fix(chat): prevent cross-conversation message leakage and tool call duplication

mateclaw-ui/src/composables/chat/useChat.ts

@@ -129,6 +129,14 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   const heartbeat = ref<HeartbeatData | null>(null)
   /** Track which conversation the current stream belongs to */
   let streamConversationId = ''
+  /** 判断事件是否属于已过期的对话（防止旧流事件污染新会话） */
+  function isStaleEvent(data: any): boolean {
+    const eventConvId = data?.conversationId
+    if (eventConvId && streamConversationId && eventConvId !== streamConversationId) {
+      return true
+    }
+    return false
+  }
   /** 已处理的 approval pendingId 集合（幂等去重） */
   const processedApprovalIds = new Set<string>()
 
@@ -191,6 +199,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   // ===== SSE 事件处理器 =====
 
   stream.on('content_delta', (data) => {
+    if (isStaleEvent(data)) return
     if (currentAssistantId.value) {
       appendMessageContent(currentAssistantId.value, data.delta || '', 'text')
       if (['thinking', 'reasoning', 'drafting_answer', 'preparing_context'].includes(streamPhase.value)) {
@@ -212,6 +221,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('thinking_delta', (data) => {
+    if (isStaleEvent(data)) return
     if (currentAssistantId.value) {
       appendMessageContent(currentAssistantId.value, data.delta || '', 'thinking')
       if (streamPhase.value !== 'summarizing_observations') {
@@ -237,6 +247,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('message_start', (data) => {
+    if (isStaleEvent(data)) return
     if (data?.role !== 'assistant') return
     const currentMsg = currentAssistantId.value ? getMessage(currentAssistantId.value) : null
     if (currentMsg?.role === 'assistant') {
@@ -258,6 +269,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('warning', (data) => {
+    if (isStaleEvent(data)) return
     console.warn('[Chat] Warning from server:', data.delta || data.message || data)
     if (currentAssistantId.value) {
       const msg = getMessage(currentAssistantId.value)
@@ -274,6 +286,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('message_complete', (data) => {
+    if (isStaleEvent(data)) return
     if (currentAssistantId.value) {
       const msg = getMessage(currentAssistantId.value)
       if (msg?.status === 'failed') {
@@ -329,6 +342,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('done', (data) => {
+    if (isStaleEvent(data)) return
     console.log('[useChat] done event received:', {
       status: data.status,
       promptTokens: data.promptTokens,
@@ -386,6 +400,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
 
   let errorFired = false
   stream.on('error', (data) => {
+    if (isStaleEvent(data)) return
     const errorInfo: ChatErrorInfo = data.errorInfo
       || (data.errorType ? classifyBackendError(data) : { category: 'unknown', retryable: true, timestamp: Date.now() })
     if (currentAssistantId.value) {
@@ -421,6 +436,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   // ===== Agent 事件处理 =====
 
   stream.on('tool_call_started', (data) => {
+    if (isStaleEvent(data)) return
     streamPhase.value = 'executing_tool'
     if (currentAssistantId.value) {
       const msg = getMessage(currentAssistantId.value)
@@ -452,6 +468,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('tool_call_completed', (data) => {
+    if (isStaleEvent(data)) return
     if (currentAssistantId.value) {
       const msg = getMessage(currentAssistantId.value)
       if (msg) {
@@ -487,6 +504,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   // ===== Browser 执行事件 =====
 
   stream.on('browser_action', (data) => {
+    if (isStaleEvent(data)) return
     if (currentAssistantId.value) {
       const msg = getMessage(currentAssistantId.value)
       if (msg) {
@@ -510,6 +528,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('phase', (data) => {
+    if (isStaleEvent(data)) return
     const phase = data.phase as StreamPhase
     if (phase) {
       streamPhase.value = phase
@@ -530,6 +549,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('plan_created', (data) => {
+    if (isStaleEvent(data)) return
     if (currentAssistantId.value) {
       const msg = getMessage(currentAssistantId.value)
       if (msg) {
@@ -546,6 +566,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('plan_step_started', (data) => {
+    if (isStaleEvent(data)) return
     if (currentAssistantId.value) {
       const msg = getMessage(currentAssistantId.value)
       if (msg) {
@@ -564,6 +585,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('plan_step_completed', (data) => {
+    if (isStaleEvent(data)) return
     if (currentAssistantId.value) {
       const msg = getMessage(currentAssistantId.value)
       if (msg) {
@@ -587,6 +609,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   // ===== 工具审批事件（带幂等去重） =====
 
   stream.on('tool_approval_requested', (data) => {
+    if (isStaleEvent(data)) return
     // 幂等去重：同一 pendingId 只处理一次
     if (data.pendingId && processedApprovalIds.has(data.pendingId)) {
       console.log('[useChat] Duplicate approval request ignored:', data.pendingId)
@@ -641,6 +664,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('tool_approval_resolved', (data) => {
+    if (isStaleEvent(data)) return
     const targetMsg = messages.value.findLast((m) => {
       if (m.role !== 'assistant') return false
       const metadata = parseMetadata((m as any).metadata)
@@ -717,6 +741,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('turn_interrupted', (data) => {
+    if (isStaleEvent(data)) return
     console.log('[useChat] Turn interrupted, hasQueuedMessage:', data.hasQueuedMessage)
     // 当前 turn 已中断，等待后端自动启动排队消息
     // 如果后端会自动续跑，前端不需要做额外操作
@@ -734,6 +759,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
   })
 
   stream.on('queued_input_started', (data) => {
+    if (isStaleEvent(data)) return
     console.log('[useChat] Queued input started:', data.message)
     // 后端已开始处理排队消息
     // 1. 先用排队的内容创建用户消息（此时上一轮回答已完成，顺序正确）
@@ -753,6 +779,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
 
   // ===== 异步任务完成事件（视频生成、图片生成等） =====
   stream.on('async_task_completed', (data) => {
+    if (isStaleEvent(data)) return
     console.log('[useChat] Async task completed:', data)
     if (data.success && streamConversationId) {
       let mediaPart: MessageContentPart | null = null
@@ -853,6 +880,11 @@ export function useChat(options: UseChatOptions): UseChatReturn {
       clearTimeout(stopFallbackTimer)
       stopFallbackTimer = null
     }
+    // 切换会话时断开旧流，防止旧事件污染新会话
+    if (streamConversationId && streamConversationId !== conversationId) {
+      stream.disconnect()
+      currentAssistantId.value = null
+    }
     error.value = null
     errorFired = false
     streamConversationId = conversationId

mateclaw-ui/src/utils/messageReconcile.ts

@@ -124,27 +124,30 @@ function mergeMetadata(localMetaRaw: any, fetchedMetaRaw: any): Record<string, a
 function mergeAssistantMessages(localMsg: Message, fetchedMsg: Message): Message {
   const localRichness = messageRichness(localMsg)
   const fetchedRichness = messageRichness(fetchedMsg)
-  const richerMsg = localRichness >= fetchedRichness ? localMsg : fetchedMsg
+  const richer = localRichness >= fetchedRichness ? localMsg : fetchedMsg
+  const poorer = richer === localMsg ? fetchedMsg : localMsg
+
+  // 取 richer 的 contentParts 而非合并两边，避免 tool_call 因序列化差异导致重复
+  const localParts = Array.isArray(localMsg.contentParts) ? localMsg.contentParts : []
+  const fetchedParts = Array.isArray(fetchedMsg.contentParts) ? fetchedMsg.contentParts : []
+  const contentParts = localParts.length >= fetchedParts.length ? localParts : fetchedParts
 
   return {
-    ...fetchedMsg,
+    ...poorer,
+    ...richer,
+    // 用 fetched 的持久化字段覆盖（id、status、tokens 等）
+    id: fetchedMsg.id || localMsg.id,
     content: (fetchedMsg.content?.length || 0) >= (localMsg.content?.length || 0)
       ? fetchedMsg.content
       : localMsg.content,
-    contentParts: mergeContentParts(
-      Array.isArray(localMsg.contentParts) ? localMsg.contentParts : [],
-      Array.isArray(fetchedMsg.contentParts) ? fetchedMsg.contentParts : [],
-    ),
+    contentParts,
     metadata: mergeMetadata(localMsg.metadata, fetchedMsg.metadata),
-    attachments: fetchedMsg.attachments?.length ? fetchedMsg.attachments : localMsg.attachments,
-    errorInfo: fetchedMsg.errorInfo || localMsg.errorInfo,
-    thinkingExpanded: localMsg.thinkingExpanded ?? fetchedMsg.thinkingExpanded,
     status: fetchedMsg.status || localMsg.status,
     promptTokens: fetchedMsg.promptTokens ?? localMsg.promptTokens,
     completionTokens: fetchedMsg.completionTokens ?? localMsg.completionTokens,
     createTime: fetchedMsg.createTime || localMsg.createTime,
     conversationId: fetchedMsg.conversationId || localMsg.conversationId,
-    role: richerMsg.role,
+    thinkingExpanded: localMsg.thinkingExpanded ?? fetchedMsg.thinkingExpanded,
   }
 }
 
@@ -185,9 +188,17 @@ export function reconcileMessages(local: Message[], fetched: Message[]): Message
   }
 
   // 保留 fetched 中不存在的本地 assistant 消息（防止 lagging snapshot 丢弃刚完成的消息）
+  // 推断当前对话 ID：取 fetched 中第一条消息的 conversationId
+  const fetchedConversationId = fetched.length > 0 ? (fetched[0] as any).conversationId : ''
   for (const lm of local) {
     const lid = String(lm.id)
     if (!matchedLocalIds.has(lid) && lm.role === 'assistant') {
+      // 跳过不属于当前对话的本地消息，防止跨对话污染
+      // 无 conversationId 的 orphan 消息也不保留
+      const lmConvId = (lm as any).conversationId
+      if (!lmConvId || (fetchedConversationId && lmConvId !== fetchedConversationId)) {
+        continue
+      }
       // 检查是否是 fetched 末尾之后的消息（刚完成，DB 还没返回）
       const lastFetchedTime = result.length > 0 ? result[result.length - 1].createTime : ''
       if (!lastFetchedTime || (lm.createTime && lm.createTime >= lastFetchedTime)) {

mateclaw-ui/src/views/ChatConsole.vue

@@ -777,8 +777,12 @@ async function refreshCurrentConversationMessages(conversationId: string) {
   try {
     const res: any = await conversationApi.listMessages(conversationId)
     const fetched = extractMessages(res).messages.map((msg: Message) => normalizeMessage(msg))
+    // 过滤掉不属于当前对话的本地消息，防止跨对话污染
+    const currentMessages = messages.value.filter(
+      (m: any) => !m.conversationId || m.conversationId === conversationId
+    )
     // 逐条 reconcile：只接受更丰富的版本，防止 poorer DB 快照覆盖 local rich message
-    messages.value = reconcileMessages(messages.value, fetched)
+    messages.value = reconcileMessages(currentMessages, fetched)
   } catch (e) {
     console.warn('[ChatConsole] Failed to refresh current conversation messages:', e)
   }

text.txt

@@ -0,0 +1,77 @@
+请基于以下已确认事实，分析并修复前端 Chat 会话串线问题。不要泛泛而谈，直接围绕时序、状态隔离、消息归属和可验证修复方案展开。
+
+问题背景：
+
+用户连续两次问了同一句话“你有记忆里有啥”，系统创建了两个不同的 conversationId，并且后端日志显示这两个会话都是独立、正常完成的：
+
+1. 第一次会话：
+- conversationId: conv_1775859743291_z9pav7
+- SSE chat 建立时间：2026-04-11 06:22:32
+- user message 已落库
+- assistant message 已落库
+- done 已发送
+- stream fully completed
+
+2. 第二次会话：
+- conversationId: conv_1775859773188_335bjk
+- SSE chat 建立时间：2026-04-11 06:22:55
+- user message 已落库
+- assistant message 已落库
+- done 已发送
+- stream fully completed
+
+关键信号：
+
+- 两次请求是两个不同 conversationId。
+- 服务端日志没有显示 approval / awaiting_approval / interrupt / queued_input 相关链路。
+- 第一个会话已经完成后，前端仍然发了一次 stop，请求日志为 `stopped=false`，这说明 stop 到达时旧流已经结束，不是服务端还在跑旧流。
+- 因此，这更像是前端本地状态污染、会话切换时序竞争、或 reconcile 逻辑把旧本地消息错误带入新会话，而不是后端把旧会话内容串到了新会话。
+
+当前高优先级怀疑点：
+
+1. 切换会话 / 新建会话时，只调用了 stopChatGeneration()，但没有等待旧 SSE 流和本地状态完全清理。
+- 这会导致旧流晚到的事件（delta / done / error）在新会话已经创建 assistant 占位消息后，继续命中新会话的共享状态。
+- useChat 内部当前使用共享的 `currentAssistantId`、`streamConversationId`、`messages`，如果不做 conversation 级别隔离，就有天然串线风险。
+
+2. 审批占位 assistant message 在某些路径下创建后没有 conversationId。
+- 当前 refresh / reconcile 的过滤逻辑对 `!conversationId` 的本地消息仍可能放行。
+- 这类 orphan message 可能被错误并入后续任意会话。
+
+3. reconcileMessages() 当前有“保留 fetched 中不存在的本地 assistant 消息”的策略。
+- 这个策略本来是为了防止 lagging snapshot 丢刚完成的 rich message。
+- 但如果 local 里混入了旧会话消息、orphan message、或者未彻底清理的占位消息，就会把错误消息保留下来。
+
+你的任务：
+
+1. 先明确判断：
+- 根因是否主要在前端状态管理，而非后端 conversation/message 落库。
+- 哪一条最可能导致“上一轮消息出现在新会话”。
+
+2. 给出修复方案，要求具体到代码层面：
+- 会话切换 / 新建会话时，如何确保旧流彻底解绑。
+- 如何避免旧流事件写入当前会话。
+- `currentAssistantId` / `streamConversationId` 是否应该按 conversation 隔离，还是至少在事件处理时校验 conversationId。
+- 所有本地新建 message 是否必须强制携带 conversationId。
+- reconcileMessages() 是否应该完全禁止保留非当前 conversation 的本地消息。
+- 对 `conversationId` 为空的本地消息，应该如何处理。
+
+3. 给出建议的防御性约束：
+- 每个 SSE 事件落地前必须校验所属 conversationId。
+- onStreamEnd / reconnect / refreshCurrentConversationMessages 只能作用于当前会话。
+- 新会话开始前，旧会话的 placeholder / generating message 必须被清理或隔离。
+
+4. 输出格式要求：
+- 先给“根因判断”。
+- 再给“最小修复方案”。
+- 再给“更稳妥的长期方案”。
+- 最后给“如何验证修复有效”，至少覆盖：
+  - 连续快速新建会话并发送相同问题
+  - 旧会话刚结束时立刻切新会话
+  - refreshCurrentConversationMessages 在流结束后执行
+  - orphan assistant message / 空 conversationId message 不得污染新会话
+
+补充要求：
+
+- 不要只说“加锁”或“避免 race condition”，要明确到状态变量、事件处理器、过滤条件和消息生命周期。
+- 如果你认为某个现有修补不够，请直接指出为什么不够。
+- 如果需要改 reconcileMessages，请说明保留本地 assistant message 的边界条件。