// Package session owns the daemon's in-memory state of truth and persists a // 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 ( "context" "errors" "fmt" "log" "path/filepath" "sort" "strings" "sync" "time" "antidrift/internal/ai" "antidrift/internal/domain" "antidrift/internal/enforce" "antidrift/internal/evidence" "antidrift/internal/knowledge" "antidrift/internal/statemachine" "antidrift/internal/store" "antidrift/internal/tasks" "github.com/google/uuid" ) const ( unavailableTitle = "(evidence unavailable)" sessionRetention = 30 * 24 * time.Hour ) const coachTimeout = 60 * time.Second const ( coachIdle = "idle" coachPending = "pending" coachReady = "ready" coachError = "error" ) const tasksTimeout = 30 * time.Second const ( tasksIdle = "idle" tasksPending = "pending" tasksReady = "ready" tasksError = "error" ) const knowledgeTimeout = 10 * time.Second const ( knowledgeIdle = "idle" knowledgePending = "pending" knowledgeReady = "ready" knowledgeAbsent = "absent" knowledgeError = "error" ) const reflectionTimeout = 30 * time.Second const reflectionHistoryN = 5 // reflectionTopBuckets caps how many per-window time buckets the finished-session // block lists for the reviewer, keeping the prompt compact. const reflectionTopBuckets = 3 const ( reflectionIdle = "idle" reflectionPending = "pending" reflectionReady = "ready" reflectionAbsent = "absent" ) var ErrNotPlanning = errors.New("session: coaching is only available while planning") var ErrNotActive = errors.New("session: only available while a commitment is active") // 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 auditPath string sessionsDir string clock func() time.Time onChange func() latestWindow evidence.WindowSnapshot stats *EvidenceStats outcomePending string coach ai.Coach coachStatus string coachProposal *ai.Proposal coachErr string coachGen int tasksProvider tasks.Provider tasksStatus string tasksList []tasks.Task tasksGen int knowledgeSrc knowledge.Source knowledgeStatus string knowledgeText string // cached grounding the coach reads knowledgePath string // selected path; "" = adapter default knowledgeChars int knowledgeGen int reviewer ai.Reviewer reflectionStatus string reflectionRecap string carryForward string // latest-wins takeaway; grounds the next coach reflectionGen int allowedClasses []string // durable: the active session's allowed window classes enforcementLevel domain.EnforcementLevel // durable: block enables window-minimize enforcement guard enforce.Guard judge ai.DriftJudge driftStatus string driftReason string driftGen int nudgeEpoch int // identifies the current on-task stretch; nudge staleness guard lastJudgedAt time.Time judgedClasses map[string]ai.Verdict nudge ai.Nudger recentTitles []string // in-memory ring of recent distinct titles this session nudgeMessage string // current soft advisory ("" = none) lastNudgedAt time.Time } // 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, auditPath: filepath.Join(dir, "audit.jsonl"), sessionsDir: filepath.Join(dir, "sessions"), clock: time.Now, outcomePending: s.OutcomePending, reflectionStatus: s.ReflectionStatus, reflectionRecap: s.ReflectionRecap, carryForward: s.CarryForward, } if s.DeadlineUnixSecs > 0 { c.deadline = time.Unix(s.DeadlineUnixSecs, 0) } if c.runtimeState == "" { c.runtimeState = domain.RuntimeLocked } _ = store.PruneOlderThan(c.sessionsDir, sessionRetention, c.clock()) if c.runtimeState == domain.RuntimeActive && s.SessionID != "" { c.allowedClasses = s.AllowedWindowClasses c.enforcementLevel = s.EnforcementLevel // 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 } // 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() defer c.mu.Unlock() return c.stateLocked() } // Deadline returns the active commitment deadline, or the zero time. func (c *Controller) Deadline() time.Time { c.mu.Lock() defer c.mu.Unlock() return c.deadline } func (c *Controller) persistLocked() error { 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 } snap.AllowedWindowClasses = c.allowedClasses snap.EnforcementLevel = c.enforcementLevel snap.ReflectionStatus = c.reflectionStatus snap.ReflectionRecap = c.reflectionRecap snap.CarryForward = c.carryForward return store.Save(c.snapshotPath, snap) } // EnterPlanning moves Locked -> Planning. func (c *Controller) EnterPlanning() error { c.mu.Lock() defer c.mu.Unlock() next, err := statemachine.TransitionRuntime(c.runtimeState, statemachine.EnterPlanning) if err != nil { return err } c.runtimeState = next c.resetCoachLocked() c.startTasksFetchLocked() c.startKnowledgeFetchLocked() return c.persistLocked() } // SetCoach injects the AI coach. Mirrors SetOnChange. A nil coach makes // RequestCoach degrade gracefully. func (c *Controller) SetCoach(coach ai.Coach) { c.mu.Lock() c.coach = coach c.mu.Unlock() } // resetCoachLocked returns coach state to idle and invalidates any in-flight // request. Caller holds mu. func (c *Controller) resetCoachLocked() { c.coachStatus = coachIdle c.coachProposal = nil c.coachErr = "" c.coachGen++ } // SetTasks injects the Tasks provider. Mirrors SetCoach. A nil provider keeps // the planning tasks list absent. func (c *Controller) SetTasks(p tasks.Provider) { c.mu.Lock() c.tasksProvider = p c.mu.Unlock() } // startTasksFetchLocked kicks off an asynchronous Today() fetch when a provider // is set. Mirrors RequestCoach: generation-guarded, discards stale or // post-planning results, and notifies on completion. Caller holds mu. func (c *Controller) startTasksFetchLocked() { c.tasksList = nil if c.tasksProvider == nil { c.tasksStatus = tasksIdle return } c.tasksGen++ gen := c.tasksGen c.tasksStatus = tasksPending provider := c.tasksProvider go func() { ctx, cancel := context.WithTimeout(context.Background(), tasksTimeout) defer cancel() list, err := provider.Today(ctx) c.mu.Lock() if gen != c.tasksGen || c.runtimeState != domain.RuntimePlanning { c.mu.Unlock() return // stale or left planning: discard } if err != nil { c.tasksStatus = tasksError c.tasksList = nil } else { c.tasksStatus = tasksReady c.tasksList = list } c.mu.Unlock() c.notify() }() } // SetKnowledge injects the Knowledge source. Mirrors SetTasks. A nil source // keeps the planning knowledge view absent and the coach ungrounded. func (c *Controller) SetKnowledge(s knowledge.Source) { c.mu.Lock() c.knowledgeSrc = s c.mu.Unlock() } // SetKnowledgePath selects an explicit profile path (session-only; not // persisted). While planning, it re-loads immediately so the indicator and the // cached grounding update. An empty path resets to the adapter default. func (c *Controller) SetKnowledgePath(path string) { c.mu.Lock() c.knowledgePath = strings.TrimSpace(path) planning := c.runtimeState == domain.RuntimePlanning if planning { c.startKnowledgeFetchLocked() } c.mu.Unlock() if planning { c.notify() } } // startKnowledgeFetchLocked kicks off an asynchronous Load when a source is set. // Mirrors startTasksFetchLocked: generation-guarded, discards stale or // post-planning results, and notifies on completion. The loaded text is cached // in knowledgeText for the coach to read. Caller holds mu. func (c *Controller) startKnowledgeFetchLocked() { c.knowledgeText = "" c.knowledgeChars = 0 if c.knowledgeSrc == nil { c.knowledgeStatus = knowledgeIdle return } c.knowledgeGen++ gen := c.knowledgeGen c.knowledgeStatus = knowledgePending src := c.knowledgeSrc path := c.knowledgePath go func() { ctx, cancel := context.WithTimeout(context.Background(), knowledgeTimeout) defer cancel() prof, err := src.Load(ctx, path) c.mu.Lock() if gen != c.knowledgeGen || c.runtimeState != domain.RuntimePlanning { c.mu.Unlock() return // stale or left planning: discard } if err != nil { c.knowledgeStatus = knowledgeError c.knowledgeText = "" c.knowledgeChars = 0 if prof.Path != "" { c.knowledgePath = prof.Path } } else if prof.Text == "" { c.knowledgeStatus = knowledgeAbsent c.knowledgeText = "" c.knowledgeChars = 0 c.knowledgePath = prof.Path } else { c.knowledgeStatus = knowledgeReady c.knowledgeText = prof.Text c.knowledgeChars = len(prof.Text) c.knowledgePath = prof.Path } c.mu.Unlock() c.notify() }() } // SetReviewer injects the AI reviewer. A nil reviewer keeps reflection idle and // leaves the coach ungrounded by any carry-forward. func (c *Controller) SetReviewer(r ai.Reviewer) { c.mu.Lock() c.reviewer = r c.mu.Unlock() } // startReflectionFetchLocked kicks off an asynchronous reflection when a // reviewer is set, on entering Review. Unlike the tasks/knowledge fetches, the // completion guard is generation-only (not state-gated): the carry-forward must // still apply if the user clicks End before the reviewer returns. A superseded // review (a later session's fetch) bumps the generation and discards this one. // The recap and carry-forward are cleared up front so a failed/slow reviewer // never leaves stale data from the previous session. Caller holds mu. func (c *Controller) startReflectionFetchLocked() { c.reflectionRecap = "" c.carryForward = "" if c.reviewer == nil { c.reflectionStatus = reflectionIdle return } c.reflectionGen++ gen := c.reflectionGen c.reflectionStatus = reflectionPending reviewer := c.reviewer finished := c.buildReflectionFinishedLocked() // Read the history synchronously, here under the lock, on purpose: it must // happen-before End appends the just-finished session to the audit chain, so // that session is excluded from "recent history" and not double-counted (it // is already carried in `finished`). Moving this into the goroutine would // race with End's append and reintroduce that double-count. The read is // bounded to reflectionHistoryN summaries and runs once per Review entry, not // on any hot path. history := buildReflectionHistory(c.auditPath) go func() { ctx, cancel := context.WithTimeout(context.Background(), reflectionTimeout) defer cancel() refl, err := reviewer.Review(ctx, finished, history) c.mu.Lock() if gen != c.reflectionGen { c.mu.Unlock() return // superseded review: discard } if err != nil || strings.TrimSpace(refl.Recap) == "" { c.reflectionStatus = reflectionAbsent c.reflectionRecap = "" c.carryForward = "" } else { c.reflectionStatus = reflectionReady c.reflectionRecap = refl.Recap c.carryForward = refl.CarryForward } _ = c.persistLocked() c.mu.Unlock() c.notify() }() } // buildReflectionFinishedLocked renders the just-finished session as a compact // block for the reviewer. Caller holds mu; c.stats/c.commitment are still set // (End clears them, but enterReview runs before End). Reuses bucketViews for the // per-window totals, already sorted desc by seconds. func (c *Controller) buildReflectionFinishedLocked() string { var na, sc string if c.commitment != nil { na, sc = c.commitment.NextAction, c.commitment.SuccessCondition } outcome := c.outcomePending if outcome == "" { outcome = "completed" } var b strings.Builder fmt.Fprintf(&b, "Next action: %s\n", na) fmt.Fprintf(&b, "Success condition: %s\n", sc) fmt.Fprintf(&b, "Outcome: %s\n", outcome) if c.stats != nil { fmt.Fprintf(&b, "Context switches: %d\n", c.stats.SwitchCount) for i, bv := range bucketViews(c.stats.Buckets) { if i >= reflectionTopBuckets { break } fmt.Fprintf(&b, "- %s ยท %s: %dm\n", bv.Class, bv.Title, bv.Seconds/60) } } return strings.TrimRight(b.String(), "\n") } // buildReflectionHistory renders the last few prior sessions as compact lines. // The just-finished session is not yet in the chain (End appends it), so it is // not double-counted. Returns "" when there is no usable history. func buildReflectionHistory(auditPath string) string { sums, err := store.RecentSessions(auditPath, reflectionHistoryN) if err != nil || len(sums) == 0 { return "" } var b strings.Builder for _, s := range sums { top := "" if len(s.Buckets) > 0 { top = fmt.Sprintf(", top %s %dm", s.Buckets[0].Class, s.Buckets[0].Seconds/60) } fmt.Fprintf(&b, "- %s: %s (%d switches%s)\n", s.Outcome, s.NextAction, s.SwitchCount, top) } return strings.TrimRight(b.String(), "\n") } // AllowedClassesForTest exposes the session allowed classes for tests. func (c *Controller) AllowedClassesForTest() []string { c.mu.Lock() defer c.mu.Unlock() return append([]string(nil), c.allowedClasses...) } // EnforcementLevelForTest exposes the active session's enforcement level. Tests // only. func (c *Controller) EnforcementLevelForTest() domain.EnforcementLevel { c.mu.Lock() defer c.mu.Unlock() return c.enforcementLevel } // composedGroundingLocked combines the standing profile (knowledge port) with // the latest carry-forward takeaway into the single free-form grounding string // the coach already accepts. Caller holds mu. func (c *Controller) composedGroundingLocked() string { g := c.knowledgeText if c.carryForward != "" { if g != "" { g += "\n\n" } g += "Last session's takeaway: " + c.carryForward } return g } // RequestCoach starts an async coach call for intent. Returns ErrNotPlanning if // not in planning; otherwise never a hard error (failures surface as coach // state). The proposal is ephemeral and never persisted. func (c *Controller) RequestCoach(intent string) error { c.mu.Lock() if c.runtimeState != domain.RuntimePlanning { c.mu.Unlock() return ErrNotPlanning } if c.coach == nil { c.coachStatus = coachError c.coachErr = "coach unavailable" c.coachProposal = nil c.mu.Unlock() c.notify() return nil } c.coachGen++ gen := c.coachGen c.coachStatus = coachPending c.coachErr = "" c.coachProposal = nil coach := c.coach grounding := c.composedGroundingLocked() c.mu.Unlock() c.notify() go func() { ctx, cancel := context.WithTimeout(context.Background(), coachTimeout) defer cancel() prop, err := coach.Coach(ctx, intent, grounding) c.mu.Lock() if gen != c.coachGen || c.runtimeState != domain.RuntimePlanning { c.mu.Unlock() return // stale or left planning: discard } if err != nil { c.coachStatus = coachError c.coachErr = coachErrorMessage(err) c.coachProposal = nil } else { c.coachStatus = coachReady c.coachProposal = &prop c.coachErr = "" } c.mu.Unlock() c.notify() }() return nil } func coachErrorMessage(err error) string { switch { case errors.Is(err, ai.ErrEmptyResponse), errors.Is(err, ai.ErrNoJSON), errors.Is(err, ai.ErrInvalidProposal): return "coach returned an unusable response" case errors.Is(err, context.DeadlineExceeded): return "coach timed out" default: return "coach unavailable" } } // 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, allowedClasses []string, level domain.EnforcementLevel) error { c.mu.Lock() defer c.mu.Unlock() commitment, err := domain.NewManual(nextAction, successCondition, timebox) if err != nil { return err } commitment.State, err = statemachine.TransitionCommitment(commitment.State, statemachine.CommitmentActivate) if err != nil { return err } next, err := statemachine.TransitionRuntime(c.runtimeState, statemachine.ActivateAccepted) if err != nil { return err } now := c.clock() c.runtimeState = next c.commitment = &commitment c.deadline = now.Add(timebox) c.outcomePending = "" c.resetCoachLocked() c.allowedClasses = append([]string(nil), allowedClasses...) c.enforcementLevel = level c.resetDriftLocked() 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 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) if err != nil { return err } if c.commitment != nil { completed, err := statemachine.TransitionCommitment(c.commitment.State, statemachine.Complete) if err != nil { return err } 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 c.startReflectionFetchLocked() return c.persistLocked() } // 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() next, err := statemachine.TransitionRuntime(c.runtimeState, statemachine.EndWorkPeriod) 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 = "" c.allowedClasses = nil c.enforcementLevel = "" c.resetDriftLocked() 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, } }