refactor(config): one effective-user-config loader replaces 9 hand-rolled raw→overlay→expand pipelines
Every defaults-free config reader (gateway runtime, TUI gateway, cron
scheduler + job snapshot, `hermes send` env bridge, doctor memory section,
hermes_cli/main early parse, hermes_time, hermes_logging, the gateway
fallback-chain refresh) re-implemented "read config.yaml + managed overlay +
${VAR} expansion" by hand, in three different orders, and none of them
replayed the model-key canonicalization or the last-known-good recovery that
load_config() gained. An admin-pinned `${VAR}` expanded on one surface and was
bridged literally on another; `model: {name: x}` resolved to an empty model
everywhere except the gateway.
hermes_cli/config_effective.py::load_user_config_effective is the one
primitive: user file → ${VAR} → managed overlay → _normalize_root_model_keys,
no DEFAULT_CONFIG merge, sharing read_raw_config's parse cache and serving the
last good parse (in-process, then backups/config/*.good.*) on torn YAML;
`fail_closed=True` raises for the one caller that keeps its own last-good
state (the fallback-chain refresh). gateway/run.py::_load_gateway_runtime_config
is deleted — it was _load_gateway_config plus expansion, and _load_gateway_config
now expands.
Behavior change: _load_bridge_config, send_cmd._load_hermes_env and
doctor_state._doctor_memory_config expanded BEFORE the overlay; they now match
load_config (managed `${VAR}` expands against the process env only). All nine
sites gain model-key canonicalization and last-good recovery.
send_cmd._load_hermes_env now routes its .env read through
env_loader._load_dotenv_with_fallback so the credential sanitizer runs.
This commit is contained in:
+2
-6
@@ -1497,16 +1497,12 @@ def _resolve_default_model_snapshot() -> Optional[str]:
|
||||
"""Default model resolved as the ticker's ``run_job`` does, so unpinned jobs can snapshot it and
|
||||
keep running on it after a later swap. ``None`` on missing config or failure ("no snapshot")."""
|
||||
try:
|
||||
from hermes_cli.config import _expand_env_vars, read_user_config_raw
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
|
||||
cfg_path = get_hermes_home() / "config.yaml"
|
||||
if not cfg_path.exists():
|
||||
return None
|
||||
cfg = read_user_config_raw(cfg_path)
|
||||
with contextlib.suppress(Exception):
|
||||
from hermes_cli import managed_scope
|
||||
cfg = managed_scope.apply_managed_overlay(cfg)
|
||||
cfg = _expand_env_vars(cfg)
|
||||
cfg = load_user_config_effective(cfg_path)
|
||||
cron_cfg = cfg.get("cron") or {}
|
||||
if isinstance(cron_cfg, dict):
|
||||
cron_model = cron_cfg.get("model")
|
||||
|
||||
+3
-8
@@ -39,7 +39,7 @@ from hermes_constants import get_hermes_home
|
||||
from cron.env_settings import cron_env_setting
|
||||
from hermes_cli._subprocess_compat import windows_hide_flags
|
||||
from hermes_cli.config import (
|
||||
_expand_env_vars, load_config, load_config_readonly, resolve_cron_model_drift_defaults)
|
||||
load_config, load_config_readonly, resolve_cron_model_drift_defaults)
|
||||
from hermes_cli.fallback_config import get_fallback_chain
|
||||
from hermes_time import now as _hermes_now
|
||||
from agent.interrupt_compat import request_hard_interrupt
|
||||
@@ -1378,15 +1378,10 @@ def _load_cron_job_config(job: dict, job_id: str, job_name: str) -> _CronJobConf
|
||||
_cfg: dict = {}
|
||||
_model_cfg: Any = {}
|
||||
try:
|
||||
from hermes_cli.config import read_user_config_raw
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
_cfg_path = str(_get_hermes_home() / "config.yaml")
|
||||
if os.path.exists(_cfg_path):
|
||||
_cfg = read_user_config_raw(Path(_cfg_path))
|
||||
# Honor administrator-pinned managed scope (fail-open; no-op without managed scope).
|
||||
with contextlib.suppress(Exception):
|
||||
from hermes_cli import managed_scope
|
||||
_cfg = managed_scope.apply_managed_overlay(_cfg)
|
||||
_cfg = _expand_env_vars(_cfg)
|
||||
_cfg = load_user_config_effective(Path(_cfg_path))
|
||||
# Coerce null to {} so a falsy default never clobbers a resolved env value.
|
||||
_model_cfg = _cfg.get("model") or {}
|
||||
_cron_cfg_for_model = _cfg.get("cron") or {}
|
||||
|
||||
+11
-66
@@ -2021,19 +2021,10 @@ def _bridge_config_to_env(_cfg: dict) -> None:
|
||||
|
||||
|
||||
def _load_bridge_config(config_path: Path) -> dict:
|
||||
"""Raw config read for the presence-sensitive env bridge, with the managed overlay applied. Raw (not
|
||||
defaults-merged) so only keys the user wrote are bridged, else all of DEFAULT_CONFIG would be
|
||||
exported; the overlay applies BEFORE bridging so pinned values win in env too."""
|
||||
from hermes_cli.config import _expand_env_vars, read_user_config_raw
|
||||
cfg = _expand_env_vars(read_user_config_raw(config_path))
|
||||
if not isinstance(cfg, dict):
|
||||
cfg = {}
|
||||
try:
|
||||
from hermes_cli import managed_scope
|
||||
cfg = managed_scope.apply_managed_overlay(cfg)
|
||||
except Exception:
|
||||
pass
|
||||
return cfg
|
||||
"""Effective USER config (no defaults) for the presence-sensitive env bridge: only keys the user
|
||||
or the managed layer wrote get bridged, else all of DEFAULT_CONFIG would be exported."""
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
return load_user_config_effective(config_path)
|
||||
|
||||
|
||||
_config_path = _hermes_home / 'config.yaml'
|
||||
@@ -2385,8 +2376,7 @@ def _try_resolve_fallback_provider() -> dict | None:
|
||||
"""Attempt to resolve credentials from the fallback_model/fallback_providers config."""
|
||||
from hermes_cli.runtime_provider import resolve_runtime_provider
|
||||
try:
|
||||
# Canonical loader so managed overlay / ${VAR} expansion reach the fallback chain.
|
||||
cfg = _load_gateway_runtime_config()
|
||||
cfg = _load_gateway_config()
|
||||
fb_list = get_fallback_chain(cfg)
|
||||
if not fb_list:
|
||||
return None
|
||||
@@ -2791,51 +2781,18 @@ def _gateway_config_home() -> Path:
|
||||
|
||||
|
||||
def _load_gateway_config(config_path: "Path | None" = None) -> dict:
|
||||
"""Load and parse a gateway config.yaml, returning {} on any error (fail-open).
|
||||
Defaults to the active gateway home (``_hermes_home`` monkeypatches apply); multiplexers pass a path.
|
||||
"""The effective user config.yaml (managed overlay, ``${VAR}`` expansion, model-key canon; no
|
||||
DEFAULT_CONFIG merge) — ``{}`` on any error (fail-open). Defaults to the active gateway home
|
||||
(``_hermes_home`` monkeypatches apply); multiplexers pass a path.
|
||||
"""
|
||||
if config_path is None:
|
||||
config_path = _gateway_config_home() / 'config.yaml'
|
||||
raw: dict = {}
|
||||
used_canonical = False
|
||||
try:
|
||||
from hermes_cli.config import get_config_path, read_raw_config
|
||||
# Fast path via shared cache when the path is canonical; else direct read (test monkeypatches).
|
||||
if config_path == get_config_path():
|
||||
raw = read_raw_config()
|
||||
used_canonical = True
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
return load_user_config_effective(config_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not used_canonical:
|
||||
try:
|
||||
if config_path.exists():
|
||||
import yaml
|
||||
with open(config_path, 'r', encoding='utf-8') as f:
|
||||
raw = yaml.safe_load(f) or {}
|
||||
except Exception:
|
||||
logger.debug("Could not load gateway config from %s", config_path)
|
||||
raw = {}
|
||||
|
||||
# Neither read_raw_config() nor yaml.safe_load carries the managed merge; overlay on both paths.
|
||||
try:
|
||||
from hermes_cli import managed_scope
|
||||
raw = managed_scope.apply_managed_overlay(raw if isinstance(raw, dict) else {})
|
||||
except Exception:
|
||||
pass
|
||||
if not isinstance(raw, dict):
|
||||
logger.debug("Could not load gateway config from %s", config_path, exc_info=True)
|
||||
return {}
|
||||
# Canonicalize model-id aliases (model.name/model.model → model.default) and migrate stale root
|
||||
# provider/base_url: the gateway bypasses load_config(), else ``model: {name: <id>}`` is empty.
|
||||
try:
|
||||
# The gateway bypasses load_config() (it reads raw YAML for speed), so the normalization that
|
||||
# load_config() applies must be replayed here or the gateway would resolve an empty model for
|
||||
# ``model: {name: <id>}`` configs while the CLI resolves it correctly. See issue #34500. Fail-open.
|
||||
from hermes_cli.config import _normalize_root_model_keys
|
||||
raw = _normalize_root_model_keys(raw)
|
||||
except Exception:
|
||||
pass
|
||||
return raw
|
||||
|
||||
|
||||
def _checkpoint_agent_kwargs(config: dict | None) -> dict:
|
||||
@@ -2855,18 +2812,6 @@ def _checkpoint_agent_kwargs(config: dict | None) -> dict:
|
||||
"checkpoint_max_file_size_mb": cp_cfg.get("max_file_size_mb", defaults["max_file_size_mb"])}
|
||||
|
||||
|
||||
def _load_gateway_runtime_config() -> dict:
|
||||
"""Load gateway config for runtime reads, expanding supported ``${VAR}`` refs.
|
||||
Expansion failures are deliberately NOT swallowed: an unexpanded dict would mask the bug fixed here.
|
||||
"""
|
||||
cfg = _load_gateway_config()
|
||||
if not isinstance(cfg, dict) or not cfg:
|
||||
return {}
|
||||
from hermes_cli.config import _expand_env_vars
|
||||
expanded = _expand_env_vars(cfg)
|
||||
return expanded if isinstance(expanded, dict) else {}
|
||||
|
||||
|
||||
def _resolve_gateway_model(config: dict | None = None) -> str:
|
||||
"""Read model from config.yaml (single source of truth), else temporary AIAgents (e.g. /compress)
|
||||
use the hardcoded default, which fails under openai-codex."""
|
||||
|
||||
@@ -887,7 +887,7 @@ class GatewayAdapterLifecycleMixin:
|
||||
default profile owns the single shared listener and a secondary's port-binders are built in
|
||||
shared-listener mode (``/p/<profile>/...``) by ``_start_one_profile_adapters``."""
|
||||
from gateway.run import (
|
||||
MultiplexConfigError, _load_gateway_runtime_config,
|
||||
MultiplexConfigError, _load_gateway_config,
|
||||
_own_policy_open_startup_violation, _profile_runtime_scope,
|
||||
)
|
||||
from gateway.config import load_gateway_config
|
||||
@@ -895,7 +895,7 @@ class GatewayAdapterLifecycleMixin:
|
||||
# Hydrate external secret sources off-loop ONCE: sync hydration would stall every heartbeat.
|
||||
await asyncio.to_thread(hydrate_profile_secret_sources, profile_home)
|
||||
with _profile_runtime_scope(profile_home, hydrate_secrets=False):
|
||||
profile_runtime_cfg = _load_gateway_runtime_config()
|
||||
profile_runtime_cfg = _load_gateway_config()
|
||||
from hermes_cli.plugins import discover_plugins
|
||||
discover_plugins()
|
||||
# This profile's `hooks:` block: start() registered before any profile scope existed.
|
||||
|
||||
@@ -45,8 +45,8 @@ class GatewayConfigLoadersMixin:
|
||||
@staticmethod
|
||||
def _cfg_str(section: str, key: str) -> str:
|
||||
"""``<section>.<key>`` from the gateway runtime config as a stripped string ("" when unset)."""
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
return str(cfg_get(_load_gateway_runtime_config(), section, key, default="") or "").strip()
|
||||
from gateway.run import _load_gateway_config
|
||||
return str(cfg_get(_load_gateway_config(), section, key, default="") or "").strip()
|
||||
|
||||
@classmethod
|
||||
def _env_or_cfg_str(cls, env_var: str, section: str, key: str) -> str:
|
||||
@@ -60,10 +60,10 @@ class GatewayConfigLoadersMixin:
|
||||
HERMES_PREFILL_MESSAGES_FILE env wins, then top-level prefill_messages_file in config.yaml,
|
||||
then legacy agent.prefill_messages_file. Relative paths resolve from ~/.hermes/.
|
||||
"""
|
||||
from gateway.run import _gateway_config_home, _load_gateway_runtime_config
|
||||
from gateway.run import _gateway_config_home, _load_gateway_config
|
||||
file_path = os.getenv("HERMES_PREFILL_MESSAGES_FILE", "")
|
||||
if not file_path:
|
||||
cfg = _load_gateway_runtime_config()
|
||||
cfg = _load_gateway_config()
|
||||
file_path = str(
|
||||
cfg.get("prefill_messages_file", "") or cfg_get(cfg, "agent", "prefill_messages_file", default="") or ""
|
||||
)
|
||||
@@ -89,11 +89,11 @@ class GatewayConfigLoadersMixin:
|
||||
@staticmethod
|
||||
def _load_ephemeral_system_prompt() -> str:
|
||||
"""HERMES_EPHEMERAL_SYSTEM_PROMPT env first, then ``display.personality`` / ``agent.system_prompt``."""
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
from gateway.run import _load_gateway_config
|
||||
prompt = os.getenv("HERMES_EPHEMERAL_SYSTEM_PROMPT", "")
|
||||
if prompt:
|
||||
return prompt
|
||||
return resolve_ephemeral_system_prompt_from_config(_load_gateway_runtime_config())
|
||||
return resolve_ephemeral_system_prompt_from_config(_load_gateway_config())
|
||||
|
||||
def _channel_override(self, platform: Platform, chat_id: str, thread_id, parent_id):
|
||||
"""``channel_overrides`` entry for this channel/thread, or None (also when no config is bound)."""
|
||||
@@ -151,9 +151,9 @@ class GatewayConfigLoadersMixin:
|
||||
|
||||
Closes #21256.
|
||||
"""
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
from gateway.run import _load_gateway_config
|
||||
from hermes_constants import resolve_reasoning_config
|
||||
return resolve_reasoning_config(_load_gateway_runtime_config(), model)
|
||||
return resolve_reasoning_config(_load_gateway_config(), model)
|
||||
|
||||
@staticmethod
|
||||
def _parse_reasoning_command_args(raw_args: str) -> tuple[str, bool]:
|
||||
@@ -234,8 +234,8 @@ class GatewayConfigLoadersMixin:
|
||||
@staticmethod
|
||||
def _load_show_reasoning() -> bool:
|
||||
"""``display.show_reasoning`` toggle."""
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
return is_truthy_value(cfg_get(_load_gateway_runtime_config(), "display", "show_reasoning"), default=False)
|
||||
from gateway.run import _load_gateway_config
|
||||
return is_truthy_value(cfg_get(_load_gateway_config(), "display", "show_reasoning"), default=False)
|
||||
|
||||
@classmethod
|
||||
def _load_busy_input_mode(cls) -> str:
|
||||
@@ -330,12 +330,12 @@ class GatewayConfigLoadersMixin:
|
||||
"""Env var (non-empty) else ``agent.<cfg_key>``; warn once when a supplied value fails to parse.
|
||||
|
||||
``0`` is a valid value; the parser falls back to ``default`` on garbage."""
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
from gateway.run import _load_gateway_config
|
||||
env_raw = os.getenv(env_var)
|
||||
if env_raw is not None and str(env_raw).strip() != "":
|
||||
raw: object = env_raw
|
||||
else:
|
||||
raw = cfg_get(_load_gateway_runtime_config(), "agent", cfg_key, default=None)
|
||||
raw = cfg_get(_load_gateway_config(), "agent", cfg_key, default=None)
|
||||
value = parse(raw)
|
||||
if raw is not None and str(raw).strip() != "":
|
||||
cls._warn_unparsable_timeout(cfg_key, raw, default)
|
||||
@@ -363,8 +363,8 @@ class GatewayConfigLoadersMixin:
|
||||
@classmethod
|
||||
def _load_signal_interrupt_grace_timeout(cls) -> float:
|
||||
"""``gateway.signal_interrupt_grace_timeout``: unexpected-signal post-interrupt grace in seconds."""
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
raw = cfg_get(_load_gateway_runtime_config(), "gateway", "signal_interrupt_grace_timeout", default=None)
|
||||
from gateway.run import _load_gateway_config
|
||||
raw = cfg_get(_load_gateway_config(), "gateway", "signal_interrupt_grace_timeout", default=None)
|
||||
value = parse_signal_interrupt_grace_timeout(raw)
|
||||
if raw is not None and raw != "":
|
||||
cls._warn_unparsable_timeout(
|
||||
@@ -385,11 +385,11 @@ class GatewayConfigLoadersMixin:
|
||||
the AMBIENT profile — callers deciding for another profile's event enter its scope first
|
||||
(``_completion_event_scope``). The env override reads through the secret scope so a served
|
||||
secondary sees its own ``.env`` value, not the launch profile's ``os.environ``."""
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
from gateway.run import _load_gateway_config
|
||||
from gateway.authz_mixin import _platform_gate_env
|
||||
mode = _platform_gate_env("HERMES_BACKGROUND_NOTIFICATIONS")
|
||||
if not mode:
|
||||
raw = cfg_get(_load_gateway_runtime_config(), "display", "background_process_notifications")
|
||||
raw = cfg_get(_load_gateway_config(), "display", "background_process_notifications")
|
||||
if raw is False:
|
||||
mode = "off"
|
||||
elif raw not in {None, ""}:
|
||||
@@ -403,19 +403,19 @@ class GatewayConfigLoadersMixin:
|
||||
@staticmethod
|
||||
def _load_provider_routing() -> dict:
|
||||
"""OpenRouter provider routing preferences (canonical fail-open loader: managed overlay + ${VAR})."""
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
from gateway.run import _load_gateway_config
|
||||
try:
|
||||
return _load_gateway_runtime_config().get("provider_routing", {}) or {}
|
||||
return _load_gateway_config().get("provider_routing", {}) or {}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
@staticmethod
|
||||
def _load_fallback_model() -> list | None:
|
||||
"""Fallback chain: ``fallback_providers`` (kept first) merged with legacy ``fallback_model``."""
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
from gateway.run import _load_gateway_config
|
||||
try:
|
||||
# Canonical gateway loader (fail-open): managed overlay + ${VAR} expansion apply here too.
|
||||
return get_fallback_chain(_load_gateway_runtime_config()) or None
|
||||
return get_fallback_chain(_load_gateway_config()) or None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
@@ -443,23 +443,14 @@ class GatewayConfigLoadersMixin:
|
||||
by_home = self._fallback_model_by_home = {}
|
||||
home_key = hermes_home_key(home)
|
||||
try:
|
||||
from hermes_cli.config import read_user_config_raw
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
cfg_path = home / "config.yaml"
|
||||
if not cfg_path.exists():
|
||||
by_home[home_key] = self._fallback_model = None
|
||||
return self._fallback_model
|
||||
# Raw primitive (raises on parse failure) is required here: the canonical fail-open
|
||||
# loader would return {} on a torn mid-edit write and WIPE the last known-good chain.
|
||||
# The overlay/expansion below fixes the managed-scope/${VAR} drift without losing that.
|
||||
cfg = read_user_config_raw(cfg_path)
|
||||
with suppress(Exception):
|
||||
from hermes_cli import managed_scope
|
||||
cfg = managed_scope.apply_managed_overlay(cfg)
|
||||
with suppress(Exception):
|
||||
from hermes_cli.config import _expand_env_vars
|
||||
expanded = _expand_env_vars(cfg)
|
||||
if isinstance(expanded, dict):
|
||||
cfg = expanded
|
||||
# fail_closed: a torn mid-edit write must raise so the per-home last known-good chain
|
||||
# below survives, instead of being WIPED by an empty fail-open result.
|
||||
cfg = load_user_config_effective(cfg_path, fail_closed=True)
|
||||
except Exception:
|
||||
logger.debug("fallback_providers refresh: config.yaml read failed; keeping last known-good chain", exc_info=True)
|
||||
self._fallback_model = by_home.get(home_key, self._fallback_model)
|
||||
|
||||
@@ -950,8 +950,8 @@ class GatewaySlashCommandsMixin(
|
||||
return EphemeralReply("Busy input mode could not be saved to config. Mode unchanged.")
|
||||
profile_name = self._busy_profile_name_for_source(event.source)
|
||||
if profile_name:
|
||||
from gateway.run import _load_gateway_runtime_config
|
||||
self._snapshot_profile_busy_modes(profile_name, _load_gateway_runtime_config())
|
||||
from gateway.run import _load_gateway_config
|
||||
self._snapshot_profile_busy_modes(profile_name, _load_gateway_config())
|
||||
else:
|
||||
self._busy_input_mode = arg
|
||||
# busy_input_mode is also the source of truth for the text mode — re-derive it so the
|
||||
|
||||
@@ -67,9 +67,13 @@ Do not add a surface-specific goal parser. ACP has no goal command or goal loop
|
||||
(`gateway_timeout`; `terminal.cwd` → `TERMINAL_CWD`). `MESSAGING_CWD` is removed and `TERMINAL_CWD`
|
||||
in `.env` is deprecated — the loader warns; canonical is `terminal.cwd`.
|
||||
- **Three loaders — know which you're in:** `load_cli_config()` (CLI, `cli.py`); `load_config()`
|
||||
(`hermes tools/setup`, most subcommands, `hermes_cli/config.py`, merges `DEFAULT_CONFIG`); raw
|
||||
YAML (gateway runtime, `gateway/run.py` + `gateway/config.py`). If the CLI sees a key and the
|
||||
gateway doesn't (or vice versa), you're on the wrong loader — check `DEFAULT_CONFIG` coverage.
|
||||
(`hermes tools/setup`, most subcommands, `hermes_cli/config.py`, merges `DEFAULT_CONFIG`);
|
||||
`hermes_cli/config_effective.py::load_user_config_effective()` (gateway runtime via
|
||||
`gateway/run.py::_load_gateway_config`, TUI gateway `_load_cfg`, cron, `hermes send`, doctor,
|
||||
`hermes_time`/`hermes_logging`: user file + managed overlay + `${VAR}` expansion + model-key
|
||||
canon, NO defaults — for presence-sensitive readers). If the CLI sees a key and the gateway
|
||||
doesn't (or vice versa), you're on the wrong loader — check `DEFAULT_CONFIG` coverage. Never
|
||||
hand-roll raw-read → overlay → expand; `read_user_config_raw` is for write-back round-trips only.
|
||||
- **Working directory:** CLI uses `os.getcwd()`; messaging uses `terminal.cwd`, bridged to
|
||||
`TERMINAL_CWD` for child tools.
|
||||
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
"""The effective USER config: config.yaml + managed overlay + ``${VAR}`` expansion, no defaults.
|
||||
|
||||
``load_config()`` merges ``DEFAULT_CONFIG`` first, which is wrong for readers that treat a
|
||||
missing key as "unset" (the gateway's presence-sensitive env bridge, ``cfg == {}`` sentinels,
|
||||
cron model pinning) — so nine surfaces used to hand-roll raw-read → overlay → expand in
|
||||
differing orders and none of them replayed the model-key canonicalization or the last-known-good
|
||||
recovery ``load_config()`` gained. This module is that one primitive.
|
||||
|
||||
Order matches ``_load_config_impl``: the user layer is expanded BEFORE the managed overlay so a
|
||||
managed ``${VAR}`` resolves against the process environment only (``apply_managed_overlay``
|
||||
expands it) and can never be re-resolved through a profile's secret scope
|
||||
(docs/design/managed-scope.md §4.1). ``read_user_config_raw`` stays the write-back primitive.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional, Tuple
|
||||
|
||||
from hermes_cli import config as _config
|
||||
from hermes_cli import managed_scope
|
||||
from utils import fast_safe_load
|
||||
|
||||
# path -> raw user mapping from the last successful parse in this process; served (through the
|
||||
# normal pipeline) when the file is later found mid-edit as broken YAML.
|
||||
_LAST_GOOD_USER_RAW: Dict[str, Dict[str, Any]] = {}
|
||||
# path -> (user_mtime_ns, user_size, managed_mtime_ns, managed_size, effective, env_snapshot).
|
||||
_EFFECTIVE_CACHE: Dict[str, Tuple[int, int, int, int, Dict[str, Any], Dict[str, Optional[str]]]] = {}
|
||||
|
||||
|
||||
def _effective(raw: Dict[str, Any]) -> Dict[str, Any]:
|
||||
expanded = _config._expand_env_vars(raw)
|
||||
merged = managed_scope.apply_managed_overlay(expanded if isinstance(expanded, dict) else {})
|
||||
return _config._normalize_root_model_keys(merged if isinstance(merged, dict) else {})
|
||||
|
||||
|
||||
def _recover_user_raw(config_path: Path, path_key: str, exc: Exception) -> Dict[str, Any]:
|
||||
"""Last-known-good raw user mapping after a parse failure: this process's last good parse,
|
||||
else the newest ``good`` copy in backups/config/, else ``{}`` (warned as defaults)."""
|
||||
raw = _LAST_GOOD_USER_RAW.get(path_key)
|
||||
fallback = "last-known-good"
|
||||
if raw is None:
|
||||
from hermes_cli.config_backups import load_newest_good_backup
|
||||
raw = load_newest_good_backup(config_path)
|
||||
fallback = "last-known-good-backup"
|
||||
_config._warn_config_parse_failure(config_path, exc, fallback=fallback if raw is not None else "defaults")
|
||||
return copy.deepcopy(raw) if raw is not None else {}
|
||||
|
||||
|
||||
def load_user_config_effective(config_path: Optional[Path] = None, *, fail_closed: bool = False) -> Dict[str, Any]:
|
||||
"""User ``config.yaml`` → ``${VAR}`` expansion → managed overlay → model-key canonicalization.
|
||||
NO ``DEFAULT_CONFIG`` merge: a key absent from the file (and from the managed layer) is absent
|
||||
here, so ``{}`` sentinels and presence-sensitive bridges keep working. An absent file is an
|
||||
empty user layer (the managed layer still applies). Returns a fresh deepcopy.
|
||||
|
||||
Broken YAML: ``fail_closed=True`` raises the parse error (for callers that keep their own
|
||||
last-good state); otherwise the last successfully parsed user file — in-process first, then
|
||||
the newest ``backups/config/*.good.*`` copy — is served through the same pipeline, so a
|
||||
mid-edit torn write never silently drops user overrides (same contract as ``load_config``).
|
||||
Cached on the user + managed file signatures and the values of every referenced env var."""
|
||||
if config_path is None:
|
||||
config_path = _config.get_config_path()
|
||||
path_key = str(config_path)
|
||||
with _config._CONFIG_LOCK:
|
||||
user_sig, cache_sig = _config._load_config_cache_sig(config_path)
|
||||
cached = _EFFECTIVE_CACHE.get(path_key)
|
||||
if cached is not None and cache_sig is not None and cached[:4] == cache_sig:
|
||||
if all(_config._env_ref_lookup(k) == v for k, v in cached[5].items()):
|
||||
return copy.deepcopy(cached[4])
|
||||
|
||||
raw: Dict[str, Any] = {}
|
||||
recovered = False
|
||||
raw_hit = _config._RAW_CONFIG_CACHE.get(path_key)
|
||||
if user_sig is not None and raw_hit is not None and raw_hit[:2] == user_sig:
|
||||
raw = copy.deepcopy(raw_hit[2]) # one parse per process, shared with read_raw_config()
|
||||
_LAST_GOOD_USER_RAW.setdefault(path_key, copy.deepcopy(raw))
|
||||
elif user_sig is not None:
|
||||
try:
|
||||
with open(config_path, encoding="utf-8") as f:
|
||||
loaded = fast_safe_load(f)
|
||||
except Exception as exc:
|
||||
if fail_closed:
|
||||
raise
|
||||
raw, recovered = _recover_user_raw(config_path, path_key, exc), True
|
||||
else:
|
||||
raw = loaded if isinstance(loaded, dict) else {}
|
||||
_config._RAW_CONFIG_CACHE[path_key] = (*user_sig, copy.deepcopy(raw))
|
||||
_LAST_GOOD_USER_RAW[path_key] = copy.deepcopy(raw)
|
||||
# Same copy load_config keeps: a fresh process recovers from it (see _recover_user_raw).
|
||||
from hermes_cli.config_backups import backup_config
|
||||
backup_config(config_path, "good")
|
||||
|
||||
env_snapshot = _config._env_ref_snapshot(raw)
|
||||
managed = managed_scope.load_managed_config()
|
||||
if managed:
|
||||
_config._env_ref_snapshot(managed, env_snapshot)
|
||||
effective = _effective(raw)
|
||||
# A recovered result is never cached under the corrupt file's signature: a later
|
||||
# ``fail_closed`` caller must still see the parse error, not a cache hit.
|
||||
if cache_sig is not None and not recovered:
|
||||
_EFFECTIVE_CACHE[path_key] = (*cache_sig, copy.deepcopy(effective), env_snapshot)
|
||||
else:
|
||||
_EFFECTIVE_CACHE.pop(path_key, None)
|
||||
return effective
|
||||
@@ -27,15 +27,11 @@ def _doctor_memory_config(hermes_home: Path | None = None) -> dict:
|
||||
"""Return the effective memory section used by doctor diagnostics."""
|
||||
from hermes_cli.doctor import HERMES_HOME
|
||||
try:
|
||||
from hermes_cli.config import _expand_env_vars, read_user_config_raw
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
config_path = (hermes_home if hermes_home is not None else HERMES_HOME) / "config.yaml"
|
||||
if not config_path.exists():
|
||||
return {}
|
||||
config = _expand_env_vars(read_user_config_raw(config_path))
|
||||
with warn_on_error(""):
|
||||
from hermes_cli import managed_scope
|
||||
config = managed_scope.apply_managed_overlay(config)
|
||||
section = config.get("memory") if isinstance(config, dict) else None
|
||||
section = load_user_config_effective(config_path).get("memory")
|
||||
return section if isinstance(section, dict) else {}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
+5
-11
@@ -604,20 +604,14 @@ load_hermes_dotenv(
|
||||
# is read from the same parse to avoid a second full load_config() (~17ms).
|
||||
_FORCE_IPV4_EARLY = False
|
||||
try:
|
||||
# read_raw_config()'s (mtime, size)-keyed cache means this SAME parse serves
|
||||
# hermes_logging and later raw reads: 3-4 config.yaml parses become one.
|
||||
from hermes_cli.config import read_raw_config as _read_raw_early
|
||||
# The effective-config cache (shared raw parse with read_raw_config()) means this SAME parse
|
||||
# serves hermes_logging, hermes_time and later raw reads: 3-4 config.yaml parses become one.
|
||||
# Managed overlay included: administrator-pinned redact_secrets / force_ipv4 win here too.
|
||||
from hermes_cli.config_effective import load_user_config_effective as _load_effective_early
|
||||
|
||||
_cfg_path = get_hermes_home() / "config.yaml"
|
||||
if _cfg_path.exists():
|
||||
_early_cfg_raw = _read_raw_early() or {}
|
||||
# Managed scope overlay: administrator-pinned redact_secrets /
|
||||
# force_ipv4 must win here too (load_config isn't usable yet). Fail-open.
|
||||
try:
|
||||
from hermes_cli import managed_scope
|
||||
_early_cfg_raw = managed_scope.apply_managed_overlay(_early_cfg_raw)
|
||||
except Exception:
|
||||
pass
|
||||
_early_cfg_raw = _load_effective_early(_cfg_path)
|
||||
if "HERMES_REDACT_SECRETS" not in os.environ:
|
||||
_early_sec_cfg = _early_cfg_raw.get("security", {})
|
||||
if isinstance(_early_sec_cfg, dict):
|
||||
|
||||
+9
-36
@@ -134,58 +134,31 @@ def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int:
|
||||
def _load_hermes_env() -> None:
|
||||
"""Populate ``os.environ`` from ``~/.hermes/.env`` AND bridge top-level ``config.yaml`` keys into
|
||||
the environment so the gateway config loader sees platform credentials and home channels."""
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
except Exception:
|
||||
load_dotenv = None # type: ignore[assignment]
|
||||
import os
|
||||
try:
|
||||
from hermes_cli.config import get_hermes_home
|
||||
home = get_hermes_home()
|
||||
except Exception:
|
||||
return
|
||||
env_path = home / ".env"
|
||||
if load_dotenv and env_path.exists():
|
||||
if env_path.exists():
|
||||
try:
|
||||
# utf-8-sig strips a leading BOM (PowerShell 5.1 / Notepad); plain "utf-8" would keep
|
||||
# U+FEFF on the first key name and silently drop it from os.environ.
|
||||
load_dotenv(str(env_path), override=True, encoding="utf-8-sig")
|
||||
except UnicodeDecodeError:
|
||||
try: # utf-8-sig can't strip a BOM once we fall back to latin-1.
|
||||
import codecs
|
||||
import io
|
||||
raw = env_path.read_bytes().removeprefix(codecs.BOM_UTF8)
|
||||
load_dotenv(stream=io.StringIO(raw.decode("latin-1")), override=True)
|
||||
except Exception:
|
||||
pass
|
||||
from hermes_cli.env_loader import _load_dotenv_with_fallback
|
||||
_load_dotenv_with_fallback(env_path, override=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Bridge top-level config.yaml scalars into the environment (never overriding existing values).
|
||||
import os
|
||||
# Bridge top-level scalars the user (or the managed layer) actually wrote — never DEFAULT_CONFIG —
|
||||
# into the environment, without overriding existing values.
|
||||
config_path = home / "config.yaml"
|
||||
if not config_path.exists():
|
||||
return
|
||||
try:
|
||||
# Raw read is deliberate — only keys the user actually wrote get bridged.
|
||||
from hermes_cli.config import read_user_config_raw
|
||||
raw = read_user_config_raw(config_path)
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
cfg = load_user_config_effective(config_path)
|
||||
except Exception:
|
||||
return
|
||||
try:
|
||||
from hermes_cli.config import _expand_env_vars
|
||||
raw = _expand_env_vars(raw)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Managed scope: administrator-pinned values win here too (fail-open via the helper).
|
||||
try:
|
||||
from hermes_cli import managed_scope
|
||||
raw = managed_scope.apply_managed_overlay(raw if isinstance(raw, dict) else {})
|
||||
except Exception:
|
||||
pass
|
||||
if not isinstance(raw, dict):
|
||||
return
|
||||
for key, val in raw.items():
|
||||
for key, val in cfg.items():
|
||||
if isinstance(val, (str, int, float, bool)) and key not in os.environ:
|
||||
os.environ[key] = str(val)
|
||||
|
||||
|
||||
+5
-11
@@ -604,12 +604,12 @@ def _add_rotating_handler(
|
||||
def _read_logging_config():
|
||||
"""Best-effort read of ``logging.*`` from config.yaml."""
|
||||
try:
|
||||
# Prefer the shared (mtime, size)-keyed raw-config cache so this reuses
|
||||
# hermes_cli.main's early parse (one config.yaml parse per process);
|
||||
# fall back to a direct parse for bare hermes_logging consumers.
|
||||
# Prefer the shared effective-config cache (managed overlay included, so an administrator
|
||||
# can pin logging.*) so this reuses hermes_cli.main's early parse (one config.yaml parse
|
||||
# per process); fall back to a direct parse for bare hermes_logging consumers.
|
||||
try:
|
||||
from hermes_cli.config import read_raw_config as _rrc
|
||||
cfg = _rrc() or {}
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
cfg = load_user_config_effective(get_config_path())
|
||||
except Exception:
|
||||
from utils import fast_safe_load
|
||||
config_path = get_config_path()
|
||||
@@ -619,12 +619,6 @@ def _read_logging_config():
|
||||
cfg = fast_safe_load(f) or {}
|
||||
if not cfg:
|
||||
return (None, None, None)
|
||||
# Managed scope: an administrator can pin logging.* too (fail-open overlay).
|
||||
try:
|
||||
from hermes_cli import managed_scope
|
||||
cfg = managed_scope.apply_managed_overlay(cfg)
|
||||
except Exception:
|
||||
pass
|
||||
log_cfg = cfg.get("logging", {})
|
||||
if isinstance(log_cfg, dict):
|
||||
return (log_cfg.get("level"), log_cfg.get("max_size_mb"), log_cfg.get("backup_count"))
|
||||
|
||||
+6
-10
@@ -48,22 +48,18 @@ def _resolve_timezone_name() -> str:
|
||||
if tz_env:
|
||||
return tz_env
|
||||
try:
|
||||
# Prefer the shared cached raw-config reader (mtime-keyed + libyaml): a direct safe_load of
|
||||
# a large config.yaml costs ~100 ms and this ran inside the FIRST system prompt build.
|
||||
# Prefer the shared cached effective-config loader (mtime-keyed + libyaml, managed overlay
|
||||
# included so an administrator can pin ``timezone``): a direct safe_load of a large
|
||||
# config.yaml costs ~100 ms and this ran inside the FIRST system prompt build. The bare
|
||||
# parse is the stdlib-safe fallback for bootstrap consumers without hermes_cli importable.
|
||||
try:
|
||||
from hermes_cli.config import read_raw_config
|
||||
cfg = read_raw_config() or {}
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
cfg = load_user_config_effective(get_config_path())
|
||||
except Exception:
|
||||
import yaml
|
||||
config_path = get_config_path()
|
||||
cfg = (yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}) if config_path.exists() else {}
|
||||
if cfg:
|
||||
# Managed scope: an administrator can pin ``timezone`` too (fail-open overlay).
|
||||
try:
|
||||
from hermes_cli import managed_scope
|
||||
cfg = managed_scope.apply_managed_overlay(cfg)
|
||||
except Exception:
|
||||
pass
|
||||
tz_cfg = cfg.get("timezone", "")
|
||||
if isinstance(tz_cfg, str) and tz_cfg.strip():
|
||||
return tz_cfg.strip()
|
||||
|
||||
@@ -104,7 +104,7 @@ def test_fast_auto_and_cold_parse_and_slash_command(monkeypatch):
|
||||
for raw, expected in (("auto", "auto"), ("COLD", "cold"), ("fast", "priority"), ("", None), ("bogus", None)):
|
||||
assert cli_mod._parse_service_tier_config(raw) == expected
|
||||
monkeypatch.setattr(
|
||||
"gateway.run._load_gateway_runtime_config", lambda: {"agent": {"service_tier": raw}}
|
||||
"gateway.run._load_gateway_config", lambda: {"agent": {"service_tier": raw}}
|
||||
)
|
||||
assert GatewayRunner._load_service_tier() == expected
|
||||
assert DEFAULT_CONFIG["agent"]["service_tier"] == ""
|
||||
|
||||
@@ -1087,7 +1087,8 @@ class TestRunJobConfigLogging:
|
||||
"""Verify that config.yaml parse failures are logged, not silently swallowed."""
|
||||
|
||||
def test_bad_config_yaml_is_logged(self, caplog, tmp_path):
|
||||
"""When config.yaml is malformed, a warning should be logged."""
|
||||
"""When config.yaml is malformed, the shared config loader warns loudly (and serves the
|
||||
last known-good copy instead of silently dropping the user's overrides)."""
|
||||
bad_yaml = tmp_path / "config.yaml"
|
||||
bad_yaml.write_text("invalid: yaml: [[[bad")
|
||||
|
||||
@@ -1116,11 +1117,11 @@ class TestRunJobConfigLogging:
|
||||
mock_agent.run_conversation.return_value = {"final_response": "ok"}
|
||||
mock_agent_cls.return_value = mock_agent
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="cron.scheduler"):
|
||||
with caplog.at_level(logging.WARNING):
|
||||
run_job(job)
|
||||
|
||||
assert any("failed to load config.yaml" in r.message for r in caplog.records), \
|
||||
f"Expected 'failed to load config.yaml' warning in logs, got: {[r.message for r in caplog.records]}"
|
||||
assert any("Failed to parse" in r.message and "config.yaml" in r.message for r in caplog.records), \
|
||||
f"Expected a config.yaml parse warning in logs, got: {[r.message for r in caplog.records]}"
|
||||
|
||||
|
||||
class TestRunJobConfigEnvVarExpansion:
|
||||
|
||||
@@ -75,7 +75,7 @@ class TestBusyCommandPersistence:
|
||||
# emulate the write that the mocked save_config_value skipped.
|
||||
monkeypatch.setattr(
|
||||
gateway_run,
|
||||
"_load_gateway_runtime_config",
|
||||
"_load_gateway_config",
|
||||
lambda: {"display": {"busy_input_mode": new_mode}},
|
||||
)
|
||||
monkeypatch.delenv("HERMES_GATEWAY_BUSY_TEXT_MODE", raising=False)
|
||||
|
||||
@@ -172,7 +172,7 @@ def test_turn_route_merges_fast_mode_with_provider_request_overrides():
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_agent_preserves_provider_request_overrides_on_gateway_path(monkeypatch):
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: {})
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
|
||||
monkeypatch.setattr(gateway_run, "_resolve_gateway_model", lambda config=None: "gpt-5.4")
|
||||
monkeypatch.setattr(
|
||||
gateway_run,
|
||||
@@ -228,7 +228,7 @@ async def test_reused_agent_turn_merges_request_overrides_not_overwrite(monkeypa
|
||||
fast-mode key while the provider extra_body survives.
|
||||
"""
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: {})
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
|
||||
monkeypatch.setattr(gateway_run, "_resolve_gateway_model", lambda config=None: "gpt-5.4")
|
||||
monkeypatch.setattr(
|
||||
gateway_run,
|
||||
|
||||
@@ -157,7 +157,7 @@ async def test_session_fast_override_beats_config_default(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
|
||||
monkeypatch.setattr(
|
||||
gateway_run,
|
||||
"_load_gateway_runtime_config",
|
||||
"_load_gateway_config",
|
||||
lambda: {"agent": {"service_tier": "fast"}},
|
||||
)
|
||||
monkeypatch.setattr(gateway_run, "_resolve_gateway_model", lambda config=None: "gpt-5.4")
|
||||
|
||||
@@ -301,7 +301,7 @@ class TestSecondaryProfileFatalRecovery:
|
||||
)
|
||||
monkeypatch.setattr(runner, "_connect_adapter_with_timeout", connect)
|
||||
monkeypatch.setattr(runner, "_connect_initial_adapter_with_timeout", connect)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: {})
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
|
||||
monkeypatch.setattr(runner, "_snapshot_profile_busy_modes", lambda *a, **k: None)
|
||||
monkeypatch.setattr("hermes_cli.plugins.discover_plugins", lambda: None)
|
||||
if entry == "startup":
|
||||
@@ -335,7 +335,7 @@ class TestSecondaryProfileFatalRecovery:
|
||||
synced = []
|
||||
runner._sync_voice_mode_state_to_adapter = synced.append
|
||||
monkeypatch.setattr("hermes_cli.env_loader.hydrate_profile_secret_sources", lambda h: {})
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: {})
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
|
||||
monkeypatch.setattr(runner, "_snapshot_profile_busy_modes", lambda *a, **k: None)
|
||||
monkeypatch.setattr("hermes_cli.plugins.discover_plugins", lambda: None)
|
||||
|
||||
|
||||
@@ -444,7 +444,7 @@ async def test_effective_mode_uses_startup_snapshot_without_rereading_config(
|
||||
def fail_config_read():
|
||||
raise AssertionError("busy-mode lookup reread config after startup")
|
||||
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", fail_config_read)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", fail_config_read)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", fail_config_read)
|
||||
|
||||
assert runner._effective_busy_input_mode(source) == "steer"
|
||||
|
||||
@@ -21,7 +21,7 @@ class TestGatewayPerModelReasoningConfig:
|
||||
},
|
||||
},
|
||||
}
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: fake_cfg)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: fake_cfg)
|
||||
|
||||
result = gateway_run.GatewayRunner._load_reasoning_config()
|
||||
assert result is not None
|
||||
@@ -42,7 +42,7 @@ class TestGatewayPerModelReasoningConfig:
|
||||
"reasoning_effort": False, # YAML boolean, not string
|
||||
},
|
||||
}
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: fake_cfg)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: fake_cfg)
|
||||
|
||||
result = gateway_run.GatewayRunner._load_reasoning_config()
|
||||
assert result is not None
|
||||
@@ -69,7 +69,7 @@ class TestGatewaySessionEffectiveModel:
|
||||
},
|
||||
},
|
||||
}
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: fake_cfg)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: fake_cfg)
|
||||
|
||||
# Session switched (session-only) to claude-opus-4.5 — its override
|
||||
# must win over the config default model's override.
|
||||
@@ -111,7 +111,7 @@ class TestApiServerPerModelReasoning:
|
||||
from gateway.platforms.api_server import APIServerAdapter
|
||||
|
||||
monkeypatch.setattr(
|
||||
gateway_run, "_load_gateway_runtime_config", lambda: self._cfg(),
|
||||
gateway_run, "_load_gateway_config", lambda: self._cfg(),
|
||||
)
|
||||
adapter = APIServerAdapter(PlatformConfig())
|
||||
|
||||
|
||||
@@ -100,7 +100,7 @@ def _setup_monkeypatches(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
|
||||
monkeypatch.setattr(
|
||||
gateway_run,
|
||||
"_load_gateway_runtime_config",
|
||||
"_load_gateway_config",
|
||||
lambda: {"agent": {"model": "test-model"}},
|
||||
)
|
||||
monkeypatch.setattr(gateway_run, "_resolve_gateway_model", lambda config=None: "test-model")
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
"""Invariants for ``hermes_cli.config_effective.load_user_config_effective`` — the one loader every
|
||||
defaults-free config reader (gateway runtime, TUI gateway, cron, ``hermes send`` bridge, doctor,
|
||||
bootstrap modules) goes through."""
|
||||
import textwrap
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def homes(tmp_path, monkeypatch):
|
||||
home = tmp_path / "home"
|
||||
home.mkdir()
|
||||
managed = tmp_path / "managed"
|
||||
managed.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
monkeypatch.setenv("HERMES_MANAGED_DIR", str(managed))
|
||||
monkeypatch.setenv("FIXTURE_USER_KEY", "user-secret")
|
||||
monkeypatch.setenv("FIXTURE_MANAGED_URL", "https://managed.example")
|
||||
_reset_caches()
|
||||
return home, managed
|
||||
|
||||
|
||||
def _reset_caches():
|
||||
import hermes_cli.config as cfg
|
||||
from hermes_cli import config_effective, managed_scope
|
||||
|
||||
cfg._LOAD_CONFIG_CACHE.clear()
|
||||
cfg._RAW_CONFIG_CACHE.clear()
|
||||
config_effective._EFFECTIVE_CACHE.clear()
|
||||
config_effective._LAST_GOOD_USER_RAW.clear()
|
||||
managed_scope.invalidate_managed_cache()
|
||||
|
||||
|
||||
def _write(path, body):
|
||||
path.write_text(textwrap.dedent(body), encoding="utf-8")
|
||||
_reset_caches()
|
||||
|
||||
|
||||
USER_YAML = """
|
||||
model:
|
||||
name: user/model
|
||||
api_key: ${FIXTURE_USER_KEY}
|
||||
provider: custom
|
||||
display:
|
||||
skin: user-skin
|
||||
"""
|
||||
MANAGED_YAML = """
|
||||
model:
|
||||
base_url: ${FIXTURE_MANAGED_URL}
|
||||
display:
|
||||
skin: managed-skin
|
||||
"""
|
||||
|
||||
|
||||
def _legacy_gateway_pipeline(config_path):
|
||||
"""The pre-unification gateway sequence (raw read → overlay → model-key canon → ${VAR} expansion)."""
|
||||
from hermes_cli import managed_scope
|
||||
from hermes_cli.config import _expand_env_vars, _normalize_root_model_keys, read_user_config_raw
|
||||
|
||||
raw = managed_scope.apply_managed_overlay(read_user_config_raw(config_path))
|
||||
return _expand_env_vars(_normalize_root_model_keys(raw))
|
||||
|
||||
|
||||
def test_effective_equals_legacy_gateway_pipeline_and_carries_no_defaults(homes):
|
||||
"""Contract: the shared loader returns byte-for-byte what the gateway's hand-rolled pipeline did
|
||||
for a user file with a managed overlay and ``${VAR}`` refs on both layers — so per-message
|
||||
gateway config reads (and the system prompt built from them) do not change — while never
|
||||
merging DEFAULT_CONFIG (a missing key stays missing)."""
|
||||
from hermes_cli.config import DEFAULT_CONFIG
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
|
||||
home, managed = homes
|
||||
_write(home / "config.yaml", USER_YAML)
|
||||
_write(managed / "config.yaml", MANAGED_YAML)
|
||||
|
||||
effective = load_user_config_effective(home / "config.yaml")
|
||||
|
||||
assert effective == _legacy_gateway_pipeline(home / "config.yaml")
|
||||
assert effective["model"] == {
|
||||
"default": "user/model", "provider": "custom", "api_key": "user-secret",
|
||||
"base_url": "https://managed.example"}
|
||||
assert effective["display"]["skin"] == "managed-skin"
|
||||
assert "provider" not in effective # root key migrated under ``model`` (canonicalization applied)
|
||||
assert "agent" not in effective and "agent" in DEFAULT_CONFIG # no DEFAULT_CONFIG merge
|
||||
|
||||
|
||||
def test_broken_yaml_serves_last_good_and_fail_closed_raises(homes):
|
||||
"""A torn mid-edit write must not silently drop user overrides: the fail-open path serves the last
|
||||
successfully parsed user file through the same pipeline; ``fail_closed`` surfaces the error to
|
||||
callers that keep their own last-good state."""
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
|
||||
home, _ = homes
|
||||
_write(home / "config.yaml", USER_YAML)
|
||||
good = load_user_config_effective(home / "config.yaml")
|
||||
|
||||
(home / "config.yaml").write_text("model: [unterminated", encoding="utf-8")
|
||||
_reset_caches_keep_last_good()
|
||||
|
||||
assert load_user_config_effective(home / "config.yaml") == good
|
||||
with pytest.raises(Exception):
|
||||
load_user_config_effective(home / "config.yaml", fail_closed=True)
|
||||
|
||||
|
||||
def _reset_caches_keep_last_good():
|
||||
import hermes_cli.config as cfg
|
||||
from hermes_cli import config_effective
|
||||
|
||||
cfg._RAW_CONFIG_CACHE.clear()
|
||||
config_effective._EFFECTIVE_CACHE.clear()
|
||||
@@ -313,16 +313,17 @@ def test_load_hermes_env_latin1_fallback_still_loads(tmp_path, monkeypatch):
|
||||
|
||||
def test_load_hermes_env_latin1_fallback_overrides_shell(tmp_path, monkeypatch):
|
||||
"""The stream-based latin-1 fallback must keep override=True semantics:
|
||||
the .env value wins over a stale shell export, same as the primary path."""
|
||||
the .env value wins over a stale shell export, same as the primary path. (A non-credential
|
||||
key name: ``*_TOKEN`` values are ASCII-sanitized by the shared loader, by design.)"""
|
||||
import os
|
||||
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
# 0xE9 forces the UnicodeDecodeError \u2192 latin-1 stream fallback.
|
||||
(hermes_home / ".env").write_bytes(b"SEND_OVR_TOKEN=caf\xe9-file\n")
|
||||
(hermes_home / ".env").write_bytes(b"SEND_OVR_LABEL=caf\xe9-file\n")
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.setenv("SEND_OVR_TOKEN", "stale-shell-value")
|
||||
monkeypatch.setenv("SEND_OVR_LABEL", "stale-shell-value")
|
||||
|
||||
from importlib import reload
|
||||
import hermes_cli.config as _hc_config
|
||||
@@ -330,7 +331,7 @@ def test_load_hermes_env_latin1_fallback_overrides_shell(tmp_path, monkeypatch):
|
||||
|
||||
send_cmd._load_hermes_env()
|
||||
|
||||
assert os.environ.get("SEND_OVR_TOKEN") == "caf\xe9-file"
|
||||
assert os.environ.get("SEND_OVR_LABEL") == "caf\xe9-file"
|
||||
|
||||
def test_load_hermes_env_fallback_read_error_is_swallowed(tmp_path, monkeypatch):
|
||||
"""An I/O error inside the latin-1 fallback must not escape \u2014 the send
|
||||
|
||||
@@ -74,7 +74,7 @@ class TestTUIPerModelReasoningConfig:
|
||||
},
|
||||
}
|
||||
monkeypatch.setattr(tui_server, "_load_cfg", lambda: fake_cfg)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: fake_cfg)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: fake_cfg)
|
||||
|
||||
tui_result = tui_server._load_reasoning_config()
|
||||
gw_result = gateway_run.GatewayRunner._load_reasoning_config()
|
||||
|
||||
+10
-26
@@ -541,7 +541,7 @@ def _configured_cwd_from_cfg(cfg: dict | None) -> str | None:
|
||||
def _profile_configured_cwd(profile_home: Path | None) -> str | None:
|
||||
"""A non-launch profile's ``terminal.cwd`` from ITS config.yaml (fail-open → None): the process-global
|
||||
``TERMINAL_CWD`` belongs to the *launch* profile, and load_config() resolves the ACTIVE profile, so
|
||||
read the file directly through the _load_cfg pipeline.
|
||||
read that file through the same effective-config pipeline as ``_load_cfg``.
|
||||
|
||||
A new session bound to another profile must take its workspace from THAT profile's config, not the stale
|
||||
env var (issue #40334). Returns an absolute, existing directory, or None for placeholders / missing /
|
||||
@@ -550,9 +550,9 @@ def _profile_configured_cwd(profile_home: Path | None) -> str | None:
|
||||
if profile_home is None:
|
||||
return None
|
||||
with contextlib.suppress(Exception):
|
||||
from hermes_cli.config import read_user_config_raw
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
p = Path(profile_home) / "config.yaml"
|
||||
return _configured_cwd_from_cfg(_expand_cfg(_apply_managed(read_user_config_raw(p)))) if p.exists() else None
|
||||
return _configured_cwd_from_cfg(load_user_config_effective(p)) if p.exists() else None
|
||||
return None
|
||||
|
||||
|
||||
@@ -1159,31 +1159,15 @@ def _load_cfg_raw() -> dict:
|
||||
return {}
|
||||
|
||||
|
||||
def _expand_cfg(cfg: dict) -> dict:
|
||||
"""``${ENV_VAR}`` expansion (same as ``load_config_readonly``); non-dict results keep the input."""
|
||||
from hermes_cli.config import _expand_env_vars
|
||||
expanded = _expand_env_vars(cfg)
|
||||
return expanded if isinstance(expanded, dict) else cfg
|
||||
|
||||
|
||||
def _load_cfg() -> dict:
|
||||
"""Behavioral config read: raw user file + managed overlay + ${VAR} expansion — ``load_config_readonly``
|
||||
minus the DEFAULT_CONFIG merge (callers treat a missing key as "unset"; merging would break
|
||||
``_load_cfg() == {}`` sentinels). Never pass the result to ``_save_cfg`` (use ``_load_cfg_raw()``)."""
|
||||
cfg = _apply_managed(_load_cfg_raw())
|
||||
"""Behavioral config read: the effective USER config (managed overlay, ``${VAR}`` expansion, model-key
|
||||
canon) minus the DEFAULT_CONFIG merge — callers treat a missing key as "unset", so merging would break
|
||||
``_load_cfg() == {}`` sentinels. Fail-open to ``{}``. Never pass the result to ``_save_cfg`` (use
|
||||
``_load_cfg_raw()``)."""
|
||||
with contextlib.suppress(Exception):
|
||||
cfg = _expand_cfg(cfg)
|
||||
return cfg
|
||||
|
||||
|
||||
def _apply_managed(cfg: dict) -> dict:
|
||||
"""Overlay administrator-pinned managed-scope values (read-side only, fail-open): this backend builds
|
||||
config independently of load_config, so managed skin/reasoning_effort/service_tier/provider_routing
|
||||
would otherwise be silently ignored."""
|
||||
with contextlib.suppress(Exception):
|
||||
from hermes_cli import managed_scope
|
||||
return managed_scope.apply_managed_overlay(cfg if isinstance(cfg, dict) else {})
|
||||
return cfg
|
||||
from hermes_cli.config_effective import load_user_config_effective
|
||||
return load_user_config_effective(_active_config_path())
|
||||
return {}
|
||||
|
||||
|
||||
def _save_cfg(cfg: dict):
|
||||
|
||||
Reference in New Issue
Block a user