From fd057f6ac4e9148bb4055883ae1d0e7f49242692 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Thu, 15 Oct 2026 01:02:28 -0700 Subject: [PATCH] fix(runner): keep OpenCode turns alive through oversized SSE events Retain bounded OpenCode SSE events without restarting the event stream. Preserve canonical tool identity, status, output tails, original UTF-8 sizes, and truthful truncation markers. Bound id-less event deduplication with hashes. Cover fragmented and complete large frames, exact-limit split separators, bulk input retention pressure, truncation-record exhaustion, and unpaired surrogate byte counts. Full current-head CI and Apex 5/5 passed. The unrelated local post-tool answer assertion was reproduced on unchanged baseline source. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: Paperclip --- packages/paperclip-runner/README.md | 13 + .../opencode/opencode-server-driver.test.ts | 57 ++- .../opencode/opencode-server-driver.ts | 81 +-- .../src/drivers/opencode/sse-frames.test.ts | 240 +++++++++ .../src/drivers/opencode/sse-frames.ts | 473 ++++++++++++++++++ .../paperclip-runner/src/provider-events.ts | 47 +- 6 files changed, 870 insertions(+), 41 deletions(-) create mode 100644 packages/paperclip-runner/src/drivers/opencode/sse-frames.test.ts create mode 100644 packages/paperclip-runner/src/drivers/opencode/sse-frames.ts diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 4171bcd5842..3cfbb0de740 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -732,3 +732,16 @@ canonical event retained for that command. The header marker `🖇️ v0.1.2` identifies the current console iteration. `create_task` accepts an optional initial `status` of `backlog` or `todo`. Use `backlog` when the user wants a saved task or plan without execution: assignment and the initial plan are committed without scheduling a wake, even when dependencies are already complete. Omitting status preserves immediate delegation (`todo`, or `blocked` for unresolved dependencies). + +### Oversized OpenCode events + +The OpenCode driver keeps SSE frames up to 1,048,576 characters exact. It compacts longer +frames while retaining tool identity, status, and bounded output excerpts. This +also applies when a complete large frame arrives inside one network chunk. +Canonical tool rows mark shortened output, report its original byte count, and +do not claim a digest for text that was not retained. A structure that cannot +fit the compacted bound produces a diagnostic; the event stream continues. + +Tool output and error fields have reserved retention space and size records, +so large input values cannot consume their output tails or make excerpts look +complete. A separator split between reads does not count against the frame limit. diff --git a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts index 75032bd82c7..6e9c9b84597 100644 --- a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts @@ -2438,7 +2438,7 @@ describe("OpenCodeServerDriver", () => { } }); - it("rejects malformed and oversized SSE frames", async () => { + it("rejects malformed SSE frames", async () => { const stream = (value: string) => new ReadableStream({ start(controller) { @@ -2454,9 +2454,58 @@ describe("OpenCodeServerDriver", () => { } }; await expect(consume("data: {bad json}\n\n")).rejects.toThrow(); - await expect(consume(`data: ${"x".repeat(1_048_577)}`)).rejects.toThrow( - "payload limit", - ); + }); + + it("keeps the turn and the tool outcome when OpenCode sends an event above the payload limit", async () => { + await chmod(fixture, 0o755); + const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-oversized-sse-")); + const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-oversized-sse-workspace-")); + roots.push(root, workspace); + let stream!: ReadableStreamDefaultController; + let streamOpened!: () => void; + const opened = new Promise(resolve => { streamOpened = resolve; }); + const driver = new OpenCodeServerDriver({ + model: "openrouter/deepseek/deepseek-v4-flash-0731", runtimeDirectory: root, command: fixture, + environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" }, + fetch: async (input, init) => String(input).endsWith("/event") + ? new Response(new ReadableStream({ start(controller) { stream = controller; streamOpened(); } }), { headers: { "Content-Type": "text/event-stream" } }) + : fetch(input, init), + }); + const session = await driver.openSession({ runId: "oversized-sse", normalizedSessionId: "oversized-sse", workingDirectory: workspace }); + try { + const sessionID = session.ids().providerSessionId; + const events = collectTurnEvents(session.events()); + await session.startTurn({ message: { role: "user", text: "run the full suite" } }); + await opened; + const tail = "\nTests: 1 failed, 812 passed\nexit status 1"; + const output = "test output line\n".repeat(150_000) + tail; + const send = (value: unknown) => { + const bytes = new TextEncoder().encode(`data: ${JSON.stringify(value)}\n\n`); + for (let index = 0; index < bytes.length; index += 64 * 1024) + stream.enqueue(bytes.subarray(index, index + 64 * 1024)); + }; + send({ type: "message.updated", properties: { info: { id: "msg_oversized", role: "assistant", sessionID } } }); + send({ type: "message.part.updated", properties: { part: { + id: "prt_oversized", messageID: "msg_oversized", sessionID, type: "tool", callID: "call_oversized", tool: "bash", + state: { status: "completed", input: { command: "pnpm test" }, output, title: "pnpm test", metadata: { exit: 1, output }, time: { start: 1, end: 2 } }, + } } }); + send({ type: "session.idle", properties: { sessionID } }); + const observed = await events; + expect(observed.at(-1)?.eventType).toBe("turn.completed"); + const diagnostics = observed.filter(event => event.eventType === "harness.diagnostic"); + expect(diagnostics.map(event => event.payload.code)).toEqual(["opencode_sse_event_compacted"]); + const tool = observed.find(event => event.eventType === "tool.execution.completed" && event.payload.executionId === "prt_oversized"); + expect(tool?.payload).toMatchObject({ + name: "bash", + status: "completed", + outputBytes: Buffer.byteLength(output), + outputTruncated: true, + outputDigest: null, + }); + expect(String(tool?.payload.output).endsWith(tail)).toBe(true); + } finally { + await session.close({ reason: "test complete" }); + } }); it("restarts OpenCode when session startup fails transiently", async () => { diff --git a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts index 4ce6a5b9dd9..489ed50a52a 100644 --- a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts +++ b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts @@ -1,5 +1,5 @@ import { spawn, type ChildProcess } from "node:child_process"; -import { randomBytes } from "node:crypto"; +import { createHash, randomBytes } from "node:crypto"; import { accessSync, constants, readFileSync, realpathSync, statSync } from "node:fs"; import { createRequire } from "node:module"; import { @@ -20,6 +20,7 @@ import { pipeline } from "node:stream/promises"; import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; import { parseOpenCodeReasoningMode, type OpenCodeReasoningMode } from "./reasoning-mode.js"; +import { annotateOmittedProviderText, parseOpenCodeSseFrames } from "./sse-frames.js"; import { CODEX_SKILLLESS_BASE_INSTRUCTIONS, @@ -494,6 +495,7 @@ class OpenCodeHarnessSession implements HarnessSession { }> = []; readonly #messageUsageFingerprints = new Map(); readonly #workspaceChangesByTurn = new Map>(); + #reportedSseCompaction = false; readonly #emittedFileReferences = new Set(); readonly #pendingRuntimeRequests = new Map< string, @@ -1359,8 +1361,31 @@ class OpenCodeHarnessSession implements HarnessSession { reason: "Opened the OpenCode server-sent event stream", }); } - for await (const frame of parseSseFrames(response.body)) { + for await (const frame of parseOpenCodeSseFrames(response.body)) { if (this.#closed) return; + const compaction = frame.compaction; + if (compaction?.omitted) { + // Even the compacted structure exceeded its bound. Report the + // frame instead of ending the stream or dropping it silently. + this.#emit("harness.diagnostic", { + code: "opencode_sse_event_omitted", + message: `OpenCode sent a ${compaction.frameChars}-character event that could not be retained within the payload bound.`, + providerEventType: compaction.eventType, + frameChars: compaction.frameChars, + }); + continue; + } + if (compaction && !this.#reportedSseCompaction) { + this.#reportedSseCompaction = true; + this.#emit("harness.diagnostic", { + code: "opencode_sse_event_compacted", + message: `OpenCode sent a ${compaction.frameChars}-character event. Long strings were shortened to keep the event within the payload bound.`, + providerEventType: compaction.eventType, + frameChars: compaction.frameChars, + truncatedStrings: compaction.truncatedStrings, + omittedChars: compaction.omittedChars, + }); + } const frameId = this.#runtime.trace?.frame({ direction: "provider_to_client", @@ -1371,22 +1396,26 @@ class OpenCodeHarnessSession implements HarnessSession { let event: unknown; try { event = JSON.parse(frame.data); + if (compaction) annotateOmittedProviderText(event, compaction); if (frameId) { this.#runtime.trace?.interpretation({ frameId, stage: "typescript_opencode_sse_parse", - ruleId: "opencode.sse.json", + ruleId: compaction ? "opencode.sse.compacted_json" : "opencode.sse.json", disposition: "mapped", fieldMappings: [ { inputPath: "$raw.data", outputPath: "$providerEvent", action: "normalized", - reason: - "Decoded the exact SSE data payload as an OpenCode event", + reason: compaction + ? "Decoded the compacted SSE data payload as an OpenCode event" + : "Decoded the exact SSE data payload as an OpenCode event", }, ], - reason: "OpenCode SSE frame parsed as JSON", + reason: compaction + ? `OpenCode SSE frame of ${compaction.frameChars} characters parsed as JSON after ${compaction.truncatedStrings} long strings were shortened (${compaction.omittedChars} characters omitted)` + : "OpenCode SSE frame parsed as JSON", }); } } catch (error) { @@ -1486,7 +1515,11 @@ class OpenCodeHarnessSession implements HarnessSession { const type = text(event.type); // Id-less deltas are additive: two identical chunks can be legitimate // adjacent tokens. Content deduplication would silently lose the second. - const eventId = text(event.id) || (type === "message.part.delta" ? null : canonicalJson(value)); + // Key id-less events by digest: a retained canonical copy of each recent + // event would hold up to 10,000 large tool payloads in memory. + const eventId = text(event.id) || (type === "message.part.delta" + ? null + : `sha256:${createHash("sha256").update(canonicalJson(value)).digest("hex")}`); if (eventId !== null && this.#seenProviderEvents.has(eventId)) return; if (eventId !== null) this.#seenProviderEvents.add(eventId); if (this.#seenProviderEvents.size > 10_000) @@ -2829,37 +2862,15 @@ async function reservePort(): Promise { return port; } -async function* parseSseFrames( - stream: ReadableStream, -): AsyncIterable<{ raw: string; data: string }> { - const decoder = new TextDecoder(); - let buffer = ""; - for await (const chunk of stream as unknown as AsyncIterable) { - buffer += decoder.decode(chunk, { stream: true }); - if (buffer.length > 1_048_576) - throw new Error("OpenCode SSE event exceeded the retained payload limit"); - let boundary: RegExpExecArray | null; - while ((boundary = /\r?\n\r?\n/.exec(buffer)) !== null) { - const rawFrame = buffer.slice(0, boundary.index); - const raw = buffer.slice(0, boundary.index + boundary[0].length); - const frame = rawFrame.replaceAll("\r", ""); - buffer = buffer.slice(boundary.index + boundary[0].length); - const data = frame - .split("\n") - .filter((line) => line.startsWith("data:")) - .map((line) => line.slice(5).trimStart()) - .join("\n"); - if (!data || data === "[DONE]") continue; - yield { raw, data }; - } - } -} - async function* parseSse( stream: ReadableStream, ): AsyncIterable { - for await (const frame of parseSseFrames(stream)) - yield JSON.parse(frame.data); + for await (const frame of parseOpenCodeSseFrames(stream)) { + if (frame.compaction?.omitted) continue; + const event = JSON.parse(frame.data) as unknown; + if (frame.compaction) annotateOmittedProviderText(event, frame.compaction); + yield event; + } } function sanitizedEnvironment( diff --git a/packages/paperclip-runner/src/drivers/opencode/sse-frames.test.ts b/packages/paperclip-runner/src/drivers/opencode/sse-frames.test.ts new file mode 100644 index 00000000000..6d1eaa758c3 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/opencode/sse-frames.test.ts @@ -0,0 +1,240 @@ +import { describe, expect, it } from "vitest"; + +import { canonicalProviderEventsFromOpenCodePart } from "../../provider-events.js"; +import { + OPENCODE_SSE_LIMITS, + annotateOmittedProviderText, + parseOpenCodeSseFrames, + type OpenCodeSseFrame, +} from "./sse-frames.js"; + +function stream(text: string, chunkChars = 64 * 1024): ReadableStream { + const bytes = new TextEncoder().encode(text); + return new ReadableStream({ + start(controller) { + for (let index = 0; index < bytes.length; index += chunkChars) + controller.enqueue(bytes.subarray(index, index + chunkChars)); + controller.close(); + }, + }); +} + +async function frames(text: string, chunkChars?: number): Promise { + const collected: OpenCodeSseFrame[] = []; + for await (const frame of parseOpenCodeSseFrames(stream(text, chunkChars))) + collected.push(frame); + return collected; +} + +function sse(value: unknown, newline = "\n"): string { + return `data: ${JSON.stringify(value)}${newline}${newline}`; +} + +function completedToolPart(output: string) { + return { + type: "message.part.updated", + properties: { + sessionID: "ses_1", + part: { + id: "prt_tool", + messageID: "msg_1", + sessionID: "ses_1", + type: "tool", + callID: "call_1", + tool: "bash", + state: { + status: "completed", + input: { command: "pnpm test", description: "Run tests" }, + output, + title: "Run tests", + metadata: { exit: 1, output, description: "Run tests" }, + time: { start: 1, end: 2 }, + }, + }, + }, + }; +} + +describe("parseOpenCodeSseFrames", () => { + it("keeps frames below the limit exact even when one chunk holds more than the limit", async () => { + const events = Array.from({ length: 40 }, (_, index) => ({ + type: "message.part.updated", + properties: { index, text: "x".repeat(32 * 1024) }, + })); + const text = events.map((event) => sse(event)).join(""); + expect(text.length).toBeGreaterThan(OPENCODE_SSE_LIMITS.frameChars); + const parsed = await frames(text, text.length); + expect(parsed).toHaveLength(40); + expect(parsed.every((frame) => frame.compaction === undefined)).toBe(true); + expect(parsed.map((frame) => JSON.parse(frame.data))).toEqual(events); + }); + + it("bounds a complete oversized event inside one chunk without changing its neighbors", async () => { + const before = { type: "server.connected", properties: {} }; + const after = { type: "session.idle", properties: { sessionID: "ses_1" } }; + const output = "begin\n" + "x".repeat(1_200_000) + "\nexit status 1"; + const text = sse(before) + sse(completedToolPart(output)) + sse(after); + const parsed = await frames(text, text.length); + expect(parsed).toHaveLength(3); + expect(JSON.parse(parsed[0]!.data)).toEqual(before); + expect(parsed[0]!.compaction).toBeUndefined(); + expect(parsed[1]!.compaction).toMatchObject({ omitted: false, truncatedStrings: 2 }); + expect(parsed[1]!.data.length).toBeLessThan(OPENCODE_SSE_LIMITS.frameChars); + const event = JSON.parse(parsed[1]!.data); + annotateOmittedProviderText(event, parsed[1]!.compaction!); + const [canonical] = canonicalProviderEventsFromOpenCodePart(event.properties.part); + expect(canonical?.payload).toMatchObject({ + status: "completed", outputBytes: Buffer.byteLength(output), outputTruncated: true, outputDigest: null, + }); + expect(String(canonical?.payload.output).endsWith("exit status 1")).toBe(true); + expect(JSON.parse(parsed[2]!.data)).toEqual(after); + expect(parsed[2]!.compaction).toBeUndefined(); + }); + + it("compacts an oversized tool event and keeps its outcome and the next event", async () => { + const head = "PASS early suite\n"; + const tail = "\nFAIL src/last.test.ts\nTests: 1 failed, 812 passed\nexit status 1"; + const output = head + "é ✓ \"quoted\" \\ 🙂 line\n".repeat(80_000) + tail; + const next = { type: "session.idle", properties: { sessionID: "ses_1" } }; + const parsed = await frames(sse(completedToolPart(output), "\r\n") + sse(next), 4093); + + expect(parsed).toHaveLength(2); + const [oversized, following] = parsed; + expect(oversized!.compaction).toMatchObject({ + omitted: false, + eventType: "message.part.updated", + truncatedStrings: 2, + }); + expect(oversized!.compaction!.frameChars).toBeGreaterThan(2 * output.length); + expect(oversized!.data.length).toBeLessThan(OPENCODE_SSE_LIMITS.frameChars); + expect(oversized!.compaction!.truncations).toEqual([ + expect.objectContaining({ + path: ["properties", "part", "state", "output"], + chars: output.length, + bytes: Buffer.byteLength(output), + }), + expect.objectContaining({ + path: ["properties", "part", "state", "metadata", "output"], + chars: output.length, + bytes: Buffer.byteLength(output), + }), + ]); + + const event = JSON.parse(oversized!.data); + const part = event.properties.part; + expect(part).toMatchObject({ + id: "prt_tool", + messageID: "msg_1", + type: "tool", + callID: "call_1", + tool: "bash", + state: { + status: "completed", + input: { command: "pnpm test", description: "Run tests" }, + title: "Run tests", + metadata: { exit: 1, description: "Run tests" }, + time: { start: 1, end: 2 }, + }, + }); + expect(part.state.output.startsWith(head)).toBe(true); + expect(part.state.output.endsWith(tail)).toBe(true); + expect(part.state.output).toMatch(/…\[\d+ characters omitted\]…/); + expect(JSON.parse(following!.data)).toEqual(next); + expect(following!.compaction).toBeUndefined(); + + annotateOmittedProviderText(event, oversized!.compaction!); + const [canonical] = canonicalProviderEventsFromOpenCodePart(part); + expect(canonical).toMatchObject({ + eventType: "tool.execution.completed", + payload: { + name: "bash", + status: "completed", + outputBytes: Buffer.byteLength(output), + outputTruncated: true, + outputDigest: null, + }, + }); + expect(String(canonical!.payload.output).endsWith(tail)).toBe(true); + expect(String(canonical!.payload.output).length).toBeLessThanOrEqual(64 * 1024); + expect(JSON.stringify(part)).not.toContain("omittedProviderText"); + }); + + it("counts unpaired surrogate output as the original UTF-8 text", async () => { + const output = "\ud800" + "x".repeat(1_200_000) + "\udc00"; + const [frame] = await frames(sse(completedToolPart(output)), 4093); + expect(frame!.compaction!.truncations).toEqual([ + expect.objectContaining({ bytes: Buffer.byteLength(output) }), + expect.objectContaining({ bytes: Buffer.byteLength(output) }), + ]); + }); + + it.each([[12, 64 * 1024], [80, 100_000]])("prioritizes output after %i large input values", async (count, inputChars) => { + const output = "o".repeat(300_000) + "\nfinal error: command failed\nexit status 1"; + const event = completedToolPart(output); + Object.assign(event.properties.part.state.input, { + values: Array.from({ length: count }, () => "i".repeat(inputChars)), + }); + const [frame] = await frames(sse(event)); + const compacted = JSON.parse(frame!.data); + annotateOmittedProviderText(compacted, frame!.compaction!); + const [canonical] = canonicalProviderEventsFromOpenCodePart(compacted.properties.part); + expect(canonical?.payload).toMatchObject({ + outputBytes: Buffer.byteLength(output), outputTruncated: true, outputDigest: null, + }); + expect(String(canonical?.payload.output).endsWith("final error: command failed\nexit status 1")).toBe(true); + expect(frame!.compaction!.truncations.length).toBeLessThanOrEqual(OPENCODE_SSE_LIMITS.truncations); + }); + + it.each([["\n", 1], ["\r\n", 1], ["\r\n", 2], ["\r\n", 3]] as const)( + "keeps an exact-limit frame with separator %j split after %i bytes", async (newline, split) => { + const empty = `data: ${JSON.stringify({ type: "message.part.updated", properties: { text: "" } })}`; + const event = { type: "message.part.updated", properties: { text: "x".repeat(OPENCODE_SSE_LIMITS.frameChars - empty.length) } }; + const raw = `data: ${JSON.stringify(event)}`; + expect(raw.length).toBe(OPENCODE_SSE_LIMITS.frameChars); + const [frame] = await frames(raw + newline + newline, raw.length + split); + expect(frame!.compaction).toBeUndefined(); + expect(JSON.parse(frame!.data)).toEqual(event); + }, + ); + + it("does not report a separator blank line before an oversized frame", async () => { + const small = { type: "server.connected", properties: {} }; + const large = completedToolPart("y".repeat(1_200_000)); + const parsed = await frames(`${sse(small)}\n${sse(large)}`, 1_100_000); + expect(parsed.map((frame) => frame.compaction?.omitted ?? null)).toEqual([null, false]); + expect(JSON.parse(parsed[1]!.data).properties.part.state.status).toBe("completed"); + }); + + it("reports a frame whose compacted structure still exceeds the bound", async () => { + const value = { + type: "session.diff", + properties: { files: Array.from({ length: 500_000 }, (_, index) => index) }, + }; + const next = { type: "session.idle", properties: {} }; + const parsed = await frames(sse(value) + sse(next)); + expect(parsed).toHaveLength(2); + expect(parsed[0]).toMatchObject({ + data: "", + raw: "", + compaction: { omitted: true, eventType: "session.diff" }, + }); + expect(JSON.parse(parsed[1]!.data)).toEqual(next); + }); + + it("keeps short identity strings once the long-string budget is spent", async () => { + const files = Array.from({ length: 20 }, (_, index) => ({ + file: `src/file-${index}.ts`, + before: "b".repeat(100_000), + after: "a".repeat(100_000), + status: "modified", + })); + const [frame] = await frames(sse({ type: "session.diff", properties: { files } })); + const event = JSON.parse(frame!.data); + expect(event.properties.files).toHaveLength(20); + expect(event.properties.files.map((file: { file: string }) => file.file)).toEqual( + files.map((file) => file.file), + ); + expect(event.properties.files.every((file: { status: string }) => file.status === "modified")).toBe(true); + expect(frame!.data.length).toBeLessThan(OPENCODE_SSE_LIMITS.frameChars); + }); +}); diff --git a/packages/paperclip-runner/src/drivers/opencode/sse-frames.ts b/packages/paperclip-runner/src/drivers/opencode/sse-frames.ts new file mode 100644 index 00000000000..ab23fc3f580 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/opencode/sse-frames.ts @@ -0,0 +1,473 @@ +import { + OMITTED_PROVIDER_TEXT, + type OmittedProviderText, +} from "../../provider-events.js"; + +/** + * Bounds for the OpenCode `/event` stream. A frame within `frameChars` is + * retained exactly. A larger frame keeps its JSON structure and short strings, + * while long strings keep a head and a tail. A tool part therefore keeps its + * identity, status and output tail even when OpenCode attaches megabytes of + * command output or file contents to one event. + */ +export const OPENCODE_SSE_LIMITS = { + frameChars: 1_048_576, + shortStringChars: 256, + stringHeadChars: 4 * 1024, + stringTailChars: 64 * 1024, + stringBudgetChars: 768 * 1024, + compactedChars: 2 * 1_048_576, + truncations: 64, +} as const; + +export interface OpenCodeSseTruncation { + path: Array; + chars: number; + bytes: number; + omittedChars: number; +} + +export interface OpenCodeSseCompaction { + /** Characters of SSE data received for the oversized frame. */ + frameChars: number; + truncatedStrings: number; + omittedChars: number; + /** Shortened string values, in stream order. Bounded by `truncations`. */ + truncations: OpenCodeSseTruncation[]; + /** Top-level OpenCode event type, when the frame exposed one. */ + eventType: string | null; + /** + * The compacted structure still exceeded `compactedChars`, so no event + * could be retained. The consumer must report this frame, not parse it. + */ + omitted: boolean; +} + +export interface OpenCodeSseFrame { + raw: string; + data: string; + compaction?: OpenCodeSseCompaction; +} + +export async function* parseOpenCodeSseFrames( + stream: ReadableStream, +): AsyncIterable { + const decoder = new TextDecoder(); + const parser = new OpenCodeSseParser(); + for await (const chunk of stream as unknown as AsyncIterable) + yield* parser.push(decoder.decode(chunk, { stream: true })); +} + +/** + * Records the real size of each string that compaction shortened on the + * object that owns it, so canonical tool payloads report the provider's + * output size instead of the excerpt size. + */ +export function annotateOmittedProviderText( + event: unknown, + compaction: OpenCodeSseCompaction, +): void { + for (const truncation of compaction.truncations) { + const key = truncation.path.at(-1); + if (typeof key !== "string") continue; + let owner: unknown = event; + for (const segment of truncation.path.slice(0, -1)) { + owner = + typeof owner === "object" && owner !== null + ? (owner as Record)[segment] + : undefined; + } + if (typeof owner !== "object" || owner === null || Array.isArray(owner)) + continue; + const target = owner as Record; + target[OMITTED_PROVIDER_TEXT] = { + ...(target[OMITTED_PROVIDER_TEXT] as Record | undefined), + [key]: { chars: truncation.chars, bytes: truncation.bytes }, + }; + } +} + +class OpenCodeSseParser { + #buffer = ""; + #oversized: OversizedSseFrame | null = null; + + push(text: string): OpenCodeSseFrame[] { + const frames: OpenCodeSseFrame[] = []; + let rest = text; + while (rest) { + if (this.#oversized) { + const consumed = this.#oversized.write(rest); + if (consumed < 0) return frames; + frames.push(this.#oversized.finish()); + this.#oversized = null; + rest = rest.slice(consumed); + continue; + } + this.#buffer += rest; + rest = ""; + let boundary: RegExpExecArray | null; + while ((boundary = /\r?\n\r?\n/.exec(this.#buffer)) !== null) { + const rawFrame = this.#buffer.slice(0, boundary.index); + const raw = this.#buffer.slice(0, boundary.index + boundary[0].length); + this.#buffer = this.#buffer.slice(boundary.index + boundary[0].length); + if (rawFrame.length > OPENCODE_SSE_LIMITS.frameChars) { + const oversized = new OversizedSseFrame(); + oversized.write(raw); + frames.push(oversized.finish()); + continue; + } + const data = rawFrame + .replaceAll("\r", "") + .split("\n") + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice(5).trimStart()) + .join("\n"); + if (!data || data === "[DONE]") continue; + frames.push({ raw, data }); + } + // Only the unterminated frame remains. Stream it through the compactor + // instead of retaining it or abandoning the whole event stream. + const separatorChars = /(?:\r\n\r|\r\n|[\r\n])$/.exec(this.#buffer)?.[0].length ?? 0; + if (this.#buffer.length - separatorChars > OPENCODE_SSE_LIMITS.frameChars) { + this.#oversized = new OversizedSseFrame(); + rest = this.#buffer; + this.#buffer = ""; + } + } + return frames; + } +} + +/** Streams the `data:` lines of one oversized SSE frame into the compactor. */ +class OversizedSseFrame { + readonly #json = new BoundedJsonCompactor(); + #lineStarted = false; + #frameStarted = false; + #lineKind: "pending" | "data" | "other" = "pending"; + #lineHead = ""; + #dataLines = 0; + + /** Returns the index after the frame's blank line, or -1 if it continues. */ + write(text: string): number { + let index = 0; + while (index < text.length) { + const newline = text.indexOf("\n", index); + const end = newline < 0 ? text.length : newline; + if (end > index) this.#segment(text.slice(index, end).replaceAll("\r", "")); + if (newline < 0) return -1; + index = newline + 1; + // A blank line ends the frame. Blank lines before its first field are + // separator leftovers from the previous frame. + if (!this.#lineStarted) { + if (this.#frameStarted) return index; + continue; + } + this.#lineStarted = false; + this.#lineKind = "pending"; + this.#lineHead = ""; + } + return -1; + } + + finish(): OpenCodeSseFrame { + const { data, compaction } = this.#json.finish(); + return { raw: data ? `data: ${data}\n\n` : "", data, compaction }; + } + + #segment(segment: string): void { + if (!segment) return; + this.#lineStarted = true; + this.#frameStarted = true; + let value = segment; + if (this.#lineKind === "pending") { + const head = this.#lineHead + value; + if (head.length < 5) { + if ("data:".startsWith(head)) this.#lineHead = head; + else this.#lineKind = "other"; + return; + } + if (!head.startsWith("data:")) { + this.#lineKind = "other"; + return; + } + this.#lineKind = "data"; + this.#lineHead = ""; + if (this.#dataLines++ > 0) this.#json.write("\n"); + value = head.slice(5); + } + if (this.#lineKind === "data") this.#json.write(value); + } +} + +const TOOL_OUTCOME_TEXT_PATHS = new Set([ + ["properties", "part", "state", "output"], + ["properties", "part", "state", "error"], + ["properties", "part", "output"], +].map(path => JSON.stringify(path))); + +interface JsonContainer { + array: boolean; + key: string | number | null; + index: number; + expectKey: boolean; +} + +/** + * Re-emits a JSON document with long strings shortened. Structure, numbers, + * literals and short strings are copied, so a valid input yields a valid + * output. Memory is bounded by the retained output plus one string's head and + * tail. + */ +class BoundedJsonCompactor { + #out: string[] = []; + #outChars = 0; + #stack: JsonContainer[] = []; + #pendingKey: string | null = null; + #string: JsonStringReader | null = null; + #failed = false; + #inputChars = 0; + #budget: number = OPENCODE_SSE_LIMITS.stringBudgetChars - TOOL_OUTCOME_TEXT_PATHS.size * + (OPENCODE_SSE_LIMITS.stringHeadChars + OPENCODE_SSE_LIMITS.stringTailChars); + #truncatedStrings = 0; + #omittedChars = 0; + #truncations: OpenCodeSseTruncation[] = []; + #eventType: string | null = null; + + write(text: string): void { + this.#inputChars += text.length; + let index = 0; + while (index < text.length && !this.#failed) { + if (this.#string) { + index = this.#string.consume(text, index); + if (this.#string.failed) this.#fail(); + else if (this.#string.done) this.#endString(this.#string); + continue; + } + const char = text[index++]!; + const top = this.#stack.at(-1); + switch (char) { + case '"': + this.#string = new JsonStringReader( + Boolean(top && !top.array && top.expectKey), + ); + break; + case "{": + case "[": + this.#stack.push({ + array: char === "[", + key: this.#position(), + index: 0, + expectKey: char === "{", + }); + this.#emit(char); + break; + case "}": + case "]": + this.#stack.pop(); + this.#emit(char); + break; + case ":": + if (top) top.expectKey = false; + this.#emit(char); + break; + case ",": + if (top?.array) top.index += 1; + else if (top) top.expectKey = true; + this.#emit(char); + break; + case " ": + case "\t": + case "\n": + case "\r": + break; + default: + this.#emit(char); + } + } + } + + finish(): { data: string; compaction: OpenCodeSseCompaction } { + const complete = + !this.#failed && !this.#string && this.#stack.length === 0 && this.#outChars > 0; + return { + data: complete ? this.#out.join("") : "", + compaction: { + frameChars: this.#inputChars, + truncatedStrings: this.#truncatedStrings, + omittedChars: this.#omittedChars, + truncations: this.#truncations, + eventType: this.#eventType, + omitted: !complete, + }, + }; + } + + #position(): string | number | null { + const top = this.#stack.at(-1); + if (!top) return null; + return top.array ? top.index : this.#pendingKey; + } + + #endString(reader: JsonStringReader): void { + this.#string = null; + const limits = OPENCODE_SSE_LIMITS; + const path = [ + ...this.#stack.slice(1).map((container) => container.key ?? ""), + this.#position() ?? "", + ]; + const pathKey = JSON.stringify(path); + const outcomeText = !reader.isKey && TOOL_OUTCOME_TEXT_PATHS.has(pathKey); + const budget = outcomeText ? limits.stringHeadChars + limits.stringTailChars : this.#budget; + let value: string; + if ( + !reader.split && + (reader.chars <= limits.shortStringChars || reader.chars <= budget) + ) { + value = reader.full; + if (!outcomeText && reader.chars > limits.shortStringChars) this.#budget -= reader.chars; + } else { + // The tail carries a command's exit status and final errors, so it has + // priority over the head when the event budget is nearly spent. + const head = reader.split ? reader.head : reader.full; + const tail = reader.split ? reader.tail : reader.full; + const tailKeep = Math.min(limits.stringTailChars, budget, reader.chars); + const headKeep = Math.min( + limits.stringHeadChars, + budget - tailKeep, + reader.chars - tailKeep, + ); + const omittedChars = reader.chars - headKeep - tailKeep; + value = `${head.slice(0, headKeep)}…[${omittedChars} characters omitted]…${tail.slice(tail.length - tailKeep)}`; + if (!outcomeText) this.#budget -= headKeep + tailKeep; + this.#truncatedStrings += 1; + this.#omittedChars += omittedChars; + if (!reader.isKey) { + if (outcomeText) { + const existing = this.#truncations.findIndex(entry => JSON.stringify(entry.path) === pathKey); + if (existing >= 0) this.#truncations.splice(existing, 1); + if (this.#truncations.length >= limits.truncations) { + const expendable = this.#truncations.findIndex(entry => !TOOL_OUTCOME_TEXT_PATHS.has(JSON.stringify(entry.path))); + if (expendable >= 0) this.#truncations.splice(expendable, 1); + } + } + if (this.#truncations.length < limits.truncations) { + this.#truncations.push({ path, chars: reader.chars, bytes: reader.bytes, omittedChars }); + } + } + } + if (reader.isKey) this.#pendingKey = value; + else if (this.#stack.length === 1 && this.#pendingKey === "type") + this.#eventType = value.slice(0, 160); + this.#emit(JSON.stringify(value)); + } + + #emit(value: string): void { + this.#outChars += value.length; + if (this.#outChars > OPENCODE_SSE_LIMITS.compactedChars) { + this.#fail(); + return; + } + this.#out.push(value); + } + + #fail(): void { + this.#failed = true; + this.#out = []; + this.#string = null; + } +} + +/** Decodes one JSON string incrementally and keeps a bounded head and tail. */ +class JsonStringReader { + readonly isKey: boolean; + full = ""; + head = ""; + tail = ""; + chars = 0; + bytes = 0; + split = false; + done = false; + failed = false; + #escape = ""; + #pendingHighSurrogate = ""; + + constructor(isKey: boolean) { + this.isKey = isKey; + } + + /** Consumes string content from `index`; returns the next unread index. */ + consume(text: string, start: number): number { + let index = start; + const special = /["\\]/g; + while (index < text.length) { + if (this.#escape) { + this.#escape += text[index++]; + const length = this.#escape[1] === "u" ? 6 : 2; + if (this.#escape.length < length) continue; + const decoded = decodeJsonEscape(this.#escape); + this.#escape = ""; + if (decoded === null) { + this.failed = true; + return text.length; + } + this.#append(decoded); + continue; + } + special.lastIndex = index; + const match = special.exec(text); + const stop = match ? match.index : text.length; + if (stop > index) this.#append(text.slice(index, stop)); + if (!match) return text.length; + index = stop + 1; + if (match[0] === '"') { + this.bytes += Buffer.byteLength(this.#pendingHighSurrogate); + this.#pendingHighSurrogate = ""; + this.done = true; + return index; + } + this.#escape = "\\"; + } + return index; + } + + #append(value: string): void { + const { stringHeadChars, stringTailChars } = OPENCODE_SSE_LIMITS; + this.chars += value.length; + const pending = this.#pendingHighSurrogate + value; + const lastCode = pending.charCodeAt(pending.length - 1); + const endsHigh = lastCode >= 0xd800 && lastCode <= 0xdbff; + this.#pendingHighSurrogate = endsHigh ? pending.slice(-1) : ""; + this.bytes += Buffer.byteLength(endsHigh ? pending.slice(0, -1) : pending); + if (!this.split) { + this.full += value; + if (this.full.length <= stringHeadChars + stringTailChars) return; + this.head = this.full.slice(0, stringHeadChars); + this.tail = this.full.slice(stringHeadChars); + this.full = ""; + this.split = true; + } else { + this.tail += value; + } + if (this.tail.length > 2 * stringTailChars) + this.tail = this.tail.slice(-stringTailChars); + } +} + +const SIMPLE_ESCAPES: Record = { + '"': '"', + "\\": "\\", + "/": "/", + b: "\b", + f: "\f", + n: "\n", + r: "\r", + t: "\t", +}; + +function decodeJsonEscape(sequence: string): string | null { + if (sequence[1] !== "u") return SIMPLE_ESCAPES[sequence[1]!] ?? null; + const hex = sequence.slice(2); + return /^[0-9a-fA-F]{4}$/.test(hex) + ? String.fromCharCode(Number.parseInt(hex, 16)) + : null; +} diff --git a/packages/paperclip-runner/src/provider-events.ts b/packages/paperclip-runner/src/provider-events.ts index e163706c53d..ef0c0d26ffe 100644 --- a/packages/paperclip-runner/src/provider-events.ts +++ b/packages/paperclip-runner/src/provider-events.ts @@ -119,6 +119,28 @@ export interface CanonicalProviderEvent { const MAX_OUTPUT = 64 * 1024; +/** + * Set by a bounded provider transport on the object that owns a string it + * shortened before parsing, keyed by property name. JSON serialization skips + * symbol keys, so the marker never reaches persisted payloads. + */ +export const OMITTED_PROVIDER_TEXT = Symbol("paperclip.omittedProviderText"); +export interface OmittedProviderText { + chars: number; + bytes: number; +} + +function omittedProviderText( + owner: Record, + key: string, +): OmittedProviderText | null { + const marker = (owner as Record)[OMITTED_PROVIDER_TEXT]; + const entry = record(record(marker)[key]); + return Number.isSafeInteger(entry.chars) && Number.isSafeInteger(entry.bytes) + ? { chars: Number(entry.chars), bytes: Number(entry.bytes) } + : null; +} + function record(value: unknown): Record { return typeof value === "object" && value !== null && !Array.isArray(value) ? (value as Record) @@ -430,6 +452,7 @@ function digest(value: string): string { function boundedOutput( value: string, + omitted: OmittedProviderText | null = null, ): Pick< Record, "output" | "outputBytes" | "outputTruncated" | "outputDigest" @@ -438,6 +461,15 @@ function boundedOutput( const output = Buffer.from(value) .subarray(Math.max(0, bytes - MAX_OUTPUT)) .toString("utf8"); + // The transport retained only an excerpt. Report the provider's real size; + // a digest of the excerpt would misidentify the output, so none is claimed. + if (omitted) + return { + output, + outputBytes: omitted.bytes, + outputTruncated: true, + outputDigest: null, + }; return { output, outputBytes: bytes, @@ -986,7 +1018,15 @@ export function canonicalProviderEventsFromOpenCodePart( ); // OpenCode places terminal tool failures in `state.error`, not `output`. // Preserve the bounded provider message so a failed row is actionable. - const output = text(state.output, text(part.output, text(state.error))); + const outputOwner: [Record, string] | null = + typeof state.output === "string" + ? [state, "output"] + : typeof part.output === "string" + ? [part, "output"] + : typeof state.error === "string" + ? [state, "error"] + : null; + const output = outputOwner ? text(outputOwner[0][outputOwner[1]]) : ""; const payload = { schema: "paperclip.tool.execution.v1", executionId: itemId, @@ -1010,7 +1050,10 @@ export function canonicalProviderEventsFromOpenCodePart( exitCode: integer(state.exit), progress: status === "running" ? text(state.title).slice(0, 4000) || null : null, - ...boundedOutput(output), + ...boundedOutput( + output, + outputOwner ? omittedProviderText(...outputOwner) : null, + ), }; const eventType = nativeStatus === "pending"