bundle/v0/presentation/bind_verified_analysis.py

from __future__ import annotations

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


ROOT = Path(__file__).resolve().parent
PAPER_ROOT = ROOT.parent
CLAIMS_LOCK = PAPER_ROOT / "configs" / "official_claims_lock.json"
THEOREM_AUDIT = PAPER_ROOT / "configs" / "theorem-5.1-audit.json"
HYPERGRADIENT = PAPER_ROOT / ".openresearch" / "artifacts" / "validation" / "hypergradient.json"
HYPERGRADIENT_SIDECAR = HYPERGRADIENT.with_suffix(".sha256")
ANALYSIS_COPY = "bound-analysis.json"
LOCK_FILE = "evidence-lock.json"
BROWSER_DATA = "evidence-data.js"
BROWSER_PREFIX = "window.__DECISION_ATLAS_EVIDENCE__=Object.freeze("
BROWSER_SUFFIX = ");\n"
CLAIM_IDS = ["A1", "A2", "A3", "A4", "A5", "A6"]
EXPECTED_VERDICTS = {
    "A1": "partially_verified",
    "A2": "inconclusive",
    "A3": "inconclusive",
    "A4": "inconclusive",
    "A5": "inconclusive",
    "A6": "partially_verified",
}


class EvidenceBindingError(ValueError):
    pass


def _canonical(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + b"\n"


def _sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _sha256_value(value: Any) -> str:
    return "sha256:" + hashlib.sha256(_canonical(value).rstrip(b"\n")).hexdigest()


def _load(path: Path, label: str) -> dict[str, Any]:
    if not path.is_file():
        raise EvidenceBindingError(f"missing {label}: {path}")
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise EvidenceBindingError(f"invalid JSON in {label}: {path}") from exc
    if not isinstance(value, dict):
        raise EvidenceBindingError(f"{label} must be an object")
    return value


def _atomic(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    try:
        with os.fdopen(descriptor, "wb") as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)


def _verify_hypergradient() -> tuple[dict[str, Any], str]:
    receipt = _load(HYPERGRADIENT, "hypergradient receipt")
    sidecar = HYPERGRADIENT_SIDECAR.read_text(encoding="utf-8").strip().split()
    if len(sidecar) != 2 or sidecar[1] != HYPERGRADIENT.name:
        raise EvidenceBindingError("hypergradient sidecar schema changed")
    digest = _sha256_bytes(HYPERGRADIENT.read_bytes())
    if sidecar[0] != digest:
        raise EvidenceBindingError("hypergradient receipt differs from its pinned sidecar")
    if receipt.get("evidence_scale") != "BOUNDED_COMPONENT_VALIDATION_NOT_PAPER_SCALE":
        raise EvidenceBindingError("hypergradient scope was widened")
    return receipt, digest


def _verify_source(source: Path) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], str, str]:
    analysis = _load(source, "sealed reconciled analysis")
    analysis_without_hash = dict(analysis)
    declared_hash = analysis_without_hash.pop("analysis_payload_sha256", None)
    if declared_hash != _sha256_value(analysis_without_hash):
        raise EvidenceBindingError("sealed analysis payload hash is invalid")
    if analysis.get("kind") != "six_anchored_claim_full_matrix_capped5000_analysis":
        raise EvidenceBindingError("analysis is not the six-claim reconciled receipt")
    bindings = analysis.get("bindings", {})
    completeness = analysis.get("completeness", {})
    eligibility = analysis.get("claim_eligibility", {})
    if bindings.get("validated_success_count") != 14000 or bindings.get("expected_task_count") != 14000:
        raise EvidenceBindingError("analysis is not bound to all 14,000 tasks")
    if completeness.get("complete") is not True or completeness.get("rejected") != []:
        raise EvidenceBindingError("analysis completeness gate failed")
    if eligibility.get("eligible") is not True or eligibility.get("validation_provenance", {}).get("recovered_task_count") != 5:
        raise EvidenceBindingError("sealed recovery lineage is absent")
    if analysis.get("claim_verdicts") != EXPECTED_VERDICTS:
        raise EvidenceBindingError("analysis verdicts differ from the conservative six-claim contract")

    claims_lock = _load(CLAIMS_LOCK, "official claims lock")
    active = claims_lock.get("official_active_claims")
    if not isinstance(active, list) or [item.get("id") for item in active] != CLAIM_IDS:
        raise EvidenceBindingError("official claims lock is not the ordered A1 through A6 list")
    source_claims = analysis.get("official_claims", {}).get("active_claims")
    if source_claims != active:
        raise EvidenceBindingError("analysis claims differ from the pinned official claims")
    theorem = _load(THEOREM_AUDIT, "Theorem 5.1 audit")
    if theorem.get("recommended_c2_verdict") != "inconclusive":
        raise EvidenceBindingError("Theorem 5.1 boundary was weakened")
    hypergradient, hypergradient_digest = _verify_hypergradient()
    return analysis, claims_lock, theorem, hypergradient_digest, _sha256_bytes(source.read_bytes())


