diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py
index a5e1317612..38956b2f61 100644
--- a/src/openai/lib/streaming/_deltas.py
+++ b/src/openai/lib/streaming/_deltas.py
@@ -2,15 +2,52 @@
 
 from ..._utils import is_dict, is_list
 
+#: Maximum gap padded between logical tool-call indexes.  A stream with a
+#: huge sparse index (e.g. index 1,000,000) must not allocate storage
+#: proportional to the numeric index; entries beyond this bound are appended
+#: at the end and still found by the index-based merge.
+_MAX_INDEX_PADDING = 1024
+
+
+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:
+            # Coalesce duplicate-index entries before storing so the snapshot
+            # starts in a clean state, and normalize single entries to their
+            # logical slot. (#3201)
+            if is_list(delta_value) and any(is_dict(x) for x in delta_value):
+                delta_value = _coalesce_list_by_index(delta_value)
             acc[key] = delta_value
             continue
 
         acc_value = acc[key]
         if acc_value is None:
+            # Coalesce duplicate-index entries here too — a prior chunk may
+            # have set acc[key] to None via a delta that only contained the
+            # key without a value, and now the actual list arrives. (#3201)
+            if is_list(delta_value) and any(is_dict(x) for x in delta_value):
+                delta_value = _coalesce_list_by_index(delta_value)
             acc[key] = delta_value
             continue
 
@@ -25,7 +62,14 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
             continue
 
         if isinstance(acc_value, str) and isinstance(delta_value, str):
-            acc_value += delta_value
+            # Only streamed fields accumulate.  Repeated metadata (e.g. a
+            # duplicate-index entry repeating `id` or `function.name` from a
+            # speculative decoder) must be replaced, not concatenated —
+            # otherwise the value becomes `call_abccall_abc`. (#3201)
+            if key in ("content", "refusal", "arguments"):
+                acc_value += delta_value
+            else:
+                acc_value = delta_value
         elif isinstance(acc_value, (int, float)) and isinstance(delta_value, (int, float)):
             acc_value += delta_value
         elif is_dict(acc_value) and is_dict(delta_value):
@@ -33,13 +77,29 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
         elif is_list(acc_value) and is_list(delta_value):
             # for lists of non-dictionary items we'll only ever get new entries
             # in the array, existing entries will never be changed
-            if all(isinstance(x, (str, int, float)) for x in acc_value):
+            if acc_value and all(isinstance(x, (str, int, float)) for x in acc_value):
                 acc_value.extend(delta_value)
                 continue
 
+            # Coalesce the incoming list so duplicate-index entries are merged
+            # before placement — covers the empty-acc fast path (an explicit
+            # `tool_calls: []` from a prior chunk) and any un-coalesced first
+            # chunk. (#3201)
+            if any(is_dict(x) for x in delta_value):
+                delta_value = _coalesce_list_by_index(delta_value)
+
+            # Build an index map once so merging is O(n) instead of O(n²).
+            index_map: dict[int, list[int]] = {}
+            for i, existing in enumerate(acc_value):
+                if is_dict(existing) and isinstance(existing.get("index"), int):
+                    index_map.setdefault(existing["index"], []).append(i)
+
             for delta_entry in delta_value:
                 if not is_dict(delta_entry):
                     raise TypeError(f"Unexpected list delta entry is not a dictionary: {delta_entry}")
+                if _is_placeholder(delta_entry):
+                    # Gap-filler from coalescing — nothing to merge.
+                    continue
 
                 try:
                     index = delta_entry["index"]
@@ -49,16 +109,100 @@ 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.
+                positions = index_map.get(index)
+                if positions:
+                    for pos in positions:
+                        acc_value[pos] = accumulate_delta(acc_value[pos], delta_entry)
+                    continue
 
-                    acc_value[index] = accumulate_delta(acc_entry, delta_entry)
+                # 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.  Bound the padding so a
+                # huge sparse index cannot allocate storage proportional to
+                # its value.
+                if len(acc_value) <= index:
+                    if index - len(acc_value) <= _MAX_INDEX_PADDING:
+                        while len(acc_value) < index:
+                            acc_value.append({})
+                        acc_value.append(delta_entry)
+                    else:
+                        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)
+                index_map.setdefault(index, []).append(len(acc_value) - 1)
 
         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.  A single entry whose index does not
