scripts/build_release.py

#!/usr/bin/env python3
"""Build deterministic Hugging Face model and Space upload directories."""

from __future__ import annotations

import argparse
import ast
import errno
import hashlib
import json
import math
import os
import shutil
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SOURCE_ROOT = ROOT / "src"
if str(SOURCE_ROOT) not in sys.path:
    sys.path.insert(0, str(SOURCE_ROOT))

_SPACE_CARD_COLORS = frozenset(
    {"red", "yellow", "green", "blue", "indigo", "purple", "pink", "gray"}
)
_SPACE_SHORT_DESCRIPTION_MAX_LENGTH = 60
_SPACE_RUNTIME_REQUIREMENTS = frozenset(
    {
        "numpy==2.3.5",
        "onnxruntime==1.26.0",
        "soundfile==0.14.0",
        "spaces==0.51.1",
    }
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--model", default="artifacts/final/model.onnx")
    parser.add_argument("--metadata", default="artifacts/final/model_metadata.json")
    parser.add_argument("--metrics", default="artifacts/final/test_metrics.json")
    parser.add_argument(
        "--export-manifest",
        help="export_manifest.json (default: next to --model)",
    )
    parser.add_argument("--output", default="release")
    parser.add_argument(
        "--allow-development-artifact",
        action="store_true",
        help="Package a non-final artifact, conspicuously marked as development-only",
    )
    parser.add_argument(
        "--frozen-manifest",
        help="required hash-bound freeze manifest for a final official-test release",
    )
    parser.add_argument(
        "--include-synthetic-controller-replay",
        action="store_true",
        help="package the canonical hash-bound synthetic replay fixture and decisions",
    )
    return parser.parse_args()


def _resolve(value: str) -> Path:
    path = Path(value)
    return (path if path.is_absolute() else ROOT / path).resolve()


def _copy_file(source: Path, destination: Path) -> None:
    if not source.is_file():
        raise FileNotFoundError(source)
    if source.is_symlink():
        raise ValueError(f"refusing to package symlink: {source}")
    destination.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(source, destination)


def _space_card_frontmatter(path: Path) -> dict[str, str]:
    """Parse the scalar fields used by the packaged Hugging Face Space card."""

    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except OSError as exc:
        raise SystemExit(f"Space README is unreadable: {path}") from exc
    if not lines or lines[0].strip() != "---":
        raise SystemExit("Space README must start with YAML frontmatter")
    try:
        closing_index = next(
            index for index, line in enumerate(lines[1:], start=1) if line.strip() == "---"
        )
    except StopIteration as exc:
        raise SystemExit("Space README frontmatter is not closed") from exc

    metadata: dict[str, str] = {}
    for line in lines[1:closing_index]:
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            continue
        if ":" not in stripped:
            raise SystemExit("Space README frontmatter must contain scalar key/value pairs")
        key, value = stripped.split(":", 1)
        key = key.strip()
        value = value.strip().strip("'\"")
        if not key or key in metadata:
            raise SystemExit(f"Space README has an invalid or duplicate metadata key: {key!r}")
        metadata[key] = value
    return metadata


def _validate_space_card(path: Path) -> None:
    metadata = _space_card_frontmatter(path)
    if metadata.get("sdk") != "gradio":
        raise SystemExit("Space README sdk must be 'gradio'")
    if metadata.get("app_file") != "app.py":
        raise SystemExit("Space README app_file must be 'app.py'")
    if metadata.get("python_version") != "3.12":
        raise SystemExit("Space README python_version must match the ZeroGPU Python 3.12 runtime")
    for field in ("colorFrom", "colorTo"):
        color = metadata.get(field)
        if color not in _SPACE_CARD_COLORS:
            allowed = ", ".join(sorted(_SPACE_CARD_COLORS))
            raise SystemExit(f"Space README {field} must be one of: {allowed}")
    short_description = metadata.get("short_description", "")
    if not short_description:
        raise SystemExit("Space README short_description must be non-empty")
    if len(short_description) > _SPACE_SHORT_DESCRIPTION_MAX_LENGTH:
        raise SystemExit(
            "Space README short_description must be at most "
            f"{_SPACE_SHORT_DESCRIPTION_MAX_LENGTH} characters"
        )


def _validate_space_requirements(path: Path) -> None:
    try:
        requirements = frozenset(
            line.strip()
            for line in path.read_text(encoding="utf-8").splitlines()
            if line.strip() and not line.lstrip().startswith("#")
        )
    except OSError as exc:
        raise SystemExit(f"Space requirements are unreadable: {path}") from exc
    if requirements != _SPACE_RUNTIME_REQUIREMENTS:
        raise SystemExit(
            "Space requirements must use the reviewed CPython-3.12-compatible runtime pins"
        )


def _validate_space_app(path: Path) -> None:
    """Require ZeroGPU to wrap the real Gradio inference callback."""

    try:
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
    except (OSError, SyntaxError) as exc:
        raise SystemExit(f"Space app is unreadable or invalid Python: {path}") from exc

    imports_spaces = any(
        isinstance(node, ast.Import)
        and any(alias.name == "spaces" and alias.asname in {None, "spaces"} for alias in node.names)
        for node in ast.walk(tree)
    )
    if not imports_spaces:
        raise SystemExit("Space app must import spaces for ZeroGPU")

    analyze_function = next(
        (
            node
            for node in tree.body
            if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef)
            and node.name == "analyze_turn"
        ),
        None,
    )
    if analyze_function is None:
        raise SystemExit("Space app must define the analyze_turn callback")

    def is_spaces_gpu(decorator: ast.expr) -> bool:
        target = decorator.func if isinstance(decorator, ast.Call) else decorator
        return (
            isinstance(target, ast.Attribute)
            and target.attr == "GPU"
            and isinstance(target.value, ast.Name)
            and target.value.id == "spaces"
        )

    if not any(is_spaces_gpu(item) for item in analyze_function.decorator_list):
        raise SystemExit("Space analyze_turn callback must use @spaces.GPU")

    callback_is_bound = any(
        isinstance(node, ast.Call)
        and isinstance(node.func, ast.Attribute)
        and node.func.attr == "click"
        and any(
            keyword.arg == "fn"
            and isinstance(keyword.value, ast.Name)
            and keyword.value.id == "analyze_turn"
            for keyword in node.keywords
        )
        for node in ast.walk(tree)
    )
    if not callback_is_bound:
        raise SystemExit("Space Analyze button must bind the decorated analyze_turn callback")


