config.py

"""Configuration loader for the hermes-node plugin.

Precedence (highest to lowest):

  1. Environment variables (``HERMES_NODES_*``).
  2. YAML config file at ``~/.hermes/hermes-node.yaml`` (path overridable
     via ``load_config(config_path=...)``).
  3. Built-in defaults baked into :class:`NodeServerConfig`.

Keys (file format matches env-var names, minus the ``HERMES_NODES_``
prefix and lowercased):

  * ``host``                      (str)   default ``"127.0.0.1"``
  * ``connect_host``               (str)   resolved at load time — the
    loopback address that is actually reachable on this machine.
    The loader probes ``host`` first, then ``"localhost"`` as fallback,
    and stores the result here. HTTP clients (tools, CLI) always use
    this value rather than ``host`` to avoid connecting to an
    unreachable loopback address. Stored under the env-var
    ``HERMES_NODES_CONNECT_HOST`` so operators can override it if needed.
  * ``port``                      (int)   default ``7000``
  * ``tls_cert_path``             (str|None) default ``None``
  * ``tls_key_path``              (str|None) default ``None``
  * ``token_store_path``          (str)   default ``~/.hermes/nodes/tokens.json``
  * ``token_encryption_key_env``  (str)   default ``"HERMES_NODES_TOKEN_KEY"``
    — the *name* of the env var that holds the Fernet key, not the key itself.
  * ``audit_log_path``            (str)   default ``~/.hermes/logs/nodes-audit.log``
    — path to the append-only JSONL audit log (FR-5.1). The audit
    module also accepts ``HERMES_NODES_AUDIT_LOG_PATH`` as a
    direct override; see :mod:`hermes_nodes_plugin.audit`.
  * ``audit_retention_days``      (int)   default ``365``
    — how long rotated audit files are kept before being purged
    (FR-5.4). The audit module also accepts
    ``HERMES_NODES_AUDIT_RETENTION_DAYS`` as a direct override.
  * ``handshake_timeout_seconds`` (float) default ``10.0``
    — max time the server waits for the ``hello`` and ``auth``
    messages on the inbound handshake. A parked WSS is a trivial
    DoS (one coroutine + FD per open socket); bounding the wait
    caps the resource cost. On timeout the server sends a
    structured ``hello_err`` / ``auth_err`` (reason
    ``handshake_timeout``) and closes with 4004. Issue #13.

  * ``heartbeat_stale_seconds``   (int)   default ``60``
    — a node whose last inbound message is older than this is
    considered dead (PROTOCOL §6) and the runner's background
    sweep will close its WebSocket. Override with
    ``HERMES_NODES_HEARTBEAT_STALE_SECONDS``.
  * ``heartbeat_sweep_interval_seconds`` (int)   default ``30``
    — how often the runner's background sweep checks the
    registry for stale connections. Must be ``> 0``. Override
    with ``HERMES_NODES_HEARTBEAT_SWEEP_INTERVAL_SECONDS``.
  * ``rate_limit_per_node`` (int)   default ``100``
    — FR-2.6: max calls/second per node the WSS server will
    accept from a single connected node before closing with
    code 4004 ("Rate limit exceeded" in PROTOCOL §4). The
    limiter is a sliding 1-second window keyed on
    ``node_name``; a node that bursts past the cap is dropped
    with a structured ``rate_limit`` error frame. ``<= 0``
    disables the limiter (fail-open; the limiter logs a warning
    at construction so a typo'd ``=0`` is visible). A
    non-integer value is a config error. Override with
    ``HERMES_NODES_RATE_LIMIT`` (the env-var name is the spec
    literal; the dataclass field uses the longer name for
    clarity at call sites).

Type coercion rules (applied uniformly to env and file values):

  * ``port`` is parsed via :func:`int`. A non-integer value raises
    :class:`ConfigError` with the offending key + value.
  * All other keys are strings (paths/hosts) or ``None``.

The loader never writes to disk and never imports optional deps beyond
``pyyaml`` (which is declared as a runtime dep). It is safe to call at
import time of the plugin.
"""

from __future__ import annotations

import os
import socket
from collections.abc import Mapping
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any

import yaml

from .errors import ConfigError

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

DEFAULT_CONFIG_PATH = Path("~/.hermes/hermes-node.yaml").expanduser()
DEFAULT_TOKEN_STORE_PATH = Path("~/.hermes/nodes/tokens.json")
DEFAULT_TOKEN_STORE_STR = "~/.hermes/nodes/tokens.json"

# Env-var prefix. Task spec only named a handful of vars; we map all six
# config keys uniformly to keep precedence rules easy to explain.
_ENV_PREFIX = "HERMES_NODES_"

# ---------------------------------------------------------------------------|
# Connection probing                                                        |
# ---------------------------------------------------------------------------|

def probe_connect_host(candidate: str, port: int, timeout: float = 1.0) -> str | None:
    """Return ``candidate`` if a TCP socket can connect to it on ``port``.

    Returns ``None`` when the connection is refused or times out, indicating
    that this loopback address is not reachable on this machine.
    """
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        sock.connect((candidate, port))
        sock.close()
        return candidate
    except (TimeoutError, OSError):
        return None


def resolve_connect_host(*, host: str, port: int) -> str:
    """Resolve the connectable loopback address for HTTP clients.

    Probing strategy (tried in order, first reachable wins):
      1. ``host`` as configured (usually ``127.0.0.1``)
      2. ``localhost``

    If the env var ``HERMES_NODES_CONNECT_HOST`` is set, it is used directly
    with no probing (operator override — they know what they're doing).
    """
    override = os.environ.get("HERMES_NODES_CONNECT_HOST")
    if override:
        return override

    if probe_connect_host(host, port) is not None:
        return host
    if probe_connect_host("localhost", port) is not None:
        return "localhost"
    # Neither is reachable — fall back to host as last resort; the
    # connection will fail at call time with a clear error.
    return host

# ---------------------------------------------------------------------------|
# Dataclass                                                                 |
# ---------------------------------------------------------------------------|

@dataclass(frozen=True)
class NodeServerConfig:
    """Resolved configuration for the node WSS server.

    Construct via :func:`load_config` rather than directly; the loader is
    the only piece that knows about env-var / file precedence. Direct
    construction is supported for tests and for the rare caller that
    already has a fully-resolved mapping.
    """

    host: str = "127.0.0.1"
    # Resolved at load time — the loopback address that is actually reachable.
    # See probe_connect_host() for the probing logic.
    connect_host: str = "127.0.0.1"
    port: int = 7000
    tls_cert_path: str | None = None
    tls_key_path: str | None = None
    token_store_path: str = DEFAULT_TOKEN_STORE_STR
    # Name of the env var that holds the Fernet key for the token store.
    # Not the key itself — see REQUIREMENTS.md FR-4.1/FR-4.2.
    token_encryption_key_env: str = "HERMES_NODES_TOKEN_KEY"
    # Path to the append-only JSONL audit log (FR-5.1). The audit
    # module also accepts ``HERMES_NODES_AUDIT_LOG_PATH`` as a
    # direct override, which beats this value. The default is
    # duplicated as a literal here to avoid a circular import
    # (``audit`` imports this dataclass); keep in sync with
    # ``hermes_nodes_plugin.audit.DEFAULT_AUDIT_LOG_PATH``.
    audit_log_path: str = "~/.hermes/logs/nodes-audit.log"
    # Retention window for rotated audit files (FR-5.4). The audit
    # module also accepts ``HERMES_NODES_AUDIT_RETENTION_DAYS``.
    # See ``hermes_nodes_plugin.audit.DEFAULT_RETENTION_DAYS`` for
    # the source of truth (kept as a literal here to break the
    # ``audit`` ↔ ``config`` import cycle).
    audit_retention_days: int = 365
    # Handshake read timeout (issue #13). A single bound for both
    # the hello and the auth recv — they are two phases of the same
    # handshake, and splitting them invites an operator to set one
    # tight and the other lax. 10s is generous for a small envelope
    # on any plausible network. ``__post_init__`` enforces > 0.
    handshake_timeout_seconds: float = 10.0

    # PROTOCOL §6: a node is considered dead after this many seconds
    # without any inbound message. The runner's background sweep
    # closes the WebSocket of any node whose ``last_heartbeat`` is
    # older than this threshold. See issue #19.
    heartbeat_stale_seconds: int = 60
    # How often the runner's background sweep runs. Lower means
    # faster cleanup of dead nodes; higher means less registry
    # churn. Must be ``> 0``. See issue #19.
    heartbeat_sweep_interval_seconds: int = 30

    # FR-2.6 per-node sliding-window rate limit. A node whose
    # inbound call count within a 1-second window exceeds this
    # cap gets a ``rate_limit`` error frame and the WSS is closed
    # with 4004. The dispatcher reads this once at app
    # construction; the cap is in-process and resets on server
    # boot. ``<= 0`` disables the limiter (fail-open; the
    # limiter itself logs the disable). The dataclass does NOT
    # raise on ``<= 0`` because ``rate_limit_per_node=0`` is a
    # legitimate "unlimited" operator choice. Env var:
    # ``HERMES_NODES_RATE_LIMIT`` (spec-literal, NOT derived
    # from the field name).
    rate_limit_per_node: int = 100

    # Number of times to retry an exec/read/write call when the
    # target node is not connected or the server is unreachable.
    # Each retry uses exponential backoff (1s, 2s, 4s... capped
    # at max_retry_backoff). Set to 0 to disable retries.
    # Env var: ``HERMES_NODES_MAX_RETRIES``.
    max_retries: int = 3

    # Maximum backoff between retries, in seconds. The actual
    # delay is min(backoff_seconds * 2^attempt, max_retry_backoff).
    # Env var: ``HERMES_NODES_RETRY_BACKOFF_SECONDS``.
    retry_backoff_seconds: float = 2.0

    # -- exec_chunk streaming (issue #116) ---------------------------------
    # Directory where exec_chunk temp files are assembled. Must exist at
    # server start (the lifespan creates it). Env var:
    # ``HERMES_NODES_EXEC_CHUNK_DIR``.
    exec_chunk_dir: str = "/tmp/hermes-exec-chunks"

    # Max total size in MiB of all in-flight exec_chunk temp files.
    # A new chunk that would push the total past this cap is rejected
    # with a ``chunk_size_limit`` error frame. ``<= 0`` means unlimited.
    # Env var: ``HERMES_NODES_EXEC_CHUNK_MAX_TOTAL_MB``.
    exec_chunk_max_total_mb: int = 50

    # Max age of an in-flight chunk (seconds). If a chunk sits unclosed
    # for longer than this, the background sweep removes it and completes
    # the waiter with a timeout error. Must be ``> 0``.
    # Env var: ``HERMES_NODES_EXEC_CHUNK_TIMEOUT_S``.
    exec_chunk_timeout_s: int = 30

    # Max payload size (bytes) of a single exec_chunk message.
    # Must be ``> 0``. Env var: ``HERMES_NODES_EXEC_CHUNK_SIZE``.
    exec_chunk_size: int = 65536

    def __post_init__(self) -> None:
        # TLS partial-config is the most common deployment footgun: an
        # operator sets tls_cert_path but forgets tls_key_path (or vice
        # versa), the server then fails to bind with a confusing ssl
        # error, and nobody knows why. Catch it at load time instead.
        if (self.tls_cert_path is None) != (self.tls_key_path is None):
            raise ConfigError(
                "TLS is partially configured: tls_cert_path and tls_key_path "
                "must both be set or both be unset. "
                f"Got tls_cert_path={self.tls_cert_path!r}, "
                f"tls_key_path={self.tls_key_path!r}."
            )
        if self.audit_retention_days <= 0:
            raise ConfigError(
                f"audit_retention_days must be > 0, got {self.audit_retention_days!r}"
            )
        if self.handshake_timeout_seconds <= 0:
            raise ConfigError(
                f"handshake_timeout_seconds must be > 0, got "
                f"{self.handshake_timeout_seconds!r}"
            )

        if self.heartbeat_stale_seconds <= 0:
            raise ConfigError(
                f"heartbeat_stale_seconds must be > 0, got {self.heartbeat_stale_seconds!r}"
            )
        if self.heartbeat_sweep_interval_seconds <= 0:
            raise ConfigError(
                "heartbeat_sweep_interval_seconds must be > 0, "
                f"got {self.heartbeat_sweep_interval_seconds!r}"
            )

        if self.exec_chunk_timeout_s <= 0:
            raise ConfigError(
                f"exec_chunk_timeout_s must be > 0, got "
                f"{self.exec_chunk_timeout_s!r}"
            )
        if self.exec_chunk_size <= 0:
            raise ConfigError(
                f"exec_chunk_size must be > 0, got "
                f"{self.exec_chunk_size!r}"
            )

    # -- predicates ---------------------------------------------------------

    def uses_tls(self) -> bool:
        """True when the server should listen with TLS (direct mode).

        Reverse-proxied deployments (the default) leave both TLS paths
        unset and the plugin listens on plain HTTP behind nginx/Caddy.
        """
        return self.tls_cert_path is not None and self.tls_key_path is not None

    # -- env helpers --------------------------------------------------------

    def token_encryption_key(self, env: Mapping[str, str] | None = None) -> str | None:
        """Read the Fernet key from the env var named by ``token_encryption_key_env``.

        Returns ``None`` when the var is unset/empty. The caller (token
        store, server startup) is responsible for turning that ``None``
        into the user-facing "refuse to start" error per FR-4.2 — this
        method is intentionally permissive so tests can construct configs
        without a real key.
        """
        src = env if env is not None else os.environ
        return src.get(self.token_encryption_key_env) or None

# ---------------------------------------------------------------------------
# Loader
# ---------------------------------------------------------------------------

def _env_name(key: str) -> str:
    return f"{_ENV_PREFIX}{key.upper()}"

def _coerce_port(value: Any, *, source: str, key: str) -> int:
    try:
        port = int(value)
    except (TypeError, ValueError) as exc:
        raise ConfigError(f"{source}: {key} must be an integer, got {value!r}") from exc
    if not (1 <= port <= 65535):
        raise ConfigError(f"{source}: {key} must be in 1..65535, got {port}")
    return port

def _coerce_optional_str(value: Any) -> str | None:
    """Treat ``None``, empty string, and YAML null as "unset"."""
    if value is None:
        return None
    if isinstance(value, str) and value.strip() == "":
        return None
    return str(value)

def _load_yaml(path: Path) -> dict[str, Any]:
    """Read a YAML config file. Missing file → empty dict (caller falls back to defaults)."""
    if not path.exists():
        return {}
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8"))
    except yaml.YAMLError as exc:
        raise ConfigError(f"failed to parse YAML config at {path}: {exc}") from exc
    if data is None:
        return {}
    if not isinstance(data, dict):
        raise ConfigError(
            f"YAML config at {path} must be a mapping at the top level, "
            f"got {type(data).__name__}"
        )
    return data

def _read_file_value(file_data: dict[str, Any], key: str) -> Any:
    """Look up a key in the YAML file, accepting both lower and upper case.

    File authors are split — some write ``host: 127.0.0.1`` (mirroring
    dataclass field names), others write ``HOST: 127.0.0.1`` (mirroring
    env-var names). We accept both so the operator doesn't have to think
    about it.
    """
    if key in file_data:
        return file_data[key]
    upper = key.upper()
    if upper in file_data:
        return file_data[upper]
    return None

def _resolve_str(
    *,
    key: str,
    env: Mapping[str, str],
    file_data: dict[str, Any],
) -> str | None:
    """Apply env > file precedence for a string field. Returns ``None`` if neither sets it."""
    env_name = _env_name(key)
    if env_name in env:
        return env[env_name]
    return _read_file_value(file_data, key)

def _build(
    *,
    env: Mapping[str, str],
    file_data: dict[str, Any],
) -> NodeServerConfig:
    """Apply precedence (env > file > dataclass default) per key.

    Strategy: for every key, compute ``raw`` (env beats file, else file,
    else dataclass default), then coerce through type-specific helpers.
    Doing it as one pass per key keeps precedence uniform and avoids
    the "env set tls_cert but file set tls_key" asymmetry surprises.
    """
    # -- host (str) ---------------------------------------------------------
    host = _resolve_str(key="host", env=env, file_data=file_data)
    if host is not None:
        host = str(host)

    # -- port (int) ---------------------------------------------------------
    port_raw: Any = None
    port_source = "default"
    if env.get(_env_name("port")) is not None:
        port_raw = env[_env_name("port")]
        port_source = "env"
    elif _read_file_value(file_data, "port") is not None:
        port_raw = _read_file_value(file_data, "port")
        port_source = "file"
    port: int | None = None
    if port_raw is not None:
        port = _coerce_port(port_raw, source=port_source, key="port")

    # -- TLS paths (optional str) ------------------------------------------
    tls_cert = _coerce_optional_str(
        _resolve_str(key="tls_cert_path", env=env, file_data=file_data)
    )
    tls_key = _coerce_optional_str(
        _resolve_str(key="tls_key_path", env=env, file_data=file_data)
    )

    # -- token store path (str) --------------------------------------------
    token_store_raw = _resolve_str(key="token_store_path", env=env, file_data=file_data)

    # -- token encryption key env var name (str) ---------------------------
    key_env_raw = _resolve_str(
        key="token_encryption_key_env", env=env, file_data=file_data
    )

    # -- audit log path (str) ----------------------------------------------
    audit_log_raw = _resolve_str(key="audit_log_path", env=env, file_data=file_data)

    # -- audit retention days (int) ----------------------------------------
    audit_retention_raw: Any = None
    audit_retention_source = "default"
    if env.get("HERMES_NODES_AUDIT_RETENTION_DAYS") is not None:
        audit_retention_raw = env["HERMES_NODES_AUDIT_RETENTION_DAYS"]
        audit_retention_source = "env"
    elif _read_file_value(file_data, "audit_retention_days") is not None:
        audit_retention_raw = _read_file_value(file_data, "audit_retention_days")
        audit_retention_source = "file"
    audit_retention: int | None = None
    if audit_retention_raw is not None:
        try:
            audit_retention = int(audit_retention_raw)
        except (TypeError, ValueError) as exc:
            raise ConfigError(
                f"{audit_retention_source}: audit_retention_days must be an integer, "
                f"got {audit_retention_raw!r}"
            ) from exc
        if audit_retention <= 0:
            raise ConfigError(
                f"{audit_retention_source}: audit_retention_days must be > 0, "
                f"got {audit_retention}"
            )

    # -- handshake timeout (float) -----------------------------------------
    # Issue #13. Bounded waits on the hello + auth recv, applied
    # uniformly through the standard env > file > default chain.
    handshake_timeout_raw: Any = None
    handshake_timeout_source = "default"
    if env.get(_env_name("handshake_timeout_seconds")) is not None:
        handshake_timeout_raw = env[_env_name("handshake_timeout_seconds")]
        handshake_timeout_source = "env"
    elif _read_file_value(file_data, "handshake_timeout_seconds") is not None:
        handshake_timeout_raw = _read_file_value(file_data, "handshake_timeout_seconds")
        handshake_timeout_source = "file"
    handshake_timeout: float | None = None
    if handshake_timeout_raw is not None:
        try:
            handshake_timeout = float(handshake_timeout_raw)
        except (TypeError, ValueError) as exc:
            raise ConfigError(
                f"{handshake_timeout_source}: handshake_timeout_seconds must be a number, "
                f"got {handshake_timeout_raw!r}"
            ) from exc
        if handshake_timeout <= 0:
            raise ConfigError(
                f"{handshake_timeout_source}: handshake_timeout_seconds must be > 0, "
                f"got {handshake_timeout}"
            )

    # -- heartbeat stale seconds (int) -------------------------------------
    # PROTOCOL §6: a node is dead after this many seconds without any
    # inbound message. The runner's background sweep (issue #19) uses
    # this to decide who's stale.
    heartbeat_stale_raw: Any = None
    heartbeat_stale_source = "default"
    if env.get("HERMES_NODES_HEARTBEAT_STALE_SECONDS") is not None:
        heartbeat_stale_raw = env["HERMES_NODES_HEARTBEAT_STALE_SECONDS"]
        heartbeat_stale_source = "env"
    elif _read_file_value(file_data, "heartbeat_stale_seconds") is not None:
        heartbeat_stale_raw = _read_file_value(file_data, "heartbeat_stale_seconds")
        heartbeat_stale_source = "file"
    heartbeat_stale: int | None = None
    if heartbeat_stale_raw is not None:
        try:
            heartbeat_stale = int(heartbeat_stale_raw)
        except (TypeError, ValueError) as exc:
            raise ConfigError(
                f"{heartbeat_stale_source}: heartbeat_stale_seconds must be an integer, "
                f"got {heartbeat_stale_raw!r}"
            ) from exc
        if heartbeat_stale <= 0:
            raise ConfigError(
                f"{heartbeat_stale_source}: heartbeat_stale_seconds must be > 0, "
                f"got {heartbeat_stale}"
            )

    # -- heartbeat sweep interval seconds (int) ----------------------------
    # How often the background sweep runs. Decoupled from the stale
    # threshold so operators can tune cleanup latency independently
    # of the dead-node definition.
    sweep_interval_raw: Any = None
    sweep_interval_source = "default"
    if env.get("HERMES_NODES_HEARTBEAT_SWEEP_INTERVAL_SECONDS") is not None:
        sweep_interval_raw = env["HERMES_NODES_HEARTBEAT_SWEEP_INTERVAL_SECONDS"]
        sweep_interval_source = "env"
    elif _read_file_value(file_data, "heartbeat_sweep_interval_seconds") is not None:
        sweep_interval_raw = _read_file_value(file_data, "heartbeat_sweep_interval_seconds")
        sweep_interval_source = "file"
    sweep_interval: int | None = None
    if sweep_interval_raw is not None:
        try:
            sweep_interval = int(sweep_interval_raw)
        except (TypeError, ValueError) as exc:
            raise ConfigError(
                f"{sweep_interval_source}: heartbeat_sweep_interval_seconds must be an integer, "
                f"got {sweep_interval_raw!r}"
            ) from exc
        if sweep_interval <= 0:
            raise ConfigError(
                f"{sweep_interval_source}: heartbeat_sweep_interval_seconds must be > 0, "
                f"got {sweep_interval}"
            )

    # -- rate limit per node (int, FR-2.6) --------------------------------
    # Spec literal env-var name ``HERMES_NODES_RATE_LIMIT`` (NOT
    # derived from the dataclass field name ``rate_limit_per_node``).
    # ``<= 0`` is a legitimate "unlimited" operator choice and is
    # not a config error — the limiter logs a warning at construction.
    rate_limit_raw: Any = None
    rate_limit_source = "default"
    if env.get("HERMES_NODES_RATE_LIMIT") is not None:
        rate_limit_raw = env["HERMES_NODES_RATE_LIMIT"]
        rate_limit_source = "env"
    elif _read_file_value(file_data, "rate_limit_per_node") is not None:
        rate_limit_raw = _read_file_value(file_data, "rate_limit_per_node")
        rate_limit_source = "file"
    rate_limit: int | None = None
    if rate_limit_raw is not None:
        # Empty string from the env (e.g. ``export HERMES_NODES_RATE_LIMIT=``)
        # falls through to the default. Some shells normalise unset vars
        # this way and the operator clearly didn't intend to set a value.
        if isinstance(rate_limit_raw, str) and rate_limit_raw.strip() == "":
            rate_limit = None
        else:
            try:
                rate_limit = int(rate_limit_raw)
            except (TypeError, ValueError) as exc:
                raise ConfigError(
                    f"{rate_limit_source}: rate_limit_per_node must be an integer, "
                    f"got {rate_limit_raw!r}"
                ) from exc

    # -- max retries (int) ---------------------------------------------------
    max_retries_raw: Any = None
    max_retries_source = "default"
    if env.get(_env_name("max_retries")) is not None:
        max_retries_raw = env[_env_name("max_retries")]
        max_retries_source = "env"
    elif _read_file_value(file_data, "max_retries") is not None:
        max_retries_raw = _read_file_value(file_data, "max_retries")
        max_retries_source = "file"
    max_retries_val: int | None = None
    if max_retries_raw is not None:
        if isinstance(max_retries_raw, str) and max_retries_raw.strip() == "":
            max_retries_val = None
        else:
            try:
                max_retries_val = int(max_retries_raw)
            except (TypeError, ValueError) as exc:
                raise ConfigError(
                    f"{max_retries_source}: max_retries must be an integer, "
                    f"got {max_retries_raw!r}"
                ) from exc
            if max_retries_val < 0:
                raise ConfigError(
                    f"{max_retries_source}: max_retries must be >= 0, "
                    f"got {max_retries_val}"
                )

    # -- retry backoff seconds (float) --------------------------------------
    retry_backoff_raw: Any = None
    retry_backoff_source = "default"
    if env.get(_env_name("retry_backoff_seconds")) is not None:
        retry_backoff_raw = env[_env_name("retry_backoff_seconds")]
        retry_backoff_source = "env"
    elif _read_file_value(file_data, "retry_backoff_seconds") is not None:
        retry_backoff_raw = _read_file_value(file_data, "retry_backoff_seconds")
        retry_backoff_source = "file"
    retry_backoff_val: float | None = None
    if retry_backoff_raw is not None:
        if isinstance(retry_backoff_raw, str) and retry_backoff_raw.strip() == "":
            retry_backoff_val = None
        else:
            try:
                retry_backoff_val = float(retry_backoff_raw)
            except (TypeError, ValueError) as exc:
                raise ConfigError(
                    f"{retry_backoff_source}: retry_backoff_seconds must be a number, "
                    f"got {retry_backoff_raw!r}"
                ) from exc
            if retry_backoff_val <= 0:
                raise ConfigError(
                    f"{retry_backoff_source}: retry_backoff_seconds must be > 0, "
                    f"got {retry_backoff_val}"
                )

    # -- exec_chunk_dir (str) -------------------------------------------------
    exec_chunk_dir_raw = _resolve_str(
        key="exec_chunk_dir", env=env, file_data=file_data
    )

    # -- exec_chunk_max_total_mb (int, <=0 = unlimited) -----------------------
    exec_chunk_max_raw: Any = None
    exec_chunk_max_source = "default"
    if env.get(_env_name("exec_chunk_max_total_mb")) is not None:
        exec_chunk_max_raw = env[_env_name("exec_chunk_max_total_mb")]
        exec_chunk_max_source = "env"
    elif _read_file_value(file_data, "exec_chunk_max_total_mb") is not None:
        exec_chunk_max_raw = _read_file_value(file_data, "exec_chunk_max_total_mb")
        exec_chunk_max_source = "file"
    exec_chunk_max_val: int | None = None
    if exec_chunk_max_raw is not None:
        if isinstance(exec_chunk_max_raw, str) and exec_chunk_max_raw.strip() == "":
            exec_chunk_max_val = None
        else:
            try:
                exec_chunk_max_val = int(exec_chunk_max_raw)
            except (TypeError, ValueError) as exc:
                raise ConfigError(
                    f"{exec_chunk_max_source}: exec_chunk_max_total_mb must be an integer, "
                    f"got {exec_chunk_max_raw!r}"
                ) from exc

    # -- exec_chunk_timeout_s (int, must be >0) ------------------------------
    exec_chunk_timeout_raw: Any = None
    exec_chunk_timeout_source = "default"
    if env.get(_env_name("exec_chunk_timeout_s")) is not None:
        exec_chunk_timeout_raw = env[_env_name("exec_chunk_timeout_s")]
        exec_chunk_timeout_source = "env"
    elif _read_file_value(file_data, "exec_chunk_timeout_s") is not None:
        exec_chunk_timeout_raw = _read_file_value(file_data, "exec_chunk_timeout_s")
        exec_chunk_timeout_source = "file"
    exec_chunk_timeout_val: int | None = None
    if exec_chunk_timeout_raw is not None:
        if isinstance(exec_chunk_timeout_raw, str) and exec_chunk_timeout_raw.strip() == "":
            exec_chunk_timeout_val = None
        else:
            try:
                exec_chunk_timeout_val = int(exec_chunk_timeout_raw)
            except (TypeError, ValueError) as exc:
                raise ConfigError(
                    f"{exec_chunk_timeout_source}: exec_chunk_timeout_s must be an integer, "
                    f"got {exec_chunk_timeout_raw!r}"
                ) from exc
            if exec_chunk_timeout_val <= 0:
                raise ConfigError(
                    f"{exec_chunk_timeout_source}: exec_chunk_timeout_s must be > 0, "
                    f"got {exec_chunk_timeout_val}"
                )

    # -- exec_chunk_size (int, must be >0) ----------------------------------
    exec_chunk_size_raw: Any = None
    exec_chunk_size_source = "default"
    if env.get(_env_name("exec_chunk_size")) is not None:
        exec_chunk_size_raw = env[_env_name("exec_chunk_size")]
        exec_chunk_size_source = "env"
    elif _read_file_value(file_data, "exec_chunk_size") is not None:
        exec_chunk_size_raw = _read_file_value(file_data, "exec_chunk_size")
        exec_chunk_size_source = "file"
    exec_chunk_size_val: int | None = None
    if exec_chunk_size_raw is not None:
        if isinstance(exec_chunk_size_raw, str) and exec_chunk_size_raw.strip() == "":
            exec_chunk_size_val = None
        else:
            try:
                exec_chunk_size_val = int(exec_chunk_size_raw)
            except (TypeError, ValueError) as exc:
                raise ConfigError(
                    f"{exec_chunk_size_source}: exec_chunk_size must be an integer, "
                    f"got {exec_chunk_size_raw!r}"
                ) from exc
            if exec_chunk_size_val <= 0:
                raise ConfigError(
                    f"{exec_chunk_size_source}: exec_chunk_size must be > 0, "
                    f"got {exec_chunk_size_val}"
                )

    # Now assemble. We use a partial dict + NodeServerConfig defaults for
    # any key we didn't resolve — dataclass handles the "default" leg of
    # the precedence chain.
    resolved: dict[str, Any] = {}
    if host is not None:
        resolved["host"] = host
    if port is not None:
        resolved["port"] = port
    if tls_cert is not None:
        resolved["tls_cert_path"] = tls_cert
    if tls_key is not None:
        resolved["tls_key_path"] = tls_key
    if token_store_raw is not None:
        resolved["token_store_path"] = str(token_store_raw)
    if key_env_raw is not None:
        resolved["token_encryption_key_env"] = str(key_env_raw)
    if audit_log_raw is not None:
        resolved["audit_log_path"] = str(audit_log_raw)
    if audit_retention is not None:
        resolved["audit_retention_days"] = audit_retention
    if handshake_timeout is not None:
        resolved["handshake_timeout_seconds"] = handshake_timeout

    if heartbeat_stale is not None:
        resolved["heartbeat_stale_seconds"] = heartbeat_stale
    if sweep_interval is not None:
        resolved["heartbeat_sweep_interval_seconds"] = sweep_interval
    if rate_limit is not None:
        resolved["rate_limit_per_node"] = rate_limit
    if max_retries_val is not None:
        resolved["max_retries"] = max_retries_val
    if retry_backoff_val is not None:
        resolved["retry_backoff_seconds"] = retry_backoff_val
    if exec_chunk_dir_raw is not None:
        resolved["exec_chunk_dir"] = str(exec_chunk_dir_raw)
    if exec_chunk_max_val is not None:
        resolved["exec_chunk_max_total_mb"] = exec_chunk_max_val
    if exec_chunk_timeout_val is not None:
        resolved["exec_chunk_timeout_s"] = exec_chunk_timeout_val
    if exec_chunk_size_val is not None:
        resolved["exec_chunk_size"] = exec_chunk_size_val

    return NodeServerConfig(**resolved)

