From 8aaef593391e8c865b069f77c02a24d3bc1db620 Mon Sep 17 00:00:00 2001 From: Serhi <63193792+serhiq@users.noreply.github.com> Date: Fri, 16 Oct 2026 01:24:06 +0200 Subject: [PATCH] fix(pi-local): add stdout compaction to stop quadratic run-log growth (#10249) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The `pi_local` adapter runs the `pi` CLI in `--mode json` and records its NDJSON stream as a run-log per turn > - In JSON mode `pi` emits one `message_update` event per streaming token, each carrying the **full cumulative assistant message**, so a single turn's run-log grows quadratically with output length (measured ~77 MB for one thinking-model turn) > - The root cause is in `pi`, but the Pi project declined to add a compact mode (earendil-works/pi#3905, closed the same day), so Paperclip has to filter on the consumer side > - This PR adds a dedicated, config-gated compactor at the `pi_local` stdout boundary (`stdoutLogMode`: `raw` | `compact`, default `compact`) that strips the cumulative snapshots while preserving streaming deltas, terminal events, and parser/UI output > - The benefit is linear run-log growth (~150× smaller in practice) with **no change** to `parsePiJsonl` results, token usage, or live typing — keeping storage and replay bounded ## Linked Issues or Issue Description - **Refs #5699** — same root cause ("full accumulated partial state" → 46 MB log, per that PR). This PR takes a **different, single-concern approach**: a dedicated `stdout-compaction.ts` module with an explicit `stdoutLogMode` config (default `compact`, fail-safe fallback to `raw`), scoped to `pi_local` only. #5699 instead filters inline in `buffered-on-log.ts`, bundles an unrelated `claude-local` change, and is currently **conflicting** + stale (last updated 2026-05-14). Happy to coordinate with @ukint-vs on a merged direction. - **Refs earendil-works/pi#3905** — proposed adding a compact JSON mode upstream in `pi`; closed by the maintainer, which is why this filters on the Paperclip side rather than the producer. - **Related #1846** (run-log/retention TTL) — complementary: this PR reduces what's stored per run; #1846 addresses how long rows live. - No public GitHub issue exists that is an exact duplicate of the `pi_local` quadratic-growth symptom; the problem is tracked via #5699 above. (No internal/instance-local references used.) ## What Changed - **`packages/adapters/pi-local/src/server/stdout-compaction.ts`** (new, 123 lines) — `createPiStdoutLogCompactor(mode)`. `raw` passes lines through unchanged; `compact` strips the cumulative `message`/`assistantMessageEvent.partial` from `message_update` events while preserving streaming deltas, terminal events, and passthrough of unknown event types. Default mode `compact`; an invalid explicit mode fails safely to `raw`. - **`packages/adapters/pi-local/src/server/execute.ts`** — read the mode via `resolvePiStdoutLogMode(config)` and apply the compactor at the stdout → `onLog` (run-log) boundary. - **`packages/adapters/pi-local/src/index.ts`** — expose the `stdoutLogMode` config option (documented inline: `string`, optional, default `"compact"`). - **`packages/adapters/pi-local/src/ui/parse-stdout.ts`** — confirm the stdout parser consumes compacted output unchanged. - **Tests** — new `stdout-compaction.test.ts` (359 lines) covering modes, edge cases, and `parsePiJsonl` replay equivalence; updated `execute.remote.test.ts`. ## Verification - **Unit:** `pnpm --filter @paperclipai/adapter-pi-local exec vitest run` → **121/121 passing (12 files)**. - **Typecheck:** `pnpm --filter @paperclipai/adapter-pi-local exec tsc --noEmit` → **clean, 0 errors**. - **Real-data replay:** a real `pi --mode json` stream (`zai/glm-5.2`, thinking=high) compacted line-by-line through this compactor: - Raw **76.96 MB** → Compact **525.5 KB** = **99.33% reduction (≈150×)**. - 4503 lines in, 4503 out (**0 dropped**); **0** `message_update` with a cumulative `.message` remaining; **4489** streaming deltas preserved. - **Parser equivalence:** `parsePiJsonl(raw) ≡ parsePiJsonl(compact)` — **0 errors** on both; identical `finalMessage` (13,369 chars); identical `usage` (`inputTokens 2896, outputTokens 4491, cachedInputTokens 6848`). Result, token accounting, and live-typing are unchanged. ## Risks - **Behavioral shift:** default `compact` changes what is stored in run-logs vs today. Operators who need the verbatim stream can set `stdoutLogMode: "raw"`. Flagging for reviewer preference — happy to flip the default to `raw` for upstream if that's preferred. - **Ordering with secret redaction (#8027 / #9856):** compaction is structural and runs at the stdout boundary **before** redaction; it only removes cumulative duplicates and never alters token content, so it does not change what redaction sees. - **Unknown event types** pass through unchanged (forward-compatible with future `pi` events). - Low risk otherwise — the change is contained to the `pi_local` adapter. ## Model Used OpenAI `gpt-5.6-luna` — via pi's `openai-codex` provider (272K context window). Used for implementation and for the real-data measurement in Verification. Reasoning/thinking: extended thinking enabled at level `high` (pi `--thinking high`; maps to OpenAI reasoning effort). `gpt-5.6-luna` is a reasoning-capable model. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip Co-authored-by: commitperclip[bot] <284812584+commitperclip[bot]@users.noreply.github.com> --- packages/adapters/pi-local/src/index.ts | 14 + .../src/server/execute.remote.test.ts | 109 ++++++ .../adapters/pi-local/src/server/execute.ts | 19 +- .../src/server/stdout-compaction.test.ts | 359 ++++++++++++++++++ .../pi-local/src/server/stdout-compaction.ts | 128 +++++++ .../adapters/pi-local/src/ui/parse-stdout.ts | 11 +- 6 files changed, 634 insertions(+), 6 deletions(-) create mode 100644 packages/adapters/pi-local/src/server/stdout-compaction.test.ts create mode 100644 packages/adapters/pi-local/src/server/stdout-compaction.ts diff --git a/packages/adapters/pi-local/src/index.ts b/packages/adapters/pi-local/src/index.ts index 7face86512e..1a8cefee19b 100644 --- a/packages/adapters/pi-local/src/index.ts +++ b/packages/adapters/pi-local/src/index.ts @@ -32,6 +32,20 @@ Core fields: Operational fields: - timeoutSec (number, optional): run timeout in seconds - graceSec (number, optional): SIGTERM grace period in seconds +- stdoutLogMode (string, optional, default "compact"): controls how Pi --mode json stdout is + compacted before entering the Paperclip run log. Pi emits cumulative state + in every message_update event, which makes run logs grow ~quadratically. + Modes: + - "raw": legacy passthrough, no filtering (full transcript; large logs). + - "compact" (default): strips cumulative copies (message_update.message, + assistantMessageEvent.partial), converts tool_execution_update to a + throttled content-free Paperclip progress event. Streaming deltas are + preserved — parsePiJsonl output and live UI typing are unchanged. + In compact mode, if a run is terminated before tool_execution_end, its + partial tool output is not retained; use "raw" when full forensic output is required. + Oversized individual lines are handled by the server's run-log pipeline + (redaction + head/tail cap) regardless of mode. Invalid explicit mode values + fail safe to "raw" rather than enabling a lossy transformation. Notes: - Pi supports multiple providers and models. Use \`pi --list-models\` to list available options. diff --git a/packages/adapters/pi-local/src/server/execute.remote.test.ts b/packages/adapters/pi-local/src/server/execute.remote.test.ts index 6f443902ff3..4195a49639b 100644 --- a/packages/adapters/pi-local/src/server/execute.remote.test.ts +++ b/packages/adapters/pi-local/src/server/execute.remote.test.ts @@ -777,4 +777,113 @@ describe("pi remote execution", () => { expect(userPrompt).not.toContain("CUSTOM POLICY run-instructions-policy"); expect(userPrompt).not.toContain("Execution contract:"); }); + + it("compacts streamed chunks while parsing the untouched raw stdout result", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-pi-remote-log-boundary-")); + cleanupDirs.push(rootDir); + const workspaceDir = path.join(rootDir, "workspace"); + await mkdir(workspaceDir, { recursive: true }); + + const sessionLine = JSON.stringify({ type: "session", id: "session-boundary", cwd: "/remote/workspace" }); + const messageLine = JSON.stringify({ + type: "message_update", + message: { + role: "assistant", + content: [{ type: "text", text: "Hello 😀" }], + }, + assistantMessageEvent: { + type: "text_delta", + contentIndex: 0, + delta: "Hello 😀", + partial: { + role: "assistant", + content: [{ type: "text", text: "Hello 😀" }], + }, + }, + }); + const toolUpdateLine = JSON.stringify({ + type: "tool_execution_update", + toolCallId: "tc-boundary", + toolName: "bash", + args: { command: "printf progress" }, + partialResult: { + content: [{ type: "text", text: "large cumulative output".repeat(1_000) }], + details: {}, + }, + }); + // CRLF on the passthrough session line, LF on the transformed message, + // and no newline on the final line exercise every buffering exit path. + const rawStdout = `${sessionLine}\r\n${messageLine}\n${toolUpdateLine}`; + + runChildProcess.mockImplementationOnce(async (...args: unknown[]) => { + const options = args[3] as { + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; + }; + const emojiSplit = rawStdout.indexOf("😀") + 1; + await options.onLog?.("stdout", rawStdout.slice(0, emojiSplit)); + await options.onLog?.("stderr", "diagnostic stderr\n"); + await options.onLog?.("stdout", rawStdout.slice(emojiSplit)); + return { + exitCode: 0, + signal: null, + timedOut: false, + stdout: rawStdout, + stderr: "diagnostic stderr\n", + pid: 456, + startedAt: new Date().toISOString(), + }; + }); + + const logs: Array<{ stream: "stdout" | "stderr"; chunk: string }> = []; + const result = await execute({ + runId: "run-log-boundary", + agent: { + id: "agent-1", + companyId: "company-1", + name: "Pi Builder", + adapterType: "pi_local", + adapterConfig: {}, + }, + runtime: { + sessionId: null, + sessionParams: null, + sessionDisplayId: null, + taskKey: null, + }, + config: { + command: "pi", + model: "openai/gpt-5.4-mini", + stdoutLogMode: "compact", + }, + context: { + paperclipWorkspace: { cwd: workspaceDir, source: "project_primary" }, + }, + executionTransport: { + remoteExecution: { + host: "127.0.0.1", + port: 2222, + username: "fixture", + remoteWorkspacePath: "/remote/workspace", + remoteCwd: "/remote/workspace", + privateKey: "PRIVATE KEY", + knownHosts: "[127.0.0.1]:2222 ssh-ed25519 AAAA", + strictHostKeyChecking: true, + }, + }, + onLog: async (stream, chunk) => { + logs.push({ stream, chunk }); + }, + }); + + const stdoutLog = logs.filter((entry) => entry.stream === "stdout").map((entry) => entry.chunk).join(""); + expect(stdoutLog).toContain(`${sessionLine}\r\n`); + expect(stdoutLog).toContain('"type":"message_update"'); + expect(stdoutLog).toContain('"delta":"Hello 😀"'); + expect(stdoutLog).toContain('"sourceEventType":"tool_execution_update"'); + expect(stdoutLog).not.toContain("large cumulative output"); + expect(logs).toContainEqual({ stream: "stderr", chunk: "diagnostic stderr\n" }); + + expect(result.summary).toBe("Hello 😀"); + expect(result.resultJson).toEqual({ stdout: rawStdout, stderr: "diagnostic stderr\n" }); + }); }); diff --git a/packages/adapters/pi-local/src/server/execute.ts b/packages/adapters/pi-local/src/server/execute.ts index d2daf7e6a85..ad68313578b 100644 --- a/packages/adapters/pi-local/src/server/execute.ts +++ b/packages/adapters/pi-local/src/server/execute.ts @@ -56,6 +56,7 @@ import { } from "@paperclipai/adapter-utils/server-utils"; import { shellQuote } from "@paperclipai/adapter-utils/ssh"; import { isPiUnknownSessionError, parsePiJsonl, createPiJsonlParser } from "./parse.js"; +import { createPiStdoutCompactor, resolvePiStdoutLogMode } from "./stdout-compaction.js"; import { ensurePiModelConfiguredAndAvailable } from "./models.js"; import { preparePiRuntimeConfig } from "./runtime-config.js"; import { SANDBOX_INSTALL_COMMAND } from "../index.js"; @@ -241,6 +242,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { const notes = [...preparedRuntimeConfig.notes]; + if (stdoutLogMode !== "raw") { + notes.push(`Pi stdout run-log compaction mode: ${stdoutLogMode}`); + } if (!resolvedInstructionsFilePath) return notes; if (instructionsReadFailed) { notes.push( @@ -716,6 +721,14 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const compacted = compactStdoutLine(line); + if (compacted !== null) { + await onLog("stdout", compacted + suffix); + } + }; const bufferedOnLog = async (stream: "stdout" | "stderr", chunk: string) => { if (stream === "stderr") { // Pass stderr through immediately (not JSONL) @@ -723,16 +736,14 @@ export async function execute(ctx: AdapterExecutionContext): Promise, assistantMessageEvent: Record) { + return JSON.stringify({ + type: "message_update", + message, + assistantMessageEvent: { ...assistantMessageEvent, partial: message }, + }); +} + +/** + * Protocol-faithful Pi 0.74 multi-turn sequence derived from pi-agent-core's + * event order and published AgentEvent/AssistantMessageEvent shapes. + */ +function buildTranscript() { + const usage1 = { + input: 100, + output: 20, + cacheRead: 10, + cacheWrite: 0, + totalTokens: 130, + cost: { input: 0.001, output: 0.0004, cacheRead: 0.0001, cacheWrite: 0, total: 0.0015 }, + }; + const usage2 = { + input: 120, + output: 40, + cacheRead: 12, + cacheWrite: 0, + totalTokens: 172, + cost: { input: 0.0012, output: 0.0008, cacheRead: 0.00012, cacheWrite: 0, total: 0.00212 }, + }; + const userMessage = { role: "user", content: "List files, then summarize.", timestamp: 1 }; + const firstAssistant = { + role: "assistant", + content: [ + { type: "thinking", thinking: "I should inspect the workspace." }, + { type: "toolCall", id: "tc-1", name: "bash", arguments: { command: "ls" } }, + ], + api: "openai-responses", + provider: "openai", + model: "gpt-5", + usage: usage1, + stopReason: "toolUse", + timestamp: 2, + }; + const toolResult = { + role: "toolResult", + toolCallId: "tc-1", + toolName: "bash", + content: [{ type: "text", text: "file1\nfile2" }], + details: { truncation: null, fullOutputPath: null }, + isError: false, + timestamp: 3, + }; + const finalAssistant = { + role: "assistant", + content: [ + { type: "thinking", thinking: "The directory contains two files." }, + { type: "text", text: "Hello world!" }, + ], + api: "openai-responses", + provider: "openai", + model: "gpt-5", + responseId: "resp-final", + usage: usage2, + stopReason: "stop", + timestamp: 4, + }; + + const thinkingStart = { ...firstAssistant, content: [{ type: "thinking", thinking: "" }] }; + const thinkingPartial = { ...firstAssistant, content: [{ type: "thinking", thinking: "I should inspect" }] }; + const thinkingComplete = { + ...firstAssistant, + content: [{ type: "thinking", thinking: "I should inspect the workspace." }], + }; + const toolCallStart = { + ...firstAssistant, + content: [ + { type: "thinking", thinking: "I should inspect the workspace." }, + { type: "toolCall", id: "tc-1", name: "bash", arguments: {} }, + ], + }; + const toolCallPartial = { + ...firstAssistant, + content: [ + { type: "thinking", thinking: "I should inspect the workspace." }, + { type: "toolCall", id: "tc-1", name: "bash", arguments: { command: "l" } }, + ], + }; + const finalThinkingStart = { + ...finalAssistant, + content: [{ type: "thinking", thinking: "" }], + }; + const finalThinkingPartial = { + ...finalAssistant, + content: [{ type: "thinking", thinking: "The directory contains" }], + }; + const finalThinkingComplete = { + ...finalAssistant, + content: [{ type: "thinking", thinking: "The directory contains two files." }], + }; + const finalTextStart = { + ...finalAssistant, + content: [ + { type: "thinking", thinking: "The directory contains two files." }, + { type: "text", text: "" }, + ], + }; + const finalTextPartial = { + ...finalAssistant, + content: [ + { type: "thinking", thinking: "The directory contains two files." }, + { type: "text", text: "Hello" }, + ], + }; + + return [ + JSON.stringify({ type: "session", id: "s-1", cwd: "/tmp/work" }), + JSON.stringify({ type: "agent_start" }), + JSON.stringify({ type: "turn_start" }), + JSON.stringify({ type: "message_start", message: { ...firstAssistant, content: [] } }), + assistantUpdate(thinkingStart, { type: "thinking_start", contentIndex: 0 }), + assistantUpdate(thinkingPartial, { type: "thinking_delta", contentIndex: 0, delta: "I should inspect" }), + assistantUpdate(thinkingComplete, { + type: "thinking_end", + contentIndex: 0, + content: "I should inspect the workspace.", + }), + assistantUpdate(toolCallStart, { type: "toolcall_start", contentIndex: 1 }), + assistantUpdate(toolCallPartial, { type: "toolcall_delta", contentIndex: 1, delta: '"ls"' }), + assistantUpdate(firstAssistant, { + type: "toolcall_end", + contentIndex: 1, + toolCall: firstAssistant.content[1], + }), + JSON.stringify({ type: "message_end", message: firstAssistant }), + JSON.stringify({ + type: "tool_execution_start", + toolCallId: "tc-1", + toolName: "bash", + args: { command: "ls" }, + }), + toolExecutionUpdate("tc-1", "file"), + toolExecutionUpdate("tc-1", "file1\nfile2"), + JSON.stringify({ + type: "tool_execution_end", + toolCallId: "tc-1", + toolName: "bash", + result: { content: toolResult.content, details: toolResult.details }, + isError: false, + }), + JSON.stringify({ type: "message_start", message: toolResult }), + JSON.stringify({ type: "message_end", message: toolResult }), + JSON.stringify({ type: "turn_end", message: firstAssistant, toolResults: [toolResult] }), + JSON.stringify({ type: "turn_start" }), + JSON.stringify({ type: "message_start", message: { ...finalAssistant, content: [] } }), + assistantUpdate(finalThinkingStart, { type: "thinking_start", contentIndex: 0 }), + assistantUpdate(finalThinkingPartial, { + type: "thinking_delta", + contentIndex: 0, + delta: "The directory contains", + }), + assistantUpdate(finalThinkingComplete, { + type: "thinking_end", + contentIndex: 0, + content: "The directory contains two files.", + }), + assistantUpdate(finalTextStart, { type: "text_start", contentIndex: 1 }), + assistantUpdate(finalTextPartial, { type: "text_delta", contentIndex: 1, delta: "Hello" }), + assistantUpdate(finalAssistant, { type: "text_delta", contentIndex: 1, delta: " world!" }), + assistantUpdate(finalAssistant, { type: "text_end", contentIndex: 1, content: "Hello world!" }), + JSON.stringify({ type: "message_end", message: finalAssistant }), + JSON.stringify({ type: "turn_end", message: finalAssistant, toolResults: [] }), + JSON.stringify({ + type: "agent_end", + messages: [userMessage, firstAssistant, toolResult, finalAssistant], + }), + JSON.stringify({ type: "agent_settled" }), + ]; +} + +function filterTranscript(lines: string[], mode: PiStdoutLogMode): string[] { + const compact = createPiStdoutCompactor(mode); + return lines.map((l) => compact(l)).filter((l): l is string => l !== null); +} + +describe("resolvePiStdoutLogMode", () => { + it("defaults missing values to compact but fails invalid values safely to raw", () => { + expect(resolvePiStdoutLogMode({})).toBe("compact"); + expect(resolvePiStdoutLogMode({ stdoutLogMode: undefined })).toBe("compact"); + expect(resolvePiStdoutLogMode({ stdoutLogMode: "bogus" })).toBe("raw"); + expect(resolvePiStdoutLogMode({ stdoutLogMode: 42 })).toBe("raw"); + }); + + it("accepts explicit modes case-insensitively", () => { + expect(resolvePiStdoutLogMode({ stdoutLogMode: "raw" })).toBe("raw"); + expect(resolvePiStdoutLogMode({ stdoutLogMode: " Compact " })).toBe("compact"); + }); +}); + +describe("raw mode", () => { + it("passes all input through unchanged", () => { + const lines = [ + ...buildTranscript(), + "not json at all", + "{broken", + "42", + "[1,2,3]", + '"just a string"', + "line with \u2028 separator", + ]; + expect(filterTranscript(lines, "raw")).toEqual(lines); + }); +}); + +describe("compact mode", () => { + it("strips cumulative message and partial from message_update, keeps delta", () => { + const out = filterTranscript([messageUpdate("Hello", "Hello")], "compact"); + expect(out).toHaveLength(1); + const event = JSON.parse(out[0]); + expect(event.type).toBe("message_update"); + expect(event.message).toBeUndefined(); + expect(event.assistantMessageEvent.type).toBe("text_delta"); + expect(event.assistantMessageEvent.delta).toBe("Hello"); + expect(event.assistantMessageEvent.contentIndex).toBe(1); + expect(event.assistantMessageEvent.partial).toBeUndefined(); + }); + + it("preserves terminal and unknown events byte-for-byte (including CRLF)", () => { + const terminal = buildTranscript().filter((l) => { + const t = JSON.parse(l).type; + return !["message_update", "tool_execution_update"].includes(t); + }); + const unknown = JSON.stringify({ type: "brand_new_pi_event", payload: { big: "z".repeat(1000) } }); + // Add CRLF to a terminal line and verify it survives byte-for-byte. + const crlfLine = terminal[0] + "\r"; + const out = filterTranscript([...terminal, unknown, crlfLine], "compact"); + expect(out).toEqual([...terminal, unknown, crlfLine]); + }); + + it("passes malformed lines through unchanged", () => { + const weird = ["not json", "{broken", "42", "[1,2]"]; + expect(filterTranscript(weird, "compact")).toEqual(weird); + }); + + it("shrinks a cumulative transcript to ~linear size", () => { + const n = 50; + const lines: string[] = []; + let text = ""; + for (let i = 0; i < n; i++) { + text += `chunk-${i} `; + lines.push(messageUpdate(`chunk-${i} `, text)); + } + const rawBytes = lines.join("\n").length; + const compactBytes = filterTranscript(lines, "compact").join("\n").length; + expect(compactBytes).toBeLessThan(rawBytes / 10); + }); +}); + +describe("tool_execution_update liveness", () => { + it("compact emits schema-distinct Paperclip progress at 0/5000ms boundaries", () => { + let now = 0; + const compact = createPiStdoutCompactor("compact", { now: () => now }); + const line = toolExecutionUpdate("tc-1", "a".repeat(50_000)); + + const first = compact(line); + now = 4_999; + const beforeBoundary = compact(line); + now = 5_000; + const atBoundary = compact(line); + + expect(first).not.toBeNull(); + expect(beforeBoundary).toBeNull(); + expect(atBoundary).not.toBeNull(); + expect(JSON.parse(first!)).toEqual({ + type: "paperclip_progress", + sourceEventType: "tool_execution_update", + toolCallId: "tc-1", + }); + expect(JSON.parse(atBoundary!)).toEqual(JSON.parse(first!)); + }); + + it("raw mode passes partialResult through unchanged", () => { + const line = toolExecutionUpdate("tc-4", "c".repeat(100)); + expect(filterTranscript([line], "raw")).toEqual([line]); + }); +}); + +describe("replay through Paperclip parsers", () => { + const raw = buildTranscript(); + + it("compact preserves full parsePiJsonl output (finalMessage, usage, toolCalls, errors, messages)", () => { + const filtered = filterTranscript(raw, "compact"); + const fromRaw = parsePiJsonl(raw.join("\n")); + const fromFiltered = parsePiJsonl(filtered.join("\n")); + + expect(fromFiltered).toEqual(fromRaw); + expect(fromFiltered.finalMessage).toBe("Hello world!"); + }); + + it("compact produces identical full UI transcript entries", () => { + const render = (lines: string[]) => { + resetParserState(); + return lines.flatMap((line) => parsePiStdoutLine(line, "2026-07-24T00:00:00Z")); + }; + expect(render(filterTranscript(raw, "compact"))).toEqual(render(raw)); + }); + + it("Paperclip progress events stay out of the UI transcript", () => { + const event = { + type: "paperclip_progress", + sourceEventType: "tool_execution_update", + toolCallId: "tc-1", + }; + expect(parsePiStdoutLine(JSON.stringify(event), "2026-07-24T00:00:00Z")).toEqual([]); + }); +}); diff --git a/packages/adapters/pi-local/src/server/stdout-compaction.ts b/packages/adapters/pi-local/src/server/stdout-compaction.ts new file mode 100644 index 00000000000..aeec4156b7f --- /dev/null +++ b/packages/adapters/pi-local/src/server/stdout-compaction.ts @@ -0,0 +1,128 @@ +import { performance } from "node:perf_hooks"; +import { parseJson } from "@paperclipai/adapter-utils/server-utils"; + +/** + * Stateful JSONL compaction for Pi `--mode json` stdout before it enters + * the Paperclip run-log pipeline. + * + * Pi emits `message_update` events that repeat the FULL cumulative assistant + * message on every streaming delta. Persisting each line makes run-log size + * grow ~quadratically. In `compact` mode, stripping the cumulative copies + * keeps growth linear without changing `parsePiJsonl` output or UI live typing. + * + * No truncation here: it would run before the server's secret redaction and + * can split a secret across the cut boundary. The server re-caps after + * redaction; oversized content is its responsibility. + * + * Tool update streams become throttled Paperclip progress events so the + * stale-run watchdog's `lastOutputAt` stays fresh during long tools. + */ + +const PI_STDOUT_LOG_MODES = ["raw", "compact"] as const; + +export type PiStdoutLogMode = (typeof PI_STDOUT_LOG_MODES)[number]; + +const DEFAULT_PI_STDOUT_LOG_MODE: PiStdoutLogMode = "compact"; + +const PROGRESS_INTERVAL_MS = 5_000; + +function isPiStdoutLogMode(value: string): value is PiStdoutLogMode { + return PI_STDOUT_LOG_MODES.includes(value as PiStdoutLogMode); +} + +export function resolvePiStdoutLogMode(config: Record): PiStdoutLogMode { + if (config.stdoutLogMode === undefined || config.stdoutLogMode === null) { + return DEFAULT_PI_STDOUT_LOG_MODE; + } + if (typeof config.stdoutLogMode !== "string") return "raw"; + const normalized = config.stdoutLogMode.trim().toLowerCase(); + if (isPiStdoutLogMode(normalized)) { + return normalized; + } + // Invalid explicit values fail safe to the lossless legacy stream rather + // than silently enabling a lossy transformation. + return "raw"; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function parseLineEvent(line: string): Record | null { + const trimmed = line.endsWith("\r") ? line.slice(0, -1) : line; + const event = parseJson(trimmed); + return isRecord(event) ? event : null; +} + +function slimAssistantMessageEvent(update: unknown): unknown { + if (!isRecord(update)) return update; + const { partial: _partial, ...rest } = update; + return rest; +} + +type Clock = () => number; + +function createThrottle(intervalMs: number, now: Clock): () => boolean { + let lastAt = Number.NEGATIVE_INFINITY; + return () => { + const current = now(); + if (current - lastAt < intervalMs) return false; + lastAt = current; + return true; + }; +} + +function progressTick(allow: () => boolean, toolCallId?: unknown): string | null { + if (!allow()) return null; + // `paperclip_` prefix marks this as adapter metadata, not a Pi event. + return JSON.stringify({ + type: "paperclip_progress", + sourceEventType: "tool_execution_update", + ...(typeof toolCallId === "string" ? { toolCallId } : {}), + }); +} + +type PiStdoutCompactor = (line: string) => string | null; + +interface PiStdoutCompactorOptions { + now?: Clock; +} + +function createCompactCompactor(now: Clock): PiStdoutCompactor { + const allowToolTick = createThrottle(PROGRESS_INTERVAL_MS, now); + return (line) => { + const event = parseLineEvent(line); + if (!event) return line; + + if (event.type === "tool_execution_update") { + return progressTick(allowToolTick, event.toolCallId); + } + // Non-`message_update` events pass through untouched — including + // `tool_execution_end`, whose `.result` can be tens of MiB (large + // tool/bash output). Capping that is an intentional separate follow-up: + // truncation here would run before the server's secret redaction and + // could split a secret across the cut boundary. See #5699. + if (event.type !== "message_update") return line; + + const { message: _cumulativeMessage, ...rest } = event; + return JSON.stringify({ + ...rest, + ...(event.assistantMessageEvent !== undefined + ? { assistantMessageEvent: slimAssistantMessageEvent(event.assistantMessageEvent) } + : {}), + }); + }; +} + +export function createPiStdoutCompactor( + mode: PiStdoutLogMode, + options: PiStdoutCompactorOptions = {}, +): PiStdoutCompactor { + const now = options.now ?? (() => performance.now()); + switch (mode) { + case "raw": + return (line) => line; + case "compact": + return createCompactCompactor(now); + } +} diff --git a/packages/adapters/pi-local/src/ui/parse-stdout.ts b/packages/adapters/pi-local/src/ui/parse-stdout.ts index 3dde76c61f1..9b377412b4d 100644 --- a/packages/adapters/pi-local/src/ui/parse-stdout.ts +++ b/packages/adapters/pi-local/src/ui/parse-stdout.ts @@ -54,8 +54,15 @@ export function parsePiStdoutLine(line: string, ts: string): TranscriptEntry[] { const type = asString(parsed.type); - // RPC protocol messages - filter these out (internal implementation detail) - if (type === "response" || type === "extension_ui_request" || type === "extension_ui_response" || type === "extension_error") { + // RPC protocol messages and adapter-generated progress metadata are + // internal implementation details, not transcript entries. + if ( + type === "response" || + type === "extension_ui_request" || + type === "extension_ui_response" || + type === "extension_error" || + type === "paperclip_progress" + ) { return []; }