quantization/release_gate.py

#!/usr/bin/env python3
"""Create a content-addressed, fail-closed model release manifest."""

from __future__ import annotations

import argparse
import hashlib
import importlib.metadata
import json
import math
import platform
import re
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any


PROJECT_ROOT = Path(__file__).resolve().parent.parent
import sys

if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tiff", ".tif", ".webp", ".bmp"}
HARD_PROTECTED_GROUPS = {"vision_encoder", "lm_head"}
EVIDENCE_PROTECTED_GROUPS = {"vision_projector", "token_embeddings"}
CANDIDATE_METADATA_FILES = (
    "config.json",
    "model.safetensors.index.json",
    "processor_config.json",
    "special_tokens_map.json",
    "tokenizer.json",
    "tokenizer_config.json",
    "chat_template.jinja",
    "precision_map.json",
    "quantization_summary.json",
)
REQUIRED_CANDIDATE_METADATA_FILES = {
    "config.json",
    "processor_config.json",
    "tokenizer.json",
    "precision_map.json",
    "quantization_summary.json",
}
RELEASE_PROJECT_FILES = (
    "LICENSE",
    "pyproject.toml",
    "generation_config.json",
    "preprocessing_config.json",
    "quantization_config.json",
    "docs/PRD.md",
    "docs/TECHNICAL_SPEC.md",
    "docs/adr/0001-verified-ocr-aware-quantization.md",
    "docs/adr/0002-calibrated-affine8-lm-head.md",
    "src/unlimited_ocr/__init__.py",
    "src/unlimited_ocr/engine.py",
    "src/unlimited_ocr/pipeline.py",
    "src/unlimited_ocr/preprocessing.py",
    "src/unlimited_ocr/output.py",
    "src/unlimited_ocr/pdf.py",
    "src/unlimited_ocr/cli.py",
    "src/unlimited_ocr/profiles.py",
    "src/unlimited_ocr/server.py",
    "src/unlimited_ocr/adapter_registry.json",
    "quantization/layer_sensitivity.py",
    "quantization/calibrate_precision.py",
    "quantization/mixed_precision_convert.py",
    "quantization/precision_map.json",
    "quantization/release_gate.py",
    "quantization/run_pipeline.py",
    "benchmarks/evaluate_cer.py",
    "benchmarks/evaluate_tables.py",
    "benchmarks/normalize_output.py",
    "benchmarks/run_accuracy.py",
    "benchmarks/run_performance.py",
    "benchmarks/rswa_validation.py",
    "benchmarks/datasets.md",
    "examples/single_image.py",
    "examples/multi_page_pdf.py",
    "examples/batch_directory.py",
)
CALIBRATION_EVIDENCE_KEYS = {
    "bfloat16": (
        "calibration_bfloat16_accuracy",
        "calibration_bfloat16_performance",
    ),
    "mxfp8": (
        "calibration_mxfp8_accuracy",
        "calibration_mxfp8_performance",
    ),
    "affine8": (
        "calibration_affine8_accuracy",
        "calibration_affine8_performance",
    ),
}
CALIBRATION_BASELINE_EVIDENCE_KEYS = {
    "calibration_baseline_accuracy",
    "calibration_reference_performance",
}
CALIBRATION_RAW_EVIDENCE_NAMES = frozenset(
    name for pair in CALIBRATION_EVIDENCE_KEYS.values() for name in pair
) | frozenset(CALIBRATION_BASELINE_EVIDENCE_KEYS)
RELEASE_EVIDENCE_NAMES = (
    frozenset(
        {
            "bf16_accuracy",
            "reference_accuracy",
            "candidate_accuracy",
            "reference_performance",
            "candidate_performance",
            "candidate_rswa",
            "sensitivity_results",
            "calibration_results",
            "generated_precision_map",
            "provenance",
        }
    )
    | CALIBRATION_RAW_EVIDENCE_NAMES
)
DEFAULT_THRESHOLDS = {
    # n≈12 held-out macro CER is noisy; allow 1.5pp absolute vs BF16.
    "max_cer_delta_vs_bf16": 0.015,
    "max_cer_delta_vs_reference": 0.005,
    "max_digit_cer_delta_vs_bf16": 0.01,
    "max_table_score_degradation_vs_bf16": 0.01,
    "min_tps_ratio_vs_reference": 0.90,
    "max_weight_size_gb": 4.5,
    "min_rswa_tokens": 8192,
    # Forced-min-token R-SWA runs suppress EOS through 8k; some residual
    # loopiness is expected. Cap still catches fully degenerate collapse.
    "max_rswa_repetition_rate": 0.25,
}
MLX_RELEASE_SCHEMA_VERSION = 3
MLX_RELEASE_GATE_NAMES = (
    "weights_are_distinct",
    "weight_size_gb",
    "native_model_metadata",
    "candidate_source_provenance",
    "evaluation_coverage",
    "held_out_evaluation_dataset",
    "accuracy_aggregates_recomputed",
    "same_evaluation_samples",
    "model_identities",
    "immutable_model_revisions",
    "same_accuracy_recipe",
    "candidate_cer_vs_bf16",
    "candidate_cer_vs_reference",
    "candidate_digit_cer_vs_bf16",
    "candidate_table_score_vs_bf16",
    "performance_aggregates_recomputed",
    "same_performance_setup",
    "candidate_tps_vs_reference",
    "rswa_8k_bounded",
    "provenance_matches_release",
    "calibration_recomputed",
    "sensitivity_matches_calibration_dataset",
    "candidate_precision_map_matches_evidence",
    "precision_map_reproducible",
)


def _reject_json_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number: {value}")


def _json_digest(value: Any) -> str:
    payload = json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    ).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def load_json_object(path: str | Path) -> dict:
    """Load a JSON object and reject malformed release evidence."""
    path = Path(path)
    try:
        value = json.loads(
            path.read_text(encoding="utf-8"),
            parse_constant=_reject_json_constant,
        )
    except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
        raise ValueError(f"Invalid JSON file: {path}") from exc
    if not isinstance(value, dict):
        raise ValueError(f"JSON file must contain an object: {path}")
    return value


def calibration_raw_evidence_paths(
    artifacts_dir: str | Path,
    calibration: dict,
) -> dict[str, Path]:
    """Resolve the eight allowlisted raw calibration artifacts safely."""
    artifacts_dir = Path(artifacts_dir)
    if artifacts_dir.is_symlink() or not artifacts_dir.is_dir():
        raise FileNotFoundError(
            f"Calibration artifacts directory not found: {artifacts_dir}"
        )
    calibration_inputs = calibration.get("input_artifacts")
    if not isinstance(calibration_inputs, dict):
        raise ValueError("Calibration input artifact records are missing")
    paths = {}
    filenames = set()
    for name in CALIBRATION_RAW_EVIDENCE_NAMES:
        record = calibration_inputs.get(name)
        filename = record.get("filename") if isinstance(record, dict) else None
        if (
            not isinstance(filename, str)
            or not filename
            or Path(filename).name != filename
            or filename in filenames
        ):
            raise ValueError(f"Calibration input filename is invalid: {name}")
        filenames.add(filename)
        paths[name] = artifacts_dir / filename
    return paths


def sha256_file(path: str | Path, chunk_size: int = 8 * 1024 * 1024) -> str:
    """Return the SHA-256 digest of a file without loading it into memory."""
    if (
        not isinstance(chunk_size, int)
        or isinstance(chunk_size, bool)
        or chunk_size < 1
    ):
        raise ValueError("chunk_size must be a positive integer")
    digest = hashlib.sha256()
    with Path(path).open("rb") as handle:
        while chunk := handle.read(chunk_size):
            digest.update(chunk)
    return digest.hexdigest()


def release_files_manifest(
    base_dir: str | Path,
    relative_paths: tuple[str, ...],
    *,
    required: bool,
) -> dict[str, Any]:
    """Hash an allowlisted set of regular files for publication approval."""
    base_dir = Path(base_dir)
    if base_dir.is_symlink():
        raise ValueError(f"Release directory must not be a symbolic link: {base_dir}")
    if len(set(relative_paths)) != len(relative_paths):
        raise ValueError("Release file allowlist contains duplicates")
    records: list[dict[str, Any]] = []
    aggregate = hashlib.sha256()
    missing: list[str] = []
    for relative in relative_paths:
        pure_path = PurePosixPath(relative)
        if (
            not relative
            or pure_path.is_absolute()
            or pure_path.as_posix() != relative
            or any(part in {"", ".", ".."} for part in pure_path.parts)
        ):
            raise ValueError(f"Unsafe release file path: {relative!r}")
        path = base_dir.joinpath(*pure_path.parts)
        parents = [
            base_dir.joinpath(*pure_path.parts[:index])
            for index in range(1, len(pure_path.parts) + 1)
        ]
        if any(parent.is_symlink() for parent in parents):
            raise ValueError(f"Release files must not be symbolic links: {relative}")
        if not path.is_file():
            if required:
                missing.append(relative)
            continue
        size = path.stat().st_size
        digest = sha256_file(path)
        aggregate.update(relative.encode("utf-8"))
        aggregate.update(b"\0")
        aggregate.update(str(size).encode("ascii"))
        aggregate.update(b"\0")
        aggregate.update(digest.encode("ascii"))
        aggregate.update(b"\n")
        records.append({"path": relative, "size": size, "sha256": digest})
    if missing:
        raise FileNotFoundError(
            "Required release files are missing: " + ", ".join(missing)
        )
    return {"files": records, "aggregate_sha256": aggregate.hexdigest()}


