diff --git a/packages/types/src/events.ts b/packages/types/src/events.ts index 88e4b73cf8..de31cd43e1 100644 --- a/packages/types/src/events.ts +++ b/packages/types/src/events.ts @@ -55,6 +55,8 @@ export enum RooCodeEventName { SwarmSessionStarted = "swarmSessionStarted", SwarmSessionEnded = "swarmSessionEnded", WorkerRegistered = "workerRegistered", + WorkerIdle = "workerIdle", + WorkerShutdown = "workerShutdown", // Evals EvalPass = "evalPass", @@ -120,6 +122,8 @@ export const rooCodeEventsSchema = z.object({ [RooCodeEventName.SwarmSessionStarted]: z.tuple([z.string(), z.string()]), // [sessionId, leaderTaskId] [RooCodeEventName.SwarmSessionEnded]: z.tuple([z.string(), z.string()]), // [sessionId, leaderTaskId] [RooCodeEventName.WorkerRegistered]: z.tuple([z.string(), z.string(), z.string(), z.string()]), // [sessionId, taskId, agentName, agentColor] + [RooCodeEventName.WorkerIdle]: z.tuple([z.string(), z.string()]), // [sessionId, taskId] + [RooCodeEventName.WorkerShutdown]: z.tuple([z.string(), z.string()]), // [sessionId, taskId] [RooCodeEventName.ModeChanged]: z.tuple([z.string()]), [RooCodeEventName.ProviderProfileChanged]: z.tuple([z.object({ name: z.string(), provider: z.string() })]), @@ -299,6 +303,16 @@ export const taskEventSchema = z.discriminatedUnion("eventName", [ payload: rooCodeEventsSchema.shape[RooCodeEventName.WorkerRegistered], taskId: z.number().optional(), }), + z.object({ + eventName: z.literal(RooCodeEventName.WorkerIdle), + payload: rooCodeEventsSchema.shape[RooCodeEventName.WorkerIdle], + taskId: z.number().optional(), + }), + z.object({ + eventName: z.literal(RooCodeEventName.WorkerShutdown), + payload: rooCodeEventsSchema.shape[RooCodeEventName.WorkerShutdown], + taskId: z.number().optional(), + }), // Evals z.object({ diff --git a/packages/types/src/task.ts b/packages/types/src/task.ts index 4f6eb344ea..2034e7c1dd 100644 --- a/packages/types/src/task.ts +++ b/packages/types/src/task.ts @@ -89,6 +89,8 @@ export type TaskProviderEvents = { [RooCodeEventName.SwarmSessionStarted]: [sessionId: string, leaderTaskId: string] [RooCodeEventName.SwarmSessionEnded]: [sessionId: string, leaderTaskId: string] [RooCodeEventName.WorkerRegistered]: [sessionId: string, taskId: string, agentName: string, agentColor: string] + [RooCodeEventName.WorkerIdle]: [sessionId: string, taskId: string] + [RooCodeEventName.WorkerShutdown]: [sessionId: string, taskId: string] } /** diff --git a/src/core/swarm/IMailboxService.ts b/src/core/swarm/IMailboxService.ts new file mode 100644 index 0000000000..6c698e9ec4 --- /dev/null +++ b/src/core/swarm/IMailboxService.ts @@ -0,0 +1,30 @@ +import type { TeammateMessage, TeammateMessageType } from "@roo-code/types" + +/** + * Abstraction over a mailbox backend. + * - InMemoryMailbox — in-process workers (P3, zero latency) + * - FileMailbox — cross-process workers (P5, uses proper-lockfile) + */ +export interface IMailboxService { + /** Deliver a message to the agent identified by `to`. */ + send(to: string, msg: TeammateMessage): Promise + + /** Peek at all pending messages for `agentId` without removing them. */ + read(agentId: string): Promise + + /** Remove the message at `idx` for `agentId`. */ + markRead(agentId: string, idx: number): Promise + + /** + * Wait for the next message whose type is in `types`. + * Removes the matched message from the queue before returning it. + * Returns null if the timeout elapses before a matching message arrives. + */ + waitForMessage( + agentId: string, + types: TeammateMessageType[], + opts?: { timeoutMs?: number }, + ): Promise + + dispose(): void +} diff --git a/src/core/swarm/InMemoryMailbox.ts b/src/core/swarm/InMemoryMailbox.ts new file mode 100644 index 0000000000..f8aa1cbe2b --- /dev/null +++ b/src/core/swarm/InMemoryMailbox.ts @@ -0,0 +1,82 @@ +import EventEmitter from "events" + +import type { TeammateMessage, TeammateMessageType } from "@roo-code/types" + +import type { IMailboxService } from "./IMailboxService" + +/** + * In-process mailbox backed by a plain Map + EventEmitter. + * Delivery is synchronous (within the same event loop turn) so there is zero + * polling latency. A configurable timeout provides a safety fallback. + */ +export class InMemoryMailbox implements IMailboxService { + private queues = new Map() + private emitter = new EventEmitter() + + constructor() { + // Suppress MaxListenersExceededWarning when many workers are waiting simultaneously. + this.emitter.setMaxListeners(0) + } + + async send(to: string, msg: TeammateMessage): Promise { + if (!this.queues.has(to)) this.queues.set(to, []) + this.queues.get(to)!.push(msg) + this.emitter.emit(`msg:${to}`) + } + + async read(agentId: string): Promise { + return [...(this.queues.get(agentId) ?? [])] + } + + async markRead(agentId: string, idx: number): Promise { + this.queues.get(agentId)?.splice(idx, 1) + } + + async waitForMessage( + agentId: string, + types: TeammateMessageType[], + opts: { timeoutMs?: number } = {}, + ): Promise { + const { timeoutMs = 60_000 } = opts + + // Check the existing queue before registering a listener. + const match = this.dequeueMatching(agentId, types) + if (match) return match + + return new Promise((resolve) => { + let settled = false + + const timer = setTimeout(() => { + if (settled) return + settled = true + this.emitter.off(`msg:${agentId}`, handler) + resolve(null) + }, timeoutMs) + + const handler = () => { + if (settled) return + const msg = this.dequeueMatching(agentId, types) + if (!msg) return // Message was of a different type — keep waiting. + settled = true + clearTimeout(timer) + this.emitter.off(`msg:${agentId}`, handler) + resolve(msg) + } + + this.emitter.on(`msg:${agentId}`, handler) + }) + } + + dispose(): void { + this.queues.clear() + this.emitter.removeAllListeners() + } + + private dequeueMatching(agentId: string, types: TeammateMessageType[]): TeammateMessage | null { + const q = this.queues.get(agentId) + if (!q) return null + const idx = q.findIndex((m) => types.includes(m.type)) + if (idx < 0) return null + return q.splice(idx, 1)[0] + } +} diff --git a/src/core/swarm/MailboxManager.ts b/src/core/swarm/MailboxManager.ts new file mode 100644 index 0000000000..d1fddea482 --- /dev/null +++ b/src/core/swarm/MailboxManager.ts @@ -0,0 +1,96 @@ +import type { TeammateMessage } from "@roo-code/types" + +import { InMemoryMailbox } from "./InMemoryMailbox" + +/** + * Manages one InMemoryMailbox per swarm session. + * A session's mailbox is only created when the session opts in to persistent + * workers (`persistent: true` on spawnConcurrentChildren). + * + * All public methods are no-ops when the session has no mailbox, so callers + * don't need to guard against missing sessions. + */ +export class MailboxManager { + private mailboxes = new Map() + + createMailbox(sessionId: string): void { + if (!this.mailboxes.has(sessionId)) { + this.mailboxes.set(sessionId, new InMemoryMailbox()) + } + } + + getMailbox(sessionId: string): InMemoryMailbox | undefined { + return this.mailboxes.get(sessionId) + } + + /** + * Called by a worker when it finishes a turn. + * Stores an `idle_notification` in the worker's own queue so the leader's + * `on(WorkerIdle)` handler can read it if needed. + */ + async notifyIdle( + sessionId: string, + workerId: string, + summary: string, + payload?: Record, + ): Promise { + const mailbox = this.mailboxes.get(sessionId) + if (!mailbox) return + await mailbox.send(`leader:${sessionId}`, { + type: "idle_notification", + from: workerId, + to: `leader:${sessionId}`, + payload: { workerId, summary, ...payload }, + ts: Date.now(), + }) + } + + /** Leader assigns a new task to an idle worker. */ + async assignTask(sessionId: string, workerId: string, message: string): Promise { + const mailbox = this.mailboxes.get(sessionId) + if (!mailbox) throw new Error(`[MailboxManager] No mailbox for session "${sessionId}"`) + await mailbox.send(workerId, { + type: "task_assignment", + from: `leader:${sessionId}`, + to: workerId, + payload: { message }, + ts: Date.now(), + }) + } + + /** Leader tells a worker to stop after its current idle period. */ + async shutdownWorker(sessionId: string, workerId: string): Promise { + const mailbox = this.mailboxes.get(sessionId) + if (!mailbox) return + await mailbox.send(workerId, { + type: "shutdown_request", + from: `leader:${sessionId}`, + to: workerId, + ts: Date.now(), + }) + } + + /** + * Waits for the worker's next `task_assignment` or `shutdown_request`. + * Returns null if the timeout fires first or the session has no mailbox. + */ + async waitForNextMessage( + sessionId: string, + workerId: string, + opts?: { timeoutMs?: number }, + ): Promise { + const mailbox = this.mailboxes.get(sessionId) + if (!mailbox) return null + return mailbox.waitForMessage(workerId, ["task_assignment", "shutdown_request"], opts) + } + + destroyMailbox(sessionId: string): void { + this.mailboxes.get(sessionId)?.dispose() + this.mailboxes.delete(sessionId) + } + + dispose(): void { + for (const mailbox of this.mailboxes.values()) mailbox.dispose() + this.mailboxes.clear() + } +} diff --git a/src/core/tools/AttemptCompletionTool.ts b/src/core/tools/AttemptCompletionTool.ts index 2e7a8b7a09..23a5cb0dd2 100644 --- a/src/core/tools/AttemptCompletionTool.ts +++ b/src/core/tools/AttemptCompletionTool.ts @@ -8,6 +8,8 @@ import { formatResponse } from "../prompts/responses" import { Package } from "../../shared/package" import type { ToolUse } from "../../shared/tools" import { t } from "../../i18n" +import type { MailboxManager } from "../swarm/MailboxManager" +import type { SwarmRegistry } from "../swarm/SwarmRegistry" import { BaseTool, ToolCallbacks } from "./BaseTool" @@ -42,6 +44,12 @@ interface DelegationProvider { payload?: Record failed?: boolean }): Promise + /** SwarmRegistry — used to find which session a task belongs to. */ + swarmRegistry?: SwarmRegistry + /** MailboxManager — only present when the session was started with persistent:true. */ + mailboxManager?: MailboxManager + /** EventEmitter — used to emit swarm lifecycle events. */ + emit?(event: string, ...args: unknown[]): boolean } export class AttemptCompletionTool extends BaseTool<"attempt_completion"> { @@ -118,6 +126,7 @@ export class AttemptCompletionTool extends BaseTool<"attempt_completion"> { if (delegation === "delegated") { this.emitTaskCompleted(task) } + // "reassigned" — worker continues; do NOT emit TaskCompleted if (delegation !== "continue") return } else { // Unexpected status (undefined or "delegated") - log error and skip delegation @@ -160,9 +169,10 @@ export class AttemptCompletionTool extends BaseTool<"attempt_completion"> { /** * Handles the common delegation flow when a subtask completes. * Returns: - * - "delegated" when completion was approved and parent resumed - * - "denied" when user denied finishing the subtask - * - "continue" when caller should fall through to normal completion ask flow + * - "delegated" completion was approved and parent resumed (or concurrent Promise resolved) + * - "reassigned" worker received a new task_assignment; LLM loop continues without completing + * - "denied" user denied finishing the subtask + * - "continue" caller should fall through to normal completion ask flow */ private async delegateToParent( task: Task, @@ -170,7 +180,7 @@ export class AttemptCompletionTool extends BaseTool<"attempt_completion"> { provider: DelegationProvider, askFinishSubTaskApproval: () => Promise, pushToolResult: (result: string) => void, - ): Promise<"delegated" | "denied" | "continue"> { + ): Promise<"delegated" | "reassigned" | "denied" | "continue"> { const didApprove = await askFinishSubTaskApproval() if (!didApprove) { @@ -178,8 +188,6 @@ export class AttemptCompletionTool extends BaseTool<"attempt_completion"> { return "denied" } - pushToolResult("") - // If the result is valid JSON, pass it as a structured payload alongside the summary. let completionPayload: Record | undefined try { @@ -192,11 +200,40 @@ export class AttemptCompletionTool extends BaseTool<"attempt_completion"> { } // Concurrent path: if the parent task is still alive in the provider's task map, - // this child was spawned concurrently — resolve its completion Promise rather than - // recreating the parent from history. + // this child was spawned concurrently — either enter the idle loop (persistent) or + // resolve its completion Promise immediately. const parentIsAlive = typeof provider.getTaskById === "function" && provider.getTaskById(task.parentTaskId!) !== undefined if (parentIsAlive && typeof provider.resolveChildCompletion === "function") { + // Idle loop — only for persistent swarm sessions (a mailbox exists for the session). + const session = provider.swarmRegistry?.getSessionForTask(task.taskId) + const mailbox = session ? provider.mailboxManager?.getMailbox(session.sessionId) : undefined + + if (session && mailbox) { + // Notify the leader that this worker is idle. + await provider.mailboxManager!.notifyIdle(session.sessionId, task.taskId, result, completionPayload) + provider.emit?.(RooCodeEventName.WorkerIdle, session.sessionId, task.taskId) + + // Wait for leader to either assign a new task or shut us down. + const nextMsg = await provider.mailboxManager!.waitForNextMessage(session.sessionId, task.taskId, { + timeoutMs: 300_000, // 5-minute safety timeout + }) + + if (nextMsg?.type === "task_assignment" && typeof nextMsg.payload?.message === "string") { + // Give worker a new task — do NOT resolve the parent's completion Promise. + pushToolResult( + `[Task complete]\n\nNew task assigned by swarm leader:\n\n${nextMsg.payload.message}`, + ) + return "reassigned" + } + + // shutdown_request or timeout — fall through to complete normally. + if (nextMsg?.type === "shutdown_request") { + provider.emit?.(RooCodeEventName.WorkerShutdown, session.sessionId, task.taskId) + } + } + + pushToolResult("") await provider.resolveChildCompletion({ childTaskId: task.taskId, summary: result, @@ -206,6 +243,7 @@ export class AttemptCompletionTool extends BaseTool<"attempt_completion"> { } // Sequential path: parent was disposed; recreate it from history. + pushToolResult("") await provider.reopenParentFromDelegation({ parentTaskId: task.parentTaskId!, childTaskId: task.taskId, diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index 17ed434d22..78d58c8f11 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -79,6 +79,7 @@ import { MdmService } from "../../services/mdm/MdmService" import { SkillsManager } from "../../services/skills/SkillsManager" import { TeamsManager } from "../../services/teams/TeamsManager" import { SwarmRegistry } from "../swarm/SwarmRegistry" +import { MailboxManager } from "../swarm/MailboxManager" import { fileExistsAtPath } from "../../utils/fs" import { setTtsEnabled, setTtsSpeed } from "../../utils/tts" @@ -156,6 +157,7 @@ export class ClineProvider protected skillsManager?: SkillsManager protected teamsManager?: TeamsManager protected swarmRegistry: SwarmRegistry = new SwarmRegistry() + protected mailboxManager: MailboxManager = new MailboxManager() private marketplaceManager: MarketplaceManager private mdmService?: MdmService private taskCreationCallback: (task: Task) => void @@ -744,6 +746,7 @@ export class ClineProvider this.skillsManager = undefined this.teamsManager = undefined this.swarmRegistry.dispose() + this.mailboxManager.dispose() this.marketplaceManager?.cleanup() this.customModesManager?.dispose() this.taskHistoryStore.dispose() @@ -3828,8 +3831,11 @@ export class ClineProvider parentTaskId: string tasks: Array<{ mode: string; message: string; worktree?: string; todos?: TodoItem[]; role?: string }> abortOnChildFailure?: boolean + /** When true, workers enter an idle loop after each turn and await task_assignment + * or shutdown_request instead of completing immediately. */ + persistent?: boolean }): Promise; error?: string }>> { - const { parentTaskId, tasks, abortOnChildFailure = false } = params + const { parentTaskId, tasks, abortOnChildFailure = false, persistent = false } = params const parent = this.tasks.get(parentTaskId) if (!parent) { @@ -3842,6 +3848,9 @@ export class ClineProvider // Create a swarm session so every worker gets a stable identity + color. const sessionId = parentTaskId this.swarmRegistry.createSession(sessionId, parentTaskId) + if (persistent) { + this.mailboxManager.createMailbox(sessionId) + } this.emit(RooCodeEventName.SwarmSessionStarted, sessionId, parentTaskId) // Mark parent as swarm leader in its history item. @@ -3984,12 +3993,29 @@ export class ClineProvider `[spawnConcurrentChildren] All ${tasks.length} children completed for parent ${parentTaskId}. failures=${hadFailures}`, ) + if (persistent) { + this.mailboxManager.destroyMailbox(sessionId) + } this.swarmRegistry.destroySession(sessionId) this.emit(RooCodeEventName.SwarmSessionEnded, sessionId, parentTaskId) return results } + /** Assign a new task to an idle persistent worker. */ + public async assignTaskToWorker(sessionId: string, workerId: string, message: string): Promise { + await this.mailboxManager.assignTask(sessionId, workerId, message) + } + + /** Send shutdown_request to all workers in a session so they resolve and the parent can collect results. */ + public async shutdownWorkers(sessionId: string): Promise { + const session = this.swarmRegistry.getSession(sessionId) + if (!session) return + await Promise.all( + Object.keys(session.teammates).map((workerId) => this.mailboxManager.shutdownWorker(sessionId, workerId)), + ) + } + /** * Called by AttemptCompletionTool when a concurrent child completes. * Resolves the Promise registered in childCompletionHandlers for this child,