feat(chat): realtime sync for external channel conversations

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 则委托渠道渲染流式事件
+            // 注册到 ChatStreamTracker：让 graph 节点广播的事件（phase / content_delta / tool_call_* 等）
-            if (adapter instanceof StreamingChannelAdapter streamingAdapter) {
+            // 能被 ChatConsole observer 订阅到。不注册 → broadcast() 会因 state==null 短路丢弃。
-                processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity);
+            // 同步 DB stream_status 让侧栏列表可以识别"该渠道对话正在运行"，从而触发 selectConversation 的 reconnect 分支。
-            } else {
+            streamTracker.register(conversationId);
-                // 同步路径：直接获取完整回复
+            streamTracker.incrementFlux(conversationId);
-                String reply = agentService.chat(agentId, promptText, conversationId);
+            conversationService.updateStreamStatus(conversationId, "running");
-
+            try {
-                // 检查 chat 过程中是否产生了审批 pending
+                // 流式路径：渠道实现了 StreamingChannelAdapter 则委托渠道渲染流式事件
-                PendingApproval newPending = approvalService.findPendingByConversation(conversationId);
+                if (adapter instanceof StreamingChannelAdapter streamingAdapter) {
-                if (newPending != null) {
+                    processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity);
-                    // 有审批需求：不保存 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);
+                    String reply = agentService.chat(agentId, promptText, conversationId);
-                    publishConversationCompletedEvent(agentId, conversationId, message.getContent(), reply);
-                    adapter.renderAndSend(replyTarget, reply);
-                    log.info("[{}] Reply sent to {}: {}chars",
-                            adapter.getChannelType(), replyTarget, reply.length());
 
-                    // 语音回复：异步 TTS 合成并追加发送（先文本后语音，不阻塞）
+                    // 检查 chat 过程中是否产生了审批 pending
-                    maybeGenerateVoiceReply(message, adapter, replyTarget, conversationId, reply, channelEntity);
+                    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());
                 }
             }
 

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

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<string>()
   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) {
+    if (lm) {
-      result.push(fm)
+      if (fm.role !== 'assistant') {
-    } else if (fm.role !== 'assistant') {
+        result.push(fm)
-      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(fm)
-      result.push(mergeAssistantMessages(lm, fm))
-      matchedLocalIds.add(fid)
     }
   }
 

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) {