feat(gateway): account-aware adapter registry + inbound stamping (#8287, 3/6)

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" rather than trip the fail-closed 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:
+                # A named account inside a secondary profile is not a
+                # supported combination yet — fail closed rather than guess 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
@@ -731,6 +732,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 +1944,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()

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, which is every single-bot gateway.
+        # The runner stamps this after construction when it starts named
+        # account adapters; ``build_source`` copies it onto every inbound
+        # ``SessionSource.account`` so session keys, busy guards, and
+        # outbound routing all 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``).
@@ -7095,6 +7102,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 serves (#8287). This is the single
+            # inbound-construction site every platform's normal-event path
+            # flows through, so stamping here rather than 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, which keeps their sources byte-identical.
+            account=getattr(self, "account_name", None),
             role_authorized=role_authorized,
             auto_thread_created=auto_thread_created,
             auto_thread_initial_name=auto_thread_initial_name,