diff --git a/src/agents/models/openai_responses.py b/src/agents/models/openai_responses.py index f66c6afb7b..f533219d7c 100644 --- a/src/agents/models/openai_responses.py +++ b/src/agents/models/openai_responses.py @@ -45,6 +45,7 @@ from ..logger import logger from ..model_settings import MCPToolChoice from ..retry import ModelRetryAdvice, ModelRetryAdviceRequest +from ..run_internal.items import drop_orphaned_messages_after_consumed_reasoning from ..tool import ( ApplyPatchTool, CodeInterpreterTool, @@ -742,6 +743,12 @@ def _build_response_create_kwargs( list_input = ItemHelpers.input_to_new_input_list(input) list_input = _to_dump_compatible(list_input) list_input = self._remove_openai_responses_api_incompatible_fields(list_input) + # Official OpenAI tolerates a handoff's trailing message whose reasoning was consumed by the + # tool call, but strict Responses endpoints (e.g. Azure OpenAI) reject it with a 400. Strip + # that orphaned assistant message only for non-official endpoints; official OpenAI receives + # the items untouched, per the Responses API guidance to pass items through as-is. + if not is_official_openai_client(self._get_client()): + list_input = drop_orphaned_messages_after_consumed_reasoning(list_input) if model_settings.parallel_tool_calls and tools: parallel_tool_calls: bool | Omit = True diff --git a/src/agents/run_internal/items.py b/src/agents/run_internal/items.py index aadba1d361..4dbb95b10e 100644 --- a/src/agents/run_internal/items.py +++ b/src/agents/run_internal/items.py @@ -29,6 +29,7 @@ "local_shell_call": "local_shell_call_output", "tool_search_call": "tool_search_output", } +_CALL_OUTPUT_TYPES: frozenset[str] = frozenset(_TOOL_CALL_TO_OUTPUT_TYPE.values()) __all__ = [ "ReasoningItemIdPolicy", @@ -37,6 +38,7 @@ "TOOL_CALL_SESSION_TITLE_KEY", "copy_input_items", "drop_orphan_function_calls", + "drop_orphaned_messages_after_consumed_reasoning", "ensure_input_item_format", "prepare_model_input_items", "run_item_to_input_item", @@ -179,6 +181,63 @@ def _drop_reasoning_items_preceding_dropped_calls( return [entry for idx, entry in enumerate(items) if idx not in excluded] +def drop_orphaned_messages_after_consumed_reasoning( + items: list[TResponseInputItem], +) -> list[TResponseInputItem]: + """Drop assistant message items orphaned because their reasoning item was consumed by a call. + + Some Responses endpoints (notably Azure OpenAI) require every assistant message item to be + paired with its own reasoning item. When a reasoning-enabled agent hands off, the model emits + ``[reasoning, function_call, message]`` in a single response: the reasoning is consumed by the + ``function_call``, so the trailing message has no reasoning of its own. Replaying that shape + triggers a 400: ``Item 'msg_...' of type 'message' was provided without its required + 'reasoning' item``. + + This walks the flat item list and drops any assistant message that follows a tool call which + consumed the most recent reasoning item, until the next call-output item ends the sequence. + Only ``assistant`` messages are dropped -- user/system/developer messages are always kept so a + resumed turn's new input is never discarded. Reasoning items and tool calls are preserved. + + This is intentionally NOT part of the shared run pipeline: official OpenAI tolerates the shape + (it ignores reasoning items that are not relevant), so callers apply this only when targeting a + strict, non-official Responses endpoint. It is the message-side counterpart to + :func:`drop_orphan_function_calls`, which removes calls without outputs and their reasoning. + """ + fresh_reasoning = False # True when the most-recent reasoning item is not yet consumed. + consumed_by_call = False # True after a tool call consumes the fresh reasoning item. + result: list[TResponseInputItem] = [] + + for item in items: + if not isinstance(item, dict): + result.append(item) + continue + item_type = item.get("type") + + if item_type == "reasoning": + fresh_reasoning = True + consumed_by_call = False + result.append(item) + elif item_type in _TOOL_CALL_TO_OUTPUT_TYPE: + if fresh_reasoning: + fresh_reasoning = False + consumed_by_call = True # The reasoning is now consumed by this call. + result.append(item) + elif item_type in _CALL_OUTPUT_TYPES: + # A call output ends the call sequence. Reset so a turn with no trailing message does + # not bleed consumed_by_call into a later agent's legitimately reasoning-less message. + consumed_by_call = False + result.append(item) + elif item_type == "message": + if not consumed_by_call or item.get("role") != "assistant": + result.append(item) + # else: orphaned assistant message -- drop it without resetting so that any further + # assistant messages in the same turn are dropped until a call output resets the flag. + else: + result.append(item) + + return result + + def ensure_input_item_format(item: TResponseInputItem) -> TResponseInputItem: """Ensure a single item is normalized for model input.""" coerced = _coerce_to_dict(item) diff --git a/tests/models/test_openai_responses.py b/tests/models/test_openai_responses.py index d86d435167..676663c7cd 100644 --- a/tests/models/test_openai_responses.py +++ b/tests/models/test_openai_responses.py @@ -7,7 +7,7 @@ import httpx import pytest -from openai import NOT_GIVEN, APIConnectionError, AsyncOpenAI, RateLimitError, omit +from openai import NOT_GIVEN, APIConnectionError, AsyncOpenAI, BadRequestError, RateLimitError, omit from openai.types.responses import ResponseCompletedEvent, ResponseErrorEvent from openai.types.responses.response_create_params import ContextManagement from openai.types.shared.reasoning import Reasoning @@ -25,6 +25,7 @@ trace, ) from agents.exceptions import ModelBehaviorError, UserError +from agents.items import TResponseInputItem from agents.models._retry_runtime import ( provider_managed_retries_disabled, websocket_pre_event_retries_disabled, @@ -3849,3 +3850,161 @@ def test_websocket_pre_event_disconnect_retry_respects_websocket_retry_disable() with websocket_pre_event_retries_disabled(True): assert _should_retry_pre_event_websocket_disconnect() is False + + +# --------------------------------------------------------------------------- +# Orphaned handoff message handling (reasoning consumed by a tool call). +# +# When a reasoning-enabled agent hands off, the model emits +# ``[reasoning, function_call, message]`` in one response. The reasoning is consumed by the call, +# so the trailing assistant message is orphaned. Official OpenAI tolerates this shape, but strict +# Responses endpoints (e.g. Azure OpenAI) reject it with a 400. The Responses model strips that +# message only for non-official endpoints; official OpenAI receives the items untouched. +# --------------------------------------------------------------------------- + + +def _orphaned_handoff_input() -> list[TResponseInputItem]: + return [ + cast(TResponseInputItem, {"type": "reasoning", "id": "rs_1", "summary": []}), + cast( + TResponseInputItem, + {"type": "function_call", "call_id": "fc_1", "name": "transfer", "arguments": "{}"}, + ), + cast( + TResponseInputItem, + {"type": "message", "role": "assistant", "content": "Transferring you now."}, + ), + cast( + TResponseInputItem, + {"type": "function_call_output", "call_id": "fc_1", "output": "ok"}, + ), + ] + + +def _has_orphaned_assistant_message(items: list[Any]) -> bool: + return any( + isinstance(item, dict) + and item.get("type") == "message" + and item.get("role") == "assistant" + for item in items + ) + + +class _CapturingResponses: + def __init__(self) -> None: + self.kwargs: dict[str, Any] = {} + + async def create(self, **kwargs: Any) -> Any: + self.kwargs = kwargs + return get_response_obj([]) + + +class _CapturingClient: + def __init__(self, base_url: str) -> None: + self.responses = _CapturingResponses() + self.base_url = httpx.URL(base_url) + + +async def _capture_sent_input(base_url: str, input_items: list[TResponseInputItem]) -> list[Any]: + client = _CapturingClient(base_url) + model = OpenAIResponsesModel(model="gpt-5", openai_client=cast(Any, client)) + await model.get_response( + system_instructions=None, + input=input_items, + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + ) + return cast(list[Any], client.responses.kwargs["input"]) + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_non_official_endpoint_strips_orphaned_handoff_message() -> None: + sent_input = await _capture_sent_input( + "https://my-resource.openai.azure.com/openai/v1/", _orphaned_handoff_input() + ) + + assert not _has_orphaned_assistant_message(sent_input), ( + "The orphaned handoff message must be stripped for non-official (Azure) endpoints." + ) + # Reasoning and the tool call/output are preserved -- only the orphaned message is removed. + assert [cast(dict[str, Any], item)["type"] for item in sent_input] == [ + "reasoning", + "function_call", + "function_call_output", + ] + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_official_endpoint_preserves_orphaned_handoff_message() -> None: + sent_input = await _capture_sent_input( + "https://api.openai.com/v1/", _orphaned_handoff_input() + ) + + assert _has_orphaned_assistant_message(sent_input), ( + "Official OpenAI must receive items untouched, per the Responses API guidance." + ) + + +def _reject_orphaned_message_like_azure(items: list[Any]) -> None: + """Reproduce Azure's validation: an assistant message right after a tool call is rejected.""" + for previous, item in zip(items, items[1:], strict=False): + if ( + isinstance(item, dict) + and item.get("type") == "message" + and item.get("role") == "assistant" + and isinstance(previous, dict) + and previous.get("type") == "function_call" + ): + request = httpx.Request("POST", "https://my-resource.openai.azure.com/openai/responses") + raise BadRequestError( + "Item 'msg_1' of type 'message' was provided " + "without its required 'reasoning' item.", + response=httpx.Response(400, request=request), + body=None, + ) + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_non_official_endpoint_avoids_reasoning_item_400_on_handoff() -> None: + orphaned = _orphaned_handoff_input() + + # Document the actual reported failure: the unstripped shape is a hard 400 on Azure. + with pytest.raises(BadRequestError): + _reject_orphaned_message_like_azure(cast(list[Any], orphaned)) + + class _ValidatingResponses: + def __init__(self) -> None: + self.kwargs: dict[str, Any] = {} + + async def create(self, **kwargs: Any) -> Any: + self.kwargs = kwargs + _reject_orphaned_message_like_azure(cast(list[Any], kwargs["input"])) + return get_response_obj([]) + + class _ValidatingClient: + def __init__(self) -> None: + self.responses = _ValidatingResponses() + self.base_url = httpx.URL("https://my-resource.openai.azure.com/openai/v1/") + + client = _ValidatingClient() + model = OpenAIResponsesModel(model="gpt-5", openai_client=cast(Any, client)) + + # With the fix, the seam strips the orphaned message before create() validates, so the strict + # endpoint no longer raises the 400. + await model.get_response( + system_instructions=None, + input=orphaned, + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + ) + + assert not _has_orphaned_assistant_message(cast(list[Any], client.responses.kwargs["input"])) diff --git a/tests/test_run_internal_items.py b/tests/test_run_internal_items.py index c7997930d1..b5a58e5ca2 100644 --- a/tests/test_run_internal_items.py +++ b/tests/test_run_internal_items.py @@ -313,6 +313,140 @@ def test_drop_orphan_function_calls_keeps_reasoning_chain_before_non_dropped_ite ] +def test_drop_orphaned_messages_after_consumed_reasoning_drops_handoff_message() -> None: + # A reasoning-enabled handoff emits [reasoning, function_call, message]; the reasoning is + # consumed by the call, so the trailing assistant message is orphaned and must be dropped. + payload: list[Any] = [ + cast(TResponseInputItem, {"type": "reasoning", "id": "rs_1", "summary": []}), + cast( + TResponseInputItem, + {"type": "function_call", "call_id": "fc_1", "name": "transfer", "arguments": "{}"}, + ), + cast( + TResponseInputItem, + {"type": "message", "role": "assistant", "content": "Transferring you now."}, + ), + cast( + TResponseInputItem, + {"type": "function_call_output", "call_id": "fc_1", "output": "ok"}, + ), + ] + + filtered = run_items.drop_orphaned_messages_after_consumed_reasoning( + cast(list[TResponseInputItem], payload) + ) + + assert not any( + isinstance(entry, dict) + and entry.get("type") == "message" + and entry.get("role") == "assistant" + for entry in filtered + ) + # The reasoning, call, and output are preserved -- only the orphaned message is removed. + assert [cast(dict[str, Any], entry)["type"] for entry in filtered] == [ + "reasoning", + "function_call", + "function_call_output", + ] + + +def test_drop_orphaned_messages_after_consumed_reasoning_preserves_user_message() -> None: + # A resumed turn appends a new user message after an interrupted [reasoning, function_call]. + # User input must never be discarded even though consumed_by_call is set. + payload: list[Any] = [ + cast(TResponseInputItem, {"type": "reasoning", "id": "rs_1", "summary": []}), + cast( + TResponseInputItem, + {"type": "function_call", "call_id": "fc_1", "name": "transfer", "arguments": "{}"}, + ), + cast( + TResponseInputItem, + {"type": "message", "role": "user", "content": "Actually, cancel that."}, + ), + ] + + filtered = run_items.drop_orphaned_messages_after_consumed_reasoning( + cast(list[TResponseInputItem], payload) + ) + + user_messages = [ + entry for entry in filtered if isinstance(entry, dict) and entry.get("role") == "user" + ] + assert len(user_messages) == 1 + assert cast(dict[str, Any], user_messages[0])["content"] == "Actually, cancel that." + + +def test_drop_orphaned_messages_after_consumed_reasoning_drops_multiple_messages() -> None: + # Every orphaned assistant message before the call output is dropped, not just the first. + payload: list[Any] = [ + cast(TResponseInputItem, {"type": "reasoning", "id": "rs_1", "summary": []}), + cast( + TResponseInputItem, + {"type": "function_call", "call_id": "fc_1", "name": "transfer", "arguments": "{}"}, + ), + cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": "one"}), + cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": "two"}), + cast( + TResponseInputItem, + {"type": "function_call_output", "call_id": "fc_1", "output": "ok"}, + ), + ] + + filtered = run_items.drop_orphaned_messages_after_consumed_reasoning( + cast(list[TResponseInputItem], payload) + ) + + assert not any( + isinstance(entry, dict) and entry.get("type") == "message" for entry in filtered + ) + + +def test_drop_orphaned_messages_after_consumed_reasoning_keeps_unconsumed_message() -> None: + # When no tool call consumed the reasoning, the assistant message keeps its own reasoning and + # must be preserved. + payload: list[Any] = [ + cast(TResponseInputItem, {"type": "reasoning", "id": "rs_1", "summary": []}), + cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": "Hi there."}), + ] + + filtered = run_items.drop_orphaned_messages_after_consumed_reasoning( + cast(list[TResponseInputItem], payload) + ) + + assert filtered == payload + + +def test_drop_orphaned_messages_after_consumed_reasoning_resets_after_call_output() -> None: + # After a call output ends the sequence, a later agent's legitimately reasoning-less assistant + # message must not be dropped (the consumed flag must not bleed across turns). + payload: list[Any] = [ + cast(TResponseInputItem, {"type": "reasoning", "id": "rs_1", "summary": []}), + cast( + TResponseInputItem, + {"type": "function_call", "call_id": "fc_1", "name": "transfer", "arguments": "{}"}, + ), + cast( + TResponseInputItem, + {"type": "function_call_output", "call_id": "fc_1", "output": "ok"}, + ), + cast( + TResponseInputItem, + {"type": "message", "role": "assistant", "content": "Here is the answer."}, + ), + ] + + filtered = run_items.drop_orphaned_messages_after_consumed_reasoning( + cast(list[TResponseInputItem], payload) + ) + + assert any( + isinstance(entry, dict) + and entry.get("type") == "message" + and entry.get("content") == "Here is the answer." + for entry in filtered + ) + + def test_normalize_and_ensure_input_item_format_keep_non_dict_entries() -> None: item = cast(TResponseInputItem, "raw-item") assert run_items.ensure_input_item_format(item) == item