fix: idle self-shutdown for the app-server broker to stop stale brokers accumulating

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

@@ -8,6 +8,7 @@ import process from "node:process";
 import { parseArgs } from "./lib/args.mjs";
 import { BROKER_BUSY_RPC_CODE, CodexAppServerClient } from "./lib/app-server.mjs";
 import { parseBrokerEndpoint } from "./lib/broker-endpoint.mjs";
+import { clearBrokerSession } from "./lib/broker-lifecycle.mjs";
 
 const STREAMING_METHODS = new Set(["turn/start", "review/start", "thread/compact/start"]);
 
@@ -71,6 +72,54 @@ async function main() {
   let activeStreamThreadIds = null;
   const sockets = new Set();
 
+  // Idle self-shutdown: a broker is spawned detached+unref'd per session and is reused by
+  // endpoint-readiness, keyed by cwd. When a session ends (or crashes, or its host process is
+  // replaced) the broker keeps its app-server child alive with no client connected, and is never
+  // torn down because teardown only ever targets the *current* cwd's stale entry. Across sessions in
+  // distinct cwds these accumulate. Exit when no client has been connected for an idle window so a
+  // dead session's broker reaps itself. Disabled when the window is <= 0.
+  //
+  // Why this never races a live broker: a client connects per logical operation and disconnects when
+  // it finishes (the broker sits CLIENTLESS between turns and is reused via endpoint-readiness, not a
+  // held socket). A long-running turn keeps its socket connected for the whole turn, so sockets.size
+  // stays > 0 and the timer never arms. The window therefore only elapses after a genuine idle gap
+  // (no turn for IDLE_SHUTDOWN_MS), after which a fresh respawn is cheap. Keep the default well above
+  // expected inter-turn think-time — do NOT lower it toward reconnect latency.
+  const IDLE_SHUTDOWN_ENV = "CODEX_BROKER_IDLE_SHUTDOWN_MS"; // value-constant, mirrors PID_FILE_ENV/LOG_FILE_ENV
+  const IDLE_SHUTDOWN_MS = (() => {
+    const raw = Number(process.env[IDLE_SHUTDOWN_ENV]);
+    return Number.isFinite(raw) ? raw : 30 * 60 * 1000; // default 30 min; set 0 (or negative) to disable
+  })();
+  let idleTimer = null;
+  let serverRef = null;
+
+  function clearIdleTimer() {
+    if (idleTimer) {
+      clearTimeout(idleTimer);
+      idleTimer = null;
+    }
+  }
+
+  function armIdleTimer() {
+    clearIdleTimer();
+    if (IDLE_SHUTDOWN_MS <= 0 || sockets.size > 0 || !serverRef) {
+      return;
+    }
+    idleTimer = setTimeout(() => {
+      // Re-check under the timer: only exit if still idle (no client reconnected meanwhile).
+      if (sockets.size === 0) {
+        // Clear the persisted broker.json for this cwd BEFORE exiting. Unlike SIGTERM/RPC shutdown
+        // (driven by a caller that also manages session state), this self-exit is autonomous: if we
+        // left broker.json pointing at the now-removed socket, a later `reuseExistingBroker: true`
+        // caller (e.g. `codex setup --json`) would load the stale endpoint without re-probing and
+        // report ready:false / connect ENOENT until something else cleans state.
+        try { clearBrokerSession(cwd); } catch { /* best-effort; teardown still proceeds */ }
+        shutdown(serverRef).then(() => process.exit(0));
+      }
+    }, IDLE_SHUTDOWN_MS);
+    idleTimer.unref?.(); // never keep the process alive solely for this timer
+  }
+
   function clearSocketOwnership(socket) {
     if (activeRequestSocket === socket) {
       activeRequestSocket = null;
@@ -103,8 +152,11 @@ async function main() {
     for (const socket of sockets) {
       socket.end();
     }
+    // Stop accepting new connections BEFORE any async teardown, so a client connecting during
+    // appClient.close() can't slip past the idle-path re-check and get dropped mid-handshake.
+    const serverClosed = new Promise((resolve) => server.close(resolve));
     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);
     }
@@ -117,6 +169,7 @@ async function main() {
 
   const server = net.createServer((socket) => {
     sockets.add(socket);
+    clearIdleTimer(); // a client is connected; cancel any pending idle shutdown
     socket.setEncoding("utf8");
     let buffer = "";
 
@@ -225,11 +278,13 @@ async function main() {
     socket.on("close", () => {
       sockets.delete(socket);
       clearSocketOwnership(socket);
+      armIdleTimer(); // last client gone -> start the idle countdown to self-reap
     });
 
     socket.on("error", () => {
       sockets.delete(socket);
       clearSocketOwnership(socket);
+      armIdleTimer();
     });
   });
 
@@ -243,7 +298,10 @@ async function main() {
     process.exit(0);
   });
 
-  server.listen(listenTarget.path);
+  serverRef = server;
+  server.listen(listenTarget.path, () => {
+    armIdleTimer(); // a broker spawned but never connected to should also self-reap
+  });
 }
 
 main().catch((error) => {

tests/broker-idle-shutdown.test.mjs

@@ -0,0 +1,188 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import net from "node:net";
+import path from "node:path";
+import fs from "node:fs";
+import { spawn } from "node:child_process";
+import { fileURLToPath } from "node:url";
+
+import { makeTempDir } from "./helpers.mjs";
+import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs";
+import {
+  saveBrokerSession,
+  loadBrokerSession
+} from "../plugins/codex/scripts/lib/broker-lifecycle.mjs";
+
+const BROKER = fileURLToPath(
+  new URL("../plugins/codex/scripts/app-server-broker.mjs", import.meta.url)
+);
+
+// Drive the broker process directly over its unix socket. The broker connects to a `codex app-server`
+// child on startup; we install the fake-codex fixture on PATH so the broker boots deterministically in
+// CI (no real Codex login required). CODEX_BROKER_IDLE_SHUTDOWN_MS is set small so the idle path is
+// exercised in milliseconds.
+
+function brokerEnv(binDir, idleMs) {
+  return { ...buildEnv(binDir), CODEX_BROKER_IDLE_SHUTDOWN_MS: String(idleMs) };
+}
+
+function spawnBroker(env, endpoint, cwd) {
+  const args = [BROKER, "serve", "--endpoint", endpoint];
+  if (cwd) {
+    args.push("--cwd", cwd);
+  }
+  return spawn(process.execPath, args, {
+    env,
+    stdio: ["ignore", "ignore", "ignore"]
+  });
+}
+
+async function waitForSocket(sockPath, timeoutMs = 20000) {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    if (fs.existsSync(sockPath)) {
+      const ok = await new Promise((resolve) => {
+        const c = net.createConnection({ path: sockPath });
+        c.on("connect", () => { c.end(); resolve(true); });
+        c.on("error", () => resolve(false));
+      });
+      if (ok) return true;
+    }
+    await new Promise((r) => setTimeout(r, 50));
+  }
+  return false;
+}
+
+function waitForExit(child, timeoutMs) {
+  return new Promise((resolve) => {
+    let done = false;
+    const t = setTimeout(() => { if (!done) { done = true; resolve(false); } }, timeoutMs);
+    child.on("exit", () => { if (!done) { done = true; clearTimeout(t); resolve(true); } });
+  });
+}
+
+function setup(t, idleMs) {
+  const binDir = makeTempDir();
+  installFakeCodex(binDir);
+  // Make `node` discoverable on the fixture PATH too (buildEnv prepends binDir; the broker spawns
+  // with process.execPath directly, but the fake codex shebang needs node on PATH).
+  try { fs.symlinkSync(process.execPath, path.join(binDir, "node")); } catch { /* may exist */ }
+  const sockDir = makeTempDir("cxc-idle-");
+  const sock = path.join(sockDir, "broker.sock");
+  const endpoint = `unix:${sock}`;
+  const child = spawnBroker(brokerEnv(binDir, idleMs), endpoint);
+  t.after(() => {
+    try { child.kill("SIGKILL"); } catch { /* already gone */ }
+    try { fs.rmSync(sockDir, { recursive: true, force: true }); } catch { /* ignore */ }
+  });
+  return { child, sock };
+}
+
+// Timing strategy (must be robust when the whole suite runs in parallel and saturates CPU):
+//  - IDLE = 3s: large enough that the broker reliably stays up long enough to be observed before its
+//    idle timer could fire, even under load. Small enough that the self-exit tests finish quickly.
+//  - waitForSocket default 20s: broker boot = spawn fake-codex + app-server handshake, which can be
+//    slow under contention; readiness must NOT be gated on the short side.
+//  - exit waits are a generous multiple of IDLE so a loaded scheduler still observes the exit.
+const IDLE = 3000;
+
+test("broker self-exits after the idle window when no client ever connects", async (t) => {
+  const { child, sock } = setup(t, IDLE);
+  assert.equal(await waitForSocket(sock), true, "broker should come up with the fake codex on PATH");
+  // Never connect a client -> the startup-armed idle timer fires and exits the process.
+  assert.equal(
+    await waitForExit(child, IDLE * 5),
+    true,
+    "broker should self-exit after the idle window with no client"
+  );
+});
+
+test("a connected client cancels the idle timer (broker stays up)", async (t) => {
+  const { child, sock } = setup(t, IDLE);
+  assert.equal(await waitForSocket(sock), true);
+  const client = net.createConnection({ path: sock });
+  await new Promise((r) => client.on("connect", r));
+  // Hold the client open well past the idle window; the broker must NOT exit while connected.
+  const exitedWhileHeld = await waitForExit(child, IDLE * 2);
+  client.end();
+  assert.equal(exitedWhileHeld, false, "broker must not idle-exit while a client is connected");
+});
+
+test("broker re-arms and exits after a client connects then disconnects", async (t) => {
+  const { child, sock } = setup(t, IDLE);
+  assert.equal(await waitForSocket(sock), true);
+  const client = net.createConnection({ path: sock });
+  await new Promise((r) => client.on("connect", r));
+  client.end();
+  // After the last client leaves, the idle countdown re-arms and the broker reaps itself.
+  assert.equal(
+    await waitForExit(child, IDLE * 5),
+    true,
+    "broker should self-exit after the last client disconnects and the idle window elapses"
+  );
+});
+
+test("idle shutdown is disabled when the window is <= 0", async (t) => {
+  const { child, sock } = setup(t, 0);
+  assert.equal(await waitForSocket(sock), true);
+  assert.equal(
+    await waitForExit(child, IDLE),
+    false,
+    "broker must stay up when idle shutdown is disabled (<=0)"
+  );
+});
+
+test("unset env falls back to the (long) default, not disabled or instant-exit", async (t) => {
+  // Spawn with CODEX_BROKER_IDLE_SHUTDOWN_MS unset: Number(undefined) -> NaN -> 30-min default.
+  // The broker must stay up through a short window (proving it neither disabled nor used a tiny value).
+  const binDir = makeTempDir();
+  installFakeCodex(binDir);
+  try { fs.symlinkSync(process.execPath, path.join(binDir, "node")); } catch { /* may exist */ }
+  const sockDir = makeTempDir("cxc-idle-");
+  const sock = path.join(sockDir, "broker.sock");
+  const env = buildEnv(binDir);
+  delete env.CODEX_BROKER_IDLE_SHUTDOWN_MS; // ensure truly unset -> default path
+  const child = spawnBroker(env, `unix:${sock}`);
+  t.after(() => {
+    try { child.kill("SIGKILL"); } catch { /* gone */ }
+    try { fs.rmSync(sockDir, { recursive: true, force: true }); } catch { /* ignore */ }
+  });
+  assert.equal(await waitForSocket(sock), true);
+  assert.equal(
+    await waitForExit(child, IDLE),
+    false,
+    "with the env unset the broker uses the long (30-min) default and must NOT exit quickly"
+  );
+});
+
+test("idle self-exit clears the persisted broker.json for the cwd", async (t) => {
+  // Regression for the reuseExistingBroker:true stale-endpoint path: after an autonomous idle exit,
+  // broker.json must NOT be left pointing at the removed socket (else `codex setup --json` and other
+  // reuse callers load a dead endpoint without re-probing).
+  const binDir = makeTempDir();
+  installFakeCodex(binDir);
+  try { fs.symlinkSync(process.execPath, path.join(binDir, "node")); } catch { /* may exist */ }
+  const sockDir = makeTempDir("cxc-idle-");
+  const workspace = makeTempDir("cxc-ws-"); // the broker --cwd; broker.json is keyed off this
+  const sock = path.join(sockDir, "broker.sock");
+
+  // Seed a persisted broker session for this workspace, as ensureBrokerSession would have.
+  saveBrokerSession(workspace, { endpoint: `unix:${sock}`, pidFile: null, logFile: null, sessionDir: sockDir, pid: null });
+  assert.notEqual(loadBrokerSession(workspace), null, "precondition: broker.json exists before idle exit");
+
+  const child = spawnBroker(brokerEnv(binDir, IDLE), `unix:${sock}`, workspace);
+  t.after(() => {
+    try { child.kill("SIGKILL"); } catch { /* gone */ }
+    try { fs.rmSync(sockDir, { recursive: true, force: true }); } catch { /* ignore */ }
+    try { fs.rmSync(workspace, { recursive: true, force: true }); } catch { /* ignore */ }
+  });
+
+  assert.equal(await waitForSocket(sock), true);
+  assert.equal(await waitForExit(child, IDLE * 5), true, "broker should idle-exit");
+  // After the autonomous idle exit, the persisted session must be cleared.
+  assert.equal(
+    loadBrokerSession(workspace),
+    null,
+    "broker.json must be cleared after idle self-exit so reuse callers don't load a dead endpoint"
+  );
+});