feat(pilot): minimal AI review bot for Gitea Actions
Ships a working pragent pilot ahead of the framework build (design doc deferred). Single stdlib-only reviewer script fetched at runtime by a per-repo Gitea Action; reviews fire only on PRs with the AI-REVIEW label; model is glm-5.2:cloud via the on-network headroom proxy; fail-open. - pilot/ai_review.py: fetch PR diff, call model, post review as pragent-bot - pilot/workflow-template.yml: per-repo Gitea Action gated on AI-REVIEW label - pilot/README.md: onboarding (bot collaborator + secret + workflow + label) - tests/pilot/test_ai_review.py: 15 unit tests for pure helpers (no network) - README/design doc: note pilot is the bootstrap; framework build deferred Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Marcos
@@ -0,0 +1,221 @@
|
||||
#!/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 run() -> int:
|
||||
api = _need("GITEA_API")
|
||||
repo = _need("GITEA_REPOSITORY")
|
||||
index = _need("PR_INDEX")
|
||||
token = _need("PRAGENT_BOT_TOKEN")
|
||||
ollama_url = _need("OLLAMA_URL")
|
||||
model = _need("OLLAMA_MODEL")
|
||||
title = os.environ.get("PR_TITLE", "")
|
||||
body = os.environ.get("PR_BODY", "")
|
||||
sha = os.environ.get("PRAGENT_SHA", "")
|
||||
max_tokens = int(os.environ.get("OLLAMA_MAX_TOKENS", "6000"))
|
||||
max_chars = int(os.environ.get("DIFF_MAX_CHARS", "150000"))
|
||||
|
||||
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 0
|
||||
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))
|
||||
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 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(run())
|
||||
Reference in New Issue
Block a user