From fafa5d63f940c37b4ad7adf3d8d7fbe098803a06 Mon Sep 17 00:00:00 2001 From: Bradley Axen Date: Sat, 10 Oct 2026 21:28:08 -0700 Subject: [PATCH] perf(codex): one app-server for every agent Codex agents each kept their own `codex app-server` (~300-400 MB with its MCP children) until the agent stopped running. With agents now running in every joined community, that cost grew with every Codex agent used. Everything agent-specific already travels with the thread: cwd, developer instructions and dynamic tools go in thread/start, and tool calls carry their threadId. So the runtime now owns one server, opened on first use in ~/.buzz, and finds a tool call's agent by thread and turn across all agents. Entries keep their conversations and saved bindings but no longer own a process. A stopped agent can no longer take its process down with it, so it interrupts its own turns, then cleans the background shells its threads left (a dev server from a finished turn, say). Starting the same agent again waits for that, so it never resumes a thread mid-teardown. The server ends with the last Codex agent, on dispose, or on a crash (the next mention reopens it). live.mjs passed the relay as an object since the per-agent relay change; it now passes a function, expands ~ like the desktop host, and checks two agents on one real server. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Bradley Axen --- src/bundled/codex/README.md | 17 ++-- src/bundled/codex/index.ts | 4 +- src/bundled/codex/live.mjs | 55 +++++++++++- src/bundled/codex/runtime.test.ts | 129 ++++++++++++++++++++++++++++ src/bundled/codex/runtime.ts | 135 ++++++++++++++++++++---------- 5 files changed, 287 insertions(+), 53 deletions(-) diff --git a/src/bundled/codex/README.md b/src/bundled/codex/README.md index b13bedc13..dd92278f5 100644 --- a/src/bundled/codex/README.md +++ b/src/bundled/codex/README.md @@ -2,8 +2,10 @@ A bundled desktop plugin, on by default on macOS and Linux, using the native process host introduced in #731. Agents2 owns the identity, admission, configuration, and signed Buzz actions. This plugin -owns Codex work after delivery handover. It starts one `codex app-server` per agent -and uses independent Codex threads for Buzz conversations. +owns Codex work after delivery handover. It starts one `codex app-server` for all +agents and uses an independent Codex thread for each agent's Buzz conversation. +Each thread carries its agent's workspace, instructions and tools, so one crash +fails every agent's current turn; the next mention reopens the server. ## Try it @@ -47,8 +49,10 @@ current turn finishes. Expect the marker. Thread scope is the default. Channel scope shares a session across channel threads; DMs always share a conversation. Workspace changes start a fresh session. Bindings persist locally under the community and agent identity. Earlier bindings without dynamic tools are ignored; new conversations start fresh. Switching -communities does not end its server; removing the agent, leaving or disconnecting -from its community, or disabling the plugin does. +communities does not end an agent's work. Removing the agent, leaving or +disconnecting from its community interrupts its turns and ends shells they left +running; the server ends when no +Codex agent that has worked is still running, or when the plugin is disabled. If a saved Codex thread no longer exists, the next mention starts a fresh one. To inspect a saved session, see [View a Codex session](../../../docs/view-codex-session.md). @@ -77,8 +81,9 @@ workspace and a simulated relay. It sends no Buzz messages. It checks native she and file tools, the native coding prompt, conversation context, default steering, idle follow-ups, process shutdown, saved-thread reuse, changed and cleared instructions, missing-thread recovery, and disabled inherited MCP tools. It also -checks real dynamic Buzz callbacks, memory, workspace file uploads, and normal -final sends without aborted-turn markers. Set +checks real dynamic Buzz callbacks, memory, workspace file uploads, normal +final sends without aborted-turn markers, and two agents sharing one server with +their own workspace, instructions and replies. Set `BUZZ_CODEX_TEST_MODEL` to another available model if needed. It prints only sanitized outcomes; it does not save raw protocol or session logs. diff --git a/src/bundled/codex/index.ts b/src/bundled/codex/index.ts index 995a06de6..37195bef2 100644 --- a/src/bundled/codex/index.ts +++ b/src/bundled/codex/index.ts @@ -16,8 +16,8 @@ export const apply: PluginModule["apply"] = (ctx) => { }); const runtime = new CodexRuntime(start, ctx.agents2.relay); const typeKey = `${ctx.pluginOwner?.id ?? "buzz.codex"}/codex`; - // Agents that stop running stop their servers; switching communities or a - // reconnect does not. + // Agents that stop running stop their turns, and the last one stops the + // server; switching communities or a reconnect does not. const sync = () => { const snapshot = ctx.agents2.snapshot(); if (snapshot.status === "loading") return; diff --git a/src/bundled/codex/live.mjs b/src/bundled/codex/live.mjs index fe197614b..dceae90ae 100644 --- a/src/bundled/codex/live.mjs +++ b/src/bundled/codex/live.mjs @@ -2,7 +2,7 @@ import assert from "node:assert/strict"; import { spawn } from "node:child_process"; import { access, mkdtemp, readFile, writeFile } from "node:fs/promises"; -import { tmpdir } from "node:os"; +import { homedir, tmpdir } from "node:os"; import { join } from "node:path"; import { build } from "rolldown"; import { JSDOM } from "jsdom"; @@ -20,6 +20,7 @@ const { CodexRuntime } = await import(join(workspace, "runtime.mjs")); const { AppServer } = await import(join(workspace, "rpc.mjs")); const children = new Set(); const notices = []; +let servers = 0; const sent = []; let acceptedSteers = 0; const nativeSpawn = async (_id, options = {}) => { @@ -46,6 +47,7 @@ const nativeSpawn = async (_id, options = {}) => { exited, }; } + servers++; const child = spawn( "codex", [ @@ -58,7 +60,8 @@ const nativeSpawn = async (_id, options = {}) => { `mcp_servers={"fixture.tools"={command="sh",args=["-c",${JSON.stringify(`touch '${join(workspace, "mcp-must-not-start")}'`)}]}}`, ], { - cwd: options.cwd, + // Like the desktop host, expand ~ in a process's working directory. + cwd: options.cwd?.replace(/^~(?=\/|$)/, homedir()), detached: true, env: Object.fromEntries( Object.entries(process.env).filter( @@ -169,7 +172,7 @@ const relay = { }, }), }; -const runtime = new CodexRuntime(nativeSpawn, relay, store); +const runtime = new CodexRuntime(nativeSpawn, () => relay, store); runtime.sync([{ pubkey }], scope); let id = 0; const delivery = (content) => { @@ -469,6 +472,52 @@ try { console.log( "PASS real dynamic callbacks, resumed tools, shared reads, memory, workspace file read/upload and final send without duplicate replies", ); + const other = "9".repeat(64); + const otherWorkspace = await mkdtemp(join(tmpdir(), "buzz-codex-other-")); + const otherPublished = []; + runtime.sync([{ pubkey }, { pubkey: other }]); + const before = servers; + const first = delivery( + "Reply exactly FIRST_AGENT_OK. No coding tools. Send the reply with buzz.send.", + ); + const second = delivery( + "Run pwd, then reply with its output. Send the reply with buzz.send.", + ); + await runtime.run(first); + await runtime.run({ + ...second, + agent: { + ...second.agent, + pubkey: other, + name: "Other", + publish: async (event) => { + otherPublished.push(event); + return { ...event, id: "8".repeat(64) }; + }, + }, + config: { + ...second.config, + workspace: otherWorkspace, + instructions: `${second.config.instructions} Every final answer MUST contain OTHER_AGENT.`, + }, + }); + await condition( + () => + published.length === 8 && + otherPublished.length === 1 && + runtime.sessions(pubkey)[0]?.status === "Idle" && + runtime.sessions(other)[0]?.status === "Idle", + "two agents on one server", + ); + // The previous step's server is still open and serves both agents. + assert.equal(servers, before); + assert.match(published.at(-1).content, /FIRST_AGENT_OK/); + assert.doesNotMatch(published.at(-1).content, /OTHER_AGENT/); + assert.match(otherPublished[0].content, /OTHER_AGENT/); + assert(otherPublished[0].content.includes(otherWorkspace.split("/").at(-1))); + console.log( + "PASS two agents share one server with their own workspace, instructions and replies", + ); } finally { await inspect?.close(); runtime.dispose(); diff --git a/src/bundled/codex/runtime.test.ts b/src/bundled/codex/runtime.test.ts index 058db43cb..b2d6599e0 100644 --- a/src/bundled/codex/runtime.test.ts +++ b/src/bundled/codex/runtime.test.ts @@ -421,6 +421,135 @@ it("hands over promptly and runs conversations independently on one server", asy tags: expect.arrayContaining([["e", root, "", "reply"]]), }); }); +/** A second agent on the same runtime, with its own identity and publisher. */ +function second(f: ReturnType) { + const other = "9".repeat(64); + const base = f.delivery("unused").agent; + const agent = { + ...base, + pubkey: other, + name: "Other", + publish: vi.fn(async (event: unknown) => ({ + ...(event as object), + id: "8".repeat(64), + })), + } as unknown as Delivery["agent"]; + f.runtime.sync([{ pubkey } as Agent, { pubkey: other } as Agent]); + return { + agent, + delivery: (content: string) => ({ + ...f.delivery(content, "7".repeat(64)), + agent, + }), + }; +} +const contents = (publish: { mock: { calls: unknown[][] } }) => + publish.mock.calls.map(([event]) => (event as { content: string }).content); +const servers = (f: ReturnType) => + f.spawn.mock.calls.filter(([id]) => id === "app-server"); +it("serves every agent from one server and routes each tool call to its own agent", async () => { + const f = fixture(); + const b = second(f); + await f.runtime.run(f.delivery("@Codex first")); + await f.runtime.run(b.delivery("@Other second")); + await vi.waitFor(() => expect(f.starts()).toHaveLength(2)); + expect(servers(f)).toHaveLength(1); + expect(f.starts()[1]?.params).toMatchObject({ + additionalContext: { + "buzz.instructions": { value: expect.stringContaining("Other") }, + }, + }); + await f.complete(1, "FROM OTHER"); + await f.complete(0, "FROM CODEX"); + await vi.waitFor(() => expect(f.publish).toHaveBeenCalledTimes(1)); + await vi.waitFor(() => expect(b.agent.publish).toHaveBeenCalledTimes(1)); + expect(contents(f.publish)).toEqual(["FROM CODEX"]); + expect(contents(b.agent.publish as never)).toEqual(["FROM OTHER"]); + expect(f.runtime.sessions("9".repeat(64))[0]?.status).toBe("Idle"); +}); +it("stops a removed agent's turn without ending the server the others share", async () => { + const f = fixture(); + const b = second(f); + await f.runtime.run(f.delivery("work")); + await f.runtime.run(b.delivery("other work")); + await vi.waitFor(() => expect(f.starts()).toHaveLength(2)); + const removed = (f.starts()[1]?.params as { threadId: string } | undefined) + ?.threadId; + f.runtime.sync([{ pubkey } as Agent]); + await vi.waitFor(() => + expect( + f.sent.some( + (wire) => + wire.method === "thread/unsubscribe" && + (wire.params as { threadId: string }).threadId === removed, + ), + ).toBe(true), + ); + for (const method of ["turn/interrupt", "thread/backgroundTerminals/clean"]) + expect(f.sent.find((wire) => wire.method === method)?.params).toMatchObject( + { threadId: removed }, + ); + expect(f.process.end).not.toHaveBeenCalled(); + expect(f.runtime.sessions("9".repeat(64))).toHaveLength(0); + await f.complete(0, "STILL HERE"); + await vi.waitFor(() => expect(f.publish).toHaveBeenCalledTimes(1)); + expect(contents(f.publish)).toEqual(["STILL HERE"]); + expect(b.agent.publish).not.toHaveBeenCalled(); + f.runtime.sync([]); + await f.exits.promise; + expect(f.process.end).toHaveBeenCalled(); +}); +it("ends a stopped agent's leftover shells and lets it start again only after letting go", async () => { + const f = fixture(); + const b = second(f); + await f.runtime.run(f.delivery("work")); + await f.runtime.run(b.delivery("start a dev server")); + await vi.waitFor(() => expect(f.starts()).toHaveLength(2)); + const thread = (f.starts()[1]?.params as { threadId: string } | undefined) + ?.threadId; + await f.complete(1, "SERVER RUNNING"); + await vi.waitFor(() => expect(b.agent.publish).toHaveBeenCalledTimes(1)); + await f.runtime.run(b.delivery("busy again")); + await vi.waitFor(() => expect(f.starts()).toHaveLength(3)); + const index = (method: string) => + f.sent + .map( + (wire) => + wire.method === method && + (wire.params as { threadId: string }).threadId === thread, + ) + .lastIndexOf(true); + const before = index("thread/unsubscribe"); + f.runtime.sync([{ pubkey } as Agent]); + f.runtime.sync([{ pubkey } as Agent, { pubkey: "9".repeat(64) } as Agent]); + await f.runtime.run(b.delivery("back")); + await vi.waitFor(() => expect(f.starts()).toHaveLength(4)); + expect(index("thread/unsubscribe")).toBeGreaterThan(before); + expect(index("thread/backgroundTerminals/clean")).toBeGreaterThan( + index("thread/unsubscribe"), + ); + expect(index("thread/resume")).toBeGreaterThan( + index("thread/backgroundTerminals/clean"), + ); + expect(f.process.end).not.toHaveBeenCalled(); + await f.complete(3, "BACK"); + await f.complete(0, "STILL HERE"); + await vi.waitFor(() => expect(f.publish).toHaveBeenCalledTimes(1)); + await vi.waitFor(() => expect(b.agent.publish).toHaveBeenCalledTimes(2)); + expect(contents(b.agent.publish as never)).toEqual([ + "SERVER RUNNING", + "BACK", + ]); +}); +it("reopens the shared server after it exits", async () => { + const f = fixture(); + await f.runtime.run(f.delivery("work")); + await vi.waitFor(() => expect(f.starts()).toHaveLength(1)); + f.exits.resolve(3); + await vi.waitFor(() => expect(f.publish).toHaveBeenCalledTimes(1)); + await f.runtime.run(f.delivery("again")); + await vi.waitFor(() => expect(servers(f)).toHaveLength(2)); +}); it("serializes owner follow-ups and routes tool replies to the latest accepted request", async () => { const f = fixture(); await f.runtime.run(f.delivery("first")); diff --git a/src/bundled/codex/runtime.ts b/src/bundled/codex/runtime.ts index 27c29768f..7f0f9cef7 100644 --- a/src/bundled/codex/runtime.ts +++ b/src/bundled/codex/runtime.ts @@ -9,6 +9,7 @@ import { appClient } from "../../buzz-mcp/app-client"; import { TOOLS, callTool } from "../../buzz-mcp/tools"; import { config, + defaults, absoluteWorkspace, validWorkspace, type Config, @@ -42,6 +43,8 @@ type Request = { }; type Saved = { threadId: string; workspace: string }; type Active = { + /** The server this turn runs on; steering must reach the same process. */ + rpc: AppServer; threadId?: string; turnId?: string; ready: Promise; @@ -60,11 +63,17 @@ type Lane = { type Entry = { storageKey: string; abort: AbortController; - rpc: AppServer; - opening: Promise; + /** Settles once the same agent's previous run has let go of its threads. */ + after: Promise; lanes: Map; saved: Record; }; +type Server = { + rpc: AppServer; + opening: Promise; + /** Threads this process has loaded, and the agent each belongs to. */ + threads: Map; +}; export type SessionView = { key: string; status: string; detail: string }; const input = (text: string) => [{ type: "text", text, text_elements: [] }]; const message = (error: unknown) => @@ -76,9 +85,13 @@ function textOf(event: EventData, name: string) { } /** Agents2 owns admission/identity. This plugin owns work after hand-over, and - * uses one app-server per agent with independent Codex threads per conversation. */ + * uses one app-server for every agent. Each conversation is its own Codex + * thread carrying its agent's workspace, instructions and tools, and tool + * calls find their agent by thread and turn. */ export class CodexRuntime { private entries = new Map(); + private server: Server | undefined; + private stopping = new Map>(); private views = new Map(); private listeners = new Set<() => void>(); constructor( @@ -96,6 +109,8 @@ export class CodexRuntime { sessions = (pubkey: string): readonly SessionView[] => this.views.get(pubkey) ?? EMPTY; private status(pubkey: string, key: string, status: string, detail = "") { + // A stopped agent's draining turns don't bring its sessions back. + if (!this.entries.has(pubkey)) return; const others = this.sessions(pubkey).filter((view) => view.key !== key); this.views.set(pubkey, [ ...others.slice(-31), @@ -108,19 +123,68 @@ export class CodexRuntime { for (const [pubkey, entry] of this.entries) if (!present.has(pubkey)) { entry.abort.abort(); - void entry.rpc.close(); this.entries.delete(pubkey); this.views.delete(pubkey); + if (this.server) this.stop(this.server, pubkey, entry); } + if (!this.entries.size) this.close(); for (const listener of this.listeners) listener(); } + /** Other agents share the server, so a stopped agent's turns interrupt + * themselves. Then the shells its finished turns left are ended, which its + * own process used to do by exiting. Until then, the same agent starting + * again waits rather than resuming a thread that is still letting go. */ + private stop(server: Server, pubkey: string, entry: Entry) { + const stopped = (async () => { + await Promise.all( + [...entry.lanes.values()].map((lane) => + lane.tail.catch(() => undefined), + ), + ); + for (const [threadId, owner] of server.threads) + if (owner === pubkey) { + server.threads.delete(threadId); + await server.rpc + .request("thread/backgroundTerminals/clean", { threadId }) + .catch(() => undefined); + } + })(); + this.stopping.set(pubkey, stopped); + void stopped.finally(() => { + if (this.stopping.get(pubkey) === stopped) this.stopping.delete(pubkey); + }); + } dispose() { - for (const entry of this.entries.values()) { - entry.abort.abort(); - void entry.rpc.close(); - } + for (const entry of this.entries.values()) entry.abort.abort(); this.entries.clear(); this.views.clear(); + this.stopping.clear(); + this.close(); + } + private close() { + void this.server?.rpc.close(); + this.server = undefined; + } + private open() { + if (this.server) return this.server; + const rpc = new AppServer((call) => this.tool(call)); + // Threads set their own cwd; the process just needs a directory that exists. + const server: Server = { + rpc, + opening: rpc.open(this.spawn, defaults.workspace), + threads: new Map(), + }; + // An exit fails current work and permits the next ordinary mention to reopen. + rpc.onMessage((wire) => { + if (wire.method === "connection/closed" && this.server === server) + this.server = undefined; + }); + void server.opening.catch(() => { + if (this.server === server) this.server = undefined; + void rpc.close(); + }); + this.server = server; + return server; } private entry(request: Request) { const pubkey = request.agent.pubkey; @@ -143,28 +207,13 @@ export class CodexRuntime { } catch { /* Start fresh after malformed local data. */ } - const rpc = new AppServer((call) => this.tool(entry, call)); const entry: Entry = { storageKey: this.storageKey(request), abort: new AbortController(), - rpc, + after: this.stopping.get(pubkey) ?? Promise.resolve(), saved, lanes: new Map(), - opening: rpc.open(this.spawn, request.settings.workspace), }; - // An exit fails current work and permits the next ordinary mention to reopen. - rpc.onMessage((wire) => { - if ( - wire.method === "connection/closed" && - this.entries.get(pubkey) === entry - ) { - this.entries.delete(pubkey); - } - }); - void entry.opening.catch(() => { - if (this.entries.get(pubkey) === entry) this.entries.delete(pubkey); - void rpc.close(); - }); this.entries.set(pubkey, entry); return entry; } @@ -233,7 +282,7 @@ export class CodexRuntime { const entry = this.entry(request); let lane = entry.lanes.get(key); if (!lane) { - lane = { tail: Promise.resolve() }; + lane = { tail: entry.after }; entry.lanes.set(key, lane); } if (lane.active) { @@ -253,7 +302,7 @@ export class CodexRuntime { this.start(request, text, key, lane, entry); return; } - await entry.rpc.request("turn/steer", { + await active.rpc.request("turn/steer", { threadId: active.threadId, expectedTurnId: active.turnId, input: input( @@ -327,8 +376,9 @@ export class CodexRuntime { ], }); } - private async tool(entry: Entry, call: ToolCall): Promise { - const active = [...entry.lanes.values()] + private async tool(call: ToolCall): Promise { + const active = [...this.entries.values()] + .flatMap((entry) => [...entry.lanes.values()]) .map((lane) => lane.active) .find( (active) => @@ -447,12 +497,14 @@ export class CodexRuntime { let ready!: () => void; let notReady!: (error: unknown) => void; const turnAbort = new AbortController(); + const server = this.open(); const signal = AbortSignal.any([ entry.abort.signal, turnAbort.signal, AbortSignal.timeout(29 * 60_000), ]); const active: Active = { + rpc: server.rpc, ready: new Promise((resolve, reject) => { ready = resolve; notReady = reject; @@ -476,7 +528,7 @@ export class CodexRuntime { void completed.catch(() => undefined); const abort = () => fail(signal.reason); signal.addEventListener("abort", abort, { once: true }); - const off = entry.rpc.onMessage((wire) => { + const off = server.rpc.onMessage((wire) => { const p = wire.params as | { threadId?: string; @@ -517,7 +569,7 @@ export class CodexRuntime { } }); try { - const rpc = await entry.opening; + const rpc = await server.opening; signal.throwIfAborted(); // The process host expands ~/.buzz and creates it; Codex's RPC needs // an absolute path rather than shell-style home notation. @@ -610,6 +662,7 @@ export class CodexRuntime { }); } active.threadId = started.thread.id; + server.threads.set(active.threadId, agent.pubkey); signal.throwIfAborted(); await rpc.request("thread/name/set", { threadId: active.threadId, @@ -659,23 +712,21 @@ export class CodexRuntime { this.status(agent.pubkey, key, "Idle"); } } catch (error) { + const ended = active.finished; active.finished = true; notReady(error); await active.steering; - if ( - signal.aborted && - active.threadId && - active.turnId && - !entry.abort.signal.aborted - ) { + // Other agents share the server, so even a stopped agent interrupts + // its turn rather than ending the process. + if (signal.aborted && !ended && active.threadId && active.turnId) { try { - await entry.rpc.request("turn/interrupt", { + await server.rpc.request("turn/interrupt", { threadId: active.threadId, turnId: active.turnId, }); - await this.terminals(entry.rpc, active.threadId); + await this.terminals(server.rpc, active.threadId); } catch { - await entry.rpc.close(); + await server.rpc.close(); } } if (!entry.abort.signal.aborted) { @@ -691,10 +742,10 @@ export class CodexRuntime { off(); // Release the subscription before the next resume: loaded-thread rejoin // ignores instruction/config overrides while a client is subscribed. - if (active.threadId && !entry.abort.signal.aborted) - await entry.rpc + if (active.threadId) + await server.rpc .request("thread/unsubscribe", { threadId: active.threadId }) - .catch(() => entry.rpc.close()); + .catch(() => server.rpc.close()); if (lane.active === active) delete lane.active; } }