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

pilot: dedupe + repo config + inline comments w/ suggestions

Browse Source 
- Dedupe: Gitea-as-state. Scan existing reviews for a hidden
  <!-- pragent:sha=... --> marker matching the head sha; skip if present
  (kills duplicate reviews on label-toggle / re-fire). Prior review bodies
  fed back as 'already said' context (light framework §6.1).
- Repo-local focus: optional .pr-review.json at repo root
  ({focus,exclude_paths,languages,instructions}), fetched at head ref.
- Inline comments + apply-able suggestions: model emits JSON findings
  {severity,path,line,problem,fix,suggestion}; diff hunks parsed into valid
  (path,new_line) RIGHT-side anchors; anchored findings become positional
  review comments with a ```suggestion fence (Gitea apply-button);
  unanchored findings fold into the summary body.
- Tests: parse_diff_anchors, parse_findings (tolerant JSON), split_findings,
  inline_comment_body, summary_bullets, parse_repo_config, reviewed_shas,
  prior_review_bodies, sha-marker. 35 pass.
- Bump OLLAMA_MAX_TOKENS default 6000 -> 8000 (suggestions add length).

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Marcos
2026-08-17 19:59:36 +00:00
parent 789fb38bae
commit 90cea84f6f
4 changed files with 734 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pilot/ai_review.py
+463 -30
View File
@@ -1,51 +1,102 @@
#!/usr/bin/env python3
"""pragent pilot — minimal AI PR reviewer.
Runs as a Gitea Actions step. 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 a PR review authored by pragent-bot.
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 the process always exits 0 so it can never block CI.
and review_pr never raises. Stdlib only — no pip install.
Stdlib only — no pip install, fast cold start in CI.
Env:
GITEA_API base URL of the in-cluster Gitea, e.g. http://gitea-http.gitea.svc.cluster.local:3000
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 (github.event.pull_request.head.sha)
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 6000
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.
For each issue, output exactly one line in this format:
- [SEVERITY] path:line — concise problem. suggested fix.
where SEVERITY is one of: critical, high, medium, low.
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:
- Skip nitpicks, pure formatting, and praise.
- If the diff is clean, output exactly: No issues found.
- Be concise. At most ~15 findings, highest severity first.
- Do not restate the diff. Do not include a summary header. Just the findings lines."""
- `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)
@@ -80,16 +131,53 @@ def parse_text_blocks(content: list) -> str:
def format_review_body(findings: str, model: str, sha: str) -> str:
"""Format the posted review body. Findings empty -> "No issues found."."""
"""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.". 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."
return f"{header}\n\n{findings}"
marker = SHA_MARKER.format(sha=sha) if sha else ""
body = f"{header}\n\n{findings}"
if marker:
body += f"\n{marker}"
return body
def build_user_prompt(title: str, body: str, diff: str) -> str:
parts = [f"## PR\nTitle: {title or '(none)'}"]
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:
@@ -99,6 +187,256 @@ def build_user_prompt(title: str, body: str, diff: str) -> str:
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 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.
"""
if not text:
return []
s = text.strip()
# Strip a single wrapping code fence if present.
if s.startswith("```"):
s = re.sub(r"^```[a-zA-Z]*\n?", "", s)
s = re.sub(r"\n?```$", "", s).strip()
data = None
try:
data = json.loads(s)
except json.JSONDecodeError:
obj = _extract_first_json_object(s)
if obj is not None:
try:
data = json.loads(obj)
except json.JSONDecodeError:
data = None
if not isinstance(data, dict):
return []
findings = data.get("findings")
if not isinstance(findings, list):
return []
out = []
for f in findings:
if not isinstance(f, dict):
continue
path = f.get("path")
line = f.get("line")
if not isinstance(path, str) or not path.strip():
continue
if not isinstance(line, int) or line < 1:
continue
sev = str(f.get("severity", "medium")).strip().lower()
if sev not in SEVERITIES:
sev = "medium"
out.append({
"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(),
})
return out
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.
"""
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```"
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 ""
lines.append(f"- **[{f['severity'].upper()}]** `{loc}` — {f['problem']}{fix}")
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
# ---------------------------------------------------------------------------
@@ -147,6 +485,35 @@ def fetch_pr_diff(api: str, repo: str, index: str, token: str, max_chars: int) -
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,
@@ -167,6 +534,7 @@ def call_model(ollama_url: str, model: str, system: str, user: str, max_tokens:
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.
@@ -175,6 +543,33 @@ def post_review(api: str, repo: str, index: str, token: str, body: str) -> None:
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`:
{path, side:"RIGHT", line, body}. The body carries the ```suggestion
fence when the model produced replacement code.
"""
comments = [
{
"path": f["path"],
"side": "RIGHT",
"line": f["line"],
"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
# ---------------------------------------------------------------------------
@@ -197,23 +592,61 @@ def review_pr(
token: str,
ollama_url: str,
model: str,
max_tokens: int = 6000,
max_tokens: int = 8000,
max_chars: int = 150000,
) -> bool:
"""Run one review and post it as a PR comment.
"""Run one review and post it as `pragent-bot`.
Returns True on success, False on failure (failure note is posted when
possible). Never raises — fail-open by design. Both the CI `run()` entry
point and the central webhook server call this.
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
user_prompt = build_user_prompt(title, body, diff)
findings = call_model(ollama_url, model, SYSTEM_PROMPT, user_prompt, max_tokens)
post_review(api, repo, index, token, format_review_body(findings, model, sha))
config = fetch_repo_config(api, repo, sha, token)
prior = prior_review_bodies(reviews, sha)
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.
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)
post_inline_review(api, repo, index, token, summary_body, anchored)
print(
f"pragent: reviewed {repo}#{index} sha={sha[:8]} "
f"findings={len(findings)} inline={len(anchored)}",
flush=True,
)
return True
except Exception as e: # fail-open
try:
@@ -235,7 +668,7 @@ def run() -> int:
token=_need("PRAGENT_BOT_TOKEN"),
ollama_url=_need("OLLAMA_URL"),
model=_need("OLLAMA_MODEL"),
max_tokens=int(os.environ.get("OLLAMA_MAX_TOKENS", "6000")),
max_tokens=int(os.environ.get("OLLAMA_MAX_TOKENS", "8000")),
max_chars=int(os.environ.get("DIFF_MAX_CHARS", "150000")),
)
return 0