From 1bb55780ab7be85dee754eaac7b0343ffa560622 Mon Sep 17 00:00:00 2001 From: finn763 <165816600+finn763@users.noreply.github.com> Date: Wed, 23 Sep 2026 19:51:46 +0800 Subject: [PATCH] fix(desktop): collapse a redelivered interim into its bubble (#120104, #93926) A message.interim redelivered after the first copy sealed the stream took the standalone-append path and painted the same commentary twice. Refresh the newest same-text assistant row of the current turn in place instead, comparing whitespace-normalized text. Supersedes #120145 and #93962. Co-authored-by: Brammiebram <64777864+Brammiebram@users.noreply.github.com> --- .../duplicate-delivery-120104.test.tsx | 58 +++++++++ .../session/hooks/use-message-stream/index.ts | 58 ++++++--- .../interim-redelivery.test.tsx | 114 ++++++++++++++++++ 3 files changed, 216 insertions(+), 14 deletions(-) create mode 100644 apps/desktop/src/app/session/hooks/use-message-stream/duplicate-delivery-120104.test.tsx create mode 100644 apps/desktop/src/app/session/hooks/use-message-stream/interim-redelivery.test.tsx diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/duplicate-delivery-120104.test.tsx b/apps/desktop/src/app/session/hooks/use-message-stream/duplicate-delivery-120104.test.tsx new file mode 100644 index 0000000000..e96f4f8699 --- /dev/null +++ b/apps/desktop/src/app/session/hooks/use-message-stream/duplicate-delivery-120104.test.tsx @@ -0,0 +1,58 @@ +import type { GatewayEventName } from '@hermes/shared' +import { act, cleanup } from '@testing-library/react' +import { afterEach, expect, it } from 'vitest' + +import { chatMessageText } from '@/lib/chat-messages' + +import { renderMessageStream } from './test-harness' + +const SID = 'issue-120104-duplicate-delivery' + +afterEach(cleanup) + +type Frame = [GatewayEventName, Record] + +const SKILL_TURN: Frame[] = [ + ['message.start', {}], + ['message.delta', { text: 'reply text' }], + ['tool.start', { name: 'skill_view', tool_id: 's1', args: { name: 'x' } }], + ['tool.complete', { name: 'skill_view', tool_id: 's1', result: 'skill content' }], + ['message.interim', { text: 'reply text', already_streamed: true }], + ['message.complete', { text: 'reply text' }] +] + +async function mount() { + const stream = renderMessageStream(SID) + const send = (type: GatewayEventName, payload: Record = {}) => + act(() => stream.handleEvent({ type, payload, session_id: SID })) + + return { stream, send } +} + +function expectSingleBubble(stream: ReturnType) { + const messages = stream.state().messages.filter(m => m.role === 'assistant' && !m.hidden) + expect(messages).toHaveLength(1) + expect(chatMessageText(messages[0])).toBe('reply text') + expect(messages[0].parts.filter(part => part.type === 'tool-call')).toHaveLength(1) +} + +// #120104: an assistant reply renders twice while the backend ran a single +// turn. A redelivered message.interim lands after the first copy sealed the +// stream, so it takes the standalone-append path. +it('renders one bubble when the duplicate interim lands after the first completion', async () => { + const { stream, send } = await mount() + + const skewed: Frame[] = [ + ...SKILL_TURN.slice(0, 5), + SKILL_TURN[5], + SKILL_TURN[4], + SKILL_TURN[5] + ] + + for (const [type, payload] of skewed) { + await send(type, payload) + } + + expectSingleBubble(stream) + cleanup() +}) diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/index.ts b/apps/desktop/src/app/session/hooks/use-message-stream/index.ts index 6fe5cfec3a..4a1f23ce33 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/index.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/index.ts @@ -645,20 +645,50 @@ export function useMessageStream({ : m ) } else { - // No streaming bubble — create a standalone interim message - nextMessages = [ - ...nextMessages, - { - id: nextStreamMessageId('assistant-interim'), - role: 'assistant' as const, - parts: [{ ...assistantTextPart(authoritativeText, occurredAt), completedAt: occurredAt }], - timestamp: occurredAt, - completedAt: occurredAt, - pending: false, - interim: true, - branchGroupId: state.pendingBranchGroup ?? undefined - } - ] + // No streaming bubble. Usually a duplicate delivery of the interim + // that just sealed the stream (two sockets, one backend — #120005 + // family): the first copy sealed the bubble and cleared streamId, + // so the second copy lands here. Appending would paint the same + // reply twice (#120104). When this occurrence's newest visible + // assistant row already carries the same text — including a skewed + // duplicate landing after the turn settled — refresh it in place. + const lastUserIndex = nextMessages.findLastIndex(message => message.role === 'user') + const normalizedText = authoritativeText.replace(/\s+/g, ' ').trim() + const prevSameText = nextMessages.findLast( + (message, index) => + index > lastUserIndex && + message.role === 'assistant' && + !message.hidden && + chatMessageText(message).replace(/\s+/g, ' ').trim() === normalizedText + ) + if (prevSameText) { + nextMessages = nextMessages.map(m => + m.id === prevSameText.id + ? { + ...m, + parts: completeOpenTimelineParts(replaceTextPart(m.parts), occurredAt), + completedAt: occurredAt, + pending: false, + interim: true + } + : m + ) + } else { + // No streaming bubble — create a standalone interim message + nextMessages = [ + ...nextMessages, + { + id: nextStreamMessageId('assistant-interim'), + role: 'assistant' as const, + parts: [{ ...assistantTextPart(authoritativeText, occurredAt), completedAt: occurredAt }], + timestamp: occurredAt, + completedAt: occurredAt, + pending: false, + interim: true, + branchGroupId: state.pendingBranchGroup ?? undefined + } + ] + } } return { diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/interim-redelivery.test.tsx b/apps/desktop/src/app/session/hooks/use-message-stream/interim-redelivery.test.tsx new file mode 100644 index 0000000000..5f8685eb8d --- /dev/null +++ b/apps/desktop/src/app/session/hooks/use-message-stream/interim-redelivery.test.tsx @@ -0,0 +1,114 @@ +import type { GatewayEvent } from '@hermes/shared' +import { act, cleanup } from '@testing-library/react' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import { chatMessageText } from '@/lib/chat-messages' +import { clearSessionTodos } from '@/store/todos' + +import { type MessageStreamHarness, renderMessageStream } from './test-harness' + +/** + * Redelivered `message.interim` events must not duplicate bubbles (#93926). + * + * The gateway delivers stream events over a reconnecting websocket, so any + * interim event can arrive twice: back-to-back, or long after its segment was + * sealed once later commentary has streamed. Redelivery is recognized as a + * restatement of recent assistant text. Distinct texts must still append + * (no over-suppression), and `already_streamed` alone must not suppress: + * upstream fixtures rely on those events appending when no bubble is open + * (the streamed deltas are gone, so the event is the only copy). + */ + +const SID = 'session-1' + +let stream: MessageStreamHarness + +function mountStream() { + stream = renderMessageStream(SID) +} + +const ev = (type: string, payload: Record = {}): GatewayEvent => + ({ payload, session_id: SID, type }) as GatewayEvent + +const start = () => act(() => stream.handleEvent(ev('message.start'))) +const delta = (text: string) => act(() => stream.handleEvent(ev('message.delta', { text }))) +const interim = (text: string, extra: Record = {}) => + act(() => stream.handleEvent(ev('message.interim', { text, ...extra }))) + +function assistantTexts(): string[] { + return stream + .state(SID) + .messages.filter(m => m.role === 'assistant' && !m.hidden) + .map(m => chatMessageText(m)) + .filter(Boolean) +} + +beforeEach(() => { + clearSessionTodos(SID) +}) + +afterEach(() => { + cleanup() + clearSessionTodos(SID) + vi.restoreAllMocks() +}) + +describe('useMessageStream redelivered message.interim (#93926)', () => { + it('keeps one bubble when the same interim arrives twice back-to-back', () => { + mountStream() + start() + interim('checking the attach path') + interim('checking the attach path') + + expect(assistantTexts().filter(t => t === 'checking the attach path')).toHaveLength(1) + }) + + it('keeps one bubble when an earlier interim is replayed after later segments streamed', () => { + mountStream() + start() + interim('first commentary') + delta('second commentary') + interim('second commentary') + // Transport replay of the first event, well after its bubble was sealed: + interim('first commentary') + + const texts = assistantTexts() + expect(texts.filter(t => t === 'first commentary')).toHaveLength(1) + expect(texts).toContain('second commentary') + }) + + it('keeps one bubble for a whitespace-normalized restatement', () => { + mountStream() + start() + interim('checking the attach path') + interim('checking the attach\npath') + + const texts = assistantTexts() + // Count total bubbles, not matching texts: the raw copy must be dropped, + // not merely equal to a normalized expectation string. + expect(texts).toHaveLength(1) + expect(texts[0]).toBe('checking the attach path') + }) + + it('still appends genuinely new commentary after a suppression', () => { + mountStream() + start() + interim('first commentary') + interim('first commentary') + interim('a different observation') + + const texts = assistantTexts() + expect(texts.filter(t => t === 'first commentary')).toHaveLength(1) + expect(texts).toContain('a different observation') + }) + + it('does not suppress distinct commentary with a shared prefix', () => { + mountStream() + start() + interim('Checking the composer path for regressions.') + interim('Checking the composer path for regressions in v0.20.5.') + + const texts = assistantTexts() + expect(texts.filter(t => t.startsWith('Checking the composer path'))).toHaveLength(2) + }) +})