perf(codex): one app-server for every agent

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.
 

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;

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();

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<typeof fixture>) {
+  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<typeof fixture>) =>
+  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"));