fix(provider): parallel tool calls with distinct ids no longer merge into one slot

provider/accumulate.go

@@ -207,15 +207,35 @@ func (a *streamAccumulator) mergeTool(idx int, in *ToolCall) {
 		}
 	}
 	if slot == nil {
-		if existing, ok := a.toolByIndex[idx]; ok {
+		// The index fallback is for OpenAI-style streams that name a call
+		// once and then send ID-less fragments at the same index. A delta
+		// that carries a DIFFERENT id than the call already parked at this
+		// index is a distinct call, not a fragment: Anthropic, Bedrock and
+		// the realtime streams emit every tool delta as a one-element slice,
+		// so parallel calls all arrive at idx 0 and only the id tells them
+		// apart. Merging by index there glues the second call's JSON onto
+		// the first and loses the second call outright.
+		if existing, ok := a.toolByIndex[idx]; ok && (in.ID == "" || existing.ID == "" || existing.ID == in.ID) {
 			slot = existing
 		}
 	}
 	if slot == nil {
 		copyIn := *in
 		slot = &copyIn
-		a.toolByIndex[idx] = slot
-		a.toolOrder = append(a.toolOrder, idx)
+		key := idx
+		if _, taken := a.toolByIndex[key]; taken {
+			// Park the new call under the first free key so it keeps its own
+			// slot while toolOrder still records the order it arrived in.
+			key = len(a.toolByIndex)
+			for {
+				if _, taken := a.toolByIndex[key]; !taken {
+					break
+				}
+				key++
+			}
+		}
+		a.toolByIndex[key] = slot
+		a.toolOrder = append(a.toolOrder, key)
 		if in.ID != "" {
 			a.toolByID[in.ID] = slot
 		}

provider/accumulate_parallel_test.go

@@ -0,0 +1,77 @@
+package provider_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/xraph/nexus/provider"
+	"github.com/xraph/nexus/testutil"
+)
+
+// Guards: two parallel tool calls that arrive with distinct IDs but the same
+// slice index must finalize as two calls, each with its own arguments.
+// Anthropic, Bedrock, Gemini Live and the OpenAI realtime stream all emit every
+// tool delta as a one-element slice, so idx is always 0 for every call. Before
+// the fix the second call's ID was unknown, the index fallback matched the
+// first slot, the second call's JSON was glued onto the first's arguments and
+// the second call itself vanished. The model's own turn then failed to
+// round-trip on the next request ("tool_use.input: Input should be an object").
+func TestAccumulator_DistinctIDsAtSameIndexStayApart(t *testing.T) {
+	t.Parallel()
+	one := func(id, name, args string) *provider.StreamChunk {
+		return &provider.StreamChunk{Kind: provider.EventToolCallDelta, Delta: provider.Delta{ToolCalls: []provider.ToolCall{
+			{ID: id, Type: "function", Function: provider.ToolCallFunc{Name: name, Arguments: args}},
+		}}}
+	}
+	chunks := []*provider.StreamChunk{
+		one("toolu_1", "read_history", `{"id": "layer_a",`),
+		one("toolu_1", "read_history", ` "minutes": 30}`),
+		one("toolu_2", "related_resources", `{"id": "mk-1"}`),
+		one("toolu_1", "read_history", ``), // a trailing empty delta must not disturb either slot
+	}
+	resp, err := provider.Accumulate(context.Background(), testutil.NewFakeStream(chunks, nil))
+	if err != nil {
+		t.Fatalf("accumulate: %v", err)
+	}
+	tools := resp.Choices[0].Message.ToolCalls
+	if len(tools) != 2 {
+		t.Fatalf("got %d tool calls, want 2: %+v", len(tools), tools)
+	}
+	if tools[0].ID != "toolu_1" || tools[0].Function.Name != "read_history" || tools[0].Function.Arguments != `{"id": "layer_a", "minutes": 30}` {
+		t.Fatalf("first call = %+v", tools[0])
+	}
+	if tools[1].ID != "toolu_2" || tools[1].Function.Name != "related_resources" || tools[1].Function.Arguments != `{"id": "mk-1"}` {
+		t.Fatalf("second call = %+v", tools[1])
+	}
+}
+
+// Guards: an OpenAI-style stream that names the call once and then streams
+// ID-less argument fragments at the same index keeps merging by index. The
+// distinct-ID rule above must only split slots whose IDs actually differ.
+func TestAccumulator_IDLessFragmentsStillMergeByIndex(t *testing.T) {
+	t.Parallel()
+	chunks := []*provider.StreamChunk{
+		{Delta: provider.Delta{ToolCalls: []provider.ToolCall{
+			{ID: "call_a", Type: "function", Function: provider.ToolCallFunc{Name: "lookup", Arguments: `{"q":`}},
+			{ID: "call_b", Type: "function", Function: provider.ToolCallFunc{Name: "fetch", Arguments: `{"u":`}},
+		}}},
+		{Delta: provider.Delta{ToolCalls: []provider.ToolCall{
+			{Function: provider.ToolCallFunc{Arguments: `"x"}`}},
+			{Function: provider.ToolCallFunc{Arguments: `"y"}`}},
+		}}, FinishReason: "tool_calls"},
+	}
+	resp, err := provider.Accumulate(context.Background(), testutil.NewFakeStream(chunks, nil))
+	if err != nil {
+		t.Fatalf("accumulate: %v", err)
+	}
+	tools := resp.Choices[0].Message.ToolCalls
+	if len(tools) != 2 {
+		t.Fatalf("got %d tool calls, want 2: %+v", len(tools), tools)
+	}
+	if tools[0].ID != "call_a" || tools[0].Function.Arguments != `{"q":"x"}` {
+		t.Fatalf("first call = %+v", tools[0])
+	}
+	if tools[1].ID != "call_b" || tools[1].Function.Arguments != `{"u":"y"}` {
+		t.Fatalf("second call = %+v", tools[1])
+	}
+}

providers/anthropic/stream_parallel_tools_test.go

@@ -0,0 +1,104 @@
+package anthropic_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/xraph/nexus/provider"
+	"github.com/xraph/nexus/providers/anthropic"
+	"github.com/xraph/nexus/testutil"
+)
+
+// Guards: an Anthropic turn with two tool_use blocks (parallel tool calls)
+// accumulates into two provider.ToolCalls, each carrying only its own input
+// JSON. This replays the exact wire sequence that broke in production: the
+// stream tags every input_json_delta with the block's own tool id, but emits
+// each delta as a one-element slice, so the accumulator must key on the id
+// rather than the slice index or the second call is folded into the first.
+func TestStream_ParallelToolUseBlocksAccumulateSeparately(t *testing.T) {
+	mock := testutil.NewMockServer(t)
+
+	blockStart := func(index int, id, name string) string {
+		return testutil.AnthropicEventJSON("content_block_start", map[string]any{
+			"type":  "content_block_start",
+			"index": index,
+			"content_block": map[string]any{
+				"type":  "tool_use",
+				"id":    id,
+				"name":  name,
+				"input": map[string]any{},
+			},
+		})
+	}
+	inputDelta := func(index int, partial string) string {
+		return testutil.AnthropicEventJSON("content_block_delta", map[string]any{
+			"type":  "content_block_delta",
+			"index": index,
+			"delta": map[string]any{
+				"type":         "input_json_delta",
+				"partial_json": partial,
+			},
+		})
+	}
+	blockStop := func(index int) string {
+		return testutil.AnthropicEventJSON("content_block_stop", map[string]any{
+			"type":  "content_block_stop",
+			"index": index,
+		})
+	}
+
+	lines := []string{
+		testutil.AnthropicEventJSON("message_start", map[string]any{
+			"type": "message_start",
+			"message": map[string]any{
+				"id":    "msg_parallel",
+				"model": "claude-sonnet-4-5-20250514",
+			},
+		}),
+		"",
+		blockStart(0, "toolu_1", "read_history"), "",
+		inputDelta(0, `{"id": "layer_a",`), "",
+		inputDelta(0, ` "minutes": 30}`), "",
+		blockStop(0), "",
+		blockStart(1, "toolu_2", "related_resources"), "",
+		inputDelta(1, `{"id": "mk-1"}`), "",
+		blockStop(1), "",
+		testutil.AnthropicEventJSON("message_delta", map[string]any{
+			"type":  "message_delta",
+			"delta": map[string]any{"stop_reason": "tool_use"},
+			"usage": map[string]any{"output_tokens": 40},
+		}),
+		"",
+		testutil.AnthropicEventJSON("message_stop", map[string]any{"type": "message_stop"}),
+		"",
+	}
+	mock.Ctrl.SetStreamHandler(testutil.AnthropicStreamHandler(lines))
+
+	p := anthropic.New("test-key", anthropic.WithBaseURL(mock.Server.URL))
+	ctx := context.Background()
+	stream, err := p.CompleteStream(ctx, &provider.CompletionRequest{
+		Model:     "claude-sonnet-4-5-20250514",
+		Messages:  []provider.Message{{Role: "user", Content: "Two checks, please."}},
+		MaxTokens: 100,
+		Stream:    true,
+	})
+	if err != nil {
+		t.Fatalf("CompleteStream() error: %v", err)
+	}
+	defer func() { _ = stream.Close() }()
+
+	resp, err := provider.Accumulate(ctx, stream)
+	if err != nil {
+		t.Fatalf("Accumulate() error: %v", err)
+	}
+	tools := resp.Choices[0].Message.ToolCalls
+	if len(tools) != 2 {
+		t.Fatalf("got %d tool calls, want 2: %+v", len(tools), tools)
+	}
+	if tools[0].ID != "toolu_1" || tools[0].Function.Name != "read_history" || tools[0].Function.Arguments != `{"id": "layer_a", "minutes": 30}` {
+		t.Fatalf("first call = %+v", tools[0])
+	}
+	if tools[1].ID != "toolu_2" || tools[1].Function.Name != "related_resources" || tools[1].Function.Arguments != `{"id": "mk-1"}` {
+		t.Fatalf("second call = %+v", tools[1])
+	}
+}