fix(executor): support nested subtask decomposition (subtask-with-own-subtasks correctly blocks, cascades, and recovers)

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 &copy
 }
 
-// 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.