feat(executor): add DAG auto-cascade-fail + role-typed subtask spawning (Phase 6)

Two prerequisites for safe parallel evaluator fan-out (later phase):

1. Auto-cascade-fail: previously, a failed task's dependents just sat
   PENDING/QUEUED forever (or until something eventually tried to dispatch
   them and discovered the dependency was dead). Pool.cascadeFail now fires
   right after a task lands in a terminal failure state (FAILED/TIMED_OUT/
   CANCELLED/BUDGET_EXCEEDED, from handleRunResult, the budget-gate reject
   path, and the checkDepsReady dependency-failure path), recursively
   cancelling every not-yet-run dependent (PENDING/QUEUED only -- RUNNING and
   terminal states are left alone) with a message referencing the upstream
   failure. A visited-set guards recursion, which turned out to be
   load-bearing rather than defense-in-depth: task creation does not prevent
   dependency cycles anywhere in this codebase.

   Correction to an earlier assumption: internal/executor's
   waitForDependencies is dead code, never called. The live mechanism is
   checkDepsReady, invoked synchronously in execute() with a self-requeue via
   time.AfterFunc. Added a freshness re-check (GetTask, bail if no longer
   QUEUED) at the top of that block so a task cascade-cancelled while sitting
   in the requeue loop stops silently instead of hitting an invalid
   CANCELLED->CANCELLED transition or, worse, still getting dispatched.

2. storeChannel.SpawnSubtask hardcoded Agent.Type: "claude" on every spawned
   child regardless of what role it should play -- a hard blocker for a
   Planner/Builder task spawning role-typed evaluator subtasks. SubtaskSpec
   (internal/agentchannel) gains a Role field; when set, the child task gets
   Agent.Role instead of a hardcoded Type, so Phase 5's role-resolution picks
   provider/model from that role's escalation ladder. spec.Role == "" (every
   existing caller) preserves today's exact behavior byte-for-byte -- proven
   by an explicit regression test, not just new-feature coverage. Threaded
   the new `role` parameter through both spawn_subtask transports: the native
   tool-use loop (internal/agentloop/tools.go) and the MCP tool exposed to
   ContainerRunner-driven claude/gemini agents (internal/executor/agentmcp.go).

go build/vet/test -race -count=1 all pass, full suite (20 packages).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V1moSNCJRcP6kykA4tyUSs

1 files changed, 222 insertions, 0 deletions

