修复流式回复错位到上一轮消息

src/client/stores/chat.ts

@@ -733,6 +733,9 @@ export const useChatStore = defineStore('chat', () => {
     await socket.request('session.interrupt', { session_id: current.runtimeSessionId! }, 15_000)
     current.isStreaming = false
     current.isQueued = false
+    current.messages = current.messages.map(message => message.role === 'assistant' && message.isStreaming
+      ? { ...message, stage: message.stage === 'streaming' ? 'settled' : message.stage, isStreaming: false }
+      : message)
     clearInflightMarker(current.route.profile, current.route.sessionId)
   }
 

src/client/utils/messageReducer.ts

@@ -75,8 +75,28 @@ function eventMessageId(payload: Record<string, unknown>, fallback: string): str
   return string(payload.message_id ?? payload.messageId ?? payload.id ?? payload.run_id ?? payload.runId, fallback)
 }
 
-function lastStreamingAssistant(messages: ChatMessage[]): ChatMessage | undefined {
-  return [...messages].reverse().find(message => message.role === 'assistant' && message.isStreaming)
+function lastUserMessage(messages: ChatMessage[]): ChatMessage | undefined {
+  return [...messages].reverse().find(message => message.role === 'user')
+}
+
+function lastStreamingAssistant(messages: ChatMessage[], afterLatestUser = false): ChatMessage | undefined {
+  let latestUserIndex = -1
+  if (afterLatestUser) {
+    for (let index = messages.length - 1; index >= 0; index -= 1) {
+      if (messages[index].role === 'user') {
+        latestUserIndex = index
+        break
+      }
+    }
+  }
+  return messages.slice(latestUserIndex + 1).reverse()
+    .find(message => message.role === 'assistant' && message.isStreaming)
+}
+
+function settleSupersededStreams(messages: ChatMessage[], current?: ChatMessage): ChatMessage[] {
+  return messages.map(message => message.role === 'assistant' && message.isStreaming && message !== current
+    ? { ...message, stage: message.stage === 'streaming' ? 'settled' : message.stage, isStreaming: false }
+    : message)
 }
 
 function toolFromEvent(payload: Record<string, unknown>, status: ToolCall['status']): ToolCall {
@@ -111,8 +131,13 @@ function updateStreamingMessage(
   payload: Record<string, unknown>,
   mode: 'start' | 'delta' | 'reasoning' | 'complete' | 'failed',
 ): ChatMessage[] {
-  const current = lastStreamingAssistant(state.messages)
-  const fallbackId = current?.id ?? `stream:${state.route.profile}:${state.route.sessionId}`
+  // A start event belongs to the turn after the latest user row. An interrupted
+  // 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 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)
@@ -124,6 +149,10 @@ function updateStreamingMessage(
   if (mode === 'reasoning') reasoning = `${reasoning ?? ''}${delta || full}`
   const message: ChatMessage = {
     id,
+    // Keep a stable local identity while Hermes promotes a synthetic stream ID
+    // to its authoritative message/run ID on a later event.
+    clientMessageId: existing?.clientMessageId
+      ?? (existing?.id.startsWith('stream:') ? existing.id : id.startsWith('stream:') ? id : undefined),
     serverMessageId: id.startsWith('stream:') ? undefined : id,
     sessionId: state.route.sessionId,
     profile: state.route.profile,
@@ -137,7 +166,8 @@ function updateStreamingMessage(
     toolCalls: existing?.toolCalls,
     raw: payload as JsonValue,
   }
-  return mergeChatMessages(state.messages, [message])
+  const messages = mode === 'start' ? settleSupersededStreams(state.messages, current) : state.messages
+  return mergeChatMessages(messages, [message])
 }
 
 export function applyChatEvent(state: ChatRouteState, event: RpcEventFrame['params']): ChatRouteState {

tests/client/chatStore.test.ts

@@ -91,6 +91,32 @@ describe('chat message reducer', () => {
     expect(current.usage?.totalTokens).toBe(30)
   })
 
+  it('starts a new assistant row after a user prompt when an interrupted row is still marked streaming', () => {
+    const interrupted = {
+      ...message('assistant-old', '上一轮残留'), role: 'assistant' as const, timestamp: 20,
+      stage: 'streaming' as const, isStreaming: true,
+    }
+    const prompt = { ...message('user-new', '再试试'), timestamp: 30, stage: 'accepted' as const }
+    const initial = { ...state(), messages: [interrupted, prompt] }
+
+    let current = applyChatEvent(initial, event('run.started', {}))
+    current = applyChatEvent(current, event('message.delta', { delta: '新一轮回答' }))
+
+    expect(current.messages).toHaveLength(3)
+    expect(current.messages.map(item => item.content)).toEqual(['上一轮残留', '再试试', '新一轮回答'])
+    expect(current.messages[0]).toMatchObject({ id: 'assistant-old', isStreaming: false })
+    expect(current.messages[2]).toMatchObject({ role: 'assistant', isStreaming: true })
+  })
+
+  it('promotes a synthetic streaming row to a server message id without duplicating it', () => {
+    const initial = { ...state(), messages: [{ ...message('user-new', '开始'), timestamp: 30 }] }
+    let current = applyChatEvent(initial, event('run.started', {}))
+    current = applyChatEvent(current, event('message.delta', { message_id: 'assistant-new', delta: '回答' }))
+
+    expect(current.messages).toHaveLength(2)
+    expect(current.messages[1]).toMatchObject({ id: 'assistant-new', content: '回答', isStreaming: true })
+  })
+
   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', {