From be88b31cc1387e95ff0717d941ef7e545e52d4bd Mon Sep 17 00:00:00 2001
From: Shakti Prasad Mohapatra <rkfshakti@gmail.com>
Date: Mon, 20 Jul 2026 23:39:41 +0530
Subject: [PATCH 1/9] fix: merge list entries by logical index instead of
 physical position (#3201)

accumulate_delta assumed that an indexed list entry's 'index' field
matches its physical position in the Python list. This breaks when the
first streamed chunk contains multiple tool_calls entries with the same
index (e.g. from speculative decoding with vLLM).

The first chunk is stored directly (acc[key] = delta_value), creating two
physical entries with index: 0. Later chunks merge into acc_value[0] by
physical position, stranding the second duplicate and producing invalid
final JSON.

Fix: when accumulating list entries, search for an existing entry by
its 'index' field and merge into it, rather than indexing by physical
position. New entries are inserted at their logical index.
---
 src/openai/lib/streaming/_deltas.py | 24 +++++---
 tests/lib/streaming/test_deltas.py  | 88 +++++++++++++++++++++++++++++
 2 files changed, 104 insertions(+), 8 deletions(-)
 create mode 100644 tests/lib/streaming/test_deltas.py

diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py
index a5e1317612..9c305f32b9 100644
--- a/src/openai/lib/streaming/_deltas.py
+++ b/src/openai/lib/streaming/_deltas.py
@@ -49,15 +49,23 @@ 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.
+                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
+                        break
 
-                    acc_value[index] = accumulate_delta(acc_entry, delta_entry)
+                if not found:
+                    # Ensure the list is large enough
+                    while len(acc_value) <= index:
+                        acc_value.append({})
+                    acc_value[index] = delta_entry
 
         acc[key] = acc_value
 
diff --git a/tests/lib/streaming/test_deltas.py b/tests/lib/streaming/test_deltas.py
new file mode 100644
index 0000000000..00f346c035
--- /dev/null
+++ b/tests/lib/streaming/test_deltas.py
@@ -0,0 +1,88 @@
+"""Tests for the streaming delta accumulator."""
+
+from __future__ import annotations
+
+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 = {
+            "tool_calls": [
+                {
+                    "index": 0,
+                    "id": "call_abc",
+                    "function": {"name": "list_files"},
+                    "type": "function",
+                },
+                {
+                    "index": 0,
+                    "function": {"arguments": ' {"'},
+                },
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = 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 = {
+            "tool_calls": [
+                {
+                    "index": 0,
+                    "function": {"arguments": 'path": "."}'},
+                }
+            ]
+        }
+        result = accumulate_delta(acc, delta)
+        calls = 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 = {
+            "tool_calls": [
+                {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"},
+            ]
+        }
+        delta2 = {
+            "tool_calls": [
+                {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"},
+            ]
+        }
+        result = accumulate_delta(acc, delta1)
+        result = accumulate_delta(result, delta2)
+        calls = 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 = {"content": " world"}
+        result = accumulate_delta(acc, delta)
+        assert result["content"] == "hello world"
\ No newline at end of file

From e2f4550426d5e1ef68daf69760d9090c296d0af8 Mon Sep 17 00:00:00 2001
From: Shakti Prasad Mohapatra <rkfshakti@gmail.com>
Date: Mon, 20 Jul 2026 23:54:57 +0530
Subject: [PATCH 2/9] fix: coalesce duplicate-index entries when first chunk is
 stored directly (#3201)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Codex P1 review: when tool_calls is first added to the snapshot it is
copied directly (acc[key] = delta_value), so a first chunk with two
entries at the same index creates two physical entries that later merges
can't fix — the merge only hits the first matching entry and breaks,
leaving the earlier duplicate stranded.

Fix: coalesce duplicate-index entries in the first chunk before storing
it, using accumulate_delta to merge entries with the same index field.

Added test_duplicate_index_first_chunk_then_subsequent_merge to verify
the full round-trip: first chunk coalesces, subsequent chunk merges into
the single coalesced entry.
---
 src/openai/lib/streaming/_deltas.py | 36 +++++++++++++++++++++++++++++
 tests/lib/streaming/test_deltas.py  | 30 +++++++++++++++++++++++-
 2 files changed, 65 insertions(+), 1 deletion(-)

diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py
index 9c305f32b9..dc23b85faf 100644
--- a/src/openai/lib/streaming/_deltas.py
+++ b/src/openai/lib/streaming/_deltas.py
@@ -6,6 +6,12 @@
 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
 
@@ -70,3 +76,33 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
         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)
+    """
+    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:
+            result.append(entry)
+    return result
diff --git a/tests/lib/streaming/test_deltas.py b/tests/lib/streaming/test_deltas.py
index 00f346c035..1d613a374f 100644
--- a/tests/lib/streaming/test_deltas.py
+++ b/tests/lib/streaming/test_deltas.py
@@ -85,4 +85,32 @@ def test_string_accumulation_unchanged(self) -> None:
         acc: dict[object, object] = {"content": "hello"}
         delta = {"content": " world"}
         result = accumulate_delta(acc, delta)
-        assert result["content"] == "hello world"
\ No newline at end of file
+        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 = {
+            "tool_calls": [
+                {"index": 0, "id": "call_abc", "function": {"name": "list_files"}, "type": "function"},
+                {"index": 0, "function": {"arguments": ' {"'}},
+            ]
+        }
+        result = accumulate_delta(acc, delta1)
+        calls = 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 = {
+            "tool_calls": [
+                {"index": 0, "function": {"arguments": 'path": "."}'}},
+            ]
+        }
+        result = accumulate_delta(result, delta2)
+        calls = 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"
\ No newline at end of file

From 04d3bd42bbc3407ffa679d9b7fea7f2543ea778d Mon Sep 17 00:00:00 2001
From: Shakti Prasad Mohapatra <rkfshakti@gmail.com>
Date: Thu, 23 Jul 2026 23:23:31 +0530
Subject: [PATCH 3/9] fix: merge delta into all matching index entries, not
 just the first

When acc_value already contains duplicate-index entries (e.g. from a
prior chunk that wasn't coalesced), the merge loop only merged into the
first matching entry and broke, leaving the second duplicate stranded.
Remove the break so all matching entries get the delta merged in.

Addresses Codex P1 review feedback.
---
 src/openai/lib/streaming/_deltas.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py
index dc23b85faf..e37d1101fa 100644
--- a/src/openai/lib/streaming/_deltas.py
+++ b/src/openai/lib/streaming/_deltas.py
@@ -60,12 +60,15 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
                 # 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
-                        break
 
                 if not found:
                     # Ensure the list is large enough

From 73cac3598ffd9f42aa80796d11185acf699e3d91 Mon Sep 17 00:00:00 2001
From: Shakti Prasad Mohapatra <rkfshakti@gmail.com>
Date: Fri, 7 Aug 2026 12:30:08 +0530
Subject: [PATCH 4/9] fix: normalize first-chunk duplicates, fix sparse-index
 data loss, fix types
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Addresses review feedback from @jbeckwith-oai on #3521:

1. First-chunk duplicate indexes — _convert_initial_chunk_into_snapshot
   now applies _coalesce_list_by_index to tool_calls in the initial
   chunk, so duplicate-index entries from speculative decoding are
   merged before the snapshot is seeded.

2. Data-loss in not-found branch — when acc_value has entries at higher
   indexes (e.g. [{"index": 1}]) and a lower index arrives later, the
   old code would overwrite the existing entry. Now appends instead of
   assigning by position. Added test_sparse_out_of_order_indexes_no_data_loss.

3. Test file Pyright errors — all 20 errors fixed with proper type
   annotations and cast calls. Pyright and Ruff both pass clean.
---
 src/openai/lib/streaming/_deltas.py           |  43 ++++++-
 src/openai/lib/streaming/chat/_completions.py |  22 +++-
 tests/lib/streaming/test_deltas.py            | 119 ++++++++++++++++--
 3 files changed, 162 insertions(+), 22 deletions(-)

diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py
index e37d1101fa..3201cc992f 100644
--- a/src/openai/lib/streaming/_deltas.py
+++ b/src/openai/lib/streaming/_deltas.py
@@ -71,10 +71,29 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
                         found = True
 
                 if not found:
-                    # Ensure the list is large enough
-                    while len(acc_value) <= index:
-                        acc_value.append({})
-                    acc_value[index] = 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.
+                    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 an empty
+                        # placeholder ({}), replace it in-place.  Otherwise
+                        # insert at the correct position to keep the list
+                        # addressable by logical index.
+                        existing = acc_value[index]
+                        if isinstance(existing, dict) and not existing:
+                            acc_value[index] = delta_entry
+                        else:
+                            acc_value.insert(index, delta_entry)
 
         acc[key] = acc_value
 
@@ -89,6 +108,10 @@ def _coalesce_list_by_index(lst: list[object]) -> list[object]:
     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:
@@ -107,5 +130,15 @@ def _coalesce_list_by_index(lst: list[object]) -> list[object]:
                 found = True
                 break
         if not found:
-            result.append(entry)
+            # 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` or shift if occupied
+            existing = result[index]
+            if isinstance(existing, dict) and not existing:
+                result[index] = entry
+            else:
+                result.insert(index, entry)
     return result
diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py
index f9dec645b6..60cbc4a684 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,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,
                         },
                     ),
                 )
@@ -743,9 +749,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(
diff --git a/tests/lib/streaming/test_deltas.py b/tests/lib/streaming/test_deltas.py
index 1d613a374f..aa1594d0ce 100644
--- a/tests/lib/streaming/test_deltas.py
+++ b/tests/lib/streaming/test_deltas.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import Any, cast
+
 from openai.lib.streaming._deltas import accumulate_delta
 
 
@@ -11,7 +13,7 @@ class TestAccumulateDelta:
     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 = {
+        delta: dict[object, object] = {
             "tool_calls": [
                 {
                     "index": 0,
@@ -26,7 +28,7 @@ def test_duplicate_index_first_chunk_merges(self) -> None:
             ]
         }
         result = accumulate_delta(acc, delta)
-        calls = result["tool_calls"]
+        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
@@ -47,7 +49,7 @@ def test_duplicate_index_subsequent_chunk_merges(self) -> None:
                 }
             ]
         }
-        delta = {
+        delta: dict[object, object] = {
             "tool_calls": [
                 {
                     "index": 0,
@@ -56,26 +58,26 @@ def test_duplicate_index_subsequent_chunk_merges(self) -> None:
             ]
         }
         result = accumulate_delta(acc, delta)
-        calls = result["tool_calls"]
+        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 = {
+        delta1: dict[object, object] = {
             "tool_calls": [
                 {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"},
             ]
         }
-        delta2 = {
+        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 = result["tool_calls"]
+        calls = cast(list[dict[str, Any]], result["tool_calls"])
         assert len(calls) == 2
         assert calls[0]["index"] == 0
         assert calls[1]["index"] == 1
@@ -83,7 +85,7 @@ def test_different_indexes_accumulate_separately(self) -> None:
     def test_string_accumulation_unchanged(self) -> None:
         """Basic string accumulation should still work."""
         acc: dict[object, object] = {"content": "hello"}
-        delta = {"content": " world"}
+        delta: dict[object, object] = {"content": " world"}
         result = accumulate_delta(acc, delta)
         assert result["content"] == "hello world"
 
@@ -91,26 +93,117 @@ 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 = {
+        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 = result["tool_calls"]
+        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 = {
+        delta2: dict[object, object] = {
             "tool_calls": [
                 {"index": 0, "function": {"arguments": 'path": "."}'}},
             ]
         }
         result = accumulate_delta(result, delta2)
-        calls = result["tool_calls"]
+        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"
\ No newline at end of file
+        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"

From d388b25aad8153c8e4b47fd4535ad49185a239d4 Mon Sep 17 00:00:00 2001
From: Shakti Prasad Mohapatra <rkfshakti@gmail.com>
Date: Fri, 7 Aug 2026 17:54:17 +0530
Subject: [PATCH 5/9] fix: detect dumped placeholders after model_dump
 round-trip

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}). The previous
check (isinstance(existing, dict) and not existing) only matched
empty {} placeholders, so a later-arriving entry at the same index
was inserted before the dumped placeholder instead of replacing it,
shifting higher-index entries and breaking tool_calls[index] lookups.

Added _is_placeholder() helper that detects both empty {} and
all-None dumped placeholders. Applied in both accumulate_delta and
_coalesce_list_by_index. Added regression tests
test_dumped_placeholder_replaced_not_shifted and
test_coalesce_dumped_placeholder_replaced.
---
 src/openai/lib/streaming/_deltas.py | 39 ++++++++++++++++----
 tests/lib/streaming/test_deltas.py  | 55 +++++++++++++++++++++++++++++
 2 files changed, 87 insertions(+), 7 deletions(-)

diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py
index 3201cc992f..79b26c7bb6 100644
--- a/src/openai/lib/streaming/_deltas.py
+++ b/src/openai/lib/streaming/_deltas.py
@@ -3,6 +3,27 @@
 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:
@@ -85,12 +106,14 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
                         acc_value.append(delta_entry)
                     else:
                         # The list is large enough but no entry has this
-                        # index.  If the slot at `index` is an empty
-                        # placeholder ({}), replace it in-place.  Otherwise
-                        # insert at the correct position to keep the list
-                        # addressable by logical index.
+                        # 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 isinstance(existing, dict) and not existing:
+                        if _is_placeholder(existing):
                             acc_value[index] = delta_entry
                         else:
                             acc_value.insert(index, delta_entry)
@@ -135,9 +158,11 @@ def _coalesce_list_by_index(lst: list[object]) -> list[object]:
             # logical index.
             while len(result) <= index:
                 result.append({})
-            # Replace the placeholder at `index` or shift if occupied
+            # 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 isinstance(existing, dict) and not existing:
+            if _is_placeholder(existing):
                 result[index] = entry
             else:
                 result.insert(index, entry)
diff --git a/tests/lib/streaming/test_deltas.py b/tests/lib/streaming/test_deltas.py
index aa1594d0ce..039a05fb35 100644
--- a/tests/lib/streaming/test_deltas.py
+++ b/tests/lib/streaming/test_deltas.py
@@ -207,3 +207,58 @@ def test_coalesce_list_by_index_sorts_by_logical_index(self) -> None:
         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"

From 37eb8f1476f2fc6cc179d41d633b05d539c59ac1 Mon Sep 17 00:00:00 2001
From: Shakti Prasad Mohapatra <rkfshakti@gmail.com>
Date: Tue, 11 Aug 2026 08:18:15 +0530
Subject: [PATCH 6/9] fix: coalesce at all list boundaries, fix _build_events
 index lookup, add integration tests

Address all three blockers from jbeckwith-oai's review:

1. Normalize at the canonical list boundary including initial/None paths.
   The `acc_value is None` fast path in accumulate_delta now coalesces
   duplicate-index entries before storing, same as the `key not in acc`
   path. _convert_initial_chunk_into_snapshot and the IndexError path in
   _accumulate_chunk both coalesce tool_calls via _coalesce_list_by_index.

2. Fix data-loss when a lower index arrives after a higher one. The
   _build_events method used tool_calls[tool_call_delta.index] as a
   physical position lookup, which raised IndexError for out-of-order
   arrivals. Changed to find by logical index field instead.

3. Fix pyright errors in the test file. Cast snapshot to Any (matching
   the pattern in test_completions_streaming.py) so strict pyright
   reports 0 errors on the test file.

Added two integration-level ChatCompletionStreamState regression tests
using the exact chunk shapes from #3201: duplicate-index first chunk and
sparse out-of-order indexes.
---
 src/openai/lib/streaming/_deltas.py           |   5 +
 src/openai/lib/streaming/chat/_completions.py |  20 ++-
 tests/lib/streaming/test_deltas.py            | 138 ++++++++++++++++++
 3 files changed, 158 insertions(+), 5 deletions(-)

diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py
index 79b26c7bb6..4a483e08c3 100644
--- a/src/openai/lib/streaming/_deltas.py
+++ b/src/openai/lib/streaming/_deltas.py
@@ -38,6 +38,11 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
 
         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 len(delta_value) > 1:
+                delta_value = _coalesce_list_by_index(delta_value)
             acc[key] = delta_value
             continue
 
diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py
index 60cbc4a684..a8ee728cb0 100644
--- a/src/openai/lib/streaming/chat/_completions.py
+++ b/src/openai/lib/streaming/chat/_completions.py
@@ -412,9 +412,10 @@ def _accumulate_chunk(self, chunk: ChatCompletionChunk) -> ParsedChatCompletionS
                 # 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"]))
+                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(
@@ -538,7 +539,16 @@ 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]
+                    # Find the tool call by logical index, not physical
+                    # position.  When entries arrive out of order (e.g.
+                    # index 1 before index 0), the physical position does
+                    # not match the logical index. (#3201)
+                    tool_call = next(
+                        (tc for tc in tool_calls if tc.index == tool_call_delta.index),
+                        None,
+                    )
+                    if tool_call is None:
+                        continue
 
                     if tool_call.type == "function":
                         assert tool_call_delta.function is not None
@@ -749,7 +759,7 @@ 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()
+        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
diff --git a/tests/lib/streaming/test_deltas.py b/tests/lib/streaming/test_deltas.py
index 039a05fb35..1e6ba5b423 100644
--- a/tests/lib/streaming/test_deltas.py
+++ b/tests/lib/streaming/test_deltas.py
@@ -262,3 +262,141 @@ def test_coalesce_dumped_placeholder_replaced(self) -> None:
         assert calls[1]["id"] == "call_b"
         assert calls[2]["index"] == 2
         assert calls[2]["id"] == "call_c"
+
+
+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"

From ed3420cfedc3251b56ffe17e60a255cb03083971 Mon Sep 17 00:00:00 2001
From: Shakti Prasad Mohapatra <rkfshakti@gmail.com>
Date: Thu, 13 Aug 2026 21:03:50 +0530
Subject: [PATCH 7/9] fix: use positional index lookup with bounds check
 instead of non-existent .index attr

ParsedFunctionToolCall does not have an .index field, so the previous
tc.index lookup caused Pyright errors. After coalescing in
_coalesce_list_by_index, the physical position in the list matches the
logical index, so use tool_call_delta.index with bounds checking instead.
---
 src/openai/lib/streaming/chat/_completions.py | 16 +++++++---------
 1 file changed, 7 insertions(+), 9 deletions(-)

diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py
index a8ee728cb0..0196a08e17 100644
--- a/src/openai/lib/streaming/chat/_completions.py
+++ b/src/openai/lib/streaming/chat/_completions.py
@@ -539,16 +539,14 @@ def _build_events(
                 assert tool_calls is not None
 
                 for tool_call_delta in choice.delta.tool_calls:
-                    # Find the tool call by logical index, not physical
-                    # position.  When entries arrive out of order (e.g.
-                    # index 1 before index 0), the physical position does
-                    # not match the logical index. (#3201)
-                    tool_call = next(
-                        (tc for tc in tool_calls if tc.index == tool_call_delta.index),
-                        None,
-                    )
-                    if tool_call is None:
+                    # 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

From 847d3172546dc71a178a90a86beaa9276ab75735 Mon Sep 17 00:00:00 2001
From: Shakti Prasad Mohapatra <rkfshakti@gmail.com>
Date: Tue, 25 Aug 2026 14:36:03 +0530
Subject: [PATCH 8/9] fix: skip metadata-only tool-call deltas in arguments
 event emission
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Codex P2 review (Aug 22): 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, aborting the stream on the assertion.

Only emit the arguments-delta event when the raw delta actually has a
function payload. Regression test: metadata-only delta no longer aborts
the stream and emits no arguments-delta event.
---
 src/openai/lib/streaming/chat/_completions.py | 10 ++-
 tests/lib/chat/test_completions_streaming.py  | 74 +++++++++++++++++++
 2 files changed, 83 insertions(+), 1 deletion(-)

diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py
index 0196a08e17..0988c47db8 100644
--- a/src/openai/lib/streaming/chat/_completions.py
+++ b/src/openai/lib/streaming/chat/_completions.py
@@ -549,7 +549,15 @@ def _build_events(
                     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,
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()

From e30c05424da0df427cd9daef9acf007cad7ec155 Mon Sep 17 00:00:00 2001
From: Shakti Prasad Mohapatra <rkfshakti@gmail.com>
Date: Wed, 9 Sep 2026 17:07:59 +0530
Subject: [PATCH 9/9] fix: bound sparse-index padding, coalesce empty lists,
 avoid premature done events

---
 src/openai/lib/streaming/_deltas.py           | 158 ++++++++++-------
 src/openai/lib/streaming/chat/_completions.py |  22 ++-
 tests/lib/streaming/test_deltas.py            | 166 ++++++++++++++++++
 3 files changed, 277 insertions(+), 69 deletions(-)

diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py
index 4a483e08c3..38956b2f61 100644
--- a/src/openai/lib/streaming/_deltas.py
+++ b/src/openai/lib/streaming/_deltas.py
@@ -2,6 +2,12 @@
 
 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.
@@ -27,11 +33,10 @@ def _is_placeholder(entry: object) -> bool:
 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:
+            # 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
@@ -41,7 +46,7 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
             # 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 len(delta_value) > 1:
+            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
@@ -57,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):
@@ -65,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"]
@@ -90,38 +118,43 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) ->
                 # 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
-
-                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:
+                positions = index_map.get(index)
+                if positions:
+                    for pos in positions:
+                        acc_value[pos] = accumulate_delta(acc_value[pos], delta_entry)
+                    continue
+
+                # 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:
-                        # 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_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
 
@@ -139,36 +172,37 @@ def _coalesce_list_by_index(lst: list[object]) -> list[object]:
 
     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.
+    logical index as physical position.  A single entry whose index does not
+    match its position is normalized to its logical slot as well.
     """
-    result: list[object] = []
+    merged: dict[int, object] = {}
+    tail: list[object] = []
     for entry in lst:
         if not is_dict(entry):
-            result.append(entry)
+            tail.append(entry)
             continue
         index = entry.get("index")
         if not isinstance(index, int):
-            result.append(entry)
+            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
-        # 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
+        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 0988c47db8..bbcb4b9c71 100644
--- a/src/openai/lib/streaming/chat/_completions.py
+++ b/src/openai/lib/streaming/chat/_completions.py
@@ -640,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
 
diff --git a/tests/lib/streaming/test_deltas.py b/tests/lib/streaming/test_deltas.py
index 1e6ba5b423..735265d21c 100644
--- a/tests/lib/streaming/test_deltas.py
+++ b/tests/lib/streaming/test_deltas.py
@@ -263,6 +263,78 @@ def test_coalesce_dumped_placeholder_replaced(self) -> None:
         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
@@ -400,3 +472,97 @@ def test_sparse_out_of_order_through_stream_state(self) -> 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}'