IBM Quantum בחינם: כך הרצתי מחקר Qiskit על מחשב קוונטי עם 156 קיוביטים

from qiskit import QuantumCircuit
from qiskit.quantum_info import Statevector


def main() -> None:
    circuit = QuantumCircuit(2)
    circuit.h(0)
    circuit.cx(0, 1)

    probabilities = Statevector.from_instruction(circuit).probabilities_dict()
    expected_states = {"00", "11"}

    if set(probabilities) != expected_states:
        raise RuntimeError(f"Unexpected Bell-state result: {probabilities}")

    print(f"Local Bell-state test passed: {probabilities}")


if __name__ == "__main__":
    main()

import argparse
import os
from pathlib import Path

from qiskit_ibm_runtime import QiskitRuntimeService


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
TOKEN_FILE = Path(
    os.environ.get("HQC_IBM_TOKEN_FILE", PACKAGE_ROOT / ".secrets" / "ibm-api-key")
)
INSTANCE_FILE = Path(
    os.environ.get(
        "HQC_IBM_INSTANCE_FILE",
        PACKAGE_ROOT / ".secrets" / "ibm-instance-crn",
    )
)


def private_secret(path: Path, label: str) -> str:
    if not path.is_file():
        raise SystemExit(f"{label} is not installed at {path}")
    mode = path.stat().st_mode & 0o777
    if mode & 0o077:
        raise SystemExit(
            f"{label} permissions are too broad ({mode:o}); expected mode 600"
        )
    value = path.read_text(encoding="utf-8").strip()
    if not value:
        raise SystemExit(f"{label} file is empty")
    return value


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Authenticate and list operational IBM backends. Metadata only; "
            "this file has no sampler.run() call."
        )
    )
    parser.add_argument("--token-file", type=Path, default=TOKEN_FILE)
    parser.add_argument("--instance-file", type=Path, default=INSTANCE_FILE)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    token = private_secret(args.token_file, "IBM Quantum API key")
    instance = private_secret(args.instance_file, "IBM Quantum instance CRN")
    service = QiskitRuntimeService(
        channel="ibm_quantum_platform",
        token=token,
        instance=instance,
    )
    backends = service.backends(operational=True)

    print(f"IBM Quantum connection passed; operational backends: {len(backends)}")


if __name__ == "__main__":
    main()

from __future__ import annotations

from itertools import product
from math import comb
from random import Random

import numpy as np


RNG = Random(20260723)
TOL = 1e-9


def subsets(n: int) -> range:
    return range(1 << n)


def is_subset(s: int, a: int) -> bool:
    return (s & ~a) == 0


def scalar_union_convolution(
    q: dict[int, complex], r: dict[int, complex], n: int
) -> dict[int, complex]:
    result = {s: 0j for s in subsets(n)}
    for u, v in product(subsets(n), repeat=2):
        result[u | v] += q[u] * r[v]
    return result


def matrix_union_convolution(
    q: dict[int, np.ndarray], r: dict[int, np.ndarray], n: int
) -> dict[int, np.ndarray]:
    dim = next(iter(q.values())).shape[0]
    result = {s: np.zeros((dim, dim), dtype=complex) for s in subsets(n)}
    for u, v in product(subsets(n), repeat=2):
        result[u | v] += q[u] @ r[v]
    return result


def module_union_convolution(
    h: dict[int, np.ndarray], psi: dict[int, np.ndarray], n: int
) -> dict[int, np.ndarray]:
    dim = next(iter(psi.values())).shape[0]
    result = {s: np.zeros(dim, dtype=complex) for s in subsets(n)}
    for u, v in product(subsets(n), repeat=2):
        result[u | v] += h[u] @ psi[v]
    return result


def zeta_scalar(q: dict[int, complex], n: int) -> dict[int, complex]:
    return {
        a: sum(q[s] for s in subsets(n) if is_subset(s, a))
        for a in subsets(n)
    }


def zeta_matrix(q: dict[int, np.ndarray], n: int) -> dict[int, np.ndarray]:
    dim = next(iter(q.values())).shape[0]
    return {
        a: sum(
            (q[s] for s in subsets(n) if is_subset(s, a)),
            np.zeros((dim, dim), dtype=complex),
        )
        for a in subsets(n)
    }


def zeta_vector(q: dict[int, np.ndarray], n: int) -> dict[int, np.ndarray]:
    dim = next(iter(q.values())).shape[0]
    return {
        a: sum(
            (q[s] for s in subsets(n) if is_subset(s, a)),
            np.zeros(dim, dtype=complex),
        )
        for a in subsets(n)
    }


def mobius_scalar(e: dict[int, complex], n: int) -> dict[int, complex]:
    return {
        s: sum(
            ((-1) ** ((s.bit_count() - t.bit_count()))) * e[t]
            for t in subsets(n)
            if is_subset(t, s)
        )
        for s in subsets(n)
    }


def mobius_matrix(e: dict[int, np.ndarray], n: int) -> dict[int, np.ndarray]:
    dim = next(iter(e.values())).shape[0]
    result: dict[int, np.ndarray] = {}
    for s in subsets(n):
        value = np.zeros((dim, dim), dtype=complex)
        for t in subsets(n):
            if is_subset(t, s):
                value += ((-1) ** (s.bit_count() - t.bit_count())) * e[t]
        result[s] = value
    return result


def dict_close(
    left: dict[int, complex], right: dict[int, complex], tol: float = TOL
) -> bool:
    return all(abs(left[k] - right[k]) <= tol for k in left)


def matrix_dict_close(
    left: dict[int, np.ndarray], right: dict[int, np.ndarray], tol: float = TOL
) -> bool:
    return all(np.allclose(left[k], right[k], atol=tol, rtol=0) for k in left)


def random_scalar_field(n: int) -> dict[int, complex]:
    return {
        s: complex(RNG.randint(-3, 3), RNG.randint(-3, 3)) for s in subsets(n)
    }


def random_matrix_field(n: int, dim: int = 2) -> dict[int, np.ndarray]:
    return {
        s: np.array(
            [
                [
                    complex(RNG.randint(-2, 2), RNG.randint(-2, 2))
                    for _ in range(dim)
                ]
                for _ in range(dim)
            ],
            dtype=complex,
        )
        for s in subsets(n)
    }


def positive_sqrt(matrix: np.ndarray) -> np.ndarray:
    values, vectors = np.linalg.eigh(matrix)
    values = np.clip(values, 0.0, None)
    return (vectors * np.sqrt(values)) @ vectors.conj().T


def amplitude(
    psi: dict[int, np.ndarray], open_mask: int, n: int
) -> np.ndarray:
    dim = next(iter(psi.values())).shape[0]
    return sum(
        (psi[s] for s in subsets(n) if is_subset(s, open_mask)),
        np.zeros(dim, dtype=complex),
    )


def weight(psi: dict[int, np.ndarray], open_mask: int, n: int) -> float:
    value = amplitude(psi, open_mask, n)
    return float(np.vdot(value, value).real)


def interference(
    psi: dict[int, np.ndarray], blocks: list[int], n: int
) -> float:
    order = len(blocks)
    total = 0.0
    for selected in subsets(order):
        open_mask = 0
        for index, block in enumerate(blocks):
            if selected & (1 << index):
                open_mask |= block
        total += ((-1) ** (order - selected.bit_count())) * weight(
            psi, open_mask, n
        )
    return total


def polynomial_weight_coefficients(m: int) -> tuple[int, int, int]:
    pair_count = comb(m, 2)
    return (m * m, 2 * m * pair_count, pair_count * pair_count)


def polynomial_linear_combination(
    terms: list[tuple[int, tuple[int, int, int]]]
) -> tuple[int, int, int]:
    return tuple(
        sum(multiplier * coefficients[index] for multiplier, coefficients in terms)
        for index in range(3)
    )


def finite_difference(
    function: dict[int, complex], i: int, a: int
) -> complex:
    assert not (a & (1 << i))
    return function[a | (1 << i)] - function[a]


def run() -> None:
    checks: list[str] = []

    for n in range(1, 6):
        q = random_scalar_field(n)
        r = random_scalar_field(n)
        s = random_scalar_field(n)
        left = scalar_union_convolution(
            scalar_union_convolution(q, r, n), s, n
        )
        right = scalar_union_convolution(
            q, scalar_union_convolution(r, s, n), n
        )
        assert dict_close(left, right)

        unit = {mask: (1 + 0j if mask == 0 else 0j) for mask in subsets(n)}
        assert dict_close(scalar_union_convolution(q, unit, n), q)
        assert dict_close(scalar_union_convolution(unit, q, n), q)

        q_zeta = zeta_scalar(q, n)
        r_zeta = zeta_scalar(r, n)
        product_zeta = zeta_scalar(scalar_union_convolution(q, r, n), n)
        assert all(
            abs(product_zeta[a] - q_zeta[a] * r_zeta[a]) <= TOL
            for a in subsets(n)
        )
        assert dict_close(mobius_scalar(q_zeta, n), q)
    checks.append("scalar algebra, unit, zeta homomorphism and Möbius inversion")

    n = 3
    q_m = random_matrix_field(n)
    r_m = random_matrix_field(n)
    q_dagger = {s: value.conj().T for s, value in q_m.items()}
    r_dagger = {s: value.conj().T for s, value in r_m.items()}
    lhs = {
        s: value.conj().T
        for s, value in matrix_union_convolution(q_m, r_m, n).items()
    }
    rhs = matrix_union_convolution(r_dagger, q_dagger, n)
    assert matrix_dict_close(lhs, rhs)
    checks.append("noncommutative involution identity")

    rho = matrix_union_convolution(q_dagger, q_m, n)
    rho_visible = zeta_matrix(rho, n)
    for visible in rho_visible.values():
        assert np.allclose(visible, visible.conj().T, atol=TOL, rtol=0)
        assert np.linalg.eigvalsh(visible).min() >= -TOL
    checks.append("context-wise positivity of q†⋆q")

    positive_contexts: dict[int, np.ndarray] = {}
    for a in subsets(n):
        raw = random_matrix_field(0)[0]
        positive_contexts[a] = raw.conj().T @ raw
    roots = {a: positive_sqrt(value) for a, value in positive_contexts.items()}
    reconstructed_q = mobius_matrix(roots, n)
    reconstructed_q_dagger = {
        s: value.conj().T for s, value in reconstructed_q.items()
    }
    reconstructed_rho = matrix_union_convolution(
        reconstructed_q_dagger, reconstructed_q, n
    )
    assert matrix_dict_close(zeta_matrix(reconstructed_rho, n), positive_contexts)
    checks.append("converse context-positive square representation")

    for n in range(2, 7):
        r_degree = 2
        s_degree = 1
        q = {
            mask: (
                complex(RNG.randint(1, 3), 0)
                if mask.bit_count() <= r_degree
                else 0j
            )
            for mask in subsets(n)
        }
        r = {
            mask: (
                complex(RNG.randint(1, 3), 0)
                if mask.bit_count() <= s_degree
                else 0j
            )
            for mask in subsets(n)
        }
        convolution = scalar_union_convolution(q, r, n)
        assert all(
            abs(value) <= TOL or mask.bit_count() <= r_degree + s_degree
            for mask, value in convolution.items()
        )
    checks.append("degree filtration inequality")

    n = 6
    degree = 2
    psi = {
        mask: (
            np.array(
                [
                    complex(RNG.randint(-2, 2), RNG.randint(-2, 2)),
                    complex(RNG.randint(-2, 2), RNG.randint(-2, 2)),
                ]
            )
            if mask and mask.bit_count() <= degree
            else np.zeros(2, dtype=complex)
        )
        for mask in subsets(n)
    }
    atomic_blocks = [1 << i for i in range(5)]
    assert abs(interference(psi, atomic_blocks, n)) <= 1e-7
    disjoint_blocks = [0b000011, 0b001100, 0b010000, 0b100000, 0]
    assert abs(interference(psi, disjoint_blocks, n)) <= 1e-7
    checks.append("I_n vanishing for n>2r, including disjoint blocks")

    i3_coefficients = polynomial_linear_combination(
        [
            (1, polynomial_weight_coefficients(3)),
            (-3, polynomial_weight_coefficients(2)),
            (3, polynomial_weight_coefficients(1)),
        ]
    )
    i4_coefficients = polynomial_linear_combination(
        [
            (1, polynomial_weight_coefficients(4)),
            (-4, polynomial_weight_coefficients(3)),
            (6, polynomial_weight_coefficients(2)),
            (-4, polynomial_weight_coefficients(1)),
        ]
    )
    assert i3_coefficients == (0, 6, 6)
    assert i4_coefficients == (0, 0, 6)
    checks.append("four-slit formulas I3=6g+6g² and I4=6g²")

    outcomes = []
    for _ in range(4):
        outcome_field = {
            mask: np.array(
                [
                    complex(RNG.randint(-2, 2), RNG.randint(-2, 2)),
                    complex(RNG.randint(-2, 2), RNG.randint(-2, 2)),
                ]
            )
            for mask in subsets(3)
        }
        outcomes.append(outcome_field)
    context = 0b111
    raw_weights = [weight(field, context, 3) for field in outcomes]
    probabilities = [value / sum(raw_weights) for value in raw_weights]
    assert min(probabilities) >= -TOL
    assert abs(sum(probabilities) - 1.0) <= TOL
    checks.append("outcome normalization when denominator is nonzero")

    h = random_matrix_field(3)
    psi_vector = {
        mask: np.array(
            [
                complex(RNG.randint(-2, 2), RNG.randint(-2, 2)),
                complex(RNG.randint(-2, 2), RNG.randint(-2, 2)),
            ]
        )
        for mask in subsets(3)
    }
    visible_module = zeta_vector(module_union_convolution(h, psi_vector, 3), 3)
    visible_h = zeta_matrix(h, 3)
    visible_psi = zeta_vector(psi_vector, 3)
    assert all(
        np.allclose(
            visible_module[a], visible_h[a] @ visible_psi[a], atol=TOL, rtol=0
        )
        for a in subsets(3)
    )
    checks.append("context-wise Schrödinger module identity")

    h_low = {
        0: np.array([[0.0]], dtype=complex),
        1: np.array([[1.0]], dtype=complex),
        2: np.array([[1.0]], dtype=complex),
        3: np.array([[0.0]], dtype=complex),
    }
    psi_low = {
        0: np.array([0.0], dtype=complex),
        1: np.array([1.0], dtype=complex),
        2: np.array([1.0], dtype=complex),
        3: np.array([0.0], dtype=complex),
    }
    derivative = module_union_convolution(h_low, psi_low, 2)
    assert abs(derivative[3][0]) > TOL
    checks.append("counterexample: low-degree Schrödinger dynamics need not preserve HQC_r")

    n = 4
    coefficients = random_scalar_field(n)
    visible = zeta_scalar(coefficients, n)
    recovered: dict[int, complex] = {}
    for s in subsets(n):
        value = 0j
        for t in subsets(n):
            if is_subset(t, s):
                value += ((-1) ** (s.bit_count() - t.bit_count())) * visible[t]
        recovered[s] = value
    assert dict_close(recovered, coefficients)

    f = random_scalar_field(n)
    g = random_scalar_field(n)
    i = 2
    for a in subsets(n):
        if a & (1 << i):
            continue
        fg = {mask: f[mask] * g[mask] for mask in subsets(n)}
        lhs = finite_difference(fg, i, a)
        delta_f = finite_difference(f, i, a)
        delta_g = finite_difference(g, i, a)
        rhs = delta_f * g[a] + f[a] * delta_g + delta_f * delta_g
        assert abs(lhs - rhs) <= TOL
    checks.append("finite-difference coefficient recovery and product rule")

    print("HQC local verification passed")
    for check in checks:
        print(f"PASS: {check}")


if __name__ == "__main__":
    run()

#!/usr/bin/env python3
"""Prepare, transpile, and estimate an HQC/Sorkin control experiment.

SAFETY: This file cannot submit an IBM Quantum workload.  It intentionally
does not import Sampler, Estimator, Session, or Batch, and contains no runtime
``run`` call.  It only reads backend metadata, performs exact local
state-vector checks, transpiles locally, and writes reproducible artifacts.

The experiment is a four-path Sorkin null benchmark:

* two qubits encode four paths;
* one ancilla is a loss/pass flag;
* preparation and detection remain fixed;
* each context changes only the reversible path filter;
* no post-selection or per-context normalization is used.

This can measure the hardware/noise floor for I3 and I4 under ordinary quantum
mechanics.  It cannot by itself establish physics beyond quantum mechanics.
"""

from __future__ import annotations

import argparse
import hashlib
import itertools
import json
import math
import os
import random
from collections import defaultdict
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from qiskit import QuantumCircuit, qpy
from qiskit.circuit import ParameterVector
from qiskit.quantum_info import Statevector
from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager
from qiskit_ibm_runtime import QiskitRuntimeService


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
ROOT = Path(os.environ.get("HQC_WORKDIR", PACKAGE_ROOT / "work" / "new-run"))
TOKEN_FILE = Path(
    os.environ.get("HQC_IBM_TOKEN_FILE", PACKAGE_ROOT / ".secrets" / "ibm-api-key")
)
INSTANCE_FILE = Path(
    os.environ.get(
        "HQC_IBM_INSTANCE_FILE",
        PACKAGE_ROOT / ".secrets" / "ibm-instance-crn",
    )
)
OUTPUT_JSON = ROOT / "HQC_IBM_PARAMETRIC_DRY_RUN_PLAN.json"
OUTPUT_QPY = ROOT / "HQC_IBM_PARAMETRIC_TEMPLATE.qpy"
REJECTED_JSON = ROOT / "HQC_IBM_DRY_RUN_PLAN_CONTEXT_SPECIFIC_REJECTED.json"
REJECTED_QPY = ROOT / "HQC_IBM_TRANSPILED_CONTEXT_SPECIFIC_REJECTED.qpy"

BACKEND_NAMES = ("ibm_fez", "ibm_marrakesh", "ibm_kingston")
SHOTS = 1_024
PRIMARY_BLOCKS = 16
SIGN_CONTROL_BLOCKS = 4
PHASE_CONTROL_BLOCKS = 4
READOUT_REPEATS_PER_POSITION = 4
SEED = 20_260_723
QUICK_USAGE_SECONDS_PER_EXECUTION = 0.00035
QUICK_USAGE_FIXED_OVERHEAD_SECONDS = 2.0
PHASE_CONTROL_LOGICAL = (0.0, math.pi / 5, 2 * math.pi / 5, 3 * math.pi / 5)


def secret(path: Path) -> str:
    if not path.is_file():
        raise SystemExit(f"Missing private credential file: {path}")
    value = path.read_text(encoding="utf-8").strip()
    if not value:
        raise SystemExit(f"Private credential file is empty: {path}")
    return value


def open_paths(mask: int) -> tuple[int, ...]:
    return tuple(i for i in range(4) if mask & (1 << i))


def decompose_uniform_rotation_angles(angles: list[Any], start: int, end: int, reverse: bool) -> None:
    """Symbolic Shende/Möttönen decomposition used by Qiskit's UCRYGate.

    Qiskit's numeric UCRYGate currently rejects unbound parameters.  Keeping
    this small symbolic form lets us transpile one circuit once and provide
    all 16 contexts only as late parameter bindings.
    """

    half = (end - start) // 2
    for index in range(start, start + half):
        first = angles[index]
        second = angles[index + half]
        if not reverse:
            angles[index] = (first + second) / 2
            angles[index + half] = (first - second) / 2
        else:
            angles[index + half] = (first + second) / 2
            angles[index] = (first - second) / 2
    if half > 1:
        decompose_uniform_rotation_angles(angles, start, start + half, False)
        decompose_uniform_rotation_angles(angles, start + half, end, True)


def parametric_context_template() -> tuple[QuantumCircuit, ParameterVector, ParameterVector]:
    """Build one topology-identical template shared by every path context."""

    theta = ParameterVector("theta", 4)
    phase = ParameterVector("phase", 2)
    beta: list[Any] = list(theta)
    decompose_uniform_rotation_angles(beta, 0, len(beta), False)

    qc = QuantumCircuit(3, 3, name="sorkin_four_path_parametric")
    qc.metadata = {
        "kind": "sorkin_parametric_template",
        "parameter_semantics": "theta[path]=0 if open, pi if blocked",
        "phase_semantics": (
            "phase=(delta1-delta0, delta2-delta0); "
            "phase-control uses an additive four-path phase pattern"
        ),
    }

    # Uniform amplitude 1/2 on the four path basis states.
    qc.h(0)
    qc.h(1)

    # Additive path phases remain in the same ISA for Primary, sign-control,
    # and phase-control bindings, without adding another two-qubit gate.
    qc.p(phase[0], 0)
    qc.p(phase[1], 1)

    # Uniformly controlled Ry on the loss flag.  Four symbolic Ry gates and
    # four CNOTs remain in every context; only theta bindings change.
    controls = (0, 1)
    for index, angle in enumerate(beta):
        qc.ry(angle, 2)
        if index != len(beta) - 1:
            value = index + 1
            control_index = (value & -value).bit_length() - 1
        else:
            control_index = len(controls) - 1
        qc.cx(controls[control_index], 2)

    # Project the surviving path amplitude back onto the fixed uniform mode.
    qc.h(0)
    qc.h(1)
    qc.measure((0, 1, 2), (0, 1, 2))
    return qc, theta, phase


def cyclic_mapping(offset: int) -> tuple[int, int, int, int]:
    """Map logical path i to physical code (i+offset) mod 4."""

    return tuple((logical + offset) % 4 for logical in range(4))


def context_binding(
    logical_mask: int,
    mapping: tuple[int, int, int, int],
    closed_angle: float,
    logical_phases: tuple[float, float, float, float],
) -> tuple[list[float], list[float], int]:
    """Return theta, diagonal phase parameters, and the physical open mask."""

    physical_open_mask = 0
    delta = [0.0] * 4
    for logical, physical in enumerate(mapping):
        if logical_mask & (1 << logical):
            physical_open_mask |= 1 << physical
        delta[physical] = logical_phases[logical]

    theta_values = [
        0.0 if physical_open_mask & (1 << physical) else closed_angle
        for physical in range(4)
    ]
    phase_values = [delta[1] - delta[0], delta[2] - delta[0]]
    reconstructed_delta3 = delta[0] + phase_values[0] + phase_values[1]
    if not math.isclose(delta[3], reconstructed_delta3, abs_tol=1e-12):
        raise ValueError(
            "Requested physical path phases are not additive on two path bits"
        )
    return theta_values, phase_values, physical_open_mask


def calibration_circuit(state: int, position: str) -> QuantumCircuit:
    qc = QuantumCircuit(3, 3, name=f"readout_{position}_{state:03b}")
    qc.metadata = {
        "kind": "readout_calibration",
        "prepared_state": state,
        "position": position,
    }
    for qubit in range(3):
        if state & (1 << qubit):
            qc.x(qubit)
    qc.measure((0, 1, 2), (0, 1, 2))
    return qc


def exact_pass_probability(circuit: QuantumCircuit) -> float:
    bare = circuit.remove_final_measurements(inplace=False)
    return float(Statevector.from_instruction(bare).probabilities_dict().get("000", 0.0))


def verify_exact_bindings(
    template: QuantumCircuit,
    theta: ParameterVector,
    phase: ParameterVector,
    rows: list[dict[str, Any]],
) -> dict[str, Any]:
    """Verify every block locally and return ideal Sorkin contrasts."""

    blocks: dict[str, dict[int, float]] = defaultdict(dict)
    block_summaries: list[dict[str, Any]] = []

    for row in rows:
        parameter_map = {
            **dict(zip(theta, row["theta_radians"])),
            **dict(zip(phase, row["phase_radians"])),
        }
        circuit = template.assign_parameters(parameter_map)
        observed = exact_pass_probability(circuit)

        logical_mask = int(row["logical_mask"], 2)
        logical_phases = row["logical_path_phases"]
        amplitude = sum(
            complex(
                math.cos(logical_phases[path]),
                math.sin(logical_phases[path]),
            )
            for path in open_paths(logical_mask)
        )
        expected = abs(amplitude / 4.0) ** 2
        if not math.isclose(observed, expected, rel_tol=0.0, abs_tol=1e-12):
            raise AssertionError(
                f"Local state-vector mismatch for {row['block_id']} "
                f"mask {row['logical_mask']}: {observed} != {expected}"
            )
        blocks[row["block_id"]][logical_mask] = observed
        row["ideal_probability_000"] = observed

    def interference(probabilities: dict[int, float], path_ids: tuple[int, ...]) -> float:
        total = 0.0
        n = len(path_ids)
        for subset_bits in range(1 << n):
            mask = 0
            size = 0
            for offset, path in enumerate(path_ids):
                if subset_bits & (1 << offset):
                    mask |= 1 << path
                    size += 1
            total += ((-1) ** (n - size)) * probabilities[mask]
        return total

    for block_id, probabilities in sorted(blocks.items()):
        if set(probabilities) != set(range(16)):
            raise AssertionError(f"Block {block_id} does not contain all 16 contexts")
        i3_values = {
            "".join(map(str, paths)): interference(probabilities, paths)
            for paths in itertools.combinations(range(4), 3)
        }
        i4 = interference(probabilities, (0, 1, 2, 3))
        if any(abs(value) > 1e-12 for value in (*i3_values.values(), i4)):
            raise AssertionError(
                f"Ordinary-QM Sorkin null identities failed locally in {block_id}"
            )
        block_summaries.append({"block_id": block_id, "i3": i3_values, "i4": i4})

    return {
        "verified_parameter_bindings": len(rows),
        "verified_complete_blocks": len(blocks),
        "block_null_results": block_summaries,
    }


def instruction_property(target: Any, name: str, qargs: tuple[int, ...]) -> Any | None:
    if name not in target.operation_names:
        return None
    return target[name].get(qargs)


def best_line_triplet(backend: Any) -> dict[str, Any]:
    """Pick a low-error connected q0-center-q1 triplet from calibration data."""

    adjacency: dict[int, set[int]] = defaultdict(set)
    for qargs in backend.target["cz"]:
        if len(qargs) == 2:
            a, b = qargs
            adjacency[a].add(b)
            adjacency[b].add(a)

    candidates: list[dict[str, Any]] = []
    for center, neighbors in adjacency.items():
        for left, right in itertools.combinations(sorted(neighbors), 2):
            cz_left = instruction_property(backend.target, "cz", (left, center))
            cz_right = instruction_property(backend.target, "cz", (right, center))
            if cz_left is None:
                cz_left = instruction_property(backend.target, "cz", (center, left))
            if cz_right is None:
                cz_right = instruction_property(backend.target, "cz", (center, right))
            measure_props = [
                instruction_property(backend.target, "measure", (q,))
                for q in (left, right, center)
            ]
            if cz_left is None or cz_right is None or any(p is None for p in measure_props):
                continue
            raw_cz_errors = (cz_left.error, cz_right.error)
            raw_readout_errors = tuple(p.error for p in measure_props)
            if any(value is None for value in (*raw_cz_errors, *raw_readout_errors)):
                continue
            cz_errors = [float(value) for value in raw_cz_errors]
            readout_errors = [float(value) for value in raw_readout_errors]
            if not all(
                math.isfinite(value) and value >= 0.0
                for value in (*cz_errors, *readout_errors)
            ):
                continue
            cz_error = sum(cz_errors)
            readout_error = sum(readout_errors)
            # Weight entangling-gate errors more heavily because the Toffoli
            # decomposition invokes them repeatedly.
            score = 4.0 * cz_error + readout_error
            candidates.append(
                {
                    "layout": [left, right, center],
                    "score": score,
                    "cz_error_sum": cz_error,
                    "cz_errors": cz_errors,
                    "readout_error_sum": readout_error,
                    "readout_errors": readout_errors,
                }
            )
    if not candidates:
        raise RuntimeError(f"No connected three-qubit line found on {backend.name}")
    return min(candidates, key=lambda row: row["score"])


def build_bindings() -> list[dict[str, Any]]:
    rng = random.Random(SEED)
    bindings: list[dict[str, Any]] = []

    # Four strata ensure that Primary, sign, and phase controls appear across
    # the beginning, middle, and end rather than in three long time segments.
    for stratum in range(4):
        block_descriptors: list[dict[str, Any]] = []
        for local_primary in range(4):
            block_index = stratum * 4 + local_primary
            block_descriptors.append(
                {
                    "type": "primary",
                    "index": block_index,
                    "closed_angle": math.pi,
                    "logical_phases": (0.0, 0.0, 0.0, 0.0),
                    "mapping": cyclic_mapping(block_index % 4),
                }
            )
        block_descriptors.extend(
            (
                {
                    "type": "sign",
                    "index": stratum,
                    "closed_angle": -math.pi,
                    "logical_phases": (0.0, 0.0, 0.0, 0.0),
                    "mapping": cyclic_mapping(stratum),
                },
                {
                    "type": "phase",
                    "index": stratum,
                    "closed_angle": math.pi,
                    "logical_phases": PHASE_CONTROL_LOGICAL,
                    "mapping": cyclic_mapping(0),
                },
            )
        )
        rng.shuffle(block_descriptors)

        for descriptor in block_descriptors:
            masks = list(range(16))
            rng.shuffle(masks)
            block_id = f"{descriptor['type']}-{descriptor['index']:02d}"
            mapping = descriptor["mapping"]
            logical_phases = descriptor["logical_phases"]
            for order_in_block, logical_mask in enumerate(masks):
                theta_values, phase_values, physical_mask = context_binding(
                    logical_mask,
                    mapping,
                    descriptor["closed_angle"],
                    logical_phases,
                )
                bindings.append(
                    {
                        "block_id": block_id,
                        "experiment_type": descriptor["type"],
                        "stratum": stratum,
                        "order_in_block": order_in_block,
                        "logical_mask": f"{logical_mask:04b}",
                        "logical_open_paths": list(open_paths(logical_mask)),
                        "logical_to_physical_mapping": list(mapping),
                        "physical_mask": f"{physical_mask:04b}",
                        "closed_angle_radians": descriptor["closed_angle"],
                        "logical_path_phases": list(logical_phases),
                        "theta_radians": theta_values,
                        "phase_radians": phase_values,
                    }
                )

    expected = (
        PRIMARY_BLOCKS + SIGN_CONTROL_BLOCKS + PHASE_CONTROL_BLOCKS
    ) * 16
    if len(bindings) != expected:
        raise AssertionError(f"Expected {expected} parameter bindings, got {len(bindings)}")
    return bindings


def build_calibrations() -> list[QuantumCircuit]:
    circuits: list[QuantumCircuit] = []
    for position in ("before", "after"):
        for repeat in range(READOUT_REPEATS_PER_POSITION):
            for state in range(8):
                circuit = calibration_circuit(state, position)
                circuit.name = f"{circuit.name}_r{repeat}"
                circuit.metadata = dict(circuit.metadata or {})
                circuit.metadata["repeat"] = repeat
                circuits.append(circuit)
    return circuits


def transpile_candidate(
    backend: Any,
    layout: list[int],
    template: QuantumCircuit,
    calibrations: list[QuantumCircuit],
    bindings: list[dict[str, Any]],
) -> dict[str, Any]:
    pass_manager = generate_preset_pass_manager(
        target=backend.target,
        optimization_level=3,
        initial_layout=layout,
        seed_transpiler=SEED,
    )
    transpiled_template = pass_manager.run(template)
    transpiled_calibrations = pass_manager.run(calibrations)

    template_duration = float(
        transpiled_template.estimate_duration(backend.target, unit="s")
    )
    calibration_durations = [
        float(circuit.estimate_duration(backend.target, unit="s"))
        for circuit in transpiled_calibrations
    ]
    operation_totals: dict[str, int] = defaultdict(int)
    for name, count in transpiled_template.count_ops().items():
        operation_totals[name] += int(count) * len(bindings)
    for circuit in transpiled_calibrations:
        for name, count in circuit.count_ops().items():
            operation_totals[name] += int(count)

    effective_circuits = len(bindings) + len(transpiled_calibrations)
    executions = effective_circuits * SHOTS
    exact_baseline = (
        QUICK_USAGE_FIXED_OVERHEAD_SECONDS
        + (
            len(bindings) * (backend.default_rep_delay + template_duration)
            + sum(
                backend.default_rep_delay + duration
                for duration in calibration_durations
            )
        )
        * SHOTS
    )
    quick_estimate = (
        QUICK_USAGE_FIXED_OVERHEAD_SECONDS
        + QUICK_USAGE_SECONDS_PER_EXECUTION * executions
    )
    conservative = max(exact_baseline, quick_estimate) * 1.5

    return {
        "qpy_circuits": [transpiled_template, *transpiled_calibrations],
        "summary": {
            "backend": backend.name,
            "layout_logical_q0_q1_flag": layout,
            "effective_num_circuits": effective_circuits,
            "parameter_bindings": len(bindings),
            "readout_calibration_circuits": len(transpiled_calibrations),
            "qpy_circuits": 1 + len(transpiled_calibrations),
            "shots_per_circuit": SHOTS,
            "num_executions": executions,
            "primary_template_depth": transpiled_template.depth(),
            "primary_template_duration_seconds": template_duration,
            "primary_template_operations": {
                name: int(count)
                for name, count in sorted(transpiled_template.count_ops().items())
            },
            "calibration_depth_min": min(c.depth() for c in transpiled_calibrations),
            "calibration_depth_max": max(c.depth() for c in transpiled_calibrations),
            "calibration_duration_seconds_min": min(calibration_durations),
            "calibration_duration_seconds_max": max(calibration_durations),
            "operation_totals": dict(sorted(operation_totals.items())),
            "usage_estimate_formula_seconds": exact_baseline,
            "usage_estimate_ibm_quick_seconds": quick_estimate,
            "usage_estimate_with_50pct_margin_seconds": conservative,
            "hard_no_go_threshold_seconds": 480.0,
        },
    }


