diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java index fdada970..6955a0ed 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java @@ -764,22 +764,42 @@ public class AgentService { // ==================== StreamDelta ==================== public record StreamDelta(String content, String thinking, String eventType, Map eventData, - boolean persistenceOnly, boolean segmentOnly) { + boolean persistenceOnly, boolean segmentOnly, ContentKind kind) { // 兼容构造器（广播+持久化） public StreamDelta(String content, String thinking) { - this(content, thinking, null, null, false, false); + this(content, thinking, null, null, false, false, null); } // 显式 5-参构造器：保留旧调用点对 (content, thinking, eventType, eventData, persistenceOnly) 的兼容 public StreamDelta(String content, String thinking, String eventType, Map eventData, boolean persistenceOnly) { - this(content, thinking, eventType, eventData, persistenceOnly, false); + this(content, thinking, eventType, eventData, persistenceOnly, false, null); + } + + // 兼容构造器：kind 出现之前的 6 参 canonical 形态 + public StreamDelta(String content, String thinking, String eventType, + Map eventData, boolean persistenceOnly, boolean segmentOnly) { + this(content, thinking, eventType, eventData, persistenceOnly, segmentOnly, null); } /** 仅用于持久化，不再广播（内容已由 NodeStreamingChatHelper 实时广播过） */ public static StreamDelta persistOnly(String content, String thinking) { - return new StreamDelta(content, thinking, null, null, true, false); + return new StreamDelta(content, thinking, null, null, true, false, null); + } + + /** {@link #persistOnly(String, String)} 带内容语义标注的变体。 */ + public static StreamDelta persistOnly(String content, String thinking, ContentKind kind) { + return new StreamDelta(content, thinking, null, null, true, false, kind); + } + + /** + * Final-answer content of the terminal turn. {@code alreadyStreamed} + * decides broadcast suppression exactly like the persistOnly/plain + * split at the emission sites did before the kind tag existed. + */ + public static StreamDelta finalAnswer(String content, boolean alreadyStreamed) { + return new StreamDelta(content, null, null, null, alreadyStreamed, false, ContentKind.FINAL_ANSWER); } /** @@ -803,15 +823,20 @@ public class AgentService { * persisted content field via this flavor. */ public static StreamDelta segmentOnly(String content, String thinking) { - return new StreamDelta(content, thinking, null, null, true, true); + return new StreamDelta(content, thinking, null, null, true, true, null); + } + + /** {@link #segmentOnly(String, String)} 带内容语义标注的变体。 */ + public static StreamDelta segmentOnly(String content, String thinking, ContentKind kind) { + return new StreamDelta(content, thinking, null, null, true, true, kind); } public static StreamDelta empty() { - return new StreamDelta(null, null, null, null, false, false); + return new StreamDelta(null, null, null, null, false, false, null); } public static StreamDelta event(String type, Map data) { - return new StreamDelta(null, null, type, data, false, false); + return new StreamDelta(null, null, type, data, false, false, null); } public boolean isEvent() { diff --git a/mateclaw-server/src/main/java/vip/mate/agent/ContentKind.java b/mateclaw-server/src/main/java/vip/mate/agent/ContentKind.java new file mode 100644 index 00000000..11ed51ac --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/ContentKind.java @@ -0,0 +1,42 @@ +package vip.mate.agent; + +import java.util.Locale; + +/** + * Semantic category of a content-bearing stream delta, assigned once at the + * producer (the agent graph) where the classification inputs — whether the + * completion carried tool calls and whether any tool observation preceded the + * text this turn — are definitively known. + * + *

Downstream consumers (web segment persistence, IM channel adapters, the + * SSE client) MUST read this tag instead of re-deriving the category from + * stream structure. Deltas from producers that predate this tag carry + * {@code null}; consumers fall back to their legacy structural handling in + * that case. + */ +public enum ContentKind { + + /** + * Text emitted in a completion that also carries tool calls, before any + * tool observation this turn. Not grounded in this turn's results — it may + * be process narration or a fully fabricated "rehearsal" of the outcome. + * Provisional: replaced by the next content of the same turn if one + * arrives, kept only when the turn produces no later content at all. + */ + PRE_TOOL_NARRATION, + + /** + * Intermediate narration emitted after at least one tool observation this + * turn (even when the same completion issues further tool calls). Grounded + * in real results; never replaced. + */ + GROUNDED_NARRATION, + + /** Final-answer text of the terminal turn. */ + FINAL_ANSWER; + + /** Stable lower-case token used in persisted segment metadata and SSE payloads. */ + public String wireName() { + return name().toLowerCase(Locale.ROOT); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java index 369dff8a..f7b60c60 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java @@ -13,6 +13,7 @@ import reactor.core.publisher.Mono; import vip.mate.agent.AgentService; import vip.mate.agent.AgentState; import vip.mate.agent.BaseAgent; +import vip.mate.agent.ContentKind; import vip.mate.agent.delegation.DelegatedUsageAccumulator; import vip.mate.agent.GraphEventPublisher; import vip.mate.agent.StructuredStreamCapable; @@ -251,15 +252,16 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC String streamed = output.state().value(STREAMED_CONTENT).orElse(""); if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) { lastEmittedStreamedContent.set(streamed); - deltas.add(streamedContentDelta(isFinalAnswerTurn, streamed)); + deltas.add(streamedContentDelta(isFinalAnswerTurn, + output.state().value(NEEDS_TOOL_CALL, false), + output.state().value(CURRENT_ITERATION, 0), + streamed)); } if (isFinalAnswerTurn && finalAnswerEmitted.compareAndSet(false, true)) { String answer = extractFinalAnswer(output); if (answer != null && !answer.isEmpty()) { - deltas.add(contentAlreadyStreamed - ? AgentService.StreamDelta.persistOnly(answer, null) - : new AgentService.StreamDelta(answer, null)); + deltas.add(AgentService.StreamDelta.finalAnswer(answer, contentAlreadyStreamed)); } } String thinking = extractFinalThinking(output); @@ -422,15 +424,16 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC String streamed = output.state().value(STREAMED_CONTENT).orElse(""); if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) { lastEmittedStreamedContent.set(streamed); - deltas.add(streamedContentDelta(isFinalAnswerTurn, streamed)); + deltas.add(streamedContentDelta(isFinalAnswerTurn, + output.state().value(NEEDS_TOOL_CALL, false), + output.state().value(CURRENT_ITERATION, 0), + streamed)); } if (isFinalAnswerTurn && finalAnswerEmitted.compareAndSet(false, true)) { String answer = extractFinalAnswer(output); if (answer != null && !answer.isEmpty()) { - deltas.add(contentAlreadyStreamed - ? AgentService.StreamDelta.persistOnly(answer, null) - : new AgentService.StreamDelta(answer, null)); + deltas.add(AgentService.StreamDelta.finalAnswer(answer, contentAlreadyStreamed)); } } @@ -641,15 +644,37 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC * renderers (copy / TTS / history reload) showing the full text. * * + *

Beyond flavor, this is the single assignment point for the delta's + * {@link ContentKind}: the graph is the only layer that definitively knows + * whether the completion carried tool calls ({@code NEEDS_TOOL_CALL}) and + * whether any tool observation preceded the text this turn + * ({@code CURRENT_ITERATION} — ObservationNode increments it after each + * observed round, so iteration 0 means "no observation yet"). Downstream + * consumers read the tag instead of re-deriving it from stream structure. + * + *

    + *
  • terminal turn → {@code FINAL_ANSWER};
    • + *
  • completion carries tool calls and no observation happened yet this + * turn → {@code PRE_TOOL_NARRATION} (provisional, may be replaced by + * the turn's next content);
    • + *
  • otherwise → {@code GROUNDED_NARRATION} (follows an observation, or + * closed its completion without tool calls — never replaced).
    • + *
+ * *

Package-private so the unit test can pin the decision without standing * up a full StateGraph fixture. Returning {@code null} for blank input is the * caller's responsibility — this helper just decides flavor for non-blank * content. */ - static AgentService.StreamDelta streamedContentDelta(boolean isFinalAnswerTurn, String streamed) { - return isFinalAnswerTurn - ? AgentService.StreamDelta.persistOnly(streamed, null) - : AgentService.StreamDelta.segmentOnly(streamed, null); + static AgentService.StreamDelta streamedContentDelta(boolean isFinalAnswerTurn, boolean carriesToolCalls, + int iteration, String streamed) { + if (isFinalAnswerTurn) { + return AgentService.StreamDelta.persistOnly(streamed, null, ContentKind.FINAL_ANSWER); + } + ContentKind kind = carriesToolCalls && iteration == 0 + ? ContentKind.PRE_TOOL_NARRATION + : ContentKind.GROUNDED_NARRATION; + return AgentService.StreamDelta.segmentOnly(streamed, null, kind); } private boolean hasFinalAnswer(NodeOutput output) { 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 748d979c..daa40aaf 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 @@ -192,13 +192,20 @@ public final class AgentStreamAccumulator { } // segments: 追加到当前 running content segment，或创建新的 var seg = findLastRunning("content"); - if (seg != null) { - seg.put("text", seg.getOrDefault("text", "") + delta.content()); - } else { + if (seg == null) { finalizeRunningSegments("thinking"); - var s = newSegment("content"); - s.put("text", delta.content()); - segments.add(s); + seg = newSegment("content"); + seg.put("text", delta.content()); + segments.add(seg); + } else { + seg.put("text", seg.getOrDefault("text", "") + delta.content()); + } + // Producer-assigned content semantics (first writer wins — a + // segment never legitimately changes kind mid-flight). Absent on + // deltas from producers that predate the tag; consumers fall back + // to structural detection for such segments. + if (delta.kind() != null && !seg.containsKey("kind")) { + seg.put("kind", delta.kind().wireName()); } } diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/StateGraphReActAgentStreamedContentDeltaTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/StateGraphReActAgentStreamedContentDeltaTest.java index c9dd092a..3aa5302c 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/graph/StateGraphReActAgentStreamedContentDeltaTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/StateGraphReActAgentStreamedContentDeltaTest.java @@ -3,8 +3,11 @@ package vip.mate.agent.graph; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import vip.mate.agent.AgentService; +import vip.mate.agent.ContentKind; +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; /** @@ -21,49 +24,94 @@ import static org.junit.jupiter.api.Assertions.assertTrue; * read {@code content}, not segments, so {@code segmentOnly} for that case would * shrink the visible message to the warning alone. * - *

The helper under test embodies the corrected contract. + *

The helper is also the single assignment point of {@link ContentKind}: the + * graph knows definitively whether the completion carried tool calls and whether + * any observation preceded the text this turn, so downstream consumers read the + * tag instead of re-deriving the category from stream structure. */ class StateGraphReActAgentStreamedContentDeltaTest { @Test - @DisplayName("intermediate iteration (no FINAL_ANSWER yet) → segmentOnly — narration stays out of content") - void intermediateIteration_routedToSegmentsOnly() { + @DisplayName("iteration 0 with tool calls → segmentOnly + PRE_TOOL_NARRATION — provisional rehearsal") + void preToolNarration_taggedProvisional() { AgentService.StreamDelta d = StateGraphReActAgent.streamedContentDelta( /* isFinalAnswerTurn */ false, - "I'll search for X."); + /* carriesToolCalls */ true, + /* iteration */ 0, + "先加载 skill，然后逐个查询。"); assertTrue(d.persistenceOnly(), "segmentOnly implies persistenceOnly — no re-broadcast (NodeStreamingChatHelper already pushed it)"); assertTrue(d.segmentOnly(), "intermediate narration MUST set segmentOnly so content.append is skipped"); - // Sanity: the content payload survives the wrap. - org.junit.jupiter.api.Assertions.assertEquals("I'll search for X.", d.content()); + assertEquals(ContentKind.PRE_TOOL_NARRATION, d.kind(), + "text alongside tool calls with zero observations this turn is not grounded — provisional"); + assertEquals("先加载 skill，然后逐个查询。", d.content()); } @Test - @DisplayName("evidence-insufficient terminal turn (FINAL_ANSWER set) → persistOnly — answer body persists to content") + @DisplayName("iteration ≥1 narration → segmentOnly + GROUNDED_NARRATION even when the completion issues more tool calls") + void postObservationNarration_taggedGrounded() { + AgentService.StreamDelta d = StateGraphReActAgent.streamedContentDelta( + false, /* carriesToolCalls */ true, /* iteration */ 1, "第一间空闲，继续查下一间。"); + + assertTrue(d.segmentOnly()); + assertEquals(ContentKind.GROUNDED_NARRATION, d.kind(), + "an observation already happened this turn — the narration is grounded and never replaced"); + } + + @Test + @DisplayName("iteration 0 without tool calls (non-terminal) → GROUNDED_NARRATION — conservative, never replaced") + void noToolCallCompletion_taggedGrounded() { + AgentService.StreamDelta d = StateGraphReActAgent.streamedContentDelta( + false, /* carriesToolCalls */ false, /* iteration */ 0, "narrative without tools"); + + assertEquals(ContentKind.GROUNDED_NARRATION, d.kind(), + "a completion that closed without tool calls has no later observation to defer to — keep it"); + } + + @Test + @DisplayName("evidence-insufficient terminal turn (FINAL_ANSWER set) → persistOnly + FINAL_ANSWER kind") void evidenceInsufficientFinalTurn_routedToPersistOnly() { // Regression: STREAMED_CONTENT here is the rejected answer body; FINAL_ANSWER // is only the "[证据不足]" warning. Persisting the streamed body keeps // mate_message.content readable through single-segment renderers. AgentService.StreamDelta d = StateGraphReActAgent.streamedContentDelta( - /* isFinalAnswerTurn */ true, + /* isFinalAnswerTurn */ true, false, 2, "The answer is 42. References: [1] [2] [3]."); assertTrue(d.persistenceOnly(), "persistOnly suppresses re-broadcast — content was already streamed live"); assertFalse(d.segmentOnly(), "persistOnly variant MUST NOT set segmentOnly — content.append needs to run"); - org.junit.jupiter.api.Assertions.assertEquals( - "The answer is 42. References: [1] [2] [3].", d.content()); + assertEquals(ContentKind.FINAL_ANSWER, d.kind()); + assertEquals("The answer is 42. References: [1] [2] [3].", d.content()); + } + + @Test + @DisplayName("finalAnswer factory carries FINAL_ANSWER kind in both broadcast flavors") + void finalAnswerFactory_taggedFinal() { + AgentService.StreamDelta streamed = AgentService.StreamDelta.finalAnswer("done", true); + assertTrue(streamed.persistenceOnly(), "already-streamed answer must not re-broadcast"); + assertEquals(ContentKind.FINAL_ANSWER, streamed.kind()); + + AgentService.StreamDelta fresh = AgentService.StreamDelta.finalAnswer("done", false); + assertFalse(fresh.persistenceOnly(), "un-streamed answer still broadcasts"); + assertEquals(ContentKind.FINAL_ANSWER, fresh.kind()); + } + + @Test + @DisplayName("legacy factories keep kind null — pre-tag producers stay distinguishable") + void legacyFactories_kindNull() { + assertNull(AgentService.StreamDelta.segmentOnly("x", null).kind()); + assertNull(AgentService.StreamDelta.persistOnly("x", null).kind()); + assertNull(new AgentService.StreamDelta("x", null).kind()); } @Test @DisplayName("both flavors leave thinking null — STREAMED_CONTENT routing only carries text content") void thinkingFieldNeverSet() { - org.junit.jupiter.api.Assertions.assertNull( - StateGraphReActAgent.streamedContentDelta(false, "x").thinking()); - org.junit.jupiter.api.Assertions.assertNull( - StateGraphReActAgent.streamedContentDelta(true, "x").thinking()); + assertNull(StateGraphReActAgent.streamedContentDelta(false, true, 0, "x").thinking()); + assertNull(StateGraphReActAgent.streamedContentDelta(true, false, 1, "x").thinking()); } } diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/AgentStreamAccumulatorKindTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/AgentStreamAccumulatorKindTest.java new file mode 100644 index 00000000..00beb518 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/AgentStreamAccumulatorKindTest.java @@ -0,0 +1,101 @@ +package vip.mate.channel.web; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import vip.mate.agent.AgentService.StreamDelta; +import vip.mate.agent.ContentKind; + +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.assertTrue; + +/** + * Pins that producer-assigned {@link ContentKind} tags survive segment + * accumulation into persisted {@code metadata.segments}, and that untagged + * deltas (pre-tag producers) leave the field absent so consumers can fall + * back to structural detection. + */ +class AgentStreamAccumulatorKindTest { + + private static final AgentStreamAccumulator.Sink NOOP_SINK = new AgentStreamAccumulator.Sink() { + @Override public void broadcast(String conversationId, String eventName, Object payload) { } + @Override public void updatePhase(String conversationId, String phase) { } + }; + + private static JsonNode segmentsOf(AgentStreamAccumulator acc, ObjectMapper mapper) throws Exception { + return mapper.readTree(acc.toMetadataJson()).path("segments"); + } + + @Test + @DisplayName("kind tags land on persisted content segments; tool segments stay untagged") + void kindTagsPersisted() throws Exception { + ObjectMapper mapper = new ObjectMapper(); + AgentStreamAccumulator acc = new AgentStreamAccumulator(mapper, NOOP_SINK); + String cid = "conv-1"; + + acc.accept(StreamDelta.segmentOnly("先查询两间会议室的环境数据。", null, + ContentKind.PRE_TOOL_NARRATION), cid); + acc.accept(StreamDelta.event("tool_call_started", + Map.of("toolCallId", "t1", "toolName", "envQuery", "arguments", "{}")), cid); + acc.accept(StreamDelta.event("tool_call_completed", + Map.of("toolCallId", "t1", "toolName", "envQuery", "result", "data={}", "success", true)), cid); + acc.accept(StreamDelta.finalAnswer("接口返回为空，无环境数据。", true), cid); + + JsonNode segments = segmentsOf(acc, mapper); + assertEquals(3, segments.size(), "content + tool_call + content"); + assertEquals("pre_tool_narration", segments.get(0).path("kind").asText()); + assertEquals("tool_call", segments.get(1).path("type").asText()); + assertFalse(segments.get(1).has("kind"), "kind is content-segment semantics only"); + assertEquals("final_answer", segments.get(2).path("kind").asText()); + + assertEquals("接口返回为空，无环境数据。", acc.getContent(), + "segmentOnly narration must stay out of the persisted top-level content"); + } + + @Test + @DisplayName("untagged deltas leave kind absent — legacy consumers keep structural fallback") + void untaggedDeltaHasNoKind() throws Exception { + ObjectMapper mapper = new ObjectMapper(); + AgentStreamAccumulator acc = new AgentStreamAccumulator(mapper, NOOP_SINK); + + acc.accept(StreamDelta.segmentOnly("legacy narration", null), "conv-2"); + + JsonNode segments = segmentsOf(acc, mapper); + assertEquals(1, segments.size()); + assertFalse(segments.get(0).has("kind")); + } + + @Test + @DisplayName("first writer wins — appended deltas cannot re-kind a running segment") + void firstWriterWins() throws Exception { + ObjectMapper mapper = new ObjectMapper(); + AgentStreamAccumulator acc = new AgentStreamAccumulator(mapper, NOOP_SINK); + String cid = "conv-3"; + + acc.accept(StreamDelta.segmentOnly("part one ", null, ContentKind.GROUNDED_NARRATION), cid); + acc.accept(StreamDelta.segmentOnly("part two", null, ContentKind.PRE_TOOL_NARRATION), cid); + + JsonNode segments = segmentsOf(acc, mapper); + assertEquals(1, segments.size(), "second delta appends into the running segment"); + assertEquals("grounded_narration", segments.get(0).path("kind").asText()); + assertTrue(segments.get(0).path("text").asText().endsWith("part two")); + } + + @Test + @DisplayName("kind fills in late when the segment opener was untagged") + void lateKindFillsUntaggedSegment() throws Exception { + ObjectMapper mapper = new ObjectMapper(); + AgentStreamAccumulator acc = new AgentStreamAccumulator(mapper, NOOP_SINK); + String cid = "conv-4"; + + acc.accept(StreamDelta.segmentOnly("opener ", null), cid); + acc.accept(StreamDelta.segmentOnly("tail", null, ContentKind.GROUNDED_NARRATION), cid); + + JsonNode segments = segmentsOf(acc, mapper); + assertEquals("grounded_narration", segments.get(0).path("kind").asText()); + } +}