source/scripts/validate_sub100m_benchmark.py

"""Validate the frozen, no-results sub-100M benchmark scaffold."""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import unicodedata
from collections.abc import Mapping, Sequence
from pathlib import Path, PurePosixPath
from typing import Any, NoReturn

BENCHMARK_ID = "mobile-actions-sub100m-v1"
CATALOG_SCHEMA = "barunaction-sub100m-catalog-template-v1"
LANE_SCHEMA = "barunaction-sub100m-lane-contracts-v1"
EXPECTED_BUNDLE_FILES = {"catalog.json", "lane-contracts.json"}
LANES = {"matched_adaptation", "off_the_shelf"}
MOVING_REVISIONS = {"latest", "main", "master"}
SHA256 = re.compile(r"[0-9a-f]{64}")
GIT_COMMIT = re.compile(r"[0-9a-f]{40}")


class ValidationError(ValueError):
    """A benchmark scaffold artifact violates its frozen public contract."""


def _reject_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValidationError(f"duplicate JSON key {key!r}")
        result[key] = value
    return result


def _reject_constant(value: str) -> NoReturn:
    raise ValidationError(f"non-finite JSON value {value!r}")


def _load_json(path: Path) -> Mapping[str, Any]:
    try:
        payload = json.loads(
            path.read_text(encoding="utf-8"),
            object_pairs_hook=_reject_pairs,
            parse_constant=_reject_constant,
        )
    except ValidationError:
        raise
    except (OSError, UnicodeError, json.JSONDecodeError) as error:
        raise ValidationError(f"cannot read strict JSON from {path}") from error
    if not isinstance(payload, Mapping):
        raise ValidationError(f"{path} must contain a JSON object")
    return payload


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


def _normalize(value: Any) -> Any:
    if isinstance(value, str):
        normalized = unicodedata.normalize("NFC", value)
        if normalized != value:
            raise ValidationError("JSON strings must already be NFC-normalized")
        return normalized
    if isinstance(value, Mapping):
        return {str(key): _normalize(item) for key, item in value.items()}
    if isinstance(value, list):
        return [_normalize(item) for item in value]
    if isinstance(value, float):
        raise ValidationError("benchmark manifests must not contain floating-point values")
    return value


def _file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    try:
        with path.open("rb") as handle:
            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                digest.update(chunk)
    except OSError as error:
        raise ValidationError(f"cannot hash {path}") from error
    return digest.hexdigest()


def _require_sha256(value: Any, path: str) -> str:
    if not isinstance(value, str) or SHA256.fullmatch(value) is None:
        raise ValidationError(f"{path} must be a lowercase SHA-256")
    return value


