#!/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())