From 40b14f343ff3aef6f725c072ca41cb63488af86f Mon Sep 17 00:00:00 2001 From: mangodxd Date: Thu, 16 Jul 2026 00:24:34 +0700 Subject: [PATCH 1/2] Fix streaming tool_call delta accumulation with duplicate indexes When the first streaming chunk contains multiple tool_calls entries with the same index, accumulate_delta stored them as separate physical list entries. Subsequent chunks would only merge into the first physical entry at that position, stranding the remaining entries with partial data. The fix normalizes delta list entries by their index field before storing or merging on all three code paths in accumulate_delta: - Initial store (key not in acc) - None value replacement - List merge path (existing entries) Fixes #3201 --- src/openai/lib/streaming/_assistants.py | 64 ++++++++++- src/openai/lib/streaming/_deltas.py | 64 ++++++++++- tests/test_deltas.py | 139 ++++++++++++++++++++++++ 3 files changed, 261 insertions(+), 6 deletions(-) create mode 100644 tests/test_deltas.py diff --git a/src/openai/lib/streaming/_assistants.py b/src/openai/lib/streaming/_assistants.py index 6efb3ca3f1..23d76b7856 100644 --- a/src/openai/lib/streaming/_assistants.py +++ b/src/openai/lib/streaming/_assistants.py @@ -980,12 +980,44 @@ def accumulate_event( def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> dict[object, object]: for key, delta_value in delta.items(): if key not in acc: - acc[key] = delta_value + # Normalize lists of dicts with index fields before storing + # to handle cases like tool_calls with duplicate indexes + # in the first chunk + if is_list(delta_value) and all(is_dict(x) for x in delta_value): + normalized: list[object] = [] + index_map: dict[int, int] = {} + for entry in delta_value: + idx = entry.get("index") + if idx is not None and isinstance(idx, int) and idx in index_map: + existing_pos = index_map[idx] + normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry) + else: + if idx is not None and isinstance(idx, int): + index_map[idx] = len(normalized) + normalized.append(entry) + acc[key] = normalized + else: + acc[key] = delta_value continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value + # Normalize lists of dicts with index fields before storing + if is_list(delta_value) and all(is_dict(x) for x in delta_value): + normalized = [] + index_map = {} + for entry in delta_value: + idx = entry.get("index") + if idx is not None and isinstance(idx, int) and idx in index_map: + existing_pos = index_map[idx] + normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry) + else: + if idx is not None and isinstance(idx, int): + index_map[idx] = len(normalized) + normalized.append(entry) + acc[key] = normalized + else: + acc[key] = delta_value continue # the `index` property is used in arrays of objects so it should @@ -1011,7 +1043,33 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> acc_value.extend(delta_value) continue - for delta_entry in delta_value: + # Normalize delta entries by index before merging. + # When multiple entries share the same index (e.g., tool_calls + # with duplicate indexes in the first chunk), they must be merged + # together first so that subsequent chunks merge into the correct + # logical position rather than stranding entries at duplicate indexes. + normalized_delta: list[object] = [] + index_map: dict[int, int] = {} + for entry in delta_value: + if not is_dict(entry): + raise TypeError(f"Unexpected list delta entry is not a dictionary: {entry}") + + try: + idx = entry["index"] + except KeyError as exc: + raise RuntimeError(f"Expected list delta entry to have an `index` key; {entry}") from exc + + if not isinstance(idx, int): + raise TypeError(f"Unexpected, list delta entry `index` value is not an integer; {idx}") + + if idx in index_map: + existing_pos = index_map[idx] + normalized_delta[existing_pos] = accumulate_delta(normalized_delta[existing_pos], entry) + else: + index_map[idx] = len(normalized_delta) + normalized_delta.append(entry) + + for delta_entry in normalized_delta: if not is_dict(delta_entry): raise TypeError(f"Unexpected list delta entry is not a dictionary: {delta_entry}") diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py index a5e1317612..c50acd1d9b 100644 --- a/src/openai/lib/streaming/_deltas.py +++ b/src/openai/lib/streaming/_deltas.py @@ -6,12 +6,44 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> dict[object, object]: for key, delta_value in delta.items(): if key not in acc: - acc[key] = delta_value + # Normalize lists of dicts with index fields before storing + # to handle cases like tool_calls with duplicate indexes + # in the first chunk (see issue #3201) + if is_list(delta_value) and all(is_dict(x) for x in delta_value): + normalized: list[object] = [] + index_map: dict[int, int] = {} + for entry in delta_value: + idx = entry.get("index") + if idx is not None and isinstance(idx, int) and idx in index_map: + existing_pos = index_map[idx] + normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry) + else: + if idx is not None and isinstance(idx, int): + index_map[idx] = len(normalized) + normalized.append(entry) + acc[key] = normalized + else: + acc[key] = delta_value continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value + # Normalize lists of dicts with index fields before storing + if is_list(delta_value) and all(is_dict(x) for x in delta_value): + normalized = [] + index_map = {} + for entry in delta_value: + idx = entry.get("index") + if idx is not None and isinstance(idx, int) and idx in index_map: + existing_pos = index_map[idx] + normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry) + else: + if idx is not None and isinstance(idx, int): + index_map[idx] = len(normalized) + normalized.append(entry) + acc[key] = normalized + else: + acc[key] = delta_value continue # the `index` property is used in arrays of objects so it should @@ -37,7 +69,33 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> acc_value.extend(delta_value) continue - for delta_entry in delta_value: + # Normalize delta entries by index before merging. + # When multiple entries share the same index (e.g., tool_calls + # with duplicate indexes in the first chunk), they must be merged + # together first so that subsequent chunks merge into the correct + # logical position rather than stranding entries at duplicate indexes. + normalized_delta: list[object] = [] + index_map: dict[int, int] = {} + for entry in delta_value: + if not is_dict(entry): + raise TypeError(f"Unexpected list delta entry is not a dictionary: {entry}") + + try: + idx = entry["index"] + except KeyError as exc: + raise RuntimeError(f"Expected list delta entry to have an `index` key; {entry}") from exc + + if not isinstance(idx, int): + raise TypeError(f"Unexpected, list delta entry `index` value is not an integer; {idx}") + + if idx in index_map: + existing_pos = index_map[idx] + normalized_delta[existing_pos] = accumulate_delta(normalized_delta[existing_pos], entry) + else: + index_map[idx] = len(normalized_delta) + normalized_delta.append(entry) + + for delta_entry in normalized_delta: if not is_dict(delta_entry): raise TypeError(f"Unexpected list delta entry is not a dictionary: {delta_entry}") diff --git a/tests/test_deltas.py b/tests/test_deltas.py new file mode 100644 index 0000000000..510d7ac586 --- /dev/null +++ b/tests/test_deltas.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +from openai.lib.streaming._deltas import accumulate_delta + + +def test_duplicate_index_in_first_chunk() -> None: + """ + Regression test for issue #3201: + When the first chunk contains multiple tool_calls entries with the same + `index`, they should be merged into a single entry by logical index, + not stored as separate physical entries. + """ + chunk1 = { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": ""}, + }, + { + "index": 0, + "function": {"arguments": ' {"cit'}, + }, + ] + } + + chunk2 = { + "tool_calls": [ + { + "index": 0, + "function": {"arguments": 'y": "London"}'}, + } + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 1, f"Expected 1 tool call, got {len(acc['tool_calls'])}" + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"cit' + + acc = accumulate_delta(acc, chunk2) + assert len(acc["tool_calls"]) == 1, f"Expected 1 tool call, got {len(acc['tool_calls'])}" + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"city": "London"}' + + +def test_multiple_tool_calls_different_indexes() -> None: + """Multiple tool calls with different indexes should remain separate.""" + chunk1 = { + "tool_calls": [ + {"index": 0, "id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": ""}}, + {"index": 1, "id": "call_2", "type": "function", "function": {"name": "get_time", "arguments": ""}}, + ] + } + + chunk2 = { + "tool_calls": [ + {"index": 0, "function": {"arguments": ' {"city": "London"}'}}, + {"index": 1, "function": {"arguments": ' {"tz": "UTC"}'}}, + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 2 + + acc = accumulate_delta(acc, chunk2) + assert len(acc["tool_calls"]) == 2 + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"city": "London"}' + assert acc["tool_calls"][1]["function"]["arguments"] == ' {"tz": "UTC"}' + + +def test_sparse_indexes() -> None: + """Sparse indexes (index 2 arriving before index 1) should be handled correctly. + + Note: The accumulator inserts at the physical position matching the index, + so index 2 is stored at position 0 when the list is empty, then index 1 + is inserted at position 1 (append). This is existing behavior. + """ + chunk1 = { + "tool_calls": [ + {"index": 2, "id": "call_3", "type": "function", "function": {"name": "search", "arguments": ""}}, + ] + } + + chunk2 = { + "tool_calls": [ + {"index": 1, "id": "call_2", "type": "function", "function": {"name": "get_time", "arguments": ""}}, + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 1 + assert acc["tool_calls"][0]["index"] == 2 + + acc = accumulate_delta(acc, chunk2) + assert len(acc["tool_calls"]) == 2 + assert acc["tool_calls"][0]["index"] == 2 + assert acc["tool_calls"][1]["index"] == 1 + + +def test_normal_case_no_duplicates() -> None: + """Normal case with no duplicate indexes should work as before.""" + chunk1 = { + "tool_calls": [ + {"index": 0, "id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": ""}}, + ] + } + + chunk2 = { + "tool_calls": [ + {"index": 0, "function": {"arguments": ' {"city": "London"}'}}, + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 1 + + acc = accumulate_delta(acc, chunk2) + assert len(acc["tool_calls"]) == 1 + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"city": "London"}' + + +def test_three_duplicate_indexes_in_first_chunk() -> None: + """Three entries with the same index in the first chunk should all be merged.""" + chunk1 = { + "tool_calls": [ + {"index": 0, "id": "call_1", "type": "function", "function": {"name": "test", "arguments": ""}}, + {"index": 0, "function": {"arguments": ' {"a"'}}, + {"index": 0, "function": {"arguments": ': 1}'}}, + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 1 + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"a": 1}' From 3eba24ca8ff27babdb07563abd634a89b48e60a2 Mon Sep 17 00:00:00 2001 From: ncttjz Date: Thu, 16 Jul 2026 00:33:20 +0700 Subject: [PATCH 2/2] Normalize first chunk snapshot through accumulate_delta The initial chunk snapshot was created by dumping choice.delta.to_dict() directly into the message field, bypassing accumulate_delta's list index normalization. When the first SSE chunk contains multiple tool_call deltas with the same index, they were stored as separate physical list items, causing subsequent delta fragments to merge into only the first entry while stranding the duplicates with partial data. Fix: route the initial snapshot through accumulate_delta({}, delta) so duplicate-indexed tool_calls are merged/normalized from the start. --- src/openai/lib/streaming/chat/_completions.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py index 5f072cafbd..d5a3cee360 100644 --- a/src/openai/lib/streaming/chat/_completions.py +++ b/src/openai/lib/streaming/chat/_completions.py @@ -744,7 +744,10 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh for choice in chunk.choices: choices[choice.index] = { **choice.model_dump(exclude_unset=True, exclude={"delta"}), - "message": choice.delta.to_dict(), + "message": accumulate_delta( + cast("dict[object, object]", {}), + cast("dict[object, object]", choice.delta.to_dict()), + ), } return cast(