#!/usr/bin/env python3 """Independent P3-R1 ledger, mathematics, and score verifier. This file deliberately imports no function from p3_hardware_runner.py. It reconstructs the phase-state Born probabilities, dimensional lift grammar, SHA-256 chain, chronology, and e-process score from the stored artifacts. """ from __future__ import annotations import argparse import cmath import hashlib import json import math from pathlib import Path import random import sys from typing import Any class VerificationFailure(RuntimeError): pass def encoded(value: Any) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def object_hash(value: Any) -> str: return hashlib.sha256(encoded(value)).hexdigest() def file_hash(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: while True: block = handle.read(1 << 20) if not block: return digest.hexdigest() digest.update(block) def load_object(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as handle: value = json.load(handle) if not isinstance(value, dict): raise VerificationFailure(f"{path} is not a JSON object") return value def load_rows(path: Path) -> list[dict[str, Any]]: rows = [] with path.open("r", encoding="utf-8") as handle: for number, line in enumerate(handle, 1): if not line.strip(): continue value = json.loads(line) if not isinstance(value, dict): raise VerificationFailure(f"{path}:{number} is not a JSON object") rows.append(value) return rows def renormalize(values: list[float]) -> list[float]: if min(values) <= 0: raise VerificationFailure("nonpositive probability") total = sum(values) return [value / total for value in values] def independent_born(config: dict[str, Any]) -> dict[str, list[float]]: phases = config["experiment"]["state"]["phases_rad_q2q1q0"] source = [cmath.rect(1 / math.sqrt(8), float(phi)) for phi in phases] z = [abs(value) ** 2 for value in source] x = [] for output in range(8): amplitude = 0j for basis, value in enumerate(source): parity = (basis & output).bit_count() % 2 amplitude += (-1 if parity else 1) * value / math.sqrt(8) x.append(abs(amplitude) ** 2) return {"Z_ALL": renormalize(z), "X_ALL": renormalize(x)} def winding_map(index: int, label: str) -> int: if label == "forward": return index if label == "reverse": return (-index) % 8 if label == "gray": return index ^ (index >> 1) raise VerificationFailure(f"unknown permutation {label}") def independent_candidate( baseline: list[float], candidate: dict[str, Any], trial: int, setting: str ) -> list[float]: values = list(baseline) mapped = winding_map(trial % 8, candidate["permutation"]) receiver = mapped ^ int(candidate["setting_masks"][setting]) source = receiver ^ int(candidate["opposite_mask"]) shift = float(candidate["transfer_fraction"]) * min(values[receiver], values[source]) values[receiver] += shift values[source] -= shift return renormalize(values) def independent_setting(candidate: dict[str, Any], trial: int) -> list[float]: preferred = winding_map(trial % 8, candidate["permutation"]) & 1 edge = float(candidate["setting_probe_edge"]) values = [0.5 - edge, 0.5 - edge] values[preferred] = 0.5 + edge return values def close_vectors(left: list[float], right: list[float], tolerance: float = 2e-14) -> bool: return len(left) == len(right) and max(abs(a - b) for a, b in zip(left, right)) <= tolerance def stable_logsumexp(values: list[float]) -> float: maximum = max(values) return maximum + math.log(sum(math.exp(value - maximum) for value in values)) def independent_property_audit(trials: int, seed: int) -> tuple[float, float]: """Stress marginal preservation and the composite-null one-step bound.""" rng = random.Random(seed) max_marginal_error = 0.0 max_composite_surplus = -math.inf labels = ("forward", "reverse", "gray") for _ in range(trials): baseline = renormalize([10 ** rng.uniform(-3, 1) for _ in range(8)]) candidate = { "permutation": rng.choice(labels), "transfer_fraction": rng.uniform(1e-5, 0.95), "opposite_mask": 4, "setting_masks": {"Z_ALL": rng.randrange(8)}, } mean = [0.0] * 8 for winding in range(8): vector = independent_candidate(baseline, candidate, winding, "Z_ALL") if abs(sum(vector) - 1) > 2e-14 or min(vector) <= 0: raise VerificationFailure("randomized candidate lost probability validity") mean = [left + right / 8 for left, right in zip(mean, vector)] max_marginal_error = max( max_marginal_error, max(abs(left - right) for left, right in zip(mean, baseline)), ) null_probability = renormalize([rng.random() + 1e-6 for _ in range(8)]) upper = [min(1.0, value + rng.random() * (1 - value)) for value in null_probability] predictor = renormalize([rng.random() + 1e-6 for _ in range(8)]) multiplier = sum( probability * prediction / bound for probability, prediction, bound in zip(null_probability, predictor, upper) ) max_composite_surplus = max(max_composite_surplus, multiplier - 1) if multiplier > 1 + 2e-14: raise VerificationFailure("composite-null one-step supermartingale bound failed") return max_marginal_error, max_composite_surplus def verify(run_dir: Path, property_trials: int, property_seed: int) -> dict[str, Any]: config_path = run_dir / "frozen_config.json" commitments_path = run_dir / "commitments.jsonl" settings_path = run_dir / "settings.jsonl" outcomes_path = run_dir / "outcomes.jsonl" manifest_path = run_dir / "manifest.json" result_manifest_path = run_dir / "result_manifest.json" for path in (config_path, commitments_path, settings_path, outcomes_path, manifest_path, result_manifest_path): if not path.exists(): raise VerificationFailure(f"missing {path.name}") config = load_object(config_path) manifest = load_object(manifest_path) result_manifest = load_object(result_manifest_path) commitments = load_rows(commitments_path) settings = load_rows(settings_path) outcomes = load_rows(outcomes_path) trial_count = int(config["experiment"]["trial_count"]) if not (len(commitments) == len(settings) == len(outcomes) == trial_count): raise VerificationFailure("ledger lengths disagree with trial count") if file_hash(config_path) != manifest["frozen_config_sha256"]: raise VerificationFailure("frozen config digest mismatch") if file_hash(commitments_path) != manifest["commitments_sha256"]: raise VerificationFailure("commitment file digest mismatch") if file_hash(settings_path) != manifest["settings_sha256"]: raise VerificationFailure("setting file digest mismatch") if file_hash(manifest_path) != result_manifest["pre_run_manifest_sha256"]: raise VerificationFailure("post-run manifest does not bind the pre-run manifest") if file_hash(outcomes_path) != result_manifest["outcomes_sha256"]: raise VerificationFailure("outcome file digest mismatch") candidates = config["lift_grammar"]["candidates"] if object_hash(candidates) != config["lift_grammar"]["candidate_grammar_sha256"]: raise VerificationFailure("config candidate grammar digest mismatch") if object_hash(candidates) != manifest["candidate_grammar_sha256"]: raise VerificationFailure("manifest candidate grammar digest mismatch") kraft = sum(2 ** -int(candidate["code_bits"]) for candidate in candidates) cash = float(config["evidence"]["mixture_cash_weight"]) if abs(kraft + cash - 1) > 1e-14: raise VerificationFailure("Kraft weights and cash do not sum to one") baseline = independent_born(config) config_digest = manifest["frozen_config_sha256"] prior: str | None = None max_commit_ns = 0 for trial, stored in enumerate(commitments): if stored["trial_id"] != trial or stored["config_sha256"] != config_digest: raise VerificationFailure(f"bad prediction identity at trial {trial}") if stored["previous_prediction_sha256"] != prior: raise VerificationFailure(f"broken chain at trial {trial}") material = dict(stored) declared = material.pop("prediction_sha256") if object_hash(material) != declared: raise VerificationFailure(f"bad prediction digest at trial {trial}") predictions = stored["predictions_by_candidate_and_setting"] for candidate in candidates: name = candidate["name"] for setting in config["experiment"]["settings"]: expected = independent_candidate(baseline[setting], candidate, trial, setting) actual = predictions[name]["outcome_by_setting"][setting] if not close_vectors(expected, actual): raise VerificationFailure(f"prediction mismatch: {trial}, {name}, {setting}") if not close_vectors( independent_setting(candidate, trial), predictions[name]["setting_distribution"] ): raise VerificationFailure(f"setting-prediction mismatch: {trial}, {name}") prior = declared max_commit_ns = max(max_commit_ns, int(stored["committed_unix_ns"])) if prior != manifest["prediction_chain_tip_sha256"]: raise VerificationFailure("chain tip mismatch") min_setting_ns = min(int(value["generated_unix_ns"]) for value in settings) if min_setting_ns <= max_commit_ns: raise VerificationFailure("setting generation is not delayed until after all predictions") is_synthetic = config.get("status") == "SYNTHETIC_DRY_RUN" if not is_synthetic: calibration_path = run_dir / "calibration.json" if not calibration_path.exists(): raise VerificationFailure("live package lacks calibration artifact") calibration_digest = file_hash(calibration_path) if calibration_digest != config["null"]["calibration_artifact_sha256"]: raise VerificationFailure("calibration digest differs from config") if calibration_digest != manifest.get("calibration_sha256"): raise VerificationFailure("calibration digest differs from pre-run manifest") calibration = load_object(calibration_path) if calibration.get("run_id") != config["run_id"]: raise VerificationFailure("calibration run_id differs from config") if calibration.get("backend_name") != config["hardware"]["backend_name"]: raise VerificationFailure("calibration backend differs from config") calibration_records = calibration.get("records", []) by_setting = {record["setting"]: record for record in calibration_records} setting_names = config["experiment"]["settings"] outcome_names = config["experiment"]["outcomes"] if len(calibration_records) != len(setting_names) or set(by_setting) != set(setting_names): raise VerificationFailure("calibration does not contain one record per setting") delta = float(config["null"]["calibration_failure_probability"]) drift = float(config["null"]["predeclared_drift_margin_per_outcome"]) cell_count = len(setting_names) * len(outcome_names) expected_envelope: dict[str, list[float]] = {} for setting_name in setting_names: record = by_setting[setting_name] shots = int(record["shots"]) counts = record["counts"] if shots <= 0 or sum(int(value) for value in counts.values()) != shots: raise VerificationFailure("calibration count total is invalid") radius = math.sqrt(math.log(cell_count / delta) / (2 * shots)) expected_envelope[setting_name] = [ min(1.0, int(counts.get(outcome_name, 0)) / shots + radius + drift) for outcome_name in outcome_names ] if expected_envelope != config["null"]["upper_envelope_by_setting"]: raise VerificationFailure("frozen envelope is not reconstructed by calibration") jobs_path = run_dir / "provider_jobs.jsonl" if not jobs_path.exists() or file_hash(jobs_path) != result_manifest.get("provider_jobs_sha256"): raise VerificationFailure("provider job ledger is absent or unsealed") for trial, (commitment, setting, outcome) in enumerate(zip(commitments, settings, outcomes)): if setting["trial_id"] != trial or outcome["trial_id"] != trial: raise VerificationFailure(f"trial order mismatch at {trial}") if setting["prediction_sha256"] != commitment["prediction_sha256"]: raise VerificationFailure(f"setting link mismatch at {trial}") if outcome["setting_name"] != setting["name"]: raise VerificationFailure(f"outcome setting mismatch at {trial}") if int(outcome["job_submitted_unix_ns"]) <= int(setting["generated_unix_ns"]): raise VerificationFailure(f"submission chronology mismatch at {trial}") if int(outcome["result_retrieved_unix_ns"]) < int(outcome["job_submitted_unix_ns"]): raise VerificationFailure(f"retrieval chronology mismatch at {trial}") if not is_synthetic and outcome["backend_name"] != config["hardware"]["backend_name"]: raise VerificationFailure(f"backend mismatch at {trial}") preservation_errors: dict[str, float] = {} for candidate in candidates: for setting in config["experiment"]["settings"]: mean = [0.0] * 8 for winding in range(8): vector = independent_candidate(baseline[setting], candidate, winding, setting) mean = [left + right / 8 for left, right in zip(mean, vector)] preservation_errors[f"{candidate['name']}:{setting}"] = max( abs(left - right) for left, right in zip(mean, baseline[setting]) ) max_preservation_error = max(preservation_errors.values()) if max_preservation_error > 2e-14: raise VerificationFailure("candidate grammar does not preserve eight-cycle marginals") envelope = config["null"]["upper_envelope_by_setting"] if not isinstance(envelope, dict): raise VerificationFailure("null envelope is absent") names = [candidate["name"] for candidate in candidates] outcomes_order = config["experiment"]["outcomes"] settings_order = config["experiment"]["settings"] setting_null = config["experiment"]["setting_probabilities"] outcome_logs = [0.0] * len(candidates) setting_logs = [0.0] * len(candidates) outcome_max = [0.0] * len(candidates) setting_max = [0.0] * len(candidates) mixture_outcome_max = 0.0 mixture_setting_max = 0.0 weight_logs = [-int(candidate["code_bits"]) * math.log(2) for candidate in candidates] cash_log = math.log(cash) for commitment, setting, outcome in zip(commitments, settings, outcomes): observed = outcomes_order.index(outcome["bitstring_q2q1q0"]) chosen = settings_order.index(setting["name"]) for index, name in enumerate(names): stored = commitment["predictions_by_candidate_and_setting"][name] outcome_logs[index] += math.log( stored["outcome_by_setting"][setting["name"]][observed] / envelope[setting["name"]][observed] ) setting_logs[index] += math.log( stored["setting_distribution"][chosen] / setting_null[chosen] ) outcome_max[index] = max(outcome_max[index], outcome_logs[index]) setting_max[index] = max(setting_max[index], setting_logs[index]) mixture_outcome_max = max( mixture_outcome_max, stable_logsumexp([cash_log] + [w + e for w, e in zip(weight_logs, outcome_logs)]), ) mixture_setting_max = max( mixture_setting_max, stable_logsumexp([cash_log] + [w + e for w, e in zip(weight_logs, setting_logs)]), ) outcome_threshold = math.log(1 / float(config["evidence"]["outcome_alpha"])) setting_threshold = math.log(1 / float(config["evidence"]["setting_leak_alpha"])) outcome_cross = mixture_outcome_max >= outcome_threshold setting_cross = mixture_setting_max >= setting_threshold report = { "verification": "PASS", "run_id": config["run_id"], "trial_count": trial_count, "physical_data": not is_synthetic, "max_eight_cycle_marginal_error": max_preservation_error, "candidate_final_log_e": dict(zip(names, outcome_logs)), "candidate_max_log_e": dict(zip(names, outcome_max)), "candidate_final_setting_log_e": dict(zip(names, setting_logs)), "candidate_max_setting_log_e": dict(zip(names, setting_max)), "mixture_max_log_e": mixture_outcome_max, "mixture_outcome_log_threshold": outcome_threshold, "mixture_max_setting_log_e": mixture_setting_max, "mixture_setting_log_threshold": setting_threshold, "outcome_crossing": outcome_cross, "setting_crossing": setting_cross, "signal_channel": "NOT_TESTED_SINGLE_DEVICE", "claim": "NO_PHYSICAL_CLAIM_SYNTHETIC" if is_synthetic else ( "DISQUALIFIED_SETTING_LEAKAGE" if setting_cross else "DECLARED_NULL_REJECTED_PENDING_W5_CONTROLS" if outcome_cross else "NO_DECLARED_LIFT_DETECTED" ), } random_marginal_error, composite_surplus = independent_property_audit( property_trials, property_seed ) report["randomized_property_audit"] = { "trials": property_trials, "seed": property_seed, "max_marginal_error": random_marginal_error, "max_composite_multiplier_minus_one": composite_surplus, "status": "PASS", } score_path = run_dir / "score.json" if score_path.exists(): production = load_object(score_path) if file_hash(score_path) != result_manifest["score_sha256"]: raise VerificationFailure("score file digest mismatch") comparisons = ( (production["mixture_max_log_e"], mixture_outcome_max), (production["mixture_max_setting_log_e"], mixture_setting_max), ) if any(abs(left - right) > 2e-12 for left, right in comparisons): raise VerificationFailure("production and independent scores disagree") report["production_score_agreement"] = "PASS" return report def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--run-dir", type=Path, required=True) parser.add_argument("--property-trials", type=int, default=5000) parser.add_argument("--property-seed", type=int, default=918273) return parser.parse_args() def main() -> None: try: args = parse_args() if args.property_trials <= 0: raise VerificationFailure("property trial count must be positive") report = verify(args.run_dir, args.property_trials, args.property_seed) except (VerificationFailure, KeyError, ValueError, TypeError, json.JSONDecodeError) as exc: print(f"INDEPENDENT VERIFICATION FAIL: {exc}", file=sys.stderr) raise SystemExit(2) from exc print(json.dumps(report, indent=2, sort_keys=True)) if __name__ == "__main__": main()