import logging from collections.abc import AsyncIterator from typing import Any, cast import pytest from openai.types.chat.chat_completion import ChatCompletion, Choice as ChatCompletionChoice from openai.types.chat.chat_completion_chunk import ( ChatCompletionChunk, Choice, ChoiceDelta, ChoiceDeltaToolCall, ChoiceDeltaToolCallFunction, ChoiceLogprobs, ) from openai.types.chat.chat_completion_message import ChatCompletionMessage from openai.types.chat.chat_completion_token_logprob import ( ChatCompletionTokenLogprob, TopLogprob, ) from openai.types.completion_usage import ( CompletionTokensDetails, CompletionUsage, PromptTokensDetails, ) from openai.types.responses import ( Response, ResponseCompletedEvent, ResponseFunctionToolCall, ResponseOutputMessage, ResponseOutputRefusal, ResponseOutputText, ResponseReasoningItem, ) from agents import Agent, Runner, function_tool from agents.exceptions import ModelBehaviorError, UserError from agents.model_settings import ModelSettings from agents.models.chatcmpl_stream_handler import ( ChatCmplStreamHandler, Part, SequenceNumber, StreamingState, _BufferedToolCall, _merge_buffered_metadata, _StreamOutputLayout, ) from agents.models.interface import ModelTracing from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel from agents.models.openai_provider import OpenAIProvider from tests.utils.simple_session import SimpleListSession async def _empty_chat_completion_stream() -> AsyncIterator[ChatCompletionChunk]: chunks: list[ChatCompletionChunk] = [] for chunk in chunks: yield chunk def _empty_response() -> Response: return Response( id="resp-id", created_at=0, model="fake-model", object="response", output=[], tool_choice="none", tools=[], parallel_tool_calls=False, ) async def _completion_stream( *chunks: ChatCompletionChunk, ) -> AsyncIterator[ChatCompletionChunk]: for chunk in chunks: yield chunk async def _collect_handler_events( *chunks: ChatCompletionChunk, model: str | None = None, ) -> list[Any]: return [ event async for event in ChatCmplStreamHandler.handle_stream( _empty_response(), cast(Any, _completion_stream(*chunks)), model=model ) ] async def _collect_buffered_tool_call_chunks( *chunks: ChatCompletionChunk, ) -> list[ChatCompletionChunk]: return [ chunk async for chunk in ChatCmplStreamHandler.buffer_tool_call_stream( _completion_stream(*chunks) ) ] @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_yields_events_for_text_content(monkeypatch) -> None: """ Validate that `stream_response` emits the correct sequence of events when streaming a simple assistant message consisting of plain text content. We simulate two chunks of text returned from the chat completion stream. """ # Create two chunks that will be emitted by the fake stream. chunk1 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(content="He"))], ) # Mark last chunk with usage so stream_response knows this is final. chunk2 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(content="llo"))], usage=CompletionUsage( completion_tokens=5, prompt_tokens=7, total_tokens=12, prompt_tokens_details=PromptTokensDetails.model_validate( {"cached_tokens": 2, "cache_write_tokens": 4} ), completion_tokens_details=CompletionTokensDetails(reasoning_tokens=3), ), ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2): yield c # Patch _fetch_response to inject our fake stream async def patched_fetch_response(self, *args, **kwargs): # `_fetch_response` is expected to return a Response skeleton and the async stream resp = Response( id="resp-id", created_at=0, model="fake-model", object="response", output=[], tool_choice="none", tools=[], parallel_tool_calls=False, ) return resp, 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) # We expect a response.created, then a response.output_item.added, content part added, # two content delta events (for "He" and "llo"), a content part done, the assistant message # output_item.done, and finally response.completed. # There should be 8 events in total. assert len(output_events) == 8 # First event indicates creation. assert output_events[0].type == "response.created" # The output item added and content part added events should mark the assistant message. assert output_events[1].type == "response.output_item.added" assert output_events[2].type == "response.content_part.added" # Two text delta events. assert output_events[3].type == "response.output_text.delta" assert output_events[3].delta == "He" assert output_events[4].type == "response.output_text.delta" assert output_events[4].delta == "llo" # After streaming, the content part and item should be marked done. assert output_events[5].type == "response.content_part.done" assert output_events[6].type == "response.output_item.done" # Last event indicates completion of the stream. assert output_events[7].type == "response.completed" # The completed response should have one output message with full text. completed_resp = output_events[7].response assert isinstance(completed_resp.output[0], ResponseOutputMessage) assert isinstance(completed_resp.output[0].content[0], ResponseOutputText) assert completed_resp.output[0].content[0].text == "Hello" assert completed_resp.usage, "usage should not be None" assert completed_resp.usage.input_tokens == 7 assert completed_resp.usage.output_tokens == 5 assert completed_resp.usage.total_tokens == 12 assert completed_resp.usage.input_tokens_details.cached_tokens == 2 assert getattr(completed_resp.usage.input_tokens_details, "cache_write_tokens", None) == 4 assert completed_resp.usage.output_tokens_details.reasoning_tokens == 3 @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_close_closes_provider_stream_with_async_close( monkeypatch, ) -> None: chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(content="Hi"))], ) class ClosableChatStream: def __init__(self) -> None: self._yielded = False self.close_calls = 0 def __aiter__(self) -> "ClosableChatStream": return self async def __anext__(self) -> ChatCompletionChunk: if self._yielded: raise StopAsyncIteration self._yielded = True return chunk async def close(self) -> None: self.close_calls += 1 provider_stream = ClosableChatStream() async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), provider_stream monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider(use_responses=False).get_model("gpt-4") stream = 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, ) stream_agen = cast(Any, stream) event = await stream_agen.__anext__() assert event.type == "response.created" await stream_agen.aclose() assert provider_stream.close_calls == 1 @pytest.mark.asyncio async def test_stream_handler_filters_multiple_choices_by_default( caplog: pytest.LogCaptureFixture, ) -> None: caplog.set_level(logging.WARNING, logger="openai.agents") chunks = [ ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=1, delta=ChoiceDelta(content="ignored-first"))], ), ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[ Choice(index=0, delta=ChoiceDelta(content="kept")), Choice(index=1, delta=ChoiceDelta(content="ignored-second")), ], ), ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=2, delta=ChoiceDelta(content="ignored-third"))], usage=CompletionUsage(completion_tokens=1, prompt_tokens=2, total_tokens=3), ), ] async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for chunk in chunks: yield chunk events = [ event async for event in ChatCmplStreamHandler.handle_stream( _empty_response(), cast(Any, fake_stream()) ) ] text_delta_events = [event for event in events if event.type == "response.output_text.delta"] assert [event.delta for event in text_delta_events] == ["kept"] completed_event = next(event for event in events if event.type == "response.completed") assert isinstance(completed_event, ResponseCompletedEvent) assert isinstance(completed_event.response.output[0], ResponseOutputMessage) text_part = completed_event.response.output[0].content[0] assert isinstance(text_part, ResponseOutputText) assert text_part.text == "kept" assert completed_event.response.usage assert completed_event.response.usage.total_tokens == 3 choice_warnings = [ record for record in caplog.records if "multiple choices or nonzero choice indexes" in record.getMessage() ] assert len(choice_warnings) == 1 @pytest.mark.asyncio async def test_stream_handler_keeps_empty_choice_usage_chunks() -> None: chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[], usage=CompletionUsage(completion_tokens=1, prompt_tokens=2, total_tokens=3), ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk events = [ event async for event in ChatCmplStreamHandler.handle_stream( _empty_response(), cast(Any, fake_stream()) ) ] assert [event.type for event in events] == ["response.created", "response.completed"] completed_event = events[-1] assert isinstance(completed_event, ResponseCompletedEvent) assert completed_event.response.output == [] assert completed_event.response.usage assert completed_event.response.usage.total_tokens == 3 @pytest.mark.asyncio async def test_stream_handler_rejects_multiple_choices_in_strict_mode() -> None: chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[ Choice(index=0, delta=ChoiceDelta(content="first")), Choice(index=1, delta=ChoiceDelta(content="second")), ], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk with pytest.raises(UserError, match="multiple choices or nonzero"): async for _ in ChatCmplStreamHandler.handle_stream( _empty_response(), cast(Any, fake_stream()), strict_feature_validation=True ): pass @pytest.mark.asyncio async def test_stream_handler_rejects_nonzero_choice_index_in_strict_mode() -> None: chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=1, delta=ChoiceDelta(content="second"))], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk with pytest.raises(UserError, match="multiple choices or nonzero"): async for _ in ChatCmplStreamHandler.handle_stream( _empty_response(), cast(Any, fake_stream()), strict_feature_validation=True ): pass @pytest.mark.asyncio async def test_buffer_tool_call_stream_merges_provider_metadata() -> None: tool_call_delta1 = ChoiceDeltaToolCall( index=0, id="tool-id", function=ChoiceDeltaToolCallFunction(name="my_func", arguments='{"a":'), type="function", ) tool_call_delta1_any = cast(Any, tool_call_delta1) tool_call_delta1_any.provider_specific_fields = { "nested": {"keep": "provider", "stable": {"value": 1}}, "replace": "old", } tool_call_delta1_any.extra_content = { "google": {"thought_signature": "sig-1", "stable": {"value": "kept"}} } tool_call_delta2 = ChoiceDeltaToolCall( index=0, id=None, function=ChoiceDeltaToolCallFunction(name=None, arguments="1}"), type="function", ) tool_call_delta2_any = cast(Any, tool_call_delta2) tool_call_delta2_any.provider_specific_fields = { "nested": {"stable": {}, "new": "provider"}, "replace": "new", } tool_call_delta2_any.extra_content = {"google": {"stable": {}, "new": "extra"}} 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]))], ) buffered_chunks = await _collect_buffered_tool_call_chunks(chunk1, chunk2) assert len(buffered_chunks) == 1 buffered_delta = buffered_chunks[0].choices[0].delta assert buffered_delta.tool_calls buffered_tool_call = buffered_delta.tool_calls[0] assert buffered_tool_call.function assert buffered_tool_call.function.arguments == '{"a":1}' assert cast(Any, buffered_tool_call).provider_specific_fields == { "nested": {"keep": "provider", "stable": {"value": 1}, "new": "provider"}, "replace": "new", } assert cast(Any, buffered_tool_call).extra_content == { "google": {"thought_signature": "sig-1", "stable": {"value": "kept"}, "new": "extra"} } def test_stream_handler_internal_part_stores_text_and_type() -> None: part = Part(text="hello", type="output_text") assert part.text == "hello" assert part.type == "output_text" def test_merge_buffered_metadata_keeps_existing_scalar_when_empty_dict_arrives() -> None: merged = _merge_buffered_metadata( {"stable": "keep-me"}, {"stable": {}, "new": {}}, ) assert merged == {"stable": "keep-me", "new": {}} def test_stream_output_layout_rejects_unknown_function_call_index() -> None: layout = _StreamOutputLayout() with pytest.raises(KeyError, match="Function call index 9 has not been tracked"): layout.function_call_output_index(StreamingState(), 9) @pytest.mark.parametrize( ("buffered_call", "message"), [ ( _BufferedToolCall(index=0, name="my_func"), "without a tool call id", ), ( _BufferedToolCall(index=0, call_id="tool-id"), "without a function name", ), ], ) def test_buffered_tool_call_delta_requires_id_and_name( buffered_call: _BufferedToolCall, message: str, ) -> None: with pytest.raises(ModelBehaviorError, match=message): ChatCmplStreamHandler._buffered_tool_call_delta(buffered_call) def test_function_call_item_omits_provider_data_when_absent() -> None: function_call = ResponseFunctionToolCall( id="fake-id", call_id="call-id", arguments="", name="my_func", type="function_call", ) item = ChatCmplStreamHandler._function_call_item( StreamingState(), function_call, arguments="{}", ) assert item.arguments == "{}" assert "provider_data" not in item.model_dump() def test_finish_reasoning_summary_part_clears_invalid_active_index() -> None: reasoning_item = ResponseReasoningItem(id="fake-id", summary=[], type="reasoning") state = StreamingState( reasoning_content_index_and_output=(0, reasoning_item), active_reasoning_summary_index=0, ) events = list(ChatCmplStreamHandler._finish_reasoning_summary_part(state, SequenceNumber())) assert events == [] assert state.active_reasoning_summary_index is None @pytest.mark.asyncio async def test_buffer_tool_call_stream_preserves_empty_choice_chunks() -> None: chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[], ) buffered_chunks = await _collect_buffered_tool_call_chunks(chunk) assert buffered_chunks == [chunk] @pytest.mark.asyncio async def test_buffer_tool_call_stream_keeps_passthrough_index_passthrough() -> None: custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( index=0, id="custom-id", type="custom", ) function_tool_call_delta = ChoiceDeltaToolCall( index=0, id="function-id", function=ChoiceDeltaToolCallFunction(name="my_func", arguments="{}"), type="function", ) chunk1 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[custom_tool_call_delta]))], ) chunk2 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[function_tool_call_delta]))], ) buffered_chunks = await _collect_buffered_tool_call_chunks(chunk1, chunk2) assert len(buffered_chunks) == 2 assert buffered_chunks[0].choices[0].delta.tool_calls == [custom_tool_call_delta] assert buffered_chunks[1].choices[0].delta.tool_calls == [function_tool_call_delta] @pytest.mark.parametrize( ("delta", "expected"), [ (None, False), (ChoiceDelta(), False), (ChoiceDelta(content="text"), True), (ChoiceDelta.model_construct(refusal="blocked"), True), (ChoiceDelta.model_construct(reasoning_content="summary"), True), (ChoiceDelta.model_construct(reasoning="scratchpad"), True), (ChoiceDelta.model_construct(thinking_blocks=[{"thinking": "hidden"}]), True), ], ) def test_stream_handler_detects_passthrough_delta_shapes( delta: ChoiceDelta | None, expected: bool, ) -> None: assert ChatCmplStreamHandler._delta_has_passthrough_output(delta) is expected @pytest.mark.asyncio async def test_stream_handler_ignores_choice_without_delta() -> None: chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice.model_construct(index=0, delta=None)], ) events = await _collect_handler_events(chunk) assert [event.type for event in events] == ["response.created", "response.completed"] completed_event = events[-1] assert isinstance(completed_event, ResponseCompletedEvent) assert completed_event.response.output == [] @pytest.mark.asyncio async def test_stream_handler_converts_third_party_reasoning_text() -> None: reasoning_delta1 = ChoiceDelta.model_construct(reasoning="think ") reasoning_delta2 = ChoiceDelta.model_construct(reasoning="hard") chunks = [ ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=reasoning_delta1)], ), ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=reasoning_delta2)], ), ] events = await _collect_handler_events(*chunks, model="third-party") reasoning_delta_events = [ event for event in events if event.type == "response.reasoning_text.delta" ] assert [event.delta for event in reasoning_delta_events] == ["think ", "hard"] reasoning_done_event = next( event for event in events if event.type == "response.output_item.done" and isinstance(event.item, ResponseReasoningItem) ) reasoning_done_item = cast(ResponseReasoningItem, reasoning_done_event.item) assert reasoning_done_item.content assert cast(Any, reasoning_done_item.content[0]).text == "think hard" completed_event = next(event for event in events if event.type == "response.completed") assert isinstance(completed_event, ResponseCompletedEvent) completed_reasoning_item = completed_event.response.output[0] assert isinstance(completed_reasoning_item, ResponseReasoningItem) assert completed_reasoning_item.content assert cast(Any, completed_reasoning_item.content[0]).text == "think hard" assert completed_reasoning_item.model_dump().get("provider_data") == { "model": "third-party", "response_id": "chunk-id", } @pytest.mark.asyncio async def test_stream_handler_preserves_thinking_blocks_with_reasoning_summary() -> None: delta = ChoiceDelta.model_construct( reasoning_content="summary", thinking_blocks=[ {"thinking": "hidden one ", "signature": "sig-1"}, {"thinking": "hidden two", "signature": "sig-2"}, ], ) chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=delta)], ) events = await _collect_handler_events(chunk) completed_event = next(event for event in events if event.type == "response.completed") reasoning_item = completed_event.response.output[0] assert isinstance(reasoning_item, ResponseReasoningItem) assert reasoning_item.summary[0].text == "summary" assert reasoning_item.content assert cast(Any, reasoning_item.content[0]).text == "hidden one hidden two" assert reasoning_item.encrypted_content == "sig-2" @pytest.mark.asyncio async def test_stream_handler_adds_third_party_reasoning_text_to_summary_item() -> None: chunks = [ ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[ Choice(index=0, delta=ChoiceDelta.model_construct(reasoning_content="summary")) ], ), ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta.model_construct(reasoning="details"))], ), ] events = await _collect_handler_events(*chunks) completed_event = next(event for event in events if event.type == "response.completed") reasoning_item = completed_event.response.output[0] assert isinstance(reasoning_item, ResponseReasoningItem) assert reasoning_item.summary[0].text == "summary" assert reasoning_item.content assert cast(Any, reasoning_item.content[0]).text == "details" @pytest.mark.asyncio async def test_stream_handler_orders_refusal_after_reasoning_and_text() -> None: chunks = [ ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[ Choice(index=0, delta=ChoiceDelta.model_construct(reasoning_content="summary")) ], ), ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(content="partial"))], ), ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta.model_construct(refusal="blocked"))], ), ] events = await _collect_handler_events(*chunks) completed_event = next(event for event in events if event.type == "response.completed") assistant_item = completed_event.response.output[1] assert isinstance(assistant_item, ResponseOutputMessage) assert isinstance(assistant_item.content[0], ResponseOutputText) assert isinstance(assistant_item.content[1], ResponseOutputRefusal) assert assistant_item.content[0].text == "partial" assert assistant_item.content[1].refusal == "blocked" @pytest.mark.asyncio async def test_stream_handler_places_text_after_existing_refusal_part() -> None: chunks = [ ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta.model_construct(refusal="blocked"))], ), ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(content="partial"))], ), ] events = await _collect_handler_events(*chunks) text_part_added = next( event for event in events if event.type == "response.content_part.added" and isinstance(event.part, ResponseOutputText) ) assert text_part_added.content_index == 1 completed_event = next(event for event in events if event.type == "response.completed") assistant_item = completed_event.response.output[0] assert isinstance(assistant_item, ResponseOutputMessage) assert isinstance(assistant_item.content[0], ResponseOutputText) assert isinstance(assistant_item.content[1], ResponseOutputRefusal) assert assistant_item.content[0].text == "partial" assert assistant_item.content[1].refusal == "blocked" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_passes_strict_validation_to_stream_handler(monkeypatch) -> None: chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=1, delta=ChoiceDelta(content="ignored"))], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, strict_feature_validation=True, ).get_model("gpt-4") with pytest.raises(UserError, match="multiple choices or nonzero"): 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, ): pass @pytest.mark.allow_call_model_methods @pytest.mark.asyncio @pytest.mark.parametrize( ("previous_response_id", "conversation_id", "expected_param"), [ ("resp_123", None, "previous_response_id"), (None, "conv_123", "conversation_id"), ], ) async def test_stream_response_warns_and_ignores_server_managed_conversation_state_by_default( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, previous_response_id: str | None, conversation_id: str | None, expected_param: str, ) -> None: called = False async def patched_fetch_response(self, *args, **kwargs): nonlocal called called = True return _empty_response(), _empty_chat_completion_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider(use_responses=False).get_model("gpt-4") caplog.set_level(logging.WARNING, logger="openai.agents") 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=previous_response_id, conversation_id=conversation_id, prompt=None, ): pass assert expected_param in caplog.text assert "Ignoring unsupported server-managed conversation state" in caplog.text assert called is True @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_warns_and_ignores_prompt_by_default( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: captured_prompt: Any = None async def patched_fetch_response(self, *args, **kwargs): nonlocal captured_prompt captured_prompt = kwargs.get("prompt") return _empty_response(), _empty_chat_completion_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider(use_responses=False).get_model("gpt-4") caplog.set_level(logging.WARNING, logger="openai.agents") async for _ 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=cast(Any, {"id": "pmpt_123"}), ): pass assert "Reusable prompts are only supported by the Responses API" in caplog.text assert "Ignoring `prompt`" in caplog.text assert captured_prompt is None @pytest.mark.allow_call_model_methods @pytest.mark.asyncio @pytest.mark.parametrize( ("previous_response_id", "conversation_id", "expected_param"), [ ("resp_123", None, "previous_response_id"), (None, "conv_123", "conversation_id"), ], ) async def test_stream_response_rejects_server_managed_conversation_state_in_strict_mode( monkeypatch: pytest.MonkeyPatch, previous_response_id: str | None, conversation_id: str | None, expected_param: str, ) -> None: called = False async def patched_fetch_response(self, *args, **kwargs): nonlocal called called = True raise AssertionError("_fetch_response should not be called") monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, strict_feature_validation=True, ).get_model("gpt-4") with pytest.raises(UserError, match="server-managed conversation state") as exc_info: 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=previous_response_id, conversation_id=conversation_id, prompt=None, ): pass assert expected_param in str(exc_info.value) assert called is False @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_rejects_prompt_in_strict_mode(monkeypatch) -> None: async def patched_fetch_response(self, *args, **kwargs): raise AssertionError("_fetch_response should not run when prompt is unsupported") monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, strict_feature_validation=True, ).get_model("gpt-4") with pytest.raises(UserError, match="Reusable prompts"): async for _ 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=cast(Any, {"id": "pmpt_123"}), ): pass @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_includes_logprobs(monkeypatch) -> None: chunk1 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[ Choice( index=0, delta=ChoiceDelta(content="Hi"), logprobs=ChoiceLogprobs( content=[ ChatCompletionTokenLogprob( token="Hi", logprob=-0.5, bytes=[1], top_logprobs=[TopLogprob(token="Hi", logprob=-0.5, bytes=[1])], ) ] ), ) ], ) chunk2 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[ Choice( index=0, delta=ChoiceDelta(content=" there"), logprobs=ChoiceLogprobs( content=[ ChatCompletionTokenLogprob( token=" there", logprob=-0.25, bytes=[2], top_logprobs=[TopLogprob(token=" there", logprob=-0.25, bytes=[2])], ) ] ), ) ], usage=CompletionUsage( completion_tokens=5, prompt_tokens=7, total_tokens=12, prompt_tokens_details=PromptTokensDetails(cached_tokens=2), completion_tokens_details=CompletionTokensDetails(reasoning_tokens=3), ), ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2): yield c 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, 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) text_delta_events = [ event for event in output_events if event.type == "response.output_text.delta" ] assert len(text_delta_events) == 2 assert [lp.token for lp in text_delta_events[0].logprobs] == ["Hi"] assert [lp.token for lp in text_delta_events[1].logprobs] == [" there"] completed_event = next(event for event in output_events if event.type == "response.completed") assert isinstance(completed_event, ResponseCompletedEvent) completed_resp = completed_event.response assert isinstance(completed_resp.output[0], ResponseOutputMessage) text_part = completed_resp.output[0].content[0] assert isinstance(text_part, ResponseOutputText) assert text_part.text == "Hi there" assert text_part.logprobs is not None assert [lp.token for lp in text_part.logprobs] == ["Hi", " there"] @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_accumulates_logprobs_across_many_deltas(monkeypatch) -> None: # Each content delta carries its own logprobs, and the streamed output text part must # accumulate all of them in order across the whole stream. tokens = ["a", "b", "c", "d", "e"] def make_chunk(token: str) -> ChatCompletionChunk: return ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[ Choice( index=0, delta=ChoiceDelta(content=token), logprobs=ChoiceLogprobs( content=[ ChatCompletionTokenLogprob( token=token, logprob=-0.5, bytes=[1], top_logprobs=[TopLogprob(token=token, logprob=-0.5, bytes=[1])], ) ] ), ) ], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for token in tokens: yield make_chunk(token) 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, 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) completed_event = next(event for event in output_events if event.type == "response.completed") assert isinstance(completed_event, ResponseCompletedEvent) completed_resp = completed_event.response assert isinstance(completed_resp.output[0], ResponseOutputMessage) text_part = completed_resp.output[0].content[0] assert isinstance(text_part, ResponseOutputText) assert text_part.text == "".join(tokens) assert text_part.logprobs is not None assert [lp.token for lp in text_part.logprobs] == tokens @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_yields_events_for_refusal_content(monkeypatch) -> None: """ Validate that when the model streams a refusal string instead of normal content, `stream_response` emits the appropriate sequence of events including `response.refusal.delta` events for each chunk of the refusal message and constructs a completed assistant message with a `ResponseOutputRefusal` part. """ # Simulate refusal text coming in two pieces, like content but using the `refusal` # field on the delta rather than `content`. chunk1 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(refusal="No"))], ) chunk2 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(refusal="Thanks"))], usage=CompletionUsage(completion_tokens=2, prompt_tokens=2, total_tokens=4), ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2): yield c 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, 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) # Expect sequence similar to text: created, output_item.added, content part added, # two refusal delta events, content part done, output_item.done, completed. assert len(output_events) == 8 assert output_events[0].type == "response.created" assert output_events[1].type == "response.output_item.added" assert output_events[2].type == "response.content_part.added" assert output_events[3].type == "response.refusal.delta" assert output_events[3].delta == "No" assert output_events[4].type == "response.refusal.delta" assert output_events[4].delta == "Thanks" assert output_events[5].type == "response.content_part.done" assert output_events[6].type == "response.output_item.done" assert output_events[7].type == "response.completed" completed_resp = output_events[7].response assert isinstance(completed_resp.output[0], ResponseOutputMessage) refusal_part = completed_resp.output[0].content[0] assert isinstance(refusal_part, ResponseOutputRefusal) assert refusal_part.refusal == "NoThanks" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_yields_events_for_tool_call(monkeypatch) -> None: """ Validate that `stream_response` emits the correct sequence of events when the model is streaming a function/tool call instead of plain text. The function call will be split across two chunks. """ # Simulate a single tool call with complete function name in first chunk # and arguments split across chunks (reflecting real OpenAI API behavior) tool_call_delta1 = ChoiceDeltaToolCall( index=0, id="tool-id", function=ChoiceDeltaToolCallFunction(name="my_func", arguments="arg1"), type="function", ) tool_call_delta2 = ChoiceDeltaToolCall( index=0, id="tool-id", function=ChoiceDeltaToolCallFunction(name=None, arguments="arg2"), 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 c in (chunk1, chunk2): yield c 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, 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) # Sequence should be: response.created, then after loop we expect function call-related events: # one response.output_item.added for function call, a response.function_call_arguments.delta, # a response.output_item.done, and finally response.completed. assert output_events[0].type == "response.created" # The next three events are about the tool call. assert output_events[1].type == "response.output_item.added" # The added item should be a ResponseFunctionToolCall. added_fn = output_events[1].item assert isinstance(added_fn, ResponseFunctionToolCall) assert added_fn.name == "my_func" # Name should be complete from first chunk assert added_fn.arguments == "" # Arguments start empty assert output_events[2].type == "response.function_call_arguments.delta" assert output_events[2].delta == "arg1" # First argument chunk assert output_events[3].type == "response.function_call_arguments.delta" assert output_events[3].delta == "arg2" # Second argument chunk assert output_events[4].type == "response.output_item.done" assert output_events[5].type == "response.completed" # Final function call should have complete arguments final_fn = output_events[4].item assert isinstance(final_fn, ResponseFunctionToolCall) assert final_fn.name == "my_func" assert final_fn.arguments == "arg1arg2" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_buffers_tool_call_deltas_when_enabled(monkeypatch) -> None: tool_call_delta1 = ChoiceDeltaToolCall( index=0, id="tool-id", function=ChoiceDeltaToolCallFunction(name="my_func", arguments="arg1"), type="function", ) tool_call_delta2 = ChoiceDeltaToolCall( index=0, id=None, function=ChoiceDeltaToolCallFunction(name=None, arguments="arg2"), 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): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, buffer_streamed_tool_calls=True, ).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) argument_delta_events = [ event for event in output_events if event.type == "response.function_call_arguments.delta" ] assert len(argument_delta_events) == 1 assert argument_delta_events[0].delta == "arg1arg2" done_event = next(event for event in output_events if event.type == "response.output_item.done") final_fn = done_event.item assert isinstance(final_fn, ResponseFunctionToolCall) assert final_fn.call_id == "tool-id" assert final_fn.name == "my_func" assert final_fn.arguments == "arg1arg2" completed_event = next(event for event in output_events if event.type == "response.completed") assert isinstance(completed_event, ResponseCompletedEvent) assert completed_event.response.usage assert completed_event.response.usage.total_tokens == 2 @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_buffered_tool_call_before_text_replays_as_single_assistant_session_message() -> None: tool_call_delta = ChoiceDeltaToolCall( index=0, id="call_lookup_status", function=ChoiceDeltaToolCallFunction(name="lookup_status", arguments="{}"), type="function", ) tool_first_chunk = ChatCompletionChunk( id="chunk-tool", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[tool_call_delta]))], ) later_text_chunk = ChatCompletionChunk( id="chunk-text", created=1, model="fake", object="chat.completion.chunk", choices=[ Choice( index=0, delta=ChoiceDelta(content="I'll look that up first."), ) ], usage=CompletionUsage(completion_tokens=5, prompt_tokens=5, total_tokens=10), ) final_text_chunk = ChatCompletionChunk( id="chunk-final", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(content="first run done"))], usage=CompletionUsage(completion_tokens=3, prompt_tokens=7, total_tokens=10), ) async def first_turn_stream() -> AsyncIterator[ChatCompletionChunk]: yield tool_first_chunk yield later_text_chunk async def final_turn_stream() -> AsyncIterator[ChatCompletionChunk]: yield final_text_chunk class DummyCompletions: def __init__(self) -> None: self.calls: list[dict[str, Any]] = [] async def create(self, **kwargs: Any) -> Any: self.calls.append(kwargs) call_number = len(self.calls) if kwargs["stream"] is True: if call_number == 1: return first_turn_stream() if call_number == 2: return final_turn_stream() raise AssertionError(f"Unexpected streamed call {call_number}") return ChatCompletion( id="resp-id", created=0, model="fake", object="chat.completion", choices=[ ChatCompletionChoice( index=0, finish_reason="stop", message=ChatCompletionMessage( role="assistant", content="second run done", ), ) ], usage=None, ) class DummyClient: def __init__(self, completions: DummyCompletions) -> None: self.chat = type("_Chat", (), {"completions": completions})() self.base_url = "http://fake" def lookup_status() -> str: return "lookup result" completions = DummyCompletions() model = OpenAIChatCompletionsModel( model="gpt-4", openai_client=DummyClient(completions), # type: ignore[arg-type] buffer_streamed_tool_calls=True, ) agent = Agent( name="test", model=model, tools=[function_tool(lookup_status, name_override="lookup_status")], ) session = SimpleListSession() first_result = Runner.run_streamed(agent, input="first question", session=session) async for _ in first_result.stream_events(): pass assert first_result.final_output == "first run done" await Runner.run(agent, input="second question", session=session) assert len(completions.calls) == 3 replayed_messages = completions.calls[2]["messages"] assert [message["role"] for message in replayed_messages] == [ "user", "assistant", "tool", "assistant", "user", ] assistant_with_tool = cast(dict[str, Any], replayed_messages[1]) assert assistant_with_tool["content"] == "I'll look that up first." assert len(assistant_with_tool["tool_calls"]) == 1 tool_call = assistant_with_tool["tool_calls"][0] assert tool_call["id"] == "call_lookup_status" assert tool_call["function"] == {"name": "lookup_status", "arguments": "{}"} tool_message = cast(dict[str, Any], replayed_messages[2]) assert tool_message["tool_call_id"] == "call_lookup_status" assert tool_message["content"] == "lookup result" assert replayed_messages[3]["content"] == "first run done" assert replayed_messages[4]["content"] == "second question" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_buffers_tool_call_usage_chunk_without_replay( monkeypatch, ) -> None: tool_call_delta = ChoiceDeltaToolCall( index=0, id="tool-id", function=ChoiceDeltaToolCallFunction(name="my_func", arguments="arg1"), type="function", ) chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[tool_call_delta]))], usage=CompletionUsage(completion_tokens=1, prompt_tokens=1, total_tokens=2), ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, buffer_streamed_tool_calls=True, ).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) argument_delta_events = [ event for event in output_events if event.type == "response.function_call_arguments.delta" ] assert len(argument_delta_events) == 1 assert argument_delta_events[0].delta == "arg1" function_done_events = [ event for event in output_events if event.type == "response.output_item.done" and isinstance(event.item, ResponseFunctionToolCall) ] assert len(function_done_events) == 1 completed_event = next(event for event in output_events if event.type == "response.completed") assert isinstance(completed_event, ResponseCompletedEvent) assert completed_event.response.usage assert completed_event.response.usage.total_tokens == 2 @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_buffers_tool_call_provider_fields(monkeypatch) -> None: tool_call_delta1 = ChoiceDeltaToolCall( index=0, id="tool-id", function=ChoiceDeltaToolCallFunction(name="my_func", arguments=None), type="function", ) cast(Any, tool_call_delta1).provider_specific_fields = {"thought_signature": "thought-sig"} tool_call_delta2 = ChoiceDeltaToolCall( index=0, id=None, function=ChoiceDeltaToolCallFunction(name=None, arguments="arg1"), type="function", ) chunk1 = ChatCompletionChunk( id="chunk-id", created=1, model="gemini/gemini-3-pro", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[tool_call_delta1]))], ) chunk2 = ChatCompletionChunk( id="chunk-id", created=1, model="gemini/gemini-3-pro", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[tool_call_delta2]))], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for chunk in (chunk1, chunk2): yield chunk async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, buffer_streamed_tool_calls=True, ).get_model("gemini/gemini-3-pro") 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) function_done_events = [ event for event in output_events if event.type == "response.output_item.done" and isinstance(event.item, ResponseFunctionToolCall) ] assert len(function_done_events) == 1 provider_data = function_done_events[0].item.model_dump().get("provider_data", {}) assert provider_data["thought_signature"] == "thought-sig" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_buffered_tool_calls_raise_for_missing_tool_call_delta( monkeypatch, ) -> None: chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, buffer_streamed_tool_calls=True, ).get_model("gpt-4") with pytest.raises(ModelBehaviorError, match="finish_reason='tool_calls'"): 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, ): pass @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_buffered_tool_calls_preserve_nonzero_choice_validation(monkeypatch) -> None: tool_call_delta = ChoiceDeltaToolCall( index=0, id="tool-id", function=ChoiceDeltaToolCallFunction(name="my_func", arguments="arg"), type="function", ) chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=1, delta=ChoiceDelta(tool_calls=[tool_call_delta]))], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, strict_feature_validation=True, buffer_streamed_tool_calls=True, ).get_model("gpt-4") with pytest.raises(UserError, match="multiple choices or nonzero choice indexes"): 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, ): pass @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_buffered_tool_calls_do_not_merge_nonzero_choice_tool_call_indexes( monkeypatch, ) -> None: choice_zero_tool_call = ChoiceDeltaToolCall( index=0, id="choice-zero-tool-id", function=ChoiceDeltaToolCallFunction(name="choice_zero_func", arguments="choice-zero"), type="function", ) choice_one_tool_call = ChoiceDeltaToolCall( index=0, id="choice-one-tool-id", function=ChoiceDeltaToolCallFunction(name="choice_one_func", arguments="choice-one"), type="function", ) chunk = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[ Choice(index=0, delta=ChoiceDelta(tool_calls=[choice_zero_tool_call])), Choice(index=1, delta=ChoiceDelta(tool_calls=[choice_one_tool_call])), ], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, buffer_streamed_tool_calls=True, ).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) function_done_events = [ event for event in output_events if event.type == "response.output_item.done" and isinstance(event.item, ResponseFunctionToolCall) ] assert len(function_done_events) == 1 final_fn = function_done_events[0].item assert isinstance(final_fn, ResponseFunctionToolCall) assert final_fn.call_id == "choice-zero-tool-id" assert final_fn.name == "choice_zero_func" assert final_fn.arguments == "choice-zero" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_buffered_tool_calls_preserve_custom_tool_call_strict_error( monkeypatch, ) -> None: custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( index=0, id="tool-call-123", type="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]), finish_reason="tool_calls", ) ], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, strict_feature_validation=True, buffer_streamed_tool_calls=True, ).get_model("gpt-4") with pytest.raises(UserError, match="Custom tool calls are not supported"): 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, ): pass @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_buffered_tool_calls_ignore_custom_tool_call_by_default(monkeypatch) -> None: custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( index=0, id="tool-call-123", type="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]), finish_reason="tool_calls", ) ], ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider( use_responses=False, buffer_streamed_tool_calls=True, ).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) completed_event = next(event for event in output_events if event.type == "response.completed") assert isinstance(completed_event, ResponseCompletedEvent) assert completed_event.response.output == [] @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_with_custom_tool_call_raises_in_strict_mode(monkeypatch) -> None: custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( index=0, id="tool-call-123", type="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 fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk 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, fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider(use_responses=False, strict_feature_validation=True).get_model("gpt-4") with pytest.raises(UserError, match="Custom tool calls are not supported"): 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, ): pass @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_ignores_custom_tool_call_chunks_by_default(monkeypatch) -> None: custom_tool_call_delta = ChoiceDeltaToolCall.model_construct( index=0, id="tool-call-123", type="custom", ) omitted_type_tool_call_delta = ChoiceDeltaToolCall.model_construct( index=0, function=ChoiceDeltaToolCallFunction(name="custom_tool", arguments="payload"), ) chunks = [ ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[custom_tool_call_delta]))], ), ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[omitted_type_tool_call_delta]))], ), ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(content="done"))], ), ] async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for chunk in chunks: yield chunk async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider(use_responses=False).get_model("gpt-4") 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, ): events.append(event) function_call_events = [] for event in events: item = getattr(event, "item", None) if isinstance(item, ResponseFunctionToolCall): function_call_events.append(event) assert function_call_events == [] completed_event = events[-1] assert isinstance(completed_event, ResponseCompletedEvent) assert all( not isinstance(item, ResponseFunctionToolCall) for item in completed_event.response.output ) assert len(completed_event.response.output) == 1 message = completed_event.response.output[0] assert isinstance(message, ResponseOutputMessage) assert len(message.content) == 1 assert isinstance(message.content[0], ResponseOutputText) assert message.content[0].text == "done" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_yields_real_time_function_call_arguments(monkeypatch) -> None: """ Validate that `stream_response` emits function call arguments in real-time as they are received, not just at the end. This test simulates the real OpenAI API behavior where function name comes first, then arguments are streamed incrementally. """ # Simulate realistic OpenAI API chunks: name first, then arguments incrementally tool_call_delta1 = ChoiceDeltaToolCall( index=0, id="tool-call-123", function=ChoiceDeltaToolCallFunction(name="write_file", arguments=""), type="function", ) tool_call_delta2 = ChoiceDeltaToolCall( index=0, function=ChoiceDeltaToolCallFunction(arguments='{"filename": "'), type="function", ) tool_call_delta3 = ChoiceDeltaToolCall( index=0, function=ChoiceDeltaToolCallFunction(arguments='test.py", "content": "'), type="function", ) tool_call_delta4 = ChoiceDeltaToolCall( index=0, function=ChoiceDeltaToolCallFunction(arguments='print(hello)"}'), 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]))], ) chunk3 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[tool_call_delta3]))], ) chunk4 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[tool_call_delta4]))], usage=CompletionUsage(completion_tokens=1, prompt_tokens=1, total_tokens=2), ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2, chunk3, chunk4): yield c 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, 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) # Extract events by type created_events = [e for e in output_events if e.type == "response.created"] output_item_added_events = [e for e in output_events if e.type == "response.output_item.added"] function_args_delta_events = [ e for e in output_events if e.type == "response.function_call_arguments.delta" ] output_item_done_events = [e for e in output_events if e.type == "response.output_item.done"] completed_events = [e for e in output_events if e.type == "response.completed"] # Verify event structure assert len(created_events) == 1 assert len(output_item_added_events) == 1 assert len(function_args_delta_events) == 3 # Three incremental argument chunks assert len(output_item_done_events) == 1 assert len(completed_events) == 1 # Verify the function call started as soon as we had name and ID added_event = output_item_added_events[0] assert isinstance(added_event.item, ResponseFunctionToolCall) assert added_event.item.name == "write_file" assert added_event.item.call_id == "tool-call-123" assert added_event.item.arguments == "" # Should be empty at start # Verify real-time argument streaming expected_deltas = ['{"filename": "', 'test.py", "content": "', 'print(hello)"}'] for i, delta_event in enumerate(function_args_delta_events): assert delta_event.delta == expected_deltas[i] assert delta_event.item_id == "__fake_id__" # FAKE_RESPONSES_ID assert delta_event.output_index == 0 # Verify completion event has full arguments done_event = output_item_done_events[0] assert isinstance(done_event.item, ResponseFunctionToolCall) assert done_event.item.name == "write_file" assert done_event.item.arguments == '{"filename": "test.py", "content": "print(hello)"}' # Verify final response completed_event = completed_events[0] function_call_output = completed_event.response.output[0] 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_fallback_function_calls_have_unique_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 c in (chunk1, chunk2): yield c 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, 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_indexes = [ event.output_index for event in output_events if event.type == "response.output_item.added" ] delta_indexes = [ event.output_index for event in output_events if event.type == "response.function_call_arguments.delta" ] done_indexes = [ event.output_index for event in output_events if event.type == "response.output_item.done" ] assert added_indexes == [0, 1] assert delta_indexes == [0, 1] assert done_indexes == [0, 1] @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_fallback_function_call_keeps_index_before_streamed_call(monkeypatch) -> None: fallback_first = ChoiceDeltaToolCall( index=0, function=ChoiceDeltaToolCallFunction( name="fallback_first", arguments='{"a": 1}', ), type="function", ) streamed_second_start = ChoiceDeltaToolCall( index=1, id="tool-call-2", function=ChoiceDeltaToolCallFunction( name="streamed_second", arguments="", ), type="function", ) streamed_second_args = ChoiceDeltaToolCall( index=1, function=ChoiceDeltaToolCallFunction(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_first]))], ) chunk2 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[streamed_second_start]))], ) chunk3 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[streamed_second_args]))], usage=CompletionUsage(completion_tokens=1, prompt_tokens=1, total_tokens=2), ) async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2, chunk3): yield c 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, 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) completed = next( event.response for event in output_events if event.type == "response.completed" ) assert [ item.name for item in completed.output if isinstance(item, ResponseFunctionToolCall) ] == [ "fallback_first", "streamed_second", ] added_by_name = { event.item.name: event.output_index for event in output_events if event.type == "response.output_item.added" and isinstance(event.item, ResponseFunctionToolCall) } delta_indexes = [ event.output_index for event in output_events if event.type == "response.function_call_arguments.delta" ] done_by_name = { event.item.name: event.output_index for event in output_events if event.type == "response.output_item.done" and isinstance(event.item, ResponseFunctionToolCall) } assert added_by_name == {"fallback_first": 0, "streamed_second": 1} assert delta_indexes == [1, 0] assert done_by_name == {"streamed_second": 1, "fallback_first": 0} @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_fallback_function_call_before_text_uses_final_output_index( monkeypatch, ) -> None: fallback_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_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) @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_streamed_function_call_before_text_keeps_realtime_order( monkeypatch, ) -> None: streamed_call_start = ChoiceDeltaToolCall( index=0, id="tool-call-1", function=ChoiceDeltaToolCallFunction(name="first_tool", arguments=""), type="function", ) streamed_call_args = ChoiceDeltaToolCall( index=0, function=ChoiceDeltaToolCallFunction(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_call_start]))], ) chunk2 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[streamed_call_args]))], ) chunk3 = 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, 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_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_tool_event.output_index == 0 assert added_message_event.output_index == 1 assert [event.output_index for event in delta_events] == [0] assert done_tool_event.output_index == 0 assert done_message_event.output_index == 1 assert isinstance(completed_event.response.output[0], ResponseFunctionToolCall) assert isinstance(completed_event.response.output[1], ResponseOutputMessage) @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_mixed_function_calls_before_text_keep_tracked_order( monkeypatch, ) -> None: fallback_first = ChoiceDeltaToolCall( index=0, function=ChoiceDeltaToolCallFunction(name="fallback_first", arguments='{"a": 1}'), type="function", ) streamed_second_start = ChoiceDeltaToolCall( index=1, id="tool-call-2", function=ChoiceDeltaToolCallFunction(name="streamed_second", arguments=""), type="function", ) streamed_second_args = ChoiceDeltaToolCall( index=1, function=ChoiceDeltaToolCallFunction(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_first]))], ) chunk2 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[streamed_second_start]))], ) chunk3 = ChatCompletionChunk( id="chunk-id", created=1, model="fake", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(tool_calls=[streamed_second_args]))], ) chunk4 = 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, chunk3, chunk4): 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" ] 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_indexes = { event.item.name: event.output_index for event in added_events if isinstance(event.item, ResponseFunctionToolCall) } assert added_tool_indexes == {"streamed_second": 1, "fallback_first": 0} assert added_message_event.output_index == 2 assert {event.delta: event.output_index for event in delta_events} == { '{"b": 2}': 1, '{"a": 1}': 0, } assert isinstance(completed_event.response.output[0], ResponseFunctionToolCall) assert isinstance(completed_event.response.output[1], ResponseFunctionToolCall) assert isinstance(completed_event.response.output[2], ResponseOutputMessage)