def candidate_metadata_manifest(model_dir: str | Path) -> dict[str, Any]:
    """Hash every candidate metadata file the publisher may upload."""
    manifest = release_files_manifest(
        model_dir,
        CANDIDATE_METADATA_FILES,
        required=False,
    )
    present = {record["path"] for record in manifest["files"]}
    missing = sorted(REQUIRED_CANDIDATE_METADATA_FILES - present)
    if missing:
        raise FileNotFoundError(
            "Candidate is missing required files: " + ", ".join(missing)
        )
    return manifest


def model_weight_manifest(
    model_dir: str | Path,
    *,
    allow_symlinks: bool = False,
) -> dict:
    """Hash Safetensors files and create a stable aggregate digest."""
    if not isinstance(allow_symlinks, bool):
        raise TypeError("allow_symlinks must be a boolean")
    model_dir = Path(model_dir)
    if model_dir.is_symlink():
        raise ValueError("Model directory must not be a symbolic link")
    weight_candidates = sorted(model_dir.glob("*.safetensors"))
    symlinks = [path.name for path in weight_candidates if path.is_symlink()]
    if symlinks and not allow_symlinks:
        raise ValueError(
            "Safetensors weights must not be symbolic links: " + ", ".join(symlinks)
        )
    weight_paths = [path for path in weight_candidates if path.is_file()]
    if not weight_paths:
        raise FileNotFoundError(f"No Safetensors weights found in: {model_dir}")

    aggregate = hashlib.sha256()
    files = []
    total_size = 0
    for path in weight_paths:
        size = path.stat().st_size
        digest = sha256_file(path)
        total_size += size
        aggregate.update(path.name.encode("utf-8"))
        aggregate.update(b"\0")
        aggregate.update(str(size).encode("ascii"))
        aggregate.update(b"\0")
        aggregate.update(digest.encode("ascii"))
        aggregate.update(b"\n")
        files.append({"name": path.name, "size": size, "sha256": digest})

    return {
        "files": files,
        "total_size_bytes": total_size,
        "total_size_gb": total_size / (1024**3),
        "aggregate_sha256": aggregate.hexdigest(),
    }


def _declared_dataset_path(
    value: Any,
    *,
    directory: str,
    allowed_suffixes: set[str],
) -> str:
    """Validate a normalized, direct child path in a dataset manifest."""
    if (
        not isinstance(value, str)
        or not value
        or value.strip() != value
        or "\\" in value
    ):
        raise ValueError(f"manifest sample requires a valid {directory} path")
    path = PurePosixPath(value)
    if (
        path.is_absolute()
        or len(path.parts) != 2
        or path.parts[0] != directory
        or path.as_posix() != value
        or path.suffix.lower() not in allowed_suffixes
    ):
        raise ValueError(f"Unsafe or unsupported dataset path: {value}")
    return value


def dataset_manifest(eval_dir: str | Path) -> dict:
    """Validate and content-hash an exact OCR image/ground-truth dataset."""
    eval_dir = Path(eval_dir)
    images_dir = eval_dir / "images"
    ground_truth_dir = eval_dir / "ground_truth"
    manifest_path = eval_dir / "manifest.json"
    if eval_dir.is_symlink() or images_dir.is_symlink() or ground_truth_dir.is_symlink():
        raise ValueError("Evaluation dataset directories must not be symbolic links")
    if not images_dir.is_dir() or not ground_truth_dir.is_dir():
        raise FileNotFoundError("Evaluation dataset requires images/ and ground_truth/")
    if manifest_path.is_symlink():
        raise ValueError("Evaluation manifest must not be a symbolic link")
    declared = load_json_object(manifest_path)

    image_entries = sorted(images_dir.iterdir())
    invalid_image_entries = [
        path.name
        for path in image_entries
        if path.is_symlink()
        or not path.is_file()
        or path.suffix.lower() not in IMAGE_EXTENSIONS
    ]
    if invalid_image_entries:
        raise ValueError(
            "Evaluation images directory contains unsupported entries: "
            + ", ".join(invalid_image_entries)
        )
    image_candidates = image_entries
    image_symlinks = [path.name for path in image_candidates if path.is_symlink()]
    if image_symlinks:
        raise ValueError(
            "Evaluation images must not be symbolic links: " + ", ".join(image_symlinks)
        )
    image_paths = [path for path in image_candidates if path.is_file()]
    if not image_paths:
        raise ValueError(f"Evaluation dataset contains no images: {images_dir}")
    duplicate_stems = sorted(
        stem
        for stem, count in Counter(path.stem for path in image_paths).items()
        if count > 1
    )
    if duplicate_stems:
        raise ValueError("Duplicate image stems: " + ", ".join(duplicate_stems))

    ground_truth_entries = sorted(ground_truth_dir.iterdir())
    invalid_ground_truth_entries = [
        path.name
        for path in ground_truth_entries
        if path.is_symlink() or not path.is_file() or path.suffix.lower() != ".txt"
    ]
    if invalid_ground_truth_entries:
        raise ValueError(
            "Ground-truth directory contains unsupported entries: "
            + ", ".join(invalid_ground_truth_entries)
        )
    ground_truth_candidates = ground_truth_entries
    ground_truth_symlinks = [
        path.name for path in ground_truth_candidates if path.is_symlink()
    ]
    if ground_truth_symlinks:
        raise ValueError(
            "Ground-truth files must not be symbolic links: "
            + ", ".join(ground_truth_symlinks)
        )
    ground_truth_paths = [path for path in ground_truth_candidates if path.is_file()]

    declared_samples = declared.get("samples")
    if not isinstance(declared_samples, list) or not declared_samples:
        raise ValueError("manifest.json must declare at least one sample")
    samples_by_image: dict[str, dict[str, Any]] = {}
    declared_ground_truth: set[str] = set()
    declared_ids: set[int] = set()
    for index, sample in enumerate(declared_samples, start=1):
        if not isinstance(sample, dict):
            raise ValueError(f"Manifest sample {index} must be an object")
        relative_image = _declared_dataset_path(
            sample.get("image"),
            directory="images",
            allowed_suffixes=IMAGE_EXTENSIONS,
        )
        relative_ground_truth = _declared_dataset_path(
            sample.get("ground_truth"),
            directory="ground_truth",
            allowed_suffixes={".txt"},
        )
        expected_ground_truth = f"ground_truth/{PurePosixPath(relative_image).stem}.txt"
        if relative_ground_truth != expected_ground_truth:
            raise ValueError(
                f"Ground truth must match the image stem: {relative_image}"
            )
        category = sample.get("category")
        sample_id = sample.get("id")
        if (
            not isinstance(sample_id, int)
            or isinstance(sample_id, bool)
            or sample_id < 1
            or sample_id in declared_ids
        ):
            raise ValueError(f"Manifest sample {index} requires a unique positive ID")
        declared_ids.add(sample_id)
        if (
            not isinstance(category, str)
            or not category.strip()
            or category != category.strip()
        ):
            raise ValueError(f"Manifest sample {index} requires a category")
        if relative_image in samples_by_image:
            raise ValueError(f"Duplicate declared image: {relative_image}")
        if relative_ground_truth in declared_ground_truth:
            raise ValueError(
                f"Duplicate declared ground truth: {relative_ground_truth}"
            )
        samples_by_image[relative_image] = sample
        declared_ground_truth.add(relative_ground_truth)

    actual_images = {path.relative_to(eval_dir).as_posix() for path in image_paths}
    actual_ground_truth = {
        path.relative_to(eval_dir).as_posix() for path in ground_truth_paths
    }
    missing_images = sorted(set(samples_by_image) - actual_images)
    undeclared_images = sorted(actual_images - set(samples_by_image))
    missing_ground_truth = sorted(declared_ground_truth - actual_ground_truth)
    undeclared_ground_truth = sorted(actual_ground_truth - declared_ground_truth)
    if missing_images:
        raise FileNotFoundError(
            "Manifest references missing images: " + ", ".join(missing_images)
        )
    if undeclared_images:
        raise ValueError(
            "Images are missing from manifest.json: " + ", ".join(undeclared_images)
        )
    if missing_ground_truth:
        raise FileNotFoundError(
            "Manifest references missing ground truth: "
            + ", ".join(missing_ground_truth)
        )
    if undeclared_ground_truth:
        raise ValueError(
            "Ground-truth files are missing from manifest.json: "
            + ", ".join(undeclared_ground_truth)
        )

    content_digest = hashlib.sha256()
    files = []
    category_counts: Counter[str] = Counter()
    digit_samples = 0
    cjk_samples = 0
    table_samples = 0
    for image_path in image_paths:
        relative_image = image_path.relative_to(eval_dir).as_posix()
        sample = samples_by_image[relative_image]
        relative_ground_truth = str(sample["ground_truth"])
        gt_path = eval_dir / relative_ground_truth
        try:
            ground_truth = gt_path.read_text(encoding="utf-8")
        except (OSError, UnicodeError) as exc:
            raise ValueError(f"Ground truth is not valid UTF-8: {gt_path}") from exc
        if not ground_truth.strip():
            raise ValueError(f"Ground truth is empty: {gt_path}")
        try:
            from PIL import Image

            with Image.open(image_path) as image:
                image.verify()
                if image.width < 1 or image.height < 1:
                    raise ValueError("image has no pixels")
        except (OSError, ValueError) as exc:
            raise ValueError(f"Evaluation image is unreadable: {image_path}") from exc

        category_counts[str(sample["category"])] += 1
        digit_samples += int(
            any(
                character.isascii() and character.isdigit()
                for character in ground_truth
            )
        )
        cjk_samples += int(
            any(
                "\u3040" <= character <= "\u30ff"
                or "\u3400" <= character <= "\u9fff"
                or "\uac00" <= character <= "\ud7af"
                for character in ground_truth
            )
        )
        table_samples += int(
            "<table" in ground_truth.lower() or "| ---" in ground_truth
        )

        for path in (image_path, gt_path):
            relative = path.relative_to(eval_dir).as_posix()
            digest = sha256_file(path)
            size = path.stat().st_size
            content_digest.update(relative.encode("utf-8"))
            content_digest.update(b"\0")
            content_digest.update(digest.encode("ascii"))
            content_digest.update(b"\n")
            files.append({"path": relative, "size": size, "sha256": digest})

    content_digest.update(manifest_path.read_bytes())
    return {
        "name": eval_dir.name,
        "num_samples": len(image_paths),
        "num_digit_samples": digit_samples,
        "num_cjk_samples": cjk_samples,
        "num_table_samples": table_samples,
        "category_counts": dict(sorted(category_counts.items())),
        "manifest_sha256": sha256_file(manifest_path),
        "content_sha256": content_digest.hexdigest(),
        "files": files,
    }


