From 37cfcaa5a06f37582e7a87ee19f7e2f604d18fa9 Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Tue, 3 Nov 2026 15:08:02 -0600 Subject: [PATCH 01/16] fix: expire idle companion brokers and serialize startup --- README.md | 7 + plugins/codex/scripts/app-server-broker.mjs | 39 +++- .../codex/scripts/lib/broker-lifecycle.mjs | 64 +++++- tests/broker-lifecycle.test.mjs | 195 ++++++++++++++++++ 4 files changed, 297 insertions(+), 8 deletions(-) create mode 100644 tests/broker-lifecycle.test.mjs diff --git a/README.md b/README.md index 937a3037b..9a4caa737 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,13 @@ One simple first run is: ## Usage +Commands in a workspace reuse a shared Codex app-server. It shuts down after +60 seconds without connected clients, pending requests, or running turns, so +completed conversations can be archived from Codex even if Claude Code's session +cleanup did not run. Later commands start another server and can resume saved +conversations. Set `CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS` to a positive number +of milliseconds to change this idle interval. + ### `/codex:review` Runs a normal Codex review on your current work. It gives you the same quality of code review as running `/review` inside Codex directly. diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 1954274fe..e8b9aed0e 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -70,6 +70,24 @@ async function main() { let activeStreamSocket = null; let activeStreamThreadIds = null; const sockets = new Set(); + const activeTurns = new Set(); + let pendingRequests = 0; + let idleTimer = null; + let shuttingDown = false; + const configuredIdleMs = Number(process.env.CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS); + const idleTimeoutMs = Number.isFinite(configuredIdleMs) && configuredIdleMs > 0 ? configuredIdleMs : 60000; + + function updateIdleTimer() { + clearTimeout(idleTimer); + idleTimer = null; + if (shuttingDown || sockets.size || pendingRequests || activeTurns.size) return; + idleTimer = setTimeout(() => { + shutdown(server).then(() => process.exit(0)).catch((error) => { + process.stderr.write(`${error.message}\n`); + process.exit(1); + }); + }, idleTimeoutMs); + } function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { @@ -82,6 +100,11 @@ async function main() { } function routeNotification(message) { + // A disconnected caller must not make its running turn eligible for shutdown. + const turnKey = `${message.params?.threadId}:${message.params?.turn?.id}`; + if (message.method === "turn/started") activeTurns.add(turnKey); + if (message.method === "turn/completed") activeTurns.delete(turnKey); + updateIdleTimer(); const target = activeRequestSocket ?? activeStreamSocket; if (!target) { return; @@ -100,6 +123,9 @@ async function main() { } async function shutdown(server) { + if (shuttingDown) return; + shuttingDown = true; + clearTimeout(idleTimer); for (const socket of sockets) { socket.end(); } @@ -117,6 +143,7 @@ async function main() { const server = net.createServer((socket) => { sockets.add(socket); + updateIdleTimer(); socket.setEncoding("utf8"); let buffer = ""; @@ -182,6 +209,7 @@ async function main() { } if (allowInterruptDuringActiveStream) { + pendingRequests += 1; try { const result = await appClient.request(message.method, message.params ?? {}); send(socket, { id: message.id, result }); @@ -190,12 +218,16 @@ async function main() { id: message.id, error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) }); + } finally { + pendingRequests -= 1; + updateIdleTimer(); } continue; } const isStreaming = STREAMING_METHODS.has(message.method); activeRequestSocket = socket; + pendingRequests += 1; try { const result = await appClient.request(message.method, message.params ?? {}); @@ -218,6 +250,9 @@ async function main() { if (activeStreamSocket === socket && !isStreaming) { activeStreamSocket = null; } + } finally { + pendingRequests -= 1; + updateIdleTimer(); } } }); @@ -225,11 +260,13 @@ async function main() { socket.on("close", () => { sockets.delete(socket); clearSocketOwnership(socket); + updateIdleTimer(); }); socket.on("error", () => { sockets.delete(socket); clearSocketOwnership(socket); + updateIdleTimer(); }); }); @@ -243,7 +280,7 @@ async function main() { process.exit(0); }); - server.listen(listenTarget.path); + server.listen(listenTarget.path, updateIdleTimer); } main().catch((error) => { diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index ef763819c..36a4538cb 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -5,6 +5,7 @@ import path from "node:path"; import process from "node:process"; import { spawn } from "node:child_process"; import { fileURLToPath } from "node:url"; +import { randomUUID } from "node:crypto"; import { createBrokerEndpoint, parseBrokerEndpoint } from "./broker-endpoint.mjs"; import { resolveStateDir } from "./state.mjs"; @@ -40,19 +41,22 @@ export async function waitForBrokerEndpoint(endpoint, timeoutMs = 2000) { return false; } -export async function sendBrokerShutdown(endpoint) { +export async function sendBrokerShutdown(endpoint, { timeoutMs = 1000 } = {}) { await new Promise((resolve) => { const socket = connectToEndpoint(endpoint); + const finish = () => { + clearTimeout(timer); + socket.destroy(); + resolve(); + }; + const timer = setTimeout(finish, timeoutMs); socket.setEncoding("utf8"); socket.on("connect", () => { socket.write(`${JSON.stringify({ id: 1, method: "broker/shutdown", params: {} })}\n`); }); - socket.on("data", () => { - socket.end(); - resolve(); - }); - socket.on("error", resolve); - socket.on("close", resolve); + socket.on("data", finish); + socket.on("error", finish); + socket.on("close", finish); }); } @@ -111,6 +115,52 @@ async function isBrokerEndpointReady(endpoint) { } export async function ensureBrokerSession(cwd, options = {}) { + // Serialize the readiness check, spawn and state publication across processes. + // Otherwise simultaneous first commands each leave a detached broker behind. + const stateFile = resolveBrokerStateFile(cwd); + fs.mkdirSync(path.dirname(stateFile), { recursive: true }); + const lockFile = `${stateFile}.lock`; + const owner = { pid: process.pid, token: randomUUID() }; + const deadline = Date.now() + (options.timeoutMs ?? 2000) + 3000; + while (true) { + try { + const fd = fs.openSync(lockFile, "wx"); + try { + fs.writeFileSync(fd, JSON.stringify(owner)); + } finally { + fs.closeSync(fd); + } + break; + } catch (error) { + if (error.code !== "EEXIST") throw error; + try { + const text = fs.readFileSync(lockFile, "utf8"); + let stale = false; + if (text) { + const previous = JSON.parse(text); + if (Number.isInteger(previous.pid) && previous.pid > 0) { + try { process.kill(previous.pid, 0); } catch (err) { stale = err.code === "ESRCH"; } + } + } else { + // A process may die between exclusive creation and writing its PID. + stale = Date.now() - fs.statSync(lockFile).mtimeMs > 30000; + } + if (stale && fs.readFileSync(lockFile, "utf8") === text) fs.unlinkSync(lockFile); + } catch (err) { + if (err.code !== "ENOENT") throw err; + } + if (Date.now() >= deadline) throw new Error("Timed out waiting for Codex broker startup lock."); + await new Promise((resolve) => setTimeout(resolve, 50)); + } + } + try { + return await startBrokerSession(cwd, options); + } finally { + if (JSON.parse(fs.readFileSync(lockFile, "utf8")).token === owner.token) fs.unlinkSync(lockFile); + } +} + +async function startBrokerSession(cwd, options) { const existing = loadBrokerSession(cwd); if (existing && (await isBrokerEndpointReady(existing.endpoint))) { return existing; diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs new file mode 100644 index 000000000..1d285ddd3 --- /dev/null +++ b/tests/broker-lifecycle.test.mjs @@ -0,0 +1,195 @@ +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 { makeTempDir, writeExecutable } from "./helpers.mjs"; +import { ensureBrokerSession, sendBrokerShutdown } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; +import { terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; +import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; + +const LIFECYCLE_URL = new URL("../plugins/codex/scripts/lib/broker-lifecycle.mjs", import.meta.url).href; + +async function waitFor(predicate, timeoutMs = 3000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + assert.fail("Timed out waiting for broker lifecycle transition"); +} + +function alive(pid) { + try { process.kill(pid, 0); return true; } catch { return false; } +} + +function connect(endpoint) { + const socket = net.createConnection({ path: parseBrokerEndpoint(endpoint).path }); + let buffer = ""; + let nextId = 1; + const pending = new Map(); + socket.setEncoding("utf8"); + socket.on("data", (chunk) => { + buffer += chunk; + let end; + while ((end = buffer.indexOf("\n")) !== -1) { + const message = JSON.parse(buffer.slice(0, end)); + buffer = buffer.slice(end + 1); + const request = pending.get(message.id); + if (request) { + pending.delete(message.id); + if (message.error) request.reject(new Error(message.error.message)); + else request.resolve(message.result); + } + } + }); + return { + socket, + request(method, params = {}) { + const id = nextId++; + return new Promise((resolve, reject) => { + pending.set(id, { resolve, reject }); + socket.write(`${JSON.stringify({ id, method, params })}\n`); + }); + } + }; +} + +async function fixture(t) { + const root = makeTempDir(); + const bin = path.join(root, "bin"); + fs.mkdirSync(bin); + writeExecutable(path.join(bin, "codex"), `#!/usr/bin/env node +const fs = require('node:fs'); +const readline = require('node:readline'); +fs.appendFileSync(process.env.TEST_STARTS, process.pid + '\\n'); +const send = (message) => process.stdout.write(JSON.stringify(message) + '\\n'); +readline.createInterface({ input: process.stdin }).on('line', (line) => { + const m = JSON.parse(line); + if (m.id === undefined) return; + const reply = (result) => send({ id: m.id, result }); + if (m.method === 'turn/start') { + reply({ turn: { id: 'turn_1' } }); + send({ method: 'turn/started', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }); + setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }), 700); + } else if (m.method === 'test/slow') { + setTimeout(() => reply({ done: true }), 700); + } else reply({}); +}); +`); + const env = { + ...process.env, + PATH: `${bin}${path.delimiter}${process.env.PATH}`, + TEST_STARTS: path.join(root, "starts"), + CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS: "200" + }; + const sessions = []; + t.after(async () => { + for (const session of sessions) { + if (alive(session.pid)) terminateProcessTree(session.pid); + } + await waitFor(() => sessions.every((s) => !alive(s.pid))); + fs.rmSync(root, { recursive: true, force: true }); + fs.rmSync(resolveStateDir(root), { recursive: true, force: true }); + for (const session of sessions) fs.rmSync(session.sessionDir, { recursive: true, force: true }); + }); + return { root, env, sessions, start: async () => { + const session = await ensureBrokerSession(root, { env }); + assert.ok(session); + sessions.push(session); + return session; + } }; +} + +test("idle broker exits and releases its app-server after the client disconnects", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("initialize"); + const childPid = Number(fs.readFileSync(f.env.TEST_STARTS, "utf8").trim()); + await new Promise((resolve) => setTimeout(resolve, 300)); + assert.ok(alive(session.pid), "connected clients keep the broker alive"); + client.socket.destroy(); + await waitFor(() => !alive(session.pid) && !alive(childPid), 1500); + const replacement = await f.start(); + assert.notEqual(replacement.pid, session.pid, "next command can start a replacement broker"); +}); + +test("disconnecting a client does not expire an active turn", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("turn/start", { threadId: "thr_1" }); + client.socket.destroy(); + await new Promise((resolve) => setTimeout(resolve, 400)); + assert.ok(alive(session.pid), "turn outlives both the socket and idle timeout"); + await waitFor(() => !alive(session.pid), 1800); +}); + +test("disconnecting a client does not expire an in-flight RPC", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + const client = connect(session.endpoint); + client.request("test/slow"); + await new Promise((resolve) => client.socket.write("", resolve)); + await new Promise((resolve) => setTimeout(resolve, 50)); + client.socket.destroy(); + await new Promise((resolve) => setTimeout(resolve, 400)); + assert.ok(alive(session.pid), "pending request outlives idle timeout"); + await waitFor(() => !alive(session.pid), 1800); +}); + +test("concurrent processes start only one broker for a workspace", { timeout: 10000 }, async (t) => { + const f = await fixture(t); + f.env.CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS = "3000"; + const launch = () => new Promise((resolve, reject) => { + const child = spawn(process.execPath, ["--input-type=module", "-e", ` + import { ensureBrokerSession } from ${JSON.stringify(LIFECYCLE_URL)}; + const session = await ensureBrokerSession(process.cwd()); + console.log(JSON.stringify(session)); + `], { cwd: f.root, env: f.env, stdio: ["ignore", "pipe", "pipe"] }); + let stdout = "", stderr = ""; + child.stdout.on("data", (c) => { stdout += c; }); + child.stderr.on("data", (c) => { stderr += c; }); + child.on("error", reject); + child.on("exit", (code) => { + if (code !== 0) return reject(new Error(stderr)); + const session = JSON.parse(stdout); + if (session) f.sessions.push(session); + resolve(session); + }); + }); + const sessions = await Promise.all([launch(), launch()]); + assert.ok(sessions.every(Boolean)); + assert.equal(sessions[0].endpoint, sessions[1].endpoint); + assert.equal(fs.readFileSync(f.env.TEST_STARTS, "utf8").trim().split("\n").length, 1); +}); + +test("shutdown is bounded when a broker accepts a connection but never replies", { timeout: 3000 }, async (t) => { + const root = makeTempDir(); + const endpoint = process.platform === "win32" ? `pipe:\\\\.\\pipe\\${path.basename(root)}` : `unix:${path.join(root, "broker.sock")}`; + const sockets = new Set(); + const server = net.createServer((socket) => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); }); + await new Promise((resolve) => server.listen(parseBrokerEndpoint(endpoint).path, resolve)); + t.after(() => { + for (const socket of sockets) socket.destroy(); + server.close(); + fs.rmSync(root, { recursive: true, force: true }); + }); + await sendBrokerShutdown(endpoint, { timeoutMs: 100 }); +}); + +test("startup recovers a lock left by a process that exited", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + await new Promise((resolve) => child.on("exit", resolve)); + const stateDir = resolveStateDir(f.root); + fs.mkdirSync(stateDir, { recursive: true }); + fs.writeFileSync(path.join(stateDir, "broker.json.lock"), JSON.stringify({ pid: child.pid, token: "dead-owner" })); + const session = await f.start(); + assert.ok(alive(session.pid)); + assert.equal(fs.existsSync(path.join(stateDir, "broker.json.lock")), false); +}); From 1b470f3e60324a84585b2714a7181a3e6169f6fa Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Tue, 3 Nov 2026 21:27:20 -0600 Subject: [PATCH 02/16] refactor: simplify broker request and lock handling --- plugins/codex/scripts/app-server-broker.mjs | 28 ++++------ .../codex/scripts/lib/broker-lifecycle.mjs | 54 ++++++++++++------- 2 files changed, 46 insertions(+), 36 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index e8b9aed0e..8b329d5b7 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -208,30 +208,19 @@ async function main() { continue; } - if (allowInterruptDuringActiveStream) { - pendingRequests += 1; - try { - const result = await appClient.request(message.method, message.params ?? {}); - send(socket, { id: message.id, result }); - } catch (error) { - send(socket, { - id: message.id, - error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) - }); - } finally { - pendingRequests -= 1; - updateIdleTimer(); - } - continue; - } - const isStreaming = STREAMING_METHODS.has(message.method); - activeRequestSocket = socket; + if (!allowInterruptDuringActiveStream) { + activeRequestSocket = socket; + } pendingRequests += 1; try { const result = await appClient.request(message.method, message.params ?? {}); send(socket, { id: message.id, result }); + // Interrupts share request bookkeeping without taking stream ownership. + if (allowInterruptDuringActiveStream) { + continue; + } if (isStreaming) { activeStreamSocket = socket; activeStreamThreadIds = buildStreamThreadIds(message.method, message.params ?? {}, result); @@ -244,6 +233,9 @@ async function main() { id: message.id, error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) }); + if (allowInterruptDuringActiveStream) { + continue; + } if (activeRequestSocket === socket) { activeRequestSocket = null; } diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index 36a4538cb..4bb7d0748 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -114,6 +114,33 @@ async function isBrokerEndpointReady(endpoint) { } } +function removeStaleBrokerLock(lockFile) { + try { + const text = fs.readFileSync(lockFile, "utf8"); + let stale = false; + if (text) { + const owner = JSON.parse(text); + if (Number.isInteger(owner.pid) && owner.pid > 0) { + try { + process.kill(owner.pid, 0); + } catch (error) { + stale = error.code === "ESRCH"; + } + } + } else { + // A process may die between exclusive creation and writing its PID. + stale = Date.now() - fs.statSync(lockFile).mtimeMs > 30000; + } + if (stale && fs.readFileSync(lockFile, "utf8") === text) { + fs.unlinkSync(lockFile); + } + } catch (error) { + if (error.code !== "ENOENT") { + throw error; + } + } +} + export async function ensureBrokerSession(cwd, options = {}) { // Serialize the readiness check, spawn and state publication across processes. // Otherwise simultaneous first commands each leave a detached broker behind. @@ -132,31 +159,22 @@ export async function ensureBrokerSession(cwd, options = {}) { } break; } catch (error) { - if (error.code !== "EEXIST") throw error; - try { - const text = fs.readFileSync(lockFile, "utf8"); - let stale = false; - if (text) { - const previous = JSON.parse(text); - if (Number.isInteger(previous.pid) && previous.pid > 0) { - try { process.kill(previous.pid, 0); } catch (err) { stale = err.code === "ESRCH"; } - } - } else { - // A process may die between exclusive creation and writing its PID. - stale = Date.now() - fs.statSync(lockFile).mtimeMs > 30000; - } - if (stale && fs.readFileSync(lockFile, "utf8") === text) fs.unlinkSync(lockFile); - } catch (err) { - if (err.code !== "ENOENT") throw err; + if (error.code !== "EEXIST") { + throw error; + } + removeStaleBrokerLock(lockFile); + if (Date.now() >= deadline) { + throw new Error("Timed out waiting for Codex broker startup lock."); } - if (Date.now() >= deadline) throw new Error("Timed out waiting for Codex broker startup lock."); await new Promise((resolve) => setTimeout(resolve, 50)); } } try { return await startBrokerSession(cwd, options); } finally { - if (JSON.parse(fs.readFileSync(lockFile, "utf8")).token === owner.token) fs.unlinkSync(lockFile); + if (JSON.parse(fs.readFileSync(lockFile, "utf8")).token === owner.token) { + fs.unlinkSync(lockFile); + } } } From 65b1926b4bb67dc4f1422972cd91d5061414aa5a Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Thu, 5 Nov 2026 16:29:51 -0600 Subject: [PATCH 03/16] fix: close broker listener before draining and recover stale state Idle shutdown now stops admitting clients before the slow app-server close, so a command arriving mid-drain starts a replacement broker instead of failing its first request. Reuse-existing callers verify the saved endpoint is reachable and fall back to a direct app-server when the broker expired. Startup lock recovery treats partial owner JSON like an empty lock (age check), and lock release no longer throws if the lock was reclaimed. Co-Authored-By: Claude Sonnet 5.5 --- plugins/codex/scripts/app-server-broker.mjs | 5 +- plugins/codex/scripts/lib/app-server.mjs | 6 ++- .../codex/scripts/lib/broker-lifecycle.mjs | 38 ++++++++++----- tests/broker-lifecycle.test.mjs | 47 ++++++++++++++++++- 4 files changed, 79 insertions(+), 17 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 8b329d5b7..c7f26fdbe 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -126,11 +126,14 @@ async function main() { if (shuttingDown) return; shuttingDown = true; clearTimeout(idleTimer); + // Stop admitting clients before the app-server close, which can take seconds. + // A client admitted while draining would initialize fine, then fail its first request. + const closed = 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 closed; if (listenTarget.kind === "unix" && fs.existsSync(listenTarget.path)) { fs.unlinkSync(listenTarget.path); } diff --git a/plugins/codex/scripts/lib/app-server.mjs b/plugins/codex/scripts/lib/app-server.mjs index 72b30a764..49163d756 100644 --- a/plugins/codex/scripts/lib/app-server.mjs +++ b/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); @@ -338,7 +338,9 @@ 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; + // A broker that expired while idle leaves its saved endpoint behind. + const saved = loadBrokerSession(cwd)?.endpoint ?? null; + brokerEndpoint = saved && (await isBrokerEndpointReady(saved)) ? saved : null; } if (!brokerEndpoint && !options.reuseExistingBroker) { const brokerSession = await ensureBrokerSession(cwd, { env: options.env }); diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index 4bb7d0748..560b2f1b2 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -103,7 +103,7 @@ export function clearBrokerSession(cwd) { } } -async function isBrokerEndpointReady(endpoint) { +export async function isBrokerEndpointReady(endpoint) { if (!endpoint) { return false; } @@ -117,18 +117,21 @@ async function isBrokerEndpointReady(endpoint) { function removeStaleBrokerLock(lockFile) { try { const text = fs.readFileSync(lockFile, "utf8"); + let owner = null; + try { + owner = JSON.parse(text); + } catch { + // Exclusive creation and the owner write are separate steps, so a process that + // dies between them leaves an empty or partial file. Fall through to the age check. + } let stale = false; - if (text) { - const owner = JSON.parse(text); - if (Number.isInteger(owner.pid) && owner.pid > 0) { - try { - process.kill(owner.pid, 0); - } catch (error) { - stale = error.code === "ESRCH"; - } + if (Number.isInteger(owner?.pid) && owner.pid > 0) { + try { + process.kill(owner.pid, 0); + } catch (error) { + stale = error.code === "ESRCH"; } } else { - // A process may die between exclusive creation and writing its PID. stale = Date.now() - fs.statSync(lockFile).mtimeMs > 30000; } if (stale && fs.readFileSync(lockFile, "utf8") === text) { @@ -141,6 +144,17 @@ function removeStaleBrokerLock(lockFile) { } } +function releaseBrokerLock(lockFile, owner) { + // Only remove a lock we still own; another process may have reclaimed it. + try { + if (JSON.parse(fs.readFileSync(lockFile, "utf8")).token === owner.token) { + fs.unlinkSync(lockFile); + } + } catch { + // Already removed or replaced by a reclaiming process. + } +} + export async function ensureBrokerSession(cwd, options = {}) { // Serialize the readiness check, spawn and state publication across processes. // Otherwise simultaneous first commands each leave a detached broker behind. @@ -172,9 +186,7 @@ export async function ensureBrokerSession(cwd, options = {}) { try { return await startBrokerSession(cwd, options); } finally { - if (JSON.parse(fs.readFileSync(lockFile, "utf8")).token === owner.token) { - fs.unlinkSync(lockFile); - } + releaseBrokerLock(lockFile, owner); } } diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 1d285ddd3..2e9db04a3 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -6,6 +6,7 @@ import assert from "node:assert/strict"; import { spawn } from "node:child_process"; import { makeTempDir, writeExecutable } from "./helpers.mjs"; +import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; import { ensureBrokerSession, sendBrokerShutdown } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; import { parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; import { terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; @@ -67,7 +68,10 @@ const fs = require('node:fs'); const readline = require('node:readline'); fs.appendFileSync(process.env.TEST_STARTS, process.pid + '\\n'); const send = (message) => process.stdout.write(JSON.stringify(message) + '\\n'); -readline.createInterface({ input: process.stdin }).on('line', (line) => { +if (process.env.TEST_SLOW_EXIT) process.on('SIGTERM', () => {}); +const rl = readline.createInterface({ input: process.stdin }); +if (process.env.TEST_SLOW_EXIT) rl.on('close', () => setTimeout(() => process.exit(0), 800)); +rl.on('line', (line) => { const m = JSON.parse(line); if (m.id === undefined) return; const reply = (result) => send({ id: m.id, result }); @@ -193,3 +197,44 @@ test("startup recovers a lock left by a process that exited", { timeout: 6000 }, assert.ok(alive(session.pid)); assert.equal(fs.existsSync(path.join(stateDir, "broker.json.lock")), false); }); + +test("startup recovers a lock whose owner write was cut short", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const stateDir = resolveStateDir(f.root); + const lockFile = path.join(stateDir, "broker.json.lock"); + fs.mkdirSync(stateDir, { recursive: true }); + fs.writeFileSync(lockFile, '{"pid":12'); + const old = new Date(Date.now() - 60000); + fs.utimesSync(lockFile, old, old); + const session = await f.start(); + assert.ok(alive(session.pid)); + assert.equal(fs.existsSync(lockFile), false); +}); + +test("a broker draining after idle expiry stops admitting clients", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + f.env.TEST_SLOW_EXIT = "1"; + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("initialize"); + client.socket.destroy(); + // Idle expiry has fired; the fake app-server holds the broker open for ~800ms more. + await new Promise((resolve) => setTimeout(resolve, 400)); + assert.ok(alive(session.pid), "broker is still draining"); + const late = connect(session.endpoint); + const outcome = await new Promise((resolve) => { + late.socket.on("connect", () => resolve("connected")); + late.socket.on("error", () => resolve("refused")); + }); + late.socket.destroy(); + assert.equal(outcome, "refused"); +}); + +test("reuse-existing callers recover when the saved broker expired while idle", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + await waitFor(() => !alive(session.pid), 1500); + const client = await CodexAppServerClient.connect(f.root, { env: f.env, reuseExistingBroker: true }); + t.after(() => client.close()); + assert.equal(client.transport, "direct"); +}); From 0f1a5f96095959988ea19c70460403c651a44267 Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Thu, 5 Nov 2026 16:47:37 -0600 Subject: [PATCH 04/16] test: add the Windows codex.cmd shim to the lifecycle fixture Co-Authored-By: Claude Sonnet 5.5 --- tests/broker-lifecycle.test.mjs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 2e9db04a3..9513dec47 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -84,6 +84,10 @@ rl.on('line', (line) => { } else reply({}); }); `); + // Same Windows wrapper as fake-codex-fixture.mjs: spawn with shell: true resolves codex.cmd. + if (process.platform === "win32") { + fs.writeFileSync(path.join(bin, "codex.cmd"), `@echo off\r\nnode "%~dp0codex" %*\r\n`, "utf8"); + } const env = { ...process.env, PATH: `${bin}${path.delimiter}${process.env.PATH}`, From eb7e0cfc6fee347e546c62bb577d0d03b11c82bf Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Fri, 6 Nov 2026 16:40:59 -0600 Subject: [PATCH 05/16] fix: end idle guard on final answer, serialize stale-lock reclaim, ignore zombies in test teardown Co-Authored-By: Claude Sonnet 5.5 --- plugins/codex/scripts/app-server-broker.mjs | 14 ++++++- .../codex/scripts/lib/broker-lifecycle.mjs | 28 +++++++++++++- tests/broker-lifecycle.test.mjs | 37 ++++++++++++++++++- 3 files changed, 75 insertions(+), 4 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index c7f26fdbe..47e00982f 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -101,9 +101,21 @@ async function main() { function routeNotification(message) { // A disconnected caller must not make its running turn eligible for shutdown. - const turnKey = `${message.params?.threadId}:${message.params?.turn?.id}`; + const turnKey = `${message.params?.threadId}:${message.params?.turn?.id ?? message.params?.turnId}`; if (message.method === "turn/started") activeTurns.add(turnKey); if (message.method === "turn/completed") activeTurns.delete(turnKey); + if ( + message.method === "item/completed" && + message.params?.item?.type === "agentMessage" && + message.params.item.phase === "final_answer" + ) { + // lib/codex.mjs completes a turn from its final answer when turn/completed never + // arrives, so a final answer ends the turn here too. Same grace as the client's. + setTimeout(() => { + activeTurns.delete(turnKey); + updateIdleTimer(); + }, 250).unref(); + } updateIdleTimer(); const target = activeRequestSocket ?? activeStreamSocket; if (!target) { diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index 560b2f1b2..9f66c4af2 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -115,6 +115,26 @@ export async function isBrokerEndpointReady(endpoint) { } function removeStaleBrokerLock(lockFile) { + // Reclaimers serialize on a guard so a descheduled one cannot unlink a lock that + // another reclaimer already replaced. A stale lock's owner is dead (or never finished + // writing), so only reclaimers ever unlink it. + const guard = `${lockFile}.reclaim`; + try { + fs.closeSync(fs.openSync(guard, "wx")); + } catch (error) { + if (error.code !== "EEXIST") { + throw error; + } + try { + // A reclaimer that died holding the guard must not block recovery forever. + if (Date.now() - fs.statSync(guard).mtimeMs > 10000) { + fs.unlinkSync(guard); + } + } catch { + // Released by its holder in the meantime. + } + return; + } try { const text = fs.readFileSync(lockFile, "utf8"); let owner = null; @@ -134,13 +154,19 @@ function removeStaleBrokerLock(lockFile) { } else { stale = Date.now() - fs.statSync(lockFile).mtimeMs > 30000; } - if (stale && fs.readFileSync(lockFile, "utf8") === text) { + if (stale) { fs.unlinkSync(lockFile); } } catch (error) { if (error.code !== "ENOENT") { throw error; } + } finally { + try { + fs.unlinkSync(guard); + } catch { + // Already removed. + } } } diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 9513dec47..26dd0a03e 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -24,7 +24,15 @@ async function waitFor(predicate, timeoutMs = 3000) { } function alive(pid) { - try { process.kill(pid, 0); return true; } catch { return false; } + try { process.kill(pid, 0); } catch { return false; } + // Without a reaping init (some containers) an exited process lingers as a zombie + // that still answers kill(pid, 0). + try { + const stat = fs.readFileSync(`/proc/${pid}/stat`, "utf8"); + return stat.slice(stat.lastIndexOf(")") + 2, stat.lastIndexOf(")") + 3) !== "Z"; + } catch { + return true; + } } function connect(endpoint) { @@ -78,7 +86,8 @@ rl.on('line', (line) => { if (m.method === 'turn/start') { reply({ turn: { id: 'turn_1' } }); send({ method: 'turn/started', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }); - setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }), 700); + if (process.env.TEST_INFERRED) setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }), 100); + else setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }), 700); } else if (m.method === 'test/slow') { setTimeout(() => reply({ done: true }), 700); } else reply({}); @@ -137,6 +146,16 @@ test("disconnecting a client does not expire an active turn", { timeout: 6000 }, await waitFor(() => !alive(session.pid), 1800); }); +test("a turn finished by its final answer alone does not pin the broker", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + f.env.TEST_INFERRED = "1"; + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("turn/start", { threadId: "thr_1" }); + client.socket.destroy(); + await waitFor(() => !alive(session.pid), 1500); +}); + test("disconnecting a client does not expire an in-flight RPC", { timeout: 6000 }, async (t) => { const f = await fixture(t); const session = await f.start(); @@ -215,6 +234,20 @@ test("startup recovers a lock whose owner write was cut short", { timeout: 6000 assert.equal(fs.existsSync(lockFile), false); }); +test("startup recovers when a reclaimer died holding the reclaim guard", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const stateDir = resolveStateDir(f.root); + const lockFile = path.join(stateDir, "broker.json.lock"); + fs.mkdirSync(stateDir, { recursive: true }); + fs.writeFileSync(lockFile, JSON.stringify({ pid: 2 ** 22 + 1, token: "dead-owner" })); + fs.writeFileSync(`${lockFile}.reclaim`, ""); + const old = new Date(Date.now() - 60000); + fs.utimesSync(`${lockFile}.reclaim`, old, old); + const session = await f.start(); + assert.ok(alive(session.pid)); + assert.equal(fs.existsSync(`${lockFile}.reclaim`), false); +}); + test("a broker draining after idle expiry stops admitting clients", { timeout: 6000 }, async (t) => { const f = await fixture(t); f.env.TEST_SLOW_EXIT = "1"; From 7bfff43d43217e3c088b772148c47993eee14e1f Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Fri, 6 Nov 2026 16:57:52 -0600 Subject: [PATCH 06/16] fix: treat zombie lock owners as exited when reclaiming the startup lock Co-Authored-By: Claude Sonnet 5.5 --- .../codex/scripts/lib/broker-lifecycle.mjs | 7 ++--- plugins/codex/scripts/lib/process.mjs | 18 ++++++++++++ tests/broker-lifecycle.test.mjs | 29 +++++++++++-------- 3 files changed, 37 insertions(+), 17 deletions(-) diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index 9f66c4af2..efee280c0 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -7,6 +7,7 @@ import { spawn } from "node:child_process"; import { fileURLToPath } from "node:url"; import { randomUUID } from "node:crypto"; 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"; @@ -146,11 +147,7 @@ function removeStaleBrokerLock(lockFile) { } let stale = false; if (Number.isInteger(owner?.pid) && owner.pid > 0) { - try { - process.kill(owner.pid, 0); - } catch (error) { - stale = error.code === "ESRCH"; - } + stale = !isProcessAlive(owner.pid); } else { stale = Date.now() - fs.statSync(lockFile).mtimeMs > 30000; } diff --git a/plugins/codex/scripts/lib/process.mjs b/plugins/codex/scripts/lib/process.mjs index dd8fc3751..374e4f2c6 100644 --- a/plugins/codex/scripts/lib/process.mjs +++ b/plugins/codex/scripts/lib/process.mjs @@ -1,3 +1,4 @@ +import fs from "node:fs"; import { spawnSync } from "node:child_process"; import process from "node:process"; @@ -54,6 +55,23 @@ function looksLikeMissingProcessMessage(text) { return /not found|no running instance|cannot find|does not exist|no such process/i.test(text); } +export function isProcessAlive(pid) { + try { + process.kill(pid, 0); + } catch (error) { + // EPERM means the process exists but belongs to someone else. + return error.code !== "ESRCH"; + } + // Without a reaping init (some containers) an exited process lingers as a zombie + // that still answers kill(pid, 0). + try { + const stat = fs.readFileSync(`/proc/${pid}/stat`, "utf8"); + return stat[stat.lastIndexOf(")") + 2] !== "Z"; + } catch { + return true; + } +} + export function terminateProcessTree(pid, options = {}) { if (!Number.isFinite(pid)) { return { attempted: false, delivered: false, method: null }; diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 26dd0a03e..6912c8a17 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -9,7 +9,7 @@ import { makeTempDir, writeExecutable } from "./helpers.mjs"; import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; import { ensureBrokerSession, sendBrokerShutdown } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; import { parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; -import { terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; +import { isProcessAlive, terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; const LIFECYCLE_URL = new URL("../plugins/codex/scripts/lib/broker-lifecycle.mjs", import.meta.url).href; @@ -23,17 +23,7 @@ async function waitFor(predicate, timeoutMs = 3000) { assert.fail("Timed out waiting for broker lifecycle transition"); } -function alive(pid) { - try { process.kill(pid, 0); } catch { return false; } - // Without a reaping init (some containers) an exited process lingers as a zombie - // that still answers kill(pid, 0). - try { - const stat = fs.readFileSync(`/proc/${pid}/stat`, "utf8"); - return stat.slice(stat.lastIndexOf(")") + 2, stat.lastIndexOf(")") + 3) !== "Z"; - } catch { - return true; - } -} +const alive = isProcessAlive; function connect(endpoint) { const socket = net.createConnection({ path: parseBrokerEndpoint(endpoint).path }); @@ -248,6 +238,21 @@ test("startup recovers when a reclaimer died holding the reclaim guard", { timeo assert.equal(fs.existsSync(`${lockFile}.reclaim`), false); }); +test("startup recovers a lock owned by an exited zombie process", { timeout: 6000, skip: !fs.existsSync("/proc/self/stat") }, async (t) => { + const f = await fixture(t); + // perl forks a child that exits at once, then sleeps without wait(), leaving a zombie. + const parent = spawn("perl", ["-e", '$| = 1; my $p = fork; exit 0 unless $p; print "$p\\n"; sleep 30'], { stdio: ["ignore", "pipe", "ignore"] }); + t.after(() => parent.kill()); + const pid = await new Promise((resolve) => parent.stdout.once("data", (c) => resolve(Number(String(c).trim())))); + await waitFor(() => fs.readFileSync(`/proc/${pid}/stat`, "utf8").split(") ")[1][0] === "Z"); + assert.equal(isProcessAlive(pid), false, "zombie counts as exited"); + const stateDir = resolveStateDir(f.root); + fs.mkdirSync(stateDir, { recursive: true }); + fs.writeFileSync(path.join(stateDir, "broker.json.lock"), JSON.stringify({ pid, token: "zombie-owner" })); + const session = await f.start(); + assert.ok(alive(session.pid)); +}); + test("a broker draining after idle expiry stops admitting clients", { timeout: 6000 }, async (t) => { const f = await fixture(t); f.env.TEST_SLOW_EXIT = "1"; From 594acbdf7ff17e8569aebc35c1ce855f21a96963 Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Fri, 6 Nov 2026 17:37:37 -0600 Subject: [PATCH 07/16] fix: tolerate concurrent broker cleanup and skip zombie test without perl Co-Authored-By: Claude Sonnet 5.5 --- plugins/codex/scripts/app-server-broker.mjs | 9 ++++---- .../codex/scripts/lib/broker-lifecycle.mjs | 23 ++++++++++++++----- tests/broker-lifecycle.test.mjs | 4 ++-- 3 files changed, 24 insertions(+), 12 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 47e00982f..1e78101a8 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -146,11 +146,12 @@ async function main() { } await appClient.close().catch(() => {}); await closed; - if (listenTarget.kind === "unix" && fs.existsSync(listenTarget.path)) { - fs.unlinkSync(listenTarget.path); + // A replacement broker's teardown may already have removed these. + if (listenTarget.kind === "unix") { + fs.rmSync(listenTarget.path, { force: true }); } - if (pidFile && fs.existsSync(pidFile)) { - fs.unlinkSync(pidFile); + if (pidFile) { + fs.rmSync(pidFile, { force: true }); } } diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index efee280c0..89f3a433b 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -273,6 +273,17 @@ async function startBrokerSession(cwd, options) { return session; } +function removeIfPresent(file) { + // A draining broker removes its own pid file and socket concurrently with a replacement. + try { + fs.unlinkSync(file); + } catch (error) { + if (error.code !== "ENOENT") { + throw error; + } + } +} + export function teardownBrokerSession({ endpoint = null, pidFile, logFile, sessionDir = null, pid = null, killProcess = null }) { if (Number.isFinite(pid) && killProcess) { try { @@ -282,19 +293,19 @@ export function teardownBrokerSession({ endpoint = null, pidFile, logFile, sessi } } - if (pidFile && fs.existsSync(pidFile)) { - fs.unlinkSync(pidFile); + if (pidFile) { + removeIfPresent(pidFile); } - if (logFile && fs.existsSync(logFile)) { - fs.unlinkSync(logFile); + if (logFile) { + removeIfPresent(logFile); } if (endpoint) { try { const target = parseBrokerEndpoint(endpoint); - if (target.kind === "unix" && fs.existsSync(target.path)) { - fs.unlinkSync(target.path); + if (target.kind === "unix") { + removeIfPresent(target.path); } } catch { // Ignore malformed or already-removed broker endpoints during teardown. diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 6912c8a17..0963457d9 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -3,7 +3,7 @@ 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 { spawn, spawnSync } from "node:child_process"; import { makeTempDir, writeExecutable } from "./helpers.mjs"; import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; @@ -238,7 +238,7 @@ test("startup recovers when a reclaimer died holding the reclaim guard", { timeo assert.equal(fs.existsSync(`${lockFile}.reclaim`), false); }); -test("startup recovers a lock owned by an exited zombie process", { timeout: 6000, skip: !fs.existsSync("/proc/self/stat") }, async (t) => { +test("startup recovers a lock owned by an exited zombie process", { timeout: 6000, skip: !fs.existsSync("/proc/self/stat") || spawnSync("perl", ["-e", "1"]).error !== undefined }, async (t) => { const f = await fixture(t); // perl forks a child that exits at once, then sleeps without wait(), leaving a zombie. const parent = spawn("perl", ["-e", '$| = 1; my $p = fork; exit 0 unless $p; print "$p\\n"; sleep 30'], { stdio: ["ignore", "pipe", "ignore"] }); From 36b93a876ecffa0a513fbad2ad6d1a0881c685e7 Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Mon, 9 Nov 2026 08:36:29 -0600 Subject: [PATCH 08/16] test: reproduce concurrent broker cleanup deterministically --- tests/broker-lifecycle.test.mjs | 59 ++++++++++++++++++++++++++++++++- 1 file changed, 58 insertions(+), 1 deletion(-) diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 0963457d9..f87b60737 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -7,7 +7,7 @@ import { spawn, spawnSync } from "node:child_process"; import { makeTempDir, writeExecutable } from "./helpers.mjs"; import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; -import { ensureBrokerSession, sendBrokerShutdown } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { ensureBrokerSession, sendBrokerShutdown, teardownBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; import { parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; import { isProcessAlive, terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; @@ -199,6 +199,63 @@ test("shutdown is bounded when a broker accepts a connection but never replies", await sendBrokerShutdown(endpoint, { timeoutMs: 100 }); }); +for (const file of ["pidFile", "logFile"]) { + test(`replacement teardown tolerates concurrent removal of its ${file}`, (t) => { + const root = makeTempDir(); + t.after(() => fs.rmSync(root, { recursive: true, force: true })); + const target = path.join(root, file); + fs.writeFileSync(target, "old broker"); + const unlink = fs.unlinkSync; + let raced = false; + t.mock.method(fs, "unlinkSync", (candidate) => { + if (candidate === target && !raced) { + raced = true; + // The draining broker removes the file just before replacement teardown does. + unlink(candidate); + } + return unlink(candidate); + }); + + assert.doesNotThrow(() => teardownBrokerSession({ [file]: target })); + assert.equal(raced, true, "exercise removal at the unlink boundary"); + assert.equal(fs.existsSync(target), false); + }); +} + +test("draining broker tolerates replacement teardown removing its pid file", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const hook = path.join(f.root, "cleanup-race.cjs"); + const marker = path.join(f.root, "cleanup-raced"); + // Inject the race inside the real detached broker, without timing sleeps or production hooks. + fs.writeFileSync(hook, ` +const fs = require('node:fs'); +const path = require('node:path'); +if (path.basename(process.argv[1] || '') === 'app-server-broker.mjs') { + const unlink = fs.unlinkSync; + for (const method of ['unlinkSync', 'rmSync']) { + const remove = fs[method]; + fs[method] = function(file, ...args) { + if (path.basename(String(file)) === 'broker.pid' && !fs.existsSync(${JSON.stringify(marker)})) { + unlink(file); + fs.writeFileSync(${JSON.stringify(marker)}, 'removed by replacement'); + } + return remove.call(this, file, ...args); + }; + } +} +`); + f.env.NODE_OPTIONS = `${f.env.NODE_OPTIONS ?? ""} --require ${JSON.stringify(hook)}`; + const session = await f.start(); + const client = connect(session.endpoint); + t.after(() => client.socket.destroy()); + await client.request("broker/shutdown"); + await waitFor(() => !alive(session.pid)); + + assert.equal(fs.readFileSync(marker, "utf8"), "removed by replacement"); + assert.doesNotMatch(fs.readFileSync(session.logFile, "utf8"), /ENOENT/); + assert.equal(fs.existsSync(session.pidFile), false); +}); + test("startup recovers a lock left by a process that exited", { timeout: 6000 }, async (t) => { const f = await fixture(t); const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); From 555ec0c746963f515e8aa17bf243eb3b1391b3de Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Mon, 9 Nov 2026 13:45:53 -0600 Subject: [PATCH 09/16] fix: wait for collaborations before releasing a turn, serialize SessionEnd cleanup, judge reclaim guard by owner liveness Co-Authored-By: Claude Sonnet 5.5 --- plugins/codex/scripts/app-server-broker.mjs | 50 +++++++--- .../codex/scripts/lib/broker-lifecycle.mjs | 91 ++++++++++++------- .../codex/scripts/session-lifecycle-hook.mjs | 25 +++-- tests/broker-lifecycle.test.mjs | 54 ++++++++++- 4 files changed, 161 insertions(+), 59 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 1e78101a8..126aec1a3 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -71,6 +71,8 @@ async function main() { let activeStreamThreadIds = null; const sockets = new Set(); const activeTurns = new Set(); + const pendingCollaborations = new Map(); + const finalAnswered = new Map(); let pendingRequests = 0; let idleTimer = null; let shuttingDown = false; @@ -89,6 +91,16 @@ async function main() { }, idleTimeoutMs); } + function settleFinalAnswer(turnKey, threadId) { + if (pendingCollaborations.get(threadId)?.size) return; + setTimeout(() => { + if (pendingCollaborations.get(threadId)?.size || !finalAnswered.has(turnKey)) return; + finalAnswered.delete(turnKey); + activeTurns.delete(turnKey); + updateIdleTimer(); + }, 250).unref(); + } + function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { activeRequestSocket = null; @@ -103,18 +115,32 @@ async function main() { // A disconnected caller must not make its running turn eligible for shutdown. const turnKey = `${message.params?.threadId}:${message.params?.turn?.id ?? message.params?.turnId}`; if (message.method === "turn/started") activeTurns.add(turnKey); - if (message.method === "turn/completed") activeTurns.delete(turnKey); - if ( - message.method === "item/completed" && - message.params?.item?.type === "agentMessage" && - message.params.item.phase === "final_answer" - ) { - // lib/codex.mjs completes a turn from its final answer when turn/completed never - // arrives, so a final answer ends the turn here too. Same grace as the client's. - setTimeout(() => { - activeTurns.delete(turnKey); - updateIdleTimer(); - }, 250).unref(); + const threadId = message.params?.threadId; + if (message.method === "turn/completed") { + activeTurns.delete(turnKey); + finalAnswered.delete(turnKey); + } + // lib/codex.mjs completes a turn from its final answer when turn/completed never + // arrives, once the thread's collaborations drain. Mirror that so such a turn does not + // pin the broker, and an unfinished collaboration is not cut off by idle expiry. + const item = message.params?.item; + if (item?.type === "collabAgentToolCall") { + const pending = pendingCollaborations.get(threadId) ?? new Set(); + if (message.method === "item/started" || item.status === "inProgress") { + pending.add(item.id); + } else if (message.method === "item/completed") { + pending.delete(item.id); + } + pendingCollaborations.set(threadId, pending); + if (message.method === "item/completed") { + for (const [key, owner] of finalAnswered) { + if (owner === threadId) settleFinalAnswer(key, owner); + } + } + } + if (message.method === "item/completed" && item?.type === "agentMessage" && item.phase === "final_answer") { + finalAnswered.set(turnKey, threadId); + settleFinalAnswer(turnKey, threadId); } updateIdleTimer(); const target = activeRequestSocket ?? activeStreamSocket; diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index 89f3a433b..31fc04d44 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -104,6 +104,14 @@ export function clearBrokerSession(cwd) { } } +export function clearBrokerSessionIfCurrent(cwd, endpoint) { + // A replacement may have published its own state since the caller loaded this one. + const current = loadBrokerSession(cwd); + if (!current || current.endpoint === endpoint) { + clearBrokerSession(cwd); + } +} + export async function isBrokerEndpointReady(endpoint) { if (!endpoint) { return false; @@ -115,44 +123,58 @@ export async function isBrokerEndpointReady(endpoint) { } } -function removeStaleBrokerLock(lockFile) { - // Reclaimers serialize on a guard so a descheduled one cannot unlink a lock that - // another reclaimer already replaced. A stale lock's owner is dead (or never finished - // writing), so only reclaimers ever unlink it. +function readLockOwner(file) { + const text = fs.readFileSync(file, "utf8"); + try { + return { text, owner: JSON.parse(text) }; + } catch { + // Exclusive creation and the owner write are separate steps, so a process that + // dies between them leaves an empty or partial file. + return { text, owner: null }; + } +} + +function isLockStale(file, owner, maxAgeMs) { + if (Number.isInteger(owner?.pid) && owner.pid > 0) { + return !isProcessAlive(owner.pid); + } + return Date.now() - fs.statSync(file).mtimeMs > maxAgeMs; +} + +function unlinkIfUnchanged(file, text) { + // Tokens are unique per acquisition, so unchanged text means the same acquisition. + if (fs.readFileSync(file, "utf8") === text) { + fs.unlinkSync(file); + } +} + +function removeStaleBrokerLock(lockFile, guardOwner) { + // Reclaimers serialize on a guard so two of them cannot both act on the same stale lock. + // The guard is judged by its owner's liveness like the lock is, never by age alone, + // so a slow but live reclaimer is not revoked. const guard = `${lockFile}.reclaim`; try { - fs.closeSync(fs.openSync(guard, "wx")); + fs.writeFileSync(guard, JSON.stringify(guardOwner), { flag: "wx" }); } catch (error) { if (error.code !== "EEXIST") { throw error; } try { - // A reclaimer that died holding the guard must not block recovery forever. - if (Date.now() - fs.statSync(guard).mtimeMs > 10000) { - fs.unlinkSync(guard); + const held = readLockOwner(guard); + if (isLockStale(guard, held.owner, 10000)) { + unlinkIfUnchanged(guard, held.text); + } + } catch (guardError) { + if (guardError.code !== "ENOENT") { + throw guardError; } - } catch { - // Released by its holder in the meantime. } return; } try { - const text = fs.readFileSync(lockFile, "utf8"); - let owner = null; - try { - owner = JSON.parse(text); - } catch { - // Exclusive creation and the owner write are separate steps, so a process that - // dies between them leaves an empty or partial file. Fall through to the age check. - } - let stale = false; - if (Number.isInteger(owner?.pid) && owner.pid > 0) { - stale = !isProcessAlive(owner.pid); - } else { - stale = Date.now() - fs.statSync(lockFile).mtimeMs > 30000; - } - if (stale) { - fs.unlinkSync(lockFile); + const held = readLockOwner(lockFile); + if (isLockStale(lockFile, held.owner, 30000)) { + unlinkIfUnchanged(lockFile, held.text); } } catch (error) { if (error.code !== "ENOENT") { @@ -160,7 +182,7 @@ function removeStaleBrokerLock(lockFile) { } } finally { try { - fs.unlinkSync(guard); + unlinkIfUnchanged(guard, JSON.stringify(guardOwner)); } catch { // Already removed. } @@ -178,9 +200,10 @@ function releaseBrokerLock(lockFile, owner) { } } -export async function ensureBrokerSession(cwd, options = {}) { - // Serialize the readiness check, spawn and state publication across processes. - // Otherwise simultaneous first commands each leave a detached broker behind. +export async function withBrokerLock(cwd, options, fn) { + // Serialize broker startup, state publication and teardown across processes. + // Otherwise simultaneous first commands each leave a detached broker behind, and a + // session-end cleanup can erase a replacement's state. const stateFile = resolveBrokerStateFile(cwd); fs.mkdirSync(path.dirname(stateFile), { recursive: true }); const lockFile = `${stateFile}.lock`; @@ -199,7 +222,7 @@ export async function ensureBrokerSession(cwd, options = {}) { if (error.code !== "EEXIST") { throw error; } - removeStaleBrokerLock(lockFile); + removeStaleBrokerLock(lockFile, { pid: process.pid, token: randomUUID() }); if (Date.now() >= deadline) { throw new Error("Timed out waiting for Codex broker startup lock."); } @@ -207,12 +230,16 @@ export async function ensureBrokerSession(cwd, options = {}) { } } try { - return await startBrokerSession(cwd, options); + return await fn(); } finally { releaseBrokerLock(lockFile, owner); } } +export function ensureBrokerSession(cwd, options = {}) { + return withBrokerLock(cwd, options, () => startBrokerSession(cwd, options)); +} + async function startBrokerSession(cwd, options) { const existing = loadBrokerSession(cwd); if (existing && (await isBrokerEndpointReady(existing.endpoint))) { diff --git a/plugins/codex/scripts/session-lifecycle-hook.mjs b/plugins/codex/scripts/session-lifecycle-hook.mjs index 778571e6c..94049b9ea 100644 --- a/plugins/codex/scripts/session-lifecycle-hook.mjs +++ b/plugins/codex/scripts/session-lifecycle-hook.mjs @@ -6,12 +6,13 @@ import process from "node:process"; import { terminateProcessTree } from "./lib/process.mjs"; import { BROKER_ENDPOINT_ENV } from "./lib/app-server.mjs"; import { - clearBrokerSession, + clearBrokerSessionIfCurrent, LOG_FILE_ENV, loadBrokerSession, PID_FILE_ENV, sendBrokerShutdown, - teardownBrokerSession + teardownBrokerSession, + withBrokerLock } from "./lib/broker-lifecycle.mjs"; import { loadState, resolveStateFile, saveState } from "./lib/state.mjs"; import { TRANSCRIPT_PATH_ENV } from "./lib/claude-session-transfer.mjs"; @@ -102,15 +103,19 @@ async function handleSessionEnd(input) { } cleanupSessionJobs(cwd, input.session_id || process.env[SESSION_ID_ENV]); - teardownBrokerSession({ - endpoint: brokerEndpoint, - pidFile, - logFile, - sessionDir, - pid, - killProcess: terminateProcessTree + // Closing the listener lets a replacement start while this broker drains. Teardown + // runs under the startup lock and clears saved state only if it is still this broker's. + await withBrokerLock(cwd, {}, () => { + teardownBrokerSession({ + endpoint: brokerEndpoint, + pidFile, + logFile, + sessionDir, + pid, + killProcess: terminateProcessTree + }); + clearBrokerSessionIfCurrent(cwd, brokerEndpoint); }); - clearBrokerSession(cwd); } async function main() { diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index f87b60737..88a03f5b0 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -7,7 +7,7 @@ import { spawn, spawnSync } from "node:child_process"; import { makeTempDir, writeExecutable } from "./helpers.mjs"; import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; -import { ensureBrokerSession, sendBrokerShutdown, teardownBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { clearBrokerSessionIfCurrent, ensureBrokerSession, loadBrokerSession, saveBrokerSession, sendBrokerShutdown, teardownBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; import { parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; import { isProcessAlive, terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; @@ -76,7 +76,11 @@ rl.on('line', (line) => { if (m.method === 'turn/start') { reply({ turn: { id: 'turn_1' } }); send({ method: 'turn/started', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }); - if (process.env.TEST_INFERRED) setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }), 100); + if (process.env.TEST_COLLAB) { + send({ method: 'item/started', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status: 'inProgress' } } }); + setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }), 100); + setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status: 'completed' } } }), 900); + } else if (process.env.TEST_INFERRED) setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }), 100); else setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }), 700); } else if (m.method === 'test/slow') { setTimeout(() => reply({ done: true }), 700); @@ -146,6 +150,21 @@ test("a turn finished by its final answer alone does not pin the broker", { time await waitFor(() => !alive(session.pid), 1500); }); +test("a final answer does not release the broker while a collaboration is still running", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + f.env.TEST_COLLAB = "1"; + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("turn/start", { threadId: "thr_1" }); + const childPid = Number(fs.readFileSync(f.env.TEST_STARTS, "utf8").trim()); + client.socket.destroy(); + // Final answer at ~100ms, collaboration completes at ~900ms; idle timeout is 200ms. + await new Promise((resolve) => setTimeout(resolve, 700)); + assert.ok(alive(childPid), "app-server stays up while the collaboration runs"); + assert.ok(alive(session.pid), "broker stays up while the collaboration runs"); + await waitFor(() => !alive(session.pid), 2500); +}); + test("disconnecting a client does not expire an in-flight RPC", { timeout: 6000 }, async (t) => { const f = await fixture(t); const session = await f.start(); @@ -287,14 +306,39 @@ test("startup recovers when a reclaimer died holding the reclaim guard", { timeo const lockFile = path.join(stateDir, "broker.json.lock"); fs.mkdirSync(stateDir, { recursive: true }); fs.writeFileSync(lockFile, JSON.stringify({ pid: 2 ** 22 + 1, token: "dead-owner" })); - fs.writeFileSync(`${lockFile}.reclaim`, ""); - const old = new Date(Date.now() - 60000); - fs.utimesSync(`${lockFile}.reclaim`, old, old); + fs.writeFileSync(`${lockFile}.reclaim`, JSON.stringify({ pid: 2 ** 22 + 2, token: "dead-reclaimer" })); const session = await f.start(); assert.ok(alive(session.pid)); assert.equal(fs.existsSync(`${lockFile}.reclaim`), false); }); +test("a live reclaimer is not revoked however long it holds the guard", { timeout: 8000 }, async (t) => { + const f = await fixture(t); + const stateDir = resolveStateDir(f.root); + const lockFile = path.join(stateDir, "broker.json.lock"); + fs.mkdirSync(stateDir, { recursive: true }); + fs.writeFileSync(lockFile, JSON.stringify({ pid: 2 ** 22 + 1, token: "dead-owner" })); + const guard = JSON.stringify({ pid: process.pid, token: "slow-reclaimer" }); + fs.writeFileSync(`${lockFile}.reclaim`, guard); + const old = new Date(Date.now() - 60000); + fs.utimesSync(`${lockFile}.reclaim`, old, old); + await assert.rejects(ensureBrokerSession(f.root, { env: f.env, timeoutMs: 50 }), /Timed out waiting/); + assert.equal(fs.readFileSync(`${lockFile}.reclaim`, "utf8"), guard, "guard untouched"); + assert.equal(fs.existsSync(lockFile), true, "stale lock left for the live reclaimer"); +}); + +test("session-end cleanup keeps a replacement broker's saved state", () => { + const root = makeTempDir(); + const replacement = { endpoint: "unix:/tmp/replacement.sock", pidFile: null, logFile: null, sessionDir: null, pid: 1 }; + saveBrokerSession(root, replacement); + clearBrokerSessionIfCurrent(root, "unix:/tmp/old.sock"); + assert.deepEqual(loadBrokerSession(root), replacement); + clearBrokerSessionIfCurrent(root, replacement.endpoint); + assert.equal(loadBrokerSession(root), null); + fs.rmSync(resolveStateDir(root), { recursive: true, force: true }); + fs.rmSync(root, { recursive: true, force: true }); +}); + test("startup recovers a lock owned by an exited zombie process", { timeout: 6000, skip: !fs.existsSync("/proc/self/stat") || spawnSync("perl", ["-e", "1"]).error !== undefined }, async (t) => { const f = await fixture(t); // perl forks a child that exits at once, then sleeps without wait(), leaving a zombie. From beecc5ce4c1fba4294a707d0d6d82528d2b0e917 Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Mon, 9 Nov 2026 14:03:14 -0600 Subject: [PATCH 10/16] fix: let SessionEnd fall back when the startup lock is held, drop empty collaboration sets Co-Authored-By: Claude Sonnet 5.5 --- plugins/codex/scripts/app-server-broker.mjs | 6 +++++- .../codex/scripts/session-lifecycle-hook.mjs | 14 ++++++++++++-- tests/broker-lifecycle.test.mjs | 17 +++++++++++++++++ 3 files changed, 34 insertions(+), 3 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 126aec1a3..f91b84dd5 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -131,7 +131,11 @@ async function main() { } else if (message.method === "item/completed") { pending.delete(item.id); } - pendingCollaborations.set(threadId, pending); + if (pending.size) { + pendingCollaborations.set(threadId, pending); + } else { + pendingCollaborations.delete(threadId); + } if (message.method === "item/completed") { for (const [key, owner] of finalAnswered) { if (owner === threadId) settleFinalAnswer(key, owner); diff --git a/plugins/codex/scripts/session-lifecycle-hook.mjs b/plugins/codex/scripts/session-lifecycle-hook.mjs index 94049b9ea..823562270 100644 --- a/plugins/codex/scripts/session-lifecycle-hook.mjs +++ b/plugins/codex/scripts/session-lifecycle-hook.mjs @@ -105,7 +105,7 @@ async function handleSessionEnd(input) { cleanupSessionJobs(cwd, input.session_id || process.env[SESSION_ID_ENV]); // Closing the listener lets a replacement start while this broker drains. Teardown // runs under the startup lock and clears saved state only if it is still this broker's. - await withBrokerLock(cwd, {}, () => { + const cleanup = () => { teardownBrokerSession({ endpoint: brokerEndpoint, pidFile, @@ -115,7 +115,17 @@ async function handleSessionEnd(input) { killProcess: terminateProcessTree }); clearBrokerSessionIfCurrent(cwd, brokerEndpoint); - }); + }; + try { + await withBrokerLock(cwd, {}, cleanup); + } catch (error) { + // A starter that holds the lock too long must not leave this session's broker behind. + // The conditional clear still protects a replacement's state. + if (!/Timed out waiting for Codex broker startup lock/.test(error.message)) { + throw error; + } + cleanup(); + } } async function main() { diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 88a03f5b0..72aca3af4 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -327,6 +327,23 @@ test("a live reclaimer is not revoked however long it holds the guard", { timeou assert.equal(fs.existsSync(lockFile), true, "stale lock left for the live reclaimer"); }); +test("session-end hook still tears down its broker when the startup lock is held", { timeout: 15000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + const lockFile = path.join(resolveStateDir(f.root), "broker.json.lock"); + fs.writeFileSync(lockFile, JSON.stringify({ pid: process.pid, token: "busy-starter" })); + t.after(() => fs.rmSync(lockFile, { force: true })); + const hook = new URL("../plugins/codex/scripts/session-lifecycle-hook.mjs", import.meta.url).pathname; + const child = spawn(process.execPath, [hook, "SessionEnd"], { cwd: f.root, env: f.env, stdio: ["pipe", "ignore", "pipe"] }); + child.stdin.end(JSON.stringify({ hook_event_name: "SessionEnd", cwd: f.root, session_id: "s1" })); + let stderr = ""; + child.stderr.on("data", (c) => { stderr += c; }); + const code = await new Promise((resolve) => child.on("exit", resolve)); + assert.equal(code, 0, stderr); + assert.equal(alive(session.pid), false); + assert.equal(loadBrokerSession(f.root), null); +}); + test("session-end cleanup keeps a replacement broker's saved state", () => { const root = makeTempDir(); const replacement = { endpoint: "unix:/tmp/replacement.sock", pidFile: null, logFile: null, sessionDir: null, pid: 1 }; From 3bd48081da12cdf4fc38993295d0ae0604ed4f1b Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Mon, 9 Nov 2026 14:15:07 -0600 Subject: [PATCH 11/16] fix: clear a thread's pending collaborations when its turn starts or ends Co-Authored-By: Claude Sonnet 5.5 --- plugins/codex/scripts/app-server-broker.mjs | 8 +++++++- tests/broker-lifecycle.test.mjs | 18 +++++++++++++++++- 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index f91b84dd5..4d45e81b2 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -114,11 +114,17 @@ async function main() { function routeNotification(message) { // A disconnected caller must not make its running turn eligible for shutdown. const turnKey = `${message.params?.threadId}:${message.params?.turn?.id ?? message.params?.turnId}`; - if (message.method === "turn/started") activeTurns.add(turnKey); const threadId = message.params?.threadId; + if (message.method === "turn/started") { + activeTurns.add(turnKey); + pendingCollaborations.delete(threadId); + } if (message.method === "turn/completed") { activeTurns.delete(turnKey); finalAnswered.delete(turnKey); + // Collaborations belong to the turn that started them; an interrupted turn may never + // have seen them complete, and a resumed thread must not inherit them. + pendingCollaborations.delete(threadId); } // lib/codex.mjs completes a turn from its final answer when turn/completed never // arrives, once the thread's collaborations drain. Mirror that so such a turn does not diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 72aca3af4..4a2c36a18 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -76,7 +76,12 @@ rl.on('line', (line) => { if (m.method === 'turn/start') { reply({ turn: { id: 'turn_1' } }); send({ method: 'turn/started', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }); - if (process.env.TEST_COLLAB) { + if (m.params && m.params.mode === 'interrupted') { + send({ method: 'item/started', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status: 'inProgress' } } }); + setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1', status: 'interrupted' } } }), 100); + } else if (m.params && m.params.mode === 'final-only') { + setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm2', phase: 'final_answer', text: 'done' } } }), 100); + } else if (process.env.TEST_COLLAB) { send({ method: 'item/started', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status: 'inProgress' } } }); setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }), 100); setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status: 'completed' } } }), 900); @@ -165,6 +170,17 @@ test("a final answer does not release the broker while a collaboration is still await waitFor(() => !alive(session.pid), 2500); }); +test("a resumed thread does not inherit an interrupted turn's collaboration", { timeout: 8000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("turn/start", { threadId: "thr_1", mode: "interrupted" }); + await new Promise((resolve) => setTimeout(resolve, 300)); + await client.request("turn/start", { threadId: "thr_1", mode: "final-only" }); + client.socket.destroy(); + await waitFor(() => !alive(session.pid), 2500); +}); + test("disconnecting a client does not expire an in-flight RPC", { timeout: 6000 }, async (t) => { const f = await fixture(t); const session = await f.start(); From 41900ff898648012abfa5a4afac2f0c914942819 Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Mon, 9 Nov 2026 15:44:08 -0600 Subject: [PATCH 12/16] test: build the hook path with fileURLToPath URL.pathname keeps %20 escapes and a leading slash before Windows drive letters, so the session-end e2e test could not spawn the hook from a checkout path containing spaces. Co-Authored-By: Claude Sonnet 5.5 --- tests/broker-lifecycle.test.mjs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 4a2c36a18..02156473d 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -1,6 +1,7 @@ import fs from "node:fs"; import net from "node:net"; import path from "node:path"; +import { fileURLToPath } from "node:url"; import test from "node:test"; import assert from "node:assert/strict"; import { spawn, spawnSync } from "node:child_process"; @@ -349,7 +350,7 @@ test("session-end hook still tears down its broker when the startup lock is held const lockFile = path.join(resolveStateDir(f.root), "broker.json.lock"); fs.writeFileSync(lockFile, JSON.stringify({ pid: process.pid, token: "busy-starter" })); t.after(() => fs.rmSync(lockFile, { force: true })); - const hook = new URL("../plugins/codex/scripts/session-lifecycle-hook.mjs", import.meta.url).pathname; + const hook = fileURLToPath(new URL("../plugins/codex/scripts/session-lifecycle-hook.mjs", import.meta.url)); const child = spawn(process.execPath, [hook, "SessionEnd"], { cwd: f.root, env: f.env, stdio: ["pipe", "ignore", "pipe"] }); child.stdin.end(JSON.stringify({ hook_event_name: "SessionEnd", cwd: f.root, session_id: "s1" })); let stderr = ""; From 637e47b5f4daf0d3ea0b06e74aa7b5c342400092 Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Mon, 9 Nov 2026 16:02:54 -0600 Subject: [PATCH 13/16] test: wait for the app-server to see the in-flight RPC before disconnecting An empty write callback plus a 50ms sleep did not prove the broker had registered the request, so a slow runner could disconnect first and test idle expiry instead of pending-request retention. Co-Authored-By: Claude Sonnet 5.5 --- tests/broker-lifecycle.test.mjs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 02156473d..6833d1613 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -89,6 +89,7 @@ rl.on('line', (line) => { } else if (process.env.TEST_INFERRED) setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }), 100); else setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }), 700); } else if (m.method === 'test/slow') { + fs.writeFileSync(process.env.TEST_SLOW_SEEN, '1'); setTimeout(() => reply({ done: true }), 700); } else reply({}); }); @@ -101,6 +102,7 @@ rl.on('line', (line) => { ...process.env, PATH: `${bin}${path.delimiter}${process.env.PATH}`, TEST_STARTS: path.join(root, "starts"), + TEST_SLOW_SEEN: path.join(root, "slow-seen"), CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS: "200" }; const sessions = []; @@ -187,8 +189,8 @@ test("disconnecting a client does not expire an in-flight RPC", { timeout: 6000 const session = await f.start(); const client = connect(session.endpoint); client.request("test/slow"); - await new Promise((resolve) => client.socket.write("", resolve)); - await new Promise((resolve) => setTimeout(resolve, 50)); + // The app-server seeing the request proves the broker has registered it as pending. + await waitFor(() => fs.existsSync(f.env.TEST_SLOW_SEEN), 2000); client.socket.destroy(); await new Promise((resolve) => setTimeout(resolve, 400)); assert.ok(alive(session.pid), "pending request outlives idle timeout"); From 5301a69f67516a38b9250fa672cb4d074198be97 Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Mon, 9 Nov 2026 17:47:31 -0600 Subject: [PATCH 14/16] refactor: trim to the idle-shutdown fix Keep the idle timer with its active-turn and pending-RPC guards and the saved-endpoint readiness check. Drop the startup lock, reclaim guard, zombie detection, SessionEnd changes and collaboration tracking: with idle expiry a duplicate broker from a startup race exits on its own. The full version lives on pecord/companion-broker-hardening. Co-Authored-By: Claude Sonnet 5.5 --- plugins/codex/scripts/app-server-broker.mjs | 96 ++---- .../codex/scripts/lib/broker-lifecycle.mjs | 169 +---------- plugins/codex/scripts/lib/process.mjs | 18 -- .../codex/scripts/session-lifecycle-hook.mjs | 37 +-- tests/broker-lifecycle.test.mjs | 275 +----------------- 5 files changed, 58 insertions(+), 537 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 4d45e81b2..e8b9aed0e 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -71,8 +71,6 @@ async function main() { let activeStreamThreadIds = null; const sockets = new Set(); const activeTurns = new Set(); - const pendingCollaborations = new Map(); - const finalAnswered = new Map(); let pendingRequests = 0; let idleTimer = null; let shuttingDown = false; @@ -91,16 +89,6 @@ async function main() { }, idleTimeoutMs); } - function settleFinalAnswer(turnKey, threadId) { - if (pendingCollaborations.get(threadId)?.size) return; - setTimeout(() => { - if (pendingCollaborations.get(threadId)?.size || !finalAnswered.has(turnKey)) return; - finalAnswered.delete(turnKey); - activeTurns.delete(turnKey); - updateIdleTimer(); - }, 250).unref(); - } - function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { activeRequestSocket = null; @@ -113,45 +101,9 @@ async function main() { function routeNotification(message) { // A disconnected caller must not make its running turn eligible for shutdown. - const turnKey = `${message.params?.threadId}:${message.params?.turn?.id ?? message.params?.turnId}`; - const threadId = message.params?.threadId; - if (message.method === "turn/started") { - activeTurns.add(turnKey); - pendingCollaborations.delete(threadId); - } - if (message.method === "turn/completed") { - activeTurns.delete(turnKey); - finalAnswered.delete(turnKey); - // Collaborations belong to the turn that started them; an interrupted turn may never - // have seen them complete, and a resumed thread must not inherit them. - pendingCollaborations.delete(threadId); - } - // lib/codex.mjs completes a turn from its final answer when turn/completed never - // arrives, once the thread's collaborations drain. Mirror that so such a turn does not - // pin the broker, and an unfinished collaboration is not cut off by idle expiry. - const item = message.params?.item; - if (item?.type === "collabAgentToolCall") { - const pending = pendingCollaborations.get(threadId) ?? new Set(); - if (message.method === "item/started" || item.status === "inProgress") { - pending.add(item.id); - } else if (message.method === "item/completed") { - pending.delete(item.id); - } - if (pending.size) { - pendingCollaborations.set(threadId, pending); - } else { - pendingCollaborations.delete(threadId); - } - if (message.method === "item/completed") { - for (const [key, owner] of finalAnswered) { - if (owner === threadId) settleFinalAnswer(key, owner); - } - } - } - if (message.method === "item/completed" && item?.type === "agentMessage" && item.phase === "final_answer") { - finalAnswered.set(turnKey, threadId); - settleFinalAnswer(turnKey, threadId); - } + const turnKey = `${message.params?.threadId}:${message.params?.turn?.id}`; + if (message.method === "turn/started") activeTurns.add(turnKey); + if (message.method === "turn/completed") activeTurns.delete(turnKey); updateIdleTimer(); const target = activeRequestSocket ?? activeStreamSocket; if (!target) { @@ -174,20 +126,16 @@ async function main() { if (shuttingDown) return; shuttingDown = true; clearTimeout(idleTimer); - // Stop admitting clients before the app-server close, which can take seconds. - // A client admitted while draining would initialize fine, then fail its first request. - const closed = new Promise((resolve) => server.close(resolve)); for (const socket of sockets) { socket.end(); } await appClient.close().catch(() => {}); - await closed; - // A replacement broker's teardown may already have removed these. - if (listenTarget.kind === "unix") { - fs.rmSync(listenTarget.path, { force: true }); + await new Promise((resolve) => server.close(resolve)); + if (listenTarget.kind === "unix" && fs.existsSync(listenTarget.path)) { + fs.unlinkSync(listenTarget.path); } - if (pidFile) { - fs.rmSync(pidFile, { force: true }); + if (pidFile && fs.existsSync(pidFile)) { + fs.unlinkSync(pidFile); } } @@ -260,19 +208,30 @@ async function main() { continue; } - const isStreaming = STREAMING_METHODS.has(message.method); - if (!allowInterruptDuringActiveStream) { - activeRequestSocket = socket; + if (allowInterruptDuringActiveStream) { + pendingRequests += 1; + try { + const result = await appClient.request(message.method, message.params ?? {}); + send(socket, { id: message.id, result }); + } catch (error) { + send(socket, { + id: message.id, + error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) + }); + } finally { + pendingRequests -= 1; + updateIdleTimer(); + } + continue; } + + const isStreaming = STREAMING_METHODS.has(message.method); + activeRequestSocket = socket; pendingRequests += 1; try { const result = await appClient.request(message.method, message.params ?? {}); send(socket, { id: message.id, result }); - // Interrupts share request bookkeeping without taking stream ownership. - if (allowInterruptDuringActiveStream) { - continue; - } if (isStreaming) { activeStreamSocket = socket; activeStreamThreadIds = buildStreamThreadIds(message.method, message.params ?? {}, result); @@ -285,9 +244,6 @@ async function main() { id: message.id, error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) }); - if (allowInterruptDuringActiveStream) { - continue; - } if (activeRequestSocket === socket) { activeRequestSocket = null; } diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index 31fc04d44..ae49becee 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -5,9 +5,7 @@ import path from "node:path"; import process from "node:process"; import { spawn } from "node:child_process"; import { fileURLToPath } from "node:url"; -import { randomUUID } from "node:crypto"; 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"; @@ -42,22 +40,19 @@ export async function waitForBrokerEndpoint(endpoint, timeoutMs = 2000) { return false; } -export async function sendBrokerShutdown(endpoint, { timeoutMs = 1000 } = {}) { +export async function sendBrokerShutdown(endpoint) { await new Promise((resolve) => { const socket = connectToEndpoint(endpoint); - const finish = () => { - clearTimeout(timer); - socket.destroy(); - resolve(); - }; - const timer = setTimeout(finish, timeoutMs); socket.setEncoding("utf8"); socket.on("connect", () => { socket.write(`${JSON.stringify({ id: 1, method: "broker/shutdown", params: {} })}\n`); }); - socket.on("data", finish); - socket.on("error", finish); - socket.on("close", finish); + socket.on("data", () => { + socket.end(); + resolve(); + }); + socket.on("error", resolve); + socket.on("close", resolve); }); } @@ -104,14 +99,6 @@ export function clearBrokerSession(cwd) { } } -export function clearBrokerSessionIfCurrent(cwd, endpoint) { - // A replacement may have published its own state since the caller loaded this one. - const current = loadBrokerSession(cwd); - if (!current || current.endpoint === endpoint) { - clearBrokerSession(cwd); - } -} - export async function isBrokerEndpointReady(endpoint) { if (!endpoint) { return false; @@ -123,124 +110,7 @@ export async function isBrokerEndpointReady(endpoint) { } } -function readLockOwner(file) { - const text = fs.readFileSync(file, "utf8"); - try { - return { text, owner: JSON.parse(text) }; - } catch { - // Exclusive creation and the owner write are separate steps, so a process that - // dies between them leaves an empty or partial file. - return { text, owner: null }; - } -} - -function isLockStale(file, owner, maxAgeMs) { - if (Number.isInteger(owner?.pid) && owner.pid > 0) { - return !isProcessAlive(owner.pid); - } - return Date.now() - fs.statSync(file).mtimeMs > maxAgeMs; -} - -function unlinkIfUnchanged(file, text) { - // Tokens are unique per acquisition, so unchanged text means the same acquisition. - if (fs.readFileSync(file, "utf8") === text) { - fs.unlinkSync(file); - } -} - -function removeStaleBrokerLock(lockFile, guardOwner) { - // Reclaimers serialize on a guard so two of them cannot both act on the same stale lock. - // The guard is judged by its owner's liveness like the lock is, never by age alone, - // so a slow but live reclaimer is not revoked. - const guard = `${lockFile}.reclaim`; - try { - fs.writeFileSync(guard, JSON.stringify(guardOwner), { flag: "wx" }); - } catch (error) { - if (error.code !== "EEXIST") { - throw error; - } - try { - const held = readLockOwner(guard); - if (isLockStale(guard, held.owner, 10000)) { - unlinkIfUnchanged(guard, held.text); - } - } catch (guardError) { - if (guardError.code !== "ENOENT") { - throw guardError; - } - } - return; - } - try { - const held = readLockOwner(lockFile); - if (isLockStale(lockFile, held.owner, 30000)) { - unlinkIfUnchanged(lockFile, held.text); - } - } catch (error) { - if (error.code !== "ENOENT") { - throw error; - } - } finally { - try { - unlinkIfUnchanged(guard, JSON.stringify(guardOwner)); - } catch { - // Already removed. - } - } -} - -function releaseBrokerLock(lockFile, owner) { - // Only remove a lock we still own; another process may have reclaimed it. - try { - if (JSON.parse(fs.readFileSync(lockFile, "utf8")).token === owner.token) { - fs.unlinkSync(lockFile); - } - } catch { - // Already removed or replaced by a reclaiming process. - } -} - -export async function withBrokerLock(cwd, options, fn) { - // Serialize broker startup, state publication and teardown across processes. - // Otherwise simultaneous first commands each leave a detached broker behind, and a - // session-end cleanup can erase a replacement's state. - const stateFile = resolveBrokerStateFile(cwd); - fs.mkdirSync(path.dirname(stateFile), { recursive: true }); - const lockFile = `${stateFile}.lock`; - const owner = { pid: process.pid, token: randomUUID() }; - const deadline = Date.now() + (options.timeoutMs ?? 2000) + 3000; - while (true) { - try { - const fd = fs.openSync(lockFile, "wx"); - try { - fs.writeFileSync(fd, JSON.stringify(owner)); - } finally { - fs.closeSync(fd); - } - break; - } catch (error) { - if (error.code !== "EEXIST") { - throw error; - } - removeStaleBrokerLock(lockFile, { pid: process.pid, token: randomUUID() }); - if (Date.now() >= deadline) { - throw new Error("Timed out waiting for Codex broker startup lock."); - } - await new Promise((resolve) => setTimeout(resolve, 50)); - } - } - try { - return await fn(); - } finally { - releaseBrokerLock(lockFile, owner); - } -} - -export function ensureBrokerSession(cwd, options = {}) { - return withBrokerLock(cwd, options, () => startBrokerSession(cwd, options)); -} - -async function startBrokerSession(cwd, options) { +export async function ensureBrokerSession(cwd, options = {}) { const existing = loadBrokerSession(cwd); if (existing && (await isBrokerEndpointReady(existing.endpoint))) { return existing; @@ -300,17 +170,6 @@ async function startBrokerSession(cwd, options) { return session; } -function removeIfPresent(file) { - // A draining broker removes its own pid file and socket concurrently with a replacement. - try { - fs.unlinkSync(file); - } catch (error) { - if (error.code !== "ENOENT") { - throw error; - } - } -} - export function teardownBrokerSession({ endpoint = null, pidFile, logFile, sessionDir = null, pid = null, killProcess = null }) { if (Number.isFinite(pid) && killProcess) { try { @@ -320,19 +179,19 @@ export function teardownBrokerSession({ endpoint = null, pidFile, logFile, sessi } } - if (pidFile) { - removeIfPresent(pidFile); + if (pidFile && fs.existsSync(pidFile)) { + fs.unlinkSync(pidFile); } - if (logFile) { - removeIfPresent(logFile); + if (logFile && fs.existsSync(logFile)) { + fs.unlinkSync(logFile); } if (endpoint) { try { const target = parseBrokerEndpoint(endpoint); - if (target.kind === "unix") { - removeIfPresent(target.path); + if (target.kind === "unix" && fs.existsSync(target.path)) { + fs.unlinkSync(target.path); } } catch { // Ignore malformed or already-removed broker endpoints during teardown. diff --git a/plugins/codex/scripts/lib/process.mjs b/plugins/codex/scripts/lib/process.mjs index 374e4f2c6..dd8fc3751 100644 --- a/plugins/codex/scripts/lib/process.mjs +++ b/plugins/codex/scripts/lib/process.mjs @@ -1,4 +1,3 @@ -import fs from "node:fs"; import { spawnSync } from "node:child_process"; import process from "node:process"; @@ -55,23 +54,6 @@ function looksLikeMissingProcessMessage(text) { return /not found|no running instance|cannot find|does not exist|no such process/i.test(text); } -export function isProcessAlive(pid) { - try { - process.kill(pid, 0); - } catch (error) { - // EPERM means the process exists but belongs to someone else. - return error.code !== "ESRCH"; - } - // Without a reaping init (some containers) an exited process lingers as a zombie - // that still answers kill(pid, 0). - try { - const stat = fs.readFileSync(`/proc/${pid}/stat`, "utf8"); - return stat[stat.lastIndexOf(")") + 2] !== "Z"; - } catch { - return true; - } -} - export function terminateProcessTree(pid, options = {}) { if (!Number.isFinite(pid)) { return { attempted: false, delivered: false, method: null }; diff --git a/plugins/codex/scripts/session-lifecycle-hook.mjs b/plugins/codex/scripts/session-lifecycle-hook.mjs index 823562270..778571e6c 100644 --- a/plugins/codex/scripts/session-lifecycle-hook.mjs +++ b/plugins/codex/scripts/session-lifecycle-hook.mjs @@ -6,13 +6,12 @@ import process from "node:process"; import { terminateProcessTree } from "./lib/process.mjs"; import { BROKER_ENDPOINT_ENV } from "./lib/app-server.mjs"; import { - clearBrokerSessionIfCurrent, + clearBrokerSession, LOG_FILE_ENV, loadBrokerSession, PID_FILE_ENV, sendBrokerShutdown, - teardownBrokerSession, - withBrokerLock + teardownBrokerSession } from "./lib/broker-lifecycle.mjs"; import { loadState, resolveStateFile, saveState } from "./lib/state.mjs"; import { TRANSCRIPT_PATH_ENV } from "./lib/claude-session-transfer.mjs"; @@ -103,29 +102,15 @@ async function handleSessionEnd(input) { } cleanupSessionJobs(cwd, input.session_id || process.env[SESSION_ID_ENV]); - // Closing the listener lets a replacement start while this broker drains. Teardown - // runs under the startup lock and clears saved state only if it is still this broker's. - const cleanup = () => { - teardownBrokerSession({ - endpoint: brokerEndpoint, - pidFile, - logFile, - sessionDir, - pid, - killProcess: terminateProcessTree - }); - clearBrokerSessionIfCurrent(cwd, brokerEndpoint); - }; - try { - await withBrokerLock(cwd, {}, cleanup); - } catch (error) { - // A starter that holds the lock too long must not leave this session's broker behind. - // The conditional clear still protects a replacement's state. - if (!/Timed out waiting for Codex broker startup lock/.test(error.message)) { - throw error; - } - cleanup(); - } + teardownBrokerSession({ + endpoint: brokerEndpoint, + pidFile, + logFile, + sessionDir, + pid, + killProcess: terminateProcessTree + }); + clearBrokerSession(cwd); } async function main() { diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 6833d1613..39fce9218 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -1,20 +1,17 @@ import fs from "node:fs"; import net from "node:net"; import path from "node:path"; -import { fileURLToPath } from "node:url"; import test from "node:test"; import assert from "node:assert/strict"; -import { spawn, spawnSync } from "node:child_process"; +import { spawn } from "node:child_process"; import { makeTempDir, writeExecutable } from "./helpers.mjs"; import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; -import { clearBrokerSessionIfCurrent, ensureBrokerSession, loadBrokerSession, saveBrokerSession, sendBrokerShutdown, teardownBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { ensureBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; import { parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; -import { isProcessAlive, terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; +import { terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; -const LIFECYCLE_URL = new URL("../plugins/codex/scripts/lib/broker-lifecycle.mjs", import.meta.url).href; - async function waitFor(predicate, timeoutMs = 3000) { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { @@ -24,7 +21,9 @@ async function waitFor(predicate, timeoutMs = 3000) { assert.fail("Timed out waiting for broker lifecycle transition"); } -const alive = isProcessAlive; +function alive(pid) { + try { process.kill(pid, 0); return true; } catch { return false; } +} function connect(endpoint) { const socket = net.createConnection({ path: parseBrokerEndpoint(endpoint).path }); @@ -67,9 +66,7 @@ const fs = require('node:fs'); const readline = require('node:readline'); fs.appendFileSync(process.env.TEST_STARTS, process.pid + '\\n'); const send = (message) => process.stdout.write(JSON.stringify(message) + '\\n'); -if (process.env.TEST_SLOW_EXIT) process.on('SIGTERM', () => {}); const rl = readline.createInterface({ input: process.stdin }); -if (process.env.TEST_SLOW_EXIT) rl.on('close', () => setTimeout(() => process.exit(0), 800)); rl.on('line', (line) => { const m = JSON.parse(line); if (m.id === undefined) return; @@ -77,17 +74,7 @@ rl.on('line', (line) => { if (m.method === 'turn/start') { reply({ turn: { id: 'turn_1' } }); send({ method: 'turn/started', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }); - if (m.params && m.params.mode === 'interrupted') { - send({ method: 'item/started', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status: 'inProgress' } } }); - setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1', status: 'interrupted' } } }), 100); - } else if (m.params && m.params.mode === 'final-only') { - setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm2', phase: 'final_answer', text: 'done' } } }), 100); - } else if (process.env.TEST_COLLAB) { - send({ method: 'item/started', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status: 'inProgress' } } }); - setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }), 100); - setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status: 'completed' } } }), 900); - } else if (process.env.TEST_INFERRED) setTimeout(() => send({ method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }), 100); - else setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }), 700); + setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }), 700); } else if (m.method === 'test/slow') { fs.writeFileSync(process.env.TEST_SLOW_SEEN, '1'); setTimeout(() => reply({ done: true }), 700); @@ -148,42 +135,6 @@ test("disconnecting a client does not expire an active turn", { timeout: 6000 }, await waitFor(() => !alive(session.pid), 1800); }); -test("a turn finished by its final answer alone does not pin the broker", { timeout: 6000 }, async (t) => { - const f = await fixture(t); - f.env.TEST_INFERRED = "1"; - const session = await f.start(); - const client = connect(session.endpoint); - await client.request("turn/start", { threadId: "thr_1" }); - client.socket.destroy(); - await waitFor(() => !alive(session.pid), 1500); -}); - -test("a final answer does not release the broker while a collaboration is still running", { timeout: 6000 }, async (t) => { - const f = await fixture(t); - f.env.TEST_COLLAB = "1"; - const session = await f.start(); - const client = connect(session.endpoint); - await client.request("turn/start", { threadId: "thr_1" }); - const childPid = Number(fs.readFileSync(f.env.TEST_STARTS, "utf8").trim()); - client.socket.destroy(); - // Final answer at ~100ms, collaboration completes at ~900ms; idle timeout is 200ms. - await new Promise((resolve) => setTimeout(resolve, 700)); - assert.ok(alive(childPid), "app-server stays up while the collaboration runs"); - assert.ok(alive(session.pid), "broker stays up while the collaboration runs"); - await waitFor(() => !alive(session.pid), 2500); -}); - -test("a resumed thread does not inherit an interrupted turn's collaboration", { timeout: 8000 }, async (t) => { - const f = await fixture(t); - const session = await f.start(); - const client = connect(session.endpoint); - await client.request("turn/start", { threadId: "thr_1", mode: "interrupted" }); - await new Promise((resolve) => setTimeout(resolve, 300)); - await client.request("turn/start", { threadId: "thr_1", mode: "final-only" }); - client.socket.destroy(); - await waitFor(() => !alive(session.pid), 2500); -}); - test("disconnecting a client does not expire an in-flight RPC", { timeout: 6000 }, async (t) => { const f = await fixture(t); const session = await f.start(); @@ -197,218 +148,6 @@ test("disconnecting a client does not expire an in-flight RPC", { timeout: 6000 await waitFor(() => !alive(session.pid), 1800); }); -test("concurrent processes start only one broker for a workspace", { timeout: 10000 }, async (t) => { - const f = await fixture(t); - f.env.CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS = "3000"; - const launch = () => new Promise((resolve, reject) => { - const child = spawn(process.execPath, ["--input-type=module", "-e", ` - import { ensureBrokerSession } from ${JSON.stringify(LIFECYCLE_URL)}; - const session = await ensureBrokerSession(process.cwd()); - console.log(JSON.stringify(session)); - `], { cwd: f.root, env: f.env, stdio: ["ignore", "pipe", "pipe"] }); - let stdout = "", stderr = ""; - child.stdout.on("data", (c) => { stdout += c; }); - child.stderr.on("data", (c) => { stderr += c; }); - child.on("error", reject); - child.on("exit", (code) => { - if (code !== 0) return reject(new Error(stderr)); - const session = JSON.parse(stdout); - if (session) f.sessions.push(session); - resolve(session); - }); - }); - const sessions = await Promise.all([launch(), launch()]); - assert.ok(sessions.every(Boolean)); - assert.equal(sessions[0].endpoint, sessions[1].endpoint); - assert.equal(fs.readFileSync(f.env.TEST_STARTS, "utf8").trim().split("\n").length, 1); -}); - -test("shutdown is bounded when a broker accepts a connection but never replies", { timeout: 3000 }, async (t) => { - const root = makeTempDir(); - const endpoint = process.platform === "win32" ? `pipe:\\\\.\\pipe\\${path.basename(root)}` : `unix:${path.join(root, "broker.sock")}`; - const sockets = new Set(); - const server = net.createServer((socket) => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); }); - await new Promise((resolve) => server.listen(parseBrokerEndpoint(endpoint).path, resolve)); - t.after(() => { - for (const socket of sockets) socket.destroy(); - server.close(); - fs.rmSync(root, { recursive: true, force: true }); - }); - await sendBrokerShutdown(endpoint, { timeoutMs: 100 }); -}); - -for (const file of ["pidFile", "logFile"]) { - test(`replacement teardown tolerates concurrent removal of its ${file}`, (t) => { - const root = makeTempDir(); - t.after(() => fs.rmSync(root, { recursive: true, force: true })); - const target = path.join(root, file); - fs.writeFileSync(target, "old broker"); - const unlink = fs.unlinkSync; - let raced = false; - t.mock.method(fs, "unlinkSync", (candidate) => { - if (candidate === target && !raced) { - raced = true; - // The draining broker removes the file just before replacement teardown does. - unlink(candidate); - } - return unlink(candidate); - }); - - assert.doesNotThrow(() => teardownBrokerSession({ [file]: target })); - assert.equal(raced, true, "exercise removal at the unlink boundary"); - assert.equal(fs.existsSync(target), false); - }); -} - -test("draining broker tolerates replacement teardown removing its pid file", { timeout: 6000 }, async (t) => { - const f = await fixture(t); - const hook = path.join(f.root, "cleanup-race.cjs"); - const marker = path.join(f.root, "cleanup-raced"); - // Inject the race inside the real detached broker, without timing sleeps or production hooks. - fs.writeFileSync(hook, ` -const fs = require('node:fs'); -const path = require('node:path'); -if (path.basename(process.argv[1] || '') === 'app-server-broker.mjs') { - const unlink = fs.unlinkSync; - for (const method of ['unlinkSync', 'rmSync']) { - const remove = fs[method]; - fs[method] = function(file, ...args) { - if (path.basename(String(file)) === 'broker.pid' && !fs.existsSync(${JSON.stringify(marker)})) { - unlink(file); - fs.writeFileSync(${JSON.stringify(marker)}, 'removed by replacement'); - } - return remove.call(this, file, ...args); - }; - } -} -`); - f.env.NODE_OPTIONS = `${f.env.NODE_OPTIONS ?? ""} --require ${JSON.stringify(hook)}`; - const session = await f.start(); - const client = connect(session.endpoint); - t.after(() => client.socket.destroy()); - await client.request("broker/shutdown"); - await waitFor(() => !alive(session.pid)); - - assert.equal(fs.readFileSync(marker, "utf8"), "removed by replacement"); - assert.doesNotMatch(fs.readFileSync(session.logFile, "utf8"), /ENOENT/); - assert.equal(fs.existsSync(session.pidFile), false); -}); - -test("startup recovers a lock left by a process that exited", { timeout: 6000 }, async (t) => { - const f = await fixture(t); - const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); - await new Promise((resolve) => child.on("exit", resolve)); - const stateDir = resolveStateDir(f.root); - fs.mkdirSync(stateDir, { recursive: true }); - fs.writeFileSync(path.join(stateDir, "broker.json.lock"), JSON.stringify({ pid: child.pid, token: "dead-owner" })); - const session = await f.start(); - assert.ok(alive(session.pid)); - assert.equal(fs.existsSync(path.join(stateDir, "broker.json.lock")), false); -}); - -test("startup recovers a lock whose owner write was cut short", { timeout: 6000 }, async (t) => { - const f = await fixture(t); - const stateDir = resolveStateDir(f.root); - const lockFile = path.join(stateDir, "broker.json.lock"); - fs.mkdirSync(stateDir, { recursive: true }); - fs.writeFileSync(lockFile, '{"pid":12'); - const old = new Date(Date.now() - 60000); - fs.utimesSync(lockFile, old, old); - const session = await f.start(); - assert.ok(alive(session.pid)); - assert.equal(fs.existsSync(lockFile), false); -}); - -test("startup recovers when a reclaimer died holding the reclaim guard", { timeout: 6000 }, async (t) => { - const f = await fixture(t); - const stateDir = resolveStateDir(f.root); - const lockFile = path.join(stateDir, "broker.json.lock"); - fs.mkdirSync(stateDir, { recursive: true }); - fs.writeFileSync(lockFile, JSON.stringify({ pid: 2 ** 22 + 1, token: "dead-owner" })); - fs.writeFileSync(`${lockFile}.reclaim`, JSON.stringify({ pid: 2 ** 22 + 2, token: "dead-reclaimer" })); - const session = await f.start(); - assert.ok(alive(session.pid)); - assert.equal(fs.existsSync(`${lockFile}.reclaim`), false); -}); - -test("a live reclaimer is not revoked however long it holds the guard", { timeout: 8000 }, async (t) => { - const f = await fixture(t); - const stateDir = resolveStateDir(f.root); - const lockFile = path.join(stateDir, "broker.json.lock"); - fs.mkdirSync(stateDir, { recursive: true }); - fs.writeFileSync(lockFile, JSON.stringify({ pid: 2 ** 22 + 1, token: "dead-owner" })); - const guard = JSON.stringify({ pid: process.pid, token: "slow-reclaimer" }); - fs.writeFileSync(`${lockFile}.reclaim`, guard); - const old = new Date(Date.now() - 60000); - fs.utimesSync(`${lockFile}.reclaim`, old, old); - await assert.rejects(ensureBrokerSession(f.root, { env: f.env, timeoutMs: 50 }), /Timed out waiting/); - assert.equal(fs.readFileSync(`${lockFile}.reclaim`, "utf8"), guard, "guard untouched"); - assert.equal(fs.existsSync(lockFile), true, "stale lock left for the live reclaimer"); -}); - -test("session-end hook still tears down its broker when the startup lock is held", { timeout: 15000 }, async (t) => { - const f = await fixture(t); - const session = await f.start(); - const lockFile = path.join(resolveStateDir(f.root), "broker.json.lock"); - fs.writeFileSync(lockFile, JSON.stringify({ pid: process.pid, token: "busy-starter" })); - t.after(() => fs.rmSync(lockFile, { force: true })); - const hook = fileURLToPath(new URL("../plugins/codex/scripts/session-lifecycle-hook.mjs", import.meta.url)); - const child = spawn(process.execPath, [hook, "SessionEnd"], { cwd: f.root, env: f.env, stdio: ["pipe", "ignore", "pipe"] }); - child.stdin.end(JSON.stringify({ hook_event_name: "SessionEnd", cwd: f.root, session_id: "s1" })); - let stderr = ""; - child.stderr.on("data", (c) => { stderr += c; }); - const code = await new Promise((resolve) => child.on("exit", resolve)); - assert.equal(code, 0, stderr); - assert.equal(alive(session.pid), false); - assert.equal(loadBrokerSession(f.root), null); -}); - -test("session-end cleanup keeps a replacement broker's saved state", () => { - const root = makeTempDir(); - const replacement = { endpoint: "unix:/tmp/replacement.sock", pidFile: null, logFile: null, sessionDir: null, pid: 1 }; - saveBrokerSession(root, replacement); - clearBrokerSessionIfCurrent(root, "unix:/tmp/old.sock"); - assert.deepEqual(loadBrokerSession(root), replacement); - clearBrokerSessionIfCurrent(root, replacement.endpoint); - assert.equal(loadBrokerSession(root), null); - fs.rmSync(resolveStateDir(root), { recursive: true, force: true }); - fs.rmSync(root, { recursive: true, force: true }); -}); - -test("startup recovers a lock owned by an exited zombie process", { timeout: 6000, skip: !fs.existsSync("/proc/self/stat") || spawnSync("perl", ["-e", "1"]).error !== undefined }, async (t) => { - const f = await fixture(t); - // perl forks a child that exits at once, then sleeps without wait(), leaving a zombie. - const parent = spawn("perl", ["-e", '$| = 1; my $p = fork; exit 0 unless $p; print "$p\\n"; sleep 30'], { stdio: ["ignore", "pipe", "ignore"] }); - t.after(() => parent.kill()); - const pid = await new Promise((resolve) => parent.stdout.once("data", (c) => resolve(Number(String(c).trim())))); - await waitFor(() => fs.readFileSync(`/proc/${pid}/stat`, "utf8").split(") ")[1][0] === "Z"); - assert.equal(isProcessAlive(pid), false, "zombie counts as exited"); - const stateDir = resolveStateDir(f.root); - fs.mkdirSync(stateDir, { recursive: true }); - fs.writeFileSync(path.join(stateDir, "broker.json.lock"), JSON.stringify({ pid, token: "zombie-owner" })); - const session = await f.start(); - assert.ok(alive(session.pid)); -}); - -test("a broker draining after idle expiry stops admitting clients", { timeout: 6000 }, async (t) => { - const f = await fixture(t); - f.env.TEST_SLOW_EXIT = "1"; - const session = await f.start(); - const client = connect(session.endpoint); - await client.request("initialize"); - client.socket.destroy(); - // Idle expiry has fired; the fake app-server holds the broker open for ~800ms more. - await new Promise((resolve) => setTimeout(resolve, 400)); - assert.ok(alive(session.pid), "broker is still draining"); - const late = connect(session.endpoint); - const outcome = await new Promise((resolve) => { - late.socket.on("connect", () => resolve("connected")); - late.socket.on("error", () => resolve("refused")); - }); - late.socket.destroy(); - assert.equal(outcome, "refused"); -}); - test("reuse-existing callers recover when the saved broker expired while idle", { timeout: 6000 }, async (t) => { const f = await fixture(t); const session = await f.start(); From 5f2688cba84b4184d7fe82efe376065d9665074c Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Thu, 12 Nov 2026 09:59:52 -0600 Subject: [PATCH 15/16] fix: restore broker drain, final-answer and cleanup-race handling, clear saved pid on exit The trim to the idle-shutdown fix dropped four behaviours the idle exit depends on: - An idle broker that kept its listener open while the app-server closed admitted a replacement client that then failed its first RPC. Stop accepting before the close. - A turn finished by its final answer alone (no turn/completed) pinned the broker, and so the conversation's writer lock, forever. Mirror the client's inferred completion, including waiting for pending collaborations and not letting a resumed thread inherit an interrupted turn's. - Cleanup used existsSync then unlinkSync on files a replacement's teardown removes concurrently; the loser threw ENOENT. Use rmSync with force. An expired broker also left broker.json pointing at its pid, which SessionEnd passes to terminateProcessTree after the OS may have reused it. The broker now clears the saved session on exit, only when it still names this broker's endpoint, so a replacement's state is kept. Co-Authored-By: Claude Sonnet 5.5 --- plugins/codex/scripts/app-server-broker.mjs | 99 ++++++++++---- .../codex/scripts/lib/broker-lifecycle.mjs | 24 ++-- tests/broker-lifecycle.test.mjs | 125 +++++++++++++++++- 3 files changed, 211 insertions(+), 37 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index e8b9aed0e..cace53dbc 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/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 { clearBrokerSessionIfCurrent } from "./lib/broker-lifecycle.mjs"; const STREAMING_METHODS = new Set(["turn/start", "review/start", "thread/compact/start"]); @@ -71,6 +72,8 @@ async function main() { let activeStreamThreadIds = null; const sockets = new Set(); const activeTurns = new Set(); + const pendingCollaborations = new Map(); + const finalAnswered = new Map(); let pendingRequests = 0; let idleTimer = null; let shuttingDown = false; @@ -89,6 +92,16 @@ async function main() { }, idleTimeoutMs); } + function settleFinalAnswer(turnKey, threadId) { + if (pendingCollaborations.get(threadId)?.size) return; + setTimeout(() => { + if (pendingCollaborations.get(threadId)?.size || !finalAnswered.has(turnKey)) return; + finalAnswered.delete(turnKey); + activeTurns.delete(turnKey); + updateIdleTimer(); + }, 250).unref(); + } + function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { activeRequestSocket = null; @@ -101,9 +114,45 @@ async function main() { function routeNotification(message) { // A disconnected caller must not make its running turn eligible for shutdown. - const turnKey = `${message.params?.threadId}:${message.params?.turn?.id}`; - if (message.method === "turn/started") activeTurns.add(turnKey); - if (message.method === "turn/completed") activeTurns.delete(turnKey); + const turnKey = `${message.params?.threadId}:${message.params?.turn?.id ?? message.params?.turnId}`; + const threadId = message.params?.threadId; + if (message.method === "turn/started") { + activeTurns.add(turnKey); + pendingCollaborations.delete(threadId); + } + if (message.method === "turn/completed") { + activeTurns.delete(turnKey); + finalAnswered.delete(turnKey); + // Collaborations belong to the turn that started them; an interrupted turn may never + // have seen them complete, and a resumed thread must not inherit them. + pendingCollaborations.delete(threadId); + } + // lib/codex.mjs completes a turn from its final answer when turn/completed never + // arrives, once the thread's collaborations drain. Mirror that so such a turn does not + // pin the broker, and an unfinished collaboration is not cut off by idle expiry. + const item = message.params?.item; + if (item?.type === "collabAgentToolCall") { + const pending = pendingCollaborations.get(threadId) ?? new Set(); + if (message.method === "item/started" || item.status === "inProgress") { + pending.add(item.id); + } else if (message.method === "item/completed") { + pending.delete(item.id); + } + if (pending.size) { + pendingCollaborations.set(threadId, pending); + } else { + pendingCollaborations.delete(threadId); + } + if (message.method === "item/completed") { + for (const [key, owner] of finalAnswered) { + if (owner === threadId) settleFinalAnswer(key, owner); + } + } + } + if (message.method === "item/completed" && item?.type === "agentMessage" && item.phase === "final_answer") { + finalAnswered.set(turnKey, threadId); + settleFinalAnswer(turnKey, threadId); + } updateIdleTimer(); const target = activeRequestSocket ?? activeStreamSocket; if (!target) { @@ -126,17 +175,23 @@ async function main() { if (shuttingDown) return; shuttingDown = true; clearTimeout(idleTimer); + // Stop admitting clients before the app-server close, which can take seconds. + // A client admitted while draining would initialize fine, then fail its first request. + const closed = new Promise((resolve) => server.close(resolve)); for (const socket of sockets) { socket.end(); } await appClient.close().catch(() => {}); - await new Promise((resolve) => server.close(resolve)); - if (listenTarget.kind === "unix" && fs.existsSync(listenTarget.path)) { - fs.unlinkSync(listenTarget.path); + await closed; + // A replacement broker's teardown may already have removed these. + if (listenTarget.kind === "unix") { + fs.rmSync(listenTarget.path, { force: true }); } - if (pidFile && fs.existsSync(pidFile)) { - fs.unlinkSync(pidFile); + if (pidFile) { + fs.rmSync(pidFile, { force: true }); } + // The saved pid would otherwise outlive this process, and SessionEnd signals it. + clearBrokerSessionIfCurrent(cwd, endpoint); } appClient.setNotificationHandler(routeNotification); @@ -208,30 +263,19 @@ async function main() { continue; } - if (allowInterruptDuringActiveStream) { - pendingRequests += 1; - try { - const result = await appClient.request(message.method, message.params ?? {}); - send(socket, { id: message.id, result }); - } catch (error) { - send(socket, { - id: message.id, - error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) - }); - } finally { - pendingRequests -= 1; - updateIdleTimer(); - } - continue; - } - const isStreaming = STREAMING_METHODS.has(message.method); - activeRequestSocket = socket; + if (!allowInterruptDuringActiveStream) { + activeRequestSocket = socket; + } pendingRequests += 1; try { const result = await appClient.request(message.method, message.params ?? {}); send(socket, { id: message.id, result }); + // Interrupts share request bookkeeping without taking stream ownership. + if (allowInterruptDuringActiveStream) { + continue; + } if (isStreaming) { activeStreamSocket = socket; activeStreamThreadIds = buildStreamThreadIds(message.method, message.params ?? {}, result); @@ -244,6 +288,9 @@ async function main() { id: message.id, error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) }); + if (allowInterruptDuringActiveStream) { + continue; + } if (activeRequestSocket === socket) { activeRequestSocket = null; } diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index ae49becee..0d3568e18 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -93,9 +93,14 @@ export function saveBrokerSession(cwd, session) { } export function clearBrokerSession(cwd) { - const stateFile = resolveBrokerStateFile(cwd); - if (fs.existsSync(stateFile)) { - fs.unlinkSync(stateFile); + // An exiting broker and a SessionEnd cleanup can both clear the same state. + fs.rmSync(resolveBrokerStateFile(cwd), { force: true }); +} + +export function clearBrokerSessionIfCurrent(cwd, endpoint) { + // A replacement may have published its own state since this broker started. + if (loadBrokerSession(cwd)?.endpoint === endpoint) { + clearBrokerSession(cwd); } } @@ -179,19 +184,20 @@ export function teardownBrokerSession({ endpoint = null, pidFile, logFile, sessi } } - if (pidFile && fs.existsSync(pidFile)) { - fs.unlinkSync(pidFile); + // A draining broker removes its own pid file and socket concurrently with a replacement. + if (pidFile) { + fs.rmSync(pidFile, { force: true }); } - if (logFile && fs.existsSync(logFile)) { - fs.unlinkSync(logFile); + if (logFile) { + fs.rmSync(logFile, { force: true }); } if (endpoint) { try { const target = parseBrokerEndpoint(endpoint); - if (target.kind === "unix" && fs.existsSync(target.path)) { - fs.unlinkSync(target.path); + if (target.kind === "unix") { + fs.rmSync(target.path, { force: true }); } } catch { // Ignore malformed or already-removed broker endpoints during teardown. diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 39fce9218..16b5d8110 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -7,7 +7,7 @@ import { spawn } from "node:child_process"; import { makeTempDir, writeExecutable } from "./helpers.mjs"; import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; -import { ensureBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { clearBrokerSessionIfCurrent, ensureBrokerSession, loadBrokerSession, saveBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; import { parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; import { terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; @@ -66,7 +66,11 @@ const fs = require('node:fs'); const readline = require('node:readline'); fs.appendFileSync(process.env.TEST_STARTS, process.pid + '\\n'); const send = (message) => process.stdout.write(JSON.stringify(message) + '\\n'); +if (process.env.TEST_SLOW_EXIT) process.on('SIGTERM', () => {}); const rl = readline.createInterface({ input: process.stdin }); +if (process.env.TEST_SLOW_EXIT) rl.on('close', () => setTimeout(() => process.exit(0), 800)); +const finalAnswer = { method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }; +const collab = (method, status) => ({ method, params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status } } }); rl.on('line', (line) => { const m = JSON.parse(line); if (m.id === undefined) return; @@ -74,7 +78,16 @@ rl.on('line', (line) => { if (m.method === 'turn/start') { reply({ turn: { id: 'turn_1' } }); send({ method: 'turn/started', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }); - setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }), 700); + if (m.params.mode === 'interrupted') { + send(collab('item/started', 'inProgress')); + setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1', status: 'interrupted' } } }), 100); + } else if (m.params.mode === 'final-only') { + setTimeout(() => send(finalAnswer), 100); + } else if (m.params.mode === 'final-with-collab') { + send(collab('item/started', 'inProgress')); + setTimeout(() => send(finalAnswer), 100); + setTimeout(() => send(collab('item/completed', 'completed')), 900); + } else setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1' } } }), 700); } else if (m.method === 'test/slow') { fs.writeFileSync(process.env.TEST_SLOW_SEEN, '1'); setTimeout(() => reply({ done: true }), 700); @@ -156,3 +169,111 @@ test("reuse-existing callers recover when the saved broker expired while idle", t.after(() => client.close()); assert.equal(client.transport, "direct"); }); + +test("a turn finished by its final answer alone does not pin the broker", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("turn/start", { threadId: "thr_1", mode: "final-only" }); + client.socket.destroy(); + await waitFor(() => !alive(session.pid), 1500); +}); + +test("a final answer does not release the broker while a collaboration is still running", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("turn/start", { threadId: "thr_1", mode: "final-with-collab" }); + const childPid = Number(fs.readFileSync(f.env.TEST_STARTS, "utf8").trim()); + client.socket.destroy(); + // Final answer at ~100ms, collaboration completes at ~900ms; idle timeout is 200ms. + await new Promise((resolve) => setTimeout(resolve, 700)); + assert.ok(alive(childPid), "app-server stays up while the collaboration runs"); + assert.ok(alive(session.pid), "broker stays up while the collaboration runs"); + await waitFor(() => !alive(session.pid), 2500); +}); + +test("a resumed thread does not inherit an interrupted turn's collaboration", { timeout: 8000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("turn/start", { threadId: "thr_1", mode: "interrupted" }); + await new Promise((resolve) => setTimeout(resolve, 300)); + await client.request("turn/start", { threadId: "thr_1", mode: "final-only" }); + client.socket.destroy(); + await waitFor(() => !alive(session.pid), 2500); +}); + +test("a broker draining after idle expiry stops admitting clients", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + f.env.TEST_SLOW_EXIT = "1"; + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("initialize"); + client.socket.destroy(); + // Idle expiry has fired; the fake app-server holds the broker open for ~800ms more. + await new Promise((resolve) => setTimeout(resolve, 400)); + assert.ok(alive(session.pid), "broker is still draining"); + const late = connect(session.endpoint); + const outcome = await new Promise((resolve) => { + late.socket.on("connect", () => resolve("connected")); + late.socket.on("error", () => resolve("refused")); + }); + late.socket.destroy(); + assert.equal(outcome, "refused"); +}); + +test("draining broker tolerates replacement teardown removing its pid file", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const hook = path.join(f.root, "cleanup-race.cjs"); + const marker = path.join(f.root, "cleanup-raced"); + // Inject the race inside the real detached broker, without timing sleeps or production hooks. + fs.writeFileSync(hook, ` +const fs = require('node:fs'); +const path = require('node:path'); +if (path.basename(process.argv[1] || '') === 'app-server-broker.mjs') { + const unlink = fs.unlinkSync; + for (const method of ['unlinkSync', 'rmSync']) { + const remove = fs[method]; + fs[method] = function(file, ...args) { + if (path.basename(String(file)) === 'broker.pid' && !fs.existsSync(${JSON.stringify(marker)})) { + unlink(file); + fs.writeFileSync(${JSON.stringify(marker)}, 'removed by replacement'); + } + return remove.call(this, file, ...args); + }; + } +} +`); + f.env.NODE_OPTIONS = `${f.env.NODE_OPTIONS ?? ""} --require ${JSON.stringify(hook)}`; + const session = await f.start(); + const client = connect(session.endpoint); + t.after(() => client.socket.destroy()); + await client.request("broker/shutdown"); + await waitFor(() => !alive(session.pid)); + + assert.equal(fs.readFileSync(marker, "utf8"), "removed by replacement"); + assert.doesNotMatch(fs.readFileSync(session.logFile, "utf8"), /ENOENT/); + assert.equal(fs.existsSync(session.pidFile), false); +}); + +test("an expired broker clears its saved session so SessionEnd never signals a reused pid", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + assert.equal(loadBrokerSession(f.root)?.pid, session.pid); + await waitFor(() => !alive(session.pid), 1500); + assert.equal(loadBrokerSession(f.root), null); +}); + +test("clearing a saved session keeps a replacement broker's state", (t) => { + const f = { root: makeTempDir() }; + t.after(() => { + fs.rmSync(f.root, { recursive: true, force: true }); + fs.rmSync(resolveStateDir(f.root), { recursive: true, force: true }); + }); + saveBrokerSession(f.root, { endpoint: "unix:/tmp/new.sock", pid: 2 }); + clearBrokerSessionIfCurrent(f.root, "unix:/tmp/old.sock"); + assert.equal(loadBrokerSession(f.root)?.pid, 2); + clearBrokerSessionIfCurrent(f.root, "unix:/tmp/new.sock"); + assert.equal(loadBrokerSession(f.root), null); +}); From 1860c00e0e34d9f14f57cea4ae0f269830d69b6b Mon Sep 17 00:00:00 2001 From: pecord-ent Date: Thu, 12 Nov 2026 12:53:49 -0600 Subject: [PATCH 16/16] Restart the broker's final-answer grace timer on each collaboration drain Each drain armed a new 250ms timeout without cancelling the previous one, so an older timeout could release the turn before the latest grace period ended. A collaboration starting in that window was then unprotected once the caller disconnected, and idle expiry could close its app-server. lib/codex.mjs keeps a single resettable timer per turn; the broker now does the same. Co-Authored-By: Claude Sonnet 5.5 --- plugins/codex/scripts/app-server-broker.mjs | 16 ++++++++++++-- tests/broker-lifecycle.test.mjs | 24 ++++++++++++++++++++- 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index cace53dbc..6d9378144 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -74,6 +74,7 @@ async function main() { const activeTurns = new Set(); const pendingCollaborations = new Map(); const finalAnswered = new Map(); + const settleTimers = new Map(); let pendingRequests = 0; let idleTimer = null; let shuttingDown = false; @@ -92,14 +93,24 @@ async function main() { }, idleTimeoutMs); } + function cancelSettle(turnKey) { + clearTimeout(settleTimers.get(turnKey)); + settleTimers.delete(turnKey); + } + + // One timer per turn: a collaboration that drains again restarts the grace period. function settleFinalAnswer(turnKey, threadId) { + cancelSettle(turnKey); if (pendingCollaborations.get(threadId)?.size) return; - setTimeout(() => { + const timer = setTimeout(() => { + settleTimers.delete(turnKey); if (pendingCollaborations.get(threadId)?.size || !finalAnswered.has(turnKey)) return; finalAnswered.delete(turnKey); activeTurns.delete(turnKey); updateIdleTimer(); - }, 250).unref(); + }, 250); + timer.unref(); + settleTimers.set(turnKey, timer); } function clearSocketOwnership(socket) { @@ -123,6 +134,7 @@ async function main() { if (message.method === "turn/completed") { activeTurns.delete(turnKey); finalAnswered.delete(turnKey); + cancelSettle(turnKey); // Collaborations belong to the turn that started them; an interrupted turn may never // have seen them complete, and a resumed thread must not inherit them. pendingCollaborations.delete(threadId); diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 16b5d8110..0fc6f686d 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -70,7 +70,7 @@ if (process.env.TEST_SLOW_EXIT) process.on('SIGTERM', () => {}); const rl = readline.createInterface({ input: process.stdin }); if (process.env.TEST_SLOW_EXIT) rl.on('close', () => setTimeout(() => process.exit(0), 800)); const finalAnswer = { method: 'item/completed', params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'agentMessage', id: 'm1', phase: 'final_answer', text: 'done' } } }; -const collab = (method, status) => ({ method, params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id: 'c1', status } } }); +const collab = (method, status, id = 'c1') => ({ method, params: { threadId: 'thr_1', turnId: 'turn_1', item: { type: 'collabAgentToolCall', id, status } } }); rl.on('line', (line) => { const m = JSON.parse(line); if (m.id === undefined) return; @@ -83,6 +83,14 @@ rl.on('line', (line) => { setTimeout(() => send({ method: 'turn/completed', params: { threadId: 'thr_1', turn: { id: 'turn_1', status: 'interrupted' } } }), 100); } else if (m.params.mode === 'final-only') { setTimeout(() => send(finalAnswer), 100); + } else if (m.params.mode === 'final-with-repeated-drains') { + send(collab('item/started', 'inProgress')); + setTimeout(() => send(finalAnswer), 50); + setTimeout(() => send(collab('item/completed', 'completed')), 100); + setTimeout(() => send(collab('item/started', 'inProgress', 'c2')), 150); + setTimeout(() => send(collab('item/completed', 'completed', 'c2')), 200); + setTimeout(() => send(collab('item/started', 'inProgress', 'c3')), 400); + setTimeout(() => send(collab('item/completed', 'completed', 'c3')), 1100); } else if (m.params.mode === 'final-with-collab') { send(collab('item/started', 'inProgress')); setTimeout(() => send(finalAnswer), 100); @@ -193,6 +201,20 @@ test("a final answer does not release the broker while a collaboration is still await waitFor(() => !alive(session.pid), 2500); }); +test("repeated collaboration drains share one grace timer, so a later collaboration keeps the turn open", { timeout: 6000 }, async (t) => { + const f = await fixture(t); + const session = await f.start(); + const client = connect(session.endpoint); + await client.request("turn/start", { threadId: "thr_1", mode: "final-with-repeated-drains" }); + const childPid = Number(fs.readFileSync(f.env.TEST_STARTS, "utf8").trim()); + client.socket.destroy(); + // The first drain's timer would fire at ~350ms and release the turn before the third + // collaboration starts at 400ms; the 200ms idle timeout would then close the app-server. + await new Promise((resolve) => setTimeout(resolve, 900)); + assert.ok(alive(childPid), "app-server stays up while the third collaboration runs"); + await waitFor(() => !alive(session.pid), 2500); +}); + test("a resumed thread does not inherit an interrupted turn's collaboration", { timeout: 8000 }, async (t) => { const f = await fixture(t); const session = await f.start();