diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 06996f673f..869c8ad04d 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -360,7 +360,26 @@ class GatewayAuthorizationMixin: def _adapter_profile_for_source(self, source: SessionSource) -> Optional[str]: """Resolve the transport-owning profile for adapter policy lookups.""" owner = self._transport_owner(source) - return owner[1] if owner is not None else getattr(source, "profile", None) + if owner is not None: + return owner[1] + from gateway.session_identity import identity_of + identity = identity_of(source) + if identity is not None and identity.multiplexed: + return None if identity.transport_profile == "default" else identity.transport_profile + return getattr(source, "profile", None) + + def _restored_source(self, entry) -> Optional[SessionSource]: + """``entry.origin`` with its identity re-pinned from the routing entry's persisted + ``transport_profile`` (no live adapter: the restored row of the transport matrix). Every path + that revives a session from durable state — auto-resume, heartbeat restore, plugin injection, + background-process events — reads the origin through here, so the receiving bot decides + delivery and authorization after a restart, not the runtime profile's heuristics.""" + source = getattr(entry, "origin", None) + if source is None: + return None + from gateway.session_identity import restore_identity + restore_identity(source, runner=self, transport_profile=getattr(entry, "transport_profile", None)) + return source def _adapter_flag(self, platform, name: str, profile) -> bool: """Adapter-declared boolean, False when unknown. ``authorization_is_upstream`` (relay: a trusted diff --git a/gateway/run_heartbeat_restore.py b/gateway/run_heartbeat_restore.py index b430ba676e..c4022fb66d 100644 --- a/gateway/run_heartbeat_restore.py +++ b/gateway/run_heartbeat_restore.py @@ -48,10 +48,11 @@ async def restore_heartbeat_watches(runner) -> None: if entry.origin is None or not entry.session_id or entry.suspended: continue try: - with runner._profile_scope_for_source(entry.origin): + source = runner._restored_source(entry) + with runner._profile_scope_for_source(source): manager = HeartbeatManager(entry.session_id) if manager.is_active(): - restored.append((entry.session_key, entry.origin, entry.session_id)) + restored.append((entry.session_key, source, entry.session_id)) except Exception: logger.debug("heartbeat restore for %s failed", entry.session_key, exc_info=True) return restored diff --git a/gateway/run_inbound.py b/gateway/run_inbound.py index 17db75bd8d..c4c4fea783 100644 --- a/gateway/run_inbound.py +++ b/gateway/run_inbound.py @@ -1857,7 +1857,8 @@ class GatewayInboundMixin: if entry is None or entry.origin is None or not _accepting(): return False - source = dataclasses.replace(entry.origin) + from gateway.session_identity import replace_source + source = replace_source(self._restored_source(entry)) try: authorized = self._is_user_authorized_for_source(source, allow_adapter_delegation=False) except Exception: diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index 891c1222e1..412667a2fb 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -969,7 +969,7 @@ class GatewayNotificationsMixin: self.session_store._ensure_loaded() entry = self.session_store._entries.get(session_key) if entry and getattr(entry, "origin", None): - return entry.origin + return self._restored_source(entry) except Exception as exc: logger.debug("Synthetic process-event session-store lookup failed for %s: %s", session_key, exc) cached_source = self._get_cached_session_source(session_key) diff --git a/gateway/run_startup.py b/gateway/run_startup.py index c20155a2e0..28e8c6c5e9 100644 --- a/gateway/run_startup.py +++ b/gateway/run_startup.py @@ -566,7 +566,7 @@ class GatewayStartupMixin: # Already being resumed (e.g. scheduled at startup, still in-flight) — no second turn. if self._is_session_running(entry.session_key): continue - source = entry.origin + source = self._restored_source(entry) adapter = self._delivery_adapter_for(source) if adapter is None: logger.debug( @@ -816,7 +816,7 @@ class GatewayStartupMixin: ) with suppress(Exception): from hermes_cli.profiles import get_active_profile_name - _profile = get_active_profile_name() + _profile = get_active_profile_name() # launch profile, pre-identity (boot log) if _profile and _profile != "default": logger.info("Active profile: %s", _profile) _write_runtime_status_quiet(gateway_state="starting", exit_reason=None, clear_profile_platforms=True) diff --git a/gateway/session.py b/gateway/session.py index 4a65b28287..a2be18cbbc 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -14,6 +14,7 @@ from typing import Dict, List, Optional, Any from .config import Platform, GatewayConfig, HomeChannel from .whatsapp_identity import canonical_whatsapp_identifier +from gateway.session_identity import transport_profile_of from gateway.session_persistence import SessionPersistenceMixin, _DB_UNPINNED from gateway.session_recovery import SessionRecoveryMixin from gateway.session_lifecycle import SessionLifecycleMixin, _iso, _new_session_id, _now, _parse_iso @@ -519,6 +520,10 @@ class SessionEntry: # Session-scoped /model override (model/provider/base_url ONLY — never credentials, see # sanitize_model_override). Persisted so a restart keeps the chosen model. model_override: Optional[Dict[str, str]] = None + # Profile owning the bot that received this lane's traffic (``RoutingIdentity.transport_profile``, + # "default" spelled out). The key namespace only says where the turn RUNS; after a restart this is + # what says which bot may deliver to it. None = unknown (row predates the field, or standalone). + transport_profile: Optional[str] = None # Fields (de)serialized verbatim, in wire order (``from_dict`` reads them with # ``data.get(name, )``), split around the three ISO-datetime/token keys. @@ -548,6 +553,8 @@ class SessionEntry: if self.model_override: # Defence-in-depth against an unsanitized dict stored directly. result["model_override"] = sanitize_model_override(self.model_override) + if self.transport_profile: + result["transport_profile"] = self.transport_profile if self.origin: result["origin"] = self.origin.to_dict() return result @@ -578,6 +585,7 @@ class SessionEntry: defaults = {f.name: f.default for f in fields(cls)} plain = {n: data.get(n, defaults[n]) for n in cls._PLAIN_FIELDS + cls._RESET_FIELDS} plain["expiry_finalized"] = data.get("expiry_finalized", data.get("memory_flushed", False)) + transport_profile = data.get("transport_profile") return cls( session_key=session_key, session_id=session_id, created_at=datetime.fromisoformat(data["created_at"]), @@ -586,7 +594,9 @@ class SessionEntry: chat_type=data.get("chat_type", "dm"), metadata=dict(data.get("metadata") or {}), last_resume_marked_at=_parse_iso(data.get("last_resume_marked_at")), active_turn_token=token, active_turn_started_at=started_at, - model_override=sanitize_model_override(data.get("model_override")), **plain, + model_override=sanitize_model_override(data.get("model_override")), + transport_profile=transport_profile if isinstance(transport_profile, str) and transport_profile else None, + **plain, ) @@ -1013,7 +1023,7 @@ class SessionStore( origin=source, display_name=source.chat_name, platform=source.platform, chat_type=source.chat_type, was_auto_reset=decision.reset_reason is not None, auto_reset_reason=decision.reset_reason, reset_had_activity=decision.reset_had_activity, - prev_session_id=decision.prev_session_id, + prev_session_id=decision.prev_session_id, transport_profile=transport_profile_of(source), ) with self._lock: current = self._entries.get(session_key) @@ -1044,9 +1054,11 @@ class SessionStore( entry.last_prompt_tokens = last_prompt_tokens # Snapshot peer fields under _lock so a concurrent reset/heal cannot tear the row. peer_sid, peer_origin, peer_name = entry.session_id, entry.origin, entry.display_name + peer_transport = entry.transport_profile # Metadata-only: single-row UPSERT, outside ``_lock``. self._save_entry(session_key) - self._record_gateway_session_peer(peer_sid, session_key, peer_origin, display_name=peer_name) + self._record_gateway_session_peer( + peer_sid, session_key, peer_origin, display_name=peer_name, transport_profile=peer_transport) def get_session_metadata(self, session_key: str, key: str, default: Any = None) -> Any: """Return a metadata value stored on a live session entry.""" @@ -1117,7 +1129,7 @@ class SessionStore( new_entry = SessionEntry( session_key=session_key, session_id=session_id, created_at=now, updated_at=now, origin=old_entry.origin, platform=old_entry.platform, chat_type=old_entry.chat_type, - **fields, + transport_profile=old_entry.transport_profile, **fields, ) self._entries[session_key] = new_entry self._save() @@ -1212,6 +1224,7 @@ class SessionStore( self._record_gateway_session_peer( target_session_id, session_key, new_entry.origin, display_name=new_entry.display_name, include_compression_ancestors=True, + transport_profile=new_entry.transport_profile, ) return new_entry diff --git a/gateway/session_identity.py b/gateway/session_identity.py index c1408fb95f..fa4512d3ec 100644 --- a/gateway/session_identity.py +++ b/gateway/session_identity.py @@ -99,6 +99,13 @@ def clear_identity(source: Any) -> None: source.profile = None +def transport_profile_of(source: Any) -> Optional[str]: + """The receiving bot's profile to persist alongside a routing entry (``SessionEntry.transport_profile``); + None outside multiplexing or when nothing resolved the source (an unknown transport is never guessed).""" + identity = identity_of(source) + return identity.transport_profile if identity is not None and identity.multiplexed else None + + def replace_source(source: "SessionSource", **changes: Any) -> "SessionSource": """:func:`dataclasses.replace` that keeps the wire-invisible provenance (transport ref, authorization home, identity). A plain ``replace`` silently produces a source the runner @@ -136,6 +143,46 @@ def canonical_identity( return None +def restore_identity( + source: "SessionSource", *, runner: Any, transport_profile: Optional[str], +) -> Optional[RoutingIdentity]: + """Pin the identity of a source rebuilt from durable state (``SessionEntry.origin``, a + ``sessions`` row, a cached copy) — no live adapter, so ``transport=None``: the restored row of the + transport matrix, where delivery goes through the persisted transport owner or fails closed. + + *transport_profile* is what the routing index persisted at ingress (``SessionEntry.transport_profile``); + ``None`` = a row written before the column existed, whose transport is unknown → nothing is pinned + and the legacy heuristics (``_is_shared_bot_satellite``) keep deciding. Standalone gateways have + nothing to restore (one bot, one home). + """ + transport_name = _name(transport_profile) + if transport_name is None: + return None + if not bool(getattr(getattr(runner, "config", None), "multiplex_profiles", False)): + return None + existing = identity_of(source) + if existing is not None: + return existing + from hermes_cli.profiles import get_profile_dir + from hermes_constants import get_process_hermes_home + + primary_profile = _name(getattr(runner, "_primary_profile_name", None)) or "default" + runtime_name = _name(getattr(source, "profile", None)) or primary_profile + authorization_home = ( + Path(get_process_hermes_home()) if transport_name == primary_profile + else get_profile_dir(transport_name)) + runtime_home = ( + authorization_home if runtime_name == transport_name + else Path(runner._resolve_profile_home_for_source(source))) + source._authorization_profile_home = authorization_home + identity = RoutingIdentity( + transport_profile=transport_name, runtime_profile=runtime_name, + authorization_home=authorization_home, runtime_home=runtime_home, + multiplexed=True, transport=None) + setattr(source, _IDENTITY_ATTR, identity) + return identity + + def resolve_identity( source: "SessionSource", *, runner: Any, adapter: Any = None, transport_profile: Optional[str] = None, primary_home: Optional[Path] = None, diff --git a/gateway/session_recovery.py b/gateway/session_recovery.py index 770f69295a..a466af5a25 100644 --- a/gateway/session_recovery.py +++ b/gateway/session_recovery.py @@ -156,11 +156,12 @@ class SessionRecoveryMixin: had_activity = row.get("_has_messages") if had_activity is None: had_activity = bool(row.get("message_count") or 0) or last_activity is not None + from gateway.session_identity import transport_profile_of return SessionEntry( session_key=session_key, session_id=str(row["id"]), created_at=created_at, updated_at=updated_at, origin=source, display_name=source.chat_name, platform=source.platform, chat_type=source.chat_type, - reset_had_activity=bool(had_activity)) + reset_had_activity=bool(had_activity), transport_profile=transport_profile_of(source)) def _find_gateway_session_row( self, *, session_key: str, source: SessionSource, allow_peer_fallback: bool, @@ -324,14 +325,18 @@ class SessionRecoveryMixin: def _record_gateway_session_peer( self, session_id: str, session_key: str, source: Optional[SessionSource], - display_name: Optional[str] = None, include_compression_ancestors: bool = False) -> None: - """Persist the routing peer for an existing gateway session row.""" + display_name: Optional[str] = None, include_compression_ancestors: bool = False, + transport_profile: Optional[str] = None) -> None: + """Persist the routing peer for an existing gateway session row. ``transport_profile`` is the + entry's persisted receiving-bot profile; when the caller has no entry it is read off the + source's pinned identity (None = unknown, the column keeps whatever an earlier writer set).""" db = self._db_for_key(session_key) if not db or not source: return recorder = getattr(db, "record_gateway_session_peer", None) if not callable(recorder): return + from gateway.session_identity import transport_profile_of peer = dict( source=source.platform.value, user_id=source.user_id, session_key=session_key, chat_id=source.chat_id, chat_type=source.chat_type, thread_id=source.thread_id) @@ -339,7 +344,8 @@ class SessionRecoveryMixin: recorder( session_id, **peer, display_name=display_name or source.chat_name, origin_json=_origin_json(source), - include_compression_ancestors=include_compression_ancestors) + include_compression_ancestors=include_compression_ancestors, + transport_profile=transport_profile or transport_profile_of(source)) except TypeError: try: # older SessionDB without display_name/origin_json kwargs recorder(session_id, **peer) @@ -410,6 +416,7 @@ class SessionRecoveryMixin: """kwargs for ``SessionDB.create_session``. Identity (origin_json) and lineage (parent/_reset_from) land atomically in the INSERT so a crash right after cannot strand the row unroutable.""" + from gateway.session_identity import transport_profile_of return { "session_id": session_id, "source": source_value, @@ -419,6 +426,7 @@ class SessionRecoveryMixin: "chat_type": origin.chat_type if origin else None, "thread_id": origin.thread_id if origin else None, "profile_name": origin.profile if origin else None, + "transport_profile": transport_profile_of(origin), "origin_json": _origin_json(origin), "display_name": display_name, "parent_session_id": parent_session_id, diff --git a/hermes_state_common.py b/hermes_state_common.py index 5868886951..03b3467991 100644 --- a/hermes_state_common.py +++ b/hermes_state_common.py @@ -388,6 +388,7 @@ CREATE TABLE IF NOT EXISTS sessions ( compression_ineffective_count INTEGER NOT NULL DEFAULT 0, compression_recovery_deadline REAL, profile_name TEXT, + transport_profile TEXT, rewind_count INTEGER NOT NULL DEFAULT 0, archived INTEGER NOT NULL DEFAULT 0, pinned INTEGER NOT NULL DEFAULT 0, diff --git a/hermes_state_gateway.py b/hermes_state_gateway.py index 8d2bfa652d..13af59e250 100644 --- a/hermes_state_gateway.py +++ b/hermes_state_gateway.py @@ -216,7 +216,8 @@ class SessionGatewayMixin: def record_gateway_session_peer( self, session_id: str, *, source: str, user_id: str = None, session_key: str = None, chat_id: str = None, chat_type: str = None, thread_id: str = None, display_name: str = None, - origin_json: str = None, include_compression_ancestors: bool = False) -> None: + origin_json: str = None, include_compression_ancestors: bool = False, + transport_profile: str = None) -> None: """Persist the gateway routing peer for an existing session row. ``display_name`` / ``origin_json``: ``None`` leaves the stored value untouched (consumers read routing data from state.db, not sessions.json). ``include_compression_ancestors`` keeps a compression lineage on one routing peer @@ -230,7 +231,9 @@ class SessionGatewayMixin: """ if not session_id or not session_key: return - identity = (session_key, source, user_id, chat_id, chat_type, thread_id, display_name, origin_json) + identity = ( + session_key, source, user_id, chat_id, chat_type, thread_id, display_name, origin_json, + transport_profile) ancestors = include_compression_ancestors query_params = [session_id, *identity] if ancestors else [*identity, session_id] def _do(conn): @@ -240,7 +243,8 @@ class SessionGatewayMixin: SET session_key = ?, source = ?, user_id = ?, chat_id = ?, chat_type = ?, thread_id = ?, display_name = COALESCE(?, display_name), - origin_json = COALESCE(?, origin_json) + origin_json = COALESCE(?, origin_json), + transport_profile = COALESCE(?, transport_profile) {"WHERE id IN (SELECT id FROM compression_lineage)" if ancestors else "WHERE id = ?"}""", query_params, ) @@ -252,20 +256,21 @@ class SessionGatewayMixin: """INSERT INTO sessions ( id, source, user_id, session_key, chat_id, chat_type, thread_id, display_name, origin_json, - profile_name, started_at + profile_name, transport_profile, started_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET session_key = COALESCE(sessions.session_key, excluded.session_key), chat_id = COALESCE(sessions.chat_id, excluded.chat_id), chat_type = COALESCE(sessions.chat_type, excluded.chat_type), thread_id = COALESCE(sessions.thread_id, excluded.thread_id), display_name = COALESCE(sessions.display_name, excluded.display_name), - origin_json = COALESCE(sessions.origin_json, excluded.origin_json)""", + origin_json = COALESCE(sessions.origin_json, excluded.origin_json), + transport_profile = COALESCE(sessions.transport_profile, excluded.transport_profile)""", # Same ownership stamp as _insert_session_row: an unowned (NULL) row # vanishes from profile-keyed consumers. (session_id, source, user_id, session_key, chat_id, chat_type, thread_id, display_name, - origin_json, self._own_profile_name(), time.time()), + origin_json, self._own_profile_name(), transport_profile, time.time()), ) self._execute_write(_do) diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py index 4208283d3d..5d82156ad6 100644 --- a/hermes_state_sessions.py +++ b/hermes_state_sessions.py @@ -213,7 +213,7 @@ _SAME_KEY_NAMESPACE_SQL = ( _UPSERT_KEEP_EXISTING_SQL = ",\n".join( f" {col} = COALESCE(sessions.{col}, excluded.{col})" for col in ( "session_key", "chat_id", "chat_type", "thread_id", "parent_session_id", "cwd", "profile_name", - "git_repo_root", "origin_json", "display_name", + "transport_profile", "git_repo_root", "origin_json", "display_name", ) ) @@ -240,6 +240,7 @@ _INHERIT_PARENT_ROUTING_SQL = ( "UPDATE sessions\n SET " + _INHERIT_SEP.join(_inherit_col_sql(c) for c in ( "user_id", "session_key", "chat_id", "chat_type", "thread_id", "display_name", "origin_json", + "transport_profile", )) + "\n WHERE id = ? AND parent_session_id IS NOT NULL\n" " AND EXISTS (\n" @@ -285,6 +286,7 @@ class SessionSessionsMixin: chat_id: str = None, chat_type: str = None, thread_id: str = None, parent_session_id: str = None, cwd: str = None, profile_name: Optional[str] = None, git_repo_root: str = None, origin_json: str = None, display_name: str = None, + transport_profile: Optional[str] = None, ) -> None: """Upsert a session row, never overwriting what an earlier writer set (the gateway creates a bare row before create_session carries the real model/prompt) — the one exception is the @@ -323,10 +325,10 @@ class SessionSessionsMixin: """INSERT INTO sessions ( id, source, user_id, session_key, chat_id, chat_type, thread_id, model, model_config, system_prompt, system_prompt_hash, - parent_session_id, cwd, profile_name, git_repo_root, + parent_session_id, cwd, profile_name, transport_profile, git_repo_root, origin_json, display_name, started_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET source = CASE WHEN sessions.source = 'unknown' @@ -367,8 +369,8 @@ class SessionSessionsMixin: ( session_id, source, user_id, session_key, chat_id, chat_type, thread_id, model, json.dumps(model_config) if model_config else None, system_prompt_hash, - parent_session_id, cwd, profile_name, git_repo_root, origin_json, display_name, - time.time(), + parent_session_id, cwd, profile_name, transport_profile, git_repo_root, origin_json, + display_name, time.time(), ), ) if system_prompt_hash is not None: diff --git a/tests/gateway/test_routing_save_fast_path.py b/tests/gateway/test_routing_save_fast_path.py index c7ea83335e..6951efe265 100644 --- a/tests/gateway/test_routing_save_fast_path.py +++ b/tests/gateway/test_routing_save_fast_path.py @@ -207,7 +207,7 @@ class TestPeerRecordConsistency: monkeypatch.setattr( store, "_record_gateway_session_peer", - lambda sid, key, origin, display_name=None: recorded.append( + lambda sid, key, origin, display_name=None, **_kw: recorded.append( (sid, key, display_name) ), ) diff --git a/tests/gateway/test_session_identity_restore.py b/tests/gateway/test_session_identity_restore.py new file mode 100644 index 0000000000..9749e920ac --- /dev/null +++ b/tests/gateway/test_session_identity_restore.py @@ -0,0 +1,154 @@ +"""Identity survives restore (#88715 phase 5). + +A restarted multiplexed gateway rebuilds every lane from the routing index; the key namespace says +where the lane runs but not which bot received it. ``SessionEntry.transport_profile`` persists that +bot, ``_restored_source`` re-pins the identity, and delivery goes through that bot or fails closed. +Real ``GatewayRunner`` resolvers and a real ``SessionStore`` over a temp ``HERMES_HOME`` — no +patched predicates. +""" + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.pairing import PairingStore +from gateway.platforms.base import BasePlatformAdapter +from gateway.profile_routing import parse_profile_routes +from gateway.session import SessionEntry, SessionStore +from gateway.session_identity import identity_of, resolve_identity + + +class _Stub(BasePlatformAdapter): + pass + + +_Stub.__abstractmethods__ = frozenset() + + +def _stub(platform, runner, label): + adapter = _Stub.__new__(_Stub) + adapter.platform, adapter.gateway_runner, adapter.label = platform, runner, label + adapter.config = PlatformConfig(enabled=True, extra={}) + adapter._pending_messages, adapter._active_sessions = {}, {} + return adapter + + +_ROUTES = [ + # Satellite ``ops`` drains through the default bot for this chat ... + {"name": "admin-dm", "platform": "telegram", "profile": "ops", "chat_id": "72719239"}, + # ... and is ALSO the routed runtime for a chat that team_b's OWN bot receives. + {"name": "b-to-ops", "platform": "telegram", "profile": "ops", "chat_id": "555", "bot_profile": "team_b"}, +] + + +def _runner(home, *, multiplex=True): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + runner.config = GatewayConfig(multiplex_profiles=multiplex, sessions_dir=home / "sessions") + runner.config.platforms = {Platform.TELEGRAM: PlatformConfig(enabled=True, extra={})} + runner.config.profile_routes = parse_profile_routes(list(_ROUTES) if multiplex else []) + runner.pairing_store = PairingStore(profile="default") + runner.pairing_stores = {} + runner._primary_profile_name = "default" + primary = _stub(Platform.TELEGRAM, runner, "PRIMARY") + team_b = _stub(Platform.TELEGRAM, runner, "TEAM_B") + team_b.set_owner_profile("team_b") + runner.adapters = {Platform.TELEGRAM: primary} + runner._profile_adapters = {"team_b": {Platform.TELEGRAM: team_b}, "ops": {}} + return SimpleNamespace(runner=runner, home=home, primary=primary, team_b=team_b) + + +@pytest.fixture +def mux(tmp_path, monkeypatch): + home = tmp_path / "hh" + for name in ("ops", "team_b"): + (home / "profiles" / name).mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(home)) + served = [("default", home), ("ops", home / "profiles" / "ops"), ("team_b", home / "profiles" / "team_b")] + with patch("hermes_cli.profiles.profiles_to_serve", return_value=served), \ + patch("hermes_cli.profiles.get_profile_dir", side_effect=lambda n: home if n == "default" else home / "profiles" / n), \ + patch("hermes_cli.profiles.profile_exists", return_value=True): + yield _runner(home) + + +def _restart(rig, entry: SessionEntry): + """A fresh process: the entry comes back through its wire dict, the runner is rebuilt.""" + fresh = _runner(rig.home) + restored = SessionEntry.from_dict(entry.to_dict()) + return fresh, fresh.runner._restored_source(restored), restored + + +def test_restored_lane_delivers_through_the_bot_that_received_it_never_the_default_by_heuristic(mux, tmp_path): + """Chat 555 arrives on team_b's bot and runs as ``ops``. Before the restart the transport ref + answers; after it only the persisted ``transport_profile`` can — without it the shared-bot + heuristic (ops IS a satellite of the default bot, for chat 72719239) hands the lane to the + DEFAULT bot. The satellite lane itself keeps its default-bot egress, the row lands in state.db, + and a pre-column entry (``transport_profile`` absent) still resolves as before.""" + store = SessionStore(sessions_dir=mux.home / "sessions", config=mux.runner.config) + mux.runner.session_store = store + + routed = mux.team_b.build_source(chat_id="555", chat_type="dm", user_id="555") + identity = resolve_identity(routed, runner=mux.runner, transport_profile="team_b") + assert (identity.transport_profile, identity.runtime_profile) == ("team_b", "ops") + entry = store.get_or_create_session(routed) + assert entry.session_key.startswith("agent:ops:") and entry.transport_profile == "team_b" + assert entry.to_dict()["transport_profile"] == "team_b" + row = store._db_for_key(entry.session_key).get_session(entry.session_id) + assert row["transport_profile"] == "team_b" and row["profile_name"] == "ops" + + fresh, source, restored = _restart(mux, entry) + assert restored.transport_profile == "team_b" + assert fresh.runner._transport_owner(source) is None # no live provenance survives a restart + restored_identity = identity_of(source) + assert restored_identity is not None and restored_identity.transport is None + assert (restored_identity.transport_profile, restored_identity.runtime_profile) == ("team_b", "ops") + assert restored_identity.authorization_home == mux.home / "profiles" / "team_b" + assert restored_identity.runtime_home == mux.home / "profiles" / "ops" + assert fresh.runner._adapter_for_source(source) is fresh.team_b + assert fresh.runner._adapter_profile_for_source(source) == "team_b" + assert fresh.runner._authorization_home_for_source(source) == mux.home / "profiles" / "team_b" + # Fail closed: team_b's bot did not reconnect → nothing delivers; the default bot never does. + fresh.runner._profile_adapters["team_b"] = {} + assert fresh.runner._adapter_for_source(source) is None + + # The satellite lane (shared default bot, runtime ops) keeps its default-bot egress. + shared = mux.primary.build_source(chat_id="72719239", chat_type="dm", user_id="72719239") + resolve_identity(shared, runner=mux.runner) + shared_entry = store.get_or_create_session(shared) + assert shared_entry.transport_profile == "default" + fresh2, shared_source, _ = _restart(mux, shared_entry) + assert identity_of(shared_source).transport_profile == "default" + assert fresh2.runner._adapter_for_source(shared_source) is fresh2.primary + + # A routing entry written before the column existed: nothing is pinned, old chain unchanged. + legacy = entry.to_dict() + legacy.pop("transport_profile") + fresh3 = _runner(mux.home) + legacy_source = fresh3.runner._restored_source(SessionEntry.from_dict(legacy)) + assert identity_of(legacy_source) is None + assert fresh3.runner._adapter_for_source(legacy_source) is fresh3.primary # the heuristic, as before + + +def test_standalone_gateway_persists_nothing_and_keys_stay_agent_main(tmp_path, monkeypatch): + """Control: outside multiplexing there is one bot and one home — no transport is recorded, the + wire dict is byte-identical to before, and a restored source resolves as it always did.""" + home = tmp_path / "solo" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + solo = _runner(home, multiplex=False) + store = SessionStore(sessions_dir=home / "sessions", config=solo.runner.config) + source = solo.primary.build_source(chat_id="4040", chat_type="dm", user_id="4040") + resolve_identity(source, runner=solo.runner) + entry = store.get_or_create_session(source) + assert entry.session_key == "agent:main:telegram:dm:4040" + assert entry.transport_profile is None and "transport_profile" not in entry.to_dict() + assert store._db_for_key(entry.session_key).get_session(entry.session_id)["transport_profile"] is None + fresh = _runner(home, multiplex=False) + restored = fresh.runner._restored_source(SessionEntry.from_dict(entry.to_dict())) + assert identity_of(restored) is None + assert fresh.runner._adapter_for_source(restored) is fresh.primary + assert fresh.runner._session_key_for_source(restored) == "agent:main:telegram:dm:4040"