fix(desktop): keep a pending streamed reply over a divergent same-turn store partial

Fixes #80151

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.')
+  })
+})

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()