Fix streaming tool_call delta accumulation with duplicate indexes

src/openai/lib/streaming/_assistants.py

@@ -980,12 +980,44 @@ def accumulate_event(
 def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> dict[object, object]:
     for key, delta_value in delta.items():
         if key not in acc:
-            acc[key] = delta_value
+            # Normalize lists of dicts with index fields before storing
+            # to handle cases like tool_calls with duplicate indexes
+            # in the first chunk
+            if is_list(delta_value) and all(is_dict(x) for x in delta_value):
+                normalized: list[object] = []
+                index_map: dict[int, int] = {}
+                for entry in delta_value:
+                    idx = entry.get("index")
+                    if idx is not None and isinstance(idx, int) and idx in index_map:
+                        existing_pos = index_map[idx]
+                        normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry)
+                    else:
+                        if idx is not None and isinstance(idx, int):
+                            index_map[idx] = len(normalized)
+                        normalized.append(entry)
+                acc[key] = normalized
+            else:
+                acc[key] = delta_value
             continue
 
         acc_value = acc[key]
         if acc_value is None:
-            acc[key] = delta_value
+            # Normalize lists of dicts with index fields before storing
+            if is_list(delta_value) and all(is_dict(x) for x in delta_value):
+                normalized = []
+                index_map = {}
+                for entry in delta_value:
+                    idx = entry.get("index")
+                    if idx is not None and isinstance(idx, int) and idx in index_map:
+                        existing_pos = index_map[idx]
+                        normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry)
+                    else:
+                        if idx is not None and isinstance(idx, int):
+                            index_map[idx] = len(normalized)
+                        normalized.append(entry)
+                acc[key] = normalized
+            else:
+                acc[key] = delta_value
             continue
 
         # the `index` property is used in arrays of objects so it should
@@ -1011,7 +1043,33 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
                 acc_value.extend(delta_value)
                 continue
 
-            for delta_entry in delta_value:
+            # Normalize delta entries by index before merging.
+            # When multiple entries share the same index (e.g., tool_calls
+            # with duplicate indexes in the first chunk), they must be merged
+            # together first so that subsequent chunks merge into the correct
+            # logical position rather than stranding entries at duplicate indexes.
+            normalized_delta: list[object] = []
+            index_map: dict[int, int] = {}
+            for entry in delta_value:
+                if not is_dict(entry):
+                    raise TypeError(f"Unexpected list delta entry is not a dictionary: {entry}")
+
+                try:
+                    idx = entry["index"]
+                except KeyError as exc:
+                    raise RuntimeError(f"Expected list delta entry to have an `index` key; {entry}") from exc
+
+                if not isinstance(idx, int):
+                    raise TypeError(f"Unexpected, list delta entry `index` value is not an integer; {idx}")
+
+                if idx in index_map:
+                    existing_pos = index_map[idx]
+                    normalized_delta[existing_pos] = accumulate_delta(normalized_delta[existing_pos], entry)
+                else:
+                    index_map[idx] = len(normalized_delta)
+                    normalized_delta.append(entry)
+
+            for delta_entry in normalized_delta:
                 if not is_dict(delta_entry):
                     raise TypeError(f"Unexpected list delta entry is not a dictionary: {delta_entry}")
 

src/openai/lib/streaming/_deltas.py

@@ -6,12 +6,44 @@
 def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> dict[object, object]:
     for key, delta_value in delta.items():
         if key not in acc:
-            acc[key] = delta_value
+            # Normalize lists of dicts with index fields before storing
+            # to handle cases like tool_calls with duplicate indexes
+            # in the first chunk (see issue #3201)
+            if is_list(delta_value) and all(is_dict(x) for x in delta_value):
+                normalized: list[object] = []
+                index_map: dict[int, int] = {}
+                for entry in delta_value:
+                    idx = entry.get("index")
+                    if idx is not None and isinstance(idx, int) and idx in index_map:
+                        existing_pos = index_map[idx]
+                        normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry)
+                    else:
+                        if idx is not None and isinstance(idx, int):
+                            index_map[idx] = len(normalized)
+                        normalized.append(entry)
+                acc[key] = normalized
+            else:
+                acc[key] = delta_value
             continue
 
         acc_value = acc[key]
         if acc_value is None:
