fix: reject history-rewriting handoffs with server-managed conversation

src/agents/run.py

@@ -664,6 +664,9 @@ def _with_reasoning_item_id_policy(result: RunResult) -> RunResult:
                                 context_wrapper=context_wrapper,
                                 run_config=run_config,
                                 run_state=run_state,
+                                server_managed_conversation=(
+                                    server_conversation_tracker is not None
+                                ),
                             )
 
                             if run_state._last_processed_response is not None:

src/agents/run_internal/run_loop.py

@@ -615,6 +615,7 @@ async def _save_stream_items_without_count(
                         context_wrapper=context_wrapper,
                         run_config=run_config,
                         run_state=run_state,
+                        server_managed_conversation=server_conversation_tracker is not None,
                     )
 
                     tool_use_tracker.record_processed_response(
@@ -1431,6 +1432,7 @@ async def rewind_model_request() -> None:
         hooks=hooks,
         context_wrapper=context_wrapper,
         run_config=run_config,
+        server_managed_conversation=server_conversation_tracker is not None,
         tool_use_tracker=tool_use_tracker,
         event_queue=streamed_result._event_queue,
     )
@@ -1557,6 +1559,7 @@ async def run_single_turn(
         hooks=hooks,
         context_wrapper=context_wrapper,
         run_config=run_config,
+        server_managed_conversation=server_conversation_tracker is not None,
         tool_use_tracker=tool_use_tracker,
     )
 

src/agents/run_internal/turn_resolution.py

@@ -293,6 +293,7 @@ async def execute_handoffs(
     hooks: RunHooks[TContext],
     context_wrapper: RunContextWrapper[TContext],
     run_config: RunConfig,
+    server_managed_conversation: bool = False,
     nest_handoff_history_fn: Callable[..., HandoffInputData] | None = None,
 ) -> SingleStepResult:
     """Execute a handoff and prepare the next turn for the new agent."""
@@ -319,6 +320,21 @@ def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffIn
     actual_handoff = run_handoffs[0]
     with handoff_span(from_agent=agent.name) as span_handoff:
         handoff = actual_handoff.handoff
+        input_filter = handoff.input_filter or (
+            run_config.handoff_input_filter if run_config else None
+        )
+        handoff_nest_setting = handoff.nest_handoff_history
+        should_nest_history = (
+            handoff_nest_setting
+            if handoff_nest_setting is not None
+            else run_config.nest_handoff_history
+        )
+        if server_managed_conversation and (input_filter or should_nest_history):
+            raise UserError(
+                "Server-managed conversation cannot be combined with handoff input filtering "
+                "or history nesting. Disable input_filter/nest_handoff_history or avoid "
+                "conversation_id, previous_response_id, or auto_previous_response_id."
+            )
         new_agent: Agent[Any] = await handoff.on_invoke_handoff(
             context_wrapper, actual_handoff.tool_call.arguments
         )
@@ -363,15 +379,6 @@ def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffIn
             ),
         )
 
-        input_filter = handoff.input_filter or (
-            run_config.handoff_input_filter if run_config else None
-        )
-        handoff_nest_setting = handoff.nest_handoff_history
-        should_nest_history = (
-            handoff_nest_setting
-            if handoff_nest_setting is not None
-            else run_config.nest_handoff_history
-        )
         handoff_input_data: HandoffInputData | None = None
         session_step_items: list[RunItem] | None = None
         if input_filter or should_nest_history:
@@ -507,6 +514,7 @@ async def execute_tools_and_side_effects(
     hooks: RunHooks[TContext],
     context_wrapper: RunContextWrapper[TContext],
     run_config: RunConfig,
+    server_managed_conversation: bool = False,
 ) -> SingleStepResult:
     """Run one turn of the loop, coordinating tools, approvals, guardrails, and handoffs."""
 
@@ -596,6 +604,7 @@ async def execute_tools_and_side_effects(
             hooks=hooks,
             context_wrapper=context_wrapper,
             run_config=run_config,
+            server_managed_conversation=server_managed_conversation,
         )
 
     tool_final_output = await _maybe_finalize_from_tool_results(
@@ -673,6 +682,7 @@ async def resolve_interrupted_turn(
     context_wrapper: RunContextWrapper[TContext],
     run_config: RunConfig,
     run_state: RunState | None = None,
+    server_managed_conversation: bool = False,
     nest_handoff_history_fn: Callable[..., HandoffInputData] | None = None,
 ) -> SingleStepResult:
     """Continue a turn that was previously interrupted waiting for tool approval."""
@@ -1241,6 +1251,7 @@ def _add_unmatched_pending(approval: ToolApprovalItem) -> None:
             hooks=hooks,
             context_wrapper=context_wrapper,
             run_config=run_config,
+            server_managed_conversation=server_managed_conversation,
             nest_handoff_history_fn=nest_history,
         )
 
@@ -1694,6 +1705,7 @@ async def get_single_step_result_from_response(
     hooks: RunHooks[TContext],
     context_wrapper: RunContextWrapper[TContext],
     run_config: RunConfig,
+    server_managed_conversation: bool = False,
     tool_use_tracker,
     event_queue: asyncio.Queue[StreamEvent | QueueCompleteSentinel] | None = None,
 ) -> SingleStepResult:
@@ -1725,4 +1737,5 @@ async def get_single_step_result_from_response(
         hooks=hooks,
         context_wrapper=context_wrapper,
         run_config=run_config,
+        server_managed_conversation=server_managed_conversation,
     )

tests/test_agent_runner.py

@@ -40,6 +40,7 @@
     retry_policies,
     tool_namespace,
 )
+from agents.extensions.handoff_filters import remove_all_tools
 from agents.agent import ToolsToFinalOutputResult
 from agents.computer import Computer
 from agents.items import (
@@ -2704,6 +2705,61 @@ async def test_run_streamed_rejects_session_with_resumed_conversation_state():
         Runner.run_streamed(agent, state, session=session)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "run_kwargs",
+    [
+        {"conversation_id": "conv-test"},
+        {"previous_response_id": "resp-test"},
+        {"auto_previous_response_id": True},
+    ],
+)
+async def test_run_rejects_handoff_input_filter_with_server_managed_conversation(
+    run_kwargs: dict[str, Any],
+):
+    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=remove_all_tools)],
+    )
+
+    triage_model.add_multiple_turn_outputs([[get_handoff_tool_call(delegate)]])
+
+    with pytest.raises(UserError, match="Server-managed conversation cannot be combined"):
+        await Runner.run(triage, input="user_message", **run_kwargs)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "run_kwargs",
+    [
+        {"conversation_id": "conv-test"},
+        {"previous_response_id": "resp-test"},
+        {"auto_previous_response_id": True},
+    ],
+)
+async def test_run_rejects_handoff_history_nesting_with_server_managed_conversation(
+    run_kwargs: dict[str, Any],
+):
+    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_handoff_tool_call(delegate)]])
+
+    with pytest.raises(UserError, match="Server-managed conversation cannot be combined"):
+        await Runner.run(
+            triage,
+            input="user_message",
+            run_config=RunConfig(nest_handoff_history=True),
+            **run_kwargs,
+        )
+
+
 @pytest.mark.asyncio
 async def test_multi_turn_previous_response_id_passed_between_runs():
     """Test that previous_response_id is passed to the model on subsequent runs."""