feat(gateway): multi-account Telegram — one gateway, N bots, isolated sessions (#8287)

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_<ACCOUNT> (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@<account>:<chat_id>, 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

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

gateway/authz_mixin.py

@@ -94,6 +94,7 @@ def _authorization_adapter(
         self,
         platform: Optional[Platform],
         profile: Optional[str] = None,
+        account: Optional[str] = None,
     ):
         """Resolve the live adapter whose intake policy should gate authorization.
 
@@ -102,11 +103,30 @@ def _authorization_adapter(
         ``self.adapters``. ``SessionSource.profile`` selects which map to consult.
         When a stamped profile has its own adapter registry entry, the default
         profile's same-platform adapter must not be consulted as a fallback.
+
+        Multi-account gateways (#8287) add a second dimension the same way:
+        named-account adapters live in ``_account_adapters[platform][account]``
+        while the default account uses ``self.adapters``. A stamped account
+        with no registry entry fails closed for the same reason a profile
+        does — replying out the wrong bot is worse than not replying.
         """
         if not platform:
             return None
         profile_name = (profile or "").strip() or None
+        # Coerce defensively: a MagicMock/SimpleNamespace source auto-creates
+        # a truthy ``account`` attribute (AGENTS.md pitfall #17), which must
+        # read as "default account", not trip the fail-closed account branch.
+        account_name = account.strip() if isinstance(account, str) else None
+        account_name = account_name or None
+        if account_name == "default":
+            account_name = None
         if profile_name and profile_name != "default":
+            if account_name:
+                # Named account inside a secondary profile is not a supported
+                # combination yet — fail closed rather than guessing a bot
+                # (#8287). Checked before the active-profile fast path so an
+                # account under a named active profile also fails closed.
+                return None
             active_profile = None
             active_profile_fn = getattr(self, "_active_profile_name", None)
             if callable(active_profile_fn):
@@ -124,6 +144,9 @@ def _authorization_adapter(
             # (e.g. its adapter failed to connect) must NOT fall back to the
             # default profile's adapter — that sends replies out the wrong bot.
             return None
+        if account_name:
+            account_adapters = getattr(self, "_account_adapters", None) or {}
+            return (account_adapters.get(platform) or {}).get(account_name)
         adapters = getattr(self, "adapters", None) or {}
         return adapters.get(platform)
 
@@ -152,6 +175,7 @@ def _adapter_for_source(self, source: Optional[SessionSource]):
         return self._authorization_adapter(
             getattr(source, "platform", None),
             getattr(source, "profile", None),
+            getattr(source, "account", None),
         )
 
     def _registered_transport_adapter(self, source: SessionSource):

gateway/config.py

@@ -11,6 +11,7 @@
 import logging
 import os
 import json
+import re
 from pathlib import Path
 from dataclasses import asdict, dataclass, field, is_dataclass
 from typing import Dict, List, Optional, Any, Callable
@@ -447,6 +448,66 @@ def _scan_bundled_plugin_platforms(cls) -> set:
 }
 
 
+def derive_account_platform_config(
+    platform: "Platform",
+    platform_config: "PlatformConfig",
+    account_block: Optional[dict],
+) -> "PlatformConfig":
+    """Derive a per-account ``PlatformConfig`` for a named bot account (#8287).
+
+    The consumer sees an ordinary ``PlatformConfig`` — the account's own token,
+    its own ``home_channel``, and account-block settings overriding the
+    platform-level ``extra`` — so adapters and the send path stay
+    account-agnostic. The ``accounts`` map itself is stripped from the derived
+    ``extra`` so a derived config can never recurse into another account.
+
+    Shared by the gateway's account-adapter startup and ``send_message``'s
+    per-account routing, so both resolve an account identically.
+    """
+    import dataclasses as _dc
+
+    merged_extra = {
+        k: v for k, v in (platform_config.extra or {}).items() if k != "accounts"
+    }
+    home_channel = platform_config.home_channel
+    token = platform_config.token
+    for key, value in (account_block or {}).items():
+        if key == "token":
+            token = value
+        elif key == "home_channel" and isinstance(value, dict):
+            # The platform is implicit inside its own account block.
+            _hc = dict(value)
+            _hc.setdefault("platform", platform.value)
+            home_channel = HomeChannel.from_dict(_hc)
+        else:
+            merged_extra[key] = value
+    return _dc.replace(
+        platform_config,
+        token=token,
+        home_channel=home_channel,
+        extra=merged_extra,
+    )
+
+
+def resolve_platform_account(
+    platform_ref: str,
+) -> tuple[str, Optional[str]]:
+    """Split a ``platform[@account]`` reference into ``(platform, account)``.
+
+    ``"telegram"`` -> ``("telegram", None)``; ``"telegram@support"`` ->
+    ``("telegram", "support")``. The account is lowercased to match the
+    normalization applied when accounts are parsed from config/env. ``default``
+    resolves to ``None`` so it is spelled the same everywhere.
+    """
+    base, sep, account = (platform_ref or "").partition("@")
+    if not sep:
+        return base, None
+    account = account.strip().lower()
+    if not account or account == "default":
+        return base, None
+    return base, account
+
+
 def platform_binds_port(platform_value: str, extra: Optional[dict] = None) -> bool:
     """Return True when *platform_value* actually binds a port for *extra* config.
 
@@ -731,6 +792,38 @@ def from_dict(cls, data: Dict[str, Any]) -> "PlatformConfig":
         if _typing_text is None:
             _typing_text = extra.get("typing_status_text")
 
+        # Multi-account blocks (#8287): ``accounts:`` may arrive top-level
+        # (``platforms.telegram.accounts`` in YAML) or bridged into extra by
+        # the shared-key loop. Normalize account names (lowercased) into
+        # ``extra["accounts"]`` so the adapter registry has a single read
+        # path. Tokens are secrets and load from ``<PLATFORM>_BOT_TOKEN_<ACCOUNT>``
+        # env vars in ``_apply_env_overrides`` — a ``token`` key inside a YAML
+        # account block is honored for parity but ``.env`` is the supported
+        # home for credentials.
+        _accounts = data.get("accounts")
+        if _accounts is None:
+            _accounts = extra.get("accounts")
+        if isinstance(_accounts, dict):
+            _norm_accounts: Dict[str, Any] = {}
+            for _acct_name, _acct_block in _accounts.items():
+                _acct_key = str(_acct_name).strip().lower()
+                if not _acct_key:
+                    continue
+                # Account names become a session-key namespace suffix
+                # (``agent:main@<account>``), so the charset is restricted:
+                # ``:`` would break key splitting, ``@`` the suffix parse.
+                if not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", _acct_key):
+                    logger.warning(
+                        "Ignoring platform account %r: names must match "
+                        "[a-z0-9][a-z0-9_-]* (they become session-key "
+                        "namespace suffixes)",
+                        _acct_name,
+                    )
+                    continue
+                _norm_accounts[_acct_key] = _coerce_dict(_acct_block)
+            if _norm_accounts:
+                extra["accounts"] = _norm_accounts
+
         channel_overrides: Dict[str, ChannelOverride] = {}
         raw_overrides = data.get("channel_overrides") or {}
         if isinstance(raw_overrides, dict):
@@ -1911,6 +2004,40 @@ def _enable_from_env(platform: Platform) -> PlatformConfig:
     if telegram_token:
         telegram_config = _enable_from_env(Platform.TELEGRAM)
         telegram_config.token = telegram_token
+
+    # Multi-account tokens (#8287): ``TELEGRAM_BOT_TOKEN_<ACCOUNT>`` declares
+    # an additional bot account named ``<account>`` (lowercased). The
+    # unsuffixed ``TELEGRAM_BOT_TOKEN`` remains the default account, so
+    # single-bot setups are byte-identical to before. Candidate names are
+    # enumerated from the process env (dotenv loads ``.env`` there) and each
+    # value is read back through ``getenv`` so profile-scoped secrets win
+    # when a scope is active. Behavioral per-account settings (allowlists,
+    # home channels, display names) belong in ``platforms.telegram.accounts``
+    # in config.yaml — env vars carry only the credential.
+    _tg_account_prefix = "TELEGRAM_BOT_TOKEN_"
+    for _env_name in sorted(os.environ):
+        if not _env_name.startswith(_tg_account_prefix):
+            continue
+        _acct_name = _env_name[len(_tg_account_prefix):].strip().lower()
+        _acct_token = getenv(_env_name)
+        if not _acct_name or not _acct_token:
+            continue
+        # Same charset rule as the YAML block: names become session-key
+        # namespace suffixes.
+        if not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", _acct_name):
+            logger.warning(
+                "Ignoring %s: account names must match [a-z0-9][a-z0-9_-]*",
+                _env_name,
+            )
+            continue
+        _tg_cfg = _enable_from_env(Platform.TELEGRAM)
+        _tg_accounts = _tg_cfg.extra.setdefault("accounts", {})
+        if not isinstance(_tg_accounts, dict):
+            _tg_accounts = {}
+            _tg_cfg.extra["accounts"] = _tg_accounts
+        _acct_block = _tg_accounts.setdefault(_acct_name, {})
+        if isinstance(_acct_block, dict):
+            _acct_block["token"] = _acct_token
 
     # Reply threading mode for Telegram (off/first/all)
     telegram_reply_mode = getenv("TELEGRAM_REPLY_TO_MODE", "").lower()