"""config.yaml / gateway.json → ``GatewayConfig.from_dict`` schema (the ``load_gateway_config`` phases). Precedence for top-level keys: key-presence at the TOP LEVEL of config.yaml wins; the nested ``gateway.`` form (what ``hermes config set gateway.`` produces) is consulted only when the top-level key is absent — not merely falsy/mistyped — so a present-but-empty top-level value is never silently replaced by the nested one. Both overwrite whatever legacy gateway.json set. """ import contextlib import json import logging import os from pathlib import Path from typing import Any, Optional from gateway.config import UNAUTHORIZED_DM_BEHAVIORS, Platform, PlatformConfig, _coerce_dict, _dict_slot, _normalize_choice # Logger name parity with the origin module: records stay under "gateway.config". logger = logging.getLogger("gateway.config") def load_legacy_gateway_json(home: Path) -> Any: """Legacy ``gateway.json`` base layer (config.yaml keys always win). Malformed → ``{}`` + warning.""" path = home / "gateway.json" if not path.exists(): return {} try: with open(path, "r", encoding="utf-8-sig") as f: data = json.load(f) or {} logger.info("Loaded legacy %s — consider moving settings to config.yaml", path) return data except Exception as e: logger.warning("Failed to load %s: %s", path, e) return {} # --- top-level key bridging ---------------------------------------------------- # # Top-level settings are also accepted nested under ``gateway:`` (what ``hermes config set # gateway.`` produces). This loader builds gw_data FLAT and never forwards the yaml ``gateway:`` # section, so even keys GatewayConfig.from_dict can fall back on itself (loop_watchdog*, # multiplex_profiles, ...) must be bridged here or they are silently ignored on real startup. # # Fallback modes (how the nested ``gateway.`` form is consulted): # "presence": top-level key present → its value; else nested key present. # "gwdata": like "presence", but nested only when NOTHING (config.yaml or gateway.json) set it yet. # "none": top-level VALUE is None → nested value. # "dict": top-level value is not a mapping → nested value; accepted only if a mapping. # "nested": nested form only (no top-level spelling is bridged). # True while GATEWAY_ALLOW_ALL_USERS in os.environ is the bridge's own write (from config.yaml), not an # operator's env var: only then may a reload overwrite/clear it, and restart env builders drop it so a # child gateway re-derives the grant from its config.yaml instead of inheriting a stale open posture. _BRIDGED_ALLOW_ALL_USERS = False def bridged_allow_all_users() -> Optional[str]: """``os.environ['GATEWAY_ALLOW_ALL_USERS']`` when it is the bridge's own write, else None.""" return os.environ.get("GATEWAY_ALLOW_ALL_USERS") if _BRIDGED_ALLOW_ALL_USERS else None def drop_bridged_env(env: dict) -> dict: """Remove the bridge-owned ``GATEWAY_ALLOW_ALL_USERS`` from a child-process env (operator-set stays).""" if bridged_allow_all_users() is not None: env.pop("GATEWAY_ALLOW_ALL_USERS", None) return env def _quick_commands_ok(value: Any) -> bool: if isinstance(value, dict): return True logger.warning( "Ignoring invalid quick_commands in config.yaml (expected mapping, got %s)", type(value).__name__, ) return False def _dm_behavior_choice(value: Any, default: str = "pair") -> str: return _normalize_choice(value, UNAUTHORIZED_DM_BEHAVIORS, default) def _presence(*keys: str) -> tuple: return tuple((k, k, "presence", None, None) for k in keys) # (yaml key, gw_data key, mode, accept(value) -> bool, transform(value)) _TOPLEVEL_BRIDGE: tuple = ( ("quick_commands", "quick_commands", "none", _quick_commands_ok, 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("room_link_url"), ("profile_routes", "profile_routes", "none", lambda v: isinstance(v, list), None), *_presence("max_concurrent_sessions"), ("systemd_watchdog_seconds", "systemd_watchdog_seconds", "nested", None, None), ("streaming", "streaming", "dict", None, None), *_presence( "reset_triggers", "always_log_local", "write_sessions_json", "loop_watchdog", "loop_watchdog_probe_interval_s", "loop_watchdog_probe_timeout_s", "loop_watchdog_max_strikes", "filter_silence_narration", ), ("unauthorized_dm_behavior", "unauthorized_dm_behavior", "presence", None, _dm_behavior_choice), *_presence("unauthorized_dm_decline_message"), ) def _bridge_lookup(yaml_cfg: dict, gateway_section: Any, gw_data: dict, key: str, mode: str) -> tuple: """Return ``(found, value)`` for *key* under the given fallback *mode*.""" nested = isinstance(gateway_section, dict) if mode in ("presence", "gwdata"): if key in yaml_cfg: return True, yaml_cfg[key] if nested and key in gateway_section and (mode == "presence" or key not in gw_data): return True, gateway_section[key] return False, None if mode == "nested": return (True, gateway_section[key]) if nested and key in gateway_section else (False, None) value = yaml_cfg.get(key) if mode == "none": if value is None and nested: value = gateway_section.get(key) return value is not None, value # "dict" if not isinstance(value, dict) and nested: value = gateway_section.get(key) return isinstance(value, dict), value def bridge_toplevel_keys(yaml_cfg: dict, gateway_section: Any, gw_data: dict) -> None: for yaml_key, gw_key, mode, accept, transform in _TOPLEVEL_BRIDGE: found, value = _bridge_lookup(yaml_cfg, gateway_section, gw_data, yaml_key, mode) if not found or (accept is not None and not accept(value)): continue gw_data[gw_key] = transform(value) if transform else value # --- platform sections ----------------------------------------------------------- def merge_platform_sections(yaml_cfg: dict, gateway_cfg: Any, gw_data: dict) -> dict: """Merge every place a platform block may live into ``gw_data["platforms"]`` and return it. Order (later wins on shared keys, ``extra`` deep-merged so gateway.json defaults survive): ``gateway.platforms.*`` → top-level ``platforms.*`` → ``gateway.`` subsections (nested first so top-level config keeps precedence, matching the gateway.streaming fallback). An ``enabled`` key in any block sets the ``_enabled_explicit`` marker consumed by the env pass. Top-level adapter keys (``gateway.api_server.port: 8642``) reach ``extra`` in ``PlatformConfig.from_dict``. """ platforms_data = _dict_slot(gw_data, "platforms") def merge(source_platforms: Any) -> None: if not isinstance(source_platforms, dict): return for plat_name, plat_block in source_platforms.items(): if not isinstance(plat_block, dict): continue existing = platforms_data.get(plat_name, {}) existing = existing if isinstance(existing, dict) else {} merged_extra = {**existing.get("extra", {}), **plat_block.get("extra", {})} if "enabled" in plat_block: merged_extra["_enabled_explicit"] = True merged = {**existing, **plat_block} if merged_extra: merged["extra"] = merged_extra platforms_data[plat_name] = merged nested_gateway = gateway_cfg if isinstance(gateway_cfg, dict) else {} merge(nested_gateway.get("platforms")) merge(yaml_cfg.get("platforms")) merge({k: v for k, v in nested_gateway.items() if k != "platforms" and isinstance(v, dict) and _is_platform_name(k)}) return platforms_data def _is_platform_name(key: Any) -> bool: try: Platform(key) except (ValueError, AttributeError): return False return True def platform_section(yaml_cfg: dict, name: str, gateway_platforms: Any) -> tuple: """``(section, is_toplevel)`` for platform *name*: a top-level ``:`` block wins; otherwise the block under ``gateway.platforms`` / ``platforms`` so shared-key bridging and adapter hooks still run for nested-only configs.""" section = yaml_cfg.get(name) toplevel = isinstance(section, dict) if not toplevel: nested = (src[name] for src in (gateway_platforms, yaml_cfg.get("platforms")) if isinstance(src, dict) and isinstance(src.get(name), dict)) section = next(nested, section) return section, toplevel def _str_keyed(value: Any) -> Any: return {str(k): v for k, v in value.items()} if isinstance(value, dict) else value _TELEGRAM = frozenset({Platform.TELEGRAM}) _DISCORD_SLACK = frozenset({Platform.DISCORD, Platform.SLACK}) def _plain(*keys: str) -> tuple: """Keys copied verbatim into ``extra`` for every platform.""" return tuple((k, None, None) for k in keys) # (key, platforms-or-None, transform-or-None), in ``extra`` insertion order. # ``"dm"`` defers to the global unauthorized_dm_behavior. _SHARED_KEYS: tuple = ( ("unauthorized_dm_behavior", None, "dm"), ("notice_delivery", None, lambda v: _normalize_choice(v, {"public", "private"}, "public")), *_plain("reply_prefix", "reply_in_thread", "cron_continuable_surface", "require_mention", "send_read_receipts"), ("allowed_chats", _TELEGRAM, None), ("group_allowed_chats", _TELEGRAM, None), ("allowed_topics", _TELEGRAM, None), *_plain("free_response_channels", "mention_patterns", "exclusive_bot_mentions"), ("observe_unmentioned_group_messages", _TELEGRAM, None), *_plain( "dm_policy", "allow_from", "allow_admin_from", "user_allowed_commands", "group_policy", "group_allow_from", "group_allow_admin_from", "group_user_allowed_commands", ), ("channel_skill_bindings", _DISCORD_SLACK, None), ("channel_prompts", None, _str_keyed), *_plain("gateway_restart_notification", "typing_indicator", "typing_status_text"), ) def _bridged_keys(plat: Platform, platform_cfg: dict, gw_data: dict, *, root_block: bool = False) -> dict: """Shared-key bridge; a ROOT-level ``:`` block (which ``merge_platform_sections`` never copies into ``platforms_data``) also gets its adapter keys promoted into ``extra``, with the same typed-key exclusion and explicit-``extra`` precedence as ``PlatformConfig.from_dict``.""" bridged: dict = {} if root_block: typed = PlatformConfig._TYPED_KEYS | {"channel_overrides"} bridged.update({k: v for k, v in platform_cfg.items() if k not in typed}) bridged.update(_coerce_dict(platform_cfg.get("extra", {}))) for key, only, transform in _SHARED_KEYS: if key not in platform_cfg or (only is not None and plat not in only): continue if transform == "dm": bridged[key] = _dm_behavior_choice(platform_cfg[key], gw_data.get("unauthorized_dm_behavior", "pair")) else: bridged[key] = transform(platform_cfg[key]) if transform else platform_cfg[key] return bridged def shared_loop_targets(registry) -> list: """Built-in platforms plus registered plugin platforms (so plugin authors get shared-key bridging).""" targets: list = list(Platform) for entry in registry.plugin_entries() if registry is not None else (): with contextlib.suppress(ValueError, KeyError): if (plat := Platform(entry.name)) not in targets: targets.append(plat) return targets def snapshot_authored_extra(platforms_data: dict) -> dict: """``{plat_name: extra}`` as the user authored it under ``platforms..extra``, captured right after ``merge_platform_sections`` and BEFORE either copy site runs. Reading ``platforms_data`` live at a copy site would mistake values the earlier site bridged from the block's own ``extra:`` sub-dict for authored ones (``telegram: {reply_to_mode: all, extra: {reply_to_mode: off}}`` must still resolve to ``all``). Copies, never creates slots.""" return { name: dict(_coerce_dict(plat.get("extra"))) for name, plat in platforms_data.items() if isinstance(plat, dict) } def _apply_managed_extra(authored: dict, managed: dict, platforms_data: dict) -> dict: """Remove from *authored* (in place) every key the administrator pinned in a managed ``:`` block: the user's ``platforms..extra`` must not outrank it. A key the managed layer itself sets under ``platforms..extra`` stays authored.""" if not managed: return {} from hermes_cli.config import _deep_merge managed_platforms = merge_platform_sections(managed, managed.get("gateway"), {}) hook_pins = {} for name, extra in authored.items(): root = _coerce_dict(managed.get(name)) nested = _coerce_dict(managed_platforms.get(name)) pinned = set(root) | set(_coerce_dict(root.get("extra"))) | set(nested) for key in (pinned - set(_coerce_dict(nested.get("extra")))) & set(extra): del extra[key] # Cross-shape merges can replace managed gateway.platforms.extra with the # user platforms.extra. Restore values, not just their key membership. managed_extra = _coerce_dict(nested.get("extra")) extra.update(_deep_merge(extra, managed_extra)) # Restore at the destination too: an unrelated root block need not mention # any pinned key, so the authored-conflict overlay alone cannot carry it. if managed_extra: destination = _dict_slot(_dict_slot(platforms_data, name), "extra") destination.update(_deep_merge(destination, managed_extra)) hook_pins[name] = {key: extra[key] for key in managed_extra if key not in PlatformConfig._TYPED_KEYS} return hook_pins def _authored_wins(extra: dict, block: dict, plat_name: str, *, toplevel: bool, warned: Optional[set] = None) -> dict: """Effective ``:`` block for YAML→extra/env bridging: *block* (top-level or nested) overlaid with the keys the user authored in ``platforms..extra``. Returns a NEW dict and never mutates *block* or *extra*. For every key present in both with DIFFERENT values, ONE warning logs that the authored extra took precedence (equal values are silent). *warned* dedupes across both copy sites within one ``load_yaml_layer`` pass; ``None`` gives the caller a fresh local set. """ if warned is None: warned = set() where = f"the top-level '{plat_name}:' block" if toplevel else f"platforms.{plat_name}" def overlay_onto(source: dict) -> dict: overlay = {k: extra[k] for k in source if k in extra} for key, value in overlay.items(): if source[key] != value and (plat_name, key) not in warned: warned.add((plat_name, key)) logger.warning( "%s set in both %s and platforms.%s.extra; platforms.%s.extra took precedence", key, where, plat_name, plat_name, ) return {**source, **overlay} effective = overlay_onto(block) # A root block may carry its own ``extra:`` sub-dict (``_bridged_keys`` merges it with # ``PlatformConfig.from_dict`` semantics); the authored extra outranks that too. if isinstance(block.get("extra"), dict): effective["extra"] = overlay_onto(block["extra"]) # Hooks consume direct keys before global fallbacks. Expose the resolved # authored subdict values there too, without promoting typed platform fields. effective.update({k: extra[k] for k in block["extra"] if k in extra and k not in PlatformConfig._TYPED_KEYS}) return effective def bridge_platform_shared_keys( yaml_cfg: dict, gateway_platforms: Any, gw_data: dict, platforms_data: dict, targets: list, warned: Optional[set] = None, authored: Optional[dict] = None, ) -> None: """Copy shared keys (allow_from, require_mention, …) from each platform's YAML section into ``extra``. ``enabled`` is only written from a TOP-LEVEL block: for nested-only configs ``merge_platform_sections`` already merged it with the correct precedence. An explicit top-level enable/disable sets ``_enabled_explicit`` so the env pass honors ``enabled: false`` for migrated plugin platforms instead of re-enabling them on token/SDK presence. """ if authored is None: authored = snapshot_authored_extra(platforms_data) for plat in targets: if plat == Platform.LOCAL: continue platform_cfg, cfg_toplevel = platform_section(yaml_cfg, plat.value, gateway_platforms) if not isinstance(platform_cfg, dict): continue # An authored ``platforms..extra`` key beats the same key in the (top-level or # nested) block, so the shared-key bridge works on the effective block. effective = _authored_wins( authored.get(plat.value, {}), platform_cfg, plat.value, toplevel=cfg_toplevel, warned=warned) bridged = _bridged_keys(plat, effective, gw_data, root_block=cfg_toplevel) has_channel_overrides = "channel_overrides" in platform_cfg if has_channel_overrides and isinstance(platform_cfg.get("channel_overrides"), dict): plat_data = _dict_slot(platforms_data, plat.value) _dict_slot(plat_data, "extra") plat_data["channel_overrides"] = { str(cid): ov_data for cid, ov_data in platform_cfg["channel_overrides"].items() if isinstance(ov_data, dict) } enabled_was_explicit = cfg_toplevel and "enabled" in platform_cfg if not bridged and not enabled_was_explicit and not has_channel_overrides: continue plat_data = _dict_slot(platforms_data, plat.value) extra = _dict_slot(plat_data, "extra") if enabled_was_explicit: plat_data["enabled"] = platform_cfg["enabled"] extra["_enabled_explicit"] = True extra.update(bridged) 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, ) -> 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()``.""" if registry is None: return if authored is None: authored = snapshot_authored_extra(platforms_data) for entry in registry.all_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 # ``GatewayConfig.from_dict``. if entry.apply_yaml_config_fn is None: continue platform_cfg, cfg_toplevel = platform_section(yaml_cfg, entry.name, gateway_platforms) if not isinstance(platform_cfg, dict): continue # Overlay the authored extra BEFORE the hook so the block the hook sees — and # therefore the values it bridges to env — already carry the authored nested # value for keys present in both (adapter readers are env-first). effective = _authored_wins( authored.get(entry.name, {}), platform_cfg, entry.name, toplevel=cfg_toplevel, warned=warned) # Restored managed pins must reach hooks before raw-YAML/global fallbacks # can seed a stale env value. Other authored-only keys keep existing behavior. effective.update((managed_extra or {}).get(entry.name, {})) try: seeded = entry.apply_yaml_config_fn(yaml_cfg, effective) except Exception as e: logger.debug("apply_yaml_config_fn for %s raised: %s", entry.name, e) continue if isinstance(seeded, dict) and seeded: # Safe: seeded values already equal the authored ones for keys present in both. _dict_slot(_dict_slot(platforms_data, entry.name), "extra").update(seeded) def bridge_core_env_settings(yaml_cfg: dict, platforms_data: dict) -> None: """The YAML→env bridges that stay in core (per-platform ones live in plugin hooks). Top-level ``require_mention`` → Telegram when the ``telegram:`` section has none: users write it alongside ``group_sessions_per_user`` expecting it to work, and the telegram plugin's hook only runs when a telegram block exists. Signal ``require_mention`` → ``SIGNAL_REQUIRE_MENTION`` (env wins). ``allow_all_users`` (top-level or ``gateway.allow_all_users``) → ``GATEWAY_ALLOW_ALL_USERS``: every allow-all reader (authz mixin, startup access check, own-policy adapters, plugin gates) consults that env var, so the bridge is the one seam that makes the YAML key reach all of them (#110690). Platform values are ALWAYS seeded into the owning platform's ``extra`` (the adapters read extra first); the process-env write is skipped while a multiplexed secondary profile's scope is active — the loader runs inside ``_profile_runtime_scope`` for every secondary, and a first-writer-wins write there would make the secondary's policy the DEFAULT profile's (#80099 class). A secondary profile sets ``GATEWAY_ALLOW_ALL_USERS`` in its own ``.env`` like every other scoped authorization gate. """ global _BRIDGED_ALLOW_ALL_USERS from gateway.platforms._shared import profile_scoped skip_env_bridge = profile_scoped() gateway_section = yaml_cfg.get("gateway") allow_all = yaml_cfg.get("allow_all_users") if allow_all is None and isinstance(gateway_section, dict): allow_all = gateway_section.get("allow_all_users") # Only a value this bridge wrote may be overwritten/cleared by a later load (config flipped to # false + reload); an operator's explicit env var still wins. Only a truthy grant is exported: # presence-based readers treat any non-empty value as "auth configured". if not skip_env_bridge and (_BRIDGED_ALLOW_ALL_USERS or not os.getenv("GATEWAY_ALLOW_ALL_USERS")): _BRIDGED_ALLOW_ALL_USERS = str(allow_all).lower() in {"true", "1", "yes"} if _BRIDGED_ALLOW_ALL_USERS: os.environ["GATEWAY_ALLOW_ALL_USERS"] = "true" # The key was inert before it was bridged, so a forgotten line silently flips the # posture to open — name the grant source at startup. logger.warning( "config.yaml allow_all_users: true grants every sender on every platform access " "(bridged to GATEWAY_ALLOW_ALL_USERS; an explicit env var wins)." ) else: os.environ.pop("GATEWAY_ALLOW_ALL_USERS", None) tl_require_mention = yaml_cfg.get("require_mention") if tl_require_mention is not None and "require_mention" not in (yaml_cfg.get("telegram") or {}): tg_plat = platforms_data.setdefault(Platform.TELEGRAM.value, {}) tg_plat.setdefault("extra", {}).setdefault("require_mention", tl_require_mention) # Also bridge to the TELEGRAM_REQUIRE_MENTION env var that the adapter reads at runtime. This used # to live in the telegram_cfg block in core; it stays in core because it keys off the TOP-LEVEL # require_mention (not a telegram: block), so the telegram plugin's apply_yaml_config_fn hook — # which only runs when a telegram config block exists — can't cover the no-telegram-block case # (#3979). if not skip_env_bridge and not os.getenv("TELEGRAM_REQUIRE_MENTION"): os.environ["TELEGRAM_REQUIRE_MENTION"] = str(tl_require_mention).lower() # Telegram settings → env vars / extra: migrated to the telegram plugin's apply_yaml_config_fn hook # (plugins/platforms/telegram/adapter.py). #41112 / #3823. # WhatsApp settings → env vars: migrated to the whatsapp plugin's apply_yaml_config_fn hook # (plugins/platforms/whatsapp/adapter.py). #41112 / #3823. signal_cfg = yaml_cfg.get("signal", {}) if isinstance(signal_cfg, dict) and "require_mention" in signal_cfg: sig_plat = platforms_data.setdefault(Platform.SIGNAL.value, {}) sig_plat.setdefault("extra", {}).setdefault("require_mention", signal_cfg["require_mention"]) if not skip_env_bridge and not os.getenv("SIGNAL_REQUIRE_MENTION"): os.environ["SIGNAL_REQUIRE_MENTION"] = str(signal_cfg["require_mention"]).lower() def read_yaml_layers(home: Path) -> dict: """User ``config.yaml`` with the managed overlay applied — the YAML the gateway loader sees. Raises on a malformed user file (the loader then falls back to env + gateway.json WITHOUT the managed layer). An ABSENT user file is an empty layer, not a reason to skip the administrator's values: a fleet host with no ``config.yaml`` must still honor them. Any pre-activation predicate (``gateway.relay.relay_explicitly_disabled``) reads through here so it cannot disagree with ``load_gateway_config()`` on which files count. """ config_yaml_path = home / "config.yaml" yaml_cfg: dict = {} if config_yaml_path.exists(): # The installer seeds config.yaml by copying cli-config.yaml.example (~120 KB, almost all # comments), and nothing caches this loader — so the pure-Python parser dominates the load. from utils import fast_safe_load with open(config_yaml_path, encoding="utf-8-sig") as f: yaml_cfg = fast_safe_load(f) or {} from hermes_cli.config import _expand_env_vars # ${VAR} / ${env:VAR} expansion — the same primitive the CLI loader applies, so platform # adapter settings (webhook secret, api_server key, teams credentials) arrive resolved # instead of as literal refs. User layer first: the managed overlay is expanded by # apply_managed_overlay itself and must not be re-resolved (config_effective._effective order). yaml_cfg = _expand_env_vars(yaml_cfg) # Managed scope: overlay administrator-pinned values (this loader bypasses # hermes_cli.config.load_config, so managed quick_commands / stt would otherwise be ignored). from hermes_cli import managed_scope return managed_scope.apply_managed_overlay(yaml_cfg) def load_yaml_layer(home: Path, gw_data: dict) -> None: """Overlay ``read_yaml_layers`` onto *gw_data* in place. Raises on any failure (caller warns + falls back).""" yaml_cfg = read_yaml_layers(home) if not yaml_cfg: return gateway_section = yaml_cfg.get("gateway") bridge_toplevel_keys(yaml_cfg, gateway_section, gw_data) gateway_platforms = gateway_section.get("platforms") if isinstance(gateway_section, dict) else None platforms_data = merge_platform_sections(yaml_cfg, gateway_section, gw_data) try: from hermes_cli.plugins import discover_plugins discover_plugins() # idempotent from gateway.platform_registry import platform_registry as registry except Exception as e: logger.debug("plugin discovery skipped: %s", e) registry = None targets = shared_loop_targets(registry) # "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, {})) from hermes_cli import managed_scope managed_extra = _apply_managed_extra(authored, managed_scope.apply_managed_overlay({}), platforms_data) warned: set = set() # one warning per conflicting (platform, key) across both copy sites 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) bridge_core_env_settings(yaml_cfg, platforms_data)