From 8722645d3e7eb42bb7cb9e3dfd83fe983ac8b4f6 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Tue, 5 May 2026 13:38:50 +0200 Subject: [PATCH] Python: Core: notify agent of external AgentModeProvider mode changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the operating mode is changed externally (e.g. via a slash-command handler calling set_agent_mode), the agent's chat history still shows the prior set_mode tool call near the end. Updating only the system instructions is insufficient — models tend to anchor on the recent tool call and ignore the new mode. Mirror the .NET AgentModeProvider behavior: when set_agent_mode detects an actual mode change, record the previous mode in provider state. On the next before_run, the provider pops that flag and injects a user-role notification message announcing the switch, so the most recent context unambiguously reflects the current mode. The agent-driven set_mode tool path bypasses this so it does not trigger a redundant notification on its own change. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../core/agent_framework/_harness/_mode.py | 46 +++++++++++-- .../core/tests/core/test_harness_mode.py | 66 +++++++++++++++++++ 2 files changed, 106 insertions(+), 6 deletions(-) diff --git a/python/packages/core/agent_framework/_harness/_mode.py b/python/packages/core/agent_framework/_harness/_mode.py index a79285b14c..e34df0dffa 100644 --- a/python/packages/core/agent_framework/_harness/_mode.py +++ b/python/packages/core/agent_framework/_harness/_mode.py @@ -9,6 +9,7 @@ from .._feature_stage import ExperimentalFeature, experimental from .._sessions import AgentSession, ContextProvider, SessionContext from .._tools import tool +from .._types import Message DEFAULT_MODE_SOURCE_ID = "agent_mode" DEFAULT_MODE_INSTRUCTIONS = ( @@ -22,6 +23,10 @@ "\n" "You are currently operating in the {current_mode} mode.\n" ) +DEFAULT_MODE_CHANGE_NOTIFICATION = ( + '[Mode changed: The operating mode has been switched from "{previous_mode}" to "{current_mode}". ' + 'You must now adjust your behavior to match the "{current_mode}" mode.]' +) DEFAULT_MODE_DESCRIPTIONS: dict[str, str] = { "plan": ( "Use this mode when analyzing requirements, breaking down tasks, and creating plans. " @@ -36,6 +41,8 @@ ), } +_PREVIOUS_MODE_STATE_KEY = "previous_mode_for_notification" + def _get_mode_state(session: AgentSession, *, source_id: str) -> dict[str, Any]: """Return the mutable session state used by the mode provider.""" @@ -126,6 +133,12 @@ def set_agent_mode( ) -> str: """Set the current operating mode in session state. + External callers (e.g., a slash-command handler) should use this helper rather than mutating + session state directly. When the mode actually changes, the prior mode is recorded so that the + next :meth:`AgentModeProvider.before_run` invocation injects a user message announcing the + switch — system instructions alone are insufficient to redirect a model that has already seen + its own ``set_mode`` tool call earlier in the chat history. + Args: session: The agent session to update the mode in. mode: The new mode to set. @@ -143,7 +156,14 @@ def set_agent_mode( normalized_modes = _normalize_available_modes(tuple(available_modes or DEFAULT_MODE_DESCRIPTIONS)) normalized_mode = _normalize_mode(mode, available_modes=normalized_modes) provider_state = _get_mode_state(session, source_id=source_id) + previous_mode = provider_state.get("current_mode") provider_state["current_mode"] = normalized_mode + # When the mode is changed externally (i.e. not via the agent's own ``set_mode`` tool), record the + # prior mode so the next ``before_run`` can inject a user message announcing the switch. Without + # that injection, the model often anchors on the earlier ``set_mode`` tool call in the chat history + # and keeps behaving as if it were still in that mode — system instructions alone are insufficient. + if isinstance(previous_mode, str) and previous_mode != normalized_mode: + provider_state[_PREVIOUS_MODE_STATE_KEY] = previous_mode return normalized_mode @@ -232,16 +252,19 @@ async def before_run( default_mode=self.default_mode, available_modes=self.available_modes, ) + # Pop the external-mode-change marker (set by ``set_agent_mode``) before injecting tools so + # the agent only sees the notification once. + provider_state = _get_mode_state(session, source_id=self.source_id) + previous_mode = provider_state.pop(_PREVIOUS_MODE_STATE_KEY, None) @tool(name="set_mode", approval_mode="never_require") def set_mode(mode: str) -> str: """Switch the agent's operating mode.""" - normalized_mode = set_agent_mode( - session, - mode, - source_id=self.source_id, - available_modes=self.available_modes, - ) + # The agent invoked the tool itself, so it knows the mode just changed — bypass + # ``set_agent_mode`` to avoid triggering a notification message on the next turn. + normalized_mode = _normalize_mode(mode, available_modes=self._mode_display_names) + tool_state = _get_mode_state(session, source_id=self.source_id) + tool_state["current_mode"] = normalized_mode return json.dumps({"mode": normalized_mode, "message": f"Mode changed to '{normalized_mode}'."}) @tool(name="get_mode", approval_mode="never_require") @@ -260,3 +283,14 @@ def get_mode() -> str: [self._build_instructions(current_mode)], ) context.extend_tools(self.source_id, [set_mode, get_mode]) + if isinstance(previous_mode, str) and previous_mode != current_mode: + # Inject a user-role message announcing the external mode change. System instructions + # always render first in the chat history, so the agent can otherwise stay anchored to + # the most recent ``set_mode`` tool call rather than the new mode. + previous_display = self._mode_display_names.get(previous_mode, previous_mode) + current_display = self._mode_display_names.get(current_mode, current_mode) + notification = DEFAULT_MODE_CHANGE_NOTIFICATION.format( + previous_mode=previous_display, + current_mode=current_display, + ) + context.extend_messages(self, [Message(role="user", contents=[notification])]) diff --git a/python/packages/core/tests/core/test_harness_mode.py b/python/packages/core/tests/core/test_harness_mode.py index d11653bd27..cb63007343 100644 --- a/python/packages/core/tests/core/test_harness_mode.py +++ b/python/packages/core/tests/core/test_harness_mode.py @@ -189,3 +189,69 @@ def test_get_agent_mode_falls_back_when_stored_mode_not_in_available_modes() -> current = get_agent_mode(session, default_mode="draft", available_modes=("draft", "final")) assert current == "draft" assert session.state[DEFAULT_MODE_SOURCE_ID]["current_mode"] == "draft" + + +def test_set_agent_mode_records_previous_mode_for_external_change_notification() -> None: + """External mode changes via ``set_agent_mode`` should record the previous mode for notification.""" + session = AgentSession(session_id="session-1") + set_agent_mode(session, "plan") + set_agent_mode(session, "execute") + + assert session.state[DEFAULT_MODE_SOURCE_ID]["current_mode"] == "execute" + assert session.state[DEFAULT_MODE_SOURCE_ID]["previous_mode_for_notification"] == "plan" + + +def test_set_agent_mode_no_op_does_not_record_previous_mode() -> None: + """Setting the same mode should not queue a notification.""" + session = AgentSession(session_id="session-1") + set_agent_mode(session, "plan") + set_agent_mode(session, "plan") + + assert "previous_mode_for_notification" not in session.state[DEFAULT_MODE_SOURCE_ID] + + +async def test_agent_mode_provider_injects_user_message_after_external_change( + chat_client_base: SupportsChatGetResponse, +) -> None: + """External mode changes should inject a user message announcing the switch on the next run.""" + session = AgentSession(session_id="session-1") + provider = AgentModeProvider() + agent = Agent(client=chat_client_base, context_providers=[provider]) + + # First run: agent uses set_mode tool to switch to execute. The tool path must NOT queue a + # notification because the agent already saw its own tool call in the chat history. + _, first_options = await agent._prepare_session_and_messages( # type: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Plan first."])], + ) + set_mode_tool = _tool_by_name(first_options["tools"], "set_mode") + await set_mode_tool.invoke(arguments={"mode": "execute"}) + assert "previous_mode_for_notification" not in session.state[provider.source_id] + + # Now an external caller (e.g., a /mode slash command) switches the mode back to plan. + set_agent_mode(session, "plan", source_id=provider.source_id) + assert session.state[provider.source_id]["previous_mode_for_notification"] == "execute" + + # Next run: the provider should inject a user message announcing the change and clear the flag. + second_context, second_options = await agent._prepare_session_and_messages( # type: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Carry on."])], + ) + instructions = second_options["instructions"] + assert isinstance(instructions, str) + assert "You are currently operating in the plan mode." in instructions + + notification_messages = [message for message in second_context.context_messages.get(provider.source_id, [])] + assert len(notification_messages) == 1 + assert notification_messages[0].role == "user" + assert "Mode changed" in notification_messages[0].text + assert '"execute"' in notification_messages[0].text + assert '"plan"' in notification_messages[0].text + assert "previous_mode_for_notification" not in session.state[provider.source_id] + + # Third run with no further external change must not re-inject the notification. + third_context, _ = await agent._prepare_session_and_messages( # type: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Status?"])], + ) + assert third_context.context_messages.get(provider.source_id, []) == []