nisse/nisse-dev/src/nisse_dev/config.py

"""Agent configuration (spec A5: nothing hardcoded).

Configuration is a YAML file: roles, connections, and the tuning knobs (see
docs/Tuning-Guide.md). It holds no secrets. A credential is written as a
reference, ``token: ${NISSE_LLM_TOKEN}``, and 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. The alternative — failing at first
use — surfaces a missing token twenty minutes into a milestone, on the call
that mattered. Unknown keys are errors too: a misspelt tuning knob that was
silently ignored would look exactly like a knob that has no effect.
"""

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

import yaml

from nisse_dev.llm import DEFAULT_MAX_OUTPUT_TOKENS, RoleMap
from nisse_dev.matrix import MatrixConfig

DEFAULT_CONFIG_FILE = "nisse-dev.yaml"

REQUIRED_KEYS = (
    "state_db",
    "workspace",
    "project",
    "llm.base_url",
    "llm.token",
    "roles.executor",
    "roles.planner",
    "roles.delegate",
    "roles.consultant",
    "matrix.homeserver",
    "matrix.user_id",
    "matrix.token",
    "matrix.room_id",
    "matrix.operators",
    "gitea.url",
    "gitea.token",
    "gitea.repo",
    "gitea.operators",
)
OPTIONAL_KEYS = ("llm.max_output_tokens", "gitea.base_branch")

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


class ConfigError(RuntimeError):
    pass


# --- tuning ----------------------------------------------------------------
# The defaults are the values the agent ran with before they were
# configurable, so an absent `tuning:` section changes nothing.


@dataclass(frozen=True)
class ExecutorTuning:
    max_turns: int = 24
    failure_limit: int = 2


@dataclass(frozen=True)
class ExpertTuning:
    consult_budget: int = 3


@dataclass(frozen=True)
class DelegationTuning:
    max_turns: int = 30
    max_tokens: int = 200_000
    wall_clock_s: float = 900.0


@dataclass(frozen=True)
class ToolsTuning:
    command_timeout_s: int = 120
    max_output_chars: int = 4000


@dataclass(frozen=True)
class ContextTuning:
    keep_recent_turns: int = 6
    safety_margin: float = 0.1


@dataclass(frozen=True)
class Tuning:
    executor: ExecutorTuning = field(default_factory=ExecutorTuning)
    expert: ExpertTuning = field(default_factory=ExpertTuning)
    delegation: DelegationTuning = field(default_factory=DelegationTuning)
    tools: ToolsTuning = field(default_factory=ToolsTuning)
    context: ContextTuning = field(default_factory=ContextTuning)


_TUNING_SECTIONS: dict[str, type] = {
    f.name: f.default_factory  # type: ignore[misc]
    for f in fields(Tuning)
}


def _parse_tuning(raw: Any, errors: list[str]) -> 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 = _TUNING_SECTIONS.get(name)
        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
            number = _positive(value, float if spec.type in (float, "float") else int)
            if number is None:
                errors.append(f"tuning.{name}.{key}: must be a positive number")
                continue
            values[key] = number
        sections[name] = cls(**values)
    margin = sections.get("context", ContextTuning()).safety_margin
    if margin >= 1:
        errors.append("tuning.context.safety_margin: must be less than 1")
    return Tuning(**sections)


def _positive(value: Any, kind: type) -> int | float | None:
    # 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:
        return None
    return kind(value)


# --- loading ---------------------------------------------------------------


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


