"""User-authorization mixin for ``GatewayRunner``: may this user/chat talk to the agent,
the per-adapter DM policy, the unauthorized-DM behavior, and the bot loop guard.

``gateway.run`` is never imported at module import time (cycle). The unauthorized-DM method still
logs through ``gateway.run``'s logger so its records keep that name.
"""

from __future__ import annotations

import contextlib
import logging
import os
import threading
from pathlib import Path
from typing import Optional

from gateway.bot_loop_guard import BotLoopGuard
from gateway.config import Platform
from gateway.session import SessionSource
from gateway.whatsapp_identity import (
    expand_whatsapp_aliases as _expand_whatsapp_auth_aliases,
    normalize_whatsapp_identifier as _normalize_whatsapp_identifier,
)

_GROUP_CHAT_TYPES = frozenset({"group", "forum", "channel"})
_GROUP_FORUM_TYPES = frozenset({"group", "forum"})
_TRUTHY = frozenset({"true", "1", "yes"})
_BOT_LOOP_GUARD_INIT_LOCK = threading.Lock()
logger = logging.getLogger(__name__)

# ``<PLATFORM>_ALLOWED_USERS`` / ``<PLATFORM>_ALLOW_ALL_USERS``, the group-chat allowlist env and
# whether an allowlist suppresses pairing come from each platform's ``PlatformEntry``
# (``allowed_users_env`` / ``allow_all_env`` / ``group_allowed_chats_env`` /
# ``allowlist_suppresses_pairing``), so a plugin platform declares them like a bundled one.
_GROUP_USER_ENV = {Platform.TELEGRAM: "TELEGRAM_GROUP_ALLOWED_USERS"}


# Gate reads use the shared per-profile isolated reader (allowlist leak under multiplex, #72348).
from gateway.platforms._shared import decode_json_list_literal as _decode_json_list_literal
from gateway.platforms._shared import extra_or_secret as _extra_or_secret
from gateway.platforms._shared import platform_gate_env as _auth_env


def _env_truthy(name: str) -> bool:
    return _auth_env(name).lower() in _TRUTHY


def authz_field(name, field, entry):
    """Late-bound :func:`gateway.platform_env.authz_field` (entry field, else the bundled declaration)."""
    from gateway.platform_env import authz_field as _field
    return _field(name, field, entry)


def _group_chat_env(platform, entry=None) -> str:
    """The platform's group-chat allowlist env name (``PlatformEntry.group_allowed_chats_env``), or ""."""
    if platform is None:
        return ""
    return authz_field(platform.value, "group_allowed_chats_env", entry or _registry_entry(platform)) or ""


def _registry_entry(platform):
    """Platform-registry entry for a (plugin) platform, or None. Discovers first (idempotent): the
    allowlist env names live only on the entry, so an undiscovered process must not read "no list"."""
    if platform is None:
        return None
    with contextlib.suppress(Exception):
        from hermes_cli.plugins import discover_plugins
        discover_plugins()
    with contextlib.suppress(Exception):
        from gateway.platform_registry import platform_registry

        return platform_registry.get(platform.value)
    return None


def _coerce_allow_set(raw) -> set[str]:
    """Parse an allowlist (YAML list, JSON list literal string, or comma-separated scalar) into a set of strings."""
    if raw is None:
        return set()
    raw = _decode_json_list_literal(raw)
    if isinstance(raw, list):
        return {str(part).strip() for part in raw if str(part).strip()}
    return {part.strip() for part in str(raw).split(",") if part.strip()}


def _allows(allowed: set[str], candidate: Optional[str]) -> bool:
    return "*" in allowed or candidate in allowed


def _adapter_config_extra(adapter) -> dict:
    return getattr(getattr(adapter, "config", None), "extra", None) or {}


# Nostr npub -> hex (Buzz): ``BUZZ_ALLOWED_USERS`` accepts hex or ``npub1…`` but inbound pubkeys
# are always hex. Pure stdlib; mirrors plugins/platforms/buzz/adapter.py.
# Without decoding, the central allowlist comparison string-matches the raw npub against the hex pubkey and
# an operator who listed only their npub sees every message rejected ("Unauthorized user: <hex pubkey>",
# #78428).
_BECH32_CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
_BECH32_GENERATOR = (0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3)


def _bech32_polymod(values):
    chk = 1
    for value in values:
        top = chk >> 25
        chk = (chk & 0x1FFFFFF) << 5 ^ value
        for i, gen in enumerate(_BECH32_GENERATOR):
            chk ^= gen if ((top >> i) & 1) else 0
    return chk


def _bech32_hrp_expand(hrp: str):
    return [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp]


def _convertbits(data, frombits: int, tobits: int, pad: bool = True):
    acc = 0
    bits = 0
    ret = []
    maxv = (1 << tobits) - 1
    for value in data:
        if value < 0 or (value >> frombits):
            return None
        acc = (acc << frombits) | value
        bits += frombits
        while bits >= tobits:
            bits -= tobits
            ret.append((acc >> bits) & maxv)
    if pad and bits:
        ret.append((acc << (tobits - bits)) & maxv)
    elif not pad and (bits >= frombits or ((acc << (tobits - bits)) & maxv)):
        return None
    return ret


