One bridge process per provider artifact; delete the codex per-thread lane (WS4 L3)

apps/host-daemon/src/command-router.ts

@@ -74,7 +74,6 @@ interface ProviderExecutionLane {
 interface ProviderProcessLaneKeyArgs {
   environmentId: string;
   providerId: string | null;
-  threadId: string;
 }
 
 interface CreateProviderExecutionLaneArgs extends ProviderProcessLaneKeyArgs {
@@ -122,7 +121,6 @@ export interface CommandRouterOptions {
 }
 
 const HOST_COMMAND_LIFECYCLE_LOG_THRESHOLD_MS = 1_000;
-const CODEX_PROVIDER_ID = "codex";
 
 function elapsedMs(startedAtMs: number): number {
   return performance.now() - startedAtMs;
@@ -528,14 +526,13 @@ export class CommandRouter {
   }
 
   private getProviderProcessLaneKey(args: ProviderProcessLaneKeyArgs): string {
-    // Legacy or thread.stop paths can lack provider ownership. Bucket them
-    // together per environment so unknown ownership stays conservative without
-    // serializing unrelated environments.
+    // One process lane per provider per environment, mirroring the runtime's
+    // one process per provider artifact. Legacy or thread.stop paths can lack
+    // provider ownership. Bucket them together per environment so unknown
+    // ownership stays conservative without serializing unrelated
+    // environments.
     const providerKey = args.providerId ?? "unknown-provider";
-    if (providerKey !== CODEX_PROVIDER_ID) {
-      return `${args.environmentId}\0${providerKey}`;
-    }
-    return `${args.environmentId}\0${providerKey}\0thread:${args.threadId}`;
+    return `${args.environmentId}\0${providerKey}`;
   }
 
   private getProviderSessionLaneKey(
@@ -551,7 +548,6 @@ export class CommandRouter {
     const processKey = this.getProviderProcessLaneKey({
       environmentId: args.environmentId,
       providerId: args.providerId,
-      threadId: args.threadId,
     });
     return {
       processKey,
@@ -579,7 +575,6 @@ export class CommandRouter {
       processMode,
       providerId: identity.providerId,
       sessionId,
-      threadId: identity.threadId,
     });
   }
 
@@ -665,31 +660,27 @@ export class CommandRouter {
           processMode: "read",
           providerId: command.providerId,
           sessionId: `thread:${command.threadId}`,
-          threadId: command.threadId,
         });
       case "turn.submit":
         return this.createProviderExecutionLane({
           environmentId: command.environmentId,
           processMode: "read",
           providerId: command.resumeContext.providerId,
           sessionId: `provider-thread:${command.resumeContext.providerThreadId}`,
-          threadId: command.threadId,
         });
       case "thread.archive":
         return this.createProviderExecutionLane({
           environmentId: command.environmentId,
           processMode: "read",
           providerId: command.providerId,
           sessionId: `provider-thread:${command.providerThreadId}`,
-          threadId: command.threadId,
         });
       case "interactive.resolve":
         return this.createProviderExecutionLane({
           environmentId: command.environmentId,
           processMode: "read",
           providerId: command.providerId,
           sessionId: `provider-thread:${command.providerThreadId}`,
-          threadId: command.threadId,
         });
       case "thread.stop":
       case "thread.plan.cancel": {

apps/host-daemon/test/command/command-router.test.ts

@@ -428,7 +428,11 @@ describe("CommandRouter", () => {
     await runtimeManager.shutdownAll();
   });
 
-  it("does not route separate codex threads through one provider process lane", async () => {
+  it("routes every codex thread through the provider's one process lane", async () => {
+    // One process per provider artifact (the codex bridge supervises its own
+    // per-thread children): a thread.stop holds the codex process's write
+    // lane for the length of its dispatch, and another codex thread's
+    // turn.submit waits on it, exactly as for every other provider.
     const harness = createHarness({ workspacePath: "/tmp/env-router" });
     await harness.manager.ensureEnvironment({
       environmentId: "env-router",
@@ -477,12 +481,13 @@ describe("CommandRouter", () => {
     });
     await flushAsyncWork();
 
-    expect(harness.runtimeState.ranTurnText).toBe("codex other thread");
-    const turnResponse = await turnTask;
-    expect(turnResponse.ok).toBe(true);
+    expect(harness.runtimeState.ranTurnText).not.toBe("codex other thread");
 
     releaseStop.resolve();
     const stopResponse = await stopTask;
     expect(stopResponse.ok).toBe(true);
+    const turnResponse = await turnTask;
+    expect(turnResponse.ok).toBe(true);
+    expect(harness.runtimeState.ranTurnText).toBe("codex other thread");
   });
 });

packages/agent-runtime/src/runtime.codex-topology.test.ts

@@ -0,0 +1,258 @@
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { fileURLToPath } from "node:url";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import type { ThreadEvent } from "@bb/domain";
+import { createAgentRuntime } from "./runtime.js";
+import {
+  createScriptedEchoLaunch,
+  createScriptedEchoRequestRecord,
+  fullRuntimeOptions,
+  waitForRuntimeState,
+  waitForThreadAgentMessageText,
+  withBridgeLaunch,
+} from "./test/runtime-test-harness.js";
+import { promptTextInput } from "./test/prompt-input.js";
+import type { AgentRuntime, AgentRuntimeBridgeLaunch } from "./types.js";
+
+/**
+ * Process topology after L3: the runtime runs ONE bridge process per provider
+ * artifact, and the codex bridge supervises one `codex app-server` child per
+ * live thread underneath it. This drives the real codex bridge module through
+ * the runtime (the way the daemon does) with the fake app-server as its child,
+ * and counts children through the fake's process log.
+ */
+
+const codexBridgeModulePath = fileURLToPath(
+  new URL(
+    "../../../plugins/provider-codex/src/bridge/bridge.ts",
+    import.meta.url,
+  ),
+);
+const fakeAppServerPath = fileURLToPath(
+  new URL(
+    "../../../plugins/provider-codex/src/bridge/fake-codex-app-server.mjs",
+    import.meta.url,
+  ),
+);
+
+interface CodexTopologyRuntime {
+  events: ThreadEvent[];
+  runtime: AgentRuntime;
+  /** App-server children the fake logged as spawned. */
+  spawned(): number;
+  /** App-server children the fake logged as exited. */
+  exited(): number;
+  /** Distinct bridge processes that spawned a child (parent pids). */
+  bridges(): number;
+  /** Bridge process exits the runtime reported. */
+  bridgeExits: { expected: boolean }[];
+  launch(digest: string): AgentRuntimeBridgeLaunch;
+}
+
+describe("codex process topology", () => {
+  let workspaceDir: string;
+  const runtimes: AgentRuntime[] = [];
+
+  beforeEach(() => {
+    workspaceDir = mkdtempSync(join(tmpdir(), "bb-codex-topology-"));
+  });
+
+  afterEach(async () => {
+    await Promise.all(runtimes.splice(0).map((runtime) => runtime.shutdown()));
+    rmSync(workspaceDir, { recursive: true, force: true });
+  });
+
+  function createCodexTopologyRuntime(): CodexTopologyRuntime {
+    const processLogPath = join(workspaceDir, "app-server-processes.log");
+    const scriptPath = join(workspaceDir, "fake-codex-script.json");
+    writeFileSync(
+      scriptPath,
+      JSON.stringify({
+        processLogPath,
+        archiveStatePath: join(workspaceDir, "fake-codex-archived.json"),
+      }),
+    );
+    const events: ThreadEvent[] = [];
+    const bridgeExits: { expected: boolean }[] = [];
+    const record = createScriptedEchoRequestRecord();
+    const launch = (digest: string): AgentRuntimeBridgeLaunch =>
+      createScriptedEchoLaunch({
+        pluginId: "provider-codex",
+        digest,
+        modulePath: codexBridgeModulePath,
+      });
+    const runtime = withBridgeLaunch(
+      createAgentRuntime({
+        workspacePath: workspaceDir,
+        env: {
+          ...record.env,
+          BB_CODEX_BRIDGE_APP_SERVER_COMMAND: process.execPath,
+          BB_CODEX_BRIDGE_APP_SERVER_ARGS: JSON.stringify([
+            fakeAppServerPath,
+            scriptPath,
+          ]),
+        },
+        onEvent: (event) => events.push(event),
+        onProcessExit: (info) => bridgeExits.push({ expected: info.expected }),
+        onToolCall: async () => ({ contentItems: [], success: true }),
+      }),
+      launch("codex-v1"),
+    );
+    runtimes.push(runtime);
+    const readLog = (): string[] => {
+      try {
+        return readFileSync(processLogPath, "utf8")
+          .split("\n")
+          .filter((line) => line.length > 0);
+      } catch {
+        return [];
+      }
+    };
+    const spawnLines = (): string[] =>
+      readLog().filter((line) => line.startsWith("spawn:"));
+    return {
+      events,
+      runtime,
+      spawned: () => spawnLines().length,
+      exited: () => readLog().filter((line) => line.startsWith("exit:")).length,
+      bridges: () => new Set(spawnLines().map((line) => line.split(":")[2])).size,
+      bridgeExits,
+      launch,
+    };
+  }
+
+  /** The bb threads the runtime hosts, out of the ones this test starts. */
+  function hosted(runtime: AgentRuntime): string[] {
+    return ["t1", "t2", "t3", "t4"].filter((threadId) =>
+      runtime.hasThread(threadId),
+    );
+  }
+
+  async function startCodexThread(
+    runtime: AgentRuntime,
+    threadId: string,
+    bridgeLaunch?: AgentRuntimeBridgeLaunch,
+  ): Promise<string> {
+    const { providerThreadId } = await runtime.startThread({
+      ...(bridgeLaunch === undefined ? {} : { bridgeLaunch }),
+      environmentId: "env-1",
+      projectId: "p1",
+      providerId: "codex",
+      threadId,
+      options: fullRuntimeOptions,
+    });
+    return providerThreadId;
+  }
+
+  it("runs N codex threads on one bridge process with one app-server child each, and reaps the children on stop, archive, and bridge retirement", async () => {
+    const topology = createCodexTopologyRuntime();
+    const { runtime, events } = topology;
+
+    await startCodexThread(runtime, "t1");
+    await startCodexThread(runtime, "t2");
+    await startCodexThread(runtime, "t3");
+
+    // N concurrent threads: ONE bridge process, N app-server children.
+    expect(runtime.listRunningProviders()).toEqual(["codex"]);
+    expect(hosted(runtime)).toEqual(["t1", "t2", "t3"]);
+    expect(topology.spawned()).toBe(3);
+    expect(topology.bridges()).toBe(1);
+    expect(topology.exited()).toBe(0);
+
+    // A turn runs on one of them without touching the others.
+    await runtime.runTurn({
+      clientRequestId: "creq_cdxtpgy222",
+      threadId: "t2",
+      input: [promptTextInput({ text: "hello" })],
+      options: fullRuntimeOptions,
+    });
+    await waitForThreadAgentMessageText({
+      events,
+      providerId: "codex",
+      runtime,
+      text: "hello from codex turn",
+      threadId: "t2",
+    });
+    expect(topology.spawned()).toBe(3);
+
+    // thread/stop (release): that thread's child dies; the bridge and the
+    // other children stay.
+    await runtime.stopThread({ threadId: "t1" });
+    await waitForRuntimeState({
+      label: "t1's app-server child exited",
+      predicate: () => topology.exited() === 1,
+      timeoutMs: 5_000,
+    });
+    expect(runtime.listRunningProviders()).toEqual(["codex"]);
+    expect(hosted(runtime)).toEqual(["t2", "t3"]);
+
+    // thread/archive: the same for the archived thread.
+    const session2 = runtime.getProviderSession("t2");
+    if (!session2) throw new Error("expected a codex session for t2");
+    await runtime.archiveThread({
+      providerId: "codex",
+      providerThreadId: session2.providerThreadId,
+      threadId: "t2",
+    });
+    await waitForRuntimeState({
+      label: "t2's app-server child exited",
+      predicate: () => topology.exited() === 2,
+      timeoutMs: 5_000,
+    });
+    expect(runtime.listRunningProviders()).toEqual(["codex"]);
+    expect(hosted(runtime)).toEqual(["t3"]);
+    expect(topology.spawned()).toBe(3);
+
+    // A plugin update ships a new artifact: the next thread starts on a new
+    // bridge process (its own child), the old bridge keeps serving t3 until
+    // t3 is released, then retires — and its last child dies with it.
+    const v2 = topology.launch("codex-v2");
+    await startCodexThread(runtime, "t4", v2);
+    expect(topology.spawned()).toBe(4);
+    expect(topology.bridges()).toBe(2);
+    expect(topology.exited()).toBe(2);
+    expect(topology.bridgeExits).toEqual([]);
+
+    await runtime.stopThread({ threadId: "t3" });
+    await waitForRuntimeState({
+      label: "the superseded bridge and t3's child exited",
+      predicate: () =>
+        topology.exited() === 3 && topology.bridgeExits.length === 1,
+      timeoutMs: 10_000,
+    });
+    expect(topology.bridgeExits).toEqual([{ expected: true }]);
+    expect(hosted(runtime)).toEqual(["t4"]);
+    expect(topology.spawned()).toBe(4);
+
+    // The archived session is still resumable on the current bridge once it
+    // is unarchived. The unarchive has no live child to use, so the bridge
+    // runs it on a one-shot maintenance child (spawned and reaped); the
+    // resume then gets a fresh child of its own.
+    await runtime.unarchiveThread({
+      providerId: "codex",
+      providerThreadId: session2.providerThreadId,
+      threadId: "t2",
+      bridgeLaunch: v2,
+    });
+    await runtime.resumeThread({
+      bridgeLaunch: v2,
+      environmentId: "env-1",
+      projectId: "p1",
+      providerId: "codex",
+      providerThreadId: session2.providerThreadId,
+      threadId: "t2",
+      options: fullRuntimeOptions,
+    });
+    expect(hosted(runtime)).toEqual(["t2", "t4"]);
+    await waitForRuntimeState({
+      label: "the maintenance child was reaped",
+      predicate: () => topology.exited() === 4,
+      timeoutMs: 5_000,
+    });
+    expect(topology.spawned()).toBe(6);
+    expect(topology.bridges()).toBe(2);
+    expect(topology.bridgeExits).toHaveLength(1);
+  }, 60_000);
+});