pr_preflight.py

#!/usr/bin/env python3
"""Resolve PR metadata, then run the shared deterministic check manifest."""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass
from pathlib import Path

from pr_metadata import TransientPRMetadataError, PullRequestMetadata, load_from_api, load_from_event_file, load_from_gh
from run_checks import (
    MANIFEST_RELATIVE_PATH,
    detect_platform,
    load_manifest,
    manifest_changed_check_ids,
    resolve_checks,
)


@dataclass(frozen=True)
class Check:
    name: str
    reason: str


def _resolve_repo_root() -> Path:
    """Return the worktree root even when the linked-worktree Git context is bare."""
    try:
        return Path(run_git(Path.cwd(), "rev-parse", "--show-toplevel"))
    except subprocess.CalledProcessError:
        return Path.cwd()


def run_git(root: Path, *args: str) -> str:
    result = subprocess.run(
        ["git", *args],
        cwd=root,
        check=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        encoding="utf-8",
    )
    return result.stdout.strip()


def run_python_capture(root: Path, *args: str, errors: str = "strict") -> subprocess.CompletedProcess[str]:
    """Run an owned Python check with a UTF-8 pipe contract on every host."""
    env = os.environ.copy()
    env["PYTHONIOENCODING"] = "utf-8:backslashreplace"
    return subprocess.run(
        [sys.executable, *args],
        cwd=root,
        env=env,
        check=False,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        encoding="utf-8",
        errors=errors,
    )


def current_branch(root: Path) -> str:
    """Return the current branch without inheriting the Windows host locale."""
    return subprocess.run(
        ["git", "symbolic-ref", "--short", "-q", "HEAD"],
        cwd=root,
        check=False,
        stdout=subprocess.PIPE,
        text=True,
        encoding="utf-8",
        errors="backslashreplace",
    ).stdout.strip()


def configure_output_streams() -> None:
    """Keep direct Windows preflight output UTF-8 and non-fatal."""
    for stream in (sys.stdout, sys.stderr):
        reconfigure = getattr(stream, "reconfigure", None)
        if not callable(reconfigure):
            continue
        options = {"errors": "backslashreplace"}
        if os.name == "nt":
            options["encoding"] = "utf-8"
        reconfigure(**options)


def changed_files(root: Path, base: str, head: str, *, include_worktree: bool = False) -> list[str]:
    output = run_git(
        root,
        "diff",
        "--name-only",
        "--no-renames",
        "--diff-filter=ACMRTD",
        f"{base}...{head}",
    )
    files = set(output.splitlines())
    if include_worktree and head == "HEAD":
        files.update(run_git(root, "diff", "--name-only", "--no-renames", "--diff-filter=ACMRTD", "HEAD").splitlines())
        files.update(run_git(root, "ls-files", "--others", "--exclude-standard").splitlines())
    return sorted(path for path in files if path)


def select_checks(
    files: list[str],
    lane: str = "ci",
    platform: str | None = None,
    *,
    metadata_only: bool = False,
    root: Path | None = None,
    base: str | None = None,
    head: str = "HEAD",
) -> list[Check]:
    root = root or Path(__file__).resolve().parents[2]
    manifest = load_manifest(root / MANIFEST_RELATIVE_PATH)
    changed_ids = (
        manifest_changed_check_ids(root, base, head, include_worktree=lane == "local")
        if base is not None and MANIFEST_RELATIVE_PATH in files
        else None
    )
    return [
        Check(check.id, check.reason)
        for check in resolve_checks(
            manifest,
            files,
            lane,
            platform=platform or detect_platform(),
            manifest_changed_ids=changed_ids,
        )
        if not metadata_only or check.requires_pr_body
    ]