def _copy_runtime(destination: Path) -> None:
    package_root = destination / "src" / "turn_detection"
    package_root.mkdir(parents=True, exist_ok=True)
    _copy_file(ROOT / "src/turn_detection/__init__.py", package_root / "__init__.py")
    shutil.copytree(
        ROOT / "src/turn_detection/runtime",
        package_root / "runtime",
        dirs_exist_ok=True,
        ignore=shutil.ignore_patterns("__pycache__", "*.pyc"),
    )


def _reject_symlinks(source: Path) -> None:
    symlinks = [path for path in source.rglob("*") if path.is_symlink()]
    if symlinks:
        raise ValueError(f"refusing to package symlink: {symlinks[0]}")


def _copy_project_source(destination: Path, *, include_synthetic_replay: bool) -> None:
    """Copy the reviewable training/inference source without local caches or data."""

    _reject_symlinks(ROOT / "src" / "turn_detection")
    shutil.copytree(
        ROOT / "src" / "turn_detection",
        destination / "src" / "turn_detection",
        dirs_exist_ok=True,
        ignore=shutil.ignore_patterns("__pycache__", "*.pyc"),
    )
    for script in sorted((ROOT / "scripts").iterdir()):
        if script.is_file() and script.suffix in {".py", ".sh"}:
            _copy_file(script, destination / "scripts" / script.name)
    for name in (
        ".env.example",
        "app.py",
        "pyproject.toml",
        "Makefile",
        "REPORT.md",
        "DATA_CARD.md",
        "MODEL_CARD.md",
        "NOTICE",
    ):
        source = ROOT / name
        if source.is_file():
            _copy_file(source, destination / name)
    for source in sorted(ROOT.glob("requirements-*.txt")):
        _copy_file(source, destination / source.name)
    _copy_file(
        ROOT / "space" / "requirements.txt",
        destination / "space" / "requirements.txt",
    )
    _copy_file(ROOT / "README.md", destination / "SOURCE_README.md")
    if (ROOT / "docs").is_dir():
        _reject_symlinks(ROOT / "docs")
        shutil.copytree(
            ROOT / "docs",
            destination / "docs",
            dirs_exist_ok=True,
            ignore=shutil.ignore_patterns("__pycache__", "*.pyc"),
        )
    if (ROOT / "deployment").is_dir():
        _reject_symlinks(ROOT / "deployment")
        shutil.copytree(
            ROOT / "deployment",
            destination / "deployment",
            dirs_exist_ok=True,
            ignore=shutil.ignore_patterns("__pycache__", "*.pyc"),
        )
    safe_reports = {
        "dataset_snapshot_status.json",
        "environment_snapshot.json",
        "partial_baseline_metrics.json",
        "partial_iid_split.json",
        "partial_shard_audit.json",
        "partial_source_holdout_baseline_metrics.json",
        "partial_source_holdout_comparison.json",
        "partial_source_holdout_split.json",
        "partial_source_holdout_tinytcn_metrics.json",
        "partial_tinytcn_benchmarks.json",
        "partial_tinytcn_comparison.json",
        "partial_tinytcn_e2e_benchmark.json",
        "partial_tinytcn_failures.json",
        "partial_tinytcn_metrics.json",
        "partial_tinytcn_onnx_benchmark.json",
        "partial_tinytcn_pytorch_benchmark.json",
        "partial_tinytcn_silence_sensitivity.json",
    }
    for name in sorted(safe_reports):
        source = ROOT / "reports" / name
        if source.is_file():
            _copy_file(source, destination / "reports" / name)
    for name in (
        "assignments.jsonl",
        "assignments.summary.json",
    ):
        source = ROOT / "data" / "collection" / name
        if source.is_file():
            _copy_file(source, destination / "data" / "collection" / name)
    if include_synthetic_replay:
        for source, relative in (
            (
                ROOT / "data" / "collection" / "controller_replay_fixture.jsonl",
                Path("data/collection/controller_replay_fixture.jsonl"),
            ),
            (
                ROOT / "reports" / "controller_replay_integration.jsonl",
                Path("reports/controller_replay_integration.jsonl"),
            ),
            (
                ROOT / "reports" / "controller_replay_integration.summary.json",
                Path("reports/controller_replay_integration.summary.json"),
            ),
        ):
            _copy_file(source, destination / relative)
    for source in sorted((ROOT / "tests").glob("test_*.py")):
        _copy_file(source, destination / "tests" / source.name)


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


