#!/usr/bin/env python3 """Reuse exact-SHA PR unit evidence; uncertainty always requests fresh units.""" import json import os from pathlib import Path import re import subprocess import time WORKFLOW = ".github/workflows/ci.yml" UNIT_JOBS = ( "unit (self-hosted, Linux, kitsubito)", "unit (self-hosted, Windows, hfenduleam)", ) class Unproven(Exception): pass def require(condition, reason): if not condition: raise Unproven(reason) def positive_id(value): return type(value) is int and value > 0 class GitHub: def __init__(self): self.deadline = time.monotonic() + 90 def get(self, endpoint): remaining = self.deadline - time.monotonic() require(remaining > 0, "api-deadline") try: response = subprocess.run( ["gh", "api", "--hostname", "github.com", "--method", "GET", "-H", "Accept: application/vnd.github+json", "-H", "X-GitHub-Api-Version: 2022-11-28", endpoint], capture_output=True, text=True, timeout=min(15, remaining), ) except subprocess.TimeoutExpired: raise Unproven("api-timeout") from None except OSError: raise Unproven("gh-unavailable") from None # Never print CLI stderr: neither credentials nor untrusted API text belong # in workflow commands. Failure categories are fixed, named strings. require(response.returncode == 0, "api-error") try: result = json.loads(response.stdout) except (ValueError, TypeError): raise Unproven("api-malformed-json") from None require(isinstance(result, dict), "api-malformed-object") return result def collection(api, endpoint, key): """Read every page and reject truncation, duplicates and changing counts.""" rows = [] ids = set() total = None separator = "&" if "?" in endpoint else "?" for page in range(1, 11): response = api.get(f"{endpoint}{separator}per_page=100&page={page}") count = response.get("total_count") batch = response.get(key) require(type(count) is int and 0 <= count < 1000, "api-count-unproven") require(isinstance(batch, list) and len(batch) <= 100, "api-malformed-page") if total is None: total = count require(count == total, "api-pagination-changed") for row in batch: require(isinstance(row, dict) and positive_id(row.get("id")), "api-malformed-row") require(row["id"] not in ids, "api-duplicate-row") ids.add(row["id"]) rows.extend(batch) require(len(rows) <= total, "api-count-mismatch") if len(rows) == total: return rows require(len(batch) == 100, "api-incomplete-page") raise Unproven("api-pagination-limit") def same_repository(value, repo, repo_id): return (isinstance(value, dict) and value.get("full_name") == repo and type(value.get("id")) is int and value["id"] == repo_id) def run_matches(run, repo, repo_id, workflow_id, sha): return ( isinstance(run, dict) and positive_id(run.get("id")) and positive_id(run.get("run_attempt")) and type(run.get("workflow_id")) is int and run["workflow_id"] == workflow_id and run.get("path") == WORKFLOW and run.get("event") == "pull_request" and run.get("head_sha") == sha and same_repository(run.get("repository"), repo, repo_id) # Same-repository PRs only: a fork's green jobs cannot authorize reuse. and same_repository(run.get("head_repository"), repo, repo_id) and run.get("status") == "completed" and run.get("conclusion") == "success" ) # [impl->REQ-CI-PUSH-MAIN-UNIT-REUSE] def decide(env, api): if env.get("GITHUB_EVENT_NAME") != "push" or env.get("GITHUB_REF") != "refs/heads/main": return True, "not-push-main" repo = env.get("GITHUB_REPOSITORY", "") sha = env.get("GITHUB_SHA", "") repo_id_text = env.get("GITHUB_REPOSITORY_ID", "") require(env.get("GITHUB_SERVER_URL") == "https://github.com", "unsupported-server") require(re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repo) is not None, "invalid-repository") require(re.fullmatch(r"[0-9a-f]{40}", sha) is not None, "invalid-sha") require(repo_id_text.isascii() and repo_id_text.isdecimal() and int(repo_id_text) > 0, "invalid-repository-id") require(bool(env.get("GH_TOKEN")), "missing-token") repo_id = int(repo_id_text) root = f"repos/{repo}/actions" workflow = api.get(f"{root}/workflows/ci.yml") require(positive_id(workflow.get("id")) and workflow.get("path") == WORKFLOW and workflow.get("name") == "ci", "workflow-identity-mismatch") workflow_id = workflow["id"] runs = collection(api, f"{root}/workflows/{workflow_id}/runs?event=pull_request&head_sha={sha}", "workflow_runs") reason = "no-exact-successful-pr-run" for candidate in runs: if not run_matches(candidate, repo, repo_id, workflow_id, sha): continue run_id = candidate["id"] endpoint = f"{root}/runs/{run_id}" current = api.get(endpoint) if not run_matches(current, repo, repo_id, workflow_id, sha) or current["id"] != run_id: reason = "run-no-longer-qualifies" continue attempt = current["run_attempt"] attempt_run = api.get(f"{endpoint}/attempts/{attempt}") require(run_matches(attempt_run, repo, repo_id, workflow_id, sha) and attempt_run["id"] == run_id and attempt_run["run_attempt"] == attempt, "attempt-identity-or-result-mismatch") # The attempt-specific endpoint cannot silently assemble a green pair # from Windows in attempt 1 and Linux in a failed-jobs-only attempt 2. jobs = collection(api, f"{endpoint}/attempts/{attempt}/jobs", "jobs") require(all(job.get("run_id") == run_id and job.get("run_attempt") == attempt and job.get("head_sha") == sha and job.get("status") == "completed" for job in jobs), "jobs-incomplete-or-mixed-attempt") units = [job for job in jobs if job.get("name") in UNIT_JOBS] if (len(units) != len(UNIT_JOBS) or {job["name"] for job in units} != set(UNIT_JOBS) or any(job.get("conclusion") != "success" for job in units)): reason = "both-unit-jobs-not-successful" continue # A rerun starting while pages were read invalidates this snapshot. final = api.get(endpoint) require(run_matches(final, repo, repo_id, workflow_id, sha) and final["id"] == run_id and final["run_attempt"] == attempt, "run-changed-during-proof") return False, f"exact-pr-proof run={run_id} attempt={attempt} sha={sha}" return True, reason def main(): try: run_unit, reason = decide(os.environ, GitHub()) except Unproven as error: run_unit, reason = True, str(error) except Exception: # Unexpected response shapes and local failures are uncertainty, not a skip. run_unit, reason = True, "internal-error" value = "true" if run_unit else "false" print(f"UNIT_REUSE run-unit={value} reason={reason}") output = os.environ.get("GITHUB_OUTPUT") if output: with Path(output).open("a", encoding="utf-8") as stream: stream.write(f"run-unit={value}\n") return 0 if __name__ == "__main__": raise SystemExit(main())