"""Tests for gateway /reasoning command and hot reload behavior.""" import asyncio import inspect import sys import types from unittest.mock import AsyncMock, MagicMock import pytest import yaml import gateway.run as gateway_run from gateway.config import Platform from gateway.platforms.base import MessageEvent from gateway.session import SessionSource def _make_event(text="/reasoning", platform=Platform.TELEGRAM, user_id="12345", chat_id="67890"): """Build a MessageEvent for testing.""" source = SessionSource( platform=platform, user_id=user_id, chat_id=chat_id, user_name="testuser", ) return MessageEvent(text=text, source=source) def _make_runner(): """Create a bare GatewayRunner without calling __init__.""" runner = object.__new__(gateway_run.GatewayRunner) runner.adapters = {} runner._ephemeral_system_prompt = "" runner._prefill_messages = [] runner._reasoning_config = None runner._show_reasoning = False runner._provider_routing = {} runner._fallback_model = None runner._running_agents = {} runner.hooks = MagicMock() runner.hooks.emit = AsyncMock() runner.hooks.loaded_hooks = [] runner._session_db = None runner._get_or_create_gateway_honcho = lambda session_key: (None, None) return runner class _CapturingAgent: """Fake agent that records init kwargs for assertions.""" last_init = None def __init__(self, *args, **kwargs): type(self).last_init = dict(kwargs) self.tools = [] def run_conversation(self, user_message: str, conversation_history=None, task_id=None): return { "final_response": "ok", "messages": [], "api_calls": 1, } class TestReasoningCommand: @pytest.mark.asyncio async def test_reasoning_in_help_output(self): runner = _make_runner() event = _make_event(text="/help") result = await runner._handle_help_command(event) assert "/reasoning [level|show|hide]" in result def test_reasoning_is_known_command(self): source = inspect.getsource(gateway_run.GatewayRunner._handle_message) assert '"reasoning"' in source @pytest.mark.asyncio async def test_reasoning_command_reloads_current_state_from_config(self, tmp_path, monkeypatch): hermes_home = tmp_path / "hermes" hermes_home.mkdir() config_path = hermes_home / "config.yaml" config_path.write_text( "agent:\n reasoning_effort: none\ndisplay:\n show_reasoning: true\n", encoding="utf-8", ) monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home) monkeypatch.delenv("HERMES_REASONING_EFFORT", raising=False) runner = _make_runner() runner._reasoning_config = {"enabled": True, "effort": "xhigh"} runner._show_reasoning = False result = await runner._handle_reasoning_command(_make_event("/reasoning")) assert "**Effort:** `none (disabled)`" in result assert "**Display:** on ✓" in result assert runner._reasoning_config == {"enabled": False} assert runner._show_reasoning is True @pytest.mark.asyncio async def test_handle_reasoning_command_updates_config_and_cache(self, tmp_path, monkeypatch): hermes_home = tmp_path / "hermes" hermes_home.mkdir() config_path = hermes_home / "config.yaml" config_path.write_text("agent:\n reasoning_effort: medium\n", encoding="utf-8") monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home) monkeypatch.delenv("HERMES_REASONING_EFFORT", raising=False) runner = _make_runner() runner._reasoning_config = {"enabled": True, "effort": "medium"} result = await runner._handle_reasoning_command(_make_event("/reasoning low")) saved = yaml.safe_load(config_path.read_text(encoding="utf-8")) assert saved["agent"]["reasoning_effort"] == "low" assert runner._reasoning_config == {"enabled": True, "effort": "low"} assert "takes effect on next message" in result def test_run_agent_reloads_reasoning_config_per_message(self, tmp_path, monkeypatch): hermes_home = tmp_path / "hermes" hermes_home.mkdir() (hermes_home / "config.yaml").write_text("agent:\n reasoning_effort: low\n", encoding="utf-8") monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home) monkeypatch.setattr(gateway_run, "_env_path", hermes_home / ".env") monkeypatch.setattr(gateway_run, "load_dotenv", lambda *args, **kwargs: None) monkeypatch.setattr( gateway_run, "_resolve_runtime_agent_kwargs", lambda: { "provider": "openrouter", "api_mode": "chat_completions", "base_url": "https://openrouter.ai/api/v1", "api_key": "test-key", }, ) monkeypatch.delenv("HERMES_REASONING_EFFORT", raising=False) fake_run_agent = types.ModuleType("run_agent") fake_run_agent.AIAgent = _CapturingAgent monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) _CapturingAgent.last_init = None runner = _make_runner() runner._reasoning_config = {"enabled": True, "effort": "xhigh"} source = SessionSource( platform=Platform.LOCAL, chat_id="cli", chat_name="CLI", chat_type="dm", user_id="user-1", ) result = asyncio.run( runner._run_agent( message="ping", context_prompt="", history=[], source=source, session_id="session-1", session_key="agent:main:local:dm", ) ) assert result["final_response"] == "ok" assert _CapturingAgent.last_init is not None assert _CapturingAgent.last_init["reasoning_config"] == {"enabled": True, "effort": "low"} def test_run_agent_prefers_config_over_stale_reasoning_env(self, tmp_path, monkeypatch): hermes_home = tmp_path / "hermes" hermes_home.mkdir() (hermes_home / "config.yaml").write_text("agent:\n reasoning_effort: none\n", encoding="utf-8") monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home) monkeypatch.setattr(gateway_run, "_env_path", hermes_home / ".env") monkeypatch.setattr(gateway_run, "load_dotenv", lambda *args, **kwargs: None) monkeypatch.setattr( gateway_run, "_resolve_runtime_agent_kwargs", lambda: { "provider": "openrouter", "api_mode": "chat_completions", "base_url": "https://openrouter.ai/api/v1", "api_key": "test-key", }, ) monkeypatch.setenv("HERMES_REASONING_EFFORT", "low") fake_run_agent = types.ModuleType("run_agent") fake_run_agent.AIAgent = _CapturingAgent monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) _CapturingAgent.last_init = None runner = _make_runner() source = SessionSource( platform=Platform.LOCAL, chat_id="cli", chat_name="CLI", chat_type="dm", user_id="user-1", ) result = asyncio.run( runner._run_agent( message="ping", context_prompt="", history=[], source=source, session_id="session-1", session_key="agent:main:local:dm", ) ) assert result["final_response"] == "ok" assert _CapturingAgent.last_init is not None assert _CapturingAgent.last_init["reasoning_config"] == {"enabled": False} def test_run_agent_includes_enabled_mcp_servers_in_gateway_toolsets(self, tmp_path, monkeypatch): hermes_home = tmp_path / "hermes" hermes_home.mkdir() (hermes_home / "config.yaml").write_text( "platform_toolsets:\n" " cli: [web, memory]\n" "mcp_servers:\n" " exa:\n" " url: https://mcp.exa.ai/mcp\n" " web-search-prime:\n" " url: https://api.z.ai/api/mcp/web_search_prime/mcp\n", encoding="utf-8", ) monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home) monkeypatch.setattr(gateway_run, "_env_path", hermes_home / ".env") monkeypatch.setattr(gateway_run, "load_dotenv", lambda *args, **kwargs: None) monkeypatch.setattr( gateway_run, "_resolve_runtime_agent_kwargs", lambda: { "provider": "openrouter", "api_mode": "chat_completions", "base_url": "https://openrouter.ai/api/v1", "api_key": "test-key", }, ) fake_run_agent = types.ModuleType("run_agent") fake_run_agent.AIAgent = _CapturingAgent monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) _CapturingAgent.last_init = None runner = _make_runner() source = SessionSource( platform=Platform.LOCAL, chat_id="cli", chat_name="CLI", chat_type="dm", user_id="user-1", ) result = asyncio.run( runner._run_agent( message="ping", context_prompt="", history=[], source=source, session_id="session-1", session_key="agent:main:local:dm", ) ) assert result["final_response"] == "ok" assert _CapturingAgent.last_init is not None enabled_toolsets = set(_CapturingAgent.last_init["enabled_toolsets"]) assert "web" in enabled_toolsets assert "memory" in enabled_toolsets assert "exa" in enabled_toolsets assert "web-search-prime" in enabled_toolsets def test_run_agent_homeassistant_uses_default_platform_toolset(self, tmp_path, monkeypatch): hermes_home = tmp_path / "hermes" hermes_home.mkdir() (hermes_home / "config.yaml").write_text("", encoding="utf-8") monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home) monkeypatch.setattr(gateway_run, "_env_path", hermes_home / ".env") monkeypatch.setattr(gateway_run, "load_dotenv", lambda *args, **kwargs: None) monkeypatch.setattr( gateway_run, "_resolve_runtime_agent_kwargs", lambda: { "provider": "openrouter", "api_mode": "chat_completions", "base_url": "https://openrouter.ai/api/v1", "api_key": "test-key", }, ) fake_run_agent = types.ModuleType("run_agent") fake_run_agent.AIAgent = _CapturingAgent monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) _CapturingAgent.last_init = None runner = _make_runner() source = SessionSource( platform=Platform.HOMEASSISTANT, chat_id="ha", chat_name="Home Assistant", chat_type="dm", user_id="user-1", ) result = asyncio.run( runner._run_agent( message="ping", context_prompt="", history=[], source=source, session_id="session-1", session_key="agent:main:homeassistant:dm", ) ) assert result["final_response"] == "ok" assert _CapturingAgent.last_init is not None assert "homeassistant" in set(_CapturingAgent.last_init["enabled_toolsets"])