def format_failure_class_suggest(payload: dict) -> str:
    """Render manual failure-class guidance alongside invariant suggestions.

    Failure classes deliberately do not infer a classification from paths or a
    diff. This formatter therefore supplies the required field and structured
    choices while making the author-owned decision explicit.
    """
    lines = ["## Failure class (fixes)", ""]
    if not payload.get("requires_declaration"):
        lines.extend(["No `fix:` commits were detected; no declaration is required.", ""])
        return "\n".join(lines)

    patch = payload.get("pr_body_patch", {})
    declaration = (
        patch.get("text", "Failure-Class: none\n").strip() if isinstance(patch, dict) else "Failure-Class: none"
    )
    lines.extend(
        [
            declaration,
            "",
            "<!-- A `fix:` commit is in this diff. Choose manually: this command does not infer a class from paths or diffs.",
            "Before opening the PR, inspect a relevant class with `scripts/failure-class explain FC-<slug> --format json`; replace `none` only if an existing class applies, or use `new` for a genuinely new class.",
            _candidate_heading(payload),
        ]
    )
    for candidate in payload.get("advisory_candidates", []):
        if isinstance(candidate, dict):
            lines.append(f"- {candidate['id']}: {candidate['violated_contract']}")
    lines.extend(["-->", ""])
    return "\n".join(lines)


def _candidate_heading(payload: dict) -> str:
    """Label the candidate list honestly.

    `failure-class prepare` lists the classes whose advisory scope_hints overlap the
    change, falling back to the whole registry when none match. Calling a narrowed list
    "Available classes" would imply the others do not apply, which is a classification
    the CLI does not make.
    """
    shown = payload.get("candidates_shown")
    total = payload.get("candidates_total")
    if isinstance(shown, int) and isinstance(total, int) and shown < total:
        return (
            f"Classes whose scope_hints overlap this change ({shown} of {total}; advisory only "
            "— run `scripts/failure-class prepare --all-candidates` for the rest):"
        )
    return "Available classes:"


def resolve_pr_metadata(
    root: Path,
    body_file: Path | None,
    repository: str | None,
    pr_number: int | None,
    event_payload_file: Path | None,
) -> PullRequestMetadata | None:
    if body_file is None:
        env_body = os.getenv("OMI_PR_BODY_FILE", "").strip()
        if env_body:
            body_file = Path(env_body)
    if body_file:
        resolved = body_file.expanduser()
        if not resolved.is_file():
            raise RuntimeError(f"PR body file not found: {resolved}")
        return PullRequestMetadata(
            number=0,
            body=resolved.read_text(encoding="utf-8"),
            updated_at="local file",
            labels=(),
            source=str(resolved.resolve()),
        )
    if repository and pr_number:
        try:
            return load_from_api(repository, pr_number, os.getenv("GITHUB_TOKEN", ""))
        except TransientPRMetadataError as api_error:
            if event_payload_file is None:
                raise
            try:
                metadata = load_from_event_file(event_payload_file, pr_number)
            except RuntimeError as event_error:
                raise RuntimeError(f"{api_error}; GitHub event fallback failed: {event_error}") from event_error
            print(f"WARNING: {api_error}; using the PR snapshot from {event_payload_file}.", file=sys.stderr)
            return metadata
    try:
        return load_from_gh(root)
    except RuntimeError as exc:
        print(f"PR metadata: unavailable ({exc})")
        print("If invariant citations are required, rerun with --pr-body-file <draft.md>")
        print("or set OMI_PR_BODY_FILE, or run: scripts/pr-preflight --suggest")
        return None


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base", default="origin/main")
    parser.add_argument("--head", default="HEAD")
    parser.add_argument("--lane", choices=("local", "ci"), default="ci")
    parser.add_argument("--pr-body-file", type=Path)
    parser.add_argument(
        "--metadata-only",
        action="store_true",
        help="Validate only metadata-dependent contracts, reporting all errors in one pass.",
    )
    parser.add_argument("--repository", help="GitHub repository as owner/name; requires --pr-number")
    parser.add_argument("--pr-number", type=int, help="Load current PR metadata through the GitHub API")
    parser.add_argument(
        "--event-payload-file",
        type=Path,
        help="Use the triggering pull_request event only after transient GitHub API metadata failures",
    )
    parser.add_argument("--head-branch", help="PR head branch, used for release-changelog policy")
    parser.add_argument("--list", action="store_true", help="Print selected checks without running them")
    parser.add_argument(
        "--suggest",
        action="store_true",
        help="Print paste-ready product-invariant and failure-class PR guidance for the diff and exit 0",
    )
    parser.add_argument("--root", type=Path)
    return parser.parse_args()


