diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py index c2fb6b549c..14ae940002 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,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 diff --git a/python/packages/orchestrations/tests/test_magentic.py b/python/packages/orchestrations/tests/test_magentic.py index af411cacb8..6551f72495 100644 --- a/python/packages/orchestrations/tests/test_magentic.py +++ b/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