diff --git a/gateway/channel_directory.py b/gateway/channel_directory.py index 5653e7599c..34331c4091 100644 --- a/gateway/channel_directory.py +++ b/gateway/channel_directory.py @@ -157,10 +157,11 @@ async def build_channel_directory(adapters: dict[Any, Any]) -> dict[str, Any]: 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__. + # Plugin platforms are dynamic enum members missing from Platform.__members__. Only connected + # adapters are discovered, so loaded entries suffice (never import an unused adapter here). with contextlib.suppress(Exception): from gateway.platform_registry import platform_registry - for entry in platform_registry.plugin_entries(): + for entry in platform_registry.loaded_entries(): await _discover(entry.name) _apply_channel_aliases(platforms) directory = {"updated_at": datetime.now().isoformat(), "platforms": platforms} diff --git a/gateway/config_env.py b/gateway/config_env.py index ef6c65b17f..3c2280dbe1 100644 --- a/gateway/config_env.py +++ b/gateway/config_env.py @@ -448,7 +448,12 @@ def _enable_plugin_platforms_from_env(config: GatewayConfig) -> None: from hermes_cli.plugins import discover_plugins discover_plugins() # idempotent from gateway.platform_registry import platform_registry - for entry in platform_registry.plugin_entries(): + from gateway.platform_activation import configured_plugin_entries + # Only platforms with activation evidence (config block, env/.env, credential pool) are + # imported; an unconfigured adapter has nothing to enable. See gateway.platform_activation. + # config.yaml blocks already materialized their adapters in load_yaml_layer (and every merged + # block is in config.platforms), so the YAML is not parsed a second time here. + for entry in configured_plugin_entries(platform_registry, yaml_cfg={}, platform_names=config.platforms): _enable_plugin_platform(config, entry) except Exception as e: logger.debug("Plugin platform enable pass failed: %s", e) diff --git a/gateway/config_loader.py b/gateway/config_loader.py index 5c52aacc28..3328704ceb 100644 --- a/gateway/config_loader.py +++ b/gateway/config_loader.py @@ -244,10 +244,15 @@ def _bridged_keys(plat: Platform, platform_cfg: dict, gw_data: dict, *, root_blo return bridged -def shared_loop_targets(registry) -> list: - """Built-in platforms plus registered plugin platforms (so plugin authors get shared-key bridging).""" +def shared_loop_targets(registry, entries=None) -> list: + """Built-in platforms plus registered plugin platforms (so plugin authors get shared-key bridging). + + *entries* are the plugin entries to bridge (``load_yaml_layer`` passes only the configured + ones, see ``gateway.platform_activation``); None means every registered plugin platform.""" targets: list = list(Platform) - for entry in registry.plugin_entries() if registry is not None else (): + if entries is None: + entries = registry.plugin_entries() if registry is not None else () + for entry in entries: with contextlib.suppress(ValueError, KeyError): if (plat := Platform(entry.name)) not in targets: targets.append(plat) @@ -379,14 +384,16 @@ def bridge_platform_shared_keys( def apply_plugin_yaml_hooks( yaml_cfg: dict, gateway_platforms: Any, platforms_data: dict, registry, warned: Optional[set] = None, authored: Optional[dict] = None, managed_extra: Optional[dict] = None, + entries: Optional[list] = None, ) -> None: """Plugin-owned YAML→env config bridges (``PlatformEntry.apply_yaml_config_fn``). Order: shared-key - loop → this dispatch → core-only bridges (require_mention/signal) → ``_apply_env_overrides()``.""" + loop → this dispatch → core-only bridges (require_mention/signal) → ``_apply_env_overrides()``. + *entries* limits the dispatch to those platforms (None = every registered entry).""" if registry is None: return if authored is None: authored = snapshot_authored_extra(platforms_data) - for entry in registry.all_entries(): + for entry in registry.all_entries() if entries is None else entries: # Plugin-owned YAML→env config bridges (#24836). See ``PlatformEntry.apply_yaml_config_fn`` for the # hook contract. Order: shared-key loop (above) → this dispatch → legacy hardcoded blocks (below; # no-op when a hook already set their env var) → ``_apply_env_overrides()`` after @@ -529,7 +536,14 @@ def load_yaml_layer(home: Path, gw_data: dict) -> None: logger.debug("plugin discovery skipped: %s", e) registry = None - targets = shared_loop_targets(registry) + # Only platforms this profile configures (a block above, a gateway.json row, their env/.env) are + # materialized: importing every deferred adapter to read its hooks cost each cold start ~0.5 s. + entries = None + if registry is not None: + from gateway.platform_activation import activation_predicate + entries = registry.configured_entries(activation_predicate( + yaml_cfg=yaml_cfg, platform_names=platforms_data, home=home, registry=registry)) + targets = shared_loop_targets(registry, None if entries is None else [e for e in entries if e.source == "plugin"]) # "Authored" is config.yaml only: a scratch merge leaves out the legacy gateway.json extra # already in ``platforms_data``, which stays the base layer every config.yaml key overrides. authored = snapshot_authored_extra(merge_platform_sections(yaml_cfg, gateway_section, {})) @@ -539,5 +553,5 @@ def load_yaml_layer(home: Path, gw_data: dict) -> None: bridge_platform_shared_keys( yaml_cfg, gateway_platforms, gw_data, platforms_data, targets, warned=warned, authored=authored) apply_plugin_yaml_hooks(yaml_cfg, gateway_platforms, platforms_data, registry, - warned=warned, authored=authored, managed_extra=managed_extra) + warned=warned, authored=authored, managed_extra=managed_extra, entries=entries) bridge_core_env_settings(yaml_cfg, platforms_data) diff --git a/gateway/platform_activation.py b/gateway/platform_activation.py new file mode 100644 index 0000000000..46f5f8c65d --- /dev/null +++ b/gateway/platform_activation.py @@ -0,0 +1,142 @@ +"""Which deferred platform adapters a gateway config/env pass must import. + +Bundled platform plugins register lazily (``PlatformRegistry.register_deferred``); importing all +~20 of them costs every cold gateway start about half a second (discord, telegram, google_chat +SDKs, telegram's i18n catalog) although a fresh profile configures none. The config passes in +``gateway.config_loader`` / ``gateway.config_env`` only need a platform's hooks +(``apply_yaml_config_fn``, ``env_enablement_fn``, ``is_connected``) when the profile could be +configuring that platform, so they materialize only names with *activation evidence*: + +* a config block for it (top-level ``:``, ``platforms.``, ``gateway.platforms.``, + ``gateway.``, or a legacy ``gateway.json`` row) — including a sibling block such as + ``wecom_callback`` (registered by the ``wecom`` plugin); +* an env var (process env, the profile ``.env``, or the active secret scope) under the platform's + prefix (``TELEGRAM_*``, ``GOOGLE_CHAT_*``) or named in its manifest ``requires_env`` + (``TWILIO_*`` for sms); +* a credential-pool record for it in the profile ``auth.json`` (Photon stores project creds there). + +The rule over-approximates on purpose: a false positive costs one import, a false negative would +leave a configured platform down. Any lookup by name (``platform_registry.get``) still imports a +deferred adapter on demand. +""" + +from __future__ import annotations + +import json +import logging +import os +from pathlib import Path +from typing import Any, Callable, Iterable, Optional + +logger = logging.getLogger(__name__) + + +def _yaml_block_names(yaml_cfg: Any) -> set[str]: + """Keys of every mapping a platform block may live in (``platform_section`` / + ``merge_platform_sections`` read the same places): ``:``, ``platforms.``, + ``gateway.``, ``gateway.platforms.``.""" + if not isinstance(yaml_cfg, dict): + return set() + gateway: dict = yaml_cfg["gateway"] if isinstance(yaml_cfg.get("gateway"), dict) else {} + parents = (yaml_cfg, yaml_cfg.get("platforms"), gateway, gateway.get("platforms")) + return {str(k) for parent in parents if isinstance(parent, dict) + for k, v in parent.items() if isinstance(v, dict)} + + +def _env_names(home: Optional[Path]) -> set[str]: + names = {k for k, v in os.environ.items() if str(v).strip()} + try: + from agent.secret_scope import current_secret_scope, load_env_file + scope = current_secret_scope() + if scope is not None: + names.update(k for k, v in scope.items() if str(v).strip()) + if home is not None: + names.update(k for k, v in load_env_file(home / ".env").items() if str(v).strip()) + # Values an external secret source (Bitwarden, 1Password) resolved for this home, and + # the administrator-managed .env: both feed the profile scope the adapters read. + from hermes_cli.env_loader import get_secret_source_values + names.update(k for k, v in get_secret_source_values(home).items() if str(v).strip()) + from hermes_cli.managed_scope import load_managed_env + names.update(k for k, v in load_managed_env().items() if str(v).strip()) + except Exception: + logger.debug("platform activation: env scan degraded", exc_info=True) + return names + + +def _credential_pool_names(home: Optional[Path]) -> set[str]: + if home is None: + return set() + try: + with open(home / "auth.json", encoding="utf-8-sig") as handle: + pool = (json.load(handle) or {}).get("credential_pool") or {} + return {str(k) for k in pool} if isinstance(pool, dict) else set() + except FileNotFoundError: + return set() + except Exception: + logger.debug("platform activation: auth.json unreadable", exc_info=True) + return set() + + +def _matches(name: str, candidates: Iterable[str]) -> bool: + lowered = name.lower() + return any(c.lower() == lowered or c.lower().startswith(lowered + "_") for c in candidates) + + +def _legacy_gateway_json_names(home: Path) -> set[str]: + try: + with open(home / "gateway.json", encoding="utf-8-sig") as handle: + rows = (json.load(handle) or {}).get("platforms") or {} + return {str(k) for k in rows} if isinstance(rows, dict) else set() + except FileNotFoundError: + return set() + except (OSError, ValueError, AttributeError): + # Unreadable or not a {"platforms": {...}} document: the YAML/env/pool evidence still applies. + logger.debug("platform activation: gateway.json unreadable", exc_info=True) + return set() + + +def activation_predicate( + *, yaml_cfg: Any = None, platform_names: Iterable[str] = (), home: Optional[Path] = None, + registry: Any = None, +) -> Callable[[str], bool]: + """``name -> bool``: does this profile show evidence of configuring platform *name*? + + *yaml_cfg* is the loaded config.yaml layer (read from *home* when omitted), *platform_names* + any already-merged platform keys (``GatewayConfig.platforms``). *home* defaults to the active + Hermes home. + """ + if home is None: + from hermes_constants import get_hermes_home + home = get_hermes_home() + if yaml_cfg is None and home is not None: + try: + from gateway.config_loader import read_yaml_layers + yaml_cfg = read_yaml_layers(home) + except Exception: + logger.debug("platform activation: config.yaml unreadable", exc_info=True) + blocks = _yaml_block_names(yaml_cfg) | {str(getattr(n, "value", n)) for n in platform_names} + if home is not None: + blocks |= _legacy_gateway_json_names(home) + env = _env_names(home) + upper_env = {e.upper() for e in env} + pools = _credential_pool_names(home) + + def has_evidence(name: str) -> bool: + if _matches(name, blocks) or _matches(name, pools): + return True + prefix = name.upper().replace("-", "_") + "_" + if any(e.startswith(prefix) for e in upper_env): + return True + declared = registry.activation_env(name) if registry is not None else frozenset() + return any(d.upper() in upper_env for d in declared) + + return has_evidence + + +def configured_plugin_entries(registry: Any, **evidence: Any) -> list: + """Plugin entries for platforms this profile configures (see module docstring); already-loaded + entries are always included. Never imports an adapter with no activation evidence.""" + if registry is None: + return [] + wanted = activation_predicate(registry=registry, **evidence) + return [e for e in registry.configured_entries(wanted) if e.source == "plugin"] diff --git a/gateway/platform_registry.py b/gateway/platform_registry.py index cd35d032f3..2ec5558541 100644 --- a/gateway/platform_registry.py +++ b/gateway/platform_registry.py @@ -156,6 +156,10 @@ class PlatformRegistry: # A failed loader is no longer discoverable, but its identity remains # until ownership teardown can CAS-restore the displaced predecessor. self._consumed_loaders: dict[_LoadKey, _Loader] = {} + # Manifest-declared env names per deferred platform (``requires_env``), recorded without + # importing the adapter so ``gateway.platform_activation`` can tell a configured platform + # from an unconfigured one. A hint only: a stale entry costs one extra import, never a miss. + self._activation_env: dict[str, frozenset[str]] = {} @staticmethod def current_scope_key() -> str: @@ -197,6 +201,16 @@ class PlatformRegistry: if name not in entries: deferred[name] = loader + def note_activation_env(self, name: str, env_names) -> None: + """Record the env names a deferred platform's manifest declares (see ``_activation_env``).""" + names = frozenset(str(n).strip() for n in env_names or () if str(n).strip()) + with self._lock: + self._activation_env[name] = self._activation_env.get(name, frozenset()) | names + + def activation_env(self, name: str) -> frozenset[str]: + with self._lock: + return self._activation_env.get(name, frozenset()) + def snapshot_registration( self, name: str, *, scope: Optional[str] = None ) -> tuple[Optional[PlatformEntry], Optional[_Loader]]: @@ -289,10 +303,8 @@ class PlatformRegistry: with self._lock: return (scope, name) in self._cancelled_inflight - def _resolve_all(self) -> None: - """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() + def _pending_names(self, active_scope: str) -> list[str]: + """Deferred (or in-flight) names for *active_scope*, scoped first, without loading them.""" with self._lock: _entries, scoped_deferred = self._scope_maps(active_scope) scoped_names = set(scoped_deferred) @@ -302,10 +314,37 @@ class PlatformRegistry: scoped_names.add(name) elif inflight_scope is None: global_names.add(name) + return [*sorted(scoped_names), *sorted(global_names - scoped_names)] + + def _resolve_all(self, wanted: Optional[Callable[[str], bool]] = None) -> None: + """Run every pending deferred loader, or only those *wanted* accepts (``all_entries`` / + ``plugin_entries`` / ``configured_entries``; CLI chat never iterates the full set).""" + active_scope = self.current_scope_key() # 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) + for name in self._pending_names(active_scope): + if wanted is None or wanted(name): + self._resolve(name, active_scope) + + def pending_names(self) -> list[str]: + """Names registered as deferred loaders that have not been imported yet (current scope).""" + return self._pending_names(self.current_scope_key()) + + def configured_entries(self, wanted: Callable[[str], bool]) -> list[PlatformEntry]: + """Loaded entries, after materializing only the deferred platforms *wanted* accepts. + + The gateway's config/env passes need each platform's hooks only for platforms the profile + actually configures; importing every bundled adapter (~20 SDKs) just to read metadata cost + every cold start ~0.5 s. *wanted* is :func:`gateway.platform_activation.has_activation_evidence` + (or a test stub); entries that are already loaded are always returned.""" + self._resolve_all(wanted) + with self._lock: + return list({**self._entries, **self._scoped_entries.get(self.current_scope_key(), {})}.values()) + + def loaded_entries(self) -> list[PlatformEntry]: + """Entries already materialized (never imports a deferred adapter).""" + with self._lock: + return list({**self._entries, **self._scoped_entries.get(self.current_scope_key(), {})}.values()) def register(self, entry: PlatformEntry, *, scope: Optional[str] = None) -> None: """Register a platform adapter entry (last writer wins on name clash).""" diff --git a/gateway/run_startup.py b/gateway/run_startup.py index 286136d53b..9316309e87 100644 --- a/gateway/run_startup.py +++ b/gateway/run_startup.py @@ -1033,7 +1033,11 @@ class GatewayStartupMixin: allow_all_vars = ["GATEWAY_ALLOW_ALL_USERS", *self._BUILTIN_ALLOW_ALL_VARS] with suppress(Exception): from gateway.platform_registry import platform_registry - entries = platform_registry.plugin_entries() + # Enabled plugin platforms were materialized by the config passes; an unconfigured + # adapter's allowlist var cannot gate anything, so never import one just to read it. + entries = [e for e in (*platform_registry.loaded_entries(), *filter(None, ( + platform_registry.get(p.value) for p, cfg in self.config.platforms.items() if cfg.enabled))) + if e.source == "plugin"] allowed_vars += [e.allowed_users_env for e in entries if e.allowed_users_env] allow_all_vars += [e.allow_all_env for e in entries if e.allow_all_env] # An API-server/webhook/local-only gateway has no messaging sender to gate, so the diff --git a/hermes_cli/plugins_loader.py b/hermes_cli/plugins_loader.py index 36f97b0a72..d5af7c2f07 100644 --- a/hermes_cli/plugins_loader.py +++ b/hermes_cli/plugins_loader.py @@ -307,6 +307,10 @@ class PluginLoaderMixin: previous = platform_registry.snapshot_registration(platform_name, scope=scope) platform_registry.register_deferred(platform_name, _loader, scope=scope) + # Lets the gateway config pass recognise an env-only setup (sms: TWILIO_*) without + # importing the adapter; see gateway.platform_activation. + platform_registry.note_activation_env( + platform_name, [e.get("name") if isinstance(e, dict) else e for e in manifest.requires_env or ()]) current = platform_registry.snapshot_registration(platform_name, scope=scope) if current[0] is None and current[1] is _loader: self._plugin_platform_names.add(platform_name) diff --git a/tests/gateway/test_cold_platform_activation.py b/tests/gateway/test_cold_platform_activation.py new file mode 100644 index 0000000000..463193b1ca --- /dev/null +++ b/tests/gateway/test_cold_platform_activation.py @@ -0,0 +1,73 @@ +"""Cold gateway config load imports only the platform adapters the profile configures. + +``platform_registry.plugin_entries()`` materializes every deferred bundled adapter (discord, +telegram, google_chat SDKs, telegram's i18n catalog: ~0.5 s per cold start). The gateway config / +env passes must resolve only platforms with activation evidence (config block, env or .env under +the platform prefix / manifest ``requires_env``, credential pool), and a configured platform — +env-only included — must still be enabled. +""" + +import json +import os +import subprocess +import sys +import textwrap +from pathlib import Path + +import pytest + +REPO = Path(__file__).resolve().parents[2] + +_PROBE = textwrap.dedent(""" + import json, sys + from gateway.config import load_gateway_config + from gateway.platform_registry import platform_registry + config = load_gateway_config() + print(json.dumps({ + "enabled": sorted(p.value for p, c in config.platforms.items() if c.enabled), + "adapters": sorted(m.split("__", 1)[1].split(".")[0] for m in sys.modules + if m.startswith("hermes_plugins.platforms__") and m.endswith(".adapter")), + "registered": sorted(platform_registry.registered_names()), + })) +""") + + +def _load(tmp_path, *, config_yaml="", env=None, dotenv=""): + home = tmp_path / ".hermes" + home.mkdir() + (home / "config.yaml").write_text(config_yaml, encoding="utf-8") + (home / ".env").write_text(dotenv, encoding="utf-8") + # A clean child: the suite process has already imported adapters, and the developer's shell + # may carry platform env vars that are genuine activation evidence. + child_env = {k: v for k, v in os.environ.items() + if k in {"PATH", "LANG", "TZ", "HERMES_RUNTIME_DIR", "TMPDIR", "SYSTEMROOT"}} + child_env.update(HOME=str(tmp_path), HERMES_HOME=str(home), PYTHONPATH=str(REPO), **(env or {})) + out = subprocess.run([sys.executable, "-c", _PROBE], cwd=str(REPO), env=child_env, + capture_output=True, text=True, timeout=120) + assert out.returncode == 0, out.stderr[-2000:] + return json.loads(out.stdout.strip().splitlines()[-1]) + + +def test_unconfigured_profile_imports_no_platform_adapter(tmp_path): + result = _load(tmp_path) + assert result["enabled"] == [] + assert result["adapters"] == [], f"unconfigured adapters imported at config load: {result['adapters']}" + # Still registered (deferred): a later lookup by name imports it on demand. + assert {"discord", "telegram", "google_chat", "slack"} <= set(result["registered"]) + + +@pytest.mark.parametrize("case", ["env", "dotenv", "manifest_env", "yaml_sibling"]) +def test_configured_platform_still_enabled_and_only_it_is_imported(tmp_path, case): + kwargs, want, module = { + "env": ({"env": {"TELEGRAM_BOT_TOKEN": "123456:fake-token-abcdef"}}, "telegram", "telegram"), + "dotenv": ({"dotenv": "TELEGRAM_BOT_TOKEN=123456:fake-token-abcdef\n"}, "telegram", "telegram"), + # sms is enabled by TWILIO_* (its manifest requires_env), not an SMS_ prefix. + "manifest_env": ({"env": {"TWILIO_ACCOUNT_SID": "ACfake", "TWILIO_AUTH_TOKEN": "x", + "TWILIO_PHONE_NUMBER": "+15550000000"}}, "sms", "sms"), + # wecom_callback is registered by the wecom plugin. + "yaml_sibling": ({"config_yaml": "platforms:\n wecom_callback:\n enabled: true\n extra:\n" + " corp_id: c\n corp_secret: s\n"}, "wecom_callback", "wecom"), + }[case] + result = _load(tmp_path, **kwargs) + assert want in result["enabled"] + assert result["adapters"] == [module], result