dongarandClaude Opus 5.5
d49435d4e5
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m16s
Two setup failures found while calibrating the new benchmark tasks, both of which would also fail real milestones. Devstral's chat template (Mistral) rejected any turn in which the model wrote some text before a tool call: the OpenAI renderer made one message per event, so the turn became a text message and a tool-call message in a row, and llama.cpp answered 500. The renderer now folds consecutive user or assistant messages into one, which is the OpenAI format for a turn anyway. Checked against the live server: the captured failing request no longer occurs, over six real runs. LiteLLMClient gave up on any request after a fixed 120 s. A dense 27B model writing a whole module in one turn takes longer, so the executor escalated work that was going well. llm.request_timeout_s now sets it for both clients, default 600. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
376 lines
13 KiB
Python
376 lines
13 KiB
Python
"""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, DEFAULT_REQUEST_TIMEOUT_S, 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", "llm.request_timeout_s", "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
|
|
request_timeout_s: float = DEFAULT_REQUEST_TIMEOUT_S
|
|
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")
|
|
|
|
request_timeout = values.get("llm.request_timeout_s", DEFAULT_REQUEST_TIMEOUT_S)
|
|
try:
|
|
request_timeout = float(request_timeout)
|
|
if request_timeout <= 0:
|
|
raise ValueError
|
|
except (TypeError, ValueError):
|
|
errors.append("llm.request_timeout_s: must be a positive number of seconds")
|
|
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,
|
|
request_timeout_s=request_timeout,
|
|
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,
|
|
}
|