#!/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. 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. 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 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) 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 DIFF_MAX_CHARS (optional) diff truncation cap, default 150000 """ import json import os import sys import urllib.error import urllib.request REVIEW_HEADER = "🤖 **AI Review** · pragent pilot · {model} · `{sha}`" 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. 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. 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.""" # --------------------------------------------------------------------------- # 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) -> str: """Format the posted review body. Findings empty -> "No issues found.".""" 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}" def build_user_prompt(title: str, body: str, diff: str) -> str: parts = [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) # --------------------------------------------------------------------------- # 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 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: 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}") # --------------------------------------------------------------------------- # 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 = 6000, max_chars: int = 150000, ) -> bool: """Run one review and post it as a PR comment. 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. """ try: 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)) 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", "6000")), max_chars=int(os.environ.get("DIFF_MAX_CHARS", "150000")), ) return 0 if __name__ == "__main__": sys.exit(run())