def prepare_online(
    *,
    workdir: Path,
    backend_names: tuple[str, ...],
    token_file: Path,
    instance_file: Path,
    no_overwrite: bool,
) -> None:
    global ROOT, TOKEN_FILE, INSTANCE_FILE, OUTPUT_JSON, OUTPUT_QPY
    global REJECTED_JSON, REJECTED_QPY

    ROOT = workdir.resolve()
    TOKEN_FILE = token_file.resolve()
    INSTANCE_FILE = instance_file.resolve()
    OUTPUT_JSON = ROOT / "HQC_IBM_PARAMETRIC_DRY_RUN_PLAN.json"
    OUTPUT_QPY = ROOT / "HQC_IBM_PARAMETRIC_TEMPLATE.qpy"
    REJECTED_JSON = (
        PACKAGE_ROOT
        / "evidence"
        / "rejected-design"
        / "HQC_IBM_DRY_RUN_PLAN_CONTEXT_SPECIFIC_REJECTED.json"
    )
    REJECTED_QPY = (
        PACKAGE_ROOT
        / "evidence"
        / "rejected-design"
        / "HQC_IBM_TRANSPILED_CONTEXT_SPECIFIC_REJECTED.qpy"
    )
    if no_overwrite and (OUTPUT_JSON.exists() or OUTPUT_QPY.exists()):
        raise SystemExit(
            "Refusing to overwrite an existing plan/QPY. Choose a fresh "
            "--workdir or pass --overwrite explicitly."
        )
    ROOT.mkdir(parents=True, exist_ok=True)
    template, theta, phase = parametric_context_template()
    bindings = build_bindings()
    exact = verify_exact_bindings(template, theta, phase, bindings)
    calibrations = build_calibrations()

    service = QiskitRuntimeService(
        channel="ibm_quantum_platform",
        token=secret(TOKEN_FILE),
        instance=secret(INSTANCE_FILE),
    )

    backend_rows: list[dict[str, Any]] = []
    prepared: dict[str, dict[str, Any]] = {}
    for name in backend_names:
        backend = service.backend(name)
        status = backend.status()
        triplet = best_line_triplet(backend)
        backend_properties = backend.properties()
        calibration_timestamp = backend_properties.last_update_date
        calibration_age_seconds = (
            datetime.now(UTC) - calibration_timestamp
        ).total_seconds()
        qubit_properties = [
            backend.target.qubit_properties[qubit]
            for qubit in triplet["layout"]
        ]
        compiled = transpile_candidate(
            backend,
            triplet["layout"],
            template,
            calibrations,
            bindings,
        )
        row = {
            **compiled["summary"],
            "operational": bool(status.operational),
            "pending_jobs_at_preparation": int(status.pending_jobs),
            "triplet_calibration_score": triplet["score"],
            "triplet_cz_error_sum": triplet["cz_error_sum"],
            "triplet_cz_errors": triplet["cz_errors"],
            "triplet_readout_error_sum": triplet["readout_error_sum"],
            "triplet_readout_errors": triplet["readout_errors"],
            "calibration_timestamp_utc": calibration_timestamp.isoformat(),
            "calibration_age_seconds_at_preparation": calibration_age_seconds,
            "qubit_t1_seconds": [float(props.t1) for props in qubit_properties],
            "qubit_t2_seconds": [float(props.t2) for props in qubit_properties],
        }
        row["quality_gates_snapshot"] = {
            "max_cz_error_below_0_015": max(row["triplet_cz_errors"]) < 0.015,
            "max_readout_error_below_0_03": (
                max(row["triplet_readout_errors"]) < 0.03
            ),
            "circuit_below_10pct_min_t1_t2": (
                row["primary_template_duration_seconds"]
                < 0.1
                * min(
                    *row["qubit_t1_seconds"],
                    *row["qubit_t2_seconds"],
                )
            ),
            "calibration_younger_than_4h": calibration_age_seconds < 4 * 3600,
            "no_more_than_6_entanglers": (
                row["primary_template_operations"].get("cz", 0) <= 6
            ),
        }
        backend_rows.append(row)
        prepared[name] = compiled

    viable = [
        row
        for row in backend_rows
        if row["operational"]
        and all(row["quality_gates_snapshot"].values())
        and row["usage_estimate_with_50pct_margin_seconds"]
        <= row["hard_no_go_threshold_seconds"]
    ]
    if not viable:
        raise RuntimeError("No operational candidate backend")
    selected = min(
        viable,
        key=lambda row: (
            row["triplet_calibration_score"],
            row["usage_estimate_formula_seconds"],
        ),
    )
    selected_compiled = prepared[selected["backend"]]["qpy_circuits"]

    with OUTPUT_QPY.open("wb") as handle:
        qpy.dump(selected_compiled, handle)

    qpy_sha256 = hashlib.sha256(OUTPUT_QPY.read_bytes()).hexdigest()
    plan = {
        "schema": "hqc-sorkin-parametric-dry-run-plan-v3",
        "prepared_at_utc": datetime.now(UTC).isoformat(),
        "safety": {
            "hardware_job_submitted": False,
            "qpu_seconds_consumed_by_this_script": 0,
            "requires_netanel_explicit_final_approval": True,
            "purpose": (
                "Ordinary-QM four-path Sorkin null benchmark and hardware noise-floor "
                "measurement; not a proof of HQC or beyond-QM physics."
            ),
            "rejected_context_specific_primary": True,
            "hard_no_go_if_reestimated_above_seconds": 480.0,
        },
        "experiment": {
            "logical_qubits": 3,
            "contexts": 16,
            "primary_blocks": PRIMARY_BLOCKS,
            "sign_control_blocks": SIGN_CONTROL_BLOCKS,
            "phase_control_blocks": PHASE_CONTROL_BLOCKS,
            "readout_repeats_per_position": READOUT_REPEATS_PER_POSITION,
            "readout_calibration_circuits": len(calibrations),
            "effective_total_circuits": len(bindings) + len(calibrations),
            "single_shared_parametric_primary_template": True,
            "parameter_order": [
                *(str(parameter) for parameter in theta),
                *(str(parameter) for parameter in phase),
            ],
            "parameter_bindings": bindings,
            "shots_per_circuit": SHOTS,
            "fixed_preparation_and_detector": True,
            "postselection": False,
            "per_context_normalization": False,
            "runtime_submission_contract": {
                "pubs": 88,
                "schedule": (
                    "32 readout-before PUBs; 24 experiment-block PUBs with "
                    "16 parameter bindings each; 32 readout-after PUBs"
                ),
                "effective_configurations": len(bindings) + len(calibrations),
                "gate_twirling": False,
                "measurement_twirling": False,
                "dynamical_decoupling": False,
                "classified_measurement": True,
                "max_execution_time_seconds": 480,
                "automatic_retry": False,
            },
            "expected_standard_qm": exact,
        },
        "backend_candidates": backend_rows,
        "selected_snapshot": selected,
        "artifacts": {
            "transpiled_qpy": str(OUTPUT_QPY),
            "transpiled_qpy_sha256": qpy_sha256,
            "rejected_context_specific_plan": str(REJECTED_JSON),
            "rejected_context_specific_qpy": str(REJECTED_QPY),
        },
        "limitations": [
            "A generic IBM QPU implements standard quantum mechanics.",
            "A nonzero measured I3 or I4 can be caused by noise, drift, readout error, or context-dependent compiled circuits.",
            "Programming HQC target probabilities into context-specific circuits would demonstrate an encoding, not new physics.",
            "Backend calibration can change; repeat metadata selection and transpilation immediately before an approved submission.",
            "The usage estimates are formulas and safety heuristics, not a billing guarantee.",
            "Gate and measurement twirling are explicitly disabled in the primary job so server-side randomizations cannot differ across path contexts.",
            "The guarded runtime runner requires a fresh approval capsule, an exact approval hash, an exact confirmation phrase, and an atomic one-job lock.",
        ],
    }
    OUTPUT_JSON.write_text(
        json.dumps(plan, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    OUTPUT_JSON.chmod(0o600)
    OUTPUT_QPY.chmod(0o600)

    print("LOCAL/METADATA-ONLY PREPARATION PASSED")
    print("No IBM Quantum workload was submitted; QPU usage: 0 seconds")
    print(f"Selected metadata snapshot: {selected['backend']}")
    print(f"Physical layout [path0, path1, flag]: {selected['layout_logical_q0_q1_flag']}")
    print(
        f"Effective circuits: {selected['effective_num_circuits']}; "
        f"shots each: {selected['shots_per_circuit']}"
    )
    print(
        "Conservative estimate with 50% margin: "
        f"{selected['usage_estimate_with_50pct_margin_seconds']:.1f} seconds"
    )
    print(f"Plan: {OUTPUT_JSON}")
    print(f"Transpiled snapshot: {OUTPUT_QPY}")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Prepare a new IBM metadata snapshot. This command never submits "
            "a QPU job, but it does contact IBM and writes new artifacts."
        )
    )
    subparsers = parser.add_subparsers(dest="command", required=True)
    prepare = subparsers.add_parser(
        "prepare-online",
        help="contact IBM for metadata and write a new plan/QPY under work/new-run",
    )
    prepare.add_argument(
        "--workdir",
        type=Path,
        default=ROOT,
        help="new-run directory; frozen evidence is never used as an output",
    )
    prepare.add_argument(
        "--backend",
        action="append",
        dest="backends",
        help="candidate backend; repeat for multiple backends",
    )
    prepare.add_argument("--token-file", type=Path, default=TOKEN_FILE)
    prepare.add_argument("--instance-file", type=Path, default=INSTANCE_FILE)
    overwrite = prepare.add_mutually_exclusive_group()
    overwrite.add_argument(
        "--no-overwrite",
        action="store_true",
        dest="no_overwrite",
        help="refuse existing plan/QPY (the default)",
    )
    overwrite.add_argument(
        "--overwrite",
        action="store_false",
        dest="no_overwrite",
        help="explicitly replace plan/QPY inside the selected workdir",
    )
    prepare.set_defaults(no_overwrite=True)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    if args.command == "prepare-online":
        prepare_online(
            workdir=args.workdir,
            backend_names=tuple(args.backends or BACKEND_NAMES),
            token_file=args.token_file,
            instance_file=args.instance_file,
            no_overwrite=args.no_overwrite,
        )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Guarded IBM Runtime runner for the HQC/Sorkin null benchmark.

The default command is a fully local structural check.  Metadata preflight is
read-only.  A hardware submission is possible only through the ``submit``
subcommand and only when all of the following are present:

* a fresh preflight approval capsule;
* the exact approval hash printed by that preflight;
* the literal one-job confirmation phrase;
* no pre-existing atomic submission-intent lock.

