| """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} |
|
|
