Claude
506 lines
20 KiB
Python
506 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
from . import pipeline
|
|
from .pipeline import *
|
|
|
|
# Repo config + existing-review helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
# Caps on `.pr-review.json`. The file is committed config, not free-form model
|
|
# input, and every byte of it lands in the prompt — bound it so a bloated (or
|
|
# hostile) config can't crowd out the diff or blow the context window.
|
|
CONFIG_MAX_LIST_ITEMS = 32
|
|
CONFIG_MAX_ITEM_CHARS = 200
|
|
CONFIG_MAX_INSTRUCTIONS_CHARS = 4000
|
|
CONFIG_MAX_PATTERNS_ITEMS = 16 # allow + deny separately, total 32 entries
|
|
CONFIG_MAX_FINDINGS = 30
|
|
CONFIG_MAX_STATIC_MESSAGE_CHARS = 400 # free-text banner, mirror of instructions
|
|
MAX_BUDGET_STEPS = 100
|
|
MAX_BUDGET_TOKENS = 2_000_000
|
|
MAX_BUDGET_SECONDS = 3_600
|
|
MAX_BUDGET_LENSES = 8
|
|
MAX_BUDGET_COST_USD = 100.0
|
|
|
|
STYLES = frozenset(STYLE_DEFAULTS)
|
|
SEVERITY_VALUES = frozenset(SEVERITIES)
|
|
|
|
|
|
def parse_repo_config(raw: str) -> dict:
|
|
"""Parse a .pr-review.json blob tolerantly. Returns {} on any failure.
|
|
|
|
List fields are capped at CONFIG_MAX_LIST_ITEMS entries of
|
|
CONFIG_MAX_ITEM_CHARS each; `instructions` at CONFIG_MAX_INSTRUCTIONS_CHARS;
|
|
`patterns.allow` / `patterns.deny` each capped at CONFIG_MAX_PATTERNS_ITEMS
|
|
of CONFIG_MAX_ITEM_CHARS.
|
|
|
|
Recognised keys (all optional):
|
|
focus, exclude_paths, languages, instructions — text steer
|
|
static_message ≤ CONFIG_MAX_STATIC_MESSAGE_CHARS — banner under header
|
|
style strict|balanced|lenient — default: balanced
|
|
severity_threshold low|medium|high|critical — default: per style
|
|
max_findings 1..CONFIG_MAX_FINDINGS — default: per style
|
|
exclude_tests bool — default: False
|
|
require_tests bool — default: False
|
|
patterns {allow:[…], deny:[…]} — post-filter globs
|
|
model <key of cost_model.PRICES> — per-repo override
|
|
cost_target <key of cost_model.PRICES> — see equivalent_cost
|
|
budget {max_steps, max_total_tokens, max_output_tokens,
|
|
max_duration_seconds, max_lenses, max_equivalent_cost_usd}
|
|
additional_context_urls list[str] (≤ 8) — see fetch_additional_context
|
|
"""
|
|
if not raw:
|
|
return {}
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
|
|
def _str_list(v):
|
|
if isinstance(v, list) and all(isinstance(x, str) for x in v):
|
|
return [x[:CONFIG_MAX_ITEM_CHARS] for x in v[:CONFIG_MAX_LIST_ITEMS]]
|
|
return None
|
|
|
|
out: dict = {}
|
|
for k in ("focus", "exclude_paths", "languages"):
|
|
s = _str_list(data.get(k))
|
|
if s is not None:
|
|
out[k] = s
|
|
|
|
instr = data.get("instructions")
|
|
if isinstance(instr, str) and instr.strip():
|
|
out["instructions"] = instr.strip()[:CONFIG_MAX_INSTRUCTIONS_CHARS]
|
|
|
|
sm = data.get("static_message")
|
|
if isinstance(sm, str) and sm.strip():
|
|
out["static_message"] = sm.strip()[:CONFIG_MAX_STATIC_MESSAGE_CHARS]
|
|
|
|
style = data.get("style")
|
|
if isinstance(style, str) and style.strip().lower() in STYLES:
|
|
out["style"] = style.strip().lower()
|
|
|
|
thresh = data.get("severity_threshold")
|
|
if isinstance(thresh, str) and thresh.strip().lower() in SEVERITY_VALUES:
|
|
out["severity_threshold"] = thresh.strip().lower()
|
|
|
|
mf = data.get("max_findings")
|
|
if isinstance(mf, int) and not isinstance(mf, bool) and 1 <= mf <= CONFIG_MAX_FINDINGS:
|
|
out["max_findings"] = mf
|
|
elif isinstance(mf, str) and mf.strip().isdigit():
|
|
n = int(mf.strip())
|
|
if 1 <= n <= CONFIG_MAX_FINDINGS:
|
|
out["max_findings"] = n
|
|
|
|
for bk in ("exclude_tests", "require_tests"):
|
|
if isinstance(data.get(bk), bool):
|
|
out[bk] = data[bk]
|
|
|
|
pat = data.get("patterns")
|
|
if isinstance(pat, dict):
|
|
allow = _str_list(pat.get("allow"))
|
|
deny = _str_list(pat.get("deny"))
|
|
patterns = {}
|
|
if allow is not None:
|
|
patterns["allow"] = allow[:CONFIG_MAX_PATTERNS_ITEMS]
|
|
if deny is not None:
|
|
patterns["deny"] = deny[:CONFIG_MAX_PATTERNS_ITEMS]
|
|
if patterns:
|
|
out["patterns"] = patterns
|
|
|
|
ct = data.get("cost_target")
|
|
if isinstance(ct, str) and ct.strip():
|
|
out["cost_target"] = ct.strip()
|
|
|
|
# Per-repo model override. Validated against cost_model.PRICES so the value
|
|
# is usable both as the opencode subprocess ref and as the REVIEW_HEADER
|
|
# label (see _resolve_display_model precedence). Unknown values are dropped
|
|
# with a stderr pointer to the valid set — silently ignoring would mask
|
|
# typos from repo admins.
|
|
raw_model = data.get("model")
|
|
if raw_model is not None:
|
|
if isinstance(raw_model, str) and raw_model.strip():
|
|
from cost_model import PRICES # lazy: ollama path dep-free
|
|
candidate = raw_model.strip()
|
|
if candidate in PRICES:
|
|
out["model"] = candidate
|
|
else:
|
|
print(
|
|
f"pragent: .pr-review.json:model={candidate!r} not in "
|
|
f"cost_model.PRICES (valid: {', '.join(sorted(PRICES))}); "
|
|
f"dropping",
|
|
file=sys.stderr, flush=True,
|
|
)
|
|
|
|
acu = data.get("additional_context_urls")
|
|
if isinstance(acu, list):
|
|
urls: list[str] = []
|
|
for x in acu:
|
|
if isinstance(x, str):
|
|
u = x.strip()
|
|
if u:
|
|
urls.append(u)
|
|
if urls:
|
|
# Cap is also enforced later by _resolve_additional_context_urls;
|
|
# this just stops a 10k-entry file from making the config huge.
|
|
out["additional_context_urls"] = urls[:8]
|
|
|
|
# Multi-lens reviewers roster. Absent / empty list = the 5-lens default
|
|
# in pilot/opencode_review.py (security, docs, code-quality, tests, perf).
|
|
# This is the cheap trigger: once the config declares `reviewers[]`, the
|
|
# orchestrator spawns one opencode subprocess per lens in parallel. Set
|
|
# to `[]` to opt out (single-primary fallback). Capped at 8.
|
|
rev = _parse_reviewers_array(data.get("reviewers"))
|
|
if rev is not None:
|
|
out["reviewers"] = rev
|
|
|
|
# Triage (cheap pre-filter that picks a subset of lenses). Off by default
|
|
# to keep the parse deterministic; the orchestrator's own default is
|
|
# to enable it when `reviewers[]` is present.
|
|
tr = _parse_triage_object(data.get("triage"))
|
|
if tr is not None:
|
|
out["triage"] = tr
|
|
|
|
# Repo-level kill-switch: `enabled: false` lets a maintainer pause the bot
|
|
# for this repo without removing the file (handy during a flaky provider
|
|
# outage). Always written so callers can do `cfg.get("enabled") is False`
|
|
# without a separate default — the file itself is committed, so we treat
|
|
# absent / wrong-type as an explicit off rather than as "config missing".
|
|
en = data.get("enabled")
|
|
out["enabled"] = en if isinstance(en, bool) else False
|
|
|
|
# Compare-against roster: list of `cost_model.PRICES` keys the render layer
|
|
# uses to print equivalent-cost lines (one per key) for maintainer
|
|
# budgeting. Unknown keys are dropped with a stderr line so a typo is loud.
|
|
# Lazy import: `cost_model` has no dep on `ai_review`, and the ollama
|
|
# fallback path never hits this branch — keep import-time cost low there.
|
|
from cost_model import PRICES as _PRICES
|
|
ca = data.get("compare_against")
|
|
if isinstance(ca, list):
|
|
cleaned: list[str] = []
|
|
for x in ca:
|
|
if isinstance(x, str) and x.strip() in _PRICES:
|
|
cleaned.append(x.strip())
|
|
elif isinstance(x, str):
|
|
print(
|
|
f"pragent: ignoring compare_against entry {x!r} "
|
|
f"(not in cost_model.PRICES); valid: {', '.join(sorted(_PRICES))}",
|
|
file=sys.stderr, flush=True,
|
|
)
|
|
if cleaned:
|
|
out["compare_against"] = cleaned[:12]
|
|
|
|
budget = _parse_budget(data.get("budget"))
|
|
if budget:
|
|
out["budget"] = budget
|
|
|
|
return out
|
|
|
|
|
|
def _parse_budget(raw) -> dict:
|
|
"""Sanitize optional per-review resource limits from trusted config."""
|
|
if not isinstance(raw, dict):
|
|
return {}
|
|
out: dict = {}
|
|
integer_limits = {
|
|
"max_steps": (1, MAX_BUDGET_STEPS),
|
|
"max_total_tokens": (1, MAX_BUDGET_TOKENS),
|
|
"max_output_tokens": (1, MAX_BUDGET_TOKENS),
|
|
"max_duration_seconds": (1, MAX_BUDGET_SECONDS),
|
|
"max_lenses": (1, MAX_BUDGET_LENSES),
|
|
}
|
|
for key, (lo, hi) in integer_limits.items():
|
|
value = raw.get(key)
|
|
if isinstance(value, int) and not isinstance(value, bool):
|
|
if lo <= value <= hi:
|
|
out[key] = value
|
|
elif isinstance(value, str) and value.strip().isdigit():
|
|
number = int(value.strip())
|
|
if lo <= number <= hi:
|
|
out[key] = number
|
|
cost = raw.get("max_equivalent_cost_usd")
|
|
if isinstance(cost, (int, float)) and not isinstance(cost, bool):
|
|
if 0 < float(cost) <= MAX_BUDGET_COST_USD:
|
|
out["max_equivalent_cost_usd"] = float(cost)
|
|
elif isinstance(cost, str):
|
|
try:
|
|
number = float(cost.strip())
|
|
except ValueError:
|
|
number = 0
|
|
if 0 < number <= MAX_BUDGET_COST_USD:
|
|
out["max_equivalent_cost_usd"] = number
|
|
return out
|
|
|
|
|
|
def _parse_reviewers_array(raw) -> list[dict] | None:
|
|
"""Sanitize `.pr-review.json:reviewers[]` to a list of dicts.
|
|
|
|
Hard caps: 8 entries (default-reviewers.xml-bound), 200 chars per string
|
|
field. Untyped / non-list → None (caller keeps the default). Fields we
|
|
don't know about are dropped (no schema drift allowed).
|
|
"""
|
|
if not isinstance(raw, list):
|
|
return None
|
|
cap = 8
|
|
out: list[dict] = []
|
|
for entry in raw[:cap]:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
spec: dict = {}
|
|
rid = entry.get("id")
|
|
if isinstance(rid, str) and rid.strip():
|
|
cand = rid.strip()[:CONFIG_MAX_ITEM_CHARS]
|
|
# Same id shape required by opencode_review.parse_reviewers_config:
|
|
# kebab-case so it maps 1:1 to .opencode/agents/<id>.md
|
|
import re as _re
|
|
if _re.match(r"^[a-z0-9][a-z0-9-]{0,31}$", cand):
|
|
spec["id"] = cand
|
|
if not spec.get("id"):
|
|
continue
|
|
for sk in ("agent_file", "model"):
|
|
sv = entry.get(sk)
|
|
if isinstance(sv, str) and sv.strip():
|
|
spec[sk] = sv.strip()[:CONFIG_MAX_ITEM_CHARS]
|
|
sf = entry.get("severity_floor")
|
|
if isinstance(sf, str) and sf.strip().lower() in SEVERITY_VALUES:
|
|
spec["severity_floor"] = sf.strip().lower()
|
|
mf = entry.get("max_findings")
|
|
if isinstance(mf, int) and not isinstance(mf, bool) and 1 <= mf <= CONFIG_MAX_FINDINGS:
|
|
spec["max_findings"] = mf
|
|
act = entry.get("activation")
|
|
if isinstance(act, str) and act.strip().lower() in ("auto", "always", "off"):
|
|
spec["activation"] = act.strip().lower()
|
|
skip = entry.get("skip_if_all_changed_paths")
|
|
if isinstance(skip, str) and skip.strip():
|
|
spec["skip_if_all_changed_paths"] = skip.strip()[:CONFIG_MAX_ITEM_CHARS]
|
|
globs = entry.get("hotpath_globs")
|
|
if isinstance(globs, list):
|
|
cleaned = [g for g in globs if isinstance(g, str) and g.strip()]
|
|
if cleaned:
|
|
spec["hotpath_globs"] = [
|
|
g.strip()[:CONFIG_MAX_ITEM_CHARS]
|
|
for g in cleaned[:CONFIG_MAX_LIST_ITEMS]
|
|
]
|
|
out.append(spec)
|
|
return out
|
|
|
|
|
|
def _parse_triage_object(raw) -> dict | None:
|
|
"""Sanitize `.pr-review.json:triage` to a dict.
|
|
|
|
Returns `None` when absent. When the value is malformed (not an object),
|
|
returns `{"enabled": False}` so a typo disables triage rather than
|
|
silently making the orchestrator error.
|
|
"""
|
|
if raw is None:
|
|
return None
|
|
if not isinstance(raw, dict):
|
|
return {"enabled": False}
|
|
out: dict = {}
|
|
if isinstance(raw.get("enabled"), bool):
|
|
out["enabled"] = raw["enabled"]
|
|
if isinstance(raw.get("model"), str) and raw["model"].strip():
|
|
out["model"] = raw["model"].strip()[:CONFIG_MAX_ITEM_CHARS]
|
|
ml = raw.get("max_lenses")
|
|
if isinstance(ml, int) and not isinstance(ml, bool) and 1 <= ml <= 8:
|
|
out["max_lenses"] = ml
|
|
return out
|
|
|
|
|
|
def effective_config(config: dict | None) -> dict:
|
|
"""Apply STYLE_DEFAULTS for any field the config didn't pin.
|
|
|
|
Returns a NEW dict combining the user's `.pr-review.json` (if any) with the
|
|
derived `max_findings` / `severity_threshold`. Style itself is preserved
|
|
so downstream code can branch on it.
|
|
"""
|
|
style = (config or {}).get("style", "balanced")
|
|
max_findings, severity_threshold = STYLE_DEFAULTS.get(style, STYLE_DEFAULTS["balanced"])
|
|
out = dict(config or {})
|
|
out.setdefault("style", style)
|
|
out.setdefault("max_findings", max_findings)
|
|
out.setdefault("severity_threshold", severity_threshold)
|
|
return out
|
|
|
|
|
|
_TEST_PATH_RE = re.compile(
|
|
r"(?:^|/)("
|
|
r"[^/]*[Tt]est\.[A-Za-z]+" # FooTest.java / foo_test.py
|
|
r"|[^/]*\.[Tt]est\.[A-Za-z]+" # foo.Test.java
|
|
r"|[^/]*_test\.py" # foo_test.py
|
|
r"|test_[^/]*\.py" # test_foo.py
|
|
r"|__tests__/[^/]+" # __tests__/foo.js
|
|
r"|[^/]*\.spec\.[A-Za-z]+" # foo.spec.ts
|
|
r")$"
|
|
)
|
|
|
|
|
|
def is_test_path(path: str) -> bool:
|
|
"""Heuristic: is `path` a test file by name/path convention?
|
|
|
|
Conservative — false positives cost real findings; false negatives just
|
|
produce one extra line in the summary. Patterns: `FooTest.java`,
|
|
`foo_test.py`, `test_foo.py`, `__tests__/foo.js`, `foo.spec.ts`, anything
|
|
ending in `.Test.java`.
|
|
"""
|
|
if not path:
|
|
return False
|
|
return bool(_TEST_PATH_RE.search(path))
|
|
|
|
|
|
def _glob_to_regex(glob: str) -> re.Pattern:
|
|
"""Translate a shell-style glob to a compiled regex.
|
|
|
|
Supports `*` (any chars except `/`), `**` (any chars including `/`),
|
|
`?` (single non-`/` char). Other characters are escaped. Used by
|
|
`apply_repo_config` to test `patterns.allow` / `patterns.deny` globs.
|
|
"""
|
|
out = []
|
|
i = 0
|
|
while i < len(glob):
|
|
c = glob[i]
|
|
if c == "*":
|
|
if i + 1 < len(glob) and glob[i + 1] == "*":
|
|
out.append(".*")
|
|
i += 2
|
|
# swallow a following `/` so `**/x` and `x/**/y` behave
|
|
if i < len(glob) and glob[i] == "/":
|
|
i += 1
|
|
continue
|
|
out.append("[^/]*")
|
|
elif c == "?":
|
|
out.append("[^/]")
|
|
else:
|
|
out.append(re.escape(c))
|
|
i += 1
|
|
return re.compile("^" + "".join(out) + "$")
|
|
|
|
|
|
def apply_repo_config(
|
|
findings: list[dict],
|
|
config: dict | None,
|
|
changed_paths: list[str] | None = None,
|
|
) -> tuple[list[dict], list[dict]]:
|
|
"""Filter + cap findings per `.pr-review.json` rules. Returns (kept, dropped).
|
|
|
|
Filters applied (in order):
|
|
1. `exclude_tests` + test-path heuristic → drop test files
|
|
2. `exclude_paths` glob match → drop matched paths
|
|
3. `patterns.deny` glob match → drop matched paths
|
|
4. `patterns.allow` (if non-empty) → keep ONLY matched paths
|
|
5. `severity_threshold` → drop below threshold
|
|
6. `max_findings` → keep first N (highest-severity-first)
|
|
7. `require_tests` → append a low-severity finding
|
|
if changed paths include non-test files but no test files changed
|
|
alongside them (caller passes `changed_paths` from the brief).
|
|
"""
|
|
eff = effective_config(config)
|
|
keep: list[dict] = []
|
|
drop: list[dict] = []
|
|
deny_globs = [_glob_to_regex(g) for g in (eff.get("patterns", {}) or {}).get("deny", [])]
|
|
allow_globs = [_glob_to_regex(g) for g in (eff.get("patterns", {}) or {}).get("allow", [])]
|
|
deny_path_globs = [_glob_to_regex(g) for g in eff.get("exclude_paths", [])]
|
|
threshold_rank = SEVERITY_RANK[eff["severity_threshold"]]
|
|
|
|
for f in findings:
|
|
path = f.get("path", "")
|
|
if eff.get("exclude_tests") and is_test_path(path):
|
|
drop.append(f); continue
|
|
if any(rx.search(path) for rx in deny_path_globs):
|
|
drop.append(f); continue
|
|
if any(rx.search(path) for rx in deny_globs):
|
|
drop.append(f); continue
|
|
if allow_globs and not any(rx.search(path) for rx in allow_globs):
|
|
drop.append(f); continue
|
|
sev_rank = SEVERITY_RANK.get(f.get("severity", "low"), 0)
|
|
if sev_rank < threshold_rank:
|
|
drop.append(f); continue
|
|
keep.append(f)
|
|
|
|
cap = eff["max_findings"]
|
|
if len(keep) > cap:
|
|
dropped = keep[cap:]
|
|
keep = keep[:cap]
|
|
drop.extend(dropped)
|
|
|
|
if eff.get("require_tests") and changed_paths is not None:
|
|
non_test = [p for p in changed_paths if not is_test_path(p)]
|
|
any_test = any(is_test_path(p) for p in changed_paths)
|
|
if non_test and not any_test:
|
|
keep.append({
|
|
"severity": "low",
|
|
"path": non_test[0],
|
|
"line": 1,
|
|
"problem": "no test file changed alongside this behavioral change (require_tests=true)",
|
|
"fix": "add a unit test exercising the changed branch",
|
|
"suggestion": "",
|
|
"reference": "",
|
|
"_config_synthetic": True,
|
|
})
|
|
|
|
return keep, drop
|
|
|
|
|
|
def reviewed_shas(reviews: list[dict]) -> set[str]:
|
|
"""Pull every `<!-- pragent:sha=... -->` marker out of a PR's reviews."""
|
|
shas: set[str] = set()
|
|
for r in reviews or []:
|
|
body = r.get("body") or ""
|
|
for m in pipeline._SHA_MARKER_RE.finditer(body):
|
|
shas.add(m.group(1))
|
|
return shas
|
|
|
|
|
|
def prior_review_bodies(reviews: list[dict], current_sha: str, limit: int = 6) -> list[str]:
|
|
"""Bodies of prior bot reviews (older shas), newest-first, bounded."""
|
|
out = []
|
|
for r in reviews or []:
|
|
body = (r.get("body") or "").strip()
|
|
if not body:
|
|
continue
|
|
shas = pipeline._SHA_MARKER_RE.findall(body)
|
|
# Skip the current sha (that would be a self-reference) and non-bot
|
|
# noise; keep reviews that carry our marker.
|
|
if not shas:
|
|
continue
|
|
if current_sha and current_sha in shas:
|
|
continue
|
|
out.append(body)
|
|
return out[:limit]
|
|
|
|
|
|
def compact_prior_reviews(prior_bodies: list[str]) -> list[str]:
|
|
"""Squeeze prior review bodies down to just the finding bullets.
|
|
|
|
Each prior review's prose ("this PR adds eval() — risky") is noise when the
|
|
model already has the diff; the only thing it needs to *not repeat* is what
|
|
was already flagged. We extract lines matching `-\\s*\\*\\*[SEV]\\*\\*`
|
|
plus their directly-attached location reference (so `[CRITICAL]` stays
|
|
anchored to `path:line`), drop the rest, and return one bullet-list per
|
|
prior review. A prior review that had no parseable findings becomes an
|
|
empty string and is dropped.
|
|
|
|
Local import keeps the ollama path dep-free (extract_finding_bullets lives
|
|
in pilot/diff_compress.py).
|
|
"""
|
|
from diff_compress import extract_finding_bullets
|
|
out = []
|
|
for body in prior_bodies or []:
|
|
bullets = extract_finding_bullets(body)
|
|
if bullets:
|
|
out.append("\n".join(bullets))
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|