From b35c29767c382bbcf7050853a2c65c6ccea4b71c Mon Sep 17 00:00:00 2001 From: matevip Date: Fri, 17 Apr 2026 08:25:32 +0800 Subject: [PATCH] feat(chat): realtime sync for external channel conversations --- .../mate/channel/ChannelMessageRouter.java | 79 +++++++++++++------ mateclaw-ui/src/composables/chat/useChat.ts | 27 ++++++- mateclaw-ui/src/utils/messageReconcile.ts | 45 +++++++++-- mateclaw-ui/src/views/ChatConsole.vue | 69 +++++++++++++++- 4 files changed, 185 insertions(+), 35 deletions(-) diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java index 343a8580..d4be0191 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java @@ -9,6 +9,7 @@ import vip.mate.approval.PendingApproval; import vip.mate.channel.model.ChannelEntity; import vip.mate.channel.notification.ApprovalNotificationService; import vip.mate.channel.service.ChannelService; +import vip.mate.channel.web.ChatStreamTracker; import org.springframework.context.ApplicationEventPublisher; import vip.mate.memory.event.ConversationCompletedEvent; import vip.mate.tts.TtsService; @@ -50,6 +51,7 @@ public class ChannelMessageRouter { private final ApplicationEventPublisher eventPublisher; private final TtsService ttsService; private final ObjectMapper objectMapper; + private final ChatStreamTracker streamTracker; /** 队列条目：封装消息及其路由上下文 */ private record QueueEntry(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) {} @@ -93,7 +95,8 @@ public class ChannelMessageRouter { ApprovalNotificationService approvalNotificationService, ApplicationEventPublisher eventPublisher, TtsService ttsService, - ObjectMapper objectMapper) { + ObjectMapper objectMapper, + ChatStreamTracker streamTracker) { this.agentService = agentService; this.conversationService = conversationService; this.channelService = channelService; @@ -103,6 +106,7 @@ public class ChannelMessageRouter { this.eventPublisher = eventPublisher; this.ttsService = ttsService; this.objectMapper = objectMapper; + this.streamTracker = streamTracker; } // ==================== 防抖辅助类 ==================== @@ -417,31 +421,58 @@ public class ChannelMessageRouter { // 构建 prompt（语音输入时注入场景提示词） String promptText = buildPromptFromParts(message.getContent(), parts, message.getInputMode()); - // 流式路径：渠道实现了 StreamingChannelAdapter 则委托渠道渲染流式事件 - if (adapter instanceof StreamingChannelAdapter streamingAdapter) { - processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity); - } else { - // 同步路径：直接获取完整回复 - String reply = agentService.chat(agentId, promptText, conversationId); - - // 检查 chat 过程中是否产生了审批 pending - PendingApproval newPending = approvalService.findPendingByConversation(conversationId); - if (newPending != null) { - // 有审批需求：不保存 LLM 的审批占位回复到 DB，直接从 pending 元数据构建通知 - String approvalNotice = buildApprovalNotice(newPending); - adapter.renderAndSend(replyTarget, approvalNotice); - log.info("[{}] Approval triggered during chat, sent notice (NOT saved to DB): tool={}", - adapter.getChannelType(), newPending.getToolName()); + // 注册到 ChatStreamTracker：让 graph 节点广播的事件（phase / content_delta / tool_call_* 等） + // 能被 ChatConsole observer 订阅到。不注册 → broadcast() 会因 state==null 短路丢弃。 + // 同步 DB stream_status 让侧栏列表可以识别"该渠道对话正在运行"，从而触发 selectConversation 的 reconnect 分支。 + streamTracker.register(conversationId); + streamTracker.incrementFlux(conversationId); + conversationService.updateStreamStatus(conversationId, "running"); + try { + // 流式路径：渠道实现了 StreamingChannelAdapter 则委托渠道渲染流式事件 + if (adapter instanceof StreamingChannelAdapter streamingAdapter) { + processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity); } else { - // 正常回复：保存并发送 - conversationService.saveMessage(conversationId, "assistant", reply); - publishConversationCompletedEvent(agentId, conversationId, message.getContent(), reply); - adapter.renderAndSend(replyTarget, reply); - log.info("[{}] Reply sent to {}: {}chars", - adapter.getChannelType(), replyTarget, reply.length()); + // 同步路径：直接获取完整回复 + String reply = agentService.chat(agentId, promptText, conversationId); - // 语音回复：异步 TTS 合成并追加发送（先文本后语音，不阻塞） - maybeGenerateVoiceReply(message, adapter, replyTarget, conversationId, reply, channelEntity); + // 检查 chat 过程中是否产生了审批 pending + PendingApproval newPending = approvalService.findPendingByConversation(conversationId); + if (newPending != null) { + // 有审批需求：不保存 LLM 的审批占位回复到 DB，直接从 pending 元数据构建通知 + String approvalNotice = buildApprovalNotice(newPending); + adapter.renderAndSend(replyTarget, approvalNotice); + log.info("[{}] Approval triggered during chat, sent notice (NOT saved to DB): tool={}", + adapter.getChannelType(), newPending.getToolName()); + } else { + // 正常回复：保存并发送 + conversationService.saveMessage(conversationId, "assistant", reply); + publishConversationCompletedEvent(agentId, conversationId, message.getContent(), reply); + adapter.renderAndSend(replyTarget, reply); + log.info("[{}] Reply sent to {}: {}chars", + adapter.getChannelType(), replyTarget, reply.length()); + + // 给 observer 推送完整的 assistant 消息以便前端一次性渲染为气泡 + // （在 content_delta 事件可能没开的同步路径下作为兜底） + streamTracker.broadcastObject(conversationId, "message_complete", Map.of( + "conversationId", conversationId, + "content", reply, + "timestamp", System.currentTimeMillis())); + + // 语音回复：异步 TTS 合成并追加发送（先文本后语音，不阻塞） + maybeGenerateVoiceReply(message, adapter, replyTarget, conversationId, reply, channelEntity); + } + } + } finally { + // 广播 done 事件让 observer 前端收尾 + streamTracker.broadcastObject(conversationId, "done", Map.of( + "conversationId", conversationId, + "status", "completed", + "timestamp", System.currentTimeMillis())); + streamTracker.completeAndConsumeIfLast(conversationId); + try { + conversationService.updateStreamStatus(conversationId, "idle"); + } catch (Exception e) { + log.debug("Failed to reset stream_status for {}: {}", conversationId, e.getMessage()); } } diff --git a/mateclaw-ui/src/composables/chat/useChat.ts b/mateclaw-ui/src/composables/chat/useChat.ts index aa13d4f6..9cab6bc6 100644 --- a/mateclaw-ui/src/composables/chat/useChat.ts +++ b/mateclaw-ui/src/composables/chat/useChat.ts @@ -1118,6 +1118,17 @@ export function useChat(options: UseChatOptions): UseChatReturn { const convId = streamConversationId const assistantId = currentAssistantId.value + // 仅在前端确实在生成时才触发停止（含 SSE 接收中 / reconnect 中 / 审批等待中）。 + // 否则只是"旁观者"身份，不能把对方（渠道用户）的 agent run 也一起杀掉。 + const activelyStreaming = isGenerating.value + || streamPhase.value === 'reconnecting' + || streamPhase.value === 'awaiting_approval' + + if (!activelyStreaming) { + // 没有真正在流 → 什么都不做，让调用方直接走 resetForNewConversation + return + } + // 先取消排队消息 messageQueue.clear() @@ -1190,8 +1201,22 @@ export function useChat(options: UseChatOptions): UseChatReturn { errorFired = false phaseInfo.value = null - // 创建 assistant 占位消息用于接收重连后的流数据 resetCurrentTurnState() + + // 清理尾部的空 assistant 消息（来自上一轮被误杀的 run 留下的空壳，或 placeholder 遗留）， + // 避免与即将重连产生的 streaming 气泡共存形成"重复两条"假象。 + while (messages.value.length > 0) { + const tail = messages.value[messages.value.length - 1] + if (tail && tail.role === 'assistant' + && tail.conversationId === conversationId + && !tail.content + && (!tail.contentParts || tail.contentParts.length === 0)) { + messages.value.pop() + } else { + break + } + } + const assistantMessage = createAssistantMessage('', conversationId) ;(assistantMessage as any)._turnId = activeTurnId currentAssistantId.value = assistantMessage.id as string diff --git a/mateclaw-ui/src/utils/messageReconcile.ts b/mateclaw-ui/src/utils/messageReconcile.ts index feff58cd..66e8c033 100644 --- a/mateclaw-ui/src/utils/messageReconcile.ts +++ b/mateclaw-ui/src/utils/messageReconcile.ts @@ -174,6 +174,16 @@ function mergeAssistantMessages(localMsg: Message, fetchedMsg: Message): Message * - assistant 消息：比较 richness，取更高分的版本 * - 本地有但 fetched 没有的 assistant 消息：保留（防止 lagging snapshot 丢消息） */ +/** + * 判断是否为"客户端临时 id"（UUID / 带字母的非数字 id）。 + * 服务端 id 是雪花算法生成的 Long（全数字字符串）。 + */ +function isClientId(id: any): boolean { + if (id === null || id === undefined) return true + const s = String(id) + return s === '' || !/^\d+$/.test(s) +} + export function reconcileMessages(local: Message[], fetched: Message[]): Message[] { if (!local.length) return fetched if (!fetched.length) return local @@ -186,19 +196,40 @@ export function reconcileMessages(local: Message[], fetched: Message[]): Message const matchedLocalIds = new Set() const result: Message[] = [] + // 收集未被 id 匹配过的本地 assistant（通常是流式产生的 client-uuid placeholder）， + // 供 fetched 端新 assistant"认领"它们的 timeline，避免两条并排。 + const unclaimedLocalAssistants: Message[] = [] + for (const lm of local) { + if (lm.role === 'assistant' && isClientId(lm.id)) { + unclaimedLocalAssistants.push(lm) + } + } + for (const fm of fetched) { const fid = String(fm.id) const lm = localMap.get(fid) - if (!lm) { - result.push(fm) - } else if (fm.role !== 'assistant') { - result.push(fm) + if (lm) { + if (fm.role !== 'assistant') { + result.push(fm) + } else { + // assistant 消息：不要整条覆盖，合并 fetched 的持久化字段与 local 的 richer timeline + result.push(mergeAssistantMessages(lm, fm)) + } matchedLocalIds.add(fid) + continue + } + + // fetched 里这条本地没有 + if (fm.role === 'assistant' && unclaimedLocalAssistants.length > 0) { + // 尝试"认领"本地一个 client-uuid placeholder： + // 取队首（最早的未认领），合并 richness 后采用 fetched 的持久化 id/时间。 + // 这消除了"本地流式气泡"+"刚落库的 DB assistant"同时存在的重复。 + const claimed = unclaimedLocalAssistants.shift()! + matchedLocalIds.add(String(claimed.id)) + result.push(mergeAssistantMessages(claimed, fm)) } else { - // assistant 消息：不要整条覆盖，合并 fetched 的持久化字段与 local 的 richer timeline - result.push(mergeAssistantMessages(lm, fm)) - matchedLocalIds.add(fid) + result.push(fm) } } diff --git a/mateclaw-ui/src/views/ChatConsole.vue b/mateclaw-ui/src/views/ChatConsole.vue index d4846791..f70663fb 100644 --- a/mateclaw-ui/src/views/ChatConsole.vue +++ b/mateclaw-ui/src/views/ChatConsole.vue @@ -684,6 +684,44 @@ function handleKeyboardShortcuts(e: KeyboardEvent) { } } +// 轮询定时器：让 ChatConsole 能实时感知外部渠道（WeChat/DingTalk/…）推进来的新消息， +// 无需 F5 即可看到侧栏列表更新和选中会话的消息/流状态。 +let activityPollTimer: number | null = null +const ACTIVITY_POLL_MS = 4000 + +async function pollActivity() { + // 页面不可见时不轮询，避免切到别的标签还在空耗 + if (typeof document !== 'undefined' && document.hidden) return + try { + await loadConversations() + } catch { + // 静默失败，下一轮再试 + } + // 自己没在生成时才刷新当前选中会话的消息 + 探测是否该接入流 + if (currentConversationId.value && !isGenerating.value && streamPhase.value !== 'awaiting_approval') { + const cid = currentConversationId.value + try { + const statusRes: any = await conversationApi.getStatus(cid) + if (currentConversationId.value !== cid) return + const running = statusRes?.data?.streamStatus === 'running' + if (running) { + // 外部渠道正在跑： + // 1. 先从 DB 拉消息，把刚插入的 user 消息（"你在干什么"之类）带进来， + // 否则只接入流的话前端只能看到 assistant content_delta，看不到用户问题。 + // 2. 再接入流，让后续 content_delta 实时累积到 assistant 气泡。 + await refreshCurrentConversationMessages(cid) + if (currentConversationId.value !== cid || isGenerating.value) return + await reconnectStream(cid) + } else { + // 不在跑：从 DB 对齐消息（新 user 消息 / 刚落库 assistant 会合并进来） + await refreshCurrentConversationMessages(cid) + } + } catch { + // 忽略探测失败 + } + } +} + onMounted(async () => { document.addEventListener('keydown', handleKeyboardShortcuts) document.addEventListener('click', handleCodeCopy) @@ -696,6 +734,7 @@ onMounted(async () => { mediumQuery.addEventListener('change', handleConvMediumChange) await Promise.all([loadAgents(), loadModelState(), loadConversations()]) await hydrateStateFromRoute() + activityPollTimer = window.setInterval(pollActivity, ACTIVITY_POLL_MS) }) onBeforeUnmount(() => { @@ -704,6 +743,10 @@ onBeforeUnmount(() => { disposeECharts() mobileQuery?.removeEventListener('change', handleMobileChange) mediumQuery?.removeEventListener('change', handleConvMediumChange) + if (activityPollTimer !== null) { + clearInterval(activityPollTimer) + activityPollTimer = null + } stopChatGeneration() // 释放所有附件的 ObjectURL，防止内存泄漏 revokeAllPreviewUrls() @@ -834,7 +877,11 @@ function syncRouteState() { async function selectConversation(conv: Conversation) { if (isMobile.value) convPanelOpen.value = false - resetStreamingState() + // 切换到不同会话时才 reset（含 stop 旧流）；点同一个会话时只刷新状态，避免误杀正在跑的 observer 流 + const switchingAway = currentConversationId.value !== conv.conversationId + if (switchingAway) { + resetStreamingState() + } currentConversationId.value = conv.conversationId selectedAgentId.value = conv.agentId || selectedAgentId.value const requestedConvId = conv.conversationId @@ -842,7 +889,10 @@ async function selectConversation(conv: Conversation) { const res: any = await conversationApi.listMessages(requestedConvId) // Stale guard：await 返回后确认仍是当前会话，否则丢弃 if (currentConversationId.value !== requestedConvId) return - messages.value = extractMessages(res).messages.map((msg: Message) => normalizeMessage(msg)) + // 点同一个会话时，若已有 SSE 在跑就不要覆盖本地消息状态 + if (switchingAway || !isGenerating.value) { + messages.value = extractMessages(res).messages.map((msg: Message) => normalizeMessage(msg)) + } // Hydrate pending approvals：恢复刷新后丢失的审批卡片 try { @@ -875,7 +925,20 @@ async function selectConversation(conv: Conversation) { // hydration 失败不影响正常使用 } - if (currentConversationId.value === requestedConvId && conv.streamStatus === 'running') { + // 决定是否重连 SSE： + // - 快照 streamStatus==='running' → 直接重连 + // - 否则探测实时状态（兜底处理：渠道消息进入后侧栏快照未刷新时，仍能接入运行中的流） + let shouldReconnect = conv.streamStatus === 'running' + if (!shouldReconnect) { + try { + const statusRes: any = await conversationApi.getStatus(requestedConvId) + if (currentConversationId.value !== requestedConvId) return + shouldReconnect = statusRes?.data?.streamStatus === 'running' + } catch { + // 探测失败不阻断主流程 + } + } + if (currentConversationId.value === requestedConvId && shouldReconnect) { await reconnectStream(requestedConvId) } } catch (e) {