def dataset_separation(
    calibration_dataset: dict,
    evaluation_dataset: dict,
) -> dict[str, Any]:
    """Describe whether model-selection images are absent from final evaluation."""
    if not isinstance(calibration_dataset, dict) or not isinstance(
        evaluation_dataset, dict
    ):
        raise TypeError("Dataset manifests must be dictionaries")

    def file_hashes(dataset: dict, prefix: str) -> set[str]:
        return {
            str(record["sha256"])
            for record in dataset.get("files", [])
            if isinstance(record, dict)
            and isinstance(record.get("path"), str)
            and record["path"].startswith(prefix)
            and isinstance(record.get("sha256"), str)
            and record["sha256"]
        }

    calibration_images = file_hashes(calibration_dataset, "images/")
    evaluation_images = file_hashes(evaluation_dataset, "images/")
    calibration_text = file_hashes(calibration_dataset, "ground_truth/")
    evaluation_text = file_hashes(evaluation_dataset, "ground_truth/")
    image_overlap = sorted(calibration_images & evaluation_images)
    ground_truth_overlap = sorted(calibration_text & evaluation_text)
    distinct_digests = (
        isinstance(calibration_dataset.get("content_sha256"), str)
        and isinstance(evaluation_dataset.get("content_sha256"), str)
        and calibration_dataset["content_sha256"]
        != evaluation_dataset["content_sha256"]
    )
    return {
        "passed": bool(calibration_images)
        and bool(evaluation_images)
        and bool(calibration_text)
        and bool(evaluation_text)
        and distinct_digests
        and not image_overlap
        and not ground_truth_overlap,
        "calibration_digest": calibration_dataset.get("content_sha256"),
        "evaluation_digest": evaluation_dataset.get("content_sha256"),
        "overlapping_image_sha256": image_overlap,
        "overlapping_ground_truth_sha256": ground_truth_overlap,
    }


def _pattern_is_within(child: str, parent: str) -> bool:
    """Return whether a precision rule is the parent itself or a descendant."""
    return child == parent or child.startswith(parent + ".")


def _finite_number(value: Any) -> float | None:
    if (
        not isinstance(value, (int, float))
        or isinstance(value, bool)
        or not math.isfinite(float(value))
    ):
        return None
    return float(value)


def generate_precision_map(
    base_map: dict,
    sensitivity_results: dict,
    *,
    calibration_results: dict | None = None,
    cer_threshold: float = 0.02,
    digit_cer_threshold: float = 0.02,
    table_degradation_threshold: float = 0.01,
) -> dict:
    """Turn group sensitivity measurements into executable top-level rules."""
    from quantization.layer_sensitivity import LAYER_GROUPS

    for name, value in (
        ("cer_threshold", cer_threshold),
        ("digit_cer_threshold", digit_cer_threshold),
        ("table_degradation_threshold", table_degradation_threshold),
    ):
        if _finite_number(value) is None or value < 0:
            raise ValueError(f"{name} must be a finite non-negative number")
    groups = sensitivity_results.get("layer_groups")
    if not isinstance(groups, dict) or not groups:
        raise ValueError("Sensitivity results contain no layer_groups")
    generated = {
        pattern: precision
        for pattern, precision in base_map.items()
        if not pattern.startswith("_")
    }
    decisions = {}
    for group_name, patterns in LAYER_GROUPS.items():
        result = groups.get(group_name)
        if not isinstance(result, dict):
            raise ValueError(f"Sensitivity result missing group: {group_name}")
        reasons = []
        if result.get("status") != "success":
            reasons.append(f"status={result.get('status', 'missing')}")
        for key, threshold in (
            ("cer_delta", cer_threshold),
            ("digit_cer_delta", digit_cer_threshold),
            ("table_score_degradation", table_degradation_threshold),
        ):
            value = _finite_number(result.get(key))
            if value is None:
                reasons.append(f"{key}=missing-or-non-finite")
            elif value > threshold:
                reasons.append(f"{key}={value:.6f}")
        measured_values = [
            _finite_number(result.get("cer_delta")),
            _finite_number(result.get("digit_cer_delta")),
            _finite_number(result.get("table_score_degradation")),
        ]
        explicit_non_degradation = result.get("status") == "success" and all(
            value is not None and value <= 0 for value in measured_values
        )
        if group_name in HARD_PROTECTED_GROUPS:
            reasons.append("protected OCR-critical group")
        elif group_name in EVIDENCE_PROTECTED_GROUPS and not explicit_non_degradation:
            reasons.append(
                "protected unless all measured quality deltas are non-degrading"
            )

        retain_bf16 = bool(reasons)
        if retain_bf16:
            for group_pattern in patterns:
                for existing_pattern in list(generated):
                    if _pattern_is_within(existing_pattern, group_pattern):
                        generated[existing_pattern] = "bfloat16"
                generated[group_pattern] = "bfloat16"
        elif group_name in EVIDENCE_PROTECTED_GROUPS and explicit_non_degradation:
            for group_pattern in patterns:
                for existing_pattern in list(generated):
                    if _pattern_is_within(existing_pattern, group_pattern):
                        generated[existing_pattern] = "mxfp8"
                generated[group_pattern] = "mxfp8"
        decisions[group_name] = {
            "precision": (
                "bfloat16"
                if retain_bf16
                else "mxfp8"
                if group_name in EVIDENCE_PROTECTED_GROUPS
                else "base-map"
            ),
            "reasons": reasons
            or (
                ["all measured quality deltas were non-degrading"]
                if group_name in EVIDENCE_PROTECTED_GROUPS
                else ["within thresholds"]
            ),
        }

    calibration_summary = None
    if calibration_results is not None:
        overrides = calibration_results.get("precision_overrides")
        if not isinstance(overrides, dict) or not overrides:
            raise ValueError("Calibration results contain no precision_overrides")
        expected_pattern = "language_model.lm_head"
        if (
            set(overrides) != {expected_pattern}
            or calibration_results.get("target_pattern") != expected_pattern
        ):
            raise ValueError("Calibration may override only language_model.lm_head")
        selected = calibration_results.get("selected")
        experiments = calibration_results.get("experiments")
        if (
            not isinstance(selected, dict)
            or not isinstance(experiments, list)
            or not experiments
        ):
            raise ValueError("Calibration selection evidence is incomplete")
        allowed_precisions = {"bfloat16", "mxfp8", "affine8"}
        for pattern, precision in overrides.items():
            if not isinstance(pattern, str) or precision not in allowed_precisions:
                raise ValueError(
                    f"Invalid calibrated precision override: {pattern}={precision}"
                )
            matching = [
                experiment
                for experiment in experiments
                if isinstance(experiment, dict)
                and experiment.get("label") == selected.get("label")
                and experiment.get("precision") == selected.get("precision")
                and experiment.get("passed") is True
            ]
            if selected.get("precision") != precision or len(matching) != 1:
                raise ValueError(
                    "Calibrated override does not match one passing experiment"
                )
            generated[pattern] = precision
            for group_name, patterns in LAYER_GROUPS.items():
                if pattern in patterns:
                    decisions[group_name] = {
                        "precision": precision,
                        "reasons": [
                            "selected by joint quality/throughput calibration: "
                            + str(calibration_results.get("selected", {}).get("label"))
                        ],
                    }
        calibration_summary = {
            "selected": calibration_results.get("selected"),
            "selection_policy": calibration_results.get("selection_policy"),
        }

    generated["_generated_from"] = {
        "source_model": sensitivity_results.get("model_path"),
        "thresholds": {
            "cer_delta": cer_threshold,
            "digit_cer_delta": digit_cer_threshold,
            "table_score_degradation": table_degradation_threshold,
        },
        "decisions": decisions,
        "calibration": calibration_summary,
    }
    return generated


