From a183519d693c4b2c07a4dadc2b9aa2c8cb188322 Mon Sep 17 00:00:00 2001 From: matevip Date: Wed, 22 Jul 2026 18:09:59 +0800 Subject: [PATCH] feat(channel): extract the per-turn stream accumulator and share it between web SSE and IM sync paths --- .../mate/channel/ChannelMessageRouter.java | 193 +++---- .../channel/web/AgentStreamAccumulator.java | 530 ++++++++++++++++++ .../ChannelMessageRouterNarrationTest.java | 9 +- 3 files changed, 617 insertions(+), 115 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/channel/web/AgentStreamAccumulator.java 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 a3e20e21..7ee3d4af 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.channel.event.ChannelMessageReceivedEvent; import vip.mate.channel.model.ChannelEntity; import vip.mate.channel.notification.ApprovalNotificationService; import vip.mate.channel.service.ChannelService; +import vip.mate.channel.web.AgentStreamAccumulator; import vip.mate.channel.web.ChatStreamTracker; import vip.mate.exception.MateClawException; import vip.mate.memory.event.ConversationCompletionPublisher; @@ -157,30 +158,6 @@ public class ChannelMessageRouter { return currentMergedLength > LONG_TEXT_THRESHOLD ? LONG_DEBOUNCE_MS : DEBOUNCE_MS; } - /** - * 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; @@ -802,12 +779,10 @@ public class ChannelMessageRouter { // Sync path for non-streaming IM adapters (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(); + // would silently drop the tool/plan events the Web Console + // mirror and the persisted execution metadata both need. + // Instead we consume chatStructuredStream directly through + // the shared accumulator (reply text, metadata, live mirror). final String channelType = adapter.getChannelType(); // Channel-level toggle for relaying per-stage narration as // standalone messages mid-run. Shares the key the streaming @@ -817,35 +792,25 @@ public class ChannelMessageRouter { // still see it via the live broadcast). final boolean relayNarration = channelConfigBoolean( channelEntity, "stream_progress", true); - // Token usage + model attribution: capture _usage_final event emitted at stream end - final int[] usage = {0, 0, 0, 0, 0}; // [prompt, completion, cacheRead, cacheWrite, reasoning] - final String[] modelInfo = {null, null}; // [runtimeModel, runtimeProvider] + // Shared accumulator: builds the segments/toolCalls metadata + // the Web console renders for history, mirrors events + + // deltas to live Web observers, and captures token usage + + // model attribution (_usage_final is consumed internally). + // Reply text also comes from it — same semantics as the + // legacy collector: persistOnly deltas included (DirectAnswerNode- + // routed answers arrive as persistOnly when CONTENT_STREAMED=true + // and IM channels still need the text for the outgoing reply), + // segmentOnly narration excluded (issue #120). + AgentStreamAccumulator accumulator = newAccumulator(); agentService.chatStructuredStream(agentId, promptText, conversationId, message.getSenderId(), chatOrigin) .doOnNext(delta -> { - if (delta.isEvent()) { - if ("_usage_final".equals(delta.eventType())) { - Map data = delta.eventData(); - usage[0] = ((Number) data.getOrDefault("promptTokens", 0)).intValue(); - usage[1] = ((Number) data.getOrDefault("completionTokens", 0)).intValue(); - usage[2] = ((Number) data.getOrDefault("cacheReadTokens", 0)).intValue(); - usage[3] = ((Number) data.getOrDefault("cacheWriteTokens", 0)).intValue(); - usage[4] = ((Number) data.getOrDefault("reasoningTokens", 0)).intValue(); - Object model = data.get("runtimeModelName"); - Object provider = data.get("runtimeProviderId"); - if (model != null) modelInfo[0] = model.toString(); - if (provider != null) modelInfo[1] = provider.toString(); - } - mirrorPlanEventToTracker(conversationId, delta, channelType); - } else if (delta.segmentOnly()) { + accumulator.accept(delta, conversationId); + if (!delta.isEvent() && delta.segmentOnly()) { // Per-stage narration ("Let me look that up…"), emitted as // one complete delta per agent loop iteration. Relay it // immediately as its own outgoing message so the user sees - // progress mid-run, and keep it out of the reply accumulator: - // concatenated narrations read as a wall of text in the - // final reply, and persisting them feeds the next turn's - // LLM history an unanswered chain of stated intents that - // replay mistakes for unfinished work (issue #120). + // progress mid-run. String narration = delta.content() != null ? delta.content().trim() : ""; if (relayNarration && !narration.isEmpty() && replyTarget != null) { try { @@ -857,16 +822,10 @@ public class ChannelMessageRouter { channelType, sendErr.getMessage()); } } - } 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(); + String reply = accumulator.getContent(); // The IM sync path bypasses FinalAnswerNode, so hallucinated // /api/v1/files/generated/{id} URLs (LLM wrote a fake link @@ -905,9 +864,19 @@ public class ChannelMessageRouter { // error turns must not pollute memory extraction. boolean isError = errorClassifier.isErrorReply(reply); String status = isError ? "error" : "completed"; + // Persist the full execution record (parts + metadata) + // so the Web console renders IM-routed turns exactly + // like Web direct chats. The content column keeps the + // scrubbed reply text that actually went out. MessageEntity saved = conversationService.saveMessage( - conversationId, "assistant", reply, null, status, - usage[0], usage[1], usage[2], usage[3], usage[4], modelInfo[0], modelInfo[1], null); + conversationId, "assistant", reply, + accumulator.toAssistantParts(), status, + accumulator.getPromptTokens(), accumulator.getCompletionTokens(), + accumulator.getCacheReadTokens(), accumulator.getCacheWriteTokens(), + accumulator.getReasoningTokens(), + blankToNull(accumulator.getRuntimeModelName()), + blankToNull(accumulator.getRuntimeProviderId()), + accumulator.toMetadataJson()); savedAssistantId = saved != null ? saved.getId() : null; if (!isError) { publishConversationCompletedEvent(agentId, conversationId, message.getContent(), reply, chatOrigin); @@ -1066,6 +1035,30 @@ public class ChannelMessageRouter { } } + /** + * Build a per-turn accumulator wired to the stream tracker, so live Web + * observers of an IM conversation receive the same event fan-out as Web + * direct chats, and the persisted metadata matches byte-for-byte. + */ + private AgentStreamAccumulator newAccumulator() { + return new AgentStreamAccumulator(objectMapper, new AgentStreamAccumulator.Sink() { + @Override + public void broadcast(String conversationId, String eventName, Object payload) { + streamTracker.broadcastObject(conversationId, eventName, payload); + } + + @Override + public void updatePhase(String conversationId, String phase) { + streamTracker.updatePhase(conversationId, phase); + } + }); + } + + /** Map the accumulator's empty-string defaults back to SQL NULL. */ + private static String blankToNull(String s) { + return s == null || s.isBlank() ? null : s; + } + /** * 流式处理路径（渠道无关） *

@@ -1074,30 +1067,6 @@ 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) { @@ -1105,33 +1074,22 @@ public class ChannelMessageRouter { log.info("[{}] Streaming processing started: conversationId={}", channelType, conversationId); try { - // Step 1: 产生事件流（RFC-063r §2.5: forward ChatOrigin so tools see channelId） + // Step 1: 产生事件流（forward ChatOrigin so tools see channelId） 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. - // Token usage + model attribution: capture _usage_final event emitted at stream end - final int[] usage = {0, 0, 0, 0, 0}; // [prompt, completion, cacheRead, cacheWrite, reasoning] - final String[] modelInfo = {null, null}; // [runtimeModel, runtimeProvider] - Flux mirroredStream = stream.doOnNext(delta -> { - if (delta.isEvent() && "_usage_final".equals(delta.eventType())) { - Map data = delta.eventData(); - usage[0] = ((Number) data.getOrDefault("promptTokens", 0)).intValue(); - usage[1] = ((Number) data.getOrDefault("completionTokens", 0)).intValue(); - usage[2] = ((Number) data.getOrDefault("cacheReadTokens", 0)).intValue(); - usage[3] = ((Number) data.getOrDefault("cacheWriteTokens", 0)).intValue(); - usage[4] = ((Number) data.getOrDefault("reasoningTokens", 0)).intValue(); - Object model = data.get("runtimeModelName"); - Object provider = data.get("runtimeProviderId"); - if (model != null) modelInfo[0] = model.toString(); - if (provider != null) modelInfo[1] = provider.toString(); - } - mirrorPlanEventToTracker(conversationId, delta, channelType); - }); + // Feed every delta through the shared accumulator before the + // adapter consumes the Flux. The accumulator builds the same + // segments/toolCalls metadata the Web SSE path persists (so the + // console renders the execution timeline for IM-routed turns), + // mirrors tool/plan/content events to any live Web observer of + // this conversation, and captures token usage + model + // attribution. Internal bookkeeping events (_usage_final, + // _routing_decision) are consumed inside the accumulator and + // never reach subscribers. + AgentStreamAccumulator accumulator = newAccumulator(); + Flux mirroredStream = stream.doOnNext(delta -> + accumulator.accept(delta, conversationId)); // Step 2: 委托渠道渲染（渠道内部消费 Flux 并处理 UI 更新） String finalContent = streamingAdapter.processStream(mirroredStream, message, conversationId); @@ -1150,9 +1108,20 @@ public class ChannelMessageRouter { } else if (finalContent != null && !finalContent.isBlank()) { boolean isError = errorClassifier.isErrorReply(finalContent); String status = isError ? "error" : "completed"; + // Persist the full execution record — parts (text/thinking/ + // tool_call) and metadata (segments/toolCalls/plan/…) — so + // the Web console renders IM-routed turns exactly like Web + // direct chats. The content column keeps the adapter's final + // text (the adapter may have post-processed it). MessageEntity saved = conversationService.saveMessage( - conversationId, "assistant", finalContent, null, status, - usage[0], usage[1], usage[2], usage[3], usage[4], modelInfo[0], modelInfo[1], null); + conversationId, "assistant", finalContent, + accumulator.toAssistantParts(), status, + accumulator.getPromptTokens(), accumulator.getCompletionTokens(), + accumulator.getCacheReadTokens(), accumulator.getCacheWriteTokens(), + accumulator.getReasoningTokens(), + blankToNull(accumulator.getRuntimeModelName()), + blankToNull(accumulator.getRuntimeProviderId()), + accumulator.toMetadataJson()); if (!isError) { publishConversationCompletedEvent(agentId, conversationId, promptText, finalContent, chatOrigin); } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/AgentStreamAccumulator.java b/mateclaw-server/src/main/java/vip/mate/channel/web/AgentStreamAccumulator.java new file mode 100644 index 00000000..4cc6ba2d --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/AgentStreamAccumulator.java @@ -0,0 +1,530 @@ +package vip.mate.channel.web; + +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import vip.mate.agent.AgentService; +import vip.mate.agent.GraphEventPublisher; +import vip.mate.workspace.conversation.model.MessageContentPart; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * 流式累积器 — 收集 StreamDelta 事件，持久化到 DB。 + *

+ * 维护两份数据： + *

+ * 两份数据从同一事件流构建，保证一致。segments 保留了 thinking → tools → content + * 的真实交错顺序，toolCalls 是 segments 中 tool_call 类型的平铺视图。 + *

+ * Shared by the Web SSE path ({@code ChatController}) and the IM channel + * router — live fan-out side effects go through the injected {@link Sink} + * so each caller keeps its own broadcast semantics. Internal bookkeeping + * events ({@code _usage_final}, {@code _routing_decision}) are consumed + * here and never reach the sink. + */ +@Slf4j +public final class AgentStreamAccumulator { + + /** + * Live fan-out hooks. The accumulator itself only builds the persisted + * metadata/parts; anything a subscriber should see in real time is + * delegated here. + */ + public interface Sink { + /** Broadcast a named event to live subscribers of the conversation. */ + void broadcast(String conversationId, String eventName, Object payload); + + /** Update the conversation's current phase indicator. */ + void updatePhase(String conversationId, String phase); + } + + /** Markdown link pointing at a generated-file download URL. Used to + * surface generated artifacts in the run-overview rail. */ + private static final Pattern GENERATED_FILE_LINK_PATTERN = + Pattern.compile("\\[([^\\]]+)\\]\\(((?:https?://[^/\\s)\\]]+)?/api/v1/files/generated/[A-Za-z0-9-]+)\\)"); + + private final ObjectMapper objectMapper; + private final Sink sink; + + private final StringBuilder content = new StringBuilder(); + private final StringBuilder thinking = new StringBuilder(); + private final List> toolCalls = new ArrayList<>(); + /** 有序事件时间线 — 前端分段渲染的权威数据源 */ + private final List> segments = new ArrayList<>(); + private final List> browserActions = new ArrayList<>(); + private final List warnings = new ArrayList<>(); + private final List> planStepResults = new ArrayList<>(); + /** Tool names whose returnDirect output was folded into the assistant message */ + private final List directToolNames = new ArrayList<>(); + /** Generated file artifacts extracted from tool results — surfaced in the run-overview rail. */ + private final List> generatedFiles = new ArrayList<>(); + private int segCounter = 0; + private int promptTokens = 0; + private int completionTokens = 0; + private int cacheReadTokens = 0; + private int cacheWriteTokens = 0; + private int reasoningTokens = 0; + private String runtimeModelName = ""; + private String runtimeProviderId = ""; + private boolean awaitingApproval = false; + private String currentPhase = ""; + /** + * Graph-emitted FinishReason for the turn (e.g. {@code "incomplete"}, + * {@code "stopped"}, {@code "evidence_insufficient"}). Sourced from + * the {@code finish_reason} {@link GraphEventPublisher} + * event that {@code FinalAnswerNode} attaches to its PENDING_EVENTS + * output — same pipeline the SSE accumulator already drains, so the + * value is delivered alongside the assistant content (not via a + * sibling SSE-only broadcast that would bypass this accumulator). + * Persisted into message metadata so downstream filters + * (memory promotion gate) see a machine-readable status instead of + * having to guess from text. Empty string until the event arrives. + */ + private String finishReason = ""; + /** + * Recovery affordance payload from {@link GraphEventPublisher#feedback}. + * Persisted into {@code metadata.feedbackEvent} so a page reload still + * surfaces the retry/regenerate/report card on the failed assistant + * bubble. Null when the turn ended cleanly. + */ + private Map feedbackEvent = null; + private Long planId = null; + private List planSteps = List.of(); + private Integer currentPlanStep = null; + private Map pendingApproval = null; + /** + * Multimodal sidecar routing decision for this turn (null when no + * routing happened). Captured from the {@code _routing_decision} + * event emitted before the graph stream and folded into + * {@code metadata.routing} on persistence so the chat UI can show + * which sidecar (if any) was invoked. + */ + private Map routingDecision = null; + + public AgentStreamAccumulator(ObjectMapper objectMapper, Sink sink) { + this.objectMapper = objectMapper; + this.sink = sink; + } + + public synchronized void accept(AgentService.StreamDelta delta, String conversationId) { + if (delta == null) return; + + if (delta.isEvent()) { + if ("_usage_final".equals(delta.eventType())) { + Map data = delta.eventData(); + promptTokens = ((Number) data.getOrDefault("promptTokens", 0)).intValue(); + completionTokens = ((Number) data.getOrDefault("completionTokens", 0)).intValue(); + cacheReadTokens = ((Number) data.getOrDefault("cacheReadTokens", 0)).intValue(); + cacheWriteTokens = ((Number) data.getOrDefault("cacheWriteTokens", 0)).intValue(); + reasoningTokens = ((Number) data.getOrDefault("reasoningTokens", 0)).intValue(); + runtimeModelName = String.valueOf(data.getOrDefault("runtimeModelName", "")); + runtimeProviderId = String.valueOf(data.getOrDefault("runtimeProviderId", "")); + return; + } + if ("phase".equals(delta.eventType())) { + String phase = String.valueOf(delta.eventData().getOrDefault("phase", "")); + if (!phase.isBlank()) { + currentPhase = phase; + sink.updatePhase(conversationId, phase); + // phase 切换时关闭 running 的 content/thinking segment，保留边界 + finalizeRunningSegments("content", "thinking"); + } + } + if ("finish_reason".equals(delta.eventType())) { + Object reason = delta.eventData().get("reason"); + if (reason != null) { + // Last-write-wins: graph normally fires this exactly once + // at FinalAnswerNode completion. Replay paths that re-enter + // the graph after approval will emit a fresh value, which + // is the correct behavior — the latest reason is what gets + // persisted with the assistant message. + finishReason = String.valueOf(reason); + } + } + if (GraphEventPublisher.EVENT_FEEDBACK.equals(delta.eventType())) { + // Snapshot the affordance payload so it persists into + // message metadata. The same event is also rebroadcast + // live (via the sink fall-through below) so an + // already-mounted UI sees it instantly without + // waiting for the message-save round trip. + feedbackEvent = delta.eventData(); + } + if (GraphEventPublisher.EVENT_ROUTING_DECISION.equals(delta.eventType())) { + // Captured at turn start; persisted under metadata.routing so the + // chat UI can render which sidecar (if any) was invoked. Internal + // event — return early to skip rebroadcast on IM channels. + routingDecision = delta.eventData(); + return; + } + accumulateToolEvent(delta.eventType(), delta.eventData(), conversationId); + try { + sink.broadcast(conversationId, delta.eventType(), delta.eventData()); + } catch (Exception e) { + log.warn("Failed to broadcast event {}: {}", delta.eventType(), e.getMessage()); + } + return; + } + + // content_delta + if (delta.content() != null && !delta.content().isBlank()) { + // segmentOnly deltas route per-iteration narration to the + // segments timeline only — the persisted top-level content + // field stays clean so it carries the final answer span, + // not "我来…让我…" concatenations across iterations (issue + // #120 narration leg). segmentOnly implies persistenceOnly, + // so no broadcast either. + if (!delta.segmentOnly()) { + content.append(delta.content()); + } + sink.updatePhase(conversationId, "drafting_answer"); + if (!delta.persistenceOnly()) { + sink.broadcast(conversationId, "content_delta", Map.of("delta", delta.content())); + } + // segments: 追加到当前 running content segment，或创建新的 + var seg = findLastRunning("content"); + if (seg != null) { + seg.put("text", seg.getOrDefault("text", "") + delta.content()); + } else { + finalizeRunningSegments("thinking"); + var s = newSegment("content"); + s.put("text", delta.content()); + segments.add(s); + } + } + + // thinking_delta + if (delta.thinking() != null && !delta.thinking().isBlank()) { + if (!delta.segmentOnly()) { + thinking.append(delta.thinking()); + } + if (!delta.persistenceOnly()) { + sink.broadcast(conversationId, "thinking_delta", Map.of("delta", delta.thinking())); + } + var seg = findLastRunning("thinking"); + if (seg != null) { + seg.put("thinkingText", seg.getOrDefault("thinkingText", "") + delta.thinking()); + } else { + var s = newSegment("thinking"); + s.put("thinkingText", delta.thinking()); + segments.add(s); + } + } + } + + public boolean isAwaitingApproval() { return awaitingApproval; } + + private void accumulateToolEvent(String eventType, Map data, String conversationId) { + if ("tool_approval_requested".equals(eventType)) { + awaitingApproval = true; + currentPhase = "awaiting_approval"; + pendingApproval = new LinkedHashMap<>(); + pendingApproval.put("pendingId", data.getOrDefault("pendingId", "")); + pendingApproval.put("toolName", data.getOrDefault("toolName", "")); + pendingApproval.put("arguments", data.getOrDefault("arguments", "")); + pendingApproval.put("reason", data.getOrDefault("reason", "")); + pendingApproval.put("status", "pending_approval"); + if (data.containsKey("findings")) pendingApproval.put("findings", data.get("findings")); + if (data.containsKey("maxSeverity")) pendingApproval.put("maxSeverity", data.get("maxSeverity")); + if (data.containsKey("summary")) pendingApproval.put("summary", data.get("summary")); + sink.updatePhase(conversationId, "awaiting_approval"); + } else if ("tool_approval_resolved".equals(eventType)) { + if (pendingApproval != null) { + pendingApproval.put("status", + "approved".equals(String.valueOf(data.getOrDefault("decision", ""))) ? "approved" : "denied"); + } + } else if ("plan_created".equals(eventType)) { + Object rawPlanId = data.get("planId"); + if (rawPlanId instanceof Number n) { + planId = n.longValue(); + } else if (rawPlanId != null) { + try { planId = Long.valueOf(String.valueOf(rawPlanId)); } catch (Exception ignored) {} + } + Object steps = data.get("steps"); + if (steps instanceof List list) { + planSteps = list.stream().map(String::valueOf).toList(); + planStepResults.clear(); + for (int i = 0; i < planSteps.size(); i++) { + planStepResults.add(null); + } + } + currentPlanStep = 0; + } else if ("plan_step_started".equals(eventType)) { + Object idx = data.get("index"); + if (idx instanceof Number n) { + currentPlanStep = n.intValue(); + } + } else if ("plan_step_completed".equals(eventType)) { + Object idx = data.get("index"); + if (idx instanceof Number n) { + int index = n.intValue(); + currentPlanStep = index; + ensurePlanStepCapacity(index + 1); + Map stepResult = new LinkedHashMap<>(); + stepResult.put("result", data.getOrDefault("result", "")); + stepResult.put("status", "completed"); + planStepResults.set(index, stepResult); + } + } else if ("browser_action".equals(eventType)) { + browserActions.add(new LinkedHashMap<>(data)); + } else if ("warning".equals(eventType)) { + String warning = String.valueOf(data.getOrDefault("message", + data.getOrDefault("delta", ""))); + if (!warning.isBlank()) { + warnings.add(warning); + } + } else if ("tool_call_started".equals(eventType)) { + // toolCalls（兼容） + Map tc = new LinkedHashMap<>(); + // toolCallId is required for history replay to pair the persisted + // assistant tool_call with its tool_response — providers reject any + // sequence whose ids don't match. Always record it (empty string + // when the upstream event didn't carry one, e.g. forced tool calls). + tc.put("toolCallId", String.valueOf(data.getOrDefault("toolCallId", ""))); + tc.put("name", data.getOrDefault("toolName", "")); + tc.put("arguments", data.getOrDefault("arguments", "")); + tc.put("status", "running"); + toolCalls.add(tc); + // segments: 关闭 running thinking/content，插入 tool_call + finalizeRunningSegments("thinking", "content"); + var seg = newSegment("tool_call"); + seg.put("toolCallId", String.valueOf(data.getOrDefault("toolCallId", ""))); + seg.put("toolName", data.getOrDefault("toolName", "")); + seg.put("toolArgs", data.getOrDefault("arguments", "")); + segments.add(seg); + } else if ("tool_direct_result".equals(eventType)) { + // returnDirect tool — track the tool name so history replay can + // render a "data returned directly by tool" badge. The actual + // textual content reaches the user/persistence layer through the + // regular content_delta path (FinalAnswerNode's FINAL_ANSWER → + // StateGraphReActAgent → StreamDelta), so we intentionally do + // NOT add a content-bearing segment here to avoid the user + // seeing the same text twice. + String toolName = String.valueOf(data.getOrDefault("toolName", "")); + if (!toolName.isBlank() && !directToolNames.contains(toolName)) { + directToolNames.add(toolName); + } + } else if ("tool_call_completed".equals(eventType)) { + String toolName = String.valueOf(data.getOrDefault("toolName", "")); + String toolCallId = String.valueOf(data.getOrDefault("toolCallId", "")); + // toolCalls（兼容）— prefer toolCallId match so parallel calls of + // the same tool don't collide on the running+toolName fallback. + for (int i = toolCalls.size() - 1; i >= 0; i--) { + Map tc = toolCalls.get(i); + boolean matches = (!toolCallId.isEmpty() + && toolCallId.equals(String.valueOf(tc.getOrDefault("toolCallId", "")))) + || (toolCallId.isEmpty() + && "running".equals(tc.get("status")) + && toolName.equals(tc.get("name"))); + if (matches) { + tc.put("result", data.getOrDefault("result", "")); + tc.put("success", data.getOrDefault("success", true)); + tc.put("status", "completed"); + break; + } + } + // segments: 标记对应 tool_call 完成 + for (int i = segments.size() - 1; i >= 0; i--) { + var seg = segments.get(i); + if (!"tool_call".equals(seg.get("type"))) continue; + boolean matches = (!toolCallId.isEmpty() + && toolCallId.equals(String.valueOf(seg.getOrDefault("toolCallId", "")))) + || (toolCallId.isEmpty() + && "running".equals(seg.get("status")) + && toolName.equals(seg.get("toolName"))); + if (matches) { + seg.put("status", "completed"); + seg.put("toolResult", data.getOrDefault("result", "")); + seg.put("toolSuccess", data.getOrDefault("success", true)); + break; + } + } + // Extract generated-file links from the tool result so the + // run-overview rail can surface artifacts without re-scanning + // segments on the frontend. + extractGeneratedFiles(String.valueOf(data.getOrDefault("result", "")), toolName); + } + } + + /** Scan a tool result for markdown links pointing at generated-file + * download URLs and collect them into {@link #generatedFiles}. + * De-duplicates by URL so a link echoed in later tool results doesn't + * produce duplicate entries in the run-overview rail. */ + private void extractGeneratedFiles(String result, String toolName) { + if (result == null || result.isBlank()) return; + Matcher m = GENERATED_FILE_LINK_PATTERN.matcher(result); + while (m.find()) { + String url = m.group(2); + boolean dup = generatedFiles.stream() + .anyMatch(f -> url.equals(String.valueOf(f.get("url")))); + if (dup) continue; + Map file = new LinkedHashMap<>(); + file.put("filename", m.group(1)); + file.put("url", url); + file.put("toolName", toolName); + generatedFiles.add(file); + } + } + + private void ensurePlanStepCapacity(int size) { + while (planStepResults.size() < size) { + planStepResults.add(null); + } + } + + // ==================== Segment helpers ==================== + + private Map newSegment(String type) { + Map seg = new LinkedHashMap<>(); + seg.put("id", type.substring(0, 2) + "-" + segCounter++); + seg.put("type", type); + seg.put("status", "running"); + return seg; + } + + private Map findLastRunning(String type) { + for (int i = segments.size() - 1; i >= 0; i--) { + var seg = segments.get(i); + if (type.equals(seg.get("type")) && "running".equals(seg.get("status"))) return seg; + } + return null; + } + + private void finalizeRunningSegments(String... types) { + var typeSet = Set.of(types); + for (var seg : segments) { + if ("running".equals(seg.get("status")) && typeSet.contains(seg.get("type"))) { + seg.put("status", "completed"); + } + } + } + + // ==================== 原有访问器 ==================== + + public String getContent() { return content.toString().trim(); } + public String getThinking() { return thinking.toString().trim(); } + public int getPromptTokens() { return promptTokens; } + public int getCompletionTokens() { return completionTokens; } + public int getCacheReadTokens() { return cacheReadTokens; } + public int getCacheWriteTokens() { return cacheWriteTokens; } + public int getReasoningTokens() { return reasoningTokens; } + public String getRuntimeModelName() { return runtimeModelName; } + public String getRuntimeProviderId() { return runtimeProviderId; } + public String getCurrentPhase() { return currentPhase; } + public String getFinishReason() { return finishReason; } + public boolean segmentsEmpty() { return segments.isEmpty(); } + + public synchronized List toAssistantParts() { + List parts = new ArrayList<>(); + if (!getContent().isBlank()) { + MessageContentPart textPart = new MessageContentPart(); + textPart.setType("text"); + textPart.setText(getContent()); + parts.add(textPart); + } + if (!getThinking().isBlank()) { + MessageContentPart thinkingPart = new MessageContentPart(); + thinkingPart.setType("thinking"); + thinkingPart.setText(getThinking()); + parts.add(thinkingPart); + } + for (Map tc : toolCalls) { + try { + parts.add(MessageContentPart.toolCall(objectMapper.writeValueAsString(tc))); + } catch (Exception e) { + log.warn("Failed to serialize tool call: {}", e.getMessage()); + } + } + return parts; + } + + private void finalizeToolCalls() { + for (Map tc : toolCalls) { + if ("running".equals(tc.get("status"))) tc.put("status", "completed"); + } + } + + /** + * 生成 metadata JSON：包含 toolCalls + segments。 + * toolCalls 保留兼容旧 UI，segments 是按事件顺序的完整时间线。 + */ + public synchronized String toMetadataJson() { + finalizeToolCalls(); + finalizeRunningSegments("thinking", "content", "tool_call"); + SegmentSupersedeDetector.markSuperseded(segments); + try { + Map metadata = new LinkedHashMap<>(); + if (!toolCalls.isEmpty()) { + metadata.put("toolCalls", toolCalls); + } + if (!segments.isEmpty()) { + metadata.put("segments", segments); + } + if (!currentPhase.isBlank()) { + metadata.put("currentPhase", currentPhase); + } + if (planId != null || !planSteps.isEmpty() || currentPlanStep != null) { + Map plan = new LinkedHashMap<>(); + if (planId != null) plan.put("planId", planId); + if (!planSteps.isEmpty()) plan.put("steps", planSteps); + if (currentPlanStep != null) plan.put("currentStep", currentPlanStep); + if (planStepResults.stream().anyMatch(Objects::nonNull)) { + plan.put("stepResults", planStepResults); + } + metadata.put("plan", plan); + } + if (pendingApproval != null && !pendingApproval.isEmpty()) { + metadata.put("pendingApproval", pendingApproval); + } + if (!browserActions.isEmpty()) { + metadata.put("browserActions", browserActions); + } + if (!warnings.isEmpty()) { + metadata.put("warnings", warnings); + } + if (!directToolNames.isEmpty()) { + // Only the tool names go into metadata — the full content + // already lives in mate_message.content (assembled by + // FinalAnswerNode). UI uses this to badge historical + // messages as "data returned directly by tool". + metadata.put("directToolNames", directToolNames); + } + if (!generatedFiles.isEmpty()) { + metadata.put("generatedFiles", generatedFiles); + } + if (!finishReason.isEmpty()) { + // Surface graph FinishReason so MemorySummarizationGate and + // any other downstream consumer can branch on a structured + // status (e.g. skip INCOMPLETE / STOPPED / ERROR_FALLBACK + // turns from long-term memory promotion) instead of doing + // brittle text matching on the assistant content. + metadata.put("finishReason", finishReason); + } + if (feedbackEvent != null && !feedbackEvent.isEmpty()) { + // Persist the recovery-affordance payload so the + // retry/regenerate/report card survives page reload. + // Stored as-is (errorType, errorMessage, actions, + // timestamp) — frontend MessageBubble reads + // metadata.feedbackEvent and renders one button per + // entry in `actions`. + metadata.put("feedbackEvent", feedbackEvent); + } + if (routingDecision != null && !routingDecision.isEmpty()) { + metadata.put("routing", routingDecision); + } + return objectMapper.writeValueAsString(metadata); + } catch (Exception e) { + log.warn("Failed to serialize metadata: {}", e.getMessage()); + return "{}"; + } + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java index 9a946713..fb89bc84 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java @@ -162,11 +162,14 @@ class ChannelMessageRouterNarrationTest { process.invoke(router, message, adapter, channel, "telegram:alice"); } - /** The persisted assistant row must hold the final-answer span only. */ + /** The persisted assistant row must hold the final-answer span only. + * Parts and metadata now carry the execution record (segments etc.), + * so they are non-null — content semantics are what this asserts. */ void verifyPersistedAssistantContent(String expected) { verify(conversationService).saveMessage( - eq("telegram:alice"), eq("assistant"), eq(expected), isNull(), eq("completed"), - eq(0), eq(0), eq(0), eq(0), eq(0), isNull(), isNull(), isNull()); + eq("telegram:alice"), eq("assistant"), eq(expected), anyList(), eq("completed"), + eq(0), eq(0), eq(0), eq(0), eq(0), isNull(), isNull(), + argThat((String metadata) -> metadata != null && metadata.contains("\"segments\""))); } /** processMessage swallows failures into a generic error reply — assert none fired. */