Python: Core: notify agent of external AgentModeProvider mode changes

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])])

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, []) == []