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 <noreply@anthropic.com>
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
authored and GitHub committed 2026-10-15 01:02:28 -07:00
1 parent af4191f45e
commit fd057f6ac4
6 files changed
+870 -41

No files matched your search

+13
View File
@@ -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.
@@ -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<Uint8Array>({
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<Uint8Array>;
let streamOpened!: () => void;
const opened = new Promise<void>(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<Uint8Array>({ 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 () => {
@@ -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<string, string>();
readonly #workspaceChangesByTurn = new Map<string, Record<string, unknown>>();
#reportedSseCompaction = false;
readonly #emittedFileReferences = new Set<string>();
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<number> {
return port;
}
async function* parseSseFrames(
stream: ReadableStream<Uint8Array>,
): AsyncIterable<{ raw: string; data: string }> {
const decoder = new TextDecoder();
let buffer = "";
for await (const chunk of stream as unknown as AsyncIterable<Uint8Array>) {
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<Uint8Array>,
): AsyncIterable<unknown> {
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(
@@ -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<Uint8Array> {
const bytes = new TextEncoder().encode(text);
return new ReadableStream<Uint8Array>({
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<OpenCodeSseFrame[]> {
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);
});
});
@@ -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<string | number>;
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<Uint8Array>,
): AsyncIterable<OpenCodeSseFrame> {
const decoder = new TextDecoder();
const parser = new OpenCodeSseParser();
for await (const chunk of stream as unknown as AsyncIterable<Uint8Array>)
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<string | number, unknown>)[segment]
: undefined;
}
if (typeof owner !== "object" || owner === null || Array.isArray(owner))
continue;
const target = owner as Record<PropertyKey, unknown>;
target[OMITTED_PROVIDER_TEXT] = {
...(target[OMITTED_PROVIDER_TEXT] as Record<string, OmittedProviderText> | 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<string, string> = {
'"': '"',
"\\": "\\",
"/": "/",
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;
}
@@ -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<string, unknown>,
key: string,
): OmittedProviderText | null {
const marker = (owner as Record<PropertyKey, unknown>)[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<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
? (value as Record<string, unknown>)
@@ -430,6 +452,7 @@ function digest(value: string): string {
function boundedOutput(
value: string,
omitted: OmittedProviderText | null = null,
): Pick<
Record<string, unknown>,
"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, unknown>, 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"