From 55fffee36cde4ad183d67982c6fc8110b1b7ca8a Mon Sep 17 00:00:00 2001 From: Nico Bailon Date: Mon, 27 Jul 2026 10:57:27 +0000 Subject: [PATCH] fix(pi-subagents): bound stdout recovery and restore print mode --- extensions/pi-subagents/CHANGELOG.md | 1 + .../pi-subagents/src/extension/index.ts | 213 +++++++++--------- .../pi-subagents/src/runs/shared/pi-args.ts | 6 +- .../test/unit/minimal-subagents.test.ts | 76 +++++-- 4 files changed, 167 insertions(+), 129 deletions(-) diff --git a/extensions/pi-subagents/CHANGELOG.md b/extensions/pi-subagents/CHANGELOG.md index a2de98a..93676ff 100644 --- a/extensions/pi-subagents/CHANGELOG.md +++ b/extensions/pi-subagents/CHANGELOG.md @@ -14,6 +14,7 @@ ### Added ### Fixed +- Run minimal child subagents in print mode instead of persisting cumulative JSON streaming events, and bound legacy output recovery so oversized logs cannot exceed V8's string limit. - Prevent the parent-death lifeline watcher from keeping successfully settled subagents alive indefinitely. ## [0.24.2] - 2026-05-10 diff --git a/extensions/pi-subagents/src/extension/index.ts b/extensions/pi-subagents/src/extension/index.ts index 471197f..c119f27 100644 --- a/extensions/pi-subagents/src/extension/index.ts +++ b/extensions/pi-subagents/src/extension/index.ts @@ -258,6 +258,9 @@ function extractFinalOutput(stdout: string): string { if (event.message?.role === "assistant") { const text = extractTextFromMessageContent(event.message.content); if (text.trim()) lastAssistant = text.trim(); + } else if (!event.type) { + // Print-mode answers may themselves be valid JSON. + rawLines.push(line); } } catch { rawLines.push(line); @@ -266,6 +269,94 @@ function extractFinalOutput(stdout: string): string { return lastAssistant || rawLines.join("\n").trim(); } +const MAX_RECOVERY_FILE_BYTES = 16 * 1024 * 1024; + +function hasNonEmptyFile(filePath: string | undefined): boolean { + if (!filePath) return false; + try { + return fs.statSync(filePath).size > 0; + } catch { + return false; + } +} + +function readRecoveryFile( + filePath: string | undefined, + label: string, +): { text: string; issue?: string } { + if (!filePath) return { text: "" }; + try { + const size = fs.statSync(filePath).size; + if (size > MAX_RECOVERY_FILE_BYTES) { + return { + text: "", + issue: `${label} exceeds the ${MAX_RECOVERY_FILE_BYTES}-byte recovery limit: ${filePath}`, + }; + } + return { text: fs.readFileSync(filePath, "utf-8") }; + } catch (error) { + const code = + error && typeof error === "object" && "code" in error + ? String(error.code) + : undefined; + if (code === "ENOENT") return { text: "" }; + return { + text: "", + issue: `Could not read ${label} ${filePath}: ${error instanceof Error ? error.message : String(error)}`, + }; + } +} + +function recoverFinalOutput(record: PersistedSubagentRecord): { + output: string; + issue?: string; +} { + try { + recordDiscoveredSessionFile(record); + } catch { + // Discovery is best-effort; stdout remains the compatibility fallback. + } + let issue: string | undefined; + for (const [filePath, label] of [ + [record.sessionFile, "child session"], + [record.stdoutFile, "child stdout"], + ] as const) { + const recovered = readRecoveryFile(filePath, label); + issue ??= recovered.issue; + const output = extractFinalOutput(recovered.text); + if (output) return { output }; + } + return { output: "", issue }; +} + +function ensureResultFile( + record: PersistedSubagentRecord, + stderr: { text: string; issue?: string }, +): void { + if (!record.outputFile || hasNonEmptyFile(record.outputFile)) return; + try { + const recovered = recoverFinalOutput(record); + if (record.error) { + fs.writeFileSync(record.outputFile, "(error)\n", { mode: 0o600 }); + } else if (recovered.output) { + fs.writeFileSync(record.outputFile, `${recovered.output}\n`, { + mode: 0o600, + }); + } else if (stderr.text.trim()) { + record.error = stderr.text.trim(); + fs.writeFileSync(record.outputFile, "(error)\n", { mode: 0o600 }); + } else if (recovered.issue || stderr.issue) { + record.error = recovered.issue ?? stderr.issue; + fs.writeFileSync(record.outputFile, "(error)\n", { mode: 0o600 }); + } else { + fs.writeFileSync(record.outputFile, "(no output)\n", { mode: 0o600 }); + } + } catch (error) { + record.error ??= + `Could not create subagent result: ${error instanceof Error ? error.message : String(error)}`; + } +} + function refreshRecord(record: PersistedSubagentRecord): { record: PersistedSubagentRecord; changed: boolean; @@ -274,40 +365,12 @@ function refreshRecord(record: PersistedSubagentRecord): { return { record, changed: false }; } const refreshed: PersistedSubagentRecord = { ...record }; - const stdout = fs.existsSync(refreshed.stdoutFile) - ? fs.readFileSync(refreshed.stdoutFile, "utf-8") - : ""; - const stderr = fs.existsSync(refreshed.stderrFile) - ? fs.readFileSync(refreshed.stderrFile, "utf-8") - : ""; refreshed.running = false; refreshed.updatedAt = Date.now(); refreshed.completedAt ??= Date.now(); - // Check if subagent already wrote to the result file. - const hasExistingResult = - refreshed.outputFile && - fs.existsSync(refreshed.outputFile) && - fs.readFileSync(refreshed.outputFile, "utf-8").trim().length > 0; - - if (!hasExistingResult) { - const finalOutput = extractFinalOutput(stdout); - if (finalOutput && refreshed.outputFile) { - fs.writeFileSync(refreshed.outputFile, `${finalOutput}\n`, { - mode: 0o600, - }); - } else if (stderr.trim()) { - // Subagent produced only stderr, no stdout output. - if (!refreshed.error) refreshed.error = stderr.trim(); - if (refreshed.outputFile) { - fs.writeFileSync(refreshed.outputFile, "(error)\n", { mode: 0o600 }); - } - } else if (refreshed.outputFile) { - // Edge case: neither stdout nor stderr produced content. - // Write a placeholder so the parent gets a valid result file. - fs.writeFileSync(refreshed.outputFile, "(no output)\n", { mode: 0o600 }); - } - } + const stderr = readRecoveryFile(refreshed.stderrFile, "child stderr"); + ensureResultFile(refreshed, stderr); refreshed.result = refreshed.outputFile; return { record: refreshed, changed: true }; } @@ -907,37 +970,7 @@ function startChild( } lifelines.set(record.id, lifeline); - // Monitor stdout for terminal provider errors (e.g. HTTP 402). - // When stopReason==="error" is emitted, the child Pi process may - // stay alive indefinitely. Kill it so the close handler finalizes - // the record promptly. - let terminalErrorText: string | undefined; - let errorTimer: NodeJS.Timeout | undefined; - child.stdout.on("data", (chunk: Buffer) => { - // Write to file (replaces pipe). - stdoutStream.write(chunk); - if (terminalErrorText) return; - const text = chunk.toString("utf-8"); - if (/"stopReason"\s*:\s*"error"/.test(text)) { - const errMatch = text.match(/"errorMessage"\s*:\s*"([^"]+)"/); - terminalErrorText = errMatch?.[1] ?? "Provider error"; - if (!record.error) { - record.error = terminalErrorText; - } - errorTimer = setTimeout(() => { - errorTimer = undefined; - if (child.exitCode === null && child.signalCode === null) { - child.kill("SIGTERM"); - setTimeout(() => { - if (child.exitCode === null && child.signalCode === null) { - child.kill("SIGKILL"); - } - }, 1000).unref(); - } - }, 1000); - errorTimer.unref(); - } - }); + child.stdout.pipe(stdoutStream); child.stderr.pipe(stderrStream); const done = new Promise((resolve) => { @@ -950,12 +983,6 @@ function startChild( if (finalized) return; finalized = true; - // Clear error-detection timer if still pending. - if (errorTimer) { - clearTimeout(errorTimer); - errorTimer = undefined; - } - await Promise.all([ new Promise((r) => stdoutStream.end(r)), new Promise((r) => stderrStream.end(r)), @@ -966,55 +993,25 @@ function startChild( closeLifeline(record.id); cleanupTempDir(built.tempDir); - const stdout = fs.existsSync(record.stdoutFile) - ? fs.readFileSync(record.stdoutFile, "utf-8") - : ""; - const stderr = fs.existsSync(record.stderrFile) - ? fs.readFileSync(record.stderrFile, "utf-8") - : ""; - record.running = false; record.completedAt = Date.now(); record.updatedAt = Date.now(); - // Only assign a code-based error for actual non-zero exit - // codes (not null signals) and when we don't already have - // a provider error from stdout monitoring. - if (code !== null && code !== 0 && !record.error) + const stderr = readRecoveryFile(record.stderrFile, "child stderr"); + if (code !== null && code !== 0 && !record.error) { record.error = - stderr.trim() || + stderr.text.trim() || + stderr.issue || `Subagent exited with code ${code}${signal ? ` (${signal})` : ""}`; - - // Check if subagent already wrote to result file - const hasExistingResult = - record.outputFile && - fs.existsSync(record.outputFile) && - fs.readFileSync(record.outputFile, "utf-8").trim().length > 0; - - if (!hasExistingResult) { - // Subagent did not write to result file — auto-save final output - const finalOutput = extractFinalOutput(stdout); - // When we have a provider error captured from stdout - // monitoring, use the (error) fallback. - if (record.error && record.outputFile) { - fs.writeFileSync(record.outputFile, "(error)\n", { mode: 0o600 }); - } else if (finalOutput && record.outputFile) { - fs.writeFileSync(record.outputFile, `${finalOutput}\n`, { - mode: 0o600, - }); - } else if (stderr.trim()) { - // Subagent produced only stderr, no stdout output. - if (!record.error) record.error = stderr.trim(); - if (record.outputFile) { - fs.writeFileSync(record.outputFile, "(error)\n", { mode: 0o600 }); - } - } else if (record.outputFile) { - // Edge case: neither stdout nor stderr produced content. - fs.writeFileSync(record.outputFile, "(no output)\n", { mode: 0o600 }); - } } + + ensureResultFile(record, stderr); record.result = record.outputFile; - recordDiscoveredSessionFile(record); + try { + recordDiscoveredSessionFile(record); + } catch { + // Session-file discovery is best-effort after terminal persistence. + } upsertRecord(record); diff --git a/extensions/pi-subagents/src/runs/shared/pi-args.ts b/extensions/pi-subagents/src/runs/shared/pi-args.ts index 98c7cb2..b8d5d64 100644 --- a/extensions/pi-subagents/src/runs/shared/pi-args.ts +++ b/extensions/pi-subagents/src/runs/shared/pi-args.ts @@ -65,9 +65,9 @@ export function buildPiArgs(input: BuildPiArgsInput): BuildPiArgsResult { if (input.model) args.push("--model", input.model); - // Use JSON mode so child output is machine-parseable for error - // detection and result extraction. - args.push("--mode", "json"); + // Child stdout is only a fallback for the final answer. Full JSON event + // streams repeat cumulative message state and can grow to hundreds of MB. + args.push("--print"); // Add only the minimal runtime extension. Do not disable normal extensions, // skills, tools, MCP direct tools, or project context. diff --git a/extensions/pi-subagents/test/unit/minimal-subagents.test.ts b/extensions/pi-subagents/test/unit/minimal-subagents.test.ts index 1b9eac5..31f87c5 100644 --- a/extensions/pi-subagents/test/unit/minimal-subagents.test.ts +++ b/extensions/pi-subagents/test/unit/minimal-subagents.test.ts @@ -478,6 +478,8 @@ test("child pi args do not restrict tools skills extensions or MCP", () => { assert(!built.args.includes("--tools")); assert.equal(built.env.MCP_DIRECT_TOOLS, undefined); assert(built.args.includes("--extension")); + assert(built.args.includes("--print")); + assert.equal(built.args.includes("--mode"), false); }); test("buildPiArgs supports sessionId with sessionDir", () => { @@ -840,6 +842,31 @@ test("auto-saves final assistant message to result file when subagent does not w } }); +test("auto-saves a print-mode answer that is valid JSON", async () => { + const mockPi = createMockPi(); + mockPi.install(); + mockPi.onCall({ output: '{"ok":true}', exitCode: 0 }); + const { sessionId, ctx } = makeTestCtx("pi-subagents-autosave-json-text"); + const { spawnTool } = registerTestTools(() => {}); + try { + const result = await spawnTool.execute( + "autosave-json-child", + { task: "return JSON" }, + new AbortController().signal, + undefined, + ctx, + ); + await waitForPersistedRecord(sessionId, result.details.id); + assert.equal( + fs.readFileSync(result.details.resultPath, "utf-8"), + '{"ok":true}\n', + ); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +}); + // Test 7: Subagent-written result file takes precedence (Requirement 5) test("subagent-written result file content is preserved, not overwritten", async () => { const mockPi = createMockPi(); @@ -1776,6 +1803,28 @@ test("cohort: reconcile preserves cohort metadata", async () => { } }); +test("reconcile never reads an oversized legacy stdout log into a string", async () => { + const { sessionId, ctx } = makeTestCtx("pi-subagents-oversized-stdout"); + const fake = makeFakeCtx(sessionId, ctx.cwd, false); + const dir = path.join(sessionId, "subagents", "oversized-a"); + fs.mkdirSync(dir, { recursive: true }); + const record = { id: "oversized-a", parentSessionId: sessionId, cwd: ctx.cwd, taskPreview: "x", running: true, outputFile: path.join(dir, "result.log"), stdoutFile: path.join(dir, "stdout.log"), stderrFile: path.join(dir, "stderr.log"), createdAt: Date.now(), updatedAt: Date.now() }; + fs.writeFileSync(record.stdoutFile, ""); + fs.truncateSync(record.stdoutFile, 16 * 1024 * 1024 + 1); + fs.writeFileSync(record.stderrFile, ""); + fs.writeFileSync(storeFile(sessionId), JSON.stringify({ records: [record] }, null, 2)); + const { handlers } = registerTestTools(() => {}); + try { + await handlers.get("session_start")(undefined, fake.ctx); + const persisted = readPersistedRecord(sessionId, "oversized-a"); + assert.equal(persisted.running, false); + assert.match(persisted.error, /exceeds the 16777216-byte recovery limit/); + assert.equal(fs.readFileSync(record.outputFile, "utf-8"), "(error)\n"); + } finally { + cleanupTestCtx(ctx, sessionId); + } +}); + // ── Lifeline: process-death cascade via anonymous pipe ── import { PI_SUBAGENT_LIFELINE_FD } from "../../src/runs/shared/subagent-prompt-runtime.ts"; @@ -2225,16 +2274,14 @@ test("model inheritance: explicit model overrides inherited model", async () => // ── Bug: fatal-error lifecycle ── -test("error lifecycle: provider error marks child failed promptly", async () => { +test("error lifecycle: provider runtime failure marks child failed promptly", async () => { const mockPi = createMockPi(); mockPi.install(); - // Simulate an HTTP 402 payment-required error where child stays alive + // The child prompt runtime turns terminal provider failures into stderr and + // a non-zero exit; its prompt lifecycle behavior is tested separately above. mockPi.onCall({ - output: "payment required", - stopReason: "error", - errorMessage: "HTTP 402 Payment Required", - exitCode: 0, - keepAliveAfterFinalMessageMs: 20_000, // would hang for 20s + stderr: "HTTP 402 Payment Required\n", + exitCode: 1, }); const { sessionId, ctx } = makeTestCtx("pi-subagents-error-lifecycle"); @@ -2249,7 +2296,6 @@ test("error lifecycle: provider error marks child failed promptly", async () => ctx, ); - // Wait up to 10s: 1s error-detection grace + SIGTERM + process exit. let record: Record | undefined; for (let i = 0; i < 200; i++) { record = readPersistedRecord(sessionId, result.details.id); @@ -2281,10 +2327,8 @@ test("error lifecycle: no duplicate completion notification on provider error", const mockPi = createMockPi(); mockPi.install(); mockPi.onCall({ - output: "payment required", - stopReason: "error", - errorMessage: "HTTP 402 Payment Required", - exitCode: 0, + stderr: "HTTP 402 Payment Required\n", + exitCode: 1, }); const { sessionId, ctx } = makeTestCtx("pi-subagents-error-dedup"); @@ -2347,11 +2391,8 @@ test("error lifecycle: lifeline cleanup unchanged after provider error", async ( const mockPi = createMockPi(); mockPi.install(); mockPi.onCall({ - output: "payment required", - stopReason: "error", - errorMessage: "HTTP 402 Payment Required", - exitCode: 0, - keepAliveAfterFinalMessageMs: 30_000, + stderr: "HTTP 402 Payment Required\n", + exitCode: 1, }); const { sessionId, ctx } = makeTestCtx("pi-subagents-error-lifeline"); @@ -2369,7 +2410,6 @@ test("error lifecycle: lifeline cleanup unchanged after provider error", async ( ctx, ); - // Wait up to 10s for error detection + kill. let record: Record | undefined; for (let i = 0; i < 200; i++) { record = readPersistedRecord(sessionId, result.details.id);