diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 645ba385a3..eb57a7f334 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -218,31 +218,33 @@ class GatewayAuthorizationMixin: return getattr(self, "_profile_adapters", None) or {} def _authorization_adapter(self, platform: Optional[Platform], profile: Optional[str] = None): - """Live adapter whose intake policy gates authorization. - - Secondary-profile adapters live in ``_profile_adapters[profile]``; the primary profile owns - ``self.adapters``. ``_profile_adapters`` is consulted BEFORE the active profile name: multiplex - turns override ``HERMES_HOME`` so ``_active_profile_name()`` reports the secondary profile - mid-turn, and treating it as primary would hand it the default bot. + """Live adapter whose intake policy gates authorization (``_adapters_for_profile`` for the + profile rule). ``None`` when the profile has no adapter for *platform*. """ if not platform: return None + return self._adapters_for_profile(profile).get(platform) + + def _adapters_for_profile(self, profile: Optional[str]) -> dict: + """The live adapter map *profile* may deliver through: ``_profile_adapters[p]`` for a + secondary, ``self.adapters`` only for the primary/default. ``_profile_adapters`` is consulted + BEFORE the active profile name: multiplex turns override ``HERMES_HOME`` so + ``_active_profile_name()`` reports the secondary profile mid-turn, and treating it as primary + would hand it the default bot. A named profile with no map gets ``{}`` — fail closed: a + secondary whose adapter failed to connect must NOT fall back to the default profile's adapter + (replies, tool sends, marker-file notices out the wrong bot).""" profile_name = (profile or "").strip() or None - if profile_name and profile_name != "default": - profile_adapters = self._profile_adapters_map() - if profile_name in profile_adapters: - return profile_adapters[profile_name].get(platform) - # Identity captured at construction, not the per-turn HERMES_HOME-derived name. - primary_profile = getattr(self, "_primary_profile_name", None) - if not primary_profile: - with contextlib.suppress(Exception): - primary_profile = self._active_profile_name() - if profile_name == primary_profile: - return self._primary_adapters().get(platform) - # Fail closed: a secondary profile whose adapter failed to connect must NOT - # fall back to the default profile's adapter (replies out the wrong bot). - return None - return self._primary_adapters().get(platform) + if not profile_name or profile_name == "default": + return self._primary_adapters() + profile_adapters = self._profile_adapters_map() + if profile_name in profile_adapters: + return profile_adapters[profile_name] + # Identity captured at construction, not the per-turn HERMES_HOME-derived name. + primary_profile = getattr(self, "_primary_profile_name", None) + if not primary_profile: + with contextlib.suppress(Exception): + primary_profile = self._active_profile_name() + return self._primary_adapters() if profile_name == primary_profile else {} def _adapter_for_source(self, source: Optional[SessionSource]): """Resolve the live adapter for an inbound ``SessionSource``.""" diff --git a/gateway/run_goals.py b/gateway/run_goals.py index ba7cc3eed4..7e4688b9aa 100644 --- a/gateway/run_goals.py +++ b/gateway/run_goals.py @@ -10,8 +10,8 @@ from __future__ import annotations import asyncio import logging import time -from contextlib import suppress -from typing import TYPE_CHECKING, Any +from contextlib import nullcontext, suppress +from typing import TYPE_CHECKING, Any, Optional from gateway.platforms.event import MessageEvent, MessageType @@ -365,8 +365,11 @@ class GatewayGoalsMixin: if msg and source is not None: await self._defer_goal_status_notice_after_delivery(source, msg) - async def _loop_wakeup_fire_one(self, sid: str, state: Any, now: float, warned_no_route: set) -> None: - """Inject one due /loop wakeup into its session, applying every deferral rule.""" + async def _loop_wakeup_fire_one( + self, sid: str, state: Any, now: float, warned_no_route: set, profile: Optional[str] = None, + ) -> None: + """Inject one due /loop wakeup into its session, applying every deferral rule. ``profile`` is + the store being scanned (None = default); a ``profile`` persisted in the route wins.""" from hermes_cli.loops import LoopManager, goal_blocks_loop_tick if state.awaiting_response or now < state.next_due_at: @@ -376,12 +379,17 @@ class GatewayGoalsMixin: chat_id = route.get("chat_id", "") if not platform_name or not chat_id: return # CLI / TUI-owned loop — their own schedulers drive it. - adapter = next((a for p, a in self.adapters.items() if p.value == platform_name), None) + profile = route.get("profile") or profile + # The loop's OWN profile's adapter map, fail closed: ``self.adapters`` is the default profile's, + # so a secondary session's wakeup would inject via the default bot on a bare chat_id (a + # Telegram DM lands in the user's chat with the other bot). + adapters = self._adapters_for_profile(profile) + adapter = next((a for p, a in adapters.items() if p.value == platform_name), None) if adapter is None: if sid not in warned_no_route: warned_no_route.add(sid) logger.debug( - "loop wakeup: no adapter for platform %r (session %s)", platform_name, sid, + "loop wakeup: no adapter for platform %r (session %s, profile %s)", platform_name, sid, profile, ) return @@ -393,6 +401,8 @@ class GatewayGoalsMixin: }) if source is None: return + if profile and not getattr(source, "profile", None): + source.profile = profile # session key + runtime scope of the injected turn session_key = None with suppress(Exception): session_key = self._session_key_for_source(source) @@ -435,21 +445,37 @@ class GatewayGoalsMixin: """Fire due /loop wakeups for idle gateway sessions: a coarse ticker scans persisted loops (SessionDB ``loop:*`` rows) and injects each due prompt via the synthetic-message path. Deferrals: session running a turn (FIFO would race the live turn); active non-parked /goal - (goal owns the idle boundary); no routing metadata (one-time warning).""" + (goal owns the idle boundary); no routing metadata (one-time warning). + + Multiplex: one gateway-wide task, so ``list_active_loops`` alone reads only the launch home's + store — a ``/loop`` set from a secondary profile's chat would never fire. Every served + profile's store is scanned under its own runtime scope (same shape as ``_handoff_watcher``), + and each hit is fired against that profile's adapters.""" + from gateway.run import _async_profile_runtime_scope, _handoff_watch_scopes await asyncio.sleep(5) # let platforms finish connecting warned_no_route: set = set() + + def _scope(profile_home): + return (_async_profile_runtime_scope(profile_home) if profile_home is not None + else nullcontext()) + + async def _scan_one_store(profile_name: Optional[str]) -> None: + from hermes_cli.loops import list_active_loops + + # Warm once per scan: the scan reads every persisted loop and a cold cache would + # run the state.db init on the loop thread before the first read. + await self._warm_goals_session_db("loop wakeup") + # Off-loop too: the read is lock-free under WAL but convoys on the writer lock without it. + active_loops = await self._run_in_executor_with_context(list_active_loops) + now = time.time() + for sid, state in active_loops: + await self._loop_wakeup_fire_one(sid, state, now, warned_no_route, profile_name) + while self._running: try: - from hermes_cli.loops import list_active_loops - - # Warm once per scan: the scan reads every persisted loop and a cold cache would - # run the state.db init on the loop thread before the first read. - await self._warm_goals_session_db("loop wakeup") - # Off-loop too: the read is lock-free under WAL but convoys on the writer lock without it. - active_loops = await self._run_in_executor_with_context(list_active_loops) - now = time.time() - for sid, state in active_loops: - await self._loop_wakeup_fire_one(sid, state, now, warned_no_route) + for profile_name, profile_home in _handoff_watch_scopes(self): + async with _scope(profile_home): + await _scan_one_store(profile_name) except Exception as exc: logger.debug("loop wakeup watcher error: %s", exc) await asyncio.sleep(interval) diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index 41f8a55ee6..39bde5a82e 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -421,6 +421,19 @@ class GatewayNotificationsMixin: prompt=_hermes_home / ".update_prompt.json", response=_hermes_home / ".update_response", ) + @staticmethod + def _marker_profile(data: dict) -> Optional[str]: + """Owning profile of a persisted restart/update marker: explicit ``profile``, else the + ``agent::`` lane of its ``session_key`` (markers written before ``profile`` was + persisted); ``None`` = default profile.""" + profile = str(data.get("profile") or "").strip() + if profile: + return profile + parts = str(data.get("session_key") or "").split(":") + if len(parts) >= 5 and parts[0] == "agent" and parts[1] not in ("main", ""): + return parts[1] + return None + def _resolve_update_target(self, paths: "_UpdatePaths") -> Optional["_UpdateTarget"]: """Resolve adapter/chat/session for update watcher messages from the pending marker.""" for path in (paths.claimed, paths.pending): @@ -434,7 +447,9 @@ class GatewayNotificationsMixin: if not (platform_str and chat_id): continue # BASE: an incomplete marker falls through to the next path, not "unresolved" platform = Platform(platform_str) - adapter = self.adapters.get(platform) + # The requester's OWN profile bot (marker ``profile``, else the ``agent::`` key + # lane); a bare self.adapters lookup is the default bot under multiplex. + adapter = self._authorization_adapter(platform, self._marker_profile(pending)) if not adapter: return None metadata = self._pending_marker_metadata(platform, chat_id, pending, adapter) @@ -627,7 +642,7 @@ class GatewayNotificationsMixin: exit_code = self._update_exit_code(paths) output = paths.output.read_bytes().decode("utf-8", errors="replace") if paths.output.exists() else "" platform = Platform(platform_str) - adapter = self.adapters.get(platform) + adapter = self._authorization_adapter(platform, self._marker_profile(pending)) if chat_id and not adapter: # Target platform not reconnected yet (common right after the update's restart): keep the # markers for a later retry instead of silently losing the notification. @@ -671,7 +686,10 @@ class GatewayNotificationsMixin: if not platform_str or not chat_id: return None platform = Platform(platform_str) - transport = resolve_delivery_transport(platform, self.config, self.adapters) + # Relay-aware transport over the REQUESTER'S profile adapter map; ``self.adapters`` is the + # default profile's, so a secondary's "restarted" notice would leave through the wrong bot. + transport = resolve_delivery_transport( + platform, self.config, self._adapters_for_profile(self._marker_profile(data))) if transport is None: logger.debug("Restart notification skipped: no live transport for %s", platform_str) return None diff --git a/gateway/run_shutdown.py b/gateway/run_shutdown.py index 812aa3d853..53933399b9 100644 --- a/gateway/run_shutdown.py +++ b/gateway/run_shutdown.py @@ -877,7 +877,9 @@ class GatewayShutdownMixin: return len(notified) async def _shutdown_notification_target(self, session_key: str): - """``(source, platform_str, chat_id, thread_id)``: persisted origin > cached source > parsed key.""" + """``(source, platform_str, chat_id, thread_id, profile)``: persisted origin > cached source > + parsed key. ``profile`` is the owning profile from the source or the ``agent::`` key + namespace (``None`` = default) so the notice leaves through that profile's bot.""" from gateway.run import _parse_session_key source = None try: @@ -890,11 +892,15 @@ class GatewayShutdownMixin: if source is None: source = self._get_cached_session_source(session_key) if source is not None: - return source, source.platform.value, str(source.chat_id), source.thread_id - _parsed = _parse_session_key(session_key) + return source, source.platform.value, str(source.chat_id), source.thread_id, getattr(source, "profile", None) + parts = session_key.split(":") + profile = parts[1] if len(parts) >= 5 and parts[0] == "agent" and parts[1] != "main" else None + # _parse_session_key only understands the ``agent:main:`` lane; a secondary's key is parsed on + # the same shape with its profile carried separately. + _parsed = _parse_session_key(":".join(["agent", "main", *parts[2:]]) if profile else session_key) if not _parsed: return None - return None, _parsed["platform"], _parsed["chat_id"], _parsed.get("thread_id") + return None, _parsed["platform"], _parsed["chat_id"], _parsed.get("thread_id"), profile async def _send_shutdown_notice( self, adapter, chat_id: str, msg: str, kind: str, platform_str: str, **send_kwargs @@ -946,13 +952,18 @@ class GatewayShutdownMixin: target = await self._shutdown_notification_target(session_key) if target is None: continue - source, platform_str, chat_id, thread_id = target + source, platform_str, chat_id, thread_id, profile = target dedup_key = _notice_target_key(platform_str, chat_id, thread_id) if dedup_key in notified: continue try: platform = Platform(platform_str) - adapter = self.adapters.get(platform) + # The session's OWN profile's bot (transport ref → profile map), never a bare + # self.adapters hit: under multiplex that is the default bot, so a secondary session's + # "Gateway shutting down" would land in the user's chat with the wrong bot. + adapter = self._adapter_for_source(source) if source is not None else None + if adapter is None: + adapter = self._authorization_adapter(platform, profile) if not adapter: continue if not self._notice_allowed(platform, "active session"): diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 27d688203c..4a778d3172 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -103,14 +103,17 @@ def _execute(command: str, **ctx_kwargs): def _restart_notify_payload(event: MessageEvent) -> dict: - """Requester routing info so the new gateway process can notify them once back online.""" + """Requester routing info so the new gateway process can notify them once back online. + ``profile`` is persisted so the notice leaves through the requester's own profile bot after the + restart (a bare platform lookup would resolve the default profile's adapter).""" source = event.source data = {"platform": source.platform.value if source.platform else None, "chat_id": source.chat_id, "chat_type": source.chat_type} if source.delivered_via_upstream_relay is True: data["delivered_via_upstream_relay"] = True data.update({k: getattr(source, k) for k in ("user_id", "scope_id") if getattr(source, k)}) - optional = (("thread_id", source.thread_id), ("message_id", event.message_id)) + optional = (("thread_id", source.thread_id), ("message_id", event.message_id), + ("profile", getattr(source, "profile", None))) data.update({k: v for k, v in optional if v}) return data @@ -208,10 +211,11 @@ class GatewaySlashCommandsMixin( return self._thread_metadata_for_source(event.source, self._reply_anchor_for_event(event)) def _adapter_and_key_for(self, event: MessageEvent): - """``(adapter, session_key)`` for the event's source, either None when no source.""" + """``(adapter, session_key)`` for the event's source, either None when no source. The source's + OWN transport (profile-aware, fail-closed) — ``self.adapters`` is the default profile's map.""" if not event.source: return None, None - return self.adapters.get(event.source.platform), self._session_key_for_source(event.source) + return self._adapter_for_source(event.source), self._session_key_for_source(event.source) def _telegramized_command_reply(self, event: MessageEvent, text: str) -> str: from gateway.run import _telegramize_command_mentions @@ -255,7 +259,7 @@ class GatewaySlashCommandsMixin( (WeCom msgtype:"stream"), which need it sent directly with control-lane metadata (reliable proactive send, not the finalized reply stream). ``is not True``: mocks auto-create attrs.""" source = event.source - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) # the receiving bot, not the default profile's if adapter: adapter.resume_typing_for_chat(source.chat_id) # agent is about to continue if getattr(adapter, "SUPPORTS_NATIVE_STREAMING", False) is not True: @@ -1234,7 +1238,10 @@ class GatewaySlashCommandsMixin( "platform": src.platform.value, "chat_id": src.chat_id, "chat_type": src.chat_type, "user_id": src.user_id, "session_key": self._session_key_for_source(src), "timestamp": datetime.now().isoformat()} - pending.update({k: v for k, v in (("thread_id", src.thread_id), ("message_id", event.message_id)) if v}) + # ``profile``: the update watcher (possibly the NEXT gateway process) must answer through the + # requester's own profile bot, not the default profile's adapter for the same platform. + pending.update({k: v for k, v in (("thread_id", src.thread_id), ("message_id", event.message_id), + ("profile", getattr(src, "profile", None))) if v}) _tmp_pending = pending_path.with_suffix(".tmp") _tmp_pending.write_text(json.dumps(pending), encoding="utf-8") _tmp_pending.replace(pending_path) diff --git a/gateway/slash_commands_goals.py b/gateway/slash_commands_goals.py index 8b74e7a4cb..422beb0d4f 100644 --- a/gateway/slash_commands_goals.py +++ b/gateway/slash_commands_goals.py @@ -280,14 +280,15 @@ class GatewayGoalCommandsMixin: mgr = LoopManager(session_id=sid) # New loops capture the event's routing so the idle loop-wakeup watcher can inject ticks - # here after a restart; best-effort, empty fields dropped. + # here after a restart; best-effort, empty fields dropped. ``profile`` pins the wakeup to this + # session's own bot under multiplex (the watcher must never fire it through the default bot). route: dict = {} try: src = event.source if src is not None: platform = getattr(src, "platform", "") route = {"platform": platform.value if hasattr(platform, "value") else str(platform or "")} - for key in ("chat_id", "chat_type", "thread_id", "user_id", "user_name"): + for key in ("chat_id", "chat_type", "thread_id", "user_id", "user_name", "profile"): route[key] = str(getattr(src, key, "") or "") route = {k: v for k, v in route.items() if v} except Exception: diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 32d754f8bd..4521301cd1 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -3916,20 +3916,45 @@ class DiscordAdapter(DiscordMediaMixin, BasePlatformAdapter): logger.debug("[Discord] Could not schedule admin notify task: %s", e) return False + @staticmethod + async def _alert_adapters_and_config(runner, profile): + """``(adapter_map, gateway_config)`` of the profile owning this adapter. Default/primary: the + runner's own. Secondary: ``_profile_adapters[profile]`` and the config loaded under that + profile's runtime scope (its ``home_channel`` entries live in ITS config.yaml).""" + if not profile: + return runner.adapters, runner.config + adapters = runner._adapters_for_profile(profile) + if adapters is runner.adapters: # profile IS the primary + return adapters, runner.config + from gateway.config import load_gateway_config + from gateway.run import _async_profile_runtime_scope + from hermes_cli.profiles import get_profile_dir + async with _async_profile_runtime_scope(get_profile_dir(profile)): + return adapters, load_gateway_config() + async def _notify_unauthorized_slash( self, user_name: str, user_id: str, chan_id, guild_id, command_text: str, reason: str, ) -> None: """Best-effort operator alert: TELEGRAM first, then SLACK; no-op without a home channel. - A soft failure (``SendResult(success=False)``, e.g. rate-limit) continues the fallback chain.""" + A soft failure (``SendResult(success=False)``, e.g. rate-limit) continues the fallback chain. + Under multiplex the alert stays inside THIS adapter's profile: its own adapter map (fail closed + when the profile has no Telegram/Slack bot) and its own home channels — never the default + profile's bot or channel, which is what a bare ``runner.adapters`` lookup resolves.""" runner = getattr(self, "gateway_runner", None) if not runner: return + profile = getattr(self, "_owner_profile", None) + try: + adapters, config = await self._alert_adapters_and_config(runner, profile) + except Exception as e: + logger.debug("[Discord] Admin notify: profile %r resolution failed: %s", profile, e) + return for target in (Platform.TELEGRAM, Platform.SLACK): try: - adapter = runner.adapters.get(target) + adapter = adapters.get(target) if not adapter: continue - home = runner.config.get_home_channel(target) + home = config.get_home_channel(target) if not home or not getattr(home, "chat_id", None): continue msg = ( diff --git a/plugins/platforms/wecom/adapter.py b/plugins/platforms/wecom/adapter.py index afe198b738..772d433e6b 100644 --- a/plugins/platforms/wecom/adapter.py +++ b/plugins/platforms/wecom/adapter.py @@ -754,11 +754,12 @@ async def _send_via(adapter, chat_id, message, *, live: bool): async def _standalone_send(pconfig, chat_id, message, *, thread_id=None, media_files=None, force_document=False): """Reuse the live gateway adapter in-process, else connect ephemerally (WeCom allows ONE - WebSocket per bot — a second connection kicks the first).""" + WebSocket per bot — a second connection kicks the first). The live adapter is the ACTIVE + PROFILE's (``_live_adapter``): a bare ``runner.adapters`` hit is the default profile's bot under + multiplex, so a secondary profile's send would leave with the wrong identity.""" try: - from gateway.run import _gateway_runner_ref - runner = _gateway_runner_ref() - adapter = runner.adapters.get(Platform.WECOM) if runner is not None else None + from tools.send_message_senders import _live_adapter + _, adapter = _live_adapter(Platform.WECOM) except Exception: adapter = None if adapter is not None: diff --git a/tests/gateway/test_multiplex_notice_egress_profile_adapter.py b/tests/gateway/test_multiplex_notice_egress_profile_adapter.py new file mode 100644 index 0000000000..40a93f1efd --- /dev/null +++ b/tests/gateway/test_multiplex_notice_egress_profile_adapter.py @@ -0,0 +1,104 @@ +"""Out-of-band gateway egress for a SECONDARY profile session leaves through that profile's bot. + +Shutdown/restart/update notices and ``/loop`` wakeups resolved their adapter by bare platform from +``runner.adapters`` — the default profile's map — so a secondary session's notice landed in the +user's chat with the wrong bot (and for ``/loop`` the wakeup ran the default profile's turn). They +must use the session's own profile adapter and fail closed when that profile has none. +""" +import json +from types import SimpleNamespace + +import pytest + +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.platforms.event import MessageEvent, MessageType +from gateway.run import GatewayRunner +from gateway.session import SessionSource +from gateway.slash_commands import _restart_notify_payload + + +class _Adapter: + def __init__(self): + self.sent, self.handled = [], [] + + async def send(self, chat_id, content=None, metadata=None, **kw): + self.sent.append(chat_id) + return SimpleNamespace(success=True, message_id="m", error=None) + + async def handle_message(self, event): + self.handled.append(event.text) + + +def _runner(): + r = object.__new__(GatewayRunner) + r.config = GatewayConfig(platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="t")}) + r.adapters = {Platform.TELEGRAM: _Adapter()} + r._profile_adapters = {"sec": {Platform.TELEGRAM: _Adapter()}, "nobot": {}} + r._primary_profile_name = "default" + r.session_store = None + r._session_sources = None + r._running_agents = {} + r._restart_requested = False + r._restart_command_source = None + return r + + +@pytest.mark.asyncio +async def test_shutdown_notice_for_secondary_session_uses_its_own_bot(): + r = _runner() + r._running_agents["agent:sec:telegram:dm:42"] = object() + r._running_agents["agent:nobot:telegram:dm:43"] = object() + r._snapshot_running_agents = lambda: list(r._running_agents) + + await r._notify_active_sessions_of_shutdown() + + assert r._profile_adapters["sec"][Platform.TELEGRAM].sent == ["42"] + # The default bot never speaks for a secondary session — not even one whose bot is down. + assert r.adapters[Platform.TELEGRAM].sent == [] + + +@pytest.mark.asyncio +async def test_restart_marker_from_secondary_session_notifies_via_its_own_bot(tmp_path, monkeypatch): + import gateway.run as gateway_run + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + r = _runner() + src = SessionSource(platform=Platform.TELEGRAM, chat_id="42", chat_type="dm", user_id="42", profile="sec") + event = MessageEvent(text="/restart", message_type=MessageType.TEXT, source=src, message_id="7") + (tmp_path / ".restart_notify.json").write_text(json.dumps(_restart_notify_payload(event))) + + assert await r._send_restart_notification() == ("telegram", "42", None) + assert r._profile_adapters["sec"][Platform.TELEGRAM].sent == ["42"] + assert r.adapters[Platform.TELEGRAM].sent == [] + + +@pytest.mark.asyncio +async def test_loop_wakeup_from_secondary_route_fires_through_its_own_bot(monkeypatch): + from hermes_cli import loops + + class _Mgr: + state = SimpleNamespace(ticks_fired=1) + + def __init__(self, session_id=None): + pass + + def is_due(self, now): + return True + + def fire_tick(self): + return "tick" + + def complete_tick(self, *_): + return {} + + monkeypatch.setattr(loops, "LoopManager", _Mgr) + monkeypatch.setattr(loops, "goal_blocks_loop_tick", lambda sid: False) + r = _runner() + r._running = True + route = {"platform": "telegram", "chat_id": "42", "chat_type": "dm", "user_id": "42", "profile": "sec"} + state = SimpleNamespace(awaiting_response=False, next_due_at=0, route=route) + + await r._loop_wakeup_fire_one("sid", state, 1e12, set()) + await r._loop_wakeup_fire_one("sid2", SimpleNamespace(**{**vars(state), "route": {**route, "profile": "nobot"}}), 1e12, set()) + + assert r._profile_adapters["sec"][Platform.TELEGRAM].handled == ["tick"] + assert r.adapters[Platform.TELEGRAM].handled == [] diff --git a/tests/tools/test_send_message_multiplex_profile_adapter.py b/tests/tools/test_send_message_multiplex_profile_adapter.py new file mode 100644 index 0000000000..6bbcb3909d --- /dev/null +++ b/tests/tools/test_send_message_multiplex_profile_adapter.py @@ -0,0 +1,47 @@ +"""send_message must deliver through the ACTIVE PROFILE's gateway adapter under multiplex. + +``runner.adapters`` holds the default profile's bots; a secondary profile's turn that resolved the +live adapter by bare platform posted (and reacted) with the default bot's identity. The lookup must +honour ``_profile_adapters[profile]`` and fail closed (``None`` → scoped standalone sender / error) +when the profile has no adapter for that platform — never the default bot. +""" +from pathlib import Path + +import pytest + +from gateway.config import Platform +from gateway.run import GatewayRunner, _profile_runtime_scope +from tools.send_message_senders import _live_adapter + + +@pytest.fixture +def mux_runner(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + (home / "profiles" / "sec").mkdir(parents=True) + (home / "profiles" / "nobot").mkdir(parents=True) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + monkeypatch.setenv("HERMES_HOME", str(home)) + default_slack, sec_slack = object(), object() + runner = object.__new__(GatewayRunner) + runner.adapters = {Platform.SLACK: default_slack} + runner._profile_adapters = {"sec": {Platform.SLACK: sec_slack}, "nobot": {}} + runner._primary_profile_name = "default" + import gateway.run as gateway_run + monkeypatch.setattr(gateway_run, "_gateway_runner_ref", lambda: runner) + return home, default_slack, sec_slack + + +def test_secondary_profile_turn_resolves_its_own_adapter(mux_runner): + home, default_slack, sec_slack = mux_runner + with _profile_runtime_scope(home / "profiles" / "sec", {}): + _, adapter = _live_adapter(Platform.SLACK) + assert adapter is sec_slack + _, adapter = _live_adapter(Platform.SLACK) # default scope still gets the default bot + assert adapter is default_slack + + +def test_profile_without_adapter_fails_closed_never_default_bot(mux_runner): + home, default_slack, _ = mux_runner + with _profile_runtime_scope(home / "profiles" / "nobot", {}): + _, adapter = _live_adapter(Platform.SLACK) + assert adapter is None diff --git a/tools/send_message_senders.py b/tools/send_message_senders.py index 166eec23c0..1e1e353017 100644 --- a/tools/send_message_senders.py +++ b/tools/send_message_senders.py @@ -292,7 +292,13 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No def _live_adapter(platform, *, lookup_failed_warning=None): """``(runner, adapter)`` for the in-process gateway; ``(None, None)`` standalone (cron); ``(runner, None)`` when the lookup fails — logged when a warning is given, never silently - swallowed (a silent fall-through could recreate a reconnect storm).""" + swallowed (a silent fall-through could recreate a reconnect storm). + + Multiplex: the adapter is the ACTIVE PROFILE's (``_profile_adapters[profile]``), never a bare + ``runner.adapters`` hit — that map holds the default profile's bots, so a secondary profile's turn + would post/react with the default bot's identity. A profile with no adapter for the platform + yields ``None`` (fail closed → the caller's scoped standalone sender or an error), never the + default bot. Same resolver shape as ``hermes_cli/platform_actions.py::_resolve_adapter``.""" try: from gateway.run import _gateway_runner_ref runner = _gateway_runner_ref() @@ -301,7 +307,11 @@ def _live_adapter(platform, *, lookup_failed_warning=None): if runner is None: return None, None try: - return runner, runner.adapters.get(platform) + resolve = getattr(runner, "_authorization_adapter", None) + if not callable(resolve): # bare runner stubs without the authz mixin + return runner, runner.adapters.get(platform) + from hermes_cli.profiles import get_active_profile_name + return runner, resolve(platform, get_active_profile_name()) except Exception: if lookup_failed_warning: logger.warning(lookup_failed_warning, exc_info=True) diff --git a/website/docs/user-guide/multi-profile-gateways.md b/website/docs/user-guide/multi-profile-gateways.md index 79636a2173..4eb5adc975 100644 --- a/website/docs/user-guide/multi-profile-gateways.md +++ b/website/docs/user-guide/multi-profile-gateways.md @@ -240,6 +240,14 @@ secrets or chat history to a reply. Kanban, profile-scoped skills/memory/SOUL, and model routing all behave per-profile exactly as they do with separate gateways. +Outbound identity is per profile too. A turn running for profile `P` that calls +the `send_message` tool (send, react, media) posts through `P`'s own bot; +so do the "Gateway shutting down/restarted" and `/update` notices for `P`'s +sessions, `/loop` wakeups set from `P`'s chats, and the Discord +unauthorized-slash operator alert of `P`'s Discord bot (to `P`'s home +channel). If `P` has no connected bot for that platform the send fails with a +clear error — it never falls back to the default profile's bot. + ### Serving selected profiles By default, `gateway.multiplex_profiles: true` serves every valid named profile