No retry path exists.  Collection of an existing job is a separate read-only
operation and can never create a new job.
"""

from __future__ import annotations

import argparse
import hashlib
import hmac
import json
import math
import os
import platform
import shutil
import sys
import tempfile
from collections import OrderedDict
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any, Iterable

import numpy as np
import qiskit
import qiskit_ibm_runtime
import scipy
from qiskit import QuantumCircuit, qpy
from qiskit.primitives.containers import BitArray, BindingsArray, SamplerPub
from qiskit_ibm_runtime import QiskitRuntimeService, SamplerV2
from qiskit_ibm_runtime.json import RuntimeEncoder
from qiskit_ibm_runtime.options import SamplerOptions
from qiskit_ibm_runtime.utils import validate_isa_circuits


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
ROOT = Path(os.environ.get("HQC_WORKDIR", PACKAGE_ROOT / "work" / "new-run"))
PLAN_FILE = ROOT / "HQC_IBM_PARAMETRIC_DRY_RUN_PLAN.json"
QPY_FILE = ROOT / "HQC_IBM_PARAMETRIC_TEMPLATE.qpy"
ANALYZER_FILE = PACKAGE_ROOT / "portable-source" / "hqc_analyze_counts.py"
DESIGN_FILE = PACKAGE_ROOT / "research" / "HQC_IBM_EXPERIMENT_DESIGN.md"
RESEARCH_DRAFT_FILE = PACKAGE_ROOT / "research" / "HQC_REVISED_RESEARCH_DRAFT_HE.md"
REQUIREMENTS_FILE = PACKAGE_ROOT / "requirements.txt"
CAPSULE_FILE = ROOT / "HQC_FINAL_APPROVAL_CAPSULE.json"
MANIFEST_FILE = ROOT / "HQC_SUBMISSION_MANIFEST.json"
INTENT_FILE = ROOT / "HQC_SUBMISSION_INTENT.json"
RECEIPT_FILE = ROOT / "HQC_SUBMISSION_RECEIPT.json"

TOKEN_FILE = Path(
    os.environ.get("HQC_IBM_TOKEN_FILE", PACKAGE_ROOT / ".secrets" / "ibm-api-key")
)
INSTANCE_FILE = Path(
    os.environ.get(
        "HQC_IBM_INSTANCE_FILE",
        PACKAGE_ROOT / ".secrets" / "ibm-instance-crn",
    )
)

PLAN_SCHEMA = "hqc-sorkin-parametric-dry-run-plan-v3"
RAW_SCHEMA = "hqc-ibm-raw-counts-v2"
MANIFEST_SCHEMA = "hqc-ibm-submission-manifest-v1"
CAPSULE_SCHEMA = "hqc-ibm-approval-capsule-v1"
RECEIPT_SCHEMA = "hqc-ibm-submission-receipt-v1"
EXPORT_SCHEMA = "hqc-ibm-export-receipt-v1"

EXPECTED_SHOTS = 1_024
EXPECTED_BINDINGS = 384
EXPECTED_CALIBRATIONS = 64
EXPECTED_CONFIGURATIONS = 448
EXPECTED_BLOCKS = {"primary": 16, "sign": 4, "phase": 4}
MAX_EXECUTION_SECONDS = 480
CAPSULE_TTL_MINUTES = 20
CONFIRMATION_PHRASE = "I_APPROVE_ONE_QPU_JOB"
APPROVAL_DOMAIN = "HQC-IBM-NULL-APPROVAL-V1"


def configure_workdir(path: Path) -> None:
    """Point every mutable/new-run artifact at one explicit work directory."""

    global ROOT, PLAN_FILE, QPY_FILE, CAPSULE_FILE, MANIFEST_FILE
    global INTENT_FILE, RECEIPT_FILE

    ROOT = path.resolve()
    PLAN_FILE = ROOT / "HQC_IBM_PARAMETRIC_DRY_RUN_PLAN.json"
    QPY_FILE = ROOT / "HQC_IBM_PARAMETRIC_TEMPLATE.qpy"
    CAPSULE_FILE = ROOT / "HQC_FINAL_APPROVAL_CAPSULE.json"
    MANIFEST_FILE = ROOT / "HQC_SUBMISSION_MANIFEST.json"
    INTENT_FILE = ROOT / "HQC_SUBMISSION_INTENT.json"
    RECEIPT_FILE = ROOT / "HQC_SUBMISSION_RECEIPT.json"


def utc_now() -> datetime:
    return datetime.now(UTC)


def iso_utc(value: datetime | None = None) -> str:
    return (value or utc_now()).astimezone(UTC).isoformat()


def canonical_json(value: Any) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def sha256_file(path: Path) -> str:
    return sha256_bytes(path.read_bytes())


def required_finite_number(
    value: Any,
    label: str,
    *,
    minimum: float | None = None,
    maximum: float | None = None,
    strictly_positive: bool = False,
) -> float:
    """Coerce required calibration metadata without treating missing data as zero."""

    if value is None or isinstance(value, bool):
        raise ValueError(f"Missing numeric calibration value: {label}")
    try:
        number = float(value)
    except (TypeError, ValueError) as error:
        raise ValueError(f"Invalid numeric calibration value: {label}") from error
    if not math.isfinite(number):
        raise ValueError(f"Non-finite calibration value: {label}")
    if strictly_positive and number <= 0.0:
        raise ValueError(f"Calibration value must be positive: {label}")
    if minimum is not None and number < minimum:
        raise ValueError(f"Calibration value is below its valid range: {label}")
    if maximum is not None and number > maximum:
        raise ValueError(f"Calibration value is above its valid range: {label}")
    return number


def write_private_json(path: Path, payload: Any, *, exclusive: bool = False) -> None:
    data = json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
    if exclusive:
        descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        try:
            with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
                handle.write(data)
                handle.flush()
                os.fsync(handle.fileno())
        except BaseException:
            raise
        return

    temporary = path.with_name(f".{path.name}.tmp-{os.getpid()}")
    descriptor = os.open(
        temporary,
        os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
        0o600,
    )
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        path.chmod(0o600)
    finally:
        if temporary.exists():
            temporary.unlink()


def write_private_runtime_json(path: Path, payload: Any) -> None:
    """Write a RuntimeEncoder payload completely before it can be published."""

    data = (
        json.dumps(
            payload,
            cls=RuntimeEncoder,
            ensure_ascii=False,
            indent=2,
        )
        + "\n"
    )
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
    except BaseException:
        raise


def private_secret(path: Path) -> str:
    if not path.is_file():
        raise RuntimeError(f"Missing private credential file: {path}")
    mode = path.stat().st_mode & 0o777
    if mode & 0o077:
        raise RuntimeError(f"Credential permissions are too broad ({mode:o}): {path}")
    value = path.read_text(encoding="utf-8").strip()
    if not value:
        raise RuntimeError(f"Credential file is empty: {path}")
    return value


def load_json(path: Path) -> dict[str, Any]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError(f"Expected a JSON object in {path}")
    return payload


@dataclass(frozen=True)
class PubSpec:
    pub_index: int
    kind: str
    circuit_index: int
    effective_configurations: int
    block_id: str | None = None
    experiment_type: str | None = None
    binding_rows: tuple[dict[str, Any], ...] = ()
    calibration: dict[str, Any] | None = None

    def serializable(self) -> dict[str, Any]:
        result: dict[str, Any] = {
            "pub_index": self.pub_index,
            "kind": self.kind,
            "circuit_index": self.circuit_index,
            "effective_configurations": self.effective_configurations,
        }
        if self.block_id is not None:
            result["block_id"] = self.block_id
        if self.experiment_type is not None:
            result["experiment_type"] = self.experiment_type
        if self.binding_rows:
            result["binding_indices"] = [
                int(row["binding_index"]) for row in self.binding_rows
            ]
        if self.calibration is not None:
            result["calibration"] = dict(self.calibration)
        return result


def load_artifacts() -> tuple[dict[str, Any], list[QuantumCircuit]]:
    if not PLAN_FILE.is_file() or not QPY_FILE.is_file():
        raise RuntimeError("Missing frozen plan or QPY artifact")
    plan = load_json(PLAN_FILE)
    if plan.get("schema") != PLAN_SCHEMA:
        raise ValueError(f"Unexpected plan schema: {plan.get('schema')!r}")
    with QPY_FILE.open("rb") as handle:
        circuits = list(qpy.load(handle))
    return plan, circuits


def validate_counts_shape(plan: dict[str, Any]) -> list[dict[str, Any]]:
    experiment = plan.get("experiment")
    if not isinstance(experiment, dict):
        raise ValueError("Plan is missing experiment metadata")
    if int(experiment.get("shots_per_circuit", -1)) != EXPECTED_SHOTS:
        raise ValueError("Frozen plan does not use exactly 1,024 shots")
    if int(experiment.get("effective_total_circuits", -1)) != EXPECTED_CONFIGURATIONS:
        raise ValueError("Frozen plan does not contain 448 effective configurations")

    rows = experiment.get("parameter_bindings")
    if not isinstance(rows, list) or len(rows) != EXPECTED_BINDINGS:
        raise ValueError("Frozen plan must contain exactly 384 parameter bindings")

    expected_types: dict[str, str] = {}
    type_counts: dict[str, int] = {name: 0 for name in EXPECTED_BLOCKS}
    seen_keys: set[tuple[str, str]] = set()
    seen_orders: set[tuple[str, int]] = set()
    blocks: OrderedDict[str, list[dict[str, Any]]] = OrderedDict()

    normalized: list[dict[str, Any]] = []
    for binding_index, original in enumerate(rows):
        if not isinstance(original, dict):
            raise ValueError(f"Binding {binding_index} is not an object")
        row = dict(original)
        row["binding_index"] = binding_index
        block_id = str(row.get("block_id", ""))
        experiment_type = str(row.get("experiment_type", ""))
        logical_mask = str(row.get("logical_mask", ""))
        order_in_block = row.get("order_in_block")
        if experiment_type not in EXPECTED_BLOCKS:
            raise ValueError(f"Unknown experiment type in binding {binding_index}")
        if not block_id.startswith(f"{experiment_type}-"):
            raise ValueError(f"Block/type mismatch in binding {binding_index}")
        if len(logical_mask) != 4 or any(bit not in "01" for bit in logical_mask):
            raise ValueError(f"Invalid logical mask in binding {binding_index}")
        if not isinstance(order_in_block, int) or not 0 <= order_in_block < 16:
            raise ValueError(f"Invalid order_in_block in binding {binding_index}")
        key = (block_id, logical_mask)
        order_key = (block_id, order_in_block)
        if key in seen_keys or order_key in seen_orders:
            raise ValueError(f"Duplicate binding key in block {block_id}")
        seen_keys.add(key)
        seen_orders.add(order_key)
        if block_id in expected_types and expected_types[block_id] != experiment_type:
            raise ValueError(f"Inconsistent type for block {block_id}")
        expected_types[block_id] = experiment_type
        blocks.setdefault(block_id, []).append(row)
        normalized.append(row)

    for block_id, block_rows in blocks.items():
        if len(block_rows) != 16:
            raise ValueError(f"Block {block_id} does not have 16 bindings")
        if {row["logical_mask"] for row in block_rows} != {
            f"{mask:04b}" for mask in range(16)
        }:
            raise ValueError(f"Block {block_id} does not contain all masks")
        if {int(row["order_in_block"]) for row in block_rows} != set(range(16)):
            raise ValueError(f"Block {block_id} has an invalid local order")
        type_counts[str(block_rows[0]["experiment_type"])] += 1

    if type_counts != EXPECTED_BLOCKS:
        raise ValueError(f"Unexpected block counts: {type_counts}")
    return normalized


def validate_qpy(
    plan: dict[str, Any],
    circuits: list[QuantumCircuit],
) -> tuple[QuantumCircuit, dict[tuple[str, int, int], tuple[int, QuantumCircuit]]]:
    expected_hash = (
        plan.get("artifacts", {}).get("transpiled_qpy_sha256")
        if isinstance(plan.get("artifacts"), dict)
        else None
    )
    actual_hash = sha256_file(QPY_FILE)
    if expected_hash != actual_hash:
        raise ValueError("QPY hash does not match the frozen plan")
    if len(circuits) != 65:
        raise ValueError(f"Expected 65 QPY circuits, got {len(circuits)}")

    template = circuits[0]
    expected_parameter_names = set(plan["experiment"]["parameter_order"])
    actual_parameter_names = {str(parameter) for parameter in template.parameters}
    if actual_parameter_names != expected_parameter_names or len(actual_parameter_names) != 6:
        raise ValueError(
            f"Template parameter mismatch: {sorted(actual_parameter_names)}"
        )

    selected = plan.get("selected_snapshot")
    if not isinstance(selected, dict):
        raise ValueError("Frozen plan has no selected backend snapshot")
    layout = selected.get("layout_logical_q0_q1_flag")
    if (
        not isinstance(layout, list)
        or len(layout) != 3
        or len(set(map(int, layout))) != 3
    ):
        raise ValueError("Frozen plan has no valid [P0, P1, L] layout")
    expected_layout = list(map(int, layout))
    expected_qubits = set(expected_layout)
    exact_cz_edges = {
        (expected_layout[0], expected_layout[2]),
        (expected_layout[1], expected_layout[2]),
    }
    for circuit_index, circuit in enumerate(circuits):
        if circuit.layout is None:
            raise ValueError(f"QPY circuit {circuit_index} has no transpile layout")
        final_layout = list(
            map(
                int,
                circuit.layout.final_index_layout(filter_ancillas=True),
            )
        )
        if final_layout != expected_layout:
            raise ValueError(
                f"QPY circuit {circuit_index} layout {final_layout} does not "
                f"match frozen layout {expected_layout}"
            )
        used_qubits = {
            circuit.find_bit(qubit).index
            for instruction in circuit.data
            if instruction.operation.name != "barrier"
            for qubit in instruction.qubits
        }
        if used_qubits != expected_qubits:
            raise ValueError(
                f"QPY circuit {circuit_index} uses physical qubits "
                f"{sorted(used_qubits)}, expected {sorted(expected_qubits)}"
            )
        for instruction in circuit.data:
            if len(instruction.qubits) != 2:
                continue
            qargs = tuple(
                circuit.find_bit(qubit).index
                for qubit in instruction.qubits
            )
            if instruction.operation.name != "cz" or qargs not in exact_cz_edges:
                raise ValueError(
                    f"QPY circuit {circuit_index} has unexpected two-qubit "
                    f"instruction {instruction.operation.name}{qargs}"
                )
    if int(template.count_ops().get("cz", 0)) != 4:
        raise ValueError("Parametric template must contain exactly four CZ gates")

    calibrations: dict[tuple[str, int, int], tuple[int, QuantumCircuit]] = {}
    for circuit_index, circuit in enumerate(circuits[1:], start=1):
        metadata = dict(circuit.metadata or {})
        if metadata.get("kind") != "readout_calibration":
            raise ValueError(f"QPY circuit {circuit_index} is not a calibration")
        position = str(metadata.get("position"))
        repeat = metadata.get("repeat")
        state = metadata.get("prepared_state")
        if position not in {"before", "after"}:
            raise ValueError(f"Invalid calibration position at circuit {circuit_index}")
        if not isinstance(repeat, int) or repeat not in range(4):
            raise ValueError(f"Invalid calibration repeat at circuit {circuit_index}")
        if not isinstance(state, int) or state not in range(8):
            raise ValueError(f"Invalid prepared state at circuit {circuit_index}")
        key = (position, repeat, state)
        if key in calibrations:
            raise ValueError(f"Duplicate QPY calibration {key}")
        if circuit.parameters:
            raise ValueError(f"Calibration circuit {circuit_index} is parameterized")
        calibrations[key] = (circuit_index, circuit)

    expected_keys = {
        (position, repeat, state)
        for position in ("before", "after")
        for repeat in range(4)
        for state in range(8)
    }
    if set(calibrations) != expected_keys:
        raise ValueError("QPY calibration set is incomplete")
    return template, calibrations


def row_parameter_map(row: dict[str, Any]) -> dict[str, float]:
    values = {
        f"theta[{index}]": float(value)
        for index, value in enumerate(row["theta_radians"])
    }
    values.update(
        {
            f"phase[{index}]": float(value)
            for index, value in enumerate(row["phase_radians"])
        }
    )
    return values


def build_pub_specs(
    rows: list[dict[str, Any]],
    calibrations: dict[tuple[str, int, int], tuple[int, QuantumCircuit]],
) -> list[PubSpec]:
    specs: list[PubSpec] = []

    def add_calibrations(position: str) -> None:
        for repeat in range(4):
            for state in range(8):
                circuit_index, _ = calibrations[(position, repeat, state)]
                specs.append(
                    PubSpec(
                        pub_index=len(specs),
                        kind="readout_calibration",
                        circuit_index=circuit_index,
                        effective_configurations=1,
                        calibration={
                            "position": position,
                            "repeat": repeat,
                            "prepared_state": state,
                        },
                    )
                )

    add_calibrations("before")

    blocks: OrderedDict[str, list[dict[str, Any]]] = OrderedDict()
    for row in rows:
        blocks.setdefault(str(row["block_id"]), []).append(row)
    for block_id, block_rows in blocks.items():
        ordered_rows = tuple(
            sorted(block_rows, key=lambda row: int(row["order_in_block"]))
        )
        specs.append(
            PubSpec(
                pub_index=len(specs),
                kind="experiment",
                circuit_index=0,
                effective_configurations=16,
                block_id=block_id,
                experiment_type=str(ordered_rows[0]["experiment_type"]),
                binding_rows=ordered_rows,
            )
        )

    add_calibrations("after")

    if len(specs) != 88:
        raise AssertionError(f"Expected 88 PUBs, got {len(specs)}")
    if sum(spec.effective_configurations for spec in specs) != EXPECTED_CONFIGURATIONS:
        raise AssertionError("PUB schedule does not contain 448 configurations")
    if [spec.pub_index for spec in specs] != list(range(len(specs))):
        raise AssertionError("PUB indices are not contiguous")
    return specs


def build_sampler_pubs(
    template: QuantumCircuit,
    calibrations: dict[tuple[str, int, int], tuple[int, QuantumCircuit]],
    specs: list[PubSpec],
) -> list[SamplerPub]:
    by_circuit_index = {
        circuit_index: circuit
        for circuit_index, circuit in calibrations.values()
    }
    parameter_by_name = {str(parameter): parameter for parameter in template.parameters}
    ordered_parameters = tuple(
        parameter_by_name[name]
        for name in sorted(parameter_by_name)
    )

    pubs: list[SamplerPub] = []
    for spec in specs:
        if spec.kind == "readout_calibration":
            circuit = by_circuit_index[spec.circuit_index]
            pubs.append(SamplerPub(circuit, shots=EXPECTED_SHOTS))
            continue
        matrix = np.asarray(
            [
                [
                    row_parameter_map(row)[str(parameter)]
                    for parameter in ordered_parameters
                ]
                for row in spec.binding_rows
            ],
            dtype=float,
        )
        bindings = BindingsArray({ordered_parameters: matrix})
        pub = SamplerPub(
            template,
            parameter_values=bindings,
            shots=EXPECTED_SHOTS,
        )
        if pub.parameter_values.shape != (16,):
            raise ValueError(
                f"Unexpected binding shape for {spec.block_id}: "
                f"{pub.parameter_values.shape}"
            )
        pubs.append(pub)
    return pubs


def schedule_payload(specs: Iterable[PubSpec]) -> list[dict[str, Any]]:
    return [spec.serializable() for spec in specs]


def structural_bundle() -> dict[str, Any]:
    plan, circuits = load_artifacts()
    rows = validate_counts_shape(plan)
    template, calibrations = validate_qpy(plan, circuits)
    specs = build_pub_specs(rows, calibrations)
    pubs = build_sampler_pubs(template, calibrations, specs)
    schedule = schedule_payload(specs)
    schedule_digest = sha256_bytes(canonical_json(schedule))
    return {
        "plan": plan,
        "circuits": circuits,
        "rows": rows,
        "template": template,
        "calibrations": calibrations,
        "specs": specs,
        "pubs": pubs,
        "schedule": schedule,
        "schedule_sha256": schedule_digest,
    }


def instruction_property(target: Any, name: str, qargs: tuple[int, ...]) -> Any | None:
    if name not in target.operation_names:
        return None
    return target[name].get(qargs)


def validate_frozen_circuits_against_target(
    circuits: list[QuantumCircuit],
    target: Any,
    layout: list[int],
) -> None:
    """Require every frozen QPY circuit and exact directed CZ edge to be ISA."""

    validate_isa_circuits(circuits, target)
    left, right, center = map(int, layout)
    for edge in ((left, center), (right, center)):
        if instruction_property(target, "cz", edge) is None:
            raise ValueError(
                f"Frozen QPY requires unavailable directed CZ edge {edge}"
            )


def current_backend_snapshot(
    service: QiskitRuntimeService,
    backend_name: str,
    layout: list[int],
    circuits: list[QuantumCircuit],
) -> dict[str, Any]:
    backend = service.backend(backend_name)
    status = backend.status()
    if len(layout) != 3 or len(set(layout)) != 3:
        raise ValueError("Invalid physical layout")
    if len(circuits) != 65:
        raise ValueError("Expected all 65 frozen QPY circuits for target validation")
    template = circuits[0]
    validate_frozen_circuits_against_target(
        circuits,
        backend.target,
        layout,
    )
    left, right, center = map(int, layout)

    cz_rows: list[dict[str, Any]] = []
    for edge in ((left, center), (right, center)):
        prop = instruction_property(backend.target, "cz", edge)
        if prop is None:
            raise ValueError(f"Missing CZ edge for {edge}")
        cz_rows.append(
            {
                "qargs": list(edge),
                "error": required_finite_number(
                    prop.error,
                    f"CZ error {edge}",
                    minimum=0.0,
                    maximum=1.0,
                ),
                "duration_seconds": required_finite_number(
                    prop.duration,
                    f"CZ duration {edge}",
                    strictly_positive=True,
                ),
            }
        )

    readout_rows: list[dict[str, Any]] = []
    for qubit in layout:
        prop = instruction_property(backend.target, "measure", (int(qubit),))
        if prop is None:
            raise ValueError(f"Missing measurement calibration for qubit {qubit}")
        readout_rows.append(
            {
                "qubit": int(qubit),
                "error": required_finite_number(
                    prop.error,
                    f"readout error qubit {qubit}",
                    minimum=0.0,
                    maximum=1.0,
                ),
                "duration_seconds": required_finite_number(
                    prop.duration,
                    f"readout duration qubit {qubit}",
                    strictly_positive=True,
                ),
            }
        )

    properties = backend.properties()
    calibration_timestamp = properties.last_update_date
    if calibration_timestamp is None or calibration_timestamp.tzinfo is None:
        raise ValueError("Backend calibration timestamp is missing or timezone-naive")
    age_seconds = (utc_now() - calibration_timestamp).total_seconds()
    qubit_properties = [backend.target.qubit_properties[int(q)] for q in layout]
    if any(item is None for item in qubit_properties):
        raise ValueError("Backend qubit properties are incomplete for the frozen layout")
    t1 = [
        required_finite_number(
            item.t1,
            f"T1 qubit {qubit}",
            strictly_positive=True,
        )
        for qubit, item in zip(layout, qubit_properties, strict=True)
    ]
    t2 = [
        required_finite_number(
            item.t2,
            f"T2 qubit {qubit}",
            strictly_positive=True,
        )
        for qubit, item in zip(layout, qubit_properties, strict=True)
    ]
    duration = required_finite_number(
        template.estimate_duration(backend.target, unit="s"),
        "transpiled template duration",
        strictly_positive=True,
    )
    rep_delay = required_finite_number(
        backend.default_rep_delay,
        "backend default repetition delay",
        minimum=0.0,
    )
    operation_counts = {
        str(name): int(count)
        for name, count in sorted(template.count_ops().items())
    }
    quality_gates = {
        "operational": bool(status.operational),
        "calibration_younger_than_4h": 0.0 <= age_seconds < 4 * 3600,
        "max_cz_error_below_0_015": max(row["error"] for row in cz_rows) < 0.015,
        "max_readout_error_below_0_03": (
            max(row["error"] for row in readout_rows) < 0.03
        ),
        "no_swap": operation_counts.get("swap", 0) == 0,
        "no_more_than_6_entanglers": operation_counts.get("cz", 0) <= 6,
        "circuit_below_10pct_min_t1_t2": duration < 0.1 * min(*t1, *t2),
    }
    return {
        "backend": backend.name,
        "operational": bool(status.operational),
        "pending_jobs": int(status.pending_jobs),
        "layout_logical_q0_q1_flag": list(map(int, layout)),
        "calibration_timestamp_utc": calibration_timestamp.isoformat(),
        "calibration_age_seconds": age_seconds,
        "cz": cz_rows,
        "readout": readout_rows,
        "t1_seconds": t1,
        "t2_seconds": t2,
        "circuit_duration_seconds": duration,
        "template_depth": int(template.depth()),
        "template_operations": operation_counts,
        "default_rep_delay_seconds": rep_delay,
        "quality_gates": quality_gates,
    }


def runtime_options_payload() -> dict[str, Any]:
    return {
        "default_shots": EXPECTED_SHOTS,
        "max_execution_time": MAX_EXECUTION_SECONDS,
        "dynamical_decoupling": {"enable": False},
        "execution": {
            "init_qubits": True,
            "meas_type": "classified",
        },
        "twirling": {
            "enable_gates": False,
            "enable_measure": False,
        },
    }


def estimate_usage(snapshot: dict[str, Any]) -> dict[str, float]:
    executions = EXPECTED_CONFIGURATIONS * EXPECTED_SHOTS
    duration = float(snapshot["circuit_duration_seconds"])
    rep_delay = float(snapshot["default_rep_delay_seconds"])
    formula = 2.0 + executions * (rep_delay + duration)
    ibm_quick = 2.0 + executions * 0.00035
    conservative = 1.5 * max(formula, ibm_quick)
    return {
        "executions": float(executions),
        "metadata_formula_seconds": formula,
        "ibm_quick_formula_seconds": ibm_quick,
        "with_50pct_margin_seconds": conservative,
        "one_ms_per_execution_seconds": executions * 0.001,
        "hard_max_seconds": float(MAX_EXECUTION_SECONDS),
    }


def approval_relevant_snapshot(snapshot: dict[str, Any]) -> dict[str, Any]:
    """Remove only fields expected to drift without changing the approved run."""

    payload = json.loads(json.dumps(snapshot))
    payload.pop("pending_jobs", None)
    payload.pop("calibration_age_seconds", None)
    quality_gates = payload.get("quality_gates")
    if isinstance(quality_gates, dict):
        quality_gates.pop("calibration_younger_than_4h", None)
    return payload


def verify_approved_snapshot_is_current(
    current: dict[str, Any],
    frozen: dict[str, Any],
    capsule: dict[str, Any],
) -> None:
    """Require a new approval for any meaningful calibration/estimate change."""

    if approval_relevant_snapshot(current) != approval_relevant_snapshot(frozen):
        raise PermissionError(
            "Approval-relevant backend snapshot changed; run preflight again"
        )
    current_estimate = estimate_usage(current)
    if current_estimate != capsule.get("usage_estimate"):
        raise PermissionError(
            "Approved QPU estimate changed; run preflight again"
        )


def package_versions() -> dict[str, str]:
    return {
        "python": platform.python_version(),
        "qiskit": qiskit.__version__,
        "qiskit_ibm_runtime": qiskit_ibm_runtime.__version__,
        "numpy": np.__version__,
        "scipy": scipy.__version__,
    }


def build_manifest(
    bundle: dict[str, Any],
    backend_snapshot: dict[str, Any] | None,
) -> dict[str, Any]:
    manifest_core: dict[str, Any] = {
        "schema": MANIFEST_SCHEMA,
        "created_at_utc": iso_utc(),
        "purpose": (
            "Standard-QM four-path Sorkin I3/I4 null benchmark and IBM "
            "hardware noise-floor measurement; not a proof of HQC."
        ),
        "artifacts": {
            "plan": str(PLAN_FILE),
            "plan_sha256": sha256_file(PLAN_FILE),
            "qpy": str(QPY_FILE),
            "qpy_sha256": sha256_file(QPY_FILE),
            "runner": str(Path(__file__).resolve()),
            "runner_sha256": sha256_file(Path(__file__).resolve()),
            "analyzer": str(ANALYZER_FILE),
            "analyzer_sha256": (
                sha256_file(ANALYZER_FILE) if ANALYZER_FILE.is_file() else None
            ),
            "design": str(DESIGN_FILE),
            "design_sha256": (
                sha256_file(DESIGN_FILE) if DESIGN_FILE.is_file() else None
            ),
            "research_draft": str(RESEARCH_DRAFT_FILE),
            "research_draft_sha256": (
                sha256_file(RESEARCH_DRAFT_FILE)
                if RESEARCH_DRAFT_FILE.is_file()
                else None
            ),
            "requirements": str(REQUIREMENTS_FILE),
            "requirements_sha256": sha256_file(REQUIREMENTS_FILE),
        },
        "experiment": {
            "shots_per_configuration": EXPECTED_SHOTS,
            "parameter_bindings": EXPECTED_BINDINGS,
            "readout_calibrations": EXPECTED_CALIBRATIONS,
            "effective_configurations": EXPECTED_CONFIGURATIONS,
            "pubs": len(bundle["specs"]),
            "schedule_sha256": bundle["schedule_sha256"],
            "schedule": bundle["schedule"],
            "postselection": False,
            "per_context_normalization": False,
        },
        "runtime_options": runtime_options_payload(),
        "versions": package_versions(),
        "backend_snapshot": backend_snapshot,
        "safety": {
            "default_is_local_check": True,
            "preflight_is_metadata_only": True,
            "explicit_approval_hash_required": True,
            "explicit_confirmation_phrase_required": True,
            "atomic_single_submission_lock": str(INTENT_FILE),
            "automatic_retry": False,
        },
    }
    digest = sha256_bytes(canonical_json(manifest_core))
    return {**manifest_core, "manifest_sha256": digest}


def verify_manifest_digest(manifest: dict[str, Any]) -> str:
    """Verify that every manifest field is covered by its canonical digest."""

    stored = str(manifest.get("manifest_sha256", ""))
    if len(stored) != 64:
        raise PermissionError("Submission manifest has no valid SHA-256 digest")
    core = {
        key: value
        for key, value in manifest.items()
        if key != "manifest_sha256"
    }
    actual = sha256_bytes(canonical_json(core))
    if not hmac.compare_digest(actual, stored):
        raise PermissionError("Submission manifest failed canonical digest verification")
    return stored


def verify_current_artifacts(
    bundle: dict[str, Any],
    manifest: dict[str, Any],
) -> None:
    """Bind collection/submission to the exact frozen files and PUB schedule."""

    artifacts = manifest.get("artifacts")
    experiment = manifest.get("experiment")
    if not isinstance(artifacts, dict) or not isinstance(experiment, dict):
        raise PermissionError("Submission manifest is missing frozen artifacts")
    current_hashes = {
        "plan_sha256": sha256_file(PLAN_FILE),
        "qpy_sha256": sha256_file(QPY_FILE),
        "runner_sha256": sha256_file(Path(__file__).resolve()),
        "analyzer_sha256": (
            sha256_file(ANALYZER_FILE) if ANALYZER_FILE.is_file() else None
        ),
        "design_sha256": (
            sha256_file(DESIGN_FILE) if DESIGN_FILE.is_file() else None
        ),
        "research_draft_sha256": (
            sha256_file(RESEARCH_DRAFT_FILE)
            if RESEARCH_DRAFT_FILE.is_file()
            else None
        ),
        "requirements_sha256": sha256_file(REQUIREMENTS_FILE),
    }
    for field, current in current_hashes.items():
        if artifacts.get(field) != current:
            raise PermissionError(f"Frozen artifact changed: {field}")
    if experiment.get("schedule_sha256") != bundle.get("schedule_sha256"):
        raise PermissionError("Frozen PUB schedule digest changed")
    if experiment.get("schedule") != bundle.get("schedule"):
        raise PermissionError("Frozen PUB schedule payload changed")


def verify_capsule_manifest_consistency(
    capsule: dict[str, Any],
    manifest: dict[str, Any],
) -> None:
    """Ensure the approved target/options/estimate are those used by the manifest."""

    manifest_digest = verify_manifest_digest(manifest)
    if capsule.get("manifest_sha256") != manifest_digest:
        raise PermissionError("Capsule/manifest digest mismatch")
    if capsule.get("backend_snapshot") != manifest.get("backend_snapshot"):
        raise PermissionError("Capsule/manifest backend snapshot mismatch")
    if capsule.get("runtime_options") != manifest.get("runtime_options"):
        raise PermissionError("Capsule/manifest Runtime options mismatch")
    snapshot = manifest.get("backend_snapshot")
    if not isinstance(snapshot, dict):
        raise PermissionError("Submission manifest has no backend snapshot")
    if capsule.get("usage_estimate") != estimate_usage(snapshot):
        raise PermissionError("Capsule/manifest QPU estimate mismatch")


def build_capsule(manifest: dict[str, Any]) -> dict[str, Any]:
    if manifest.get("backend_snapshot") is None:
        raise ValueError("A metadata-only backend snapshot is required for approval")
    created = utc_now()
    expires = created + timedelta(minutes=CAPSULE_TTL_MINUTES)
    capsule_core = {
        "schema": CAPSULE_SCHEMA,
        "created_at_utc": iso_utc(created),
        "expires_at_utc": iso_utc(expires),
        "manifest_sha256": manifest["manifest_sha256"],
        "backend_snapshot": manifest["backend_snapshot"],
        "usage_estimate": estimate_usage(manifest["backend_snapshot"]),
        "runtime_options": manifest["runtime_options"],
        "approval_scope": (
            "Exactly one IBM Runtime Sampler job for the frozen standard-QM "
            "Sorkin null benchmark; no retry or mutation."
        ),
    }
    approval_hash = sha256_bytes(
        APPROVAL_DOMAIN.encode("ascii") + b"\0" + canonical_json(capsule_core)
    )
    return {**capsule_core, "approval_hash": approval_hash}


def create_service() -> QiskitRuntimeService:
    return QiskitRuntimeService(
        channel="ibm_quantum_platform",
        token=private_secret(TOKEN_FILE),
        instance=private_secret(INSTANCE_FILE),
    )


def local_check() -> dict[str, Any]:
    bundle = structural_bundle()
    manifest = build_manifest(bundle, None)
    print("LOCAL STRUCTURAL CHECK PASSED")
    print("No IBM service call; no hardware job; QPU usage: 0 seconds")
    print(f"PUBs: {len(bundle['pubs'])}")
    print(f"Effective configurations: {EXPECTED_CONFIGURATIONS}")
    print(f"Schedule SHA-256: {bundle['schedule_sha256']}")
    return manifest


def ensure_preflight_is_unlocked() -> None:
    if INTENT_FILE.exists() or RECEIPT_FILE.exists():
        raise PermissionError(
            "A submission intent/receipt already exists; preflight cannot replace "
            "the frozen approval artifacts"
        )


def metadata_preflight() -> dict[str, Any]:
    ensure_preflight_is_unlocked()
    bundle = structural_bundle()
    plan = bundle["plan"]
    selected = plan.get("selected_snapshot")
    if not isinstance(selected, dict):
        raise ValueError("Plan has no selected backend snapshot")
    backend_name = str(selected.get("backend"))
    layout = selected.get("layout_logical_q0_q1_flag")
    if not isinstance(layout, list):
        raise ValueError("Plan has no selected physical layout")

    service = create_service()
    snapshot = current_backend_snapshot(
        service,
        backend_name,
        list(map(int, layout)),
        bundle["circuits"],
    )
    if not all(snapshot["quality_gates"].values()):
        failed = [
            name
            for name, passed in snapshot["quality_gates"].items()
            if not passed
        ]
        raise RuntimeError(f"NO-GO backend quality gates: {failed}")
    usage = estimate_usage(snapshot)
    if usage["with_50pct_margin_seconds"] > MAX_EXECUTION_SECONDS:
        raise RuntimeError(
            "NO-GO: conservative QPU estimate exceeds 480 seconds"
        )

    manifest = build_manifest(bundle, snapshot)
    capsule = build_capsule(manifest)
    write_private_json(MANIFEST_FILE, manifest)
    write_private_json(CAPSULE_FILE, capsule)
    print("METADATA-ONLY PREFLIGHT PASSED")
    print("No hardware job was submitted; QPU usage: 0 seconds")
    print(f"Backend: {snapshot['backend']}")
    print(f"Layout [P0, P1, L]: {snapshot['layout_logical_q0_q1_flag']}")
    print(f"Pending jobs: {snapshot['pending_jobs']}")
    print(
        "Conservative estimate: "
        f"{usage['with_50pct_margin_seconds']:.1f} seconds"
    )
    print(f"Approval expires: {capsule['expires_at_utc']}")
    print(f"Approval hash: {capsule['approval_hash']}")
    print(f"Required phrase: {CONFIRMATION_PHRASE}")
    return capsule


def parse_iso(value: str) -> datetime:
    parsed = datetime.fromisoformat(value)
    if parsed.tzinfo is None:
        raise ValueError(f"Timestamp must include a timezone: {value}")
    return parsed.astimezone(UTC)


def verify_capsule(
    bundle: dict[str, Any],
    approval_hash: str,
    confirmation: str,
) -> tuple[dict[str, Any], dict[str, Any]]:
    if confirmation != CONFIRMATION_PHRASE:
        raise PermissionError("Exact one-job confirmation phrase is required")
    if not CAPSULE_FILE.is_file() or not MANIFEST_FILE.is_file():
        raise PermissionError("Run a fresh metadata preflight first")
    capsule = load_json(CAPSULE_FILE)
    manifest = load_json(MANIFEST_FILE)
    if capsule.get("schema") != CAPSULE_SCHEMA:
        raise PermissionError("Unexpected approval capsule schema")
    if manifest.get("schema") != MANIFEST_SCHEMA:
        raise PermissionError("Unexpected submission manifest schema")
    if not hmac.compare_digest(
        str(capsule.get("approval_hash", "")),
        approval_hash,
    ):
        raise PermissionError("Approval hash does not match the frozen capsule")
    if parse_iso(str(capsule["expires_at_utc"])) <= utc_now():
        raise PermissionError("Approval capsule expired; run preflight again")

    verify_capsule_manifest_consistency(capsule, manifest)
    verify_current_artifacts(bundle, manifest)
    current_manifest = build_manifest(bundle, manifest.get("backend_snapshot"))
    stable_keys = ("artifacts", "experiment", "runtime_options", "versions")
    for key in stable_keys:
        if current_manifest[key] != manifest.get(key):
            raise PermissionError(f"Frozen manifest changed in field {key}")
    expected_core = {
        key: value
        for key, value in capsule.items()
        if key != "approval_hash"
    }
    expected_hash = sha256_bytes(
        APPROVAL_DOMAIN.encode("ascii") + b"\0" + canonical_json(expected_core)
    )
    if not hmac.compare_digest(expected_hash, approval_hash):
        raise PermissionError("Approval capsule failed canonical verification")
    return manifest, capsule


def sampler_options(manifest: dict[str, Any]) -> SamplerOptions:
    approval_prefix = str(manifest["manifest_sha256"])[:16]
    return SamplerOptions(
        max_execution_time=MAX_EXECUTION_SECONDS,
        default_shots=EXPECTED_SHOTS,
        dynamical_decoupling={"enable": False},
        execution={
            "init_qubits": True,
            "meas_type": "classified",
        },
        twirling={
            "enable_gates": False,
            "enable_measure": False,
        },
        environment={
            "job_tags": [
                "hqc-sorkin-null",
                f"approval-{approval_prefix}",
                "no-auto-retry",
            ],
            "private": True,
        },
    )


def checkpoint_returned_job(intent: dict[str, Any], job_id: str) -> dict[str, Any]:
    """Persist the IBM job ID before any receipt I/O can fail."""

    normalized = str(job_id).strip()
    if not normalized:
        raise ValueError("IBM Runtime returned an empty job ID")
    checkpoint = {
        **intent,
        "state": "job_returned_receipt_pending",
        "job_id": normalized,
        "job_id_checkpointed_at_utc": iso_utc(),
    }
    write_private_json(INTENT_FILE, checkpoint)
    return checkpoint


def submit_one_job(approval_hash: str, confirmation: str) -> None:
    """The only function in this file that can create an IBM workload."""

    if INTENT_FILE.exists() or RECEIPT_FILE.exists():
        raise PermissionError(
            "Submission lock/receipt already exists; automatic retry is forbidden"
        )
    bundle = structural_bundle()
    manifest, capsule = verify_capsule(bundle, approval_hash, confirmation)

    service = create_service()
    frozen_snapshot = manifest["backend_snapshot"]
    backend = service.backend(str(frozen_snapshot["backend"]))
    current = current_backend_snapshot(
        service,
        backend.name,
        list(map(int, frozen_snapshot["layout_logical_q0_q1_flag"])),
        bundle["circuits"],
    )
    verify_approved_snapshot_is_current(
        current,
        frozen_snapshot,
        capsule,
    )
    if not all(current["quality_gates"].values()):
        raise PermissionError("Backend no longer passes every quality gate")
    if estimate_usage(current)["with_50pct_margin_seconds"] > MAX_EXECUTION_SECONDS:
        raise PermissionError("Fresh QPU estimate exceeds the hard cap")

    intent = {
        "schema": "hqc-ibm-submission-intent-v1",
        "created_at_utc": iso_utc(),
        "state": "submission_intent_locked",
        "approval_hash": approval_hash,
        "manifest_sha256": manifest["manifest_sha256"],
        "backend": backend.name,
        "automatic_retry": False,
    }
    write_private_json(INTENT_FILE, intent, exclusive=True)

    sampler = SamplerV2(
        mode=backend,
        options=sampler_options(manifest),
    )
    try:
        job = sampler.run(bundle["pubs"])
    except BaseException as error:
        intent["state"] = "submission_outcome_uncertain_no_retry"
        intent["error_type"] = type(error).__name__
        intent["error_message"] = str(error)
        write_private_json(INTENT_FILE, intent)
        raise

    job_id: str | None = None
    try:
        job_id = str(job.job_id()).strip()
        intent = checkpoint_returned_job(intent, job_id)
    except BaseException as error:
        intent["state"] = "job_returned_checkpoint_failed_no_retry"
        if job_id:
            intent["job_id"] = job_id
        intent["error_type"] = type(error).__name__
        intent["error_message"] = str(error)
        try:
            write_private_json(INTENT_FILE, intent)
        except BaseException:
            pass
        raise

    receipt = {
        "schema": RECEIPT_SCHEMA,
        "submitted_at_utc": iso_utc(),
        "state": "submitted",
        "job_id": job_id,
        "backend": backend.name,
        "approval_hash": approval_hash,
        "manifest_sha256": manifest["manifest_sha256"],
        "capsule_sha256": sha256_file(CAPSULE_FILE),
        "automatic_retry": False,
        "collection_command": (
            f"{Path(__file__).resolve()} collect --job-id {job_id}"
        ),
    }
    write_private_json(RECEIPT_FILE, receipt, exclusive=True)
    intent["state"] = "submitted_receipt_written"
    intent["job_id"] = job_id
    write_private_json(INTENT_FILE, intent)
    print("ONE IBM RUNTIME JOB SUBMITTED")
    print(f"Job ID: {job_id}")
    print("No retry will be attempted automatically.")


def only_bit_array(data: Any) -> tuple[str, BitArray]:
    candidates: list[tuple[str, BitArray]] = []
    for name in data:
        value = getattr(data, name)
        if isinstance(value, BitArray):
            candidates.append((str(name), value))
    if len(candidates) != 1:
        raise ValueError(
            f"Expected exactly one BitArray result field, got {[n for n, _ in candidates]}"
        )
    name, bit_array = candidates[0]
    if bit_array.num_bits != 3 or bit_array.num_shots != EXPECTED_SHOTS:
        raise ValueError(
            f"Unexpected BitArray dimensions: bits={bit_array.num_bits}, "
            f"shots={bit_array.num_shots}"
        )
    return name, bit_array


def counts_for(bit_array: BitArray, location: int | None = None) -> dict[str, int]:
    counts = bit_array.get_counts(loc=location)
    normalized: dict[str, int] = {}
    for bitstring, count in counts.items():
        bits = str(bitstring).replace(" ", "")
        if len(bits) != 3 or any(bit not in "01" for bit in bits):
            raise ValueError(f"Invalid classified count key: {bitstring!r}")
        if not isinstance(count, int) or count < 0:
            raise ValueError(f"Invalid classified count: {count!r}")
        normalized[bits] = normalized.get(bits, 0) + count
    if sum(normalized.values()) != EXPECTED_SHOTS:
        raise ValueError("Counts do not sum to the frozen shots value")
    return dict(sorted(normalized.items()))


def execution_window_for_pub(
    execution_spans: Any,
    pub_index: int,
) -> tuple[str | None, str | None, str]:
    """Return the full IBM execution-span envelope that references a PUB.

    ``ExecutionSpans`` is global PrimitiveResult metadata, not PubResult
    metadata.  Its span objects expose the exact PUB indices that contributed
    data in each hardware time window.  We never infer bracketing from only
    the first timestamp, and we never invent a window when metadata is absent.
    """

    if execution_spans is None:
        return None, None, "unavailable"
    matching = [
        span
        for span in execution_spans
        if pub_index in getattr(span, "pub_idxs", ())
    ]
    if not matching:
        return None, None, "unavailable"
    start = min(span.start for span in matching)
    stop = max(span.stop for span in matching)
    if stop <= start:
        raise ValueError(f"Invalid IBM execution-span envelope for PUB {pub_index}")
    return iso_utc(start), iso_utc(stop), "runtime_execution_spans"


def execution_time_for_pub(
    execution_spans: Any,
    pub_index: int,
) -> tuple[str | None, str]:
    """Backward-compatible start-time accessor for local tooling."""

    start, _stop, source = execution_window_for_pub(
        execution_spans,
        pub_index,
    )
    return start, source


def json_safe(value: Any) -> Any:
    return json.loads(json.dumps(value, cls=RuntimeEncoder))


def schedule_order_status(pub_metadata: list[dict[str, Any]]) -> bool | str:
    """Verify order and readout bracketing from complete PUB time windows.

    Runtime spans can overlap because of classical processing.  Such overlap
    does not prove physical reordering, but it also cannot prove the intended
    order, so the result is ``"unknown"`` rather than ``True``.
    """

    if len(pub_metadata) != 88:
        return False
    for pub_index, row in enumerate(pub_metadata):
        if row.get("pub_index") != pub_index:
            return False
        if pub_index < 32:
            expected_kind = "readout_calibration"
            expected_position = "before"
        elif pub_index < 56:
            expected_kind = "experiment"
            expected_position = None
        else:
            expected_kind = "readout_calibration"
            expected_position = "after"
        if row.get("kind") != expected_kind:
            return False
        if row.get("calibration_position") != expected_position:
            return False

    parsed: list[tuple[datetime, datetime] | None] = []
    missing = False
    for row in pub_metadata:
        start_value = row.get("execution_start_utc")
        stop_value = row.get("execution_stop_utc")
        if start_value is None or stop_value is None:
            missing = True
            parsed.append(None)
            continue
        try:
            start = parse_iso(str(start_value))
            stop = parse_iso(str(stop_value))
        except (TypeError, ValueError):
            return False
        if stop <= start:
            return False
        parsed.append((start, stop))

    known = [
        (index, interval)
        for index, interval in enumerate(parsed)
        if interval is not None
    ]
    for (_earlier_index, earlier), (_later_index, later) in zip(
        known,
        known[1:],
    ):
        if later[0] < earlier[0]:
            return False
    if missing:
        return "unknown"

    intervals = [interval for interval in parsed if interval is not None]
    if any(
        earlier[1] > later[0]
        for earlier, later in zip(intervals, intervals[1:])
    ):
        return "unknown"
    return True


def safe_job_id(job_id: str) -> str:
    normalized = str(job_id).strip()
    if not normalized or any(
        not (character.isalnum() or character in "-_")
        for character in normalized
    ):
        raise ValueError("Unexpected IBM Runtime job ID format")
    return normalized


def fsync_directory(path: Path) -> None:
    descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def completed_export_path(job_id: str, manifest_sha256: str) -> Path | None:
    """Return a verified completed export, or None when no export exists."""

    normalized = safe_job_id(job_id)
    export_dir = ROOT / f"HQC_IBM_EXPORT_{normalized}"
    if not export_dir.exists():
        return None
    if not export_dir.is_dir():
        raise PermissionError("Export target exists but is not a directory")
    raw_path = export_dir / f"HQC_IBM_RAW_COUNTS_{normalized}.json"
    full_result_path = (
        export_dir / f"HQC_IBM_PRIMITIVE_RESULT_{normalized}.json"
    )
    receipt_path = export_dir / f"HQC_IBM_EXPORT_RECEIPT_{normalized}.json"
    if not all(path.is_file() for path in (raw_path, full_result_path, receipt_path)):
        raise PermissionError("Existing export bundle is incomplete")
    export_receipt = load_json(receipt_path)
    if export_receipt.get("schema") != EXPORT_SCHEMA:
        raise PermissionError("Unexpected export receipt schema")
    if export_receipt.get("job_id") != normalized:
        raise PermissionError("Export receipt job ID mismatch")
    if export_receipt.get("manifest_sha256") != manifest_sha256:
        raise PermissionError("Export receipt manifest mismatch")
    if export_receipt.get("raw_counts_sha256") != sha256_file(raw_path):
        raise PermissionError("Exported raw counts failed digest verification")
    if export_receipt.get("primitive_result_sha256") != sha256_file(
        full_result_path
    ):
        raise PermissionError("Exported PrimitiveResult failed digest verification")
    return raw_path


def write_result_export_bundle(
    job_id: str,
    raw_payload: dict[str, Any],
    result: Any,
) -> Path:
    """Publish raw data, full Runtime result, and receipt as one directory."""

    normalized = safe_job_id(job_id)
    manifest_sha256 = str(raw_payload["manifest_sha256"])
    existing = completed_export_path(normalized, manifest_sha256)
    if existing is not None:
        return existing

    final_dir = ROOT / f"HQC_IBM_EXPORT_{normalized}"
    staging_dir = Path(
        tempfile.mkdtemp(
            prefix=f".HQC_IBM_EXPORT_{normalized}.tmp-",
            dir=ROOT,
        )
    )
    raw_name = f"HQC_IBM_RAW_COUNTS_{normalized}.json"
    full_name = f"HQC_IBM_PRIMITIVE_RESULT_{normalized}.json"
    receipt_name = f"HQC_IBM_EXPORT_RECEIPT_{normalized}.json"
    try:
        raw_staging = staging_dir / raw_name
        full_staging = staging_dir / full_name
        write_private_json(raw_staging, raw_payload, exclusive=True)
        write_private_runtime_json(full_staging, result)
        export_receipt = {
            "schema": EXPORT_SCHEMA,
            "created_at_utc": iso_utc(),
            "job_id": normalized,
            "manifest_sha256": manifest_sha256,
            "raw_counts": raw_name,
            "raw_counts_sha256": sha256_file(raw_staging),
            "primitive_result": full_name,
            "primitive_result_sha256": sha256_file(full_staging),
            "records": len(raw_payload["records"]),
        }
        write_private_json(
            staging_dir / receipt_name,
            export_receipt,
            exclusive=True,
        )
        fsync_directory(staging_dir)
        try:
            os.rename(staging_dir, final_dir)
        except OSError:
            if final_dir.exists():
                verified = completed_export_path(normalized, manifest_sha256)
                if verified is not None:
                    return verified
            raise
        fsync_directory(ROOT)
        return final_dir / raw_name
    finally:
        if staging_dir.exists():
            shutil.rmtree(staging_dir)


def collect_existing_job(job_id: str) -> Path:
    job_id = safe_job_id(job_id)
    if not RECEIPT_FILE.is_file():
        raise PermissionError("No local submission receipt exists")
    receipt = load_json(RECEIPT_FILE)
    if receipt.get("schema") != RECEIPT_SCHEMA:
        raise ValueError("Unexpected receipt schema")
    if not hmac.compare_digest(str(receipt.get("job_id")), job_id):
        raise PermissionError("Requested job ID does not match the local receipt")

    bundle = structural_bundle()
    manifest = load_json(MANIFEST_FILE)
    manifest_digest = verify_manifest_digest(manifest)
    if manifest_digest != receipt.get("manifest_sha256"):
        raise PermissionError("Receipt/manifest mismatch")
    verify_current_artifacts(bundle, manifest)
    backend_snapshot = manifest.get("backend_snapshot")
    if not isinstance(backend_snapshot, dict):
        raise PermissionError("Submission manifest has no backend snapshot")
    if receipt.get("backend") != backend_snapshot.get("backend"):
        raise PermissionError("Receipt/manifest backend mismatch")
    completed = completed_export_path(job_id, manifest_digest)
    if completed is not None:
        print("EXISTING VERIFIED EXPORT FOUND")
        print("No IBM service call and no new QPU workload.")
        print(f"Raw counts: {completed}")
        return completed

    service = create_service()
    job = service.job(job_id)
    result = job.result()
    specs: list[PubSpec] = bundle["specs"]
    if len(result) != len(specs):
        raise ValueError(f"Expected {len(specs)} PUB results, got {len(result)}")
    primitive_metadata = getattr(result, "metadata", {})
    execution_spans = None
    if isinstance(primitive_metadata, dict):
        execution = primitive_metadata.get("execution")
        if isinstance(execution, dict):
            execution_spans = execution.get("execution_spans")

    records: list[dict[str, Any]] = []
    execution_index = 0
    pub_metadata: list[dict[str, Any]] = []
    for spec, pub_result in zip(specs, result, strict=True):
        field_name, bit_array = only_bit_array(pub_result.data)
        execution_start, execution_stop, time_source = execution_window_for_pub(
            execution_spans,
            spec.pub_index,
        )
        calibration_position = (
            (spec.calibration or {}).get("position")
            if spec.kind == "readout_calibration"
            else None
        )
        pub_metadata.append(
            {
                "pub_index": spec.pub_index,
                "kind": spec.kind,
                "calibration_position": calibration_position,
                "block_id": spec.block_id,
                "result_field": field_name,
                "metadata": json_safe(pub_result.metadata),
                "executed_at_utc": execution_start,
                "execution_start_utc": execution_start,
                "execution_stop_utc": execution_stop,
                "execution_time_source": time_source,
            }
        )

        if spec.kind == "readout_calibration":
            calibration = dict(spec.calibration or {})
            records.append(
                {
                    "kind": "readout_calibration",
                    **calibration,
                    "shots": EXPECTED_SHOTS,
                    "counts": counts_for(bit_array),
                    "execution_index": execution_index,
                    "executed_at_utc": execution_start,
                    "execution_start_utc": execution_start,
                    "execution_stop_utc": execution_stop,
                    "execution_time_source": time_source,
                    "pub_index": spec.pub_index,
                }
            )
            execution_index += 1
            continue

        if bit_array.shape != (16,):
            raise ValueError(
                f"Experiment PUB {spec.pub_index} has shape {bit_array.shape}, not (16,)"
            )
        for location, row in enumerate(spec.binding_rows):
            records.append(
                {
                    "kind": "experiment",
                    "block_id": row["block_id"],
                    "experiment_type": row["experiment_type"],
                    "stratum": row["stratum"],
                    "order_in_block": row["order_in_block"],
                    "logical_mask": row["logical_mask"],
                    "logical_to_physical_mapping": row[
                        "logical_to_physical_mapping"
                    ],
                    "shots": EXPECTED_SHOTS,
                    "counts": counts_for(bit_array, location),
                    "execution_index": execution_index,
                    "executed_at_utc": execution_start,
                    "execution_start_utc": execution_start,
                    "execution_stop_utc": execution_stop,
                    "execution_time_source": time_source,
                    "pub_index": spec.pub_index,
                    "binding_index": row["binding_index"],
                }
            )
            execution_index += 1

    if len(records) != EXPECTED_CONFIGURATIONS:
        raise AssertionError(f"Expected 448 exported records, got {len(records)}")

    metrics: Any
    try:
        metrics = json_safe(job.metrics())
    except BaseException as error:
        metrics = {"unavailable": type(error).__name__}
    usage: Any
    try:
        usage = json_safe(job.usage())
    except BaseException as error:
        usage = {"unavailable": type(error).__name__}

    schedule_order_preserved = schedule_order_status(pub_metadata)

    raw_payload = {
        "schema": RAW_SCHEMA,
        "source": "ibm-runtime-sampler-v2",
        "exported_at_utc": iso_utc(),
        "job_id": job_id,
        "backend": receipt["backend"],
        "layout": manifest["backend_snapshot"]["layout_logical_q0_q1_flag"],
        "shots_per_circuit": EXPECTED_SHOTS,
        "plan_sha256": manifest["artifacts"]["plan_sha256"],
        "qpy_sha256": manifest["artifacts"]["qpy_sha256"],
        "manifest_sha256": manifest["manifest_sha256"],
        "schedule_order_preserved": schedule_order_preserved,
        "execution_bracketing_verified": schedule_order_preserved,
        "execution_time_source": (
            "runtime_execution_spans"
            if any(row["execution_start_utc"] for row in pub_metadata)
            else "unavailable"
        ),
        "runtime_versions": manifest["versions"],
        "job_metrics": metrics,
        "job_usage": usage,
        "primitive_metadata": json_safe(primitive_metadata),
        "pub_metadata": pub_metadata,
        "records": records,
    }
    raw_path = write_result_export_bundle(job_id, raw_payload, result)
    print("EXISTING JOB RESULT COLLECTED")
    print("No new job was submitted; collection consumes no new QPU workload.")
    print(f"Raw counts: {raw_path}")
    return raw_path


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Guarded HQC/IBM Sorkin null-benchmark runner"
    )
    parser.add_argument(
        "--workdir",
        type=Path,
        default=ROOT,
        help=(
            "directory containing the newly prepared plan/QPY and all "
            "new-run state; frozen evidence is never the default"
        ),
    )
    subparsers = parser.add_subparsers(dest="command")
    subparsers.add_parser("check", help="fully local structural validation")
    subparsers.add_parser("preflight", help="metadata-only IBM preflight")

    submit = subparsers.add_parser(
        "submit",
        help="submit exactly one approved job; never used by default",
    )
    submit.add_argument("--approval-hash", required=True)
    submit.add_argument("--confirmation", required=True)

    collect = subparsers.add_parser(
        "collect",
        help="collect an existing receipted job; cannot submit",
    )
    collect.add_argument("--job-id", required=True)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    configure_workdir(args.workdir)
    command = args.command or "check"
    if command == "check":
        local_check()
    elif command == "preflight":
        metadata_preflight()
    elif command == "submit":
        submit_one_job(args.approval_hash, args.confirmation)
    elif command == "collect":
        collect_existing_job(args.job_id)
    else:
        raise SystemExit(f"Unknown command: {command}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Local, pre-registered analysis for the HQC/IBM Sorkin null benchmark.

The module deliberately has no IBM or Qiskit imports and cannot submit work.
It validates an exported 448-record counts payload against the frozen plan
and manifest, preserves raw and independently readout-corrected results, and
implements the hierarchical bootstrap and conservative decision rules in
sections 8-9 of ``HQC_IBM_EXPERIMENT_DESIGN.md``.
"""

