From 997cd8b56bc086a02b9c7c006dd62b07b9fcd2f3 Mon Sep 17 00:00:00 2001 From: Claude Sonnet 5 Date: Fri, 3 Jul 2026 23:22:42 +0000 Subject: 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 Claude-Session: https://claude.ai/code/session_01V1moSNCJRcP6kykA4tyUSs --- internal/executor/cascade_test.go | 222 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 222 insertions(+) create mode 100644 internal/executor/cascade_test.go (limited to 'internal/executor/cascade_test.go') 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) + } +} -- cgit v1.2.3