fix(gateway): keep topic bindings aligned after compression

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

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]

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

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,20 +88,44 @@ 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(
         "OPENAI_BASE_URL=https://new.example/v1\nHERMES_INFERENCE_PROVIDER=custom\n",
         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"

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.

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