From 54a2c3c0a345954f62be50549efd9fb9a298b19c Mon Sep 17 00:00:00 2001 From: matevip Date: Tue, 4 Aug 2026 06:14:55 -0400 Subject: [PATCH] fix(wecom): drop pre-tool rehearsal narrations from permanent bubbles --- .../channel/wecom/WeComChannelAdapter.java | 83 +++++++++++++++---- .../channel/wecom/WeComProcessStreamTest.java | 74 +++++++++++++++-- 2 files changed, 133 insertions(+), 24 deletions(-) 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 c4bc205b..fe4b72dd 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 @@ -1416,7 +1416,7 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea } }) .blockLast(Duration.ofMinutes(10)); - outcome = new StreamOutcome(null, false); + outcome = new StreamOutcome(null, false, false); } else { outcome = consumeWithProgress(stream, message, replyTarget, ctx, contentAccumulator); } @@ -1428,9 +1428,14 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea // (routine once the answer is short — the model restates it before the // last tool call), publishing both puts the identical bubble on screen // twice. Same text → drop the narration, the final answer covers it. + // A pre-tool narration (no observation behind it, tools ran after it) + // is likewise dropped once a grounded answer exists: whatever it says + // — process narration or a predicted result — it was written before + // this turn's observations, and IM bubbles cannot be retracted later. String pendingNarration = outcome.pendingNarration(); if (!finalContent.isBlank()) { - if (pendingNarration != null && !sameOutboundText(pendingNarration, finalContent)) { + if (pendingNarration != null && !outcome.pendingNarrationPreTool() + && !sameOutboundText(pendingNarration, finalContent)) { publishNarrationBubble(replyTarget, pendingNarration); } renderAndSend(replyTarget, finalContent); @@ -1454,10 +1459,39 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea * which only {@link #processStream} can act on (it needs the final answer * first). * - * @param pendingNarration the last per-stage narration, still unpublished - * @param approvalPending a tool call is parked on human approval + * @param pendingNarration the last per-stage narration, still unpublished + * @param pendingNarrationPreTool the pending narration was written before any + * tool observation and tools ran after it — not + * grounded, so it must not become a permanent + * bubble once a grounded answer exists + * @param approvalPending a tool call is parked on human approval */ - private record StreamOutcome(String pendingNarration, boolean approvalPending) {} + private record StreamOutcome(String pendingNarration, boolean pendingNarrationPreTool, + boolean approvalPending) {} + + /** + * A per-stage narration held back from publication, with the structural + * facts needed to decide later whether it may become a permanent bubble. + * + * @param text the narration text + * @param followsToolResult a tool observation completed between the previous + * narration (or turn start) and this one — the text + * was written with real output in hand + * @param toolMark tool-observation count at staging time; a higher + * count later means tools ran after this narration + */ + private record StagedNarration(String text, boolean followsToolResult, int toolMark) { + + /** + * Whether this narration is superseded by whatever content follows it: + * it was not grounded in an observation, and tool calls ran after it — + * the same structural rule the Web segments timeline applies, evaluated + * online. The caller only asks once follow-up content exists. + */ + boolean supersededAt(int toolResultsNow) { + return !followsToolResult && toolResultsNow > toolMark; + } + } /** Compare two outbound texts the way the user sees them (post-filter, trimmed). */ private boolean sameOutboundText(String a, String b) { @@ -1520,11 +1554,16 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea // as the new progress bubble, so chat chronology stays intact and // the final answer always lands in the newest bubble. AtomicReference liveCtx = new AtomicReference<>(initialCtx); - AtomicReference pendingNarration = new AtomicReference<>(); + AtomicReference pendingNarration = new AtomicReference<>(); + 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(); + } flushNow = progress.onEvent(delta.eventType(), delta.eventData()); if (standaloneToolMessages) { maybeSendToolEventMessage(replyTarget, delta.eventType(), delta.eventData()); @@ -1537,22 +1576,32 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea // LLM history with unanswered chain-of-thought. // // Publishing lags one narration behind: the newest one is only - // staged (visible live in the bubble, not yet finalized) so - // processStream can still drop it if the final answer turns out - // to be the same text. Without the lag the user reads the same - // paragraph in two adjacent bubbles. + // staged (visible live in the bubble, not yet finalized) so a + // later decision can still drop it — when the final answer + // turns out to be the same text, or when it proves to be a + // pre-tool rehearsal. A narration written before any tool + // observation, with tool calls running after it, is not + // grounded in this turn's results (it may even be a predicted + // result table); once grounded content follows, it is dropped + // instead of becoming an unretractable permanent bubble. + // Narrations written after an observation keep standalone + // value (e.g. intermediate results) and publish as before. String narration = delta.content() != null ? delta.content().trim() : ""; if (!narration.isEmpty()) { - String previous = pendingNarration.getAndSet(narration); + int mark = toolResults.get(); + StagedNarration staged = new StagedNarration( + narration, mark > lastNarrationToolMark[0], mark); + lastNarrationToolMark[0] = mark; + StagedNarration previous = pendingNarration.getAndSet(staged); progress.onNarration(narration); - if (previous != null) { + if (previous != null && !previous.supersededAt(mark)) { WeComReplyContext ctx = liveCtx.get(); if (replyContexts.get(replyTarget) == ctx) { - liveCtx.set(rollProgressBubble(replyTarget, ctx, previous, progress)); + liveCtx.set(rollProgressBubble(replyTarget, ctx, previous.text(), progress)); } else { // Bubble already force-finished (180s ceiling) — the // narration still goes out as a plain message. - sendMessage(replyTarget, previous); + sendMessage(replyTarget, previous.text()); } } flushNow = true; @@ -1589,7 +1638,11 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea } }).blockLast(Duration.ofMinutes(10)); - return new StreamOutcome(pendingNarration.get(), progress.isApprovalPending()); + StagedNarration last = pendingNarration.get(); + return new StreamOutcome( + last != null ? last.text() : null, + last != null && last.supersededAt(toolResults.get()), + progress.isApprovalPending()); } /** diff --git a/mateclaw-server/src/test/java/vip/mate/channel/wecom/WeComProcessStreamTest.java b/mateclaw-server/src/test/java/vip/mate/channel/wecom/WeComProcessStreamTest.java index 6c219cbf..16130349 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/wecom/WeComProcessStreamTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/wecom/WeComProcessStreamTest.java @@ -154,7 +154,7 @@ class WeComProcessStreamTest { } @Test - @DisplayName("stage narrations roll the bubble: each stage finishes its own bubble, final answer excludes them") + @DisplayName("post-tool narrations roll the bubble; the pre-tool opener is dropped, final answer excludes both") void stageNarrationsRollBubbles() throws Exception { TestableAdapter adapter = newAdapter("{\"progress_interval_ms\": 0}"); seedReplyContext(adapter, "alice", "req-1", "stream-1"); @@ -176,18 +176,74 @@ class WeComProcessStreamTest { List> streamBodies = streamBodies(adapter.drainFrames()); List> finished = streamBodies.stream() .filter(s -> Boolean.TRUE.equals(s.get("finish"))).toList(); - // Three finished bubbles in chronological order: narration #1, - // narration #2, final answer — each on its own stream id. - assertEquals(3, finished.size(), "each stage plus the final answer closes one bubble"); - assertTrue(String.valueOf(finished.get(0).get("content")).contains("我先查一下当前时间")); - assertTrue(String.valueOf(finished.get(1).get("content")).contains("再查会议室")); - assertTrue(String.valueOf(finished.get(2).get("content")).contains("已预约")); - assertEquals(3, finished.stream().map(s -> s.get("id")).distinct().count(), + // Two finished bubbles: the observation-grounded narration #2 and the + // final answer. Narration #1 ran before any tool observation and tools + // ran after it — a pre-tool rehearsal never becomes a permanent bubble + // (it stays visible only transiently in the live progress snapshot). + assertEquals(2, finished.size(), + "grounded narration + final answer close one bubble each; the rehearsal closes none"); + assertTrue(finished.stream().noneMatch( + s -> String.valueOf(s.get("content")).contains("我先查一下当前时间")), + "the pre-tool rehearsal must not finalize a bubble of its own"); + assertTrue(String.valueOf(finished.get(0).get("content")).contains("再查会议室")); + assertTrue(String.valueOf(finished.get(1).get("content")).contains("已预约")); + assertEquals(2, finished.stream().map(s -> s.get("id")).distinct().count(), "each finished bubble must ride its own stream id"); - // The first narration finalizes the original placeholder stream. + // The first published narration finalizes the original placeholder stream. assertEquals("stream-1", finished.get(0).get("id")); } + @Test + @DisplayName("a pre-tool rehearsal pending at stream end is dropped once a grounded answer exists") + void preToolRehearsalDroppedForGroundedAnswer() throws Exception { + TestableAdapter adapter = newAdapter("{\"progress_interval_ms\": 0}"); + seedReplyContext(adapter, "alice", "req-1", "stream-1"); + + // The model writes a full predicted result (fabricated numbers) before + // its first tool call; the real observation then produces the answer. + Flux stream = Flux.just( + StreamDelta.segmentOnly("环境监测结果：温度 29.0°C，湿度 63.0%。需要进一步操作吗？", null), + StreamDelta.event("tool_call_started", + Map.of("toolCallId", "c1", "toolName", "query_env")), + StreamDelta.event("tool_call_completed", + Map.of("toolCallId", "c1", "toolName", "query_env", "success", true)), + new StreamDelta("接口返回为空，所有会议室均无环境数据。", null)); + + String result = adapter.processStream(stream, inbound("alice"), "wecom:alice"); + + assertEquals("接口返回为空，所有会议室均无环境数据。", result); + List> finished = streamBodies(adapter.drainFrames()).stream() + .filter(s -> Boolean.TRUE.equals(s.get("finish"))).toList(); + assertEquals(1, finished.size(), "only the grounded answer may close a bubble"); + String content = String.valueOf(finished.get(0).get("content")); + assertTrue(content.contains("接口返回为空")); + assertFalse(content.contains("29.0"), "the fabricated rehearsal must never reach the user: " + content); + } + + @Test + @DisplayName("a pre-tool narration is still published when the turn produced no answer at all") + void preToolNarrationKeptWhenNoAnswer() throws Exception { + TestableAdapter adapter = newAdapter("{\"progress_interval_ms\": 0}"); + seedReplyContext(adapter, "alice", "req-1", "stream-1"); + + // No content after the tool ran — the narration is everything the user + // gets (approval park / stop / empty answer). With no replacement + // content it is not superseded, so it must still close the bubble. + Flux stream = Flux.just( + StreamDelta.segmentOnly("我先调用工具查询状态：", null), + StreamDelta.event("tool_call_started", + Map.of("toolCallId", "c1", "toolName", "query_env")), + StreamDelta.event("tool_call_completed", + Map.of("toolCallId", "c1", "toolName", "query_env", "success", true))); + + assertEquals("", adapter.processStream(stream, inbound("alice"), "wecom:alice")); + + List> finished = streamBodies(adapter.drainFrames()).stream() + .filter(s -> Boolean.TRUE.equals(s.get("finish"))).toList(); + assertEquals(1, finished.size(), "the narration must close the bubble when nothing else can"); + assertTrue(String.valueOf(finished.get(0).get("content")).contains("我先调用工具查询状态")); + } + @Test @DisplayName("stream_progress=false degrades to accumulate-then-send with no interim overwrites") void progressDisabledDegrades() throws Exception {