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

1 files changed, 53 insertions, 24 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)
 	}