def _point_series(points: dict[str, Any], prefix: str) -> list[dict[str, Any]]:
    output = []
    for key, value in points.items():
        if not key.startswith(prefix):
            continue
        sample_size = int(key.rsplit("n", 1)[1])
        output.append({
            "sample_size": sample_size,
            "mean": value["mean"],
            "lower": value["ci"]["lower"],
            "upper": value["ci"]["upper"],
            "distribution_count": value["distribution_count"],
            "observation_count": value["observation_count"],
        })
    return sorted(output, key=lambda item: item["sample_size"])


def _presentation_payload(analysis: dict[str, Any], claims_lock: dict[str, Any], theorem: dict[str, Any], source_digest: str, hypergradient_digest: str) -> dict[str, Any]:
    active = {item["id"]: item["text"] for item in claims_lock["official_active_claims"]}
    a4 = analysis["A4"]["anchored_setup_audit"]
    coverage = _point_series(analysis["A5"]["coverage_reporting_points"], "portfolio_gaussian/")
    regression = _point_series(analysis["A6"]["improvement_reporting_points"], "regression_absolute_main/")
    payload: dict[str, Any] = {
        "schema_version": 4,
        "paper": {"openreview_id": "K1EPPO9t2c", "submission": "12512", "title": "Loss-Aware Distributionally Robust Optimization via Trainable Optimal Transport Ambiguity Sets"},
        "seal": {
            "source_analysis_file_sha256": source_digest,
            "analysis_payload_sha256": analysis["analysis_payload_sha256"],
            "aggregate_identity": analysis["bindings"]["aggregate_identity"],
            "aggregate_results_sha256": analysis["bindings"]["aggregate_results_sha256"],
            "manifest_hash": analysis["bindings"]["manifest_hash"],
            "recovery_attestation_hash": analysis["claim_eligibility"]["validation_provenance"]["attestation_hash"],
            "official_claims_lock_sha256": _sha256_bytes(CLAIMS_LOCK.read_bytes()),
            "theorem_audit_sha256": _sha256_bytes(THEOREM_AUDIT.read_bytes()),
            "hypergradient_receipt_sha256": hypergradient_digest,
        },
        "matrix": {
            "validated_rows": 14000,
            "recovered_rows": 5,
            "rejected_rows": 0,
            "solver_acceptance_fraction": analysis["A1"]["accepted_solver_fraction"],
            "censored_at_5000": analysis["stopping"]["overall"]["counts"]["censored_at_5000"],
        },
        "claims": {
            "A1": {"verdict": analysis["A1"]["verdict"], "official_text": active["A1"], "title": "Did the complete bilevel pipeline run?", "answer": analysis["A1"]["rationale"], "facts": [["Validated tasks", 14000], ["Accepted solver fraction", analysis["A1"]["accepted_solver_fraction"]], ["Required tasks capped", analysis["A1"]["censored_required_task_count"]]], "downgrade": analysis["A1"]["strongest_downgrade_argument"]},
            "A2": {"verdict": analysis["A2"]["verdict"], "official_text": active["A2"], "title": "Does Theorem 5.1 follow as stated?", "answer": analysis["A2"]["rationale"], "facts": [["Audit assumptions", len(theorem.get("assumptions", []))], ["Executed horizon", "finite and capped"], ["Theorem verdict", theorem["recommended_c2_verdict"]]], "downgrade": analysis["A2"]["guardrail"]},
            "A3": {"verdict": analysis["A3"]["verdict"], "official_text": active["A3"], "title": "Did we validate the nonsmooth hypergradient route?", "answer": analysis["A3"]["rationale"], "facts": [["Pinned component routes", 5], ["Receipt scope", "bounded components"], ["Paper-scale receipt", "not bound"]], "downgrade": "The pinned component receipt checks implemented gradients, but it does not bind Algorithm 1 at nonsmooth active-set boundaries."},
            "A4": {"verdict": analysis["A4"]["verdict"], "official_text": active["A4"], "title": "Does the exact Figure 2 trend reproduce?", "answer": analysis["A4"]["rationale"], "facts": [["Matched fields", sum(a4[key]["status"] == "match" for key in ("k", "J", "n_b", "gamma", "beta"))], ["Required fields", 5], ["Figure 2 formula", a4["relative_improvement_estimand"]["status"]]], "downgrade": "The available sample-size sweep is a sensitivity extension, not the anchored Figure 2 estimand."},
            "A5": {"verdict": analysis["A5"]["verdict"], "official_text": active["A5"], "title": "Did the learned set preserve 90% coverage?", "answer": analysis["A5"]["rationale"], "facts": [["Target", analysis["A5"]["coverage_target"]], ["Sample sizes", len(coverage)], ["All lower bounds at target", analysis["A5"]["criteria"]["all_lower_bounds_at_target"]]], "downgrade": "Every distribution-first upper confidence bound is below the predeclared 0.90 target, but capped tasks block a terminal falsified verdict."},
            "A6": {"verdict": analysis["A6"]["verdict"], "official_text": active["A6"], "title": "Was regression loss lower in all ten trials?", "answer": analysis["A6"]["rationale"], "facts": [["Independent trials", 10], ["Sample sizes", len(regression)], ["Every trial mean positive", analysis["A6"]["criteria"]["every_one_of_ten_trial_means_positive"]]], "downgrade": "The distribution-level mean is positive at each sample size, but the claim says consistently across all ten trials and that stricter test fails."},
        },
        "charts": {"coverage": coverage, "regression": regression},
        "limits": analysis["limitations"],
    }
    payload["derived_payload_sha256"] = _sha256_value(payload)
    return payload


