diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ProvisionalContentTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/ProvisionalContentTracker.java new file mode 100644 index 00000000..7796a5ea --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/ProvisionalContentTracker.java @@ -0,0 +1,180 @@ +package vip.mate.channel; + +import io.micrometer.core.instrument.Metrics; +import lombok.extern.slf4j.Slf4j; +import vip.mate.agent.ContentKind; + +import java.util.List; +import java.util.Map; + +/** + * Single authority for the "provisional narration" lifecycle shared by every + * user-facing surface. + * + *

A {@link ContentKind#PRE_TOOL_NARRATION} span is written before any tool + * observation of its turn, in a completion that goes on to call tools — it may + * be process narration or a fully fabricated rehearsal of the result. The + * policy, identical everywhere: + * + *

+ * + *

Grounded narrations and final answers never stage — they publish + * directly. Producers that predate the kind tag emit {@code null} kinds; the + * streaming API accepts a caller-supplied structural fallback signal for that + * case, and the segment-marking API leaves untagged timelines to the legacy + * structural detector. + * + *

Instances are single-turn and not thread-safe — create one per stream + * consumption, confine to the consuming thread (matches how channel adapters + * drain a turn's {@code Flux} today). + */ +@Slf4j +public final class ProvisionalContentTracker { + + /** Marker value stored in {@code supersededReason} — same wire value the + * legacy structural detector writes, so the UI needs no new vocabulary. */ + public static final String REASON_PRE_TOOL_CONTENT_REPLACED = + "pre_tool_content_replaced_by_post_tool_answer"; + + private static final String METRIC_SUPERSEDED = "mateclaw.narration.superseded"; + + /** Where the supersede happened — metric tag, one value per surface. */ + private final String surface; + + private String pendingText; + private boolean pendingProvisional; + + public ProvisionalContentTracker(String surface) { + this.surface = surface; + } + + /** + * 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). + * + * @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) + */ + public String stageNarration(String text, ContentKind kind, boolean observedSinceLast) { + boolean provisional = kind != null + ? kind == ContentKind.PRE_TOOL_NARRATION + : !observedSinceLast; + String previous = pendingText; + boolean previousProvisional = pendingProvisional; + pendingText = text; + pendingProvisional = provisional; + if (previous == null) { + return null; + } + if (previousProvisional) { + recordSuperseded(previous); + return null; + } + return previous; + } + + /** + * 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). + * + * @param hasFinalContent whether the turn produced a final answer — with + * one, a provisional narration is superseded; with + * none, even a provisional narration commits (no + * replacement exists) + */ + public String settle(boolean hasFinalContent) { + String text = pendingText; + boolean provisional = pendingProvisional; + pendingText = null; + pendingProvisional = false; + if (text == null) { + return null; + } + if (provisional && hasFinalContent) { + recordSuperseded(text); + return null; + } + return text; + } + + private void recordSuperseded(String text) { + log.info("[{}] provisional narration superseded by later content ({} chars dropped from permanent output)", + surface, text.length()); + Metrics.counter(METRIC_SUPERSEDED, "surface", surface).increment(); + } + + // ==================== Persisted-timeline marking ==================== + + /** + * Whether the persisted segments timeline carries producer-assigned kind + * tags — i.e. whether {@link #markSuperseded(List, String)} is applicable + * or the caller should fall back to structural detection. + */ + public static boolean hasKindTags(List> segments) { + if (segments == null) { + return false; + } + for (Map seg : segments) { + if ("content".equals(seg.get("type")) && seg.get("kind") != null) { + return true; + } + } + return false; + } + + /** + * Kind-driven counterpart of the structural supersede scan: a + * {@code pre_tool_narration} content segment is marked superseded by the + * first content segment that follows it; grounded narrations and final + * answers are never marked. Mutates segment maps in place with the same + * three keys the structural detector writes ({@code superseded}, + * {@code supersededBySegmentId}, {@code supersededReason}). + */ + public static void markSuperseded(List> segments, String surface) { + if (segments == null || segments.isEmpty()) { + return; + } + String preToolWire = ContentKind.PRE_TOOL_NARRATION.wireName(); + for (int i = 0; i < segments.size(); i++) { + Map seg = segments.get(i); + if (!"content".equals(seg.get("type")) + || !preToolWire.equals(seg.get("kind")) + || Boolean.TRUE.equals(seg.get("superseded"))) { + continue; + } + Map replacement = nextContent(segments, i + 1); + if (replacement == null) { + continue; // turn produced no later content — the narration stands + } + seg.put("superseded", true); + seg.put("supersededBySegmentId", String.valueOf(replacement.getOrDefault("id", ""))); + seg.put("supersededReason", REASON_PRE_TOOL_CONTENT_REPLACED); + log.info("[{}] provisional narration segment {} superseded by segment {}", + surface, seg.get("id"), replacement.get("id")); + Metrics.counter(METRIC_SUPERSEDED, "surface", surface).increment(); + } + } + + private static Map nextContent(List> segments, int from) { + for (int i = from; i < segments.size(); i++) { + if ("content".equals(segments.get(i).get("type"))) { + return segments.get(i); + } + } + return null; + } +} 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 index daa40aaf..8e0ca63c 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/AgentStreamAccumulator.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/AgentStreamAccumulator.java @@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import vip.mate.agent.AgentService; import vip.mate.agent.GraphEventPublisher; +import vip.mate.channel.ProvisionalContentTracker; import vip.mate.workspace.conversation.model.MessageContentPart; import java.util.ArrayList; @@ -472,7 +473,14 @@ public final class AgentStreamAccumulator { public synchronized String toMetadataJson() { finalizeToolCalls(); finalizeRunningSegments("thinking", "content", "tool_call"); - SegmentSupersedeDetector.markSuperseded(segments); + // Producer-tagged timelines use the kind-driven authority; untagged + // ones (pre-tag producers, replayed legacy turns) keep the structural + // scan as fallback. + if (ProvisionalContentTracker.hasKindTags(segments)) { + ProvisionalContentTracker.markSuperseded(segments, "web"); + } else { + SegmentSupersedeDetector.markSuperseded(segments); + } try { Map metadata = new LinkedHashMap<>(); if (!toolCalls.isEmpty()) { 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 fe4b72dd..f2089e8f 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 @@ -7,6 +7,7 @@ import vip.mate.agent.AgentService.StreamDelta; import vip.mate.channel.AbstractChannelAdapter; import vip.mate.channel.ChannelMessage; import vip.mate.channel.ChannelMessageRouter; +import vip.mate.channel.ProvisionalContentTracker; import vip.mate.channel.StreamingChannelAdapter; import vip.mate.workspace.core.service.ChatUploadLocationResolver; import vip.mate.channel.ExponentialBackoff; @@ -1416,7 +1417,7 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea } }) .blockLast(Duration.ofMinutes(10)); - outcome = new StreamOutcome(null, false, false); + outcome = new StreamOutcome(null, false); } else { outcome = consumeWithProgress(stream, message, replyTarget, ctx, contentAccumulator); } @@ -1432,10 +1433,11 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea // 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(); + String pendingNarration = outcome.tracker() != null + ? outcome.tracker().settle(!finalContent.isBlank()) + : null; if (!finalContent.isBlank()) { - if (pendingNarration != null && !outcome.pendingNarrationPreTool() - && !sameOutboundText(pendingNarration, finalContent)) { + if (pendingNarration != null && !sameOutboundText(pendingNarration, finalContent)) { publishNarrationBubble(replyTarget, pendingNarration); } renderAndSend(replyTarget, finalContent); @@ -1459,39 +1461,14 @@ 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 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 + * @param tracker the narration lifecycle tracker holding the last + * per-stage narration, still unresolved — settled by + * {@code processStream} once the final answer is + * known; {@code null} on the degraded path where + * narration is never staged + * @param approvalPending a tool call is parked on human approval */ - 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; - } - } + private record StreamOutcome(ProvisionalContentTracker tracker, boolean approvalPending) {} /** Compare two outbound texts the way the user sees them (post-filter, trimmed). */ private boolean sameOutboundText(String a, String b) { @@ -1554,7 +1531,7 @@ 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<>(); + ProvisionalContentTracker tracker = new ProvisionalContentTracker("wecom"); AtomicInteger toolResults = new AtomicInteger(); final int[] lastNarrationToolMark = {0}; final long[] lastFlushAt = {0L}; @@ -1575,33 +1552,30 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea // a wall of text, and persisted they pollute the next turn's // 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 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. + // Publishing lags one narration behind via the shared + // lifecycle tracker: the newest narration is only staged + // (visible live in the bubble, not yet finalized) so a later + // decision can still drop it — a pre-tool rehearsal must not + // become an unretractable permanent bubble once grounded + // content follows. The producer-assigned kind decides; for + // untagged deltas the tracker falls back to the observation + // counter (a tool result completed since the last narration + // 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(); - StagedNarration staged = new StagedNarration( - narration, mark > lastNarrationToolMark[0], mark); + boolean observedSinceLast = mark > lastNarrationToolMark[0]; lastNarrationToolMark[0] = mark; - StagedNarration previous = pendingNarration.getAndSet(staged); progress.onNarration(narration); - if (previous != null && !previous.supersededAt(mark)) { + String publishable = tracker.stageNarration(narration, delta.kind(), observedSinceLast); + if (publishable != null) { WeComReplyContext ctx = liveCtx.get(); if (replyContexts.get(replyTarget) == ctx) { - liveCtx.set(rollProgressBubble(replyTarget, ctx, previous.text(), progress)); + liveCtx.set(rollProgressBubble(replyTarget, ctx, publishable, progress)); } else { // Bubble already force-finished (180s ceiling) — the // narration still goes out as a plain message. - sendMessage(replyTarget, previous.text()); + sendMessage(replyTarget, publishable); } } flushNow = true; @@ -1638,11 +1612,7 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea } }).blockLast(Duration.ofMinutes(10)); - StagedNarration last = pendingNarration.get(); - return new StreamOutcome( - last != null ? last.text() : null, - last != null && last.supersededAt(toolResults.get()), - progress.isApprovalPending()); + return new StreamOutcome(tracker, progress.isApprovalPending()); } /** diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ProvisionalContentTrackerTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ProvisionalContentTrackerTest.java new file mode 100644 index 00000000..09c14831 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/ProvisionalContentTrackerTest.java @@ -0,0 +1,189 @@ +package vip.mate.channel; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import vip.mate.agent.ContentKind; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * State machine of the shared provisional-narration lifecycle, plus the + * kind-driven persisted-timeline marking. The streaming scenarios mirror the + * ones the structural detector's tests pin for the web timeline — the two + * paths must agree on every case a producer can actually emit. + */ +class ProvisionalContentTrackerTest { + + // ==================== streaming state machine ==================== + + @Test + @DisplayName("pre-tool narration superseded by the next narration — never published") + void preToolNarrationSupersededByNextNarration() { + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + + assertNull(t.stageNarration("先查询会议室数据。", ContentKind.PRE_TOOL_NARRATION, false), + "first narration has no predecessor to publish"); + String publishable = t.stageNarration("第一间空闲，继续。", ContentKind.GROUNDED_NARRATION, true); + assertNull(publishable, "provisional predecessor is superseded, not published"); + + assertEquals("第一间空闲，继续。", t.settle(true), + "the grounded successor itself settles publishable"); + } + + @Test + @DisplayName("grounded narration publishes when the next narration arrives") + void groundedNarrationPublishesOnNext() { + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + + t.stageNarration("时间拿到了，再查会议室：", ContentKind.GROUNDED_NARRATION, true); + String publishable = t.stageNarration("会议室也查完了。", ContentKind.GROUNDED_NARRATION, true); + assertEquals("时间拿到了，再查会议室：", publishable); + } + + @Test + @DisplayName("pre-tool narration superseded by the final answer at settle") + void preToolNarrationSupersededByFinalAnswer() { + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + + t.stageNarration("环境监测结果：温度 29.0°C。", ContentKind.PRE_TOOL_NARRATION, false); + assertNull(t.settle(true), "fabricated rehearsal must not survive once a grounded answer exists"); + } + + @Test + @DisplayName("pre-tool narration commits when the turn produced no answer at all") + void preToolNarrationCommitsWithoutAnswer() { + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + + t.stageNarration("我先调用工具查询状态：", ContentKind.PRE_TOOL_NARRATION, false); + assertEquals("我先调用工具查询状态：", t.settle(false), + "with no replacement content the narration is everything the user gets"); + } + + @Test + @DisplayName("grounded narration always settles publishable") + void groundedNarrationSettles() { + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + t.stageNarration("查完了，结果如下。", ContentKind.GROUNDED_NARRATION, true); + assertEquals("查完了，结果如下。", t.settle(true)); + } + + @Test + @DisplayName("settle clears state — second settle finds nothing") + void settleClearsState() { + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + t.stageNarration("x", ContentKind.GROUNDED_NARRATION, true); + t.settle(true); + assertNull(t.settle(true)); + } + + @Test + @DisplayName("null kind falls back to the observation-counter rule") + void nullKindFallsBackToCounter() { + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + + // No observation since turn start → provisional (pre-tag online rule). + t.stageNarration("我先查一下：", null, false); + assertNull(t.settle(true), "untagged pre-tool narration still dropped for a grounded answer"); + + // Observation completed since last narration → grounded. + ProvisionalContentTracker t2 = new ProvisionalContentTracker("test"); + t2.stageNarration("拿到结果了。", null, true); + assertEquals("拿到结果了。", t2.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). + ProvisionalContentTracker t = new ProvisionalContentTracker("test"); + t.stageNarration("预演内容", ContentKind.PRE_TOOL_NARRATION, true); + assertNull(t.settle(true), "kind is authoritative — counter signal ignored when tagged"); + } + + // ==================== persisted-timeline marking ==================== + + private static Map content(String id, String kind) { + Map seg = new LinkedHashMap<>(); + seg.put("id", id); + seg.put("type", "content"); + seg.put("text", "t-" + id); + if (kind != null) { + seg.put("kind", kind); + } + return seg; + } + + private static Map tool(String id) { + Map seg = new LinkedHashMap<>(); + seg.put("id", id); + seg.put("type", "tool_call"); + return seg; + } + + @Test + @DisplayName("hasKindTags only reacts to tagged content segments") + void hasKindTagsDetection() { + assertFalse(ProvisionalContentTracker.hasKindTags(null)); + assertFalse(ProvisionalContentTracker.hasKindTags(List.of(content("c0", null), tool("t1")))); + assertTrue(ProvisionalContentTracker.hasKindTags( + List.of(content("c0", "pre_tool_narration")))); + } + + @Test + @DisplayName("pre_tool segment marked superseded by the first later content segment") + void marksPreToolSegment() { + List> segments = new ArrayList<>(List.of( + content("c0", "pre_tool_narration"), + tool("t1"), + content("c2", "final_answer"))); + + ProvisionalContentTracker.markSuperseded(segments, "test"); + + assertEquals(true, segments.get(0).get("superseded")); + assertEquals("c2", segments.get(0).get("supersededBySegmentId")); + assertEquals(ProvisionalContentTracker.REASON_PRE_TOOL_CONTENT_REPLACED, + segments.get(0).get("supersededReason")); + assertFalse(segments.get(2).containsKey("superseded")); + } + + @Test + @DisplayName("grounded and final segments never marked; trailing pre_tool without replacement stands") + void groundedNeverMarkedAndTrailingPreToolStands() { + List> segments = new ArrayList<>(List.of( + content("c0", "grounded_narration"), + tool("t1"), + content("c2", "pre_tool_narration"))); + + ProvisionalContentTracker.markSuperseded(segments, "test"); + + assertFalse(segments.get(0).containsKey("superseded"), + "grounded narration is never replaced"); + assertFalse(segments.get(2).containsKey("superseded"), + "no later content exists — the narration is everything the user gets"); + } + + @Test + @DisplayName("chained tool segments between narration and answer don't block marking") + void chainedToolsBetween() { + List> segments = new ArrayList<>(List.of( + content("c0", "pre_tool_narration"), + tool("t1"), + tool("t2"), + content("c3", "final_answer"))); + + ProvisionalContentTracker.markSuperseded(segments, "test"); + + assertEquals(true, segments.get(0).get("superseded")); + assertEquals("c3", segments.get(0).get("supersededBySegmentId")); + } +}