def _npub_to_hex(npub: str) -> Optional[str]:
    """Decode an ``npub1…`` bech32 string to a 64-char hex pubkey, else None."""
    npub = npub.strip().lower()
    if not npub.startswith("npub1"):
        return None
    try:
        data = [_BECH32_CHARSET.index(c) for c in npub[len("npub1"):]]
    except ValueError:
        return None
    if _bech32_polymod(_bech32_hrp_expand("npub") + data) != 1:
        return None
    decoded = _convertbits(data[:-6], 5, 8, pad=False)
    if decoded is None or len(decoded) != 32:
        return None
    return bytes(decoded).hex()


def _normalize_nostr_allow_entries(entries: set) -> set:
    """Add the hex form of every valid ``npub1…`` entry; invalid entries are kept as-is.

    Hex entries pass through unchanged; each valid ``npub1…`` entry is decoded and its 64-char hex form
    added, so either form authorizes the same identity (#78428). Invalid entries are kept as-is (they simply
    never match an inbound hex pubkey).
    """
    return set(entries) | {h for e in entries if e.lower().startswith("npub1") and (h := _npub_to_hex(e))}


def _principal_matches_allowlist(source, user_id: str, allowed_ids: set) -> bool:
    """Whether *user_id* (under any platform-specific alias) is in *allowed_ids*."""
    check_ids = {user_id}

    # WhatsApp (Baileys + Cloud): phone<->LID / JID aliases match the same principal.
    if source.platform in {Platform.WHATSAPP, Platform.WHATSAPP_CLOUD}:
        allowed_ids = set().union(*(_expand_whatsapp_auth_aliases(a) for a in allowed_ids)) or allowed_ids
        check_ids.update(_expand_whatsapp_auth_aliases(user_id))
        normalized_user_id = _normalize_whatsapp_identifier(user_id)
        if normalized_user_id:
            check_ids.add(normalized_user_id)

    platform_value = source.platform.value if source.platform is not None else None
    # SimpleX: SIMPLEX_ALLOWED_USERS matches only the stable numeric contactId (user_id).
    # The display name (user_name) is attacker-controlled — any contact can adopt another
    # contact's display name, so matching it would bypass the allowlist (#44729).
    # Buzz: allowlist may hold npub or hex; inbound pubkeys are hex.
    if platform_value == "buzz":
        # Buzz (Nostr-based): BUZZ_ALLOWED_USERS accepts npub or hex, but inbound event pubkeys are always
        # 64-char hex. Decode npub entries to hex so an operator who listed only their npub authorizes the
        # same identity as the hex form (#78428). Hex entries pass through unchanged, so existing hex-only
        # allowlists keep working.
        allowed_ids = _normalize_nostr_allow_entries(allowed_ids)
        hex_user = _npub_to_hex(user_id) if user_id.startswith("npub") else None
        if hex_user:
            check_ids.add(hex_user)
    return bool(check_ids & allowed_ids)


