Contain Codex child stdin release races

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<void> =>
+      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);
 

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<T>(): {
+  promise: Promise<T>;
+  resolve(value: T): void;
+} {
+  let resolvePromise: ((value: T) => void) | undefined;
+  const promise = new Promise<T>((resolve) => {
+    resolvePromise = resolve;
+  });
+  return {
+    promise,
+    resolve(value) {
+      resolvePromise?.(value);
+    },
+  };
+}
+
+async function stopConnection(
+  connection: CodexAppServerConnection,
+  exit: Promise<CodexAppServerExitInfo>,
+): Promise<void> {
+  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<CodexAppServerExitInfo>();
+    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<CodexAppServerExitInfo>();
+    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<CodexAppServerExitInfo>();
+    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<void>();
+    const exited = deferred<CodexAppServerExitInfo>();
+    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);
+});