def _load_json(path: Path) -> dict:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise ValueError(f"invalid JSON: {path}") from exc
    if not isinstance(payload, dict):
        raise ValueError(f"expected JSON object: {path}")
    return payload


def _same_value(left: object, right: object) -> bool:
    if isinstance(left, int | float) and isinstance(right, int | float):
        return math.isclose(float(left), float(right), rel_tol=0.0, abs_tol=1e-9)
    return left == right


def _is_hex_sha256(value: object) -> bool:
    return (
        isinstance(value, str)
        and len(value) == 64
        and all(character in "0123456789abcdef" for character in value.lower())
    )


def _deployment_source_paths() -> list[Path]:
    paths = [
        *sorted((ROOT / "src" / "turn_detection").rglob("*.py")),
        *sorted((ROOT / "scripts").glob("*.py")),
        *sorted((ROOT / "scripts").glob("*.sh")),
        *(
            path
            for path in sorted((ROOT / "deployment").rglob("*"))
            if path.is_file() and "__pycache__" not in path.parts and path.suffix != ".pyc"
        ),
        ROOT / "app.py",
        ROOT / "pyproject.toml",
        ROOT / "space" / "requirements.txt",
        *sorted(ROOT.glob("requirements-*.txt")),
    ]
    return sorted(set(paths), key=lambda path: path.relative_to(ROOT).as_posix())


def _validate_local_evidence(evidence: object, label: str) -> None:
    if not isinstance(evidence, dict) or not isinstance(evidence.get("path"), str):
        raise SystemExit(f"export manifest is missing hash-bound {label}")
    unresolved = ROOT / evidence["path"]
    if unresolved.is_symlink() or not unresolved.is_file():
        raise SystemExit(f"bound {label} is missing or not a regular file: {evidence['path']}")
    path = unresolved.resolve()
    try:
        path.relative_to(ROOT)
    except ValueError as exc:
        raise SystemExit(f"bound {label} escapes project root") from exc
    if evidence.get("bytes") != path.stat().st_size or evidence.get("sha256") != _sha256(path):
        raise SystemExit(f"bound {label} changed after export")


def _validate_source_inventory(export_manifest: dict) -> None:
    evidence_items = export_manifest.get("source_files")
    if not isinstance(evidence_items, list) or not evidence_items:
        raise SystemExit("export manifest has no hash-bound executable source inventory")
    expected_paths = {path.relative_to(ROOT).as_posix() for path in _deployment_source_paths()}
    observed_paths: list[str] = []
    for index, evidence in enumerate(evidence_items):
        if not isinstance(evidence, dict) or not isinstance(evidence.get("path"), str):
            raise SystemExit(f"invalid source inventory entry {index}")
        observed_paths.append(evidence["path"])
        _validate_local_evidence(evidence, f"source file {evidence['path']}")
    if len(observed_paths) != len(set(observed_paths)):
        raise SystemExit("export source inventory contains duplicate paths")
    if set(observed_paths) != expected_paths:
        raise SystemExit("executable source surface changed after export; export again")
    serialized = json.dumps(evidence_items, sort_keys=True, separators=(",", ":")).encode("utf-8")
    if export_manifest.get("source_inventory_sha256") != hashlib.sha256(serialized).hexdigest():
        raise SystemExit("export source inventory digest is invalid")


