nisse/nisse-core/src/nisse_core/config.py

"""Configuration any agent shares (architecture L9: nothing hardcoded).

Configuration is a YAML file holding no secrets: a credential is written as a
reference, ``token: ${NISSE_LLM_TOKEN}``, resolved from the environment, so
the file can be reviewed, shared, or committed without leaking. The
environment supplies those references and nothing else: every setting has
exactly one home.

Every value is collected and validated **before** anything is built, and a
single error names every problem at once; unknown keys are errors too. These
are the pieces an agent's own configuration is built from: reading the file,
resolving references, the people allowed to instruct it, tuning sections,
the LLM's numbers, and the knowledge vault.
"""

import re
from collections.abc import Mapping
from dataclasses import dataclass, fields
from pathlib import Path
from typing import Any

import yaml

from nisse_core.knowledge.vault import ACCESS, DEFAULT_IGNORE
from nisse_core.models.llm import DEFAULT_MAX_OUTPUT_TOKENS, DEFAULT_REQUEST_TIMEOUT_S

REFERENCE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")


class ConfigError(RuntimeError):
    pass


def people(raw: Any) -> frozenset[str]:
    """Operator allowlists: a YAML list, or comma-separated in the
    environment. Blanks are dropped so a trailing comma cannot admit an empty
    identity."""
    parts = raw if isinstance(raw, list) else str(raw).split(",")
    return frozenset(str(part).strip() for part in parts if str(part).strip())


def flatten(tree: Mapping[str, Any], prefix: str = "") -> dict[str, Any]:
    flat: dict[str, Any] = {}
    for key, value in tree.items():
        dotted = f"{prefix}{key}"
        if isinstance(value, dict) and dotted != "tuning":
            flat.update(flatten(value, f"{dotted}."))
        else:
            flat[dotted] = value
    return flat


def resolve(value: Any, key: str, env: Mapping[str, str], errors: list[str]) -> Any:
    """Replace ${VAR} references with the environment's value. A reference to
    an unset variable is an error, never an empty string: an empty token would
    pass for a value and fail at the first request."""
    if not isinstance(value, str):
        return value

    def lookup(match: re.Match[str]) -> str:
        name = match.group(1)
        found = (env.get(name) or "").strip()
        if not found:
            errors.append(f"{key}: references ${{{name}}}, which is unset or empty")
        return found

    return REFERENCE.sub(lookup, value)


def read_file(path: Path, errors: list[str]) -> dict[str, Any]:
    try:
        loaded = yaml.safe_load(path.read_text())
    except yaml.YAMLError as exc:
        errors.append(f"{path}: not valid YAML ({exc})")
        return {}
    if loaded is None:
        return {}
    if not isinstance(loaded, dict):
        errors.append(f"{path}: must be a mapping at the top level")
        return {}
    return loaded


def positive(value: Any, kind: type, *, zero: bool = False) -> int | float | None:
    """``value`` as ``kind`` if above 0 (or 0 itself, given ``zero``)."""
    # bool is an int in Python; `true` for a turn limit is a mistake.
    if isinstance(value, bool) or not isinstance(value, int | float):
        return None
    if kind is int and not float(value).is_integer():
        return None
    if value < 0 or (value == 0 and not zero):
        return None
    return kind(value)


