bundle/v0/scripts/validate_theorem_audit.py

from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
from typing import Any

from loss_aware_dro_repro.core import ContractError, canonical_bytes, load_json, load_plan, plan_hash


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_SOURCE = ROOT / "configs" / "theorem-5.1-audit.json"
DEFAULT_ARTIFACT = ROOT / ".openresearch" / "artifacts" / "validation" / "theorem-5.1.json"
FROZEN_AUDIT_SHA256 = "a3e889405cdd71b97c34852dbd08f02f9d2cc946df4066f3dfe6f58618007842"

EXPECTED_TOP_LEVEL_KEYS = {
    "schema_version",
    "audit_id",
    "claim_id",
    "plan_hash",
    "paper",
    "source_evidence",
    "evidence_levels",
    "theorem_statement",
    "assumptions",
    "conclusion_qualifications",
    "finite_experiment_boundary",
    "all_assumptions_pass",
    "paper_theorem_as_written",
    "executed_algorithm_within_theorem_scope",
    "recommended_c2_verdict",
    "falsification_status",
    "guardrails",
}
EXPECTED_ASSUMPTION_IDS = {f"A{index}" for index in range(1, 13)}
ALLOWED_BRANCHES = {"both", "diminishing", "constant", "differentiability"}
ALLOWED_CLASSIFICATIONS = {
    "stated_theorem_assumption",
    "supporting_paper_assumption",
    "sufficient_appendix_condition",
    "cited_result_bridge",
    "domain_recurrence_bridge",
    "cited_result_condition",
    "consequence_or_alternative",
    "execution_scope_condition",
}
ALLOWED_STATUSES = {
    "established",
    "stated_not_instance_certified",
    "sufficient_route_not_certified",
    "bridge_not_established",
    "not_exercised",
    "not_empirically_decidable",
    "not_instance_certified",
    "dependent_on_unresolved_A10",
    "contradicted_by_execution",
}
BLOCKING_CLASSIFICATIONS = {
    "stated_theorem_assumption",
    "cited_result_bridge",
    "domain_recurrence_bridge",
    "cited_result_condition",
    "execution_scope_condition",
}
EXPECTED_SOURCE_HASHES = {
    "paper": "566fe932f33d868d848c7b3928354a70036fbb69d0f90d951e3ffabc85bf2bd1",
    "davis": "dafc5c774365cdca133026d537c7206d80335c6490430437f783c329919e7f5b",
    "bolte_v1": "705e09125146549cacf9aa0c871f96bc408686e3c780bc9e724f3f691813d5c8",
    "bolte_v2": "10b59a7411ef928627492cac6887ebde5b24a1dbdce87b82bd9218545fa1ecf5",
}
EXPECTED_AUTHOR_COMMIT = "6e8d18f023f4f2e920af22dc1ef32b292dbb9631"


def _require_exact_bool(value: Any, field: str) -> None:
    if type(value) is not bool:
        raise ContractError(f"{field} must be a JSON boolean")


def _require_nonempty_strings(value: Any, field: str) -> None:
    if not isinstance(value, list) or not value:
        raise ContractError(f"{field} must be a non-empty list")
    if any(not isinstance(item, str) or not item.strip() for item in value):
        raise ContractError(f"{field} must contain only non-empty strings")


def _sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        while chunk := handle.read(1024 * 1024):
            digest.update(chunk)
    return digest.hexdigest()


def _bound_artifact(root: Path, relative: Any, source_id: str) -> Path:
    if not isinstance(relative, str) or not relative.strip():
        raise ContractError(f"{source_id} artifact_path must be a non-empty relative path")
    candidate = Path(relative)
    if candidate.is_absolute():
        raise ContractError(f"{source_id} artifact_path must be relative")
    resolved_root = root.resolve()
    resolved = (resolved_root / candidate).resolve()
    if not resolved.is_relative_to(resolved_root):
        raise ContractError(f"{source_id} artifact_path escapes the lane root")
    if not resolved.is_file():
        raise ContractError(f"{source_id} bound source artifact is missing: {relative}")
    return resolved


