refactor: restructure into proper project layout + improve docs
Project layout: - src/: application modules (run_server, auto_merge_bot, health-check, sync-key, trivial_merge, callback_server, start_server) - scripts/: setup/deployment helpers (setup_all, setup_app, generate_manifest) - templates/: manifest.json (GitHub App manifest template) - docs/ + CONTRIBUTING.md: documentation Improvements: - flake.nix: added pr-agent-auto-merge wrapper binary, updated installPhase paths - deploy.yml: syntax check covers all modules including health-check.py and sync-key.py - README.md: comprehensive with architecture, layout, dev, ops, deployment - CONTRIBUTING.md: standards and testing checklist - .gitignore: added *.log, *.pid, .env.* - Cleanup: removed duplicate manifest_current.json / manifest_final.json - Fix: health-check.py docstring updated to claude-opus-5 Verification: - ✅ python3 -m py_compile: all 11 modules pass - ✅ nix flake check: passes
This commit is contained in:
asepharyana
@@ -0,0 +1,294 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PR-Agent Auto-Approve + Auto-Merge Bot
|
||||
Runs periodically (cron), finds open PRs that have been reviewed by PR-Agent,
|
||||
approves them and enables auto-merge.
|
||||
"""
|
||||
import os, sys, json, time, hmac, hashlib, asyncio
|
||||
from pathlib import Path
|
||||
|
||||
# ── Config ──
|
||||
APP_ID = os.environ.get("GITHUB_APP_ID", "4319749")
|
||||
PRIVATE_KEY_PATH = os.environ.get("PRIVATE_KEY_PATH", "/var/lib/pr-agent-server/private-key.pem")
|
||||
WEBHOOK_SECRET = os.environ.get("GITHUB_WEBHOOK_SECRET", "")
|
||||
BASE_URL = os.environ.get("GITHUB_API_BASE", "https://api.github.com")
|
||||
|
||||
def get_jwt():
|
||||
import jwt as pyjwt
|
||||
with open(PRIVATE_KEY_PATH) as f:
|
||||
key = f.read()
|
||||
now = int(time.time())
|
||||
payload = {"iat": now - 60, "exp": now + 600, "iss": APP_ID}
|
||||
return pyjwt.encode(payload, key, algorithm="RS256")
|
||||
|
||||
def get_installation_token(installation_id: int) -> str:
|
||||
"""Get installation access token"""
|
||||
import httpx
|
||||
jwt_token = get_jwt()
|
||||
with httpx.Client() as client:
|
||||
r = client.post(
|
||||
f"{BASE_URL}/app/installations/{installation_id}/access_tokens",
|
||||
headers={"Authorization": f"Bearer {jwt_token}", "Accept": "application/vnd.github.v3+json"}
|
||||
)
|
||||
return r.json().get("token", "")
|
||||
|
||||
def get_all_installations() -> list:
|
||||
"""Get all app installations"""
|
||||
jwt_token = get_jwt()
|
||||
import httpx
|
||||
with httpx.Client() as client:
|
||||
r = client.get(
|
||||
f"{BASE_URL}/app/installations",
|
||||
headers={"Authorization": f"Bearer {jwt_token}", "Accept": "application/vnd.github.v3+json"}
|
||||
)
|
||||
return r.json()
|
||||
|
||||
def get_installation_repos(installation_id: int, token: str) -> list:
|
||||
"""Get repos for an installation"""
|
||||
import httpx
|
||||
with httpx.Client() as client:
|
||||
r = client.get(
|
||||
f"{BASE_URL}/installation/repositories",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
||||
)
|
||||
return r.json().get("repositories", [])
|
||||
|
||||
def get_open_prs(token: str, repo_full: str) -> list:
|
||||
"""Get open PRs in a repo"""
|
||||
import httpx
|
||||
with httpx.Client() as client:
|
||||
r = client.get(
|
||||
f"{BASE_URL}/repos/{repo_full}/pulls?state=open&sort=updated&direction=desc",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
||||
)
|
||||
return r.json()
|
||||
|
||||
def get_pr_reviews(token: str, repo_full: str, pr_number: int) -> list:
|
||||
"""Get reviews for a PR"""
|
||||
import httpx
|
||||
with httpx.Client() as client:
|
||||
r = client.get(
|
||||
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}/reviews",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
||||
)
|
||||
return r.json()
|
||||
|
||||
def post_discord_notification(repo_full: str, pr_number: int, status: str, summary: str = "", score: str = "", url: str = ""):
|
||||
"""Fire-and-forget Discord notification via the server's internal endpoint."""
|
||||
import httpx
|
||||
notify_url = os.environ.get("PR_AGENT_NOTIFY_URL", "http://127.0.0.1:4002/api/v1/notify_review")
|
||||
try:
|
||||
with httpx.Client(timeout=5) as client:
|
||||
client.post(notify_url, json={
|
||||
"repo": repo_full,
|
||||
"pr": pr_number,
|
||||
"status": status,
|
||||
"summary": summary[:500],
|
||||
"score": str(score),
|
||||
"url": url,
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def get_pr_comments(token: str, repo_full: str, pr_number: int) -> list:
|
||||
"""Get issue comments for a PR"""
|
||||
import httpx
|
||||
with httpx.Client() as client:
|
||||
r = client.get(
|
||||
f"{BASE_URL}/repos/{repo_full}/issues/{pr_number}/comments",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
||||
)
|
||||
return r.json()
|
||||
|
||||
def approve_pr(token: str, repo_full: str, pr_number: int) -> bool:
|
||||
"""Submit APPROVE review"""
|
||||
import httpx
|
||||
with httpx.Client() as client:
|
||||
r = client.post(
|
||||
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}/reviews",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"},
|
||||
json={"event": "APPROVE", "body": "✅ Auto-approved by PR-Agent bot."}
|
||||
)
|
||||
return r.status_code == 200
|
||||
|
||||
def merge_pr(token: str, repo_full: str, pr_number: int) -> tuple:
|
||||
"""Attempt to merge the PR"""
|
||||
import httpx
|
||||
with httpx.Client() as client:
|
||||
# Get PR info for SHA
|
||||
pr_r = client.get(
|
||||
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
|
||||
)
|
||||
if pr_r.status_code != 200:
|
||||
return False, f"Can't get PR: {pr_r.status_code}"
|
||||
|
||||
pr_data = pr_r.json()
|
||||
sha = pr_data.get("head", {}).get("sha", "")
|
||||
mergeable = pr_data.get("mergeable", False)
|
||||
|
||||
if mergeable is False:
|
||||
return False, "PR not mergeable (conflicts or checks pending)"
|
||||
|
||||
# Try merge
|
||||
merge_r = client.put(
|
||||
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}/merge",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"},
|
||||
json={
|
||||
"commit_title": f"Auto-merge PR #{pr_number}",
|
||||
"merge_method": "merge",
|
||||
"sha": sha
|
||||
}
|
||||
)
|
||||
if merge_r.status_code == 200:
|
||||
return True, f"Merged: {merge_r.json().get('sha', '')}"
|
||||
else:
|
||||
return False, f"Merge failed: {merge_r.status_code} - {merge_r.json().get('message', '')}"
|
||||
|
||||
def has_bot_comment_with_review(comments: list) -> tuple:
|
||||
"""Check if PR-Agent has posted a review comment and extract quality"""
|
||||
bot_name = "mytheclipsebotreview"
|
||||
for c in comments:
|
||||
if c.get("user", {}).get("login", "").startswith(bot_name):
|
||||
body = c.get("body", "")
|
||||
# Check for PR Reviewer Guide (successful review)
|
||||
if "PR Reviewer Guide" in body:
|
||||
# Extract score if available
|
||||
score = extract_score(body)
|
||||
return True, score
|
||||
return False, 0
|
||||
|
||||
def extract_score(body: str) -> int:
|
||||
"""Extract review score from bot comment"""
|
||||
import re
|
||||
# Look for score patterns like "Score: 8" or "⏱️ Estimated effort"
|
||||
# For now, assume passing if we got a review without errors
|
||||
return 8
|
||||
|
||||
def main():
|
||||
print("=" * 60)
|
||||
print(f"PR-Agent Auto-Approve/Merge Bot - {time.ctime()}")
|
||||
print("=" * 60)
|
||||
|
||||
# Get installations
|
||||
installations = get_all_installations()
|
||||
print(f"Found {len(installations)} installation(s)")
|
||||
|
||||
for inst in installations:
|
||||
inst_id = inst["id"]
|
||||
account = inst["account"]["login"]
|
||||
print(f"\n📦 Installation {inst_id} - @{account}")
|
||||
|
||||
# Get token
|
||||
token = get_installation_token(inst_id)
|
||||
if not token:
|
||||
print(f" ❌ Failed to get token")
|
||||
continue
|
||||
|
||||
# Get repos
|
||||
repos = get_installation_repos(inst_id, token)
|
||||
print(f" Repos: {len(repos)}")
|
||||
|
||||
for repo in repos:
|
||||
repo_full = repo["full_name"]
|
||||
print(f"\n 📁 {repo_full}")
|
||||
|
||||
# Get open PRs
|
||||
prs = get_open_prs(token, repo_full)
|
||||
print(f" Open PRs: {len(prs)}")
|
||||
|
||||
for pr in prs[:5]: # Max 5 per repo
|
||||
pr_num = pr["number"]
|
||||
pr_title = pr["title"]
|
||||
pr_user = pr["user"]["login"]
|
||||
pr_author = pr_user
|
||||
|
||||
print(f" 🔀 PR #{pr_num}: {pr_title[:50]}...")
|
||||
|
||||
# Skip bot PRs
|
||||
if "[bot]" in pr_author or pr_author == "mytheclipsebotreview":
|
||||
print(f" ⏭️ Bot PR, skipping")
|
||||
continue
|
||||
|
||||
# Check if already approved/merged
|
||||
if pr.get("merged", False):
|
||||
print(f" ✅ Already merged")
|
||||
continue
|
||||
|
||||
# Check reviews
|
||||
reviews = get_pr_reviews(token, repo_full, pr_num)
|
||||
bot_approved = any(
|
||||
r.get("user", {}).get("login", "").startswith("mytheclipsebotreview")
|
||||
and r.get("state") == "APPROVED"
|
||||
for r in reviews
|
||||
)
|
||||
|
||||
if bot_approved:
|
||||
print(f" ✅ Already approved. Trying merge...")
|
||||
success, msg = merge_pr(token, repo_full, pr_num)
|
||||
print(f" {'✅' if success else '❌'} Merge: {msg}")
|
||||
continue
|
||||
|
||||
# Check bot comments for review
|
||||
comments = get_pr_comments(token, repo_full, pr_num)
|
||||
has_review, score = has_bot_comment_with_review(comments)
|
||||
|
||||
# ── TRIVIAL PR FAST-PATH ──
|
||||
# Docs-only / version bumps / dependabot / tiny diffs with green
|
||||
# CI skip the AI-fix + score gate and merge directly.
|
||||
if has_review:
|
||||
changed_files, total_lines = [], 0
|
||||
is_trivial = False
|
||||
try:
|
||||
from trivial_merge import (
|
||||
is_trivial_pr, get_pr_changed_files, check_ci_passed,
|
||||
merge_pr as trivial_merge,
|
||||
)
|
||||
changed_files, total_lines = get_pr_changed_files(token, repo_full, pr_num)
|
||||
is_trivial = is_trivial_pr(pr_title, pr_author, changed_files, total_lines)
|
||||
except Exception as e:
|
||||
is_trivial = False
|
||||
print(f" ⚠️ trivial check failed: {e}")
|
||||
|
||||
if is_trivial:
|
||||
print(f" ⚡ TRIVIAL PR ({total_lines} lines, {len(changed_files)} files). Fast-path approve+merge...")
|
||||
ci_ok, ci_msg = check_ci_passed(token, repo_full, pr.get("head", {}).get("sha", ""))
|
||||
if not ci_ok:
|
||||
print(f" ⏳ CI not green: {ci_msg}")
|
||||
continue
|
||||
if approve_pr(token, repo_full, pr_num):
|
||||
print(f" ✅ Approved (trivial)")
|
||||
time.sleep(1)
|
||||
success, msg = trivial_merge(token, repo_full, pr_num, pr.get("head", {}).get("sha", ""))
|
||||
print(f" {'✅ Merged!' if success else '❌ ' + msg}")
|
||||
post_discord_notification(repo_full, pr_num, "done" if success else "failed",
|
||||
summary=f"Trivial PR auto-merged ({total_lines} lines)" if success else f"Trivial merge failed: {msg}",
|
||||
score=score, url=pr.get("html_url", ""))
|
||||
continue
|
||||
|
||||
if has_review and score >= 5:
|
||||
print(f" 📝 Review found (score: {score}). Approving + merging...")
|
||||
|
||||
# Approve
|
||||
if approve_pr(token, repo_full, pr_num):
|
||||
print(f" ✅ Approved!")
|
||||
else:
|
||||
print(f" ❌ Approve failed")
|
||||
continue
|
||||
|
||||
# Small delay
|
||||
time.sleep(2)
|
||||
|
||||
# Merge
|
||||
success, msg = merge_pr(token, repo_full, pr_num)
|
||||
print(f" {'✅ Merged!' if success else '❌ ' + msg}")
|
||||
elif has_review and score < 5:
|
||||
print(f" ⏭️ Review score too low ({score})")
|
||||
else:
|
||||
print(f" ⏳ No bot review yet")
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("Done!")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,54 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Quick callback server to receive GitHub App credentials after manifest creation"""
|
||||
import json, os, sys
|
||||
sys.path.insert(0, os.path.expanduser("~/hermes-agent/.venv/lib/python3.12/site-packages"))
|
||||
|
||||
from fastapi import FastAPI, Request
|
||||
import uvicorn
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/setup/callback")
|
||||
@app.post("/setup/callback")
|
||||
async def callback(request: Request):
|
||||
params = dict(request.query_params)
|
||||
print(f"[CALLBACK] Received params: {json.dumps(params, indent=2)}")
|
||||
|
||||
# If we got a code, exchange it for credentials
|
||||
if "code" in params:
|
||||
import httpx
|
||||
code = params["code"]
|
||||
print(f"[CALLBACK] Exchanging code: {code[:20]}...")
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.post(
|
||||
f"https://api.github.com/app-manifests/{code}/conversions",
|
||||
headers={"Accept": "application/vnd.github.v3+json"}
|
||||
)
|
||||
if resp.status_code == 201:
|
||||
data = resp.json()
|
||||
# Save credentials
|
||||
creds = {
|
||||
"app_id": data.get("id"),
|
||||
"app_slug": data.get("slug"),
|
||||
"pem": data.get("pem"),
|
||||
"webhook_secret": data.get("webhook_secret"),
|
||||
"client_id": data.get("client_id"),
|
||||
"client_secret": data.get("client_secret")
|
||||
}
|
||||
with open("/opt/pr-agent-server/app_credentials.json", "w") as f:
|
||||
json.dump(creds, f, indent=2)
|
||||
print(f"[CALLBACK] App created! ID: {creds['app_id']}, Slug: {creds['app_slug']}")
|
||||
return {"status": "success", "app_id": creds["app_id"], "app_slug": creds["app_slug"]}
|
||||
else:
|
||||
print(f"[CALLBACK] Exchange failed: {resp.status_code} - {resp.text}")
|
||||
return {"status": "error", "detail": resp.text}
|
||||
|
||||
return {"status": "waiting", "params": params}
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
return {"status": "ok"}
|
||||
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(app, host="0.0.0.0", port=3000, log_level="info")
|
||||
@@ -0,0 +1,157 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PR-Agent Model Health Watchdog
|
||||
===============================
|
||||
Runs every 10 minutes via Hermes no_agent cron. Tests the exact model config
|
||||
the pr-agent server uses (primary + fallbacks) against 9router via raw HTTP.
|
||||
|
||||
Output contract (no_agent cron):
|
||||
- OK → empty stdout (silent, $0 idle)
|
||||
- FAIL → one-line alert + detail (delivered to Discord/home channel)
|
||||
|
||||
Design: alert only when EVERY configured model fails (primary AND all
|
||||
fallbacks). If any model works, the server's own fallback chain will succeed,
|
||||
so the system is healthy even if the primary is down/slow. This prevents
|
||||
false alerts from a single slow/failed model.
|
||||
"""
|
||||
import os, sys, json, hashlib, subprocess
|
||||
from pathlib import Path
|
||||
|
||||
BWS_SECRET_ID = "2aef2194-971d-4dae-99dd-b49a0041f97c"
|
||||
ROUTER_BASE = "https://9router.asepharyana.my.id/v1"
|
||||
PRIMARY = "openai/claude-opus-5"
|
||||
FALLBACKS = ["openai/claude-sonnet-5", "openai/claude-haiku-4-5-20251001", "openai/ATLAS", "openai/gemini", "openai/text", "openai/deepseek-v4-flash-free"]
|
||||
# Caddy 9router route is now response_header_timeout 120s / read 300s.
|
||||
# LLM combo TTFT often 30-40s+. Give the check room to complete.
|
||||
HTTP_TIMEOUT = 150
|
||||
CONSECUTIVE_FAIL_FILE = Path("/tmp/pr-agent-health-fail-count")
|
||||
|
||||
# ── key from BWS ────────────────────────────────────────────────────────────
|
||||
def _read_token() -> str:
|
||||
"""Read BWS token. Direct read fails for non-root (root:bws 640), so fall
|
||||
back to `sudo -n cat` (cron user `code` is in sudo group, NOPASSWD)."""
|
||||
for path in (Path("/etc/bws-token"),):
|
||||
try:
|
||||
if path.is_file():
|
||||
return path.read_text().strip()
|
||||
except PermissionError:
|
||||
pass
|
||||
try:
|
||||
r = subprocess.run(["sudo", "-n", "cat", "/etc/bws-token"],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
if r.returncode == 0:
|
||||
return r.stdout.strip()
|
||||
except Exception:
|
||||
pass
|
||||
return ""
|
||||
|
||||
def get_key() -> str:
|
||||
token = os.environ.get("BWS_ACCESS_TOKEN", "")
|
||||
if not token:
|
||||
token = _read_token()
|
||||
if not token:
|
||||
return ""
|
||||
env = {**os.environ, "BWS_ACCESS_TOKEN": token}
|
||||
try:
|
||||
r = subprocess.run(
|
||||
["/usr/local/bin/bws", "secret", "get", BWS_SECRET_ID, "--output", "env"],
|
||||
capture_output=True, text=True, timeout=30, env=env,
|
||||
)
|
||||
if r.returncode != 0:
|
||||
return ""
|
||||
# Value is shell-quoted KEY="value" — take first line only. BWS sometimes
|
||||
# appends "# one or more secrets have been commented-out..."; only the
|
||||
# first line is the real key value.
|
||||
line = r.stdout.split("\n")[0]
|
||||
if "=" not in line:
|
||||
return ""
|
||||
val = line.split("=", 1)[1].strip().strip('"')
|
||||
if len(val) < 10:
|
||||
return ""
|
||||
return val
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
# ── health check ────────────────────────────────────────────────────────────
|
||||
def check_model(model: str, key: str) -> tuple:
|
||||
"""Returns (ok: bool, detail: str). Uses raw HTTP (no litellm dependency).
|
||||
|
||||
NOTE: litellm strips the 'openai/' provider prefix before sending the
|
||||
request body. 9router resolves bare aliases (e.g. 'claude-opus-5') to
|
||||
its own routing; WITH the prefix it tries the 'openai' provider upstream,
|
||||
which has no credentials → 404 'No active credentials for provider: openai'.
|
||||
So we strip the prefix here to mirror exactly what the server sends.
|
||||
"""
|
||||
bare = model.split("/", 1)[-1] if "/" in model else model
|
||||
import urllib.request, urllib.error
|
||||
body = json.dumps({
|
||||
"model": bare,
|
||||
"messages": [{"role": "user", "content": "Reply with the single word OK"}],
|
||||
"max_tokens": 10,
|
||||
}).encode()
|
||||
req = urllib.request.Request(
|
||||
f"{ROUTER_BASE}/chat/completions",
|
||||
data=body,
|
||||
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as r:
|
||||
return r.status == 200, f"HTTP {r.status}"
|
||||
except urllib.error.HTTPError as e:
|
||||
err = e.read().decode(errors="replace")[:160].replace("\n", " ")
|
||||
return False, f"HTTP {e.code}: {err}"
|
||||
except Exception as e:
|
||||
return False, f"{type(e).__name__}: {str(e)[:120]}"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
key = get_key()
|
||||
if not key:
|
||||
print("⚠️ pr-agent health: cannot fetch router key from BWS (bws unavailable)")
|
||||
return 1
|
||||
|
||||
results = {}
|
||||
ok_somewhere = False
|
||||
results[PRIMARY] = check_model(PRIMARY, key)
|
||||
ok_somewhere = ok_somewhere or results[PRIMARY][0]
|
||||
if not ok_somewhere:
|
||||
for fb in FALLBACKS:
|
||||
results[fb] = check_model(fb, key)
|
||||
if results[fb][0]:
|
||||
ok_somewhere = True
|
||||
break # bound runtime; one working model is enough
|
||||
else:
|
||||
# ensure every fallback appears in results for the report
|
||||
for fb in FALLBACKS:
|
||||
results.setdefault(fb, (False, "not tested (prior model failed)"))
|
||||
else:
|
||||
for fb in FALLBACKS:
|
||||
results.setdefault(fb, (True, "not checked (primary ok)"))
|
||||
|
||||
# Any model working = server's fallback chain will succeed = healthy.
|
||||
if ok_somewhere:
|
||||
CONSECUTIVE_FAIL_FILE.unlink(missing_ok=True)
|
||||
return 0
|
||||
|
||||
# Every model failed. Count consecutive to avoid flapping on 1-off glitch.
|
||||
failures = [f"{m} → {d}" for m, (ok, d) in results.items() if not ok]
|
||||
n = 1
|
||||
if CONSECUTIVE_FAIL_FILE.exists():
|
||||
try:
|
||||
n = int(CONSECUTIVE_FAIL_FILE.read_text().strip()) + 1
|
||||
except ValueError:
|
||||
n = 1
|
||||
CONSECUTIVE_FAIL_FILE.write_text(str(n))
|
||||
|
||||
if n < 2:
|
||||
return 0
|
||||
|
||||
detail = " | ".join(failures)
|
||||
key_hash = hashlib.sha256(key.encode()).hexdigest()[:8]
|
||||
print(f"🚨 pr-agent MODELS FAILING ({n} consecutive checks)\n{detail}\nkey hash {key_hash}")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,268 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PR-Agent GitHub App + manifest callback server"""
|
||||
import os, sys, json, time, glob
|
||||
from pathlib import Path
|
||||
|
||||
# Configurable paths (systemd Nix deployment keeps secrets outside the store)
|
||||
APP_DIR = os.environ.get("PR_AGENT_APP_DIR", "/var/lib/pr-agent-server")
|
||||
private_key_path = os.environ.get(
|
||||
"PRIVATE_KEY_PATH", os.path.join(APP_DIR, "private-key.pem")
|
||||
)
|
||||
omni_key_path = os.environ.get(
|
||||
"OMNIROUTE_KEY_PATH", os.path.join(APP_DIR, "omniroute_key")
|
||||
)
|
||||
|
||||
with open(private_key_path) as f:
|
||||
private_key = f.read()
|
||||
|
||||
os.environ["GITHUB__DEPLOYMENT_TYPE"] = "app"
|
||||
os.environ["GITHUB__APP_ID"] = os.environ.get("GITHUB_APP_ID", "4319749")
|
||||
os.environ["GITHUB__PRIVATE_KEY"] = private_key
|
||||
os.environ["GITHUB__WEBHOOK_SECRET"] = os.environ.get("GITHUB_WEBHOOK_SECRET", "")
|
||||
|
||||
with open(omni_key_path) as f:
|
||||
omni_key = f.read().strip()
|
||||
os.environ["OPENAI__API_BASE"] = os.environ.get(
|
||||
"OPENAI_API_BASE", "https://omniroute.imrnes.team/v1"
|
||||
)
|
||||
os.environ["OPENAI__KEY"] = omni_key
|
||||
os.environ["CONFIG__MODEL"] = os.environ.get("PR_AGENT_MODEL", "openai/claude-opus-5")
|
||||
os.environ["CONFIG__FALLBACK_MODELS"] = os.environ.get(
|
||||
"PR_AGENT_FALLBACK_MODELS",
|
||||
'["openai/claude-sonnet-5","openai/claude-haiku-4-5-20251001","openai/ATLAS","openai/gemini","openai/text","openai/deepseek-v4-flash-free"]',
|
||||
)
|
||||
os.environ["CONFIG__CUSTOM_MODEL_MAX_TOKENS"] = os.environ.get(
|
||||
"PR_AGENT_MAX_TOKENS", "128000"
|
||||
)
|
||||
os.environ["GITHUB_APP__PR_COMMANDS"] = os.environ.get(
|
||||
"PR_AGENT_PR_COMMANDS",
|
||||
'["/review --pr_reviewer.require_score_review=true --pr_reviewer.require_security_review=true","/describe","/improve"]',
|
||||
)
|
||||
|
||||
# Analytics folder for PR-Agent structured logs (analytics=True records)
|
||||
ANALYTICS_DIR = os.environ.get("PR_AGENT_ANALYTICS_DIR", "/var/lib/pr-agent-server/analytics")
|
||||
os.makedirs(ANALYTICS_DIR, exist_ok=True)
|
||||
os.environ["CONFIG__ANALYTICS_FOLDER"] = ANALYTICS_DIR
|
||||
|
||||
# Discord webhook for notifications (from BWS secret DISCORD_WEBHOOK_URL)
|
||||
DISCORD_WEBHOOK_URL = os.environ.get("DISCORD_WEBHOOK_URL", "")
|
||||
DISCORD_ALERT_WEBHOOK_URL = os.environ.get("DISCORD_ALERT_WEBHOOK_URL", "")
|
||||
|
||||
sys.path.insert(0, APP_DIR)
|
||||
from pr_agent.servers.github_app import app as pr_agent_app, router as pr_router
|
||||
from fastapi import FastAPI, Request
|
||||
import uvicorn
|
||||
import httpx
|
||||
|
||||
from starlette.middleware import Middleware
|
||||
from starlette_context.middleware import RawContextMiddleware
|
||||
from fastapi.responses import PlainTextResponse, JSONResponse
|
||||
|
||||
app = FastAPI(middleware=[Middleware(RawContextMiddleware)])
|
||||
app.include_router(pr_router)
|
||||
|
||||
|
||||
# ── Analytics / Metrics ─────────────────────────────────────────────────────
|
||||
def _read_analytics_logs(max_files: int = 5) -> list:
|
||||
"""Parse PR-Agent analytics JSON logs (analytics=True records).
|
||||
|
||||
Real log lines look like:
|
||||
{"text": "...", "record": {"elapsed": {...}, "extra": {"command": "...", "pr_url": "..."},
|
||||
"file": {...}, "function": "...", "level": {"name": "INFO", ...},
|
||||
"message": "...", "module": "...", "process": {...}, "thread": {...},
|
||||
"time": {"repr": "2026-08-04 ...", "timestamp": ...}}}
|
||||
"""
|
||||
records = []
|
||||
files = sorted(glob.glob(os.path.join(ANALYTICS_DIR, "pr-agent.*.log")))
|
||||
for f in files[-max_files:]:
|
||||
try:
|
||||
with open(f) as fh:
|
||||
for line in fh:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
rec = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
# PR-Agent wraps under "record": {...}
|
||||
if "record" in rec and isinstance(rec["record"], dict):
|
||||
rec = rec["record"]
|
||||
extra = rec.get("extra", {}) or {}
|
||||
if "artifact" in extra and isinstance(extra["artifact"], dict):
|
||||
extra.update(extra.pop("artifact"))
|
||||
rec["_extra"] = extra
|
||||
rec["_file"] = Path(f).name
|
||||
records.append(rec)
|
||||
except FileNotFoundError:
|
||||
continue
|
||||
return records
|
||||
|
||||
|
||||
@app.get("/api/metrics")
|
||||
async def metrics():
|
||||
"""Prometheus-style metrics for the PR-Agent server."""
|
||||
records = _read_analytics_logs()
|
||||
total = len(records)
|
||||
failed = 0
|
||||
success = 0
|
||||
command_counts = {}
|
||||
model_failures = {}
|
||||
for rec in records:
|
||||
extra = rec.get("_extra", {})
|
||||
cmd = extra.get("command", "unknown")
|
||||
command_counts[cmd] = command_counts.get(cmd, 0) + 1
|
||||
msg = rec.get("message", "")
|
||||
if "Failed to generate" in msg or "error" in msg.lower() and rec.get("level", {}).get("name", "") == "WARNING":
|
||||
failed += 1
|
||||
model = extra.get("model", "unknown")
|
||||
model_failures[model] = model_failures.get(model, 0) + 1
|
||||
else:
|
||||
success += 1
|
||||
lines = [
|
||||
"# HELP pr_agent_requests_total Total PR-Agent analytics events",
|
||||
"# TYPE pr_agent_requests_total counter",
|
||||
f'pr_agent_requests_total{{status="success"}} {success}',
|
||||
f'pr_agent_requests_total{{status="failed"}} {failed}',
|
||||
"# HELP pr_agent_requests_by_command PR-Agent events by command",
|
||||
"# TYPE pr_agent_requests_by_command counter",
|
||||
]
|
||||
for cmd, cnt in sorted(command_counts.items()):
|
||||
lines.append(f'pr_agent_requests_by_command{{command="{cmd}"}} {cnt}')
|
||||
lines.append("# HELP pr_agent_model_failures PR-Agent model failures by model")
|
||||
lines.append("# TYPE pr_agent_model_failures counter")
|
||||
for model, cnt in sorted(model_failures.items()):
|
||||
lines.append(f'pr_agent_model_failures{{model="{model}"}} {cnt}')
|
||||
return PlainTextResponse(
|
||||
"\n".join(lines) + "\n",
|
||||
media_type="text/plain; version=0.0.4; charset=utf-8",
|
||||
)
|
||||
|
||||
|
||||
@app.get("/api/analytics")
|
||||
async def analytics():
|
||||
"""JSON analytics summary — recent events + failure breakdown."""
|
||||
records = _read_analytics_logs()
|
||||
recent = []
|
||||
for rec in records[-30:]:
|
||||
extra = rec.get("_extra", {})
|
||||
recent.append(
|
||||
{
|
||||
"time": rec.get("time", {}).get("repr", ""),
|
||||
"command": extra.get("command", ""),
|
||||
"message": rec.get("message", ""),
|
||||
"pr_url": extra.get("pr_url", ""),
|
||||
"model": extra.get("model", ""),
|
||||
"level": rec.get("level", {}).get("name", ""),
|
||||
}
|
||||
)
|
||||
failures = [r for r in records if "Failed to generate" in r.get("message", "")]
|
||||
return {
|
||||
"total_events": len(records),
|
||||
"failure_count": len(failures),
|
||||
"recent": recent,
|
||||
"failures": [
|
||||
{
|
||||
"time": r.get("time", {}).get("repr", ""),
|
||||
"command": r.get("_extra", {}).get("command", ""),
|
||||
"model": r.get("_extra", {}).get("model", ""),
|
||||
"message": r.get("message", "")[:200],
|
||||
}
|
||||
for r in failures[-20:]
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
# ── Discord notifications ───────────────────────────────────────────────────
|
||||
async def _send_discord(webhook: str, content: str, title: str = "", color: int = 0x5865F2):
|
||||
"""Fire-and-forget Discord webhook message. Never raises."""
|
||||
if not webhook:
|
||||
return False
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
resp = await client.post(
|
||||
webhook,
|
||||
json={
|
||||
"username": "PR-Agent Ops",
|
||||
"embeds": [{"title": title, "description": content[:4000], "color": color}],
|
||||
},
|
||||
)
|
||||
return resp.status_code in (200, 204)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
@app.post("/api/v1/notify_review")
|
||||
async def notify_review(request: Request):
|
||||
"""Internal endpoint: pr-agent/queue worker posts here after a review completes."""
|
||||
try:
|
||||
body = await request.json()
|
||||
except Exception:
|
||||
body = {}
|
||||
repo = body.get("repo", "")
|
||||
pr_num = body.get("pr", "")
|
||||
status = body.get("status", "done") # done | failed
|
||||
summary = body.get("summary", "")
|
||||
score = body.get("score", "")
|
||||
url = body.get("url", "")
|
||||
|
||||
if status == "failed":
|
||||
await _send_discord(
|
||||
DISCORD_ALERT_WEBHOOK_URL or DISCORD_WEBHOOK_URL,
|
||||
f"**{repo}** PR #{pr_num} review FAILED\n```{summary}```\n{url}",
|
||||
title="🚨 PR-Agent Review Failed",
|
||||
color=0xED4245,
|
||||
)
|
||||
else:
|
||||
await _send_discord(
|
||||
DISCORD_WEBHOOK_URL,
|
||||
f"**{repo}** PR #{pr_num} reviewed" + (f" — score {score}/10" if score else "") + f"\n{summary}\n{url}",
|
||||
title="✅ PR-Agent Review Complete",
|
||||
color=0x57F287,
|
||||
)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@app.get("/setup/callback")
|
||||
@app.post("/setup/callback")
|
||||
async def callback(request: Request):
|
||||
params = dict(request.query_params)
|
||||
if "code" in params:
|
||||
code = params["code"]
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.post(
|
||||
f"https://api.github.com/app-manifests/{code}/conversions",
|
||||
headers={"Accept": "application/vnd.github.v3+json"},
|
||||
)
|
||||
if resp.status_code == 201:
|
||||
data = resp.json()
|
||||
creds = {
|
||||
"app_id": data.get("id"),
|
||||
"pem": data.get("pem"),
|
||||
"webhook_secret": data.get("webhook_secret"),
|
||||
"slug": data.get("slug"),
|
||||
}
|
||||
with open(os.path.join(APP_DIR, "credentials_callback.json"), "w") as f:
|
||||
json.dump(creds, f, indent=2)
|
||||
return {
|
||||
"status": "success",
|
||||
"app_id": creds["app_id"],
|
||||
"slug": creds["slug"],
|
||||
}
|
||||
return {"status": "ok", "message": "callback received"}
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
return {"status": "ok", "model": os.environ.get("PR_AGENT_MODEL", "")}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
port = int(os.environ.get("PORT", "3000"))
|
||||
print(f"PR-Agent GitHub App server starting...")
|
||||
print(f" App ID: {os.environ.get('GITHUB_APP_ID', '')}")
|
||||
print(f" Model: {os.environ.get('PR_AGENT_MODEL', 'openai/claude-opus-5')} via omniroute")
|
||||
print(f" Endpoint: /api/v1/github_webhooks")
|
||||
print(f" Analytics: {ANALYTICS_DIR}")
|
||||
print(f" Port: {port}")
|
||||
uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")
|
||||
@@ -0,0 +1,15 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PR-Agent GitHub Webhook Server - Start Script"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
# Add the pr-agent package to path
|
||||
sys.path.insert(0, os.path.expanduser("~/hermes-agent/.venv/lib/python3.12/site-packages"))
|
||||
|
||||
from pr_agent.servers.github_app import app
|
||||
import uvicorn
|
||||
|
||||
if __name__ == '__main__':
|
||||
port = int(os.environ.get("PORT", "4002"))
|
||||
print(f"Starting PR-Agent GitHub App server on 0.0.0.0:{port}")
|
||||
uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")
|
||||
@@ -0,0 +1,90 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PR-Agent key auto-sync — fetch the working 9router key from Bitwarden Secrets
|
||||
Manager (BWS) and write it to the on-disk omniroute_key file IF it differs.
|
||||
|
||||
Why: the on-disk key file is the single source of truth for the running
|
||||
server (run_server.py reads it at startup). If BWS gets updated (key
|
||||
rotation) and the file isn't refreshed, the server silently starts failing
|
||||
with 401s — exactly what happened 2026-08-04 (stale 35-char key for 14h).
|
||||
|
||||
This script is invoked by systemd ExecStartPre= so every service start /
|
||||
restart re-syncs the key before uvicorn boots. It is idempotent and
|
||||
fail-open (on any BWS error it leaves the existing file untouched so the
|
||||
service can still start).
|
||||
"""
|
||||
import os, sys, subprocess, hashlib
|
||||
from pathlib import Path
|
||||
|
||||
APP_DIR = Path(os.environ.get("PR_AGENT_APP_DIR", "/var/lib/pr-agent-server"))
|
||||
KEYFILE = APP_DIR / "omniroute_key"
|
||||
# BWS secret that holds the working router key
|
||||
BWS_SECRET_ID = os.environ.get("BWS_ROUTER_KEY_SECRET_ID", "2aef2194-971d-4dae-99dd-b49a0041f97c")
|
||||
PROJECT_ID = "27210268-6134-47b3-9a68-b4980079d1ec"
|
||||
|
||||
|
||||
def sha(s: str) -> str:
|
||||
return hashlib.sha256(s.encode()).hexdigest()
|
||||
|
||||
|
||||
def bws_get_secret_value(secret_id: str) -> str:
|
||||
"""Fetch a BWS secret value. Returns '' on any failure (fail-open)."""
|
||||
token = os.environ.get("BWS_ACCESS_TOKEN", "")
|
||||
if not token and Path("/etc/bws-token").is_file():
|
||||
token = Path("/etc/bws-token").read_text().strip()
|
||||
if not token:
|
||||
print("sync-key: no BWS_ACCESS_TOKEN", file=sys.stderr)
|
||||
return ""
|
||||
env = {**os.environ, "BWS_ACCESS_TOKEN": token}
|
||||
try:
|
||||
r = subprocess.run(
|
||||
["/usr/local/bin/bws", "secret", "get", secret_id, "--output", "env"],
|
||||
capture_output=True, text=True, timeout=30, env=env,
|
||||
)
|
||||
if r.returncode != 0:
|
||||
print(f"sync-key: bws get failed rc={r.returncode}: {r.stderr[:200]}", file=sys.stderr)
|
||||
return ""
|
||||
# Value is shell-quoted KEY="value" — take first line, strip quotes
|
||||
line = r.stdout.split("\n")[0]
|
||||
if "=" not in line:
|
||||
return ""
|
||||
val = line.split("=", 1)[1].strip()
|
||||
# Remove wrapping quotes (shlex could be used; simple strip is fine for keys)
|
||||
if val.startswith('"') and val.endswith('"'):
|
||||
val = val[1:-1]
|
||||
elif val.startswith("'") and val.endswith("'"):
|
||||
val = val[1:-1]
|
||||
return val
|
||||
except Exception as e:
|
||||
print(f"sync-key: bws error {type(e).__name__}: {str(e)[:200]}", file=sys.stderr)
|
||||
return ""
|
||||
|
||||
|
||||
def main() -> int:
|
||||
new_key = bws_get_secret_value(BWS_SECRET_ID).strip()
|
||||
if not new_key:
|
||||
print("sync-key: no key from BWS, leaving existing file", file=sys.stderr)
|
||||
return 0 # fail-open
|
||||
|
||||
if KEYFILE.exists():
|
||||
old = KEYFILE.read_text().strip()
|
||||
if old == new_key:
|
||||
print("sync-key: key already up to date (no change)")
|
||||
return 0
|
||||
|
||||
# Write key with correct owner/perms (pr-agent user)
|
||||
try:
|
||||
import pwd
|
||||
pw = pwd.getpwnam("pr-agent")
|
||||
KEYFILE.write_text(new_key + "\n")
|
||||
os.chmod(KEYFILE, 0o600)
|
||||
os.chown(KEYFILE, pw.pw_uid, pw.pw_gid)
|
||||
print(f"sync-key: updated {KEYFILE} ({sha(new_key)[:12]}...)")
|
||||
return 0
|
||||
except Exception as e:
|
||||
print(f"sync-key: write failed {type(e).__name__}: {str(e)[:200]}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,136 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PR-Agent Trivial PR Auto-Merge — enhances auto_merge_bot.py with trivial-PR fast-path.
|
||||
|
||||
Logic:
|
||||
- PR with bot review comment ("PR Reviewer Guide") + score >= 5
|
||||
- AND is_trivial (docs-only, version bump, dependency bump, < small diff)
|
||||
→ skip AI fix, approve + merge directly if CI is green.
|
||||
|
||||
This file is imported/executed by auto_merge_bot.py; keep it standalone and
|
||||
dependency-light (httpx, pyjwt).
|
||||
"""
|
||||
import os, re, time, json
|
||||
from pathlib import Path
|
||||
|
||||
# ── Config ──
|
||||
APP_ID = os.environ.get("GITHUB_APP_ID", "4319749")
|
||||
PRIVATE_KEY_PATH = os.environ.get("PRIVATE_KEY_PATH", "/var/lib/pr-agent-server/private-key.pem")
|
||||
BASE_URL = os.environ.get("GITHUB_API_BASE", "https://api.github.com")
|
||||
BOT_LOGIN = os.environ.get("PR_AGENT_BOT_LOGIN", "mytheclipsebotreview")
|
||||
TRIVIAL_MAX_DIFF_LINES = int(os.environ.get("TRIVIAL_MAX_DIFF_LINES", "100"))
|
||||
TRIVIAL_MAX_FILES = int(os.environ.get("TRIVIAL_MAX_FILES", "5"))
|
||||
|
||||
TRIVIAL_TITLE_RE = re.compile(
|
||||
r"(dependabot|update|upgrade|bump|chore\(deps\)|pin dependencies|"
|
||||
r"docs?[:\(]|version|release|backport|typo|fix typo|minor|patch)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
TRIVIAL_FILE_RE = re.compile(
|
||||
r"(\.md$|\.txt$|\.lock$|\.gitignore$|\.dockerignore$|README|LICENSE|"
|
||||
r"CHANGELOG|package\.json$|pyproject\.toml$|Cargo\.toml$|go\.mod$|Gemfile\.lock$|"
|
||||
r"requirements.*\.txt$|\.github/workflows/|\.env\.example$)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def is_trivial_pr(title: str, author: str, changed_files: list, total_lines: int) -> bool:
|
||||
"""Determine if a PR is 'trivial' — safe to auto-merge without AI fix."""
|
||||
if author == BOT_LOGIN or "[bot]" in author:
|
||||
return True
|
||||
if total_lines > TRIVIAL_MAX_DIFF_LINES:
|
||||
return False
|
||||
if len(changed_files) > TRIVIAL_MAX_FILES:
|
||||
return False
|
||||
if TRIVIAL_TITLE_RE.search(title):
|
||||
return True
|
||||
# all changed files trivial?
|
||||
if changed_files and all(TRIVIAL_FILE_RE.search(f) for f in changed_files):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def get_pr_changed_files(token: str, repo_full: str, pr_number: int) -> tuple:
|
||||
"""Return (changed_files: list, total_added+deleted: int) via GitHub API."""
|
||||
import httpx
|
||||
files = []
|
||||
total = 0
|
||||
page = 1
|
||||
with httpx.Client(timeout=30) as client:
|
||||
while True:
|
||||
r = client.get(
|
||||
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}/files",
|
||||
params={"per_page": 100, "page": page},
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"},
|
||||
)
|
||||
if r.status_code != 200:
|
||||
break
|
||||
batch = r.json()
|
||||
if not batch:
|
||||
break
|
||||
for f in batch:
|
||||
files.append(f.get("filename", ""))
|
||||
total += f.get("additions", 0) + f.get("deletions", 0)
|
||||
if len(batch) < 100:
|
||||
break
|
||||
page += 1
|
||||
return files, total
|
||||
|
||||
|
||||
def check_ci_passed(token: str, repo_full: str, sha: str) -> tuple:
|
||||
"""Check GitHub check-runs/status for a SHA. Returns (ok, msg)."""
|
||||
import httpx
|
||||
with httpx.Client(timeout=30) as client:
|
||||
r = client.get(
|
||||
f"{BASE_URL}/repos/{repo_full}/commits/{sha}/check-runs",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"},
|
||||
)
|
||||
if r.status_code == 200:
|
||||
data = r.json()
|
||||
runs = data.get("check_runs", [])
|
||||
if not runs:
|
||||
return True, "No CI configured"
|
||||
for run in runs:
|
||||
status = run.get("status", "")
|
||||
conclusion = run.get("conclusion")
|
||||
if status != "completed":
|
||||
return False, f"Check pending: {run.get('name','?')}"
|
||||
if conclusion not in ("success", "neutral", "skipped"):
|
||||
return False, f"Check failed: {run.get('name','?')} → {conclusion}"
|
||||
return True, f"CI green ({len(runs)} checks)"
|
||||
# fallback to statuses
|
||||
r2 = client.get(
|
||||
f"{BASE_URL}/repos/{repo_full}/commits/{sha}/status",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"},
|
||||
)
|
||||
if r2.status_code == 200:
|
||||
st = r2.json().get("state", "")
|
||||
if st == "success":
|
||||
return True, "Status success"
|
||||
if st == "pending":
|
||||
return False, "Status pending"
|
||||
return False, f"Status {st}"
|
||||
return True, "No CI configured"
|
||||
|
||||
|
||||
def approve_pr(token: str, repo_full: str, pr_number: int) -> int:
|
||||
import httpx
|
||||
with httpx.Client(timeout=30) as client:
|
||||
r = client.post(
|
||||
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}/reviews",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"},
|
||||
json={"event": "APPROVE", "body": "✅ Auto-approved (trivial PR)."},
|
||||
)
|
||||
return r.status_code
|
||||
|
||||
|
||||
def merge_pr(token: str, repo_full: str, pr_number: int, sha: str) -> tuple:
|
||||
import httpx
|
||||
with httpx.Client(timeout=30) as client:
|
||||
r = client.put(
|
||||
f"{BASE_URL}/repos/{repo_full}/pulls/{pr_number}/merge",
|
||||
headers={"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"},
|
||||
json={"commit_title": f"Auto-merge trivial PR #{pr_number}", "merge_method": "squash", "sha": sha},
|
||||
)
|
||||
if r.status_code == 200:
|
||||
return True, f"Merged: {r.json().get('sha','?')}"
|
||||
return False, f"Merge failed: {r.status_code} - {r.json().get('message','')}"
|
||||
Reference in New Issue
Block a user