Python: add option to disable return-to-previous handoff routing

python/packages/orchestrations/README.md

@@ -51,6 +51,19 @@ workflow = (
 )
 ```
 
+By default, each user response returns to the agent that requested it. To have the start agent re-evaluate every user
+turn, disable return-to-previous routing:
+
+```python
+workflow = (
+    HandoffBuilder()
+    .participants([triage, billing, support])
+    .with_start_agent(triage)
+    .enable_return_to_previous(False)
+    .build()
+)
+```
+
 ### GroupChatBuilder
 
 Orchestrator-directed multi-agent conversations:

python/packages/orchestrations/agent_framework_orchestrations/_handoff.py

@@ -241,6 +241,7 @@ def __init__(
         *,
         agent_session: AgentSession | None = None,
         is_start_agent: bool = False,
+        user_response_target_id: str | None = None,
         termination_condition: TerminationCondition | None = None,
         autonomous_mode: bool = False,
         autonomous_mode_prompt: str | None = None,
@@ -254,6 +255,8 @@ def __init__(
             agent_session: Optional AgentSession that manages the agent's execution context
             is_start_agent: Whether this agent is the starting agent in the handoff workflow.
                             There can only be one starting agent in a handoff workflow.
+            user_response_target_id: Optional executor ID that should handle user responses.
+                                     Defaults to this executor.
             termination_condition: Optional callable that determines when to terminate the workflow
             autonomous_mode: Whether the agent should operate involve external systems after
                              a response that does not trigger a handoff or before the turn
@@ -271,6 +274,7 @@ def __init__(
         self._handoff_targets = {handoff.target_id for handoff in handoffs}
         self._termination_condition = termination_condition
         self._is_start_agent = is_start_agent
+        self._user_response_target_id = user_response_target_id
 
         # Autonomous mode members
         self._autonomous_mode = autonomous_mode
@@ -510,8 +514,15 @@ async def handle_response(
         # Broadcast the user response to all other agents
         await self._broadcast_messages(response, ctx)
 
-        # Append the user response messages to the cache
+        # Keep this executor synchronized even when another agent handles the response.
         self._cache.extend(response)
+        if self._user_response_target_id is not None and self._user_response_target_id != self.id:
+            await ctx.send_message(
+                AgentExecutorRequest(messages=[], should_respond=True),
+                target_id=self._user_response_target_id,
+            )
+            return
+
         await self._run_agent_and_emit(ctx)
 
     async def _broadcast_messages(
@@ -623,6 +634,8 @@ class HandoffBuilder:
     Agents can hand off to other agents using `.add_handoff()`. This provides a decentralized
     approach to multi-agent collaboration. Handoffs can be configured using `.add_handoff`. If
     none are specified, all agents can hand off to all others by default (making a mesh topology).
+    By default, user responses return to the agent that requested them. Use
+    `.enable_return_to_previous(False)` to route user responses through the start agent instead.
 
     Participants must be ``Agent`` instances. ``SupportsAgentRun`` protocol implementors that
     are not ``Agent`` subclasses are not supported because handoff workflows require cloning,
@@ -690,6 +703,7 @@ def __init__(
 
         # Handoff related members
         self._handoff_config: dict[str, set[HandoffConfiguration]] = {}
+        self._return_to_previous: bool = True
 
         # Checkpoint related members
         self._checkpoint_storage: CheckpointStorage | None = checkpoint_storage
@@ -864,6 +878,33 @@ def with_start_agent(self, agent: Agent) -> "HandoffBuilder":
 
         return self
 
+    def enable_return_to_previous(self, enabled: bool = True) -> "HandoffBuilder":
+        """Configure whether user responses return to the agent that requested them.
+
+        Return-to-previous routing is enabled by default. Disable it to route each user response
+        through the configured start agent so that agent can re-evaluate the conversation before
+        handing off again.
+
+        Args:
+            enabled: Whether user responses should return to the agent that requested them.
+
+        Returns:
+            Self for method chaining.
+
+        Example:
+
+        .. code-block:: python
+
+            workflow = (
+                HandoffBuilder(participants=[triage, billing, support])
+                .with_start_agent(triage)
+                .enable_return_to_previous(False)
+                .build()
+            )
+        """
+        self._return_to_previous = enabled
+        return self
+
     def with_autonomous_mode(
         self,
         *,
@@ -1155,6 +1196,7 @@ def _resolve_executors(
                 agent=agent,
                 handoffs=handoffs.get(resolved_id, []),
                 is_start_agent=(id == self._start_id),
+                user_response_target_id=None if self._return_to_previous else self._start_id,
                 termination_condition=self._termination_condition,
                 autonomous_mode=autonomous_mode,
                 autonomous_mode_prompt=self._autonomous_mode_prompts.get(id, None),

python/packages/orchestrations/tests/test_handoff.py

@@ -84,6 +84,7 @@ def __init__(
         self._name = name
         self._handoff_to = handoff_to
         self._call_index = 0
+        self.received_messages: list[list[Message]] = []
 
     def _inner_get_response(
         self,
@@ -93,6 +94,7 @@ def _inner_get_response(
         options: Mapping[str, Any],
         **kwargs: Any,
     ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
+        self.received_messages.append(list(messages))
         if stream:
             return self._build_streaming_response(options=dict(options))
 
@@ -303,6 +305,71 @@ def _request_text(event: WorkflowEvent[Any]) -> str:
     return messages[-1].text or ""
 
 
+@pytest.mark.parametrize(
+    ("return_to_previous", "expected_start_agent_calls"),
+    [
+        param(None, 1, id="default"),
+        param(True, 1, id="enabled"),
+        param(False, 2, id="disabled"),
+    ],
+)
+async def test_handoff_user_response_routing(
+    return_to_previous: bool | None,
+    expected_start_agent_calls: int,
+) -> None:
+    """User responses return to the prior agent unless configured to revisit the start agent."""
+    triage_client = MockChatClient(name="triage", handoff_to="specialist")
+    specialist_client = MockChatClient(name="specialist")
+    triage = Agent(
+        id="triage",
+        name="triage",
+        client=triage_client,
+        require_per_service_call_history_persistence=True,
+    )
+    specialist = Agent(
+        id="specialist",
+        name="specialist",
+        client=specialist_client,
+        require_per_service_call_history_persistence=True,
+    )
+    builder = HandoffBuilder(
+        participants=_as_handoff_agents(triage, specialist),
+        termination_condition=lambda _: False,
+    ).with_start_agent(_as_handoff_agent(triage))
+    if return_to_previous is True:
+        builder.enable_return_to_previous()
+    elif return_to_previous is False:
+        builder.enable_return_to_previous(False)
+    workflow = builder.build()
+
+    first_events = await _drain(workflow.run("Need specialist help", stream=True))
+    first_request = _latest_request_info_event(first_events)
+    assert first_request.source_executor_id == specialist.name
+    assert len(triage_client.received_messages) == 1
+    assert len(specialist_client.received_messages) == 1
+
+    second_events = await _drain(
+        workflow.run(
+            stream=True,
+            responses={first_request.request_id: [Message(role="user", contents=["Additional details"])]},
+        )
+    )
+    second_request = _latest_request_info_event(second_events)
+
+    assert second_request.source_executor_id == specialist.name
+    assert len(triage_client.received_messages) == expected_start_agent_calls
+    assert len(specialist_client.received_messages) == 2
+    if return_to_previous is False:
+        assert any(
+            message.role == "user" and message.text == "Additional details"
+            for message in triage_client.received_messages[-1]
+        )
+    assert any(
+        message.role == "user" and message.text == "Additional details"
+        for message in specialist_client.received_messages[-1]
+    )
+
+
 async def test_resume_keeps_prior_user_context_for_same_agent() -> None:
     """Ensure same-agent request_info resumes retain prior turn context."""
     refund_agent = Agent(