#!/usr/bin/env python3 """P3-R1 hardware-ready Born-outcome pilot. The standard-library path freezes and audits an append-only prediction/setting/ outcome ledger and runs seeded synthetic controls. IBM submission is optional and imports Qiskit only inside the two hardware commands. No credential is accepted on the command line or written to an artifact. This program is an experiment runner, not evidence that nature contains an outcome-predictive hidden variable. """ from __future__ import annotations import argparse import cmath import hashlib import importlib.metadata import json import math import os from pathlib import Path import random import secrets import shutil import sys import time from typing import Any, Iterable import uuid SCHEMA_VERSION = "p3-hardware-run/1.0" TRIAL_SCHEMA_VERSION = "p3-trial/1.0" PLACEHOLDER_PREFIX = "REPLACE_" class P3Error(RuntimeError): """A fail-closed protocol or provenance error.""" def canonical_bytes(value: Any) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def sha256_value(value: Any) -> str: return hashlib.sha256(canonical_bytes(value)).hexdigest() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1 << 20), b""): digest.update(block) return digest.hexdigest() def load_json(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as handle: value = json.load(handle) if not isinstance(value, dict): raise P3Error(f"{path} must contain one JSON object") return value def write_json(path: Path, value: Any) -> None: temporary = path.with_suffix(path.suffix + ".tmp") with temporary.open("w", encoding="utf-8") as handle: json.dump(value, handle, indent=2, sort_keys=True) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) def write_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> None: temporary = path.with_suffix(path.suffix + ".tmp") with temporary.open("w", encoding="utf-8") as handle: for row in rows: handle.write(canonical_bytes(row).decode() + "\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) def read_jsonl(path: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] with path.open("r", encoding="utf-8") as handle: for line_number, line in enumerate(handle, 1): if not line.strip(): continue value = json.loads(line) if not isinstance(value, dict): raise P3Error(f"{path}:{line_number} is not an object") rows.append(value) return rows def normalize(values: list[float]) -> list[float]: if not values or any(not math.isfinite(x) or x <= 0 for x in values): raise P3Error("probability vectors must be finite and strictly positive") total = sum(values) return [x / total for x in values] def ideal_born_probabilities(config: dict[str, Any]) -> dict[str, list[float]]: """Exact ideal probabilities for the frozen eight-phase state.""" phases = config["experiment"]["state"]["phases_rad_q2q1q0"] if len(phases) != 8: raise P3Error("the Q1 state must contain exactly eight phases") amplitudes = [cmath.exp(1j * float(phi)) / math.sqrt(8) for phi in phases] result: dict[str, list[float]] = {} for setting in config["experiment"]["settings"]: if setting == "Z_ALL": probabilities = [abs(amplitude) ** 2 for amplitude in amplitudes] elif setting == "X_ALL": transformed: list[complex] = [] for y in range(8): total = sum( ((-1) ** ((x & y).bit_count())) * amplitudes[x] for x in range(8) ) / math.sqrt(8) transformed.append(total) probabilities = [abs(amplitude) ** 2 for amplitude in transformed] else: raise P3Error(f"unsupported measurement setting {setting!r}") result[setting] = normalize([max(float(p), 1e-15) for p in probabilities]) return result def permuted_winding(winding: int, permutation: str) -> int: if permutation == "forward": return winding if permutation == "reverse": return (-winding) % 8 if permutation == "gray": return winding ^ (winding >> 1) raise P3Error(f"unsupported winding permutation {permutation!r}") def candidate_outcome_distribution( reference: list[float], candidate: dict[str, Any], trial_id: int, setting: str ) -> list[float]: q = list(reference) winding = trial_id % 8 mapped = permuted_winding(winding, candidate["permutation"]) target = mapped ^ int(candidate["setting_masks"][setting]) donor = target ^ int(candidate["opposite_mask"]) fraction = float(candidate["transfer_fraction"]) transfer = fraction * min(q[target], q[donor]) q[target] += transfer q[donor] -= transfer return normalize(q) def candidate_setting_distribution( candidate: dict[str, Any], trial_id: int, setting_count: int ) -> list[float]: if setting_count != 2: raise P3Error("P3-R1 setting tripwire currently requires two settings") mapped = permuted_winding(trial_id % 8, candidate["permutation"]) preferred = mapped & 1 edge = float(candidate["setting_probe_edge"]) probabilities = [0.5 - edge, 0.5 - edge] probabilities[preferred] = 0.5 + edge return probabilities def expected_grammar_hash(config: dict[str, Any]) -> str: return sha256_value(config["lift_grammar"]["candidates"]) def validate_config(config: dict[str, Any], live: bool) -> tuple[list[str], list[str]]: errors: list[str] = [] blockers: list[str] = [] if config.get("schema_version") != SCHEMA_VERSION: errors.append(f"schema_version must be {SCHEMA_VERSION}") experiment = config.get("experiment", {}) if experiment.get("qubit_count") != 3: errors.append("P3-R1 requires exactly three qubits") if experiment.get("outcomes") != [f"{i:03b}" for i in range(8)]: errors.append("outcomes must be the ordered q2q1q0 bitstrings 000 through 111") if experiment.get("settings") != ["Z_ALL", "X_ALL"]: errors.append("settings must be the frozen order Z_ALL, X_ALL") trial_count = experiment.get("trial_count") if not isinstance(trial_count, int) or trial_count <= 0: errors.append("trial_count must be a positive integer") elif trial_count != config.get("evidence", {}).get("maximum_trials"): errors.append("trial_count and evidence.maximum_trials must agree") probabilities = experiment.get("setting_probabilities") if probabilities != [0.5, 0.5]: errors.append("P3-R1 requires exactly balanced delayed settings") phases = experiment.get("state", {}).get("phases_rad_q2q1q0") if not isinstance(phases, list) or len(phases) != 8: errors.append("exactly eight state phases are required") candidates = config.get("lift_grammar", {}).get("candidates", []) if not candidates: errors.append("the lift grammar is empty") else: kraft = sum(2.0 ** -int(item.get("code_bits", -1)) for item in candidates) cash = float(config.get("evidence", {}).get("mixture_cash_weight", -1)) if abs(kraft + cash - 1.0) > 1e-12: errors.append("Kraft weights plus mixture cash weight must equal one") actual_hash = expected_grammar_hash(config) declared_hash = config.get("lift_grammar", {}).get("candidate_grammar_sha256") if actual_hash != declared_hash: errors.append( f"candidate grammar hash mismatch: declared {declared_hash}, actual {actual_hash}" ) for item in candidates: fraction = item.get("transfer_fraction") edge = item.get("setting_probe_edge") if not isinstance(fraction, (int, float)) or not 0 < fraction < 1: errors.append(f"{item.get('name')}: transfer_fraction must lie in (0,1)") if not isinstance(edge, (int, float)) or not 0 < edge < 0.5: errors.append(f"{item.get('name')}: setting_probe_edge must lie in (0,0.5)") try: reference = ideal_born_probabilities(config) for setting, vector in reference.items(): if abs(sum(vector) - 1.0) > 1e-12: errors.append(f"ideal probabilities for {setting} are not normalized") except (KeyError, TypeError, ValueError, P3Error) as exc: errors.append(f"invalid ideal-state definition: {exc}") run_id = str(config.get("run_id", "")) backend = str(config.get("hardware", {}).get("backend_name", "")) if not run_id or run_id.startswith(PLACEHOLDER_PREFIX): blockers.append("fresh non-placeholder run_id is required") else: try: uuid.UUID(run_id) except ValueError: blockers.append("run_id must be a canonical UUID") if not backend or backend.startswith(PLACEHOLDER_PREFIX): blockers.append("explicit accessible backend_name is required") envelope = config.get("null", {}).get("upper_envelope_by_setting") if envelope is None: blockers.append("calibrated upper_envelope_by_setting is required") elif isinstance(envelope, dict): for setting in experiment.get("settings", []): vector = envelope.get(setting) if not isinstance(vector, list) or len(vector) != 8: errors.append(f"null envelope for {setting} must have eight entries") continue if any(not isinstance(x, (int, float)) or not 0 < x <= 1 for x in vector): errors.append(f"null envelope for {setting} must lie coordinatewise in (0,1]") if sum(vector) < 1 - 1e-12: errors.append(f"null envelope for {setting} cannot sum below one") else: errors.append("upper_envelope_by_setting must be an object or null") if not config.get("null", {}).get("calibration_artifact_sha256"): blockers.append("calibration artifact hash is required") if config.get("status") != "CALIBRATED_NOT_SUBMITTED": blockers.append("status must be CALIBRATED_NOT_SUBMITTED") if config.get("scope", {}).get("spacelike_separation") is not False: errors.append("P3-R1 must remain explicitly single-device and non-spacelike") if live and blockers: errors.extend(f"LIVE BLOCKER: {item}" for item in blockers) return errors, blockers def prepare_run_directory(run_dir: Path) -> None: if run_dir.exists() and any(run_dir.iterdir()): raise P3Error(f"refusing to overwrite nonempty run directory {run_dir}") run_dir.mkdir(parents=True, exist_ok=True) def prediction_record( config: dict[str, Any], config_hash: str, trial_id: int, previous_hash: str | None, committed_ns: int, ) -> dict[str, Any]: reference = ideal_born_probabilities(config) predictions: dict[str, Any] = {} for candidate in config["lift_grammar"]["candidates"]: predictions[candidate["name"]] = { "outcome_by_setting": { setting: candidate_outcome_distribution( reference[setting], candidate, trial_id, setting ) for setting in config["experiment"]["settings"] }, "setting_distribution": candidate_setting_distribution( candidate, trial_id, len(config["experiment"]["settings"]) ), } material = { "schema_version": TRIAL_SCHEMA_VERSION, "run_id": config["run_id"], "trial_id": trial_id, "config_sha256": config_hash, "previous_prediction_sha256": previous_hash, "committed_unix_ns": committed_ns, "predictions_by_candidate_and_setting": predictions, } return {**material, "prediction_sha256": sha256_value(material)} def setting_index_from_rng( config: dict[str, Any], trial_id: int, rng: random.Random | None, setting_mode: str, ) -> int: if setting_mode == "setting-leak": candidate = config["lift_grammar"]["candidates"][0] preferred = permuted_winding(trial_id % 8, candidate["permutation"]) & 1 draw = rng.random() if rng is not None else secrets.randbelow(10_000) / 10_000 return preferred if draw < 0.8 else 1 - preferred if rng is not None: return rng.randrange(2) return secrets.randbelow(2) def freeze_run( config: dict[str, Any], run_dir: Path, rng: random.Random | None = None, setting_mode: str = "independent", calibration_path: Path | None = None, ) -> dict[str, Any]: prepare_run_directory(run_dir) if calibration_path is not None: copied_calibration = run_dir / "calibration.json" shutil.copyfile(calibration_path, copied_calibration) if sha256_file(copied_calibration) != config["null"]["calibration_artifact_sha256"]: raise P3Error("copied calibration digest does not match frozen config") frozen_config = run_dir / "frozen_config.json" write_json(frozen_config, config) config_hash = sha256_file(frozen_config) commitments: list[dict[str, Any]] = [] previous_hash: str | None = None last_ns = time.time_ns() for trial_id in range(config["experiment"]["trial_count"]): last_ns = max(last_ns + 1, time.time_ns()) record = prediction_record(config, config_hash, trial_id, previous_hash, last_ns) commitments.append(record) previous_hash = record["prediction_sha256"] commitment_path = run_dir / "commitments.jsonl" write_jsonl(commitment_path, commitments) settings: list[dict[str, Any]] = [] last_setting_ns = max(last_ns + 1, time.time_ns()) for trial_id, commitment in enumerate(commitments): setting_index = setting_index_from_rng(config, trial_id, rng, setting_mode) last_setting_ns = max(last_setting_ns + 1, time.time_ns()) settings.append( { "schema_version": TRIAL_SCHEMA_VERSION, "run_id": config["run_id"], "trial_id": trial_id, "config_sha256": config_hash, "name": config["experiment"]["settings"][setting_index], "rng_source": ( "seeded_python_random_SYNTHETIC_ONLY" if rng is not None else "python_secrets_system_rng" ), "generated_unix_ns": last_setting_ns, "prediction_sha256": commitment["prediction_sha256"], } ) setting_path = run_dir / "settings.jsonl" write_jsonl(setting_path, settings) manifest = { "schema_version": SCHEMA_VERSION, "run_id": config["run_id"], "status": "PREDICTIONS_AND_SETTINGS_FROZEN", "frozen_config_sha256": config_hash, "commitments_sha256": sha256_file(commitment_path), "settings_sha256": sha256_file(setting_path), "calibration_sha256": ( sha256_file(run_dir / "calibration.json") if calibration_path is not None else None ), "candidate_grammar_sha256": expected_grammar_hash(config), "prediction_chain_tip_sha256": commitments[-1]["prediction_sha256"], "last_prediction_commit_unix_ns": commitments[-1]["committed_unix_ns"], "first_setting_generated_unix_ns": settings[0]["generated_unix_ns"], "external_timestamp_anchor": None, "warning": "Local hashes detect later alteration relative to this manifest; externally timestamp before a court-grade submission.", } write_json(run_dir / "manifest.json", manifest) return manifest def sample_index(probabilities: list[float], rng: random.Random) -> int: draw = rng.random() cumulative = 0.0 for index, probability in enumerate(probabilities): cumulative += probability if draw <= cumulative: return index return len(probabilities) - 1 def make_synthetic_outcomes( run_dir: Path, world: str, rng: random.Random ) -> None: config = load_json(run_dir / "frozen_config.json") manifest = load_json(run_dir / "manifest.json") commitments = read_jsonl(run_dir / "commitments.jsonl") settings = read_jsonl(run_dir / "settings.jsonl") reference = ideal_born_probabilities(config) first_candidate = config["lift_grammar"]["candidates"][0]["name"] outcomes: list[dict[str, Any]] = [] submitted_ns = max(item["generated_unix_ns"] for item in settings) + 1 last_ns = submitted_ns for commitment, setting in zip(commitments, settings): if world == "null": probabilities = reference[setting["name"]] elif world in {"planted", "setting-leak"}: probabilities = commitment["predictions_by_candidate_and_setting"][first_candidate][ "outcome_by_setting" ][setting["name"]] else: raise P3Error(f"unsupported synthetic world {world}") outcome_index = sample_index(probabilities, rng) last_ns += 1 outcomes.append( { "schema_version": TRIAL_SCHEMA_VERSION, "run_id": config["run_id"], "trial_id": commitment["trial_id"], "config_sha256": manifest["frozen_config_sha256"], "bitstring_q2q1q0": config["experiment"]["outcomes"][outcome_index], "backend_name": "synthetic_reference", "provider_job_id": f"synthetic-{world}", "job_submitted_unix_ns": submitted_ns, "result_retrieved_unix_ns": last_ns, "setting_name": setting["name"], } ) write_jsonl(run_dir / "outcomes.jsonl", outcomes) def validate_ledgers(run_dir: Path, require_outcomes: bool = True) -> dict[str, Any]: config_path = run_dir / "frozen_config.json" commitment_path = run_dir / "commitments.jsonl" setting_path = run_dir / "settings.jsonl" manifest_path = run_dir / "manifest.json" for path in (config_path, commitment_path, setting_path, manifest_path): if not path.exists(): raise P3Error(f"missing required artifact {path}") config = load_json(config_path) manifest = load_json(manifest_path) if sha256_file(config_path) != manifest["frozen_config_sha256"]: raise P3Error("frozen config hash mismatch") if sha256_file(commitment_path) != manifest["commitments_sha256"]: raise P3Error("commitment ledger hash mismatch") if sha256_file(setting_path) != manifest["settings_sha256"]: raise P3Error("setting ledger hash mismatch") if expected_grammar_hash(config) != manifest["candidate_grammar_sha256"]: raise P3Error("candidate grammar changed after manifest") is_synthetic = config.get("status") == "SYNTHETIC_DRY_RUN" if not is_synthetic: calibration_path = run_dir / "calibration.json" if not calibration_path.exists(): raise P3Error("live run package is missing its calibration artifact") calibration_hash = sha256_file(calibration_path) if calibration_hash != config["null"]["calibration_artifact_sha256"]: raise P3Error("calibration digest differs from frozen config") if calibration_hash != manifest.get("calibration_sha256"): raise P3Error("calibration digest differs from manifest") calibration = load_json(calibration_path) if calibrated_envelope(config, calibration) != config["null"]["upper_envelope_by_setting"]: raise P3Error("frozen null envelope is not the declared function of calibration") commitments = read_jsonl(commitment_path) settings = read_jsonl(setting_path) expected_count = config["experiment"]["trial_count"] if len(commitments) != expected_count or len(settings) != expected_count: raise P3Error("ledger trial count does not match frozen config") previous_hash: str | None = None seen: set[int] = set() for expected_id, record in enumerate(commitments): if record.get("trial_id") != expected_id or expected_id in seen: raise P3Error("commitment trial IDs are missing, duplicated, or reordered") seen.add(expected_id) if record.get("previous_prediction_sha256") != previous_hash: raise P3Error(f"prediction hash chain breaks at trial {expected_id}") declared = record.get("prediction_sha256") material = {key: value for key, value in record.items() if key != "prediction_sha256"} if sha256_value(material) != declared: raise P3Error(f"prediction commitment changed at trial {expected_id}") if record.get("config_sha256") != manifest["frozen_config_sha256"]: raise P3Error(f"prediction config hash mismatch at trial {expected_id}") previous_hash = declared if previous_hash != manifest["prediction_chain_tip_sha256"]: raise P3Error("prediction chain tip does not match manifest") max_commit_time = max(item["committed_unix_ns"] for item in commitments) min_setting_time = min(item["generated_unix_ns"] for item in settings) if min_setting_time <= max_commit_time: raise P3Error("a delayed setting was generated before all predictions were committed") for expected_id, (commitment, setting) in enumerate(zip(commitments, settings)): if setting.get("trial_id") != expected_id: raise P3Error("setting trial IDs are missing, duplicated, or reordered") if setting.get("prediction_sha256") != commitment["prediction_sha256"]: raise P3Error(f"setting is not linked to trial {expected_id} commitment") if setting.get("name") not in config["experiment"]["settings"]: raise P3Error(f"unknown setting at trial {expected_id}") outcomes: list[dict[str, Any]] | None = None outcome_path = run_dir / "outcomes.jsonl" if require_outcomes: if not outcome_path.exists(): raise P3Error("outcome ledger is missing") outcomes = read_jsonl(outcome_path) if len(outcomes) != expected_count: raise P3Error("outcome count does not match frozen trial count") frozen_backend = config["hardware"]["backend_name"] for expected_id, (setting, outcome) in enumerate(zip(settings, outcomes)): if outcome.get("trial_id") != expected_id: raise P3Error("outcome trial IDs are missing, duplicated, or reordered") if outcome.get("setting_name") != setting["name"]: raise P3Error(f"outcome-setting mismatch at trial {expected_id}") if outcome.get("job_submitted_unix_ns", 0) <= setting["generated_unix_ns"]: raise P3Error(f"job submission precedes setting at trial {expected_id}") if outcome.get("result_retrieved_unix_ns", 0) < outcome["job_submitted_unix_ns"]: raise P3Error(f"result retrieval precedes submission at trial {expected_id}") if outcome.get("bitstring_q2q1q0") not in config["experiment"]["outcomes"]: raise P3Error(f"invalid outcome bitstring at trial {expected_id}") if not is_synthetic and outcome.get("backend_name") != frozen_backend: raise P3Error(f"backend mismatch at trial {expected_id}") return { "config": config, "manifest": manifest, "commitments": commitments, "settings": settings, "outcomes": outcomes, "chronology_gate": "PASS", "hash_gate": "PASS", } def logsumexp(values: list[float]) -> float: peak = max(values) return peak + math.log(sum(math.exp(value - peak) for value in values)) def write_result_manifest(run_dir: Path) -> dict[str, Any]: """Seal post-measurement artifacts without modifying the pre-run manifest.""" pre_manifest = run_dir / "manifest.json" outcome_path = run_dir / "outcomes.jsonl" score_path = run_dir / "score.json" if not outcome_path.exists() or not score_path.exists(): raise P3Error("cannot seal a result before outcomes and score exist") value = { "schema_version": "p3-result-manifest/1.0", "pre_run_manifest_sha256": sha256_file(pre_manifest), "outcomes_sha256": sha256_file(outcome_path), "score_sha256": sha256_file(score_path), "provider_jobs_sha256": ( sha256_file(run_dir / "provider_jobs.jsonl") if (run_dir / "provider_jobs.jsonl").exists() else None ), "sealed_unix_ns": time.time_ns(), } write_json(run_dir / "result_manifest.json", value) return value def score_run(run_dir: Path) -> dict[str, Any]: ledger = validate_ledgers(run_dir, require_outcomes=True) config = ledger["config"] commitments = ledger["commitments"] settings = ledger["settings"] outcomes = ledger["outcomes"] envelope = config["null"]["upper_envelope_by_setting"] if envelope is None: raise P3Error("scoring requires a frozen composite-null envelope") names = [item["name"] for item in config["lift_grammar"]["candidates"]] code_bits = [int(item["code_bits"]) for item in config["lift_grammar"]["candidates"]] outcome_logs = [0.0] * len(names) setting_logs = [0.0] * len(names) outcome_maxima = [0.0] * len(names) setting_maxima = [0.0] * len(names) outcomes_order = config["experiment"]["outcomes"] settings_order = config["experiment"]["settings"] setting_null = config["experiment"]["setting_probabilities"] cash_log = math.log(float(config["evidence"]["mixture_cash_weight"])) weights_log = [-bits * math.log(2) for bits in code_bits] mixture_outcome_max = 0.0 mixture_setting_max = 0.0 for commitment, setting, outcome in zip(commitments, settings, outcomes): y = outcomes_order.index(outcome["bitstring_q2q1q0"]) s = settings_order.index(setting["name"]) for index, name in enumerate(names): prediction = commitment["predictions_by_candidate_and_setting"][name] q_y = prediction["outcome_by_setting"][setting["name"]][y] u_y = envelope[setting["name"]][y] outcome_logs[index] += math.log(q_y / u_y) q_s = prediction["setting_distribution"][s] setting_logs[index] += math.log(q_s / setting_null[s]) outcome_maxima[index] = max(outcome_maxima[index], outcome_logs[index]) setting_maxima[index] = max(setting_maxima[index], setting_logs[index]) mixture_outcome = logsumexp([cash_log] + [w + e for w, e in zip(weights_log, outcome_logs)]) mixture_setting = logsumexp([cash_log] + [w + e for w, e in zip(weights_log, setting_logs)]) mixture_outcome_max = max(mixture_outcome_max, mixture_outcome) mixture_setting_max = max(mixture_setting_max, mixture_setting) alpha_outcome = float(config["evidence"]["outcome_alpha"]) alpha_setting = float(config["evidence"]["setting_leak_alpha"]) log_outcome_threshold = math.log(1 / alpha_outcome) log_setting_threshold = math.log(1 / alpha_setting) outcome_crossing = mixture_outcome_max >= log_outcome_threshold setting_crossing = mixture_setting_max >= log_setting_threshold physical_data = config.get("status") != "SYNTHETIC_DRY_RUN" gates = { "hash": "PASS", "chronology": "PASS", "outcome_evidence": "CROSS" if outcome_crossing else "NO_CROSS", "setting_leakage": "CROSS_DISQUALIFY" if setting_crossing else "NO_CROSS_NOT_EQUIVALENCE", "signal_channel": "NOT_TESTED_SINGLE_DEVICE", "physical_data": "YES" if physical_data else "NO_SYNTHETIC", } if not physical_data: verdict = "SYNTHETIC_ONLY_NO_PHYSICAL_CLAIM" elif outcome_crossing and not setting_crossing: verdict = "DECLARED_SINGLE_DEVICE_NULL_REJECTED_PENDING_ALL_W5_CONTROLS" elif outcome_crossing and setting_crossing: verdict = "DISQUALIFIED_SETTING_LEAKAGE" else: verdict = "NO_DECLARED_LIFT_DETECTED" report = { "schema_version": "p3-score/1.0", "run_id": config["run_id"], "trial_count": len(commitments), "candidate_final_log_e": dict(zip(names, outcome_logs)), "candidate_max_log_e": dict(zip(names, outcome_maxima)), "candidate_final_setting_log_e": dict(zip(names, setting_logs)), "candidate_max_setting_log_e": dict(zip(names, setting_maxima)), "mixture_max_log_e": mixture_outcome_max, "mixture_outcome_log_threshold": log_outcome_threshold, "mixture_max_setting_log_e": mixture_setting_max, "mixture_setting_log_threshold": log_setting_threshold, "gates": gates, "verdict": verdict, "claim_ceiling": config["scope"]["maximum_claim"], } write_json(run_dir / "score.json", report) write_result_manifest(run_dir) return report def require_qiskit() -> tuple[Any, Any, Any, Any]: try: from qiskit import QuantumCircuit from qiskit.circuit.library import DiagonalGate from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager from qiskit_ibm_runtime import QiskitRuntimeService, SamplerV2 except ImportError as exc: raise P3Error( "IBM mode requires qiskit~=2.5.2 and qiskit-ibm-runtime~=0.47.0; " "the standard-library audit and dry-run remain available" ) from exc return QuantumCircuit, DiagonalGate, generate_preset_pass_manager, (QiskitRuntimeService, SamplerV2) def build_circuit(config: dict[str, Any], setting: str) -> Any: QuantumCircuit, DiagonalGate, _, _ = require_qiskit() circuit = QuantumCircuit(3) circuit.h(range(3)) phases = config["experiment"]["state"]["phases_rad_q2q1q0"] circuit.append(DiagonalGate([cmath.exp(1j * float(phi)) for phi in phases]), range(3)) if setting == "X_ALL": circuit.h(range(3)) elif setting != "Z_ALL": raise P3Error(f"unsupported setting {setting}") circuit.measure_all() return circuit def ibm_service_and_backend(config: dict[str, Any]) -> tuple[Any, Any, Any, Any]: _, _, pass_manager_factory, runtime = require_qiskit() QiskitRuntimeService, SamplerV2 = runtime env_name = config["hardware"]["instance_crn_env"] instance = os.environ.get(env_name) service = QiskitRuntimeService(instance=instance) if instance else QiskitRuntimeService() backend = service.backend(config["hardware"]["backend_name"]) return backend, pass_manager_factory, SamplerV2, service def extract_counts(pub_result: Any) -> dict[str, int]: try: raw = pub_result.data.meas.get_counts() except Exception as exc: # provider object has version-dependent concrete types raise P3Error("Sampler result does not expose data.meas.get_counts()") from exc return {str(key).replace(" ", ""): int(value) for key, value in raw.items()} def ibm_calibrate(config_path: Path, output: Path) -> dict[str, Any]: config = load_json(config_path) errors, blockers = validate_config(config, live=False) hardware_blockers = [item for item in blockers if "run_id" in item or "backend" in item] if errors or hardware_blockers: raise P3Error("; ".join(errors + hardware_blockers)) backend, pass_manager_factory, SamplerV2, _ = ibm_service_and_backend(config) circuits = [build_circuit(config, setting) for setting in config["experiment"]["settings"]] manager = pass_manager_factory( backend=backend, optimization_level=config["hardware"]["optimization_level"] ) isa_circuits = [manager.run(circuit) for circuit in circuits] sampler = SamplerV2(mode=backend) submitted_ns = time.time_ns() shots = int(config["experiment"]["calibration_shots_per_setting"]) job = sampler.run(isa_circuits, shots=shots) result = job.result() records = [] for setting, pub_result in zip(config["experiment"]["settings"], result): records.append({"setting": setting, "counts": extract_counts(pub_result), "shots": shots}) artifact = { "schema_version": "p3-calibration/1.0", "run_id": config["run_id"], "backend_name": config["hardware"]["backend_name"], "provider_job_id": job.job_id(), "submitted_unix_ns": submitted_ns, "retrieved_unix_ns": time.time_ns(), "qiskit_version": importlib.metadata.version("qiskit"), "qiskit_ibm_runtime_version": importlib.metadata.version("qiskit-ibm-runtime"), "records": records, "warning": "Calibration sampling bounds do not by themselves bound later hardware drift.", } write_json(output, artifact) return artifact def calibrated_envelope( config: dict[str, Any], calibration: dict[str, Any] ) -> dict[str, list[float]]: """Deterministically reconstruct the predeclared fixed-sample envelope.""" if calibration.get("schema_version") != "p3-calibration/1.0": raise P3Error("unsupported calibration schema") if calibration.get("backend_name") != config["hardware"]["backend_name"]: raise P3Error("calibration backend differs from frozen backend") if calibration.get("run_id") != config["run_id"]: raise P3Error("calibration run_id differs from config run_id") settings = config["experiment"]["settings"] outcomes = config["experiment"]["outcomes"] raw_records = calibration.get("records", []) records = {record["setting"]: record for record in raw_records} if len(raw_records) != len(settings) or set(records) != set(settings): raise P3Error("calibration must contain exactly one record per setting") delta = float(config["null"]["calibration_failure_probability"]) drift = float(config["null"]["predeclared_drift_margin_per_outcome"]) if not 0 < delta < 1 or not 0 <= drift < 1: raise P3Error("invalid calibration failure probability or drift margin") cell_count = len(settings) * len(outcomes) envelope: dict[str, list[float]] = {} for setting in settings: if setting not in records: raise P3Error(f"calibration lacks setting {setting}") record = records[setting] shots = int(record["shots"]) counts = record["counts"] if not isinstance(counts, dict) or not set(counts).issubset(set(outcomes)): raise P3Error(f"invalid calibration outcome key for {setting}") if shots <= 0 or any(int(x) < 0 for x in counts.values()) or sum(int(x) for x in counts.values()) != shots: raise P3Error(f"invalid calibration counts for {setting}") radius = math.sqrt(math.log(cell_count / delta) / (2 * shots)) envelope[setting] = [ min(1.0, int(counts.get(outcome, 0)) / shots + radius + drift) for outcome in outcomes ] return envelope def build_envelope(config_path: Path, calibration_path: Path, output: Path) -> dict[str, Any]: config = load_json(config_path) errors, _ = validate_config(config, live=False) if errors: raise P3Error("; ".join(errors)) calibration = load_json(calibration_path) envelope = calibrated_envelope(config, calibration) activated = json.loads(json.dumps(config)) activated["status"] = "CALIBRATED_NOT_SUBMITTED" activated["null"]["upper_envelope_by_setting"] = envelope activated["null"]["calibration_artifact_sha256"] = sha256_file(calibration_path) write_json(output, activated) return activated def validate_job_coverage(job_records: list[dict[str, Any]], trial_count: int) -> int: """Return the number of contiguously covered trials or fail closed.""" covered = 0 for record in job_records: if int(record["first_trial_id"]) != covered: raise P3Error("provider job coverage is not contiguous") count = int(record["trial_count"]) if count <= 0: raise P3Error("provider job has a nonpositive trial count") covered += count if covered > trial_count: raise P3Error("provider jobs cover more trials than the frozen run") return covered def ibm_retrieve(run_dir: Path) -> None: """Recover immutable outcomes from already submitted provider job IDs.""" ledger = validate_ledgers(run_dir, require_outcomes=False) config = ledger["config"] errors, _ = validate_config(config, live=True) if errors: raise P3Error("; ".join(errors)) if (run_dir / "outcomes.jsonl").exists(): raise P3Error("final outcome ledger already exists") job_path = run_dir / "provider_jobs.jsonl" if not job_path.exists(): raise P3Error("no provider jobs are available for retrieval") job_records = read_jsonl(job_path) settings = ledger["settings"] if validate_job_coverage(job_records, len(settings)) != len(settings): raise P3Error("not every frozen trial has a submitted provider job") _, _, _, service = ibm_service_and_backend(config) outcomes: list[dict[str, Any]] = [] for record in job_records: start = int(record["first_trial_id"]) count = int(record["trial_count"]) batch = settings[start : start + count] job = service.job(record["provider_job_id"]) result = job.result() if len(result) != len(batch): raise P3Error("provider result length differs from submitted PUB count") retrieved_ns = time.time_ns() for setting, pub_result in zip(batch, result): counts = extract_counts(pub_result) if sum(counts.values()) != 1 or len(counts) != 1: raise P3Error("each validation PUB must return exactly one shot") outcomes.append( { "schema_version": TRIAL_SCHEMA_VERSION, "run_id": config["run_id"], "trial_id": setting["trial_id"], "config_sha256": ledger["manifest"]["frozen_config_sha256"], "bitstring_q2q1q0": next(iter(counts)), "backend_name": config["hardware"]["backend_name"], "provider_job_id": record["provider_job_id"], "job_submitted_unix_ns": record["submitted_unix_ns"], "result_retrieved_unix_ns": max(retrieved_ns, record["submitted_unix_ns"]), "setting_name": setting["name"], } ) write_jsonl(run_dir / "outcomes.partial.jsonl", outcomes) write_jsonl(run_dir / "outcomes.jsonl", outcomes) report = score_run(run_dir) print(json.dumps(report, indent=2, sort_keys=True)) def ibm_submit(run_dir: Path) -> None: """Submit every still-unsubmitted batch, then retrieve without duplicates.""" ledger = validate_ledgers(run_dir, require_outcomes=False) config = ledger["config"] errors, _ = validate_config(config, live=True) if errors: raise P3Error("; ".join(errors)) if (run_dir / "outcomes.jsonl").exists(): raise P3Error("final outcome ledger already exists") backend, pass_manager_factory, SamplerV2, _ = ibm_service_and_backend(config) settings = ledger["settings"] manager = pass_manager_factory( backend=backend, optimization_level=config["hardware"]["optimization_level"] ) batch_size = int(config["hardware"]["pub_batch_size"]) job_path = run_dir / "provider_jobs.jsonl" job_records = read_jsonl(job_path) if job_path.exists() else [] covered = validate_job_coverage(job_records, len(settings)) sampler = SamplerV2(mode=backend) for start in range(covered, len(settings), batch_size): batch = settings[start : start + batch_size] circuits = [manager.run(build_circuit(config, item["name"])) for item in batch] submitted_ns = time.time_ns() job = sampler.run(circuits, shots=1) job_records.append( { "provider_job_id": job.job_id(), "backend_name": config["hardware"]["backend_name"], "first_trial_id": start, "trial_count": len(batch), "submitted_unix_ns": submitted_ns, } ) write_jsonl(job_path, job_records) ibm_retrieve(run_dir) def cmd_validate_config(args: argparse.Namespace) -> None: config = load_json(args.config) errors, blockers = validate_config(config, live=args.live) report = { "structural_errors": errors, "live_blockers": blockers, "structural_status": "PASS" if not [e for e in errors if not e.startswith("LIVE BLOCKER")] else "FAIL", "live_status": "READY" if not errors and not blockers else "BLOCKED", "candidate_grammar_sha256_actual": expected_grammar_hash(config), "ideal_born_probabilities": ideal_born_probabilities(config), } print(json.dumps(report, indent=2, sort_keys=True)) if errors: raise P3Error("configuration validation failed") def cmd_dry_run(args: argparse.Namespace) -> None: source = load_json(args.config) errors, _ = validate_config(source, live=False) if errors: raise P3Error("; ".join(errors)) config = json.loads(json.dumps(source)) config["status"] = "SYNTHETIC_DRY_RUN" config["run_id"] = f"dry-{args.world}-{args.seed}" if args.trials is not None: if args.trials <= 0: raise P3Error("--trials must be positive") config["experiment"]["trial_count"] = args.trials config["evidence"]["maximum_trials"] = args.trials config["hardware"]["backend_name"] = "synthetic_reference" config["null"]["upper_envelope_by_setting"] = ideal_born_probabilities(config) config["null"]["calibration_artifact_sha256"] = "0" * 64 rng = random.Random(args.seed) setting_mode = "setting-leak" if args.world == "setting-leak" else "independent" freeze_run(config, args.run_dir, rng=rng, setting_mode=setting_mode) make_synthetic_outcomes(args.run_dir, args.world, rng) report = score_run(args.run_dir) print(json.dumps(report, indent=2, sort_keys=True)) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) validate = subparsers.add_parser("validate-config") validate.add_argument("--config", type=Path, required=True) validate.add_argument("--live", action="store_true", help="turn activation blockers into errors") validate.set_defaults(function=cmd_validate_config) dry = subparsers.add_parser("dry-run") dry.add_argument("--config", type=Path, required=True) dry.add_argument("--run-dir", type=Path, required=True) dry.add_argument("--world", choices=("null", "planted", "setting-leak"), required=True) dry.add_argument("--seed", type=int, required=True) dry.add_argument("--trials", type=int, help="synthetic-only trial-count override") dry.set_defaults(function=cmd_dry_run) freeze = subparsers.add_parser("freeze") freeze.add_argument("--config", type=Path, required=True) freeze.add_argument("--calibration", type=Path, required=True) freeze.add_argument("--run-dir", type=Path, required=True) freeze.set_defaults(function=lambda args: freeze_command(args)) calibrate = subparsers.add_parser("ibm-calibrate") calibrate.add_argument("--config", type=Path, required=True) calibrate.add_argument("--output", type=Path, required=True) calibrate.set_defaults(function=lambda args: print(json.dumps(ibm_calibrate(args.config, args.output), indent=2))) envelope = subparsers.add_parser("build-envelope") envelope.add_argument("--config", type=Path, required=True) envelope.add_argument("--calibration", type=Path, required=True) envelope.add_argument("--output", type=Path, required=True) envelope.set_defaults(function=lambda args: print(json.dumps(build_envelope(args.config, args.calibration, args.output), indent=2))) submit = subparsers.add_parser("ibm-submit") submit.add_argument("--run-dir", type=Path, required=True) submit.set_defaults(function=lambda args: ibm_submit(args.run_dir)) retrieve = subparsers.add_parser("ibm-retrieve") retrieve.add_argument("--run-dir", type=Path, required=True) retrieve.set_defaults(function=lambda args: ibm_retrieve(args.run_dir)) score = subparsers.add_parser("score") score.add_argument("--run-dir", type=Path, required=True) score.set_defaults(function=lambda args: print(json.dumps(score_run(args.run_dir), indent=2, sort_keys=True))) return parser.parse_args() def freeze_command(args: argparse.Namespace) -> None: config = load_json(args.config) errors, _ = validate_config(config, live=True) if errors: raise P3Error("; ".join(errors)) if not args.calibration.exists(): raise P3Error("calibration artifact does not exist") if sha256_file(args.calibration) != config["null"]["calibration_artifact_sha256"]: raise P3Error("calibration digest does not match frozen config") calibration = load_json(args.calibration) if calibration.get("backend_name") != config["hardware"]["backend_name"]: raise P3Error("calibration backend differs from frozen backend") if calibration.get("run_id") != config["run_id"]: raise P3Error("calibration run_id differs from frozen run_id") if calibrated_envelope(config, calibration) != config["null"]["upper_envelope_by_setting"]: raise P3Error("config envelope is not the declared function of calibration") manifest = freeze_run(config, args.run_dir, calibration_path=args.calibration) print(json.dumps(manifest, indent=2, sort_keys=True)) def main() -> None: try: args = parse_args() args.function(args) except P3Error as exc: print(f"P3 REFUSAL: {exc}", file=sys.stderr) raise SystemExit(2) from exc if __name__ == "__main__": main()