Python: Fix Magentic manager duplicating conversation history

python/packages/orchestrations/agent_framework_orchestrations/_magentic.py

@@ -15,7 +15,6 @@
 from agent_framework import (
     AgentResponse,
     AgentResponseUpdate,
-    AgentSession,
     Message,
     SupportsAgentRun,
 )
@@ -569,7 +568,6 @@ def __init__(
         )
 
         self._agent: SupportsAgentRun = agent
-        self._session: AgentSession = self._agent.create_session()
         self.task_ledger: _MagenticTaskLedger | None = task_ledger
 
         # Prompts may be overridden if needed
@@ -597,8 +595,20 @@ async def _complete(
 
         The agent's run method is called which applies the agent's configured options
         (temperature, seed, instructions, etc.).
+
+        A *fresh* session is created for every call instead of reusing a persistent one.
+        The manager already passes the complete conversation it wants the model to see
+        on each call (see ``plan``, ``replan`` and ``create_progress_ledger``, which all
+        build ``[*magentic_context.chat_history, ...]``). Reusing a single accumulating
+        session would make the agent's history provider (the default
+        ``InMemoryHistoryProvider`` for local sessions) reload every previously sent /
+        received message and prepend it to the input, so the task, facts and plan would
+        be duplicated and compound on every round. A throwaway session keeps each call
+        stateless while still propagating a non-``None`` session, so any context
+        providers configured on the manager agent are still invoked (regression #4371).
         """
-        response: AgentResponse = await self._agent.run(messages, session=self._session)
+        session = self._agent.create_session()
+        response: AgentResponse = await self._agent.run(messages, session=session)
         if not response.messages:
             raise RuntimeError("Agent returned no messages in response.")
         if len(response.messages) > 1:
@@ -743,7 +753,6 @@ def on_checkpoint_save(self) -> dict[str, Any]:
         state: dict[str, Any] = {}
         if self.task_ledger is not None:
             state["task_ledger"] = self.task_ledger.to_dict()
-        state["agent_session"] = self._session.to_dict()
         return state
 
     @override
@@ -754,12 +763,6 @@ def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
                 self.task_ledger = _MagenticTaskLedger.from_dict(ledger)
             except Exception:  # pragma: no cover - defensive
                 logger.warning("Failed to restore manager task ledger from checkpoint state")
-        session_payload = state.get("agent_session")
-        if session_payload is not None:
-            try:
-                self._session = AgentSession.from_dict(session_payload)
-            except Exception:  # pragma: no cover - defensive
-                logger.warning("Failed to restore manager agent session from checkpoint state")
 
 
 # endregion Magentic Manager

python/packages/orchestrations/tests/test_magentic.py

@@ -8,10 +8,13 @@
 
 import pytest
 from agent_framework import (
+    Agent,
     AgentResponse,
     AgentResponseUpdate,
     AgentSession,
     BaseAgent,
+    BaseChatClient,
+    ChatResponse,
     Content,
     Executor,
     Message,
@@ -1207,39 +1210,95 @@ async def _run() -> AgentResponse[Any]:
 
     await mgr.plan(ctx.clone())
 
-    # plan() calls _complete twice (facts + plan), both should receive the same session
+    # plan() calls _complete twice (facts + plan). Each call must receive a non-None
+    # session so context providers configured on the manager agent are still invoked
+    # (the original intent of regression #4371).
     assert len(captured_sessions) == 2
     assert all(s is not None for s in captured_sessions), "session must be passed to agent.run()"
-    assert captured_sessions[0] is captured_sessions[1], "same session instance must be reused across calls"
-    assert captured_sessions[0] is mgr._session
+    # Each call must use a *fresh* session rather than one shared, accumulating session.
+    # The manager re-passes the full conversation on every call, so a reused session
+    # would make the agent's history provider re-inject prior turns and duplicate the
+    # task/facts/plan each round. See test_standard_manager_does_not_duplicate_history.
+    assert captured_sessions[0] is not captured_sessions[1], "each call must use a fresh session"
+
+
+async def test_standard_manager_does_not_duplicate_history():
+    """Regression: the manager must not re-send already-sent turns to the model.
+
+    The manager rebuilds the full conversation it wants the model to see on every call
+    (``[*chat_history, facts_user]`` then ``[*chat_history, facts_user, facts_msg, plan_user]``).
+    Previously it reused one persistent ``AgentSession`` across calls, so the agent's
+    default ``InMemoryHistoryProvider`` reloaded the first call's stored messages and
+    prepended them to the second call's input, duplicating the facts pre-survey (and, over
+    multiple rounds, compounding the whole task/facts/plan). This drives a real ``Agent``
+    through the real session machinery and asserts no such duplication reaches the client.
+    """
+    facts_marker = "Below I will present you a request."
+    plan_marker = "Fantastic. To address this request"
+
+    class RecordingChatClient(BaseChatClient):
+        """Captures the exact message list handed to the model on each call."""
+
+        def __init__(self) -> None:
+            super().__init__()
+            self.calls: list[list[Message]] = []
+
+        @override
+        def _inner_get_response(self, *, messages, stream, options, **kwargs):  # type: ignore[override]
+            # Snapshot the fully-merged messages (session history + input) the model sees.
+            self.calls.append(list(messages))
+
+            async def _get() -> ChatResponse:
+                return ChatResponse(messages=Message(role="assistant", contents=["recorded"]))
 
+            return _get()
 
-def test_standard_manager_checkpoint_preserves_session():
-    """Verify that checkpoint save/restore preserves the manager's session identity."""
-    agent = StubManagerAgent()
+    client = RecordingChatClient()
+    agent = Agent(name="MagenticManager", client=client)
     mgr = StandardMagenticManager(agent=agent)
