From 9062273ef59f5eff3ebcf47095fb8788301def63 Mon Sep 17 00:00:00 2001 From: davincidreams Date: Sat, 25 Apr 2026 08:05:22 -0400 Subject: [PATCH] feat(swarm-p5): file-backed mailbox for cross-process workers FileMailbox implements IMailboxService using a JSON file per agent with proper-lockfile for concurrent-write safety. Polling replaces the EventEmitter used by InMemoryMailbox. - FileMailbox: ~//.json, 500ms poll interval - Race-safe inbox creation via exclusive O_EXCL write (flag "wx") - MailboxManager: createFileMailbox(sessionId, baseDir?) alongside existing createMailbox(); map type widened to IMailboxService Co-Authored-By: Claude Sonnet 4.6 --- docs/prd-swarm-multi-agent.md | 2 +- src/core/swarm/FileMailbox.ts | 157 +++++++++++++++++++++++++++++++ src/core/swarm/MailboxManager.ts | 29 +++++- 3 files changed, 184 insertions(+), 4 deletions(-) create mode 100644 src/core/swarm/FileMailbox.ts diff --git a/docs/prd-swarm-multi-agent.md b/docs/prd-swarm-multi-agent.md index 1fa9ad70d8..3dda6a0118 100644 --- a/docs/prd-swarm-multi-agent.md +++ b/docs/prd-swarm-multi-agent.md @@ -1,6 +1,6 @@ # PRD: Swarm / True Multi-Agent Execution in Moo Code -**Status:** In progress — Phase 1 shipped; Teams system shipped +**Status:** In progress — P1, PT, P2, P3, P4, P5 shipped **Author:** davincidreams **Branch:** `feat/swarm-prd` **Created:** 2026-04-25 diff --git a/src/core/swarm/FileMailbox.ts b/src/core/swarm/FileMailbox.ts new file mode 100644 index 0000000000..487abf665e --- /dev/null +++ b/src/core/swarm/FileMailbox.ts @@ -0,0 +1,157 @@ +import fs from "fs/promises" +import path from "path" + +import lockfile from "proper-lockfile" + +import type { TeammateMessage, TeammateMessageType } from "@roo-code/types" + +import type { IMailboxService } from "./IMailboxService" + +const POLL_INTERVAL_MS = 500 + +/** + * File-backed mailbox for cross-process swarm workers. + * + * Each agent's queue lives at `/.json`. + * Concurrent writes are serialised with proper-lockfile (advisory lock + * via a sibling `.lock` directory); polling replaces the EventEmitter + * used by InMemoryMailbox. + * + * baseDir is typically `~/.roo/swarm//`. + */ +export class FileMailbox implements IMailboxService { + private _disposed = false + + constructor(private readonly baseDir: string) {} + + // ------------------------------------------------------------------------- + // IMailboxService + // ------------------------------------------------------------------------- + + async send(to: string, msg: TeammateMessage): Promise { + const p = await this.ensureInbox(to) + await this.withLock(p, async () => { + const queue = await this.readRaw(p) + queue.push(msg) + await fs.writeFile(p, JSON.stringify(queue, null, 2), "utf-8") + }) + } + + async read(agentId: string): Promise { + const p = this.inboxPath(agentId) + return this.readRaw(p) + } + + async markRead(agentId: string, idx: number): Promise { + const p = await this.ensureInbox(agentId) + await this.withLock(p, async () => { + const queue = await this.readRaw(p) + queue.splice(idx, 1) + await fs.writeFile(p, JSON.stringify(queue, null, 2), "utf-8") + }) + } + + async waitForMessage( + agentId: string, + types: TeammateMessageType[], + opts: { timeoutMs?: number } = {}, + ): Promise { + const { timeoutMs = 60_000 } = opts + + if (this._disposed) return null + + const immediate = await this.dequeueMatching(agentId, types) + if (immediate) return immediate + + return new Promise((resolve) => { + let settled = false + + const finish = (msg: TeammateMessage | null) => { + if (settled) return + settled = true + clearTimeout(timer) + clearInterval(poll) + resolve(msg) + } + + const timer = setTimeout(() => finish(null), timeoutMs) + + const poll = setInterval(async () => { + if (settled || this._disposed) { + finish(null) + return + } + const msg = await this.dequeueMatching(agentId, types) + if (msg) finish(msg) + }, POLL_INTERVAL_MS) + }) + } + + dispose(): void { + this._disposed = true + } + + // ------------------------------------------------------------------------- + // Internals + // ------------------------------------------------------------------------- + + private inboxPath(agentId: string): string { + // Sanitise agentId so it's safe as a filename. + const safe = agentId.replace(/[^a-zA-Z0-9_-]/g, "_") + return path.join(this.baseDir, `${safe}.json`) + } + + /** Ensure the inbox file and its parent directory exist. Race-safe. */ + private async ensureInbox(agentId: string): Promise { + const p = this.inboxPath(agentId) + await fs.mkdir(path.dirname(p), { recursive: true }) + try { + // wx = exclusive create; throws EEXIST if the file is already there + await fs.writeFile(p, "[]", { flag: "wx", encoding: "utf-8" }) + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err + } + return p + } + + private async readRaw(filePath: string): Promise { + try { + const raw = await fs.readFile(filePath, "utf-8") + return JSON.parse(raw) as TeammateMessage[] + } catch { + return [] + } + } + + private async withLock(filePath: string, fn: () => Promise): Promise { + const release = await lockfile.lock(filePath, { + retries: { retries: 10, minTimeout: 50, maxTimeout: 200 }, + }) + try { + await fn() + } finally { + await release() + } + } + + /** Read, find the first matching message, remove it, write back. */ + private async dequeueMatching(agentId: string, types: TeammateMessageType[]): Promise { + const p = this.inboxPath(agentId) + try { + await fs.access(p) + } catch { + return null // inbox file doesn't exist yet + } + + let result: TeammateMessage | null = null + await this.withLock(p, async () => { + const queue = await this.readRaw(p) + const idx = queue.findIndex((m) => types.includes(m.type)) + if (idx >= 0) { + result = queue.splice(idx, 1)[0] + await fs.writeFile(p, JSON.stringify(queue, null, 2), "utf-8") + } + }) + return result + } +} diff --git a/src/core/swarm/MailboxManager.ts b/src/core/swarm/MailboxManager.ts index d1fddea482..be755ad8a2 100644 --- a/src/core/swarm/MailboxManager.ts +++ b/src/core/swarm/MailboxManager.ts @@ -1,25 +1,48 @@ +import os from "os" +import path from "path" + import type { TeammateMessage } from "@roo-code/types" +import type { IMailboxService } from "./IMailboxService" import { InMemoryMailbox } from "./InMemoryMailbox" +import { FileMailbox } from "./FileMailbox" /** - * Manages one InMemoryMailbox per swarm session. + * Manages one mailbox per swarm session. * A session's mailbox is only created when the session opts in to persistent * workers (`persistent: true` on spawnConcurrentChildren). * + * Two backends are available: + * - InMemoryMailbox — in-process workers (zero latency, default) + * - FileMailbox — cross-process workers (polls JSON file with lockfile) + * * 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() + private mailboxes = new Map() + /** Create an in-process (memory) mailbox for the session. */ createMailbox(sessionId: string): void { if (!this.mailboxes.has(sessionId)) { this.mailboxes.set(sessionId, new InMemoryMailbox()) } } - getMailbox(sessionId: string): InMemoryMailbox | undefined { + /** + * Create a file-backed mailbox for the session. + * Used by cross-process workers (P5/P6). + * Files land at `/.json`. + * Defaults to `~/.roo/swarm//` when baseDir is omitted. + */ + createFileMailbox(sessionId: string, baseDir?: string): void { + if (!this.mailboxes.has(sessionId)) { + const dir = baseDir ?? path.join(os.homedir(), ".roo", "swarm", sessionId) + this.mailboxes.set(sessionId, new FileMailbox(dir)) + } + } + + getMailbox(sessionId: string): IMailboxService | undefined { return this.mailboxes.get(sessionId) }