from __future__ import annotations

import argparse
import hashlib
import hmac
import json
import math
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Iterable

import numpy as np
from scipy.optimize import minimize
from scipy.stats import chi2, f_oneway, linregress, norm


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
ROOT = PACKAGE_ROOT / "work"
PLAN_FILE = (
    PACKAGE_ROOT
    / "evidence"
    / "frozen-pre-run"
    / "HQC_IBM_PARAMETRIC_DRY_RUN_PLAN.json"
)
MANIFEST_FILE = (
    PACKAGE_ROOT / "evidence" / "submission" / "HQC_SUBMISSION_MANIFEST.json"
)
PRE_RUN_MANIFEST_FILE = (
    PACKAGE_ROOT
    / "evidence"
    / "frozen-pre-run"
    / "HQC_PRE_RUN_MANIFEST.sha256"
)
QPY_FILE = (
    PACKAGE_ROOT
    / "evidence"
    / "frozen-pre-run"
    / "HQC_IBM_PARAMETRIC_TEMPLATE.qpy"
)
FROZEN_ANALYZER_FILE = (
    PACKAGE_ROOT / "archive" / "original-frozen" / "hqc_analyze_counts.py"
)
SYNTHETIC_COUNTS = ROOT / "HQC_SYNTHETIC_COUNTS.json"
SYNTHETIC_ANALYSIS = ROOT / "HQC_SYNTHETIC_ANALYSIS.json"

RAW_SCHEMA = "hqc-ibm-raw-counts-v2"
ANALYSIS_SCHEMA = "hqc-ibm-analysis-v2"
DEFAULT_BOOTSTRAP_REPS = 20_000
DEFAULT_SEED = 20_260_723
CONTRAST_NAMES = (
    "I3_012",
    "I3_013",
    "I3_023",
    "I3_123",
    "I4_0123",
)
VARIANTS = ("raw", "mitigated_before", "mitigated_after")
EXPERIMENT_TYPES = ("primary", "sign", "phase")
FIVE_SIGMA_TWO_SIDED_ALPHA = float(2.0 * norm.sf(5.0))
SUBMISSION_MANIFEST_SCHEMA = "hqc-ibm-submission-manifest-v1"
EXPORT_RECEIPT_SCHEMA = "hqc-ibm-export-receipt-v1"


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def canonical_json_bytes(value: Any) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def replace_nonfinite_with_null(value: Any) -> tuple[Any, int]:
    """Return strict-JSON-safe data without changing finite statistics.

    The frozen analyzer produced 15 NaN values when IBM exposed one aggregate
    execution timestamp for every PUB. The published result replaced only
    those non-finite regression fields with JSON null. This portable layer
    makes that post-run serialization step explicit and reproducible.
    """

    if isinstance(value, float) and not math.isfinite(value):
        return None, 1
    if isinstance(value, list):
        replaced = 0
        items = []
        for item in value:
            normalized, count = replace_nonfinite_with_null(item)
            items.append(normalized)
            replaced += count
        return items, replaced
    if isinstance(value, dict):
        replaced = 0
        items = {}
        for key, item in value.items():
            normalized, count = replace_nonfinite_with_null(item)
            items[key] = normalized
            replaced += count
        return items, replaced
    return value, 0


def _is_sha256(value: Any) -> bool:
    return (
        isinstance(value, str)
        and len(value) == 64
        and all(character in "0123456789abcdef" for character in value)
    )


def _parse_pre_run_manifest(path: Path) -> dict[str, str]:
    entries: dict[str, str] = {}
    for line_number, raw_line in enumerate(
        path.read_text(encoding="utf-8").splitlines(), start=1
    ):
        line = raw_line.strip()
        if not line:
            continue
        parts = line.split(maxsplit=1)
        if len(parts) != 2 or not _is_sha256(parts[0]):
            raise ValueError(f"Malformed pre-run manifest line {line_number}")
        filename = parts[1].lstrip("*")
        if filename in entries:
            raise ValueError(f"Duplicate pre-run manifest entry: {filename}")
        entries[filename] = parts[0]
    return entries


def _safe_job_id(value: Any) -> str:
    if not isinstance(value, str):
        raise ValueError("Unexpected IBM Runtime job ID format")
    normalized = value.strip()
    if not normalized or any(
        not (character.isalnum() or character in "-_")
        for character in normalized
    ):
        raise ValueError("Unexpected IBM Runtime job ID format")
    return normalized


def load_frozen_contract(
    plan_path: Path = PLAN_FILE,
    manifest_path: Path = MANIFEST_FILE,
    qpy_path: Path = QPY_FILE,
    *,
    allow_pre_run_fallback: bool = False,
) -> dict[str, Any]:
    plan = json.loads(plan_path.read_text(encoding="utf-8"))
    if not manifest_path.is_file():
        if allow_pre_run_fallback and PRE_RUN_MANIFEST_FILE.is_file():
            entries = _parse_pre_run_manifest(PRE_RUN_MANIFEST_FILE)
            plan_digest = sha256_file(plan_path)
            qpy_digest = sha256_file(qpy_path)
            if entries.get(plan_path.name) != plan_digest:
                raise ValueError("Frozen plan hash does not match the pre-run manifest")
            if entries.get(qpy_path.name) != qpy_digest:
                raise ValueError("Frozen QPY hash does not match the pre-run manifest")
            if plan.get("artifacts", {}).get("transpiled_qpy_sha256") != qpy_digest:
                raise ValueError("Frozen plan and QPY hash disagree")
            experiment = plan.get("experiment", {})
            if (
                experiment.get("effective_total_circuits") != 448
                or len(experiment.get("parameter_bindings", [])) != 384
                or experiment.get("readout_calibration_circuits") != 64
            ):
                raise ValueError(
                    "Frozen plan does not describe the registered 448-record run"
                )
            fallback_digest = sha256_file(PRE_RUN_MANIFEST_FILE)
            fallback_manifest = {
                "schema": "hqc-pre-run-local-self-test-v1",
                "manifest_sha256": fallback_digest,
                "artifacts": {
                    "plan_sha256": plan_digest,
                    "qpy_sha256": qpy_digest,
                    "analyzer_sha256": sha256_file(FROZEN_ANALYZER_FILE),
                },
                "experiment": {
                    "shots_per_configuration": 1024,
                    "parameter_bindings": 384,
                    "readout_calibrations": 64,
                    "effective_configurations": 448,
                },
                "versions": {"analyzer": ANALYSIS_SCHEMA, "mode": "local-self-test"},
                "backend_snapshot": plan["selected_snapshot"],
            }
            return {
                "plan": plan,
                "plan_sha256": plan_digest,
                "qpy_sha256": qpy_digest,
                "manifest_sha256": fallback_digest,
                "manifest_file_sha256": fallback_digest,
                "manifest": fallback_manifest,
                "contract_mode": "pre-run-local-self-test",
            }
        raise ValueError(
            f"Submission manifest is missing: {manifest_path}. "
            "Run the metadata-only preflight before analyzing IBM results."
        )
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    if not isinstance(manifest, dict):
        raise ValueError("Submission manifest must be a JSON object")
    if manifest.get("schema") != SUBMISSION_MANIFEST_SCHEMA:
        raise ValueError("Unsupported submission manifest schema")
    internal_digest = manifest.get("manifest_sha256")
    if not _is_sha256(internal_digest):
        raise ValueError("Submission manifest has an invalid internal digest")
    manifest_core = {
        key: value for key, value in manifest.items() if key != "manifest_sha256"
    }
    computed_internal_digest = hashlib.sha256(
        canonical_json_bytes(manifest_core)
    ).hexdigest()
    if not hmac.compare_digest(internal_digest, computed_internal_digest):
        raise ValueError("Submission manifest internal digest does not match its content")

    plan_digest = sha256_file(plan_path)
    qpy_digest = sha256_file(qpy_path)
    artifacts = manifest.get("artifacts")
    if not isinstance(artifacts, dict):
        raise ValueError("Submission manifest artifacts must be an object")
    if artifacts.get("plan_sha256") != plan_digest:
        raise ValueError("Frozen plan hash does not match the submission manifest")
    if artifacts.get("qpy_sha256") != qpy_digest:
        raise ValueError("Frozen QPY hash does not match the submission manifest")
    analyzer_digests = {
        sha256_file(FROZEN_ANALYZER_FILE),
        sha256_file(Path(__file__).resolve()),
    }
    if artifacts.get("analyzer_sha256") not in analyzer_digests:
        raise ValueError("Analyzer hash does not match the submission manifest")
    if plan.get("artifacts", {}).get("transpiled_qpy_sha256") != qpy_digest:
        raise ValueError("Frozen plan and QPY hash disagree")
    experiment = plan.get("experiment", {})
    if (
        experiment.get("effective_total_circuits") != 448
        or len(experiment.get("parameter_bindings", [])) != 384
        or experiment.get("readout_calibration_circuits") != 64
    ):
        raise ValueError("Frozen plan does not describe the registered 448-record run")
    manifest_experiment = manifest.get("experiment")
    if not isinstance(manifest_experiment, dict):
        raise ValueError("Submission manifest experiment must be an object")
    expected_manifest_shape = {
        "shots_per_configuration": 1024,
        "parameter_bindings": 384,
        "readout_calibrations": 64,
        "effective_configurations": 448,
    }
    for key, expected in expected_manifest_shape.items():
        if manifest_experiment.get(key) != expected:
            raise ValueError(f"Submission manifest experiment.{key} is not {expected}")
    backend_snapshot = manifest.get("backend_snapshot")
    if not isinstance(backend_snapshot, dict):
        raise ValueError("Submission manifest lacks a backend snapshot")
    if not isinstance(manifest.get("versions"), dict):
        raise ValueError("Submission manifest versions must be an object")
    return {
        "plan": plan,
        "plan_sha256": plan_digest,
        "qpy_sha256": qpy_digest,
        "manifest_sha256": internal_digest,
        "manifest_file_sha256": sha256_file(manifest_path),
        "manifest": manifest,
        "contract_mode": "submission-manifest",
    }


def verify_export_provenance(
    input_path: Path,
    payload: dict[str, Any],
    contract: dict[str, Any],
) -> dict[str, Any]:
    """Verify the immutable raw-counts/receipt link for a real analysis."""

    if not input_path.is_file():
        raise ValueError(f"Raw input file is missing: {input_path}")
    job_id = _safe_job_id(payload.get("job_id"))
    receipt_path = input_path.with_name(
        f"HQC_IBM_EXPORT_RECEIPT_{job_id}.json"
    )
    if not receipt_path.is_file():
        raise ValueError(
            f"Matching export receipt is missing: {receipt_path.name}"
        )
    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
    if not isinstance(receipt, dict):
        raise ValueError("Export receipt must be a JSON object")
    if receipt.get("schema") != EXPORT_RECEIPT_SCHEMA:
        raise ValueError("Unsupported export receipt schema")
    if receipt.get("job_id") != job_id:
        raise ValueError("Export receipt job ID does not match raw input")
    if receipt.get("manifest_sha256") != contract["manifest_sha256"]:
        raise ValueError("Export receipt manifest does not match frozen contract")
    if payload.get("manifest_sha256") != receipt.get("manifest_sha256"):
        raise ValueError("Raw input and export receipt manifest hashes disagree")
    if receipt.get("raw_counts") != input_path.name:
        raise ValueError("Export receipt points to a different raw-counts file")
    expected_raw_digest = receipt.get("raw_counts_sha256")
    if not _is_sha256(expected_raw_digest):
        raise ValueError("Export receipt raw-counts digest is invalid")
    actual_raw_digest = sha256_file(input_path)
    if not hmac.compare_digest(expected_raw_digest, actual_raw_digest):
        raise ValueError("Raw counts failed export-receipt digest verification")
    if receipt.get("records") != 448:
        raise ValueError("Export receipt does not attest exactly 448 records")
    return {
        "mode": "verified-export-receipt",
        "raw_input_file": input_path.name,
        "raw_input_sha256": actual_raw_digest,
        "export_receipt_file": receipt_path.name,
        "export_receipt_sha256": sha256_file(receipt_path),
        "receipt_schema": receipt["schema"],
        "job_id": job_id,
        "manifest_sha256": contract["manifest_sha256"],
    }


def _require_int(value: Any, label: str, *, minimum: int | None = None) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise ValueError(f"{label} must be an integer")
    if minimum is not None and value < minimum:
        raise ValueError(f"{label} must be at least {minimum}")
    return value


def _parse_iso_timestamp(value: Any, label: str) -> float | None:
    if value is None:
        return None
    if not isinstance(value, str) or not value:
        raise ValueError(f"{label} must be an ISO timestamp or null")
    normalized = value.replace("Z", "+00:00")
    try:
        parsed = datetime.fromisoformat(normalized)
    except ValueError as error:
        raise ValueError(f"{label} is not a valid ISO timestamp") from error
    if parsed.tzinfo is None:
        raise ValueError(f"{label} must include a timezone")
    return parsed.timestamp()


def counts_vector(counts: Any, shots: int, label: str) -> np.ndarray:
    if not isinstance(counts, dict) or not counts:
        raise ValueError(f"{label}.counts must be a non-empty object")
    vector = np.zeros(8, dtype=np.int64)
    normalized_keys: set[str] = set()
    for bitstring, count in counts.items():
        if not isinstance(bitstring, str):
            raise ValueError(f"{label} contains a non-string count key")
        bits = bitstring.replace(" ", "")
        if len(bits) != 3 or any(bit not in "01" for bit in bits):
            raise ValueError(
                f"{label} expected a three-bit classified count key, got {bitstring!r}"
            )
        if bits in normalized_keys:
            raise ValueError(f"{label} contains duplicate normalized key {bits}")
        normalized_keys.add(bits)
        count_value = _require_int(count, f"{label}.counts[{bitstring!r}]", minimum=0)
        vector[int(bits, 2)] = count_value
    if int(vector.sum()) != shots:
        raise ValueError(
            f"{label} count total {int(vector.sum())} does not equal shots {shots}"
        )
    return vector


def _expected_record_descriptors(plan: dict[str, Any]) -> list[dict[str, Any]]:
    experiment = plan["experiment"]
    repeats = int(experiment["readout_repeats_per_position"])
    descriptors: list[dict[str, Any]] = []
    for repeat in range(repeats):
        for prepared_state in range(8):
            descriptors.append(
                {
                    "kind": "readout_calibration",
                    "position": "before",
                    "repeat": repeat,
                    "prepared_state": prepared_state,
                    "pub_index": repeat * 8 + prepared_state,
                }
            )
    for binding_index, binding in enumerate(experiment["parameter_bindings"]):
        descriptors.append(
            {
                "kind": "experiment",
                "block_id": binding["block_id"],
                "experiment_type": binding["experiment_type"],
                "stratum": binding["stratum"],
                "order_in_block": binding["order_in_block"],
                "logical_mask": binding["logical_mask"],
                "logical_to_physical_mapping": binding[
                    "logical_to_physical_mapping"
                ],
                "pub_index": 32 + binding_index // 16,
                "binding_index": binding_index,
            }
        )
    for repeat in range(repeats):
        for prepared_state in range(8):
            descriptors.append(
                {
                    "kind": "readout_calibration",
                    "position": "after",
                    "repeat": repeat,
                    "prepared_state": prepared_state,
                    "pub_index": 56 + repeat * 8 + prepared_state,
                }
            )
    if len(descriptors) != 448:
        raise ValueError("Internal frozen descriptor count is not 448")
    return descriptors


def _derived_execution_order_status(
    records: list[dict[str, Any]],
) -> bool | str:
    """Derive a conservative 88-PUB order/bracketing status from records."""

    windows_by_pub: dict[
        int, tuple[float | None, float | None, str]
    ] = {}
    for record in records:
        pub_index = int(record["pub_index"])
        window = (
            record["_execution_start"],
            record["_execution_stop"],
            record["execution_time_source"],
        )
        if pub_index in windows_by_pub and windows_by_pub[pub_index] != window:
            raise ValueError(
                f"Records for PUB {pub_index} do not share one execution window"
            )
        windows_by_pub[pub_index] = window
    if set(windows_by_pub) != set(range(88)):
        raise ValueError("Payload does not contain the complete 88-PUB schedule")
    ordered = [windows_by_pub[index] for index in range(88)]
    if any(start is None or stop is None for start, stop, _source in ordered):
        return "unknown"
    complete = [
        (float(start), float(stop))
        for start, stop, _source in ordered
        if start is not None and stop is not None
    ]
    if all(
        previous_stop <= next_start
        for (_previous_start, previous_stop), (next_start, _next_stop)
        in zip(complete, complete[1:])
    ):
        return True
    if any(
        next_start < previous_start
        for (previous_start, _previous_stop), (next_start, _next_stop)
        in zip(complete, complete[1:])
    ):
        return False
    return "unknown"


def _validate_pub_metadata(
    payload: dict[str, Any],
    records: list[dict[str, Any]],
    plan: dict[str, Any],
) -> list[dict[str, Any]]:
    metadata = payload.get("pub_metadata")
    if not isinstance(metadata, list) or len(metadata) != 88:
        raise ValueError("Payload pub_metadata must contain exactly 88 PUBs")
    records_by_pub: dict[int, list[dict[str, Any]]] = defaultdict(list)
    for record in records:
        records_by_pub[int(record["pub_index"])].append(record)
    experiment_bindings = plan["experiment"]["parameter_bindings"]
    expected_experiment_blocks = [
        str(experiment_bindings[offset]["block_id"])
        for offset in range(0, len(experiment_bindings), 16)
    ]
    normalized: list[dict[str, Any]] = []
    for pub_index, row in enumerate(metadata):
        label = f"pub_metadata[{pub_index}]"
        if not isinstance(row, dict):
            raise ValueError(f"{label} must be an object")
        if _require_int(
            row.get("pub_index"), f"{label}.pub_index", minimum=0
        ) != pub_index:
            raise ValueError(f"{label}.pub_index is out of order")
        if pub_index < 32:
            expected_kind = "readout_calibration"
            expected_position: str | None = "before"
            expected_block: str | None = None
        elif pub_index < 56:
            expected_kind = "experiment"
            expected_position = None
            expected_block = expected_experiment_blocks[pub_index - 32]
        else:
            expected_kind = "readout_calibration"
            expected_position = "after"
            expected_block = None
        if row.get("kind") != expected_kind:
            raise ValueError(f"{label}.kind does not match the frozen PUB schedule")
        if row.get("calibration_position") != expected_position:
            raise ValueError(
                f"{label}.calibration_position does not match the frozen schedule"
            )
        if row.get("block_id") != expected_block:
            raise ValueError(f"{label}.block_id does not match the frozen schedule")
        start = _parse_iso_timestamp(
            row.get("execution_start_utc"),
            f"{label}.execution_start_utc",
        )
        stop = _parse_iso_timestamp(
            row.get("execution_stop_utc"),
            f"{label}.execution_stop_utc",
        )
        executed_at = _parse_iso_timestamp(
            row.get("executed_at_utc"), f"{label}.executed_at_utc"
        )
        if (start is None) != (stop is None):
            raise ValueError(
                f"{label} execution start/stop must both be set or both be null"
            )
        if start is not None and stop is not None and stop <= start:
            raise ValueError(f"{label} execution window must satisfy start < stop")
        if executed_at != start:
            raise ValueError(
                f"{label}.executed_at_utc must equal execution_start_utc"
            )
        source = row.get("execution_time_source")
        if not isinstance(source, str) or not source:
            raise ValueError(f"{label}.execution_time_source must be non-empty")
        pub_records = records_by_pub.get(pub_index, [])
        expected_count = 16 if expected_kind == "experiment" else 1
        if len(pub_records) != expected_count:
            raise ValueError(
                f"PUB {pub_index} has {len(pub_records)} records, "
                f"expected {expected_count}"
            )
        for record in pub_records:
            if (
                record["_execution_start"] != start
                or record["_execution_stop"] != stop
                or record["execution_time_source"] != source
            ):
                raise ValueError(
                    f"PUB {pub_index} metadata and record windows disagree"
                )
        normalized_row = dict(row)
        normalized_row["_execution_start"] = start
        normalized_row["_execution_stop"] = stop
        normalized.append(normalized_row)
    return normalized


