mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-15 05:42:34 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ae8114932e |
+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