claude
2232 lines
92 KiB
Python
2232 lines
92 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}` · Merge confidence: {confidence}"
|
||
# 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}) -->")
|
||
|
||
SEVERITIES = ("critical", "high", "medium", "low", "trivial", "info")
|
||
# Severity rank — higher = more severe. Used by `apply_repo_config` to drop
|
||
# findings below `severity_threshold`. critical=4, high=3, medium=2, low=1,
|
||
# trivial=0, info=-1.
|
||
SEVERITY_RANK = {"info": -1, "trivial": 0, "low": 1, "medium": 2, "high": 3, "critical": 4}
|
||
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"
|
||
|
||
# Default roster of paid providers shown in the equivalent-cost table when
|
||
# `.pr-review.json` does not pin `compare_against`. The pilot is free-tier only,
|
||
# so this list is the operator's budgeting signal — it answers "what would this
|
||
# have cost on a mainstream paid API?". Override per-repo via
|
||
# `.pr-review.json:compare_against` (capped at 12 entries; unknown keys are
|
||
# dropped with a stderr line at parse time).
|
||
DEFAULT_COMPARE_AGAINST = ("claude-sonnet-5", "gpt-5", "gemini-2.5-pro", "grok-4.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|trivial|info",
|
||
"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>"
|
||
}
|
||
],
|
||
"walkthrough": ["2-6 short bullets, file- or change-grouped, plain prose"],
|
||
"risk_verdict": "Low|Medium|High|Critical risk: <one-line concrete reason>",
|
||
"test_coverage": "Tests added" | "Tests changed" | "No tests for behavioral change" | "No test files in repo"
|
||
}
|
||
|
||
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).
|
||
- `walkthrough`: 2-6 short bullets, file- or change-grouped, plain prose.
|
||
Default to `[]` when the diff is trivial. Backward compatible: parsers
|
||
default to `[]` if absent.
|
||
- `risk_verdict`: exactly one line. Lead with "Low|Medium|High|Critical risk:"
|
||
followed by a concrete reason. Default to `""` when not applicable.
|
||
Backward compatible: parsers default to `""` if absent.
|
||
- `test_coverage`: short string. One of "Tests added" / "Tests changed" /
|
||
"No tests for behavioral change" / "No test files in repo". Default to `""`
|
||
when not applicable. Backward compatible: parsers default to `""` if absent.
|
||
- 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 fmt_tokens(n) -> str:
|
||
"""1234567 -> '1,234,567 (1.2M)'; 0 -> '0'; <1000 -> comma-only; None/negative -> '?'.
|
||
|
||
Always returns the full comma-separated number; the short suffix is a
|
||
parenthetical for fast scanning. Caps at B; the cost model never exceeds M.
|
||
"""
|
||
if n is None:
|
||
return "?"
|
||
if not isinstance(n, (int, float)) or n < 0:
|
||
return "?"
|
||
n = int(n)
|
||
if n < 1000:
|
||
return f"{n:,}"
|
||
if n < 1_000_000:
|
||
return f"{n:,} ({n / 1000:.1f}K)"
|
||
if n < 1_000_000_000:
|
||
return f"{n:,} ({n / 1_000_000:.1f}M)"
|
||
return f"{n:,} ({n / 1_000_000_000:.1f}B)"
|
||
|
||
|
||
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 _int_env(name: str, default: int) -> int:
|
||
"""Read an int from the environment, falling back on anything unparseable.
|
||
|
||
A typo in a tuning knob must not take down a review that is already
|
||
mid-flight — the operator gets a stderr line and the default instead.
|
||
"""
|
||
raw = os.environ.get(name, "")
|
||
if not str(raw).strip():
|
||
return default
|
||
try:
|
||
return int(str(raw).strip())
|
||
except (TypeError, ValueError):
|
||
print(
|
||
f"pragent: ignoring {name}={raw!r} (not an integer); using {default}",
|
||
file=sys.stderr, flush=True,
|
||
)
|
||
return default
|
||
|
||
|
||
# 1-5 merge-verdict score (higher = safer). Buckets:
|
||
# 5 = clean (or low/info/trivial only — nothing worth blocking on)
|
||
# 4 = medium present
|
||
# 3 = high present (operator should at least look)
|
||
# 1 = critical present (block the merge by default)
|
||
# Cross-lens agreement on any finding takes one more off, floored at 1.
|
||
_CONFIDENCE_BADGE = {5: "🟢", 4: "🟢", 3: "🟡", 2: "🟠", 1: "🔴"}
|
||
|
||
|
||
def merge_confidence(findings: list[dict], *, multi_lens_observed: bool = False) -> int:
|
||
"""1-5 merge verdict: higher = safer.
|
||
|
||
Tier drops driven by the most severe finding present:
|
||
- critical → 1
|
||
- high → 3
|
||
- medium → 4
|
||
- else → 5 (low / trivial / info / unknown → no drop)
|
||
|
||
An extra -1 when cross-lens agreement was observed on any finding
|
||
(``multi_lens_observed``). The flag is passed in explicitly because the
|
||
raw ``_multi_lens`` marker is stripped from findings by the time they
|
||
reach this function — first by ``opencode_review.run_lenses_review``
|
||
(the ``_``-prefix scrub) and again by ``_normalize_finding`` (the
|
||
7-key schema rebuild). The caller (``review_pr``) must capture the
|
||
signal before those strips fire. Final score is clamped to [1, 5] so
|
||
a critical + multi_lens combo doesn't go negative.
|
||
"""
|
||
if not findings:
|
||
return 5
|
||
max_rank = max(SEVERITY_RANK.get(f.get("severity", "low"), 0) for f in findings)
|
||
if max_rank >= SEVERITY_RANK["critical"]:
|
||
score = 1
|
||
elif max_rank >= SEVERITY_RANK["high"]:
|
||
score = 3
|
||
elif max_rank >= SEVERITY_RANK["medium"]:
|
||
score = 4
|
||
else:
|
||
score = 5
|
||
if multi_lens_observed:
|
||
score -= 1
|
||
return max(1, min(5, score))
|
||
|
||
|
||
def format_review_body(
|
||
findings: str,
|
||
model: str,
|
||
sha: str,
|
||
summary: str = "",
|
||
usage_section: str = "",
|
||
*,
|
||
summary_changes: list[str] | None = None,
|
||
risks: list[str] | None = None,
|
||
findings_for_table: list[dict] | None = None,
|
||
inline_count: int = 0,
|
||
confidence: int = 5,
|
||
walkthrough: list[str] | None = None,
|
||
risk_verdict: str = "",
|
||
test_coverage: str = "",
|
||
) -> str:
|
||
"""Format the posted review summary body.
|
||
|
||
Layout (per the operator's format guide):
|
||
|
||
* Header line (``🤖 AI Review …``) including the merge-confidence badge.
|
||
* **Summary of Changes** — 2–4 bullets of what the PR introduces
|
||
(`summary_changes`); falls back to the opencode prose `summary` if
|
||
the agent didn't emit the list.
|
||
* **Risk Verdict** — one-line "<level> risk: <reason>" verdict
|
||
(`risk_verdict`); omitted when empty.
|
||
* **Walkthrough** — up to 6 file- or change-grouped bullets
|
||
(`walkthrough`); the file part is wrapped in backticks so paths
|
||
render as code in Gitea. Omitted when empty.
|
||
* **Test Coverage** — short `test_coverage` string ("Tests added" /
|
||
etc.); omitted when empty.
|
||
* **Key Risks & Concerns** — bullets of potential bugs/edge cases
|
||
found across the diff (`risks`).
|
||
* **Findings Overview** — a Markdown table (severity / location /
|
||
one-line problem) covering ALL findings, anchored or not.
|
||
* Unanchored bullets — findings with no post-change line to anchor
|
||
(the inline ones are posted separately as Gitea review comments).
|
||
* AI Usage & Run Details — wrapped in a ``<details>`` collapsible so
|
||
the body stays scannable; cost lines stay inside it.
|
||
* Hidden SHA marker — for the dedupe pass.
|
||
|
||
`confidence` is a 1-5 merge verdict rendered as `<N>/5 <badge>` in the
|
||
header. Clamped to [1, 5] so a stray value (e.g. 0 from a missing
|
||
finding list) doesn't print a broken badge.
|
||
|
||
Empty `summary_changes` + empty `risks` + empty `summary` collapse into
|
||
a single "Summary of Changes: _no summary provided._" line so the body
|
||
never looks half-rendered.
|
||
"""
|
||
score = max(1, min(5, confidence))
|
||
badge = _CONFIDENCE_BADGE.get(score, "🟢")
|
||
confidence_str = f"{score}/5 {badge}"
|
||
header = REVIEW_HEADER.format(
|
||
model=model,
|
||
sha=sha[:8] if sha else "unknown",
|
||
confidence=confidence_str,
|
||
)
|
||
parts: list[str] = [header]
|
||
|
||
# --- Summary of Changes ---
|
||
sc = list(summary_changes or [])
|
||
if not sc and summary:
|
||
sc = _string_list(summary)
|
||
if sc:
|
||
sc = sc[:4]
|
||
items = "\n".join(f"- {item}" for item in sc)
|
||
parts.append(f"### Summary of Changes\n\n{items}")
|
||
else:
|
||
parts.append("### Summary of Changes\n\n_No summary provided._")
|
||
|
||
# --- Risk Verdict ---
|
||
if risk_verdict:
|
||
parts.append(f"### Risk Verdict\n\n{risk_verdict}")
|
||
|
||
# --- Walkthrough ---
|
||
wt = list(walkthrough or [])
|
||
if wt:
|
||
wt = wt[:6]
|
||
rendered = []
|
||
for item in wt:
|
||
# Items typically look like "a.py — adds X" (em-dash separator).
|
||
# Wrap the file path in backticks so it renders as code in the
|
||
# Gitea markdown body; leave the description as plain prose. When
|
||
# no separator is present, render the whole line as plain prose
|
||
# (the agent's "plain prose" fallback for change-grouped bullets).
|
||
if " — " in item:
|
||
path, _, rest = item.partition(" — ")
|
||
rendered.append(f"- `{path}` — {rest}")
|
||
else:
|
||
rendered.append(f"- {item}")
|
||
parts.append(f"### Walkthrough\n\n" + "\n".join(rendered))
|
||
|
||
# --- Test Coverage ---
|
||
if test_coverage:
|
||
parts.append(f"### Test Coverage\n\n{test_coverage}")
|
||
|
||
# --- Key Risks & Concerns ---
|
||
rs = list(risks or [])
|
||
if rs:
|
||
items = "\n".join(f"- {item}" for item in rs)
|
||
parts.append(f"### Key Risks & Concerns\n\n{items}")
|
||
else:
|
||
parts.append("### Key Risks & Concerns\n\n_None identified._")
|
||
|
||
# --- Findings Overview (table) ---
|
||
table = findings_table(findings_for_table or [])
|
||
if table:
|
||
n_inline = inline_count
|
||
n_total = len(findings_for_table or [])
|
||
if n_inline:
|
||
heading = f"### Findings Overview\n\n_{n_inline} inline comment(s); {n_total} total._"
|
||
else:
|
||
heading = f"### Findings Overview\n\n_{n_total} finding(s)._"
|
||
parts.append(f"{heading}\n\n{table}")
|
||
|
||
# --- Unanchored bullets ---
|
||
fb = (findings or "").strip()
|
||
if fb:
|
||
parts.append(fb)
|
||
|
||
# --- Collapsible usage ---
|
||
if usage_section:
|
||
parts.append(usage_section.strip())
|
||
|
||
# --- Hidden marker ---
|
||
marker = SHA_MARKER.format(sha=sha) if sha else ""
|
||
|
||
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 build_user_prompt(
|
||
title: str,
|
||
body: str,
|
||
diff: str,
|
||
config: dict | None = None,
|
||
prior_reviews: list[str] | None = None,
|
||
additional_context: str = "",
|
||
) -> str:
|
||
"""Assemble the user prompt: repo config + additional context + 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 additional_context:
|
||
# Repo-provided static background (architecture summary, module map,
|
||
# conventions, glossary, …). Cached for the review; the agent reads
|
||
# this ONCE per review and the prompt-cached prefix absorbs it on
|
||
# later steps — much cheaper than re-discovering the same facts from
|
||
# the source tree on every PR.
|
||
parts.append(
|
||
"## Repo-provided context (.pr-review.json:additional_context_urls "
|
||
"+ PRAGENT_ADDITIONAL_CONTEXT_URL — cached per review)\n" + additional_context
|
||
)
|
||
|
||
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], list[str], list[str], list[str], str, str]:
|
||
"""Parse the opengine's stdout into a 7-tuple:
|
||
(summary, findings, summary_changes, risks,
|
||
walkthrough, risk_verdict, test_coverage)
|
||
|
||
Accepts `{"summary": "...", "summary_changes": [...], "risks": [...],
|
||
"walkthrough": [...], "risk_verdict": "...", "test_coverage": "...",
|
||
"findings": [...]}` (the opencode pragent agent), the legacy 4-field
|
||
shape, or a bare `[...]` of finding dicts. The three new fields
|
||
(`walkthrough`, `risk_verdict`, `test_coverage`) default to empty
|
||
list / empty strings when absent — older outputs and the bare-array
|
||
shape stay backward compatible.
|
||
|
||
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 "", [], [], [], [], "", ""
|
||
summary = ""
|
||
summary_changes: list[str] = []
|
||
risks: list[str] = []
|
||
walkthrough: list[str] = []
|
||
risk_verdict = ""
|
||
test_coverage = ""
|
||
findings_raw = None
|
||
if isinstance(data, dict):
|
||
summary = str(data.get("summary", "") or "").strip()
|
||
summary_changes = _string_list(data.get("summary_changes"))
|
||
risks = _string_list(data.get("risks"))
|
||
walkthrough = _string_list(data.get("walkthrough"))
|
||
risk_verdict = str(data.get("risk_verdict", "") or "").strip()
|
||
test_coverage = str(data.get("test_coverage", "") or "").strip()
|
||
findings_raw = data.get("findings")
|
||
elif isinstance(data, list):
|
||
# Bare array: each item is a finding; no summary/sections.
|
||
findings_raw = data
|
||
else:
|
||
return "", [], [], [], [], "", ""
|
||
out = []
|
||
if isinstance(findings_raw, list):
|
||
for f in findings_raw:
|
||
n = _normalize_finding(f)
|
||
if n is not None:
|
||
out.append(n)
|
||
return summary, out, summary_changes, risks, walkthrough, risk_verdict, test_coverage
|
||
|
||
|
||
def _string_list(value) -> list[str]:
|
||
"""Coerce a JSON value into a list of non-empty strings.
|
||
|
||
Accepts a list of strings, a single string (split on lines/bullets), or
|
||
anything else (returns []). Used for `summary_changes` and `risks`,
|
||
which some agents emit as one big string instead of a list.
|
||
"""
|
||
if isinstance(value, list):
|
||
return [str(v).strip() for v in value if str(v).strip()]
|
||
if isinstance(value, str):
|
||
s = value.strip()
|
||
if not s:
|
||
return []
|
||
# Split on newlines OR on lines that start with "- " / "* " (markdown
|
||
# bullets). Strip the bullet markers.
|
||
out: list[str] = []
|
||
for line in s.splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
if line[:2] in ("- ", "* "):
|
||
line = line[2:].strip()
|
||
if line:
|
||
out.append(line)
|
||
return out
|
||
return []
|
||
|
||
|
||
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, "")
|
||
|
||
|
||
_SEVERITY_EMOJI = {
|
||
"critical": "🔴",
|
||
"high": "🔴",
|
||
"medium": "🟡",
|
||
"low": "🔵",
|
||
"trivial": "⚪",
|
||
"info": "⚪",
|
||
"nit": "⚪",
|
||
}
|
||
|
||
# Severities whose own name is rendered verbatim (uppercased) in the badge.
|
||
# Anything outside this set falls back to "INFO" so the badge label stays
|
||
# a clean short token regardless of what the model emits.
|
||
_BADGED_SEVERITY_LABELS = frozenset({
|
||
"critical", "high", "medium", "low", "trivial", "info", "nit",
|
||
})
|
||
|
||
|
||
def _severity_badge(severity: str) -> str:
|
||
"""Render the severity as emoji + uppercase label (e.g. ``🔴 [HIGH]``)."""
|
||
sev = (severity or "").lower()
|
||
emoji = _SEVERITY_EMOJI.get(sev, "⚪")
|
||
label = sev.upper() if sev in _BADGED_SEVERITY_LABELS else "INFO"
|
||
return f"{emoji} [{label}]"
|
||
|
||
|
||
def _format_reference(ref: str) -> str:
|
||
"""Render a reference URL as a clean Markdown hyperlink.
|
||
|
||
``"https://example.com/x"`` → ``"[example.com/x](https://example.com/x)"``.
|
||
Accepts the bare URL form so older findings still render readably; drops
|
||
anything that doesn't look like a URL rather than embedding raw text in
|
||
parens (the spec says: never print raw URLs).
|
||
"""
|
||
ref = (ref or "").strip()
|
||
if not ref:
|
||
return ""
|
||
if not (ref.startswith("http://") or ref.startswith("https://")):
|
||
# Non-URL text (e.g. a CVE id, a doc title). Render as plain text —
|
||
# `[CVE-2024-1](CVE-2024-1)` would render as a broken *relative* link
|
||
# in Gitea, which is worse than no link at all.
|
||
return ref
|
||
# Strip the scheme + www. for the visible label so the link text is short.
|
||
visible = ref
|
||
for prefix in ("https://", "http://"):
|
||
if visible.startswith(prefix):
|
||
visible = visible[len(prefix):]
|
||
break
|
||
if visible.startswith("www."):
|
||
visible = visible[4:]
|
||
# Drop trailing slash + truncate any path noise past 60 chars.
|
||
visible = visible.rstrip("/")
|
||
if len(visible) > 60:
|
||
visible = visible[:57] + "…"
|
||
return f"[{visible}]({ref})"
|
||
|
||
|
||
def inline_comment_body(f: dict) -> str:
|
||
"""Render one finding as a positional review-comment body.
|
||
|
||
Shape:
|
||
* Severity badge with emoji (🔴 HIGH / 🟡 MEDIUM / 🔵 LOW / ⚪ INFO).
|
||
* 1–2 short paragraphs: ``problem`` + optional ``fix``.
|
||
* ``suggestion`` block (Gitea/Forgejo apply-on-click) when the model
|
||
produced replacement code. Language-tagged fences are reserved for
|
||
cross-file patterns the suggestion block can't carry.
|
||
* Reference as a Markdown hyperlink (``[label](url)``) — never a raw URL.
|
||
* Per-comment attributed output tokens (`🪙 ~N tok (P% · attributed)`)
|
||
when the caller passed `compute_attribution` data. Hidden when the
|
||
finding has no attributed tokens (e.g. legacy callers / ollama path
|
||
without usage metering).
|
||
"""
|
||
badge = _severity_badge(f.get("severity", "medium"))
|
||
body = f"{badge} {f.get('problem', '').strip()}"
|
||
fix = (f.get("fix") or "").strip()
|
||
if fix:
|
||
body += f"\n\n**Fix:** {fix}"
|
||
suggestion = (f.get("suggestion") or "").strip()
|
||
if suggestion:
|
||
# `suggestion` fence is the standard one-click-apply block in
|
||
# Gitea/Forgejo/GitHub. The agent's replacement lines must already be
|
||
# indented as in the target file.
|
||
body += f"\n\n```suggestion\n{suggestion}\n```"
|
||
ref_md = _format_reference(f.get("reference", ""))
|
||
if ref_md:
|
||
body += f"\n\n🔗 **Reference:** {ref_md}"
|
||
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🪙 ~{fmt_tokens(tok)} tok ({pct:.0f}% · attributed output)"
|
||
return body
|
||
|
||
|
||
def summary_bullets(findings: list[dict]) -> str:
|
||
"""Render unanchored findings as PR-level bullets.
|
||
|
||
Used for findings that couldn't be anchored to a post-change line (no
|
||
inline comment posted). Each bullet carries severity, location, problem,
|
||
fix, and a Markdown-linked reference.
|
||
"""
|
||
lines = []
|
||
for f in findings:
|
||
loc = f"{f['path']}:{f['line']}" if f["line"] else f["path"]
|
||
badge = _severity_badge(f.get("severity", "medium"))
|
||
problem = f.get("problem", "").strip()
|
||
body = f"- {badge} `{loc}` — {problem}"
|
||
fix = (f.get("fix") or "").strip()
|
||
if fix:
|
||
body += f"\n - **Fix:** {fix}"
|
||
ref_md = _format_reference(f.get("reference", ""))
|
||
if ref_md:
|
||
body += f"\n - 🔗 **Reference:** {ref_md}"
|
||
lines.append(body)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def findings_table(findings: list[dict]) -> str:
|
||
"""Render ALL findings as a Markdown table for the PR-level comment.
|
||
|
||
Columns: severity emoji, location (path:line), and a one-line summary.
|
||
Findings with empty location collapse to just the severity + summary.
|
||
"""
|
||
if not findings:
|
||
return ""
|
||
header = "| Severity | Location | Finding |\n|---|---|---|"
|
||
rows = []
|
||
for f in findings:
|
||
badge = _severity_badge(f.get("severity", "medium"))
|
||
path = (f.get("path") or "").strip()
|
||
line = f.get("line")
|
||
loc = f"`{path}:{line}`" if line else (f"`{path}`" if path else "_(no location)_")
|
||
problem = (f.get("problem") or "").strip()
|
||
# Escape pipes inside the finding text so the table stays valid.
|
||
problem_esc = problem.replace("|", "\\|").replace("\n", " ")
|
||
rows.append(f"| {badge} | {loc} | {problem_esc} |")
|
||
return "\n".join([header, *rows])
|
||
|
||
|
||
def _render_collapsible_usage(usage: dict | None, model: str, config: dict | None) -> str:
|
||
"""Render the telemetry as a collapsible ``<details>`` block.
|
||
|
||
Empty string when `usage` is None. The equivalent-cost table is the
|
||
operator's budgeting signal — the pilot runs on a free tier, so the
|
||
`actual` line is $0.00; the table shows what the same measured tokens
|
||
would bill on mainstream paid APIs (configurable via `compare_against`,
|
||
defaulting to ``DEFAULT_COMPARE_AGAINST``). The row matching `cost_target`
|
||
is bolded so the price target stands out. The whole table is omitted when
|
||
every row would be $0 (no work done).
|
||
"""
|
||
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 ""
|
||
cost_target, price_err = _resolve_price_target(config)
|
||
if price_err:
|
||
# Surface config typos loudly but do not pollute the posted summary
|
||
# body — typos at the table-row level would render as English
|
||
# mid-table and look like a model error.
|
||
print(f"pragent: {price_err}", file=sys.stderr, flush=True)
|
||
# Lazy: cost_model has no dep on ai_review, and the ollama path
|
||
# never reaches this branch.
|
||
from cost_model import PRICES as _PRICES
|
||
cfg = config or {}
|
||
compare: list[str] = list(cfg.get("compare_against") or DEFAULT_COMPARE_AGAINST)
|
||
# Always include the resolved cost_target (env + config), even when the
|
||
# operator pinned a different `compare_against` roster — the price target
|
||
# row is the one maintainers eyeball against. Skip silently if the key
|
||
# isn't a known Price (e.g. a typo that slipped past stderr earlier).
|
||
if cost_target in _PRICES and cost_target not in compare:
|
||
compare.append(cost_target)
|
||
eq_rows: list[str] = []
|
||
for key in compare:
|
||
if key not in _PRICES:
|
||
continue
|
||
c = equivalent_cost(usage, key)
|
||
if c <= 0:
|
||
continue
|
||
label = _PRICES[key].name
|
||
cost_str = f"${c:.4f}" if c < 0.01 else f"${c:.2f}"
|
||
bold = "**" if key == cost_target else ""
|
||
eq_rows.append(f"| {bold}{label}{bold} | {cost_str} |")
|
||
in_tok = usage.get("input", 0)
|
||
out_tok = usage.get("output", 0)
|
||
reason_tok = usage.get("reasoning", 0)
|
||
cache_r = usage.get("cache_read", 0)
|
||
cache_w = usage.get("cache_write", 0)
|
||
total = usage.get("total", 0)
|
||
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 output is *attributed* (one model pass "
|
||
"produces all findings; output split by each finding's body weight)."
|
||
)
|
||
lines = [
|
||
"<details>",
|
||
"<summary>🔋 AI Usage & Run Details</summary>",
|
||
"",
|
||
f"- **Model / Engine**: `{model}` · opencode · {usage.get('steps', 0)} steps · {dur_s}",
|
||
f"- **Total Tokens**: {fmt_tokens(in_tok)} in / {fmt_tokens(out_tok)} out "
|
||
f"({fmt_tokens(reason_tok)} reasoning, cache {fmt_tokens(cache_r)} read / "
|
||
f"{fmt_tokens(cache_w)} write, {fmt_tokens(total)} total)",
|
||
f"- **Actual**: {actual_s}{actual_note}",
|
||
f"- **Scope**: {scope}",
|
||
]
|
||
if eq_rows:
|
||
lines.append("")
|
||
lines.append("- **Equivalent cost on paid providers** (this run's tokens):")
|
||
lines.append("")
|
||
lines.append("| Provider | Cost |")
|
||
lines.append("|---|---:|")
|
||
lines.extend(eq_rows)
|
||
# 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)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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
|
||
additional_context_urls list[str] (≤ 8) — see fetch_additional_context
|
||
"""
|
||
if not raw:
|
||
return {}
|
||
try:
|
||
data = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return {}
|
||
if not isinstance(data, dict):
|
||
return {}
|
||
|
||
def _str_list(v):
|
||
if isinstance(v, list) and all(isinstance(x, str) for x in v):
|
||
return [x[:CONFIG_MAX_ITEM_CHARS] for x in v[:CONFIG_MAX_LIST_ITEMS]]
|
||
return None
|
||
|
||
out: dict = {}
|
||
for k in ("focus", "exclude_paths", "languages"):
|
||
s = _str_list(data.get(k))
|
||
if s is not None:
|
||
out[k] = s
|
||
|
||
instr = data.get("instructions")
|
||
if isinstance(instr, str) and instr.strip():
|
||
out["instructions"] = instr.strip()[:CONFIG_MAX_INSTRUCTIONS_CHARS]
|
||
|
||
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()
|
||
|
||
acu = data.get("additional_context_urls")
|
||
if isinstance(acu, list):
|
||
urls: list[str] = []
|
||
for x in acu:
|
||
if isinstance(x, str):
|
||
u = x.strip()
|
||
if u:
|
||
urls.append(u)
|
||
if urls:
|
||
# Cap is also enforced later by _resolve_additional_context_urls;
|
||
# this just stops a 10k-entry file from making the config huge.
|
||
out["additional_context_urls"] = urls[:8]
|
||
|
||
# Multi-lens reviewers roster. Absent / empty list = the 5-lens default
|
||
# in pilot/opencode_review.py (security, docs, code-quality, tests, perf).
|
||
# This is the cheap trigger: once the config declares `reviewers[]`, the
|
||
# orchestrator spawns one opencode subprocess per lens in parallel. Set
|
||
# to `[]` to opt out (single-primary fallback). Capped at 8.
|
||
rev = _parse_reviewers_array(data.get("reviewers"))
|
||
if rev is not None:
|
||
out["reviewers"] = rev
|
||
|
||
# Triage (cheap pre-filter that picks a subset of lenses). Off by default
|
||
# to keep the parse deterministic; the orchestrator's own default is
|
||
# to enable it when `reviewers[]` is present.
|
||
tr = _parse_triage_object(data.get("triage"))
|
||
if tr is not None:
|
||
out["triage"] = tr
|
||
|
||
# Repo-level kill-switch: `enabled: false` lets a maintainer pause the bot
|
||
# for this repo without removing the file (handy during a flaky provider
|
||
# outage). Always written so callers can do `cfg.get("enabled") is False`
|
||
# without a separate default — the file itself is committed, so we treat
|
||
# absent / wrong-type as an explicit off rather than as "config missing".
|
||
en = data.get("enabled")
|
||
out["enabled"] = en if isinstance(en, bool) else False
|
||
|
||
# Compare-against roster: list of `cost_model.PRICES` keys the render layer
|
||
# uses to print equivalent-cost lines (one per key) for maintainer
|
||
# budgeting. Unknown keys are dropped with a stderr line so a typo is loud.
|
||
# Lazy import: `cost_model` has no dep on `ai_review`, and the ollama
|
||
# fallback path never hits this branch — keep import-time cost low there.
|
||
from cost_model import PRICES as _PRICES
|
||
ca = data.get("compare_against")
|
||
if isinstance(ca, list):
|
||
cleaned: list[str] = []
|
||
for x in ca:
|
||
if isinstance(x, str) and x.strip() in _PRICES:
|
||
cleaned.append(x.strip())
|
||
elif isinstance(x, str):
|
||
print(
|
||
f"pragent: ignoring compare_against entry {x!r} "
|
||
f"(not in cost_model.PRICES); valid: {', '.join(sorted(_PRICES))}",
|
||
file=sys.stderr, flush=True,
|
||
)
|
||
if cleaned:
|
||
out["compare_against"] = cleaned[:12]
|
||
|
||
return out
|
||
|
||
|
||
def _parse_reviewers_array(raw) -> list[dict] | None:
|
||
"""Sanitize `.pr-review.json:reviewers[]` to a list of dicts.
|
||
|
||
Hard caps: 8 entries (default-reviewers.xml-bound), 200 chars per string
|
||
field. Untyped / non-list → None (caller keeps the default). Fields we
|
||
don't know about are dropped (no schema drift allowed).
|
||
"""
|
||
if not isinstance(raw, list):
|
||
return None
|
||
cap = 8
|
||
out: list[dict] = []
|
||
for entry in raw[:cap]:
|
||
if not isinstance(entry, dict):
|
||
continue
|
||
spec: dict = {}
|
||
rid = entry.get("id")
|
||
if isinstance(rid, str) and rid.strip():
|
||
cand = rid.strip()[:CONFIG_MAX_ITEM_CHARS]
|
||
# Same id shape required by opencode_review.parse_reviewers_config:
|
||
# kebab-case so it maps 1:1 to .opencode/agents/<id>.md
|
||
import re as _re
|
||
if _re.match(r"^[a-z0-9][a-z0-9-]{0,31}$", cand):
|
||
spec["id"] = cand
|
||
if not spec.get("id"):
|
||
continue
|
||
for sk in ("agent_file", "model"):
|
||
sv = entry.get(sk)
|
||
if isinstance(sv, str) and sv.strip():
|
||
spec[sk] = sv.strip()[:CONFIG_MAX_ITEM_CHARS]
|
||
sf = entry.get("severity_floor")
|
||
if isinstance(sf, str) and sf.strip().lower() in SEVERITY_VALUES:
|
||
spec["severity_floor"] = sf.strip().lower()
|
||
mf = entry.get("max_findings")
|
||
if isinstance(mf, int) and not isinstance(mf, bool) and 1 <= mf <= CONFIG_MAX_FINDINGS:
|
||
spec["max_findings"] = mf
|
||
act = entry.get("activation")
|
||
if isinstance(act, str) and act.strip().lower() in ("auto", "always", "off"):
|
||
spec["activation"] = act.strip().lower()
|
||
skip = entry.get("skip_if_all_changed_paths")
|
||
if isinstance(skip, str) and skip.strip():
|
||
spec["skip_if_all_changed_paths"] = skip.strip()[:CONFIG_MAX_ITEM_CHARS]
|
||
globs = entry.get("hotpath_globs")
|
||
if isinstance(globs, list):
|
||
cleaned = [g for g in globs if isinstance(g, str) and g.strip()]
|
||
if cleaned:
|
||
spec["hotpath_globs"] = [
|
||
g.strip()[:CONFIG_MAX_ITEM_CHARS]
|
||
for g in cleaned[:CONFIG_MAX_LIST_ITEMS]
|
||
]
|
||
out.append(spec)
|
||
return out
|
||
|
||
|
||
def _parse_triage_object(raw) -> dict | None:
|
||
"""Sanitize `.pr-review.json:triage` to a dict.
|
||
|
||
Returns `None` when absent. When the value is malformed (not an object),
|
||
returns `{"enabled": False}` so a typo disables triage rather than
|
||
silently making the orchestrator error.
|
||
"""
|
||
if raw is None:
|
||
return None
|
||
if not isinstance(raw, dict):
|
||
return {"enabled": False}
|
||
out: dict = {}
|
||
if isinstance(raw.get("enabled"), bool):
|
||
out["enabled"] = raw["enabled"]
|
||
if isinstance(raw.get("model"), str) and raw["model"].strip():
|
||
out["model"] = raw["model"].strip()[:CONFIG_MAX_ITEM_CHARS]
|
||
ml = raw.get("max_lenses")
|
||
if isinstance(ml, int) and not isinstance(ml, bool) and 1 <= ml <= 8:
|
||
out["max_lenses"] = ml
|
||
return out
|
||
|
||
|
||
def effective_config(config: dict | None) -> dict:
|
||
"""Apply STYLE_DEFAULTS for any field the config didn't pin.
|
||
|
||
Returns a NEW dict combining the user's `.pr-review.json` (if any) with the
|
||
derived `max_findings` / `severity_threshold`. Style itself is preserved
|
||
so downstream code can branch on it.
|
||
"""
|
||
style = (config or {}).get("style", "balanced")
|
||
max_findings, severity_threshold = STYLE_DEFAULTS.get(style, STYLE_DEFAULTS["balanced"])
|
||
out = dict(config or {})
|
||
out.setdefault("style", style)
|
||
out.setdefault("max_findings", max_findings)
|
||
out.setdefault("severity_threshold", severity_threshold)
|
||
return out
|
||
|
||
|
||
_TEST_PATH_RE = re.compile(
|
||
r"(?:^|/)("
|
||
r"[^/]*[Tt]est\.[A-Za-z]+" # FooTest.java / foo_test.py
|
||
r"|[^/]*\.[Tt]est\.[A-Za-z]+" # foo.Test.java
|
||
r"|[^/]*_test\.py" # foo_test.py
|
||
r"|test_[^/]*\.py" # test_foo.py
|
||
r"|__tests__/[^/]+" # __tests__/foo.js
|
||
r"|[^/]*\.spec\.[A-Za-z]+" # foo.spec.ts
|
||
r")$"
|
||
)
|
||
|
||
|
||
def is_test_path(path: str) -> bool:
|
||
"""Heuristic: is `path` a test file by name/path convention?
|
||
|
||
Conservative — false positives cost real findings; false negatives just
|
||
produce one extra line in the summary. Patterns: `FooTest.java`,
|
||
`foo_test.py`, `test_foo.py`, `__tests__/foo.js`, `foo.spec.ts`, anything
|
||
ending in `.Test.java`.
|
||
"""
|
||
if not path:
|
||
return False
|
||
return bool(_TEST_PATH_RE.search(path))
|
||
|
||
|
||
def _glob_to_regex(glob: str) -> re.Pattern:
|
||
"""Translate a shell-style glob to a compiled regex.
|
||
|
||
Supports `*` (any chars except `/`), `**` (any chars including `/`),
|
||
`?` (single non-`/` char). Other characters are escaped. Used by
|
||
`apply_repo_config` to test `patterns.allow` / `patterns.deny` globs.
|
||
"""
|
||
out = []
|
||
i = 0
|
||
while i < len(glob):
|
||
c = glob[i]
|
||
if c == "*":
|
||
if i + 1 < len(glob) and glob[i + 1] == "*":
|
||
out.append(".*")
|
||
i += 2
|
||
# swallow a following `/` so `**/x` and `x/**/y` behave
|
||
if i < len(glob) and glob[i] == "/":
|
||
i += 1
|
||
continue
|
||
out.append("[^/]*")
|
||
elif c == "?":
|
||
out.append("[^/]")
|
||
else:
|
||
out.append(re.escape(c))
|
||
i += 1
|
||
return re.compile("^" + "".join(out) + "$")
|
||
|
||
|
||
def apply_repo_config(
|
||
findings: list[dict],
|
||
config: dict | None,
|
||
changed_paths: list[str] | None = None,
|
||
) -> tuple[list[dict], list[dict]]:
|
||
"""Filter + cap findings per `.pr-review.json` rules. Returns (kept, dropped).
|
||
|
||
Filters applied (in order):
|
||
1. `exclude_tests` + test-path heuristic → drop test files
|
||
2. `exclude_paths` glob match → drop matched paths
|
||
3. `patterns.deny` glob match → drop matched paths
|
||
4. `patterns.allow` (if non-empty) → keep ONLY matched paths
|
||
5. `severity_threshold` → drop below threshold
|
||
6. `max_findings` → keep first N (highest-severity-first)
|
||
7. `require_tests` → append a low-severity finding
|
||
if changed paths include non-test files but no test files changed
|
||
alongside them (caller passes `changed_paths` from the brief).
|
||
"""
|
||
eff = effective_config(config)
|
||
keep: list[dict] = []
|
||
drop: list[dict] = []
|
||
deny_globs = [_glob_to_regex(g) for g in (eff.get("patterns", {}) or {}).get("deny", [])]
|
||
allow_globs = [_glob_to_regex(g) for g in (eff.get("patterns", {}) or {}).get("allow", [])]
|
||
deny_path_globs = [_glob_to_regex(g) for g in eff.get("exclude_paths", [])]
|
||
threshold_rank = SEVERITY_RANK[eff["severity_threshold"]]
|
||
|
||
for f in findings:
|
||
path = f.get("path", "")
|
||
if eff.get("exclude_tests") and is_test_path(path):
|
||
drop.append(f); continue
|
||
if any(rx.search(path) for rx in deny_path_globs):
|
||
drop.append(f); continue
|
||
if any(rx.search(path) for rx in deny_globs):
|
||
drop.append(f); continue
|
||
if allow_globs and not any(rx.search(path) for rx in allow_globs):
|
||
drop.append(f); continue
|
||
sev_rank = SEVERITY_RANK.get(f.get("severity", "low"), 0)
|
||
if sev_rank < threshold_rank:
|
||
drop.append(f); continue
|
||
keep.append(f)
|
||
|
||
cap = eff["max_findings"]
|
||
if len(keep) > cap:
|
||
dropped = keep[cap:]
|
||
keep = keep[:cap]
|
||
drop.extend(dropped)
|
||
|
||
if eff.get("require_tests") and changed_paths is not None:
|
||
non_test = [p for p in changed_paths if not is_test_path(p)]
|
||
any_test = any(is_test_path(p) for p in changed_paths)
|
||
if non_test and not any_test:
|
||
keep.append({
|
||
"severity": "low",
|
||
"path": non_test[0],
|
||
"line": 1,
|
||
"problem": "no test file changed alongside this behavioral change (require_tests=true)",
|
||
"fix": "add a unit test exercising the changed branch",
|
||
"suggestion": "",
|
||
"reference": "",
|
||
"_config_synthetic": True,
|
||
})
|
||
|
||
return keep, drop
|
||
|
||
|
||
def reviewed_shas(reviews: list[dict]) -> set[str]:
|
||
"""Pull every `<!-- pragent:sha=... -->` marker out of a PR's reviews."""
|
||
shas: set[str] = set()
|
||
for r in reviews or []:
|
||
body = r.get("body") or ""
|
||
for m in _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)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Additional context URLs — static repo-provided background fetched once
|
||
# per review and injected into the brief. The idea is the cheap reusable
|
||
# knowledge (architecture summary, module map, conventions, glossary, past
|
||
# incident write-ups, …) lives in a versioned file the maintainers control,
|
||
# so the agent doesn't have to re-read the source tree to rediscover it on
|
||
# every PR. Cached by URL for the lifetime of the process.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Hard caps — these guard against a single repo-config entry pulling down a
|
||
# 2 MB doc and blowing the brief budget. Per-URL truncation keeps the worst
|
||
# case bounded; total truncation caps the sum across URLs.
|
||
_ADDITIONAL_CONTEXT_MAX_URLS = 8
|
||
_ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS = 4000
|
||
_ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS = 16_000
|
||
_ADDITIONAL_CONTEXT_TIMEOUT_S = 5
|
||
# Module-level cache, keyed by URL. The webhook server is a single Python
|
||
# process per pod and reviews happen sequentially, so this stays bounded.
|
||
_ADDITIONAL_CONTEXT_CACHE: dict[str, str] = {}
|
||
|
||
|
||
def _parse_additional_context_env(value: str) -> list[str]:
|
||
"""Comma-split an env var into a deduped, ordered URL list."""
|
||
if not value:
|
||
return []
|
||
seen: set[str] = set()
|
||
out: list[str] = []
|
||
for piece in value.split(","):
|
||
u = piece.strip()
|
||
if u and u not in seen:
|
||
seen.add(u)
|
||
out.append(u)
|
||
return out
|
||
|
||
|
||
def _resolve_additional_context_urls(config: dict | None) -> list[str]:
|
||
"""Merge the env var `PRAGENT_ADDITIONAL_CONTEXT_URL` with the per-repo
|
||
config field `additional_context_urls`. Env var wins on ordering — it
|
||
appears first so a one-off override can shadow a stale config entry."""
|
||
env = _parse_additional_context_env(os.environ.get("PRAGENT_ADDITIONAL_CONTEXT_URL", ""))
|
||
cfg_raw = (config or {}).get("additional_context_urls") or []
|
||
cfg: list[str] = []
|
||
if isinstance(cfg_raw, list):
|
||
for x in cfg_raw:
|
||
if isinstance(x, str):
|
||
u = x.strip()
|
||
if u and u not in set(env):
|
||
cfg.append(u)
|
||
merged = env + cfg
|
||
return merged[:_ADDITIONAL_CONTEXT_MAX_URLS]
|
||
|
||
|
||
def _fetch_one_additional_context(url: str) -> str | None:
|
||
"""Fetch a single URL. Returns the body (UTF-8, truncated) or None on
|
||
any failure — never raises; additional-context is best-effort.
|
||
|
||
Reject non-http(s) schemes defensively so a misconfigured `file://` or
|
||
`javascript:` URL cannot escape the pod. Cap per-URL size before parsing
|
||
to avoid a 50 MB response landing in memory.
|
||
"""
|
||
try:
|
||
parsed = urllib.parse.urlparse(url)
|
||
except ValueError:
|
||
return None
|
||
if parsed.scheme not in ("http", "https"):
|
||
return None
|
||
try:
|
||
req = urllib.request.Request(url, headers={"User-Agent": "pragent/1.0 (+context)"})
|
||
with urllib.request.urlopen(req, timeout=_ADDITIONAL_CONTEXT_TIMEOUT_S) as r:
|
||
raw = r.read(_ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS + 1)
|
||
if len(raw) > _ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS:
|
||
raw = raw[:_ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS]
|
||
truncated = True
|
||
else:
|
||
truncated = False
|
||
body = raw.decode("utf-8", errors="replace")
|
||
except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError, ValueError):
|
||
return None
|
||
if truncated:
|
||
body += "\n…[truncated]"
|
||
return body
|
||
|
||
|
||
def fetch_additional_context(urls: list[str]) -> str:
|
||
"""Fetch a list of URLs, join into one string for the brief. Cached.
|
||
|
||
Empty when no URLs are given. Best-effort: a URL that errors is logged
|
||
to stderr and skipped — never aborts the review. Each fetched body is
|
||
truncated to `_ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS` and the joined
|
||
output to `_ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS`. Already-cached URLs
|
||
are not refetched.
|
||
"""
|
||
if not urls:
|
||
return ""
|
||
blocks: list[str] = []
|
||
total = 0
|
||
for url in urls:
|
||
if url in _ADDITIONAL_CONTEXT_CACHE:
|
||
body = _ADDITIONAL_CONTEXT_CACHE[url]
|
||
else:
|
||
body = _fetch_one_additional_context(url) or ""
|
||
_ADDITIONAL_CONTEXT_CACHE[url] = body
|
||
if not body:
|
||
continue
|
||
block = f"### {url}\n\n{body}"
|
||
if total + len(block) > _ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS:
|
||
remaining = _ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS - total
|
||
if remaining <= 80:
|
||
break
|
||
block = block[:remaining] + "\n…[truncated]"
|
||
blocks.append(block)
|
||
total = _ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS
|
||
break
|
||
blocks.append(block)
|
||
total += len(block)
|
||
return "\n\n".join(blocks)
|
||
|
||
|
||
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,
|
||
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".
|
||
|
||
The opencode engine's measured token/cost usage is always rendered as a
|
||
`## 🔋 AI usage` section on the review body and an attributed `🪙 ~N tok`
|
||
line on each inline comment when usage data is available (i.e. when the
|
||
opencode subprocess returned a `usage` dict). No-op on the ollama fallback
|
||
(no usage available — `usage` is None).
|
||
|
||
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_env("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 = ""
|
||
# Static repo-provided context (architecture summary, module map, …)
|
||
# fetched once from `additional_context_urls` (env + .pr-review.json).
|
||
# Cheap, cached, capped — see fetch_additional_context.
|
||
additional_context = fetch_additional_context(_resolve_additional_context_urls(config))
|
||
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}"
|
||
# 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, _walkthrough, _risk_verdict, _test_coverage = 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 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 = _render_collapsible_usage(usage, model, config=config) if usage else ""
|
||
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, additional_context)
|
||
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 = []
|
||
# Capture cross-lens agreement BEFORE apply_repo_config — by the time
|
||
# findings land in `review_pr` the `_multi_lens` marker has already
|
||
# been scrubbed (once by `opencode_review.run_lenses_review`'s
|
||
# `_`-prefix strip, again by `_normalize_finding`'s 7-key rebuild),
|
||
# so `merge_confidence` cannot read it off the dict. We scan here as
|
||
# the convergence point for both engine paths; in practice the kwarg
|
||
# currently always passes False, but the structural plumbing is
|
||
# correct for any future code path that preserves the flag.
|
||
multi_lens = any(f.get("_multi_lens") for f in findings)
|
||
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,
|
||
)
|
||
|
||
# Compute attribution so inline comments + the table can show per-comment
|
||
# estimates. Only meaningful when we have measured usage.
|
||
if usage and usage.get("output"):
|
||
compute_attribution(findings, usage["output"])
|
||
usage_section = _render_collapsible_usage(usage, model, config=config) if usage else ""
|
||
|
||
# Anchor against the RAW diff, never the compressed one. Compression
|
||
# drops context lines, so a finding on a line that survived in the file
|
||
# but not in the prompt would be demoted to a bullet for no reason.
|
||
# (compress_diff renumbers its hunks, so both are line-accurate; the
|
||
# raw diff is simply the complete set.)
|
||
anchors = parse_diff_anchors(raw_diff)
|
||
anchored, unanchored = split_findings(findings, anchors)
|
||
|
||
# Summary body: unanchored bullets fall through to a "Unanchored notes"
|
||
# section; the structured Findings Overview table covers both anchored
|
||
# + unanchored so reviewers see the full set even if inline comments
|
||
# are collapsed.
|
||
bullets = summary_bullets(unanchored)
|
||
summary_parts = []
|
||
if bullets:
|
||
summary_parts.append("### Unanchored Notes\n\n" + bullets)
|
||
# 1-5 merge verdict for the header badge. Computed AFTER filtering +
|
||
# anchoring so the verdict reflects what the operator sees (a critical
|
||
# finding that fails to anchor is still a critical finding). The
|
||
# default 5 keeps any failure path (e.g. empty findings) green.
|
||
# Cross-lens agreement is passed in via kwarg (see multi_lens scan
|
||
# above) because the `_multi_lens` flag is stripped before findings
|
||
# reach this call.
|
||
confidence = merge_confidence(findings, multi_lens_observed=multi_lens)
|
||
summary_body = format_review_body(
|
||
"\n\n".join(summary_parts), model, sha,
|
||
summary=review_summary,
|
||
usage_section=usage_section,
|
||
summary_changes=summary_changes,
|
||
risks=risks,
|
||
findings_for_table=findings,
|
||
inline_count=len(anchored),
|
||
confidence=confidence,
|
||
)
|
||
|
||
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_env("OLLAMA_MAX_TOKENS", 8000),
|
||
max_chars=_int_env("DIFF_MAX_CHARS", 150000),
|
||
base_ref=os.environ.get("PR_BASE_REF", ""),
|
||
)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(run()) |