def _validate_onnx_runtime(model: Path, metadata: dict, export_manifest: dict) -> None:
    """Load and execute the exact graph instead of trusting self-reported parity."""

    try:
        import numpy as np
        import onnxruntime as ort
    except ImportError as exc:
        raise SystemExit(
            "release validation requires numpy and onnxruntime; use the export environment"
        ) from exc
    try:
        options = ort.SessionOptions()
        options.intra_op_num_threads = 1
        options.inter_op_num_threads = 1
        session = ort.InferenceSession(
            str(model), sess_options=options, providers=["CPUExecutionProvider"]
        )
    except Exception as exc:
        raise SystemExit(f"released ONNX graph cannot be loaded: {exc}") from exc

    input_names = [item.name for item in session.get_inputs()]
    output_names = [item.name for item in session.get_outputs()]
    if input_names != export_manifest.get("input_names"):
        raise SystemExit("actual ONNX inputs do not match export manifest")
    if output_names != export_manifest.get("output_names"):
        raise SystemExit("actual ONNX outputs do not match export manifest")
    if input_names != [
        metadata.get("input_features_name", "log_mel"),
        metadata.get("frame_mask_name", "frame_mask"),
    ]:
        raise SystemExit("actual ONNX inputs do not match runtime metadata")
    if output_names != [metadata.get("endpoint_output_name") or "endpoint_probability"]:
        raise SystemExit("actual ONNX output does not match runtime metadata")

    frontend = metadata["frontend"]
    frames = max(
        2,
        int(
            round(
                float(frontend["max_seconds"])
                * int(frontend["sample_rate"])
                / int(frontend["hop_length"])
            )
        ),
    )
    features = np.zeros((1, int(frontend["n_mels"]), frames), dtype=np.float32)
    mask = np.ones((1, frames), dtype=np.float32)
    try:
        output = session.run(
            output_names,
            {input_names[0]: features, input_names[1]: mask},
        )[0]
    except Exception as exc:
        raise SystemExit(f"released ONNX graph failed representative inference: {exc}") from exc
    values = np.asarray(output)
    if values.size != 1 or not bool(np.isfinite(values).all()):
        raise SystemExit("released ONNX graph returned invalid batch-one output")
    probability = float(values.reshape(-1)[0])
    if not 0.0 <= probability <= 1.0:
        raise SystemExit("released ONNX output is not a probability in [0, 1]")


def _validate_export(
    model: Path,
    metadata: dict,
    export_manifest: dict,
    *,
    development_only: bool,
) -> None:
    if model.suffix.lower() != ".onnx":
        raise SystemExit("Hugging Face/Space release requires an actual .onnx model")
    if export_manifest.get("format_version") != 2:
        raise SystemExit("release requires export manifest format_version=2; export again")
    for key in (
        "model_name",
        "architecture",
        "frontend",
        "threshold",
        "parameter_count",
        "training_status",
        "development_only",
        "data_scope",
        "data_revision",
    ):
        if metadata.get(key) is None:
            raise SystemExit(f"model metadata is missing {key}")
    if metadata.get("output_type") != "probability":
        raise SystemExit("released ONNX must expose endpoint probability, not logits")
    if not isinstance(metadata.get("frontend"), dict):
        raise SystemExit("model metadata frontend must be an object")
    for key in ("sample_rate", "hop_length", "n_mels", "max_seconds"):
        value = metadata["frontend"].get(key)
        if not isinstance(value, int | float) or float(value) <= 0:
            raise SystemExit(f"model metadata frontend has invalid {key}")
    if (
        isinstance(metadata.get("threshold"), bool)
        or not isinstance(metadata.get("threshold"), int | float)
        or not 0.0 <= float(metadata["threshold"]) <= 1.0
    ):
        raise SystemExit("model metadata threshold must be in [0, 1]")
    controller = metadata.get("controller")
    if not isinstance(controller, dict):
        raise SystemExit("model metadata controller must be an object")
    for key in (
        "endpoint_threshold",
        "long_pause_threshold",
        "min_silence_ms",
        "relax_after_ms",
        "max_silence_ms",
    ):
        value = controller.get(key)
        if isinstance(value, bool) or not isinstance(value, int | float):
            raise SystemExit(f"model metadata controller has invalid {key}")
    if not _same_value(controller["endpoint_threshold"], metadata["threshold"]):
        raise SystemExit("controller endpoint threshold does not match model threshold")
    if (
        not 0.0
        <= float(controller["long_pause_threshold"])
        <= float(controller["endpoint_threshold"])
        <= 1.0
    ):
        raise SystemExit("model metadata controller has invalid probability thresholds")
    if (
        not 0.0
        <= float(controller["min_silence_ms"])
        <= float(controller["relax_after_ms"])
        <= float(controller["max_silence_ms"])
    ):
        raise SystemExit("model metadata controller has invalid silence bounds")
    confirmations = controller.get("required_confirmations")
    if isinstance(confirmations, bool) or not isinstance(confirmations, int) or confirmations < 1:
        raise SystemExit("model metadata controller has invalid required_confirmations")
    if (
        isinstance(metadata.get("parameter_count"), bool)
        or not isinstance(metadata.get("parameter_count"), int)
        or metadata["parameter_count"] <= 0
    ):
        raise SystemExit("model metadata parameter_count must be positive")
    if export_manifest.get("task") != "audio-turn-end-detection":
        raise SystemExit("export manifest has the wrong or missing task")
    if export_manifest.get("input_names") != ["log_mel", "frame_mask"]:
        raise SystemExit("export manifest has the wrong model inputs")
    if export_manifest.get("output_names") != ["endpoint_probability"]:
        raise SystemExit("export manifest has the wrong model output")
    _validate_local_evidence(export_manifest.get("resolved_config"), "resolved config")
    training_data = export_manifest.get("training_data")
    if not isinstance(training_data, dict):
        raise SystemExit("export manifest has no training-data provenance")
    if not _same_value(training_data.get("revision"), metadata.get("data_revision")):
        raise SystemExit("training-data revision does not match runtime metadata")
    if not _same_value(training_data.get("scope"), metadata.get("data_scope")):
        raise SystemExit("training-data scope does not match runtime metadata")
    sources = training_data.get("sources")
    if not isinstance(sources, dict) or "train_source" not in sources:
        raise SystemExit("export manifest has no training source")
    for name, evidence in sources.items():
        _validate_local_evidence(evidence, name)
    _validate_source_inventory(export_manifest)
    try:
        fp32 = export_manifest["files"]["fp32"]
    except (KeyError, TypeError) as exc:
        raise SystemExit("export manifest is missing files.fp32") from exc
    if fp32.get("sha256") != _sha256(model) or fp32.get("bytes") != model.stat().st_size:
        raise SystemExit("model bytes do not match export_manifest.json")
    parity = export_manifest.get("parity", {}).get("fp32_max_abs_error")
    if not isinstance(parity, int | float) or not math.isfinite(parity) or parity > 1e-4:
        raise SystemExit("export manifest needs passing FP32 ONNX parity <= 1e-4")
    bindings = {
        "development_only": (development_only, export_manifest.get("development_only")),
        "training_status": (
            metadata.get("training_status"),
            export_manifest.get("training_status"),
        ),
        "data_scope": (metadata.get("data_scope"), export_manifest.get("data_scope")),
        "data_revision": (
            metadata.get("data_revision"),
            export_manifest.get("data_revision"),
        ),
        "architecture/model_type": (
            metadata.get("architecture"),
            export_manifest.get("model_type"),
        ),
        "threshold": (metadata.get("threshold"), export_manifest.get("threshold")),
        "controller": (metadata.get("controller"), export_manifest.get("controller")),
        "parameter_count": (
            metadata.get("parameter_count"),
            export_manifest.get("parameter_count"),
        ),
    }
    for name, (metadata_value, export_value) in bindings.items():
        if not _same_value(metadata_value, export_value):
            raise SystemExit(f"metadata/export manifest mismatch for {name}")
    _validate_onnx_runtime(model, metadata, export_manifest)


