scripts/publish_hf.py

#!/usr/bin/env python3
"""Validate and publish reproducible model and Gradio Space folders to Hugging Face."""

from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import random
import socket
import sys
import tempfile
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, TypeVar

ROOT = Path(__file__).resolve().parents[1]
_HUB_MANAGED_FILES = {".gitattributes"}
_TRANSIENT_HTTP_STATUS = {408, 425, 429, 500, 502, 503, 504}
_FAILED_SPACE_STAGES = {
    "BUILD_ERROR",
    "CONFIG_ERROR",
    "DELETING",
    "NO_APP_FILE",
    "PAUSED",
    "RUNTIME_ERROR",
    "STOPPED",
}
_INTERMEDIATE_SPACE_STAGES = {
    "APP_STARTING",
    "BUILDING",
    "RUNNING_APP_STARTING",
    "RUNNING_BUILDING",
}
_T = TypeVar("_T")


class PublishFailure(RuntimeError):
    """A sanitized, actionable publication failure safe to show to a user."""


class SpaceRuntimeFailure(PublishFailure):
    """A Space build/runtime failure carrying only its non-secret stage."""

    def __init__(self, message: str, *, stage: str | None) -> None:
        super().__init__(message)
        self.stage = stage


@dataclass(frozen=True)
class RetryPolicy:
    max_attempts: int = 4
    base_seconds: float = 1.0
    max_seconds: float = 30.0
    request_timeout_seconds: float = 120.0
    sleeper: Callable[[float], None] = time.sleep
    jitter: Callable[[float, float], float] = random.uniform


@dataclass(frozen=True)
class RemoteVerification:
    matches: bool
    paths: frozenset[str]
    reason: str


@dataclass(frozen=True)
class ComponentResult:
    repo_id: str
    repo_type: str
    commit_sha: str
    commit_url: str
    action: str
    verified: bool = True

    def as_dict(self) -> dict[str, Any]:
        return {
            "repo_id": self.repo_id,
            "repo_type": self.repo_type,
            "commit_sha": self.commit_sha,
            "commit_url": self.commit_url,
            "action": self.action,
            "verified": self.verified,
        }


@dataclass(frozen=True)
class SpaceRuntimeEvidence:
    stage: str
    build_transition_observed: bool
    commit_bound: bool = False


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def _validate_inventory(release_dir: Path, manifest: dict[str, Any]) -> None:
    inventory = manifest.get("files")
    if not isinstance(inventory, dict) or not inventory:
        raise SystemExit("release manifest has no file inventory; rebuild the release")
    manifest_path = release_dir / "release_manifest.json"
    if manifest_path.is_symlink() or not manifest_path.is_file():
        raise SystemExit("release manifest must be a regular file")
    symlinks = [path for path in release_dir.rglob("*") if path.is_symlink()]
    if symlinks:
        raise SystemExit("release contents contain a symlink; rebuild")
    expected_paths = set(inventory)
    actual_paths = {
        path.relative_to(release_dir).as_posix()
        for path in release_dir.rglob("*")
        if path.is_file() and path != manifest_path
    }
    if actual_paths != expected_paths:
        raise SystemExit("release contents differ from the hash-bound file inventory; rebuild")
    for relative, evidence in inventory.items():
        path = (release_dir / relative).resolve()
        try:
            path.relative_to(release_dir.resolve())
        except ValueError as exc:
            raise SystemExit(f"release inventory escapes its root: {relative}") from exc
        if path.is_symlink() or not path.is_file() or not isinstance(evidence, dict):
            raise SystemExit(f"invalid release inventory entry: {relative}")
        if evidence.get("bytes") != path.stat().st_size or evidence.get("sha256") != _sha256(path):
            raise SystemExit(f"release file failed integrity validation: {relative}")
    serialized = json.dumps(inventory, sort_keys=True, separators=(",", ":")).encode()
    expected_digest = hashlib.sha256(serialized).hexdigest()
    if manifest.get("release_inventory_sha256") != expected_digest:
        raise SystemExit("release inventory digest is invalid; rebuild")


