perf(gateway): config load imports only the platform adapters a profile configures

A cold `hermes chat -q` in a fresh home spent ~0.5 s in load_gateway_config: shared_loop_targets,
apply_plugin_yaml_hooks and the env enable pass called platform_registry.plugin_entries(), which
imports every deferred bundled adapter (discord, telegram + its i18n catalog, google_chat, ... 21
in all) just to read hook metadata, with zero platforms configured. The next boot-path
iterators (_start_check_access_policy, build_channel_directory) did the same.

The config/env passes now resolve only platforms with activation evidence
(gateway/platform_activation.py): a config block (top-level, platforms.*, gateway.platforms.*,
gateway.<name>, gateway.json rows, sibling names such as wecom_callback), an env/.env/secret-scope/
secret-source/managed-env var under the platform prefix or named in the manifest's requires_env
(recorded at deferred registration, so sms's TWILIO_* counts), or a credential-pool record
(photon). The rule over-approximates: a false positive costs one import. The allowlist reminder
reads loaded + enabled entries; the channel directory only discovers connected adapters, so it
reads loaded entries. Lookups by name still import on demand.

Fresh home: 0 of 21 adapters imported at config load (was 21). Env-only telegram/sms/irc/
google_chat, .env-only telegram, YAML discord/wecom_callback and auth.json photon enable exactly as
before and import only their own adapter.

Review: https://github.com/NousResearch/hermes-agent/pull/106742#issuecomment-6073624820
(cherry picked from commit c244bdfc070a7ea67e7ded2f3dc022fd7ce6a65c)

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}

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)

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)

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 ``<name>:``, ``platforms.<name>``, ``gateway.platforms.<name>``,
+  ``gateway.<name>``, 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): ``<name>:``, ``platforms.<name>``,
+    ``gateway.<name>``, ``gateway.platforms.<name>``."""
+    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"]

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)."""

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

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)

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