def _validate_metrics(
    metrics: dict,
    metadata: dict,
    export_manifest: dict,
    *,
    development_only: bool,
) -> None:
    expected_split = {"validation", "development", "dev"} if development_only else {"test"}
    if str(metrics.get("split", "")).lower() not in expected_split:
        raise SystemExit(f"metrics split must be one of {sorted(expected_split)} for this release")
    if development_only and metrics.get("official_test") is True:
        raise SystemExit("development metrics cannot claim official-test scope")
    if not development_only:
        if metrics.get("official_test") is not True:
            raise SystemExit("final metrics must be marked official_test=true")
        if metrics.get("dataset_revision") != "0500378e8ed6d38e37b016e24d261e8e6c6a6859":
            raise SystemExit("final metrics use the wrong official-test revision")
        freeze_hash = metrics.get("freeze_manifest_sha256")
        if not _is_hex_sha256(freeze_hash):
            raise SystemExit("final metrics are not bound to a freeze manifest")
    bindings = {
        "development_only": (development_only, metrics.get("development_only")),
        "training_status": (metadata.get("training_status"), metrics.get("training_status")),
        "data_scope": (metadata.get("data_scope"), metrics.get("data_scope")),
        "data_revision": (metadata.get("data_revision"), metrics.get("data_revision")),
        "threshold": (metadata.get("threshold"), metrics.get("threshold")),
        "checkpoint_sha256": (
            export_manifest.get("checkpoint", {}).get("sha256"),
            metrics.get("checkpoint_sha256"),
        ),
    }
    for name, (expected, actual) in bindings.items():
        if expected is None or not _same_value(expected, actual):
            raise SystemExit(f"metrics are not bound to the exported artifact: {name}")

    measured = metrics.get("metrics")
    if not isinstance(measured, dict):
        raise SystemExit("metrics artifact has no measured metrics object")
    integer_fields = ("count", "positive_count", "negative_count", "tp", "fp", "tn", "fn")
    for name in integer_fields:
        value = measured.get(name)
        if not isinstance(value, int) or isinstance(value, bool) or value < 0:
            raise SystemExit(f"metrics artifact has invalid {name}")
    count = measured["count"]
    positives = measured["positive_count"]
    negatives = measured["negative_count"]
    if count < 2 or positives < 1 or negatives < 1 or positives + negatives != count:
        raise SystemExit("metrics artifact needs a non-empty two-class evaluation")
    if measured["tp"] + measured["fn"] != positives:
        raise SystemExit("metrics positive confusion counts are inconsistent")
    if measured["tn"] + measured["fp"] != negatives:
        raise SystemExit("metrics negative confusion counts are inconsistent")
    if not _same_value(measured.get("threshold"), metadata.get("threshold")):
        raise SystemExit("measured threshold does not match the released threshold")
    for name in ("roc_auc", "average_precision", "brier_score", "log_loss"):
        value = measured.get(name)
        if not isinstance(value, int | float) or not math.isfinite(float(value)):
            raise SystemExit(f"metrics artifact has invalid {name}")
    if not development_only:
        catalog = _load_json(ROOT / "configs" / "datasets.json")
        expected_rows = catalog.get("official_test", {}).get("expected_rows")
        if count != expected_rows:
            raise SystemExit(
                f"official-test metrics must contain exactly {expected_rows} examples, got {count}"
            )


