Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Python: Core: address AgentModeProvider review feedback
- default_mode now defaults to None and falls back to the first configured
  mode, decoupling the kwarg from the built-in 'plan'/'execute' set.
- get_agent_mode catches ValueError when a previously persisted mode is no
  longer in available_modes and resets to the default mode (matching the
  non-string recovery branch). Added regression coverage for both behaviors.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
  • Loading branch information
eavanvalkenburg and Copilot committed May 5, 2026
commit f3ddb884beddd5539a7ca5519c79ddb9e173f34d
34 changes: 24 additions & 10 deletions python/packages/core/agent_framework/_harness/_mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,12 +72,19 @@ def _normalize_mode(mode: str, *, available_modes: Mapping[str, str]) -> str:
return normalized


def _resolve_default_mode(default_mode: str | None, *, available_modes: Mapping[str, str]) -> str:
"""Resolve the default mode, falling back to the first configured mode when omitted."""
if default_mode is None:
return next(iter(available_modes))
return _normalize_mode(default_mode, available_modes=available_modes)


@experimental(feature_id=ExperimentalFeature.HARNESS)
def get_agent_mode(
session: AgentSession,
*,
source_id: str = DEFAULT_MODE_SOURCE_ID,
default_mode: str = "plan",
default_mode: str | None = None,
available_modes: Sequence[str] | None = None,
) -> str:
"""Get the current operating mode from session state.
Expand All @@ -87,20 +94,26 @@ def get_agent_mode(

Keyword Args:
source_id: Unique source ID for the provider state.
default_mode: Initial mode used when no mode is stored yet.
default_mode: Initial mode used when no mode is stored yet. When omitted, the first entry of
``available_modes`` is used.
available_modes: Supported modes to validate against. Defaults to the built-in modes.

Returns:
The current mode string.
"""
normalized_modes = _normalize_available_modes(tuple(available_modes or DEFAULT_MODE_DESCRIPTIONS))
normalized_default_mode = _normalize_mode(default_mode, available_modes=normalized_modes)
normalized_default_mode = _resolve_default_mode(default_mode, available_modes=normalized_modes)
provider_state = _get_mode_state(session, source_id=source_id)
current_mode = provider_state.get("current_mode")
Comment thread
eavanvalkenburg marked this conversation as resolved.
Resolved
if not isinstance(current_mode, str):
provider_state["current_mode"] = normalized_default_mode
return normalized_default_mode
return _normalize_mode(current_mode, available_modes=normalized_modes)
if isinstance(current_mode, str):
try:
return _normalize_mode(current_mode, available_modes=normalized_modes)
except ValueError:
# Stored mode is no longer in the configured set (e.g. available_modes was reconfigured).
# Fall through and reset to the default mode.
pass
provider_state["current_mode"] = normalized_default_mode
return normalized_default_mode


@experimental(feature_id=ExperimentalFeature.HARNESS)
Expand Down Expand Up @@ -157,7 +170,7 @@ def __init__(
self,
source_id: str = DEFAULT_MODE_SOURCE_ID,
*,
default_mode: str = "plan",
default_mode: str | None = None,
mode_descriptions: Mapping[str, str] | None = None,
instructions: str | None = None,
) -> None:
Expand All @@ -167,7 +180,8 @@ def __init__(
source_id: Unique source ID for the provider.

Keyword Args:
default_mode: Initial mode used when no mode is stored yet.
default_mode: Initial mode used when no mode is stored yet. When omitted, the first entry of
``mode_descriptions`` is used.
mode_descriptions: Mapping of supported modes to descriptions of when and how to use each mode.
instructions: Custom instructions for using the mode tools. The instructions can contain an
``{available_modes}`` placeholder for the configured list of modes and a ``{current_mode}`` placeholder
Expand All @@ -183,7 +197,7 @@ def __init__(
raise ValueError("mode_descriptions must contain at least one mode.")
self.mode_descriptions = {mode.strip().lower(): description for mode, description in mode_descriptions.items()}
self.available_modes = tuple(self._mode_display_names)
self.default_mode = _normalize_mode(default_mode, available_modes=self._mode_display_names)
self.default_mode = _resolve_default_mode(default_mode, available_modes=self._mode_display_names)
self.instructions = instructions

def _build_instructions(self, current_mode: str) -> str:
Expand Down
22 changes: 22 additions & 0 deletions python/packages/core/tests/core/test_harness_mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -167,3 +167,25 @@ async def test_agent_mode_context_provider_updates_agent_mode(
assert json.loads(set_result[0].text) == {"mode": "execute", "message": "Mode changed to 'execute'."}
assert get_agent_mode(session, source_id=provider.source_id) == "execute"
assert set_agent_mode(session, "plan", source_id=provider.source_id) == "plan"


def test_default_mode_falls_back_to_first_available_mode() -> None:
"""When ``default_mode`` is omitted, helpers and provider should use the first configured mode."""
session = AgentSession(session_id="session-1")

assert get_agent_mode(session, available_modes=("draft", "final")) == "draft"

provider = AgentModeProvider(mode_descriptions={"Draft": "Draft it.", "Final": "Finalize it."})
assert provider.default_mode == "draft"


def test_get_agent_mode_falls_back_when_stored_mode_not_in_available_modes() -> None:
"""A previously persisted mode that is no longer configured should be reset to the default."""
session = AgentSession(session_id="session-1")
set_agent_mode(session, "execute")
assert session.state[DEFAULT_MODE_SOURCE_ID]["current_mode"] == "execute"

# Reconfigure with a smaller mode set that no longer includes "execute".
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"
Loading