From d2b3f8d5ccb8c38d4b5444aa9065b7d85439c783 Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:56:53 +0800 Subject: [PATCH 01/11] fix(chat-completions): tolerate a missing tool call index when buffering streamed tool calls With buffer_streamed_tool_calls=True, the buffered path keyed tool calls on tool_call_delta.index. The OpenAI SDK's lenient chunk parsing leaves that index as None when an OpenAI-compatible provider omits it, so the replayed ChoiceDeltaToolCall failed pydantic validation and sorting the buffered calls raised TypeError once None and int keys coexisted. The unbuffered path already handled the same stream. Replay indexed calls first and give the index-less call the next free index when building the buffered chunk. Co-Authored-By: Claude Fable 5.1 --- src/agents/models/chatcmpl_stream_handler.py | 25 ++++- .../test_openai_chatcompletions_stream.py | 106 ++++++++++++++++++ 2 files changed, 128 insertions(+), 3 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 289e116b06..477df86149 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -157,6 +157,13 @@ class _BufferedToolCall: extra_content: dict[str, Any] | None = None +def _buffered_tool_call_order(buffered_call: _BufferedToolCall) -> tuple[bool, int]: + """Sort indexed tool calls by index and keep an index-less call after them.""" + if isinstance(buffered_call.index, int): + return (False, buffered_call.index) + return (True, 0) + + def _merge_buffered_metadata( current: dict[str, Any] | None, incoming: dict[str, Any], @@ -341,6 +348,7 @@ def _accumulate_tool_call_delta( @staticmethod def _buffered_tool_call_delta( buffered_call: _BufferedToolCall, + fallback_index: int = 0, ) -> ChoiceDeltaToolCall: if not buffered_call.call_id: raise ModelBehaviorError( @@ -353,7 +361,9 @@ def _buffered_tool_call_delta( ) tool_call_delta = ChoiceDeltaToolCall( - index=buffered_call.index, + # Lenient chunk parsing leaves the index as None when the provider omitted it, + # and the replayed delta needs a real index. + index=buffered_call.index if isinstance(buffered_call.index, int) else fallback_index, id=buffered_call.call_id, function=ChoiceDeltaToolCallFunction( name=buffered_call.name, @@ -376,9 +386,18 @@ def _buffered_tool_calls_chunk( template_chunk: ChatCompletionChunk, buffered_calls: dict[int, _BufferedToolCall], ) -> ChatCompletionChunk: + # OpenAI-compatible providers may omit the tool call index, which lenient chunk + # parsing leaves as None. Every index-less delta accumulates under that single key, + # so replay the indexed calls in index order and give the index-less call the next + # free index instead of failing on a None/int comparison. + ordered_calls = sorted(buffered_calls.values(), key=_buffered_tool_call_order) + fallback_index = ( + max((call.index for call in ordered_calls if isinstance(call.index, int)), default=-1) + + 1 + ) tool_call_deltas = [ - cls._buffered_tool_call_delta(buffered_call) - for _, buffered_call in sorted(buffered_calls.items()) + cls._buffered_tool_call_delta(buffered_call, fallback_index=fallback_index) + for buffered_call in ordered_calls ] choice = Choice( index=0, diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 5c79493769..817f302040 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -5,6 +5,7 @@ import httpx2 import pytest +from openai._models import construct_type from openai.types.chat.chat_completion import ChatCompletion, Choice as ChatCompletionChoice from openai.types.chat.chat_completion_chunk import ( ChatCompletionChunk, @@ -107,6 +108,44 @@ async def _collect_buffered_tool_call_chunks( ] +async def _collect_buffered_handler_events(*chunks: ChatCompletionChunk) -> list[Any]: + return [ + event + async for event in ChatCmplStreamHandler.handle_stream( + _empty_response(), + cast(Any, ChatCmplStreamHandler.buffer_tool_call_stream(_completion_stream(*chunks))), + ) + ] + + +def _lenient_chunk(delta: dict[str, Any], finish_reason: str | None = None) -> ChatCompletionChunk: + # Build the chunk the way ``AsyncStream`` does: ``construct_type`` does not validate the + # provider payload, so a tool call delta that omits ``index`` keeps ``index=None`` instead + # of failing validation. That is the shape OpenAI-compatible providers can produce. + return cast( + ChatCompletionChunk, + construct_type( + type_=ChatCompletionChunk, + value={ + "id": "chunk-id", + "object": "chat.completion.chunk", + "created": 1, + "model": "fake", + "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}], + }, + ), + ) + + +def _completed_function_calls(events: list[Any]) -> list[tuple[str, str, str]]: + completed = cast(ResponseCompletedEvent, events[-1]) + return [ + (item.call_id, item.name, item.arguments) + for item in completed.response.output + if isinstance(item, ResponseFunctionToolCall) + ] + + def _url_citation( url: str = "https://example.com/weather", title: str = "Weather", @@ -858,6 +897,73 @@ async def test_buffer_tool_call_stream_keeps_passthrough_index_passthrough() -> assert buffered_chunks[1].choices[0].delta.tool_calls == [function_tool_call_delta] +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_tolerates_missing_tool_call_index() -> None: + chunks = ( + _lenient_chunk( + { + "role": "assistant", + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": '{"a":'}, + } + ], + } + ), + _lenient_chunk({"tool_calls": [{"function": {"arguments": "1}"}}]}), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + first_delta = chunks[0].choices[0].delta + assert first_delta.tool_calls and first_delta.tool_calls[0].index is None + + unbuffered_events = await _collect_handler_events(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + expected_calls = [("call_1", "my_func", '{"a":1}')] + assert _completed_function_calls(unbuffered_events) == expected_calls + assert _completed_function_calls(buffered_events) == expected_calls + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_orders_missing_index_after_indexed_calls() -> None: + chunks = ( + _lenient_chunk( + { + "role": "assistant", + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "first_func", "arguments": "{}"}, + } + ], + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_2", + "type": "function", + "function": {"name": "second_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + unbuffered_events = await _collect_handler_events(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + expected_calls = [("call_1", "first_func", "{}"), ("call_2", "second_func", "{}")] + assert _completed_function_calls(unbuffered_events) == expected_calls + assert _completed_function_calls(buffered_events) == expected_calls + + @pytest.mark.parametrize( ("delta", "expected"), [ From 993094149bec8894e3cd2ed6587b5a38eb827310 Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Wed, 2 Sep 2026 14:55:01 +0800 Subject: [PATCH 02/11] Fix missing-index tool call buffering edge cases --- src/agents/models/chatcmpl_stream_handler.py | 45 ++++++---- .../test_openai_chatcompletions_stream.py | 89 +++++++++++++++++++ 2 files changed, 117 insertions(+), 17 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 477df86149..ab492b3822 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -149,7 +149,7 @@ def _open_thinking_block(self) -> dict[str, Any]: class _BufferedToolCall: """Accumulates a streamed Chat Completions function tool call.""" - index: int + index: int | None call_id: str | None = None name: str | None = None arguments: str = "" @@ -314,12 +314,24 @@ def _delta_has_passthrough_output(delta: ChoiceDelta | None) -> bool: @staticmethod def _accumulate_tool_call_delta( - buffered_calls: dict[int, _BufferedToolCall], + buffered_calls: dict[int | None, _BufferedToolCall], tool_call_delta: ChoiceDeltaToolCall, ) -> None: + tool_call_index = tool_call_delta.index + if not isinstance(tool_call_index, int) and not tool_call_delta.id: + if None in buffered_calls: + tool_call_index = None + elif len(buffered_calls) == 1: + tool_call_index = next(iter(buffered_calls)) + elif len(buffered_calls) > 1: + raise ModelBehaviorError( + "Chat Completions tool call delta omitted an index while multiple " + "function tool calls were being buffered." + ) + buffered_call = buffered_calls.setdefault( - tool_call_delta.index, - _BufferedToolCall(index=tool_call_delta.index), + tool_call_index, + _BufferedToolCall(index=tool_call_index), ) if tool_call_delta.id: @@ -361,8 +373,6 @@ def _buffered_tool_call_delta( ) tool_call_delta = ChoiceDeltaToolCall( - # Lenient chunk parsing leaves the index as None when the provider omitted it, - # and the replayed delta needs a real index. index=buffered_call.index if isinstance(buffered_call.index, int) else fallback_index, id=buffered_call.call_id, function=ChoiceDeltaToolCallFunction( @@ -384,17 +394,14 @@ def _buffered_tool_call_delta( def _buffered_tool_calls_chunk( cls, template_chunk: ChatCompletionChunk, - buffered_calls: dict[int, _BufferedToolCall], + buffered_calls: dict[int | None, _BufferedToolCall], + passthrough_tool_call_indexes: set[int], ) -> ChatCompletionChunk: - # OpenAI-compatible providers may omit the tool call index, which lenient chunk - # parsing leaves as None. Every index-less delta accumulates under that single key, - # so replay the indexed calls in index order and give the index-less call the next - # free index instead of failing on a None/int comparison. ordered_calls = sorted(buffered_calls.values(), key=_buffered_tool_call_order) - fallback_index = ( - max((call.index for call in ordered_calls if isinstance(call.index, int)), default=-1) - + 1 - ) + occupied_indexes = passthrough_tool_call_indexes | { + call.index for call in ordered_calls if isinstance(call.index, int) + } + fallback_index = max(occupied_indexes, default=-1) + 1 tool_call_deltas = [ cls._buffered_tool_call_delta(buffered_call, fallback_index=fallback_index) for buffered_call in ordered_calls @@ -412,7 +419,7 @@ async def buffer_tool_call_stream( stream: AsyncIterator[ChatCompletionChunk], ) -> AsyncIterator[ChatCompletionChunk]: """Buffer streamed function tool-call deltas until they are complete.""" - buffered_calls: dict[int, _BufferedToolCall] = {} + buffered_calls: dict[int | None, _BufferedToolCall] = {} passthrough_tool_call_indexes: set[int] = set() saw_passthrough_tool_call = False last_chunk: ChatCompletionChunk | None = None @@ -477,7 +484,11 @@ async def buffer_tool_call_stream( if buffered_calls: if last_chunk is None: return - yield cls._buffered_tool_calls_chunk(last_chunk, buffered_calls) + yield cls._buffered_tool_calls_chunk( + last_chunk, + buffered_calls, + passthrough_tool_call_indexes, + ) @staticmethod def _merged_provider_data( diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 817f302040..358d9ccac5 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -964,6 +964,95 @@ async def test_buffer_tool_call_stream_orders_missing_index_after_indexed_calls( assert _completed_function_calls(buffered_events) == expected_calls +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_avoids_passthrough_index_for_missing_index() -> None: + custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( + index=0, + id="custom-id", + type="custom", + ) + custom_chunk = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[custom_tool_call_delta]))], + ) + chunks = ( + custom_chunk, + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_events = await _collect_buffered_handler_events(*chunks) + + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_merges_missing_index_continuation() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": '{"a":'}, + } + ] + } + ), + _lenient_chunk({"tool_calls": [{"function": {"arguments": "1}"}}]}), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_events = await _collect_buffered_handler_events(*chunks) + + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_ambiguous_missing_index_continuation() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "first_func", "arguments": '{"a":'}, + }, + { + "index": 1, + "id": "call_2", + "type": "function", + "function": {"name": "second_func", "arguments": '{"b":'}, + }, + ] + } + ), + _lenient_chunk({"tool_calls": [{"function": {"arguments": "1}"}}]}), + ) + + with pytest.raises( + ModelBehaviorError, match="omitted an index while multiple function tool calls" + ): + await _collect_buffered_handler_events(*chunks) + + @pytest.mark.parametrize( ("delta", "expected"), [ From dacd81d21e30dd7c1f9910a214e3b1912881a1b6 Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Wed, 2 Sep 2026 15:29:50 +0800 Subject: [PATCH 03/11] Handle missing passthrough tool call indexes --- src/agents/models/chatcmpl_stream_handler.py | 3 +- .../test_openai_chatcompletions_stream.py | 35 +++++++++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index ab492b3822..12038083cf 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -450,7 +450,8 @@ async def buffer_tool_call_stream( elif cls._should_buffer_tool_call_delta(tool_call_delta): cls._accumulate_tool_call_delta(buffered_calls, tool_call_delta) else: - passthrough_tool_call_indexes.add(tool_call_delta.index) + if isinstance(tool_call_delta.index, int): + passthrough_tool_call_indexes.add(tool_call_delta.index) saw_passthrough_tool_call = True remaining_tool_calls.append(tool_call_delta) diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 358d9ccac5..929881fb99 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -999,6 +999,41 @@ async def test_buffer_tool_call_stream_avoids_passthrough_index_for_missing_inde assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] +@pytest.mark.parametrize("function_tool_call_index", [0, None]) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_ignores_missing_passthrough_index( + function_tool_call_index: int | None, +) -> None: + custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( + index=None, + id="custom-id", + type="custom", + ) + custom_chunk = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[custom_tool_call_delta]))], + ) + function_tool_call: dict[str, Any] = { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + if function_tool_call_index is not None: + function_tool_call["index"] = function_tool_call_index + chunks = ( + custom_chunk, + _lenient_chunk({"tool_calls": [function_tool_call]}), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_events = await _collect_buffered_handler_events(*chunks) + + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + + @pytest.mark.asyncio async def test_buffer_tool_call_stream_merges_missing_index_continuation() -> None: chunks = ( From 3a2eef5c45dc7d5ccf6be9f1fa4eb8c0dcdd81fe Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:21:40 +0800 Subject: [PATCH 04/11] Resolve missing-index tool call identity --- src/agents/models/chatcmpl_stream_handler.py | 155 ++++- .../test_openai_chatcompletions_stream.py | 588 +++++++++++++++++- 2 files changed, 726 insertions(+), 17 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 12038083cf..ccfe0ce316 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -157,11 +157,25 @@ class _BufferedToolCall: extra_content: dict[str, Any] | None = None -def _buffered_tool_call_order(buffered_call: _BufferedToolCall) -> tuple[bool, int]: - """Sort indexed tool calls by index and keep an index-less call after them.""" - if isinstance(buffered_call.index, int): - return (False, buffered_call.index) - return (True, 0) +def _buffered_tool_calls_in_replay_order( + buffered_calls: dict[int | None, _BufferedToolCall], +) -> list[_BufferedToolCall]: + """Sort indexed calls while preserving where the index-less call first appeared.""" + indexed_calls = sorted( + (call for call in buffered_calls.values() if isinstance(call.index, int)), + key=lambda call: cast(int, call.index), + ) + if None not in buffered_calls: + return indexed_calls + + unindexed_position = 0 + for index in buffered_calls: + if index is None: + break + unindexed_position += 1 + + indexed_calls.insert(unindexed_position, buffered_calls[None]) + return indexed_calls def _merge_buffered_metadata( @@ -318,16 +332,55 @@ def _accumulate_tool_call_delta( tool_call_delta: ChoiceDeltaToolCall, ) -> None: tool_call_index = tool_call_delta.index - if not isinstance(tool_call_index, int) and not tool_call_delta.id: - if None in buffered_calls: - tool_call_index = None - elif len(buffered_calls) == 1: - tool_call_index = next(iter(buffered_calls)) - elif len(buffered_calls) > 1: + if not isinstance(tool_call_index, int): + matching_indexes = [ + index + for index, buffered_call in buffered_calls.items() + if tool_call_delta.id and buffered_call.call_id == tool_call_delta.id + ] + if len(matching_indexes) == 1: + tool_call_index = matching_indexes[0] + elif len(matching_indexes) > 1: + raise ModelBehaviorError( + "Chat Completions tool call delta omitted an index and the same ID " + "matched multiple buffered calls." + ) + elif tool_call_delta.id and None in buffered_calls and buffered_calls[None].call_id: raise ModelBehaviorError( - "Chat Completions tool call delta omitted an index while multiple " - "function tool calls were being buffered." + "Chat Completions tool call delta omitted an index with a new ID while " + "another index-less call was being buffered." ) + else: + function_name = tool_call_delta.function.name if tool_call_delta.function else None + if function_name and None in buffered_calls: + buffered_name = buffered_calls[None].name + if buffered_name and buffered_name != function_name: + raise ModelBehaviorError( + "Chat Completions tool call delta omitted an index and used a " + "different function name from the buffered index-less call." + ) + + if not tool_call_delta.id and function_name: + if None in buffered_calls: + tool_call_index = None + elif len(buffered_calls) == 1: + sole_index = next(iter(buffered_calls)) + sole_name = buffered_calls[sole_index].name + if not sole_name or sole_name == function_name: + tool_call_index = sole_index + else: + tool_call_index = None + else: + tool_call_index = None + elif not tool_call_delta.id and None in buffered_calls: + tool_call_index = None + elif not tool_call_delta.id and len(buffered_calls) == 1: + tool_call_index = next(iter(buffered_calls)) + elif not tool_call_delta.id and len(buffered_calls) > 1: + raise ModelBehaviorError( + "Chat Completions tool call delta omitted an index while multiple " + "function tool calls were being buffered." + ) buffered_call = buffered_calls.setdefault( tool_call_index, @@ -397,7 +450,7 @@ def _buffered_tool_calls_chunk( buffered_calls: dict[int | None, _BufferedToolCall], passthrough_tool_call_indexes: set[int], ) -> ChatCompletionChunk: - ordered_calls = sorted(buffered_calls.values(), key=_buffered_tool_call_order) + ordered_calls = _buffered_tool_calls_in_replay_order(buffered_calls) occupied_indexes = passthrough_tool_call_indexes | { call.index for call in ordered_calls if isinstance(call.index, int) } @@ -421,6 +474,8 @@ async def buffer_tool_call_stream( """Buffer streamed function tool-call deltas until they are complete.""" buffered_calls: dict[int | None, _BufferedToolCall] = {} passthrough_tool_call_indexes: set[int] = set() + passthrough_tool_call_indexes_by_id: dict[str, int | None] = {} + saw_unindexed_passthrough_tool_call = False saw_passthrough_tool_call = False last_chunk: ChatCompletionChunk | None = None @@ -444,14 +499,84 @@ async def buffer_tool_call_stream( if tool_call_deltas := (delta.tool_calls if delta and delta.tool_calls else None): remaining_tool_calls: list[ChoiceDeltaToolCall] = [] for tool_call_delta in tool_call_deltas: - if tool_call_delta.index in passthrough_tool_call_indexes: + is_unindexed_untyped_continuation = ( + not isinstance(tool_call_delta.index, int) + and getattr(tool_call_delta, "type", None) is None + and tool_call_delta.function is None + ) + buffered_id_matches = [ + buffered_call + for buffered_call in buffered_calls.values() + if tool_call_delta.id and buffered_call.call_id == tool_call_delta.id + ] + is_unindexed_passthrough_continuation = ( + is_unindexed_untyped_continuation + and ( + tool_call_delta.id in passthrough_tool_call_indexes_by_id + or (saw_unindexed_passthrough_tool_call and not tool_call_delta.id) + ) + ) + if is_unindexed_passthrough_continuation and buffered_id_matches: + raise ModelBehaviorError( + "Chat Completions tool call delta omitted an index and its ID " + "matched both a buffered function call and a passthrough call." + ) + + if ( + tool_call_delta.index in passthrough_tool_call_indexes + or is_unindexed_passthrough_continuation + ): + if passthrough_id := tool_call_delta.id: + owner_index = passthrough_tool_call_indexes_by_id.get( + passthrough_id + ) + if isinstance(owner_index, int) and not isinstance( + tool_call_delta.index, int + ): + tool_call_delta = tool_call_delta.model_copy( + update={"index": owner_index} + ) + passthrough_tool_call_indexes_by_id.setdefault( + passthrough_id, + tool_call_delta.index + if isinstance(tool_call_delta.index, int) + else None, + ) saw_passthrough_tool_call = True remaining_tool_calls.append(tool_call_delta) + elif ( + is_unindexed_untyped_continuation + and saw_passthrough_tool_call + and ( + not tool_call_delta.id + or ( + bool( + set(tool_call_delta.model_extra or {}) + - {"provider_specific_fields", "extra_content"} + ) + and not buffered_id_matches + ) + ) + ): + raise ModelBehaviorError( + "Chat Completions tool call delta omitted an index, type, and " + "function payload after a passthrough call, so it could not be " + "attributed safely." + ) elif cls._should_buffer_tool_call_delta(tool_call_delta): cls._accumulate_tool_call_delta(buffered_calls, tool_call_delta) else: if isinstance(tool_call_delta.index, int): passthrough_tool_call_indexes.add(tool_call_delta.index) + else: + saw_unindexed_passthrough_tool_call = True + if tool_call_delta.id: + passthrough_tool_call_indexes_by_id.setdefault( + tool_call_delta.id, + tool_call_delta.index + if isinstance(tool_call_delta.index, int) + else None, + ) saw_passthrough_tool_call = True remaining_tool_calls.append(tool_call_delta) diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 929881fb99..6549684b4f 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -964,6 +964,82 @@ async def test_buffer_tool_call_stream_orders_missing_index_after_indexed_calls( assert _completed_function_calls(buffered_events) == expected_calls +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_preserves_missing_index_call_arrival_order() -> None: + chunks = ( + _lenient_chunk( + { + "role": "assistant", + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "first_func", "arguments": "{}"}, + } + ], + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_2", + "type": "function", + "function": {"name": "second_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + unbuffered_events = await _collect_handler_events(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + expected_calls = [("call_1", "first_func", "{}"), ("call_2", "second_func", "{}")] + assert _completed_function_calls(unbuffered_events) == expected_calls + assert _completed_function_calls(buffered_events) == expected_calls + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_sorts_indexed_calls_by_index() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 1, + "id": "call_2", + "type": "function", + "function": {"name": "second_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "first_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_events = await _collect_buffered_handler_events(*chunks) + + assert _completed_function_calls(buffered_events) == [ + ("call_1", "first_func", "{}"), + ("call_2", "second_func", "{}"), + ] + + @pytest.mark.asyncio async def test_buffer_tool_call_stream_avoids_passthrough_index_for_missing_index() -> None: custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( @@ -1034,8 +1110,280 @@ async def test_buffer_tool_call_stream_ignores_missing_passthrough_index( assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] +@pytest.mark.parametrize("continuation_id", [None, "custom-id"]) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_forwards_missing_index_passthrough_continuation( + continuation_id: str | None, +) -> None: + continuation: dict[str, Any] = {"custom": {"input": "nt(1)"}} + if continuation_id is not None: + continuation["id"] = continuation_id + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "pri"}, + } + ] + } + ), + _lenient_chunk({"tool_calls": [continuation]}), + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + assert buffered_chunks[0].choices[0].delta.tool_calls == chunks[0].choices[0].delta.tool_calls + assert buffered_chunks[1].choices[0].delta.tool_calls == chunks[1].choices[0].delta.tool_calls + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + + +@pytest.mark.parametrize("function_tool_call_index", [1, None], ids=["indexed", "unindexed"]) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_forwards_indexed_passthrough_continuation_by_id( + function_tool_call_index: int | None, +) -> None: + function_tool_call: dict[str, Any] = { + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + if function_tool_call_index is not None: + function_tool_call["index"] = function_tool_call_index + function_tool_call["id"] = "call_1" + + chunks = [ + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "pri"}, + } + ] + } + ), + _lenient_chunk({"tool_calls": [function_tool_call]}), + _lenient_chunk({"tool_calls": [{"id": "custom-id", "custom": {"input": "nt(1)"}}]}), + ] + if function_tool_call_index is None: + chunks.append(_lenient_chunk({"tool_calls": [{"id": "call_1"}]})) + chunks.append(_lenient_chunk({}, finish_reason="tool_calls")) + + buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + continuation_tool_calls = buffered_chunks[1].choices[0].delta.tool_calls + assert continuation_tool_calls + assert continuation_tool_calls[0].index == 0 + assert continuation_tool_calls[0].id == "custom-id" + assert continuation_tool_calls[0].model_extra == {"custom": {"input": "nt(1)"}} + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_records_passthrough_id_from_indexed_continuation() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "type": "custom", + "custom": {"name": "code_exec", "input": "pri"}, + } + ] + } + ), + _lenient_chunk( + {"tool_calls": [{"index": 0, "id": "custom-id", "custom": {"input": "nt("}}]} + ), + _lenient_chunk( + { + "tool_calls": [ + { + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({"tool_calls": [{"id": "custom-id", "custom": {"input": "1)"}}]}), + _lenient_chunk({"tool_calls": [{"id": "call_1"}]}), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + continuation_tool_calls = buffered_chunks[2].choices[0].delta.tool_calls + assert continuation_tool_calls + assert continuation_tool_calls[0].index == 0 + assert continuation_tool_calls[0].id == "custom-id" + assert continuation_tool_calls[0].model_extra == {"custom": {"input": "1)"}} + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + + @pytest.mark.asyncio -async def test_buffer_tool_call_stream_merges_missing_index_continuation() -> None: +async def test_buffer_tool_call_stream_rejects_cross_domain_passthrough_id() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "shared-id", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 1, + "id": "shared-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "print(1)"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "id": "shared-id", + "extra_content": {"google": {"thought_signature": "sig"}}, + } + ] + } + ), + ) + + with pytest.raises(ModelBehaviorError, match="matched both"): + await _collect_buffered_tool_call_chunks(*chunks) + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_allows_function_metadata_on_late_id() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "print(1)"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "extra_content": {"google": {"thought_signature": "sig"}}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) + unbuffered_events = await _collect_handler_events(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + replayed_tool_calls = buffered_chunks[-1].choices[0].delta.tool_calls + assert replayed_tool_calls + assert cast(Any, replayed_tool_calls[0]).extra_content == { + "google": {"thought_signature": "sig"} + } + expected_calls = [("call_1", "my_func", "{}")] + assert _completed_function_calls(unbuffered_events) == expected_calls + assert _completed_function_calls(buffered_events) == expected_calls + + +@pytest.mark.parametrize("continuation_id", [None, "unknown-id"], ids=["without-id", "unknown-id"]) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_ambiguous_unindexed_passthrough_continuation( + continuation_id: str | None, +) -> None: + continuation: dict[str, Any] = {"custom": {"input": "nt(1)"}} + if continuation_id is not None: + continuation["id"] = continuation_id + + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "pri"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 1, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({"tool_calls": [continuation]}), + ) + + with pytest.raises(ModelBehaviorError, match="could not be attributed"): + await _collect_buffered_tool_call_chunks(*chunks) + + +@pytest.mark.parametrize("continuation_name", [None, "my_func"]) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_merges_missing_index_continuation( + continuation_name: str | None, +) -> None: + continuation_function = {"arguments": "1}"} + if continuation_name is not None: + continuation_function["name"] = continuation_name chunks = ( _lenient_chunk( { @@ -1049,7 +1397,7 @@ async def test_buffer_tool_call_stream_merges_missing_index_continuation() -> No ] } ), - _lenient_chunk({"tool_calls": [{"function": {"arguments": "1}"}}]}), + _lenient_chunk({"tool_calls": [{"function": continuation_function}]}), _lenient_chunk({}, finish_reason="tool_calls"), ) @@ -1058,6 +1406,242 @@ async def test_buffer_tool_call_stream_merges_missing_index_continuation() -> No assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_merges_late_id_into_missing_index_call() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "type": "function", + "function": {"name": "my_func", "arguments": '{"a":'}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"arguments": "1}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + unbuffered_events = await _collect_handler_events(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + expected_calls = [("call_1", "my_func", '{"a":1}')] + assert _completed_function_calls(unbuffered_events) == expected_calls + assert _completed_function_calls(buffered_events) == expected_calls + + +@pytest.mark.parametrize( + "opening_name", + ["my_func", None], + ids=["same-name", "fills-missing-name"], +) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_merges_named_continuation_into_missing_index_slot( + opening_name: str | None, +) -> None: + opening_function = {"arguments": '{"a":'} + if opening_name is not None: + opening_function["name"] = opening_name + + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "type": "function", + "function": opening_function, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "type": "function", + "function": {"name": "my_func", "arguments": "1}"}, + } + ] + } + ), + _lenient_chunk({"tool_calls": [{"id": "call_1"}]}), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + unbuffered_events = await _collect_handler_events(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + expected_calls = [("call_1", "my_func", '{"a":1}')] + assert _completed_function_calls(unbuffered_events) == expected_calls + assert _completed_function_calls(buffered_events) == expected_calls + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_opens_named_missing_index_call_beside_indexed_call() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "first_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "type": "function", + "function": {"name": "second_func", "arguments": '{"b":'}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_2", + "type": "function", + "function": {"arguments": "1}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + unbuffered_events = await _collect_handler_events(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + expected_calls = [ + ("call_1", "first_func", "{}"), + ("call_2", "second_func", '{"b":1}'), + ] + assert _completed_function_calls(unbuffered_events) == expected_calls + assert _completed_function_calls(buffered_events) == expected_calls + + +@pytest.mark.parametrize("continuation_id", [None, "call_2"], ids=["without-id", "with-new-id"]) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_different_name_for_missing_index_slot( + continuation_id: str | None, +) -> None: + continuation: dict[str, Any] = { + "type": "function", + "function": {"name": "second_func", "arguments": "{}"}, + } + if continuation_id is not None: + continuation["id"] = continuation_id + + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "type": "function", + "function": {"name": "first_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({"tool_calls": [continuation]}), + ) + + with pytest.raises(ModelBehaviorError, match="different function name"): + await _collect_buffered_handler_events(*chunks) + + +@pytest.mark.parametrize( + "continuation_function", + [ + {"arguments": "1}"}, + {"name": "my_func", "arguments": "1}"}, + ], + ids=["arguments-only", "repeated-name"], +) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_merges_missing_index_continuation_by_id( + continuation_function: dict[str, str], +) -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": '{"a":'}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": continuation_function, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_events = await _collect_buffered_handler_events(*chunks) + + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] + + +def test_accumulate_tool_call_delta_rejects_ambiguous_repeated_id() -> None: + buffered_calls = { + 0: _BufferedToolCall(index=0, call_id="call_1", name="first_func"), + 1: _BufferedToolCall(index=1, call_id="call_1", name="second_func"), + } + continuation = ChoiceDeltaToolCall.model_construct( + index=None, + id="call_1", + type="function", + function=ChoiceDeltaToolCallFunction(arguments="{}"), + ) + + with pytest.raises(ModelBehaviorError, match="same ID matched multiple buffered calls"): + ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, continuation) + + +def test_accumulate_tool_call_delta_rejects_new_id_for_occupied_missing_index() -> None: + buffered_calls = { + None: _BufferedToolCall(index=None, call_id="call_1", name="first_func"), + } + new_call = ChoiceDeltaToolCall.model_construct( + index=None, + id="call_2", + type="function", + function=ChoiceDeltaToolCallFunction(name="second_func", arguments="{}"), + ) + + with pytest.raises(ModelBehaviorError, match="new ID while another index-less call"): + ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, new_call) + + @pytest.mark.asyncio async def test_buffer_tool_call_stream_rejects_ambiguous_missing_index_continuation() -> None: chunks = ( From 10195b50c6a8e7e2533bfc3b623a1b7aff220b1e Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:36:55 +0800 Subject: [PATCH 05/11] Reconcile late tool call indexes by ID --- src/agents/models/chatcmpl_stream_handler.py | 41 +++++- .../test_openai_chatcompletions_stream.py | 126 ++++++++++++++++++ 2 files changed, 161 insertions(+), 6 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index ccfe0ce316..37fd791cab 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -332,12 +332,28 @@ def _accumulate_tool_call_delta( tool_call_delta: ChoiceDeltaToolCall, ) -> None: tool_call_index = tool_call_delta.index - if not isinstance(tool_call_index, int): - matching_indexes = [ - index - for index, buffered_call in buffered_calls.items() - if tool_call_delta.id and buffered_call.call_id == tool_call_delta.id - ] + matching_indexes = [ + index + for index, buffered_call in buffered_calls.items() + if tool_call_delta.id and buffered_call.call_id == tool_call_delta.id + ] + if isinstance(tool_call_index, int): + if None in matching_indexes: + if len(matching_indexes) > 1: + raise ModelBehaviorError( + "Chat Completions tool call delta supplied an index and the same ID " + "matched multiple buffered calls." + ) + if tool_call_index in buffered_calls: + raise ModelBehaviorError( + "Chat Completions tool call delta supplied an index already used by " + "another buffered call." + ) + + buffered_call = buffered_calls.pop(None) + buffered_call.index = tool_call_index + buffered_calls[tool_call_index] = buffered_call + else: if len(matching_indexes) == 1: tool_call_index = matching_indexes[0] elif len(matching_indexes) > 1: @@ -522,6 +538,19 @@ async def buffer_tool_call_stream( "matched both a buffered function call and a passthrough call." ) + unindexed_buffered_call = buffered_calls.get(None) + if ( + isinstance(tool_call_delta.index, int) + and tool_call_delta.index in passthrough_tool_call_indexes + and tool_call_delta.id + and unindexed_buffered_call is not None + and unindexed_buffered_call.call_id == tool_call_delta.id + ): + raise ModelBehaviorError( + "Chat Completions tool call delta supplied an index already used " + "by a passthrough call." + ) + if ( tool_call_delta.index in passthrough_tool_call_indexes or is_unindexed_passthrough_continuation diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 6549684b4f..6f3d89137f 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -1,6 +1,7 @@ import asyncio import logging from collections.abc import AsyncIterator +from dataclasses import replace from typing import Any, cast import httpx2 @@ -1406,6 +1407,50 @@ async def test_buffer_tool_call_stream_merges_missing_index_continuation( assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_late_index_used_by_passthrough() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "print(1)"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": '{"a":'}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "id": "call_1", + "type": "function", + "function": {"arguments": "1}"}, + } + ] + } + ), + ) + + with pytest.raises(ModelBehaviorError, match="index already used by a passthrough call"): + await _collect_buffered_tool_call_chunks(*chunks) + + @pytest.mark.asyncio async def test_buffer_tool_call_stream_merges_late_id_into_missing_index_call() -> None: chunks = ( @@ -1441,6 +1486,51 @@ async def test_buffer_tool_call_stream_merges_late_id_into_missing_index_call() assert _completed_function_calls(buffered_events) == expected_calls +@pytest.mark.parametrize("continuation_name", [None, "my_func"]) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_reconciles_late_index_by_id( + continuation_name: str | None, +) -> None: + continuation_function = {"arguments": "1}"} + if continuation_name is not None: + continuation_function["name"] = continuation_name + + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": '{"a":'}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "id": "call_1", + "type": "function", + "function": continuation_function, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + replayed_tool_calls = buffered_chunks[-1].choices[0].delta.tool_calls + assert replayed_tool_calls + assert [tool_call.index for tool_call in replayed_tool_calls] == [2] + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] + + @pytest.mark.parametrize( "opening_name", ["my_func", None], @@ -1642,6 +1732,42 @@ def test_accumulate_tool_call_delta_rejects_new_id_for_occupied_missing_index() ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, new_call) +def test_accumulate_tool_call_delta_rejects_late_index_collision() -> None: + unindexed_call = _BufferedToolCall(index=None, call_id="call_1", name="first_func") + indexed_call = _BufferedToolCall(index=2, call_id="call_2", name="second_func") + buffered_calls = {None: unindexed_call, 2: indexed_call} + expected_calls = {None: replace(unindexed_call), 2: replace(indexed_call)} + continuation = ChoiceDeltaToolCall.model_construct( + index=2, + id="call_1", + type="function", + function=ChoiceDeltaToolCallFunction(arguments="{}"), + ) + + with pytest.raises(ModelBehaviorError, match="index already used"): + ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, continuation) + + assert buffered_calls == expected_calls + + +def test_accumulate_tool_call_delta_rejects_ambiguous_late_index() -> None: + unindexed_call = _BufferedToolCall(index=None, call_id="call_1", name="first_func") + indexed_call = _BufferedToolCall(index=1, call_id="call_1", name="second_func") + buffered_calls = {None: unindexed_call, 1: indexed_call} + expected_calls = {None: replace(unindexed_call), 1: replace(indexed_call)} + continuation = ChoiceDeltaToolCall.model_construct( + index=2, + id="call_1", + type="function", + function=ChoiceDeltaToolCallFunction(arguments="{}"), + ) + + with pytest.raises(ModelBehaviorError, match="same ID matched multiple buffered calls"): + ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, continuation) + + assert buffered_calls == expected_calls + + @pytest.mark.asyncio async def test_buffer_tool_call_stream_rejects_ambiguous_missing_index_continuation() -> None: chunks = ( From 122fda0c616ecffe5ecb531afd9914dbed6a4a08 Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:30:10 +0800 Subject: [PATCH 06/11] Handle late indexed continuation ownership --- src/agents/models/chatcmpl_stream_handler.py | 81 ++++++- .../test_openai_chatcompletions_stream.py | 229 ++++++++++++++++++ 2 files changed, 297 insertions(+), 13 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 37fd791cab..937ab5e795 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -332,6 +332,7 @@ def _accumulate_tool_call_delta( tool_call_delta: ChoiceDeltaToolCall, ) -> None: tool_call_index = tool_call_delta.index + function_name = tool_call_delta.function.name if tool_call_delta.function else None matching_indexes = [ index for index, buffered_call in buffered_calls.items() @@ -353,6 +354,21 @@ def _accumulate_tool_call_delta( buffered_call = buffered_calls.pop(None) buffered_call.index = tool_call_index buffered_calls[tool_call_index] = buffered_call + elif ( + not tool_call_delta.id + and None in buffered_calls + and tool_call_index not in buffered_calls + ): + buffered_name = buffered_calls[None].name + if not function_name or not buffered_name or function_name == buffered_name: + if len(buffered_calls) > 1: + raise ModelBehaviorError( + "Chat Completions tool call delta supplied a new index without an ID " + "while multiple function calls were being buffered." + ) + buffered_call = buffered_calls.pop(None) + buffered_call.index = tool_call_index + buffered_calls[tool_call_index] = buffered_call else: if len(matching_indexes) == 1: tool_call_index = matching_indexes[0] @@ -367,7 +383,6 @@ def _accumulate_tool_call_delta( "another index-less call was being buffered." ) else: - function_name = tool_call_delta.function.name if tool_call_delta.function else None if function_name and None in buffered_calls: buffered_name = buffered_calls[None].name if buffered_name and buffered_name != function_name: @@ -515,36 +530,52 @@ async def buffer_tool_call_stream( if tool_call_deltas := (delta.tool_calls if delta and delta.tool_calls else None): remaining_tool_calls: list[ChoiceDeltaToolCall] = [] for tool_call_delta in tool_call_deltas: - is_unindexed_untyped_continuation = ( - not isinstance(tool_call_delta.index, int) - and getattr(tool_call_delta, "type", None) is None + is_untyped_continuation = ( + getattr(tool_call_delta, "type", None) is None and tool_call_delta.function is None ) + is_unindexed_untyped_continuation = ( + not isinstance(tool_call_delta.index, int) and is_untyped_continuation + ) buffered_id_matches = [ buffered_call for buffered_call in buffered_calls.values() if tool_call_delta.id and buffered_call.call_id == tool_call_delta.id ] + is_passthrough_continuation_by_id = ( + is_untyped_continuation + and bool(tool_call_delta.id) + and tool_call_delta.id in passthrough_tool_call_indexes_by_id + ) is_unindexed_passthrough_continuation = ( is_unindexed_untyped_continuation and ( - tool_call_delta.id in passthrough_tool_call_indexes_by_id + is_passthrough_continuation_by_id or (saw_unindexed_passthrough_tool_call and not tool_call_delta.id) ) ) - if is_unindexed_passthrough_continuation and buffered_id_matches: + if is_passthrough_continuation_by_id and buffered_id_matches: raise ModelBehaviorError( - "Chat Completions tool call delta omitted an index and its ID " - "matched both a buffered function call and a passthrough call." + "Chat Completions tool call delta ID matched both a buffered " + "function call and a passthrough call." ) unindexed_buffered_call = buffered_calls.get(None) if ( isinstance(tool_call_delta.index, int) and tool_call_delta.index in passthrough_tool_call_indexes - and tool_call_delta.id and unindexed_buffered_call is not None - and unindexed_buffered_call.call_id == tool_call_delta.id + and ( + ( + tool_call_delta.id + and unindexed_buffered_call.call_id == tool_call_delta.id + ) + or ( + not tool_call_delta.id + and len(buffered_calls) == 1 + and tool_call_delta.function is not None + ) + ) ): raise ModelBehaviorError( "Chat Completions tool call delta supplied an index already used " @@ -553,15 +584,39 @@ async def buffer_tool_call_stream( if ( tool_call_delta.index in passthrough_tool_call_indexes + or is_passthrough_continuation_by_id or is_unindexed_passthrough_continuation ): if passthrough_id := tool_call_delta.id: owner_index = passthrough_tool_call_indexes_by_id.get( passthrough_id ) - if isinstance(owner_index, int) and not isinstance( - tool_call_delta.index, int - ): + if isinstance(tool_call_delta.index, int): + promotes_unindexed_passthrough_owner = ( + is_passthrough_continuation_by_id + and not isinstance(owner_index, int) + ) + if isinstance(owner_index, int) and ( + owner_index != tool_call_delta.index + ): + raise ModelBehaviorError( + "Chat Completions passthrough tool call delta supplied " + "a different index from its buffered ID owner." + ) + if tool_call_delta.index in buffered_calls: + raise ModelBehaviorError( + "Chat Completions passthrough tool call delta supplied " + "an index already used by a buffered function call." + ) + passthrough_tool_call_indexes.add(tool_call_delta.index) + passthrough_tool_call_indexes_by_id[passthrough_id] = ( + tool_call_delta.index + ) + if promotes_unindexed_passthrough_owner: + tool_call_delta = tool_call_delta.model_copy( + update={"type": "custom"} + ) + elif isinstance(owner_index, int): tool_call_delta = tool_call_delta.model_copy( update={"index": owner_index} ) diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 6f3d89137f..9f85db983f 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -1154,6 +1154,132 @@ async def test_buffer_tool_call_stream_forwards_missing_index_passthrough_contin assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_promotes_indexed_passthrough_continuation_by_id() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "pri"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "id": "custom-id", + "custom": {"input": "nt("}, + } + ] + } + ), + _lenient_chunk({"tool_calls": [{"id": "custom-id", "custom": {"input": "1)"}}]}), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + indexed_continuation = buffered_chunks[1].choices[0].delta.tool_calls + unindexed_continuation = buffered_chunks[2].choices[0].delta.tool_calls + assert indexed_continuation and indexed_continuation[0].index == 2 + assert unindexed_continuation and unindexed_continuation[0].index == 2 + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_indexed_passthrough_continuation_collision() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "pri"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "id": "custom-id", + "custom": {"input": "nt(1)"}, + } + ] + } + ), + ) + + with pytest.raises(ModelBehaviorError, match="index already used by a buffered function"): + await _collect_buffered_tool_call_chunks(*chunks) + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_passthrough_continuation_index_change() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 1, + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "pri"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "id": "custom-id", + "custom": {"input": "nt(1)"}, + } + ] + } + ), + ) + + with pytest.raises(ModelBehaviorError, match="different index from its buffered ID owner"): + await _collect_buffered_tool_call_chunks(*chunks) + + @pytest.mark.parametrize("function_tool_call_index", [1, None], ids=["indexed", "unindexed"]) @pytest.mark.asyncio async def test_buffer_tool_call_stream_forwards_indexed_passthrough_continuation_by_id( @@ -1531,6 +1657,91 @@ async def test_buffer_tool_call_stream_reconciles_late_index_by_id( assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_reconciles_late_index_without_id_for_sole_call() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": '{"a":'}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "type": "function", + "function": {"arguments": "1}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + replayed_tool_calls = buffered_chunks[-1].choices[0].delta.tool_calls + assert replayed_tool_calls + assert [tool_call.index for tool_call in replayed_tool_calls] == [2] + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_keeps_different_named_idless_late_index_distinct() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "first_func", "arguments": '{"a":1}'}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "type": "function", + "function": {"name": "second_func", "arguments": '{"b":'}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "id": "call_2", + "type": "function", + "function": {"arguments": "1}"}, + } + ] + } + ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_events = await _collect_buffered_handler_events(*chunks) + + assert _completed_function_calls(buffered_events) == [ + ("call_1", "first_func", '{"a":1}'), + ("call_2", "second_func", '{"b":1}'), + ] + + @pytest.mark.parametrize( "opening_name", ["my_func", None], @@ -1768,6 +1979,24 @@ def test_accumulate_tool_call_delta_rejects_ambiguous_late_index() -> None: assert buffered_calls == expected_calls +def test_accumulate_tool_call_delta_rejects_ambiguous_idless_late_index() -> None: + unindexed_call = _BufferedToolCall(index=None, call_id="call_1", name="first_func") + indexed_call = _BufferedToolCall(index=0, call_id="call_2", name="second_func") + buffered_calls = {None: unindexed_call, 0: indexed_call} + expected_calls = {None: replace(unindexed_call), 0: replace(indexed_call)} + continuation = ChoiceDeltaToolCall.model_construct( + index=2, + id=None, + type="function", + function=ChoiceDeltaToolCallFunction(arguments="{}"), + ) + + with pytest.raises(ModelBehaviorError, match="new index without an ID"): + ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, continuation) + + assert buffered_calls == expected_calls + + @pytest.mark.asyncio async def test_buffer_tool_call_stream_rejects_ambiguous_missing_index_continuation() -> None: chunks = ( From 9628029f0c7b5dd4c340f1209e7d593d6ebab77a Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Thu, 3 Sep 2026 23:27:44 +0800 Subject: [PATCH 07/11] fix(chat-completions): reject ambiguous streamed tool call ownership --- src/agents/models/chatcmpl_stream_handler.py | 17 +++- .../test_openai_chatcompletions_stream.py | 83 +++++++++++++++++++ 2 files changed, 99 insertions(+), 1 deletion(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 937ab5e795..3dbe613b07 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -551,7 +551,11 @@ async def buffer_tool_call_stream( is_unindexed_untyped_continuation and ( is_passthrough_continuation_by_id - or (saw_unindexed_passthrough_tool_call and not tool_call_delta.id) + or ( + saw_unindexed_passthrough_tool_call + and not tool_call_delta.id + and None not in buffered_calls + ) ) ) if is_passthrough_continuation_by_id and buffered_id_matches: @@ -560,6 +564,17 @@ async def buffer_tool_call_stream( "function call and a passthrough call." ) + if ( + isinstance(tool_call_delta.index, int) + and tool_call_delta.index in passthrough_tool_call_indexes + and tool_call_delta.id + and buffered_id_matches + ): + raise ModelBehaviorError( + "Chat Completions tool call delta supplied an index already used " + "by a passthrough call and an ID owned by a buffered function call." + ) + unindexed_buffered_call = buffered_calls.get(None) if ( isinstance(tool_call_delta.index, int) diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 9f85db983f..306fe1a42e 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -1462,6 +1462,89 @@ async def test_buffer_tool_call_stream_allows_function_metadata_on_late_id() -> assert _completed_function_calls(buffered_events) == expected_calls +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_ambiguous_mixed_unindexed_delta() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "print(1)"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk( + {"tool_calls": [{"extra_content": {"google": {"thought_signature": "sig"}}}]} + ), + ) + + with pytest.raises(ModelBehaviorError, match="could not be attributed"): + await _collect_buffered_tool_call_chunks(*chunks) + + +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_conflicting_index_and_id_owners() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "custom-id", + "type": "custom", + "custom": {"name": "code_exec", "input": "print(1)"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 1, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "extra_content": {"google": {"thought_signature": "sig"}}, + } + ] + } + ), + ) + + buffered = ChatCmplStreamHandler.buffer_tool_call_stream(_completion_stream(*chunks)) + first_chunk = await anext(buffered) + first_tool_calls = first_chunk.choices[0].delta.tool_calls + assert first_tool_calls and first_tool_calls[0].id == "custom-id" + + with pytest.raises(ModelBehaviorError, match="index already used by a passthrough call"): + await anext(buffered) + + @pytest.mark.parametrize("continuation_id", [None, "unknown-id"], ids=["without-id", "unknown-id"]) @pytest.mark.asyncio async def test_buffer_tool_call_stream_rejects_ambiguous_unindexed_passthrough_continuation( From 14dd22b64a0d7d656710931a371c6b872044d24e Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Thu, 3 Sep 2026 23:39:42 +0800 Subject: [PATCH 08/11] fix(chat-completions): promote ID-less passthrough indexes --- src/agents/models/chatcmpl_stream_handler.py | 61 +++++++++++++------ .../test_openai_chatcompletions_stream.py | 51 ++++++++++++++-- 2 files changed, 90 insertions(+), 22 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 3dbe613b07..1d23371830 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -506,7 +506,7 @@ async def buffer_tool_call_stream( buffered_calls: dict[int | None, _BufferedToolCall] = {} passthrough_tool_call_indexes: set[int] = set() passthrough_tool_call_indexes_by_id: dict[str, int | None] = {} - saw_unindexed_passthrough_tool_call = False + unindexed_passthrough_tool_call_count = 0 saw_passthrough_tool_call = False last_chunk: ChatCompletionChunk | None = None @@ -537,6 +537,10 @@ async def buffer_tool_call_stream( is_unindexed_untyped_continuation = ( not isinstance(tool_call_delta.index, int) and is_untyped_continuation ) + passthrough_payload_keys = set(tool_call_delta.model_extra or {}) - { + "provider_specific_fields", + "extra_content", + } buffered_id_matches = [ buffered_call for buffered_call in buffered_calls.values() @@ -547,15 +551,14 @@ async def buffer_tool_call_stream( and bool(tool_call_delta.id) and tool_call_delta.id in passthrough_tool_call_indexes_by_id ) - is_unindexed_passthrough_continuation = ( - is_unindexed_untyped_continuation + is_idless_passthrough_continuation = ( + is_untyped_continuation + and not tool_call_delta.id + and unindexed_passthrough_tool_call_count == 1 + and None not in buffered_calls and ( - is_passthrough_continuation_by_id - or ( - saw_unindexed_passthrough_tool_call - and not tool_call_delta.id - and None not in buffered_calls - ) + not isinstance(tool_call_delta.index, int) + or bool(passthrough_payload_keys) ) ) if is_passthrough_continuation_by_id and buffered_id_matches: @@ -575,6 +578,16 @@ async def buffer_tool_call_stream( "by a passthrough call and an ID owned by a buffered function call." ) + if ( + is_idless_passthrough_continuation + and isinstance(tool_call_delta.index, int) + and tool_call_delta.index in buffered_calls + ): + raise ModelBehaviorError( + "Chat Completions passthrough tool call delta supplied an index " + "already used by a buffered function call." + ) + unindexed_buffered_call = buffered_calls.get(None) if ( isinstance(tool_call_delta.index, int) @@ -600,7 +613,7 @@ async def buffer_tool_call_stream( if ( tool_call_delta.index in passthrough_tool_call_indexes or is_passthrough_continuation_by_id - or is_unindexed_passthrough_continuation + or is_idless_passthrough_continuation ): if passthrough_id := tool_call_delta.id: owner_index = passthrough_tool_call_indexes_by_id.get( @@ -628,6 +641,7 @@ async def buffer_tool_call_stream( tool_call_delta.index ) if promotes_unindexed_passthrough_owner: + unindexed_passthrough_tool_call_count -= 1 tool_call_delta = tool_call_delta.model_copy( update={"type": "custom"} ) @@ -641,6 +655,23 @@ async def buffer_tool_call_stream( if isinstance(tool_call_delta.index, int) else None, ) + elif ( + is_idless_passthrough_continuation + and isinstance(tool_call_delta.index, int) + and tool_call_delta.index not in passthrough_tool_call_indexes + ): + passthrough_tool_call_indexes.add(tool_call_delta.index) + for passthrough_id, owner_index in list( + passthrough_tool_call_indexes_by_id.items() + ): + if owner_index is None: + passthrough_tool_call_indexes_by_id[passthrough_id] = ( + tool_call_delta.index + ) + unindexed_passthrough_tool_call_count = 0 + tool_call_delta = tool_call_delta.model_copy( + update={"type": "custom"} + ) saw_passthrough_tool_call = True remaining_tool_calls.append(tool_call_delta) elif ( @@ -648,13 +679,7 @@ async def buffer_tool_call_stream( and saw_passthrough_tool_call and ( not tool_call_delta.id - or ( - bool( - set(tool_call_delta.model_extra or {}) - - {"provider_specific_fields", "extra_content"} - ) - and not buffered_id_matches - ) + or (bool(passthrough_payload_keys) and not buffered_id_matches) ) ): raise ModelBehaviorError( @@ -668,7 +693,7 @@ async def buffer_tool_call_stream( if isinstance(tool_call_delta.index, int): passthrough_tool_call_indexes.add(tool_call_delta.index) else: - saw_unindexed_passthrough_tool_call = True + unindexed_passthrough_tool_call_count += 1 if tool_call_delta.id: passthrough_tool_call_indexes_by_id.setdefault( tool_call_delta.id, diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 306fe1a42e..d5a73c73b5 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -1206,7 +1206,9 @@ async def test_buffer_tool_call_stream_promotes_indexed_passthrough_continuation @pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_indexed_passthrough_continuation_collision() -> None: +async def test_buffer_tool_call_stream_promotes_indexed_passthrough_continuation_without_id() -> ( + None +): chunks = ( _lenient_chunk( { @@ -1219,11 +1221,13 @@ async def test_buffer_tool_call_stream_rejects_indexed_passthrough_continuation_ ] } ), + _lenient_chunk({"tool_calls": [{"index": 2, "custom": {"input": "nt("}}]}), + _lenient_chunk({"tool_calls": [{"id": "custom-id", "custom": {"input": "1)"}}]}), _lenient_chunk( { "tool_calls": [ { - "index": 2, + "index": 0, "id": "call_1", "type": "function", "function": {"name": "my_func", "arguments": "{}"}, @@ -1231,17 +1235,56 @@ async def test_buffer_tool_call_stream_rejects_indexed_passthrough_continuation_ ] } ), + _lenient_chunk({}, finish_reason="tool_calls"), + ) + + buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) + buffered_events = await _collect_buffered_handler_events(*chunks) + + indexed_continuation = buffered_chunks[1].choices[0].delta.tool_calls + unindexed_continuation = buffered_chunks[2].choices[0].delta.tool_calls + assert indexed_continuation and indexed_continuation[0].index == 2 + assert unindexed_continuation and unindexed_continuation[0].index == 2 + assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + + +@pytest.mark.parametrize("continuation_id", [None, "custom-id"]) +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_indexed_passthrough_continuation_collision( + continuation_id: str | None, +) -> None: + continuation: dict[str, Any] = { + "index": 2, + "custom": {"input": "nt(1)"}, + } + if continuation_id is not None: + continuation["id"] = continuation_id + + chunks = ( _lenient_chunk( { "tool_calls": [ { - "index": 2, "id": "custom-id", - "custom": {"input": "nt(1)"}, + "type": "custom", + "custom": {"name": "code_exec", "input": "pri"}, } ] } ), + _lenient_chunk( + { + "tool_calls": [ + { + "index": 2, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk({"tool_calls": [continuation]}), ) with pytest.raises(ModelBehaviorError, match="index already used by a buffered function"): From 9a2eadddb8e034fd55f30806628a8f16cc7181c3 Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Thu, 3 Sep 2026 23:54:35 +0800 Subject: [PATCH 09/11] fix(chat-completions): reject same-name owner ambiguity --- src/agents/models/chatcmpl_stream_handler.py | 7 +++- .../test_openai_chatcompletions_stream.py | 38 ++++++++++++++++--- 2 files changed, 38 insertions(+), 7 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 1d23371830..552f6e058f 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -397,8 +397,13 @@ def _accumulate_tool_call_delta( elif len(buffered_calls) == 1: sole_index = next(iter(buffered_calls)) sole_name = buffered_calls[sole_index].name - if not sole_name or sole_name == function_name: + if not sole_name: tool_call_index = sole_index + elif sole_name == function_name: + raise ModelBehaviorError( + "Chat Completions tool call delta omitted an index and ID while " + "another buffered call used the same function name." + ) else: tool_call_index = None else: diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index d5a73c73b5..9880ca0d1c 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -1629,14 +1629,9 @@ async def test_buffer_tool_call_stream_rejects_ambiguous_unindexed_passthrough_c await _collect_buffered_tool_call_chunks(*chunks) -@pytest.mark.parametrize("continuation_name", [None, "my_func"]) @pytest.mark.asyncio -async def test_buffer_tool_call_stream_merges_missing_index_continuation( - continuation_name: str | None, -) -> None: +async def test_buffer_tool_call_stream_merges_arguments_only_missing_index_continuation() -> None: continuation_function = {"arguments": "1}"} - if continuation_name is not None: - continuation_function["name"] = continuation_name chunks = ( _lenient_chunk( { @@ -1659,6 +1654,37 @@ async def test_buffer_tool_call_stream_merges_missing_index_continuation( assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] +@pytest.mark.asyncio +async def test_buffer_tool_call_stream_rejects_ambiguous_same_named_unindexed_opening() -> None: + chunks = ( + _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ), + _lenient_chunk( + { + "tool_calls": [ + { + "type": "function", + "function": {"name": "my_func", "arguments": '{"b":'}, + } + ] + } + ), + ) + + with pytest.raises(ModelBehaviorError, match="same function name"): + await _collect_buffered_tool_call_chunks(*chunks) + + @pytest.mark.asyncio async def test_buffer_tool_call_stream_rejects_late_index_used_by_passthrough() -> None: chunks = ( From 0046b18a23f2da5ff2f630dd6d60a21625ea23b6 Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Fri, 4 Sep 2026 00:13:02 +0800 Subject: [PATCH 10/11] refactor(chat-completions): narrow missing-index buffering --- src/agents/models/chatcmpl_stream_handler.py | 338 +---- .../test_openai_chatcompletions_stream.py | 1261 ++--------------- 2 files changed, 189 insertions(+), 1410 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 552f6e058f..a9d0166447 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -157,27 +157,6 @@ class _BufferedToolCall: extra_content: dict[str, Any] | None = None -def _buffered_tool_calls_in_replay_order( - buffered_calls: dict[int | None, _BufferedToolCall], -) -> list[_BufferedToolCall]: - """Sort indexed calls while preserving where the index-less call first appeared.""" - indexed_calls = sorted( - (call for call in buffered_calls.values() if isinstance(call.index, int)), - key=lambda call: cast(int, call.index), - ) - if None not in buffered_calls: - return indexed_calls - - unindexed_position = 0 - for index in buffered_calls: - if index is None: - break - unindexed_position += 1 - - indexed_calls.insert(unindexed_position, buffered_calls[None]) - return indexed_calls - - def _merge_buffered_metadata( current: dict[str, Any] | None, incoming: dict[str, Any], @@ -338,91 +317,62 @@ def _accumulate_tool_call_delta( for index, buffered_call in buffered_calls.items() if tool_call_delta.id and buffered_call.call_id == tool_call_delta.id ] - if isinstance(tool_call_index, int): - if None in matching_indexes: - if len(matching_indexes) > 1: - raise ModelBehaviorError( - "Chat Completions tool call delta supplied an index and the same ID " - "matched multiple buffered calls." - ) - if tool_call_index in buffered_calls: - raise ModelBehaviorError( - "Chat Completions tool call delta supplied an index already used by " - "another buffered call." - ) - buffered_call = buffered_calls.pop(None) - buffered_call.index = tool_call_index - buffered_calls[tool_call_index] = buffered_call - elif ( - not tool_call_delta.id - and None in buffered_calls - and tool_call_index not in buffered_calls - ): - buffered_name = buffered_calls[None].name - if not function_name or not buffered_name or function_name == buffered_name: - if len(buffered_calls) > 1: - raise ModelBehaviorError( - "Chat Completions tool call delta supplied a new index without an ID " - "while multiple function calls were being buffered." - ) - buffered_call = buffered_calls.pop(None) - buffered_call.index = tool_call_index - buffered_calls[tool_call_index] = buffered_call - else: - if len(matching_indexes) == 1: - tool_call_index = matching_indexes[0] - elif len(matching_indexes) > 1: + if len(matching_indexes) > 1: + raise ModelBehaviorError( + "Chat Completions tool call delta matched multiple buffered calls." + ) + + if isinstance(tool_call_index, int): + if matching_indexes and matching_indexes[0] != tool_call_index: raise ModelBehaviorError( - "Chat Completions tool call delta omitted an index and the same ID " - "matched multiple buffered calls." + "Chat Completions tool call index and ID identified different buffered calls." ) - elif tool_call_delta.id and None in buffered_calls and buffered_calls[None].call_id: + elif matching_indexes: + tool_call_index = matching_indexes[0] + elif tool_call_delta.id: + if None in buffered_calls and buffered_calls[None].call_id: raise ModelBehaviorError( - "Chat Completions tool call delta omitted an index with a new ID while " + "Chat Completions tool call delta omitted an index for a new call while " "another index-less call was being buffered." ) - else: - if function_name and None in buffered_calls: - buffered_name = buffered_calls[None].name - if buffered_name and buffered_name != function_name: - raise ModelBehaviorError( - "Chat Completions tool call delta omitted an index and used a " - "different function name from the buffered index-less call." - ) - - if not tool_call_delta.id and function_name: - if None in buffered_calls: - tool_call_index = None - elif len(buffered_calls) == 1: - sole_index = next(iter(buffered_calls)) - sole_name = buffered_calls[sole_index].name - if not sole_name: - tool_call_index = sole_index - elif sole_name == function_name: - raise ModelBehaviorError( - "Chat Completions tool call delta omitted an index and ID while " - "another buffered call used the same function name." - ) - else: - tool_call_index = None - else: - tool_call_index = None - elif not tool_call_delta.id and None in buffered_calls: - tool_call_index = None - elif not tool_call_delta.id and len(buffered_calls) == 1: - tool_call_index = next(iter(buffered_calls)) - elif not tool_call_delta.id and len(buffered_calls) > 1: + tool_call_index = None + elif function_name: + if None in buffered_calls: + if buffered_calls[None].name: raise ModelBehaviorError( - "Chat Completions tool call delta omitted an index while multiple " - "function tool calls were being buffered." + "Chat Completions tool call delta omitted both index and ID, so its " + "function-call owner could not be attributed safely." ) + elif any(call.name == function_name for call in buffered_calls.values()): + raise ModelBehaviorError( + "Chat Completions tool call delta omitted both index and ID, so its " + "function-call owner could not be attributed safely." + ) + else: + tool_call_index = None + elif len(buffered_calls) == 1: + tool_call_index = next(iter(buffered_calls)) + elif len(buffered_calls) > 1: + raise ModelBehaviorError( + "Chat Completions tool call delta omitted both index and ID while multiple " + "function calls were being buffered." + ) buffered_call = buffered_calls.setdefault( tool_call_index, _BufferedToolCall(index=tool_call_index), ) + if tool_call_delta.id and buffered_call.call_id not in (None, tool_call_delta.id): + raise ModelBehaviorError( + "Chat Completions tool call delta changed the ID of a buffered call." + ) + if function_name and buffered_call.name not in (None, function_name): + raise ModelBehaviorError( + "Chat Completions tool call delta changed the name of a buffered function call." + ) + if tool_call_delta.id: buffered_call.call_id = tool_call_delta.id @@ -449,6 +399,7 @@ def _accumulate_tool_call_delta( @staticmethod def _buffered_tool_call_delta( buffered_call: _BufferedToolCall, + *, fallback_index: int = 0, ) -> ChoiceDeltaToolCall: if not buffered_call.call_id: @@ -486,11 +437,17 @@ def _buffered_tool_calls_chunk( buffered_calls: dict[int | None, _BufferedToolCall], passthrough_tool_call_indexes: set[int], ) -> ChatCompletionChunk: - ordered_calls = _buffered_tool_calls_in_replay_order(buffered_calls) - occupied_indexes = passthrough_tool_call_indexes | { + ordered_calls = sorted( + buffered_calls.values(), + key=lambda call: ( + not isinstance(call.index, int), + call.index if isinstance(call.index, int) else 0, + ), + ) + used_indexes = passthrough_tool_call_indexes | { call.index for call in ordered_calls if isinstance(call.index, int) } - fallback_index = max(occupied_indexes, default=-1) + 1 + fallback_index = max(used_indexes, default=-1) + 1 tool_call_deltas = [ cls._buffered_tool_call_delta(buffered_call, fallback_index=fallback_index) for buffered_call in ordered_calls @@ -510,8 +467,6 @@ async def buffer_tool_call_stream( """Buffer streamed function tool-call deltas until they are complete.""" buffered_calls: dict[int | None, _BufferedToolCall] = {} passthrough_tool_call_indexes: set[int] = set() - passthrough_tool_call_indexes_by_id: dict[str, int | None] = {} - unindexed_passthrough_tool_call_count = 0 saw_passthrough_tool_call = False last_chunk: ChatCompletionChunk | None = None @@ -525,8 +480,6 @@ async def buffer_tool_call_stream( passthrough_choices: list[Choice] = [] for choice in chunk.choices: if choice.index != 0: - if choice.delta and choice.delta.tool_calls: - saw_passthrough_tool_call = True passthrough_choices.append(choice) continue @@ -535,177 +488,40 @@ async def buffer_tool_call_stream( if tool_call_deltas := (delta.tool_calls if delta and delta.tool_calls else None): remaining_tool_calls: list[ChoiceDeltaToolCall] = [] for tool_call_delta in tool_call_deltas: - is_untyped_continuation = ( - getattr(tool_call_delta, "type", None) is None - and tool_call_delta.function is None - ) - is_unindexed_untyped_continuation = ( - not isinstance(tool_call_delta.index, int) and is_untyped_continuation - ) - passthrough_payload_keys = set(tool_call_delta.model_extra or {}) - { - "provider_specific_fields", - "extra_content", - } - buffered_id_matches = [ - buffered_call - for buffered_call in buffered_calls.values() - if tool_call_delta.id and buffered_call.call_id == tool_call_delta.id - ] - is_passthrough_continuation_by_id = ( - is_untyped_continuation - and bool(tool_call_delta.id) - and tool_call_delta.id in passthrough_tool_call_indexes_by_id - ) - is_idless_passthrough_continuation = ( - is_untyped_continuation - and not tool_call_delta.id - and unindexed_passthrough_tool_call_count == 1 - and None not in buffered_calls - and ( - not isinstance(tool_call_delta.index, int) - or bool(passthrough_payload_keys) - ) - ) - if is_passthrough_continuation_by_id and buffered_id_matches: - raise ModelBehaviorError( - "Chat Completions tool call delta ID matched both a buffered " - "function call and a passthrough call." - ) - - if ( - isinstance(tool_call_delta.index, int) - and tool_call_delta.index in passthrough_tool_call_indexes - and tool_call_delta.id - and buffered_id_matches - ): - raise ModelBehaviorError( - "Chat Completions tool call delta supplied an index already used " - "by a passthrough call and an ID owned by a buffered function call." - ) - - if ( - is_idless_passthrough_continuation - and isinstance(tool_call_delta.index, int) - and tool_call_delta.index in buffered_calls - ): - raise ModelBehaviorError( - "Chat Completions passthrough tool call delta supplied an index " - "already used by a buffered function call." - ) - - unindexed_buffered_call = buffered_calls.get(None) - if ( - isinstance(tool_call_delta.index, int) - and tool_call_delta.index in passthrough_tool_call_indexes - and unindexed_buffered_call is not None - and ( - ( - tool_call_delta.id - and unindexed_buffered_call.call_id == tool_call_delta.id - ) - or ( - not tool_call_delta.id - and len(buffered_calls) == 1 - and tool_call_delta.function is not None - ) - ) - ): - raise ModelBehaviorError( - "Chat Completions tool call delta supplied an index already used " - "by a passthrough call." - ) - - if ( - tool_call_delta.index in passthrough_tool_call_indexes - or is_passthrough_continuation_by_id - or is_idless_passthrough_continuation - ): - if passthrough_id := tool_call_delta.id: - owner_index = passthrough_tool_call_indexes_by_id.get( - passthrough_id - ) - if isinstance(tool_call_delta.index, int): - promotes_unindexed_passthrough_owner = ( - is_passthrough_continuation_by_id - and not isinstance(owner_index, int) - ) - if isinstance(owner_index, int) and ( - owner_index != tool_call_delta.index - ): - raise ModelBehaviorError( - "Chat Completions passthrough tool call delta supplied " - "a different index from its buffered ID owner." - ) - if tool_call_delta.index in buffered_calls: - raise ModelBehaviorError( - "Chat Completions passthrough tool call delta supplied " - "an index already used by a buffered function call." - ) - passthrough_tool_call_indexes.add(tool_call_delta.index) - passthrough_tool_call_indexes_by_id[passthrough_id] = ( - tool_call_delta.index - ) - if promotes_unindexed_passthrough_owner: - unindexed_passthrough_tool_call_count -= 1 - tool_call_delta = tool_call_delta.model_copy( - update={"type": "custom"} - ) - elif isinstance(owner_index, int): - tool_call_delta = tool_call_delta.model_copy( - update={"index": owner_index} - ) - passthrough_tool_call_indexes_by_id.setdefault( - passthrough_id, - tool_call_delta.index - if isinstance(tool_call_delta.index, int) - else None, - ) - elif ( - is_idless_passthrough_continuation - and isinstance(tool_call_delta.index, int) - and tool_call_delta.index not in passthrough_tool_call_indexes + if tool_call_delta.index in passthrough_tool_call_indexes: + if tool_call_delta.id and any( + buffered_call.call_id == tool_call_delta.id + for buffered_call in buffered_calls.values() ): - passthrough_tool_call_indexes.add(tool_call_delta.index) - for passthrough_id, owner_index in list( - passthrough_tool_call_indexes_by_id.items() - ): - if owner_index is None: - passthrough_tool_call_indexes_by_id[passthrough_id] = ( - tool_call_delta.index - ) - unindexed_passthrough_tool_call_count = 0 - tool_call_delta = tool_call_delta.model_copy( - update={"type": "custom"} + raise ModelBehaviorError( + "Chat Completions tool call index and ID identified different " + "tool call owners." ) saw_passthrough_tool_call = True remaining_tool_calls.append(tool_call_delta) - elif ( - is_unindexed_untyped_continuation - and saw_passthrough_tool_call - and ( - not tool_call_delta.id - or (bool(passthrough_payload_keys) and not buffered_id_matches) - ) - ): - raise ModelBehaviorError( - "Chat Completions tool call delta omitted an index, type, and " - "function payload after a passthrough call, so it could not be " - "attributed safely." - ) elif cls._should_buffer_tool_call_delta(tool_call_delta): + if ( + saw_passthrough_tool_call + and tool_call_delta.index is None + and tool_call_delta.id is None + and tool_call_delta.function is None + ): + raise ModelBehaviorError( + "Chat Completions tool call delta could not be attributed " + "safely between buffered and passthrough calls." + ) cls._accumulate_tool_call_delta(buffered_calls, tool_call_delta) else: + if ( + isinstance(tool_call_delta.index, int) + and tool_call_delta.index in buffered_calls + ): + raise ModelBehaviorError( + "Chat Completions tool call index identified both a buffered " + "function call and a passthrough call." + ) if isinstance(tool_call_delta.index, int): passthrough_tool_call_indexes.add(tool_call_delta.index) - else: - unindexed_passthrough_tool_call_count += 1 - if tool_call_delta.id: - passthrough_tool_call_indexes_by_id.setdefault( - tool_call_delta.id, - tool_call_delta.index - if isinstance(tool_call_delta.index, int) - else None, - ) saw_passthrough_tool_call = True remaining_tool_calls.append(tool_call_delta) diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 9880ca0d1c..a60b02df18 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -1,7 +1,6 @@ import asyncio import logging from collections.abc import AsyncIterator -from dataclasses import replace from typing import Any, cast import httpx2 @@ -120,9 +119,7 @@ async def _collect_buffered_handler_events(*chunks: ChatCompletionChunk) -> list def _lenient_chunk(delta: dict[str, Any], finish_reason: str | None = None) -> ChatCompletionChunk: - # Build the chunk the way ``AsyncStream`` does: ``construct_type`` does not validate the - # provider payload, so a tool call delta that omits ``index`` keeps ``index=None`` instead - # of failing validation. That is the shape OpenAI-compatible providers can produce. + """Construct a chunk as AsyncStream does, without validating a missing tool-call index.""" return cast( ChatCompletionChunk, construct_type( @@ -932,7 +929,6 @@ async def test_buffer_tool_call_stream_orders_missing_index_after_indexed_calls( chunks = ( _lenient_chunk( { - "role": "assistant", "tool_calls": [ { "index": 0, @@ -940,77 +936,13 @@ async def test_buffer_tool_call_stream_orders_missing_index_after_indexed_calls( "type": "function", "function": {"name": "first_func", "arguments": "{}"}, } - ], - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_2", - "type": "function", - "function": {"name": "second_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - unbuffered_events = await _collect_handler_events(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - expected_calls = [("call_1", "first_func", "{}"), ("call_2", "second_func", "{}")] - assert _completed_function_calls(unbuffered_events) == expected_calls - assert _completed_function_calls(buffered_events) == expected_calls - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_preserves_missing_index_call_arrival_order() -> None: - chunks = ( - _lenient_chunk( - { - "role": "assistant", - "tool_calls": [ - { - "id": "call_1", - "type": "function", - "function": {"name": "first_func", "arguments": "{}"}, - } - ], - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_2", - "type": "function", - "function": {"name": "second_func", "arguments": "{}"}, - } ] } ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - unbuffered_events = await _collect_handler_events(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - expected_calls = [("call_1", "first_func", "{}"), ("call_2", "second_func", "{}")] - assert _completed_function_calls(unbuffered_events) == expected_calls - assert _completed_function_calls(buffered_events) == expected_calls - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_sorts_indexed_calls_by_index() -> None: - chunks = ( _lenient_chunk( { "tool_calls": [ { - "index": 1, "id": "call_2", "type": "function", "function": {"name": "second_func", "arguments": "{}"}, @@ -1018,18 +950,6 @@ async def test_buffer_tool_call_stream_sorts_indexed_calls_by_index() -> None: ] } ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_1", - "type": "function", - "function": {"name": "first_func", "arguments": "{}"}, - } - ] - } - ), _lenient_chunk({}, finish_reason="tool_calls"), ) @@ -1042,21 +962,9 @@ async def test_buffer_tool_call_stream_sorts_indexed_calls_by_index() -> None: @pytest.mark.asyncio -async def test_buffer_tool_call_stream_avoids_passthrough_index_for_missing_index() -> None: - custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( - index=0, - id="custom-id", - type="custom", - ) - custom_chunk = ChatCompletionChunk( - id="chunk-id", - created=1, - model="fake", - object="chat.completion.chunk", - choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[custom_tool_call_delta]))], - ) +async def test_missing_index_replay_avoids_passthrough_index_collision() -> None: chunks = ( - custom_chunk, + _lenient_chunk({"tool_calls": [{"index": 0, "id": "custom-id", "type": "custom"}]}), _lenient_chunk( { "tool_calls": [ @@ -1068,1115 +976,170 @@ async def test_buffer_tool_call_stream_avoids_passthrough_index_for_missing_inde ] } ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - buffered_events = await _collect_buffered_handler_events(*chunks) - - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] - - -@pytest.mark.parametrize("function_tool_call_index", [0, None]) -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_ignores_missing_passthrough_index( - function_tool_call_index: int | None, -) -> None: - custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( - index=None, - id="custom-id", - type="custom", - ) - custom_chunk = ChatCompletionChunk( - id="chunk-id", - created=1, - model="fake", - object="chat.completion.chunk", - choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[custom_tool_call_delta]))], - ) - function_tool_call: dict[str, Any] = { - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - if function_tool_call_index is not None: - function_tool_call["index"] = function_tool_call_index - chunks = ( - custom_chunk, - _lenient_chunk({"tool_calls": [function_tool_call]}), - _lenient_chunk({}, finish_reason="tool_calls"), ) - buffered_events = await _collect_buffered_handler_events(*chunks) + buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + assert buffered_chunks[0].choices[0].delta.tool_calls == chunks[0].choices[0].delta.tool_calls + replayed_calls = buffered_chunks[-1].choices[0].delta.tool_calls + assert replayed_calls and replayed_calls[0].index == 1 + assert replayed_calls[0].id == "call_1" -@pytest.mark.parametrize("continuation_id", [None, "custom-id"]) @pytest.mark.asyncio -async def test_buffer_tool_call_stream_forwards_missing_index_passthrough_continuation( - continuation_id: str | None, -) -> None: - continuation: dict[str, Any] = {"custom": {"input": "nt(1)"}} - if continuation_id is not None: - continuation["id"] = continuation_id +async def test_missing_index_continuation_merges_into_sole_buffered_function() -> None: chunks = ( _lenient_chunk( { "tool_calls": [ { - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "pri"}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [continuation]}), - _lenient_chunk( - { - "tool_calls": [ - { + "index": 2, "id": "call_1", "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, + "function": {"name": "my_func", "arguments": '{"a":'}, } ] } ), - _lenient_chunk({}, finish_reason="tool_calls"), + _lenient_chunk({"tool_calls": [{"function": {"arguments": "1}"}}]}), ) buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) + replayed_calls = buffered_chunks[-1].choices[0].delta.tool_calls - assert buffered_chunks[0].choices[0].delta.tool_calls == chunks[0].choices[0].delta.tool_calls - assert buffered_chunks[1].choices[0].delta.tool_calls == chunks[1].choices[0].delta.tool_calls - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + assert replayed_calls and replayed_calls[0].index == 2 + assert replayed_calls[0].function + assert replayed_calls[0].function.arguments == '{"a":1}' @pytest.mark.asyncio -async def test_buffer_tool_call_stream_promotes_indexed_passthrough_continuation_by_id() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "pri"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 2, - "id": "custom-id", - "custom": {"input": "nt("}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [{"id": "custom-id", "custom": {"input": "1)"}}]}), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk({}, finish_reason="tool_calls"), +async def test_missing_index_continuation_uses_a_unique_function_call_id() -> None: + opening = _lenient_chunk( + { + "tool_calls": [ + { + "index": index, + "id": f"call_{index}", + "type": "function", + "function": {"name": f"func_{index}", "arguments": "start"}, + } + for index in range(2) + ] + } + ) + continuation = _lenient_chunk( + {"tool_calls": [{"id": "call_1", "function": {"arguments": "-end"}}]} ) - buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) + buffered_chunks = await _collect_buffered_tool_call_chunks(opening, continuation) + replayed_calls = buffered_chunks[-1].choices[0].delta.tool_calls - indexed_continuation = buffered_chunks[1].choices[0].delta.tool_calls - unindexed_continuation = buffered_chunks[2].choices[0].delta.tool_calls - assert indexed_continuation and indexed_continuation[0].index == 2 - assert unindexed_continuation and unindexed_continuation[0].index == 2 - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + assert replayed_calls and [ + call.function.arguments for call in replayed_calls if call.function + ] == [ + "start", + "start-end", + ] @pytest.mark.asyncio -async def test_buffer_tool_call_stream_promotes_indexed_passthrough_continuation_without_id() -> ( - None -): +async def test_missing_index_continuation_rejects_multiple_buffered_functions() -> None: chunks = ( _lenient_chunk( { "tool_calls": [ { - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "pri"}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [{"index": 2, "custom": {"input": "nt("}}]}), - _lenient_chunk({"tool_calls": [{"id": "custom-id", "custom": {"input": "1)"}}]}), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_1", + "index": index, + "id": f"call_{index}", "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, + "function": {"name": f"func_{index}", "arguments": "{}"}, } ] } - ), - _lenient_chunk({}, finish_reason="tool_calls"), + ) + for index in range(2) ) + continuation = _lenient_chunk({"tool_calls": [{"function": {"arguments": "tail"}}]}) - buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - indexed_continuation = buffered_chunks[1].choices[0].delta.tool_calls - unindexed_continuation = buffered_chunks[2].choices[0].delta.tool_calls - assert indexed_continuation and indexed_continuation[0].index == 2 - assert unindexed_continuation and unindexed_continuation[0].index == 2 - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] + with pytest.raises(ModelBehaviorError, match="multiple function calls"): + await _collect_buffered_tool_call_chunks(*chunks, continuation) -@pytest.mark.parametrize("continuation_id", [None, "custom-id"]) @pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_indexed_passthrough_continuation_collision( - continuation_id: str | None, -) -> None: - continuation: dict[str, Any] = { - "index": 2, - "custom": {"input": "nt(1)"}, - } - if continuation_id is not None: - continuation["id"] = continuation_id - - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "pri"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 2, - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [continuation]}), +async def test_anonymous_delta_rejects_mixed_unindexed_call_owners() -> None: + custom = _lenient_chunk( + {"tool_calls": [{"id": "custom-id", "type": "custom", "custom": {"input": "x"}}]} + ) + function = _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ) + anonymous = _lenient_chunk( + {"tool_calls": [{"extra_content": {"google": {"thought_signature": "sig"}}}]} ) - with pytest.raises(ModelBehaviorError, match="index already used by a buffered function"): - await _collect_buffered_tool_call_chunks(*chunks) + with pytest.raises(ModelBehaviorError, match="could not be attributed safely"): + await _collect_buffered_tool_call_chunks(custom, function, anonymous) @pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_passthrough_continuation_index_change() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 1, - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "pri"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 2, - "id": "custom-id", - "custom": {"input": "nt(1)"}, - } - ] - } - ), +async def test_tool_call_index_and_id_owner_conflict_fails_closed() -> None: + custom = _lenient_chunk({"tool_calls": [{"index": 0, "id": "custom-id", "type": "custom"}]}) + function = _lenient_chunk( + { + "tool_calls": [ + { + "index": 1, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ) + conflicting = _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "extra_content": {"google": {"thought_signature": "sig"}}, + } + ] + } ) - with pytest.raises(ModelBehaviorError, match="different index from its buffered ID owner"): - await _collect_buffered_tool_call_chunks(*chunks) + with pytest.raises(ModelBehaviorError, match="different tool call owners"): + await _collect_buffered_tool_call_chunks(custom, function, conflicting) -@pytest.mark.parametrize("function_tool_call_index", [1, None], ids=["indexed", "unindexed"]) @pytest.mark.asyncio -async def test_buffer_tool_call_stream_forwards_indexed_passthrough_continuation_by_id( - function_tool_call_index: int | None, -) -> None: - function_tool_call: dict[str, Any] = { - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - if function_tool_call_index is not None: - function_tool_call["index"] = function_tool_call_index - function_tool_call["id"] = "call_1" +async def test_same_name_is_not_used_to_infer_a_missing_index_owner() -> None: + indexed = _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "same_func", "arguments": "{}"}, + } + ] + } + ) + unindexed = _lenient_chunk( + {"tool_calls": [{"type": "function", "function": {"name": "same_func"}}]} + ) - chunks = [ - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "pri"}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [function_tool_call]}), - _lenient_chunk({"tool_calls": [{"id": "custom-id", "custom": {"input": "nt(1)"}}]}), - ] - if function_tool_call_index is None: - chunks.append(_lenient_chunk({"tool_calls": [{"id": "call_1"}]})) - chunks.append(_lenient_chunk({}, finish_reason="tool_calls")) - - buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - continuation_tool_calls = buffered_chunks[1].choices[0].delta.tool_calls - assert continuation_tool_calls - assert continuation_tool_calls[0].index == 0 - assert continuation_tool_calls[0].id == "custom-id" - assert continuation_tool_calls[0].model_extra == {"custom": {"input": "nt(1)"}} - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_records_passthrough_id_from_indexed_continuation() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "type": "custom", - "custom": {"name": "code_exec", "input": "pri"}, - } - ] - } - ), - _lenient_chunk( - {"tool_calls": [{"index": 0, "id": "custom-id", "custom": {"input": "nt("}}]} - ), - _lenient_chunk( - { - "tool_calls": [ - { - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [{"id": "custom-id", "custom": {"input": "1)"}}]}), - _lenient_chunk({"tool_calls": [{"id": "call_1"}]}), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - continuation_tool_calls = buffered_chunks[2].choices[0].delta.tool_calls - assert continuation_tool_calls - assert continuation_tool_calls[0].index == 0 - assert continuation_tool_calls[0].id == "custom-id" - assert continuation_tool_calls[0].model_extra == {"custom": {"input": "1)"}} - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", "{}")] - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_cross_domain_passthrough_id() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "shared-id", - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 1, - "id": "shared-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "print(1)"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "id": "shared-id", - "extra_content": {"google": {"thought_signature": "sig"}}, - } - ] - } - ), - ) - - with pytest.raises(ModelBehaviorError, match="matched both"): - await _collect_buffered_tool_call_chunks(*chunks) - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_allows_function_metadata_on_late_id() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "print(1)"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_1", - "extra_content": {"google": {"thought_signature": "sig"}}, - } - ] - } - ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) - unbuffered_events = await _collect_handler_events(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - replayed_tool_calls = buffered_chunks[-1].choices[0].delta.tool_calls - assert replayed_tool_calls - assert cast(Any, replayed_tool_calls[0]).extra_content == { - "google": {"thought_signature": "sig"} - } - expected_calls = [("call_1", "my_func", "{}")] - assert _completed_function_calls(unbuffered_events) == expected_calls - assert _completed_function_calls(buffered_events) == expected_calls - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_ambiguous_mixed_unindexed_delta() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "print(1)"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk( - {"tool_calls": [{"extra_content": {"google": {"thought_signature": "sig"}}}]} - ), - ) - - with pytest.raises(ModelBehaviorError, match="could not be attributed"): - await _collect_buffered_tool_call_chunks(*chunks) - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_conflicting_index_and_id_owners() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "print(1)"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 1, - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_1", - "extra_content": {"google": {"thought_signature": "sig"}}, - } - ] - } - ), - ) - - buffered = ChatCmplStreamHandler.buffer_tool_call_stream(_completion_stream(*chunks)) - first_chunk = await anext(buffered) - first_tool_calls = first_chunk.choices[0].delta.tool_calls - assert first_tool_calls and first_tool_calls[0].id == "custom-id" - - with pytest.raises(ModelBehaviorError, match="index already used by a passthrough call"): - await anext(buffered) - - -@pytest.mark.parametrize("continuation_id", [None, "unknown-id"], ids=["without-id", "unknown-id"]) -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_ambiguous_unindexed_passthrough_continuation( - continuation_id: str | None, -) -> None: - continuation: dict[str, Any] = {"custom": {"input": "nt(1)"}} - if continuation_id is not None: - continuation["id"] = continuation_id - - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "pri"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 1, - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [continuation]}), - ) - - with pytest.raises(ModelBehaviorError, match="could not be attributed"): - await _collect_buffered_tool_call_chunks(*chunks) - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_merges_arguments_only_missing_index_continuation() -> None: - continuation_function = {"arguments": "1}"} - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": '{"a":'}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [{"function": continuation_function}]}), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - buffered_events = await _collect_buffered_handler_events(*chunks) - - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_ambiguous_same_named_unindexed_opening() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "type": "function", - "function": {"name": "my_func", "arguments": '{"b":'}, - } - ] - } - ), - ) - - with pytest.raises(ModelBehaviorError, match="same function name"): - await _collect_buffered_tool_call_chunks(*chunks) - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_late_index_used_by_passthrough() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 2, - "id": "custom-id", - "type": "custom", - "custom": {"name": "code_exec", "input": "print(1)"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": '{"a":'}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 2, - "id": "call_1", - "type": "function", - "function": {"arguments": "1}"}, - } - ] - } - ), - ) - - with pytest.raises(ModelBehaviorError, match="index already used by a passthrough call"): - await _collect_buffered_tool_call_chunks(*chunks) - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_merges_late_id_into_missing_index_call() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "type": "function", - "function": {"name": "my_func", "arguments": '{"a":'}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_1", - "type": "function", - "function": {"arguments": "1}"}, - } - ] - } - ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - unbuffered_events = await _collect_handler_events(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - expected_calls = [("call_1", "my_func", '{"a":1}')] - assert _completed_function_calls(unbuffered_events) == expected_calls - assert _completed_function_calls(buffered_events) == expected_calls - - -@pytest.mark.parametrize("continuation_name", [None, "my_func"]) -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_reconciles_late_index_by_id( - continuation_name: str | None, -) -> None: - continuation_function = {"arguments": "1}"} - if continuation_name is not None: - continuation_function["name"] = continuation_name - - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": '{"a":'}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 2, - "id": "call_1", - "type": "function", - "function": continuation_function, - } - ] - } - ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - replayed_tool_calls = buffered_chunks[-1].choices[0].delta.tool_calls - assert replayed_tool_calls - assert [tool_call.index for tool_call in replayed_tool_calls] == [2] - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_reconciles_late_index_without_id_for_sole_call() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": '{"a":'}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 2, - "type": "function", - "function": {"arguments": "1}"}, - } - ] - } - ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - buffered_chunks = await _collect_buffered_tool_call_chunks(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - replayed_tool_calls = buffered_chunks[-1].choices[0].delta.tool_calls - assert replayed_tool_calls - assert [tool_call.index for tool_call in replayed_tool_calls] == [2] - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_keeps_different_named_idless_late_index_distinct() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_1", - "type": "function", - "function": {"name": "first_func", "arguments": '{"a":1}'}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 2, - "type": "function", - "function": {"name": "second_func", "arguments": '{"b":'}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "index": 2, - "id": "call_2", - "type": "function", - "function": {"arguments": "1}"}, - } - ] - } - ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - buffered_events = await _collect_buffered_handler_events(*chunks) - - assert _completed_function_calls(buffered_events) == [ - ("call_1", "first_func", '{"a":1}'), - ("call_2", "second_func", '{"b":1}'), - ] - - -@pytest.mark.parametrize( - "opening_name", - ["my_func", None], - ids=["same-name", "fills-missing-name"], -) -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_merges_named_continuation_into_missing_index_slot( - opening_name: str | None, -) -> None: - opening_function = {"arguments": '{"a":'} - if opening_name is not None: - opening_function["name"] = opening_name - - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "type": "function", - "function": opening_function, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "type": "function", - "function": {"name": "my_func", "arguments": "1}"}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [{"id": "call_1"}]}), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - unbuffered_events = await _collect_handler_events(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - expected_calls = [("call_1", "my_func", '{"a":1}')] - assert _completed_function_calls(unbuffered_events) == expected_calls - assert _completed_function_calls(buffered_events) == expected_calls - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_opens_named_missing_index_call_beside_indexed_call() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_1", - "type": "function", - "function": {"name": "first_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "type": "function", - "function": {"name": "second_func", "arguments": '{"b":'}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_2", - "type": "function", - "function": {"arguments": "1}"}, - } - ] - } - ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - unbuffered_events = await _collect_handler_events(*chunks) - buffered_events = await _collect_buffered_handler_events(*chunks) - - expected_calls = [ - ("call_1", "first_func", "{}"), - ("call_2", "second_func", '{"b":1}'), - ] - assert _completed_function_calls(unbuffered_events) == expected_calls - assert _completed_function_calls(buffered_events) == expected_calls - - -@pytest.mark.parametrize("continuation_id", [None, "call_2"], ids=["without-id", "with-new-id"]) -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_different_name_for_missing_index_slot( - continuation_id: str | None, -) -> None: - continuation: dict[str, Any] = { - "type": "function", - "function": {"name": "second_func", "arguments": "{}"}, - } - if continuation_id is not None: - continuation["id"] = continuation_id - - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "type": "function", - "function": {"name": "first_func", "arguments": "{}"}, - } - ] - } - ), - _lenient_chunk({"tool_calls": [continuation]}), - ) - - with pytest.raises(ModelBehaviorError, match="different function name"): - await _collect_buffered_handler_events(*chunks) - - -@pytest.mark.parametrize( - "continuation_function", - [ - {"arguments": "1}"}, - {"name": "my_func", "arguments": "1}"}, - ], - ids=["arguments-only", "repeated-name"], -) -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_merges_missing_index_continuation_by_id( - continuation_function: dict[str, str], -) -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_1", - "type": "function", - "function": {"name": "my_func", "arguments": '{"a":'}, - } - ] - } - ), - _lenient_chunk( - { - "tool_calls": [ - { - "id": "call_1", - "type": "function", - "function": continuation_function, - } - ] - } - ), - _lenient_chunk({}, finish_reason="tool_calls"), - ) - - buffered_events = await _collect_buffered_handler_events(*chunks) - - assert _completed_function_calls(buffered_events) == [("call_1", "my_func", '{"a":1}')] - - -def test_accumulate_tool_call_delta_rejects_ambiguous_repeated_id() -> None: - buffered_calls = { - 0: _BufferedToolCall(index=0, call_id="call_1", name="first_func"), - 1: _BufferedToolCall(index=1, call_id="call_1", name="second_func"), - } - continuation = ChoiceDeltaToolCall.model_construct( - index=None, - id="call_1", - type="function", - function=ChoiceDeltaToolCallFunction(arguments="{}"), - ) - - with pytest.raises(ModelBehaviorError, match="same ID matched multiple buffered calls"): - ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, continuation) - - -def test_accumulate_tool_call_delta_rejects_new_id_for_occupied_missing_index() -> None: - buffered_calls = { - None: _BufferedToolCall(index=None, call_id="call_1", name="first_func"), - } - new_call = ChoiceDeltaToolCall.model_construct( - index=None, - id="call_2", - type="function", - function=ChoiceDeltaToolCallFunction(name="second_func", arguments="{}"), - ) - - with pytest.raises(ModelBehaviorError, match="new ID while another index-less call"): - ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, new_call) - - -def test_accumulate_tool_call_delta_rejects_late_index_collision() -> None: - unindexed_call = _BufferedToolCall(index=None, call_id="call_1", name="first_func") - indexed_call = _BufferedToolCall(index=2, call_id="call_2", name="second_func") - buffered_calls = {None: unindexed_call, 2: indexed_call} - expected_calls = {None: replace(unindexed_call), 2: replace(indexed_call)} - continuation = ChoiceDeltaToolCall.model_construct( - index=2, - id="call_1", - type="function", - function=ChoiceDeltaToolCallFunction(arguments="{}"), - ) - - with pytest.raises(ModelBehaviorError, match="index already used"): - ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, continuation) - - assert buffered_calls == expected_calls - - -def test_accumulate_tool_call_delta_rejects_ambiguous_late_index() -> None: - unindexed_call = _BufferedToolCall(index=None, call_id="call_1", name="first_func") - indexed_call = _BufferedToolCall(index=1, call_id="call_1", name="second_func") - buffered_calls = {None: unindexed_call, 1: indexed_call} - expected_calls = {None: replace(unindexed_call), 1: replace(indexed_call)} - continuation = ChoiceDeltaToolCall.model_construct( - index=2, - id="call_1", - type="function", - function=ChoiceDeltaToolCallFunction(arguments="{}"), - ) - - with pytest.raises(ModelBehaviorError, match="same ID matched multiple buffered calls"): - ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, continuation) - - assert buffered_calls == expected_calls - - -def test_accumulate_tool_call_delta_rejects_ambiguous_idless_late_index() -> None: - unindexed_call = _BufferedToolCall(index=None, call_id="call_1", name="first_func") - indexed_call = _BufferedToolCall(index=0, call_id="call_2", name="second_func") - buffered_calls = {None: unindexed_call, 0: indexed_call} - expected_calls = {None: replace(unindexed_call), 0: replace(indexed_call)} - continuation = ChoiceDeltaToolCall.model_construct( - index=2, - id=None, - type="function", - function=ChoiceDeltaToolCallFunction(arguments="{}"), - ) - - with pytest.raises(ModelBehaviorError, match="new index without an ID"): - ChatCmplStreamHandler._accumulate_tool_call_delta(buffered_calls, continuation) - - assert buffered_calls == expected_calls - - -@pytest.mark.asyncio -async def test_buffer_tool_call_stream_rejects_ambiguous_missing_index_continuation() -> None: - chunks = ( - _lenient_chunk( - { - "tool_calls": [ - { - "index": 0, - "id": "call_1", - "type": "function", - "function": {"name": "first_func", "arguments": '{"a":'}, - }, - { - "index": 1, - "id": "call_2", - "type": "function", - "function": {"name": "second_func", "arguments": '{"b":'}, - }, - ] - } - ), - _lenient_chunk({"tool_calls": [{"function": {"arguments": "1}"}}]}), - ) - - with pytest.raises( - ModelBehaviorError, match="omitted an index while multiple function tool calls" - ): - await _collect_buffered_handler_events(*chunks) + with pytest.raises(ModelBehaviorError, match="could not be attributed safely"): + await _collect_buffered_tool_call_chunks(indexed, unindexed) @pytest.mark.parametrize( From e321a8f9a0592b39e1620785343ea4e143c87484 Mon Sep 17 00:00:00 2001 From: Rio Yu <52408936+rioyu123@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:36:46 +0800 Subject: [PATCH 11/11] fix(chat-completions): preserve passthrough stream behavior --- src/agents/models/chatcmpl_stream_handler.py | 56 ++++--- .../test_openai_chatcompletions_stream.py | 154 +++++++++++++++++- 2 files changed, 184 insertions(+), 26 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index a9d0166447..07c39cdf92 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -318,11 +318,6 @@ def _accumulate_tool_call_delta( if tool_call_delta.id and buffered_call.call_id == tool_call_delta.id ] - if len(matching_indexes) > 1: - raise ModelBehaviorError( - "Chat Completions tool call delta matched multiple buffered calls." - ) - if isinstance(tool_call_index, int): if matching_indexes and matching_indexes[0] != tool_call_index: raise ModelBehaviorError( @@ -435,7 +430,7 @@ def _buffered_tool_calls_chunk( cls, template_chunk: ChatCompletionChunk, buffered_calls: dict[int | None, _BufferedToolCall], - passthrough_tool_call_indexes: set[int], + passthrough_tool_call_indexes: set[int | None], ) -> ChatCompletionChunk: ordered_calls = sorted( buffered_calls.values(), @@ -444,9 +439,9 @@ def _buffered_tool_calls_chunk( call.index if isinstance(call.index, int) else 0, ), ) - used_indexes = passthrough_tool_call_indexes | { - call.index for call in ordered_calls if isinstance(call.index, int) - } + used_indexes = { + index for index in passthrough_tool_call_indexes if isinstance(index, int) + } | {call.index for call in ordered_calls if isinstance(call.index, int)} fallback_index = max(used_indexes, default=-1) + 1 tool_call_deltas = [ cls._buffered_tool_call_delta(buffered_call, fallback_index=fallback_index) @@ -466,7 +461,7 @@ async def buffer_tool_call_stream( ) -> AsyncIterator[ChatCompletionChunk]: """Buffer streamed function tool-call deltas until they are complete.""" buffered_calls: dict[int | None, _BufferedToolCall] = {} - passthrough_tool_call_indexes: set[int] = set() + passthrough_tool_call_indexes: set[int | None] = set() saw_passthrough_tool_call = False last_chunk: ChatCompletionChunk | None = None @@ -480,6 +475,8 @@ async def buffer_tool_call_stream( passthrough_choices: list[Choice] = [] for choice in chunk.choices: if choice.index != 0: + if choice.delta and choice.delta.tool_calls: + saw_passthrough_tool_call = True passthrough_choices.append(choice) continue @@ -488,28 +485,40 @@ async def buffer_tool_call_stream( if tool_call_deltas := (delta.tool_calls if delta and delta.tool_calls else None): remaining_tool_calls: list[ChoiceDeltaToolCall] = [] for tool_call_delta in tool_call_deltas: - if tool_call_delta.index in passthrough_tool_call_indexes: - if tool_call_delta.id and any( + matches_buffered_id = bool( + tool_call_delta.id + and any( buffered_call.call_id == tool_call_delta.id for buffered_call in buffered_calls.values() - ): - raise ModelBehaviorError( - "Chat Completions tool call index and ID identified different " - "tool call owners." - ) - saw_passthrough_tool_call = True - remaining_tool_calls.append(tool_call_delta) - elif cls._should_buffer_tool_call_delta(tool_call_delta): + ) + ) + is_unindexed_function_delta = tool_call_delta.index is None and ( + tool_call_delta.function is not None + or tool_call_delta.type == "function" + or matches_buffered_id + ) + if ( + tool_call_delta.index in passthrough_tool_call_indexes + and not is_unindexed_function_delta + ): if ( - saw_passthrough_tool_call - and tool_call_delta.index is None + tool_call_delta.index is None and tool_call_delta.id is None and tool_call_delta.function is None + and None in buffered_calls ): raise ModelBehaviorError( "Chat Completions tool call delta could not be attributed " "safely between buffered and passthrough calls." ) + if matches_buffered_id: + raise ModelBehaviorError( + "Chat Completions tool call index and ID identified different " + "tool call owners." + ) + saw_passthrough_tool_call = True + remaining_tool_calls.append(tool_call_delta) + elif cls._should_buffer_tool_call_delta(tool_call_delta): cls._accumulate_tool_call_delta(buffered_calls, tool_call_delta) else: if ( @@ -520,8 +529,7 @@ async def buffer_tool_call_stream( "Chat Completions tool call index identified both a buffered " "function call and a passthrough call." ) - if isinstance(tool_call_delta.index, int): - passthrough_tool_call_indexes.add(tool_call_delta.index) + passthrough_tool_call_indexes.add(tool_call_delta.index) saw_passthrough_tool_call = True remaining_tool_calls.append(tool_call_delta) diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index a60b02df18..75993695eb 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -118,7 +118,12 @@ async def _collect_buffered_handler_events(*chunks: ChatCompletionChunk) -> list ] -def _lenient_chunk(delta: dict[str, Any], finish_reason: str | None = None) -> ChatCompletionChunk: +def _lenient_chunk( + delta: dict[str, Any], + finish_reason: str | None = None, + *, + choice_index: int = 0, +) -> ChatCompletionChunk: """Construct a chunk as AsyncStream does, without validating a missing tool-call index.""" return cast( ChatCompletionChunk, @@ -129,7 +134,9 @@ def _lenient_chunk(delta: dict[str, Any], finish_reason: str | None = None) -> C "object": "chat.completion.chunk", "created": 1, "model": "fake", - "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}], + "choices": [ + {"index": choice_index, "delta": delta, "finish_reason": finish_reason} + ], }, ), ) @@ -986,6 +993,79 @@ async def test_missing_index_replay_avoids_passthrough_index_collision() -> None assert replayed_calls[0].id == "call_1" +@pytest.mark.asyncio +async def test_unindexed_passthrough_continuation_remains_passthrough() -> None: + opening = _lenient_chunk( + {"tool_calls": [{"id": "custom-id", "type": "custom", "custom": {"input": "start"}}]} + ) + continuation = _lenient_chunk({"tool_calls": [{"custom": {"input": "-end"}}]}) + + buffered_chunks = await _collect_buffered_tool_call_chunks(opening, continuation) + + assert [chunk.choices[0].delta.tool_calls for chunk in buffered_chunks] == [ + opening.choices[0].delta.tool_calls, + continuation.choices[0].delta.tool_calls, + ] + + +@pytest.mark.asyncio +async def test_indexed_function_does_not_block_unindexed_passthrough_continuation() -> None: + indexed_function = _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ) + custom_opening = _lenient_chunk( + {"tool_calls": [{"id": "custom-id", "type": "custom", "custom": {"input": "start"}}]} + ) + custom_continuation = _lenient_chunk({"tool_calls": [{"custom": {"input": "-end"}}]}) + + buffered_chunks = await _collect_buffered_tool_call_chunks( + indexed_function, + custom_opening, + custom_continuation, + ) + + assert [chunk.choices[0].delta.tool_calls for chunk in buffered_chunks[:-1]] == [ + custom_opening.choices[0].delta.tool_calls, + custom_continuation.choices[0].delta.tool_calls, + ] + replayed_calls = buffered_chunks[-1].choices[0].delta.tool_calls + assert replayed_calls and replayed_calls[0].id == "call_1" + + +@pytest.mark.asyncio +async def test_nonzero_choice_tool_call_suppresses_choice_zero_finish_error() -> None: + nonzero_choice = _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + }, + choice_index=1, + ) + choice_zero_finish = _lenient_chunk({}, finish_reason="tool_calls") + + buffered_chunks = await _collect_buffered_tool_call_chunks( + nonzero_choice, + choice_zero_finish, + ) + + assert buffered_chunks == [nonzero_choice] + + @pytest.mark.asyncio async def test_missing_index_continuation_merges_into_sole_buffered_function() -> None: chunks = ( @@ -1042,6 +1122,27 @@ async def test_missing_index_continuation_uses_a_unique_function_call_id() -> No ] +@pytest.mark.asyncio +async def test_late_index_for_unindexed_function_is_rejected() -> None: + opening = _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": '{"a":'}, + } + ] + } + ) + late_index = _lenient_chunk( + {"tool_calls": [{"index": 0, "id": "call_1", "function": {"arguments": "1}"}}]} + ) + + with pytest.raises(ModelBehaviorError, match="index and ID identified different"): + await _collect_buffered_tool_call_chunks(opening, late_index) + + @pytest.mark.asyncio async def test_missing_index_continuation_rejects_multiple_buffered_functions() -> None: chunks = ( @@ -1065,6 +1166,55 @@ async def test_missing_index_continuation_rejects_multiple_buffered_functions() await _collect_buffered_tool_call_chunks(*chunks, continuation) +@pytest.mark.asyncio +async def test_parallel_unindexed_function_openings_are_rejected() -> None: + first = _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "first_func", "arguments": "{}"}, + } + ] + } + ) + second = _lenient_chunk( + { + "tool_calls": [ + { + "id": "call_2", + "type": "function", + "function": {"name": "second_func", "arguments": "{}"}, + } + ] + } + ) + + with pytest.raises(ModelBehaviorError, match="omitted an index for a new call"): + await _collect_buffered_tool_call_chunks(first, second) + + +@pytest.mark.asyncio +async def test_passthrough_call_rejects_a_buffered_function_index() -> None: + function = _lenient_chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "my_func", "arguments": "{}"}, + } + ] + } + ) + custom = _lenient_chunk({"tool_calls": [{"index": 0, "id": "custom-id", "type": "custom"}]}) + + with pytest.raises(ModelBehaviorError, match="identified both a buffered function call"): + await _collect_buffered_tool_call_chunks(function, custom) + + @pytest.mark.asyncio async def test_anonymous_delta_rejects_mixed_unindexed_call_owners() -> None: custom = _lenient_chunk(