fix(broker): stop tearing down a live broker that misses the readiness probe

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

@@ -99,23 +99,108 @@ export function clearBrokerSession(cwd) {
   }
 }
 
+// A broker that is mid-turn keeps its event loop busy, so 150 ms is not enough
+// budget for it to accept a probe connection. Failing the probe used to be treated
+// as "the broker is gone", which then tore down a perfectly healthy broker.
+const BROKER_READY_PROBE_TIMEOUT_MS = 3000;
+
+// Stale-lock timeout for the broker creation lock below.
+const BROKER_LOCK_STALE_MS = 30000;
+const BROKER_LOCK_WAIT_MS = 15000;
+
 async function isBrokerEndpointReady(endpoint) {
   if (!endpoint) {
     return false;
   }
   try {
-    return await waitForBrokerEndpoint(endpoint, 150);
+    return await waitForBrokerEndpoint(endpoint, BROKER_READY_PROBE_TIMEOUT_MS);
   } catch {
     return false;
   }
 }
 
+// A failed probe only proves the broker did not answer in time; it does not prove
+// the process is gone. Never tear down a broker whose process is still alive.
+function isBrokerProcessAlive(pid) {
+  if (!Number.isFinite(pid) || pid <= 0) {
+    return false;
+  }
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch (error) {
+    // ESRCH is the only code that proves the process is gone. EPERM (and anything
+    // else) means we cannot tell, so assume it is alive and leave it alone.
+    return error?.code !== "ESRCH";
+  }
+}
+
+function brokerLockPath(cwd) {
+  return path.join(resolveStateDir(cwd), "broker.lock");
+}
+
+// Concurrent clients can all observe `loadBrokerSession() === null` and each spawn a
+// broker; the last writer wins and the other brokers are orphaned together with the
+// app-servers (and in-flight turns) they own. Serialize the check-then-create window.
+async function withBrokerLock(cwd, fn) {
+  const lockFile = brokerLockPath(cwd);
+  fs.mkdirSync(path.dirname(lockFile), { recursive: true });
+
+  const deadline = Date.now() + BROKER_LOCK_WAIT_MS;
+  let held = false;
+  while (Date.now() < deadline) {
+    try {
+      const fd = fs.openSync(lockFile, "wx");
+      fs.writeSync(fd, String(process.pid));
+      fs.closeSync(fd);
+      held = true;
+      break;
+    } catch (error) {
+      if (error?.code !== "EEXIST") {
+        throw error;
+      }
+      try {
+        if (Date.now() - fs.statSync(lockFile).mtimeMs > BROKER_LOCK_STALE_MS) {
+          fs.unlinkSync(lockFile);
+          continue;
+        }
+      } catch {
+        // The holder released it between statSync and now; retry.
+      }
+      await new Promise((resolve) => setTimeout(resolve, 100));
+    }
+  }
+
+  try {
+    // Falling through without the lock keeps the previous (racy) behaviour rather
+    // than failing the caller outright.
+    return await fn();
+  } finally {
+    if (held) {
+      try {
+        fs.unlinkSync(lockFile);
+      } catch {
+        // Already removed by a stale-lock sweep.
+      }
+    }
+  }
+}
+
 export async function ensureBrokerSession(cwd, options = {}) {
+  return withBrokerLock(cwd, () => ensureBrokerSessionLocked(cwd, options));
+}
+
+async function ensureBrokerSessionLocked(cwd, options = {}) {
   const existing = loadBrokerSession(cwd);
   if (existing && (await isBrokerEndpointReady(existing.endpoint))) {
     return existing;
   }
 
+  if (existing && isBrokerProcessAlive(existing.pid)) {
+    // Alive but slow to answer: it is busy, not dead. Reuse it.
+    return existing;
+  }
+
   if (existing) {
     teardownBrokerSession({
       endpoint: existing.endpoint ?? null,

tests/broker-lifecycle.test.mjs

@@ -0,0 +1,83 @@
+import fs from "node:fs";
+import path from "node:path";
+import test from "node:test";
+import assert from "node:assert/strict";
+
+import { makeTempDir } from "./helpers.mjs";
+import { ensureBrokerSession, saveBrokerSession, loadBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs";
+
+test("ensureBrokerSession reuses a live broker whose endpoint does not answer the probe", async () => {
+  const workspace = makeTempDir();
+  // Use this test process as the "broker": it is unquestionably alive and it is not
+  // listening on the endpoint, so the readiness probe must fail. A broker that is
+  // mid-turn behaves the same way — its event loop is busy and it misses the probe
+  // window. Failing the probe must not be treated as "the broker is gone".
+  // Keep the broker's own session dir separate from the workspace: the buggy path
+  // unlinks sessionDir, and we still need the workspace to exist afterwards.
+  const sessionDir = path.join(workspace, "broker-session");
+  fs.mkdirSync(sessionDir, { recursive: true });
+  const session = {
+    endpoint: "unix:/tmp/codex-plugin-test-not-listening/broker.sock",
+    pidFile: path.join(sessionDir, "broker.pid"),
+    logFile: path.join(sessionDir, "broker.log"),
+    sessionDir,
+    pid: process.pid
+  };
+  fs.writeFileSync(session.pidFile, String(process.pid));
+  fs.writeFileSync(session.logFile, "");
+  saveBrokerSession(workspace, session);
+
+  let killed = false;
+  const result = await ensureBrokerSession(workspace, {
+    killProcess: () => {
+      killed = true;
+    }
+  });
+
+  assert.equal(killed, false, "a live broker must never be killed after a failed probe");
+  assert.equal(result?.pid, process.pid, "the live broker session must be reused");
+  assert.equal(loadBrokerSession(workspace)?.pid, process.pid, "its state must be kept");
+  assert.equal(fs.existsSync(session.logFile), true, "its log must not be unlinked");
+});
+
+test("ensureBrokerSession serializes concurrent callers so only one broker is created", async () => {
+  const workspace = makeTempDir();
+  const created = [];
+  let counter = 0;
+
+  // Both callers start with no persisted session. Without the lock they each spawn a
+  // broker and the last writer wins, orphaning the other broker and its app-server.
+  const options = {
+    createBrokerEndpoint: (sessionDir) => `unix:${sessionDir}/broker.sock`,
+    scriptPath: path.join(workspace, "fake-broker.mjs"),
+    timeoutMs: 1000
+  };
+  fs.writeFileSync(
+    options.scriptPath,
+    [
+      'import net from "node:net";',
+      'const endpointIndex = process.argv.indexOf("--endpoint");',
+      'const target = process.argv[endpointIndex + 1].replace(/^unix:/, "");',
+      "net.createServer(() => {}).listen(target);",
+      "setTimeout(() => {}, 60000);"
+    ].join("\n")
+  );
+
+  const spy = { ...options, onSpawn: () => created.push(++counter) };
+  const [a, b] = await Promise.all([
+    ensureBrokerSession(workspace, spy),
+    ensureBrokerSession(workspace, spy)
+  ]);
+
+  assert.ok(a, "first caller must get a session");
+  assert.ok(b, "second caller must get a session");
+  assert.equal(a.endpoint, b.endpoint, "concurrent callers must share one broker");
+
+  for (const session of [a, b]) {
+    try {
+      if (session?.pid) process.kill(session.pid);
+    } catch {
+      // Already gone.
+    }
+  }
+});