From ae8114932e3d4efd692d0da681ded77f3d4df467 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Tue, 6 Oct 2026 23:14:10 +0200 Subject: [PATCH] server: keep streamed OpenCode v2 chunks from holding their whole message Each streamed text or reasoning chunk was cut from the accumulated message with slice, which can share the source string's storage. The timeline store keeps every chunk, so a 128 KB reasoning part retained about 128 MB: every chunk held the full message as it stood when the chunk arrived. Long OpenCode v2 reasoning streams exhausted the daemon worker's heap. The timeline now emits an independent copy of each suffix, so retained memory follows the length of the text. --- .../test-utils/v2-streamed-text-memory.ts | 48 +++++++++++++++++++ .../providers/opencode/v2/timeline.test.ts | 37 ++++++++++++++ .../agent/providers/opencode/v2/timeline.ts | 11 ++++- 3 files changed, 94 insertions(+), 2 deletions(-) create mode 100644 packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts b/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts new file mode 100644 index 0000000000..d9d254d9de --- /dev/null +++ b/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts @@ -0,0 +1,48 @@ +import { TimelineProjection } from "../../../timeline-projection.js"; +import { V2Timeline } from "../v2/timeline.js"; + +// Streams one reasoning part through the timeline into the daemon's projection and +// prints the bytes still held once the part is complete. +const mode = process.argv[2]; +const chunks = Number(process.argv[3]); +if (mode !== "delta" && mode !== "snapshot") throw new Error("Mode must be delta or snapshot"); +if (!global.gc) throw new Error("Run with --expose-gc"); + +global.gc(); +const before = process.memoryUsage().heapUsed; +const timeline = new V2Timeline(); +const projection = new TimelineProjection(); +const part = { assistantMessageID: "answer", type: "reasoning" as const, ordinal: 0 }; +timeline.startPart(part); +let text = ""; +for (let index = 0; index < chunks; index++) { + const delta = `chunk-${index}:`.padEnd(64, String.fromCharCode(65 + (index % 26))); + text += delta; + const event = + mode === "delta" + ? timeline.delta({ ...part, delta }) + : timeline.messages([ + { + id: "answer", + type: "assistant", + agent: "build", + model: { providerID: "test", id: "model" }, + time: { created: 2 }, + content: [{ type: "reasoning", text }], + }, + ])[0]; + if (event?.type !== "timeline") throw new Error("Expected a timeline event"); + projection.append({ + seq: index + 1, + timestamp: "2026-10-01T00:00:00.000Z", + turnId: "turn", + item: event.item, + }); +} +global.gc(); +const retainedBytes = process.memoryUsage().heapUsed - before; +const [row] = projection.getRows(); +if (row?.item.type !== "reasoning" || row.item.text !== text) { + throw new Error("Projection lost the streamed text"); +} +process.stdout.write(String(retainedBytes)); diff --git a/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts b/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts index 1bbd86490a..3a365b6f00 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts @@ -1,4 +1,6 @@ import type { SessionMessageAssistant } from "@opencode/client"; +import { execFileSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; import { expect, test } from "vitest"; import { V2Timeline } from "./timeline.js"; @@ -54,3 +56,38 @@ test("a completed edit carries the replaced and replacement text", () => { }, ]); }); + +test("a surrogate pair split across deltas streams back intact", () => { + const timeline = new V2Timeline(); + const part = { assistantMessageID: "answer", type: "text" as const, ordinal: 0 }; + timeline.startPart(part); + const chunks = ["a\uD83D", "\uDE00b"].map((delta) => { + const event = timeline.delta({ ...part, delta }); + return event?.type === "timeline" && event.item.type === "assistant_message" + ? event.item.text + : null; + }); + expect(chunks).toEqual(["a\uD83D", "\uDE00b"]); + expect(chunks.join("")).toBe("a😀b"); +}); + +function retainedBytesAfterStreaming(mode: "delta" | "snapshot", chunks: number): number { + const fixture = fileURLToPath( + new URL("../test-utils/v2-streamed-text-memory.ts", import.meta.url), + ); + const output = execFileSync( + process.execPath, + ["--expose-gc", "--import", "tsx", fixture, mode, String(chunks)], + { encoding: "utf8" }, + ); + return Number(output); +} + +test.each(["delta", "snapshot"] as const)( + "a long %s-streamed part holds memory in proportion to its text", + (mode) => { + const chunks = 2000; + const textBytes = chunks * 64; + expect(retainedBytesAfterStreaming(mode, chunks)).toBeLessThan(textBytes * 32); + }, +); diff --git a/packages/server/src/server/agent/providers/opencode/v2/timeline.ts b/packages/server/src/server/agent/providers/opencode/v2/timeline.ts index 3bc0a5bf76..1301060c97 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/timeline.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/timeline.ts @@ -8,6 +8,13 @@ function textPartKey(messageID: string, part: "text" | "reasoning", ordinal: num return `${messageID}:${part}:${ordinal}`; } +// A `slice` can share its source's storage, so every streamed chunk kept by the timeline +// store would hold the whole text it was cut from. Copying the UTF-16 code units makes +// the chunk independent; UTF-8 would corrupt a surrogate pair split across chunks. +function detachedSuffix(text: string, start: number): string { + return Buffer.from(text.slice(start), "utf16le").toString("utf16le"); +} + // Text and reasoning each number parts from zero, independent of snapshot position. export class V2Timeline { private readonly content = new Map(); @@ -36,7 +43,7 @@ export class V2Timeline { if (!text.startsWith(emitted)) throw new Error("OpenCode changed previously emitted message content"); this.content.set(key, text); - const suffix = text.slice(emitted.length); + const suffix = detachedSuffix(text, emitted.length); const item: AgentTimelineItem = event.type === "text" ? { type: "assistant_message", text: suffix, messageId: event.assistantMessageID } @@ -139,7 +146,7 @@ export class V2Timeline { if (!part.text.startsWith(previous)) throw new Error("OpenCode changed previously emitted message content"); this.content.set(key, part.text); - const suffix = part.text.slice(previous.length); + const suffix = detachedSuffix(part.text, previous.length); if (part.type === "text") push({ type: "assistant_message", text: suffix, messageId: messageID }); else push({ type: "reasoning", text: suffix });