mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-14 13:22:37 +00:00
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.
This commit is contained in:
+48
@@ -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));
|
||||
@@ -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);
|
||||
},
|
||||
);
|
||||
|
||||
@@ -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<string, string>();
|
||||
@@ -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 });
|
||||
|
||||
Reference in New Issue
Block a user