From aaca35ecfc610e5cc4cfb68442ebec5604a271d0 Mon Sep 17 00:00:00 2001 From: c <37263590+Aphroq@users.noreply.github.com> Date: Mon, 4 May 2026 06:05:56 +0000 Subject: [PATCH 1/6] Fix chat completions fallback output indexes --- src/agents/models/chatcmpl_stream_handler.py | 9 ++- .../test_openai_chatcompletions_stream.py | 75 +++++++++++++++++++ 2 files changed, 81 insertions(+), 3 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index a862a13783..5ef5fb4d79 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -624,6 +624,7 @@ async def handle_stream( ) # Send completion events for function calls + fallback_emitted_count = 0 for index, function_call in state.function_calls.items(): if state.function_call_streaming.get(index, False): # Function call was streamed, just send the completion event @@ -668,6 +669,7 @@ async def handle_stream( fallback_starting_index += sum( 1 for streaming in state.function_call_streaming.values() if streaming ) + fallback_output_index = fallback_starting_index + fallback_emitted_count # Build function call kwargs, include provider_data if present fallback_func_call_kwargs: dict[str, Any] = { @@ -690,23 +692,24 @@ async def handle_stream( # Send all events at once (backward compatibility) yield ResponseOutputItemAddedEvent( item=ResponseFunctionToolCall(**fallback_func_call_kwargs), - output_index=fallback_starting_index, + output_index=fallback_output_index, type="response.output_item.added", sequence_number=sequence_number.get_and_increment(), ) yield ResponseFunctionCallArgumentsDeltaEvent( delta=function_call.arguments, item_id=FAKE_RESPONSES_ID, - output_index=fallback_starting_index, + output_index=fallback_output_index, type="response.function_call_arguments.delta", sequence_number=sequence_number.get_and_increment(), ) yield ResponseOutputItemDoneEvent( item=ResponseFunctionToolCall(**fallback_func_call_kwargs), - output_index=fallback_starting_index, + output_index=fallback_output_index, type="response.output_item.done", sequence_number=sequence_number.get_and_increment(), ) + fallback_emitted_count += 1 # Finally, send the Response completed event outputs: list[ResponseOutputItem] = [] diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 847aef8da9..3e5938d227 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -553,3 +553,78 @@ async def patched_fetch_response(self, *args, **kwargs): assert isinstance(function_call_output, ResponseFunctionToolCall) assert function_call_output.name == "write_file" assert function_call_output.arguments == '{"filename": "test.py", "content": "print(hello)"}' + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_stream_response_fallback_tool_calls_use_distinct_output_indexes(monkeypatch) -> None: + tool_call_delta1 = ChoiceDeltaToolCall( + index=0, + function=ChoiceDeltaToolCallFunction(name="first_tool", arguments='{"a": 1}'), + type="function", + ) + tool_call_delta2 = ChoiceDeltaToolCall( + index=1, + function=ChoiceDeltaToolCallFunction(name="second_tool", arguments='{"b": 2}'), + type="function", + ) + chunk1 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[tool_call_delta1]))], + ) + chunk2 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[tool_call_delta2]))], + usage=CompletionUsage(completion_tokens=1, prompt_tokens=1, total_tokens=2), + ) + + async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: + for chunk in (chunk1, chunk2): + yield chunk + + async def patched_fetch_response(self, *args, **kwargs): + response = Response( + id="resp-id", + created_at=0, + model="fake-model", + object="response", + output=[], + tool_choice="none", + tools=[], + parallel_tool_calls=False, + ) + return response, fake_stream() + + monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) + model = OpenAIProvider(use_responses=False).get_model("gpt-4") + output_events = [] + + async for event in model.stream_response( + system_instructions=None, + input="", + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ): + output_events.append(event) + + added_events = [event for event in output_events if event.type == "response.output_item.added"] + delta_events = [ + event for event in output_events if event.type == "response.function_call_arguments.delta" + ] + done_events = [event for event in output_events if event.type == "response.output_item.done"] + + assert [event.output_index for event in added_events] == [0, 1] + assert [event.output_index for event in delta_events] == [0, 1] + assert [event.output_index for event in done_events] == [0, 1] From c303f92361a98524007ba63cbb72764ce9098467 Mon Sep 17 00:00:00 2001 From: c <37263590+Aphroq@users.noreply.github.com> Date: Mon, 4 May 2026 06:17:55 +0000 Subject: [PATCH 2/6] fix: emit integer chat stream output indexes --- src/agents/models/chatcmpl_stream_handler.py | 30 +++++++++----------- tests/models/test_reasoning_content.py | 18 ++++++++++++ 2 files changed, 31 insertions(+), 17 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 5ef5fb4d79..338cab7f1a 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -84,6 +84,10 @@ def get_and_increment(self) -> int: class ChatCmplStreamHandler: + @staticmethod + def _assistant_message_output_index(state: StreamingState) -> int: + return 1 if state.reasoning_content_index_and_output is not None else 0 + @classmethod def _finish_reasoning_summary_part( cls, @@ -341,16 +345,14 @@ async def handle_stream( # Notify consumers of the start of a new output message + first content part yield ResponseOutputItemAddedEvent( item=assistant_item, - output_index=state.reasoning_content_index_and_output - is not None, # fixed 0 -> 0 or 1 + output_index=cls._assistant_message_output_index(state), type="response.output_item.added", sequence_number=sequence_number.get_and_increment(), ) yield ResponseContentPartAddedEvent( content_index=state.text_content_index_and_output[0], item_id=FAKE_RESPONSES_ID, - output_index=state.reasoning_content_index_and_output - is not None, # fixed 0 -> 0 or 1 + output_index=cls._assistant_message_output_index(state), part=ResponseOutputText( text="", type="output_text", @@ -374,8 +376,7 @@ async def handle_stream( content_index=state.text_content_index_and_output[0], delta=delta.content, item_id=FAKE_RESPONSES_ID, - output_index=state.reasoning_content_index_and_output - is not None, # fixed 0 -> 0 or 1 + output_index=cls._assistant_message_output_index(state), type="response.output_text.delta", sequence_number=sequence_number.get_and_increment(), logprobs=delta_logprobs, @@ -415,15 +416,14 @@ async def handle_stream( # Notify downstream that assistant message + first content part are starting yield ResponseOutputItemAddedEvent( item=assistant_item, - output_index=state.reasoning_content_index_and_output - is not None, # fixed 0 -> 0 or 1 + output_index=cls._assistant_message_output_index(state), type="response.output_item.added", sequence_number=sequence_number.get_and_increment(), ) yield ResponseContentPartAddedEvent( content_index=state.refusal_content_index_and_output[0], item_id=FAKE_RESPONSES_ID, - output_index=(1 if state.reasoning_content_index_and_output else 0), + output_index=cls._assistant_message_output_index(state), part=ResponseOutputRefusal( refusal="", type="refusal", @@ -436,8 +436,7 @@ async def handle_stream( content_index=state.refusal_content_index_and_output[0], delta=delta.refusal, item_id=FAKE_RESPONSES_ID, - output_index=state.reasoning_content_index_and_output - is not None, # fixed 0 -> 0 or 1 + output_index=cls._assistant_message_output_index(state), type="response.refusal.delta", sequence_number=sequence_number.get_and_increment(), ) @@ -603,8 +602,7 @@ async def handle_stream( yield ResponseContentPartDoneEvent( content_index=state.text_content_index_and_output[0], item_id=FAKE_RESPONSES_ID, - output_index=state.reasoning_content_index_and_output - is not None, # fixed 0 -> 0 or 1 + output_index=cls._assistant_message_output_index(state), part=state.text_content_index_and_output[1], type="response.content_part.done", sequence_number=sequence_number.get_and_increment(), @@ -616,8 +614,7 @@ async def handle_stream( yield ResponseContentPartDoneEvent( content_index=state.refusal_content_index_and_output[0], item_id=FAKE_RESPONSES_ID, - output_index=state.reasoning_content_index_and_output - is not None, # fixed 0 -> 0 or 1 + output_index=cls._assistant_message_output_index(state), part=state.refusal_content_index_and_output[1], type="response.content_part.done", sequence_number=sequence_number.get_and_increment(), @@ -750,8 +747,7 @@ async def handle_stream( # send a ResponseOutputItemDone for the assistant message yield ResponseOutputItemDoneEvent( item=assistant_msg, - output_index=state.reasoning_content_index_and_output - is not None, # fixed 0 -> 0 or 1 + output_index=cls._assistant_message_output_index(state), type="response.output_item.done", sequence_number=sequence_number.get_and_increment(), ) diff --git a/tests/models/test_reasoning_content.py b/tests/models/test_reasoning_content.py index 2f583b4017..dd11824a48 100644 --- a/tests/models/test_reasoning_content.py +++ b/tests/models/test_reasoning_content.py @@ -160,6 +160,24 @@ async def patched_fetch_response(self, *args, **kwargs): assert content_delta_events[0].delta == "The answer" assert content_delta_events[1].delta == " is 42" + assistant_message_index_events = [] + for event in output_events: + event_any = cast(Any, event) + if event.type in {"response.output_item.added", "response.output_item.done"}: + if event_any.item.type == "message": + assistant_message_index_events.append(event_any) + elif event.type in { + "response.content_part.added", + "response.output_text.delta", + "response.content_part.done", + }: + assistant_message_index_events.append(event_any) + + assert assistant_message_index_events + for event in assistant_message_index_events: + assert event.output_index == 1 + assert type(event.output_index) is int + # verify the final response contains both types of content response_event = output_events[-1] assert response_event.type == "response.completed" From 49142b5573002c824e4468c3c21a9809eb8b8bcb Mon Sep 17 00:00:00 2001 From: c <37263590+Aphroq@users.noreply.github.com> Date: Mon, 4 May 2026 11:45:51 +0000 Subject: [PATCH 3/6] fix: stabilize chat completions tool call indexes --- src/agents/models/chatcmpl_stream_handler.py | 58 ++--- .../test_openai_chatcompletions_stream.py | 208 ++++++++++++++++++ tests/models/test_reasoning_content.py | 98 ++++++++- 3 files changed, 324 insertions(+), 40 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 338cab7f1a..c94011a49b 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -88,6 +88,19 @@ class ChatCmplStreamHandler: def _assistant_message_output_index(state: StreamingState) -> int: return 1 if state.reasoning_content_index_and_output is not None else 0 + @staticmethod + def _function_call_output_base(state: StreamingState) -> int: + output_index = 0 + if state.reasoning_content_index_and_output: + output_index += 1 + if state.text_content_index_and_output or state.refusal_content_index_and_output: + output_index += 1 + return output_index + + @classmethod + def _next_function_call_output_index(cls, state: StreamingState) -> int: + return cls._function_call_output_base(state) + len(state.function_calls) + @classmethod def _finish_reasoning_summary_part( cls, @@ -447,6 +460,9 @@ async def handle_stream( if delta.tool_calls: for tc_delta in delta.tool_calls: if tc_delta.index not in state.function_calls: + state.function_call_output_idx[tc_delta.index] = ( + cls._next_function_call_output_index(state) + ) state.function_calls[tc_delta.index] = ResponseFunctionToolCall( id=FAKE_RESPONSES_ID, arguments="", @@ -526,25 +542,9 @@ async def handle_stream( and function_call.name and function_call.call_id ): - # Calculate the output index for this function call - function_call_starting_index = 0 - if state.reasoning_content_index_and_output: - function_call_starting_index += 1 - if state.text_content_index_and_output: - function_call_starting_index += 1 - if state.refusal_content_index_and_output: - function_call_starting_index += 1 - - # Add offset for already started function calls - function_call_starting_index += sum( - 1 for streaming in state.function_call_streaming.values() if streaming - ) - # Mark this function call as streaming and store its output index state.function_call_streaming[tc_delta.index] = True - state.function_call_output_idx[tc_delta.index] = ( - function_call_starting_index - ) + function_call_output_index = state.function_call_output_idx[tc_delta.index] # Send initial function call added event func_call_item = ResponseFunctionToolCall( @@ -569,7 +569,7 @@ async def handle_stream( func_call_item.provider_data = merged_provider_data # type: ignore[attr-defined] yield ResponseOutputItemAddedEvent( item=func_call_item, - output_index=function_call_starting_index, + output_index=function_call_output_index, type="response.output_item.added", sequence_number=sequence_number.get_and_increment(), ) @@ -592,12 +592,7 @@ async def handle_stream( for event in cls._finish_reasoning_item(state, sequence_number): yield event - function_call_starting_index = 0 - if state.reasoning_content_index_and_output: - function_call_starting_index += 1 - if state.text_content_index_and_output: - function_call_starting_index += 1 # Send end event for this content part yield ResponseContentPartDoneEvent( content_index=state.text_content_index_and_output[0], @@ -609,7 +604,6 @@ async def handle_stream( ) if state.refusal_content_index_and_output: - function_call_starting_index += 1 # Send end event for this content part yield ResponseContentPartDoneEvent( content_index=state.refusal_content_index_and_output[0], @@ -621,7 +615,6 @@ async def handle_stream( ) # Send completion events for function calls - fallback_emitted_count = 0 for index, function_call in state.function_calls.items(): if state.function_call_streaming.get(index, False): # Function call was streamed, just send the completion event @@ -654,19 +647,7 @@ async def handle_stream( else: # Function call was not streamed (fallback to old behavior) # This handles edge cases where function name never arrived - fallback_starting_index = 0 - if state.reasoning_content_index_and_output: - fallback_starting_index += 1 - if state.text_content_index_and_output: - fallback_starting_index += 1 - if state.refusal_content_index_and_output: - fallback_starting_index += 1 - - # Add offset for already started function calls - fallback_starting_index += sum( - 1 for streaming in state.function_call_streaming.values() if streaming - ) - fallback_output_index = fallback_starting_index + fallback_emitted_count + fallback_output_index = state.function_call_output_idx[index] # Build function call kwargs, include provider_data if present fallback_func_call_kwargs: dict[str, Any] = { @@ -706,7 +687,6 @@ async def handle_stream( type="response.output_item.done", sequence_number=sequence_number.get_and_increment(), ) - fallback_emitted_count += 1 # Finally, send the Response completed event outputs: list[ResponseOutputItem] = [] diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 3e5938d227..ce1ea6eb7d 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -628,3 +628,211 @@ async def patched_fetch_response(self, *args, **kwargs): assert [event.output_index for event in added_events] == [0, 1] assert [event.output_index for event in delta_events] == [0, 1] assert [event.output_index for event in done_events] == [0, 1] + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_stream_response_mixed_tool_calls_use_final_output_indexes(monkeypatch) -> None: + fallback_tool_call = ChoiceDeltaToolCall( + index=0, + function=ChoiceDeltaToolCallFunction(name="first_tool", arguments='{"a": 1}'), + type="function", + ) + streamed_tool_call = ChoiceDeltaToolCall( + index=1, + id="second-tool-call-id", + function=ChoiceDeltaToolCallFunction(name="second_tool", arguments='{"b": 2}'), + type="function", + ) + chunk1 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[fallback_tool_call]))], + ) + chunk2 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[streamed_tool_call]))], + usage=CompletionUsage(completion_tokens=1, prompt_tokens=1, total_tokens=2), + ) + + async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: + for chunk in (chunk1, chunk2): + yield chunk + + async def patched_fetch_response(self, *args, **kwargs): + response = Response( + id="resp-id", + created_at=0, + model="fake-model", + object="response", + output=[], + tool_choice="none", + tools=[], + parallel_tool_calls=False, + ) + return response, fake_stream() + + monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) + model = OpenAIProvider(use_responses=False).get_model("gpt-4") + output_events = [] + + async for event in model.stream_response( + system_instructions=None, + input="", + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ): + output_events.append(event) + + added_events = [event for event in output_events if event.type == "response.output_item.added"] + delta_events = [ + event for event in output_events if event.type == "response.function_call_arguments.delta" + ] + done_events = [event for event in output_events if event.type == "response.output_item.done"] + completed_event = next(event for event in output_events if event.type == "response.completed") + + added_event_indexes = {} + for event in added_events: + assert isinstance(event.item, ResponseFunctionToolCall) + added_event_indexes[event.item.name] = event.output_index + + done_event_indexes = {} + for event in done_events: + assert isinstance(event.item, ResponseFunctionToolCall) + done_event_indexes[event.item.name] = event.output_index + + completed_output_names = [] + for output in completed_event.response.output: + assert isinstance(output, ResponseFunctionToolCall) + completed_output_names.append(output.name) + + assert added_event_indexes == { + "first_tool": 0, + "second_tool": 1, + } + assert {event.delta: event.output_index for event in delta_events} == { + '{"a": 1}': 0, + '{"b": 2}': 1, + } + assert done_event_indexes == { + "first_tool": 0, + "second_tool": 1, + } + assert completed_output_names == ["first_tool", "second_tool"] + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_stream_response_text_before_mixed_tool_calls_offsets_tool_indexes( + monkeypatch, +) -> None: + fallback_tool_call = ChoiceDeltaToolCall( + index=0, + function=ChoiceDeltaToolCallFunction(name="first_tool", arguments='{"a": 1}'), + type="function", + ) + streamed_tool_call = ChoiceDeltaToolCall( + index=1, + id="second-tool-call-id", + function=ChoiceDeltaToolCallFunction(name="second_tool", arguments='{"b": 2}'), + type="function", + ) + chunk1 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(content="Preparing tools"))], + ) + chunk2 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[fallback_tool_call]))], + ) + chunk3 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[streamed_tool_call]))], + usage=CompletionUsage(completion_tokens=1, prompt_tokens=1, total_tokens=2), + ) + + async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: + for chunk in (chunk1, chunk2, chunk3): + yield chunk + + async def patched_fetch_response(self, *args, **kwargs): + response = Response( + id="resp-id", + created_at=0, + model="fake-model", + object="response", + output=[], + tool_choice="none", + tools=[], + parallel_tool_calls=False, + ) + return response, fake_stream() + + monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) + model = OpenAIProvider(use_responses=False).get_model("gpt-4") + output_events = [] + + async for event in model.stream_response( + system_instructions=None, + input="", + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ): + output_events.append(event) + + added_events = [event for event in output_events if event.type == "response.output_item.added"] + delta_events = [ + event for event in output_events if event.type == "response.function_call_arguments.delta" + ] + done_events = [event for event in output_events if event.type == "response.output_item.done"] + completed_event = next(event for event in output_events if event.type == "response.completed") + + added_tool_indexes = {} + for event in added_events: + if isinstance(event.item, ResponseFunctionToolCall): + added_tool_indexes[event.item.name] = event.output_index + + done_tool_indexes = {} + for event in done_events: + if isinstance(event.item, ResponseFunctionToolCall): + done_tool_indexes[event.item.name] = event.output_index + + assert added_tool_indexes == {"first_tool": 1, "second_tool": 2} + assert {event.delta: event.output_index for event in delta_events} == { + '{"a": 1}': 1, + '{"b": 2}': 2, + } + assert done_tool_indexes == {"first_tool": 1, "second_tool": 2} + assert isinstance(completed_event.response.output[0], ResponseOutputMessage) + completed_tool_outputs = completed_event.response.output[1:] + completed_tool_names = [] + for output in completed_tool_outputs: + assert isinstance(output, ResponseFunctionToolCall) + completed_tool_names.append(output.name) + assert completed_tool_names == ["first_tool", "second_tool"] diff --git a/tests/models/test_reasoning_content.py b/tests/models/test_reasoning_content.py index dd11824a48..2bbac34892 100644 --- a/tests/models/test_reasoning_content.py +++ b/tests/models/test_reasoning_content.py @@ -5,7 +5,12 @@ import pytest from openai.types.chat import ChatCompletion, ChatCompletionChunk, ChatCompletionMessage -from openai.types.chat.chat_completion_chunk import Choice, ChoiceDelta +from openai.types.chat.chat_completion_chunk import ( + Choice, + ChoiceDelta, + ChoiceDeltaToolCall, + ChoiceDeltaToolCallFunction, +) from openai.types.completion_usage import ( CompletionTokensDetails, CompletionUsage, @@ -13,6 +18,7 @@ ) from openai.types.responses import ( Response, + ResponseFunctionToolCall, ResponseOutputMessage, ResponseOutputText, ResponseReasoningItem, @@ -41,6 +47,12 @@ def create_reasoning_delta(content: str) -> dict[str, Any]: } +def create_tool_call_delta( + tool_calls: list[ChoiceDeltaToolCall], +) -> dict[str, Any]: + return {"content": None, "role": None, "function_call": None, "tool_calls": tool_calls} + + def create_chunk(delta: dict[str, Any], include_usage: bool = False) -> ChatCompletionChunk: """Create a ChatCompletionChunk with the given delta""" # Create a ChoiceDelta object from the dictionary @@ -193,6 +205,90 @@ async def patched_fetch_response(self, *args, **kwargs): assert response_event.response.output[1].content[0].text == "The answer is 42" +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_stream_response_reasoning_before_mixed_tool_calls_offsets_tool_indexes( + monkeypatch, +) -> None: + fallback_tool_call = ChoiceDeltaToolCall( + index=0, + function=ChoiceDeltaToolCallFunction(name="first_tool", arguments='{"a": 1}'), + type="function", + ) + streamed_tool_call = ChoiceDeltaToolCall( + index=1, + id="second-tool-call-id", + function=ChoiceDeltaToolCallFunction(name="second_tool", arguments='{"b": 2}'), + type="function", + ) + chunks = [ + create_chunk(create_reasoning_delta("Let me think")), + create_chunk(create_tool_call_delta([fallback_tool_call])), + create_chunk(create_tool_call_delta([streamed_tool_call]), include_usage=True), + ] + + async def patched_fetch_response(self, *args, **kwargs): + resp = Response( + id="resp-id", + created_at=0, + model="fake-model", + object="response", + output=[], + tool_choice="none", + tools=[], + parallel_tool_calls=False, + ) + return resp, create_fake_stream(chunks) + + monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) + model = OpenAIProvider(use_responses=False).get_model("gpt-4") + output_events = [] + async for event in model.stream_response( + system_instructions=None, + input="", + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ): + output_events.append(event) + + added_events = [event for event in output_events if event.type == "response.output_item.added"] + delta_events = [ + event for event in output_events if event.type == "response.function_call_arguments.delta" + ] + done_events = [event for event in output_events if event.type == "response.output_item.done"] + completed_event = next(event for event in output_events if event.type == "response.completed") + + added_tool_indexes = {} + for event in added_events: + if isinstance(event.item, ResponseFunctionToolCall): + added_tool_indexes[event.item.name] = event.output_index + + done_tool_indexes = {} + for event in done_events: + if isinstance(event.item, ResponseFunctionToolCall): + done_tool_indexes[event.item.name] = event.output_index + + assert added_tool_indexes == {"first_tool": 1, "second_tool": 2} + assert {event.delta: event.output_index for event in delta_events} == { + '{"a": 1}': 1, + '{"b": 2}': 2, + } + assert done_tool_indexes == {"first_tool": 1, "second_tool": 2} + assert isinstance(completed_event.response.output[0], ResponseReasoningItem) + completed_tool_outputs = completed_event.response.output[1:] + completed_tool_names = [] + for output in completed_tool_outputs: + assert isinstance(output, ResponseFunctionToolCall) + completed_tool_names.append(output.name) + assert completed_tool_names == ["first_tool", "second_tool"] + + @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_keeps_reasoning_item_open_across_interleaved_text( From bb4f9ca157d9d4719f8d59804f0afdce0a789f17 Mon Sep 17 00:00:00 2001 From: c <37263590+Aphroq@users.noreply.github.com> Date: Mon, 4 May 2026 22:02:39 +0000 Subject: [PATCH 4/6] fix: handle late text after fallback tool calls --- src/agents/models/chatcmpl_stream_handler.py | 19 ++-- .../test_openai_chatcompletions_stream.py | 92 +++++++++++++++++++ 2 files changed, 104 insertions(+), 7 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index c94011a49b..b221458fb7 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -98,8 +98,12 @@ def _function_call_output_base(state: StreamingState) -> int: return output_index @classmethod - def _next_function_call_output_index(cls, state: StreamingState) -> int: - return cls._function_call_output_base(state) + len(state.function_calls) + def _function_call_output_index(cls, state: StreamingState, function_call_index: int) -> int: + for offset, index in enumerate(state.function_calls): + if index == function_call_index: + return cls._function_call_output_base(state) + offset + + raise KeyError(f"Function call index {function_call_index} has not been tracked") @classmethod def _finish_reasoning_summary_part( @@ -460,9 +464,6 @@ async def handle_stream( if delta.tool_calls: for tc_delta in delta.tool_calls: if tc_delta.index not in state.function_calls: - state.function_call_output_idx[tc_delta.index] = ( - cls._next_function_call_output_index(state) - ) state.function_calls[tc_delta.index] = ResponseFunctionToolCall( id=FAKE_RESPONSES_ID, arguments="", @@ -544,7 +545,11 @@ async def handle_stream( ): # Mark this function call as streaming and store its output index state.function_call_streaming[tc_delta.index] = True - function_call_output_index = state.function_call_output_idx[tc_delta.index] + function_call_output_index = cls._function_call_output_index( + state, + tc_delta.index, + ) + state.function_call_output_idx[tc_delta.index] = function_call_output_index # Send initial function call added event func_call_item = ResponseFunctionToolCall( @@ -647,7 +652,7 @@ async def handle_stream( else: # Function call was not streamed (fallback to old behavior) # This handles edge cases where function name never arrived - fallback_output_index = state.function_call_output_idx[index] + fallback_output_index = cls._function_call_output_index(state, index) # Build function call kwargs, include provider_data if present fallback_func_call_kwargs: dict[str, Any] = { diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index ce1ea6eb7d..ef0bdefd95 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -836,3 +836,95 @@ async def patched_fetch_response(self, *args, **kwargs): assert isinstance(output, ResponseFunctionToolCall) completed_tool_names.append(output.name) assert completed_tool_names == ["first_tool", "second_tool"] + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_stream_response_fallback_tool_call_before_text_uses_final_output_index( + monkeypatch, +) -> None: + fallback_tool_call = ChoiceDeltaToolCall( + index=0, + function=ChoiceDeltaToolCallFunction(name="first_tool", arguments='{"a": 1}'), + type="function", + ) + chunk1 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[fallback_tool_call]))], + ) + chunk2 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(content="answer"))], + usage=CompletionUsage(completion_tokens=1, prompt_tokens=1, total_tokens=2), + ) + + async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: + for chunk in (chunk1, chunk2): + yield chunk + + async def patched_fetch_response(self, *args, **kwargs): + response = Response( + id="resp-id", + created_at=0, + model="fake-model", + object="response", + output=[], + tool_choice="none", + tools=[], + parallel_tool_calls=False, + ) + return response, fake_stream() + + monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) + model = OpenAIProvider(use_responses=False).get_model("gpt-4") + output_events = [] + + async for event in model.stream_response( + system_instructions=None, + input="", + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ): + output_events.append(event) + + added_events = [event for event in output_events if event.type == "response.output_item.added"] + delta_events = [ + event for event in output_events if event.type == "response.function_call_arguments.delta" + ] + done_events = [event for event in output_events if event.type == "response.output_item.done"] + completed_event = next(event for event in output_events if event.type == "response.completed") + + added_message_event = next( + event for event in added_events if isinstance(event.item, ResponseOutputMessage) + ) + added_tool_event = next( + event for event in added_events if isinstance(event.item, ResponseFunctionToolCall) + ) + done_message_event = next( + event for event in done_events if isinstance(event.item, ResponseOutputMessage) + ) + done_tool_event = next( + event for event in done_events if isinstance(event.item, ResponseFunctionToolCall) + ) + + assert added_message_event.output_index == 0 + assert added_tool_event.output_index == 1 + assert [event.output_index for event in delta_events] == [1] + assert done_message_event.output_index == 0 + assert done_tool_event.output_index == 1 + + assert isinstance(completed_event.response.output[0], ResponseOutputMessage) + assert isinstance(completed_event.response.output[1], ResponseFunctionToolCall) + assert completed_event.response.output[1].name == "first_tool" From abb916e82cb5c5f0ede8781268b81e731fc6d5f3 Mon Sep 17 00:00:00 2001 From: c <37263590+Aphroq@users.noreply.github.com> Date: Tue, 5 May 2026 01:13:17 +0000 Subject: [PATCH 5/6] fix: defer streamed tool indexes until order is known --- src/agents/models/chatcmpl_stream_handler.py | 211 +++++++++++------- .../test_openai_chatcompletions_stream.py | 93 ++++++++ tests/models/test_reasoning_content.py | 79 +++++++ 3 files changed, 302 insertions(+), 81 deletions(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index b221458fb7..4cf5d74d9c 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -65,7 +65,9 @@ class StreamingState: function_calls: dict[int, ResponseFunctionToolCall] = field(default_factory=dict) # Fields for real-time function call streaming function_call_streaming: dict[int, bool] = field(default_factory=dict) + function_call_events_emitted: dict[int, bool] = field(default_factory=dict) function_call_output_idx: dict[int, int] = field(default_factory=dict) + function_call_pending_argument_deltas: dict[int, list[str]] = field(default_factory=dict) # Store accumulated thinking text and signature for Anthropic compatibility thinking_text: str = "" thinking_signature: str | None = None @@ -105,6 +107,84 @@ def _function_call_output_index(cls, state: StreamingState, function_call_index: raise KeyError(f"Function call index {function_call_index} has not been tracked") + @staticmethod + def _merged_provider_data( + state: StreamingState, + function_call: ResponseFunctionToolCall, + ) -> dict[str, Any] | None: + if not ( + state.provider_data + or (hasattr(function_call, "provider_data") and function_call.provider_data) + ): + return None + + merged_provider_data = state.provider_data.copy() if state.provider_data else {} + if hasattr(function_call, "provider_data") and function_call.provider_data: + merged_provider_data.update(function_call.provider_data) + return merged_provider_data + + @classmethod + def _function_call_item( + cls, + state: StreamingState, + function_call: ResponseFunctionToolCall, + *, + arguments: str, + ) -> ResponseFunctionToolCall: + function_call_kwargs: dict[str, Any] = { + "id": FAKE_RESPONSES_ID, + "call_id": function_call.call_id, + "arguments": arguments, + "name": function_call.name, + "type": "function_call", + } + + if merged_provider_data := cls._merged_provider_data(state, function_call): + function_call_kwargs["provider_data"] = merged_provider_data + + return ResponseFunctionToolCall(**function_call_kwargs) + + @staticmethod + def _can_emit_function_call_events(state: StreamingState) -> bool: + return bool(state.text_content_index_and_output or state.refusal_content_index_and_output) + + @classmethod + def _flush_pending_function_call_events( + cls, + state: StreamingState, + sequence_number: SequenceNumber, + *, + force: bool = False, + ) -> Iterator[TResponseStreamEvent]: + if not (force or cls._can_emit_function_call_events(state)): + return + + for index, function_call in state.function_calls.items(): + if not state.function_call_streaming.get(index, False): + continue + if state.function_call_events_emitted.get(index, False): + continue + + output_index = cls._function_call_output_index(state, index) + state.function_call_output_idx[index] = output_index + state.function_call_events_emitted[index] = True + + yield ResponseOutputItemAddedEvent( + item=cls._function_call_item(state, function_call, arguments=""), + output_index=output_index, + type="response.output_item.added", + sequence_number=sequence_number.get_and_increment(), + ) + + for delta in state.function_call_pending_argument_deltas.pop(index, []): + yield ResponseFunctionCallArgumentsDeltaEvent( + delta=delta, + item_id=FAKE_RESPONSES_ID, + output_index=output_index, + type="response.function_call_arguments.delta", + sequence_number=sequence_number.get_and_increment(), + ) + @classmethod def _finish_reasoning_summary_part( cls, @@ -405,6 +485,11 @@ async def handle_stream( state.text_content_index_and_output[1].logprobs = ( existing_logprobs + output_logprobs ) + for event in cls._flush_pending_function_call_events( + state, + sequence_number, + ): + yield event # Handle refusals (model declines to answer) # This is always set by the OpenAI API, but not by others e.g. LiteLLM @@ -459,6 +544,11 @@ async def handle_stream( ) # Accumulate the refusal string in the output part state.refusal_content_index_and_output[1].refusal += delta.refusal + for event in cls._flush_pending_function_call_events( + state, + sequence_number, + ): + yield event # Handle tool calls with real-time streaming support if delta.tool_calls: @@ -472,6 +562,8 @@ async def handle_stream( call_id="", ) state.function_call_streaming[tc_delta.index] = False + state.function_call_events_emitted[tc_delta.index] = False + state.function_call_pending_argument_deltas[tc_delta.index] = [] tc_function = tc_delta.function @@ -543,41 +635,7 @@ async def handle_stream( and function_call.name and function_call.call_id ): - # Mark this function call as streaming and store its output index state.function_call_streaming[tc_delta.index] = True - function_call_output_index = cls._function_call_output_index( - state, - tc_delta.index, - ) - state.function_call_output_idx[tc_delta.index] = function_call_output_index - - # Send initial function call added event - func_call_item = ResponseFunctionToolCall( - id=FAKE_RESPONSES_ID, - call_id=function_call.call_id, - arguments="", # Start with empty arguments - name=function_call.name, - type="function_call", - ) - # Merge provider_data from state and function_call (e.g. thought_signature) - if state.provider_data or ( - hasattr(function_call, "provider_data") and function_call.provider_data - ): - merged_provider_data = ( - state.provider_data.copy() if state.provider_data else {} - ) - if ( - hasattr(function_call, "provider_data") - and function_call.provider_data - ): - merged_provider_data.update(function_call.provider_data) - func_call_item.provider_data = merged_provider_data # type: ignore[attr-defined] - yield ResponseOutputItemAddedEvent( - item=func_call_item, - output_index=function_call_output_index, - type="response.output_item.added", - sequence_number=sequence_number.get_and_increment(), - ) # Stream arguments if we've started streaming this function call if ( @@ -585,18 +643,36 @@ async def handle_stream( and tc_function and tc_function.arguments ): - output_index = state.function_call_output_idx[tc_delta.index] - yield ResponseFunctionCallArgumentsDeltaEvent( - delta=tc_function.arguments, - item_id=FAKE_RESPONSES_ID, - output_index=output_index, - type="response.function_call_arguments.delta", - sequence_number=sequence_number.get_and_increment(), - ) + if state.function_call_events_emitted.get(tc_delta.index, False): + output_index = state.function_call_output_idx[tc_delta.index] + yield ResponseFunctionCallArgumentsDeltaEvent( + delta=tc_function.arguments, + item_id=FAKE_RESPONSES_ID, + output_index=output_index, + type="response.function_call_arguments.delta", + sequence_number=sequence_number.get_and_increment(), + ) + else: + state.function_call_pending_argument_deltas[tc_delta.index].append( + tc_function.arguments + ) + + for event in cls._flush_pending_function_call_events( + state, + sequence_number, + ): + yield event for event in cls._finish_reasoning_item(state, sequence_number): yield event + for event in cls._flush_pending_function_call_events( + state, + sequence_number, + force=True, + ): + yield event + if state.text_content_index_and_output: # Send end event for this content part yield ResponseContentPartDoneEvent( @@ -625,26 +701,12 @@ async def handle_stream( # Function call was streamed, just send the completion event output_index = state.function_call_output_idx[index] - # Build function call kwargs, include provider_data if present - func_call_kwargs: dict[str, Any] = { - "id": FAKE_RESPONSES_ID, - "call_id": function_call.call_id, - "arguments": function_call.arguments, - "name": function_call.name, - "type": "function_call", - } - - # Merge provider_data from state and function_call (e.g. thought_signature) - if state.provider_data or ( - hasattr(function_call, "provider_data") and function_call.provider_data - ): - merged_provider_data = state.provider_data.copy() if state.provider_data else {} - if hasattr(function_call, "provider_data") and function_call.provider_data: - merged_provider_data.update(function_call.provider_data) - func_call_kwargs["provider_data"] = merged_provider_data - yield ResponseOutputItemDoneEvent( - item=ResponseFunctionToolCall(**func_call_kwargs), + item=cls._function_call_item( + state, + function_call, + arguments=function_call.arguments, + ), output_index=output_index, type="response.output_item.done", sequence_number=sequence_number.get_and_increment(), @@ -653,28 +715,15 @@ async def handle_stream( # Function call was not streamed (fallback to old behavior) # This handles edge cases where function name never arrived fallback_output_index = cls._function_call_output_index(state, index) - - # Build function call kwargs, include provider_data if present - fallback_func_call_kwargs: dict[str, Any] = { - "id": FAKE_RESPONSES_ID, - "call_id": function_call.call_id, - "arguments": function_call.arguments, - "name": function_call.name, - "type": "function_call", - } - - # Merge provider_data from state and function_call (e.g. thought_signature) - if state.provider_data or ( - hasattr(function_call, "provider_data") and function_call.provider_data - ): - merged_provider_data = state.provider_data.copy() if state.provider_data else {} - if hasattr(function_call, "provider_data") and function_call.provider_data: - merged_provider_data.update(function_call.provider_data) - fallback_func_call_kwargs["provider_data"] = merged_provider_data + fallback_func_call_item = cls._function_call_item( + state, + function_call, + arguments=function_call.arguments, + ) # Send all events at once (backward compatibility) yield ResponseOutputItemAddedEvent( - item=ResponseFunctionToolCall(**fallback_func_call_kwargs), + item=fallback_func_call_item, output_index=fallback_output_index, type="response.output_item.added", sequence_number=sequence_number.get_and_increment(), @@ -687,7 +736,7 @@ async def handle_stream( sequence_number=sequence_number.get_and_increment(), ) yield ResponseOutputItemDoneEvent( - item=ResponseFunctionToolCall(**fallback_func_call_kwargs), + item=fallback_func_call_item, output_index=fallback_output_index, type="response.output_item.done", sequence_number=sequence_number.get_and_increment(), diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index ef0bdefd95..0295900ca7 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -928,3 +928,96 @@ async def patched_fetch_response(self, *args, **kwargs): assert isinstance(completed_event.response.output[0], ResponseOutputMessage) assert isinstance(completed_event.response.output[1], ResponseFunctionToolCall) assert completed_event.response.output[1].name == "first_tool" + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_stream_response_streamed_tool_call_before_text_uses_final_output_index( + monkeypatch, +) -> None: + streamed_tool_call = ChoiceDeltaToolCall( + index=0, + id="first-tool-call-id", + function=ChoiceDeltaToolCallFunction(name="first_tool", arguments='{"a": 1}'), + type="function", + ) + chunk1 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[streamed_tool_call]))], + ) + chunk2 = ChatCompletionChunk( + id="chunk-id", + created=1, + model="fake", + object="chat.completion.chunk", + choices=[Choice(index=0, delta=ChoiceDelta(content="answer"))], + usage=CompletionUsage(completion_tokens=1, prompt_tokens=1, total_tokens=2), + ) + + async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: + for chunk in (chunk1, chunk2): + yield chunk + + async def patched_fetch_response(self, *args, **kwargs): + response = Response( + id="resp-id", + created_at=0, + model="fake-model", + object="response", + output=[], + tool_choice="none", + tools=[], + parallel_tool_calls=False, + ) + return response, fake_stream() + + monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) + model = OpenAIProvider(use_responses=False).get_model("gpt-4") + output_events = [] + + async for event in model.stream_response( + system_instructions=None, + input="", + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ): + output_events.append(event) + + added_events = [event for event in output_events if event.type == "response.output_item.added"] + delta_events = [ + event for event in output_events if event.type == "response.function_call_arguments.delta" + ] + done_events = [event for event in output_events if event.type == "response.output_item.done"] + completed_event = next(event for event in output_events if event.type == "response.completed") + + added_message_event = next( + event for event in added_events if isinstance(event.item, ResponseOutputMessage) + ) + added_tool_event = next( + event for event in added_events if isinstance(event.item, ResponseFunctionToolCall) + ) + done_message_event = next( + event for event in done_events if isinstance(event.item, ResponseOutputMessage) + ) + done_tool_event = next( + event for event in done_events if isinstance(event.item, ResponseFunctionToolCall) + ) + + assert added_message_event.output_index == 0 + assert added_tool_event.output_index == 1 + assert [event.output_index for event in delta_events] == [1] + assert done_message_event.output_index == 0 + assert done_tool_event.output_index == 1 + + assert isinstance(completed_event.response.output[0], ResponseOutputMessage) + assert isinstance(completed_event.response.output[1], ResponseFunctionToolCall) + assert completed_event.response.output[1].name == "first_tool" diff --git a/tests/models/test_reasoning_content.py b/tests/models/test_reasoning_content.py index 2bbac34892..4415dbbee6 100644 --- a/tests/models/test_reasoning_content.py +++ b/tests/models/test_reasoning_content.py @@ -289,6 +289,85 @@ async def patched_fetch_response(self, *args, **kwargs): assert completed_tool_names == ["first_tool", "second_tool"] +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_stream_response_reasoning_then_streamed_tool_before_text_uses_final_output_index( + monkeypatch, +) -> None: + streamed_tool_call = ChoiceDeltaToolCall( + index=0, + id="tool-call-id", + function=ChoiceDeltaToolCallFunction(name="lookup_tool", arguments='{"query": "x"}'), + type="function", + ) + chunks = [ + create_chunk(create_reasoning_delta("Let me think")), + create_chunk(create_tool_call_delta([streamed_tool_call])), + create_chunk(create_content_delta("answer"), include_usage=True), + ] + + async def patched_fetch_response(self, *args, **kwargs): + resp = Response( + id="resp-id", + created_at=0, + model="fake-model", + object="response", + output=[], + tool_choice="none", + tools=[], + parallel_tool_calls=False, + ) + return resp, create_fake_stream(chunks) + + monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) + model = OpenAIProvider(use_responses=False).get_model("gpt-4") + output_events = [] + async for event in model.stream_response( + system_instructions=None, + input="", + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ): + output_events.append(event) + + added_events = [event for event in output_events if event.type == "response.output_item.added"] + delta_events = [ + event for event in output_events if event.type == "response.function_call_arguments.delta" + ] + done_events = [event for event in output_events if event.type == "response.output_item.done"] + completed_event = next(event for event in output_events if event.type == "response.completed") + + added_message_event = next( + event for event in added_events if isinstance(event.item, ResponseOutputMessage) + ) + added_tool_event = next( + event for event in added_events if isinstance(event.item, ResponseFunctionToolCall) + ) + done_message_event = next( + event for event in done_events if isinstance(event.item, ResponseOutputMessage) + ) + done_tool_event = next( + event for event in done_events if isinstance(event.item, ResponseFunctionToolCall) + ) + + assert added_message_event.output_index == 1 + assert added_tool_event.output_index == 2 + assert [event.output_index for event in delta_events] == [2] + assert done_message_event.output_index == 1 + assert done_tool_event.output_index == 2 + + assert isinstance(completed_event.response.output[0], ResponseReasoningItem) + assert isinstance(completed_event.response.output[1], ResponseOutputMessage) + assert isinstance(completed_event.response.output[2], ResponseFunctionToolCall) + assert completed_event.response.output[2].name == "lookup_tool" + + @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_keeps_reasoning_item_open_across_interleaved_text( From 949b4846d3133afe93896e994458d6e6f5568413 Mon Sep 17 00:00:00 2001 From: c <37263590+Aphroq@users.noreply.github.com> Date: Thu, 7 May 2026 01:32:42 +0000 Subject: [PATCH 6/6] test: keep streaming timing test in one context --- .../test_openai_chatcompletions_stream.py | 67 ++++++++++++------- 1 file changed, 42 insertions(+), 25 deletions(-) diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index a872ca2859..d5925e6ac1 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -1,5 +1,6 @@ import asyncio from collections.abc import AsyncIterator +from typing import Any import pytest from openai.types.chat.chat_completion_chunk import ( @@ -622,32 +623,48 @@ async def patched_fetch_response(self, *args, **kwargs): monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider(use_responses=False).get_model("gpt-4") - events = model.stream_response( - system_instructions=None, - input="", - model_settings=ModelSettings(), - tools=[], - output_schema=None, - handoffs=[], - tracing=ModelTracing.DISABLED, - previous_response_id=None, - conversation_id=None, - prompt=None, - ) + observed_events: asyncio.Queue[Any] = asyncio.Queue() + output_events = [] - assert (await asyncio.wait_for(anext(events), timeout=1)).type == "response.created" - added_event = await asyncio.wait_for(anext(events), timeout=1) - assert added_event.type == "response.output_item.added" - assert isinstance(added_event.item, ResponseFunctionToolCall) - assert added_event.item.name == "write_file" - delta_event = await asyncio.wait_for(anext(events), timeout=1) - assert delta_event.type == "response.function_call_arguments.delta" - assert delta_event.delta == '{"filename": "' - - allow_finish.set() - remaining_events = [] - async for event in events: - remaining_events.append(event) + async def collect_events() -> None: + async for event in model.stream_response( + system_instructions=None, + input="", + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ): + output_events.append(event) + await observed_events.put(event) + + collect_task = asyncio.create_task(collect_events()) + try: + assert (await asyncio.wait_for(observed_events.get(), timeout=1)).type == "response.created" + added_event = await asyncio.wait_for(observed_events.get(), timeout=1) + assert added_event.type == "response.output_item.added" + assert isinstance(added_event.item, ResponseFunctionToolCall) + assert added_event.item.name == "write_file" + delta_event = await asyncio.wait_for(observed_events.get(), timeout=1) + assert delta_event.type == "response.function_call_arguments.delta" + assert delta_event.delta == '{"filename": "' + assert not collect_task.done() + + allow_finish.set() + await asyncio.wait_for(collect_task, timeout=1) + finally: + if not collect_task.done(): + collect_task.cancel() + try: + await collect_task + except asyncio.CancelledError: + pass + + remaining_events = output_events[3:] assert [event.type for event in remaining_events] == [ "response.function_call_arguments.delta",