scripts/snapshot_kernel_contracts.py

#!/usr/bin/env python3
"""Snapshot immutable Hugging Face Kernel revisions into reviewable contracts.

The pin file is deliberately hand-reviewed. This script never discovers or
advances a revision. It only resolves the exact 40-character revisions already
present in registry/kernel-pins.json and records their file digests.
"""

from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
import re
import urllib.request


ROOT = Path(__file__).resolve().parents[1]
PINS_PATH = ROOT / "registry" / "kernel-pins.json"
CONTRACTS_DIR = ROOT / "contracts"
HEX40 = re.compile(r"^[0-9a-f]{40}$")


def get_json(url: str) -> object:
    request = urllib.request.Request(
        url,
        headers={"Accept": "application/json", "User-Agent": "szl-kernels-live/1.0"},
    )
    with urllib.request.urlopen(request, timeout=30) as response:
        return json.load(response)


def get_bytes(url: str) -> bytes:
    request = urllib.request.Request(
        url,
        headers={"User-Agent": "szl-kernels-live/1.0"},
    )
    with urllib.request.urlopen(request, timeout=30) as response:
        return response.read()


def canonical_json(value: object) -> bytes:
    return json.dumps(
        value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")


def load_pins() -> dict:
    pins = json.loads(PINS_PATH.read_text(encoding="utf-8"))
    kernels = pins.get("kernels")
    if not isinstance(kernels, list) or len(kernels) != 10:
        raise ValueError("pin registry must contain exactly ten kernels")
    seen: set[str] = set()
    for kernel in kernels:
        kernel_id = kernel.get("id")
        revision = kernel.get("revision")
        if not isinstance(kernel_id, str) or kernel_id in seen:
            raise ValueError(f"invalid or duplicate kernel id: {kernel_id!r}")
        if not isinstance(revision, str) or not HEX40.fullmatch(revision):
            raise ValueError(f"{kernel_id}: revision must be a full lowercase SHA")
        seen.add(kernel_id)
    return pins


def snapshot_kernel(kernel: dict, recorded_at: str) -> dict:
    kernel_id = kernel["id"]
    revision = kernel["revision"]
    detail = get_json(f"https://huggingface.co/api/kernels/{kernel_id}")
    if not isinstance(detail, dict):
        raise ValueError(f"{kernel_id}: malformed Hugging Face response")
    if detail.get("sha") != revision:
        raise ValueError(
            f"{kernel_id}: live head {detail.get('sha')} differs from reviewed pin "
            f"{revision}; review and update the pin explicitly"
        )

    tree_url = (
        f"https://huggingface.co/api/kernels/{kernel_id}/tree/{revision}/"
        "build/torch-cpu?recursive=true&expand=false"
    )
    tree = get_json(tree_url)
    if not isinstance(tree, list):
        raise ValueError(f"{kernel_id}: malformed build tree")

    files: list[dict] = []
    for entry in tree:
        if not isinstance(entry, dict) or entry.get("type") != "file":
            continue
        path = entry.get("path")
        if (
            not isinstance(path, str)
            or not path.startswith("build/torch-cpu/")
            or ".." in Path(path).parts
        ):
            raise ValueError(f"{kernel_id}: unsafe build path {path!r}")
        content = get_bytes(
            f"https://huggingface.co/kernels/{kernel_id}/resolve/{revision}/{path}"
        )
        expected_size = entry.get("size")
        if expected_size != len(content):
            raise ValueError(
                f"{kernel_id}:{path}: API size {expected_size} != {len(content)}"
            )
        files.append(
            {
                "path": path,
                "size": len(content),
                "git_oid": entry.get("oid"),
                "sha256": hashlib.sha256(content).hexdigest(),
            }
        )
    files.sort(key=lambda row: row["path"])
    if not files:
        raise ValueError(f"{kernel_id}: no build/torch-cpu files found")

    tree_digest = hashlib.sha256(canonical_json(files)).hexdigest()
    status = kernel["portfolio_status"]
    replacement = "SZLHOLDINGS/szl-kernels" if status == "RETAINED_COMPATIBILITY" else None
    dependencies = kernel.get("dependencies", [])
    optional_dependencies = kernel.get("optional_dependencies", [])
    package = kernel["probe"]["package"]

    return {
        "$schema": "../../schemas/kernel-contract.schema.json",
        "schema_version": "1.0.0",
        "id": kernel_id,
        "title": kernel["title"],
        "summary": kernel["summary"],
        "revision": revision,
        "recorded_at": recorded_at,
        "artifact": {
            "repo_type": "kernel",
            "build": "torch-cpu",
            "tree_url": (
                f"https://huggingface.co/kernels/{kernel_id}/tree/{revision}/"
                "build/torch-cpu"
            ),
            "files": files,
            "tree_digest_sha256": tree_digest,
        },
        "loading": {
            "method": "kernels.get_kernel",
            "trust_remote_code": True,
            "revision_required": True,
            "example": (
                "from kernels import get_kernel\n"
                f'kernel = get_kernel("{kernel_id}", '
                f'revision="{revision}", trust_remote_code=True)'
            ),
        },
        "runtime": {
            "classification": "PYTHON_GOVERNANCE_KERNEL",
            "packages": kernel["packages"],
            "dependencies": dependencies,
            "optional_dependencies": optional_dependencies,
            "python": ">=3.9",
            "artifact_build": "torch-cpu",
            "hf_declared_driver_families": [],
            "measured_compatibility": {
                "status": "MEASURED",
                "receipt": "../../evidence/kernel-selfcheck-20260726.json",
                "environment": "Windows amd64, Python 3.11, CPU execution",
            },
        },
        "probe": kernel["probe"],
        "source_binding": kernel["source_binding"],
        "limitations": kernel["limitations"],
        "deprecation": {
            "status": status,
            "replacement": replacement,
        },
        "generation": {
            "tool": "scripts/snapshot_kernel_contracts.py",
            "mode": "reviewed immutable pin; no automatic revision advancement",
            "primary_package": package,
        },
    }


def write_json(path: Path, value: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(value, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
        newline="\n",
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--output",
        type=Path,
        default=CONTRACTS_DIR,
        help="Contract output directory (defaults to ./contracts)",
    )
    args = parser.parse_args()
    pins = load_pins()
    contracts = [
        snapshot_kernel(kernel, pins["recorded_at"]) for kernel in pins["kernels"]
    ]

    output = args.output.resolve()
    for contract, seed in zip(contracts, pins["kernels"], strict=True):
        write_json(output / seed["slug"] / "contract.json", contract)
    index = {
        "schema_version": "1.0.0",
        "recorded_at": pins["recorded_at"],
        "doctrine": pins["doctrine"],
        "kernels": [
            {
                "id": contract["id"],
                "slug": seed["slug"],
                "title": contract["title"],
                "summary": contract["summary"],
                "revision": contract["revision"],
                "tree_digest_sha256": contract["artifact"]["tree_digest_sha256"],
                "portfolio_status": contract["deprecation"]["status"],
                "source_status": contract["source_binding"]["status"],
                "contract": f"{seed['slug']}/contract.json",
                "limitations": contract["limitations"],
            }
            for contract, seed in zip(contracts, pins["kernels"], strict=True)
        ],
    }
    write_json(output / "index.json", index)
    print(f"wrote {len(contracts)} immutable contracts to {output}")
    return 0


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