From a01a9f8bcf786dd58f2c6a2acdc3871bc1ca805d Mon Sep 17 00:00:00 2001 From: SP Son <1376128+seungpyoson@users.noreply.github.com> Date: Tue, 25 Aug 2026 02:03:03 +0900 Subject: [PATCH 1/3] fix: make app-server connection loss terminal --- plugins/codex/scripts/app-server-broker.mjs | 34 ++- plugins/codex/scripts/lib/app-server.mjs | 238 ++++++++++++------ .../codex/scripts/lib/broker-lifecycle.mjs | 3 +- plugins/codex/scripts/lib/codex.mjs | 12 + plugins/codex/scripts/lib/process.mjs | 23 +- tests/app-server-lifecycle.test.mjs | 220 ++++++++++++++++ tests/broker-lifecycle.test.mjs | 45 ++++ tests/fake-codex-fixture.mjs | 60 +++++ tests/helpers.mjs | 1 + tests/process.test.mjs | 37 +++ tests/runtime.test.mjs | 147 ++++++++++- 11 files changed, 720 insertions(+), 100 deletions(-) create mode 100644 tests/app-server-lifecycle.test.mjs create mode 100644 tests/broker-lifecycle.test.mjs diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 1954274fe..14e39c012 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -70,6 +70,7 @@ async function main() { let activeStreamSocket = null; let activeStreamThreadIds = null; const sockets = new Set(); + let shutdownPromise = null; function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { @@ -99,18 +100,25 @@ async function main() { } } - async function shutdown(server) { - 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); - } - if (pidFile && fs.existsSync(pidFile)) { - fs.unlinkSync(pidFile); + function shutdown(server) { + if (shutdownPromise) { + return shutdownPromise; } + shutdownPromise = Promise.resolve().then(async () => { + const serverClosed = new Promise((resolve) => server.close(resolve)); + for (const socket of sockets) { + socket.destroy(); + } + await appClient.close().catch(() => {}); + await serverClosed; + if (listenTarget.kind === "unix" && fs.existsSync(listenTarget.path)) { + fs.unlinkSync(listenTarget.path); + } + if (pidFile && fs.existsSync(pidFile)) { + fs.unlinkSync(pidFile); + } + }); + return shutdownPromise; } appClient.setNotificationHandler(routeNotification); @@ -233,6 +241,10 @@ async function main() { }); }); + appClient.onTerminal(() => { + void shutdown(server).finally(() => process.exit(1)); + }); + process.on("SIGTERM", async () => { await shutdown(server); process.exit(0); diff --git a/plugins/codex/scripts/lib/app-server.mjs b/plugins/codex/scripts/lib/app-server.mjs index 72b30a764..27bd2aee1 100644 --- a/plugins/codex/scripts/lib/app-server.mjs +++ b/plugins/codex/scripts/lib/app-server.mjs @@ -22,6 +22,11 @@ const PLUGIN_MANIFEST = JSON.parse(fs.readFileSync(PLUGIN_MANIFEST_URL, "utf8")) export const BROKER_ENDPOINT_ENV = "CODEX_COMPANION_APP_SERVER_ENDPOINT"; export const BROKER_BUSY_RPC_CODE = -32001; +const CHILD_STDIN_GRACE_MS = 1000; +const CHILD_TERMINATION_GRACE_MS = 1000; +const APP_SERVER_INITIALIZE_TIMEOUT_MS = 15000; +const BROKER_SOCKET_CLOSE_GRACE_MS = 1000; + /** @type {ClientInfo} */ const DEFAULT_CLIENT_INFO = { title: "Codex Plugin", @@ -54,6 +59,16 @@ function createProtocolError(message, data) { return error; } +function settlesWithin(promise, timeoutMs) { + return new Promise((resolve) => { + const timer = setTimeout(() => resolve(false), timeoutMs); + promise.then(() => { + clearTimeout(timer); + resolve(true); + }); + }); +} + class AppServerClientBase { constructor(cwd, options = {}) { this.cwd = cwd; @@ -61,15 +76,15 @@ class AppServerClientBase { this.pending = new Map(); this.nextId = 1; this.stderr = ""; - this.closed = false; - this.exitError = null; + this.terminalCause = null; + this.terminalListeners = new Set(); /** @type {AppServerNotificationHandler | null} */ this.notificationHandler = null; this.lineBuffer = ""; this.transport = "unknown"; - this.exitPromise = new Promise((resolve) => { - this.resolveExit = resolve; + this.terminalPromise = new Promise((resolve) => { + this.resolveTerminal = resolve; }); } @@ -77,6 +92,15 @@ class AppServerClientBase { this.notificationHandler = handler; } + onTerminal(listener) { + if (this.terminalCause) { + listener(this.terminalCause); + return () => {}; + } + this.terminalListeners.add(listener); + return () => this.terminalListeners.delete(listener); + } + /** * @template {AppServerMethod} M * @param {M} method @@ -84,8 +108,8 @@ class AppServerClientBase { * @returns {Promise>} */ request(method, params) { - if (this.closed) { - throw new Error("codex app-server client is closed."); + if (this.terminalCause) { + throw this.terminalCause; } const id = this.nextId; @@ -98,12 +122,27 @@ class AppServerClientBase { } notify(method, params = {}) { - if (this.closed) { + if (this.terminalCause) { return; } this.sendMessage({ method, params }); } + async initializeProtocol() { + const initializeTimeout = setTimeout(() => { + this.transitionToTerminal(createProtocolError("codex app-server initialization timed out.")); + }, APP_SERVER_INITIALIZE_TIMEOUT_MS); + try { + await this.request("initialize", { + clientInfo: this.options.clientInfo ?? DEFAULT_CLIENT_INFO, + capabilities: this.options.capabilities ?? DEFAULT_CAPABILITIES + }); + } finally { + clearTimeout(initializeTimeout); + } + this.notify("initialized", {}); + } + handleChunk(chunk) { this.lineBuffer += chunk; let newlineIndex = this.lineBuffer.indexOf("\n"); @@ -116,6 +155,9 @@ class AppServerClientBase { } handleLine(line) { + if (this.terminalCause) { + return; + } if (!line.trim()) { return; } @@ -124,7 +166,12 @@ class AppServerClientBase { try { message = JSON.parse(line); } catch (error) { - this.handleExit(createProtocolError(`Failed to parse codex app-server JSONL: ${error.message}`, { line })); + this.transitionToTerminal(createProtocolError(`Failed to parse codex app-server JSONL: ${error.message}`, { line })); + return; + } + + if (typeof message !== "object" || message === null || Array.isArray(message)) { + this.transitionToTerminal(createProtocolError("Invalid codex app-server JSONL message: expected an object.", { line })); return; } @@ -149,7 +196,11 @@ class AppServerClientBase { } if (message.method && this.notificationHandler) { - this.notificationHandler(/** @type {AppServerNotification} */ (message)); + try { + this.notificationHandler(/** @type {AppServerNotification} */ (message)); + } catch (error) { + this.transitionToTerminal(error); + } } } @@ -160,23 +211,39 @@ class AppServerClientBase { }); } - handleExit(error) { - if (this.exitResolved) { + transitionToTerminal(error) { + if (this.terminalCause) { return; } - this.exitResolved = true; - this.exitError = error ?? null; + this.terminalCause = error ?? new Error("codex app-server connection closed."); for (const pending of this.pending.values()) { - pending.reject(this.exitError ?? new Error("codex app-server connection closed.")); + pending.reject(this.terminalCause); } this.pending.clear(); - this.resolveExit(undefined); + for (const listener of this.terminalListeners) { + listener(this.terminalCause); + } + this.terminalListeners.clear(); + this.resolveTerminal(undefined); + } + + sendMessage(message) { + if (this.terminalCause) { + return false; + } + try { + this.writeMessage(message); + return true; + } catch (error) { + this.transitionToTerminal(error); + return false; + } } - sendMessage(_message) { - throw new Error("sendMessage must be implemented by subclasses."); + writeMessage(_message) { + throw new Error("writeMessage must be implemented by subclasses."); } } @@ -184,6 +251,7 @@ class SpawnedCodexAppServerClient extends AppServerClientBase { constructor(cwd, options = {}) { super(cwd, options); this.transport = "direct"; + this.closePromise = null; } async initialize() { @@ -192,18 +260,32 @@ class SpawnedCodexAppServerClient extends AppServerClientBase { env: this.options.env ?? process.env, stdio: ["pipe", "pipe", "pipe"], shell: process.platform === "win32" ? (process.env.SHELL || true) : false, + detached: process.platform !== "win32", windowsHide: true }); + this.childExitedPromise = new Promise((resolve) => { + this.proc.once("close", resolve); + }); + this.proc.stdout.setEncoding("utf8"); this.proc.stderr.setEncoding("utf8"); this.proc.stderr.on("data", (chunk) => { this.stderr += chunk; }); + this.proc.stdin.on("error", (error) => { + this.transitionToTerminal(error); + }); + + this.proc.stdout.on("end", () => { + if (!this.terminalCause) { + this.transitionToTerminal(createProtocolError("codex app-server stdout closed before the connection ended.")); + } + }); this.proc.on("error", (error) => { - this.handleExit(error); + this.transitionToTerminal(error); }); this.proc.on("exit", (code, signal) => { @@ -214,7 +296,7 @@ class SpawnedCodexAppServerClient extends AppServerClientBase { : createProtocolError( `codex app-server exited unexpectedly (${signal ? `signal ${signal}` : `exit ${code}`}).${stderr ? `\n${stderr}` : ""}` ); - this.handleExit(detail); + this.transitionToTerminal(detail); }); this.readline = readline.createInterface({ input: this.proc.stdout }); @@ -222,50 +304,47 @@ class SpawnedCodexAppServerClient extends AppServerClientBase { this.handleLine(line); }); - await this.request("initialize", { - clientInfo: this.options.clientInfo ?? DEFAULT_CLIENT_INFO, - capabilities: this.options.capabilities ?? DEFAULT_CAPABILITIES - }); - this.notify("initialized", {}); + await this.initializeProtocol(); } async close() { - if (this.closed) { - await this.exitPromise; - return; + if (this.closePromise) { + return this.closePromise; } + this.closePromise = Promise.resolve().then(async () => { + if (this.readline) { + this.readline.close(); + } - this.closed = true; - - if (this.readline) { - this.readline.close(); - } + if (!this.proc || !this.childExitedPromise) { + return; + } + if (this.proc.exitCode === null && !this.proc.killed) { + this.proc.stdin.end(); + } + if (await settlesWithin(this.childExitedPromise, CHILD_STDIN_GRACE_MS)) { + return; + } - if (this.proc && !this.proc.killed) { - this.proc.stdin.end(); - setTimeout(() => { - if (this.proc && !this.proc.killed && this.proc.exitCode === null) { - // On Windows with shell: true, the direct child is cmd.exe. - // Use terminateProcessTree to kill the entire tree including - // the grandchild node process. - if (process.platform === "win32") { - try { - terminateProcessTree(this.proc.pid); - } catch { - // Best-effort cleanup inside an unref'd timer — swallow errors - // to avoid crashing the host process during shutdown. - } - } else { - this.proc.kill("SIGTERM"); - } + try { + terminateProcessTree(this.proc.pid); + } catch { + // The child may have exited between the grace deadline and termination. + } + if (!(await settlesWithin(this.childExitedPromise, CHILD_TERMINATION_GRACE_MS))) { + try { + terminateProcessTree(this.proc.pid, { signal: "SIGKILL" }); + } catch { + // The process tree may have exited during escalation. } - }, 50).unref?.(); - } - - await this.exitPromise; + } + await this.childExitedPromise; + }); + this.transitionToTerminal(new Error("codex app-server client is closed.")); + return this.closePromise; } - sendMessage(message) { + writeMessage(message) { const line = `${JSON.stringify(message)}\n`; const stdin = this.proc?.stdin; if (!stdin) { @@ -280,49 +359,55 @@ class BrokerCodexAppServerClient extends AppServerClientBase { super(cwd, options); this.transport = "broker"; this.endpoint = options.brokerEndpoint; + this.closePromise = null; } async initialize() { await new Promise((resolve, reject) => { const target = parseBrokerEndpoint(this.endpoint); this.socket = net.createConnection({ path: target.path }); + this.socketClosedPromise = new Promise((resolve) => { + this.socket.once("close", resolve); + }); this.socket.setEncoding("utf8"); this.socket.on("connect", resolve); this.socket.on("data", (chunk) => { this.handleChunk(chunk); }); this.socket.on("error", (error) => { - if (!this.exitResolved) { + if (!this.terminalCause) { reject(error); } - this.handleExit(error); + this.transitionToTerminal(error); }); this.socket.on("close", () => { - this.handleExit(this.exitError); + this.transitionToTerminal(this.terminalCause); }); }); - await this.request("initialize", { - clientInfo: this.options.clientInfo ?? DEFAULT_CLIENT_INFO, - capabilities: this.options.capabilities ?? DEFAULT_CAPABILITIES - }); - this.notify("initialized", {}); + await this.initializeProtocol(); } async close() { - if (this.closed) { - await this.exitPromise; - return; - } - - this.closed = true; - if (this.socket) { - this.socket.end(); + if (this.closePromise) { + return this.closePromise; } - await this.exitPromise; + this.closePromise = Promise.resolve().then(async () => { + if (this.socket && !this.socket.destroyed) { + this.socket.end(); + } + if (this.socketClosedPromise && !(await settlesWithin(this.socketClosedPromise, BROKER_SOCKET_CLOSE_GRACE_MS))) { + this.socket.destroy(); + } + if (this.socketClosedPromise) { + await this.socketClosedPromise; + } + }); + this.transitionToTerminal(new Error("codex app-server client is closed.")); + return this.closePromise; } - sendMessage(message) { + writeMessage(message) { const line = `${JSON.stringify(message)}\n`; const socket = this.socket; if (!socket) { @@ -348,7 +433,12 @@ export class CodexAppServerClient { const client = brokerEndpoint ? new BrokerCodexAppServerClient(cwd, { ...options, brokerEndpoint }) : new SpawnedCodexAppServerClient(cwd, options); - await client.initialize(); - return client; + try { + await client.initialize(); + return client; + } catch (error) { + await client.close().catch(() => {}); + throw error; + } } } diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index ef763819c..962d7eba3 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -6,6 +6,7 @@ import process from "node:process"; import { spawn } from "node:child_process"; import { fileURLToPath } from "node:url"; import { createBrokerEndpoint, parseBrokerEndpoint } from "./broker-endpoint.mjs"; +import { terminateProcessTree } from "./process.mjs"; import { resolveStateDir } from "./state.mjs"; export const PID_FILE_ENV = "CODEX_COMPANION_APP_SERVER_PID_FILE"; @@ -154,7 +155,7 @@ export async function ensureBrokerSession(cwd, options = {}) { logFile, sessionDir, pid: child.pid ?? null, - killProcess: options.killProcess ?? null + killProcess: options.killProcess ?? terminateProcessTree }); return null; } diff --git a/plugins/codex/scripts/lib/codex.mjs b/plugins/codex/scripts/lib/codex.mjs index fead00cc4..ac116bfcc 100644 --- a/plugins/codex/scripts/lib/codex.mjs +++ b/plugins/codex/scripts/lib/codex.mjs @@ -307,6 +307,7 @@ function createTurnCaptureState(threadId, options = {}) { resolveCompletion = resolve; rejectCompletion = reject; }); + void completion.catch(() => {}); return { threadId, @@ -370,6 +371,15 @@ function completeTurn(state, turn = null, options = {}) { state.resolveCompletion(state); } +function failTurn(state, error) { + if (state.completed) { + return; + } + clearCompletionTimer(state); + state.completed = true; + state.rejectCompletion(error); +} + function scheduleInferredCompletion(state) { if (state.completed || state.finalTurn || !state.finalAnswerSeen) { return; @@ -559,6 +569,7 @@ function applyTurnNotification(state, message) { async function captureTurn(client, threadId, startRequest, options = {}) { const state = createTurnCaptureState(threadId, options); const previousHandler = client.notificationHandler; + const removeTerminalListener = client.onTerminal((error) => failTurn(state, error)); client.setNotificationHandler((message) => { if (!state.turnId) { @@ -606,6 +617,7 @@ async function captureTurn(client, threadId, startRequest, options = {}) { return await state.completion; } finally { clearCompletionTimer(state); + removeTerminalListener(); client.setNotificationHandler(previousHandler ?? null); } } diff --git a/plugins/codex/scripts/lib/process.mjs b/plugins/codex/scripts/lib/process.mjs index dd8fc3751..7a3ce5a8c 100644 --- a/plugins/codex/scripts/lib/process.mjs +++ b/plugins/codex/scripts/lib/process.mjs @@ -62,6 +62,7 @@ export function terminateProcessTree(pid, options = {}) { const platform = options.platform ?? process.platform; const runCommandImpl = options.runCommandImpl ?? runCommand; const killImpl = options.killImpl ?? process.kill.bind(process); + const signal = options.signal ?? "SIGTERM"; if (platform === "win32") { const result = runCommandImpl("taskkill", ["/PID", String(pid), "/T", "/F"], { @@ -98,22 +99,18 @@ export function terminateProcessTree(pid, options = {}) { } try { - killImpl(-pid, "SIGTERM"); + killImpl(-pid, signal); return { attempted: true, delivered: true, method: "process-group" }; - } catch (error) { - if (error?.code !== "ESRCH") { - try { - killImpl(pid, "SIGTERM"); - return { attempted: true, delivered: true, method: "process" }; - } catch (innerError) { - if (innerError?.code === "ESRCH") { - return { attempted: true, delivered: false, method: "process" }; - } - throw innerError; + } catch { + try { + killImpl(pid, signal); + return { attempted: true, delivered: true, method: "process" }; + } catch (error) { + if (error?.code === "ESRCH") { + return { attempted: true, delivered: false, method: "process" }; } + throw error; } - - return { attempted: true, delivered: false, method: "process-group" }; } } diff --git a/tests/app-server-lifecycle.test.mjs b/tests/app-server-lifecycle.test.mjs new file mode 100644 index 000000000..6ecac2540 --- /dev/null +++ b/tests/app-server-lifecycle.test.mjs @@ -0,0 +1,220 @@ +import assert from "node:assert/strict"; +import fs from "node:fs"; +import path from "node:path"; +import process from "node:process"; +import test from "node:test"; + +import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; +import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs"; +import { makeTempDir } from "./helpers.mjs"; + +function settleWithin(promise, timeoutMs = 250) { + let timer; + return Promise.race([ + promise.then( + (value) => ({ status: "fulfilled", value }), + (reason) => ({ status: "rejected", reason }) + ), + new Promise((resolve) => { + timer = setTimeout(() => resolve({ status: "timeout" }), timeoutMs); + }) + ]).finally(() => clearTimeout(timer)); +} + +function processIsAlive(pid) { + try { + process.kill(pid, 0); + return true; + } catch (error) { + if (error?.code === "ESRCH") return false; + throw error; + } +} + +test("a request made after the app-server child exits rejects instead of hanging", async () => { + const binDir = makeTempDir("codex-plugin-lifecycle-"); + installFakeCodex(binDir, "exit-after-initialize"); + + const client = await CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }); + + await client.terminalPromise; + const outcome = await settleWithin( + Promise.resolve().then(() => client.request("account/read", { refreshToken: false })) + ); + + assert.equal(outcome.status, "rejected"); + assert.match(outcome.reason.message, /connection closed|exited|stdout closed/i); + await client.close(); +}); + +test("a JSON null protocol line rejects initialization instead of crashing the host", async () => { + const binDir = makeTempDir("codex-plugin-null-line-"); + installFakeCodex(binDir, "null-on-initialize"); + + await assert.rejects( + CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }), + /invalid codex app-server JSONL message/i + ); +}); + +test("protocol EOF rejects initialization and reaps the live child", { skip: process.platform === "win32" }, async () => { + const binDir = makeTempDir("codex-plugin-stdout-eof-"); + const statePath = path.join(binDir, "fake-codex-state.json"); + installFakeCodex(binDir, "stdout-eof-on-initialize"); + let pid = null; + + try { + const outcome = await settleWithin( + CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }), + 2000 + ); + pid = JSON.parse(fs.readFileSync(statePath, "utf8")).pid; + + assert.equal(outcome.status, "rejected"); + assert.match(outcome.reason.message, /stdout closed|connection closed/i); + assert.equal(processIsAlive(pid), false); + } finally { + if (pid && processIsAlive(pid)) { + process.kill(pid, "SIGKILL"); + } + } +}); + +test("a broken app-server stdin becomes the single terminal cause", { skip: process.platform === "win32" }, async () => { + const binDir = makeTempDir("codex-plugin-stdin-eof-"); + installFakeCodex(binDir, "stdin-eof-after-initialize"); + const client = await CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }); + + try { + const outcome = await settleWithin(client.terminalPromise, 2000); + assert.equal(outcome.status, "fulfilled"); + assert.match(client.terminalCause.message, /EPIPE|broken pipe|write/i); + assert.throws(() => client.request("account/read", {}), (error) => error === client.terminalCause); + } finally { + await client.close(); + } +}); + +test("close is idempotent when terminal listeners re-enter teardown", async () => { + const binDir = makeTempDir("codex-plugin-close-reentry-"); + installFakeCodex(binDir); + const client = await CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }); + let listenerClose = null; + client.onTerminal(() => { + listenerClose = client.close(); + }); + + await client.close(); + await listenerClose; + assert.equal(processIsAlive(client.proc.pid), false); +}); + +test("close lets a cooperative app-server finish before escalation", { skip: process.platform === "win32" }, async () => { + const binDir = makeTempDir("codex-plugin-clean-exit-"); + installFakeCodex(binDir, "slow-clean-exit"); + const client = await CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }); + const exited = new Promise((resolve) => { + client.proc.once("exit", (code, signal) => resolve({ code, signal })); + }); + + await client.close(); + assert.deepEqual(await exited, { code: 0, signal: null }); +}); + +test("initialization failure reaps the owned app-server child before rejecting", async () => { + const binDir = makeTempDir("codex-plugin-init-reap-"); + const statePath = path.join(binDir, "fake-codex-state.json"); + installFakeCodex(binDir, "malformed-on-initialize"); + let pid = null; + + try { + await assert.rejects( + CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }), + /Failed to parse codex app-server JSONL/ + ); + pid = JSON.parse(fs.readFileSync(statePath, "utf8")).pid; + assert.equal(processIsAlive(pid), false); + } finally { + if (pid && processIsAlive(pid)) { + process.kill(pid, "SIGKILL"); + } + } +}); + +test("forced teardown reaps an uncooperative launcher and its descendant", { skip: process.platform === "win32" }, async () => { + const binDir = makeTempDir("codex-plugin-force-reap-"); + const statePath = path.join(binDir, "fake-codex-state.json"); + installFakeCodex(binDir, "uncooperative-tree-on-initialize"); + let launcherPid = null; + let descendantPid = null; + + try { + const outcome = await settleWithin( + CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }), + 3000 + ); + const state = JSON.parse(fs.readFileSync(statePath, "utf8")); + launcherPid = state.pid; + descendantPid = state.descendantPid; + + assert.equal(outcome.status, "rejected"); + assert.match(outcome.reason.message, /failed to parse/i); + assert.equal(processIsAlive(launcherPid), false); + assert.equal(processIsAlive(descendantPid), false); + } finally { + for (const pid of [launcherPid, descendantPid]) { + if (pid && processIsAlive(pid)) { + process.kill(pid, "SIGKILL"); + } + } + } +}); + +test("silent initialization is bounded and reaps the owned child", { timeout: 22000 }, async () => { + const binDir = makeTempDir("codex-plugin-silent-init-"); + const statePath = path.join(binDir, "fake-codex-state.json"); + installFakeCodex(binDir, "silent-on-initialize"); + let pid = null; + + try { + const outcome = await settleWithin( + CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }), + 18000 + ); + pid = JSON.parse(fs.readFileSync(statePath, "utf8")).pid; + assert.equal(outcome.status, "rejected"); + assert.match(outcome.reason.message, /initialization timed out/i); + assert.equal(processIsAlive(pid), false); + } finally { + if (pid && processIsAlive(pid)) { + process.kill(pid, "SIGKILL"); + } + } +}); diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs new file mode 100644 index 000000000..600443e75 --- /dev/null +++ b/tests/broker-lifecycle.test.mjs @@ -0,0 +1,45 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import path from "node:path"; + +import { ensureBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { makeTempDir, writeExecutable } from "./helpers.mjs"; + +function processIsAlive(pid) { + try { + process.kill(pid, 0); + return true; + } catch (error) { + if (error?.code === "ESRCH") return false; + throw error; + } +} + +test("failed broker readiness terminates the detached broker", async () => { + const cwd = makeTempDir("codex-plugin-broker-timeout-"); + const scriptPath = path.join(cwd, "silent-broker.mjs"); + const pidPath = path.join(cwd, "broker-child.pid"); + let pid = null; + writeExecutable( + scriptPath, + `import fs from "node:fs";\nfs.writeFileSync(${JSON.stringify(pidPath)}, String(process.pid));\nsetInterval(() => {}, 1000);\n` + ); + + try { + const session = await ensureBrokerSession(cwd, { + scriptPath, + timeoutMs: 200 + }); + + assert.equal(session, null); + pid = Number(fs.readFileSync(pidPath, "utf8")); + await new Promise((resolve) => setTimeout(resolve, 50)); + assert.equal(processIsAlive(pid), false); + } finally { + if (pid && processIsAlive(pid)) { + process.kill(pid, "SIGKILL"); + } + fs.rmSync(cwd, { recursive: true, force: true }); + } +}); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index f83c96a0d..072686b33 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -9,6 +9,7 @@ export function installFakeCodex(binDir, behavior = "review-ok") { const scriptPath = path.join(binDir, "codex"); const source = `#!/usr/bin/env node const fs = require("node:fs"); +const { spawn } = require("node:child_process"); const crypto = require("node:crypto"); const path = require("node:path"); const readline = require("node:readline"); @@ -272,7 +273,11 @@ if (args[0] !== "app-server") { } const bootState = loadState(); bootState.appServerStarts = (bootState.appServerStarts || 0) + 1; +bootState.pid = process.pid; saveState(bootState); +if (BEHAVIOR === "broker-fails-then-exit-before-turn-start-response" && bootState.appServerStarts === 1) { + process.exit(1); +} const rl = readline.createInterface({ input: process.stdin }); rl.on("line", (line) => { @@ -288,7 +293,46 @@ rl.on("line", (line) => { case "initialize": state.capabilities = message.params.capabilities || null; saveState(state); + if (BEHAVIOR === "null-on-initialize") { + process.stdout.write("null\\n"); + setTimeout(() => process.exit(0), 20); + break; + } + if (BEHAVIOR === "malformed-on-initialize") { + process.on("SIGTERM", () => setTimeout(() => process.exit(0), 100)); + setInterval(() => {}, 1000); + process.stdout.write("not-json\\n"); + break; + } + if (BEHAVIOR === "uncooperative-tree-on-initialize") { + const descendant = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); + state.descendantPid = descendant.pid; + saveState(state); + process.on("SIGTERM", () => {}); + setInterval(() => {}, 1000); + process.stdout.write("not-json\\n"); + break; + } + if (BEHAVIOR === "silent-on-initialize") { + process.on("SIGTERM", () => setTimeout(() => process.exit(0), 100)); + setInterval(() => {}, 1000); + break; + } + if (BEHAVIOR === "stdout-eof-on-initialize") { + process.on("SIGTERM", () => setTimeout(() => process.exit(0), 100)); + setInterval(() => {}, 1000); + fs.closeSync(1); + break; + } send({ id: message.id, result: { userAgent: "fake-codex-app-server" } }); + if (BEHAVIOR === "stdin-eof-after-initialize") { + process.on("SIGTERM", () => setTimeout(() => process.exit(0), 100)); + setInterval(() => {}, 1000); + fs.closeSync(0); + } + if (BEHAVIOR === "exit-after-initialize") { + setTimeout(() => process.exit(0), 20); + } break; case "initialized": @@ -452,8 +496,19 @@ rl.on("line", (line) => { prompt }; saveState(state); + if ( + BEHAVIOR === "exit-before-turn-start-response" || + BEHAVIOR === "broker-fails-then-exit-before-turn-start-response" + ) { + process.exit(0); + } send({ id: message.id, result: { turn: buildTurn(turnId) } }); + if (BEHAVIOR === "exit-during-turn") { + setTimeout(() => process.exit(0), 20); + break; + } + const payload = message.params.outputSchema && message.params.outputSchema.properties && message.params.outputSchema.properties.verdict ? structuredReviewPayload(prompt) : taskPayload(prompt, thread.name && thread.name.startsWith("Codex Companion Task") && prompt.includes("Continue from the current thread state")); @@ -638,6 +693,11 @@ rl.on("line", (line) => { send({ id: message.id, error: { code: -32000, message: error.message } }); } }); +rl.on("close", () => { + if (BEHAVIOR === "slow-clean-exit") { + setTimeout(() => process.exit(0), 250); + } +}); `; writeExecutable(scriptPath, source); diff --git a/tests/helpers.mjs b/tests/helpers.mjs index d6981197a..41ced54dd 100644 --- a/tests/helpers.mjs +++ b/tests/helpers.mjs @@ -18,6 +18,7 @@ export function run(command, args, options = {}) { env: options.env, encoding: "utf8", input: options.input, + timeout: options.timeout, shell: options.shell ?? (process.platform === "win32" && !path.isAbsolute(command)), windowsHide: true }); diff --git a/tests/process.test.mjs b/tests/process.test.mjs index 80e0715b0..5fa7fdc1b 100644 --- a/tests/process.test.mjs +++ b/tests/process.test.mjs @@ -53,3 +53,40 @@ test("terminateProcessTree treats missing Windows processes as already stopped", assert.equal(outcome.result.status, 128); assert.match(outcome.result.stdout, /not found/i); }); + +test("terminateProcessTree falls back to the POSIX process when no process group exists", () => { + const calls = []; + const outcome = terminateProcessTree(1234, { + platform: "darwin", + killImpl(pid, signal) { + calls.push({ pid, signal }); + if (pid === -1234) { + const error = new Error("no such process group"); + error.code = "ESRCH"; + throw error; + } + } + }); + + assert.deepEqual(calls, [ + { pid: -1234, signal: "SIGTERM" }, + { pid: 1234, signal: "SIGTERM" } + ]); + assert.equal(outcome.delivered, true); + assert.equal(outcome.method, "process"); +}); + +test("terminateProcessTree forwards a forced POSIX signal to the process group", () => { + const calls = []; + const outcome = terminateProcessTree(1234, { + platform: "linux", + signal: "SIGKILL", + killImpl(pid, signal) { + calls.push({ pid, signal }); + } + }); + + assert.deepEqual(calls, [{ pid: -1234, signal: "SIGKILL" }]); + assert.equal(outcome.delivered, true); + assert.equal(outcome.method, "process-group"); +}); diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 8f276835b..b7f1b82fe 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -3,11 +3,13 @@ import path from "node:path"; import test from "node:test"; import assert from "node:assert/strict"; import { spawn } from "node:child_process"; +import net from "node:net"; import { fileURLToPath } from "node:url"; import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs"; import { initGitRepo, makeTempDir, run } from "./helpers.mjs"; -import { loadBrokerSession, saveBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; +import { ensureBrokerSession, loadBrokerSession, saveBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); @@ -28,6 +30,18 @@ async function waitFor(predicate, { timeoutMs = 5000, intervalMs = 50 } = {}) { throw new Error("Timed out waiting for condition."); } +function processIsAlive(pid) { + try { + process.kill(pid, 0); + return true; + } catch (error) { + if (error?.code === "ESRCH") { + return false; + } + throw error; + } +} + test("setup reports ready when fake codex is installed and authenticated", () => { const binDir = makeTempDir(); installFakeCodex(binDir); @@ -193,6 +207,137 @@ test("task runs without auth preflight so Codex can refresh an expired session", assert.match(result.stdout, /Handled the requested task/); }); +test("Codex death terminates the broker and a later command starts a fresh broker", async () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "exit-during-turn"); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + const env = buildEnv(binDir); + + try { + const result = run("node", [SCRIPT, "task", "observe connection loss"], { + cwd: repo, + env, + timeout: 5000 + }); + + assert.equal(result.error, undefined, result.error?.message); + assert.notEqual(result.status, 0); + assert.match(result.stderr, /app-server.*(closed|exited)|connection.*closed/i); + + const firstSession = loadBrokerSession(repo); + assert.ok(firstSession); + const firstEndpoint = parseBrokerEndpoint(firstSession.endpoint); + await waitFor( + () => !processIsAlive(firstSession.pid) && (firstEndpoint.kind !== "unix" || !fs.existsSync(firstEndpoint.path)), + { timeoutMs: 3000 } + ); + + const secondResult = run("node", [SCRIPT, "task", "start after connection loss"], { + cwd: repo, + env, + timeout: 5000 + }); + assert.equal(secondResult.error, undefined, secondResult.error?.message); + assert.notEqual(secondResult.status, 0); + + const secondSession = loadBrokerSession(repo); + assert.ok(secondSession); + assert.notEqual(secondSession.pid, firstSession.pid); + const fakeState = JSON.parse(fs.readFileSync(path.join(binDir, "fake-codex-state.json"), "utf8")); + assert.equal(fakeState.appServerStarts, 2); + } finally { + run("node", [SESSION_HOOK, "SessionEnd"], { + cwd: repo, + env, + input: JSON.stringify({ hook_event_name: "SessionEnd", cwd: repo }), + timeout: 2000 + }); + } +}); + +test("connection loss before turn/start responds without an unhandled rejection", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "broker-fails-then-exit-before-turn-start-response"); + initGitRepo(repo); + const env = { + ...buildEnv(binDir), + CLAUDE_PLUGIN_DATA: path.join(binDir, "plugin-data") + }; + const runner = path.join(binDir, "run-turn.mjs"); + fs.writeFileSync( + runner, + [ + `const { runAppServerTurn } = await import(${JSON.stringify(path.join(PLUGIN_ROOT, "scripts", "lib", "codex.mjs"))});`, + "const unhandled = [];", + 'process.on("unhandledRejection", (error) => unhandled.push(error?.message ?? String(error)));', + "let caught = null;", + 'try { await runAppServerTurn(process.argv[2], { prompt: "observe early connection loss" }); } catch (error) { caught = error.message; }', + "await new Promise((resolve) => setTimeout(resolve, 100));", + 'process.stdout.write(JSON.stringify({ caught, unhandled }) + "\\n");' + ].join("\n") + ); + + try { + const result = run("node", [runner, repo], { + cwd: repo, + env, + timeout: 5000 + }); + + assert.equal(result.error, undefined, result.error?.message); + assert.equal(result.status, 0, result.stderr); + const outcome = JSON.parse(result.stdout); + assert.match(outcome.caught, /app-server.*(closed|exited)|connection.*closed/i); + assert.deepEqual(outcome.unhandled, []); + } finally { + run("node", [SESSION_HOOK, "SessionEnd"], { + cwd: repo, + env, + input: JSON.stringify({ hook_event_name: "SessionEnd", cwd: repo }), + timeout: 5000 + }); + } +}); + +test("Codex death shuts down the broker with a half-open client", async () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir); + initGitRepo(repo); + const env = { + ...buildEnv(binDir), + CLAUDE_PLUGIN_DATA: path.join(binDir, "plugin-data") + }; + const session = await ensureBrokerSession(repo, { env }); + assert.ok(session); + const target = parseBrokerEndpoint(session.endpoint); + const socket = net.createConnection({ path: target.path, allowHalfOpen: true }); + await new Promise((resolve, reject) => { + socket.once("connect", resolve); + socket.once("error", reject); + }); + socket.on("end", () => {}); + + try { + const fakeState = JSON.parse(fs.readFileSync(path.join(binDir, "fake-codex-state.json"), "utf8")); + process.kill(fakeState.pid, "SIGKILL"); + await waitFor(() => !processIsAlive(session.pid), { timeoutMs: 3000 }); + } finally { + socket.destroy(); + run("node", [SESSION_HOOK, "SessionEnd"], { + cwd: repo, + env, + input: JSON.stringify({ hook_event_name: "SessionEnd", cwd: repo }), + timeout: 5000 + }); + } +}); + test("transfer delegates the current Claude session directly to native import", () => { const home = makeTempDir(); const repo = path.join(home, "repo"); From 92ed77c7f47c0e7c27a048f743efd840a5e2038e Mon Sep 17 00:00:00 2001 From: SP Son <1376128+seungpyoson@users.noreply.github.com> Date: Sat, 29 Aug 2026 16:13:57 +0900 Subject: [PATCH 2/3] fix: address app-server lifecycle review --- plugins/codex/scripts/lib/app-server.mjs | 16 +++++--- tests/app-server-lifecycle.test.mjs | 51 +++++++++++++----------- tests/broker-lifecycle.test.mjs | 12 +----- tests/fake-codex-fixture.mjs | 16 ++++++-- tests/helpers.mjs | 23 +++++++++++ tests/runtime.test.mjs | 51 ++++++++++++++++++------ 6 files changed, 112 insertions(+), 57 deletions(-) diff --git a/plugins/codex/scripts/lib/app-server.mjs b/plugins/codex/scripts/lib/app-server.mjs index 27bd2aee1..514ae65f4 100644 --- a/plugins/codex/scripts/lib/app-server.mjs +++ b/plugins/codex/scripts/lib/app-server.mjs @@ -24,6 +24,7 @@ export const BROKER_BUSY_RPC_CODE = -32001; const CHILD_STDIN_GRACE_MS = 1000; const CHILD_TERMINATION_GRACE_MS = 1000; +const CHILD_EXIT_DIAGNOSTIC_GRACE_MS = 250; const APP_SERVER_INITIALIZE_TIMEOUT_MS = 15000; const BROKER_SOCKET_CLOSE_GRACE_MS = 1000; @@ -260,7 +261,6 @@ class SpawnedCodexAppServerClient extends AppServerClientBase { env: this.options.env ?? process.env, stdio: ["pipe", "pipe", "pipe"], shell: process.platform === "win32" ? (process.env.SHELL || true) : false, - detached: process.platform !== "win32", windowsHide: true }); @@ -279,16 +279,22 @@ class SpawnedCodexAppServerClient extends AppServerClientBase { }); this.proc.stdout.on("end", () => { - if (!this.terminalCause) { - this.transitionToTerminal(createProtocolError("codex app-server stdout closed before the connection ended.")); - } + // Child close follows stream EOF for ordinary process exits and carries + // the exit status after stderr has drained. + const diagnosticTimer = setTimeout(() => { + if (!this.terminalCause) { + this.transitionToTerminal(createProtocolError("codex app-server stdout closed before the connection ended.")); + } + }, CHILD_EXIT_DIAGNOSTIC_GRACE_MS); + diagnosticTimer.unref?.(); + this.proc.once("close", () => clearTimeout(diagnosticTimer)); }); this.proc.on("error", (error) => { this.transitionToTerminal(error); }); - this.proc.on("exit", (code, signal) => { + this.proc.on("close", (code, signal) => { const stderr = this.stderr.trim(); const detail = code === 0 diff --git a/tests/app-server-lifecycle.test.mjs b/tests/app-server-lifecycle.test.mjs index 6ecac2540..f34177445 100644 --- a/tests/app-server-lifecycle.test.mjs +++ b/tests/app-server-lifecycle.test.mjs @@ -6,7 +6,7 @@ import test from "node:test"; import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs"; import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs"; -import { makeTempDir } from "./helpers.mjs"; +import { makeTempDir, processIsAlive } from "./helpers.mjs"; function settleWithin(promise, timeoutMs = 250) { let timer; @@ -21,16 +21,6 @@ function settleWithin(promise, timeoutMs = 250) { ]).finally(() => clearTimeout(timer)); } -function processIsAlive(pid) { - try { - process.kill(pid, 0); - return true; - } catch (error) { - if (error?.code === "ESRCH") return false; - throw error; - } -} - test("a request made after the app-server child exits rejects instead of hanging", async () => { const binDir = makeTempDir("codex-plugin-lifecycle-"); installFakeCodex(binDir, "exit-after-initialize"); @@ -63,6 +53,24 @@ test("a JSON null protocol line rejects initialization instead of crashing the h ); }); +test("a startup exit preserves its code and stderr when stdout closes first", async () => { + const binDir = makeTempDir("codex-plugin-startup-exit-"); + installFakeCodex(binDir, "startup-exit-after-stdout-eof"); + + await assert.rejects( + CodexAppServerClient.connect(binDir, { + disableBroker: true, + env: buildEnv(binDir) + }), + (error) => { + assert.match(error.message, /exit 17/i); + assert.match(error.message, /fake codex startup failure/i); + assert.doesNotMatch(error.message, /stdout closed before/i); + return true; + } + ); +}); + test("protocol EOF rejects initialization and reaps the live child", { skip: process.platform === "win32" }, async () => { const binDir = makeTempDir("codex-plugin-stdout-eof-"); const statePath = path.join(binDir, "fake-codex-state.json"); @@ -75,7 +83,7 @@ test("protocol EOF rejects initialization and reaps the live child", { skip: pro disableBroker: true, env: buildEnv(binDir) }), - 2000 + 3000 ); pid = JSON.parse(fs.readFileSync(statePath, "utf8")).pid; @@ -162,12 +170,11 @@ test("initialization failure reaps the owned app-server child before rejecting", } }); -test("forced teardown reaps an uncooperative launcher and its descendant", { skip: process.platform === "win32" }, async () => { +test("forced teardown reaps an uncooperative app-server", { skip: process.platform === "win32" }, async () => { const binDir = makeTempDir("codex-plugin-force-reap-"); const statePath = path.join(binDir, "fake-codex-state.json"); - installFakeCodex(binDir, "uncooperative-tree-on-initialize"); - let launcherPid = null; - let descendantPid = null; + installFakeCodex(binDir, "uncooperative-on-initialize"); + let pid = null; try { const outcome = await settleWithin( @@ -178,18 +185,14 @@ test("forced teardown reaps an uncooperative launcher and its descendant", { ski 3000 ); const state = JSON.parse(fs.readFileSync(statePath, "utf8")); - launcherPid = state.pid; - descendantPid = state.descendantPid; + pid = state.pid; assert.equal(outcome.status, "rejected"); assert.match(outcome.reason.message, /failed to parse/i); - assert.equal(processIsAlive(launcherPid), false); - assert.equal(processIsAlive(descendantPid), false); + assert.equal(processIsAlive(pid), false); } finally { - for (const pid of [launcherPid, descendantPid]) { - if (pid && processIsAlive(pid)) { - process.kill(pid, "SIGKILL"); - } + if (pid && processIsAlive(pid)) { + process.kill(pid, "SIGKILL"); } } }); diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs index 600443e75..4ee4c9b0b 100644 --- a/tests/broker-lifecycle.test.mjs +++ b/tests/broker-lifecycle.test.mjs @@ -4,17 +4,7 @@ import fs from "node:fs"; import path from "node:path"; import { ensureBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; -import { makeTempDir, writeExecutable } from "./helpers.mjs"; - -function processIsAlive(pid) { - try { - process.kill(pid, 0); - return true; - } catch (error) { - if (error?.code === "ESRCH") return false; - throw error; - } -} +import { makeTempDir, processIsAlive, writeExecutable } from "./helpers.mjs"; test("failed broker readiness terminates the detached broker", async () => { const cwd = makeTempDir("codex-plugin-broker-timeout-"); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index 072686b33..711869a46 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -298,21 +298,29 @@ rl.on("line", (line) => { setTimeout(() => process.exit(0), 20); break; } + if (BEHAVIOR === "startup-exit-after-stdout-eof") { + fs.writeSync(2, "fake codex startup failure\\n"); + fs.closeSync(1); + setTimeout(() => process.exit(17), 20); + break; + } if (BEHAVIOR === "malformed-on-initialize") { process.on("SIGTERM", () => setTimeout(() => process.exit(0), 100)); setInterval(() => {}, 1000); process.stdout.write("not-json\\n"); break; } - if (BEHAVIOR === "uncooperative-tree-on-initialize") { - const descendant = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); - state.descendantPid = descendant.pid; - saveState(state); + if (BEHAVIOR === "uncooperative-on-initialize") { process.on("SIGTERM", () => {}); setInterval(() => {}, 1000); process.stdout.write("not-json\\n"); break; } + if (BEHAVIOR === "spawn-tree-after-initialize") { + const descendant = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); + state.descendantPid = descendant.pid; + saveState(state); + } if (BEHAVIOR === "silent-on-initialize") { process.on("SIGTERM", () => setTimeout(() => process.exit(0), 100)); setInterval(() => {}, 1000); diff --git a/tests/helpers.mjs b/tests/helpers.mjs index 41ced54dd..01c6df398 100644 --- a/tests/helpers.mjs +++ b/tests/helpers.mjs @@ -12,6 +12,29 @@ export function writeExecutable(filePath, source) { fs.writeFileSync(filePath, source, { encoding: "utf8", mode: 0o755 }); } +export function processIsAlive(pid) { + try { + process.kill(pid, 0); + } catch (error) { + if (error?.code === "ESRCH") return false; + throw error; + } + + if (process.platform === "linux") { + try { + const stat = fs.readFileSync(`/proc/${pid}/stat`, "utf8"); + const commandEnd = stat.lastIndexOf(")"); + const state = stat.slice(commandEnd + 2, commandEnd + 3); + return state !== "Z" && state !== "X"; + } catch (error) { + if (error?.code === "ENOENT") return false; + throw error; + } + } + + return true; +} + export function run(command, args, options = {}) { return spawnSync(command, args, { cwd: options.cwd, diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index b7f1b82fe..9dff70ec8 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -7,9 +7,10 @@ import net from "node:net"; import { fileURLToPath } from "node:url"; import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs"; -import { initGitRepo, makeTempDir, run } from "./helpers.mjs"; +import { initGitRepo, makeTempDir, processIsAlive, run } from "./helpers.mjs"; import { parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; import { ensureBrokerSession, loadBrokerSession, saveBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; +import { terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); @@ -30,18 +31,6 @@ async function waitFor(predicate, { timeoutMs = 5000, intervalMs = 50 } = {}) { throw new Error("Timed out waiting for condition."); } -function processIsAlive(pid) { - try { - process.kill(pid, 0); - return true; - } catch (error) { - if (error?.code === "ESRCH") { - return false; - } - throw error; - } -} - test("setup reports ready when fake codex is installed and authenticated", () => { const binDir = makeTempDir(); installFakeCodex(binDir); @@ -304,6 +293,42 @@ test("connection loss before turn/start responds without an unhandled rejection" } }); +test("force-killing the broker process group also terminates Codex and its descendants", { skip: process.platform === "win32" }, async () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "spawn-tree-after-initialize"); + initGitRepo(repo); + const env = { + ...buildEnv(binDir), + CLAUDE_PLUGIN_DATA: path.join(binDir, "plugin-data") + }; + const pids = []; + + try { + const session = await ensureBrokerSession(repo, { env }); + assert.ok(session); + const fakeState = JSON.parse(fs.readFileSync(path.join(binDir, "fake-codex-state.json"), "utf8")); + pids.push(session.pid, fakeState.pid, fakeState.descendantPid); + assert.ok(pids.every(Number.isFinite)); + + const outcome = terminateProcessTree(session.pid, { signal: "SIGKILL" }); + assert.equal(outcome.method, "process-group"); + await waitFor(() => pids.every((pid) => !processIsAlive(pid)), { timeoutMs: 3000 }); + } finally { + for (const pid of pids) { + if (Number.isFinite(pid) && processIsAlive(pid)) { + process.kill(pid, "SIGKILL"); + } + } + run("node", [SESSION_HOOK, "SessionEnd"], { + cwd: repo, + env, + input: JSON.stringify({ hook_event_name: "SessionEnd", cwd: repo }), + timeout: 5000 + }); + } +}); + test("Codex death shuts down the broker with a half-open client", async () => { const repo = makeTempDir(); const binDir = makeTempDir(); From ca51d1ba18ce73d23285ab2edad479eaff387874 Mon Sep 17 00:00:00 2001 From: SP Son <1376128+seungpyoson@users.noreply.github.com> Date: Sat, 29 Aug 2026 16:38:09 +0900 Subject: [PATCH 3/3] test: trigger broken app-server stdin --- tests/app-server-lifecycle.test.mjs | 7 +++++++ tests/fake-codex-fixture.mjs | 4 +++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/tests/app-server-lifecycle.test.mjs b/tests/app-server-lifecycle.test.mjs index f34177445..1f713b454 100644 --- a/tests/app-server-lifecycle.test.mjs +++ b/tests/app-server-lifecycle.test.mjs @@ -106,9 +106,16 @@ test("a broken app-server stdin becomes the single terminal cause", { skip: proc }); try { + const requestOutcomePromise = settleWithin( + Promise.resolve().then(() => client.request("account/read", { refreshToken: false })), + 2000 + ); const outcome = await settleWithin(client.terminalPromise, 2000); + const requestOutcome = await requestOutcomePromise; assert.equal(outcome.status, "fulfilled"); assert.match(client.terminalCause.message, /EPIPE|broken pipe|write/i); + assert.equal(requestOutcome.status, "rejected"); + assert.equal(requestOutcome.reason, client.terminalCause); assert.throws(() => client.request("account/read", {}), (error) => error === client.terminalCause); } finally { await client.close(); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index 711869a46..643073209 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -332,12 +332,14 @@ rl.on("line", (line) => { fs.closeSync(1); break; } - send({ id: message.id, result: { userAgent: "fake-codex-app-server" } }); if (BEHAVIOR === "stdin-eof-after-initialize") { process.on("SIGTERM", () => setTimeout(() => process.exit(0), 100)); setInterval(() => {}, 1000); + fs.writeSync(1, JSON.stringify({ id: message.id, result: { userAgent: "fake-codex-app-server" } }) + "\\n"); fs.closeSync(0); + break; } + send({ id: message.id, result: { userAgent: "fake-codex-app-server" } }); if (BEHAVIOR === "exit-after-initialize") { setTimeout(() => process.exit(0), 20); }