diff --git a/internal/executor/cascade_test.go b/internal/executor/cascade_test.go
new file mode 100644
index 0000000..5e5f923
--- /dev/null
+++ b/internal/executor/cascade_test.go
@@ -0,0 +1,222 @@
+package executor
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/thepeterstone/claudomator/internal/storage"
+	"github.com/thepeterstone/claudomator/internal/task"
+)
+
+// depTask returns a task with the given DependsOn and state, using the same
+// shape makeTask() produces (see executor_test.go) so it satisfies
+// task.Validate-adjacent expectations (non-nil Tags/DependsOn slices etc).
+func depTask(id string, dependsOn []string, state task.State) *task.Task {
+	tk := makeTask(id)
+	if dependsOn == nil {
+		dependsOn = []string{}
+	}
+	tk.DependsOn = dependsOn
+	tk.State = state
+	return tk
+}
+
+func cascadeTestPool(t *testing.T, store Store) *Pool {
+	t.Helper()
+	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))
+	return NewPool(2, map[string]Runner{"claude": &mockRunner{}}, store, logger)
+}
+
+// TestPool_CascadeFail_SingleLevel: A fails; B depends on A and is PENDING.
+// cascadeFail must mark B CANCELLED with a message referencing A.
+func TestPool_CascadeFail_SingleLevel(t *testing.T) {
+	store := testStore(t)
+	a := depTask("cf-a", nil, task.StateFailed)
+	b := depTask("cf-b", []string{"cf-a"}, task.StatePending)
+	if err := store.CreateTask(a); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.CreateTask(b); err != nil {
+		t.Fatal(err)
+	}
+
+	pool := cascadeTestPool(t, store)
+	pool.cascadeFail("cf-a", "build failed")
+
+	got, err := store.GetTask("cf-b")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.State != task.StateCancelled {
+		t.Fatalf("cf-b state: want CANCELLED, got %s", got.State)
+	}
+
+	execs, err := store.ListExecutions("cf-b")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(execs) != 1 {
+		t.Fatalf("expected 1 execution recorded for cf-b, got %d", len(execs))
+	}
+	if execs[0].Status != "CANCELLED" {
+		t.Errorf("execution status: want CANCELLED, got %q", execs[0].Status)
+	}
+	if !strings.Contains(execs[0].ErrorMsg, "cf-a") {
+		t.Errorf("execution ErrorMsg should reference upstream task cf-a, got %q", execs[0].ErrorMsg)
+	}
+}
+
+// TestPool_CascadeFail_MultiLevelRecursion: A fails, B depends on A, C
+// depends on B. Both B and C must end up CANCELLED — cascadeFail must
+// recurse through the whole subtree, not just cancel direct dependents.
+func TestPool_CascadeFail_MultiLevelRecursion(t *testing.T) {
+	store := testStore(t)
+	a := depTask("cfm-a", nil, task.StateFailed)
+	b := depTask("cfm-b", []string{"cfm-a"}, task.StatePending)
+	c := depTask("cfm-c", []string{"cfm-b"}, task.StatePending)
+	for _, tk := range []*task.Task{a, b, c} {
+		if err := store.CreateTask(tk); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	pool := cascadeTestPool(t, store)
+	pool.cascadeFail("cfm-a", "build failed")
+
+	gotB, err := store.GetTask("cfm-b")
+	if err != nil {
+		t.Fatal(err)
+	}
+	gotC, err := store.GetTask("cfm-c")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if gotB.State != task.StateCancelled {
+		t.Errorf("cfm-b state: want CANCELLED, got %s", gotB.State)
+	}
+	if gotC.State != task.StateCancelled {
+		t.Errorf("cfm-c state: want CANCELLED, got %s", gotC.State)
+	}
+}
+
+// TestPool_CascadeFail_LeavesRunningAndCompletedAlone: dependents that are
+// already RUNNING or already terminal (COMPLETED) when the cascade fires
+// must be left untouched, not force-cancelled.
+func TestPool_CascadeFail_LeavesRunningAndCompletedAlone(t *testing.T) {
+	store := testStore(t)
+	a := depTask("cfr-a", nil, task.StateFailed)
+	running := depTask("cfr-running", []string{"cfr-a"}, task.StateRunning)
+	completed := depTask("cfr-completed", []string{"cfr-a"}, task.StateCompleted)
+	for _, tk := range []*task.Task{a, running, completed} {
+		if err := store.CreateTask(tk); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	pool := cascadeTestPool(t, store)
+	pool.cascadeFail("cfr-a", "build failed")
+
+	gotRunning, err := store.GetTask("cfr-running")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if gotRunning.State != task.StateRunning {
+		t.Errorf("cfr-running state: want unchanged RUNNING, got %s", gotRunning.State)
+	}
+
+	gotCompleted, err := store.GetTask("cfr-completed")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if gotCompleted.State != task.StateCompleted {
+		t.Errorf("cfr-completed state: want unchanged COMPLETED, got %s", gotCompleted.State)
+	}
+
+	// No execution records should have been created for either — cascadeFail
+	// should not have touched them at all.
+	if execs, _ := store.ListExecutions("cfr-running"); len(execs) != 0 {
+		t.Errorf("expected no executions recorded for cfr-running, got %d", len(execs))
+	}
+	if execs, _ := store.ListExecutions("cfr-completed"); len(execs) != 0 {
+		t.Errorf("expected no executions recorded for cfr-completed, got %d", len(execs))
+	}
+}
+
+// TestPool_CascadeFail_CycleGuard proves cascadeFail terminates on a
+// dependency cycle instead of recursing forever. task.Validate and
+// storage.CreateTask do not reject dependency cycles today (there is no
+// cycle-detection anywhere in task creation), so this guard is load-bearing,
+// not just defense-in-depth.
+func TestPool_CascadeFail_CycleGuard(t *testing.T) {
+	store := testStore(t)
+	a := depTask("cyc-a", []string{"cyc-b"}, task.StatePending)
+	b := depTask("cyc-b", []string{"cyc-a"}, task.StatePending)
+	if err := store.CreateTask(a); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.CreateTask(b); err != nil {
+		t.Fatal(err)
+	}
+
+	pool := cascadeTestPool(t, store)
+
+	done := make(chan struct{})
+	go func() {
+		pool.cascadeFail("cyc-a", "boom")
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("cascadeFail did not terminate on a dependency cycle — infinite recursion?")
+	}
+
+	gotA, err := store.GetTask("cyc-a")
+	if err != nil {
+		t.Fatal(err)
+	}
+	gotB, err := store.GetTask("cyc-b")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if gotA.State != task.StateCancelled {
+		t.Errorf("cyc-a state: want CANCELLED, got %s", gotA.State)
+	}
+	if gotB.State != task.StateCancelled {
+		t.Errorf("cyc-b state: want CANCELLED, got %s", gotB.State)
+	}
+}
+
+// TestPool_CascadeFail_WiredIntoHandleRunResult verifies cascadeFail is
+// actually invoked when a task's run terminates in a failure state via the
+// normal handleRunResult path (not just when called directly), using a real
+// store so ListDependents sees real data.
+func TestPool_CascadeFail_WiredIntoHandleRunResult(t *testing.T) {
+	store := testStore(t)
+	a := depTask("hrr-cascade-a", nil, task.StateRunning)
+	b := depTask("hrr-cascade-b", []string{"hrr-cascade-a"}, task.StatePending)
+	if err := store.CreateTask(a); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.CreateTask(b); err != nil {
+		t.Fatal(err)
+	}
+
+	pool := cascadeTestPool(t, store)
+	execRec := &storage.Execution{ID: "hrr-cascade-exec", TaskID: a.ID, Status: "RUNNING"}
+	pool.handleRunResult(context.Background(), a, execRec, errors.New("boom"), "claude")
+	<-pool.resultCh
+
+	gotB, err := store.GetTask("hrr-cascade-b")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if gotB.State != task.StateCancelled {
+		t.Errorf("hrr-cascade-b state: want CANCELLED after upstream FAILED, got %s", gotB.State)
+	}
+}