fix: merge list entries by logical index instead of physical position (#3201)

src/openai/lib/streaming/_deltas.py

@@ -3,9 +3,36 @@
 from ..._utils import is_dict, is_list
 
 
+def _is_placeholder(entry: object) -> bool:
+    """Detect a gap-filler placeholder that should be replaced in-place.
+
+    When a sparse tool-call stream emits index 0 then 2, the gap at index 1
+    is padded with an empty ``{}``.  After the snapshot is round-tripped
+    through ``model_dump`` (which happens on the next chunk), that placeholder
+    is no longer empty — it becomes a dict of unset tool-call fields such as
+    ``{"id": None, "function": None, "type": None}``.  Both forms must be
+    detected so a later-arriving entry at the same index *replaces* the
+    placeholder instead of being inserted before it (which would shift
+    higher-index entries and break ``tool_calls[index]`` lookups).
+    """
+    if not is_dict(entry):
+        return False
+    # Empty placeholder from the padding path.
+    if not entry:
+        return True
+    # Dumped placeholder: every value is None (or the dict is empty).
+    return all(v is None for v in entry.values())
+
+
 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:
+            # When the first chunk contains a list with multiple entries at the
+            # same index (e.g. from speculative decoding), storing it directly
+            # would leave duplicate entries that later merges can't fix. (#3201)
+            # Coalesce duplicate-index entries before storing.
+            if is_list(delta_value) and len(delta_value) > 1:
+                delta_value = _coalesce_list_by_index(delta_value)
             acc[key] = delta_value
             continue
 
@@ -49,16 +76,94 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
                 if not isinstance(index, int):
                     raise TypeError(f"Unexpected, list delta entry `index` value is not an integer; {index}")
 
-                try:
-                    acc_entry = acc_value[index]
-                except IndexError:
-                    acc_value.insert(index, delta_entry)
-                else:
-                    if not is_dict(acc_entry):
-                        raise TypeError("not handled yet")
+                # Merge by logical index, not physical position. (#3201)
+                # When the first chunk contains multiple entries with the same
+                # index (e.g. from speculative decoding), the physical position
+                # does not match the logical index. Find the existing entry by
+                # its index field and merge into it.
+                #
+                # If acc_value already contains duplicate-index entries
+                # (e.g. from a prior chunk that wasn't coalesced), merge into
+                # all of them so none are stranded.
+                found = False
+                for i, existing in enumerate(acc_value):
+                    if is_dict(existing) and existing.get("index") == index:
+                        acc_value[i] = accumulate_delta(existing, delta_entry)
+                        found = True
 
-                    acc_value[index] = accumulate_delta(acc_entry, delta_entry)
+                if not found:
+                    # Add the new entry.  Don't assume the logical index is a
+                    # safe physical slot — if acc_value already has entries at
+                    # higher indexes (e.g. [{"index": 1, ...}] and index 0
+                    # arrives), acc_value[index] would overwrite the existing
+                    # entry.  Place the entry at the position matching the
+                    # logical index so downstream code that does
+                    # tool_calls[index] (treating logical index as physical
+                    # position) reads the right entry.
+                    if len(acc_value) <= index:
+                        while len(acc_value) < index:
+                            acc_value.append({})
+                        acc_value.append(delta_entry)
+                    else:
+                        # The list is large enough but no entry has this
+                        # index.  If the slot at `index` is a placeholder
+                        # (empty {} or a dumped placeholder with only None
+                        # values from a model_dump round-trip), replace it
+                        # in-place.  Otherwise insert at the correct
+                        # position to keep the list addressable by logical
+                        # index.
+                        existing = acc_value[index]
+                        if _is_placeholder(existing):
+                            acc_value[index] = delta_entry
+                        else:
+                            acc_value.insert(index, delta_entry)
 
         acc[key] = acc_value
 
     return acc
+
+
+def _coalesce_list_by_index(lst: list[object]) -> list[object]:
+    """Merge list entries that share the same ``index`` field into a single entry.
+
+    When the first streamed chunk contains multiple entries with the same
+    ``index`` (e.g. from speculative decoding), storing the list directly would
+    leave duplicate entries. This function coalesces them by merging entries
+    with the same index using :func:`accumulate_delta`, so the snapshot starts
+    in a clean state. (#3201)
+
+    The result is sorted by the ``index`` field so the list stays addressable
+    by logical index — downstream code does ``tool_calls[index]`` treating
+    logical index as physical position.
+    """
+    result: list[object] = []
+    for entry in lst:
+        if not is_dict(entry):
+            result.append(entry)
+            continue
+        index = entry.get("index")
+        if not isinstance(index, int):
+            result.append(entry)
+            continue
+        # Find an existing entry with the same index
+        found = False
+        for i, existing in enumerate(result):
+            if is_dict(existing) and existing.get("index") == index:
+                result[i] = accumulate_delta(existing, entry)
+                found = True
+                break
+        if not found:
+            # Place at the position matching the logical index, padding
+            # with empty dicts if needed, so the list is addressable by
+            # logical index.
+            while len(result) <= index:
+                result.append({})
+            # Replace the placeholder at `index` (empty {} or a dumped
+            # placeholder with only None values from a model_dump round-trip)
+            # or shift if occupied by a real entry.
+            existing = result[index]
+            if _is_placeholder(existing):
+                result[index] = entry
+            else:
+                result.insert(index, entry)
+    return result

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

@@ -22,9 +22,9 @@
     FunctionToolCallArgumentsDoneEvent,
     FunctionToolCallArgumentsDeltaEvent,
 )
