gitea_admin/pragent
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(review): multi-lens orchestration — 5 parallel opencode subprocesses (security/docs/code-quality/tests/perf)

Browse Source 
On by default, opt-out via "reviewers": []. 290 tests pass.

- pilot/opencode_review.py: ReviewerSpec dataclass, default_reviewers(),
  parse_reviewers_config(), parse_triage_config(), resolve_reviewers(),
  _normalize_lens_finding(), posthash() (matches feedback.py scheme),
  _agreement_hash() (severity-free for cross-lens promotion), _tone_strip(),
  synthesize() 7-stage (severity_floor → tone-strip → length cap → per-lens
  max → per-file cap → dedup by _posthash → cross-lens severity promote →
  per-PR cap), run_lenses() (ThreadPoolExecutor pool=4), triage(),
  _intersect_with_triage(), _filter_by_skip_if(), run_lenses_review().
  run() routes to fan-out when config.reviewers[] present or PRAGENT_REVIEWERS=1.
- pilot/ai_review.py: parse_repo_config learns reviewers[] and triage objects
  (id regex /^[a-z0-9][a-z0-9-]{0,31}$/, 8-entry cap, agent_file/model/
  severity_floor/max_findings/activation/skip_if_all_changed_paths/hotpath_globs).
  review_pr branches to opencode_review.run_lenses_review when configured.
  _render_collapsible_usage shows lenses: ... line when present.
- .opencode/agents/{docs,code-quality,triage}.md: 3 new lens subagents.
- .opencode/skills/lens-orchestration/SKILL.md: strict-JSON contract every
  lens subagent MUST honor.
- .opencode/agents/pragent.md: slim to coordinator; no more hardcoded
  @security/@tests/@perf delegation; loads lens-orchestration skill.
- .opencode/README.md: rewrite 'Add a review lens' recipe for multi-lens.
- pilot/README-webhook.md: new 'Multi-lens pipeline' section (diagram +
  default roster + config schema + env vars + cross-lens dedup contract).
- tests: 36 new tests (test_ai_review.py +12 reviewers/triage/usage,
  test_opencode_review.py +24 orchestration). posthash golden-vector matches
  feedback.py exactly across 5 severity × 2 line cases.
