From 5e5b05abd96d932f70d2bd97129cbf1965ecef74 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Fri, 3 Jul 2026 00:03:50 +0800 Subject: [PATCH 1/3] =?UTF-8?q?fix:=20streamed=20messages=20pushed=20into?= =?UTF-8?q?=20scrollback=20twice=20=E2=80=94=20commit=20finalize=20via=20u?= =?UTF-8?q?psert,=20not=20a=20second=20push?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The #50 fix covered only #artificiallyStreamText. The main streaming path still called #persistAndBuffer at BOTH stream start (text_delta creates the message) and finalize (text_done), and the same double-push lived in #flushActiveAssistant (interrupt/turn-boundary flush) and #finalizeActiveThinking (every thinking block). Consequences: - scrollback.replay carried two entries per streamed messageId; clients rendered the message twice and the web virtualizer's messageId-keyed caches collided (the residual cause of the 'intermittent message overlap' that #73 partially fixed) - the memory chunker received the stream-start push with empty content, emitting a prompt-only user_turn episode and then a promptless assistant_turn — every plain turn fragmented into two half-episodes - byte accounting drifted negative (push #1 accounted the empty size, eviction subtracted the grown size twice), permanently disabling the 20MB scrollback cap Fixes: - ScrollbackBuffer now records the accounted size per entry and upserts by messageId: re-pushing a buffered id re-accounts the existing entry in place (keeping its replay position) instead of appending a duplicate. Eviction subtracts exactly what was added — negative drift is structurally impossible. updateMessage is O(1) via the id index (was a front-to-back scan). - Session stream-start sites push to scrollback only; the new #commitStreamed emits the durable transcript row and the chunker event exactly once, at finalize, with final content. #artificiallyStreamText drops its bespoke reset-and-updateMessage dance for the same helper. - #seq now seeds past the persisted transcript tail on resume instead of restarting at 0, making seq usable as a monotonic replay cursor. Tests: session-stream-commit.test.ts pins one-scrollback-entry-per- messageId across all four finalize paths (text_done, thinking_done, batch-reply artificial streaming, turn-boundary flush), buffer upsert + byte-cap accounting under by-reference growth, chunker episode pairing (including a test documenting the pre-fix fragmentation), and seq continuation after resume. Co-Authored-By: Claude Fable 5 --- src/daemon/scrollback.ts | 84 +++-- src/daemon/session-manager.ts | 6 +- src/daemon/session.ts | 72 ++-- src/tests/session-stream-commit.test.ts | 446 ++++++++++++++++++++++++ 4 files changed, 558 insertions(+), 50 deletions(-) create mode 100644 src/tests/session-stream-commit.test.ts diff --git a/src/daemon/scrollback.ts b/src/daemon/scrollback.ts index f31e224..cecfce6 100644 --- a/src/daemon/scrollback.ts +++ b/src/daemon/scrollback.ts @@ -24,8 +24,27 @@ const DEFAULT_CONFIG: ScrollbackConfig = { maxBytes: 20 * 1024 * 1024, // 20MB }; +/** + * Internal wrapper that records the byte size that was actually accounted + * into `#bytes` for this entry. Streamed messages are held by reference and + * grow in place between push and finalize; eviction must subtract exactly + * what was added, never the current (grown) serialized size — otherwise the + * counter drifts negative and the byte cap stops evicting. + */ +interface Entry { + msg: DaemonMessage; + size: number; +} + +function messageIdOf(msg: DaemonMessage): string | undefined { + return msg.type === "session.message" + ? (msg as { messageId?: string }).messageId + : undefined; +} + export class ScrollbackBuffer { - #entries: DaemonMessage[] = []; + #entries: Entry[] = []; + #byId = new Map(); #bytes = 0; #config: ScrollbackConfig; @@ -35,10 +54,30 @@ export class ScrollbackBuffer { /** * Push a message into the buffer. Evicts oldest entries if limits are exceeded. + * + * Upserts by messageId: pushing a message whose messageId is already + * buffered re-accounts the existing entry in place (keeping its position) + * instead of appending a second entry. Duplicate entries for one messageId + * corrupt scrollback.replay — clients render the message twice and + * virtualizers keyed on messageId collide (the #50 bug class). */ push(msg: DaemonMessage): void { - this.#entries.push(msg); - this.#bytes += JSON.stringify(msg).length; + const messageId = messageIdOf(msg); + if (messageId !== undefined) { + const existing = this.#byId.get(messageId); + if (existing) { + const size = JSON.stringify(msg).length; + this.#bytes += size - existing.size; + existing.msg = msg; + existing.size = size; + this.#evict(); + return; + } + } + const entry: Entry = { msg, size: JSON.stringify(msg).length }; + this.#entries.push(entry); + if (messageId !== undefined) this.#byId.set(messageId, entry); + this.#bytes += entry.size; this.#evict(); } @@ -49,8 +88,11 @@ export class ScrollbackBuffer { this.#bytes > this.#config.maxBytes ) { const evicted = this.#entries.shift(); - if (evicted) { - this.#bytes -= JSON.stringify(evicted).length; + if (!evicted) break; + this.#bytes -= evicted.size; + const id = messageIdOf(evicted.msg); + if (id !== undefined && this.#byId.get(id) === evicted) { + this.#byId.delete(id); } } } @@ -60,16 +102,18 @@ export class ScrollbackBuffer { * Returns a snapshot — safe to iterate while new messages arrive. */ read(): DaemonMessage[] { - return [...this.#entries]; + return this.#entries.map((e) => e.msg); } /** * Read messages after a given timestamp (for incremental catch-up). */ readSince(timestamp: string): DaemonMessage[] { - return this.#entries.filter( - (msg) => "timestamp" in msg && (msg as { timestamp: string }).timestamp > timestamp, - ); + return this.#entries + .map((e) => e.msg) + .filter( + (msg) => "timestamp" in msg && (msg as { timestamp: string }).timestamp > timestamp, + ); } /** @@ -77,20 +121,13 @@ export class ScrollbackBuffer { * transitions so scrollback replay shows final states, not intermediate. */ updateMessage(messageId: string, updater: (msg: DaemonMessage) => void): void { - for (const entry of this.#entries) { - if (entry.type === "session.message" && (entry as { messageId?: string }).messageId === messageId) { - // Re-account bytes around the in-place mutation. Tool entries are - // pushed small (no output) then mutated to carry large output; without - // adjusting #bytes here, eviction later subtracts the grown size that - // was never added, drifting #bytes negative and defeating the byte cap. - const before = JSON.stringify(entry).length; - updater(entry); - const after = JSON.stringify(entry).length; - this.#bytes += after - before; - this.#evict(); - return; - } - } + const entry = this.#byId.get(messageId); + if (!entry) return; + updater(entry.msg); + const after = JSON.stringify(entry.msg).length; + this.#bytes += after - entry.size; + entry.size = after; + this.#evict(); } /** Number of entries currently buffered. */ @@ -106,6 +143,7 @@ export class ScrollbackBuffer { /** Clear the buffer. */ clear(): void { this.#entries = []; + this.#byId.clear(); this.#bytes = 0; } } diff --git a/src/daemon/session-manager.ts b/src/daemon/session-manager.ts index 711762c..1d38b9c 100644 --- a/src/daemon/session-manager.ts +++ b/src/daemon/session-manager.ts @@ -157,10 +157,12 @@ export class SessionManager { onModels: (m) => this.#cacheModels(m), }); - // Restore scrollback from transcript + // Restore scrollback from transcript, seeding the seq counter past + // the persisted tail so new appends continue the monotonic sequence. const entries = await this.#transcriptStore.loadTranscript(meta.sessionId); const messages = entries.map((e) => e.message); - session.restoreScrollback(messages); + const maxSeq = entries.reduce((max, e) => Math.max(max, e.seq), -1); + session.restoreScrollback(messages, maxSeq + 1); this.#sessions.set(session.id, session); // Resume is NOT a creation — don't burn a slot in the diff --git a/src/daemon/session.ts b/src/daemon/session.ts index 38291c7..31a71b2 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -931,7 +931,14 @@ export class Session { await this.#transcriptStore.delete(this.id); } - restoreScrollback(messages: DaemonMessage[]): void { + restoreScrollback(messages: DaemonMessage[], nextSeq?: number): void { + // Seed the transcript sequence counter past the loaded log's tail. + // Without this, post-restart appends restart at seq 0 — harmless for + // loadTranscript (which orders by file position) but it makes seq + // unusable as a monotonic replay cursor. + if (nextSeq !== undefined && nextSeq > this.#seq) { + this.#seq = nextSeq; + } for (const msg of messages) { if (msg.type !== "session.message") continue; // Reconcile tool calls frozen in a non-terminal phase (streaming / @@ -1797,7 +1804,12 @@ export class Session { case "text_delta": { if (!this.#activeAssistantMsg) { this.#activeAssistantMsg = this.#makeMessage("assistant", "", this.#agentIdentity, []); - this.#persistAndBuffer(this.#activeAssistantMsg); + // Scrollback only — no transcript row, no chunker event. The buffer + // holds the message by reference so clients attaching mid-stream see + // it grow; the durable row and the chunker event are emitted once, + // with final content, by #commitStreamed. Feeding the chunker an + // empty assistant message here would emit a promptless half-episode. + this.#scrollback.push(this.#activeAssistantMsg); this.#broadcastRaw(this.#activeAssistantMsg); if (this.#status === "tool_running") this.#setStatus("thinking"); } @@ -1823,7 +1835,7 @@ export class Session { if (this.#activeAssistantMsg) { this.#activeAssistantMsg.content = event.content; this.#activeAssistantMsg.parts = [{ kind: "text", text: event.content, markdown: true }]; - this.#persistAndBuffer(this.#activeAssistantMsg); + this.#commitStreamed(this.#activeAssistantMsg); this.#broadcastRaw(this.#activeAssistantMsg); this.#activeAssistantMsg = null; } else if (event.content) { @@ -1842,7 +1854,9 @@ export class Session { this.#finalizeActiveThinking(); this.#activeThinkingMsg = this.#makeMessage("thinking", "", this.#agentIdentity, []); this.#activeThinkingIndex = event.blockIndex ?? null; - this.#persistAndBuffer(this.#activeThinkingMsg); + // Scrollback only — see the text_delta note; committed by + // #finalizeActiveThinking → #commitStreamed. + this.#scrollback.push(this.#activeThinkingMsg); this.#broadcastRaw(this.#activeThinkingMsg); } if (event.content) { @@ -2093,7 +2107,7 @@ export class Session { if (!m.content || m.content.length === 0) { m.content = "(no output)"; } - this.#persistAndBuffer(m); + this.#commitStreamed(m); this.#broadcastRaw(m); } @@ -2119,7 +2133,8 @@ export class Session { const msg = this.#makeMessage("assistant", "", this.#agentIdentity, []); this.#activeAssistantMsg = msg; - this.#persistAndBuffer(msg); + // Scrollback only — committed with final content below. + this.#scrollback.push(msg); this.#broadcastRaw(msg); for (let pos = 0; pos < content.length; pos += charsPerStep) { @@ -2138,24 +2153,9 @@ export class Session { } if (this.#activeAssistantMsg !== msg) return; // interrupted on last frame - const finalParts: ContentPart[] = [{ kind: "text", text: content, markdown: true }]; - // Reset to the placeholder size so updateMessage measures the correct - // before/after byte delta — the buffer holds msg by reference, so - // mutations here are visible to the accounting logic inside updateMessage. - msg.content = ""; - msg.parts = []; - // Do NOT call #persistAndBuffer again — it would push a second scrollback entry - // for the same messageId, causing duplicate messages on scrollback.replay. - // The updater sets final content/parts inside the buffer's size-accounting pass. - this.#scrollback.updateMessage(msg.messageId, (entry) => { - const sm = entry as SessionMessage; - sm.content = content; - sm.parts = finalParts; - }); - this.#transcriptStore.append(this.id, msg, this.#seq++).catch((e) => { - console.error(`[codeoid/session ${this.id}] transcript append failed: ${e instanceof Error ? e.message : String(e)}`); - }); - this.#chunker?.onMessage(msg); + msg.content = content; + msg.parts = [{ kind: "text", text: content, markdown: true }]; + this.#commitStreamed(msg); this.#broadcastRaw(msg); this.#activeAssistantMsg = null; } @@ -2175,7 +2175,7 @@ export class Session { if (!m.content || m.content.length === 0) { m.content = "(reasoning elided)"; } - this.#persistAndBuffer(m); + this.#commitStreamed(m); this.#broadcastRaw(m); } @@ -2260,6 +2260,28 @@ export class Session { }; } + /** + * Commit the final content of a streamed message. The message was pushed + * into scrollback (by reference) at stream start and grew in place via + * deltas; ScrollbackBuffer.push upserts by messageId, so this re-accounts + * the existing entry — or re-adds it if it was evicted mid-stream — without + * ever creating a duplicate. A second entry per messageId corrupts + * scrollback.replay: clients render the message twice and virtualizers + * keyed on messageId collide (the #50 bug class). The durable transcript + * row and the memory-chunker event are emitted here exactly once, with + * final content, so plain turns produce one user+assistant episode instead + * of two half-episodes. + */ + #commitStreamed(msg: SessionMessage): void { + this.#scrollback.push(msg); + this.#transcriptStore.append(this.id, msg, this.#seq++).catch((e) => { + console.error( + `[codeoid/session ${this.id}] transcript append failed: ${e instanceof Error ? e.message : String(e)}`, + ); + }); + this.#chunker?.onMessage(msg); + } + /** Persist to transcript + scrollback buffer + memory chunker */ #persistAndBuffer(msg: SessionMessage): void { this.#scrollback.push(msg); diff --git a/src/tests/session-stream-commit.test.ts b/src/tests/session-stream-commit.test.ts new file mode 100644 index 0000000..840cf7a --- /dev/null +++ b/src/tests/session-stream-commit.test.ts @@ -0,0 +1,446 @@ +/** + * Streamed-message commit regression tests — the #50 bug class. + * + * A streamed message (assistant text, thinking) is pushed into scrollback + * once at stream start (empty, held by reference so mid-stream attaches see + * it grow) and must be COMMITTED — never pushed again — when the stream + * finalizes. A second push for the same messageId puts two entries in the + * ring buffer; scrollback.replay then renders the message twice and + * virtualizers keyed on messageId collide (web UI overlapping rows). + * + * #50 fixed this only inside #artificiallyStreamText. These tests pin the + * fix across ALL finalize paths: + * + * C1 text_delta → text_done (normal streamed turn) + * C2 thinking_delta → thinking_done (reasoning blocks) + * C3 text_done only (batch reply → artificial streaming) + * C4 text_delta → turn_done (turn ends mid-stream → flush) + * C5 ScrollbackBuffer byte accounting never drifts negative and the + * byte cap keeps evicting after by-reference growth + * C6 EpisodeChunker sees one commit-time assistant message per turn → + * one combined user+assistant episode (not two half-episodes) + * C7 #seq resumes past the persisted transcript tail after restart + */ + +import { describe, it, expect, beforeEach, afterEach } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { randomUUID } from "node:crypto"; +import { Store } from "../daemon/store.js"; +import { TranscriptStore } from "../daemon/transcript.js"; +import { ScrollbackBuffer } from "../daemon/scrollback.js"; +import { Session, type AttachedClient } from "../daemon/session.js"; +import { MockSessionProvider } from "../daemon/providers/mock/session-provider.js"; +import { mockResult } from "../daemon/providers/mock/index.js"; +import { EpisodeChunker } from "../daemon/memory/index.js"; +import type { Episode } from "../daemon/memory/types.js"; +import type { DaemonMessage, AuthContext, SessionMessage } from "../protocol/types.js"; +import { SYSTEM_IDENTITY } from "../protocol/types.js"; +import type { ProviderEvent } from "../daemon/providers/interface.js"; + +// ── Fixtures (same shape as session-integration.test.ts) ───────────────────── + +const TEST_AUTH: AuthContext = { + sub: "user:test-stream-commit", + scopes: [], + delegationDepth: 0, + accountId: "acc-stream", + projectId: "proj-stream", +}; + +let tmp: string; +let store: Store; +let transcriptStore: TranscriptStore; + +beforeEach(() => { + tmp = mkdtempSync(join(tmpdir(), "codeoid-stream-commit-")); + store = new Store(join(tmp, "codeoid.db")); + transcriptStore = new TranscriptStore(join(tmp, "transcripts")); +}); + +afterEach(async () => { + // Yield so fire-and-forget writes from the previous test settle before + // the store closes and the tmp dir is removed (see session-integration). + await new Promise((r) => setTimeout(r, 100)); + try { store.close(); } catch {} + try { rmSync(tmp, { recursive: true, force: true }); } catch {} +}); + +function makeSession(provider: MockSessionProvider, name = "stream-commit-test"): Session { + const id = randomUUID(); + store.createSession({ + id, + name, + workdir: tmp, + status: "idle", + createdBy: TEST_AUTH.sub, + createdAt: new Date().toISOString(), + attachedClients: 0, + accountId: TEST_AUTH.accountId!, + projectId: TEST_AUTH.projectId!, + }); + return new Session({ + name, + workdir: tmp, + auth: TEST_AUTH, + store, + transcriptStore, + existingId: id, + _testProvider: provider, + }); +} + +function makeClient(id = randomUUID()): { client: AttachedClient; received: DaemonMessage[] } { + const received: DaemonMessage[] = []; + return { + client: { id, auth: TEST_AUTH, send: (msg) => received.push(msg) }, + received, + }; +} + +function waitForIdle(session: Session, timeoutMs = 8000): Promise { + if (session.status === "idle" || session.status === "error") return Promise.resolve(); + return new Promise((resolve, reject) => { + const watcherId = randomUUID(); + const timer = setTimeout(() => { + session.detach(watcherId); + reject(new Error(`session did not reach idle within ${timeoutMs}ms — status=${session.status}`)); + }, timeoutMs); + const watcher: AttachedClient = { + id: watcherId, + auth: TEST_AUTH, + send(msg) { + if (msg.type === "session.status_change" && + (msg.status === "idle" || msg.status === "error")) { + clearTimeout(timer); + session.detach(watcherId); + resolve(); + } + }, + }; + session.attach(watcher); + }); +} + +/** Attach a fresh client and return the scrollback.replay it receives. */ +function replayFor(session: Session): SessionMessage[] { + const { client, received } = makeClient(); + session.attach(client); + session.detach(client.id); + const replay = received.find((m) => m.type === "scrollback.replay"); + if (!replay) return []; + return (replay as { messages: SessionMessage[] }).messages; +} + +/** Assert every messageId appears exactly once; returns messages of a role. */ +function assertNoDuplicates(messages: SessionMessage[]): void { + const seen = new Map(); + for (const m of messages) { + seen.set(m.messageId, (seen.get(m.messageId) ?? 0) + 1); + } + const dupes = [...seen.entries()].filter(([, n]) => n > 1); + expect(dupes).toEqual([]); +} + +const turnDone: ProviderEvent = { type: "turn_done", result: mockResult() }; + +// ── C1: normal streamed turn ────────────────────────────────────────────────── + +describe("C1 – text_delta → text_done commits exactly one scrollback entry", () => { + it("replay after a streamed turn has one assistant entry with final content", async () => { + const provider = new MockSessionProvider("claude", [ + [ + { type: "text_delta", content: "Hello " }, + { type: "text_delta", content: "world" }, + { type: "text_done", content: "Hello world" }, + turnDone, + ], + ]); + const session = makeSession(provider); + + await session.send("greet me", TEST_AUTH); + await waitForIdle(session); + + const replay = replayFor(session); + assertNoDuplicates(replay); + const assistant = replay.filter((m) => m.role === "assistant"); + expect(assistant).toHaveLength(1); + expect(assistant[0]!.content).toBe("Hello world"); + }); +}); + +// ── C2: thinking blocks ─────────────────────────────────────────────────────── + +describe("C2 – thinking stream commits exactly one scrollback entry", () => { + it("replay after a thinking block has one thinking entry with full content", async () => { + const provider = new MockSessionProvider("claude", [ + [ + { type: "thinking_delta", content: "hmm ", blockIndex: 0 }, + { type: "thinking_delta", content: "got it", blockIndex: 0 }, + { type: "thinking_done", blockIndex: 0 }, + { type: "text_delta", content: "answer" }, + { type: "text_done", content: "answer" }, + turnDone, + ], + ]); + const session = makeSession(provider); + + await session.send("think hard", TEST_AUTH); + await waitForIdle(session); + + const replay = replayFor(session); + assertNoDuplicates(replay); + const thinking = replay.filter((m) => m.role === "thinking"); + expect(thinking).toHaveLength(1); + expect(thinking[0]!.content).toBe("hmm got it"); + }); +}); + +// ── C3: batch reply → artificial streaming (the original #50 path) ─────────── + +describe("C3 – text_done without deltas (artificial streaming) commits one entry", () => { + it("replay after a batch reply has one assistant entry", async () => { + const provider = new MockSessionProvider("claude", [ + [ + { type: "text_done", content: "batch reply, no streaming" }, + turnDone, + ], + ]); + const session = makeSession(provider); + + await session.send("reply in batch", TEST_AUTH); + await waitForIdle(session); + + const replay = replayFor(session); + assertNoDuplicates(replay); + const assistant = replay.filter((m) => m.role === "assistant"); + expect(assistant).toHaveLength(1); + expect(assistant[0]!.content).toBe("batch reply, no streaming"); + }); +}); + +// ── C4: turn ends mid-stream → #flushActiveAssistant ───────────────────────── + +describe("C4 – turn boundary without text_done commits the partial exactly once", () => { + it("replay after a flushed partial has one assistant entry with the partial content", async () => { + const provider = new MockSessionProvider("claude", [ + [ + { type: "text_delta", content: "partial rep" }, + turnDone, // no text_done — flush path commits the partial + ], + ]); + const session = makeSession(provider); + + await session.send("get cut off", TEST_AUTH); + await waitForIdle(session); + + const replay = replayFor(session); + assertNoDuplicates(replay); + const assistant = replay.filter((m) => m.role === "assistant"); + expect(assistant).toHaveLength(1); + expect(assistant[0]!.content).toBe("partial rep"); + }); +}); + +// ── C5: ScrollbackBuffer accounting under by-reference growth ───────────────── + +describe("C5 – ScrollbackBuffer upsert + byte accounting", () => { + function msg(id: string, content: string): SessionMessage { + return { + type: "session.message", + sessionId: "s1", + messageId: id, + role: "assistant", + content, + identity: SYSTEM_IDENTITY, + timestamp: new Date().toISOString(), + }; + } + + it("re-pushing the same messageId upserts instead of appending", () => { + const buf = new ScrollbackBuffer(); + const m = msg("m1", ""); + buf.push(m); + m.content = "grown by reference during streaming"; + buf.push(m); // commit — must not create a second entry + expect(buf.length).toBe(1); + expect((buf.read()[0] as SessionMessage).content).toBe( + "grown by reference during streaming", + ); + }); + + it("bytes match the final serialized size after by-reference growth + re-push", () => { + const buf = new ScrollbackBuffer(); + const m = msg("m1", ""); + buf.push(m); + m.content = "x".repeat(10_000); + buf.push(m); + expect(buf.bytes).toBe(JSON.stringify(m).length); + }); + + it("byte cap keeps evicting after streamed growth (no negative drift)", () => { + // Pre-fix, the empty-push/full-subtract asymmetry drove #bytes negative, + // permanently disabling the byte cap. Stream many messages through the + // push-grow-push lifecycle and verify the cap still holds. + const buf = new ScrollbackBuffer({ maxBytes: 5_000, maxEntries: 1_000 }); + for (let i = 0; i < 50; i++) { + const m = msg(`m${i}`, ""); + buf.push(m); + m.content = "y".repeat(1_000); + buf.push(m); + } + expect(buf.bytes).toBeGreaterThanOrEqual(0); + expect(buf.bytes).toBeLessThanOrEqual(5_000); + expect(buf.length).toBeLessThan(50); + }); + + it("eviction forgets the id: a later push appends a fresh entry", () => { + const buf = new ScrollbackBuffer({ maxEntries: 2, maxBytes: 1024 * 1024 }); + const m1 = msg("m1", "a"); + buf.push(m1); + buf.push(msg("m2", "b")); + buf.push(msg("m3", "c")); // evicts m1 + expect(buf.length).toBe(2); + buf.push(msg("m1", "reborn")); // must append, not resurrect the old slot + expect(buf.length).toBe(2); // m2 evicted by the cap + const ids = buf.read().map((e) => (e as SessionMessage).messageId); + expect(ids).toEqual(["m3", "m1"]); + }); + + it("updateMessage re-accounts against the recorded size", () => { + const buf = new ScrollbackBuffer(); + const m = msg("m1", "small"); + buf.push(m); + buf.updateMessage("m1", (entry) => { + (entry as SessionMessage).content = "z".repeat(5_000); + }); + expect(buf.bytes).toBe(JSON.stringify(m).length); + }); +}); + +// ── C6: chunker episode pairing ─────────────────────────────────────────────── + +describe("C6 – one commit-time assistant message → one combined episode", () => { + function sm(role: SessionMessage["role"], content: string): SessionMessage { + return { + type: "session.message", + sessionId: "s1", + messageId: randomUUID(), + role, + content, + identity: SYSTEM_IDENTITY, + timestamp: new Date().toISOString(), + }; + } + + it("user + finalized assistant yields a single user_turn episode with both halves", () => { + const episodes: Omit[] = []; + const chunker = new EpisodeChunker( + { workspaceId: "w1", sessionId: "s1", createdBy: TEST_AUTH.sub }, + (ep) => episodes.push(ep), + ); + // Post-fix message sequence: the session feeds the chunker ONCE per + // streamed message, at commit time, with final content. + chunker.onMessage(sm("user", "write me a haiku")); + chunker.onMessage(sm("assistant", "an old silent pond")); + + expect(episodes).toHaveLength(1); + expect(episodes[0]!.kind).toBe("user_turn"); + expect(episodes[0]!.content).toContain("write me a haiku"); + expect(episodes[0]!.content).toContain("an old silent pond"); + }); + + it("documents the pre-fix fragmentation: an empty stream-start assistant splits the turn", () => { + // This is what the session used to emit (push at stream start with empty + // content, push again at finalize) and why #commitStreamed must be the + // only chunker feed for streamed messages. + const episodes: Omit[] = []; + const chunker = new EpisodeChunker( + { workspaceId: "w1", sessionId: "s1", createdBy: TEST_AUTH.sub }, + (ep) => episodes.push(ep), + ); + chunker.onMessage(sm("user", "write me a haiku")); + chunker.onMessage(sm("assistant", "")); // stream-start push (pre-fix) + chunker.onMessage(sm("assistant", "an old silent pond")); // finalize push + + expect(episodes).toHaveLength(2); + expect(episodes[0]!.content).not.toContain("an old silent pond"); + expect(episodes[1]!.kind).toBe("assistant_turn"); // promptless half-episode + }); +}); + +// ── C7: seq resumes past the persisted tail ─────────────────────────────────── + +describe("C7 – transcript seq seeds from the loaded log on resume", () => { + it("appends after restoreScrollback continue the sequence instead of restarting at 0", async () => { + const sessionId = randomUUID(); + const mkMsg = (id: string, role: SessionMessage["role"], content: string): SessionMessage => ({ + type: "session.message", + sessionId, + messageId: id, + role, + content, + identity: SYSTEM_IDENTITY, + timestamp: new Date().toISOString(), + }); + + // Simulate a prior daemon lifetime: three persisted rows, seq 0..2. + await transcriptStore.append(sessionId, mkMsg("p1", "user", "old prompt"), 0); + await transcriptStore.append(sessionId, mkMsg("p2", "assistant", "old reply"), 1); + await transcriptStore.append(sessionId, mkMsg("p3", "user", "old prompt 2"), 2); + + const provider = new MockSessionProvider("claude", [ + [ + { type: "text_delta", content: "new reply" }, + { type: "text_done", content: "new reply" }, + turnDone, + ], + ]); + store.createSession({ + id: sessionId, + name: "resume-seq", + workdir: tmp, + status: "idle", + createdBy: TEST_AUTH.sub, + createdAt: new Date().toISOString(), + attachedClients: 0, + accountId: TEST_AUTH.accountId!, + projectId: TEST_AUTH.projectId!, + }); + const session = new Session({ + name: "resume-seq", + workdir: tmp, + auth: TEST_AUTH, + store, + transcriptStore, + existingId: sessionId, + _testProvider: provider, + }); + + // Mirror SessionManager.resumeSessions: load, seed seq past the tail. + const entries = await transcriptStore.loadTranscript(sessionId); + const maxSeq = entries.reduce((max, e) => Math.max(max, e.seq), -1); + session.restoreScrollback(entries.map((e) => e.message), maxSeq + 1); + expect(maxSeq).toBe(2); + + await session.send("new prompt", TEST_AUTH); + await waitForIdle(session); + // Let the fire-and-forget transcript appends settle. + await new Promise((r) => setTimeout(r, 150)); + + const raw = await Bun.file(transcriptStore.transcriptPath(sessionId)).text(); + const rows = raw + .split("\n") + .filter((l) => l.trim()) + .map((l) => JSON.parse(l) as { seq: number }); + const newSeqs = rows.slice(3).map((r) => r.seq); + expect(newSeqs.length).toBeGreaterThan(0); + // Every post-resume row continues past the persisted tail. + for (const s of newSeqs) expect(s).toBeGreaterThanOrEqual(3); + // And the sequence is strictly increasing (no reuse). + for (let i = 1; i < newSeqs.length; i++) { + expect(newSeqs[i]!).toBeGreaterThan(newSeqs[i - 1]!); + } + }); +}); From b01879e6f4c1471834ab599c105d88be1149be9d Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Fri, 3 Jul 2026 00:12:41 +0800 Subject: [PATCH 2/3] fix: account scrollback bytes as UTF-8, add live-session chunker regression test CodeRabbit review follow-ups on #74: - String.length counts UTF-16 code units; use Buffer.byteLength so the 20MB cap holds for non-ASCII payloads - end-to-end test that a real Session + MemoryEngine ingests exactly one combined user+assistant episode per streamed turn (a stray stream-start chunker feed would fail it) Co-Authored-By: Claude Fable 5 --- src/daemon/scrollback.ts | 14 ++++- src/tests/session-stream-commit.test.ts | 73 ++++++++++++++++++++++++- 2 files changed, 82 insertions(+), 5 deletions(-) diff --git a/src/daemon/scrollback.ts b/src/daemon/scrollback.ts index cecfce6..0937d05 100644 --- a/src/daemon/scrollback.ts +++ b/src/daemon/scrollback.ts @@ -42,6 +42,14 @@ function messageIdOf(msg: DaemonMessage): string | undefined { : undefined; } +/** + * Serialized size in real UTF-8 bytes. `String.length` counts UTF-16 code + * units, undercounting non-ASCII payloads against the byte cap. + */ +function serializedSizeOf(msg: DaemonMessage): number { + return Buffer.byteLength(JSON.stringify(msg), "utf8"); +} + export class ScrollbackBuffer { #entries: Entry[] = []; #byId = new Map(); @@ -66,7 +74,7 @@ export class ScrollbackBuffer { if (messageId !== undefined) { const existing = this.#byId.get(messageId); if (existing) { - const size = JSON.stringify(msg).length; + const size = serializedSizeOf(msg); this.#bytes += size - existing.size; existing.msg = msg; existing.size = size; @@ -74,7 +82,7 @@ export class ScrollbackBuffer { return; } } - const entry: Entry = { msg, size: JSON.stringify(msg).length }; + const entry: Entry = { msg, size: serializedSizeOf(msg) }; this.#entries.push(entry); if (messageId !== undefined) this.#byId.set(messageId, entry); this.#bytes += entry.size; @@ -124,7 +132,7 @@ export class ScrollbackBuffer { const entry = this.#byId.get(messageId); if (!entry) return; updater(entry.msg); - const after = JSON.stringify(entry.msg).length; + const after = serializedSizeOf(entry.msg); this.#bytes += after - entry.size; entry.size = after; this.#evict(); diff --git a/src/tests/session-stream-commit.test.ts b/src/tests/session-stream-commit.test.ts index 840cf7a..c135205 100644 --- a/src/tests/session-stream-commit.test.ts +++ b/src/tests/session-stream-commit.test.ts @@ -33,7 +33,12 @@ import { ScrollbackBuffer } from "../daemon/scrollback.js"; import { Session, type AttachedClient } from "../daemon/session.js"; import { MockSessionProvider } from "../daemon/providers/mock/session-provider.js"; import { mockResult } from "../daemon/providers/mock/index.js"; -import { EpisodeChunker } from "../daemon/memory/index.js"; +import { + EpisodeChunker, + MemoryEngine, + SqliteEpisodeStore, +} from "../daemon/memory/index.js"; +import type { Embedder } from "../daemon/memory/embedder.js"; import type { Episode } from "../daemon/memory/types.js"; import type { DaemonMessage, AuthContext, SessionMessage } from "../protocol/types.js"; import { SYSTEM_IDENTITY } from "../protocol/types.js"; @@ -67,7 +72,31 @@ afterEach(async () => { try { rmSync(tmp, { recursive: true, force: true }); } catch {} }); -function makeSession(provider: MockSessionProvider, name = "stream-commit-test"): Session { +/** Deterministic embedder so MemoryEngine runs offline (same as memory.test.ts). */ +class StubEmbedder implements Embedder { + readonly modelName = "stub-embed"; + readonly dimensions = 8; + async init(): Promise {} + async embed(texts: string[]): Promise { + return texts.map((t) => { + const v = new Float32Array(this.dimensions); + for (let i = 0; i < t.length; i++) { + v[i % this.dimensions]! += t.charCodeAt(i) / 1000; + } + let norm = 0; + for (let i = 0; i < v.length; i++) norm += v[i]! * v[i]!; + norm = Math.sqrt(norm) || 1; + for (let i = 0; i < v.length; i++) v[i] = v[i]! / norm; + return v; + }); + } +} + +function makeSession( + provider: MockSessionProvider, + name = "stream-commit-test", + memory?: MemoryEngine, +): Session { const id = randomUUID(); store.createSession({ id, @@ -88,6 +117,7 @@ function makeSession(provider: MockSessionProvider, name = "stream-commit-test") transcriptStore, existingId: id, _testProvider: provider, + memory, }); } @@ -317,6 +347,15 @@ describe("C5 – ScrollbackBuffer upsert + byte accounting", () => { }); expect(buf.bytes).toBe(JSON.stringify(m).length); }); + + it("accounts UTF-8 bytes, not UTF-16 code units", () => { + const buf = new ScrollbackBuffer(); + const m = msg("m1", "नमस्ते 🙏 — multi-byte content"); + buf.push(m); + const json = JSON.stringify(m); + expect(buf.bytes).toBe(Buffer.byteLength(json, "utf8")); + expect(buf.bytes).toBeGreaterThan(json.length); // .length would undercount + }); }); // ── C6: chunker episode pairing ─────────────────────────────────────────────── @@ -351,6 +390,36 @@ describe("C6 – one commit-time assistant message → one combined episode", () expect(episodes[0]!.content).toContain("an old silent pond"); }); + it("live session feeds the chunker exactly once per streamed turn", async () => { + // End-to-end version of the pairing test: a real Session with a real + // MemoryEngine. A stray chunker feed at stream start (the pre-fix + // behavior) would ingest a prompt-only user_turn plus a promptless + // assistant_turn — this asserts exactly one combined episode lands. + const memStore = new SqliteEpisodeStore(join(tmp, "memory.db")); + const engine = new MemoryEngine({ store: memStore, embedder: new StubEmbedder() }); + await engine.init(); + + const provider = new MockSessionProvider("claude", [ + [ + { type: "text_delta", content: "an old " }, + { type: "text_delta", content: "silent pond" }, + { type: "text_done", content: "an old silent pond" }, + turnDone, + ], + ]); + const session = makeSession(provider, "chunker-e2e", engine); + + await session.send("write me a haiku", TEST_AUTH); + await waitForIdle(session); + + const episodes = memStore.listEpisodesForSession(session.id); + expect(episodes).toHaveLength(1); + expect(episodes[0]!.kind).toBe("user_turn"); + expect(episodes[0]!.content).toContain("write me a haiku"); + expect(episodes[0]!.content).toContain("an old silent pond"); + memStore.close(); + }); + it("documents the pre-fix fragmentation: an empty stream-start assistant splits the turn", () => { // This is what the session used to emit (push at stream start with empty // content, push again at finalize) and why #commitStreamed must be the From 3cf82048acc20824d167ca49469242c3d2b1f362 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Fri, 3 Jul 2026 00:16:09 +0800 Subject: [PATCH 3/3] fix: StubEmbedder missing close() from the Embedder interface Co-Authored-By: Claude Fable 5 --- src/tests/session-stream-commit.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/tests/session-stream-commit.test.ts b/src/tests/session-stream-commit.test.ts index c135205..d04d23d 100644 --- a/src/tests/session-stream-commit.test.ts +++ b/src/tests/session-stream-commit.test.ts @@ -77,6 +77,7 @@ class StubEmbedder implements Embedder { readonly modelName = "stub-embed"; readonly dimensions = 8; async init(): Promise {} + async close(): Promise {} async embed(texts: string[]): Promise { return texts.map((t) => { const v = new Float32Array(this.dimensions);