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 <noreply@anthropic.com>
This commit is contained in:
davincidreams 2026-04-25 05:07:43 -04:00
parent 862d263cb8
commit 5aecb08211
4 changed files with 396 additions and 14 deletions

View file

@ -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<string, unknown> }) => 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<typeof vi.fn> }> = []
const provider = {
tasks: new Map<string, any>([[parentTaskId, parentTask]]),
focusedTaskId: parentTaskId as string | undefined,
leaderTaskId: parentTaskId as string | undefined,
childCompletionHandlers: new Map<string, CompletionHandler>(),
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")
})
})

View file

@ -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,
},
},

View file

@ -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 {

View file

@ -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<Array<{ taskId: string; summary: string; payload?: Record<string, unknown>; 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<string, unknown>
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<string, unknown>; 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)