fix(bot-mode): route managed messaging-gateway sessions

agent/conversation_loop.py

@@ -588,36 +588,29 @@ def _print_billing_or_entitlement_guidance(
 
 
 def _bot_chat_prompt_stale(agent, stored_prompt: str) -> bool:
-    """Bot Chat capability epoch check for a stored prompt.
-
-    The stored prompt embeds a capability fingerprint; a mismatch is a deliberate
-    once-per-change rebuild. Unstamped prompts never match; probe failures fail closed
-    to "reuse" so the cache is kept. Legacy upgrade: a Bot Chat prompt predating the
-    epoch mechanism gets ONE title-gated migration rebuild; the stamped result cannot
-    re-fire."""
+    """Return whether routed Bot Mode prompt state must be rebuilt once."""
     try:
         from tools.bot_mode_probe import (
-            BOT_CHAT_TITLE,
+            bot_mode_session_state,
             stored_bot_chat_prompt_needs_upgrade,
             stored_prompt_capability_stale,
+            stored_prompt_has_bot_mode_protocol,
         )
         home = None
         try:
             from agent.system_prompt import _agent_home
             home = _agent_home(agent)
         except Exception:
             pass
-        if stored_prompt_capability_stale(stored_prompt, home):
+        routed = bool(bot_mode_session_state(agent)["session_kind"])
+        if not routed and stored_prompt_has_bot_mode_protocol(stored_prompt):
             return True
-        if not getattr(agent, "_bot_mode_protocol", True):
+        if not routed:
             return False
-        title = str(getattr(agent, "_session_title_hint", "") or "").strip()
-        if not title and agent._session_db and agent.session_id:
-            try:
-                title = str(agent._session_db.get_session_title(agent.session_id) or "").strip()
-            except Exception:
-                title = ""
-        return title == BOT_CHAT_TITLE and bool(stored_bot_chat_prompt_needs_upgrade(stored_prompt, home))
+        return bool(
+            stored_prompt_capability_stale(stored_prompt, home)
+            or stored_bot_chat_prompt_needs_upgrade(stored_prompt, home)
+        )
     except Exception:
         return False
 
@@ -667,7 +660,6 @@ def _restore_or_build_system_prompt(agent, system_message, conversation_history)
                 "adopt the new capability surface (one-time prefix-cache break).",
                 agent.session_id,
             )
-            agent._session_title_hint = "Bot Chat"
             # The skills index cache (LRU + disk snapshot) does not watch the skills
             # dir; a capability refresh must rebuild THROUGH it or new skills are lost.
             try:

agent/system_prompt.py

@@ -312,24 +312,18 @@ def _skills_prompt(agent: Any) -> str:
 
 
 def _bot_mode_parts(agent: Any) -> List[str]:
-    """Bot Mode teammate protocol — only in a bot's canonical "Bot Chat" session.
-    Marks the prompt timeless (the volatile date line is dropped) since a birth
-    date pinned in a months-long session is misinformation."""
+    """Bot Mode protocol for canonical Bot Chats and routed human messaging sessions."""
     parts: List[str] = []
     try:
-        from tools.bot_mode_probe import BOT_CHAT_TITLE, epoch_line, get_bot_mode_protocol_section
-        _title = str(getattr(agent, "_session_title_hint", "") or "").strip()
-        if not _title:
-            _sdb = getattr(agent, "_session_db", None)
-            _sid = getattr(agent, "session_id", None)
-            _title = str((_sdb.get_session_title(_sid) if (_sdb and _sid) else None) or "").strip()
-        _bot_section = get_bot_mode_protocol_section(_agent_home(agent)) if _title == BOT_CHAT_TITLE else None
-        if _bot_section:
-            parts.append(_bot_section)
-            # Capability epoch lets the restore path rebuild ONCE per
-            # user-initiated capability change in an eternal session.
-            parts.append(epoch_line(_agent_home(agent)))
-            agent._bot_chat_timeless_prompt = True
+        from tools.bot_mode_probe import bot_mode_session_state, epoch_line, get_bot_mode_protocol_section
+        if bot_mode_session_state(agent)["session_kind"]:
+            _bot_section = get_bot_mode_protocol_section(_agent_home(agent))
+            if _bot_section:
+                parts.append(_bot_section)
+                # Capability epoch lets the restore path rebuild ONCE per
+                # user-initiated capability change in an eternal session.
+                parts.append(epoch_line(_agent_home(agent)))
+                agent._bot_chat_timeless_prompt = True
     except Exception:
         pass
     return parts

