felixm/antidrift
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

M1: controller evidence stats, accumulation, replay, audit-at-End

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Felix Martin
2026-05-31 12:49:19 -04:00
parent 1646321d0b
commit df384afb7c
2 changed files with 404 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/session/session.go
+274 -19
View File
@@ -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},
}
}