From fa26bf38923d16f6c98bd48c6ff71ee90dc815b8 Mon Sep 17 00:00:00 2001 From: Chen Yefan Date: Tue, 21 Jul 2026 10:36:09 +0800 Subject: [PATCH] fix(api): supersede streamed fragment rows with their assistant snapshot Streamed replies persist message.completed before the aggregated message.added snapshot arrives, so the timeline folded the stream and then rendered the snapshot again as a separate assistant message. When the driver additionally loses the message_start frame, per-envelope uuids fracture one reply into per-fragment messages and the log shows 'P' / 'ong. ...' / full-text rows (YEF-884). Track fragment rows closed without a snapshot per stream key and let a snapshot whose text prefix-matches the concatenated fragments replace them in place, keeping the first fragment's timeline position. Repairs existing rows at read time, including data written by older drivers. --- .../domain/session-event-stream-fold.ts | 103 +++++++++++-- .../tests/session-event-stream-fold.test.ts | 135 ++++++++++++++++++ 2 files changed, 224 insertions(+), 14 deletions(-) diff --git a/apps/api/src/modules/sessions/domain/session-event-stream-fold.ts b/apps/api/src/modules/sessions/domain/session-event-stream-fold.ts index 57c56f54..1ef897df 100644 --- a/apps/api/src/modules/sessions/domain/session-event-stream-fold.ts +++ b/apps/api/src/modules/sessions/domain/session-event-stream-fold.ts @@ -7,6 +7,14 @@ import type { RuntimeEventId, SessionRunId } from "@mosoo/id"; // events. The merge rules mirror the pre-persistence compactor // (runtime-event-compaction.ts): deltas append, snapshots prefer the longer // prefix-matching text. +// +// Rows carry no message identity, so a stream whose driver-side identity +// fractured (a dropped message_start splits one reply across several +// started/delta groups, YEF-884) still folds into several fragment rows, and +// the trailing message.added snapshot lands after its stream already closed. +// To heal both shapes, closed fragment rows are kept as supersede candidates: +// a snapshot whose text prefix-matches the concatenated fragments replaces +// them with a single row instead of rendering a duplicate. export interface StreamFoldableSessionEventRow { content_text: string; @@ -59,6 +67,11 @@ interface OpenStreamGroup { runId: SessionRunId | null; } +interface ClosedFragmentSegment { + content: string; + outputIndex: number; +} + function appendStreamText(current: string, next: string): string { if (next.length === 0) { return current; @@ -87,6 +100,14 @@ function mergeSnapshotText(current: string, next: string): string { return `${current}${next}`; } +function isSnapshotOfFragments(fragments: string, snapshot: string): boolean { + if (fragments.length === 0 || snapshot.length === 0) { + return false; + } + + return snapshot.startsWith(fragments) || fragments.startsWith(snapshot); +} + function createOpenStreamGroup( row: R, classification: StreamRowClassification, @@ -140,15 +161,60 @@ export function foldStreamedSessionEventRows { - const output: R[] = []; + const output: (R | null)[] = []; const openGroups = new Map>(); + const fragmentSegments = new Map(); - function closeGroup(group: OpenStreamGroup): void { + function closeGroupAsFragment(key: string, group: OpenStreamGroup): void { const folded = createFoldedStreamRow(group); - if (folded !== null) { + if (folded === null) { + return; + } + + output.push(folded); + const segments = fragmentSegments.get(key) ?? []; + segments.push({ content: folded.content_text, outputIndex: output.length - 1 }); + fragmentSegments.set(key, segments); + } + + // A snapshot row is the authoritative text of the message it closes. When + // its text extends (or repeats) the fragment rows already emitted for the + // same stream key, those fragments were partial views of this snapshot: + // collapse them into one row at the first fragment's timeline position. + function closeGroupWithSnapshot(key: string, folded: R | null): void { + if (folded === null) { + return; + } + + const segments = fragmentSegments.get(key) ?? []; + fragmentSegments.delete(key); + const fragments = segments.map((segment) => segment.content).join(""); + + if (!isSnapshotOfFragments(fragments, folded.content_text)) { + output.push(folded); + return; + } + + const firstSegment = segments[0]; + const anchorRow = firstSegment === undefined ? null : output[firstSegment.outputIndex]; + + if (firstSegment === undefined || anchorRow === null || anchorRow === undefined) { output.push(folded); + return; } + + for (const segment of segments) { + output[segment.outputIndex] = null; + } + + output[firstSegment.outputIndex] = { + ...folded, + content_text: + folded.content_text.length >= fragments.length ? folded.content_text : fragments, + occurred_at: anchorRow.occurred_at, + seq: anchorRow.seq, + }; } for (const row of rows) { @@ -162,7 +228,7 @@ export function foldStreamedSessionEventRows row !== null) }; } return { openStreamRows: [...openGroups.values()] .flatMap((group) => group.rows) .toSorted((a, b) => a.seq - b.seq), - rows: output, + rows: output.filter((row): row is R => row !== null), }; } diff --git a/apps/api/tests/session-event-stream-fold.test.ts b/apps/api/tests/session-event-stream-fold.test.ts index 0491f643..2523d068 100644 --- a/apps/api/tests/session-event-stream-fold.test.ts +++ b/apps/api/tests/session-event-stream-fold.test.ts @@ -212,6 +212,141 @@ describe("session event stream folding", () => { expect(secondFold.rows[0]).toMatchObject({ content_text: "流式输出完成", id: "m-end" }); }); + test("supersedes a closed stream with its trailing snapshot instead of duplicating it", () => { + // Streamed replies persist message.completed (at message_stop) before the + // aggregated assistant snapshot arrives, so the snapshot lands after its + // stream already closed and must replace the folded row, not repeat it. + const folded = foldStreamedSessionEventRows( + [ + row({ content: "Message updated.", eventType: "message.started", id: "m-start", seq: 1 }), + row({ content: "P", eventType: "message.delta", id: "m-1", seq: 2 }), + row({ + content: "ong. What would you like to work on?", + eventType: "message.delta", + id: "m-2", + seq: 3, + }), + row({ content: "Message updated.", eventType: "message.completed", id: "m-end", seq: 4 }), + row({ + content: "Pong. What would you like to work on?", + eventType: "message.added", + id: "m-added", + seq: 5, + }), + ], + { flushOpenStreams: true }, + ); + + expect(folded.rows).toHaveLength(1); + expect(folded.rows[0]).toMatchObject({ + content_text: "Pong. What would you like to work on?", + event_type: "message.added", + id: "m-added", + seq: 1, + }); + }); + + test("collapses a fractured stream whose snapshot matches the concatenated fragments", () => { + // YEF-884: a dropped message_start fractures one reply into per-fragment + // messages, so the rows arrive as started/delta pairs per fragment plus a + // final full snapshot. The snapshot equals the fragment concatenation and + // must fold everything into a single timeline entry. + const folded = foldStreamedSessionEventRows( + [ + row({ content: "Message updated.", eventType: "message.started", id: "s-1", seq: 1 }), + row({ content: "P", eventType: "message.delta", id: "m-1", seq: 2 }), + row({ content: "Message updated.", eventType: "message.started", id: "s-2", seq: 3 }), + row({ + content: "ong. What would you like to work on?", + eventType: "message.delta", + id: "m-2", + seq: 4, + }), + row({ content: "Message updated.", eventType: "message.started", id: "s-3", seq: 5 }), + row({ + content: "Pong. What would you like to work on?", + eventType: "message.delta", + id: "m-3", + seq: 6, + }), + row({ + content: "Pong. What would you like to work on?", + eventType: "message.added", + id: "m-added", + seq: 7, + }), + row({ content: "Message updated.", eventType: "message.completed", id: "c-3", seq: 8 }), + row({ content: "Message updated.", eventType: "message.completed", id: "c-1", seq: 9 }), + row({ content: "Message updated.", eventType: "message.completed", id: "c-2", seq: 10 }), + row({ + content: "run-1", + eventType: "run.completed", + id: "r-done", + processType: "run.completed", + seq: 11, + }), + ], + { flushOpenStreams: true }, + ); + + expect(folded.rows.map((entry) => entry.content_text)).toEqual([ + "Pong. What would you like to work on?", + "run-1", + ]); + expect(folded.rows[0]).toMatchObject({ id: "m-added", seq: 1 }); + }); + + test("supersedes a truncated stream with the longer prefix-matching snapshot", () => { + const folded = foldStreamedSessionEventRows( + [ + row({ content: "Final ans", eventType: "message.delta", id: "m-1", seq: 1 }), + row({ content: "Message updated.", eventType: "message.completed", id: "m-end", seq: 2 }), + row({ content: "Final answer.", eventType: "message.added", id: "m-added", seq: 3 }), + ], + { flushOpenStreams: true }, + ); + + expect(folded.rows).toHaveLength(1); + expect(folded.rows[0]).toMatchObject({ + content_text: "Final answer.", + id: "m-added", + seq: 1, + }); + }); + + test("keeps a snapshot that does not extend the closed stream as its own message", () => { + const folded = foldStreamedSessionEventRows( + [ + row({ content: "第一条进度", eventType: "message.delta", id: "m-1", seq: 1 }), + row({ content: "Message updated.", eventType: "message.completed", id: "m-end", seq: 2 }), + row({ content: "另一条最终回复", eventType: "message.added", id: "m-added", seq: 3 }), + ], + { flushOpenStreams: true }, + ); + + expect(folded.rows.map((entry) => entry.content_text)).toEqual([ + "第一条进度", + "另一条最终回复", + ]); + }); + + test("consumes fragments per snapshot across a multi-message turn", () => { + const folded = foldStreamedSessionEventRows( + [ + row({ content: "进度说明", eventType: "message.delta", id: "m-1", seq: 1 }), + row({ content: "Message updated.", eventType: "message.completed", id: "c-1", seq: 2 }), + row({ content: "进度说明", eventType: "message.added", id: "a-1", seq: 3 }), + row({ content: "最终回复", eventType: "message.delta", id: "m-2", seq: 4 }), + row({ content: "Message updated.", eventType: "message.completed", id: "c-2", seq: 5 }), + row({ content: "最终回复", eventType: "message.added", id: "a-2", seq: 6 }), + ], + { flushOpenStreams: true }, + ); + + expect(folded.rows.map((entry) => entry.content_text)).toEqual(["进度说明", "最终回复"]); + expect(folded.rows.map((entry) => entry.id)).toEqual(["a-1", "a-2"]); + }); + test("drops streams that never carried text", () => { const folded = foldStreamedSessionEventRows( [