def _require_mapping(value: Any, path: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise ValidationError(f"{path} must be an object")
    return value


def _require_list(value: Any, path: str) -> list[Any]:
    if not isinstance(value, list):
        raise ValidationError(f"{path} must be an array")
    return value


def _require_relative_source(repo_root: Path, relative: Any, expected: Any, path: str) -> None:
    if not isinstance(relative, str):
        raise ValidationError(f"{path} source path must be a string")
    pure = PurePosixPath(relative)
    if pure.is_absolute() or ".." in pure.parts or str(pure) != relative:
        raise ValidationError(f"{path} source path must be normalized and repository-relative")
    expected_sha256 = _require_sha256(expected, f"{path} source SHA-256")
    source = repo_root / relative
    if not source.is_file() or source.is_symlink():
        raise ValidationError(f"{path} source file is missing or is a symlink")
    if _file_sha256(source) != expected_sha256:
        raise ValidationError(f"{path} source SHA-256 mismatch")


def _validate_contract(container: Mapping[str, Any], path: str) -> Mapping[str, Any]:
    contract = _require_mapping(container.get("contract"), f"{path}.contract")
    expected = _require_sha256(container.get("contract_sha256"), f"{path}.contract_sha256")
    if _canonical_sha256(contract) != expected:
        raise ValidationError(f"{path} contract SHA-256 mismatch")
    return contract


def _validate_lane_contracts(repo_root: Path, payload: Mapping[str, Any]) -> None:
    if payload.get("schema_version") != LANE_SCHEMA:
        raise ValidationError("unsupported lane-contract schema")
    if payload.get("benchmark_id") != BENCHMARK_ID:
        raise ValidationError("lane-contract benchmark_id mismatch")
    if payload.get("state") != "lane_contracts_frozen_catalog_unpopulated_no_runs":
        raise ValidationError("lane contracts must identify the unpopulated, no-run scaffold")

    scope = _require_mapping(payload.get("scope"), "$.scope")
    if scope.get("literal_exhaustive_coverage_claimed") is not False:
        raise ValidationError("the catalog must not claim literal exhaustive coverage")
    if scope.get("future_discoveries_require_a_new_catalog_version_before_scoring") is not True:
        raise ValidationError("future catalog additions must require a pre-score version change")

    eligibility = _require_mapping(payload.get("eligibility"), "$.eligibility")
    if eligibility.get("parameter_operator") != "<":
        raise ValidationError("parameter eligibility must use a strict less-than operator")
    if eligibility.get("parameter_limit_exclusive") != 100_000_000:
        raise ValidationError("parameter eligibility must be exactly <100,000,000")
    for key in (
        "immutable_revision_required",
        "license_required",
        "public_ungated_checkpoint_required",
        "standalone_local_checkpoint_required",
        "text_generation_required",
    ):
        if eligibility.get(key) is not True:
            raise ValidationError(f"eligibility.{key} must be true")
    for key in ("estimated_or_rounded_parameter_counts_allowed", "remote_api_only_allowed"):
        if eligibility.get(key) is not False:
            raise ValidationError(f"eligibility.{key} must be false")

    data = _require_mapping(payload.get("data"), "$.data")
    if data.get("official_evaluation_rows_allowed") is not False:
        raise ValidationError("official evaluation access must remain forbidden")
    if data.get("development_rows") != 756 or data.get("training_rows") != 7937:
        raise ValidationError("Mobile Actions population sizes changed")
    revision = data.get("dataset_revision")
    if not isinstance(revision, str) or GIT_COMMIT.fullmatch(revision) is None:
        raise ValidationError("$.data.dataset_revision must be a 40-character git commit")
    for key in (
        "development_manifest_sha256",
        "source_sha256",
        "training_manifest_sha256",
    ):
        _require_sha256(data.get(key), f"$.data.{key}")

    prompt = _require_mapping(payload.get("prompt"), "$.prompt")
    _validate_contract(prompt, "$.prompt")
    _require_relative_source(
        repo_root,
        prompt.get("renderer_source_path"),
        prompt.get("renderer_source_sha256"),
        "$.prompt",
    )

    decoding = _require_mapping(payload.get("decoding"), "$.decoding")
    decoding_contract = _validate_contract(decoding, "$.decoding")
    if decoding_contract.get("algorithm") != "unconstrained_deterministic_greedy":
        raise ValidationError("decoding must remain deterministic greedy")
    if decoding_contract.get("max_new_tokens") != 192:
        raise ValidationError("decoding max_new_tokens changed")
    _require_relative_source(
        repo_root,
        decoding.get("implementation_path"),
        decoding.get("implementation_sha256"),
        "$.decoding",
    )

    scoring = _require_mapping(payload.get("scoring"), "$.scoring")
    scoring_contract = _validate_contract(scoring, "$.scoring")
    if scoring_contract.get("callable") != "barunlm.evaluation.mobile_actions.write_scores":
        raise ValidationError("the protocol must reuse the existing Mobile Actions scorer")
    sources = _require_mapping(scoring.get("source_files"), "$.scoring.source_files")
    if set(sources) != {
        "src/barunlm/evaluation/action_ir.py",
        "src/barunlm/evaluation/evaluator.py",
        "src/barunlm/evaluation/mobile_actions.py",
    }:
        raise ValidationError("scorer source closure changed")
    for relative, digest in sources.items():
        _require_relative_source(repo_root, relative, digest, "$.scoring")

    lanes = _require_mapping(payload.get("lanes"), "$.lanes")
    if set(lanes) != LANES:
        raise ValidationError("the two benchmark lanes must remain separate")
    off_the_shelf = _require_mapping(lanes["off_the_shelf"], "$.lanes.off_the_shelf")
    off_contract = _validate_contract(off_the_shelf, "$.lanes.off_the_shelf")
    if (
        off_contract.get("parameter_updates") != 0
        or off_contract.get("adapters_added") is not False
    ):
        raise ValidationError("off-the-shelf lane may not adapt the model")
    matched = _require_mapping(lanes["matched_adaptation"], "$.lanes.matched_adaptation")
    matched_contract = _validate_contract(matched, "$.lanes.matched_adaptation")
    if matched.get("requires_no_prior_mobile_actions_adaptation") is not True:
        raise ValidationError("matched adaptation must exclude already task-adapted bases")
    if matched_contract.get("training_manifest_sha256") != data.get("training_manifest_sha256"):
        raise ValidationError("matched adaptation training population changed")
    if matched_contract.get("passes_over_training_manifest") != 1:
        raise ValidationError("matched adaptation must remain a one-pass treatment")


def _validate_entry(entry: Any, contract: Mapping[str, Any], limit: int) -> None:
    record = _require_mapping(entry, "$.entries[]")
    required = set(
        _require_list(contract.get("required_fields"), "$.entry_contract.required_fields")
    )
    if set(record) != required:
        raise ValidationError("catalog entry fields differ from the frozen entry contract")
    revision = record.get("revision")
    if not isinstance(revision, str) or not revision or revision.lower() in MOVING_REVISIONS:
        raise ValidationError("catalog entries require a non-moving revision")
    if record.get("generation_interface") not in {"causal_lm", "seq2seq_lm"}:
        raise ValidationError("catalog entry is not a supported text-generation model")

    access = _require_mapping(record.get("access"), "$.entries[].access")
    if access.get("public") is not True or access.get("gated") is not False:
        raise ValidationError("catalog checkpoints must be public and ungated")
    _require_sha256(access.get("verification_receipt_sha256"), "access receipt")

    parameters = _require_mapping(record.get("parameters"), "$.entries[].parameters")
    total = parameters.get("exact_unique_total")
    if isinstance(total, bool) or not isinstance(total, int) or not 0 < total < limit:
        raise ValidationError("catalog exact unique parameter count is not strictly below 100M")
    if not isinstance(parameters.get("unique_storages"), int):
        raise ValidationError("catalog parameter audit lacks an exact storage count")
    _require_sha256(parameters.get("count_receipt_sha256"), "parameter count receipt")

    artifacts = _require_mapping(record.get("artifacts"), "$.entries[].artifacts")
    for field in _require_list(
        contract.get("artifact_required_fields"), "$.entry_contract.artifact_required_fields"
    ):
        _require_sha256(artifacts.get(field), f"artifact {field}")

    lane_eligibility = _require_mapping(record.get("lane_eligibility"), "lane_eligibility")
    if set(lane_eligibility) != LANES:
        raise ValidationError("catalog entry must classify both lanes")
    eligible = 0
    for lane, decision in lane_eligibility.items():
        decision_map = _require_mapping(decision, f"lane_eligibility.{lane}")
        if decision_map.get("status") not in {"eligible", "ineligible"}:
            raise ValidationError("lane eligibility status must be eligible or ineligible")
        reasons = _require_list(decision_map.get("reason_codes"), "lane reason_codes")
        if reasons != sorted(set(reasons)):
            raise ValidationError("lane reason codes must be unique and sorted")
        eligible += decision_map.get("status") == "eligible"
    if eligible == 0:
        raise ValidationError("catalog entry is ineligible for both lanes")


def _validate_catalog(payload: Mapping[str, Any], lane_sha256: str) -> None:
    if payload.get("schema_version") != CATALOG_SCHEMA:
        raise ValidationError("unsupported catalog schema")
    if payload.get("benchmark_id") != BENCHMARK_ID:
        raise ValidationError("catalog benchmark_id mismatch")
    if payload.get("lane_contracts_sha256") != lane_sha256:
        raise ValidationError("catalog does not bind the exact lane-contract file")
    boundary = _require_mapping(payload.get("catalog_boundary"), "$.catalog_boundary")
    if boundary.get("literal_exhaustive_coverage_claimed") is not False:
        raise ValidationError("catalog must not claim literal exhaustive coverage")
    if boundary.get("scores_may_influence_membership") is not False:
        raise ValidationError("model scores may not influence catalog membership")

    entries = _require_list(payload.get("entries"), "$.entries")
    contract = _require_mapping(payload.get("entry_contract"), "$.entry_contract")
    if set(_require_list(contract.get("lane_names"), "$.entry_contract.lane_names")) != LANES:
        raise ValidationError("entry contract must classify both benchmark lanes")
    if boundary.get("roster_frozen") is False:
        if entries or payload.get("results_status") != "no_model_runs_or_results":
            raise ValidationError(
                "the unfrozen catalog template must contain no entries or results"
            )
    elif boundary.get("roster_frozen") is True:
        if not entries:
            raise ValidationError("a frozen roster may not be empty")
    else:
        raise ValidationError("catalog roster_frozen must be boolean")

    seen: set[str] = set()
    limit = 100_000_000
    for entry in entries:
        _validate_entry(entry, contract, limit)
        entry_id = _require_mapping(entry, "$.entries[]").get("entry_id")
        if not isinstance(entry_id, str) or not entry_id or entry_id in seen:
            raise ValidationError("catalog entry IDs must be unique non-empty strings")
        seen.add(entry_id)
    if [entry.get("entry_id") for entry in entries] != sorted(seen):
        raise ValidationError("catalog entries must be sorted by entry_id")

    exclusions = _require_list(payload.get("exclusions"), "$.exclusions")
    if exclusions and boundary.get("roster_frozen") is not True:
        raise ValidationError("exclusions belong only to a researched frozen roster")


def validate_sub100m_scaffold(repo_root: str | Path) -> dict[str, Any]:
    root = Path(repo_root).resolve()
    bundle = root / "benchmarks" / "sub100m"
    names = {path.name for path in bundle.iterdir() if path.is_file()}
    if names != EXPECTED_BUNDLE_FILES:
        raise ValidationError("sub100m must contain only catalog.json and lane-contracts.json")
    lanes_path = bundle / "lane-contracts.json"
    catalog_path = bundle / "catalog.json"
    lanes = _load_json(lanes_path)
    catalog = _load_json(catalog_path)
    _normalize(lanes)
    _normalize(catalog)
    _validate_lane_contracts(root, lanes)
    lanes_sha256 = _file_sha256(lanes_path)
    _validate_catalog(catalog, lanes_sha256)
    return {
        "benchmark_id": BENCHMARK_ID,
        "catalog_entries": len(_require_list(catalog.get("entries"), "$.entries")),
        "catalog_sha256": _file_sha256(catalog_path),
        "lane_contracts_sha256": lanes_sha256,
        "literal_exhaustive_coverage_claimed": False,
        "model_runs": 0,
        "ok": True,
        "scorer_callable": "barunlm.evaluation.mobile_actions.write_scores",
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--repository",
        type=Path,
        default=Path(__file__).resolve().parents[1],
        help="repository root containing benchmarks/sub100m",
    )
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        result = validate_sub100m_scaffold(args.repository)
    except ValidationError as error:
        print(json.dumps({"error": str(error), "ok": False}, sort_keys=True))
        return 2
    print(json.dumps(result, sort_keys=True))
    return 0


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