def validate_payload(
    payload: dict[str, Any],
    contract: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Validate and normalize the complete exported counts contract."""

    contract = contract or load_frozen_contract()
    plan = contract["plan"]
    experiment = plan["experiment"]
    snapshot = contract["manifest"]["backend_snapshot"]

    if not isinstance(payload, dict):
        raise ValueError("Raw payload must be a JSON object")
    if payload.get("schema") != RAW_SCHEMA:
        raise ValueError("Unsupported raw-counts schema")
    for key in ("plan_sha256", "qpy_sha256", "manifest_sha256"):
        if payload.get(key) != contract[key]:
            raise ValueError(f"Payload {key} does not match the frozen artifact")
    if payload.get("backend") != snapshot["backend"]:
        raise ValueError("Payload backend does not match the transpiled frozen plan")
    if payload.get("layout") != snapshot["layout_logical_q0_q1_flag"]:
        raise ValueError("Payload layout does not match the submission manifest")
    if payload.get("shots_per_circuit") != experiment["shots_per_circuit"]:
        raise ValueError("Payload shots_per_circuit does not match the frozen plan")
    if _safe_job_id(payload.get("job_id")) != payload.get("job_id"):
        raise ValueError("Payload job_id must be a normalized safe identifier")
    if not isinstance(payload.get("runtime_versions"), dict):
        raise ValueError("Payload runtime_versions must be an object")
    if payload["runtime_versions"] != contract["manifest"]["versions"]:
        raise ValueError("Payload runtime_versions do not match the submission manifest")
    schedule_order = payload.get("schedule_order_preserved")
    if schedule_order is None:
        schedule_order = "unknown"
    if schedule_order not in (True, False, "unknown"):
        raise ValueError(
            "schedule_order_preserved must be true, false, null, or 'unknown'"
        )
    bracketing_status = payload.get("execution_bracketing_verified")
    if bracketing_status not in (True, False, "unknown"):
        raise ValueError(
            "execution_bracketing_verified must be true, false, or 'unknown'"
        )
    if not isinstance(payload.get("execution_time_source"), str):
        raise ValueError("execution_time_source must be a string")

    records = payload.get("records")
    if not isinstance(records, list) or len(records) != 448:
        raise ValueError("Payload must contain exactly 448 records")
    descriptors = _expected_record_descriptors(plan)
    shots_expected = int(experiment["shots_per_circuit"])
    normalized_records: list[dict[str, Any]] = []
    seen_experiments: set[tuple[str, str]] = set()
    seen_calibrations: set[tuple[str, int, int]] = set()
    timestamps: list[float | None] = []

    for index, (record, expected) in enumerate(zip(records, descriptors)):
        label = f"records[{index}]"
        if not isinstance(record, dict):
            raise ValueError(f"{label} must be an object")
        execution_index = _require_int(
            record.get("execution_index"), f"{label}.execution_index", minimum=0
        )
        if execution_index != index:
            raise ValueError(f"{label} is out of the frozen execution order")
        if record.get("kind") != expected["kind"]:
            raise ValueError(f"{label}.kind does not match the frozen plan")
        pub_index = _require_int(
            record.get("pub_index"), f"{label}.pub_index", minimum=0
        )
        if pub_index != expected["pub_index"]:
            raise ValueError(f"{label}.pub_index does not match the frozen plan")
        shots = _require_int(record.get("shots"), f"{label}.shots", minimum=1)
        if shots != shots_expected:
            raise ValueError(f"{label}.shots does not equal {shots_expected}")
        vector = counts_vector(record.get("counts"), shots, label)
        timestamp = _parse_iso_timestamp(
            record.get("executed_at_utc"), f"{label}.executed_at_utc"
        )
        execution_start = _parse_iso_timestamp(
            record.get("execution_start_utc"),
            f"{label}.execution_start_utc",
        )
        execution_stop = _parse_iso_timestamp(
            record.get("execution_stop_utc"),
            f"{label}.execution_stop_utc",
        )
        if (execution_start is None) != (execution_stop is None):
            raise ValueError(
                f"{label} execution start/stop must both be set or both be null"
            )
        if (
            execution_start is not None
            and execution_stop is not None
            and execution_stop <= execution_start
        ):
            raise ValueError(f"{label} execution window must satisfy start < stop")
        if timestamp != execution_start:
            raise ValueError(
                f"{label}.executed_at_utc must equal execution_start_utc"
            )
        if (
            not isinstance(record.get("execution_time_source"), str)
            or not record["execution_time_source"]
        ):
            raise ValueError(f"{label}.execution_time_source must be non-empty")
        timestamps.append(timestamp)

        normalized = dict(record)
        normalized["_counts_vector"] = vector
        normalized["_timestamp"] = timestamp
        normalized["_execution_start"] = execution_start
        normalized["_execution_stop"] = execution_stop
        if expected["kind"] == "readout_calibration":
            if "binding_index" in record:
                raise ValueError(
                    f"{label}.binding_index is forbidden for calibration records"
                )
            for key in ("position", "repeat", "prepared_state"):
                if record.get(key) != expected[key]:
                    raise ValueError(f"{label}.{key} does not match the frozen plan")
            prepared_state = _require_int(
                record["prepared_state"], f"{label}.prepared_state", minimum=0
            )
            if prepared_state > 7:
                raise ValueError(f"{label}.prepared_state must be between 0 and 7")
            calibration_key = (
                record["position"],
                int(record["repeat"]),
                prepared_state,
            )
            if calibration_key in seen_calibrations:
                raise ValueError(f"Duplicate calibration record {calibration_key}")
            seen_calibrations.add(calibration_key)
        else:
            binding_index = _require_int(
                record.get("binding_index"),
                f"{label}.binding_index",
                minimum=0,
            )
            if binding_index != expected["binding_index"]:
                raise ValueError(
                    f"{label}.binding_index does not match the frozen plan"
                )
            for key in (
                "block_id",
                "experiment_type",
                "stratum",
                "order_in_block",
                "logical_mask",
                "logical_to_physical_mapping",
            ):
                if record.get(key) != expected[key]:
                    raise ValueError(f"{label}.{key} does not match the frozen plan")
            logical_mask = record["logical_mask"]
            if (
                not isinstance(logical_mask, str)
                or len(logical_mask) != 4
                or any(bit not in "01" for bit in logical_mask)
            ):
                raise ValueError(f"{label}.logical_mask must be four binary digits")
            experiment_key = (record["block_id"], logical_mask)
            if experiment_key in seen_experiments:
                raise ValueError(f"Duplicate experiment record {experiment_key}")
            seen_experiments.add(experiment_key)
        normalized_records.append(normalized)

    if len(seen_experiments) != 384 or len(seen_calibrations) != 64:
        raise ValueError("Payload does not contain the registered record inventory")
    non_null_times = [value for value in timestamps if value is not None]
    timestamps_are_monotonic = not any(
        later < earlier for earlier, later in zip(non_null_times, non_null_times[1:])
    )
    if schedule_order is True and non_null_times and not timestamps_are_monotonic:
        raise ValueError(
            "schedule_order_preserved=true contradicts execution timestamps"
        )
    normalized_pub_metadata = _validate_pub_metadata(
        payload, normalized_records, plan
    )
    derived_order_status = _derived_execution_order_status(normalized_records)
    if schedule_order is True and derived_order_status is not True:
        raise ValueError(
            "schedule_order_preserved=true is not verified by execution windows"
        )
    if bracketing_status is True and derived_order_status is not True:
        raise ValueError(
            "execution_bracketing_verified=true is not verified by execution windows"
        )
    normalized_payload = dict(payload)
    normalized_payload["schedule_order_preserved"] = schedule_order
    normalized_payload["execution_bracketing_verified"] = bracketing_status
    normalized_payload["_derived_execution_order_status"] = derived_order_status
    normalized_payload["_timestamps_are_monotonic"] = timestamps_are_monotonic
    normalized_payload["records"] = normalized_records
    normalized_payload["pub_metadata"] = normalized_pub_metadata
    return normalized_payload


def assignment_matrix(
    records: Iterable[dict[str, Any]], position: str
) -> np.ndarray:
    grouped = {state: np.zeros(8, dtype=float) for state in range(8)}
    for record in records:
        if (
            record["kind"] == "readout_calibration"
            and record["position"] == position
        ):
            grouped[int(record["prepared_state"])] += record["_counts_vector"]
    matrix = np.zeros((8, 8), dtype=float)
    for prepared in range(8):
        total = float(grouped[prepared].sum())
        if total <= 0:
            raise ValueError(
                f"Missing {position} readout calibration for state {prepared:03b}"
            )
        matrix[:, prepared] = grouped[prepared] / total
    return matrix


def mitigate(probabilities: np.ndarray, assignment: np.ndarray) -> np.ndarray:
    """Constrained least squares on the probability simplex.

    Near-identity assignment matrices use the exact unconstrained solution
    when it is already on the simplex. A small SLSQP fallback enforces both
    non-negativity and unit sum for the exceptional case.
    """

    try:
        candidate = np.linalg.solve(assignment, probabilities)
    except np.linalg.LinAlgError:
        candidate = np.linalg.lstsq(assignment, probabilities, rcond=None)[0]
    if (
        np.all(candidate >= -1e-12)
        and np.all(candidate <= 1.0 + 1e-12)
        and abs(float(candidate.sum()) - 1.0) <= 1e-9
    ):
        corrected = np.clip(candidate, 0.0, 1.0)
        return corrected / corrected.sum()

    objective = lambda vector: float(
        np.dot(assignment @ vector - probabilities, assignment @ vector - probabilities)
    )
    gradient = lambda vector: 2.0 * assignment.T @ (
        assignment @ vector - probabilities
    )
    initial = np.clip(candidate, 0.0, 1.0)
    if initial.sum() <= 0:
        initial = np.full(8, 1.0 / 8.0)
    else:
        initial /= initial.sum()
    fit = minimize(
        objective,
        initial,
        jac=gradient,
        bounds=[(0.0, 1.0)] * 8,
        constraints={"type": "eq", "fun": lambda vector: vector.sum() - 1.0},
        method="SLSQP",
        options={"ftol": 1e-12, "maxiter": 200},
    )
    if not fit.success:
        raise ValueError(f"Readout correction failed: {fit.message}")
    corrected = np.clip(fit.x, 0.0, 1.0)
    if corrected.sum() <= 0:
        raise ValueError("Readout correction produced a zero distribution")
    return corrected / corrected.sum()


def _contrast_terms(paths: tuple[int, ...]) -> dict[int, int]:
    terms: dict[int, int] = {}
    n = len(paths)
    for subset_bits in range(1 << n):
        mask = 0
        size = 0
        for offset, path in enumerate(paths):
            if subset_bits & (1 << offset):
                mask |= 1 << path
                size += 1
        terms[mask] = (-1) ** (n - size)
    return terms


CONTRAST_TERMS = {
    "I3_012": _contrast_terms((0, 1, 2)),
    "I3_013": _contrast_terms((0, 1, 3)),
    "I3_023": _contrast_terms((0, 2, 3)),
    "I3_123": _contrast_terms((1, 2, 3)),
    "I4_0123": _contrast_terms((0, 1, 2, 3)),
}


def contrasts(probabilities: dict[int, float]) -> dict[str, float]:
    if set(probabilities) != set(range(16)):
        missing = sorted(set(range(16)) - set(probabilities))
        extra = sorted(set(probabilities) - set(range(16)))
        raise ValueError(f"Incomplete block; missing={missing}, extra={extra}")
    return {
        name: float(sum(sign * probabilities[mask] for mask, sign in terms.items()))
        for name, terms in CONTRAST_TERMS.items()
    }


def block_summary(probabilities: dict[int, float]) -> dict[str, float]:
    result = contrasts(probabilities)
    result.update(
        {
            "p_empty": float(probabilities[0]),
            "p_full": float(probabilities[15]),
            "mean_singletons": float(
                np.mean([value for mask, value in probabilities.items() if mask.bit_count() == 1])
            ),
            "mean_pairs": float(
                np.mean([value for mask, value in probabilities.items() if mask.bit_count() == 2])
            ),
            "mean_triples": float(
                np.mean([value for mask, value in probabilities.items() if mask.bit_count() == 3])
            ),
        }
    )
    return result


def holm_adjust(p_values: dict[str, float]) -> dict[str, float]:
    ordered = sorted(p_values, key=lambda name: (p_values[name], name))
    adjusted: dict[str, float] = {}
    running = 0.0
    total = len(ordered)
    for rank, name in enumerate(ordered):
        candidate = min(1.0, (total - rank) * float(p_values[name]))
        running = max(running, candidate)
        adjusted[name] = running
    return adjusted


def _safe_sign(value: float, tolerance: float = 1e-15) -> int:
    if value > tolerance:
        return 1
    if value < -tolerance:
        return -1
    return 0


def _mean_metrics(
    block_rows: dict[str, dict[str, Any]],
    block_ids: Iterable[str],
    variant: str,
) -> dict[str, float]:
    ids = list(block_ids)
    return {
        name: float(np.mean([block_rows[block_id][variant][name] for block_id in ids]))
        for name in CONTRAST_NAMES
    }


def _bootstrap_summary(values: np.ndarray, point: float) -> dict[str, Any]:
    valid = values[np.isfinite(values)]
    if valid.size == 0:
        return {
            "estimate": point,
            "bootstrap_standard_error": None,
            "bootstrap_95pct_low": None,
            "bootstrap_95pct_high": None,
            "valid_replicates": 0,
        }
    standard_error = float(valid.std(ddof=1)) if valid.size > 1 else 0.0
    low, high = np.percentile(valid, [2.5, 97.5])
    if standard_error == 0.0:
        p_value = 1.0 if abs(point) <= 1e-15 else 0.0
    else:
        p_value = float(2.0 * norm.sf(abs(point / standard_error)))
    return {
        "estimate": float(point),
        "bootstrap_standard_error": standard_error,
        "bootstrap_95pct_low": float(low),
        "bootstrap_95pct_high": float(high),
        "valid_replicates": int(valid.size),
        "normal_two_sided_p": p_value,
    }


def _bootstrap_stability_detail(values: np.ndarray) -> dict[str, Any]:
    """Check split-half CI endpoint stability against 10% of full CI width."""

    midpoint = len(values) // 2
    full = values[np.isfinite(values)]
    first = values[:midpoint]
    first = first[np.isfinite(first)]
    second = values[midpoint:]
    second = second[np.isfinite(second)]
    if full.size < 20 or first.size < 10 or second.size < 10:
        return {
            "stable": False,
            "reason": "insufficient_valid_replicates",
            "valid_replicates": int(full.size),
            "first_half_valid_replicates": int(first.size),
            "second_half_valid_replicates": int(second.size),
            "full_bootstrap_95pct": [None, None],
            "first_half_bootstrap_95pct": [None, None],
            "second_half_bootstrap_95pct": [None, None],
            "full_interval_width": None,
            "endpoint_tolerance": None,
            "maximum_half_endpoint_difference": None,
        }
    full_bounds = np.percentile(full, [2.5, 97.5])
    first_bounds = np.percentile(first, [2.5, 97.5])
    second_bounds = np.percentile(second, [2.5, 97.5])
    width = float(full_bounds[1] - full_bounds[0])
    endpoint_difference = float(
        np.max(np.abs(first_bounds - second_bounds))
    )
    if width <= 1e-15:
        tolerance = 1e-12
        stable = endpoint_difference <= tolerance
    else:
        tolerance = 0.10 * width
        stable = endpoint_difference <= tolerance
    return {
        "stable": bool(stable),
        "reason": None if stable else "split_half_ci_endpoints_changed",
        "valid_replicates": int(full.size),
        "first_half_valid_replicates": int(first.size),
        "second_half_valid_replicates": int(second.size),
        "full_bootstrap_95pct": [
            float(full_bounds[0]),
            float(full_bounds[1]),
        ],
        "first_half_bootstrap_95pct": [
            float(first_bounds[0]),
            float(first_bounds[1]),
        ],
        "second_half_bootstrap_95pct": [
            float(second_bounds[0]),
            float(second_bounds[1]),
        ],
        "full_interval_width": width,
        "endpoint_tolerance": tolerance,
        "maximum_half_endpoint_difference": endpoint_difference,
    }


def _bootstrap_is_stable(values: np.ndarray) -> bool:
    return bool(_bootstrap_stability_detail(values)["stable"])


def _resample_assignment(
    calibration_records: dict[str, dict[int, dict[int, dict[str, Any]]]],
    position: str,
    rng: np.random.Generator,
) -> np.ndarray:
    repeats = sorted(calibration_records[position])
    selected_repeats = rng.choice(repeats, size=len(repeats), replace=True)
    grouped = {state: np.zeros(8, dtype=float) for state in range(8)}
    for repeat in selected_repeats:
        for state in range(8):
            record = calibration_records[position][int(repeat)][state]
            probabilities = record["_counts_vector"] / record["shots"]
            grouped[state] += rng.multinomial(record["shots"], probabilities)
    matrix = np.zeros((8, 8), dtype=float)
    for state in range(8):
        matrix[:, state] = grouped[state] / grouped[state].sum()
    return matrix


def _point_block_rows(
    records: list[dict[str, Any]],
    before: np.ndarray,
    after: np.ndarray,
    before_available: bool,
    after_available: bool,
) -> tuple[dict[str, dict[str, Any]], dict[str, dict[int, dict[str, Any]]]]:
    by_block_records: dict[str, dict[int, dict[str, Any]]] = defaultdict(dict)
    for record in records:
        if record["kind"] == "experiment":
            by_block_records[record["block_id"]][
                int(record["logical_mask"], 2)
            ] = record
    rows: dict[str, dict[str, Any]] = {}
    for block_id, mask_records in by_block_records.items():
        raw_probabilities = {
            mask: float(record["_counts_vector"][0] / record["shots"])
            for mask, record in mask_records.items()
        }
        first_record = next(iter(mask_records.values()))
        row: dict[str, Any] = {
            "block_id": block_id,
            "experiment_type": first_record["experiment_type"],
            "stratum": first_record["stratum"],
            "logical_to_physical_mapping": first_record[
                "logical_to_physical_mapping"
            ],
            "execution_index_mean": float(
                np.mean([record["execution_index"] for record in mask_records.values()])
            ),
            "executed_at_mean": (
                None
                if not all(record["_timestamp"] is not None for record in mask_records.values())
                else float(
                    np.mean(
                        [record["_timestamp"] for record in mask_records.values()]
                    )
                )
            ),
            "raw": block_summary(raw_probabilities),
        }
        rows[block_id] = row

    for variant, matrix, available in (
        ("mitigated_before", before, before_available),
        ("mitigated_after", after, after_available),
    ):
        corrected_by_block: dict[str, dict[str, float]] = {}
        if available:
            try:
                for block_id, mask_records in by_block_records.items():
                    corrected_probabilities: dict[int, float] = {}
                    for mask, record in mask_records.items():
                        corrected = mitigate(
                            record["_counts_vector"] / record["shots"], matrix
                        )
                        if not np.all(np.isfinite(corrected)):
                            raise FloatingPointError(
                                "Readout correction produced non-finite values"
                            )
                        corrected_probabilities[mask] = float(corrected[0])
                    corrected_by_block[block_id] = block_summary(
                        corrected_probabilities
                    )
            except (
                ValueError,
                RuntimeError,
                FloatingPointError,
                np.linalg.LinAlgError,
            ):
                corrected_by_block = {}
        point_variant_available = len(corrected_by_block) == len(rows)
        for block_id in rows:
            rows[block_id][variant] = (
                corrected_by_block[block_id]
                if point_variant_available
                else None
            )
    return rows, by_block_records


def _ordered_primary_blocks(
    block_rows: dict[str, dict[str, Any]],
    payload: dict[str, Any],
) -> tuple[list[str], str, bool]:
    primary_ids = [
        block_id
        for block_id, row in block_rows.items()
        if row["experiment_type"] == "primary"
    ]
    if all(block_rows[block_id]["executed_at_mean"] is not None for block_id in primary_ids):
        return (
            sorted(primary_ids, key=lambda block_id: block_rows[block_id]["executed_at_mean"]),
            "executed_at_utc",
            True,
        )
    if payload["schedule_order_preserved"] is True:
        return (
            sorted(
                primary_ids,
                key=lambda block_id: block_rows[block_id]["execution_index_mean"],
            ),
            "execution_index",
            True,
        )
    return sorted(primary_ids), "unavailable", False


def _hierarchical_bootstrap(
    validated: dict[str, Any],
    block_rows: dict[str, dict[str, Any]],
    by_block_records: dict[str, dict[int, dict[str, Any]]],
    *,
    repetitions: int,
    seed: int,
    before_available: bool,
    after_available: bool,
    ordered_primary: list[str],
) -> dict[str, Any]:
    if repetitions < 20:
        raise ValueError("bootstrap_reps must be at least 20")
    rng = np.random.default_rng(seed)
    records = validated["records"]
    calibration_records: dict[
        str, dict[int, dict[int, dict[str, Any]]]
    ] = defaultdict(lambda: defaultdict(dict))
    for record in records:
        if record["kind"] == "readout_calibration":
            calibration_records[record["position"]][record["repeat"]][
                record["prepared_state"]
            ] = record

    block_ids_by_type = {
        experiment_type: sorted(
            block_id
            for block_id, row in block_rows.items()
            if row["experiment_type"] == experiment_type
        )
        for experiment_type in EXPERIMENT_TYPES
    }
    samples = {
        variant: {
            experiment_type: {
                name: np.full(repetitions, np.nan, dtype=float)
                for name in CONTRAST_NAMES
            }
            for experiment_type in EXPERIMENT_TYPES
        }
        for variant in VARIANTS
    }
    early_samples = {
        variant: {
            name: np.full(repetitions, np.nan, dtype=float)
            for name in CONTRAST_NAMES
        }
        for variant in VARIANTS
    }
    late_samples = {
        variant: {
            name: np.full(repetitions, np.nan, dtype=float)
            for name in CONTRAST_NAMES
        }
        for variant in VARIANTS
    }
    condition_numbers = {
        "before": np.full(repetitions, np.nan),
        "after": np.full(repetitions, np.nan),
    }

    early_ids = ordered_primary[:8]
    late_ids = ordered_primary[8:]
    for replicate in range(repetitions):
        before_matrix: np.ndarray | None = None
        after_matrix: np.ndarray | None = None
        before_condition = math.inf
        after_condition = math.inf
        try:
            before_matrix = _resample_assignment(
                calibration_records, "before", rng
            )
            before_condition = float(np.linalg.cond(before_matrix))
            condition_numbers["before"][replicate] = before_condition
        except (ValueError, RuntimeError, FloatingPointError, np.linalg.LinAlgError):
            before_matrix = None
        try:
            after_matrix = _resample_assignment(calibration_records, "after", rng)
            after_condition = float(np.linalg.cond(after_matrix))
            condition_numbers["after"][replicate] = after_condition
        except (ValueError, RuntimeError, FloatingPointError, np.linalg.LinAlgError):
            after_matrix = None
        replicate_available = {
            "raw": True,
            "mitigated_before": bool(
                before_available
                and before_matrix is not None
                and math.isfinite(before_condition)
                and before_condition <= 10.0
            ),
            "mitigated_after": bool(
                after_available
                and after_matrix is not None
                and math.isfinite(after_condition)
                and after_condition <= 10.0
            ),
        }

        selected_by_type = {
            experiment_type: rng.choice(
                ids, size=len(ids), replace=True
            ).tolist()
            for experiment_type, ids in block_ids_by_type.items()
        }
        selected_groups = {
            **{
                f"type:{experiment_type}": selected
                for experiment_type, selected in selected_by_type.items()
            },
            "early": rng.choice(
                early_ids, size=len(early_ids), replace=True
            ).tolist(),
            "late": rng.choice(
                late_ids, size=len(late_ids), replace=True
            ).tolist(),
        }

        # Draw multinomial counts independently for every selected block
        # occurrence.  A block selected twice therefore contributes two
        # independent within-block draws, as required by the hierarchical
        # bootstrap, while every analysis variant sees the same paired draws.
        occurrence_distributions: dict[
            str, list[dict[int, np.ndarray]]
        ] = {}
        group_summaries: dict[
            str, dict[str, list[dict[str, float]]]
        ] = {}
        for group_name, selected_ids in selected_groups.items():
            occurrence_distributions[group_name] = []
            raw_summaries: list[dict[str, float]] = []
            for block_id in selected_ids:
                distributions: dict[int, np.ndarray] = {}
                for mask, record in by_block_records[block_id].items():
                    empirical = record["_counts_vector"] / record["shots"]
                    resampled_vector = rng.multinomial(
                        record["shots"], empirical
                    )
                    distributions[mask] = (
                        resampled_vector / record["shots"]
                    )
                occurrence_distributions[group_name].append(distributions)
                raw_summaries.append(
                    block_summary(
                        {
                            mask: float(distribution[0])
                            for mask, distribution in distributions.items()
                        }
                    )
                )
            group_summaries[group_name] = {"raw": raw_summaries}

        for variant, matrix in (
            ("mitigated_before", before_matrix),
            ("mitigated_after", after_matrix),
        ):
            if not replicate_available[variant] or matrix is None:
                continue
            try:
                corrected_groups: dict[
                    str, list[dict[str, float]]
                ] = {}
                for group_name, occurrences in occurrence_distributions.items():
                    corrected_summaries: list[dict[str, float]] = []
                    for distributions in occurrences:
                        corrected_probabilities: dict[int, float] = {}
                        for mask, distribution in distributions.items():
                            corrected = mitigate(distribution, matrix)
                            if not np.all(np.isfinite(corrected)):
                                raise FloatingPointError(
                                    "Readout correction produced non-finite values"
                                )
                            corrected_probabilities[mask] = float(corrected[0])
                        corrected_summaries.append(
                            block_summary(corrected_probabilities)
                        )
                    corrected_groups[group_name] = corrected_summaries
                for group_name, summaries in corrected_groups.items():
                    group_summaries[group_name][variant] = summaries
            except (
                ValueError,
                RuntimeError,
                FloatingPointError,
                np.linalg.LinAlgError,
            ):
                # One failed fit invalidates this mitigation variant for the
                # entire paired replicate, while raw and the other variant
                # remain usable.  The preallocated NaNs preserve that fact.
                replicate_available[variant] = False

        for variant in VARIANTS:
            if not replicate_available[variant]:
                continue
            for experiment_type in EXPERIMENT_TYPES:
                summaries = group_summaries[f"type:{experiment_type}"][
                    variant
                ]
                for name in CONTRAST_NAMES:
                    samples[variant][experiment_type][name][replicate] = float(
                        np.mean([summary[name] for summary in summaries])
                    )
            for name in CONTRAST_NAMES:
                early_samples[variant][name][replicate] = float(
                    np.mean(
                        [
                            summary[name]
                            for summary in group_summaries["early"][variant]
                        ]
                    )
                )
                late_samples[variant][name][replicate] = float(
                    np.mean(
                        [
                            summary[name]
                            for summary in group_summaries["late"][variant]
                        ]
                    )
                )

    return {
        "samples": samples,
        "early_samples": early_samples,
        "late_samples": late_samples,
        "condition_numbers": condition_numbers,
        "repetitions": repetitions,
        "seed": seed,
    }


def _bootstrap_variant_validity(bootstrap: dict[str, Any]) -> dict[str, Any]:
    """Summarize usable paired replicates for each primary analysis variant."""

    repetitions = int(bootstrap["repetitions"])
    output: dict[str, Any] = {}
    for variant in VARIANTS:
        metric_fractions = {
            name: float(
                np.mean(
                    np.isfinite(
                        bootstrap["samples"][variant]["primary"][name]
                    )
                )
            )
            for name in CONTRAST_NAMES
        }
        joint_valid = np.ones(repetitions, dtype=bool)
        for name in CONTRAST_NAMES:
            joint_valid &= np.isfinite(
                bootstrap["samples"][variant]["primary"][name]
            )
        output[variant] = {
            "primary_contrast_valid_fraction_by_name": metric_fractions,
            "minimum_primary_contrast_valid_fraction": float(
                min(metric_fractions.values())
            ),
            "mean_primary_contrast_valid_fraction": float(
                np.mean(list(metric_fractions.values()))
            ),
            "joint_primary_valid_fraction": float(np.mean(joint_valid)),
            "joint_primary_valid_replicates": int(np.count_nonzero(joint_valid)),
        }
    return output


def _bootstrap_stability_analysis(bootstrap: dict[str, Any]) -> dict[str, Any]:
    """Apply the registered split-half stability gate independently."""

    variants: dict[str, Any] = {}
    flags: list[str] = []
    for variant in VARIANTS:
        contrast_results: dict[str, Any] = {}
        for name in CONTRAST_NAMES:
            detail = _bootstrap_stability_detail(
                bootstrap["samples"][variant]["primary"][name]
            )
            contrast_results[name] = detail
            if not detail["stable"]:
                flags.append(f"{variant}:{name}")
        variants[variant] = {
            "all_primary_contrasts_stable": all(
                result["stable"] for result in contrast_results.values()
            ),
            "contrasts": contrast_results,
        }
    return {
        "criterion": (
            "split-half 95% CI endpoint differences must be no greater "
            "than 10% of the full 95% CI width"
        ),
        "variants": variants,
        "inconsistency_flags": sorted(set(flags)),
    }


def _paired_variant_agreement(bootstrap: dict[str, Any]) -> dict[str, Any]:
    """Paired bootstrap checks for dependence on readout correction choice."""

    pair_definitions = (
        ("raw", "mitigated_before"),
        ("raw", "mitigated_after"),
        ("mitigated_before", "mitigated_after"),
    )
    repetitions = int(bootstrap["repetitions"])
    contrasts_output: dict[str, Any] = {}
    flags: list[str] = []
    for name in CONTRAST_NAMES:
        pair_output: dict[str, Any] = {}
        for left, right in pair_definitions:
            pair_name = f"{left}_minus_{right}"
            left_values = bootstrap["samples"][left]["primary"][name]
            right_values = bootstrap["samples"][right]["primary"][name]
            valid = np.isfinite(left_values) & np.isfinite(right_values)
            differences = (left_values - right_values)[valid]
            if differences.size:
                low, high = np.percentile(differences, [2.5, 97.5])
                bounds: list[float | None] = [float(low), float(high)]
                mean_difference: float | None = float(np.mean(differences))
                excludes_zero = bool(low > 0.0 or high < 0.0)
            else:
                bounds = [None, None]
                mean_difference = None
                excludes_zero = False
            if excludes_zero:
                flags.append(f"{name}:{pair_name}")
            pair_output[pair_name] = {
                "mean_paired_difference": mean_difference,
                "bootstrap_95pct_low": bounds[0],
                "bootstrap_95pct_high": bounds[1],
                "valid_replicates": int(differences.size),
                "valid_fraction": float(differences.size / repetitions),
                "excludes_zero": excludes_zero,
            }
        contrasts_output[name] = pair_output
    return {
        "method": "paired hierarchical bootstrap differences",
        "contrasts": contrasts_output,
        "inconsistency_flags": sorted(set(flags)),
    }


def _aggregate_statistics(
    block_rows: dict[str, dict[str, Any]],
    bootstrap: dict[str, Any],
) -> tuple[dict[str, Any], dict[str, dict[str, float]]]:
    statistics: dict[str, Any] = {}
    holm_by_variant: dict[str, dict[str, float]] = {}
    for variant in VARIANTS:
        point_available = all(
            row.get(variant) is not None for row in block_rows.values()
        )
        bootstrap_available = all(
            np.any(
                np.isfinite(
                    bootstrap["samples"][variant][experiment_type][name]
                )
            )
            for experiment_type in EXPERIMENT_TYPES
            for name in CONTRAST_NAMES
        )
        if not point_available or not bootstrap_available:
            statistics[variant] = {"available": False}
            continue
        statistics[variant] = {"available": True}
        for experiment_type in EXPERIMENT_TYPES:
            ids = [
                block_id
                for block_id, row in block_rows.items()
                if row["experiment_type"] == experiment_type
            ]
            points = _mean_metrics(block_rows, ids, variant)
            metrics = {
                name: _bootstrap_summary(
                    bootstrap["samples"][variant][experiment_type][name],
                    points[name],
                )
                for name in CONTRAST_NAMES
            }
            statistics[variant][experiment_type] = metrics
        p_values = {
            name: statistics[variant]["primary"][name]["normal_two_sided_p"]
            for name in CONTRAST_NAMES
        }
        holm = holm_adjust(p_values)
        holm_by_variant[variant] = holm
        for name in CONTRAST_NAMES:
            statistics[variant]["primary"][name]["holm_adjusted_p"] = holm[name]
            statistics[variant]["primary"][name]["holm_reject_0_05"] = bool(
                holm[name] < 0.05
            )
            statistics[variant]["primary"][name]["holm_reject_5sigma"] = bool(
                holm[name] < FIVE_SIGMA_TWO_SIDED_ALPHA
            )
    return statistics, holm_by_variant


def _drift_analysis(
    block_rows: dict[str, dict[str, Any]],
    by_block_records: dict[str, dict[int, dict[str, Any]]],
    bootstrap: dict[str, Any],
    ordered_primary: list[str],
    time_basis: str,
    time_available: bool,
) -> dict[str, Any]:
    early_ids = ordered_primary[:8]
    late_ids = ordered_primary[8:]
    anchor_results: dict[str, Any] = {}
    drift_flags: list[str] = []
    for anchor in ("p_empty", "p_full"):
        early = np.array([block_rows[block_id]["raw"][anchor] for block_id in early_ids])
        late = np.array([block_rows[block_id]["raw"][anchor] for block_id in late_ids])
        delta = float(early.mean() - late.mean())
        standard_error = float(
            math.sqrt(
                (early.var(ddof=1) / len(early) if len(early) > 1 else 0.0)
                + (late.var(ddof=1) / len(late) if len(late) > 1 else 0.0)
            )
        )
        exceeds_three_se = (
            abs(delta) > 0.0 if standard_error == 0.0 else abs(delta) > 3.0 * standard_error
        )
        flagged = abs(delta) > 0.02 or exceeds_three_se
        if flagged:
            drift_flags.append(f"{anchor}_early_late")
        anchor_results[anchor] = {
            "early_mean": float(early.mean()),
            "late_mean": float(late.mean()),
            "difference": delta,
            "standard_error": standard_error,
            "exceeds_0_02_or_3se": flagged,
        }

    slopes: dict[str, Any] = {}
    if time_available:
        if time_basis == "executed_at_utc":
            raw_time = np.array(
                [block_rows[block_id]["executed_at_mean"] for block_id in ordered_primary]
            )
            span = raw_time.max() - raw_time.min()
            if span == 0:
                time_available = False
                times = None
            else:
                times = (raw_time - raw_time.min()) / span
        else:
            times = np.linspace(0.0, 1.0, len(ordered_primary))
    if time_available:
        assert times is not None
        for name in CONTRAST_NAMES:
            values = np.array(
                [block_rows[block_id]["raw"][name] for block_id in ordered_primary]
            )
            if np.allclose(values, values[0]):
                slope, p_value = 0.0, 1.0
            else:
                regression = linregress(times, values)
                slope, p_value = float(regression.slope), float(regression.pvalue)
            primary_mean = float(values.mean())
            predicted_change = abs(slope)
            comparable = predicted_change >= abs(primary_mean)
            flagged = p_value < 0.01 and comparable
            if flagged:
                drift_flags.append(f"{name}_time_slope")
            slopes[name] = {
                "slope_per_run_span": slope,
                "two_sided_p": p_value,
                "predicted_change": predicted_change,
                "primary_mean": primary_mean,
                "significant_and_comparable": flagged,
            }
    else:
        drift_flags.append("execution_time_unavailable")
        for name in CONTRAST_NAMES:
            values = np.array(
                [block_rows[block_id]["raw"][name] for block_id in ordered_primary]
            )
            slopes[name] = {
                "slope_per_run_span": None,
                "two_sided_p": None,
                "predicted_change": None,
                "primary_mean": float(values.mean()),
                "significant_and_comparable": False,
            }

    overdispersion: dict[str, Any] = {}
    for name in CONTRAST_NAMES:
        values: list[float] = []
        variances: list[float] = []
        for block_id in ordered_primary:
            values.append(block_rows[block_id]["raw"][name])
            variance = 0.0
            for mask in CONTRAST_TERMS[name]:
                record = by_block_records[block_id][mask]
                probability = float(record["_counts_vector"][0] / record["shots"])
                variance += probability * (1.0 - probability) / record["shots"]
            variances.append(variance)
        values_array = np.array(values)
        variances_array = np.array(variances)
        if np.all(variances_array <= 0):
            q_statistic, p_value = 0.0, 1.0
        else:
            safe_variances = np.maximum(variances_array, 1e-15)
            weights = 1.0 / safe_variances
            weighted_mean = float(np.sum(weights * values_array) / weights.sum())
            q_statistic = float(
                np.sum(weights * (values_array - weighted_mean) ** 2)
            )
            p_value = float(chi2.sf(q_statistic, len(values_array) - 1))
        stable = _bootstrap_is_stable(
            bootstrap["samples"]["raw"]["primary"][name]
        )
        flagged = p_value < 0.01 and not stable
        if flagged:
            drift_flags.append(f"{name}_overdispersion_unstable")
        overdispersion[name] = {
            "cochran_q": q_statistic,
            "degrees_of_freedom": len(values_array) - 1,
            "p_value": p_value,
            "bootstrap_stable": stable,
            "inconclusive": flagged,
        }
    return {
        "time_basis": time_basis,
        "time_available": time_available,
        "early_block_ids": early_ids,
        "late_block_ids": late_ids,
        "anchors": anchor_results,
        "slopes": slopes,
        "overdispersion": overdispersion,
        "flags": sorted(set(drift_flags)),
    }


def _systematics_analysis(
    block_rows: dict[str, dict[str, Any]],
    bootstrap: dict[str, Any],
    ordered_primary: list[str],
) -> dict[str, Any]:
    ids_by_type = {
        experiment_type: [
            block_id
            for block_id, row in block_rows.items()
            if row["experiment_type"] == experiment_type
        ]
        for experiment_type in EXPERIMENT_TYPES
    }
    early_ids, late_ids = ordered_primary[:8], ordered_primary[8:]
    output: dict[str, Any] = {}
    for name in CONTRAST_NAMES:
        sign_components: list[float] = []
        time_components: list[float] = []
        variant_points: dict[str, Any] = {}
        for variant in VARIANTS:
            if any(block_rows[block_id].get(variant) is None for block_id in block_rows):
                continue
            primary = _mean_metrics(
                block_rows, ids_by_type["primary"], variant
            )[name]
            sign = _mean_metrics(block_rows, ids_by_type["sign"], variant)[name]
            phase = _mean_metrics(block_rows, ids_by_type["phase"], variant)[name]
            early = _mean_metrics(block_rows, early_ids, variant)[name]
            late = _mean_metrics(block_rows, late_ids, variant)[name]
            sign_components.append(abs(primary - sign) / 2.0)
            time_components.append(abs(early - late) / 2.0)
            variant_points[variant] = {
                "primary": primary,
                "sign_control": sign,
                "phase_control": phase,
                "early": early,
                "late": late,
            }
        before_samples = bootstrap["samples"]["mitigated_before"]["primary"][name]
        after_samples = bootstrap["samples"]["mitigated_after"]["primary"][name]
        valid = np.isfinite(before_samples) & np.isfinite(after_samples)
        if np.any(valid):
            readout_component = float(
                np.percentile(
                    np.abs(before_samples[valid] - after_samples[valid]), 95.0
                )
            )
        else:
            readout_component = math.inf
        sign_component = max(sign_components, default=math.inf)
        time_component = max(time_components, default=math.inf)
        b_sys_numeric = max(sign_component, time_component, readout_component)
        b_sys = b_sys_numeric if math.isfinite(b_sys_numeric) else None
        output[name] = {
            "variants": variant_points,
            "sign_half_difference_max": (
                sign_component if math.isfinite(sign_component) else None
            ),
            "early_late_half_difference_max": (
                time_component if math.isfinite(time_component) else None
            ),
            "readout_bootstrap_95pct_full_difference": (
                readout_component if math.isfinite(readout_component) else None
            ),
            "B_sys": b_sys,
        }
    return output


def _control_analysis(
    block_rows: dict[str, dict[str, Any]],
    bootstrap: dict[str, Any],
) -> dict[str, Any]:
    results: dict[str, Any] = {}
    inconsistent: list[str] = []
    for variant in VARIANTS:
        if any(row.get(variant) is None for row in block_rows.values()):
            results[variant] = {"available": False}
            continue
        variant_result: dict[str, Any] = {"available": True}
        mapping_groups: dict[tuple[int, ...], list[str]] = defaultdict(list)
        for block_id, row in block_rows.items():
            if row["experiment_type"] == "primary":
                mapping_groups[tuple(row["logical_to_physical_mapping"])].append(
                    block_id
                )
        for name in CONTRAST_NAMES:
            primary_samples = bootstrap["samples"][variant]["primary"][name]
            sign_samples = bootstrap["samples"][variant]["sign"][name]
            phase_samples = bootstrap["samples"][variant]["phase"][name]
            sign_difference = primary_samples - sign_samples
            phase_difference = primary_samples - phase_samples
            sign_valid = sign_difference[np.isfinite(sign_difference)]
            phase_valid = phase_difference[np.isfinite(phase_difference)]
            sign_bounds = (
                [None, None]
                if sign_valid.size == 0
                else [float(value) for value in np.percentile(sign_valid, [2.5, 97.5])]
            )
            phase_bounds = (
                [None, None]
                if phase_valid.size == 0
                else [float(value) for value in np.percentile(phase_valid, [2.5, 97.5])]
            )
            sign_inconsistent = (
                sign_bounds[0] is not None
                and (sign_bounds[0] > 0.0 or sign_bounds[1] < 0.0)
            )
            phase_inconsistent = (
                phase_bounds[0] is not None
                and (phase_bounds[0] > 0.0 or phase_bounds[1] < 0.0)
            )
            groups = [
                [block_rows[block_id][variant][name] for block_id in ids]
                for ids in mapping_groups.values()
            ]
            flattened_groups = np.concatenate(
                [np.asarray(group, dtype=float) for group in groups]
            )
            if np.allclose(flattened_groups, flattened_groups[0]):
                mapping_p = 1.0
            else:
                mapping_p = float(f_oneway(*groups).pvalue)
                if not math.isfinite(mapping_p):
                    mapping_p = 1.0
            mapping_inconsistent = mapping_p < 0.01
            if sign_inconsistent:
                inconsistent.append(f"{variant}:{name}:sign")
            if phase_inconsistent:
                inconsistent.append(f"{variant}:{name}:phase")
            if mapping_inconsistent:
                inconsistent.append(f"{variant}:{name}:mapping")
            variant_result[name] = {
                "primary_minus_sign_bootstrap_95pct": sign_bounds,
                "primary_minus_phase_bootstrap_95pct": phase_bounds,
                "mapping_anova_p": mapping_p,
                "inconsistent": bool(
                    sign_inconsistent or phase_inconsistent or mapping_inconsistent
                ),
            }
        results[variant] = variant_result
    results["inconsistency_flags"] = sorted(set(inconsistent))
    return results


def _classify(
    statistics: dict[str, Any],
    readout: dict[str, Any],
    drift: dict[str, Any],
    systematics: dict[str, Any],
    controls: dict[str, Any],
    variant_agreement: dict[str, Any],
    bootstrap_stability: dict[str, Any],
    execution_integrity: dict[str, Any],
    block_rows: dict[str, dict[str, Any]],
) -> dict[str, Any]:
    flags: list[str] = []
    if (
        not readout["condition_number_acceptable_before"]
        or not readout["condition_number_acceptable_after"]
    ):
        flags.append("readout_condition_number_above_10")
    if (
        readout["condition_number_acceptable_before"]
        and not readout["point_mitigation_fit_available_before"]
    ):
        flags.append("readout_mitigation_fit_failed:before")
    if (
        readout["condition_number_acceptable_after"]
        and not readout["point_mitigation_fit_available_after"]
    ):
        flags.append("readout_mitigation_fit_failed:after")
    if not readout["drift_below_or_equal_0_02"]:
        flags.append("readout_matrix_drift_above_0_02")
    valid_before = readout["bootstrap_valid_fraction_before"]
    valid_after = readout["bootstrap_valid_fraction_after"]
    if valid_before < 0.95 or valid_after < 0.95:
        flags.append("more_than_5pct_invalid_mitigation_bootstraps")
    flags.extend(drift["flags"])
    flags.extend(controls["inconsistency_flags"])
    flags.extend(
        f"mitigation_variant_disagreement:{flag}"
        for flag in variant_agreement["inconsistency_flags"]
    )
    flags.extend(
        f"bootstrap_unstable:{flag}"
        for flag in bootstrap_stability["inconsistency_flags"]
    )
    if execution_integrity["schedule_order_preserved"] is not True:
        flags.append("execution_order_not_verified")
    if execution_integrity["execution_bracketing_verified"] is not True:
        flags.append("readout_bracketing_not_verified")

    available_variants = [
        variant for variant in VARIANTS if statistics[variant]["available"]
    ]
    all_null = len(available_variants) == 3
    all_candidate: list[str] = []
    for name in CONTRAST_NAMES:
        if len(available_variants) != 3:
            all_null = False
            continue
        for variant in available_variants:
            metric = statistics[variant]["primary"][name]
            if metric["holm_adjusted_p"] < 0.05:
                all_null = False
            if not (
                metric["bootstrap_95pct_low"] <= 0.0
                <= metric["bootstrap_95pct_high"]
            ):
                all_null = False
        point_signs = {
            variant: _safe_sign(
                statistics[variant]["primary"][name]["estimate"]
            )
            for variant in available_variants
        }
        primary_sign = point_signs["raw"]
        candidate = (
            primary_sign != 0
            and all(sign == primary_sign for sign in point_signs.values())
            and all(
                statistics[variant]["primary"][name]["holm_adjusted_p"]
                < FIVE_SIGMA_TWO_SIDED_ALPHA
                for variant in available_variants
            )
            and systematics[name]["B_sys"] is not None
            and abs(statistics["raw"]["primary"][name]["estimate"])
            > 5.0 * systematics[name]["B_sys"]
        )
        if candidate:
            for variant in available_variants:
                points = systematics[name]["variants"][variant]
                required_signs = (
                    _safe_sign(points["sign_control"]),
                    _safe_sign(points["phase_control"]),
                )
                primary_value = points["primary"]
                control_values = (
                    points["sign_control"],
                    points["phase_control"],
                )
                mapping_values = {
                    float(
                        np.mean(
                            [
                                row[variant][name]
                                for row in block_rows.values()
                                if row["experiment_type"] == "primary"
                                and tuple(row["logical_to_physical_mapping"])
                                == mapping
                            ]
                        )
                    )
                    for mapping in {
                        tuple(row["logical_to_physical_mapping"])
                        for row in block_rows.values()
                        if row["experiment_type"] == "primary"
                    }
                }
                mapping_signs = {_safe_sign(value) for value in mapping_values}
                same_magnitude = all(
                    abs(value - primary_value) <= 0.5 * abs(primary_value)
                    for value in (*control_values, *mapping_values)
                )
                if (
                    any(sign != primary_sign for sign in required_signs)
                    or any(sign != primary_sign for sign in mapping_signs)
                    or not same_magnitude
                ):
                    candidate = False
        if candidate:
            all_candidate.append(name)

    flags = sorted(set(flags))
    if not flags and all_candidate:
        label = "candidate_anomaly"
    elif not flags and all_null:
        label = "null_validation"
    else:
        label = "inconclusive"
    return {
        "label": label,
        "candidate_contrasts": all_candidate if label == "candidate_anomaly" else [],
        "inconclusive_flags": flags,
        "wording": {
            "null_validation": "We did not reject H0 at the present sensitivity.",
            "inconclusive": "Systematics, drift, controls, or uncertainty prevent a null/anomaly claim.",
            "candidate_anomaly": "Candidate anomaly only; independent replication is required and this is not evidence for HQC.",
        }[label],
    }


def analyze(
    payload: dict[str, Any],
    *,
    bootstrap_reps: int = DEFAULT_BOOTSTRAP_REPS,
    seed: int = DEFAULT_SEED,
    contract: dict[str, Any] | None = None,
    provenance: dict[str, Any] | None = None,
) -> dict[str, Any]:
    contract = contract or load_frozen_contract()
    validated = validate_payload(payload, contract)
    records = validated["records"]
    before = assignment_matrix(records, "before")
    after = assignment_matrix(records, "after")
    before_condition = float(np.linalg.cond(before))
    after_condition = float(np.linalg.cond(after))
    before_available = bool(
        math.isfinite(before_condition) and before_condition <= 10.0
    )
    after_available = bool(
        math.isfinite(after_condition) and after_condition <= 10.0
    )
    max_change = float(np.max(np.abs(before - after)))

    block_rows, by_block_records = _point_block_rows(
        records, before, after, before_available, after_available
    )
    if len(block_rows) != 24:
        raise ValueError("Validated input did not produce exactly 24 blocks")
    before_condition_acceptable = before_available
    after_condition_acceptable = after_available
    before_available = bool(
        before_available
        and all(
            row.get("mitigated_before") is not None
            for row in block_rows.values()
        )
    )
    after_available = bool(
        after_available
        and all(
            row.get("mitigated_after") is not None
            for row in block_rows.values()
        )
    )
    ordered_primary, time_basis, time_available = _ordered_primary_blocks(
        block_rows, validated
    )
    bootstrap = _hierarchical_bootstrap(
        validated,
        block_rows,
        by_block_records,
        repetitions=bootstrap_reps,
        seed=seed,
        before_available=before_available,
        after_available=after_available,
        ordered_primary=ordered_primary,
    )
    statistics, _ = _aggregate_statistics(block_rows, bootstrap)
    drift = _drift_analysis(
        block_rows,
        by_block_records,
        bootstrap,
        ordered_primary,
        time_basis,
        time_available,
    )
    systematics = _systematics_analysis(block_rows, bootstrap, ordered_primary)
    controls = _control_analysis(block_rows, bootstrap)
    bootstrap_validity = _bootstrap_variant_validity(bootstrap)
    variant_agreement = _paired_variant_agreement(bootstrap)
    bootstrap_stability = _bootstrap_stability_analysis(bootstrap)
    readout = {
        "before_assignment_matrix": before.tolist(),
        "after_assignment_matrix": after.tolist(),
        "before_condition_number": (
            before_condition if math.isfinite(before_condition) else None
        ),
        "after_condition_number": (
            after_condition if math.isfinite(after_condition) else None
        ),
        "max_element_change_before_after": max_change,
        "mitigated_analysis_available": bool(before_available and after_available),
        "condition_number_acceptable_before": before_condition_acceptable,
        "condition_number_acceptable_after": after_condition_acceptable,
        "point_mitigation_fit_available_before": before_available,
        "point_mitigation_fit_available_after": after_available,
        "drift_below_or_equal_0_02": bool(max_change <= 0.02),
        "bootstrap_condition_fraction_before": float(
            np.mean(
                np.isfinite(bootstrap["condition_numbers"]["before"])
                & (bootstrap["condition_numbers"]["before"] <= 10.0)
            )
        ),
        "bootstrap_condition_fraction_after": float(
            np.mean(
                np.isfinite(bootstrap["condition_numbers"]["after"])
                & (bootstrap["condition_numbers"]["after"] <= 10.0)
            )
        ),
        "bootstrap_valid_fraction_before": bootstrap_validity[
            "mitigated_before"
        ]["joint_primary_valid_fraction"],
        "bootstrap_valid_fraction_after": bootstrap_validity[
            "mitigated_after"
        ]["joint_primary_valid_fraction"],
        "bootstrap_variant_validity": bootstrap_validity,
    }
    execution_integrity = {
        "schedule_order_preserved": validated["schedule_order_preserved"],
        "execution_bracketing_verified": validated[
            "execution_bracketing_verified"
        ],
        "derived_execution_order_status": validated[
            "_derived_execution_order_status"
        ],
        "claim_eligible": bool(
            validated["schedule_order_preserved"] is True
            and validated["execution_bracketing_verified"] is True
            and validated["_derived_execution_order_status"] is True
        ),
    }
    classification = _classify(
        statistics,
        readout,
        drift,
        systematics,
        controls,
        variant_agreement,
        bootstrap_stability,
        execution_integrity,
        block_rows,
    )

    public_blocks = []
    for block_id in sorted(block_rows):
        row = dict(block_rows[block_id])
        public_blocks.append(row)
    result = {
        "schema": ANALYSIS_SCHEMA,
        "source": validated.get("source"),
        "job_id": validated["job_id"],
        "backend": validated["backend"],
        "layout": validated["layout"],
        "runtime_versions": validated["runtime_versions"],
        "execution_time_source": validated["execution_time_source"],
        "schedule_order_preserved": validated["schedule_order_preserved"],
        "execution_bracketing_verified": validated[
            "execution_bracketing_verified"
        ],
        "artifact_hashes": {
            "plan_sha256": validated["plan_sha256"],
            "qpy_sha256": validated["qpy_sha256"],
            "manifest_sha256": validated["manifest_sha256"],
            "manifest_file_sha256": contract["manifest_file_sha256"],
            "raw_input_sha256": (
                provenance.get("raw_input_sha256")
                if provenance is not None
                else None
            ),
            "export_receipt_sha256": (
                provenance.get("export_receipt_sha256")
                if provenance is not None
                else None
            ),
        },
        "provenance": provenance
        or {
            "mode": "in-memory-or-local-self-test",
            "raw_input_sha256": None,
            "export_receipt_sha256": None,
        },
        "safety": {
            "analysis_is_local": True,
            "hardware_job_submitted_by_this_script": False,
            "ibm_or_qiskit_imports": False,
        },
        "validation": {
            "records": len(records),
            "experiment_records": 384,
            "readout_calibration_records": 64,
            "complete_blocks": len(block_rows),
            "frozen_contract_match": True,
            "contract_mode": contract["contract_mode"],
            "execution_timestamps_monotonic_in_record_order": validated[
                "_timestamps_are_monotonic"
            ],
            "execution_windows_complete": all(
                record["_execution_start"] is not None
                and record["_execution_stop"] is not None
                for record in records
            ),
        },
        "bootstrap": {
            "method": "hierarchical block + multinomial counts + calibration-repeat bootstrap",
            "repetitions": bootstrap_reps,
            "seed": seed,
        },
        "readout": readout,
        "execution_integrity": execution_integrity,
        "blocks": public_blocks,
        "statistics": statistics,
        "drift": drift,
        "controls": controls,
        "variant_agreement": variant_agreement,
        "bootstrap_stability": bootstrap_stability,
        "systematics": systematics,
        "classification": classification,
        "interpretation_guardrails": [
            "A nonzero I3/I4 is first treated as noise, drift, SPAM, or implementation dependence.",
            "No postselection and no per-context renormalization are permitted.",
            "Raw, before-calibrated, and after-calibrated results are all reported.",
            "Sign and phase controls never increase primary significance.",
            "This IBM benchmark cannot prove HQC or physics beyond ordinary quantum mechanics.",
        ],
    }
    return result


def ideal_synthetic_payload(
    plan: dict[str, Any] | None = None,
    contract: dict[str, Any] | None = None,
) -> dict[str, Any]:
    contract = contract or load_frozen_contract(allow_pre_run_fallback=True)
    plan = plan or contract["plan"]
    manifest = contract["manifest"]
    shots = int(plan["experiment"]["shots_per_circuit"])
    descriptors = _expected_record_descriptors(plan)
    bindings = iter(plan["experiment"]["parameter_bindings"])
    base_time = datetime(2026, 7, 23, 15, 30, tzinfo=timezone.utc)
    records: list[dict[str, Any]] = []
    for execution_index, descriptor in enumerate(descriptors):
        execution_start = base_time + timedelta(
            milliseconds=2 * int(descriptor["pub_index"])
        )
        execution_stop = execution_start + timedelta(milliseconds=1)
        timestamp = execution_start.isoformat()
        execution_fields = {
            "executed_at_utc": timestamp,
            "execution_start_utc": timestamp,
            "execution_stop_utc": execution_stop.isoformat(),
            "execution_time_source": "synthetic",
        }
        if descriptor["kind"] == "readout_calibration":
            state = descriptor["prepared_state"]
            records.append(
                {
                    **descriptor,
                    "shots": shots,
                    "counts": {f"{state:03b}": shots},
                    "execution_index": execution_index,
                    **execution_fields,
                }
            )
            continue
        binding = next(bindings)
        probability = float(binding["ideal_probability_000"])
        count_000 = int(round(shots * probability))
        counts: dict[str, int] = {}
        if count_000:
            counts["000"] = count_000
        if shots - count_000:
            counts["001"] = shots - count_000
        records.append(
            {
                **descriptor,
                "shots": shots,
                "counts": counts,
                "execution_index": execution_index,
                **execution_fields,
            }
        )
    pub_metadata: list[dict[str, Any]] = []
    bindings = plan["experiment"]["parameter_bindings"]
    for pub_index in range(88):
        execution_start = base_time + timedelta(milliseconds=2 * pub_index)
        execution_stop = execution_start + timedelta(milliseconds=1)
        if pub_index < 32:
            kind = "readout_calibration"
            calibration_position: str | None = "before"
            block_id: str | None = None
        elif pub_index < 56:
            kind = "experiment"
            calibration_position = None
            block_id = str(bindings[(pub_index - 32) * 16]["block_id"])
        else:
            kind = "readout_calibration"
            calibration_position = "after"
            block_id = None
        pub_metadata.append(
            {
                "pub_index": pub_index,
                "kind": kind,
                "calibration_position": calibration_position,
                "block_id": block_id,
                "result_field": "synthetic_counts",
                "metadata": {},
                "executed_at_utc": execution_start.isoformat(),
                "execution_start_utc": execution_start.isoformat(),
                "execution_stop_utc": execution_stop.isoformat(),
                "execution_time_source": "synthetic",
            }
        )
    return {
        "schema": RAW_SCHEMA,
        "source": "ideal-local-self-test",
        "plan_sha256": contract["plan_sha256"],
        "qpy_sha256": contract["qpy_sha256"],
        "manifest_sha256": contract["manifest_sha256"],
        "backend": manifest["backend_snapshot"]["backend"],
        "layout": manifest["backend_snapshot"]["layout_logical_q0_q1_flag"],
        "shots_per_circuit": shots,
        "job_id": "LOCAL-SELF-TEST",
        "runtime_versions": manifest["versions"],
        "schedule_order_preserved": True,
        "execution_bracketing_verified": True,
        "execution_time_source": "synthetic",
        "pub_metadata": pub_metadata,
        "records": records,
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", type=Path)
    parser.add_argument("--output", type=Path)
    parser.add_argument("--self-test", action="store_true")
    parser.add_argument("--manifest", type=Path, default=MANIFEST_FILE)
    parser.add_argument(
        "--bootstrap-reps", type=int, default=DEFAULT_BOOTSTRAP_REPS
    )
    parser.add_argument("--seed", type=int, default=DEFAULT_SEED)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    ROOT.mkdir(parents=True, exist_ok=True)
    contract = load_frozen_contract(
        manifest_path=args.manifest,
        allow_pre_run_fallback=args.self_test,
    )
    if args.self_test:
        payload = ideal_synthetic_payload(contract=contract)
        input_path = SYNTHETIC_COUNTS
        output_path = args.output or SYNTHETIC_ANALYSIS
        input_path.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
        provenance = {
            "mode": "local-self-test",
            "raw_input_file": input_path.name,
            "raw_input_sha256": sha256_file(input_path),
            "export_receipt_file": None,
            "export_receipt_sha256": None,
            "manifest_sha256": contract["manifest_sha256"],
        }
    else:
        if args.input is None:
            raise SystemExit("--input is required unless --self-test is used")
        input_path = args.input
        output_path = args.output or input_path.with_name(
            f"{input_path.stem}-analysis.json"
        )
        payload = json.loads(input_path.read_text(encoding="utf-8"))
        provenance = verify_export_provenance(
            input_path, payload, contract
        )

    result = analyze(
        payload,
        bootstrap_reps=args.bootstrap_reps,
        seed=args.seed,
        contract=contract,
        provenance=provenance,
    )
    result, replaced_nonfinite = replace_nonfinite_with_null(result)
    output_path.write_text(
        json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + "\n",
        encoding="utf-8",
    )
    input_path.chmod(0o600)
    output_path.chmod(0o600)
    if args.self_test:
        for variant in VARIANTS:
            if not result["statistics"][variant]["available"]:
                raise AssertionError(f"Ideal self-test lacks {variant}")
            for name in CONTRAST_NAMES:
                estimate = result["statistics"][variant]["primary"][name][
                    "estimate"
                ]
                if abs(estimate) > 1e-12:
                    raise AssertionError(
                        f"Ideal {variant} self-test failed: {name}={estimate}"
                    )
        if result["classification"]["label"] != "null_validation":
            raise AssertionError(
                f"Ideal self-test classification is {result['classification']['label']}"
            )

    print("LOCAL HQC ANALYSIS PASSED")
    print("No IBM/Qiskit imports, no hardware submission, QPU usage: 0 seconds")
    print(f"Records validated: {result['validation']['records']}")
    print(f"Blocks analyzed: {result['validation']['complete_blocks']}")
    print(f"Bootstrap repetitions: {result['bootstrap']['repetitions']}")
    print(f"Classification: {result['classification']['label']}")
    print(f"Non-finite regression fields serialized as null: {replaced_nonfinite}")
    print(f"Output: {output_path}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Run the safe end-to-end reproduction path; never contacts IBM."""

from __future__ import annotations

import argparse
import os
import subprocess
import sys
from pathlib import Path


PACKAGE_ROOT = Path(__file__).resolve().parents[1]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Offline verification only. This command has no path to "
            "SamplerV2, IBM credentials, collection, or job submission."
        )
    )
    parser.add_argument(
        "--quick",
        action="store_true",
        help=(
            "explicit alias for the safe default: 1000 synthetic and "
            "64 real-export bootstrap repetitions"
        ),
    )
    parser.add_argument(
        "--bootstrap-reps",
        type=int,
        default=1_000,
        help=(
            "Synthetic bootstrap repetitions. Use 20000 for the full "
            "registered analysis; 1000 keeps the quick path null_validation."
        ),
    )
    parser.add_argument(
        "--real-bootstrap-reps",
        type=int,
        default=64,
        help="Fast real-export smoke replay; use 20000 for the published analysis.",
    )
    parser.add_argument("--skip-tests", action="store_true")
    return parser.parse_args()


