scripts/upload_huggingface_surface.py

#!/usr/bin/env python3
"""Upload one prepared Hub surface and record its compare-and-swap commit."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any

COMMIT_PATTERN = re.compile(r"[0-9a-f]{40}")
MARKER_SCHEMA = "noqt.huggingface_sync.v1"
UPLOAD_SCHEMA = "noqt.huggingface_upload.v1"
REMOTE_MARKER = "HUGGINGFACE_SYNC.json"
# One-time first-migration bridge for the known public Space parent that
# predates the source marker.  The bridge is intentionally bound to both the
# exact parent and the independently observed source commit; all other
# missing, malformed, or unreadable markers remain fail-closed.
LEGACY_BOOTSTRAP_PARENT = "c089755e699c75c68e926613d59219bf4feebac5"
LEGACY_BOOTSTRAP_SOURCE = "bf6910dbd83d30a50a486f84ac0fa96a0244e23e"


class UploadError(RuntimeError):
    """Raised when a Hub upload cannot be bound safely."""


def _sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _write_once(path: Path, value: dict[str, Any]) -> None:
    path = Path(path)
    if path.exists() or path.is_symlink():
        raise UploadError(f"output already exists: {path}")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def _is_missing_remote_marker_error(error: BaseException) -> bool:
    """Recognize only the Hub client's missing-entry exception variants."""

    # huggingface_hub 1.7.1 exposes both names from huggingface_hub.errors;
    # older clients called the remote 404 RemoteEntryNotFoundError.  Do not
    # classify transport/auth/permission errors as a missing marker.
    try:
        from huggingface_hub import errors as hub_errors
    except (ImportError, AttributeError):
        return False
    missing_types = tuple(
        error_type
        for name in ("EntryNotFoundError", "RemoteEntryNotFoundError")
        if isinstance(error_type := getattr(hub_errors, name, None), type)
    )
    return bool(missing_types) and isinstance(error, missing_types)


def _remote_source_revision(api: Any, repo_id: str, repo_type: str, parent: str, token: str) -> str:
    try:
        with tempfile.TemporaryDirectory(prefix="hf-sync-parent-marker-") as marker_dir:
            marker_path = Path(
                api.hf_hub_download(
                    repo_id=repo_id,
                    filename=REMOTE_MARKER,
                    repo_type=repo_type,
                    revision=parent,
                    local_dir=marker_dir,
                    token=token,
                )
            )
            if marker_path.is_symlink() or not marker_path.is_file():
                raise UploadError("remote source marker is not a regular file")
            marker = json.loads(marker_path.read_text(encoding="utf-8"))
    except UploadError:
        raise
    except Exception as error:
        if parent == LEGACY_BOOTSTRAP_PARENT and _is_missing_remote_marker_error(error):
            return LEGACY_BOOTSTRAP_SOURCE
        raise UploadError("remote source marker is unreadable; distribution remains INCOMPLETE") from error
    if not isinstance(marker, dict):
        raise UploadError("remote source marker is not an object; distribution remains INCOMPLETE")
    source_revision = marker.get("source_revision")
    if marker.get("schema") != MARKER_SCHEMA or not isinstance(source_revision, str) or not COMMIT_PATTERN.fullmatch(source_revision):
        raise UploadError("remote source marker has no exact source commit; distribution remains INCOMPLETE")
    return source_revision


def _require_source_ancestry(source_root: Path, ancestor: str, descendant: str) -> None:
    source_root = Path(source_root)
    if source_root.is_symlink():
        raise UploadError("source root must not be a symlink")
    try:
        source_root = source_root.resolve(strict=True)
    except OSError as error:
        raise UploadError("source root is unavailable; distribution remains INCOMPLETE") from error
    if not source_root.is_dir():
        raise UploadError("source root is not a directory; distribution remains INCOMPLETE")
    try:
        result = subprocess.run(
            ["git", "-C", str(source_root), "merge-base", "--is-ancestor", ancestor, descendant],
            check=False,
            capture_output=True,
            text=True,
        )
    except OSError as error:
        raise UploadError("cannot verify remote source ancestry; distribution remains INCOMPLETE") from error
    if result.returncode != 0:
        raise UploadError("remote source is not an ancestor of the candidate; distribution remains INCOMPLETE")