def validate_candidate_metadata(model_dir: str | Path) -> dict:
    """Validate native Unlimited-OCR MXFP8 metadata."""
    model_dir = Path(model_dir)
    config = load_json_object(model_dir / "config.json")
    processor = load_json_object(model_dir / "processor_config.json")
    precision_map = load_json_object(model_dir / "precision_map.json")
    summary = load_json_object(model_dir / "quantization_summary.json")
    architectures = config.get("architectures")
    quantization = config.get("quantization") or config.get("quantization_config")
    text_config = config.get("text_config") or config.get("language_config") or config
    window = (
        text_config.get("sliding_window_size")
        if isinstance(text_config, dict)
        else None
    )
    if window is None and isinstance(text_config, dict):
        window = text_config.get("sliding_window")
    precision_rules = {
        pattern: precision
        for pattern, precision in precision_map.items()
        if not pattern.startswith("_")
    }
    quantized_modules = summary.get("quantized_modules")
    module_precisions = summary.get("quantized_module_precisions")
    precision_counts = summary.get("quantized_precision_counts")
    valid_modules = (
        isinstance(quantized_modules, list)
        and bool(quantized_modules)
        and all(isinstance(name, str) and bool(name) for name in quantized_modules)
        and len(set(quantized_modules)) == len(quantized_modules)
        and isinstance(module_precisions, dict)
        and set(module_precisions) == set(quantized_modules)
        and all(
            precision in {"mxfp8", "affine8"}
            for precision in module_precisions.values()
        )
    )
    actual_precision_counts = (
        dict(sorted(Counter(module_precisions.values()).items()))
        if valid_modules
        else None
    )
    tokenizer_config_path = model_dir / "tokenizer_config.json"
    tokenizer_processor_ok = True
    if tokenizer_config_path.is_file():
        tokenizer_config = load_json_object(tokenizer_config_path)
        tokenizer_processor_ok = (
            tokenizer_config.get("processor_class") == "UnlimitedOCRHFProcessor"
        )

    checks = {
        "architecture": isinstance(architectures, list)
        and "UnlimitedOCRForCausalLM" in architectures,
        "model_type": config.get("model_type") == "unlimited-ocr",
        "mxfp8": isinstance(quantization, dict) and quantization.get("mode") == "mxfp8",
        "sliding_window": isinstance(window, int)
        and not isinstance(window, bool)
        and window > 0,
        "processor_class": processor.get("processor_class")
        == "UnlimitedOCRHFProcessor"
        and tokenizer_processor_ok,
        "sft_format": processor.get("sft_format") == "unlimitedocr",
        "precision_map": bool(precision_rules)
        and all(
            isinstance(pattern, str)
            and bool(pattern)
            and precision in {"bfloat16", "mxfp8", "affine8"}
            for pattern, precision in precision_rules.items()
        ),
        "quantization_summary": summary.get("method") == "mxfp8"
        and summary.get("group_size") == 32
        and summary.get("bits") == 8
        and isinstance(summary.get("source_model"), str)
        and bool(summary["source_model"])
        and isinstance(summary.get("source_revision"), str)
        and bool(re.fullmatch(r"[0-9a-f]{40}", summary["source_revision"]))
        and summary.get("precision_map_sha256") == _json_digest(precision_map)
        and valid_modules
        and summary.get("quantized_module_count") == len(quantized_modules)
        and precision_counts == actual_precision_counts,
        "protected_modules_preserved": valid_modules
        and not any(
            name.startswith(("sam_model.", "vision_model."))
            or name.endswith(".mlp.gate")
            for name in quantized_modules
        ),
    }
    return {
        "passed": all(checks.values()),
        "checks": checks,
        "sliding_window": window,
        "source_model": summary.get("source_model"),
        "source_revision": summary.get("source_revision"),
    }


def _number(payload: dict, key: str) -> float | None:
    return _finite_number(payload.get(key))


def validate_release_thresholds(thresholds: dict | None) -> dict[str, float | int]:
    if thresholds is not None and not isinstance(thresholds, dict):
        raise TypeError("thresholds must be a dictionary or None")
    unexpected = sorted(set(thresholds or {}) - set(DEFAULT_THRESHOLDS))
    if unexpected:
        raise ValueError("Unknown release thresholds: " + ", ".join(unexpected))
    limits = {**DEFAULT_THRESHOLDS, **(thresholds or {})}
    for name, value in limits.items():
        if name == "min_rswa_tokens":
            if not isinstance(value, int) or isinstance(value, bool) or value < 1:
                raise ValueError("min_rswa_tokens must be a positive integer")
            continue
        number = _finite_number(value)
        if number is None or number < 0:
            raise ValueError(f"{name} must be a finite non-negative number")
        limits[name] = number
    if limits["max_rswa_repetition_rate"] > 1:
        raise ValueError("max_rswa_repetition_rate must be in [0, 1]")
    return limits


def _gate(name: str, actual: Any, limit: Any, passed: bool, detail: str) -> dict:
    return {
        "name": name,
        "passed": bool(passed),
        "actual": actual,
        "limit": limit,
        "detail": detail,
    }


def _positive_integer(payload: dict, key: str, *, minimum: int = 1) -> int | None:
    value = payload.get(key)
    if not isinstance(value, int) or isinstance(value, bool) or value < minimum:
        return None
    return value


def _accuracy_files(payload: dict) -> set[str] | None:
    per_file = payload.get("per_file")
    if not isinstance(per_file, list) or not per_file:
        return None
    names = [
        item.get("file")
        for item in per_file
        if isinstance(item, dict) and isinstance(item.get("file"), str)
    ]
    return set(names) if len(names) == len(per_file) == len(set(names)) else None


def _official_mlx_accuracy_recipe(payload: dict) -> bool:
    settings = payload.get("generation_settings")
    return (
        payload.get("backend") == "mlx"
        and payload.get("prompt") == "<image>document parsing."
        and payload.get("profile") == "accurate"
        and _positive_integer(payload, "max_tokens") is not None
        and isinstance(settings, dict)
        and _number(settings, "temperature") == 0.0
        and _number(settings, "top_p") == 1.0
        and _number(settings, "repetition_penalty") == 1.0
        and _positive_integer(settings, "no_repeat_ngram_size") == 35
        and _positive_integer(settings, "ngram_window") == 128
    )


def _performance_runs_complete(payload: dict) -> bool:
    num_runs = _positive_integer(payload, "num_runs", minimum=3)
    max_tokens = _positive_integer(payload, "max_tokens")
    runs = payload.get("runs")
    if (
        num_runs is None
        or max_tokens is None
        or not isinstance(runs, list)
        or len(runs) != num_runs
    ):
        return False
    return all(
        isinstance(run, dict)
        and _positive_integer(run, "run") == index
        and _positive_integer(run, "tokens_generated") == max_tokens
        and run.get("tokens_generated_source") == "mlx-vlm token count"
        and (_number(run, "tokens_per_second") or 0) > 0
        and run.get("tokens_per_second_source") == "mlx-vlm generation_tps"
        and (_number(run, "elapsed_seconds") or 0) > 0
        and run.get("finish_reason") == "length"
        for index, run in enumerate(runs, start=1)
    )


def _mean_matches(value: object, values: list[float]) -> bool:
    number = _finite_number(value)
    if number is None or not values:
        return False
    expected = sum(values) / len(values)
    return math.isclose(number, expected, rel_tol=1e-12, abs_tol=1e-12)


def _accuracy_aggregates_valid(payload: dict) -> bool:
    """Recompute every release-relevant accuracy aggregate from per-file rows."""
    per_file = payload.get("per_file")
    if not isinstance(per_file, list) or not per_file:
        return False
    cer_values = []
    digit_values = []
    cjk_values = []
    table_values = []
    repetition_values = []
    elapsed_values = []
    for item in per_file:
        if not isinstance(item, dict):
            return False
        cer = _number(item, "cer")
        digit_cer = _number(item, "digit_cer")
        cjk_cer = _number(item, "cjk_cer")
        repetition = _number(item, "repetition_rate")
        elapsed = _number(item, "elapsed_seconds")
        digit_count = item.get("ref_digit_count")
        cjk_count = item.get("ref_cjk_count")
        table_score = item.get("table_score")
        if (
            cer is None
            or cer < 0
            or digit_cer is None
            or digit_cer < 0
            or cjk_cer is None
            or cjk_cer < 0
            or repetition is None
            or not 0 <= repetition <= 1
            or elapsed is None
            or elapsed < 0
            or not isinstance(digit_count, int)
            or isinstance(digit_count, bool)
            or digit_count < 0
            or not isinstance(cjk_count, int)
            or isinstance(cjk_count, bool)
            or cjk_count < 0
        ):
            return False
        cer_values.append(cer)
        repetition_values.append(repetition)
        elapsed_values.append(elapsed)
        if digit_count > 0:
            digit_values.append(digit_cer)
        if cjk_count > 0:
            cjk_values.append(cjk_cer)
        if table_score is not None:
            score = _finite_number(table_score)
            if score is None or not 0 <= score <= 1:
                return False
            table_values.append(score)

    total_time = _number(payload, "total_time_seconds")
    return (
        payload.get("num_images") == len(per_file)
        and payload.get("num_samples") == len(per_file)
        and payload.get("num_digit_samples") == len(digit_values)
        and payload.get("num_cjk_samples") == len(cjk_values)
        and payload.get("num_table_samples") == len(table_values)
        and _mean_matches(payload.get("mean_cer"), cer_values)
        and _mean_matches(payload.get("mean_digit_cer"), digit_values)
        and _mean_matches(payload.get("mean_cjk_cer"), cjk_values)
        and _mean_matches(payload.get("mean_table_score"), table_values)
        and _mean_matches(payload.get("mean_repetition_rate"), repetition_values)
        and total_time is not None
        and math.isclose(
            total_time,
            sum(elapsed_values),
            rel_tol=1e-12,
            abs_tol=1e-12,
        )
    )


