bundle/v0/scripts/validate_scientific_canary.py

#!/usr/bin/env python3
from __future__ import annotations

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


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

from loss_aware_dro_repro.canary import source_tree_hash  # noqa: E402
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


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 validate(canary_dir: Path, manifest_path: Path) -> list[str]:
    errors: list[str] = []
    canary_dir = canary_dir.resolve()
    manifest_path = manifest_path.resolve()
    receipt_path = canary_dir / "receipt.json"
    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}"]

    lineage = receipt.get("raw_lineage", {})
    trace_path = (canary_dir / lineage.get("trace", "")).resolve()
    if not _inside(trace_path, canary_dir) or not trace_path.is_file():
        return ["trace is missing or escapes the immutable canary directory"]
    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:
        return [f"cannot read trace: {exc}"]
    if not trace:
        errors.append("trace must be nonempty")
    if lineage.get("trace_sha256") != _hash(trace_path):
        errors.append("trace hash mismatch")
    if lineage.get("trace_rows") != len(trace):
        errors.append("trace row count mismatch")

    manifest_runs = manifest.get("runs", [])
    matching = [run for run in manifest_runs if (LANE_ROOT / run.get("receipt", "")).resolve() == receipt_path]
    if len(matching) != 1:
        errors.append("manifest must bind exactly one run to this receipt")
        run = {}
    else:
        run = matching[0]
    if run.get("receipt_sha256") != _hash(receipt_path):
        errors.append("manifest receipt hash mismatch")
    if run.get("trace_sha256") != lineage.get("trace_sha256"):
        errors.append("manifest trace hash mismatch")
    if run.get("canary_id") != receipt.get("canary_id"):
        errors.append("manifest canary id mismatch")

    inputs = receipt.get("scientific_inputs", {})
    try:
        config_path = (LANE_ROOT / inputs["config_path"]).resolve()
        plan_path = (LANE_ROOT / inputs["plan_path"]).resolve()
        parameters_path = (LANE_ROOT / inputs["published_parameters_path"]).resolve()
    except KeyError as exc:
        errors.append(f"missing scientific input binding: {exc}")
        config_path = plan_path = parameters_path = LANE_ROOT / "__missing__"
    for label, path in (("config", config_path), ("plan", plan_path), ("published parameters", parameters_path)):
        if not _inside(path, LANE_ROOT) or not path.is_file():
            errors.append(f"{label} input is missing or escaping")
    if config_path.is_file():
        config = load_json(config_path)
        if inputs.get("config_sha256") != _hash(config_path):
            errors.append("selected config byte hash mismatch")
        if receipt.get("canary_config_hash") != sha256_value(config):
            errors.append("selected config canonical hash mismatch")
        if run.get("config") != config_path.relative_to(LANE_ROOT).as_posix():
            errors.append("manifest selected config path mismatch")
        if run.get("config_hash") != receipt.get("canary_config_hash"):
            errors.append("manifest selected config hash mismatch")
        if config.get("canary_id") != receipt.get("canary_id"):
            errors.append("selected config canary id mismatch")
    else:
        config = {}
    if plan_path.is_file():
        plan = load_plan(plan_path)
        if inputs.get("plan_sha256") != _hash(plan_path):
            errors.append("plan byte hash mismatch")
        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")
    else:
        plan = {}
    if parameters_path.is_file() and inputs.get("published_parameters_sha256") != _hash(parameters_path):
        errors.append("published-parameter hash mismatch")
    if config_path.is_file() and receipt.get("source_tree_hash") != source_tree_hash(config_path):
        errors.append("live source/config/input tree differs from receipt")
    if run.get("source_tree_hash") != receipt.get("source_tree_hash"):
        errors.append("manifest source tree hash mismatch")

    if plan and config:
        tasks = {task["task_id"]: task for task in expand_tasks(plan)}
        task = tasks.get(config.get("task_selector"))
        if task is None:
            errors.append("selected task missing from plan")
        else:
            if receipt.get("task_id") != task["task_id"] or receipt.get("task_hash") != task["task_hash"]:
                errors.append("task id/hash binding mismatch")
            samples, metadata = generate_dataset(task)
            if receipt.get("dataset", {}).get("fingerprint") != metadata["fingerprint"]:
                errors.append("dataset fingerprint mismatch")
            if receipt.get("dataset", {}).get("shape") != list(samples.shape):
                errors.append("dataset shape mismatch")
            if receipt.get("dataset", {}).get("seeds") != task["seeds"]:
                errors.append("dataset seed binding mismatch")

    if receipt.get("claim_eligible") is not False:
        errors.append("canary must not be claim eligible")
    if receipt.get("scientific_verdicts") != {"C1": "HOLD", "C2": "HOLD", "C3": "HOLD"}:
        errors.append("scientific verdicts must remain HOLD")
    checks = receipt.get("gradient_checks", {})
    if checks.get("all_pass") is not True:
        errors.append("gradient checks did not pass")
    threshold = checks.get("threshold", 0.0)
    for name in ("gelbrich", "conic_value", "combined_outer_active_penalty"):
        if checks.get(name, {}).get("relative_error", math.inf) > threshold:
            errors.append(f"{name} exceeds gradient threshold")
    if checks.get("combined_outer_active_penalty", {}).get("penalty", 0.0) <= 0:
        errors.append("coverage penalty was not active in its gradient check")
    residual_threshold = receipt.get("solver", {}).get("threshold", 0.0)
    if residual_threshold != config.get("solver_residual_max"):
        errors.append("solver threshold differs from selected config")
    if receipt.get("solver", {}).get("max_residual", math.inf) > residual_threshold:
        errors.append("solver receipt exceeds residual threshold")
    required_residuals = {"primal", "dual", "equality", "cone", "dual_cone", "complementarity", "duality_gap"}
    for index, row in enumerate(trace):
        if row.get("iteration") != index:
            errors.append(f"trace iteration mismatch at row {index}")
        if row.get("solver_status") not in {"optimal", "optimal_inaccurate"}:
            errors.append(f"bad solver status at row {index}")
        residuals = row.get("solver_residuals", {})
        if set(residuals) != required_residuals:
            errors.append(f"incomplete residual certificate at row {index}")
        elif any(not math.isfinite(float(value)) or float(value) > residual_threshold for value in residuals.values()):
            errors.append(f"bad solver residual at row {index}")
        if not math.isfinite(float(row.get("total_objective", math.nan))):
            errors.append(f"non-finite objective at row {index}")
    optimization = receipt.get("optimization", {})
    if optimization.get("iterations") != len(trace):
        errors.append("terminal iteration count mismatch")
    if optimization.get("terminal_solver_status") not in {"optimal", "optimal_inaccurate"}:
        errors.append("terminal solver status is not accepted")
    terminal_residuals = optimization.get("terminal_solver_residuals", {})
    if set(terminal_residuals) != required_residuals:
        errors.append("terminal residual certificate is incomplete")
    elif any(not math.isfinite(float(value)) or float(value) > residual_threshold for value in terminal_residuals.values()):
        errors.append("terminal solver residual exceeds threshold")
    if trace and optimization.get("terminal_L") != trace[-1].get("L_next"):
        errors.append("terminal factor is not bound to the final trace update")
    if trace and optimization.get("stopping") != trace[-1].get("stopping"):
        errors.append("terminal stopping receipt is not bound to the final trace row")
    if config.get("relative_nonpenalized_objective_improvement_tolerance") is not None:
        stopping = optimization.get("stopping", {})
        if stopping.get("tolerance") != config.get("relative_nonpenalized_objective_improvement_tolerance"):
            errors.append("stopping tolerance differs from selected config")
        if stopping.get("reason") not in {"relative_nonpenalized_objective_improvement_below_tolerance", "maximum_outer_iterations_reached"}:
            errors.append("terminal stopping reason is missing or invalid")
    return errors


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("canary_dir", type=Path)
    parser.add_argument("--manifest", type=Path, required=True)
    args = parser.parse_args()
    errors = validate(args.canary_dir, args.manifest)
    if errors:
        print("SCIENTIFIC CANARY INVALID")
        for error in errors:
            print(f"- {error}")
        return 2
    print("SCIENTIFIC CANARY VALID: manifest, inputs, task, dataset, gradient, full residual, terminal, lineage, and HOLD gates pass")
    return 0


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