From 32dd6e7911dfa18bb50520870d182883fbd02401 Mon Sep 17 00:00:00 2001 From: matevip Date: Tue, 4 Aug 2026 22:50:42 -0400 Subject: [PATCH] fix(channel): sync-path IM channels stop relaying pre-tool rehearsals verbatim --- .../mate/channel/ChannelMessageRouter.java | 52 +++++++++++--- .../channel/ProvisionalContentTracker.java | 40 ++++++++--- .../channel/wecom/WeComChannelAdapter.java | 9 +-- .../ChannelMessageRouterNarrationTest.java | 68 +++++++++++++++++-- .../ProvisionalContentTrackerTest.java | 64 ++++++++++++----- 5 files changed, 182 insertions(+), 51 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 d5eb3a7c..3a32a716 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java @@ -884,30 +884,47 @@ public class ChannelMessageRouter { // and IM channels still need the text for the outgoing reply), // segmentOnly narration excluded (issue #120). AgentStreamAccumulator accumulator = newAccumulator(); + // Narration lifecycle: relayed messages on this path are + // permanent (IM messages cannot be retracted), so per-stage + // narration publishes one behind through the shared tracker — + // a pre-tool rehearsal (possibly a fabricated result table) + // is dropped once later content supersedes it instead of + // reaching the user verbatim. + final ProvisionalContentTracker narrationTracker = + new ProvisionalContentTracker(channelType); agentService.chatStructuredStream(agentId, promptText, conversationId, message.getSenderId(), chatOrigin) .doOnNext(delta -> { accumulator.accept(delta, conversationId); - if (!delta.isEvent() && delta.segmentOnly()) { + if (delta.isEvent()) { + if ("tool_call_completed".equals(delta.eventType())) { + narrationTracker.onToolObservation(); + } + return; + } + if (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. + // one complete delta per agent loop iteration, each becoming + // its own outgoing message so the user sees progress mid-run. String narration = delta.content() != null ? delta.content().trim() : ""; if (relayNarration && !narration.isEmpty() && replyTarget != null) { - try { - adapter.renderAndSend(replyTarget, narration); - } catch (Exception sendErr) { - // A failed progress send must not abort the agent - // run — the final reply still goes out below. - log.warn("[{}] Narration relay failed (non-fatal): {}", - channelType, sendErr.getMessage()); + String publishable = narrationTracker.stageNarration(narration, delta.kind()); + if (publishable != null) { + relayNarrationSafely(adapter, replyTarget, publishable); } } } }) .blockLast(Duration.ofMinutes(10)); String reply = accumulator.getContent(); + // The last narration was held back until the answer was + // known: superseded → dropped, otherwise it still goes out + // (before the final reply) unless it duplicates it. + String heldNarration = narrationTracker.settle(!reply.isBlank()); + if (heldNarration != null && replyTarget != null + && !heldNarration.equals(reply.trim())) { + relayNarrationSafely(adapter, replyTarget, heldNarration); + } // The IM sync path bypasses FinalAnswerNode, so hallucinated // /api/v1/files/generated/{id} URLs (LLM wrote a fake link @@ -1290,6 +1307,19 @@ public class ChannelMessageRouter { return s == null || s.isBlank() ? null : s; } + /** + * Send a progress narration as its own outgoing message. A failed send + * must not abort the agent run — the final reply still goes out. + */ + private void relayNarrationSafely(ChannelAdapter adapter, String replyTarget, String narration) { + try { + adapter.renderAndSend(replyTarget, narration); + } catch (Exception sendErr) { + log.warn("[{}] Narration relay failed (non-fatal): {}", + adapter.getChannelType(), sendErr.getMessage()); + } + } + /** * 流式处理路径（渠道无关） *

diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ProvisionalContentTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/ProvisionalContentTracker.java index 7796a5ea..6fc17d59 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ProvisionalContentTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ProvisionalContentTracker.java @@ -48,38 +48,53 @@ public final class ProvisionalContentTracker { /** Where the supersede happened — metric tag, one value per surface. */ private final String surface; + /** Tool observations completed so far this turn (caller-reported). */ + private int observations; + /** Observation count at the time of the most recent staging. */ + private int lastStageMark; + private String pendingText; private boolean pendingProvisional; + /** Observation count when the pending narration was staged. */ + private int pendingMark; public ProvisionalContentTracker(String surface) { this.surface = surface; } + /** Report a completed tool observation (a {@code tool_call_completed} event). */ + public void onToolObservation() { + observations++; + } + /** * Stage a per-round narration. Returns the previous staged * narration if the new arrival makes it publishable, or {@code null} when * there is nothing to publish (no previous, or the previous was - * provisional and is now superseded by this later content). + * provisional and tool observations since its staging mean this later + * content supersedes it). * - * @param kind producer-assigned kind; {@code null} for - * pre-tag producers - * @param observedSinceLast structural fallback used only when {@code kind} - * is null: whether a tool observation completed - * since the previous narration (the pre-tag - * online rule) + * @param kind producer-assigned kind; {@code null} for pre-tag producers, + * in which case a narration counts as provisional when no + * observation completed since the previous staging (the + * pre-tag online rule) */ - public String stageNarration(String text, ContentKind kind, boolean observedSinceLast) { + public String stageNarration(String text, ContentKind kind) { + boolean observedSinceLast = observations > lastStageMark; boolean provisional = kind != null ? kind == ContentKind.PRE_TOOL_NARRATION : !observedSinceLast; String previous = pendingText; boolean previousProvisional = pendingProvisional; + int previousMark = pendingMark; pendingText = text; pendingProvisional = provisional; + pendingMark = observations; + lastStageMark = observations; if (previous == null) { return null; } - if (previousProvisional) { + if (previousProvisional && observations > previousMark) { recordSuperseded(previous); return null; } @@ -89,7 +104,8 @@ public final class ProvisionalContentTracker { /** * Resolve the staged narration at turn end. Returns the text to publish, * or {@code null} when nothing remains (no staged narration, or it was - * provisional and the turn produced final content that replaces it). + * provisional, tools ran after it, and the turn produced final content + * that replaces it). * * @param hasFinalContent whether the turn produced a final answer — with * one, a provisional narration is superseded; with @@ -99,12 +115,14 @@ public final class ProvisionalContentTracker { public String settle(boolean hasFinalContent) { String text = pendingText; boolean provisional = pendingProvisional; + int mark = pendingMark; pendingText = null; pendingProvisional = false; + pendingMark = 0; if (text == null) { return null; } - if (provisional && hasFinalContent) { + if (provisional && hasFinalContent && observations > mark) { recordSuperseded(text); return null; } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java index f2089e8f..028f243c 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java @@ -1532,14 +1532,12 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea // the final answer always lands in the newest bubble. AtomicReference liveCtx = new AtomicReference<>(initialCtx); ProvisionalContentTracker tracker = new ProvisionalContentTracker("wecom"); - AtomicInteger toolResults = new AtomicInteger(); - final int[] lastNarrationToolMark = {0}; final long[] lastFlushAt = {0L}; stream.doOnNext(delta -> { boolean flushNow = false; if (delta.isEvent()) { if ("tool_call_completed".equals(delta.eventType())) { - toolResults.incrementAndGet(); + tracker.onToolObservation(); } flushNow = progress.onEvent(delta.eventType(), delta.eventData()); if (standaloneToolMessages) { @@ -1563,11 +1561,8 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea // means this one was written with real output in hand). String narration = delta.content() != null ? delta.content().trim() : ""; if (!narration.isEmpty()) { - int mark = toolResults.get(); - boolean observedSinceLast = mark > lastNarrationToolMark[0]; - lastNarrationToolMark[0] = mark; progress.onNarration(narration); - String publishable = tracker.stageNarration(narration, delta.kind(), observedSinceLast); + String publishable = tracker.stageNarration(narration, delta.kind()); if (publishable != null) { WeComReplyContext ctx = liveCtx.get(); if (replyContexts.get(replyTarget) == ctx) { 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 200d547f..9233a621 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java @@ -25,11 +25,14 @@ import static org.mockito.Mockito.*; /** * Sync-path narration routing for non-streaming IM adapters. * - *

Per-iteration narration deltas ({@code segmentOnly}) must be relayed as - * standalone messages the moment they arrive and stay out of the accumulated - * final reply — otherwise multiple iterations glue into a wall of text and - * the persisted assistant message pollutes the next turn's LLM history with - * unanswered stated intents (issue #120). + *

Per-iteration narration deltas ({@code segmentOnly}) are relayed as + * standalone messages and stay out of the accumulated final reply — otherwise + * multiple iterations glue into a wall of text and the persisted assistant + * message pollutes the next turn's LLM history with unanswered stated intents + * (issue #120). Publishing lags one narration behind through the shared + * provisional-content tracker: messages on this path are permanent, so a + * pre-tool rehearsal (possibly a fabricated result table) must be dropped + * once later content supersedes it instead of reaching the user verbatim. */ class ChannelMessageRouterNarrationTest { @@ -100,6 +103,61 @@ class ChannelMessageRouterNarrationTest { f.verifyNoProcessingError(); } + @Test + @DisplayName("kind-tagged pre-tool rehearsal is dropped once a grounded answer exists") + void preToolRehearsalDroppedForGroundedAnswer() throws Exception { + Fixture f = new Fixture(); + String rehearsal = "环境监测结果：温度 29.0°C，湿度 63.0%。"; + f.streamReturns( + StreamDelta.segmentOnly(rehearsal, null, vip.mate.agent.ContentKind.PRE_TOOL_NARRATION), + StreamDelta.event("tool_call_completed", + java.util.Map.of("toolCallId", "t1", "toolName", "envQuery", + "result", "data={}", "success", true)), + StreamDelta.persistOnly("接口返回为空，所有会议室均无环境数据。", null)); + + f.process("各会议室的环境情况"); + + verify(f.adapter, never()).renderAndSend("reply-1", rehearsal); + verify(f.adapter).renderAndSend("reply-1", "接口返回为空，所有会议室均无环境数据。"); + f.verifyPersistedAssistantContent("接口返回为空，所有会议室均无环境数据。"); + f.verifyNoProcessingError(); + } + + @Test + @DisplayName("kind-tagged pre-tool narration still goes out when the turn produced no answer") + void preToolNarrationKeptWhenNoAnswer() throws Exception { + Fixture f = new Fixture(); + f.streamReturns( + StreamDelta.segmentOnly("我先调用工具查询状态：", null, + vip.mate.agent.ContentKind.PRE_TOOL_NARRATION), + StreamDelta.event("tool_call_completed", + java.util.Map.of("toolCallId", "t1", "toolName", "q", + "result", "x", "success", true))); + + f.process("查一下状态"); + + verify(f.adapter).renderAndSend("reply-1", "我先调用工具查询状态："); + } + + @Test + @DisplayName("untagged narration is dropped when a tool ran after it and later content followed") + void untaggedNarrationDroppedAfterObservation() throws Exception { + Fixture f = new Fixture(); + f.streamReturns( + StreamDelta.segmentOnly("我先查一下天气", null), + StreamDelta.event("tool_call_completed", + java.util.Map.of("toolCallId", "t1", "toolName", "weather", + "result", "晴", "success", true)), + StreamDelta.segmentOnly("查到了，整理结果", null), + StreamDelta.persistOnly("今天晴。", null)); + + f.process("帮我查天气"); + + verify(f.adapter, never()).renderAndSend("reply-1", "我先查一下天气"); + verify(f.adapter).renderAndSend("reply-1", "查到了，整理结果"); + verify(f.adapter).renderAndSend("reply-1", "今天晴。"); + } + @Test @DisplayName("persistOnly and plain content deltas still accumulate into one reply") void nonNarrationDeltasStillAccumulate() throws Exception { diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ProvisionalContentTrackerTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ProvisionalContentTrackerTest.java index 09c14831..7197ce95 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/ProvisionalContentTrackerTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/ProvisionalContentTrackerTest.java @@ -29,9 +29,10 @@ class ProvisionalContentTrackerTest { void preToolNarrationSupersededByNextNarration() { ProvisionalContentTracker t = new ProvisionalContentTracker("test"); - assertNull(t.stageNarration("先查询会议室数据。", ContentKind.PRE_TOOL_NARRATION, false), + assertNull(t.stageNarration("先查询会议室数据。", ContentKind.PRE_TOOL_NARRATION), "first narration has no predecessor to publish"); - String publishable = t.stageNarration("第一间空闲，继续。", ContentKind.GROUNDED_NARRATION, true); + t.onToolObservation(); + String publishable = t.stageNarration("第一间空闲，继续。", ContentKind.GROUNDED_NARRATION); assertNull(publishable, "provisional predecessor is superseded, not published"); assertEquals("第一间空闲，继续。", t.settle(true), @@ -43,8 +44,10 @@ class ProvisionalContentTrackerTest { void groundedNarrationPublishesOnNext() { ProvisionalContentTracker t = new ProvisionalContentTracker("test"); - t.stageNarration("时间拿到了，再查会议室：", ContentKind.GROUNDED_NARRATION, true); - String publishable = t.stageNarration("会议室也查完了。", ContentKind.GROUNDED_NARRATION, true); + t.onToolObservation(); + t.stageNarration("时间拿到了，再查会议室：", ContentKind.GROUNDED_NARRATION); + t.onToolObservation(); + String publishable = t.stageNarration("会议室也查完了。", ContentKind.GROUNDED_NARRATION); assertEquals("时间拿到了，再查会议室：", publishable); } @@ -53,7 +56,8 @@ class ProvisionalContentTrackerTest { void preToolNarrationSupersededByFinalAnswer() { ProvisionalContentTracker t = new ProvisionalContentTracker("test"); - t.stageNarration("环境监测结果：温度 29.0°C。", ContentKind.PRE_TOOL_NARRATION, false); + t.stageNarration("环境监测结果：温度 29.0°C。", ContentKind.PRE_TOOL_NARRATION); + t.onToolObservation(); assertNull(t.settle(true), "fabricated rehearsal must not survive once a grounded answer exists"); } @@ -62,16 +66,28 @@ class ProvisionalContentTrackerTest { void preToolNarrationCommitsWithoutAnswer() { ProvisionalContentTracker t = new ProvisionalContentTracker("test"); - t.stageNarration("我先调用工具查询状态：", ContentKind.PRE_TOOL_NARRATION, false); + t.stageNarration("我先调用工具查询状态：", ContentKind.PRE_TOOL_NARRATION); + t.onToolObservation(); assertEquals("我先调用工具查询状态：", t.settle(false), "with no replacement content the narration is everything the user gets"); } + @Test + @DisplayName("pre-tool narration with no tool run after it survives settle — nothing replaced it") + void preToolNarrationWithoutToolsAfterSurvives() { + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + + t.stageNarration("我先调用工具：", ContentKind.PRE_TOOL_NARRATION); + assertEquals("我先调用工具：", t.settle(true), + "supersede requires an observation after staging — e.g. all tools denied leaves nothing to defer to"); + } + @Test @DisplayName("grounded narration always settles publishable") void groundedNarrationSettles() { ProvisionalContentTracker t = new ProvisionalContentTracker("test"); - t.stageNarration("查完了，结果如下。", ContentKind.GROUNDED_NARRATION, true); + t.onToolObservation(); + t.stageNarration("查完了，结果如下。", ContentKind.GROUNDED_NARRATION); assertEquals("查完了，结果如下。", t.settle(true)); } @@ -79,7 +95,7 @@ class ProvisionalContentTrackerTest { @DisplayName("settle clears state — second settle finds nothing") void settleClearsState() { ProvisionalContentTracker t = new ProvisionalContentTracker("test"); - t.stageNarration("x", ContentKind.GROUNDED_NARRATION, true); + t.stageNarration("x", ContentKind.GROUNDED_NARRATION); t.settle(true); assertNull(t.settle(true)); } @@ -87,26 +103,40 @@ class ProvisionalContentTrackerTest { @Test @DisplayName("null kind falls back to the observation-counter rule") void nullKindFallsBackToCounter() { + // No observation before staging → provisional; a tool ran after → superseded. ProvisionalContentTracker t = new ProvisionalContentTracker("test"); - - // No observation since turn start → provisional (pre-tag online rule). - t.stageNarration("我先查一下：", null, false); + t.stageNarration("我先查一下：", null); + t.onToolObservation(); assertNull(t.settle(true), "untagged pre-tool narration still dropped for a grounded answer"); - // Observation completed since last narration → grounded. + // Observation completed before staging → grounded, publishes. ProvisionalContentTracker t2 = new ProvisionalContentTracker("test"); - t2.stageNarration("拿到结果了。", null, true); + t2.onToolObservation(); + t2.stageNarration("拿到结果了。", null); assertEquals("拿到结果了。", t2.settle(true)); } + @Test + @DisplayName("untagged narrations with no tool activity between them all publish — legacy relay preserved") + void untaggedNoToolStreamKeepsRelay() { + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + t.stageNarration("我先查一下天气", null); + assertEquals("我先查一下天气", t.stageNarration("再帮你汇总结果", null), + "no observation between the two — the predecessor was not replaced by anything grounded"); + assertEquals("再帮你汇总结果", t.settle(true)); + } + @Test @DisplayName("producer kind outranks the counter signal when both are present") void kindOutranksCounter() { - // The counter says an observation happened, but the producer knows the - // text was emitted before any observation of this turn (the two can - // disagree when multiple narrations land between two completions). + // The counter says an observation preceded the narration, but the + // producer knows the text was emitted before any observation of this + // turn (the two can disagree when multiple narrations land between + // two completions). Kind wins. ProvisionalContentTracker t = new ProvisionalContentTracker("test"); - t.stageNarration("预演内容", ContentKind.PRE_TOOL_NARRATION, true); + t.onToolObservation(); + t.stageNarration("预演内容", ContentKind.PRE_TOOL_NARRATION); + t.onToolObservation(); assertNull(t.settle(true), "kind is authoritative — counter signal ignored when tagged"); }