diff --git a/plugins/codex/CHANGELOG.md b/plugins/codex/CHANGELOG.md index d647561bb..bcd5d5c69 100644 --- a/plugins/codex/CHANGELOG.md +++ b/plugins/codex/CHANGELOG.md @@ -1,5 +1,10 @@ # Changelog +## Unreleased + +- Stop an orphaned shared Codex app-server broker after 15 minutes with no connected clients. Active foreground and background jobs keep their broker connection open, and the timeout can be configured with `CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS` (`0` disables the safety timer). +- Close the broker listener before asynchronous child cleanup and safely reject reconnects already queued during shutdown. + ## 1.0.0 - Initial version of the Codex plugin for Claude Code diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 1954274fe..0231f8103 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -10,6 +10,21 @@ import { BROKER_BUSY_RPC_CODE, CodexAppServerClient } from "./lib/app-server.mjs import { parseBrokerEndpoint } from "./lib/broker-endpoint.mjs"; const STREAMING_METHODS = new Set(["turn/start", "review/start", "thread/compact/start"]); +const BROKER_IDLE_TIMEOUT_ENV = "CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS"; +const DEFAULT_BROKER_IDLE_TIMEOUT_MS = 15 * 60 * 1000; + +function resolveIdleTimeoutMs(env = process.env) { + const rawValue = env[BROKER_IDLE_TIMEOUT_ENV]; + if (rawValue == null || rawValue === "") { + return DEFAULT_BROKER_IDLE_TIMEOUT_MS; + } + + const parsed = Number(rawValue); + if (!Number.isFinite(parsed) || parsed < 0) { + return DEFAULT_BROKER_IDLE_TIMEOUT_MS; + } + return Math.floor(parsed); +} function buildStreamThreadIds(method, params, result) { const threadIds = new Set(); @@ -70,6 +85,16 @@ async function main() { let activeStreamSocket = null; let activeStreamThreadIds = null; const sockets = new Set(); + const idleTimeoutMs = resolveIdleTimeoutMs(); + let idleTimer = null; + let shuttingDown = false; + + function cancelIdleShutdown() { + if (idleTimer) { + clearTimeout(idleTimer); + idleTimer = null; + } + } function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { @@ -100,11 +125,20 @@ async function main() { } async function shutdown(server) { + if (shuttingDown) { + return; + } + shuttingDown = true; + cancelIdleShutdown(); + // Stop accepting connections before awaiting child cleanup. Otherwise a + // reconnect can slip into the async shutdown window and inherit a closing + // app-server client. + const serverClosed = 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 serverClosed; if (listenTarget.kind === "unix" && fs.existsSync(listenTarget.path)) { fs.unlinkSync(listenTarget.path); } @@ -113,9 +147,33 @@ async function main() { } } + function scheduleIdleShutdown(server) { + cancelIdleShutdown(); + if (shuttingDown || idleTimeoutMs === 0 || sockets.size > 0 || activeRequestSocket || activeStreamSocket) { + return; + } + + idleTimer = setTimeout(async () => { + idleTimer = null; + if (sockets.size > 0 || activeRequestSocket || activeStreamSocket) { + return; + } + await shutdown(server); + process.exit(0); + }, idleTimeoutMs); + } + appClient.setNotificationHandler(routeNotification); const server = net.createServer((socket) => { + if (shuttingDown) { + // An already-accepted connection event can be delivered after + // server.close(). Reject it decisively and absorb a simultaneous reset. + socket.on("error", () => {}); + socket.destroy(); + return; + } + cancelIdleShutdown(); sockets.add(socket); socket.setEncoding("utf8"); let buffer = ""; @@ -225,11 +283,13 @@ async function main() { socket.on("close", () => { sockets.delete(socket); clearSocketOwnership(socket); + scheduleIdleShutdown(server); }); socket.on("error", () => { sockets.delete(socket); clearSocketOwnership(socket); + scheduleIdleShutdown(server); }); }); @@ -243,7 +303,9 @@ async function main() { process.exit(0); }); - server.listen(listenTarget.path); + server.listen(listenTarget.path, () => { + scheduleIdleShutdown(server); + }); } main().catch((error) => { diff --git a/tests/broker-idle.test.mjs b/tests/broker-idle.test.mjs new file mode 100644 index 000000000..6f3860d46 --- /dev/null +++ b/tests/broker-idle.test.mjs @@ -0,0 +1,126 @@ +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 { once } from "node:events"; +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; + +import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs"; +import { makeTempDir } from "./helpers.mjs"; +import { sendBrokerShutdown, waitForBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; + +const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); +const BROKER_SCRIPT = path.join(ROOT, "plugins", "codex", "scripts", "app-server-broker.mjs"); +const IDLE_TIMEOUT_ENV = "CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS"; + +function spawnTestBroker({ idleTimeoutMs }) { + const workspace = makeTempDir(); + const binDir = makeTempDir(); + const sessionDir = makeTempDir("codex-plugin-broker-"); + const socketPath = path.join(sessionDir, "broker.sock"); + const endpoint = `unix:${socketPath}`; + const pidFile = path.join(sessionDir, "broker.pid"); + installFakeCodex(binDir); + + const child = spawn( + process.execPath, + [BROKER_SCRIPT, "serve", "--endpoint", endpoint, "--cwd", workspace, "--pid-file", pidFile], + { + cwd: workspace, + env: { + ...buildEnv(binDir), + [IDLE_TIMEOUT_ENV]: String(idleTimeoutMs) + }, + stdio: ["ignore", "pipe", "pipe"] + } + ); + + return { child, endpoint, pidFile, socketPath }; +} + +async function waitForExit(child, timeoutMs = 3000) { + if (child.exitCode != null || child.signalCode != null) { + return; + } + await Promise.race([ + once(child, "exit"), + new Promise((_, reject) => setTimeout(() => reject(new Error("Timed out waiting for broker to exit.")), timeoutMs)) + ]); +} + +function terminateBroker(child) { + if (child.exitCode == null && child.signalCode == null) { + child.kill("SIGTERM"); + } +} + +test("broker exits and removes runtime files after its last client stays disconnected", async (t) => { + const broker = spawnTestBroker({ idleTimeoutMs: 150 }); + t.after(() => terminateBroker(broker.child)); + + assert.equal(await waitForBrokerEndpoint(broker.endpoint), true); + await waitForExit(broker.child); + + assert.equal(broker.child.exitCode, 0); + assert.equal(fs.existsSync(broker.socketPath), false); + assert.equal(fs.existsSync(broker.pidFile), false); +}); + +test("broker idle shutdown waits until a connected client disconnects", async (t) => { + const broker = spawnTestBroker({ idleTimeoutMs: 150 }); + t.after(() => terminateBroker(broker.child)); + + assert.equal(await waitForBrokerEndpoint(broker.endpoint), true); + const socket = net.createConnection({ path: broker.socketPath }); + await once(socket, "connect"); + + await new Promise((resolve) => setTimeout(resolve, 350)); + assert.equal(broker.child.exitCode, null); + assert.equal(broker.child.signalCode, null); + + socket.end(); + await once(socket, "close"); + await waitForExit(broker.child); + assert.equal(broker.child.exitCode, 0); +}); + +test("broker idle timeout can be disabled", async (t) => { + const broker = spawnTestBroker({ idleTimeoutMs: 0 }); + t.after(() => terminateBroker(broker.child)); + + assert.equal(await waitForBrokerEndpoint(broker.endpoint), true); + await new Promise((resolve) => setTimeout(resolve, 350)); + assert.equal(broker.child.exitCode, null); + assert.equal(broker.child.signalCode, null); + + terminateBroker(broker.child); + await waitForExit(broker.child); +}); + +test("broker shuts down cleanly while new clients race to connect", async (t) => { + const broker = spawnTestBroker({ idleTimeoutMs: 0 }); + t.after(() => terminateBroker(broker.child)); + + assert.equal(await waitForBrokerEndpoint(broker.endpoint), true); + const reconnects = Array.from( + { length: 50 }, + () => + new Promise((resolve) => { + const socket = net.createConnection({ path: broker.socketPath }); + const timer = setTimeout(() => socket.destroy(), 1000); + const finish = () => { + clearTimeout(timer); + resolve(); + }; + socket.on("connect", () => socket.end()); + socket.on("error", finish); + socket.on("close", finish); + }) + ); + + await Promise.all([sendBrokerShutdown(broker.endpoint), ...reconnects]); + await waitForExit(broker.child); + assert.equal(broker.child.exitCode, 0); +}); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index f83c96a0d..e0d3957f4 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -653,6 +653,9 @@ export function buildEnv(binDir) { const sep = process.platform === "win32" ? ";" : ":"; return { ...process.env, - PATH: `${binDir}${sep}${process.env.PATH}` + PATH: `${binDir}${sep}${process.env.PATH}`, + // Production keeps an idle broker warm for 15 minutes. Tests only need a + // brief reuse window and should not leave dozens of detached helpers. + CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS: "2000" }; }