Marcos
4129f217fa
Gitea 1.26.x's POST /pulls/{i}/reviews does NOT honor the line/side fields
used by newer Gitea — it silently drops them, leaving the comment unpositioned.
Gitea then renders a file-level review comment on EVERY diff line of the file,
so a 5-finding review on a 25-line diff showed ~125 comment blocks in the
Files Changed view (the flood reported on canalhandia PR #2).
The 1.26 schema anchors inline review comments with new_position (line in the
post-change file) + old_position: 0. f["line"] is already a validated
post-change (RIGHT-side) line from split_findings, so it maps directly to
new_position. Verified: new_position=98 -> position=98 + populated diff_hunk
(positioned, renders on line 98 only); the old line/side form -> position=0,
empty diff_hunk (unpositioned).
49 tests pass (no test asserted the POST payload shape).
Co-Authored-By: Claude <noreply@anthropic.com>
785 lines
30 KiB
Python
785 lines
30 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 apply-able ```suggestion blocks where the model could produce
|
|
them 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 ```suggestion fence so Gitea renders an
|
|
apply-button. 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)
|
|
PRAGENT_BOT_TOKEN bot access token (repo secret)
|
|
PRAGENT_SHA head SHA to tag the review
|
|
OLLAMA_URL headroom proxy URL, e.g. http://100.74.17.70: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.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")
|
|
REPO_CONFIG_FILE = ".pr-review.json"
|
|
|
|
SYSTEM_PROMPT = """You are a senior, pragmatic code reviewer. Review the pull request diff below.
|
|
|
|
Report ONLY real, actionable issues: correctness bugs, security problems, risky
|
|
changes, missing tests for changed behaviour, and breaking API/contract changes.
|
|
Honour any repo-specific focus / instructions given in the prompt; if focus is
|
|
given, weight those areas higher, but do not ignore a critical issue outside them.
|
|
|
|
Output STRICT JSON only — no prose, no markdown fences. Shape:
|
|
{
|
|
"findings": [
|
|
{
|
|
"severity": "critical|high|medium|low",
|
|
"path": "file path exactly as it appears in the diff (`+++ b/` side)",
|
|
"line": <int, the NEW-file line number the issue is on, within the diff>,
|
|
"problem": "one line: what is wrong",
|
|
"fix": "one line: how to fix it",
|
|
"suggestion": "<exact replacement lines for that location, or empty string if you cannot produce safe replacement code>"
|
|
}
|
|
]
|
|
}
|
|
|
|
Rules:
|
|
- `line` MUST be a line number that exists in the post-change version of `path`
|
|
(i.e. a context line or an added `+` line shown in the diff). Never a removed
|
|
line. If you are unsure of the exact line, set `line` to the closest context
|
|
line you CAN see in the diff.
|
|
- `suggestion` is the literal new code that should replace the flagged line(s).
|
|
Keep it minimal — just the changed lines, indented as they would appear in the
|
|
file. Leave it empty ("") if a safe textual replacement is not possible (e.g.
|
|
a missing test, an architectural note).
|
|
- Skip nitpicks, pure formatting, and praise. At most ~15 findings, highest
|
|
severity first.
|
|
- If the diff is clean, output: {"findings": []}
|
|
- Do NOT repeat anything already covered in "PREVIOUS REVIEWS" — only surface
|
|
new or still-unresolved issues."""
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pure helpers (unit-tested, no network)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def truncate_diff(text: str, max_chars: int) -> tuple[str, bool, int]:
|
|
"""Return (text, was_truncated, original_len). Never raises on bad input."""
|
|
if text is None:
|
|
return "", False, 0
|
|
orig_len = len(text)
|
|
if orig_len <= max_chars:
|
|
return text, False, orig_len
|
|
return text[:max_chars] + f"\n\n[diff truncated at {max_chars} characters]\n", True, orig_len
|
|
|
|
|
|
def parse_text_blocks(content: list) -> str:
|
|
"""Join `type:"text"` blocks from an Anthropic /v1/messages response.
|
|
|
|
Drops `thinking` blocks (glm-5.2:cloud is a reasoning model and emits them).
|
|
Tolerates missing/malformed blocks by skipping them.
|
|
"""
|
|
if not isinstance(content, list):
|
|
return ""
|
|
out = []
|
|
for block in content:
|
|
if not isinstance(block, dict):
|
|
continue
|
|
if block.get("type") == "text" and isinstance(block.get("text"), str):
|
|
out.append(block["text"])
|
|
return "\n".join(out).strip()
|
|
|
|
|
|
def format_review_body(findings: str, model: str, sha: str, summary: str = "") -> str:
|
|
"""Format the posted review summary body.
|
|
|
|
`findings` is the bullet text for findings that could NOT be anchored inline
|
|
(or, on the legacy/no-inline path, the whole review). Empty -> "No issues
|
|
found.". `summary` (optional, opencode engine) is rendered as a "Summary"
|
|
section right under the header. The hidden sha marker is always appended
|
|
for the dedupe pass.
|
|
"""
|
|
header = REVIEW_HEADER.format(model=model, sha=sha[:8] if sha else "unknown")
|
|
findings = (findings or "").strip()
|
|
if not findings:
|
|
findings = "No issues found."
|
|
marker = SHA_MARKER.format(sha=sha) if sha else ""
|
|
parts = [header]
|
|
if summary:
|
|
parts.append(summary.strip())
|
|
parts.append(findings)
|
|
body = "\n\n".join(parts)
|
|
if marker:
|
|
body += f"\n{marker}"
|
|
return body
|
|
|
|
|
|
def build_user_prompt(
|
|
title: str,
|
|
body: str,
|
|
diff: str,
|
|
config: dict | None = None,
|
|
prior_reviews: list[str] | None = None,
|
|
) -> str:
|
|
"""Assemble the user prompt: repo config + prior reviews + PR meta + diff."""
|
|
parts: list[str] = []
|
|
|
|
if config:
|
|
cfg_lines = []
|
|
if config.get("focus"):
|
|
cfg_lines.append("Focus areas: " + ", ".join(config["focus"]))
|
|
if config.get("exclude_paths"):
|
|
cfg_lines.append("Ignore paths: " + ", ".join(config["exclude_paths"]))
|
|
if config.get("languages"):
|
|
cfg_lines.append("Languages: " + ", ".join(config["languages"]))
|
|
if config.get("instructions"):
|
|
cfg_lines.append("Instructions:\n" + str(config["instructions"]).strip())
|
|
if cfg_lines:
|
|
parts.append("## Repo review config (.pr-review.json)\n" + "\n".join(cfg_lines))
|
|
|
|
if prior_reviews:
|
|
joined = "\n\n---\n\n".join(prior_reviews)
|
|
if len(joined) > 8000:
|
|
joined = joined[:8000] + "\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 (" " or anything else within a hunk)
|
|
if raw.startswith(" "):
|
|
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:
|
|
"""Return the substring of the last fenced ```json block in text, or None.
|
|
Falls back to _extract_first_json_object when no fence is present."""
|
|
s = text or ""
|
|
# Find all ```json ... ``` fenced blocks; take the last.
|
|
blocks = list(re.finditer(r"```(?:json)?\s*(\{.*?\})\s*```", s, re.DOTALL))
|
|
if blocks:
|
|
return blocks[-1].group(1)
|
|
return _extract_first_json_object(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.
|
|
"""
|
|
data = _parse_json_tolerant(text)
|
|
if not isinstance(data, dict):
|
|
return []
|
|
findings = data.get("findings")
|
|
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
|
|
|
|
|
|
def parse_review_output(text: str) -> tuple[str, list[dict]]:
|
|
"""Parse the opengine's stdout into (summary, findings).
|
|
|
|
Accepts `{"summary": "...", "findings": [...]}` (the opencode pragent agent)
|
|
or a bare `{"findings": [...]}`. `summary` defaults to "". Uses the LAST
|
|
```json fenced block (the pragent agent emits JSON as the final block), with
|
|
a tolerant fallback. Never raises.
|
|
"""
|
|
blob = _last_json_block(text)
|
|
if blob is None:
|
|
return "", []
|
|
try:
|
|
data = json.loads(blob)
|
|
except json.JSONDecodeError:
|
|
return "", []
|
|
if not isinstance(data, dict):
|
|
return "", []
|
|
summary = str(data.get("summary", "") or "").strip()
|
|
findings = data.get("findings")
|
|
out = []
|
|
if isinstance(findings, list):
|
|
for f in findings:
|
|
n = _normalize_finding(f)
|
|
if n is not None:
|
|
out.append(n)
|
|
return summary, out
|
|
|
|
|
|
def _parse_json_tolerant(text: str) -> dict | None:
|
|
"""Parse a JSON object from text: try the last fenced block, then a direct
|
|
parse, then the first balanced object. Returns None on any failure."""
|
|
if not text:
|
|
return None
|
|
blob = _last_json_block(text)
|
|
if blob is not None:
|
|
try:
|
|
d = json.loads(blob)
|
|
if isinstance(d, dict):
|
|
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):
|
|
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):
|
|
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
|
|
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 == "{":
|
|
depth += 1
|
|
elif c == "}":
|
|
depth -= 1
|
|
if depth == 0:
|
|
return s[start:i + 1]
|
|
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 inline_comment_body(f: dict) -> str:
|
|
"""Render one finding as a positional review-comment body.
|
|
|
|
Includes a ```suggestion fence only if the model produced non-empty
|
|
replacement code. Gitea renders that as an apply-able suggestion. Appends a
|
|
`📎 ref:` link when the finding carries a `reference` URL.
|
|
"""
|
|
sev = f["severity"].upper()
|
|
body = f"**[{sev}]** {f['problem']}"
|
|
if f["fix"]:
|
|
body += f"\n\nFix: {f['fix']}"
|
|
if f["suggestion"]:
|
|
body += f"\n\n```suggestion\n{f['suggestion']}\n```"
|
|
ref = f.get("reference", "")
|
|
if ref:
|
|
body += f"\n\n📎 ref: {ref}"
|
|
return body
|
|
|
|
|
|
def summary_bullets(findings: list[dict]) -> str:
|
|
"""Render unanchored findings as summary-body bullets (no line anchor)."""
|
|
lines = []
|
|
for f in findings:
|
|
loc = f"{f['path']}:{f['line']}" if f["line"] else f["path"]
|
|
fix = f" — fix: {f['fix']}" if f["fix"] else ""
|
|
ref = f" ({f.get('reference', '')})" if f.get("reference") else ""
|
|
lines.append(f"- **[{f['severity'].upper()}]** `{loc}` — {f['problem']}{fix}{ref}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Repo config + existing-review helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_repo_config(raw: str) -> dict:
|
|
"""Parse a .pr-review.json blob tolerantly. Returns {} on any failure."""
|
|
if not raw:
|
|
return {}
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
out = {}
|
|
for k in ("focus", "exclude_paths", "languages"):
|
|
v = data.get(k)
|
|
if isinstance(v, list) and all(isinstance(x, str) for x in v):
|
|
out[k] = v
|
|
instr = data.get("instructions")
|
|
if isinstance(instr, str) and instr.strip():
|
|
out["instructions"] = instr.strip()
|
|
return out
|
|
|
|
|
|
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]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Network helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _http(method: str, url: str, token: str, body: dict | None = None, accept: str = "application/json") -> tuple[int, bytes]:
|
|
headers = {"Authorization": f"token {token}", "Accept": accept}
|
|
data = None
|
|
if body is not None:
|
|
data = json.dumps(body).encode()
|
|
headers["Content-Type"] = "application/json"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=180) as r:
|
|
return r.status, r.read()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read()
|
|
except urllib.error.URLError as e:
|
|
raise RuntimeError(f"network error: {e.reason}") from e
|
|
|
|
|
|
def gitea_get(api: str, repo: str, path: str, token: str, accept: str = "application/json") -> tuple[int, bytes]:
|
|
return _http("GET", f"{api}/api/v1/repos/{repo}/{path}", token, None, accept)
|
|
|
|
|
|
def gitea_post(api: str, repo: str, path: str, token: str, body: dict) -> tuple[int, bytes]:
|
|
return _http("POST", f"{api}/api/v1/repos/{repo}/{path}", token, body)
|
|
|
|
|
|
def fetch_pr_diff(api: str, repo: str, index: str, token: str, max_chars: int) -> tuple[str, bool, int]:
|
|
"""Get the unified diff. Try the `.diff` suffix first, fall back to the
|
|
files endpoint (join `patch` fields) if the server does not serve .diff."""
|
|
status, raw = gitea_get(api, repo, f"pulls/{index}.diff", token, accept="text/plain")
|
|
if status == 200:
|
|
return truncate_diff(raw.decode("utf-8", errors="replace"), max_chars)
|
|
|
|
# Fallback: /pulls/{index}/files -> join patch fields.
|
|
status, raw = gitea_get(api, repo, f"pulls/{index}/files", token)
|
|
if status != 200:
|
|
raise RuntimeError(f"could not fetch diff: .diff={status}, files={status}")
|
|
files = json.loads(raw)
|
|
joined = []
|
|
for f in files:
|
|
h = f.get("filename", "?")
|
|
joined.append(f"--- {h}\n+++ {h}\n{f.get('patch', '(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, sha: str, token: str) -> dict:
|
|
"""Fetch .pr-review.json from the PR's head ref. {} if absent/unreadable."""
|
|
if not sha:
|
|
return {}
|
|
status, raw = gitea_get(api, repo, f"contents/{REPO_CONFIG_FILE}?ref={sha}", 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 the
|
|
```suggestion fence 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 so the findings still land somewhere.
|
|
post_review(api, repo, index, token, summary)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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,
|
|
) -> 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).
|
|
|
|
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
|
|
|
|
diff, _truncated, _orig = fetch_pr_diff(api, repo, index, token, max_chars)
|
|
if not 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, sha, token)
|
|
prior = prior_review_bodies(reviews, sha)
|
|
|
|
engine = os.environ.get("PRAGENT_ENGINE", "opencode").strip().lower()
|
|
review_summary = ""
|
|
if engine == "opencode":
|
|
# The review "brain" runs on opencode: it gets the checked-out repo,
|
|
# the brief, and the pragent agent factory; returns stdout with a
|
|
# summary + findings JSON. We parse + anchor + post here.
|
|
import opencode_review # local import keeps the ollama path dep-free
|
|
# opencode wants a provider-prefixed model ref (headroom/glm-5.2:cloud);
|
|
# `model` here is the bare id (OLLAMA_MODEL). OPENCODE_MODEL overrides
|
|
# with the full ref; otherwise we prefix the configured provider.
|
|
oc_model = os.environ.get("OPENCODE_MODEL") or f"headroom/{model}"
|
|
stdout = 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,
|
|
)
|
|
review_summary, findings = parse_review_output(stdout)
|
|
if not findings and not review_summary:
|
|
# opencode produced nothing parseable — fall back to a note.
|
|
post_review(api, repo, index, token, format_review_body(
|
|
"AI review produced no parseable output.", model, sha))
|
|
return True
|
|
else:
|
|
user_prompt = build_user_prompt(title, body, diff, config, prior)
|
|
raw_findings = call_model(ollama_url, model, SYSTEM_PROMPT, user_prompt, max_tokens)
|
|
findings = parse_findings(raw_findings)
|
|
|
|
anchors = parse_diff_anchors(diff)
|
|
anchored, unanchored = split_findings(findings, anchors)
|
|
|
|
# Summary body: the unanchored bullets (or "No issues found."), plus a
|
|
# one-line note when inline comments were posted so the summary isn't
|
|
# empty-looking. The opencode engine also carries a prose summary.
|
|
bullets = summary_bullets(unanchored)
|
|
summary_parts = []
|
|
if anchored:
|
|
summary_parts.append(f"_{len(anchored)} inline comment(s) posted below._")
|
|
if bullets:
|
|
summary_parts.append(bullets)
|
|
if not summary_parts:
|
|
summary_parts.append("No issues found.")
|
|
summary_body = format_review_body("\n\n".join(summary_parts), model, sha, summary=review_summary)
|
|
|
|
post_inline_review(api, repo, index, token, summary_body, anchored)
|
|
print(
|
|
f"pragent: reviewed {repo}#{index} sha={sha[:8]} "
|
|
f"engine={engine} findings={len(findings)} inline={len(anchored)}",
|
|
flush=True,
|
|
)
|
|
return True
|
|
except Exception as e: # fail-open
|
|
try:
|
|
post_review(api, repo, index, token, format_review_body(f"⚠️ AI review failed: {e}", model, sha))
|
|
except Exception as e2:
|
|
print(f"pragent: could not post failure note: {e2}", file=sys.stderr)
|
|
print(f"pragent: review failed: {e}", file=sys.stderr)
|
|
return False
|
|
|
|
|
|
def run() -> int:
|
|
review_pr(
|
|
api=_need("GITEA_API"),
|
|
repo=_need("GITEA_REPOSITORY"),
|
|
index=_need("PR_INDEX"),
|
|
title=os.environ.get("PR_TITLE", ""),
|
|
body=os.environ.get("PR_BODY", ""),
|
|
sha=os.environ.get("PRAGENT_SHA", ""),
|
|
token=_need("PRAGENT_BOT_TOKEN"),
|
|
ollama_url=_need("OLLAMA_URL"),
|
|
model=_need("OLLAMA_MODEL"),
|
|
max_tokens=int(os.environ.get("OLLAMA_MAX_TOKENS", "8000")),
|
|
max_chars=int(os.environ.get("DIFF_MAX_CHARS", "150000")),
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(run()) |