From a3e6724cf4cd671d6145870b9e528c3ba03cd343 Mon Sep 17 00:00:00 2001 From: matevip Date: Fri, 17 Apr 2026 10:03:56 +0800 Subject: [PATCH] fix(chat): channel conversation sync + running indicator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ChannelMessageRouter: include assistantMessageId in message_complete and done broadcasts so ChatConsole observers can reconcile the streaming placeholder to the persisted DB row by id instead of falling back to the FIFO 'claim' heuristic (which occasionally dropped the assistant bubble on external channel conversations). Capture the id from ConversationService.saveMessage in both the sync agentService.chat path and the streaming processWithStreaming path; switch to HashMap since Map.of rejects null values when save is skipped (e.g. under approval). - ChatConsole: add a 'running' indicator on the sidebar so users can tell which conversations have an in-flight agent run. Pulsing amber dot on the channel icon (both expanded and collapsed modes) plus a '生成中…' / 'Generating…' pill in expanded mode. - ChatConsole: don't cancel the previous conversation's streaming run when switching conversations — let it keep running in the background and reconcile when the user comes back. --- .../mate/channel/ChannelMessageRouter.java | 44 ++++++--- mateclaw-ui/src/views/ChatConsole.vue | 94 ++++++++++++++++++- 2 files changed, 120 insertions(+), 18 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 d4be0191..d334833a 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java @@ -15,6 +15,7 @@ import vip.mate.memory.event.ConversationCompletedEvent; import vip.mate.tts.TtsService; import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.model.MessageContentPart; +import vip.mate.workspace.conversation.model.MessageEntity; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; @@ -22,6 +23,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.*; @@ -427,10 +429,15 @@ public class ChannelMessageRouter { streamTracker.register(conversationId); streamTracker.incrementFlux(conversationId); conversationService.updateStreamStatus(conversationId, "running"); + // 捕获已保存 assistant 的 DB id，供 finally 的 done 事件带出来 —— 前端 useChat.done 处理器 + // 依赖 data.assistantMessageId 把本地流式 placeholder 的 client-uuid 升级为 DB id，这样 + // 紧接着 refreshCurrentConversationMessages 的 reconcile 能按 id 干净匹配，不会走"认领"兜底 + // 导致气泡丢失。 + Long savedAssistantId = null; try { // 流式路径：渠道实现了 StreamingChannelAdapter 则委托渠道渲染流式事件 if (adapter instanceof StreamingChannelAdapter streamingAdapter) { - processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity); + savedAssistantId = processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity); } else { // 同步路径：直接获取完整回复 String reply = agentService.chat(agentId, promptText, conversationId); @@ -445,7 +452,8 @@ public class ChannelMessageRouter { adapter.getChannelType(), newPending.getToolName()); } else { // 正常回复：保存并发送 - conversationService.saveMessage(conversationId, "assistant", reply); + MessageEntity saved = conversationService.saveMessage(conversationId, "assistant", reply); + savedAssistantId = saved != null ? saved.getId() : null; publishConversationCompletedEvent(agentId, conversationId, message.getContent(), reply); adapter.renderAndSend(replyTarget, reply); log.info("[{}] Reply sent to {}: {}chars", @@ -453,21 +461,29 @@ public class ChannelMessageRouter { // 给 observer 推送完整的 assistant 消息以便前端一次性渲染为气泡 // （在 content_delta 事件可能没开的同步路径下作为兜底） - streamTracker.broadcastObject(conversationId, "message_complete", Map.of( - "conversationId", conversationId, - "content", reply, - "timestamp", System.currentTimeMillis())); + Map msgCompletePayload = new HashMap<>(); + msgCompletePayload.put("conversationId", conversationId); + msgCompletePayload.put("content", reply); + if (savedAssistantId != null) { + msgCompletePayload.put("assistantMessageId", savedAssistantId); + } + msgCompletePayload.put("timestamp", System.currentTimeMillis()); + streamTracker.broadcastObject(conversationId, "message_complete", msgCompletePayload); // 语音回复：异步 TTS 合成并追加发送（先文本后语音，不阻塞） maybeGenerateVoiceReply(message, adapter, replyTarget, conversationId, reply, channelEntity); } } } finally { - // 广播 done 事件让 observer 前端收尾 - streamTracker.broadcastObject(conversationId, "done", Map.of( - "conversationId", conversationId, - "status", "completed", - "timestamp", System.currentTimeMillis())); + // 广播 done 事件让 observer 前端收尾（HashMap 允许 null 值缺失，Map.of 不允许） + Map donePayload = new HashMap<>(); + donePayload.put("conversationId", conversationId); + donePayload.put("status", "completed"); + if (savedAssistantId != null) { + donePayload.put("assistantMessageId", savedAssistantId); + } + donePayload.put("timestamp", System.currentTimeMillis()); + streamTracker.broadcastObject(conversationId, "done", donePayload); streamTracker.completeAndConsumeIfLast(conversationId); try { conversationService.updateStreamStatus(conversationId, "idle"); @@ -499,7 +515,7 @@ public class ChannelMessageRouter { * - StreamingChannelAdapter 负责渲染（AI Card / 卡片更新 / 文本累积等） * - Router 负责后续的审批检查、消息持久化、事件发布 */ - private void processWithStreaming(ChannelMessage message, StreamingChannelAdapter streamingAdapter, + private Long processWithStreaming(ChannelMessage message, StreamingChannelAdapter streamingAdapter, String conversationId, Long agentId, String promptText, ChannelEntity channelEntity) { String channelType = streamingAdapter.getChannelType(); @@ -521,7 +537,7 @@ public class ChannelMessageRouter { log.info("[{}] Approval triggered during streaming (NOT saved to DB): tool={}", channelType, newPending.getToolName()); } else if (finalContent != null && !finalContent.isBlank()) { - conversationService.saveMessage(conversationId, "assistant", finalContent); + MessageEntity saved = conversationService.saveMessage(conversationId, "assistant", finalContent); publishConversationCompletedEvent(agentId, conversationId, promptText, finalContent); log.info("[{}] Streaming completed: contentLen={}", channelType, finalContent.length()); @@ -531,6 +547,7 @@ public class ChannelMessageRouter { maybeGenerateVoiceReply(message, streamingAdapter, replyTarget, conversationId, finalContent, channelEntity); } + return saved != null ? saved.getId() : null; } } catch (Exception e) { @@ -543,6 +560,7 @@ public class ChannelMessageRouter { log.error("[{}] Failed to send streaming error message: {}", channelType, sendErr.getMessage()); } } + return null; } // ==================== 审批重放 ==================== diff --git a/mateclaw-ui/src/views/ChatConsole.vue b/mateclaw-ui/src/views/ChatConsole.vue index f70663fb..676f192d 100644 --- a/mateclaw-ui/src/views/ChatConsole.vue +++ b/mateclaw-ui/src/views/ChatConsole.vue @@ -63,11 +63,19 @@ v-for="conv in group.items" :key="conv.conversationId" class="conv-item" - :class="{ active: currentConversationId === conv.conversationId }" + :class="{ + active: currentConversationId === conv.conversationId, + 'is-running': conv.streamStatus === 'running', + }" @click="selectConversation(conv)" >
+
-
{{ conv.title }}
+
+ {{ conv.title }} + + + {{ $t('chat.streamGenerating') }} + +
{{ $t('chat.messages', { count: conv.messageCount }) }} · @@ -877,10 +895,14 @@ function syncRouteState() { async function selectConversation(conv: Conversation) { if (isMobile.value) convPanelOpen.value = false - // 切换到不同会话时才 reset（含 stop 旧流）；点同一个会话时只刷新状态，避免误杀正在跑的 observer 流 + // 切换到不同会话：只清理本地 UI/SSE（resetForNewConversation 会 stream.disconnect + 清变量）， + // 但不 POST /chat/{A}/stop —— 让 A 的后台 agent run 跑到完成。 + // 用户之后回到 A：pollActivity / selectConversation 的 /status 探测会自动 reconnect 接回实时流； + // 若 A 已完成，refreshCurrentConversationMessages 会从 DB 拉完整结果。 + // 点同一个会话则完全不 reset，避免打断正在观察的流。 const switchingAway = currentConversationId.value !== conv.conversationId if (switchingAway) { - resetStreamingState() + resetForNewConversation() } currentConversationId.value = conv.conversationId selectedAgentId.value = conv.agentId || selectedAgentId.value @@ -1740,12 +1762,64 @@ function handleCodeCopy(e: MouseEvent) { .conv-icon { color: var(--mc-text-tertiary); flex-shrink: 0; + position: relative; } .conv-item.active .conv-icon { color: var(--mc-primary); } +/* 正在执行：图标右上角脉冲小点（折叠与展开态均可见） */ +.conv-running-dot { + position: absolute; + top: -2px; + right: -2px; + width: 7px; + height: 7px; + border-radius: 50%; + background: #fbbf24; + box-shadow: 0 0 4px rgba(251, 191, 36, 0.6), 0 0 0 2px var(--mc-bg-primary, #fff); + animation: pulse-dot 1.2s infinite; + pointer-events: none; +} + +.conv-item.is-running { + background: color-mix(in srgb, #fbbf24 8%, transparent); +} + +.conv-item.is-running:hover { + background: color-mix(in srgb, #fbbf24 14%, var(--mc-bg-sunken)); +} + +.conv-item.is-running.active { + background: var(--mc-primary-bg); +} + +/* 展开态：标题右侧"生成中..."小徽章 */ +.conv-running-badge { + display: inline-flex; + align-items: center; + gap: 4px; + flex-shrink: 0; + font-size: 10px; + font-weight: 500; + color: #b45309; + background: rgba(251, 191, 36, 0.15); + border: 1px solid rgba(251, 191, 36, 0.3); + padding: 1px 6px 1px 5px; + border-radius: 10px; + line-height: 1.3; + white-space: nowrap; +} + +.conv-running-badge-pulse { + width: 6px; + height: 6px; + border-radius: 50%; + background: #f59e0b; + animation: pulse-dot 1.2s infinite; +} + .conv-info { flex: 1; overflow: hidden; @@ -1755,9 +1829,19 @@ function handleCodeCopy(e: MouseEvent) { font-size: 13px; font-weight: 500; color: var(--mc-text-primary); - white-space: nowrap; + display: flex; + align-items: center; + gap: 6px; + min-width: 0; +} + +/* 标题文本本身承担省略号；flex 父级上的 overflow:hidden 会阻止 ellipsis 正常工作 */ +.conv-title > span:first-child { overflow: hidden; text-overflow: ellipsis; + white-space: nowrap; + min-width: 0; + flex: 1 1 auto; } .conv-item.active .conv-title {