From 5aecb0821115bddc51426809618c078a314f4996 Mon Sep 17 00:00:00 2001 From: davincidreams Date: Sat, 25 Apr 2026 05:07:43 -0400 Subject: [PATCH] fix(swarm): schema, mode-switch race, abortOnChildFailure for concurrent spawn Three targeted fixes: 1. Tool schema: add concurrent and abortOnChildFailure params to spawn_parallel_tasks so the LLM can discover and use them. Update description to explain sequential vs concurrent trade-offs. 2. Mode-switch race: spawnConcurrentChildren was using tasks.map(async) which ran all handleModeSwitch calls concurrently, causing each child to read whatever mode happened to win the race. Rewrite to a sequential for-loop for setup (mode switch + createTask), then a tight sync loop for all child.start() calls so execution is still concurrent. 3. abortOnChildFailure for concurrent mode: when a child rejects and the flag is set, immediately abort all sibling tasks via removeClineFromStack and reject their completion Promises. Previously this flag was only wired for the sequential queue-drain path. Add 7 unit tests covering: aggregated results, serialised mode switches, simultaneous start ordering, per-child error collection, sibling abort, event emission, and worktree creation. Co-Authored-By: Claude Sonnet 4.6 --- src/__tests__/spawnConcurrentChildren.spec.ts | 323 ++++++++++++++++++ .../native-tools/spawn_parallel_tasks.ts | 20 +- src/core/tools/SpawnParallelTasksTool.ts | 1 + src/core/webview/ClineProvider.ts | 66 +++- 4 files changed, 396 insertions(+), 14 deletions(-) create mode 100644 src/__tests__/spawnConcurrentChildren.spec.ts diff --git a/src/__tests__/spawnConcurrentChildren.spec.ts b/src/__tests__/spawnConcurrentChildren.spec.ts new file mode 100644 index 0000000000..bc4762bc65 --- /dev/null +++ b/src/__tests__/spawnConcurrentChildren.spec.ts @@ -0,0 +1,323 @@ +// npx vitest run __tests__/spawnConcurrentChildren.spec.ts + +import { describe, it, expect, vi } from "vitest" +import { ClineProvider } from "../core/webview/ClineProvider" +import { RooCodeEventName } from "@roo-code/types" + +vi.mock("@roo-code/telemetry", () => ({ + TelemetryService: { + instance: { + captureParallelTaskSpawned: vi.fn(), + captureParallelTaskCompleted: vi.fn(), + captureParallelTaskChildFailed: vi.fn(), + captureWorktreeCreated: vi.fn(), + captureWorktreeDeleted: vi.fn(), + }, + }, +})) + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +type CompletionHandler = { + resolve: (result: { summary: string; payload?: Record }) => void + reject: (reason: Error) => void +} + +function buildProvider(opts: { parentTaskId: string }) { + const { parentTaskId } = opts + + const parentTask = { taskId: parentTaskId, workspacePath: "/workspace", parentTask: undefined } + + const childTaskCounter = { n: 0 } + const createdChildren: Array<{ taskId: string; start: ReturnType }> = [] + + const provider = { + tasks: new Map([[parentTaskId, parentTask]]), + focusedTaskId: parentTaskId as string | undefined, + leaderTaskId: parentTaskId as string | undefined, + childCompletionHandlers: new Map(), + taskEventListeners: new WeakMap(), + log: vi.fn(), + + // Mock handleModeSwitch — synchronous to avoid ordering issues in tests + handleModeSwitch: vi.fn().mockResolvedValue(undefined), + + // Mock _createWorktreeForTask + _createWorktreeForTask: vi.fn().mockResolvedValue("/worktree/path"), + + // Mock createTask — each call returns a new unique child Task stub + createTask: vi.fn().mockImplementation(async () => { + const id = `child-${++childTaskCounter.n}` + const child = { + taskId: id, + workspacePath: "/workspace", + parentTask: parentTask, + start: vi.fn(), + } + createdChildren.push(child) + // Register in the tasks map as addClineToStack would + provider.tasks.set(id, child) + provider.focusedTaskId = id + return child + }), + + getTaskWithId: vi.fn().mockImplementation(async (id: string) => ({ + historyItem: { id, worktreePath: undefined, workspace: "/workspace" }, + })), + updateTaskHistory: vi.fn().mockResolvedValue(undefined), + + // Mock removeClineFromStack so abort-siblings tests can track calls + removeClineFromStack: vi.fn().mockImplementation(async (opts?: { taskId?: string }) => { + const id = opts?.taskId ?? provider.focusedTaskId + if (id) provider.tasks.delete(id) + }), + + emit: vi.fn(), + } + + return { provider, parentTask, createdChildren, childTaskCounter } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +describe("ClineProvider.spawnConcurrentChildren()", () => { + it("creates and starts all children, returns aggregated results", async () => { + const { provider, createdChildren } = buildProvider({ parentTaskId: "parent-1" }) + + // Schedule resolutions after children are registered + const resolveAll = () => { + for (const [taskId, handler] of provider.childCompletionHandlers) { + handler.resolve({ summary: `result-of-${taskId}` }) + } + } + + // Intercept start() calls to trigger resolutions + let startCount = 0 + const originalCreateTask = provider.createTask + provider.createTask = vi.fn().mockImplementation(async (...args) => { + const child = await originalCreateTask(...args) + const originalStart = child.start + child.start = vi.fn().mockImplementation(() => { + originalStart() + startCount++ + if (startCount === 2) resolveAll() + }) + return child + }) + + const results = await (ClineProvider.prototype as any).spawnConcurrentChildren.call(provider, { + parentTaskId: "parent-1", + tasks: [ + { mode: "code", message: "Task A" }, + { mode: "debug", message: "Task B" }, + ], + }) + + expect(results).toHaveLength(2) + expect(results[0].summary).toBe("result-of-child-1") + expect(results[1].summary).toBe("result-of-child-2") + expect(results.every((r: { error?: string }) => !r.error)).toBe(true) + }) + + it("serialises handleModeSwitch calls (no concurrent mode-switch race)", async () => { + const { provider } = buildProvider({ parentTaskId: "parent-1" }) + const modeOrder: string[] = [] + + provider.handleModeSwitch = vi.fn().mockImplementation(async (mode: string) => { + modeOrder.push(`start-${mode}`) + await Promise.resolve() // yield + modeOrder.push(`end-${mode}`) + }) + + let startCount = 0 + const originalCreateTask = provider.createTask + provider.createTask = vi.fn().mockImplementation(async (...args) => { + const child = await originalCreateTask(...args) + child.start = vi.fn().mockImplementation(() => { + startCount++ + if (startCount === 2) { + for (const [, handler] of provider.childCompletionHandlers) { + handler.resolve({ summary: "done" }) + } + } + }) + return child + }) + + await (ClineProvider.prototype as any).spawnConcurrentChildren.call(provider, { + parentTaskId: "parent-1", + tasks: [ + { mode: "code", message: "A" }, + { mode: "debug", message: "B" }, + ], + }) + + // The calls must be fully interleaved in order — start-code, end-code, start-debug, end-debug + // (not start-code, start-debug, end-code, end-debug which would be the concurrent-map race) + expect(modeOrder).toEqual(["start-code", "end-code", "start-debug", "end-debug"]) + }) + + it("all children start() are called before any await on completions", async () => { + const { provider } = buildProvider({ parentTaskId: "parent-1" }) + const startLog: string[] = [] + + const originalCreateTask = provider.createTask + provider.createTask = vi.fn().mockImplementation(async (...args) => { + const child = await originalCreateTask(...args) + child.start = vi.fn().mockImplementation(() => { + startLog.push(child.taskId) + // Do NOT resolve yet — verifying start was called for all before awaiting + }) + return child + }) + + // Let resolutions happen after the start loop + setImmediate(() => { + for (const [, handler] of provider.childCompletionHandlers) { + handler.resolve({ summary: "done" }) + } + }) + + await (ClineProvider.prototype as any).spawnConcurrentChildren.call(provider, { + parentTaskId: "parent-1", + tasks: [ + { mode: "code", message: "A" }, + { mode: "code", message: "B" }, + { mode: "code", message: "C" }, + ], + }) + + // All three starts fired before we awaited any completion + expect(startLog).toEqual(["child-1", "child-2", "child-3"]) + }) + + it("collects results with errors when children fail (no abortOnChildFailure)", async () => { + const { provider } = buildProvider({ parentTaskId: "parent-1" }) + + let startCount = 0 + const originalCreateTask = provider.createTask + provider.createTask = vi.fn().mockImplementation(async (...args) => { + const child = await originalCreateTask(...args) + child.start = vi.fn().mockImplementation(() => { + startCount++ + const handler = provider.childCompletionHandlers.get(child.taskId) + if (startCount === 1) { + handler?.reject(new Error("child-1 failed")) + } else { + handler?.resolve({ summary: "child-2 succeeded" }) + } + }) + return child + }) + + const results = await (ClineProvider.prototype as any).spawnConcurrentChildren.call(provider, { + parentTaskId: "parent-1", + tasks: [ + { mode: "code", message: "A" }, + { mode: "code", message: "B" }, + ], + abortOnChildFailure: false, + }) + + expect(results).toHaveLength(2) + const failed = results.find((r: { error?: string }) => r.error) + const succeeded = results.find((r: { error?: string }) => !r.error) + expect(failed?.error).toBe("child-1 failed") + expect(succeeded?.summary).toBe("child-2 succeeded") + }) + + it("aborts sibling tasks when abortOnChildFailure is true and a child fails", async () => { + const { provider } = buildProvider({ parentTaskId: "parent-1" }) + + let startCount = 0 + const originalCreateTask = provider.createTask + provider.createTask = vi.fn().mockImplementation(async (...args) => { + const child = await originalCreateTask(...args) + child.start = vi.fn().mockImplementation(() => { + startCount++ + const handler = provider.childCompletionHandlers.get(child.taskId) + if (startCount === 1) { + // child-1 fails immediately + handler?.reject(new Error("child-1 bombed")) + } + // child-2 never resolves on its own — it should be aborted + }) + return child + }) + + const results = await (ClineProvider.prototype as any).spawnConcurrentChildren.call(provider, { + parentTaskId: "parent-1", + tasks: [ + { mode: "code", message: "A" }, + { mode: "code", message: "B" }, + ], + abortOnChildFailure: true, + }) + + expect(results).toHaveLength(2) + // Both results should carry errors + expect(results.every((r: { error?: string }) => Boolean(r.error))).toBe(true) + // removeClineFromStack should have been called for child-2 (the sibling) + expect(provider.removeClineFromStack).toHaveBeenCalledWith( + expect.objectContaining({ taskId: "child-2", skipDelegationRepair: true }), + ) + }) + + it("emits TaskSpawned and TaskDelegated events for each child", async () => { + const { provider } = buildProvider({ parentTaskId: "parent-1" }) + + let startCount = 0 + const originalCreateTask = provider.createTask + provider.createTask = vi.fn().mockImplementation(async (...args) => { + const child = await originalCreateTask(...args) + child.start = vi.fn().mockImplementation(() => { + startCount++ + if (startCount === 2) { + for (const [, h] of provider.childCompletionHandlers) h.resolve({ summary: "done" }) + } + }) + return child + }) + + await (ClineProvider.prototype as any).spawnConcurrentChildren.call(provider, { + parentTaskId: "parent-1", + tasks: [ + { mode: "code", message: "A" }, + { mode: "code", message: "B" }, + ], + }) + + expect(provider.emit).toHaveBeenCalledWith(RooCodeEventName.TaskSpawned, "child-1") + expect(provider.emit).toHaveBeenCalledWith(RooCodeEventName.TaskSpawned, "child-2") + expect(provider.emit).toHaveBeenCalledWith(RooCodeEventName.TaskDelegated, "parent-1", "child-1") + expect(provider.emit).toHaveBeenCalledWith(RooCodeEventName.TaskDelegated, "parent-1", "child-2") + }) + + it("creates worktrees when spec.worktree is specified", async () => { + const { provider } = buildProvider({ parentTaskId: "parent-1" }) + + let startCount = 0 + const originalCreateTask = provider.createTask + provider.createTask = vi.fn().mockImplementation(async (...args) => { + const child = await originalCreateTask(...args) + child.start = vi.fn().mockImplementation(() => { + startCount++ + if (startCount === 1) { + provider.childCompletionHandlers.get(child.taskId)?.resolve({ summary: "done" }) + } + }) + return child + }) + + await (ClineProvider.prototype as any).spawnConcurrentChildren.call(provider, { + parentTaskId: "parent-1", + tasks: [{ mode: "code", message: "A", worktree: "auto" }], + }) + + expect(provider._createWorktreeForTask).toHaveBeenCalledWith("/workspace", "auto", "parent-1") + }) +}) diff --git a/src/core/prompts/tools/native-tools/spawn_parallel_tasks.ts b/src/core/prompts/tools/native-tools/spawn_parallel_tasks.ts index 3816e546ac..9c55d5f8aa 100644 --- a/src/core/prompts/tools/native-tools/spawn_parallel_tasks.ts +++ b/src/core/prompts/tools/native-tools/spawn_parallel_tasks.ts @@ -1,8 +1,12 @@ import type OpenAI from "openai" -const SPAWN_PARALLEL_TASKS_DESCRIPTION = `Spawn multiple subtasks that execute sequentially and whose results are aggregated before returning to you. +const SPAWN_PARALLEL_TASKS_DESCRIPTION = `Spawn multiple subtasks and collect their aggregated results before continuing. -Use this when you need to split work into independent chunks (e.g., implement N features, analyze N files, run N experiments) and collect all results before proceeding. Each task runs to completion before the next starts. When all tasks finish, you receive their aggregated results as a JSON array. +**Execution modes:** +- \`concurrent: false\` (default) — tasks run one at a time; the parent task is suspended until all finish. Lower resource use; good for dependent or quota-sensitive work. +- \`concurrent: true\` — all tasks start simultaneously and run in parallel while the parent stays active. Best for truly independent work where wall-clock time matters. + +Use this when you need to split work into independent chunks (e.g., implement N features, analyze N files, run N experiments). When all tasks finish, you receive their aggregated results as a JSON array. CRITICAL: This tool MUST be called alone. Do NOT call it alongside other tools in the same turn.` @@ -44,8 +48,18 @@ export default { }, minItems: 2, }, + concurrent: { + type: ["boolean", "null"], + description: + "When true, all tasks run simultaneously (parent stays active). When false or omitted, tasks run sequentially.", + }, + abortOnChildFailure: { + type: ["boolean", "null"], + description: + "When true and concurrent is also true, abort all remaining sibling tasks as soon as one fails. Has no effect in sequential mode (use the default false to collect all results regardless of failures).", + }, }, - required: ["tasks"], + required: ["tasks", "concurrent", "abortOnChildFailure"], additionalProperties: false, }, }, diff --git a/src/core/tools/SpawnParallelTasksTool.ts b/src/core/tools/SpawnParallelTasksTool.ts index ae36cd564f..68d67e07e2 100644 --- a/src/core/tools/SpawnParallelTasksTool.ts +++ b/src/core/tools/SpawnParallelTasksTool.ts @@ -105,6 +105,7 @@ export class SpawnParallelTasksTool extends BaseTool<"spawn_parallel_tasks"> { worktree: t.spec.worktree || undefined, todos: t.todoItems, })), + abortOnChildFailure, }) pushToolResult(JSON.stringify(results, null, 2)) } else { diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index 11e2fc97c6..7b9739d258 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -3811,8 +3811,9 @@ export class ClineProvider public async spawnConcurrentChildren(params: { parentTaskId: string tasks: Array<{ mode: string; message: string; worktree?: string; todos?: TodoItem[] }> + abortOnChildFailure?: boolean }): Promise; error?: string }>> { - const { parentTaskId, tasks } = params + const { parentTaskId, tasks, abortOnChildFailure = false } = params const parent = this.tasks.get(parentTaskId) if (!parent) { @@ -3822,8 +3823,22 @@ export class ClineProvider TelemetryService.instance.captureParallelTaskSpawned(parentTaskId, tasks.length) this.log(`[spawnConcurrentChildren] Spawning ${tasks.length} concurrent children for parent ${parentTaskId}`) - const childPromises = tasks.map(async (spec) => { - // Switch mode and optionally create worktree before spawning. + // Phase 1: sequential setup — mode switch and task creation must be serialised to + // prevent handleModeSwitch() from racing (it mutates global provider mode state). + // Task.start() is NOT called yet; that happens in phase 2. + type ChildEntry = { + child: Task + taskIdForError: string + completionPromise: Promise<{ + taskId: string + summary: string + payload?: Record + error?: string + }> + } + const entries: ChildEntry[] = [] + + for (const spec of tasks) { try { await this.handleModeSwitch(spec.mode as any) } catch { @@ -3851,7 +3866,6 @@ export class ClineProvider } } - // Register a completion handler before starting the child. const completionPromise = new Promise<{ taskId: string summary: string @@ -3864,24 +3878,54 @@ export class ClineProvider }) }) + entries.push({ child, taskIdForError: child.taskId, completionPromise }) + } + + // Phase 2: start all children in a tight sync loop so they run concurrently. + for (const { child } of entries) { child.start() this.emit(RooCodeEventName.TaskSpawned, child.taskId) this.emit(RooCodeEventName.TaskDelegated, parentTaskId, child.taskId) + } - return completionPromise - }) + // Phase 3: await completions. + // When abortOnChildFailure is set, the first rejection aborts all remaining siblings. + let abortTriggered = false + const abortSiblings = async (failedTaskId: string) => { + if (abortTriggered) return + abortTriggered = true + for (const { child } of entries) { + if (child.taskId === failedTaskId) continue + const handler = this.childCompletionHandlers.get(child.taskId) + this.childCompletionHandlers.delete(child.taskId) + // Abort the running child task. + await this.removeClineFromStack({ taskId: child.taskId, skipDelegationRepair: true }) + handler?.reject(new Error("Aborted: sibling task failed")) + TelemetryService.instance.captureParallelTaskChildFailed(parentTaskId, child.taskId) + } + this.log(`[spawnConcurrentChildren] Aborted remaining siblings after child ${failedTaskId} failed`) + } - // Await all children; collect results even for failed ones (Promise.allSettled semantics). - const settled = await Promise.allSettled(childPromises) + const wrappedPromises = entries.map(({ child, completionPromise }) => + completionPromise.then( + (r) => r, + async (err: unknown) => { + if (abortOnChildFailure) { + await abortSiblings(child.taskId) + } + throw err + }, + ), + ) + + const settled = await Promise.allSettled(wrappedPromises) const results: Array<{ taskId: string; summary: string; payload?: Record; error?: string }> = settled.map((s, i) => { if (s.status === "fulfilled") { return s.value } - // Rejected means child threw; synthesize an error result. - const taskId = tasks[i] ? `child-${i}` : `child-${i}` const msg = s.reason instanceof Error ? s.reason.message : String(s.reason) - return { taskId, summary: msg, error: msg } + return { taskId: entries[i]?.taskIdForError ?? `child-${i}`, summary: msg, error: msg } }) const hadFailures = results.some((r) => r.error !== undefined)