fix: #2258 add normalized to_input_list mode for filtered handoff follow-ups (#2667)

src/agents/result.py

@@ -29,7 +29,7 @@ from .items import (
 )
 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 @@ class RunResultBase(abc.ABC):
 
         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."""
-        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)
+    def to_input_list(
+        self,
+        *,
+        mode: ToInputListMode = "preserve_all",
+    ) -> list[TResponseInputItem]:
+        """Create an input-item view of this run.
 
-        return original_items + new_items
+        ``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)
+        reasoning_item_id_policy = getattr(self, "_reasoning_item_id_policy", None)
+        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:

src/agents/run.py

@@ -798,6 +798,11 @@ class AgentRunner:
                                 )
                                 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 @@ class AgentRunner:
                         )
                         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 @@ class AgentRunner:
                             )
                             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 @@ class AgentRunner:
         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

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

src/agents/run_internal/run_loop.py

@@ -483,6 +483,11 @@ async def start_streaming(
         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 start_streaming(
                     )
                     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 start_streaming(
                 )
                 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

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"

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"

tests/test_handoff_history_duplication.py

@@ -5,8 +5,10 @@ function_call and function_call_output items are NOT duplicated
 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 import (
 )
 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 @@ from agents.items import (
     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 @@ class TestHandoffHistoryDuplicationFix:
         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