"""Tests for model switching functionality.""" from collections.abc import Iterator from typing import Any from unittest.mock import AsyncMock, Mock, patch import pytest from deepagents_code import model_config from deepagents_code.app import ( DeepAgentsApp, _extract_model_params_flag, _format_model_params, ) from deepagents_code.config import settings from deepagents_code.model_config import ( ModelSpec, ProviderAuthSource, ProviderAuthState, ProviderAuthStatus, clear_caches, ) from deepagents_code.remote_client import RemoteAgent from deepagents_code.widgets.messages import AppMessage, ErrorMessage _CONFIGURED_AUTH_STATUS = ProviderAuthStatus( state=ProviderAuthState.CONFIGURED, provider="anthropic", env_var="ANTHROPIC_API_KEY", source=ProviderAuthSource.ENV, ) """Generic non-blocking auth status for tests that bypass the credential check.""" def _make_remote_agent() -> RemoteAgent: """Create a RemoteAgent pointing at a dummy URL for test scaffolding.""" return RemoteAgent("http://test:0") class _FakeModelResult: """Minimal model result for `_switch_model` tests.""" def __init__( self, *, model_name: str, provider: str, context_limit: int, unsupported_modalities: frozenset[str] = frozenset(), ) -> None: self.model_name = model_name self.provider = provider self.context_limit = context_limit self.unsupported_modalities = unsupported_modalities def apply_to_settings(self) -> None: """Mirror `ModelResult.apply_to_settings()` for test isolation.""" settings.model_name = self.model_name settings.model_provider = self.provider settings.model_context_limit = self.context_limit settings.model_unsupported_modalities = self.unsupported_modalities @pytest.fixture(autouse=True) def _restore_settings() -> Iterator[None]: """Save and restore global settings mutated by tests.""" original_name = settings.model_name original_provider = settings.model_provider original_context_limit = settings.model_context_limit original_modalities = settings.model_unsupported_modalities yield settings.model_name = original_name settings.model_provider = original_provider settings.model_context_limit = original_context_limit settings.model_unsupported_modalities = original_modalities @pytest.fixture(autouse=True) def mock_create_model() -> Iterator[Mock]: """Avoid provider package imports while preserving metadata updates.""" context_limits = { "anthropic:claude-opus-4-5": 200_000, "anthropic:claude-sonnet-4-5": 200_000, "fireworks:llama-v3p1-70b": 131_072, "ollama:llama3": 8_192, "openai:gpt-5.5": 128_000, } def fake_create_model( model_spec: str, *, extra_kwargs: dict[str, object] | None = None, profile_overrides: dict[str, object] | None = None, ) -> _FakeModelResult: del extra_kwargs, profile_overrides parsed = ModelSpec.try_parse(model_spec) if parsed is None: provider = "openai" model_name = model_spec else: provider = parsed.provider model_name = parsed.model context_limit = context_limits.get(f"{provider}:{model_name}", 65_536) return _FakeModelResult( model_name=model_name, provider=provider, context_limit=context_limit, ) with patch( "deepagents_code.config.create_model", side_effect=fake_create_model, ) as mock: yield mock class TestFormatModelParams: """Tests for the `_format_model_params` rendering helper.""" def test_none_returns_empty_string(self) -> None: """`None` produces no suffix so callers can concatenate unconditionally.""" assert _format_model_params(None) == "" def test_empty_dict_returns_empty_string(self) -> None: """An empty dict has no params worth echoing — collapse to empty.""" assert _format_model_params({}) == "" def test_single_key_renders_with_leading_space(self) -> None: """Single key renders as a space-prefixed suffix that callers append.""" assert _format_model_params({"num_ctx": 16384}) == ( ' with model params {"num_ctx": 16384}' ) def test_keys_are_sorted_regardless_of_insertion_order(self) -> None: """`sort_keys=True` must produce stable output across call sites and dicts. Insertion-reversed keys would render in insertion order without `sort_keys=True`, so this test would fail if the flag were dropped. """ params = {"temperature": 0.2, "num_ctx": 16384} assert _format_model_params(params) == ( ' with model params {"num_ctx": 16384, "temperature": 0.2}' ) def test_string_values_are_json_escaped(self) -> None: """Values containing quotes must be JSON-escaped, not interpolated raw.""" assert _format_model_params({"stop": '"end"'}) == ( ' with model params {"stop": "\\"end\\""}' ) class TestModelSwitchNoOp: """Tests for no-op when switching to the same model.""" async def test_no_message_when_switching_to_same_model(self) -> None: """Switching to the already-active model should not print 'Switched to'. This is a regression test for the bug where selecting the same model from the model selector would print "Switched to X" even though no actual switch occurred. """ app = DeepAgentsApp() # Replace method with mock to track calls (hence ignore) app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() # Set current model settings.model_name = "claude-opus-4-5" settings.model_provider = "anthropic" captured_messages: list[str] = [] original_init = AppMessage.__init__ def capture_init(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_init(self, message, **kwargs) with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch.object(AppMessage, "__init__", capture_init), ): # Attempt to switch to the same model await app._switch_model("anthropic:claude-opus-4-5") # Should show "Already using" message, not "Switched to" # Type checker doesn't track that _mount_message was replaced with mock app._mount_message.assert_called_once() # ty: ignore assert len(captured_messages) == 1 assert "Already using" in captured_messages[0] assert "Switched to" not in captured_messages[0] assert app._model_switching is False async def test_duplicate_same_model_message_is_suppressed(self) -> None: """Repeated same-model switches should only emit one unchanged notice.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "claude-opus-4-5" settings.model_provider = "anthropic" captured_messages: list[str] = [] original_init = AppMessage.__init__ def capture_init(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_init(self, message, **kwargs) with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch.object(AppMessage, "__init__", capture_init), ): await app._switch_model("anthropic:claude-opus-4-5") await app._switch_model("anthropic:claude-opus-4-5") app._mount_message.assert_called_once() # ty: ignore assert captured_messages == ["Already using anthropic:claude-opus-4-5"] assert app._model_switching is False async def test_same_model_changed_params_emit_new_message(self) -> None: """A changed same-model notice should still be shown to the user.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "claude-opus-4-5" settings.model_provider = "anthropic" captured_messages: list[str] = [] original_init = AppMessage.__init__ def capture_init(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_init(self, message, **kwargs) with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch.object(AppMessage, "__init__", capture_init), ): await app._switch_model("anthropic:claude-opus-4-5") await app._switch_model( "anthropic:claude-opus-4-5", extra_kwargs={"temperature": 0.2}, ) assert app._mount_message.call_count == 2 # ty: ignore assert captured_messages == [ "Already using anthropic:claude-opus-4-5", ( "Already using anthropic:claude-opus-4-5 " 'with model params {"temperature": 0.2}' ), ] assert app._model_switching is False async def test_same_model_can_skip_unchanged_message(self) -> None: """Onboarding can re-select the active model without adding chat noise.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "claude-opus-4-5" settings.model_provider = "anthropic" with patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ): await app._switch_model( "anthropic:claude-opus-4-5", announce_unchanged=False ) assert app._model_override == "anthropic:claude-opus-4-5" assert app._model_params_override is None app._mount_message.assert_not_called() # ty: ignore assert app._model_switching is False async def test_same_model_with_new_params_applies_overrides(self) -> None: """`/model --model-params {...}` should apply params per-session. Regression test for the bug where the early return on the already-active-model branch silently dropped `--model-params`, leaving `_model_params_override` unset. """ app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "claude-opus-4-5" settings.model_provider = "anthropic" captured_messages: list[str] = [] original_init = AppMessage.__init__ def capture_init(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_init(self, message, **kwargs) with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch.object(AppMessage, "__init__", capture_init), ): await app._switch_model( "anthropic:claude-opus-4-5", extra_kwargs={"num_ctx": 16384, "temperature": 0.2}, ) assert app._model_override == "anthropic:claude-opus-4-5" assert app._model_params_override == { "num_ctx": 16384, "temperature": 0.2, } assert len(captured_messages) == 1 message = captured_messages[0] assert message.startswith("Already using anthropic:claude-opus-4-5") # Stable, key-sorted JSON in the echoed suffix. assert 'with model params {"num_ctx": 16384, "temperature": 0.2}' in message async def test_same_model_without_params_clears_prior_override(self) -> None: """Re-selecting the same model with no params must clear stale params. Regression test for the asymmetry where the regular-switch path always wrote `_model_params_override = extra_kwargs` (clearing on `None`) but the already-active branch left previously-set params in place. """ app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "claude-opus-4-5" settings.model_provider = "anthropic" # Simulate a prior `/model --model-params {...}` call. app._model_override = "anthropic:claude-opus-4-5" app._model_params_override = {"num_ctx": 16384} with patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ): await app._switch_model("anthropic:claude-opus-4-5") assert app._model_override == "anthropic:claude-opus-4-5" assert app._model_params_override is None class TestModelSwitchErrorHandling: """Tests for error handling in _switch_model.""" async def test_missing_credentials_shows_error(self) -> None: """_switch_model shows error when provider credentials are missing.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() # Set a different current model settings.model_name = "gpt-5.5" settings.model_provider = "openai" captured_errors: list[str] = [] original_init = ErrorMessage.__init__ def capture_init(self: ErrorMessage, message: str, **kwargs: Any) -> None: captured_errors.append(message) original_init(self, message, **kwargs) with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=ProviderAuthStatus( state=ProviderAuthState.MISSING, provider="anthropic", env_var="ANTHROPIC_API_KEY", ), ), patch.object(ErrorMessage, "__init__", capture_init), ): await app._switch_model("anthropic:claude-sonnet-4-5") app._mount_message.assert_called_once() # ty: ignore assert len(captured_errors) == 1 assert "Missing credentials" in captured_errors[0] assert "ANTHROPIC_API_KEY" in captured_errors[0] assert app._model_switching is False async def test_save_recent_model_failure_shows_warning(self) -> None: """Permission error saving recent model shows error, no success message.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "gpt-5.5" settings.model_provider = "openai" captured_errors: list[str] = [] original_err_init = ErrorMessage.__init__ def capture_err(self: ErrorMessage, message: str, **kwargs: Any) -> None: captured_errors.append(message) original_err_init(self, message, **kwargs) captured_messages: list[str] = [] original_app_init = AppMessage.__init__ def capture_app(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_app_init(self, message, **kwargs) with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch("deepagents_code.model_config.save_recent_model", return_value=False), patch.object(ErrorMessage, "__init__", capture_err), patch.object(AppMessage, "__init__", capture_app), ): await app._switch_model("anthropic:claude-sonnet-4-5") # Should warn about save failure assert len(captured_errors) == 1 assert "could not save" in captured_errors[0].lower() assert "~/.deepagents/" in captured_errors[0] # Should NOT show success message when save fails assert not any("Switched to" in m for m in captured_messages) assert app._model_override == "anthropic:claude-sonnet-4-5" async def test_remote_agent_sets_model_override(self) -> None: """With remote agent, sets model override for ConfigurableModelMiddleware.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "gpt-5.5" settings.model_provider = "openai" captured_messages: list[str] = [] original_init = AppMessage.__init__ def capture_init(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_init(self, message, **kwargs) with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch( "deepagents_code.model_config.save_recent_model", return_value=True ) as mock_save, patch.object(AppMessage, "__init__", capture_init), ): await app._switch_model("anthropic:claude-sonnet-4-5") assert app._model_override == "anthropic:claude-sonnet-4-5" assert app._model_params_override is None mock_save.assert_called_once() assert settings.model_name == "claude-sonnet-4-5" assert settings.model_provider == "anthropic" assert any("Switched to" in m for m in captured_messages) async def test_remote_agent_refreshes_model_metadata( self, mock_create_model: Mock ) -> None: """Switching models should refresh derived settings like context size.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() app._profile_override = {"max_input_tokens": 180_000} settings.model_name = "gpt-5.5" settings.model_provider = "openai" settings.model_context_limit = 128_000 with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch("deepagents_code.model_config.save_recent_model", return_value=True), ): await app._switch_model( "anthropic:claude-sonnet-4-5", extra_kwargs={"temperature": 0.7}, ) assert settings.model_name == "claude-sonnet-4-5" assert settings.model_provider == "anthropic" assert settings.model_context_limit == 200_000 mock_create_model.assert_called_once_with( "anthropic:claude-sonnet-4-5", extra_kwargs={"temperature": 0.7}, profile_overrides={"max_input_tokens": 180_000}, ) async def test_remote_agent_sets_model_params_override(self) -> None: """With remote agent, extra_kwargs are stored as _model_params_override.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "gpt-5.5" settings.model_provider = "openai" with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch("deepagents_code.model_config.save_recent_model", return_value=True), ): await app._switch_model( "anthropic:claude-sonnet-4-5", extra_kwargs={"temperature": 0.7, "max_tokens": 1024}, ) assert app._model_override == "anthropic:claude-sonnet-4-5" assert app._model_params_override == { "temperature": 0.7, "max_tokens": 1024, } async def test_switched_to_message_echoes_params(self) -> None: """The 'Switched to' confirmation should echo `--model-params`.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "gpt-5.5" settings.model_provider = "openai" captured_messages: list[str] = [] original_init = AppMessage.__init__ def capture_init(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_init(self, message, **kwargs) with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch("deepagents_code.model_config.save_recent_model", return_value=True), patch.object(AppMessage, "__init__", capture_init), ): await app._switch_model( "anthropic:claude-sonnet-4-5", extra_kwargs={"temperature": 0.7, "num_ctx": 16384}, ) assert any( m == "Switched to anthropic:claude-sonnet-4-5 with model params " '{"num_ctx": 16384, "temperature": 0.7}' for m in captured_messages ) class TestModelSwitchConcurrencyGuard: """Tests for _model_switching concurrency guard.""" async def test_concurrent_model_switch_blocked(self) -> None: """Second _switch_model call is rejected while first is in-flight.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._model_switching = True captured_messages: list[str] = [] original_init = AppMessage.__init__ def capture_init(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_init(self, message, **kwargs) with patch.object(AppMessage, "__init__", capture_init): await app._switch_model("anthropic:claude-sonnet-4-5") app._mount_message.assert_called_once() # ty: ignore assert len(captured_messages) == 1 assert "already in progress" in captured_messages[0] async def test_model_switching_flag_reset_on_success(self) -> None: """_model_switching resets to False after a successful switch.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "gpt-5.5" settings.model_provider = "openai" with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch("deepagents_code.model_config.save_recent_model", return_value=True), ): await app._switch_model("anthropic:claude-sonnet-4-5") assert app._model_switching is False class TestModelSwitchSessionReadiness: """Tests for gating model switch on server-backed session readiness.""" async def test_defers_switch_while_connecting(self) -> None: """A direct /model switch fired before `ServerReady` is queued, not failed.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore notify_mock = Mock() app.notify = notify_mock # ty: ignore app._agent = None app._connecting = True await app._switch_model("anthropic:claude-sonnet-4-5") assert len(app._deferred_actions) == 1 action = app._deferred_actions[0] assert action.kind == "model_switch" notify_mock.assert_called_once() app._mount_message.assert_not_called() # ty: ignore # Guard flag cleared so the deferred retry isn't a no-op. assert app._model_switching is False async def test_errors_when_agent_missing_and_not_connecting(self) -> None: """Server-startup failure (no agent, not connecting) still errors.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore notify_mock = Mock() app.notify = notify_mock # ty: ignore app._agent = None app._connecting = False captured_errors: list[str] = [] original_init = ErrorMessage.__init__ def capture_init(self: ErrorMessage, message: str, **kwargs: Any) -> None: captured_errors.append(message) original_init(self, message, **kwargs) with patch.object(ErrorMessage, "__init__", capture_init): await app._switch_model("anthropic:claude-sonnet-4-5") assert app._deferred_actions == [] notify_mock.assert_not_called() assert any("server-backed session" in msg for msg in captured_errors) async def test_deferred_switch_completes_after_server_ready(self) -> None: """End-to-end: defer during connect, drain after ready, switch completes.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore notify_mock = Mock() app.notify = notify_mock # ty: ignore app._agent = None app._connecting = True settings.model_name = "gpt-5.5" settings.model_provider = "openai" with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch("deepagents_code.model_config.save_recent_model", return_value=True), ): await app._switch_model("anthropic:claude-sonnet-4-5") assert len(app._deferred_actions) == 1 # Simulate `ServerReady`: agent arrives, connecting flips off, drain runs. app._agent = _make_remote_agent() app._connecting = False await app._maybe_drain_deferred() assert app._deferred_actions == [] assert app._model_override == "anthropic:claude-sonnet-4-5" assert settings.model_name == "claude-sonnet-4-5" assert settings.model_provider == "anthropic" assert app._model_switching is False class TestModelSwitchFailedStartupRecovery: """Tests for `/model` recovery after a failed initial server startup.""" async def test_retries_startup_with_new_model(self) -> None: """Failed startup + `/model` retries `_start_server_background`. Regression: when the CLI launches without the API key for the configured model, `_start_server_background` raises `ModelConfigError` before the server comes up, leaving the user unable to switch via `/model` (the old code path bailed with "Model switching requires a server-backed session"). `/model` should now rewire deferred-startup state and re-run the worker. """ app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = None app._connecting = False app._server_startup_error = "ModelConfigError: ANTHROPIC_API_KEY not set" app._server_kwargs = { "assistant_id": None, "model_name": "anthropic:claude-opus-4-5", "model_params": None, "interactive": True, } app._model_kwargs = { "model_spec": "anthropic:claude-opus-4-5", "extra_kwargs": None, "profile_overrides": None, } run_worker_mock = Mock() app.run_worker = run_worker_mock # ty: ignore with patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ): await app._switch_model("anthropic:claude-sonnet-4-5") # Failure state cleared and a fresh startup worker scheduled. assert app._server_startup_error is None assert app._connecting is True run_worker_mock.assert_called_once() worker_args, worker_kwargs = run_worker_mock.call_args assert worker_args[0] == app._start_server_background assert worker_kwargs.get("group") == "server-startup" # Deferred-startup kwargs rewired with the new spec. assert app._model_kwargs == { "model_spec": "anthropic:claude-sonnet-4-5", "extra_kwargs": None, "profile_overrides": None, } assert app._server_kwargs["model_name"] == "anthropic:claude-sonnet-4-5" assert app._model_switching is False async def test_retry_with_still_missing_credentials_errors(self) -> None: """Retrying with creds still missing surfaces the credentials error. Avoids looping right back into the same `ModelConfigError` by applying the standard tri-state credentials check before re-launching the startup worker. """ app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = None app._connecting = False app._server_startup_error = "ModelConfigError: ANTHROPIC_API_KEY not set" app._server_kwargs = { "assistant_id": None, "model_name": "anthropic:claude-opus-4-5", "model_params": None, "interactive": True, } run_worker_mock = Mock() app.run_worker = run_worker_mock # ty: ignore captured_errors: list[str] = [] original_init = ErrorMessage.__init__ def capture_init(self: ErrorMessage, message: str, **kwargs: Any) -> None: captured_errors.append(message) original_init(self, message, **kwargs) with ( patch( "deepagents_code.model_config.get_provider_auth_status", return_value=ProviderAuthStatus( state=ProviderAuthState.MISSING, provider="anthropic", env_var="ANTHROPIC_API_KEY", ), ), patch.object(ErrorMessage, "__init__", capture_init), ): await app._switch_model("anthropic:claude-sonnet-4-5") # No worker scheduled, failure state preserved so the user can retry. run_worker_mock.assert_not_called() assert app._server_startup_error == ( "ModelConfigError: ANTHROPIC_API_KEY not set" ) assert app._connecting is False assert any( "Missing credentials" in msg and "ANTHROPIC_API_KEY" in msg for msg in captured_errors ) async def test_remote_server_mode_keeps_original_error(self) -> None: """In remote-server mode (no `_server_kwargs`), recovery is not possible. The CLI doesn't own the subprocess so it can't restart it; fall back to the existing "server-backed session" error rather than silently no-op'ing. """ app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = None app._connecting = False app._server_startup_error = "RuntimeError: connection refused" app._server_kwargs = None run_worker_mock = Mock() app.run_worker = run_worker_mock # ty: ignore captured_errors: list[str] = [] original_init = ErrorMessage.__init__ def capture_init(self: ErrorMessage, message: str, **kwargs: Any) -> None: captured_errors.append(message) original_init(self, message, **kwargs) with patch.object(ErrorMessage, "__init__", capture_init): await app._switch_model("anthropic:claude-sonnet-4-5") run_worker_mock.assert_not_called() assert any("server-backed session" in msg for msg in captured_errors) class TestModelSwitchConfigProvider: """Tests for switching to config-file-defined providers.""" def setup_method(self) -> None: """Clear model config cache before each test.""" clear_caches() async def test_switch_to_config_provider_no_whitelist_error(self, tmp_path) -> None: """Switching to a provider not in PROVIDER_API_KEY_ENV succeeds. Previously this would error with "Unknown provider". Now it switches immediately in the server-backed session. """ config_path = tmp_path / "config.toml" config_path.write_text(""" [models.providers.fireworks] models = ["llama-v3p1-70b"] api_key_env = "FIREWORKS_API_KEY" """) app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "gpt-5.5" settings.model_provider = "openai" captured_messages: list[str] = [] original_app_init = AppMessage.__init__ def capture_app(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_app_init(self, message, **kwargs) with ( patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path), patch.dict("os.environ", {"FIREWORKS_API_KEY": "test-key"}), patch( "deepagents_code.model_config.save_recent_model", return_value=True ) as mock_save, patch.object(AppMessage, "__init__", capture_app), ): await app._switch_model("fireworks:llama-v3p1-70b") mock_save.assert_called_once_with("fireworks:llama-v3p1-70b") assert app._model_override == "fireworks:llama-v3p1-70b" assert settings.model_name == "llama-v3p1-70b" assert settings.model_provider == "fireworks" # Should succeed, not show "Unknown provider" assert any( "Switched to fireworks:llama-v3p1-70b" in m for m in captured_messages ) assert not any("Unknown provider" in m for m in captured_messages) async def test_switch_config_provider_missing_credentials(self, tmp_path) -> None: """Config provider with missing credentials shows appropriate error.""" config_path = tmp_path / "config.toml" config_path.write_text(""" [models.providers.fireworks] models = ["llama-v3p1-70b"] api_key_env = "FIREWORKS_API_KEY" """) app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "gpt-5.5" settings.model_provider = "openai" captured_errors: list[str] = [] original_err_init = ErrorMessage.__init__ def capture_err(self: ErrorMessage, message: str, **kwargs: Any) -> None: captured_errors.append(message) original_err_init(self, message, **kwargs) with ( patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path), patch.dict("os.environ", {}, clear=True), patch.object(ErrorMessage, "__init__", capture_err), ): await app._switch_model("fireworks:llama-v3p1-70b") app._mount_message.assert_called_once() # ty: ignore assert len(captured_errors) == 1 assert "Missing credentials" in captured_errors[0] assert "FIREWORKS_API_KEY" in captured_errors[0] async def test_switch_to_ollama_no_key_required(self, tmp_path) -> None: """Ollama (no api_key_env) passes credential check and switches.""" config_path = tmp_path / "config.toml" config_path.write_text(""" [models.providers.ollama] models = ["llama3"] """) app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "gpt-5.5" settings.model_provider = "openai" captured_messages: list[str] = [] original_app_init = AppMessage.__init__ def capture_app(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_app_init(self, message, **kwargs) with ( patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path), patch( "deepagents_code.model_config.save_recent_model", return_value=True ) as mock_save, patch.object(AppMessage, "__init__", capture_app), ): await app._switch_model("ollama:llama3") mock_save.assert_called_once_with("ollama:llama3") assert app._model_override == "ollama:llama3" assert settings.model_name == "llama3" assert settings.model_provider == "ollama" assert any("Switched to ollama:llama3" in m for m in captured_messages) class TestModelSwitchBareModelName: """Tests for _switch_model with bare model names (no provider prefix).""" async def test_bare_model_name_auto_detects_provider(self) -> None: """Bare model name like 'gpt-5.5' auto-detects provider and switches.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "claude-sonnet-4-5" settings.model_provider = "anthropic" captured_messages: list[str] = [] original_init = AppMessage.__init__ def capture_init(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_init(self, message, **kwargs) with ( patch("deepagents_code.config.detect_provider", return_value="openai"), patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch( "deepagents_code.model_config.save_recent_model", return_value=True ) as mock_save, patch.object(AppMessage, "__init__", capture_init), ): await app._switch_model("gpt-5.5") mock_save.assert_called_once_with("openai:gpt-5.5") assert app._model_override == "openai:gpt-5.5" assert settings.model_name == "gpt-5.5" assert settings.model_provider == "openai" assert any("Switched to openai:gpt-5.5" in m for m in captured_messages) async def test_bare_model_name_missing_credentials(self) -> None: """Bare model name shows credential error when provider creds are missing.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "claude-sonnet-4-5" settings.model_provider = "anthropic" captured_errors: list[str] = [] original_init = ErrorMessage.__init__ def capture_init(self: ErrorMessage, message: str, **kwargs: Any) -> None: captured_errors.append(message) original_init(self, message, **kwargs) with ( patch("deepagents_code.config.detect_provider", return_value="openai"), patch( "deepagents_code.model_config.get_provider_auth_status", return_value=ProviderAuthStatus( state=ProviderAuthState.MISSING, provider="openai", env_var="OPENAI_API_KEY", ), ), patch.object(ErrorMessage, "__init__", capture_init), ): await app._switch_model("gpt-5.5") app._mount_message.assert_called_once() # ty: ignore assert len(captured_errors) == 1 assert "Missing credentials" in captured_errors[0] assert "OPENAI_API_KEY" in captured_errors[0] async def test_bare_model_name_already_using(self) -> None: """Bare model name matching current model shows 'Already using'.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore app._agent = _make_remote_agent() settings.model_name = "gpt-5.5" settings.model_provider = "openai" captured_messages: list[str] = [] original_init = AppMessage.__init__ def capture_init(self: AppMessage, message: str, **kwargs: Any) -> None: captured_messages.append(message) original_init(self, message, **kwargs) with ( patch("deepagents_code.config.detect_provider", return_value="openai"), patch( "deepagents_code.model_config.get_provider_auth_status", return_value=_CONFIGURED_AUTH_STATUS, ), patch.object(AppMessage, "__init__", capture_init), ): await app._switch_model("gpt-5.5") app._mount_message.assert_called_once() # ty: ignore assert len(captured_messages) == 1 assert "Already using" in captured_messages[0] class TestExtractModelParamsFlag: """Tests for _extract_model_params_flag helper.""" def test_no_flag(self) -> None: """Returns original string and None when flag absent.""" remaining, params = _extract_model_params_flag("anthropic:claude-sonnet-4-5") assert remaining == "anthropic:claude-sonnet-4-5" assert params is None def test_single_quoted_json(self) -> None: """Extracts JSON from single-quoted value.""" raw = """--model-params '{"temperature": 0.7}' anthropic:claude-sonnet-4-5""" remaining, params = _extract_model_params_flag(raw) assert remaining == "anthropic:claude-sonnet-4-5" assert params == {"temperature": 0.7} def test_double_quoted_json_with_escaped_quotes(self) -> None: """Extracts JSON from double-quoted value with escaped inner quotes.""" raw = '--model-params "{\\"temperature\\": 0.7}" anthropic:claude-sonnet-4-5' remaining, params = _extract_model_params_flag(raw) assert remaining == "anthropic:claude-sonnet-4-5" assert params == {"temperature": 0.7} def test_bare_braces(self) -> None: """Extracts JSON from unquoted braces with balanced matching.""" raw = '--model-params {"temperature": 0.7, "max_tokens": 100}' remaining, params = _extract_model_params_flag(raw) assert remaining == "" assert params == {"temperature": 0.7, "max_tokens": 100} def test_bare_braces_with_model_after(self) -> None: """Model arg after bare-brace JSON is preserved.""" raw = '--model-params {"temperature":0.7} anthropic:claude-sonnet-4-5' remaining, params = _extract_model_params_flag(raw) assert remaining == "anthropic:claude-sonnet-4-5" assert params == {"temperature": 0.7} def test_model_before_flag(self) -> None: """Model arg before --model-params is preserved.""" raw = "anthropic:claude-sonnet-4-5 --model-params '{\"temperature\": 0.7}'" remaining, params = _extract_model_params_flag(raw) assert remaining == "anthropic:claude-sonnet-4-5" assert params == {"temperature": 0.7} def test_missing_value_raises(self) -> None: """Raises ValueError when --model-params has no value.""" with pytest.raises(ValueError, match="requires a JSON object"): _extract_model_params_flag("--model-params") def test_invalid_json_raises(self) -> None: """Raises ValueError with hint for malformed JSON.""" with pytest.raises(ValueError, match=r"Invalid JSON.*Expected format"): _extract_model_params_flag("--model-params '{not json}'") def test_non_dict_json_raises(self) -> None: """Raises TypeError when JSON is not an object.""" with pytest.raises(TypeError, match="must be a JSON object"): _extract_model_params_flag("--model-params '[1, 2, 3]'") def test_unclosed_quote_raises(self) -> None: """Raises ValueError for unclosed quote.""" with pytest.raises(ValueError, match="Unclosed"): _extract_model_params_flag("""--model-params '{"temperature": 0.7}""") def test_unbalanced_braces_raises(self) -> None: """Raises ValueError for unbalanced braces.""" with pytest.raises(ValueError, match="Unbalanced"): _extract_model_params_flag('--model-params {"temperature": 0.7') def test_with_default_flag(self) -> None: """Works alongside --default flag.""" raw = ( """--model-params '{"temperature": 0.7}' """ "--default anthropic:claude-sonnet-4-5" ) remaining, params = _extract_model_params_flag(raw) assert remaining == "--default anthropic:claude-sonnet-4-5" assert params == {"temperature": 0.7} def test_empty_object(self) -> None: """Empty JSON object is valid.""" remaining, params = _extract_model_params_flag("--model-params '{}'") assert remaining == "" assert params == {} class TestModelCommandIntegration: """Tests for /model command handler integration.""" async def test_invalid_model_params_shows_error(self) -> None: """/model with invalid --model-params JSON shows error.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore captured_errors: list[str] = [] original_init = ErrorMessage.__init__ def capture_init(self: ErrorMessage, message: str, **kwargs: Any) -> None: captured_errors.append(message) original_init(self, message, **kwargs) with patch.object(ErrorMessage, "__init__", capture_init): await app._handle_command("/model --model-params '{bad}'") assert len(captured_errors) == 1 assert "Invalid JSON" in captured_errors[0] assert "Expected format" in captured_errors[0] async def test_model_params_with_default_rejected(self) -> None: """/model --model-params with --default shows error.""" app = DeepAgentsApp() app._mount_message = AsyncMock() # ty: ignore captured_errors: list[str] = [] original_init = ErrorMessage.__init__ def capture_init(self: ErrorMessage, message: str, **kwargs: Any) -> None: captured_errors.append(message) original_init(self, message, **kwargs) cmd = ( """/model --model-params '{"temperature": 0.7}' """ "--default anthropic:claude-sonnet-4-5" ) with patch.object(ErrorMessage, "__init__", capture_init): await app._handle_command(cmd) assert len(captured_errors) == 1 assert "cannot be used with --default" in captured_errors[0]