Prevent orphaned MCP stdio spin

Co-Authored-By: OpenAI Codex <noreply@openai.com>

Author: xuio

dist/index.js

@@ -21548,8 +21548,16 @@ function loggingDiagnostics(env = process.env) {
   };
 }
 function writeDefaultLog(line) {
-  process.stderr.write(`${line}
-`);
+  try {
+    if (!process.stderr.destroyed && process.stderr.writable) {
+      process.stderr.write(`${line}
+`, (error2) => {
+        if (error2) lastLogFileError = error2.message;
+      });
+    }
+  } catch (error2) {
+    lastLogFileError = error2 instanceof Error ? error2.message : String(error2);
+  }
   const logFile = process.env.CODEX_SUBAGENTS_LOG_FILE?.trim();
   if (!logFile) return;
   try {
@@ -23503,6 +23511,21 @@ function createProgressReporter(extra, options = {}) {
   return { send, flush };
 }
 
+// src/stdio.ts
+var brokenStdioCodes = /* @__PURE__ */ new Set([
+  "EPIPE",
+  "ECONNRESET",
+  "ERR_STREAM_DESTROYED",
+  "ERR_STREAM_WRITE_AFTER_END"
+]);
+function isBrokenStdioError(error2) {
+  if (!error2 || typeof error2 !== "object") return false;
+  const value = error2;
+  if (typeof value.code === "string" && brokenStdioCodes.has(value.code)) return true;
+  const message = typeof value.message === "string" ? value.message.toLowerCase() : "";
+  return message.includes("broken pipe") || message.includes("stream has been destroyed") || message.includes("write after end");
+}
+
 // src/response.ts
 var singleAgentLimits = {
   finalMessageChars: 12e3,
@@ -26222,7 +26245,7 @@ async function loggedToolCall(tool, args, extra, run) {
     return errorResult(error2, tool);
   }
 }
-function installTransportLogging(transport) {
+function installTransportLogging(transport, shutdown) {
   const previousOnMessage = transport.onmessage;
   transport.onmessage = (message) => {
     logger.rawDebug("mcp.transport.inbound", {
@@ -26232,6 +26255,10 @@ function installTransportLogging(transport) {
   };
   const previousOnError = transport.onerror;
   transport.onerror = (error2) => {
+    if (isBrokenStdioError(error2)) {
+      shutdown("mcp_transport_broken_stdio", 0, 500);
+      return;
+    }
     logger.error("mcp.transport.error", { error: errorForLog(error2) });
     previousOnError?.(error2);
   };
@@ -26243,6 +26270,10 @@ function installTransportLogging(transport) {
     try {
       await send(message);
     } catch (error2) {
+      if (isBrokenStdioError(error2)) {
+        shutdown("mcp_transport_send_broken_stdio", 0, 500);
+        return;
+      }
       logger.error("mcp.transport.send_failed", { error: errorForLog(error2) });
       throw error2;
     }
@@ -28456,34 +28487,59 @@ registerCleanupHandler(async (reason) => {
 });
 function installProcessCleanup() {
   let shutdownStarted = false;
-  const shutdown = (reason, exitCode) => {
+  const shutdown = (reason, exitCode, graceMs = 2500) => {
     if (shutdownStarted) return;
     shutdownStarted = true;
-    void cleanupRuntime(reason).finally(() => {
+    const forceExit = exitCode === void 0 ? void 0 : setTimeout(() => process.exit(exitCode), Math.max(1e3, graceMs + 1e3));
+    forceExit?.unref();
+    void cleanupRuntime(reason, graceMs).finally(() => {
+      if (forceExit) clearTimeout(forceExit);
       if (exitCode !== void 0) process.exit(exitCode);
     });
   };
+  const shutdownOnBrokenStdio = (reason) => (error2) => {
+    if (isBrokenStdioError(error2)) {
+      shutdown(reason, 0, 500);
+      return;
+    }
+    logger.error(`${reason}.error`, { error: errorForLog(error2) });
+  };
   process.once("SIGINT", () => shutdown("SIGINT", 130));
   process.once("SIGTERM", () => shutdown("SIGTERM", 143));
-  process.stdin.once("close", () => shutdown("stdin_close"));
+  process.stdin.once("close", () => shutdown("stdin_close", 0, 500));
+  process.stdin.once("end", () => shutdown("stdin_end", 0, 500));
+  process.stdin.once("error", shutdownOnBrokenStdio("stdin"));
+  process.stdout.once("error", shutdownOnBrokenStdio("stdout"));
+  process.stderr.once("error", shutdownOnBrokenStdio("stderr"));
+  return shutdown;
 }
 async function main() {
-  installProcessCleanup();
+  const shutdown = installProcessCleanup();
   process.on("unhandledRejection", (error2) => {
+    if (isBrokenStdioError(error2)) {
+      shutdown("unhandled_broken_stdio", 0, 500);
+      return;
+    }
     logger.error("process.unhandled_rejection", { error: errorForLog(error2) });
   });
   process.on("uncaughtException", (error2) => {
+    if (isBrokenStdioError(error2)) {
+      shutdown("uncaught_broken_stdio", 0, 500);
+      return;
+    }
     logger.error("process.uncaught_exception", { error: errorForLog(error2) });
+    shutdown("uncaught_exception", 1, 500);
   });
   logger.info("server.starting", {
     logging: loggingDiagnostics()
   });
   const transport = new StdioServerTransport();
-  installTransportLogging(transport);
+  installTransportLogging(transport, shutdown);
   await server.connect(transport);
   logger.info("server.connected", { transport: "stdio" });
 }
 main().catch((error2) => {
+  if (isBrokenStdioError(error2)) process.exit(0);
   logger.error("server.start_failed", { error: errorForLog(error2) });
   process.exit(1);
 });

package.json

@@ -33,6 +33,7 @@
     "test:stress": "node test/stress-mcp.mjs",
     "test:progress": "node test/progress-mcp.mjs",
     "test:advanced": "node test/advanced-mcp.mjs",
+    "test:stdio-shutdown": "node test/stdio-shutdown.mjs",
     "test:dev-link": "node test/dev-link.mjs",
     "test:plugin-manifest": "node test/plugin-manifest.mjs",
     "test:codex-runtime": "node test/codex-runtime-probe.mjs",
@@ -46,8 +47,8 @@
     "test:claude-real-codex": "node test/claude-real-codex.mjs",
     "test:claude-real-session": "node test/claude-real-session.mjs",
     "test:real-soak": "node test/real-soak.mjs",
-    "test:ci": "npm run build && npm run test:plugin-manifest && npm test && npm run smoke:mcp && npm run test:reliability && npm run test:stress && npm run test:progress && npm run test:advanced && npm run test:dev-link",
-    "test:comprehensive": "npm run build && npm test && npm run smoke:mcp && npm run test:reliability && npm run test:stress && npm run test:progress && npm run test:advanced && npm run test:codex-runtime && npm run test:app-server-contract && npm run test:real-matrix && npm run validate:plugin && npm run test:claude-desktop",
+    "test:ci": "npm run build && npm run test:plugin-manifest && npm test && npm run test:stdio-shutdown && npm run smoke:mcp && npm run test:reliability && npm run test:stress && npm run test:progress && npm run test:advanced && npm run test:dev-link",
+    "test:comprehensive": "npm run build && npm test && npm run test:stdio-shutdown && npm run smoke:mcp && npm run test:reliability && npm run test:stress && npm run test:progress && npm run test:advanced && npm run test:codex-runtime && npm run test:app-server-contract && npm run test:real-matrix && npm run validate:plugin && npm run test:claude-desktop",
     "test:claude-desktop": "node test/claude-desktop-cli.mjs"
   },
   "keywords": [

src/index.ts

@@ -31,6 +31,7 @@ import { errorForLog, logger, loggingDiagnostics, makeLogId, summarizeRawTraffic
 import { recoveryForAgentResult, recoveryForError, recoveryForWait } from "./recovery.js";
 import { redactJsonValue, redactSensitiveText } from "./redaction.js";
 import { createProgressReporter, type ProgressOptions, type ProgressReporter, type ToolExtra } from "./progress.js";
+import { isBrokenStdioError } from "./stdio.js";
 import {
   compactAgentResultForMcp,
   compactAgentResultsForMcp,
@@ -1002,7 +1003,7 @@ async function loggedToolCall(
   }
 }
 
-function installTransportLogging(transport: StdioServerTransport): void {
+function installTransportLogging(transport: StdioServerTransport, shutdown: (reason: string, exitCode?: number, graceMs?: number) => void): void {
   const previousOnMessage = transport.onmessage;
   transport.onmessage = (message) => {
     logger.rawDebug("mcp.transport.inbound", {
@@ -1013,6 +1014,10 @@ function installTransportLogging(transport: StdioServerTransport): void {
 
   const previousOnError = transport.onerror;
   transport.onerror = (error) => {
+    if (isBrokenStdioError(error)) {
+      shutdown("mcp_transport_broken_stdio", 0, 500);
+      return;
+    }
     logger.error("mcp.transport.error", { error: errorForLog(error) });
     previousOnError?.(error);
   };
@@ -1025,6 +1030,10 @@ function installTransportLogging(transport: StdioServerTransport): void {
     try {
       await send(message);
     } catch (error) {
+      if (isBrokenStdioError(error)) {
+        shutdown("mcp_transport_send_broken_stdio", 0, 500);
+        return;
+      }
       logger.error("mcp.transport.send_failed", { error: errorForLog(error) });
       throw error;
     }
@@ -3788,40 +3797,67 @@ registerCleanupHandler(async (reason) => {
   await sessionManager.shutdown(reason);
 });
 
-function installProcessCleanup(): void {
+function installProcessCleanup(): (reason: string, exitCode?: number, graceMs?: number) => void {
   let shutdownStarted = false;
-  const shutdown = (reason: string, exitCode?: number) => {
+  const shutdown = (reason: string, exitCode?: number, graceMs = 2_500) => {
     if (shutdownStarted) return;
     shutdownStarted = true;
-    void cleanupRuntime(reason).finally(() => {
+    const forceExit = exitCode === undefined
+      ? undefined
+      : setTimeout(() => process.exit(exitCode), Math.max(1_000, graceMs + 1_000));
+    forceExit?.unref();
+    void cleanupRuntime(reason, graceMs).finally(() => {
+      if (forceExit) clearTimeout(forceExit);
       if (exitCode !== undefined) process.exit(exitCode);
     });
   };
+  const shutdownOnBrokenStdio = (reason: string) => (error: unknown) => {
+    if (isBrokenStdioError(error)) {
+      shutdown(reason, 0, 500);
+      return;
+    }
+    logger.error(`${reason}.error`, { error: errorForLog(error) });
+  };
 
   process.once("SIGINT", () => shutdown("SIGINT", 130));
   process.once("SIGTERM", () => shutdown("SIGTERM", 143));
-  process.stdin.once("close", () => shutdown("stdin_close"));
+  process.stdin.once("close", () => shutdown("stdin_close", 0, 500));
+  process.stdin.once("end", () => shutdown("stdin_end", 0, 500));
+  process.stdin.once("error", shutdownOnBrokenStdio("stdin"));
+  process.stdout.once("error", shutdownOnBrokenStdio("stdout"));
+  process.stderr.once("error", shutdownOnBrokenStdio("stderr"));
+  return shutdown;
 }
 
 async function main(): Promise<void> {
-  installProcessCleanup();
+  const shutdown = installProcessCleanup();
   process.on("unhandledRejection", (error) => {
+    if (isBrokenStdioError(error)) {
+      shutdown("unhandled_broken_stdio", 0, 500);
+      return;
+    }
     logger.error("process.unhandled_rejection", { error: errorForLog(error) });
   });
   process.on("uncaughtException", (error) => {
+    if (isBrokenStdioError(error)) {
+      shutdown("uncaught_broken_stdio", 0, 500);
+      return;
+    }
     logger.error("process.uncaught_exception", { error: errorForLog(error) });
+    shutdown("uncaught_exception", 1, 500);
   });
 
   logger.info("server.starting", {
     logging: loggingDiagnostics(),
   });
   const transport = new StdioServerTransport();
-  installTransportLogging(transport);
+  installTransportLogging(transport, shutdown);
   await server.connect(transport);
   logger.info("server.connected", { transport: "stdio" });
 }
 
 main().catch((error) => {
+  if (isBrokenStdioError(error)) process.exit(0);
   logger.error("server.start_failed", { error: errorForLog(error) });
   process.exit(1);
 });

src/logging.ts

@@ -203,7 +203,16 @@ export function loggingDiagnostics(env: NodeJS.ProcessEnv = process.env): Record
 }
 
 function writeDefaultLog(line: string): void {
-  process.stderr.write(`${line}\n`);
+  try {
+    if (!process.stderr.destroyed && process.stderr.writable) {
+      process.stderr.write(`${line}\n`, (error) => {
+        if (error) lastLogFileError = error.message;
+      });
+    }
+  } catch (error) {
+    lastLogFileError = error instanceof Error ? error.message : String(error);
+  }
+
   const logFile = process.env.CODEX_SUBAGENTS_LOG_FILE?.trim();
   if (!logFile) return;
   try {

src/stdio.ts

@@ -0,0 +1,18 @@
+const brokenStdioCodes = new Set([
+  "EPIPE",
+  "ECONNRESET",
+  "ERR_STREAM_DESTROYED",
+  "ERR_STREAM_WRITE_AFTER_END",
+]);
+
+export function isBrokenStdioError(error: unknown): boolean {
+  if (!error || typeof error !== "object") return false;
+  const value = error as { code?: unknown; message?: unknown };
+  if (typeof value.code === "string" && brokenStdioCodes.has(value.code)) return true;
+  const message = typeof value.message === "string" ? value.message.toLowerCase() : "";
+  return (
+    message.includes("broken pipe") ||
+    message.includes("stream has been destroyed") ||
+    message.includes("write after end")
+  );
+}

test/app-server-hardening.test.ts

@@ -263,7 +263,7 @@ describe("app-server hardening", () => {
       return calls.some((call) => call.method === "process/sigterm");
     }, 10_000);
     expect(sawSigterm).toBe(true);
-  });
+  }, 15_000);
 
   it("keeps concurrent app-server sessions isolated", async () => {
     const manager = new CodexSessionManager();

test/stdio-shutdown.mjs

@@ -0,0 +1,70 @@
+import { spawn } from "node:child_process";
+import { once } from "node:events";
+import { mkdtemp, rm } from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { setTimeout as delay } from "node:timers/promises";
+
+const root = process.cwd();
+const stateDir = await mkdtemp(path.join(os.tmpdir(), "codex-subagents-stdio-"));
+
+function assert(condition, message, details) {
+  if (!condition) {
+    throw new Error(`${message}${details ? `\n${JSON.stringify(details, null, 2)}` : ""}`);
+  }
+}
+
+async function waitForExit(child, timeoutMs) {
+  const exited = once(child, "exit").then(([code, signal]) => ({ code, signal }));
+  return Promise.race([exited, delay(timeoutMs).then(() => undefined)]);
+}
+
+try {
+  const child = spawn(path.join(root, "dist/index.js"), [], {
+    cwd: root,
+    stdio: ["pipe", "pipe", "pipe"],
+    env: {
+      ...process.env,
+      CODEX_SUBAGENTS_SESSION_STATE_FILE: path.join(stateDir, "sessions.json"),
+      CODEX_SUBAGENTS_LOG_LEVEL: "debug",
+    },
+  });
+
+  let stderr = "";
+  child.stderr.setEncoding("utf8");
+  child.stderr.on("data", (chunk) => {
+    stderr += chunk;
+  });
+
+  await delay(250);
+  child.stdout.destroy();
+  child.stdin.write(`${JSON.stringify({
+    jsonrpc: "2.0",
+    id: 1,
+    method: "initialize",
+    params: {
+      protocolVersion: "2025-06-18",
+      capabilities: {},
+      clientInfo: { name: "stdio-shutdown-test", version: "0.1.0" },
+    },
+  })}\n`);
+
+  const result = await waitForExit(child, 5_000);
+  if (!result) {
+    child.kill("SIGKILL");
+    throw new Error(`MCP server did not exit after stdout disconnect.\n${stderr}`);
+  }
+  assert(result.code === 0, "MCP server should exit cleanly after stdout disconnect", {
+    result,
+    stderr,
+  });
+  assert(
+    !stderr.includes("uncaught_exception") && !stderr.includes("unhandled_rejection"),
+    "broken stdio should not loop through uncaught exception logging",
+    stderr,
+  );
+} finally {
+  await rm(stateDir, { recursive: true, force: true });
+}
+
+console.log("Stdio shutdown test passed");