fix(pr-monitor): REST transport — multi-node monitor runs exhaust GraphQL

pmoves/tools/pr_monitor.py

@@ -6,6 +6,7 @@
 import argparse
 import json
 import subprocess
+import sys
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
@@ -107,6 +108,31 @@ def _run_json(cmd: list[str]) -> Any:
     return json.loads(payload)
 
 
+def _run_json_pages(repo_path: str, *, list_key: str | None = None) -> list[Any]:
+    """Every page of a REST list endpoint, flattened.
+
+    `gh api --paginate` alone concatenates one JSON document per page, which
+    `json.loads` cannot read past the first page. `--slurp` wraps the pages in
+    one outer array; this flattens it. Endpoints that return an object with a
+    list inside (check-runs -> `check_runs`) pass `list_key` so each page's
+    list is concatenated instead of the page objects.
+    """
+    sep = "&" if "?" in repo_path else "?"
+    pages = _run_json(["gh", "api", "--paginate", "--slurp", f"{repo_path}{sep}per_page=100"]) or []
+    if not isinstance(pages, list):
+        pages = [pages]
+    out: list[Any] = []
+    for page in pages:
+        if list_key is not None:
+            if isinstance(page, dict):
+                out.extend(page.get(list_key) or [])
+        elif isinstance(page, list):
+            out.extend(page)
+        elif page is not None:
+            out.append(page)
+    return out
+
+
 def _parse_origin_remote() -> str | None:
     """Parse owner/repo from the origin remote URL.
 
@@ -201,7 +227,21 @@ def _review_thread_flags(repo: str, number: int) -> dict[str, tuple[bool, bool]]
         ]
         if cursor:
             cmd.extend(["-F", f"after={cursor}"])
-        payload = _run_json(cmd)
+        try:
+            payload = _run_json(cmd)
+        except RuntimeError as exc:
+            # Review threads are GraphQL-only (no REST equivalent), so thread
+            # resolved/outdated state degrades — loudly — when the GraphQL
+            # budget is exhausted. Comment bodies still flow: they come from
+            # the REST pulls endpoints and carry the learnings.
+            if "rate limit" in str(exc).lower():
+                print(
+                    f"warn: GraphQL rate-limited fetching review threads for #{number}; "
+                    "resolved/outdated flags unknown for this PR",
+                    file=sys.stderr,
+                )
+                break
+            raise
         if not isinstance(payload, dict):
             break
         data = payload.get("data")
@@ -451,45 +491,137 @@ def _collect_review_summary(repo: str, number: int, detail: dict[str, Any], *, h
 def _pr_numbers(repo: str, base: str, state: str, explicit_prs: list[int]) -> list[int]:
     if explicit_prs:
         return sorted(set(explicit_prs))
-    rows = _run_json(
-        [
-            "gh",
-            "pr",
-            "list",
-            "--repo",
-            repo,
-            "--state",
-            state,
-            "--base",
-            base,
-            "--json",
-            "number",
-        ]
-    )
-    if not isinstance(rows, list):
-        return []
+    # REST transport (operator direction 2026-09-05): `gh pr list --json` rides
+    # GraphQL, and multi-node monitor runs (B850 + SPARK + CI bots) exhaust the
+    # 5000/h GraphQL budget while REST stays plentiful. The pulls list endpoint
+    # carries everything this needs — number, base, state; `merged` maps to
+    # closed+filter because REST has no merged state of its own.
+    rest_state = "closed" if state == "merged" else state
+    rows = _run_json_pages(f"repos/{repo}/pulls?state={rest_state}&base={base}")
     out: list[int] = []
     for row in rows:
-        if isinstance(row, dict):
-            number = row.get("number")
-            if isinstance(number, int):
-                out.append(number)
+        if not isinstance(row, dict):
+            continue
+        if state == "merged" and not row.get("merged_at"):
+            continue
+        number = row.get("number")
+        if isinstance(number, int):
+            out.append(number)
     return sorted(set(out))
 
 
+def _pr_detail(repo: str, number: int) -> dict[str, Any]:
+    """PR detail via REST, shaped like `gh pr view --json` output.
+
+    `gh pr view --json` rides GraphQL, and multi-node monitor runs (B850 +
+    SPARK + CI bots in one hour) exhaust the 5000/h GraphQL budget while REST
+    stays plentiful — measured live 2026-09-05. Every field below has a REST
+    equivalent except review-thread resolved/outdated state, which has none
+    and degrades separately in `_review_thread_flags`.
+    """
+    detail = _run_json(["gh", "api", f"repos/{repo}/pulls/{number}"])
+    if not isinstance(detail, dict):
+        raise RuntimeError(f"invalid PR payload for #{number}")
+    head = detail.get("head") or {}
+    base = detail.get("base") or {}
+    head_sha = str(head.get("sha") or "")
+
+    comments = _run_json_pages(f"repos/{repo}/issues/{number}/comments")
+    reviews = _run_json_pages(f"repos/{repo}/pulls/{number}/reviews")
+
+    # reviewDecision: latest SUBMITTED review per author wins (GitHub's own
+    # semantics — PENDING is a draft review and does not count); any APPROVED
+    # -> APPROVED, else any CHANGES_REQUESTED -> CHANGES_REQUESTED.
+    # GitHub's own semantics, over EVERY page of reviews in submission order:
+    # PENDING is a draft and does not count; COMMENTED never changes an author's
+    # standing; DISMISSED clears it; the latest APPROVED / CHANGES_REQUESTED is
+    # the author's standing; and one outstanding CHANGES_REQUESTED blocks no
+    # matter how many approvals exist -- change requests are checked FIRST.
+    latest_by_author: dict[str, str] = {}
+    for review in reviews:
+        if not isinstance(review, dict):
+            continue
+        user = review.get("user") or {}
+        login = str(user.get("login") or "")
+        state = str(review.get("state") or "").upper()
+        if not login:
+            continue
+        if state in {"APPROVED", "CHANGES_REQUESTED"}:
+            latest_by_author[login] = state
+        elif state == "DISMISSED":
+            latest_by_author.pop(login, None)
+    states = set(latest_by_author.values())
+    if "CHANGES_REQUESTED" in states:
+        review_decision = "CHANGES_REQUESTED"
+    elif "APPROVED" in states:
+        review_decision = "APPROVED"
+    else:
+        review_decision = "REVIEW_REQUIRED"
+
+    mergeable = {True: "MERGEABLE", False: "CONFLICTING"}.get(detail.get("mergeable"), "UNKNOWN")
+    state_map = {
+        "clean": "CLEAN",
+        "has_hooks": "HAS_HOOKS",
+        "unstable": "UNSTABLE",
+        "blocked": "BLOCKED",
+        "dirty": "DIRTY",
+        "draft": "DRAFT",
+    }
+    merge_state = state_map.get(str(detail.get("mergeable_state") or ""), "UNKNOWN")
+
+    rollup: list[dict[str, Any]] = []
+    if head_sha:
+        # Every page of check-runs: a busy PR exceeds the 30-per-page default and
+        # a truncated list under-counts pending/failed runs.
+        for run in _run_json_pages(f"repos/{repo}/commits/{head_sha}/check-runs", list_key="check_runs"):
+            if isinstance(run, dict):
+                rollup.append(
+                    {
+                        "__typename": "CheckRun",
+                        "status": run.get("status"),
+                        "conclusion": run.get("conclusion"),
+                    }
+                )
+        # Legacy commit statuses (external CI / bots post these, not check-runs)
+        # are what the GraphQL rollup folded in as StatusContext.
+        combined = _run_json(["gh", "api", f"repos/{repo}/commits/{head_sha}/status"]) or {}
+        statuses = combined.get("statuses") if isinstance(combined, dict) else None
+        for status in statuses or []:
+            if isinstance(status, dict):
+                rollup.append(
+                    {
+                        "__typename": "StatusContext",
+                        "state": status.get("state"),
+                        "context": status.get("context"),
+                    }
+                )
+
+    def _author_body_url(row: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "author": row.get("user"),
+            "body": row.get("body"),
+            "url": row.get("html_url"),
+        }
+
+    return {
+        "number": detail.get("number"),
+        "title": detail.get("title"),
+        "url": detail.get("html_url"),
+        "headRefName": head.get("ref"),
+        "headRefOid": head_sha,
+        "baseRefName": base.get("ref"),
+        "mergeable": mergeable,
+        "mergeStateStatus": merge_state,
+        "reviewDecision": review_decision,
+        "isDraft": bool(detail.get("draft")),
+        "statusCheckRollup": rollup,
+        "comments": [_author_body_url(c) for c in comments if isinstance(c, dict)],
+        "reviews": [_author_body_url(r) for r in reviews if isinstance(r, dict)],
+    }
+
+
 def _pr_summary(repo: str, number: int) -> PrSummary:
-    detail = _run_json(
-        [
-            "gh",
-            "pr",
-            "view",
-            str(number),
-            "--repo",
-            repo,
-            "--json",
-            "number,title,url,headRefName,headRefOid,baseRefName,mergeable,mergeStateStatus,reviewDecision,isDraft,statusCheckRollup,comments,reviews",
-        ]
-    )
+    detail = _pr_detail(repo, number)
     if not isinstance(detail, dict):
         raise RuntimeError(f"invalid PR payload for #{number}")
     checks = _check_summary(detail.get("statusCheckRollup"))