Marcos
2b1cf750b7
Four defects, all found reviewing PR #9 (two of them by pragent-bot's own review of that PR, which the anchoring bug then misplaced): * compress_diff dropped context lines but copied the original `@@` hunk header verbatim, so the header no longer described the lines beneath it. parse_diff_anchors then walked stale headers and produced anchor sets shifted by the number of elided lines, misplacing inline comments or demoting them to bullets. Each surviving run of lines is now re-emitted as its own hunk with a recomputed `@@ -a,b +c,d @@`, so the output stays a valid unified diff whose numbers describe the real post-change file. The pseudo-marker `@@ … N context line(s) omitted … @@` is gone; it parsed as a hunk header and reset the anchor counter to 0. Anchoring additionally runs on the raw diff now, so the prompt window can never shrink the anchorable set. * compress_diff's `_FILE_HEADER` regex matched diff *body* lines: a removed YAML `---` separator or an added `++` line was read as a file header, truncating the hunk and dropping its `@@` header with it. Body detection is now prefix-based, with a full-shape hunk-header regex. * extract_finding_bullets could not match the bullets pragent itself posts: summary_bullets renders an emoji severity badge between the `-` and the `[SEV]` tag, which the regex rejected, so compact_prior_reviews always returned [] and every re-review repeated its previous findings. * triage returning `{"lenses":[]}` — documented in .opencode/agents/triage.md as "no lens has surface, skip the fan-out" — ran every lens instead, since _intersect_with_triage mapped an empty selection to "all" and the call site had a second `or reviewers` fallback. `[]` and None are now distinct outcomes: `[]` skips, None fails open. A roster naming only unknown lens ids now fails open rather than silencing the review. The skip path returns a well-formed empty-findings response instead of "", which had landed in ai_review's unparseable-output branch and posted "AI review produced no parseable output" — a malfunction message for a normal verdict. Also: non-URL references (a CVE id, a doc title) rendered as `[CVE-2024-1234](CVE-2024-1234)`, a broken relative link in Gitea — now plain text. PRAGENT_DIFF_CONTEXT and friends parse through _int_env, so a typo logs and falls back instead of killing a review mid-flight. Removed format_usage_section, dead since the collapsible usage block replaced it and carrying a duplicate copy of the price-target logic. Tests: 290 -> 301. New coverage for hunk-header fidelity before/after compression, header-shaped content lines, the bullet round-trip against the real renderer, and triage's three outcomes (previously untested). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B11e8TZZxJyzHW7jj7KWUN
2002 lines
81 KiB
Python
2002 lines
81 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 _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
|
||
|
||
|
||
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,
|
||
) -> str:
|
||
"""Format the posted review summary body.
|
||
|
||
Layout (per the operator's format guide):
|
||
|
||
* Header line (``🤖 AI Review …``).
|
||
* **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.
|
||
* **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.
|
||
|
||
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.
|
||
"""
|
||
header = REVIEW_HEADER.format(model=model, sha=sha[:8] if sha else "unknown")
|
||
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._")
|
||
|
||
# --- 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]]:
|
||
"""Parse the opengine's stdout into (summary, findings, summary_changes, risks).
|
||
|
||
Accepts `{"summary": "...", "summary_changes": [...], "risks": [...],
|
||
"findings": [...]}` (the opencode pragent agent), `{"findings": [...]}`,
|
||
or a bare `[...]` of finding dicts. `summary_changes` and `risks` default
|
||
to empty lists; older outputs without them still parse fine. 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] = []
|
||
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"))
|
||
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
|
||
|
||
|
||
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": "🔵",
|
||
"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 {"critical", "high", "medium", "low"} 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🪙 ~{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 cost-equivalent line is always
|
||
shown (it's the operator's budgeting signal). The `actual` line is shown
|
||
but the FREE-TIER note is collapsed into a single short clause.
|
||
"""
|
||
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 ""
|
||
price_key, price_err = _resolve_price_target(config)
|
||
from cost_model import PRICES
|
||
eq = equivalent_cost(usage, price_key)
|
||
eq_s = f"${eq:.4f}" if eq else "$0.00"
|
||
eq_label = PRICES[price_key].name
|
||
eq_note = (
|
||
f" _(price target: `{price_key}`; {price_err})_"
|
||
if price_err else ""
|
||
)
|
||
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**: {in_tok} in / {out_tok} out ({reason_tok} reasoning, cache {cache_r} read / {cache_w} write, {total} total)",
|
||
f"- **Est. cost on {eq_label}**: {eq_s}{eq_note}",
|
||
f"- **Actual**: {actual_s}{actual_note}",
|
||
f"- **Scope**: {scope}",
|
||
]
|
||
# 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
|
||
|
||
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,
|
||
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_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 = 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 = _render_collapsible_usage(usage, model, config=config) if report_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 = []
|
||
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 AND the PR asked
|
||
# for it.
|
||
if report_usage and usage and usage.get("output"):
|
||
compute_attribution(findings, usage["output"])
|
||
usage_section = _render_collapsible_usage(usage, model, config=config) if report_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)
|
||
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),
|
||
)
|
||
|
||
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()) |