def _validate_synthetic_replay(
    fixture: Path,
    decisions: Path,
    summary_path: Path,
    controller: object,
) -> None:
    paths = (fixture, decisions, summary_path)
    if not any(path.exists() for path in paths):
        return
    if not all(path.is_file() and not path.is_symlink() for path in paths):
        raise SystemExit("synthetic controller replay package is incomplete")
    summary = _load_json(summary_path)
    if summary.get("format_version") != 1:
        raise SystemExit("controller replay summary has an unsupported format")
    if summary.get("evidence_scope") != "synthetic_integration":
        raise SystemExit("controller replay report is not labelled synthetic integration")
    if not _same_value(summary.get("controller_config"), controller):
        raise SystemExit("controller replay policy differs from released model metadata")
    for label, path, evidence in (
        ("input", fixture, summary.get("input")),
        ("decisions", decisions, summary.get("decisions")),
    ):
        if not isinstance(evidence, dict):
            raise SystemExit(f"controller replay summary has no {label} evidence")
        if evidence.get("bytes") != path.stat().st_size or evidence.get("sha256") != _sha256(path):
            raise SystemExit(f"controller replay {label} does not match its hash evidence")
    if summary.get("duplicate_response_emissions") != 0:
        raise SystemExit("controller integration emitted duplicate response edges")


def _validate_benchmark(
    path: Path,
    artifact_evidence: object,
    *,
    expected_scope: str | None = None,
) -> None:
    if not isinstance(artifact_evidence, dict):
        raise SystemExit(f"benchmark {path.name} has no bound artifact evidence")
    report = _load_json(path)
    if report.get("artifact_bytes") != artifact_evidence.get("bytes") or report.get(
        "artifact_sha256"
    ) != artifact_evidence.get("sha256"):
        raise SystemExit(f"benchmark {path.name} is stale for the released artifact")
    if expected_scope is not None and report.get("scope") != expected_scope:
        raise SystemExit(f"benchmark {path.name} has the wrong measurement scope")
    for name in ("threads", "batch_size", "measured_iterations"):
        value = report.get(name)
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise SystemExit(f"benchmark {path.name} has invalid {name}")
    p95 = report.get("warm_latency_ms", {}).get("p95")
    if (
        isinstance(p95, bool)
        or not isinstance(p95, int | float)
        or not math.isfinite(float(p95))
        or float(p95) < 0.0
    ):
        raise SystemExit(f"benchmark {path.name} has invalid p95 latency")


def _file_inventory(root: Path) -> dict[str, dict[str, int | str]]:
    return {
        path.relative_to(root).as_posix(): {
            "bytes": path.stat().st_size,
            "sha256": _sha256(path),
        }
        for path in sorted(root.rglob("*"))
        if path.is_file()
    }


def _inventory_sha256(inventory: dict[str, dict[str, int | str]]) -> str:
    serialized = json.dumps(inventory, sort_keys=True, separators=(",", ":")).encode()
    return hashlib.sha256(serialized).hexdigest()


def _replace_release_directory(staged: Path, destination: Path) -> None:
    """Swap a fully built staging tree into place with rollback on rename failure."""

    backup: Path | None = None
    if destination.exists():
        backup = Path(
            tempfile.mkdtemp(prefix=f".{destination.name}.previous-", dir=destination.parent)
        )
        backup.rmdir()
        os.replace(destination, backup)
    try:
        os.replace(staged, destination)
    except Exception:
        if backup is not None and backup.exists() and not destination.exists():
            os.replace(backup, destination)
        raise
    if backup is not None:
        for attempt in range(3):
            try:
                shutil.rmtree(backup)
                break
            except FileNotFoundError:
                break
            except OSError as exc:
                if exc.errno != errno.ENOTEMPTY or attempt == 2:
                    raise


