Add files via upload

internal/multiagent/eino_adk_run_loop.go

@@ -573,6 +573,8 @@ func runEinoADKAgentLoop(ctx context.Context, args *einoADKRunLoopArgs, baseMsgs
 			var subAssistantBuf string
 			var subReplyStreamID string
 			var mainAssistantBuf string
+			// 已通过 response_delta 推到前端的正文（与 monitor.js normalizeStreamingDeltaJs 累积一致）
+			var mainAssistWireAccum string
 			var mainAssistDupTarget string // 非空表示本段主助手流需缓冲至 EOF，与 execute 输出比对去重
 			var reasoningBuf string
 			var prevReasoningDisplay string // UI 用：剥离 Claude 内部 signature 尾缀后的累计展示
@@ -681,6 +683,7 @@ func runEinoADKAgentLoop(ctx context.Context, args *einoADKRunLoopArgs, baseMsgs
 										"einoAgent":       ev.AgentName,
 										"orchestration":   orchMode,
 									})
+									mainAssistWireAccum, _ = normalizeStreamingDelta(mainAssistWireAccum, contentDelta)
 								}
 							}
 						} else if !streamsMainAssistant(ev.AgentName) {
@@ -726,21 +729,29 @@ func runEinoADKAgentLoop(ctx context.Context, args *einoADKRunLoopArgs, baseMsgs
 						}
 					} else if s != "" {
 						if progress != nil {
-							progress("response_start", "", map[string]interface{}{
+							// 仅用 TrimSpace 与 execute 比对；推到 UI 的必须是 mainAssistantBuf，
-								"conversationId":     conversationID,
+							// 否则尾部空白/换行与已流式前缀不一致时，前端 normalize 会走拼接路径造成叠字。
-								"mcpExecutionIds":    snapshotMCPIDs(),
+							_, eofTail := normalizeStreamingDelta(mainAssistWireAccum, mainAssistantBuf)
-								"messageGeneratedBy": "eino:" + ev.AgentName,
+							if eofTail != "" {
-								"einoRole":           "orchestrator",
+								if !streamHeaderSent {
-								"einoAgent":          ev.AgentName,
+									progress("response_start", "", map[string]interface{}{
-								"orchestration":      orchMode,
+										"conversationId":     conversationID,
-							})
+										"mcpExecutionIds":    snapshotMCPIDs(),
-							progress("response_delta", s, map[string]interface{}{
+										"messageGeneratedBy": "eino:" + ev.AgentName,
-								"conversationId":  conversationID,
+										"einoRole":           "orchestrator",
-								"mcpExecutionIds": snapshotMCPIDs(),
+										"einoAgent":          ev.AgentName,
-								"einoRole":        "orchestrator",
+										"orchestration":      orchMode,
-								"einoAgent":       ev.AgentName,
+									})
-								"orchestration":   orchMode,
+								}
-							})
+								progress("response_delta", eofTail, map[string]interface{}{
+									"conversationId":  conversationID,
+									"mcpExecutionIds": snapshotMCPIDs(),
+									"einoRole":        "orchestrator",
+									"einoAgent":       ev.AgentName,
+									"orchestration":   orchMode,
+								})
+								mainAssistWireAccum, _ = normalizeStreamingDelta(mainAssistWireAccum, eofTail)
+							}
 						}
 						lastAssistant = s
 						runAccumulatedMsgs = append(runAccumulatedMsgs, schema.AssistantMessage(s, nil))

internal/multiagent/normalize_streaming_eof_test.go

@@ -0,0 +1,22 @@
+package multiagent
+
+import (
+	"strings"
+	"testing"
+)
+
+// Eino execute 去重分支 EOF flush 须以 mainAssistantBuf 为基准计算 tail，
+// 若误用 TrimSpace(mainAssistantBuf)，会与已推前缀在空白处失配，normalize 走拼接路径叠字。
+func TestNormalizeStreamingDelta_eofTailUsesRawBufNotTrim(t *testing.T) {
+	wireAccum := "phrase "
+	rawFull := "phrase \n"
+	_, tail := normalizeStreamingDelta(wireAccum, rawFull)
+	if want := "\n"; tail != want {
+		t.Fatalf("tail=%q want %q", tail, want)
+	}
+
+	nextWrong, badTail := normalizeStreamingDelta(wireAccum, strings.TrimSpace(rawFull))
+	if badTail != "phrase" || nextWrong != "phrase phrase" {
+		t.Fatalf("trimmed full vs wire prefix mismatch should concat-append; got next=%q badTail=%q", nextWrong, badTail)
+	}
+}