diff --git a/config.example.yaml b/config.example.yaml index 1ab39970..51020d81 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1,5 +1,12 @@ # Nerve — Personal AI Assistant Configuration # Copy to config.yaml and customize. Secrets go in config.local.yaml (gitignored). +# +# Any string value may reference an environment variable — the recommended way +# to keep secrets out of committed/shared config: +# key: ${MY_VAR} # required — load fails if MY_VAR is unset +# key: ${MY_VAR:-default} # optional — uses default when unset/empty +# Only the braced ${...} form is interpolated; a bare "$" is left untouched +# (so bcrypt hashes / jwt secrets are safe). Use $$ for a literal "$". # Core workspace: ~/nerve-workspace # Path to workspace with SOUL.md, MEMORY.md, etc. diff --git a/docs/config.md b/docs/config.md index 5717937e..0c9da276 100644 --- a/docs/config.md +++ b/docs/config.md @@ -13,6 +13,51 @@ A blank value (`runs_dir:` with nothing after it, or `""`) means *unset*, so the documented default applies. That includes `gateway.ssl.cert`/`key`: blank means TLS is off, not TLS with an empty certificate path. +## Environment Variable References + +Any string value in `config.yaml` or `config.local.yaml` may reference an +environment variable, so secrets can come from the environment (or a secret +store) instead of being written into a file: + +```yaml +anthropic_api_key: ${ANTHROPIC_API_KEY} # required: load fails if unset +gateway: + host: ${BIND_HOST:-127.0.0.1} # optional: default when unset or empty +``` + +| Form | Behavior | +|------|----------| +| `${VAR}` | Required. Loading fails with an error listing every unresolved variable. Only an *unset* variable is an error; `VAR=""` resolves to an empty string. | +| `${VAR:-default}` | Optional. Uses `default` when `VAR` is unset or empty (shell `:-` semantics). | +| `$$` | A literal `$`, so `$${X}` yields the text `${X}`. | + +Only the braced `${...}` form is interpolated. A bare `$` is never touched, so +bcrypt `password_hash` values (`$2b$...`), jwt secrets and connection strings +are safe as written. Interpolation runs once, after `config.local.yaml` is +merged on top, so either file may use references. + +Resolved values arrive as strings and are converted back to the field's declared +type, including `int | None`, `list[int]` and `list[str]`. So `port: ${PORT}` and +`enabled: ${FEATURE}` behave the same as literal YAML values. + +- Booleans accept `true/false`, `1/0`, `yes/no`, `on/off`, `y/n`, `t/f` + (case-insensitive). `enabled: ${FLAG}` with `FLAG=false` is **off**: the + string is parsed, not tested for truthiness. An empty value (`FLAG=`) and a + bare `enabled:` are also off, so blanking a variable reliably disables a + feature. +- On a list field, **one reference is one element.** `accounts: ${GMAIL}` with + `GMAIL=a@example.com` yields `["a@example.com"]`. Numeric lists + (`list[int]`) additionally split on commas, because a comma cannot occur + inside a number — so `exclude_chats: ${SKIP}` with `SKIP=-100,-200` yields two + ids. String lists are never split, since a comma is legal inside the value: + the default `langfuse.redact_patterns` are regexes containing `{20,}`. To set + several strings, write a YAML list. +- An unrecognized value is logged with the field that owns it, and that field + keeps its documented default. Integers are parsed with `int()`, so `"1.5"` + and `"1e3"` are rejected rather than truncated. +- Defaults are not re-scanned: `${A:-${B}}` yields the literal `${B}` when `A` + is unset. Use a single reference instead. + ## Config Directory Resolution `nerve` commands locate the config directory via a waterfall, so they work diff --git a/nerve/cli.py b/nerve/cli.py index 44eaa3a7..1b3d3566 100644 --- a/nerve/cli.py +++ b/nerve/cli.py @@ -32,6 +32,7 @@ from nerve import paths from nerve.config import ( + ConfigError, RESUME_QUEUE_FILE, load_config, resolve_config_dir, @@ -185,7 +186,13 @@ def main(ctx: click.Context, config_dir: str | None, verbose: bool) -> None: resolved_dir, config_source = resolve_config_dir(config_dir) if not resolved_dir.is_absolute(): resolved_dir = resolved_dir.resolve() - config = load_config(resolved_dir) + try: + config = load_config(resolved_dir) + except ConfigError as e: + # Render config errors (e.g. an unresolved required ${ENV_VAR}) as a + # clean message + non-zero exit instead of a raw traceback — this + # callback runs before every subcommand, including `nerve doctor`. + raise click.ClickException(str(e)) from e set_config(config) ctx.ensure_object(dict) ctx.obj["config"] = config diff --git a/nerve/coerce.py b/nerve/coerce.py new file mode 100644 index 00000000..3f8a6f47 --- /dev/null +++ b/nerve/coerce.py @@ -0,0 +1,300 @@ +"""Scalar coercion at the config boundary. + +YAML gives a real ``bool`` for ``enabled: false`` and a real ``int`` for +``port: 8900``. A ``${ENV_VAR}`` reference does not: interpolation is a string +substitution, so the value that reaches a builder is always ``str``. +``bool("false")`` is ``True``, and an int field left as ``"8900"`` raises at +its first arithmetic or ``socket.bind`` use. + +That is not a corner case here. Under lockdown the machine-local config layers +are discarded, so every per-machine value *has* to be an env reference in the +tracked settings — precisely the shape that breaks. A quoted YAML scalar +(``enabled: "false"``) reaches the same code path with no env var involved at +all. + +Coercion is applied at the ``from_dict`` boundary (see :func:`coerced`) rather +than in ``__init__``, so constructing a config object in code keeps exact-type +semantics and a test can still pass a deliberately odd value. The decorator +goes on *every* ``from_dict``, including the ones whose dataclass currently has +no numeric or boolean field, so that adding one later needs no ceremony. + +Coercers are *lenient* about values they cannot parse: a bad value in a section +that isn't even active must not stop the daemon from starting. They are logged +with the owning ``Class.field`` so the line is actually actionable. + +This module lives outside ``nerve.config`` so that dataclasses defined in other +packages (which ``nerve.config`` imports) can use it without an import cycle. +Keep it dependency-free. +""" + +from __future__ import annotations + +import dataclasses +import functools +import logging +import types +import typing +from typing import Any + +logger = logging.getLogger(__name__) + +TRUTHY = frozenset({"1", "true", "yes", "on", "y", "t"}) +FALSY = frozenset({"0", "false", "no", "off", "n", "f"}) + +# Sentinel for "this value could not be converted" — distinct from None, which +# is a legitimate value for an Optional field. +_UNCONVERTIBLE = object() + + +def _log_reject(value: Any, default: Any, kind: str, label: str) -> None: + logger.warning( + "%signoring non-%s config value %r, using %r", + f"{label}: " if label else "", + kind, + value, + default, + ) + + +def lenient_int(value: Any, default: int, *, label: str = "") -> int: + """Best-effort ``int`` coercion, falling back to ``default``. + + Note that this is ``int()``, so ``"1.5"`` and ``"1e3"`` are rejected + rather than truncated — they fall back to the default with a warning. + """ + try: + return int(value) if value is not None else default + except (TypeError, ValueError): + _log_reject(value, default, "integer", label) + return default + + +def lenient_float(value: Any, default: float, *, label: str = "") -> float: + """Best-effort ``float`` coercion, falling back to ``default``.""" + try: + return float(value) if value is not None else default + except (TypeError, ValueError): + _log_reject(value, default, "numeric", label) + return default + + +def as_bool(value: Any, default: bool, *, label: str = "") -> bool: + """Best-effort ``bool`` coercion accepting the usual YAML/shell spellings. + + Three cases, and the distinction between them matters: + + * ``None`` (a bare ``enabled:`` in YAML) and ``""`` (a ``${VAR}`` whose + variable is set but empty) are **false**. They were false before this + module existed — ``bool(None)`` and ``bool("")`` — and an empty env var + is the natural way to spell "off" for a value that must be an env + reference under lockdown. Returning the field's default here would make + a blanked-out flag silently keep whatever the tracked config said. + * A recognized spelling is parsed, not tested for truthiness. This is the + whole point: ``bool("false")`` is ``True``. + * Anything else falls back to the field's declared default and is logged. + Falling back rather than guessing means a typo can't flip a flag to the + opposite of both what it says and what the config declares. + """ + if isinstance(value, bool): + return value + if value is None: + return False + if isinstance(value, (int, float)): + return bool(value) + if isinstance(value, str): + text = value.strip().lower() + if not text: + return False + if text in TRUTHY: + return True + if text in FALSY: + return False + _log_reject(value, default, "boolean", label) + return default + + +_COERCERS: dict[type, Any] = {bool: as_bool, int: lenient_int, float: lenient_float} + + +def _convert_element(base: type, value: Any) -> Any: + """Strict conversion for a list element — no default to fall back to.""" + if type(value) is base: + return value + if value is None: + # Already the answer for the numeric types (``int(None)`` raises), but + # spelled out because ``str(None)`` would otherwise turn a YAML ``~`` + # entry into the literal text "None" instead of flagging it. + return _UNCONVERTIBLE + if base is bool: + if isinstance(value, (int, float)): + return bool(value) + if isinstance(value, str): + text = value.strip().lower() + if text in TRUTHY: + return True + if text in FALSY: + return False + return _UNCONVERTIBLE + try: + return base(value) + except (TypeError, ValueError): + return _UNCONVERTIBLE + + +def _scalar_to_list(value: Any, base: type) -> list[Any] | None: + """Normalize a non-list value on a ``list[X]`` field, or ``None`` to skip. + + A ``${VAR}`` reference on a list field interpolates to a plain string, and + lockdown makes env references mandatory for per-machine values, which + ``telegram.allowed_users`` and ``sync.telegram.exclude_chats`` both are. + Leaving such a string alone moves the problem to the consumers rather than + avoiding it: both call ``set()`` on the value, and ``set("123")`` is + ``{"1", "2", "3"}``. An allowlist then fails closed, but an exclude list + fails open — an int chat id never equals a one-character string, so an + excluded chat is ingested. + + **How a string becomes a list depends on the element type**, because the two + cases have different safe answers: + + * ``list[int]``/``list[float]``/``list[bool]`` split on comma, matching how + the bootstrap wizard parses this same field. A comma cannot occur inside a + valid number, so the separator is unambiguous. + * ``list[str]`` is **wrapped as a single element**, never split. A comma is + legal inside a string — three of the four default + ``langfuse.redact_patterns`` are regexes containing ``{20,}`` — so + splitting would corrupt values that are correct as written. One ``${VAR}`` + is one element; several values are spelled as a YAML list. + + A blank string yields the empty list rather than one empty entry, and a lone + non-string scalar (``exclude_chats: 42``) is wrapped so it reaches the + element coercion below. + """ + if value is None: + return [] + if isinstance(value, str): + if base is str: + return [value] if value.strip() else [] + return [part.strip() for part in value.split(",") if part.strip()] + if isinstance(value, (bool, int, float)): + return [value] + return None + + +def _classify(declared: Any) -> tuple[str, type] | None: + """Map a declared annotation onto ``(kind, base_type)``. + + Handles the three shapes that actually appear in the config dataclasses: + a bare scalar, ``X | None``, and ``list[X]``. Anything else is left alone. + + ``str`` is absent from ``_COERCERS`` — a bare ``str`` field needs no + coercion, since interpolation already yields one — but ``list[str]`` does, + because the container is the thing that gets lost. See + :func:`_scalar_to_list`. + """ + if declared in _COERCERS: + return ("scalar", declared) + origin = typing.get_origin(declared) + args = typing.get_args(declared) + if origin in (typing.Union, types.UnionType): + non_none = [a for a in args if a is not type(None)] + if len(non_none) == 1 and non_none[0] in _COERCERS: + return ("optional", non_none[0]) + return None + if origin is list and len(args) == 1 and (args[0] in _COERCERS or args[0] is str): + return ("list", args[0]) + return None + + +@functools.lru_cache(maxsize=None) +def scalar_fields(cls: type) -> tuple[tuple[str, str, type, Any], ...]: + """``(name, kind, base_type, default)`` for each coercible field.""" + hints = typing.get_type_hints(cls) + out: list[tuple[str, str, type, Any]] = [] + for f in dataclasses.fields(cls): + classified = _classify(hints.get(f.name)) + if classified is None: + continue + kind, base = classified + if f.default is not dataclasses.MISSING: + default = f.default + elif f.default_factory is not dataclasses.MISSING: + default = f.default_factory() + else: + default = base() + out.append((f.name, kind, base, default)) + return tuple(out) + + +def coerce_scalars(obj: Any) -> Any: + """Force declared bool/int/float fields to their annotated type.""" + cls_name = type(obj).__name__ + for name, kind, base, default in scalar_fields(type(obj)): + value = getattr(obj, name) + label = f"{cls_name}.{name}" + + if kind == "list": + widened = False + if not isinstance(value, list): + normalized = _scalar_to_list(value, base) + if normalized is None: + logger.warning( + "%s: ignoring non-list config value %r, leaving it as-is", + label, + value, + ) + continue + value = normalized + widened = True + converted = [_convert_element(base, v) for v in value] + if any(c is _UNCONVERTIBLE for c in converted): + # Keep the offending entries verbatim rather than dropping + # them: silently shrinking a list of allowed users or excluded + # chats is worse than a type that stands out. + for original, result in zip(value, converted): + if result is _UNCONVERTIBLE: + _log_reject(original, original, base.__name__, f"{label}[]") + converted = [ + original if result is _UNCONVERTIBLE else result + for original, result in zip(value, converted) + ] + if ( + widened + or converted != value + or any(type(a) is not type(b) for a, b in zip(converted, value)) + ): + setattr(obj, name, converted) + continue + + if kind == "optional" and value is None: + continue + + # `type() is` rather than isinstance: a bool sitting in an int field, + # or an int in a float field, should still be normalized. + if type(value) is not base: + setattr(obj, name, _COERCERS[base](value, default, label=label)) + return obj + + +def coerced(fn): + """Decorator for ``from_dict``: coerce declared scalars on the way out. + + Goes *under* ``@classmethod``:: + + @classmethod + @coerced + def from_dict(cls, d: dict) -> GatewayConfig: + ... + + (The other order raises ``TypeError: 'classmethod' object is not + callable`` at import, so it cannot silently no-op.) + + Builders should hand raw YAML values straight to the constructor. An eager + ``bool(d.get("enabled", True))`` inside the builder defeats this entirely, + because ``bool("false")`` is already ``True`` by the time it arrives. + """ + + @functools.wraps(fn) + def wrapper(cls, *args, **kwargs): + return coerce_scalars(fn(cls, *args, **kwargs)) + + return wrapper diff --git a/nerve/config.py b/nerve/config.py index 5a9e5853..ae5e2d11 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -15,6 +15,8 @@ from typing import Any from nerve import paths +from nerve.coerce import coerced as _coerced +from nerve.coerce import lenient_int as _lenient_int import yaml @@ -93,12 +95,103 @@ def _setting_str(value: object, default: str = "") -> str: return str(value or "").strip() or default +def _str_list(value: object, *, clean: bool = False) -> object: + """Copy a ``list[str]`` field's value, or hand a non-list straight through. + + ``list("a@b.com")`` is nineteen single-character entries, and a builder that + widens the value itself gets there before ``@_coerced`` can intervene — at + which point nothing downstream can tell the result from a genuine list. So a + non-list is returned untouched for :func:`nerve.coerce.coerce_scalars` to + wrap as one element, which is what a ``${VAR}`` reference on a list field + means. + + ``clean`` additionally strips each entry and drops the blanks, for the fields + that were already doing that. + """ + if not isinstance(value, list): + return value + if clean: + return [s for s in (str(v or "").strip() for v in value) if s] + return list(value) +class ConfigError(ValueError): + """Raised when configuration cannot be loaded (e.g. an unresolved + required ``${ENV_VAR}`` reference).""" + + +# Matches an escaped ``$$`` (literal dollar) or a ``${...}`` reference. +# We deliberately only interpolate the *braced* form so that values which +# legitimately contain a bare ``$`` — bcrypt ``password_hash`` (``$2b$..``), +# jwt secrets, connection strings — are never touched. +_ENV_REF_RE = re.compile(r"\$\$|\$\{([^}]*)\}") + + +def _interpolate_str(value: str, missing: list[str]) -> str: + """Resolve ``${VAR}`` / ``${VAR:-default}`` references in a single string. + + * ``${VAR}`` — required; if unset, the name is appended to + ``missing`` and the reference is left intact (the caller raises). + * ``${VAR:-default}`` — use ``default`` when VAR is unset *or* empty + (shell ``:-`` semantics). + * ``$$`` — an escaped literal ``$`` (so ``$${X}`` yields the + literal text ``${X}``). + """ + + def _replace(match: re.Match[str]) -> str: + if match.group(0) == "$$": + return "$" + expr = match.group(1) + if ":-" in expr: + name, default = expr.split(":-", 1) + name = name.strip() + resolved = os.environ.get(name) + return resolved if resolved else default + name = expr.strip() + if not name: + # `${}` — almost certainly a typo. Leave it intact rather than + # emitting a confusing "missing variable: ''" error. + return match.group(0) + if name in os.environ: + return os.environ[name] + missing.append(name) + return match.group(0) + + return _ENV_REF_RE.sub(_replace, value) + + +def _interpolate_env(obj: Any, missing: list[str]) -> Any: + """Recursively resolve ``${ENV_VAR}`` references in all string values of a + merged config structure. Non-string leaves are returned unchanged.""" + if isinstance(obj, dict): + return {k: _interpolate_env(v, missing) for k, v in obj.items()} + if isinstance(obj, list): + return [_interpolate_env(v, missing) for v in obj] + if isinstance(obj, str): + return _interpolate_str(obj, missing) + return obj + + +def _resolve_env_refs(merged: dict[str, Any]) -> dict[str, Any]: + """Interpolate ``${ENV_VAR}`` references across a merged config dict, + raising :class:`ConfigError` listing every unresolved required variable.""" + missing: list[str] = [] + resolved = _interpolate_env(merged, missing) + if missing: + names = ", ".join(sorted(set(missing))) + raise ConfigError( + f"Unresolved required environment variable(s) in config: {names}. " + "Set them in the environment, or use ${VAR:-default} to supply a " + "fallback." + ) + return resolved + + @dataclass class SSLConfig: cert: Path | None = None key: Path | None = None @classmethod + @_coerced def from_dict(cls, d: dict) -> SSLConfig: return cls(cert=_expand_path(d.get("cert")), key=_expand_path(d.get("key"))) @@ -114,6 +207,7 @@ class GatewayConfig: ssl: SSLConfig = field(default_factory=SSLConfig) @classmethod + @_coerced def from_dict(cls, d: dict) -> GatewayConfig: return cls( host=d.get("host", "0.0.0.0"), @@ -143,6 +237,7 @@ def is_bedrock(self) -> bool: return self.type == "bedrock" @classmethod + @_coerced def from_dict(cls, d: dict) -> ProviderConfig: return cls( type=d.get("type", "anthropic"), @@ -176,12 +271,13 @@ class PromptRewriteConfig: timeout_seconds: float = 45.0 @classmethod + @_coerced def from_dict(cls, d: dict) -> PromptRewriteConfig: return cls( - enabled=bool(d.get("enabled", True)), + enabled=d.get("enabled", True), model=d.get("model", ""), - max_tokens=int(d.get("max_tokens", 1024)), - timeout_seconds=float(d.get("timeout_seconds", 45.0)), + max_tokens=d.get("max_tokens", 1024), + timeout_seconds=d.get("timeout_seconds", 45.0), ) @@ -278,6 +374,7 @@ def resolved_cron_backend(self) -> str: return self.cron_backend or self.backend @classmethod + @_coerced def from_dict(cls, d: dict) -> AgentConfig: return cls( backend=str(d.get("backend", "claude")).strip().lower(), @@ -289,28 +386,22 @@ def from_dict(cls, d: dict) -> AgentConfig: str(k): str(v or "") for k, v in (d.get("model_aliases") or {}).items() }, - models=[ - s for s in ( - str(m or "").strip() for m in (d.get("models") or []) - ) if s - ], + models=_str_list(d.get("models"), clean=True), max_turns=d.get("max_turns", 100), max_concurrent=d.get("max_concurrent", 32), thinking=str(d.get("thinking", "max")), effort=str(d.get("effort", "max")), cron_effort=str(d.get("cron_effort", "medium")), context_1m=d.get("context_1m", True), - context_1m_excluded_models=list( - d.get("context_1m_excluded_models", []) or [] + context_1m_excluded_models=_str_list( + d.get("context_1m_excluded_models") ), cache_ttl=str(d.get("cache_ttl", "5m")), - cache_ttl_excluded_models=list( - d.get("cache_ttl_excluded_models", []) or [] - ), - cli_idle_timeout_seconds=int(d.get("cli_idle_timeout_seconds", 900)), - background_agent_permissions=bool( - d.get("background_agent_permissions", True) + cache_ttl_excluded_models=_str_list( + d.get("cache_ttl_excluded_models") ), + cli_idle_timeout_seconds=d.get("cli_idle_timeout_seconds", 900), + background_agent_permissions=d.get("background_agent_permissions", True), prompt_rewrite=PromptRewriteConfig.from_dict(d.get("prompt_rewrite") or {}), ) @@ -341,6 +432,7 @@ class TelegramConfig: dm_policy: str = "pairing" @classmethod + @_coerced def from_dict(cls, d: dict) -> TelegramConfig: dm_policy = d.get("dm_policy", "pairing") if dm_policy not in ("pairing", "open"): @@ -353,7 +445,13 @@ def from_dict(cls, d: dict) -> TelegramConfig: return cls( enabled=d.get("enabled", True), bot_token=d.get("bot_token", ""), - allowed_users=[int(u) for u in d.get("allowed_users", []) or []], + # Deliberately uncast. The declared list[int] is converted after + # construction, which logs an unconvertible entry and keeps it + # verbatim; casting here raises instead, so one bad value in a + # section that may not even be enabled would stop the daemon + # booting. It also stopped a bare string being read character by + # character — "123" became the three user IDs 1, 2 and 3. + allowed_users=d.get("allowed_users") or [], stream_mode=d.get("stream_mode", "partial"), dm_policy=dm_policy, ) @@ -374,6 +472,7 @@ class TelegramSyncConfig: condense: bool = False @classmethod + @_coerced def from_dict(cls, d: dict) -> TelegramSyncConfig: return cls( enabled=d.get("enabled", True), @@ -404,6 +503,7 @@ class GmailSyncConfig: condense_prompt: str = "" # Custom prompt for LLM condensation (overrides default) @classmethod + @_coerced def from_dict(cls, d: dict) -> GmailSyncConfig: return cls( enabled=d.get("enabled", True), @@ -444,6 +544,7 @@ class GitHubSyncConfig: deny_actors: list[str] = field(default_factory=list) @classmethod + @_coerced def from_dict(cls, d: dict) -> GitHubSyncConfig: return cls( enabled=d.get("enabled", True), @@ -474,6 +575,7 @@ class GitHubEventsSyncConfig: model: str = "" @classmethod + @_coerced def from_dict(cls, d: dict) -> GitHubEventsSyncConfig: return cls( enabled=d.get("enabled", False), @@ -507,6 +609,7 @@ class GitHubReposSyncConfig: model: str = "" @classmethod + @_coerced def from_dict(cls, d: dict) -> GitHubReposSyncConfig: return cls( enabled=d.get("enabled", False), @@ -540,16 +643,17 @@ class CodexOriginConfig: transport: dict = field(default_factory=dict) @classmethod + @_coerced def from_dict(cls, d: dict) -> CodexOriginConfig: return cls( id=d.get("id", "local"), type=d.get("type", "local_rollout"), - enabled=bool(d.get("enabled", True)), + enabled=d.get("enabled", True), path=_setting_str(d.get("path"), "~/.codex/sessions"), archive_path=_setting_str( d.get("archive_path"), "~/.codex/archived_sessions" ), - poll_interval_seconds=float(d.get("poll_interval_seconds", 2.0)), + poll_interval_seconds=d.get("poll_interval_seconds", 2.0), transport=d.get("transport", {}), ) @@ -571,10 +675,11 @@ class CodexWorkspaceFilterConfig: explicit_paths: list[str] = field(default_factory=list) @classmethod + @_coerced def from_dict(cls, d: dict) -> CodexWorkspaceFilterConfig: return cls( mode=str(d.get("mode", "nerve_workspace")), - explicit_paths=list(d.get("explicit_paths", [])), + explicit_paths=_str_list(d.get("explicit_paths")), ) @@ -594,6 +699,7 @@ class CodexSyncConfig: store_encrypted_reasoning: bool = True @classmethod + @_coerced def from_dict(cls, d: dict) -> CodexSyncConfig: raw_origins = d.get("origins", []) origins = [ @@ -602,12 +708,12 @@ def from_dict(cls, d: dict) -> CodexSyncConfig: if isinstance(o, dict) ] return cls( - enabled=bool(d.get("enabled", False)), + enabled=d.get("enabled", False), workspace_filter=CodexWorkspaceFilterConfig.from_dict( d.get("workspace_filter", {}), ), origins=origins, - store_encrypted_reasoning=bool(d.get("store_encrypted_reasoning", True)), + store_encrypted_reasoning=d.get("store_encrypted_reasoning", True), ) @@ -623,6 +729,7 @@ class SyncConfig: consumer_cursor_ttl_days: int = 2 # Consumer cursors expire after N days of inactivity @classmethod + @_coerced def from_dict(cls, d: dict) -> SyncConfig: return cls( telegram=TelegramSyncConfig.from_dict(d.get("telegram", {})), @@ -642,6 +749,7 @@ class MemoryCategoryConfig: description: str @classmethod + @_coerced def from_dict(cls, d: dict) -> MemoryCategoryConfig: return cls(name=d["name"], description=d.get("description", "")) @@ -658,6 +766,7 @@ class MemoryConfig: categories: list[MemoryCategoryConfig] = field(default_factory=list) @classmethod + @_coerced def from_dict(cls, d: dict) -> MemoryConfig: default_dsn = f"sqlite:///{paths.memu_sqlite()}" raw_cats = d.get("categories", []) @@ -668,8 +777,8 @@ def from_dict(cls, d: dict) -> MemoryConfig: fast_model=d.get("fast_model", "claude-haiku-4-5-20251001"), embed_model=d.get("embed_model", ""), sqlite_dsn=d.get("sqlite_dsn", default_dsn), - semantic_dedup_threshold=float(d.get("semantic_dedup_threshold", 0.85)), - knowledge_filter=bool(d.get("knowledge_filter", False)), + semantic_dedup_threshold=d.get("semantic_dedup_threshold", 0.85), + knowledge_filter=d.get("knowledge_filter", False), categories=categories, ) @@ -683,6 +792,7 @@ class CronConfig: gate_plugins_dir: Path = field(default_factory=lambda: paths.cron_dir() / "gates") @classmethod + @_coerced def from_dict(cls, d: dict) -> CronConfig: return cls( jobs_file=_expand_path(d.get("jobs_file")) or paths.cron_dir() / "jobs.yaml", @@ -712,16 +822,17 @@ class BackupConfig: notify_on_success: bool = False # low-priority digest line @classmethod + @_coerced def from_dict(cls, d: dict) -> BackupConfig: return cls( - enabled=bool(d.get("enabled", False)), + enabled=d.get("enabled", False), target_dir=_setting_str(d.get("target_dir")), - interval_hours=int(d.get("interval_hours", 24)), - retention_count=int(d.get("retention_count", 7)), - include_workspace=bool(d.get("include_workspace", True)), - workspace_excludes=list(d.get("workspace_excludes", []) or []), - notify_on_failure=bool(d.get("notify_on_failure", True)), - notify_on_success=bool(d.get("notify_on_success", False)), + interval_hours=d.get("interval_hours", 24), + retention_count=d.get("retention_count", 7), + include_workspace=d.get("include_workspace", True), + workspace_excludes=_str_list(d.get("workspace_excludes")), + notify_on_failure=d.get("notify_on_failure", True), + notify_on_success=d.get("notify_on_success", False), ) @@ -734,6 +845,7 @@ class ReviewLoopLegConfig: effort: str = "" # "" = source default @classmethod + @_coerced def from_dict(cls, d: dict, default_engine: str) -> ReviewLoopLegConfig: engine = str(d.get("engine", default_engine)) if engine not in ("claude-workflow", "codex-ultracode"): @@ -743,8 +855,8 @@ def from_dict(cls, d: dict, default_engine: str) -> ReviewLoopLegConfig: ) return cls( engine=engine, - model=str(d.get("model", "")), - effort=str(d.get("effort", "")), + model=d.get("model", ""), + effort=d.get("effort", ""), ) @@ -783,10 +895,13 @@ class ReviewLoopConfig: reconcile_interval_seconds: int = 600 @classmethod + @_coerced def from_dict(cls, d: dict) -> ReviewLoopConfig: adoption = d.get("criteria_adoption", "no") if adoption is False: adoption = "no" # YAML parses an unquoted `no` as boolean False + # Normalized, not coerced: this one is a closed set, so the value is + # lowered and checked here rather than left to the scalar coercer. adoption = str(adoption).lower() if adoption not in ("no", "ask", "auto"): raise ValueError( @@ -794,25 +909,25 @@ def from_dict(cls, d: dict) -> ReviewLoopConfig: f"'auto', got {adoption!r}" ) return cls( - enabled=bool(d.get("enabled", True)), + enabled=d.get("enabled", True), implementer=ReviewLoopLegConfig.from_dict( d.get("implementer", {}) or {}, "claude-workflow", ), verifier=ReviewLoopLegConfig.from_dict( d.get("verifier", {}) or {}, "codex-ultracode", ), - max_iterations=int(d.get("max_iterations", 3)), - default_budget_usd=float(d.get("default_budget_usd", 10.0)), - min_leg_budget_usd=float(d.get("min_leg_budget_usd", 0.5)), - verifier_reserve_fraction=float(d.get("verifier_reserve_fraction", 0.15)), + max_iterations=d.get("max_iterations", 3), + default_budget_usd=d.get("default_budget_usd", 10.0), + min_leg_budget_usd=d.get("min_leg_budget_usd", 0.5), + verifier_reserve_fraction=d.get("verifier_reserve_fraction", 0.15), criteria_adoption=adoption, - max_new_criteria_per_iteration=int(d.get("max_new_criteria_per_iteration", 3)), - max_discovered_criteria=int(d.get("max_discovered_criteria", 12)), - discovery_grace_rounds=int(d.get("discovery_grace_rounds", 2)), - auto_reissue_implementer=bool(d.get("auto_reissue_implementer", False)), - escalation_reproposals=int(d.get("escalation_reproposals", 2)), - verifier_sandbox=str(d.get("verifier_sandbox", "workspace-write")), - reconcile_interval_seconds=int(d.get("reconcile_interval_seconds", 600)), + max_new_criteria_per_iteration=d.get("max_new_criteria_per_iteration", 3), + max_discovered_criteria=d.get("max_discovered_criteria", 12), + discovery_grace_rounds=d.get("discovery_grace_rounds", 2), + auto_reissue_implementer=d.get("auto_reissue_implementer", False), + escalation_reproposals=d.get("escalation_reproposals", 2), + verifier_sandbox=d.get("verifier_sandbox", "workspace-write"), + reconcile_interval_seconds=d.get("reconcile_interval_seconds", 600), ) @@ -848,15 +963,20 @@ class WorkflowRunsConfig: review_loop: ReviewLoopConfig = field(default_factory=ReviewLoopConfig) @classmethod + @_coerced def from_dict(cls, d: dict) -> WorkflowRunsConfig: + # Raw values through: an eager cast here happens before interpolated + # strings are converted, and `bool("false")` is already True. That + # matters most for allow_unbudgeted, whose whole job is to refuse + # runs that would spend without a cap. return cls( - enabled=bool(d.get("enabled", True)), + enabled=d.get("enabled", True), runs_dir=_expand_path(d.get("runs_dir")) or paths.nerve_path("workflow-runs"), - poll_interval_seconds=int(d.get("poll_interval_seconds", 60)), - warn_fraction=float(d.get("warn_fraction", 0.8)), - kill_grace_seconds=int(d.get("kill_grace_seconds", 30)), - max_concurrent_runs=int(d.get("max_concurrent_runs", 2)), - allow_unbudgeted=bool(d.get("allow_unbudgeted", False)), + poll_interval_seconds=d.get("poll_interval_seconds", 60), + warn_fraction=d.get("warn_fraction", 0.8), + kill_grace_seconds=d.get("kill_grace_seconds", 30), + max_concurrent_runs=d.get("max_concurrent_runs", 2), + allow_unbudgeted=d.get("allow_unbudgeted", False), review_loop=ReviewLoopConfig.from_dict(d.get("review_loop", {}) or {}), ) @@ -876,8 +996,9 @@ class HouseOfAgentsConfig: enabled: bool = False @classmethod + @_coerced def from_dict(cls, d: dict) -> HouseOfAgentsConfig: - return cls(enabled=bool(d.get("enabled", False))) + return cls(enabled=d.get("enabled", False)) @dataclass @@ -892,6 +1013,7 @@ class SessionsConfig: star_project_hook: bool = False # opt-in; fire an internal agent turn on star/unstar transition @classmethod + @_coerced def from_dict(cls, d: dict) -> SessionsConfig: return cls( archive_after_days=d.get("archive_after_days", 30), @@ -901,7 +1023,7 @@ def from_dict(cls, d: dict) -> SessionsConfig: memorize_interval_minutes=d.get("memorize_interval_minutes", 30), sticky_period_minutes=d.get("sticky_period_minutes", 120), client_idle_timeout_minutes=d.get("client_idle_timeout_minutes", 60), - star_project_hook=bool(d.get("star_project_hook", False)), + star_project_hook=d.get("star_project_hook", False), ) @@ -928,12 +1050,13 @@ class RetentionConfig: interval_hours: int = 24 @classmethod + @_coerced def from_dict(cls, d: dict) -> RetentionConfig: return cls( - enabled=bool(d.get("enabled", False)), - retention_days=max(1, int(d.get("retention_days", 90))), - retention_full_days=max(1, int(d.get("retention_full_days", 30))), - interval_hours=max(1, int(d.get("interval_hours", 24))), + enabled=d.get("enabled", False), + retention_days=max(1, _lenient_int(d.get("retention_days"), 90)), + retention_full_days=max(1, _lenient_int(d.get("retention_full_days"), 30)), + interval_hours=max(1, _lenient_int(d.get("interval_hours"), 24)), ) @@ -943,6 +1066,7 @@ class AuthConfig: jwt_secret: str = "" @classmethod + @_coerced def from_dict(cls, d: dict) -> AuthConfig: return cls( password_hash=d.get("password_hash", ""), @@ -963,6 +1087,7 @@ class NotificationsConfig: }) @classmethod + @_coerced def from_dict(cls, d: dict) -> NotificationsConfig: return cls( channels=d.get("channels", ["web", "telegram"]), @@ -981,6 +1106,7 @@ class ChannelsConfig: """Global channel settings.""" @classmethod + @_coerced def from_dict(cls, d: dict) -> ChannelsConfig: return cls() @@ -992,6 +1118,7 @@ class DockerConfig: extra_mounts: list[str] = field(default_factory=list) # e.g. ["~/code:/code"] @classmethod + @_coerced def from_dict(cls, d: dict) -> DockerConfig: return cls( extra_mounts=d.get("extra_mounts", []), @@ -1011,6 +1138,7 @@ class ProxyConfig: log_file: Path = field(default_factory=lambda: paths.nerve_path("proxy.log")) @classmethod + @_coerced def from_dict(cls, d: dict) -> ProxyConfig: return cls( enabled=d.get("enabled", False), @@ -1057,11 +1185,12 @@ def openai_base_url(self) -> str: return f"http://{self.host}:{self.port}/v1" @classmethod + @_coerced def from_dict(cls, d: dict) -> OllamaConfig: return cls( - enabled=bool(d.get("enabled", False)), + enabled=d.get("enabled", False), host=d.get("host", "127.0.0.1"), - port=int(d.get("port", 11434)), + port=d.get("port", 11434), ) @@ -1084,11 +1213,12 @@ class McpEndpointConfig: include_hoa: bool = False # Expose the deprecated hoa_* stub tools to external clients @classmethod + @_coerced def from_dict(cls, d: dict) -> McpEndpointConfig: return cls( - enabled=bool(d.get("enabled", False)), + enabled=d.get("enabled", False), path=str(d.get("path", "/mcp/v1")), - include_hoa=bool(d.get("include_hoa", False)), + include_hoa=d.get("include_hoa", False), ) @@ -1111,10 +1241,11 @@ class ExternalAgentTargetConfig: token: str = "" # deprecated; never persisted @classmethod + @_coerced def from_dict(cls, d: dict) -> ExternalAgentTargetConfig: return cls( name=str(d.get("name", "")), - enabled=bool(d.get("enabled", True)), + enabled=d.get("enabled", True), token="", ) @@ -1145,6 +1276,7 @@ class ExternalAgentsConfig: targets: list[ExternalAgentTargetConfig] = field(default_factory=list) @classmethod + @_coerced def from_dict(cls, d: dict) -> ExternalAgentsConfig: raw_targets = d.get("targets", []) targets: list[ExternalAgentTargetConfig] = [] @@ -1153,23 +1285,13 @@ def from_dict(cls, d: dict) -> ExternalAgentsConfig: if isinstance(raw, dict) and raw.get("name"): targets.append(ExternalAgentTargetConfig.from_dict(raw)) return cls( - enabled=bool(d.get("enabled", True)), - sync_interval_minutes=int(d.get("sync_interval_minutes", 15)), + enabled=d.get("enabled", True), + sync_interval_minutes=d.get("sync_interval_minutes", 15), conflict_policy=str(d.get("conflict_policy", "backup")), targets=targets, ) -def _lenient_int(value: Any, default: int) -> int: - """Best-effort int coercion — malformed inactive config must not - brick startup (validate() reports problems when the section is live).""" - try: - return int(value) if value is not None else default - except (TypeError, ValueError): - logger.warning("Ignoring non-integer config value %r", value) - return default - - _CODEX_APPROVAL_POLICIES = ("never", "on-request", "untrusted") _CODEX_SANDBOX_MODES = ("read-only", "workspace-write", "danger-full-access") @@ -1204,16 +1326,17 @@ class UltracodeConfig: max_agents: int = 8 @classmethod + @_coerced def from_dict(cls, raw: dict | None) -> "UltracodeConfig": d = raw or {} return cls( - enabled=bool(d.get("enabled", False)), - auto_install=bool(d.get("auto_install", True)), + enabled=d.get("enabled", False), + auto_install=d.get("auto_install", True), repository=str(d.get("repository") or cls.repository), revision=str(d.get("revision") or cls.revision), version=str(d.get("version") or cls.version), - dashboard=bool(d.get("dashboard", False)), - ui=bool(d.get("ui", False)), + dashboard=d.get("dashboard", False), + ui=d.get("ui", False), default_transport=str(d.get("default_transport") or "exec"), max_concurrency=_lenient_int(d.get("max_concurrency"), 2), default_token_budget=_lenient_int( @@ -1276,6 +1399,7 @@ class CodexConfig: ultracode: UltracodeConfig = field(default_factory=UltracodeConfig) @classmethod + @_coerced def from_dict(cls, d: dict) -> "CodexConfig": pricing = {k: dict(v) for k, v in _DEFAULT_CODEX_PRICING.items()} raw_pricing = d.get("pricing") or {} @@ -1316,7 +1440,7 @@ def from_dict(cls, d: dict) -> "CodexConfig": sandbox=str(d.get("sandbox", "danger-full-access")), approval_policy=str(d.get("approval_policy", "never")), effort_map=effort_map, - web_search=bool(d.get("web_search", True)), + web_search=d.get("web_search", True), tool_timeout_sec=_lenient_int(d.get("tool_timeout_sec"), 3600), turn_idle_timeout_seconds=_lenient_int( d.get("turn_idle_timeout_seconds"), 0, @@ -1369,6 +1493,7 @@ class McpServerConfig: headers: dict[str, str] = field(default_factory=dict) @classmethod + @_coerced def from_dict(cls, name: str, d: dict) -> McpServerConfig: return cls( name=name, @@ -1530,13 +1655,19 @@ class LangfuseConfig: ) @classmethod + @_coerced def from_dict(cls, d: dict) -> "LangfuseConfig": + # A bare ``redact_patterns:`` parses to None, which the shared list + # handling reads as the empty list. That is the right answer for most + # fields and the wrong one here, because it would turn secret redaction + # off on a blank line. An explicit ``[]`` still means off. + patterns = d.get("redact_patterns", _DEFAULT_LANGFUSE_REDACT_PATTERNS) return cls( public_key=d.get("public_key", ""), secret_key=d.get("secret_key", ""), host=d.get("host", "https://cloud.langfuse.com"), - redact_patterns=list( - d.get("redact_patterns", _DEFAULT_LANGFUSE_REDACT_PATTERNS), + redact_patterns=_str_list( + _DEFAULT_LANGFUSE_REDACT_PATTERNS if patterns is None else patterns ), ) @@ -1567,6 +1698,7 @@ def enabled(self) -> bool: return bool(self.api_key and self.instance_id) @classmethod + @_coerced def from_dict(cls, d: dict) -> "XmemoryConfig": return cls( api_key=d.get("api_key", ""), @@ -1574,7 +1706,7 @@ def from_dict(cls, d: dict) -> "XmemoryConfig": api_url=d.get("api_url", "https://api.xmemory.ai"), extraction_logic=d.get("extraction_logic", "deep"), read_mode=d.get("read_mode", "single-answer"), - timeout=float(d.get("timeout", 60.0)), + timeout=d.get("timeout", 60.0), ) @@ -1773,6 +1905,7 @@ def _validate_backend_config(self) -> None: ) @classmethod + @_coerced def from_dict(cls, d: dict) -> NerveConfig: config = cls._build_from_dict(d) config._validate_backend_config() @@ -1842,6 +1975,7 @@ def load_mcp_servers(config_dir: Path | None = None) -> list[McpServerConfig]: local = yaml.safe_load(f) or {} merged = _deep_merge(base, local) + merged = _resolve_env_refs(merged) return _parse_mcp_servers(merged) @@ -1945,6 +2079,10 @@ def load_config(config_dir: Path | None = None) -> NerveConfig: merged = _deep_merge(base, local) + # Resolve ${ENV_VAR} references before typing the config so secrets can be + # supplied from the environment rather than committed to tracked files. + merged = _resolve_env_refs(merged) + # Surface typos and stale keys instead of silently ignoring them. for warning in validate_config_keys(merged): logger.warning("config: %s", warning) diff --git a/tests/test_config_env.py b/tests/test_config_env.py new file mode 100644 index 00000000..8a051364 --- /dev/null +++ b/tests/test_config_env.py @@ -0,0 +1,589 @@ +"""Tests for ${ENV_VAR} interpolation in config loading.""" + +import pytest + +from nerve.config import ConfigError, _resolve_env_refs, load_config + + +class TestResolveEnvRefs: + def test_required_ref_resolves_from_env(self, monkeypatch): + monkeypatch.setenv("NERVE_X", "resolved") + assert _resolve_env_refs({"k": "${NERVE_X}"}) == {"k": "resolved"} + + def test_ref_embedded_in_larger_string(self, monkeypatch): + monkeypatch.setenv("HOST", "db.internal") + out = _resolve_env_refs({"url": "https://${HOST}:8443/x"}) + assert out == {"url": "https://db.internal:8443/x"} + + def test_default_used_when_unset(self, monkeypatch): + monkeypatch.delenv("NERVE_Y", raising=False) + assert _resolve_env_refs({"k": "${NERVE_Y:-fallback}"}) == {"k": "fallback"} + + def test_default_used_when_empty(self, monkeypatch): + monkeypatch.setenv("NERVE_Y", "") + assert _resolve_env_refs({"k": "${NERVE_Y:-fallback}"}) == {"k": "fallback"} + + def test_env_wins_over_default(self, monkeypatch): + monkeypatch.setenv("NERVE_Y", "actual") + assert _resolve_env_refs({"k": "${NERVE_Y:-fallback}"}) == {"k": "actual"} + + def test_missing_required_raises_listing_var(self, monkeypatch): + monkeypatch.delenv("NERVE_MISSING", raising=False) + with pytest.raises(ConfigError) as ei: + _resolve_env_refs({"k": "${NERVE_MISSING}"}) + assert "NERVE_MISSING" in str(ei.value) + + def test_multiple_missing_all_listed(self, monkeypatch): + monkeypatch.delenv("MISS_A", raising=False) + monkeypatch.delenv("MISS_B", raising=False) + with pytest.raises(ConfigError) as ei: + _resolve_env_refs({"a": "${MISS_A}", "b": {"c": "${MISS_B}"}}) + msg = str(ei.value) + assert "MISS_A" in msg and "MISS_B" in msg + + def test_recurses_dicts_and_lists(self, monkeypatch): + monkeypatch.setenv("V", "x") + out = _resolve_env_refs({"a": ["${V}", {"b": "${V}"}], "n": 5, "flag": True}) + assert out == {"a": ["x", {"b": "x"}], "n": 5, "flag": True} + + def test_bare_dollar_values_untouched(self): + """Critical: bcrypt hashes / jwt secrets contain $ but no braces.""" + bcrypt = "$2b$12$abcdefghijklmnopqrstuv" + out = _resolve_env_refs({"password_hash": bcrypt, "s": "cost=$5"}) + assert out == {"password_hash": bcrypt, "s": "cost=$5"} + + def test_double_dollar_escapes_to_literal(self): + assert _resolve_env_refs({"k": "$${NOT_A_VAR}"}) == {"k": "${NOT_A_VAR}"} + + def test_non_string_leaves_pass_through(self): + assert _resolve_env_refs({"n": 1, "b": False, "z": None}) == { + "n": 1, "b": False, "z": None, + } + + def test_adjacent_refs(self, monkeypatch): + monkeypatch.setenv("A", "x") + monkeypatch.setenv("B", "y") + assert _resolve_env_refs({"k": "${A}${B}"}) == {"k": "xy"} + + def test_required_empty_value_accepted(self, monkeypatch): + """${VAR} (required) checks only *unset*; VAR="" yields "" (POSIX).""" + monkeypatch.setenv("EMPTY_VAR", "") + assert _resolve_env_refs({"k": "${EMPTY_VAR}"}) == {"k": ""} + + def test_falsy_looking_values_kept_not_defaulted(self, monkeypatch): + """Non-empty strings like "0"/"false"/" " are truthy → kept, not + replaced by the :- default.""" + for val in ("0", "false", " "): + monkeypatch.setenv("FLAG", val) + assert _resolve_env_refs({"k": "${FLAG:-D}"}) == {"k": val} + + def test_empty_name_left_intact(self): + assert _resolve_env_refs({"k": "a${}b"}) == {"k": "a${}b"} + + +class TestLoadMcpServersInterpolation: + def test_mcp_server_env_ref_resolved(self, tmp_path, monkeypatch): + from nerve.config import load_mcp_servers + + monkeypatch.setenv("MCP_TOKEN", "sk-mcp-secret") + (tmp_path / "config.yaml").write_text( + "mcp_servers:\n" + " demo:\n" + " type: http\n" + " url: https://example.com/mcp\n" + " headers:\n" + " Authorization: Bearer ${MCP_TOKEN}\n", + encoding="utf-8", + ) + servers = load_mcp_servers(tmp_path) + demo = next(s for s in servers if s.name == "demo") + # The secret was interpolated from the environment, not left as ${...}. + assert "sk-mcp-secret" in str(demo.headers) + assert "${MCP_TOKEN}" not in str(demo.headers) + + def test_mcp_missing_required_var_raises(self, tmp_path, monkeypatch): + from nerve.config import load_mcp_servers + + monkeypatch.delenv("MCP_NOPE", raising=False) + (tmp_path / "config.yaml").write_text( + "mcp_servers:\n" + " demo:\n" + " type: http\n" + " url: https://example.com/mcp\n" + " headers:\n" + " Authorization: Bearer ${MCP_NOPE}\n", + encoding="utf-8", + ) + with pytest.raises(ConfigError): + load_mcp_servers(tmp_path) + + +class TestCleanCliError: + def test_missing_var_renders_clean_error_not_traceback(self, tmp_path, monkeypatch): + from click.testing import CliRunner + + from nerve.cli import main + + monkeypatch.delenv("NOPE_CLI", raising=False) + (tmp_path / "config.yaml").write_text( + "anthropic_api_key: ${NOPE_CLI}\n", encoding="utf-8" + ) + result = CliRunner().invoke(main, ["-c", str(tmp_path), "doctor"]) + assert result.exit_code != 0 + assert "NOPE_CLI" in result.output + assert "Traceback" not in result.output + + +class TestLoadConfigInterpolation: + def test_end_to_end_from_env(self, tmp_path, monkeypatch): + monkeypatch.setenv("MY_ANTHROPIC", "sk-from-env") + (tmp_path / "config.yaml").write_text( + "anthropic_api_key: ${MY_ANTHROPIC}\n", encoding="utf-8" + ) + cfg = load_config(tmp_path) + assert cfg.anthropic_api_key == "sk-from-env" + + def test_missing_required_var_fails_load(self, tmp_path, monkeypatch): + monkeypatch.delenv("NOPE_KEY", raising=False) + (tmp_path / "config.yaml").write_text( + "anthropic_api_key: ${NOPE_KEY}\n", encoding="utf-8" + ) + with pytest.raises(ConfigError): + load_config(tmp_path) + + def test_default_in_config_file(self, tmp_path, monkeypatch): + # BIND_HOST is a plausible name to already have exported in a dev shell + # or CI job; without the delenv this passes on whatever value is lying + # around and never exercises the `:-` fallback it claims to. + monkeypatch.delenv("BIND_HOST", raising=False) + (tmp_path / "config.yaml").write_text( + 'gateway:\n host: "${BIND_HOST:-127.0.0.1}"\n', encoding="utf-8" + ) + cfg = load_config(tmp_path) + assert cfg.gateway.host == "127.0.0.1" + + def test_env_wins_over_default_in_config_file(self, tmp_path, monkeypatch): + """The other half of the pair: a set variable must beat the fallback. + + On its own, `test_default_in_config_file` also passes if `:-` handling + collapses to "always take the default", which would ignore every value + an operator actually exported. + """ + monkeypatch.setenv("BIND_HOST", "10.0.0.5") + (tmp_path / "config.yaml").write_text( + 'gateway:\n host: "${BIND_HOST:-127.0.0.1}"\n', encoding="utf-8" + ) + cfg = load_config(tmp_path) + assert cfg.gateway.host == "10.0.0.5" + + def test_local_overlay_secret_from_env(self, tmp_path, monkeypatch): + """config.local.yaml can reference env too, merged on top of base.""" + monkeypatch.setenv("OAI", "sk-openai-env") + (tmp_path / "config.yaml").write_text( + "openai_api_key: placeholder\n", encoding="utf-8" + ) + (tmp_path / "config.local.yaml").write_text( + "openai_api_key: ${OAI}\n", encoding="utf-8" + ) + cfg = load_config(tmp_path) + assert cfg.openai_api_key == "sk-openai-env" + + +class TestScalarCoercion: + """`${VAR}` interpolation is a string substitution — it erases YAML types. + + `bool("false")` is True and an int field left as `"8900"` raises at first + use, so without coercion an env-referenced flag silently means the opposite + of what it says. This matters more under lockdown, where the machine-local + layers are dropped and every per-machine value *has* to be an env ref. + """ + + def test_as_bool_spellings(self): + from nerve.coerce import as_bool + + for text in ("false", "False", " FALSE ", "0", "no", "off", "n", "f"): + assert as_bool(text, True) is False, text + for text in ("true", "TRUE", "1", "yes", "on", "y", "t"): + assert as_bool(text, False) is True, text + + def test_as_bool_passes_through_real_bools(self): + from nerve.coerce import as_bool + + assert as_bool(True, False) is True + assert as_bool(False, True) is False + + def test_as_bool_falls_back_rather_than_enabling(self): + """An unparseable value must never turn a feature on by accident.""" + from nerve.coerce import as_bool + + assert as_bool("maybe", False) is False + assert as_bool("maybe", True) is True + assert as_bool([], False) is False + assert as_bool([], True) is True + + def test_as_bool_null_and_empty_are_off(self): + """Not "fall back to the default" — these were falsy before coercion, + and `FLAG=` must stay a working way to switch something off.""" + from nerve.coerce import as_bool + + assert as_bool(None, True) is False + assert as_bool("", True) is False + assert as_bool(" ", True) is False + + def test_as_bool_numeric(self): + from nerve.coerce import as_bool + + assert as_bool(1, False) is True + assert as_bool(0, True) is False + + def test_lenient_coercers_fall_back_on_junk(self): + from nerve.coerce import lenient_float, lenient_int + + assert lenient_int("8900", 1) == 8900 + assert lenient_int("junk", 7) == 7 + assert lenient_int(None, 7) == 7 + assert lenient_float("1.5", 0.0) == 1.5 + assert lenient_float("junk", 2.5) == 2.5 + + def test_env_ref_on_bool_field_end_to_end(self, tmp_path, monkeypatch): + """The headline case: a flag disabled via env must actually be off.""" + monkeypatch.setenv("MCP_ON", "false") + (tmp_path / "config.yaml").write_text( + "mcp_endpoint:\n enabled: ${MCP_ON}\n", encoding="utf-8" + ) + assert load_config(tmp_path).mcp_endpoint.enabled is False + + def test_env_ref_on_int_field_end_to_end(self, tmp_path, monkeypatch): + """`port: ${PORT}` is the literal example in docs/config.md.""" + monkeypatch.setenv("PORT", "9001") + (tmp_path / "config.yaml").write_text( + "gateway:\n port: ${PORT}\n", encoding="utf-8" + ) + port = load_config(tmp_path).gateway.port + assert port == 9001 and isinstance(port, int) + + def test_quoted_yaml_bool_without_any_env_var(self, tmp_path): + """Reachable with no interpolation at all — just a quoted scalar.""" + (tmp_path / "config.yaml").write_text( + 'retention:\n enabled: "false"\n', encoding="utf-8" + ) + assert load_config(tmp_path).retention.enabled is False + + def test_null_and_empty_are_off_not_default(self, tmp_path): + """A bare `key:` and an empty ${VAR} must mean off, not "keep default". + + Both were falsy before coercion existed (`bool(None)`, `bool("")`). + Returning the field's default instead would silently *enable* the ten + bool fields whose default is True — including + agent.background_agent_permissions, which grants a catch-all + permission hook. + """ + (tmp_path / "config.yaml").write_text( + "agent:\n" + " background_agent_permissions:\n" + "backup:\n" + " include_workspace:\n" + " notify_on_failure:\n", + encoding="utf-8", + ) + cfg = load_config(tmp_path) + assert cfg.agent.background_agent_permissions is False + assert cfg.backup.include_workspace is False + assert cfg.backup.notify_on_failure is False + + def test_empty_env_var_disables(self, tmp_path, monkeypatch): + """`FLAG=` is the natural "off" spelling when the value must be a ref.""" + monkeypatch.setenv("FLAG", "") + (tmp_path / "config.yaml").write_text( + "backup:\n include_workspace: ${FLAG}\n", encoding="utf-8" + ) + assert load_config(tmp_path).backup.include_workspace is False + + def test_optional_int_field_is_coerced(self, tmp_path, monkeypatch): + """`int | None` counts too — None stays None, a string becomes int.""" + monkeypatch.setenv("TGC", "-1001") + (tmp_path / "config.yaml").write_text( + "notifications:\n telegram_chat_id: ${TGC}\n", encoding="utf-8" + ) + assert load_config(tmp_path).notifications.telegram_chat_id == -1001 + + def test_list_of_int_elements_are_coerced(self, tmp_path, monkeypatch): + """sources/telegram.py compares against these with `in` — a str entry + never matches, so the chat you meant to exclude gets synced anyway.""" + monkeypatch.setenv("CHAT", "-1001") + (tmp_path / "config.yaml").write_text( + 'sync:\n telegram:\n exclude_chats: ["${CHAT}", 42]\n', + encoding="utf-8", + ) + assert load_config(tmp_path).sync.telegram.exclude_chats == [-1001, 42] + + def test_unconvertible_list_element_is_kept_not_dropped(self, tmp_path): + """Silently shrinking an allowlist/denylist is worse than a loud type.""" + (tmp_path / "config.yaml").write_text( + 'sync:\n telegram:\n exclude_chats: [7, "nope"]\n', encoding="utf-8" + ) + assert load_config(tmp_path).sync.telegram.exclude_chats == [7, "nope"] + + def test_bad_allowed_user_does_not_stop_the_daemon_booting( + self, tmp_path, monkeypatch, + ): + """An unresolvable ref in an allowlist must degrade, not crash. + + telegram may not even be enabled on this host, and a config that + refuses to load takes down every other subsystem with it. + """ + monkeypatch.setenv("TG_USERS", "not-a-number") + (tmp_path / "config.yaml").write_text( + 'telegram:\n allowed_users: ["${TG_USERS}", "8900"]\n', + encoding="utf-8", + ) + cfg = load_config(tmp_path) + # Kept verbatim so it stands out, and 8900 still resolves. The set() + # the Telegram channel builds from this holds a str that no int user + # id can match, so the bad entry fails closed. + assert cfg.telegram.allowed_users == ["not-a-number", 8900] + + def test_a_bare_string_allowlist_is_not_read_character_by_character( + self, tmp_path, + ): + """`allowed_users: "123"` must not authorize users 1, 2 and 3.""" + (tmp_path / "config.yaml").write_text( + 'telegram:\n allowed_users: "123"\n', encoding="utf-8" + ) + users = load_config(tmp_path).telegram.allowed_users + assert users != [1, 2, 3] + # Widened to one entry rather than left as a str: leaving the string + # alone only moves the character-splitting to the consumers, which all + # call set() on this. + assert users == [123] + + def test_comma_separated_env_ref_becomes_a_list(self, tmp_path, monkeypatch): + """The only way to spell a multi-value list in a single env var. + + Lockdown discards the machine-local layers, so a per-machine value like + an allowlist *has* to be a ${VAR} reference in the tracked settings — + and interpolation hands the builder one flat string. + """ + monkeypatch.setenv("TG_USERS", " 123 , 456 ") + (tmp_path / "config.yaml").write_text( + "telegram:\n allowed_users: ${TG_USERS}\n", encoding="utf-8" + ) + assert load_config(tmp_path).telegram.allowed_users == [123, 456] + + def test_blank_list_ref_is_empty_not_one_empty_entry(self, tmp_path, monkeypatch): + """An empty allowlist is what re-arms Telegram pairing mode. + + A single blank entry would leave the list truthy, and + _is_authorized's `if not self._allowed_users` gate would never fire. + """ + monkeypatch.setenv("TG_USERS", "") + (tmp_path / "config.yaml").write_text( + "telegram:\n allowed_users: ${TG_USERS}\n", encoding="utf-8" + ) + assert load_config(tmp_path).telegram.allowed_users == [] + + def test_lone_scalar_on_a_list_field_is_wrapped(self, tmp_path): + """`exclude_chats: -1001` is an easy thing to write and mean.""" + (tmp_path / "config.yaml").write_text( + "sync:\n telegram:\n exclude_chats: -1001\n", encoding="utf-8" + ) + assert load_config(tmp_path).sync.telegram.exclude_chats == [-1001] + + def test_string_exclude_chats_actually_excludes(self, tmp_path, monkeypatch): + """The fail-*open* half, reproduced the way the source consumes it. + + sources/telegram.py builds `set(exclude_chats)` and tests an int chat + id for membership. Against the raw string "-1001" that set is a bag of + single characters, nothing matches, and the excluded chat is synced. + """ + monkeypatch.setenv("SKIP", "-1001,-1002") + (tmp_path / "config.yaml").write_text( + "sync:\n telegram:\n exclude_chats: ${SKIP}\n", encoding="utf-8" + ) + excluded = set(load_config(tmp_path).sync.telegram.exclude_chats) + assert -1001 in excluded + assert -1002 in excluded + + def test_non_scalar_on_a_list_field_is_left_alone(self, tmp_path): + """A mapping isn't a list spelled oddly — don't invent a reading.""" + (tmp_path / "config.yaml").write_text( + "telegram:\n allowed_users:\n nope: 1\n", encoding="utf-8" + ) + assert load_config(tmp_path).telegram.allowed_users == {"nope": 1} + + def test_every_declared_scalar_survives_a_string_value(self): + """Sweep the whole dataclass tree so a new field can't reintroduce this. + + A builder that eagerly casts (`bool(d.get("enabled", True))`) defeats + @coerced, because bool("false") is already True by the time the + decorator sees it. This catches that, and catches a new config + dataclass whose from_dict forgets the decorator. + + Scope, so the green tick isn't read as more than it is: only + dataclasses reachable as attributes of `nerve.config` with a callable + `from_dict`. Values read straight off the merged dict (e.g. the + top-level `lockdown` flag) and dataclasses in other modules that + nerve.config doesn't re-export are NOT covered here. + """ + import dataclasses + import inspect + import typing + + import nerve.config as cfg + from nerve.coerce import _classify + + probes = { + bool: [("false", False), ("true", True), ("0", False), ("on", True)], + int: [("8900", 8900)], + float: [("1.5", 1.5)], + # Only ever reached as list[str] — a bare str field needs no + # coercion, so _classify leaves it alone. The value is deliberately + # several characters long and contains a comma, because both of the + # ways a string can be mistaken for a collection (iterating it, + # splitting it) turn this into more than one entry. + str: [("alpha,beta", "alpha,beta")], + } + two_arg = {"McpServerConfig"} # from_dict(cls, name, d) + broken, checked = [], 0 + for _name, klass in sorted(vars(cfg).items()): + if not (inspect.isclass(klass) and dataclasses.is_dataclass(klass)): + continue + from_dict = getattr(klass, "from_dict", None) + if not callable(from_dict): + continue + hints = typing.get_type_hints(klass) + for f in dataclasses.fields(klass): + classified = _classify(hints.get(f.name)) + if classified is None: + continue + kind, base = classified + checked += 1 + for raw, want in probes[base]: + value = [raw] if kind == "list" else raw + expected = [want] if kind == "list" else want + args = ("probe", {f.name: value}) if klass.__name__ in two_arg \ + else ({f.name: value},) + got = getattr(from_dict(*args), f.name) + if got != expected or ( + kind != "list" and type(got) is not base + ): + broken.append( + f"{klass.__name__}.{f.name} ({kind} {base.__name__}): " + f"{value!r} -> {type(got).__name__}={got!r}" + ) + # A ${VAR} reference on a list field arrives as a bare + # scalar, not a one-element list, and it has to become + # exactly one element. This is the probe that separates a + # wrap from a widen: `list("alpha,beta")` is ten entries and + # a comma split is two, and once either has happened nothing + # downstream can tell the result from a genuine list. + if kind == "list": + args = ("probe", {f.name: raw}) if klass.__name__ in two_arg \ + else ({f.name: raw},) + got = getattr(from_dict(*args), f.name) + if got != [want]: + broken.append( + f"{klass.__name__}.{f.name} ({kind} " + f"{base.__name__}): bare {raw!r} -> {got!r}, " + f"expected exactly one element" + ) + # Every probe above happens to convert, so a builder that casts + # eagerly still looks fine on a list. An unresolvable ref is the + # case that separates them: the decorator logs and keeps it, an + # eager cast raises out of load_config. + if kind == "list" and base is not str: + args = ("probe", {f.name: ["not-a-number"]}) \ + if klass.__name__ in two_arg else ({f.name: ["not-a-number"]},) + try: + got = getattr(from_dict(*args), f.name) + except Exception as e: # noqa: BLE001 — that IS the failure + broken.append( + f"{klass.__name__}.{f.name} ({kind} {base.__name__}): " + f"unconvertible entry raised {type(e).__name__}: {e}" + ) + else: + if got != ["not-a-number"]: + broken.append( + f"{klass.__name__}.{f.name} ({kind} " + f"{base.__name__}): unconvertible entry became " + f"{got!r} instead of being kept verbatim" + ) + # A bare `key:` in YAML must not be read as "use the default". + if kind == "scalar" and base is bool: + args = ("probe", {f.name: None}) if klass.__name__ in two_arg \ + else ({f.name: None},) + got = getattr(from_dict(*args), f.name) + if got is not False: + broken.append(f"{klass.__name__}.{f.name}: null -> {got!r}") + assert checked > 60, f"probe only reached {checked} fields — did it break?" + assert not broken, "fields that ignore their declared type:\n" + "\n".join(broken) + + def test_every_declared_list_field_wraps_a_bare_scalar(self): + """Walk the declared annotations, not the coercion's own idea of them. + + The sweep above asks `_classify` which fields it handles, so a type the + coercion silently ignores is invisible to it — a field it skips is a + field that test never probes. That is exactly how `list[str]` stayed + broken while the sweep stayed green. This one derives its field set from + `typing.get_type_hints`, so a `list[X]` the coercion does not cover fails + here instead of disappearing. + + The rule: a `${VAR}` reference on a list field interpolates to a bare + string and must arrive as exactly one element. Both ways of getting that + wrong — widening it (`list("a@b.com")` is nineteen entries) and splitting + it (`redact_patterns` defaults contain `{20,}`) — produce a list nothing + downstream can distinguish from a genuine one. + """ + import dataclasses + import inspect + import typing + + import nerve.config as cfg + + bare = { + str: ("alpha,beta", ["alpha,beta"]), + int: ("8900", [8900]), + float: ("1.5", [1.5]), + bool: ("true", [True]), + } + two_arg = {"McpServerConfig"} # from_dict(cls, name, d) + broken, checked, skipped = [], 0, [] + for _name, klass in sorted(vars(cfg).items()): + if not (inspect.isclass(klass) and dataclasses.is_dataclass(klass)): + continue + from_dict = getattr(klass, "from_dict", None) + if not callable(from_dict): + continue + hints = typing.get_type_hints(klass) + for f in dataclasses.fields(klass): + declared = hints.get(f.name) + if typing.get_origin(declared) is not list: + continue + args = typing.get_args(declared) + element = args[0] if args else None + if element not in bare: + skipped.append((f"{klass.__name__}.{f.name}", element)) + continue + raw, expected = bare[element] + checked += 1 + call = ("probe", {f.name: raw}) if klass.__name__ in two_arg \ + else ({f.name: raw},) + got = getattr(from_dict(*call), f.name) + if got != expected: + broken.append( + f"{klass.__name__}.{f.name} (list[{element.__name__}]): " + f"bare {raw!r} -> {got!r}, expected {expected!r}" + ) + assert checked >= 19, f"probe reached only {checked} list fields — did it break?" + assert not broken, ( + "list fields that do not wrap a bare scalar as a single element:\n" + + "\n".join(broken) + ) + # The skip bucket may only ever hold lists of dataclasses, where a bare + # string is not shorthand for a mapping and the builder fails loudly on + # its own. A new `list[]` must not land here unprobed. + for name, element in skipped: + assert dataclasses.is_dataclass(element), ( + f"{name} is list[{element}], which has no probe above — add one " + "rather than leaving the field unchecked" + )