From 5fd5c221b3a783ff15d65f13860863a74178bad2 Mon Sep 17 00:00:00 2001 From: Chukwuebuka-2003 Date: Sun, 12 Jul 2026 15:54:26 +0000 Subject: [PATCH 1/2] fix(streaming): merge duplicate-index tool_call deltas in first chunk When the first streamed chunk carries multiple tool_calls entries with the same `index` (e.g. tool name + start of arguments at index 0), the accumulator stored them as separate physical list entries. Every subsequent delta merged only into the first entry, leaving the orphan fragment stranded and producing invalid final JSON. The root cause is two early-return paths in `accumulate_delta` that assign the raw list before checking for duplicate logical indices. The same path exists in the assistants API's copy of the function. Fix: 1. Add `_normalize_indexed_list()` that merges entries sharing the same integer `index` into a single entry before first storage. 2. Apply it at both early-return branches (key absent, value None). 3. Also normalize the initial snapshot in `_convert_initial_chunk_into_snapshot` so that a first chunk with duplicate indices doesn't seed bad state. Applies to both `_deltas.py` (chat completions) and the duplicated copy in `_assistants.py`. Fixes #3201 --- src/openai/lib/streaming/_assistants.py | 48 ++++++++++++++++++- src/openai/lib/streaming/_deltas.py | 48 ++++++++++++++++++- src/openai/lib/streaming/chat/_completions.py | 8 +++- 3 files changed, 98 insertions(+), 6 deletions(-) diff --git a/src/openai/lib/streaming/_assistants.py b/src/openai/lib/streaming/_assistants.py index 6efb3ca3f1..0069c56daf 100644 --- a/src/openai/lib/streaming/_assistants.py +++ b/src/openai/lib/streaming/_assistants.py @@ -977,15 +977,59 @@ def accumulate_event( return current_message_snapshot, new_content +def _normalize_indexed_list(items: list[object]) -> list[object]: + """Merge list entries that share the same integer ``index`` key. + + When the first delta chunk for a key carries multiple entries with the + same ``index`` (e.g. tool name **and** the start of arguments both at + index 0), those entries *must* be merged into a single logical entry. + Without this step the accumulator stores them as separate physical list + entries; the second entry is then orphaned because every subsequent delta + merges into ``acc_value[index]`` (the first one), leaving argument + fragments stranded and producing invalid final JSON. + """ + if not items: + return items + if not all(is_dict(item) for item in items): + return items + + # Count occurrences of each integer index. + index_counts: dict[int, int] = {} + for item in items: + idx = item.get("index") + if isinstance(idx, int): + index_counts[idx] = index_counts.get(idx, 0) + 1 + + # No duplicate indices -> nothing to normalise. + if not index_counts or max(index_counts.values()) <= 1: + return items + + # Merge entries that share the same index. + merged: dict[int, dict[object, object]] = {} + for item in items: + idx = item.get("index") + if isinstance(idx, int): + if idx in merged: + merged[idx] = accumulate_delta(merged[idx], dict(item)) + else: + merged[idx] = dict(item) + + # Rebuild the list ordered by index. + return [ + {**entry, "index": idx} + for idx, entry in sorted(merged.items()) + ] + + 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 + acc[key] = _normalize_indexed_list(delta_value) if is_list(delta_value) else delta_value continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value + acc[key] = _normalize_indexed_list(delta_value) if is_list(delta_value) else delta_value continue # the `index` property is used in arrays of objects so it should diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py index a5e1317612..1e6a0f4603 100644 --- a/src/openai/lib/streaming/_deltas.py +++ b/src/openai/lib/streaming/_deltas.py @@ -3,15 +3,59 @@ from ..._utils import is_dict, is_list +def _normalize_indexed_list(items: list[object]) -> list[object]: + """Merge list entries that share the same integer ``index`` key. + + When the first delta chunk for a key carries multiple entries with the + same ``index`` (e.g. tool name **and** the start of arguments both at + index 0), those entries *must* be merged into a single logical entry. + Without this step the accumulator stores them as separate physical list + entries; the second entry is then orphaned because every subsequent delta + merges into ``acc_value[index]`` (the first one), leaving argument + fragments stranded and producing invalid final JSON. + """ + if not items: + return items + if not all(is_dict(item) for item in items): + return items + + # Count occurrences of each integer index. + index_counts: dict[int, int] = {} + for item in items: + idx = item.get("index") + if isinstance(idx, int): + index_counts[idx] = index_counts.get(idx, 0) + 1 + + # No duplicate indices -> nothing to normalise. + if not index_counts or max(index_counts.values()) <= 1: + return items + + # Merge entries that share the same index. + merged: dict[int, dict[object, object]] = {} + for item in items: + idx = item.get("index") + if isinstance(idx, int): + if idx in merged: + merged[idx] = accumulate_delta(merged[idx], dict(item)) + else: + merged[idx] = dict(item) + + # Rebuild the list ordered by index. + return [ + {**entry, "index": idx} + for idx, entry in sorted(merged.items()) + ] + + 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 + acc[key] = _normalize_indexed_list(delta_value) if is_list(delta_value) else delta_value continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value + acc[key] = _normalize_indexed_list(delta_value) if is_list(delta_value) else delta_value continue # the `index` property is used in arrays of objects so it should diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py index 5f072cafbd..3e4ffe00b6 100644 --- a/src/openai/lib/streaming/chat/_completions.py +++ b/src/openai/lib/streaming/chat/_completions.py @@ -22,7 +22,7 @@ FunctionToolCallArgumentsDoneEvent, FunctionToolCallArgumentsDeltaEvent, ) -from .._deltas import accumulate_delta +from .._deltas import _normalize_indexed_list, accumulate_delta from ...._types import Omit, IncEx, omit from ...._utils import is_given, consume_sync_iterator, consume_async_iterator from ...._compat import model_dump @@ -742,9 +742,13 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh choices = cast("list[object]", data["choices"]) for choice in chunk.choices: + message = choice.delta.to_dict() + tool_calls = message.get("tool_calls") + if isinstance(tool_calls, list): + message["tool_calls"] = _normalize_indexed_list(tool_calls) choices[choice.index] = { **choice.model_dump(exclude_unset=True, exclude={"delta"}), - "message": choice.delta.to_dict(), + "message": message, } return cast( From ec1c5accae9780605ec51c8de82980e371112bc7 Mon Sep 17 00:00:00 2001 From: Chukwuebuka-2003 Date: Sun, 12 Jul 2026 16:12:24 +0000 Subject: [PATCH 2/2] fix(streaming): normalize tool_calls in mid-stream new-choice path too The IndexError path in _accumulate_chunk creates a snapshot for a choice that appears after the first chunk (multi-choice streams). The raw choice.delta.to_dict() was stored directly without normalizing duplicate-index tool_call entries, leaving the same orphan bug as the initial snapshot. Apply _normalize_indexed_list() here as well. --- src/openai/lib/streaming/chat/_completions.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py index 3e4ffe00b6..5f72bda6fd 100644 --- a/src/openai/lib/streaming/chat/_completions.py +++ b/src/openai/lib/streaming/chat/_completions.py @@ -409,13 +409,17 @@ def _accumulate_chunk(self, chunk: ChatCompletionChunk) -> ParsedChatCompletionS elif TYPE_CHECKING: # type: ignore[unreachable] assert_never(prev_tool) except IndexError: + message = choice.delta.to_dict() + tool_calls = message.get("tool_calls") + if isinstance(tool_calls, list): + message["tool_calls"] = _normalize_indexed_list(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": message, }, ), )