def run(*parts: str) -> None:
    env = dict(os.environ)
    env["HQC_OFFLINE"] = "1"
    subprocess.run(
        [sys.executable, *parts],
        cwd=PACKAGE_ROOT,
        env=env,
        check=True,
    )


def main() -> None:
    args = parse_args()
    if args.quick:
        args.bootstrap_reps = 1_000
        args.real_bootstrap_reps = 64
    print("SAFE OFFLINE PIPELINE")
    print("IBM/network operations: disabled; QPU submission code is not invoked")

    run("scripts/scan_public_release.py")
    run("portable-source/verify_local.py")
    run("portable-source/hqc_local_verify.py")
    run("scripts/verify_bindings.py")
    run("scripts/verify_archive.py")
    run("tests/test_hqc_revised_draft.py")
    run("tests/test_release_regressions.py")
    if not args.skip_tests:
        run("scripts/run_tests.py")
    run(
        "scripts/generate_synthetic.py",
        "--bootstrap-reps",
        str(args.bootstrap_reps),
        "--output",
        "work/verify-all-synthetic-analysis.json",
    )
    run(
        "scripts/replay_real_analysis.py",
        "--bootstrap-reps",
        str(args.real_bootstrap_reps),
        "--output",
        "work/verify-all-real-analysis.json",
    )

    print("ALL SAFE REPRODUCTION GATES PASSED")
    print("No IBM connection; no credential read; no hardware invocation; 0 QPU jobs")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Verify all 384 parameter bindings and 24 Sorkin blocks offline."""

from __future__ import annotations

import json
import math
import sys
from pathlib import Path


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PACKAGE_ROOT / "portable-source"))

import hqc_prepare_sorkin_experiment as preparation  # noqa: E402


def main() -> None:
    template, theta, phase = preparation.parametric_context_template()
    bindings = preparation.build_bindings()
    result = preparation.verify_exact_bindings(
        template,
        theta,
        phase,
        bindings,
    )
    values = [
        float(value)
        for block in result["block_null_results"]
        for value in (*block["i3"].values(), block["i4"])
    ]
    maximum = max(abs(value) for value in values)

    if result["verified_parameter_bindings"] != 384:
        raise AssertionError("Expected exactly 384 verified bindings")
    if result["verified_complete_blocks"] != 24:
        raise AssertionError("Expected exactly 24 complete blocks")
    if not math.isfinite(maximum) or maximum > 1e-12:
        raise AssertionError(f"Unexpected local Sorkin residual: {maximum}")

    output = PACKAGE_ROOT / "work" / "binding-verification.json"
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(
        json.dumps(
            {
                **result,
                "maximum_absolute_sorkin_residual": maximum,
                "network_used": False,
                "qpu_jobs_submitted": 0,
            },
            ensure_ascii=False,
            indent=2,
        )
        + "\n",
        encoding="utf-8",
    )

    print("OFFLINE STATEVECTOR VERIFICATION PASSED")
    print("IBM/network calls: 0; QPU jobs: 0")
    print(f"Parameter bindings: {result['verified_parameter_bindings']}")
    print(f"Complete blocks: {result['verified_complete_blocks']}")
    print(f"Maximum |I3/I4|: {maximum:.3e}")
    print(f"Output: {output.relative_to(PACKAGE_ROOT)}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Generate and analyze the complete 448-record synthetic payload offline."""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from pathlib import Path


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
ANALYZER = PACKAGE_ROOT / "portable-source" / "hqc_analyze_counts.py"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--bootstrap-reps", type=int, default=20_000)
    parser.add_argument("--seed", type=int, default=20_260_723)
    parser.add_argument(
        "--output",
        type=Path,
        default=PACKAGE_ROOT / "work" / "synthetic-analysis.json",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    args.output.parent.mkdir(parents=True, exist_ok=True)
    subprocess.run(
        [
            sys.executable,
            str(ANALYZER),
            "--self-test",
            "--output",
            str(args.output),
            "--bootstrap-reps",
            str(args.bootstrap_reps),
            "--seed",
            str(args.seed),
        ],
        cwd=PACKAGE_ROOT,
        check=True,
    )
    result = json.loads(args.output.read_text(encoding="utf-8"))
    if result["validation"]["records"] != 448:
        raise AssertionError("Synthetic analysis did not validate 448 records")
    if result["validation"]["complete_blocks"] != 24:
        raise AssertionError("Synthetic analysis did not validate 24 blocks")
    if result["classification"]["label"] != "null_validation":
        raise AssertionError("Ideal synthetic payload was not null_validation")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Replay the published IBM raw-counts analysis without contacting IBM."""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from pathlib import Path


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
ANALYZER = PACKAGE_ROOT / "portable-source" / "hqc_analyze_counts.py"
RAW = (
    PACKAGE_ROOT
    / "evidence"
    / "ibm-run"
    / "HQC_IBM_RAW_COUNTS_d9h516jsbqfc73epj6jg.json"
)
PUBLISHED = (
    PACKAGE_ROOT
    / "evidence"
    / "ibm-run"
    / "HQC_IBM_ANALYSIS_d9h516jsbqfc73epj6jg.json"
)
MANIFEST = (
    PACKAGE_ROOT / "evidence" / "submission" / "HQC_SUBMISSION_MANIFEST.json"
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", type=Path, default=RAW)
    parser.add_argument("--manifest", type=Path, default=MANIFEST)
    parser.add_argument("--published-analysis", type=Path, default=PUBLISHED)
    parser.add_argument("--bootstrap-reps", type=int, default=20_000)
    parser.add_argument("--seed", type=int, default=20_260_723)
    parser.add_argument(
        "--output",
        type=Path,
        default=PACKAGE_ROOT / "work" / "reproduced-real-analysis.json",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    args.output.parent.mkdir(parents=True, exist_ok=True)
    subprocess.run(
        [
            sys.executable,
            str(ANALYZER),
            "--input",
            str(args.input),
            "--manifest",
            str(args.manifest),
            "--output",
            str(args.output),
            "--bootstrap-reps",
            str(args.bootstrap_reps),
            "--seed",
            str(args.seed),
        ],
        cwd=PACKAGE_ROOT,
        check=True,
    )
    result = json.loads(args.output.read_text(encoding="utf-8"))
    if result["validation"]["records"] != 448:
        raise AssertionError("Real replay did not validate 448 records")
    if result["classification"]["label"] != "inconclusive":
        raise AssertionError(
            f"Unexpected real-run classification: {result['classification']['label']}"
        )
    drift = result["drift"]
    if drift["time_available"] is not False:
        raise AssertionError("Aggregate zero-span timing must be unavailable")
    if set(drift["slopes"]) != {
        "I3_012",
        "I3_013",
        "I3_023",
        "I3_123",
        "I4_0123",
    }:
        raise AssertionError("Portable drift schema must retain five slope entries")
    null_regression_fields = sum(
        row[field] is None
        for row in drift["slopes"].values()
        for field in (
            "slope_per_run_span",
            "two_sided_p",
            "predicted_change",
        )
    )
    if null_regression_fields != 15:
        raise AssertionError(
            f"Expected 15 null regression fields, got {null_regression_fields}"
        )
    if "execution_time_unavailable" not in drift["flags"]:
        raise AssertionError("Missing execution_time_unavailable drift flag")
    published = json.loads(args.published_analysis.read_text(encoding="utf-8"))
    point_estimates_match = all(
        result["statistics"]["raw"]["primary"][name]["estimate"]
        == published["statistics"]["raw"]["primary"][name]["estimate"]
        for name in (
            "I3_012",
            "I3_013",
            "I3_023",
            "I3_123",
            "I4_0123",
        )
    )
    if not point_estimates_match:
        raise AssertionError("Point estimates differ from the published analysis")

    registered_statistics_checked = (
        args.bootstrap_reps == 20_000 and args.seed == 20_260_723
    )
    registered_statistics_match = None
    if registered_statistics_checked:
        fields = (
            "bootstrap_95pct_low",
            "bootstrap_95pct_high",
            "holm_adjusted_p",
        )
        registered_statistics_match = all(
            result["statistics"]["raw"]["primary"][name][field]
            == published["statistics"]["raw"]["primary"][name][field]
            for name in (
                "I3_012",
                "I3_013",
                "I3_023",
                "I3_123",
                "I4_0123",
            )
            for field in fields
        )
        if not registered_statistics_match:
            raise AssertionError(
                "20,000-replicate raw confidence intervals/Holm values "
                "differ from the published analysis"
            )

    comparison = {
        "point_estimates_match": True,
        "classification_matches": (
            result["classification"]["label"]
            == published["classification"]["label"]
        ),
        "registered_statistics_checked": registered_statistics_checked,
        "registered_raw_ci_and_holm_match": registered_statistics_match,
        "byte_equality_expected": False,
        "published_time_available": published["drift"]["time_available"],
        "portable_time_available": result["drift"]["time_available"],
        "published_slope_entries": len(published["drift"]["slopes"]),
        "portable_slope_entries": len(result["drift"]["slopes"]),
        "reason": (
            "The public portable analyzer treats an aggregate zero-span "
            "ExecutionSpan as unavailable. The frozen analyzer attempted "
            "linregress and its 15 NaN values were manually serialized as null."
        ),
    }
    comparison_path = args.output.with_suffix(".comparison.json")
    comparison_path.write_text(
        json.dumps(comparison, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    print("REAL IBM EXPORT REPLAY PASSED")
    print("IBM/network calls: 0; QPU jobs: 0")
    print("Classification: inconclusive")
    print("Published point estimates: exact match")
    if registered_statistics_checked:
        print("Published raw 95% CIs and Holm values: exact match (5/5)")
    else:
        print(
            "Published CI/Holm comparison: skipped in smoke mode; "
            "use --bootstrap-reps 20000 --seed 20260723"
        )
    print("Byte-for-byte analysis equality: not expected (documented timing fix)")
    print(f"Comparison: {comparison_path}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Verify the frozen pre-run inventory and the exported IBM evidence."""

from __future__ import annotations

import hashlib
import hmac
import json
from pathlib import Path
from typing import Any


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
FROZEN = PACKAGE_ROOT / "archive" / "original-frozen"
PRE_RUN = PACKAGE_ROOT / "evidence" / "frozen-pre-run"
SUBMISSION = PACKAGE_ROOT / "evidence" / "submission"
IBM_RUN = PACKAGE_ROOT / "evidence" / "ibm-run"


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def canonical_json(value: Any) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def path_for_frozen_name(name: str) -> Path:
    if name in {
        "HQC_IBM_PARAMETRIC_DRY_RUN_PLAN.json",
        "HQC_IBM_PARAMETRIC_TEMPLATE.qpy",
    }:
        return PRE_RUN / name
    return FROZEN / name


def verify_pre_run_manifest() -> int:
    manifest_path = PRE_RUN / "HQC_PRE_RUN_MANIFEST.sha256"
    checked = 0
    for line_number, raw in enumerate(
        manifest_path.read_text(encoding="utf-8").splitlines(),
        start=1,
    ):
        line = raw.strip()
        if not line:
            continue
        expected, name = line.split(maxsplit=1)
        path = path_for_frozen_name(name.lstrip("*"))
        if not path.is_file():
            raise FileNotFoundError(f"Missing pre-run file: {path}")
        actual = sha256_file(path)
        if not hmac.compare_digest(expected, actual):
            raise AssertionError(
                f"Pre-run hash mismatch on line {line_number}: {path.name}"
            )
        checked += 1
    if checked != 17:
        raise AssertionError(f"Expected 17 pre-run entries, got {checked}")
    return checked


def verify_submission_contract() -> dict[str, str]:
    manifest_path = SUBMISSION / "HQC_SUBMISSION_MANIFEST.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    internal = manifest["manifest_sha256"]
    core = {key: value for key, value in manifest.items() if key != "manifest_sha256"}
    computed = hashlib.sha256(canonical_json(core)).hexdigest()
    if not hmac.compare_digest(internal, computed):
        raise AssertionError("Submission manifest internal digest mismatch")

    artifacts = manifest["artifacts"]
    plan = PRE_RUN / "HQC_IBM_PARAMETRIC_DRY_RUN_PLAN.json"
    qpy = PRE_RUN / "HQC_IBM_PARAMETRIC_TEMPLATE.qpy"
    if artifacts["plan_sha256"] != sha256_file(plan):
        raise AssertionError("Submission manifest plan hash mismatch")
    if artifacts["qpy_sha256"] != sha256_file(qpy):
        raise AssertionError("Submission manifest QPY hash mismatch")

    return {
        "manifest_sha256": internal,
        "plan_sha256": artifacts["plan_sha256"],
        "qpy_sha256": artifacts["qpy_sha256"],
    }


def verify_export() -> dict[str, Any]:
    receipts = list(IBM_RUN.glob("HQC_IBM_EXPORT_RECEIPT_*.json"))
    if len(receipts) != 1:
        raise AssertionError("Expected exactly one IBM export receipt")
    receipt = json.loads(receipts[0].read_text(encoding="utf-8"))
    raw = IBM_RUN / receipt["raw_counts"]
    primitive = IBM_RUN / receipt["primitive_result"]
    if receipt["raw_counts_sha256"] != sha256_file(raw):
        raise AssertionError("IBM raw-counts digest mismatch")
    if receipt["primitive_result_sha256"] != sha256_file(primitive):
        raise AssertionError("IBM PrimitiveResult digest mismatch")
    payload = json.loads(raw.read_text(encoding="utf-8"))
    if len(payload["records"]) != 448:
        raise AssertionError("IBM raw export does not contain 448 records")
    return {
        "job_id": receipt["job_id"],
        "records": receipt["records"],
        "raw_counts_sha256": receipt["raw_counts_sha256"],
        "primitive_result_sha256": receipt["primitive_result_sha256"],
    }


def main() -> None:
    pre_run_files = verify_pre_run_manifest()
    contract = verify_submission_contract()
    export = verify_export()
    output = PACKAGE_ROOT / "work" / "archive-verification.json"
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(
        json.dumps(
            {
                "pre_run_files_verified": pre_run_files,
                "submission": contract,
                "ibm_export": export,
                "network_used": False,
                "qpu_jobs_submitted": 0,
            },
            ensure_ascii=False,
            indent=2,
        )
        + "\n",
        encoding="utf-8",
    )
    print("FROZEN ARCHIVE VERIFICATION PASSED")
    print("IBM/network calls: 0; QPU jobs: 0")
    print(f"Pre-run files: {pre_run_files}/17")
    print(f"IBM records: {export['records']}/448")
    print(f"Plan SHA-256: {contract['plan_sha256']}")
    print(f"QPY SHA-256: {contract['qpy_sha256']}")
    print(f"Output: {output.relative_to(PACKAGE_ROOT)}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Run the exact 56 frozen unit tests with network access blocked."""

from __future__ import annotations

import os
import socket
import sys
import unittest
from pathlib import Path


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
TESTS = PACKAGE_ROOT / "tests"
sys.path[:0] = [
    str(TESTS),
    str(PACKAGE_ROOT / "dangerous"),
    str(PACKAGE_ROOT / "portable-source"),
]
os.environ["HQC_OFFLINE"] = "1"
os.environ["HQC_WORKDIR"] = str(
    PACKAGE_ROOT / "evidence" / "frozen-pre-run"
)


def blocked_network(*_args, **_kwargs):
    raise RuntimeError("Network access is disabled by scripts/run_tests.py")


def main() -> None:
    socket.create_connection = blocked_network  # type: ignore[assignment]
    socket.socket.connect = blocked_network  # type: ignore[method-assign]

    suite = unittest.defaultTestLoader.loadTestsFromNames(
        [
            "test_hqc_preparation",
            "test_hqc_runtime_job",
            "test_hqc_analysis",
        ]
    )
    count = suite.countTestCases()
    if count != 56:
        raise AssertionError(f"Expected exactly 56 tests, discovered {count}")
    result = unittest.TextTestRunner(verbosity=2).run(suite)
    if not result.wasSuccessful():
        raise SystemExit(1)
    print("EXACT TEST INVENTORY PASSED: 56/56")
    print("Network was blocked; QPU jobs submitted: 0")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Fail when the public/portable release contains credential material."""

from __future__ import annotations

import re
from pathlib import Path


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
PORTABLE_AREAS = (
    "portable-source",
    "dangerous",
    "online",
    "scripts",
    "research",
    "tests",
)
TEXT_SUFFIXES = {
    "",
    ".md",
    ".txt",
    ".py",
    ".json",
    ".toml",
    ".yaml",
    ".yml",
    ".cff",
}
SKIP_PARTS = {
    ".git",
    ".venv",
    "__pycache__",
    "site-packages",
    "work",
}
FORBIDDEN_GLOBAL = {
    "IBM CRN value": re.compile(r"crn:v1:", re.IGNORECASE),
    "private key": re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
}
HARD_CODED_SECRET = re.compile(
    r"""(?ix)
    \b(?:api[_-]?key|token|password|secret)\b
    \s*[:=]\s*
    ["']
    (?!<|your-|example|placeholder)
    [A-Za-z0-9_./+=-]{24,}
    ["']
    """
)


def text_files(root: Path):
    for path in root.rglob("*"):
        relative_parts = path.relative_to(root).parts
        if (
            path.is_file()
            and path.suffix.lower() in TEXT_SUFFIXES
            and not SKIP_PARTS.intersection(relative_parts)
            and path.resolve() != Path(__file__).resolve()
        ):
            yield path


def main() -> None:
    failures: list[str] = []
    for area in PORTABLE_AREAS:
        for path in text_files(PACKAGE_ROOT / area):
            text = path.read_text(encoding="utf-8", errors="replace")
            if "/root/" in text:
                failures.append(
                    f"internal absolute path in {path.relative_to(PACKAGE_ROOT)}"
                )
            if HARD_CODED_SECRET.search(text):
                failures.append(
                    f"possible hard-coded secret in {path.relative_to(PACKAGE_ROOT)}"
                )

    for path in text_files(PACKAGE_ROOT):
        text = path.read_text(encoding="utf-8", errors="replace")
        for label, pattern in FORBIDDEN_GLOBAL.items():
            if pattern.search(text):
                failures.append(
                    f"{label} in {path.relative_to(PACKAGE_ROOT)}"
                )

    if failures:
        raise SystemExit("PUBLIC RELEASE SCAN FAILED\n- " + "\n- ".join(failures))

    print("PUBLIC RELEASE SCAN PASSED")
    print("No CRN values, private keys, hard-coded secrets, or portable /root paths.")
    print(
        "Historical archive files may mention credential filenames only; "
        "they contain no credential values."
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Safety and calibration-quality tests for local/metadata-only preparation."""

from __future__ import annotations

import ast
import math
import unittest
from pathlib import Path
from types import SimpleNamespace

import hqc_prepare_sorkin_experiment as preparation


class FakeInstructionMap:
    def __init__(self, entries: dict[tuple[int, ...], object]) -> None:
        self.entries = entries

    def __iter__(self):
        return iter(self.entries)

    def get(self, qargs: tuple[int, ...]):
        return self.entries.get(qargs)


class FakeTarget:
    def __init__(self, cz: dict, measure: dict) -> None:
        self.operation_names = {"cz", "measure"}
        self.maps = {
            "cz": FakeInstructionMap(cz),
            "measure": FakeInstructionMap(measure),
        }

    def __getitem__(self, name: str):
        return self.maps[name]


class FakeBackend:
    name = "fake_backend"

    def __init__(self, target: FakeTarget) -> None:
        self.target = target


def prop(error: float | None, duration: float = 1e-7):
    return SimpleNamespace(error=error, duration=duration)


class PreparationTests(unittest.TestCase):
    def test_missing_calibration_error_is_never_treated_as_zero(self) -> None:
        target = FakeTarget(
            cz={(0, 1): prop(None), (2, 1): prop(0.002)},
            measure={
                (0,): prop(0.01),
                (1,): prop(0.01),
                (2,): prop(0.01),
            },
        )
        with self.assertRaises(RuntimeError):
            preparation.best_line_triplet(FakeBackend(target))

    def test_nonfinite_calibration_is_rejected(self) -> None:
        target = FakeTarget(
            cz={(0, 1): prop(math.nan), (2, 1): prop(0.002)},
            measure={
                (0,): prop(0.01),
                (1,): prop(0.01),
                (2,): prop(0.01),
            },
        )
        with self.assertRaises(RuntimeError):
            preparation.best_line_triplet(FakeBackend(target))

    def test_finite_triplet_is_selected_with_real_errors(self) -> None:
        target = FakeTarget(
            cz={(0, 1): prop(0.001), (2, 1): prop(0.002)},
            measure={
                (0,): prop(0.010),
                (1,): prop(0.011),
                (2,): prop(0.012),
            },
        )
        selected = preparation.best_line_triplet(FakeBackend(target))
        self.assertEqual(selected["layout"], [0, 2, 1])
        self.assertAlmostEqual(selected["cz_error_sum"], 0.003)
        self.assertAlmostEqual(selected["readout_error_sum"], 0.033)

    def test_preparation_has_no_runtime_submission_call(self) -> None:
        source = Path(preparation.__file__).read_text(encoding="utf-8")
        tree = ast.parse(source)
        imported_names = {
            alias.name
            for node in ast.walk(tree)
            if isinstance(node, (ast.Import, ast.ImportFrom))
            for alias in node.names
        }
        self.assertFalse(
            imported_names
            & {
                "SamplerV2",
                "EstimatorV2",
                "Session",
                "Batch",
            }
        )
        runtime_run_calls = [
            node
            for node in ast.walk(tree)
            if isinstance(node, ast.Call)
            and isinstance(node.func, ast.Attribute)
            and node.func.attr == "run"
            and isinstance(node.func.value, ast.Name)
            and node.func.value.id in {"sampler", "estimator", "job"}
        ]
        self.assertEqual(runtime_run_calls, [])


if __name__ == "__main__":
    unittest.main(verbosity=2)

#!/usr/bin/env python3
"""Local safety and structural tests for the guarded IBM Runtime runner."""

from __future__ import annotations

import ast
import copy
import json
import tempfile
import unittest
from datetime import UTC, datetime, timedelta
from pathlib import Path
from unittest.mock import patch

import numpy as np
from qiskit.primitives.containers import BitArray

import hqc_runtime_job as runner


def ordered_pub_metadata() -> list[dict[str, object]]:
    base = datetime(2026, 7, 23, 16, 0, 0, tzinfo=UTC)
    rows: list[dict[str, object]] = []
    for pub_index in range(88):
        start = base + timedelta(seconds=2 * pub_index)
        stop = start + timedelta(seconds=1)
        if pub_index < 32:
            kind = "readout_calibration"
            position: str | None = "before"
        elif pub_index < 56:
            kind = "experiment"
            position = None
        else:
            kind = "readout_calibration"
            position = "after"
        rows.append(
            {
                "pub_index": pub_index,
                "kind": kind,
                "calibration_position": position,
                "executed_at_utc": start.isoformat(),
                "execution_start_utc": start.isoformat(),
                "execution_stop_utc": stop.isoformat(),
            }
        )
    return rows


class RuntimeRunnerTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.bundle = runner.structural_bundle()

    def test_exact_frozen_shape(self) -> None:
        self.assertEqual(len(self.bundle["rows"]), 384)
        self.assertEqual(len(self.bundle["calibrations"]), 64)
        self.assertEqual(len(self.bundle["specs"]), 88)
        self.assertEqual(len(self.bundle["pubs"]), 88)
        self.assertEqual(
            sum(spec.effective_configurations for spec in self.bundle["specs"]),
            448,
        )
        experiment_indices = [
            int(row["binding_index"])
            for spec in self.bundle["specs"]
            if spec.kind == "experiment"
            for row in spec.binding_rows
        ]
        self.assertEqual(experiment_indices, list(range(384)))

    def test_calibrations_wrap_experiment(self) -> None:
        specs = self.bundle["specs"]
        self.assertTrue(all(spec.kind == "readout_calibration" for spec in specs[:32]))
        self.assertTrue(all(spec.kind == "experiment" for spec in specs[32:56]))
        self.assertTrue(all(spec.kind == "readout_calibration" for spec in specs[56:]))
        self.assertEqual(
            [spec.calibration["position"] for spec in specs[:32]],
            ["before"] * 32,
        )
        self.assertEqual(
            [spec.calibration["position"] for spec in specs[56:]],
            ["after"] * 32,
        )

    def test_all_qpy_circuits_match_the_exact_frozen_layout(self) -> None:
        expected = self.bundle["plan"]["selected_snapshot"][
            "layout_logical_q0_q1_flag"
        ]
        self.assertEqual(expected, [89, 91, 90])
        for circuit in self.bundle["circuits"]:
            self.assertEqual(
                circuit.layout.final_index_layout(filter_ancillas=True),
                expected,
            )
            used = {
                circuit.find_bit(qubit).index
                for instruction in circuit.data
                if instruction.operation.name != "barrier"
                for qubit in instruction.qubits
            }
            self.assertEqual(used, set(expected))

        tampered_plan = copy.deepcopy(self.bundle["plan"])
        tampered_plan["selected_snapshot"]["layout_logical_q0_q1_flag"] = [
            90,
            91,
            89,
        ]
        with self.assertRaises(ValueError):
            runner.validate_qpy(
                tampered_plan,
                self.bundle["circuits"],
            )

    def test_target_validation_rejects_reverse_only_cz_edges(self) -> None:
        class FakeTarget:
            operation_names = {"cz"}

            def __init__(self, edges: set[tuple[int, int]]) -> None:
                self.edges = edges

            def __getitem__(self, name: str) -> dict[tuple[int, int], object]:
                self.assert_name = name
                return {edge: object() for edge in self.edges}

        layout = [89, 91, 90]
        exact = FakeTarget({(89, 90), (91, 90)})
        with patch.object(runner, "validate_isa_circuits") as validate:
            runner.validate_frozen_circuits_against_target(
                self.bundle["circuits"],
                exact,
                layout,
            )
            validate.assert_called_once_with(
                self.bundle["circuits"],
                exact,
            )

        reverse_only = FakeTarget({(90, 89), (90, 91)})
        with (
            patch.object(runner, "validate_isa_circuits"),
            self.assertRaises(ValueError),
        ):
            runner.validate_frozen_circuits_against_target(
                self.bundle["circuits"],
                reverse_only,
                layout,
            )

    def test_preflight_and_submit_share_target_validation_gate(self) -> None:
        source = Path(runner.__file__).read_text(encoding="utf-8")
        tree = ast.parse(source)
        functions = {
            node.name: node
            for node in tree.body
            if isinstance(node, ast.FunctionDef)
        }
        snapshot_calls = [
            node
            for node in ast.walk(functions["current_backend_snapshot"])
            if isinstance(node, ast.Call)
            and isinstance(node.func, ast.Name)
            and node.func.id == "validate_frozen_circuits_against_target"
        ]
        self.assertEqual(len(snapshot_calls), 1)
        for function_name in ("metadata_preflight", "submit_one_job"):
            calls = [
                node
                for node in ast.walk(functions[function_name])
                if isinstance(node, ast.Call)
                and isinstance(node.func, ast.Name)
                and node.func.id == "current_backend_snapshot"
            ]
            self.assertEqual(
                len(calls),
                1,
                msg=f"{function_name} must pass through target validation",
            )

    def test_parameter_mapping_uses_names_not_incidental_order(self) -> None:
        pairs = [
            (spec, pub)
            for spec, pub in zip(
                self.bundle["specs"],
                self.bundle["pubs"],
                strict=True,
            )
            if spec.kind == "experiment"
        ]
        self.assertEqual(len(pairs), 24)
        for spec, pub in pairs:
            actual = pub.parameter_values.as_array()
            parameter_names = [
                str(parameter) for parameter in pub.circuit.parameters
            ]
            expected = np.asarray(
                [
                    [
                        runner.row_parameter_map(row)[name]
                        for name in parameter_names
                    ]
                    for row in spec.binding_rows
                ],
                dtype=float,
            )
            np.testing.assert_allclose(actual, expected, rtol=0.0, atol=0.0)
            self.assertEqual(
                parameter_names,
                [
                    "phase[0]",
                    "phase[1]",
                    "theta[0]",
                    "theta[1]",
                    "theta[2]",
                    "theta[3]",
                ],
            )

    def test_twirling_and_dd_are_explicitly_disabled(self) -> None:
        options = runner.runtime_options_payload()
        self.assertIs(options["twirling"]["enable_gates"], False)
        self.assertIs(options["twirling"]["enable_measure"], False)
        self.assertNotIn("num_randomizations", options["twirling"])
        self.assertIs(options["dynamical_decoupling"]["enable"], False)
        self.assertEqual(options["max_execution_time"], 480)

    def test_bitarray_counts_contract(self) -> None:
        bit_array = BitArray.from_samples(["000"] * 700 + ["111"] * 324, num_bits=3)
        counts = runner.counts_for(bit_array)
        self.assertEqual(counts, {"000": 700, "111": 324})

    def test_execution_time_is_taken_only_from_matching_spans(self) -> None:
        class FakeSpan:
            def __init__(
                self,
                start: datetime,
                stop: datetime,
                pub_idxs: list[int],
            ) -> None:
                self.start = start
                self.stop = stop
                self.pub_idxs = pub_idxs

        later = datetime(2026, 7, 23, 16, 0, 2, tzinfo=UTC)
        earlier = datetime(2026, 7, 23, 16, 0, 1, tzinfo=UTC)
        spans = [
            FakeSpan(
                later,
                later + timedelta(seconds=2),
                [10],
            ),
            FakeSpan(
                earlier,
                earlier + timedelta(seconds=2),
                [10, 11],
            ),
            FakeSpan(
                datetime(2026, 7, 23, 15, 0, tzinfo=UTC),
                datetime(2026, 7, 23, 15, 0, 1, tzinfo=UTC),
                [9],
            ),
        ]
        start, stop, source = runner.execution_window_for_pub(spans, 10)
        self.assertEqual(start, earlier.isoformat())
        self.assertEqual(stop, (later + timedelta(seconds=2)).isoformat())
        self.assertEqual(source, "runtime_execution_spans")
        timestamp, source = runner.execution_time_for_pub(spans, 10)
        self.assertEqual(timestamp, earlier.isoformat())
        self.assertEqual(source, "runtime_execution_spans")
        self.assertEqual(
            runner.execution_time_for_pub(spans, 12),
            (None, "unavailable"),
        )
        self.assertEqual(
            runner.execution_window_for_pub(spans, 12),
            (None, None, "unavailable"),
        )

    def test_schedule_order_uses_explicit_unknown(self) -> None:
        complete = ordered_pub_metadata()
        self.assertIs(runner.schedule_order_status(complete), True)

        missing = copy.deepcopy(complete)
        missing[40]["execution_stop_utc"] = None
        self.assertEqual(runner.schedule_order_status(missing), "unknown")

        overlap = copy.deepcopy(complete)
        overlap[1]["execution_start_utc"] = (
            datetime.fromisoformat(str(overlap[0]["execution_stop_utc"]))
            - timedelta(milliseconds=500)
        ).isoformat()
        overlap[1]["executed_at_utc"] = overlap[1]["execution_start_utc"]
        self.assertEqual(runner.schedule_order_status(overlap), "unknown")

        equal_start = copy.deepcopy(complete)
        equal_start[1]["execution_start_utc"] = equal_start[0][
            "execution_start_utc"
        ]
        equal_start[1]["executed_at_utc"] = equal_start[1][
            "execution_start_utc"
        ]
        self.assertEqual(runner.schedule_order_status(equal_start), "unknown")

        reordered = copy.deepcopy(complete)
        reordered[1]["execution_start_utc"] = (
            datetime.fromisoformat(str(reordered[0]["execution_start_utc"]))
            - timedelta(seconds=1)
        ).isoformat()
        reordered[1]["executed_at_utc"] = reordered[1][
            "execution_start_utc"
        ]
        self.assertIs(runner.schedule_order_status(reordered), False)

        broken_bracket = copy.deepcopy(complete)
        broken_bracket[32]["kind"] = "readout_calibration"
        broken_bracket[32]["calibration_position"] = "before"
        self.assertIs(
            runner.schedule_order_status(broken_bracket),
            False,
        )

    def test_missing_calibration_values_are_never_zero_filled(self) -> None:
        for value in (None, float("nan"), float("inf"), -float("inf")):
            with self.subTest(value=value):
                with self.assertRaises(ValueError):
                    runner.required_finite_number(value, "test")
        self.assertEqual(
            runner.required_finite_number(
                0.0,
                "valid zero error",
                minimum=0.0,
                maximum=1.0,
            ),
            0.0,
        )
        with self.assertRaises(ValueError):
            runner.required_finite_number(
                0.0,
                "invalid zero duration",
                strictly_positive=True,
            )

    def test_manifest_digest_covers_backend_snapshot(self) -> None:
        snapshot = {
            "backend": "fake_backend",
            "layout_logical_q0_q1_flag": [1, 2, 3],
            "circuit_duration_seconds": 0.000001,
            "default_rep_delay_seconds": 0.001,
        }
        manifest = runner.build_manifest(self.bundle, snapshot)
        self.assertEqual(
            runner.verify_manifest_digest(manifest),
            manifest["manifest_sha256"],
        )
        tampered = copy.deepcopy(manifest)
        tampered["backend_snapshot"]["backend"] = "tampered_backend"
        with self.assertRaises(PermissionError):
            runner.verify_manifest_digest(tampered)

    def test_capsule_and_manifest_must_match_exactly(self) -> None:
        snapshot = {
            "backend": "fake_backend",
            "layout_logical_q0_q1_flag": [1, 2, 3],
            "circuit_duration_seconds": 0.000001,
            "default_rep_delay_seconds": 0.001,
        }
        manifest = runner.build_manifest(self.bundle, snapshot)
        capsule = runner.build_capsule(manifest)
        runner.verify_capsule_manifest_consistency(capsule, manifest)
        tampered = copy.deepcopy(capsule)
        tampered["backend_snapshot"]["backend"] = "another_backend"
        with self.assertRaises(PermissionError):
            runner.verify_capsule_manifest_consistency(tampered, manifest)

    def test_current_artifacts_include_preregistration_and_research_draft(
        self,
    ) -> None:
        manifest = runner.build_manifest(self.bundle, None)
        runner.verify_current_artifacts(self.bundle, manifest)
        artifacts = manifest["artifacts"]
        self.assertEqual(
            artifacts["design_sha256"],
            runner.sha256_file(runner.DESIGN_FILE),
        )
        self.assertEqual(
            artifacts["research_draft_sha256"],
            runner.sha256_file(runner.RESEARCH_DRAFT_FILE),
        )
        self.assertEqual(
            artifacts["requirements_sha256"],
            runner.sha256_file(runner.REQUIREMENTS_FILE),
        )
        self.assertEqual(
            manifest["versions"]["scipy"],
            runner.scipy.__version__,
        )
        tampered = copy.deepcopy(manifest)
        tampered["artifacts"]["plan_sha256"] = "0" * 64
        with self.assertRaises(PermissionError):
            runner.verify_current_artifacts(self.bundle, tampered)

    def test_approval_relevant_snapshot_and_estimate_are_frozen(self) -> None:
        frozen = {
            "backend": "fake_backend",
            "operational": True,
            "pending_jobs": 7,
            "layout_logical_q0_q1_flag": [89, 91, 90],
            "calibration_timestamp_utc": "2026-07-23T16:00:00+00:00",
            "calibration_age_seconds": 60.0,
            "cz": [
                {
                    "qargs": [89, 90],
                    "error": 0.002,
                    "duration_seconds": 0.00000008,
                },
                {
                    "qargs": [91, 90],
                    "error": 0.003,
                    "duration_seconds": 0.00000008,
                },
            ],
            "readout": [
                {
                    "qubit": qubit,
                    "error": 0.01,
                    "duration_seconds": 0.000001,
                }
                for qubit in (89, 91, 90)
            ],
            "t1_seconds": [0.0002, 0.00021, 0.00022],
            "t2_seconds": [0.00015, 0.00016, 0.00017],
            "circuit_duration_seconds": 0.000004,
            "template_depth": 26,
            "template_operations": {"cz": 4, "measure": 3},
            "default_rep_delay_seconds": 0.00025,
            "quality_gates": {
                "operational": True,
                "calibration_younger_than_4h": True,
                "max_cz_error_below_0_015": True,
                "max_readout_error_below_0_03": True,
                "no_swap": True,
                "no_more_than_6_entanglers": True,
                "circuit_below_10pct_min_t1_t2": True,
            },
        }
        capsule = {"usage_estimate": runner.estimate_usage(frozen)}

        queue_and_age_only = copy.deepcopy(frozen)
        queue_and_age_only["pending_jobs"] = 99
        queue_and_age_only["calibration_age_seconds"] = 7_200.0
        queue_and_age_only["quality_gates"][
            "calibration_younger_than_4h"
        ] = False
        runner.verify_approved_snapshot_is_current(
            queue_and_age_only,
            frozen,
            capsule,
        )

        changed_calibration = copy.deepcopy(queue_and_age_only)
        changed_calibration["cz"][0]["error"] = 0.004
        with self.assertRaises(PermissionError):
            runner.verify_approved_snapshot_is_current(
                changed_calibration,
                frozen,
                capsule,
            )

        changed_estimate = copy.deepcopy(capsule)
        changed_estimate["usage_estimate"][
            "with_50pct_margin_seconds"
        ] += 1.0
        with self.assertRaises(PermissionError):
            runner.verify_approved_snapshot_is_current(
                queue_and_age_only,
                frozen,
                changed_estimate,
            )

    def test_submit_rechecks_approved_snapshot_before_intent(self) -> None:
        source = Path(runner.__file__).read_text(encoding="utf-8")
        tree = ast.parse(source)
        submit_function = next(
            node
            for node in tree.body
            if isinstance(node, ast.FunctionDef)
            and node.name == "submit_one_job"
        )
        calls: dict[str, list[int]] = {
            "current_backend_snapshot": [],
            "verify_approved_snapshot_is_current": [],
            "intent_write": [],
        }
        for node in ast.walk(submit_function):
            if not isinstance(node, ast.Call):
                continue
            if (
                isinstance(node.func, ast.Name)
                and node.func.id in (
                    "current_backend_snapshot",
                    "verify_approved_snapshot_is_current",
                )
            ):
                calls[node.func.id].append(node.lineno)
            if (
                isinstance(node.func, ast.Name)
                and node.func.id == "write_private_json"
                and node.args
                and isinstance(node.args[0], ast.Name)
                and node.args[0].id == "INTENT_FILE"
                and any(
                    keyword.arg == "exclusive"
                    and isinstance(keyword.value, ast.Constant)
                    and keyword.value.value is True
                    for keyword in node.keywords
                )
            ):
                calls["intent_write"].append(node.lineno)
        self.assertEqual(len(calls["current_backend_snapshot"]), 1)
        self.assertEqual(
            len(calls["verify_approved_snapshot_is_current"]),
            1,
        )
        self.assertEqual(len(calls["intent_write"]), 1)
        self.assertLess(
            calls["current_backend_snapshot"][0],
            calls["verify_approved_snapshot_is_current"][0],
        )
        self.assertLess(
            calls["verify_approved_snapshot_is_current"][0],
            calls["intent_write"][0],
        )

    def test_raw_schema_is_v2_for_execution_windows(self) -> None:
        self.assertEqual(runner.RAW_SCHEMA, "hqc-ibm-raw-counts-v2")

    def test_preflight_refuses_to_replace_post_submission_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            intent = root / "intent.json"
            receipt = root / "receipt.json"
            intent.write_text("{}\n", encoding="utf-8")
            with (
                patch.object(runner, "INTENT_FILE", intent),
                patch.object(runner, "RECEIPT_FILE", receipt),
                patch.object(
                    runner,
                    "structural_bundle",
                    side_effect=AssertionError("must fail before artifact reads"),
                ),
                patch.object(
                    runner,
                    "create_service",
                    side_effect=AssertionError("must not contact IBM"),
                ),
            ):
                with self.assertRaises(PermissionError):
                    runner.metadata_preflight()

    def test_exclusive_lock_cannot_be_recreated(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            lock = Path(directory) / "intent.json"
            runner.write_private_json(lock, {"state": "first"}, exclusive=True)
            with self.assertRaises(FileExistsError):
                runner.write_private_json(
                    lock,
                    {"state": "second"},
                    exclusive=True,
                )
            self.assertEqual(
                json.loads(lock.read_text(encoding="utf-8"))["state"],
                "first",
            )

    def test_job_id_is_checkpointed_before_receipt(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            intent_path = Path(directory) / "intent.json"
            runner.write_private_json(
                intent_path,
                {"state": "submission_intent_locked"},
            )
            with patch.object(runner, "INTENT_FILE", intent_path):
                checkpoint = runner.checkpoint_returned_job(
                    {"state": "submission_intent_locked"},
                    "job-safe-123",
                )
            stored = json.loads(intent_path.read_text(encoding="utf-8"))
            self.assertEqual(stored["job_id"], "job-safe-123")
            self.assertEqual(stored["state"], "job_returned_receipt_pending")
            self.assertEqual(checkpoint["job_id"], "job-safe-123")

        source = Path(runner.__file__).read_text(encoding="utf-8")
        tree = ast.parse(source)
        submit_function = next(
            node
            for node in tree.body
            if isinstance(node, ast.FunctionDef)
            and node.name == "submit_one_job"
        )
        checkpoint_calls: list[int] = []
        receipt_writes: list[int] = []
        for node in ast.walk(submit_function):
            if not isinstance(node, ast.Call):
                continue
            if (
                isinstance(node.func, ast.Name)
                and node.func.id == "checkpoint_returned_job"
            ):
                checkpoint_calls.append(node.lineno)
            if (
                isinstance(node.func, ast.Name)
                and node.func.id == "write_private_json"
                and node.args
                and isinstance(node.args[0], ast.Name)
                and node.args[0].id == "RECEIPT_FILE"
            ):
                receipt_writes.append(node.lineno)
        self.assertEqual(len(checkpoint_calls), 1)
        self.assertEqual(len(receipt_writes), 1)
        self.assertLess(checkpoint_calls[0], receipt_writes[0])

    def test_result_export_is_an_atomic_directory_bundle(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            payload = {
                "manifest_sha256": "a" * 64,
                "records": [{"kind": "test"}],
            }
            with patch.object(runner, "ROOT", root):
                raw_path = runner.write_result_export_bundle(
                    "job-safe-123",
                    payload,
                    {"runtime": "result"},
                )
                self.assertTrue(raw_path.is_file())
                export_dir = raw_path.parent
                self.assertTrue(
                    (
                        export_dir
                        / "HQC_IBM_PRIMITIVE_RESULT_job-safe-123.json"
                    ).is_file()
                )
                verified = runner.completed_export_path(
                    "job-safe-123",
                    "a" * 64,
                )
                self.assertEqual(verified, raw_path)
                self.assertEqual(export_dir.stat().st_mode & 0o777, 0o700)
                for path in export_dir.iterdir():
                    self.assertEqual(path.stat().st_mode & 0o777, 0o600)

    def test_failed_export_never_publishes_a_partial_bundle(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            payload = {
                "manifest_sha256": "b" * 64,
                "records": [{"kind": "test"}],
            }
            with (
                patch.object(runner, "ROOT", root),
                patch.object(
                    runner,
                    "write_private_runtime_json",
                    side_effect=OSError("synthetic I/O failure"),
                ),
            ):
                with self.assertRaises(OSError):
                    runner.write_result_export_bundle(
                        "job-safe-456",
                        payload,
                        {"runtime": "result"},
                    )
            self.assertFalse(
                (root / "HQC_IBM_EXPORT_job-safe-456").exists()
            )
            self.assertEqual(list(root.iterdir()), [])

    def test_approval_phrase_is_mandatory_before_files_are_read(self) -> None:
        with self.assertRaises(PermissionError):
            runner.verify_capsule(self.bundle, "0" * 64, "wrong phrase")

    def test_only_one_runtime_submission_callsite_exists(self) -> None:
        source = Path(runner.__file__).read_text(encoding="utf-8")
        tree = ast.parse(source)
        run_calls = []
        for node in ast.walk(tree):
            if (
                isinstance(node, ast.Call)
                and isinstance(node.func, ast.Attribute)
                and node.func.attr == "run"
                and isinstance(node.func.value, ast.Name)
                and node.func.value.id == "sampler"
            ):
                run_calls.append(node.lineno)
        self.assertEqual(len(run_calls), 1)
        submit_function = next(
            node
            for node in tree.body
            if isinstance(node, ast.FunctionDef)
            and node.name == "submit_one_job"
        )
        start = submit_function.lineno
        end = submit_function.end_lineno or start
        self.assertTrue(start <= run_calls[0] <= end)

    def test_collect_has_no_submission_call(self) -> None:
        source = Path(runner.__file__).read_text(encoding="utf-8")
        tree = ast.parse(source)
        collect_function = next(
            node
            for node in tree.body
            if isinstance(node, ast.FunctionDef)
            and node.name == "collect_existing_job"
        )
        run_calls = [
            node
            for node in ast.walk(collect_function)
            if isinstance(node, ast.Call)
            and isinstance(node.func, ast.Attribute)
            and node.func.attr == "run"
        ]
        self.assertEqual(run_calls, [])

    def test_local_check_does_not_create_submission_files(self) -> None:
        before = {
            path: path.exists()
            for path in (runner.INTENT_FILE, runner.RECEIPT_FILE)
        }
        runner.local_check()
        after = {
            path: path.exists()
            for path in (runner.INTENT_FILE, runner.RECEIPT_FILE)
        }
        self.assertEqual(before, after)


if __name__ == "__main__":
    unittest.main()

#!/usr/bin/env python3
"""Regression and malformed-input tests for the local HQC analyzer."""

from __future__ import annotations

import copy
import hashlib
import json
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch

import numpy as np

import hqc_analyze_counts as hqc


class HQCAnalysisTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        plan = json.loads(hqc.PLAN_FILE.read_text(encoding="utf-8"))
        manifest_core = {
            "schema": "hqc-local-unit-test-contract-v1",
            "artifacts": {
                "plan_sha256": hqc.sha256_file(hqc.PLAN_FILE),
                "qpy_sha256": hqc.sha256_file(hqc.QPY_FILE),
                "analyzer_sha256": hqc.sha256_file(
                    Path(hqc.__file__).resolve()
                ),
            },
            "experiment": {
                "shots_per_configuration": 1024,
                "parameter_bindings": 384,
                "readout_calibrations": 64,
                "effective_configurations": 448,
            },
            "versions": {"fixture": "local-unit-test"},
            "backend_snapshot": plan["selected_snapshot"],
        }
        manifest = dict(manifest_core)
        manifest["manifest_sha256"] = hashlib.sha256(
            hqc.canonical_json_bytes(manifest_core)
        ).hexdigest()
        cls.contract = {
            "plan": plan,
            "plan_sha256": manifest["artifacts"]["plan_sha256"],
            "qpy_sha256": manifest["artifacts"]["qpy_sha256"],
            "manifest_sha256": manifest["manifest_sha256"],
            "manifest_file_sha256": manifest["manifest_sha256"],
            "manifest": manifest,
            "contract_mode": "local-unit-test-fixture",
        }
        cls.payload = hqc.ideal_synthetic_payload(contract=cls.contract)

    def analyze(self, payload: dict | None = None, *, reps: int = 32) -> dict:
        return hqc.analyze(
            payload or copy.deepcopy(self.payload),
            bootstrap_reps=reps,
            seed=123456,
            contract=self.contract,
        )

    def test_frozen_inventory_and_ideal_null(self) -> None:
        with patch.object(
            hqc,
            "_bootstrap_stability_analysis",
            return_value={
                "criterion": "unit-test stable override",
                "variants": {},
                "inconsistency_flags": [],
            },
        ):
            result = self.analyze(reps=64)
        self.assertEqual(result["validation"]["records"], 448)
        self.assertEqual(result["validation"]["experiment_records"], 384)
        self.assertEqual(result["validation"]["readout_calibration_records"], 64)
        self.assertEqual(result["validation"]["complete_blocks"], 24)
        self.assertEqual(result["classification"]["label"], "null_validation")
        self.assertEqual(
            result["variant_agreement"]["inconsistency_flags"], []
        )
        for variant in hqc.VARIANTS:
            self.assertTrue(result["statistics"][variant]["available"])
            for name in hqc.CONTRAST_NAMES:
                self.assertAlmostEqual(
                    result["statistics"][variant]["primary"][name]["estimate"],
                    0.0,
                    places=12,
                )
                self.assertFalse(
                    result["statistics"][variant]["primary"][name][
                        "holm_reject_0_05"
                    ]
                )

    def test_bootstrap_stability_independently_blocks_null(self) -> None:
        result = self.analyze(reps=64)
        stability_flags = [
            flag
            for flag in result["classification"]["inconclusive_flags"]
            if flag.startswith("bootstrap_unstable:")
        ]
        self.assertTrue(stability_flags)
        self.assertEqual(result["classification"]["label"], "inconclusive")
        self.assertFalse(
            any(
                item["inconclusive"]
                for item in result["drift"]["overdispersion"].values()
            )
        )

    def test_deterministic_bootstrap(self) -> None:
        first = self.analyze(reps=24)
        second = self.analyze(reps=24)
        self.assertEqual(
            json.dumps(first, sort_keys=True, allow_nan=False),
            json.dumps(second, sort_keys=True, allow_nan=False),
        )

    def test_holm_known_vector(self) -> None:
        adjusted = hqc.holm_adjust({"a": 0.01, "b": 0.03, "c": 0.04})
        self.assertAlmostEqual(adjusted["a"], 0.03)
        self.assertAlmostEqual(adjusted["b"], 0.06)
        self.assertAlmostEqual(adjusted["c"], 0.06)

    def test_known_interference_values(self) -> None:
        probabilities = {mask: 0.0 for mask in range(16)}
        probabilities[0] = 0.1
        probabilities[1] = 0.2
        probabilities[2] = 0.3
        probabilities[3] = 0.4
        probabilities[4] = 0.5
        probabilities[5] = 0.6
        probabilities[6] = 0.7
        probabilities[7] = 0.8
        expected_i3_012 = (
            probabilities[7]
            - probabilities[6]
            - probabilities[5]
            - probabilities[3]
            + probabilities[4]
            + probabilities[2]
            + probabilities[1]
            - probabilities[0]
        )
        self.assertAlmostEqual(
            hqc.contrasts(probabilities)["I3_012"], expected_i3_012
        )

    def test_simplex_mitigation_recovers_known_distribution(self) -> None:
        assignment = np.eye(8) * 0.96
        assignment += (np.ones((8, 8)) - np.eye(8)) * (0.04 / 7.0)
        truth = np.array([0.40, 0.20, 0.10, 0.08, 0.07, 0.06, 0.05, 0.04])
        observed = assignment @ truth
        recovered = hqc.mitigate(observed, assignment)
        np.testing.assert_allclose(recovered, truth, atol=1e-10)
        self.assertAlmostEqual(float(recovered.sum()), 1.0)
        self.assertTrue(np.all(recovered >= 0.0))

    def test_endian_calibration_identity(self) -> None:
        validated = hqc.validate_payload(
            copy.deepcopy(self.payload), self.contract
        )
        before = hqc.assignment_matrix(validated["records"], "before")
        after = hqc.assignment_matrix(validated["records"], "after")
        np.testing.assert_array_equal(before, np.eye(8))
        np.testing.assert_array_equal(after, np.eye(8))

    def test_injected_i4_is_not_misclassified_as_validation(self) -> None:
        payload = copy.deepcopy(self.payload)
        for record in payload["records"]:
            if (
                record["kind"] == "experiment"
                and record["experiment_type"] == "primary"
                and record["logical_mask"] == "1111"
            ):
                record["counts"] = {"000": 960, "001": 64}
        result = self.analyze(payload, reps=48)
        estimate = result["statistics"]["raw"]["primary"]["I4_0123"][
            "estimate"
        ]
        self.assertAlmostEqual(estimate, -0.0625)
        self.assertEqual(result["classification"]["label"], "inconclusive")

    def test_readout_drift_above_threshold_is_inconclusive(self) -> None:
        payload = copy.deepcopy(self.payload)
        for record in payload["records"]:
            if (
                record["kind"] == "readout_calibration"
                and record["position"] == "after"
                and record["prepared_state"] == 0
            ):
                record["counts"] = {"000": 960, "001": 64}
        result = self.analyze(payload, reps=24)
        self.assertFalse(result["readout"]["drift_below_or_equal_0_02"])
        self.assertEqual(result["classification"]["label"], "inconclusive")
        self.assertIn(
            "readout_matrix_drift_above_0_02",
            result["classification"]["inconclusive_flags"],
        )

    def test_b_sys_uses_sign_half_difference(self) -> None:
        payload = copy.deepcopy(self.payload)
        for record in payload["records"]:
            if (
                record["kind"] == "experiment"
                and record["experiment_type"] == "sign"
                and record["logical_mask"] == "1111"
            ):
                record["counts"] = {"000": 960, "001": 64}
        result = self.analyze(payload, reps=32)
        systematic = result["systematics"]["I4_0123"]
        self.assertAlmostEqual(
            systematic["sign_half_difference_max"], 0.03125
        )
        self.assertGreaterEqual(systematic["B_sys"], 0.03125)

    def test_mapping_specific_artifact_is_flagged(self) -> None:
        payload = copy.deepcopy(self.payload)
        for record in payload["records"]:
            if (
                record["kind"] == "experiment"
                and record["experiment_type"] == "primary"
                and record["logical_to_physical_mapping"] == [0, 1, 2, 3]
                and record["logical_mask"] == "1111"
            ):
                record["counts"] = {"000": 896, "001": 128}
        result = self.analyze(payload, reps=32)
        self.assertTrue(
            result["controls"]["raw"]["I4_0123"]["inconsistent"]
        )
        self.assertTrue(
            any(
                flag.endswith(":I4_0123:mapping")
                for flag in result["controls"]["inconsistency_flags"]
            )
        )

    def test_singular_readout_disables_mitigation(self) -> None:
        payload = copy.deepcopy(self.payload)
        for record in payload["records"]:
            if record["kind"] == "readout_calibration":
                record["counts"] = {"000": record["shots"]}
        result = self.analyze(payload, reps=20)
        self.assertFalse(result["readout"]["mitigated_analysis_available"])
        self.assertFalse(result["statistics"]["mitigated_before"]["available"])
        self.assertFalse(result["statistics"]["mitigated_after"]["available"])
        self.assertEqual(result["classification"]["label"], "inconclusive")
        self.assertIn(
            "more_than_5pct_invalid_mitigation_bootstraps",
            result["classification"]["inconclusive_flags"],
        )
        json.dumps(result, allow_nan=False)

    def test_unavailable_execution_time_is_inconclusive_not_invalid(self) -> None:
        payload = copy.deepcopy(self.payload)
        payload["schedule_order_preserved"] = "unknown"
        payload["execution_bracketing_verified"] = "unknown"
        payload["execution_time_source"] = "unavailable"
        for record in payload["records"]:
            record["executed_at_utc"] = None
            record["execution_start_utc"] = None
            record["execution_stop_utc"] = None
            record["execution_time_source"] = "unavailable"
        for row in payload["pub_metadata"]:
            row["executed_at_utc"] = None
            row["execution_start_utc"] = None
            row["execution_stop_utc"] = None
            row["execution_time_source"] = "unavailable"
        result = self.analyze(payload, reps=20)
        self.assertFalse(result["drift"]["time_available"])
        self.assertEqual(result["classification"]["label"], "inconclusive")
        self.assertIn(
            "execution_time_unavailable",
            result["classification"]["inconclusive_flags"],
        )
        self.assertIn(
            "readout_bracketing_not_verified",
            result["classification"]["inconclusive_flags"],
        )

    def test_verified_schedule_rejects_missing_execution_windows(self) -> None:
        payload = copy.deepcopy(self.payload)
        payload["schedule_order_preserved"] = True
        payload["execution_bracketing_verified"] = True
        payload["execution_time_source"] = "submission_order"
        for record in payload["records"]:
            record["executed_at_utc"] = None
            record["execution_start_utc"] = None
            record["execution_stop_utc"] = None
            record["execution_time_source"] = "unavailable"
        for row in payload["pub_metadata"]:
            row["executed_at_utc"] = None
            row["execution_start_utc"] = None
            row["execution_stop_utc"] = None
            row["execution_time_source"] = "unavailable"
        with self.assertRaisesRegex(ValueError, "not verified"):
            hqc.validate_payload(payload, self.contract)

    def test_null_schedule_flag_is_accepted_as_unknown(self) -> None:
        payload = copy.deepcopy(self.payload)
        payload["schedule_order_preserved"] = None
        validated = hqc.validate_payload(payload, self.contract)
        self.assertEqual(validated["schedule_order_preserved"], "unknown")

    def test_nonmonotonic_timestamps_are_sorted_when_schedule_not_preserved(
        self,
    ) -> None:
        payload = copy.deepcopy(self.payload)
        payload["schedule_order_preserved"] = False
        payload["execution_bracketing_verified"] = False
        payload["execution_time_source"] = "runtime-span"
        start = datetime(2026, 7, 23, 16, 0, tzinfo=timezone.utc)
        for record in payload["records"]:
            pub_index = record["pub_index"]
            execution_start = start + timedelta(
                milliseconds=2 * (87 - pub_index)
            )
            record["executed_at_utc"] = execution_start.isoformat()
            record["execution_start_utc"] = execution_start.isoformat()
            record["execution_stop_utc"] = (
                execution_start + timedelta(milliseconds=1)
            ).isoformat()
            record["execution_time_source"] = "runtime-span"
        for row in payload["pub_metadata"]:
            execution_start = start + timedelta(
                milliseconds=2 * (87 - row["pub_index"])
            )
            row["executed_at_utc"] = execution_start.isoformat()
            row["execution_start_utc"] = execution_start.isoformat()
            row["execution_stop_utc"] = (
                execution_start + timedelta(milliseconds=1)
            ).isoformat()
            row["execution_time_source"] = "runtime-span"
        result = self.analyze(payload, reps=20)
        self.assertFalse(
            result["validation"][
                "execution_timestamps_monotonic_in_record_order"
            ]
        )
        self.assertEqual(result["drift"]["time_basis"], "executed_at_utc")
        self.assertTrue(result["drift"]["time_available"])

        payload["schedule_order_preserved"] = True
        with self.assertRaisesRegex(ValueError, "contradicts execution timestamps"):
            hqc.validate_payload(payload, self.contract)

    def test_overlapping_windows_are_diagnostic_but_not_claim_eligible(
        self,
    ) -> None:
        payload = copy.deepcopy(self.payload)
        payload["schedule_order_preserved"] = "unknown"
        payload["execution_bracketing_verified"] = "unknown"
        payload["execution_time_source"] = "runtime-span"
        start = datetime(2026, 7, 23, 16, 30, tzinfo=timezone.utc)
        for record in payload["records"]:
            execution_start = start + timedelta(
                milliseconds=record["pub_index"]
            )
            record["executed_at_utc"] = execution_start.isoformat()
            record["execution_start_utc"] = execution_start.isoformat()
            record["execution_stop_utc"] = (
                execution_start + timedelta(milliseconds=10)
            ).isoformat()
            record["execution_time_source"] = "runtime-span"
        for row in payload["pub_metadata"]:
            execution_start = start + timedelta(
                milliseconds=row["pub_index"]
            )
            row["executed_at_utc"] = execution_start.isoformat()
            row["execution_start_utc"] = execution_start.isoformat()
            row["execution_stop_utc"] = (
                execution_start + timedelta(milliseconds=10)
            ).isoformat()
            row["execution_time_source"] = "runtime-span"
        result = self.analyze(payload, reps=20)
        self.assertEqual(
            result["execution_integrity"]["derived_execution_order_status"],
            "unknown",
        )
        self.assertFalse(result["execution_integrity"]["claim_eligible"])
        self.assertEqual(result["drift"]["time_basis"], "executed_at_utc")
        self.assertIn(
            "execution_order_not_verified",
            result["classification"]["inconclusive_flags"],
        )
        self.assertIn(
            "readout_bracketing_not_verified",
            result["classification"]["inconclusive_flags"],
        )

    def test_bootstrap_mitigation_failure_is_nan_not_abort(self) -> None:
        validated = hqc.validate_payload(
            copy.deepcopy(self.payload), self.contract
        )
        before = hqc.assignment_matrix(validated["records"], "before")
        after = hqc.assignment_matrix(validated["records"], "after")
        block_rows, by_block_records = hqc._point_block_rows(
            validated["records"], before, after, True, True
        )
        ordered_primary, _, _ = hqc._ordered_primary_blocks(
            block_rows, validated
        )
        with patch.object(
            hqc, "mitigate", side_effect=ValueError("synthetic SLSQP failure")
        ):
            bootstrap = hqc._hierarchical_bootstrap(
                validated,
                block_rows,
                by_block_records,
                repetitions=20,
                seed=123,
                before_available=True,
                after_available=True,
                ordered_primary=ordered_primary,
            )
        self.assertTrue(
            np.all(
                np.isfinite(
                    bootstrap["samples"]["raw"]["primary"]["I4_0123"]
                )
            )
        )
        self.assertTrue(
            np.all(
                np.isnan(
                    bootstrap["samples"]["mitigated_before"]["primary"][
                        "I4_0123"
                    ]
                )
            )
        )
        self.assertTrue(
            np.all(
                np.isnan(
                    bootstrap["samples"]["mitigated_after"]["primary"][
                        "I4_0123"
                    ]
                )
            )
        )

    def test_bootstrap_resamples_every_selected_block_occurrence(self) -> None:
        validated = hqc.validate_payload(
            copy.deepcopy(self.payload), self.contract
        )
        before = hqc.assignment_matrix(validated["records"], "before")
        after = hqc.assignment_matrix(validated["records"], "after")
        block_rows, by_block_records = hqc._point_block_rows(
            validated["records"], before, after, True, True
        )
        ordered_primary, _, _ = hqc._ordered_primary_blocks(
            block_rows, validated
        )
        real_rng = np.random.Generator(np.random.PCG64(321))

        class CountingRng:
            def __init__(self) -> None:
                self.multinomial_calls = 0

            def choice(self, *args, **kwargs):
                return real_rng.choice(*args, **kwargs)

            def multinomial(self, *args, **kwargs):
                self.multinomial_calls += 1
                return real_rng.multinomial(*args, **kwargs)

        counting_rng = CountingRng()
        with patch.object(
            hqc.np.random, "default_rng", return_value=counting_rng
        ):
            hqc._hierarchical_bootstrap(
                validated,
                block_rows,
                by_block_records,
                repetitions=20,
                seed=321,
                before_available=True,
                after_available=True,
                ordered_primary=ordered_primary,
            )
        # Per replicate: 64 calibration draws plus 40 selected block
        # occurrences times 16 contexts.  The older, incorrect unique-block
        # implementation made only 24*16 experiment draws.
        self.assertEqual(
            counting_rng.multinomial_calls,
            20 * (64 + 40 * 16),
        )

    def test_point_mitigation_failure_disables_only_failed_variant(self) -> None:
        payload = copy.deepcopy(self.payload)
        original_mitigate = hqc.mitigate
        call_count = 0

        def fail_first_fit(
            probabilities: np.ndarray, assignment: np.ndarray
        ) -> np.ndarray:
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                raise ValueError("synthetic point-fit failure")
            return original_mitigate(probabilities, assignment)

        with patch.object(hqc, "mitigate", side_effect=fail_first_fit):
            result = self.analyze(payload, reps=20)
        self.assertFalse(
            result["readout"]["point_mitigation_fit_available_before"]
        )
        self.assertTrue(
            result["readout"]["point_mitigation_fit_available_after"]
        )
        self.assertFalse(result["statistics"]["mitigated_before"]["available"])
        self.assertTrue(result["statistics"]["mitigated_after"]["available"])
        self.assertEqual(result["classification"]["label"], "inconclusive")
        self.assertIn(
            "readout_mitigation_fit_failed:before",
            result["classification"]["inconclusive_flags"],
        )

    def test_paired_variant_disagreement_is_flagged(self) -> None:
        repetitions = 20
        samples = {
            variant: {
                "primary": {
                    name: np.zeros(repetitions, dtype=float)
                    for name in hqc.CONTRAST_NAMES
                }
            }
            for variant in hqc.VARIANTS
        }
        samples["mitigated_before"]["primary"]["I4_0123"][:] = 0.05
        bootstrap = {"samples": samples, "repetitions": repetitions}
        agreement = hqc._paired_variant_agreement(bootstrap)
        self.assertIn(
            "I4_0123:raw_minus_mitigated_before",
            agreement["inconsistency_flags"],
        )
        self.assertTrue(
            agreement["contrasts"]["I4_0123"][
                "raw_minus_mitigated_before"
            ]["excludes_zero"]
        )

    def test_submission_manifest_internal_digest_contract(self) -> None:
        plan = self.contract["plan"]
        manifest_core = {
            "schema": hqc.SUBMISSION_MANIFEST_SCHEMA,
            "artifacts": {
                "plan_sha256": hqc.sha256_file(hqc.PLAN_FILE),
                "qpy_sha256": hqc.sha256_file(hqc.QPY_FILE),
                "analyzer_sha256": hqc.sha256_file(
                    Path(hqc.__file__).resolve()
                ),
            },
            "experiment": {
                "shots_per_configuration": 1024,
                "parameter_bindings": 384,
                "readout_calibrations": 64,
                "effective_configurations": 448,
            },
            "versions": {"fixture": "1"},
            "backend_snapshot": plan["selected_snapshot"],
        }
        manifest = dict(manifest_core)
        manifest["manifest_sha256"] = hashlib.sha256(
            hqc.canonical_json_bytes(manifest_core)
        ).hexdigest()
        with tempfile.TemporaryDirectory() as temporary_directory:
            manifest_path = Path(temporary_directory) / "manifest.json"
            manifest_path.write_text(
                json.dumps(manifest, ensure_ascii=False), encoding="utf-8"
            )
            contract = hqc.load_frozen_contract(manifest_path=manifest_path)
            self.assertEqual(
                contract["manifest_sha256"], manifest["manifest_sha256"]
            )
            self.assertEqual(contract["contract_mode"], "submission-manifest")

            tampered = copy.deepcopy(manifest)
            tampered["versions"]["fixture"] = "2"
            manifest_path.write_text(
                json.dumps(tampered, ensure_ascii=False), encoding="utf-8"
            )
            with self.assertRaisesRegex(ValueError, "internal digest"):
                hqc.load_frozen_contract(manifest_path=manifest_path)

    def test_real_analysis_requires_submission_manifest(self) -> None:
        with tempfile.TemporaryDirectory() as temporary_directory:
            missing = Path(temporary_directory) / "missing-manifest.json"
            with self.assertRaisesRegex(ValueError, "Submission manifest is missing"):
                hqc.load_frozen_contract(
                    manifest_path=missing,
                    allow_pre_run_fallback=False,
                )

    def test_export_receipt_provenance_and_raw_tamper_detection(self) -> None:
        payload = copy.deepcopy(self.payload)
        job_id = payload["job_id"]
        with tempfile.TemporaryDirectory() as temporary_directory:
            directory = Path(temporary_directory)
            raw_path = directory / f"HQC_IBM_RAW_COUNTS_{job_id}.json"
            receipt_path = (
                directory / f"HQC_IBM_EXPORT_RECEIPT_{job_id}.json"
            )
            raw_path.write_text(
                json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
                encoding="utf-8",
            )
            receipt = {
                "schema": hqc.EXPORT_RECEIPT_SCHEMA,
                "created_at_utc": "2026-07-23T17:00:00+00:00",
                "job_id": job_id,
                "manifest_sha256": self.contract["manifest_sha256"],
                "raw_counts": raw_path.name,
                "raw_counts_sha256": hqc.sha256_file(raw_path),
                "primitive_result": "fixture.json",
                "primitive_result_sha256": "0" * 64,
                "records": 448,
            }
            receipt_path.write_text(
                json.dumps(receipt, ensure_ascii=False, indent=2) + "\n",
                encoding="utf-8",
            )
            provenance = hqc.verify_export_provenance(
                raw_path, payload, self.contract
            )
            self.assertEqual(
                provenance["raw_input_sha256"],
                receipt["raw_counts_sha256"],
            )
            self.assertEqual(
                provenance["export_receipt_sha256"],
                hqc.sha256_file(receipt_path),
            )

            raw_path.write_text(
                raw_path.read_text(encoding="utf-8") + " ",
                encoding="utf-8",
            )
            with self.assertRaisesRegex(ValueError, "digest verification"):
                hqc.verify_export_provenance(
                    raw_path, payload, self.contract
                )

    def test_real_input_without_export_receipt_is_rejected(self) -> None:
        payload = copy.deepcopy(self.payload)
        with tempfile.TemporaryDirectory() as temporary_directory:
            raw_path = (
                Path(temporary_directory)
                / f"HQC_IBM_RAW_COUNTS_{payload['job_id']}.json"
            )
            raw_path.write_text(json.dumps(payload), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "receipt is missing"):
                hqc.verify_export_provenance(
                    raw_path, payload, self.contract
                )

    def test_malformed_payloads_are_rejected(self) -> None:
        mutations: list[tuple[str, callable]] = [
            (
                "wrong plan hash",
                lambda payload: payload.__setitem__("plan_sha256", "0" * 64),
            ),
            (
                "missing record",
                lambda payload: payload["records"].pop(),
            ),
            (
                "out of order",
                lambda payload: payload["records"][33].__setitem__(
                    "execution_index", 32
                ),
            ),
            (
                "wrong shots",
                lambda payload: payload["records"][32].__setitem__("shots", 1023),
            ),
            (
                "negative count",
                lambda payload: payload["records"][32].__setitem__(
                    "counts", {"000": -1, "001": 1025}
                ),
            ),
            (
                "float count",
                lambda payload: payload["records"][32].__setitem__(
                    "counts", {"000": 512.0, "001": 512}
                ),
            ),
            (
                "wrong count total",
                lambda payload: payload["records"][32].__setitem__(
                    "counts", {"000": 1}
                ),
            ),
            (
                "invalid count key",
                lambda payload: payload["records"][32].__setitem__(
                    "counts", {"0000": 1024}
                ),
            ),
            (
                "duplicate normalized count key",
                lambda payload: payload["records"][32].__setitem__(
                    "counts", {"000": 512, "0 00": 512}
                ),
            ),
            (
                "wrong logical mask",
                lambda payload: payload["records"][32].__setitem__(
                    "logical_mask", "1111"
                ),
            ),
            (
                "wrong mapping",
                lambda payload: payload["records"][32].__setitem__(
                    "logical_to_physical_mapping", [3, 2, 1, 0]
                ),
            ),
            (
                "wrong pub index",
                lambda payload: payload["records"][32].__setitem__(
                    "pub_index", 33
                ),
            ),
            (
                "wrong binding index",
                lambda payload: payload["records"][32].__setitem__(
                    "binding_index", 1
                ),
            ),
            (
                "binding index on calibration",
                lambda payload: payload["records"][0].__setitem__(
                    "binding_index", 0
                ),
            ),
            (
                "invalid timestamp",
                lambda payload: payload["records"][0].__setitem__(
                    "executed_at_utc", "not-a-time"
                ),
            ),
            (
                "partial execution window",
                lambda payload: payload["records"][0].__setitem__(
                    "execution_stop_utc", None
                ),
            ),
            (
                "empty record time source",
                lambda payload: payload["records"][0].__setitem__(
                    "execution_time_source", ""
                ),
            ),
            (
                "invalid schedule flag",
                lambda payload: payload.__setitem__(
                    "schedule_order_preserved", "definitely-not-valid"
                ),
            ),
            (
                "invalid bracketing flag",
                lambda payload: payload.__setitem__(
                    "execution_bracketing_verified", None
                ),
            ),
        ]
        for label, mutate in mutations:
            with self.subTest(label=label):
                payload = copy.deepcopy(self.payload)
                mutate(payload)
                with self.assertRaises(ValueError):
                    hqc.validate_payload(payload, self.contract)


if __name__ == "__main__":
    unittest.main(verbosity=2)

#!/usr/bin/env python3
"""Static acceptance checks for HQC_REVISED_RESEARCH_DRAFT_HE.md."""

from pathlib import Path
import re


HERE = Path(__file__).resolve().parent
DRAFT = HERE / "HQC_REVISED_RESEARCH_DRAFT_HE.md"


def require(text: str, needle: str, label: str) -> None:
    if needle not in text:
        raise AssertionError(f"חסר תיקון קריטי: {label}: {needle!r}")


def forbid(text: str, needle: str, label: str) -> None:
    if needle in text:
        raise AssertionError(f"נמצאה טענה אסורה: {label}: {needle!r}")


def main() -> None:
    text = DRAFT.read_text(encoding="utf-8")

    if len(text) < 8_000:
        raise AssertionError("הטיוטה קצרה מדי מכדי לכלול מפרט מחקר שלם")

    forbidden = {
        "מספר חדש": "אין להציג את q כמערכת מספרים חדשה",
        "כפל חדש": "covering product אינו חדש",
        "משוואת שרדינגר החדשה": "הדינמיקה אינה חדשה אלגברית",
        "חשבון דיפרנציאלי חדש": "זהו חשבון הפרשים סופיים מוכר",
        "HQC_1=מכניקת הקוונטים הרגילה": "אין שקילות ל-QM מלאה",
        "HQC_1 = מכניקת הקוונטים הרגילה": "אין שקילות ל-QM מלאה",
        "אובייקט היסוד החדש": "ייצוג Möbius אינו אובייקט מתמטי חדש",
        "HQC היא הוכחה לפיזיקה חדשה": "אין לטעון שהמודל כבר הוכח פיזיקלית",
    }
    for needle, label in forbidden.items():
        forbid(text, needle, label)

    required = {
        "**[הגדרה]**": "תיוג הגדרות",
        "**[ידע מתמטי מוכר]**": "תיוג prior art",
        "**[למה במודל]**": "תיוג תוצאות מוכחות",
        "**[השערה פיזיקלית": "תיוג השערות פיזיקליות",
        "**[בעיה פתוחה": "תיוג בעיות פתוחות",
        "\\mathcal B": "סימון אלגברת המקדמים שאינו מתנגש עם A",
        "יהי \\(X\\) אוסף סופי": "הנחת סופיות X",
        "b_A=E_A(\\rho)^{1/2}": "השורש החיובי הנכון",
        "\\deg0=-\\infty": "מוסכמת דרגת האפס",
        "\\min\\{|X|,\\deg q+\\deg r\\}": "חסם הדרגה המתוקן",
        "F_r\\star F_s\\subseteq F_{\\min\\{|X|,r+s\\}}": "חוק הסינון",
        "n>2r": "חסם ההתאבכות",
        "לא־ריקים וזרים": "תנאי הבלוקים בלמת ההתאבכות",
        "המשקלים הלא־מנורמלים": "הפרדת משקל מהסתברות",
        "D(A)=D>0": "תנאי נרמול קבוע",
        "L_H(F_r)\\subseteq F_r": "תנאי אינווריאנטיות לדינמיקה",
        "\\psi_t(\\{1,2\\})=e^{-iht}-1": "דוגמה נגדית לשימור דרגה",
        "\\left(\\prod_{i\\in S}\\Delta_i\\right)\\Psi(\\varnothing)": (
            "שחזור הפרשים בנקודת הבסיס"
        ),
        "\\sum_{R\\subseteq A}\\psi(S\\cup R)": "נוסחת הפרשים בבסיס כללי",
        "Sorkin null benchmark": "הגדרה נכונה לניסוי IBM",
        "הניסוי אינו יכול לאמת את HQC": "מגבלת ההסקה מניסוי IBM",
        "ללא postselection": "איסור postselection",
        "אישור סופי ומפורש": "שער אישור לפני QPU",
        "אין כאן טענת בכורה": "הצהרת novelty זהירה",
    }
    for needle, label in required.items():
        require(text, needle, label)

    # כל ארבעת הסמנים האפיסטמיים צריכים להופיע יותר מפעם אחת.
    for marker in (
        r"\*\*\[הגדרה\]\*\*",
        r"\*\*\[ידע מתמטי מוכר\]\*\*",
        r"\*\*\[למה במודל\]\*\*",
    ):
        if len(re.findall(marker, text)) < 2:
            raise AssertionError(f"הסמן {marker!r} אינו משמש באופן שיטתי")

    # נרמול חייב להופיע כמנה w_y / D(A), לא כנוסחה משובשת.
    require(text, "p(y\\mid A)=", "נוסחת הסתברות מותנית")
    require(text, "\\frac{w_y(A)}{D(A)}", "המנה המנורמלת")

    # ודא שהמסמך אינו מבטיח שימור דרגה.
    if re.search(r"הדינמיקה (?:תמיד )?משמרת (?:את )?הדרגה", text):
        raise AssertionError("המסמך עדיין מבטיח שימור דרגה אוטומטי")

    print(
        "PASS: HQC revised draft includes all critical corrections, "
        "separates known mathematics from physical hypotheses, and "
        "states IBM as a null benchmark only."
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Public-package safety regressions; separate from the frozen 56 tests."""

from __future__ import annotations

import os
import importlib.util
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path


PACKAGE_ROOT = Path(__file__).resolve().parents[1]


class PublicReleaseRegressions(unittest.TestCase):
    def run_help(self, relative: str, *args: str, env: dict | None = None):
        return subprocess.run(
            [sys.executable, str(PACKAGE_ROOT / relative), *args, "--help"],
            cwd=PACKAGE_ROOT,
            env=env,
            capture_output=True,
            text=True,
            timeout=30,
            check=False,
        )

    def test_prepare_help_does_not_write_workdir(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            env = dict(os.environ)
            env["HQC_WORKDIR"] = directory
            result = self.run_help(
                "portable-source/hqc_prepare_sorkin_experiment.py",
                env=env,
            )
            self.assertEqual(result.returncode, 0, result.stderr)
            self.assertEqual(list(Path(directory).iterdir()), [])

    def test_online_help_does_not_read_missing_credentials(self) -> None:
        env = dict(os.environ)
        env["HQC_IBM_TOKEN_FILE"] = "/definitely/missing/token"
        env["HQC_IBM_INSTANCE_FILE"] = "/definitely/missing/instance"
        result = self.run_help("online/check_ibm_access.py", env=env)
        self.assertEqual(result.returncode, 0, result.stderr)

    def test_live_runner_defaults_to_new_run_not_frozen_evidence(self) -> None:
        source = (PACKAGE_ROOT / "dangerous" / "hqc_runtime_job.py").read_text(
            encoding="utf-8"
        )
        self.assertIn('PLAN_FILE = ROOT / "HQC_IBM_PARAMETRIC_DRY_RUN_PLAN.json"', source)
        self.assertIn('QPY_FILE = ROOT / "HQC_IBM_PARAMETRIC_TEMPLATE.qpy"', source)
        header = source.split("def utc_now", maxsplit=1)[0]
        self.assertNotIn('"frozen-pre-run"', header)

    def test_safe_orchestrator_has_no_live_code_path(self) -> None:
        source = (PACKAGE_ROOT / "scripts" / "verify_all.py").read_text(
            encoding="utf-8"
        )
        self.assertNotIn("dangerous/", source)
        self.assertNotIn("online/", source)
        self.assertNotIn("prepare-online", source)
        self.assertNotIn("sampler.run", source)

    def test_public_scan_ignores_virtualenv_dependencies(self) -> None:
        scanner_path = PACKAGE_ROOT / "scripts" / "scan_public_release.py"
        spec = importlib.util.spec_from_file_location(
            "hqc_public_release_scanner",
            scanner_path,
        )
        self.assertIsNotNone(spec)
        self.assertIsNotNone(spec.loader)
        scanner = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(scanner)

        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            dependency = root / ".venv" / "lib" / "site-packages" / "vendor.py"
            dependency.parent.mkdir(parents=True)
            dependency.write_text(
                'example = "' + "crn" + ':v1:not-a-release-secret"\n',
                encoding="utf-8",
            )
            public = root / "public.py"
            public.write_text("value = 1\n", encoding="utf-8")
            discovered = {
                path.relative_to(root)
                for path in scanner.text_files(root)
            }
            self.assertEqual(discovered, {Path("public.py")})


if __name__ == "__main__":
    unittest.main(verbosity=2)