From f5f27cb870a9c49feb4ffbb076899add06c5af0b Mon Sep 17 00:00:00 2001 From: Molecule AI Infra-SRE Date: Mon, 11 May 2026 19:08:04 +0000 Subject: [PATCH] =?UTF-8?q?fix(ci):=20gate-check-v3=20=E2=80=94=203=20bug?= =?UTF-8?q?=20fixes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug 1 (self-referential failure loop, #544): signal_6_ci now filters out its own prior status from check_statuses before evaluating, preventing a gate-check-v3 → failure → re-reads self → failure cycle. Bug 2 (hardcoded base branch, #544): signal_6_ci now uses the PR's actual base branch ref instead of hardcoded 'main'. Caller passes PR data to avoid redundant API call. Bug 3 (comment-post 403, #543): Wrapped POST/PATCH comment-post in try/except for HTTPError 403. Logs a warning and skips posting when the token lacks write:repository scope — verdict still drives exit code correctly. Also removed 3 lines of dead code at the end of format_comment (unreachable return after prior return). Co-Authored-By: Claude Opus 4.7 --- tools/gate-check-v3/gate_check.py | 57 +++++++++++++++++++------------ 1 file changed, 35 insertions(+), 22 deletions(-) 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