diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 990d306303..0c5fea66a0 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -223,9 +223,9 @@ class GatewayAuthorizationMixin: """Resolve the live adapter for an inbound ``SessionSource``.""" if source is None: return None - transport_adapter = self._registered_transport_adapter(source) - if transport_adapter is not None: - return transport_adapter + owner = self._transport_owner(source) + if owner is not None: + return owner[0] # Relay ingress keeps the underlying platform on the source, but delivery must use the one # process-level RelayAdapter owning the connector socket; a profile-aware lookup would # silently disable streaming/typing/tool progress. @@ -243,8 +243,8 @@ class GatewayAuthorizationMixin: return True, profile return False, None - def _registered_transport_adapter(self, source: SessionSource): - """The registered adapter that created *source*, if retained. + 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 @@ -255,17 +255,13 @@ class GatewayAuthorizationMixin: platform = getattr(source, "platform", None) if adapter is None or platform is None: return None - registered, _ = self._owning_profile(adapter, platform) - return adapter if registered else None + registered, profile = self._owning_profile(adapter, platform) + return (adapter, profile) if registered else None def _adapter_profile_for_source(self, source: SessionSource) -> Optional[str]: """Resolve the transport-owning profile for adapter policy lookups.""" - adapter = self._registered_transport_adapter(source) - if adapter is not None: - registered, profile = self._owning_profile(adapter, getattr(source, "platform", None)) - if registered: - return profile - return getattr(source, "profile", None) + owner = self._transport_owner(source) + return owner[1] if owner is not None else getattr(source, "profile", None) def _adapter_flag(self, platform, name: str, profile) -> bool: """Adapter-declared boolean, False when unknown. ``authorization_is_upstream`` (relay: a trusted diff --git a/gateway/channel_directory.py b/gateway/channel_directory.py index edac38888e..c895b27a68 100644 --- a/gateway/channel_directory.py +++ b/gateway/channel_directory.py @@ -18,15 +18,13 @@ from utils import atomic_json_write logger = logging.getLogger(__name__) -# Paths resolve lazily: a multiplexed gateway serves several profile homes from -# one process, so an import-time constant would pin every profile to whichever -# home imported first. These globals are explicit overrides (tests patch them); -# ``None`` means "resolve from the current home". +# Paths resolve lazily: a multiplexed gateway serves several profile homes from one +# process, so an import-time constant would pin every profile to whichever home imported +# first. These globals are explicit overrides (tests patch them); None = current home. DIRECTORY_PATH: Optional[Path] = None -# User-maintained friendly-name overlay, re-applied on every build AND load -# (hand-edits to the regenerated channel_directory.json don't survive); also -# lets a chat be pre-named before its first traffic. -# Format: {"": {"": "", ...}, ...} +# User-maintained friendly-name overlay {"": {"": ""}}, +# re-applied on every build AND load (hand-edits to the regenerated +# channel_directory.json don't survive); also lets a chat be pre-named before first traffic. CHANNEL_ALIASES_PATH: Optional[Path] = None # Slack refresh failures recur on every timed rebuild (missing scope, revoked @@ -47,13 +45,17 @@ def _aliases_path() -> Path: return CHANNEL_ALIASES_PATH or get_hermes_home() / "channel_aliases.json" +def _read_json(path: Path) -> Any: + with open(path, encoding="utf-8") as f: + return json.load(f) + + def _load_json_dict(path: Path) -> Dict[str, Any]: """Read a JSON object from *path*; {} when missing, unreadable, or not a dict.""" if not path.exists(): return {} try: - with open(path, encoding="utf-8") as f: - data = json.load(f) + data = _read_json(path) return data if isinstance(data, dict) else {} except Exception: return {} @@ -62,8 +64,8 @@ def _load_json_dict(path: Path) -> Dict[str, Any]: def _apply_channel_aliases(platforms: Dict[str, Any]) -> None: """Overlay friendly names onto directory entries by chat_id, in place. - An aliased id not yet discovered gets a placeholder entry so a - freshly-created group is addressable by name before its first message. + An aliased id not yet discovered gets a placeholder entry so a freshly-created + group is addressable by name before its first message. """ for plat_name, id_map in _load_json_dict(_aliases_path()).items(): if not isinstance(id_map, dict): @@ -93,28 +95,23 @@ def _normalize_channel_query(value: str) -> str: def _channel_target_name(platform_name: str, channel: Dict[str, Any]) -> str: """Human-facing target label for a channel entry.""" name = channel["name"] - if platform_name == "discord" and channel.get("guild"): - return f"#{name}" - if platform_name != "discord" and channel.get("type"): - return f"{name} ({channel['type']})" - return name + if platform_name == "discord": + return f"#{name}" if channel.get("guild") else name + return f"{name} ({channel['type']})" if channel.get("type") else name def _session_entry_id(origin: Dict[str, Any]) -> Optional[str]: chat_id = origin.get("chat_id") if not chat_id: return None - thread_id = origin.get("thread_id") - return f"{chat_id}:{thread_id}" if thread_id else str(chat_id) + return f"{chat_id}:{thread_id}" if (thread_id := origin.get("thread_id")) else str(chat_id) def _session_entry_name(origin: Dict[str, Any]) -> str: base_name = origin.get("chat_name") or origin.get("user_name") or str(origin.get("chat_id")) - thread_id = origin.get("thread_id") - if not thread_id: + if not (thread_id := origin.get("thread_id")): return base_name - topic_label = origin.get("chat_topic") or f"topic {thread_id}" - return f"{base_name} / {topic_label}" + return f"{base_name} / {origin.get('chat_topic') or f'topic {thread_id}'}" def _report_slack_failure(team_id: str, error_code: Optional[str], detail: str) -> None: @@ -137,9 +134,7 @@ def _report_slack_failure(team_id: str, error_code: Optional[str], detail: str) async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]: """Build the directory from connected adapters + session data and persist it.""" from gateway.config import Platform - platforms: Dict[str, List[Dict[str, str]]] = {} - for platform, adapter in adapters.items(): try: list_channels = getattr(adapter, "list_channels", None) @@ -154,21 +149,14 @@ async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]: platforms["slack"] = await _build_slack(adapter) except Exception as e: logger.warning("Channel directory: failed to build %s: %s", platform.value, e) - - # Platforms without channel enumeration get session-based discovery, but - # only when connected in THIS gateway process: origins for disabled or - # decommissioned platforms must not resurface as stale send targets. + # Platforms without channel enumeration get session-based discovery, but only when + # connected in THIS gateway process: origins for disabled or decommissioned + # platforms must not resurface as stale send targets. adapter_platform_names = {getattr(p, "value", str(p)) for p in adapters} - async def _discover(plat_name: str) -> None: - if ( - plat_name in _SKIP_SESSION_DISCOVERY - or plat_name in platforms - or plat_name not in adapter_platform_names - ): + if plat_name in _SKIP_SESSION_DISCOVERY or plat_name in platforms or plat_name not in adapter_platform_names: return platforms[plat_name] = await asyncio.to_thread(_build_from_sessions, plat_name) - for plat in Platform: await _discover(plat.value) # Plugin platforms are dynamic enum members missing from Platform.__members__. @@ -176,16 +164,12 @@ async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]: from gateway.platform_registry import platform_registry for entry in platform_registry.plugin_entries(): await _discover(entry.name) - _apply_channel_aliases(platforms) - directory = {"updated_at": datetime.now().isoformat(), "platforms": platforms} - try: await asyncio.to_thread(atomic_json_write, _directory_path(), directory) except Exception as e: logger.warning("Channel directory: failed to write: %s", e) - return directory @@ -195,19 +179,16 @@ def _build_discord(adapter) -> List[Dict[str, str]]: client = getattr(adapter, "_client", None) if not client: return channels - try: import discord as _discord # noqa: F401 — SDK presence check except ImportError: return channels - for guild in client.guilds: # Forum channels (type 15): creating a message auto-spawns a thread post. forums = getattr(guild, "forum_channels", None) or [] for chs, ch_type in ((guild.text_channels, "channel"), (forums, "forum")): for ch in chs: channels.append({"id": str(ch.id), "name": ch.name, "guild": guild.name, "type": ch_type}) - # DM-capable users aren't reachable via guild enumeration; they come from sessions. channels.extend(_build_from_sessions("discord")) return channels @@ -233,9 +214,7 @@ def _normalize_adapter_channels(raw_channels: Any) -> List[Dict[str, Any]]: if not channel_id or not name or channel_id in seen_ids: continue entry: Dict[str, Any] = {"id": channel_id, "name": name, "type": str(raw.get("type") or "dm")} - for key in ("thread_id", "guild"): - if raw.get(key): - entry[key] = str(raw.get(key)) + entry.update({key: str(raw[key]) for key in ("thread_id", "guild") if raw.get(key)}) channels.append(entry) seen_ids.add(channel_id) return channels @@ -257,10 +236,7 @@ async def _slack_team_channels(team_id: str, client, seen_ids: set) -> List[Dict cursor: Optional[str] = None for _page in range(20): # safety cap on pagination response = await client.users_conversations( - types="public_channel,private_channel", - exclude_archived=True, - limit=200, - cursor=cursor, + types="public_channel,private_channel", exclude_archived=True, limit=200, cursor=cursor, ) if not response.get("ok"): error_code = response.get("error", "unknown") @@ -289,7 +265,6 @@ async def _slack_resolve_raw_names(client, channels: List[Dict[str, Any]]) -> No unresolved_by_base.setdefault(_slack_base_id(entry["id"]), []).append(entry) if not unresolved_by_base: return - async def _resolve_base(base_id: str, entries: list) -> None: try: resp = await client.conversations_info(channel=base_id) @@ -311,7 +286,6 @@ async def _slack_resolve_raw_names(client, channels: List[Dict[str, Any]]) -> No entry["type"] = resolved_type except Exception as e: logger.debug("Channel directory: failed to resolve %s: %s", base_id, e) - await asyncio.gather(*[_resolve_base(bid, ents) for bid, ents in unresolved_by_base.items()]) @@ -321,12 +295,10 @@ async def _build_slack(adapter) -> List[Dict[str, Any]]: team_clients = getattr(adapter, "_team_clients", None) or {} if not team_clients: return await asyncio.to_thread(_build_from_sessions, "slack") - channels: List[Dict[str, Any]] = [] seen_ids: set = set() for team_id, client in team_clients.items(): channels.extend(await _slack_team_channels(team_id, client, seen_ids)) - # Merge session-history DM/group entries, naming raw-ID entries from the # API-discovered channels where the base conversation ID is known. api_name_lookup = {ch["id"]: ch["name"] for ch in channels} @@ -338,25 +310,18 @@ async def _build_slack(adapter) -> List[Dict[str, Any]]: entry["name"] = api_name_lookup[_slack_base_id(eid)] channels.append(entry) seen_ids.add(eid) - await _slack_resolve_raw_names(next(iter(team_clients.values())), channels) return channels def _build_from_sessions(platform_name: str) -> List[Dict[str, str]]: - """Pull known channels/contacts from gateway session origin data. - - state.db is the primary source; sessions.json is the fallback for - pre-migration databases. - """ + """Known channels/contacts from session origins: state.db first, sessions.json fallback (pre-migration).""" return _build_from_sessions_db(platform_name) or _build_from_sessions_json(platform_name) def _entries_from_origins(platform_name: str, source: str, origins_fn) -> List[Dict[str, Any]]: - """Deduped directory entries for the (origin, chat_type) pairs yielded by ``origins_fn()``. - - Appends incrementally so a mid-iteration failure keeps the entries read so far. - """ + """Deduped entries for the (origin, chat_type) pairs from ``origins_fn()``; a mid-iteration + failure keeps entries read so far.""" entries: List[Dict[str, Any]] = [] try: seen_ids = set() @@ -393,7 +358,6 @@ def _build_from_sessions_db(platform_name: str) -> List[Dict[str, str]]: if not isinstance(origin, dict) or not origin: origin = {"chat_id": row.get("chat_id"), "thread_id": row.get("thread_id"), "chat_name": row.get("display_name")} yield origin, row.get("chat_type") or "dm" - return _entries_from_origins(platform_name, "state.db session read failed", _origins) @@ -402,18 +366,14 @@ def _build_from_sessions_json(platform_name: str) -> List[Dict[str, str]]: sessions_path = get_hermes_home() / "sessions" / "sessions.json" if not sessions_path.exists(): return [] - def _origins() -> Iterable[Tuple[Dict[str, Any], Any]]: - with open(sessions_path, encoding="utf-8") as f: - data = json.load(f) - for _key, session in data.items(): + for _key, session in _read_json(sessions_path).items(): # Keys starting with "_" (e.g. the gateway's "_README") are metadata sentinels. if str(_key).startswith("_") or not isinstance(session, dict): continue origin = session.get("origin") or {} if origin.get("platform") == platform_name: yield origin, session.get("chat_type", "dm") - return _entries_from_origins(platform_name, "failed to read sessions", _origins) @@ -424,8 +384,7 @@ def load_directory() -> Dict[str, Any]: directory_path = _directory_path() if directory_path.exists(): with contextlib.suppress(Exception): - with open(directory_path, encoding="utf-8") as f: - data = json.load(f) + data = _read_json(directory_path) # Aliases apply on read too, so new names take effect between timed rebuilds. _apply_channel_aliases(data.setdefault("platforms", {})) return data @@ -436,37 +395,27 @@ def load_directory() -> Dict[str, Any]: def lookup_channel_type(platform_name: str, chat_id: str) -> Optional[str]: """Channel ``type`` string (e.g. ``"channel"``, ``"forum"``) for *chat_id*, or None if unknown.""" - for ch in load_directory().get("platforms", {}).get(platform_name, []): - if ch.get("id") == chat_id: - return ch.get("type") - return None + channels = load_directory().get("platforms", {}).get(platform_name, []) + return next((ch.get("type") for ch in channels if ch.get("id") == chat_id), None) def resolve_channel_name(platform_name: str, name: str) -> Optional[str]: - """Resolve a human-friendly channel name to a numeric ID (case-insensitive, first match wins). - - Accepts e.g. Discord "bot-home" / "#bot-home" / "GuildName/bot-home", Telegram - display or group names, Slack "engineering" / "#engineering". - """ + """Resolve a friendly channel name (e.g. "bot-home", "#bot-home", "GuildName/bot-home", + Slack "#engineering") to an ID; case-insensitive, first match wins.""" channels = load_directory().get("platforms", {}).get(platform_name, []) if not channels: return None - - # 0. Exact ID match — case-sensitive, no normalization, so raw platform IDs - # (e.g. Slack "C0B0QV5434G") work even when _parse_target_ref's format - # guard didn't recognize them as explicit. + # 0. Exact ID match — case-sensitive, no normalization, so raw platform IDs (e.g. Slack + # "C0B0QV5434G") work even when _parse_target_ref's format guard didn't recognize them. raw = name.strip() for ch in channels: if ch.get("id") == raw: return ch["id"] - query = _normalize_channel_query(name) - # 1. Exact name match, including the display labels shown by send_message(action="list") for ch in channels: if query in (_normalize_channel_query(ch["name"]), _normalize_channel_query(_channel_target_name(platform_name, ch))): return ch["id"] - # 2. Guild-qualified match for Discord ("GuildName/channel") if "/" in query: guild_part, ch_part = query.rsplit("/", 1) @@ -474,30 +423,23 @@ def resolve_channel_name(platform_name: str, name: str) -> Optional[str]: guild = ch.get("guild", "").strip().lower() if guild == guild_part and _normalize_channel_query(ch["name"]) == ch_part: return ch["id"] - # 3. Partial prefix match (only if unambiguous) matches = [ch for ch in channels if _normalize_channel_query(ch["name"]).startswith(query)] - if len(matches) == 1: - return matches[0]["id"] - - return None + return matches[0]["id"] if len(matches) == 1 else None def format_directory_for_display(platforms: Optional[Dict[str, Any]] = None) -> str: """Format the channel directory as a human-readable list for the model. ``platforms`` overrides the on-disk directory (``hermes send --list`` merges in - configured-but-undiscovered platforms). An empty channel list renders a - "(no channels discovered yet)" hint: the platform is still a valid send target. + configured-but-undiscovered platforms); an empty channel list renders a "(no channels + discovered yet)" hint because the platform is still a valid send target. """ if platforms is None: platforms = load_directory().get("platforms", {}) - if not platforms: return "No messaging platforms connected or no channels discovered yet." - lines = ["Available messaging targets:\n"] - for plat_name, channels in sorted(platforms.items()): if not channels: lines.append(f"{plat_name.title()}:") @@ -521,8 +463,6 @@ def format_directory_for_display(platforms: Optional[Dict[str, Any]] = None) -> lines.append(f"{plat_name.title()}:") lines.extend(f" {plat_name}:{_channel_target_name(plat_name, ch)}" for ch in channels) lines.append("") - lines.append('Use these as the "target" parameter when sending.') lines.append('Bare platform name (e.g. "telegram") sends to home channel.') - return "\n".join(lines) diff --git a/gateway/config.py b/gateway/config.py index f330b3c65f..a0a20c0491 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -105,19 +105,20 @@ def _normalize_transport_token(value: Any) -> str: return str(value).strip().lower() or "auto" -def _coerce_float(value: Any, default: float) -> float: +def _coerce_num(cast, value: Any, default): + # OverflowError: ``int(float("inf"))`` — non-finite YAML must degrade, not abort loading. try: - return default if value is None else float(value) - except (TypeError, ValueError): + return default if value is None else cast(value) + except (TypeError, ValueError, OverflowError): return default +def _coerce_float(value: Any, default: float) -> float: + return _coerce_num(float, value, default) + + def _coerce_int(value: Any, default: int) -> int: - # OverflowError: ``int(float("inf"))`` — non-finite YAML must degrade, not abort loading. - try: - return default if value is None else int(value) - except (TypeError, ValueError, OverflowError): - return default + return _coerce_num(int, value, default) def _coerce_optional_positive_int(value: Any, key: str) -> Optional[int]: diff --git a/gateway/config_env.py b/gateway/config_env.py index eb67de3967..9b3aed0f0e 100644 --- a/gateway/config_env.py +++ b/gateway/config_env.py @@ -168,32 +168,26 @@ def _env_reply_mode(config: GatewayConfig, platform: Platform, env: str) -> None config.platforms.setdefault(platform, PlatformConfig()).reply_to_mode = mode -def _enable_from_env(config: GatewayConfig, platform: Platform) -> PlatformConfig: +def _enable_from_env( + config: GatewayConfig, platform: Platform, *, pop_marker: bool = False, warn: bool = True +) -> PlatformConfig: """Enable *platform* on env credentials unless config.yaml explicitly disabled it. - READS (does not pop) ``_enabled_explicit``: the plugin-enable pass also needs it.""" - if platform not in config.platforms: - config.platforms[platform] = PlatformConfig(enabled=True) - return config.platforms[platform] - platform_config = config.platforms[platform] - if not platform_config.enabled: - if platform_config.extra.get("_enabled_explicit", False): - _warn_explicit_disable_beats_env(platform) - else: - platform_config.enabled = True - return platform_config - - -def _enable_port_bound_from_env(config: GatewayConfig, platform: Platform) -> PlatformConfig: - """Enable a port-binding platform unless config.yaml explicitly disabled it. A multiplex secondary profile pins ``enabled: false`` to share the default profile's listener yet inherits the process env; without this guard env presence would force-enable it and trip - MultiplexConfigError. POPs the marker (terminal branch). + MultiplexConfigError. By default the ``_enabled_explicit`` marker is READ (the plugin-enable + and relay passes still need it) and the disable is warned once; port-binding platforms POP it + (terminal branch) and stay silent. """ platform_config = config.platforms.setdefault(platform, PlatformConfig()) - explicit = platform_config.extra.pop("_enabled_explicit", False) - if not explicit or platform_config.enabled: + extra = platform_config.extra + explicit = extra.pop("_enabled_explicit", False) if pop_marker else extra.get("_enabled_explicit", False) + if platform_config.enabled: + return platform_config + if not explicit: platform_config.enabled = True + elif warn: + _warn_explicit_disable_beats_env(platform) return platform_config @@ -305,7 +299,7 @@ def _api_server(config: GatewayConfig) -> None: key = getenv("API_SERVER_KEY") if not _has_usable_api_server_key(key): return - extra = _enable_port_bound_from_env(config, Platform.API_SERVER).extra + extra = _enable_from_env(config, Platform.API_SERVER, pop_marker=True, warn=False).extra extra["key"] = key _csv_extras(extra, (("cors_origins", getenv("API_SERVER_CORS_ORIGINS")),)) _env_extras(extra, (("port", "API_SERVER_PORT", _INT), ("host", "API_SERVER_HOST"), ("model_name", "API_SERVER_MODEL_NAME"))) @@ -313,7 +307,7 @@ def _api_server(config: GatewayConfig) -> None: def _webhook(config: GatewayConfig) -> None: if is_truthy_value(getenv("WEBHOOK_ENABLED")): - extra = _enable_port_bound_from_env(config, Platform.WEBHOOK).extra + extra = _enable_from_env(config, Platform.WEBHOOK, pop_marker=True, warn=False).extra _env_extras(extra, (("port", "WEBHOOK_PORT", _INT), ("secret", "WEBHOOK_SECRET"))) @@ -326,9 +320,8 @@ def _msgraph_webhook(config: GatewayConfig) -> None: or client_state or resources or allowed_cidrs): return msgraph_cfg = config.platforms.setdefault(Platform.MSGRAPH_WEBHOOK, PlatformConfig()) - # Same guard as webhook, but READ (don't pop) the marker: the relay pass still consults it. - if enabled and (not msgraph_cfg.extra.get("_enabled_explicit", False) or msgraph_cfg.enabled): - msgraph_cfg.enabled = True + if enabled: # same guard as webhook, but READ (don't pop) the marker: the relay pass still consults it + _enable_from_env(config, Platform.MSGRAPH_WEBHOOK, warn=False) _env_extras(msgraph_cfg.extra, (("port", "MSGRAPH_WEBHOOK_PORT", _INT),)) if client_state: msgraph_cfg.extra["client_state"] = client_state diff --git a/gateway/config_loader.py b/gateway/config_loader.py index 7f35552d5b..ec6dbad729 100644 --- a/gateway/config_loader.py +++ b/gateway/config_loader.py @@ -58,10 +58,6 @@ def _quick_commands_ok(value: Any) -> bool: return False -def _is_dict(value: Any) -> bool: - return isinstance(value, dict) - - def _dm_behavior_choice(value: Any, default: str = "pair") -> str: return _normalize_choice(value, {"pair", "ignore"}, default) @@ -74,7 +70,7 @@ def _presence(*keys: str) -> tuple: _TOPLEVEL_BRIDGE: tuple = ( ("session_reset", "default_reset_policy", "presence", lambda v: bool(v) and isinstance(v, dict), None), ("quick_commands", "quick_commands", "none", _quick_commands_ok, None), - ("stt", "stt", "presence", _is_dict, None), + ("stt", "stt", "presence", lambda v: isinstance(v, dict), None), *_presence("stt_echo_transcripts", "group_sessions_per_user", "thread_sessions_per_user"), ("multiplex_profiles", "multiplex_profiles", "gwdata", None, None), *_presence("multiplex_profile_allowlist", "room_link_url"), diff --git a/gateway/pairing.py b/gateway/pairing.py index 7fefdb017e..278980316a 100644 --- a/gateway/pairing.py +++ b/gateway/pairing.py @@ -1,12 +1,10 @@ """DM pairing: code-based approval of new users on messaging platforms. -Unknown users receive a one-time pairing code that the bot owner approves via -the CLI, instead of maintaining static user-ID allowlists. - -Security properties (OWASP + NIST SP 800-63-4): 8-char codes from a 32-char -unambiguous alphabet via ``secrets``, 1-hour expiry, max 3 pending per -platform, 1 request per user per 10 min, lockout after 5 failed approvals, -chmod 0600 data files, codes never logged. Storage: ~/.hermes/pairing/ +Unknown users receive a one-time pairing code that the bot owner approves via the +CLI, instead of static user-ID allowlists. Security properties (OWASP + NIST SP +800-63-4): 8-char codes from a 32-char unambiguous alphabet via ``secrets``, 1-hour +expiry, max 3 pending per platform, 1 request per user per 10 min, lockout after 5 +failed approvals, chmod 0600 data files, codes never logged. Storage: ~/.hermes/pairing/ """ import contextlib @@ -21,15 +19,8 @@ import time from pathlib import Path from typing import Optional -from gateway.whatsapp_identity import ( - expand_whatsapp_aliases, - normalize_whatsapp_identifier, -) -from hermes_constants import ( - get_default_hermes_root, - get_hermes_dir, - get_hermes_home, -) +from gateway.whatsapp_identity import expand_whatsapp_aliases, normalize_whatsapp_identifier +from hermes_constants import get_default_hermes_root, get_hermes_dir, get_hermes_home from utils import atomic_replace logger = logging.getLogger(__name__) @@ -53,9 +44,7 @@ PAIRING_DIR = None def _default_pairing_dir() -> Path: - if PAIRING_DIR is not None: - return PAIRING_DIR - return get_hermes_dir("platforms/pairing", "pairing") + return PAIRING_DIR if PAIRING_DIR is not None else get_hermes_dir("platforms/pairing", "pairing") # Platform value -> allowlist env var. Approving a code also writes the user into @@ -63,36 +52,25 @@ def _default_pairing_dir() -> Path: # the visible source of truth. Platforms absent here (or with no allowlist # configured) keep the pairing store as the sole grant record (authz union). _PLATFORM_ALLOWLIST_ENV = { - "telegram": "TELEGRAM_ALLOWED_USERS", - "discord": "DISCORD_ALLOWED_USERS", - "whatsapp": "WHATSAPP_ALLOWED_USERS", - "whatsapp_cloud": "WHATSAPP_CLOUD_ALLOWED_USERS", - "slack": "SLACK_ALLOWED_USERS", - "signal": "SIGNAL_ALLOWED_USERS", - "email": "EMAIL_ALLOWED_USERS", - "sms": "SMS_ALLOWED_USERS", - "mattermost": "MATTERMOST_ALLOWED_USERS", - "matrix": "MATRIX_ALLOWED_USERS", - "dingtalk": "DINGTALK_ALLOWED_USERS", - "feishu": "FEISHU_ALLOWED_USERS", - "wecom": "WECOM_ALLOWED_USERS", - "wecom_callback": "WECOM_CALLBACK_ALLOWED_USERS", - "weixin": "WEIXIN_ALLOWED_USERS", - "bluebubbles": "BLUEBUBBLES_ALLOWED_USERS", - "qqbot": "QQ_ALLOWED_USERS", - "yuanbao": "YUANBAO_ALLOWED_USERS", + "telegram": "TELEGRAM_ALLOWED_USERS", "discord": "DISCORD_ALLOWED_USERS", + "whatsapp": "WHATSAPP_ALLOWED_USERS", "whatsapp_cloud": "WHATSAPP_CLOUD_ALLOWED_USERS", + "slack": "SLACK_ALLOWED_USERS", "signal": "SIGNAL_ALLOWED_USERS", + "email": "EMAIL_ALLOWED_USERS", "sms": "SMS_ALLOWED_USERS", + "mattermost": "MATTERMOST_ALLOWED_USERS", "matrix": "MATRIX_ALLOWED_USERS", + "dingtalk": "DINGTALK_ALLOWED_USERS", "feishu": "FEISHU_ALLOWED_USERS", + "wecom": "WECOM_ALLOWED_USERS", "wecom_callback": "WECOM_CALLBACK_ALLOWED_USERS", + "weixin": "WEIXIN_ALLOWED_USERS", "bluebubbles": "BLUEBUBBLES_ALLOWED_USERS", + "qqbot": "QQ_ALLOWED_USERS", "yuanbao": "YUANBAO_ALLOWED_USERS", } def _allowlist_env_for_platform(platform: str) -> Optional[str]: """Allowlist env var name for ``platform`` (plugin registry fallback), or None.""" platform = (platform or "").lower().strip() - env_var = _PLATFORM_ALLOWLIST_ENV.get(platform) - if env_var: + if env_var := _PLATFORM_ALLOWLIST_ENV.get(platform): return env_var with contextlib.suppress(Exception): from gateway.platform_registry import platform_registry - return platform_registry.get(platform).allowed_users_env or None return None @@ -139,14 +117,13 @@ def _matching_ids(platform: str, approved: dict, user_id: str) -> list: def _read_allowlist_env(env_var: str) -> str: """Read a platform allowlist env var through the profile secret scope. - Under multiplexing the process env may hold ANOTHER profile's allowlist, so a - scoped miss must return empty rather than borrow it. Unscoped callers keep the - legacy ``os.getenv`` read. Writes go through ``save_env_value``/``remove_env_value``, - which target the active profile's ``.env`` / installed scope, not ``os.environ``. + Under multiplexing the process env may hold ANOTHER profile's allowlist, so a scoped + miss must return empty rather than borrow it; unscoped callers keep the legacy + ``os.getenv`` read. Writes (``save_env_value``/``remove_env_value``) target the + active profile's ``.env`` / installed scope, not ``os.environ``. """ with contextlib.suppress(Exception): from agent.secret_scope import UnscopedSecretError, get_secret - try: return (get_secret(env_var) or "").strip() except UnscopedSecretError: @@ -169,7 +146,6 @@ def _write_allowlist_env(env_var: str, ids: list) -> None: """Best-effort persist (empty list removes the key); the pairing store grant still authorizes via the union.""" with contextlib.suppress(Exception): from hermes_cli.config import save_env_value, remove_env_value - save_env_value(env_var, ",".join(ids)) if ids else remove_env_value(env_var) @@ -189,12 +165,10 @@ def _iter_live_gateway_adapters(): runner = None with contextlib.suppress(Exception): from gateway.run import _gateway_runner_ref - runner = _gateway_runner_ref() if runner is None: return - mappings = [getattr(runner, "adapters", None) or {}] - mappings.extend((getattr(runner, "_profile_adapters", None) or {}).values()) + mappings = [getattr(runner, "adapters", None) or {}, *(getattr(runner, "_profile_adapters", None) or {}).values()] for mapping in mappings: for adapter in (mapping or {}).values(): if adapter is not None: @@ -203,16 +177,13 @@ def _iter_live_gateway_adapters(): def _adapter_platform_name(adapter) -> str: value = getattr(getattr(adapter, "platform", None), "value", None) - if value: - return str(value).strip().lower() - return str(getattr(adapter, "name", None) or "").strip().lower() + return str(value or getattr(adapter, "name", None) or "").strip().lower() def _purge_allowlist_entries(entries, platform: str, user_id: str): """Drop alias-equivalent allowlist entries while preserving ``*``.""" def keep(entry) -> bool: return str(entry).strip() == "*" or not _user_ids_match(platform, str(entry), str(user_id)) - if isinstance(entries, str): return ",".join(filter(keep, _split_allowlist(entries))) if isinstance(entries, (set, frozenset)): @@ -233,15 +204,11 @@ def _sync_live_adapter_allowlist_remove(platform: str, user_id: str) -> None: continue if hasattr(adapter, "_allow_from"): with contextlib.suppress(Exception): - adapter._allow_from = _purge_allowlist_entries( - set(adapter._allow_from or ()), platform_name, user_id - ) + adapter._allow_from = _purge_allowlist_entries(set(adapter._allow_from or ()), platform_name, user_id) extra = getattr(getattr(adapter, "config", None), "extra", None) if isinstance(extra, dict) and "allow_from" in extra: with contextlib.suppress(Exception): - extra["allow_from"] = _purge_allowlist_entries( - extra.get("allow_from"), platform_name, user_id - ) + extra["allow_from"] = _purge_allowlist_entries(extra.get("allow_from"), platform_name, user_id) def _sync_allowlist_remove(platform: str, user_id: str) -> None: @@ -249,10 +216,11 @@ def _sync_allowlist_remove(platform: str, user_id: str) -> None: Approve mirrors a normalized phone while revoke is often given a JID/device form, so matching uses alias rules -- exact delete would leave the sender authorized. + An unconfigured allowlist is left alone (config-only snapshots are not touched). """ configured = _configured_allowlist(platform) if configured is None: - return # No allowlist configured — do not touch config-only snapshots. + return env_var, ids = configured remaining = _purge_allowlist_entries(ids, platform, user_id) if len(remaining) == len(ids): @@ -342,28 +310,31 @@ def _is_hashed_entry(entry) -> bool: return isinstance(entry, dict) and "salt" in entry and "hash" in entry +def _entry_created_at(info): + """Numeric ``created_at`` of a pending entry, or None for malformed/legacy entries.""" + created_at = info.get("created_at") if isinstance(info, dict) else None + return created_at if isinstance(created_at, (int, float)) else None + + class PairingStore: """Pairing codes and approved user lists. - Files per platform: ``{platform}-pending.json``, ``{platform}-approved.json``, - plus shared ``_rate_limits.json``. With ``profile=""`` storage resolves from - that profile's HERMES_HOME exactly as ``hermes -p pairing ...`` does, so - multiplex gateways and profile-scoped CLI approvals share one whitelist. + Files per platform: ``{platform}-pending.json``, ``{platform}-approved.json``, plus + shared ``_rate_limits.json``. With ``profile=""`` storage resolves from that + profile's HERMES_HOME exactly as ``hermes -p pairing ...`` does, so multiplex + gateways and profile-scoped CLI approvals share one whitelist. """ def __init__(self, profile: Optional[str] = None): + profile_home = None if profile: root = get_default_hermes_root() profile_home = root if profile == "default" else root / "profiles" / profile - self._dir = get_hermes_dir("platforms/pairing", "pairing", home=profile_home) - else: - profile_home = None - self._dir = _default_pairing_dir() + self._dir = get_hermes_dir("platforms/pairing", "pairing", home=profile_home) if profile else _default_pairing_dir() self._dir.mkdir(parents=True, exist_ok=True) # Merge the alternate old/new layout so upgrades cannot split approvals. _migrate_split_pairing_dirs(home=profile_home, active=self._dir) - # Adapters run concurrently in threads sharing one store. - self._lock = threading.RLock() + self._lock = threading.RLock() # adapters run concurrently in threads sharing one store self._profile = profile # for diagnostics / log lines @property @@ -383,6 +354,9 @@ class PairingStore: _load_json = staticmethod(_load_json_file) _save_json = staticmethod(_save_json_file) + def _platforms(self, platform: Optional[str], suffix: str) -> list: + return [platform] if platform else self._all_platforms(suffix) + # ----- Approved users ----- def is_approved(self, platform: str, user_id: str) -> bool: @@ -391,11 +365,11 @@ class PairingStore: def list_approved(self, platform: str = None) -> list: """List approved users, optionally filtered by platform.""" - results = [] - for p in [platform] if platform else self._all_platforms("approved"): - for uid, info in self._load_json(self._approved_path(p)).items(): - results.append({"platform": p, "user_id": uid, **info}) - return results + return [ + {"platform": p, "user_id": uid, **info} + for p in self._platforms(platform, "approved") + for uid, info in self._load_json(self._approved_path(p)).items() + ] def _approve_user(self, platform: str, user_id: str, user_name: str = "") -> None: """Add a user to the approved list. Must be called under self._lock.""" @@ -403,10 +377,7 @@ class PairingStore: normalized_user_id = _normalize_user_id(platform, user_id) for approved_user_id in _matching_ids(platform, approved, normalized_user_id): del approved[approved_user_id] - approved[normalized_user_id] = { - "user_name": user_name, - "approved_at": time.time(), - } + approved[normalized_user_id] = {"user_name": user_name, "approved_at": time.time()} self._save_json(self._approved_path(platform), approved) # Mirror the grant into the operator's allowlist when one is configured. _sync_allowlist_add(platform, normalized_user_id) @@ -436,11 +407,9 @@ class PairingStore: """Remove a pending request and approve its user. Must hold self._lock.""" del pending[matched_key] self._save_json(self._pending_path(platform), pending) - # A successful approval proves legitimacy, so the persisted brute-force streak # must not carry over (isolated typos would accumulate into a spurious lockout). self._reset_failed_attempts(platform) - result = {"user_id": matched_entry["user_id"], "user_name": matched_entry.get("user_name", "")} self._approve_user(platform, result["user_id"], result["user_name"]) return result @@ -448,29 +417,23 @@ class PairingStore: def generate_code(self, platform: str, user_id: str, user_name: str = "") -> Optional[str]: """Generate a pairing code for a new user. - Returns None if the user is rate-limited, the platform hit - MAX_PENDING_PER_PLATFORM, or the platform is locked out. Only a salted - SHA-256 hash of the code is persisted, never the plaintext. + Returns None if the user is rate-limited, the platform hit MAX_PENDING_PER_PLATFORM, + or the platform is locked out. Only a salted SHA-256 hash of the code is persisted. """ with self._lock: self._cleanup_expired(platform) normalized_user_id = _normalize_user_id(platform, user_id) - if self._is_locked_out(platform) or self._is_rate_limited(platform, user_id): return None pending = self._load_json(self._pending_path(platform)) if len(pending) >= MAX_PENDING_PER_PLATFORM: return None - code = "".join(secrets.choice(ALPHABET) for _ in range(CODE_LENGTH)) salt = os.urandom(16) # Keyed by a random entry id, not the code itself. pending[secrets.token_hex(8)] = { - "hash": self._hash_code(code, salt), - "salt": salt.hex(), - "user_id": normalized_user_id, - "user_name": user_name, - "created_at": time.time(), + "hash": self._hash_code(code, salt), "salt": salt.hex(), + "user_id": normalized_user_id, "user_name": user_name, "created_at": time.time(), } self._save_json(self._pending_path(platform), pending) self._record_rate_limit(platform, user_id) @@ -479,18 +442,16 @@ class PairingStore: def approve_code(self, platform: str, code: str) -> Optional[dict]: """Approve a pairing code and add its user to the approved list. - Returns ``{user_id, user_name}``, or ``None`` if the code is invalid/expired OR - the platform is locked out (disambiguate with ``_is_locked_out``). Constant-time + Returns ``{user_id, user_name}``, or ``None`` if the code is invalid/expired OR the + platform is locked out (disambiguate with ``_is_locked_out``). Constant-time salted-hash compare; legacy plaintext entries are ignored and pruned at TTL. """ with self._lock: self._cleanup_expired(platform) code = code.upper().strip() - # Before the lookup, or an already-issued valid code would bypass lockout. if self._is_locked_out(platform): return None - pending = self._load_json(self._pending_path(platform)) # Skip legacy/malformed entries so an in-place upgrade doesn't crash. for entry_id, entry in pending.items(): @@ -502,7 +463,6 @@ class PairingStore: continue if secrets.compare_digest(self._hash_code(code, salt), entry["hash"]): return self._finish_approval(platform, pending, entry_id, entry) - self._record_failed_attempt(platform) return None @@ -526,7 +486,6 @@ class PairingStore: request_id = str(request_id or "").strip().lower() if not request_id: return None - pending = self._load_json(self._pending_path(platform)) for entry_id, entry in pending.items(): if _is_hashed_entry(entry) and secrets.compare_digest(str(entry_id).lower(), request_id): @@ -538,20 +497,17 @@ class PairingStore: for :meth:`approve_request`; legacy pre-hash entries report an empty id).""" results = [] with self._lock: - for p in [platform] if platform else self._all_platforms("pending"): + for p in self._platforms(platform, "pending"): self._cleanup_expired(p) for entry_id, info in self._load_json(self._pending_path(p)).items(): - if not isinstance(info, dict): - continue - created_at = info.get("created_at") - if not isinstance(created_at, (int, float)): + created_at = _entry_created_at(info) + if created_at is None: continue is_modern = isinstance(info.get("hash"), str) and isinstance(info.get("salt"), str) results.append({ "platform": p, "request_id": str(entry_id) if is_modern else "", - "user_id": info.get("user_id", ""), - "user_name": info.get("user_name", ""), + "user_id": info.get("user_id", ""), "user_name": info.get("user_name", ""), "age_minutes": int((time.time() - created_at) / 60), }) return results @@ -560,35 +516,40 @@ class PairingStore: """Clear all pending requests. Returns count removed.""" with self._lock: count = 0 - for p in [platform] if platform else self._all_platforms("pending"): + for p in self._platforms(platform, "pending"): count += len(self._load_json(self._pending_path(p))) self._save_json(self._pending_path(p), {}) return count # ----- Rate limiting and lockout ----- + def _limits(self) -> dict: + return self._load_json(self._rate_limit_path()) + + def _save_limits(self, limits: dict) -> None: + self._save_json(self._rate_limit_path(), limits) + def _is_rate_limited(self, platform: str, user_id: str) -> bool: """Whether a user (under any alias) has requested a code too recently.""" - limits = self._load_json(self._rate_limit_path()) + limits = self._limits() return any( (time.time() - limits.get(f"{platform}:{alias}", 0)) < RATE_LIMIT_SECONDS for alias in _user_id_aliases(platform, user_id) ) def _record_rate_limit(self, platform: str, user_id: str) -> None: - limits = self._load_json(self._rate_limit_path()) + limits = self._limits() now = time.time() for alias in _user_id_aliases(platform, user_id): limits[f"{platform}:{alias}"] = now - self._save_json(self._rate_limit_path(), limits) + self._save_limits(limits) def _is_locked_out(self, platform: str) -> bool: - limits = self._load_json(self._rate_limit_path()) - return time.time() < limits.get(f"_lockout:{platform}", 0) + return time.time() < self._limits().get(f"_lockout:{platform}", 0) def _record_failed_attempt(self, platform: str) -> None: """Record a failed approval attempt; triggers lockout after MAX_FAILED_ATTEMPTS.""" - limits = self._load_json(self._rate_limit_path()) + limits = self._limits() fail_key = f"_failures:{platform}" fails = limits.get(fail_key, 0) + 1 limits[fail_key] = fails @@ -597,15 +558,15 @@ class PairingStore: limits[fail_key] = 0 print(f"[pairing] Platform {platform} locked out for {LOCKOUT_SECONDS}s " f"after {MAX_FAILED_ATTEMPTS} failed attempts", flush=True) - self._save_json(self._rate_limit_path(), limits) + self._save_limits(limits) def _reset_failed_attempts(self, platform: str) -> None: """Clear the failed-approval counter after a success (it tracks *consecutive* failures).""" - limits = self._load_json(self._rate_limit_path()) + limits = self._limits() fail_key = f"_failures:{platform}" if limits.get(fail_key): limits[fail_key] = 0 - self._save_json(self._rate_limit_path(), limits) + self._save_limits(limits) # ----- Cleanup ----- @@ -616,9 +577,7 @@ class PairingStore: now = time.time() expired = [ entry_id for entry_id, info in pending.items() - if not isinstance(info, dict) - or not isinstance(info.get("created_at"), (int, float)) - or (now - info["created_at"]) > CODE_TTL_SECONDS + if (created := _entry_created_at(info)) is None or (now - created) > CODE_TTL_SECONDS ] if expired: self._save_json(path, {k: v for k, v in pending.items() if k not in expired}) @@ -626,5 +585,5 @@ class PairingStore: def _all_platforms(self, suffix: str) -> list: """Platforms that have a ``-.json`` data file (``_``-prefixed files are shared state).""" tail = f"-{suffix}.json" - platforms = [f.name.replace(tail, "") for f in self._dir.iterdir() if f.name.endswith(tail)] + platforms = (f.name.replace(tail, "") for f in self._dir.iterdir() if f.name.endswith(tail)) return [p for p in platforms if not p.startswith("_")] diff --git a/gateway/platform_registry.py b/gateway/platform_registry.py index 8ceb493b36..97d9ffd1a6 100644 --- a/gateway/platform_registry.py +++ b/gateway/platform_registry.py @@ -1,11 +1,10 @@ """Platform adapter registry. -Platform adapters (built-in and plugin) self-register here so the gateway can -discover and instantiate them without hardcoded if/elif chains. Plugins -register via ``PluginContext.register_platform()``; ``GatewayRunner -._create_adapter()`` consults the registry first and falls back to the legacy -built-in path. Plugin side: ``platform_registry.register(PlatformEntry(...))``; -gateway side: ``platform_registry.create_adapter("irc", platform_config)``. +Adapters (built-in and plugin) self-register here so the gateway can discover and +instantiate them without hardcoded if/elif chains. Plugins register via +``PluginContext.register_platform()``; ``GatewayRunner._create_adapter()`` consults the +registry first, then the legacy built-in path. Plugin side: ``platform_registry +.register(PlatformEntry(...))``; gateway side: ``create_adapter("irc", platform_config)``. """ import logging @@ -26,7 +25,6 @@ def _plugin_scope_from_callable(callback: Callable) -> Optional[str]: """Infer a plugin profile from code registered outside PluginContext.""" try: from tools.registry import registry as tool_registry - return tool_registry.plugin_scope_for_callable(callback) except (ImportError, AttributeError): return None @@ -37,9 +35,7 @@ def _caller_plugin_scope() -> Optional[str]: module_name = sys._getframe(2).f_globals.get("__name__", "") or "" except Exception: return None - return _plugin_scope_from_callable( - type("_Caller", (), {"__module__": module_name}) - ) + return _plugin_scope_from_callable(type("_Caller", (), {"__module__": module_name})) @dataclass @@ -49,14 +45,13 @@ class PlatformEntry: name: str # config.yaml identifier (e.g. "irc") label: str # human-readable adapter_factory: Callable[[Any], Any] # PlatformConfig -> adapter (factory allows custom init) - # PASSIVE dependency probe (deps importable RIGHT NOW). Must be side-effect free: it + # PASSIVE dependency probe (deps importable RIGHT NOW); must be side-effect free since it # runs from status displays and the config enablement pass, which may never pip-install. check_fn: Callable[[], bool] validate_config: Optional[Callable[[Any], bool]] = None # None = let connect() fail descriptively - # ACTIVE installer, run by ``create_adapter()`` only when ``check_fn`` is False (the - # platform is enabled+configured and about to connect). None = a False check_fn is a - # hard block. The split exists because one field either installed from every status - # display or never at all. + # ACTIVE installer, run by ``create_adapter()`` only when ``check_fn`` is False (platform + # enabled+configured, about to connect); None = a False check_fn is a hard block. Split + # from check_fn because one field either installed from every status display or never. ensure_deps_fn: Optional[Callable[[], bool]] = None # Connected/configured for this PlatformConfig (``get_connected_platforms``, setup UI); # None falls back to ``validate_config`` or ``check_fn``. @@ -73,13 +68,12 @@ class PlatformEntry: emoji: str = "🔌" # CLI/gateway display allow_update_command: bool = True # /update may be issued from this platform platform_hint: str = "" # injected into the system prompt; empty = none - # ``() -> Optional[dict]`` of ``extra`` fields to seed when auto-enabled from env. Runs - # in ``_apply_env_overrides`` BEFORE adapter construction so ``gateway status`` sees it. + # ``() -> Optional[dict]`` of ``extra`` fields to seed when auto-enabled from env; runs in + # ``_apply_env_overrides`` BEFORE adapter construction so ``gateway status`` sees it. env_enablement_fn: Optional[Callable[[], Optional[dict]]] = None - # YAML->env bridge ``(yaml_cfg, platform_cfg) -> Optional[dict]`` merged into ``extra``; - # runs after the shared-key loop, before ``_apply_env_overrides``. May set ``os.environ`` - # (guard with ``not os.getenv(...)`` to keep env > YAML). Contract: - # website/docs/developer-guide/adding-platform-adapters.md. + # YAML->env bridge ``(yaml_cfg, platform_cfg) -> Optional[dict]`` merged into ``extra``; runs + # after the shared-key loop, before ``_apply_env_overrides``. May set ``os.environ`` (guard + # with ``not os.getenv(...)`` to keep env > YAML). Contract: docs/developer-guide/adding-platform-adapters.md. apply_yaml_config_fn: Optional[Callable[[dict, dict], Optional[dict]]] = None cron_deliver_env_var: str = "" # home-channel env var read for cron ``deliver=`` # ``(target_ref) -> Optional[(chat_id, thread_id)]`` run before channel-directory @@ -88,30 +82,26 @@ class PlatformEntry: # Post-resolution validation: True accept, False reject, non-empty str = reject + diagnostic. validate_target_ref_fn: Optional[Callable[[str], bool | str]] = None # Whole-request delivery ``(args, normalized_chat_id, platform_name, pconfig)``, sync or - # async. Prefer standalone_sender_fn when the standard send contract suffices. + # async; prefer standalone_sender_fn when the standard send contract suffices. send_message_handler: Optional[Callable[[dict, str, str, Any], Any]] = None - # Out-of-process sender for cron without a co-resident gateway:: - # async (pconfig, chat_id, message, *, thread_id=None, media_files=None, force_document=False) -> dict - # returning ``{"success": True, "message_id": ...}`` or ``{"error": str}``. + # Out-of-process sender for cron without a co-resident gateway: + # ``async (pconfig, chat_id, message, *, thread_id=None, media_files=None, force_document=False) + # -> {"success": True, "message_id": ...} | {"error": str}``. standalone_sender_fn: Optional[Callable[..., Awaitable[dict]]] = None class PlatformRegistry: - """Central registry of platform adapters. - - Registrations are serialized, and concurrent lazy lookups share an - in-flight event while the loader runs outside the registry lock. - """ + """Central registry of platform adapters. Registrations are serialized; concurrent + lazy lookups share an in-flight event while the loader runs outside the registry lock.""" def __init__(self) -> None: self._lock = threading.RLock() - # Process-global registrations (e.g. the built-in relay). - self._entries: dict[str, PlatformEntry] = {} + self._entries: dict[str, PlatformEntry] = {} # process-global (e.g. the built-in relay) # Plugin adapters are isolated per resolved HERMES_HOME and overlay the # process-global entries for lookups in that profile's runtime scope. self._scoped_entries: dict[str, dict[str, PlatformEntry]] = {} # Deferred loaders: name -> callable importing the owning plugin module (which calls - # register()). Eagerly importing ~20 SDK-heavy adapters added seconds to every + # register()); eagerly importing ~20 SDK-heavy adapters added seconds to every # `hermes` invocation, so the import happens only when a lookup asks for it. self._deferred: dict[str, _Loader] = {} self._scoped_deferred: dict[str, dict[str, _Loader]] = {} @@ -157,8 +147,8 @@ class PlatformRegistry: # -- deferred loading ---------------------------------------------------- def register_deferred(self, name: str, loader: _Loader, *, scope: Optional[str] = None) -> None: - """Register a lazy loader (imports the plugin module, which must call :meth:`register`). - Runs at most once, on first lookup; a concrete registration drops it.""" + """Register a lazy loader (imports the plugin module, which must call :meth:`register`); + runs at most once, on first lookup; a concrete registration drops it.""" with self._lock: entries, deferred = self._scope_maps(scope, create=True) self._consumed_loaders.pop((scope, name), None) @@ -168,21 +158,14 @@ class PlatformRegistry: def snapshot_registration( self, name: str, *, scope: Optional[str] = None ) -> tuple[Optional[PlatformEntry], Optional[_Loader]]: - """Concrete and deferred state for *name* without resolving it. - - Lets the plugin ledger restore a deferred loader displaced by a concrete - registration without importing the displaced adapter as a side effect. - """ + """Concrete and deferred state for *name* without resolving it, so the plugin ledger can + restore a deferred loader displaced by a concrete registration without importing it.""" with self._lock: return self._registration_state(scope, name) def restore_registration( - self, - name: str, - current: tuple[Optional[PlatformEntry], Optional[_Loader]], - previous: tuple[Optional[PlatformEntry], Optional[_Loader]], - *, - scope: Optional[str] = None, + self, name: str, current: tuple[Optional[PlatformEntry], Optional[_Loader]], + previous: tuple[Optional[PlatformEntry], Optional[_Loader]], *, scope: Optional[str] = None, ) -> bool: """Restore a registration if its full state is still *current* (CAS): a later registration is never removed, and deferred loaders are part of the state.""" @@ -190,17 +173,12 @@ class PlatformRegistry: entry, loader = self._registration_state(scope, name, create=True) if entry is not current[0] or loader is not current[1]: return False - entries, deferred = self._scope_maps(scope) load_key = (scope, name) - previous_entry, previous_loader = previous - if previous_entry is None: - entries.pop(name, None) - else: - entries[name] = previous_entry - if previous_loader is None: - deferred.pop(name, None) - else: - deferred[name] = previous_loader + for mapping, value in zip(self._scope_maps(scope), previous): + if value is None: + mapping.pop(name, None) + else: + mapping[name] = value if load_key in self._inflight: self._cancelled_inflight.add(load_key) self._consumed_loaders.pop(load_key, None) @@ -210,8 +188,6 @@ class PlatformRegistry: def _resolve(self, name: str, scope: Optional[str] = None) -> None: """Run the deferred loader for *name* if one is pending.""" loader: Optional[_Loader] = None - event: Optional[threading.Event] = None - load_key: _LoadKey is_loader = False with self._lock: active_scope = scope or self.current_scope_key() @@ -223,11 +199,10 @@ class PlatformRegistry: if event is None and name not in entries: loader = deferred.pop(name, None) if event is None and loader is None and name not in entries: + load_key = global_key event = self._inflight.get(global_key) - load_key = global_key - if event is None and loader is None and name not in entries: - loader = self._deferred.pop(name, None) - load_key = global_key + if event is None: + loader = self._deferred.pop(name, None) if event is None and loader is not None: event = threading.Event() self._inflight[load_key] = event @@ -239,15 +214,12 @@ class PlatformRegistry: if not is_loader and self._inflight_owners.get(load_key) == threading.get_ident(): logger.warning("Deferred platform '%s' recursively requested while loading", name) return - if not is_loader: event.wait() - # Teardown may have restored an older deferred generation while - # cancelling the one we waited for; resolve that predecessor now - # instead of returning a one-shot false negative. + # Teardown may have restored an older deferred generation while cancelling the one + # we waited for; resolve that predecessor instead of a one-shot false negative. self._resolve(name, active_scope) return - try: loader() except Exception as e: @@ -272,8 +244,8 @@ class PlatformRegistry: return (scope, name) in self._cancelled_inflight def _resolve_all(self) -> None: - """Run every pending deferred loader (only ``all_entries``/``plugin_entries`` call - this, from paths that need every adapter; CLI chat never iterates the full set).""" + """Run every pending deferred loader (only ``all_entries``/``plugin_entries`` call this; + CLI chat never iterates the full set).""" active_scope = self.current_scope_key() with self._lock: _entries, scoped_deferred = self._scope_maps(active_scope) @@ -284,8 +256,8 @@ class PlatformRegistry: scoped_names.add(name) elif inflight_scope is None: global_names.add(name) - # Load outside the registry lock; each name has an in-flight event so - # concurrent readers wait for the same materialization. + # Load outside the registry lock; each name has an in-flight event so concurrent + # readers wait for the same materialization. for name in (*sorted(scoped_names), *sorted(global_names)): self._resolve(name, active_scope) @@ -304,9 +276,7 @@ class PlatformRegistry: deferred.pop(entry.name, None) prev = entries.get(entry.name) if prev is not None: - logger.info( - "Platform '%s' re-registered (was %s, now %s)", entry.name, prev.source, entry.source - ) + logger.info("Platform '%s' re-registered (was %s, now %s)", entry.name, prev.source, entry.source) entries[entry.name] = entry logger.debug("Registered platform adapter: %s (%s)", entry.name, entry.source) @@ -328,10 +298,8 @@ class PlatformRegistry: """True when a lookup of *name* must run/await a deferred loader (lock held).""" _entries, deferred = self._scope_maps(scope) return ( - name in deferred - or (name not in self._entries and name in self._deferred) - or (scope, name) in self._inflight - or (None, name) in self._inflight + name in deferred or (name not in self._entries and name in self._deferred) + or (scope, name) in self._inflight or (None, name) in self._inflight ) def get(self, name: str) -> Optional[PlatformEntry]: @@ -350,9 +318,7 @@ class PlatformRegistry: """Return all registered platform entries.""" self._resolve_all() with self._lock: - entries = dict(self._entries) - entries.update(self._scoped_entries.get(self.current_scope_key(), {})) - return list(entries.values()) + return list({**self._entries, **self._scoped_entries.get(self.current_scope_key(), {})}.values()) def plugin_entries(self) -> list[PlatformEntry]: """Return only plugin-registered platform entries.""" @@ -366,43 +332,35 @@ class PlatformRegistry: return entries.keys() | deferred.keys() | self._entries.keys() | self._deferred.keys() def is_registered(self, name: str) -> bool: - # A deferred (not-yet-imported) platform still counts as registered so - # cheap membership checks (toolset resolution, webhook deliver-target - # checks) never trigger a heavy import. + # A deferred (not-yet-imported) platform still counts as registered so cheap membership + # checks (toolset resolution, webhook deliver-target checks) never trigger a heavy import. with self._lock: scope = self.current_scope_key() entries, _deferred = self._scope_maps(scope) return name in entries or name in self._entries or self._load_pending(scope, name) def create_adapter(self, name: str, config: Any) -> Optional[Any]: - """Create an adapter instance for *name*. - - Returns None when no entry exists, deps are missing and cannot be - installed, ``validate_config`` fails, or the factory raises. - """ + """Create an adapter instance for *name*; None when no entry exists, deps are missing + and cannot be installed, ``validate_config`` fails, or the factory raises.""" entry = self.get(name) if entry is None: return None - - try: - deps_ok = bool(entry.check_fn()) - except Exception as e: - logger.warning("Platform '%s' check_fn raised: %s", entry.label, e) - deps_ok = False + def _probe(fn: Callable[[], Any], failure_msg: str) -> bool: + try: + return bool(fn()) + except Exception as e: + logger.warning(failure_msg, entry.label, e) + return False + deps_ok = _probe(entry.check_fn, "Platform '%s' check_fn raised: %s") if not deps_ok and entry.ensure_deps_fn is not None: # The ONE place the active installer runs: the platform is enabled+configured # and about to connect, so an install is what the user wants. logger.info("Platform '%s' dependencies missing — attempting install...", entry.label) - try: - deps_ok = bool(entry.ensure_deps_fn()) - except Exception as e: - logger.warning("Platform '%s' dependency install raised: %s", entry.label, e) - deps_ok = False + deps_ok = _probe(entry.ensure_deps_fn, "Platform '%s' dependency install raised: %s") if not deps_ok: hint = f" ({entry.install_hint})" if entry.install_hint else "" logger.warning("Platform '%s' requirements not met%s", entry.label, hint) return None - if entry.validate_config is not None: try: if not entry.validate_config(config): @@ -411,13 +369,10 @@ class PlatformRegistry: except Exception as e: logger.warning("Platform '%s' config validation error: %s", entry.label, e) return None - try: return entry.adapter_factory(config) except Exception as e: - logger.error( - "Failed to create adapter for platform '%s': %s", entry.label, e, exc_info=True - ) + logger.error("Failed to create adapter for platform '%s': %s", entry.label, e, exc_info=True) return None diff --git a/gateway/profile_routing.py b/gateway/profile_routing.py index da403e2dad..7a5f90c863 100644 --- a/gateway/profile_routing.py +++ b/gateway/profile_routing.py @@ -146,7 +146,7 @@ def match_profile_route( thread_id: Optional[str] = None, parent_chat_id: Optional[str] = None, ) -> Optional[ProfileRoute]: """Return the first (most specific) matching route, or None.""" - return next( - (r for r in routes if r.matches(platform, guild_id=guild_id, chat_id=chat_id, thread_id=thread_id, parent_chat_id=parent_chat_id)), - None, - ) + for route in routes: + if route.matches(platform, guild_id=guild_id, chat_id=chat_id, thread_id=thread_id, parent_chat_id=parent_chat_id): + return route + return None