| |
| """Validate the independent T90 official-quality audit for ELIGIBLE vision models.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import csv |
| import hashlib |
| import json |
| import shlex |
| import sys |
| from collections import Counter |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| VISION_TASKS = {"vision_classification", "object_detection", "semantic_segmentation"} |
|
|
|
|
| def sha256_file(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def md5_file(path: Path) -> str: |
| digest = hashlib.md5(usedforsecurity=False) |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def atomic_json(path: Path, value: Any) -> None: |
| temporary = path.with_suffix(path.suffix + ".tmp") |
| temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n") |
| temporary.replace(path) |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--project-root", type=Path, default=Path.cwd()) |
| parser.add_argument("--audit", type=Path, required=True) |
| parser.add_argument("--output", type=Path, required=True) |
| args = parser.parse_args() |
| root = args.project_root.absolute() |
| audit_path = args.audit if args.audit.is_absolute() else root / args.audit |
| audit = json.loads(audit_path.read_text()) |
| checks: list[dict[str, Any]] = [] |
|
|
| def check(name: str, condition: bool, detail: Any) -> None: |
| checks.append({"name": name, "pass": bool(condition), "detail": detail}) |
|
|
| with (root / "model_registry.csv").open(newline="") as handle: |
| registry = list(csv.DictReader(handle)) |
| expected = { |
| row["model_id"] |
| for row in registry |
| if row["eligibility"] == "ELIGIBLE" and row["task"] in VISION_TASKS |
| } |
| models = audit["models"] |
| by_id = {row["model_id"]: row for row in models} |
| check("audit_ids_unique", len(by_id) == len(models), len(by_id)) |
| check("all_eligible_vision_models", set(by_id) == expected, {"expected": sorted(expected), "actual": sorted(by_id)}) |
| decisions = Counter(row["decision"] for row in models) |
| check("decision_vocabulary", set(decisions) <= {"EXECUTED_PASS", "EXECUTABLE", "EXTERNAL_BLOCKED"}, decisions) |
| check("summary_counts", audit["summary"]["models"] == len(models) == 16 and audit["summary"]["executed_pass"] == decisions["EXECUTED_PASS"] and audit["summary"]["executable_or_running"] == decisions["EXECUTABLE"] and audit["summary"]["external_blocked"] == decisions["EXTERNAL_BLOCKED"], {"summary": audit["summary"], "decisions": decisions}) |
| blocked_missing_detail = [row["model_id"] for row in models if row["decision"] == "EXTERNAL_BLOCKED" and (not row.get("blocker") or not row.get("resolution"))] |
| check("blocked_resolution_conditions", not blocked_missing_detail, blocked_missing_detail) |
| check("segmentation_threshold_policy", all(by_id[model_id]["official_threshold"] is None and by_id[model_id]["expected_acceptance_status"] == "MEASURED_NO_ACCEPTANCE_THRESHOLD" for model_id in ["SG06", "SG07"]), {model_id: by_id[model_id] for model_id in ["SG06", "SG07"]}) |
|
|
| executed_details: dict[str, Any] = {} |
| for model_id, threshold, expected_count in [("VC01", 0.8, 1000), ("VC02", 0.85, 200)]: |
| quality_root = root / "models" / "vision_classification" / model_id / "quality_evaluation" |
| summary = json.loads((quality_root / "results" / "quality_summary.json").read_text()) |
| validation = json.loads((quality_root / "results" / "validation_report.json").read_text()) |
| manifest = json.loads((quality_root / "execution_manifest.json").read_text()) |
| check(f"{model_id}_summary_pass", summary["status"] == "PASS" and summary["acceptance_status"] == "PASS", {"status": summary["status"], "acceptance": summary["acceptance_status"]}) |
| check(f"{model_id}_threshold", summary["threshold"] == threshold and summary["quality"]["public_int8"]["accuracy"] >= threshold, {"threshold": summary["threshold"], "accuracy": summary["quality"]["public_int8"]["accuracy"]}) |
| check(f"{model_id}_generic_quality_fields", all(summary["quality"][variant]["format"] == "tflite" and summary["quality"][variant]["sample_count"] == expected_count for variant in ["fp32", "public_int8"]), summary["quality"]) |
| check(f"{model_id}_resume", summary["resume"]["rows_reused"] == expected_count and summary["resume"]["rows_evaluated"] == 0, summary["resume"]) |
| check(f"{model_id}_validator", validation["status"] == "PASS" and all(item["pass"] for item in validation["checks"]), {"status": validation["status"], "checks": len(validation["checks"])}) |
| check(f"{model_id}_execution_manifest", manifest["overall"]["status"] == "PASS" and not manifest["overall"]["forbidden_operations_performed"], manifest["overall"]) |
| check(f"{model_id}_prohibited_operations", not any(summary["prohibited_operations"].values()), summary["prohibited_operations"]) |
| executed_details[model_id] = {"summary": summary["quality"], "validation_checks": len(validation["checks"]), "resume": summary["resume"]} |
|
|
| cifar = root / "research" / "downloads" / "cifar10" / "cifar-10-python.tar.gz" |
| transport = by_id["VC02"]["transport_provenance"] |
| check("vc02_transport_sha256_identity", sha256_file(cifar) == transport["downloaded_sha256"] == transport["mirror_lfs_pointer_sha256"] == transport["keras_v3_12_canonical_url_sha256"], sha256_file(cifar)) |
| check("vc02_transport_md5_identity", md5_file(cifar) == transport["downloaded_md5"] == transport["official_page_md5"], md5_file(cifar)) |
| check("vc02_canonical_source_unchanged", transport["canonical_source_changed"] is False and transport["canonical_url"] == "https://www.cs.toronto.edu/~kriz/cifar-10-python.tar.gz", transport) |
| check("vc02_canonical_failure_logged", transport["canonical_probe_exit_code"] == 28 and (root / transport["canonical_probe_stdout"]).is_file() and (root / transport["canonical_probe_stderr"]).is_file(), {"exit_code": transport["canonical_probe_exit_code"]}) |
|
|
| passed = all(item["pass"] for item in checks) |
| report = { |
| "stage": "T90_VISION_OFFICIAL_QUALITY_AUDIT_VALIDATION", |
| "status": "PASS" if passed else "FAIL", |
| "failure_code": None if passed else "FAIL_ANALYSIS", |
| "command": shlex.join([sys.executable, *sys.argv]), |
| "tool_versions": {"python": sys.version.split()[0]}, |
| "audit": str(audit_path), |
| "audit_sha256": sha256_file(audit_path), |
| "eligible_vision_model_count": len(expected), |
| "decision_counts": dict(decisions), |
| "checks": checks, |
| "executed_details": executed_details, |
| } |
| output = args.output if args.output.is_absolute() else root / args.output |
| atomic_json(output, report) |
| print(json.dumps({"status": report["status"], "checks": len(checks), "passed": sum(item["pass"] for item in checks)}, sort_keys=True)) |
| return 0 if passed else 1 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