def _validate_provenance(release_dir: Path) -> None:
    """Check that the portable inner provenance describes the two upload folders."""

    path = release_dir / "model" / "release_provenance.json"
    try:
        provenance = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise SystemExit("release_provenance.json is missing or invalid") from exc
    inventories = provenance.get("files")
    if not isinstance(inventories, dict):
        raise SystemExit("release provenance has no component inventories; rebuild")
    expected_digest = hashlib.sha256(
        json.dumps(inventories, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()
    if provenance.get("source_and_artifact_inventory_sha256") != expected_digest:
        raise SystemExit("release provenance inventory digest is invalid; rebuild")

    for component in ("model", "space"):
        expected = inventories.get(component)
        if not isinstance(expected, dict):
            raise SystemExit(f"release provenance is missing the {component} inventory")
        folder = release_dir / component
        actual: dict[str, dict[str, int | str]] = {}
        for file_path in folder.rglob("*"):
            if not file_path.is_file():
                continue
            relative = file_path.relative_to(folder).as_posix()
            if component == "model" and relative == "release_provenance.json":
                continue
            actual[relative] = {
                "bytes": file_path.stat().st_size,
                "sha256": _sha256(file_path),
            }
        if actual != expected:
            raise SystemExit(f"release provenance disagrees with the {component} folder; rebuild")


def _validated_release_status(release_dir: Path, manifest: dict[str, Any]) -> bool:
    """Derive policy from inventoried files, not mutable outer-manifest text."""

    metadata_path = release_dir / "model" / "model_metadata.json"
    provenance_path = release_dir / "model" / "release_provenance.json"
    try:
        metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
        provenance = json.loads(provenance_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise SystemExit("inventoried release policy files are invalid") from exc
    if not isinstance(metadata, dict) or not isinstance(provenance, dict):
        raise SystemExit("inventoried release policy files must be JSON objects")
    development_value = metadata.get("development_only")
    if not isinstance(development_value, bool):
        raise SystemExit("inventoried metadata development_only must be a boolean")
    development_only = development_value
    if not development_only and str(metadata.get("training_status", "")).lower() != "final":
        raise SystemExit("inventoried metadata has an invalid final-release status")
    if provenance.get("development_only") is not development_only:
        raise SystemExit("release provenance disagrees with model metadata")
    if manifest.get("development_only") is not development_only:
        raise SystemExit("outer release status disagrees with inventoried metadata")
    expected_metrics = "development_metrics.json" if development_only else "test_metrics.json"
    if manifest.get("metrics_file") != expected_metrics:
        raise SystemExit("outer metrics classification disagrees with inventoried metadata")
    expected_scope = "development" if development_only else "official_test"
    if manifest.get("metrics_scope") != expected_scope:
        raise SystemExit("outer metrics scope disagrees with inventoried metadata")
    if manifest.get("has_test_metrics") is not (not development_only):
        raise SystemExit("outer test-metrics flag disagrees with inventoried metadata")
    if not (release_dir / "model" / expected_metrics).is_file():
        raise SystemExit(f"inventoried release is missing {expected_metrics}")
    model_path = (release_dir / "model" / str(manifest.get("model_file", ""))).resolve()
    try:
        model_path.relative_to((release_dir / "model").resolve())
    except ValueError as exc:
        raise SystemExit("outer model_file escapes the model release directory") from exc
    if model_path.is_symlink() or not model_path.is_file():
        raise SystemExit("outer model_file is not an inventoried regular file")
    export_path = release_dir / "model" / "export_manifest.json"
    metrics_path = release_dir / "model" / expected_metrics
    bindings = {
        "model_sha256": _sha256(model_path),
        "metadata_sha256": _sha256(metadata_path),
        "export_manifest_sha256": _sha256(export_path),
        "metrics_sha256": _sha256(metrics_path),
    }
    for name, actual in bindings.items():
        if manifest.get(name) != actual:
            raise SystemExit(f"outer release manifest has invalid {name}")
    return development_only


def _load_env_token(path: Path) -> str | None:
    if not path.is_file():
        return None
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        if key.strip() in {"HF_TOKEN", "hf_token", "HUGGING_FACE_HUB_TOKEN"}:
            token = value.strip().strip("'\"")
            return token or None
    return None


def _http_status(exc: BaseException) -> int | None:
    seen: set[int] = set()
    current: BaseException | None = exc
    while current is not None and id(current) not in seen:
        seen.add(id(current))
        response = getattr(current, "response", None)
        status = getattr(response, "status_code", None)
        if isinstance(status, int):
            return status
        direct_status = getattr(current, "status_code", None)
        if isinstance(direct_status, int):
            return direct_status
        current = current.__cause__ or current.__context__
    return None


def _exception_chain(exc: BaseException) -> list[BaseException]:
    chain: list[BaseException] = []
    seen: set[int] = set()
    current: BaseException | None = exc
    while current is not None and id(current) not in seen:
        seen.add(id(current))
        chain.append(current)
        current = current.__cause__ or current.__context__
    return chain


def _is_transient(exc: BaseException) -> bool:
    status = _http_status(exc)
    if status is not None:
        return status in _TRANSIENT_HTTP_STATUS or 500 <= status <= 599
    for item in _exception_chain(exc):
        if isinstance(item, (ConnectionError, TimeoutError, socket.gaierror)):
            return True
        module = type(item).__module__.split(".", 1)[0]
        name = type(item).__name__.lower()
        if module in {"httpcore", "httpx", "requests", "urllib3"}:
            hierarchy = {base.__name__.lower() for base in type(item).__mro__}
            if hierarchy & {
                "connectionerror",
                "networkerror",
                "protocolerror",
                "timeoutexception",
                "transporterror",
            } or any(
                marker in name
                for marker in (
                    "close",
                    "connect",
                    "network",
                    "protocol",
                    "read",
                    "timeout",
                    "write",
                )
            ):
                return True
    return False


def _failure_kind(exc: BaseException) -> str:
    status = _http_status(exc)
    if status in {401, 403}:
        return f"authentication/authorization rejection (HTTP {status})"
    if status == 402:
        return (
            "Hugging Face plan/hardware entitlement required (HTTP 402); "
            "verify free ZeroGPU eligibility and an available slot for "
            "--space-hardware zero-a10g, otherwise upgrade the account plan"
        )
    if status is not None:
        return f"HTTP {status}"
    chain = _exception_chain(exc)
    if any(isinstance(item, socket.gaierror) for item in chain):
        return "DNS resolution error"
    if any(
        isinstance(item, TimeoutError) or "timeout" in type(item).__name__.lower() for item in chain
    ):
        return "network timeout"
    if any(
        isinstance(item, ConnectionError) or "connect" in type(item).__name__.lower()
        for item in chain
    ):
        return "connection error"
    if any(
        type(item).__module__.split(".", 1)[0] in {"httpcore", "httpx", "requests", "urllib3"}
        and any(
            marker in type(item).__name__.lower()
            for marker in ("close", "network", "protocol", "read", "write")
        )
        for item in chain
    ):
        return "network transport error"
    return type(exc).__name__


def _delay(policy: RetryPolicy, attempt: int) -> float:
    unjittered = min(policy.max_seconds, policy.base_seconds * (2 ** max(0, attempt - 1)))
    return max(0.0, policy.jitter(unjittered * 0.8, unjittered * 1.2))


def _call_with_retry(label: str, operation: Callable[[], _T], policy: RetryPolicy) -> _T:
    for attempt in range(1, policy.max_attempts + 1):
        try:
            return operation()
        except Exception as exc:
            transient = _is_transient(exc)
            if not transient or attempt == policy.max_attempts:
                raise PublishFailure(
                    f"{label} failed after {attempt} attempt(s): {_failure_kind(exc)}"
                ) from None
            delay = _delay(policy, attempt)
            print(
                f"{label}: transient {_failure_kind(exc)}; "
                f"retrying {attempt + 1}/{policy.max_attempts} in {delay:.1f}s",
                file=sys.stderr,
            )
            policy.sleeper(delay)
    raise AssertionError("unreachable")


def _is_not_found(exc: BaseException) -> bool:
    return _http_status(exc) == 404


def _repo_info_or_none(
    api: Any,
    *,
    repo_id: str,
    repo_type: str,
    policy: RetryPolicy,
) -> Any | None:
    for attempt in range(1, policy.max_attempts + 1):
        try:
            return api.repo_info(
                repo_id=repo_id,
                repo_type=repo_type,
                timeout=policy.request_timeout_seconds,
            )
        except Exception as exc:
            if _is_not_found(exc):
                return None
            if not _is_transient(exc) or attempt == policy.max_attempts:
                raise PublishFailure(
                    f"inspect {repo_type} repository failed after {attempt} attempt(s): "
                    f"{_failure_kind(exc)}"
                ) from None
            delay = _delay(policy, attempt)
            print(
                f"inspect {repo_type} repository: transient {_failure_kind(exc)}; "
                f"retrying {attempt + 1}/{policy.max_attempts} in {delay:.1f}s",
                file=sys.stderr,
            )
            policy.sleeper(delay)
    raise AssertionError("unreachable")


def _component_inventory(manifest: Mapping[str, Any], component: str) -> dict[str, dict[str, Any]]:
    prefix = f"{component}/"
    inventory = manifest.get("files", {})
    return {
        relative[len(prefix) :]: evidence
        for relative, evidence in inventory.items()
        if relative.startswith(prefix)
    }


def _verify_remote(
    api: Any,
    snapshot_download: Callable[..., str],
    *,
    repo_id: str,
    repo_type: str,
    revision: str,
    expected: Mapping[str, Mapping[str, Any]],
    token: str,
    policy: RetryPolicy,
) -> RemoteVerification:
    remote_paths = frozenset(
        _call_with_retry(
            f"list pinned files for {repo_type} repository",
            lambda: api.list_repo_files(
                repo_id=repo_id,
                repo_type=repo_type,
                revision=revision,
            ),
            policy,
        )
    )
    expected_paths = set(expected)
    missing = expected_paths - remote_paths
    unexpected = set(remote_paths) - expected_paths - _HUB_MANAGED_FILES
    if missing or unexpected:
        details: list[str] = []
        if missing:
            details.append(f"{len(missing)} missing")
        if unexpected:
            details.append(f"{len(unexpected)} unexpected")
        return RemoteVerification(False, remote_paths, ", ".join(details))

    with tempfile.TemporaryDirectory(prefix="hf-publish-verify-") as cache_dir:
        snapshot_path = Path(
            _call_with_retry(
                f"download pinned snapshot for {repo_type} repository",
                lambda: snapshot_download(
                    repo_id=repo_id,
                    repo_type=repo_type,
                    revision=revision,
                    cache_dir=cache_dir,
                    force_download=True,
                    token=token,
                ),
                policy,
            )
        )
        for relative, evidence in expected.items():
            path = snapshot_path / relative
            if not path.is_file():
                return RemoteVerification(False, remote_paths, f"missing snapshot file {relative}")
            if path.stat().st_size != evidence.get("bytes") or _sha256(path) != evidence.get(
                "sha256"
            ):
                return RemoteVerification(False, remote_paths, f"hash mismatch for {relative}")
    return RemoteVerification(True, remote_paths, "exact package mirror")


def _repo_url(repo_id: str, repo_type: str) -> str:
    if repo_type == "space":
        return f"https://huggingface.co/spaces/{repo_id}"
    return f"https://huggingface.co/{repo_id}"


def _commit_url(repo_id: str, repo_type: str, revision: str) -> str:
    return f"{_repo_url(repo_id, repo_type)}/commit/{revision}"


def _validate_repo_settings(info: Any, *, repo_id: str, repo_type: str, private: bool) -> None:
    actual_id = str(getattr(info, "id", repo_id))
    if actual_id.casefold() != repo_id.casefold():
        raise PublishFailure(f"Hub returned an unexpected {repo_type} repository identity")
    actual_private = getattr(info, "private", None)
    if actual_private is not None and bool(actual_private) is not private:
        visibility = "private" if actual_private else "public"
        requested = "private" if private else "public"
        raise PublishFailure(
            f"existing {repo_type} repository is {visibility}, but {requested} was requested"
        )
    if repo_type == "space":
        sdk = getattr(info, "sdk", None)
        if sdk is not None and str(sdk).casefold() != "gradio":
            raise PublishFailure(f"existing Space SDK is {sdk!r}, not 'gradio'")


def _ensure_repo(
    api: Any,
    *,
    repo_id: str,
    repo_type: str,
    private: bool,
    space_hardware: str | None,
    policy: RetryPolicy,
) -> Any:
    info = _repo_info_or_none(api, repo_id=repo_id, repo_type=repo_type, policy=policy)
    if info is None:
        create_kwargs: dict[str, Any] = {
            "repo_id": repo_id,
            "repo_type": repo_type,
            "private": private,
            "exist_ok": True,
        }
        if repo_type == "space":
            create_kwargs["space_sdk"] = "gradio"
            if space_hardware is not None:
                create_kwargs["space_hardware"] = space_hardware
        _call_with_retry(
            f"create {repo_type} repository",
            lambda: api.create_repo(**create_kwargs),
            policy,
        )
        info = _call_with_retry(
            f"inspect created {repo_type} repository",
            lambda: api.repo_info(
                repo_id=repo_id,
                repo_type=repo_type,
                timeout=policy.request_timeout_seconds,
            ),
            policy,
        )
    _validate_repo_settings(info, repo_id=repo_id, repo_type=repo_type, private=private)
    auth_check = getattr(api, "auth_check", None)
    if callable(auth_check):
        _call_with_retry(
            f"verify write access for {repo_type} repository",
            lambda: auth_check(repo_id=repo_id, repo_type=repo_type, write=True),
            policy,
        )
    return info


def _current_verification(
    api: Any,
    snapshot_download: Callable[..., str],
    *,
    repo_id: str,
    repo_type: str,
    expected: Mapping[str, Mapping[str, Any]],
    token: str,
    policy: RetryPolicy,
) -> tuple[Any, RemoteVerification]:
    for stability_attempt in range(1, policy.max_attempts + 1):
        before = _call_with_retry(
            f"inspect current {repo_type} repository head",
            lambda: api.repo_info(
                repo_id=repo_id,
                repo_type=repo_type,
                timeout=policy.request_timeout_seconds,
            ),
            policy,
        )
        revision = str(getattr(before, "sha", "") or "")
        if revision:
            verification = _verify_remote(
                api,
                snapshot_download,
                repo_id=repo_id,
                repo_type=repo_type,
                revision=revision,
                expected=expected,
                token=token,
                policy=policy,
            )
        else:
            verification = RemoteVerification(False, frozenset(), "repository has no commit")
        after = _call_with_retry(
            f"recheck current {repo_type} repository head",
            lambda: api.repo_info(
                repo_id=repo_id,
                repo_type=repo_type,
                timeout=policy.request_timeout_seconds,
            ),
            policy,
        )
        after_revision = str(getattr(after, "sha", "") or "")
        if after_revision == revision:
            return after, verification
        if stability_attempt == policy.max_attempts:
            raise PublishFailure(
                f"{repo_type} repository head kept changing during pinned verification"
            )
        print(
            f"{repo_type} repository head changed during pinned verification; "
            f"retrying {stability_attempt + 1}/{policy.max_attempts}",
            file=sys.stderr,
        )
    raise AssertionError("unreachable")


def _publish_component(
    api: Any,
    snapshot_download: Callable[..., str],
    *,
    folder: Path,
    expected: Mapping[str, Mapping[str, Any]],
    repo_id: str,
    repo_type: str,
    private: bool,
    token: str,
    replace_existing: bool,
    release_kind: str,
    inventory_digest: str,
    policy: RetryPolicy,
    space_hardware: str | None = None,
) -> ComponentResult:
    _ensure_repo(
        api,
        repo_id=repo_id,
        repo_type=repo_type,
        private=private,
        space_hardware=space_hardware if repo_type == "space" else None,
        policy=policy,
    )
    info, initial = _current_verification(
        api,
        snapshot_download,
        repo_id=repo_id,
        repo_type=repo_type,
        expected=expected,
        token=token,
        policy=policy,
    )
    base_sha = str(getattr(info, "sha", "") or "")
    if initial.matches:
        return ComponentResult(
            repo_id=repo_id,
            repo_type=repo_type,
            commit_sha=base_sha,
            commit_url=_commit_url(repo_id, repo_type, base_sha),
            action="already_current",
        )

    meaningful_existing = set(initial.paths) - _HUB_MANAGED_FILES
    if meaningful_existing and not replace_existing:
        raise PublishFailure(
            f"refusing to replace divergent existing {repo_type} repository {repo_id!r} "
            f"({len(meaningful_existing)} remote path(s)); inspect it and pass "
            "--replace-existing explicitly"
        )

    message_verb = "Deploy" if repo_type == "space" else "Publish"
    upload_kwargs = {
        "repo_id": repo_id,
        "repo_type": repo_type,
        "folder_path": str(folder),
        "commit_message": f"{message_verb} Tiny Hinglish Turn Detector {release_kind}",
        "commit_description": f"release inventory sha256: {inventory_digest}",
        "delete_patterns": "*",
        "parent_commit": base_sha or None,
    }
    for attempt in range(1, policy.max_attempts + 1):
        try:
            commit = api.upload_folder(**upload_kwargs)
        except Exception as exc:
            current_info, current = _current_verification(
                api,
                snapshot_download,
                repo_id=repo_id,
                repo_type=repo_type,
                expected=expected,
                token=token,
                policy=policy,
            )
            current_sha = str(getattr(current_info, "sha", "") or "")
            if current.matches:
                return ComponentResult(
                    repo_id=repo_id,
                    repo_type=repo_type,
                    commit_sha=current_sha,
                    commit_url=_commit_url(repo_id, repo_type, current_sha),
                    action="reconciled_after_error",
                )
            if current_sha != base_sha:
                raise PublishFailure(
                    f"{repo_type} repository head changed during upload and is not an exact mirror; "
                    "refusing to overwrite the concurrent commit"
                ) from None
            if not _is_transient(exc) or attempt == policy.max_attempts:
                raise PublishFailure(
                    f"upload {repo_type} repository failed after {attempt} attempt(s): "
                    f"{_failure_kind(exc)}; remote head was reconciled and remains unchanged"
                ) from None
            delay = _delay(policy, attempt)
            print(
                f"upload {repo_type} repository: transient {_failure_kind(exc)}; "
                f"remote head unchanged, retrying {attempt + 1}/{policy.max_attempts} "
                f"in {delay:.1f}s",
                file=sys.stderr,
            )
            policy.sleeper(delay)
            continue

        returned_commit_sha = str(getattr(commit, "oid", "") or "")
        if not returned_commit_sha:
            raise PublishFailure(
                f"Hub returned no commit SHA after uploading {repo_type} repository"
            )
        current_info, current = _current_verification(
            api,
            snapshot_download,
            repo_id=repo_id,
            repo_type=repo_type,
            expected=expected,
            token=token,
            policy=policy,
        )
        current_sha = str(getattr(current_info, "sha", "") or "")
        if current_sha != returned_commit_sha and not current.matches:
            raise PublishFailure(
                f"{repo_type} repository head changed after upload and is not an exact mirror"
            )
        if not current.matches:
            raise PublishFailure(
                f"uploaded {repo_type} repository failed pinned remote verification: {current.reason}"
            )
        commit_sha = current_sha
        if commit_sha == returned_commit_sha:
            commit_url = str(getattr(commit, "commit_url", "") or "") or _commit_url(
                repo_id, repo_type, commit_sha
            )
        else:
            commit_url = _commit_url(repo_id, repo_type, commit_sha)
        return ComponentResult(
            repo_id=repo_id,
            repo_type=repo_type,
            commit_sha=commit_sha,
            commit_url=commit_url,
            action="uploaded",
        )
    raise AssertionError("unreachable")


def _space_stage(runtime: Any) -> str:
    stage = getattr(runtime, "stage", runtime)
    stage = getattr(stage, "value", stage)
    return str(stage).upper()


def _wait_for_space(
    api: Any,
    *,
    repo_id: str,
    timeout_seconds: float,
    poll_seconds: float,
    policy: RetryPolicy,
    require_build_transition: bool,
    monotonic: Callable[[], float] = time.monotonic,
) -> SpaceRuntimeEvidence:
    deadline = monotonic() + timeout_seconds
    last_stage: str | None = None
    transition_observed = False
    while True:
        runtime = _call_with_retry(
            "inspect Space runtime",
            lambda: api.get_space_runtime(repo_id=repo_id),
            policy,
        )
        last_stage = _space_stage(runtime)
        if last_stage in _FAILED_SPACE_STAGES:
            raise SpaceRuntimeFailure(
                f"Space did not become healthy; terminal runtime stage is {last_stage}",
                stage=last_stage,
            )
        if last_stage in _INTERMEDIATE_SPACE_STAGES:
            transition_observed = True
        if last_stage == "RUNNING" and (transition_observed or not require_build_transition):
            return SpaceRuntimeEvidence(
                stage=last_stage,
                build_transition_observed=transition_observed,
            )
        remaining = deadline - monotonic()
        if remaining <= 0:
            transition_note = (
                "no new build/start transition was observed"
                if require_build_transition and not transition_observed
                else f"last stage: {last_stage}"
            )
            raise SpaceRuntimeFailure(
                f"Space content is verified, but runtime did not reach RUNNING within "
                f"{timeout_seconds:g}s ({transition_note})",
                stage=last_stage,
            )
        policy.sleeper(min(poll_seconds, remaining))


def _utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _write_receipt(path: Path, receipt: Mapping[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
    )
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            json.dump(receipt, handle, indent=2, sort_keys=True)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        temporary.chmod(0o600)
        os.replace(temporary, path)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise


def _receipt_path(raw: str, release_dir: Path) -> Path:
    path = Path(raw)
    if not path.is_absolute():
        path = ROOT / path
    path = path.resolve()
    try:
        path.relative_to(release_dir.resolve())
    except ValueError:
        return path
    raise SystemExit("publish receipt must be outside the inventoried release directory")


def _execute_publish(
    args: argparse.Namespace,
    *,
    release_dir: Path,
    manifest: dict[str, Any],
    development_only: bool,
    token: str,
    api: Any,
    snapshot_download: Callable[..., str],
    policy: RetryPolicy,
) -> dict[str, Any]:
    receipt_path = _receipt_path(args.receipt, release_dir)
    receipt: dict[str, Any] = {
        "format_version": 1,
        "release_inventory_sha256": manifest["release_inventory_sha256"],
        "development_only": development_only,
        "requested_owner": args.username,
        "targets": {
            "model": f"{args.username}/{args.model_name}",
            "space": f"{args.username}/{args.space_name}",
        },
        "space_creation": {
            "sdk": "gradio",
            "hardware_if_created": args.space_hardware,
        },
        "status": "starting",
        "updated_at": _utc_now(),
    }
    _write_receipt(receipt_path, receipt)
    try:
        identity = _call_with_retry("authenticate with Hugging Face", api.whoami, policy)
        authenticated_name = str(identity.get("name", "")) if isinstance(identity, dict) else ""
        if authenticated_name.casefold() != args.username.casefold():
            raise PublishFailure(
                f"authenticated Hugging Face user {authenticated_name!r} does not match "
                f"requested owner {args.username!r}"
            )
    except PublishFailure as exc:
        receipt["status"] = "failed"
        receipt["failure_stage"] = "authentication"
        receipt["failure"] = str(exc)
        receipt["updated_at"] = _utc_now()
        _write_receipt(receipt_path, receipt)
        raise
    except Exception as exc:
        failure = PublishFailure(f"authentication failed unexpectedly: {_failure_kind(exc)}")
        receipt["status"] = "failed"
        receipt["failure_stage"] = "authentication"
        receipt["failure"] = str(failure)
        receipt["updated_at"] = _utc_now()
        _write_receipt(receipt_path, receipt)
        raise failure from None

    model_id = f"{authenticated_name}/{args.model_name}"
    space_id = f"{authenticated_name}/{args.space_name}"
    receipt["authenticated_owner"] = authenticated_name
    receipt["targets"] = {"model": model_id, "space": space_id}
    receipt["status"] = "authenticated"
    receipt["updated_at"] = _utc_now()
    _write_receipt(receipt_path, receipt)
    release_kind = "development preview" if development_only else "final release"

    stage = "model"
    try:
        _validate_inventory(release_dir, manifest)
        model = _publish_component(
            api,
            snapshot_download,
            folder=release_dir / "model",
            expected=_component_inventory(manifest, "model"),
            repo_id=model_id,
            repo_type="model",
            private=args.private,
            token=token,
            replace_existing=args.replace_existing,
            release_kind=release_kind,
            inventory_digest=manifest["release_inventory_sha256"],
            policy=policy,
        )
        receipt["model"] = model.as_dict()
        receipt["status"] = "model_verified"
        receipt["updated_at"] = _utc_now()
        _write_receipt(receipt_path, receipt)

        stage = "space"
        _validate_inventory(release_dir, manifest)
        space = _publish_component(
            api,
            snapshot_download,
            folder=release_dir / "space",
            expected=_component_inventory(manifest, "space"),
            repo_id=space_id,
            repo_type="space",
            private=args.private,
            token=token,
            replace_existing=args.replace_existing,
            release_kind=release_kind,
            inventory_digest=manifest["release_inventory_sha256"],
            policy=policy,
            space_hardware=args.space_hardware,
        )
        receipt["space"] = space.as_dict()
        receipt["status"] = "space_content_verified"
        receipt["updated_at"] = _utc_now()
        _write_receipt(receipt_path, receipt)

        stage = "space_runtime"
        runtime_evidence = _wait_for_space(
            api,
            repo_id=space_id,
            timeout_seconds=args.space_timeout_seconds,
            poll_seconds=args.space_poll_seconds,
            policy=policy,
            require_build_transition=space.action != "already_current",
        )
        # A concurrent push after the build can make RUNNING refer to content other than this release.
        current_info, current = _current_verification(
            api,
            snapshot_download,
            repo_id=space_id,
            repo_type="space",
            expected=_component_inventory(manifest, "space"),
            token=token,
            policy=policy,
        )
        if not current.matches:
            raise PublishFailure("Space head changed before runtime verification completed")
        current_sha = str(getattr(current_info, "sha", "") or "")
        receipt["space"]["commit_sha"] = current_sha
        receipt["space"]["commit_url"] = _commit_url(space_id, "space", current_sha)
        receipt["space"]["runtime_stage"] = runtime_evidence.stage
        receipt["space"]["build_transition_observed"] = runtime_evidence.build_transition_observed
        receipt["space"]["runtime_commit_bound"] = runtime_evidence.commit_bound
        receipt["status"] = "content_verified_runtime_running"
        receipt["updated_at"] = _utc_now()
        _write_receipt(receipt_path, receipt)
        return receipt
    except (PublishFailure, SystemExit) as exc:
        receipt["status"] = "failed"
        receipt["failure_stage"] = stage
        receipt["failure"] = str(exc)
        if isinstance(exc, SpaceRuntimeFailure):
            receipt.setdefault("space", {})["runtime_stage"] = exc.stage
            receipt["space"]["runtime_commit_bound"] = False
        receipt["updated_at"] = _utc_now()
        _write_receipt(receipt_path, receipt)
        raise
    except Exception as exc:
        failure = PublishFailure(f"{stage} failed unexpectedly: {_failure_kind(exc)}")
        receipt["status"] = "failed"
        receipt["failure_stage"] = stage
        receipt["failure"] = str(failure)
        receipt["updated_at"] = _utc_now()
        _write_receipt(receipt_path, receipt)
        raise failure from None


def _positive_int(raw: str) -> int:
    value = int(raw)
    if value < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return value


def _nonnegative_float(raw: str) -> float:
    value = float(raw)
    if not math.isfinite(value) or value < 0:
        raise argparse.ArgumentTypeError("must be finite and non-negative")
    return value


def _positive_float(raw: str) -> float:
    value = float(raw)
    if not math.isfinite(value) or value <= 0:
        raise argparse.ArgumentTypeError("must be finite and greater than zero")
    return value


def _bounded_http_client_factory(
    httpx_module: Any,
    *,
    connect_seconds: float,
    read_seconds: float,
    write_seconds: float,
    pool_seconds: float,
) -> Callable[[], Any]:
    timeout = httpx_module.Timeout(
        connect=connect_seconds,
        read=read_seconds,
        write=write_seconds,
        pool=pool_seconds,
    )

    def factory() -> Any:
        return httpx_module.Client(follow_redirects=True, timeout=timeout)

    return factory


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--username", default="suvradeepp")
    parser.add_argument("--model-name", default="tiny-hinglish-turn-detector")
    parser.add_argument("--space-name", default="tiny-hinglish-turn-detector")
    parser.add_argument(
        "--space-hardware",
        default="zero-a10g",
        help=(
            "Hardware requested only when creating a new Space (default: zero-a10g, "
            "the free-account-compatible ZeroGPU tier)"
        ),
    )
    parser.add_argument("--release-dir", default="release")
    parser.add_argument("--receipt", default="reports/hf_publish_receipt.json")
    parser.add_argument("--private", action="store_true")
    parser.add_argument(
        "--replace-existing",
        action="store_true",
        help="Explicitly allow replacing a non-empty divergent existing target repository",
    )
    parser.add_argument("--max-attempts", type=_positive_int, default=4)
    parser.add_argument("--retry-base-seconds", type=_nonnegative_float, default=1.0)
    parser.add_argument("--space-timeout-seconds", type=_nonnegative_float, default=900.0)
    parser.add_argument("--space-poll-seconds", type=_positive_float, default=10.0)
    parser.add_argument("--http-connect-timeout-seconds", type=_positive_float, default=10.0)
    parser.add_argument("--http-read-timeout-seconds", type=_positive_float, default=120.0)
    parser.add_argument("--http-write-timeout-seconds", type=_positive_float, default=120.0)
    parser.add_argument("--http-pool-timeout-seconds", type=_positive_float, default=10.0)
    parser.add_argument(
        "--allow-development-release",
        action="store_true",
        help="Allow publishing a manifest explicitly marked development_only",
    )
    parser.add_argument(
        "--execute",
        action="store_true",
        help="Perform external repository creation/uploads; otherwise only validate offline",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    release_dir = Path(args.release_dir)
    if not release_dir.is_absolute():
        release_dir = ROOT / release_dir
    model_folder = release_dir / "model"
    space_folder = release_dir / "space"
    for path in (model_folder, space_folder, release_dir / "release_manifest.json"):
        if not path.exists():
            raise SystemExit(f"missing release artifact: {path}; run scripts/build_release.py")
    try:
        manifest = json.loads((release_dir / "release_manifest.json").read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError) as exc:
        raise SystemExit("release_manifest.json is missing or invalid") from exc
    _validate_inventory(release_dir, manifest)
    _validate_provenance(release_dir)
    development_only = _validated_release_status(release_dir, manifest)
    if development_only and not args.allow_development_release:
        raise SystemExit(
            "refusing to publish a development-only artifact without --allow-development-release"
        )
    if not args.execute:
        release_kind = "development" if development_only else "final"
        print(
            f"validated {release_kind} release folders for {args.username} entirely offline; "
            "pass --execute to publish"
        )
        return 0

    token = (
        os.environ.get("HF_TOKEN")
        or os.environ.get("HUGGING_FACE_HUB_TOKEN")
        or _load_env_token(ROOT / ".env")
    )
    if not token:
        raise SystemExit("HF token missing; set HF_TOKEN or keep it in ignored .env")
    try:
        import httpx
        from huggingface_hub import HfApi, close_session, set_client_factory, snapshot_download
    except ImportError as exc:
        raise SystemExit(
            "install the exact publishing dependencies: python -m pip install -r "
            "requirements-publish.txt"
        ) from exc

    set_client_factory(
        _bounded_http_client_factory(
            httpx,
            connect_seconds=args.http_connect_timeout_seconds,
            read_seconds=args.http_read_timeout_seconds,
            write_seconds=args.http_write_timeout_seconds,
            pool_seconds=args.http_pool_timeout_seconds,
        )
    )
    api = HfApi(token=token)
    policy = RetryPolicy(
        max_attempts=args.max_attempts,
        base_seconds=args.retry_base_seconds,
        request_timeout_seconds=args.http_read_timeout_seconds,
    )
    try:
        receipt = _execute_publish(
            args,
            release_dir=release_dir,
            manifest=manifest,
            development_only=development_only,
            token=token,
            api=api,
            snapshot_download=snapshot_download,
            policy=policy,
        )
    except PublishFailure as exc:
        raise SystemExit(str(exc)) from None
    finally:
        close_session()
    print(f"model: {_repo_url(receipt['targets']['model'], 'model')}")
    print(f"model commit: {receipt['model']['commit_url']}")
    print(f"space: {_repo_url(receipt['targets']['space'], 'space')}")
    print(f"space commit: {receipt['space']['commit_url']}")
    print(f"publication receipt: {_receipt_path(args.receipt, release_dir)}")
    return 0


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