mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-23 16:36:23 +00:00
47fb20c0bd
* fix(fast): default /fast to session scope on CLI and gateway Completes the session-first policy from #67946 for the /fast toggle (the remaining half of #54084). A bare /fast fast|normal now applies to the current session only; --global persists agent.service_tier to config.yaml. Gateway: new _session_service_tier_overrides dict (registered in _CONVERSATION_SCOPED_STATE so /new clears it) resolved at both agent turn sites via _resolve_session_service_tier(); the /fast handler and its choice picker apply session overrides and evict the cached agent. The TUI config.set fast path was already session-scoped. CLI: /fast parses --global (parity with /reasoning); bare toggles mutate self.service_tier only. * fix(sessions): /new resets model, reasoning, and fast to config defaults /new and /reset are full conversation boundaries: session-scoped runtime overrides do not carry into the next session (#48055, #23131). CLI new_session(): clears the one-turn model restore, re-derives service_tier from config, and — when the session's model differs from the config default — switches back via the shared switch_model() pipeline (live agent swap included; best-effort so an unreachable default never blocks /new). TUI _reset_session_agent(): stops forwarding model_override / create_reasoning_override / create_service_tier_override into the rebuilt agent and pops the pins so later rebuilds can't resurrect them. The gateway already cleared its per-session overrides via _clear_conversation_scope on /new. Cross-session contamination stays impossible: nothing here touches process-global env or other sessions' pins.
241 lines
9.5 KiB
Python
241 lines
9.5 KiB
Python
"""Tests for the gateway interactive choice picker (/reasoning, /fast).
|
|
|
|
The picker mirrors the /model picker architecture: the gateway gates on the
|
|
adapter *type* exposing ``send_choice_picker``, sends a flat choice list, and
|
|
falls back to the text status card when the platform has no picker or the
|
|
send fails. Selection flows through the same application path as the typed
|
|
command, so picker and typed arguments can never diverge.
|
|
"""
|
|
|
|
import asyncio
|
|
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, SendResult
|
|
from gateway.session import SessionSource
|
|
|
|
|
|
def _make_event(text="/reasoning", platform=Platform.TELEGRAM, user_id="12345", chat_id="67890"):
|
|
source = SessionSource(
|
|
platform=platform,
|
|
user_id=user_id,
|
|
chat_id=chat_id,
|
|
user_name="testuser",
|
|
)
|
|
return MessageEvent(text=text, source=source)
|
|
|
|
|
|
class _PickerAdapter:
|
|
"""Adapter whose *type* exposes ``send_choice_picker`` (the gate the
|
|
handler checks via ``getattr(type(adapter), 'send_choice_picker', None)``)."""
|
|
|
|
def __init__(self, success=True):
|
|
self.calls = []
|
|
self._success = success
|
|
|
|
async def send_choice_picker(self, **kwargs):
|
|
self.calls.append(kwargs)
|
|
return SendResult(success=self._success, message_id="m1")
|
|
|
|
|
|
class _NoPickerAdapter:
|
|
"""Adapter with no choice-picker capability."""
|
|
|
|
|
|
def _make_runner(adapter=None):
|
|
runner = object.__new__(gateway_run.GatewayRunner)
|
|
runner.adapters = {}
|
|
runner._ephemeral_system_prompt = ""
|
|
runner._prefill_messages = []
|
|
runner._reasoning_config = None
|
|
runner._session_reasoning_overrides = {}
|
|
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)
|
|
runner._adapter_for_source = lambda source: adapter
|
|
runner._thread_metadata_for_source = lambda source, anchor=None: {}
|
|
runner._reply_anchor_for_event = lambda event: None
|
|
return runner
|
|
|
|
|
|
class TestReasoningChoicePicker:
|
|
@pytest.mark.asyncio
|
|
async def test_bare_reasoning_sends_picker_when_adapter_supports_it(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
|
|
adapter = _PickerAdapter()
|
|
runner = _make_runner(adapter)
|
|
|
|
result = await runner._handle_reasoning_command(_make_event("/reasoning"))
|
|
|
|
assert result is None # picker sent — adapter owns the response
|
|
assert len(adapter.calls) == 1
|
|
call = adapter.calls[0]
|
|
values = [c["value"] for c in call["choices"]]
|
|
# Full canonical ladder + none + subcommands, in order
|
|
from hermes_constants import VALID_REASONING_EFFORTS
|
|
assert values[0] == "none"
|
|
assert values[1:1 + len(VALID_REASONING_EFFORTS)] == list(VALID_REASONING_EFFORTS)
|
|
assert values[-3:] == ["reset", "show", "hide"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bare_reasoning_falls_back_to_text_without_picker(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
|
|
runner = _make_runner(_NoPickerAdapter())
|
|
|
|
result = await runner._handle_reasoning_command(_make_event("/reasoning"))
|
|
|
|
assert isinstance(result, str)
|
|
assert "/reasoning" in result # text status card
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bare_reasoning_falls_back_to_text_when_picker_send_fails(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
|
|
adapter = _PickerAdapter(success=False)
|
|
runner = _make_runner(adapter)
|
|
|
|
result = await runner._handle_reasoning_command(_make_event("/reasoning"))
|
|
|
|
assert isinstance(result, str)
|
|
assert len(adapter.calls) == 1 # attempted, then fell back
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_typed_argument_never_sends_picker(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
|
|
adapter = _PickerAdapter()
|
|
runner = _make_runner(adapter)
|
|
|
|
result = await runner._handle_reasoning_command(_make_event("/reasoning high"))
|
|
|
|
assert isinstance(result, str)
|
|
assert adapter.calls == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_picker_selection_applies_same_as_typed(self, tmp_path, monkeypatch):
|
|
"""The picker's on_choice_selected must produce the identical state
|
|
change as typing the argument (single application path)."""
|
|
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
|
|
adapter = _PickerAdapter()
|
|
runner = _make_runner(adapter)
|
|
event = _make_event("/reasoning")
|
|
session_key = runner._session_key_for_source(event.source)
|
|
|
|
await runner._handle_reasoning_command(event)
|
|
on_choice = adapter.calls[0]["on_choice_selected"]
|
|
|
|
reply = await on_choice(event.source.chat_id, "ultra")
|
|
|
|
assert "ultra" in reply
|
|
override = runner._session_reasoning_overrides.get(session_key)
|
|
assert override == {"enabled": True, "effort": "ultra"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_picker_selection_of_current_level_marks_is_current(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
|
|
(tmp_path / "config.yaml").write_text(
|
|
yaml.safe_dump({"agent": {"reasoning_effort": "xhigh"}}), encoding="utf-8"
|
|
)
|
|
adapter = _PickerAdapter()
|
|
runner = _make_runner(adapter)
|
|
|
|
await runner._handle_reasoning_command(_make_event("/reasoning"))
|
|
|
|
current = [c["value"] for c in adapter.calls[0]["choices"] if c.get("is_current")]
|
|
assert current == ["xhigh"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_picker_show_choice_toggles_display(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
|
|
adapter = _PickerAdapter()
|
|
runner = _make_runner(adapter)
|
|
event = _make_event("/reasoning")
|
|
|
|
await runner._handle_reasoning_command(event)
|
|
on_choice = adapter.calls[0]["on_choice_selected"]
|
|
await on_choice(event.source.chat_id, "show")
|
|
|
|
assert runner._show_reasoning is True
|
|
saved = yaml.safe_load((tmp_path / "config.yaml").read_text(encoding="utf-8"))
|
|
assert saved["display"]["platforms"]["telegram"]["show_reasoning"] is True
|
|
|
|
|
|
class TestFastChoicePicker:
|
|
def _patch_fast_support(self, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
|
|
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
|
|
monkeypatch.setattr(gateway_run, "_resolve_gateway_model", lambda cfg: "gpt-5.6")
|
|
import hermes_cli.models as models_mod
|
|
monkeypatch.setattr(models_mod, "model_supports_fast_mode", lambda m: True)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bare_fast_sends_picker_when_adapter_supports_it(self, tmp_path, monkeypatch):
|
|
self._patch_fast_support(monkeypatch, tmp_path)
|
|
adapter = _PickerAdapter()
|
|
runner = _make_runner(adapter)
|
|
|
|
result = await runner._handle_fast_command(_make_event("/fast"))
|
|
|
|
assert result is None
|
|
values = [c["value"] for c in adapter.calls[0]["choices"]]
|
|
assert values == ["fast", "normal"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fast_picker_selection_is_session_scoped(self, tmp_path, monkeypatch):
|
|
"""A bare /fast picker tap applies a session override, not a config write."""
|
|
self._patch_fast_support(monkeypatch, tmp_path)
|
|
adapter = _PickerAdapter()
|
|
runner = _make_runner(adapter)
|
|
event = _make_event("/fast")
|
|
|
|
await runner._handle_fast_command(event)
|
|
on_choice = adapter.calls[0]["on_choice_selected"]
|
|
await on_choice(event.source.chat_id, "fast")
|
|
|
|
assert runner._service_tier == "priority"
|
|
assert runner._session_service_tier_overrides
|
|
assert not (tmp_path / "config.yaml").exists()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fast_picker_global_flag_persists_service_tier(self, tmp_path, monkeypatch):
|
|
"""A /fast --global picker tap persists agent.service_tier to config."""
|
|
self._patch_fast_support(monkeypatch, tmp_path)
|
|
adapter = _PickerAdapter()
|
|
runner = _make_runner(adapter)
|
|
event = _make_event("/fast --global")
|
|
|
|
await runner._handle_fast_command(event)
|
|
on_choice = adapter.calls[0]["on_choice_selected"]
|
|
await on_choice(event.source.chat_id, "fast")
|
|
|
|
assert runner._service_tier == "priority"
|
|
saved = yaml.safe_load((tmp_path / "config.yaml").read_text(encoding="utf-8"))
|
|
assert saved["agent"]["service_tier"] == "fast"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bare_fast_falls_back_to_text_without_picker(self, tmp_path, monkeypatch):
|
|
self._patch_fast_support(monkeypatch, tmp_path)
|
|
runner = _make_runner(_NoPickerAdapter())
|
|
|
|
result = await runner._handle_fast_command(_make_event("/fast"))
|
|
|
|
assert isinstance(result, str)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_typed_fast_argument_never_sends_picker(self, tmp_path, monkeypatch):
|
|
self._patch_fast_support(monkeypatch, tmp_path)
|
|
adapter = _PickerAdapter()
|
|
runner = _make_runner(adapter)
|
|
|
|
result = await runner._handle_fast_command(_make_event("/fast normal"))
|
|
|
|
assert isinstance(result, str)
|
|
assert adapter.calls == []
|