From b44b19ac85d4d649662ef5e90ba6e56fceff02e7 Mon Sep 17 00:00:00 2001 From: Erwan Lee Pesle Date: Sun, 5 Jul 2026 09:53:54 +0800 Subject: [PATCH 1/2] fix(client): reap the bridge process tree on close killAgentIfRunning() signalled only the bridge PID, so processes the bridge itself spawned (the model process, MCP servers such as serena) were never signalled and survived as orphans, accumulating across runs until the agent app-server choked and new sessions hung at session/new. Spawn the bridge in its own process group (detached on Unix) and, on close, signal the whole group via a negative PID. The group signal is gated on an agentGroupLeader flag so acpx never signals its own process group; Windows and any non-detached path keep the previous single-process kill. Adds a real-process regression test: a mock agent forks a long-lived grandchild and records its PID; after AcpClient.close() that PID must be dead. Co-Authored-By: Codex Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01DeuCHKuxEBNVTxHm9bSmwL --- src/acp/client.ts | 24 ++++++++++++-- test/client.test.ts | 79 +++++++++++++++++++++++++++++++++++++++++++++ test/mock-agent.ts | 38 ++++++++++++++++++++++ 3 files changed, 138 insertions(+), 3 deletions(-) diff --git a/src/acp/client.ts b/src/acp/client.ts index b9830229..84233a8e 100644 --- a/src/acp/client.ts +++ b/src/acp/client.ts @@ -442,6 +442,7 @@ export class AcpClient { private agentStartedAt?: string; private lastAgentExit?: AgentExitInfo; private lastKnownPid?: number; + private agentGroupLeader = false; private readonly promptPermissionFailures = new Map(); private readonly pendingConnectionRequests = new Set(); private readonly modelConfigIds = new Map(); @@ -710,11 +711,16 @@ export class AcpClient { private async spawnAgentProcess( plan: AgentLaunchPlan, ): Promise> { + const agentGroupLeader = process.platform !== "win32"; const spawnedChild = spawn( plan.spawnCommand, plan.args, - buildSpawnCommandOptions(plan.spawnCommand, plan.spawnOptions), + buildSpawnCommandOptions(plan.spawnCommand, { + ...plan.spawnOptions, + detached: agentGroupLeader, + }), ) as ChildProcessByStdio; + this.agentGroupLeader = agentGroupLeader; try { await waitForSpawn(spawnedChild); } catch (error) { @@ -1369,6 +1375,7 @@ export class AcpClient { this.initResult = undefined; this.connection = undefined; this.agent = undefined; + this.agentGroupLeader = false; } private async terminateAgentProcess( @@ -1409,9 +1416,20 @@ export class AcpClient { return alreadyExited; } try { - child.kill(signal); + if (this.agentGroupLeader && child.pid !== undefined) { + // Detached Unix bridges lead their own process group (pgid == pid), so + // a negative PID targets only that group and its descendants. The + // agentGroupLeader gate prevents signaling acpx's own group. + process.kill(-child.pid, signal); + } else { + child.kill(signal); + } } catch { - // best effort + try { + child.kill(signal); + } catch { + // best effort + } } return await waitForChildExit(child, waitMs); } diff --git a/test/client.test.ts b/test/client.test.ts index 3483c39f..4f52359d 100644 --- a/test/client.test.ts +++ b/test/client.test.ts @@ -1,7 +1,9 @@ import assert from "node:assert/strict"; +import fs from "node:fs/promises"; import path from "node:path"; import { PassThrough } from "node:stream"; import test from "node:test"; +import { fileURLToPath } from "node:url"; import type { RequestPermissionRequest, RequestPermissionResponse } from "@agentclientprotocol/sdk"; import { AcpClient, @@ -20,6 +22,10 @@ import { PermissionPromptUnavailableError, UnsupportedPromptContentError, } from "../src/errors.js"; +import { isProcessAlive } from "../src/process-liveness.js"; +import { fileExists, withTempDir } from "./runtime-test-helpers.js"; + +const MOCK_AGENT_PATH = fileURLToPath(new URL("./mock-agent.js", import.meta.url)); test("parseAcpJsonMessageLine ignores non-object JSON values", () => { for (const line of ["1", "null", '"diagnostic"', "[]", "[{}]"]) { @@ -1276,6 +1282,53 @@ test("AcpClient start fails fast when the agent exits during initialize", async assert(Date.now() - startedAt < 2_000); }); +test("AcpClient close terminates long-lived bridge grandchildren", async () => { + if (process.platform === "win32") { + return; + } + + await withTempDir("acpx-client-grandchild-", async (tempDir) => { + const grandchildPidFile = path.join(tempDir, "grandchild.pid"); + const client = makeClient({ + agentCommand: [ + JSON.stringify(process.execPath), + JSON.stringify(MOCK_AGENT_PATH), + "--ignore-sigterm", + "--stay-alive-after-stdin-end", + "--grandchild-pid-file", + JSON.stringify(grandchildPidFile), + ].join(" "), + }); + let grandchildPid: number | undefined; + + try { + await client.start(); + await waitUntil(() => fileExists(grandchildPidFile)); + + grandchildPid = Number((await fs.readFile(grandchildPidFile, "utf8")).trim()); + assert( + Number.isInteger(grandchildPid) && grandchildPid > 0, + "grandchild PID must be a positive integer", + ); + assert.equal(isProcessAlive(grandchildPid), true, "grandchild must be alive before close"); + + await client.close(); + + const grandchildExited = await waitUntil(() => + Promise.resolve(!isProcessAlive(grandchildPid)), + ); + assert.equal( + grandchildExited, + true, + `grandchild process ${grandchildPid} must be dead after AcpClient.close()`, + ); + } finally { + await client.close(); + killProcessIfAlive(grandchildPid); + } + }); +}); + test("AcpClient close resets in-memory state and shuts down terminal manager", async () => { const client = makeClient(); const internals = asInternals(client); @@ -1358,6 +1411,32 @@ function asInternals(client: AcpClient): ClientInternals { return client as unknown as ClientInternals; } +async function waitUntil( + condition: () => Promise, + timeoutMs = 2_000, + pollMs = 50, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await condition()) { + return true; + } + await new Promise((resolve) => setTimeout(resolve, pollMs)); + } + return false; +} + +function killProcessIfAlive(pid: number | undefined): void { + if (pid === undefined || !isProcessAlive(pid)) { + return; + } + try { + process.kill(pid, "SIGKILL"); + } catch { + // best effort cleanup for the intentional RED leak + } +} + function makePermissionRequest( sessionId: string, kind: RequestPermissionRequest["toolCall"]["kind"], diff --git a/test/mock-agent.ts b/test/mock-agent.ts index 840cb135..18923832 100644 --- a/test/mock-agent.ts +++ b/test/mock-agent.ts @@ -1,5 +1,6 @@ #!/usr/bin/env node +import { spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; import { writeFileSync } from "node:fs"; import path from "node:path"; @@ -68,8 +69,11 @@ type MockAgentOptions = { loadReplayText: string; ignoreSigterm: boolean; cancelDelayMs: number; + stayAliveAfterStdinEnd: boolean; /** If set, the agent writes its PID to this path at startup (before ACP handshake). */ pidFile?: string; + /** If set, the agent spawns a long-lived child and writes its PID to this path. */ + grandchildPidFile?: string; }; type SessionState = { @@ -379,7 +383,9 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { let ignoreSigterm = false; let cancelDelayMs = 0; let hangOnNewSession = false; + let stayAliveAfterStdinEnd = false; let pidFile: string | undefined; + let grandchildPidFile: string | undefined; for (let index = 0; index < argv.length; index += 1) { const token = argv[index]; @@ -513,6 +519,11 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { continue; } + if (token === "--stay-alive-after-stdin-end") { + stayAliveAfterStdinEnd = true; + continue; + } + if (token === "--cancel-delay-ms") { cancelDelayMs = parsePositiveIntegerOption(argv, index + 1, token); index += 1; @@ -530,6 +541,12 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { continue; } + if (token === "--grandchild-pid-file") { + grandchildPidFile = parseOptionValue(argv, index + 1, token); + index += 1; + continue; + } + if (token === "--claude-agent-acp") { continue; } @@ -596,7 +613,9 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { loadReplayText, ignoreSigterm, cancelDelayMs, + stayAliveAfterStdinEnd, pidFile, + grandchildPidFile, }; } @@ -1481,18 +1500,37 @@ const input = Readable.toWeb(process.stdin) as ReadableStream; const stream = ndJsonStream(output, input); const mockAgentOptions = parseMockAgentOptions(process.argv.slice(2)); +function spawnLongLivedGrandchild(pidFile: string): void { + const grandchild = spawn(process.execPath, ["--eval", "setInterval(() => {}, 1000);"], { + stdio: "ignore", + }); + if (!grandchild.pid) { + throw new Error("long-lived grandchild did not receive a PID"); + } + grandchild.unref(); + writeFileSync(pidFile, `${grandchild.pid}\n`, "utf8"); +} + // Write PID to a file before doing anything else so that the parent can track // this bridge process and verify it is dead after queue-owner shutdown. if (mockAgentOptions.pidFile) { writeFileSync(mockAgentOptions.pidFile, `${process.pid}\n`, "utf8"); } +if (mockAgentOptions.grandchildPidFile) { + spawnLongLivedGrandchild(mockAgentOptions.grandchildPidFile); +} + if (mockAgentOptions.ignoreSigterm) { process.on("SIGTERM", () => { // Intentionally ignore to exercise ACP client SIGKILL fallback behavior. }); } +if (mockAgentOptions.stayAliveAfterStdinEnd) { + setInterval(() => undefined, 1_000); +} + const connection = new AgentSideConnection( (agentConnection) => new MockAgent(agentConnection, mockAgentOptions), stream, From 6c5a4f0e577ff677786f2b289e7785f59e83846a Mon Sep 17 00:00:00 2001 From: Erwan Lee Pesle Date: Sun, 5 Jul 2026 22:06:58 +0800 Subject: [PATCH 2/2] feat(acp): reap bridge trees on abrupt owner death --- native/lifeline.c | 137 ++++++++++++ package.json | 8 +- scripts/build-native-lifeline.mjs | 48 ++++ src/acp/client.ts | 215 ++++++++++++++++-- src/acp/lifeline.ts | 255 +++++++++++++++++++++ test/client.test.ts | 344 +++++++++++++++++++++++++++++ test/lifeline-test-helper.ts | 37 ++++ test/lifeline.test.ts | 337 ++++++++++++++++++++++++++++ test/mock-agent.ts | 34 ++- test/package-scripts.test.ts | 10 + test/queue-owner-lifecycle.test.ts | 169 ++++++++++++++ 11 files changed, 1562 insertions(+), 32 deletions(-) create mode 100644 native/lifeline.c create mode 100644 scripts/build-native-lifeline.mjs create mode 100644 src/acp/lifeline.ts create mode 100644 test/lifeline-test-helper.ts create mode 100644 test/lifeline.test.ts diff --git a/native/lifeline.c b/native/lifeline.c new file mode 100644 index 00000000..9ceae182 --- /dev/null +++ b/native/lifeline.c @@ -0,0 +1,137 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define GROUP_POLL_TIMEOUT_MS 100 + +static int parse_pgid(const char *value, pid_t *pgid) { + errno = 0; + char *end = NULL; + long parsed = strtol(value, &end, 10); + if (errno == ERANGE || end == value || *end != '\0' || parsed <= 1) { + return -1; + } + + pid_t narrowed = (pid_t)parsed; + if ((long)narrowed != parsed) { + return -1; + } + + *pgid = narrowed; + return 0; +} + +static int process_group_alive(pid_t pgid) { + if (kill(-pgid, 0) == 0) { + return 1; + } + if (errno == EPERM) { + return 1; + } + if (errno == ESRCH) { + return 0; + } + return 1; +} + +static void reap_group(pid_t pgid) { + kill(-pgid, SIGTERM); + usleep(200000); + /* + * The process group can outlive its leader. Probe the group itself so + * TERM-resistant descendants still get the SIGKILL fallback. + */ + if (kill(-pgid, 0) == 0 || errno == EPERM) { + kill(-pgid, SIGKILL); + } +} + +static int should_reap_after_owner_eof(pid_t bridge_pgid) { + return process_group_alive(bridge_pgid) ? 1 : 0; +} + +static int wait_for_owner_pipe(pid_t bridge_pgid) { + struct pollfd pfd = { + .fd = STDIN_FILENO, + .events = POLLIN, + }; + char buffer[32]; + + for (;;) { + int ready = poll(&pfd, 1, GROUP_POLL_TIMEOUT_MS); + if (ready == -1) { + if (errno == EINTR) { + continue; + } + perror("poll"); + return -1; + } + + if (ready == 0) { + if (!process_group_alive(bridge_pgid)) { + return 0; + } + continue; + } + + if ((pfd.revents & POLLIN) != 0) { + ssize_t nread = read(STDIN_FILENO, buffer, sizeof(buffer)); + if (nread == -1) { + if (errno == EINTR) { + continue; + } + perror("read"); + return -1; + } + if (nread == 0) { + return should_reap_after_owner_eof(bridge_pgid); + } + if (memchr(buffer, 'R', (size_t)nread) != NULL) { + return 0; + } + } + + if ((pfd.revents & POLLHUP) != 0) { + return should_reap_after_owner_eof(bridge_pgid); + } + + if ((pfd.revents & (POLLERR | POLLNVAL)) != 0) { + /* + * The lifeline fd itself is broken: we can no longer observe the owner. + * Degrade to pre-lifeline behavior (exit without reaping) rather than + * risk killing a live session, and avoid a POLLERR busy-loop. + */ + return -1; + } + + if (!process_group_alive(bridge_pgid)) { + return 0; + } + } +} + +int main(int argc, char **argv) { + if (argc != 2) { + fprintf(stderr, "usage: lifeline \n"); + return 2; + } + + pid_t bridge_pgid = 0; + if (parse_pgid(argv[1], &bridge_pgid) != 0) { + fprintf(stderr, "invalid bridge pgid\n"); + return 2; + } + + int result = wait_for_owner_pipe(bridge_pgid); + if (result == 1) { + reap_group(bridge_pgid); + return 0; + } + return result == 0 ? 0 : 1; +} diff --git a/package.json b/package.json index 682241e2..f8f1e8cf 100644 --- a/package.json +++ b/package.json @@ -22,6 +22,7 @@ }, "files": [ "dist", + "native", "skills", "README.md", "LICENSE" @@ -35,8 +36,9 @@ "./package.json": "./package.json" }, "scripts": { - "build": "tsdown src/cli.ts src/flows.ts src/runtime.ts --format esm --dts --clean --platform node --target node22 --no-fixedExtension", - "build:quiet": "tsdown --logLevel silent src/cli.ts src/flows.ts src/runtime.ts --format esm --dts --clean --platform node --target node22 --no-fixedExtension", + "build": "tsdown src/cli.ts src/flows.ts src/runtime.ts --format esm --dts --clean --platform node --target node22 --no-fixedExtension && pnpm run build:native", + "build:native": "node scripts/build-native-lifeline.mjs", + "build:quiet": "tsdown --logLevel silent src/cli.ts src/flows.ts src/runtime.ts --format esm --dts --clean --platform node --target node22 --no-fixedExtension && pnpm run build:native", "build:test": "node -e \"require('node:fs').rmSync('dist-test',{recursive:true,force:true})\" && tsgo -p tsconfig.test.json", "check": "pnpm run format:check && pnpm run typecheck && pnpm run lint && pnpm run build && pnpm run viewer:typecheck && pnpm run viewer:build && pnpm run test:coverage", "check:changed": "pnpm run check", @@ -62,7 +64,7 @@ "mutate": "stryker run", "perf:report": "tsx scripts/perf-report.ts", "precommit": "pnpm exec lint-staged && pnpm run -s build", - "prepack": "tsdown --logLevel silent src/cli.ts src/flows.ts src/runtime.ts --format esm --dts --clean --platform node --target node22 --no-fixedExtension", + "prepack": "tsdown --logLevel silent src/cli.ts src/flows.ts src/runtime.ts --format esm --dts --clean --platform node --target node22 --no-fixedExtension && node -e \"require('node:fs').rmSync('dist/native',{recursive:true,force:true})\"", "prepare": "husky", "test": "pnpm run build && pnpm run build:test && node --test dist-test/test/*.test.js", "test:changed": "pnpm run test:coverage", diff --git a/scripts/build-native-lifeline.mjs b/scripts/build-native-lifeline.mjs new file mode 100644 index 00000000..cc7494ed --- /dev/null +++ b/scripts/build-native-lifeline.mjs @@ -0,0 +1,48 @@ +import { spawnSync } from "node:child_process"; +import { mkdirSync } from "node:fs"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); +const targetPlatform = process.env.npm_config_platform ?? process.platform; +const targetArch = process.env.npm_config_arch ?? process.arch; + +if (targetPlatform !== "darwin" && targetPlatform !== "linux") { + process.exit(0); +} + +if (targetPlatform !== process.platform) { + console.warn( + `[acpx] skipping ${targetPlatform} lifeline helper build on ${process.platform} host`, + ); + process.exit(0); +} + +if (targetArch !== process.arch) { + console.warn( + `[acpx] skipping ${targetPlatform}-${targetArch} lifeline helper build on ${process.arch} host`, + ); + process.exit(0); +} + +const cc = process.env.CC ?? "cc"; +const source = path.join(repoRoot, "native", "lifeline.c"); +const outDir = path.join(repoRoot, "dist", "native"); +const output = path.join(outDir, `lifeline-${targetPlatform}-${targetArch}`); + +mkdirSync(outDir, { recursive: true }); + +const result = spawnSync(cc, ["-O2", source, "-o", output], { + stdio: "inherit", +}); + +if (result.error) { + const code = "code" in result.error ? result.error.code : undefined; + if (code === "ENOENT") { + console.warn(`[acpx] skipping ${targetPlatform} lifeline helper build: cc not found`); + process.exit(0); + } + throw result.error; +} + +process.exit(result.status ?? 1); diff --git a/src/acp/client.ts b/src/acp/client.ts index 84233a8e..e90c06c3 100644 --- a/src/acp/client.ts +++ b/src/acp/client.ts @@ -100,6 +100,7 @@ import { } from "./client-process.js"; import { extractAcpError } from "./error-shapes.js"; import { isAcpMessageObject, isSessionUpdateNotification } from "./jsonrpc.js"; +import { ensureLifelineHelper } from "./lifeline.js"; import { modelStateFromConfigOptions, modelStateFromSessionResponse, @@ -443,6 +444,7 @@ export class AcpClient { private lastAgentExit?: AgentExitInfo; private lastKnownPid?: number; private agentGroupLeader = false; + private lifelineWatchdog?: ChildProcess; private readonly promptPermissionFailures = new Map(); private readonly pendingConnectionRequests = new Set(); private readonly modelConfigIds = new Map(); @@ -711,7 +713,11 @@ export class AcpClient { private async spawnAgentProcess( plan: AgentLaunchPlan, ): Promise> { - const agentGroupLeader = process.platform !== "win32"; + const lifelineHelper = + process.platform !== "win32" + ? await ensureLifelineHelper({ log: (message) => this.log(message) }) + : undefined; + const agentGroupLeader = process.platform !== "win32" && lifelineHelper !== undefined; const spawnedChild = spawn( plan.spawnCommand, plan.args, @@ -723,12 +729,65 @@ export class AcpClient { this.agentGroupLeader = agentGroupLeader; try { await waitForSpawn(spawnedChild); + await this.startLifelineWatchdog(spawnedChild, lifelineHelper ?? null); } catch (error) { + this.stopLifelineWatchdog(); throw new AgentSpawnError(this.options.agentCommand, error); } return requireAgentStdio(spawnedChild); } + private async startLifelineWatchdog( + child: ChildProcess, + helperOverride?: string | null, + ): Promise { + this.stopLifelineWatchdog(); + if (!this.agentGroupLeader || child.pid === undefined) { + return; + } + + const helper = + helperOverride === undefined + ? await ensureLifelineHelper({ log: (message) => this.log(message) }) + : helperOverride; + if (!helper) { + return; + } + + try { + /* + * The bridge is already spawned before this helper can inherit the pipe, + * so an owner SIGKILL in that tiny gap can still orphan it. Once the + * watchdog starts, pipe EOF is the owner identity and death signal. + */ + const watchdog = spawn(helper, [String(child.pid)], { + detached: true, + stdio: ["pipe", "ignore", "ignore"], + windowsHide: true, + }); + this.lifelineWatchdog = watchdog; + const clearWatchdog = () => { + if (this.lifelineWatchdog === watchdog) { + this.lifelineWatchdog = undefined; + } + }; + await waitForSpawn(watchdog); + watchdog.once("exit", clearWatchdog); + watchdog.once("error", () => { + clearWatchdog(); + }); + watchdog.unref(); + } catch { + // The watchdog is best-effort and only guards abrupt owner death. The bridge was + // already spawned detached, so agentGroupLeader stays true to preserve the group-kill + // on cooperative close; a failed watchdog only forgoes abrupt-death reaping for a plain + // owner-PID SIGKILL / crash / OOM, which an attached bridge would not survive either (the + // one case attaching would still cover is a group-wide `kill -9 -` / terminal + // SIGHUP). Downgrading agentGroupLeader here would instead break the close-time group-kill. + this.stopLifelineWatchdog(); + } + } + private createConnection( stream: { readable: ReadableStream; @@ -838,15 +897,19 @@ export class AcpClient { error: unknown, ): Promise { params.startupFailure.dispose(); - const normalizedError = await this.normalizeInitializeError( - error, - params.child, - params.startupStderr, - ); + let normalizedError: unknown = error; try { - params.child.kill(); - } catch { - // best effort + normalizedError = await this.normalizeInitializeError( + error, + params.child, + params.startupStderr, + ); + } finally { + try { + await this.terminateInitializeFailedProcess(params.child); + } finally { + this.stopLifelineWatchdog(); + } } if (params.launch.geminiAcp && error instanceof TimeoutError) { throw new GeminiAcpStartupTimeoutError( @@ -1337,8 +1400,12 @@ export class AcpClient { await this.terminalManager.shutdown(); const agent = this.agent; - if (agent) { - await this.terminateAgentProcess(agent); + try { + if (agent) { + await this.terminateAgentProcess(agent); + } + } finally { + this.stopLifelineWatchdog(); } if (this.pendingConnectionRequests.size > 0) { this.rejectPendingConnectionRequests( @@ -1376,6 +1443,7 @@ export class AcpClient { this.connection = undefined; this.agent = undefined; this.agentGroupLeader = false; + this.lifelineWatchdog = undefined; } private async terminateAgentProcess( @@ -1383,17 +1451,53 @@ export class AcpClient { ): Promise { const stdinCloseGraceMs = resolveAgentCloseAfterStdinEndMs(this.options.agentCommand); this.endAgentStdin(child); - let exited = await waitForChildExit(child, stdinCloseGraceMs); - exited = await this.killAgentIfRunning(child, exited, "SIGTERM", AGENT_CLOSE_TERM_GRACE_MS); - if (!exited) { - this.log(`agent did not exit after ${AGENT_CLOSE_TERM_GRACE_MS}ms; forcing SIGKILL`); - exited = await this.killAgentIfRunning(child, exited, "SIGKILL", AGENT_CLOSE_KILL_GRACE_MS); + let childExited = await waitForChildExit(child, stdinCloseGraceMs); + + if (!childExited || this.hasLiveAgentProcessGroup(child)) { + this.signalAgentProcess(child, "SIGTERM"); + const treeCleanedUp = await this.waitForAgentTreeCleanup(child, AGENT_CLOSE_TERM_GRACE_MS); + childExited = !isChildProcessRunning(child); + + if (!treeCleanedUp) { + if (!childExited) { + this.log(`agent did not exit after ${AGENT_CLOSE_TERM_GRACE_MS}ms; forcing SIGKILL`); + } + this.signalAgentProcess(child, "SIGKILL"); + await this.waitForAgentTreeCleanup(child, AGENT_CLOSE_KILL_GRACE_MS); + childExited = !isChildProcessRunning(child); + } } // Ensure stdio handles don't keep this process alive after close() returns. + this.detachAgentHandles(child, !childExited); + } + + private async terminateInitializeFailedProcess( + child: ChildProcessByStdio, + ): Promise { + this.signalAgentProcess(child, "SIGTERM"); + let exited = await waitForChildExit(child, AGENT_CLOSE_TERM_GRACE_MS); + if (!exited || this.agentGroupLeader) { + this.signalAgentProcess(child, "SIGKILL"); + exited = (await waitForChildExit(child, AGENT_CLOSE_KILL_GRACE_MS)) || exited; + } this.detachAgentHandles(child, !exited); } + private stopLifelineWatchdog(): void { + const watchdog = this.lifelineWatchdog; + this.lifelineWatchdog = undefined; + if (!watchdog?.stdin || watchdog.stdin.destroyed) { + return; + } + try { + watchdog.stdin.once("error", () => {}); + watchdog.stdin.end("R"); + } catch { + // best effort + } + } + private endAgentStdin(child: ChildProcessByStdio): void { // Closing stdin is the most graceful shutdown signal for stdio-based ACP agents. if (child.stdin.destroyed) { @@ -1406,32 +1510,51 @@ export class AcpClient { } } - private async killAgentIfRunning( + private async waitForAgentTreeCleanup( child: ChildProcessByStdio, - alreadyExited: boolean, - signal: NodeJS.Signals, waitMs: number, ): Promise { - if (alreadyExited || !isChildProcessRunning(child)) { - return alreadyExited; + if (this.agentGroupLeader && child.pid !== undefined && process.platform !== "win32") { + return await waitForChildAndProcessGroupExit(child, child.pid, waitMs); } + return await waitForChildExit(child, waitMs); + } + + private hasLiveAgentProcessGroup( + child: ChildProcessByStdio, + ): boolean { + return ( + this.agentGroupLeader && + child.pid !== undefined && + process.platform !== "win32" && + hasLiveProcessGroup(child.pid) + ); + } + + private signalAgentProcess( + child: ChildProcessByStdio, + signal: NodeJS.Signals, + ): void { try { if (this.agentGroupLeader && child.pid !== undefined) { // Detached Unix bridges lead their own process group (pgid == pid), so // a negative PID targets only that group and its descendants. The // agentGroupLeader gate prevents signaling acpx's own group. process.kill(-child.pid, signal); - } else { - child.kill(signal); + return; } + child.kill(signal); + return; } catch { + if (this.agentGroupLeader && !isChildProcessRunning(child)) { + return; + } try { child.kill(signal); } catch { // best effort } } - return await waitForChildExit(child, waitMs); } private detachAgentHandles(agent: ChildProcess, unref: boolean): void { @@ -2039,3 +2162,47 @@ export class AcpClient { await this.waitForSessionUpdateDrain(options?.idleMs ?? 0, options?.timeoutMs ?? 0); } } + +async function waitForChildAndProcessGroupExit( + child: ChildProcess, + processGroupId: number, + timeoutMs: number, +): Promise { + const deadline = Date.now() + Math.max(0, timeoutMs); + + while (true) { + if (!isChildProcessRunning(child) && !hasLiveProcessGroup(processGroupId)) { + return true; + } + + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) { + return false; + } + + await delay(Math.min(DRAIN_POLL_INTERVAL_MS, remainingMs)); + } +} + +function hasLiveProcessGroup(processGroupId: number): boolean { + try { + process.kill(-processGroupId, 0); + return true; + } catch (error) { + return getErrorCode(error) === "EPERM"; + } +} + +function getErrorCode(error: unknown): string | undefined { + if (typeof error !== "object" || error === null || !("code" in error)) { + return undefined; + } + const code = error.code; + return typeof code === "string" ? code : undefined; +} + +async function delay(ms: number): Promise { + await new Promise((resolve) => { + setTimeout(resolve, ms); + }); +} diff --git a/src/acp/lifeline.ts b/src/acp/lifeline.ts new file mode 100644 index 00000000..6bfd328e --- /dev/null +++ b/src/acp/lifeline.ts @@ -0,0 +1,255 @@ +import { execFile } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import fs from "node:fs"; +import fsp from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; +import { getAcpxVersion } from "../version.js"; + +const execFileAsync = promisify(execFile); +const LIFELINE_HELPER_ENV = "ACPX_LIFELINE_HELPER"; +const failedCompileCachePaths = new Set(); +const reportedCompileFailures = new Set(); +let cachedPackageRoot: string | undefined; + +type EnsureLifelineHelperOptions = { + arch?: string; + log?: (message: string) => void; + platform?: NodeJS.Platform; +}; + +type LifelineTarget = { + arch: string; + cachePath: string; + platform: NodeJS.Platform; +}; + +type FastPathResult = + | { + helper?: string; + status: "done"; + } + | { + status: "compile"; + }; + +function unique(values: string[]): string[] { + return [...new Set(values)]; +} + +function isExecutableFile(candidate: string): boolean { + try { + if (!fs.statSync(candidate).isFile()) { + return false; + } + fs.accessSync(candidate, fs.constants.X_OK); + return true; + } catch { + return false; + } +} + +function resolveEnvOverride(override: string): string | undefined { + if (!path.isAbsolute(override)) { + return undefined; + } + return isExecutableFile(override) ? override : undefined; +} + +function configuredEnvOverride(): string | undefined { + return process.env[LIFELINE_HELPER_ENV]?.trim() || undefined; +} + +function isSupportedPlatform(platform: NodeJS.Platform): boolean { + return platform === "darwin" || platform === "linux"; +} + +function lifelineHelperName(platform: NodeJS.Platform, arch: string): string { + return `lifeline-${platform}-${arch}`; +} + +function moduleDir(): string { + return path.dirname(fileURLToPath(import.meta.url)); +} + +function findPackageRoot(): string | undefined { + if (cachedPackageRoot !== undefined) { + return cachedPackageRoot; + } + + let current = moduleDir(); + while (true) { + if (isAcpxPackageRoot(current)) { + cachedPackageRoot = current; + return cachedPackageRoot; + } + const parent = path.dirname(current); + if (parent === current) { + return undefined; + } + current = parent; + } +} + +function isAcpxPackageRoot(candidate: string): boolean { + try { + const parsed = JSON.parse(fs.readFileSync(path.join(candidate, "package.json"), "utf8")) as { + name?: unknown; + }; + return parsed.name === "acpx"; + } catch { + return false; + } +} + +function prebuiltCandidates(fileName: string): string[] { + const base = moduleDir(); + const packageRoot = findPackageRoot(); + return unique([ + path.join(base, "native", fileName), + ...(packageRoot ? [path.join(packageRoot, "dist", "native", fileName)] : []), + ]); +} + +function sourceCandidates(fileName: string): string[] { + const packageRoot = findPackageRoot(); + return packageRoot ? [path.join(packageRoot, "native", fileName)] : []; +} + +function findExecutable(candidates: string[]): string | undefined { + return candidates.find(isExecutableFile); +} + +function findExistingFile(candidates: string[]): string | undefined { + return candidates.find((candidate) => fs.existsSync(candidate)); +} + +function cachePathFor(platform: NodeJS.Platform, arch: string): string { + return path.join( + os.homedir(), + ".acpx", + "native", + `lifeline-${getAcpxVersion()}-${platform}-${arch}`, + ); +} + +function resolveCachedHelper(platform: NodeJS.Platform, arch: string): string | undefined { + const cachePath = cachePathFor(platform, arch); + return isExecutableFile(cachePath) ? cachePath : undefined; +} + +export function resolveLifelineHelper( + platform: NodeJS.Platform = process.platform, + arch: string = process.arch, +): string | undefined { + const override = configuredEnvOverride(); + if (override) { + return resolveEnvOverride(override); + } + + if (!isSupportedPlatform(platform)) { + return undefined; + } + + return ( + findExecutable(prebuiltCandidates(lifelineHelperName(platform, arch))) ?? + resolveCachedHelper(platform, arch) + ); +} + +export async function ensureLifelineHelper( + options: EnsureLifelineHelperOptions = {}, +): Promise { + const target = lifelineTarget(options); + const fastPath = resolveEnsureFastPath(target); + if (fastPath.status === "done") { + return fastPath.helper; + } + if (hasRememberedCompileFailure(target.cachePath)) { + return undefined; + } + + const source = findExistingFile(sourceCandidates("lifeline.c")); + if (!source) { + await rememberCompileFailure(target.cachePath, "lifeline source not found", options.log); + return undefined; + } + + try { + await compileLifelineSource(source, target.cachePath); + return isExecutableFile(target.cachePath) ? target.cachePath : undefined; + } catch (error) { + await rememberCompileFailure(target.cachePath, formatCompileFailure(error), options.log); + return undefined; + } +} + +function lifelineTarget(options: EnsureLifelineHelperOptions): LifelineTarget { + const platform = options.platform ?? process.platform; + const arch = options.arch ?? process.arch; + return { + arch, + cachePath: cachePathFor(platform, arch), + platform, + }; +} + +function resolveEnsureFastPath(target: LifelineTarget): FastPathResult { + const resolved = resolveLifelineHelper(target.platform, target.arch); + if (resolved || configuredEnvOverride() || !isSupportedPlatform(target.platform)) { + return { + helper: resolved, + status: "done", + }; + } + return { status: "compile" }; +} + +function hasRememberedCompileFailure(cachePath: string): boolean { + return failedCompileCachePaths.has(cachePath) || fs.existsSync(`${cachePath}.failed`); +} + +async function compileLifelineSource(source: string, cachePath: string): Promise { + await fsp.mkdir(path.dirname(cachePath), { recursive: true }); + const tmpPath = `${cachePath}.${process.pid}.${randomUUID()}.tmp`; + try { + await execFileAsync(process.env.CC ?? "cc", ["-O2", "-o", tmpPath, source]); + await renameCompiledHelper(tmpPath, cachePath); + } catch (error) { + await fsp.rm(tmpPath, { force: true }).catch(() => {}); + throw error; + } +} + +async function renameCompiledHelper(tmpPath: string, cachePath: string): Promise { + try { + await fsp.rename(tmpPath, cachePath); + } catch (error) { + if (fs.existsSync(cachePath)) { + await fsp.rm(tmpPath, { force: true }).catch(() => {}); + return; + } + throw error; + } +} + +async function rememberCompileFailure( + cachePath: string, + reason: string, + log: ((message: string) => void) | undefined, +): Promise { + failedCompileCachePaths.add(cachePath); + await fsp.mkdir(path.dirname(cachePath), { recursive: true }).catch(() => {}); + await fsp.writeFile(`${cachePath}.failed`, `${reason}\n`, "utf8").catch(() => {}); + if (reportedCompileFailures.has(cachePath)) { + return; + } + reportedCompileFailures.add(cachePath); + log?.(`lifeline helper lazy compile failed: ${reason}`); +} + +function formatCompileFailure(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} diff --git a/test/client.test.ts b/test/client.test.ts index 4f52359d..e26f8701 100644 --- a/test/client.test.ts +++ b/test/client.test.ts @@ -1,4 +1,6 @@ import assert from "node:assert/strict"; +import { spawn, type ChildProcess } from "node:child_process"; +import { once } from "node:events"; import fs from "node:fs/promises"; import path from "node:path"; import { PassThrough } from "node:stream"; @@ -23,6 +25,7 @@ import { UnsupportedPromptContentError, } from "../src/errors.js"; import { isProcessAlive } from "../src/process-liveness.js"; +import { LIFELINE_HELPER_ENV } from "./lifeline-test-helper.js"; import { fileExists, withTempDir } from "./runtime-test-helpers.js"; const MOCK_AGENT_PATH = fileURLToPath(new URL("./mock-agent.js", import.meta.url)); @@ -119,6 +122,10 @@ type ClientInternals = { kill: (signal?: NodeJS.Signals) => void; unref: () => void; }; + agentGroupLeader: boolean; + lifelineWatchdog?: ChildProcess; + startLifelineWatchdog?: (child: ChildProcess) => Promise; + stopLifelineWatchdog?: () => void; activePrompt?: | { sessionId: string; @@ -1282,6 +1289,60 @@ test("AcpClient start fails fast when the agent exits during initialize", async assert(Date.now() - startedAt < 2_000); }); +test("AcpClient start group-kills bridge descendants when initialize fails", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline watchdog is POSIX-only"); + return; + } + + await withTempDir("acpx-client-init-failure-grandchild-", async (tempDir) => { + const helper = await writeLifelineFixture(tempDir); + const grandchildPidFile = path.join(tempDir, "grandchild.pid"); + const client = makeClient({ + agentCommand: [ + JSON.stringify(process.execPath), + JSON.stringify(MOCK_AGENT_PATH), + "--fail-initialize", + "--grandchild-pid-file", + JSON.stringify(grandchildPidFile), + ].join(" "), + }); + let grandchildPid: number | undefined; + + await withEnv({ [LIFELINE_HELPER_ENV]: helper }, async () => { + try { + const startResult = client.start().then( + () => ({ type: "resolved" as const }), + (error: unknown) => ({ type: "rejected" as const, error }), + ); + const sawGrandchildPid = await waitUntil(() => fileExists(grandchildPidFile)); + assert.equal(sawGrandchildPid, true, "grandchild PID file must be written"); + grandchildPid = Number((await fs.readFile(grandchildPidFile, "utf8")).trim()); + assert( + Number.isInteger(grandchildPid) && grandchildPid > 0, + "grandchild PID must be valid", + ); + + const result = await startResult; + assert.equal(result.type, "rejected"); + assert.match(String(result.error), /initialize failed/); + + const grandchildExited = await waitUntil(() => + Promise.resolve(!isProcessAlive(grandchildPid)), + ); + assert.equal( + grandchildExited, + true, + `grandchild process ${grandchildPid} must be dead after initialize failure`, + ); + } finally { + await client.close(); + killProcessIfAlive(grandchildPid); + } + }); + }); +}); + test("AcpClient close terminates long-lived bridge grandchildren", async () => { if (process.platform === "win32") { return; @@ -1329,6 +1390,229 @@ test("AcpClient close terminates long-lived bridge grandchildren", async () => { }); }); +test("AcpClient close reaps bridge descendants after cooperative bridge exit", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline watchdog is POSIX-only"); + return; + } + + await withTempDir("acpx-client-cooperative-grandchild-", async (tempDir) => { + const helper = await writeLifelineFixture(tempDir); + const grandchildPidFile = path.join(tempDir, "grandchild.pid"); + const client = makeClient({ + agentCommand: [ + JSON.stringify(process.execPath), + JSON.stringify(MOCK_AGENT_PATH), + "--grandchild-pid-file", + JSON.stringify(grandchildPidFile), + "--grandchild-ignore-sigterm", + ].join(" "), + }); + let grandchildPid: number | undefined; + + await withEnv({ [LIFELINE_HELPER_ENV]: helper }, async () => { + try { + await client.start(); + await waitUntil(() => fileExists(grandchildPidFile)); + + grandchildPid = Number((await fs.readFile(grandchildPidFile, "utf8")).trim()); + assert( + Number.isInteger(grandchildPid) && grandchildPid > 0, + "grandchild PID must be a positive integer", + ); + assert.equal(isProcessAlive(grandchildPid), true, "grandchild must be alive before close"); + + await client.close(); + + const grandchildExited = await waitUntil(() => + Promise.resolve(!isProcessAlive(grandchildPid)), + ); + assert.equal( + grandchildExited, + true, + `grandchild process ${grandchildPid} must be dead after cooperative bridge exit`, + ); + } finally { + await client.close(); + killProcessIfAlive(grandchildPid); + } + }); + }); +}); + +test("AcpClient releases the lifeline watchdog on cooperative close", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline watchdog is POSIX-only"); + return; + } + + await withTempDir("acpx-client-lifeline-close-", async (tempDir) => { + const helper = await writeLifelineFixture(tempDir); + const client = makeClient({ + agentCommand: [JSON.stringify(process.execPath), JSON.stringify(MOCK_AGENT_PATH)].join(" "), + }); + let watchdog: ChildProcess | undefined; + + await withEnv({ [LIFELINE_HELPER_ENV]: helper }, async () => { + try { + await client.start(); + watchdog = asInternals(client).lifelineWatchdog; + assert(watchdog?.pid, "lifeline watchdog handle must be retained"); + + await client.close(); + await waitForChildExit(watchdog, 2_000); + + assert.equal(watchdog.exitCode, 0, "watchdog must exit cleanly after release"); + assert.equal(watchdog.signalCode, null, "watchdog must not be killed by raw PID signal"); + assert.equal(asInternals(client).lifelineWatchdog, undefined); + } finally { + await client.close(); + killProcessIfAlive(watchdog?.pid); + } + }); + }); +}); + +test("AcpClient keeps the bridge attached when the lifeline helper is unavailable", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline watchdog is POSIX-only"); + return; + } + + await withTempDir("acpx-client-lifeline-unavailable-", async (tempDir) => { + const missingHelper = path.join(tempDir, "missing-lifeline-helper"); + const bridgePidFile = path.join(tempDir, "bridge.pid"); + const client = makeClient({ + agentCommand: [ + JSON.stringify(process.execPath), + JSON.stringify(MOCK_AGENT_PATH), + "--stay-alive-after-stdin-end", + "--pid-file", + JSON.stringify(bridgePidFile), + ].join(" "), + }); + let bridgePid: number | undefined; + + await withEnv({ [LIFELINE_HELPER_ENV]: missingHelper }, async () => { + try { + await client.start(); + await waitUntil(() => fileExists(bridgePidFile)); + + bridgePid = Number((await fs.readFile(bridgePidFile, "utf8")).trim()); + assert(Number.isInteger(bridgePid) && bridgePid > 0, "bridge PID must be valid"); + assert.equal(asInternals(client).agentGroupLeader, false); + assert.equal(asInternals(client).lifelineWatchdog, undefined); + + await client.close(); + + const bridgeExited = await waitUntil(() => Promise.resolve(!isProcessAlive(bridgePid))); + assert.equal(bridgeExited, true, `bridge process ${bridgePid} must be dead after close`); + } finally { + await client.close(); + killProcessIfAlive(bridgePid); + } + }); + }); +}); + +test("AcpClient still group-kills bridge descendants at close when the lifeline watchdog fails to start", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline watchdog is POSIX-only"); + return; + } + + await withTempDir("acpx-client-lifeline-watchdog-fail-", async (tempDir) => { + const brokenHelper = path.join(tempDir, "broken-lifeline-helper"); + // An invalid shebang makes the spawn fail with ENOENT on both macOS and Linux. + // A shebang-less file would instead hit the execvp -> /bin/sh fallback on Linux + // (spawn succeeds, then exits quickly), which would not exercise the watchdog-failure path. + await fs.writeFile(brokenHelper, "#!/nonexistent/acpx-lifeline-test-interpreter\n", "utf8"); + await fs.chmod(brokenHelper, 0o755); + const grandchildPidFile = path.join(tempDir, "grandchild.pid"); + const client = makeClient({ + agentCommand: [ + JSON.stringify(process.execPath), + JSON.stringify(MOCK_AGENT_PATH), + "--ignore-sigterm", + "--stay-alive-after-stdin-end", + "--grandchild-pid-file", + JSON.stringify(grandchildPidFile), + ].join(" "), + }); + let grandchildPid: number | undefined; + + await withEnv({ [LIFELINE_HELPER_ENV]: brokenHelper }, async () => { + try { + await client.start(); + await waitUntil(() => fileExists(grandchildPidFile)); + + grandchildPid = Number((await fs.readFile(grandchildPidFile, "utf8")).trim()); + assert( + Number.isInteger(grandchildPid) && grandchildPid > 0, + "grandchild PID must be valid", + ); + assert.equal(asInternals(client).lifelineWatchdog, undefined); + assert.equal(asInternals(client).agentGroupLeader, true); + + await client.close(); + + const grandchildExited = await waitUntil(() => + Promise.resolve(!isProcessAlive(grandchildPid)), + ); + assert.equal( + grandchildExited, + true, + `grandchild ${grandchildPid} must be group-killed at close`, + ); + } finally { + await client.close(); + killProcessIfAlive(grandchildPid); + } + }); + }); +}); + +test("AcpClient stops the previous lifeline watchdog before replacing it", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline watchdog is POSIX-only"); + return; + } + + await withTempDir("acpx-client-lifeline-replace-", async (tempDir) => { + const helper = await writeLifelineFixture(tempDir); + const client = makeClient(); + const internals = asInternals(client); + const firstBridge = spawnDetachedIdleProcess(); + const secondBridge = spawnDetachedIdleProcess(); + let firstWatchdog: ChildProcess | undefined; + let secondWatchdog: ChildProcess | undefined; + + await withEnv({ [LIFELINE_HELPER_ENV]: helper }, async () => { + try { + internals.agentGroupLeader = true; + await internals.startLifelineWatchdog?.(firstBridge); + firstWatchdog = internals.lifelineWatchdog; + assert(firstWatchdog?.pid, "first watchdog must be retained"); + + await internals.startLifelineWatchdog?.(secondBridge); + secondWatchdog = internals.lifelineWatchdog; + assert(secondWatchdog?.pid, "replacement watchdog must be retained"); + assert.notEqual(secondWatchdog.pid, firstWatchdog.pid); + + await waitForChildExit(firstWatchdog, 2_000); + assert.equal(firstWatchdog.exitCode, 0, "old watchdog must receive release"); + assert.equal(firstWatchdog.signalCode, null, "old watchdog must not be killed by PID"); + } finally { + internals.stopLifelineWatchdog?.(); + killProcessIfAlive(firstWatchdog?.pid); + killProcessIfAlive(secondWatchdog?.pid); + killProcessGroupIfAlive(firstBridge.pid); + killProcessGroupIfAlive(secondBridge.pid); + } + }); + }); +}); + test("AcpClient close resets in-memory state and shuts down terminal manager", async () => { const client = makeClient(); const internals = asInternals(client); @@ -1426,6 +1710,55 @@ async function waitUntil( return false; } +async function waitForChildExit(child: ChildProcess, timeoutMs: number): Promise { + if (child.exitCode !== null || child.signalCode !== null) { + return; + } + + let timer: NodeJS.Timeout | undefined; + try { + await Promise.race([ + once(child, "exit"), + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("child did not exit in time")), timeoutMs); + }), + ]); + } finally { + if (timer) { + clearTimeout(timer); + } + } +} + +async function writeLifelineFixture(tempDir: string): Promise { + const helper = path.join(tempDir, "lifeline-helper"); + await fs.writeFile( + helper, + `#!/usr/bin/env node +process.stdin.setEncoding("utf8"); +process.stdin.on("data", (chunk) => { + if (chunk.includes("R")) { + process.exit(0); + } +}); +process.stdin.on("end", () => process.exit(4)); +setInterval(() => {}, 1000); +`, + "utf8", + ); + await fs.chmod(helper, 0o755); + return helper; +} + +function spawnDetachedIdleProcess(): ChildProcess { + const child = spawn(process.execPath, ["--eval", "setInterval(() => {}, 1000);"], { + detached: true, + stdio: "ignore", + }); + child.unref(); + return child; +} + function killProcessIfAlive(pid: number | undefined): void { if (pid === undefined || !isProcessAlive(pid)) { return; @@ -1437,6 +1770,17 @@ function killProcessIfAlive(pid: number | undefined): void { } } +function killProcessGroupIfAlive(pid: number | undefined): void { + if (pid === undefined || !isProcessAlive(pid)) { + return; + } + try { + process.kill(-pid, "SIGKILL"); + } catch { + killProcessIfAlive(pid); + } +} + function makePermissionRequest( sessionId: string, kind: RequestPermissionRequest["toolCall"]["kind"], diff --git a/test/lifeline-test-helper.ts b/test/lifeline-test-helper.ts new file mode 100644 index 00000000..3f155e96 --- /dev/null +++ b/test/lifeline-test-helper.ts @@ -0,0 +1,37 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { resolveLifelineHelper } from "../src/acp/lifeline.js"; + +export const LIFELINE_HELPER_ENV = "ACPX_LIFELINE_HELPER"; + +async function fileExists(filePath: string): Promise { + try { + await fs.access(filePath); + return true; + } catch { + return false; + } +} + +export async function resolveTestLifelineHelper(): Promise { + if (process.platform === "win32") { + return undefined; + } + + const helperName = `lifeline-${process.platform}-${process.arch}`; + const builtHelper = path.resolve("dist", "native", helperName); + const previous = process.env[LIFELINE_HELPER_ENV]; + + try { + if (await fileExists(builtHelper)) { + process.env[LIFELINE_HELPER_ENV] = builtHelper; + } + return resolveLifelineHelper(); + } finally { + if (previous === undefined) { + delete process.env[LIFELINE_HELPER_ENV]; + } else { + process.env[LIFELINE_HELPER_ENV] = previous; + } + } +} diff --git a/test/lifeline.test.ts b/test/lifeline.test.ts new file mode 100644 index 00000000..7fed4f5b --- /dev/null +++ b/test/lifeline.test.ts @@ -0,0 +1,337 @@ +import assert from "node:assert/strict"; +import { execFile, spawn, type ChildProcess } from "node:child_process"; +import { once } from "node:events"; +import fs from "node:fs/promises"; +import path from "node:path"; +import test from "node:test"; +import { ensureLifelineHelper, resolveLifelineHelper } from "../src/acp/lifeline.js"; +import { isProcessAlive } from "../src/process-liveness.js"; +import { getAcpxVersion } from "../src/version.js"; +import { LIFELINE_HELPER_ENV, resolveTestLifelineHelper } from "./lifeline-test-helper.js"; +import { fileExists, withTempDir } from "./runtime-test-helpers.js"; + +test("resolveLifelineHelper ignores helper binaries from the current working directory", async () => { + const previousCwd = process.cwd(); + await withTempDir("acpx-lifeline-cwd-", async (tempDir) => { + const fakeHelper = path.join(tempDir, "dist", "native", "lifeline-darwin-testarch"); + await fs.mkdir(path.dirname(fakeHelper), { recursive: true }); + await fs.writeFile(fakeHelper, "", "utf8"); + // Make it executable so the assertion proves the CWD is ignored, not merely + // that the file fails the X_OK resolution check. + await fs.chmod(fakeHelper, 0o755); + + process.chdir(tempDir); + try { + assert.equal(resolveLifelineHelper("darwin", "testarch"), undefined); + } finally { + process.chdir(previousCwd); + } + }); +}); + +test("resolveLifelineHelper accepts an existing absolute env override", async () => { + const previous = process.env[LIFELINE_HELPER_ENV]; + await withTempDir("acpx-lifeline-env-", async (tempDir) => { + const helper = path.join(tempDir, "lifeline-helper"); + await fs.writeFile(helper, "", "utf8"); + await fs.chmod(helper, 0o755); + process.env[LIFELINE_HELPER_ENV] = helper; + + try { + assert.equal(resolveLifelineHelper("darwin", "testarch"), helper); + } finally { + if (previous === undefined) { + delete process.env[LIFELINE_HELPER_ENV]; + } else { + process.env[LIFELINE_HELPER_ENV] = previous; + } + } + }); +}); + +test("resolveLifelineHelper rejects an existing non-executable absolute env override", async () => { + await withTempDir("acpx-lifeline-env-nonexec-", async (tempDir) => { + const helper = path.join(tempDir, "lifeline-helper"); + await fs.writeFile(helper, "", { encoding: "utf8", mode: 0o644 }); + await fs.chmod(helper, 0o644); + + await withEnv({ [LIFELINE_HELPER_ENV]: helper }, async () => { + assert.equal(resolveLifelineHelper("darwin", "testarch"), undefined); + }); + }); +}); + +test("ensureLifelineHelper compiles the packaged lifeline source into the user cache", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline helper is POSIX-only"); + return; + } + if (!(await commandExists(process.env.CC ?? "cc"))) { + t.skip("cc is unavailable"); + return; + } + + await withTempDir("acpx-lifeline-cache-", async (homeDir) => { + const arch = `cache-test-${process.pid}`; + await withEnv( + { + HOME: homeDir, + [LIFELINE_HELPER_ENV]: undefined, + }, + async () => { + const helper = await ensureLifelineHelper({ arch }); + assert(helper, "helper must compile into cache"); + assert.equal(path.dirname(helper), path.join(homeDir, ".acpx", "native")); + assert.match( + path.basename(helper), + new RegExp(`^lifeline-${escapeRegExp(getAcpxVersion())}-`), + ); + + const firstStat = await fs.stat(helper); + const second = await ensureLifelineHelper({ arch }); + const secondStat = await fs.stat(helper); + + assert.equal(second, helper); + assert.equal(secondStat.mtimeMs, firstStat.mtimeMs, "cache hit must not recompile"); + }, + ); + }); +}); + +test("ensureLifelineHelper records lazy compile failures and does not retry in process", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline helper is POSIX-only"); + return; + } + + await withTempDir("acpx-lifeline-cache-fail-", async (homeDir) => { + const arch = `cache-fail-test-${process.pid}`; + const messages: string[] = []; + await withEnv( + { + CC: path.join(homeDir, "missing-cc"), + HOME: homeDir, + [LIFELINE_HELPER_ENV]: undefined, + }, + async () => { + assert.equal( + await ensureLifelineHelper({ arch, log: (message: string) => messages.push(message) }), + undefined, + ); + assert.equal( + await ensureLifelineHelper({ arch, log: (message: string) => messages.push(message) }), + undefined, + ); + + const nativeDir = path.join(homeDir, ".acpx", "native"); + const markers = (await fs.readdir(nativeDir)).filter((entry) => entry.endsWith(".failed")); + assert.equal(markers.length, 1); + assert.equal(messages.length, 1, "compile failure should log once per process"); + }, + ); + }); +}); + +test("native lifeline disarms when the bridge process group disappears before owner EOF", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline helper is POSIX-only"); + return; + } + + const helper = await resolveTestLifelineHelper(); + if (!helper) { + t.skip("lifeline helper binary is unavailable"); + return; + } + + const bridge = spawn(process.execPath, ["--eval", "setTimeout(() => process.exit(0), 50);"], { + detached: true, + stdio: "ignore", + }); + assert(bridge.pid, "bridge must receive a PID"); + const bridgeExit = once(bridge, "exit"); + bridge.unref(); + + let watchdog: ChildProcess | undefined; + + try { + watchdog = spawn(helper, [String(bridge.pid)], { + stdio: ["pipe", "ignore", "ignore"], + }); + await once(watchdog, "spawn"); + await bridgeExit; + + await waitForChildExit(watchdog, 2_000); + assert.equal(watchdog.exitCode, 0, "lifeline should exit cleanly after disarming"); + } finally { + killProcessIfAlive(watchdog?.pid); + killProcessGroupIfAlive(bridge.pid); + } +}); + +test("native lifeline SIGKILLs TERM-resistant descendants after the bridge exits", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline helper is POSIX-only"); + return; + } + + const helper = await resolveTestLifelineHelper(); + if (!helper) { + t.skip("lifeline helper binary is unavailable"); + return; + } + + await withTempDir("acpx-lifeline-term-resistant-descendant-", async (tempDir) => { + const bridgePidFile = path.join(tempDir, "bridge.pid"); + const grandchildPidFile = path.join(tempDir, "grandchild.pid"); + const grandchildScript = "process.on('SIGTERM', () => {}); setInterval(() => {}, 1000);"; + const bridgeScript = ` +const { spawn } = require("node:child_process"); +const fs = require("node:fs"); +const grandchild = spawn(process.execPath, ["--eval", ${JSON.stringify(grandchildScript)}], { + stdio: "ignore", +}); +if (!grandchild.pid) { + process.exit(2); +} +grandchild.unref(); +fs.writeFileSync(${JSON.stringify(grandchildPidFile)}, String(grandchild.pid) + "\\n", "utf8"); +fs.writeFileSync(${JSON.stringify(bridgePidFile)}, String(process.pid) + "\\n", "utf8"); +setTimeout(() => process.exit(0), 50); +`; + + const bridge = spawn(process.execPath, ["--eval", bridgeScript], { + detached: true, + stdio: "ignore", + }); + assert(bridge.pid, "bridge must receive a PID"); + const bridgeExit = once(bridge, "exit"); + bridge.unref(); + + let grandchildPid: number | undefined; + let watchdog: ChildProcess | undefined; + + try { + assert.equal(await waitUntil(() => fileExists(bridgePidFile)), true); + assert.equal(await waitUntil(() => fileExists(grandchildPidFile)), true); + grandchildPid = Number((await fs.readFile(grandchildPidFile, "utf8")).trim()); + assert(Number.isInteger(grandchildPid) && grandchildPid > 0, "grandchild PID must be valid"); + assert.equal(isProcessAlive(grandchildPid), true, "grandchild must start alive"); + + watchdog = spawn(helper, [String(bridge.pid)], { + stdio: ["pipe", "ignore", "ignore"], + }); + await once(watchdog, "spawn"); + await bridgeExit; + watchdog.stdin?.end(); + await once(watchdog, "exit"); + + assert.equal( + await waitUntil(() => Promise.resolve(!isProcessAlive(grandchildPid))), + true, + "TERM-resistant grandchild must be killed by lifeline SIGKILL fallback", + ); + } finally { + killProcessIfAlive(watchdog?.pid); + killProcessGroupIfAlive(bridge.pid); + killProcessIfAlive(grandchildPid); + } + }); +}); + +async function commandExists(command: string): Promise { + return await new Promise((resolve) => { + const child = execFile(command, ["--version"], (error: Error | null) => { + resolve(!error); + }); + child.on("error", () => resolve(false)); + }); +} + +function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"); +} + +async function waitUntil( + condition: () => Promise, + timeoutMs = 2_000, + pollMs = 50, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await condition()) { + return true; + } + await new Promise((resolve) => setTimeout(resolve, pollMs)); + } + return false; +} + +async function waitForChildExit(child: ChildProcess, timeoutMs: number): Promise { + if (child.exitCode !== null || child.signalCode !== null) { + return; + } + + let timer: NodeJS.Timeout | undefined; + try { + await Promise.race([ + once(child, "exit"), + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("child did not exit in time")), timeoutMs); + }), + ]); + } finally { + if (timer) { + clearTimeout(timer); + } + } +} + +function killProcessIfAlive(pid: number | undefined): void { + if (pid === undefined || !isProcessAlive(pid)) { + return; + } + try { + process.kill(pid, "SIGKILL"); + } catch { + // best effort cleanup + } +} + +function killProcessGroupIfAlive(pgid: number | undefined): void { + if (pgid === undefined) { + return; + } + try { + process.kill(-pgid, "SIGKILL"); + } catch { + // best effort cleanup + } +} + +async function withEnv( + entries: Record, + run: () => Promise, +): Promise { + const previous = new Map(); + for (const [key, value] of Object.entries(entries)) { + previous.set(key, process.env[key]); + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + } + + try { + return await run(); + } finally { + for (const [key, value] of previous) { + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + } + } +} diff --git a/test/mock-agent.ts b/test/mock-agent.ts index 18923832..0ee08e8e 100644 --- a/test/mock-agent.ts +++ b/test/mock-agent.ts @@ -70,10 +70,12 @@ type MockAgentOptions = { ignoreSigterm: boolean; cancelDelayMs: number; stayAliveAfterStdinEnd: boolean; + failInitialize: boolean; /** If set, the agent writes its PID to this path at startup (before ACP handshake). */ pidFile?: string; /** If set, the agent spawns a long-lived child and writes its PID to this path. */ grandchildPidFile?: string; + grandchildIgnoreSigterm: boolean; }; type SessionState = { @@ -384,8 +386,10 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { let cancelDelayMs = 0; let hangOnNewSession = false; let stayAliveAfterStdinEnd = false; + let failInitialize = false; let pidFile: string | undefined; let grandchildPidFile: string | undefined; + let grandchildIgnoreSigterm = false; for (let index = 0; index < argv.length; index += 1) { const token = argv[index]; @@ -524,6 +528,11 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { continue; } + if (token === "--fail-initialize") { + failInitialize = true; + continue; + } + if (token === "--cancel-delay-ms") { cancelDelayMs = parsePositiveIntegerOption(argv, index + 1, token); index += 1; @@ -547,6 +556,11 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { continue; } + if (token === "--grandchild-ignore-sigterm") { + grandchildIgnoreSigterm = true; + continue; + } + if (token === "--claude-agent-acp") { continue; } @@ -614,8 +628,10 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { ignoreSigterm, cancelDelayMs, stayAliveAfterStdinEnd, + failInitialize, pidFile, grandchildPidFile, + grandchildIgnoreSigterm, }; } @@ -767,6 +783,10 @@ class MockAgent implements Agent { } async initialize(): Promise { + if (this.options.failInitialize) { + throw RequestError.internalError({ reason: "requested failure" }, "initialize failed"); + } + const sessionCapabilities = { ...(this.options.supportsCloseSession ? { close: {} } : {}), ...(this.options.supportsListSessions ? { list: {} } : {}), @@ -1500,10 +1520,11 @@ const input = Readable.toWeb(process.stdin) as ReadableStream; const stream = ndJsonStream(output, input); const mockAgentOptions = parseMockAgentOptions(process.argv.slice(2)); -function spawnLongLivedGrandchild(pidFile: string): void { - const grandchild = spawn(process.execPath, ["--eval", "setInterval(() => {}, 1000);"], { - stdio: "ignore", - }); +function spawnLongLivedGrandchild(pidFile: string, ignoreSigterm: boolean): void { + const script = ignoreSigterm + ? "process.on('SIGTERM', () => {}); setInterval(() => {}, 1000);" + : "setInterval(() => {}, 1000);"; + const grandchild = spawn(process.execPath, ["--eval", script], { stdio: "ignore" }); if (!grandchild.pid) { throw new Error("long-lived grandchild did not receive a PID"); } @@ -1518,7 +1539,10 @@ if (mockAgentOptions.pidFile) { } if (mockAgentOptions.grandchildPidFile) { - spawnLongLivedGrandchild(mockAgentOptions.grandchildPidFile); + spawnLongLivedGrandchild( + mockAgentOptions.grandchildPidFile, + mockAgentOptions.grandchildIgnoreSigterm, + ); } if (mockAgentOptions.ignoreSigterm) { diff --git a/test/package-scripts.test.ts b/test/package-scripts.test.ts index 4156c5f9..2cbc5ad6 100644 --- a/test/package-scripts.test.ts +++ b/test/package-scripts.test.ts @@ -4,6 +4,7 @@ import path from "node:path"; import test from "node:test"; type PackageJson = { + files?: string[]; scripts?: Record; dependencies?: Record; devDependencies?: Record; @@ -84,3 +85,12 @@ test("test scripts build packaged output before running package-bin smoke tests" assert.match(pkg.scripts?.test ?? "", /^pnpm run build && pnpm run build:test && /); assert.match(pkg.scripts?.["test:coverage"] ?? "", /^pnpm run build && pnpm run build:test && /); }); + +test("package ships lifeline source but prepack does not ship host-native helper binaries", () => { + const pkg = readPackageJson(); + const prepack = pkg.scripts?.prepack ?? ""; + + assert(pkg.files?.includes("native"), "native source directory must be included in package"); + assert.doesNotMatch(prepack, /\bbuild:native\b/); + assert.match(prepack, /dist\/native/); +}); diff --git a/test/queue-owner-lifecycle.test.ts b/test/queue-owner-lifecycle.test.ts index 67963b56..1f26afc4 100644 --- a/test/queue-owner-lifecycle.test.ts +++ b/test/queue-owner-lifecycle.test.ts @@ -18,6 +18,7 @@ import { describe, it } from "node:test"; import { fileURLToPath } from "node:url"; import { isProcessAlive } from "../src/cli/queue/lease-store.js"; import { queueLockFilePath, queueSocketPath } from "../src/cli/queue/paths.js"; +import { LIFELINE_HELPER_ENV, resolveTestLifelineHelper } from "./lifeline-test-helper.js"; import { makeSessionRecord, withTempHome, writeSessionRecordFile } from "./runtime-test-helpers.js"; const CLI_PATH = fileURLToPath(new URL("../src/cli.js", import.meta.url)); @@ -96,6 +97,38 @@ function waitForProcessExit( }); } +async function readPidFile(pidFilePath: string, label: string): Promise { + const raw = (await fs.readFile(pidFilePath, "utf8")).trim(); + const pid = Number(raw); + assert(Number.isInteger(pid) && pid > 0, `${label} PID must be a positive integer`); + return pid; +} + +async function waitForProcessesToExit( + pids: number[], + timeoutMs = 2_000, + pollMs = 50, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (pids.every((pid) => !isProcessAlive(pid))) { + return; + } + await new Promise((resolve) => setTimeout(resolve, pollMs)); + } +} + +function killProcessIfAlive(pid: number | undefined): void { + if (pid === undefined || !isProcessAlive(pid)) { + return; + } + try { + process.kill(pid, "SIGKILL"); + } catch { + // best effort cleanup for intentionally orphaned RED processes + } +} + describe("queue owner lifecycle — graceful SIGTERM shutdown", () => { it("exits with code 0 and releases its lease when it receives SIGTERM", async () => { if (process.platform === "win32") { @@ -303,6 +336,142 @@ describe("queue owner lifecycle — graceful SIGTERM shutdown", () => { }); }); +describe("queue owner lifecycle — bridge lifeline on abrupt owner death", () => { + it("kills the bridge process tree when the queue owner is SIGKILLed", async (t) => { + if (process.platform === "win32") { + t.skip("lifeline watchdog is POSIX-only"); + return; + } + + const helper = await resolveTestLifelineHelper(); + if (!helper) { + t.skip("lifeline helper binary is unavailable"); + return; + } + + await withTempHome("acpx-lifecycle-lifeline-sigkill-", async (homeDir) => { + const cwd = path.join(homeDir, "workspace"); + await fs.mkdir(cwd, { recursive: true }); + + const bridgePidFilePath = path.join(homeDir, "mock-agent-lifeline.pid"); + const grandchildPidFilePath = path.join(homeDir, "mock-agent-lifeline-grandchild.pid"); + + const record = makeSessionRecord({ + acpxRecordId: "lifecycle-lifeline-sigkill-test", + acpSessionId: "lifecycle-lifeline-sigkill-session", + agentCommand: [ + "node", + JSON.stringify(MOCK_AGENT_PATH), + "--pid-file", + JSON.stringify(bridgePidFilePath), + "--grandchild-pid-file", + JSON.stringify(grandchildPidFilePath), + "--grandchild-ignore-sigterm", + "--stay-alive-after-stdin-end", + ].join(" "), + cwd, + }); + await writeSessionRecordFile(homeDir, record); + + const socketPath = queueSocketPath(record.acpxRecordId, homeDir); + const payload = JSON.stringify({ + sessionId: record.acpxRecordId, + permissionMode: "approve-reads", + }); + + const child = spawn(process.execPath, [CLI_PATH, "__queue-owner"], { + env: { + ...process.env, + HOME: homeDir, + ACPX_QUEUE_OWNER_PAYLOAD: payload, + [LIFELINE_HELPER_ENV]: helper, + }, + stdio: ["ignore", "ignore", "pipe"], + }); + + const stderrChunks: Buffer[] = []; + child.stderr?.on("data", (chunk: Buffer) => stderrChunks.push(chunk)); + + let queueSocket: net.Socket | undefined; + let lines: readline.Interface | undefined; + let bridgePid: number | undefined; + let grandchildPid: number | undefined; + + try { + await waitUntil(() => fileExists(socketPath)); + + queueSocket = await new Promise((resolve, reject) => { + const socket = net.createConnection(socketPath); + socket.setEncoding("utf8"); + socket.once("connect", () => resolve(socket)); + socket.once("error", reject); + }); + + queueSocket.write( + `${JSON.stringify({ + type: "submit_prompt", + requestId: "req-lifeline-sigkill-test", + message: "sleep 10000", + permissionMode: "approve-reads", + waitForCompletion: true, + })}\n`, + ); + + lines = readline.createInterface({ input: queueSocket }); + const iter = lines[Symbol.asyncIterator](); + const acceptedRaw = await Promise.race([ + iter.next(), + new Promise((_, reject) => + setTimeout(() => reject(new Error("timeout waiting for accepted")), 5_000), + ), + ]); + const accepted = JSON.parse((acceptedRaw as IteratorYieldResult).value) as { + type: string; + }; + assert.equal(accepted.type, "accepted", "queue owner must acknowledge the prompt"); + + await waitUntil(() => fileExists(bridgePidFilePath), 8_000); + await waitUntil(() => fileExists(grandchildPidFilePath), 8_000); + + bridgePid = await readPidFile(bridgePidFilePath, "bridge"); + grandchildPid = await readPidFile(grandchildPidFilePath, "grandchild"); + assert.equal(isProcessAlive(bridgePid), true, "bridge must be alive before SIGKILL"); + assert.equal( + isProcessAlive(grandchildPid), + true, + "grandchild must be alive before SIGKILL", + ); + + child.kill("SIGKILL"); + const { signal } = await waitForProcessExit(child, 5_000); + const stderr = Buffer.concat(stderrChunks).toString("utf8"); + assert.equal(signal, "SIGKILL", `queue owner must be SIGKILLed; stderr=${stderr}`); + + await waitForProcessesToExit([bridgePid, grandchildPid], 2_000); + assert.deepEqual( + { + bridgeAliveAfterOwnerSigkill: isProcessAlive(bridgePid), + grandchildAliveAfterOwnerSigkill: isProcessAlive(grandchildPid), + }, + { + bridgeAliveAfterOwnerSigkill: false, + grandchildAliveAfterOwnerSigkill: false, + }, + "bridge and grandchild must die after abrupt queue-owner death", + ); + } finally { + lines?.close(); + queueSocket?.destroy(); + if (child.exitCode == null && child.signalCode == null) { + child.kill("SIGKILL"); + } + killProcessIfAlive(bridgePid); + killProcessIfAlive(grandchildPid); + } + }); + }); +}); + describe("queue owner lifecycle — bridge process death on SIGTERM", () => { // Verifies that when the queue owner is SIGTERMed while a prompt is in // flight, the agent bridge process (mock-agent) is also killed by the