read_ci_verdict.py

#!/usr/bin/env python3
"""Read a PR's CI verdict from the machine field, never the display.

Decides PASS/FAIL on the *decisional* gates by reading ``statusCheckRollup[].name``
and ``.conclusion``. The gate policy is loaded from ``.github/required-checks.json``
(sub-delivery #6), the canonical source first wired by the doc-drift ``gate`` kind
(family G7). That file carries two name sets:

  - ``required``   -- every check the branch protection requires to merge
                      (quality, tests, security, AND "Analyze (python)").
  - ``decisional`` -- the subset this reader decides the verdict on
                      (quality, tests, security).

Their difference IS the required-check boundary, now declared as data instead of hard-coded:
"Analyze (python)" is required-to-merge but non-decisional (its SARIF upload is
blocked on the free private tier, so it shows nominally red). The reader decides
on ``decisional`` only; a red non-required check is ignored *by construction*.

This module is the committed twin of METHOD's rule "champ machine, pas
affichage": the verdict-reading is engraved once instead of being re-typed as a
fragile ``gh ... --jq 'select(...)'`` one-liner every session (the canal that bit
in E89, a full-width pipe smuggled into a hand-typed jq select).

The core (``decide``, ``load_policy``) is pure and offline-testable; the thin CLI
shell calls ``gh pr view <pr> --json statusCheckRollup`` and feeds the core.
Separating them is the oracle hierarchy: the live ``gh`` call ratifies, the local
parser motivates, and the red floor stays reproducible without network or auth.
"""

from __future__ import annotations

import json
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path

# The canonical gate policy: an object under .github/, resolved from this file's
# own location (never the cwd), so the reader works from wherever it is invoked.
_REPO_ROOT = Path(__file__).resolve().parent.parent
REQUIRED_CHECKS_PATH = _REPO_ROOT / ".github" / "required-checks.json"

# The only conclusion string that counts as green. Anything else -- FAILURE,
# CANCELLED, TIMED_OUT, in-progress (None) -- is not a pass (promote on the
# observed result, not the promise).
_GREEN = "SUCCESS"


# Floor-coverage claim (METHOD section 2): this tool carries no error vocabulary,
# so CODES is an explicit empty set (never silence), and FLOOR names its red floor.
CODES = frozenset()
FLOOR = "test_read_ci_verdict.py"


class PolicyError(ValueError):
    """The required-checks policy file is missing or unusable."""


@dataclass(frozen=True)
class Policy:
    required: tuple[str, ...]  # every merge-required check name
    decisional: tuple[str, ...]  # the subset the verdict is decided on


def _name_list(data: dict, key: str, path: Path) -> tuple[str, ...]:
    value = data.get(key)
    if not isinstance(value, list) or not all(isinstance(name, str) for name in value):
        raise PolicyError(f"'{key}' must be a JSON array of strings: {path}")
    if not value:
        raise PolicyError(f"'{key}' list is empty: {path}")
    return tuple(value)


def load_policy(path: Path) -> Policy:
    """Load the gate policy from the single-source JSON object.

    The file is a JSON object with ``required`` and ``decisional`` arrays of
    check names -- the declared source this reader consumes (and a future
    ruleset-setter reuses). Raises (never ``sys.exit`` -- this is core, the shell
    translates to an exit code) on every way the declaration can be unusable,
    because an unusable policy must fail loudly, never default to something that
    lets a PR through:

      - file absent,
      - malformed JSON, or not a JSON object,
      - either name set missing, not an array of strings, or empty.
    """
    try:
        raw = path.read_text(encoding="utf-8")
    except OSError as exc:
        raise PolicyError(f"required-checks file not found: {path}") from exc
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise PolicyError(f"required-checks file is not valid JSON: {path}") from exc
    if not isinstance(data, dict):
        raise PolicyError(f"required-checks must be a JSON object: {path}")
    return Policy(
        required=_name_list(data, "required", path),
        decisional=_name_list(data, "decisional", path),
    )


@dataclass(frozen=True)
class CheckResult:
    name: str
    present: bool  # the required check appears in the rollup at all
    conclusion: str | None  # its conclusion (None if absent or not yet concluded)
    ok: bool  # present AND conclusion == SUCCESS


@dataclass(frozen=True)
class Verdict:
    ok: bool  # all decisional checks present and green
    results: tuple[CheckResult, ...]  # one per decisional check, in `decisional` order

    def render(self) -> str:
        head = "PASS" if self.ok else "FAIL"
        lines = [f"CI verdict: {head}"]
        for r in self.results:
            if not r.present:
                mark, detail = "MISSING", "absent from rollup"
            elif r.ok:
                mark, detail = "ok", str(r.conclusion)
            else:
                mark, detail = "FAIL", (r.conclusion or "pending")
            lines.append(f"  {r.name}: {mark} ({detail})")
        return "\n".join(lines)


def decide(rollup, decisional: tuple[str, ...]) -> Verdict:
    """Decide the verdict from a statusCheckRollup list.

    ``rollup`` is the parsed ``statusCheckRollup`` array: a list of dicts, each
    a CheckRun (carries ``conclusion``) or a StatusContext (carries ``state``).
    ``decisional`` is the gate names the verdict is decided on (see
    ``load_policy``). Only those names move the verdict; everything else is
    ignored -- so a red non-decisional check (CodeQL) does not block.
    """
    seen: dict[str, str | None] = {}
    for entry in rollup:
        name = entry.get("name")
        if not name:
            continue
        conclusion = entry.get("conclusion")
        if not conclusion and entry.get("state"):
            conclusion = entry.get("state")  # StatusContext shape
        # Last entry wins on duplicate names (re-runs land later in the array).
        seen[name] = conclusion

    results = tuple(
        CheckResult(
            name=n,
            present=n in seen,
            conclusion=seen.get(n),
            ok=(n in seen and seen.get(n) == _GREEN),
        )
        for n in decisional
    )
    return Verdict(ok=all(r.ok for r in results), results=results)


def fetch_rollup(pr: str) -> list:
    """Fetch the statusCheckRollup for a PR via gh (the ratifying oracle)."""
    proc = subprocess.run(  # fixed argv, no shell (B603/B607 skipped via pyproject)
        ["gh", "pr", "view", str(pr), "--json", "statusCheckRollup"],
        capture_output=True,
        text=True,
        check=True,
    )
    data = json.loads(proc.stdout)
    return data.get("statusCheckRollup", [])


def main(argv: list[str]) -> int:
    if len(argv) != 1:
        print("usage: read_ci_verdict.py <pr-number>", file=sys.stderr)
        return 2
    try:
        policy = load_policy(REQUIRED_CHECKS_PATH)
    except PolicyError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    rollup = fetch_rollup(argv[0])
    verdict = decide(rollup, policy.decisional)
    print(verdict.render())
    return 0 if verdict.ok else 1


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))