diff options
Diffstat (limited to 'internal/scheduler')
| -rw-r--r-- | internal/scheduler/story_orchestrator.go | 106 | ||||
| -rw-r--r-- | internal/scheduler/story_orchestrator_test.go | 255 |
2 files changed, 328 insertions, 33 deletions
diff --git a/internal/scheduler/story_orchestrator.go b/internal/scheduler/story_orchestrator.go index d190ba7..a526552 100644 --- a/internal/scheduler/story_orchestrator.go +++ b/internal/scheduler/story_orchestrator.go @@ -212,9 +212,21 @@ func (o *StoryOrchestrator) logf(msg string, args ...any) { } } -// 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). +// processStory advances every currently-READY builder-role node in a +// story's tree -- root or nested, any depth -- by up to one stage per tick +// (each node's own progression through Evaluators -> Arbitration -> +// approve/reject is handled by processBuilderNode, below). This no longer +// stops at the first node that isn't ready yet: a story's tree can have +// several builder nodes progressing independently and concurrently. In +// particular, a still-BLOCKED root can have a nested subtask that has +// already reached READY and needs its own arbitration before it can turn +// COMPLETED -- and internal/executor.Pool.maybeUnblockParent requires every +// subtask's task.CurrentAttempt to resolve to COMPLETED before promoting a +// BLOCKED parent. Stopping at root the way the single-root version did would +// deadlock any story with nested subtasks: root can never become READY +// while the very node that would produce that COMPLETED sits unprocessed, +// waiting for a tick that never reaches past root's own not-yet-READY +// state. func (o *StoryOrchestrator) processStory(ctx context.Context, st *story.Story) { if st.Status == "NEEDS_FIX" { o.ensureFixAttempt(ctx, st) @@ -226,20 +238,41 @@ func (o *StoryOrchestrator) processStory(ctx context.Context, st *story.Story) { o.logf("story orchestrator: resolve current root attempt", "storyID", st.ID, "rootTaskID", st.RootTaskID, "error", err) return } - // root is NOT auto-accepted here. Unlike evaluator/arbitration tasks, - // root is builder-role: its own completion must mean "verified by - // arbitration", not merely "the agent finished" -- so it stays READY - // through the whole Evaluators/Arbitration cycle below, and only - // finalizeArbitration's approval branch ever promotes it to COMPLETED. - // READY already satisfies a dependent's DependsOn (see - // executor.depDoneStates), so evaluators/arbitration can depend on a - // still-READY root without issue. - if root.State != task.StateReady { - return // not yet finished (still running/blocked); or already COMPLETED, meaning a prior tick's finalizeArbitration already approved it -- nothing further to do + if root.State == task.StateCompleted { + // The whole tree is already fully resolved: root only ever reaches + // COMPLETED via finalizeArbitration's approval branch, which -- + // per the maybeUnblockParent invariant above -- can't happen until + // every nested builder node beneath it is COMPLETED too. Nothing + // left to walk. + return } - // Stage 1: Builder -> Evaluators (+ story -> VALIDATING). - evaluators, ok := o.ensureEvaluators(ctx, st, root) + for _, node := range o.taskTree(root.ID) { + if node.Agent.Role != "builder" || node.State != task.StateReady { + continue + } + o.processBuilderNode(ctx, st, node) + } +} + +// processBuilderNode drives one READY builder-role node -- root or nested -- +// through Evaluators -> Arbitration -> approve/reject, advancing it by at +// most one stage per tick (the next tick picks up where this one left off). +// This is the exact per-root body processStory ran directly before this +// design recursed into nested subtask trees; extracting it into its own +// function changed nothing about what it does, only how many nodes call it +// per tick. +// +// node is NOT auto-accepted here. Unlike evaluator/arbitration tasks, node +// is builder-role: its own completion must mean "verified by arbitration", +// not merely "the agent finished" -- so it stays READY through the whole +// Evaluators/Arbitration cycle below, and only finalizeArbitration's +// approval branch ever promotes it to COMPLETED. READY already satisfies a +// dependent's DependsOn (see executor.depDoneStates), so evaluators/ +// arbitration can depend on a still-READY node without issue. +func (o *StoryOrchestrator) processBuilderNode(ctx context.Context, st *story.Story, node *task.Task) { + // Stage 1: Builder -> Evaluators. + evaluators, ok := o.ensureEvaluators(ctx, st, node) if !ok { return // not all 4 could be found/created yet; retry next tick } @@ -260,19 +293,20 @@ func (o *StoryOrchestrator) processStory(ctx context.Context, st *story.Story) { } // Stage 2: Evaluators -> Arbitration. - arbitration, ok := o.ensureArbitration(ctx, st, root, evaluators) + arbitration, ok := o.ensureArbitration(ctx, st, node, evaluators) if !ok { return } arbitration = o.autoAccept(st, arbitration) - // Stage 3: Arbitration -> approve (root: READY -> COMPLETED, story -> - // REVIEW_READY) or reject (story -> NEEDS_FIX; a fix attempt is spawned - // next tick by ensureFixAttempt, depending on the still-READY root -- - // task.CurrentAttempt resolves future lookups of this position forward - // to it). + // Stage 3: Arbitration -> approve (node: READY -> COMPLETED, and if node + // is the story's root, story -> REVIEW_READY) or reject (if node is the + // root, story -> NEEDS_FIX and a fix attempt is spawned next tick by + // ensureFixAttempt; otherwise finalizeArbitration spawns the nested + // fix-attempt directly). task.CurrentAttempt resolves future lookups of + // this position forward to whatever fix-attempt gets spawned. if arbitration.State == task.StateCompleted { - o.finalizeArbitration(ctx, st, root, arbitration) + o.finalizeArbitration(ctx, st, node, arbitration) } } @@ -309,18 +343,24 @@ func (o *StoryOrchestrator) autoAccept(st *story.Story, t *task.Task) *task.Task 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 +// ensureEvaluators returns the 4 Evaluator tasks fanned out from node (root +// or nested -- any builder-role task processStory's tree walk has found +// READY), spawning any missing ones. Idempotency is structural, not a marker +// on the story: it looks at node'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) — +// same node never spawns duplicates (test (b) in the phase description) — // even across a process restart, unlike a purely in-memory guard would be. +// story.Status is only ever set to VALIDATING here when node IS the story's +// root (node.ParentTaskID == "") -- mirroring finalizeArbitration's own +// isRoot gate on story.Status writes; a nested node's evaluators spawning +// must not touch story-level bookkeeping that only the root position has +// any claim to. // 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) +func (o *StoryOrchestrator) ensureEvaluators(ctx context.Context, st *story.Story, node *task.Task) ([]*task.Task, bool) { + dependents, err := o.Store.ListDependents(node.ID) if err != nil { - o.logf("story orchestrator: list root dependents", "storyID", st.ID, "error", err) + o.logf("story orchestrator: list node dependents", "storyID", st.ID, "error", err) return nil, false } found := make(map[string]*task.Task, len(evaluatorRoles)) @@ -332,7 +372,7 @@ func (o *StoryOrchestrator) ensureEvaluators(ctx context.Context, st *story.Stor } } - criteria := root.AcceptanceCriteria + criteria := node.AcceptanceCriteria if len(criteria) == 0 { criteria = st.AcceptanceCriteria } @@ -342,8 +382,8 @@ func (o *StoryOrchestrator) ensureEvaluators(ctx context.Context, st *story.Stor 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\n\nAcceptance criteria:\n%s", root.ID, r, st.Name, st.Spec, formatAcceptanceCriteria(criteria))) + nt, err := o.spawnRoleTask(ctx, fmt.Sprintf("%s: %s", r, st.Name), r, []string{node.ID}, "", node, + fmt.Sprintf("Evaluate the changes made by task %s against the %q dimension for story %q.\n\nStory spec:\n%s\n\nAcceptance criteria:\n%s", node.ID, r, st.Name, st.Spec, formatAcceptanceCriteria(criteria))) if err != nil { o.logf("story orchestrator: spawn evaluator", "storyID", st.ID, "role", r, "error", err) continue @@ -352,7 +392,7 @@ func (o *StoryOrchestrator) ensureEvaluators(ctx context.Context, st *story.Stor spawnedAny = true } - if spawnedAny { + if spawnedAny && node.ParentTaskID == "" { st.Status = "VALIDATING" if err := o.Store.UpdateStory(st); err != nil { o.logf("story orchestrator: update story to VALIDATING", "storyID", st.ID, "error", err) diff --git a/internal/scheduler/story_orchestrator_test.go b/internal/scheduler/story_orchestrator_test.go index 03d4437..aa7fff1 100644 --- a/internal/scheduler/story_orchestrator_test.go +++ b/internal/scheduler/story_orchestrator_test.go @@ -1672,3 +1672,258 @@ func TestStoryOrchestrator_Retro_EmitsRetroCaptured_WhenRetroTaskCompletes(t *te t.Fatalf("expected still exactly 1 retro_captured event after repeated ticks, got %d", len(captured)) } } + +// TestStoryOrchestrator_ProcessesNestedReadyBuilderNode_WhileRootStillBlocked +// proves the deadlock the single-root version of processStory had: a nested +// builder subtask reaching READY while its parent (root) is still BLOCKED +// must still get its own Evaluators spawned this tick -- root can only ever +// become READY once this nested node's own arbitration approves it (see +// internal/executor.Pool.maybeUnblockParent, which requires every +// subtask's task.CurrentAttempt to resolve to COMPLETED before promoting a +// BLOCKED parent). Root itself, still BLOCKED, gets nothing spawned for it. +func TestStoryOrchestrator_ProcessesNestedReadyBuilderNode_WhileRootStillBlocked(t *testing.T) { + store := newFakeStoryStore() + root := builderTask("deadlock-root", task.StateBlocked) + store.tasks[root.ID] = root + st := newStoryWithRoot("deadlock-story", root.ID, "IN_PROGRESS") + store.stories[st.ID] = st + + nested := builderTask("deadlock-nested", task.StateReady) + nested.ParentTaskID = root.ID + store.tasks[nested.ID] = nested + + pool := &fakePool{} + orch := &StoryOrchestrator{Store: store, Pool: pool} + orch.Tick(context.Background()) + + deps, err := store.ListDependents(nested.ID) + if err != nil { + t.Fatal(err) + } + if len(deps) != 4 { + t.Fatalf("expected 4 evaluator tasks spawned for the nested READY node even though root is still BLOCKED, got %d", len(deps)) + } + + rootDeps, _ := store.ListDependents(root.ID) + if len(rootDeps) != 0 { + t.Fatalf("root itself must not get evaluators spawned while still BLOCKED, got %d dependents", len(rootDeps)) + } + rootAfter, err := store.GetTask(root.ID) + if err != nil { + t.Fatal(err) + } + if rootAfter.State != task.StateBlocked { + t.Errorf("root state must be untouched: want BLOCKED, got %v", rootAfter.State) + } + + stories, _ := store.ListStories(storage.StoryFilter{}) + var gotStory *story.Story + for _, s := range stories { + if s.ID == st.ID { + gotStory = s + } + } + if gotStory.Status != "IN_PROGRESS" { + t.Errorf("story Status = %q, want unchanged IN_PROGRESS (nested evaluator spawn must not touch story-level status)", gotStory.Status) + } +} + +// TestStoryOrchestrator_ProcessStory_MultipleSiblingNestedNodes_BothProcessedSameTick +// proves the tree walk processes every qualifying node in a single tick, not +// just the first one it finds -- a real behavioral change from the old +// single-root processStory, which returned as soon as it found one node not +// ready to progress. +func TestStoryOrchestrator_ProcessStory_MultipleSiblingNestedNodes_BothProcessedSameTick(t *testing.T) { + store := newFakeStoryStore() + root := builderTask("multi-root", task.StateBlocked) + store.tasks[root.ID] = root + st := newStoryWithRoot("multi-story", root.ID, "IN_PROGRESS") + store.stories[st.ID] = st + + nested1 := builderTask("multi-nested-1", task.StateReady) + nested1.ParentTaskID = root.ID + store.tasks[nested1.ID] = nested1 + + nested2 := builderTask("multi-nested-2", task.StateReady) + nested2.ParentTaskID = root.ID + store.tasks[nested2.ID] = nested2 + + pool := &fakePool{} + orch := &StoryOrchestrator{Store: store, Pool: pool} + orch.Tick(context.Background()) + + deps1, _ := store.ListDependents(nested1.ID) + deps2, _ := store.ListDependents(nested2.ID) + if len(deps1) != 4 { + t.Errorf("nested1: expected 4 evaluator tasks, got %d", len(deps1)) + } + if len(deps2) != 4 { + t.Errorf("nested2: expected 4 evaluator tasks, got %d", len(deps2)) + } + if pool.submitCount() != 8 { + t.Fatalf("expected 8 pool submissions (4 evaluators x 2 sibling nodes) in a single tick, got %d", pool.submitCount()) + } +} + +// TestStoryOrchestrator_ProcessStory_RootCompleted_NoOp proves the +// short-circuit added for a fully-resolved tree: once root itself is +// COMPLETED (only possible after finalizeArbitration approved it, which +// itself requires every nested node beneath it to already be COMPLETED -- +// see maybeUnblockParent), processStory does nothing further, not even a +// tree walk. +func TestStoryOrchestrator_ProcessStory_RootCompleted_NoOp(t *testing.T) { + store := newFakeStoryStore() + root := builderTask("done-root", task.StateCompleted) + store.tasks[root.ID] = root + st := newStoryWithRoot("done-story", root.ID, "REVIEW_READY") + store.stories[st.ID] = st + + pool := &fakePool{} + orch := &StoryOrchestrator{Store: store, Pool: pool} + orch.Tick(context.Background()) + + if pool.submitCount() != 0 { + t.Fatalf("expected no submissions once root is COMPLETED, got %d", pool.submitCount()) + } +} + +// TestStoryOrchestrator_EndToEnd_NestedSubtask exercises the full recursive +// loop this piece adds: a nested builder subtask reaches READY while root is +// still BLOCKED, gets its own Evaluators -> Arbitration -> approval cycle +// entirely independent of root, reaches COMPLETED -- at which point (in a +// real deployment) internal/executor.Pool.maybeUnblockParent would see every +// one of root's subtasks resolve to COMPLETED via task.CurrentAttempt and +// promote root BLOCKED -> READY. That promotion is simulated directly here +// (store.setTaskState) since maybeUnblockParent lives in a different +// package/component this fake doesn't exercise. Root then goes through the +// exact same cycle for itself, ending at COMPLETED with the story at +// REVIEW_READY -- proving the two levels compose without any special-casing +// between them. +func TestStoryOrchestrator_EndToEnd_NestedSubtask(t *testing.T) { + store := newFakeStoryStore() + root := builderTask("e2e-nested-root", task.StateBlocked) + store.tasks[root.ID] = root + st := newStoryWithRoot("e2e-nested-story", root.ID, "IN_PROGRESS") + store.stories[st.ID] = st + + nested := builderTask("e2e-nested-child", task.StateReady) + nested.ParentTaskID = root.ID + store.tasks[nested.ID] = nested + + pool := &fakePool{} + orch := &StoryOrchestrator{Store: store, Pool: pool} + + // Tick 1: nested is READY -> its 4 evaluators spawn. Root, still + // BLOCKED, gets nothing. + orch.Tick(context.Background()) + nestedDeps, _ := store.ListDependents(nested.ID) + if len(nestedDeps) != 4 { + t.Fatalf("expected 4 evaluators for nested, got %d", len(nestedDeps)) + } + + // Evaluators' executions succeed one by one (READY, not COMPLETED). + for i, d := range nestedDeps { + store.setTaskState(d.ID, task.StateReady) + store.setTaskSummary(d.ID, fmt.Sprintf("nested verdict %d", i)) + orch.Tick(context.Background()) + } + for _, d := range nestedDeps { + got, err := store.GetTask(d.ID) + if err != nil { + t.Fatal(err) + } + if got.State != task.StateCompleted { + t.Fatalf("nested evaluator %s should be auto-accepted to COMPLETED, got %v", d.ID, got.State) + } + } + + nestedArbs := store.dependentsWithRole(nestedDeps[0].ID, "planner") + if len(nestedArbs) != 1 { + t.Fatalf("expected exactly 1 arbitration task for nested, got %d", len(nestedArbs)) + } + nestedArb := nestedArbs[0] + + // Arbitration approves nested. + store.setTaskState(nestedArb.ID, task.StateReady) + store.setTaskSummary(nestedArb.ID, "approved") + payload, _ := json.Marshal(struct { + Approved bool `json:"approved"` + Reasoning string `json:"reasoning"` + }{Approved: true, Reasoning: "meets criteria"}) + if err := store.CreateEvent(&event.Event{TaskID: nestedArb.ID, Kind: event.KindVerdictReported, Actor: event.ActorAgent, Payload: payload}); err != nil { + t.Fatalf("seed nested verdict event: %v", err) + } + orch.Tick(context.Background()) + + nestedAfter, err := store.GetTask(nested.ID) + if err != nil { + t.Fatal(err) + } + if nestedAfter.State != task.StateCompleted { + t.Fatalf("nested should be promoted to COMPLETED now that its arbitration approved it, got %v", nestedAfter.State) + } + + // Story status must still be untouched by the nested node's own + // approval -- only root's own approval may set REVIEW_READY. + statusOf := func() string { + stories, _ := store.ListStories(storage.StoryFilter{}) + for _, s := range stories { + if s.ID == st.ID { + return s.Status + } + } + return "" + } + if statusOf() != "IN_PROGRESS" { + t.Fatalf("story status must remain IN_PROGRESS after only the nested node's approval, got %q", statusOf()) + } + + // Simulate what internal/executor.Pool.maybeUnblockParent would do in a + // real deployment now that nested's task.CurrentAttempt resolves to + // COMPLETED: promote root BLOCKED -> READY. + store.setTaskState(root.ID, task.StateReady) + + // Tick: root is now READY -> its own 4 evaluators spawn. + orch.Tick(context.Background()) + rootDeps, _ := store.ListDependents(root.ID) + if len(rootDeps) != 4 { + t.Fatalf("expected 4 evaluators for root, got %d", len(rootDeps)) + } + if statusOf() != "VALIDATING" { + t.Fatalf("expected VALIDATING once root's own evaluators spawn, got %q", statusOf()) + } + + for i, d := range rootDeps { + store.setTaskState(d.ID, task.StateReady) + store.setTaskSummary(d.ID, fmt.Sprintf("root verdict %d", i)) + orch.Tick(context.Background()) + } + + rootArbs := store.dependentsWithRole(rootDeps[0].ID, "planner") + if len(rootArbs) != 1 { + t.Fatalf("expected exactly 1 arbitration task for root, got %d", len(rootArbs)) + } + rootArb := rootArbs[0] + + store.setTaskState(rootArb.ID, task.StateReady) + store.setTaskSummary(rootArb.ID, "approved") + rootPayload, _ := json.Marshal(struct { + Approved bool `json:"approved"` + Reasoning string `json:"reasoning"` + }{Approved: true, Reasoning: "meets criteria"}) + if err := store.CreateEvent(&event.Event{TaskID: rootArb.ID, Kind: event.KindVerdictReported, Actor: event.ActorAgent, Payload: rootPayload}); err != nil { + t.Fatalf("seed root verdict event: %v", err) + } + orch.Tick(context.Background()) + + rootAfter, err := store.GetTask(root.ID) + if err != nil { + t.Fatal(err) + } + if rootAfter.State != task.StateCompleted { + t.Fatalf("root should be promoted to COMPLETED now that its own arbitration approved it, got %v", rootAfter.State) + } + if statusOf() != "REVIEW_READY" { + t.Fatalf("expected REVIEW_READY once root's own arbitration approves it, got %q", statusOf()) + } +} |