from __future__ import annotations import importlib import sys import types as pytypes from collections.abc import AsyncIterator from typing import Any, Literal, cast import pytest from openai.types.chat import ( ChatCompletion, ChatCompletionChunk, ChatCompletionMessage, ChatCompletionMessageFunctionToolCall, ) from openai.types.chat.chat_completion import Choice from openai.types.chat.chat_completion_chunk import ChoiceDelta from openai.types.completion_usage import CompletionUsage, PromptTokensDetails from openai.types.responses import Response, ResponseCompletedEvent, ResponseOutputMessage from openai.types.responses.response_error_event import ResponseErrorEvent from openai.types.responses.response_failed_event import ResponseFailedEvent from openai.types.responses.response_incomplete_event import ResponseIncompleteEvent from openai.types.responses.response_output_text import ResponseOutputText from openai.types.responses.response_usage import ( InputTokensDetails, OutputTokensDetails, ResponseUsage, ) from pydantic import BaseModel from agents import ( Agent, Handoff, ModelBehaviorError, ModelSettings, ModelTracing, Tool, TResponseInputItem, __version__, ) from agents.exceptions import UserError from agents.models.chatcmpl_helpers import HEADERS_OVERRIDE from agents.models.fake_id import FAKE_RESPONSES_ID class FakeAnyLLMProvider: def __init__( self, *, supports_responses: bool, chat_response: Any | None = None, responses_response: Any | None = None, ) -> None: self.SUPPORTS_RESPONSES = supports_responses self.chat_response = chat_response self.responses_response = responses_response self.chat_calls: list[dict[str, Any]] = [] self.responses_calls: list[dict[str, Any]] = [] self.private_responses_calls: list[dict[str, Any]] = [] async def acompletion(self, **kwargs: Any) -> Any: self.chat_calls.append(kwargs) return self.chat_response async def aresponses(self, **kwargs: Any) -> Any: self.responses_calls.append(kwargs) return self.responses_response async def _aresponses(self, params: Any, **kwargs: Any) -> Any: self.private_responses_calls.append({"params": params, "kwargs": kwargs}) return self.responses_response def _import_any_llm_module( monkeypatch: pytest.MonkeyPatch, provider: FakeAnyLLMProvider, ) -> tuple[Any, list[dict[str, Any]]]: create_calls: list[dict[str, Any]] = [] class FakeAnyLLMFactory: @staticmethod def create(provider_name: str, api_key: str | None = None, api_base: str | None = None): create_calls.append( { "provider_name": provider_name, "api_key": api_key, "api_base": api_base, } ) return provider fake_any_llm: Any = pytypes.ModuleType("any_llm") fake_any_llm.AnyLLM = FakeAnyLLMFactory sys.modules.pop("agents.extensions.models.any_llm_model", None) monkeypatch.setitem(sys.modules, "any_llm", fake_any_llm) module = importlib.import_module("agents.extensions.models.any_llm_model") monkeypatch.setattr(module, "AnyLLM", FakeAnyLLMFactory, raising=True) return module, create_calls def _chat_completion(text: str) -> ChatCompletion: return ChatCompletion( id="chatcmpl_123", created=0, model="fake-model", object="chat.completion", choices=[ Choice( index=0, finish_reason="stop", message=ChatCompletionMessage(role="assistant", content=text), ) ], usage=CompletionUsage( completion_tokens=5, prompt_tokens=7, total_tokens=12, prompt_tokens_details=PromptTokensDetails.model_validate( {"cached_tokens": 2, "cache_write_tokens": 4} ), ), ) def _responses_output(text: str) -> list[Any]: return [ ResponseOutputMessage( id="msg_123", role="assistant", status="completed", type="message", content=[ ResponseOutputText( text=text, type="output_text", annotations=[], logprobs=[], ) ], ) ] def _response(text: str, response_id: str = "resp_123") -> Response: return Response( id=response_id, created_at=123, model="fake-model", object="response", output=_responses_output(text), tool_choice="none", tools=[], parallel_tool_calls=False, usage=ResponseUsage( input_tokens=11, output_tokens=13, total_tokens=24, input_tokens_details=InputTokensDetails.model_validate( {"cache_write_tokens": 0, "cached_tokens": 0} ), output_tokens_details=OutputTokensDetails(reasoning_tokens=0), ), ) def _chat_completion_with_tool_call(*, thought_signature: str) -> ChatCompletion: return ChatCompletion( id="chatcmpl_tool_123", created=0, model="fake-model", object="chat.completion", choices=[ Choice( index=0, finish_reason="tool_calls", message=ChatCompletionMessage( role="assistant", content="Calling a tool.", tool_calls=[ ChatCompletionMessageFunctionToolCall.model_validate( { "id": "call_123", "type": "function", "function": { "name": "get_weather", "arguments": '{"city":"Paris"}', }, "extra_content": { "google": {"thought_signature": thought_signature} }, } ) ], ), ) ], usage=CompletionUsage( completion_tokens=5, prompt_tokens=7, total_tokens=12, prompt_tokens_details=PromptTokensDetails(cached_tokens=0), ), ) class GenericChatCompletionPayload(BaseModel): id: str created: int model: str object: str choices: list[Any] usage: Any async def _empty_chat_stream() -> AsyncIterator[ChatCompletionChunk]: if False: yield ChatCompletionChunk( id="chunk_123", created=0, model="fake-model", object="chat.completion.chunk", choices=[Choice(index=0, delta=ChoiceDelta(), finish_reason=None)], ) @pytest.mark.allow_call_model_methods @pytest.mark.asyncio @pytest.mark.parametrize("override_ua", [None, "test_user_agent"]) async def test_user_agent_header_any_llm_chat(override_ua: str | None, monkeypatch) -> None: provider = FakeAnyLLMProvider(supports_responses=False, chat_response=_chat_completion("Hello")) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openrouter/openai/gpt-5.4-mini") expected_ua = override_ua or f"Agents/Python {__version__}" if override_ua is not None: token = HEADERS_OVERRIDE.set({"User-Agent": override_ua}) else: token = None try: await model.get_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id=None, conversation_id=None, prompt=None, ) finally: if token is not None: HEADERS_OVERRIDE.reset(token) assert provider.chat_calls[0]["extra_headers"]["User-Agent"] == expected_ua @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_chat_path_is_used_when_responses_are_unsupported(monkeypatch) -> None: provider = FakeAnyLLMProvider(supports_responses=False, chat_response=_chat_completion("Hello")) module, create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openrouter/openai/gpt-5.4-mini", api_key="router-key") response = await model.get_response( system_instructions="You are terse.", input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id="resp_prev", conversation_id="conv_123", prompt=None, ) assert create_calls == [ { "provider_name": "openrouter", "api_key": "router-key", "api_base": None, } ] assert len(provider.chat_calls) == 1 assert provider.responses_calls == [] assert provider.chat_calls[0]["model"] == "openai/gpt-5.4-mini" assert response.response_id is None assert response.output[0].content[0].text == "Hello" assert response.usage.input_tokens_details.cached_tokens == 2 assert getattr(response.usage.input_tokens_details, "cache_write_tokens", None) == 4 @pytest.mark.allow_call_model_methods @pytest.mark.asyncio @pytest.mark.parametrize( "chat_response", [ pytest.param(_chat_completion("Hello").model_dump(), id="dict"), pytest.param( GenericChatCompletionPayload.model_validate(_chat_completion("Hello").model_dump()), id="basemodel", ), ], ) async def test_any_llm_chat_path_normalizes_non_stream_payloads( monkeypatch, chat_response: Any, ) -> None: provider = FakeAnyLLMProvider(supports_responses=False, chat_response=chat_response) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openrouter/openai/gpt-5.4-mini") response = await model.get_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id=None, conversation_id=None, prompt=None, ) assert response.response_id is None assert response.output[0].content[0].text == "Hello" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_chat_path_preserves_gemini_tool_call_metadata(monkeypatch) -> None: provider = FakeAnyLLMProvider( supports_responses=False, chat_response=_chat_completion_with_tool_call(thought_signature="sig_123"), ) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="gemini/gemini-2.0-flash") response = await model.get_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id=None, conversation_id=None, prompt=None, ) function_calls = [ item for item in response.output if getattr(item, "type", None) == "function_call" ] assert len(function_calls) == 1 provider_data = function_calls[0].model_dump()["provider_data"] assert provider_data["model"] == "gemini/gemini-2.0-flash" assert provider_data["response_id"] == "chatcmpl_tool_123" assert provider_data["thought_signature"] == "sig_123" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_responses_path_is_used_when_supported(monkeypatch) -> None: provider = FakeAnyLLMProvider(supports_responses=True, responses_response=_response("Hello")) module, create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="gpt-5.4-mini", api_key="openai-key") response = await model.get_response( system_instructions="You are terse.", input="hi", model_settings=ModelSettings(store=True), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id="resp_prev", conversation_id="conv_123", prompt=None, ) assert create_calls == [ { "provider_name": "openai", "api_key": "openai-key", "api_base": None, } ] assert provider.chat_calls == [] assert provider.responses_calls == [] assert len(provider.private_responses_calls) == 1 params = provider.private_responses_calls[0]["params"] kwargs = provider.private_responses_calls[0]["kwargs"] assert params.model == "gpt-5.4-mini" assert params.previous_response_id == "resp_prev" assert params.conversation == "conv_123" assert kwargs["extra_headers"]["User-Agent"] == f"Agents/Python {__version__}" assert response.response_id == "resp_123" assert response.output[0].content[0].text == "Hello" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_can_force_chat_completions_when_responses_are_supported(monkeypatch) -> None: provider = FakeAnyLLMProvider( supports_responses=True, chat_response=_chat_completion("Hello from chat"), responses_response=_response("Hello from responses"), ) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openai/gpt-4.1-mini", api="chat_completions") response = await model.get_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id="resp_prev", conversation_id="conv_123", prompt=None, ) assert len(provider.chat_calls) == 1 assert provider.responses_calls == [] assert response.response_id is None assert response.output[0].content[0].text == "Hello from chat" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_forced_responses_errors_when_provider_does_not_support_it( monkeypatch, ) -> None: provider = FakeAnyLLMProvider(supports_responses=False, chat_response=_chat_completion("Hello")) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openrouter/openai/gpt-4.1-mini", api="responses") with pytest.raises(UserError, match="does not support the Responses API"): await model.get_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id=None, conversation_id=None, prompt=None, ) @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_stream_uses_chat_handler_when_responses_are_unsupported(monkeypatch) -> None: provider = FakeAnyLLMProvider(supports_responses=False, chat_response=_empty_chat_stream()) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel completed = ResponseCompletedEvent( type="response.completed", response=_response("Hello from stream"), sequence_number=1, ) async def fake_handle_stream(response, stream, model=None): assert model == "openrouter/openai/gpt-5.4-mini" async for _chunk in stream: pass yield completed monkeypatch.setattr(module.ChatCmplStreamHandler, "handle_stream", fake_handle_stream) model = AnyLLMModel(model="openrouter/openai/gpt-5.4-mini") events = [ event async for event in model.stream_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id=None, conversation_id=None, prompt=None, ) ] assert [event.type for event in events] == ["response.completed"] @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_stream_passthrough_uses_responses_when_supported(monkeypatch) -> None: async def response_stream() -> AsyncIterator[ResponseCompletedEvent]: yield ResponseCompletedEvent( type="response.completed", response=_response("Hello from responses stream"), sequence_number=1, ) provider = FakeAnyLLMProvider(supports_responses=True, responses_response=response_stream()) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openai/gpt-5.4-mini") events = [ event async for event in model.stream_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id="resp_prev", conversation_id="conv_123", prompt=None, ) ] assert [event.type for event in events] == ["response.completed"] assert provider.responses_calls == [] assert provider.private_responses_calls[0]["params"].previous_response_id == "resp_prev" assert provider.private_responses_calls[0]["params"].conversation == "conv_123" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio @pytest.mark.parametrize( ("terminal_event_type", "terminal_event_cls"), [ ("response.incomplete", ResponseIncompleteEvent), ("response.failed", ResponseFailedEvent), ], ) async def test_any_llm_responses_stream_rejects_failed_terminal_events( monkeypatch, terminal_event_type: str, terminal_event_cls: type[Any], ) -> None: async def response_stream() -> AsyncIterator[Any]: yield terminal_event_cls( type=terminal_event_type, response=_response("partial", response_id="resp-terminal"), sequence_number=1, ) provider = FakeAnyLLMProvider(supports_responses=True, responses_response=response_stream()) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openai/gpt-5.4-mini") events = [] with pytest.raises(ModelBehaviorError, match=terminal_event_type): async for event in model.stream_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id=None, conversation_id=None, prompt=None, ): events.append(event) assert len(events) == 1 assert events[0].type == terminal_event_type assert events[0].response.id == "resp-terminal" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_responses_stream_rejects_error_event(monkeypatch) -> None: async def response_stream() -> AsyncIterator[ResponseErrorEvent]: yield ResponseErrorEvent( type="error", code="invalid_request_error", message="bad request", param=None, sequence_number=1, ) provider = FakeAnyLLMProvider(supports_responses=True, responses_response=response_stream()) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openai/gpt-5.4-mini") events = [] with pytest.raises(ModelBehaviorError, match="invalid_request_error"): async for event in model.stream_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id=None, conversation_id=None, prompt=None, ): events.append(event) assert len(events) == 1 assert events[0].type == "error" assert events[0].code == "invalid_request_error" @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_responses_path_passes_transport_kwargs_via_private_provider_api( monkeypatch, ) -> None: provider = FakeAnyLLMProvider(supports_responses=True, responses_response=_response("Hello")) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openai/gpt-5.4-mini") await model.get_response( system_instructions=None, input="hi", model_settings=ModelSettings( extra_headers={"X-Test-Header": "test"}, extra_query={"trace": "1"}, extra_body={"foo": "bar"}, ), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id=None, conversation_id=None, prompt=None, ) assert provider.responses_calls == [] assert len(provider.private_responses_calls) == 1 call = provider.private_responses_calls[0] assert call["kwargs"]["extra_headers"]["X-Test-Header"] == "test" assert call["kwargs"]["extra_query"] == {"trace": "1"} assert call["kwargs"]["extra_body"] == {"foo": "bar"} @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_prompt_requests_fail_fast(monkeypatch) -> None: provider = FakeAnyLLMProvider(supports_responses=True, responses_response=_response("Hello")) module, _create_calls = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="openai/gpt-5.4-mini") with pytest.raises(Exception, match="prompt-managed requests"): await model.get_response( system_instructions=None, input="hi", model_settings=ModelSettings(), tools=[], output_schema=None, handoffs=[], tracing=ModelTracing.DISABLED, previous_response_id=None, conversation_id=None, prompt={"id": "pmpt_123"}, ) def test_any_llm_responses_input_sanitizer_strips_none_fields_from_reasoning_items() -> None: pytest.importorskip( "any_llm", reason="`any-llm-sdk` is only available when the optional dependency is installed.", ) from agents.extensions.models.any_llm_model import AnyLLMModel model = AnyLLMModel(model="openai/gpt-5.4-mini") raw_input = [ { "id": "rid1", "summary": [{"text": "why", "type": "summary_text"}], "type": "reasoning", "content": [{"type": "reasoning_text", "text": "thinking"}], "status": None, "encrypted_content": None, } ] cleaned = model._sanitize_any_llm_responses_input(raw_input) assert cleaned == [ { "id": "rid1", "summary": [{"text": "why", "type": "summary_text"}], "type": "reasoning", "content": [{"type": "reasoning_text", "text": "thinking"}], } ] ResponsesParams = importlib.import_module("any_llm.types.responses").ResponsesParams params = ResponsesParams(model="dummy", input=cleaned) assert isinstance(params.input, list) @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_any_llm_responses_path_sanitizes_replayed_items_before_validation() -> None: pytest.importorskip( "any_llm", reason="`any-llm-sdk` is only available when the optional dependency is installed.", ) from agents.extensions.models.any_llm_model import AnyLLMModel class ValidatingProvider: SUPPORTS_RESPONSES = True def __init__(self) -> None: self.private_responses_calls: list[dict[str, Any]] = [] async def aresponses(self, **kwargs: Any) -> Any: raise AssertionError("public aresponses path should not be used in this test") async def _aresponses(self, params: Any, **kwargs: Any) -> Response: self.private_responses_calls.append({"params": params, "kwargs": kwargs}) return _response("Hello from sanitized replay") class TestAnyLLMModel(AnyLLMModel): def __init__(self, provider: ValidatingProvider) -> None: super().__init__(model="openai/gpt-5.4-mini", api="responses") self._provider = provider def _get_provider(self) -> Any: return self._provider provider = ValidatingProvider() model = TestAnyLLMModel(provider) tools: list[Tool] = [] handoffs: list[Handoff[Any, Agent[Any]]] = [] stream_flag: Literal[False] = False replay_input = cast( list[TResponseInputItem], [ {"role": "user", "content": "What's the weather in Tokyo?"}, { "id": FAKE_RESPONSES_ID, "summary": [ {"text": "I should call the weather tool first.", "type": "summary_text"} ], "type": "reasoning", "content": [{"type": "reasoning_text", "text": "thinking"}], "status": None, "provider_data": {"model": "anthropic/fake-responses-model"}, }, { "id": FAKE_RESPONSES_ID, "arguments": '{"city": "Tokyo"}', "call_id": "call_weather_123", "name": "get_weather", "type": "function_call", "status": None, "provider_data": {"model": "anthropic/fake-responses-model"}, }, { "type": "function_call_output", "call_id": "call_weather_123", "output": "The weather in Tokyo is sunny and 22°C.", }, ], ) response = await model._fetch_responses_response( system_instructions=None, input=replay_input, model_settings=ModelSettings(), tools=tools, output_schema=None, handoffs=handoffs, previous_response_id=None, conversation_id=None, stream=stream_flag, prompt=None, ) assert response.id == "resp_123" assert len(provider.private_responses_calls) == 1 params = provider.private_responses_calls[0]["params"] assert params.input == [ {"role": "user", "content": "What's the weather in Tokyo?"}, { "arguments": '{"city": "Tokyo"}', "call_id": "call_weather_123", "name": "get_weather", "type": "function_call", }, { "type": "function_call_output", "call_id": "call_weather_123", "output": "The weather in Tokyo is sunny and 22°C.", }, ] def test_any_llm_provider_passes_api_override() -> None: pytest.importorskip( "any_llm", reason="`any-llm-sdk` is only available when the optional dependency is installed.", ) from agents.extensions.models.any_llm_model import AnyLLMModel from agents.extensions.models.any_llm_provider import AnyLLMProvider provider = AnyLLMProvider(api="chat_completions") model = provider.get_model("openai/gpt-4.1-mini") assert isinstance(model, AnyLLMModel) assert model.api == "chat_completions" def test_any_llm_reasoning_objects_prefer_content_attributes_over_iterable_pairs() -> None: pytest.importorskip( "any_llm", reason="`any-llm-sdk` is only available when the optional dependency is installed.", ) from any_llm.types.completion import Reasoning from agents.extensions.models.any_llm_model import _extract_any_llm_reasoning_text delta = pytypes.SimpleNamespace(reasoning=Reasoning(content="用户")) assert _extract_any_llm_reasoning_text(delta) == "用户" def test_any_llm_split_does_not_duplicate_content_or_thinking(monkeypatch) -> None: """Splitting multi-tool assistant messages must not duplicate text/thinking blocks. Anthropic's extended thinking API rejects requests that include the same signed thinking block more than once, and duplicated assistant text corrupts conversation history. Only the first split should retain content, thinking_blocks, and reasoning_content; subsequent splits should carry the tool_call alone. """ provider = FakeAnyLLMProvider(supports_responses=False) module, _ = _import_any_llm_module(monkeypatch, provider) AnyLLMModel = module.AnyLLMModel model = AnyLLMModel(model="anthropic/claude-3-5-sonnet") messages: list[Any] = [ {"role": "user", "content": "Search both"}, { "role": "assistant", "content": "Looking up both queries.", "thinking_blocks": [{"type": "thinking", "thinking": "plan", "signature": "sig_abc"}], "reasoning_content": "internal plan", "tool_calls": [ { "id": "call_1", "type": "function", "function": {"name": "s", "arguments": "{}"}, }, { "id": "call_2", "type": "function", "function": {"name": "s", "arguments": "{}"}, }, ], }, {"role": "tool", "tool_call_id": "call_1", "content": "ok1"}, {"role": "tool", "tool_call_id": "call_2", "content": "ok2"}, ] result = model._fix_tool_message_ordering(messages) assistants = [m for m in result if m.get("role") == "assistant"] assert len(assistants) == 2 # First split keeps the shared fields. assert assistants[0].get("content") == "Looking up both queries." assert "thinking_blocks" in assistants[0] assert "reasoning_content" in assistants[0] # Second split must NOT duplicate them. assert "content" not in assistants[1] assert "thinking_blocks" not in assistants[1] assert "reasoning_content" not in assistants[1] # Tool calls are still split one-per-message. assert assistants[0]["tool_calls"][0]["id"] == "call_1" assert assistants[1]["tool_calls"][0]["id"] == "call_2"