From d3f0526355ecc7e0fb5bf888c4bf999a4a994595 Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Mon, 1 Jun 2026 17:57:26 -0400 Subject: [PATCH] Split async AI roles into roles.go Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/session/roles.go | 375 ++++++++++++++++++++++++++++++++++++ internal/session/session.go | 364 ---------------------------------- 2 files changed, 375 insertions(+), 364 deletions(-) create mode 100644 internal/session/roles.go diff --git a/internal/session/roles.go b/internal/session/roles.go new file mode 100644 index 0000000..d1a8436 --- /dev/null +++ b/internal/session/roles.go @@ -0,0 +1,375 @@ +package session + +import ( + "antidrift/internal/ai" + "antidrift/internal/domain" + "antidrift/internal/knowledge" + "antidrift/internal/store" + "antidrift/internal/tasks" + "context" + "errors" + "fmt" + "strings" + "time" +) + +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" +) + +// 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") +} + +// 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" + } +} diff --git a/internal/session/session.go b/internal/session/session.go index c9870b1..8f2110a 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -5,13 +5,10 @@ package session import ( - "context" "errors" - "fmt" "log" "path/filepath" "sort" - "strings" "sync" "time" @@ -32,49 +29,6 @@ const ( 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") @@ -247,247 +201,6 @@ func (c *Controller) EnterPlanning() error { 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() @@ -503,83 +216,6 @@ func (c *Controller) EnforcementLevelForTest() domain.EnforcementLevel { 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.