def validate_audit(audit: dict[str, Any], *, root: Path = ROOT) -> None:
    if not isinstance(audit, dict):
        raise ContractError("theorem audit must be a JSON object")
    if set(audit) != EXPECTED_TOP_LEVEL_KEYS:
        missing = sorted(EXPECTED_TOP_LEVEL_KEYS - set(audit))
        extra = sorted(set(audit) - EXPECTED_TOP_LEVEL_KEYS)
        raise ContractError(f"theorem audit keys changed; missing={missing}, extra={extra}")
    if type(audit["schema_version"]) is not int or audit["schema_version"] != 2:
        raise ContractError("unsupported theorem audit schema")
    if audit["audit_id"] != "theorem-5.1-c2-v2" or audit["claim_id"] != "C2":
        raise ContractError("theorem audit identity changed")

    plan = load_plan(root / "configs" / "paper_scale.json")
    if audit["plan_hash"] != plan_hash(plan):
        raise ContractError("theorem audit plan_hash does not match paper-scale plan")
    paper_meta = load_json(root / "paper.json")
    paper = audit["paper"]
    expected_paper_fields = {
        "submission": paper_meta["submission"],
        "openreview": paper_meta["openreview"],
        "arxiv": paper_meta["arxiv"],
        "theorem": "Theorem 5.1",
        "section": "Section 5.1",
        "ideal_update": "Algorithm 1",
    }
    for field, expected in expected_paper_fields.items():
        if paper.get(field) != expected:
            raise ContractError(f"paper.{field} changed")
    if paper.get("objective_equations") != ["(14)", "(15)", "(16)", "(17)"]:
        raise ContractError("paper objective equation map changed")

    levels = audit["evidence_levels"]
    if set(levels) != {
        "stated_theorem_assumption",
        "sufficient_appendix_condition",
        "released_instance_certification",
    } or any(not isinstance(value, str) or not value.strip() for value in levels.values()):
        raise ContractError("evidence-level distinctions changed")

    sources = audit["source_evidence"]
    if set(sources) != {"paper", "davis", "bolte_v1", "bolte_v2", "author_code"}:
        raise ContractError("primary source set changed")
    for source_id, expected_hash in EXPECTED_SOURCE_HASHES.items():
        source = sources[source_id]
        if source.get("source_archive_sha256") != expected_hash:
            raise ContractError(f"{source_id} recorded source archive hash changed")
        if not str(source.get("url", "")).startswith("https://arxiv.org/"):
            raise ContractError(f"{source_id} must cite a primary arXiv URL")
        _require_nonempty_strings(source.get("citations"), f"source_evidence.{source_id}.citations")
        artifact = _bound_artifact(root, source.get("artifact_path"), source_id)
        if _sha256_file(artifact) != expected_hash:
            raise ContractError(f"{source_id} bound source artifact hash mismatch")
    author = sources["author_code"]
    if author.get("commit") != EXPECTED_AUTHOR_COMMIT:
        raise ContractError("author code commit changed")
    if author["commit"] != paper_meta["official_source"]["revision"]:
        raise ContractError("author code commit does not match paper.json")
    _require_nonempty_strings(author.get("citations"), "source_evidence.author_code.citations")

    assumptions = audit["assumptions"]
    if not isinstance(assumptions, list) or not assumptions:
        raise ContractError("assumptions must be a non-empty list")
    ids = {item.get("id") for item in assumptions if isinstance(item, dict)}
    if ids != EXPECTED_ASSUMPTION_IDS or len(assumptions) != len(EXPECTED_ASSUMPTION_IDS):
        raise ContractError("theorem assumption inventory changed")
    required_assumption_keys = {
        "id",
        "classification",
        "branch",
        "requirement",
        "paper_evidence",
        "author_code_evidence",
        "independent_code_evidence",
        "status",
        "reason",
    }
    for item in assumptions:
        if not isinstance(item, dict) or set(item) != required_assumption_keys:
            raise ContractError("assumption record keys changed")
        if item["classification"] not in ALLOWED_CLASSIFICATIONS:
            raise ContractError(f"invalid classification for {item['id']}")
        if item["branch"] not in ALLOWED_BRANCHES:
            raise ContractError(f"invalid branch for {item['id']}")
        if item["status"] not in ALLOWED_STATUSES:
            raise ContractError(f"invalid status for {item['id']}")
        for field in ("requirement", "reason"):
            if not isinstance(item[field], str) or not item[field].strip():
                raise ContractError(f"{item['id']}.{field} must be non-empty")
        for field in ("paper_evidence", "author_code_evidence", "independent_code_evidence"):
            _require_nonempty_strings(item[field], f"{item['id']}.{field}")

    _require_exact_bool(audit["all_assumptions_pass"], "all_assumptions_pass")
    _require_exact_bool(
        audit["executed_algorithm_within_theorem_scope"],
        "executed_algorithm_within_theorem_scope",
    )
    unresolved_blockers = [
        item["id"]
        for item in assumptions
        if item["classification"] in BLOCKING_CLASSIFICATIONS and item["status"] != "established"
    ]
    if audit["all_assumptions_pass"] != (not unresolved_blockers):
        raise ContractError("all_assumptions_pass is inconsistent with blocking audit records")
    if audit["all_assumptions_pass"]:
        raise ContractError("current source evidence does not support a passing theorem audit")
    if audit["executed_algorithm_within_theorem_scope"]:
        raise ContractError("current author and independent executions are modified recurrences")
    if audit["paper_theorem_as_written"] != "not_established_by_cited_results":
        raise ContractError("the conclusion-overreach finding changed")
    if audit["recommended_c2_verdict"] != "inconclusive":
        raise ContractError("C2 must remain inconclusive with unresolved conditions")
    if audit["falsification_status"] != "not_falsified":
        raise ContractError("the audit does not contain a certified counterexample")

    boundary = audit["finite_experiment_boundary"]
    if set(boundary) != {"can_establish", "cannot_establish"}:
        raise ContractError("finite experiment boundary keys changed")
    _require_nonempty_strings(boundary["can_establish"], "finite_experiment_boundary.can_establish")
    _require_nonempty_strings(boundary["cannot_establish"], "finite_experiment_boundary.cannot_establish")
    _require_nonempty_strings(audit["guardrails"], "guardrails")
    if not isinstance(audit["conclusion_qualifications"], list) or not audit["conclusion_qualifications"]:
        raise ContractError("conclusion_qualifications must be non-empty")

    content_hash = hashlib.sha256(canonical_bytes(audit)).hexdigest()
    if content_hash != FROZEN_AUDIT_SHA256:
        raise ContractError("frozen theorem-audit content changed without a reviewed schema revision")


def render_audit(audit: dict[str, Any]) -> bytes:
    return (json.dumps(audit, indent=2, ensure_ascii=False) + "\n").encode("utf-8")


def emit_audit(source: Path = DEFAULT_SOURCE, artifact: Path = DEFAULT_ARTIFACT) -> Path:
    audit = load_json(source)
    validate_audit(audit, root=ROOT)
    payload = render_audit(audit)
    artifact.parent.mkdir(parents=True, exist_ok=True)
    temporary = artifact.with_name(artifact.name + ".tmp")
    temporary.write_bytes(payload)
    temporary.replace(artifact)
    return artifact


def main() -> int:
    parser = argparse.ArgumentParser(description="Validate and emit the Theorem 5.1 C2 audit")
    parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE)
    parser.add_argument("--artifact", type=Path, default=DEFAULT_ARTIFACT)
    args = parser.parse_args()
    emitted = emit_audit(args.source, args.artifact)
    print(emitted)
    return 0


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