diff options
| -rw-r--r-- | internal/executor/executor.go | 77 | ||||
| -rw-r--r-- | internal/executor/executor_test.go | 138 | ||||
| -rw-r--r-- | internal/task/task.go | 2 |
3 files changed, 192 insertions, 25 deletions
diff --git a/internal/executor/executor.go b/internal/executor/executor.go index e2eb7ce..2505280 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -512,21 +512,24 @@ func (p *Pool) handleRunResult(ctx context.Context, t *task.Task, exec *storage. p.mu.Lock() p.consecutiveFailures[agentType] = 0 p.mu.Unlock() - if t.ParentTaskID == "" { - subtasks, subErr := p.store.ListSubtasks(t.ID) - if subErr != nil { - p.logger.Error("failed to list subtasks", "taskID", t.ID, "error", subErr) + // Check for pending subtasks uniformly, regardless of whether t is + // itself top-level or a subtask -- a subtask that decomposed further + // (nested spawn_subtask) must block on its own children exactly like + // a top-level task does; only the READY-vs-COMPLETED choice below + // depends on whether t has its own ParentTaskID. + subtasks, subErr := p.store.ListSubtasks(t.ID) + if subErr != nil { + p.logger.Error("failed to list subtasks", "taskID", t.ID, "error", subErr) + } + if subErr == nil && len(subtasks) > 0 { + exec.Status = "BLOCKED" + if err := p.store.UpdateTaskState(t.ID, task.StateBlocked); err != nil { + p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateBlocked, "error", err) } - if subErr == nil && len(subtasks) > 0 { - exec.Status = "BLOCKED" - if err := p.store.UpdateTaskState(t.ID, task.StateBlocked); err != nil { - p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateBlocked, "error", err) - } - } else { - exec.Status = "READY" - if err := p.store.UpdateTaskState(t.ID, task.StateReady); err != nil { - p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateReady, "error", err) - } + } else if t.ParentTaskID == "" { + exec.Status = "READY" + if err := p.store.UpdateTaskState(t.ID, task.StateReady); err != nil { + p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateReady, "error", err) } } else { exec.Status = "COMPLETED" @@ -1121,10 +1124,16 @@ func (p *Pool) RecoverStaleQueued(ctx context.Context) { } } -// RecoverStaleBlocked promotes any BLOCKED or QUEUED parent task to READY when -// all of its subtasks are already COMPLETED. This handles the case where the -// server was restarted after subtasks finished but before maybeUnblockParent -// could fire. +// RecoverStaleBlocked promotes any BLOCKED or QUEUED task to READY (or, for a +// task that is itself a subtask, straight to COMPLETED — see +// maybeUnblockParent) when all of its own subtasks are already COMPLETED. +// This handles the case where the server was restarted after subtasks +// finished but before maybeUnblockParent could fire. Iterates every +// BLOCKED/QUEUED task, not just top-level ones — a subtask that itself +// decomposed (nested spawn_subtask) needs the exact same recovery +// maybeUnblockParent already provides; maybeUnblockParent's own +// "len(subtasks) == 0" guard is what correctly no-ops for a task that isn't +// actually a parent of anything, so no separate filter is needed here. // Call this once on server startup, after RecoverStaleRunning and RecoverStaleQueued. func (p *Pool) RecoverStaleBlocked() { for _, state := range []task.State{task.StateBlocked, task.StateQueued} { @@ -1134,9 +1143,6 @@ func (p *Pool) RecoverStaleBlocked() { continue } for _, t := range tasks { - if t.ParentTaskID != "" { - continue // only promote actual parents - } p.maybeUnblockParent(t.ID) } } @@ -1213,9 +1219,24 @@ func withFailureHistory(t *task.Task, execs []*storage.Execution, err error) *ta return © } -// maybeUnblockParent transitions the parent task to READY if all of its subtasks -// are in the COMPLETED state. Handles both BLOCKED parents (ran, created subtasks, -// paused) and QUEUED parents (subtasks created before parent ran). +// maybeUnblockParent transitions the parent task once all of its subtasks +// are in the COMPLETED state: to READY if parentID is itself top-level +// (awaiting a human accept, same as ever), or straight to COMPLETED if +// parentID is itself a subtask (ParentTaskID set) -- mirroring +// handleRunResult's own subtask-completion branch, since a subtask never +// needs a human accept regardless of whether it got there by finishing its +// own direct work or by having all of its delegated children finish. That +// COMPLETED transition recurses into maybeUnblockParent(parent.ParentTaskID) +// so a chain of nested decomposition (subtask spawns subtask spawns +// subtask...) fully cascades to completion when the deepest leaves finish, +// not just one level. +// +// Handles both BLOCKED parents (ran, created subtasks, paused) and QUEUED +// parents (subtasks created before parent ran) for the READY path only — +// the COMPLETED-instead-of-READY branch only applies to BLOCKED parents, +// since a subtask can only have already-completed children after it has +// itself run (BLOCKED), never while still QUEUED, and task.ValidTransition +// doesn't permit QUEUED → COMPLETED regardless. func (p *Pool) maybeUnblockParent(parentID string) { parent, err := p.store.GetTask(parentID) if err != nil { @@ -1240,6 +1261,14 @@ func (p *Pool) maybeUnblockParent(parentID string) { return } } + if parent.State == task.StateBlocked && parent.ParentTaskID != "" { + if err := p.store.UpdateTaskState(parentID, task.StateCompleted); err != nil { + p.logger.Error("maybeUnblockParent: update parent state", "parentID", parentID, "error", err) + return + } + p.maybeUnblockParent(parent.ParentTaskID) + return + } if err := p.store.UpdateTaskState(parentID, task.StateReady); err != nil { p.logger.Error("maybeUnblockParent: update parent state", "parentID", parentID, "error", err) } diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go index 12d052c..d89a9eb 100644 --- a/internal/executor/executor_test.go +++ b/internal/executor/executor_test.go @@ -1031,6 +1031,144 @@ func TestPool_Submit_NotLastSubtask_ParentStaysBlocked(t *testing.T) { } } +// TestPool_Submit_SubtaskWithOwnSubtasks_GoesBlocked proves the fix for a +// real pre-existing bug: a subtask (ParentTaskID != "") that itself spawns +// subtasks must go BLOCKED when its own agent turn ends, not COMPLETED -- +// mirroring exactly how a top-level task with subtasks already behaves +// (TestPool_Submit_TopLevel_WithSubtasks_GoesBlocked). Before this fix, +// handleRunResult only checked for pending subtasks when ParentTaskID == "", +// so a decomposing subtask was marked COMPLETED the instant its own turn +// ended, ignoring the children it just spawned. +func TestPool_Submit_SubtaskWithOwnSubtasks_GoesBlocked(t *testing.T) { + store := testStore(t) + runner := &mockRunner{} + runners := map[string]Runner{"claude": runner} + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := NewPool(2, runners, store, logger) + + middle := makeTask("middle-with-own-subtasks") + middle.ParentTaskID = "grandparent-1" // middle is itself a subtask + store.CreateTask(middle) + + // middle spawned its own child, but that child hasn't been submitted yet. + grandchild := makeTask("grandchild-of-middle") + grandchild.ParentTaskID = middle.ID + store.CreateTask(grandchild) + + if err := pool.Submit(context.Background(), middle); err != nil { + t.Fatalf("submit: %v", err) + } + + result := <-pool.Results() + if result.Err != nil { + t.Errorf("expected no error, got: %v", result.Err) + } + if result.Execution.Status != "BLOCKED" { + t.Errorf("status: want BLOCKED, got %q", result.Execution.Status) + } + got, _ := store.GetTask(middle.ID) + if got.State != task.StateBlocked { + t.Errorf("task state: want BLOCKED, got %v (this is the bug: a decomposing subtask must not skip straight to COMPLETED)", got.State) + } +} + +// TestPool_Submit_GrandchildCompletion_CascadesThroughNestedParents proves +// the full 3-level recursive propagation: when the deepest leaf completes, +// its immediate parent (itself a subtask) goes straight to COMPLETED -- +// not READY, since subtasks never need a human accept -- which then +// recursively unblocks the true top-level root to READY. +func TestPool_Submit_GrandchildCompletion_CascadesThroughNestedParents(t *testing.T) { + store := testStore(t) + runner := &mockRunner{} + runners := map[string]Runner{"claude": runner} + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := NewPool(2, runners, store, logger) + + root := makeTask("nested-root") + root.State = task.StateBlocked // already ran, delegated to middle + store.CreateTask(root) + + middle := makeTask("nested-middle") + middle.ParentTaskID = root.ID + middle.State = task.StateBlocked // already ran, delegated to grandchild + store.CreateTask(middle) + + grandchild := makeTask("nested-grandchild") + grandchild.ParentTaskID = middle.ID + store.CreateTask(grandchild) // fresh, about to be submitted + + if err := pool.Submit(context.Background(), grandchild); err != nil { + t.Fatalf("submit: %v", err) + } + + result := <-pool.Results() + if result.Err != nil { + t.Errorf("expected no error, got: %v", result.Err) + } + if result.Execution.Status != "COMPLETED" { + t.Errorf("grandchild status: want COMPLETED, got %q", result.Execution.Status) + } + + gotMiddle, err := store.GetTask(middle.ID) + if err != nil { + t.Fatalf("get middle: %v", err) + } + if gotMiddle.State != task.StateCompleted { + t.Errorf("middle state: want COMPLETED (it's a subtask, no accept needed), got %v", gotMiddle.State) + } + + gotRoot, err := store.GetTask(root.ID) + if err != nil { + t.Fatalf("get root: %v", err) + } + if gotRoot.State != task.StateReady { + t.Errorf("root state: want READY (top-level, awaiting accept), got %v", gotRoot.State) + } +} + +// TestPool_RecoverStaleBlocked_PromotesNestedSubtaskParent proves +// RecoverStaleBlocked no longer skips BLOCKED subtask-parents -- before +// this fix, its "only promote actual parents" filter (ParentTaskID == "") +// meant a restarted server could never recover a BLOCKED subtask-parent +// once its own children finished, leaving it stuck forever. +func TestPool_RecoverStaleBlocked_PromotesNestedSubtaskParent(t *testing.T) { + store := testStore(t) + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := NewPool(2, map[string]Runner{"claude": &mockRunner{}}, store, logger) + + root := makeTask("recover-nested-root") + root.State = task.StateBlocked + store.CreateTask(root) + + middle := makeTask("recover-nested-middle") + middle.ParentTaskID = root.ID + middle.State = task.StateBlocked + store.CreateTask(middle) + + grandchild := makeTask("recover-nested-grandchild") + grandchild.ParentTaskID = middle.ID + grandchild.State = task.StateCompleted + store.CreateTask(grandchild) + + pool.RecoverStaleBlocked() + + gotMiddle, err := store.GetTask(middle.ID) + if err != nil { + t.Fatalf("get middle: %v", err) + } + if gotMiddle.State != task.StateCompleted { + t.Errorf("middle state: want COMPLETED, got %v", gotMiddle.State) + } + + gotRoot, err := store.GetTask(root.ID) + if err != nil { + t.Fatalf("get root: %v", err) + } + if gotRoot.State != task.StateReady { + t.Errorf("root state: want READY, got %v", gotRoot.State) + } +} + // TestPool_Submit_ParentNotBlocked_NoTransition verifies that completing a subtask // does not change the parent's state when the parent is not BLOCKED. func TestPool_Submit_ParentNotBlocked_NoTransition(t *testing.T) { diff --git a/internal/task/task.go b/internal/task/task.go index 35da9a0..a95107a 100644 --- a/internal/task/task.go +++ b/internal/task/task.go @@ -162,7 +162,7 @@ var validTransitions = map[State]map[State]bool{ StateTimedOut: {StateQueued: true}, // retry or resume StateCancelled: {StateQueued: true}, // restart StateBudgetExceeded: {StateQueued: true}, // retry - StateBlocked: {StateQueued: true, StateReady: true, StateCancelled: true}, + StateBlocked: {StateQueued: true, StateReady: true, StateCompleted: true, StateCancelled: true}, } // ValidTransition returns true if moving from the current state to next is allowed. |