From a470f374f92d486e6437e68423a3cdd3f736a24f Mon Sep 17 00:00:00 2001 From: Kazuhiro Sera Date: Sat, 14 Mar 2026 09:55:00 +0900 Subject: [PATCH] fix: #2258 add normalized to_input_list mode for filtered handoff follow-ups --- src/agents/result.py | 69 ++++++-- src/agents/run.py | 16 ++ .../run_internal/agent_runner_helpers.py | 1 + src/agents/run_internal/run_loop.py | 11 ++ tests/test_agent_runner.py | 4 + tests/test_agent_runner_streamed.py | 8 + tests/test_handoff_history_duplication.py | 161 +++++++++++++++++- 7 files changed, 258 insertions(+), 12 deletions(-) diff --git a/src/agents/result.py b/src/agents/result.py index 36d303c3d5..774c90dc4e 100644 --- a/src/agents/result.py +++ b/src/agents/result.py @@ -29,7 +29,7 @@ ) from .logger import logger from .run_context import RunContextWrapper -from .run_internal.items import run_item_to_input_item +from .run_internal.items import run_items_to_input_items from .run_internal.run_steps import ( NextStepInterruption, ProcessedResponse, @@ -110,6 +110,40 @@ def _populate_state_from_result( return state +ToInputListMode = Literal["preserve_all", "normalized"] + + +def _input_items_for_result( + result: RunResultBase, + *, + mode: ToInputListMode, + reasoning_item_id_policy: Literal["preserve", "omit"] | None, +) -> list[TResponseInputItem]: + """Return input items for the requested result view. + + ``preserve_all`` keeps the full converted history from ``new_items``. ``normalized`` returns + the canonical continuation input when handoff filtering rewrote model history, otherwise it + falls back to the same converted history. + """ + session_items = run_items_to_input_items(result.new_items, reasoning_item_id_policy) + if mode == "preserve_all": + return session_items + if mode != "normalized": + raise ValueError(f"Unsupported to_input_list mode: {mode}") + if not getattr(result, "_replay_from_model_input_items", False): + # Most runs never rewrite continuation history, so normalized stays identical to the + # historical preserve-all view unless the runner explicitly marked a divergence. + return session_items + + model_input_items = getattr(result, "_model_input_items", None) + if not isinstance(model_input_items, list): + return session_items + + # When the runner marks a divergence, generated_items already reflect the continuation input + # chosen for the next local run after applying handoff/input filtering. + return run_items_to_input_items(model_input_items, reasoning_item_id_policy) + + @dataclass class RunResultBase(abc.ABC): input: str | list[TResponseInputItem] @@ -145,6 +179,12 @@ class RunResultBase(abc.ABC): _trace_state: TraceState | None = field(default=None, init=False, repr=False) """Serialized trace metadata captured during the run.""" + _replay_from_model_input_items: bool = field(default=False, init=False, repr=False) + """Whether replay helpers should prefer `_model_input_items` over `new_items`. + + This is only set when the runner preserved extra session history items that should not be + replayed into the next local run, such as nested handoff history or filtered handoff input. + """ @classmethod def __get_pydantic_core_schema__( @@ -208,18 +248,25 @@ def final_output_as(self, cls: type[T], raise_if_incorrect_type: bool = False) - return cast(T, self.final_output) - def to_input_list(self) -> list[TResponseInputItem]: - """Creates a new input list, merging the original input with all the new items generated.""" + def to_input_list( + self, + *, + mode: ToInputListMode = "preserve_all", + ) -> list[TResponseInputItem]: + """Create an input-item view of this run. + + ``mode="preserve_all"`` keeps the historical behavior of converting ``new_items`` into a + full plain-item history. ``mode="normalized"`` prefers the canonical continuation input + when handoff filtering rewrote model history, while remaining identical for ordinary runs. + """ original_items: list[TResponseInputItem] = ItemHelpers.input_to_new_input_list(self.input) - new_items: list[TResponseInputItem] = [] reasoning_item_id_policy = getattr(self, "_reasoning_item_id_policy", None) - for item in self.new_items: - converted = run_item_to_input_item(item, reasoning_item_id_policy) - if converted is None: - continue - new_items.append(converted) - - return original_items + new_items + replay_items = _input_items_for_result( + self, + mode=mode, + reasoning_item_id_policy=reasoning_item_id_policy, + ) + return original_items + replay_items @property def agent_tool_invocation(self) -> AgentToolInvocation | None: diff --git a/src/agents/run.py b/src/agents/run.py index 355fcc0967..047d454d35 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -798,6 +798,11 @@ def _with_reasoning_item_id_policy(result: RunResult) -> RunResult: ) result._current_turn = current_turn result._model_input_items = list(generated_items) + # Keep normalized replay aligned with the model-facing + # continuation whenever session history preserved extra items. + result._replay_from_model_input_items = list( + generated_items + ) != list(session_items) if run_state is not None: result._trace_state = run_state._trace_state if session_persistence_enabled: @@ -932,6 +937,9 @@ def _with_reasoning_item_id_policy(result: RunResult) -> RunResult: ) result._current_turn = max_turns result._model_input_items = list(generated_items) + result._replay_from_model_input_items = list(generated_items) != list( + session_items + ) if run_state is not None: result._trace_state = run_state._trace_state if session_persistence_enabled and include_in_history: @@ -1200,6 +1208,9 @@ def _with_reasoning_item_id_policy(result: RunResult) -> RunResult: ) result._current_turn = current_turn result._model_input_items = list(generated_items) + result._replay_from_model_input_items = list(generated_items) != list( + session_items + ) if run_state is not None: result._current_turn_persisted_item_count = ( run_state._current_turn_persisted_item_count @@ -1591,6 +1602,11 @@ def run_streamed( streamed_result._model_input_items = ( list(run_state._generated_items) if run_state is not None else [] ) + streamed_result._replay_from_model_input_items = ( + list(run_state._generated_items) != list(run_state._session_items) + if run_state is not None + else False + ) streamed_result._reasoning_item_id_policy = resolved_reasoning_item_id_policy if run_state is not None: streamed_result._trace_state = run_state._trace_state diff --git a/src/agents/run_internal/agent_runner_helpers.py b/src/agents/run_internal/agent_runner_helpers.py index 9523e80a04..776e406703 100644 --- a/src/agents/run_internal/agent_runner_helpers.py +++ b/src/agents/run_internal/agent_runner_helpers.py @@ -271,6 +271,7 @@ def build_interruption_result( ) result._current_turn = current_turn result._model_input_items = list(generated_items) + result._replay_from_model_input_items = list(generated_items) != list(session_items) if run_state is not None: result._current_turn_persisted_item_count = run_state._current_turn_persisted_item_count result._trace_state = run_state._trace_state diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index 0220a3a4b7..3d21d89fda 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -483,6 +483,11 @@ def _sync_conversation_tracking_from_tracker() -> None: streamed_result._state = run_state if run_state is not None: streamed_result._model_input_items = list(run_state._generated_items) + # Streamed follow-ups need the same normalized replay signal as sync runs when the + # runner's continuation differs from the richer session history. + streamed_result._replay_from_model_input_items = list(run_state._generated_items) != list( + run_state._session_items + ) if run_state is not None: run_state._conversation_id = conversation_id @@ -627,6 +632,9 @@ async def _save_stream_items_without_count( ) streamed_result._model_input_items = generated_items streamed_result.new_items = base_session_items + list(turn_session_items) + streamed_result._replay_from_model_input_items = list( + streamed_result._model_input_items + ) != list(streamed_result.new_items) if run_state is not None: update_run_state_after_resume( run_state, @@ -914,6 +922,9 @@ async def _save_stream_items_without_count( ) turn_session_items = session_items_for_turn(turn_result) streamed_result.new_items.extend(turn_session_items) + streamed_result._replay_from_model_input_items = list( + streamed_result._model_input_items + ) != list(streamed_result.new_items) store_setting = current_agent.model_settings.resolve( run_config.model_settings ).store diff --git a/tests/test_agent_runner.py b/tests/test_agent_runner.py index 971a165f1d..8b07297167 100644 --- a/tests/test_agent_runner.py +++ b/tests/test_agent_runner.py @@ -1146,6 +1146,10 @@ async def test_structured_output(): "should have input: conversation summary, function call, function call result, message, " "handoff, handoff output, preamble message, tool call, tool call result, final output" ) + assert len(result.to_input_list(mode="normalized")) == 6, ( + "should have normalized replay input: conversation summary, carried-forward message, " + "preamble message, tool call, tool call result, final output" + ) assert result.last_agent == agent_1, "should have handed off to agent_1" assert result.final_output == Foo(bar="baz"), "should have structured output" diff --git a/tests/test_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py index c3db6e06e4..0e729fed37 100644 --- a/tests/test_agent_runner_streamed.py +++ b/tests/test_agent_runner_streamed.py @@ -669,6 +669,10 @@ async def test_structured_output(): "should have input: conversation summary, function call, function call result, message, " "handoff, handoff output, preamble message, tool call, tool call result, final output" ) + assert len(result.to_input_list(mode="normalized")) == 6, ( + "should have normalized replay input: conversation summary, carried-forward message, " + "preamble message, tool call, tool call result, final output" + ) assert result.last_agent == agent_1, "should have handed off to agent_1" assert result.final_output == Foo(bar="baz"), "should have structured output" @@ -1398,6 +1402,10 @@ async def test_streaming_events(): "should have input: conversation summary, function call, function call result, message, " "handoff, handoff output, tool call, tool call result, final output" ) + assert len(result.to_input_list(mode="normalized")) == 5, ( + "should have normalized replay input: conversation summary, carried-forward message, " + "tool call, tool call result, final output" + ) assert result.last_agent == agent_1, "should have handed off to agent_1" assert result.final_output == Foo(bar="baz"), "should have structured output" diff --git a/tests/test_handoff_history_duplication.py b/tests/test_handoff_history_duplication.py index 9afc65e7d8..d26357de5c 100644 --- a/tests/test_handoff_history_duplication.py +++ b/tests/test_handoff_history_duplication.py @@ -5,8 +5,10 @@ in the input sent to the next agent. """ +import json from typing import Any, cast +import pytest from openai.types.responses import ( ResponseFunctionToolCall, ResponseOutputMessage, @@ -14,7 +16,7 @@ ) from openai.types.responses.response_reasoning_item import ResponseReasoningItem, Summary -from agents import Agent +from agents import Agent, RunConfig, Runner, function_tool, handoff from agents.handoffs import HandoffInputData, nest_handoff_history from agents.items import ( HandoffCallItem, @@ -26,6 +28,9 @@ ToolCallOutputItem, ) +from .fake_model import FakeModel +from .test_responses import get_function_tool_call, get_handoff_tool_call, get_text_message + def _create_mock_agent() -> Agent: """Create a mock agent for testing.""" @@ -365,3 +370,157 @@ def test_full_handoff_scenario_no_duplication(self): assert len(function_call_outputs) == 0, ( "No function_call_output items should be in model input" ) + + +@pytest.mark.asyncio +async def test_to_input_list_normalized_uses_filtered_continuation_after_nested_handoff() -> None: + triage_model = FakeModel() + delegate_model = FakeModel() + + delegate = Agent(name="delegate", model=delegate_model) + triage = Agent(name="triage", model=triage_model, handoffs=[delegate]) + + triage_model.add_multiple_turn_outputs( + [[get_text_message("triage summary"), get_handoff_tool_call(delegate)]] + ) + delegate_model.add_multiple_turn_outputs( + [ + [get_text_message("resolution")], + [get_text_message("followup answer")], + ] + ) + + result = await Runner.run( + triage, + input="user_question", + run_config=RunConfig(nest_handoff_history=True), + ) + + preserve_all_input = result.to_input_list() + normalized_input = result.to_input_list(mode="normalized") + preserve_all_types = [ + item.get("type", "message") for item in preserve_all_input if isinstance(item, dict) + ] + normalized_types = [ + item.get("type", "message") for item in normalized_input if isinstance(item, dict) + ] + + assert len(preserve_all_input) == 5 + assert "function_call" in preserve_all_types + assert "function_call_output" in preserve_all_types + assert len(normalized_input) == 3 + assert "function_call" not in normalized_types + assert "function_call_output" not in normalized_types + + follow_up_input = normalized_input + [{"role": "user", "content": "follow up?"}] + follow_up_result = await Runner.run(delegate, input=follow_up_input) + + assert follow_up_result.final_output == "followup answer" + assert delegate_model.last_turn_args["input"] == follow_up_input + + +@pytest.mark.asyncio +async def test_to_input_list_normalized_keeps_delegate_tool_items_after_nested_handoff() -> None: + async def lookup_weather(city: str) -> str: + return f"weather:{city}" + + triage_model = FakeModel() + delegate_model = FakeModel() + + delegate = Agent( + name="delegate", + model=delegate_model, + tools=[function_tool(lookup_weather, name_override="lookup_weather")], + ) + triage = Agent(name="triage", model=triage_model, handoffs=[delegate]) + + triage_model.add_multiple_turn_outputs( + [[get_text_message("triage summary"), get_handoff_tool_call(delegate)]] + ) + delegate_model.add_multiple_turn_outputs( + [ + [ + get_text_message("delegate preamble"), + get_function_tool_call("lookup_weather", json.dumps({"city": "Tokyo"})), + ], + [get_text_message("resolution")], + ] + ) + + result = await Runner.run( + triage, + input="user_question", + run_config=RunConfig(nest_handoff_history=True), + ) + + preserve_all_input = result.to_input_list() + normalized_input = result.to_input_list(mode="normalized") + preserve_all_function_calls = [ + cast(dict[str, Any], item) + for item in preserve_all_input + if isinstance(item, dict) and item.get("type") == "function_call" + ] + preserve_all_function_outputs = [ + cast(dict[str, Any], item) + for item in preserve_all_input + if isinstance(item, dict) and item.get("type") == "function_call_output" + ] + function_calls = [ + cast(dict[str, Any], item) + for item in normalized_input + if isinstance(item, dict) and item.get("type") == "function_call" + ] + function_outputs = [ + cast(dict[str, Any], item) + for item in normalized_input + if isinstance(item, dict) and item.get("type") == "function_call_output" + ] + + assert len(preserve_all_function_calls) == 2 + assert len(preserve_all_function_outputs) == 2 + assert len(function_calls) == 1 + assert function_calls[0]["name"] == "lookup_weather" + assert len(function_outputs) == 1 + assert function_outputs[0]["output"] == "weather:Tokyo" + + +@pytest.mark.asyncio +async def test_to_input_list_normalized_uses_custom_filter_input_items() -> None: + def keep_messages_only(data: HandoffInputData) -> HandoffInputData: + return data.clone( + input_items=tuple( + item for item in data.new_items if isinstance(item, MessageOutputItem) + ) + ) + + triage_model = FakeModel() + delegate_model = FakeModel() + + delegate = Agent(name="delegate", model=delegate_model) + triage = Agent( + name="triage", + model=triage_model, + handoffs=[handoff(delegate, input_filter=keep_messages_only)], + ) + + triage_model.add_multiple_turn_outputs( + [[get_text_message("triage summary"), get_handoff_tool_call(delegate)]] + ) + delegate_model.add_multiple_turn_outputs([[get_text_message("resolution")]]) + + result = await Runner.run(triage, input="user_question") + preserve_all_input = result.to_input_list() + normalized_input = result.to_input_list(mode="normalized") + preserve_all_types = [ + item.get("type", "message") for item in preserve_all_input if isinstance(item, dict) + ] + normalized_types = [ + item.get("type", "message") for item in normalized_input if isinstance(item, dict) + ] + + assert len(preserve_all_input) == 5 + assert "function_call" in preserve_all_types + assert "function_call_output" in preserve_all_types + assert len(normalized_input) == 3 + assert "function_call" not in normalized_types + assert "function_call_output" not in normalized_types