diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 1954274fe..8e53e2335 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -2,12 +2,15 @@ import fs from "node:fs"; import net from "node:net"; +import os from "node:os"; import path from "node:path"; 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 } from "./lib/broker-lifecycle.mjs"; +import { terminateProcessTree } from "./lib/process.mjs"; const STREAMING_METHODS = new Set(["turn/start", "review/start", "thread/compact/start"]); @@ -48,11 +51,12 @@ 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 ] [--managed-session-dir]"); } const { options } = parseArgs(argv, { - valueOptions: ["cwd", "pid-file", "endpoint"] + valueOptions: ["cwd", "pid-file", "endpoint"], + booleanOptions: ["managed-session-dir"] }); if (!options.endpoint) { @@ -63,14 +67,64 @@ async function main() { const endpoint = String(options.endpoint); const listenTarget = parseBrokerEndpoint(endpoint); const pidFile = options["pid-file"] ? path.resolve(options["pid-file"]) : null; + const managedSessionDir = options["managed-session-dir"] === true; writePidFile(pidFile); const appClient = await CodexAppServerClient.connect(cwd, { disableBroker: true }); let activeRequestSocket = null; let activeStreamSocket = null; let activeStreamThreadIds = null; + let inFlightRequests = 0; + let shuttingDown = false; + let shutdownPromise = null; + let serverRef = null; const sockets = new Set(); + // Bound each forwarded request so a hung app server cannot pin + // inFlightRequests forever, which would permanently disarm idle + // self-shutdown and make the broker unkillable by anything but a signal. + // Requests resolve at acceptance (streams ride notifications), so the + // default is generous; <= 0 disables the bound. + const requestTimeoutRaw = (process.env.CODEX_BROKER_REQUEST_TIMEOUT_MS ?? "").trim(); + const requestTimeoutMs = /^-?\d+$/.test(requestTimeoutRaw) + ? Math.min(Number(requestTimeoutRaw), 2 ** 31 - 1) + : 10 * 60 * 1000; + + // Forward a request to the app server while counting it as in-flight, so idle + // self-shutdown can never fire while real work is running, even if the calling + // client disconnected mid-request (which clears activeRequestSocket). + async function forwardAppRequest(method, params) { + inFlightRequests += 1; + try { + const request = appClient.request(method, params); + if (requestTimeoutMs <= 0) { + return await request; + } + return await Promise.race([ + request, + new Promise((_, reject) => { + const timer = setTimeout(() => { + // The race cannot cancel the underlying request: releasing + // ownership while it might still execute would let a clientless + // turn keep running (and route its notifications to a later + // client). A request unanswered for this long means the app + // server is wedged, so terminate the whole broker instead — + // shutdown closes the app-server child with it, and callers + // respawn a fresh broker on demand. + reject(new Error(`Shared broker request ${method} timed out after ${requestTimeoutMs}ms; broker shutting down.`)); + setTimeout(() => process.exit(1), 5000).unref(); // backstop if cleanup hangs + shutdown(serverRef).finally(() => process.exit(1)); + }, requestTimeoutMs); + timer.unref(); + request.finally(() => clearTimeout(timer)).catch(() => {}); + }) + ]); + } finally { + inFlightRequests -= 1; + scheduleIdleShutdown(); + } + } + function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { activeRequestSocket = null; @@ -95,27 +149,169 @@ async function main() { if (activeRequestSocket === target) { activeRequestSocket = null; } + // Releasing stream ownership can be the last activity on this broker; + // without rearming here an abandoned cwd would never become idle. + scheduleIdleShutdown(); } } } - async function shutdown(server) { + // Every shutdown path shares one promise: a second caller (e.g. SIGTERM + // landing during an idle shutdown) awaits the same cleanup instead of + // returning early and letting its process.exit() abort the first caller's + // cleanup mid-flight. + function shutdown(server) { + if (!shutdownPromise) { + shuttingDown = true; + shutdownPromise = performShutdown(server); + } + return shutdownPromise; + } + + async function performShutdown(server) { + // Whole-shutdown backstop: whatever below wedges, this process ends. The + // state record is cleared first, so a replacement broker is never blocked + // on this one finishing its cleanup. + setTimeout(() => process.exit(1), 15000).unref(); + // Retire this broker's state record first, while its socket is still the + // live one for this cwd: no replacement broker can have been spawned yet, + // so the guarded clear cannot race a newer record, and clients probing + // from here on fall back to starting a fresh broker instead of connecting + // to a dying one. Guarded on the endpoint matching, so a record that was + // already replaced (e.g. this broker was deemed unresponsive) is kept. + try { + if (loadBrokerSession(cwd)?.endpoint === endpoint) { + clearBrokerSession(cwd); + } + } catch { + // Ignore unreadable or already-removed state records. + } 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); + // Bound the app-server close: close() escalates only as far as SIGTERM, + // so a child that ignores it would wedge this shutdown indefinitely + // (record already cleared, session dir still present, both processes + // alive while the next command spawns a replacement). After the deadline, + // deliver a real force-kill; the child's exit settles the dangling close. + // On POSIX that must be SIGKILL to the pid itself: the child is not + // detached, so a process-group signal (kill(-pid)) hits ESRCH and + // terminateProcessTree delivers nothing there. On Windows the child is a + // cmd.exe wrapper, so the taskkill /T /F tree kill is the right tool. + await Promise.race([ + appClient.close().catch(() => {}), + new Promise((resolve) => { + const timer = setTimeout(() => { + try { + if (appClient.proc && appClient.proc.exitCode === null) { + if (process.platform === "win32") { + terminateProcessTree(appClient.proc.pid); + } else { + appClient.proc.kill("SIGKILL"); + } + } + } catch { + // Best effort; the whole-shutdown backstop above still applies. + } + resolve(); + }, 5000); + timer.unref(); + }) + ]); + if (server) { + await new Promise((resolve) => server.close(resolve)); } - if (pidFile && fs.existsSync(pidFile)) { - fs.unlinkSync(pidFile); + // Remove the broker's own session directory (socket, pid file, log) so a + // clean exit leaves nothing behind for the reaper to GC. Recursive removal + // requires the spawner to have declared it created the directory + // (--managed-session-dir, set by spawnBrokerProcess after mkdtemp) AND the + // plugin's own layout (cxc- prefix directly under the OS temp dir), so a + // manual invocation can never have a caller-selected directory deleted, + // even one named to look like ours; only the broker's own files are + // unlinked there. + const sessionDir = pidFile + ? path.dirname(pidFile) + : listenTarget.kind === "unix" + ? path.dirname(listenTarget.path) + : null; + try { + if (sessionDir && isManagedSessionDir(sessionDir)) { + fs.rmSync(sessionDir, { recursive: true, force: true }); + } else { + if (pidFile && fs.existsSync(pidFile)) { + fs.unlinkSync(pidFile); + } + if (listenTarget.kind === "unix" && fs.existsSync(listenTarget.path)) { + fs.unlinkSync(listenTarget.path); + } + } + } catch { + // Ignore already-removed files or directories. + } + } + + function isManagedSessionDir(dir) { + if (!managedSessionDir) { + return false; + } + try { + return ( + path.basename(dir).startsWith("cxc-") && + fs.realpathSync(path.dirname(dir)) === fs.realpathSync(os.tmpdir()) + ); + } catch { + return false; } } appClient.setNotificationHandler(routeNotification); + // Idle self-shutdown: a broker is spawned per working directory and is reused + // across sessions, so no external actor can safely decide it is done. Instead + // the broker exits itself once it has had no connections and no in-flight work + // for CODEX_BROKER_IDLE_MS (default 30 min; <= 0 disables). Callers respawn one + // on demand, so exiting when idle is safe and stops brokers from accumulating. + // Strict integer parsing: parseInt would truncate "30m" to 30ms and accept + // scientific notation, silently inverting an "effectively never" intent into + // near-instant shutdown. Malformed values fall back to the default, and the + // value is clamped below Node's 2^31-1 setTimeout ceiling (beyond it the + // timer fires after 1ms). + const idleRaw = (process.env.CODEX_BROKER_IDLE_MS ?? "").trim(); + const idleTimeoutMs = /^-?\d+$/.test(idleRaw) + ? Math.min(Number(idleRaw), 2 ** 31 - 1) + : 30 * 60 * 1000; + let idleTimer = null; + function cancelIdleShutdown() { + if (idleTimer) { + clearTimeout(idleTimer); + idleTimer = null; + } + } + function isIdle() { + return sockets.size === 0 && inFlightRequests === 0 && !activeRequestSocket && !activeStreamSocket; + } + function scheduleIdleShutdown() { + cancelIdleShutdown(); + if (idleTimeoutMs <= 0 || !isIdle()) { + return; + } + idleTimer = setTimeout(() => { + if (isIdle()) { + shutdown(server).finally(() => process.exit(0)); + } + }, idleTimeoutMs); + idleTimer.unref(); // never keep the process alive solely to fire this timer + } + const server = net.createServer((socket) => { + if (shuttingDown) { + // A connection that lands between shutdown starting and server.close() + // taking effect would otherwise hold the close (and the process exit) + // open until the client goes away on its own. + socket.destroy(); + return; + } + cancelIdleShutdown(); sockets.add(socket); socket.setEncoding("utf8"); let buffer = ""; @@ -183,7 +379,7 @@ async function main() { if (allowInterruptDuringActiveStream) { try { - const result = await appClient.request(message.method, message.params ?? {}); + const result = await forwardAppRequest(message.method, message.params ?? {}); send(socket, { id: message.id, result }); } catch (error) { send(socket, { @@ -198,9 +394,13 @@ async function main() { activeRequestSocket = socket; try { - const result = await appClient.request(message.method, message.params ?? {}); + const result = await forwardAppRequest(message.method, message.params ?? {}); send(socket, { id: message.id, result }); - if (isStreaming) { + // A socket that disconnected during the await already had its + // ownership cleared by the close handler; assigning it here would + // strand a dead socket in activeStreamSocket, busy-rejecting other + // clients and keeping isIdle() false with no event left to clear it. + if (isStreaming && !socket.destroyed) { activeStreamSocket = socket; activeStreamThreadIds = buildStreamThreadIds(message.method, message.params ?? {}, result); } @@ -225,14 +425,30 @@ async function main() { socket.on("close", () => { sockets.delete(socket); clearSocketOwnership(socket); + scheduleIdleShutdown(); }); socket.on("error", () => { sockets.delete(socket); clearSocketOwnership(socket); + scheduleIdleShutdown(); }); }); + serverRef = server; + + // If the codex app-server child exits or its connection is lost, this broker + // can never serve another request, but its socket keeps accepting: endpoint + // probes pass, the reaper skips the live pid, and nothing external kills + // brokers anymore. Exit instead; callers respawn a fresh broker on demand. + Promise.resolve(appClient.exitPromise) + .catch(() => {}) + .then(() => { + if (!shuttingDown) { + shutdown(server).finally(() => process.exit(1)); + } + }); + process.on("SIGTERM", async () => { await shutdown(server); process.exit(0); @@ -243,7 +459,9 @@ async function main() { process.exit(0); }); - server.listen(listenTarget.path); + server.listen(listenTarget.path, () => { + scheduleIdleShutdown(); // exit if nobody ever connects + }); } main().catch((error) => { diff --git a/plugins/codex/scripts/lib/app-server.mjs b/plugins/codex/scripts/lib/app-server.mjs index 72b30a764..111bd93ab 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, loadBrokerSession, waitForBrokerEndpoint } 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,14 @@ 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; + // Validate the recorded endpoint before reusing it: the record can be + // stale when the broker died without a clean exit. A stale record then + // behaves like no record, and the caller falls back to a direct app + // server instead of erroring against a dead socket. + const recordedEndpoint = loadBrokerSession(cwd)?.endpoint ?? null; + if (recordedEndpoint && (await waitForBrokerEndpoint(recordedEndpoint, 150).catch(() => false))) { + brokerEndpoint = recordedEndpoint; + } } 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 ef763819c..515f140f9 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -12,8 +12,21 @@ export const PID_FILE_ENV = "CODEX_COMPANION_APP_SERVER_PID_FILE"; export const LOG_FILE_ENV = "CODEX_COMPANION_APP_SERVER_LOG_FILE"; const BROKER_STATE_FILE = "broker.json"; +const MANAGED_MARKER_FILE = "broker.managed"; + export function createBrokerSessionDir(prefix = "cxc-") { - return fs.mkdtempSync(path.join(os.tmpdir(), prefix)); + const sessionDir = fs.mkdtempSync(path.join(os.tmpdir(), prefix)); + // Persisted ownership record: only directories this plugin created carry the + // marker, and only marked directories are ever removed recursively (by the + // reaper below; the broker's own shutdown gets the equivalent signal via + // --managed-session-dir). A caller-selected directory that merely looks like + // ours never gains the marker, so it is never deleted. + fs.writeFileSync( + path.join(sessionDir, MANAGED_MARKER_FILE), + "Created by the codex plugin (createBrokerSessionDir); safe to remove recursively.\n", + "utf8" + ); + return sessionDir; } function connectToEndpoint(endpoint) { @@ -58,7 +71,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], { + // --managed-session-dir: this spawner created the session directory + // (createBrokerSessionDir's mkdtemp), so the broker may remove the whole + // directory on clean exit. Manual invocations lack the flag and keep theirs. + const child = spawn(process.execPath, [scriptPath, "serve", "--endpoint", endpoint, "--cwd", cwd, "--pid-file", pidFile, "--managed-session-dir"], { cwd, env, detached: true, @@ -201,9 +217,87 @@ export function teardownBrokerSession({ endpoint = null, pidFile, logFile, sessi const resolvedSessionDir = sessionDir ?? (pidFile ? path.dirname(pidFile) : logFile ? path.dirname(logFile) : null); if (resolvedSessionDir && fs.existsSync(resolvedSessionDir)) { try { + const marker = path.join(resolvedSessionDir, MANAGED_MARKER_FILE); + if (fs.existsSync(marker)) { + fs.unlinkSync(marker); + } fs.rmdirSync(resolvedSessionDir); } catch { // Ignore non-empty or missing directories. } } } + +function readBrokerPid(sessionDir) { + try { + const pid = Number.parseInt(fs.readFileSync(path.join(sessionDir, "broker.pid"), "utf8").trim(), 10); + return Number.isInteger(pid) && pid > 1 ? pid : null; // reject 0/negative/NaN + } catch { + return null; + } +} + +export function isPidAlive(pid) { + if (!Number.isInteger(pid) || pid <= 1) { + return false; + } + try { + process.kill(pid, 0); // signal 0 only checks existence + return true; + } catch (error) { + return error?.code === "EPERM"; // exists but owned by another user + } +} + +// GC leaked broker session directories. A broker is spawned per working +// directory and reused across sessions, and it now exits itself once idle, +// removing its own directory (see app-server-broker.mjs). This only cleans up +// after a broker that died WITHOUT that clean exit (e.g. it was killed): its +// directory is left behind with a now-dead PID. Deletion requires the +// persisted ownership marker createBrokerSessionDir writes, so a directory +// this plugin did not create (a manual --pid-file location, however named) is +// never removed regardless of what its pid file says. A live PID is never +// inspected or signalled, so this can neither interrupt a session sharing a +// broker nor signal an unrelated process that reused a stale PID. And a +// marked directory whose broker.pid is missing, empty, or unparseable +// (possibly a torn write from a broker still starting up) is left alone +// rather than racing the writer; the cost is that a permanently corrupt pid +// file leaks its (tiny) directory, where the alternative was deleting a live +// broker's socket out from under it. +export async function reapBrokerSessions({ tmpDir = os.tmpdir() } = {}) { + let entries; + try { + entries = fs.readdirSync(tmpDir, { withFileTypes: true }); + } catch { + return; + } + + for (const entry of entries) { + if (!entry.isDirectory() || !entry.name.startsWith("cxc-")) { + continue; + } + const sessionDir = path.join(tmpDir, entry.name); + if (!fs.existsSync(path.join(sessionDir, MANAGED_MARKER_FILE))) { + continue; // no ownership marker: not created by this plugin, never delete + } + if (!fs.existsSync(path.join(sessionDir, "broker.pid"))) { + continue; // a broker removes its own dir on clean exit; nothing to do + } + + const pid = readBrokerPid(sessionDir); + if (pid === null) { + continue; // empty/unparseable pid file: possibly mid-write, leave it alone + } + if (isPidAlive(pid)) { + continue; // live broker: leave it entirely alone (it self-exits when idle) + } + + // The broker process is gone but left its directory behind (killed, not a + // clean exit). Remove the leftover; the PID is dead so nothing is signalled. + try { + fs.rmSync(sessionDir, { recursive: true, force: true }); + } catch { + // Ignore already-removed directories. + } + } +} diff --git a/plugins/codex/scripts/lib/codex.mjs b/plugins/codex/scripts/lib/codex.mjs index fead00cc4..ae4dd43d2 100644 --- a/plugins/codex/scripts/lib/codex.mjs +++ b/plugins/codex/scripts/lib/codex.mjs @@ -41,7 +41,8 @@ import path from "node:path"; import { readJsonFile } from "./fs.mjs"; import { BROKER_BUSY_RPC_CODE, BROKER_ENDPOINT_ENV, CodexAppServerClient } from "./app-server.mjs"; -import { loadBrokerSession } from "./broker-lifecycle.mjs"; +import { parseBrokerEndpoint } from "./broker-endpoint.mjs"; +import { isPidAlive, loadBrokerSession } from "./broker-lifecycle.mjs"; import { binaryAvailable } from "./process.mjs"; const SERVICE_NAME = "claude_code_codex_plugin"; @@ -612,16 +613,29 @@ async function captureTurn(client, threadId, startRequest, options = {}) { async function withAppServer(cwd, fn) { let client = null; + let connectFailed = false; try { - client = await CodexAppServerClient.connect(cwd); + try { + client = await CodexAppServerClient.connect(cwd); + } catch (error) { + connectFailed = true; + throw error; + } const result = await fn(client); await client.close(); return result; } catch (error) { - const brokerRequested = client?.transport === "broker" || Boolean(process.env[BROKER_ENDPOINT_ENV]); + // connect() itself throwing means fn never ran, so a direct retry is + // always safe regardless of the error's shape; this covers racing a + // broker's idle self-shutdown at every phase (socket gone: ENOENT or + // ECONNREFUSED; accepted then destroyed: ECONNRESET or a closed-connection + // error during initialize). Post-connect failures retry only on the + // broker-busy rejection, which the broker raises before any work runs; + // retrying fn after other mid-flight failures could replay side effects + // (e.g. a thread created before the failing request). const shouldRetryDirect = - (client?.transport === "broker" && error?.rpcCode === BROKER_BUSY_RPC_CODE) || - (brokerRequested && (error?.code === "ENOENT" || error?.code === "ECONNREFUSED")); + connectFailed || + (client?.transport === "broker" && error?.rpcCode === BROKER_BUSY_RPC_CODE); if (client) { await client.close().catch(() => {}); @@ -903,8 +917,46 @@ export function getCodexAvailability(cwd) { }; } +// Weak fallback liveness signal for endpoints with no recorded broker pid: a +// unix socket file that no longer exists cannot back a live broker (the broker +// removes it on clean shutdown; the session-start reaper removes it after an +// unclean death). Pipe endpoints (Windows) cannot be checked without +// connecting and are reported as-is. +function isBrokerEndpointPresent(endpoint) { + try { + const target = parseBrokerEndpoint(endpoint); + return target.kind === "unix" ? fs.existsSync(target.path) : true; + } catch { + return false; + } +} + +// A state record can outlive its broker when the process died without a clean +// exit (a crash leaves both broker.json and the socket file behind until a +// reaper runs). Prefer the recorded process identity: a dead pid means no +// shared runtime regardless of what is on disk. Records without a usable pid +// fall back to endpoint presence. +function isBrokerSessionLive(session) { + if (!session?.endpoint) { + return false; + } + if (Number.isInteger(session.pid) && session.pid > 1) { + return isPidAlive(session.pid); + } + return isBrokerEndpointPresent(session.endpoint); +} + export function getSessionRuntimeStatus(env = process.env, cwd = process.cwd()) { - const endpoint = env?.[BROKER_ENDPOINT_ENV] ?? loadBrokerSession(cwd)?.endpoint ?? null; + const envEndpoint = env?.[BROKER_ENDPOINT_ENV]; + let endpoint = null; + if (envEndpoint != null) { + // An env override takes precedence even when empty: an empty value masks + // the recorded session, matching the previous behavior. + endpoint = envEndpoint && isBrokerEndpointPresent(envEndpoint) ? envEndpoint : null; + } else { + const session = loadBrokerSession(cwd); + endpoint = session && isBrokerSessionLive(session) ? session.endpoint : null; + } if (endpoint) { return { mode: "shared", @@ -957,7 +1009,7 @@ export async function getCodexAuthStatus(cwd, options = {}) { } } -export async function interruptAppServerTurn(cwd, { threadId, turnId }) { +export async function interruptAppServerTurn(cwd, { threadId, turnId, skipAvailabilityCheck = false }) { if (!threadId || !turnId) { return { attempted: false, @@ -967,14 +1019,20 @@ export async function interruptAppServerTurn(cwd, { threadId, turnId }) { }; } - const availability = getCodexAvailability(cwd); - if (!availability.available) { - return { - attempted: false, - interrupted: false, - transport: null, - detail: availability.detail - }; + // The availability probe spawns codex synchronously (twice) and cannot be + // preempted by a caller's timeout race. A caller that has already verified a + // live broker endpoint (the SessionEnd hook) skips it: a responding broker + // proves the runtime exists. + if (!skipAvailabilityCheck) { + const availability = getCodexAvailability(cwd); + if (!availability.available) { + return { + attempted: false, + interrupted: false, + transport: null, + detail: availability.detail + }; + } } let client = null; diff --git a/plugins/codex/scripts/session-lifecycle-hook.mjs b/plugins/codex/scripts/session-lifecycle-hook.mjs index 778571e6c..490ff12fe 100644 --- a/plugins/codex/scripts/session-lifecycle-hook.mjs +++ b/plugins/codex/scripts/session-lifecycle-hook.mjs @@ -5,20 +5,19 @@ import process from "node:process"; import { terminateProcessTree } from "./lib/process.mjs"; import { BROKER_ENDPOINT_ENV } from "./lib/app-server.mjs"; -import { - clearBrokerSession, - LOG_FILE_ENV, - loadBrokerSession, - PID_FILE_ENV, - sendBrokerShutdown, - teardownBrokerSession -} from "./lib/broker-lifecycle.mjs"; -import { loadState, resolveStateFile, saveState } from "./lib/state.mjs"; +import { loadBrokerSession, reapBrokerSessions, waitForBrokerEndpoint } from "./lib/broker-lifecycle.mjs"; +import { interruptAppServerTurn } from "./lib/codex.mjs"; +import { loadState, readJobFile, resolveJobFile, resolveStateFile, saveState } from "./lib/state.mjs"; import { TRANSCRIPT_PATH_ENV } from "./lib/claude-session-transfer.mjs"; import { resolveWorkspaceRoot } from "./lib/workspace.mjs"; export const SESSION_ID_ENV = "CODEX_COMPANION_SESSION_ID"; const PLUGIN_DATA_ENV = "CLAUDE_PLUGIN_DATA"; +// The SessionEnd hook runs under a 5-second timeout (hooks/hooks.json). Cap +// the total time spent on turn interrupts well below that, so an app server +// that never answers turn/interrupt cannot get the hook killed before the +// runner termination and state cleanup below it have run. +const INTERRUPT_BUDGET_MS = 2000; function readHookInput() { const raw = fs.readFileSync(0, "utf8").trim(); @@ -39,7 +38,24 @@ function appendEnvVar(name, value) { fs.appendFileSync(process.env.CLAUDE_ENV_FILE, `export ${name}=${shellEscape(value)}\n`, "utf8"); } -function cleanupSessionJobs(cwd, sessionId) { +function readJobTurn(workspaceRoot, job) { + let stored = {}; + try { + const jobFile = resolveJobFile(workspaceRoot, job.id); + if (fs.existsSync(jobFile)) { + stored = readJobFile(jobFile) ?? {}; + } + } catch { + // A corrupt or unreadable job file must not fail the whole hook; fall + // back to the ids on the state entry. + } + return { + threadId: stored.threadId ?? job.threadId ?? null, + turnId: stored.turnId ?? job.turnId ?? null + }; +} + +async function cleanupSessionJobs(cwd, sessionId) { if (!cwd || !sessionId) { return; } @@ -56,10 +72,50 @@ function cleanupSessionJobs(cwd, sessionId) { return; } - for (const job of removedJobs) { - const stillRunning = job.status === "queued" || job.status === "running"; - if (!stillRunning) { - continue; + const runningJobs = removedJobs.filter((job) => job.status === "queued" || job.status === "running"); + + // A running job's Codex turn executes inside the shared app server, which + // deliberately outlives this session. Interrupt each owned turn (as + // handleCancel does), so killing the runner cannot leave a clientless turn + // running in the broker until it completes or the broker idles out. The + // interrupt is skipped when no live broker exists: the turn died with its + // app server. Each runner is terminated immediately after its own + // interrupt, not in a later pass, so a runner that exits in response to + // the interrupt cannot have its PID reused (and the reused PID killed) + // while other jobs' interrupts are still awaited. + let brokerAlive = false; + if (runningJobs.length > 0) { + // Same endpoint precedence as CodexAppServerClient.connect: an env-provided + // endpoint wins over the recorded one, so the broker probed here is the + // broker the interrupt below will actually reach. + const brokerEndpoint = process.env[BROKER_ENDPOINT_ENV] || loadBrokerSession(cwd)?.endpoint || null; + brokerAlive = brokerEndpoint + ? await waitForBrokerEndpoint(brokerEndpoint, 150).catch(() => false) + : false; + } + + const interruptDeadline = Date.now() + INTERRUPT_BUDGET_MS; + for (const job of runningJobs) { + const remainingMs = interruptDeadline - Date.now(); + if (brokerAlive && remainingMs > 0) { + const { threadId, turnId } = readJobTurn(workspaceRoot, job); + if (threadId && turnId) { + try { + // Race the interrupt against the remaining budget; an abandoned + // attempt is simply left behind (main exits explicitly). + await Promise.race([ + // skipAvailabilityCheck: the endpoint probe above proved the + // runtime exists, and the availability check's synchronous spawns + // would block the event loop, making this budget race ineffective. + interruptAppServerTurn(cwd, { threadId, turnId, skipAvailabilityCheck: true }), + new Promise((resolve) => { + setTimeout(resolve, remainingMs).unref(); + }) + ]); + } catch { + // Ignore interrupt failures during session shutdown. + } + } } try { terminateProcessTree(job.pid ?? Number.NaN); @@ -68,49 +124,35 @@ function cleanupSessionJobs(cwd, sessionId) { } } + // Re-load before saving: the awaited interrupts above can take a while, and + // saving the stale snapshot from the top of this function would clobber any + // state written by a concurrent session in the meantime. + const currentState = loadState(workspaceRoot); saveState(workspaceRoot, { - ...state, - jobs: state.jobs.filter((job) => job.sessionId !== sessionId) + ...currentState, + jobs: currentState.jobs.filter((job) => job.sessionId !== sessionId) }); } -function handleSessionStart(input) { +async function handleSessionStart(input) { appendEnvVar(SESSION_ID_ENV, input.session_id); appendEnvVar(TRANSCRIPT_PATH_ENV, input.transcript_path); appendEnvVar(PLUGIN_DATA_ENV, process.env[PLUGIN_DATA_ENV]); + // GC broker session dirs left behind by brokers that have already exited + // (they self-shut-down when idle). Live brokers are never touched. + await reapBrokerSessions(); } async function handleSessionEnd(input) { const cwd = input.cwd || process.cwd(); - const brokerSession = - loadBrokerSession(cwd) ?? - (process.env[BROKER_ENDPOINT_ENV] - ? { - endpoint: process.env[BROKER_ENDPOINT_ENV], - pidFile: process.env[PID_FILE_ENV] ?? null, - logFile: process.env[LOG_FILE_ENV] ?? null - } - : null); - const brokerEndpoint = brokerSession?.endpoint ?? null; - const pidFile = brokerSession?.pidFile ?? null; - const logFile = brokerSession?.logFile ?? null; - const sessionDir = brokerSession?.sessionDir ?? null; - const pid = brokerSession?.pid ?? null; - - if (brokerEndpoint) { - await sendBrokerShutdown(brokerEndpoint); - } - - cleanupSessionJobs(cwd, input.session_id || process.env[SESSION_ID_ENV]); - teardownBrokerSession({ - endpoint: brokerEndpoint, - pidFile, - logFile, - sessionDir, - pid, - killProcess: terminateProcessTree - }); - clearBrokerSession(cwd); + await cleanupSessionJobs(cwd, input.session_id || process.env[SESSION_ID_ENV]); + // Do not shut down or kill this cwd's broker, and do not clear its state + // record: both are shared with any concurrent session on the same cwd. The + // broker exits itself once idle and removes its own record then + // (app-server-broker.mjs), so a session ending leaves it entirely alone. + // GC the directories of brokers that have already exited. A live broker, + // including one a concurrent session is still using, is never touched. + await reapBrokerSessions(); } async function main() { @@ -118,7 +160,7 @@ async function main() { const eventName = process.argv[2] ?? input.hook_event_name ?? ""; if (eventName === "SessionStart") { - handleSessionStart(input); + await handleSessionStart(input); return; } @@ -127,7 +169,14 @@ async function main() { } } -main().catch((error) => { - process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); - process.exit(1); -}); +main() + .then(() => { + // Exit explicitly: an interrupt attempt abandoned by the budget race can + // hold sockets or child processes that would otherwise keep the hook + // process alive until the harness timeout kills it. + process.exit(0); + }) + .catch((error) => { + process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); + process.exit(1); + }); diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs new file mode 100644 index 000000000..34657c0ba --- /dev/null +++ b/tests/broker-lifecycle.test.mjs @@ -0,0 +1,128 @@ +import fs from "node:fs"; +import path from "node:path"; +import test from "node:test"; +import assert from "node:assert/strict"; + +import { makeTempDir } from "./helpers.mjs"; +import { + clearBrokerSession, + loadBrokerSession, + reapBrokerSessions, + saveBrokerSession +} from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { getSessionRuntimeStatus } from "../plugins/codex/scripts/lib/codex.mjs"; + +function makeSessionDir(tmpDir, name, pidContents, { managed = true } = {}) { + const sessionDir = path.join(tmpDir, name); + fs.mkdirSync(sessionDir, { recursive: true }); + if (managed) { + fs.writeFileSync(path.join(sessionDir, "broker.managed"), "test marker\n", "utf8"); + } + if (pidContents !== undefined) { + fs.writeFileSync(path.join(sessionDir, "broker.pid"), pidContents, "utf8"); + } + return sessionDir; +} + +function findDeadPid() { + // Spawn-free approach: walk down from a high pid until one is not alive. + for (let pid = 999_999; pid > 900_000; pid -= 1) { + try { + process.kill(pid, 0); + } catch (error) { + if (error.code === "ESRCH") { + return pid; + } + } + } + throw new Error("could not find a dead pid to test with"); +} + +test("reapBrokerSessions removes only dirs whose recorded broker pid is dead", async () => { + const tmpDir = makeTempDir(); + const deadPid = findDeadPid(); + + const deadDir = makeSessionDir(tmpDir, "cxc-dead", `${deadPid}\n`); + const liveDir = makeSessionDir(tmpDir, "cxc-live", `${process.pid}\n`); + const pidlessDir = makeSessionDir(tmpDir, "cxc-pidless"); + const tornDir = makeSessionDir(tmpDir, "cxc-torn", ""); + const garbageDir = makeSessionDir(tmpDir, "cxc-garbage", "not-a-pid\n"); + const zeroDir = makeSessionDir(tmpDir, "cxc-zero", "0\n"); + const unrelatedDir = makeSessionDir(tmpDir, "other-prefix", `${deadPid}\n`); + const unmarkedDir = makeSessionDir(tmpDir, "cxc-user-work", `${deadPid}\n`, { managed: false }); + fs.writeFileSync(path.join(unmarkedDir, "important.txt"), "user file", "utf8"); + + await reapBrokerSessions({ tmpDir }); + + assert.equal(fs.existsSync(deadDir), false, "dead-pid dir should be removed"); + assert.equal( + fs.existsSync(path.join(unmarkedDir, "important.txt")), + true, + "a dir without the ownership marker is never deleted, dead pid or not" + ); + assert.equal(fs.existsSync(liveDir), true, "live-pid dir must never be touched"); + assert.equal(fs.existsSync(pidlessDir), true, "dir without broker.pid is not a broker session"); + assert.equal(fs.existsSync(tornDir), true, "empty pid file may be a torn write; leave it"); + assert.equal(fs.existsSync(garbageDir), true, "unparseable pid file is left alone"); + assert.equal(fs.existsSync(zeroDir), true, "pid 0 is never treated as a signalable broker"); + assert.equal(fs.existsSync(unrelatedDir), true, "non cxc- prefixed dirs are ignored"); +}); + +test("getSessionRuntimeStatus reports shared only while the recorded broker pid is alive", () => { + const workspace = makeTempDir(); + const sessionDir = makeTempDir(); + const endpoint = `unix:${path.join(sessionDir, "broker.sock")}`; + + saveBrokerSession(workspace, { endpoint, pid: process.pid }); + assert.equal(getSessionRuntimeStatus({}, workspace).mode, "shared"); + + saveBrokerSession(workspace, { endpoint, pid: findDeadPid() }); + assert.equal(getSessionRuntimeStatus({}, workspace).mode, "direct"); + + clearBrokerSession(workspace); + assert.equal(getSessionRuntimeStatus({}, workspace).mode, "direct"); +}); + +test("getSessionRuntimeStatus falls back to socket presence for records without a pid", () => { + const workspace = makeTempDir(); + const sessionDir = makeTempDir(); + const socketPath = path.join(sessionDir, "broker.sock"); + + saveBrokerSession(workspace, { endpoint: `unix:${socketPath}` }); + assert.equal(getSessionRuntimeStatus({}, workspace).mode, "direct", "missing socket file means no runtime"); + + fs.writeFileSync(socketPath, "", "utf8"); + assert.equal(getSessionRuntimeStatus({}, workspace).mode, "shared", "present socket is the best available signal"); + + clearBrokerSession(workspace); +}); + +test("getSessionRuntimeStatus env override wins and an empty override masks the record", () => { + const workspace = makeTempDir(); + const sessionDir = makeTempDir(); + const socketPath = path.join(sessionDir, "broker.sock"); + fs.writeFileSync(socketPath, "", "utf8"); + + saveBrokerSession(workspace, { endpoint: `unix:${socketPath}`, pid: process.pid }); + + const env = { CODEX_COMPANION_APP_SERVER_ENDPOINT: "" }; + assert.equal(getSessionRuntimeStatus(env, workspace).mode, "direct", "empty env override masks the record"); + + env.CODEX_COMPANION_APP_SERVER_ENDPOINT = `unix:${socketPath}`; + assert.equal(getSessionRuntimeStatus(env, workspace).mode, "shared"); + + const missing = path.join(sessionDir, "gone.sock"); + env.CODEX_COMPANION_APP_SERVER_ENDPOINT = `unix:${missing}`; + assert.equal(getSessionRuntimeStatus(env, workspace).mode, "direct", "stale env endpoint is not reported as live"); + + clearBrokerSession(workspace); +}); + +test("loadBrokerSession round-trips and clearBrokerSession removes the record", () => { + const workspace = makeTempDir(); + const session = { endpoint: "unix:/tmp/nowhere.sock", pid: 12345 }; + saveBrokerSession(workspace, session); + assert.deepEqual(loadBrokerSession(workspace), session); + clearBrokerSession(workspace); + assert.equal(loadBrokerSession(workspace), null); +}); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index f83c96a0d..727b6cdf9 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -653,6 +653,10 @@ export function buildEnv(binDir) { const sep = process.platform === "win32" ? ";" : ":"; return { ...process.env, - PATH: `${binDir}${sep}${process.env.PATH}` + PATH: `${binDir}${sep}${process.env.PATH}`, + // Brokers exit themselves when idle; without a short window here, every + // broker-spawning test would leave a broker + fake-codex pair running for + // the default 30 minutes after the suite finishes. + CODEX_BROKER_IDLE_MS: process.env.CODEX_BROKER_IDLE_MS ?? "30000" }; } diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 8f276835b..c157e210f 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -2240,7 +2240,10 @@ test("setup and status honor --cwd when reading shared session runtime", () => { const invocationWorkspace = makeTempDir(); saveBrokerSession(targetWorkspace, { - endpoint: "unix:/tmp/fake-broker.sock" + endpoint: "unix:/tmp/fake-broker.sock", + // Status only reports a shared runtime for a record whose broker is still + // live; recording this (alive) test process satisfies the pid check. + pid: process.pid }); const status = run("node", [SCRIPT, "status", "--cwd", targetWorkspace], {