-from .._deltas import accumulate_delta
+from .._deltas import accumulate_delta, _coalesce_list_by_index
 from ...._types import Omit, IncEx, omit
-from ...._utils import is_given, consume_sync_iterator, consume_async_iterator
+from ...._utils import is_list, is_given, consume_sync_iterator, consume_async_iterator
 from ...._compat import model_dump
 from ...._models import build, construct_type
 from ..._parsing import (
@@ -409,13 +409,19 @@ def _accumulate_chunk(self, chunk: ChatCompletionChunk) -> ParsedChatCompletionS
                     elif TYPE_CHECKING:  # type: ignore[unreachable]
                         assert_never(prev_tool)
             except IndexError:
+                # A new choice appeared that wasn't in the initial chunk.
+                # Coalesce tool_calls by index to handle duplicate-index entries
+                # from speculative decoding, same as _convert_initial_chunk_into_snapshot.
+                delta_dict = choice.delta.to_dict()
+                if is_list(delta_dict.get("tool_calls")):
+                    delta_dict["tool_calls"] = _coalesce_list_by_index(cast("list[object]", delta_dict["tool_calls"]))
                 choice_snapshot = cast(
                     ParsedChoiceSnapshot,
                     construct_type(
                         type_=ParsedChoiceSnapshot,
                         value={
                             **choice.model_dump(exclude_unset=True, exclude={"delta"}),
-                            "message": choice.delta.to_dict(),
+                            "message": delta_dict,
                         },
                     ),
                 )
@@ -742,9 +748,17 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh
     choices = cast("list[object]", data["choices"])
 
     for choice in chunk.choices:
+        message_dict = choice.delta.to_dict()
+        # Coalesce duplicate-index tool_calls in the initial chunk. (#3201)
+        # When the first chunk contains multiple tool_calls with the same index
+        # (e.g. from speculative decoding), storing them directly would leave
+        # duplicate entries that later merges can't fix.
+        tool_calls = message_dict.get("tool_calls")
+        if is_list(tool_calls) and len(tool_calls) > 1:
+            message_dict["tool_calls"] = _coalesce_list_by_index(tool_calls)
         choices[choice.index] = {
             **choice.model_dump(exclude_unset=True, exclude={"delta"}),
-            "message": choice.delta.to_dict(),
+            "message": message_dict,
         }
 
     return cast(