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

src/agents/run.py

@@ -3,7 +3,7 @@
 import asyncio
 import contextlib
 import warnings
-from typing import Union, cast
+from typing import Any, Union, cast
 
 from typing_extensions import Unpack
 
@@ -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
@@ -409,6 +410,21 @@ async def run(
         if run_config is None:
             run_config = RunConfig()
 
+        def validate_conversation_settings(active_agent_for_validation: Agent[Any]) -> None:
+            validate_session_conversation_settings(
+                session,
+                conversation_id=conversation_id,
+                previous_response_id=previous_response_id,
+                auto_previous_response_id=auto_previous_response_id,
+            )
+            validate_server_conversation_handoff_settings(
+                active_agent_for_validation,
+                run_config,
+                conversation_id=conversation_id,
+                previous_response_id=previous_response_id,
+                auto_previous_response_id=auto_previous_response_id,
+            )
+
         is_resumed_state = isinstance(input, RunState)
         run_state: RunState[TContext] | None = None
         starting_input = input if not is_resumed_state else None
@@ -432,11 +448,8 @@ async def run(
                 previous_response_id=previous_response_id,
                 auto_previous_response_id=auto_previous_response_id,
             )
-            validate_session_conversation_settings(
-                session,
-                conversation_id=conversation_id,
-                previous_response_id=previous_response_id,
-                auto_previous_response_id=auto_previous_response_id,
+            validate_conversation_settings(
+                run_state._current_agent if run_state._current_agent else starting_agent
             )
             starting_input = run_state._original_input
             original_user_input = copy_input_items(run_state._original_input)
@@ -453,12 +466,7 @@ async def run(
             raw_input = cast(Union[str, list[TResponseInputItem]], input)
             original_user_input = raw_input
 
-            validate_session_conversation_settings(
-                session,
-                conversation_id=conversation_id,
-                previous_response_id=previous_response_id,
-                auto_previous_response_id=auto_previous_response_id,
-            )
+            validate_conversation_settings(starting_agent)
 
             server_manages_conversation = (
                 conversation_id is not None
@@ -1422,6 +1430,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 +1472,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,110 @@ 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):
+                    # Still validate handoff-level settings even when target resolution is deferred.
+                    edges.append((current, handoff))
+                    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,