This commit is contained in:
pragent-bot
2026-08-20 22:16:21 +00:00
parent e8ebc54362
commit 78bcf6a9a0
11 changed files with 1847 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pilot/README-webhook.md
+156 -2
View File
@@ -225,6 +225,82 @@ so the `/tmp/pragent-work` emptyDir is writable.
[csa]: https://labs.cloudsecurityalliance.org/research/csa-research-note-comment-control-github-prompt-injection-20/
## Multi-lens pipeline (5 default lenses, on by default)
Default `AI-REVIEW` runs spawn **one opencode subprocess per lens in parallel**
and synthesize the merged findings before posting. Cheaper than 5 sequential
reviews because the headroom proxy caches the byte-identical brief across
lens calls (lenses 2..N hit cache).
```
Gitea webhook
│
▼
pilot/ai_review.review_pr
│ resolve config + sort changed paths
▼
pilot/opencode_review.run_lenses_review
│ spawn 1..N subprocesses (default 5)
▼
┌── security ──┐ ┌── docs ──┐ ┌── code-quality ──┐ ┌── tests ──┐ ┌── perf ──┐
│ opencode │ │ opencode │ │ opencode │ │ opencode │ │ opencode │
│ subprocess │ │ subprocess│ │ subprocess │ │ subprocess│ │ subprocess│
└──────┬────────┘ └─────┬────┘ └─────────┬────────┘ └─────┬──────┘ └─────┬─────┘
└──────────── synthesise (dedup, severity promote, cap) ─────────────┘
│
▼
post_inline_review (existing path, unchanged)
```
**Default roster** (5 lenses, all on `headroom/glm-5.2:cloud`):
| id | severity_floor | max_findings | target |
|---------------|----------------|--------------|--------|
| `security` | low | 12 | auth, crypto, secrets, SQL, file I/O, supply chain |
| `docs` | low | 8 | README, CHANGELOG, docstrings, code-fence breakage |
| `code-quality`| low | 8 | dead code, hidden complexity, suppressed errors |
| `tests` | low | 8 | coverage gaps for changed logic, missing assertions |
| `perf` | medium | 6 | hot-path globs, O(n²) loops, N+1 queries |
Set `.pr-review.json: "reviewers": []` to opt out (single-primary fallback).
**Per-lens config** (drop-in):
```jsonc
{
"reviewers": [
{ "id": "security", "severity_floor": "high", "max_findings": 10 },
{ "id": "docs", "activation": "off" },
{ "id": "perf", "skip_if_all_changed_paths": "docs/**" },
{ "id": "my-lens", "agent_file": ".opencode/agents/my-lens.md", "model": "headroom/glm-5.2:cloud" }
],
"triage": { "enabled": true, "max_lenses": 4 },
"max_findings": 7
}
```
Triage (off by default, but `enabled: true` recommended) runs a tiny
primary agent that picks a subset of lenses based on the diff's changed
files. Fail-open: if triage errors, all lenses run.
**Env vars:**
| var | default | effect |
|-----|---------|--------|
| `PRAGENT_MAX_PARALLEL_LENSES` | 4 | cap concurrency |
| `PRAGENT_LENS_TIMEOUT` | 540 | per-lens subprocess timeout (s) |
| `PRAGENT_REVIEWERS` | unset | force multi-lens fan-out even without `reviewers[]` |
**Cross-lens dedup:** synthesiser drops duplicates by
`sha256[:16](path|line|severity|problem[:80])` (matches the feedback DB's
`posthash`), then promotes multi-lens agreement by one severity step
(never past critical). A `[multi-lens]` tag is added so the summary
section can flag it.
**Adding a new lens**: drop `.opencode/agents/<id>.md` (use an existing
one as a template), then add one entry to `reviewers[]`. That's it — no
Python change, no image rebuild.
## Repo-local focus: `.pr-review.json` (optional)
Drop a `.pr-review.json` at the repo root (committed on the PR's branch, or on
@@ -256,6 +332,74 @@ from what maintainers merged, not from the branch under review. A PR that
Bad/missing file fails open to defaults. Fields are capped (32 list items ×
200 chars; `instructions` 4000 chars). The bot's `read:repository` scope reads it.
## Feedback loop (reactions → daily report)
The bot learns from how humans react to its reviews. The loop has three parts:
1. **Harvest** (every PR webhook). `pilot/feedback_harvest.py` walks back over
the PR's bot-authored reviews + inline comments + their reactions + their
reply threads + their resolved/unresolved state, and writes everything into
`/data/feedback.db` (SQLite, on the `pragent-feedback-data` PVC). It runs
inside the webhook pod, before the new review is scheduled — piggy-backs on
the webhook so there is no second cron just for harvesting. ~50 ms per PR.
3. **Analyze** (`pilot/feedback_analyze.py`). Aggregates findings by `posthash`
(a sha256 of `path:line:severity:problem`) and computes per-finding scores:
- **false-positive score** = `-1` reactions + unresolved status + negation-
phrase replies ("false positive", "intentional", "not a bug"…) − upvotes
− resolved.
- **accepted-pattern score** = upvotes + resolved − downvotes − unresolved
− negation replies.
- **restraint** = fraction of reviewed PRs the bot left a finding on. The
DoorDash rule (2026-07-06, [ZenML recap](https://www.zenml.io/blog/llmops-database)):
*excessive noise on clean code is its own failure mode*. Above ~25% the
report flags ⚠️.
Renders markdown: top-N false-positive candidates, top-N accepted patterns,
a case-review queue (every disagreement with full context), and a
"where to action this" footer.
4. **Deliver** (`pilot/feedback_post.py`). Posts the markdown as a comment on
a single long-lived issue `pragent feedback roll-up` in `gitea_admin/pragent`.
Comments are append-only history — one per run, timestamped.
The daily CronJob (`k8s/pragent-feedback-cronjob.yaml`, schedule `7 3 * * *`)
runs `feedback_post.py`. The webhook pod has `PRAGENT_FEEDBACK_DB=/data/feedback.db`;
an empty / unset value disables harvesting (CI-step pod never gets the PVC).
Human reactions are **not ground truth** — authors accept/reject for workflow
reasons as often as for technical ones (DoorDash lesson). Treat the top-N lists
as a *case-review queue*, not a directive. Re-read the PR before adding
anything to `.pr-review.json:instructions` or the cross-repo `architecture.md`.
### Acting on the report
- **Per-repo**: add a `patterns.deny` glob to `.pr-review.json`, raise the
`severity_threshold`, or amend `instructions` — all read live at the next
review.
- **Cross-repo**: append accepted patterns to the shared
`PRAGENT_ADDITIONAL_CONTEXT_URL` document on Nexus raw-hosted (e.g.
`canalhandia/architecture.md`). The next review picks it up via the
prompt-cached prefix → ~0 marginal cost on step 2+.
- **Benchmark gate** (DoorDash pattern): before changing the model / prompt /
context window, replay the labeled `posthash` corpus against a candidate
change. If a candidate flips ≥ 1 currently-accepted finding into
false-positive, drop it.
### Manual ops
```bash
# ad-hoc report (no post)
python3 pilot/feedback_analyze.py --db /data/feedback.db --out /tmp/report.md
# ad-hoc report for a window
python3 pilot/feedback_analyze.py --db /data/feedback.db --since 1755000000
# force-run the cron now
kubectl -n pragent create job --from cronjob/pragent-feedback pragent-fb-now
kubectl -n pragent logs -l app=pragent-feedback --tail=30
# pause the cron
kubectl -n pragent patch cronjob pragent-feedback -p '{"spec":{"suspend":true}}'
```
## One-time per-owner setup: register a user-level webhook
Gitea **system webhooks** (one webhook for the whole instance — the ideal) are
@@ -396,9 +540,17 @@ typescript-language-server / eslint / ruff) is built locally and imported into
microk8s containerd — it is **not** pulled from a registry (`imagePullPolicy:
Never`). The webhook secret + bot token are a Secret (`pragent-webhook`). An
emptyDir at `/tmp/pragent-work` holds the per-review checkout + the warmed
opencode runtime. Verified: a regular pod on kubernets reaches both
opencode runtime. The PVC `pragent-feedback-data` (1 Gi, microk8s-hostpath,
ReadWriteOnce) is mounted at `/data` and holds the SQLite file the feedback
loop reads + writes — both the webhook pod and the daily CronJob pod share it.
Verified: a regular pod on kubernets reaches both
`<model-proxy-host>:8789` (headroom/glm) and `gitea-http.gitea.svc.cluster.local:3000`.
The feedback CronJob lives in `~/k8s/pragent-feedback-cronjob.yaml` — same
image, same PVC, schedule `7 3 * * *` (nudge off the round-hour). It runs
`feedback_post.py`, which posts the daily report to the `pragent feedback
roll-up` issue in `gitea_admin/pragent`.
Build + deploy after editing the pilot scripts or the factory:
```bash
@@ -422,7 +574,9 @@ Env on the Deployment: `PRAGENT_ENGINE`, `OPENCODE_MODEL`,
`PRAGENT_WORK_ROOT`, `PRAGENT_REVIEW_TIMEOUT`, `GITEA_API`, `OLLAMA_URL`,
`OLLAMA_MODEL`, `OLLAMA_MAX_TOKENS`, `DIFF_MAX_CHARS`,
`PRAGENT_ADDITIONAL_CONTEXT_URL` (optional, see "Repo-provided static
context" above), `PRAGENT_MAX_CONCURRENT_REVIEWS`, `PRAGENT_MAX_BODY_BYTES`
context" above), `PRAGENT_FEEDBACK_DB` (defaults to `/data/feedback.db` on
the webhook; empty / unset disables harvesting — the CI-step path doesn't
get the PVC), `PRAGENT_MAX_CONCURRENT_REVIEWS`, `PRAGENT_MAX_BODY_BYTES`
are literals;
`WEBHOOK_SECRET` + `PRAGENT_BOT_TOKEN` come from the Secret. The image now runs
as uid 10001 — add `securityContext: {runAsNonRoot: true, runAsUser: 10001,
pilot/ai_review.py
+123 -8
View File
@@ -1129,9 +1129,17 @@ def _render_collapsible_usage(usage: dict | None, model: str, config: dict | Non
f"- **Est. cost on {eq_label}**: {eq_s}{eq_note}",
f"- **Actual**: {actual_s}{actual_note}",
f"- **Scope**: {scope}",
"",
"</details>",
]
# Multi-lens fan-out: surface the lens roster + summed steps so the user
# can see which lenses contributed (and that triage didn't drop them all).
lenses = usage.get("lenses")
if lenses:
ls = usage.get("lens_steps", usage.get("steps", 0))
lines.append(
f"- **Lenses**: {', '.join(f'`{x}`' for x in lenses)} "
f"({len(lenses)} parallel subprocesses, {ls} summed steps)"
)
lines += ["", "</details>"]
return "\n".join(lines)
@@ -1245,6 +1253,97 @@ def parse_repo_config(raw: str) -> dict:
# 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
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
@@ -1814,13 +1913,29 @@ def review_pr(
# `model` here is the bare id (OLLAMA_MODEL). OPENCODE_MODEL overrides
# with the full ref; otherwise we prefix the configured provider.
oc_model = os.environ.get("OPENCODE_MODEL") or f"headroom/{model}"
stdout, usage = opencode_review.run(
api=api, repo=repo, index=index, sha=sha, token=token,
title=title, body=body, diff=diff, config=config,
prior_reviews=prior, model=oc_model,
compression_note=compression_note,
additional_context=additional_context,
# Multi-lens fan-out: when the repo declared `reviewers[]` (or the
# operator pinned PRAGENT_REVIEWERS=1), spawn one opencode subprocess
# per lens in parallel and synthesize. Falls through to the legacy
# single-primary path when neither is set.
use_lenses = bool((config or {}).get("reviewers")) or bool(
os.environ.get("PRAGENT_REVIEWERS")
)
if use_lenses and hasattr(opencode_review, "run_lenses_review"):
stdout, usage = opencode_review.run_lenses_review(
api=api, repo=repo, index=index, sha=sha, token=token,
title=title, body=body, diff=diff, config=config,
prior_reviews=prior, model=oc_model,
compression_note=compression_note,
additional_context=additional_context,
)
else:
stdout, usage = opencode_review.run(
api=api, repo=repo, index=index, sha=sha, token=token,
title=title, body=body, diff=diff, config=config,
prior_reviews=prior, model=oc_model,
compression_note=compression_note,
additional_context=additional_context,
)
review_summary, findings, summary_changes, risks = parse_review_output(stdout)
if not findings and not review_summary:
# The findings JSON was missing or malformed. Don't discard the
pilot/opencode_review.py
+800
View File
@@ -510,6 +510,788 @@ _PROMPT = (
)
# ---------------------------------------------------------------------------
# Multi-lens orchestration (config-driven fan-out + synthesis)
# ---------------------------------------------------------------------------
#
# When `.pr-review.json:reviewers[]` is configured (or PRAGENT_REVIEWERS=1), the
# `run()` entry point forks N parallel opencode subprocesses — one per lens
# (security, docs, code-quality, tests, perf by default). Each runs in a
# shared workdir, reads the same brief, and emits its own findings JSON.
# `synthesize()` then merges + dedups by posthash (the same key the feedback
# loop uses, so FP-vote data lines up automatically). Absent/empty reviewers[]
# falls back to the legacy single-primary path (no behavior change).
#
# 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)}
@_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 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"]
# ---------------------------------------------------------------------------
# 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]
# ---------------------------------------------------------------------------
# Per-lens subprocess + parallel fan-out
# ---------------------------------------------------------------------------
def _extract_json_object(text: str) -> dict | None:
"""Last balanced {...} JSON object in text, or None. Tolerant: scans for
a ```json fence first, then falls back to a balanced-brace scan of the
whole text. Reused by `_run_one_lens` to parse a lens's output."""
if not text:
return None
# 1. Try the last ```json ... ``` fence.
fences = list(re.finditer(r"```(?:json)?\s*\n", text))
for m in reversed(fences):
start = m.end()
# find the matching ```
end = text.find("```", start)
if end == -1:
continue
block = text[start:end].strip()
try:
obj = json.loads(block)
except json.JSONDecodeError:
# balanced-brace scan inside the block
for cand in _balanced_jsons(block):
try:
return json.loads(cand)
except json.JSONDecodeError:
continue
continue
if isinstance(obj, dict):
return obj
if isinstance(obj, list) and obj and isinstance(obj[0], dict):
return {"findings": obj}
# 2. Balanced scan over the whole text.
for cand in reversed(list(_balanced_jsons(text))):
try:
obj = json.loads(cand)
except json.JSONDecodeError:
continue
if isinstance(obj, dict):
return obj
if isinstance(obj, list) and obj and isinstance(obj[0], dict):
return {"findings": obj}
return None
def _balanced_jsons(text: str):
"""Yield each top-level balanced {...} substring (greedy on the inside)."""
depth = 0
start = None
for i, ch in enumerate(text):
if ch == "{":
if depth == 0:
start = i
depth += 1
elif ch == "}":
if depth > 0:
depth -= 1
if depth == 0 and start is not None:
yield text[start:i + 1]
start = None
def _run_one_lens(
workdir: str,
spec: ReviewerSpec,
model: str,
factory_root: str,
) -> tuple[list[dict], dict | None, str]:
"""Run one lens subprocess. Returns (findings, usage, lens_id).
findings are RAW lens shape ({title, body, ruleId, severity, path, line,
suggestion, reference}) — normalize in `synthesize()`. Empty list on
failure (does NOT abort siblings — fail-open per-lens).
"""
bin_ = _opencode_bin()
home = _shared_home()
_warm_opencode(home, model)
env = _build_env(home)
agent_path = spec.agent_path(factory_root)
prompt = (
f"You are the {spec.id} lens. Read .pragent/brief.md, load the "
f"lens-orchestration skill (mandatory), and return STRICT JSON "
f"findings per that skill. Cap at {spec.max_findings} findings, "
f"severity >= {spec.severity_floor}. The agent markdown you should "
f"load is at {agent_path} (it sets your role + permissions)."
)
cmd = [
bin_, "run", "--pure", "--format", "json",
"--agent", spec.id, "--dir", workdir, "--model", model,
prompt,
]
try:
proc = subprocess.run(
cmd, cwd=workdir, env=env, capture_output=True, text=True,
stdin=subprocess.DEVNULL, timeout=LENS_TIMEOUT_S,
)
except subprocess.TimeoutExpired:
print(f"pragent: lens {spec.id} timed out after {LENS_TIMEOUT_S}s", flush=True)
return [], None, spec.id
except Exception as e:
print(f"pragent: lens {spec.id} crashed: {e}", flush=True)
return [], None, spec.id
text, usage = parse_opencode_events(proc.stdout or "")
if not text.strip():
print(
f"pragent: lens {spec.id} empty text (rc={proc.returncode}); "
f"stderr tail: {(proc.stderr or '')[-500:]}",
flush=True,
)
return [], usage, spec.id
obj = _extract_json_object(text)
if obj is None:
print(f"pragent: lens {spec.id} produced no parseable JSON", flush=True)
return [], usage, spec.id
raw_findings = obj.get("findings") or []
if not isinstance(raw_findings, list):
return [], usage, spec.id
normalized = []
for raw in raw_findings:
n = _normalize_lens_finding(raw, spec, model)
if n is not None:
normalized.append(n)
print(
f"pragent: lens {spec.id} findings={len(normalized)} "
f"raw={len(raw_findings)} ok=1",
flush=True,
)
return normalized, usage, spec.id
def run_lenses(
workdir: str,
reviewers: list[ReviewerSpec],
default_model: str,
factory_root: str,
) -> dict[str, tuple[list[dict], dict | None]]:
"""Fan out N lens subprocesses in parallel. Returns lens_id → (findings, usage).
Uses a thread pool (stdlib `concurrent.futures.ThreadPoolExecutor`) — the
work is I/O-bound subprocess wait, not CPU. `MAX_PARALLEL_LENSES` bounds
concurrency so a config that asks for 20 lenses doesn't fork-bomb the pod.
"""
if not reviewers:
return {}
pool_size = min(len(reviewers), MAX_PARALLEL_LENSES)
out: dict[str, tuple[list[dict], dict | None]] = {}
with _cf.ThreadPoolExecutor(max_workers=pool_size) as ex:
futures = {
ex.submit(
_run_one_lens, workdir, spec,
spec.model or default_model, factory_root,
): spec
for spec in reviewers
}
for fut in _cf.as_completed(futures):
spec = futures[fut]
try:
findings, usage, _ = fut.result()
except Exception as e:
print(f"pragent: lens {spec.id} worker crashed: {e}", flush=True)
findings, usage = [], None
out[spec.id] = (findings, usage)
return out
def triage(
workdir: str,
triage_cfg: dict,
reviewers: list[ReviewerSpec],
default_model: str,
factory_root: str,
) -> list[str] | None:
"""Run the triage agent. Returns the lens subset with surface, or None to
mean "all reviewers" (fail-open on any error).
`triage_cfg.enabled = False` → skip triage, return None.
"""
if not triage_cfg.get("enabled", True):
return None
bin_ = _opencode_bin()
home = _shared_home()
_warm_opencode(home, default_model)
env = _build_env(home)
lens_ids = [r.id for r in reviewers]
prompt = (
f"You are the triage agent. Read .pragent/brief.md. "
f"Available lens ids: {','.join(lens_ids)}. "
f"Return STRICT JSON on a single line: {{\"lenses\":[\"<id>\",...]}}. "
f"Include a lens only if the diff gives it real surface. "
f"Empty list = no lenses needed. No prose."
)
cmd = [
bin_, "run", "--pure", "--format", "json",
"--agent", "triage", "--dir", workdir, "--model", default_model,
prompt,
]
try:
proc = subprocess.run(
cmd, cwd=workdir, env=env, capture_output=True, text=True,
stdin=subprocess.DEVNULL, timeout=120,
)
except (subprocess.TimeoutExpired, Exception) as e:
print(f"pragent: triage crashed: {e}; falling back to all lenses", flush=True)
return None
text, _ = parse_opencode_events(proc.stdout or "")
obj = _extract_json_object(text) if text.strip() else None
if obj is None:
print("pragent: triage no parseable output; falling back to all lenses", flush=True)
return None
lenses = obj.get("lenses")
if not isinstance(lenses, list):
return None
valid = [lid for lid in lenses if isinstance(lid, str) and lid in lens_ids]
cap = triage_cfg.get("max_lenses", 5)
selected = valid[:cap]
print(f"pragent: triage selected {selected}", flush=True)
return selected
def _intersect_with_triage(
reviewers: list[ReviewerSpec], selected_ids: list[str]
) -> list[ReviewerSpec]:
"""Filter `reviewers` to those named by `selected_ids`, preserving the
original order. Lenses in `selected_ids` not present in `reviewers` are
dropped silently. `None` or empty list → no triage, return all."""
if not selected_ids:
return list(reviewers)
sel = set(selected_ids)
return [r for r in reviewers if r.id in sel]
def _filter_by_skip_if(
reviewers: list[ReviewerSpec], changed_paths: list[str]
) -> list[ReviewerSpec]:
"""Drop a lens whose `skip_if_all_changed_paths` matches ALL changed paths.
Pure path-glob check; cheap; runs before triage so we don't pay for an
opencode subprocess we'll skip anyway."""
import fnmatch
out = []
for r in reviewers:
pat = r.skip_if_all_changed_paths.strip()
if pat and changed_paths and all(
fnmatch.fnmatch(p, pat) for p in changed_paths
):
continue
out.append(r)
return out
def merge_usage(parts: list[dict | None]) -> dict:
"""Sum a list of usage dicts (one per lens) into one. Missing fields are
treated as 0; `steps` is summed; `duration_s` becomes the max."""
base = _new_usage()
base["duration_s"] = 0.0
for u in parts:
if not u:
continue
for k in base:
if isinstance(base[k], (int, float)):
base[k] += u.get(k, 0) or 0
return base
# ---------------------------------------------------------------------------
# Multi-lens entry point
# ---------------------------------------------------------------------------
def run_lenses_review(
*,
api: str,
repo: str,
index: str,
sha: str,
token: str,
title: str,
body: str,
diff: str,
config: dict | None,
prior_reviews: list[str] | None,
model: str,
compression_note: str = "",
additional_context: str = "",
) -> tuple[str, dict | None]:
"""Fan-out + synthesize path. Returns (merged-text, merged-usage).
`text` is a synthesized prose summary + the merged findings JSON (the
downstream `ai_review.parse_review_output` expects the same shape it
always has: prose + a final ```json fence with the legacy schema).
"""
os.makedirs(WORK_ROOT, exist_ok=True)
workdir = tempfile.mkdtemp(prefix=f"{repo.replace('/', '_')}-{sha[:8]}-", dir=WORK_ROOT)
keep = bool(os.environ.get("PRAGENT_KEEP_WORK"))
t0 = time.monotonic()
try:
fetch_archive(api, repo, sha, token, workdir)
sanitize_workdir(workdir)
write_brief(
workdir,
repo=repo, index=index, sha=sha, title=title, description=body,
diff=diff, config=config, prior_reviews=prior_reviews,
compression_note=compression_note,
additional_context=additional_context,
)
drop_factory(workdir)
reviewers = resolve_reviewers(config)
if not reviewers:
# Edge case: reviewers[] present but every entry had activation:off.
# Fall back to single-primary.
return _fallback_single_primary(
workdir=workdir, model=model,
)
triage_cfg = parse_triage_config((config or {}).get("triage"))
changed_paths = changed_files(diff)
reviewers = _filter_by_skip_if(reviewers, changed_paths)
selected = triage(
workdir, triage_cfg, reviewers, model, _factory_dir(),
)
if selected is not None:
reviewers = _intersect_with_triage(reviewers, selected) or reviewers
if not reviewers:
return "", None
factory_root = _factory_dir()
results = run_lenses(workdir, reviewers, model, factory_root)
# Merge findings + usage across lenses
findings_per_lens = {lid: r[0] for lid, r in results.items()}
merged = synthesize(findings_per_lens, reviewers)
merged_usage = merge_usage([r[1] for r in results.values()])
# Build a synthetic text response that ai_review.parse_review_output
# can consume (prose summary + final ```json fence with legacy schema).
lens_names = ", ".join(sorted({f["_lens"] for f in merged})) or "—"
sev_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0}
for f in merged:
sev_counts[f["severity"]] = sev_counts.get(f["severity"], 0) + 1
summary = (
f"Multi-lens review of {repo}#{index} "
f"(sha {sha[:8]}). Lenses: {lens_names}. "
f"Findings: critical={sev_counts['critical']} "
f"high={sev_counts['high']} medium={sev_counts['medium']} "
f"low={sev_counts['low']}."
)
# Strip internal _lens/_posthash/_ruleId/_multi_lens/_lens_model keys from
# the merged findings so the legacy parser doesn't see them. (They
# remain in the DB via feedback_harvest which re-derives posthash.)
clean_findings = [
{k: v for k, v in f.items() if not k.startswith("_")}
for f in merged
]
text = (
f"{summary}\n\n"
f"## Findings (multi-lens)\n\n"
f"```json\n{json.dumps({'summary': summary, 'findings': clean_findings}, indent=2)}\n```\n"
)
if merged_usage is not None:
merged_usage["duration_s"] = round(time.monotonic() - t0, 1)
merged_usage["lenses"] = sorted(results.keys())
merged_usage["lens_steps"] = merged_usage.get("steps", 0)
return text, merged_usage
finally:
if not keep:
shutil.rmtree(workdir, ignore_errors=True)
def _fallback_single_primary(workdir: str, model: str) -> tuple[str, dict | None]:
"""Used when reviewers[] resolves to empty (all activation:off)."""
try:
text, usage = run_opencode(workdir, model)
return text, usage
except Exception as e:
print(f"pragent: fallback single-primary failed: {e}", flush=True)
return "", None
def _shared_home() -> str:
"""A persistent shared HOME for opencode across reviews.
@@ -705,7 +1487,25 @@ def run(
`additional_context`: pre-fetched markdown from
`additional_context_urls` / `PRAGENT_ADDITIONAL_CONTEXT_URL`. Rendered as
its own brief section. Empty string by default.
Routing:
* If `config:reviewers[]` is present OR `PRAGENT_REVIEWERS=1` env is set,
delegate to `run_lenses_review` (parallel fan-out + synth).
* Otherwise, the legacy single-primary path (calls `run_opencode`).
The no-config branch is the no-regression gate.
"""
use_fanout = bool((config or {}).get("reviewers")) or bool(
os.environ.get("PRAGENT_REVIEWERS")
)
if use_fanout:
return run_lenses_review(
api=api, repo=repo, index=index, sha=sha, token=token,
title=title, body=body, diff=diff, config=config,
prior_reviews=prior_reviews, model=model,
compression_note=compression_note,
additional_context=additional_context,
)
os.makedirs(WORK_ROOT, exist_ok=True)
workdir = tempfile.mkdtemp(prefix=f"{repo.replace('/', '_')}-{sha[:8]}-", dir=WORK_ROOT)
keep = bool(os.environ.get("PRAGENT_KEEP_WORK"))