From 4bc847741d5d589dfd5204b222587ec71755d87b Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Mon, 28 Sep 2026 20:24:35 -0500 Subject: [PATCH] fix(desktop): keep a pending streamed reply over a divergent same-turn store partial Fixes #80151 --- .../streamed-reply-survives-sync.test.ts | 112 ++++++++++++++++++ .../hooks/use-session-actions/utils.ts | 71 ++++++++++- 2 files changed, 178 insertions(+), 5 deletions(-) diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/streamed-reply-survives-sync.test.ts b/apps/desktop/src/app/session/hooks/use-session-actions/streamed-reply-survives-sync.test.ts index db066af9b0..0371723d9f 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/streamed-reply-survives-sync.test.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/streamed-reply-survives-sync.test.ts @@ -100,3 +100,115 @@ describe('preserveLocalPendingTurnMessages — settled reply over an empty tool expect(preserveLocalPendingTurnMessages(next, [liveUser, settledStream([textPart('Model is up.')])])).toEqual(next) }) }) + +// #80151: the backend persists segments while the turn runs, so switching away +// mid-stream and back hydrates the store's flat mid-turn partial next to the +// still-streaming local copy. That partial need not be a byte prefix of the +// streamed text, so the strict-prefix pairing dropped the pre-switch content. +describe('preserveLocalPendingTurnMessages — divergent same-turn store partial (#80151)', () => { + it('keeps a pending streamed reply when the store holds a same-turn partial whose flat text is not a byte prefix', () => { + const previous = [ + message('user-live', 'user', [textPart('inspect the logs')]), + message( + 'assistant-stream-live', + 'assistant', + [textPart('Reading the logs now.'), tool('call-1'), textPart('Found the stale lock.')], + { pending: true } + ) + ] + + const next = [ + message('user-stored', 'user', [textPart('inspect the logs')], { rowId: 300 }), + message('assistant-stored-partial', 'assistant', [textPart('Reading the logs now.\n\nFound the stale')], { + pending: false, + rowId: 301 + }) + ] + + const merged = preserveLocalPendingTurnMessages(next, previous) + + expect(merged).toHaveLength(2) + expect(merged[1]).toMatchObject({ id: 'assistant-stream-live', rowId: 301 }) + expect(merged[1].parts.some(part => part.type === 'tool-call')).toBe(true) + expect(chatMessageText(merged[1])).toContain('Found the stale lock.') + }) + + it('keeps a pending streamed reply over a committed same-turn partial proven by shared tool-call ids', () => { + // A tool-heavy turn: the store committed the tool round plus the segment + // after it while the turn was still running. The committed text is not a + // prefix of the streamed text (the stream also holds the pre-tool + // commentary), but the shared tool-call id proves the same turn. + const previous = [ + message('user-live', 'user', [textPart('inspect the logs')]), + message( + 'assistant-stream-live', + 'assistant', + [textPart('Reading the logs now.'), tool('call-9'), textPart('Removed the stale lock.')], + { pending: true } + ) + ] + + const next = [ + message('user-stored', 'user', [textPart('inspect the logs')], { rowId: 310 }), + message('assistant-stored-partial', 'assistant', [tool('call-9'), textPart('Removed the stale lock.')], { + pending: false, + rowId: 311 + }) + ] + + const merged = preserveLocalPendingTurnMessages(next, previous) + + expect(merged).toHaveLength(2) + expect(merged[1]).toMatchObject({ id: 'assistant-stream-live', rowId: 311 }) + expect(chatMessageText(merged[1])).toContain('Reading the logs now.') + expect(merged[1].parts.some(part => part.type === 'tool-call')).toBe(true) + }) + + it('does not let a pending stream claim a different turn\'s committed answer at the same ordinal', () => { + // No shared tool-call ids and no folded prefix relation: the committed row + // is an earlier answer to the same resent prompt. The widening must not + // let the unrelated pending stream take its slot, and the stale-copy rule + // keeps dropping the local row. + const previous = [ + message('user-live', 'user', [textPart('inspect the logs')]), + message('assistant-stream-live', 'assistant', [textPart('Fresh streaming reply, different wording.')], { + pending: true + }) + ] + + const next = [ + message('user-stored', 'user', [textPart('inspect the logs')], { rowId: 320 }), + message('assistant-stored-old-answer', 'assistant', [textPart('An older, unrelated answer text.')], { + pending: false, + rowId: 321 + }) + ] + + const merged = preserveLocalPendingTurnMessages(next, previous) + + expect(merged).toEqual(next) + }) + + it('does not trade a committed partial for a pending stream that is behind its text', () => { + // Same turn (shared tool-call id, folded prefix), but the local stream has + // LESS text than the store already persisted — replacing would lose visible + // content, so the committed partial keeps the slot. + const previous = [ + message('user-live', 'user', [textPart('inspect the logs')]), + message('assistant-stream-live', 'assistant', [textPart('Reading the'), tool('call-4')], { pending: true }) + ] + + const next = [ + message('user-stored', 'user', [textPart('inspect the logs')], { rowId: 330 }), + message('assistant-stored-partial', 'assistant', [tool('call-4'), textPart('Reading the logs now. Done.')], { + pending: false, + rowId: 331 + }) + ] + + const merged = preserveLocalPendingTurnMessages(next, previous) + + expect(merged[1]).toMatchObject({ id: 'assistant-stored-partial', rowId: 331 }) + expect(chatMessageText(merged[1])).toContain('Reading the logs now. Done.') + }) +}) diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts b/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts index 5f97a2b42d..87816f6e51 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts @@ -554,6 +554,47 @@ const isGatewaySystemMarker = (message: ChatMessage): boolean => const hasStreamedContent = (message: ChatMessage): boolean => chatMessageText(message).trim().length > 0 || hasStructuralParts(message) +/** + * #80151: the store's flat projection and the streamed parts join the same + * segments with different separators (blank lines around folded tool rounds, + * reference lines), so byte-prefix pairing misses the same turn. Compare the + * answer text with reference lines stripped and separators folded away. + */ +const foldAnswerTextForCompare = (text: string): string => textWithoutReferenceLines(text).replace(/\s+/g, '') + +/** + * #80151: may a still-pending local stream claim a COMMITTED row whose answer + * text the stream provably holds? With the turn proven by shared tool-call + * ids (call ids are unique to the turn), the committed text is the mid-turn + * segment the store flushed — it must be one of the stream's own folded text + * segments, or a prefix of one (a mid-segment flush). Without that proof the + * whole folded answer must strictly extend the committed text, mirroring the + * live-shell prefix rule. An empty-text committed row is never claimable + * here: empty prose carries no identity (#114543); the settled-final shell + * case is preserveLocalPendingTurnMessages' own rule (#123047). + */ +const streamedAnswerHoldsCommittedText = (local: ChatMessage, authoritative: ChatMessage): boolean => { + const foldedAuthoritative = foldAnswerTextForCompare(chatMessageText(authoritative)) + + if (!foldedAuthoritative.length) { + return false + } + + const authoritativeToolIds = toolCallIdsOf(authoritative) + + if (authoritativeToolIds.length > 0 && authoritativeToolIds.every(id => toolCallIdsOf(local).includes(id))) { + const localSegments = local.parts.flatMap(part => + part.type === 'text' ? [foldAnswerTextForCompare(part.text)] : [] + ) + + return localSegments.some(segment => segment.startsWith(foldedAuthoritative)) + } + + const foldedLocal = foldAnswerTextForCompare(chatMessageText(local)) + + return foldedLocal.length > foldedAuthoritative.length && foldedLocal.startsWith(foldedAuthoritative) +} + /** * May the cached local row stand in for this authoritative assistant? * @@ -563,25 +604,45 @@ const hasStreamedContent = (message: ChatMessage): boolean => * — or the stream id — of a genuine stored reply. A retained failure snapshot * (`inflight.error`, projected with empty text) is never a shell: repainting it * from the local partial would hide the error and mark the turn healthy again. + * + * #80151: a COMMITTED row at the same ordinal can be this turn's mid-turn + * partial — the backend persists segments while the turn runs, so switching + * chats mid-stream and back hydrates the store's flat projection next to the + * still-streaming local copy, and that projection need not be a byte prefix of + * the streamed text. Only a still-PENDING local row may claim it, and only + * with same-turn proof: every tool-call id the committed row names also + * streams locally (call ids are unique to the turn, so a different turn at + * the same ordinal — a resent prompt answered earlier — cannot pass), or the + * committed answer text is a strict folded prefix of the streamed text. In + * both arms the streamed copy must cover the committed text, so a complete + * earlier answer is never traded for a shorter local row. */ const localPendingSupersedes = (local: ChatMessage, authoritative: ChatMessage): boolean => { if (local.role !== 'assistant' || !isLiveTailRow(local)) { return false } - if (!isLiveTailRow(authoritative) || authoritative.error) { + if (authoritative.error) { return false } const authoritativeText = chatMessageText(authoritative).trim() - if (!authoritativeText.length) { - return hasStreamedContent(local) + if (isLiveTailRow(authoritative)) { + if (!authoritativeText.length) { + return hasStreamedContent(local) + } + + const localText = chatMessageText(local).trim() + + return localText.length > authoritativeText.length && isStrictAnswerTextExtension(localText, authoritativeText) } - const localText = chatMessageText(local).trim() + if (local.pending !== true) { + return false + } - return localText.length > authoritativeText.length && isStrictAnswerTextExtension(localText, authoritativeText) + return hasStreamedContent(local) && streamedAnswerHoldsCommittedText(local, authoritative) } const answerText = (message: ChatMessage) => textWithoutReferenceLines(chatMessageText(message)).trim()