From 92a7508dac30bcdde9c4e9ccec2d4c9191f1283f Mon Sep 17 00:00:00 2001 From: matevip Date: Thu, 30 Apr 2026 08:54:32 +0800 Subject: [PATCH] fix(channel): mirror plan-execute events to web sse for im-routed conversations --- .../mate/channel/ChannelMessageRouter.java | 87 ++++++++++++++++++- 1 file changed, 84 insertions(+), 3 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 4540e0b1..f1bc8772 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java @@ -24,9 +24,11 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.time.Duration; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.*; import java.util.concurrent.locks.ReentrantLock; @@ -89,6 +91,30 @@ public class ChannelMessageRouter { /** 防抖等待时间（毫秒） */ private static final long DEBOUNCE_MS = 500; + /** + * Plan-Execute SSE events that the Web Console mirror needs to see when + * a conversation runs through an IM channel. + *

+ * The agent emits these via {@code GraphEventPublisher} and they ride on + * the {@code chatStructuredStream} Flux as {@code StreamDelta.event(...)}. + * Web direct chats already broadcast them via the ChatController + * accumulator. IM channels (DingTalk + the seven sync-path adapters) + * historically dropped them — DingTalk's {@code processStreamAsText} + * only consumes {@code delta.content()}, and the sync {@code chat()} + * collector explicitly filters {@code delta.isEvent()} out. The whitelist + * is applied in the IM stream path so PlanStepsPanel renders correctly + * when an operator monitors an IM conversation in the Web Console. + *

+ * Whitelist (not pass-through) so Web-side accumulator-internal events + * like {@code _usage_final} or future agent-internal markers don't leak + * to subscribers. + */ + private static final Set MIRRORED_PLAN_EVENTS = Set.of( + "plan_created", + "plan_step_started", + "plan_step_completed" + ); + /** 是否已关闭 */ private volatile boolean shutdown = false; @@ -456,8 +482,31 @@ public class ChannelMessageRouter { if (adapter instanceof StreamingChannelAdapter streamingAdapter) { savedAssistantId = processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity, chatOrigin); } else { - // 同步路径：直接获取完整回复 - String reply = agentService.chat(agentId, promptText, conversationId, chatOrigin); + // Sync path for non-streaming IM adapters (feishu / wecom / weixin / + // slack / discord / qq / telegram). We can't use agentService.chat() + // because its collector filters out `delta.isEvent()` deltas — that + // would silently drop plan_created / plan_step_* events that the Web + // Console mirror needs to render PlanStepsPanel. Instead we consume + // chatStructuredStream directly: content gets accumulated for the IM + // reply, and whitelisted plan events are mirrored to ChatStreamTracker + // for any Web SSE viewer of the same conversationId. + StringBuilder replyAccumulator = new StringBuilder(); + final String channelType = adapter.getChannelType(); + agentService.chatStructuredStream(agentId, promptText, conversationId, + message.getSenderId(), chatOrigin) + .doOnNext(delta -> { + if (delta.isEvent()) { + mirrorPlanEventToTracker(conversationId, delta, channelType); + } else if (delta.content() != null) { + // Match the legacy agentService.chat() behavior: include + // persistOnly deltas too. DirectAnswerNode-routed answers + // arrive as persistOnly when CONTENT_STREAMED=true and IM + // channels still need the text for the outgoing reply. + replyAccumulator.append(delta.content()); + } + }) + .blockLast(Duration.ofMinutes(10)); + String reply = replyAccumulator.toString(); // 检查 chat 过程中是否产生了审批 pending PendingApproval newPending = approvalService.findPendingByConversation(conversationId); @@ -542,6 +591,30 @@ public class ChannelMessageRouter { * - StreamingChannelAdapter 负责渲染（AI Card / 卡片更新 / 文本累积等） * - Router 负责后续的审批检查、消息持久化、事件发布 */ + /** + * Forward whitelisted Plan-Execute SSE events to ChatStreamTracker so a + * Web Console viewer of an IM-routed conversation sees PlanStepsPanel. + *

+ * Bounded to {@link #MIRRORED_PLAN_EVENTS} — see the constant's javadoc + * for why this is a whitelist rather than a pass-through. Failures here + * are best-effort and never propagate, since dropping a UI update is + * preferable to derailing the channel reply. + */ + private void mirrorPlanEventToTracker(String conversationId, + AgentService.StreamDelta delta, + String channelTypeForLog) { + String eventType = delta.eventType(); + if (eventType == null || !MIRRORED_PLAN_EVENTS.contains(eventType)) { + return; + } + try { + streamTracker.broadcastObject(conversationId, eventType, delta.eventData()); + } catch (Exception ex) { + log.debug("[{}] Failed to mirror plan event {}: {}", + channelTypeForLog, eventType, ex.getMessage()); + } + } + private Long processWithStreaming(ChannelMessage message, StreamingChannelAdapter streamingAdapter, String conversationId, Long agentId, String promptText, ChannelEntity channelEntity, ChatOrigin chatOrigin) { @@ -553,8 +626,16 @@ public class ChannelMessageRouter { Flux stream = agentService.chatStructuredStream( agentId, promptText, conversationId, message.getSenderId(), chatOrigin); + // Mirror plan-execute SSE events to ChatStreamTracker before the + // adapter consumes the Flux. DingTalkChannelAdapter.processStreamAsText + // only reads `delta.content()` and would otherwise eat plan_created / + // plan_step_* events, leaving the Web Console mirror with no + // PlanStepsPanel for IM-routed conversations. + Flux mirroredStream = stream.doOnNext(delta -> + mirrorPlanEventToTracker(conversationId, delta, channelType)); + // Step 2: 委托渠道渲染（渠道内部消费 Flux 并处理 UI 更新） - String finalContent = streamingAdapter.processStream(stream, message, conversationId); + String finalContent = streamingAdapter.processStream(mirroredStream, message, conversationId); // Step 3: 审批检查 + 持久化（渠道无关逻辑，由 Router 统一处理） PendingApproval newPending = approvalService.findPendingByConversation(conversationId);