diff --git a/packages/agent-runtime/src/runtime.codex-topology.test.ts b/packages/agent-runtime/src/runtime.codex-topology.test.ts index 0b0f629a4e..c0ec1e4059 100644 --- a/packages/agent-runtime/src/runtime.codex-topology.test.ts +++ b/packages/agent-runtime/src/runtime.codex-topology.test.ts @@ -2,7 +2,7 @@ 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 { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { ThreadEvent } from "@bb/domain"; import { createAgentRuntime } from "./runtime.js"; import { @@ -131,11 +131,11 @@ describe("codex process topology", () => { return { events, runtime, - childPids: () => - spawnLines().map((line) => Number(line.split(":")[1])), + childPids: () => spawnLines().map((line) => Number(line.split(":")[1])), spawned: () => spawnLines().length, exited: () => readLog().filter((line) => line.startsWith("exit:")).length, - bridges: () => new Set(spawnLines().map((line) => line.split(":")[2])).size, + bridges: () => + new Set(spawnLines().map((line) => line.split(":")[2])).size, bridgeExits, launch, }; @@ -316,21 +316,63 @@ describe("codex process topology", () => { }, 30_000); it("releases the thread on the bridge when a construction times out on the runtime's side", async () => { - // The fake answers thread/start after 800 ms; the runtime gives up at - // 200 ms. The bridge still constructs the session (and its child) when - // the answer lands — and must be told to drop it. + // Keep the runtime's request clock still until the real child is running, + // then cross its existing 200 ms deadline deterministically. Starting the + // wall clock before a subprocess is scheduled does not prove that a child + // was ever under construction — under contention it can be released + // before executing its first line, making a spawn/exit assertion + // impossible by construction. + const realSetTimeout = setTimeout; + const sleepReal = (ms: number): Promise => + new Promise((resolve) => { + realSetTimeout(resolve, ms); + }); const topology = createCodexTopologyRuntime({ - fakeScript: { startDelayMs: 800 }, + fakeScript: { stallThreadStart: true }, threadCreationTimeoutMs: 200, }); const { runtime } = topology; - await expect(startCodexThread(runtime, "t1")).rejects.toThrow(/timed out/i); + vi.useFakeTimers(); + try { + const startOutcome = startCodexThread(runtime, "t1").then( + (providerThreadId) => ({ + status: "resolved" as const, + providerThreadId, + }), + (error: unknown) => ({ status: "rejected" as const, error }), + ); + for (let attempt = 0; topology.spawned() === 0; attempt += 1) { + if (attempt >= 1_000) { + throw new Error("The fake app-server child never started"); + } + await sleepReal(10); + } + + await vi.advanceTimersByTimeAsync(201); + const outcome = await startOutcome; + if (outcome.status === "resolved") { + throw new Error( + `Expected thread construction to time out, but it resolved as ${outcome.providerThreadId}`, + ); + } + expect(outcome.error).toBeInstanceOf(Error); + expect(String(outcome.error)).toMatch(/timed out: thread\/start/i); + } finally { + vi.useRealTimers(); + } + expect(runtime.hasThread("t1")).toBe(false); + const [childPid] = topology.childPids(); + if (childPid === undefined) { + throw new Error("Expected the stalled construction to spawn a child"); + } await waitForRuntimeState({ - label: "the late-constructed child was released", - predicate: () => topology.spawned() === 1 && topology.exited() === 1, + label: "the stalled construction child exited gracefully", + predicate: () => topology.exited() === 1 && !isAlive(childPid), timeoutMs: 10_000, }); + expect(topology.spawned()).toBe(1); + expect(topology.exited()).toBe(1); expect(runtime.listRunningProviders()).toEqual(["codex"]); }, 30_000); diff --git a/plugins/provider-codex/src/bridge/app-server-connection.test.ts b/plugins/provider-codex/src/bridge/app-server-connection.test.ts new file mode 100644 index 0000000000..6c56f41680 --- /dev/null +++ b/plugins/provider-codex/src/bridge/app-server-connection.test.ts @@ -0,0 +1,243 @@ +import { setTimeout as delay } from "node:timers/promises"; +import { describe, expect, it } from "vitest"; +import { z } from "zod"; +import { + CodexAppServerExitedError, + createCodexAppServerConnection, + type CodexAppServerConnection, + type CodexAppServerExitInfo, +} from "./app-server-connection.js"; + +const EPIPE_PAYLOAD_SIZE = 1024 * 1024; + +function deferred(): { + promise: Promise; + resolve(value: T): void; +} { + let resolvePromise: ((value: T) => void) | undefined; + const promise = new Promise((resolve) => { + resolvePromise = resolve; + }); + return { + promise, + resolve(value) { + resolvePromise?.(value); + }, + }; +} + +async function stopConnection( + connection: CodexAppServerConnection, + exit: Promise, +): Promise { + if (!connection.exited) { + connection.kill(); + } + await exit; +} + +function childRequestLine(): string { + return `${JSON.stringify({ + jsonrpc: "2.0", + id: "child-request", + method: "fixture/approval", + params: {}, + })}\n`; +} + +describe("codex app-server connection", () => { + it("keeps pending requests alive for final protocol output after child exit", async () => { + const exited = deferred(); + const lateResponseLine = `${JSON.stringify({ + jsonrpc: "2.0", + id: 1, + result: { thread: { id: "thread-from-final-output" } }, + })}\n`; + const descendantScript = [ + `const line = ${JSON.stringify(lateResponseLine)};`, + "setTimeout(() => process.stdout.write(line, () => process.exit(0)), 250);", + ].join(""); + const childScript = [ + 'const { spawn } = require("node:child_process");', + 'process.stdin.once("data", () => {', + `process.stdout.write(${JSON.stringify(childRequestLine())}, () => {`, + `spawn(process.execPath, ["-e", ${JSON.stringify(descendantScript)}], { stdio: ["ignore", 1, "ignore"] });`, + "process.exit(7);", + "});", + "});", + ].join(""); + const connection = createCodexAppServerConnection({ + command: process.execPath, + args: ["-e", childScript], + cwd: process.cwd(), + env: process.env, + recordThreadId: null, + onNotification: () => undefined, + onRequest: (_method, _params, responder) => { + setTimeout(() => responder.result({ decision: "accept" }), 100); + }, + onExit: exited.resolve, + }); + + try { + await expect( + connection.request({ + method: "thread/start", + resultSchema: z.object({ + thread: z.object({ id: z.string() }), + }), + }), + ).resolves.toEqual({ + thread: { id: "thread-from-final-output" }, + }); + await expect(exited.promise).resolves.toMatchObject({ + code: 7, + signal: null, + }); + } finally { + await stopConnection(connection, exited.promise); + } + }, 30_000); + + it("preserves exit status and stderr while final output drains", async () => { + const exited = deferred(); + const descendantScript = "setTimeout(() => process.exit(0), 250);"; + const childScript = [ + 'const { spawn } = require("node:child_process");', + 'process.stdin.once("data", () => {', + 'process.stderr.write("fixture stderr\\n");', + `process.stdout.write(${JSON.stringify(childRequestLine())}, () => {`, + `spawn(process.execPath, ["-e", ${JSON.stringify(descendantScript)}], { stdio: ["ignore", 1, "ignore"] });`, + "process.exit(7);", + "});", + "});", + ].join(""); + const connection = createCodexAppServerConnection({ + command: process.execPath, + args: ["-e", childScript], + cwd: process.cwd(), + env: process.env, + recordThreadId: null, + onNotification: () => undefined, + onRequest: (_method, _params, responder) => { + setTimeout(() => responder.result({ decision: "accept" }), 100); + }, + onExit: exited.resolve, + }); + + try { + await expect( + connection.request({ + method: "thread/start", + resultSchema: z.unknown(), + }), + ).rejects.toThrow( + "codex app-server exited (code 7, signal null): fixture stderr", + ); + await expect(exited.promise).resolves.toEqual({ + code: 7, + signal: null, + stderrTail: "fixture stderr", + spawnFailed: false, + }); + } finally { + await stopConnection(connection, exited.promise); + } + }, 30_000); + + it("preserves exit status when EPIPE precedes the exit event", async () => { + const exited = deferred(); + const connection = createCodexAppServerConnection({ + command: process.execPath, + args: [ + "-e", + 'process.stderr.write("fixture stderr\\n"); process.exit(7);', + ], + cwd: process.cwd(), + env: process.env, + recordThreadId: null, + onNotification: () => undefined, + onRequest: () => undefined, + onExit: exited.resolve, + }); + + try { + await expect( + connection.request({ + method: "thread/start", + params: { payload: "x".repeat(EPIPE_PAYLOAD_SIZE) }, + resultSchema: z.unknown(), + }), + ).rejects.toThrow(/codex app-server exited \(code 7, signal null\)/); + await expect(exited.promise).resolves.toMatchObject({ + code: 7, + signal: null, + stderrTail: expect.stringContaining("fixture stderr"), + spawnFailed: false, + }); + } finally { + await stopConnection(connection, exited.promise); + } + }, 30_000); + + it("makes a broken child stdin immediately terminal", async () => { + const ready = deferred(); + const exited = deferred(); + const connection = createCodexAppServerConnection({ + command: process.execPath, + args: [ + "-e", + [ + 'require("node:fs").closeSync(0);', + `process.stdout.write(${JSON.stringify( + `${JSON.stringify({ jsonrpc: "2.0", method: "ready" })}\n`, + )});`, + 'process.on("SIGTERM", () => {});', + "setTimeout(() => process.exit(0), 1000);", + ].join(""), + ], + cwd: process.cwd(), + env: process.env, + recordThreadId: null, + onNotification(method) { + if (method === "ready") ready.resolve(); + }, + onRequest: () => undefined, + onExit: exited.resolve, + }); + + try { + await ready.promise; + const pendingRequest = connection.request({ + method: "thread/start", + params: { payload: "x".repeat(EPIPE_PAYLOAD_SIZE) }, + resultSchema: z.unknown(), + }); + const requestWithDeadline = Promise.race([ + pendingRequest, + delay(500).then(() => { + throw new Error("Codex request remained pending after stdin closed"); + }), + ]); + + await expect(requestWithDeadline).rejects.toBeInstanceOf( + CodexAppServerExitedError, + ); + expect(connection.exited).toBe(true); + await expect( + connection.request({ + method: "thread/resume", + resultSchema: z.unknown(), + }), + ).rejects.toBeInstanceOf(CodexAppServerExitedError); + await expect(exited.promise).resolves.toMatchObject({ + code: null, + signal: "SIGKILL", + stderrTail: expect.stringMatching(/stdin failed \(EPIPE\)/), + spawnFailed: false, + }); + } finally { + await stopConnection(connection, exited.promise); + } + }, 30_000); +}); diff --git a/plugins/provider-codex/src/bridge/app-server-connection.ts b/plugins/provider-codex/src/bridge/app-server-connection.ts index dbe1b913d3..01e426ded3 100644 --- a/plugins/provider-codex/src/bridge/app-server-connection.ts +++ b/plugins/provider-codex/src/bridge/app-server-connection.ts @@ -22,6 +22,7 @@ import type { z } from "zod"; const STDERR_TAIL_MAX_CHUNKS = 40; const CLOSE_AFTER_EXIT_GRACE_MS = 1_000; const KILL_ESCALATION_MS = 4_000; +const CLOSED_STDIN_ERROR_CODES = new Set(["EPIPE", "ERR_STREAM_DESTROYED"]); export interface CodexAppServerRequestResponder { result(value: unknown): void; @@ -111,6 +112,14 @@ function parseChildLine(line: string): ParsedChildMessage | null { return parsed as ParsedChildMessage; } +function isClosedChildStdinError(error: Error): boolean { + return ( + "code" in error && + typeof error.code === "string" && + CLOSED_STDIN_ERROR_CODES.has(error.code) + ); +} + export function createCodexAppServerConnection( options: CreateCodexAppServerConnectionOptions, ): CodexAppServerConnection { @@ -132,15 +141,16 @@ export function createCodexAppServerConnection( code: number | null; signal: NodeJS.Signals | null; } | null = null; + let killStarted = false; + let stdinFailure: CodexAppServerExitedError | null = null; let closeGraceTimer: NodeJS.Timeout | null = null; let stdoutLines: Interface | null = null; - function writeLine(message: object): void { - const stdin = child.stdin; - if (!stdin || stdin.destroyed || !stdin.writable) { - return; + function pushStderrChunk(chunk: string): void { + stderrChunks.push(chunk); + if (stderrChunks.length > STDERR_TAIL_MAX_CHUNKS) { + stderrChunks.shift(); } - stdin.write(JSON.stringify(message) + "\n"); } function rejectAllPending(error: Error): void { @@ -153,10 +163,62 @@ export function createCodexAppServerConnection( pending.clear(); } - function finalizeExit(status: { - code: number | null; - signal: NodeJS.Signals | null; - }): void { + function killChild(): void { + if (finalized || killStarted) { + return; + } + killStarted = true; + const escalation = setTimeout(() => { + if (!finalized) { + child.kill("SIGKILL"); + } + }, KILL_ESCALATION_MS); + escalation.unref?.(); + child.kill("SIGTERM"); + } + + function handleBrokenStdin(error: Error): void { + if (finalized || exitStatus !== null || stdinFailure !== null) { + return; + } + const code = + "code" in error && typeof error.code === "string" + ? ` (${error.code})` + : ""; + const detail = `stdin failed${code}: ${error.message}`; + stdinFailure = new CodexAppServerExitedError(`codex app-server ${detail}`); + pushStderrChunk(detail); + // Once the child stops reading requests the protocol cannot recover. Kill + // it immediately, then let its exit/close events finalize the connection: + // an EPIPE can arrive just before `exit`, and that boundary preserves the + // real code or signal plus any final stdout still in flight. + killStarted = true; + child.kill("SIGKILL"); + } + + function writeLine(message: object): void { + if (stdinFailure !== null) { + return; + } + const stdin = child.stdin; + if (!stdin || stdin.destroyed || !stdin.writable) { + // `exit` intentionally precedes finalization while stdout drains. A + // response during that grace must not replace the real exit status or + // discard a final protocol message still buffered on stdout. + if (exitStatus !== null) { + return; + } + handleBrokenStdin(new Error("stdin is not writable")); + return; + } + stdin.write(JSON.stringify(message) + "\n"); + } + + function finalizeConnection( + status: { code: number | null; signal: NodeJS.Signals | null }, + stderrTail: string, + pendingError: CodexAppServerExitedError, + ): void { if (finalized) { return; } @@ -170,8 +232,18 @@ export function createCodexAppServerConnection( stdoutLines?.close(); child.stdout?.destroy(); child.stderr?.destroy(); + rejectAllPending(pendingError); + options.onExit({ ...status, stderrTail, spawnFailed }); + } + + function finalizeExit(status: { + code: number | null; + signal: NodeJS.Signals | null; + }): void { const stderrTail = stderrChunks.join("\n"); - rejectAllPending( + finalizeConnection( + status, + stderrTail, new CodexAppServerExitedError( `codex app-server exited (code ${status.code ?? "null"}, signal ${status.signal ?? "null"})${ stderrTail ? `: ${stderrTail}` : "" @@ -179,7 +251,6 @@ export function createCodexAppServerConnection( { spawnFailed }, ), ); - options.onExit({ ...status, stderrTail, spawnFailed }); } if (child.stdout) { @@ -255,19 +326,23 @@ export function createCodexAppServerConnection( terminal: false, }); stderrLines.on("line", (line) => { - stderrChunks.push(line); - if (stderrChunks.length > STDERR_TAIL_MAX_CHUNKS) { - stderrChunks.shift(); - } + pushStderrChunk(line); }); } child.on("error", (error) => { spawnFailed = true; - stderrChunks.push(error.message); + pushStderrChunk(error.message); finalizeExit({ code: null, signal: null }); }); + child.stdin?.on("error", (error) => { + if (!isClosedChildStdinError(error)) { + throw error; + } + handleBrokenStdin(error); + }); + child.on("exit", (code, signal) => { exitStatus = { code: code ?? null, signal: signal ?? null }; // Prefer `close` (stdio fully drained) so the child's final protocol @@ -285,7 +360,7 @@ export function createCodexAppServerConnection( return { get exited() { - return finalized; + return finalized || stdinFailure !== null; }, request({ method, params, resultSchema, timeoutMs }) { @@ -296,6 +371,9 @@ export function createCodexAppServerConnection( }), ); } + if (stdinFailure !== null) { + return Promise.reject(stdinFailure); + } const id = nextRequestId; nextRequestId += 1; return new Promise((resolve, reject) => { @@ -332,23 +410,14 @@ export function createCodexAppServerConnection( }, notify(method, params) { - if (finalized) { + if (finalized || stdinFailure !== null) { return; } writeLine({ jsonrpc: "2.0", method, params }); }, kill() { - if (finalized) { - return; - } - const escalation = setTimeout(() => { - if (!finalized) { - child.kill("SIGKILL"); - } - }, KILL_ESCALATION_MS); - escalation.unref?.(); - child.kill("SIGTERM"); + killChild(); }, }; } 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 74793f9e61..d7c910edb9 100644 --- a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs +++ b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs @@ -164,8 +164,8 @@ const archivedThreadIds = new Set(); * see the children die on release, archive, and bridge shutdown. */ const processLogPath = script?.processLogPath ?? null; -/** `startDelayMs`: answer `thread/start` only after this many milliseconds. */ -const startDelayMs = script?.startDelayMs ?? 0; +/** `stallThreadStart`: leave `thread/start` pending until this process dies. */ +const stallThreadStart = script?.stallThreadStart ?? false; function logProcessStep(step) { if (processLogPath === null) { @@ -174,11 +174,11 @@ function logProcessStep(step) { appendFileSync(processLogPath, `${step}:${process.pid}:${process.ppid}\n`); } -logProcessStep("spawn"); process.on("SIGTERM", () => { logProcessStep("exit"); process.exit(0); }); +logProcessStep("spawn"); let scriptedTurnIndex = 0; function readArchivedThreadIds() { @@ -341,8 +341,8 @@ async function handleRequest(message) { respond(id, {}); return; case "thread/start": { - if (startDelayMs > 0) { - await new Promise((resolve) => setTimeout(resolve, startDelayMs)); + if (stallThreadStart) { + await new Promise(() => undefined); } threadCounter += 1; const threadId = `codex-fx-${process.pid}-${threadCounter}`;