nisse/nisse-review/src/nisse_review/config.py

"""nisse-review's configuration (architecture L9: nothing hardcoded; R5).

A YAML file holding no secrets, built from ``nisse_core.config``: the LLM and
its roles, Gitea, the state store, the vault, and tuning. Credentials are
``${VAR}`` references resolved from the environment.
"""

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

from nisse_core.config import (
    ConfigError,
    VaultConfig,
    flatten,
    llm_numbers,
    parse_tuning,
    parse_vault,
    people,
    read_file,
    resolve,
)
from nisse_core.models.llm import (
    DEFAULT_MAX_OUTPUT_TOKENS,
    DEFAULT_REQUEST_TIMEOUT_S,
    RoleMap,
)

DEFAULT_CONFIG_FILE = "nisse-review.yaml"

REQUIRED_KEYS = (
    "state_db",
    "llm.base_url",
    "llm.token",
    "roles.reviewer",
    "roles.expert",
    "gitea.url",
    "gitea.token",
    "gitea.operators",
)
OPTIONAL_KEYS = (
    "llm.max_output_tokens",
    "llm.request_timeout_s",
    # Patterns of the repositories it reviews ("dev/nisse", "dev/*", "*/*"),
    # public ones included; unset, every one it was given (R1, ``access``).
    "repositories",
)


@dataclass(frozen=True)
class ReviewTuning:
    poll_s: float = 60.0  # how often Gitea is asked for ready PRs
    max_turns: int = 40  # a review run's turn cap
    diff_chars: int = 60000  # the diff in the review's context; files on demand
    attempts: int = 2  # failed runs on one PR before saying so on it
    # A new head waits for its CI to finish: at most ci_wait_s while checks
    # are pending, and ci_grace_s for checks to appear at all. 0: don't wait.
    ci_wait_s: float = field(default=3600.0, metadata={"zero": True})
    ci_grace_s: float = field(default=120.0, metadata={"zero": True})
    # Reviews and reply sets before it sums up for the operator (R2).
    max_rounds: int = 4


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


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


@dataclass(frozen=True)
class Tuning:
    review: ReviewTuning = field(default_factory=ReviewTuning)
    expert: ExpertTuning = field(default_factory=ExpertTuning)
    context: ContextTuning = field(default_factory=ContextTuning)


@dataclass(frozen=True)
class ReviewConfig:
    state_db: Path
    llm_base_url: str
    llm_token: str = field(repr=False)
    # The reviewer drives (R5); the helper role isn't used, so it is the
    # reviewer's model too.
    roles: RoleMap
    gitea_url: str
    gitea_token: str = field(repr=False)
    gitea_operators: frozenset[str]  # whose comments are the operator's (R4)
    max_output_tokens: int = DEFAULT_MAX_OUTPUT_TOKENS
    request_timeout_s: float = DEFAULT_REQUEST_TIMEOUT_S
    tuning: Tuning = field(default_factory=Tuning)
    vault: VaultConfig | None = None
    repositories: frozenset[str] | None = None  # None: all it can read

    @classmethod
    def load(cls, env: Mapping[str, str] = os.environ) -> "ReviewConfig":
        """The file named by NISSE_CONFIG, or ./nisse-review.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}")
        errors: list[str] = []
        return cls.from_tree(read_file(path, errors), env, errors)

    @classmethod
    def from_tree(
        cls,
        tree: Mapping[str, Any],
        env: Mapping[str, str] = os.environ,
        errors: list[str] | None = None,
    ) -> "ReviewConfig":
        errors = [] if errors is None else errors
        tuning = parse_tuning(tree.get("tuning"), Tuning, errors)
        if tuning.context.safety_margin >= 1:
            errors.append("tuning.context.safety_margin: must be less than 1")
        vault = parse_vault(tree.get("vault"), errors)
        flat = flatten({k: v for k, v in tree.items() if k != "vault"})
        flat.pop("tuning", None)
        known = set(REQUIRED_KEYS) | set(OPTIONAL_KEYS)
        errors += [f"{key}: unknown setting" for key in sorted(set(flat) - known)]
        values = {
            key: resolve(value, key, env, errors)
            for key, value in flat.items()
            if key in known
        }
        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, timeout = llm_numbers(values, errors)
        patterns = people(values["repositories"]) if "repositories" in values else None
        for pattern in sorted(patterns or ()):
            owner, slash, name = pattern.partition("/")
            if not (owner and slash and name) or "/" in name:
                errors.append(f"repositories: {pattern!r} is not owner/name")
        if errors:
            raise ConfigError("cannot start: " + "; ".join(errors))
        reviewer = str(values["roles.reviewer"])
        return cls(
            state_db=Path(values["state_db"]).expanduser(),
            llm_base_url=str(values["llm.base_url"]),
            llm_token=str(values["llm.token"]),
            roles=RoleMap(
                driver=reviewer, helper=reviewer, expert=str(values["roles.expert"])
            ),
            gitea_url=str(values["gitea.url"]),
            gitea_token=str(values["gitea.token"]),
            gitea_operators=people(values["gitea.operators"]),
            max_output_tokens=max_output,
            request_timeout_s=timeout,
            tuning=tuning,
            vault=vault,
            repositories=patterns,
        )