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
+ * 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