142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Swarm Governor — prevents PR pileup by enforcing merge discipline.
|
|
|
|
Runs as a pre-flight check before any swarm dispatch cycle.
|
|
If the open PR count exceeds the threshold, the swarm is paused
|
|
until PRs are reviewed, merged, or closed.
|
|
|
|
Usage:
|
|
python3 swarm_governor.py --check # Exit 0 if clear, 1 if blocked
|
|
python3 swarm_governor.py --report # Print status report
|
|
python3 swarm_governor.py --enforce # Close lowest-priority stale PRs
|
|
|
|
Environment:
|
|
GITEA_URL — Gitea instance URL (default: https://forge.alexanderwhitestone.com)
|
|
GITEA_TOKEN — API token
|
|
SWARM_MAX_OPEN — Max open PRs before blocking (default: 15)
|
|
SWARM_STALE_DAYS — Days before a PR is considered stale (default: 3)
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
GITEA_URL = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com")
|
|
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
|
|
MAX_OPEN = int(os.environ.get("SWARM_MAX_OPEN", "15"))
|
|
STALE_DAYS = int(os.environ.get("SWARM_STALE_DAYS", "3"))
|
|
|
|
# Repos to govern
|
|
REPOS = [
|
|
"Timmy_Foundation/the-nexus",
|
|
"Timmy_Foundation/timmy-config",
|
|
"Timmy_Foundation/timmy-home",
|
|
"Timmy_Foundation/fleet-ops",
|
|
"Timmy_Foundation/hermes-agent",
|
|
"Timmy_Foundation/the-beacon",
|
|
]
|
|
|
|
def api(path):
|
|
"""Call Gitea API."""
|
|
url = f"{GITEA_URL}/api/v1{path}"
|
|
req = urllib.request.Request(url)
|
|
if GITEA_TOKEN:
|
|
req.add_header("Authorization", f"token {GITEA_TOKEN}")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
return []
|
|
|
|
def get_open_prs():
|
|
"""Get all open PRs across governed repos."""
|
|
all_prs = []
|
|
for repo in REPOS:
|
|
prs = api(f"/repos/{repo}/pulls?state=open&limit=50")
|
|
for pr in prs:
|
|
pr["_repo"] = repo
|
|
age = (datetime.now(timezone.utc) -
|
|
datetime.fromisoformat(pr["created_at"].replace("Z", "+00:00")))
|
|
pr["_age_days"] = age.days
|
|
pr["_stale"] = age.days >= STALE_DAYS
|
|
all_prs.extend(prs)
|
|
return all_prs
|
|
|
|
def check():
|
|
"""Check if swarm should be allowed to dispatch."""
|
|
prs = get_open_prs()
|
|
total = len(prs)
|
|
stale = sum(1 for p in prs if p["_stale"])
|
|
|
|
if total > MAX_OPEN:
|
|
print(f"BLOCKED: {total} open PRs (max {MAX_OPEN}). {stale} stale.")
|
|
print(f"Review and merge before dispatching new work.")
|
|
return 1
|
|
else:
|
|
print(f"CLEAR: {total}/{MAX_OPEN} open PRs. {stale} stale.")
|
|
return 0
|
|
|
|
def report():
|
|
"""Print full status report."""
|
|
prs = get_open_prs()
|
|
by_repo = {}
|
|
for pr in prs:
|
|
by_repo.setdefault(pr["_repo"], []).append(pr)
|
|
|
|
print(f"{'='*60}")
|
|
print(f"SWARM GOVERNOR REPORT — {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
|
|
print(f"{'='*60}")
|
|
print(f"Total open PRs: {len(prs)} (max: {MAX_OPEN})")
|
|
print(f"Status: {'BLOCKED' if len(prs) > MAX_OPEN else 'CLEAR'}")
|
|
print()
|
|
|
|
for repo, repo_prs in sorted(by_repo.items()):
|
|
print(f" {repo}: {len(repo_prs)} open")
|
|
by_author = {}
|
|
for pr in repo_prs:
|
|
by_author.setdefault(pr["user"]["login"], []).append(pr)
|
|
for author, author_prs in sorted(by_author.items(), key=lambda x: -len(x[1])):
|
|
stale_count = sum(1 for p in author_prs if p["_stale"])
|
|
stale_str = f" ({stale_count} stale)" if stale_count else ""
|
|
print(f" {author}: {len(author_prs)}{stale_str}")
|
|
|
|
# Highlight stale PRs
|
|
stale_prs = [p for p in prs if p["_stale"]]
|
|
if stale_prs:
|
|
print(f"\nStale PRs (>{STALE_DAYS} days):")
|
|
for pr in sorted(stale_prs, key=lambda p: p["_age_days"], reverse=True):
|
|
print(f" #{pr['number']} ({pr['_age_days']}d) [{pr['_repo'].split('/')[1]}] {pr['title'][:60]}")
|
|
|
|
def enforce():
|
|
"""Close stale PRs that are blocking the queue."""
|
|
prs = get_open_prs()
|
|
if len(prs) <= MAX_OPEN:
|
|
print("Queue is clear. Nothing to enforce.")
|
|
return 0
|
|
|
|
# Sort by staleness, close oldest first
|
|
stale = sorted([p for p in prs if p["_stale"]], key=lambda p: p["_age_days"], reverse=True)
|
|
to_close = len(prs) - MAX_OPEN
|
|
|
|
print(f"Need to close {to_close} PRs to get under {MAX_OPEN}.")
|
|
for pr in stale[:to_close]:
|
|
print(f" Would close: #{pr['number']} ({pr['_age_days']}d) [{pr['_repo'].split('/')[1]}] {pr['title'][:50]}")
|
|
|
|
print(f"\nDry run — add --force to actually close.")
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "--check"
|
|
if cmd == "--check":
|
|
sys.exit(check())
|
|
elif cmd == "--report":
|
|
report()
|
|
elif cmd == "--enforce":
|
|
enforce()
|
|
else:
|
|
print(f"Usage: {sys.argv[0]} [--check|--report|--enforce]")
|
|
sys.exit(1)
|