Wire debounced cached async drift judgment into RecordWindow

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-05-31 17:14:20 -04:00
parent 2977b903c2
commit 4e7676560d
2 changed files with 245 additions and 0 deletions
@@ -181,6 +181,10 @@ func New(snapshotPath string) (*Controller, error) {
 	_ = store.PruneOlderThan(c.sessionsDir, sessionRetention, c.clock())
 	if c.runtimeState == domain.RuntimeActive && s.SessionID != "" {
 		c.allowedClasses = s.AllowedWindowClasses
+		// Drift state is not persisted: recompute fresh after restart to avoid
+		// stale interrupts. This also initializes judgedClasses (the pipeline
+		// writes to it) so a restored Active session never panics on a nil map.
+		c.resetDriftLocked()
 		c.replayStats(s.SessionID)
 	}
 	return c, nil
@@ -590,9 +594,100 @@ func (c *Controller) RecordWindow(snap evidence.WindowSnapshot) {
 	now := c.clock()
 	_ = store.AppendFocus(c.sessionsDir, c.stats.SessionID, focusEvent(now, snap))
 	c.applyEvent(now, snap)
+	launch := c.evaluateDriftLocked(now, snap)
+	c.mu.Unlock()
+	if launch != nil {
+		go launch()
+	}
+	c.notify()
+}
+
+// evaluateDriftLocked runs the local-first drift pipeline for one observation
+// and updates synchronous drift state. When an async judgment is warranted it
+// returns the judging closure; the caller runs it in a goroutine after
+// releasing the mutex (the M2 RequestCoach discipline). Caller holds mu and has
+// already verified the runtime is Active.
+func (c *Controller) evaluateDriftLocked(now time.Time, snap evidence.WindowSnapshot) func() {
+	class, title := snap.Class, snap.Title
+
+	// 1. Local match: authoritative on-task, no LLM.
+	ac := domain.AllowedContext{WindowClasses: c.allowedClasses}
+	if evidence.MatchesAllowed(ac, class, title) {
+		c.driftStatus = driftOnTask
+		c.driftReason = ""
+		return nil
+	}
+
+	// 2. Per-class cache.
+	if v, ok := c.judgedClasses[class]; ok {
+		c.applyVerdictLocked(v)
+		return nil
+	}
+
+	// 3. No judge wired: never block; leave idle.
+	if c.judge == nil {
+		c.driftStatus = driftIdle
+		c.driftReason = ""
+		return nil
+	}
+
+	// 4. Debounce: at most one judgment per driftDebounce window.
+	if !c.lastJudgedAt.IsZero() && now.Sub(c.lastJudgedAt) < driftDebounce {
+		return nil
+	}
+
+	prevStatus, prevReason := c.driftStatus, c.driftReason
+	c.driftGen++
+	gen := c.driftGen
+	c.lastJudgedAt = now
+	c.driftStatus = driftPending
+	c.driftReason = ""
+	judge := c.judge
+	commitment := ""
+	if c.commitment != nil {
+		commitment = c.commitment.NextAction + " — " + c.commitment.SuccessCondition
+	}
+
+	return func() {
+		ctx, cancel := context.WithTimeout(context.Background(), driftTimeout)
+		defer cancel()
+		v, err := judge.JudgeDrift(ctx, commitment, class, title)
+
+		c.mu.Lock()
+		if gen != c.driftGen || c.runtimeState != domain.RuntimeActive {
+			c.mu.Unlock()
+			return // stale: a newer judgment started or the session ended
+		}
+		if err != nil {
+			// Degrade: never fabricate drift. Revert the optimistic pending.
+			if c.driftStatus == driftPending {
+				c.driftStatus = prevStatus
+				c.driftReason = prevReason
+			}
+			c.mu.Unlock()
+			log.Printf("session: drift judge failed: %v", err)
+			c.notify()
+			return
+		}
+		c.judgedClasses[class] = v
+		if c.stats != nil && c.stats.Current.Class == class {
+			c.applyVerdictLocked(v)
+		}
 		c.mu.Unlock()
 		c.notify()
 	}
+}
+
+// applyVerdictLocked maps a verdict onto drift state. Caller holds mu.
+func (c *Controller) applyVerdictLocked(v ai.Verdict) {
+	if v.OnTask {
+		c.driftStatus = driftOnTask
+		c.driftReason = ""
+		return
+	}
+	c.driftStatus = driftDrifting
+	c.driftReason = v.Reason
+}

 // applyEvent advances stats by one observation: it credits the prior segment to
 // the prior bucket, counts a context switch on key change, and records the new
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"path/filepath"
+	"sync/atomic"
 	"testing"
 	"time"

@@ -386,6 +387,155 @@ func TestOnTaskAppendsCurrentClass(t *testing.T) {
 	}
 }