class GatewayAuthorizationMixin:
    """User/chat authorization methods for ``GatewayRunner``."""

    # ``getattr(self, ...)`` throughout: test helpers build bare runners via ``object.__new__``
    # without ``adapters`` / ``config``.

    def _primary_adapters(self) -> dict:
        return getattr(self, "adapters", None) or {}

    def _profile_adapters_map(self) -> dict:
        return getattr(self, "_profile_adapters", None) or {}

    def _authorization_adapter(self, platform: Optional[Platform], profile: Optional[str] = None):
        """Live adapter whose intake policy gates authorization (``_adapters_for_profile`` for the
        profile rule). ``None`` when the profile has no adapter for *platform*.
        """
        if not platform:
            return None
        return self._adapters_for_profile(profile).get(platform)

    def _adapters_for_profile(self, profile: Optional[str]) -> dict:
        """The live adapter map *profile* may deliver through: ``_profile_adapters[p]`` for a
        secondary, ``self.adapters`` only for the primary/default. ``_profile_adapters`` is consulted
        BEFORE the active profile name: multiplex turns override ``HERMES_HOME`` so
        ``_active_profile_name()`` reports the secondary profile mid-turn, and treating it as primary
        would hand it the default bot. A named profile with no map gets ``{}`` — fail closed: a
        secondary whose adapter failed to connect must NOT fall back to the default profile's adapter
        (replies, tool sends, marker-file notices out the wrong bot)."""
        profile_name = (profile or "").strip() or None
        if not profile_name or profile_name == "default":
            return self._primary_adapters()
        profile_adapters = self._profile_adapters_map()
        if profile_name in profile_adapters:
            adapters = profile_adapters[profile_name]
            from cron.scheduler_preflight import has_credential_platform
            if has_credential_platform(adapters) or not self._is_shared_bot_satellite(profile_name):
                return adapters
            # Its own LOCAL adapter stays layered over the borrowed primary bots.
            return {**self._primary_adapters(), **adapters}
        # Identity captured at construction, not the per-turn HERMES_HOME-derived name.
        primary_profile = getattr(self, "_primary_profile_name", None)
        if not primary_profile:
            with contextlib.suppress(Exception):
                primary_profile = self._active_profile_name()
        return self._primary_adapters() if profile_name == primary_profile else {}

    def _is_shared_bot_satellite(self, profile_name: str) -> bool:
        """A served profile with NO adapter of its own that a ``profile_routes`` entry targets through the
        default profile's bot: it drains through the primary's adapters (gateway/AGENTS.md). Its
        ``_profile_adapters`` entry is the ``{}`` startup placeholder; a secondary connected on ANY
        platform is its own credential boundary and never borrows the primary. Restored/cached sources
        carry no transport ref, so this is what keeps heartbeats, completions and goal notices for such a
        profile deliverable after a restart (the same rule ``kanban_watchers_notifier`` and cron apply)."""
        config = getattr(self, "config", None)
        if not getattr(config, "multiplex_profiles", False):
            return False
        # A bot of its own that is down (failed to connect, or dropped) still makes it a credential boundary.
        if (getattr(self, "_profile_bot_platforms", None) or {}).get(profile_name):
            return False
        routes = getattr(config, "profile_routes", None) or []
        if not any(r.enabled and r.profile == profile_name and r.bot_profile is None for r in routes):
            return False
        from gateway.run import _multiplex_profile_homes
        try:
            return profile_name in {name for name, _home in _multiplex_profile_homes(config)}
        except Exception:
            return False

    def _intake_adapter_for(self, source: Optional[SessionSource]):
        """The adapter that RECEIVED *source*'s event — the only one whose intake policy (ignored
        channels, relay fronting, re-dispatch of a still-live event) may act on it.

        Live provenance only: the registered transport that built the source, the process-level
        RelayAdapter for relay-delivered events, or the receiving bot named by a pinned
        ``RoutingIdentity`` (``transport_profile``) when its adapter reconnected. A source with none
        of these (restored row, hand-built) fails closed under multiplexing — nothing can say which
        bot admitted it, and guessing the runtime profile's bot is exactly the shared-bot /
        route-override confusion the identity exists to end. A standalone gateway owns one adapter
        per platform, so that adapter is the receiver by construction.
        """
        if source is None:
            return None
        owner = self._transport_owner(source)
        if owner is not None:
            return owner[0]
        # Relay ingress keeps the underlying platform on the source, but the one process-level
        # RelayAdapter owns the connector socket; a profile-aware lookup would silently disable
        # streaming/typing/tool progress.
        if getattr(source, "delivered_via_upstream_relay", False) is True:
            return self._primary_adapters().get(Platform.RELAY)
        platform = getattr(source, "platform", None)
        from gateway.session_identity import identity_of
        identity = identity_of(source)
        # A LOCAL session has no receiving bot.  Its default transport is inferred as the
        # primary profile, but the policy and delivery adapter belong to the session authority.
        # Let the delivery fallback resolve that profile instead of treating a lazily-created
        # primary LocalSessionAdapter as the receiver (#139060).
        if (identity is not None and identity.multiplexed
                and not (platform == Platform.LOCAL and identity.transport_inferred)):
            return self._adapters_for_profile(identity.transport_profile).get(platform) if platform else None
        if not getattr(getattr(self, "config", None), "multiplex_profiles", False):
            return self._primary_adapters().get(platform) if platform else None
        return None

    def _delivery_adapter_for(self, source: Optional[SessionSource]):
        """The adapter that ANSWERS *source*: sends, edits, typing, progress, pickers, pending slots.

        The receiving bot whenever it is known (:meth:`_intake_adapter_for` — a routed event replies
        through the bot that received it, whether the runtime is a shared-bot satellite or a profile
        with a bot of its own). With no live provenance the unique owner of ``(platform,
        runtime_profile)`` delivers — ``_adapters_for_profile`` holds one adapter per platform per
        profile, a shared-bot satellite drains through the primary, and a disconnected secondary
        fails closed to ``{}`` rather than borrowing the default bot. Restored sessions, cron,
        kanban and completion notices all rely on this row. Matrix: gateway/AGENTS.md § Profile scope.
        """
        if source is None:
            return None
        from gateway.session_detached import detached_route_of
        if detached_route_of(source):
            # A conversation its chat moved past answers its app viewers only, through the local
            # adapter's event-stream delivery (gateway/session_detached.py), never the chat.
            from gateway.session_identity import identity_of
            identity = identity_of(source)
            profile = identity.runtime_profile if identity is not None else getattr(source, "profile", None)
            return self._adapters_for_profile(profile).get(Platform.LOCAL)
        adapter = self._intake_adapter_for(source)
        if adapter is not None:
            return adapter
        # A pinned identity NAMES the receiving bot (live or restored from ``transport_profile``).
        # If that bot has no adapter right now it is offline: fail closed rather than fall through to
        # the runtime profile's bot — that fallthrough is the "restored lane answers from the wrong
        # bot" row the identity exists to close. Only an identity-less source (legacy row, bare
        # fixture) uses the unique-owner heuristic below.
        from gateway.session_identity import identity_of
        identity = identity_of(source)
        if identity is not None and identity.multiplexed and not identity.transport_inferred:
            return None
        # No identity, or one whose transport was only inferred (hand-built source, pre-column row):
        # the unique owner of ``(platform, runtime_profile)`` delivers.
        # ``getattr``: test fixtures build bare SimpleNamespace sources without ``profile``.
        return self._authorization_adapter(getattr(source, "platform", None), getattr(source, "profile", None))

    def _owning_profile(self, adapter, platform):
        """Return (registered, profile) for a live adapter: profile is None for primary."""
        if adapter is self._primary_adapters().get(platform):
            return True, None
        for profile, profile_adapters in self._profile_adapters_map().items():
            if adapter is profile_adapters.get(platform):
                return True, profile
        return False, None

    def _transport_owner(self, source: SessionSource):
        """``(adapter, profile)`` of the registered adapter that created *source*, if retained; else None.

        ``source.profile`` may differ from the adapter profile when one shared credential serves
        several routed runtimes; ``build_source`` keeps the receiving adapter as provenance so replies
        stay on that transport. Restored/hand-built sources fall back (fail-closed) to profile lookup.
        """
        adapter_ref = getattr(source, "_transport_adapter_ref", None)
        adapter = adapter_ref() if callable(adapter_ref) else None
        platform = getattr(source, "platform", None)
        if adapter is None or platform is None:
            return None
        registered, profile = self._owning_profile(adapter, platform)
        return (adapter, profile) if registered else None

    def _authorization_home_for_source(self, source: SessionSource):
        """HERMES_HOME whose allowlist admits *source*: the identity's transport home (or the
        ingress-stamped one), else the home of the profile owning the adapter that delivers it.
        ``None`` = authorize in the ambient scope (multiplex off, or no live adapter — the check then
        fails closed on its own).

        Inside a routed satellite's turn the ambient scope is the satellite's, whose ``.env`` has no
        token/allowlist; every authorization decision made mid-turn (``/topic``, sibling ``/stop``, plugin
        injection, voice, auto-resume) must read the admitting bot's allowlist instead."""
        from gateway.session_identity import identity_of
        identity = identity_of(source)
        if identity is not None:
            return identity.authorization_home if identity.multiplexed else None
        stamped = getattr(source, "_authorization_profile_home", None)
        if stamped is not None:
            return Path(stamped)
        if not getattr(getattr(self, "config", None), "multiplex_profiles", False):
            return None
        # A restored/hand-built source is admitted by the bot that will answer it (auto-resume,
        # plugin injection) — the same unique-owner row delivery uses.
        adapter = self._delivery_adapter_for(source)
        if adapter is None:
            return None
        _registered, profile = self._owning_profile(adapter, getattr(source, "platform", None))
        if profile is None:
            from hermes_constants import get_process_hermes_home
            return get_process_hermes_home()  # the primary bot's home, never the per-turn override
        from hermes_cli.profiles import get_profile_dir
        return get_profile_dir(profile)

    def _adapter_profile_for_source(self, source: SessionSource) -> Optional[str]:
        """Resolve the transport-owning profile for adapter policy lookups."""
        owner = self._transport_owner(source)
        if owner is not None:
            return owner[1]
        from gateway.session_identity import identity_of
        identity = identity_of(source)
        if identity is not None and identity.multiplexed:
            return None if identity.transport_profile == "default" else identity.transport_profile
        return getattr(source, "profile", None)

    def _restored_source(self, entry) -> Optional[SessionSource]:
        """``entry.origin`` with its identity re-pinned from the routing entry's persisted
        ``transport_profile`` (no live adapter: the restored row of the transport matrix). Every path
        that revives a session from durable state — auto-resume, heartbeat restore, plugin injection,
        background-process events — reads the origin through here, so the receiving bot decides
        delivery and authorization after a restart, not the runtime profile's heuristics."""
        source = getattr(entry, "origin", None)
        if source is None:
            return None
        from gateway.session_identity import restore_identity
        restore_identity(source, runner=self, transport_profile=getattr(entry, "transport_profile", None))
        return source

    def _adapter_flag(self, platform, name: str, profile) -> bool:
        """Adapter-declared boolean, False when unknown. ``authorization_is_upstream`` (relay: a trusted
        authenticated upstream decides) is honored directly; ``enforces_own_access_policy`` (WeCom, Weixin,
        Yuanbao, QQBot, WhatsApp gate at intake) is NOT "already authorized" — those adapters default to
        ``open``, so ``_is_user_authorized`` only trusts them under an actual ``allowlist`` policy."""
        if not platform:
            return False
        adapter = self._authorization_adapter(platform, profile)
        return adapter is not None and bool(getattr(adapter, name, False))

    def _config_extra(self, platform) -> dict:
        """``config.platforms[platform].extra`` as a dict ({} when absent)."""
        platforms = getattr(getattr(self, "config", None), "platforms", None)
        extra = getattr(platforms.get(platform), "extra", None) if platforms is not None else None
        return extra if isinstance(extra, dict) else {}

    def _adapter_setting(self, platform, attr: str, extra_key: str, profile):
        """Live adapter's resolved ``attr`` (folds in the ``<PLATFORM>_*`` env override),
        else ``config.extra[extra_key]`` for bare runners with no adapter."""
        adapter = self._authorization_adapter(platform, profile)
        value = getattr(adapter, attr, None) if adapter is not None else None
        if value is None:
            value = self._config_extra(platform).get(extra_key)
        return value

    def _adapter_policy(self, platform, kind: str, profile) -> str:
        """Lowercased effective ``dm_policy`` (open/allowlist/disabled/pairing) or ``group_policy``
        (open/allowlist/disabled) for *kind* in {"dm", "group"}; ``""`` if unknown."""
        if not platform:
            return ""
        return str(self._adapter_setting(platform, f"_{kind}_policy", f"{kind}_policy", profile) or "").strip().lower()

    def _adapter_group_has_sender_allowlist(
        self, platform: Optional[Platform], chat_id: Optional[str], *, profile: Optional[str] = None
    ) -> bool:
        """Whether a per-group sender allowlist (WeCom ``groups.<id>.allow_from``) gated this message:
        a group may be open at the chat level while restricting senders, so reaching the gateway
        means the adapter already checked that list."""
        if not platform or not chat_id:
            return False
        groups = self._adapter_setting(platform, "_groups", "groups", profile)
        if not isinstance(groups, dict):
            return False
        chat_id_str = str(chat_id)
        group_cfg = groups.get(chat_id_str)
        if not isinstance(group_cfg, dict):
            lowered = chat_id_str.lower()
            group_cfg = next(
                (v for k, v in groups.items() if isinstance(k, str) and k.lower() == lowered and isinstance(v, dict)),
                groups.get("*"),
            )
        if not isinstance(group_cfg, dict):
            return False
        sender_allow = group_cfg.get("allow_from") or group_cfg.get("allowFrom")
        if isinstance(sender_allow, str):
            return bool(sender_allow.strip())
        return isinstance(sender_allow, (list, tuple, set)) and any(str(item).strip() for item in sender_allow)

    def _pairing_store_for(self, source: SessionSource):
        """Per-profile PairingStore for a source, else the global ``self.pairing_store``."""
        per_profile = getattr(self, "pairing_stores", None) or {}
        profile = getattr(source, "profile", None)
        return per_profile[profile] if profile and profile in per_profile else getattr(self, "pairing_store", None)

    def _adapter_extra_for_source(self, source) -> dict:
        return _adapter_config_extra(self._delivery_adapter_for(source))

    def _own_policy_authorizes(self, source, user_id, is_group, adapter_profile) -> Optional[bool]:
        """Own-policy adapter verdict when no env allowlist exists; None = no verdict.

        Trusted only when the effective policy for THIS chat type is ``allowlist``: ``open`` forwards
        EVERY sender (the fail-open SECURITY.md §2.6 forbids), ``disabled`` never forwards, ``pairing``
        forwards unpaired DMs for the handshake (already denied by the pairing-store check).
        Anything else → default-deny.
        """
        if is_group and self._adapter_group_has_sender_allowlist(source.platform, source.chat_id, profile=adapter_profile):
            return True
        if self._adapter_policy(source.platform, "group" if is_group else "dm", adapter_profile) != "allowlist":
            return None
        # Re-check DMs via the live adapter's ``_is_dm_allowed`` when present: pairing revoke can clear
        # WHATSAPP_ALLOWED_USERS while a construction-time snapshot would keep authorizing until
        # restart. Others keep the historical rubber-stamp.
        if not is_group:
            adapter = self._authorization_adapter(source.platform, profile=adapter_profile)
            dm_check = getattr(adapter, "_is_dm_allowed", None) if adapter is not None else None
            if callable(dm_check):
                return bool(dm_check(user_id))
        return True

    def _adapter_extra_allowlist_authorizes(self, source, user_id, is_group) -> bool:
        """Adapters (e.g. Telegram) that gate via config.extra.allow_from / group_allow_from
        without setting enforces_own_access_policy."""
        adapter = self._delivery_adapter_for(source)
        if adapter is None:
            return False
        extra = _adapter_config_extra(adapter)
        # ``group_allow_from`` is a separate group list (Telegram); an adapter with a single user
        # allowlist (Discord) declares ``allow_from_covers_groups`` so its groups read ``allow_from``.
        groups_use_allow_from = getattr(adapter, "allow_from_covers_groups", False) is True
        group_key = "allow_from" if groups_use_allow_from else "group_allow_from"
        adapter_allow = extra.get(group_key if is_group else "allow_from")
        if not adapter_allow:
            # Plugin platforms (Buzz, DingTalk) spell their env allowlist as ``extra.allowed_users``;
            # under multiplex only the default profile's list reaches the env (first-writer-wins
            # bridge), so read the live adapter's.
            entry = _registry_entry(source.platform)
            if authz_field(source.platform.value, "allowed_users_env", entry):
                # Buzz) carry the same operator-configured allowlist in
                # ``PlatformConfig.extra.allowed_users``. An absent/empty entry changes nothing here — the
                # default-deny below still applies. See #82871, #98738.
                adapter_allow = extra.get("allowed_users")
        if not adapter_allow:
            return False
        allowed = _coerce_allow_set(adapter_allow)
        normalize = getattr(adapter, "normalize_user_id", None)
        if callable(normalize):
            # Ids and entries may spell the same principal differently (Buzz hex vs npub).
            allowed = {normalize(entry) or entry for entry in allowed}
        return _allows(allowed, user_id)

    def _adapter_resolved_allowlist_ids(self, source) -> set[str]:
        """IDs an adapter resolved from username-shaped allowlist entries at connect time (Discord).

        The per-turn .env hot-reload restores RAW usernames, so from the second turn on the env
        allowlist holds usernames while user_id is numeric. Never a widening: the empty-allowlist
        branch already returned and adapters only resolve operator-written entries. Only called with
        a non-empty platform allowlist so group/global-only configs never consult adapter memory;
        type-checked so mocks cannot auto-truthy in.
        """
        adapter = resolved_ids = None
        with contextlib.suppress(Exception):
            adapter = self._delivery_adapter_for(source)
        resolver = getattr(adapter, "resolved_allowlist_user_ids", None)
        if callable(resolver):
            with contextlib.suppress(Exception):
                resolved_ids = resolver()
        if not isinstance(resolved_ids, (set, frozenset, list, tuple)):
            return set()
        return {s for e in resolved_ids if isinstance(e, (str, int)) and (s := str(e).strip())}

    def _chat_scoped_grant(self, source, adapter_profile, is_group: bool, allow_adapter_delegation: bool) -> bool:
        """Grants that need no ``user_id`` (checked before the no-user-id guard)."""
        # Trusted-upstream delegation (relay): the connector authenticates this gateway's WS and
        # resolves owner bindings BEFORE delivering, so there is no local RELAY_ALLOWED_USERS. Not a
        # fail-open: fires only for events actually delivered over the relay WS
        # (``delivered_via_upstream_relay``) or whose adapter declares ``authorization_is_upstream``.
        # The delivery marker is PRIMARY because a relayed message carries the UNDERLYING platform,
        # not ``Platform.RELAY``. ``is True``: a MagicMock stand-in must not auto-truthy into authz.
        if allow_adapter_delegation and (
            source.delivered_via_upstream_relay is True
            or self._adapter_flag(source.platform, "authorization_is_upstream", adapter_profile)
        ):
            return True
        # Chat-scoped group allowlists must work with ``user_id is None`` (anonymous admins,
        # sender_chat posts, channel broadcasts).
        if is_group and source.chat_id:
            chat_allowlist_env = _group_chat_env(source.platform)
            if chat_allowlist_env and _allows(_coerce_allow_set(_auth_env(chat_allowlist_env)), source.chat_id):
                return True
            # config.yaml fallback (``extra.group_allowed_chats``): Telegram observe-unmentioned mode
            # strips user_id, so the env-only check above misses it.
            with contextlib.suppress(Exception):
                adapter_group_allowed = self._adapter_extra_for_source(source).get("group_allowed_chats")
                if adapter_group_allowed and _allows(_coerce_allow_set(adapter_group_allowed), source.chat_id):
                    return True
        # Bots admitted by the platform's ``allow_bots_env`` (scoped env → the routed adapter's YAML
        # ``allow_bots`` → none) bypass the human allowlist (#4466). Checked here, before the no-user-id
        # guard: Slack Workflow Builder posts arrive as subtype=bot_message with user=None. The YAML rung
        # is what a secondary profile has: its config is never bridged into the process env. No entry
        # (plugin not loaded) = no bypass: fails closed.
        if getattr(source, "is_bot", False):
            # Idempotent: the env name lives on the plugin's entry, so a runner that never ran the
            # config passes (which discover) must still see it rather than silently deny.
            with contextlib.suppress(Exception):
                from hermes_cli.plugins import discover_plugins
                discover_plugins()
            allow_bots_var = authz_field(source.platform.value, "allow_bots_env", _registry_entry(source.platform)) or ""
            if allow_bots_var:
                extra = {}
                with contextlib.suppress(Exception):
                    extra = self._adapter_extra_for_source(source)
                mode = str(_extra_or_secret(extra, "allow_bots", allow_bots_var, "none")).lower().strip()
                if mode in {"mentions", "all"}:
                    return True
        return False

    def _legacy_telegram_chat_grant(self, source, group_user_allowlist: str) -> bool:
        """TELEGRAM_GROUP_ALLOWED_USERS was once (mis)used as a chat-ID allowlist; "-"-prefixed
        values are chat IDs, honor them and warn once."""
        from gateway.run import logger
        legacy_chat_ids = {v.strip() for v in group_user_allowlist.split(",") if v.strip().startswith("-")}
        if not legacy_chat_ids:
            return False
        if not getattr(self, "_warned_telegram_group_users_legacy", False):
            logger.warning(
                "TELEGRAM_GROUP_ALLOWED_USERS contains chat-ID-shaped values "
                "(%s). Treating them as chat IDs for backward compatibility. "
                "Move chat IDs to TELEGRAM_GROUP_ALLOWED_CHATS — the _USERS var "
                "is now for sender user IDs.",
                ",".join(sorted(legacy_chat_ids)),
            )
            self._warned_telegram_group_users_legacy = True
        return source.chat_id in legacy_chat_ids

    def _bot_loop_guard_instance(self) -> BotLoopGuard:
        guard = getattr(self, "_bot_loop_guard", None)
        if guard is None:
            with _BOT_LOOP_GUARD_INIT_LOCK:
                guard = getattr(self, "_bot_loop_guard", None)
                if guard is None:
                    guard = self._bot_loop_guard = BotLoopGuard()
        return guard

    def _bot_loop_guard_conversation(self, source: SessionSource) -> tuple:
        # One budget per conversation, not per sender pair: a per-pair key would hand N bots N budgets.
        platform = source.platform.value if source.platform else ""
        return (self._adapter_profile_for_source(source) or "", platform, str(source.chat_id or ""))

    def _admit_bot_message(self, source: SessionSource) -> bool:
        """Count one authorized bot-authored inbound message. False when it trips the budget or the chat is cooling down.
        The inbound handler calls this once per message; ``_is_user_authorized`` only peeks because it is asked several times."""
        if not getattr(source, "is_bot", False):
            return True
        allowed, state = self._bot_loop_guard_instance().admit(self._bot_loop_guard_conversation(source))
        if state == "tripped":
            logger.warning(
                "Bot loop guard is dropping bot messages in %s chat %s: bot %s sent one message too many "
                "for the window, cooling down (gateway.bot_loop_guard in config.yaml).",
                source.platform.value if source.platform else "", source.chat_id, source.user_id,
            )
        return allowed

    def _is_user_authorized(self, source: SessionSource, *, allow_adapter_delegation: bool = True) -> bool:
        """Whether a user may use the bot.

        Order: trusted-upstream delegation, chat-scoped group allowlists, ``{PLATFORM}_ALLOW_BOTS``,
        per-platform allow-all, adapter role auth, pairing store, env/config allowlists,
        ``GATEWAY_ALLOW_ALL_USERS``, default deny. A bot-authored message that any of these admits
        is still refused while its chat's loop guard is cooling down.
        """
        if not self._principal_authorized(source, allow_adapter_delegation=allow_adapter_delegation):
            return False
        if not getattr(source, "is_bot", False):
            return True
        # The guard judges the final verdict: a chat allowlist admits a bot before the ALLOW_BOTS block runs.
        return not self._bot_loop_guard_instance().blocked(self._bot_loop_guard_conversation(source))

    def _principal_authorized(self, source: SessionSource, *, allow_adapter_delegation: bool) -> bool:
        """The allowlist verdict alone, before the bot loop guard."""
        from gateway.session_local import authorize_local_source
        local_verdict = authorize_local_source(self, source)
        if local_verdict is not None:
            return local_verdict

        # Webhook events are HMAC-verified; a ``trusted_inbound`` platform's events come from the
        # service its adapter authenticated to (no human sender to allowlist).
        if source.platform == Platform.WEBHOOK or getattr(_registry_entry(source.platform), "trusted_inbound", False) is True:
            return True

        adapter_profile = self._adapter_profile_for_source(source)
        is_group = source.chat_type in _GROUP_CHAT_TYPES
        is_group_or_forum = source.chat_type in _GROUP_FORUM_TYPES
        if self._chat_scoped_grant(source, adapter_profile, is_group, allow_adapter_delegation):
            return True
        user_id = source.user_id
        if not user_id:
            return False

        entry = _registry_entry(source.platform)
        platform_allow_env = authz_field(source.platform.value, "allowed_users_env", entry) or ""
        platform_allow_all_var = authz_field(source.platform.value, "allow_all_env", entry) or ""
        if platform_allow_all_var and _env_truthy(platform_allow_all_var):
            return True
        # Adapter-verified role auth (Discord DISCORD_ALLOWED_ROLES). ``is True``: no MagicMock pass.
        if allow_adapter_delegation and getattr(source, "role_authorized", False) is True:
            return True
        # Pairing store: a first-class grant created only by an operator approving a code. Honored as
        # a UNION with the allowlist (approval also mirrors into it).
        pairing_store = self._pairing_store_for(source)
        if pairing_store is not None and pairing_store.is_approved(source.platform.value if source.platform else "", user_id):
            return True

        platform_allowlist = _auth_env(platform_allow_env)
        group_user_allowlist = _auth_env(_GROUP_USER_ENV.get(source.platform, "")) if is_group_or_forum else ""
        group_chat_allowlist = _auth_env(_group_chat_env(source.platform, entry)) if is_group_or_forum else ""
        global_allowlist = _auth_env("GATEWAY_ALLOWED_USERS")

        if not (platform_allowlist or group_user_allowlist or group_chat_allowlist or global_allowlist):
            # No env allowlist: own-policy adapters gate at intake (see _own_policy_authorizes).
            if allow_adapter_delegation and self._adapter_flag(source.platform, "enforces_own_access_policy", adapter_profile):
                verdict = self._own_policy_authorizes(source, user_id, is_group, adapter_profile)
                if verdict is not None:
                    return verdict
            if self._adapter_extra_allowlist_authorizes(source, user_id, is_group):
                return True
            return _env_truthy("GATEWAY_ALLOW_ALL_USERS")

        if is_group_or_forum and source.chat_id:
            # Telegram group traffic authorized by chat ID (TELEGRAM_GROUP_ALLOWED_USERS gates the sender).
            if group_chat_allowlist and _allows(_coerce_allow_set(group_chat_allowlist), source.chat_id):
                return True
            if (
                source.platform == Platform.TELEGRAM and group_user_allowlist
                and self._legacy_telegram_chat_grant(source, group_user_allowlist)
            ):
                return True

        # TELEGRAM_GROUP_ALLOWED_USERS is group-scoped (no DM access); TELEGRAM_ALLOWED_USERS is platform-wide.
        allowed_ids = (
            _coerce_allow_set(platform_allowlist)
            | _coerce_allow_set(group_user_allowlist)
            | _coerce_allow_set(global_allowlist)
        )
        if platform_allowlist:
            allowed_ids |= self._adapter_resolved_allowlist_ids(source)
        return "*" in allowed_ids or _principal_matches_allowlist(source, user_id, allowed_ids)

    def _get_unauthorized_dm_behavior(self, platform: Optional[Platform], *, profile: Optional[str] = None) -> str:
        """How unauthorized DMs are handled ("pair" / "ignore" / "decline") for a platform.

        Order: explicit per-platform config; Email → "ignore" (inboxes hold arbitrary mail); explicit
        non-default global; adapter dm_policy (pairing → "pair", allowlist/disabled → "ignore"); any
        configured allowlist → "ignore" (spamming unknown contacts with codes is noisy and leaks); else "pair".

        1. 2. Email defaults to ``"ignore"`` unless explicitly opted into pairing. 3. Explicit global
        ``unauthorized_dm_behavior`` in config — wins for chat-shaped platforms when no per-platform
        override is set. 4. When an adapter-level DM policy opts into pairing or silent drop, honor it. 5.
        When an allowlist (``PLATFORM_ALLOWED_USERS``, ``PLATFORM_GROUP_ALLOWED_USERS`` /
        ``PLATFORM_GROUP_ALLOWED_CHATS``, or ``GATEWAY_ALLOWED_USERS``) is configured, default to
        ``"ignore"`` — the allowlist signals that the owner has deliberately restricted access; spamming
        unknown contacts with pairing codes is both noisy and a potential info-leak. (#9337) 6.
        """
        config = getattr(self, "config", None)
        if (
            config and hasattr(config, "get_unauthorized_dm_behavior") and platform
            and "unauthorized_dm_behavior" in self._config_extra(platform)
        ):
            return config.get_unauthorized_dm_behavior(platform)
        if platform == Platform.EMAIL:
            return "ignore"
        if config and hasattr(config, "unauthorized_dm_behavior") and config.unauthorized_dm_behavior != "pair":
            return config.unauthorized_dm_behavior

        allowlist_keys = ["GATEWAY_ALLOWED_USERS"]
        if platform:
            dm_policy = self._adapter_policy(platform, "dm", profile)
            if not dm_policy:
                dm_policy = str(self._config_extra(platform).get("dm_policy") or "").strip().lower()
            if dm_policy == "pairing":
                return "pair"
            if dm_policy in {"allowlist", "disabled"}:
                return "ignore"
            entry = _registry_entry(platform)
            # A group allowlist alone must not silence DM pairing: only platforms that opt in.
            if authz_field(platform.value, "allowlist_suppresses_pairing", entry):
                allowlist_keys = [authz_field(platform.value, "allowed_users_env", entry) or "",
                                  _GROUP_USER_ENV.get(platform), _group_chat_env(platform, entry), *allowlist_keys]
        if any(key and _auth_env(key).strip() for key in allowlist_keys):
            return "ignore"
        return "pair"