"""Validate and analyse the fixed H50 recovery protocol using the standard library. This program checks structure, arithmetic and local evidence-file hashes. Physical calibration and experiment attestations require separate assessment. """ import argparse import hashlib import json import math from pathlib import Path import sys ARMS = ("exchange", "swap", "qmm_output", "global_inverse") AXES = ("XX", "YY", "ZZ") SHOTS = 4096 ALPHA = 0.01 THRESHOLD = 0.562 BIAS_LIMIT = 0.02 PROTOCOL = "h50-fixed-4096-v1" INTAKE_VERSION = "0.1.0" ATTESTATIONS = ( "raw_unmitigated_counts", "independent_shots", "fixed_decoder_and_axes", "no_postselection_or_early_stopping", "settings_randomized_and_interleaved", "protocol_fixed_before_outcomes", ) CALIBRATION_FIELDS = ("b_prep", "epsilon_XX", "epsilon_YY", "epsilon_ZZ", "b_drift") EVIDENCE_ROLES = ("protocol", "native_circuits", "layout", "calibration", "execution", "exclusions") ARM_ROLES = { "exchange": "primary", "swap": "validation_control", "qmm_output": "comparison_only", "global_inverse": "comparison_only", } def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError(f"Duplicate JSON key: {key}") result[key] = value return result def reject_constant(value): raise ValueError(f"Non-finite JSON number: {value}") def read_json(path): return json.loads( Path(path).read_text(encoding="utf-8"), object_pairs_hook=unique_object, parse_constant=reject_constant, ) def nonnegative_number(value, label): if type(value) not in (int, float) or not math.isfinite(value) or value < 0: raise ValueError(f"{label} must be a finite non-negative number") return value def correlator(histogram, label): if not isinstance(histogram, dict) or not histogram: raise ValueError(f"{label} requires a raw count histogram") if set(histogram) - {"00", "01", "10", "11"}: raise ValueError(f"{label} accepts only two-bit outcomes 00, 01, 10, 11") if any(type(value) is not int or value < 0 for value in histogram.values()): raise ValueError(f"{label} requires non-negative integer raw counts") if sum(histogram.values()) != SHOTS: raise ValueError(f"{label} must contain exactly {SHOTS} shots") even = histogram.get("00", 0) + histogram.get("11", 0) odd = histogram.get("01", 0) + histogram.get("10", 0) return (even - odd) / SHOTS def evidence_checks(records, folder, issues): if not isinstance(records, list): raise ValueError("evidence must be a list") verified_roles = set() for index, record in enumerate(records): if not isinstance(record, dict): raise ValueError(f"evidence[{index}] must be an object") role = record.get("role") if role not in EVIDENCE_ROLES: raise ValueError(f"Unknown evidence role: {role}") name = record.get("path") digest = record.get("sha256") if not isinstance(name, str) or not name: issues.append(f"Missing evidence path for {role}") continue path = (folder / name).resolve() if Path(name).is_absolute() or not path.is_relative_to(folder): issues.append(f"Evidence path escapes the run folder for {role}") continue if not isinstance(digest, str) or len(digest) != 64: issues.append(f"Missing SHA-256 digest for {role}") continue if not path.is_file(): issues.append(f"Evidence file unavailable for {role}: {name}") continue checksum = hashlib.sha256() with path.open("rb") as stream: for block in iter(lambda: stream.read(1024 * 1024), b""): checksum.update(block) if checksum.hexdigest() != digest.lower(): issues.append(f"Evidence checksum mismatch for {role}: {name}") continue verified_roles.add(role) for role in EVIDENCE_ROLES: if role not in verified_roles: issues.append(f"Required evidence role unverified: {role}") return sorted(verified_roles) def analyze(data, folder): if not isinstance(data, dict): raise ValueError("The run record must be an object") if data.get("schema_version") != 1 or type(data.get("schema_version")) is not int: raise ValueError("schema_version must be integer 1") if data.get("protocol_id") != PROTOCOL: raise ValueError(f"protocol_id must be {PROTOCOL}") kind = data.get("dataset_kind") if kind not in ("hardware", "archived_hardware", "synthetic"): raise ValueError("dataset_kind must be hardware, archived_hardware or synthetic") counts = data.get("counts") if not isinstance(counts, dict) or set(counts) != set(ARMS): raise ValueError(f"counts must contain exactly the four arms {ARMS}") estimates = {} radius = 0.75 * math.sqrt(2 * math.log(6 / ALPHA) / SHOTS) for arm in ARMS: if not isinstance(counts[arm], dict) or set(counts[arm]) != set(AXES): raise ValueError(f"{arm} requires exactly XX, YY and ZZ") values = {axis: correlator(counts[arm][axis], f"{arm}.{axis}") for axis in AXES} estimate = (1 + values["XX"] - values["YY"] + values["ZZ"]) / 4 estimates[arm] = { "correlators": values, "bell_overlap_estimate": estimate, "numerical_threshold_crossed": estimate > THRESHOLD, "role": ARM_ROLES[arm], } issues = [] if kind == "synthetic": issues.append("Synthetic demonstrations cannot complete the hardware evidence gate") if not isinstance(data.get("run_id"), str) or not data["run_id"].strip(): issues.append("Run identifier is missing") device = data.get("device") if not isinstance(device, dict): raise ValueError("device must be an object") for field in ("provider", "backend"): if not isinstance(device.get(field), str) or not device[field].strip(): issues.append(f"Device {field} is missing") jobs = device.get("job_ids") if not isinstance(jobs, list) or not jobs or any(not isinstance(job, str) or not job.strip() for job in jobs): issues.append("Execution job identifiers are missing") attestations = data.get("attestations") if not isinstance(attestations, dict): raise ValueError("attestations must be an object") if set(attestations) - set(ATTESTATIONS): raise ValueError("Unknown attestation name") for name in ATTESTATIONS: value = attestations.get(name) if value is not None and type(value) is not bool: raise ValueError(f"Attestation {name} must be boolean or null") if value is not True: issues.append(f"Protocol condition requires assessment: {name}") calibration = data.get("calibration") if not isinstance(calibration, dict): raise ValueError("calibration must be an object") bias = None if any(calibration.get(field) is None for field in CALIBRATION_FIELDS): issues.append("Calibration bias allocation is incomplete") else: numbers = {field: nonnegative_number(calibration[field], field) for field in CALIBRATION_FIELDS} bias = numbers["b_prep"] + numbers["b_drift"] + sum(numbers[f"epsilon_{axis}"] for axis in AXES) / 4 if not math.isfinite(bias): raise ValueError("The combined calibration bias must be finite") if bias > BIAS_LIMIT: issues.append("Calibration bias exceeds the fixed 0.02 budget") calibration_failure = calibration.get("bound_failure_probability") if calibration_failure is None: issues.append("Calibration-bound failure probability is unspecified") else: nonnegative_number(calibration_failure, "bound_failure_probability") if calibration_failure > 1: raise ValueError("bound_failure_probability must be at most 1") verified_roles = evidence_checks(data.get("evidence", []), folder.resolve(), issues) if bias is not None: for arm in ("exchange", "swap"): estimate = estimates[arm]["bell_overlap_estimate"] estimates[arm]["individual_conditional_interval"] = [ max(0, min(1, estimate - radius - bias)), max(0, min(1, estimate + radius + bias)), ] ready = not issues primary_pass = estimates["exchange"]["numerical_threshold_crossed"] control_pass = estimates["swap"]["numerical_threshold_crossed"] outcome = "quality_requirements_pending" if ready: if not control_pass: outcome = "validation_control_failed" elif primary_pass: outcome = "conditional_primary_witness_pass" else: outcome = "primary_recovery_not_certified_by_fixed_rule" return { "intake_version": INTAKE_VERSION, "protocol_id": PROTOCOL, "dataset_kind": kind, "run_id": data.get("run_id"), "measurement_shots": len(ARMS) * len(AXES) * SHOTS, "raw_shot_alpha": ALPHA, "statistical_radius": radius, "fixed_observed_threshold": THRESHOLD, "declared_total_bias_bound": bias, "calibration_bound_failure_probability": calibration_failure, "primary_failure_probability_union_bound": None if calibration_failure is None else min(1, ALPHA + calibration_failure), "estimates": estimates, "verified_evidence_roles": verified_roles, "record_checks_complete": ready, "quality_issues": issues, "conditional_rule_outcome": outcome, "level7_status": "pending_gate_assessment", "scope": "Physical validity of attestations and calibration bounds requires separate assessment. File hashes verify file identity. The program assigns no development level.", "uncertainty_scope": "Intervals are individual and conditional on the declared sampling and bias assumptions. Comparison arms have no confirmatory multiplicity claim. A missed threshold alone is insufficient to reject a candidate.", } def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("run", type=Path, help="Path to the completed run JSON") args = parser.parse_args() try: report = analyze(read_json(args.run), args.run.resolve().parent) except (ValueError, OSError) as error: print(f"Invalid run record: {error}", file=sys.stderr) return 2 print(json.dumps(report, indent=2, allow_nan=False)) return 0 if __name__ == "__main__": raise SystemExit(main())