mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-31 19:16:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix(gateway): deliver relay-backed homes after restart

Browse source
This commit is contained in:
Victor Kyriazakos 2026-07-23 14:10:02 +00:00 • committed by Teknium
parent 6de7c0f7a7
commit 45a408f41a
10 changed files with 878 additions and 77 deletions
Download patch file Download diff file Expand all files Collapse all files

61
cron/scheduler.py
View file

@ -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

42
gateway/config.py
View file

@ -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

103
gateway/delivery.py
View file

@ -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

74
gateway/relay/adapter.py
View file

@ -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(

60
gateway/run.py
View file

@ -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",

81
gateway/slash_commands.py
View file

@ -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)

122
tests/cron/test_scheduler.py
View file

@ -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

59
tests/gateway/test_config.py
View file

@ -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:

183
tests/gateway/test_delivery.py
View file

@ -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 = []

170
tests/gateway/test_restart_notification.py
View file

@ -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."""