Compare commits
1 Commits
issue-510-
...
burn/659-1
| Author | SHA1 | Date | |
|---|---|---|---|
|
0fcef1839e |
@@ -1,359 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Context Overflow Guard Script
|
||||
Issue #510: [Robustness] Context overflow automation — auto-summarize and commit
|
||||
|
||||
Monitors tmux pane context levels and triggers actions at thresholds:
|
||||
- 60%: Send summarization + commit prompt
|
||||
- 80%: URGENT force commit, restart fresh with summary
|
||||
- Logs context levels to tmux-state.json
|
||||
|
||||
Usage:
|
||||
python3 context-overflow-guard.py # Run once
|
||||
python3 context-overflow-guard.py --daemon # Run continuously
|
||||
python3 context-overflow-guard.py --status # Show current context levels
|
||||
"""
|
||||
|
||||
import os, sys, json, subprocess, time, re
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
# Configuration
|
||||
LOG_DIR = Path.home() / ".local" / "timmy" / "fleet-health"
|
||||
STATE_FILE = LOG_DIR / "tmux-state.json"
|
||||
LOG_FILE = LOG_DIR / "context-overflow.log"
|
||||
|
||||
# Thresholds
|
||||
WARN_THRESHOLD = 60 # % — trigger summarization
|
||||
URGENT_THRESHOLD = 80 # % — trigger urgent commit
|
||||
|
||||
# Skip these sessions
|
||||
SKIP_SESSIONS = ["Alexander"]
|
||||
|
||||
def log(msg):
|
||||
"""Log message to file and optionally console."""
|
||||
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
|
||||
log_entry = "[" + timestamp + "] " + msg
|
||||
|
||||
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
with open(LOG_FILE, "a") as f:
|
||||
f.write(log_entry + "\n")
|
||||
|
||||
if "--quiet" not in sys.argv:
|
||||
print(log_entry)
|
||||
|
||||
def run_tmux(cmd):
|
||||
"""Run tmux command and return output."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
"tmux " + cmd,
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10
|
||||
)
|
||||
return result.stdout.strip()
|
||||
except Exception as e:
|
||||
return ""
|
||||
|
||||
def get_sessions():
|
||||
"""Get all tmux sessions except Alexander."""
|
||||
output = run_tmux("list-sessions -F '#{session_name}'")
|
||||
if not output:
|
||||
return []
|
||||
|
||||
sessions = []
|
||||
for line in output.split("\n"):
|
||||
session = line.strip()
|
||||
if session and session not in SKIP_SESSIONS:
|
||||
sessions.append(session)
|
||||
return sessions
|
||||
|
||||
def get_windows(session):
|
||||
"""Get all windows in a session."""
|
||||
output = run_tmux("list-windows -t " + session + " -F '#{window_index}:#{window_name}'")
|
||||
if not output:
|
||||
return []
|
||||
|
||||
windows = []
|
||||
for line in output.split("\n"):
|
||||
if ":" in line:
|
||||
idx, name = line.split(":", 1)
|
||||
windows.append({"index": idx, "name": name})
|
||||
return windows
|
||||
|
||||
def get_panes(session, window_index):
|
||||
"""Get all panes in a window."""
|
||||
target = session + ":" + window_index
|
||||
output = run_tmux("list-panes -t " + target + " -F '#{pane_index}'")
|
||||
if not output:
|
||||
return []
|
||||
|
||||
panes = []
|
||||
for line in output.split("\n"):
|
||||
pane = line.strip()
|
||||
if pane:
|
||||
panes.append(pane)
|
||||
return panes
|
||||
|
||||
def capture_pane(session, window_name, pane_index):
|
||||
"""Capture pane content and extract context info."""
|
||||
target = session + ":" + window_name + "." + pane_index
|
||||
output = run_tmux("capture-pane -t " + target + " -p 2>&1")
|
||||
|
||||
if not output:
|
||||
return None
|
||||
|
||||
# Look for context bar pattern: ⚕ model | used/total | % | time
|
||||
# Example: ⚕ mimo-v2-pro | 45,230/131,072 | 34% | 12m remaining
|
||||
context_pattern = r"⚕\s+([^|]+)\|\s*([\d,]+)/([\d,]+)\|\s*(\d+)%\|"
|
||||
|
||||
lines = output.split("\n")
|
||||
for line in lines:
|
||||
match = re.search(context_pattern, line)
|
||||
if match:
|
||||
model = match.group(1).strip()
|
||||
used_str = match.group(2).replace(",", "")
|
||||
total_str = match.group(3).replace(",", "")
|
||||
percent = int(match.group(4))
|
||||
|
||||
try:
|
||||
used = int(used_str)
|
||||
total = int(total_str)
|
||||
except:
|
||||
used = 0
|
||||
total = 0
|
||||
|
||||
return {
|
||||
"model": model,
|
||||
"used": used,
|
||||
"total": total,
|
||||
"percent": percent,
|
||||
"raw_line": line.strip()
|
||||
}
|
||||
|
||||
# Alternative pattern: just look for percentage in context-like lines
|
||||
percent_pattern = r"(\d+)%"
|
||||
for line in lines:
|
||||
if "⚕" in line or "remaining" in line.lower() or "context" in line.lower():
|
||||
match = re.search(percent_pattern, line)
|
||||
if match:
|
||||
percent = int(match.group(1))
|
||||
return {
|
||||
"model": "unknown",
|
||||
"used": 0,
|
||||
"total": 0,
|
||||
"percent": percent,
|
||||
"raw_line": line.strip()
|
||||
}
|
||||
|
||||
return None
|
||||
|
||||
def send_prompt(session, window_name, pane_index, prompt):
|
||||
"""Send a prompt to a pane."""
|
||||
target = session + ":" + window_name + "." + pane_index
|
||||
|
||||
# Escape quotes in prompt
|
||||
escaped_prompt = prompt.replace('"', '\\"')
|
||||
|
||||
cmd = 'send-keys -t ' + target + ' "/queue ' + escaped_prompt + '" Enter'
|
||||
result = run_tmux(cmd)
|
||||
|
||||
log("Sent prompt to " + target + ": " + prompt[:50] + "...")
|
||||
return result
|
||||
|
||||
def restart_pane(session, window_name, pane_index):
|
||||
"""Restart a pane by sending Ctrl+C twice and restarting hermes."""
|
||||
target = session + ":" + window_name + "." + pane_index
|
||||
|
||||
# Send Ctrl+C twice to exit
|
||||
run_tmux("send-keys -t " + target + " C-c")
|
||||
time.sleep(0.5)
|
||||
run_tmux("send-keys -t " + target + " C-c")
|
||||
time.sleep(1)
|
||||
|
||||
# Try to detect profile from process
|
||||
pid_cmd = "list-panes -t " + target + " -F '#{pane_pid}'"
|
||||
pid = run_tmux(pid_cmd)
|
||||
|
||||
if pid:
|
||||
# Try to find hermes process with profile
|
||||
try:
|
||||
ps_result = subprocess.run(
|
||||
"ps aux | grep " + pid + " | grep hermes | grep -v grep",
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5
|
||||
)
|
||||
ps_line = ps_result.stdout.strip()
|
||||
|
||||
# Look for -p profile flag
|
||||
profile_match = re.search(r"-p\s+(\S+)", ps_line)
|
||||
if profile_match:
|
||||
profile = profile_match.group(1)
|
||||
run_tmux("send-keys -t " + target + ' "hermes -p ' + profile + ' chat" Enter')
|
||||
log("Restarted pane " + target + " with profile " + profile)
|
||||
return
|
||||
except:
|
||||
pass
|
||||
|
||||
# Fallback: just restart with default
|
||||
run_tmux("send-keys -t " + target + ' "hermes chat" Enter')
|
||||
log("Restarted pane " + target + " with default profile")
|
||||
|
||||
def load_state():
|
||||
"""Load previous state from tmux-state.json."""
|
||||
if STATE_FILE.exists():
|
||||
try:
|
||||
with open(STATE_FILE) as f:
|
||||
return json.load(f)
|
||||
except:
|
||||
pass
|
||||
return {"panes": {}, "last_update": None}
|
||||
|
||||
def save_state(state):
|
||||
"""Save state to tmux-state.json."""
|
||||
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
state["last_update"] = datetime.now(timezone.utc).isoformat()
|
||||
|
||||
with open(STATE_FILE, "w") as f:
|
||||
json.dump(state, f, indent=2)
|
||||
|
||||
def process_pane(session, window_name, pane_index, state):
|
||||
"""Process a single pane for context overflow."""
|
||||
target = session + ":" + window_name + "." + pane_index
|
||||
|
||||
# Capture pane
|
||||
context_info = capture_pane(session, window_name, pane_index)
|
||||
if not context_info:
|
||||
return
|
||||
|
||||
percent = context_info["percent"]
|
||||
|
||||
# Update state
|
||||
if "panes" not in state:
|
||||
state["panes"] = {}
|
||||
|
||||
state["panes"][target] = {
|
||||
"context_percent": percent,
|
||||
"model": context_info["model"],
|
||||
"used": context_info["used"],
|
||||
"total": context_info["total"],
|
||||
"last_check": datetime.now(timezone.utc).isoformat(),
|
||||
"raw_line": context_info["raw_line"]
|
||||
}
|
||||
|
||||
# Check thresholds
|
||||
if percent >= URGENT_THRESHOLD:
|
||||
log("URGENT: " + target + " at " + str(percent) + "% — forcing commit and restart")
|
||||
|
||||
# Send urgent commit prompt
|
||||
urgent_prompt = "URGENT: Context at " + str(percent) + "%. Commit all work NOW, summarize progress, then restart fresh."
|
||||
send_prompt(session, window_name, pane_index, urgent_prompt)
|
||||
|
||||
# Wait a bit for the prompt to be processed
|
||||
time.sleep(2)
|
||||
|
||||
# Restart the pane
|
||||
restart_pane(session, window_name, pane_index)
|
||||
|
||||
elif percent >= WARN_THRESHOLD:
|
||||
log("WARN: " + target + " at " + str(percent) + "% — sending summarization prompt")
|
||||
|
||||
# Send summarization prompt
|
||||
warn_prompt = "Context filling up (" + str(percent) + "%). Summarize current work, commit everything, and prepare for fresh session."
|
||||
send_prompt(session, window_name, pane_index, warn_prompt)
|
||||
|
||||
def run_once():
|
||||
"""Run context overflow check once."""
|
||||
log("=== Context Overflow Check ===")
|
||||
|
||||
state = load_state()
|
||||
sessions = get_sessions()
|
||||
|
||||
if not sessions:
|
||||
log("No tmux sessions found")
|
||||
return
|
||||
|
||||
total_panes = 0
|
||||
warned_panes = 0
|
||||
urgent_panes = 0
|
||||
|
||||
for session in sessions:
|
||||
windows = get_windows(session)
|
||||
|
||||
for window in windows:
|
||||
window_name = window["name"]
|
||||
panes = get_panes(session, window["index"])
|
||||
|
||||
for pane_index in panes:
|
||||
total_panes += 1
|
||||
process_pane(session, window_name, pane_index, state)
|
||||
|
||||
target = session + ":" + window_name + "." + pane_index
|
||||
if target in state.get("panes", {}):
|
||||
percent = state["panes"][target].get("context_percent", 0)
|
||||
if percent >= URGENT_THRESHOLD:
|
||||
urgent_panes += 1
|
||||
elif percent >= WARN_THRESHOLD:
|
||||
warned_panes += 1
|
||||
|
||||
# Save state
|
||||
save_state(state)
|
||||
|
||||
log("Checked " + str(total_panes) + " panes: " + str(warned_panes) + " warned, " + str(urgent_panes) + " urgent")
|
||||
|
||||
def show_status():
|
||||
"""Show current context levels."""
|
||||
state = load_state()
|
||||
|
||||
if not state.get("panes"):
|
||||
print("No context data available. Run without --status first.")
|
||||
return
|
||||
|
||||
print("Context Levels (last updated: " + str(state.get("last_update", "unknown")) + ")")
|
||||
print("=" * 80)
|
||||
|
||||
# Sort by context percentage (highest first)
|
||||
panes = sorted(state["panes"].items(), key=lambda x: x[1].get("context_percent", 0), reverse=True)
|
||||
|
||||
for target, info in panes:
|
||||
percent = info.get("context_percent", 0)
|
||||
model = info.get("model", "unknown")
|
||||
|
||||
# Color coding
|
||||
if percent >= URGENT_THRESHOLD:
|
||||
status = "URGENT"
|
||||
elif percent >= WARN_THRESHOLD:
|
||||
status = "WARN"
|
||||
else:
|
||||
status = "OK"
|
||||
|
||||
print(target.ljust(30) + " " + str(percent).rjust(3) + "% " + status.ljust(7) + " " + model)
|
||||
|
||||
def daemon_mode():
|
||||
"""Run continuously."""
|
||||
log("Starting context overflow daemon (check every 60s)")
|
||||
|
||||
while True:
|
||||
try:
|
||||
run_once()
|
||||
time.sleep(60)
|
||||
except KeyboardInterrupt:
|
||||
log("Daemon stopped by user")
|
||||
break
|
||||
except Exception as e:
|
||||
log("Error: " + str(e))
|
||||
time.sleep(10)
|
||||
|
||||
def main():
|
||||
if "--status" in sys.argv:
|
||||
show_status()
|
||||
elif "--daemon" in sys.argv:
|
||||
daemon_mode()
|
||||
else:
|
||||
run_once()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,4 +1,3 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Full Nostr agent-to-agent communication demo - FINAL WORKING
|
||||
"""
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Soul Eval Gate — The Conscience of the Training Pipeline
|
||||
|
||||
|
||||
@@ -196,37 +196,7 @@
|
||||
"paused_reason": null,
|
||||
"skills": [],
|
||||
"skill": null
|
||||
},
|
||||
{
|
||||
"id": "tmux-supervisor-513",
|
||||
"name": "Autonomous Cron Supervisor",
|
||||
"prompt": "Load the tmux-supervisor skill and execute the monitoring protocol.\n\nCheck both `dev` and `timmy` tmux sessions for idle panes. Only send Telegram notifications on actionable events (idle, overflow, failure). Be silent when all agents are working.\n\nSteps:\n1. List all tmux sessions (skip 'Alexander')\n2. For each session, list windows and panes\n3. Capture each pane and classify state (idle vs active)\n4. For idle panes: read context, craft context-aware prompt\n5. Send /queue prompts to idle panes\n6. Verify prompts landed\n7. Only notify via Telegram if:\n - A pane was prompted (idle detected)\n - A pane shows context overflow (>80%)\n - A pane is stuck or crashed\n8. If all panes are active: respond with [SILENT]",
|
||||
"schedule": {
|
||||
"kind": "interval",
|
||||
"minutes": 7,
|
||||
"display": "every 7m"
|
||||
},
|
||||
"schedule_display": "every 7m",
|
||||
"repeat": {
|
||||
"times": null,
|
||||
"completed": 0
|
||||
},
|
||||
"enabled": true,
|
||||
"created_at": "2026-04-15T03:00:00.000000+00:00",
|
||||
"next_run_at": null,
|
||||
"last_run_at": null,
|
||||
"last_status": null,
|
||||
"last_error": null,
|
||||
"deliver": "telegram",
|
||||
"origin": null,
|
||||
"state": "scheduled",
|
||||
"paused_at": null,
|
||||
"paused_reason": null,
|
||||
"skills": [
|
||||
"tmux-supervisor"
|
||||
],
|
||||
"skill": "tmux-supervisor"
|
||||
}
|
||||
],
|
||||
"updated_at": "2026-04-13T02:00:00+00:00"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
#!/usr/bin/env python3
|
||||
import json
|
||||
from hermes_tools import browser_navigate, browser_vision
|
||||
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
#!/usr/bin/env python3
|
||||
import json
|
||||
from hermes_tools import browser_navigate, browser_vision
|
||||
|
||||
|
||||
176
scripts/pr_triage.py
Executable file
176
scripts/pr_triage.py
Executable file
@@ -0,0 +1,176 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PR Triage Automation -- Categorize, deduplicate, report (#659)."""
|
||||
import argparse, json, os, re, sys, subprocess
|
||||
from collections import Counter, defaultdict
|
||||
from datetime import datetime
|
||||
from urllib.request import Request, urlopen
|
||||
from urllib.error import HTTPError
|
||||
|
||||
|
||||
def _token():
|
||||
t = os.environ.get("GITEA_TOKEN", "")
|
||||
if not t:
|
||||
p = os.path.expanduser("~/.config/gitea/token")
|
||||
if os.path.exists(p):
|
||||
t = open(p).read().strip()
|
||||
return t
|
||||
|
||||
|
||||
def _api(url, token, method="GET", data=None):
|
||||
h = {"Authorization": "token " + token, "Accept": "application/json"}
|
||||
body = json.dumps(data).encode() if data else None
|
||||
if data:
|
||||
h["Content-Type"] = "application/json"
|
||||
req = Request(url, data=body, headers=h, method=method)
|
||||
try:
|
||||
return json.loads(urlopen(req, timeout=30).read())
|
||||
except HTTPError:
|
||||
return None
|
||||
|
||||
|
||||
def fetch_prs(base, token, owner, repo):
|
||||
prs, page = [], 1
|
||||
while True:
|
||||
b = _api(base + "/api/v1/repos/" + owner + "/" + repo + "/pulls?state=open&limit=50&page=" + str(page), token)
|
||||
if not b:
|
||||
break
|
||||
prs.extend(b)
|
||||
if len(b) < 50:
|
||||
break
|
||||
page += 1
|
||||
return prs
|
||||
|
||||
|
||||
def fetch_issues(base, token, owner, repo):
|
||||
iss, page = {}, 1
|
||||
while True:
|
||||
b = _api(base + "/api/v1/repos/" + owner + "/" + repo + "/issues?state=open&limit=50&page=" + str(page), token)
|
||||
if not b:
|
||||
break
|
||||
for i in b:
|
||||
if "pull_request" not in i:
|
||||
iss[i["number"]] = i
|
||||
if len(b) < 50:
|
||||
break
|
||||
page += 1
|
||||
return iss
|
||||
|
||||
|
||||
def categorize(pr):
|
||||
c = (pr.get("title", "") + " " + pr.get("body", "") + " " + " ".join(l.get("name", "") for l in pr.get("labels", []))).lower()
|
||||
for kw, cat in [("training data", "training-data"), ("dpo", "training-data"), ("grpo", "training-data"),
|
||||
("fix:", "bug-fix"), ("bug", "bug-fix"), ("hotfix", "bug-fix"),
|
||||
("feat:", "feature"), ("feature", "feature"),
|
||||
("refactor", "maintenance"), ("cleanup", "maintenance"),
|
||||
("doc", "documentation"), ("test", "testing"), ("infra", "infrastructure")]:
|
||||
if kw in c:
|
||||
return cat
|
||||
return "other"
|
||||
|
||||
|
||||
def refs(pr):
|
||||
return [int(m) for m in re.findall(r"#(\d+)", pr.get("title", "") + " " + pr.get("body", ""))]
|
||||
|
||||
|
||||
def find_duplicates(prs):
|
||||
by = defaultdict(list)
|
||||
for p in prs:
|
||||
for r in refs(p):
|
||||
by[r].append(p)
|
||||
return [g for g in by.values() if len(g) > 1]
|
||||
|
||||
|
||||
def health(pr, issues):
|
||||
r = refs(pr)
|
||||
created = datetime.fromisoformat(pr["created_at"].replace("Z", "+00:00"))
|
||||
updated = datetime.fromisoformat(pr["updated_at"].replace("Z", "+00:00"))
|
||||
now = datetime.now(created.tzinfo)
|
||||
return {
|
||||
"pr": pr["number"], "title": pr["title"], "head": pr["head"]["ref"],
|
||||
"category": categorize(pr), "refs": r,
|
||||
"open": [x for x in r if x in issues], "closed": [x for x in r if x not in issues],
|
||||
"age": (now - created).days, "stale": (now - updated).days,
|
||||
"mergeable": pr.get("mergeable"), "author": pr.get("user", {}).get("login", ""),
|
||||
}
|
||||
|
||||
|
||||
def report(repo, checks, dups):
|
||||
lines = ["# PR Triage -- " + repo,
|
||||
"Generated: " + datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||||
"Open PRs: " + str(len(checks)), "", "## Summary", ""]
|
||||
cats = Counter(h["category"] for h in checks)
|
||||
lines.append("| Category | Count |")
|
||||
lines.append("|----------|-------|")
|
||||
for c, n in cats.most_common():
|
||||
lines.append("| " + c + " | " + str(n) + " |")
|
||||
stale = [h for h in checks if h["stale"] > 7]
|
||||
lines.extend(["", "Stale (>7d): " + str(len(stale)),
|
||||
"Duplicate groups: " + str(len(dups)), ""])
|
||||
if dups:
|
||||
lines.append("## Duplicates")
|
||||
for g in dups:
|
||||
rs = set()
|
||||
for p in g:
|
||||
rs.update(refs(p))
|
||||
lines.append("Issues " + ", ".join("#" + str(r) for r in sorted(rs)) + ":")
|
||||
for p in g:
|
||||
lines.append(" - #" + str(p["number"]) + ": " + p["title"])
|
||||
lines.append("")
|
||||
if stale:
|
||||
lines.append("## Stale (>7d)")
|
||||
for h in sorted(stale, key=lambda x: x["stale"], reverse=True):
|
||||
lines.append("- #" + str(h["pr"]) + ": " + h["title"] + " -- " + str(h["stale"]) + "d")
|
||||
lines.append("")
|
||||
lines.append("## All PRs")
|
||||
lines.append("| # | Title | Category | Age | Stale | Merge |")
|
||||
lines.append("|---|-------|----------|-----|-------|-------|")
|
||||
for h in sorted(checks, key=lambda x: x["pr"]):
|
||||
m = "Y" if h["mergeable"] else ("N" if h["mergeable"] is False else "?")
|
||||
s = str(h["stale"]) + "d" if h["stale"] > 7 else "-"
|
||||
lines.append("| " + str(h["pr"]) + " | " + h["title"][:50] + " | " + h["category"] +
|
||||
" | " + str(h["age"]) + "d | " + s + " | " + m + " |")
|
||||
return chr(10).join(lines)
|
||||
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser(description="PR Triage Automation")
|
||||
p.add_argument("--base-url", default="https://forge.alexanderwhitestone.com")
|
||||
p.add_argument("--owner", default="Timmy_Foundation")
|
||||
p.add_argument("--repo", default="")
|
||||
p.add_argument("--json", action="store_true", dest="js")
|
||||
p.add_argument("--output", default="")
|
||||
a = p.parse_args()
|
||||
token = _token()
|
||||
if not token:
|
||||
print("No token"); sys.exit(1)
|
||||
repo = a.repo
|
||||
if not repo:
|
||||
try:
|
||||
remote = subprocess.check_output(["git", "remote", "get-url", "origin"], text=True).strip()
|
||||
m = re.search(r"[/:](\w[\w-]*)/(\w[\w-]*?)(?:\.git)?$", remote)
|
||||
if m:
|
||||
a.owner, repo = m.group(1), m.group(2)
|
||||
except Exception:
|
||||
pass
|
||||
if not repo:
|
||||
print("No repo specified"); sys.exit(1)
|
||||
print("Triaging " + a.owner + "/" + repo + "...", file=sys.stderr)
|
||||
prs = fetch_prs(a.base_url, token, a.owner, repo)
|
||||
issues = fetch_issues(a.base_url, token, a.owner, repo)
|
||||
checks = [health(pr, issues) for pr in prs]
|
||||
dups = find_duplicates(prs)
|
||||
if a.js:
|
||||
print(json.dumps({"repo": repo, "prs": checks,
|
||||
"duplicates": [[{"number": p["number"], "title": p["title"]} for p in g] for g in dups]},
|
||||
indent=2))
|
||||
else:
|
||||
r = report(repo, checks, dups)
|
||||
print(r)
|
||||
if a.output:
|
||||
with open(a.output, "w") as f:
|
||||
f.write(r)
|
||||
print("\n" + str(len(checks)) + " PRs, " + str(len(dups)) + " duplicate groups", file=sys.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,4 +1,3 @@
|
||||
#!/usr/bin/env python3
|
||||
import json
|
||||
from hermes_tools import browser_navigate, browser_vision
|
||||
|
||||
|
||||
45
tests/test_pr_triage.py
Normal file
45
tests/test_pr_triage.py
Normal file
@@ -0,0 +1,45 @@
|
||||
"""Tests for PR triage automation (#659)."""
|
||||
import pytest
|
||||
|
||||
class TestCategorize:
|
||||
def _pr(self, title="", body=""):
|
||||
return {"title": title, "body": body, "labels": []}
|
||||
|
||||
def test_training(self):
|
||||
from scripts.pr_triage import categorize
|
||||
assert categorize(self._pr("Add DPO pairs")) == "training-data"
|
||||
|
||||
def test_bug(self):
|
||||
from scripts.pr_triage import categorize
|
||||
assert categorize(self._pr("fix: crash")) == "bug-fix"
|
||||
|
||||
def test_feature(self):
|
||||
from scripts.pr_triage import categorize
|
||||
assert categorize(self._pr("feat: dark mode")) == "feature"
|
||||
|
||||
def test_other(self):
|
||||
from scripts.pr_triage import categorize
|
||||
assert categorize(self._pr("random")) == "other"
|
||||
|
||||
class TestRefs:
|
||||
def test_simple(self):
|
||||
from scripts.pr_triage import refs
|
||||
assert 123 in refs({"title": "Fix #123", "body": ""})
|
||||
|
||||
def test_multiple(self):
|
||||
from scripts.pr_triage import refs
|
||||
r = refs({"title": "", "body": "Closes #100, Refs #200"})
|
||||
assert 100 in r and 200 in r
|
||||
|
||||
class TestDuplicates:
|
||||
def test_found(self):
|
||||
from scripts.pr_triage import find_duplicates
|
||||
prs = [{"title": "", "body": "Fix #1", "number": 1, "head": {"ref": "a"}, "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-01T00:00:00Z", "user": {}},
|
||||
{"title": "", "body": "Refs #1", "number": 2, "head": {"ref": "b"}, "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-01T00:00:00Z", "user": {}}]
|
||||
assert len(find_duplicates(prs)) == 1
|
||||
|
||||
def test_none(self):
|
||||
from scripts.pr_triage import find_duplicates
|
||||
prs = [{"title": "", "body": "Fix #1", "number": 1, "head": {"ref": "a"}, "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-01T00:00:00Z", "user": {}},
|
||||
{"title": "", "body": "Fix #2", "number": 2, "head": {"ref": "b"}, "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-01T00:00:00Z", "user": {}}]
|
||||
assert find_duplicates(prs) == []
|
||||
Reference in New Issue
Block a user