139 lines
5.2 KiB
Python
139 lines
5.2 KiB
Python
"""Configuration model for opencode review lenses."""
|
|
|
|
import dataclasses as _dc
|
|
import os
|
|
import re
|
|
|
|
_LENS_ID_RE = re.compile(r"^[a-z0-9-]{1,32}$")
|
|
SEVERITY_ORDER = ("low", "medium", "high", "critical")
|
|
|
|
@_dc.dataclass(frozen=True)
|
|
class ReviewerSpec:
|
|
"""One lens to run. Immutable — synthesized from config once per review."""
|
|
|
|
id: str
|
|
agent_file: str = "" # default derived from id below
|
|
model: str = "" # default = the global OPENCODE_MODEL
|
|
severity_floor: str = "low" # findings below are dropped
|
|
max_findings: int = 12 # per-lens cap before synthesis
|
|
activation: str = "auto" # auto | always | off (off = exclude entirely)
|
|
skip_if_all_changed_paths: str = "" # glob; skip when every changed path matches
|
|
hotpath_globs: tuple[str, ...] = () # for triage hint only
|
|
|
|
def agent_path(self, factory_root: str) -> str:
|
|
"""Resolve the absolute path of this lens's agent markdown."""
|
|
rel = self.agent_file or f".opencode/agents/{self.id}.md"
|
|
return os.path.join(factory_root, rel)
|
|
|
|
|
|
def _coerce_str(v, default: str = "") -> str:
|
|
return str(v).strip() if isinstance(v, (str, int, float)) else default
|
|
|
|
|
|
def _coerce_int(v, default: int, lo: int, hi: int) -> int:
|
|
try:
|
|
n = int(v)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
return max(lo, min(hi, n))
|
|
|
|
|
|
def default_reviewers() -> list[ReviewerSpec]:
|
|
"""The 5-lens default when the repo's `.pr-review.json:reviewers[]` is absent.
|
|
|
|
Order matters: the synthesizer dedups by posthash and keeps the highest
|
|
severity; on tie, the FIRST-listed lens wins. So security first (most
|
|
conservative severity), then docs (additive), then code-quality + tests +
|
|
perf (additive).
|
|
"""
|
|
return [
|
|
ReviewerSpec(id="security", severity_floor="low", max_findings=12),
|
|
ReviewerSpec(id="docs", severity_floor="low", max_findings=8),
|
|
ReviewerSpec(id="code-quality", severity_floor="low", max_findings=8),
|
|
ReviewerSpec(id="tests", severity_floor="low", max_findings=8),
|
|
ReviewerSpec(id="perf", severity_floor="medium", max_findings=6),
|
|
]
|
|
|
|
|
|
def _coerce_str(v, default: str = "") -> str:
|
|
return str(v).strip() if isinstance(v, (str, int, float)) else default
|
|
|
|
|
|
def _coerce_int(v, default: int, lo: int, hi: int) -> int:
|
|
try:
|
|
n = int(v)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
return max(lo, min(hi, n))
|
|
|
|
|
|
def parse_reviewers_config(raw: dict) -> list[ReviewerSpec]:
|
|
"""Read `.pr-review.json:reviewers[]` into `list[ReviewerSpec]`.
|
|
|
|
Validates: id (kebab ≤ 32 chars), model (must contain `/` — provider/model
|
|
ref form), severity_floor ∈ SEVERITY_ORDER, max_findings ∈ [1..30],
|
|
activation ∈ {auto,always,off}, skip_if is a string. Drops invalid entries
|
|
silently. Caps the array at 8.
|
|
|
|
Returns [] on absent/invalid; the caller falls back to `default_reviewers()`.
|
|
"""
|
|
if not isinstance(raw, list):
|
|
return []
|
|
out: list[ReviewerSpec] = []
|
|
for entry in raw[:8]:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
rid = _coerce_str(entry.get("id", "")).lower()
|
|
if not _LENS_ID_RE.match(rid):
|
|
continue
|
|
model = _coerce_str(entry.get("model", ""))
|
|
if model and "/" not in model:
|
|
model = "" # must be provider/model — silent drop of bad model
|
|
sf = _coerce_str(entry.get("severity_floor", "")).lower()
|
|
if sf not in SEVERITY_ORDER:
|
|
sf = "low"
|
|
mf = _coerce_int(entry.get("max_findings"), default=12, lo=1, hi=30)
|
|
act = _coerce_str(entry.get("activation", "auto")).lower()
|
|
if act not in ("auto", "always", "off"):
|
|
act = "auto"
|
|
skip = _coerce_str(entry.get("skip_if_all_changed_paths", ""))
|
|
hot = entry.get("hotpath_globs") or []
|
|
if isinstance(hot, list):
|
|
hot = tuple(_coerce_str(g) for g in hot if _coerce_str(g))[:8]
|
|
else:
|
|
hot = ()
|
|
out.append(ReviewerSpec(
|
|
id=rid,
|
|
agent_file=_coerce_str(entry.get("agent_file", "")),
|
|
model=model,
|
|
severity_floor=sf,
|
|
max_findings=mf,
|
|
activation=act,
|
|
skip_if_all_changed_paths=skip,
|
|
hotpath_globs=hot,
|
|
))
|
|
return out
|
|
|
|
|
|
def parse_triage_config(raw: dict) -> dict:
|
|
"""`.pr-review.json:triage` → safe defaults. Always returns a dict."""
|
|
if not isinstance(raw, dict):
|
|
return {"enabled": True, "model": "", "max_lenses": 5}
|
|
enabled = bool(raw.get("enabled", True))
|
|
model = _coerce_str(raw.get("model", ""))
|
|
max_lenses = _coerce_int(raw.get("max_lenses"), default=5, lo=1, hi=8)
|
|
return {"enabled": enabled, "model": model, "max_lenses": max_lenses}
|
|
|
|
|
|
def resolve_reviewers(config: dict | None) -> list[ReviewerSpec]:
|
|
"""Pick the reviewer list: config-driven if present, else defaults.
|
|
|
|
Drops `activation: off` entries (they're config noise). The triage step
|
|
further filters by surface.
|
|
"""
|
|
cfg = config or {}
|
|
raw = cfg.get("reviewers")
|
|
parsed = parse_reviewers_config(raw) if raw is not None else []
|
|
base = parsed if parsed else default_reviewers()
|
|
return [r for r in base if r.activation != "off"]
|