fix(gateway): secondary-profile send_message, notices and /loop wakeups go out via their own bot
Under gateway.multiplex_profiles a turn running for a secondary profile P resolved its live adapter by bare platform from runner.adapters — the DEFAULT profile's map — so P's send_message tool calls (send/react/media on slack, matrix, wecom, buzz, ntfy, every plugin platform), its "Gateway shutting down/restarted" and /update notices, its /loop wakeups, and its Discord bot's unauthorized-slash operator alert all left through the default bot (Telegram DMs landed in the user's chat with the other bot). Every such door now resolves through the profile-aware, fail-closed resolver already used by the inbound reply path (authz_mixin: _adapters_for_profile / _authorization_adapter / _adapter_for_source): P's own adapter, or None → a clear error, never the default bot. - tools/send_message_senders.py::_live_adapter — resolve via runner._authorization_adapter(platform, get_active_profile_name()); shared by _send_via_adapter, _handle_react/unreact, media sends, matrix E2EE fast path and the WeCom standalone sender. - gateway/authz_mixin.py — extract _adapters_for_profile (the whole map, for relay- aware resolve_delivery_transport callers); _authorization_adapter reuses it. - gateway/run_shutdown.py — shutdown/restart notice for a running session uses the session's source transport / agent:<profile>: key lane, never self.adapters. - gateway/slash_commands.py + run_notifications.py — /restart and /update markers persist `profile`; the restart notice, update result and update prompt resolve the requester's own adapter (legacy markers fall back to the session_key lane). /goal, /heartbeat, /approve, /deny confirmations use the source's own transport. - gateway/slash_commands_goals.py + run_goals.py — /loop persists `profile` in its route; the wakeup watcher scans every served profile's store under its own scope (same shape as _handoff_watcher) and fires through that profile's adapter map. - plugins/platforms/discord/adapter.py::_notify_unauthorized_slash — alert stays in the owning profile (its adapters and its home channels). - plugins/platforms/wecom/adapter.py::_standalone_send — via _live_adapter. - docs: website/docs/user-guide/multi-profile-gateways.md (outbound identity). Tests (red on base): tests/tools/test_send_message_multiplex_profile_adapter.py, tests/gateway/test_multiplex_notice_egress_profile_adapter.py. Co-authored-by: pierrenode <298902573+pierrenode@users.noreply.github.com>
This commit is contained in:
+23
-21
@@ -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``."""
|
||||
|
||||
+43
-17
@@ -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)
|
||||
|
||||
@@ -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:<profile>:`` 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:<profile>:`` 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
|
||||
|
||||
+17
-6
@@ -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:<profile>:`` 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"):
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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 = (
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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 == []
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user