scripts/release_gate.py

#!/usr/bin/env python3
"""Fail-fast, pre-upload release gate for the public HF Space."""
from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
import urllib.request
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
LATEST_MANIFEST_URL = "https://huggingface.co/buckets/mzx/dilutionrisk-public/resolve/manifests/v1/latest.json"
REVISION = re.compile(r"^[0-9a-f]{40}$")
SEMVER = re.compile(r"^\d+\.\d+\.\d+$")


def command(arguments: list[str], *, capture: bool = True) -> str:
    result = subprocess.run(
        arguments,
        cwd=ROOT,
        check=True,
        text=True,
        capture_output=capture,
    )
    return (result.stdout or "") + (result.stderr or "")


def read_json(path: Path) -> dict:
    return json.loads(path.read_text())


def fetch_latest_manifest() -> dict:
    request = urllib.request.Request(
        LATEST_MANIFEST_URL,
        headers={"User-Agent": "dilutionrisk-release-gate/1"},
    )
    with urllib.request.urlopen(request, timeout=30) as response:
        return json.load(response)


def validate_release_files(root: Path, latest_manifest: dict) -> dict:
    release = read_json(root / "release.json")
    baseline = read_json(root / "LOAD_BASELINE.json")
    readme = (root / "README.md").read_text()
    requirements = (root / "requirements.txt").read_text()

    if not SEMVER.fullmatch(release.get("service_version", "")):
        raise ValueError("release service_version must be semantic x.y.z")
    revision = release.get("dataset_revision", "")
    if not REVISION.fullmatch(revision):
        raise ValueError("release Dataset revision must be a 40-character SHA")
    if release.get("dataset_id") != "mzx/dilutionrisk-data":
        raise ValueError("release Dataset must stay in the mzx namespace")
    if latest_manifest.get("dataset_revision") != revision:
        raise ValueError("release Dataset revision does not match the public latest manifest")
    if baseline.get("dataset_revision") != revision:
        raise ValueError("stored load baseline does not match the release Dataset revision")
    hosted = baseline.get("hosted_cpu_basic", {})
    if hosted.get("rest", {}).get("successful") != hosted.get("rest", {}).get("concurrent_requests"):
        raise ValueError("stored hosted REST baseline is missing or contains failures")
    if hosted.get("mcp", {}).get("successful") != hosted.get("mcp", {}).get("concurrent_sessions"):
        raise ValueError("stored hosted MCP baseline is missing or contains failures")
    for required in ("sdk_version: 6.20.0", "python_version: '3.13'", "mzx/dilutionrisk-mcp"):
        if required not in readme:
            raise ValueError(f"README is missing pinned release marker: {required}")
    for required in ("gradio==6.20.0", "duckdb==1.4.5", "huggingface-hub==1.8.0"):
        if required not in requirements:
            raise ValueError(f"requirements lock is missing {required}")
    return {
        "service_version": release["service_version"],
        "dataset_id": release["dataset_id"],
        "dataset_revision": revision,
        "manifest_published_at": latest_manifest.get("published_at"),
        "source_watermark": latest_manifest.get("source_watermark"),
        "baseline_service_version": baseline.get("service_version"),
        "baseline_build_sha": hosted.get("build_sha"),
    }


def run_gate(*, allow_dirty: bool = False, skip_network: bool = False) -> dict:
    command(["git", "diff", "--check"])
    status = command(["git", "status", "--porcelain", "--untracked-files=all"]).strip()
    if status and not allow_dirty:
        raise RuntimeError("release candidate tree is not clean; commit intentional changes first")

    test_output = command(
        [sys.executable, "-m", "unittest", "discover", "-s", "tests", "-v"]
    )
    print(test_output, end="")
    match = re.search(r"Ran (\d+) tests", test_output)
    if not match:
        raise RuntimeError("could not confirm unit-test count")

    manifest = (
        {"dataset_revision": read_json(ROOT / "release.json")["dataset_revision"]}
        if skip_network
        else fetch_latest_manifest()
    )
    release = validate_release_files(ROOT, manifest)
    return {
        "gate": "passed",
        "tests_passed": int(match.group(1)),
        "git_commit": command(["git", "rev-parse", "HEAD"]).strip(),
        "git_tree": command(["git", "rev-parse", "HEAD^{tree}"]).strip(),
        **release,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--allow-dirty", action="store_true")
    parser.add_argument("--skip-network", action="store_true")
    options = parser.parse_args()
    report = run_gate(
        allow_dirty=options.allow_dirty,
        skip_network=options.skip_network,
    )
    print(json.dumps(report, indent=2, sort_keys=True))
    return 0


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