From c41bda0f953cda1c98550ec40d2d5464585aa165 Mon Sep 17 00:00:00 2001 From: matevip Date: Sat, 11 Apr 2026 06:48:14 +0800 Subject: [PATCH] fix(chat): prevent cross-conversation message leakage and tool call duplication --- mateclaw-ui/src/composables/chat/useChat.ts | 32 +++++++++ mateclaw-ui/src/utils/messageReconcile.ts | 31 ++++++--- mateclaw-ui/src/views/ChatConsole.vue | 6 +- text.txt | 77 +++++++++++++++++++++ 4 files changed, 135 insertions(+), 11 deletions(-) create mode 100644 text.txt diff --git a/mateclaw-ui/src/composables/chat/useChat.ts b/mateclaw-ui/src/composables/chat/useChat.ts index f704f4c8..7f2c998c 100644 --- a/mateclaw-ui/src/composables/chat/useChat.ts +++ b/mateclaw-ui/src/composables/chat/useChat.ts @@ -129,6 +129,14 @@ export function useChat(options: UseChatOptions): UseChatReturn { const heartbeat = ref(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() @@ -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 diff --git a/mateclaw-ui/src/utils/messageReconcile.ts b/mateclaw-ui/src/utils/messageReconcile.ts index eb66fb92..1536a9a6 100644 --- a/mateclaw-ui/src/utils/messageReconcile.ts +++ b/mateclaw-ui/src/utils/messageReconcile.ts @@ -124,27 +124,30 @@ function mergeMetadata(localMetaRaw: any, fetchedMetaRaw: any): Record= 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)) { diff --git a/mateclaw-ui/src/views/ChatConsole.vue b/mateclaw-ui/src/views/ChatConsole.vue index 4e315f66..f4dae3c7 100644 --- a/mateclaw-ui/src/views/ChatConsole.vue +++ b/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) } diff --git a/text.txt b/text.txt new file mode 100644 index 00000000..84398fb8 --- /dev/null +++ b/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 的边界条件。