From be8e305124531e3c1e4196d02507f96bd6e68218 Mon Sep 17 00:00:00 2001 From: Utkarsh Date: Thu, 9 Jul 2026 21:50:47 +0530 Subject: [PATCH] fix: strip orphaned handoff message only for non-official Responses endpoints When a reasoning-enabled agent hands off, the model emits [reasoning, function_call, message] in one response. The reasoning is consumed by the function_call, leaving the trailing assistant message without its own reasoning item. Strict Responses endpoints (e.g. Azure OpenAI) reject the replayed shape with HTTP 400 ("Item 'msg_...' of type 'message' was provided without its required 'reasoning' item"), while official OpenAI tolerates it and, per the Responses API guidance, expects items since the last user message to be passed through untouched. Scope the repair to the Responses model layer, applied only when the client is not official OpenAI (is_official_openai_client). Official OpenAI and the Chat Completions/LiteLLM paths are untouched. The message also stays in session history -- it is only omitted from the outgoing payload to a strict endpoint, where keeping it would otherwise 400 the entire turn. Add a test that reproduces the actual Azure 400 and proves the fix prevents it, plus endpoint-scoping and pure-function unit tests. Co-Authored-By: Claude Opus 4.8 --- src/agents/models/openai_responses.py | 7 ++ src/agents/run_internal/items.py | 59 ++++++++++ tests/models/test_openai_responses.py | 161 +++++++++++++++++++++++++- tests/test_run_internal_items.py | 134 +++++++++++++++++++++ 4 files changed, 360 insertions(+), 1 deletion(-) 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