diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index ef763819..11615784 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -6,11 +6,150 @@ import process from "node:process"; import { spawn } from "node:child_process"; import { fileURLToPath } from "node:url"; import { createBrokerEndpoint, parseBrokerEndpoint } from "./broker-endpoint.mjs"; -import { resolveStateDir } from "./state.mjs"; +import { resolveStateDir, resolveStateRoot } from "./state.mjs"; 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 SESSION_ID_ENV = "CODEX_COMPANION_SESSION_ID"; +const BROKER_STATE_LOCK_STALE_MS = 30000; +const BROKER_STATE_LOCK_TIMEOUT_MS = 5000; +const BROKER_SHUTDOWN_TIMEOUT_MS = 1000; +const BROKER_LOCK_TIMEOUT_CODE = "EBROKERSTATELOCKTIMEOUT"; + +let brokerLockTokenSeq = 0; + +export function resolveSessionId(options = {}) { + if (options.sessionId) { + return options.sessionId; + } + const env = options.env ?? process.env; + return env[SESSION_ID_ENV] ?? null; +} + +function brokerSessionOwners(session) { + const owners = []; + if (Array.isArray(session?.sessionIds)) { + owners.push(...session.sessionIds); + } + if (session?.sessionId) { + owners.push(session.sessionId); + } + return [...new Set(owners.filter(Boolean))]; +} + +// True when the broker is still owned by a session other than `sessionId`. +// Used by SessionEnd to decide whether the cwd fallback must still run: if the +// only remaining owner is the ending session (e.g. its session-keyed teardown +// was skipped under lock contention), the broker must NOT be left behind. +export function hasOtherBrokerSessionOwners(session, sessionId) { + return brokerSessionOwners(session).some((owner) => owner !== sessionId); +} + +function withBrokerSessionOwner(session, sessionId) { + if (!sessionId) { + return session; + } + const owners = brokerSessionOwners(session); + if (!owners.includes(sessionId)) { + owners.push(sessionId); + } + return { + ...session, + sessionId: owners[0] ?? sessionId, + sessionIds: owners + }; +} + +async function sleep(ms) { + await new Promise((resolve) => setTimeout(resolve, ms)); +} + +function isBrokerLockTimeout(error) { + return error?.code === BROKER_LOCK_TIMEOUT_CODE; +} + +async function withBrokerStateFileLock(stateFile, fn, options = {}) { + const lockDir = `${stateFile}.lock`; + const tokenFile = path.join(lockDir, "owner"); + const token = `${process.pid}-${brokerLockTokenSeq += 1}`; + const timeoutMs = options.timeoutMs ?? BROKER_STATE_LOCK_TIMEOUT_MS; + const staleMs = options.staleMs ?? BROKER_STATE_LOCK_STALE_MS; + const deadline = Date.now() + timeoutMs; + + while (true) { + try { + fs.mkdirSync(lockDir); + break; + } catch (error) { + if (error?.code !== "EEXIST") { + throw error; + } + try { + const stat = fs.statSync(lockDir); + if (Date.now() - stat.mtimeMs > staleMs) { + // Reclaim atomically: rename the stale directory to a private name so + // only one reclaimer can take it — a blind rmSync lets two reclaimers + // each delete the other's freshly created lock. After the rename, + // re-check the mtime: if the directory we grabbed was refreshed after + // our stat (a fresh lock, not the stale one), put it back untouched. + const claimed = `${lockDir}.reclaim-${process.pid}-${(brokerLockTokenSeq += 1)}`; + try { + fs.renameSync(lockDir, claimed); + if (Date.now() - fs.statSync(claimed).mtimeMs > staleMs) { + fs.rmSync(claimed, { recursive: true, force: true }); + } else { + try { + fs.renameSync(claimed, lockDir); + } catch { + // A new lock already took the path; drop the moved copy. + fs.rmSync(claimed, { recursive: true, force: true }); + } + } + } catch { + // Lost the reclaim race; fall through and retry acquisition. + } + continue; + } + } catch { + continue; + } + if (Date.now() >= deadline) { + const timeout = new Error(`Timed out waiting for broker state lock: ${stateFile}`); + timeout.code = BROKER_LOCK_TIMEOUT_CODE; + throw timeout; + } + await sleep(25); + } + } + + // Stamp ownership so the finally only releases a lock we still hold — a + // stale reclaim by another waiter must not have its lock deleted from under it. + let stamped = false; + try { + fs.writeFileSync(tokenFile, token, "utf8"); + stamped = true; + } catch { + // Could not stamp (quota/permission race). We still created the lock dir, + // so release it unconditionally below rather than leaking it. + } + + try { + return await fn(); + } finally { + let owned = true; + if (stamped) { + try { + owned = fs.readFileSync(tokenFile, "utf8") === token; + } catch { + owned = false; + } + } + if (owned) { + fs.rmSync(lockDir, { recursive: true, force: true }); + } + } +} export function createBrokerSessionDir(prefix = "cxc-") { return fs.mkdtempSync(path.join(os.tmpdir(), prefix)); @@ -40,19 +179,46 @@ export async function waitForBrokerEndpoint(endpoint, timeoutMs = 2000) { return false; } -export async function sendBrokerShutdown(endpoint) { +export async function sendBrokerShutdown(endpoint, { timeoutMs = BROKER_SHUTDOWN_TIMEOUT_MS } = {}) { await new Promise((resolve) => { - const socket = connectToEndpoint(endpoint); + let settled = false; + let timer = null; + const finish = () => { + if (settled) { + return; + } + settled = true; + if (timer) { + clearTimeout(timer); + } + resolve(); + }; + // Graceful shutdown is best-effort: a corrupt/unsupported endpoint makes + // connectToEndpoint (parseBrokerEndpoint) throw synchronously. Never reject, + // so callers always fall through to the forced process/file teardown. + let socket; + try { + socket = connectToEndpoint(endpoint); + } catch { + finish(); + return; + } + if (timeoutMs > 0) { + timer = setTimeout(() => { + socket.destroy(); + finish(); + }, timeoutMs); + } socket.setEncoding("utf8"); socket.on("connect", () => { socket.write(`${JSON.stringify({ id: 1, method: "broker/shutdown", params: {} })}\n`); }); socket.on("data", () => { socket.end(); - resolve(); + finish(); }); - socket.on("error", resolve); - socket.on("close", resolve); + socket.on("error", finish); + socket.on("close", finish); }); } @@ -86,10 +252,19 @@ export function loadBrokerSession(cwd) { } } +// Write broker.json atomically (temp + rename) so a concurrent reader — e.g. an +// unlocked ownership pre-check in another session's teardown — never observes a +// half-written file and mis-parses it. +function writeBrokerStateFile(stateFile, session) { + const tmp = `${stateFile}.tmp-${process.pid}-${(brokerLockTokenSeq += 1)}`; + fs.writeFileSync(tmp, `${JSON.stringify(session, null, 2)}\n`, "utf8"); + fs.renameSync(tmp, stateFile); +} + export function saveBrokerSession(cwd, session) { const stateDir = resolveStateDir(cwd); fs.mkdirSync(stateDir, { recursive: true }); - fs.writeFileSync(resolveBrokerStateFile(cwd), `${JSON.stringify(session, null, 2)}\n`, "utf8"); + writeBrokerStateFile(resolveBrokerStateFile(cwd), session); } export function clearBrokerSession(cwd) { @@ -110,24 +285,10 @@ async function isBrokerEndpointReady(endpoint) { } } -export async function ensureBrokerSession(cwd, options = {}) { - const existing = loadBrokerSession(cwd); - if (existing && (await isBrokerEndpointReady(existing.endpoint))) { - return existing; - } - - if (existing) { - teardownBrokerSession({ - endpoint: existing.endpoint ?? null, - pidFile: existing.pidFile ?? null, - logFile: existing.logFile ?? null, - sessionDir: existing.sessionDir ?? null, - pid: existing.pid ?? null, - killProcess: options.killProcess ?? null - }); - clearBrokerSession(cwd); - } - +// Spawn a broker process and wait for it to accept connections. Returns an +// unsaved session record, or null if it never became ready. No locking or +// persistence — the caller owns those. +async function spawnReadyBroker(cwd, options) { const sessionDir = createBrokerSessionDir(); const endpointFactory = options.createBrokerEndpoint ?? createBrokerEndpoint; const endpoint = endpointFactory(sessionDir, options.platform); @@ -159,15 +320,234 @@ export async function ensureBrokerSession(cwd, options = {}) { return null; } - const session = { + return { endpoint, pidFile, logFile, sessionDir, - pid: child.pid ?? null + pid: child.pid ?? null, + sessionId: resolveSessionId(options) }; - saveBrokerSession(cwd, session); - return session; +} + +// Tear down whatever broker is recorded for cwd (best effort) so a fresh one +// can replace it. +function discardBrokerSession(cwd, session, options) { + if (!session) { + return; + } + teardownBrokerSession({ + endpoint: session.endpoint ?? null, + pidFile: session.pidFile ?? null, + logFile: session.logFile ?? null, + sessionDir: session.sessionDir ?? null, + pid: session.pid ?? null, + killProcess: options.killProcess ?? null + }); + clearBrokerSession(cwd); +} + +// Adopt cwd's recorded broker for this session if it is still live, else spawn +// and persist a fresh one. Callers run this inside the state-file lock so two +// racing sessions cannot each leave an orphaned broker with no broker.json. +async function adoptOrSpawnBroker(cwd, options) { + const current = loadBrokerSession(cwd); + if (current && (await isBrokerEndpointReady(current.endpoint))) { + const withOwner = withBrokerSessionOwner(current, resolveSessionId(options)); + if (withOwner !== current) { + saveBrokerSession(cwd, withOwner); + } + return withOwner; + } + discardBrokerSession(cwd, current, options); + + const session = await spawnReadyBroker(cwd, options); + if (!session) { + return null; + } + const withOwner = withBrokerSessionOwner(session, session.sessionId); + saveBrokerSession(cwd, withOwner); + return withOwner; +} + +export async function ensureBrokerSession(cwd, options = {}) { + const stateFile = resolveBrokerStateFile(cwd); + + // Fast path: reuse a ready broker. The readiness probe runs outside the lock, + // so the broker may have been torn down (or replaced) before we acquired it; + // trust only the locked re-read, never the pre-lock snapshot. + for (let attempt = 0; attempt < 2; attempt += 1) { + const existing = loadBrokerSession(cwd); + if (!existing || !(await isBrokerEndpointReady(existing.endpoint))) { + break; + } + let lockUnavailable = false; + const reused = await withBrokerStateFileLock(stateFile, () => { + const current = loadBrokerSession(cwd); + if (!current || current.endpoint !== existing.endpoint) { + return null; + } + const withOwner = withBrokerSessionOwner(current, resolveSessionId(options)); + if (withOwner !== current) { + saveBrokerSession(cwd, withOwner); + } + return withOwner; + }).catch((error) => { + if (isBrokerLockTimeout(error)) { + lockUnavailable = true; + return null; + } + throw error; + }); + if (reused) { + return reused; + } + if (lockUnavailable) { + break; + } + // State changed while we waited for the lock — re-probe: a replacement + // broker may already be live and reusable. + } + + // Slow path: adopt-or-spawn under the lock so concurrent spawns don't orphan + // brokers. resolveStateDir must exist before we can create the lock dir. + fs.mkdirSync(resolveStateDir(cwd), { recursive: true }); + let lockUnavailable = false; + const created = await withBrokerStateFileLock(stateFile, () => adoptOrSpawnBroker(cwd, options)).catch((error) => { + if (isBrokerLockTimeout(error)) { + lockUnavailable = true; + return null; + } + throw error; + }); + if (!lockUnavailable) { + return created; + } + + // Lock stayed busy past the timeout: proceed without it rather than block + // starting Codex. adopt-or-spawn still reuses a live recorded broker (and + // only discards one that fails its readiness probe), so a long-held lock + // never causes another session's live broker to be unlinked/orphaned. Rare, + // and degrades only to the pre-lock race window. + return adoptOrSpawnBroker(cwd, options); +} + +export async function teardownBrokersForSession( + sessionId, + { + killProcess = null, + shutdownTimeoutMs = BROKER_SHUTDOWN_TIMEOUT_MS, + lockTimeoutMs = BROKER_STATE_LOCK_TIMEOUT_MS, + budgetMs = null + } = {} +) { + if (!sessionId) { + return 0; + } + const stateRoot = resolveStateRoot(); + if (!fs.existsSync(stateRoot)) { + return 0; + } + + const deadline = budgetMs != null ? Date.now() + budgetMs : null; + let count = 0; + for (const entry of fs.readdirSync(stateRoot)) { + if (deadline != null && Date.now() >= deadline) { + break; + } + const stateFile = path.join(stateRoot, entry, BROKER_STATE_FILE); + if (!fs.existsSync(stateFile)) { + continue; + } + + // Ownership pre-check without the lock: never block on a lock held for a + // workspace this session does not own. The owner set only ever grows to + // include our sessionId (reuse) or shrinks when we ourselves remove it, so + // an unlocked read cannot falsely exclude a broker we own. A parse failure + // (e.g. a genuinely corrupt file) is NOT treated as "not ours" — fall + // through to the locked re-read, which is authoritative, rather than + // skipping a broker that might belong to this session. + let preview = null; + try { + preview = JSON.parse(fs.readFileSync(stateFile, "utf8")); + } catch { + preview = null; + } + if (preview && !brokerSessionOwners(preview).includes(sessionId)) { + continue; + } + + const remaining = deadline != null ? deadline - Date.now() : lockTimeoutMs; + if (remaining <= 0) { + break; + } + const entryLockTimeout = Math.min(lockTimeoutMs, remaining); + + try { + await withBrokerStateFileLock( + stateFile, + async () => { + if (!fs.existsSync(stateFile)) { + return; + } + + let session; + try { + session = JSON.parse(fs.readFileSync(stateFile, "utf8")); + } catch { + return; + } + const owners = brokerSessionOwners(session); + if (!owners.includes(sessionId)) { + return; + } + const remainingOwners = owners.filter((owner) => owner !== sessionId); + if (remainingOwners.length > 0) { + writeBrokerStateFile(stateFile, { + ...session, + sessionId: remainingOwners[0], + sessionIds: remainingOwners + }); + return; + } + + // Bound the graceful-shutdown wait by the remaining scan budget, not + // just the per-RPC default: several unresponsive endpoints could + // otherwise each burn shutdownTimeoutMs and push the whole scan past + // the SessionEnd hook's budget. Out of budget → skip the RPC and let + // teardownBrokerSession terminate the process directly. + if (session.endpoint) { + const shutdownWait = + deadline != null ? Math.min(shutdownTimeoutMs, deadline - Date.now()) : shutdownTimeoutMs; + if (shutdownWait > 0) { + await sendBrokerShutdown(session.endpoint, { timeoutMs: shutdownWait }); + } + } + teardownBrokerSession({ + endpoint: session.endpoint ?? null, + pidFile: session.pidFile ?? null, + logFile: session.logFile ?? null, + sessionDir: session.sessionDir ?? null, + pid: session.pid ?? null, + killProcess + }); + if (fs.existsSync(stateFile)) { + fs.unlinkSync(stateFile); + } + count += 1; + }, + { timeoutMs: entryLockTimeout } + ); + } catch (error) { + if (!isBrokerLockTimeout(error)) { + throw error; + } + // Another hook holds this entry's lock and did not release it within the + // timeout. Skip it so the rest of this session's brokers still get torn + // down instead of aborting the whole scan. + } + } + return count; } export function teardownBrokerSession({ endpoint = null, pidFile, logFile, sessionDir = null, pid = null, killProcess = null }) { diff --git a/plugins/codex/scripts/lib/state.mjs b/plugins/codex/scripts/lib/state.mjs index 2da23498..5f8e04b8 100644 --- a/plugins/codex/scripts/lib/state.mjs +++ b/plugins/codex/scripts/lib/state.mjs @@ -26,6 +26,11 @@ function defaultState() { }; } +export function resolveStateRoot() { + const pluginDataDir = process.env[PLUGIN_DATA_ENV]; + return pluginDataDir ? path.join(pluginDataDir, "state") : FALLBACK_STATE_ROOT_DIR; +} + export function resolveStateDir(cwd) { const workspaceRoot = resolveWorkspaceRoot(cwd); let canonicalWorkspaceRoot = workspaceRoot; @@ -38,8 +43,7 @@ export function resolveStateDir(cwd) { const slugSource = path.basename(workspaceRoot) || "workspace"; const slug = slugSource.replace(/[^a-zA-Z0-9._-]+/g, "-").replace(/^-+|-+$/g, "") || "workspace"; const hash = createHash("sha256").update(canonicalWorkspaceRoot).digest("hex").slice(0, 16); - const pluginDataDir = process.env[PLUGIN_DATA_ENV]; - const stateRoot = pluginDataDir ? path.join(pluginDataDir, "state") : FALLBACK_STATE_ROOT_DIR; + const stateRoot = resolveStateRoot(); return path.join(stateRoot, `${slug}-${hash}`); } diff --git a/plugins/codex/scripts/session-lifecycle-hook.mjs b/plugins/codex/scripts/session-lifecycle-hook.mjs index 778571e6..0f8bab12 100644 --- a/plugins/codex/scripts/session-lifecycle-hook.mjs +++ b/plugins/codex/scripts/session-lifecycle-hook.mjs @@ -1,17 +1,21 @@ #!/usr/bin/env node import fs from "node:fs"; +import path from "node:path"; import process from "node:process"; +import { fileURLToPath } from "node:url"; import { terminateProcessTree } from "./lib/process.mjs"; import { BROKER_ENDPOINT_ENV } from "./lib/app-server.mjs"; import { clearBrokerSession, + hasOtherBrokerSessionOwners, LOG_FILE_ENV, loadBrokerSession, PID_FILE_ENV, sendBrokerShutdown, - teardownBrokerSession + teardownBrokerSession, + teardownBrokersForSession } from "./lib/broker-lifecycle.mjs"; import { loadState, resolveStateFile, saveState } from "./lib/state.mjs"; import { TRANSCRIPT_PATH_ENV } from "./lib/claude-session-transfer.mjs"; @@ -19,6 +23,12 @@ import { resolveWorkspaceRoot } from "./lib/workspace.mjs"; export const SESSION_ID_ENV = "CODEX_COMPANION_SESSION_ID"; const PLUGIN_DATA_ENV = "CLAUDE_PLUGIN_DATA"; +// hooks.json gives the SessionEnd hook a 5s budget. Keep the session-keyed +// broker scan well under it — a short per-entry lock wait plus a total scan +// deadline — so a single contended broker.json.lock cannot starve the rest of +// cleanup before the hook is killed. +const SESSION_END_TEARDOWN_BUDGET_MS = 3000; +const SESSION_END_LOCK_TIMEOUT_MS = 750; function readHookInput() { const raw = fs.readFileSync(0, "utf8").trim(); @@ -80,8 +90,24 @@ function handleSessionStart(input) { appendEnvVar(PLUGIN_DATA_ENV, process.env[PLUGIN_DATA_ENV]); } -async function handleSessionEnd(input) { +export async function handleSessionEnd(input) { const cwd = input.cwd || process.cwd(); + const sessionId = input.session_id || process.env[SESSION_ID_ENV]; + let cleanupError = null; + try { + cleanupSessionJobs(cwd, sessionId); + } catch (error) { + cleanupError = error; + } + + if (sessionId) { + await teardownBrokersForSession(sessionId, { + killProcess: terminateProcessTree, + lockTimeoutMs: SESSION_END_LOCK_TIMEOUT_MS, + budgetMs: SESSION_END_TEARDOWN_BUDGET_MS + }); + } + const brokerSession = loadBrokerSession(cwd) ?? (process.env[BROKER_ENDPOINT_ENV] @@ -91,6 +117,12 @@ async function handleSessionEnd(input) { logFile: process.env[LOG_FILE_ENV] ?? null } : null); + if (sessionId && hasOtherBrokerSessionOwners(brokerSession, sessionId)) { + if (cleanupError) { + throw cleanupError; + } + return; + } const brokerEndpoint = brokerSession?.endpoint ?? null; const pidFile = brokerSession?.pidFile ?? null; const logFile = brokerSession?.logFile ?? null; @@ -101,7 +133,6 @@ async function handleSessionEnd(input) { await sendBrokerShutdown(brokerEndpoint); } - cleanupSessionJobs(cwd, input.session_id || process.env[SESSION_ID_ENV]); teardownBrokerSession({ endpoint: brokerEndpoint, pidFile, @@ -111,6 +142,9 @@ async function handleSessionEnd(input) { killProcess: terminateProcessTree }); clearBrokerSession(cwd); + if (cleanupError) { + throw cleanupError; + } } async function main() { @@ -127,7 +161,26 @@ async function main() { } } -main().catch((error) => { - process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); - process.exit(1); -}); +function isExecutedDirectly() { + if (!process.argv[1]) { + return false; + } + + try { + return fs.realpathSync(fileURLToPath(import.meta.url)) === + fs.realpathSync(path.resolve(process.argv[1])); + } catch { + // argv[1] may not resolve to a real path (deleted file, loader indirection); + // importing must never crash at module load. + return false; + } +} + +// Only run main() when executed directly (not when imported by tests). +const isMain = isExecutedDirectly(); +if (isMain) { + main().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 00000000..4b0eb8e3 --- /dev/null +++ b/tests/broker-lifecycle.test.mjs @@ -0,0 +1,829 @@ +import fs from "node:fs"; +import net from "node:net"; +import os from "node:os"; +import path from "node:path"; +import { spawn } from "node:child_process"; +import { pathToFileURL } from "node:url"; +import test from "node:test"; +import assert from "node:assert/strict"; + +import { makeTempDir } from "./helpers.mjs"; +import { createBrokerEndpoint, parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; +import { resolveStateDir, resolveStateRoot } from "../plugins/codex/scripts/lib/state.mjs"; +import { + ensureBrokerSession, + loadBrokerSession, + resolveSessionId, + saveBrokerSession, + teardownBrokersForSession +} from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { handleSessionEnd } from "../plugins/codex/scripts/session-lifecycle-hook.mjs"; + +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +// Minimal stand-in for app-server-broker.mjs: honors the spawn contract +// (serve --endpoint --cwd --pid-file) enough for waitForBrokerEndpoint. +const FAKE_BROKER_SCRIPT = `import fs from "node:fs"; +import net from "node:net"; + +const args = process.argv.slice(2); +const get = (name) => args[args.indexOf(name) + 1]; +const sockPath = get("--endpoint").replace(/^unix:/, ""); +const server = net.createServer((socket) => socket.end()); +server.listen(sockPath, () => { + fs.writeFileSync(get("--pid-file"), String(process.pid), "utf8"); +}); +`; + +function writeFakeBrokerScript() { + const scriptPath = path.join(makeTempDir(), "fake-broker.mjs"); + fs.writeFileSync(scriptPath, FAKE_BROKER_SCRIPT, "utf8"); + return scriptPath; +} + +const stateRootForTest = resolveStateRoot; + +test("resolveStateRoot uses CLAUDE_PLUGIN_DATA/state when set", () => { + const pluginData = makeTempDir(); + const prev = process.env.CLAUDE_PLUGIN_DATA; + process.env.CLAUDE_PLUGIN_DATA = pluginData; + try { + assert.equal(resolveStateRoot(), path.join(pluginData, "state")); + } finally { + if (prev == null) delete process.env.CLAUDE_PLUGIN_DATA; + else process.env.CLAUDE_PLUGIN_DATA = prev; + } +}); + +test("resolveStateRoot falls back to a tmp dir when CLAUDE_PLUGIN_DATA is unset", () => { + const prev = process.env.CLAUDE_PLUGIN_DATA; + delete process.env.CLAUDE_PLUGIN_DATA; + try { + assert.equal(resolveStateRoot(), path.join(os.tmpdir(), "codex-companion")); + } finally { + if (prev != null) process.env.CLAUDE_PLUGIN_DATA = prev; + } +}); + +test("resolveSessionId prefers explicit option, then env, then null", () => { + assert.equal(resolveSessionId({ sessionId: "explicit" }), "explicit"); + assert.equal(resolveSessionId({ env: { CODEX_COMPANION_SESSION_ID: "from-env" } }), "from-env"); +}); + +test("resolveSessionId reads process.env when no option/env given", () => { + const prev = process.env.CODEX_COMPANION_SESSION_ID; + process.env.CODEX_COMPANION_SESSION_ID = "proc-env"; + try { + assert.equal(resolveSessionId({}), "proc-env"); + } finally { + if (prev == null) delete process.env.CODEX_COMPANION_SESSION_ID; + else process.env.CODEX_COMPANION_SESSION_ID = prev; + } +}); + +test("resolveSessionId returns null when nothing is set", () => { + const prev = process.env.CODEX_COMPANION_SESSION_ID; + delete process.env.CODEX_COMPANION_SESSION_ID; + try { + assert.equal(resolveSessionId({ env: {} }), null); + } finally { + if (prev != null) process.env.CODEX_COMPANION_SESSION_ID = prev; + } +}); + +function writeBrokerJson(stateRoot, dirName, session) { + const dir = path.join(stateRoot, dirName); + fs.mkdirSync(dir, { recursive: true }); + const file = path.join(dir, "broker.json"); + fs.writeFileSync(file, JSON.stringify(session), "utf8"); + return file; +} + +function withPluginData(fn) { + const pluginData = makeTempDir(); + const prev = process.env.CLAUDE_PLUGIN_DATA; + process.env.CLAUDE_PLUGIN_DATA = pluginData; + return Promise.resolve(fn(pluginData)).finally(() => { + if (prev == null) delete process.env.CLAUDE_PLUGIN_DATA; + else process.env.CLAUDE_PLUGIN_DATA = prev; + }); +} + +async function withReadyBroker(fn) { + const sessionDir = makeTempDir(); + const endpoint = createBrokerEndpoint(sessionDir); + const target = parseBrokerEndpoint(endpoint); + const requests = []; + const server = net.createServer((socket) => { + socket.setEncoding("utf8"); + socket.on("data", (chunk) => { + requests.push(chunk); + socket.write(`${JSON.stringify({ id: 1, result: {} })}\n`); + socket.end(); + }); + }); + + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(target.path, () => { + server.off("error", reject); + resolve(); + }); + }); + + try { + return await fn({ endpoint, requests, sessionDir }); + } finally { + await new Promise((resolve) => server.close(resolve)); + } +} + +async function withHangingBroker(fn) { + const sessionDir = makeTempDir(); + const endpoint = createBrokerEndpoint(sessionDir); + const target = parseBrokerEndpoint(endpoint); + const requests = []; + const sockets = []; + const server = net.createServer((socket) => { + sockets.push(socket); + socket.setEncoding("utf8"); + socket.on("data", (chunk) => { + requests.push(chunk); + }); + }); + + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(target.path, () => { + server.off("error", reject); + resolve(); + }); + }); + + const destroySockets = () => { + for (const socket of sockets) { + socket.destroy(); + } + }; + + try { + return await fn({ endpoint, requests, sessionDir, destroySockets }); + } finally { + destroySockets(); + await new Promise((resolve) => server.close(resolve)); + } +} + +function waitForChild(child) { + return new Promise((resolve, reject) => { + child.once("error", reject); + child.once("exit", (code, signal) => { + resolve({ code, signal }); + }); + }); +} + +test("teardownBrokersForSession tears down a broker registered for a different cwd", async () => { + await withPluginData(async () => { + const stateRoot = stateRootForTest(); + const sessionDir = makeTempDir(); + const pidFile = path.join(sessionDir, "broker.pid"); + const logFile = path.join(sessionDir, "broker.log"); + fs.writeFileSync(pidFile, "12345\n"); + fs.writeFileSync(logFile, ""); + const brokerJson = writeBrokerJson(stateRoot, "worktree-deadbeefdeadbeef", { + endpoint: "unix:/tmp/codex-test-nonexistent.sock", + pidFile, logFile, sessionDir, pid: 12345, sessionId: "S" + }); + + const killed = []; + const count = await teardownBrokersForSession("S", { killProcess: (pid) => killed.push(pid) }); + + assert.equal(count, 1); + assert.deepEqual(killed, [12345]); + assert.equal(fs.existsSync(brokerJson), false); + }); +}); + +test("teardownBrokersForSession leaves non-matching sessionId brokers intact", async () => { + await withPluginData(async () => { + const stateRoot = stateRootForTest(); + const brokerJson = writeBrokerJson(stateRoot, "other-1111111111111111", { + endpoint: "unix:/tmp/codex-test-nonexistent2.sock", + pidFile: null, logFile: null, sessionDir: null, pid: null, sessionId: "S" + }); + const count = await teardownBrokersForSession("OTHER", { killProcess: () => {} }); + assert.equal(count, 0); + assert.equal(fs.existsSync(brokerJson), true); + }); +}); + +test("teardownBrokersForSession ignores broker.json without sessionId (legacy)", async () => { + await withPluginData(async () => { + const stateRoot = stateRootForTest(); + const brokerJson = writeBrokerJson(stateRoot, "legacy-2222222222222222", { + endpoint: "unix:/tmp/codex-test-nonexistent3.sock", pid: null + }); + const count = await teardownBrokersForSession("S", { killProcess: () => {} }); + assert.equal(count, 0); + assert.equal(fs.existsSync(brokerJson), true); + }); +}); + +test("teardownBrokersForSession is a no-op for empty sessionId", async () => { + await withPluginData(async () => { + const count = await teardownBrokersForSession("", { killProcess: () => {} }); + assert.equal(count, 0); + }); +}); + +test("reusing a ready broker transfers cleanup ownership to the later session", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint, requests, sessionDir }) => { + const cwd = makeTempDir(); + const pidFile = path.join(sessionDir, "broker.pid"); + const logFile = path.join(sessionDir, "broker.log"); + fs.writeFileSync(pidFile, "12345\n"); + fs.writeFileSync(logFile, ""); + saveBrokerSession(cwd, { + endpoint, + pidFile, + logFile, + sessionDir, + pid: 12345, + sessionId: "A" + }); + + const reused = await ensureBrokerSession(cwd, { env: { CODEX_COMPANION_SESSION_ID: "B" } }); + assert.equal(reused.endpoint, endpoint); + assert.deepEqual(loadBrokerSession(cwd).sessionIds, ["A", "B"]); + + const killed = []; + assert.equal(await teardownBrokersForSession("A", { killProcess: (pid) => killed.push(pid) }), 0); + assert.deepEqual(killed, []); + assert.equal(loadBrokerSession(cwd).sessionId, "B"); + assert.deepEqual(loadBrokerSession(cwd).sessionIds, ["B"]); + assert.equal(requests.length, 0); + + assert.equal(await teardownBrokersForSession("B", { killProcess: (pid) => killed.push(pid) }), 1); + assert.deepEqual(killed, [12345]); + assert.equal(loadBrokerSession(cwd), null); + assert.equal(requests.length, 1); + }); + }); +}); + +test("handleSessionEnd removes only the ending owner from a shared cwd broker", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint, requests, sessionDir }) => { + const cwd = makeTempDir(); + const pidFile = path.join(sessionDir, "broker.pid"); + const logFile = path.join(sessionDir, "broker.log"); + fs.writeFileSync(pidFile, "12345\n"); + fs.writeFileSync(logFile, ""); + saveBrokerSession(cwd, { + endpoint, + pidFile, + logFile, + sessionDir, + pid: 12345, + sessionId: "A", + sessionIds: ["A", "B"] + }); + + await handleSessionEnd({ cwd, session_id: "A" }); + + assert.equal(loadBrokerSession(cwd).sessionId, "B"); + assert.deepEqual(loadBrokerSession(cwd).sessionIds, ["B"]); + assert.equal(requests.length, 0); + }); + }); +}); + +test("concurrent SessionEnd hooks tear down a shared broker after the last owner exits", async () => { + await withPluginData(async (pluginData) => { + await withReadyBroker(async ({ endpoint, requests, sessionDir }) => { + const cwd = makeTempDir(); + const markerDir = makeTempDir(); + const pidFile = path.join(sessionDir, "broker.pid"); + const logFile = path.join(sessionDir, "broker.log"); + fs.writeFileSync(pidFile, "12345\n"); + fs.writeFileSync(logFile, ""); + const brokerJson = writeBrokerJson(stateRootForTest(), "worktree-race-deadbeef", { + endpoint, + pidFile, + logFile, + sessionDir, + pid: 12345, + sessionId: "A", + sessionIds: ["A", "B"] + }); + + const moduleUrl = pathToFileURL(path.resolve("plugins/codex/scripts/lib/broker-lifecycle.mjs")).href; + const script = ` + import fs from "node:fs"; + import path from "node:path"; + + const stateFile = process.env.TEST_BROKER_STATE_FILE; + const markerDir = process.env.TEST_MARKER_DIR; + const sessionId = process.env.TEST_SESSION_ID; + const otherSessionId = sessionId === "A" ? "B" : "A"; + const originalWriteFileSync = fs.writeFileSync.bind(fs); + + fs.writeFileSync = (file, data, ...args) => { + if (file === stateFile && String(data).includes('"sessionIds"')) { + originalWriteFileSync(path.join(markerDir, sessionId + ".ready"), "", "utf8"); + const otherReady = path.join(markerDir, otherSessionId + ".ready"); + const deadline = Date.now() + 500; + while (!fs.existsSync(otherReady) && Date.now() < deadline) {} + } + return originalWriteFileSync(file, data, ...args); + }; + + const { teardownBrokersForSession } = await import(process.env.TEST_BROKER_MODULE_URL); + await teardownBrokersForSession(sessionId, { killProcess: () => {} }); + `; + + const makeChild = (sessionId) => + spawn(process.execPath, ["--input-type=module", "-e", script], { + cwd: path.resolve("."), + env: { + ...process.env, + CLAUDE_PLUGIN_DATA: pluginData, + TEST_BROKER_STATE_FILE: brokerJson, + TEST_BROKER_MODULE_URL: moduleUrl, + TEST_MARKER_DIR: markerDir, + TEST_SESSION_ID: sessionId + }, + stdio: ["ignore", "pipe", "pipe"] + }); + + const childA = makeChild("A"); + const childB = makeChild("B"); + const [resultA, resultB] = await Promise.all([waitForChild(childA), waitForChild(childB)]); + + assert.deepEqual([resultA.code, resultB.code], [0, 0]); + assert.equal(fs.existsSync(brokerJson), false); + assert.equal(requests.length, 1); + }); + }); +}); + +test("handleSessionEnd still tears down session brokers when job cleanup fails", async () => { + await withPluginData(async () => { + const cwd = makeTempDir(); + const workspaceStateDir = resolveStateDir(cwd); + saveBrokerSession(cwd, { + endpoint: "unix:/tmp/codex-test-nonexistent-cleanup.sock", + pidFile: null, + logFile: null, + sessionDir: null, + pid: null, + sessionId: "S", + sessionIds: ["S"] + }); + const badLogFile = path.join(workspaceStateDir, "bad.log"); + fs.mkdirSync(badLogFile); + const stateFile = path.join(workspaceStateDir, "state.json"); + fs.writeFileSync( + stateFile, + `${JSON.stringify({ + version: 1, + config: { stopReviewGate: false }, + jobs: [{ id: "completed", status: "completed", sessionId: "S", logFile: badLogFile }] + }, null, 2)}\n`, + "utf8" + ); + + await assert.rejects(() => handleSessionEnd({ cwd, session_id: "S" }), { code: "EISDIR" }); + assert.equal(fs.existsSync(path.join(workspaceStateDir, "broker.json")), false); + }); +}); + +test("teardownBrokersForSession times out unresponsive broker shutdown requests", async () => { + await withPluginData(async () => { + await withHangingBroker(async ({ endpoint, requests, sessionDir, destroySockets }) => { + const stateRoot = stateRootForTest(); + const brokerJson = writeBrokerJson(stateRoot, "hanging-shutdown-deadbeef", { + endpoint, + pidFile: null, + logFile: null, + sessionDir, + pid: null, + sessionId: "S", + sessionIds: ["S"] + }); + + let completed = false; + const teardown = teardownBrokersForSession("S", { killProcess: () => {}, shutdownTimeoutMs: 50 }) + .then(() => { + completed = true; + }); + try { + await new Promise((resolve) => setTimeout(resolve, 150)); + assert.equal(completed, true); + } finally { + destroySockets(); + await teardown; + } + assert.equal(fs.existsSync(brokerJson), false); + assert.equal(requests.length, 1); + }); + }); +}); + +test("teardownBrokersForSession skips a same-session locked entry but tears down the rest", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint, requests, sessionDir }) => { + const stateRoot = stateRootForTest(); + + // Entry 1: owned by this session but its lock is held by a concurrent hook + // that never released it. + const lockedJson = writeBrokerJson(stateRoot, "worktree-lockedaaaaaaaa", { + endpoint: "unix:/tmp/codex-test-locked.sock", + pidFile: null, logFile: null, sessionDir: null, pid: null, sessionId: "S" + }); + fs.mkdirSync(`${lockedJson}.lock`); + + // Entry 2: a real broker owned by the same session, later in the scan. + const cleanableJson = writeBrokerJson(stateRoot, "worktree-cleanablebbbb", { + endpoint, pidFile: null, logFile: null, sessionDir, pid: null, sessionId: "S" + }); + + try { + const count = await teardownBrokersForSession("S", { killProcess: () => {}, lockTimeoutMs: 60 }); + // The locked entry is skipped (not aborted), the reachable one is cleaned. + assert.equal(count, 1); + assert.equal(fs.existsSync(cleanableJson), false); + assert.equal(fs.existsSync(lockedJson), true); + assert.equal(requests.length, 1); + } finally { + fs.rmSync(`${lockedJson}.lock`, { recursive: true, force: true }); + } + }); + }); +}); + +test("teardownBrokersForSession never waits on a lock for another session's workspace", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint, requests, sessionDir }) => { + const stateRoot = stateRootForTest(); + + // An unrelated workspace with a held lock, ordered before ours. + const otherJson = writeBrokerJson(stateRoot, "worktree-0000otheraaaa", { + endpoint: "unix:/tmp/codex-test-other.sock", + pidFile: null, logFile: null, sessionDir: null, pid: null, sessionId: "OTHER" + }); + fs.mkdirSync(`${otherJson}.lock`); + + const ourJson = writeBrokerJson(stateRoot, "worktree-9999oursbbbbb", { + endpoint, pidFile: null, logFile: null, sessionDir, pid: null, sessionId: "S" + }); + + try { + const startedAt = Date.now(); + // A large per-entry lock timeout would stall for seconds if we waited on + // the unrelated lock; the ownership pre-check must skip it outright. + const count = await teardownBrokersForSession("S", { killProcess: () => {}, lockTimeoutMs: 5000 }); + const elapsed = Date.now() - startedAt; + + assert.equal(count, 1); + assert.equal(fs.existsSync(ourJson), false); + assert.equal(fs.existsSync(otherJson), true); + assert.ok(elapsed < 1000, `expected fast teardown, took ${elapsed}ms`); + assert.equal(requests.length, 1); + } finally { + fs.rmSync(`${otherJson}.lock`, { recursive: true, force: true }); + } + }); + }); +}); + +test("teardownBrokersForSession reclaims a stale lock and still tears the broker down", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint, requests, sessionDir }) => { + const stateRoot = stateRootForTest(); + const brokerJson = writeBrokerJson(stateRoot, "worktree-stalelock1234", { + endpoint, pidFile: null, logFile: null, sessionDir, pid: null, sessionId: "S" + }); + + // A lock left behind by a crashed holder: present but aged well past the + // stale threshold, so acquisition must reclaim it instead of timing out. + const lockDir = `${brokerJson}.lock`; + fs.mkdirSync(lockDir); + const old = new Date(Date.now() - 120000); + fs.utimesSync(lockDir, old, old); + + const startedAt = Date.now(); + const count = await teardownBrokersForSession("S", { killProcess: () => {}, lockTimeoutMs: 2000 }); + const elapsed = Date.now() - startedAt; + + assert.equal(count, 1); + assert.equal(fs.existsSync(brokerJson), false); + assert.equal(requests.length, 1); + assert.ok(elapsed < 1000, `stale lock should be reclaimed promptly, took ${elapsed}ms`); + }); + }); +}); + +test("teardownBrokersForSession stops scanning once its time budget is exhausted", async () => { + await withPluginData(async () => { + const stateRoot = stateRootForTest(); + // Two same-session entries, both with held locks. With a tiny budget the + // scan must return promptly rather than spending lockTimeoutMs on each. + const a = writeBrokerJson(stateRoot, "worktree-budgetaaaaaaaa", { + endpoint: "unix:/tmp/codex-test-b1.sock", + pidFile: null, logFile: null, sessionDir: null, pid: null, sessionId: "S" + }); + const b = writeBrokerJson(stateRoot, "worktree-budgetbbbbbbbb", { + endpoint: "unix:/tmp/codex-test-b2.sock", + pidFile: null, logFile: null, sessionDir: null, pid: null, sessionId: "S" + }); + fs.mkdirSync(`${a}.lock`); + fs.mkdirSync(`${b}.lock`); + + try { + const startedAt = Date.now(); + await teardownBrokersForSession("S", { killProcess: () => {}, lockTimeoutMs: 5000, budgetMs: 200 }); + const elapsed = Date.now() - startedAt; + assert.ok(elapsed < 1500, `expected budget-bounded scan, took ${elapsed}ms`); + } finally { + fs.rmSync(`${a}.lock`, { recursive: true, force: true }); + fs.rmSync(`${b}.lock`, { recursive: true, force: true }); + } + }); +}); + +test("ensureBrokerSession spawns and persists a fresh broker when the recorded one is dead", async () => { + await withPluginData(async () => { + const cwd = makeTempDir(); + saveBrokerSession(cwd, { + endpoint: "unix:/tmp/codex-test-dead-spawn.sock", + pidFile: null, + logFile: null, + sessionDir: null, + pid: null, + sessionId: "A", + sessionIds: ["A"] + }); + + const session = await ensureBrokerSession(cwd, { + env: { CODEX_COMPANION_SESSION_ID: "B" }, + scriptPath: writeFakeBrokerScript() + }); + try { + assert.ok(session); + assert.notEqual(session.endpoint, "unix:/tmp/codex-test-dead-spawn.sock"); + assert.deepEqual(session.sessionIds, ["B"]); + // The freshly spawned broker is the one persisted (no orphan record). + assert.equal(loadBrokerSession(cwd).endpoint, session.endpoint); + assert.equal(await new Promise((resolve) => { + const socket = net.createConnection({ path: parseBrokerEndpoint(session.endpoint).path }); + socket.on("connect", () => { socket.end(); resolve(true); }); + socket.on("error", () => resolve(false)); + }), true); + } finally { + if (session?.pid) { + try { + process.kill(session.pid); + } catch { + // Already gone. + } + } + } + }); +}); + +test("ensureBrokerSession does not resurrect a broker torn down while waiting for the state lock", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint, sessionDir }) => { + const cwd = makeTempDir(); + saveBrokerSession(cwd, { + endpoint, + pidFile: null, + logFile: null, + sessionDir, + pid: null, + sessionId: "A", + sessionIds: ["A"] + }); + const stateFile = path.join(resolveStateDir(cwd), "broker.json"); + fs.mkdirSync(`${stateFile}.lock`); + + const pending = ensureBrokerSession(cwd, { + env: { CODEX_COMPANION_SESSION_ID: "B" }, + scriptPath: writeFakeBrokerScript() + }); + + // While B is parked on the lock (readiness probe already passed), A's + // SessionEnd shuts the broker down and removes broker.json. + await sleep(200); + fs.unlinkSync(stateFile); + fs.rmSync(parseBrokerEndpoint(endpoint).path, { force: true }); + fs.rmdirSync(`${stateFile}.lock`); + + const session = await pending; + try { + assert.ok(session); + assert.notEqual(session.endpoint, endpoint); + assert.deepEqual(session.sessionIds, ["B"]); + assert.equal(loadBrokerSession(cwd).endpoint, session.endpoint); + } finally { + if (session?.pid) { + try { + process.kill(session.pid); + } catch { + // Already gone. + } + } + } + }); + }); +}); + +test("ensureBrokerSession reuses a live replacement broker after losing the lock race", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint: staleEndpoint, sessionDir: staleDir }) => { + await withReadyBroker(async ({ endpoint: freshEndpoint, sessionDir: freshDir }) => { + const cwd = makeTempDir(); + saveBrokerSession(cwd, { + endpoint: staleEndpoint, + pidFile: null, + logFile: null, + sessionDir: staleDir, + pid: null, + sessionId: "A", + sessionIds: ["A"] + }); + const stateFile = path.join(resolveStateDir(cwd), "broker.json"); + fs.mkdirSync(`${stateFile}.lock`); + + const pending = ensureBrokerSession(cwd, { env: { CODEX_COMPANION_SESSION_ID: "B" } }); + + // While B waits, the stale broker is replaced by a different live one. + await sleep(200); + saveBrokerSession(cwd, { + endpoint: freshEndpoint, + pidFile: null, + logFile: null, + sessionDir: freshDir, + pid: null, + sessionId: "C", + sessionIds: ["C"] + }); + fs.rmdirSync(`${stateFile}.lock`); + + const session = await pending; + assert.equal(session.endpoint, freshEndpoint); + assert.deepEqual(session.sessionIds, ["C", "B"]); + assert.deepEqual(loadBrokerSession(cwd).sessionIds, ["C", "B"]); + }); + }); + }); +}); + +test("teardownBrokersForSession tolerates an unparseable broker.json and keeps cleaning later brokers", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint, requests, sessionDir }) => { + const stateRoot = stateRootForTest(); + + // A corrupt/half-written broker.json: the unlocked ownership pre-check + // must not treat a parse failure as "not ours" and abort or skip the + // rest of the scan. + const corruptDir = path.join(stateRoot, "worktree-badjson00000"); + fs.mkdirSync(corruptDir, { recursive: true }); + fs.writeFileSync(path.join(corruptDir, "broker.json"), "{ not valid json", "utf8"); + + const goodJson = writeBrokerJson(stateRoot, "worktree-goodjson11111", { + endpoint, pidFile: null, logFile: null, sessionDir, pid: null, sessionId: "S" + }); + + const count = await teardownBrokersForSession("S", { killProcess: () => {}, lockTimeoutMs: 200 }); + + assert.equal(count, 1); + assert.equal(fs.existsSync(goodJson), false); + assert.equal(requests.length, 1); + }); + }); +}); + +test("saveBrokerSession writes broker.json atomically", async () => { + await withPluginData(async () => { + const cwd = makeTempDir(); + saveBrokerSession(cwd, { endpoint: "unix:/tmp/x.sock", sessionId: "S", sessionIds: ["S"] }); + const dir = resolveStateDir(cwd); + // No leftover temp files, and the persisted file parses cleanly. + const leftovers = fs.readdirSync(dir).filter((name) => name.includes("broker.json.tmp")); + assert.deepEqual(leftovers, []); + assert.equal(loadBrokerSession(cwd).sessionId, "S"); + }); +}); + +test("teardownBrokersForSession tolerates a corrupt endpoint and keeps cleaning later brokers", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint, requests, sessionDir }) => { + const stateRoot = stateRootForTest(); + + // A stale record with an unsupported endpoint: sendBrokerShutdown would + // throw synchronously on it. It must be torn down best-effort, not abort + // the scan. + const corruptPid = 999999999; // non-existent; killProcess is mocked below + const corruptJson = writeBrokerJson(stateRoot, "worktree-corrupt00000", { + endpoint: "garbage-endpoint", pidFile: null, logFile: null, + sessionDir: null, pid: corruptPid, sessionId: "S" + }); + + // A healthy broker owned by the same session, later in the scan. + const goodJson = writeBrokerJson(stateRoot, "worktree-goodendpoint1", { + endpoint, pidFile: null, logFile: null, sessionDir, pid: null, sessionId: "S" + }); + + const killed = []; + const count = await teardownBrokersForSession("S", { killProcess: (pid) => killed.push(pid) }); + + assert.equal(count, 2); + assert.equal(fs.existsSync(corruptJson), false); + assert.equal(fs.existsSync(goodJson), false); + assert.ok(killed.includes(corruptPid)); + assert.equal(requests.length, 1); + }); + }); +}); + +test("handleSessionEnd falls through to cwd teardown when session teardown is skipped under lock contention", async () => { + await withPluginData(async () => { + await withReadyBroker(async ({ endpoint, requests, sessionDir }) => { + const cwd = makeTempDir(); + saveBrokerSession(cwd, { + endpoint, pidFile: null, logFile: null, sessionDir, pid: null, + sessionId: "S", sessionIds: ["S"] + }); + // Hold the broker's lock so the session-keyed teardown skips it; the + // record then still lists the ending session as its only owner. + const lockDir = `${path.join(resolveStateDir(cwd), "broker.json")}.lock`; + fs.mkdirSync(lockDir); + + try { + await handleSessionEnd({ cwd, session_id: "S" }); + // The cwd fallback (which does not take the lock) must still tear it + // down rather than leaving a broker owned only by the ended session. + assert.equal(loadBrokerSession(cwd), null); + assert.equal(requests.length, 1); + } finally { + fs.rmSync(lockDir, { recursive: true, force: true }); + } + }); + }); +}); + +test("teardownBrokersForSession caps shutdown waits to the remaining budget", async () => { + await withPluginData(async () => { + await withHangingBroker(async ({ endpoint, sessionDir, destroySockets }) => { + const stateRoot = stateRootForTest(); + writeBrokerJson(stateRoot, "worktree-hang1aaaaaaaaa", { + endpoint, pidFile: null, logFile: null, sessionDir, pid: null, + sessionId: "S", sessionIds: ["S"] + }); + writeBrokerJson(stateRoot, "worktree-hang2bbbbbbbbb", { + endpoint, pidFile: null, logFile: null, sessionDir: null, pid: null, + sessionId: "S", sessionIds: ["S"] + }); + + try { + const startedAt = Date.now(); + // Endpoints accept but never reply; a per-RPC 1s wait on each would blow + // the budget. The scan must honor budgetMs across shutdown waits. + await teardownBrokersForSession("S", { + killProcess: () => {}, + shutdownTimeoutMs: 1000, + budgetMs: 300 + }); + const elapsed = Date.now() - startedAt; + assert.ok(elapsed < 900, `expected budget-capped shutdown, took ${elapsed}ms`); + } finally { + destroySockets(); + } + }); + }); +}); + +test("handleSessionEnd tears down broker even when cwd mismatches (regression #380)", async () => { + await withPluginData(async () => { + const stateRoot = stateRootForTest(); + const sessionDir = makeTempDir(); + const pidFile = path.join(sessionDir, "broker.pid"); + fs.writeFileSync(pidFile, "999999999\n"); // non-existent pid, harmless to signal + const brokerJson = writeBrokerJson(stateRoot, "worktree-33333333deadbeef", { + endpoint: "unix:/tmp/codex-test-nonexistent4.sock", + pidFile, logFile: null, sessionDir, pid: 999999999, sessionId: "S" + }); + + // cwd is a DIFFERENT path than the broker's workspace — the cwd-based path + // would miss; the session-based path must still tear it down. + await handleSessionEnd({ cwd: makeTempDir(), session_id: "S" }); + + assert.equal(fs.existsSync(brokerJson), false); + }); +}); diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 8f276835..af99274a 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -698,6 +698,38 @@ test("session start hook exports the Claude session id, transcript path, and plu ); }); +test("session start hook runs when invoked through a symlinked plugin root", () => { + const repo = makeTempDir(); + const linkParent = makeTempDir(); + const pluginLink = path.join(linkParent, "codex-link"); + fs.symlinkSync(PLUGIN_ROOT, pluginLink, process.platform === "win32" ? "junction" : "dir"); + + const envFile = path.join(makeTempDir(), "claude-env.sh"); + fs.writeFileSync(envFile, "", "utf8"); + const pluginDataDir = makeTempDir(); + const symlinkedSessionHook = path.join(pluginLink, "scripts", "session-lifecycle-hook.mjs"); + + const result = run("node", [symlinkedSessionHook, "SessionStart"], { + cwd: repo, + env: { + ...process.env, + CLAUDE_ENV_FILE: envFile, + CLAUDE_PLUGIN_DATA: pluginDataDir + }, + input: JSON.stringify({ + hook_event_name: "SessionStart", + session_id: "sess-symlink", + cwd: repo + }) + }); + + assert.equal(result.status, 0, result.stderr); + assert.equal( + fs.readFileSync(envFile, "utf8"), + `export CODEX_COMPANION_SESSION_ID='sess-symlink'\nexport CLAUDE_PLUGIN_DATA='${pluginDataDir}'\n` + ); +}); + test("write task output focuses on the Codex result without generic follow-up hints", () => { const repo = makeTempDir(); const binDir = makeTempDir();