bundle/v0/scripts/validate_appendix_canary.py

from __future__ import annotations

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


LANE_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(LANE_ROOT / "src"))

from loss_aware_dro_repro.core import load_json, load_plan, plan_hash, sha256_value  # noqa: E402
from loss_aware_dro_repro.datasets import generate_dataset  # noqa: E402
from loss_aware_dro_repro.matrix import expand_tasks  # noqa: E402
from loss_aware_dro_repro.route_canary import source_tree_hash  # noqa: E402


def _hash(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _inside(path: Path, root: Path) -> bool:
    return path == root or root in path.parents


def _all_finite(value: Any) -> bool:
    if isinstance(value, bool) or value is None or isinstance(value, str):
        return True
    if isinstance(value, (int, float)):
        return math.isfinite(float(value))
    if isinstance(value, list):
        return all(_all_finite(item) for item in value)
    if isinstance(value, dict):
        return all(_all_finite(item) for item in value.values())
    return False


def validate(root: Path, manifest_path: Path) -> list[str]:
    root = root.resolve()
    manifest_path = manifest_path.resolve()
    receipt_path = root / "receipt.json"
    errors: list[str] = []
    try:
        receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        return [f"cannot read receipt/manifest: {exc}"]

    if (LANE_ROOT / manifest.get("receipt", "")).resolve() != receipt_path:
        errors.append("manifest receipt path mismatch")
    if manifest.get("receipt_sha256") != _hash(receipt_path):
        errors.append("manifest receipt hash mismatch")
    if manifest.get("canary_id") != receipt.get("canary_id"):
        errors.append("manifest canary id mismatch")
    if manifest.get("base_commit") != receipt.get("base_commit"):
        errors.append("manifest base commit mismatch")

    inputs = receipt.get("scientific_inputs", {})
    bound_paths: dict[str, Path] = {}
    for name in ("config", "plan", "published_gaussians", "regression_models"):
        key = f"{name}_path"
        hash_key = f"{name}_sha256"
        if key not in inputs:
            errors.append(f"missing scientific input binding: {key}")
            continue
        path = (LANE_ROOT / inputs[key]).resolve()
        bound_paths[name] = path
        if not _inside(path, LANE_ROOT) or not path.is_file():
            errors.append(f"{name} input is missing or escaping")
        elif inputs.get(hash_key) != _hash(path):
            errors.append(f"{name} byte hash mismatch")
    config_path = bound_paths.get("config")
    plan_path = bound_paths.get("plan")
    config = load_json(config_path) if config_path and config_path.is_file() else {}
    plan = load_plan(plan_path) if plan_path and plan_path.is_file() else {}
    if config:
        if receipt.get("canary_config_hash") != sha256_value(config):
            errors.append("selected config canonical hash mismatch")
        if config.get("canary_id") != receipt.get("canary_id"):
            errors.append("selected config canary id mismatch")
        if manifest.get("config") != config_path.relative_to(LANE_ROOT).as_posix():
            errors.append("manifest config path mismatch")
        if manifest.get("config_hash") != receipt.get("canary_config_hash"):
            errors.append("manifest config hash mismatch")
        if receipt.get("source_tree_hash") != source_tree_hash(config_path):
            errors.append("live source/config/input tree differs from receipt")
        if manifest.get("source_tree_hash") != receipt.get("source_tree_hash"):
            errors.append("manifest source tree hash mismatch")
    if plan:
        if receipt.get("plan_hash") != plan_hash(plan):
            errors.append("plan canonical hash mismatch")
        if manifest.get("plan_hash") != receipt.get("plan_hash"):
            errors.append("manifest plan hash mismatch")

    configured_routes = {route["name"]: route for route in config.get("routes", [])}
    receipt_routes = {route.get("route"): route for route in receipt.get("routes", [])}
    manifest_routes = {route.get("name"): route for route in manifest.get("routes", [])}
    expected = {"empirical_w1_portfolio", "absolute_regression", "squared_regression"}
    if set(configured_routes) != expected or set(receipt_routes) != expected or set(manifest_routes) != expected:
        errors.append("config/receipt/manifest route sets must exactly match the three Appendix routes")
    tasks = {task["task_id"]: task for task in expand_tasks(plan)} if plan else {}
    required_residuals = {"primal", "dual", "equality", "cone", "dual_cone", "complementarity", "duality_gap"}
    residual_threshold = float(config.get("solver_residual_max", 0.0))
    expected_iterations = config.get("max_outer_iterations")

    for name in sorted(expected):
        route_config = configured_routes.get(name, {})
        route = receipt_routes.get(name, {})
        route_manifest = manifest_routes.get(name, {})
        if route.get("task_id") != route_config.get("task_selector"):
            errors.append(f"{name}: task selector mismatch")
        if route.get("transport_estimand", {}).get("order") != route_config.get("wasserstein_order"):
            errors.append(f"{name}: transport order mismatch")
        if route.get("implementation") != route_config.get("implementation"):
            errors.append(f"{name}: implementation mismatch")
        for field, expected_value in (
            ("task_id", route.get("task_id")),
            ("wasserstein_order", route.get("transport_estimand", {}).get("order")),
            ("implementation", route.get("implementation")),
        ):
            if route_manifest.get(field) != expected_value:
                errors.append(f"{name}: manifest {field} mismatch")
        task = tasks.get(route.get("task_id"))
        if task is None:
            errors.append(f"{name}: task absent from bound plan")
        else:
            if route.get("task_hash") != task.get("task_hash"):
                errors.append(f"{name}: task hash mismatch")
            samples, metadata = generate_dataset(task)
            dataset = route.get("dataset", {})
            if dataset.get("fingerprint") != metadata["fingerprint"]:
                errors.append(f"{name}: dataset fingerprint mismatch")
            if dataset.get("shape") != list(samples.shape):
                errors.append(f"{name}: dataset shape mismatch")
            if dataset.get("seeds") != task["seeds"]:
                errors.append(f"{name}: dataset seed mismatch")
        if route.get("claim_eligible") is not False or route.get("scientific_verdicts") != {"C1": "HOLD", "C2": "HOLD", "C3": "HOLD"}:
            errors.append(f"{name}: route must remain claim-ineligible and HOLD")
        checks = route.get("gradient_checks", {})
        if checks.get("all_pass") is not True:
            errors.append(f"{name}: gradient checks failed")
        thresholds = checks.get("thresholds", {})
        if checks.get("empirical_ot", {}).get("relative_error", math.inf) > thresholds.get("empirical_ot", -1.0):
            errors.append(f"{name}: empirical OT gradient exceeds threshold")
        for check_name in ("conic_value", "combined_outer_active_penalty"):
            if checks.get(check_name, {}).get("relative_error", math.inf) > thresholds.get("conic_and_combined", -1.0):
                errors.append(f"{name}: {check_name} gradient exceeds threshold")
        solver = route.get("solver", {})
        if solver.get("threshold") != residual_threshold or solver.get("all_pass") is not True:
            errors.append(f"{name}: solver threshold/status mismatch")
        if max(
            solver.get("max_residual", math.inf),
            solver.get("maximum_ot_marginal_residual", math.inf),
            solver.get("maximum_ot_optimality_residual", math.inf),
        ) > residual_threshold:
            errors.append(f"{name}: solver/OT residual exceeds threshold")

        trace_info = route.get("raw_lineage", {})
        trace_path = (root / trace_info.get("trace", "")).resolve()
        if not _inside(trace_path, root) or not trace_path.is_file():
            errors.append(f"{name}: trace missing or escaping")
            continue
        try:
            trace = [json.loads(line) for line in trace_path.read_text(encoding="utf-8").splitlines() if line.strip()]
        except (OSError, UnicodeError, json.JSONDecodeError) as exc:
            errors.append(f"{name}: trace unreadable: {exc}")
            continue
        if not trace:
            errors.append(f"{name}: trace must be nonempty")
        trace_hash = _hash(trace_path)
        if trace_info.get("trace_sha256") != trace_hash:
            errors.append(f"{name}: trace hash mismatch")
        if (LANE_ROOT / route_manifest.get("trace", "")).resolve() != trace_path or route_manifest.get("trace_sha256") != trace_hash:
            errors.append(f"{name}: manifest trace binding mismatch")
        if trace_info.get("trace_rows") != len(trace) or len(trace) != expected_iterations:
            errors.append(f"{name}: trace count differs from receipt/config")
        previous_next = None
        for index, row in enumerate(trace):
            if not _all_finite(row):
                errors.append(f"{name}: non-finite trace value at row {index}")
            if row.get("iteration") != index:
                errors.append(f"{name}: iteration discontinuity at row {index}")
            if previous_next is not None and row.get("L") != previous_next:
                errors.append(f"{name}: metric trace discontinuity at row {index}")
            previous_next = row.get("L_next")
            if row.get("route") != name or row.get("task_id") != route.get("task_id"):
                errors.append(f"{name}: trace route/task mismatch at row {index}")
            if row.get("wasserstein_order") != route_config.get("wasserstein_order") or row.get("implementation") != route_config.get("implementation"):
                errors.append(f"{name}: trace estimand mismatch at row {index}")
            if row.get("solver_status") not in {"optimal", "optimal_inaccurate"}:
                errors.append(f"{name}: unaccepted solver status at row {index}")
            residuals = row.get("solver_residuals", {})
            if set(residuals) != required_residuals:
                errors.append(f"{name}: incomplete residual certificate at row {index}")
            elif max(float(value) for value in residuals.values()) > residual_threshold:
                errors.append(f"{name}: residual threshold exceeded at row {index}")
            if row.get("maximum_ot_marginal_residual", math.inf) > residual_threshold or row.get("maximum_ot_optimality_residual", math.inf) > residual_threshold:
                errors.append(f"{name}: OT certificate failed at row {index}")
            if not row.get("dual") or not row.get("slack") or not row.get("q"):
                errors.append(f"{name}: primal/dual/slack recomputation data missing at row {index}")
        optimization = route.get("optimization", {})
        if optimization.get("iterations") != expected_iterations:
            errors.append(f"{name}: terminal iteration count mismatch")
        if trace and optimization.get("terminal_L") != trace[-1].get("L_next"):
            errors.append(f"{name}: terminal factor is not final trace update")
        if optimization.get("terminal_solver_status") not in {"optimal", "optimal_inaccurate"}:
            errors.append(f"{name}: terminal solver status unaccepted")
        terminal_residuals = optimization.get("terminal_solver_residuals", {})
        if set(terminal_residuals) != required_residuals:
            errors.append(f"{name}: terminal residual certificate incomplete")
        elif max(float(value) for value in terminal_residuals.values()) > residual_threshold:
            errors.append(f"{name}: terminal residual threshold exceeded")
        if not optimization.get("terminal_dual") or not optimization.get("terminal_slack") or not optimization.get("terminal_q"):
            errors.append(f"{name}: terminal primal/dual/slack data missing")

    if receipt.get("claim_eligible") is not False or receipt.get("scientific_verdicts") != {"C1": "HOLD", "C2": "HOLD", "C3": "HOLD"}:
        errors.append("bundle must remain claim-ineligible with all verdicts HOLD")
    squared = receipt_routes.get("squared_regression", {}).get("optimization", {})
    if squared.get("post_solve_square") is not True:
        errors.append("squared_regression: post-solve square flag missing")
    elif abs(squared.get("terminal_lower_objective", 0.0) - squared.get("terminal_root_objective", 0.0) ** 2) > 1e-8:
        errors.append("squared_regression: squared scientific objective mismatch")
    if not _all_finite(receipt):
        errors.append("receipt contains a non-finite numeric value")
    return errors


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("output_dir", type=Path)
    parser.add_argument("--manifest", type=Path, required=True)
    args = parser.parse_args()
    errors = validate(args.output_dir, args.manifest)
    if errors:
        print("APPENDIX CANARY INVALID")
        for error in errors:
            print(f"- {error}")
        return 1
    print("APPENDIX CANARY VALID: manifest, inputs, routes, tasks, datasets, gradients, full conic/OT certificates, terminal states, traces, and HOLD gates pass")
    return 0


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