From df384afb7cc3a33489da007753978b479e18582b Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Sun, 31 May 2026 12:49:19 -0400 Subject: [PATCH] M1: controller evidence stats, accumulation, replay, audit-at-End Co-Authored-By: Claude Opus 4.8 --- internal/session/session.go | 293 +++++++++++++++++++++++++++++-- internal/session/session_test.go | 130 ++++++++++++++ 2 files changed, 404 insertions(+), 19 deletions(-) diff --git a/internal/session/session.go b/internal/session/session.go index 53207e7..43207a3 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -1,23 +1,58 @@ // Package session owns the daemon's in-memory state of truth and persists a -// snapshot on every change. Transitions go through the pure statemachine. +// snapshot on every change. Transitions go through the pure statemachine. It +// also owns per-session evidence stats: it accumulates active-window time while +// Active, logs raw focus events, and seals each session into the audit chain. package session import ( + "log" + "path/filepath" + "sort" "sync" "time" "antidrift/internal/domain" + "antidrift/internal/evidence" "antidrift/internal/statemachine" "antidrift/internal/store" + + "github.com/google/uuid" ) +const ( + unavailableTitle = "(evidence unavailable)" + sessionRetention = 30 * 24 * time.Hour +) + +// bucketKey identifies a time bucket; Title is the scrubbed title. +type bucketKey struct{ Class, Title string } + +// EvidenceStats is the in-memory accounting for the current session only. +type EvidenceStats struct { + SessionID string + StartedUnix int64 + Buckets map[bucketKey]time.Duration + SwitchCount int + Current evidence.WindowSnapshot + lastFocusAt time.Time + lastKey bucketKey + hasLast bool +} + // Controller holds runtime state and the active commitment behind a mutex. type Controller struct { - mu sync.Mutex - runtimeState domain.RuntimeState - commitment *domain.Commitment - deadline time.Time - snapshotPath string + mu sync.Mutex + runtimeState domain.RuntimeState + commitment *domain.Commitment + deadline time.Time + snapshotPath string + auditPath string + sessionsDir string + clock func() time.Time + onChange func() + latestWindow evidence.WindowSnapshot + stats *EvidenceStats + outcomePending string } // CommitmentView is the UI projection of the active commitment. @@ -28,22 +63,49 @@ type CommitmentView struct { DeadlineUnixSecs int64 `json:"deadline_unix_secs"` } +// WindowView / BucketView / EvidenceView are the evidence projection. +type WindowView struct { + Class string `json:"class"` + Title string `json:"title"` +} + +type BucketView struct { + Class string `json:"class"` + Title string `json:"title"` + Seconds int64 `json:"seconds"` +} + +type EvidenceView struct { + Available bool `json:"available"` + Reason string `json:"reason"` + Current WindowView `json:"current"` + SwitchCount int `json:"switch_count"` + Buckets []BucketView `json:"buckets"` +} + // State is the broadcastable view of the controller. type State struct { RuntimeState domain.RuntimeState `json:"runtime_state"` Commitment *CommitmentView `json:"commitment"` + Evidence *EvidenceView `json:"evidence"` } -// New loads any persisted snapshot, or starts Locked. +// New loads any persisted snapshot, prunes stale session logs, and rebuilds +// in-memory stats from the raw log if a live session was interrupted. func New(snapshotPath string) (*Controller, error) { s, err := store.Load(snapshotPath) if err != nil { return nil, err } + dir := filepath.Dir(snapshotPath) c := &Controller{ - runtimeState: s.RuntimeState, - commitment: s.Commitment, - snapshotPath: snapshotPath, + runtimeState: s.RuntimeState, + commitment: s.Commitment, + snapshotPath: snapshotPath, + auditPath: filepath.Join(dir, "audit.jsonl"), + sessionsDir: filepath.Join(dir, "sessions"), + clock: time.Now, + outcomePending: s.OutcomePending, } if s.DeadlineUnixSecs > 0 { c.deadline = time.Unix(s.DeadlineUnixSecs, 0) @@ -51,9 +113,38 @@ func New(snapshotPath string) (*Controller, error) { if c.runtimeState == "" { c.runtimeState = domain.RuntimeLocked } + _ = store.PruneOlderThan(c.sessionsDir, sessionRetention, c.clock()) + if c.runtimeState == domain.RuntimeActive && s.SessionID != "" { + c.replayStats(s.SessionID) + } return c, nil } +// SetClock overrides the time source (tests only). Call before starting a +// session. +func (c *Controller) SetClock(f func() time.Time) { + c.mu.Lock() + c.clock = f + c.mu.Unlock() +} + +// SetOnChange registers a callback fired after an evidence-driven state change +// (focus updates). It is invoked with the mutex released. +func (c *Controller) SetOnChange(f func()) { + c.mu.Lock() + c.onChange = f + c.mu.Unlock() +} + +func (c *Controller) notify() { + c.mu.Lock() + f := c.onChange + c.mu.Unlock() + if f != nil { + f() + } +} + // State returns the current broadcastable state. Safe for concurrent use. func (c *Controller) State() State { c.mu.Lock() @@ -81,14 +172,39 @@ func (c *Controller) stateLocked() State { } st.Commitment = view } + if c.stats != nil { + st.Evidence = &EvidenceView{ + Available: c.stats.Current.Health.Available, + Reason: c.stats.Current.Health.Reason, + Current: WindowView{Class: c.stats.Current.Class, Title: c.stats.Current.Title}, + SwitchCount: c.stats.SwitchCount, + Buckets: bucketViews(c.stats.Buckets), + } + } return st } +func bucketViews(buckets map[bucketKey]time.Duration) []BucketView { + out := make([]BucketView, 0, len(buckets)) + for k, d := range buckets { + out = append(out, BucketView{Class: k.Class, Title: k.Title, Seconds: int64(d.Seconds())}) + } + sort.Slice(out, func(i, j int) bool { return out[i].Seconds > out[j].Seconds }) + return out +} + func (c *Controller) persistLocked() error { - snap := store.Snapshot{RuntimeState: c.runtimeState, Commitment: c.commitment} + snap := store.Snapshot{ + RuntimeState: c.runtimeState, + Commitment: c.commitment, + OutcomePending: c.outcomePending, + } if !c.deadline.IsZero() { snap.DeadlineUnixSecs = c.deadline.Unix() } + if c.stats != nil { + snap.SessionID = c.stats.SessionID + } return store.Save(c.snapshotPath, snap) } @@ -104,8 +220,9 @@ func (c *Controller) EnterPlanning() error { return c.persistLocked() } -// StartManualCommitment validates input, activates a new commitment, and moves -// Planning -> Active. +// StartManualCommitment validates input, activates a new commitment, mints a +// session, seeds evidence stats from the latest window, and moves Planning -> +// Active. func (c *Controller) StartManualCommitment(nextAction, successCondition string, timebox time.Duration) error { c.mu.Lock() defer c.mu.Unlock() @@ -121,16 +238,31 @@ func (c *Controller) StartManualCommitment(nextAction, successCondition string, if err != nil { return err } + now := c.clock() c.runtimeState = next c.commitment = &commitment - c.deadline = time.Now().Add(timebox) + c.deadline = now.Add(timebox) + c.outcomePending = "" + + sessionID := "session-" + uuid.Must(uuid.NewV7()).String() + c.stats = &EvidenceStats{ + SessionID: sessionID, + StartedUnix: now.Unix(), + Buckets: map[bucketKey]time.Duration{}, + } + seed := c.latestWindow + _ = store.AppendFocus(c.sessionsDir, sessionID, focusEvent(now, seed)) + c.applyEvent(now, seed) return c.persistLocked() } -// Complete moves Active -> Review and marks the commitment completed. Both -// transitions are computed before any field is mutated, so a failure leaves -// state unchanged. -func (c *Controller) Complete() error { +// Complete moves Active -> Review with a "completed" outcome. +func (c *Controller) Complete() error { return c.enterReview("completed") } + +// Expire moves Active -> Review with an "expired" outcome (timebox elapsed). +func (c *Controller) Expire() error { return c.enterReview("expired") } + +func (c *Controller) enterReview(outcome string) error { c.mu.Lock() defer c.mu.Unlock() next, err := statemachine.TransitionRuntime(c.runtimeState, statemachine.CompleteForReview) @@ -144,11 +276,18 @@ func (c *Controller) Complete() error { } c.commitment.State = completed } + // Flush the final open segment, then freeze accounting. + if c.stats != nil && c.stats.hasLast { + c.stats.Buckets[c.stats.lastKey] += c.clock().Sub(c.stats.lastFocusAt) + c.stats.hasLast = false + } c.runtimeState = next + c.outcomePending = outcome return c.persistLocked() } -// End moves Review -> Locked and clears the commitment. +// End moves Review -> Locked, writes the session summary to the audit chain, +// and clears the commitment and stats. func (c *Controller) End() error { c.mu.Lock() defer c.mu.Unlock() @@ -156,8 +295,124 @@ func (c *Controller) End() error { if err != nil { return err } + if c.stats != nil { + if err := store.AppendSession(c.auditPath, c.buildSummaryLocked()); err != nil { + // State integrity over audit completeness: the transition still + // completes. Surfaced for the operator; no auto-retry in M1. + log.Printf("session: audit append failed: %v", err) + } + } c.runtimeState = next c.commitment = nil c.deadline = time.Time{} + c.stats = nil + c.outcomePending = "" return c.persistLocked() } + +func (c *Controller) buildSummaryLocked() store.SessionSummary { + buckets := make([]store.BucketTotal, 0, len(c.stats.Buckets)) + for k, d := range c.stats.Buckets { + buckets = append(buckets, store.BucketTotal{Class: k.Class, Title: k.Title, Seconds: int64(d.Seconds())}) + } + sort.Slice(buckets, func(i, j int) bool { return buckets[i].Seconds > buckets[j].Seconds }) + outcome := c.outcomePending + if outcome == "" { + outcome = "completed" + } + var na, sc string + if c.commitment != nil { + na, sc = c.commitment.NextAction, c.commitment.SuccessCondition + } + return store.SessionSummary{ + SessionID: c.stats.SessionID, + NextAction: na, + SuccessCond: sc, + Outcome: outcome, + StartedUnix: c.stats.StartedUnix, + EndedUnix: c.clock().Unix(), + SwitchCount: c.stats.SwitchCount, + Buckets: buckets, + } +} + +// RecordWindow ingests one sensor observation. It accumulates time only while +// Active, always tracks the latest window for display, and fires onChange after +// releasing the mutex. +func (c *Controller) RecordWindow(snap evidence.WindowSnapshot) { + c.mu.Lock() + c.latestWindow = snap + if c.runtimeState != domain.RuntimeActive || c.stats == nil { + c.mu.Unlock() + c.notify() + return + } + now := c.clock() + _ = store.AppendFocus(c.sessionsDir, c.stats.SessionID, focusEvent(now, snap)) + c.applyEvent(now, snap) + c.mu.Unlock() + c.notify() +} + +// 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. +func (c *Controller) applyEvent(now time.Time, snap evidence.WindowSnapshot) { + if c.stats.hasLast { + c.stats.Buckets[c.stats.lastKey] += now.Sub(c.stats.lastFocusAt) + } + newKey := keyFor(snap) + if c.stats.hasLast && newKey != c.stats.lastKey { + c.stats.SwitchCount++ + } + c.stats.lastKey = newKey + c.stats.lastFocusAt = now + c.stats.hasLast = true + c.stats.Current = snap +} + +// replayStats rebuilds in-memory stats from the raw session log after a crash. +func (c *Controller) replayStats(sessionID string) { + events, err := store.ReplaySession(c.sessionsDir, sessionID) + if err != nil || len(events) == 0 { + c.stats = &EvidenceStats{ + SessionID: sessionID, + StartedUnix: c.clock().Unix(), + Buckets: map[bucketKey]time.Duration{}, + } + return + } + c.stats = &EvidenceStats{ + SessionID: sessionID, + StartedUnix: events[0].AtUnixMillis / 1000, + Buckets: map[bucketKey]time.Duration{}, + } + for _, e := range events { + c.applyEvent(time.UnixMilli(e.AtUnixMillis), snapFromEvent(e)) + } +} + +func keyFor(snap evidence.WindowSnapshot) bucketKey { + if !snap.Health.Available { + return bucketKey{Class: "", Title: unavailableTitle} + } + return bucketKey{Class: snap.Class, Title: evidence.ScrubTitle(snap.Title)} +} + +func focusEvent(now time.Time, snap evidence.WindowSnapshot) store.FocusEvent { + return store.FocusEvent{ + AtUnixMillis: now.UnixMilli(), + Class: snap.Class, + Title: snap.Title, + Available: snap.Health.Available, + Reason: snap.Health.Reason, + } +} + +func snapFromEvent(e store.FocusEvent) evidence.WindowSnapshot { + return evidence.WindowSnapshot{ + Title: e.Title, + Class: e.Class, + Health: evidence.EvidenceHealth{Available: e.Available, Reason: e.Reason}, + } +} diff --git a/internal/session/session_test.go b/internal/session/session_test.go index 37b37a9..2ceff3c 100644 --- a/internal/session/session_test.go +++ b/internal/session/session_test.go @@ -6,6 +6,8 @@ import ( "time" "antidrift/internal/domain" + "antidrift/internal/evidence" + "antidrift/internal/store" ) func newTestController(t *testing.T) (*Controller, string) { @@ -105,3 +107,131 @@ func TestRestoredCommitmentKeepsDeadline(t *testing.T) { t.Fatalf("restored deadline: got %+v want %d", got, want) } } + +// fakeClock returns successive instants on demand. +type fakeClock struct{ now time.Time } + +func (f *fakeClock) advance(d time.Duration) { f.now = f.now.Add(d) } +func (f *fakeClock) fn() func() time.Time { return func() time.Time { return f.now } } + +func snap(class, title string) evidence.WindowSnapshot { + return evidence.WindowSnapshot{Class: class, Title: title, Health: evidence.EvidenceHealth{Available: true}} +} + +func bucketSeconds(t *testing.T, st State, class, title string) int64 { + t.Helper() + if st.Evidence == nil { + t.Fatalf("expected evidence projection") + } + for _, b := range st.Evidence.Buckets { + if b.Class == class && b.Title == title { + return b.Seconds + } + } + return -1 +} + +func TestAccumulatesBucketsAndSwitches(t *testing.T) { + c, _ := newTestController(t) + clk := &fakeClock{now: time.Unix(1000, 0)} + c.SetClock(clk.fn()) + c.RecordWindow(snap("code", "antidrift")) // latest window before start + + _ = c.EnterPlanning() + _ = c.StartManualCommitment("work", "done", 25*time.Minute) // seeds at t=1000 with code/antidrift + + clk.advance(10 * time.Second) + c.RecordWindow(snap("firefox", "docs")) // credits 10s to code/antidrift, switch -> firefox + clk.advance(20 * time.Second) + c.RecordWindow(snap("code", "antidrift")) // credits 20s to firefox/docs, switch back + + st := c.State() + if got := bucketSeconds(t, st, "code", "antidrift"); got != 10 { + t.Errorf("code bucket = %d, want 10", got) + } + if got := bucketSeconds(t, st, "firefox", "docs"); got != 20 { + t.Errorf("firefox bucket = %d, want 20", got) + } + if st.Evidence.SwitchCount != 2 { + t.Errorf("switch count = %d, want 2", st.Evidence.SwitchCount) + } +} + +func TestNoAccountingOutsideActive(t *testing.T) { + c, _ := newTestController(t) + clk := &fakeClock{now: time.Unix(1000, 0)} + c.SetClock(clk.fn()) + // Locked: RecordWindow must not create stats. + c.RecordWindow(snap("code", "x")) + if c.State().Evidence != nil { + t.Fatalf("no session: evidence should be nil") + } +} + +func TestUnavailableAccruesToReservedBucket(t *testing.T) { + c, _ := newTestController(t) + clk := &fakeClock{now: time.Unix(1000, 0)} + c.SetClock(clk.fn()) + _ = c.EnterPlanning() + _ = c.StartManualCommitment("work", "done", 25*time.Minute) // seed: empty unavailable latestWindow + // latestWindow was zero-value (unavailable) at seed time. + clk.advance(5 * time.Second) + c.RecordWindow(snap("code", "x")) // credits 5s to the reserved bucket + st := c.State() + if got := bucketSeconds(t, st, "", "(evidence unavailable)"); got != 5 { + t.Errorf("unavailable bucket = %d, want 5", got) + } +} + +func TestCrashReplayRebuildsStats(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + first, _ := New(path) + clk := &fakeClock{now: time.Unix(1000, 0)} + first.SetClock(clk.fn()) + first.RecordWindow(snap("code", "antidrift")) + _ = first.EnterPlanning() + _ = first.StartManualCommitment("work", "done", 25*time.Minute) + clk.advance(15 * time.Second) + first.RecordWindow(snap("firefox", "docs")) // 15s -> code, switch + + // Simulate crash + restart: New replays the raw log. + second, err := New(path) + if err != nil { + t.Fatalf("reopen: %v", err) + } + st := second.State() + if st.RuntimeState != domain.RuntimeActive { + t.Fatalf("restored state should be Active, got %s", st.RuntimeState) + } + if got := bucketSeconds(t, st, "code", "antidrift"); got != 15 { + t.Errorf("replayed code bucket = %d, want 15", got) + } + if st.Evidence.SwitchCount != 1 { + t.Errorf("replayed switch count = %d, want 1", st.Evidence.SwitchCount) + } +} + +func TestEndWritesAuditSummary(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "state.json") + c, _ := New(path) + clk := &fakeClock{now: time.Unix(1000, 0)} + c.SetClock(clk.fn()) + c.RecordWindow(snap("code", "antidrift")) + _ = c.EnterPlanning() + _ = c.StartManualCommitment("write plan", "plan done", 25*time.Minute) + clk.advance(30 * time.Second) + c.RecordWindow(snap("firefox", "docs")) + clk.advance(10 * time.Second) + if err := c.Complete(); err != nil { // flush: +10s to firefox/docs + t.Fatalf("complete: %v", err) + } + if err := c.End(); err != nil { + t.Fatalf("end: %v", err) + } + + auditPath := filepath.Join(dir, "audit.jsonl") + if err := store.VerifyChain(auditPath); err != nil { + t.Fatalf("chain should verify: %v", err) + } +}