From 4e7676560d85c7edd62d6963d2f3b23792024206 Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Sun, 31 May 2026 17:14:20 -0400 Subject: [PATCH] Wire debounced cached async drift judgment into RecordWindow Co-Authored-By: Claude Opus 4.8 --- internal/session/session.go | 95 ++++++++++++++++++++ internal/session/session_test.go | 150 +++++++++++++++++++++++++++++++ 2 files changed, 245 insertions(+) diff --git a/internal/session/session.go b/internal/session/session.go index 0cce2c2..870cb68 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -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,10 +594,101 @@ 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 // current window. Used by both live tracking and crash replay. Caller holds mu. diff --git a/internal/session/session_test.go b/internal/session/session_test.go index 5ca0c8d..fcd3933 100644 --- a/internal/session/session_test.go +++ b/internal/session/session_test.go @@ -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()