diff --git a/provider/accumulate.go b/provider/accumulate.go index 044b4c3..024ee71 100644 --- a/provider/accumulate.go +++ b/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 = ©In - 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 } diff --git a/provider/accumulate_parallel_test.go b/provider/accumulate_parallel_test.go new file mode 100644 index 0000000..6f58bb2 --- /dev/null +++ b/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]) + } +} diff --git a/providers/anthropic/stream_parallel_tools_test.go b/providers/anthropic/stream_parallel_tools_test.go new file mode 100644 index 0000000..eefda30 --- /dev/null +++ b/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]) + } +}