agent/turn_context.py

@@ -833,8 +833,9 @@ def build_turn_context(
         restore_or_build_system_prompt(agent, system_message, conversation_history)
     active_system_prompt = agent._cached_system_prompt
 
-    # Bot Mode DM tool — injected ONLY into a bot's canonical "Bot Chat" session (same
-    # gate as the protocol section); gate is session-stable, so cache-safe.
+    # Bot Mode DM tool — injected only into routed Bot Mode sessions (canonical
+    # Bot Chat or a classified human messaging chat bound to a real profile).
+    # The shared gate is frozen per session/source, so the tool list is cache-safe.
     try:
         from tools.bot_mode_dm import ensure_message_agent_tool
 

gateway/session_context.py

@@ -155,6 +155,20 @@ def get_session_env(name: str, default: str = "") -> str:
     return os.getenv(name, default)
 
 
+def resolve_session_source(platform: str | None) -> str:
+    """Return the authoritative source for an agent-backed session.
+
+    A task-local ``HERMES_SESSION_SOURCE`` takes precedence over the agent's
+    platform. This is shared by persistence and authorization callers so a
+    machine/task turn cannot inherit trust from a human-facing platform hint.
+    """
+    try:
+        source = get_session_env("HERMES_SESSION_SOURCE", "")
+    except Exception:
+        source = os.environ.get("HERMES_SESSION_SOURCE", "")
+    return str(source or "").strip() or platform or "cli"
+
+
 # Surfaces that are not a human chat channel (gateway binds HERMES_SESSION_PLATFORM, CLI/TUI/
 # desktop bind HERMES_SESSION_SOURCE, so both are consulted).  Default-deny: an unrecognized
 # identity counts as messaging.  Mirrors LOCAL_SESSION_SOURCE_IDS in apps/desktop session-source.ts.

run_agent.py

@@ -44,13 +44,9 @@ def _launch_cwd_for_session(source: str) -> Optional[str]:
 
 
 def _session_source_for_agent(platform: Optional[str]) -> str:
-    try:
-        from gateway.session_context import get_session_env
+    from gateway.session_context import resolve_session_source
 
-        source = get_session_env("HERMES_SESSION_SOURCE", "")
-    except Exception:
-        source = os.environ.get("HERMES_SESSION_SOURCE", "")
-    return str(source or "").strip() or platform or "cli"
+    return resolve_session_source(platform)
 
 
 def _gateway_origin_json(agent: "AIAgent") -> Optional[str]:

tests/agent/test_system_prompt_restore.py

@@ -190,6 +190,139 @@ def test_no_history_with_null_row_does_not_warn(self, caplog):
         assert not any("rebuilding" in m for m in warnings)
 
 
+class TestBotModeGatewayPromptRestore:
+    """Persisted routed chats migrate without changing session identity."""
+
+    @staticmethod
+    def _gateway_agent(tmp_path, stored_prompt):
+        home = tmp_path / ".hermes"
+        yuki_home = home / "profiles" / "yuki"
+        yuki_home.mkdir(parents=True)
+        (yuki_home / "profile.yaml").write_text(
+            "ui_meta:\n  hermes-bots:\n    shape: cloud\n", encoding="utf-8"
+        )
+        db = MagicMock()
+        db.db_path = str(yuki_home / "state.db")
+        db.get_session.return_value = {"system_prompt": stored_prompt}
+        db.get_session_title.return_value = "Group: 42"
+        agent = _make_agent(db, prebuilt_prompt="REBUILT GATEWAY PROMPT")
+        agent.platform = "discord"
+        agent._session_title_hint = None
+        agent._bot_mode_protocol = True
+        return agent
+
+    def test_unstamped_gateway_prompt_upgrades_once(self, tmp_path, monkeypatch):
+        from agent import conversation_loop
+        from tools import bot_mode_probe
+
+        bot_mode_probe._reset_cache_for_tests()
+        agent = self._gateway_agent(tmp_path, "legacy prompt without protocol")
+        monkeypatch.setattr(
+            conversation_loop, "_stored_prompt_matches_runtime", lambda *_: True
+        )
+
+        _restore_or_build_system_prompt(
+            agent, None, [{"role": "user", "content": "hi"}]
+        )
+
+        assert agent._cached_system_prompt == "REBUILT GATEWAY PROMPT"
+        assert agent._session_title_hint is None
+        agent._session_db.update_system_prompt.assert_called_once()
+
+    def test_stale_gateway_prompt_rebuild_preserves_title(self, tmp_path, monkeypatch):
+        from agent import conversation_loop
+        from tools import bot_mode_probe
+
+        bot_mode_probe._reset_cache_for_tests()
+        agent = self._gateway_agent(tmp_path, "stamped but stale prompt")
+        monkeypatch.setattr(
+            conversation_loop, "_stored_prompt_matches_runtime", lambda *_: True
+        )
+        monkeypatch.setattr(
+            bot_mode_probe, "stored_prompt_capability_stale", lambda *_: True
+        )
+
+        _restore_or_build_system_prompt(
+            agent, None, [{"role": "user", "content": "hi"}]
+        )
+
+        assert agent._cached_system_prompt == "REBUILT GATEWAY PROMPT"
+        assert agent._session_title_hint is None
+
+    @pytest.mark.parametrize(
+        ("roster_profile", "platform"),
+        ((False, "discord"), (True, "cli"), (True, "not-a-platform")),
+    )
+    def test_stale_prompt_does_not_migrate_denied_session(
+        self, tmp_path, monkeypatch, roster_profile, platform
+    ):
+        from agent import conversation_loop
+        from tools import bot_mode_probe
+
+        bot_mode_probe._reset_cache_for_tests()
+        stored = "stale capability stamp from a prior surface"
+        agent = self._gateway_agent(tmp_path, stored)
+        agent.platform = platform
+        agent._session_db.get_session_title.return_value = "Ordinary chat"
+        if not roster_profile:
+            # Keep the install managed via another profile while tombstoning
+            # the current routed profile out of profiles.list.
+            tombstone = tmp_path / ".hermes" / "profiles" / ".deleted" / "yuki"
+            tombstone.parent.mkdir(parents=True)
+            tombstone.write_text("deleted\n", encoding="utf-8")
+            other = tmp_path / ".hermes" / "profiles" / "researcher"
+            other.mkdir(parents=True)
+            (other / "profile.yaml").write_text(
+                "ui_meta:\n  hermes-bots:\n    shape: cloud\n", encoding="utf-8"
+            )
+        monkeypatch.setattr(
+            conversation_loop, "_stored_prompt_matches_runtime", lambda *_: True
+        )
+        monkeypatch.setattr(
+            bot_mode_probe, "stored_prompt_capability_stale", lambda *_: True
+        )
+
+        _restore_or_build_system_prompt(
+            agent, None, [{"role": "user", "content": "hi"}]
+        )
+
+        assert agent._cached_system_prompt == stored
+        agent._build_system_prompt.assert_not_called()
+        agent._session_db.update_system_prompt.assert_not_called()
+
+    def test_bot_mode_prompt_is_not_restored_into_denied_task_source(
+        self, tmp_path, monkeypatch
+    ):
+        """A trusted CLI Bot Chat prompt cannot cross into a tool-owned turn."""
+        from gateway.session_context import clear_session_vars, set_session_vars
+        from tools import bot_mode_probe
+
+        bot_mode_probe._reset_cache_for_tests()
+        yuki_home = tmp_path / ".hermes" / "profiles" / "yuki"
+        stored = (
+            "trusted prompt\n"
+            "## Messaging other agents\n"
+            f"{bot_mode_probe.epoch_line(yuki_home)}\n"
+            "Platform: cli"
+        )
+        agent = self._gateway_agent(tmp_path, stored)
+        agent.platform = "cli"
+        agent._session_db.get_session_title.return_value = "Bot Chat"
+
+        tokens = set_session_vars(source="tool")
+        try:
+            _restore_or_build_system_prompt(
+                agent, None, [{"role": "user", "content": "hi"}]
+            )
+        finally:
+            clear_session_vars(tokens)
+
+        assert agent._cached_system_prompt == "REBUILT GATEWAY PROMPT"
+        agent._session_db.update_system_prompt.assert_called_once_with(
+            agent.session_id, "REBUILT GATEWAY PROMPT"
+        )
+
+
 # ---------------------------------------------------------------------------
 # Byte-stability invariant
 # ---------------------------------------------------------------------------