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/agentchannel/agentchannel.go | 8 ++ internal/agentloop/tools.go | 3 + internal/executor/agentmcp.go | 2 + internal/executor/agentmcp_test.go | 19 +++ internal/executor/cascade_test.go | 222 +++++++++++++++++++++++++++++++++ internal/executor/channel.go | 30 +++-- internal/executor/channel_test.go | 82 ++++++++++++ internal/executor/executor.go | 94 ++++++++++++++ internal/executor/executor_test.go | 1 + internal/executor/nativerunner_test.go | 38 ++++++ internal/storage/db.go | 30 +++++ internal/storage/db_test.go | 76 +++++++++++ 12 files changed, 595 insertions(+), 10 deletions(-) create mode 100644 internal/executor/cascade_test.go (limited to 'internal') diff --git a/internal/agentchannel/agentchannel.go b/internal/agentchannel/agentchannel.go index 2730dfa..42eacbc 100644 --- a/internal/agentchannel/agentchannel.go +++ b/internal/agentchannel/agentchannel.go @@ -34,6 +34,14 @@ type SubtaskSpec struct { Instructions string Model string MaxBudgetUSD float64 + // Role, when non-empty, names an internal/role.RoleConfig the child task + // should dispatch through (task.AgentConfig.Role — see Phase 5's + // role-based dispatch in internal/executor.Pool.execute()). When set, the + // storeChannel implementation leaves Agent.Type/Model empty on the child + // so tier 0 of the role's escalation ladder resolves them, exactly like + // any other role-typed task. Empty (the default) preserves the prior + // hardcoded "claude" behavior for backward compatibility. + Role string } // AgentChannel is how a Runner reports agent-originated signals to the rest of diff --git a/internal/agentloop/tools.go b/internal/agentloop/tools.go index fcb17a2..fc003cb 100644 --- a/internal/agentloop/tools.go +++ b/internal/agentloop/tools.go @@ -52,6 +52,7 @@ func agentToolSpecs() []provider.ToolSpec { "instructions": strProp("complete instructions for the subtask agent"), "model": strProp("optional model override"), "max_budget_usd": map[string]any{"type": "number", "description": "optional budget cap in USD"}, + "role": strProp("optional role name to dispatch the subtask through instead of a fixed model (e.g. an evaluator role); when set, model is ignored and the role's escalation ladder picks the provider/model"), }, "required": []string{"name", "instructions"}, }, @@ -153,6 +154,7 @@ func (l *Loop) dispatchTool(ctx context.Context, name, argsJSON string) (result Instructions string `json:"instructions"` Model string `json:"model"` MaxBudgetUSD float64 `json:"max_budget_usd"` + Role string `json:"role"` } _ = json.Unmarshal([]byte(argsJSON), &a) id, ssErr := l.Channel.SpawnSubtask(ctx, agentchannel.SubtaskSpec{ @@ -160,6 +162,7 @@ func (l *Loop) dispatchTool(ctx context.Context, name, argsJSON string) (result Instructions: a.Instructions, Model: a.Model, MaxBudgetUSD: a.MaxBudgetUSD, + Role: a.Role, }) if ssErr != nil { return "", false, ssErr diff --git a/internal/executor/agentmcp.go b/internal/executor/agentmcp.go index 4368031..2177fbd 100644 --- a/internal/executor/agentmcp.go +++ b/internal/executor/agentmcp.go @@ -66,6 +66,7 @@ type spawnSubtaskInput struct { Instructions string `json:"instructions" jsonschema:"complete instructions for the subtask agent"` Model string `json:"model,omitempty" jsonschema:"optional model override, e.g. sonnet or opus"` MaxBudgetUSD float64 `json:"max_budget_usd,omitempty" jsonschema:"optional budget cap in USD"` + Role string `json:"role,omitempty" jsonschema:"optional role name to dispatch the subtask through instead of a fixed model (e.g. an evaluator role); when set, model is ignored and the role's escalation ladder picks the provider/model"` } type recordProgressInput struct { @@ -118,6 +119,7 @@ func newAgentServer(ch AgentChannel) *mcp.Server { Instructions: in.Instructions, Model: in.Model, MaxBudgetUSD: in.MaxBudgetUSD, + Role: in.Role, }) if err != nil { return nil, nil, err diff --git a/internal/executor/agentmcp_test.go b/internal/executor/agentmcp_test.go index 3973af0..4aeb7cb 100644 --- a/internal/executor/agentmcp_test.go +++ b/internal/executor/agentmcp_test.go @@ -145,6 +145,25 @@ func TestAgentServer_SpawnSubtask(t *testing.T) { } } +// TestAgentServer_SpawnSubtask_RolePassthrough proves the MCP transport +// (ContainerRunner-driven claude/gemini agents) can spawn role-typed +// subtasks too, mirroring the native tool-use loop's coverage in +// nativerunner_test.go. +func TestAgentServer_SpawnSubtask_RolePassthrough(t *testing.T) { + ch := &recordingChannel{spawnID: "sub-2"} + cs := connectInMemory(t, newAgentServer(ch)) + _, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "spawn_subtask", + Arguments: map[string]any{"name": "eval", "instructions": "evaluate", "role": "evaluator_quality"}, + }) + if err != nil { + t.Fatalf("CallTool: %v", err) + } + if len(ch.spawned) != 1 || ch.spawned[0].Role != "evaluator_quality" { + t.Errorf("subtask role not propagated: %+v", ch.spawned) + } +} + func TestAgentServer_RecordProgress(t *testing.T) { ch := &recordingChannel{} cs := connectInMemory(t, newAgentServer(ch)) 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) + } +} diff --git a/internal/executor/channel.go b/internal/executor/channel.go index debe21d..30d826c 100644 --- a/internal/executor/channel.go +++ b/internal/executor/channel.go @@ -112,20 +112,30 @@ func (c *storeChannel) PendingQuestion() (string, bool) { func (c *storeChannel) SpawnSubtask(_ context.Context, spec SubtaskSpec) (string, error) { now := time.Now().UTC() + agent := task.AgentConfig{ + Instructions: spec.Instructions, + MaxBudgetUSD: spec.MaxBudgetUSD, + } + if spec.Role != "" { + // Role-typed subtask: leave Type/Model empty so + // executor.Pool.execute() resolves tier 0 of the role's escalation + // ladder on dispatch, exactly like any other role-typed task. + agent.Role = spec.Role + } else { + // Backward-compatible default: every pre-Phase-6 caller of + // SpawnSubtask gets exactly this unchanged behavior. + agent.Type = "claude" + agent.Model = spec.Model + } child := &task.Task{ ID: uuid.NewString(), Name: spec.Name, ParentTaskID: c.taskID, - Agent: task.AgentConfig{ - Type: "claude", - Model: spec.Model, - Instructions: spec.Instructions, - MaxBudgetUSD: spec.MaxBudgetUSD, - }, - Priority: task.PriorityNormal, - State: task.StatePending, - CreatedAt: now, - UpdatedAt: now, + Agent: agent, + Priority: task.PriorityNormal, + State: task.StatePending, + CreatedAt: now, + UpdatedAt: now, } if err := c.store.CreateTask(child); err != nil { return "", err diff --git a/internal/executor/channel_test.go b/internal/executor/channel_test.go index 250e8eb..157cb0c 100644 --- a/internal/executor/channel_test.go +++ b/internal/executor/channel_test.go @@ -97,6 +97,88 @@ func TestStoreChannel_SpawnSubtask_CreatesChildWithParent(t *testing.T) { } } +// TestStoreChannel_SpawnSubtask_RoleSet verifies that when spec.Role is set, +// the child task gets Agent.Role populated and Agent.Type/Model left empty +// so executor.Pool.execute()'s role-based dispatch (Phase 5) resolves them +// from the role's escalation ladder, exactly like any other role-typed task. +func TestStoreChannel_SpawnSubtask_RoleSet(t *testing.T) { + store := &fakeChannelStore{} + ch := newStoreChannel(store, "parent-1") + id, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{ + Name: "evaluator", + Instructions: "evaluate the change", + Model: "should-be-ignored", + MaxBudgetUSD: 2.0, + Role: "evaluator_quality", + }) + if err != nil { + t.Fatalf("SpawnSubtask: %v", err) + } + if len(store.createdTasks) != 1 { + t.Fatalf("expected 1 created task, got %d", len(store.createdTasks)) + } + ct := store.createdTasks[0] + if ct.ID != id { + t.Errorf("returned id %q != created task id %q", id, ct.ID) + } + if ct.Agent.Role != "evaluator_quality" { + t.Errorf("Agent.Role: want evaluator_quality, got %q", ct.Agent.Role) + } + if ct.Agent.Type != "" { + t.Errorf("Agent.Type: want empty (resolved later by role dispatch), got %q", ct.Agent.Type) + } + if ct.Agent.Model != "" { + t.Errorf("Agent.Model: want empty (resolved later by role dispatch), got %q", ct.Agent.Model) + } + if ct.Agent.Instructions != "evaluate the change" || ct.Agent.MaxBudgetUSD != 2.0 { + t.Errorf("non-role fields not propagated: %+v", ct.Agent) + } +} + +// TestStoreChannel_SpawnSubtask_NoRole_RegressionUnchanged is a regression +// test proving that spec.Role == "" (every pre-Phase-6 caller) still +// produces exactly today's behavior: Agent.Type hardcoded to "claude" and +// Agent.Model set from spec.Model. This must keep passing unmodified by any +// future change to role-typed spawning. +func TestStoreChannel_SpawnSubtask_NoRole_RegressionUnchanged(t *testing.T) { + store := &fakeChannelStore{} + ch := newStoreChannel(store, "parent-1") + id, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{ + Name: "child", + Instructions: "do it", + Model: "sonnet", + MaxBudgetUSD: 1.5, + }) + if err != nil { + t.Fatalf("SpawnSubtask: %v", err) + } + if len(store.createdTasks) != 1 { + t.Fatalf("expected 1 created task, got %d", len(store.createdTasks)) + } + ct := store.createdTasks[0] + if ct.ID != id { + t.Errorf("returned id %q != created task id %q", id, ct.ID) + } + if ct.ParentTaskID != "parent-1" { + t.Errorf("ParentTaskID: got %q want parent-1", ct.ParentTaskID) + } + if ct.Agent.Type != "claude" { + t.Errorf("Agent.Type: want claude (unchanged backward-compat default), got %q", ct.Agent.Type) + } + if ct.Agent.Model != "sonnet" { + t.Errorf("Agent.Model: want sonnet, got %q", ct.Agent.Model) + } + if ct.Agent.Role != "" { + t.Errorf("Agent.Role: want empty, got %q", ct.Agent.Role) + } + if ct.Agent.Instructions != "do it" || ct.Agent.MaxBudgetUSD != 1.5 { + t.Errorf("agent config not propagated: %+v", ct.Agent) + } + if ct.State != task.StatePending { + t.Errorf("expected PENDING child, got %s", ct.State) + } +} + func TestStoreChannel_SpawnSubtask_PropagatesError(t *testing.T) { store := &fakeChannelStore{createTaskErr: errors.New("boom")} ch := newStoreChannel(store, "parent-1") diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 981a3ad..ed38d7d 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -45,6 +45,9 @@ type Store interface { // if none is active. Used by execute() to resolve tier 0 of a role-typed // task's escalation ladder. GetActiveRoleConfig(role string) (*storage.RoleConfigRow, error) + // ListDependents returns tasks that directly depend on taskID. Used by + // cascadeFail to proactively cancel a failed task's downstream subtree. + ListDependents(taskID string) ([]*task.Task, error) } // LogPather is an optional interface runners can implement to provide the log @@ -524,6 +527,16 @@ func (p *Pool) handleRunResult(ctx context.Context, t *task.Task, exec *storage. } } + // Proactively cancel the downstream dependency subtree the moment this + // task lands in a terminal failure state, so tasks depending on it (e.g. + // evaluator subtasks fanned out from a Builder) don't sit around waiting + // on a dependency that will never complete. BLOCKED/READY/COMPLETED are + // intentionally excluded — those aren't terminal failures. + switch exec.Status { + case "FAILED", "TIMED_OUT", "CANCELLED", "BUDGET_EXCEEDED": + p.cascadeFail(t.ID, exec.ErrorMsg) + } + summary := exec.Summary if summary == "" && exec.StdoutPath != "" { summary = extractSummary(exec.StdoutPath) @@ -550,6 +563,75 @@ func (p *Pool) handleRunResult(ctx context.Context, t *task.Task, exec *storage. p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err} } +// cascadeCancellableStates are the task states a dependent task can safely be +// preempted from: it hasn't started running yet. RUNNING, terminal states, +// and BLOCKED (subtask delegation) are all left untouched — cascadeFail only +// preempts work that hasn't started. +var cascadeCancellableStates = map[task.State]bool{ + task.StatePending: true, + task.StateQueued: true, +} + +// cascadeFail proactively cancels the downstream dependency subtree rooted at +// taskID. It is called right after a task is marked into a terminal failure +// state (FAILED/TIMED_OUT/CANCELLED/BUDGET_EXCEEDED): every direct dependent +// still waiting to run (PENDING or QUEUED) is recorded as CANCELLED with a +// message referencing the upstream failure, and the same cancellation is +// applied recursively to that dependent's own dependents — so a multi-level +// chain (A fails, B depends on A, C depends on B) cascades all the way down, +// not just one hop. Dependents that are already RUNNING or in a terminal +// state are left alone. +// +// This exists so a fan-out pattern (e.g. a Builder task with several +// Evaluator subtasks depending on it) doesn't leave those subtasks sitting +// around forever — or worse, waiting on a dependency that will never +// complete — when the upstream task fails. Without this, a dependent only +// discovers its dependency failed the next time something tries to dispatch +// it (see checkDepsReady), which may be never for a task nobody has +// submitted to the pool yet. +func (p *Pool) cascadeFail(taskID string, reason string) { + p.cascadeFailFrom(taskID, reason, make(map[string]bool)) +} + +// cascadeFailFrom does the actual work for cascadeFail, threading a visited +// set through the recursion so a pathological dependency cycle can't cause +// infinite recursion. +func (p *Pool) cascadeFailFrom(taskID string, reason string, visited map[string]bool) { + if visited[taskID] { + return + } + visited[taskID] = true + + dependents, err := p.store.ListDependents(taskID) + if err != nil { + p.logger.Error("cascadeFail: list dependents", "taskID", taskID, "error", err) + return + } + for _, dep := range dependents { + if !cascadeCancellableStates[dep.State] { + continue // already running, already terminal, or blocked — leave it alone + } + msg := fmt.Sprintf("upstream dependency %s failed: %s", taskID, reason) + now := time.Now().UTC() + exec := &storage.Execution{ + ID: uuid.New().String(), + TaskID: dep.ID, + StartTime: now, + EndTime: now, + Status: "CANCELLED", + ErrorMsg: msg, + } + if createErr := p.store.CreateExecution(exec); createErr != nil { + p.logger.Error("cascadeFail: create execution", "taskID", dep.ID, "error", createErr) + } + if stateErr := p.store.UpdateTaskState(dep.ID, task.StateCancelled); stateErr != nil { + p.logger.Error("cascadeFail: update task state", "taskID", dep.ID, "error", stateErr) + continue // couldn't actually cancel it — don't cascade further from here + } + p.logger.Info("cascade-cancelled dependent task", "taskID", dep.ID, "upstream", taskID) + p.cascadeFailFrom(dep.ID, msg, visited) + } +} // ActiveCount returns the number of currently running tasks. func (p *Pool) ActiveCount() int { @@ -839,6 +921,7 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { if err := p.store.UpdateTaskState(t.ID, task.StateBudgetExceeded); err != nil { p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateBudgetExceeded, "error", err) } + p.cascadeFail(t.ID, exec.ErrorMsg) p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: fmt.Errorf("budget cap reached for %s", agentType)} return } @@ -849,6 +932,16 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { // if a dependent task holds the slot while waiting for its dependency to run, // the dependency can never start (maxPerAgent=1). if len(t.DependsOn) > 0 { + // cascadeFail (triggered when some ancestor of t failed) may already + // have cancelled t out from under it while it sat in the AfterFunc + // requeue loop below — re-check the fresh state so we don't try (and + // fail, since CANCELLED has no valid self-transition) to re-apply a + // terminal transition, and — more importantly — so a cascade-cancelled + // task never actually gets dispatched to a runner. + if fresh, ferr := p.store.GetTask(t.ID); ferr == nil && fresh.State != task.StateQueued { + p.logger.Info("skipping dispatch: task no longer queued (likely cascade-cancelled)", "taskID", t.ID, "state", fresh.State) + return + } ready, depErr := p.checkDepsReady(t) if depErr != nil { // A dependency hit a terminal failure — cancel this task immediately. @@ -867,6 +960,7 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { if err := p.store.UpdateTaskState(t.ID, task.StateCancelled); err != nil { p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateCancelled, "error", err) } + p.cascadeFail(t.ID, depErr.Error()) p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: depErr} return } diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go index 8eb7e72..e7e445e 100644 --- a/internal/executor/executor_test.go +++ b/internal/executor/executor_test.go @@ -1166,6 +1166,7 @@ func (m *minimalMockStore) CreateEvent(_ *event.Event) error { return nil } func (m *minimalMockStore) GetActiveRoleConfig(_ string) (*storage.RoleConfigRow, error) { return nil, sql.ErrNoRows } +func (m *minimalMockStore) ListDependents(_ string) ([]*task.Task, error) { return nil, nil } func (m *minimalMockStore) lastStateUpdate() (string, task.State, bool) { m.mu.Lock() diff --git a/internal/executor/nativerunner_test.go b/internal/executor/nativerunner_test.go index ab0317c..b081b8e 100644 --- a/internal/executor/nativerunner_test.go +++ b/internal/executor/nativerunner_test.go @@ -199,6 +199,44 @@ func TestNativeRunner_Run_ToolLoop_ReportSummaryAndSpawn(t *testing.T) { } } +// TestNativeRunner_Run_ToolLoop_SpawnSubtask_RolePassthrough is an +// end-to-end-ish test (fake LLM server, real agentloop.Loop/tools.go +// dispatch, real storeChannel) proving a spawn_subtask tool call carrying a +// "role" argument reaches SubtaskSpec.Role correctly through +// internal/agentloop/tools.go's dispatchTool, and from there through +// storeChannel.SpawnSubtask into the created child task's Agent.Role. +func TestNativeRunner_Run_ToolLoop_SpawnSubtask_RolePassthrough(t *testing.T) { + srv := fakeChatServer(t, []fakeTurn{ + {toolCalls: []llm.ToolCall{toolCall("c1", "spawn_subtask", `{"name":"eval","instructions":"evaluate","role":"evaluator_quality"}`)}}, + {content: "finished"}, + }) + defer srv.Close() + + r := newLocalRunner(t, srv) + tt := localTask() + store := &fakeChannelStore{} + ch := newStoreChannel(store, tt.ID) + exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID} + + if err := r.Run(context.Background(), tt, exec, ch); err != nil { + t.Fatalf("Run: %v", err) + } + + if len(store.createdTasks) != 1 { + t.Fatalf("expected 1 spawned subtask, got %d", len(store.createdTasks)) + } + child := store.createdTasks[0] + if child.Agent.Role != "evaluator_quality" { + t.Errorf("Agent.Role: want evaluator_quality, got %q", child.Agent.Role) + } + if child.Agent.Type != "" { + t.Errorf("Agent.Type: want empty for role-typed subtask, got %q", child.Agent.Type) + } + if child.Agent.Instructions != "evaluate" { + t.Errorf("Agent.Instructions: want evaluate, got %q", child.Agent.Instructions) + } +} + func TestNativeRunner_Run_RecordProgress(t *testing.T) { srv := fakeChatServer(t, []fakeTurn{ {toolCalls: []llm.ToolCall{toolCall("c1", "record_progress", `{"message":"halfway there"}`)}}, diff --git a/internal/storage/db.go b/internal/storage/db.go index a16ad4e..d571c2e 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -287,6 +287,36 @@ func (s *DB) ListSubtasks(parentID string) ([]*task.Task, error) { return tasks, rows.Err() } +// ListDependents returns tasks that directly depend on taskID — i.e. those +// whose depends_on_json contains taskID. depends_on_json has no reverse +// index, so this is a full-table scan; the codebase already accepts this +// tradeoff for other JSON-blob columns at self-hosted scale (see the +// "Additive migration strategy is fragile" design-debt note in CLAUDE.md). +// Only direct dependents are returned — callers that need the full +// transitive downstream subtree (e.g. cascade-cancellation) must recurse. +func (s *DB) ListDependents(taskID string) ([]*task.Task, error) { + rows, err := s.db.Query(`SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json FROM tasks`) + if err != nil { + return nil, err + } + defer rows.Close() + + var dependents []*task.Task + for rows.Next() { + t, err := scanTaskRows(rows) + if err != nil { + return nil, err + } + for _, depID := range t.DependsOn { + if depID == taskID { + dependents = append(dependents, t) + break + } + } + } + return dependents, rows.Err() +} + // UpdateTaskState atomically updates a task's state, enforcing valid // transitions. The transition is attributed to the system actor; callers // acting on behalf of a user should use UpdateTaskStateBy. diff --git a/internal/storage/db_test.go b/internal/storage/db_test.go index 09bbdfc..b53234a 100644 --- a/internal/storage/db_test.go +++ b/internal/storage/db_test.go @@ -1256,4 +1256,80 @@ func TestUpdateProject(t *testing.T) { } } +func makeDepTask(id string, dependsOn []string) *task.Task { + now := time.Now().UTC() + if dependsOn == nil { + dependsOn = []string{} + } + return &task.Task{ + ID: id, Name: "Task " + id, + Agent: task.AgentConfig{Type: "claude", Instructions: "x"}, + Priority: task.PriorityNormal, + Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "linear"}, + Tags: []string{}, + DependsOn: dependsOn, + State: task.StatePending, + CreatedAt: now, UpdatedAt: now, + } +} + +// TestListDependents verifies ListDependents returns exactly the tasks that +// directly depend on the given ID, not transitive dependents and not +// unrelated tasks. +func TestListDependents(t *testing.T) { + db := testDB(t) + + // A has no deps. + // B and C both depend directly on A. + // D depends on B (transitive on A, not direct). + // E has no relation to A at all. + a := makeDepTask("ld-a", nil) + b := makeDepTask("ld-b", []string{"ld-a"}) + c := makeDepTask("ld-c", []string{"ld-a"}) + d := makeDepTask("ld-d", []string{"ld-b"}) + e := makeDepTask("ld-e", nil) + + for _, tk := range []*task.Task{a, b, c, d, e} { + if err := db.CreateTask(tk); err != nil { + t.Fatalf("CreateTask(%s): %v", tk.ID, err) + } + } + + deps, err := db.ListDependents("ld-a") + if err != nil { + t.Fatalf("ListDependents: %v", err) + } + if len(deps) != 2 { + t.Fatalf("want 2 direct dependents of ld-a, got %d: %+v", len(deps), deps) + } + got := map[string]bool{} + for _, dep := range deps { + got[dep.ID] = true + } + if !got["ld-b"] || !got["ld-c"] { + t.Errorf("expected dependents ld-b and ld-c, got %v", got) + } + if got["ld-d"] || got["ld-e"] { + t.Errorf("ListDependents must not include transitive (ld-d) or unrelated (ld-e) tasks: %v", got) + } + + // ld-d depends only on ld-b. + depsOfB, err := db.ListDependents("ld-b") + if err != nil { + t.Fatalf("ListDependents(ld-b): %v", err) + } + if len(depsOfB) != 1 || depsOfB[0].ID != "ld-d" { + t.Errorf("want [ld-d] as dependents of ld-b, got %+v", depsOfB) + } + + // A task with no dependents returns an empty slice, not an error. + depsOfE, err := db.ListDependents("ld-e") + if err != nil { + t.Fatalf("ListDependents(ld-e): %v", err) + } + if len(depsOfE) != 0 { + t.Errorf("want no dependents of ld-e, got %+v", depsOfE) + } +} + -- cgit v1.2.3