+type fakeJudge struct {
+	verdict ai.Verdict
+	err     error
+	gate    chan struct{}
+	calls   int32
+}
+
+func (f *fakeJudge) JudgeDrift(ctx context.Context, commitment, class, title string) (ai.Verdict, error) {
+	atomic.AddInt32(&f.calls, 1)
+	if f.gate != nil {
+		<-f.gate
+	}
+	return f.verdict, f.err
+}
+
+func waitDriftStatus(t *testing.T, c *Controller, want string) State {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		st := c.State()
+		if st.Drift != nil && st.Drift.Status == want {
+			return st
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("drift status never reached %q (last: %+v)", want, c.State().Drift)
+	return State{}
+}
+
+func startActive(t *testing.T, c *Controller, allowed []string) {
+	t.Helper()
+	if err := c.EnterPlanning(); err != nil {
+		t.Fatalf("planning: %v", err)
+	}
+	if err := c.StartManualCommitment("write report", "report drafted", 25*time.Minute, allowed); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+}
+
+func obs(class, title string) evidence.WindowSnapshot {
+	return evidence.WindowSnapshot{Class: class, Title: title, Health: evidence.EvidenceHealth{Available: true}}
+}
+
+func TestLocalMatchIsOnTaskWithoutJudge(t *testing.T) {
+	c, _ := newTestController(t)
+	fj := &fakeJudge{verdict: ai.Verdict{OnTask: false, Reason: "should not be called"}}
+	c.SetDriftJudge(fj)
+	startActive(t, c, []string{"code"})
+	c.RecordWindow(obs("Code", "main.go"))
+	st := c.State()
+	if st.Drift == nil || st.Drift.Status != "ontask" {
+		t.Fatalf("local match should be on-task, got %+v", st.Drift)
+	}
+	if atomic.LoadInt32(&fj.calls) != 0 {
+		t.Fatalf("judge must not be called on a local match")
+	}
+}
+
+func TestUnmatchedWindowIsJudgedDrifting(t *testing.T) {
+	c, _ := newTestController(t)
+	c.SetDriftJudge(&fakeJudge{verdict: ai.Verdict{OnTask: false, Reason: "YouTube is off-task"}})
+	startActive(t, c, []string{"code"})
+	c.RecordWindow(obs("firefox", "YouTube"))
+	st := waitDriftStatus(t, c, "drifting")
+	if st.Drift.Reason != "YouTube is off-task" {
+		t.Fatalf("reason not surfaced: %+v", st.Drift)
+	}
+}
+
+func TestPerClassCacheAvoidsSecondJudge(t *testing.T) {
+	c, _ := newTestController(t)
+	now := time.Now()
+	c.SetClock(func() time.Time { return now })
+	fj := &fakeJudge{verdict: ai.Verdict{OnTask: false, Reason: "off-task"}}
+	c.SetDriftJudge(fj)
+	startActive(t, c, []string{"code"})
+	c.RecordWindow(obs("firefox", "YouTube"))
+	waitDriftStatus(t, c, "drifting")
+	// Advance past the debounce window so a second judge call would be allowed:
+	// this isolates the per-class cache as the only reason the judge is skipped.
+	now = now.Add(2 * driftDebounce)
+	c.RecordWindow(obs("firefox", "Reddit")) // same class, should hit cache
+	time.Sleep(50 * time.Millisecond)
+	if got := atomic.LoadInt32(&fj.calls); got != 1 {
+		t.Fatalf("cached class should not re-judge, calls=%d", got)
+	}
+}
+
+func TestDebounceLimitsJudgeCalls(t *testing.T) {
+	c, _ := newTestController(t)
+	now := time.Now()
+	c.SetClock(func() time.Time { return now })
+	fj := &fakeJudge{verdict: ai.Verdict{OnTask: false, Reason: "off"}, gate: make(chan struct{})}
+	c.SetDriftJudge(fj)
+	startActive(t, c, []string{"code"})
+	c.RecordWindow(obs("firefox", "a")) // launches one (blocked on gate)
+	c.RecordWindow(obs("chrome", "b"))  // within debounce window -> suppressed
+	time.Sleep(50 * time.Millisecond)
+	if got := atomic.LoadInt32(&fj.calls); got != 1 {
+		t.Fatalf("debounce should allow one judge call, calls=%d", got)
+	}
+	close(fj.gate)
+}
+
+func TestStaleJudgmentDiscardedAfterEnd(t *testing.T) {
+	c, _ := newTestController(t)
+	fj := &fakeJudge{verdict: ai.Verdict{OnTask: false, Reason: "off"}, gate: make(chan struct{})}
+	c.SetDriftJudge(fj)
+	startActive(t, c, []string{"code"})
+	c.RecordWindow(obs("firefox", "YouTube")) // launches, blocks on gate
+	waitDriftStatus(t, c, "pending")
+	_ = c.Complete()
+	_ = c.End()
+	close(fj.gate) // release stale goroutine; must be discarded
+	time.Sleep(50 * time.Millisecond)
+	if c.State().RuntimeState != domain.RuntimeLocked {
+		t.Fatalf("should be Locked")
+	}
+}
+
+func TestNilJudgeLeavesUnmatchedIdle(t *testing.T) {
+	c, _ := newTestController(t)
+	startActive(t, c, []string{"code"})
+	c.RecordWindow(obs("firefox", "YouTube"))
+	st := c.State()
+	if st.Drift == nil || st.Drift.Status != "idle" {
+		t.Fatalf("nil judge should leave unmatched idle, got %+v", st.Drift)
+	}
+}
+
+func TestRestoredActiveSessionCanBeJudged(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.json")
+	first, err := New(path)
+	if err != nil {
+		t.Fatalf("new: %v", err)
+	}
+	_ = first.EnterPlanning()
+	if err := first.StartManualCommitment("write report", "drafted", 25*time.Minute, []string{"code"}); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	second, err := New(path)
+	if err != nil {
+		t.Fatalf("reload: %v", err)
+	}
+	second.SetDriftJudge(&fakeJudge{verdict: ai.Verdict{OnTask: false, Reason: "off-task"}})
+	second.RecordWindow(obs("firefox", "YouTube")) // must not panic on nil judgedClasses map
+	waitDriftStatus(t, second, "drifting")
+}
+
 func TestLeavingPlanningClearsCoach(t *testing.T) {
 	c, _ := newTestController(t)
 	_ = c.EnterPlanning()