"""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 ``." 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