Python: Support an autonomous handoff flow (#2497)

* Support an autonomous handoff flow.

* Simplify public API

* Address feedback

python/packages/core/tests/workflow/test_handoff.py

@@ -289,6 +289,140 @@ async def test_tool_call_handoff_detection_with_text_hint():
     assert len(specialist.calls[0]) >= 2
 
 
+async def test_autonomous_interaction_mode_yields_output_without_user_request():
+    """Ensure autonomous interaction mode yields output without requesting user input."""
+    triage = _RecordingAgent(name="triage", handoff_to="specialist")
+    specialist = _RecordingAgent(name="specialist")
+
+    workflow = (
+        HandoffBuilder(participants=[triage, specialist])
+        .set_coordinator("triage")
+        .with_interaction_mode("autonomous", autonomous_turn_limit=1)
+        .build()
+    )
+
+    events = await _drain(workflow.run_stream("Package arrived broken"))
+    assert len(triage.calls) == 1
+    assert len(specialist.calls) == 1
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert not requests, "Autonomous mode should not request additional user input"
+
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs, "Autonomous mode should yield a workflow output"
+
+    final_conversation = outputs[-1].data
+    assert isinstance(final_conversation, list)
+    conversation_list = cast(list[ChatMessage], final_conversation)
+    assert any(
+        msg.role == Role.ASSISTANT and (msg.text or "").startswith("specialist reply") for msg in conversation_list
+    )
+
+
+async def test_autonomous_continues_without_handoff_until_termination():
+    """Autonomous mode should keep invoking the same agent when no handoff occurs."""
+    worker = _RecordingAgent(name="worker")
+
+    workflow = (
+        HandoffBuilder(participants=[worker])
+        .set_coordinator(worker)
+        .with_interaction_mode("autonomous", autonomous_turn_limit=3)
+        .with_termination_condition(lambda conv: False)
+        .build()
+    )
+
+    events = await _drain(workflow.run_stream("Start"))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs, "Autonomous mode should yield output after termination condition"
+    assert len(worker.calls) == 3, "Worker should be invoked multiple times without user input"
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert not requests, "Autonomous mode should not request user input"
+
+
+async def test_autonomous_turn_limit_stops_loop():
+    """Autonomous mode should stop when the configured turn limit is reached."""
+    worker = _RecordingAgent(name="worker")
+
+    workflow = (
+        HandoffBuilder(participants=[worker])
+        .set_coordinator(worker)
+        .with_interaction_mode("autonomous", autonomous_turn_limit=2)
+        .with_termination_condition(lambda conv: False)
+        .build()
+    )
+
+    events = await _drain(workflow.run_stream("Start"))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs, "Turn limit should force a workflow output"
+    assert len(worker.calls) == 2, "Worker should stop after reaching the turn limit"
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert not requests, "Autonomous mode should not request user input"
+
+
+async def test_autonomous_routes_back_to_coordinator_when_specialist_stops():
+    """Specialist without handoff should route back to coordinator in autonomous mode."""
+    triage = _RecordingAgent(name="triage", handoff_to="specialist")
+    specialist = _RecordingAgent(name="specialist")
+
+    workflow = (
+        HandoffBuilder(participants=[triage, specialist])
+        .set_coordinator(triage)
+        .add_handoff(triage, specialist)
+        .with_interaction_mode("autonomous", autonomous_turn_limit=3)
+        .with_termination_condition(lambda conv: len(conv) >= 4)
+        .build()
+    )
+
+    events = await _drain(workflow.run_stream("Issue"))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs, "Workflow should complete without user input"
+    assert len(specialist.calls) >= 1, "Specialist should run without handoff"
+
+
+async def test_autonomous_mode_with_inline_turn_limit():
+    """Autonomous mode should respect turn limit passed via with_interaction_mode."""
+    worker = _RecordingAgent(name="worker")
+
+    workflow = (
+        HandoffBuilder(participants=[worker])
+        .set_coordinator(worker)
+        .with_interaction_mode("autonomous", autonomous_turn_limit=2)
+        .with_termination_condition(lambda conv: False)
+        .build()
+    )
+
+    events = await _drain(workflow.run_stream("Start"))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs, "Turn limit should force a workflow output"
+    assert len(worker.calls) == 2, "Worker should stop after reaching the inline turn limit"
+
+
+def test_autonomous_turn_limit_ignored_in_human_in_loop_mode(caplog):
+    """Verify that autonomous_turn_limit logs a warning when mode is human_in_loop."""
+    worker = _RecordingAgent(name="worker")
+
+    # Should not raise, but should log a warning
+    HandoffBuilder(participants=[worker]).set_coordinator(worker).with_interaction_mode(
+        "human_in_loop", autonomous_turn_limit=10
+    )
+
+    assert "autonomous_turn_limit=10 was provided but interaction_mode is 'human_in_loop'; ignoring." in caplog.text
+
+
+def test_autonomous_turn_limit_must_be_positive():
+    """Verify that autonomous_turn_limit raises an error when <= 0."""
+    worker = _RecordingAgent(name="worker")
+
+    with pytest.raises(ValueError, match="autonomous_turn_limit must be positive"):
+        HandoffBuilder(participants=[worker]).set_coordinator(worker).with_interaction_mode(
+            "autonomous", autonomous_turn_limit=0
+        )
+
+    with pytest.raises(ValueError, match="autonomous_turn_limit must be positive"):
+        HandoffBuilder(participants=[worker]).set_coordinator(worker).with_interaction_mode(
+            "autonomous", autonomous_turn_limit=-5
+        )
+
+
 def test_build_fails_without_coordinator():
     """Verify that build() raises ValueError when set_coordinator() was not called."""
     triage = _RecordingAgent(name="triage")