Marcos
5302e8dcd7
The canalhandia PR review lost all findings because the agent ran out of
context before emitting the closing json fence. Three failure modes hit
the old regex \{.*?\}:
* nested objects inside the fence truncated at the first }
* bare arrays (no {summary, findings} wrapper) returned []
* unfenced JSON in the prose tail was never reached (first not last)
Replace the regex with a balanced-brace scanner:
* _last_json_block walks the fence contents with a depth counter so
nested objects survive
* _last_balanced_json + _balanced_json_substring handle bare arrays and
prose-tail JSON when no fence is present
* _parse_json_tolerant returns list as well as dict; parse_findings and
parse_review_output accept a bare array as the outer value
Agent prompt tightened: reserve the final step for emitting the JSON
block so the analysis isn't lost when context runs out.
10 new tests in tests/pilot/test_ai_review.py cover the new shapes.
Co-Authored-By: Claude <noreply@anthropic.com>
1556 lines
63 KiB
Python
1556 lines
63 KiB
Python
#!/usr/bin/env python3
|
||
"""pragent pilot — minimal AI PR reviewer.
|
||
|
||
Runs as a Gitea Actions step OR is called by the central webhook server
|
||
(`webhook_server.py`). Fetches a PR diff, asks glm-5.2:cloud (via the on-network
|
||
headroom proxy, Anthropic /v1/messages format) to review it, and posts the
|
||
findings back as `pragent-bot` — as a **review summary** plus **inline line
|
||
comments** with a fenced suggested-fix block (tagged with the file's language so
|
||
Gitea syntax-highlights it) where the model could produce one and the line
|
||
anchors cleanly to the post-change file.
|
||
|
||
Features (pilot v2):
|
||
- **Dedupe / persistence:** Gitea itself is the source of truth. Before
|
||
reviewing, fetch the PR's existing reviews and look for a hidden
|
||
`<!-- pragent:sha=... -->` marker matching this commit. If present, skip
|
||
(no duplicate review on label-toggle / re-fire). Prior review bodies are
|
||
fed back to the model as "already said" context so a re-push synthesizes
|
||
instead of repeating (light version of design §6.1).
|
||
- **Repo-local focus:** if the repo has a `.pr-review.json` at the PR's head
|
||
ref, its `focus` / `exclude_paths` / `instructions` / `languages` steer the
|
||
review. Optional — defaults apply when absent.
|
||
- **Inline comments + suggestions:** the model emits structured JSON
|
||
findings with `path`/`line`. We parse the diff hunks to learn which
|
||
`(path, new_line)` pairs are valid post-change anchors and post each
|
||
anchored finding as a positional review comment; the `suggestion` field, if
|
||
non-empty, is wrapped in a fenced code block tagged with the file's language
|
||
(via `_lang_for_path`) so Gitea syntax-highlights it. Gitea 1.26.x has no
|
||
GitHub-style "Apply suggestion" button, so a language-tagged block is used
|
||
for highlighting instead of a ```suggestion fence. Findings that don't
|
||
anchor (bad line, unchanged file, etc.) are folded into the summary body as
|
||
plain bullets.
|
||
|
||
Fail-open by design: any error becomes a short "review failed" review comment,
|
||
and review_pr never raises. Stdlib only — no pip install.
|
||
|
||
Env (CI run() path):
|
||
GITEA_API base URL of the in-cluster Gitea
|
||
GITEA_REPOSITORY "owner/repo" of the PR (github.repository)
|
||
PR_INDEX PR number (github.event.pull_request.number)
|
||
PR_TITLE PR title
|
||
PR_BODY PR body (optional)
|
||
PR_BASE_REF base branch (.pr-review.json is read from here, not the
|
||
PR head); optional, defaults to the repo default branch
|
||
PRAGENT_BOT_TOKEN bot access token (repo secret)
|
||
PRAGENT_SHA head SHA to tag the review
|
||
OLLAMA_URL headroom proxy URL, e.g. http://model-proxy.internal:8789
|
||
OLLAMA_MODEL model id, e.g. glm-5.2:cloud
|
||
OLLAMA_MAX_TOKENS (optional) output cap, default 8000
|
||
DIFF_MAX_CHARS (optional) diff truncation cap, default 150000
|
||
"""
|
||
|
||
import base64
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
REVIEW_HEADER = "🤖 **AI Review** · pragent pilot · {model} · `{sha}`"
|
||
# Hidden marker the dedupe pass scans for. Full sha so a re-push (new sha) is
|
||
# never mistaken for an already-reviewed commit, and a label-toggle (same sha)
|
||
# is correctly skipped.
|
||
SHA_MARKER = "<!-- pragent:sha={sha} -->"
|
||
_SHA_MARKER_RE = re.compile(r"<!-- pragent:sha=([0-9a-f]{7,40}) -->")
|
||
|
||
AI_REVIEW_LABEL = "AI-REVIEW"
|
||
SEVERITIES = ("critical", "high", "medium", "low")
|
||
# Severity rank — higher = more severe. Used by `apply_repo_config` to drop
|
||
# findings below `severity_threshold`. Critical=3, high=2, medium=1, low=0.
|
||
SEVERITY_RANK = {"low": 0, "medium": 1, "high": 2, "critical": 3}
|
||
REPO_CONFIG_FILE = ".pr-review.json"
|
||
|
||
# Style → (default max_findings, default severity_threshold). Strict is
|
||
# terse/high-signal; lenient shows everything; balanced is the default for
|
||
# unconfigured repos. Repo `.pr-review.json` overrides per-field.
|
||
STYLE_DEFAULTS: dict[str, tuple[int, str]] = {
|
||
"strict": (5, "high"),
|
||
"balanced": (12, "medium"),
|
||
"lenient": (15, "low"),
|
||
}
|
||
|
||
# Default provider to compare against in the usage section. The pilot runs on
|
||
# headroom/glm-5.2:cloud at $0/MTok, so the actual line shows $0.00 — but the
|
||
# equivalent provider line lets a maintainer see what they would have paid on
|
||
# Claude/GPT for the same measured tokens. Override with PRAGENT_PRICE_TARGET
|
||
# (env) or `.pr-review.json:cost_target` (per repo).
|
||
DEFAULT_PRICE_TARGET = "claude-sonnet-5"
|
||
|
||
SYSTEM_PROMPT = """You are a senior, pragmatic code reviewer. Review the pull request diff below.
|
||
|
||
Report ONLY real, actionable issues: correctness bugs, security problems, risky
|
||
changes, missing tests for changed behaviour, and breaking API/contract changes.
|
||
Honour any repo-specific focus / instructions given in the prompt; if focus is
|
||
given, weight those areas higher, but do not ignore a critical issue outside them.
|
||
|
||
Output STRICT JSON only — no prose, no markdown fences. Shape:
|
||
{
|
||
"findings": [
|
||
{
|
||
"severity": "critical|high|medium|low",
|
||
"path": "file path exactly as it appears in the diff (`+++ b/` side)",
|
||
"line": <int, the NEW-file line number the issue is on, within the diff>,
|
||
"problem": "one line: what is wrong",
|
||
"fix": "one line: how to fix it",
|
||
"suggestion": "<exact replacement lines for that location, or empty string if you cannot produce safe replacement code>"
|
||
}
|
||
]
|
||
}
|
||
|
||
Rules:
|
||
- `line` MUST be a line number that exists in the post-change version of `path`
|
||
(i.e. a context line or an added `+` line shown in the diff). Never a removed
|
||
line. If you are unsure of the exact line, set `line` to the closest context
|
||
line you CAN see in the diff.
|
||
- `suggestion` is the literal new code that should replace the flagged line(s).
|
||
Keep it minimal — just the changed lines, indented as they would appear in the
|
||
file. Leave it empty ("") if a safe textual replacement is not possible (e.g.
|
||
a missing test, an architectural note).
|
||
- Skip nitpicks, pure formatting, and praise. At most ~15 findings, highest
|
||
severity first.
|
||
- If the diff is clean, output: {"findings": []}
|
||
- Do NOT repeat anything already covered in "PREVIOUS REVIEWS" — only surface
|
||
new or still-unresolved issues."""
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Pure helpers (unit-tested, no network)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def truncate_diff(text: str, max_chars: int) -> tuple[str, bool, int]:
|
||
"""Return (text, was_truncated, original_len). Never raises on bad input."""
|
||
if text is None:
|
||
return "", False, 0
|
||
orig_len = len(text)
|
||
if orig_len <= max_chars:
|
||
return text, False, orig_len
|
||
return text[:max_chars] + f"\n\n[diff truncated at {max_chars} characters]\n", True, orig_len
|
||
|
||
|
||
def parse_text_blocks(content: list) -> str:
|
||
"""Join `type:"text"` blocks from an Anthropic /v1/messages response.
|
||
|
||
Drops `thinking` blocks (glm-5.2:cloud is a reasoning model and emits them).
|
||
Tolerates missing/malformed blocks by skipping them.
|
||
"""
|
||
if not isinstance(content, list):
|
||
return ""
|
||
out = []
|
||
for block in content:
|
||
if not isinstance(block, dict):
|
||
continue
|
||
if block.get("type") == "text" and isinstance(block.get("text"), str):
|
||
out.append(block["text"])
|
||
return "\n".join(out).strip()
|
||
|
||
|
||
def format_review_body(findings: str, model: str, sha: str, summary: str = "", usage_section: str = "") -> str:
|
||
"""Format the posted review summary body.
|
||
|
||
`findings` is the bullet text for findings that could NOT be anchored inline
|
||
(or, on the legacy/no-inline path, the whole review). Empty -> "No issues
|
||
found.". `summary` (optional, opencode engine) is rendered as a "Summary"
|
||
section right under the header. `usage_section` (optional, shown only when
|
||
the PR carries the `AI-USAGE` label) is rendered between the summary and the
|
||
findings bullets. The hidden sha marker is always appended for the dedupe
|
||
pass.
|
||
"""
|
||
header = REVIEW_HEADER.format(model=model, sha=sha[:8] if sha else "unknown")
|
||
findings = (findings or "").strip()
|
||
if not findings:
|
||
findings = "No issues found."
|
||
marker = SHA_MARKER.format(sha=sha) if sha else ""
|
||
parts = [header]
|
||
if summary:
|
||
parts.append(summary.strip())
|
||
if usage_section:
|
||
parts.append(usage_section.strip())
|
||
parts.append(findings)
|
||
body = "\n\n".join(parts)
|
||
if marker:
|
||
body += f"\n{marker}"
|
||
return body
|
||
|
||
|
||
def _finding_weight(f: dict) -> int:
|
||
"""Body-weight used to attribute output tokens to a finding (char length of
|
||
its rendered problem + fix + suggestion). One model pass produces all
|
||
findings, so per-finding tokens can't be measured directly — we split the
|
||
measured output total by this weight as an honest attribution."""
|
||
return (
|
||
len(f.get("problem") or "")
|
||
+ len(f.get("fix") or "")
|
||
+ len(f.get("suggestion") or "")
|
||
)
|
||
|
||
|
||
def compute_attribution(findings: list[dict], output_tokens: int) -> None:
|
||
"""Stash `_tok_attrib` (attributed output tokens) and `_tok_pct` (0..1) on
|
||
each finding, splitting `output_tokens` by each finding's body weight.
|
||
Mutates in place. No-op when there are no findings or no output budget."""
|
||
if not findings or not output_tokens:
|
||
return
|
||
weights = [_finding_weight(f) for f in findings]
|
||
total_w = sum(weights)
|
||
if total_w <= 0:
|
||
# All-zero weights (no prose): split evenly.
|
||
share = output_tokens / len(findings)
|
||
for f in findings:
|
||
f["_tok_attrib"] = int(round(share))
|
||
f["_tok_pct"] = 1.0 / len(findings)
|
||
return
|
||
for f, w in zip(findings, weights):
|
||
f["_tok_attrib"] = int(round(output_tokens * w / total_w))
|
||
f["_tok_pct"] = w / total_w
|
||
|
||
|
||
def _resolve_price_target(config: dict | None) -> tuple[str, str | None]:
|
||
"""Pick which provider to compute the equivalent cost against.
|
||
|
||
Order: `.pr-review.json:cost_target` > `PRAGENT_PRICE_TARGET` env >
|
||
`DEFAULT_PRICE_TARGET` (claude-sonnet-5). Returns `(price_key, error)`.
|
||
|
||
If any of the user-set keys is unknown, falls back to the default AND
|
||
reports the error so the operator sees their typo (a config-level typo
|
||
silently picking the default would defeat the purpose of letting repos
|
||
opt into a different comparison model).
|
||
"""
|
||
from cost_model import PRICES # local import keeps ollama path dep-free
|
||
candidates: list[tuple[str, str]] = []
|
||
if isinstance(config, dict) and config.get("cost_target"):
|
||
candidates.append(("repo config", str(config["cost_target"]).strip()))
|
||
env = os.environ.get("PRAGENT_PRICE_TARGET", "").strip()
|
||
if env:
|
||
candidates.append(("PRAGENT_PRICE_TARGET env", env))
|
||
candidates.append(("default", DEFAULT_PRICE_TARGET))
|
||
|
||
chosen = DEFAULT_PRICE_TARGET
|
||
for source, key in candidates:
|
||
if key in PRICES:
|
||
chosen = key
|
||
break
|
||
else:
|
||
# No candidate was valid. Use default + report.
|
||
return chosen, (
|
||
f"unknown price target (checked {', '.join(f'{s}={k!r}' for s, k in candidates)}); "
|
||
f"valid: {', '.join(sorted(PRICES))}"
|
||
)
|
||
|
||
# Even when we picked a valid key, if the *user* set one and it was
|
||
# unknown, surface that. (We only get here if a later candidate resolved,
|
||
# so the invalid one was upstream.)
|
||
invalid = [(s, k) for s, k in candidates if k not in PRICES and s != "default"]
|
||
if invalid:
|
||
return chosen, (
|
||
f"unknown price target (set {', '.join(f'{s}={k!r}' for s, k in invalid)}); "
|
||
f"valid: {', '.join(sorted(PRICES))}; falling back to `{chosen}`"
|
||
)
|
||
return chosen, None
|
||
|
||
|
||
def equivalent_cost(usage: dict, price_key: str) -> float:
|
||
"""USD the measured usage would have billed on `price_key`'s provider.
|
||
|
||
`usage` is the dict from `parse_opencode_events` (input/output/reasoning/
|
||
cache_read/cache_write). Builds a `cost_model.Usage` and runs `cost()`. The
|
||
pilot's actual provider (headroom/glm-5.2:cloud) reports $0 — this is what
|
||
the same tokens would cost on a paid model, so maintainers can budget.
|
||
"""
|
||
from cost_model import Usage, cost, PRICES # local import: ollama path dep-free
|
||
if price_key not in PRICES:
|
||
return 0.0
|
||
u = Usage(
|
||
uncached_input=(usage.get("input", 0) - usage.get("cache_read", 0)),
|
||
cached_input=usage.get("cache_read", 0),
|
||
cache_writes=usage.get("cache_write", 0),
|
||
output=usage.get("output", 0),
|
||
)
|
||
return cost(u, PRICES[price_key])
|
||
|
||
|
||
def format_usage_section(
|
||
usage: dict | None,
|
||
findings: list[dict],
|
||
model: str,
|
||
config: dict | None = None,
|
||
) -> str:
|
||
"""Render the `## 🔋 AI usage` block for the review body.
|
||
|
||
Only called when the PR carries the `AI-USAGE` label (and the opencode
|
||
engine produced a usage dict). Reports the MEASURED total
|
||
(input/output/reasoning/cache/cost/steps/duration) plus an ATTRIBUTED
|
||
per-finding table — one model pass generates all findings, so per-comment
|
||
counts are an estimate (output split by body weight), clearly labelled.
|
||
|
||
The cost lines show TWO numbers because the pilot runs on headroom at
|
||
$0/MTok: the `actual` line is what was billed (always $0.00 today), and
|
||
the `est. cost on <provider>` line shows what the same measured tokens
|
||
would have cost on a paid model — the number a maintainer actually cares
|
||
about when budgeting. `config["cost_target"]` / `PRAGENT_PRICE_TARGET`
|
||
/ `DEFAULT_PRICE_TARGET` (claude-sonnet-5) picks the comparison provider.
|
||
|
||
Returns "" if `usage` is None.
|
||
"""
|
||
if not usage:
|
||
return ""
|
||
dur = usage.get("duration_s")
|
||
dur_s = f"{dur}s" if dur is not None else "?"
|
||
actual = usage.get("cost") or 0.0
|
||
actual_s = f"${actual:.4f}" if actual else "$0.00"
|
||
actual_note = (
|
||
"(headroom glm-5.2:cloud — free tier)"
|
||
if not actual else "(billed by provider)"
|
||
)
|
||
price_key, price_err = _resolve_price_target(config)
|
||
from cost_model import PRICES # local import keeps ollama path dep-free
|
||
eq = equivalent_cost(usage, price_key)
|
||
eq_s = f"${eq:.4f}" if eq else "$0.00"
|
||
eq_label = PRICES[price_key].name
|
||
lines = [
|
||
"## 🔋 AI usage",
|
||
"",
|
||
f"- model: `{model}` · engine: opencode · agent steps: {usage.get('steps', 0)} · duration: {dur_s}",
|
||
(
|
||
f"- tokens: {usage.get('input', 0)} in · {usage.get('output', 0)} out · "
|
||
f"{usage.get('reasoning', 0)} reasoning · cache "
|
||
f"{usage.get('cache_read', 0)} read / {usage.get('cache_write', 0)} write "
|
||
f"→ {usage.get('total', 0)} total"
|
||
),
|
||
f"- est. cost on **{eq_label}**: {eq_s}" + (
|
||
f" _(price target: `{price_key}`; "
|
||
f"{price_err})_"
|
||
if price_err else ""
|
||
),
|
||
f"- actual: {actual_s} {actual_note}",
|
||
"- scope: whole-repo checkout at head sha (agent can read any file + run linters, not just the diff) — input tokens include files read beyond the diff",
|
||
"- per-comment tokens are *attributed* (one model pass produces all findings; output split by each finding's body weight)",
|
||
]
|
||
# Per-finding attribution table.
|
||
rows = [f for f in findings if f.get("_tok_attrib") is not None]
|
||
if rows:
|
||
lines.append("")
|
||
lines.append("| severity | location | ≈out tok | % |")
|
||
lines.append("|---|---|---:|---:|")
|
||
for f in rows:
|
||
loc = f"{f['path']}:{f['line']}" if f.get("line") else f.get("path", "?")
|
||
pct = f.get("_tok_pct", 0.0) * 100
|
||
lines.append(
|
||
f"| {f.get('severity', '').upper()} | `{loc}` | "
|
||
f"{f.get('_tok_attrib', 0)} | {pct:.0f}% |"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def build_user_prompt(
|
||
title: str,
|
||
body: str,
|
||
diff: str,
|
||
config: dict | None = None,
|
||
prior_reviews: list[str] | None = None,
|
||
) -> str:
|
||
"""Assemble the user prompt: repo config + prior reviews + PR meta + diff."""
|
||
parts: list[str] = []
|
||
|
||
eff = effective_config(config) if config else {}
|
||
if eff:
|
||
cfg_lines = []
|
||
if eff.get("focus"):
|
||
cfg_lines.append("Focus areas: " + ", ".join(eff["focus"]))
|
||
if eff.get("exclude_paths"):
|
||
cfg_lines.append("Ignore paths: " + ", ".join(eff["exclude_paths"]))
|
||
if eff.get("languages"):
|
||
cfg_lines.append("Languages: " + ", ".join(eff["languages"]))
|
||
if eff.get("style"):
|
||
cfg_lines.append(f"Review style: {eff['style']} "
|
||
f"(max {eff['max_findings']} findings, threshold "
|
||
f"{eff['severity_threshold']}+)")
|
||
if eff.get("patterns", {}).get("allow"):
|
||
cfg_lines.append("Allow paths (only these are reviewed): "
|
||
+ ", ".join(eff["patterns"]["allow"]))
|
||
if eff.get("patterns", {}).get("deny"):
|
||
cfg_lines.append("Deny paths: " + ", ".join(eff["patterns"]["deny"]))
|
||
if eff.get("exclude_tests"):
|
||
cfg_lines.append("Skip test files entirely.")
|
||
if eff.get("require_tests"):
|
||
cfg_lines.append("Flag behavioral changes that don't add a test "
|
||
"alongside (added as a `low` finding).")
|
||
if eff.get("instructions"):
|
||
cfg_lines.append("Instructions:\n" + str(eff["instructions"]).strip())
|
||
if cfg_lines:
|
||
parts.append("## Repo review config (.pr-review.json)\n" + "\n".join(cfg_lines))
|
||
|
||
if prior_reviews:
|
||
joined = "\n\n---\n\n".join(prior_reviews)
|
||
if len(joined) > 4000:
|
||
joined = joined[:4000] + "\n…[prior reviews truncated]"
|
||
parts.append("## PREVIOUS REVIEWS (already posted — do NOT repeat these points)\n" + joined)
|
||
|
||
parts.append(f"## PR\nTitle: {title or '(none)'}")
|
||
if body and body.strip():
|
||
b = body.strip()
|
||
if len(b) > 4000:
|
||
b = b[:4000] + "\n…[PR body truncated]"
|
||
parts.append(f"Description:\n{b}")
|
||
parts.append(f"## Diff\n```diff\n{diff}\n```")
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Diff parsing — find valid post-change (RIGHT-side) line anchors per file
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def parse_diff_anchors(diff: str) -> dict[str, set[int]]:
|
||
"""Parse a unified diff into {path: {new_line, ...}} for lines that exist in
|
||
the post-change version (context + added lines). Removed lines are NOT
|
||
anchors (they have no RIGHT-side line). Used to validate inline comments.
|
||
|
||
Robust to:
|
||
- `diff --git a/x b/x` and `+++ b/x` path headers (uses the `b/` side)
|
||
- hunk headers `@@ -a,b +c,d @@` (new line counter starts at c)
|
||
- No-newline-at-eof markers, binary files, missing hunks.
|
||
"""
|
||
anchors: dict[str, set[int]] = {}
|
||
current_path: str | None = None
|
||
new_line = 0
|
||
for raw in (diff or "").splitlines():
|
||
# File path: prefer the `+++ b/` line (handles renames); fall back to
|
||
# `diff --git a/x b/x`'s second path.
|
||
if raw.startswith("+++ "):
|
||
p = raw[4:].strip()
|
||
if p == "/dev/null":
|
||
current_path = None
|
||
else:
|
||
current_path = _strip_path_prefix(p)
|
||
anchors.setdefault(current_path, set())
|
||
continue
|
||
if raw.startswith("diff --git "):
|
||
# `diff --git a/foo b/foo` — take the second path as a fallback in
|
||
# case the `+++` line is missing (binary). Split on " b/".
|
||
m = re.search(r" b/(.+)$", raw)
|
||
if m:
|
||
current_path = m.group(1).strip()
|
||
anchors.setdefault(current_path, set())
|
||
continue
|
||
if raw.startswith("@@"):
|
||
m = re.search(r"\+(\d+)(?:,\d+)?\s@@", raw)
|
||
new_line = int(m.group(1)) if m else 0
|
||
continue
|
||
if current_path is None:
|
||
continue
|
||
if raw.startswith("\\ No newline"):
|
||
continue
|
||
if raw.startswith("-"):
|
||
# removed line — no RIGHT-side anchor
|
||
continue
|
||
if raw.startswith("+"):
|
||
anchors[current_path].add(new_line)
|
||
new_line += 1
|
||
continue
|
||
# Context line: normally " text", but an empty context line arrives as
|
||
# "" whenever something along the way stripped trailing whitespace (some
|
||
# forges, some patch tools, copy/paste). Treating "" as "not a line"
|
||
# would desync `new_line` for the whole rest of the hunk and silently
|
||
# misplace every later inline comment in the file, so count it.
|
||
if raw.startswith(" ") or raw == "":
|
||
anchors[current_path].add(new_line)
|
||
new_line += 1
|
||
return anchors
|
||
|
||
|
||
def _strip_path_prefix(p: str) -> str:
|
||
"""`b/foo` or `foo` -> `foo`."""
|
||
if p.startswith("b/"):
|
||
return p[2:]
|
||
return p
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Model output parsing — tolerant JSON findings extraction
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _normalize_finding(f: dict) -> dict | None:
|
||
"""Validate + normalize one raw finding dict. Returns None if it's unusable
|
||
(missing path/line). Normalises severity, keeps `reference` (default "")."""
|
||
if not isinstance(f, dict):
|
||
return None
|
||
path = f.get("path")
|
||
line = f.get("line")
|
||
if not isinstance(path, str) or not path.strip():
|
||
return None
|
||
if not isinstance(line, int) or line < 1:
|
||
return None
|
||
sev = str(f.get("severity", "medium")).strip().lower()
|
||
if sev not in SEVERITIES:
|
||
sev = "medium"
|
||
reference = str(f.get("reference", "") or "").strip()
|
||
return {
|
||
"severity": sev,
|
||
"path": path.strip(),
|
||
"line": line,
|
||
"problem": str(f.get("problem", "")).strip(),
|
||
"fix": str(f.get("fix", "")).strip(),
|
||
"suggestion": str(f.get("suggestion", "") or "").strip(),
|
||
"reference": reference,
|
||
}
|
||
|
||
|
||
def _last_json_block(text: str) -> str | None:
|
||
r"""Return the substring of the last JSON object/array in text, or None.
|
||
|
||
The pragent agent emits ```json fences around its final block, but real
|
||
outputs drift:
|
||
* the fence contains nested objects (regex ``\{.*?\}`` only matches the
|
||
first ``}``, truncating the JSON — the parser then sees
|
||
``json.JSONDecodeError``);
|
||
* the fence is missing or unterminated, but a balanced JSON object sits
|
||
in the prose tail;
|
||
* the agent emits a bare array (findings only, no summary wrapper).
|
||
|
||
Strategy:
|
||
1. Find each fenced block, take the last. Inside it, walk a balanced
|
||
``{...}``/``[...]`` scanner (not a regex) so nested structures survive.
|
||
2. Fall back to a balanced scanner over the whole text, picking the LAST
|
||
balanced object/array (the agent writes its conclusion last).
|
||
"""
|
||
s = text or ""
|
||
if not s:
|
||
return None
|
||
# 1. Fenced blocks: take the last ```json ... ``` or ``` ... ``` region.
|
||
fences = list(re.finditer(r"```(?:json)?\n", s))
|
||
for m in reversed(fences):
|
||
start = m.end()
|
||
# Find the matching closing fence.
|
||
end = s.find("```", start)
|
||
if end < 0:
|
||
# Unterminated fence — try to salvage the balanced object inside.
|
||
end = len(s)
|
||
inner = s[start:end].strip()
|
||
obj = _balanced_json_substring(inner)
|
||
if obj is not None:
|
||
return obj
|
||
# 2. No (parseable) fence — scan the whole text for the LAST balanced
|
||
# object/array. The agent's conclusion is at the tail.
|
||
return _last_balanced_json(s)
|
||
|
||
|
||
def parse_findings(text: str) -> list[dict]:
|
||
"""Parse the model's JSON response into a list of finding dicts.
|
||
|
||
Tolerant: strips ```json fences, and if the model wrapped JSON in prose,
|
||
scans for the first balanced `{...}` and extracts its `findings` array.
|
||
Drops findings missing path/line or with an unknown severity (normalised).
|
||
Never raises — returns [] on any parse failure.
|
||
|
||
Also accepts a bare JSON array as the outer value: ``[{...}, {...}]`` —
|
||
some agents skip the ``{"summary":..., "findings":[...]}`` wrapper.
|
||
"""
|
||
data = _parse_json_tolerant(text)
|
||
if isinstance(data, dict):
|
||
findings = data.get("findings")
|
||
elif isinstance(data, list):
|
||
findings = data
|
||
else:
|
||
return []
|
||
if not isinstance(findings, list):
|
||
return []
|
||
out = []
|
||
for f in findings:
|
||
n = _normalize_finding(f)
|
||
if n is not None:
|
||
out.append(n)
|
||
return out
|
||
|
||
|
||
SALVAGE_MAX_CHARS = 4000
|
||
|
||
|
||
def salvage_summary(text: str, max_chars: int = SALVAGE_MAX_CHARS) -> str:
|
||
"""Recover something postable from agent output we could not parse.
|
||
|
||
An opencode run costs minutes and millions of tokens. When the findings JSON
|
||
is missing or malformed, the analysis itself is usually still there in the
|
||
prose — discarding it to post "no parseable output" throws away the whole
|
||
run and tells the maintainer nothing. This keeps the tail of the prose (the
|
||
conclusion, which is what the agent writes last), drops fenced code blocks
|
||
so a half-written JSON blob doesn't dominate, and labels it plainly as
|
||
unstructured so nobody mistakes it for a normal review.
|
||
|
||
Returns "" when there is genuinely nothing to salvage.
|
||
"""
|
||
if not text or not text.strip():
|
||
return ""
|
||
# Drop fenced blocks — a truncated ```json block is noise here.
|
||
prose = re.sub(r"```.*?```", "", text, flags=re.DOTALL)
|
||
prose = re.sub(r"```.*$", "", prose, flags=re.DOTALL) # unterminated fence
|
||
prose = prose.strip()
|
||
if not prose:
|
||
return ""
|
||
if len(prose) > max_chars:
|
||
prose = "…" + prose[-max_chars:]
|
||
return (
|
||
"⚠️ _The reviewer did not emit a parseable findings block, so there are "
|
||
"no inline comments. Its raw notes are below — treat them as unverified: "
|
||
"line numbers were not validated against the diff._\n\n" + prose
|
||
)
|
||
|
||
|
||
def parse_review_output(text: str) -> tuple[str, list[dict]]:
|
||
"""Parse the opengine's stdout into (summary, findings).
|
||
|
||
Accepts `{"summary": "...", "findings": [...]}` (the opencode pragent agent),
|
||
`{"findings": [...]}`, or a bare `[...]` of finding dicts. `summary` defaults
|
||
to "". Uses the LAST fenced block (the pragent agent emits JSON as the final
|
||
block), with a tolerant fallback that scans for the last balanced
|
||
object/array in the prose tail. Never raises.
|
||
"""
|
||
blob = _last_json_block(text)
|
||
if blob is None:
|
||
return "", []
|
||
try:
|
||
data = json.loads(blob)
|
||
except json.JSONDecodeError:
|
||
return "", []
|
||
if isinstance(data, dict):
|
||
summary = str(data.get("summary", "") or "").strip()
|
||
findings = data.get("findings")
|
||
elif isinstance(data, list):
|
||
# Bare array: each item is a finding; no summary.
|
||
summary = ""
|
||
findings = data
|
||
else:
|
||
return "", []
|
||
out = []
|
||
if isinstance(findings, list):
|
||
for f in findings:
|
||
n = _normalize_finding(f)
|
||
if n is not None:
|
||
out.append(n)
|
||
return summary, out
|
||
|
||
|
||
def _parse_json_tolerant(text: str) -> dict | list | None:
|
||
"""Parse a JSON object/array from text: try the last fenced block, then a
|
||
direct parse, then the first balanced object. Returns None on any failure.
|
||
Accepts both ``{...}`` (the pragent schema) and bare ``[...]`` arrays
|
||
(agents that skip the wrapper)."""
|
||
if not text:
|
||
return None
|
||
blob = _last_json_block(text)
|
||
if blob is not None:
|
||
try:
|
||
d = json.loads(blob)
|
||
if isinstance(d, (dict, list)):
|
||
return d
|
||
except json.JSONDecodeError:
|
||
pass
|
||
s = text.strip()
|
||
if s.startswith("```"):
|
||
s = re.sub(r"^```[a-zA-Z]*\n?", "", s)
|
||
s = re.sub(r"\n?```$", "", s).strip()
|
||
try:
|
||
d = json.loads(s)
|
||
if isinstance(d, (dict, list)):
|
||
return d
|
||
except json.JSONDecodeError:
|
||
pass
|
||
obj = _extract_first_json_object(text)
|
||
if obj is not None:
|
||
try:
|
||
d = json.loads(obj)
|
||
if isinstance(d, (dict, list)):
|
||
return d
|
||
except json.JSONDecodeError:
|
||
pass
|
||
# Last resort: the JSON lives at the tail of the prose with no fence.
|
||
# Walk the whole text for the last balanced object/array.
|
||
last = _last_balanced_json(text)
|
||
if last is not None:
|
||
try:
|
||
d = json.loads(last)
|
||
if isinstance(d, (dict, list)):
|
||
return d
|
||
except json.JSONDecodeError:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _extract_first_json_object(s: str) -> str | None:
|
||
"""Return the substring of the first balanced top-level `{ ... }` in s."""
|
||
start = s.find("{")
|
||
if start < 0:
|
||
return None
|
||
end = _scan_balanced(s, start, "{", "}")
|
||
if end is None:
|
||
return None
|
||
return s[start:end + 1]
|
||
|
||
|
||
def _last_balanced_json(s: str) -> str | None:
|
||
"""Return the substring of the LAST balanced ``{...}`` or ``[...]`` in s.
|
||
|
||
Used when the agent emits no fence: the JSON lives in the prose tail.
|
||
Picks whichever closer (object or array) appears latest in the text.
|
||
"""
|
||
if not s:
|
||
return None
|
||
last_obj = _find_last_close(s, "{", "}")
|
||
last_arr = _find_last_close(s, "[", "]")
|
||
candidates = []
|
||
if last_obj is not None:
|
||
candidates.append(last_obj)
|
||
if last_arr is not None:
|
||
candidates.append(last_arr)
|
||
if not candidates:
|
||
return None
|
||
end, opener, start = max(candidates, key=lambda t: t[0])
|
||
return s[start:end + 1]
|
||
|
||
|
||
def _balanced_json_substring(s: str) -> str | None:
|
||
"""Return the first balanced ``{...}`` or ``[...]`` substring in ``s``.
|
||
|
||
Skips past leading whitespace/non-JSON and returns the full balanced
|
||
extent (handles nested objects/arrays and string literals with braces).
|
||
"""
|
||
if not s:
|
||
return None
|
||
# Try object first; the pragent schema is an object on the outer level.
|
||
for i, c in enumerate(s):
|
||
if c == "{":
|
||
end = _scan_balanced(s, i, "{", "}")
|
||
if end is not None:
|
||
return s[i:end + 1]
|
||
break
|
||
if c == "[":
|
||
end = _scan_balanced(s, i, "[", "]")
|
||
if end is not None:
|
||
return s[i:end + 1]
|
||
break
|
||
return None
|
||
|
||
|
||
def _scan_balanced(s: str, start: int, opener: str, closer: str) -> int | None:
|
||
"""Return the index of the matching ``closer`` for ``s[start] == opener``.
|
||
|
||
Tracks string literals (with ``\\`` escapes) so braces inside strings don't
|
||
fool the depth counter. Returns None if no balance is reached.
|
||
"""
|
||
depth = 0
|
||
in_str = False
|
||
esc = False
|
||
for i in range(start, len(s)):
|
||
c = s[i]
|
||
if in_str:
|
||
if esc:
|
||
esc = False
|
||
elif c == "\\":
|
||
esc = True
|
||
elif c == '"':
|
||
in_str = False
|
||
continue
|
||
if c == '"':
|
||
in_str = True
|
||
elif c == opener:
|
||
depth += 1
|
||
elif c == closer:
|
||
depth -= 1
|
||
if depth == 0:
|
||
return i
|
||
return None
|
||
|
||
|
||
def _find_last_close(s: str, opener: str, closer: str) -> tuple[int, str, int] | None:
|
||
"""Walk ``s`` backwards from the last ``closer`` to find its matching opener.
|
||
|
||
Returns ``(close_idx, opener_char, open_idx)`` for the rightmost balanced
|
||
structure, or None if no pair exists.
|
||
"""
|
||
# Find the last `closer` candidate.
|
||
last = s.rfind(closer)
|
||
while last >= 0:
|
||
# Walk left, tracking depth from the perspective of the opener.
|
||
depth = 1
|
||
in_str = False
|
||
esc = False
|
||
for j in range(last - 1, -1, -1):
|
||
c = s[j]
|
||
if in_str:
|
||
if esc:
|
||
esc = False
|
||
elif c == "\\":
|
||
esc = True
|
||
elif c == '"':
|
||
in_str = False
|
||
continue
|
||
if c == '"':
|
||
# Approximation: we don't track quotes perfectly walking
|
||
# backwards, but strings in agent output are short and rare.
|
||
in_str = not in_str
|
||
elif c == closer:
|
||
depth += 1
|
||
elif c == opener:
|
||
depth -= 1
|
||
if depth == 0:
|
||
return (last, opener, j)
|
||
last = s.rfind(closer, 0, last)
|
||
return None
|
||
|
||
|
||
def split_findings(findings: list[dict], anchors: dict[str, set[int]]) -> tuple[list[dict], list[dict]]:
|
||
"""Split findings into (anchored, unanchored).
|
||
|
||
A finding is anchored if its path is known AND its line is a valid post-change
|
||
line for that path. Lines just outside the diff (model off-by-one) are NOT
|
||
anchored — safer to keep them as summary bullets than to drop or misplace.
|
||
"""
|
||
anchored, unanchored = [], []
|
||
for f in findings:
|
||
valid = anchors.get(f["path"])
|
||
if valid and f["line"] in valid:
|
||
anchored.append(f)
|
||
else:
|
||
unanchored.append(f)
|
||
return anchored, unanchored
|
||
|
||
|
||
def _lang_for_path(path: str) -> str:
|
||
"""Map a file extension to a chroma language tag for fenced code blocks.
|
||
|
||
Used so the suggested-fix block is syntax-highlighted in Gitea. Gitea 1.26.x
|
||
has no GitHub-style "Apply suggestion" button (the ```suggestion fence is
|
||
just an unknown-language code block → plain monospace, no apply), so we tag
|
||
the block with the file's real language for highlighting instead.
|
||
"""
|
||
ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
|
||
return {
|
||
"java": "java", "kt": "kotlin", "scala": "scala", "groovy": "groovy",
|
||
"ts": "typescript", "tsx": "tsx", "js": "javascript", "jsx": "jsx",
|
||
"mjs": "javascript", "cjs": "javascript",
|
||
"py": "python", "pyi": "python",
|
||
"go": "go", "rs": "rust", "rb": "ruby", "php": "php",
|
||
"c": "c", "h": "c", "cpp": "cpp", "cc": "cpp", "hpp": "cpp",
|
||
"cs": "csharp", "swift": "swift", "m": "objc",
|
||
"sh": "bash", "bash": "bash", "zsh": "bash",
|
||
"yml": "yaml", "yaml": "yaml", "json": "json", "jsonc": "json",
|
||
"toml": "toml", "ini": "ini", "cfg": "ini",
|
||
"html": "html", "htm": "html", "css": "css", "scss": "scss",
|
||
"xml": "xml", "svg": "xml", "sql": "sql",
|
||
"md": "markdown", "dockerfile": "dockerfile",
|
||
}.get(ext, "")
|
||
|
||
|
||
def inline_comment_body(f: dict) -> str:
|
||
"""Render one finding as a positional review-comment body.
|
||
|
||
Includes a fenced suggested-fix block only if the model produced non-empty
|
||
replacement code. The fence is tagged with the file's language (via
|
||
`_lang_for_path`) so Gitea syntax-highlights it — Gitea 1.26.x has no
|
||
GitHub-style "Apply suggestion" button (```suggestion is just an
|
||
unknown-language block there → plain monospace), so a language-tagged block
|
||
is strictly more readable and loses nothing. Appends a `📎 ref:` link when
|
||
the finding carries a `reference` URL.
|
||
"""
|
||
sev = f["severity"].upper()
|
||
body = f"**[{sev}]** {f['problem']}"
|
||
if f["fix"]:
|
||
body += f"\n\nFix: {f['fix']}"
|
||
if f["suggestion"]:
|
||
lang = _lang_for_path(f.get("path", ""))
|
||
fence = f"```{lang}" if lang else "```"
|
||
body += f"\n\n{fence}\n{f['suggestion']}\n```"
|
||
ref = f.get("reference", "")
|
||
if ref:
|
||
body += f"\n\n📎 ref: {ref}"
|
||
tok = f.get("_tok_attrib")
|
||
if tok is not None:
|
||
pct = (f.get("_tok_pct", 0.0) or 0.0) * 100
|
||
body += f"\n\n🪙 ~{tok} tok ({pct:.0f}% · attributed output)"
|
||
return body
|
||
|
||
|
||
def summary_bullets(findings: list[dict]) -> str:
|
||
"""Render unanchored findings as summary-body bullets (no line anchor)."""
|
||
lines = []
|
||
for f in findings:
|
||
loc = f"{f['path']}:{f['line']}" if f["line"] else f["path"]
|
||
fix = f" — fix: {f['fix']}" if f["fix"] else ""
|
||
ref = f" ({f.get('reference', '')})" if f.get("reference") else ""
|
||
lines.append(f"- **[{f['severity'].upper()}]** `{loc}` — {f['problem']}{fix}{ref}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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
|
||
|
||
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
|
||
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
|
||
cost_target <key of cost_model.PRICES> — see equivalent_cost
|
||
"""
|
||
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]
|
||
|
||
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()
|
||
|
||
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 _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 = _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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Network helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _http(method: str, url: str, token: str, body: dict | None = None, accept: str = "application/json") -> tuple[int, bytes]:
|
||
headers = {"Authorization": f"token {token}", "Accept": accept}
|
||
data = None
|
||
if body is not None:
|
||
data = json.dumps(body).encode()
|
||
headers["Content-Type"] = "application/json"
|
||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=180) as r:
|
||
return r.status, r.read()
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, e.read()
|
||
except urllib.error.URLError as e:
|
||
raise RuntimeError(f"network error: {e.reason}") from e
|
||
|
||
|
||
def gitea_get(api: str, repo: str, path: str, token: str, accept: str = "application/json") -> tuple[int, bytes]:
|
||
return _http("GET", f"{api}/api/v1/repos/{repo}/{path}", token, None, accept)
|
||
|
||
|
||
def gitea_post(api: str, repo: str, path: str, token: str, body: dict) -> tuple[int, bytes]:
|
||
return _http("POST", f"{api}/api/v1/repos/{repo}/{path}", token, body)
|
||
|
||
|
||
def fetch_pr_diff(api: str, repo: str, index: str, token: str, max_chars: int) -> tuple[str, bool, int]:
|
||
"""Get the unified diff. Try the `.diff` suffix first, fall back to the
|
||
files endpoint (join `patch` fields) if the server does not serve .diff."""
|
||
diff_status, raw = gitea_get(api, repo, f"pulls/{index}.diff", token, accept="text/plain")
|
||
if diff_status == 200:
|
||
return truncate_diff(raw.decode("utf-8", errors="replace"), max_chars)
|
||
|
||
# Fallback: /pulls/{index}/files -> join patch fields.
|
||
files_status, raw = gitea_get(api, repo, f"pulls/{index}/files", token)
|
||
if files_status != 200:
|
||
raise RuntimeError(
|
||
f"could not fetch diff: .diff={diff_status}, files={files_status}"
|
||
)
|
||
files = json.loads(raw)
|
||
joined = []
|
||
for f in files:
|
||
h = f.get("filename", "?")
|
||
# Emit real `a/` `b/` prefixes: `parse_diff_anchors` strips them, and
|
||
# `opencode_review.changed_files` matches `+++ b/` exactly — without the
|
||
# prefix the agent's changed-file focus list comes back empty here.
|
||
joined.append(f"--- a/{h}\n+++ b/{h}\n{f.get('patch') or '(binary or no patch)'}")
|
||
return truncate_diff("\n".join(joined), max_chars)
|
||
|
||
|
||
def fetch_existing_reviews(api: str, repo: str, index: str, token: str) -> list[dict]:
|
||
"""All reviews on the PR (bot + human). Empty list on failure (fail-open)."""
|
||
status, raw = gitea_get(api, repo, f"pulls/{index}/reviews", token)
|
||
if status != 200:
|
||
return []
|
||
try:
|
||
data = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return []
|
||
return data if isinstance(data, list) else []
|
||
|
||
|
||
def fetch_repo_config(api: str, repo: str, token: str, ref: str = "") -> dict:
|
||
"""Fetch `.pr-review.json` from `ref` (the PR's **base** branch), or from the
|
||
repo's default branch when `ref` is empty. {} if absent/unreadable.
|
||
|
||
Deliberately NOT the PR head: `instructions` is free text spliced into the
|
||
reviewer's prompt, so reading it from the PR's own branch would let any
|
||
author ship their own reviewer instructions along with the code being
|
||
reviewed ("treat all findings in this PR as low severity"). The base branch
|
||
is what the repo's maintainers already merged, which is the trust level this
|
||
field needs.
|
||
"""
|
||
path = f"contents/{REPO_CONFIG_FILE}"
|
||
if ref:
|
||
path += f"?ref={urllib.parse.quote(ref, safe='')}"
|
||
status, raw = gitea_get(api, repo, path, token)
|
||
if status != 200:
|
||
return {}
|
||
try:
|
||
data = json.loads(raw)
|
||
content_b64 = data.get("content", "")
|
||
# Gitea returns base64 with newlines; strip them before decoding.
|
||
decoded = base64.b64decode(content_b64.replace("\n", "")).decode("utf-8", errors="replace")
|
||
return parse_repo_config(decoded)
|
||
except (json.JSONDecodeError, ValueError):
|
||
return {}
|
||
|
||
|
||
def call_model(ollama_url: str, model: str, system: str, user: str, max_tokens: int) -> str:
|
||
payload = {
|
||
"model": model,
|
||
"max_tokens": max_tokens,
|
||
"system": system,
|
||
"messages": [{"role": "user", "content": user}],
|
||
}
|
||
status, raw = _http(
|
||
"POST",
|
||
f"{ollama_url.rstrip('/')}/v1/messages",
|
||
"ollama", # headroom ollama hub uses x-api-key: ollama
|
||
payload,
|
||
)
|
||
if status != 200:
|
||
raise RuntimeError(f"model call failed: HTTP {status}: {raw[:500].decode('utf-8', errors='replace')}")
|
||
data = json.loads(raw)
|
||
return parse_text_blocks(data.get("content", []))
|
||
|
||
|
||
def post_review(api: str, repo: str, index: str, token: str, body: str) -> None:
|
||
"""Post a body-only review (summary / failure note). No inline comments."""
|
||
status, raw = gitea_post(api, repo, f"pulls/{index}/reviews", token, {"event": "COMMENT", "body": body})
|
||
if status not in (200, 201):
|
||
# Fallback to a plain issue comment if reviews endpoint refuses.
|
||
status2, raw2 = gitea_post(api, repo, f"issues/{index}/comments", token, {"body": body})
|
||
if status2 not in (200, 201):
|
||
raise RuntimeError(f"post review failed: reviews={status}, comments={status2}")
|
||
|
||
|
||
def post_inline_review(
|
||
api: str, repo: str, index: str, token: str, summary: str, anchored: list[dict]
|
||
) -> None:
|
||
"""Post a review with a summary body AND positional inline comments.
|
||
|
||
Each anchored finding becomes one entry in `comments`. Gitea 1.26.x anchors
|
||
inline review comments with `new_position` (the line in the POST-change file)
|
||
+ `old_position: 0` — the `line`/`side` fields used by newer Gitea are NOT
|
||
honored here and silently leave the comment unpositioned (Gitea then renders
|
||
a file-level comment on EVERY diff line of the file, which is the flood we
|
||
hit). `f["line"]` is already a validated post-change (RIGHT-side) line from
|
||
`split_findings`, so it maps directly to `new_position`. The body carries a
|
||
language-tagged fenced code block when the model produced replacement code.
|
||
"""
|
||
comments = [
|
||
{
|
||
"path": f["path"],
|
||
"new_position": f["line"],
|
||
"old_position": 0,
|
||
"body": inline_comment_body(f),
|
||
}
|
||
for f in anchored
|
||
]
|
||
payload = {"event": "COMMENT", "body": summary, "comments": comments}
|
||
status, raw = gitea_post(api, repo, f"pulls/{index}/reviews", token, payload)
|
||
if status in (200, 201):
|
||
return
|
||
# If the inline post failed (e.g. a bad line slipped through), retry as a
|
||
# body-only review — but fold the anchored findings into the body as bullets
|
||
# first. Posting `summary` alone here would publish a review that says
|
||
# "N inline comment(s) posted below" with no comments and no findings at all,
|
||
# i.e. every finding silently lost on the one path where that matters most.
|
||
degraded = summary
|
||
if anchored:
|
||
degraded += (
|
||
"\n\n_Inline anchoring failed (Gitea returned "
|
||
f"{status}); findings listed here instead:_\n\n"
|
||
+ summary_bullets(anchored)
|
||
)
|
||
post_review(api, repo, index, token, degraded)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Main
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _need(name: str) -> str:
|
||
v = os.environ.get(name)
|
||
if not v:
|
||
raise RuntimeError(f"missing env {name}")
|
||
return v
|
||
|
||
|
||
def review_pr(
|
||
api: str,
|
||
repo: str,
|
||
index: str,
|
||
title: str,
|
||
body: str,
|
||
sha: str,
|
||
token: str,
|
||
ollama_url: str,
|
||
model: str,
|
||
max_tokens: int = 8000,
|
||
max_chars: int = 150000,
|
||
report_usage: bool = False,
|
||
base_ref: str = "",
|
||
) -> bool:
|
||
"""Run one review and post it as `pragent-bot`.
|
||
|
||
Dedupe: if a prior review already carries this commit's sha marker, skip
|
||
(no duplicate). Otherwise: fetch repo config + prior-review context, call
|
||
the model, parse JSON findings, anchor what we can to diff lines, post a
|
||
review with inline comments + suggestions (unanchored findings → summary
|
||
bullets).
|
||
|
||
`base_ref`: the PR's base branch. `.pr-review.json` is read from there (not
|
||
from the PR head) so a PR cannot ship its own reviewer instructions; empty
|
||
means "the repo's default branch".
|
||
|
||
`report_usage`: when True (PR carries the `AI-USAGE` label), the opencode
|
||
engine's measured token/cost usage is rendered as a `## 🔋 AI usage` section
|
||
on the review body and an attributed `🪙 ~N tok` line on each inline
|
||
comment. No-op on the ollama fallback (no usage available).
|
||
|
||
Returns True on success (including a deliberate skip), False on failure
|
||
(failure note posted when possible). Never raises — fail-open by design.
|
||
Both the CI `run()` entry point and the central webhook server call this.
|
||
"""
|
||
try:
|
||
reviews = fetch_existing_reviews(api, repo, index, token)
|
||
# Dedupe: already reviewed this exact commit -> nothing to do.
|
||
if sha and sha in reviewed_shas(reviews):
|
||
print(f"pragent: {repo}#{index} sha={sha[:8]} already reviewed, skipping", flush=True)
|
||
return True
|
||
|
||
raw_diff, _truncated, _orig = fetch_pr_diff(api, repo, index, token, max_chars)
|
||
if not raw_diff.strip():
|
||
post_review(api, repo, index, token, format_review_body("No diff content to review.", model, sha))
|
||
return True
|
||
|
||
config = fetch_repo_config(api, repo, token, ref=base_ref)
|
||
prior = compact_prior_reviews(prior_review_bodies(reviews, sha))
|
||
|
||
# Trim the diff to +/- hunks plus a narrow context window. The agent
|
||
# resends the brief prefix every step, so a 25k-char diff becomes
|
||
# 25k × 30-step × cached-after-step-1 = hundreds of thousands of input
|
||
# tokens. Default context=1: enough for the reviewer to see what an
|
||
# added line is replacing; the full file is on disk in the workdir
|
||
# anyway, so anything more is reading the diff twice. Tunable via
|
||
# PRAGENT_DIFF_CONTEXT (0 = +/- only; -1 = disable compression).
|
||
from diff_compress import compress_diff
|
||
ctx = int(os.environ.get("PRAGENT_DIFF_CONTEXT", "1"))
|
||
if ctx < 0:
|
||
diff = raw_diff
|
||
compression_note = ""
|
||
else:
|
||
diff, orig_chars, kept_chars = compress_diff(raw_diff, context=ctx)
|
||
if kept_chars < orig_chars:
|
||
compression_note = (
|
||
f"\n\n> _diff compressed: {orig_chars:,} → {kept_chars:,} chars "
|
||
f"(context={ctx}; PRAGENT_DIFF_CONTEXT to tune)_"
|
||
)
|
||
else:
|
||
compression_note = ""
|
||
|
||
engine = os.environ.get("PRAGENT_ENGINE", "opencode").strip().lower()
|
||
review_summary = ""
|
||
if engine == "opencode":
|
||
# The review "brain" runs on opencode: it gets the checked-out repo,
|
||
# the brief, and the pragent agent factory; returns stdout with a
|
||
# summary + findings JSON. We parse + anchor + post here.
|
||
import opencode_review # local import keeps the ollama path dep-free
|
||
# opencode wants a provider-prefixed model ref (headroom/glm-5.2:cloud);
|
||
# `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,
|
||
)
|
||
review_summary, findings = parse_review_output(stdout)
|
||
if not findings and not review_summary:
|
||
# The findings JSON was missing or malformed. Don't discard the
|
||
# run: salvage the prose, keep the usage report (the label asked
|
||
# for it, and the tokens were spent either way), and log enough
|
||
# of the raw output to diagnose why the agent went off-format.
|
||
print(
|
||
f"pragent: {repo}#{index} sha={sha[:8]} unparseable output "
|
||
f"({len(stdout)} chars); tail: {stdout[-600:]!r}",
|
||
file=sys.stderr, flush=True,
|
||
)
|
||
salvaged = salvage_summary(stdout)
|
||
usage_section = ""
|
||
if report_usage and usage:
|
||
usage_section = format_usage_section(usage, [], model, config=config)
|
||
post_review(api, repo, index, token, format_review_body(
|
||
salvaged or "AI review produced no parseable output.",
|
||
model, sha, usage_section=usage_section))
|
||
return True
|
||
else:
|
||
user_prompt = build_user_prompt(title, body + compression_note, diff, config, prior)
|
||
raw_findings = call_model(ollama_url, model, SYSTEM_PROMPT, user_prompt, max_tokens)
|
||
findings = parse_findings(raw_findings)
|
||
usage = None
|
||
|
||
# Filter / cap findings per `.pr-review.json` (style, threshold, max,
|
||
# patterns, exclude_tests). Without this every config knob would be a
|
||
# no-op — the agent has no view into the config beyond instructions.
|
||
# The synthetic require_tests finding (if any) is appended here.
|
||
try:
|
||
changed_paths = sorted({
|
||
f.get("path", "")
|
||
for f in findings
|
||
if f.get("path")
|
||
})
|
||
except Exception:
|
||
changed_paths = []
|
||
kept, _dropped = apply_repo_config(findings, config, changed_paths=changed_paths)
|
||
findings = kept
|
||
if _dropped:
|
||
print(
|
||
f"pragent: {repo}#{index} sha={sha[:8]} filtered "
|
||
f"{len(_dropped)} finding(s) per .pr-review.json "
|
||
f"(style={(config or {}).get('style', 'balanced')}, "
|
||
f"threshold={(config or {}).get('severity_threshold', '?')}, "
|
||
f"max={len(findings)})",
|
||
flush=True,
|
||
)
|
||
|
||
# Attribute output tokens to each finding (mutates finding dicts) so
|
||
# inline comments + the usage table can show a per-comment estimate.
|
||
# Only meaningful when we have measured usage AND the PR asked for it.
|
||
usage_section = ""
|
||
if report_usage and usage and usage.get("output"):
|
||
compute_attribution(findings, usage["output"])
|
||
usage_section = format_usage_section(usage, findings, model, config=config)
|
||
|
||
anchors = parse_diff_anchors(diff)
|
||
anchored, unanchored = split_findings(findings, anchors)
|
||
|
||
# Summary body: the unanchored bullets (or "No issues found."), plus a
|
||
# one-line note when inline comments were posted so the summary isn't
|
||
# empty-looking. The opencode engine also carries a prose summary.
|
||
bullets = summary_bullets(unanchored)
|
||
summary_parts = []
|
||
if anchored:
|
||
summary_parts.append(f"_{len(anchored)} inline comment(s) posted below._")
|
||
if bullets:
|
||
summary_parts.append(bullets)
|
||
if not summary_parts:
|
||
summary_parts.append("No issues found.")
|
||
summary_body = format_review_body(
|
||
"\n\n".join(summary_parts), model, sha,
|
||
summary=review_summary, usage_section=usage_section,
|
||
)
|
||
|
||
post_inline_review(api, repo, index, token, summary_body, anchored)
|
||
print(
|
||
f"pragent: reviewed {repo}#{index} sha={sha[:8]} "
|
||
f"engine={engine} findings={len(findings)} inline={len(anchored)}",
|
||
flush=True,
|
||
)
|
||
return True
|
||
except Exception as e: # fail-open
|
||
try:
|
||
post_review(api, repo, index, token, format_review_body(f"⚠️ AI review failed: {e}", model, sha))
|
||
except Exception as e2:
|
||
print(f"pragent: could not post failure note: {e2}", file=sys.stderr)
|
||
print(f"pragent: review failed: {e}", file=sys.stderr)
|
||
return False
|
||
|
||
|
||
def run() -> int:
|
||
review_pr(
|
||
api=_need("GITEA_API"),
|
||
repo=_need("GITEA_REPOSITORY"),
|
||
index=_need("PR_INDEX"),
|
||
title=os.environ.get("PR_TITLE", ""),
|
||
body=os.environ.get("PR_BODY", ""),
|
||
sha=os.environ.get("PRAGENT_SHA", ""),
|
||
token=_need("PRAGENT_BOT_TOKEN"),
|
||
ollama_url=_need("OLLAMA_URL"),
|
||
model=_need("OLLAMA_MODEL"),
|
||
max_tokens=int(os.environ.get("OLLAMA_MAX_TOKENS", "8000")),
|
||
max_chars=int(os.environ.get("DIFF_MAX_CHARS", "150000")),
|
||
base_ref=os.environ.get("PR_BASE_REF", ""),
|
||
)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(run()) |