fix: bound the lifetime of detached brokers and task workers

plugins/codex/scripts/codex-companion.mjs

@@ -24,7 +24,8 @@ import {
 import { resolveClaudeSessionPath } from "./lib/claude-session-transfer.mjs";
 import { readStdinIfPiped } from "./lib/fs.mjs";
 import { collectReviewContext, ensureGitRepository, resolveReviewTarget } from "./lib/git.mjs";
-import { binaryAvailable, terminateProcessTree } from "./lib/process.mjs";
+import { armTimeout, disarmTimeout, workerTtlMs } from "./lib/lifecycle-limits.mjs";
+import { binaryAvailable, terminateProcessTree, terminateProcessTreeAndExit } from "./lib/process.mjs";
 import { loadPromptTemplate, interpolateTemplate } from "./lib/prompts.mjs";
 import {
   generateJobId,
@@ -865,19 +866,96 @@ async function handleTaskWorker(argv) {
       logFile: storedJob.logFile ?? null
     }
   );
-  await runTrackedJob(
-    {
-      ...storedJob,
-      workspaceRoot,
-      logFile
-    },
-    () =>
-      executeTaskRun({
-        ...request,
-        onProgress: progress
-      }),
-    { logFile }
-  );
+  const releaseTtl = armWorkerTtl({ workspaceRoot, jobId: storedJob.id, storedJob, logFile });
+  try {
+    await runTrackedJob(
+      {
+        ...storedJob,
+        workspaceRoot,
+        logFile
+      },
+      () =>
+        executeTaskRun({
+          ...request,
+          onProgress: progress
+        }),
+      { logFile }
+    );
+  } finally {
+    releaseTtl();
+  }
+}
+
+/**
+ * Bound how long a detached worker may live.
+ *
+ * The worker is deliberately detached so a background task survives the session that queued it,
+ * and its immediate parent exits right after enqueue — so there is no parent to watch and nothing
+ * else that ever reclaims it. Without a ceiling a single wedged task keeps its whole process tree
+ * (app-server plus every MCP server under it) alive indefinitely.
+ *
+ * Returns a function that disarms the timer once the job finishes normally.
+ */
+/** How long the tree gets to leave on SIGTERM before the group is killed outright. */
+const WORKER_TERMINATION_GRACE_MS = 5000;
+
+function armWorkerTtl({ workspaceRoot, jobId, storedJob, logFile }) {
+  const ttlMs = workerTtlMs();
+  const timer = armTimeout(ttlMs, () => {
+    const errorMessage = `Worker exceeded its ${ttlMs}ms lifetime.`;
+
+    // Record the outcome before terminating: the process group is about to take this process down
+    // with it, and a job left at "running" with a dead pid is exactly the stale record that makes
+    // leaked workers invisible. All of it is best effort — a full disk or a deleted state
+    // directory must not be what keeps a runaway tree alive.
+    const completedAt = nowIso();
+    const terminal = {
+      status: "failed",
+      phase: "failed",
+      pid: null,
+      completedAt,
+      errorMessage
+    };
+
+    // Each of these is best effort on its own. Sharing one try means a missing log directory or a
+    // full disk would skip the terminal status too, leaving the job "running" behind a dead pid —
+    // the stale record that hides leaked workers in the first place. Status goes first, because it
+    // is the part anything else reads.
+    const attempt = (action) => {
+      try {
+        action();
+      } catch {
+        // Never let bookkeeping keep a runaway tree alive.
+      }
+    };
+
+    const recordExpiry = () => {
+      attempt(() => {
+        // Re-read rather than reusing the snapshot this timer closed over a day ago: it predates
+        // startedAt, threadId, turnId and every progress update since.
+        const current = readStoredJob(workspaceRoot, jobId) ?? storedJob;
+        writeJobFile(workspaceRoot, jobId, { ...current, ...terminal, logFile });
+      });
+      attempt(() => upsertJob(workspaceRoot, { id: jobId, ...terminal }));
+    };
+
+    recordExpiry();
+    attempt(() => appendLogLine(logFile, `${errorMessage} Terminating its process tree.`));
+
+    // Terminate the tree rather than just this process: the app-server and MCP servers underneath
+    // are the expensive part, and they do not exit on their own.
+    //
+    // SIGTERM alone is not a ceiling — a descendant that traps or ignores it keeps running, and
+    // once this worker is gone nothing is left to escalate. So the worker survives its own signal
+    // for a grace period. That means the job can finish during it and record a success over the
+    // expiry, which would be a lie: its tree is about to be killed. Writing the expiry again as
+    // the last act before the kill makes it the outcome that stands.
+    terminateProcessTreeAndExit(process.pid, {
+      graceMs: WORKER_TERMINATION_GRACE_MS,
+      beforeKill: recordExpiry
+    });
+  });
+  return () => disarmTimeout(timer);
 }
 
 async function handleStatus(argv) {

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

@@ -13,7 +13,7 @@ import process from "node:process";
 import { spawn } from "node:child_process";
 import readline from "node:readline";
 import { parseBrokerEndpoint } from "./broker-endpoint.mjs";
-import { ensureBrokerSession, loadBrokerSession } from "./broker-lifecycle.mjs";
+import { ensureBrokerSession, isBrokerEndpointReady, loadBrokerSession } from "./broker-lifecycle.mjs";
 import { terminateProcessTree } from "./process.mjs";
 
 const PLUGIN_MANIFEST_URL = new URL("../../.claude-plugin/plugin.json", import.meta.url);
@@ -22,6 +22,9 @@ const PLUGIN_MANIFEST = JSON.parse(fs.readFileSync(PLUGIN_MANIFEST_URL, "utf8"))
 export const BROKER_ENDPOINT_ENV = "CODEX_COMPANION_APP_SERVER_ENDPOINT";
 export const BROKER_BUSY_RPC_CODE = -32001;
 
+/** How long a failed connect waits for its transport to report an exit before killing it. */
+const CONNECT_CLEANUP_GRACE_MS = 5000;
+
 /** @type {ClientInfo} */
 const DEFAULT_CLIENT_INFO = {
   title: "Codex Plugin",
@@ -87,6 +90,12 @@ class AppServerClientBase {
     if (this.closed) {
       throw new Error("codex app-server client is closed.");
     }
+    // `closed` only covers a close we asked for. The transport can die on its own — between a
+    // successful connect and the very next request, for instance — and a request registered after
+    // that never resolves, because the exit that would reject it has already been reported.
+    if (this.exitResolved) {
+      throw this.exitError ?? new Error("codex app-server connection closed.");
+    }
 
     const id = this.nextId;
     this.nextId += 1;
@@ -98,7 +107,7 @@ class AppServerClientBase {
   }
 
   notify(method, params = {}) {
-    if (this.closed) {
+    if (this.closed || this.exitResolved) {
       return;
     }
     this.sendMessage({ method, params });
@@ -175,6 +184,20 @@ class AppServerClientBase {
     this.resolveExit(undefined);
   }
 
+  /**
+   * Settle the exit state when initialization failed before a transport existed.
+   *
+   * `close()` ends by awaiting `exitPromise`, and only a live transport ever resolves it. But
+   * initialization can fail before one is created — a malformed endpoint rejects while being
+   * parsed, a spawn can throw — and closing then waits for an exit that nothing will report,
+   * turning a configuration error into a hang.
+   */
+  settleExitIfNoTransport(hasTransport) {
+    if (!hasTransport) {
+      this.handleExit(this.exitError);
+    }
+  }
+
   sendMessage(_message) {
     throw new Error("sendMessage must be implemented by subclasses.");
   }
@@ -262,6 +285,7 @@ class SpawnedCodexAppServerClient extends AppServerClientBase {
       }, 50).unref?.();
     }
 
+    this.settleExitIfNoTransport(Boolean(this.proc));
     await this.exitPromise;
   }
 
@@ -319,6 +343,7 @@ class BrokerCodexAppServerClient extends AppServerClientBase {
     if (this.socket) {
       this.socket.end();
     }
+    this.settleExitIfNoTransport(Boolean(this.socket));
     await this.exitPromise;
   }
 
@@ -338,7 +363,22 @@ export class CodexAppServerClient {
     if (!options.disableBroker) {
       brokerEndpoint = options.brokerEndpoint ?? options.env?.[BROKER_ENDPOINT_ENV] ?? process.env[BROKER_ENDPOINT_ENV] ?? null;
       if (!brokerEndpoint && options.reuseExistingBroker) {
-        brokerEndpoint = loadBrokerSession(cwd)?.endpoint ?? null;
+        // Probe before trusting the record. A broker that was killed rather than shut down leaves
+        // its session behind, and connecting to that endpoint surfaces ENOENT/ECONNREFUSED as a
+        // failure of whatever the caller was doing — an authentication check, most visibly —
+        // rather than as a broker that is simply gone. Discard it and fall through to spawning.
+        // Probe before trusting the record. A broker that was killed rather than shut down leaves
+        // its session behind, and connecting to that endpoint surfaces ENOENT/ECONNREFUSED as a
+        // failure of whatever the caller was doing — an authentication check, most visibly —
+        // rather than as a broker that is simply gone. Fall through to spawning instead.
+        //
+        // The record itself is left alone: `status` and `setup` report a recorded shared runtime
+        // whether or not it answers, and reclaiming a dead broker's files belongs to
+        // `ensureBrokerSession`, which is the path that actually replaces it.
+        const persisted = loadBrokerSession(cwd)?.endpoint ?? null;
+        if (persisted && (await isBrokerEndpointReady(persisted))) {
+          brokerEndpoint = persisted;
+        }
       }
       if (!brokerEndpoint && !options.reuseExistingBroker) {
         const brokerSession = await ensureBrokerSession(cwd, { env: options.env });
@@ -348,7 +388,31 @@ export class CodexAppServerClient {
     const client = brokerEndpoint
       ? new BrokerCodexAppServerClient(cwd, { ...options, brokerEndpoint })
       : new SpawnedCodexAppServerClient(cwd, options);
-    await client.initialize();
+    try {
+      await client.initialize();
+    } catch (error) {
+      // initialize() has usually already spawned the app-server, and with it every configured MCP
+      // server. The caller never receives this object, so this is the only chance to reclaim them.
+      //
+      // Bounded, because close() waits on the transport reporting its exit: an app-server that
+      // answered with an error but ignores EOF and SIGTERM would otherwise swallow the original
+      // failure entirely and leave the caller waiting forever. Whatever is still up after the
+      // grace gets killed outright.
+      await Promise.race([
+        client.close().catch(() => {}),
+        new Promise((resolve) => {
+          setTimeout(resolve, CONNECT_CLEANUP_GRACE_MS).unref?.();
+        })
+      ]);
+      if (client.proc && client.proc.exitCode === null && !client.proc.killed) {
+        try {
+          client.proc.kill("SIGKILL");
+        } catch {
+          // Already gone.
+        }
+      }
+      throw error;
+    }
     return client;
   }
 }

plugins/codex/scripts/lib/broker-lifecycle.mjs

@@ -6,6 +6,7 @@ import process from "node:process";
 import { spawn } from "node:child_process";
 import { fileURLToPath } from "node:url";
 import { createBrokerEndpoint, parseBrokerEndpoint } from "./broker-endpoint.mjs";
+import { isProcessAlive } from "./process.mjs";
 import { resolveStateDir } from "./state.mjs";
 
 export const PID_FILE_ENV = "CODEX_COMPANION_APP_SERVER_PID_FILE";
@@ -58,7 +59,10 @@ export async function sendBrokerShutdown(endpoint) {
 
 export function spawnBrokerProcess({ scriptPath, cwd, endpoint, pidFile, logFile, env = process.env }) {
   const logFd = fs.openSync(logFile, "a");
-  const child = spawn(process.execPath, [scriptPath, "serve", "--endpoint", endpoint, "--cwd", cwd, "--pid-file", pidFile], {
+  // The broker is told its own log path, not just handed the descriptor: it has to be able to
+  // clean up after itself without consulting the shared session record, which may by then name a
+  // successor.
+  const child = spawn(process.execPath, [scriptPath, "serve", "--endpoint", endpoint, "--cwd", cwd, "--pid-file", pidFile, "--log-file", logFile], {
     cwd,
     env,
     detached: true,
@@ -99,7 +103,8 @@ export function clearBrokerSession(cwd) {
   }
 }
 
-async function isBrokerEndpointReady(endpoint) {
+/** Whether something is actually listening on a persisted endpoint, as opposed to merely recorded. */
+export async function isBrokerEndpointReady(endpoint) {
   if (!endpoint) {
     return false;
   }
@@ -116,7 +121,14 @@ export async function ensureBrokerSession(cwd, options = {}) {
     return existing;
   }
 
-  if (existing) {
+  // Only reclaim a broker we can prove is gone. The probe above waits 150ms, which a live but busy
+  // broker can miss, and on the normal path `killProcess` is null — so tearing down here would
+  // delete a running broker's socket without stopping it, leaving it unreachable while it still
+  // holds its app-server and every MCP server underneath.
+  //
+  // A live one is left exactly as it is. Once the replacement below takes over it has no clients,
+  // so its own idle shutdown reclaims both the process and its files.
+  if (existing && !isProcessAlive(existing.pid)) {
     teardownBrokerSession({
       endpoint: existing.endpoint ?? null,
       pidFile: existing.pidFile ?? null,

plugins/codex/scripts/lib/lifecycle-limits.mjs

@@ -0,0 +1,81 @@
+import process from "node:process";
+
+const DEFAULT_BROKER_IDLE_SHUTDOWN_MS = 10 * 60 * 1000;
+const DEFAULT_BROKER_STARTUP_TIMEOUT_MS = 5 * 60 * 1000;
+const DEFAULT_WORKER_TTL_MS = 24 * 60 * 60 * 1000;
+
+/** `setTimeout` truncates anything larger to a 32-bit int, firing almost immediately instead. */
+const MAX_TIMEOUT_MS = 2 ** 31 - 1;
+
+function readDurationMs(raw, fallback) {
+  if (typeof raw !== "string" || raw.trim() === "") {
+    return fallback;
+  }
+  const parsed = Number(raw.trim());
+  if (!Number.isFinite(parsed) || parsed < 0) {
+    return fallback;
+  }
+  // Clamp rather than reject: an operator asking for 30 days means "effectively never", and
+  // letting that overflow into ~1ms would kill exactly what they meant to protect.
+  return Math.min(parsed, MAX_TIMEOUT_MS);
+}
+
+/**
+ * Schedule `onExpiry`, treating `0` as "no limit".
+ *
+ * Every limit here documents `0` as disabled, but `setTimeout(fn, 0)` means "next tick" — passing
+ * a disabled limit straight through would fire the guard immediately instead of never. Returns
+ * the timer, or `null` when disabled; pass that back to {@link disarmTimeout}.
+ */
+export function armTimeout(ms, onExpiry) {
+  if (!ms) {
+    return null;
+  }
+  const timer = setTimeout(onExpiry, ms);
+  // Never hold the event loop open on the guard alone.
+  timer.unref?.();
+  return timer;
+}
+
+/** Counterpart to {@link armTimeout}; tolerates the `null` a disabled limit produces. */
+export function disarmTimeout(timer) {
+  if (timer) {
+    clearTimeout(timer);
+  }
+}
+
+/**
+ * How long the broker may sit with no connected client before shutting itself down, taking its
+ * app-server (and every MCP server under it) with it.
+ *
+ * A broker outlives the client that spawned it, so without this it survives a crashed or
+ * timed-out SessionEnd hook and holds that whole tree alive indefinitely. `0` disables the timer.
+ */
+export function brokerIdleShutdownMs(env = process.env) {
+  return readDurationMs(env.CODEX_BROKER_IDLE_SHUTDOWN_MS, DEFAULT_BROKER_IDLE_SHUTDOWN_MS);
+}
+
+/**
+ * How long the broker may spend starting up before it gives up and takes its tree down.
+ *
+ * Until the broker is listening there is no idle timer, and the client that spawned it stops
+ * waiting after a couple of seconds without killing anything — so a wedged app-server or MCP
+ * startup would strand the whole tree permanently. The default is generous because a cold start
+ * legitimately spawns every configured MCP server. `0` disables the bound.
+ */
+export function brokerStartupTimeoutMs(env = process.env) {
+  return readDurationMs(env.CODEX_BROKER_STARTUP_TIMEOUT_MS, DEFAULT_BROKER_STARTUP_TIMEOUT_MS);
+}
+
+/**
+ * Ceiling on a detached background worker's wall-clock lifetime.
+ *
+ * The worker is deliberately detached so a background task survives the session that queued it,
+ * and its immediate parent exits right after enqueue — so there is no parent to watch and nothing
+ * else that reclaims it. This is a runaway guard, not a task deadline: the default is far longer
+ * than any real background task, and still reclaims the multi-day trees that prompted it. `0`
+ * disables the ceiling.
+ */
+export function workerTtlMs(env = process.env) {
+  return readDurationMs(env.CODEX_TASK_WORKER_TTL_MS, DEFAULT_WORKER_TTL_MS);
+}