Fix missing results for parallel Eino tool calls.

Merge streaming tool outputs by CallID with ConcatMessages, pair same-name historical results, and FIFO-match duplicate IDs so concurrent nmap 1/2 and 2/2 stay distinct.

Co-authored-by: Cursor <cursoragent@cursor.com>

internal/multiagent/eino_adk_run_loop.go

@@ -443,22 +443,41 @@ func nextAgentEventWithContext(ctx context.Context, iter *adk.AsyncIterator[*adk
 
 // recvSchemaMessageStream 消费 ADK Tool 流式结果；ctx 取消时立即返回，避免 amass 等无输出时永久阻塞。
 func recvSchemaMessageStream(ctx context.Context, stream *schema.StreamReader[*schema.Message]) (content, toolCallID, toolName string, recvErr error) {
-	if stream == nil {
-		return "", "", "", nil
+	msgs, recvErr := recvSchemaToolResultMessages(ctx, stream)
+	if len(msgs) == 0 {
+		return "", "", "", recvErr
 	}
-	var buf strings.Builder
-	recvErr = recvEinoSchemaMessageStreamWithContext(ctx, stream, 8, func(chunk *schema.Message) {
-		if chunk.Content != "" {
-			buf.WriteString(chunk.Content)
+	parts := make([]string, 0, len(msgs))
+	for _, msg := range msgs {
+		if msg == nil {
+			continue
 		}
-		if tid := strings.TrimSpace(chunk.ToolCallID); tid != "" {
-			toolCallID = tid
+		parts = append(parts, msg.Content)
+		if id := strings.TrimSpace(msg.ToolCallID); id != "" {
+			toolCallID = id
 		}
-		if name := strings.TrimSpace(chunk.ToolName); name != "" {
+		if name := strings.TrimSpace(msg.ToolName); name != "" {
 			toolName = name
 		}
+	}
+	return strings.Join(parts, ""), toolCallID, toolName, recvErr
+}
+
+// recvSchemaToolResultMessages 先收齐 Tool 流，再用 Eino ConcatMessages 合并。
+// EventSender 一 call 一条流时走 ConcatMessages；并行结果被摊平进同一条流时按 CallID 分列再合并。
+func recvSchemaToolResultMessages(ctx context.Context, stream *schema.StreamReader[*schema.Message]) (msgs []*schema.Message, recvErr error) {
+	if stream == nil {
+		return nil, nil
+	}
+	var chunks []*schema.Message
+	recvErr = recvEinoSchemaMessageStreamWithContext(ctx, stream, 8, func(chunk *schema.Message) {
+		chunks = append(chunks, chunk)
 	})
-	return buf.String(), toolCallID, toolName, recvErr
+	msgs, concatErr := concatToolResultChunks(chunks)
+	if concatErr != nil && recvErr == nil {
+		return nil, concatErr
+	}
+	return msgs, recvErr
 }
 
 func buildEinoCheckpointID(orchMode string) string {

internal/multiagent/eino_adk_run_loop_stream_test.go

@@ -30,6 +30,29 @@ func TestRecvSchemaMessageStream_EOF(t *testing.T) {
 	}
 }
 
+func TestRecvSchemaToolResultMessages_SplitsParallelIDs(t *testing.T) {
+	sr, sw := schema.Pipe[*schema.Message](8)
+	_ = sw.Send(schema.ToolMessage("one-", "tc-1", schema.WithToolName("nmap")), nil)
+	_ = sw.Send(schema.ToolMessage("two-", "tc-2", schema.WithToolName("nmap")), nil)
+	_ = sw.Send(schema.ToolMessage("a", "tc-1", schema.WithToolName("nmap")), nil)
+	_ = sw.Send(schema.ToolMessage("b", "tc-2", schema.WithToolName("nmap")), nil)
+	sw.Close()
+
+	msgs, err := recvSchemaToolResultMessages(context.Background(), sr)
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	if len(msgs) != 2 {
+		t.Fatalf("msgs = %#v, want 2", msgs)
+	}
+	if msgs[0].ToolCallID != "tc-1" || msgs[0].Content != "one-a" {
+		t.Fatalf("msg 0 = %#v", msgs[0])
+	}
+	if msgs[1].ToolCallID != "tc-2" || msgs[1].Content != "two-b" {
+		t.Fatalf("msg 1 = %#v", msgs[1])
+	}
+}
+
 func TestRecvSchemaMessageStream_CapturesToolName(t *testing.T) {
 	sr, sw := schema.Pipe[*schema.Message](4)
 	_ = sw.Send(schema.ToolMessage("hello", "tc-1", schema.WithToolName("execute")), nil)

internal/multiagent/eino_agentic_event_adapter.go

@@ -31,6 +31,11 @@ func adaptAgenticEventToEinoEvents(ev *adk.TypedAgentEvent[*schema.AgenticMessag
 		return []*adk.AgentEvent{base(&adk.AgentOutput{CustomizedOutput: customized})}
 	}
 	if mv.IsStreaming {
+		// Tool 流保持 1 event ↔ 1 MessageStream，对齐 ADK EventSenderToolWrapper：
+		// 每个 CallID 在工具包装层就已经是独立事件。这里不能再按 CallID 现场拆成
+		// 多条 live pipe——drain 会阻塞读完当前流，交错的并行 chunk 会把另一列写满后死锁。
+		// 若上游仍把 ToolsNode 的 MergeStreamReaders 摊成一条流，由
+		// concatToolResultChunks 按列 ConcatMessages 恢复。
 		return []*adk.AgentEvent{base(&adk.AgentOutput{
 			MessageOutput: &adk.MessageVariant{
 				IsStreaming:   true,

internal/multiagent/eino_run_progress_tracker_test.go

@@ -98,6 +98,39 @@ func TestEinoRunProgressTrackerDedupesToolCalls(t *testing.T) {
 	}
 }
 
+func TestEinoRunProgressTrackerDedupesSameToolCallIDsWithDifferentArgs(t *testing.T) {
+	var toolCalls int
+	progress := func(eventType, _ string, _ interface{}) {
+		if eventType == "tool_call" {
+			toolCalls++
+		}
+	}
+	tracker := newEinoRunProgressTracker("deep", "lead", "conv-1", progress, nil, nil)
+	first := &schema.Message{ToolCalls: []schema.ToolCall{{
+		ID:   "call-1",
+		Type: "function",
+		Function: schema.FunctionCall{
+			Name:      "nmap",
+			Arguments: `{"host":"10.0.0.1"}`,
+		},
+	}}}
+	second := &schema.Message{ToolCalls: []schema.ToolCall{{
+		ID:   "call-1",
+		Type: "function",
+		Function: schema.FunctionCall{
+			Name:      "nmap",
+			Arguments: `{"host":"10.0.0.1","ports":"1-1024"}`,
+		},
+	}}}
+
+	tracker.EmitToolCalls(first, "lead", nil)
+	tracker.EmitToolCalls(second, "lead", nil)
+
+	if toolCalls != 1 {
+		t.Fatalf("tool call events = %d, want 1", toolCalls)
+	}
+}
+
 func TestEinoRunProgressTrackerHidesModelOutputRecoveryToolCalls(t *testing.T) {
 	var eventTypes []string
 	var marked []toolCallPendingInfo

internal/multiagent/eino_tool_result_concat.go

@@ -0,0 +1,91 @@
+package multiagent
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/cloudwego/eino/schema"
+)
+
+// concatToolResultChunks 按 Eino 原生语义合并工具结果流：
+//   - 同一 CallID（EventSender 一 call 一 event）：schema.ConcatMessages
+//   - 并行工具被摊进同一条流（ToolsNode MergeStreamReaders 扁平化后）：
+//     按 CallID 分列后再 ConcatMessages，等价于 schema.ConcatMessageArray
+func concatToolResultChunks(chunks []*schema.Message) ([]*schema.Message, error) {
+	if len(chunks) == 0 {
+		return nil, nil
+	}
+	if toolResultChunksShareCallID(chunks) {
+		merged, err := schema.ConcatMessages(chunks)
+		if err != nil {
+			return nil, err
+		}
+		return []*schema.Message{merged}, nil
+	}
+	return concatToolResultChunksByCallID(chunks)
+}
+
+func toolResultChunksShareCallID(chunks []*schema.Message) bool {
+	id := ""
+	for _, chunk := range chunks {
+		if chunk == nil {
+			continue
+		}
+		got := strings.TrimSpace(chunk.ToolCallID)
+		if got == "" {
+			continue
+		}
+		if id == "" {
+			id = got
+			continue
+		}
+		if got != id {
+			return false
+		}
+	}
+	return true
+}
+
+func concatToolResultChunksByCallID(chunks []*schema.Message) ([]*schema.Message, error) {
+	type column struct {
+		key    string
+		chunks []*schema.Message
+	}
+	var ordered []column
+	index := make(map[string]int)
+	lastKey := ""
+	anon := 0
+	for _, chunk := range chunks {
+		if chunk == nil {
+			continue
+		}
+		key := strings.TrimSpace(chunk.ToolCallID)
+		if key == "" {
+			if lastKey != "" {
+				key = lastKey
+			} else {
+				key = fmt.Sprintf("\x00anon-%d", anon)
+				anon++
+			}
+		}
+		if idx, ok := index[key]; ok {
+			ordered[idx].chunks = append(ordered[idx].chunks, chunk)
+		} else {
+			index[key] = len(ordered)
+			ordered = append(ordered, column{key: key, chunks: []*schema.Message{chunk}})
+		}
+		lastKey = key
+	}
+	out := make([]*schema.Message, 0, len(ordered))
+	for _, col := range ordered {
+		merged, err := schema.ConcatMessages(col.chunks)
+		if err != nil {
+			return nil, err
+		}
+		if strings.HasPrefix(col.key, "\x00anon-") {
+			merged.ToolCallID = ""
+		}
+		out = append(out, merged)
+	}
+	return out, nil
+}

internal/multiagent/eino_tool_result_concat_test.go

@@ -0,0 +1,41 @@
+package multiagent
+
+import (
+	"testing"
+
+	"github.com/cloudwego/eino/schema"
+)
+
+func TestConcatToolResultChunksUsesEinoConcatForSingleCall(t *testing.T) {
+	got, err := concatToolResultChunks([]*schema.Message{
+		schema.ToolMessage("hel", "call-1", schema.WithToolName("execute")),
+		schema.ToolMessage("lo", "call-1", schema.WithToolName("execute")),
+	})
+	if err != nil {
+		t.Fatalf("concat: %v", err)
+	}
+	if len(got) != 1 || got[0].ToolCallID != "call-1" || got[0].Content != "hello" || got[0].ToolName != "execute" {
+		t.Fatalf("got = %#v, want one ConcatMessages result", got)
+	}
+}
+
+func TestConcatToolResultChunksSplitsParallelCalls(t *testing.T) {
+	got, err := concatToolResultChunks([]*schema.Message{
+		schema.ToolMessage("nmap 1/2 ", "call-1", schema.WithToolName("nmap")),
+		schema.ToolMessage("nmap 2/2 ", "call-2", schema.WithToolName("nmap")),
+		schema.ToolMessage("22/tcp", "call-1", schema.WithToolName("nmap")),
+		schema.ToolMessage("80/tcp", "call-2", schema.WithToolName("nmap")),
+	})
+	if err != nil {
+		t.Fatalf("concat: %v", err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("got = %#v, want two calls", got)
+	}
+	if got[0].ToolCallID != "call-1" || got[0].Content != "nmap 1/2 22/tcp" {
+		t.Fatalf("call-1 = %#v", got[0])
+	}
+	if got[1].ToolCallID != "call-2" || got[1].Content != "nmap 2/2 80/tcp" {
+		t.Fatalf("call-2 = %#v", got[1])
+	}
+}

internal/multiagent/eino_tool_result_event_handler.go

@@ -42,27 +42,42 @@ func (h *einoToolResultEventHandler) HandleStreaming(mv *adk.MessageVariant, age
 	if h == nil || mv == nil || !mv.IsStreaming || mv.MessageStream == nil || mv.Role != schema.Tool {
 		return false
 	}
-	toolName := strings.TrimSpace(mv.ToolName)
-	content, streamToolCallID, streamToolName, recvErr := recvSchemaMessageStream(h.ctx, mv.MessageStream)
-	if toolName == "" {
-		toolName = streamToolName
+	defaultName := strings.TrimSpace(mv.ToolName)
+	msgs, recvErr := recvSchemaToolResultMessages(h.ctx, mv.MessageStream)
+	if isEinoVoluntaryCancelErr(recvErr) && len(msgs) == 0 {
+		msgs = []*schema.Message{schema.ToolMessage("已中断并继续，当前工具调用已停止。", "", schema.WithToolName(defaultName))}
 	}
-	if isEinoVoluntaryCancelErr(recvErr) && strings.TrimSpace(content) == "" {
-		content = "已中断并继续，当前工具调用已停止。"
+	if len(msgs) == 0 {
+		msgs = []*schema.Message{schema.ToolMessage("", "", schema.WithToolName(defaultName))}
 	}
-	isErr := einoToolResultIsError(toolName, content) || isEinoVoluntaryCancelErr(recvErr)
-	content = einoToolResultBody(content)
-	if streamToolCallID != "" && h.runMessages != nil {
-		h.runMessages.AppendToolMessage(content, streamToolCallID, schema.WithToolName(toolName))
-	}
-	if h.emitter != nil {
-		h.emitter.Emit(h.ctx, toolName, content, streamToolCallID, isErr, agentName)
-	}
-	if recvErr != nil && !isEinoVoluntaryCancelErr(recvErr) && h.logger != nil {
-		h.logger.Warn("eino tool result stream recv error",
-			zap.Error(recvErr),
-			zap.String("agent", agentName),
-			zap.String("tool", toolName))
+	for _, msg := range msgs {
+		if msg == nil {
+			continue
+		}
+		toolName := strings.TrimSpace(msg.ToolName)
+		if toolName == "" {
+			toolName = defaultName
+		}
+		content := msg.Content
+		if isEinoVoluntaryCancelErr(recvErr) && strings.TrimSpace(content) == "" {
+			content = "已中断并继续，当前工具调用已停止。"
+		}
+		isErr := einoToolResultIsError(toolName, content) || isEinoVoluntaryCancelErr(recvErr)
+		content = einoToolResultBody(content)
+		toolCallID := strings.TrimSpace(msg.ToolCallID)
+		if toolCallID != "" && h.runMessages != nil {
+			h.runMessages.AppendToolMessage(content, toolCallID, schema.WithToolName(toolName))
+		}
+		if h.emitter != nil {
+			h.emitter.Emit(h.ctx, toolName, content, toolCallID, isErr, agentName)
+		}
+		if recvErr != nil && !isEinoVoluntaryCancelErr(recvErr) && h.logger != nil {
+			h.logger.Warn("eino tool result stream recv error",
+				zap.Error(recvErr),
+				zap.String("agent", agentName),
+				zap.String("tool", toolName),
+				zap.String("toolCallId", toolCallID))
+		}
 	}
 	if recvErr == nil && h.confirmRecovery != nil {
 		h.confirmRecovery()

internal/multiagent/eino_tool_result_event_handler_test.go

@@ -59,6 +59,54 @@ func TestEinoToolResultEventHandlerHandlesStreamingToolResult(t *testing.T) {
 	}
 }
 
+func TestEinoToolResultEventHandlerSplitsParallelStreamingResults(t *testing.T) {
+	var events []map[string]interface{}
+	runMessages := newEinoRunMessageAccumulator(nil)
+	emitter := newEinoToolResultProgressEmitter(einoToolResultProgressEmitterConfig{
+		ConversationID: "conv-1",
+		Progress: func(eventType, _ string, data interface{}) {
+			if eventType != "tool_result" {
+				return
+			}
+			m, _ := data.(map[string]interface{})
+			events = append(events, m)
+		},
+	})
+	handler := newEinoToolResultEventHandler(einoToolResultEventHandlerConfig{
+		RunMessages: runMessages,
+		Emitter:     emitter,
+	})
+	stream := schema.StreamReaderFromArray([]*schema.Message{
+		{Role: schema.Tool, Content: "nmap 1/2 start ", ToolCallID: "call-1", ToolName: "nmap"},
+		{Role: schema.Tool, Content: "nmap 2/2 start ", ToolCallID: "call-2", ToolName: "nmap"},
+		{Role: schema.Tool, Content: "22/tcp open", ToolCallID: "call-1", ToolName: "nmap"},
+		{Role: schema.Tool, Content: "80/tcp open", ToolCallID: "call-2", ToolName: "nmap"},
+	})
+	mv := &adk.MessageVariant{
+		IsStreaming:   true,
+		Role:          schema.Tool,
+		ToolName:      "nmap",
+		MessageStream: stream,
+	}
+
+	if !handler.HandleStreaming(mv, "worker") {
+		t.Fatal("streaming tool result was not handled")
+	}
+	if len(events) != 2 {
+		t.Fatalf("events = %#v, want two tool_result", events)
+	}
+	if events[0]["toolCallId"] != "call-1" || events[0]["result"] != "nmap 1/2 start 22/tcp open" {
+		t.Fatalf("first event = %#v", events[0])
+	}
+	if events[1]["toolCallId"] != "call-2" || events[1]["result"] != "nmap 2/2 start 80/tcp open" {
+		t.Fatalf("second event = %#v", events[1])
+	}
+	msgs := runMessages.Messages()
+	if len(msgs) != 2 || msgs[0].ToolCallID != "call-1" || msgs[1].ToolCallID != "call-2" {
+		t.Fatalf("run messages = %#v", msgs)
+	}
+}
+
 func TestEinoToolResultEventHandlerHandlesMaterializedToolResult(t *testing.T) {
 	var event map[string]interface{}
 	emitter := newEinoToolResultProgressEmitter(einoToolResultProgressEmitterConfig{

internal/multiagent/runner.go

@@ -908,10 +908,37 @@ func tryEmitToolCallsOnce(
 	if _, ok := seen[sig]; ok {
 		return
 	}
+	if idSig := toolCallsStableIDSignature(msg); idSig != "" {
+		idKey := agentName + "\x1eids\x1e" + idSig
+		if _, ok := seen[idKey]; ok {
+			return
+		}
+		seen[idKey] = struct{}{}
+	}
 	seen[sig] = struct{}{}
 	emitToolCallsFromMessage(msg, agentName, orchestratorName, conversationID, orchMode, progress, subAgentToolStep, mainAgentToolStep, markPending)
 }
 
+func toolCallsStableIDSignature(msg *schema.Message) string {
+	if msg == nil || len(msg.ToolCalls) == 0 {
+		return ""
+	}
+	visible := filterVisibleToolCallsForProgress(msg.ToolCalls)
+	ids := make([]string, 0, len(visible))
+	for _, tc := range visible {
+		id := strings.TrimSpace(tc.ID)
+		if id == "" {
+			continue
+		}
+		ids = append(ids, id)
+	}
+	if len(ids) == 0 {
+		return ""
+	}
+	sort.Strings(ids)
+	return strings.Join(ids, ";")
+}
+
 func emitToolCallsFromMessage(
 	msg *schema.Message,
 	agentName, orchestratorName, conversationID, orchMode string,