From 1bed6c084b129015fe685d0e0351a4a93b2f3c74 Mon Sep 17 00:00:00 2001 From: Aditya Kumar Date: Wed, 7 Oct 2026 01:34:27 +0000 Subject: [PATCH] Python: Reset Magentic participant sessions on replan MagenticAgentExecutor.handle_magentic_reset assigned the new session to _agent_thread, an attribute nothing reads since the AgentExecutor session attribute was renamed to _session, so participants kept their whole conversation across every reset. Assign _session instead and also clear _pending_request_order with the other pending request state. --- .../_magentic.py | 3 +- .../orchestrations/tests/test_magentic.py | 50 +++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py index 6206022f043..0fb8797ab57 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py @@ -1374,8 +1374,9 @@ async def handle_magentic_reset(self, signal: MagenticResetSignal, ctx: Workflow # Request into related self._pending_agent_requests.clear() self._pending_responses_to_agent.clear() + self._pending_request_order.clear() # Reset sessions - self._agent_thread = self._agent.create_session() + self._session = self._agent.create_session() # endregion Magentic Agent Executor diff --git a/python/packages/orchestrations/tests/test_magentic.py b/python/packages/orchestrations/tests/test_magentic.py index 38a3dcc718c..c8a50db34ea 100644 --- a/python/packages/orchestrations/tests/test_magentic.py +++ b/python/packages/orchestrations/tests/test_magentic.py @@ -857,6 +857,56 @@ async def test_magentic_stall_and_reset_reach_limits(): assert output_event.data.text == "Workflow terminated due to reaching maximum reset count." +async def test_magentic_reset_gives_participants_a_fresh_session(): + """After a stall forces a reset and replan, participants must not see their earlier history.""" + + class RecordingChatClient(BaseChatClient): + """Captures the message texts handed to the model on each call.""" + + def __init__(self) -> None: + super().__init__() + self.calls: list[list[str]] = [] + + @override + def _inner_get_response(self, *, messages, stream, options, **kwargs): # type: ignore[override] + self.calls.append([m.text for m in messages]) + reply = f"reply-{len(self.calls)}" + + async def _get() -> ChatResponse: + return ChatResponse(messages=Message(role="assistant", contents=[reply])) + + return _get() + + class StallOnceManager(NotProgressingManager): + """Asks agentA, stalls once (forcing a reset and replan), asks agentA again, then finishes.""" + + def __init__(self) -> None: + super().__init__(max_round_count=10, max_stall_count=0) + self.rounds = 0 + + async def create_progress_ledger(self, magentic_context: MagenticContext) -> MagenticProgressLedger: + self.rounds += 1 + instruction = "before reset" if self.rounds == 1 else "after reset" + progress = self.rounds != 2 + return MagenticProgressLedger( + is_request_satisfied=MagenticProgressLedgerItem(reason="r", answer=self.rounds >= 4), + is_in_loop=MagenticProgressLedgerItem(reason="r", answer=not progress), + is_progress_being_made=MagenticProgressLedgerItem(reason="r", answer=progress), + next_speaker=MagenticProgressLedgerItem(reason="r", answer="agentA"), + instruction_or_question=MagenticProgressLedgerItem(reason="r", answer=instruction), + ) + + client = RecordingChatClient() + manager = StallOnceManager() + wf = MagenticBuilder(participants=[Agent(name="agentA", client=client)], manager=manager).build() + + await wf.run("test reset") + + assert manager.rounds == 4 + assert client.calls[0] == ["before reset"] + assert client.calls[1] == ["after reset"], "agentA kept its pre-reset session history" + + async def test_magentic_checkpoint_runtime_only() -> None: """Test checkpointing configured ONLY at runtime, not at build time.""" storage = InMemoryCheckpointStorage()