aitbc
4c81d9c32e
All checks were successful
Documentation Validation / validate-docs (push) Successful in 9s
- Move documentation files to docs/summaries/ - Move temporary files to temp/ directory - Keep only essential files in root directory - Improve project structure and maintainability
274 lines
9.9 KiB
Python
274 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Automated PR reviewer for multi-agent collaboration.
|
|
|
|
Fetches open PRs authored by the sibling agent, runs basic validation,
|
|
and posts an APPROVE or COMMENT review.
|
|
|
|
Usage: GITEA_TOKEN=... python3 auto_review.py
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
import shutil
|
|
from datetime import datetime
|
|
|
|
TOKEN = os.getenv("GITEA_TOKEN")
|
|
API_BASE = os.getenv("GITEA_API_BASE", "http://gitea.bubuit.net:3000/api/v1")
|
|
REPO = "oib/aitbc"
|
|
SELF = os.getenv("AGENT_NAME", "aitbc") # set this in env: aitbc or aitbc1
|
|
OTHER = "aitbc1" if SELF == "aitbc" else "aitbc"
|
|
|
|
def log(msg):
|
|
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}")
|
|
|
|
def die(msg):
|
|
log(f"FATAL: {msg}")
|
|
sys.exit(1)
|
|
|
|
def api_get(path):
|
|
cmd = ["curl", "-s", "-H", f"Authorization: token {TOKEN}", f"{API_BASE}/{path}"]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
return None
|
|
try:
|
|
return json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
def api_post(path, payload):
|
|
cmd = ["curl", "-s", "-X", "POST", "-H", f"Authorization: token {TOKEN}", "-H", "Content-Type: application/json",
|
|
f"{API_BASE}/{path}", "-d", json.dumps(payload)]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
return None
|
|
try:
|
|
return json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
def get_open_prs():
|
|
return api_get(f"repos/{REPO}/pulls?state=open") or []
|
|
|
|
def get_my_reviews(pr_number):
|
|
return api_get(f"repos/{REPO}/pulls/{pr_number}/reviews") or []
|
|
|
|
# Stability ring definitions
|
|
RING_PREFIXES = [
|
|
(0, ["packages/py/aitbc-core", "packages/py/aitbc-sdk"]), # Ring 0: Core
|
|
(1, ["apps/"]), # Ring 1: Platform services
|
|
(2, ["cli/", "analytics/", "tools/"]), # Ring 2: Application
|
|
]
|
|
RING_THRESHOLD = {0: 0.90, 1: 0.80, 2: 0.70, 3: 0.50} # Ring 3: Experimental/low
|
|
|
|
def is_test_file(path):
|
|
"""Heuristic: classify test files to downgrade ring."""
|
|
if '/tests/' in path or path.startswith('tests/') or path.endswith('_test.py'):
|
|
return True
|
|
return False
|
|
|
|
def detect_ring(workdir, base_sha, head_sha):
|
|
"""Determine the stability ring of the PR based on changed files."""
|
|
try:
|
|
# Get list of changed files between base and head
|
|
output = subprocess.run(
|
|
["git", "--git-dir", os.path.join(workdir, ".git"), "diff", "--name-only", base_sha, head_sha],
|
|
capture_output=True, text=True, check=True
|
|
).stdout
|
|
files = [f.strip() for f in output.splitlines() if f.strip()]
|
|
except subprocess.CalledProcessError:
|
|
files = []
|
|
|
|
# If all changed files are tests, treat as Ring 3 (low risk)
|
|
if files and all(is_test_file(f) for f in files):
|
|
return 3
|
|
|
|
# Find highest precedence ring (lowest number) among changed files
|
|
for ring, prefixes in sorted(RING_PREFIXES, key=lambda x: x[0]):
|
|
for p in files:
|
|
if any(p.startswith(prefix) for prefix in prefixes):
|
|
return ring
|
|
return 3 # default to Ring 3 (experimental)
|
|
|
|
def checkout_pr_branch(pr):
|
|
"""Checkout PR branch in a temporary worktree."""
|
|
tmpdir = tempfile.mkdtemp(prefix="aitbc_review_")
|
|
try:
|
|
# Clone just .git into tmp, then checkout
|
|
subprocess.run(["git", "clone", "--no-checkout", "origin", tmpdir], check=True, capture_output=True)
|
|
worktree = os.path.join(tmpdir, "wt")
|
|
os.makedirs(worktree)
|
|
subprocess.run(["git", "--git-dir", os.path.join(tmpdir, ".git"), "--work-tree", worktree, "fetch", "origin", pr['head']['ref']], check=True, capture_output=True)
|
|
subprocess.run(["git", "--git-dir", os.path.join(tmpdir, ".git"), "--work-tree", worktree, "checkout", "FETCH_HEAD"], check=True, capture_output=True)
|
|
return worktree, tmpdir
|
|
except subprocess.CalledProcessError as e:
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
log(f"Checkout failed: {e}")
|
|
return None, None
|
|
|
|
def run_checks(workdir):
|
|
"""Run validation checks. Returns (pass, score, notes)."""
|
|
notes = []
|
|
score = 0.0
|
|
|
|
# 1. Import sanity: try to import the aitbc_cli module
|
|
try:
|
|
subprocess.run([sys.executable, "-c", "import aitbc_cli.main"], check=True, cwd=workdir, capture_output=True)
|
|
notes.append("CLI imports OK")
|
|
score += 0.3
|
|
except subprocess.CalledProcessError as e:
|
|
notes.append(f"CLI import failed: {e.stderr.decode().strip()}")
|
|
return False, 0.0, "\n".join(notes)
|
|
|
|
# 2. Check if tests exist and run them (if tests/ directory present)
|
|
tests_dir = os.path.join(workdir, "tests")
|
|
if os.path.exists(tests_dir):
|
|
try:
|
|
# Run pytest quietly
|
|
result = subprocess.run([sys.executable, "-m", "pytest", "-q"], cwd=workdir, capture_output=True, text=True, timeout=60)
|
|
if result.returncode == 0:
|
|
notes.append("All tests passed")
|
|
score += 0.4
|
|
# Parse coverage if available? Not implemented
|
|
else:
|
|
notes.append(f"Tests failed (exit {result.returncode}): {result.stdout[-500:]}")
|
|
return False, score, "\n".join(notes)
|
|
except subprocess.TimeoutExpired:
|
|
notes.append("Tests timed out")
|
|
return False, score, "\n".join(notes)
|
|
except Exception as e:
|
|
notes.append(f"Test run error: {e}")
|
|
return False, score, "\n".join(notes)
|
|
else:
|
|
notes.append("No tests directory; skipping test run")
|
|
score += 0.1 # small baseline
|
|
|
|
# 3. Check for syntax errors in all Python files (quick scan)
|
|
py_files = subprocess.run(["find", workdir, "-name", "*.py", "-not", "-path", "*/.*"], capture_output=True, text=True).stdout.strip().split("\n")
|
|
syntax_errors = []
|
|
for py in py_files[:50]: # limit to first 50
|
|
if not py:
|
|
continue
|
|
result = subprocess.run([sys.executable, "-m", "py_compile", py], capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
syntax_errors.append(f"{os.path.basename(py)}: {result.stderr.strip()}")
|
|
if syntax_errors:
|
|
notes.append("Syntax errors:\n" + "\n".join(syntax_errors))
|
|
return False, score, "\n".join(notes)
|
|
else:
|
|
notes.append(f"Syntax OK ({len(py_files)} files checked)")
|
|
score += 0.2
|
|
|
|
# 4. Effort estimate: count files changed
|
|
# We can approximate by using git diff --name-only on the branch compared to main
|
|
try:
|
|
# Get the merge base with main
|
|
base = pr['base']['sha']
|
|
head = pr['head']['sha']
|
|
changed = subprocess.run(["git", "--git-dir", os.path.join(tmpdir, ".git"), "diff", "--name-only", base, head], capture_output=True, text=True).stdout.strip().split("\n")
|
|
num_files = len([f for f in changed if f])
|
|
if num_files < 5:
|
|
score += 0.1
|
|
elif num_files < 15:
|
|
score += 0.05
|
|
else:
|
|
score -= 0.05
|
|
notes.append(f"Changed files: {num_files}")
|
|
except Exception as e:
|
|
notes.append(f"Could not compute changed files: {e}")
|
|
|
|
# Normalize score to 0-1 range (max ~1.0)
|
|
score = min(max(score, 0.0), 1.0)
|
|
return True, score, "\n".join(notes)
|
|
|
|
def post_review(pr_number, body, event="COMMENT"):
|
|
payload = {"body": body, "event": event}
|
|
return api_post(f"repos/{REPO}/pulls/{pr_number}/reviews", payload)
|
|
|
|
def main():
|
|
log(f"Starting auto-review (agent={SELF}, watching for PRs from {OTHER})")
|
|
prs = get_open_prs()
|
|
if not prs:
|
|
log("No open PRs found.")
|
|
return
|
|
|
|
for pr in prs:
|
|
author = pr['user']['login']
|
|
if author != OTHER:
|
|
continue # only review sibling's PRs
|
|
|
|
pr_number = pr['number']
|
|
title = pr['title']
|
|
head = pr['head']['ref']
|
|
base = pr['base']['ref']
|
|
|
|
log(f"Reviewing PR#{pr_number}: {title} (head: {head})")
|
|
|
|
# Check if I already reviewed
|
|
reviews = get_my_reviews(pr_number)
|
|
if any(r['user']['login'] == SELF for r in reviews):
|
|
log(f"Already reviewed PR#{pr_number}; skipping")
|
|
continue
|
|
|
|
# Checkout and run tests
|
|
workdir, tmpdir = checkout_pr_branch(pr)
|
|
if not workdir:
|
|
log(f"Failed to checkout PR#{pr_number}; skipping")
|
|
continue
|
|
|
|
try:
|
|
# Determine stability ring and threshold
|
|
base_sha = pr['base']['sha']
|
|
head_sha = pr['head']['sha']
|
|
ring = detect_ring(workdir, base_sha, head_sha)
|
|
threshold = RING_THRESHOLD[ring]
|
|
|
|
ok, score, notes = run_checks(workdir)
|
|
notes = f"Ring: {ring}\nThreshold: {threshold}\n{notes}"
|
|
finally:
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
if ok and score >= threshold:
|
|
review_body = f"""**Auto-review by {SELF}**
|
|
|
|
✅ **APPROVED**
|
|
|
|
Confidence score: {score:.2f}
|
|
|
|
**Validation notes:**
|
|
{notes}
|
|
|
|
This PR meets quality criteria and can be merged."""
|
|
result = post_review(pr_number, review_body, event="APPROVE")
|
|
if result:
|
|
log(f"PR#{pr_number} APPROVED (score={score:.2f})")
|
|
else:
|
|
log(f"Failed to post approval for PR#{pr_number}")
|
|
else:
|
|
review_body = f"""**Auto-review by {SELF}**
|
|
|
|
⚠️ **CHANGES REQUESTED**
|
|
|
|
Confidence score: {score:.2f}
|
|
|
|
**Validation notes:**
|
|
{notes}
|
|
|
|
Please address the issues above and push again."""
|
|
result = post_review(pr_number, review_body, event="REQUEST_CHANGES")
|
|
if result:
|
|
log(f"PR#{pr_number} CHANGE REQUESTED (score={score:.2f})")
|
|
else:
|
|
log(f"Failed to post review for PR#{pr_number}")
|
|
|
|
log("Auto-review complete.")
|
|
|
|
if __name__ == "__main__":
|
|
if not TOKEN:
|
|
die("GITEA_TOKEN environment variable required")
|
|
if SELF not in ("aitbc", "aitbc1"):
|
|
die("AGENT_NAME must be set to 'aitbc' or 'aitbc1'")
|
|
main() |