diff options
Diffstat (limited to 'internal/scheduler/story_orchestrator.go')
| -rw-r--r-- | internal/scheduler/story_orchestrator.go | 482 |
1 files changed, 482 insertions, 0 deletions
diff --git a/internal/scheduler/story_orchestrator.go b/internal/scheduler/story_orchestrator.go new file mode 100644 index 0000000..6ad0f86 --- /dev/null +++ b/internal/scheduler/story_orchestrator.go @@ -0,0 +1,482 @@ +// This file implements Phase 7b's story orchestrator: a deterministic, +// poll-based watcher that drives a story.Story through +// Builder -> 4 parallel Evaluators -> Arbitration -> a single human +// accept-gate at the *story* level, exactly the way Scheduler (above, in +// scheduler.go) drives a role-typed task through its retry/escalation +// ladder. It lives in this package rather than internal/story for a +// structural reason, not just stylistic taste: internal/storage already +// imports internal/story (storage/story.go), so internal/story cannot +// import internal/storage back — and this orchestrator's Store interface +// needs storage.StoryFilter/*task.Task/etc. internal/scheduler already +// imports both internal/storage and internal/task (and has zero risk of +// internal/story importing scheduler back), so it's a clean home. It also +// matches the design note this phase's task description invited: "fold it +// into internal/scheduler as a sibling to the existing Scheduler". +// +// Mechanism choice: poll-based, not a handleRunResult hook. Every task this +// orchestrator cares about (the story's root Builder task, the 4 Evaluators, +// the Arbitration task) is a top-level task (ParentTaskID == ""), so per +// task.go's state machine the *only* way one of them would ever reach +// COMPLETED on its own is via a human/chatbot POST /api/tasks/{id}/accept +// (internal/api.acceptTask), READY -> COMPLETED — executor.Pool.handleRunResult +// never transitions a top-level task to COMPLETED directly (only READY, or +// BLOCKED if it has subtasks). That's exactly the gap this orchestrator +// closes itself (see autoAccept below): the whole point of the story-level +// ceremony is that a human/chatbot should only ever have to make *one* +// accept decision (the final POST /api/stories/{id}/accept, REVIEW_READY -> +// DONE) — not six (builder + 4 evaluators + arbitration) along the way. So +// on every tick, for the specific Builder/Evaluator/Arbitration tasks it +// already knows are wired into a story's pipeline, this orchestrator +// auto-accepts (READY -> COMPLETED) them itself, using the same +// state-machine-respecting write internal/api's acceptTask uses +// (Store.UpdateTaskState, which wraps storage.DB.UpdateTaskStateBy — +// validates task.ValidTransition and writes the state_change event +// atomically, not a raw/unchecked write). This is narrowly scoped: it only +// ever touches the root task a story actually tracks (st.RootTaskID) and +// that root's structurally-discovered evaluator/arbitration dependents +// (found via ensureEvaluators/ensureArbitration's own role-matching) — never +// a blanket "auto-accept every READY task" sweep. Polling (rather than a +// handleRunResult hook) is still the right mechanism regardless of who does +// the accepting: the transition happens via a *write this orchestrator +// itself performs* on a poll tick, which is inherently poll-driven, not an +// executor-package callback. +package scheduler + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/google/uuid" + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/storage" + "github.com/thepeterstone/claudomator/internal/story" + "github.com/thepeterstone/claudomator/internal/task" +) + +// evaluatorRoles is the fixed fan-out of Evaluator roles spawned once a +// story's Builder (root_task_id) task completes. Order here is preserved +// wherever evaluator tasks are collected into a slice (e.g. building the +// Arbitration task's DependsOn), purely for determinism/testability — the +// orchestrator does not otherwise care about ordering. +var evaluatorRoles = []string{ + "evaluator_quality", + "evaluator_security", + "evaluator_correctness", + "evaluator_performance", +} + +// arbitrationRole is the role assigned to the single task spawned once all +// Evaluators for a story complete. It depends on all 4 Evaluator tasks and, +// per this phase's documented simplification (see finalizeArbitration), +// always routes the story to REVIEW_READY on completion rather than parsing +// its summary for an approve/reject verdict. +const arbitrationRole = "planner" + +// StoryStore is the subset of storage.DB methods StoryOrchestrator needs. +// Defining it as an interface (mirroring executor.Store/scheduler.Store) +// allows tests to supply an in-memory fake with no real SQLite database. +type StoryStore interface { + ListStories(filter storage.StoryFilter) ([]*story.Story, error) + UpdateStory(st *story.Story) error + GetTask(id string) (*task.Task, error) + ListDependents(taskID string) ([]*task.Task, error) + CreateTask(t *task.Task) error + UpdateTaskState(id string, newState task.State) error + CreateEvent(e *event.Event) error +} + +// StoryOrchestrator polls stories with a root_task_id set and drives them +// through the Builder -> Evaluators -> Arbitration -> REVIEW_READY ceremony, +// auto-accepting (READY -> COMPLETED) the Builder/Evaluator/Arbitration tasks +// along the way (see autoAccept) so that a human/chatbot never has to touch +// POST /api/tasks/{id}/accept for any of them. The final REVIEW_READY -> DONE +// transition is the *only* remaining human/chatbot action (see internal/api's +// POST /api/stories/{id}/accept), not something this type does itself. +type StoryOrchestrator struct { + Store StoryStore + // Pool reuses the same minimal interface Scheduler depends on + // (Submit(ctx, *task.Task) error) — satisfied directly by + // *executor.Pool, declared once in scheduler.go. + Pool Pool + Logger *slog.Logger + + // handledVerdicts dedupes per-evaluator KindEvalVerdict emission within a + // single running process, keyed by evaluator task ID. Without it, an + // evaluator that completes while its siblings are still running would + // get a fresh eval_verdict event on every poll tick until the last + // sibling finishes. This mirrors Scheduler.handled exactly: an in-memory, + // per-process guard that resets on restart. That's an accepted + // simplification here for the same reason it is for Scheduler — this is + // idempotent bookkeeping on an append-only observability stream, not + // orchestration state; a restart can produce at most one duplicate + // eval_verdict event per evaluator, never an infinite loop, because the + // *structural* idempotency checks (ensureEvaluators/ensureArbitration + // below, and the story.Status=="VALIDATING" gate in + // finalizeArbitration) are what actually prevent duplicate task + // creation and duplicate story-status transitions — the two things that + // would matter if repeated forever. + handledVerdicts map[string]bool +} + +// DefaultStoryPollInterval is used by Run when pollInterval <= 0. +const DefaultStoryPollInterval = 15 * time.Second + +// Run polls all stories with a root_task_id set every pollInterval until ctx +// is cancelled. +func (o *StoryOrchestrator) Run(ctx context.Context, pollInterval time.Duration) { + if pollInterval <= 0 { + pollInterval = DefaultStoryPollInterval + } + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + o.Tick(ctx) + } + } +} + +// Tick runs a single poll pass over every story. Exported so tests can drive +// it directly without waiting on a ticker. +func (o *StoryOrchestrator) Tick(ctx context.Context) { + stories, err := o.Store.ListStories(storage.StoryFilter{}) + if err != nil { + o.logf("story orchestrator: list stories", "error", err) + return + } + for _, st := range stories { + if st.RootTaskID == "" { + continue // no execution tree yet — nothing for this orchestrator to do + } + if st.Status == "DONE" || st.Status == "CANCELLED" { + continue // terminal; this orchestrator never revives a story from here + } + o.processStory(ctx, st) + } +} + +func (o *StoryOrchestrator) logf(msg string, args ...any) { + if o.Logger != nil { + o.Logger.Warn(msg, args...) + } +} + +// processStory advances a single story by at most one stage per tick (it +// returns as soon as it finds a stage that isn't ready to progress yet — the +// next tick picks up where this one left off). +func (o *StoryOrchestrator) processStory(ctx context.Context, st *story.Story) { + root, err := o.Store.GetTask(st.RootTaskID) + if err != nil { + o.logf("story orchestrator: get root task", "storyID", st.ID, "rootTaskID", st.RootTaskID, "error", err) + return + } + root = o.autoAccept(st, root) + if root.State != task.StateCompleted { + return // Builder hasn't reached COMPLETED yet (still running, or not yet READY to auto-accept) + } + + // Stage 1: Builder -> Evaluators (+ story -> VALIDATING). + evaluators, ok := o.ensureEvaluators(ctx, st, root) + if !ok { + return // not all 4 could be found/created yet; retry next tick + } + + // Auto-accept each evaluator that's reached READY, emit per-evaluator + // verdicts, and check whether all 4 are done. + allDone := true + for i, ev := range evaluators { + ev = o.autoAccept(st, ev) + evaluators[i] = ev + o.maybeEmitVerdict(st, ev) + if ev.State != task.StateCompleted { + allDone = false + } + } + if !allDone { + return + } + + // Stage 2: Evaluators -> Arbitration. + arbitration, ok := o.ensureArbitration(ctx, st, evaluators) + if !ok { + return + } + arbitration = o.autoAccept(st, arbitration) + + // Stage 3: Arbitration -> REVIEW_READY. + if arbitration.State == task.StateCompleted { + o.finalizeArbitration(st, arbitration) + } +} + +// autoAccept transitions t from READY to COMPLETED if that's its current +// state, using the same state-machine-respecting write internal/api's +// acceptTask uses (Store.UpdateTaskState wraps storage.DB.UpdateTaskStateBy, +// which validates task.ValidTransition and writes the state_change event +// atomically — not a raw/unchecked write). Returns t unchanged if it wasn't +// READY (including if it's already COMPLETED, or still RUNNING/BLOCKED/etc.) +// or if the update failed. +// +// This is the mechanism that makes the story-level accept-gate +// (POST /api/stories/{id}/accept) the *only* human/chatbot interaction +// required to drive a story from a completed Builder run all the way to +// REVIEW_READY: without it, a human would have to separately call +// POST /api/tasks/{id}/accept on the Builder task, each of the 4 Evaluator +// tasks, and the Arbitration task, since none of those top-level tasks can +// reach COMPLETED any other way (see this file's package doc comment). +// Callers only ever pass tasks they've already established are part of a +// specific story's pipeline — the root task a story tracks via +// st.RootTaskID, or a role-matched dependent discovered by +// ensureEvaluators/ensureArbitration — so this never touches an unrelated +// READY task sitting outside any story's pipeline. +func (o *StoryOrchestrator) autoAccept(st *story.Story, t *task.Task) *task.Task { + if t.State != task.StateReady { + return t + } + if err := o.Store.UpdateTaskState(t.ID, task.StateCompleted); err != nil { + o.logf("story orchestrator: auto-accept", "storyID", st.ID, "taskID", t.ID, "error", err) + return t + } + accepted := *t + accepted.State = task.StateCompleted + return &accepted +} + +// ensureEvaluators returns the 4 Evaluator tasks fanned out from root, +// spawning any missing ones. Idempotency is structural, not a marker on the +// story: it looks at root's actual dependents and checks which of +// evaluatorRoles are already represented, so calling this repeatedly for the +// same story never spawns duplicates (test (b) in the phase description) — +// even across a process restart, unlike a purely in-memory guard would be. +// Returns ok=false if any missing evaluator couldn't be created this tick +// (transient store error); the caller retries on the next tick. +func (o *StoryOrchestrator) ensureEvaluators(ctx context.Context, st *story.Story, root *task.Task) ([]*task.Task, bool) { + dependents, err := o.Store.ListDependents(root.ID) + if err != nil { + o.logf("story orchestrator: list root dependents", "storyID", st.ID, "error", err) + return nil, false + } + found := make(map[string]*task.Task, len(evaluatorRoles)) + for _, d := range dependents { + for _, r := range evaluatorRoles { + if d.Agent.Role == r { + found[r] = d + } + } + } + + spawnedAny := false + for _, r := range evaluatorRoles { + if _, ok := found[r]; ok { + continue + } + nt, err := o.spawnRoleTask(ctx, fmt.Sprintf("%s: %s", r, st.Name), r, []string{root.ID}, root, + fmt.Sprintf("Evaluate the changes made by task %s against the %q dimension for story %q.\n\nStory spec:\n%s", root.ID, r, st.Name, st.Spec)) + if err != nil { + o.logf("story orchestrator: spawn evaluator", "storyID", st.ID, "role", r, "error", err) + continue + } + found[r] = nt + spawnedAny = true + } + + if spawnedAny { + st.Status = "VALIDATING" + if err := o.Store.UpdateStory(st); err != nil { + o.logf("story orchestrator: update story to VALIDATING", "storyID", st.ID, "error", err) + } + } + + if len(found) != len(evaluatorRoles) { + return nil, false + } + ordered := make([]*task.Task, len(evaluatorRoles)) + for i, r := range evaluatorRoles { + ordered[i] = found[r] + } + return ordered, true +} + +// ensureArbitration returns the single Arbitration task depending on all 4 +// evaluators, spawning it if it doesn't exist yet. Idempotency is again +// structural: it looks for an existing "planner"-role dependent of the first +// evaluator task that depends on every evaluator ID, rather than relying on +// story.Status (which a human can freely rewrite via PUT /api/stories/{id}). +func (o *StoryOrchestrator) ensureArbitration(ctx context.Context, st *story.Story, evaluators []*task.Task) (*task.Task, bool) { + ids := make([]string, len(evaluators)) + for i, ev := range evaluators { + ids[i] = ev.ID + } + + dependents, err := o.Store.ListDependents(evaluators[0].ID) + if err != nil { + o.logf("story orchestrator: list evaluator dependents", "storyID", st.ID, "error", err) + return nil, false + } + for _, d := range dependents { + if d.Agent.Role == arbitrationRole && dependsOnAll(d, ids) { + return d, true + } + } + + instructions := fmt.Sprintf( + "Arbitrate the 4 evaluator verdicts for story %q (task %s). Read each evaluator task's summary/events "+ + "and decide whether the story is ready to ship. Acceptance criteria:\n%s", + st.Name, st.ID, formatAcceptanceCriteria(st.AcceptanceCriteria)) + nt, err := o.spawnRoleTask(ctx, "Arbitration: "+st.Name, arbitrationRole, ids, evaluators[0], instructions) + if err != nil { + o.logf("story orchestrator: spawn arbitration", "storyID", st.ID, "error", err) + return nil, false + } + return nt, true +} + +// finalizeArbitration handles the Arbitration task reaching COMPLETED: it +// emits KindArbitrationDecided and moves the story to REVIEW_READY. +// +// Documented simplification (Phase 7b, see CLAUDE.md Design Debt): this does +// NOT parse the arbitration task's summary for an approve/reject verdict — +// it always routes to REVIEW_READY. A human or chatbot who reads the +// arbitration summary and disagrees can manually set the story to NEEDS_FIX +// via the existing PUT /api/stories/{id}. A later phase could close this gap +// by giving the arbitration task a dedicated verdict-reporting tool (e.g. a +// new AgentChannel method) whose structured output this orchestrator could +// trust instead of free-text parsing. +// +// Gated on st.Status == "VALIDATING" so repeated ticks (or a story a human +// already advanced past REVIEW_READY) don't re-emit the event or re-write the +// status — this is the one place in the orchestrator where the story's own +// status field, not a structural dependents check, is the idempotency guard, +// because by this stage there's nothing further to check structurally: the +// Arbitration task is the last task in the chain, so "does a subsequent task +// exist" isn't an available signal. +func (o *StoryOrchestrator) finalizeArbitration(st *story.Story, arbitration *task.Task) { + if st.Status != "VALIDATING" { + return + } + + payload, _ := json.Marshal(struct { + TaskID string `json:"task_id"` + Summary string `json:"summary"` + }{TaskID: arbitration.ID, Summary: arbitration.Summary}) + if err := o.Store.CreateEvent(&event.Event{ + TaskID: st.ID, + Kind: event.KindArbitrationDecided, + Actor: event.ActorSystem, + Payload: payload, + }); err != nil { + o.logf("story orchestrator: emit arbitration_decided", "storyID", st.ID, "error", err) + } + + st.Status = "REVIEW_READY" + if err := o.Store.UpdateStory(st); err != nil { + o.logf("story orchestrator: update story to REVIEW_READY", "storyID", st.ID, "error", err) + } +} + +// maybeEmitVerdict records a KindEvalVerdict event, attached to the story's +// ID (not the evaluator task's ID), the first time a given evaluator task is +// observed COMPLETED. Attaching to the story ID — the same choice +// finalizeArbitration makes for KindArbitrationDecided — means a single +// GET /api/stories/{id}/events call surfaces every verdict for a story, +// rather than requiring a client to separately fetch each evaluator task's +// own event stream and reassemble them; events.task_id has no enforced FK +// (see internal/event's doc comment), so this is exactly the tolerance the +// 7a phase already built in anticipation of this use. +func (o *StoryOrchestrator) maybeEmitVerdict(st *story.Story, ev *task.Task) { + if ev.State != task.StateCompleted { + return + } + if o.handledVerdicts == nil { + o.handledVerdicts = make(map[string]bool) + } + if o.handledVerdicts[ev.ID] { + return + } + o.handledVerdicts[ev.ID] = true + + payload, _ := json.Marshal(struct { + TaskID string `json:"task_id"` + Role string `json:"role"` + Summary string `json:"summary"` + }{TaskID: ev.ID, Role: ev.Agent.Role, Summary: ev.Summary}) + if err := o.Store.CreateEvent(&event.Event{ + TaskID: st.ID, + Kind: event.KindEvalVerdict, + Actor: event.ActorSystem, + Payload: payload, + }); err != nil { + o.logf("story orchestrator: emit eval_verdict", "storyID", st.ID, "taskID", ev.ID, "error", err) + } +} + +// spawnRoleTask creates a new role-typed, top-level task (no ParentTaskID — +// these are DAG siblings via DependsOn, not delegated subtasks; see +// internal/executor.Pool.cascadeFail's doc comment for why that distinction +// matters) and submits it to the pool. Agent.Type/Model are left empty so +// Phase 5's role-based dispatch resolves them from the role's escalation +// ladder on first dispatch (internal/executor.Pool.execute()); if no active +// role_configs row exists for the role, that same code path logs a warning +// and dispatches without role resolution — an accepted degraded mode we +// don't special-case here either. +func (o *StoryOrchestrator) spawnRoleTask(ctx context.Context, name, roleName string, dependsOn []string, template *task.Task, instructions string) (*task.Task, error) { + nt := &task.Task{ + ID: uuid.NewString(), + Name: name, + Project: template.Project, + RepositoryURL: template.RepositoryURL, + Agent: task.AgentConfig{ + Role: roleName, + Instructions: instructions, + }, + Priority: task.PriorityNormal, + Tags: []string{"story-orchestrator"}, + DependsOn: dependsOn, + Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "exponential"}, + State: task.StatePending, + } + if err := o.Store.CreateTask(nt); err != nil { + return nil, err + } + if err := o.Store.UpdateTaskState(nt.ID, task.StateQueued); err != nil { + return nil, err + } + nt.State = task.StateQueued + if err := o.Pool.Submit(ctx, nt); err != nil { + return nil, err + } + return nt, nil +} + +// dependsOnAll reports whether t.DependsOn contains every ID in ids. +func dependsOnAll(t *task.Task, ids []string) bool { + have := make(map[string]bool, len(t.DependsOn)) + for _, d := range t.DependsOn { + have[d] = true + } + for _, id := range ids { + if !have[id] { + return false + } + } + return true +} + +// formatAcceptanceCriteria renders a story's acceptance criteria as a +// markdown bullet list, or a placeholder line if there are none. +func formatAcceptanceCriteria(criteria []string) string { + if len(criteria) == 0 { + return "(none specified)" + } + out := "" + for _, c := range criteria { + out += "- " + c + "\n" + } + return out +} |