From d9530cb1d473cf9659e63d562b71b15487163b9e Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Wed, 9 Sep 2026 11:59:22 +0000 Subject: [PATCH 1/5] test: make tmux signal cleanup assertions deterministic --- tests/headless.test.ts | 52 +++++++++++++++++++++++++++++++++++++----- 1 file changed, 46 insertions(+), 6 deletions(-) diff --git a/tests/headless.test.ts b/tests/headless.test.ts index 1e92d40..0019223 100644 --- a/tests/headless.test.ts +++ b/tests/headless.test.ts @@ -5135,13 +5135,17 @@ test("CLI --tmux --delete bounds cleanup of a stuck tmux server", async () => { } }); -test("CLI --tmux --wait --delete kills its named session on SIGTERM", async () => { +async function verifyTmuxSignalCleanup(signalWhen: "launched" | "probe-started"): Promise { const dir = mkdtempSync(join(tmpdir(), "headless-test-")); + let child: ReturnType | undefined; try { const home = join(dir, "home"); const binDir = join(dir, "bin"); const workDir = join(dir, "work"); const captureFile = join(dir, "tmux.jsonl"); + const sessionFile = join(dir, "session"); + const probeReadyFile = join(dir, "probe-ready"); + const probeDrainedFile = join(dir, "probe-drained"); mkdirSync(home); mkdirSync(binDir); mkdirSync(workDir); @@ -5152,14 +5156,25 @@ test("CLI --tmux --wait --delete kills its named session on SIGTERM", async () = "#!/usr/bin/env node", "const fs = require('node:fs');", "const args = process.argv.slice(2);", + "const session = process.env.HEADLESS_TMUX_SESSION_FILE;", + "if (args[0] === 'new-session') fs.writeFileSync(session, args[3]);", + "if (args[0] === 'has-session' && process.env.HEADLESS_TMUX_PROBE_READY) {", + " fs.writeFileSync(process.env.HEADLESS_TMUX_PROBE_READY, 'ready');", + " const deadline = Date.now() + 5000;", + " while (fs.existsSync(session) && Date.now() < deadline) Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 5);", + "}", "fs.appendFileSync(process.env.HEADLESS_TMUX_CAPTURE, JSON.stringify(args) + '\\n');", - "if (args[0] === 'has-session') process.exit(0);", + "if (args[0] === 'kill-session' && fs.readFileSync(session, 'utf8') === args[2]) fs.unlinkSync(session);", + "if (args[0] === 'has-session') {", + " if (process.env.HEADLESS_TMUX_PROBE_DRAINED) fs.writeFileSync(process.env.HEADLESS_TMUX_PROBE_DRAINED, 'drained');", + " process.exit(fs.existsSync(session) ? 0 : 1);", + "}", "", ].join("\n"), ); chmodSync(tmux, 0o755); - const child = spawn( + child = spawn( process.execPath, [ "--import", @@ -5180,6 +5195,11 @@ test("CLI --tmux --wait --delete kills its named session on SIGTERM", async () = env: { ...process.env, HEADLESS_TMUX_CAPTURE: captureFile, + HEADLESS_TMUX_SESSION_FILE: sessionFile, + ...(signalWhen === "probe-started" ? { + HEADLESS_TMUX_PROBE_READY: probeReadyFile, + HEADLESS_TMUX_PROBE_DRAINED: probeDrainedFile, + } : {}), HEADLESS_TMUX_WAIT_INTERVAL_MS: "10", HOME: home, PATH: `${binDir}:${process.env.PATH ?? ""}`, @@ -5187,19 +5207,39 @@ test("CLI --tmux --wait --delete kills its named session on SIGTERM", async () = stdio: "ignore", }, ); - await waitFor(() => existsSync(captureFile) && readFileSync(captureFile, "utf8").includes("new-session")); const completion = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => { - child.on("close", (code, signal) => resolve({ code, signal })); + child!.on("close", (code, signal) => resolve({ code, signal })); }); + await waitFor(() => signalWhen === "probe-started" + ? existsSync(probeReadyFile) + : existsSync(captureFile) && readFileSync(captureFile, "utf8").includes("new-session")); child.kill("SIGTERM"); const result = await completion; assert.equal(result.signal, "SIGTERM"); + if (signalWhen === "probe-started") await waitFor(() => existsSync(probeDrainedFile)); const calls = readFileSync(captureFile, "utf8").trim().split("\n").map((line) => JSON.parse(line)); - assert.deepEqual(calls.at(-1), ["kill-session", "-t", calls[0][3]]); + assert.deepEqual(calls.filter((call) => call[0] === "kill-session"), [["kill-session", "-t", calls[0][3]]]); + assert.equal(existsSync(sessionFile), false); + if (signalWhen === "probe-started") { + assert.deepEqual(calls.at(-1), ["has-session", "-t", calls[0][3]]); + } } finally { + if (child && child.exitCode === null && child.signalCode === null) { + const completion = new Promise((resolve) => child!.once("close", resolve)); + child.kill("SIGKILL"); + await completion; + } rmSync(dir, { force: true, recursive: true }); } +} + +test("CLI --tmux --wait --delete kills its named session on SIGTERM", async () => { + await verifyTmuxSignalCleanup("launched"); +}); + +test("CLI --tmux --wait --delete permits an in-flight probe to finish after SIGTERM cleanup", async () => { + await verifyTmuxSignalCleanup("probe-started"); }); test("CLI --tmux --wait pins the Claude session id instead of injecting a marker", async () => { From d02ef0370e173b5b21718d01daf2ab3cb04dfb6a Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Wed, 9 Sep 2026 12:06:41 +0000 Subject: [PATCH 2/5] test: isolate tmux cleanup from host transcript history --- tests/headless.test.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/headless.test.ts b/tests/headless.test.ts index 0019223..084285f 100644 --- a/tests/headless.test.ts +++ b/tests/headless.test.ts @@ -5093,9 +5093,11 @@ test("CLI --tmux --delete kills a partially launched session", async () => { test("CLI --tmux --delete bounds cleanup of a stuck tmux server", async () => { const dir = mkdtempSync(join(tmpdir(), "headless-test-")); try { + const home = join(dir, "home"); const binDir = join(dir, "bin"); const workDir = join(dir, "work"); const captureFile = join(dir, "tmux.jsonl"); + mkdirSync(home); mkdirSync(binDir); mkdirSync(workDir); const tmux = join(binDir, "tmux"); @@ -5121,6 +5123,8 @@ test("CLI --tmux --delete bounds cleanup of a stuck tmux server", async () => { HEADLESS_TMUX_CAPTURE: captureFile, HEADLESS_TMUX_WAIT_FORCE_MARKER: "1", HEADLESS_TMUX_WAIT_INTERVAL_MS: "10", + HOME: home, + CODEX_HOME: join(home, ".codex"), PATH: `${binDir}:${process.env.PATH ?? ""}`, }, }, From 0c4d5dafa04bc2ed2f4764e876b34410470e69fe Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Wed, 9 Sep 2026 12:07:17 +0000 Subject: [PATCH 3/5] fix: accept successful empty Pi completions --- README.md | 10 ++ src/cli.ts | 55 ++++--- src/pi-completion.ts | 112 ++++++++++++++ src/runs.ts | 4 +- tests/modal.test.ts | 35 +++++ tests/pi-completion-cli.test.ts | 263 ++++++++++++++++++++++++++++++++ tests/pi-completion.test.ts | 215 ++++++++++++++++++++++++++ 7 files changed, 675 insertions(+), 19 deletions(-) create mode 100644 src/pi-completion.ts create mode 100644 tests/pi-completion-cli.test.ts create mode 100644 tests/pi-completion.test.ts diff --git a/README.md b/README.md index 2554d11..2e19cb5 100644 --- a/README.md +++ b/README.md @@ -172,6 +172,16 @@ estimates with reported charges. Subscription cost estimates are API list-price comparisons, not subscription charges. Auth changes affect child environments only; Headless never replaces shared login files. +### Empty Pi completions + +Pi can finish artifact-producing work with an empty final assistant message. +Headless accepts that as success when Pi's native terminal event confirms a normal +completion and the process exits successfully. Plain output contains no invented +answer; `--usage` still reports usage, and SDK results contain `finalMessage: ""`. +An incomplete lifecycle or native error remains a failure, including when earlier +assistant progress text exists. Legacy message-only output remains supported. +Artifact validation remains the caller's responsibility. + ## Native TUI Completion Use `--tmux --wait --delete` when you want Headless to launch the agent in its native TUI, wait for the final native transcript message, print that message, and then terminate the tmux session after the prompt completes. diff --git a/src/cli.ts b/src/cli.ts index e14cbac..84934c3 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -19,6 +19,7 @@ import { randomUUID } from "node:crypto"; import { dirname, join } from "node:path"; import { tmpdir } from "node:os"; import { fileURLToPath } from "node:url"; +import { PiCompletionObserver } from "./pi-completion.js"; import { buildAgentCommand, @@ -4560,6 +4561,7 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise runBilling((attempt) => execute( buildAttemptCommand(attempt.env, attempt.options), attempt.env, attempt.timeoutSeconds ?? modalTimeoutSeconds, - (text) => { attempt.observe(text); if (stdoutHandling === "capture") commandStdoutLog?.(text); }, + (text) => { + piCompletion?.write(text); + attempt.observe(text); + if (stdoutHandling === "capture") commandStdoutLog?.(text); + }, )), }) : await runBilling(async (attempt) => { @@ -4648,7 +4654,11 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise { attempt.observe(text); commandStdoutLog?.(text); }, stderr: commandStderr, + stdoutLog: (text) => { + piCompletion?.write(text); + attempt.observe(text); + commandStdoutLog?.(text); + }, stderr: commandStderr, timeoutSeconds, captureFinalMessageTrace: Boolean(parsed.sdkFormat) || (parsed.agent === "antigravity" && parsed.json && Boolean(parsed.runId)), captureRelevantTrace: Boolean(parsed.sdkFormat) || parsed.usage || (parsed.json && (Boolean(parsed.runId) || Boolean(parsed.sessionAlias))), @@ -4659,6 +4669,7 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise; + +function record(value: unknown): JsonRecord | undefined { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? value as JsonRecord : undefined; +} + +function assistantFailure(message: JsonRecord): string | undefined { + const detail = typeof message.errorMessage === "string" ? message.errorMessage.trim() : ""; + if (detail) return detail.slice(0, MAX_ERROR_CHARS); + if (message.stopReason === "error") return "Pi assistant request failed."; + if (message.stopReason === "aborted") return "Pi assistant request aborted."; + return undefined; +} + +function finalText(message: JsonRecord): string | undefined { + if (message.stopReason !== "stop" || !Array.isArray(message.content)) return undefined; + if (message.errorMessage !== undefined && typeof message.errorMessage !== "string") return undefined; + const text: string[] = []; + for (const value of message.content) { + const block = record(value); + if (block?.type === "text" && typeof block.text === "string") text.push(block.text); + else if (block?.type !== "thinking" || typeof block.thinking !== "string") return undefined; + } + return text.join("").trim(); +} + +/** Native JSONL completion only: never recursively inspect tool results or assistant prose. */ +export class PiCompletionObserver { + outcome: PiCompletionOutcome = { status: "unknown" }; + private pending = ""; + private pendingBytes = 0; + private skipping = false; + private lifecycleSeen = false; + + get observedLifecycle(): boolean { + return this.lifecycleSeen; + } + + write(chunk: string): void { + let start = 0; + while (start < chunk.length) { + const newline = chunk.indexOf("\n", start); + const end = newline < 0 ? chunk.length : newline; + if (!this.skipping) { + const segment = chunk.slice(start, end); + this.pendingBytes += Buffer.byteLength(segment); + if (this.pendingBytes > MAX_LINE_BYTES) { + this.pending = ""; + this.skipping = true; + this.invalidateSuccess(); + } else { + this.pending += segment; + } + } + if (newline < 0) break; + if (!this.skipping) this.consume(this.pending); + this.resetLine(); + start = newline + 1; + } + } + + end(): void { + if (!this.skipping && this.pending) this.consume(this.pending); + this.resetLine(); + } + + private resetLine(): void { + this.pending = ""; + this.pendingBytes = 0; + this.skipping = false; + } + + private invalidateSuccess(): void { + if (this.outcome.status === "success") this.outcome = { status: "unknown" }; + } + + private consume(line: string): void { + if (!line.trim()) return; + let event: JsonRecord | undefined; + try { event = record(JSON.parse(line)); } catch { /* Incomplete native evidence cannot preserve success. */ } + if (!event || typeof event.type !== "string") { + this.invalidateSuccess(); + return; + } + if (["agent_start", "agent_end", "agent_settled"].includes(event.type)) this.lifecycleSeen = true; + if (event.type === "agent_settled") return; + this.invalidateSuccess(); + let message: JsonRecord | undefined; + if (event.type === "agent_end" && Array.isArray(event.messages)) { + message = record(event.messages.at(-1)); + } else if (event.type === "message_end" || event.type === "turn_end") { + message = record(event.message); + } + if (message?.role !== "assistant") return; + const error = assistantFailure(message); + if (error) { + this.outcome = { status: "error", error }; + return; + } + if (event.type !== "agent_end" || (event.willRetry !== undefined && event.willRetry !== false)) return; + const finalMessage = finalText(message); + if (finalMessage !== undefined) this.outcome = { status: "success", finalMessage }; + } +} diff --git a/src/runs.ts b/src/runs.ts index c76e430..55d3fe5 100644 --- a/src/runs.ts +++ b/src/runs.ts @@ -291,7 +291,7 @@ export function updateNodeStatus( const now = new Date().toISOString(); node.status = status; node.updatedAt = now; - if (message) { + if (message !== undefined) { node.lastMessage = message; } if (metrics && Object.keys(metrics).length > 0) { @@ -322,7 +322,7 @@ export function completeIdleRunNodes(env: Env, runId: string, orchestratorNodeId } node.status = "done"; node.updatedAt = now; - if (node.nodeId === orchestratorNodeId && orchestratorMessage) { + if (node.nodeId === orchestratorNodeId && orchestratorMessage !== undefined) { node.lastMessage = orchestratorMessage; } addEvent(run, { diff --git a/tests/modal.test.ts b/tests/modal.test.ts index ed09d5e..8a10dbb 100644 --- a/tests/modal.test.ts +++ b/tests/modal.test.ts @@ -26,6 +26,7 @@ import { type ModalWriteStreamLike, } from "../src/modal.ts"; import { quoteCommand } from "../src/shell.ts"; +import { PiCompletionObserver } from "../src/pi-completion.ts"; test("default Modal image is immutable", () => { assert.equal( @@ -955,3 +956,37 @@ test("Modal billing retries share sandbox, observe captured output and mask inhe test("explicit undefined command credentials cannot be reintroduced by forwarding", () => { assert.equal(collectModalEnv({ OPENAI_API_KEY: "parent" }, { OPENAI_API_KEY: undefined }, ["OPENAI_API_KEY=explicit"]).OPENAI_API_KEY, undefined); }); + +test("Modal observes empty Pi completion before captured stdout is truncated", async () => { + const dir = mkdtempSync(join(tmpdir(), "headless-modal-pi-completion-")); + try { + const work = join(dir, "work"), remote = join(dir, "remote"); + mkdirSync(work); + mkdirSync(remote); + initGitWorkdir(work); + const trace = [ + { type: "agent_start" }, + { type: "agent_end", messages: [ + { role: "assistant", stopReason: "stop", content: [{ type: "text", text: "" }] }, + ] }, + { type: "agent_settled" }, + ].map((record) => `${JSON.stringify(record)}\n`); + const sandbox = new FakeSandbox(remote, { agentStdoutChunks: trace }); + const observer = new PiCompletionObserver(); + const result = await executeModalAgent({ + agent: "pi", appName: "test", command: { command: "pi", args: ["--mode", "json"] }, + cpu: 1, env: { HOME: join(dir, "home") }, image: DEFAULT_MODAL_IMAGE, + includeGit: false, memoryMiB: 1024, modalEnv: [], modalSecrets: [], + stdout: () => assert.fail("capture mode must not stream"), stderr: () => {}, + stdoutHandling: "capture", maxCapturedStdoutBytes: 30, timeoutSeconds: 60, + workDir: work, clientFactory: async () => new FakeModalClient(sandbox), + invoke: (execute) => execute({ command: "pi", args: ["--mode", "json"] }, {}, 50, + (text) => observer.write(text)), + }); + observer.end(); + assert.equal(result.code, 0); + assert.doesNotMatch(result.stdout, /agent_end/); + assert.deepEqual(observer.outcome, { status: "success", finalMessage: "" }); + assert.equal(sandbox.terminated, true); + } finally { rmSync(dir, { recursive: true, force: true }); } +}); diff --git a/tests/pi-completion-cli.test.ts b/tests/pi-completion-cli.test.ts new file mode 100644 index 0000000..38917ac --- /dev/null +++ b/tests/pi-completion-cli.test.ts @@ -0,0 +1,263 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { runCli } from "../src/cli.ts"; +import { readStoredSession } from "../src/sessions.ts"; +import { readRun, registerNode, updateNodeStatus } from "../src/runs.ts"; + +type Json = Record; + +function assistant(stopReason: string, text: string, input: number, cached: number): Json { + return { + role: "assistant", provider: "openai-codex", model: "gpt-5.6-sol", stopReason, + content: [{ type: "text", text }], + usage: { input, output: stopReason === "toolUse" ? 2 : 0, cacheRead: cached, + cacheWrite: 0, cost: { input: 0.04, output: 0.01, cacheRead: 0.005, cacheWrite: 0, total: 0.055 } }, + }; +} + +// Sanitized shape of run 79706: artifact work, empty native stop, then agent_settled. +function completedTrace(final = assistant("stop", "", 4, 6)): Json[] { + const tool = assistant("toolUse", "", 10, 3); + tool.content = [{ type: "toolCall", id: "write-design", name: "bash", arguments: { command: "write design" } }]; + return [ + { type: "agent_start" }, + { type: "message_end", message: tool }, + { type: "tool_execution_start", toolCallId: "write-design", toolName: "bash" }, + { type: "tool_execution_end", toolCallId: "write-design", toolName: "bash", isError: false, + result: { content: [{ type: "text", text: "shape checks ok" }] } }, + { type: "message_update", assistantMessageEvent: { type: "text_end", contentIndex: 0, content: "" } }, + { type: "message_end", message: final }, + { type: "turn_end", message: final }, + { type: "agent_end", messages: [tool, final] }, + { type: "agent_settled" }, + ]; +} + +function fixture(records = completedTrace(), exitCode = 0, rawTail = "") { + const home = mkdtempSync(join(tmpdir(), "pi-completion-cli-")); + const nativeSession = join(home, "native-session.jsonl"); + const executable = `#!${process.execPath}\n` + + `const fs = require('node:fs');\n` + + `fs.appendFileSync(${JSON.stringify(join(home, "calls"))}, JSON.stringify(process.argv.slice(2)) + '\\n');\n` + + `const records = [{type:'session',sessionId:${JSON.stringify(nativeSession)}}, ...${JSON.stringify(records)}];\n` + + `fs.writeFileSync(${JSON.stringify(nativeSession)}, records.map(JSON.stringify).join('\\n')+'\\n');\n` + + `for (const record of records) console.log(JSON.stringify(record));\n` + + `process.stdout.write(${JSON.stringify(rawTail)});\nprocess.exitCode=${exitCode};\n`; + for (const binary of ["pi", "docker"]) writeFileSync(join(home, binary), executable, { mode: 0o755 }); + const env = { HOME: home, PATH: `${home}:${process.env.PATH}`, HEADLESS_MODELS_DEV_CACHE: "" }; + return { + home, nativeSession, env, + async run(flags: string[]) { + const output: string[] = [], errors: string[] = []; + const code = await runCli(["pi", "--model", "openai-codex/gpt-5.6-sol", "--prompt", "write design", ...flags], { + env, stdout: (text) => output.push(text), stderr: (text) => errors.push(text), + }); + return { code, stdout: output.join(""), stderr: errors.join("") }; + }, + cleanup() { rmSync(home, { recursive: true, force: true }); }, + }; +} + +function jsonRows(output: string): Json[] { + return output.split("\n").flatMap((line) => { + try { return [JSON.parse(line) as Json]; } catch { return []; } + }); +} + +function assertUsage(usage: Json): void { + assert.equal(usage.inputTokens, 14); + assert.equal(usage.cacheReadTokens, 9); + assert.equal(usage.outputTokens, 2); + assert.equal(usage.totalTokens, 25); + assert.ok(Math.abs(usage.cost.total - 0.11) < 1e-12); +} + +test("Pi empty successful native completion succeeds without invented plain prose", async () => { + const f = fixture(); + try { + const result = await f.run([]); + assert.equal(result.code, 0, result.stderr); + assert.equal(result.stdout.trim(), ""); + assert.doesNotMatch(result.stderr, /could not extract final message/); + } finally { f.cleanup(); } +}); + +for (const flags of [["--usage"], ["--debug", "--usage"], ["--json", "--usage"]]) { + test(`Pi empty success retains usage exactly once with ${flags.join(" ")}`, async () => { + const f = fixture(); + try { + const result = await f.run(flags); + assert.equal(result.code, 0, result.stderr); + const reports = jsonRows(result.stdout).filter((row) => Object.keys(row).length === 1 && row.usage); + assert.equal(reports.length, 1); + assertUsage(reports[0].usage); + if (flags.includes("--json") || flags.includes("--debug")) { + assert.ok(jsonRows(result.stdout).some((row) => row.type === "agent_settled")); + } + } finally { f.cleanup(); } + }); +} + +for (const format of ["json", "ndjson"]) { + test(`Pi SDK ${format} accepts empty final completion with exact usage`, async () => { + const f = fixture(); + try { + const result = await f.run(["--sdk-format", format, "--usage"]); + assert.equal(result.code, 0, result.stderr); + const rows = jsonRows(result.stdout); + const terminal = rows.at(-1)!; + assert.equal(terminal.type, "result"); + assert.equal(terminal.data.finalMessage, ""); + assertUsage(terminal.data.usage); + assert.equal(rows.filter((row) => row.type === "result").length, 1); + } finally { f.cleanup(); } + }); +} + +for (const flags of [[], ["--json"], ["--sdk-format", "json"]]) { + test(`Pi empty completion preserves native nonzero exit in ${flags.join(" ") || "plain"}`, async () => { + const f = fixture(completedTrace(), 7); + try { assert.equal((await f.run(flags)).code, 7); } finally { f.cleanup(); } + }); +} + +for (const flags of [["--usage"], ["--json", "--usage"], ["--sdk-format", "json"]]) { + test(`Pi native error after earlier prose cannot become success in ${flags.join(" ")}`, async () => { + const previous = assistant("stop", "earlier progress", 10, 3); + const failure = { ...assistant("error", "", 4, 6), errorMessage: "provider transport failed" }; + const records = [{ type: "message_end", message: previous }, + { type: "message_end", message: failure }, { type: "agent_end", messages: [previous, failure] }, + { type: "agent_settled" }]; + const f = fixture(records); + try { + const result = await f.run(flags); + assert.equal(result.code, 1, result.stdout); + if (!flags.includes("--sdk-format")) { + const reports = jsonRows(result.stdout).filter((row) => Object.keys(row).length === 1 && row.usage); + assert.equal(reports.length, 1); + assert.equal(reports[0].usage.inputTokens, 14); + } + } finally { f.cleanup(); } + }); +} + +test("Pi absent terminal empty trace still fails plain completion", async () => { + const f = fixture([{ type: "agent_start" }]); + try { assert.equal((await f.run([])).code, 1); } finally { f.cleanup(); } +}); + +function incompleteTraces(): { name: string; records: Json[]; rawTail: string }[] { + const records = [ + { type: "message_end", message: assistant("stop", "earlier progress", 1, 0) }, + ...completedTrace(), + ]; + return [ + { name: "new unfinished agent turn", records: [...records, { type: "agent_start" }], rawTail: "" }, + { name: "truncated native tail", records, rawTail: '{"type":"message_end","message":' }, + ]; +} + +for (const trace of incompleteTraces()) { + for (const flags of [[], ["--json"], ["--sdk-format", "json"], ["--sdk-format", "ndjson"]]) { + test(`Pi ${trace.name} cannot reuse earlier prose in ${flags.join(" ") || "plain"}`, async () => { + const f = fixture(trace.records, 0, trace.rawTail); + try { + const result = await f.run(flags); + assert.equal(result.code, 1, result.stdout); + if (flags.includes("--sdk-format")) { + assert.equal(jsonRows(result.stdout).filter((row) => row.type === "result").length, 0); + } + } finally { f.cleanup(); } + }); + } + + test(`Pi ${trace.name} does not persist a named session`, async () => { + const f = fixture(trace.records, 0, trace.rawTail); + try { + const result = await f.run(["--session", "incomplete-design"]); + assert.equal(result.code, 1, result.stdout); + assert.equal(readStoredSession(f.env, "pi", "incomplete-design"), undefined); + } finally { f.cleanup(); } + }); +} + +test("Pi native failure does not persist a named session", async () => { + const failure = { ...assistant("error", "", 4, 6), errorMessage: "provider transport failed" }; + const f = fixture(completedTrace(failure)); + try { + const result = await f.run(["--session", "failed-design"]); + assert.equal(result.code, 1, result.stdout); + assert.equal(readStoredSession(f.env, "pi", "failed-design"), undefined); + } finally { f.cleanup(); } +}); + +for (const scenario of [ + { name: "empty success", records: completedTrace(), exitCode: 0, expectedCode: 0, rawTail: "" }, + { name: "native nonzero", records: completedTrace(), exitCode: 7, expectedCode: 7, rawTail: "" }, + { name: "native error", records: completedTrace(assistant("error", "", 4, 6)), + exitCode: 0, expectedCode: 1, rawTail: "" }, + ...incompleteTraces().map((trace) => ({ ...trace, exitCode: 0, expectedCode: 1 })), +]) { + test(`Pi orchestrator ${scenario.name} records the correct run completion`, async () => { + const f = fixture(scenario.records, scenario.exitCode, scenario.rawTail); + try { + for (const status of ["idle", "planned"] as const) { + registerNode(f.env, { runId: "design-run", nodeId: `${status}-worker`, role: "worker", + agent: "pi", coordination: "oneshot", status }); + } + const result = await f.run(["--run", "design-run", "--role", "orchestrator", + "--coordination", "oneshot"]); + assert.equal(result.code, scenario.expectedCode, result.stderr); + const run = readRun(f.env, "design-run")!; + assert.equal(run.nodes.orchestrator.status, scenario.expectedCode === 0 ? "done" : "failed"); + assert.equal(run.nodes["idle-worker"].status, scenario.expectedCode === 0 ? "done" : "idle"); + assert.equal(run.nodes["planned-worker"].status, "planned"); + if (scenario.expectedCode === 0) assert.equal(run.nodes.orchestrator.lastMessage, ""); + } finally { f.cleanup(); } + }); +} + +for (const role of ["orchestrator", "worker"] as const) { + test(`Pi reused ${role} clears prior prose after confirmed empty success`, async () => { + const f = fixture(); + try { + registerNode(f.env, { runId: "reused-run", nodeId: role, role, agent: "pi", + coordination: "oneshot", status: "idle" }); + updateNodeStatus(f.env, "reused-run", role, "idle", "previous invocation answer"); + const result = await f.run(["--run", "reused-run", "--role", role, "--node", role, + "--coordination", "oneshot"]); + assert.equal(result.code, 0, result.stderr); + const node = readRun(f.env, "reused-run")!.nodes[role]; + assert.equal(node.lastMessage, ""); + assert.equal(node.status, role === "orchestrator" ? "done" : "idle"); + } finally { f.cleanup(); } + }); +} + +test("Pi empty completion works through Docker transport", async () => { + const f = fixture(); + try { + const result = await f.run(["--docker", "--usage"]); + assert.equal(result.code, 0, result.stderr); + const reports = jsonRows(result.stdout).filter((row) => Object.keys(row).length === 1 && row.usage); + assert.equal(reports.length, 1); + assertUsage(reports[0].usage); + } finally { f.cleanup(); } +}); + +test("Pi empty completion persists its native session and resumes that session", async () => { + const f = fixture(); + try { + const first = await f.run(["--session", "design"]); + assert.equal(first.code, 0, first.stderr); + assert.equal(readStoredSession(f.env, "pi", "design")?.nativeId, f.nativeSession); + const second = await f.run(["--session", "design"]); + assert.equal(second.code, 0, second.stderr); + const calls = readFileSync(join(f.home, "calls"), "utf8").trim().split("\n").map((line) => JSON.parse(line)); + assert.equal(calls.length, 2); + assert.equal(calls[1][calls[1].indexOf("--session") + 1], f.nativeSession); + } finally { f.cleanup(); } +}); diff --git a/tests/pi-completion.test.ts b/tests/pi-completion.test.ts new file mode 100644 index 0000000..2950aa2 --- /dev/null +++ b/tests/pi-completion.test.ts @@ -0,0 +1,215 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { PiCompletionObserver } from "../src/pi-completion.ts"; + +const assistant = (text = "", extra = {}) => ({ + role: "assistant", content: [{ type: "text", text }], stopReason: "stop", ...extra, +}); +const terminal = (message = assistant(), extra = {}) => ({ + type: "agent_end", messages: [message], ...extra, +}); +const line = (event: unknown) => `${JSON.stringify(event)}\n`; +function observe(...events: unknown[]): PiCompletionObserver { + const observer = new PiCompletionObserver(); + for (const event of events) observer.write(line(event)); + observer.end(); + return observer; +} + +test("accepts empty terminal text without reusing earlier assistant prose", () => { + const observer = observe( + { type: "message_end", message: assistant("Working on it", { stopReason: "toolUse" }) }, + terminal(), { type: "agent_settled" }, + ); + assert.deepEqual(observer.outcome, { status: "success", finalMessage: "" }); +}); + +test("joins terminal text blocks and excludes reasoning", () => { + const message = assistant("", { content: [ + { type: "thinking", thinking: "private", text: "not visible" }, + { type: "text", text: "Final " }, { type: "text", text: "answer." }, + ] }); + assert.deepEqual(observe(terminal(message)).outcome, { + status: "success", finalMessage: "Final answer.", + }); +}); + +test("handles every split position and a complete final line without newline", () => { + const trace = line({ type: "agent_start" }) + JSON.stringify(terminal(assistant("answer 🌱"))); + for (let split = 0; split <= trace.length; split++) { + const observer = new PiCompletionObserver(); + observer.write(trace.slice(0, split)); + observer.write(trace.slice(split)); + observer.end(); + assert.deepEqual(observer.outcome, { status: "success", finalMessage: "answer 🌱" }); + } +}); + +test("message_end, turn_end, and agent_settled cannot establish successful completion", () => { + for (const event of [ + { type: "message_end", message: assistant() }, + { type: "turn_end", message: assistant() }, { type: "agent_settled" }, + ]) assert.deepEqual(observe(event).outcome, { status: "unknown" }); +}); + +test("requires last agent_end item to be a successful assistant without tools", () => { + for (const event of [ + terminal(assistant(), { messages: [] }), + terminal(assistant(), { messages: [assistant(), { role: "toolResult", content: [] }] }), + ...[undefined, "", "length", "toolUse", "pending", "completed"].map((stopReason) => + terminal(assistant("", { stopReason }))), + terminal(assistant("", { content: [{ type: "toolCall", name: "bash" }] })), + terminal(assistant("", { content: "" })), + terminal(assistant("", { content: [{ type: "text" }] })), + terminal(assistant("", { content: [{ type: "thinking" }] })), + terminal(assistant("", { errorMessage: { message: "invalid native detail" } })), + terminal(assistant(), { willRetry: true }), + ]) assert.deepEqual(observe(event).outcome, { status: "unknown" }); +}); + +test("allows empty content and reasoning-only successful terminal messages", () => { + for (const content of [[], [{ type: "thinking", thinking: "reasoning" }]]) { + assert.deepEqual(observe(terminal(assistant("", { content }))).outcome, + { status: "success", finalMessage: "" }); + } +}); + +test("native assistant failures override prior prose and successful completion", () => { + for (const type of ["message_end", "turn_end", "agent_end"]) { + for (const stopReason of ["error", "aborted"]) { + const message = assistant("stale text", { stopReason, errorMessage: "Provider rejected request" }); + const event = type === "agent_end" ? terminal(message) : { type, message }; + assert.deepEqual(observe(terminal(assistant("earlier")), event).outcome, + { status: "error", error: "Provider rejected request" }); + } + } +}); + +test("treats nonempty native errorMessage as failure even with stop reason stop", () => { + assert.deepEqual(observe(terminal(assistant("", { errorMessage: "Rejected" }))).outcome, + { status: "error", error: "Rejected" }); +}); + +test("uses bounded meaningful failure messages", () => { + for (const stopReason of ["error", "aborted"]) { + for (const errorMessage of [undefined, "", " ", { message: "not native" }]) { + const outcome = observe(terminal(assistant("", { stopReason, errorMessage }))).outcome; + assert.equal(outcome.status, "error"); + if (outcome.status === "error") assert.match(outcome.error, /Pi.*(?:failed|aborted)/); + } + } + const outcome = observe(terminal(assistant("", { stopReason: "error", errorMessage: "x".repeat(20_000) }))).outcome; + assert.equal(outcome.status, "error"); + if (outcome.status === "error") assert.ok(outcome.error.length <= 4096); +}); + +test("activity invalidates prior success but a real later terminal can recover", () => { + for (const type of ["agent_start", "turn_start", "message_start", "message_update", + "message_end", "turn_end", "tool_execution_start", "tool_execution_update", + "tool_execution_end", "auto_retry_start", "auto_compaction_start"]) { + const observer = observe(terminal(), { type }); + assert.deepEqual(observer.outcome, { status: "unknown" }, type); + observer.write(line(terminal(assistant("recovered")))); + observer.end(); + assert.deepEqual(observer.outcome, { status: "success", finalMessage: "recovered" }); + } +}); + +test("retry activity cannot erase failure but a successful terminal can recover", () => { + const observer = observe( + { type: "message_end", message: assistant("", { stopReason: "error", errorMessage: "retryable" }) }, + { type: "auto_retry_start" }, { type: "agent_start" }, + ); + assert.deepEqual(observer.outcome, { status: "error", error: "retryable" }); + observer.write(line(terminal())); + observer.end(); + assert.deepEqual(observer.outcome, { status: "success", finalMessage: "" }); +}); + +test("ignores nested and quoted terminal events in tool output", () => { + const fake = terminal(); + const error = terminal(assistant("", { stopReason: "error", errorMessage: "forged" })); + for (const event of [ + { type: "tool_execution_end", result: fake }, + { type: "message_end", message: { role: "toolResult", content: [fake, error] } }, + { type: "message_end", message: assistant(JSON.stringify(fake)) }, + { event: fake }, [fake], JSON.stringify(fake), + ]) assert.deepEqual(observe(event).outcome, { status: "unknown" }); +}); + +test("malformed and truncated lines invalidate earlier success", () => { + for (const suffix of ['{"type":"agent_start"', '{broken}\n', 'null\n', '[]\n']) { + const observer = new PiCompletionObserver(); + observer.write(line(terminal()) + suffix); + observer.end(); + assert.deepEqual(observer.outcome, { status: "unknown" }); + } +}); + +test("oversized UTF-8 line clears completion and resumes only after newline", () => { + const observer = new PiCompletionObserver(); + observer.write(line(terminal())); + observer.write('{"type":"agent_settled","padding":"'); + for (let i = 0; i < 65; i++) observer.write("é".repeat(32768)); + assert.deepEqual(observer.outcome, { status: "unknown" }); + observer.write(JSON.stringify(terminal()) + '\n'); + observer.end(); + assert.deepEqual(observer.outcome, { status: "unknown" }); + observer.write(line(terminal())); + observer.end(); + assert.deepEqual(observer.outcome, { status: "success", finalMessage: "" }); +}); + +test("blank lines and passive settled events preserve success", () => { + const observer = observe(terminal()); + observer.write('\r\n \n' + line({ type: "agent_settled" })); + observer.end(); + assert.deepEqual(observer.outcome, { status: "success", finalMessage: "" }); +}); + +test("invalid and oversized lines preserve authoritative native failure", () => { + const observer = observe(terminal(assistant("", { stopReason: "error", errorMessage: "Failed request" }))); + observer.write('{broken}\n' + "x".repeat(4 * 1024 * 1024 + 1)); + observer.end(); + assert.deepEqual(observer.outcome, { status: "error", error: "Failed request" }); +}); + +test("accepts a terminal line at the byte limit but rejects the next byte", () => { + const empty = JSON.stringify(terminal(assistant())); + const padding = 4 * 1024 * 1024 - Buffer.byteLength(empty); + for (const extra of [0, 1]) { + const observer = new PiCompletionObserver(); + observer.write(JSON.stringify(terminal(assistant("x".repeat(padding + extra))))); + observer.end(); + assert.equal(observer.outcome.status, extra ? "unknown" : "success"); + } +}); + +test("distinguishes legacy message-only traces from an incomplete native lifecycle", () => { + const legacy = observe({ type: "message_end", message: assistant("legacy answer") }); + assert.equal(legacy.observedLifecycle, false); + for (const event of [ + { type: "agent_start" }, { type: "agent_settled" }, + ]) { + const observer = observe(event); + assert.equal(observer.observedLifecycle, true); + assert.deepEqual(observer.outcome, { status: "unknown" }); + } +}); + +test("retains lifecycle evidence after later activity or malformed input invalidates success", () => { + for (const suffix of [line({ type: "agent_start" }), '{broken}\n']) { + const observer = observe(terminal(assistant("previous answer"))); + assert.equal(observer.observedLifecycle, true); + observer.write(suffix); + observer.end(); + assert.deepEqual(observer.outcome, { status: "unknown" }); + assert.equal(observer.observedLifecycle, true); + } +}); + +test("nested lifecycle envelopes do not disable legacy trace compatibility", () => { + const observer = observe({ type: "tool_execution_end", result: terminal() }); + assert.equal(observer.observedLifecycle, false); +}); From 8c90750a337f15aed10edd410a6b4f92a58b06dd Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Wed, 9 Sep 2026 12:15:13 +0000 Subject: [PATCH 4/5] fix: preserve Pi completion after threshold compaction --- src/pi-completion.ts | 20 ++++++++++++++++++++ tests/pi-completion-cli.test.ts | 19 +++++++++++++++++++ tests/pi-completion.test.ts | 31 +++++++++++++++++++++++++++++++ 3 files changed, 70 insertions(+) diff --git a/src/pi-completion.ts b/src/pi-completion.ts index 3e7e44e..f7b5f56 100644 --- a/src/pi-completion.ts +++ b/src/pi-completion.ts @@ -39,6 +39,7 @@ export class PiCompletionObserver { private pendingBytes = 0; private skipping = false; private lifecycleSeen = false; + private compactionCompletion?: Extract; get observedLifecycle(): boolean { return this.lifecycleSeen; @@ -79,9 +80,27 @@ export class PiCompletionObserver { } private invalidateSuccess(): void { + this.compactionCompletion = undefined; if (this.outcome.status === "success") this.outcome = { status: "unknown" }; } + private observeCompaction(event: JsonRecord): boolean { + if (event.type === "compaction_start" && event.reason === "threshold" && this.outcome.status === "success") { + const completion = this.outcome; + this.invalidateSuccess(); + this.compactionCompletion = completion; + return true; + } + if (event.type !== "compaction_end" || !this.compactionCompletion) return false; + const completion = this.compactionCompletion; + this.invalidateSuccess(); + if (event.reason === "threshold" && event.aborted === false && event.willRetry === false + && record(event.result) && event.errorMessage === undefined) { + this.outcome = completion; + } + return true; + } + private consume(line: string): void { if (!line.trim()) return; let event: JsonRecord | undefined; @@ -92,6 +111,7 @@ export class PiCompletionObserver { } if (["agent_start", "agent_end", "agent_settled"].includes(event.type)) this.lifecycleSeen = true; if (event.type === "agent_settled") return; + if (this.observeCompaction(event)) return; this.invalidateSuccess(); let message: JsonRecord | undefined; if (event.type === "agent_end" && Array.isArray(event.messages)) { diff --git a/tests/pi-completion-cli.test.ts b/tests/pi-completion-cli.test.ts index 38917ac..034a2f2 100644 --- a/tests/pi-completion-cli.test.ts +++ b/tests/pi-completion-cli.test.ts @@ -261,3 +261,22 @@ test("Pi empty completion persists its native session and resumes that session", assert.equal(calls[1][calls[1].indexOf("--session") + 1], f.nativeSession); } finally { f.cleanup(); } }); + +for (const text of ["", "done"]) { + for (const flags of [[], ["--sdk-format", "json"]]) { + test(`Pi threshold compaction preserves ${JSON.stringify(text)} in ${flags.join(" ") || "plain"}`, async () => { + const records = completedTrace(assistant("stop", text, 4, 6)); + records.splice(-1, 0, + { type: "compaction_start", reason: "threshold" }, + { type: "compaction_end", reason: "threshold", aborted: false, willRetry: false, + result: { summary: "Compacted history", firstKeptEntryId: "entry", tokensBefore: 100000 } }); + const f = fixture(records); + try { + const result = await f.run([...flags, "--session", "compacted"]); + assert.equal(result.code, 0, result.stderr || result.stdout); + assert.equal(flags.length ? jsonRows(result.stdout).at(-1)!.data.finalMessage : result.stdout.trim(), text); + assert.equal(readStoredSession(f.env, "pi", "compacted")?.nativeId, f.nativeSession); + } finally { f.cleanup(); } + }); + } +} diff --git a/tests/pi-completion.test.ts b/tests/pi-completion.test.ts index 2950aa2..16e79d4 100644 --- a/tests/pi-completion.test.ts +++ b/tests/pi-completion.test.ts @@ -213,3 +213,34 @@ test("nested lifecycle envelopes do not disable legacy trace compatibility", () const observer = observe({ type: "tool_execution_end", result: terminal() }); assert.equal(observer.observedLifecycle, false); }); + +const compactionStart = { type: "compaction_start", reason: "threshold" }; +const compactionEnd = { + type: "compaction_end", reason: "threshold", aborted: false, willRetry: false, + result: { summary: "Compacted history", firstKeptEntryId: "entry", tokensBefore: 100000 }, +}; + +test("restores the terminal answer after successful threshold compaction", () => { + for (const text of ["", "done"]) { + const observer = observe(terminal(assistant(text)), compactionStart); + assert.deepEqual(observer.outcome, { status: "unknown" }); + observer.write(line(compactionEnd) + line({ type: "agent_settled" })); + observer.end(); + assert.deepEqual(observer.outcome, { status: "success", finalMessage: text }); + } +}); + +test("compaction cannot revive an absent, failed, or interrupted completion", () => { + const failure = terminal(assistant("", { stopReason: "error" })); + const traces = [ + [compactionStart, compactionEnd], [terminal(), compactionEnd], + [failure, compactionStart, compactionEnd], + [terminal(), compactionStart], + ...[{ aborted: true }, { willRetry: true }, { reason: "overflow" }, { result: undefined }, + { errorMessage: "compaction failed" }].map((extra) => + [terminal(), compactionStart, { ...compactionEnd, ...extra }]), + ...[{ type: "agent_start" }, { type: "auto_retry_start" }, null].map((event) => + [terminal(), compactionStart, event, compactionEnd]), + ]; + for (const trace of traces) assert.notEqual(observe(...trace).outcome.status, "success"); +}); From 7b006e9504a900998d7ee9980469138f34df240c Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Wed, 9 Sep 2026 12:20:33 +0000 Subject: [PATCH 5/5] fix: stream oversized Pi terminal history --- src/pi-completion.ts | 35 ++-- src/pi-terminal-json.ts | 277 ++++++++++++++++++++++++++++++++ tests/pi-completion-cli.test.ts | 20 +++ tests/pi-completion.test.ts | 31 ++++ tests/pi-terminal-json.test.ts | 92 +++++++++++ 5 files changed, 444 insertions(+), 11 deletions(-) create mode 100644 src/pi-terminal-json.ts create mode 100644 tests/pi-terminal-json.test.ts diff --git a/src/pi-completion.ts b/src/pi-completion.ts index f7b5f56..84a5d22 100644 --- a/src/pi-completion.ts +++ b/src/pi-completion.ts @@ -1,3 +1,5 @@ +import { PiTerminalJson } from "./pi-terminal-json.js"; + export type PiCompletionOutcome = | { status: "unknown" } | { status: "success"; finalMessage: string } @@ -37,7 +39,7 @@ export class PiCompletionObserver { outcome: PiCompletionOutcome = { status: "unknown" }; private pending = ""; private pendingBytes = 0; - private skipping = false; + private oversizedRecord?: PiTerminalJson; private lifecycleSeen = false; private compactionCompletion?: Extract; @@ -50,33 +52,47 @@ export class PiCompletionObserver { while (start < chunk.length) { const newline = chunk.indexOf("\n", start); const end = newline < 0 ? chunk.length : newline; - if (!this.skipping) { - const segment = chunk.slice(start, end); + const segment = chunk.slice(start, end); + if (!this.oversizedRecord) { this.pendingBytes += Buffer.byteLength(segment); if (this.pendingBytes > MAX_LINE_BYTES) { + this.oversizedRecord = new PiTerminalJson(); + this.oversizedRecord.write(this.pending); this.pending = ""; - this.skipping = true; this.invalidateSuccess(); } else { this.pending += segment; } } + this.oversizedRecord?.write(segment); if (newline < 0) break; - if (!this.skipping) this.consume(this.pending); + this.finishLine(); this.resetLine(); start = newline + 1; } } end(): void { - if (!this.skipping && this.pending) this.consume(this.pending); + this.finishLine(); this.resetLine(); } private resetLine(): void { this.pending = ""; this.pendingBytes = 0; - this.skipping = false; + this.oversizedRecord = undefined; + } + + private finishLine(): void { + if (!this.oversizedRecord && !this.pending.trim()) return; + let event: JsonRecord | undefined; + if (this.oversizedRecord) { + event = this.oversizedRecord.end(); + if (event?.type !== "agent_end") event = undefined; + } else { + try { event = record(JSON.parse(this.pending)); } catch { /* Incomplete native evidence cannot preserve success. */ } + } + this.consume(event); } private invalidateSuccess(): void { @@ -101,10 +117,7 @@ export class PiCompletionObserver { return true; } - private consume(line: string): void { - if (!line.trim()) return; - let event: JsonRecord | undefined; - try { event = record(JSON.parse(line)); } catch { /* Incomplete native evidence cannot preserve success. */ } + private consume(event: JsonRecord | undefined): void { if (!event || typeof event.type !== "string") { this.invalidateSuccess(); return; diff --git a/src/pi-terminal-json.ts b/src/pi-terminal-json.ts new file mode 100644 index 0000000..53e6558 --- /dev/null +++ b/src/pi-terminal-json.ts @@ -0,0 +1,277 @@ +const MAX_BYTES = 4 * 1024 * 1024; +const MAX_DEPTH = 256; + +class BoundedText { + private parts: string[] = []; + private part: string[] = []; + private bytes = 0; + private previousHigh = false; + + append(character: string): void { + if (this.bytes > MAX_BYTES) return; + const code = character.charCodeAt(0); + this.bytes += code < 0x80 ? 1 : code < 0x800 ? 2 + : code >= 0xdc00 && code <= 0xdfff && this.previousHigh ? 1 : 3; + this.previousHigh = code >= 0xd800 && code <= 0xdbff; + if (this.bytes > MAX_BYTES) { + this.parts = []; + this.part = []; + return; + } + this.part.push(character); + if (this.part.length === 8192) { + this.parts.push(this.part.join("")); + this.part = []; + } + } + + value(): string | undefined { + return this.bytes > MAX_BYTES ? undefined : this.parts.join("") + this.part.join(""); + } +} + +type Frame = { + kind: "object" | "array"; + state: "keyOrEnd" | "key" | "colon" | "valueOrEnd" | "value" | "commaOrEnd"; + key?: string; + projectedMessages?: boolean; + lastMessage?: string; + hasMessage?: boolean; +}; + +type Capture = { depth: number; key?: string; text: BoundedText }; + +/** Validate one JSON object, retaining only the last top-level messages item. */ +export class PiTerminalJson { + private frames: Frame[] = []; + private fields = new Map(); + private fieldBytes = 2; + private failed = false; + private started = false; + private complete = false; + private capture?: Capture; + private token?: "string" | "number" | "literal"; + private keyText?: BoundedText; + private stringIsKey = false; + private escaped = false; + private unicodeRemaining = 0; + private numberState = ""; + private literal = ""; + private literalIndex = 0; + + write(chunk: string): void { + for (let i = 0; i < chunk.length && !this.failed; i++) this.character(chunk[i]); + } + + end(): Record | undefined { + if (this.token === "number" && this.numberCanEnd()) { + this.token = undefined; + this.finishValue(); + } + if (this.failed || this.token || !this.complete) return undefined; + const entries: string[] = []; + for (const [key, value] of this.fields) { + if (value === undefined) return undefined; + entries.push(`${JSON.stringify(key)}:${value}`); + } + const projected = `{${entries.join(",")}}`; + if (Buffer.byteLength(projected) > MAX_BYTES) return undefined; + try { return JSON.parse(projected) as Record; } catch { return undefined; } + } + + private character(character: string): void { + if (this.token === "number") { + if (this.numberCharacter(character)) { + this.capture?.text.append(character); + return; + } + if (!this.numberCanEnd()) { this.failed = true; return; } + this.token = undefined; + this.finishValue(); + } + if (this.token === "string") { + this.capture?.text.append(character); + this.keyText?.append(character); + this.stringCharacter(character); + return; + } + if (this.token === "literal") { + this.capture?.text.append(character); + if (character !== this.literal[this.literalIndex++]) { this.failed = true; return; } + if (this.literalIndex === this.literal.length) { + this.token = undefined; + this.finishValue(); + } + return; + } + if (character === " " || character === "\t" || character === "\r" || character === "\n") { + this.capture?.text.append(character); + return; + } + if (this.complete) { this.failed = true; return; } + const frame = this.frames.at(-1); + if (!frame) { + if (this.started || character !== "{") { this.failed = true; return; } + this.started = true; + this.frames.push({ kind: "object", state: "keyOrEnd" }); + return; + } + if (character === "}" || character === "]") { + this.closeContainer(character); + return; + } + if (frame.state === "key" || frame.state === "keyOrEnd") { + if (character !== '"') { this.failed = true; return; } + this.capture?.text.append(character); + this.startString(); + this.stringIsKey = true; + if (this.frames.length === 1) { + this.keyText = new BoundedText(); + this.keyText.append(character); + } + return; + } + if (frame.state === "colon") { + this.capture?.text.append(character); + if (character !== ":") { this.failed = true; return; } + frame.state = "value"; + return; + } + if (frame.state === "commaOrEnd") { + this.capture?.text.append(character); + if (character !== ",") { this.failed = true; return; } + frame.state = frame.kind === "object" ? "key" : "value"; + return; + } + this.startValue(character, frame); + } + + private startValue(character: string, frame: Frame): void { + const projectedMessages = this.frames.length === 1 && frame.key === "messages" && character === "["; + if (!this.capture && !projectedMessages && (this.frames.length === 1 || frame.projectedMessages)) { + this.capture = { depth: this.frames.length, key: this.frames.length === 1 ? frame.key : undefined, + text: new BoundedText() }; + } + this.capture?.text.append(character); + if (character === "{" || character === "[") { + if (this.frames.length >= MAX_DEPTH) { this.failed = true; return; } + this.frames.push({ kind: character === "{" ? "object" : "array", + state: character === "{" ? "keyOrEnd" : "valueOrEnd", projectedMessages }); + } else if (character === '"') { + this.startString(); + } else if (character === "-" || /[0-9]/.test(character)) { + this.token = "number"; + this.numberState = character === "-" ? "minus" : character === "0" ? "zero" : "integer"; + } else if (character === "t" || character === "f" || character === "n") { + this.token = "literal"; + this.literal = character === "t" ? "true" : character === "f" ? "false" : "null"; + this.literalIndex = 1; + } else this.failed = true; + } + + private startString(): void { + this.token = "string"; + this.stringIsKey = false; + this.escaped = false; + this.unicodeRemaining = 0; + } + + private stringCharacter(character: string): void { + if (this.unicodeRemaining) { + if (!/[0-9a-fA-F]/.test(character)) this.failed = true; + this.unicodeRemaining--; + } else if (this.escaped) { + this.escaped = false; + if (character === "u") this.unicodeRemaining = 4; + else if (!'"\\/bfnrt'.includes(character)) this.failed = true; + } else if (character === "\\") this.escaped = true; + else if (character === '"') { + this.token = undefined; + if (this.stringIsKey) { + const frame = this.frames.at(-1)!; + if (this.keyText) { + const raw = this.keyText.value(); + if (raw === undefined) { this.failed = true; return; } + frame.key = JSON.parse(raw) as string; + this.keyText = undefined; + } + frame.state = "colon"; + } else this.finishValue(); + } else if (character.charCodeAt(0) < 0x20) this.failed = true; + } + + private numberCharacter(character: string): boolean { + const digit = character >= "0" && character <= "9"; + switch (this.numberState) { + case "minus": + if (digit) { this.numberState = character === "0" ? "zero" : "integer"; return true; } + return false; + case "integer": + if (digit) return true; + // Both integer forms may continue with a fraction or exponent. + case "zero": + if (character === ".") { this.numberState = "dot"; return true; } + if (character === "e" || character === "E") { this.numberState = "exponent"; return true; } + return false; + case "dot": + if (digit) { this.numberState = "fraction"; return true; } + return false; + case "fraction": + if (digit) return true; + if (character === "e" || character === "E") { this.numberState = "exponent"; return true; } + return false; + case "exponent": + if (character === "+" || character === "-") { this.numberState = "exponentSign"; return true; } + if (digit) { this.numberState = "exponentDigits"; return true; } + return false; + case "exponentSign": + if (digit) { this.numberState = "exponentDigits"; return true; } + return false; + case "exponentDigits": return digit; + default: return false; + } + } + + private numberCanEnd(): boolean { + return ["zero", "integer", "fraction", "exponentDigits"].includes(this.numberState); + } + + private closeContainer(character: string): void { + const frame = this.frames.at(-1)!; + const matching = frame.kind === "object" ? character === "}" : character === "]"; + if (!matching || !["keyOrEnd", "valueOrEnd", "commaOrEnd"].includes(frame.state)) { + this.failed = true; + return; + } + this.capture?.text.append(character); + this.frames.pop(); + if (frame.projectedMessages) { + this.storeField("messages", frame.hasMessage + ? frame.lastMessage === undefined ? undefined : `[${frame.lastMessage}]` : "[]"); + } + this.finishValue(); + } + + private finishValue(): void { + const frame = this.frames.at(-1); + if (this.capture?.depth === this.frames.length) { + const { key, text } = this.capture; + if (key !== undefined) this.storeField(key, text.value()); + else if (frame?.projectedMessages) { + frame.lastMessage = text.value(); + frame.hasMessage = true; + } + this.capture = undefined; + } + if (frame) frame.state = "commaOrEnd"; + else this.complete = true; + } + + private storeField(key: string, value: string | undefined): void { + const previous = this.fields.get(key); + if (!this.fields.has(key)) this.fieldBytes += Buffer.byteLength(JSON.stringify(key)) + 2; + this.fieldBytes += Buffer.byteLength(value ?? "") - Buffer.byteLength(previous ?? ""); + if (this.fieldBytes > MAX_BYTES + 1) { this.failed = true; return; } + this.fields.set(key, value); + } +} diff --git a/tests/pi-completion-cli.test.ts b/tests/pi-completion-cli.test.ts index 034a2f2..4c44a7f 100644 --- a/tests/pi-completion-cli.test.ts +++ b/tests/pi-completion-cli.test.ts @@ -280,3 +280,23 @@ for (const text of ["", "done"]) { }); } } + +for (const text of ["", "done"]) { + for (const flags of [[], ["--sdk-format", "json"]]) { + test(`Pi cumulative terminal history preserves ${JSON.stringify(text)} in ${flags.join(" ") || "plain"}`, async () => { + const final = assistant("stop", text, 4, 6); + const records = completedTrace(final); + records[records.length - 2].messages = [ + ...Array.from({ length: 5 }, () => ({ role: "toolResult", + content: [{ type: "text", text: "x".repeat(1024 * 1024) }] })), final, + ]; + const f = fixture(records); + try { + const result = await f.run([...flags, "--session", "large-history"]); + assert.equal(result.code, 0, result.stderr || result.stdout); + assert.equal(flags.length ? jsonRows(result.stdout).at(-1)!.data.finalMessage : result.stdout.trim(), text); + assert.equal(readStoredSession(f.env, "pi", "large-history")?.nativeId, f.nativeSession); + } finally { f.cleanup(); } + }); + } +} diff --git a/tests/pi-completion.test.ts b/tests/pi-completion.test.ts index 16e79d4..1830a46 100644 --- a/tests/pi-completion.test.ts +++ b/tests/pi-completion.test.ts @@ -244,3 +244,34 @@ test("compaction cannot revive an absent, failed, or interrupted completion", () ]; for (const trace of traces) assert.notEqual(observe(...trace).outcome.status, "success"); }); + +test("accepts a small final answer after cumulative terminal history exceeds the capture limit", () => { + const history = Array.from({ length: 5 }, () => ({ + role: "toolResult", content: [{ type: "text", text: "x".repeat(1024 * 1024) }], + })); + for (const text of ["", "done"]) { + const trace = line({ type: "agent_start" }) + line(terminal(assistant(text), { + messages: [...history, assistant(text)], + })) + line({ type: "agent_settled" }); + const observer = new PiCompletionObserver(); + for (let start = 0; start < trace.length; start += 65536) observer.write(trace.slice(start, start + 65536)); + observer.end(); + assert.deepEqual(observer.outcome, { status: "success", finalMessage: text }); + } +}); + +test("large terminal history does not hide errors, retries, or malformed tails", () => { + const history = { role: "toolResult", content: [{ type: "text", text: "x".repeat(4 * 1024 * 1024) }] }; + for (const extra of [{ willRetry: true }, { messages: [history] }]) { + assert.notEqual(observe(terminal(assistant(), { messages: [history, assistant()], ...extra })).outcome.status, "success"); + } + const failure = assistant("", { stopReason: "error", errorMessage: "provider failed" }); + assert.deepEqual(observe(terminal(failure, { messages: [history, failure] })).outcome, + { status: "error", error: "provider failed" }); + for (const suffix of ["", ",broken}", '} trailing']) { + const observer = new PiCompletionObserver(); + observer.write(JSON.stringify(terminal(assistant(), { messages: [history, assistant()] })).slice(0, -1) + suffix); + observer.end(); + assert.notEqual(observer.outcome.status, "success"); + } +}); diff --git a/tests/pi-terminal-json.test.ts b/tests/pi-terminal-json.test.ts new file mode 100644 index 0000000..ee06883 --- /dev/null +++ b/tests/pi-terminal-json.test.ts @@ -0,0 +1,92 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { PiTerminalJson } from "../src/pi-terminal-json.ts"; + +function parse(source: string, chunkSize = source.length || 1) { + const parser = new PiTerminalJson(); + for (let i = 0; i < source.length; i += chunkSize) parser.write(source.slice(i, i + chunkSize)); + return parser.end(); +} + +const final = { role: "assistant", content: [{ type: "text", text: "done" }], stopReason: "stop" }; + +test("projects cumulative messages exceeding the limit while retaining the last message", () => { + const source = JSON.stringify({ type: "agent_end", messages: ["x".repeat(4 * 1024 * 1024), final] }); + assert.deepEqual(parse(source, 8191), { type: "agent_end", messages: [final] }); + assert.equal(parse(JSON.stringify({ type: "agent_end", messages: [final, "x".repeat(4 * 1024 * 1024)] })), undefined); +}); + +test("handles every character boundary and escaped top-level keys", () => { + const source = '{"messa\\u0067es":[{"ignored":[null,false,true,-2.3e+4]},' + JSON.stringify(final) + '],"type":"agent_end"}'; + for (let split = 0; split <= source.length; split++) { + const parser = new PiTerminalJson(); + parser.write(source.slice(0, split)); + parser.write(source.slice(split)); + assert.deepEqual(parser.end(), { messages: [final], type: "agent_end" }); + } +}); + +test("preserves duplicate-key semantics and does not project nested messages", () => { + assert.deepEqual(parse('{"messages":[1,2],"messages":[3,4],"type":"wrong","type":"agent_end"}'), + { messages: [4], type: "agent_end" }); + assert.deepEqual(parse('{"messages":[1],"messages":null,"nested":{"messages":[1,2]}}'), + { messages: null, nested: { messages: [1, 2] } }); + assert.deepEqual(parse('{"messages":null,"messages":[]}'), { messages: [] }); + assert.deepEqual(parse('{"__proto__":{"polluted":true},"messages":[1,2]}'), + JSON.parse('{"__proto__":{"polluted":true},"messages":[2]}')); +}); + +test("rejects malformed syntax even in discarded large history", () => { + for (const invalid of ['01', '1.', '1e+', 'tru', 'undefined', '"bad\\x"', '"bad\n"', + '{"a":1,}', '[1,]', '{"a" 1}', '[,1]', '{1:2}', '"\\u00xz"']) { + assert.equal(parse('{"messages":[' + JSON.stringify("x".repeat(4 * 1024 * 1024)) + ',' + invalid + ',1]}', 16381), undefined, invalid); + } + for (const invalid of ['{} trailing', '{}{}', '{', '[]', 'null', '{"messages":[1,2]'] ) { + assert.equal(parse(invalid, 1), undefined, invalid); + } +}); + +test("bounds nesting and the complete projected envelope", () => { + assert.equal(parse('{"messages":[' + '['.repeat(257) + '1' + ']'.repeat(257) + ',1]}'), undefined); + assert.equal(parse(JSON.stringify({ type: "agent_end", padding: "x".repeat(3 * 1024 * 1024), + messages: ["x".repeat(2 * 1024 * 1024)] })), undefined); +}); + +test("matches JSON.parse across deterministic nested values and streaming boundaries", () => { + let seed = 42; + const random = (limit: number) => { seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0; return seed % limit; }; + const scalars = [null, true, false, 0, -17, 1.3e40, -0.004, "", "quote\"\\\n\t", "日本語🌱", "\ud800"]; + function value(depth: number): unknown { + if (depth === 0 || random(3) === 0) return scalars[random(scalars.length)]; + if (random(2)) return Array.from({ length: random(4) }, () => value(depth - 1)); + return Object.fromEntries(Array.from({ length: random(4) }, (_, i) => [`key${i}`, value(depth - 1)])); + } + for (let i = 0; i < 100; i++) { + const source = JSON.stringify({ metadata: value(3), messages: Array.from({ length: random(5) }, () => value(3)), + type: "agent_end" }, null, i % 2 ? 2 : undefined); + const expected = JSON.parse(source); + expected.messages = expected.messages.slice(-1); + for (const chunkSize of [1, 7, 64]) assert.deepEqual(parse(source, chunkSize), expected); + } +}); + +test("rejects invalid delimiter, number, literal, and string mutations", () => { + const mutations = ['{"messages";[1]}', '{"messages":[1 2]}', '{"messages":[truefalse]}', + '{"messages":[+1]}', '{"messages":[-.2]}', '{"messages":[1e]}', '{"messages":[1.e2]}', + '{"messages":[false,]}', '{"messages":["\\u123"]}', '{"messages":["\\v"]}', + '{"messages":[1]}\u00a0', '{"messages":[1]}\0', '{"messages":{]}', '{"messages":{,}}']; + for (const source of mutations) { + assert.throws(() => JSON.parse(source)); + for (const chunkSize of [1, 5, 64]) assert.equal(parse(source, chunkSize), undefined, source); + } +}); + +test("applies the exact envelope byte limit across surrogate-pair chunk boundaries", () => { + const overhead = Buffer.byteLength(JSON.stringify({ messages: [""] })); + const text = "🌱".repeat(Math.floor((4 * 1024 * 1024 - overhead) / 4)); + const padding = "x".repeat(4 * 1024 * 1024 - overhead - Buffer.byteLength(text)); + const source = JSON.stringify({ messages: [text + padding] }); + assert.equal(Buffer.byteLength(source), 4 * 1024 * 1024); + assert.deepEqual(parse(source, 8191), { messages: [text + padding] }); + assert.equal(parse(JSON.stringify({ messages: [text + padding + "x"] }), 8191), undefined); +});