197 lines
7.3 KiB
Python
197 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""research_backfill_sweep.py — one-time research-pending backfill (#204).
|
|
|
|
The research gate (pipeline-gate.yml) applies `research-pending` only on
|
|
ready-label EVENTS. Standing backlog issues predate the gate → never labeled
|
|
→ never researched. This sweep applies the gate's OWN readiness semantics to
|
|
the whole backlog:
|
|
|
|
eligible = open issue, not blocked, not already research-pending,
|
|
no existing 4-phase research round (INWARD/OUTWARD/INNOVATION/
|
|
SYNTHESIS comment — research already done stays done),
|
|
all `Depends on: #N` closed (verified against the server;
|
|
a missing dep is UNSATISFIED, never assumed closed).
|
|
|
|
Nick's critique, structuralized: "are these issues with research already
|
|
done?" — the sweep detects existing research and skips it. Research is never
|
|
re-dispatched for completed work.
|
|
|
|
Dry-run by default. `--apply` is the explicit mutation. Idempotent: safe to
|
|
re-run; labeled/satisfied issues are skipped on the next pass.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
PHASES = ("INWARD", "OUTWARD", "INNOVATION", "SYNTHESIS")
|
|
DEP_RE = re.compile(r"Depends on:\s*#(\d+)", re.IGNORECASE)
|
|
|
|
|
|
class SweepError(Exception):
|
|
"""A sweep precondition or server call failed — fail loud, never guess."""
|
|
|
|
|
|
def _api(
|
|
base: str, token: str, path: str, method: str = "GET", data: Any = None, ok: tuple = (200,)
|
|
) -> tuple[int, Any]:
|
|
req = urllib.request.Request(
|
|
base + path,
|
|
data=json.dumps(data).encode() if data is not None else None,
|
|
headers={"Authorization": "token " + token, "Content-Type": "application/json"},
|
|
method=method,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
body, status = r.read(), r.status
|
|
except urllib.error.HTTPError as e:
|
|
status, body = e.code, e.read()
|
|
except urllib.error.URLError as e:
|
|
raise SweepError(f"forge unreachable at {path}: {e.reason}") from e
|
|
try:
|
|
parsed: Any = json.loads(body) if body else {}
|
|
except json.JSONDecodeError:
|
|
if status not in ok:
|
|
raise SweepError(f"{method} {path} -> {status} (non-JSON): {body[:200]}") from None
|
|
parsed = body
|
|
if method == "GET" and status not in ok:
|
|
raise SweepError(f"GET {path} -> {status}: {str(parsed)[:300]}")
|
|
return status, parsed
|
|
|
|
|
|
class ForgeAdapter:
|
|
"""Thin .get()/.post() facade over _api so logic is testable."""
|
|
|
|
def __init__(self, base: str, token: str):
|
|
self.base, self.token = base, token
|
|
|
|
def get(self, path: str) -> tuple[int, Any]:
|
|
return _api(self.base, self.token, path)
|
|
|
|
def post(self, path: str, data: Any) -> tuple[int, Any]:
|
|
return _api(self.base, self.token, path, "POST", data)
|
|
|
|
|
|
def has_research_round(comments: list[dict]) -> bool:
|
|
"""True if any comment carries all four phase markers (gate's own test)."""
|
|
for c in comments:
|
|
body = c.get("body", "")
|
|
if all(re.search(rf"\b{p}\b", body, re.IGNORECASE) for p in PHASES):
|
|
return True
|
|
return False
|
|
|
|
|
|
def open_pr_issue_refs(api) -> set[int]:
|
|
"""Issue numbers referenced by OPEN PRs (closes/fixes/resolves #N in the
|
|
body) — those issues are in-flight; research is premature."""
|
|
refs: set[int] = set()
|
|
page = 1
|
|
ref_re = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.IGNORECASE)
|
|
while True:
|
|
status, batch = api.get(f"/pulls?state=open&limit=50&page={page}")
|
|
if not isinstance(batch, list):
|
|
break
|
|
for pr in batch:
|
|
for m in ref_re.finditer(pr.get("body") or ""):
|
|
refs.add(int(m.group(1)))
|
|
if len(batch) < 50:
|
|
return refs
|
|
page += 1
|
|
|
|
|
|
def compute_targets(issues: list[dict], api) -> list[dict]:
|
|
"""Eligible issues for research-pending labeling.
|
|
|
|
`api` is a ForgeAdapter-like object exposing .get(path) -> (status, body).
|
|
Gate semantics: open, not blocked, not already labeled, no existing
|
|
research round, all deps server-verified closed (missing = unsatisfied),
|
|
and not referenced by an open PR (in-flight work doesn't need research).
|
|
"""
|
|
in_flight = open_pr_issue_refs(api)
|
|
targets = []
|
|
for i in issues:
|
|
if i.get("pull_request") or i.get("state") != "open":
|
|
continue
|
|
labels = {l["name"] for l in i.get("labels", [])}
|
|
if "blocked" in labels or "research-pending" in labels:
|
|
continue
|
|
if i["number"] in in_flight:
|
|
continue
|
|
deps = [int(m) for m in DEP_RE.findall(i.get("body") or "")]
|
|
unsatisfied = False
|
|
for d in deps:
|
|
status, dep = api.get(f"/issues/{d}")
|
|
if status != 200 or dep.get("state") != "closed":
|
|
unsatisfied = True
|
|
break
|
|
if unsatisfied:
|
|
continue
|
|
_, comments = api.get(f"/issues/{i['number']}/comments")
|
|
comments = comments if isinstance(comments, list) else []
|
|
if has_research_round(comments):
|
|
# A round exists — re-queue ONLY if its disposition says MORE_RESEARCH
|
|
# (§14: the loop continues). Latest round wins (comments are append-only).
|
|
more = any(re.search(r"Disposition:\s*MORE_RESEARCH", c.get("body", ""), re.IGNORECASE) for c in comments)
|
|
if not more:
|
|
continue
|
|
targets.append(i)
|
|
return targets
|
|
|
|
|
|
def apply(api, numbers: list[int]) -> list[int]:
|
|
"""POST research-pending for each number; returns applied list."""
|
|
for n in numbers:
|
|
api.post(f"/issues/{n}/labels", {"labels": ["research-pending"]})
|
|
return numbers
|
|
|
|
|
|
def fetch_open_issues(api) -> list[dict]:
|
|
"""Paginated open issues (PRs filtered downstream) — page cap never
|
|
silently truncates the candidate set (teach-job lesson)."""
|
|
issues: list[dict] = []
|
|
page = 1
|
|
while True:
|
|
status, batch = api.get(f"/issues?state=open&type=issues&limit=50&page={page}")
|
|
if not isinstance(batch, list):
|
|
raise SweepError(f"issues page {page} returned non-list: {str(batch)[:200]}")
|
|
issues.extend(batch)
|
|
if len(batch) < 50:
|
|
return issues
|
|
page += 1
|
|
|
|
|
|
def main_with(argv: list[str]) -> int:
|
|
parser = argparse.ArgumentParser(description="Research-pending backlog sweep")
|
|
parser.add_argument("cmd", choices=["sweep"])
|
|
parser.add_argument("--base", required=True, help="repo-scoped API base …/repos/<o>/<r>")
|
|
parser.add_argument("--token", required=True)
|
|
parser.add_argument("--apply", action="store_true", help="label for real (default dry-run)")
|
|
args = parser.parse_args(argv[1:])
|
|
|
|
api = ForgeAdapter(args.base, args.token)
|
|
issues = fetch_open_issues(api)
|
|
targets = compute_targets(issues, api)
|
|
numbers = sorted(t["number"] for t in targets)
|
|
mode = "APPLY" if args.apply else "DRY-RUN"
|
|
print(f"research-sweep[{mode}]: {len(issues)} open issues, {len(numbers)} eligible")
|
|
for t in targets:
|
|
print(f" #{t['number']}: {t['title'][:70]}")
|
|
if args.apply and numbers:
|
|
apply(api, numbers)
|
|
print(f"research-sweep: labeled {len(numbers)} issue(s) research-pending")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
return main_with(sys.argv)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|