-    original_session_id = mgr._session.session_id
+    ctx = MagenticContext(task="Is the system healthy?", participant_descriptions={"a": "desc"})
+
+    await mgr.plan(ctx.clone())
+
+    # plan() makes two model calls: the facts call, then the plan call.
+    assert len(client.calls) == 2
+    facts_call, plan_call = client.calls
+
+    # The facts pre-survey is sent once on the facts call...
+    assert sum(facts_marker in m.text for m in facts_call) == 1
+    # ...and must appear exactly once on the plan call too (the manager includes it
+    # manually). A reused/accumulating session would make it appear twice.
+    assert sum(facts_marker in m.text for m in plan_call) == 1, "facts pre-survey duplicated across calls"
+    # The plan prompt itself is present exactly once on the plan call.
+    assert sum(plan_marker in m.text for m in plan_call) == 1
+
+
+def test_standard_manager_checkpoint_preserves_task_ledger():
+    """Checkpoint save/restore round-trips the manager's task ledger (its only persisted state)."""
+    from agent_framework_orchestrations._magentic import _MagenticTaskLedger  # type: ignore
+
+    mgr = StandardMagenticManager(agent=StubManagerAgent())
+    mgr.task_ledger = _MagenticTaskLedger(
+        facts=Message("assistant", ["Custom facts"]),
+        plan=Message("assistant", ["Custom plan"]),
+    )
 
     state = mgr.on_checkpoint_save()
-    assert "agent_session" in state
+    assert "task_ledger" in state
 
-    # Restore into a fresh manager and verify session_id is preserved
-    mgr2 = StandardMagenticManager(agent=agent)
-    assert mgr2._session.session_id != original_session_id
+    mgr2 = StandardMagenticManager(agent=StubManagerAgent())
+    assert mgr2.task_ledger is None
     mgr2.on_checkpoint_restore(state)
-    assert mgr2._session.session_id == original_session_id
+    assert mgr2.task_ledger is not None
+    assert mgr2.task_ledger.facts.text == "Custom facts"
+    assert mgr2.task_ledger.plan.text == "Custom plan"
 
 
 def test_standard_manager_checkpoint_restore_empty_state():
-    """Verify that restoring from a state without agent_session leaves the session intact."""
-    agent = StubManagerAgent()
-    mgr = StandardMagenticManager(agent=agent)
-    original_session = mgr._session
-    original_session_id = original_session.session_id
-
+    """Restoring from a state without a task ledger leaves the manager unchanged."""
+    mgr = StandardMagenticManager(agent=StubManagerAgent())
     mgr.on_checkpoint_restore({})
-    assert mgr._session is original_session
-    assert mgr._session.session_id == original_session_id
+    assert mgr.task_ledger is None
 
 
 # endregion