def _performance_aggregates_valid(payload: dict) -> bool:
    if not _performance_runs_complete(payload):
        return False
    runs = payload["runs"]
    tps_values = [_number(run, "tokens_per_second") for run in runs]
    elapsed_values = [_number(run, "elapsed_seconds") for run in runs]
    memory_values = [_number(run, "peak_memory_mb") for run in runs]
    if (
        any(value is None or value <= 0 for value in tps_values)
        or any(value is None or value <= 0 for value in elapsed_values)
        or any(value is None or value < 0 for value in memory_values)
    ):
        return False
    return (
        _mean_matches(payload.get("mean_tps"), tps_values)
        and _mean_matches(payload.get("mean_elapsed_seconds"), elapsed_values)
        and _mean_matches(payload.get("mean_peak_memory_mb"), memory_values)
    )


def _calibration_artifact_record(path: Path) -> dict:
    return {
        "filename": path.name,
        "size": path.stat().st_size,
        "sha256": sha256_file(path),
    }


def validate_calibration_results(
    calibration: dict,
    *,
    bf16_accuracy: dict,
    reference_performance: dict,
    calibration_dataset: dict,
    source_revision: str,
    reference_revision: str,
    evidence: dict[str, dict],
    evidence_paths: dict[str, str | Path],
    thresholds: dict | None = None,
) -> dict:
    """Recompute calibration selection from content-addressed raw evidence."""
    limits = validate_release_thresholds(thresholds)
    if not re.fullmatch(r"[0-9a-f]{40}", source_revision):
        raise ValueError("Calibration source revision must be a lowercase commit SHA")
    if not re.fullmatch(r"[0-9a-f]{40}", reference_revision):
        raise ValueError(
            "Calibration reference revision must be a lowercase commit SHA"
        )
    if calibration.get("schema_version") != 3:
        raise ValueError("Calibration schema_version 3 is required")
    if calibration.get("target_pattern") != "language_model.lm_head":
        raise ValueError("Calibration may target only language_model.lm_head")
    if calibration.get("thresholds") != limits:
        raise ValueError("Calibration thresholds do not match release thresholds")
    if calibration.get("selection_policy") != (
        "fastest candidate passing existing quality and throughput limits"
    ):
        raise ValueError("Calibration selection policy is invalid")

    expected_input_keys = {
        "calibration_baseline_accuracy",
        "calibration_reference_performance",
        *CALIBRATION_RAW_EVIDENCE_NAMES,
    }
    input_artifacts = calibration.get("input_artifacts")
    if (
        not isinstance(input_artifacts, dict)
        or set(input_artifacts) != expected_input_keys
    ):
        raise ValueError("Calibration input artifacts are incomplete")
    for name in expected_input_keys:
        path = Path(evidence_paths[name])
        if path.is_symlink() or not path.is_file():
            raise FileNotFoundError(f"Calibration input is missing: {name}")
        if input_artifacts.get(name) != _calibration_artifact_record(path):
            raise ValueError(f"Calibration input hash does not match: {name}")
    if (
        evidence.get("calibration_baseline_accuracy") != bf16_accuracy
        or evidence.get("calibration_reference_performance")
        != reference_performance
    ):
        raise ValueError("Calibration baselines do not match raw evidence")

    if not isinstance(calibration_dataset, dict) or calibration.get(
        "dataset"
    ) != calibration_dataset:
        raise ValueError("Calibration dataset provenance does not match")
    expected_files = {
        Path(record["path"]).name
        for record in calibration_dataset.get("files", [])
        if isinstance(record, dict)
        and isinstance(record.get("path"), str)
        and record["path"].startswith("images/")
    }
    if (
        not expected_files
        or len(expected_files) != calibration_dataset.get("num_samples")
        or _accuracy_files(bf16_accuracy) != expected_files
        or bf16_accuracy.get("num_images") != calibration_dataset.get("num_samples")
        or bf16_accuracy.get("num_samples") != calibration_dataset.get("num_samples")
        or bf16_accuracy.get("num_digit_samples")
        != calibration_dataset.get("num_digit_samples")
        or bf16_accuracy.get("num_cjk_samples")
        != calibration_dataset.get("num_cjk_samples")
        or bf16_accuracy.get("num_table_samples")
        != calibration_dataset.get("num_table_samples")
    ):
        raise ValueError("Calibration baseline does not match its dataset")

    baseline_values = {
        "mean_cer": _number(bf16_accuracy, "mean_cer"),
        "mean_digit_cer": _number(bf16_accuracy, "mean_digit_cer"),
        "mean_table_score": _number(bf16_accuracy, "mean_table_score"),
    }
    reference_tps = _number(reference_performance, "mean_tps")
    if (
        any(value is None for value in baseline_values.values())
        or reference_tps is None
        or reference_tps <= 0
        or not _accuracy_aggregates_valid(bf16_accuracy)
        or not _performance_aggregates_valid(reference_performance)
        or bf16_accuracy.get("served_revision") != source_revision
        or reference_performance.get("served_revision") != reference_revision
    ):
        raise ValueError("Calibration baselines are incomplete or inconsistent")

    experiments = calibration.get("experiments")
    if not isinstance(experiments, list) or len(experiments) != len(
        CALIBRATION_EVIDENCE_KEYS
    ):
        raise ValueError("Calibration must contain exactly three experiments")
    stored_by_precision = {}
    labels = set()
    for experiment in experiments:
        if not isinstance(experiment, dict):
            raise ValueError("Calibration experiment must be an object")
        precision = experiment.get("precision")
        label = experiment.get("label")
        if (
            precision not in CALIBRATION_EVIDENCE_KEYS
            or precision in stored_by_precision
            or not isinstance(label, str)
            or not label
            or label != label.strip()
            or label in labels
        ):
            raise ValueError("Calibration experiment identities are invalid")
        stored_by_precision[precision] = experiment
        labels.add(label)
    if set(stored_by_precision) != set(CALIBRATION_EVIDENCE_KEYS):
        raise ValueError("Calibration precision coverage is incomplete")

    expected_experiments = []
    model_paths = set()
    accuracy_recipe_keys = ("prompt", "max_tokens", "profile", "generation_settings")
    performance_recipe_keys = (
        "image_path",
        "prompt",
        "max_tokens",
        "num_warmup",
        "num_runs",
        "system",
    )
    baseline_files = _accuracy_files(bf16_accuracy)
    for precision, (accuracy_key, performance_key) in CALIBRATION_EVIDENCE_KEYS.items():
        accuracy = evidence.get(accuracy_key)
        performance = evidence.get(performance_key)
        if not isinstance(accuracy, dict) or not isinstance(performance, dict):
            raise ValueError(f"Calibration raw evidence is missing for {precision}")
        model_path = accuracy.get("model_path")
        if (
            not isinstance(model_path, str)
            or not model_path
            or performance.get("model_path") != model_path
            or performance.get("served_revision")
            != accuracy.get("served_revision")
            or model_path in model_paths
        ):
            raise ValueError(f"Calibration model identity is invalid for {precision}")
        model_paths.add(model_path)
        if (
            not _official_mlx_accuracy_recipe(accuracy)
            or not _accuracy_aggregates_valid(accuracy)
            or _accuracy_files(accuracy) != baseline_files
            or any(
                accuracy.get(key) != bf16_accuracy.get(key)
                for key in accuracy_recipe_keys
            )
            or any(
                accuracy.get(key) != bf16_accuracy.get(key)
                for key in (
                    "num_images",
                    "num_samples",
                    "num_digit_samples",
                    "num_cjk_samples",
                    "num_table_samples",
                )
            )
        ):
            raise ValueError(f"Calibration accuracy recipe is invalid for {precision}")
        if not _performance_aggregates_valid(performance) or any(
            performance.get(key) != reference_performance.get(key)
            for key in performance_recipe_keys
        ):
            raise ValueError(
                f"Calibration performance recipe is invalid for {precision}"
            )

        candidate_cer = _number(accuracy, "mean_cer")
        candidate_digit = _number(accuracy, "mean_digit_cer")
        candidate_table = _number(accuracy, "mean_table_score")
        candidate_tps = _number(performance, "mean_tps")
        if None in (candidate_cer, candidate_digit, candidate_table, candidate_tps):
            raise ValueError(f"Calibration metrics are incomplete for {precision}")
        deltas = {
            "cer_vs_bf16": candidate_cer - baseline_values["mean_cer"],
            "digit_cer_vs_bf16": (candidate_digit - baseline_values["mean_digit_cer"]),
            "table_degradation_vs_bf16": (
                baseline_values["mean_table_score"] - candidate_table
            ),
            "tps_ratio_vs_reference": candidate_tps / reference_tps,
        }
        checks = {
            "cer": deltas["cer_vs_bf16"] <= limits["max_cer_delta_vs_bf16"],
            "digit_cer": deltas["digit_cer_vs_bf16"]
            <= limits["max_digit_cer_delta_vs_bf16"],
            "table_score": deltas["table_degradation_vs_bf16"]
            <= limits["max_table_score_degradation_vs_bf16"],
            "throughput": deltas["tps_ratio_vs_reference"]
            >= limits["min_tps_ratio_vs_reference"],
        }
        expected = {
            "label": stored_by_precision[precision]["label"],
            "precision": precision,
            "passed": all(checks.values()),
            "checks": checks,
            "metrics": {
                "mean_cer": candidate_cer,
                "mean_digit_cer": candidate_digit,
                "mean_table_score": candidate_table,
                "mean_tps": candidate_tps,
            },
            "deltas": deltas,
        }
        if stored_by_precision[precision] != expected:
            raise ValueError(f"Calibration claims do not recompute for {precision}")
        expected_experiments.append(expected)

    passing = [
        experiment for experiment in expected_experiments if experiment["passed"]
    ]
    if not passing:
        raise ValueError("No calibration experiment passes every release limit")
    selected_experiment = max(
        passing,
        key=lambda experiment: experiment["metrics"]["mean_tps"],
    )
    selected = {
        "label": selected_experiment["label"],
        "precision": selected_experiment["precision"],
    }
    if calibration.get("selected") != selected:
        raise ValueError("Calibration did not select the fastest passing experiment")
    if calibration.get("precision_overrides") != {
        "language_model.lm_head": selected["precision"]
    }:
        raise ValueError("Calibration precision override does not match selection")
    return {
        "selected": selected,
        "models": sorted(model_paths),
        "dataset_digest": calibration_dataset.get("content_sha256"),
        "input_artifacts": input_artifacts,
    }


