"""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))