From 581b67f3c8cacd9f1034e2b2c9c2c09de0ea80b6 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Wed, 26 Aug 2026 15:57:29 -0700 Subject: [PATCH 1/3] Handle Codex child stdin release races --- .../src/runtime.codex-topology.test.ts | 61 +++++++++++++++--- .../src/bridge/app-server-connection.test.ts | 63 +++++++++++++++++++ .../src/bridge/app-server-connection.ts | 56 +++++++++++------ .../src/bridge/fake-codex-app-server.mjs | 5 ++ 4 files changed, 157 insertions(+), 28 deletions(-) create mode 100644 plugins/provider-codex/src/bridge/app-server-connection.test.ts diff --git a/packages/agent-runtime/src/runtime.codex-topology.test.ts b/packages/agent-runtime/src/runtime.codex-topology.test.ts index 0b0f629a4e..ec5046f823 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,62 @@ 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/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, + predicate: () => !isAlive(childPid), timeoutMs: 10_000, }); + expect(topology.spawned()).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..f9409bf882 --- /dev/null +++ b/plugins/provider-codex/src/bridge/app-server-connection.test.ts @@ -0,0 +1,63 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { z } from "zod"; +import { createCodexAppServerConnection } from "./app-server-connection.js"; + +describe("codex app-server connection", () => { + const workspaces: string[] = []; + + afterEach(() => { + for (const workspace of workspaces.splice(0)) { + rmSync(workspace, { recursive: true, force: true }); + } + }); + + it("contains a broken child stdin instead of crashing the bridge", async () => { + const workspace = mkdtempSync(join(tmpdir(), "bb-codex-connection-")); + workspaces.push(workspace); + let signalReady: () => void = () => undefined; + const ready = new Promise((resolve) => { + signalReady = resolve; + }); + let signalExit: () => void = () => undefined; + const exited = new Promise((resolve) => { + signalExit = resolve; + }); + const connection = createCodexAppServerConnection({ + command: process.execPath, + args: [ + "-e", + [ + `require("node:fs").closeSync(0);`, + `process.stdout.write('${JSON.stringify({ jsonrpc: "2.0", method: "ready", params: {} })}\\n');`, + "setInterval(() => undefined, 1000);", + ].join(""), + ], + cwd: workspace, + env: process.env, + recordThreadId: "t1", + onNotification: (method) => { + if (method === "ready") signalReady(); + }, + onRequest: () => undefined, + onExit: () => signalExit(), + }); + + try { + await ready; + await expect( + connection.request({ + method: "thread/start", + resultSchema: z.object({}), + timeoutMs: 1_000, + }), + ).rejects.toThrow(/codex app-server stdin failed/i); + await exited; + expect(connection.exited).toBe(true); + } finally { + connection.kill(); + } + }); +}); diff --git a/plugins/provider-codex/src/bridge/app-server-connection.ts b/plugins/provider-codex/src/bridge/app-server-connection.ts index dbe1b913d3..4a792f262e 100644 --- a/plugins/provider-codex/src/bridge/app-server-connection.ts +++ b/plugins/provider-codex/src/bridge/app-server-connection.ts @@ -132,17 +132,10 @@ export function createCodexAppServerConnection( code: number | null; signal: NodeJS.Signals | null; } | null = null; + let killStarted = false; 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; - } - stdin.write(JSON.stringify(message) + "\n"); - } - function rejectAllPending(error: Error): void { for (const [, request] of pending) { if (request.timeout !== null) { @@ -153,6 +146,40 @@ export function createCodexAppServerConnection( pending.clear(); } + 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) { + return; + } + const connectionError = new CodexAppServerExitedError( + `codex app-server stdin failed: ${error.message}`, + ); + rejectAllPending(connectionError); + killChild(); + } + + function writeLine(message: object): void { + const stdin = child.stdin; + if (!stdin || stdin.destroyed || !stdin.writable) { + handleBrokenStdin(new Error("stdin is not writable")); + return; + } + stdin.write(JSON.stringify(message) + "\n"); + } + function finalizeExit(status: { code: number | null; signal: NodeJS.Signals | null; @@ -268,6 +295,8 @@ export function createCodexAppServerConnection( finalizeExit({ code: null, signal: null }); }); + child.stdin?.on("error", handleBrokenStdin); + child.on("exit", (code, signal) => { exitStatus = { code: code ?? null, signal: signal ?? null }; // Prefer `close` (stdio fully drained) so the child's final protocol @@ -339,16 +368,7 @@ export function createCodexAppServerConnection( }, 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..804dd10013 100644 --- a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs +++ b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs @@ -166,6 +166,8 @@ const archivedThreadIds = new Set(); 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) { @@ -341,6 +343,9 @@ async function handleRequest(message) { respond(id, {}); return; case "thread/start": { + if (stallThreadStart) { + await new Promise(() => undefined); + } if (startDelayMs > 0) { await new Promise((resolve) => setTimeout(resolve, startDelayMs)); } From a2a4acda26c53238557fbd0a3ff55583495b9b03 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Thu, 27 Aug 2026 13:10:15 -0700 Subject: [PATCH 2/3] Preserve Codex exit finalization --- .../src/runtime.codex-topology.test.ts | 7 +- .../src/bridge/app-server-connection.test.ts | 219 +++++++++++++++--- .../src/bridge/app-server-connection.ts | 65 +++++- .../src/bridge/fake-codex-app-server.mjs | 7 +- 4 files changed, 241 insertions(+), 57 deletions(-) diff --git a/packages/agent-runtime/src/runtime.codex-topology.test.ts b/packages/agent-runtime/src/runtime.codex-topology.test.ts index ec5046f823..c0ec1e4059 100644 --- a/packages/agent-runtime/src/runtime.codex-topology.test.ts +++ b/packages/agent-runtime/src/runtime.codex-topology.test.ts @@ -356,7 +356,7 @@ describe("codex process topology", () => { ); } expect(outcome.error).toBeInstanceOf(Error); - expect(String(outcome.error)).toMatch(/timed out/i); + expect(String(outcome.error)).toMatch(/timed out: thread\/start/i); } finally { vi.useRealTimers(); } @@ -367,11 +367,12 @@ describe("codex process topology", () => { throw new Error("Expected the stalled construction to spawn a child"); } await waitForRuntimeState({ - label: "the late-constructed child was released", - predicate: () => !isAlive(childPid), + 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 index f9409bf882..597c232371 100644 --- a/plugins/provider-codex/src/bridge/app-server-connection.test.ts +++ b/plugins/provider-codex/src/bridge/app-server-connection.test.ts @@ -1,63 +1,208 @@ -import { mkdtempSync, rmSync } from "node:fs"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +import { setTimeout as delay } from "node:timers/promises"; +import { describe, expect, it } from "vitest"; import { z } from "zod"; -import { createCodexAppServerConnection } from "./app-server-connection.js"; +import { + CodexAppServerExitedError, + createCodexAppServerConnection, + type CodexAppServerConnection, + type CodexAppServerExitInfo, +} from "./app-server-connection.js"; -describe("codex app-server connection", () => { - const workspaces: string[] = []; +const EPIPE_PAYLOAD_SIZE = 1024 * 1024; - afterEach(() => { - for (const workspace of workspaces.splice(0)) { - rmSync(workspace, { recursive: true, force: true }); - } +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; +} - it("contains a broken child stdin instead of crashing the bridge", async () => { - const workspace = mkdtempSync(join(tmpdir(), "bb-codex-connection-")); - workspaces.push(workspace); - let signalReady: () => void = () => undefined; - const ready = new Promise((resolve) => { - signalReady = resolve; +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, }); - let signalExit: () => void = () => undefined; - const exited = new Promise((resolve) => { - signalExit = 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("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({ jsonrpc: "2.0", method: "ready", params: {} })}\\n');`, - "setInterval(() => undefined, 1000);", + '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: workspace, + cwd: process.cwd(), env: process.env, - recordThreadId: "t1", - onNotification: (method) => { - if (method === "ready") signalReady(); + recordThreadId: null, + onNotification(method) { + if (method === "ready") ready.resolve(); }, onRequest: () => undefined, - onExit: () => signalExit(), + onExit: exited.resolve, }); try { - await ready; + 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/start", - resultSchema: z.object({}), - timeoutMs: 1_000, + method: "thread/resume", + resultSchema: z.unknown(), }), - ).rejects.toThrow(/codex app-server stdin failed/i); - await exited; - expect(connection.exited).toBe(true); + ).rejects.toBeInstanceOf(CodexAppServerExitedError); + await expect(exited.promise).resolves.toMatchObject({ + code: null, + signal: null, + stderrTail: expect.stringMatching(/stdin failed \(EPIPE\)/), + spawnFailed: false, + }); } finally { - connection.kill(); + 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 4a792f262e..c1d6fb139a 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 { @@ -161,29 +170,49 @@ export function createCodexAppServerConnection( } function handleBrokenStdin(error: Error): void { - if (finalized) { + if (finalized || exitStatus !== null) { return; } + const code = + "code" in error && typeof error.code === "string" + ? ` (${error.code})` + : ""; + const detail = `stdin failed${code}: ${error.message}`; const connectionError = new CodexAppServerExitedError( - `codex app-server stdin failed: ${error.message}`, + `codex app-server ${detail}`, + ); + const stderrTail = [...stderrChunks, detail].join("\n"); + // Once the child stops reading requests the protocol cannot recover. Mark + // the connection terminal immediately so callers can rebuild it, and use + // SIGKILL so a child that ignores SIGTERM cannot linger behind that state. + child.kill("SIGKILL"); + finalizeConnection( + { code: null, signal: null }, + stderrTail, + connectionError, ); - rejectAllPending(connectionError); - killChild(); } function writeLine(message: object): void { 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 finalizeExit(status: { - code: number | null; - signal: NodeJS.Signals | null; - }): void { + function finalizeConnection( + status: { code: number | null; signal: NodeJS.Signals | null }, + stderrTail: string, + pendingError: CodexAppServerExitedError, + ): void { if (finalized) { return; } @@ -197,8 +226,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}` : "" @@ -206,7 +245,6 @@ export function createCodexAppServerConnection( { spawnFailed }, ), ); - options.onExit({ ...status, stderrTail, spawnFailed }); } if (child.stdout) { @@ -295,7 +333,12 @@ export function createCodexAppServerConnection( finalizeExit({ code: null, signal: null }); }); - child.stdin?.on("error", handleBrokenStdin); + child.stdin?.on("error", (error) => { + if (!isClosedChildStdinError(error)) { + throw error; + } + handleBrokenStdin(error); + }); child.on("exit", (code, signal) => { exitStatus = { code: code ?? null, signal: signal ?? null }; 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 804dd10013..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,6 @@ 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; @@ -176,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() { @@ -346,9 +344,6 @@ async function handleRequest(message) { if (stallThreadStart) { await new Promise(() => undefined); } - if (startDelayMs > 0) { - await new Promise((resolve) => setTimeout(resolve, startDelayMs)); - } threadCounter += 1; const threadId = `codex-fx-${process.pid}-${threadCounter}`; notify("thread/started", { thread: { id: threadId } }); From 535d0c3483f9dc6d56de37edab9fc64c535267c6 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Thu, 27 Aug 2026 14:17:48 -0700 Subject: [PATCH 3/3] Preserve real exit status after EPIPE --- .../src/bridge/app-server-connection.test.ts | 37 ++++++++++++++- .../src/bridge/app-server-connection.ts | 46 +++++++++++-------- 2 files changed, 62 insertions(+), 21 deletions(-) diff --git a/plugins/provider-codex/src/bridge/app-server-connection.test.ts b/plugins/provider-codex/src/bridge/app-server-connection.test.ts index 597c232371..6c56f41680 100644 --- a/plugins/provider-codex/src/bridge/app-server-connection.test.ts +++ b/plugins/provider-codex/src/bridge/app-server-connection.test.ts @@ -145,6 +145,41 @@ describe("codex app-server connection", () => { } }, 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(); @@ -197,7 +232,7 @@ describe("codex app-server connection", () => { ).rejects.toBeInstanceOf(CodexAppServerExitedError); await expect(exited.promise).resolves.toMatchObject({ code: null, - signal: null, + signal: "SIGKILL", stderrTail: expect.stringMatching(/stdin failed \(EPIPE\)/), spawnFailed: false, }); diff --git a/plugins/provider-codex/src/bridge/app-server-connection.ts b/plugins/provider-codex/src/bridge/app-server-connection.ts index c1d6fb139a..01e426ded3 100644 --- a/plugins/provider-codex/src/bridge/app-server-connection.ts +++ b/plugins/provider-codex/src/bridge/app-server-connection.ts @@ -142,9 +142,17 @@ export function createCodexAppServerConnection( 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 pushStderrChunk(chunk: string): void { + stderrChunks.push(chunk); + if (stderrChunks.length > STDERR_TAIL_MAX_CHUNKS) { + stderrChunks.shift(); + } + } + function rejectAllPending(error: Error): void { for (const [, request] of pending) { if (request.timeout !== null) { @@ -170,7 +178,7 @@ export function createCodexAppServerConnection( } function handleBrokenStdin(error: Error): void { - if (finalized || exitStatus !== null) { + if (finalized || exitStatus !== null || stdinFailure !== null) { return; } const code = @@ -178,22 +186,20 @@ export function createCodexAppServerConnection( ? ` (${error.code})` : ""; const detail = `stdin failed${code}: ${error.message}`; - const connectionError = new CodexAppServerExitedError( - `codex app-server ${detail}`, - ); - const stderrTail = [...stderrChunks, detail].join("\n"); - // Once the child stops reading requests the protocol cannot recover. Mark - // the connection terminal immediately so callers can rebuild it, and use - // SIGKILL so a child that ignores SIGTERM cannot linger behind that state. + 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"); - finalizeConnection( - { code: null, signal: null }, - stderrTail, - connectionError, - ); } 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 @@ -320,16 +326,13 @@ 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 }); }); @@ -357,7 +360,7 @@ export function createCodexAppServerConnection( return { get exited() { - return finalized; + return finalized || stdinFailure !== null; }, request({ method, params, resultSchema, timeoutMs }) { @@ -368,6 +371,9 @@ export function createCodexAppServerConnection( }), ); } + if (stdinFailure !== null) { + return Promise.reject(stdinFailure); + } const id = nextRequestId; nextRequestId += 1; return new Promise((resolve, reject) => { @@ -404,7 +410,7 @@ export function createCodexAppServerConnection( }, notify(method, params) { - if (finalized) { + if (finalized || stdinFailure !== null) { return; } writeLine({ jsonrpc: "2.0", method, params });