From addbeb276c2248e27390e477613ea85737b6a2f5 Mon Sep 17 00:00:00 2001 From: Hassan Ghominejad Date: Wed, 3 Jun 2026 11:43:44 +0400 Subject: [PATCH 1/2] Python: Fix Magentic manager duplicating conversation history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _complete() reused one persistent AgentSession, so the default history provider re-injected prior turns on top of the full prompt the manager already rebuilds each call — duplicating task/facts/plan and compounding every round. Use a fresh session per call; keep self._session only for checkpointing. GroupChatOrchestrator is unaffected. Add a regression test and update the session-propagation test. --- .../_magentic.py | 16 ++++- .../orchestrations/tests/test_magentic.py | 69 ++++++++++++++++++- 2 files changed, 81 insertions(+), 4 deletions(-) diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py index 53ca4052ff..212f9ebd77 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py @@ -569,6 +569,9 @@ def __init__( ) self._agent: SupportsAgentRun = agent + # Retained only for checkpoint save/restore continuity. LLM calls in `_complete` + # use a fresh session each time so the agent's history provider cannot re-inject + # (and thereby duplicate) the conversation the manager already passes in full. self._session: AgentSession = self._agent.create_session() self.task_ledger: _MagenticTaskLedger | None = task_ledger @@ -597,8 +600,19 @@ 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 ``self._session``. + 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) + response: AgentResponse = await self._agent.run(messages, session=self._agent.create_session()) if not response.messages: raise RuntimeError("Agent returned no messages in response.") if len(response.messages) > 1: diff --git a/python/packages/orchestrations/tests/test_magentic.py b/python/packages/orchestrations/tests/test_magentic.py index 5c94d2fb14..d0d57de865 100644 --- a/python/packages/orchestrations/tests/test_magentic.py +++ b/python/packages/orchestrations/tests/test_magentic.py @@ -7,10 +7,13 @@ import pytest from agent_framework import ( + Agent, AgentResponse, AgentResponseUpdate, AgentSession, BaseAgent, + BaseChatClient, + ChatResponse, Content, Executor, Message, @@ -23,6 +26,10 @@ WorkflowRunState, handler, ) +from agent_framework_orchestrations._magentic import ( + ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT, + ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT, +) from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage from agent_framework.orchestrations import ( GroupChatRequestMessage, @@ -1165,11 +1172,67 @@ async def _run() -> AgentResponse: 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 = ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT.split("{")[0].strip() + plan_marker = ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT.split("{")[0].strip() + + 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() + + client = RecordingChatClient() + agent = Agent(name="MagenticManager", client=client) + mgr = StandardMagenticManager(agent=agent) + 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_session(): From 177920d877bcd489f75b2e8c1ebc06a83c0f6e9e Mon Sep 17 00:00:00 2001 From: Hassan Ghominejad Date: Mon, 29 Jun 2026 11:16:57 +0400 Subject: [PATCH 2/2] Python: Clean up Magentic manager per Copilot review (drop dead _session) --- .../_magentic.py | 17 ++----- .../orchestrations/tests/test_magentic.py | 44 +++++++++---------- 2 files changed, 23 insertions(+), 38 deletions(-) diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py index 212f9ebd77..fbaee46fb5 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py @@ -15,7 +15,6 @@ from agent_framework import ( AgentResponse, AgentResponseUpdate, - AgentSession, Message, SupportsAgentRun, ) @@ -569,10 +568,6 @@ def __init__( ) self._agent: SupportsAgentRun = agent - # Retained only for checkpoint save/restore continuity. LLM calls in `_complete` - # use a fresh session each time so the agent's history provider cannot re-inject - # (and thereby duplicate) the conversation the manager already passes in full. - self._session: AgentSession = self._agent.create_session() self.task_ledger: _MagenticTaskLedger | None = task_ledger # Prompts may be overridden if needed @@ -601,7 +596,7 @@ 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 ``self._session``. + 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 @@ -612,7 +607,8 @@ async def _complete( 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._agent.create_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: @@ -757,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 @@ -768,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 diff --git a/python/packages/orchestrations/tests/test_magentic.py b/python/packages/orchestrations/tests/test_magentic.py index d0d57de865..2cf17a4f50 100644 --- a/python/packages/orchestrations/tests/test_magentic.py +++ b/python/packages/orchestrations/tests/test_magentic.py @@ -26,10 +26,6 @@ WorkflowRunState, handler, ) -from agent_framework_orchestrations._magentic import ( - ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT, - ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT, -) from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage from agent_framework.orchestrations import ( GroupChatRequestMessage, @@ -1195,8 +1191,8 @@ async def test_standard_manager_does_not_duplicate_history(): 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 = ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT.split("{")[0].strip() - plan_marker = ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT.split("{")[0].strip() + 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.""" @@ -1235,32 +1231,32 @@ async def _get() -> ChatResponse: assert sum(plan_marker in m.text for m in plan_call) == 1 -def test_standard_manager_checkpoint_preserves_session(): - """Verify that checkpoint save/restore preserves the manager's session identity.""" - agent = StubManagerAgent() - mgr = StandardMagenticManager(agent=agent) - original_session_id = mgr._session.session_id +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