from hermes_state import AsyncSessionDB """Tests for gateway /status behavior and token persistence.""" from datetime import datetime import time from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest from gateway.config import GatewayConfig, Platform, PlatformConfig from gateway.platforms.base import MessageEvent from gateway.session import SessionEntry, SessionSource, build_session_key def _make_source(platform: Platform = Platform.TELEGRAM) -> SessionSource: return SessionSource( platform=platform, user_id="u1", chat_id="c1", user_name="tester", chat_type="dm", ) def _make_event(text: str, *, platform: Platform = Platform.TELEGRAM) -> MessageEvent: return MessageEvent( text=text, source=_make_source(platform), message_id="m1", ) def _make_runner(session_entry: SessionEntry, *, platform: Platform = Platform.TELEGRAM): from gateway.run import GatewayRunner runner = object.__new__(GatewayRunner) runner.config = GatewayConfig( platforms={platform: PlatformConfig(enabled=True, token="***")} ) adapter = MagicMock() adapter.send = AsyncMock() runner.adapters = {platform: adapter} runner._voice_mode = {} runner.hooks = SimpleNamespace(emit=AsyncMock(), loaded_hooks=False) runner.session_store = MagicMock() runner.session_store.get_or_create_session.return_value = session_entry runner.session_store.load_transcript.return_value = [] runner.session_store.has_any_sessions.return_value = True runner.session_store.append_to_transcript = MagicMock() runner.session_store.rewrite_transcript = MagicMock() runner.session_store.update_session = MagicMock() runner._running_agents = {} runner._session_run_generation = {} runner._pending_messages = {} runner._pending_approvals = {} runner._session_db = AsyncSessionDB(MagicMock()) runner._session_db._db.get_session_title.return_value = None # Default: no DB row โ†’ /status reports 0 tokens. Tests that exercise # the populated path override this. runner._session_db._db.get_session.return_value = None runner._reasoning_config = None runner._provider_routing = {} runner._fallback_model = None runner._agent_cache = {} runner._agent_cache_lock = MagicMock() runner._show_reasoning = False runner._is_user_authorized = lambda _source: True runner._set_session_env = lambda _context: None runner._should_send_voice_reply = lambda *_args, **_kwargs: False runner._send_voice_reply = AsyncMock() runner._capture_gateway_honcho_if_configured = lambda *args, **kwargs: None runner._emit_gateway_run_progress = AsyncMock() return runner @pytest.mark.asyncio async def test_status_command_reports_running_agent_without_interrupt(monkeypatch): session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", total_tokens=321, ) runner = _make_runner(session_entry) # Token total comes from the SQLite SessionDB, not SessionEntry. runner._session_db._db.get_session.return_value = { "input_tokens": 200, "output_tokens": 121, "cache_read_tokens": 0, "cache_write_tokens": 0, "reasoning_tokens": 0, } running_agent = MagicMock() runner._running_agents[build_session_key(_make_source())] = running_agent result = await runner._handle_message(_make_event("/status")) assert "**Session ID:** `sess-1`" in result assert "**Lifetime tokens billed:** 321" in result assert "**Agent Running:** Yes โšก" in result assert "**Title:**" not in result running_agent.interrupt.assert_not_called() assert runner._pending_messages == {} @pytest.mark.asyncio async def test_status_command_includes_session_title_when_present(): session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", total_tokens=321, ) runner = _make_runner(session_entry) runner._session_db._db.get_session_title.return_value = "My titled session" result = await runner._handle_message(_make_event("/status")) assert "**Session ID:** `sess-1`" in result assert "**Title:** My titled session" in result @pytest.mark.asyncio async def test_status_command_reads_token_totals_from_session_db(): """Regression test for #17158: /status must source token totals from the SQLite SessionDB (where run_agent.py persists them) and sum all component counts, not from SessionEntry (which the agent never writes).""" session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", total_tokens=0, # SessionEntry never gets written to โ€” always 0. ) runner = _make_runner(session_entry) runner._session_db._db.get_session.return_value = { "input_tokens": 1000, "output_tokens": 250, "cache_read_tokens": 500, "cache_write_tokens": 100, "reasoning_tokens": 50, } result = await runner._handle_message(_make_event("/status")) # 1000 + 250 + 500 + 100 + 50 = 1,900 assert "**Lifetime tokens billed:** 1,900" in result @pytest.mark.asyncio async def test_status_command_tokens_zero_when_session_db_row_missing(): """When the SessionDB has no row for the current session yet (fresh session, no agent calls), /status reports 0 without raising.""" session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", total_tokens=999, # This should be ignored. ) runner = _make_runner(session_entry) runner._session_db._db.get_session.return_value = None result = await runner._handle_message(_make_event("/status")) assert "**Lifetime tokens billed:** 0" in result @pytest.mark.asyncio async def test_status_command_includes_live_agent_model_and_context(): session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", total_tokens=0, ) runner = _make_runner(session_entry) runner._session_db._db.get_session.return_value = { "input_tokens": 1000, "output_tokens": 250, "cache_read_tokens": 0, "cache_write_tokens": 0, "reasoning_tokens": 0, "model": "openai/gpt-test", } running_agent = SimpleNamespace( model="openai/gpt-test", provider="openai", context_compressor=SimpleNamespace( last_prompt_tokens=12_345, context_length=100_000, ), interrupt=MagicMock(), ) runner._running_agents[build_session_key(_make_source())] = running_agent result = await runner._handle_message(_make_event("/status")) assert "**Model:** `openai/gpt-test` (openai)" in result assert "**Context:** 12,345 / 100,000 (12%)" in result assert "**Lifetime tokens billed:** 1,250" in result @pytest.mark.asyncio async def test_status_command_includes_persisted_model_and_context_when_agent_not_running(monkeypatch): session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", total_tokens=0, last_prompt_tokens=24_000, ) runner = _make_runner(session_entry) runner._session_db._db.get_session.return_value = { "input_tokens": 2000, "output_tokens": 500, "cache_read_tokens": 0, "cache_write_tokens": 0, "reasoning_tokens": 0, "model": "openai/gpt-persisted", "billing_provider": "openai-codex", "billing_base_url": "https://example.invalid/v1", } monkeypatch.setattr("gateway.run._load_gateway_config", lambda: {"model": {"context_length": 272_000}}) result = await runner._handle_message(_make_event("/status")) assert "**Model:** `openai/gpt-persisted` (openai-codex)" in result assert "**Context:** 24,000 / 272,000 (9%)" in result assert "**Lifetime tokens billed:** 2,500" in result @pytest.mark.asyncio async def test_status_command_includes_cached_agent_model_and_context(): session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", total_tokens=0, ) runner = _make_runner(session_entry) cached_agent = SimpleNamespace( model="anthropic/claude-sonnet-test", provider="openrouter", context_compressor=SimpleNamespace( last_prompt_tokens=10_000, context_length=200_000, ), ) runner._agent_cache = {session_entry.session_key: (cached_agent, time.time())} result = await runner._handle_message(_make_event("/status")) assert "**Model:** `anthropic/claude-sonnet-test` (openrouter)" in result assert "**Context:** 10,000 / 200,000 (5%)" in result @pytest.mark.asyncio async def test_agents_command_reports_active_agents_and_processes(monkeypatch): session_key = build_session_key(_make_source()) session_entry = SessionEntry( session_key=session_key, session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", total_tokens=0, ) runner = _make_runner(session_entry) running_agent = SimpleNamespace( session_id="sess-running", model="openrouter/test-model", interrupt=MagicMock(), get_activity_summary=lambda: {"seconds_since_activity": 0}, ) runner._running_agents[session_key] = running_agent runner._running_agents_ts = {session_key: time.time() - 8} runner._background_tasks = set() class _FakeRegistry: def list_sessions(self): return [ { "session_id": "proc-1", "status": "running", "uptime_seconds": 17, "command": "sleep 30", } ] monkeypatch.setattr("tools.process_registry.process_registry", _FakeRegistry()) result = await runner._handle_message(_make_event("/agents")) assert "**Active agents:** 1" in result assert "**Running background processes:** 1" in result assert "proc-1" in result running_agent.interrupt.assert_not_called() @pytest.mark.asyncio async def test_tasks_alias_routes_to_agents_command(monkeypatch): session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", total_tokens=0, ) runner = _make_runner(session_entry) runner._background_tasks = set() class _FakeRegistry: def list_sessions(self): return [] monkeypatch.setattr("tools.process_registry.process_registry", _FakeRegistry()) result = await runner._handle_message(_make_event("/tasks")) assert "Active Agents & Tasks" in result @pytest.mark.asyncio async def test_handle_message_persists_agent_token_counts(monkeypatch): import gateway.run as gateway_run session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) runner.session_store.load_transcript.return_value = [{"role": "user", "content": "earlier"}] runner._run_agent = AsyncMock( return_value={ "final_response": "ok", "messages": [], "tools": [], "history_offset": 0, "last_prompt_tokens": 80, "input_tokens": 120, "output_tokens": 45, "model": "openai/test-model", } ) monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}) monkeypatch.setattr( "agent.model_metadata.get_model_context_length", lambda *_args, **_kwargs: 100000, ) result = await runner._handle_message(_make_event("hello")) assert result == "ok" runner.session_store.update_session.assert_called_once_with( session_entry.session_key, last_prompt_tokens=80, ) @pytest.mark.asyncio async def test_first_run_slack_home_channel_onboarding_uses_parent_command(monkeypatch): import gateway.run as gateway_run session_entry = SessionEntry( session_key=build_session_key(_make_source(Platform.SLACK)), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.SLACK, chat_type="dm", ) runner = _make_runner(session_entry, platform=Platform.SLACK) runner.session_store.load_transcript.return_value = [] runner.session_store.has_any_sessions.return_value = False runner._run_agent = AsyncMock( return_value={ "final_response": "ok", "messages": [], "tools": [], "history_offset": 0, "last_prompt_tokens": 0, "input_tokens": 0, "output_tokens": 0, "model": "openai/test-model", } ) monkeypatch.delenv("SLACK_HOME_CHANNEL", raising=False) monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}) monkeypatch.setattr( "agent.model_metadata.get_model_context_length", lambda *_args, **_kwargs: 100000, ) result = await runner._handle_message(_make_event("hello", platform=Platform.SLACK)) assert result == "ok" runner.adapters[Platform.SLACK].send.assert_awaited_once() onboarding = runner.adapters[Platform.SLACK].send.await_args.args[1] assert "/hermes sethome" in onboarding assert "Type /sethome" not in onboarding @pytest.mark.asyncio async def test_first_run_non_slack_home_channel_onboarding_keeps_direct_command(monkeypatch): import gateway.run as gateway_run session_entry = SessionEntry( session_key=build_session_key(_make_source(Platform.TELEGRAM)), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry, platform=Platform.TELEGRAM) runner.session_store.load_transcript.return_value = [] runner.session_store.has_any_sessions.return_value = False runner._run_agent = AsyncMock( return_value={ "final_response": "ok", "messages": [], "tools": [], "history_offset": 0, "last_prompt_tokens": 0, "input_tokens": 0, "output_tokens": 0, "model": "openai/test-model", } ) monkeypatch.delenv("TELEGRAM_HOME_CHANNEL", raising=False) monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}) monkeypatch.setattr( "agent.model_metadata.get_model_context_length", lambda *_args, **_kwargs: 100000, ) result = await runner._handle_message(_make_event("hello", platform=Platform.TELEGRAM)) assert result == "ok" runner.adapters[Platform.TELEGRAM].send.assert_awaited_once() onboarding = runner.adapters[Platform.TELEGRAM].send.await_args.args[1] assert "Type /sethome" in onboarding @pytest.mark.asyncio async def test_handle_message_discards_stale_result_after_session_invalidation(monkeypatch): import gateway.run as gateway_run session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) runner.session_store.load_transcript.return_value = [{"role": "user", "content": "earlier"}] session_key = session_entry.session_key runner.adapters[Platform.TELEGRAM]._post_delivery_callbacks = {session_key: object()} async def _stale_result(**kwargs): runner._invalidate_session_run_generation(kwargs["session_key"], reason="test_stale_result") return { "final_response": "late reply", "messages": [], "tools": [], "history_offset": 0, "last_prompt_tokens": 80, "input_tokens": 120, "output_tokens": 45, "model": "openai/test-model", } runner._run_agent = AsyncMock(side_effect=_stale_result) monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}) monkeypatch.setattr( "agent.model_metadata.get_model_context_length", lambda *_args, **_kwargs: 100000, ) result = await runner._handle_message(_make_event("hello")) assert result is None runner.session_store.append_to_transcript.assert_not_called() runner.session_store.update_session.assert_not_called() assert session_key not in runner.adapters[Platform.TELEGRAM]._post_delivery_callbacks @pytest.mark.asyncio async def test_handle_message_stale_result_keeps_newer_generation_callback(monkeypatch): import gateway.run as gateway_run class _Adapter: def __init__(self): self._post_delivery_callbacks = {} async def send(self, *args, **kwargs): return None def pop_post_delivery_callback(self, session_key, *, generation=None): entry = self._post_delivery_callbacks.get(session_key) if entry is None: return None if isinstance(entry, tuple): entry_generation, callback = entry if generation is not None and entry_generation != generation: return None self._post_delivery_callbacks.pop(session_key, None) return callback if generation is not None: return None return self._post_delivery_callbacks.pop(session_key, None) session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) runner.session_store.load_transcript.return_value = [{"role": "user", "content": "earlier"}] session_key = session_entry.session_key adapter = _Adapter() runner.adapters[Platform.TELEGRAM] = adapter async def _stale_result(**kwargs): # Simulate a newer run claiming the callback slot before the stale run unwinds. runner._session_run_generation[session_key] = 2 adapter._post_delivery_callbacks[session_key] = (2, lambda: None) return { "final_response": "late reply", "messages": [], "tools": [], "history_offset": 0, "last_prompt_tokens": 80, "input_tokens": 120, "output_tokens": 45, "model": "openai/test-model", } runner._run_agent = AsyncMock(side_effect=_stale_result) monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}) monkeypatch.setattr( "agent.model_metadata.get_model_context_length", lambda *_args, **_kwargs: 100000, ) result = await runner._handle_message(_make_event("hello")) assert result is None assert session_key in adapter._post_delivery_callbacks assert adapter._post_delivery_callbacks[session_key][0] == 2 @pytest.mark.asyncio async def test_status_command_bypasses_active_session_guard(): """When an agent is running, /status must be dispatched immediately via base.handle_message โ€” not queued or treated as an interrupt (#5046).""" import asyncio from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType from gateway.session import build_session_key from gateway.config import Platform, PlatformConfig source = _make_source() session_key = build_session_key(source) handler_called_with = [] async def fake_handler(event): handler_called_with.append(event) return "๐Ÿ“Š **Hermes Gateway Status**\n**Agent Running:** Yes โšก" # Concrete subclass to avoid abstract method errors class _ConcreteAdapter(BasePlatformAdapter): platform = Platform.TELEGRAM async def connect(self, *, is_reconnect: bool = False): pass async def disconnect(self): pass async def send(self, chat_id, content, **kwargs): pass async def get_chat_info(self, chat_id): return {} platform_config = PlatformConfig(enabled=True, token="***") adapter = _ConcreteAdapter(platform_config, Platform.TELEGRAM) adapter.set_message_handler(fake_handler) sent = [] async def fake_send_with_retry(chat_id, content, reply_to=None, metadata=None): sent.append(content) adapter._send_with_retry = fake_send_with_retry # Simulate an active session interrupt_event = asyncio.Event() adapter._active_sessions[session_key] = interrupt_event event = MessageEvent( text="/status", source=source, message_id="m1", message_type=MessageType.COMMAND, ) await adapter.handle_message(event) assert handler_called_with, "/status handler was never called (event was queued or dropped)" assert sent, "/status response was never sent" assert "Agent Running" in sent[0] assert not interrupt_event.is_set(), "/status incorrectly triggered an agent interrupt" assert session_key not in adapter._pending_messages, "/status was incorrectly queued" @pytest.mark.asyncio async def test_profile_command_reports_custom_root_profile(monkeypatch, tmp_path): """Gateway /profile detects custom-root profiles (not under ~/.hermes).""" from pathlib import Path session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) profile_home = tmp_path / "profiles" / "coder" monkeypatch.setenv("HERMES_HOME", str(profile_home)) monkeypatch.setattr(Path, "home", lambda: tmp_path / "unrelated-home") result = await runner._handle_profile_command(_make_event("/profile")) assert "**Profile:** `coder`" in result assert f"**Home:** `{profile_home}`" in result @pytest.mark.asyncio async def test_profile_command_reports_source_stamped_profile(monkeypatch, tmp_path): """On a multiplexed gateway, /profile reports the profile SERVING the source (source.profile โ€” URL prefix / per-credential adapter / room map), not the multiplexer's active profile, which is always the default and made /profile answer "default" in every persona chat.""" hermes_home = tmp_path / ".hermes" profile_home = hermes_home / "profiles" / "milo" profile_home.mkdir(parents=True) session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) runner.config.multiplex_profiles = True monkeypatch.setenv("HERMES_HOME", str(hermes_home)) event = _make_event("/profile") event.source.profile = "milo" result = await runner._handle_profile_command(event) assert "**Profile:** `milo`" in result assert f"**Home:** `{profile_home}`" in result @pytest.mark.asyncio async def test_profile_command_ignores_stamp_when_multiplexing_off(monkeypatch, tmp_path): """Without ``gateway.multiplex_profiles`` a stamped source is ignored: /profile keeps reporting the active profile and the default home, mirroring the multiplex gating in ``_run_agent`` and ``_reset_notice_session_info``.""" hermes_home = tmp_path / ".hermes" profile_home = hermes_home / "profiles" / "milo" profile_home.mkdir(parents=True) session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) assert runner.config.multiplex_profiles is False monkeypatch.setenv("HERMES_HOME", str(hermes_home)) event = _make_event("/profile") event.source.profile = "milo" result = await runner._handle_profile_command(event) assert "**Profile:** `default`" in result assert f"**Home:** `{hermes_home}`" in result @pytest.mark.asyncio async def test_profile_command_unstamped_source_unchanged(monkeypatch, tmp_path): """Single-profile behavior is untouched: an unstamped source reports the active profile and the default home.""" hermes_home = tmp_path / ".hermes" hermes_home.mkdir() session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) monkeypatch.setenv("HERMES_HOME", str(hermes_home)) result = await runner._handle_profile_command(_make_event("/profile")) assert "**Profile:** `default`" in result assert f"**Home:** `{hermes_home}`" in result @pytest.mark.asyncio async def test_post_delivery_callback_generation_snapshot_happens_after_bind(): """Regression: the callback_generation snapshot in _process_message_background must happen AFTER the handler runs, not before. _hermes_run_generation is set on the interrupt event by GatewayRunner._bind_adapter_run_generation during _handle_message_with_agent. The earlier snapshot-at-task-start always captured None, which bypassed the generation-ownership check in pop_post_delivery_callback and let stale runs fire a fresher run's callbacks. """ import asyncio from gateway.platforms.base import BasePlatformAdapter source = _make_source() session_key = build_session_key(source) fired = [] class _ConcreteAdapter(BasePlatformAdapter): platform = Platform.TELEGRAM async def connect(self, *, is_reconnect: bool = False): pass async def disconnect(self): pass async def send(self, chat_id, content, **kwargs): pass async def get_chat_info(self, chat_id): return {} adapter = _ConcreteAdapter( PlatformConfig(enabled=True, token="***"), Platform.TELEGRAM ) async def fake_handler(event): # Simulate what _bind_adapter_run_generation does mid-run. interrupt_event = adapter._active_sessions.get(session_key) setattr(interrupt_event, "_hermes_run_generation", 1) # Stale run registers its callback at generation=1. adapter.register_post_delivery_callback( session_key, lambda: fired.append("older"), generation=1, ) # A fresher run overwrites with generation=2 (different dict entry). adapter.register_post_delivery_callback( session_key, lambda: fired.append("newer"), generation=2, ) return None adapter.set_message_handler(fake_handler) event = MessageEvent(text="hello", source=source, message_id="m1") await adapter.handle_message(event) tasks = list(adapter._background_tasks) assert tasks, "expected background task to be created" await asyncio.gather(*tasks) # The stale run (generation=1) must NOT fire the fresher run's callback # (generation=2). With the pre-fix code, callback_generation was snapshotted # as None before the handler ran, bypassing the ownership check and firing # "newer" anyway. assert fired == [] assert session_key in adapter._post_delivery_callbacks assert adapter._post_delivery_callbacks[session_key][0] == 2 # โ”€โ”€ /context command tests โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ def _stub_agent(**overrides) -> SimpleNamespace: """Build a stub agent with the attributes _handle_context_command reads.""" props = dict( model="openai/gpt-test", context_compressor=SimpleNamespace( last_prompt_tokens=47_231, context_length=200_000, threshold_tokens=100_000, threshold_percent=0.5, compression_count=2, _last_compression_savings_pct=63.0, ), session_api_calls=47, session_input_tokens=410_000, session_output_tokens=38_000, session_reasoning_tokens=12_000, session_total_tokens=3_158_641, session_cache_read_tokens=2_900_000, session_cache_write_tokens=48_000, ) props.update(overrides) return SimpleNamespace(**props) @pytest.mark.asyncio async def test_context_command_live_agent(): """/context with a live running agent shows the full view: gauge, compression, and throughput โ€” but NOT cache stats.""" session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) session_key = session_entry.session_key agent = _stub_agent() runner._running_agents[session_key] = agent result = await runner._handle_context_command(_make_event("/context")) assert "๐Ÿง  **Context Window**" in result assert "Model: `openai/gpt-test`" in result assert "Window: 200,000 tokens" in result assert "In use: 47,231 / 200,000 (24%)" in result assert "Headroom to limit: 152,769 tokens" in result # Compression section assert "Auto-compresses at: 100,000 (50%)" in result assert "Compressions this session: 2" in result assert "Last compression freed: 63% of context" in result # Throughput โ€” NOT cache assert "Session totals (cumulative across 47 API calls)" in result assert "Input 410,000" in result assert "Output 38,000" in result assert "Reasoning 12,000" in result assert "Total billed: 3,158,641" in result assert "each call re-sends the window above" in result # Cache stats must NOT appear (removed per design) assert "Cache read" not in result assert "Cache write" not in result assert "Cache hit" not in result assert "Hit rate" not in result @pytest.mark.asyncio async def test_context_command_over_threshold(): """When used >= threshold, the over-threshold warning is shown.""" session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-2", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) session_key = session_entry.session_key agent = _stub_agent( context_compressor=SimpleNamespace( last_prompt_tokens=150_000, context_length=200_000, threshold_tokens=100_000, threshold_percent=0.5, compression_count=5, _last_compression_savings_pct=40.0, ) ) runner._running_agents[session_key] = agent result = await runner._handle_context_command(_make_event("/context")) assert "โš ๏ธ" in result assert "Over auto-compression threshold" in result @pytest.mark.asyncio async def test_context_command_no_agent_transcript_fallback(): """When no agent is resident and session_entry has no last_prompt_tokens, /context falls back to a transcript estimate.""" session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-3", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", last_prompt_tokens=0, # No live context data ) runner = _make_runner(session_entry) # Stub the transcript so estimate_messages_tokens_rough has something to work with runner.session_store.load_transcript.return_value = [ {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there!"}, {"role": "user", "content": "What's my balance?"}, {"role": "assistant", "content": "Your balance is $1,000."}, ] result = await runner._handle_context_command(_make_event("/context")) assert "๐Ÿง  **Context Window**" in result assert "Estimated context:" in result assert "4 messages" in result @pytest.mark.asyncio async def test_context_command_no_data(): """When there's no agent, no session data, and no transcript, /context returns the no-data message.""" session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-4", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", last_prompt_tokens=0, ) runner = _make_runner(session_entry) runner.session_store.load_transcript.return_value = [] result = await runner._handle_context_command(_make_event("/context")) assert "No context data available yet" in result @pytest.mark.asyncio async def test_context_command_includes_category_breakdown(): """/context with a live agent appends the per-category estimated breakdown (plain text, no glyph grid) and the /context all hint.""" session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-5", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) agent = _stub_agent() runner._running_agents[session_entry.session_key] = agent fake_payload = { "categories": [ {"id": "system_prompt", "label": "System prompt", "tokens": 9_000}, {"id": "tool_definitions", "label": "Tool definitions", "tokens": 21_000}, ], "context_max": 200_000, "context_percent": 24, "context_used": 47_231, "estimated_total": 30_000, "model": "openai/gpt-test", } from unittest.mock import patch as _patch with _patch( "agent.context_breakdown.compute_session_context_breakdown", return_value=fake_payload, ): result = await runner._handle_context_command(_make_event("/context")) assert "Estimated usage by category" in result assert "System prompt" in result assert "9,000 tokens" in result assert "Tool definitions" in result assert "Use /context all" in result # No glyph grid on the gateway (plain-text variant) assert "ยท ยท ยท" not in result @pytest.mark.asyncio async def test_context_all_appends_expanded_listings(): """/context all appends per-toolset and per-skill cost listings.""" session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-6", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) agent = _stub_agent() runner._running_agents[session_entry.session_key] = agent fake_payload = { "categories": [ {"id": "skills", "label": "Skills", "tokens": 2_000}, ], "context_max": 200_000, "context_percent": 24, "context_used": 47_231, "estimated_total": 2_000, "model": "openai/gpt-test", } fake_details = { "skills": [ {"name": "hermes-agent", "index_tokens": 30, "skill_md_tokens": 2_500}, ], "toolsets": [ {"toolset": "terminal", "tool_count": 4, "schema_tokens": 5_100}, ], } from unittest.mock import patch as _patch with _patch( "agent.context_breakdown.compute_session_context_breakdown", return_value=fake_payload, ), _patch( "agent.context_breakdown.compute_context_details", return_value=fake_details, ): result = await runner._handle_context_command(_make_event("/context all")) assert "Toolsets by schema cost" in result assert "terminal" in result and "5,100 tokens" in result assert "Skills by cost" in result assert "hermes-agent" in result # Expanded view drops the hint assert "Use /context all" not in result @pytest.mark.asyncio async def test_context_breakdown_failure_never_breaks_command(): """A breakdown engine crash degrades gracefully โ€” the gauge still renders.""" session_entry = SessionEntry( session_key=build_session_key(_make_source()), session_id="sess-7", created_at=datetime.now(), updated_at=datetime.now(), platform=Platform.TELEGRAM, chat_type="dm", ) runner = _make_runner(session_entry) agent = _stub_agent() runner._running_agents[session_entry.session_key] = agent from unittest.mock import patch as _patch with _patch( "agent.context_breakdown.compute_session_context_breakdown", side_effect=RuntimeError("boom"), ): result = await runner._handle_context_command(_make_event("/context")) assert "๐Ÿง  **Context Window**" in result assert "In use: 47,231 / 200,000 (24%)" in result assert "Estimated usage by category" not in result