mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 00:05:49 +02:00
fix(pi-local): add stdout compaction to stop quadratic run-log growth (#10249)
## 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 <noreply@paperclip.ing>
Co-authored-by: commitperclip[bot] <284812584+commitperclip[bot]@users.noreply.github.com>
This commit is contained in:
6 files changed
+634
-6
No files matched your search
@@ -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.
|
||||
|
||||
@@ -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<void>;
|
||||
};
|
||||
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" });
|
||||
});
|
||||
});
|
||||
@@ -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<AdapterExec
|
||||
const command = asString(config.command, "pi");
|
||||
const model = asString(config.model, "").trim();
|
||||
const thinking = asString(config.thinking, "").trim();
|
||||
const stdoutLogMode = resolvePiStdoutLogMode(config);
|
||||
|
||||
// Parse model into provider and model id
|
||||
const provider = parseModelProvider(model);
|
||||
@@ -622,6 +624,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
|
||||
const commandNotes = (() => {
|
||||
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<AdapterExec
|
||||
|
||||
// Buffer stdout by lines to handle partial JSON chunks
|
||||
let stdoutBuffer = "";
|
||||
const compactStdoutLine = createPiStdoutCompactor(stdoutLogMode);
|
||||
// Compaction only affects the persisted run log; parsePiJsonl reads the unfiltered proc.stdout.
|
||||
const emitCompacted = async (line: string, suffix: string) => {
|
||||
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<AdapterExec
|
||||
return;
|
||||
}
|
||||
|
||||
// Buffer stdout and emit only complete lines
|
||||
stdoutBuffer += chunk;
|
||||
const lines = stdoutBuffer.split("\n");
|
||||
// Keep the last (potentially incomplete) line in the buffer
|
||||
stdoutBuffer = lines.pop() || "";
|
||||
|
||||
// Emit complete lines
|
||||
for (const line of lines) {
|
||||
if (line) {
|
||||
await onLog(stream, line + "\n");
|
||||
await emitCompacted(line, "\n");
|
||||
}
|
||||
}
|
||||
};
|
||||
@@ -760,7 +771,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
|
||||
// Flush any remaining buffer content
|
||||
if (stdoutBuffer) {
|
||||
await onLog("stdout", stdoutBuffer);
|
||||
await emitCompacted(stdoutBuffer, "");
|
||||
}
|
||||
|
||||
// Display output is capped by the process transport. Keep accounting
|
||||
|
||||
@@ -0,0 +1,359 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { parsePiJsonl } from "./parse.js";
|
||||
import {
|
||||
createPiStdoutCompactor,
|
||||
resolvePiStdoutLogMode,
|
||||
type PiStdoutLogMode,
|
||||
} from "./stdout-compaction.js";
|
||||
import { parsePiStdoutLine, resetParserState } from "../ui/parse-stdout.js";
|
||||
|
||||
function messageUpdate(delta: string, cumulativeText: string) {
|
||||
return JSON.stringify({
|
||||
type: "message_update",
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [
|
||||
{ type: "thinking", thinking: "x".repeat(5000) },
|
||||
{ type: "text", text: cumulativeText },
|
||||
],
|
||||
},
|
||||
assistantMessageEvent: {
|
||||
type: "text_delta",
|
||||
contentIndex: 1,
|
||||
delta,
|
||||
partial: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: cumulativeText }],
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function toolExecutionUpdate(toolCallId: string, partial: string) {
|
||||
return JSON.stringify({
|
||||
type: "tool_execution_update",
|
||||
toolCallId,
|
||||
toolName: "bash",
|
||||
args: { command: "ls" },
|
||||
partialResult: {
|
||||
content: [{ type: "text", text: partial }],
|
||||
details: { truncation: null, fullOutputPath: null },
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function assistantUpdate(message: Record<string, unknown>, assistantMessageEvent: Record<string, unknown>) {
|
||||
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([]);
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown>): 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<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function parseLineEvent(line: string): Record<string, unknown> | 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);
|
||||
}
|
||||
}
|
||||
@@ -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 [];
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user