def evaluate_release_gates(
    *,
    candidate_weights: dict,
    reference_weights: dict,
    dataset: dict,
    calibration_dataset: dict,
    metadata: dict,
    bf16_accuracy: dict,
    reference_accuracy: dict,
    candidate_accuracy: dict,
    reference_performance: dict,
    candidate_performance: dict,
    rswa: dict,
    provenance: dict,
    source_model_name: str,
    source_id: str,
    source_config_sha256: str,
    source_revision: str,
    reference_id: str,
    reference_revision: str,
    candidate_model_name: str,
    repo_id: str,
    thresholds: dict | None = None,
) -> list[dict]:
    """Evaluate every required release gate; missing values fail closed."""
    limits = validate_release_thresholds(thresholds)
    gates = []
    candidate_digest = candidate_weights.get("aggregate_sha256")
    reference_digest = reference_weights.get("aggregate_sha256")
    gates.append(
        _gate(
            "weights_are_distinct",
            candidate_digest,
            f"different from {reference_digest}",
            bool(
                candidate_digest
                and reference_digest
                and candidate_digest != reference_digest
            ),
            "Candidate aggregate digest must differ from the Sahil reference",
        )
    )
    candidate_size = _number(candidate_weights, "total_size_gb")
    gates.append(
        _gate(
            "weight_size_gb",
            candidate_size,
            limits["max_weight_size_gb"],
            candidate_size is not None
            and candidate_size <= limits["max_weight_size_gb"],
            "Candidate Safetensors size",
        )
    )
    gates.append(
        _gate(
            "native_model_metadata",
            metadata.get("checks"),
            True,
            metadata.get("passed") is True,
            "Native Unlimited-OCR, MXFP8, and R-SWA metadata",
        )
    )
    candidate_source = {
        "model": metadata.get("source_model"),
        "revision": metadata.get("source_revision"),
    }
    gates.append(
        _gate(
            "candidate_source_provenance",
            candidate_source,
            {
                "model": [source_id, source_model_name],
                "revision": source_revision,
            },
            candidate_source["model"] in {source_id, source_model_name}
            and candidate_source["revision"] == source_revision,
            "Converted metadata must bind the exact BF16 source commit",
        )
    )

    coverage = {
        "samples": dataset.get("num_samples", 0),
        "digit": dataset.get("num_digit_samples", 0),
        "cjk": dataset.get("num_cjk_samples", 0),
        "table": dataset.get("num_table_samples", 0),
    }
    gates.append(
        _gate(
            "evaluation_coverage",
            coverage,
            "all counts > 0",
            all(
                isinstance(value, int) and not isinstance(value, bool) and value > 0
                for value in coverage.values()
            ),
            "Dataset must cover ordinary text, digits, CJK, and tables",
        )
    )
    separation = dataset_separation(calibration_dataset, dataset)
    gates.append(
        _gate(
            "held_out_evaluation_dataset",
            separation,
            "distinct dataset digests and no shared image or ground-truth hashes",
            separation["passed"],
            "Final quality evidence must not reuse model-selection samples",
        )
    )
    expected_files = {
        Path(item["path"]).name
        for item in dataset.get("files", [])
        if isinstance(item, dict)
        and isinstance(item.get("path"), str)
        and item["path"].startswith("images/")
    }
    accuracy_payloads = (
        bf16_accuracy,
        reference_accuracy,
        candidate_accuracy,
    )
    gates.append(
        _gate(
            "accuracy_aggregates_recomputed",
            [_accuracy_aggregates_valid(payload) for payload in accuracy_payloads],
            [True, True, True],
            all(_accuracy_aggregates_valid(payload) for payload in accuracy_payloads),
            "Every reported accuracy aggregate must recompute from per-file rows",
        )
    )
    sample_counts = [payload.get("num_samples") for payload in accuracy_payloads]
    evaluated_files = [_accuracy_files(payload) for payload in accuracy_payloads]
    gates.append(
        _gate(
            "same_evaluation_samples",
            {
                "counts": sample_counts,
                "file_counts": [len(files or set()) for files in evaluated_files],
            },
            {"count": dataset.get("num_samples"), "files": sorted(expected_files)},
            bool(expected_files)
            and len(expected_files) == dataset.get("num_samples")
            and all(count == dataset.get("num_samples") for count in sample_counts)
            and all(
                payload.get("num_images") == dataset.get("num_samples")
                for payload in accuracy_payloads
            )
            and all(files == expected_files for files in evaluated_files),
            "All three checkpoints must run every identical evaluation file",
        )
    )
    model_identities = {
        "bf16": bf16_accuracy.get("model_path"),
        "reference": reference_accuracy.get("model_path"),
        "candidate": candidate_accuracy.get("model_path"),
        "reference_performance": reference_performance.get("model_path"),
        "candidate_performance": candidate_performance.get("model_path"),
        "rswa": rswa.get("model_path"),
    }
    gates.append(
        _gate(
            "model_identities",
            model_identities,
            {
                "bf16": source_model_name,
                "reference": reference_id,
                "candidate": candidate_model_name,
            },
            model_identities
            == {
                "bf16": source_model_name,
                "reference": reference_id,
                "candidate": candidate_model_name,
                "reference_performance": reference_id,
                "candidate_performance": candidate_model_name,
                "rswa": candidate_model_name,
            },
            "Accuracy, performance, and R-SWA evidence must identify exact models",
        )
    )
    model_revisions = {
        "bf16_accuracy": bf16_accuracy.get("served_revision"),
        "reference_accuracy": reference_accuracy.get("served_revision"),
        "candidate_accuracy": candidate_accuracy.get("served_revision"),
        "reference_performance": reference_performance.get("served_revision"),
        "candidate_performance": candidate_performance.get("served_revision"),
    }
    expected_revisions = {
        "bf16_accuracy": source_revision,
        "reference_accuracy": reference_revision,
        "candidate_accuracy": None,
        "reference_performance": reference_revision,
        "candidate_performance": None,
    }
    gates.append(
        _gate(
            "immutable_model_revisions",
            model_revisions,
            expected_revisions,
            bool(re.fullmatch(r"[0-9a-f]{40}", source_revision))
            and bool(re.fullmatch(r"[0-9a-f]{40}", reference_revision))
            and model_revisions == expected_revisions,
            "Remote source and reference evidence must name exact Hub commits",
        )
    )
    accuracy_recipe_keys = ("prompt", "max_tokens", "profile", "generation_settings")
    gates.append(
        _gate(
            "same_accuracy_recipe",
            {
                key: [payload.get(key) for payload in accuracy_payloads]
                for key in accuracy_recipe_keys
            },
            "identical official MLX OCR recipe",
            all(_official_mlx_accuracy_recipe(payload) for payload in accuracy_payloads)
            and all(
                payload.get(key) == bf16_accuracy.get(key)
                for payload in accuracy_payloads[1:]
                for key in accuracy_recipe_keys
            ),
            "Accuracy runs must use the same deterministic OCR generation settings",
        )
    )

    bf16_cer = _number(bf16_accuracy, "mean_cer")
    reference_cer = _number(reference_accuracy, "mean_cer")
    candidate_cer = _number(candidate_accuracy, "mean_cer")
    delta_bf16 = (
        candidate_cer - bf16_cer
        if candidate_cer is not None and bf16_cer is not None
        else None
    )
    delta_reference = (
        candidate_cer - reference_cer
        if candidate_cer is not None and reference_cer is not None
        else None
    )
    gates.append(
        _gate(
            "candidate_cer_vs_bf16",
            delta_bf16,
            limits["max_cer_delta_vs_bf16"],
            delta_bf16 is not None and delta_bf16 <= limits["max_cer_delta_vs_bf16"],
            "Candidate minus BF16 absolute mean CER",
        )
    )
    gates.append(
        _gate(
            "candidate_cer_vs_reference",
            delta_reference,
            limits["max_cer_delta_vs_reference"],
            delta_reference is not None
            and delta_reference <= limits["max_cer_delta_vs_reference"],
            "Candidate minus Sahil-reference absolute mean CER",
        )
    )

    bf16_digit = _number(bf16_accuracy, "mean_digit_cer")
    candidate_digit = _number(candidate_accuracy, "mean_digit_cer")
    digit_delta = (
        candidate_digit - bf16_digit
        if candidate_digit is not None and bf16_digit is not None
        else None
    )
    gates.append(
        _gate(
            "candidate_digit_cer_vs_bf16",
            digit_delta,
            limits["max_digit_cer_delta_vs_bf16"],
            digit_delta is not None
            and digit_delta <= limits["max_digit_cer_delta_vs_bf16"],
            "Candidate minus BF16 digit CER",
        )
    )

    bf16_table = _number(bf16_accuracy, "mean_table_score")
    candidate_table = _number(candidate_accuracy, "mean_table_score")
    table_degradation = (
        bf16_table - candidate_table
        if bf16_table is not None and candidate_table is not None
        else None
    )
    gates.append(
        _gate(
            "candidate_table_score_vs_bf16",
            table_degradation,
            limits["max_table_score_degradation_vs_bf16"],
            table_degradation is not None
            and table_degradation <= limits["max_table_score_degradation_vs_bf16"],
            "BF16 minus candidate mean table score",
        )
    )

    performance_keys = (
        "image_path",
        "prompt",
        "max_tokens",
        "num_warmup",
        "num_runs",
        "system",
    )
    gates.append(
        _gate(
            "performance_aggregates_recomputed",
            {
                "reference": _performance_aggregates_valid(reference_performance),
                "candidate": _performance_aggregates_valid(candidate_performance),
            },
            {"reference": True, "candidate": True},
            _performance_aggregates_valid(reference_performance)
            and _performance_aggregates_valid(candidate_performance),
            "Performance means must recompute from complete benchmark runs",
        )
    )
    gates.append(
        _gate(
            "same_performance_setup",
            {
                key: [reference_performance.get(key), candidate_performance.get(key)]
                for key in performance_keys
            },
            "identical setup with at least three complete runs",
            all(
                reference_performance.get(key) == candidate_performance.get(key)
                for key in performance_keys
            )
            and reference_performance.get("prompt") == "<image>document parsing."
            and _positive_integer(reference_performance, "max_tokens") is not None
            and _positive_integer(reference_performance, "num_warmup") is not None
            and _performance_runs_complete(reference_performance)
            and _performance_runs_complete(candidate_performance)
            and isinstance(reference_performance.get("system"), dict)
            and bool(reference_performance["system"]),
            "Performance comparisons must use the same host and benchmark recipe",
        )
    )

    reference_tps = _number(reference_performance, "mean_tps")
    candidate_tps = _number(candidate_performance, "mean_tps")
    tps_ratio = (
        candidate_tps / reference_tps
        if candidate_tps is not None and reference_tps and reference_tps > 0
        else None
    )
    gates.append(
        _gate(
            "candidate_tps_vs_reference",
            tps_ratio,
            limits["min_tps_ratio_vs_reference"],
            tps_ratio is not None and tps_ratio >= limits["min_tps_ratio_vs_reference"],
            "Candidate decode throughput divided by Sahil-reference throughput",
        )
    )
    rswa_results = rswa.get("test_results")
    long_results = (
        [
            result
            for result in rswa_results
            if isinstance(rswa_results, list)
            and isinstance(result, dict)
            and _positive_integer(result, "max_tokens") == limits["min_rswa_tokens"]
        ]
        if isinstance(rswa_results, list)
        else []
    )
    long_result = long_results[0] if len(long_results) == 1 else {}
    repetition_rate = _number(long_result, "repetition_rate")
    forced_tokens = _positive_integer(rswa, "force_min_tokens")
    generated_tokens = _positive_integer(long_result, "tokens_generated")
    pass_conditions = rswa.get("pass_conditions")
    rswa_generation = rswa.get("generation_settings")
    gates.append(
        _gate(
            "rswa_8k_bounded",
            {
                "pass_conditions": pass_conditions,
                "tokens": long_result.get("tokens_generated"),
                "repetition_rate": repetition_rate,
            },
            {
                "min_tokens": limits["min_rswa_tokens"],
                "max_repetition_rate": limits["max_rswa_repetition_rate"],
            },
            rswa.get("passed") is True
            and isinstance(pass_conditions, dict)
            and pass_conditions.get("cache_bounded") is True
            and pass_conditions.get("tps_stable") is True
            and pass_conditions.get("8k_test_passed") is True
            and rswa.get("prompt") == "<image>document parsing."
            and isinstance(rswa_generation, dict)
            and _number(rswa_generation, "temperature") == 0.0
            and _positive_integer(rswa_generation, "no_repeat_ngram_size") == 35
            and _positive_integer(rswa_generation, "ngram_window") == 128
            and forced_tokens is not None
            and forced_tokens >= limits["min_rswa_tokens"]
            and len(long_results) == 1
            and long_result.get("status") == "success"
            and long_result.get("tokens_generated_reliable") is True
            and generated_tokens is not None
            and generated_tokens >= limits["min_rswa_tokens"]
            and repetition_rate is not None
            and 0 <= repetition_rate <= limits["max_rswa_repetition_rate"],
            "8K generation must have bounded cache, stable throughput, and repetition",
        )
    )
    provenance_dataset = provenance.get("evaluation_dataset")
    provenance_calibration_dataset = provenance.get("calibration_dataset")
    source_config = provenance.get("source_config")
    gates.append(
        _gate(
            "provenance_matches_release",
            {
                "source": provenance.get("source_model"),
                "reference": provenance.get("reference_model"),
                "target": provenance.get("target_repo"),
                "source_config": source_config,
                "source_revision": provenance.get("source_revision"),
                "reference_revision": provenance.get("reference_revision"),
                "dataset_digest": (
                    provenance_dataset.get("content_sha256")
                    if isinstance(provenance_dataset, dict)
                    else None
                ),
                "calibration_dataset_digest": (
                    provenance_calibration_dataset.get("content_sha256")
                    if isinstance(provenance_calibration_dataset, dict)
                    else None
                ),
            },
            {
                "source": source_id,
                "reference": reference_id,
                "target": repo_id,
                "source_config_sha256": source_config_sha256,
                "source_revision": source_revision,
                "reference_revision": reference_revision,
                "dataset_digest": dataset.get("content_sha256"),
                "calibration_dataset_digest": calibration_dataset.get(
                    "content_sha256"
                ),
            },
            provenance.get("source_model") == source_id
            and provenance.get("reference_model") == reference_id
            and provenance.get("target_repo") == repo_id
            and provenance.get("source_revision") == source_revision
            and provenance.get("reference_revision") == reference_revision
            and isinstance(source_config, dict)
            and source_config.get("path") == "config.json"
            and source_config.get("sha256") == source_config_sha256
            and provenance_dataset == dataset
            and provenance_calibration_dataset == calibration_dataset,
            "Provenance must bind the source, selection/evaluation datasets, and target",
        )
    )
    return gates