def upload_surface(
    *,
    repo_id: str,
    repo_type: str,
    folder: Path,
    revision: str,
    source_revision: str,
    output: Path,
    source_root: Path = Path("."),
) -> dict[str, Any]:
    if repo_type != "space":
        raise UploadError("only the configured Space repository type is supported")
    if not COMMIT_PATTERN.fullmatch(source_revision):
        raise UploadError("source revision must be an exact 40-character commit")
    output = Path(output)
    if output.exists() or output.is_symlink():
        raise UploadError(f"output already exists: {output}")
    folder = Path(folder)
    if folder.is_symlink():
        raise UploadError("upload folder must not be a symlink")
    folder = folder.resolve(strict=True)
    if not folder.is_dir() or folder.is_symlink():
        raise UploadError("upload folder must be a regular directory")
    marker_path = folder / "HUGGINGFACE_SYNC.json"
    manifest_path = folder / "HUGGINGFACE_MANIFEST.json"
    try:
        marker = json.loads(marker_path.read_text(encoding="utf-8"))
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
        raise UploadError("prepared surface metadata is unreadable") from error
    if (
        marker.get("schema") != MARKER_SCHEMA
        or marker.get("source_revision") != source_revision
        or manifest.get("source_revision") != source_revision
        or marker.get("manifest_sha256") != _sha256(manifest_path)
    ):
        raise UploadError("prepared surface is not bound to the requested source revision")
    token = os.environ.get("HF_TOKEN", "")
    if not token:
        raise UploadError("HF_TOKEN is missing; distribution remains INCOMPLETE")
    try:
        from huggingface_hub import HfApi

        api = HfApi(token=token)
        parent = api.repo_info(repo_id=repo_id, repo_type=repo_type, revision=revision).sha
        if not isinstance(parent, str) or not COMMIT_PATTERN.fullmatch(parent):
            raise UploadError("remote parent revision is not an exact commit")
        remote_source_revision = _remote_source_revision(api, repo_id, repo_type, parent, token)
        _require_source_ancestry(source_root, remote_source_revision, source_revision)
        commit = api.upload_folder(
            repo_id=repo_id,
            repo_type=repo_type,
            folder_path=folder,
            path_in_repo=".",
            revision=revision,
            parent_commit=parent,
            commit_message=f"Sync GitHub source {source_revision}",
            commit_description=(
                "Canonical source: "
                f"https://github.com/noqt/Lumi-Eggcracker/commit/{source_revision}"
            ),
            delete_patterns=["*"],
            token=token,
        )
    except UploadError:
        raise
    except Exception as error:  # pragma: no cover - exercised by the remote service
        raise UploadError(f"Hub upload failed ({type(error).__name__}); distribution remains INCOMPLETE") from error
    upload_commit = getattr(commit, "oid", None)
    commit_url = getattr(commit, "commit_url", None)
    if not isinstance(upload_commit, str) or not COMMIT_PATTERN.fullmatch(upload_commit):
        raise UploadError("Hub upload returned no exact commit; distribution remains INCOMPLETE")
    if not isinstance(commit_url, str) or not commit_url.startswith("https://huggingface.co/"):
        raise UploadError("Hub upload returned no canonical commit link")
    result = {
        "schema": UPLOAD_SCHEMA,
        "repository": repo_id,
        "repo_type": repo_type,
        "branch": revision,
        "source_revision": source_revision,
        "source_url": f"https://github.com/noqt/Lumi-Eggcracker/commit/{source_revision}",
        "parent_commit": parent,
        "upload_commit": upload_commit,
        "commit_url": commit_url,
        "manifest_sha256": marker["manifest_sha256"],
    }
    _write_once(Path(output), result)
    return result


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-id", required=True)
    parser.add_argument("--repo-type", required=True)
    parser.add_argument("--folder", required=True, type=Path)
    parser.add_argument("--revision", required=True)
    parser.add_argument("--source-revision", required=True)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--source-root", type=Path, default=Path("."))
    return parser


def main() -> int:
    args = _parser().parse_args()
    try:
        result = upload_surface(
            repo_id=args.repo_id,
            repo_type=args.repo_type,
            folder=args.folder,
            revision=args.revision,
            source_revision=args.source_revision,
            output=args.output,
            source_root=args.source_root,
        )
    except UploadError as error:
        print(f"HUGGINGFACE_SYNC_INCOMPLETE: {error}", file=sys.stderr)
        return 2
    print(json.dumps({key: result[key] for key in ("source_revision", "parent_commit", "upload_commit", "manifest_sha256")}, sort_keys=True))
    return 0


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