+    match its position is normalized to its logical slot as well.
+    """
+    merged: dict[int, object] = {}
+    tail: list[object] = []
+    for entry in lst:
+        if not is_dict(entry):
+            tail.append(entry)
+            continue
+        index = entry.get("index")
+        if not isinstance(index, int):
+            if _is_placeholder(entry):
+                # Gap-filler from a previous padding pass — replaced by the
+                # real entry at the same logical index.
+                continue
+            tail.append(entry)
+            continue
+        if index in merged:
+            merged[index] = accumulate_delta(merged[index], entry)
+        else:
+            merged[index] = entry
+
+    if not merged:
+        return list(lst)
+
+    max_index = max(merged)
+    if max_index <= _MAX_INDEX_PADDING:
+        result = [merged.get(i, {}) for i in range(max_index + 1)]
+    else:
+        # Huge sparse index: materialize only up to the bound, then append
+        # the remaining entries in index order so allocation stays bounded.
+        result = [merged.get(i, {}) for i in range(_MAX_INDEX_PADDING + 1)]
+        result.extend(merged[i] for i in sorted(merged) if i > _MAX_INDEX_PADDING)
+    return result + tail
diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py
index f9dec645b6..bbcb4b9c71 100644
--- a/src/openai/lib/streaming/chat/_completions.py
+++ b/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,20 @@ 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 = cast("dict[object, object]", choice.delta.to_dict())
+                tool_calls = delta_dict.get("tool_calls")
+                if is_list(tool_calls) and len(tool_calls) > 1:
+                    delta_dict["tool_calls"] = _coalesce_list_by_index(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,
                         },
                     ),
                 )
@@ -532,10 +539,25 @@ def _build_events(
                 assert tool_calls is not None
 
                 for tool_call_delta in choice.delta.tool_calls:
-                    tool_call = tool_calls[tool_call_delta.index]
+                    # After coalescing in accumulate_delta / _coalesce_list_by_index,
+                    # the physical position in the list matches the logical index.
+                    # Use the delta's index with bounds checking to handle
+                    # sparse or out-of-order arrival. (#3201)
+                    idx = tool_call_delta.index
+                    if idx < 0 or idx >= len(tool_calls):
+                        continue
+                    tool_call = tool_calls[idx]
 
                     if tool_call.type == "function":
-                        assert tool_call_delta.function is not None
+                        # A raw delta entry may be metadata-only (e.g. a
+                        # duplicate index-0 entry carrying just `id` and
+                        # `type: "function"` while the arguments arrived in
+                        # an earlier entry). Coalescing merges them into one
+                        # snapshot entry, but the raw delta itself has no
+                        # function payload — skip it rather than aborting
+                        # the stream on the assertion below. (#3201)
+                        if tool_call_delta.function is None:
+                            continue
                         events_to_fire.append(
                             build(
                                 FunctionToolCallArgumentsDeltaEvent,
@@ -618,17 +640,25 @@ def get_done_events(
                 )
 
         for tool_call in choice_chunk.delta.tool_calls or []:
-            if self.__current_tool_call_index != tool_call.index:
+            # Only finalize the previous tool call on a *forward* index
+            # transition.  In an out-of-order stream (e.g. index 1 starts
+            # before index 0), a backward transition must not mark the
+            # higher-index call as done — its arguments may still be
+            # streaming, and finalizing it early would suppress the
+            # corrected done event when the real end arrives. (#3201)
+            if (
+                self.__current_tool_call_index is not None
+                and tool_call.index > self.__current_tool_call_index
+            ):
                 events_to_fire.extend(
                     self._content_done_events(choice_snapshot=choice_snapshot, response_format=response_format)
                 )
 
-                if self.__current_tool_call_index is not None:
-                    self._add_tool_done_event(
-                        events_to_fire=events_to_fire,
-                        choice_snapshot=choice_snapshot,
-                        tool_index=self.__current_tool_call_index,
-                    )
+                self._add_tool_done_event(
+                    events_to_fire=events_to_fire,
+                    choice_snapshot=choice_snapshot,
+                    tool_index=self.__current_tool_call_index,
+                )
 
             self.__current_tool_call_index = tool_call.index
 
@@ -743,9 +773,17 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh
     choices = cast("list[object]", data["choices"])
 
     for choice in chunk.choices:
+        message_dict = cast("dict[object, object]", 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(
diff --git a/tests/lib/chat/test_completions_streaming.py b/tests/lib/chat/test_completions_streaming.py
index 40b5a7a47c..e92f3d9fa8 100644
--- a/tests/lib/chat/test_completions_streaming.py
+++ b/tests/lib/chat/test_completions_streaming.py
@@ -1061,6 +1061,80 @@ def test_stream_obfuscation_stays_on_raw_chunks(padding: tuple[str | None, str |
     assert "obfuscation" not in completion.to_json()
 
 
+def test_metadata_only_tool_call_delta_does_not_abort_stream() -> None:
+    """A raw delta entry with type='function' but no function payload must
+    not abort the stream (regression for #3201).
+
+    When duplicate entries split a tool call's fields — one index-0 entry
+    carries function.arguments, a later index-0 entry carries only id and
+    type — coalescing merges them into one snapshot entry, but the raw
+    metadata-only delta still reaches _build_events with function=None.
+    The arguments-delta event must be skipped, not asserted on.
+    """
+    state = ChatCompletionStreamState()
+
+    # Chunk 1: index-0 entry with the function payload.
+    chunk1 = model_parse(
+        ChatCompletionChunk,
+        {
+            "id": "chatcmpl-test",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "gpt-test",
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {
+                        "role": "assistant",
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "id": "call_abc",
+                                "type": "function",
+                                "function": {"name": "list_files", "arguments": ' {"'},
+                            }
+                        ],
+                    },
+                    "finish_reason": None,
+                    "logprobs": None,
+                }
+            ],
+        },
+    )
+    list(state.handle_chunk(chunk1))
+
+    # Chunk 2: duplicate index-0 entry with only id/type — no function payload.
+    chunk2 = model_parse(
+        ChatCompletionChunk,
+        {
+            "id": "chatcmpl-test",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "gpt-test",
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "id": "call_abc",
+                                "type": "function",
+                            }
+                        ],
+                    },
+                    "finish_reason": None,
+                    "logprobs": None,
+                }
+            ],
+        },
+    )
+    # Must not raise AssertionError.
+    events = list(state.handle_chunk(chunk2))
+    assert events, "expected at least the chunk event"
+    assert all(e.type != "tool_calls.function.arguments.delta" for e in events)
+
+
 @pytest.mark.respx2(base_url=base_url)
 def test_chat_completion_state_helper(client: OpenAI, respx2_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None:
     state = ChatCompletionStreamState()
diff --git a/tests/lib/streaming/test_deltas.py b/tests/lib/streaming/test_deltas.py
new file mode 100644
index 0000000000..735265d21c
--- /dev/null
+++ b/tests/lib/streaming/test_deltas.py
@@ -0,0 +1,568 @@
+"""Tests for the streaming delta accumulator."""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+from openai.lib.streaming._deltas import accumulate_delta
+
+
+class TestAccumulateDelta:
+    """Tests for accumulate_delta — regression for #3201."""
+
+    def test_duplicate_index_first_chunk_merges(self) -> None:
+        """First chunk with two entries at the same index should merge into one."""
+        acc: dict[object, object] = {}
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {
+                    "index": 0,
+                    "id": "call_abc",
+                    "function": {"name": "list_files"},
+                    "type": "function",
+                },
+                {
+                    "index": 0,
+                    "function": {"arguments": ' {"'},
+                },
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        assert isinstance(calls, list)
+        # Should be a single entry at index 0, not two
+        assert len(calls) == 1
+        assert calls[0]["index"] == 0
+        assert calls[0]["id"] == "call_abc"
+        assert calls[0]["function"]["name"] == "list_files"
+        assert calls[0]["function"]["arguments"] == ' {"'
+
+    def test_duplicate_index_subsequent_chunk_merges(self) -> None:
+        """Subsequent chunk with same index should merge into existing entry."""
+        acc: dict[object, object] = {
+            "tool_calls": [
+                {
+                    "index": 0,
+                    "id": "call_abc",
+                    "function": {"name": "list_files", "arguments": ' {"'},
+                    "type": "function",
+                }
+            ]
+        }
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {
+                    "index": 0,
+                    "function": {"arguments": 'path": "."}'},
+                }
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        assert len(calls) == 1
+        assert calls[0]["function"]["arguments"] == ' {"path": "."}'
+
+    def test_different_indexes_accumulate_separately(self) -> None:
+        """Entries with different indexes should accumulate separately."""
+        acc: dict[object, object] = {}
+        delta1: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"},
+            ]
+        }
+        delta2: dict[object, object] = {
+            "tool_calls": [
+                {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"},
+            ]
+        }
+        result = accumulate_delta(acc, delta1)
+        result = accumulate_delta(result, delta2)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        assert len(calls) == 2
+        assert calls[0]["index"] == 0
+        assert calls[1]["index"] == 1
+
+    def test_string_accumulation_unchanged(self) -> None:
+        """Basic string accumulation should still work."""
+        acc: dict[object, object] = {"content": "hello"}
+        delta: dict[object, object] = {"content": " world"}
+        result = accumulate_delta(acc, delta)
+        assert result["content"] == "hello world"
+
+    def test_duplicate_index_first_chunk_then_subsequent_merge(self) -> None:
+        """Full round-trip: first chunk with duplicate indexes, then subsequent chunk merges correctly."""
+        acc: dict[object, object] = {}
+        # First chunk: two entries at index 0
+        delta1: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "id": "call_abc", "function": {"name": "list_files"}, "type": "function"},
+                {"index": 0, "function": {"arguments": ' {"'}},
+            ]
+        }
+        result = accumulate_delta(acc, delta1)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        assert len(calls) == 1, f"Expected 1 entry after coalescing, got {len(calls)}"
+        assert calls[0]["function"]["arguments"] == ' {"'
+
+        # Second chunk: more arguments for index 0
+        delta2: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "function": {"arguments": 'path": "."}'}},
+            ]
+        }
+        result = accumulate_delta(result, delta2)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        assert len(calls) == 1
+        assert calls[0]["function"]["arguments"] == ' {"path": "."}'
+        assert calls[0]["id"] == "call_abc"
+        assert calls[0]["function"]["name"] == "list_files"
+
+    def test_sparse_out_of_order_indexes_no_data_loss(self) -> None:
+        """Regression for the data-loss bug: if acc_value has [{"index": 1, ...}]
+        and index 0 arrives later, the index-1 entry must not be overwritten."""
+        acc: dict[object, object] = {
+            "tool_calls": [
+                {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"},
+            ]
+        }
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"},
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        # Both entries should survive
+        assert len(calls) == 2
+        # The index-1 entry should not be overwritten
+        ids = [c["id"] for c in calls]
+        assert "call_a" in ids
+        assert "call_b" in ids
+
+    def test_out_of_order_index_stays_addressable_by_logical_index(self) -> None:
+        """Regression for Codex P2: when index 1 arrives before index 0, the
+        list must stay addressable by logical index — downstream code does
+        ``tool_calls[tool_call_delta.index]`` treating logical index as
+        physical position.  If the list is ``[{"index": 1}, {"index": 0}]``
+        then ``tool_calls[0]`` returns the wrong entry."""
+        acc: dict[object, object] = {
+            "tool_calls": [
+                {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"},
+            ]
+        }
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"},
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        # The list must be addressable by logical index: calls[0] should have
+        # index 0, calls[1] should have index 1.
+        assert calls[0]["index"] == 0
+        assert calls[0]["id"] == "call_a"
+        assert calls[1]["index"] == 1
+        assert calls[1]["id"] == "call_b"
+
+    def test_gap_placeholder_replaced_not_shifted(self) -> None:
+        """Regression for Codex P2: when indexes 0 then 2 arrive, slot 1 is
+        padded with {}.  If index 1 arrives later, it must replace the
+        placeholder in-place, not insert before it (which would shift the
+        placeholder ahead of index 2, breaking tool_calls[2] lookups)."""
+        acc: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"},
+                {},
+                {"index": 2, "id": "call_c", "function": {"name": "tool_c"}, "type": "function"},
+            ]
+        }
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"},
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        # The placeholder at index 1 should be replaced, not shifted
+        assert len(calls) == 3
+        assert calls[0]["index"] == 0
+        assert calls[0]["id"] == "call_a"
+        assert calls[1]["index"] == 1
+        assert calls[1]["id"] == "call_b"
+        assert calls[2]["index"] == 2
+        assert calls[2]["id"] == "call_c"
+
+    def test_coalesce_list_by_index_sorts_by_logical_index(self) -> None:
+        """Regression for Codex P2: _coalesce_list_by_index must sort entries
+        by logical index so the list is addressable by tool_calls[index]."""
+        from openai.lib.streaming._deltas import _coalesce_list_by_index
+
+        lst: list[object] = [
+            {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"},
+            {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"},
+        ]
+        result = _coalesce_list_by_index(lst)
+        calls = cast(list[dict[str, Any]], result)
+        assert calls[0]["index"] == 0
+        assert calls[0]["id"] == "call_a"
+        assert calls[1]["index"] == 1
+        assert calls[1]["id"] == "call_b"
+
+    def test_dumped_placeholder_replaced_not_shifted(self) -> None:
+        """Regression for Codex P2: after the snapshot is round-tripped through
+        model_dump, a gap-filler {} placeholder becomes a dict of unset
+        tool-call fields (e.g. {"id": None, "function": None, "type": None}).
+        If index 1 arrives later, it must replace that dumped placeholder
+        in-place, not insert before it (which would shift the index-2 entry
+        to slot 3 and break tool_calls[2] lookups)."""
+        acc: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"},
+                # Simulates a {} placeholder after model_dump round-trip
+                {"id": None, "function": None, "type": None},
+                {"index": 2, "id": "call_c", "function": {"name": "tool_c"}, "type": "function"},
+            ]
+        }
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"},
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        # The dumped placeholder at index 1 should be replaced, not shifted
+        assert len(calls) == 3
+        assert calls[0]["index"] == 0
+        assert calls[0]["id"] == "call_a"
+        assert calls[1]["index"] == 1
+        assert calls[1]["id"] == "call_b"
+        assert calls[2]["index"] == 2
+        assert calls[2]["id"] == "call_c"
+
+    def test_coalesce_dumped_placeholder_replaced(self) -> None:
+        """Regression for Codex P2: _coalesce_list_by_index must also detect
+        dumped placeholders (all-None values from model_dump) and replace them
+        in-place instead of inserting before them."""
+        from openai.lib.streaming._deltas import _coalesce_list_by_index
+
+        lst: list[object] = [
+            {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"},
+            # Dumped placeholder at index 1 (all values None)
+            {"id": None, "function": None, "type": None},
+            {"index": 2, "id": "call_c", "function": {"name": "tool_c"}, "type": "function"},
+            # Index 1 arriving later — should replace the placeholder
+            {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"},
+        ]
+        result = _coalesce_list_by_index(lst)
+        calls = cast(list[dict[str, Any]], result)
+        assert len(calls) == 3
+        assert calls[0]["index"] == 0
+        assert calls[0]["id"] == "call_a"
+        assert calls[1]["index"] == 1
+        assert calls[1]["id"] == "call_b"
+        assert calls[2]["index"] == 2
+        assert calls[2]["id"] == "call_c"
+
+    def test_empty_accumulated_list_still_coalesces(self) -> None:
+        """Regression for Codex P2: when a prior chunk explicitly sets
+        tool_calls: [], the next chunk with duplicate-index entries must still
+        be coalesced instead of being extended verbatim."""
+        acc: dict[object, object] = {"tool_calls": []}
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "id": "call_abc", "function": {"name": "list_files"}, "type": "function"},
+                {"index": 0, "function": {"arguments": ' {"'}},
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        assert len(calls) == 1
+        assert calls[0]["id"] == "call_abc"
+        assert calls[0]["function"]["arguments"] == ' {"'
+
+    def test_single_entry_normalized_to_logical_slot(self) -> None:
+        """Regression for Codex P2: a single tool-call entry whose logical
+        index is not 0 must be padded to its logical slot, not stored at
+        physical slot 0."""
+        acc: dict[object, object] = {}
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"},
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        assert len(calls) == 2
+        assert calls[0] == {}
+        assert calls[1]["index"] == 1
+        assert calls[1]["id"] == "call_b"
+
+    def test_repeated_metadata_replaced_not_concatenated(self) -> None:
+        """Regression for Codex P2: duplicate-index entries repeating metadata
+        (id, function.name) must replace, not concatenate — otherwise the value
+        becomes call_abccall_abc."""
+        acc: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "id": "call_abc", "function": {"name": "list_files"}, "type": "function"},
+            ]
+        }
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {"index": 0, "id": "call_abc", "function": {"name": "list_files"}, "type": "function"},
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        assert len(calls) == 1
+        assert calls[0]["id"] == "call_abc"
+        assert calls[0]["function"]["name"] == "list_files"
+
+    def test_huge_sparse_index_does_not_materialize_gaps(self) -> None:
+        """Regression for Codex P2: a delta with a huge sparse index must not
+        allocate storage proportional to the numeric index."""
+        from openai.lib.streaming._deltas import _MAX_INDEX_PADDING
+
+        acc: dict[object, object] = {}
+        delta: dict[object, object] = {
+            "tool_calls": [
+                {"index": 1_000_000, "id": "call_z", "function": {"name": "tool_z"}, "type": "function"},
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
+        # Bounded allocation: no million-entry placeholder list.
+        assert len(calls) <= _MAX_INDEX_PADDING + 2
+        assert calls[-1]["index"] == 1_000_000
+        assert calls[-1]["id"] == "call_z"
+
+
+class TestChatCompletionStreamStateIntegration:
+    """Integration-level regression for #3201: feed the two problematic chunks
+    through ChatCompletionStreamState and verify the final snapshot has exactly
+    one tool call per index with merged fields."""
+
+    def test_duplicate_index_through_stream_state(self) -> None:
+        """Replay the exact issue shape from #3201 through the full stream state.
+
+        The first chunk contains two tool_calls at index 0 (one with id/name,
+        one with arguments).  The second chunk adds a delta to index 0.
+        The final snapshot must contain a single index-0 call with all fields.
+        """
+        from openai.types.chat import ChatCompletionChunk
+        from openai.lib.streaming.chat import ChatCompletionStreamState
+        from openai.types.chat.chat_completion_chunk import Choice as ChoiceChunk
+
+        chunk1 = ChatCompletionChunk.construct(
+            id="chatcmpl-1",
+            created=0,
+            model="gpt-4",
+            choices=[
+                ChoiceChunk.construct(
+                    index=0,
+                    delta={
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "id": "call_abc",
+                                "function": {"name": "list_files"},
+                                "type": "function",
+                            },
+                            {
+                                "index": 0,
+                                "function": {"arguments": ' {"'},
+                            },
+                        ]
+                    },
+                ),
+            ],
+        )
+
+        chunk2 = ChatCompletionChunk.construct(
+            id="chatcmpl-1",
+            created=0,
+            model="gpt-4",
+            choices=[
+                ChoiceChunk.construct(
+                    index=0,
+                    delta={
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "function": {"arguments": "path"},
+                            },
+                        ]
+                    },
+                ),
+            ],
+        )
+
+        state = ChatCompletionStreamState()
+        list(state.handle_chunk(chunk1))
+        list(state.handle_chunk(chunk2))
+
+        snapshot = cast(Any, state.current_completion_snapshot)
+        assert len(snapshot.choices) == 1
+        message = snapshot.choices[0].message
+        tool_calls = message.tool_calls
+        assert tool_calls is not None
+        assert len(tool_calls) == 1, f"Expected 1 tool call, got {len(tool_calls)}"
+        call = tool_calls[0]
+        assert call.id == "call_abc"
+        func = call.function
+        assert func is not None
+        assert func.name == "list_files"
+        assert func.arguments == ' {"path'
+
+    def test_sparse_out_of_order_through_stream_state(self) -> None:
+        """Index 1 arrives before index 0 — no data loss, list stays addressable."""
+        from openai.types.chat import ChatCompletionChunk
+        from openai.lib.streaming.chat import ChatCompletionStreamState
+        from openai.types.chat.chat_completion_chunk import Choice as ChoiceChunk
+
+        chunk1 = ChatCompletionChunk.construct(
+            id="chatcmpl-2",
+            created=0,
+            model="gpt-4",
+            choices=[
+                ChoiceChunk.construct(
+                    index=0,
+                    delta={
+                        "tool_calls": [
+                            {
+                                "index": 1,
+                                "id": "call_b",
+                                "function": {"name": "tool_b"},
+                                "type": "function",
+                            },
+                        ]
+                    },
+                ),
+            ],
+        )
+
+        chunk2 = ChatCompletionChunk.construct(
+            id="chatcmpl-2",
+            created=0,
+            model="gpt-4",
+            choices=[
+                ChoiceChunk.construct(
+                    index=0,
+                    delta={
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "id": "call_a",
+                                "function": {"name": "tool_a"},
+                                "type": "function",
+                            },
+                        ]
+                    },
+                ),
+            ],
+        )
+
+        state = ChatCompletionStreamState()
+        list(state.handle_chunk(chunk1))
+        list(state.handle_chunk(chunk2))
+
+        snapshot = cast(Any, state.current_completion_snapshot)
+        message = snapshot.choices[0].message
+        tool_calls = message.tool_calls
+        assert tool_calls is not None
+        assert len(tool_calls) == 2, f"Expected 2 tool calls, got {len(tool_calls)}"
+        assert tool_calls[0].id == "call_a"
+        assert tool_calls[1].id == "call_b"
+
+    def test_backward_index_transition_does_not_finalize_tool(self) -> None:
+        """Regression for Codex P2: when a higher-index tool call starts before
+        a lower one (1 -> 0), the backward transition must not mark tool call 1
+        as done — its arguments may still be streaming, and finalizing it
+        early would suppress the corrected done event."""
+        from openai.types.chat import ChatCompletionChunk
+        from openai.lib.streaming.chat import ChatCompletionStreamState
+        from openai.types.chat.chat_completion_chunk import Choice as ChoiceChunk
+
+        chunk1 = ChatCompletionChunk.construct(
+            id="chatcmpl-3",
+            created=0,
+            model="gpt-4",
+            choices=[
+                ChoiceChunk.construct(
+                    index=0,
+                    delta={
+                        "tool_calls": [
+                            {
+                                "index": 1,
+                                "id": "call_b",
+                                "function": {"name": "tool_b", "arguments": ""},
+                                "type": "function",
+                            },
+                        ]
+                    },
+                ),
+            ],
+        )
+
+        chunk2 = ChatCompletionChunk.construct(
+            id="chatcmpl-3",
+            created=0,
+            model="gpt-4",
+            choices=[
+                ChoiceChunk.construct(
+                    index=0,
+                    delta={
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "id": "call_a",
+                                "function": {"name": "tool_a", "arguments": ""},
+                                "type": "function",
+                            },
+                        ]
+                    },
+                ),
+            ],
+        )
+
+        chunk3 = ChatCompletionChunk.construct(
+            id="chatcmpl-3",
+            created=0,
+            model="gpt-4",
+            choices=[
+                ChoiceChunk.construct(
+                    index=0,
+                    delta={
+                        "tool_calls": [
+                            {
+                                "index": 1,
+                                "function": {"arguments": '{"x": 1}'},
+                            },
+                        ]
+                    },
+                ),
+            ],
+        )
+
+        state = ChatCompletionStreamState()
+        events1 = list(state.handle_chunk(chunk1))
+        events2 = list(state.handle_chunk(chunk2))
+        events3 = list(state.handle_chunk(chunk3))
+
+        # The backward transition (1 -> 0) must not finalize tool call 1 —
+        # its arguments are still streaming.  A done event for index 1 with
+        # empty arguments would be premature.
+        done_events = [
+            e
+            for e in events1 + events2 + events3
+            if getattr(e, "type", "") == "tool_calls.function.arguments.done"
+        ]
+        assert all(e.index != 1 for e in done_events), f"Premature done event for tool call 1: {done_events}"
+
+        # The final snapshot must still hold both calls with merged arguments.
+        snapshot = cast(Any, state.current_completion_snapshot)
+        tool_calls = snapshot.choices[0].message.tool_calls
+        assert tool_calls is not None
+        assert len(tool_calls) == 2
+        assert tool_calls[0].id == "call_a"
+        assert tool_calls[1].id == "call_b"
+        assert tool_calls[1].function.arguments == '{"x": 1}'