Merge PR openai#490: Stop orphaned Codex companion brokers (idle-timeout + shutdown-path corrections)

Author: axisrow

plugins/codex/CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## Unreleased
+
+- Stop an orphaned shared Codex app-server broker after 15 minutes with no connected clients. Active foreground and background jobs keep their broker connection open, and the timeout can be configured with `CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS` (`0` disables the safety timer).
+- Close the broker listener before asynchronous child cleanup and safely reject reconnects already queued during shutdown.
+
 ## 1.0.0
 
 - Initial version of the Codex plugin for Claude Code

plugins/codex/scripts/app-server-broker.mjs

@@ -10,6 +10,21 @@ import { BROKER_BUSY_RPC_CODE, CodexAppServerClient } from "./lib/app-server.mjs
 import { parseBrokerEndpoint } from "./lib/broker-endpoint.mjs";
 
 const STREAMING_METHODS = new Set(["turn/start", "review/start", "thread/compact/start"]);
+const BROKER_IDLE_TIMEOUT_ENV = "CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS";
+const DEFAULT_BROKER_IDLE_TIMEOUT_MS = 15 * 60 * 1000;
+
+function resolveIdleTimeoutMs(env = process.env) {
+  const rawValue = env[BROKER_IDLE_TIMEOUT_ENV];
+  if (rawValue == null || rawValue === "") {
+    return DEFAULT_BROKER_IDLE_TIMEOUT_MS;
+  }
+
+  const parsed = Number(rawValue);
+  if (!Number.isFinite(parsed) || parsed < 0) {
+    return DEFAULT_BROKER_IDLE_TIMEOUT_MS;
+  }
+  return Math.floor(parsed);
+}
 
 function buildStreamThreadIds(method, params, result) {
   const threadIds = new Set();
@@ -70,6 +85,16 @@ async function main() {
   let activeStreamSocket = null;
   let activeStreamThreadIds = null;
   const sockets = new Set();
+  const idleTimeoutMs = resolveIdleTimeoutMs();
+  let idleTimer = null;
+  let shuttingDown = false;
+
+  function cancelIdleShutdown() {
+    if (idleTimer) {
+      clearTimeout(idleTimer);
+      idleTimer = null;
+    }
+  }
 
   function clearSocketOwnership(socket) {
     if (activeRequestSocket === socket) {
@@ -100,11 +125,20 @@ async function main() {
   }
 
   async function shutdown(server) {
+    if (shuttingDown) {
+      return;
+    }
+    shuttingDown = true;
+    cancelIdleShutdown();
+    // Stop accepting connections before awaiting child cleanup. Otherwise a
+    // reconnect can slip into the async shutdown window and inherit a closing
+    // app-server client.
+    const serverClosed = new Promise((resolve) => server.close(resolve));
     for (const socket of sockets) {
       socket.end();
     }
     await appClient.close().catch(() => {});
-    await new Promise((resolve) => server.close(resolve));
+    await serverClosed;
     if (listenTarget.kind === "unix" && fs.existsSync(listenTarget.path)) {
       fs.unlinkSync(listenTarget.path);
     }
@@ -113,9 +147,33 @@ async function main() {
     }
   }
 
+  function scheduleIdleShutdown(server) {
+    cancelIdleShutdown();
+    if (shuttingDown || idleTimeoutMs === 0 || sockets.size > 0 || activeRequestSocket || activeStreamSocket) {
+      return;
+    }
+
+    idleTimer = setTimeout(async () => {
+      idleTimer = null;
+      if (sockets.size > 0 || activeRequestSocket || activeStreamSocket) {
+        return;
+      }
+      await shutdown(server);
+      process.exit(0);
+    }, idleTimeoutMs);
+  }
+
   appClient.setNotificationHandler(routeNotification);
 
   const server = net.createServer((socket) => {
+    if (shuttingDown) {
+      // An already-accepted connection event can be delivered after
+      // server.close(). Reject it decisively and absorb a simultaneous reset.
+      socket.on("error", () => {});
+      socket.destroy();
+      return;
+    }
+    cancelIdleShutdown();
     sockets.add(socket);
     socket.setEncoding("utf8");
     let buffer = "";
@@ -232,11 +290,13 @@ async function main() {
     socket.on("close", () => {
       sockets.delete(socket);
       clearSocketOwnership(socket);
+      scheduleIdleShutdown(server);
     });
 
     socket.on("error", () => {
       sockets.delete(socket);
       clearSocketOwnership(socket);
+      scheduleIdleShutdown(server);
     });
   });
 
@@ -250,7 +310,9 @@ async function main() {
     process.exit(0);
   });
 
-  server.listen(listenTarget.path);
+  server.listen(listenTarget.path, () => {
+    scheduleIdleShutdown(server);
+  });
 }
 
 main().catch((error) => {

tests/broker-idle.test.mjs

@@ -0,0 +1,126 @@
+import fs from "node:fs";
+import net from "node:net";
+import path from "node:path";
+import test from "node:test";
+import assert from "node:assert/strict";
+import { once } from "node:events";
+import { spawn } from "node:child_process";
+import { fileURLToPath } from "node:url";
+
+import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs";
+import { makeTempDir } from "./helpers.mjs";
+import { sendBrokerShutdown, waitForBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs";
+
+const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..");
+const BROKER_SCRIPT = path.join(ROOT, "plugins", "codex", "scripts", "app-server-broker.mjs");
+const IDLE_TIMEOUT_ENV = "CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS";
+
+function spawnTestBroker({ idleTimeoutMs }) {
+  const workspace = makeTempDir();
+  const binDir = makeTempDir();
+  const sessionDir = makeTempDir("codex-plugin-broker-");
+  const socketPath = path.join(sessionDir, "broker.sock");
+  const endpoint = `unix:${socketPath}`;
+  const pidFile = path.join(sessionDir, "broker.pid");
+  installFakeCodex(binDir);
+
+  const child = spawn(
+    process.execPath,
+    [BROKER_SCRIPT, "serve", "--endpoint", endpoint, "--cwd", workspace, "--pid-file", pidFile],
+    {
+      cwd: workspace,
+      env: {
+        ...buildEnv(binDir),
+        [IDLE_TIMEOUT_ENV]: String(idleTimeoutMs)
+      },
+      stdio: ["ignore", "pipe", "pipe"]
+    }
+  );
+
+  return { child, endpoint, pidFile, socketPath };
+}
+
+async function waitForExit(child, timeoutMs = 3000) {
+  if (child.exitCode != null || child.signalCode != null) {
+    return;
+  }
+  await Promise.race([
+    once(child, "exit"),
+    new Promise((_, reject) => setTimeout(() => reject(new Error("Timed out waiting for broker to exit.")), timeoutMs))
+  ]);
+}
+
+function terminateBroker(child) {
+  if (child.exitCode == null && child.signalCode == null) {
+    child.kill("SIGTERM");
+  }
+}
+
+test("broker exits and removes runtime files after its last client stays disconnected", async (t) => {
+  const broker = spawnTestBroker({ idleTimeoutMs: 150 });
+  t.after(() => terminateBroker(broker.child));
+
+  assert.equal(await waitForBrokerEndpoint(broker.endpoint), true);
+  await waitForExit(broker.child);
+
+  assert.equal(broker.child.exitCode, 0);
+  assert.equal(fs.existsSync(broker.socketPath), false);
+  assert.equal(fs.existsSync(broker.pidFile), false);
+});
+
+test("broker idle shutdown waits until a connected client disconnects", async (t) => {
+  const broker = spawnTestBroker({ idleTimeoutMs: 150 });
+  t.after(() => terminateBroker(broker.child));
+
+  assert.equal(await waitForBrokerEndpoint(broker.endpoint), true);
+  const socket = net.createConnection({ path: broker.socketPath });
+  await once(socket, "connect");
+
+  await new Promise((resolve) => setTimeout(resolve, 350));
+  assert.equal(broker.child.exitCode, null);
+  assert.equal(broker.child.signalCode, null);
+
+  socket.end();
+  await once(socket, "close");
+  await waitForExit(broker.child);
+  assert.equal(broker.child.exitCode, 0);
+});
+
+test("broker idle timeout can be disabled", async (t) => {
+  const broker = spawnTestBroker({ idleTimeoutMs: 0 });
+  t.after(() => terminateBroker(broker.child));
+
+  assert.equal(await waitForBrokerEndpoint(broker.endpoint), true);
+  await new Promise((resolve) => setTimeout(resolve, 350));
+  assert.equal(broker.child.exitCode, null);
+  assert.equal(broker.child.signalCode, null);
+
+  terminateBroker(broker.child);
+  await waitForExit(broker.child);
+});
+
+test("broker shuts down cleanly while new clients race to connect", async (t) => {
+  const broker = spawnTestBroker({ idleTimeoutMs: 0 });
+  t.after(() => terminateBroker(broker.child));
+
+  assert.equal(await waitForBrokerEndpoint(broker.endpoint), true);
+  const reconnects = Array.from(
+    { length: 50 },
+    () =>
+      new Promise((resolve) => {
+        const socket = net.createConnection({ path: broker.socketPath });
+        const timer = setTimeout(() => socket.destroy(), 1000);
+        const finish = () => {
+          clearTimeout(timer);
+          resolve();
+        };
+        socket.on("connect", () => socket.end());
+        socket.on("error", finish);
+        socket.on("close", finish);
+      })
+  );
+
+  await Promise.all([sendBrokerShutdown(broker.endpoint), ...reconnects]);
+  await waitForExit(broker.child);
+  assert.equal(broker.child.exitCode, 0);
+});

tests/fake-codex-fixture.mjs

@@ -749,6 +749,9 @@ export function buildEnv(binDir) {
   const sep = process.platform === "win32" ? ";" : ":";
   return {
     ...process.env,
-    PATH: `${binDir}${sep}${process.env.PATH}`
+    PATH: `${binDir}${sep}${process.env.PATH}`,
+    // Production keeps an idle broker warm for 15 minutes. Tests only need a
+    // brief reuse window and should not leave dozens of detached helpers.
+    CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS: "2000"
   };
 }