"""Agent configuration (architecture L9: 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 from pathlib import Path from typing import Any from nisse_core.config import ( REFERENCE, ConfigError, VaultConfig, flatten, llm_numbers, parse_tuning, parse_vault, people, read_file, resolve, ) from nisse_core.conversation.matrix import MatrixConfig from nisse_core.models.compact import CompactionTuning, check_tuning from nisse_core.models.llm import ( DEFAULT_MAX_OUTPUT_TOKENS, DEFAULT_REQUEST_TIMEOUT_S, RoleMap, ) DEFAULT_CONFIG_FILE = "nisse-dev.yaml" REQUIRED_KEYS = ( "state_db", "llm.base_url", "llm.token", "roles.driver", "roles.helper", "roles.expert", "matrix.homeserver", "matrix.user_id", "matrix.token", "matrix.room_id", "matrix.operators", "gitea.url", "gitea.token", "gitea.operators", ) OPTIONAL_KEYS = ( "roles.classifier", "llm.max_output_tokens", "llm.request_timeout_s", ) # Settings the old loop had (until M2.8), and what to do instead. RETIRED_KEYS = { "loop": "the loop is the only one now; remove it", "roles.planner": "use roles.driver", "roles.executor": "the driver does the work now; remove it", } # A project entry (D8), under `projects: {name: {...}}`. PROJECT_REQUIRED = ("repo", "workspace", "build_host.address", "build_host.key") PROJECT_OPTIONAL = ( "base_branch", "push_key", "push_user", "push_token", "build_host.user", "build_host.port", "labels", "secrets", "egress", "trust.merge", ) # The trust dial (L3, D5): who merges a project's PRs. "approved": the # harness merges once an operator approved the PR's current commit and its # checks pass, behind protection that requires that approval. MERGE_MODES = ("operator", "approved") # Who may see a project secret (D7, D8): the model's commands get it as an # environment variable, redacted from what comes back; harness-only ones are # never in a command the model chose. VISIBILITY = ("commands", "harness") _PROJECT_NAME = re.compile(r"[a-z0-9][a-z0-9-]*") GENERAL = "general" # the scope of conversation not about a project (L1) # --- 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 DriverTuning: max_turns: int = 40 failure_limit: int = 2 @dataclass(frozen=True) class PreflightTuning: max_turns: int = 20 # model calls a preflight run may make (M4.1) command_timeout_s: int = 900 # one command in it: installs take a while @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 # Compaction (L6): over compact_at of the prompt budget, the model's # notes replace the middle of the run, keeping at most compact_keep_turns # recent turns, in compact_to of it. compact_at 0: never. compact_at: float = field(default=0.6, metadata={"zero": True}) compact_to: float = 0.25 compact_keep_turns: int = 3 def compaction(self) -> CompactionTuning: return CompactionTuning( compact_at=self.compact_at, compact_to=self.compact_to, keep_turns=self.compact_keep_turns, ) @dataclass(frozen=True) class LoopTuning: # A restatement on a `bug` issue proceeds after this long without a # reply (D1). bug_grace_s: float = 600.0 # A `general` conversation closes after this long without a message (L1). general_idle_s: float = 86400.0 @dataclass(frozen=True) class BuildVMTuning: # An unreachable build VM is tried again this often (#137). retry_s: float = 300.0 # Once it answers, it must still answer this much later before work # goes on, so it has finished booting. settle_s: float = 180.0 @dataclass(frozen=True) class MemoryTuning: # Model calls the memory pass at an item's close may make (L6.4). max_turns: int = 10 @dataclass(frozen=True) class Tuning: driver: DriverTuning = field(default_factory=DriverTuning) preflight: PreflightTuning = field(default_factory=PreflightTuning) expert: ExpertTuning = field(default_factory=ExpertTuning) delegation: DelegationTuning = field(default_factory=DelegationTuning) tools: ToolsTuning = field(default_factory=ToolsTuning) context: ContextTuning = field(default_factory=ContextTuning) loop: LoopTuning = field(default_factory=LoopTuning) build_vm: BuildVMTuning = field(default_factory=BuildVMTuning) memory: MemoryTuning = field(default_factory=MemoryTuning) def _parse_tuning(raw: Any, errors: list[str]) -> Tuning: tuning = parse_tuning( raw, Tuning, errors, retired={"executor": "it is tuning.driver now"}, # until M2.8 ) if tuning.context.safety_margin >= 1: errors.append("tuning.context.safety_margin: must be less than 1") check_tuning(tuning.context.compaction(), "tuning.context", errors) return tuning # --- loading --------------------------------------------------------------- def _parse_projects( raw: Any, env: Mapping[str, str], errors: list[str] ) -> tuple["Project", ...]: """`projects:`, a mapping of name to entry (D8). Every problem in every entry is reported, like the rest of the file.""" if not isinstance(raw, dict) or not raw: errors.append("projects: must name at least one project") return () projects: list[Project] = [] repos: dict[str, str] = {} for name, entry in raw.items(): where = f"projects.{name}" if not isinstance(name, str) or not _PROJECT_NAME.fullmatch(name): errors.append(f"{where}: a project name is lowercase letters, digits, -") continue if name == GENERAL: errors.append(f"{where}: `{GENERAL}` is the scope of conversations (L1)") continue if not isinstance(entry, dict): errors.append(f"{where}: must be a mapping") continue project = _parse_project(name, entry, env, errors) if project is None: continue if project.repo in repos: errors.append( f"{where}.repo: {project.repo} is also project {repos[project.repo]}'s" ) repos[project.repo] = name projects.append(project) return tuple(projects) def _parse_project( name: str, entry: Mapping[str, Any], env: Mapping[str, str], errors: list[str] ) -> "Project | None": where = f"projects.{name}" # labels, secrets and egress are structured; the rest flattens. flat = flatten({k: v for k, v in entry.items() if k not in ("secrets",)}) known = set(PROJECT_REQUIRED) | set(PROJECT_OPTIONAL) for key in sorted(set(flat) - known): errors.append(f"{where}.{key}: unknown setting") values = { key: resolve(value, f"{where}.{key}", env, errors) for key, value in flat.items() if key in known } missing = [ key for key in PROJECT_REQUIRED if values.get(key) is None or (isinstance(values[key], str) and not values[key].strip()) ] if missing: errors.append(f"{where}: unset or empty — " + ", ".join(missing)) return None port = values.get("build_host.port", 22) try: port = int(port) if not 0 < port < 65536: raise ValueError except (TypeError, ValueError): errors.append(f"{where}.build_host.port: must be a TCP port number") push_user = values.get("push_user") push_token = values.get("push_token") push_key = values.get("push_key") if bool(push_user) != bool(push_token): errors.append(f"{where}: push_user and push_token: set both or neither") if push_token and push_key: errors.append(f"{where}: set a push token or a push key, not both") labels = _strings(values.get("labels", []), f"{where}.labels", errors) egress = _strings(values.get("egress", []), f"{where}.egress", errors) merge = str(values.get("trust.merge") or "operator") if merge not in MERGE_MODES: errors.append(f"{where}.trust.merge: must be one of {', '.join(MERGE_MODES)}") secrets = _parse_secrets(entry.get("secrets"), where, env, errors) return Project( name=name, repo=str(values["repo"]), workspace=Path(values["workspace"]).expanduser(), build_host=BuildHost( address=str(values["build_host.address"]), key=Path(values["build_host.key"]).expanduser(), user=str(values.get("build_host.user") or "root"), port=port if isinstance(port, int) else 22, ), base_branch=str(values.get("base_branch") or "main"), push_key=Path(push_key).expanduser() if push_key else None, push_user=str(push_user) if push_user else None, push_token=str(push_token) if push_token else None, labels=labels, secrets=secrets, egress=egress, merge=merge, ) def _strings(raw: Any, where: str, errors: list[str]) -> tuple[str, ...]: if not isinstance(raw, list) or not all( isinstance(x, str) and x.strip() for x in raw ): errors.append(f"{where}: must be a list of names") return () return tuple(x.strip() for x in raw) def _parse_secrets( raw: Any, where: str, env: Mapping[str, str], errors: list[str] ) -> dict[str, "Secret"]: """name → {value: ${VAR}, visible: commands | harness}. A value must be a reference: secrets live in secret stores, never in this file.""" if raw is None: return {} if not isinstance(raw, dict): errors.append(f"{where}.secrets: must be a mapping of name to secret") return {} secrets: dict[str, Secret] = {} for name, spec in raw.items(): at = f"{where}.secrets.{name}" if not isinstance(name, str) or not re.fullmatch(r"[A-Z_][A-Z0-9_]*", name): errors.append(f"{at}: a secret's name is an environment variable name") continue if not isinstance(spec, dict) or set(spec) - {"value", "visible"}: errors.append(f"{at}: give value and visible, nothing else") continue visible = spec.get("visible") if visible not in VISIBILITY: errors.append(f"{at}.visible: must be one of {', '.join(VISIBILITY)}") continue value = spec.get("value") if not isinstance(value, str) or not REFERENCE.fullmatch(value.strip()): errors.append(f"{at}.value: must be a ${{VAR}} reference, never inline") continue secrets[name] = Secret(resolve(value.strip(), at, env, errors), visible) return secrets @dataclass(frozen=True) class BuildHost: """The tool VM the workspace lives on (L13, D3). ``key`` is the path to the agent's SSH private key for it.""" address: str key: Path user: str = "root" port: int = 22 @dataclass(frozen=True) class Profile: """A model's capability profile (L9), from benchmarks: what one role is told about another, never the model's name.""" tools: str = "" # how reliably it keeps the tool-call contract strengths: str = "" weaknesses: str = "" def describe(self) -> str: parts = [ f"tool calls: {self.tools}" if self.tools else "", f"strong at {self.strengths}" if self.strengths else "", f"weak at {self.weaknesses}" if self.weaknesses else "", ] return "; ".join(p for p in parts if p) or "no profile recorded" def _parse_profiles(raw: Any, errors: list[str]) -> dict[str, "Profile"]: """`profiles:`, model alias -> {tools, strengths, weaknesses}.""" if raw is None: return {} if not isinstance(raw, dict): errors.append("profiles: must be a mapping of model alias to profile") return {} profiles: dict[str, Profile] = {} for model, spec in raw.items(): where = f"profiles.{model}" if not isinstance(spec, dict): errors.append(f"{where}: must be a mapping") continue unknown = set(spec) - {"tools", "strengths", "weaknesses"} if unknown: errors.append(f"{where}.{min(unknown)}: unknown setting") continue if not all(isinstance(v, str) for v in spec.values()): errors.append(f"{where}: values are text") continue profiles[str(model)] = Profile(**{k: v.strip() for k, v in spec.items()}) return profiles @dataclass(frozen=True) class Secret: value: str = field(repr=False) visible: str = "harness" # one of VISIBILITY @dataclass(frozen=True) class Project: """A project entry (D8): what makes a project known to the agent, and everything per-project the rest of the agent refers to.""" name: str repo: str workspace: Path # on the build VM build_host: BuildHost base_branch: str = "main" # What may push to the project repository, handed to each push on the # build VM and to nothing else (D4): a push-only account's HTTPS token # (push_user, push_token), or the path to a deploy key (push_key). push_key: Path | None = None push_user: str | None = None push_token: str | None = field(default=None, repr=False) labels: tuple[str, ...] = () # added to every issue's own (D1) secrets: Mapping[str, Secret] = field(default_factory=dict) egress: tuple[str, ...] = () # beyond the default; recorded, not yet enforced merge: str = "operator" # the trust dial (L3): who merges its PRs def command_secrets(self) -> dict[str, str]: """The secrets the model's commands may see, by name.""" return { name: s.value for name, s in self.secrets.items() if s.visible == "commands" } @dataclass(frozen=True) class AgentConfig: state_db: Path llm_base_url: str # Credentials never appear in a repr: a failing test or a log line # would print them. llm_token: str = field(repr=False) roles: RoleMap matrix: MatrixConfig operators: frozenset[str] gitea_url: str gitea_token: str = field(repr=False) gitea_operators: frozenset[str] projects: tuple[Project, ...] profiles: Mapping[str, Profile] = field(default_factory=dict) 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 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) projects = _parse_projects(tree.get("projects"), env, errors) profiles = _parse_profiles(tree.get("profiles"), errors) vault = parse_vault(tree.get("vault"), errors) flat = flatten( { k: v for k, v in tree.items() if k not in ("projects", "profiles", "vault") } ) flat.pop("tuning", None) known = set(REQUIRED_KEYS) | set(OPTIONAL_KEYS) for key in sorted(set(flat) - known): if key in RETIRED_KEYS: errors.append(f"{key}: retired — {RETIRED_KEYS[key]}") else: 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, request_timeout = llm_numbers(values, errors) if errors: raise ConfigError("cannot start: " + "; ".join(errors)) config = 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=str(values["roles.driver"]), helper=str(values["roles.helper"]), expert=str(values["roles.expert"]), classifier=str(values.get("roles.classifier") or ""), ), 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_operators=people(values["gitea.operators"]), projects=projects, profiles=profiles, max_output_tokens=max_output, request_timeout_s=request_timeout, tuning=tuning, vault=vault, 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. It lives on the build VM, so this only bites when the build host is this machine (tests, a local bring-up), and there it matters exactly as before. 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. """ 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)) for project in self.projects: workspace = project.workspace.expanduser().resolve() 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} (project {project.name}) 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 { "projects": ", ".join( f"{p.name} ({p.repo}, {p.build_host.address}:{p.workspace})" for p in self.projects ), "driver": self.roles.driver, "helper": self.roles.helper, }