diff --git a/tools/gate-check-v3/gate_check.py b/tools/gate-check-v3/gate_check.py index 429c2b40..38ed66d3 100644 --- a/tools/gate-check-v3/gate_check.py +++ b/tools/gate-check-v3/gate_check.py @@ -316,7 +316,7 @@ def signal_3_staleness(pr_number: int, repo: str) -> dict: # ── Signal 6: CI required-checks awareness ─────────────────────────────────── -def signal_6_ci(pr_number: int, repo: str, branch: str = "main") -> dict: +def signal_6_ci(pr_number: int, repo: str, branch: str | None = None, pr_data: dict | None = None) -> dict: """ Query combined CI status for PR head commit. Find required status checks on target branch. @@ -324,8 +324,12 @@ def signal_6_ci(pr_number: int, repo: str, branch: str = "main") -> dict: """ owner, name = repo.split("/", 1) - pr = api_get(f"/repos/{owner}/{name}/pulls/{pr_number}") - head_sha = pr["head"]["sha"] + # Re-use PR data if already fetched by caller; otherwise fetch once. + if pr_data is None: + pr_data = api_get(f"/repos/{owner}/{name}/pulls/{pr_number}") + head_sha = pr_data["head"]["sha"] + # Fall back to PR's actual base branch when no explicit branch is given + branch = branch or pr_data.get("base", {}).get("ref", "main") # Combined status of PR head combined = api_get(f"/repos/{owner}/{name}/commits/{head_sha}/status") @@ -334,9 +338,12 @@ def signal_6_ci(pr_number: int, repo: str, branch: str = "main") -> dict: # Individual check statuses # Gitea Actions uses "status" (pending/success/failure) not "state" for # individual check entries. "state" is null for pending runs. + # Exclude our own prior status to prevent self-referential failure loops. check_statuses = {} for s in combined.get("statuses") or []: - check_statuses[s["context"]] = s.get("status", "pending") + ctx = s["context"] + if "gate-check" not in ctx.lower(): + check_statuses[ctx] = s.get("status", "pending") # Try to get branch protection for required checks required_checks = [] @@ -459,21 +466,21 @@ def format_comment(repo: str, pr_number: int, verdict: str, gates: list[dict], b lines.append(f"_gate-check-v3 · repo={repo} · pr={pr_number}_") return "\n".join(lines) - lines.append("") - lines.append(f"_gate-check-v3 · repo={repo} · pr={pr_number}_") - - return "\n".join(lines) - # ── Main ───────────────────────────────────────────────────────────────────── def run(repo: str, pr_number: int, post_comment: bool = False) -> dict: try: + # Fetch PR once to get base ref for signal_6_ci + owner, name = repo.split("/", 1) + pr = api_get(f"/repos/{owner}/{name}/pulls/{pr_number}") + base_ref = pr.get("base", {}).get("ref", "main") + gates = [ signal_1_comment_scan(pr_number, repo), signal_2_reviews(pr_number, repo), signal_3_staleness(pr_number, repo), - signal_6_ci(pr_number, repo), + signal_6_ci(pr_number, repo, branch=base_ref, pr_data=pr), ] verdict, blockers = compute_verdict(gates) @@ -501,18 +508,24 @@ def run(repo: str, pr_number: int, post_comment: bool = False) -> dict: # Check if a gate-check comment already exists to avoid spamming existing = api_list(f"/repos/{owner}/{name}/issues/{pr_number}/comments") our_comments = [c for c in existing if "[gate-check-v3]" in (c.get("body") or "")] - if our_comments: - # Update latest - comment_id = our_comments[-1]["id"] - url = f"{API_BASE}/repos/{owner}/{name}/issues/comments/{comment_id}" - req = urllib.request.Request(url, data=json.dumps({"body": comment_body}).encode(), headers=headers, method="PATCH") - with urllib.request.urlopen(req) as r: - r.read() - else: - url = f"{API_BASE}/repos/{owner}/{name}/issues/{pr_number}/comments" - req = urllib.request.Request(url, data=json.dumps({"body": comment_body}).encode(), headers=headers, method="POST") - with urllib.request.urlopen(req) as r: - r.read() + try: + if our_comments: + # Update latest + comment_id = our_comments[-1]["id"] + url = f"{API_BASE}/repos/{owner}/{name}/issues/comments/{comment_id}" + req = urllib.request.Request(url, data=json.dumps({"body": comment_body}).encode(), headers=headers, method="PATCH") + with urllib.request.urlopen(req) as r: + r.read() + else: + url = f"{API_BASE}/repos/{owner}/{name}/issues/{pr_number}/comments" + req = urllib.request.Request(url, data=json.dumps({"body": comment_body}).encode(), headers=headers, method="POST") + with urllib.request.urlopen(req) as r: + r.read() + except urllib.error.HTTPError as e: + if e.code == 403: + print(f"WARN: --post-comment 403 (token scope) — verdict={verdict}; skipping comment-post", file=sys.stderr) + else: + raise return result