def _metric_summary(payload: dict) -> dict:
    keys = (
        "model_path",
        "num_samples",
        "mean_cer",
        "mean_digit_cer",
        "mean_cjk_cer",
        "mean_table_score",
        "mean_tps",
        "mean_peak_memory_mb",
        "max_tokens",
        "profile",
    )
    return {key: payload.get(key) for key in keys if key in payload}


def _runtime_versions() -> dict:
    versions = {}
    for distribution in ("mlx", "mlx-vlm", "huggingface-hub", "numpy", "Pillow"):
        try:
            versions[distribution] = importlib.metadata.version(distribution)
        except importlib.metadata.PackageNotFoundError:
            versions[distribution] = None
    return versions


def build_release_manifest(
    *,
    candidate_dir: str | Path,
    reference_dir: str | Path,
    source_dir: str | Path,
    calibration_dir: str | Path,
    eval_dir: str | Path,
    evidence_paths: dict[str, str | Path],
    repo_id: str,
    source_id: str,
    source_revision: str,
    reference_id: str,
    reference_revision: str,
    thresholds: dict | None = None,
) -> dict:
    """Build a complete release decision from on-disk evidence."""
    required = RELEASE_EVIDENCE_NAMES
    if not isinstance(evidence_paths, dict):
        raise TypeError("evidence_paths must be a dictionary")
    missing = sorted(required - set(evidence_paths))
    if missing:
        raise ValueError("Missing release evidence: " + ", ".join(missing))
    unexpected = sorted(set(evidence_paths) - required)
    if unexpected:
        raise ValueError("Unexpected release evidence: " + ", ".join(unexpected))
    evidence_files = [Path(path) for path in evidence_paths.values()]
    if len({path.name for path in evidence_files}) != len(evidence_files):
        raise ValueError("Release evidence filenames must be unique")
    symlink_evidence = [path.name for path in evidence_files if path.is_symlink()]
    if symlink_evidence:
        raise ValueError(
            "Release evidence must not be symbolic links: "
            + ", ".join(symlink_evidence)
        )
    limits = validate_release_thresholds(thresholds)
    evidence = {name: load_json_object(path) for name, path in evidence_paths.items()}

    candidate_weights = model_weight_manifest(candidate_dir)
    # Hugging Face snapshot directories use content-addressed links into the
    # immutable blob cache. Hash their resolved bytes, while keeping candidate
    # and source checkpoints link-free.
    reference_weights = model_weight_manifest(reference_dir, allow_symlinks=True)
    source_weights = model_weight_manifest(source_dir)
    calibration_dataset = dataset_manifest(calibration_dir)
    dataset = dataset_manifest(eval_dir)
    metadata = validate_candidate_metadata(candidate_dir)
    source_config_path = Path(source_dir) / "config.json"
    if source_config_path.is_symlink() or not source_config_path.is_file():
        raise FileNotFoundError(
            f"Source config must be a regular file: {source_config_path}"
        )
    source_config_sha256 = sha256_file(source_config_path)
    candidate_files = candidate_metadata_manifest(candidate_dir)
    project_files = release_files_manifest(
        PROJECT_ROOT,
        RELEASE_PROJECT_FILES,
        required=True,
    )
    gates = evaluate_release_gates(
        candidate_weights=candidate_weights,
        reference_weights=reference_weights,
        dataset=dataset,
        calibration_dataset=calibration_dataset,
        metadata=metadata,
        bf16_accuracy=evidence["bf16_accuracy"],
        reference_accuracy=evidence["reference_accuracy"],
        candidate_accuracy=evidence["candidate_accuracy"],
        reference_performance=evidence["reference_performance"],
        candidate_performance=evidence["candidate_performance"],
        rswa=evidence["candidate_rswa"],
        provenance=evidence["provenance"],
        source_model_name=Path(source_dir).name,
        source_id=source_id,
        source_config_sha256=source_config_sha256,
        source_revision=source_revision,
        reference_id=reference_id,
        reference_revision=reference_revision,
        candidate_model_name=Path(candidate_dir).name,
        repo_id=repo_id,
        thresholds=limits,
    )
    calibration_validation = validate_calibration_results(
        evidence["calibration_results"],
        bf16_accuracy=evidence["calibration_baseline_accuracy"],
        reference_performance=evidence["calibration_reference_performance"],
        calibration_dataset=calibration_dataset,
        source_revision=source_revision,
        reference_revision=reference_revision,
        evidence=evidence,
        evidence_paths=evidence_paths,
        thresholds=limits,
    )
    gates.append(
        _gate(
            "calibration_recomputed",
            calibration_validation,
            "content-addressed inputs and fastest passing experiment",
            True,
            "Calibration claims must recompute from raw benchmark evidence",
        )
    )
    sensitivity = evidence["sensitivity_results"]
    sensitivity_baseline = sensitivity.get("baseline")
    calibration_baseline = evidence["calibration_baseline_accuracy"]
    sensitivity_metric_keys = (
        "num_samples",
        "num_digit_samples",
        "num_table_samples",
        "mean_cer",
        "mean_digit_cer",
        "mean_table_score",
    )
    sensitivity_matches_calibration = (
        isinstance(sensitivity_baseline, dict)
        and sensitivity.get("dataset") == calibration_dataset
        and sensitivity.get("prompt") == calibration_baseline.get("prompt")
        and sensitivity.get("max_tokens") == calibration_baseline.get("max_tokens")
        and all(
            sensitivity_baseline.get(key) == calibration_baseline.get(key)
            for key in sensitivity_metric_keys
        )
    )
    gates.append(
        _gate(
            "sensitivity_matches_calibration_dataset",
            {
                "dataset_digest": (
                    sensitivity.get("dataset", {}).get("content_sha256")
                    if isinstance(sensitivity.get("dataset"), dict)
                    else None
                ),
                "baseline_metrics_match": sensitivity_matches_calibration,
            },
            {
                "dataset_digest": calibration_dataset.get("content_sha256"),
                "baseline_metrics_match": True,
            },
            sensitivity_matches_calibration,
            "Sensitivity decisions must use the recorded selection dataset and baseline",
        )
    )
    candidate_precision_sha = sha256_file(Path(candidate_dir) / "precision_map.json")
    evidence_precision_sha = sha256_file(evidence_paths["generated_precision_map"])
    gates.append(
        _gate(
            "candidate_precision_map_matches_evidence",
            candidate_precision_sha,
            evidence_precision_sha,
            candidate_precision_sha == evidence_precision_sha,
            "The executable candidate precision map must equal approved evidence",
        )
    )
    evidence_map = evidence["generated_precision_map"]
    map_thresholds = (
        evidence_map.get("_generated_from", {}).get("thresholds")
        if isinstance(evidence_map.get("_generated_from"), dict)
        else None
    )
    reproduce_kwargs: dict[str, Any] = {}
    if isinstance(map_thresholds, dict):
        if map_thresholds.get("cer_delta") is not None:
            reproduce_kwargs["cer_threshold"] = map_thresholds["cer_delta"]
        if map_thresholds.get("digit_cer_delta") is not None:
            reproduce_kwargs["digit_cer_threshold"] = map_thresholds[
                "digit_cer_delta"
            ]
        if map_thresholds.get("table_score_degradation") is not None:
            reproduce_kwargs["table_degradation_threshold"] = map_thresholds[
                "table_score_degradation"
            ]
    reproduced_precision_map = generate_precision_map(
        load_json_object(PROJECT_ROOT / "quantization/precision_map.json"),
        evidence["sensitivity_results"],
        calibration_results=evidence["calibration_results"],
        **reproduce_kwargs,
    )
    gates.append(
        _gate(
            "precision_map_reproducible",
            _json_digest(evidence["generated_precision_map"]),
            _json_digest(reproduced_precision_map),
            evidence["generated_precision_map"] == reproduced_precision_map,
            "Sensitivity and calibration evidence must reproduce the executable map",
        )
    )
    actual_gate_names = tuple(gate.get("name") for gate in gates)
    if actual_gate_names != MLX_RELEASE_GATE_NAMES:
        raise RuntimeError(
            "MLX release gate implementation does not match its publication contract"
        )
    artifact_hashes = {
        name: {
            "filename": Path(path).name,
            "size": Path(path).stat().st_size,
            "sha256": sha256_file(path),
        }
        for name, path in evidence_paths.items()
    }
    return {
        "schema_version": MLX_RELEASE_SCHEMA_VERSION,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "release_approved": all(gate["passed"] for gate in gates),
        "repo_id": repo_id,
        "source": {
            "id": source_id,
            "resolved_revision": source_revision,
            "weights": source_weights,
        },
        "reference": {
            "id": reference_id,
            "resolved_revision": reference_revision,
            "weights": reference_weights,
        },
        "candidate": {
            "name": Path(candidate_dir).name,
            "weights": candidate_weights,
            "files": candidate_files,
            "metadata": metadata,
        },
        "project_files": project_files,
        "calibration_dataset": calibration_dataset,
        "evaluation_dataset": dataset,
        "thresholds": limits,
        "metrics": {
            name: _metric_summary(evidence[name])
            for name in (
                "bf16_accuracy",
                "reference_accuracy",
                "candidate_accuracy",
                "reference_performance",
                "candidate_performance",
            )
        },
        "rswa": evidence["candidate_rswa"],
        "sensitivity": {
            "baseline": evidence["sensitivity_results"].get("baseline"),
            "groups": evidence["sensitivity_results"].get("layer_groups"),
        },
        "calibration": evidence["calibration_results"],
        "precision_map": evidence["generated_precision_map"],
        "artifacts": artifact_hashes,
        "gates": gates,
        "environment": {
            "platform": platform.platform(),
            "machine": platform.machine(),
            "python": platform.python_version(),
            "versions": _runtime_versions(),
        },
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="Build a fail-closed release manifest")
    parser.add_argument("--candidate-dir", required=True, type=Path)
    parser.add_argument("--reference-dir", required=True, type=Path)
    parser.add_argument("--source-dir", required=True, type=Path)
    parser.add_argument("--calibration-dir", required=True, type=Path)
    parser.add_argument("--eval-dir", required=True, type=Path)
    parser.add_argument("--artifacts-dir", required=True, type=Path)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--repo-id", required=True)
    parser.add_argument("--source-id", default="baidu/Unlimited-OCR")
    parser.add_argument("--source-revision", required=True)
    parser.add_argument(
        "--reference-id", default="sahilchachra/unlimited-ocr-mxfp8-mlx"
    )
    parser.add_argument("--reference-revision", required=True)
    args = parser.parse_args()

    evidence_paths = {
        name: args.artifacts_dir / f"{name}.json"
        for name in RELEASE_EVIDENCE_NAMES - CALIBRATION_RAW_EVIDENCE_NAMES
    }
    calibration = load_json_object(evidence_paths["calibration_results"])
    evidence_paths.update(
        calibration_raw_evidence_paths(args.artifacts_dir, calibration)
    )
    manifest = build_release_manifest(
        candidate_dir=args.candidate_dir,
        reference_dir=args.reference_dir,
        source_dir=args.source_dir,
        calibration_dir=args.calibration_dir,
        eval_dir=args.eval_dir,
        evidence_paths=evidence_paths,
        repo_id=args.repo_id,
        source_id=args.source_id,
        source_revision=args.source_revision,
        reference_id=args.reference_id,
        reference_revision=args.reference_revision,
    )
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
    )
    for gate in manifest["gates"]:
        print(
            f"[{'PASS' if gate['passed'] else 'FAIL'}] {gate['name']}: {gate['actual']}"
        )
    print(f"Release approved: {manifest['release_approved']}")
    if not manifest["release_approved"]:
        raise SystemExit(1)


if __name__ == "__main__":
    main()