From 45a408f41adc677194af146df1ec207e5a1067ce Mon Sep 17 00:00:00 2001 From: Victor Kyriazakos Date: Thu, 23 Jul 2026 14:10:02 +0000 Subject: [PATCH] fix(gateway): deliver relay-backed homes after restart --- cron/scheduler.py | 61 +++++-- gateway/config.py | 42 ++++- gateway/delivery.py | 103 +++++++++++- gateway/relay/adapter.py | 74 ++++++++- gateway/run.py | 60 ++++--- gateway/slash_commands.py | 81 ++++++--- tests/cron/test_scheduler.py | 122 ++++++++++++++ tests/gateway/test_config.py | 59 ++++++- tests/gateway/test_delivery.py | 183 ++++++++++++++++++++- tests/gateway/test_restart_notification.py | 170 ++++++++++++++++++- 10 files changed, 878 insertions(+), 77 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index 7b00bfcb0f3..c309c59c4e5 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1586,16 +1586,26 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option delivery_errors.append(msg) continue - pconfig = config.platforms.get(platform) + from gateway.delivery import resolve_delivery_transport + + transport = resolve_delivery_transport(platform, config, adapters) + if transport is not None: + pconfig = transport.config + runtime_adapter = transport.adapter + else: + # No live transport: preserve the existing standalone delivery path, + # which uses the logical platform's configured credential. + pconfig = config.platforms.get(platform) + runtime_adapter = None + if not pconfig or not pconfig.enabled: msg = f"platform '{platform_name}' not configured/enabled" logger.warning("Job '%s': %s", job["id"], msg) delivery_errors.append(msg) continue - # Prefer the live adapter when the gateway is running — this supports E2EE - # rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt. - runtime_adapter = (adapters or {}).get(platform) + # Prefer the resolved live transport when the gateway is running. This + # supports E2EE native adapters and relay-fronted logical platforms. # The live-send path (which SEEDS the flat in_channel continuation # session via _seed_cron_channel_session) needs not just a live adapter # but a running event loop to schedule the async send onto. Compute that @@ -1900,10 +1910,13 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option f"live adapter send to {platform_name}:{chat_id} " f"returned unconfirmed result ({shape}, error={err})" ) - logger.warning( - "Job '%s': %s, falling back to standalone", - job["id"], msg, - ) + if transport is not None and transport.is_relay: + logger.warning("Job '%s': %s", job["id"], msg) + else: + logger.warning( + "Job '%s': %s, falling back to standalone", + job["id"], msg, + ) target_errors.append(msg) adapter_ok = False # fall through to standalone path elif ( @@ -1929,11 +1942,20 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # skipped attachments so the drop is visible rather than silently # lost. if adapter_ok and not timed_out and media_files: + routed_media_metadata = dict(media_metadata or {}) + if transport is not None and transport.is_relay: + routed_media_metadata["_relay_logical_platform"] = platform.value + logical_home = config.get_home_channel(platform) + if logical_home is not None and logical_home.chat_id == chat_id: + if logical_home.user_id: + routed_media_metadata["user_id"] = logical_home.user_id + if logical_home.scope_id: + routed_media_metadata["scope_id"] = logical_home.scope_id _send_media_via_adapter( runtime_adapter, chat_id, media_files, - media_metadata, + routed_media_metadata or None, loop, job, platform=platform, @@ -1978,12 +2000,25 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option err_msg = f"live adapter delivery to {platform_name}:{chat_id} failed: {e}" if not any(err_msg in err for err in target_errors): target_errors.append(err_msg) - logger.warning( - "Job '%s': %s, falling back to standalone", - job["id"], err_msg, - ) + if transport is not None and transport.is_relay: + logger.warning("Job '%s': %s", job["id"], err_msg) + else: + logger.warning( + "Job '%s': %s, falling back to standalone", + job["id"], err_msg, + ) if not delivered: + if transport is not None and transport.is_relay: + # Relay owns the logical destination and its connector owns the + # platform credential. A native retry could duplicate delivery + # and cannot be authenticated correctly, so fail closed. + if not target_errors: + target_errors.append( + f"relay delivery to {platform_name}:{chat_id} failed" + ) + delivery_errors.extend(target_errors) + continue # If the interpreter is finalizing (gateway SIGTERM / restart / # OOM), scheduling any new delivery is futile — asyncio.run and a # fresh ThreadPoolExecutor both raise "cannot schedule new futures diff --git a/gateway/config.py b/gateway/config.py index 01c906cbd2e..6e72e2942a3 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -431,6 +431,11 @@ class HomeChannel: chat_id: str name: str # Human-readable name for display thread_id: Optional[str] = None + # Authenticated logical-target provenance observed by a platform adapter. + # Relay egress re-attaches these values, but the connector remains the + # authorization boundary and resolves them against its authoritative stores. + user_id: Optional[str] = None + scope_id: Optional[str] = None def to_dict(self) -> Dict[str, Any]: result = { @@ -440,6 +445,10 @@ class HomeChannel: } if self.thread_id: result["thread_id"] = self.thread_id + if self.user_id: + result["user_id"] = self.user_id + if self.scope_id: + result["scope_id"] = self.scope_id return result @classmethod @@ -449,9 +458,30 @@ class HomeChannel: chat_id=str(data["chat_id"]), name=data.get("name", "Home"), thread_id=str(data["thread_id"]) if data.get("thread_id") else None, + user_id=str(data["user_id"]) if data.get("user_id") else None, + scope_id=str(data["scope_id"]) if data.get("scope_id") else None, ) +def persist_home_channel(home: HomeChannel, *, enabled_if_new: bool = False) -> None: + """Persist a logical home without falsely enabling a Relay-fronted adapter.""" + from hermes_cli.config import load_config, save_config + + config = load_config() + platforms = config.setdefault("platforms", {}) + if not isinstance(platforms, dict): + platforms = {} + config["platforms"] = platforms + platform_config = platforms.setdefault(home.platform.value, {}) + if not isinstance(platform_config, dict): + platform_config = {} + platforms[home.platform.value] = platform_config + if enabled_if_new: + platform_config.setdefault("enabled", True) + platform_config["home_channel"] = home.to_dict() + save_config(config) + + @dataclass class SessionResetPolicy: """ @@ -1942,12 +1972,20 @@ def _apply_env_overrides(config: GatewayConfig) -> None: # send Slack messages can use it without activating the gateway adapter. config.platforms[Platform.SLACK].token = slack_token slack_home = getenv("SLACK_HOME_CHANNEL") - if slack_home and Platform.SLACK in config.platforms: - config.platforms[Platform.SLACK].home_channel = HomeChannel( + if slack_home: + slack_config = config.platforms.setdefault( + Platform.SLACK, + PlatformConfig(enabled=False), + ) + existing_home = slack_config.home_channel + same_home = existing_home is not None and existing_home.chat_id == slack_home + slack_config.home_channel = HomeChannel( platform=Platform.SLACK, chat_id=slack_home, name=getenv("SLACK_HOME_CHANNEL_NAME", ""), thread_id=getenv("SLACK_HOME_CHANNEL_THREAD_ID") or None, + user_id=existing_home.user_id if existing_home and same_home else None, + scope_id=existing_home.scope_id if existing_home and same_home else None, ) # Signal diff --git a/gateway/delivery.py b/gateway/delivery.py index 77b245d291c..3672a7c49e3 100644 --- a/gateway/delivery.py +++ b/gateway/delivery.py @@ -54,11 +54,83 @@ def _is_silence_narration(content: Optional[str]) -> bool: return False return bool(_SILENCE_NARRATION.match(stripped)) -from .config import Platform, GatewayConfig +from .config import Platform, GatewayConfig, PlatformConfig from .session import SessionSource from .dead_targets import DeadTargetRegistry +@dataclass(frozen=True) +class DeliveryTransport: + """Resolved live transport for one logical delivery platform.""" + + adapter: Any + config: Optional[PlatformConfig] + transport_platform: Platform + + @property + def is_relay(self) -> bool: + return self.transport_platform == Platform.RELAY + + async def send( + self, + logical_platform: Platform, + chat_id: str, + content: str, + metadata: Optional[Dict[str, Any]], + ) -> Any: + """Send through this transport while preserving the logical platform.""" + if self.is_relay: + return await self.adapter.send_for_platform( + logical_platform, + chat_id, + content, + metadata=metadata, + ) + return await self.adapter.send(chat_id, content, metadata=metadata) + + +def resolve_delivery_transport( + platform: Platform, + config: GatewayConfig, + adapters: Optional[Dict[Platform, Any]], +) -> Optional[DeliveryTransport]: + """Resolve a logical platform to its live delivery transport. + + A concrete native adapter always wins. Relay is eligible only when its + authenticated transport explicitly advertises that it fronts the logical + platform, which keeps restart-time delivery independent of per-chat caches + without letting Relay hijack unrelated platform targets. + """ + live_adapters = adapters or {} + native = live_adapters.get(platform) + native_config = config.platforms.get(platform) + # Preserve DeliveryRouter's historical support for explicitly supplied live + # adapters with no config block, but never let an explicitly disabled native + # adapter shadow an enabled Relay transport. + if native is not None and (native_config is None or native_config.enabled): + return DeliveryTransport( + adapter=native, + config=native_config, + transport_platform=platform, + ) + + relay = live_adapters.get(Platform.RELAY) + relay_config = config.platforms.get(Platform.RELAY) + fronts_platform = getattr(relay, "fronts_platform", None) + if ( + relay is not None + and (relay_config is None or relay_config.enabled) + and callable(fronts_platform) + and fronts_platform(platform) + ): + return DeliveryTransport( + adapter=relay, + config=relay_config, + transport_platform=Platform.RELAY, + ) + return None + + def looks_like_telegram_private_chat_id(chat_id: Optional[str]) -> bool: """True when ``chat_id`` is a positive int — Telegram's private-chat shape. @@ -392,11 +464,11 @@ class DeliveryRouter: metadata: Optional[Dict[str, Any]] ) -> Dict[str, Any]: """Deliver content to a messaging platform.""" - adapter = self.adapters.get(target.platform) - - if not adapter: + transport = resolve_delivery_transport(target.platform, self.config, self.adapters) + if transport is None: raise ValueError(f"No adapter configured for {target.platform.value}") - + adapter = transport.adapter + if not target.chat_id: raise ValueError(f"No chat ID for {target.platform.value} delivery") @@ -472,6 +544,13 @@ class DeliveryRouter: } send_metadata = dict(metadata or {}) + if transport.is_relay: + home = self.config.get_home_channel(target.platform) + if home is not None and home.chat_id == target.chat_id: + if home.user_id: + send_metadata["user_id"] = home.user_id + if home.scope_id: + send_metadata["scope_id"] = home.scope_id is_named_telegram_private_topic = False named_telegram_private_topic_name: Optional[str] = None if target.thread_id: @@ -524,7 +603,12 @@ class DeliveryRouter: send_metadata["telegram_dm_topic_reply_fallback"] = True elif "thread_id" not in send_metadata and "message_thread_id" not in send_metadata and not has_explicit_direct_topic: send_metadata["thread_id"] = target_thread_id - result = await adapter.send(target.chat_id, content, metadata=send_metadata or None) + result = await transport.send( + target.platform, + target.chat_id, + content, + metadata=send_metadata or None, + ) if _send_result_failed(result): if ( is_named_telegram_private_topic @@ -547,7 +631,12 @@ class DeliveryRouter: ) send_metadata["thread_id"] = str(refreshed_thread_id) send_metadata["telegram_dm_topic_created_for_send"] = True - result = await adapter.send(target.chat_id, content, metadata=send_metadata or None) + result = await transport.send( + target.platform, + target.chat_id, + content, + metadata=send_metadata or None, + ) if _send_result_failed(result): raise RuntimeError(_send_result_error(result) or f"{target.platform.value} delivery failed") return result diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index e743f26b290..5937eb71bc1 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -319,16 +319,24 @@ class RelayAdapter(BasePlatformAdapter): meta["user_id"] = author return meta - def _platform_is_fronted(self, platform: str) -> bool: - """Whether ``platform`` is one of the platforms this gateway fronts over - the relay (Phase 1.5). Reads the transport's advertised identity set; used - to decide whether a follow-up's platform-prefixed `kind` names a real - fronted platform worth tagging on the frame (vs. leaving egress to the - session default). Safe when the transport is absent or single-identity.""" + def fronts_platform(self, platform: Any) -> bool: + """Whether the authenticated relay transport advertises ``platform``. + + This is the restart-safe delivery ownership signal: it comes from the + configured identity set sent during handshake, not from an inbound + chat cache learned only after a user sends another message. + """ + platform_value = getattr(platform, "value", platform) + if not platform_value: + return False ids = getattr(self._transport, "_identities", None) if not ids: return False - return any(p == platform for p, _ in ids) + return any(p == str(platform_value) for p, _ in ids) + + def _platform_is_fronted(self, platform: str) -> bool: + """Backward-compatible internal alias for follow-up routing.""" + return self.fronts_platform(platform) async def on_interrupt(self, session_key: str, chat_id: str) -> None: """Bridge a connector-delivered /stop into the adapter's interrupt path. @@ -488,13 +496,27 @@ class RelayAdapter(BasePlatformAdapter): logger.debug("relay go_dormant failed", exc_info=True) return False - async def send( + async def send_for_platform( self, + logical_platform: Any, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: + """Send to an explicitly advertised logical platform over Relay. + + Scheduled and persisted-home deliveries have no fresh inbound event to + populate ``_platform_by_chat``. The shared delivery resolver calls this + method only after ``fronts_platform`` succeeds, and this method repeats + that check fail-closed before stamping the outbound frame. + """ + platform_value = getattr(logical_platform, "value", logical_platform) + if not self.fronts_platform(platform_value): + return SendResult( + success=False, + error=f"relay does not front platform {platform_value}", + ) if self._transport is None: return SendResult(success=False, error="no transport") result = await self._transport.send_outbound( @@ -505,6 +527,42 @@ class RelayAdapter(BasePlatformAdapter): "reply_to": reply_to, "metadata": self._with_scope(chat_id, metadata), }, + platform=str(platform_value), + ) + return SendResult( + success=bool(result.get("success")), + message_id=result.get("message_id"), + error=result.get("error"), + raw_response=result, + ) + + async def send( + self, + chat_id: str, + content: str, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + send_metadata = dict(metadata or {}) + explicit_platform = send_metadata.pop("_relay_logical_platform", None) + if explicit_platform: + return await self.send_for_platform( + explicit_platform, + chat_id, + content, + reply_to=reply_to, + metadata=send_metadata or None, + ) + if self._transport is None: + return SendResult(success=False, error="no transport") + result = await self._transport.send_outbound( + { + "op": "send", + "chat_id": chat_id, + "content": content, + "reply_to": reply_to, + "metadata": self._with_scope(chat_id, send_metadata), + }, platform=self._platform_by_chat.get(str(chat_id)), ) return SendResult( diff --git a/gateway/run.py b/gateway/run.py index efe6b2a0b08..3ef88007653 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2108,7 +2108,11 @@ from gateway.session import ( is_shared_multi_user_session, neutralize_untrusted_inline_text, ) -from gateway.delivery import DeliveryRouter, looks_like_telegram_private_chat_id +from gateway.delivery import ( + DeliveryRouter, + looks_like_telegram_private_chat_id, + resolve_delivery_transport, +) from gateway.turn_lease import SessionTurnLeaseRegistry from gateway.authz_mixin import GatewayAuthorizationMixin from gateway.kanban_watchers import GatewayKanbanWatchersMixin @@ -17031,10 +17035,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew return None platform = Platform(platform_str) - adapter = self.adapters.get(platform) - if not adapter: + transport = resolve_delivery_transport(platform, self.config, self.adapters) + if transport is None: logger.debug( - "Restart notification skipped: %s adapter not connected", + "Restart notification skipped: no live transport for %s", platform_str, ) return None @@ -17053,9 +17057,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew thread_id, chat_type=chat_type, reply_to_message_id=message_id, - adapter=adapter, + adapter=transport.adapter, ) - result = await adapter.send( + if data.get("delivered_via_upstream_relay") is True: + metadata = dict(metadata or {}) + if data.get("user_id"): + metadata["user_id"] = str(data["user_id"]) + if data.get("scope_id"): + metadata["scope_id"] = str(data["scope_id"]) + result = await transport.send( + platform, str(chat_id), "♻ Gateway restarted successfully. Your session continues.", metadata=_non_conversational_metadata(metadata, platform=platform), @@ -17100,13 +17111,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew skipped = skip_targets or set() message = "♻️ Gateway online — Hermes is back and ready." - for platform, adapter in self.adapters.items(): - home = self.config.get_home_channel(platform) + for platform, platform_cfg in self.config.platforms.items(): + home = platform_cfg.home_channel if not home or not home.chat_id: continue - platform_cfg = self.config.platforms.get(platform) - if platform_cfg is not None and not platform_cfg.gateway_restart_notification: + transport = resolve_delivery_transport(platform, self.config, self.adapters) + if transport is None: + continue + + if not platform_cfg.gateway_restart_notification: logger.info( "Home-channel startup notification suppressed: %s has gateway_restart_notification=false", platform.value, @@ -17122,24 +17136,24 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew platform, home.chat_id, home.thread_id, - adapter=adapter, + adapter=transport.adapter, ) - if metadata: - result = await adapter.send( + if transport.is_relay: + metadata = dict(metadata or {}) + if home.user_id: + metadata["user_id"] = home.user_id + if home.scope_id: + metadata["scope_id"] = home.scope_id + send_metadata = _non_conversational_metadata(metadata, platform=platform) + if send_metadata is not None or transport.is_relay: + result = await transport.send( + platform, str(home.chat_id), message, - metadata=_non_conversational_metadata(metadata, platform=platform), + metadata=send_metadata, ) else: - _startup_meta = _non_conversational_metadata(platform=platform) - if _startup_meta: - result = await adapter.send( - str(home.chat_id), - message, - metadata=_startup_meta, - ) - else: - result = await adapter.send(str(home.chat_id), message) + result = await transport.adapter.send(str(home.chat_id), message) if result is not None and getattr(result, "success", True) is False: logger.warning( "Home-channel startup notification failed for %s:%s: %s", diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 9b6f989ff8b..e70cc25d72d 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -32,7 +32,7 @@ from typing import Any, Optional, Union from agent.account_usage import fetch_account_usage, render_account_usage_lines from agent.i18n import t from agent.turn_context import extract_api_content_sidecar -from gateway.config import HomeChannel, Platform, PlatformConfig +from gateway.config import HomeChannel, Platform, PlatformConfig, persist_home_channel from gateway.platforms.base import EphemeralReply, MessageEvent, MessageType from gateway.session import ( AsyncSessionStore, @@ -1280,6 +1280,12 @@ class GatewaySlashCommandsMixin: "chat_id": event.source.chat_id, "chat_type": event.source.chat_type, } + if event.source.delivered_via_upstream_relay is True: + notify_data["delivered_via_upstream_relay"] = True + if event.source.user_id: + notify_data["user_id"] = event.source.user_id + if event.source.scope_id: + notify_data["scope_id"] = event.source.scope_id if event.source.thread_id: notify_data["thread_id"] = event.source.thread_id if event.message_id: @@ -2601,34 +2607,67 @@ class GatewaySlashCommandsMixin: platform_name = source.platform.value if source.platform else "unknown" chat_id = source.chat_id chat_name = source.chat_name or chat_id + if source.platform is None: + return t("gateway.set_home.save_failed", error="Missing logical platform") + + via_relay = getattr(source, "delivered_via_upstream_relay", False) is True + if via_relay: + adapter_for_source = getattr(self, "_adapter_for_source", None) + relay_adapter = adapter_for_source(source) if callable(adapter_for_source) else None + fronts_platform = getattr(relay_adapter, "fronts_platform", None) + if ( + source.platform in {None, Platform.LOCAL, Platform.RELAY} + or not getattr(source, "user_id", None) + or not callable(fronts_platform) + or not fronts_platform(source.platform) + ): + return t( + "gateway.set_home.save_failed", + error="Relay does not authenticate this logical home target", + ) - env_key = _home_target_env_var(platform_name) - thread_env_key = _home_thread_env_var(platform_name) thread_id = source.thread_id + home = HomeChannel( + platform=source.platform, + chat_id=str(chat_id), + name=chat_name, + thread_id=str(thread_id) if thread_id else None, + user_id=( + str(source.user_id) + if getattr(source, "user_id", None) + else None + ), + scope_id=( + str(source.scope_id) + if getattr(source, "scope_id", None) + else None + ), + ) - # Save to .env so it persists across restarts + # config.yaml is canonical because it can persist the authenticated + # logical-target provenance required by Relay after a restart. try: - from hermes_cli.config import save_env_value - save_env_value(env_key, str(chat_id)) - # Keep thread/topic routing explicit and clear stale values when - # /sethome is run from the parent chat instead of a thread. - save_env_value(thread_env_key, str(thread_id or "")) + persist_home_channel(home, enabled_if_new=not via_relay) except Exception as e: return t("gateway.set_home.save_failed", error=e) + # Preserve legacy home env vars for existing cron/setup consumers. + env_key = _home_target_env_var(platform_name) + thread_env_key = _home_thread_env_var(platform_name) + try: + from hermes_cli.config import save_env_value + save_env_value(env_key, str(chat_id)) + save_env_value(thread_env_key, str(thread_id or "")) + except Exception as e: + logger.warning("Home config saved but legacy env persistence failed: %s", e) + # Keep the running gateway config in sync too. The pre-restart - # notification path reads self.config before the process reloads env. - if source.platform: - platform_config = self.config.platforms.setdefault( - source.platform, - PlatformConfig(enabled=True), - ) - platform_config.home_channel = HomeChannel( - platform=source.platform, - chat_id=str(chat_id), - name=chat_name, - thread_id=str(thread_id) if thread_id else None, - ) + # notification path reads self.config before the process reloads config. + platform_config = getattr(self, "config").platforms.setdefault( + source.platform, + PlatformConfig(enabled=not via_relay), + ) + platform_config.home_channel = home return t("gateway.set_home.success", name=chat_name, chat_id=chat_id) diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 014e88ec6db..0320d2fccbd 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -736,6 +736,128 @@ class TestDeliverResultWrapping: # Media files should be forwarded separately assert kwargs["media_files"] == [(str(media_path), False)] + def test_relay_fronted_home_uses_relay_config_and_live_adapter(self, monkeypatch, tmp_path): + """Persisted Slack home survives restart without native Slack config.""" + from concurrent.futures import Future + + from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig + + relay = MagicMock() + relay.fronts_platform.side_effect = lambda platform: platform == Platform.SLACK + relay.send_for_platform = AsyncMock(return_value=MagicMock(success=True)) + relay.send_voice = AsyncMock(return_value=MagicMock(success=True)) + relay.supports_inchannel_continuable = False + + config = GatewayConfig( + platforms={ + Platform.RELAY: PlatformConfig(enabled=True), + Platform.SLACK: PlatformConfig( + enabled=False, + home_channel=HomeChannel( + platform=Platform.SLACK, + chat_id="D123", + name="Owner DM", + user_id="U123", + ), + ), + }, + ) + loop = MagicMock() + loop.is_running.return_value = True + + def fake_run_coro(coro, _loop): + import asyncio as _asyncio + + future = Future() + try: + future.set_result(_asyncio.run(coro)) + except BaseException as exc: # noqa: BLE001 + future.set_exception(exc) + return future + + standalone_send = AsyncMock(return_value={"success": True}) + media_path = self._safe_media_path(tmp_path, monkeypatch, "relay-voice.mp3") + monkeypatch.setenv("SLACK_HOME_CHANNEL", "D123") + job = { + "id": "relay-cron", + "deliver": "slack", + } + + with ( + patch("gateway.config.load_gateway_config", return_value=config), + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), + patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro), + patch("tools.send_message_tool._send_to_platform", new=standalone_send), + ): + result = _deliver_result( + job, + f"scheduled result\nMEDIA:{media_path}", + adapters={Platform.RELAY: relay}, + loop=loop, + ) + + assert result is None + relay.send_for_platform.assert_awaited_once() + args = relay.send_for_platform.await_args.args + assert args[:3] == (Platform.SLACK, "D123", "scheduled result") + assert relay.send_for_platform.await_args.kwargs["metadata"]["user_id"] == "U123" + relay.send_voice.assert_awaited_once() + media_metadata = relay.send_voice.await_args.kwargs["metadata"] + assert media_metadata["_relay_logical_platform"] == "slack" + assert media_metadata["user_id"] == "U123" + standalone_send.assert_not_awaited() + + def test_relay_fronted_delivery_failure_does_not_use_native_fallback(self): + """Connector-owned credentials must never fall through to native send.""" + from concurrent.futures import Future + + from gateway.config import GatewayConfig, Platform, PlatformConfig + + relay = MagicMock() + relay.fronts_platform.side_effect = lambda platform: platform == Platform.SLACK + relay.send_for_platform = AsyncMock( + return_value=MagicMock(success=False, error="connector unavailable") + ) + relay.supports_inchannel_continuable = False + config = GatewayConfig( + platforms={Platform.RELAY: PlatformConfig(enabled=True)}, + ) + loop = MagicMock() + loop.is_running.return_value = True + + def fake_run_coro(coro, _loop): + import asyncio as _asyncio + + future = Future() + try: + future.set_result(_asyncio.run(coro)) + except BaseException as exc: # noqa: BLE001 + future.set_exception(exc) + return future + + standalone_send = AsyncMock(return_value={"success": True}) + job = { + "id": "relay-cron-failure", + "deliver": "slack:D123", + } + + with ( + patch("gateway.config.load_gateway_config", return_value=config), + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), + patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro), + patch("tools.send_message_tool._send_to_platform", new=standalone_send), + ): + result = _deliver_result( + job, + "scheduled result", + adapters={Platform.RELAY: relay}, + loop=loop, + ) + + assert result is not None + assert "connector unavailable" in result + standalone_send.assert_not_awaited() + def test_live_adapter_sends_media_as_attachments(self, tmp_path, monkeypatch): """When a live adapter is available, MEDIA files should be sent as native platform attachments (e.g., Discord voice, Telegram audio) rather than diff --git a/tests/gateway/test_config.py b/tests/gateway/test_config.py index 163376e1a7f..451ad7012b0 100644 --- a/tests/gateway/test_config.py +++ b/tests/gateway/test_config.py @@ -23,18 +23,75 @@ from gateway.config import ( StreamingConfig, _apply_env_overrides, load_gateway_config, + persist_home_channel, ) class TestHomeChannelRoundtrip: def test_to_dict_from_dict(self): - hc = HomeChannel(platform=Platform.DISCORD, chat_id="999", name="general") + hc = HomeChannel( + platform=Platform.DISCORD, + chat_id="999", + name="general", + user_id="user-123", + scope_id="guild-456", + ) d = hc.to_dict() restored = HomeChannel.from_dict(d) assert restored.platform == Platform.DISCORD assert restored.chat_id == "999" assert restored.name == "general" + assert restored.user_id == "user-123" + assert restored.scope_id == "guild-456" + + def test_relay_only_slack_home_hydrates_disabled_with_provenance(self): + config = GatewayConfig( + platforms={ + Platform.SLACK: PlatformConfig( + enabled=False, + home_channel=HomeChannel( + platform=Platform.SLACK, + chat_id="D123", + name="Owner DM", + user_id="U123", + ), + ) + } + ) + + with patch.dict(os.environ, {"SLACK_HOME_CHANNEL": "D123"}, clear=False): + _apply_env_overrides(config) + + slack = config.platforms[Platform.SLACK] + assert slack.enabled is False + assert slack.home_channel is not None + assert slack.home_channel.chat_id == "D123" + assert slack.home_channel.user_id == "U123" + + def test_persisted_relay_home_survives_real_config_reload(self, tmp_path, monkeypatch): + monkeypatch.setenv("SLACK_HOME_CHANNEL", "D123") + monkeypatch.delenv("SLACK_BOT_TOKEN", raising=False) + home_token = set_hermes_home_override(str(tmp_path)) + try: + persist_home_channel( + HomeChannel( + platform=Platform.SLACK, + chat_id="D123", + name="Owner DM", + user_id="U123", + ) + ) + config = load_gateway_config() + finally: + reset_hermes_home_override(home_token) + + slack = config.platforms[Platform.SLACK] + assert slack.enabled is False + assert slack.token is None + assert slack.home_channel is not None + assert slack.home_channel.chat_id == "D123" + assert slack.home_channel.user_id == "U123" class TestPlatformConfigRoundtrip: diff --git a/tests/gateway/test_delivery.py b/tests/gateway/test_delivery.py index 807d9cbb4ac..a1849985c7a 100644 --- a/tests/gateway/test_delivery.py +++ b/tests/gateway/test_delivery.py @@ -1,10 +1,13 @@ """Tests for the delivery routing module.""" import pytest +from typing import Any, cast -from gateway.config import GatewayConfig, Platform +from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig from gateway.delivery import DeliveryRouter, DeliveryTarget from gateway.platforms.base import SendResult +from gateway.relay.adapter import RelayAdapter +from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor from gateway.session import SessionSource @@ -125,6 +128,98 @@ class TestPlatformNameCaseInsensitivity: assert target.platform == Platform.TELEGRAM assert target.chat_id == "12345" +class _RelayDeliveryTransport: + """Relay transport that advertises Slack and records outbound wire frames.""" + + def __init__(self): + self._identities = [("slack", "bot-1")] + self.sent = [] + + async def send_outbound(self, action, *, platform=None): + self.sent.append((action, platform)) + if not action.get("metadata", {}).get("user_id"): + return {"success": False, "error": "target not routed to an onboarded tenant"} + return {"success": True, "message_id": "relay-message-1"} + + +def _make_relay(transport): + return RelayAdapter( + PlatformConfig(enabled=True), + CapabilityDescriptor( + contract_version=CONTRACT_VERSION, + platform="slack", + label="Slack", + max_message_length=4000, + supports_draft_streaming=False, + supports_edit=True, + supports_threads=True, + markdown_dialect="slack", + len_unit="chars", + ), + transport=cast(Any, transport), + ) + + +@pytest.mark.asyncio +async def test_relay_fronted_target_delivers_without_prior_inbound_chat_state(tmp_path, monkeypatch): + """A persisted Slack home must work immediately after a gateway restart.""" + monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path) + transport = _RelayDeliveryTransport() + relay = _make_relay(transport) + config = GatewayConfig( + platforms={ + Platform.RELAY: PlatformConfig(enabled=True), + Platform.SLACK: PlatformConfig( + enabled=False, + home_channel=HomeChannel( + platform=Platform.SLACK, + chat_id="D123", + name="Owner DM", + user_id="U123", + ), + ), + }, + ) + router = DeliveryRouter(config, adapters={Platform.RELAY: relay}) + + result = await router._deliver_to_platform( + DeliveryTarget(platform=Platform.SLACK, chat_id="D123"), + "scheduled result", + metadata={"job_id": "cron-1", "user_id": "stale-user"}, + ) + + assert getattr(result, "success", False) is True + assert len(transport.sent) == 1 + action, wire_platform = transport.sent[0] + assert wire_platform == "slack" + assert action["chat_id"] == "D123" + assert action["metadata"] == {"job_id": "cron-1", "user_id": "U123"} + + +@pytest.mark.asyncio +async def test_relay_media_fallback_retains_explicit_platform_and_owner(): + """Attachment fallback cannot default to another Relay identity after restart.""" + transport = _RelayDeliveryTransport() + transport._identities = [("discord", "discord-bot"), ("slack", "slack-bot")] + relay = _make_relay(transport) + + result = await relay.send_document( + chat_id="D123", + file_path="/tmp/report.pdf", + metadata={ + "_relay_logical_platform": "slack", + "user_id": "U123", + }, + ) + + assert result.success is True + assert len(transport.sent) == 1 + action, wire_platform = transport.sent[0] + assert wire_platform == "slack" + assert action["metadata"] == {"user_id": "U123"} + assert "_relay_logical_platform" not in action["metadata"] + + class RecordingAdapter: def __init__(self): self.calls = [] @@ -141,6 +236,92 @@ class RecordingAdapter: return "38049" +@pytest.mark.asyncio +async def test_native_adapter_wins_when_relay_also_fronts_platform(tmp_path, monkeypatch): + monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path) + native = RecordingAdapter() + transport = _RelayDeliveryTransport() + relay = _make_relay(transport) + config = GatewayConfig( + platforms={ + Platform.SLACK: PlatformConfig(enabled=True), + Platform.RELAY: PlatformConfig(enabled=True), + }, + ) + router = DeliveryRouter( + config, + adapters={Platform.SLACK: native, Platform.RELAY: relay}, + ) + + await router._deliver_to_platform( + DeliveryTarget(platform=Platform.SLACK, chat_id="D123"), + "native result", + metadata=None, + ) + + assert native.calls == [ + {"chat_id": "D123", "content": "native result", "metadata": None} + ] + assert transport.sent == [] + + +@pytest.mark.asyncio +async def test_disabled_native_adapter_does_not_shadow_relay(tmp_path, monkeypatch): + monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path) + native = RecordingAdapter() + transport = _RelayDeliveryTransport() + relay = _make_relay(transport) + config = GatewayConfig( + platforms={ + Platform.SLACK: PlatformConfig( + enabled=False, + home_channel=HomeChannel( + platform=Platform.SLACK, + chat_id="D123", + name="Owner DM", + user_id="U123", + ), + ), + Platform.RELAY: PlatformConfig(enabled=True), + }, + ) + router = DeliveryRouter( + config, + adapters={Platform.SLACK: native, Platform.RELAY: relay}, + ) + + await router._deliver_to_platform( + DeliveryTarget(platform=Platform.SLACK, chat_id="D123"), + "relay result", + metadata=None, + ) + + assert native.calls == [] + assert len(transport.sent) == 1 + assert transport.sent[0][1] == "slack" + + +@pytest.mark.asyncio +async def test_relay_does_not_claim_unadvertised_platform(tmp_path, monkeypatch): + monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path) + transport = _RelayDeliveryTransport() + transport._identities = [("discord", "bot-1")] + relay = _make_relay(transport) + config = GatewayConfig( + platforms={Platform.RELAY: PlatformConfig(enabled=True)}, + ) + router = DeliveryRouter(config, adapters={Platform.RELAY: relay}) + + with pytest.raises(ValueError, match="No adapter configured for slack"): + await router._deliver_to_platform( + DeliveryTarget(platform=Platform.SLACK, chat_id="D123"), + "must not route", + metadata=None, + ) + + assert transport.sent == [] + + class StaleTopicAdapter: def __init__(self): self.calls = [] diff --git a/tests/gateway/test_restart_notification.py b/tests/gateway/test_restart_notification.py index e110eb876e0..a9cca3f65fa 100644 --- a/tests/gateway/test_restart_notification.py +++ b/tests/gateway/test_restart_notification.py @@ -7,7 +7,7 @@ from unittest.mock import AsyncMock, MagicMock import pytest import gateway.run as gateway_run -from gateway.config import HomeChannel, Platform +from gateway.config import HomeChannel, Platform, PlatformConfig from gateway.platforms.base import MessageEvent, MessageType, SendResult from gateway.session import build_session_key from tests.gateway.restart_test_helpers import ( @@ -77,6 +77,35 @@ async def test_restart_command_writes_notify_file(tmp_path, monkeypatch): assert "thread_id" not in data # no thread → omitted +@pytest.mark.asyncio +async def test_relay_restart_command_persists_authenticated_routing_provenance( + tmp_path, monkeypatch +): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + + runner, _adapter = make_restart_runner() + runner.request_restart = MagicMock(return_value=True) + source = make_restart_source(chat_id="D123") + source.platform = Platform.SLACK + source.user_id = "U123" + source.scope_id = "T123" + source.delivered_via_upstream_relay = True + event = MessageEvent( + text="/restart", + message_type=MessageType.TEXT, + source=source, + message_id="m-relay-restart", + ) + + await runner._handle_restart_command(event) + + data = json.loads((tmp_path / ".restart_notify.json").read_text()) + assert data["platform"] == "slack" + assert data["user_id"] == "U123" + assert data["scope_id"] == "T123" + assert data["delivered_via_upstream_relay"] is True + + @pytest.mark.asyncio async def test_restart_command_uses_service_restart_under_systemd(tmp_path, monkeypatch): """Under systemd (INVOCATION_ID set), /restart uses via_service=True.""" @@ -189,6 +218,7 @@ async def test_sethome_updates_running_config_for_same_process_restart(tmp_path, saved[key] = value monkeypatch.setattr("hermes_cli.config.save_env_value", _fake_save_env_value) + monkeypatch.setattr("gateway.slash_commands.persist_home_channel", lambda home, **kwargs: None) runner, _adapter = make_restart_runner() source = make_restart_source(chat_id="home-42") @@ -221,6 +251,7 @@ async def test_sethome_preserves_thread_target_for_same_process_restart(tmp_path saved[key] = value monkeypatch.setattr("hermes_cli.config.save_env_value", _fake_save_env_value) + monkeypatch.setattr("gateway.slash_commands.persist_home_channel", lambda home, **kwargs: None) runner, _adapter = make_restart_runner() source = make_restart_source(chat_id="parent-42", thread_id="topic-7") @@ -243,6 +274,67 @@ async def test_sethome_preserves_thread_target_for_same_process_restart(tmp_path assert home.thread_id == "topic-7" +@pytest.mark.asyncio +async def test_relay_sethome_persists_authenticated_logical_owner(monkeypatch): + persisted = [] + monkeypatch.setattr( + "gateway.slash_commands.persist_home_channel", + lambda home, **kwargs: persisted.append(home), + ) + monkeypatch.setattr("hermes_cli.config.save_env_value", lambda key, value: None) + + runner, _adapter = make_restart_runner() + relay = MagicMock() + relay.fronts_platform.side_effect = lambda platform: platform == Platform.SLACK + runner._adapter_for_source = lambda source: relay + source = make_restart_source(chat_id="D123") + source.platform = Platform.SLACK + source.user_id = "U123" + source.scope_id = None + source.delivered_via_upstream_relay = True + event = MessageEvent( + text="/sethome", + message_type=MessageType.TEXT, + source=source, + message_id="m-relay-home", + ) + + result = await runner._handle_set_home_command(event) + + assert "Home channel set" in result + assert len(persisted) == 1 + assert persisted[0].platform == Platform.SLACK + assert persisted[0].chat_id == "D123" + assert persisted[0].user_id == "U123" + assert runner.config.platforms[Platform.SLACK].enabled is False + + +@pytest.mark.asyncio +async def test_relay_sethome_rejects_unadvertised_platform(monkeypatch): + persist = MagicMock() + monkeypatch.setattr("gateway.slash_commands.persist_home_channel", persist) + + runner, _adapter = make_restart_runner() + relay = MagicMock() + relay.fronts_platform.return_value = False + runner._adapter_for_source = lambda source: relay + source = make_restart_source(chat_id="D123") + source.platform = Platform.SLACK + source.user_id = "U123" + source.delivered_via_upstream_relay = True + event = MessageEvent( + text="/sethome", + message_type=MessageType.TEXT, + source=source, + message_id="m-relay-home", + ) + + result = await runner._handle_set_home_command(event) + + assert "Failed to save" in result + persist.assert_not_called() + + # ── home-channel startup notifications ───────────────────────────────────── @@ -370,6 +462,42 @@ async def test_send_home_channel_startup_notification_ignores_false_send_result( adapter.send.assert_called_once() +@pytest.mark.asyncio +async def test_relay_fronted_logical_home_gets_startup_notification(tmp_path, monkeypatch): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + + runner, _native = make_restart_runner() + relay = MagicMock() + relay.fronts_platform.side_effect = lambda platform: platform == Platform.SLACK + relay.send_for_platform = AsyncMock(return_value=SendResult(success=True, message_id="home")) + runner.adapters = {Platform.RELAY: relay} + runner.config.platforms = { + Platform.RELAY: PlatformConfig(enabled=True), + Platform.SLACK: PlatformConfig( + enabled=False, + home_channel=HomeChannel( + platform=Platform.SLACK, + chat_id="D123", + name="Owner DM", + user_id="U123", + scope_id="T123", + ), + ), + } + + delivered = await runner._send_home_channel_startup_notifications() + + assert delivered == {("slack", "D123", None)} + relay.send_for_platform.assert_awaited_once() + assert relay.send_for_platform.await_args.args[:3] == ( + Platform.SLACK, + "D123", + "♻️ Gateway online — Hermes is back and ready.", + ) + assert relay.send_for_platform.await_args.kwargs["metadata"]["user_id"] == "U123" + assert relay.send_for_platform.await_args.kwargs["metadata"]["scope_id"] == "T123" + + # ── _send_restart_notification ─────────────────────────────────────────── @@ -398,6 +526,46 @@ async def test_send_restart_notification_delivers_and_cleans_up(tmp_path, monkey assert not notify_path.exists() +@pytest.mark.asyncio +async def test_relay_restart_notification_uses_logical_platform_and_owner(tmp_path, monkeypatch): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + notify_path = tmp_path / ".restart_notify.json" + notify_path.write_text( + json.dumps( + { + "platform": "slack", + "chat_id": "D123", + "chat_type": "dm", + "user_id": "U123", + "scope_id": "T123", + "delivered_via_upstream_relay": True, + } + ) + ) + + runner, _native = make_restart_runner() + relay = MagicMock() + relay.fronts_platform.side_effect = lambda platform: platform == Platform.SLACK + relay.send_for_platform = AsyncMock( + return_value=SendResult(success=True, message_id="restart") + ) + runner.adapters = {Platform.RELAY: relay} + runner.config.platforms = { + Platform.RELAY: PlatformConfig(enabled=True), + Platform.SLACK: PlatformConfig(enabled=False), + } + + delivered_target = await runner._send_restart_notification() + + assert delivered_target == ("slack", "D123", None) + relay.send_for_platform.assert_awaited_once() + assert relay.send_for_platform.await_args.args[0:2] == (Platform.SLACK, "D123") + metadata = relay.send_for_platform.await_args.kwargs["metadata"] + assert metadata["user_id"] == "U123" + assert metadata["scope_id"] == "T123" + assert not notify_path.exists() + + @pytest.mark.asyncio async def test_send_restart_notification_with_thread(tmp_path, monkeypatch): """Thread ID is passed as metadata so the message lands in the right topic."""