Python: Add stable names to orchestration workflows

python/packages/orchestrations/agent_framework_orchestrations/_concurrent.py

@@ -30,6 +30,7 @@
 from ._workflow_builder import OrchestrationWorkflowBuilder as WorkflowBuilder
 
 logger = logging.getLogger(__name__)
+DEFAULT_WORKFLOW_NAME = "Concurrent"
 
 """Concurrent builder for agent-only fan-out/fan-in workflows.
 
@@ -213,6 +214,7 @@ def summarize(results: list[AgentExecutorResponse]) -> str:
     def __init__(
         self,
         *,
+        name: str | None = None,
         participants: Sequence[SupportsAgentRun | Executor],
         checkpoint_storage: CheckpointStorage | None = None,
         output_from: Sequence[_ParticipantOutputSpecifier] | Literal["all"] | None = cast(Any, UNSET),
@@ -221,6 +223,7 @@ def __init__(
         """Initialize the ConcurrentBuilder.
 
         Args:
+            name: Optional workflow identifier. Defaults to ``"Concurrent"``.
             participants: Sequence of agent or executor instances to run in parallel.
             checkpoint_storage: Optional checkpoint storage for enabling workflow state persistence.
             output_from: Optional participant names or instances whose ``yield_output`` calls
@@ -230,6 +233,7 @@ def __init__(
                 surface as workflow ``intermediate`` events. Pass ``"all_other"`` to select every participant
                 not selected by ``output_from``. Unlisted participant outputs are hidden.
         """
+        self._name = name or DEFAULT_WORKFLOW_NAME
         self._participants: list[SupportsAgentRun | Executor] = []
         self._aggregator: Executor | None = None
         self._checkpoint_storage: CheckpointStorage | None = checkpoint_storage
@@ -421,6 +425,7 @@ def build(self) -> Workflow:
             extra_output_executors=[aggregator],
         )
         builder = WorkflowBuilder(
+            name=self._name,
             start_executor=dispatcher,
             checkpoint_storage=self._checkpoint_storage,
             output_from=designated,

python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py

@@ -69,6 +69,7 @@
     from typing_extensions import override  # pragma: no cover
 
 logger = logging.getLogger(__name__)
+DEFAULT_WORKFLOW_NAME = "GroupChat"
 
 
 @dataclass(frozen=True)
@@ -629,6 +630,7 @@ class GroupChatBuilder:
     def __init__(
         self,
         *,
+        name: str | None = None,
         participants: Sequence[SupportsAgentRun | Executor] | None = None,
         participant_factories: Sequence[Callable[[], SupportsAgentRun | Executor]] | None = None,
         # Orchestrator config (exactly one required)
@@ -646,6 +648,7 @@ def __init__(
         """Initialize the GroupChatBuilder.
 
         Args:
+            name: Optional workflow identifier. Defaults to ``"GroupChat"``.
             participants: Optional sequence of agent or executor instances for the group chat.
             participant_factories: Optional sequence of callables returning agent or executor instances.
             orchestrator_agent: An instance of Agent or a callable that produces one to manage the group chat.
@@ -665,6 +668,7 @@ def __init__(
                 surface as workflow ``intermediate`` events. Pass ``"all_other"`` to select every participant
                 not selected by ``output_from``. Unlisted participant outputs are hidden.
         """
+        self._name = name or DEFAULT_WORKFLOW_NAME
         self._participants: dict[str, SupportsAgentRun | Executor] = {}
         self._participant_factories: list[Callable[[], SupportsAgentRun | Executor]] = []
 
@@ -1039,6 +1043,7 @@ def build(self) -> Workflow:
             extra_output_executors=[orchestrator],
         )
         workflow_builder = WorkflowBuilder(
+            name=self._name,
             start_executor=orchestrator,
             checkpoint_storage=self._checkpoint_storage,
             output_from=designated,

python/packages/orchestrations/agent_framework_orchestrations/_handoff.py

@@ -71,6 +71,7 @@
 
 
 logger = logging.getLogger(__name__)
+DEFAULT_WORKFLOW_NAME = "Handoff"
 
 
 # region Handoff events
@@ -614,7 +615,7 @@ def __init__(
 
         Args:
             name: Optional workflow identifier used in logging and debugging.
-                  If not provided, a default name will be generated.
+                Defaults to ``"Handoff"``.
             participants: Optional list of ``Agent`` instances that will participate in the handoff workflow.
                           You can also call `.participants([...])` later. Each participant must have a
                           unique identifier (`.name` is preferred if set, otherwise `.id` is used).
@@ -630,7 +631,7 @@ def __init__(
                 surface as workflow ``intermediate`` events. Pass ``"all_other"`` to select every participant
                 not selected by ``output_from``. Unlisted participant outputs are hidden.
         """
-        self._name = name
+        self._name = name or DEFAULT_WORKFLOW_NAME
         self._description = description
 
         # Participant related members

python/packages/orchestrations/agent_framework_orchestrations/_magentic.py

@@ -55,6 +55,7 @@
 
 
 logger = logging.getLogger(__name__)
+DEFAULT_WORKFLOW_NAME = "Magentic"
 
 # Consistent author name for messages produced by the Magentic manager/orchestrator
 MAGENTIC_MANAGER_NAME = "magentic_manager"
@@ -1422,6 +1423,7 @@ def __init__(
         max_reset_count: int | None = None,
         max_round_count: int | None = None,
         # Existing params
+        name: str | None = None,
         enable_plan_review: bool = False,
         checkpoint_storage: CheckpointStorage | None = None,
         output_from: Sequence[_ParticipantOutputSpecifier] | Literal["all"] | None = cast(Any, UNSET),
@@ -1430,6 +1432,7 @@ def __init__(
         """Initialize the Magentic workflow builder.
 
         Args:
+            name: Optional workflow identifier. Defaults to ``"Magentic"``.
             participants: Sequence of agent or executor instances for the workflow.
             manager: Pre-configured manager instance (subclass of MagenticManagerBase).
             manager_factory: Callable that returns a new MagenticManagerBase instance.
@@ -1455,6 +1458,7 @@ def __init__(
                 surface as workflow ``intermediate`` events. Pass ``"all_other"`` to select every participant
                 not selected by ``output_from``. Unlisted participant outputs are hidden.
         """
+        self._name = name or DEFAULT_WORKFLOW_NAME
         self._participants: dict[str, SupportsAgentRun | Executor] = {}
 
         # Manager related members
@@ -1794,6 +1798,7 @@ def build(self) -> Workflow:
             extra_output_executors=[orchestrator],
         )
         workflow_builder = WorkflowBuilder(
+            name=self._name,
             start_executor=orchestrator,
             checkpoint_storage=self._checkpoint_storage,
             output_from=designated,

python/packages/orchestrations/agent_framework_orchestrations/_sequential.py

@@ -44,6 +44,7 @@
 from ._workflow_builder import OrchestrationWorkflowBuilder as WorkflowBuilder
 
 logger = logging.getLogger(__name__)
+DEFAULT_WORKFLOW_NAME = "Sequential"
 
 
 class _InputToConversation(Executor):
@@ -98,6 +99,7 @@ class SequentialBuilder:
     def __init__(
         self,
         *,
+        name: str | None = None,
         participants: Sequence[SupportsAgentRun | Executor],
         checkpoint_storage: CheckpointStorage | None = None,
         chain_only_agent_responses: bool = False,
@@ -107,6 +109,7 @@ def __init__(
         """Initialize the SequentialBuilder.
 
         Args:
+            name: Optional workflow identifier. Defaults to ``"Sequential"``.
             participants: Sequence of agent or executor instances to run sequentially.
             checkpoint_storage: Optional checkpoint storage for enabling workflow state persistence.
             chain_only_agent_responses: If True, only agent responses are chained between agents.
@@ -118,6 +121,7 @@ def __init__(
                 surface as workflow ``intermediate`` events. Pass ``"all_other"`` to select every participant
                 not selected by ``output_from``. Unlisted participant outputs are hidden.
         """
+        self._name = name or DEFAULT_WORKFLOW_NAME
         self._participants: list[SupportsAgentRun | Executor] = []
         self._checkpoint_storage: CheckpointStorage | None = checkpoint_storage
         self._chain_only_agent_responses: bool = chain_only_agent_responses
@@ -258,6 +262,7 @@ def build(self) -> Workflow:
             default_output_from=[participants[-1]],
         )
         builder = WorkflowBuilder(
+            name=self._name,
             start_executor=input_conv,
             checkpoint_storage=self._checkpoint_storage,
             output_from=designated,

python/packages/orchestrations/tests/test_concurrent.py

@@ -55,6 +55,15 @@ def test_concurrent_builder_rejects_duplicate_executors() -> None:
         ConcurrentBuilder(participants=[a, b])
 
 
+def test_concurrent_builder_uses_stable_default_and_custom_name() -> None:
+    participants = [_FakeAgentExec("agentA", "A"), _FakeAgentExec("agentB", "B")]
+
+    assert ConcurrentBuilder(participants=participants).build().name == "Concurrent"
+    assert (
+        ConcurrentBuilder(name="custom-concurrent", participants=participants).build().name == "custom-concurrent"
+    )
+
+
 async def test_concurrent_default_aggregator_emits_assistants_only() -> None:
     """Default aggregator yields a single AgentResponse with one assistant message per participant.
 
@@ -210,6 +219,7 @@ async def test_concurrent_checkpoint_resume_round_trip() -> None:
     )
 
     wf = ConcurrentBuilder(participants=list(participants), checkpoint_storage=storage).build()
+    assert wf.name == "Concurrent"
 
     baseline_output: AgentResponse | None = None
     async for ev in wf.run("checkpoint concurrent", stream=True):

python/packages/orchestrations/tests/test_group_chat.py

@@ -254,6 +254,23 @@ async def test_group_chat_builder_basic_flow() -> None:
     assert updates[0].author_name == "manager"
 
 
+def test_group_chat_builder_uses_stable_default_and_custom_name() -> None:
+    participant = StubAgent("agent", "response")
+
+    default_workflow = GroupChatBuilder(
+        participants=[participant],
+        selection_func=make_sequence_selector(),
+    ).build()
+    custom_workflow = GroupChatBuilder(
+        name="custom-group-chat",
+        participants=[participant],
+        selection_func=make_sequence_selector(),
+    ).build()
+
+    assert default_workflow.name == "GroupChat"
+    assert custom_workflow.name == "custom-group-chat"
+
+
 async def test_group_chat_as_agent_accepts_conversation() -> None:
     selector = make_sequence_selector()
     alpha = StubAgent("alpha", "ack from alpha")
@@ -495,6 +512,7 @@ def selector(state: GroupChatState) -> str:
             selection_func=selector,
         ).build()
 
+        assert workflow.name == "GroupChat"
         updates: list[AgentResponseUpdate] = []
         async for event in workflow.run("test task", stream=True):
             if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
@@ -660,6 +678,7 @@ def selector(state: GroupChatState) -> str:
             checkpoint_storage=storage,
             selection_func=selector,
         ).build()
+        assert workflow.name == "GroupChat"
 
         updates: list[AgentResponseUpdate] = []
         async for event in workflow.run("test task", stream=True):

python/packages/orchestrations/tests/test_handoff.py

@@ -261,6 +261,28 @@ async def test_handoff():
     assert request.source_executor_id == escalation.name
 
 
+def test_handoff_builder_uses_stable_default_and_custom_name() -> None:
+    participant = MockHandoffAgent(name="agent")
+
+    default_workflow = (
+        HandoffBuilder(participants=[participant], termination_condition=lambda _: True)
+        .with_start_agent(participant)
+        .build()
+    )
+    custom_workflow = (
+        HandoffBuilder(
+            name="custom-handoff",
+            participants=[participant],
+            termination_condition=lambda _: True,
+        )
+        .with_start_agent(participant)
+        .build()
+    )
+
+    assert default_workflow.name == "Handoff"
+    assert custom_workflow.name == "custom-handoff"
+
+
 def _latest_request_info_event(events: list[WorkflowEvent]) -> WorkflowEvent[Any]:
     request_events = [event for event in events if event.type == "request_info"]
     assert request_events

python/packages/orchestrations/tests/test_magentic.py

@@ -201,6 +201,24 @@ def test_magentic_builder_marks_feature_with_custom_manager() -> None:
     assert int(token.split(".", 1)[1], 16) & (1 << FeatureIndex.ORCHESTRATION_MAGENTIC)
 
 
+def test_magentic_builder_uses_stable_default_and_custom_name() -> None:
+    default_manager = FakeManager()
+    custom_manager = FakeManager()
+
+    default_workflow = MagenticBuilder(
+        participants=[DummyExec(name=default_manager.next_speaker_name)],
+        manager=default_manager,
+    ).build()
+    custom_workflow = MagenticBuilder(
+        name="custom-magentic",
+        participants=[DummyExec(name=custom_manager.next_speaker_name)],
+        manager=custom_manager,
+    ).build()
+
+    assert default_workflow.name == "Magentic"
+    assert custom_workflow.name == "custom-magentic"
+
+
 async def test_magentic_builder_returns_workflow_and_runs() -> None:
     manager = FakeManager()
     agent = StubAgent(manager.next_speaker_name, "first draft")
@@ -429,6 +447,7 @@ async def test_magentic_checkpoint_resume_round_trip():
         manager=manager1,
     ).build()
 
+    assert wf.name == "Magentic"
     task_text = "checkpoint task"
     req_event: WorkflowEvent | None = None
     async for ev in wf.run(task_text, stream=True):

python/packages/orchestrations/tests/test_sequential.py

@@ -114,6 +114,13 @@ def test_sequential_builder_rejects_empty_participants() -> None:
         SequentialBuilder(participants=[])
 
 
+def test_sequential_builder_uses_stable_default_and_custom_name() -> None:
+    participant = _EchoAgent(name="echo")
+
+    assert SequentialBuilder(participants=[participant]).build().name == "Sequential"
+    assert SequentialBuilder(name="custom-sequential", participants=[participant]).build().name == "custom-sequential"
+
+
 def test_sequential_builder_validation_rejects_invalid_executor() -> None:
     """Test that adding an invalid executor to the builder raises an error."""
     with pytest.raises(TypeCompatibilityError):
@@ -212,6 +219,7 @@ async def test_sequential_checkpoint_resume_round_trip() -> None:
 
     initial_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
     wf = SequentialBuilder(participants=list(initial_agents), checkpoint_storage=storage).build()
+    assert wf.name == "Sequential"
 
     baseline_updates: list[AgentResponseUpdate] = []
     async for ev in wf.run("checkpoint sequential", stream=True):