diff --git a/gateway/run.py b/gateway/run.py index 7b5ace07067ef..ea828d17610a3 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2287,6 +2287,38 @@ def _record_telegram_topic_binding( session_id=session_entry.session_id, ) + def _sync_telegram_topic_binding( + self, + source: SessionSource, + session_entry, + *, + reason: str, + ) -> None: + """Keep topic-mode Telegram bindings aligned with session rotations. + + Compression rotates the underlying Hermes session_id while the + Telegram topic thread_id stays the same. If the topic binding is left + pointing at the pre-compression session, the next message in that topic + gets rebound to the oversized parent transcript and can compact again. + """ + if not self._is_telegram_topic_lane(source): + return + try: + self._record_telegram_topic_binding(source, session_entry) + logger.info( + "telegram topic binding synced after %s: chat=%s thread=%s session=%s", + reason, + source.chat_id, + source.thread_id, + session_entry.session_id, + ) + except Exception: + logger.debug( + "Failed to sync Telegram topic binding after %s", + reason, + exc_info=True, + ) + def _recover_telegram_topic_thread_id( self, source: SessionSource, @@ -8530,6 +8562,11 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g if _hyg_new_sid != session_entry.session_id: session_entry.session_id = _hyg_new_sid self.session_store._save() + self._sync_telegram_topic_binding( + source, + session_entry, + reason="hygiene-compression", + ) self.session_store.rewrite_transcript( session_entry.session_id, _compressed @@ -8791,10 +8828,17 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g response = _sanitize_gateway_final_response(source.platform, response) # If the agent's session_id changed during compression, update - # session_entry so transcript writes below go to the right session. + # session_entry so transcript writes below go to the right session, + # and keep Telegram topic-mode's thread_id -> session_id binding + # from snapping the next turn back to the oversized parent session. if agent_result.get("session_id") and agent_result["session_id"] != session_entry.session_id: session_entry.session_id = agent_result["session_id"] self.session_store._save() + self._sync_telegram_topic_binding( + source, + session_entry, + reason="agent-compression", + ) # Prepend reasoning/thinking if display is enabled (per-platform) try: @@ -17152,24 +17196,19 @@ def _approval_notify_sync(approval_data: dict) -> None: # the compressed transcript, not the stale pre-compression one. agent = agent_holder[0] _session_was_split = False - if agent and session_key and hasattr(agent, 'session_id') and agent.session_id != session_id: + agent_session_id = getattr(agent, "session_id", None) if agent else None + if agent and session_key and agent_session_id and agent_session_id != session_id: _session_was_split = True logger.info( "Session split detected: %s → %s (compression)", - session_id, agent.session_id, + session_id, agent_session_id, ) - entry = self.session_store._entries.get(session_key) - if entry: - entry.session_id = agent.session_id - self.session_store._save() - # If this is a Telegram DM and source.thread_id was lost during - # the session split (synthetic / recovered event), restore it - # from the binding so _thread_metadata_for_source produces the - # correct message_thread_id instead of routing to the General - # thread. Failure here is non-fatal — we log and continue; - # worst case the message lands in General, which is the - # pre-fix behaviour. + # If Telegram delivered this topic-mode DM without the lane + # thread_id, recover it from the old session binding before + # rotating the binding to the compressed child. Looking up by + # the child session cannot work yet: the binding is precisely + # what we are about to update. if ( getattr(source, "platform", None) == Platform.TELEGRAM and getattr(source, "chat_type", None) == "dm" @@ -17178,22 +17217,32 @@ def _approval_notify_sync(approval_data: dict) -> None: ): try: _binding = self._session_db.get_telegram_topic_binding_by_session( - session_id=agent.session_id, + session_id=session_id, ) if _binding and _binding.get("thread_id"): source.thread_id = str(_binding["thread_id"]) logger.debug( - "Restored source.thread_id=%s from binding after session split %s → %s", + "Restored source.thread_id=%s from old binding before session split sync %s → %s", source.thread_id, session_id, - agent.session_id, + agent_session_id, ) except Exception: logger.debug( - "Failed to restore thread_id from binding after session split", + "Failed to restore thread_id from old binding before session split sync", exc_info=True, ) + entry = self.session_store._entries.get(session_key) + if entry: + entry.session_id = agent_session_id + self.session_store._save() + self._sync_telegram_topic_binding( + source, + entry, + reason="agent-compression", + ) + effective_session_id = getattr(agent, 'session_id', session_id) if agent else session_id # When compression created a new session, the messages list was diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 8673195748061..c8b9ed5053e28 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -2241,8 +2241,6 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None) sane_path = ":".join(path_entries) return f"""[Unit] Description={SERVICE_DESCRIPTION} -After=network-online.target -Wants=network-online.target StartLimitIntervalSec=0 [Service] diff --git a/tests/gateway/test_telegram_topic_mode.py b/tests/gateway/test_telegram_topic_mode.py index 1941bb89e208a..c16e8ed4f9b8f 100644 --- a/tests/gateway/test_telegram_topic_mode.py +++ b/tests/gateway/test_telegram_topic_mode.py @@ -266,6 +266,62 @@ async def fake_run_agent(*args, **kwargs): assert captured["session_id"] == "restored-session" +@pytest.mark.asyncio +async def test_topic_binding_follows_session_id_rotation_after_compression( + tmp_path, monkeypatch +): + import gateway.run as gateway_run + + session_db = SessionDB(db_path=tmp_path / "state.db") + session_db.enable_telegram_topic_mode(chat_id="208214988", user_id="208214988") + session_key = build_session_key(_make_source(thread_id="17585")) + session_db.create_session( + session_id="oversized-parent-session", + source="telegram", + user_id="208214988", + ) + session_db.create_session( + session_id="compressed-child-session", + source="telegram", + user_id="208214988", + parent_session_id="oversized-parent-session", + ) + session_db.bind_telegram_topic( + chat_id="208214988", + thread_id="17585", + user_id="208214988", + session_key=session_key, + session_id="oversized-parent-session", + ) + runner = _make_runner(session_db=session_db) + + async def fake_run_agent(*args, **kwargs): + assert kwargs.get("session_id") == "oversized-parent-session" + return { + "success": True, + "final_response": "compressed response", + "session_id": "compressed-child-session", + "messages": [], + "last_prompt_tokens": 1234, + } + + runner._run_agent = AsyncMock(side_effect=fake_run_agent) + + monkeypatch.setattr( + gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"} + ) + + result = await runner._handle_message(_make_event("continue", thread_id="17585")) + + assert result == "compressed response" + binding = session_db.get_telegram_topic_binding( + chat_id="208214988", + thread_id="17585", + ) + assert binding is not None + assert binding["session_id"] == "compressed-child-session" + + @pytest.mark.asyncio async def test_telegram_group_prompt_is_not_topic_lobby_even_when_dm_topic_mode_enabled( tmp_path, monkeypatch diff --git a/tests/hermes_cli/test_env_loader.py b/tests/hermes_cli/test_env_loader.py index 2523754a84ba4..e8aba9658595f 100644 --- a/tests/hermes_cli/test_env_loader.py +++ b/tests/hermes_cli/test_env_loader.py @@ -1,5 +1,6 @@ -import importlib +import json import os +import subprocess import sys from pathlib import Path @@ -87,7 +88,7 @@ def test_null_bytes_in_user_env_are_stripped(tmp_path, monkeypatch): assert os.getenv("OPENAI_API_KEY") == "sk-123" -def test_main_import_applies_user_env_over_shell_values(tmp_path, monkeypatch): +def test_main_import_applies_user_env_over_shell_values(tmp_path): home = tmp_path / "hermes" home.mkdir() (home / ".env").write_text( @@ -95,12 +96,36 @@ def test_main_import_applies_user_env_over_shell_values(tmp_path, monkeypatch): encoding="utf-8", ) - monkeypatch.setenv("HERMES_HOME", str(home)) - monkeypatch.setenv("OPENAI_BASE_URL", "https://old.example/v1") - monkeypatch.setenv("HERMES_INFERENCE_PROVIDER", "openrouter") + env = os.environ.copy() + env.update( + { + "HERMES_HOME": str(home), + "OPENAI_BASE_URL": "https://old.example/v1", + "HERMES_INFERENCE_PROVIDER": "openrouter", + } + ) - sys.modules.pop("hermes_cli.main", None) - importlib.import_module("hermes_cli.main") + # Import hermes_cli.main in a subprocess: the import intentionally mutates + # process-global environment/module state, and doing that in the pytest + # worker leaks into unrelated CLI/setup tests that run later in the same + # worker. + code = """ +import json +import os +import hermes_cli.main # noqa: F401 +print(json.dumps({ + "OPENAI_BASE_URL": os.getenv("OPENAI_BASE_URL"), + "HERMES_INFERENCE_PROVIDER": os.getenv("HERMES_INFERENCE_PROVIDER"), +})) +""" + result = subprocess.run( + [sys.executable, "-c", code], + check=True, + capture_output=True, + text=True, + env=env, + ) + loaded = json.loads(result.stdout.strip().splitlines()[-1]) - assert os.getenv("OPENAI_BASE_URL") == "https://new.example/v1" - assert os.getenv("HERMES_INFERENCE_PROVIDER") == "custom" + assert loaded["OPENAI_BASE_URL"] == "https://new.example/v1" + assert loaded["HERMES_INFERENCE_PROVIDER"] == "custom" diff --git a/tests/hermes_cli/test_gateway_service.py b/tests/hermes_cli/test_gateway_service.py index b1fcadbf4f0d7..6996d8377bab9 100644 --- a/tests/hermes_cli/test_gateway_service.py +++ b/tests/hermes_cli/test_gateway_service.py @@ -324,6 +324,8 @@ def test_user_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self unit = gateway_cli.generate_systemd_unit(system=False) assert "ExecStart=" in unit + assert "After=network-online.target" not in unit + assert "Wants=network-online.target" not in unit assert "ExecStop=" not in unit assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit @@ -388,6 +390,8 @@ def test_system_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(se assert "ExecStop=" not in unit assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit + assert "After=network-online.target" in unit + assert "Wants=network-online.target" in unit # TimeoutStopSec must exceed the default drain_timeout (60s) so # systemd doesn't SIGKILL the cgroup before post-interrupt cleanup # (tool subprocess kill, adapter disconnect) runs — issue #8202. diff --git a/tests/hermes_cli/test_update_hangup_protection.py b/tests/hermes_cli/test_update_hangup_protection.py index e5c81a45a0106..bcb6f0f50bc17 100644 --- a/tests/hermes_cli/test_update_hangup_protection.py +++ b/tests/hermes_cli/test_update_hangup_protection.py @@ -213,8 +213,16 @@ def test_wraps_stdout_and_stderr_with_mirror(self, tmp_path, monkeypatch): try: # On Windows (no SIGHUP) we still wrap stdio and create the log. assert state["installed"] is True - assert isinstance(sys.stdout, _UpdateOutputStream) - assert isinstance(sys.stderr, _UpdateOutputStream) + # Other tests may re-import hermes_cli.main in the same xdist + # worker, so class identity can differ even though the runtime + # wrapper is the same implementation. Assert the observable + # wrapper contract instead of brittle module-object identity. + assert sys.stdout.__class__.__name__ == "_UpdateOutputStream" + assert sys.stderr.__class__.__name__ == "_UpdateOutputStream" + assert getattr(sys.stdout, "_original", None) is prev_out + assert getattr(sys.stderr, "_original", None) is prev_err + assert getattr(sys.stdout, "_log", None) is state["log_file"] + assert getattr(sys.stderr, "_log", None) is state["log_file"] assert state["log_file"] is not None sys.stdout.write("checking mirror\n") diff --git a/tests/plugins/web/test_web_search_provider_plugins.py b/tests/plugins/web/test_web_search_provider_plugins.py index 47d7791977b97..ba5f597c08ca7 100644 --- a/tests/plugins/web/test_web_search_provider_plugins.py +++ b/tests/plugins/web/test_web_search_provider_plugins.py @@ -2,8 +2,8 @@ Covers: -- All eight bundled plugins (brave-free, ddgs, searxng, exa, parallel, - tavily, firecrawl, xai) instantiate and self-report the expected + +- All bundled plugins instantiate and self-report the expected capabilities + ABC-derived defaults. - Each plugin's ``is_available()`` correctly reflects env-var presence. - The web_search_registry resolves an active provider in the documented @@ -27,6 +27,33 @@ import pytest +BUNDLED_WEB_PLUGINS = [ + "brave-free", + "ddgs", + "exa", + "firecrawl", + "parallel", + "searxng", + "tavily", + "xai", +] + +BUNDLED_WEB_PLUGIN_CAPABILITIES = [ + ("brave-free", True, False, False), + ("ddgs", True, False, False), + ("searxng", True, False, False), + ("exa", True, True, False), + ("parallel", True, True, False), + ("tavily", True, True, True), + # firecrawl: search + extract + crawl. Crawl was originally + # disabled in the migration (fell through to a legacy inline + # path); the follow-up commit enabled it natively. + ("firecrawl", True, True, True), + # xAI delegates search to Grok's server-side web_search tool. + ("xai", True, False, False), +] + + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -71,40 +98,18 @@ def _isolate_env(monkeypatch: pytest.MonkeyPatch) -> None: class TestBundledPluginsRegister: - """All eight bundled web plugins discover and register correctly.""" + """All bundled web plugins discover and register correctly.""" def test_all_seven_plugins_present_in_registry(self) -> None: _ensure_plugins_loaded() from agent.web_search_registry import list_providers names = sorted(p.name for p in list_providers()) - assert names == [ - "brave-free", - "ddgs", - "exa", - "firecrawl", - "parallel", - "searxng", - "tavily", - "xai", - ] + assert names == BUNDLED_WEB_PLUGINS @pytest.mark.parametrize( "plugin_name,expected_search,expected_extract,expected_crawl", - [ - ("brave-free", True, False, False), - ("ddgs", True, False, False), - ("searxng", True, False, False), - ("exa", True, True, False), - ("parallel", True, True, False), - ("tavily", True, True, True), - # firecrawl: search + extract + crawl. Crawl was originally - # disabled in the migration (fell through to a legacy inline - # path); the follow-up commit enabled it natively. - ("firecrawl", True, True, True), - # xai: search-only via Grok's agentic web_search tool. - ("xai", True, False, False), - ], + BUNDLED_WEB_PLUGIN_CAPABILITIES, ) def test_capability_flags_match_spec( self, @@ -124,7 +129,7 @@ def test_capability_flags_match_spec( @pytest.mark.parametrize( "plugin_name", - ["brave-free", "ddgs", "searxng", "exa", "parallel", "tavily", "firecrawl", "xai"], + BUNDLED_WEB_PLUGINS, ) def test_each_plugin_has_name_and_display_name(self, plugin_name: str) -> None: _ensure_plugins_loaded() @@ -137,7 +142,7 @@ def test_each_plugin_has_name_and_display_name(self, plugin_name: str) -> None: @pytest.mark.parametrize( "plugin_name", - ["brave-free", "ddgs", "searxng", "exa", "parallel", "tavily", "firecrawl", "xai"], + BUNDLED_WEB_PLUGINS, ) def test_each_plugin_has_setup_schema(self, plugin_name: str) -> None: """``get_setup_schema()`` returns a dict the picker can consume."""