def bind(source: Path, destination: Path = ROOT) -> dict[str, Any]:
    analysis, claims_lock, theorem, hypergradient_digest, source_digest = _verify_source(source.resolve())
    payload = _presentation_payload(analysis, claims_lock, theorem, source_digest, hypergradient_digest)
    payload_bytes = _canonical(payload)
    lock = {
        "schema_version": 4,
        "payload_sha256": _sha256_bytes(payload_bytes),
        "source_analysis_file_sha256": source_digest,
        "analysis_payload_sha256": analysis["analysis_payload_sha256"],
        "official_claims_lock_sha256": _sha256_bytes(CLAIMS_LOCK.read_bytes()),
        "theorem_audit_sha256": _sha256_bytes(THEOREM_AUDIT.read_bytes()),
        "hypergradient_receipt_sha256": hypergradient_digest,
    }
    browser = BROWSER_PREFIX.encode() + json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + BROWSER_SUFFIX.encode()
    _atomic(destination / ANALYSIS_COPY, payload_bytes)
    _atomic(destination / LOCK_FILE, _canonical(lock))
    _atomic(destination / BROWSER_DATA, browser)
    return lock


def verify_bound(destination: Path = ROOT) -> dict[str, Any]:
    lock = _load(destination / LOCK_FILE, "evidence lock")
    payload = _load(destination / ANALYSIS_COPY, "bound analysis")
    if lock.get("schema_version") != 4 or payload.get("schema_version") != 4:
        raise EvidenceBindingError("bound presentation schema changed")
    payload_hash = payload.pop("derived_payload_sha256", None)
    if payload_hash != _sha256_value(payload):
        raise EvidenceBindingError("derived presentation payload hash is invalid")
    payload["derived_payload_sha256"] = payload_hash
    payload_bytes = _canonical(payload)
    expected = {
        "schema_version": 4,
        "payload_sha256": _sha256_bytes(payload_bytes),
        "source_analysis_file_sha256": payload["seal"]["source_analysis_file_sha256"],
        "analysis_payload_sha256": payload["seal"]["analysis_payload_sha256"],
        "official_claims_lock_sha256": _sha256_bytes(CLAIMS_LOCK.read_bytes()),
        "theorem_audit_sha256": _sha256_bytes(THEOREM_AUDIT.read_bytes()),
        "hypergradient_receipt_sha256": _verify_hypergradient()[1],
    }
    if lock != expected:
        raise EvidenceBindingError("evidence lock differs from pinned sources")
    if payload["seal"]["official_claims_lock_sha256"] != expected["official_claims_lock_sha256"] or payload["seal"]["theorem_audit_sha256"] != expected["theorem_audit_sha256"] or payload["seal"]["hypergradient_receipt_sha256"] != expected["hypergradient_receipt_sha256"]:
        raise EvidenceBindingError("bound payload differs from pinned scientific evidence")
    active = _load(CLAIMS_LOCK, "official claims lock")["official_active_claims"]
    if [item["id"] for item in active] != CLAIM_IDS or list(payload["claims"]) != CLAIM_IDS:
        raise EvidenceBindingError("six-claim ordering changed")
    if {key: value["verdict"] for key, value in payload["claims"].items()} != EXPECTED_VERDICTS:
        raise EvidenceBindingError("bound verdicts were strengthened")
    browser_expected = BROWSER_PREFIX.encode() + json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + BROWSER_SUFFIX.encode()
    if (destination / BROWSER_DATA).read_bytes() != browser_expected:
        raise EvidenceBindingError("browser evidence differs from the sealed presentation payload")
    return lock


def main() -> int:
    parser = argparse.ArgumentParser(description="Bind a sealed reconciled six-claim analysis to the Decision Atlas")
    parser.add_argument("command", choices=("bind", "verify-bound"))
    parser.add_argument("source", nargs="?", type=Path)
    parser.add_argument("--destination", type=Path, default=ROOT)
    args = parser.parse_args()
    try:
        result = verify_bound(args.destination) if args.command == "verify-bound" else bind(args.source, args.destination) if args.source else (_ for _ in ()).throw(EvidenceBindingError("source is required for bind"))
    except EvidenceBindingError as exc:
        parser.exit(1, f"FAIL: {exc}\n")
    print(json.dumps(result, indent=2, sort_keys=True))
    return 0


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