diff --git a/apps/host-daemon/src/command-router.ts b/apps/host-daemon/src/command-router.ts index a40cdffa00..3f2790aebf 100644 --- a/apps/host-daemon/src/command-router.ts +++ b/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,7 +660,6 @@ export class CommandRouter { processMode: "read", providerId: command.providerId, sessionId: `thread:${command.threadId}`, - threadId: command.threadId, }); case "turn.submit": return this.createProviderExecutionLane({ @@ -673,7 +667,6 @@ export class CommandRouter { processMode: "read", providerId: command.resumeContext.providerId, sessionId: `provider-thread:${command.resumeContext.providerThreadId}`, - threadId: command.threadId, }); case "thread.archive": return this.createProviderExecutionLane({ @@ -681,7 +674,6 @@ export class CommandRouter { processMode: "read", providerId: command.providerId, sessionId: `provider-thread:${command.providerThreadId}`, - threadId: command.threadId, }); case "interactive.resolve": return this.createProviderExecutionLane({ @@ -689,7 +681,6 @@ export class CommandRouter { processMode: "read", providerId: command.providerId, sessionId: `provider-thread:${command.providerThreadId}`, - threadId: command.threadId, }); case "thread.stop": case "thread.plan.cancel": { diff --git a/apps/host-daemon/test/command/command-router.test.ts b/apps/host-daemon/test/command/command-router.test.ts index c791e48fd9..81126870f2 100644 --- a/apps/host-daemon/test/command/command-router.test.ts +++ b/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"); }); }); diff --git a/packages/agent-runtime/src/runtime.codex-topology.test.ts b/packages/agent-runtime/src/runtime.codex-topology.test.ts new file mode 100644 index 0000000000..a23169a7ab --- /dev/null +++ b/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 { + 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); +}); diff --git a/packages/agent-runtime/src/runtime.process-lifecycle.test.ts b/packages/agent-runtime/src/runtime.process-lifecycle.test.ts index d541752114..42ace21d35 100644 --- a/packages/agent-runtime/src/runtime.process-lifecycle.test.ts +++ b/packages/agent-runtime/src/runtime.process-lifecycle.test.ts @@ -728,7 +728,7 @@ describe("createAgentRuntime process lifecycle", () => { return { processLog, runtime }; } - it("runs each codex thread on a separate provider process", async () => { + it("runs every codex thread on one provider process", async () => { const events: ThreadEvent[] = []; const { processLog, runtime } = createCodexRuntime({ events }); await runtime.startThread({ @@ -745,13 +745,12 @@ describe("createAgentRuntime process lifecycle", () => { providerId: "codex", options: fullRuntimeOptions, }); - await waitForRuntimeState({ - label: "two codex provider processes spawned", - predicate: () => - processLog.read().filter((line) => line.startsWith("spawn:")).length === - 2, - runtime, - }); + // One process per provider artifact: the bridge supervises its own + // per-thread children (see runtime.codex-topology.test.ts for the real + // codex bridge); the runtime never scopes a process to a thread. + expect( + processLog.read().filter((line) => line.startsWith("spawn:")), + ).toHaveLength(1); const firstSession = runtime.getProviderSession("t1"); const secondSession = runtime.getProviderSession("t2"); if (!firstSession || !secondSession) { @@ -788,8 +787,8 @@ describe("createAgentRuntime process lifecycle", () => { text: "second", threadId: "t2", }); - // Each answer carries the pid of the process that served it: two - // different processes. + // Each answer carries the pid of the process that served it: the same + // process for both threads. const pidOf = (threadId: string): string | undefined => events .filter( @@ -803,16 +802,13 @@ describe("createAgentRuntime process lifecycle", () => { ) .find((pid) => pid !== undefined); expect(pidOf("t1")).toBeDefined(); - expect(pidOf("t1")).not.toBe(pidOf("t2")); + expect(pidOf("t1")).toBe(pidOf("t2")); + // Stopping one thread releases its session; the process stays up for + // the other thread and for the provider's next thread. await runtime.stopThread({ threadId: "t1" }); - await waitForRuntimeState({ - label: "one codex provider process exited after stopping one thread", - predicate: () => - processLog.read().filter((line) => line.startsWith("exit:")).length === - 1, - runtime, - }); + expect(runtime.hasThread("t1")).toBe(false); + expect(runtime.listRunningProviders()).toEqual(["codex"]); await runtime.runTurn({ clientRequestId: "creq_2222222252", threadId: "t2", @@ -826,10 +822,13 @@ describe("createAgentRuntime process lifecycle", () => { text: "still alive", threadId: "t2", }); + expect( + processLog.read().filter((line) => line.startsWith("exit:")), + ).toHaveLength(0); await runtime.shutdown(); }); - it("stops a thread-scoped codex process when session construction fails", async () => { + it("keeps the codex provider process when one session construction fails", async () => { const events: ThreadEvent[] = []; const { processLog, runtime } = createCodexRuntime({ events, @@ -848,22 +847,13 @@ describe("createAgentRuntime process lifecycle", () => { options: fullRuntimeOptions, }), ).rejects.toThrow("no rollout found"); - // The process the failed construction spawned does not linger. - await waitForRuntimeState({ - label: "thread-scoped codex process exited after failed construction", - predicate: () => - processLog.read().filter((line) => line.startsWith("exit:")).length === - 1, - runtime, - timeoutMs: 5_000, - }); expect(runtime.getProviderSession("t1")).toBeNull(); - await waitForRuntimeState({ - label: "no codex provider process left running", - predicate: () => runtime.listRunningProviders().length === 0, - runtime, - timeoutMs: 5_000, - }); + // The process serves every codex thread in the environment, so one + // failed construction does not take it down. + expect(runtime.listRunningProviders()).toEqual(["codex"]); + expect( + processLog.read().filter((line) => line.startsWith("exit:")), + ).toHaveLength(0); await runtime.shutdown(); }); @@ -944,7 +934,7 @@ describe("createAgentRuntime process lifecycle", () => { } }); - it("reaps a codex thread process after a terminal provider error before turn start", async () => { + it("reaps a codex session after a terminal provider error before turn start", async () => { const events: ThreadEvent[] = []; const { processLog, runtime } = createCodexRuntime({ events }); try { @@ -991,19 +981,18 @@ describe("createAgentRuntime process lifecycle", () => { threadId: "t1", }), ]); - await waitForRuntimeState({ - label: "reaped codex process exited", - predicate: () => - processLog.read().filter((line) => line.startsWith("exit:")) - .length === 1, - runtime, - }); + // The session is released (thread/stop), the provider process stays. + expect(runtime.getProviderSession("t1")).toBeNull(); + expect(runtime.listRunningProviders()).toEqual(["codex"]); + expect( + processLog.read().filter((line) => line.startsWith("exit:")), + ).toHaveLength(0); } finally { await runtime.shutdown(); } }); - it("reaps an idle codex thread process and resumes it later", async () => { + it("reaps an idle codex session and resumes it later on the same process", async () => { const events: ThreadEvent[] = []; const { processLog, runtime } = createCodexRuntime({ events }); try { @@ -1059,15 +1048,10 @@ describe("createAgentRuntime process lifecycle", () => { threadId: "t1", }); expect(reapedSession.idleForMs).toBeGreaterThanOrEqual(30 * 60 * 1000); - await waitForRuntimeState({ - label: "reaped codex process exited", - predicate: () => - processLog.read().filter((line) => line.startsWith("exit:")) - .length === 1, - runtime, - }); expect(runtime.hasThread("t1")).toBe(false); expect(runtime.getProviderSession("t1")).toBeNull(); + // The session was released on the provider process, which stays up. + expect(runtime.listRunningProviders()).toEqual(["codex"]); await runtime.resumeThread({ environmentId: "env-1", @@ -1090,12 +1074,13 @@ describe("createAgentRuntime process lifecycle", () => { text: "after reap", threadId: "t1", }); + // Resumed on the same process: one spawn, no exit, one thread/resume. const logLines = processLog.read(); expect(logLines.filter((line) => line.startsWith("spawn:"))).toHaveLength( - 2, + 1, ); expect(logLines.filter((line) => line.startsWith("exit:"))).toHaveLength( - 1, + 0, ); expect( logLines.some( diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index e9ea85c113..c852f90730 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -135,7 +135,6 @@ interface ResolveProviderProcessKeyArgs { acpLaunchSpec?: HostDaemonAcpLaunchSpec; bridgeLaunch?: AgentRuntimeBridgeLaunch; providerId: string; - threadId?: string; } interface ArchiveOrUnarchiveThreadArgs { @@ -275,12 +274,12 @@ interface RequireProviderRequestPlanArgs { } /** - * The codex per-thread process lane (#130). Still load-bearing for the idle - * reap (`isThreadScopedCodexProcess`); the process-topology layer (L3) - * collapses it to one process per provider artifact. + * The one provider id the pre-experiment idle reap releases (the behavior + * bb shipped before `providerSessionReapingEnabled` extended release to every + * restorable provider). Product policy, not a process-topology fact: one + * bridge process serves every thread of a provider in the environment. */ const CODEX_PROVIDER_ID = "codex"; -const CODEX_THREAD_PROCESS_KEY_PREFIX = `${CODEX_PROVIDER_ID}\0thread:`; const THREAD_CREATION_REQUEST_TIMEOUT_MS = 2 * 60_000; async function delay(ms: number): Promise { @@ -405,19 +404,16 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { }); /** - * Codex runs one provider process per thread. The codex bridge now owns a - * per-thread `codex app-server` child internally, so this outer scoping is - * redundant for isolation — only the pre-experiment idle reap still keys - * off `isThreadScopedCodexProcess`. The process-topology layer (L3) - * collapses it to one process per provider artifact. + * One process per provider artifact: every thread of a provider in this + * environment runs on the same bridge process, and the bridge supervises + * whatever children it needs (the codex bridge runs one `codex app-server` + * per thread underneath itself). The runtime never scopes a process to a + * thread. */ function resolveProviderProcessKey( args: ResolveProviderProcessKeyArgs, ): string { - const baseKey = - args.providerId !== CODEX_PROVIDER_ID || args.threadId === undefined - ? args.providerId - : `${CODEX_THREAD_PROCESS_KEY_PREFIX}${args.threadId}`; + const baseKey = args.providerId; // A plugin-delivered bridge keys process identity by its artifact hash AND // by the declaration facts baked into the adapter at spawn (capabilities, // static provider options): a plugin can change either one alone, and @@ -441,32 +437,16 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { return providerProcesses.requireProviderProcess({ processKey, providerId }); } - function isThreadScopedCodexProcess(proc: ProviderProcess): boolean { - return ( - proc.providerId === CODEX_PROVIDER_ID && - proc.processKey.startsWith(CODEX_THREAD_PROCESS_KEY_PREFIX) - ); - } - /** * Releasing a thread is the moment a process can become retirable: a - * thread-scoped codex process has nothing left to serve, and a bridge - * process superseded by a plugin update was only being kept alive by the - * threads still running on it. + * bridge process superseded by a plugin update was only being kept alive + * by the threads still running on it. A current process stays up for the + * provider's next thread; its own per-thread children are the bridge's + * business (the codex bridge kills a thread's app-server on release). */ async function releaseIdleProviderProcess( proc: ProviderProcess, ): Promise { - if ( - isThreadScopedCodexProcess(proc) && - proc.identity.threadIds.size === 0 - ) { - await providerProcesses.shutdownProvider({ - processKey: proc.processKey, - providerId: proc.providerId, - }); - return; - } await providerProcesses.retireSupersededBridgeProcessIfIdle(proc); } @@ -1125,7 +1105,6 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { resolveProviderProcessKey({ ...(bridgeLaunch !== undefined ? { bridgeLaunch } : {}), providerId, - threadId, }); await providerProcesses.ensureProvider({ processKey, @@ -1578,18 +1557,12 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { } const runtime: AgentRuntime = { - async ensureProvider({ - providerId, - forThreadId, - acpLaunchSpec, - bridgeLaunch, - }) { + async ensureProvider({ providerId, acpLaunchSpec, bridgeLaunch }) { await providerProcesses.ensureProvider({ processKey: resolveProviderProcessKey({ ...(acpLaunchSpec !== undefined ? { acpLaunchSpec } : {}), ...(bridgeLaunch !== undefined ? { bridgeLaunch } : {}), providerId, - ...(forThreadId !== undefined ? { threadId: forThreadId } : {}), }), providerId, ...(acpLaunchSpec !== undefined ? { acpLaunchSpec } : {}), @@ -1621,11 +1594,9 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { ...(acpLaunchSpec !== undefined ? { acpLaunchSpec } : {}), ...(bridgeLaunch !== undefined ? { bridgeLaunch } : {}), providerId, - threadId, }); await runtime.ensureProvider({ providerId, - forThreadId: threadId, ...(acpLaunchSpec !== undefined ? { acpLaunchSpec } : {}), ...(bridgeLaunch !== undefined ? { bridgeLaunch } : {}), }); @@ -1824,11 +1795,9 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { ...(acpLaunchSpec !== undefined ? { acpLaunchSpec } : {}), ...(bridgeLaunch !== undefined ? { bridgeLaunch } : {}), providerId, - threadId, }); await runtime.ensureProvider({ providerId, - forThreadId: threadId, ...(acpLaunchSpec !== undefined ? { acpLaunchSpec } : {}), ...(bridgeLaunch !== undefined ? { bridgeLaunch } : {}), }); @@ -2014,11 +1983,9 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { ...(acpLaunchSpec !== undefined ? { acpLaunchSpec } : {}), ...(bridgeLaunch !== undefined ? { bridgeLaunch } : {}), providerId, - threadId, }); await runtime.ensureProvider({ providerId, - forThreadId: threadId, ...(acpLaunchSpec !== undefined ? { acpLaunchSpec } : {}), ...(bridgeLaunch !== undefined ? { bridgeLaunch } : {}), }); @@ -2674,9 +2641,9 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { return null; } - let proc: ProviderProcess; try { - proc = providerProcesses.requireProviderProcess({ + // A session whose process is gone has nothing to release. + providerProcesses.requireProviderProcess({ processKey: candidate.runtimeConfig.processKey, providerId: candidate.runtimeConfig.providerId, }); @@ -2686,11 +2653,7 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { // Open background tasks and open delegations (a codex native // sub-agent still running, or still owed a followup turn) are // live provider work; reaping the session would destroy it. - if ( - providerSessionReapingEnabled - ? backgroundWorkState.hasOpenThreadWork(candidate.threadId) - : !isThreadScopedCodexProcess(proc) - ) { + if (backgroundWorkState.hasOpenThreadWork(candidate.threadId)) { return null; } diff --git a/packages/agent-runtime/src/types.ts b/packages/agent-runtime/src/types.ts index c65e2e46a7..6cc2fafbb8 100644 --- a/packages/agent-runtime/src/types.ts +++ b/packages/agent-runtime/src/types.ts @@ -216,12 +216,6 @@ export interface AgentRuntimeBridgeLaunch { export interface EnsureProviderArgs { acpLaunchSpec?: HostDaemonAcpLaunchSpec; bridgeLaunch?: AgentRuntimeBridgeLaunch; - /** - * Providers with thread-scoped processes use this to start the process for a - * specific bb thread. Omit it for provider-scoped maintenance work such as - * model listing. - */ - forThreadId?: string; providerId: string; } diff --git a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs index ce9e172445..66e24ae6af 100644 --- a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs +++ b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs @@ -24,6 +24,7 @@ */ import { + appendFileSync, closeSync, existsSync, openSync, @@ -156,6 +157,26 @@ const archiveStatePath = script?.archiveStatePath ?? null; */ let renameEmptyRolloutFailuresLeft = script?.renameEmptyRolloutFailures ?? 0; const archivedThreadIds = new Set(); +/** + * `processLogPath`: one line per child lifecycle step (`spawn::`, + * `exit::`), so a test can count how many app-server children the + * bridge runs, how many bridge processes spawned them (distinct ppids), and + * see the children die on release, archive, and bridge shutdown. + */ +const processLogPath = script?.processLogPath ?? null; + +function logProcessStep(step) { + if (processLogPath === null) { + return; + } + appendFileSync(processLogPath, `${step}:${process.pid}:${process.ppid}\n`); +} + +logProcessStep("spawn"); +process.on("SIGTERM", () => { + logProcessStep("exit"); + process.exit(0); +}); let scriptedTurnIndex = 0; function readArchivedThreadIds() { @@ -508,5 +529,6 @@ stdinLines.on("line", (line) => { } }); stdinLines.on("close", () => { + logProcessStep("exit"); process.exit(0); }); diff --git a/scripts/provider-literal-baseline.json b/scripts/provider-literal-baseline.json index d86dc91ad3..cb16b176f8 100644 --- a/scripts/provider-literal-baseline.json +++ b/scripts/provider-literal-baseline.json @@ -1,6 +1,6 @@ { "_comment": "Provider-literal ratchet (G1). Per-file occurrence count of provider-ID references in core. May only go DOWN. Regenerate: node scripts/check-provider-literal-ratchet.mjs --write. Delete this file and the guard when empty.", - "total": 142, + "total": 136, "files": { "apps/app/src/components/tools/SkillsCollection.tsx": 3, "apps/app/src/lib/provider-icon.ts": 15, @@ -9,7 +9,6 @@ "apps/host-daemon/scripts/bundle-manifest.mjs": 1, "apps/host-daemon/src/command-handlers/list-commands.ts": 20, "apps/host-daemon/src/command-handlers/list-skills.ts": 1, - "apps/host-daemon/src/command-router.ts": 3, "apps/host-daemon/src/injected-skills.ts": 3, "apps/server/src/routes/threads/actions.ts": 1, "apps/server/src/services/ai/host-daemon-ai-provider.ts": 1, @@ -25,7 +24,7 @@ "packages/agent-runtime/src/provider-catalog.ts": 2, "packages/agent-runtime/src/provider-registry.ts": 2, "packages/agent-runtime/src/runtime-skill-roots.ts": 5, - "packages/agent-runtime/src/runtime.ts": 6, + "packages/agent-runtime/src/runtime.ts": 3, "packages/agent-runtime/src/types.ts": 3, "packages/config/src/bb-app-managed-config.ts": 4, "packages/domain/src/provider-model-catalog.ts": 3,