Python: Create Magentic managers per workflow

python/packages/orchestrations/agent_framework_orchestrations/_magentic.py

@@ -10,6 +10,7 @@
 from collections.abc import Callable, Sequence
 from dataclasses import dataclass, field
 from enum import Enum
+from functools import partial
 from typing import Any, ClassVar, Literal, TypeVar, cast
 
 from agent_framework import (
@@ -1435,9 +1436,14 @@ def __init__(
             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.
-            manager_agent: Agent instance for creating a StandardMagenticManager.
-            manager_agent_factory: Callable that returns a new agent instance for creating a StandardMagenticManager.
+                The same instance, including its mutable state, is reused by every workflow
+                built with this builder. Do not share a stateful manager across concurrent or
+                interleaved workflows; use manager_factory to create a fresh manager per build.
+            manager_factory: Callable invoked on each build to return a new MagenticManagerBase instance.
+            manager_agent: Agent instance used to create a new StandardMagenticManager on each build.
+                Each workflow has its own manager, but the supplied agent is shared.
+            manager_agent_factory: Callable invoked on each build to return a new agent instance
+                for a new StandardMagenticManager.
             task_ledger: Optional custom task ledger (used with manager_agent/manager_agent_factory).
             task_ledger_facts_prompt: Custom prompt for extracting facts.
             task_ledger_plan_prompt: Custom prompt for generating initial plan.
@@ -1691,7 +1697,8 @@ def _log_warning_if_constructor_args_provided() -> None:
             self._manager = manager
             _log_warning_if_constructor_args_provided()
         elif manager_agent is not None:
-            self._manager = StandardMagenticManager(
+            self._manager_factory = partial(
+                StandardMagenticManager,
                 agent=manager_agent,
                 task_ledger=task_ledger,
                 task_ledger_facts_prompt=task_ledger_facts_prompt,
@@ -1781,7 +1788,12 @@ def _resolve_participants(self) -> list[Executor]:
         return executors
 
     def build(self) -> Workflow:
-        """Build a Magentic workflow with the orchestrator and all agent executors."""
+        """Build a Magentic workflow with the orchestrator and all agent executors.
+
+        A new standard manager is created for each build when configured with manager_agent
+        or manager_agent_factory. An explicitly supplied manager instance is reused unchanged;
+        manager_factory is called on each build to obtain a manager.
+        """
         mark_feature_used(FeatureIndex.ORCHESTRATION_MAGENTIC)
         logger.info(f"Building Magentic workflow with {len(self._participants)} participants")
 

python/packages/orchestrations/tests/test_magentic.py

@@ -1102,12 +1102,17 @@ def agent_factory() -> SupportsAgentRun:
     assert event_count > 0
 
 
-def test_magentic_agent_factory_uses_default_max_stall_count() -> None:
+@pytest.mark.parametrize("use_agent_factory", [False, True])
+def test_magentic_manager_agent_uses_default_max_stall_count(use_agent_factory: bool) -> None:
     def agent_factory() -> SupportsAgentRun:
         return cast(SupportsAgentRun, StubManagerAgent())
 
     participant = StubAgent("agentA", "reply from agentA")
-    workflow = MagenticBuilder(participants=[participant], manager_agent_factory=agent_factory).build()
+    workflow = MagenticBuilder(
+        participants=[participant],
+        manager_agent=None if use_agent_factory else StubManagerAgent(),
+        manager_agent_factory=agent_factory if use_agent_factory else None,
+    ).build()
 
     orchestrator = next(e for e in workflow.executors.values() if isinstance(e, MagenticOrchestrator))
     manager = orchestrator._manager  # type: ignore[reportPrivateUsage]
@@ -1116,6 +1121,103 @@ def agent_factory() -> SupportsAgentRun:
     assert manager.max_stall_count == 3
 
 
+def test_magentic_manager_agent_reusable_builder() -> None:
+    agent = StubManagerAgent()
+    builder = MagenticBuilder(participants=[StubAgent("agentA", "reply")], manager_agent=agent)
+
+    assert builder._manager is None
+    workflow1 = builder.build()
+    workflow2 = builder.build()
+    orchestrator1 = next(e for e in workflow1.executors.values() if isinstance(e, MagenticOrchestrator))
+    orchestrator2 = next(e for e in workflow2.executors.values() if isinstance(e, MagenticOrchestrator))
+    manager1 = orchestrator1._manager
+    manager2 = orchestrator2._manager
+
+    assert isinstance(manager1, StandardMagenticManager)
+    assert isinstance(manager2, StandardMagenticManager)
+    assert manager1 is not manager2
+    assert manager1._agent is manager2._agent is agent
+    assert manager1.task_ledger is None
+    assert manager2.task_ledger is None
+
+
+def test_magentic_supplied_manager_reusable_builder_preserves_identity() -> None:
+    manager = FakeManager()
+    builder = MagenticBuilder(participants=[StubAgent("agentA", "reply")], manager=manager)
+
+    workflow1 = builder.build()
+    workflow2 = builder.build()
+    orchestrator1 = next(e for e in workflow1.executors.values() if isinstance(e, MagenticOrchestrator))
+    orchestrator2 = next(e for e in workflow2.executors.values() if isinstance(e, MagenticOrchestrator))
+
+    assert orchestrator1 is not orchestrator2
+    assert orchestrator1._manager is orchestrator2._manager is manager
+
+
+@pytest.mark.parametrize("use_agent_factory", [False, True])
+async def test_magentic_reusable_builder_keeps_interleaved_task_ledgers_separate(use_agent_factory: bool) -> None:
+    class RecordingChatClient(BaseChatClient):
+        def __init__(self) -> None:
+            super().__init__()
+            self.calls: list[list[Message]] = []
+
+        @override
+        def _inner_get_response(self, *, messages, stream, options, **kwargs):  # type: ignore[override]
+            self.calls.append(list(messages))
+
+            async def _get() -> ChatResponse:
+                return ChatResponse(messages=Message("assistant", [messages[-1].text]))
+
+            return _get()
+
+    client = RecordingChatClient()
+    agent = Agent(name="manager", client=client)
+    builder = MagenticBuilder(
+        participants=[StubAgent("agentA", "reply")],
+        manager_agent=None if use_agent_factory else agent,
+        manager_agent_factory=(lambda: agent) if use_agent_factory else None,
+        enable_plan_review=True,
+        task_ledger_facts_prompt="FACTS:{task}",
+        task_ledger_plan_prompt="PLAN",
+        task_ledger_facts_update_prompt="UPDATE:{task}:{old_facts}",
+        task_ledger_plan_update_prompt="REVISED PLAN",
+    )
+    workflow_a = builder.build()
+    workflow_b = builder.build()
+
+    events_a = [event async for event in workflow_a.run("A_ONLY", stream=True)]
+    events_b = [event async for event in workflow_b.run("B_ONLY", stream=True)]
+    review_a = next(event for event in events_a if event.type == "request_info")
+    review_b = next(event for event in events_b if event.type == "request_info")
+    assert isinstance(review_a.data, MagenticPlanReviewRequest)
+    assert isinstance(review_b.data, MagenticPlanReviewRequest)
+    assert len(client.calls) == 4
+
+    revised_events = [
+        event
+        async for event in workflow_a.run(
+            stream=True, responses={review_a.request_id: review_a.data.revise("Revise the plan")}
+        )
+    ]
+    assert len(client.calls) == 6
+    assert client.calls[4][-1].text == "UPDATE:A_ONLY:FACTS:A_ONLY"
+    assert all("B_ONLY" not in message.text for call in client.calls[4:] for message in call)
+    revised_review = next(event for event in revised_events if event.type == "request_info")
+    assert isinstance(revised_review.data, MagenticPlanReviewRequest)
+    assert "UPDATE:A_ONLY:FACTS:A_ONLY" in revised_review.data.plan.text
+    assert "B_ONLY" not in revised_review.data.plan.text
+
+    _ = [
+        event
+        async for event in workflow_b.run(
+            stream=True, responses={review_b.request_id: review_b.data.revise("Revise the plan")}
+        )
+    ]
+    assert len(client.calls) == 8
+    assert client.calls[6][-1].text == "UPDATE:B_ONLY:FACTS:B_ONLY"
+    assert all("A_ONLY" not in message.text for call in client.calls[6:] for message in call)
+
+
 async def test_magentic_manager_factory_reusable_builder():
     """Test that the builder can be reused to build multiple workflows with manager factory."""
     factory_call_count = 0
@@ -1140,10 +1242,12 @@ def manager_factory() -> MagenticManagerBase:
     orchestrator1 = next(e for e in wf1.executors.values() if isinstance(e, MagenticOrchestrator))
     orchestrator2 = next(e for e in wf2.executors.values() if isinstance(e, MagenticOrchestrator))
     assert orchestrator1 is not orchestrator2
+    assert orchestrator1._manager is not orchestrator2._manager
 
 
-def test_magentic_agent_factory_with_standard_manager_options():
-    """Test that agent_factory properly passes through standard manager options."""
+@pytest.mark.parametrize("use_agent_factory", [False, True])
+def test_magentic_manager_agent_with_standard_manager_options(use_agent_factory: bool) -> None:
+    """Both manager agent configurations pass through standard manager options."""
     factory_call_count = 0
 
     def agent_factory() -> SupportsAgentRun:
@@ -1174,7 +1278,8 @@ def agent_factory() -> SupportsAgentRun:
     participant = StubAgent("agentA", "reply from agentA")
     workflow = MagenticBuilder(
         participants=[participant],
-        manager_agent_factory=agent_factory,
+        manager_agent=None if use_agent_factory else StubManagerAgent(),
+        manager_agent_factory=agent_factory if use_agent_factory else None,
         task_ledger=custom_task_ledger,
         max_stall_count=custom_max_stall_count,
         max_reset_count=custom_max_reset_count,
@@ -1189,7 +1294,7 @@ def agent_factory() -> SupportsAgentRun:
     ).build()
 
     # Factory should be called during build
-    assert factory_call_count == 1
+    assert factory_call_count == (1 if use_agent_factory else 0)
 
     # Get the orchestrator and verify the manager has the custom options
     orchestrator = next(e for e in workflow.executors.values() if isinstance(e, MagenticOrchestrator))