fix(run): fail fast on incompatible handoff settings with server conversation

src/agents/run.py

@@ -54,6 +54,7 @@
     save_turn_items_if_needed,
     should_cancel_parallel_model_task_on_input_guardrail_trip,
     update_run_state_for_interruption,
+    validate_server_conversation_handoff_settings,
     validate_session_conversation_settings,
 )
 from .run_internal.approvals import approvals_from_step
@@ -438,6 +439,13 @@ async def run(
                 previous_response_id=previous_response_id,
                 auto_previous_response_id=auto_previous_response_id,
             )
+            validate_server_conversation_handoff_settings(
+                run_state._current_agent if run_state._current_agent else starting_agent,
+                run_config,
+                conversation_id=conversation_id,
+                previous_response_id=previous_response_id,
+                auto_previous_response_id=auto_previous_response_id,
+            )
             starting_input = run_state._original_input
             original_user_input = copy_input_items(run_state._original_input)
             prepared_input = normalize_resumed_input(original_user_input)
@@ -459,6 +467,13 @@ async def run(
                 previous_response_id=previous_response_id,
                 auto_previous_response_id=auto_previous_response_id,
             )
+            validate_server_conversation_handoff_settings(
+                starting_agent,
+                run_config,
+                conversation_id=conversation_id,
+                previous_response_id=previous_response_id,
+                auto_previous_response_id=auto_previous_response_id,
+            )
 
             server_manages_conversation = (
                 conversation_id is not None
@@ -1422,6 +1437,13 @@ def run_streamed(
                 previous_response_id=previous_response_id,
                 auto_previous_response_id=auto_previous_response_id,
             )
+            validate_server_conversation_handoff_settings(
+                run_state._current_agent if run_state._current_agent else starting_agent,
+                run_config,
+                conversation_id=conversation_id,
+                previous_response_id=previous_response_id,
+                auto_previous_response_id=auto_previous_response_id,
+            )
             # When resuming, use the original_input from state.
             # primeFromState will mark items as sent so prepareInput skips them
             starting_input = run_state._original_input
@@ -1457,6 +1479,13 @@ def run_streamed(
                 previous_response_id=previous_response_id,
                 auto_previous_response_id=auto_previous_response_id,
             )
+            validate_server_conversation_handoff_settings(
+                starting_agent,
+                run_config,
+                conversation_id=conversation_id,
+                previous_response_id=previous_response_id,
+                auto_previous_response_id=auto_previous_response_id,
+            )
             context_wrapper = ensure_context_wrapper(context)
             # input_for_state is the same as input_for_result here
             input_for_state = input_for_result

src/agents/run_internal/agent_runner_helpers.py

@@ -7,6 +7,7 @@
 from ..agent import Agent
 from ..exceptions import UserError
 from ..guardrail import InputGuardrailResult
+from ..handoffs import Handoff
 from ..items import ModelResponse, RunItem, ToolApprovalItem, TResponseInputItem
 from ..memory import Session
 from ..result import RunResult
@@ -45,6 +46,7 @@
     "save_turn_items_if_needed",
     "should_cancel_parallel_model_task_on_input_guardrail_trip",
     "update_run_state_for_interruption",
+    "validate_server_conversation_handoff_settings",
 ]
 
 _PARALLEL_INPUT_GUARDRAIL_CANCEL_PATCH_ID = (
@@ -103,6 +105,108 @@ def validate_session_conversation_settings(
     )
 
 
+def _is_remove_all_tools_filter(filter_fn: object | None) -> bool:
+    if filter_fn is None:
+        return False
+    module_name = getattr(filter_fn, "__module__", "")
+    function_name = getattr(filter_fn, "__name__", "")
+    return function_name == "remove_all_tools" and module_name.endswith(
+        "agents.extensions.handoff_filters"
+    )
+
+
+def _is_disabled_handoff(handoff_obj: Handoff[Any, Agent[Any]]) -> bool:
+    is_enabled = handoff_obj.is_enabled
+    return isinstance(is_enabled, bool) and not is_enabled
+
+
+def _collect_handoff_edges(
+    agent: Agent[Any],
+) -> list[tuple[Agent[Any], Handoff[Any, Agent[Any]] | None]]:
+    edges: list[tuple[Agent[Any], Handoff[Any, Agent[Any]] | None]] = []
+    visited_agents: set[int] = set()
+    queue: list[Agent[Any]] = [agent]
+
+    while queue:
+        current = queue.pop(0)
+        current_id = id(current)
+        if current_id in visited_agents:
+            continue
+        visited_agents.add(current_id)
+
+        for handoff_or_agent in current.handoffs:
+            handoff: Handoff[Any, Agent[Any]] | None = None
+            target: Agent[Any]
+            if isinstance(handoff_or_agent, Agent):
+                target = handoff_or_agent
+            elif isinstance(handoff_or_agent, Handoff):
+                handoff = handoff_or_agent
+                if _is_disabled_handoff(handoff):
+                    continue
+                target_ref = handoff._agent_ref() if handoff._agent_ref is not None else None
+                if not isinstance(target_ref, Agent):
+                    continue
+                target = target_ref
+            else:
+                continue
+
+            edges.append((target, handoff))
+            queue.append(target)
+
+    return edges
+
+
+def validate_server_conversation_handoff_settings(
+    agent: Agent[Any],
+    run_config: RunConfig,
+    *,
+    conversation_id: str | None,
+    previous_response_id: str | None,
+    auto_previous_response_id: bool,
+) -> None:
+    server_manages_conversation = (
+        conversation_id is not None or previous_response_id is not None or auto_previous_response_id
+    )
+    if not server_manages_conversation:
+        return
+
+    has_remove_all_tools = False
+    has_nested_handoff_history = False
+    for _, handoff in _collect_handoff_edges(agent):
+        input_filter = (
+            handoff.input_filter
+            if handoff and handoff.input_filter is not None
+            else run_config.handoff_input_filter
+        )
+        if _is_remove_all_tools_filter(input_filter):
+            has_remove_all_tools = True
+
+        handoff_nest_handoff_history = handoff.nest_handoff_history if handoff else None
+        should_nest_history = (
+            handoff_nest_handoff_history
+            if handoff_nest_handoff_history is not None
+            else run_config.nest_handoff_history
+        )
+        if input_filter is None and should_nest_history:
+            has_nested_handoff_history = True
+
+    if not has_remove_all_tools and not has_nested_handoff_history:
+        return
+
+    conflict_parts: list[str] = []
+    if has_nested_handoff_history:
+        conflict_parts.append("nest_handoff_history")
+    if has_remove_all_tools:
+        conflict_parts.append("remove_all_tools")
+
+    raise UserError(
+        "Server-managed conversation (conversation_id / previous_response_id / "
+        "auto_previous_response_id) is incompatible with handoff settings: "
+        + ", ".join(conflict_parts)
+        + ". Use nest_handoff_history=False and avoid remove_all_tools."
+    )
+
+
 def resolve_trace_settings(
     *,
     run_state: RunState[TContext] | None,

tests/test_agent_runner.py

@@ -36,6 +36,7 @@
 )
 from agents.agent import ToolsToFinalOutputResult
 from agents.computer import Computer
+from agents.extensions.handoff_filters import remove_all_tools
 from agents.items import (
     HandoffOutputItem,
     ModelResponse,
@@ -2096,6 +2097,150 @@ async def test_run_streamed_rejects_session_with_resumed_conversation_state():
         Runner.run_streamed(agent, state, session=session)
 
 
+@pytest.mark.asyncio
+async def test_run_rejects_server_conversation_with_nested_handoff_history():
+    model = FakeModel()
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+
+    with pytest.raises(UserError, match="Server-managed conversation"):
+        await Runner.run(
+            triage,
+            input="test",
+            conversation_id="conv-test",
+            run_config=RunConfig(nest_handoff_history=True),
+        )
+
+
+@pytest.mark.asyncio
+async def test_run_rejects_server_conversation_with_remove_all_tools_filter():
+    model = FakeModel()
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(
+        name="triage",
+        model=model,
+        handoffs=[handoff(delegate, input_filter=remove_all_tools)],
+    )
+
+    with pytest.raises(UserError, match="Server-managed conversation"):
+        await Runner.run(
+            triage,
+            input="test",
+            conversation_id="conv-test",
+        )
+
+
+@pytest.mark.asyncio
+async def test_run_rejects_server_conversation_with_global_remove_all_tools_filter():
+    model = FakeModel()
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+
+    with pytest.raises(UserError, match="Server-managed conversation"):
+        await Runner.run(
+            triage,
+            input="test",
+            conversation_id="conv-test",
+            run_config=RunConfig(handoff_input_filter=remove_all_tools),
+        )
+
+
+@pytest.mark.asyncio
+async def test_run_allows_server_conversation_with_disabled_remove_all_tools_handoff():
+    model = FakeModel()
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(
+        name="triage",
+        model=model,
+        handoffs=[handoff(delegate, input_filter=remove_all_tools, is_enabled=False)],
+    )
+
+    result = await Runner.run(
+        triage,
+        input="test",
+        conversation_id="conv-test",
+    )
+
+    assert result.last_agent == triage
+
+
+@pytest.mark.asyncio
+async def test_run_allows_server_conversation_with_disabled_nested_handoff_history():
+    model = FakeModel()
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(
+        name="triage",
+        model=model,
+        handoffs=[handoff(delegate, nest_handoff_history=True, is_enabled=False)],
+    )
+
+    result = await Runner.run(
+        triage,
+        input="test",
+        conversation_id="conv-test",
+        run_config=RunConfig(nest_handoff_history=True),
+    )
+
+    assert result.last_agent == triage
+
+
+@pytest.mark.asyncio
+async def test_run_streamed_rejects_server_conversation_with_nested_handoff_history():
+    model = FakeModel()
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+
+    with pytest.raises(UserError, match="Server-managed conversation"):
+        Runner.run_streamed(
+            triage,
+            input="test",
+            conversation_id="conv-test",
+            run_config=RunConfig(nest_handoff_history=True),
+        )
+
+
+@pytest.mark.asyncio
+async def test_run_rejects_resumed_server_conversation_with_nested_handoff_history():
+    model = FakeModel()
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+    context_wrapper = RunContextWrapper(context=None)
+    state = RunState(
+        context=context_wrapper,
+        original_input="hello",
+        starting_agent=triage,
+        conversation_id="conv-test",
+    )
+
+    with pytest.raises(UserError, match="Server-managed conversation"):
+        await Runner.run(
+            triage,
+            state,
+            run_config=RunConfig(nest_handoff_history=True),
+        )
+
+
+@pytest.mark.asyncio
+async def test_run_streamed_rejects_resumed_server_conversation_with_nested_handoff_history():
+    model = FakeModel()
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+    context_wrapper = RunContextWrapper(context=None)
+    state = RunState(
+        context=context_wrapper,
+        original_input="hello",
+        starting_agent=triage,
+        conversation_id="conv-test",
+    )
+
+    with pytest.raises(UserError, match="Server-managed conversation"):
+        Runner.run_streamed(
+            triage,
+            state,
+            run_config=RunConfig(nest_handoff_history=True),
+        )
+
+
 @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."""