修复流式分段输出回写旧消息

src/client/utils/messageReducer.ts

@@ -99,6 +99,47 @@ function settleSupersededStreams(messages: ChatMessage[], current?: ChatMessage)
     : message)
 }
 
+function streamingSegmentId(state: ChatRouteState): string {
+  const latestUser = lastUserMessage(state.messages)
+  return `stream:${state.route.profile}:${state.route.sessionId}:${latestUser?.id ?? 'orphan'}:${createId('segment')}`
+}
+
+function sealInterimMessage(state: ChatRouteState, payload: Record<string, unknown>): ChatMessage[] {
+  const content = string(payload.output ?? payload.text ?? payload.content)
+  if (!content.trim()) return state.messages
+
+  const current = lastStreamingAssistant(state.messages)
+  const occurredAt = number(payload.timestamp ?? payload.created_at, Date.now() / 1000)
+  const sealedId = current?.id ?? streamingSegmentId(state)
+  const sealed: ChatMessage = {
+    ...(current ?? {}),
+    id: sealedId,
+    clientMessageId: current?.clientMessageId ?? (sealedId.startsWith('stream:') ? sealedId : undefined),
+    sessionId: state.route.sessionId,
+    profile: state.route.profile,
+    role: 'assistant',
+    content,
+    timestamp: current?.timestamp ?? occurredAt,
+    stage: 'settled',
+    isStreaming: false,
+    raw: payload as JsonValue,
+  }
+  const nextId = streamingSegmentId(state)
+  const next: ChatMessage = {
+    id: nextId,
+    clientMessageId: nextId,
+    sessionId: state.route.sessionId,
+    profile: state.route.profile,
+    role: 'assistant',
+    content: '',
+    timestamp: occurredAt,
+    stage: 'streaming',
+    isStreaming: true,
+    raw: { stream_interim_continuation: true },
+  }
+  return mergeChatMessages(mergeChatMessages(state.messages, [sealed]), [next])
+}
+
 function toolFromEvent(payload: Record<string, unknown>, status: ToolCall['status']): ToolCall {
   return {
     id: string(payload.tool_call_id ?? payload.toolCallId ?? payload.tool_id ?? payload.id, createId('tool')),
@@ -135,13 +176,32 @@ function updateStreamingMessage(
   // turn may never receive a terminal event, so reusing any older streaming row
   // would make the new answer render above the new prompt.
   const current = lastStreamingAssistant(state.messages, mode === 'start')
-  const latestUser = lastUserMessage(state.messages)
-  const fallbackId = current?.id
-    ?? `stream:${state.route.profile}:${state.route.sessionId}:${latestUser?.id ?? 'orphan'}`
+  const fallbackId = current?.id ?? streamingSegmentId(state)
   const id = eventMessageId(payload, fallbackId)
   const existing = state.messages.find(message => message.id === id) ?? current
   const delta = string(payload.delta ?? payload.text_delta ?? payload.content_delta)
   const full = string(payload.output ?? payload.text ?? payload.content)
+  const emptyInterimContinuation = mode === 'complete'
+    && existing === current
+    && bool(record(current?.raw).stream_interim_continuation)
+    && !current?.content.trim()
+    && !current?.reasoning?.trim()
+    && !current?.toolCalls?.length
+  if (emptyInterimContinuation && current) {
+    const currentIndex = state.messages.indexOf(current)
+    const previous = state.messages.slice(0, currentIndex).reverse()
+      .find(message => message.role === 'assistant')
+    const previousContent = previous?.content.trim() ?? ''
+    const continuesPrevious = Boolean(full && previousContent
+      && (full === previousContent || full.startsWith(previousContent) || previousContent.startsWith(full)))
+    if (previous && (!full || bool(payload.response_previewed) || continuesPrevious)) {
+      return orderChatMessages(state.messages
+        .filter(message => message !== current)
+        .map(message => message === previous && full
+          ? { ...message, content: full, stage: 'settled', isStreaming: false, raw: payload as JsonValue }
+          : message))
+    }
+  }
   let content = existing?.content ?? ''
   let reasoning = existing?.reasoning
   if (mode === 'delta') content += delta || full
@@ -188,6 +248,13 @@ export function applyChatEvent(state: ChatRouteState, event: RpcEventFrame['para
       next.messages = updateStreamingMessage(state, payload, 'delta')
       next.isStreaming = true
       break
+    case 'message.interim':
+      // Hermes emits this after commentary that precedes a tool call. Seal the
+      // visible text in place and start a fresh segment, otherwise later deltas
+      // keep rewriting the older bubble until message.complete replaces it.
+      next.messages = sealInterimMessage(state, payload)
+      next.isStreaming = true
+      break
     case 'reasoning.delta':
       next.messages = updateStreamingMessage(state, payload, 'reasoning')
       next.isStreaming = true

src/client/utils/turnTrace.ts

@@ -63,6 +63,10 @@ export function buildMessageTimelineRows(messages: UiMessage[]): MessageTimeline
       if (segment.length && nextOwner !== owner) flush()
       owner = nextOwner
       segment.push(message)
+      // An interim/final assistant bubble is a real timeline boundary. Keep
+      // later running tools and text below it instead of hoisting their trace
+      // above the earlier visible message merely because the author matches.
+      if (hasVisibleMessage(message)) flush()
       continue
     }
     flush()

tests/client/chatStore.test.ts

@@ -117,6 +117,46 @@ describe('chat message reducer', () => {
     expect(current.messages[1]).toMatchObject({ id: 'assistant-new', content: '回答', isStreaming: true })
   })
 
+  it('seals interim commentary so later streamed text stays at the bottom of the turn', () => {
+    const initial = { ...state(), messages: [{ ...message('user-new', '检查项目'), timestamp: 100 }] }
+    let current = applyChatEvent(initial, event('message.start', {}))
+    current = applyChatEvent(current, event('message.delta', { text: '我先读取配置。', timestamp: 101 }))
+    current = applyChatEvent(current, event('message.interim', { text: '我先读取配置。', timestamp: 102 }))
+    current = applyChatEvent(current, event('tool.start', { tool_id: 'tool-1', name: 'read_file' }))
+    current = applyChatEvent(current, event('tool.complete', { tool_id: 'tool-1', name: 'read_file', result: 'ok' }))
+    current = applyChatEvent(current, event('message.delta', { text: '配置没有问题。', timestamp: 105 }))
+
+    expect(current.messages.map(item => item.content)).toEqual(['检查项目', '我先读取配置。', '配置没有问题。'])
+    expect(current.messages[1]).toMatchObject({ role: 'assistant', isStreaming: false })
+    expect(current.messages[2]).toMatchObject({ role: 'assistant', isStreaming: true })
+    expect(current.messages[2].toolCalls).toEqual([
+      expect.objectContaining({ id: 'tool-1', status: 'completed' }),
+    ])
+  })
+
+  it('preserves interim commentary after the final answer completes', () => {
+    const initial = { ...state(), messages: [{ ...message('user-new', '执行'), timestamp: 100 }] }
+    let current = applyChatEvent(initial, event('message.start', {}))
+    current = applyChatEvent(current, event('message.delta', { text: '开始执行。' }))
+    current = applyChatEvent(current, event('message.interim', { text: '开始执行。' }))
+    current = applyChatEvent(current, event('message.delta', { text: '执行完成。' }))
+    current = applyChatEvent(current, event('message.complete', { text: '执行完成。' }))
+
+    expect(current.messages.map(item => item.content)).toEqual(['执行', '开始执行。', '执行完成。'])
+    expect(current.messages.slice(1).every(item => item.isStreaming === false)).toBe(true)
+  })
+
+  it('settles an identical completion onto the sealed interim instead of duplicating it', () => {
+    const initial = { ...state(), messages: [{ ...message('user-new', '执行'), timestamp: 100 }] }
+    let current = applyChatEvent(initial, event('message.start', {}))
+    current = applyChatEvent(current, event('message.delta', { text: '已经完成。' }))
+    current = applyChatEvent(current, event('message.interim', { text: '已经完成。', already_streamed: true }))
+    current = applyChatEvent(current, event('message.complete', { text: '已经完成。' }))
+
+    expect(current.messages.map(item => item.content)).toEqual(['执行', '已经完成。'])
+    expect(current.messages[1]).toMatchObject({ stage: 'settled', isStreaming: false })
+  })
+
   it('treats message.complete status=error as a failed terminal row', () => {
     let current = applyChatEvent(state(), event('message.start', { message_id: 'assistant-1' }))
     current = applyChatEvent(current, event('message.complete', {

tests/client/toolTimeline.test.ts

@@ -69,6 +69,24 @@ describe('ordinary chat tool timeline', () => {
     expect(rows[2]?.kind === 'message' ? rows[2].message : undefined).toMatchObject({ content: '检查完成', reasoning: undefined, tools: undefined })
   })
 
+  it('keeps a later streaming trace below sealed interim commentary', () => {
+    const rows = buildMessageTimelineRows([
+      { id: 'user', role: 'user', content: '检查项目' },
+      { id: 'interim', role: 'assistant', content: '我先读取配置。', status: 'settled' },
+      {
+        id: 'streaming', role: 'assistant', content: '配置没有问题。', status: 'streaming',
+        tools: [{ id: 'read', name: 'read_file', status: 'running' }],
+      },
+    ])
+
+    expect(rows.map(row => [row.kind, row.id])).toEqual([
+      ['message', 'message:user'],
+      ['message', 'message:interim'],
+      ['trace', 'trace:streaming'],
+      ['message', 'message:streaming'],
+    ])
+  })
+
   it('does not merge assistant traces across agents or system events', () => {
     const rows = buildMessageTimelineRows([
       { id: 'a', role: 'assistant', author: '甲', content: '', reasoning: '甲思考' },

tests/e2e/workspace.spec.ts

@@ -708,6 +708,28 @@ test('shows a thinking animation after submit until output starts', async ({ pag
   await expect(page.locator('.thinking-indicator')).toHaveCount(0)
 })
 
+test('keeps streamed text below sealed interim commentary and its tool trace', async ({ page }) => {
+  await page.goto('/chat/session-demo?profile=yaoyao')
+  await page.locator('.composer-textarea').fill('验证流式分段')
+  await page.getByRole('button', { name: '发送消息' }).click()
+
+  const interim = page.locator('.message--assistant').filter({ hasText: '我先检查配置。' }).last()
+  const current = page.locator('.message--assistant').filter({ hasText: '配置检查完成。' }).last()
+  const trace = page.locator('.turn-trace').last()
+  await expect(interim).toBeVisible()
+  await expect(current).toBeVisible()
+  await expect(trace).toHaveClass(/turn-trace--running/)
+
+  const [interimBox, traceBox, currentBox] = await Promise.all([
+    interim.boundingBox(), trace.boundingBox(), current.boundingBox(),
+  ])
+  expect(interimBox?.y).toBeLessThan(traceBox?.y ?? 0)
+  expect(traceBox?.y).toBeLessThan(currentBox?.y ?? 0)
+
+  await expect(trace).toHaveClass(/turn-trace--success/)
+  await expect(page.getByText('配置检查完成。', { exact: true })).toHaveCount(1)
+})
+
 test('keeps the canonical logo and yaoyao-webui composer geometry', async ({ page }) => {
   const logo = page.locator('.rail__brand img')
   await expect(logo).toHaveAttribute('src', '/brand/AppIcon-1024.png')

tests/fixtures/fake-hermes.mjs

@@ -253,9 +253,25 @@ server.on('upgrade', (request, socket, head) => {
       if (requestFrame.method === 'prompt.submit') {
         respond({ status: 'accepted' })
         setTimeout(() => {
-          client.send(JSON.stringify({ jsonrpc: '2.0', method: 'event', params: { type: 'message.start', session_id: requestFrame.params.session_id, profile: 'yaoyao', payload: { message_id: 'stream-demo' } } }))
-          client.send(JSON.stringify({ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: requestFrame.params.session_id, profile: 'yaoyao', payload: { message_id: 'stream-demo', delta: '这是来自假 Gateway 的流式回复。' } } }))
-          client.send(JSON.stringify({ jsonrpc: '2.0', method: 'event', params: { type: 'message.complete', session_id: requestFrame.params.session_id, profile: 'yaoyao', payload: { message_id: 'stream-demo', text: '这是来自假 Gateway 的流式回复。', status: 'complete' } } }))
+          const emit = (type, payload) => client.send(JSON.stringify({
+            jsonrpc: '2.0', method: 'event',
+            params: { type, session_id: requestFrame.params.session_id, profile: 'yaoyao', ...(payload ? { payload } : {}) },
+          }))
+          if (requestFrame.params.text === '验证流式分段') {
+            emit('message.start')
+            emit('message.delta', { text: '我先检查配置。' })
+            emit('message.interim', { text: '我先检查配置。', already_streamed: true })
+            emit('tool.start', { tool_id: 'fake-tool', name: 'terminal', arguments: { command: 'pwd' } })
+            setTimeout(() => {
+              emit('tool.complete', { tool_id: 'fake-tool', name: 'terminal', result: { output: '/tmp' } })
+              emit('message.delta', { text: '配置检查完成。' })
+            }, 300)
+            setTimeout(() => emit('message.complete', { text: '配置检查完成。', status: 'complete' }), 2_000)
+            return
+          }
+          client.send(JSON.stringify({ jsonrpc: '2.0', method: 'event', params: { type: 'message.start', session_id: requestFrame.params.session_id, profile: 'yaoyao' } }))
+          client.send(JSON.stringify({ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: requestFrame.params.session_id, profile: 'yaoyao', payload: { text: '这是来自假 Gateway 的流式回复。' } } }))
+          client.send(JSON.stringify({ jsonrpc: '2.0', method: 'event', params: { type: 'message.complete', session_id: requestFrame.params.session_id, profile: 'yaoyao', payload: { text: '这是来自假 Gateway 的流式回复。', status: 'complete' } } }))
         }, 1200)
         return
       }