diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 1954274fe..7ace7c3a0 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -8,9 +8,37 @@ import process from "node:process"; import { parseArgs } from "./lib/args.mjs"; import { BROKER_BUSY_RPC_CODE, CodexAppServerClient } from "./lib/app-server.mjs"; import { parseBrokerEndpoint } from "./lib/broker-endpoint.mjs"; +import { + clearBrokerSession, + loadBrokerSession, + teardownBrokerSession +} from "./lib/broker-lifecycle.mjs"; +import { + armTimeout, + brokerIdleShutdownMs, + brokerStartupTimeoutMs, + disarmTimeout +} from "./lib/lifecycle-limits.mjs"; +import { terminateProcessTreeAndExit } from "./lib/process.mjs"; const STREAMING_METHODS = new Set(["turn/start", "review/start", "thread/compact/start"]); +/** How long each step of a shutdown waits before giving up and going down without it. */ +const SHUTDOWN_GRACE_MS = 5000; + +/** How many abandoned turns to keep discarding notifications for. */ +const ABANDONED_THREAD_MEMORY = 32; + +/** A deadline that never keeps the event loop alive on its own. */ +function grace(ms = SHUTDOWN_GRACE_MS) { + return new Promise((resolve) => { + setTimeout(resolve, ms).unref?.(); + }); +} + +/** Whether the app-server — and with it every configured MCP server — has been started. */ +let backendStarted = false; + function buildStreamThreadIds(method, params, result) { const threadIds = new Set(); if (params?.threadId) { @@ -22,6 +50,21 @@ function buildStreamThreadIds(method, params, result) { return threadIds; } +/** + * The single thread the started turn actually runs on. + * + * A detached review streams on the review thread it just created, not on the source thread it was + * launched from. Both need routing, but only this one will ever produce a completion — marking the + * other abandoned would leave it marked for good, and every later turn resumed on it would have + * its notifications discarded. + */ +function turnThreadId(method, params, result) { + if (method === "review/start") { + return result?.reviewThreadId ?? params?.threadId ?? null; + } + return params?.threadId ?? null; +} + function buildJsonRpcError(code, message, data) { return data === undefined ? { code, message } : { code, message, data }; } @@ -48,11 +91,11 @@ function writePidFile(pidFile) { async function main() { const [subcommand, ...argv] = process.argv.slice(2); if (subcommand !== "serve") { - throw new Error("Usage: node scripts/app-server-broker.mjs serve --endpoint [--cwd ] [--pid-file ]"); + throw new Error("Usage: node scripts/app-server-broker.mjs serve --endpoint [--cwd ] [--pid-file ] [--log-file ]"); } const { options } = parseArgs(argv, { - valueOptions: ["cwd", "pid-file", "endpoint"] + valueOptions: ["cwd", "pid-file", "endpoint", "log-file"] }); if (!options.endpoint) { @@ -63,13 +106,36 @@ async function main() { const endpoint = String(options.endpoint); const listenTarget = parseBrokerEndpoint(endpoint); const pidFile = options["pid-file"] ? path.resolve(options["pid-file"]) : null; + const logFile = options["log-file"] ? path.resolve(options["log-file"]) : null; writePidFile(pidFile); + // Connecting spawns the app-server, which in turn spawns every configured MCP server. Until the + // listener below is up there is no idle timer and no parent watching — the spawning client gives + // up after a couple of seconds and, on the normal path, kills nothing. So bound the startup + // itself: a wedged connect would otherwise strand this whole tree for good. + const startupTimeoutMs = brokerStartupTimeoutMs(); + const startupTimer = armTimeout(startupTimeoutMs, () => { + process.stderr.write(`broker startup exceeded ${startupTimeoutMs}ms; terminating\n`); + terminateProcessTreeAndExit(process.pid); + }); + const appClient = await CodexAppServerClient.connect(cwd, { disableBroker: true }); + backendStarted = true; let activeRequestSocket = null; let activeStreamSocket = null; let activeStreamThreadIds = null; const sockets = new Set(); + // Turns whose client left before the stream could be handed over: their notifications go + // nowhere, and they are interrupted rather than left running for the next client to receive. + const abandonedThreadIds = new Set(); + // Completions seen while a streaming start is still awaiting its response. Scoped to that + // window and cleared when it closes, so nothing here can outlive the handoff it belongs to. + const completedDuringHandoff = new Set(); + let pendingStreamStarts = 0; + const idleShutdownMs = brokerIdleShutdownMs(); + let idleTimer = null; + let shuttingDown = false; + let shutdownPromise = null; function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { @@ -81,42 +147,184 @@ async function main() { } } + /** + * Stop a turn whose client left before it could be handed the stream. + * + * Leaving it running is not harmless: nobody is reading it, and its notifications would be + * delivered to whichever client connects next, because routing follows whoever currently owns + * the broker rather than the turn that produced them. + */ + async function abandonStream(threadId, turnId) { + if (!threadId) { + return; + } + // Bounded: an interrupt need not be followed by a completion, and an entry left here + // silently discards every future turn on that thread. + if (abandonedThreadIds.size >= ABANDONED_THREAD_MEMORY) { + abandonedThreadIds.delete(abandonedThreadIds.values().next().value); + } + abandonedThreadIds.add(threadId); + try { + // The turn id is required alongside the thread; without it the interrupt is rejected and + // the turn we meant to stop keeps running while its thread stays marked abandoned. + await appClient.request("turn/interrupt", { threadId, turnId }); + } catch { + // Best effort: the turn may already be finishing on its own. + } + } + function routeNotification(message) { + const threadId = message.params?.threadId ?? null; + + // An abandoned turn belongs to a client that is gone. Never hand it to whoever is here now. + if (threadId && abandonedThreadIds.has(threadId)) { + if (message.method === "turn/completed") { + abandonedThreadIds.delete(threadId); + armIdleShutdown(); + } + return; + } + + // The response and its completion can arrive in one chunk, so a completion can land before the + // request continuation assigns ownership. Remember it, or that continuation would take + // ownership of a turn that is already over and hold the broker busy until the client leaves. + // + // Recorded by window rather than by identity: a detached review runs on a thread it only names + // in its response, so there is nothing to match against beforehand. The window is what makes + // this safe — an entry cannot outlive the start it raced, and the continuation checks it + // against the threads the response actually reports. + if (message.method === "turn/completed" && threadId && pendingStreamStarts > 0) { + completedDuringHandoff.add(threadId); + } + const target = activeRequestSocket ?? activeStreamSocket; if (!target) { return; } send(target, message); if (message.method === "turn/completed" && activeStreamSocket === target) { - const threadId = message.params?.threadId ?? null; if (!threadId || !activeStreamThreadIds || activeStreamThreadIds.has(threadId)) { activeStreamSocket = null; activeStreamThreadIds = null; if (activeRequestSocket === target) { activeRequestSocket = null; } + // Releasing ownership can be the moment the broker becomes idle — the client may already + // have disconnected mid-turn. Nothing else fires afterwards, so if the timer is not armed + // here it never is. + armIdleShutdown(); } } } + function isBrokerBusy() { + return sockets.size > 0 || activeRequestSocket !== null || activeStreamSocket !== null; + } + + function cancelIdleShutdown() { + disarmTimeout(idleTimer); + idleTimer = null; + } + + // A broker outlives the client that spawned it, so without this it survives a crashed or + // timed-out SessionEnd hook and keeps its app-server (and every MCP server under it) alive + // indefinitely. Re-armed whenever the last client disconnects, cancelled when one connects. + function armIdleShutdown() { + cancelIdleShutdown(); + // A shutdown already in flight must not be rescheduled behind itself; sockets closing as part + // of it would otherwise arm a timer for a broker that is on its way out. + if (shuttingDown || isBrokerBusy()) { + return; + } + idleTimer = armTimeout(idleShutdownMs, async () => { + idleTimer = null; + if (isBrokerBusy()) { + armIdleShutdown(); + return; + } + await shutdown(server).catch(() => {}); + process.exit(0); + }); + } + async function shutdown(server) { + // Every entry point — the idle timer, `broker/shutdown`, SIGTERM, SIGINT — can land while + // another is mid-flight. Run once and let the rest await that same pass. + if (shutdownPromise) { + return shutdownPromise; + } + shutdownPromise = runShutdown(server); + return shutdownPromise; + } + + async function runShutdown(server) { + cancelIdleShutdown(); + // Stop accepting before the first await. Otherwise a client can connect while we are closing + // the app-server, get a broker that looks alive but has no backend, and hold server.close() + // open on a socket nobody will serve. + shuttingDown = true; + 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); + + // A wedged app-server must not outlive the guard meant to reclaim it: waiting on it forever + // is exactly how the tree survives. + let backendClosed = false; + const settled = () => { + backendClosed = true; + }; + await Promise.race([appClient.close().then(settled, settled), grace()]); + + // `end()` only half-closes: a client holding its read side open keeps server.close() pending + // for as long as it likes, which would hang SIGTERM and broker/shutdown just as surely. + await Promise.race([closed, grace()]); + for (const socket of sockets) { + socket.destroy(); + } + + // Clean up after ourselves. When the idle timer fires there is no session-end hook to run + // teardown for us, so these would otherwise survive every expiry. + // + // The artifacts are ours unconditionally — we were told their paths at startup precisely so + // that this does not depend on the shared record, which by now may name a broker that + // superseded us while we sat idle. The record itself is the one thing we must not touch in + // that case: it belongs to whoever it points at. + try { + teardownBrokerSession({ endpoint, pidFile, logFile }); + } catch { + // Best effort; never let bookkeeping block the shutdown. + } + + try { + if (loadBrokerSession(cwd)?.endpoint === endpoint) { + clearBrokerSession(cwd); + } + } catch { + // Best effort; a record we cannot read is one we must not delete. } - if (pidFile && fs.existsSync(pidFile)) { - fs.unlinkSync(pidFile); + + // If the app-server never acknowledged the close, it is still running — and on POSIX the + // client's own fallback signals only its direct pid, so the MCP servers under it would outlive + // this broker and defeat the whole point of shutting down. Take the group with us. Done last, + // after the artifacts and the record are already cleaned up, because this ends us too. + if (!backendClosed) { + terminateProcessTreeAndExit(process.pid); } } appClient.setNotificationHandler(routeNotification); const server = net.createServer((socket) => { + if (shuttingDown) { + // Racing a shutdown already in flight: refuse cleanly so the caller falls back to starting + // its own broker rather than talking to one whose app-server is going away. + socket.destroy(); + return; + } sockets.add(socket); + cancelIdleShutdown(); socket.setEncoding("utf8"); let buffer = ""; @@ -143,6 +351,17 @@ async function main() { continue; } + // `null`, a bare number and an array are all valid JSON. Dereferencing them below would + // throw inside this async listener, which on current Node takes the whole broker down — + // detached, so nothing tears down its app-server or its session record. + if (message === null || typeof message !== "object" || Array.isArray(message)) { + send(socket, { + id: null, + error: buildJsonRpcError(-32600, "Invalid JSON-RPC message: expected an object.") + }); + continue; + } + if (message.id !== undefined && message.method === "initialize") { send(socket, { id: message.id, @@ -197,15 +416,45 @@ async function main() { const isStreaming = STREAMING_METHODS.has(message.method); activeRequestSocket = socket; + // A detached review streams on a thread it only names in its response, so a completion + // racing that response cannot be recognised by thread id in advance. Record completions + // for the duration of the start instead, and match them once the response tells us which + // threads this turn actually uses. + if (isStreaming) { + pendingStreamStarts += 1; + } + try { const result = await appClient.request(message.method, message.params ?? {}); send(socket, { id: message.id, result }); if (isStreaming) { - activeStreamSocket = socket; - activeStreamThreadIds = buildStreamThreadIds(message.method, message.params ?? {}, result); + const threadIds = buildStreamThreadIds(message.method, message.params ?? {}, result); + const finishedAlready = [...threadIds].some((id) => completedDuringHandoff.has(id)); + if (finishedAlready) { + // Over before we got here: nothing to hand over, and nothing to abandon. Marking it + // abandoned now would be permanent — interrupting a finished turn need not produce + // another completion to clear the mark, and every later turn on that thread would + // have its notifications discarded, including the one its caller is waiting for. + } else if (!sockets.has(socket)) { + // The client left while the turn was starting. Taking ownership on its behalf would + // hold the broker busy for a socket nobody reads; leaving the turn running would let + // its notifications reach the next client. Stop it instead — the turn's own thread + // only, since that is the one that will report the completion clearing the mark. + await abandonStream( + turnThreadId(message.method, message.params ?? {}, result), + result?.turn?.id ?? null + ); + } else { + activeStreamSocket = socket; + activeStreamThreadIds = threadIds; + } } if (activeRequestSocket === socket) { activeRequestSocket = null; + // The close handler could not arm the timer while this request still owned the broker, + // and for a non-streaming request no notification follows to do it later. Releasing + // ownership here can therefore be the last event the broker ever sees. + armIdleShutdown(); } } catch (error) { send(socket, { @@ -217,6 +466,17 @@ async function main() { } if (activeStreamSocket === socket && !isStreaming) { activeStreamSocket = null; + activeStreamThreadIds = null; + } + armIdleShutdown(); + } finally { + // The handoff is over either way. Once no start is in flight, a completion belongs to a + // running turn rather than to a race, so nothing recorded here may survive. + if (isStreaming) { + pendingStreamStarts -= 1; + if (pendingStreamStarts === 0) { + completedDuringHandoff.clear(); + } } } } @@ -225,11 +485,13 @@ async function main() { socket.on("close", () => { sockets.delete(socket); clearSocketOwnership(socket); + armIdleShutdown(); }); socket.on("error", () => { sockets.delete(socket); clearSocketOwnership(socket); + armIdleShutdown(); }); }); @@ -243,10 +505,32 @@ async function main() { process.exit(0); }); - server.listen(listenTarget.path); + // Startup is over once we are accepting; from here the idle timer takes over. A broker nobody + // ever connects to must not linger either, so arm it immediately. + // + // A listen failure — a stale socket path, a permission problem, an address already in use — + // must reach main()'s handler rather than surfacing as an unhandled error event, or the process + // dies with its app-server and MCP servers still running and its session record still on disk. + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(listenTarget.path, () => { + server.off("error", reject); + disarmTimeout(startupTimer); + armIdleShutdown(); + resolve(); + }); + }); } main().catch((error) => { process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); + // Once the app-server is up it has its own MCP servers under it, and this process is detached — + // exiting alone would leave that tree with no parent and no record, the leak this script exists + // to prevent. Only then, though: a bad argument fails before anything was spawned, and there is + // nothing to take down. + if (backendStarted) { + terminateProcessTreeAndExit(process.pid); + return; + } process.exit(1); }); diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs index 83df468ad..aafd7364a 100644 --- a/plugins/codex/scripts/codex-companion.mjs +++ b/plugins/codex/scripts/codex-companion.mjs @@ -24,7 +24,8 @@ import { import { resolveClaudeSessionPath } from "./lib/claude-session-transfer.mjs"; import { readStdinIfPiped } from "./lib/fs.mjs"; import { collectReviewContext, ensureGitRepository, resolveReviewTarget } from "./lib/git.mjs"; -import { binaryAvailable, terminateProcessTree } from "./lib/process.mjs"; +import { armTimeout, disarmTimeout, workerTtlMs } from "./lib/lifecycle-limits.mjs"; +import { binaryAvailable, terminateProcessTree, terminateProcessTreeAndExit } from "./lib/process.mjs"; import { loadPromptTemplate, interpolateTemplate } from "./lib/prompts.mjs"; import { generateJobId, @@ -865,19 +866,96 @@ async function handleTaskWorker(argv) { logFile: storedJob.logFile ?? null } ); - await runTrackedJob( - { - ...storedJob, - workspaceRoot, - logFile - }, - () => - executeTaskRun({ - ...request, - onProgress: progress - }), - { logFile } - ); + const releaseTtl = armWorkerTtl({ workspaceRoot, jobId: storedJob.id, storedJob, logFile }); + try { + await runTrackedJob( + { + ...storedJob, + workspaceRoot, + logFile + }, + () => + executeTaskRun({ + ...request, + onProgress: progress + }), + { logFile } + ); + } finally { + releaseTtl(); + } +} + +/** + * Bound how long a detached worker may live. + * + * The worker is deliberately detached so a background task survives the session that queued it, + * and its immediate parent exits right after enqueue — so there is no parent to watch and nothing + * else that ever reclaims it. Without a ceiling a single wedged task keeps its whole process tree + * (app-server plus every MCP server under it) alive indefinitely. + * + * Returns a function that disarms the timer once the job finishes normally. + */ +/** How long the tree gets to leave on SIGTERM before the group is killed outright. */ +const WORKER_TERMINATION_GRACE_MS = 5000; + +function armWorkerTtl({ workspaceRoot, jobId, storedJob, logFile }) { + const ttlMs = workerTtlMs(); + const timer = armTimeout(ttlMs, () => { + const errorMessage = `Worker exceeded its ${ttlMs}ms lifetime.`; + + // Record the outcome before terminating: the process group is about to take this process down + // with it, and a job left at "running" with a dead pid is exactly the stale record that makes + // leaked workers invisible. All of it is best effort — a full disk or a deleted state + // directory must not be what keeps a runaway tree alive. + const completedAt = nowIso(); + const terminal = { + status: "failed", + phase: "failed", + pid: null, + completedAt, + errorMessage + }; + + // Each of these is best effort on its own. Sharing one try means a missing log directory or a + // full disk would skip the terminal status too, leaving the job "running" behind a dead pid — + // the stale record that hides leaked workers in the first place. Status goes first, because it + // is the part anything else reads. + const attempt = (action) => { + try { + action(); + } catch { + // Never let bookkeeping keep a runaway tree alive. + } + }; + + const recordExpiry = () => { + attempt(() => { + // Re-read rather than reusing the snapshot this timer closed over a day ago: it predates + // startedAt, threadId, turnId and every progress update since. + const current = readStoredJob(workspaceRoot, jobId) ?? storedJob; + writeJobFile(workspaceRoot, jobId, { ...current, ...terminal, logFile }); + }); + attempt(() => upsertJob(workspaceRoot, { id: jobId, ...terminal })); + }; + + recordExpiry(); + attempt(() => appendLogLine(logFile, `${errorMessage} Terminating its process tree.`)); + + // Terminate the tree rather than just this process: the app-server and MCP servers underneath + // are the expensive part, and they do not exit on their own. + // + // SIGTERM alone is not a ceiling — a descendant that traps or ignores it keeps running, and + // once this worker is gone nothing is left to escalate. So the worker survives its own signal + // for a grace period. That means the job can finish during it and record a success over the + // expiry, which would be a lie: its tree is about to be killed. Writing the expiry again as + // the last act before the kill makes it the outcome that stands. + terminateProcessTreeAndExit(process.pid, { + graceMs: WORKER_TERMINATION_GRACE_MS, + beforeKill: recordExpiry + }); + }); + return () => disarmTimeout(timer); } async function handleStatus(argv) { diff --git a/plugins/codex/scripts/lib/app-server.mjs b/plugins/codex/scripts/lib/app-server.mjs index 72b30a764..d85bd6d23 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); @@ -22,6 +22,9 @@ const PLUGIN_MANIFEST = JSON.parse(fs.readFileSync(PLUGIN_MANIFEST_URL, "utf8")) export const BROKER_ENDPOINT_ENV = "CODEX_COMPANION_APP_SERVER_ENDPOINT"; export const BROKER_BUSY_RPC_CODE = -32001; +/** How long a failed connect waits for its transport to report an exit before killing it. */ +const CONNECT_CLEANUP_GRACE_MS = 5000; + /** @type {ClientInfo} */ const DEFAULT_CLIENT_INFO = { title: "Codex Plugin", @@ -87,6 +90,12 @@ class AppServerClientBase { if (this.closed) { throw new Error("codex app-server client is closed."); } + // `closed` only covers a close we asked for. The transport can die on its own — between a + // successful connect and the very next request, for instance — and a request registered after + // that never resolves, because the exit that would reject it has already been reported. + if (this.exitResolved) { + throw this.exitError ?? new Error("codex app-server connection closed."); + } const id = this.nextId; this.nextId += 1; @@ -98,7 +107,7 @@ class AppServerClientBase { } notify(method, params = {}) { - if (this.closed) { + if (this.closed || this.exitResolved) { return; } this.sendMessage({ method, params }); @@ -175,6 +184,20 @@ class AppServerClientBase { this.resolveExit(undefined); } + /** + * Settle the exit state when initialization failed before a transport existed. + * + * `close()` ends by awaiting `exitPromise`, and only a live transport ever resolves it. But + * initialization can fail before one is created — a malformed endpoint rejects while being + * parsed, a spawn can throw — and closing then waits for an exit that nothing will report, + * turning a configuration error into a hang. + */ + settleExitIfNoTransport(hasTransport) { + if (!hasTransport) { + this.handleExit(this.exitError); + } + } + sendMessage(_message) { throw new Error("sendMessage must be implemented by subclasses."); } @@ -262,6 +285,7 @@ class SpawnedCodexAppServerClient extends AppServerClientBase { }, 50).unref?.(); } + this.settleExitIfNoTransport(Boolean(this.proc)); await this.exitPromise; } @@ -319,6 +343,7 @@ class BrokerCodexAppServerClient extends AppServerClientBase { if (this.socket) { this.socket.end(); } + this.settleExitIfNoTransport(Boolean(this.socket)); await this.exitPromise; } @@ -338,7 +363,22 @@ export class CodexAppServerClient { if (!options.disableBroker) { brokerEndpoint = options.brokerEndpoint ?? options.env?.[BROKER_ENDPOINT_ENV] ?? process.env[BROKER_ENDPOINT_ENV] ?? null; if (!brokerEndpoint && options.reuseExistingBroker) { - brokerEndpoint = loadBrokerSession(cwd)?.endpoint ?? null; + // Probe before trusting the record. A broker that was killed rather than shut down leaves + // its session behind, and connecting to that endpoint surfaces ENOENT/ECONNREFUSED as a + // failure of whatever the caller was doing — an authentication check, most visibly — + // rather than as a broker that is simply gone. Discard it and fall through to spawning. + // Probe before trusting the record. A broker that was killed rather than shut down leaves + // its session behind, and connecting to that endpoint surfaces ENOENT/ECONNREFUSED as a + // failure of whatever the caller was doing — an authentication check, most visibly — + // rather than as a broker that is simply gone. Fall through to spawning instead. + // + // The record itself is left alone: `status` and `setup` report a recorded shared runtime + // whether or not it answers, and reclaiming a dead broker's files belongs to + // `ensureBrokerSession`, which is the path that actually replaces it. + const persisted = loadBrokerSession(cwd)?.endpoint ?? null; + if (persisted && (await isBrokerEndpointReady(persisted))) { + brokerEndpoint = persisted; + } } if (!brokerEndpoint && !options.reuseExistingBroker) { const brokerSession = await ensureBrokerSession(cwd, { env: options.env }); @@ -348,7 +388,31 @@ export class CodexAppServerClient { const client = brokerEndpoint ? new BrokerCodexAppServerClient(cwd, { ...options, brokerEndpoint }) : new SpawnedCodexAppServerClient(cwd, options); - await client.initialize(); + try { + await client.initialize(); + } catch (error) { + // initialize() has usually already spawned the app-server, and with it every configured MCP + // server. The caller never receives this object, so this is the only chance to reclaim them. + // + // Bounded, because close() waits on the transport reporting its exit: an app-server that + // answered with an error but ignores EOF and SIGTERM would otherwise swallow the original + // failure entirely and leave the caller waiting forever. Whatever is still up after the + // grace gets killed outright. + await Promise.race([ + client.close().catch(() => {}), + new Promise((resolve) => { + setTimeout(resolve, CONNECT_CLEANUP_GRACE_MS).unref?.(); + }) + ]); + if (client.proc && client.proc.exitCode === null && !client.proc.killed) { + try { + client.proc.kill("SIGKILL"); + } catch { + // Already gone. + } + } + throw error; + } return client; } } diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index ef763819c..43d9b602c 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -6,6 +6,7 @@ import process from "node:process"; import { spawn } from "node:child_process"; import { fileURLToPath } from "node:url"; import { createBrokerEndpoint, parseBrokerEndpoint } from "./broker-endpoint.mjs"; +import { isProcessAlive } from "./process.mjs"; import { resolveStateDir } from "./state.mjs"; export const PID_FILE_ENV = "CODEX_COMPANION_APP_SERVER_PID_FILE"; @@ -58,7 +59,10 @@ export async function sendBrokerShutdown(endpoint) { export function spawnBrokerProcess({ scriptPath, cwd, endpoint, pidFile, logFile, env = process.env }) { const logFd = fs.openSync(logFile, "a"); - const child = spawn(process.execPath, [scriptPath, "serve", "--endpoint", endpoint, "--cwd", cwd, "--pid-file", pidFile], { + // The broker is told its own log path, not just handed the descriptor: it has to be able to + // clean up after itself without consulting the shared session record, which may by then name a + // successor. + const child = spawn(process.execPath, [scriptPath, "serve", "--endpoint", endpoint, "--cwd", cwd, "--pid-file", pidFile, "--log-file", logFile], { cwd, env, detached: true, @@ -99,7 +103,8 @@ export function clearBrokerSession(cwd) { } } -async function isBrokerEndpointReady(endpoint) { +/** Whether something is actually listening on a persisted endpoint, as opposed to merely recorded. */ +export async function isBrokerEndpointReady(endpoint) { if (!endpoint) { return false; } @@ -116,7 +121,14 @@ export async function ensureBrokerSession(cwd, options = {}) { return existing; } - if (existing) { + // Only reclaim a broker we can prove is gone. The probe above waits 150ms, which a live but busy + // broker can miss, and on the normal path `killProcess` is null — so tearing down here would + // delete a running broker's socket without stopping it, leaving it unreachable while it still + // holds its app-server and every MCP server underneath. + // + // A live one is left exactly as it is. Once the replacement below takes over it has no clients, + // so its own idle shutdown reclaims both the process and its files. + if (existing && !isProcessAlive(existing.pid)) { teardownBrokerSession({ endpoint: existing.endpoint ?? null, pidFile: existing.pidFile ?? null, diff --git a/plugins/codex/scripts/lib/lifecycle-limits.mjs b/plugins/codex/scripts/lib/lifecycle-limits.mjs new file mode 100644 index 000000000..65fd9cd8e --- /dev/null +++ b/plugins/codex/scripts/lib/lifecycle-limits.mjs @@ -0,0 +1,81 @@ +import process from "node:process"; + +const DEFAULT_BROKER_IDLE_SHUTDOWN_MS = 10 * 60 * 1000; +const DEFAULT_BROKER_STARTUP_TIMEOUT_MS = 5 * 60 * 1000; +const DEFAULT_WORKER_TTL_MS = 24 * 60 * 60 * 1000; + +/** `setTimeout` truncates anything larger to a 32-bit int, firing almost immediately instead. */ +const MAX_TIMEOUT_MS = 2 ** 31 - 1; + +function readDurationMs(raw, fallback) { + if (typeof raw !== "string" || raw.trim() === "") { + return fallback; + } + const parsed = Number(raw.trim()); + if (!Number.isFinite(parsed) || parsed < 0) { + return fallback; + } + // Clamp rather than reject: an operator asking for 30 days means "effectively never", and + // letting that overflow into ~1ms would kill exactly what they meant to protect. + return Math.min(parsed, MAX_TIMEOUT_MS); +} + +/** + * Schedule `onExpiry`, treating `0` as "no limit". + * + * Every limit here documents `0` as disabled, but `setTimeout(fn, 0)` means "next tick" — passing + * a disabled limit straight through would fire the guard immediately instead of never. Returns + * the timer, or `null` when disabled; pass that back to {@link disarmTimeout}. + */ +export function armTimeout(ms, onExpiry) { + if (!ms) { + return null; + } + const timer = setTimeout(onExpiry, ms); + // Never hold the event loop open on the guard alone. + timer.unref?.(); + return timer; +} + +/** Counterpart to {@link armTimeout}; tolerates the `null` a disabled limit produces. */ +export function disarmTimeout(timer) { + if (timer) { + clearTimeout(timer); + } +} + +/** + * How long the broker may sit with no connected client before shutting itself down, taking its + * app-server (and every MCP server under it) with it. + * + * A broker outlives the client that spawned it, so without this it survives a crashed or + * timed-out SessionEnd hook and holds that whole tree alive indefinitely. `0` disables the timer. + */ +export function brokerIdleShutdownMs(env = process.env) { + return readDurationMs(env.CODEX_BROKER_IDLE_SHUTDOWN_MS, DEFAULT_BROKER_IDLE_SHUTDOWN_MS); +} + +/** + * How long the broker may spend starting up before it gives up and takes its tree down. + * + * Until the broker is listening there is no idle timer, and the client that spawned it stops + * waiting after a couple of seconds without killing anything — so a wedged app-server or MCP + * startup would strand the whole tree permanently. The default is generous because a cold start + * legitimately spawns every configured MCP server. `0` disables the bound. + */ +export function brokerStartupTimeoutMs(env = process.env) { + return readDurationMs(env.CODEX_BROKER_STARTUP_TIMEOUT_MS, DEFAULT_BROKER_STARTUP_TIMEOUT_MS); +} + +/** + * Ceiling on a detached background worker's wall-clock lifetime. + * + * The worker is deliberately detached so a background task survives the session that queued it, + * and its immediate parent exits right after enqueue — so there is no parent to watch and nothing + * else that reclaims it. This is a runaway guard, not a task deadline: the default is far longer + * than any real background task, and still reclaims the multi-day trees that prompted it. `0` + * disables the ceiling. + */ +export function workerTtlMs(env = process.env) { + return readDurationMs(env.CODEX_TASK_WORKER_TTL_MS, DEFAULT_WORKER_TTL_MS); +} diff --git a/plugins/codex/scripts/lib/process.mjs b/plugins/codex/scripts/lib/process.mjs index dd8fc3751..d3d7346ac 100644 --- a/plugins/codex/scripts/lib/process.mjs +++ b/plugins/codex/scripts/lib/process.mjs @@ -54,6 +54,59 @@ function looksLikeMissingProcessMessage(text) { return /not found|no running instance|cannot find|does not exist|no such process/i.test(text); } +/** + * Whether a pid still names a running process. + * + * Signal `0` runs the existence and permission checks without delivering anything. `EPERM` means + * the process is there but owned by someone else, which still counts as alive. + */ +export function isProcessAlive(pid, killImpl = process.kill.bind(process)) { + if (!Number.isFinite(pid) || pid <= 0) { + return false; + } + try { + killImpl(pid, 0); + return true; + } catch (error) { + return error?.code === "EPERM"; + } +} + +/** + * Terminate a process tree and make sure it is actually gone. + * + * `terminateProcessTree` sends SIGTERM and stops there, so a descendant that traps or ignores it + * survives — and when the tree being terminated is our own, we die with the signal and nothing is + * left to escalate. This keeps the caller alive through its own SIGTERM, gives the tree a grace + * period to leave on its own, then kills what remains and exits. + * + * Only meaningful for a group leader; a caller that is not one signals nothing, which is the + * existing behaviour of `terminateProcessTree`. + */ +export function terminateProcessTreeAndExit(pid, { graceMs = 5000, exitCode = 1, beforeKill } = {}) { + if (pid === process.pid) { + process.on("SIGTERM", () => {}); + } + terminateProcessTree(pid); + // Deliberately not unref'd: this timer is the escalation, and the process must stay up for it. + setTimeout(() => { + // Surviving our own SIGTERM means ordinary work keeps running during the grace period and can + // record an outcome of its own. This is the last moment before the group dies, so a caller + // that needs the final say gets it here. + try { + beforeKill?.(); + } catch { + // Never let bookkeeping stop the kill. + } + try { + process.kill(-pid, "SIGKILL"); + } catch { + // Nothing left in the group, or the caller was never its leader. + } + process.exit(exitCode); + }, graceMs); +} + export function terminateProcessTree(pid, options = {}) { if (!Number.isFinite(pid)) { return { attempted: false, delivered: false, method: null }; diff --git a/tests/app-server-connect.test.mjs b/tests/app-server-connect.test.mjs new file mode 100644 index 000000000..a3c730647 --- /dev/null +++ b/tests/app-server-connect.test.mjs @@ -0,0 +1,40 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; + +/** Reject rather than hang, so a wedged connect fails the test instead of stalling the run. */ +function withDeadline(promise, ms, label) { + let timer; + const deadline = new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(`${label} did not settle within ${ms}ms`)), ms); + }); + return Promise.race([promise, deadline]).finally(() => clearTimeout(timer)); +} + +test("connect reports a malformed broker endpoint instead of hanging", async () => { + // parseBrokerEndpoint throws before a socket exists, so cleanup has no transport to wait on. + // Closing then awaited an exit nothing would ever report, and connect() never settled — a + // configuration mistake turned into a hang. + await assert.rejects( + () => + withDeadline( + CodexAppServerClient.connect(process.cwd(), { brokerEndpoint: "not-an-endpoint" }), + 5000, + "connect" + ), + /Unsupported broker endpoint/ + ); +}); + +test("connect reports an empty broker endpoint instead of hanging", async () => { + await assert.rejects( + () => + withDeadline( + CodexAppServerClient.connect(process.cwd(), { brokerEndpoint: "unix:" }), + 5000, + "connect" + ), + /missing its path/ + ); +}); diff --git a/tests/broker-reclaim.test.mjs b/tests/broker-reclaim.test.mjs new file mode 100644 index 000000000..a1c522bc2 --- /dev/null +++ b/tests/broker-reclaim.test.mjs @@ -0,0 +1,89 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { + ensureBrokerSession, + loadBrokerSession, + saveBrokerSession, + spawnBrokerProcess +} from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; + +/** A broker session whose files exist on disk, so we can see whether they survive. */ +function plantSession(pid) { + const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "cxc-cwd-")); + const sessionDir = fs.mkdtempSync(path.join(os.tmpdir(), "cxc-session-")); + const pidFile = path.join(sessionDir, "broker.pid"); + const logFile = path.join(sessionDir, "broker.log"); + const socketPath = path.join(sessionDir, "broker.sock"); + fs.writeFileSync(pidFile, `${pid}\n`); + fs.writeFileSync(logFile, "log\n"); + fs.writeFileSync(socketPath, ""); + const session = { endpoint: `unix:${socketPath}`, pidFile, logFile, sessionDir, pid }; + saveBrokerSession(cwd, session); + return { cwd, session, socketPath }; +} + +// The endpoint never answers, so the readiness probe fails either way; what differs is whether the +// process behind the record is still alive. +const nowhere = { scriptPath: path.join(os.tmpdir(), "cxc-does-not-exist.mjs"), timeoutMs: 50 }; + +test("a broker that is merely unresponsive keeps its files", async () => { + // process.pid is unmistakably alive. Tearing this one down would delete a running broker's + // socket without stopping it — it would keep its app-server and every MCP server underneath, + // now unreachable and untracked. + const { cwd, session, socketPath } = plantSession(process.pid); + + await ensureBrokerSession(cwd, nowhere).catch(() => {}); + + assert.equal(fs.existsSync(session.pidFile), true, "pid file was removed"); + assert.equal(fs.existsSync(session.logFile), true, "log was removed"); + assert.equal(fs.existsSync(socketPath), true, "socket was removed"); + + fs.rmSync(session.sessionDir, { recursive: true, force: true }); + fs.rmSync(cwd, { recursive: true, force: true }); +}); + +test("a broker that is gone is reclaimed", async () => { + // A pid that cannot be running: process 0 is never a live user process, so this stands in for a + // broker killed outright. + const { cwd, session, socketPath } = plantSession(0); + + await ensureBrokerSession(cwd, nowhere).catch(() => {}); + + assert.equal(fs.existsSync(session.pidFile), false, "pid file survived"); + assert.equal(fs.existsSync(session.logFile), false, "log survived"); + assert.equal(fs.existsSync(socketPath), false, "socket survived"); + assert.equal(loadBrokerSession(cwd), null, "record survived"); + + fs.rmSync(session.sessionDir, { recursive: true, force: true }); + fs.rmSync(cwd, { recursive: true, force: true }); +}); + +test("a broker is told its own log path", async () => { + // Its record may name a successor by the time it shuts down, so it cannot rely on that to find + // its own artifacts — it has to be given them at startup. + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "cxc-args-")); + const argvFile = path.join(dir, "argv.json"); + const scriptPath = path.join(dir, "fake-broker.mjs"); + fs.writeFileSync( + scriptPath, + `import fs from "node:fs";\nfs.writeFileSync(${JSON.stringify(argvFile)}, JSON.stringify(process.argv.slice(2)));\n` + ); + const logFile = path.join(dir, "broker.log"); + + const child = spawnBrokerProcess({ + scriptPath, + cwd: dir, + endpoint: `unix:${path.join(dir, "broker.sock")}`, + pidFile: path.join(dir, "broker.pid"), + logFile + }); + await new Promise((resolve) => child.on("exit", resolve)); + + const argv = JSON.parse(fs.readFileSync(argvFile, "utf8")); + assert.equal(argv[argv.indexOf("--log-file") + 1], logFile); + fs.rmSync(dir, { recursive: true, force: true }); +}); diff --git a/tests/lifecycle-limits.test.mjs b/tests/lifecycle-limits.test.mjs new file mode 100644 index 000000000..0aae4ceb4 --- /dev/null +++ b/tests/lifecycle-limits.test.mjs @@ -0,0 +1,110 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { + armTimeout, + brokerIdleShutdownMs, + brokerStartupTimeoutMs, + disarmTimeout, + workerTtlMs +} from "../plugins/codex/scripts/lib/lifecycle-limits.mjs"; +import { isProcessAlive } from "../plugins/codex/scripts/lib/process.mjs"; + +const TEN_MINUTES = 10 * 60 * 1000; +const FIVE_MINUTES = 5 * 60 * 1000; +const ONE_DAY = 24 * 60 * 60 * 1000; +const MAX_TIMEOUT_MS = 2 ** 31 - 1; + +test("broker idle shutdown defaults to ten minutes", () => { + assert.equal(brokerIdleShutdownMs({}), TEN_MINUTES); + assert.equal(brokerIdleShutdownMs({ CODEX_BROKER_IDLE_SHUTDOWN_MS: "" }), TEN_MINUTES); +}); + +test("broker idle shutdown honours an override and can be disabled", () => { + assert.equal(brokerIdleShutdownMs({ CODEX_BROKER_IDLE_SHUTDOWN_MS: "5000" }), 5000); + assert.equal(brokerIdleShutdownMs({ CODEX_BROKER_IDLE_SHUTDOWN_MS: "0" }), 0); +}); + +test("broker idle shutdown falls back to the default on unusable input", () => { + for (const raw of ["abc", "-1", "NaN", " "]) { + assert.equal(brokerIdleShutdownMs({ CODEX_BROKER_IDLE_SHUTDOWN_MS: raw }), TEN_MINUTES); + } +}); + +test("broker startup timeout defaults to five minutes and is overridable", () => { + assert.equal(brokerStartupTimeoutMs({}), FIVE_MINUTES); + assert.equal(brokerStartupTimeoutMs({ CODEX_BROKER_STARTUP_TIMEOUT_MS: "1500" }), 1500); + assert.equal(brokerStartupTimeoutMs({ CODEX_BROKER_STARTUP_TIMEOUT_MS: "0" }), 0); +}); + +test("worker ttl defaults to a day", () => { + assert.equal(workerTtlMs({}), ONE_DAY); + assert.equal(workerTtlMs({ CODEX_TASK_WORKER_TTL_MS: "" }), ONE_DAY); +}); + +test("worker ttl honours an override and can be disabled", () => { + assert.equal(workerTtlMs({ CODEX_TASK_WORKER_TTL_MS: "1000" }), 1000); + assert.equal(workerTtlMs({ CODEX_TASK_WORKER_TTL_MS: "0" }), 0); +}); + +test("worker ttl falls back to the default on unusable input", () => { + for (const raw of ["soon", "-5", "NaN", " "]) { + assert.equal(workerTtlMs({ CODEX_TASK_WORKER_TTL_MS: raw }), ONE_DAY); + } +}); + +test("a disabled limit never fires", async () => { + // setTimeout(fn, 0) means "next tick", but 0 is our documented way to disable a limit. Passing + // it straight through would turn "no ceiling" into "terminate immediately". + let fired = false; + const timer = armTimeout(0, () => { + fired = true; + }); + assert.equal(timer, null); + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.equal(fired, false); + disarmTimeout(timer); // must tolerate the null a disabled limit produces +}); + +test("an enabled limit fires and can be disarmed", async () => { + let fired = false; + const timer = armTimeout(5, () => { + fired = true; + }); + assert.notEqual(timer, null); + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.equal(fired, true); + + let second = false; + disarmTimeout(armTimeout(5, () => { + second = true; + })); + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.equal(second, false); +}); + +test("liveness treats a permission error as alive and a missing pid as gone", () => { + // A broker owned by another user still holds its socket; treating EPERM as "gone" would delete + // a running broker's files. + const eperm = () => { + throw Object.assign(new Error("operation not permitted"), { code: "EPERM" }); + }; + const esrch = () => { + throw Object.assign(new Error("no such process"), { code: "ESRCH" }); + }; + assert.equal(isProcessAlive(1234, eperm), true); + assert.equal(isProcessAlive(1234, esrch), false); + assert.equal(isProcessAlive(1234, () => {}), true); + for (const bad of [null, undefined, 0, -1, Number.NaN, "123"]) { + assert.equal(isProcessAlive(bad, () => {}), false); + } + assert.equal(isProcessAlive(process.pid), true); +}); + +test("durations are clamped to what setTimeout can actually hold", () => { + // 30 days would truncate to a 32-bit int and fire almost immediately — killing precisely the + // long-running work the operator meant to protect. + const thirtyDays = String(30 * 24 * 60 * 60 * 1000); + assert.equal(workerTtlMs({ CODEX_TASK_WORKER_TTL_MS: thirtyDays }), MAX_TIMEOUT_MS); + assert.equal(brokerIdleShutdownMs({ CODEX_BROKER_IDLE_SHUTDOWN_MS: thirtyDays }), MAX_TIMEOUT_MS); +});