fix(broker): self-terminate on idle to reap orphaned shared brokers (#450)

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

@@ -11,6 +11,34 @@ import { parseBrokerEndpoint } from "./lib/broker-endpoint.mjs";
 
 const STREAMING_METHODS = new Set(["turn/start", "review/start", "thread/compact/start"]);
 
+// Broker-side idle timeout. When no client has been connected for this long the
+// broker self-terminates. This is the correctness backstop for stale ownership
+// records in broker.json: a co-owning session that exits without running its
+// SessionEnd hook (SIGKILL, OOM, crash, host reboot) or whose teardown skips the
+// entry on lock contention leaves its sessionId behind forever, so no future hook
+// will ever tear the broker down. Self-termination on idle is platform-independent
+// and needs no PID/liveness signal, so it covers the abnormal-exit orphan, the
+// dead-co-owner orphan, and the lock-contention skip in one mechanism. See #108,
+// #380, and #450.
+const IDLE_TIMEOUT_ENV = "CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS";
+const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000;
+
+// Resolve the idle timeout from the CLI flag, then the environment, then the
+// default. A value <= 0 (or a non-finite/negative override) disables the timeout
+// so the broker never self-terminates, which keeps the old behavior available for
+// callers that manage lifecycle themselves.
+function resolveIdleTimeoutMs(optionValue, env = process.env) {
+  const raw = optionValue ?? env[IDLE_TIMEOUT_ENV];
+  if (raw === undefined || raw === null || raw === "") {
+    return DEFAULT_IDLE_TIMEOUT_MS;
+  }
+  const parsed = Number(raw);
+  if (!Number.isFinite(parsed) || parsed < 0) {
+    return DEFAULT_IDLE_TIMEOUT_MS;
+  }
+  return parsed;
+}
+
 function buildStreamThreadIds(method, params, result) {
   const threadIds = new Set();
   if (params?.threadId) {
@@ -48,11 +76,11 @@ function writePidFile(pidFile) {
 async function main() {
   const [subcommand, ...argv] = process.argv.slice(2);
   if (subcommand !== "serve") {
-    throw new Error("Usage: node scripts/app-server-broker.mjs serve --endpoint <value> [--cwd <path>] [--pid-file <path>]");
+    throw new Error("Usage: node scripts/app-server-broker.mjs serve --endpoint <value> [--cwd <path>] [--pid-file <path>] [--idle-timeout <ms>]");
   }
 
   const { options } = parseArgs(argv, {
-    valueOptions: ["cwd", "pid-file", "endpoint"]
+    valueOptions: ["cwd", "pid-file", "endpoint", "idle-timeout"]
   });
 
   if (!options.endpoint) {
@@ -63,13 +91,39 @@ async function main() {
   const endpoint = String(options.endpoint);
   const listenTarget = parseBrokerEndpoint(endpoint);
   const pidFile = options["pid-file"] ? path.resolve(options["pid-file"]) : null;
+  const idleTimeoutMs = resolveIdleTimeoutMs(options["idle-timeout"]);
   writePidFile(pidFile);
 
   const appClient = await CodexAppServerClient.connect(cwd, { disableBroker: true });
   let activeRequestSocket = null;
   let activeStreamSocket = null;
   let activeStreamThreadIds = null;
   const sockets = new Set();
+  let idleTimer = null;
+
+  function disarmIdleTimer() {
+    if (idleTimer) {
+      clearTimeout(idleTimer);
+      idleTimer = null;
+    }
+  }
+
+  // Arm the idle timer whenever the broker has no connected clients. A live
+  // client connection means the broker is still in use, so we only count down
+  // while idle and cancel the moment a client connects. When the timer fires we
+  // shut the broker down gracefully and exit.
+  function armIdleTimer() {
+    disarmIdleTimer();
+    if (idleTimeoutMs <= 0 || sockets.size > 0) {
+      return;
+    }
+    idleTimer = setTimeout(() => {
+      idleTimer = null;
+      shutdown(server)
+        .catch(() => {})
+        .finally(() => process.exit(0));
+    }, idleTimeoutMs);
+  }
 
   function clearSocketOwnership(socket) {
     if (activeRequestSocket === socket) {
@@ -100,6 +154,7 @@ async function main() {
   }
 
   async function shutdown(server) {
+    disarmIdleTimer();
     for (const socket of sockets) {
       socket.end();
     }
@@ -117,6 +172,7 @@ async function main() {
 
   const server = net.createServer((socket) => {
     sockets.add(socket);
+    disarmIdleTimer();
     socket.setEncoding("utf8");
     let buffer = "";
 
@@ -225,11 +281,13 @@ async function main() {
     socket.on("close", () => {
       sockets.delete(socket);
       clearSocketOwnership(socket);
+      armIdleTimer();
     });
 
     socket.on("error", () => {
       sockets.delete(socket);
       clearSocketOwnership(socket);
+      armIdleTimer();
     });
   });
 
@@ -243,7 +301,12 @@ async function main() {
     process.exit(0);
   });
 
-  server.listen(listenTarget.path);
+  server.listen(listenTarget.path, () => {
+    // Start counting down immediately: a broker that is spawned but never
+    // receives a client (or whose only client connects briefly during the
+    // readiness probe) must still self-terminate instead of lingering.
+    armIdleTimer();
+  });
 }
 
 main().catch((error) => {

tests/broker-idle-timeout.test.mjs

@@ -0,0 +1,136 @@
+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 { spawn } from "node:child_process";
+import { fileURLToPath } from "node:url";
+
+import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs";
+import { makeTempDir } from "./helpers.mjs";
+import { createBrokerEndpoint, parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs";
+import { 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");
+
+function spawnBroker({ cwd, endpoint, env, idleTimeoutMs }) {
+  const args = [BROKER_SCRIPT, "serve", "--endpoint", endpoint, "--cwd", cwd];
+  if (idleTimeoutMs !== undefined) {
+    args.push("--idle-timeout", String(idleTimeoutMs));
+  }
+  return spawn(process.execPath, args, {
+    cwd,
+    env,
+    stdio: ["ignore", "pipe", "pipe"]
+  });
+}
+
+function waitForExit(child, { timeoutMs = 5000 } = {}) {
+  return new Promise((resolve, reject) => {
+    if (child.exitCode !== null || child.signalCode !== null) {
+      resolve({ code: child.exitCode, signal: child.signalCode });
+      return;
+    }
+    const timer = setTimeout(() => {
+      child.removeListener("exit", onExit);
+      reject(new Error("Timed out waiting for broker process to exit."));
+    }, timeoutMs);
+    function onExit(code, signal) {
+      clearTimeout(timer);
+      resolve({ code, signal });
+    }
+    child.once("exit", onExit);
+  });
+}
+
+function connectClient(endpoint) {
+  const target = parseBrokerEndpoint(endpoint);
+  return new Promise((resolve, reject) => {
+    const socket = net.createConnection({ path: target.path });
+    socket.once("connect", () => resolve(socket));
+    socket.once("error", reject);
+  });
+}
+
+function delay(ms) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+test("broker self-terminates after the idle timeout when no client is connected", async () => {
+  const binDir = makeTempDir();
+  installFakeCodex(binDir);
+  const sessionDir = makeTempDir("cxc-");
+  const endpoint = createBrokerEndpoint(sessionDir);
+  const child = spawnBroker({ cwd: sessionDir, endpoint, env: buildEnv(binDir), idleTimeoutMs: 300 });
+
+  try {
+    const ready = await waitForBrokerEndpoint(endpoint, 3000);
+    assert.equal(ready, true, "broker should accept connections before it times out");
+
+    const start = Date.now();
+    const result = await waitForExit(child, { timeoutMs: 5000 });
+    const elapsed = Date.now() - start;
+
+    assert.equal(result.code, 0, "broker should exit cleanly on idle timeout");
+    assert.ok(elapsed >= 200, `broker exited too early (${elapsed}ms)`);
+  } finally {
+    if (child.exitCode === null && child.signalCode === null) {
+      child.kill();
+    }
+  }
+});
+
+test("broker stays alive while a client is connected and exits after it disconnects", async () => {
+  const binDir = makeTempDir();
+  installFakeCodex(binDir);
+  const sessionDir = makeTempDir("cxc-");
+  const endpoint = createBrokerEndpoint(sessionDir);
+  const child = spawnBroker({ cwd: sessionDir, endpoint, env: buildEnv(binDir), idleTimeoutMs: 300 });
+
+  let socket = null;
+  try {
+    const ready = await waitForBrokerEndpoint(endpoint, 3000);
+    assert.equal(ready, true);
+
+    socket = await connectClient(endpoint);
+
+    // Hold the connection open well past the idle timeout; the broker must not
+    // self-terminate while a client is still connected.
+    await delay(900);
+    assert.equal(child.exitCode, null, "broker must stay alive while a client is connected");
+
+    socket.end();
+    socket = null;
+
+    const result = await waitForExit(child, { timeoutMs: 5000 });
+    assert.equal(result.code, 0, "broker should exit once the client disconnects and it goes idle");
+  } finally {
+    if (socket) {
+      socket.destroy();
+    }
+    if (child.exitCode === null && child.signalCode === null) {
+      child.kill();
+    }
+  }
+});
+
+test("broker with the idle timeout disabled keeps running while idle", async () => {
+  const binDir = makeTempDir();
+  installFakeCodex(binDir);
+  const sessionDir = makeTempDir("cxc-");
+  const endpoint = createBrokerEndpoint(sessionDir);
+  const child = spawnBroker({ cwd: sessionDir, endpoint, env: buildEnv(binDir), idleTimeoutMs: 0 });
+
+  try {
+    const ready = await waitForBrokerEndpoint(endpoint, 3000);
+    assert.equal(ready, true);
+
+    await delay(700);
+    assert.equal(child.exitCode, null, "broker must not self-terminate when the idle timeout is disabled");
+  } finally {
+    if (child.exitCode === null && child.signalCode === null) {
+      child.kill();
+    }
+  }
+});