diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 37a4a0f66d0f..46fc19125458 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1080,6 +1080,22 @@ platform_toolsets: # priority_mode: prepend # priority: # - my_plugin_command +# # Multi-account bots: declare each additional bot's credential in .env +# # as TELEGRAM_BOT_TOKEN_ (e.g. TELEGRAM_BOT_TOKEN_SUPPORT); +# # this yaml block carries only per-account behavioral settings. +# # Account names must match [a-z0-9][a-z0-9_-]*. The unsuffixed +# # TELEGRAM_BOT_TOKEN remains the default account, so single-bot +# # setups need no changes. Target a specific bot when delivering with +# # telegram@:, e.g. deliver: "telegram@support:123456789". +# accounts: +# support: +# display_name: "Support Bot" +# home_channel: +# chat_id: "123456789" +# sales: +# display_name: "Sales Bot" +# home_channel: +# chat_id: "987654321" # slack: # extra: # # Render live tool calls as Slack-native plan/task cards. This explicit diff --git a/cron/scheduler.py b/cron/scheduler.py index 79b5210cc1aa..570c69145e57 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1851,6 +1851,12 @@ def _resolve_single_delivery_target(job: dict, deliver_value: str) -> Optional[d if ":" in deliver_value: platform_name, rest = deliver_value.split(":", 1) + # Named bot account (#8287): "telegram@support:123" delivers through + # the support account's adapter. Split the account off before any + # platform validation/lookup — downstream maps know "telegram", not + # "telegram@support". + platform_name, _at_sep, account_name = platform_name.partition("@") + account_name = account_name.strip().lower() or None platform_key = platform_name.lower() from tools.send_message_tool import ( @@ -1881,11 +1887,14 @@ def _resolve_single_delivery_target(job: dict, deliver_value: str) -> Optional[d ): thread_id = origin.get("thread_id") - return { + target = { "platform": platform_name, "chat_id": chat_id, "thread_id": thread_id, } + if account_name: + target["account"] = account_name + return target platform_name = deliver_value if origin and origin.get("platform") == platform_name: @@ -2229,6 +2238,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option platform_name = target["platform"] chat_id = target["chat_id"] thread_id = target.get("thread_id") + account_name = target.get("account") # Diagnostic: log thread_id for topic-aware delivery debugging origin = _resolve_origin(job) or {} @@ -2489,11 +2499,29 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option from agent.async_utils import safe_schedule_threadsafe router = DeliveryRouter(config, adapters) + # Named-account routing (#8287): sync the gateway's + # account registry so account targets resolve fail-closed + # (never through the default bot). The live runner is + # reachable via the module weakref gateway/run.py keeps + # for exactly this kind of module-level consumer; when no + # gateway is running (CLI cron), the registry stays empty + # and account targets are skipped with the router's + # "No adapter configured" error. + try: + from gateway.run import _gateway_runner_ref + _runner = _gateway_runner_ref() + if _runner is not None: + router.account_adapters = getattr( + _runner, "_account_adapters", {} + ) or {} + except Exception: + pass route_target = DeliveryTarget( platform=platform, chat_id=str(chat_id), thread_id=route_thread_id, is_explicit=True, + account=account_name, ) # Pass thread routing via the target (not a bare metadata # "thread_id"): the router only applies its Telegram DM-topic diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index fae4b74a5e11..aa8e8e71dced 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -94,6 +94,7 @@ def _authorization_adapter( self, platform: Optional[Platform], profile: Optional[str] = None, + account: Optional[str] = None, ): """Resolve the live adapter whose intake policy should gate authorization. @@ -102,11 +103,30 @@ def _authorization_adapter( ``self.adapters``. ``SessionSource.profile`` selects which map to consult. When a stamped profile has its own adapter registry entry, the default profile's same-platform adapter must not be consulted as a fallback. + + Multi-account gateways (#8287) add a second dimension the same way: + named-account adapters live in ``_account_adapters[platform][account]`` + while the default account uses ``self.adapters``. A stamped account + with no registry entry fails closed for the same reason a profile + does — replying out the wrong bot is worse than not replying. """ if not platform: return None profile_name = (profile or "").strip() or None + # Coerce defensively: a MagicMock/SimpleNamespace source auto-creates + # a truthy ``account`` attribute (AGENTS.md pitfall #17), which must + # read as "default account", not trip the fail-closed account branch. + account_name = account.strip() if isinstance(account, str) else None + account_name = account_name or None + if account_name == "default": + account_name = None if profile_name and profile_name != "default": + if account_name: + # Named account inside a secondary profile is not a supported + # combination yet — fail closed rather than guessing a bot + # (#8287). Checked before the active-profile fast path so an + # account under a named active profile also fails closed. + return None active_profile = None active_profile_fn = getattr(self, "_active_profile_name", None) if callable(active_profile_fn): @@ -124,6 +144,9 @@ def _authorization_adapter( # (e.g. its adapter failed to connect) must NOT fall back to the # default profile's adapter — that sends replies out the wrong bot. return None + if account_name: + account_adapters = getattr(self, "_account_adapters", None) or {} + return (account_adapters.get(platform) or {}).get(account_name) adapters = getattr(self, "adapters", None) or {} return adapters.get(platform) @@ -152,6 +175,7 @@ def _adapter_for_source(self, source: Optional[SessionSource]): return self._authorization_adapter( getattr(source, "platform", None), getattr(source, "profile", None), + getattr(source, "account", None), ) def _registered_transport_adapter(self, source: SessionSource): diff --git a/gateway/config.py b/gateway/config.py index a00fa0f9a1ca..29652fbd1c4c 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -11,6 +11,7 @@ import logging import os import json +import re from pathlib import Path from dataclasses import asdict, dataclass, field, is_dataclass from typing import Dict, List, Optional, Any, Callable @@ -447,6 +448,66 @@ def _scan_bundled_plugin_platforms(cls) -> set: } +def derive_account_platform_config( + platform: "Platform", + platform_config: "PlatformConfig", + account_block: Optional[dict], +) -> "PlatformConfig": + """Derive a per-account ``PlatformConfig`` for a named bot account (#8287). + + The consumer sees an ordinary ``PlatformConfig`` — the account's own token, + its own ``home_channel``, and account-block settings overriding the + platform-level ``extra`` — so adapters and the send path stay + account-agnostic. The ``accounts`` map itself is stripped from the derived + ``extra`` so a derived config can never recurse into another account. + + Shared by the gateway's account-adapter startup and ``send_message``'s + per-account routing, so both resolve an account identically. + """ + import dataclasses as _dc + + merged_extra = { + k: v for k, v in (platform_config.extra or {}).items() if k != "accounts" + } + home_channel = platform_config.home_channel + token = platform_config.token + for key, value in (account_block or {}).items(): + if key == "token": + token = value + elif key == "home_channel" and isinstance(value, dict): + # The platform is implicit inside its own account block. + _hc = dict(value) + _hc.setdefault("platform", platform.value) + home_channel = HomeChannel.from_dict(_hc) + else: + merged_extra[key] = value + return _dc.replace( + platform_config, + token=token, + home_channel=home_channel, + extra=merged_extra, + ) + + +def resolve_platform_account( + platform_ref: str, +) -> tuple[str, Optional[str]]: + """Split a ``platform[@account]`` reference into ``(platform, account)``. + + ``"telegram"`` -> ``("telegram", None)``; ``"telegram@support"`` -> + ``("telegram", "support")``. The account is lowercased to match the + normalization applied when accounts are parsed from config/env. ``default`` + resolves to ``None`` so it is spelled the same everywhere. + """ + base, sep, account = (platform_ref or "").partition("@") + if not sep: + return base, None + account = account.strip().lower() + if not account or account == "default": + return base, None + return base, account + + def platform_binds_port(platform_value: str, extra: Optional[dict] = None) -> bool: """Return True when *platform_value* actually binds a port for *extra* config. @@ -731,6 +792,38 @@ def from_dict(cls, data: Dict[str, Any]) -> "PlatformConfig": if _typing_text is None: _typing_text = extra.get("typing_status_text") + # Multi-account blocks (#8287): ``accounts:`` may arrive top-level + # (``platforms.telegram.accounts`` in YAML) or bridged into extra by + # the shared-key loop. Normalize account names (lowercased) into + # ``extra["accounts"]`` so the adapter registry has a single read + # path. Tokens are secrets and load from ``_BOT_TOKEN_`` + # env vars in ``_apply_env_overrides`` — a ``token`` key inside a YAML + # account block is honored for parity but ``.env`` is the supported + # home for credentials. + _accounts = data.get("accounts") + if _accounts is None: + _accounts = extra.get("accounts") + if isinstance(_accounts, dict): + _norm_accounts: Dict[str, Any] = {} + for _acct_name, _acct_block in _accounts.items(): + _acct_key = str(_acct_name).strip().lower() + if not _acct_key: + continue + # Account names become a session-key namespace suffix + # (``agent:main@``), so the charset is restricted: + # ``:`` would break key splitting, ``@`` the suffix parse. + if not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", _acct_key): + logger.warning( + "Ignoring platform account %r: names must match " + "[a-z0-9][a-z0-9_-]* (they become session-key " + "namespace suffixes)", + _acct_name, + ) + continue + _norm_accounts[_acct_key] = _coerce_dict(_acct_block) + if _norm_accounts: + extra["accounts"] = _norm_accounts + channel_overrides: Dict[str, ChannelOverride] = {} raw_overrides = data.get("channel_overrides") or {} if isinstance(raw_overrides, dict): @@ -1911,6 +2004,40 @@ def _enable_from_env(platform: Platform) -> PlatformConfig: if telegram_token: telegram_config = _enable_from_env(Platform.TELEGRAM) telegram_config.token = telegram_token + + # Multi-account tokens (#8287): ``TELEGRAM_BOT_TOKEN_`` declares + # an additional bot account named ```` (lowercased). The + # unsuffixed ``TELEGRAM_BOT_TOKEN`` remains the default account, so + # single-bot setups are byte-identical to before. Candidate names are + # enumerated from the process env (dotenv loads ``.env`` there) and each + # value is read back through ``getenv`` so profile-scoped secrets win + # when a scope is active. Behavioral per-account settings (allowlists, + # home channels, display names) belong in ``platforms.telegram.accounts`` + # in config.yaml — env vars carry only the credential. + _tg_account_prefix = "TELEGRAM_BOT_TOKEN_" + for _env_name in sorted(os.environ): + if not _env_name.startswith(_tg_account_prefix): + continue + _acct_name = _env_name[len(_tg_account_prefix):].strip().lower() + _acct_token = getenv(_env_name) + if not _acct_name or not _acct_token: + continue + # Same charset rule as the YAML block: names become session-key + # namespace suffixes. + if not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", _acct_name): + logger.warning( + "Ignoring %s: account names must match [a-z0-9][a-z0-9_-]*", + _env_name, + ) + continue + _tg_cfg = _enable_from_env(Platform.TELEGRAM) + _tg_accounts = _tg_cfg.extra.setdefault("accounts", {}) + if not isinstance(_tg_accounts, dict): + _tg_accounts = {} + _tg_cfg.extra["accounts"] = _tg_accounts + _acct_block = _tg_accounts.setdefault(_acct_name, {}) + if isinstance(_acct_block, dict): + _acct_block["token"] = _acct_token # Reply threading mode for Telegram (off/first/all) telegram_reply_mode = getenv("TELEGRAM_REPLY_TO_MODE", "").lower() diff --git a/gateway/delivery.py b/gateway/delivery.py index fa43db6d0f92..604c4b71ed1d 100644 --- a/gateway/delivery.py +++ b/gateway/delivery.py @@ -226,6 +226,10 @@ class DeliveryTarget: thread_id: Optional[str] = None is_origin: bool = False is_explicit: bool = False # True if chat_id was explicitly specified + # Named bot account on the platform (#8287): "telegram@support:123" + # targets the support bot; origin targets inherit the account the + # message arrived on. None = the platform's default account. + account: Optional[str] = None @classmethod def parse(cls, target: str, origin: Optional[SessionSource] = None) -> "DeliveryTarget": @@ -248,6 +252,8 @@ def parse(cls, target: str, origin: Optional[SessionSource] = None) -> "Delivery chat_id=origin.chat_id, thread_id=origin.thread_id, is_origin=True, + # Reply out the same bot the message arrived on (#8287). + account=getattr(origin, "account", None), ) else: # Fallback to local if no origin @@ -256,24 +262,35 @@ def parse(cls, target: str, origin: Optional[SessionSource] = None) -> "Delivery if target_lower == "local": return cls(platform=Platform.LOCAL) - # Check for platform:chat_id or platform:chat_id:thread_id format - # Use the original case for chat_id/thread_id to preserve case-sensitive IDs + # Check for platform:chat_id or platform:chat_id:thread_id format. + # The platform segment may carry a named bot account (#8287): + # "telegram@support" / "telegram@support:123456". Use the original + # case for chat_id/thread_id to preserve case-sensitive IDs. if ":" in target_stripped: parts = target_stripped.split(":", 2) platform_str = parts[0].lower() # Platform names are case-insensitive + platform_str, _at, account = platform_str.partition("@") chat_id = parts[1] if len(parts) > 1 else None thread_id = parts[2] if len(parts) > 2 else None try: platform = Platform(platform_str) - return cls(platform=platform, chat_id=chat_id, thread_id=thread_id, is_explicit=True) + return cls( + platform=platform, + chat_id=chat_id, + thread_id=thread_id, + is_explicit=True, + account=account or None, + ) except ValueError: # Unknown platform, treat as local return cls(platform=Platform.LOCAL) - - # Just a platform name (use home channel) + + # Just a platform name (use home channel), optionally account-scoped + # ("telegram@support" → the support bot's home channel). + platform_str, _at, account = target_lower.partition("@") try: - platform = Platform(target_lower) - return cls(platform=platform) + platform = Platform(platform_str) + return cls(platform=platform, account=account or None) except ValueError: # Unknown platform, treat as local return cls(platform=Platform.LOCAL) @@ -284,11 +301,16 @@ def to_string(self) -> str: return "origin" if self.platform == Platform.LOCAL: return "local" + platform_ref = ( + f"{self.platform.value}@{self.account}" + if self.account + else self.platform.value + ) if self.chat_id and self.thread_id: - return f"{self.platform.value}:{self.chat_id}:{self.thread_id}" + return f"{platform_ref}:{self.chat_id}:{self.thread_id}" if self.chat_id: - return f"{self.platform.value}:{self.chat_id}" - return self.platform.value + return f"{platform_ref}:{self.chat_id}" + return platform_ref class DeliveryRouter: @@ -312,8 +334,25 @@ def __init__(self, config: GatewayConfig, adapters: Dict[Platform, Any] = None, """ self.config = config self.adapters = adapters or {} + # Named-account adapters (#8287): Platform -> {account -> adapter}, + # synced by the gateway runner alongside ``adapters``. Account + # targets resolve here fail-closed — never through the default bot. + self.account_adapters: Dict[Platform, Dict[str, Any]] = {} self.output_dir = get_hermes_home() / "cron" / "output" self.dead_targets = dead_targets or DeadTargetRegistry() + + def _adapter_for_target(self, target: DeliveryTarget): + """Resolve the adapter for a target, honoring its account (#8287). + + A named account with no live adapter returns None (fail closed): + delivering account-addressed content out the default bot would leak + it to the wrong audience. + """ + if target.account: + return (self.account_adapters.get(target.platform) or {}).get( + target.account + ) + return self.adapters.get(target.platform) async def deliver( self, @@ -464,10 +503,24 @@ async def _deliver_to_platform( metadata: Optional[Dict[str, Any]] ) -> Dict[str, Any]: """Deliver content to a messaging platform.""" - 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 + # Named-account targets (#8287) resolve through the account registry + # fail-closed — never fall back to the default bot. The default path + # uses main's transport resolution (relay/provenance-aware). + if target.account: + adapter = self._adapter_for_target(target) + if not adapter: + raise ValueError( + f"No adapter configured for " + f"{target.platform.value}@{target.account}" + ) + else: + 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") diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 1a044d1521f7..26772f51a6c0 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -3005,6 +3005,13 @@ def set_status_text(self, chat_id: str, text: Optional[str]) -> None: def __init__(self, config: PlatformConfig, platform: Platform): self.config = config self.platform = platform + # Bot account this adapter instance serves (#8287). None = the + # platform's default account (single-bot gateways never set it). + # The gateway stamps this after construction when it starts named + # account adapters; adapters copy it onto every inbound + # ``SessionSource.account`` so session keys, busy guards, and + # outbound routing stay per-account. + self.account_name: Optional[str] = None self._message_handler: Optional[MessageHandler] = None # Optional gateway-supplied fan-out for platform-native emoji # reaction events (see ``set_reaction_handler``). @@ -7093,6 +7100,13 @@ def build_source( parent_chat_id=str(parent_chat_id) if parent_chat_id else None, message_id=str(message_id) if message_id else None, profile=profile, + # Bot account this adapter instance serves (#8287). This is the + # single inbound-construction site every platform's normal-event + # path flows through, so stamping here (not in per-platform + # helpers) is what actually routes named-bot traffic to its own + # session key and egress adapter. None on default/single-bot + # adapters — byte-identical to before. + account=getattr(self, "account_name", None), role_authorized=role_authorized, auto_thread_created=auto_thread_created, auto_thread_initial_name=auto_thread_initial_name, @@ -7107,7 +7121,7 @@ def build_source( # routes to unserved profiles consistently without surfacing HTTP 500s. source.profile_route_rejected = profile_route_rejected return source - + @abstractmethod async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: """ diff --git a/gateway/run.py b/gateway/run.py index af607198ba8c..d65ffa8f52ef 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2450,6 +2450,7 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor build_session_key, is_shared_multi_user_session, neutralize_untrusted_inline_text, + split_key_namespace, ) from gateway.delivery import ( DeliveryRouter, @@ -3542,12 +3543,19 @@ def _parse_session_key(session_key: str) -> "dict | None": thread_id, so we leave ``thread_id`` out to avoid mis-routing. """ parts = session_key.split(":") - if len(parts) >= 5 and parts[0] == "agent" and parts[1] == "main": + # Accept the default namespace with or without a multi-account suffix + # (``agent:main`` / ``agent:main@support``, #8287) — the positional + # layout after the namespace slot is identical. Named-profile keys stay + # excluded, as before. + _ns, _account = split_key_namespace(parts[1]) if len(parts) > 1 else ("", None) + if len(parts) >= 5 and parts[0] == "agent" and _ns == "main": result = { "platform": parts[2], "chat_type": parts[3], "chat_id": parts[4], } + if _account: + result["account"] = _account if len(parts) > 5 and parts[3] in {"dm", "thread"}: result["thread_id"] = parts[5] return result @@ -6472,6 +6480,17 @@ def __init__(self, config: Optional[GatewayConfig] = None): # sites are untouched when multiplexing is off (this dict is empty). # Populated by _start_secondary_profile_adapters(). self._profile_adapters: Dict[str, Dict[Platform, BasePlatformAdapter]] = {} + # Multi-account (#8287): adapters for NAMED bot accounts live here, + # keyed by Platform then account name. self.adapters stays the + # default account's map — same pattern as _profile_adapters, so the + # ~93 existing self.adapters[...] sites are untouched when no named + # accounts are configured (this dict is empty). Populated by + # _start_account_adapters(). + self._account_adapters: Dict[Platform, Dict[str, BasePlatformAdapter]] = {} + # Named-account adapters queued for background reconnection, keyed by + # (Platform, account name) — the account-dimension mirror of + # _failed_platforms (#8287). + self._failed_account_adapters: Dict[tuple, Dict[str, Any]] = {} self._warn_if_docker_media_delivery_is_risky() _gateway_runner_ref = _weakref.ref(self) @@ -7801,6 +7820,63 @@ async def _handle_reaction_event(self, ctx: Dict[str, Any]) -> None: except Exception: logger.debug("[Gateway] reaction hook emit failed", exc_info=True) + async def _handle_account_adapter_fatal_error( + self, adapter: BasePlatformAdapter, account_name: str + ) -> None: + """Account-dimension mirror of ``_handle_adapter_fatal_error`` (#8287). + + Same contract, scoped to ``_account_adapters[platform][account]``: + stale-owner guard against the account's own registry slot, pop + + disconnect, and retryable failures queued for background + reconnection under a ``(platform, account)`` key. An account's death + never touches the default slot and never triggers the all-platforms- + down shutdown logic — the default adapter (or other accounts) may + still be healthy. + """ + platform = adapter.platform + registry = (getattr(self, "_account_adapters", None) or {}).get(platform) or {} + existing = registry.get(account_name) + if existing is not None and existing is not adapter: + logger.debug( + "Ignoring stale fatal error from a superseded %s account %r " + "adapter instance: %s", + platform.value, account_name, + adapter.fatal_error_code or "unknown", + ) + return + + logger.error( + "Fatal %s adapter error (account %r, %s): %s", + platform.value, account_name, + adapter.fatal_error_code or "unknown", + adapter.fatal_error_message or "unknown error", + ) + self._update_platform_runtime_status( + f"{platform.value}@{account_name}", + platform_state="retrying" if adapter.fatal_error_retryable else "fatal", + error_code=adapter.fatal_error_code, + error_message=adapter.fatal_error_message, + ) + + if existing is adapter: + registry.pop(account_name, None) + if not registry: + self._account_adapters.pop(platform, None) + await self._safe_adapter_disconnect(adapter, platform) + + if adapter.fatal_error_retryable: + key = (platform, account_name) + if key not in self._failed_account_adapters: + self._failed_account_adapters[key] = { + "config": adapter.config, + "attempts": 0, + "next_retry": time.monotonic(), + } + logger.info( + "%s account %r queued for background reconnection", + platform.value, account_name, + ) + async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> None: """React to an adapter failure after startup. @@ -7944,6 +8020,17 @@ async def _handle_adapter_fatal_error_detached( await self.stop() async def _handle_adapter_fatal_error_impl(self, adapter: BasePlatformAdapter) -> None: + # Named-account adapters (#8287) live in _account_adapters, not the + # platform slot: without this branch the stale-owner guard below + # would see the DEFAULT adapter occupying the slot and silently + # ignore a dying account — no reconnect, no status, no log. + _account_name = getattr(adapter, "account_name", None) + # isinstance-guard: a MagicMock adapter auto-creates a truthy + # account_name (pitfall #17); only a real named account is a str. + if isinstance(_account_name, str) and _account_name: + await self._handle_account_adapter_fatal_error(adapter, _account_name) + return + # Snapshot the current owner of this platform slot before doing # anything else. If it's neither this adapter nor empty, a different # adapter has already taken over (e.g. this is a delayed notification @@ -7992,6 +8079,9 @@ async def _handle_adapter_fatal_error_impl(self, adapter: BasePlatformAdapter) - # the same object twice. self.adapters.pop(adapter.platform, None) self.delivery_router.adapters = self.adapters + self.delivery_router.account_adapters = ( + getattr(self, "_account_adapters", {}) or {} + ) # Queue retryable failures BEFORE any disconnect await (#80598). # A half-dead transport can wedge native close() (or swallow @@ -10267,8 +10357,7 @@ async def _notify_active_sessions_of_shutdown(self) -> None: # elsewhere), which would otherwise trigger # ``RuntimeError: dictionary changed size during iteration`` — # observed in a user report during gateway shutdown. - for platform, adapter in list(self.adapters.items()): - home = self.config.get_home_channel(platform) + for platform, adapter, home in self._iter_live_adapters_with_home(snapshot=True): if not home or not home.chat_id: continue @@ -12058,7 +12147,34 @@ async def start(self) -> bool: _multiplex_skipped_platforms.append(platform) continue enabled_platform_count += 1 - + + # Accounts-only configuration (#8287): named account tokens with + # no default credential. Don't attempt a doomed token-less + # default connect (it would fail and queue reconnects forever) — + # start the named account adapters directly. + if ( + not _platform_has_bot_credential(platform, platform_config) + and isinstance((platform_config.extra or {}).get("accounts"), dict) + and platform_config.extra["accounts"] + ): + logger.info( + "%s has no default-account credential; starting named " + "accounts only.", + platform.value, + ) + _acct_connected = await self._start_account_adapters( + platform, platform_config + ) + connected_count += _acct_connected + if _acct_connected: + self._update_platform_runtime_status( + platform.value, + platform_state="connected", + error_code=None, + error_message=None, + ) + continue + adapter = self._create_adapter(platform, platform_config) if not adapter: # Distinguish between missing builtin deps and missing plugin @@ -12210,6 +12326,17 @@ async def start(self) -> bool: platform, adapter ), } + + # Named bot accounts (#8287) start independently of the default + # adapter's outcome: a bad/absent default token must never keep a + # healthy named bot offline. _start_account_adapters is a no-op + # when no accounts are configured, so single-bot platforms are + # unaffected. (The accounts-only branch above already handled the + # no-default-credential case and `continue`d before reaching here.) + connected_count += await self._start_account_adapters( + platform, platform_config + ) + if await self._abort_startup_if_shutdown_requested(): return True @@ -12340,6 +12467,7 @@ async def start(self) -> bool: if await self._abort_startup_if_shutdown_requested(): return True self.delivery_router.adapters = self.adapters + self.delivery_router.account_adapters = getattr(self, "_account_adapters", {}) or {} self._wire_teams_pipeline_runtime() self._running = True @@ -13437,6 +13565,76 @@ async def _platform_reconnect_watcher(self) -> None: continue now = time.monotonic() + + # Named-account reconnects (#8287): independent of the platform + # pass below — an account's retry cadence mirrors a platform's, + # but success re-registers into _account_adapters, never the + # default slot. getattr-guarded: partially-constructed test + # runners (and any pre-#8287 pickle/restore) may lack the dict. + _failed_accounts = getattr(self, "_failed_account_adapters", None) + for _acct_key in list(_failed_accounts.keys()) if _failed_accounts else []: + if not self._running: + return + _acct_platform, _acct_name = _acct_key + _acct_info = self._failed_account_adapters[_acct_key] + if _acct_info.get("paused") or now < _acct_info["next_retry"]: + continue + _acct_attempt = _acct_info["attempts"] + 1 + logger.info( + "Reconnecting %s account %r (attempt %d)...", + _acct_platform.value, _acct_name, _acct_attempt, + ) + _acct_adapter = None + try: + _acct_adapter = self._create_adapter( + _acct_platform, _acct_info["config"] + ) + if not _acct_adapter: + logger.warning( + "Reconnect %s account %r: adapter creation " + "returned None, removing from retry queue", + _acct_platform.value, _acct_name, + ) + del self._failed_account_adapters[_acct_key] + continue + _acct_adapter.account_name = _acct_name + self._wire_account_adapter(_acct_adapter) + _acct_ok = await self._connect_adapter_with_timeout( + _acct_adapter, _acct_platform + ) + except Exception as _acct_exc: + logger.warning( + "Reconnect %s account %r failed: %s", + _acct_platform.value, _acct_name, _acct_exc, + ) + _acct_ok = False + if _acct_ok: + self._account_adapters.setdefault(_acct_platform, {})[ + _acct_name + ] = _acct_adapter + self._sync_voice_mode_state_to_adapter(_acct_adapter) + del self._failed_account_adapters[_acct_key] + self._update_platform_runtime_status( + f"{_acct_platform.value}@{_acct_name}", + platform_state="connected", + error_code=None, + error_message=None, + ) + logger.info( + "✓ %s reconnected (account %r)", + _acct_platform.value, _acct_name, + ) + else: + if _acct_adapter is not None: + await self._safe_adapter_disconnect( + _acct_adapter, _acct_platform + ) + _acct_info["attempts"] = _acct_attempt + # Same capped exponential backoff as the platform pass. + _acct_info["next_retry"] = now + min( + 30 * (2 ** min(_acct_attempt, 6)), 1800 + ) + for platform in list(self._failed_platforms.keys()): if not self._running: return @@ -13538,6 +13736,7 @@ async def _platform_reconnect_watcher(self) -> None: if hasattr(adapter, "_voice_input_callback"): adapter._voice_input_callback = self._handle_voice_channel_input self.delivery_router.adapters = self.adapters + self.delivery_router.account_adapters = getattr(self, "_account_adapters", {}) or {} del self._failed_platforms[platform] self._update_platform_runtime_status( platform.value, @@ -14884,9 +15083,152 @@ def _adapter_credential_fingerprint(adapter: Any) -> Optional[str]: import hashlib return hashlib.sha256(("hermes-mux:" + token).encode("utf-8")).hexdigest()[:16] + @staticmethod + def _account_platform_config( + platform: Platform, + platform_config: "PlatformConfig", + account_name: str, + account_block: Dict[str, Any], + ) -> "PlatformConfig": + """Derive a per-account PlatformConfig from the platform's config (#8287). + + Thin wrapper over ``gateway.config.derive_account_platform_config``, + which is shared with ``send_message``'s per-account routing so both + resolve an account's token/home/extra identically. + """ + from gateway.config import derive_account_platform_config + + return derive_account_platform_config( + platform, platform_config, account_block + ) + + def _iter_live_adapters_with_home(self, snapshot: bool = False): + """Yield (platform, adapter, home_channel) for every live adapter. + + Default-account adapters use the platform-level home channel; named + account adapters (#8287) use their own derived config's home channel, + so each bot broadcasts to its own home. ``snapshot=True`` list()s the + maps first so adapter.send() fatal paths popping registry entries + can't break iteration (the shutdown-broadcast lesson). + """ + adapters = list(self.adapters.items()) if snapshot else self.adapters.items() + for platform, adapter in adapters: + yield platform, adapter, self.config.get_home_channel(platform) + account_map = getattr(self, "_account_adapters", None) or {} + account_items = list(account_map.items()) if snapshot else account_map.items() + for platform, accounts in account_items: + account_adapters = list(accounts.values()) if snapshot else accounts.values() + for adapter in account_adapters: + yield platform, adapter, getattr(adapter.config, "home_channel", None) + + def _queue_account_reconnect( + self, platform: Platform, account_name: str, config: "PlatformConfig" + ) -> None: + """Queue a named account for background reconnection (#8287), keyed by + (platform, account). Idempotent — an already-queued account keeps its + existing backoff state. Shared by initial-startup failures and the + fatal-error path so both feed the one account reconnect watcher.""" + key = (platform, account_name) + if key not in self._failed_account_adapters: + self._failed_account_adapters[key] = { + "config": config, + "attempts": 1, + "next_retry": time.monotonic() + 30, + } + logger.info( + "%s account %r queued for background reconnection", + platform.value, account_name, + ) + + def _wire_account_adapter(self, adapter: BasePlatformAdapter) -> None: + """Wire a named-account adapter's handlers — identical to a default + adapter's wiring in the startup loop (#8287). Shared by account + startup and account reconnect so the two can never drift.""" + adapter.set_message_handler(self._handle_message) + adapter.set_fatal_error_handler(self._handle_adapter_fatal_error) + adapter.set_session_store(self.session_store) + adapter.set_busy_session_handler(self._handle_active_session_busy_message) + adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id) + adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform)) + adapter._busy_text_mode = self._busy_text_mode + + async def _start_account_adapters( + self, platform: Platform, platform_config: "PlatformConfig" + ) -> int: + """Start one adapter per NAMED bot account on ``platform`` (#8287). + + Called after the platform's default adapter is handled. Each account + adapter is wired identically to a default adapter, stamped with its + account name (adapters copy it onto every inbound + ``SessionSource.account``), and registered in + ``_account_adapters[platform][name]`` — the account-dimension mirror + of ``_profile_adapters``. Returns the number of accounts connected. + + A failed account connect is logged and skipped: it must not block the + default account or other accounts. (Account-aware reconnect queueing + lands with the per-account delivery/reconnect consumers.) + """ + accounts = (platform_config.extra or {}).get("accounts") + if not isinstance(accounts, dict) or not accounts: + return 0 + connected = 0 + for account_name, account_block in accounts.items(): + block = account_block if isinstance(account_block, dict) else {} + if not block.get("token"): + logger.warning( + "Skipping %s account %r: no token " + "(set %s_BOT_TOKEN_%s in .env)", + platform.value, account_name, + platform.value.upper(), account_name.upper(), + ) + continue + account_config = self._account_platform_config( + platform, platform_config, account_name, block + ) + adapter = self._create_adapter(platform, account_config) + if not adapter: + logger.warning( + "No adapter available for %s account %r", + platform.value, account_name, + ) + continue + adapter.account_name = account_name + self._wire_account_adapter(adapter) + logger.info("Connecting to %s (account %r)...", platform.value, account_name) + try: + success = await self._connect_adapter_with_timeout(adapter, platform) + except Exception as exc: + logger.error( + "%s account %r failed to connect: %s", + platform.value, account_name, exc, + ) + await self._safe_adapter_disconnect(adapter, platform) + self._queue_account_reconnect(platform, account_name, account_config) + continue + if not success: + logger.warning( + "✗ %s account %r failed to connect", platform.value, account_name + ) + await self._safe_adapter_disconnect(adapter, platform) + # Queue retryable initial failures so a transient startup + # blip (network, provider hiccup) is retried by the account + # reconnect watcher — mirrors the default adapter's path. + # Non-retryable fatal errors are left alone (a bad token + # shouldn't spin forever). + if getattr(adapter, "fatal_error_retryable", True): + self._queue_account_reconnect( + platform, account_name, account_config + ) + continue + self._account_adapters.setdefault(platform, {})[account_name] = adapter + self._sync_voice_mode_state_to_adapter(adapter) + connected += 1 + logger.info("✓ %s connected (account %r)", platform.value, account_name) + return connected + def _create_adapter( - self, - platform: Platform, + self, + platform: Platform, config: Any ) -> Optional[BasePlatformAdapter]: """Create the appropriate adapter for a platform. diff --git a/gateway/session.py b/gateway/session.py index 0121518152d1..41c306d02b92 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -186,6 +186,14 @@ class SessionSource: # target is not served. Excluded from repr/equality and wire serialization. profile_route_rejected: bool = field(default=False, repr=False, compare=False) + # Bot account this inbound message arrived on (#8287). A gateway can run + # multiple bot accounts on one platform (TELEGRAM_BOT_TOKEN_); + # the receiving adapter stamps its account name here so session keys, + # busy guards, and outbound delivery all route per account. None => the + # platform's default account — byte-identical behavior to a single-bot + # gateway. + account: Optional[str] = None + # Discord auto-thread metadata. Newly auto-created Discord threads start # with a fast placeholder title from the raw message, then the gateway can # rename them after the first agent turn using the generated session title. @@ -279,6 +287,8 @@ def to_dict(self) -> Dict[str, Any]: d["message_id"] = self.message_id if self.profile: d["profile"] = self.profile + if self.account: + d["account"] = self.account if self.auto_thread_created: d["auto_thread_created"] = True if self.auto_thread_initial_name: @@ -306,6 +316,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionSource": parent_chat_id=data.get("parent_chat_id"), message_id=data.get("message_id"), profile=data.get("profile"), + account=data.get("account"), auto_thread_created=bool(data.get("auto_thread_created", False)), auto_thread_initial_name=data.get("auto_thread_initial_name"), prospective_thread_id=data.get("prospective_thread_id"), @@ -1067,7 +1078,9 @@ def is_shared_multi_user_session( return not group_sessions_per_user -def _session_key_namespace(profile: Optional[str]) -> str: +def _session_key_namespace( + profile: Optional[str], account: Optional[str] = None +) -> str: """Return the ``agent:`` namespace prefix for a session key. The historical key format is ``agent:main:::...`` where @@ -1081,10 +1094,36 @@ def _session_key_namespace(profile: Optional[str]) -> str: - named profile ``coder`` → ``agent:coder`` — keeps the same positional layout, just a different namespace, so two profiles serving the same platform/chat never collide. + + Multi-account gateways (#8287) reuse the slot the same way: a non-default + bot account is appended as ``@`` (``agent:main@support``, + ``agent:coder@support``), so the same chat reached through two bots yields + two sessions while every positional parser keeps its layout. ``:`` stays + the only separator, and account names are charset-restricted at config + parse time so ``@`` cannot appear inside a name. Readers that map the + namespace back to a profile must strip the suffix via + :func:`split_key_namespace`. """ if not profile or profile == "default": - return "agent:main" - return f"agent:{profile}" + ns = "agent:main" + else: + ns = f"agent:{profile}" + if account and account != "default": + return f"{ns}@{account}" + return ns + + +def split_key_namespace(namespace: str) -> tuple[str, Optional[str]]: + """Split a session-key namespace component into ``(profile_ns, account)``. + + ``main`` → ``("main", None)``; ``main@support`` → ``("main", "support")``. + The account suffix was introduced for multi-account gateways (#8287); + every reader that compares or maps the namespace (profile resolution, + key parsers) must strip it through here rather than assuming the raw + slot equals a profile name. + """ + base, sep, account = (namespace or "").partition("@") + return base, (account or None) if sep else None def build_session_key( @@ -1125,7 +1164,16 @@ def build_session_key( shared session per chat. - Without identifiers, messages fall back to one session per platform/chat_type. """ - ns = _session_key_namespace(profile) + # Account comes from the SOURCE, not a caller parameter: which bot + # received the message is intrinsic to the event, and reading it here + # guarantees the adapter-level guard and the session store derive the + # same key for the same event (per-key guards diverging is the #64934 + # bug class). isinstance-guard bare test fixtures whose auto-attributes + # read as a truthy non-string (AGENTS.md pitfall #17). + account = getattr(source, "account", None) + if not isinstance(account, str): + account = None + ns = _session_key_namespace(profile, account) platform = source.platform.value slack_scope_id = ( str(source.scope_id) @@ -1788,7 +1836,10 @@ def _profile_from_session_key(session_key: Optional[str]) -> Optional[str]: parts = str(session_key).split(":") if len(parts) < 2 or parts[0] != "agent": return None - namespace = parts[1] or "main" + # Strip a multi-account suffix (agent:main@support) — the account is + # not a profile and must not be resolved as one (#8287). + namespace, _account = split_key_namespace(parts[1] or "main") + namespace = namespace or "main" return "default" if namespace == "main" else namespace @staticmethod diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py index 3c65981cbb06..ff57ef79d319 100644 --- a/hermes_cli/setup.py +++ b/hermes_cli/setup.py @@ -2023,6 +2023,28 @@ def _setup_telegram(): if home_channel: save_env_value("TELEGRAM_HOME_CHANNEL", home_channel) + print() + if prompt_yes_no("Add another Telegram bot account (multi-bot gateway)?", False): + print_info("🤖 Each account is a separate bot with isolated sessions.") + print_info(" Names become session/delivery identifiers") + print_info(" (lowercase letters, digits, - and _; e.g. support, sales).") + print_info(" Per-account settings live under platforms.telegram.accounts") + print_info(" in config.yaml; target a bot with telegram@:.") + while True: + account_name = prompt("Account name (leave empty to finish)").strip().lower() + if not account_name: + break + if not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", account_name): + print_error("Names must match [a-z0-9][a-z0-9_-]* — try again.") + continue + account_token = _prompt_telegram_bot_token() + if not account_token: + continue + save_env_value( + f"TELEGRAM_BOT_TOKEN_{account_name.upper()}", account_token + ) + print_success(f"Account {account_name!r} token saved") + # _setup_slack and _write_slack_manifest_and_instruct moved to the slack # plugin: plugins/platforms/slack/adapter.py::interactive_setup (registered diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index f31cd85e6a5c..75b0a932088f 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -977,6 +977,7 @@ def _is_callback_user_authorized( user_id=normalized_user_id, user_name=str(user_name).strip() if user_name else None, thread_id=str(thread_id) if thread_id is not None else None, + account=getattr(self, "account_name", None), ) return bool(auth_fn(source)) except Exception: @@ -1054,6 +1055,7 @@ def _source_from_message_for_auth(self, message: Message): user_id=user_id, user_name=user_name, thread_id=thread_id, + account=getattr(self, "account_name", None), ) def _source_from_reaction_for_auth(self, update): diff --git a/tests/gateway/test_telegram_multi_account_adapters.py b/tests/gateway/test_telegram_multi_account_adapters.py new file mode 100644 index 000000000000..2b1a3bd02430 --- /dev/null +++ b/tests/gateway/test_telegram_multi_account_adapters.py @@ -0,0 +1,303 @@ +"""Per-account adapter registry and lifecycle — #8287. + +Named bot accounts get their own adapter instances, registered in +``_account_adapters[platform][name]`` (the account-dimension mirror of +``_profile_adapters``), each seeing an ordinary derived ``PlatformConfig``. +Resolution fails closed: a stamped account with no registry entry must never +fall back to the default bot — replying out the wrong bot is worse than not +replying. +""" + +import sys +import types +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import gateway.run as gateway_run +from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig +from gateway.session import SessionSource + + +@pytest.fixture() +def runner(monkeypatch, tmp_path): + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + return gateway_run.GatewayRunner(GatewayConfig()) + + +# ── Resolution (authz_mixin) ──────────────────────────────────────────────── + + +def test_default_account_resolves_default_adapter(runner): + default_adapter = MagicMock() + runner.adapters = {Platform.TELEGRAM: default_adapter} + assert runner._authorization_adapter(Platform.TELEGRAM) is default_adapter + assert ( + runner._authorization_adapter(Platform.TELEGRAM, account="default") + is default_adapter + ) + + +def test_named_account_resolves_its_own_adapter(runner): + default_adapter, support_adapter = MagicMock(), MagicMock() + runner.adapters = {Platform.TELEGRAM: default_adapter} + runner._account_adapters = {Platform.TELEGRAM: {"support": support_adapter}} + assert ( + runner._authorization_adapter(Platform.TELEGRAM, account="support") + is support_adapter + ) + + +def test_unknown_account_fails_closed_never_default_bot(runner): + """The wrong-bot rule: a stamped account whose adapter is missing + (failed to connect, misconfigured) must NOT fall back to the default + adapter.""" + runner.adapters = {Platform.TELEGRAM: MagicMock()} + runner._account_adapters = {} + assert runner._authorization_adapter(Platform.TELEGRAM, account="support") is None + + +def test_account_in_secondary_profile_fails_closed(runner): + runner._profile_adapters = {"coder": {Platform.TELEGRAM: MagicMock()}} + assert ( + runner._authorization_adapter( + Platform.TELEGRAM, profile="coder", account="support" + ) + is None + ) + + +def test_adapter_for_source_routes_by_account(runner): + default_adapter, support_adapter = MagicMock(), MagicMock() + runner.adapters = {Platform.TELEGRAM: default_adapter} + runner._account_adapters = {Platform.TELEGRAM: {"support": support_adapter}} + + src_default = SessionSource( + platform=Platform.TELEGRAM, chat_id="1", chat_type="dm" + ) + src_support = SessionSource( + platform=Platform.TELEGRAM, chat_id="1", chat_type="dm", account="support" + ) + assert runner._adapter_for_source(src_default) is default_adapter + assert runner._adapter_for_source(src_support) is support_adapter + + +def test_adapter_for_source_tolerates_bare_fixture(runner): + """SimpleNamespace sources without an ``account`` attr (AGENTS.md + pitfall #17) must resolve like the default account.""" + default_adapter = MagicMock() + runner.adapters = {Platform.TELEGRAM: default_adapter} + bare = SimpleNamespace(platform=Platform.TELEGRAM, profile=None) + assert runner._adapter_for_source(bare) is default_adapter + + +# ── Derived per-account config ────────────────────────────────────────────── + + +def test_account_platform_config_overrides_and_strips_accounts(): + base = PlatformConfig( + enabled=True, + token="123:default", + extra={ + "accounts": {"support": {}}, + "fallback_ips": ["1.2.3.4"], + "allowed_users": [1], + }, + ) + derived = gateway_run.GatewayRunner._account_platform_config( + Platform.TELEGRAM, + base, + "support", + { + "token": "456:support", + "allowed_users": [2, 3], + "home_channel": {"chat_id": "-100999"}, + }, + ) + assert derived.token == "456:support" + assert isinstance(derived.home_channel, HomeChannel) + assert derived.home_channel.chat_id == "-100999" + # Account block overrides platform-level extra; unrelated keys inherit. + assert derived.extra["allowed_users"] == [2, 3] + assert derived.extra["fallback_ips"] == ["1.2.3.4"] + # The accounts map itself never leaks into an account's own config. + assert "accounts" not in derived.extra + # The base config is untouched (dataclasses.replace, not mutation). + assert base.token == "123:default" + assert base.extra["allowed_users"] == [1] + + +# ── Lifecycle (_start_account_adapters) ───────────────────────────────────── + + +def _wire_lifecycle_mocks(runner, connect_results): + created = [] + + def _fake_create(platform, config): + adapter = MagicMock() + adapter.platform = platform + adapter.config = config + adapter.account_name = None + created.append(adapter) + return adapter + + runner._create_adapter = _fake_create + runner._connect_adapter_with_timeout = AsyncMock(side_effect=connect_results) + runner._safe_adapter_disconnect = AsyncMock() + runner._make_adapter_auth_check = MagicMock(return_value=lambda *a, **kw: True) + runner._sync_voice_mode_state_to_adapter = MagicMock() + runner._recover_telegram_topic_thread_id = lambda _s: None + runner._handle_message = AsyncMock() + runner._handle_adapter_fatal_error = AsyncMock() + runner._handle_active_session_busy_message = AsyncMock() + runner.session_store = MagicMock() + runner._busy_text_mode = "full" + return created + + +@pytest.mark.asyncio +async def test_start_account_adapters_registers_connected_accounts(runner): + created = _wire_lifecycle_mocks(runner, [True, True]) + cfg = PlatformConfig( + enabled=True, + token="123:default", + extra={ + "accounts": { + "support": {"token": "456:support"}, + "sales": {"token": "789:sales"}, + } + }, + ) + connected = await runner._start_account_adapters(Platform.TELEGRAM, cfg) + assert connected == 2 + registry = runner._account_adapters[Platform.TELEGRAM] + assert set(registry) == {"support", "sales"} + # Stamped before connect, with the derived (account) token. + assert registry["support"].account_name == "support" + assert registry["support"].config.token == "456:support" + # Wired like a default adapter. + registry["support"].set_message_handler.assert_called_once() + registry["support"].set_authorization_check.assert_called_once() + assert len(created) == 2 + + +@pytest.mark.asyncio +async def test_tokenless_account_is_skipped(runner): + created = _wire_lifecycle_mocks(runner, [True]) + cfg = PlatformConfig( + enabled=True, + token="123:default", + extra={"accounts": {"support": {"display_name": "no token"}}}, + ) + connected = await runner._start_account_adapters(Platform.TELEGRAM, cfg) + assert connected == 0 + assert runner._account_adapters == {} + assert created == [] + + +@pytest.mark.asyncio +async def test_failed_account_connect_skips_without_blocking_others(runner): + _wire_lifecycle_mocks(runner, [False, True]) + cfg = PlatformConfig( + enabled=True, + token="123:default", + extra={ + "accounts": { + "support": {"token": "456:support"}, + "sales": {"token": "789:sales"}, + } + }, + ) + connected = await runner._start_account_adapters(Platform.TELEGRAM, cfg) + assert connected == 1 + registry = runner._account_adapters[Platform.TELEGRAM] + assert set(registry) == {"sales"} + runner._safe_adapter_disconnect.assert_awaited() + + +@pytest.mark.asyncio +async def test_failed_initial_connect_queues_for_reconnect(runner): + """A transient startup connect failure for a named account must enter + the reconnect queue, not vanish — otherwise the account is offline until + the next full gateway restart (teknium1 review finding).""" + _wire_lifecycle_mocks(runner, [False]) # connect returns False (retryable) + cfg = PlatformConfig( + enabled=True, + token="123:default", + extra={"accounts": {"support": {"token": "456:support"}}}, + ) + connected = await runner._start_account_adapters(Platform.TELEGRAM, cfg) + assert connected == 0 + assert (Platform.TELEGRAM, "support") in runner._failed_account_adapters + + +@pytest.mark.asyncio +async def test_no_accounts_is_a_noop(runner): + _wire_lifecycle_mocks(runner, []) + cfg = PlatformConfig(enabled=True, token="123:default") + assert await runner._start_account_adapters(Platform.TELEGRAM, cfg) == 0 + assert runner._account_adapters == {} + + +# ── Inbound stamping (real TelegramAdapter) ──────────────────────────────── + + +def test_telegram_adapter_stamps_account_on_inbound_source(): + from plugins.platforms.telegram.adapter import TelegramAdapter + + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="1:x")) + adapter.account_name = "support" + + message = MagicMock() + message.chat.id = 777 + message.chat.type = "private" + message.chat.title = None + message.from_user.id = 42 + message.from_user.username = "user" + message.from_user.full_name = "User" + message.message_thread_id = None + message.is_topic_message = False + + source = adapter._source_from_message_for_auth(message) + assert source.account == "support" + assert source.platform == Platform.TELEGRAM + + # Default account stays unset — single-bot gateways are unchanged. + default_adapter = TelegramAdapter(PlatformConfig(enabled=True, token="1:y")) + assert default_adapter._source_from_message_for_auth(message).account is None + + +def test_build_source_stamps_account_on_normal_event_path(): + """The normal inbound path — every platform's regular traffic flows + through BasePlatformAdapter.build_source(), NOT the auth helper. If the + account isn't stamped here, named-bot messages get the default session + key and route replies out the default bot (the feature silently no-ops). + Regression guard for that exact miss.""" + from gateway.platforms.base import BasePlatformAdapter + + # build_source is a concrete method on the ABC; call it unbound with a + # minimal stand-in carrying the two attributes it reads off self. + class _Stub: + platform = Platform.TELEGRAM + account_name = "support" + + def _resolve_profile_for_source(self, *a, **kw): + return None + + src = BasePlatformAdapter.build_source( + _Stub(), chat_id="777", chat_type="dm", user_id="42" + ) + assert src.account == "support" + + # Default adapter (no account_name attr set) → account stays None. + class _DefaultStub(_Stub): + account_name = None + + default_src = BasePlatformAdapter.build_source( + _DefaultStub(), chat_id="777", chat_type="dm", user_id="42" + ) + assert default_src.account is None diff --git a/tests/gateway/test_telegram_multi_account_config.py b/tests/gateway/test_telegram_multi_account_config.py new file mode 100644 index 000000000000..94d4fbdca9e7 --- /dev/null +++ b/tests/gateway/test_telegram_multi_account_config.py @@ -0,0 +1,122 @@ +"""Multi-account Telegram configuration parsing — #8287. + +One gateway, N Telegram bot accounts: tokens arrive as +``TELEGRAM_BOT_TOKEN_`` env vars (secrets stay in .env), behavioral +settings as ``platforms.telegram.accounts.`` in config.yaml. The +unsuffixed ``TELEGRAM_BOT_TOKEN`` remains the default account, so single-bot +configurations parse byte-identically to before. +""" + +import pytest + +from gateway.config import Platform, PlatformConfig, load_gateway_config + + +def test_single_bot_config_has_no_accounts_key(monkeypatch, tmp_path): + """Backward compatibility: an unsuffixed token must not grow an + accounts block — existing single-bot setups stay byte-identical.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "123:default-token") + + config = load_gateway_config() + tg = config.platforms[Platform.TELEGRAM] + assert tg.token == "123:default-token" + assert "accounts" not in tg.extra + + +def test_suffixed_env_tokens_declare_accounts(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "123:default-token") + monkeypatch.setenv("TELEGRAM_BOT_TOKEN_SUPPORT", "456:support-token") + monkeypatch.setenv("TELEGRAM_BOT_TOKEN_SALES", "789:sales-token") + + config = load_gateway_config() + tg = config.platforms[Platform.TELEGRAM] + + # Default account untouched. + assert tg.token == "123:default-token" + # Suffix names are lowercased account names carrying only the credential. + accounts = tg.extra["accounts"] + assert accounts["support"]["token"] == "456:support-token" + assert accounts["sales"]["token"] == "789:sales-token" + + +def test_suffixed_token_alone_enables_platform(monkeypatch, tmp_path): + """A gateway configured with only account tokens (no default) still + enables Telegram — the registry decides which accounts to start.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False) + monkeypatch.setenv("TELEGRAM_BOT_TOKEN_SUPPORT", "456:support-token") + + config = load_gateway_config() + tg = config.platforms[Platform.TELEGRAM] + assert tg.enabled + assert tg.token is None + assert tg.extra["accounts"]["support"]["token"] == "456:support-token" + + +def test_empty_suffix_or_value_is_ignored(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "123:default-token") + monkeypatch.setenv("TELEGRAM_BOT_TOKEN_", "999:no-name") + monkeypatch.setenv("TELEGRAM_BOT_TOKEN_EMPTY", "") + + config = load_gateway_config() + tg = config.platforms[Platform.TELEGRAM] + assert "accounts" not in tg.extra + + +def test_yaml_accounts_block_parses_and_normalizes_names(): + cfg = PlatformConfig.from_dict({ + "enabled": True, + "accounts": { + "Support": {"display_name": "Support Bot", "allowed_users": [1, 2]}, + " SALES ": {"home_channel": {"chat_id": "-100123"}}, + }, + }) + accounts = cfg.extra["accounts"] + assert set(accounts) == {"support", "sales"} + assert accounts["support"]["display_name"] == "Support Bot" + assert accounts["support"]["allowed_users"] == [1, 2] + + +def test_yaml_accounts_survive_via_extra_bridge(): + """The shared-key loop can bridge accounts into extra — both routes + normalize identically (the gateway_restart_notification pattern).""" + cfg = PlatformConfig.from_dict({ + "enabled": True, + "extra": {"accounts": {"Support": {"display_name": "S"}}}, + }) + assert cfg.extra["accounts"]["support"]["display_name"] == "S" + + +def test_env_token_merges_into_yaml_account_block(monkeypatch, tmp_path): + """config.yaml declares the behavioral block; .env supplies the token. + The two merge on the same account name.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / "config.yaml").write_text( + "gateway:\n" + " platforms:\n" + " telegram:\n" + " enabled: true\n" + " accounts:\n" + " support:\n" + " display_name: Support Bot\n", + encoding="utf-8", + ) + monkeypatch.setenv("TELEGRAM_BOT_TOKEN_SUPPORT", "456:support-token") + + config = load_gateway_config() + tg = config.platforms[Platform.TELEGRAM] + support = tg.extra["accounts"]["support"] + assert support["token"] == "456:support-token" + assert support.get("display_name") == "Support Bot" + + +def test_accounts_round_trip_through_to_dict(): + cfg = PlatformConfig.from_dict({ + "enabled": True, + "accounts": {"support": {"display_name": "S"}}, + }) + rebuilt = PlatformConfig.from_dict(cfg.to_dict()) + assert rebuilt.extra["accounts"]["support"]["display_name"] == "S" diff --git a/tests/gateway/test_telegram_multi_account_delivery.py b/tests/gateway/test_telegram_multi_account_delivery.py new file mode 100644 index 000000000000..c7aed1cb34e8 --- /dev/null +++ b/tests/gateway/test_telegram_multi_account_delivery.py @@ -0,0 +1,170 @@ +"""Per-account outbound routing and lifecycle recovery — #8287. + +Outbound must honor the account dimension end-to-end: origin replies leave +through the bot the message arrived on, explicit targets can address a named +bot (``telegram@support:123``), a missing account adapter fails closed +(never the default bot), and a dying account adapter is queued for +reconnection without ever touching the default platform slot. +""" + +import sys +import types +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import gateway.run as gateway_run +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.delivery import DeliveryRouter, DeliveryTarget +from gateway.session import SessionSource + + +@pytest.fixture() +def runner(monkeypatch, tmp_path): + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + return gateway_run.GatewayRunner(GatewayConfig()) + + +# ── DeliveryTarget parsing ───────────────────────────────────────────────── + + +def test_parse_account_scoped_chat_target(): + target = DeliveryTarget.parse("telegram@support:123456") + assert target.platform == Platform.TELEGRAM + assert target.account == "support" + assert target.chat_id == "123456" + assert target.is_explicit + + +def test_parse_account_scoped_home_target(): + target = DeliveryTarget.parse("telegram@support") + assert target.platform == Platform.TELEGRAM + assert target.account == "support" + assert target.chat_id is None + + +def test_parse_plain_targets_unchanged(): + assert DeliveryTarget.parse("telegram:123").account is None + assert DeliveryTarget.parse("telegram").account is None + + +def test_origin_target_inherits_account(): + origin = SessionSource( + platform=Platform.TELEGRAM, chat_id="777", chat_type="dm", + account="support", + ) + target = DeliveryTarget.parse("origin", origin=origin) + assert target.is_origin and target.account == "support" + # Default-account origins stay account-less. + plain = SessionSource(platform=Platform.TELEGRAM, chat_id="7", chat_type="dm") + assert DeliveryTarget.parse("origin", origin=plain).account is None + + +def test_to_string_round_trips_account(): + for raw in ("telegram@support:123", "telegram@support", "telegram:123"): + assert DeliveryTarget.parse(raw).to_string() == raw + + +# ── Router resolution ────────────────────────────────────────────────────── + + +def _router(default_adapter=None, account_adapters=None): + router = DeliveryRouter(GatewayConfig()) + if default_adapter is not None: + router.adapters = {Platform.TELEGRAM: default_adapter} + router.account_adapters = account_adapters or {} + return router + + +def test_router_resolves_account_adapter(): + default_adapter, support_adapter = MagicMock(), MagicMock() + router = _router(default_adapter, {Platform.TELEGRAM: {"support": support_adapter}}) + assert ( + router._adapter_for_target(DeliveryTarget.parse("telegram@support:1")) + is support_adapter + ) + assert ( + router._adapter_for_target(DeliveryTarget.parse("telegram:1")) + is default_adapter + ) + + +def test_router_fails_closed_for_unknown_account(): + """Account-addressed content must never leave through the default bot.""" + router = _router(MagicMock()) + assert router._adapter_for_target(DeliveryTarget.parse("telegram@support:1")) is None + + +@pytest.mark.asyncio +async def test_deliver_to_platform_raises_with_account_ref(): + router = _router(MagicMock()) + with pytest.raises(ValueError, match="telegram@support"): + await router._deliver_to_platform( + DeliveryTarget.parse("telegram@support:1"), "content", None + ) + + +# ── Account fatal-error path ─────────────────────────────────────────────── + + +def _fatal_adapter(platform=Platform.TELEGRAM, account="support", retryable=True): + adapter = MagicMock() + adapter.platform = platform + adapter.account_name = account + adapter.fatal_error_code = "network" + adapter.fatal_error_message = "boom" + adapter.fatal_error_retryable = retryable + adapter.config = PlatformConfig(enabled=True, token="456:support") + return adapter + + +@pytest.mark.asyncio +async def test_account_fatal_error_queues_reconnect_not_default_slot(runner): + default_adapter = MagicMock() + runner.adapters = {Platform.TELEGRAM: default_adapter} + adapter = _fatal_adapter() + runner._account_adapters = {Platform.TELEGRAM: {"support": adapter}} + runner._safe_adapter_disconnect = AsyncMock() + + await runner._handle_adapter_fatal_error(adapter) + + # Popped from the account registry, queued under (platform, account). + assert "support" not in (runner._account_adapters.get(Platform.TELEGRAM) or {}) + key = (Platform.TELEGRAM, "support") + assert key in runner._failed_account_adapters + assert runner._failed_account_adapters[key]["config"] is adapter.config + # The default platform slot is untouched — no clobbering, no queueing. + assert runner.adapters[Platform.TELEGRAM] is default_adapter + assert Platform.TELEGRAM not in runner._failed_platforms + runner._safe_adapter_disconnect.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_stale_account_fatal_error_is_ignored(runner): + """A superseded account adapter instance (reconnect already won) must + not evict the healthy replacement.""" + replacement = MagicMock() + runner._account_adapters = {Platform.TELEGRAM: {"support": replacement}} + runner._safe_adapter_disconnect = AsyncMock() + + stale = _fatal_adapter() + await runner._handle_adapter_fatal_error(stale) + + assert runner._account_adapters[Platform.TELEGRAM]["support"] is replacement + assert (Platform.TELEGRAM, "support") not in runner._failed_account_adapters + runner._safe_adapter_disconnect.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_nonretryable_account_fatal_error_not_queued(runner): + adapter = _fatal_adapter(retryable=False) + runner._account_adapters = {Platform.TELEGRAM: {"support": adapter}} + runner._safe_adapter_disconnect = AsyncMock() + + await runner._handle_adapter_fatal_error(adapter) + + assert runner._failed_account_adapters == {} + assert "support" not in (runner._account_adapters.get(Platform.TELEGRAM) or {}) diff --git a/tests/gateway/test_telegram_multi_account_sessions.py b/tests/gateway/test_telegram_multi_account_sessions.py new file mode 100644 index 000000000000..b971465a4c05 --- /dev/null +++ b/tests/gateway/test_telegram_multi_account_sessions.py @@ -0,0 +1,108 @@ +"""Per-account session identity — #8287. + +A gateway hosting multiple bot accounts on one platform must keep their +conversations apart: the same chat reached through two bots is two sessions. +The account rides in the session-key namespace slot (``agent:main@support``) +— the same mechanism profiles use — so every positional parser +(``parts[2] == platform`` etc.) keeps its layout, and single-bot gateways +produce byte-identical keys to before. +""" + +from gateway.config import Platform +from gateway.run import _parse_session_key +from gateway.session import ( + SessionSource, + build_session_key, + split_key_namespace, +) + + +def _source(account=None, **kw): + defaults = dict( + platform=Platform.TELEGRAM, chat_id="777", chat_type="dm", user_id="777" + ) + defaults.update(kw) + return SessionSource(account=account, **defaults) + + +def test_same_chat_two_bots_two_sessions(): + """The #10455-review isolation requirement: identical chat + user via + two different bot accounts must never share a session key.""" + key_default = build_session_key(_source(account=None)) + key_support = build_session_key(_source(account="support")) + key_sales = build_session_key(_source(account="sales")) + assert len({key_default, key_support, key_sales}) == 3 + + +def test_default_account_key_is_byte_identical_to_legacy(): + """Single-bot gateways must keep every key they have ever generated.""" + assert build_session_key(_source(account=None)) == "agent:main:telegram:dm:777" + assert build_session_key(_source(account="default")) == "agent:main:telegram:dm:777" + + +def test_account_key_keeps_positional_layout(): + """The account lives in the namespace slot — platform/chat_type/chat_id + stay at parts[2:5], so positional parsers are unaffected.""" + key = build_session_key(_source(account="support")) + parts = key.split(":") + assert parts[0] == "agent" + assert parts[1] == "main@support" + assert parts[2] == "telegram" + assert parts[3] == "dm" + assert parts[4] == "777" + + +def test_profile_and_account_compose(): + key = build_session_key(_source(account="support"), profile="coder") + assert key.startswith("agent:coder@support:telegram:") + + +def test_group_and_thread_keys_carry_account(): + group_a = build_session_key( + _source(account="support", chat_type="group", chat_id="-100", user_id="9") + ) + group_b = build_session_key( + _source(account=None, chat_type="group", chat_id="-100", user_id="9") + ) + assert group_a != group_b + assert group_a.split(":")[1] == "main@support" + + +def test_source_account_round_trips_serialization(): + src = _source(account="support") + rebuilt = SessionSource.from_dict(src.to_dict()) + assert rebuilt.account == "support" + # Default account stays wire-invisible (no key emitted), like profile. + assert "account" not in _source(account=None).to_dict() + + +def test_split_key_namespace(): + assert split_key_namespace("main") == ("main", None) + assert split_key_namespace("main@support") == ("main", "support") + assert split_key_namespace("coder@support") == ("coder", "support") + assert split_key_namespace("") == ("", None) + + +def test_profile_resolution_ignores_account_suffix(): + from gateway.session import SessionStore + + resolve = SessionStore._profile_from_session_key + assert resolve("agent:main:telegram:dm:1") == "default" + assert resolve("agent:main@support:telegram:dm:1") == "default" + assert resolve("agent:coder@support:telegram:dm:1") == "coder" + + +def test_parse_session_key_accepts_account_namespace(): + parsed = _parse_session_key("agent:main@support:telegram:dm:777:42") + assert parsed == { + "platform": "telegram", + "chat_type": "dm", + "chat_id": "777", + "account": "support", + "thread_id": "42", + } + # Default-namespace behavior unchanged. + legacy = _parse_session_key("agent:main:telegram:dm:777") + assert legacy == {"platform": "telegram", "chat_type": "dm", "chat_id": "777"} + # Named-profile keys stay excluded, as before. + assert _parse_session_key("agent:coder:telegram:dm:777") is None diff --git a/tests/gateway/test_telegram_multi_account_targets.py b/tests/gateway/test_telegram_multi_account_targets.py new file mode 100644 index 000000000000..365af81265c5 --- /dev/null +++ b/tests/gateway/test_telegram_multi_account_targets.py @@ -0,0 +1,92 @@ +"""Cron delivery targets and home-channel broadcasts per account — #8287. + +Cron ``deliver`` strings can address a named bot (``telegram@support:123``), +and gateway home-channel broadcasts (startup/shutdown notices) reach every +account's own home channel, not just the platform default's. +""" + +import sys +import types +from unittest.mock import MagicMock + +import pytest + +import gateway.run as gateway_run +from cron.scheduler import _resolve_single_delivery_target +from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig + + +@pytest.fixture() +def runner(monkeypatch, tmp_path): + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + return gateway_run.GatewayRunner(GatewayConfig()) + + +# ── Cron deliver-string parsing ──────────────────────────────────────────── + + +def test_cron_target_parses_account_scoped_chat(): + target = _resolve_single_delivery_target({}, "telegram@support:123456") + assert target == { + "platform": "telegram", + "chat_id": "123456", + "thread_id": None, + "account": "support", + } + + +def test_cron_target_account_name_is_lowercased(): + target = _resolve_single_delivery_target({}, "telegram@SUPPORT:123456") + assert target["account"] == "support" + + +def test_cron_target_plain_form_carries_no_account(): + target = _resolve_single_delivery_target({}, "telegram:123456") + assert "account" not in target + assert target["platform"] == "telegram" + + +def test_cron_target_account_with_thread(): + target = _resolve_single_delivery_target({}, "telegram@sales:-100777:42") + assert target["account"] == "sales" + assert target["chat_id"] == "-100777" + assert target["thread_id"] == "42" + + +# ── Home-channel broadcast iteration ─────────────────────────────────────── + + +def test_broadcast_iter_includes_account_homes(runner): + default_adapter = MagicMock() + runner.adapters = {Platform.TELEGRAM: default_adapter} + runner.config.platforms[Platform.TELEGRAM] = PlatformConfig( + enabled=True, + home_channel=HomeChannel(platform=Platform.TELEGRAM, chat_id="111", name="Home"), + ) + + support_adapter = MagicMock() + support_adapter.config = PlatformConfig( + enabled=True, + home_channel=HomeChannel(platform=Platform.TELEGRAM, chat_id="222", name="Support Home"), + ) + runner._account_adapters = {Platform.TELEGRAM: {"support": support_adapter}} + + entries = list(runner._iter_live_adapters_with_home(snapshot=True)) + by_adapter = {id(adapter): home for _p, adapter, home in entries} + + assert len(entries) == 2 + assert by_adapter[id(default_adapter)].chat_id == "111" + assert by_adapter[id(support_adapter)].chat_id == "222" + + +def test_broadcast_iter_tolerates_account_without_home(runner): + runner.adapters = {} + bare = MagicMock() + bare.config = PlatformConfig(enabled=True) # no home_channel + runner._account_adapters = {Platform.TELEGRAM: {"support": bare}} + entries = list(runner._iter_live_adapters_with_home()) + assert len(entries) == 1 + assert entries[0][2] is None # callers skip home-less adapters diff --git a/tests/tools/test_send_message_account_routing.py b/tests/tools/test_send_message_account_routing.py new file mode 100644 index 000000000000..94aed56337d6 --- /dev/null +++ b/tests/tools/test_send_message_account_routing.py @@ -0,0 +1,172 @@ +"""Per-account `send_message` routing — #8287. + +The multi-account gateway lets one process host several bots. `send_message` +targets could not address them: `telegram@support:123` was rejected, so the +`send` consumer was the one reviewed path that stayed account-blind. These +tests cover the target parsing, the account-config derivation shared with the +gateway's adapter startup, and the fail-closed behavior for an unknown or +token-less account (never silently fall back to the default bot, which would +deliver to the wrong audience). +""" + +import pytest + +from gateway.config import ( + HomeChannel, + Platform, + PlatformConfig, + derive_account_platform_config, + resolve_platform_account, +) + + +# --------------------------------------------------------------------------- +# target parsing +# --------------------------------------------------------------------------- + + +def test_plain_platform_has_no_account(): + assert resolve_platform_account("telegram") == ("telegram", None) + + +def test_named_account_is_split_and_lowercased(): + assert resolve_platform_account("telegram@Support") == ("telegram", "support") + + +def test_default_account_spelling_resolves_to_none(): + """`@default` means the platform's default bot, spelled the same as omitting it.""" + assert resolve_platform_account("telegram@default") == ("telegram", None) + + +def test_empty_account_suffix_is_ignored(): + assert resolve_platform_account("telegram@") == ("telegram", None) + + +def test_empty_input_is_safe(): + assert resolve_platform_account("") == ("", None) + + +# --------------------------------------------------------------------------- +# derived per-account config (shared with the gateway's adapter startup) +# --------------------------------------------------------------------------- + + +def test_account_token_overrides_and_accounts_map_is_stripped(): + base = PlatformConfig( + enabled=True, + token="123:default", + extra={ + "accounts": {"support": {"token": "456:support"}}, + "fallback_ips": ["1.2.3.4"], + }, + ) + derived = derive_account_platform_config( + Platform.TELEGRAM, base, {"token": "456:support"} + ) + + assert derived.token == "456:support" + assert derived.extra["fallback_ips"] == ["1.2.3.4"] # platform extra inherited + # A derived config can never recurse into another account. + assert "accounts" not in derived.extra + # The base config is untouched (dataclasses.replace, not mutation). + assert base.token == "123:default" + assert "accounts" in base.extra + + +def test_account_home_channel_platform_is_implicit(): + base = PlatformConfig(enabled=True, token="123:default") + derived = derive_account_platform_config( + Platform.TELEGRAM, base, {"home_channel": {"chat_id": "-100999"}} + ) + assert isinstance(derived.home_channel, HomeChannel) + assert derived.home_channel.chat_id == "-100999" + assert derived.home_channel.platform == Platform.TELEGRAM + + +def test_account_block_overrides_platform_extra(): + base = PlatformConfig( + enabled=True, token="t", extra={"allowed_users": [1], "keep": "yes"} + ) + derived = derive_account_platform_config( + Platform.TELEGRAM, base, {"allowed_users": [2, 3]} + ) + assert derived.extra["allowed_users"] == [2, 3] + assert derived.extra["keep"] == "yes" + + +def test_empty_account_block_inherits_everything(): + base = PlatformConfig(enabled=True, token="123:default", extra={"a": 1}) + derived = derive_account_platform_config(Platform.TELEGRAM, base, {}) + assert derived.token == "123:default" + assert derived.extra["a"] == 1 + + +def test_runner_helper_delegates_to_the_shared_function(): + """The gateway's account startup and this send path must resolve an + account identically — one implementation, two callers.""" + from gateway.run import GatewayRunner + + base = PlatformConfig( + enabled=True, token="123:default", extra={"accounts": {"s": {}}} + ) + block = {"token": "456:support", "home_channel": {"chat_id": "-100777"}} + + via_runner = GatewayRunner._account_platform_config( + Platform.TELEGRAM, base, "support", block + ) + via_shared = derive_account_platform_config(Platform.TELEGRAM, base, block) + + assert via_runner.token == via_shared.token == "456:support" + assert via_runner.home_channel.chat_id == via_shared.home_channel.chat_id + assert via_runner.extra == via_shared.extra + + +# --------------------------------------------------------------------------- +# fail-closed on a bad account (never fall back to the default bot) +# --------------------------------------------------------------------------- + + +def _send(target, monkeypatch, accounts=None, default_token="123:default"): + """Drive send_message_tool far enough to hit account resolution, with the + gateway config stubbed so no network or live adapter is involved.""" + import tools.send_message_tool as smt + + extra = {"accounts": accounts} if accounts is not None else {} + pconfig = PlatformConfig(enabled=True, token=default_token, extra=extra) + + class _Cfg: + platforms = {Platform.TELEGRAM: pconfig} + + def get_home_channel(self, platform): + return None + + monkeypatch.setattr(smt, "load_gateway_config", lambda: _Cfg(), raising=False) + import gateway.config as gwc + + monkeypatch.setattr(gwc, "load_gateway_config", lambda: _Cfg(), raising=False) + return smt.send_message_tool({"target": target, "message": "hi"}) + + +def test_unknown_account_is_rejected_with_the_configured_list(monkeypatch): + out = _send( + "telegram@nope:123", monkeypatch, accounts={"support": {"token": "t"}} + ) + assert "nope" in out + assert "support" in out # tells the user what IS configured + assert "TELEGRAM_BOT_TOKEN_NOPE" in out # and how to add it + + +def test_account_with_no_token_is_rejected(monkeypatch): + out = _send( + "telegram@support:123", monkeypatch, accounts={"support": {"display_name": "S"}} + ) + assert "support" in out + assert "no token" in out.lower() + assert "TELEGRAM_BOT_TOKEN_SUPPORT" in out + + +def test_account_target_on_platform_without_accounts_is_rejected(monkeypatch): + """Fail closed rather than silently using the default bot's credential.""" + out = _send("telegram@support:123", monkeypatch, accounts=None) + assert "support" in out + assert "none" in out.lower() # no accounts configured diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index cf93756121c6..14e52768dcf7 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -225,7 +225,7 @@ async def _send_telegram_message_with_retry(bot, *, attempts: int = 3, **kwargs) }, "target": { "type": "string", - "description": "Delivery target. Format: 'platform' (uses home channel), 'platform:#channel-name', 'platform:chat_id', or 'platform:chat_id:thread_id' for Telegram topics and Discord threads. Examples: 'telegram', 'telegram:-1001234567890:17585', 'discord:999888777:555444333', 'discord:#bot-home', 'slack:#engineering', 'signal:+155****4567', 'matrix:!roomid:server.org', 'matrix:@user:server.org', 'ntfy:alerts-channel' (explicit ntfy topic), 'yuanbao:direct:' (DM), 'yuanbao:group:' (group chat)" + "description": "Delivery target. Format: 'platform' (uses home channel), 'platform:#channel-name', 'platform:chat_id', or 'platform:chat_id:thread_id' for Telegram topics and Discord threads. Examples: 'telegram', 'telegram:-1001234567890:17585', 'discord:999888777:555444333', 'discord:#bot-home', 'slack:#engineering', 'signal:+155****4567', 'matrix:!roomid:server.org', 'matrix:@user:server.org', 'ntfy:alerts-channel' (explicit ntfy topic), 'yuanbao:direct:' (DM), 'yuanbao:group:' (group chat). On a multi-bot gateway, prefix a named account with '@' to send through that bot's credential: 'telegram@support:123456789', or 'telegram@support' for that account's own home channel." }, "message": { "type": "string", @@ -371,6 +371,15 @@ def _handle_send(args): thread_id = None prepare_send_message_platforms() + + # Named-bot-account targets (#8287) — "telegram@support:123" sends through + # the support bot's credential. Split the account off before any platform + # lookup: downstream maps and the Platform enum know "telegram", not + # "telegram@support". The account's PlatformConfig is derived further down, + # once the platform's own config has been resolved. + from gateway.config import resolve_platform_account + + platform_name, account_name = resolve_platform_account(platform_name) if target_ref: chat_id, thread_id, resolution_error = resolve_send_target( platform_name, target_ref @@ -424,6 +433,36 @@ def _handle_send(args): else: return tool_error(f"Platform '{platform_name}' is not configured. Set up credentials in ~/.hermes/config.yaml or environment variables.") + # Per-account send (#8287): swap in the named account's derived config so + # the rest of this path — credential, home channel, platform-extra + # settings — belongs to that bot. Fail closed with an actionable error + # rather than silently falling back to the default account, which would + # deliver to the wrong audience. + if account_name: + _accounts = (pconfig.extra or {}).get("accounts") + _account_block = ( + _accounts.get(account_name) if isinstance(_accounts, dict) else None + ) + if not isinstance(_account_block, dict): + _known = ( + ", ".join(sorted(_accounts)) if isinstance(_accounts, dict) and _accounts else "none" + ) + return tool_error( + f"Unknown {platform_name} account '{account_name}'. " + f"Configured accounts: {_known}. Declare the account's token " + f"as {platform_name.upper()}_BOT_TOKEN_{account_name.upper()} " + f"in .env (and any per-account settings under " + f"platforms.{platform_name}.accounts.{account_name})." + ) + if not _account_block.get("token"): + return tool_error( + f"{platform_name} account '{account_name}' has no token. Set " + f"{platform_name.upper()}_BOT_TOKEN_{account_name.upper()} in .env." + ) + from gateway.config import derive_account_platform_config + + pconfig = derive_account_platform_config(platform, pconfig, _account_block) + from gateway.platforms.base import BasePlatformAdapter # Capture [[as_document]] directive before extract_media strips it. @@ -438,7 +477,14 @@ def _handle_send(args): used_home_channel = False if not chat_id: - home = config.get_home_channel(platform) + # A named account's own home channel wins over the platform default + # (#8287): "telegram@support" with no chat must reach the support + # bot's home, not the default bot's. + home = ( + pconfig.home_channel + if account_name and pconfig.home_channel + else config.get_home_channel(platform) + ) if not home and platform_name == "weixin": wx_home = os.getenv("WEIXIN_HOME_CHANNEL", "").strip() if wx_home: