hermes-agent/hermes_cli/config_read_errors.py

"""What Hermes does when an existing ``config.yaml`` cannot be read or parsed.

Readers fail open (``{}`` / defaults / last-known-good) so a broken file never takes the process
down; this module makes that fallback loud (one warning per file signature, a ``corrupt`` backup),
remembers the failure for :func:`get_active_config_parse_failure`, and marks every fallback as a
:class:`FailedConfigRead` so the writers refuse to persist it.
"""

from __future__ import annotations

import logging
import sys
from pathlib import Path
from typing import Any, Optional

from utils import file_signature

logger = logging.getLogger(__name__)

# (config_path, mtime_ns, size) tuples already warned about, so concurrent CLI/gateway
# loads of a broken config.yaml don't spam stderr. A changed file (new mtime) warns again.
_CONFIG_PARSE_WARNED: set = set()

# path -> (mtime_ns, size, error message) of active parse failures. Written by
# _warn_config_parse_failure() (the single funnel for every load-path parse failure) and
# probed by get_active_config_parse_failure() so provider auto-resolution can refuse to
# adopt a paid provider from env keys while the user's REAL config is unreadable.
_CONFIG_PARSE_FAILURES: dict = {}

_PARSE_FAILURE_FALLBACK_MSG = {
    "last-known-good": "Hermes is running on the settings it loaded before the edit until it is fixed, so recent changes are not applied.",
    "last-known-good-backup": "Hermes is running on your last good settings until it is fixed, so recent changes are not applied.",
    "refuse-write": "Nothing was written, so the existing file is preserved."}
_PARSE_FAILURE_DEFAULTS_MSG = (
    "Hermes is running on default settings until it is fixed, so none of your saved settings are applied.")
_PARSE_FAILURE_REPAIR_MSG = "Open it with `hermes config edit`, fix {where}, then run `hermes config check`."
_FIX_PERMS = "Fix the file permissions or move it aside first."
_FIX_YAML = (
    "Fix it with `hermes config edit` and check with `hermes config check`, or copy the newest good "
    "file from {backups} over config.yaml.")


def _yaml_error_location(exc: Exception) -> str:
    """``"line 12"`` from a PyYAML problem mark (1-based), else ``""``."""
    mark = getattr(exc, "problem_mark", None) or getattr(exc, "context_mark", None)
    line = getattr(mark, "line", None)
    return f"line {line + 1}" if isinstance(line, int) else ""


def _yaml_error_details(exc: Exception) -> str:
    """Single-line ``Details:`` text: the PyYAML problem, or the exception's first line."""
    problem = getattr(exc, "problem", None)
    text = f"{problem}" if problem else str(exc).strip()
    return " ".join(text.split())


def format_config_parse_failure(config_path: Path, exc: Exception, *, fallback: str = "defaults") -> str:
    """User copy for an unparseable config.yaml: what happened, what Hermes is doing, how to fix.
    Only the problem line/column is printed; the raw PyYAML text goes to a ``Details:`` line."""
    where = _yaml_error_location(exc)
    at = f" at {where}" if where else ""
    fallback_msg = _PARSE_FAILURE_FALLBACK_MSG.get(fallback, _PARSE_FAILURE_DEFAULTS_MSG)
    if isinstance(exc, OSError):  # the file is intact; EMFILE/EIO/sharing violation, not a YAML problem
        return f"Your settings file ({config_path}) could not be read. {fallback_msg} {_read_error_fix(exc)}"
    repair = _PARSE_FAILURE_REPAIR_MSG.format(where=where or "the problem")
    return f"Your settings file ({config_path}) has a formatting error{at}. {fallback_msg} {repair}"


def _warn_config_parse_failure(
    config_path: Path, exc: Exception, *, fallback: str = "defaults") -> None:
    """Surface a config.yaml parse failure to log and stderr (once per file signature).
    Silent fallback to ``DEFAULT_CONFIG`` drops every user override, so this must be loud.

    ``fallback`` selects the message wording: ``"defaults"`` (fresh process, nothing else to serve) or
    ``"last-known-good"`` (in-process retention of the previously loaded config — see the codex#31188 port
    in ``hermes_cli.config._load_config_impl``).
    """
    try:
        st = config_path.stat()
        sig = file_signature(st)
        key = (str(config_path), *sig)
        _CONFIG_PARSE_FAILURES[str(config_path)] = (*sig, str(exc))
    except OSError:
        key = (str(config_path), 0, 0, 0, 0)
    if key in _CONFIG_PARSE_WARNED:
        return
    _CONFIG_PARSE_WARNED.add(key)
    from hermes_cli.config_backups import backup_config
    # A read error leaves an intact file behind: no "corrupt" copy of a good file.
    backup_path = None if isinstance(exc, OSError) else backup_config(config_path, "corrupt")
    msg = format_config_parse_failure(config_path, exc, fallback=fallback)
    if backup_path is not None:
        msg += f" A copy of the broken file was saved to {backup_path}."
    logger.warning("%s Details: %s", msg, _yaml_error_details(exc))
    try:
        sys.stderr.write(f"⚠️  hermes config: {msg}\n    Details: {_yaml_error_details(exc)}\n")
        sys.stderr.flush()
    except Exception:
        pass


def get_active_config_parse_failure() -> Optional[str]:
    """Return the recorded parse error while the ACTIVE config.yaml is still byte-identical
    (mtime_ns + size + ino + ctime_ns) to the file that failed to parse; else None."""
    from hermes_cli.config import get_config_path
    try:
        record = _CONFIG_PARSE_FAILURES[str(path := get_config_path())]
        st = path.stat()
        return record[4] if file_signature(st) == record[:4] else None
    except Exception:
        return None


class FailedConfigRead(dict):
    """The fail-open fallback a config reader serves when an existing config.yaml could not be read
    or parsed (``{}``, defaults or last-known-good). Readers use it like any dict; ``save_config`` /
    ``atomic_config_write`` refuse to persist it. The writer merges by deletion, so saving a fallback
    after one transient EMFILE/EIO replaced the whole file with the fallback plus the caller's edit.
    A dict subclass so the refusal survives the load→mutate→save round trip at every call site."""

    def __init__(self, data: Any = (), *, error: Exception):
        super().__init__(data)
        self.read_error = error


def _read_error_fix(exc: OSError) -> str:
    return _FIX_PERMS if isinstance(exc, PermissionError) else (
        "Try again; run `hermes config check` if it keeps failing.")


def _refuse_failed_read(config_path: Path, data: Any) -> None:
    """Refuse to save a fallback; only a read error is worth retrying, bad YAML needs an edit."""
    if not isinstance(data, FailedConfigRead):
        return
    exc = data.read_error
    if isinstance(exc, OSError):
        raise _refuse_overwrite(config_path, "could not be read", exc, _read_error_fix(exc))
    raise _refuse_overwrite(
        config_path, "has a formatting error", exc, _FIX_YAML.format(backups=_backups_dir_display()))


def _refuse_overwrite(config_path: Path, reason: str, exc: Exception, fix: str) -> RuntimeError:
    """Error for a write that must not replace an existing config.yaml. Plain lead + ``Details:``."""
    where = _yaml_error_location(exc)
    at = f" ({where})" if where else ""
    return RuntimeError(
        f"Your settings file ({config_path}) {reason}{at}, so this change was not saved. {fix} "
        f"Details: {_yaml_error_details(exc)}")


def _backups_dir_display() -> str:
    from hermes_constants import display_hermes_home
    return f"{display_hermes_home()}/backups/config/"