From f5e72803ba36a5db13e44a3511ffa84598322daa Mon Sep 17 00:00:00 2001 From: hotragn Date: Sun, 19 Jul 2026 05:16:29 -0500 Subject: [PATCH 1/8] feat(gateway): parse multi-account Telegram config (#8287) First slice of the account-aware gateway: configuration surface only, no runtime behavior change. - TELEGRAM_BOT_TOKEN_ env vars declare additional bot accounts (lowercased names); the unsuffixed TELEGRAM_BOT_TOKEN remains the default account, so single-bot setups parse byte-identically. Tokens are secrets: env/.env is their supported home. - platforms.telegram.accounts. in config.yaml carries the behavioral per-account settings (display names, allowlists, home channels) and merges with env tokens on the account name; the block arrives top-level or bridged into extra (the same two-route pattern as gateway_restart_notification) and round-trips through to_dict. Registry, session-key, and routing slices follow in this branch per the acceptance architecture in the #10455 review. --- gateway/config.py | 46 +++++++ .../test_telegram_multi_account_config.py | 122 ++++++++++++++++++ 2 files changed, 168 insertions(+) create mode 100644 tests/gateway/test_telegram_multi_account_config.py diff --git a/gateway/config.py b/gateway/config.py index a00fa0f9a1ca..397233735f49 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -731,6 +731,26 @@ 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 _acct_key: + _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 +1931,32 @@ 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 + _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/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" From 4737cf93c9a2c08c6bb619a1c379fc09dfdc379d Mon Sep 17 00:00:00 2001 From: hotragn Date: Sun, 19 Jul 2026 05:28:10 -0500 Subject: [PATCH 2/8] feat(gateway): per-account session identity (#8287) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second slice: the same chat reached through two bot accounts is two sessions. - SessionSource.account carries which bot received the message (stamped by the adapter in the upcoming registry slice); wire-invisible when unset, serialized like profile. - The account rides in the session-key NAMESPACE slot — the same mechanism profiles use: agent:main@support / agent:coder@support. Positional parsers keep their layout (parts[2] == platform), and single-bot gateways produce byte-identical keys (locked by test). - build_session_key reads the account from the SOURCE, not a caller parameter, so 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, and this keeps that door shut by construction. - The two namespace readers are account-aware via a shared helper: _profile_from_session_key strips the suffix instead of resolving 'main@support' as a profile name, and _parse_session_key accepts the suffixed default namespace (named-profile keys stay excluded). - Account names are charset-restricted at config parse ([a-z0-9][a-z0-9_-]*) so ':' and '@' can never reach a key. Includes the #10455-review isolation test: same chat + user via two accounts yields distinct keys. --- gateway/config.py | 25 +++- gateway/run.py | 10 +- gateway/session.py | 58 +++++++++- .../test_telegram_multi_account_sessions.py | 108 ++++++++++++++++++ 4 files changed, 193 insertions(+), 8 deletions(-) create mode 100644 tests/gateway/test_telegram_multi_account_sessions.py diff --git a/gateway/config.py b/gateway/config.py index 397233735f49..e6cd6c392a59 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 @@ -746,8 +747,20 @@ def from_dict(cls, data: Dict[str, Any]) -> "PlatformConfig": _norm_accounts: Dict[str, Any] = {} for _acct_name, _acct_block in _accounts.items(): _acct_key = str(_acct_name).strip().lower() - if _acct_key: - _norm_accounts[_acct_key] = _coerce_dict(_acct_block) + 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 @@ -1949,6 +1962,14 @@ def _enable_from_env(platform: Platform) -> PlatformConfig: _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): diff --git a/gateway/run.py b/gateway/run.py index af607198ba8c..853f75110312 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 diff --git a/gateway/session.py b/gateway/session.py index 0121518152d1..2d7c801fbf12 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,13 @@ 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). ``getattr`` guards bare test fixtures (AGENTS.md pitfall). + account = getattr(source, "account", None) + ns = _session_key_namespace(profile, account) platform = source.platform.value slack_scope_id = ( str(source.scope_id) @@ -1788,7 +1833,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/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 From dc33d6b1ed23212238f505226cfeef78bf120dce Mon Sep 17 00:00:00 2001 From: hotragn Date: Sun, 19 Jul 2026 05:39:51 -0500 Subject: [PATCH 3/8] feat(gateway): account-aware adapter registry + Telegram inbound stamping (#8287) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third slice: one adapter instance per named bot account. - _account_adapters[platform][name] mirrors the proven _profile_adapters two-level registry: self.adapters stays the default account's map, so every existing consumer is untouched when no named accounts exist. - _authorization_adapter gains the account dimension with the same fail-closed contract as profiles: a stamped account with no registry entry resolves to None rather than the default bot — replying out the wrong bot is worse than not replying. _adapter_for_source reads source.account, so inbound routing follows the stamp automatically. - Each account adapter sees an ordinary derived PlatformConfig (its own token, its own home_channel with the platform implicit, account-block settings overriding platform extra, the accounts map stripped) — adapter internals stay account-agnostic. - BasePlatformAdapter.account_name identifies the serving account; the Telegram adapter copies it onto every inbound SessionSource, which is where the per-account session keys from the previous slice light up. - Startup: named accounts start after the default adapter, each independent (a failed account never blocks the others). Accounts-only configurations (tokens with no default credential) skip the doomed token-less default connect and start named accounts directly. Per-account reconnect queueing, delivery/status/cron consumers, and setup UX land in the remaining slices. --- gateway/authz_mixin.py | 20 ++ gateway/platforms/base.py | 7 + gateway/run.py | 157 ++++++++++- plugins/platforms/telegram/adapter.py | 2 + .../test_telegram_multi_account_adapters.py | 255 ++++++++++++++++++ 5 files changed, 438 insertions(+), 3 deletions(-) create mode 100644 tests/gateway/test_telegram_multi_account_adapters.py diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index fae4b74a5e11..78a40aa3a530 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,26 @@ 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 + account_name = (account or "").strip() 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 +140,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 +171,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/platforms/base.py b/gateway/platforms/base.py index 1a044d1521f7..070786257396 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``). diff --git a/gateway/run.py b/gateway/run.py index 853f75110312..7a91ea84d5ea 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -6480,6 +6480,13 @@ 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]] = {} self._warn_if_docker_media_delivery_is_risky() _gateway_runner_ref = _weakref.ref(self) @@ -12066,7 +12073,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 @@ -12129,6 +12163,12 @@ async def start(self) -> bool: retrying_since=None, ) logger.info("✓ %s connected", platform.value) + # Named bot accounts on this platform (#8287) start after + # the default account; each is independent and a failed + # account never blocks the others. + connected_count += await self._start_account_adapters( + platform, platform_config + ) else: logger.warning("✗ %s failed to connect", platform.value) # Defensive cleanup: a failed connect() may have @@ -14892,9 +14932,120 @@ 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). + + The account adapter sees an ordinary PlatformConfig — its own token, + its own home_channel, and account-block settings overriding the + platform-level extra — so adapter internals stay account-agnostic. + The ``accounts`` map itself is stripped from the derived extra. + """ + import dataclasses as _dc + + from gateway.config import HomeChannel as _HomeChannel + + 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, + ) + + 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 + 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 + 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) + continue + if not success: + logger.warning( + "✗ %s account %r failed to connect", platform.value, account_name + ) + await self._safe_adapter_disconnect(adapter, platform) + 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/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..e5a3d2039732 --- /dev/null +++ b/tests/gateway/test_telegram_multi_account_adapters.py @@ -0,0 +1,255 @@ +"""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_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 From 45a9ba85d02a40b8a0a1bcadd89456d6c1862398 Mon Sep 17 00:00:00 2001 From: hotragn Date: Sun, 19 Jul 2026 05:49:16 -0500 Subject: [PATCH 4/8] feat(gateway): per-account delivery routing + fatal-error recovery (#8287) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fourth slice: the outbound half and lifecycle recovery. - DeliveryTarget grows the account dimension: 'telegram@support:123' addresses a chat through the support bot, 'telegram@support' its home channel, and origin targets inherit the account the message arrived on — replies always leave through the bot that received them. - DeliveryRouter resolves account targets through the account registry fail-closed: account-addressed content never leaves through the default bot. Registry sync rides the existing three adapter-sync sites (shared references, so registrations propagate live). - Fatal errors: the stale-owner guard in _handle_adapter_fatal_error sees the DEFAULT adapter occupying the platform slot and would silently ignore a dying named account — no reconnect, no status, no log. Named accounts now take their own path: same contract, scoped to the account registry (stale-guard against their own slot, pop + disconnect, retryable failures queued under (platform, account)), and an account's death never touches the default slot or the all-platforms-down shutdown logic. - Reconnect watcher gains a named-account pass mirroring the platform pass (same capped exponential backoff); success re-registers into _account_adapters, never the platform slot. Adapter wiring is shared between account startup and account reconnect via _wire_account_adapter so the two can never drift. Remaining slice: cron/home-channel account targets in the runner's target-resolution paths, setup UX, and example-config docs. --- gateway/delivery.py | 81 +++++++-- gateway/run.py | 163 ++++++++++++++++- .../test_telegram_multi_account_delivery.py | 170 ++++++++++++++++++ 3 files changed, 393 insertions(+), 21 deletions(-) create mode 100644 tests/gateway/test_telegram_multi_account_delivery.py 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/run.py b/gateway/run.py index 7a91ea84d5ea..9e561aa203d1 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -6487,6 +6487,10 @@ def __init__(self, config: Optional[GatewayConfig] = None): # 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) @@ -7816,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 = self._account_adapters.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. @@ -7959,6 +8020,15 @@ 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) + if _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 @@ -8007,6 +8077,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", {} + ) # Queue retryable failures BEFORE any disconnect await (#80598). # A half-dead transport can wedge native close() (or swallow @@ -12388,6 +12461,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 = self._account_adapters self._wire_teams_pipeline_runtime() self._running = True @@ -13485,6 +13559,74 @@ 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. + for _acct_key in list(self._failed_account_adapters.keys()): + 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 @@ -13586,6 +13728,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 = self._account_adapters del self._failed_platforms[platform] self._update_platform_runtime_status( platform.value, @@ -14973,6 +15116,18 @@ def _account_platform_config( extra=merged_extra, ) + 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: @@ -15014,13 +15169,7 @@ async def _start_account_adapters( ) continue adapter.account_name = account_name - 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 + 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) 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 {}) From df207ca6055b4341758cf0a70074cbcb7b838ad9 Mon Sep 17 00:00:00 2001 From: hotragn Date: Sun, 19 Jul 2026 06:18:46 -0500 Subject: [PATCH 5/8] feat(gateway): cron account targets, per-account home broadcasts, setup UX (#8287) Final slice: the remaining consumers from the #10455 review checklist, plus the user-facing surface. - Cron 'deliver' strings accept named-account targets: telegram@support:123 routes through the support bot. The account splits off before platform validation (downstream maps know 'telegram', not 'telegram@support'), rides the resolved-target dict, and reaches DeliveryTarget(account=...). The delivery router syncs the gateway's account registry via the module-level runner weakref gateway/run.py already keeps for module consumers; with no gateway running (CLI cron), account targets fail closed with the router's 'No adapter configured' error. - Home-channel broadcasts (startup notice + shutdown notification) reach every account's own home channel: a shared _iter_live_adapters_with_home yields (platform, adapter, home) for default adapters (platform-level home) and account adapters (their derived config's home), preserving the snapshot-before-iterate shutdown lesson. - hermes gateway setup gains an optional multi-bot stanza after the home channel step: name-validated accounts, token prompts with the existing regex/retry loop, saved as TELEGRAM_BOT_TOKEN_ in .env. - cli-config.yaml.example documents platforms.telegram.accounts with the name charset, the env token convention, and the delivery syntax. --- cli-config.yaml.example | 16 ++++ cron/scheduler.py | 30 +++++- gateway/run.py | 22 ++++- hermes_cli/setup.py | 22 +++++ .../test_telegram_multi_account_targets.py | 92 +++++++++++++++++++ 5 files changed, 179 insertions(+), 3 deletions(-) create mode 100644 tests/gateway/test_telegram_multi_account_targets.py 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/run.py b/gateway/run.py index 9e561aa203d1..8208024f368a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -10355,8 +10355,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 @@ -15116,6 +15115,25 @@ def _account_platform_config( extra=merged_extra, ) + 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 _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 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/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 From 87903ff2d0109e42b162051963b3fc0711023830 Mon Sep 17 00:00:00 2001 From: hotragn Date: Sun, 19 Jul 2026 07:46:29 -0500 Subject: [PATCH 6/8] fix(gateway): stamp account on the real inbound path + independent account lifecycle (#8287) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the #67455 review — three correct findings, plus the CI failures the review surfaced. - Account is now stamped in BasePlatformAdapter.build_source(), the single construction site every platform's NORMAL inbound event flows through. The prior stamps were only on Telegram auth-helper sources, so ordinary named-bot traffic kept account=None and routed through the default session key and default egress adapter — the feature silently no-op'd for real messages. Regression test added for this exact path. - Named-account adapters start independently of the default adapter's connect outcome (moved out of the 'if success' block): a bad or absent default token no longer keeps healthy named bots offline. - Retryable initial account-connect failures now queue for background reconnection via a shared _queue_account_reconnect helper (the same queue the fatal-error path and reconnect watcher use), so a transient startup blip is retried instead of dropped until the next restart. - hermes send rejects @account targets with an actionable error rather than an opaque 'Unknown platform'; per-account send routing is a scoped follow-up. CI-failure fix: a MagicMock/SimpleNamespace source auto-creates a truthy 'account' attribute (AGENTS.md pitfall #17, already guarded for 'profile'), which tripped fail-closed account resolution. The account read is now isinstance-coerced to str-or-None in _authorization_adapter and build_session_key, so any source built without an explicit account reads as the default. --- gateway/authz_mixin.py | 6 ++- gateway/platforms/base.py | 9 +++- gateway/run.py | 46 +++++++++++++++--- gateway/session.py | 5 +- .../test_telegram_multi_account_adapters.py | 48 +++++++++++++++++++ tools/send_message_tool.py | 12 +++++ 6 files changed, 117 insertions(+), 9 deletions(-) diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 78a40aa3a530..aa8e8e71dced 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -113,7 +113,11 @@ def _authorization_adapter( if not platform: return None profile_name = (profile or "").strip() or None - account_name = (account 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": diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 070786257396..26772f51a6c0 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -7100,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, @@ -7114,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 8208024f368a..b6a9a3dcefa0 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -12235,12 +12235,6 @@ async def start(self) -> bool: retrying_since=None, ) logger.info("✓ %s connected", platform.value) - # Named bot accounts on this platform (#8287) start after - # the default account; each is independent and a failed - # account never blocks the others. - connected_count += await self._start_account_adapters( - platform, platform_config - ) else: logger.warning("✗ %s failed to connect", platform.value) # Defensive cleanup: a failed connect() may have @@ -12330,6 +12324,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 @@ -15134,6 +15139,25 @@ def _iter_live_adapters_with_home(self, snapshot: bool = False): 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 @@ -15197,12 +15221,22 @@ async def _start_account_adapters( 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) diff --git a/gateway/session.py b/gateway/session.py index 2d7c801fbf12..41c306d02b92 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1168,8 +1168,11 @@ def build_session_key( # 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). ``getattr`` guards bare test fixtures (AGENTS.md pitfall). + # 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 = ( diff --git a/tests/gateway/test_telegram_multi_account_adapters.py b/tests/gateway/test_telegram_multi_account_adapters.py index e5a3d2039732..2b1a3bd02430 100644 --- a/tests/gateway/test_telegram_multi_account_adapters.py +++ b/tests/gateway/test_telegram_multi_account_adapters.py @@ -219,6 +219,22 @@ async def test_failed_account_connect_skips_without_blocking_others(runner): 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, []) @@ -253,3 +269,35 @@ def test_telegram_adapter_stamps_account_on_inbound_source(): # 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/tools/send_message_tool.py b/tools/send_message_tool.py index cf93756121c6..a51c3ad65619 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -371,6 +371,18 @@ def _handle_send(args): thread_id = None prepare_send_message_platforms() + + # Named-bot-account targets (#8287) — e.g. "telegram@support:123" — are + # supported for cron delivery and inbound routing, but this send path + # always uses the platform's default account. Reject the @account form + # with an actionable error rather than the opaque "Unknown platform". + if "@" in platform_name: + _base, _, _acct = platform_name.partition("@") + return tool_error( + f"send targets the default {_base} account; per-account send " + f"(@{_acct}) is not supported here yet. Send via the default " + f"account, or use a cron job with deliver='{platform_name}:'." + ) if target_ref: chat_id, thread_id, resolution_error = resolve_send_target( platform_name, target_ref From 95e3beaf1f3a25711ff2a1083e54b5187dfa8bbf Mon Sep 17 00:00:00 2001 From: hotragn Date: Sun, 19 Jul 2026 08:03:04 -0500 Subject: [PATCH 7/8] fix(gateway): make account registries bare-runner safe (#8287) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI caught real regressions the account lifecycle introduced: the reconnect watcher and fatal-error handler are exercised by tests that build a partially-constructed GatewayRunner (no __init__), so they lack _account_adapters / _failed_account_adapters — and a MagicMock adapter auto-creates a truthy account_name (AGENTS.md pitfall #17), routing a default adapter's fatal error down the named-account path. - Reconnect watcher iterates _failed_account_adapters via getattr, so a bare runner sees an empty account queue instead of AttributeError. - _handle_adapter_fatal_error isinstance-guards account_name: only a real str routes to the account fatal path; a Mock reads as the default adapter. - All three delivery_router.account_adapters sync lines and the account-fatal registry read use getattr defaults. Verified: the CI-failing suites (test_platform_reconnect, test_platform_reconnect_fd_leak, test_discord_liveness) plus the runner lifecycle files all green in isolation. --- gateway/run.py | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index b6a9a3dcefa0..73e8908981ed 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -7834,7 +7834,7 @@ async def _handle_account_adapter_fatal_error( still be healthy. """ platform = adapter.platform - registry = self._account_adapters.get(platform) or {} + 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( @@ -8025,7 +8025,9 @@ async def _handle_adapter_fatal_error_impl(self, adapter: BasePlatformAdapter) - # 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) - if _account_name: + # 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 @@ -8077,8 +8079,8 @@ 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", {} + self.delivery_router.account_adapters = ( + getattr(self, "_account_adapters", {}) or {} ) # Queue retryable failures BEFORE any disconnect await (#80598). @@ -12465,7 +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 = self._account_adapters + self.delivery_router.account_adapters = getattr(self, "_account_adapters", {}) or {} self._wire_teams_pipeline_runtime() self._running = True @@ -13567,8 +13569,10 @@ async def _platform_reconnect_watcher(self) -> None: # 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. - for _acct_key in list(self._failed_account_adapters.keys()): + # 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 @@ -13732,7 +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 = self._account_adapters + self.delivery_router.account_adapters = getattr(self, "_account_adapters", {}) or {} del self._failed_platforms[platform] self._update_platform_runtime_status( platform.value, From 10df63ad4395cbfb3994e7e3c83386bd07b2a6cd Mon Sep 17 00:00:00 2001 From: hotragn Date: Wed, 29 Jul 2026 23:23:03 -0500 Subject: [PATCH 8/8] feat(gateway): per-account send_message routing (#8287) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the last account-blind consumer from teknium1's review: send_message rejected 'telegram@support:123', so the one path a reviewer flagged as incomplete stayed incomplete. send_message already builds its own PlatformConfig from gateway config rather than reaching a live adapter registry, so per-account send needs no new plumbing — just the account's config. The target's platform segment now splits into (platform, account) BEFORE any platform lookup (downstream maps and the Platform enum know 'telegram', not 'telegram@support'), and the named account's derived PlatformConfig replaces the platform's for the rest of the path: its own token, its own home channel, its own platform-extra overrides. 'telegram@support' with no chat reaches the SUPPORT bot's home channel, not the default bot's. The derivation is no longer duplicated: GatewayRunner._account_platform_config now delegates to gateway.config.derive_account_platform_config, shared with this send path, so account-adapter startup and send resolve an account's token/home/extra identically. resolve_platform_account() is the one place 'platform[@account]' is parsed, with '@default' and '@' both meaning the default account. Fails CLOSED on a bad account — unknown account, or one with no token — naming the configured accounts and the exact env var to set, rather than silently falling back to the default bot's credential and delivering to the wrong audience. Tool schema documents the '@account' target form. 13 tests: parsing (plain / named / '@default' / empty / empty-input), derivation (token override, accounts-map stripped, implicit home-channel platform, extra override, empty block, base config untouched), the runner-vs-shared equivalence, and all three fail-closed cases. Regression: 61/61 across the five multi-account suites and 50/50 across the nine existing send_message test files. --- gateway/config.py | 60 ++++++ gateway/run.py | 34 +--- .../test_send_message_account_routing.py | 172 ++++++++++++++++++ tools/send_message_tool.py | 60 ++++-- 4 files changed, 285 insertions(+), 41 deletions(-) create mode 100644 tests/tools/test_send_message_account_routing.py diff --git a/gateway/config.py b/gateway/config.py index e6cd6c392a59..29652fbd1c4c 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -448,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. diff --git a/gateway/run.py b/gateway/run.py index 73e8908981ed..d65ffa8f52ef 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -15092,36 +15092,14 @@ def _account_platform_config( ) -> "PlatformConfig": """Derive a per-account PlatformConfig from the platform's config (#8287). - The account adapter sees an ordinary PlatformConfig — its own token, - its own home_channel, and account-block settings overriding the - platform-level extra — so adapter internals stay account-agnostic. - The ``accounts`` map itself is stripped from the derived extra. + 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. """ - import dataclasses as _dc + from gateway.config import derive_account_platform_config - from gateway.config import HomeChannel as _HomeChannel - - 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, + return derive_account_platform_config( + platform, platform_config, account_block ) def _iter_live_adapters_with_home(self, snapshot: bool = False): 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 a51c3ad65619..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", @@ -372,17 +372,14 @@ def _handle_send(args): prepare_send_message_platforms() - # Named-bot-account targets (#8287) — e.g. "telegram@support:123" — are - # supported for cron delivery and inbound routing, but this send path - # always uses the platform's default account. Reject the @account form - # with an actionable error rather than the opaque "Unknown platform". - if "@" in platform_name: - _base, _, _acct = platform_name.partition("@") - return tool_error( - f"send targets the default {_base} account; per-account send " - f"(@{_acct}) is not supported here yet. Send via the default " - f"account, or use a cron job with deliver='{platform_name}:'." - ) + # 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 @@ -436,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. @@ -450,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: