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 b9830229..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, @@ -442,6 +443,8 @@ export class AcpClient { private agentStartedAt?: string; 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(); @@ -710,19 +713,81 @@ export class AcpClient { private async spawnAgentProcess( plan: AgentLaunchPlan, ): Promise> { + 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, - buildSpawnCommandOptions(plan.spawnCommand, plan.spawnOptions), + buildSpawnCommandOptions(plan.spawnCommand, { + ...plan.spawnOptions, + detached: agentGroupLeader, + }), ) as ChildProcessByStdio; + 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; @@ -832,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( @@ -1331,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( @@ -1369,6 +1442,8 @@ export class AcpClient { this.initResult = undefined; this.connection = undefined; this.agent = undefined; + this.agentGroupLeader = false; + this.lifelineWatchdog = undefined; } private async terminateAgentProcess( @@ -1376,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) { @@ -1399,21 +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); + return; + } child.kill(signal); + return; } catch { - // best effort + 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 { @@ -2021,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 3483c39f..e26f8701 100644 --- a/test/client.test.ts +++ b/test/client.test.ts @@ -1,7 +1,11 @@ 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"; import test from "node:test"; +import { fileURLToPath } from "node:url"; import type { RequestPermissionRequest, RequestPermissionResponse } from "@agentclientprotocol/sdk"; import { AcpClient, @@ -20,6 +24,11 @@ import { PermissionPromptUnavailableError, 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)); test("parseAcpJsonMessageLine ignores non-object JSON values", () => { for (const line of ["1", "null", '"diagnostic"', "[]", "[{}]"]) { @@ -113,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; @@ -1276,6 +1289,330 @@ 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; + } + + 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 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); @@ -1358,6 +1695,92 @@ 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; +} + +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; + } + try { + process.kill(pid, "SIGKILL"); + } catch { + // best effort cleanup for the intentional RED leak + } +} + +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 840cb135..0ee08e8e 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,13 @@ type MockAgentOptions = { loadReplayText: string; 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 = { @@ -379,7 +385,11 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { let ignoreSigterm = false; 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]; @@ -513,6 +523,16 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { continue; } + if (token === "--stay-alive-after-stdin-end") { + stayAliveAfterStdinEnd = true; + continue; + } + + if (token === "--fail-initialize") { + failInitialize = true; + continue; + } + if (token === "--cancel-delay-ms") { cancelDelayMs = parsePositiveIntegerOption(argv, index + 1, token); index += 1; @@ -530,6 +550,17 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { continue; } + if (token === "--grandchild-pid-file") { + grandchildPidFile = parseOptionValue(argv, index + 1, token); + index += 1; + continue; + } + + if (token === "--grandchild-ignore-sigterm") { + grandchildIgnoreSigterm = true; + continue; + } + if (token === "--claude-agent-acp") { continue; } @@ -596,7 +627,11 @@ function parseMockAgentOptions(argv: string[]): MockAgentOptions { loadReplayText, ignoreSigterm, cancelDelayMs, + stayAliveAfterStdinEnd, + failInitialize, pidFile, + grandchildPidFile, + grandchildIgnoreSigterm, }; } @@ -748,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: {} } : {}), @@ -1481,18 +1520,41 @@ const input = Readable.toWeb(process.stdin) as ReadableStream; const stream = ndJsonStream(output, input); const mockAgentOptions = parseMockAgentOptions(process.argv.slice(2)); +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"); + } + 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, + mockAgentOptions.grandchildIgnoreSigterm, + ); +} + 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, 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