Claude
352 lines
14 KiB
Python
352 lines
14 KiB
Python
"""Finding normalization and synthesis for multi-lens reviews."""
|
|
|
|
import re
|
|
import os
|
|
|
|
from ai_review import _SEVERITY_EMOJI, is_test_path
|
|
from .opencode_lens_config import ReviewerSpec, _coerce_str
|
|
from .opencode_workspace import changed_files
|
|
|
|
# Env:
|
|
# PRAGENT_MAX_PARALLEL_LENSES per-review lens fan-out cap (default 4).
|
|
# The webhook's _review_slots still bounds
|
|
# total concurrent reviews; this bounds the
|
|
# subprocess fan-out inside one review.
|
|
# PRAGENT_LENS_TIMEOUT seconds per lens subprocess (default 540).
|
|
# PRAGENT_REVIEWERS set to "1" to force the fan-out path even
|
|
# when the repo's config is absent.
|
|
|
|
import concurrent.futures as _cf
|
|
import dataclasses as _dc
|
|
|
|
|
|
MAX_PARALLEL_LENSES = int(os.environ.get("PRAGENT_MAX_PARALLEL_LENSES", "4"))
|
|
LENS_TIMEOUT_S = int(os.environ.get("PRAGENT_LENS_TIMEOUT", "540"))
|
|
|
|
# Length caps per finding field. Cheap insurance against DoorDash's "noise on
|
|
# clean code" failure mode — one lens writing 200 words + another writing 10
|
|
# bullets = inconsistent review, regardless of synthesis.
|
|
FINDING_TITLE_MAX = 120
|
|
FINDING_BODY_MAX = 600
|
|
FINDING_SUGGESTION_MAX = 280
|
|
PER_FILE_CAP = 2
|
|
PER_PR_CAP = 7
|
|
|
|
# Tone-strip regex — drops the mushy AI-tone openers that turn a finding into
|
|
# a hedge. Applied to the title AND body before length capping. DoorDash's
|
|
# same problem (different lenses wrote different prose styles); deterministic
|
|
# regex is the cheapest fix.
|
|
_TONE_STRIP_RE = re.compile(
|
|
r"^(consider|it might be worth|perhaps|maybe|i think|i would suggest|"
|
|
r"you may want to|you could|it would be better to|it's worth|"
|
|
r"one option is|one approach is|note that|be aware that|"
|
|
r"as a general rule|as a best practice)\s*[:\-—,]?\s*",
|
|
re.I,
|
|
)
|
|
|
|
# Lens id rules. Lowercase kebab-case, ≤ 32 chars. Must match `[a-z0-9-]+`.
|
|
_LENS_ID_RE = re.compile(r"^[a-z0-9-]{1,32}$")
|
|
|
|
SEVERITY_ORDER = ("low", "medium", "high", "critical")
|
|
SEVERITY_RANK = {s: i for i, s in enumerate(SEVERITY_ORDER)}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Synthesizer — normalize, filter, dedup, cap
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _normalize_lens_finding(raw: dict, spec: ReviewerSpec, model: str) -> dict | None:
|
|
"""Lens-emitted {title, body, ruleId, severity, path, line, suggestion, reference}
|
|
→ legacy schema {severity, path, line, problem, fix, suggestion, reference, _lens,
|
|
_lens_model, _ruleId, _posthash}. Returns None if path/line invalid.
|
|
|
|
The mapping:
|
|
problem ← "{title}\n\n{body}" (capped to FINDING_BODY_MAX)
|
|
fix ← "" (lens agents don't separate; let the
|
|
inline comment carry the prose)
|
|
The synthesizer + tone-strip + length-cap runs over problem before posting.
|
|
"""
|
|
if not isinstance(raw, dict):
|
|
return None
|
|
path = _coerce_str(raw.get("path", ""))
|
|
line = raw.get("line")
|
|
if not path or not isinstance(line, int) or line < 1:
|
|
return None
|
|
sev = _coerce_str(raw.get("severity", "medium")).lower()
|
|
if sev not in SEVERITY_ORDER:
|
|
sev = "medium"
|
|
title = _coerce_str(raw.get("title", ""))
|
|
body = _coerce_str(raw.get("body", ""))
|
|
if not title and not body:
|
|
return None
|
|
problem = f"{title}\n\n{body}".strip() if body else title
|
|
suggestion = _coerce_str(raw.get("suggestion", ""))[:FINDING_SUGGESTION_MAX]
|
|
reference = _coerce_str(raw.get("reference", ""))
|
|
rule_id = _coerce_str(raw.get("ruleId", "")).upper()
|
|
return {
|
|
"severity": sev,
|
|
"path": path,
|
|
"line": line,
|
|
"problem": problem,
|
|
"fix": "",
|
|
"suggestion": suggestion,
|
|
"reference": reference,
|
|
"_lens": spec.id,
|
|
"_lens_model": model,
|
|
"_ruleId": rule_id,
|
|
"_posthash": posthash(path, line, sev, problem),
|
|
}
|
|
|
|
|
|
def posthash(path: str, line: int, severity: str, problem: str) -> str:
|
|
"""sha256[:16] of `path\\nline\\nseverity\\nproblem[:80].strip().lower()`.
|
|
|
|
Identical scheme to `pilot/feedback.py::posthash` — the golden-vector
|
|
test pins equality so FP-vote data lines up across the lens pipeline and
|
|
the feedback DB without a migration. Severity participates because
|
|
"CRITICAL bug" and "LOW nit" at the same line are different signals.
|
|
"""
|
|
import hashlib
|
|
h = hashlib.sha256()
|
|
h.update(f"{path}\n".encode())
|
|
h.update(f"{line}\n".encode())
|
|
h.update(f"{severity.upper()}\n".encode())
|
|
h.update(problem[:80].strip().lower().encode())
|
|
return h.hexdigest()[:16]
|
|
|
|
|
|
def _lens_posthash(finding: dict) -> str:
|
|
"""Compute posthash on a normalized finding (which already has path/line/severity/problem)."""
|
|
return posthash(
|
|
finding.get("path", "?"),
|
|
int(finding.get("line", 0) or 0),
|
|
finding.get("severity", "low"),
|
|
finding.get("problem", ""),
|
|
)
|
|
|
|
|
|
def _agreement_hash(finding: dict) -> str:
|
|
"""Severity-free hash for cross-lens agreement detection.
|
|
|
|
Two lenses flagging the same line on the same problem at different
|
|
severities (e.g. security=high, perf=low) still count as agreement —
|
|
that's the signal `_multi_lens` should highlight. Severity-keyed
|
|
`_posthash` is what the feedback DB indexes; this is for the synthesis
|
|
step only.
|
|
"""
|
|
import hashlib
|
|
h = hashlib.sha256()
|
|
h.update(f"{finding.get('path', '?')}\n".encode())
|
|
h.update(f"{int(finding.get('line', 0) or 0)}\n".encode())
|
|
h.update(finding.get("problem", "")[:80].strip().lower().encode())
|
|
return h.hexdigest()[:16]
|
|
|
|
|
|
def _tone_strip(text: str) -> str:
|
|
"""Strip the AI-tone openers in `_TONE_STRIP_RE` from a single line/short
|
|
prose. Case-insensitive. Returns the text otherwise unchanged."""
|
|
if not text:
|
|
return text
|
|
# Apply to the first non-empty line only (body text may have multiple lines)
|
|
parts = text.split("\n", 1)
|
|
head = parts[0]
|
|
new_head = _TONE_STRIP_RE.sub("", head, count=1).strip()
|
|
if len(parts) == 1:
|
|
return new_head
|
|
return new_head + "\n" + parts[1] if new_head else parts[1]
|
|
|
|
|
|
def _cap_text(text: str, max_chars: int) -> str:
|
|
if len(text) <= max_chars:
|
|
return text
|
|
return text[: max_chars - 1].rstrip() + "…"
|
|
|
|
|
|
def _drop_below_floor(finding: dict, floor: str) -> bool:
|
|
"""True if finding should be DROPPED (severity is below the floor)."""
|
|
return SEVERITY_RANK.get(finding["severity"], 0) < SEVERITY_RANK.get(floor, 0)
|
|
|
|
|
|
def synthesize(
|
|
findings_per_lens: dict[str, list[dict]],
|
|
reviewers: list[ReviewerSpec],
|
|
*,
|
|
per_pr_cap: int = PER_PR_CAP,
|
|
per_file_cap: int = PER_FILE_CAP,
|
|
) -> list[dict]:
|
|
"""Merge + filter + dedup + cap. Returns the final findings list.
|
|
|
|
Pipeline:
|
|
1. severity_floor filter per lens
|
|
2. tone-strip + length-cap
|
|
3. per-lens max_findings cap
|
|
4. per-file cap (lowest severity dropped)
|
|
5. cross-lens dedup by posthash — keep highest severity
|
|
6. cross-lens severity promotion when 2+ lenses agree
|
|
7. per-PR cap (highest severity first)
|
|
"""
|
|
# ReviewerSpec lookup by id for per-lens knobs
|
|
by_id = {r.id: r for r in reviewers}
|
|
|
|
# 1 + 2 + 3: filter + tone-strip + length cap + per-lens cap
|
|
merged: list[dict] = []
|
|
for lens_id, items in findings_per_lens.items():
|
|
spec = by_id.get(lens_id)
|
|
if spec is None:
|
|
continue
|
|
kept = [f for f in items if not _drop_below_floor(f, spec.severity_floor)]
|
|
for f in kept:
|
|
f["problem"] = _cap_text(_tone_strip(f["problem"]), FINDING_BODY_MAX)
|
|
# Per-lens cap: top max_findings by severity, ties broken by original order
|
|
ranked = sorted(
|
|
enumerate(kept),
|
|
key=lambda kv: -SEVERITY_RANK.get(kv[1]["severity"], 0),
|
|
)[: spec.max_findings]
|
|
# Re-sort by original order so the final list reads naturally
|
|
ranked.sort(key=lambda kv: kv[0])
|
|
merged.extend(kv[1] for kv in ranked)
|
|
|
|
if not merged:
|
|
return merged
|
|
|
|
# 4: per-file cap (PER_FILE_CAP). Drop lowest severity on overflow.
|
|
by_path: dict[str, list[dict]] = {}
|
|
for f in merged:
|
|
by_path.setdefault(f["path"], []).append(f)
|
|
for path, group in by_path.items():
|
|
if len(group) <= per_file_cap:
|
|
continue
|
|
group_sorted = sorted(
|
|
group, key=lambda f: -SEVERITY_RANK.get(f["severity"], 0)
|
|
)
|
|
kept_ids = {id(f) for f in group_sorted[:per_file_cap]}
|
|
merged = [f for f in merged if f["path"] != path or id(f) in kept_ids]
|
|
|
|
# 5: dedup by posthash. Keep highest severity; on tie, first-listed lens.
|
|
lens_order = {r.id: i for i, r in enumerate(reviewers)}
|
|
by_hash: dict[str, dict] = {}
|
|
for f in merged:
|
|
h = f["_posthash"]
|
|
prev = by_hash.get(h)
|
|
if prev is None:
|
|
by_hash[h] = f
|
|
continue
|
|
prev_rank = SEVERITY_RANK.get(prev["severity"], 0)
|
|
cur_rank = SEVERITY_RANK.get(f["severity"], 0)
|
|
if cur_rank > prev_rank or (
|
|
cur_rank == prev_rank
|
|
and lens_order.get(f["_lens"], 99) < lens_order.get(prev["_lens"], 99)
|
|
):
|
|
by_hash[h] = f
|
|
deduped = list(by_hash.values())
|
|
|
|
# 6: cross-lens severity promotion. When 2+ lenses reported the same
|
|
# agreement (severity-free), promote the survivor's severity by one step
|
|
# (never past critical). Tag with `_multi_lens: True` so the summary
|
|
# section can flag it. Use `_agreement_hash` (path|line|problem) so
|
|
# different severities from different lenses still count.
|
|
multi_lens_hashes: set[str] = set()
|
|
hash_lens_count: dict[str, set[str]] = {}
|
|
for f in merged:
|
|
h = _agreement_hash(f)
|
|
hash_lens_count.setdefault(h, set()).add(f["_lens"])
|
|
for h, lenses in hash_lens_count.items():
|
|
if len(lenses) >= 2:
|
|
multi_lens_hashes.add(h)
|
|
for f in deduped:
|
|
if _agreement_hash(f) in multi_lens_hashes:
|
|
cur = SEVERITY_RANK.get(f["severity"], 0)
|
|
if cur < len(SEVERITY_ORDER) - 1:
|
|
f["severity"] = SEVERITY_ORDER[cur + 1]
|
|
f["_multi_lens"] = True
|
|
|
|
# 7: per-PR cap. Highest severity first; ties broken by lens order.
|
|
deduped.sort(
|
|
key=lambda f: (
|
|
-SEVERITY_RANK.get(f["severity"], 0),
|
|
lens_order.get(f["_lens"], 99),
|
|
)
|
|
)
|
|
return deduped[:per_pr_cap]
|
|
|
|
|
|
def _synthesize_summary_fields(
|
|
findings: list[dict],
|
|
diff: str,
|
|
changed_paths: list[str] | None = None,
|
|
) -> tuple[list[str], str, str]:
|
|
"""Synthesize review-level meta from the merged findings + diff.
|
|
|
|
Returns (walkthrough, risk_verdict, test_coverage) — the three new
|
|
top-level fields in the pragent review JSON shape
|
|
(`ai_review.parse_review_output` extracts them as the 5th, 6th, and
|
|
7th tuple elements, defaulting to `[]` / `""` when missing).
|
|
|
|
Real implementation (Task 8). Python fallback used when the lens
|
|
fan-out path is engaged (the synthesized JSON fence in `run_lenses_review`
|
|
has no model to call, so we build these fields deterministically from
|
|
the merged findings + the diff):
|
|
- walkthrough: one line per changed file. When findings exist, group
|
|
by path and pick the peak-severity problem as the headline; when
|
|
no findings exist, just announce "changed".
|
|
- risk_verdict: a one-line verdict driven by the highest severity
|
|
bucket that has any findings ("Critical risk" / "High risk" /
|
|
"Medium risk" / "Low risk").
|
|
- test_coverage: "Tests changed" if any changed path matches
|
|
`is_test_path`, else "No tests for behavioral change in `<path>`."
|
|
pointing at the first non-test path.
|
|
"""
|
|
# None-safe: callers occasionally pass None when the upstream merger
|
|
# short-circuited. Treat as empty so the for-loop and group-by below
|
|
# never crash.
|
|
findings = findings or []
|
|
# walkthrough
|
|
walkthrough: list[str] = []
|
|
if findings:
|
|
by_path: dict[str, list[dict]] = {}
|
|
for f in findings:
|
|
by_path.setdefault(f.get("path", "?"), []).append(f)
|
|
for path, group in sorted(by_path.items()):
|
|
peak = max(
|
|
group,
|
|
key=lambda x: SEVERITY_RANK.get(x.get("severity", "low"), 0),
|
|
)
|
|
problem_lines = (peak.get("problem") or "").splitlines()
|
|
problem = problem_lines[0][:80].strip() if problem_lines else ""
|
|
emoji = _SEVERITY_EMOJI.get(peak.get("severity", "low"), "⚪")
|
|
walkthrough.append(f"`{path}` — {emoji} {problem}")
|
|
else:
|
|
files = changed_paths if changed_paths is not None else changed_files(diff)
|
|
for p in files:
|
|
walkthrough.append(f"`{p}` — changed")
|
|
|
|
# risk_verdict
|
|
sev_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0}
|
|
for f in findings:
|
|
s = f.get("severity", "low")
|
|
sev_counts[s] = sev_counts.get(s, 0) + 1
|
|
if sev_counts["critical"]:
|
|
rv = f"Critical risk: {sev_counts['critical']} critical finding(s)."
|
|
elif sev_counts["high"]:
|
|
rv = f"High risk: {sev_counts['high']} high finding(s)."
|
|
elif sev_counts["medium"]:
|
|
rv = f"Medium risk: {sev_counts['medium']} medium finding(s)."
|
|
else:
|
|
rv = "Low risk: clean or minor nits only."
|
|
|
|
# test_coverage
|
|
paths = changed_paths if changed_paths is not None else changed_files(diff)
|
|
test_changed = any(is_test_path(p) for p in paths)
|
|
non_test = [p for p in paths if not is_test_path(p)]
|
|
if test_changed and non_test:
|
|
tc = "Tests changed"
|
|
elif non_test:
|
|
tc = f"No tests for behavioral change in `{non_test[0]}`."
|
|
elif test_changed:
|
|
tc = "Tests changed"
|
|
else:
|
|
tc = ""
|
|
|
|
return walkthrough, rv, tc
|