@dataclass(frozen=True)
class AgentConfig:
    state_db: Path
    workspace: Path
    project: str
    llm_base_url: str
    llm_token: str
    roles: RoleMap
    matrix: MatrixConfig
    operators: frozenset[str]
    gitea_url: str
    gitea_token: str
    gitea_repo: str
    gitea_operators: frozenset[str]
    base_branch: str = "main"
    max_output_tokens: int = DEFAULT_MAX_OUTPUT_TOKENS
    tuning: Tuning = field(default_factory=Tuning)
    config_file: Path | None = None

    @classmethod
    def load(cls, env: Mapping[str, str] = os.environ) -> "AgentConfig":
        """The file named by NISSE_CONFIG, or ./nisse-dev.yaml."""
        path = Path(env.get("NISSE_CONFIG") or DEFAULT_CONFIG_FILE).expanduser()
        if not path.is_file():
            raise ConfigError(
                f"cannot start: no config file at {path}; create it from "
                "nisse-dev.example.yaml, or point NISSE_CONFIG at one"
            )
        return cls.from_file(path, env)

    @classmethod
    def from_file(
        cls, path: Path, env: Mapping[str, str] = os.environ
    ) -> "AgentConfig":
        errors: list[str] = []
        tree = _read_file(path, errors)
        return cls._build(tree, env, errors, config_file=path)

    @classmethod
    def from_tree(
        cls, tree: Mapping[str, Any], env: Mapping[str, str] = os.environ
    ) -> "AgentConfig":
        """From an already-parsed config, as the file would give it."""
        return cls._build(tree, env, [], config_file=None)

    @classmethod
    def _build(
        cls,
        tree: Mapping[str, Any],
        env: Mapping[str, str],
        errors: list[str],
        *,
        config_file: Path | None,
    ) -> "AgentConfig":
        tuning = _parse_tuning(tree.get("tuning"), errors)
        flat = _flatten(tree)
        flat.pop("tuning", None)

        known = set(REQUIRED_KEYS) | set(OPTIONAL_KEYS)
        for key in sorted(set(flat) - known):
            errors.append(f"{key}: unknown setting")
        values = {
            key: _resolve(value, key, env, errors)
            for key, value in flat.items()
            if key in known
        }

        # Whitespace-only counts as missing: a token of spaces would pass a
        # truthiness check and fail at the first request instead.
        missing = [
            key
            for key in REQUIRED_KEYS
            if values.get(key) is None
            or (isinstance(values[key], str) and not values[key].strip())
            or (isinstance(values[key], list) and not _people(values[key]))
        ]
        if missing:
            errors.append("unset or empty — " + ", ".join(missing))

        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")
        if errors:
            raise ConfigError("cannot start: " + "; ".join(errors))

        config = cls(
            state_db=Path(values["state_db"]).expanduser(),
            workspace=Path(values["workspace"]).expanduser(),
            project=str(values["project"]),
            llm_base_url=str(values["llm.base_url"]),
            llm_token=str(values["llm.token"]),
            roles=RoleMap(
                executor=str(values["roles.executor"]),
                planner=str(values["roles.planner"]),
                delegate=str(values["roles.delegate"]),
                consultant=str(values["roles.consultant"]),
            ),
            matrix=MatrixConfig(
                homeserver=str(values["matrix.homeserver"]).rstrip("/"),
                user_id=str(values["matrix.user_id"]),
                token=str(values["matrix.token"]),
                room_id=str(values["matrix.room_id"]),
            ),
            operators=_people(values["matrix.operators"]),
            gitea_url=str(values["gitea.url"]),
            gitea_token=str(values["gitea.token"]),
            gitea_repo=str(values["gitea.repo"]),
            gitea_operators=_people(values["gitea.operators"]),
            base_branch=str(values.get("gitea.base_branch") or "main"),
            max_output_tokens=max_output,
            tuning=tuning,
            config_file=config_file,
        )
        config._refuse_secrets_inside_the_clone(env)
        return config

    def _refuse_secrets_inside_the_clone(self, env: Mapping[str, str]) -> None:
        """The workspace is a git clone the agent commits with ``git add -A``
        and pushes. Anything inside it reaches the project repository, so a
        state store or an env file placed there would publish credentials as a
        side effect of the agent working correctly — silently, and once. The
        config file holds no secrets itself, but it names where they come
        from and how the agent is deployed, which is not the project's
        business either.
        """
        workspace = self.workspace.expanduser().resolve()
        candidates = [
            ("state_db", self.state_db),
            ("NISSE_ENV", Path(env.get("NISSE_ENV", ".env"))),
        ]
        if self.config_file is not None:
            candidates.append(("NISSE_CONFIG", self.config_file))
        offenders = []
        for label, path in candidates:
            resolved = path.expanduser().resolve()
            if resolved == workspace or workspace in resolved.parents:
                offenders.append(f"{label} ({resolved})")
        if offenders:
            raise ConfigError(
                "refusing to start: these sit inside the workspace clone at "
                f"{workspace} and would be committed and pushed to the project "
                "repository — " + ", ".join(offenders)
            )

    def redacted(self) -> dict[str, str]:
        """What the agent may safely announce about how it is configured."""
        return {
            "project": self.project,
            "workspace": str(self.workspace),
            "repo": self.gitea_repo,
            "planner": self.roles.planner,
            "executor": self.roles.executor,
            "delegate": self.roles.delegate,
        }