-            acc[key] = delta_value
+            # Normalize lists of dicts with index fields before storing
+            if is_list(delta_value) and all(is_dict(x) for x in delta_value):
+                normalized = []
+                index_map = {}
+                for entry in delta_value:
+                    idx = entry.get("index")
+                    if idx is not None and isinstance(idx, int) and idx in index_map:
+                        existing_pos = index_map[idx]
+                        normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry)
+                    else:
+                        if idx is not None and isinstance(idx, int):
+                            index_map[idx] = len(normalized)
+                        normalized.append(entry)
+                acc[key] = normalized
+            else:
+                acc[key] = delta_value
             continue
 
         # the `index` property is used in arrays of objects so it should
@@ -37,7 +69,33 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
                 acc_value.extend(delta_value)
                 continue
 
-            for delta_entry in delta_value:
+            # Normalize delta entries by index before merging.
+            # When multiple entries share the same index (e.g., tool_calls
+            # with duplicate indexes in the first chunk), they must be merged
+            # together first so that subsequent chunks merge into the correct
+            # logical position rather than stranding entries at duplicate indexes.
+            normalized_delta: list[object] = []
+            index_map: dict[int, int] = {}
+            for entry in delta_value:
+                if not is_dict(entry):
+                    raise TypeError(f"Unexpected list delta entry is not a dictionary: {entry}")
+
+                try:
+                    idx = entry["index"]
+                except KeyError as exc:
+                    raise RuntimeError(f"Expected list delta entry to have an `index` key; {entry}") from exc
+
+                if not isinstance(idx, int):
+                    raise TypeError(f"Unexpected, list delta entry `index` value is not an integer; {idx}")
+
+                if idx in index_map:
+                    existing_pos = index_map[idx]
+                    normalized_delta[existing_pos] = accumulate_delta(normalized_delta[existing_pos], entry)
+                else:
+                    index_map[idx] = len(normalized_delta)
+                    normalized_delta.append(entry)
+
+            for delta_entry in normalized_delta:
                 if not is_dict(delta_entry):
                     raise TypeError(f"Unexpected list delta entry is not a dictionary: {delta_entry}")
 

src/openai/lib/streaming/chat/_completions.py