def parse_tuning(
    raw: Any,
    tuning: type,
    errors: list[str],
    *,
    retired: Mapping[str, str] | None = None,
) -> Any:
    """`tuning:` as ``tuning``, a dataclass of section dataclasses whose
    defaults are the values the agent ran with before they were
    configurable, so an absent section changes nothing. ``retired``: old
    section names, and what to do instead."""
    retired = retired or {}
    sections_of: dict[str, type] = {
        f.name: f.default_factory  # type: ignore[misc]
        for f in fields(tuning)
    }
    if raw is None:
        return tuning()
    if not isinstance(raw, dict):
        errors.append("tuning: must be a mapping")
        return tuning()
    sections: dict[str, Any] = {}
    for name, section in raw.items():
        cls = sections_of.get(name)
        if name in retired:
            errors.append(f"tuning.{name}: retired — {retired[name]}")
            continue
        if cls is None:
            errors.append(f"tuning.{name}: unknown section")
            continue
        if not isinstance(section, dict):
            errors.append(f"tuning.{name}: must be a mapping")
            continue
        known = {f.name: f for f in fields(cls)}
        values: dict[str, Any] = {}
        for key, value in section.items():
            spec = known.get(key)
            if spec is None:
                errors.append(f"tuning.{name}.{key}: unknown setting")
                continue
            # A field with metadata {"zero": True} may also be 0 ("don't wait").
            zero = bool(spec.metadata.get("zero"))
            kind = float if spec.type in (float, "float") else int
            number = positive(value, kind, zero=zero)
            if number is None:
                need = "zero or more" if zero else "a positive number"
                errors.append(f"tuning.{name}.{key}: must be {need}")
                continue
            values[key] = number
        sections[name] = cls(**values)
    return tuning(**sections)


def llm_numbers(values: Mapping[str, Any], errors: list[str]) -> tuple[int, float]:
    """`llm.max_output_tokens` and `llm.request_timeout_s`, or their
    defaults."""
    max_output = values.get("llm.max_output_tokens", DEFAULT_MAX_OUTPUT_TOKENS)
    try:
        max_output = int(max_output)
        if max_output <= 0:
            raise ValueError
    except (TypeError, ValueError):
        errors.append("llm.max_output_tokens: must be a positive integer")
        max_output = DEFAULT_MAX_OUTPUT_TOKENS
    timeout = values.get("llm.request_timeout_s", DEFAULT_REQUEST_TIMEOUT_S)
    try:
        timeout = float(timeout)
        if timeout <= 0:
            raise ValueError
    except (TypeError, ValueError):
        errors.append("llm.request_timeout_s: must be a positive number of seconds")
        timeout = DEFAULT_REQUEST_TIMEOUT_S
    return max_output, timeout


@dataclass(frozen=True)
class VaultConfig:
    """The knowledge vault (L10): where it is, and the folders this agent
    sees, each "rw" or "ro" (one writer per folder)."""

    root: Path
    folders: Mapping[str, str]
    ignore: tuple[str, ...] = DEFAULT_IGNORE


def parse_vault(raw: Any, errors: list[str]) -> VaultConfig | None:
    if raw is None:
        return None  # no vault: no note tools
    if not isinstance(raw, dict):
        errors.append("vault: must be a mapping")
        return None
    for key in sorted(set(raw) - {"root", "folders", "ignore"}):
        errors.append(f"vault.{key}: unknown setting")
    root = raw.get("root")
    if not isinstance(root, str) or not root.strip():
        errors.append("vault.root: the vault's folder on the agent host")
    folders = raw.get("folders")
    if not isinstance(folders, dict) or not folders:
        errors.append("vault.folders: a mapping of folder to rw or ro")
        folders = {}
    clean: dict[str, str] = {}
    for folder, access in folders.items():
        where = f"vault.folders.{folder}"
        name = str(folder).strip("/")
        if not name or name.startswith("/") or ".." in name.split("/"):
            errors.append(f"{where}: a folder inside the vault")
        elif access not in ACCESS:
            errors.append(f"{where}: must be one of {', '.join(ACCESS)}")
        else:
            clean[name] = access
    names = sorted(clean)
    for a in names:
        for b in names:
            if a != b and b.startswith(f"{a}/"):
                errors.append(f"vault.folders: {b} is inside {a}; one access each")
    ignore = raw.get("ignore", list(DEFAULT_IGNORE))
    if not isinstance(ignore, list) or not all(isinstance(i, str) for i in ignore):
        errors.append("vault.ignore: a list of name patterns")
        ignore = list(DEFAULT_IGNORE)
    if errors:
        return None
    return VaultConfig(Path(str(root)).expanduser(), clean, tuple(ignore))