def load_config(
    *,
    env: Mapping[str, str] | None = None,
    config_path: Path | str | None = None,
) -> NodeServerConfig:
    """Resolve a :class:`NodeServerConfig` from env, file, and defaults.

    Args:
        env: Override the env mapping (defaults to :data:`os.environ`).
            Tests use this to inject fixtures without touching the real
            process environment.
        config_path: Override the YAML file location (defaults to
            ``~/.hermes/hermes-node.yaml``). A missing file is *not* an
            error — the loader falls through to dataclass defaults.

    Returns:
        A fully-resolved, frozen :class:`NodeServerConfig`.

    Raises:
        ConfigError: YAML parse failure, bad port value, partial TLS
            config, or non-mapping top-level YAML.
    """
    if env is None:
        env = os.environ
    if config_path is None:
        path = DEFAULT_CONFIG_PATH
    else:
        path = Path(config_path).expanduser()

    file_data = _load_yaml(path)
    cfg = _build(env=env, file_data=file_data)

    # Probe for a connectable loopback address. This avoids the situation
    # where the server binds to 127.0.0.1 but the client can't reach it
    # (e.g. some container setups where localhost resolves differently).
    connect_host = resolve_connect_host(host=cfg.host, port=cfg.port)
    return replace(cfg, connect_host=connect_host)

# ---------------------------------------------------------------------------
# Convenience re-exports
# ---------------------------------------------------------------------------

__all__ = [
    "DEFAULT_CONFIG_PATH",
    "DEFAULT_TOKEN_STORE_PATH",
    "ConfigError",
    "NodeServerConfig",
    "load_config",
]