@@ -744,7 +744,10 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh
     for choice in chunk.choices:
         choices[choice.index] = {
             **choice.model_dump(exclude_unset=True, exclude={"delta"}),
-            "message": choice.delta.to_dict(),
+            "message": accumulate_delta(
+                cast("dict[object, object]", {}),
+                cast("dict[object, object]", choice.delta.to_dict()),
+            ),
         }
 
     return cast(

tests/test_deltas.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from openai.lib.streaming._deltas import accumulate_delta
+
+
+def test_duplicate_index_in_first_chunk() -> None:
+    """
+    Regression test for issue #3201:
+    When the first chunk contains multiple tool_calls entries with the same
+    `index`, they should be merged into a single entry by logical index,
+    not stored as separate physical entries.
+    """
+    chunk1 = {
+        "tool_calls": [
+            {
+                "index": 0,
+                "id": "call_1",
+                "type": "function",
+                "function": {"name": "get_weather", "arguments": ""},
+            },
+            {
+                "index": 0,
+                "function": {"arguments": ' {"cit'},
+            },
+        ]
+    }
+
+    chunk2 = {
+        "tool_calls": [
+            {
+                "index": 0,
+                "function": {"arguments": 'y": "London"}'},
+            }
+        ]
+    }
+
+    acc: dict = {}
+    acc = accumulate_delta(acc, chunk1)
+    assert len(acc["tool_calls"]) == 1, f"Expected 1 tool call, got {len(acc['tool_calls'])}"
+    assert acc["tool_calls"][0]["function"]["arguments"] == ' {"cit'
+
+    acc = accumulate_delta(acc, chunk2)
+    assert len(acc["tool_calls"]) == 1, f"Expected 1 tool call, got {len(acc['tool_calls'])}"
+    assert acc["tool_calls"][0]["function"]["arguments"] == ' {"city": "London"}'
+
+
+def test_multiple_tool_calls_different_indexes() -> None:
+    """Multiple tool calls with different indexes should remain separate."""
+    chunk1 = {
+        "tool_calls": [
+            {"index": 0, "id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": ""}},
+            {"index": 1, "id": "call_2", "type": "function", "function": {"name": "get_time", "arguments": ""}},
+        ]
+    }
+
+    chunk2 = {
+        "tool_calls": [
+            {"index": 0, "function": {"arguments": ' {"city": "London"}'}},
+            {"index": 1, "function": {"arguments": ' {"tz": "UTC"}'}},
+        ]
+    }
+
+    acc: dict = {}
+    acc = accumulate_delta(acc, chunk1)
+    assert len(acc["tool_calls"]) == 2
+
+    acc = accumulate_delta(acc, chunk2)
+    assert len(acc["tool_calls"]) == 2
+    assert acc["tool_calls"][0]["function"]["arguments"] == ' {"city": "London"}'
+    assert acc["tool_calls"][1]["function"]["arguments"] == ' {"tz": "UTC"}'
+
+
+def test_sparse_indexes() -> None:
+    """Sparse indexes (index 2 arriving before index 1) should be handled correctly.
+
+    Note: The accumulator inserts at the physical position matching the index,
+    so index 2 is stored at position 0 when the list is empty, then index 1
+    is inserted at position 1 (append). This is existing behavior.
+    """
+    chunk1 = {
+        "tool_calls": [
+            {"index": 2, "id": "call_3", "type": "function", "function": {"name": "search", "arguments": ""}},
+        ]
+    }
+
+    chunk2 = {
+        "tool_calls": [
+            {"index": 1, "id": "call_2", "type": "function", "function": {"name": "get_time", "arguments": ""}},
+        ]
+    }
+
+    acc: dict = {}
+    acc = accumulate_delta(acc, chunk1)
+    assert len(acc["tool_calls"]) == 1
+    assert acc["tool_calls"][0]["index"] == 2
+
+    acc = accumulate_delta(acc, chunk2)
+    assert len(acc["tool_calls"]) == 2
+    assert acc["tool_calls"][0]["index"] == 2
+    assert acc["tool_calls"][1]["index"] == 1
+
+
+def test_normal_case_no_duplicates() -> None:
+    """Normal case with no duplicate indexes should work as before."""
+    chunk1 = {
+        "tool_calls": [
+            {"index": 0, "id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": ""}},
+        ]
+    }
+
+    chunk2 = {
+        "tool_calls": [
+            {"index": 0, "function": {"arguments": ' {"city": "London"}'}},
+        ]
+    }
+
+    acc: dict = {}
+    acc = accumulate_delta(acc, chunk1)
+    assert len(acc["tool_calls"]) == 1
+
+    acc = accumulate_delta(acc, chunk2)
+    assert len(acc["tool_calls"]) == 1
+    assert acc["tool_calls"][0]["function"]["arguments"] == ' {"city": "London"}'
+
+
+def test_three_duplicate_indexes_in_first_chunk() -> None:
+    """Three entries with the same index in the first chunk should all be merged."""
+    chunk1 = {
+        "tool_calls": [
+            {"index": 0, "id": "call_1", "type": "function", "function": {"name": "test", "arguments": ""}},
+            {"index": 0, "function": {"arguments": ' {"a"'}},
+            {"index": 0, "function": {"arguments": ': 1}'}},
+        ]
+    }
+
+    acc: dict = {}
+    acc = accumulate_delta(acc, chunk1)
+    assert len(acc["tool_calls"]) == 1
+    assert acc["tool_calls"][0]["function"]["arguments"] == ' {"a": 1}'