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>
This commit is contained in:
@@ -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<string, unknown>]
|
||||
|
||||
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<string, unknown> = {}) =>
|
||||
act(() => stream.handleEvent({ type, payload, session_id: SID }))
|
||||
|
||||
return { stream, send }
|
||||
}
|
||||
|
||||
function expectSingleBubble(stream: ReturnType<typeof renderMessageStream>) {
|
||||
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()
|
||||
})
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<string, unknown> = {}): 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<string, unknown> = {}) =>
|
||||
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)
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user