opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-13 03:13:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(wecom): drop pre-tool rehearsal narrations from permanent bubbles

Browse Source
This commit is contained in:
matevip 2026-08-04 06:14:55 -04:00
parent 77dd866243
commit 54a2c3c0a3
2 changed files with 133 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java
View File

@ -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<WeComReplyContext> liveCtx = new AtomicReference<>(initialCtx);
AtomicReference<String> pendingNarration = new AtomicReference<>();
AtomicReference<StagedNarration> 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());
}
/**

74
mateclaw-server/src/test/java/vip/mate/channel/wecom/WeComProcessStreamTest.java
View File

@ -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<Map<String, Object>> streamBodies = streamBodies(adapter.drainFrames());
List<Map<String, Object>> 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<StreamDelta> 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<Map<String, Object>> 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<StreamDelta> 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<Map<String, Object>> 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 {