def main() -> int:
    configure_output_streams()
    args = parse_args()
    if bool(args.repository) != bool(args.pr_number):
        print("FAIL: --repository and --pr-number must be supplied together", file=sys.stderr)
        return 2
    root = (args.root or _resolve_repo_root()).resolve()
    started = time.monotonic()
    try:
        merge_base = run_git(root, "merge-base", args.base, args.head)
        files = changed_files(root, args.base, args.head, include_worktree=args.lane == "local")
    except subprocess.CalledProcessError as exc:
        print(f"FAIL: could not resolve preflight diff: {exc.stderr.strip()}", file=sys.stderr)
        return 1
    checks = select_checks(
        files,
        args.lane,
        metadata_only=args.metadata_only,
        root=root,
        base=merge_base,
        head=args.head,
    )
    summary = f"PR preflight: lane={args.lane} base={args.base} ({merge_base[:12]}) head={args.head} files={len(files)}"
    print(summary, file=sys.stderr if args.suggest else sys.stdout)
    for check in checks:
        print(f"  SELECTED {check.name}: {check.reason}", file=sys.stderr if args.suggest else sys.stdout)
    if args.list:
        return 0

    with tempfile.TemporaryDirectory(prefix="omi-pr-preflight-") as temp_dir:
        temp = Path(temp_dir)
        files_path = temp / "changed-files.txt"
        files_path.write_text("".join(f"{path}\n" for path in files), encoding="utf-8")
        if args.suggest:
            invariants = run_python_capture(
                root,
                ".github/scripts/check_product_invariants.py",
                "--changed-files",
                str(files_path),
                "--suggest",
                errors="backslashreplace",
            )
            if invariants.stdout:
                print(invariants.stdout, end="")
            if invariants.returncode:
                return invariants.returncode

            suggestion_body = temp / "suggest-pr-body.md"
            suggestion_body.write_text("", encoding="utf-8")
            failure_classes = run_python_capture(
                root,
                "scripts/failure-class",
                "prepare",
                "--base",
                args.base,
                "--head",
                args.head,
                "--pr-body-file",
                str(suggestion_body),
                "--format",
                "json",
            )
            if failure_classes.returncode:
                print("FAIL: failure-class preparation failed.", file=sys.stderr)
                if failure_classes.stdout:
                    print(failure_classes.stdout, end="", file=sys.stderr)
                return failure_classes.returncode
            try:
                payload = json.loads(failure_classes.stdout)
            except json.JSONDecodeError:
                print(
                    "FAIL: failure-class preparation returned invalid JSON.",
                    file=sys.stderr,
                )
                return 1
            print(format_failure_class_suggest(payload), end="")
            return 0

        try:
            metadata = resolve_pr_metadata(
                root, args.pr_body_file, args.repository, args.pr_number, args.event_payload_file
            )
        except RuntimeError as exc:
            print(f"FAIL: {exc}", file=sys.stderr)
            return 1
        head_branch = args.head_branch or os.getenv("GITHUB_HEAD_REF", "")
        if not head_branch:
            head_branch = current_branch(root)
        skip_changelog = head_branch.startswith("changelog/v") or os.getenv("PRE_PUSH_SKIP_DESKTOP_CHANGELOG") == "1"
        if metadata:
            print(f"PR metadata: {metadata.source}, updated_at={metadata.updated_at}")
        elif any(check.name == "product-invariants" for check in checks):
            print("PR metadata: none (product-invariants will use an empty body)")

        body_path = temp / "pr-body.txt"
        body_path.write_text(metadata.body if metadata else "", encoding="utf-8")
        command = [
            sys.executable,
            ".github/scripts/run_checks.py",
            "--lane",
            args.lane,
            "--base",
            args.base,
            "--head",
            args.head,
            "--changed-files",
            str(files_path),
            "--pr-body-file",
            str(body_path),
        ]
        if args.metadata_only:
            command.append("--metadata-only")
        if skip_changelog:
            command.append("--skip-changelog")
        result = subprocess.run(command, cwd=root, check=False)

    elapsed = time.monotonic() - started
    if result.returncode:
        print(f"PR preflight failed in {elapsed:.2f}s.", file=sys.stderr)
        return result.returncode
    print(f"PR preflight passed: {len(checks)} checks in {elapsed:.2f}s.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())