diff --git a/extensions/pi-subagents/src/extension/index.ts b/extensions/pi-subagents/src/extension/index.ts index c5801aa..cee17df 100644 --- a/extensions/pi-subagents/src/extension/index.ts +++ b/extensions/pi-subagents/src/extension/index.ts @@ -19,6 +19,7 @@ import { } from "../shared/types.ts"; import { getPiSpawnCommand } from "../runs/shared/pi-spawn.ts"; import { buildPiArgs, cleanupTempDir } from "../runs/shared/pi-args.ts"; +import { PI_SUBAGENT_LIFELINE_FD } from "../runs/shared/subagent-prompt-runtime.ts"; import { SpawnSubagentParams, type SpawnSubagentParamsLike, @@ -91,6 +92,24 @@ interface StartChildHooks { const DEFAULT_TIMEOUT_SECONDS = 600; const runningChildren = new Map(); + +// Lifeline pipes keyed by record.id. Each entry holds the parent's write end +// (child.stdin) of the anonymous lifeline pipe. The parent never writes to it; +// holding it open keeps the child's read end alive. When the parent process +// dies the kernel closes all FDs, the child sees EOF, and self-terminates. +function closeLifeline(recordId: string): void { + const stream = lifelines.get(recordId); + if (!stream) return; + lifelines.delete(recordId); + try { stream.end(); } catch { /* best-effort */ } + try { stream.destroy(); } catch { /* best-effort */ } +} + +function destroyAllLifelines(): void { + for (const id of [...lifelines.keys()]) closeLifeline(id); +} + +const lifelines = new Map(); const activeCohorts = new Map< string, { id: string; createdAt: number; turnIndex?: number } @@ -1123,9 +1142,13 @@ function startChild( ...built.env, ...getSubagentDepthEnv(resolveCurrentMaxSubagentDepth()), }; + // Use stdin (fd 0) as the anonymous lifeline pipe. + // The parent holds child.stdin open without writing; the child + // runtime watches fd 0 for EOF to detect parent death. + env[PI_SUBAGENT_LIFELINE_FD] = "0"; child = spawn(spawnSpec.command, spawnSpec.args, { cwd: record.cwd, - stdio: ["ignore", "pipe", "pipe"], + stdio: ["pipe", "pipe", "pipe"], env, }); } catch (error) { @@ -1154,6 +1177,8 @@ function startChild( if (!child.stdout || !child.stderr) { throw new Error("Subagent process stdio pipes were not created."); } + // Store the lifeline write end (child.stdin) before piping stdout/stderr. + if (child.stdin) lifelines.set(record.id, child.stdin); child.stdout.pipe(stdoutStream); child.stderr.pipe(stderrStream); @@ -1173,6 +1198,8 @@ function startChild( ]); runningChildren.delete(record.id); + // Release the lifeline on normal child completion. + closeLifeline(record.id); cleanupTempDir(built.tempDir); const stdout = fs.existsSync(record.stdoutFile) @@ -1356,6 +1383,9 @@ export default function registerSubagentExtension(pi: ExtensionAPI): void { }); pi.on("session_shutdown", (_event, _ctx) => { + // Close all lifelines: the owning session is being torn down. + // This terminates all running children for this parent session. + destroyAllLifelines(); // Clear all widget refresh timers to prevent stale UI state across extension instances. stopAllWidgetRefreshForInstance(); activeCohorts.clear(); diff --git a/extensions/pi-subagents/src/runs/shared/subagent-prompt-runtime.ts b/extensions/pi-subagents/src/runs/shared/subagent-prompt-runtime.ts index d04b261..bd76c17 100644 --- a/extensions/pi-subagents/src/runs/shared/subagent-prompt-runtime.ts +++ b/extensions/pi-subagents/src/runs/shared/subagent-prompt-runtime.ts @@ -1,4 +1,5 @@ import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; +import * as fs from "node:fs"; export const SUBAGENT_INTERCOM_SESSION_NAME_ENV = "PI_SUBAGENT_INTERCOM_SESSION_NAME"; @@ -6,6 +7,12 @@ export const SUBAGENT_RESULT_PATH_ENV = "PI_SUBAGENT_RESULT_PATH"; export const CHILD_SUBAGENT_SYSTEM_LINE = "You are a Pi subagent controlled by another Pi agent."; +/** Env var set by the parent when spawning a managed subagent. + * Value is the child-side file descriptor number of the anonymous lifeline pipe. + * The child watches this fd for EOF; when the parent process dies the kernel + * closes the write end and the child detects the EOF to self-terminate. */ +export const PI_SUBAGENT_LIFELINE_FD = "PI_SUBAGENT_LIFELINE_FD"; + const RESULT_PATH_MARKER = "Your result file:"; const RESULT_PATH_ALIASES = new Set([ "$PI_SUBAGENT_RESULT_PATH", @@ -18,6 +25,37 @@ export function rewriteSubagentPrompt(prompt: string): string { return `${CHILD_SUBAGENT_SYSTEM_LINE}\n\n${prompt}`; } +// ── Lifeline watcher: self-terminate when parent process dies ── + +function setupLifelineWatcher(): void { + const lifelineFdRaw = process.env[PI_SUBAGENT_LIFELINE_FD]; + if (lifelineFdRaw === undefined) return; + + const fd = parseInt(lifelineFdRaw, 10); + if (!Number.isFinite(fd) || fd < 0) return; + + // Create a read stream on the lifeline fd. When the parent process + // dies the kernel closes the write end of the pipe; the child sees EOF + // and the stream emits 'end' → self-terminate via SIGTERM. + // We use a dedicated ReadStream (not process.stdin) so we never + // conflict with whatever Pi may do with its own stdin handling. + // We use fd option to read directly from the lifeline fd; the path + // argument is not used when fd is supplied but required by the type. + const lifeline = fs.createReadStream("", { fd, autoClose: false }); + lifeline.on("end", () => { + process.kill(process.pid, "SIGTERM"); + }); + lifeline.on("error", () => { + // If the fd is already closed or invalid, treat as parent death. + process.kill(process.pid, "SIGTERM"); + }); + // Start flowing so libuv polls the fd for readability/EOF. + lifeline.resume(); +} + +// Run at module load time so the watcher is active before any Pi handlers fire. +setupLifelineWatcher(); + export default function registerSubagentPromptRuntime(pi: ExtensionAPI): void { pi.on("tool_call", (event) => { if (!FILE_TOOL_NAMES.has(event.toolName)) return; diff --git a/extensions/pi-subagents/test/unit/minimal-subagents.test.ts b/extensions/pi-subagents/test/unit/minimal-subagents.test.ts index f5cda0f..33b4dd4 100644 --- a/extensions/pi-subagents/test/unit/minimal-subagents.test.ts +++ b/extensions/pi-subagents/test/unit/minimal-subagents.test.ts @@ -1128,8 +1128,8 @@ test("Phase 7.8: headless liveness — stdio pipes keep parent process alive nat assert.ok(childSpawnOptions, "extension child spawn call is present"); assert.match( childSpawnOptions, - /stdio:\s*\["ignore", "pipe", "pipe"\]/, - "extension child spawn uses default pipe stdio for stdout/stderr", + /stdio:\s*\["pipe", "pipe", "pipe"\]/, + "extension child spawn uses pipe stdio with lifeline on stdin", ); assert.doesNotMatch( childSpawnOptions, @@ -1559,3 +1559,273 @@ test("cohort: reconcile preserves cohort metadata", async () => { cleanupTestCtx(ctx, sessionId); } }); + +// ── Lifeline: process-death cascade via anonymous pipe ── + +import { PI_SUBAGENT_LIFELINE_FD } from "../../src/runs/shared/subagent-prompt-runtime.ts"; + +function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +async function waitForProcessDeath(pid: number, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (!isProcessAlive(pid)) return; + await new Promise((r) => setTimeout(r, 50)); + } + throw new Error(`process ${pid} did not die within ${timeoutMs}ms`); +} + +async function waitForFile(filePath: string, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + try { + return fs.readFileSync(filePath, "utf-8"); + } catch { + await new Promise((r) => setTimeout(r, 50)); + } + } + throw new Error(`file ${filePath} not created within ${timeoutMs}ms`); +} + +test("lifeline: subagent-prompt-runtime exposes lifeline env constant", () => { + assert.equal(typeof PI_SUBAGENT_LIFELINE_FD, "string"); + assert.ok(PI_SUBAGENT_LIFELINE_FD.length > 0); +}); + +test("lifeline: session_shutdown terminates children via lifeline", async () => { + const mockPi = createMockPi(); + mockPi.install(); + mockPi.onCall({ + output: "lifeline-kill done", + exitCode: 0, + delay: 60, + keepAliveAfterFinalMessageMs: 300, + }); + + const { sessionId, ctx } = makeTestCtx("pi-subagents-lifeline-kill"); + const fake = makeFakeCtx(sessionId, ctx.cwd, false); + + const sessionShutdownHandlers: Array<(_event: unknown, ctx: unknown) => void> = []; + const registered = new Map(); + registerSubagentExtension({ + registerTool(tool: any) { registered.set(tool.name, tool); }, + sendMessage() {}, + on(event: string, handler: any) { + if (event === "session_shutdown") sessionShutdownHandlers.push(handler); + }, + } as never); + const spawnTool = registered.get("spawn_subagent"); + assert.ok(spawnTool, "spawn_subagent tool registered"); + + try { + const result = await spawnTool.execute( + "lifeline-kill-child", + { task: "lifeline kill test", timeout: 30 }, + new AbortController().signal, + undefined, + fake.ctx, + ); + const childId = result.details.id; + await waitForSubagentRecord(sessionId, childId, (r) => r.running === true); + const record = readPersistedRecord(sessionId, childId); + assert.equal(record.running, true); + assert.ok(typeof record.pid === "number"); + assert.ok(isProcessAlive(record.pid), "child must be alive before lifeline close"); + + assert.ok(sessionShutdownHandlers.length >= 1, "session_shutdown handler registered"); + sessionShutdownHandlers.forEach((h) => h(undefined, fake.ctx)); + + await waitForProcessDeath(record.pid, 2000); + assert.equal(isProcessAlive(record.pid), false, "child must die after session_shutdown"); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +}); + +test("lifeline: agent_end does NOT kill child process", async () => { + const mockPi = createMockPi(); + mockPi.install(); + mockPi.onCall({ output: "agent-end-survive done", exitCode: 0, delay: 200 }); + + const { sessionId, ctx } = makeTestCtx("pi-subagents-agent-end-survive"); + const fake = makeFakeCtx(sessionId, ctx.cwd, false); + + const agentEndHandlers: Array<(_event: unknown, ctx: unknown) => void> = []; + const registered = new Map(); + registerSubagentExtension({ + registerTool(tool: any) { registered.set(tool.name, tool); }, + sendMessage() {}, + on(event: string, handler: any) { + if (event === "agent_end") agentEndHandlers.push(handler); + }, + } as never); + const spawnTool = registered.get("spawn_subagent"); + + try { + const result = await spawnTool.execute( + "agent-end-survive-child", + { task: "survive agent_end", timeout: 30 }, + new AbortController().signal, + undefined, + fake.ctx, + ); + const childId = result.details.id; + await waitForSubagentRecord(sessionId, childId, (r) => r.running === true); + const record = readPersistedRecord(sessionId, childId); + assert.ok(isProcessAlive(record.pid), "child must be alive before agent_end"); + + assert.ok(agentEndHandlers.length >= 1); + agentEndHandlers.forEach((h) => h(undefined, fake.ctx)); + + await new Promise((r) => setTimeout(r, 80)); + assert.ok(isProcessAlive(record.pid), "child must survive agent_end"); + + await waitForSubagentRecord(sessionId, childId, (r) => r.running === false); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +}); + +test("lifeline: abrupt parent SIGKILL cascades to child termination", async () => { + const extensionPath = path.join(projectRoot, "src", "index.ts"); + const testDir = fs.mkdtempSync(path.join(os.tmpdir(), "pi-subagents-cascade-")); + const parentSessionId = testDir; + const sessionFile = path.join(testDir, "session.jsonl"); + + const parentScript = ` + import fs from "node:fs"; + import registerSubagentExtension from ${JSON.stringify(extensionPath)}; + + const registered = new Map(); + registerSubagentExtension({ + registerTool(tool) { registered.set(tool.name, tool); }, + sendMessage() {}, + on() {}, + }); + + const ctx = { + cwd: ${JSON.stringify(testDir)}, + hasUI: false, + sessionManager: { + getSessionFile: () => ${JSON.stringify(sessionFile)}, + getSessionId: () => ${JSON.stringify(sessionFile)}, + }, + }; + + const result = await registered.get("spawn_subagent").execute( + "cascade-child", + { task: "long-running cascade child", timeout: 60 }, + new AbortController().signal, + undefined, + ctx, + ); + + fs.writeFileSync(${JSON.stringify(path.join(testDir, "ready"))}, JSON.stringify({ + childId: result.details.id, + }), "utf-8"); + + setTimeout(() => {}, 60000); + `; + + const parentProc = spawn( + process.execPath, + ["--experimental-strip-types", "--input-type=module", "-e", parentScript], + { cwd: projectRoot, env: { ...process.env, PI_NO_COLOR: "1" }, stdio: ["ignore", "pipe", "pipe"] }, + ); + + try { + const readyContent = await waitForFile(path.join(testDir, "ready"), 10000); + const { childId } = JSON.parse(readyContent) as { childId: string }; + + const childRecord = readPersistedRecord(parentSessionId, childId); + assert.ok(childRecord, "child record must exist"); + assert.equal(childRecord.running, true); + const childPid = childRecord.pid; + assert.ok(typeof childPid === "number" && childPid > 0); + assert.ok(isProcessAlive(childPid), "child must be alive before parent kill"); + + assert.ok(parentProc.pid); + process.kill(parentProc.pid, "SIGKILL"); + + await waitForProcessDeath(childPid, 5000); + assert.equal(isProcessAlive(childPid), false, "child must die after parent SIGKILL"); + } finally { + try { process.kill(parentProc.pid as number, "SIGKILL"); } catch {} + try { fs.rmSync(testDir, { recursive: true, force: true }); } catch {} + } +}); + +test("lifeline: recursive cascade — grandparent death kills parent subagent", async () => { + const extensionPath = path.join(projectRoot, "src", "index.ts"); + const testDir = fs.mkdtempSync(path.join(os.tmpdir(), "pi-subagents-recursive-")); + const sessionFile = path.join(testDir, "session.jsonl"); + + const grandparentScript = ` + import fs from "node:fs"; + import registerSubagentExtension from ${JSON.stringify(extensionPath)}; + + const registered = new Map(); + registerSubagentExtension({ + registerTool(tool) { registered.set(tool.name, tool); }, + sendMessage() {}, + on() {}, + }); + + const ctx = { + cwd: ${JSON.stringify(testDir)}, + hasUI: false, + sessionManager: { + getSessionFile: () => ${JSON.stringify(sessionFile)}, + getSessionId: () => ${JSON.stringify(sessionFile)}, + }, + }; + + const result = await registered.get("spawn_subagent").execute( + "recursive-parent", + { task: "parent that spawns child", timeout: 60 }, + new AbortController().signal, + undefined, + ctx, + ); + + fs.writeFileSync(${JSON.stringify(path.join(testDir, "ready"))}, JSON.stringify({ + parentId: result.details.id, + }), "utf-8"); + + setTimeout(() => {}, 60000); + `; + + const grandparentProc = spawn( + process.execPath, + ["--experimental-strip-types", "--input-type=module", "-e", grandparentScript], + { cwd: projectRoot, env: { ...process.env, PI_NO_COLOR: "1" }, stdio: ["ignore", "pipe", "pipe"] }, + ); + + try { + const readyContent = await waitForFile(path.join(testDir, "ready"), 15000); + const { parentId } = JSON.parse(readyContent) as { parentId: string }; + + const parentRecord = readPersistedRecord(testDir, parentId); + assert.ok(parentRecord, "parent record must exist"); + const parentPid = parentRecord.pid; + assert.ok(typeof parentPid === "number" && parentPid > 0); + assert.ok(isProcessAlive(parentPid), "parent subagent must be alive"); + + process.kill(grandparentProc.pid as number, "SIGKILL"); + + await waitForProcessDeath(parentPid, 5000); + assert.equal(isProcessAlive(parentPid), false, "parent subagent must die after grandparent SIGKILL"); + } finally { + try { process.kill(grandparentProc.pid as number, "SIGKILL"); } catch {} + try { fs.rmSync(testDir, { recursive: true, force: true }); } catch {} + } +});