fix(chat): channel conversation sync + running indicator

- 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.
This commit is contained in:
matevip 2026-04-17 10:03:56 +08:00
parent b35c29767c
commit a3e6724cf4
2 changed files with 120 additions and 18 deletions

View File

@ -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<String, Object> 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<String, Object> 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;
}
// ==================== 审批重放 ====================

View File

@ -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)"
>
<div class="conv-icon">
<img :src="channelIconUrl(conv.source)" width="14" height="14" alt="" />
<span
v-if="conv.streamStatus === 'running'"
class="conv-running-dot"
:title="$t('chat.streamGenerating')"
></span>
</div>
<div v-if="!convPanelCollapsed || isMobile" class="conv-info">
<input
@ -80,7 +88,17 @@
@click.stop
ref="renameInputRef"
/>
<div v-else class="conv-title" @dblclick.stop="startRename(conv)">{{ conv.title }}</div>
<div v-else class="conv-title" @dblclick.stop="startRename(conv)">
<span>{{ conv.title }}</span>
<span
v-if="conv.streamStatus === 'running'"
class="conv-running-badge"
:title="$t('chat.streamGenerating')"
>
<span class="conv-running-badge-pulse"></span>
{{ $t('chat.streamGenerating') }}
</span>
</div>
<div class="conv-meta">
<span>{{ $t('chat.messages', { count: conv.messageCount }) }}</span>
<span class="conv-dot">·</span>
@ -877,10 +895,14 @@ function syncRouteState() {
async function selectConversation(conv: Conversation) {
if (isMobile.value) convPanelOpen.value = false
// reset stop observer
// UI/SSEresetForNewConversation stream.disconnect +
// POST /chat/{A}/stop A agent run
// ApollActivity / 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 {