Python: Core: notify agent of external AgentModeProvider mode changes (#5650)

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>
This commit is contained in:
Eduard van Valkenburg
2026-05-07 02:58:38 +00:00
committed by GitHub
co-authored by Copilot
parent cdd80c61ac
commit a95493a909
2 changed files with 106 additions and 6 deletions
@@ -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, []) == []