From feeea11a356c2d4ead825bf549732abb3b457ab8 Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Mon, 20 Jul 2026 23:39:41 +0530 Subject: [PATCH 1/2] 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 f2b61eb836851b242f6e0e81222f6b1c785b4441 Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Mon, 20 Jul 2026 23:54:57 +0530 Subject: [PATCH 2/2] 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