This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 0c2e15ace fix(ai): scope Claude text deduplication to the streamed
block (#6682)
0c2e15ace is described below
commit 0c2e15ace5406353af9616ff2dcf2b6c802ad1d6
Author: 健康宝 <[email protected]>
AuthorDate: Thu Oct 15 19:08:15 2026 +0800
fix(ai): scope Claude text deduplication to the streamed block (#6682)
---
.../studio/ops/ai/ClaudeCodeStreamParser.java | 64 +++++++++++++++--
.../studio/ops/ai/ClaudeCodeStreamParserTest.java | 81 ++++++++++++++++++++++
2 files changed, 139 insertions(+), 6 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParser.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParser.java
index 033a77d5e..6bc4e0f7e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParser.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParser.java
@@ -210,12 +210,28 @@ final class ClaudeCodeStreamParser {
/** Upstream types already reported as unhandled, so each is reported at
most once per run. */
private final Set<String> reportedUnhandled = new LinkedHashSet<>();
- /** True once a {@code text_delta} arrived: the aggregated assistant text
is then a duplicate. */
+ /** True once a {@code text_delta} arrived: fallback for captures whose
frames carry no
+ * {@code message.id}, where per-message discrimination is impossible. */
private boolean textDeltaSeen;
/** Same guard for reasoning. */
private boolean thinkingDeltaSeen;
+ /** {@code message.id}s that streamed a non-empty {@code text_delta}:
their aggregated
+ * assistant text is a duplicate. A different message's aggregate is not.
*/
+ private final Set<String> streamedTextMessageIds = new LinkedHashSet<>();
+
+ /** Same guard for reasoning blocks. */
+ private final Set<String> streamedThinkingMessageIds = new
LinkedHashSet<>();
+
+ /** The message currently being streamed, from the inner {@code
message_start}; null between. */
+ private String currentStreamingMessageId;
+
+ /** {@code message.id}s announced by an inner {@code message_start}: only
for these can a
+ * delta be attributed with certainty. Captures that never show a {@code
message_start} fall
+ * back to the run-level flags. */
+ private final Set<String> observedMessageIds = new LinkedHashSet<>();
+
/** True once any {@link AgentEvent.TextDelta} was emitted, from either
source. */
private boolean textEmitted;
@@ -363,7 +379,18 @@ final class ClaudeCodeStreamParser {
case "content_block_start" -> onContentBlockStart(event);
case "content_block_delta" -> onContentBlockDelta(event);
case "content_block_stop" -> onContentBlockStop(event);
- case "message_start", "message_delta", "message_stop", "ping" ->
List.of();
+ case "message_start" -> {
+ // The boundary of a new message: remember which one is being
+ // streamed so a delta can be attributed to exactly it.
+ currentStreamingMessageId =
event.path("message").path("id").asText("");
+ if (StringUtils.hasText(currentStreamingMessageId)) {
+ observedMessageIds.add(currentStreamingMessageId);
+ } else {
+ currentStreamingMessageId = null;
+ }
+ yield List.of();
+ }
+ case "message_delta", "message_stop", "ping" -> List.of();
default -> unhandled(STREAM_EVENT + "/" + eventType);
};
}
@@ -393,20 +420,28 @@ final class ClaudeCodeStreamParser {
String deltaType = delta.path("type").asText("");
return switch (deltaType) {
case DELTA_TEXT -> {
- textDeltaSeen = true;
String text = delta.path("text").asText("");
if (text.isEmpty()) {
+ // An empty delta proves nothing was streamed: suppressing
+ // the aggregate on its account dropped real text.
yield List.of();
}
+ textDeltaSeen = true;
+ if (currentStreamingMessageId != null) {
+ streamedTextMessageIds.add(currentStreamingMessageId);
+ }
textEmitted = true;
yield List.of(new AgentEvent.TextDelta(text));
}
case DELTA_THINKING -> {
- thinkingDeltaSeen = true;
String thinking = delta.path("thinking").asText("");
if (thinking.isEmpty()) {
yield List.of();
}
+ thinkingDeltaSeen = true;
+ if (currentStreamingMessageId != null) {
+ streamedThinkingMessageIds.add(currentStreamingMessageId);
+ }
yield List.of(new AgentEvent.ThinkingDelta(thinking,
ThinkingSource.MODEL));
}
case DELTA_INPUT_JSON -> {
@@ -456,7 +491,18 @@ final class ClaudeCodeStreamParser {
String blockType = block.path("type").asText("");
switch (blockType) {
case BLOCK_TEXT -> {
- if (textDeltaSeen) {
+ // Deduplicate only the same message's streamed text: a
+ // run-level flag dropped every later message's aggregate
+ // once any earlier message had streamed. When the capture
+ // never showed a message_start for this id, attribution
+ // is impossible and the run-level flag stays the rule.
+ String messageId =
node.path("message").path("id").asText("");
+ boolean attributionKnown = StringUtils.hasText(messageId)
+ && observedMessageIds.contains(messageId);
+ boolean streamedAlready = attributionKnown
+ ? streamedTextMessageIds.contains(messageId)
+ : textDeltaSeen;
+ if (streamedAlready) {
continue;
}
String text = block.path("text").asText("");
@@ -466,7 +512,13 @@ final class ClaudeCodeStreamParser {
}
}
case BLOCK_THINKING -> {
- if (thinkingDeltaSeen) {
+ String messageId =
node.path("message").path("id").asText("");
+ boolean attributionKnown = StringUtils.hasText(messageId)
+ && observedMessageIds.contains(messageId);
+ boolean streamedAlready = attributionKnown
+ ? streamedThinkingMessageIds.contains(messageId)
+ : thinkingDeltaSeen;
+ if (streamedAlready) {
continue;
}
String thinking = block.path("thinking").asText("");
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParserTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParserTest.java
index a41eecafb..95ee62e46 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParserTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParserTest.java
@@ -416,6 +416,87 @@ class ClaudeCodeStreamParserTest {
}
/** The content of every event of one type, concatenated in arrival order.
*/
+ @Test
+ void aggregateTextSurvivesAcrossMessagesTest() {
+ // Message one streams its text; message two only aggregates. The old
+ // run-level flag treated every later aggregate as a duplicate of the
+ // earlier stream and dropped it.
+ ClaudeCodeStreamParser parser = new ClaudeCodeStreamParser();
+ List<AgentEvent> events = new ArrayList<>();
+ for (String frame : new String[] {
+ """
+
{"type":"stream_event","session_id":"s-1","event":{"type":"message_start","message":{"id":"msg_1"}}}""",
+ """
+
{"type":"stream_event","session_id":"s-1","event":{"type":"content_block_delta","index":0,\
+ "delta":{"type":"text_delta","text":"first"}}}""",
+ """
+
{"type":"assistant","session_id":"s-1","message":{"id":"msg_1","role":"assistant",\
+ "content":[{"type":"text","text":"first"}]}}""",
+ """
+
{"type":"stream_event","session_id":"s-1","event":{"type":"message_start","message":{"id":"msg_2"}}}""",
+ """
+
{"type":"assistant","session_id":"s-1","message":{"id":"msg_2","role":"assistant",\
+ "content":[{"type":"text","text":"second"}]}}"""}) {
+ events.addAll(parser.parseLine(frame));
+ }
+ assertThat(joined(events,
AgentEvent.TextDelta.class)).isEqualTo("firstsecond");
+ }
+
+ @Test
+ void anEmptyDeltaDoesNotSuppressTheAggregateTest() {
+ ClaudeCodeStreamParser parser = new ClaudeCodeStreamParser();
+ List<AgentEvent> events = new ArrayList<>();
+ for (String frame : new String[] {
+ """
+
{"type":"stream_event","session_id":"s-1","event":{"type":"message_start","message":{"id":"msg_1"}}}""",
+ """
+
{"type":"stream_event","session_id":"s-1","event":{"type":"content_block_delta","index":0,\
+ "delta":{"type":"text_delta","text":""}}}""",
+ """
+
{"type":"assistant","session_id":"s-1","message":{"id":"msg_1","role":"assistant",\
+ "content":[{"type":"text","text":"real aggregate"}]}}"""}) {
+ events.addAll(parser.parseLine(frame));
+ }
+ assertThat(joined(events, AgentEvent.TextDelta.class)).isEqualTo("real
aggregate");
+ }
+
+ @Test
+ void sameMessageAggregateIsStillDeduplicatedTest() {
+ ClaudeCodeStreamParser parser = new ClaudeCodeStreamParser();
+ List<AgentEvent> events = new ArrayList<>();
+ for (String frame : new String[] {
+ """
+
{"type":"stream_event","session_id":"s-1","event":{"type":"message_start","message":{"id":"msg_1"}}}""",
+ """
+
{"type":"stream_event","session_id":"s-1","event":{"type":"content_block_delta","index":0,\
+ "delta":{"type":"text_delta","text":"first"}}}""",
+ """
+
{"type":"assistant","session_id":"s-1","message":{"id":"msg_1","role":"assistant",\
+ "content":[{"type":"text","text":"first"}]}}"""}) {
+ events.addAll(parser.parseLine(frame));
+ }
+ assertThat(joined(events,
AgentEvent.TextDelta.class)).isEqualTo("first");
+ }
+
+ @Test
+ void idlessCapturesKeepTheRunLevelFallbackTest() {
+ // Captures without message ids cannot discriminate per message: the
+ // aggregate of the streamed text stays suppressed run-wide, exactly
+ // as before.
+ ClaudeCodeStreamParser parser = new ClaudeCodeStreamParser();
+ List<AgentEvent> events = new ArrayList<>();
+ for (String frame : new String[] {
+ """
+
{"type":"stream_event","session_id":"s-1","event":{"type":"content_block_delta","index":0,\
+ "delta":{"type":"text_delta","text":"first"}}}""",
+ """
+
{"type":"assistant","session_id":"s-1","message":{"role":"assistant",\
+ "content":[{"type":"text","text":"first"}]}}"""}) {
+ events.addAll(parser.parseLine(frame));
+ }
+ assertThat(joined(events,
AgentEvent.TextDelta.class)).isEqualTo("first");
+ }
+
private static String joined(List<AgentEvent> events, Class<?> type) {
StringBuilder joined = new StringBuilder();
for (AgentEvent event : events) {