def main() -> int:
    args = parse_args()
    model = _resolve(args.model)
    metadata_path = _resolve(args.metadata)
    metrics_path = _resolve(args.metrics)
    export_manifest_path = (
        _resolve(args.export_manifest)
        if args.export_manifest
        else model.with_name("export_manifest.json")
    )
    destination = _resolve(args.output)
    frozen_manifest_path = _resolve(args.frozen_manifest) if args.frozen_manifest else None

    try:
        output_relative = destination.relative_to(ROOT)
    except ValueError as exc:
        raise SystemExit("--output must stay inside the project directory") from exc
    if destination == ROOT:
        raise SystemExit("refusing to replace the project root; choose a release subdirectory")
    first_component = output_relative.parts[0]
    safe_release_path = first_component == "release" or first_component.startswith("release-")
    safe_artifact_path = (
        first_component == "artifacts"
        and len(output_relative.parts) >= 2
        and output_relative.parts[1].startswith("release")
    )
    if not (safe_release_path or safe_artifact_path):
        raise SystemExit("--output must be release/, release-*/, or artifacts/release*/")
    raw_output = Path(args.output)
    if not raw_output.is_absolute():
        raw_output = ROOT / raw_output
    if raw_output.is_symlink():
        raise SystemExit("refusing a symlink release destination")

    inputs = [model, metadata_path, metrics_path, export_manifest_path]
    if frozen_manifest_path is not None:
        inputs.append(frozen_manifest_path)
    for input_path in inputs:
        if input_path == destination or destination in input_path.parents:
            raise SystemExit("release output cannot contain or replace one of its input artifacts")

    if not metadata_path.is_file():
        raise SystemExit(f"model metadata does not exist: {metadata_path}")
    metadata = _load_json(metadata_path)
    metadata_is_development = bool(metadata.get("smoke_test") or metadata.get("development_only"))
    if not metadata_is_development and str(metadata.get("training_status", "")).lower() != "final":
        raise SystemExit("a non-development release requires training_status='final'")
    if metadata_is_development and not args.allow_development_artifact:
        raise SystemExit(
            "refusing to release a smoke/development model without --allow-development-artifact"
        )
    metrics = _load_json(metrics_path) if metrics_path.is_file() else None
    if metrics is None and not args.allow_development_artifact:
        raise SystemExit("final release requires measured official-test metrics")
    if not model.is_file():
        raise SystemExit(f"model artifact does not exist: {model}")
    if not export_manifest_path.is_file():
        raise SystemExit(f"export manifest does not exist: {export_manifest_path}")
    export_manifest = _load_json(export_manifest_path)
    _validate_export(
        model,
        metadata,
        export_manifest,
        development_only=metadata_is_development,
    )
    if metrics is not None:
        _validate_metrics(
            metrics,
            metadata,
            export_manifest,
            development_only=metadata_is_development,
        )
    if args.include_synthetic_controller_replay:
        _validate_synthetic_replay(
            ROOT / "data" / "collection" / "controller_replay_fixture.jsonl",
            ROOT / "reports" / "controller_replay_integration.jsonl",
            ROOT / "reports" / "controller_replay_integration.summary.json",
            metadata.get("controller"),
        )
    _validate_space_card(ROOT / "space" / "README.md")
    _validate_space_requirements(ROOT / "space" / "requirements.txt")
    _validate_space_app(ROOT / "app.py")

    frozen_manifest: dict | None = None
    if not metadata_is_development:
        if frozen_manifest_path is None:
            raise SystemExit("final release requires --frozen-manifest")
        if frozen_manifest_path.is_symlink() or not frozen_manifest_path.is_file():
            raise SystemExit("final freeze manifest must be a regular file")
        from turn_detection.provenance import verify_freeze_manifest

        try:
            frozen_manifest = verify_freeze_manifest(frozen_manifest_path, ROOT)
        except ValueError as exc:
            raise SystemExit(f"final freeze manifest failed verification: {exc}") from exc
        if metrics is None:
            raise SystemExit("final release requires official-test metrics")
        if metrics.get("freeze_manifest_sha256") != _sha256(frozen_manifest_path):
            raise SystemExit("official-test metrics do not match --frozen-manifest")
        frozen_checkpoint = frozen_manifest.get("checkpoint", {})
        exported_checkpoint = export_manifest.get("checkpoint", {})
        if frozen_checkpoint.get("sha256") != exported_checkpoint.get("sha256"):
            raise SystemExit("exported checkpoint differs from the official-test freeze")
        if not _same_value(frozen_manifest.get("threshold"), metadata.get("threshold")):
            raise SystemExit("released threshold differs from the official-test freeze")

    destination.parent.mkdir(parents=True, exist_ok=True)
    output = Path(tempfile.mkdtemp(prefix=f".{destination.name}.staging-", dir=destination.parent))
    model_release = output / "model"
    space_release = output / "space"
    model_release.mkdir(parents=True)
    space_release.mkdir(parents=True)

    suffix = model.suffix or ".onnx"
    packaged_model_name = f"model{suffix}"
    _copy_file(model, model_release / packaged_model_name)
    _copy_file(metadata_path, model_release / "model_metadata.json")
    _copy_file(export_manifest_path, model_release / "export_manifest.json")
    metrics_scope = "development" if metadata_is_development else "official_test"
    packaged_metrics_name = (
        "development_metrics.json" if metadata_is_development else "test_metrics.json"
    )
    if metrics_path.is_file():
        _copy_file(metrics_path, model_release / packaged_metrics_name)
    if frozen_manifest_path is not None and not metadata_is_development:
        _copy_file(frozen_manifest_path, model_release / "frozen_manifest.json")
    for artifact_name in ("resolved_config.json", "history.json"):
        source = model.parent / artifact_name
        if source.is_file():
            _copy_file(source, model_release / "run_artifacts" / artifact_name)
    benchmark_contracts = {
        "cpu_benchmark.json": (export_manifest.get("checkpoint"), None),
        "onnx_benchmark.json": (
            export_manifest.get("files", {}).get("fp32"),
            "neural_model_only_log_mel_input",
        ),
        "onnx_e2e_benchmark.json": (
            export_manifest.get("files", {}).get("fp32"),
            "end_to_end_waveform_to_probability",
        ),
    }
    for artifact_name, (artifact_evidence, expected_scope) in benchmark_contracts.items():
        source = model.parent / artifact_name
        if source.is_file():
            _validate_benchmark(source, artifact_evidence, expected_scope=expected_scope)
            _copy_file(source, model_release / "run_artifacts" / artifact_name)
    baseline_model = ROOT / "artifacts" / "partial-baseline" / "model.json"
    baseline_report_path = ROOT / "reports" / "partial_baseline_metrics.json"
    if baseline_model.is_file() and baseline_report_path.is_file():
        baseline_report = _load_json(baseline_report_path)
        baseline_evidence = baseline_report.get("model")
        if not isinstance(baseline_evidence, dict):
            raise SystemExit("baseline report does not bind its model artifact")
        if baseline_evidence.get("bytes") != baseline_model.stat().st_size or baseline_evidence.get(
            "sha256"
        ) != _sha256(baseline_model):
            raise SystemExit("baseline model does not match its report evidence")
        _copy_file(
            baseline_model,
            model_release / "reference_models" / "acoustic_baseline.json",
        )
    elif baseline_model.exists() or baseline_report_path.exists():
        raise SystemExit("baseline release inputs are incomplete")
    _copy_file(ROOT / "MODEL_CARD.md", model_release / "README.md")
    _copy_file(ROOT / "LICENSE", model_release / "LICENSE")
    _copy_file(ROOT / "NOTICE", model_release / "NOTICE")
    _copy_project_source(
        model_release,
        include_synthetic_replay=args.include_synthetic_controller_replay,
    )
    for config in sorted((ROOT / "configs").glob("*")):
        if config.is_file():
            _copy_file(config, model_release / "configs" / config.name)

    _copy_file(ROOT / "space/README.md", space_release / "README.md")
    _copy_file(ROOT / "space/requirements.txt", space_release / "requirements.txt")
    _copy_file(ROOT / "LICENSE", space_release / "LICENSE")
    _copy_file(ROOT / "NOTICE", space_release / "NOTICE")
    _copy_file(ROOT / "MODEL_CARD.md", space_release / "MODEL_CARD.md")
    _copy_file(ROOT / "DATA_CARD.md", space_release / "DATA_CARD.md")
    _copy_file(ROOT / "app.py", space_release / "app.py")
    _copy_runtime(space_release)
    _copy_file(model, space_release / "artifacts" / "model.onnx")
    _copy_file(metadata_path, space_release / "artifacts" / "model_metadata.json")

    model_inventory = _file_inventory(model_release)
    space_inventory = _file_inventory(space_release)
    staged_inventory = {
        "model": model_inventory,
        "space": space_inventory,
    }
    model_provenance = {
        "format_version": 1,
        "development_only": metadata_is_development,
        "model_file": packaged_model_name,
        "metrics_file": packaged_metrics_name if metrics_path.is_file() else None,
        "metrics_scope": metrics_scope if metrics_path.is_file() else None,
        "source_and_artifact_inventory_sha256": hashlib.sha256(
            json.dumps(staged_inventory, sort_keys=True, separators=(",", ":")).encode()
        ).hexdigest(),
        "files": staged_inventory,
    }
    (model_release / "release_provenance.json").write_text(
        json.dumps(model_provenance, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )

    release_manifest = {
        "development_only": bool(args.allow_development_artifact or metadata_is_development),
        "model_file": packaged_model_name,
        "model_size_bytes": model.stat().st_size,
        "model_sha256": _sha256(model),
        "metadata_sha256": _sha256(metadata_path),
        "export_manifest_sha256": _sha256(export_manifest_path),
        "metrics_sha256": _sha256(metrics_path) if metrics_path.is_file() else None,
        "metrics_file": packaged_metrics_name if metrics_path.is_file() else None,
        "metrics_scope": metrics_scope if metrics_path.is_file() else None,
        "model_metadata": metadata,
        "has_test_metrics": bool(metrics is not None and metrics_scope == "official_test"),
    }
    release_manifest["files"] = _file_inventory(output)
    release_manifest["release_inventory_sha256"] = _inventory_sha256(release_manifest["files"])
    (output / "release_manifest.json").write_text(
        json.dumps(release_manifest, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    _replace_release_directory(output, destination)
    print(json.dumps(release_manifest, indent=2, sort_keys=True))
    return 0


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