diff --git a/gateway/AGENTS.md b/gateway/AGENTS.md index 0bb5baa37a83e..075c6ae21592a 100644 --- a/gateway/AGENTS.md +++ b/gateway/AGENTS.md @@ -164,6 +164,20 @@ gateway under the backend, and do NOT "fix" update locks by widening the tree-ki | shared bot → profile that owns its own bot | routed | receiving adapter | receiving adapter (conversation continuity; #70625's "routed bot" reading was not adopted) | | secondary-owned bot → `default` (`bot_profile`) | default (`agent:main`) | receiving adapter | receiving adapter | | restored / hand-built, no live provenance | stored `source.profile` | **`None`** | unique owner of `(platform, runtime)`; a disconnected secondary → `None`, never the default bot | +- **Identity survives the process.** `SessionEntry.transport_profile` (routing index + + `sessions.transport_profile`, nullable, reconciled by `SCHEMA_SQL`) persists the receiving bot + next to the key namespace; the namespace says where a lane RUNS, the column says which bot may + DELIVER to it. Anything reviving a session from durable state (auto-resume, heartbeat restore, + plugin injection, background-process events) reads `entry.origin` through + `authz_mixin.py::_restored_source`, which re-pins a `RoutingIdentity(transport=None)` via + `session_identity.restore_identity`; `_delivery_adapter_for` then delivers through that bot's + adapter or nothing (never the default bot by heuristic). Rows without the column (pre-PR-5) + keep the `_is_shared_bot_satellite` fallback. Deferred callbacks capture the identity/home at + command time (`/model` picker); `_run_in_executor_with_context` carries the scope over thread + hops. Relay: `_with_scope` echoes the routed `profile` on every outbound frame and `follow_up` + derives it from the key namespace so the connector stamps it on the next `passthrough_forward`. + Ambient `get_active_profile_name()` reads in `gateway/` are boot-only and marked + `# launch profile, pre-identity`; a path with an identity reads `identity.runtime_profile`. - **Token locks.** An adapter that connects with a unique credential (bot token, API key) calls `acquire_scoped_lock()` from `gateway.status` in `connect()`/`start()` and `release_scoped_lock()` in `disconnect()`/`stop()`, so two profiles cannot share one credential. Canonical: diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 06996f673fc6b..0232288eed8ed 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -300,6 +300,17 @@ def _delivery_adapter_for(self, source: Optional[SessionSource]): adapter = self._intake_adapter_for(source) if adapter is not None: return adapter + # A pinned identity NAMES the receiving bot (live or restored from ``transport_profile``). + # If that bot has no adapter right now it is offline: fail closed rather than fall through to + # the runtime profile's bot — that fallthrough is the "restored lane answers from the wrong + # bot" row the identity exists to close. Only an identity-less source (legacy row, bare + # fixture) uses the unique-owner heuristic below. + from gateway.session_identity import identity_of + identity = identity_of(source) + if identity is not None and identity.multiplexed and not identity.transport_inferred: + return None + # No identity, or one whose transport was only inferred (hand-built source, pre-column row): + # the unique owner of ``(platform, runtime_profile)`` delivers. # ``getattr``: test fixtures build bare SimpleNamespace sources without ``profile``. return self._authorization_adapter(getattr(source, "platform", None), getattr(source, "profile", None)) @@ -360,7 +371,26 @@ def _authorization_home_for_source(self, source: SessionSource): 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/platforms/api_server.py b/gateway/platforms/api_server.py index 899cb9589e347..e6180a9ba561e 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -1308,7 +1308,7 @@ def _resolve_model_name(explicit: str) -> str: profile_name = "" 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 (advertised model name) if profile and profile not in {"default", "custom"}: profile_name = profile return resolve_effective_model(explicit, profile_name, "hermes-agent") diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index 03529c3f999c7..01b4cedce664c 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -95,6 +95,17 @@ def _event_ids(event) -> Tuple[Optional[str], Optional[str]]: return message_id, getattr(event.source, "chat_id", None) +def _profile_from_session_key(session_key: str) -> Optional[str]: + """Named profile encoded in an ``agent::...`` session key; None for the legacy ``agent:main`` + namespace (single-profile gateway) so the wire frame stays byte-identical there.""" + parts = (session_key or "").split(":") + if len(parts) < 2 or parts[0] != "agent" or not parts[1]: + return None + from gateway.session import profile_from_session_key_namespace + profile = profile_from_session_key_namespace(parts[1]) + return None if profile == "default" else profile + + class RelayAdapter(BasePlatformAdapter): """Generic relay adapter advertising a connector-negotiated capability profile.""" @@ -125,6 +136,10 @@ def __init__( # platforms on one WS and a reply must egress through the platform the # inbound came from. Empty for a single-platform gateway (connector default). self._platform_by_chat: Dict[str, str] = {} + # chat_id -> Hermes profile the connector routed the inbound to (multiplex mode). Echoed + # on every outbound frame's metadata so the connector can stamp the SAME profile on the + # next passthrough_forward for that chat; empty on a single-profile gateway. + self._profile_by_chat: Dict[str, str] = {} # Chats the connector has refused (see the terminal-decline latch). # chat_id -> (thread_id, initial_name) of the auto-thread the CONNECTOR # created for our latest send; read by the semantic thread-rename lane. @@ -1042,6 +1057,7 @@ def _capture_scope(self, event) -> None: for attr, cache in ( ("user_id", self._dm_user_by_chat), ("scope_id", self._scope_by_chat), ("chat_type", self._chat_type_by_chat), + ("profile", self.__dict__.setdefault("_profile_by_chat", {})), ): value = getattr(src, attr, None) if value: @@ -1059,7 +1075,11 @@ def _with_scope(self, chat_id: str, metadata: Optional[Dict[str, Any]]) -> Dict[ first and only falls back to user_id on a route miss, so carrying both never overrides routing-table resolution.""" meta: Dict[str, Any] = dict(metadata or {}) - for key, cache in (("scope_id", self._scope_by_chat), ("user_id", self._dm_user_by_chat)): + # ``getattr``: relay tests build bare adapters via ``__new__`` without ``__init__``. + for key, cache in ( + ("scope_id", self._scope_by_chat), ("user_id", self._dm_user_by_chat), + ("profile", getattr(self, "_profile_by_chat", {})), + ): if not meta.get(key): value = cache.get(str(chat_id)) if value: @@ -1698,13 +1718,20 @@ async def send_follow_up( # default routes it. prefix = kind.split(".", 1)[0] if kind and "." in kind else None follow_up_platform = prefix if prefix and self.fronts_platform(prefix) else None + follow_up_metadata = dict(metadata or {}) + # The session key names the profile namespace the interaction ran under; carry it so the + # connector's next passthrough_forward for this interaction routes to the same profile. + if not follow_up_metadata.get("profile"): + profile = _profile_from_session_key(session_key) + if profile: + follow_up_metadata["profile"] = profile result = await self._transport.send_follow_up( { "op": "follow_up", "session_key": session_key, "kind": kind, "content": content, - "metadata": metadata or {}, + "metadata": follow_up_metadata, }, platform=follow_up_platform, ) diff --git a/gateway/run.py b/gateway/run.py index 3bf9d1605b127..8ee83f0a34d8c 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1613,7 +1613,7 @@ def _cron_tick_profile_homes(config: object) -> list[tuple[str, "Path"]]: from hermes_cli.profiles import get_active_profile_name, get_profile_dir homes = _multiplex_profile_homes(config) - active = get_active_profile_name() or "default" + active = get_active_profile_name() or "default" # launch profile, pre-identity (ticker boot) if any(name == active for name, _home in homes): return homes try: @@ -3841,9 +3841,14 @@ def _session_key_for_source(self, source: SessionSource) -> str: pass config = getattr(self, "config", None) # Mirror SessionStore._resolve_profile_for_key so this fallback yields the primary path's - # namespace: None (legacy agent:main) unless multiplexing is on, then the active profile. + # namespace: None (legacy agent:main) unless multiplexing is on, then the pinned identity's + # runtime profile, the source stamp, or the active profile. + from gateway.session_identity import identity_of + identity = identity_of(source) _profile = None - if getattr(config, "multiplex_profiles", False): + if identity is not None: + _profile = identity.session_key_profile + elif getattr(config, "multiplex_profiles", False): if source.profile: _profile = source.profile else: diff --git a/gateway/run_adapters.py b/gateway/run_adapters.py index 1c9b07afdcb4d..01212bcec7e90 100644 --- a/gateway/run_adapters.py +++ b/gateway/run_adapters.py @@ -853,7 +853,7 @@ async def _start_secondary_profile_adapters(self) -> int: from hermes_cli.profiles import get_active_profile_name except Exception: return 0 - active = get_active_profile_name() or "default" + active = get_active_profile_name() or "default" # launch profile, pre-identity (adapter boot) connected = 0 claimed = self._primary_resource_claims(active) profile_homes = _multiplex_profile_homes(self.config) diff --git a/gateway/run_heartbeat_restore.py b/gateway/run_heartbeat_restore.py index b430ba676ed60..c4022fb66db45 100644 --- a/gateway/run_heartbeat_restore.py +++ b/gateway/run_heartbeat_restore.py @@ -48,10 +48,11 @@ def scan(): 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 17db75bd8d69c..c4c4fea7831dc 100644 --- a/gateway/run_inbound.py +++ b/gateway/run_inbound.py @@ -1857,7 +1857,8 @@ def _accepting() -> bool: 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 891c1222e1d55..412667a2fb3c8 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -969,7 +969,7 @@ def _build_process_event_source(self, evt: dict): 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 c20155a2e049d..28e8c6c5e9f02 100644 --- a/gateway/run_startup.py +++ b/gateway/run_startup.py @@ -566,7 +566,7 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int: # 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 @@ def _start_log_startup_environment(self) -> None: ) 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 4a65b282877e4..a2be18cbbcddc 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -14,6 +14,7 @@ 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 @@ def to_dict(self) -> Dict[str, Any]: 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 @@ def from_dict(cls, data: Dict[str, Any]) -> "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 @@ def from_dict(cls, data: Dict[str, Any]) -> "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 @@ def _route_create( 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 @@ def update_session( 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 @@ def _replace_route_locked(self, session_key, old_entry, session_id, now, **field 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 @@ def switch_session( 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 c1408fb95f1e0..aba8789b506c8 100644 --- a/gateway/session_identity.py +++ b/gateway/session_identity.py @@ -57,6 +57,11 @@ class RoutingIdentity: # Receiving adapter; None for restored/synthetic sources (no live provenance → fail closed). # Provenance, not identity: two events from the same bot share one identity. transport: Optional[weakref.ref] = field(default=None, compare=False, hash=False) + # True when nothing named the receiving bot (no live adapter, no persisted transport_profile, + # no explicit hint) and ``transport_profile`` is the primary by default. A hand-built or + # pre-column source. Delivery may still fall back to the runtime profile's unique adapter for + # these; an identity whose transport is KNOWN (live or restored) never does. + transport_inferred: bool = field(default=False, compare=False, hash=False) @property def namespace(self) -> str: @@ -99,6 +104,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 +148,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, @@ -174,6 +226,7 @@ def resolve_identity( source._transport_adapter_ref = weakref.ref(adapter) _registered, owner_profile = runner._owning_profile(adapter, platform) transport_name = _name(transport_profile) or _name(owner_profile) or primary_profile + transport_inferred = adapter is None and _name(transport_profile) is None and _name(owner_profile) is None transport_ref = weakref.ref(adapter) if adapter is not None else None if not multiplexed: @@ -215,6 +268,6 @@ def resolve_identity( identity = RoutingIdentity( transport_profile=transport_name, runtime_profile=runtime_name, authorization_home=authorization_home, runtime_home=runtime_home, - multiplexed=True, transport=transport_ref) + multiplexed=True, transport=transport_ref, transport_inferred=transport_inferred) setattr(source, _IDENTITY_ATTR, identity) return identity diff --git a/gateway/session_recovery.py b/gateway/session_recovery.py index 770f69295a052..a466af5a25de6 100644 --- a/gateway/session_recovery.py +++ b/gateway/session_recovery.py @@ -156,11 +156,12 @@ def _ts(value, default: datetime) -> datetime: 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 @@ def _reopen_session_row(self, session_key: str, session_id: str, *, log_prefix: 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 @@ def _record_gateway_session_peer( 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 @@ def _session_create_kwargs( """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 @@ def _session_create_kwargs( "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_cli/session_schema_history.py b/hermes_cli/session_schema_history.py index a54c16e536c1a..88aa10350b981 100644 --- a/hermes_cli/session_schema_history.py +++ b/hermes_cli/session_schema_history.py @@ -203,6 +203,7 @@ def reachable_physical_layouts( ('+', 'compression_recovery_deadline', 'compression_ineffective_count'), )), ('26 2026-09-02T14:22Z 8e4366d358', (('+', 'tool_names', 'last_read_at'),)), + ('27 2026-09-19T00:10Z 922a0c3c87', (('+', 'transport_profile', 'profile_name'),)), ), ), "messages": _TableHistory( diff --git a/hermes_state_common.py b/hermes_state_common.py index 586888695124c..03b34679913d8 100644 --- a/hermes_state_common.py +++ b/hermes_state_common.py @@ -388,6 +388,7 @@ def _id_chunks(ids, size: int = _SQL_IN_CHUNK): 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 8d2bfa652d8a1..13af59e250b0b 100644 --- a/hermes_state_gateway.py +++ b/hermes_state_gateway.py @@ -216,7 +216,8 @@ def _reap_inactive_orphan_desktop_holders( 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 @@ def record_gateway_session_peer( """ 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 @@ def _do(conn): 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 @@ def _do(conn): """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 4208283d3d0e6..5d82156ad63b5 100644 --- a/hermes_state_sessions.py +++ b/hermes_state_sessions.py @@ -213,7 +213,7 @@ def classify_session_status(role: Optional[str], has_tool_calls: bool, finish_re _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 @@ def _inherit_col_sql(col: str, extra: str = "") -> str: "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 @@ def _insert_session_row( 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 @@ def _do(conn): """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 @@ def _do(conn): ( 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/relay/test_relay_passthrough.py b/tests/gateway/relay/test_relay_passthrough.py index a8e27e43351fd..e19c6d6a37f68 100644 --- a/tests/gateway/relay/test_relay_passthrough.py +++ b/tests/gateway/relay/test_relay_passthrough.py @@ -266,3 +266,42 @@ async def fake_handle(event): assert ev.source.delivered_via_upstream_relay is True + + +@pytest.mark.asyncio +async def test_routed_profile_round_trips_on_every_egress_frame(adapter, monkeypatch): + """Relay passthrough round-trip keeps ``profile`` (#88715 phase 5): the profile the connector + stamped on an inbound interaction is echoed on the chat's outbound frames and on the + ``follow_up`` addressed by the routed session key, so the connector can stamp it on the NEXT + passthrough_forward for that chat; a single-profile gateway emits no ``profile`` key at all.""" + await adapter.connect() + stub = adapter._transport + monkeypatch.setattr(adapter, "handle_message", _noop_handle) + + fwd = _interaction_forward( + { + "id": "interaction-3", "type": 2, "channel_id": "chan-9", "guild_id": "guild-7", + "data": {"name": "summarize"}, "member": {"user": {"id": "user-3", "username": "ben"}}, + }, + profile="reviewer", + ) + await stub.push_passthrough(fwd, buffer_id=None) + await adapter.send("chan-9", "done") + await adapter.send_follow_up( + session_key="agent:reviewer:discord:group:chan-9", kind="discord.interaction_token", content="x") + assert stub.sent[-1]["metadata"]["profile"] == "reviewer" + assert stub.follow_ups[-1]["metadata"]["profile"] == "reviewer" + + # Legacy namespace / unrouted chat: byte-identical frames, no profile key. + await stub.push_passthrough(_interaction_forward({ + "id": "interaction-4", "type": 2, "channel_id": "chan-1", "guild_id": "guild-7", + "data": {"name": "summarize"}, "member": {"user": {"id": "user-3"}}}), buffer_id=None) + await adapter.send("chan-1", "done") + await adapter.send_follow_up( + session_key="agent:main:discord:group:chan-1", kind="discord.interaction_token", content="x") + assert "profile" not in stub.sent[-1]["metadata"] + assert "profile" not in stub.follow_ups[-1]["metadata"] + + +async def _noop_handle(event): + return None diff --git a/tests/gateway/test_routing_save_fast_path.py b/tests/gateway/test_routing_save_fast_path.py index c7ea83335ee85..6951efe2659cc 100644 --- a/tests/gateway/test_routing_save_fast_path.py +++ b/tests/gateway/test_routing_save_fast_path.py @@ -207,7 +207,7 @@ def test_update_session_records_peer_fields_snapshotted_under_lock( 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 0000000000000..48dc1326ec784 --- /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._delivery_adapter_for(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._delivery_adapter_for(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._delivery_adapter_for(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._delivery_adapter_for(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._delivery_adapter_for(restored) is fresh.primary + assert fresh.runner._session_key_for_source(restored) == "agent:main:telegram:dm:4040" diff --git a/website/docs/developer-guide/multiplexing-gateway.md b/website/docs/developer-guide/multiplexing-gateway.md index dea5122e7db5d..b5cad80ef355b 100644 --- a/website/docs/developer-guide/multiplexing-gateway.md +++ b/website/docs/developer-guide/multiplexing-gateway.md @@ -222,6 +222,22 @@ conflating (`gateway/authz_mixin.py`): Outside multiplexing there is one adapter per platform, so both seams return it. `tests/gateway/test_multiplex_transport_matrix.py` asserts every row. +### Restore, relay, callbacks and thread hops + +The routing entry persists `transport_profile` next to the key (and the +`sessions.transport_profile` column in `state.db`), so after a restart a +revived lane still knows which bot received it: `_restored_source(entry)` +re-pins a `RoutingIdentity` with no live adapter and `_delivery_adapter_for` +delivers through that bot's adapter or fails closed — a satellite routed +through the default bot keeps answering from the default bot, a lane owned by +a secondary never falls back to the default bot's credential. Entries written +before the column existed carry `null` and keep the shared-bot heuristics. +Over the relay, every outbound frame's `metadata.profile` (and `follow_up`'s +key namespace) tells the connector which profile to stamp on the next +`passthrough_forward`, so a button press after a routed slash command stays in +the same profile. Deferred callbacks (`/model` picker) capture the routed home +at command time and the gateway's executor hops copy the ContextVar scope. + ## Control plane Desktop plugins reach the gateway only through the ws JSON-RPC door, so diff --git a/website/docs/developer-guide/relay-connector-contract.md b/website/docs/developer-guide/relay-connector-contract.md index 1a7552de68b36..bc8ffa9fb3987 100644 --- a/website/docs/developer-guide/relay-connector-contract.md +++ b/website/docs/developer-guide/relay-connector-contract.md @@ -423,6 +423,16 @@ The gateway calls the transport with action dicts. Source of truth: `get_chat_info(chat_id)` is a separate proxied call returning at least `{name, type}`. +**`metadata.profile` (multiplex round-trip).** Every chat-addressed outbound +frame's `metadata` carries the tenant discriminators the gateway captured from +the inbound (`scope_id`, `user_id`) and, on a multiplexed gateway, the Hermes +`profile` the connector routed that chat's inbound to; `follow_up` frames carry +the profile encoded in their `session_key` namespace. The connector MUST stamp +the same `profile` on the next `passthrough_forward` / `inbound` for that chat or +interaction, so a Discord button press after a slash command lands in the same +profile's session (`gateway/relay/adapter.py::_with_scope`, `send_follow_up`). +Single-profile gateways never emit the key — frames stay byte-identical. + **`send_media` (Phase 2 media egress).** Media crosses the wire BY REFERENCE: `source_url` is either (a) a **connector re-host** the gateway previously uploaded via `POST {connector}/relay/media` (raw bytes body, `Content-Type` + diff --git a/website/docs/developer-guide/session-storage.md b/website/docs/developer-guide/session-storage.md index 56430819040b5..db42da3f27a6d 100644 --- a/website/docs/developer-guide/session-storage.md +++ b/website/docs/developer-guide/session-storage.md @@ -151,7 +151,8 @@ Abridged — see `SCHEMA_SQL` in `hermes_state_common.py` (applied by `hermes_st (which also includes gateway routing metadata such as `session_key`, `chat_id`, `chat_type`, `thread_id`, `display_name`, `origin_json`, `expiry_finalized`, workspace fields `cwd` / `git_branch` / `git_repo_root`, handoff and -compression-failure fields, `profile_name`, `rewind_count`, `archived`, and +compression-failure fields, `profile_name`, `transport_profile` (the multiplex +bot that received the lane, nullable), `rewind_count`, `archived`, and `pinned`): ```sql