//! The clients against services that follow each provider's published
//! contract: each fake parses the submitted program in that provider's own
//! format, runs it on the crate's simulator, and answers in that provider's
//! result shape and bit order. A client whose request, units or bit order
//! departs from the contract reads back the wrong distribution, and these
//! tests fail.
//!
//! The fakes encode the contracts as published; the first job on each real
//! service is still [`probe_circuit`], which settles its bit order there.

use serde_json::{Value, json};
use std::collections::HashMap;
use std::f64::consts::PI;
use std::sync::Mutex;
use wai_quantum::quantum::{Circuit, Gateset};
use wai_quantum::quantum_fabric::sampling::{QPU_DEFAULT, reference_distribution, sample_kernel};
use wai_quantum::quantum_fabric::{FabricClass, Parity, Provenance};
use wai_quantum::quantum_toolchain::from_qasm;
use wai_quantum::quantum_toolchain::native::Basis;
use wai_quantum_qpu::cloud_tasks::{CloudTasks, Keys};
use wai_quantum_qpu::job::{self, Progress, Target};
use wai_quantum_qpu::ledger::Ledger;
use wai_quantum_qpu::registry::Kind;
use wai_quantum_qpu::sampler_v2::SamplerV2;
use wai_quantum_qpu::native_turns::NativeTurns;
use wai_quantum_qpu::prx_circuits::PrxCircuits;
use wai_quantum_qpu::transport::{Method, Request, Response, Scripted, Transport};
use wai_quantum_qpu::{Provider, QpuError, QpuSampler, Secret, Wait, check_probe, counts_from_probabilities, probe_circuit};

// ---------------------------------------------------------------------------
// Running what a fake receives
// ---------------------------------------------------------------------------

/// A radian angle that is a dyadic multiple of π, as the reader writes it.
fn angle_text(rad: f64) -> String {
    if rad == 0.0 {
        return "0".into();
    }
    for j in 0..=40 {
        let m = rad / PI * 2f64.powi(j);
        if (m - m.round()).abs() < 1e-6 {
            let m = m.round() as i64;
            return if j == 0 { format!("{m}*pi") } else { format!("{m}*pi/{}", 1u64 << j) };
        }
    }
    panic!("the fake received an angle that is no dyadic multiple of π: {rad}")
}

/// The probabilities of `width` physical qubits after `body`, a program in the
/// gates the reader takes on `q[0..width]`.
fn physical_distribution(width: usize, body: &str) -> Vec<f64> {
    let text = format!("OPENQASM 3.0;\nqubit[{width}] q;\n{body}");
    let c = from_qasm(&text).unwrap_or_else(|e| panic!("the fake cannot read the program: {e}\n{text}"));
    reference_distribution(&c).expect("the fake runs the program")
}

/// Exact-as-possible counts over the classical register: physical outcome `x`
/// gives `c[i] = bit measured[i] of x`.
fn register_counts(p: &[f64], measured: &[usize], shots: u64) -> HashMap<u64, u64> {
    let mut reg = vec![0.0; 1 << measured.len()];
    for (x, &px) in p.iter().enumerate() {
        let r = measured.iter().enumerate().fold(0usize, |r, (i, &q)| r | (((x >> q) & 1) << i));
        reg[r] += px;
    }
    counts_from_probabilities(&reg, shots).into_iter().enumerate().filter(|&(_, k)| k > 0).map(|(r, k)| (r as u64, k)).collect()
}

/// An angle as a task's program writes it, `0` or `m*pi/d`, in radians.
fn radians_of(text: &str) -> f64 {
    let t = text.trim();
    if t == "0" {
        return 0.0;
    }
    let (m, d) = t.split_once("*pi/").unwrap_or_else(|| panic!("an angle that is not m*pi/d: {t}"));
    m.parse::<f64>().unwrap() * PI / d.parse::<f64>().unwrap()
}

fn not_found(req: &Request) -> Response {
    Response::json(404, &json!({"error": format!("no route {} {}", req.method.as_str(), req.url)}))
}

fn body_json(req: &Request) -> Value {
    serde_json::from_slice(req.body.as_deref().expect("a body")).expect("a JSON body")
}

// ---------------------------------------------------------------------------
// Sampler V2: an IAM token, then OpenQASM 3 in rz/sx/x/cz on `$p`
// ---------------------------------------------------------------------------

const SAMPLER_EDGES: [[u16; 2]; 7] = [[0, 1], [1, 2], [3, 4], [4, 5], [0, 3], [1, 4], [2, 5]];

struct SamplerV2Service {
    jobs: Mutex<HashMap<String, (u32, Vec<String>)>>,
    /// Write `c[0]` as the most significant bit: the wrong convention, for the
    /// probe's test.
    flip_bit_order: bool,
    fail_jobs: bool,
    /// Samples the device discards before it answers, as a post-selected readout
    /// does: it answers fewer shots than it was asked for.
    discard: usize,
}

impl SamplerV2Service {
    fn new() -> Self {
        SamplerV2Service { jobs: Mutex::new(HashMap::new()), flip_bit_order: false, fail_jobs: false, discard: 0 }
    }

    fn answer(&self, req: &Request) -> Response {
        if req.url == "https://iam.test/identity/token" {
            let body = String::from_utf8(req.body.clone().unwrap()).unwrap();
            assert!(body.starts_with("grant_type=urn%3Aibm%3Aparams%3Aoauth%3Agrant-type%3Aapikey&apikey="), "{body}"); // allow-name: a protocol grant type
            return Response::json(200, &json!({"access_token": "tok-1", "token_type": "Bearer", "expires_in": 3600, "expiration": 4_000_000_000u64}));
        }
        let authed = req.header_value("Authorization") == Some("Bearer tok-1") && req.header_value("Service-CRN") == Some("crn:test") && req.header_value(wai_quantum_qpu::sampler_v2::VERSION_HEADER) == Some("2026-04-15");
        if !authed {
            return Response::json(401, &json!({"error": "unauthorised"}));
        }
        let path = req.url.strip_prefix("https://sampler.test/api/v1").unwrap_or("");
        match (req.method, path) {
            (Method::Get, "/backends/test_device/configuration") => {
                Response::json(200, &json!({"backend_name": "test_device", "n_qubits": 6, "basis_gates": ["cz", "id", "rz", "sx", "x"], "coupling_map": [[0, 1]], "gates": [{"name": "cz", "coupling_map": SAMPLER_EDGES.iter().flat_map(|e| [[e[0], e[1]], [e[1], e[0]]]).collect::<Vec<_>>()}]}))
            }
            (Method::Post, "/jobs") => {
                let b = body_json(req);
                assert_eq!(b["program_id"], "sampler");
                assert_eq!(b["params"]["version"], 2);
                let pub0 = &b["params"]["pubs"][0];
                let (qasm, shots) = (pub0[0].as_str().unwrap(), pub0[2].as_u64().unwrap());
                assert!(pub0[1].is_null());
                let mut body = String::new();
                let mut measured: Vec<(usize, usize)> = Vec::new();
                for line in qasm.lines() {
                    let line = line.trim();
                    if line.starts_with("OPENQASM") || line.starts_with("include") || line.starts_with("bit[") || line.is_empty() {
                        continue;
                    }
                    assert!(!line.starts_with("gate "), "an ISA circuit defines no gate: {line}");
                    if let Some(rest) = line.strip_prefix("c[") {
                        let (i, p) = rest.split_once("] = measure $").expect("c[i] = measure $p;");
                        measured.push((i.parse().unwrap(), p.trim_end_matches(';').parse().unwrap()));
                        continue;
                    }
                    let name: String = line.chars().take_while(|c| c.is_ascii_alphanumeric()).collect();
                    assert!(["rz", "sx", "x", "cz"].contains(&name.as_str()), "only basis gates: {line}");
                    if name == "cz" {
                        let qs: Vec<u16> = line[2..].trim_end_matches(';').split(',').map(|q| q.trim().trim_start_matches('$').parse().unwrap()).collect();
                        assert!(SAMPLER_EDGES.iter().any(|e| (e[0], e[1]) == (qs[0], qs[1]) || (e[1], e[0]) == (qs[0], qs[1])), "cz on an unconnected pair: {line}");
                    }
                    body.push_str(&line.replace('$', "q[").replace(';', "];").replace(", q[", "],q["));
                    body.push('\n');
                }
                measured.sort();
                let order: Vec<usize> = measured.iter().map(|&(_, p)| p).collect();
                let p = physical_distribution(6, &body);
                let n = order.len();
                let mut samples = Vec::new();
                for (reg, k) in register_counts(&p, &order, shots) {
                    let v = if self.flip_bit_order { (0..n).fold(0u64, |v, i| v | (((reg >> i) & 1) << (n - 1 - i))) } else { reg };
                    for _ in 0..k {
                        samples.push(format!("0x{v:x}"));
                    }
                }
                samples.truncate(samples.len().saturating_sub(self.discard));
                let id = format!("job-{}", self.jobs.lock().unwrap().len());
                self.jobs.lock().unwrap().insert(id.clone(), (0, samples));
                Response::json(200, &json!({"id": id, "backend": "test_device"}))
            }
            // This service gives no metrics (tests/sampler_v2.rs has one that does).
            (Method::Get, p) if p.starts_with("/jobs/") && p.ends_with("/metrics") => not_found(req),
            (Method::Get, p) if p.starts_with("/jobs/") && p.ends_with("/results") => {
                let id = &p["/jobs/".len()..p.len() - "/results".len()];
                let jobs = self.jobs.lock().unwrap();
                Response::json(200, &json!({"results": [{"data": {"c": {"samples": jobs[id].1}}, "metadata": {}}], "metadata": {"version": 2}}))
            }
            (Method::Get, p) if p.starts_with("/jobs/") => {
                let id = p["/jobs/".len()..].split('?').next().unwrap();
                let mut jobs = self.jobs.lock().unwrap();
                let job = jobs.get_mut(id).unwrap();
                job.0 += 1;
                let status = if self.fail_jobs { "Failed" } else if job.0 < 3 { "Queued" } else { "Completed" };
                Response::json(200, &json!({"id": id, "state": {"status": status, "reason": "the fake failed it"}}))
            }
            _ => not_found(req),
        }
    }
}

// ---------------------------------------------------------------------------
// Phased-X circuits: prx/cz in radians on named qubits, one measurement key per bit
// ---------------------------------------------------------------------------

struct PrxService {
    jobs: Mutex<HashMap<String, (u32, Value)>>,
}

const PRX_LOCI: [[&str; 2]; 5] = [["QB1", "QB2"], ["QB2", "QB3"], ["QB3", "QB4"], ["QB4", "QB5"], ["QB2", "QB5"]];

/// The star quantum computer `star-qc`: five qubits round the computational
/// resonator CR1, each coupled to it by `cz`, the third moving its state into it.
const STAR_LOCI: [[&str; 2]; 5] = [["QB1", "CR1"], ["QB2", "CR1"], ["QB3", "CR1"], ["QB4", "CR1"], ["QB5", "CR1"]];
const STAR_MOVER: &str = "QB3";

impl PrxService {
    fn answer(&self, req: &Request) -> Response {
        if req.header_value("Authorization") != Some("Bearer prx-token") {
            return Response::json(401, &json!({"detail": "unauthorised"}));
        }
        let path = req.url.strip_prefix("https://prx.test/api/v1").unwrap_or("");
        let qbs = json!([["QB1"], ["QB2"], ["QB3"], ["QB4"], ["QB5"]]);
        match (req.method, path) {
            (Method::Get, "/calibration-sets/test-qc/default/dynamic-quantum-architecture") => Response::json(
                200,
                &json!({"calibration_set_id": "00000000-0000-0000-0000-000000000000", "qubits": ["QB1", "QB2", "QB3", "QB4", "QB5"], "computational_resonators": [],
                        "gates": {"prx": {"implementations": {"drag_gaussian": {"loci": qbs}}, "default_implementation": "drag_gaussian"},
                                  "cz": {"implementations": {"tgss": {"loci": PRX_LOCI}}, "default_implementation": "tgss", "override_default_implementation": {}},
                                  "measure": {"implementations": {"constant": {"loci": qbs}}}}}),
            ),
            (Method::Get, "/calibration-sets/star-qc/default/dynamic-quantum-architecture") => Response::json(
                200,
                &json!({"calibration_set_id": "00000000-0000-0000-0000-000000000001", "qubits": ["QB1", "QB2", "QB3", "QB4", "QB5"], "computational_resonators": ["CR1"],
                        "gates": {"prx": {"implementations": {"drag_gaussian": {"loci": qbs}}, "default_implementation": "drag_gaussian"},
                                  "cz": {"implementations": {"tgss": {"loci": STAR_LOCI}}, "default_implementation": "tgss", "override_default_implementation": {}},
                                  "move": {"implementations": {"tgss_crf": {"loci": [[STAR_MOVER, "CR1"]]}}, "default_implementation": "tgss_crf"},
                                  "measure": {"implementations": {"constant": {"loci": qbs}}}}}),
            ),
            (Method::Post, "/jobs/test-qc/circuit" | "/jobs/star-qc/circuit") => {
                let star = path.contains("star-qc");
                let b = body_json(req);
                let keys: Vec<&str> = b.as_object().unwrap().keys().map(|k| k.as_str()).collect();
                assert!(keys.iter().all(|k| ["calibration_set_id", "circuits", "dd_mode", "heralding_mode", "move_gate_frame_tracking", "move_gate_validation", "shots"].contains(k)), "no field the contract does not name: {keys:?}");
                let shots = b["shots"].as_u64().unwrap();
                // The qubits QB1…QB5 at 0…4, the resonator at 5.
                let qb = |name: &Value| -> usize {
                    let n = name.as_str().unwrap();
                    if n == "CR1" { 5 } else { n.strip_prefix("QB").unwrap().parse::<usize>().unwrap() - 1 }
                };
                let mut body = String::new();
                let mut measured: Vec<(usize, usize)> = Vec::new();
                // The qubit whose state the resonator holds: the service refuses a
                // circuit that uses it out of turn.
                let mut held: Option<usize> = None;
                let refuse = |why: String| Response::json(400, &json!({"detail": why}));
                for ins in b["circuits"][0]["instructions"].as_array().unwrap() {
                    let locus = ins["locus"].as_array().unwrap();
                    if let Some(h) = held
                        && ins["name"] != "move"
                        && locus.iter().any(|q| qb(q) == h)
                    {
                        return refuse(format!("{ins}: a gate on a qubit whose state is in the resonator"));
                    }
                    match ins["name"].as_str().unwrap() {
                        "prx" => {
                            let (theta, phi, q) = (ins["args"]["angle"].as_f64().unwrap(), ins["args"]["phase"].as_f64().unwrap(), qb(&locus[0]));
                            assert!(q < 5, "prx on the resonator");
                            // PRX(θ, φ) = RZ(φ)·RX(θ)·RZ(−φ).
                            body.push_str(&format!("rz({}) q[{q}];\nrx({}) q[{q}];\nrz({}) q[{q}];\n", angle_text(-phi), angle_text(theta), angle_text(phi)));
                        }
                        "cz" => {
                            let (a, c) = (locus[0].as_str().unwrap(), locus[1].as_str().unwrap());
                            let loci: &[[&str; 2]] = if star { &STAR_LOCI } else { &PRX_LOCI };
                            assert!(loci.iter().any(|l| (l[0], l[1]) == (a, c) || (l[1], l[0]) == (a, c)), "cz on an uncalibrated pair {a}, {c}");
                            body.push_str(&format!("cz q[{}],q[{}];\n", qb(&locus[0]), qb(&locus[1])));
                        }
                        // The qubit's state into the empty resonator, or back out: the
                        // exchange, with a phase the device leaves undefined (here
                        // e^{iπ/4} on |1⟩ moved out, its conjugate on |1⟩ moved in),
                        // which only the move back cancels.
                        "move" if star => {
                            let (q, r) = (locus[0].as_str().unwrap(), locus[1].as_str().unwrap());
                            if (q, r) != (STAR_MOVER, "CR1") {
                                return refuse(format!("move on an uncalibrated locus {q}, {r}"));
                            }
                            let q = qb(&locus[0]);
                            held = match held {
                                None => Some(q),
                                Some(h) if h == q => None,
                                Some(_) => return refuse("a move into a resonator that holds another qubit's state".into()),
                            };
                            body.push_str(&format!("swap q[{q}],q[5];\np(pi/4) q[{q}];\np(-pi/4) q[5];\n"));
                        }
                        "measure" => {
                            let key = ins["args"]["key"].as_str().unwrap();
                            measured.push((key.strip_prefix('c').unwrap().parse().unwrap(), qb(&locus[0])));
                        }
                        other => panic!("not one of the service's instructions: {other}"),
                    }
                }
                if held.is_some() {
                    return refuse("the circuit ends with a state in the resonator".into());
                }
                measured.sort();
                let order: Vec<usize> = measured.iter().map(|&(_, q)| q).collect();
                assert!(order.iter().all(|&q| q < 5), "the resonator is never measured");
                let n = order.len();
                let p = physical_distribution(if star { 6 } else { 5 }, &body);
                // Keys listed last bit first, so a client must use the order given.
                let keys: Vec<String> = (0..n).rev().map(|i| format!("c{i}")).collect();
                let mut counts = serde_json::Map::new();
                for (reg, k) in register_counts(&p, &order, shots) {
                    let bits: String = (0..n).rev().map(|i| if (reg >> i) & 1 == 1 { '1' } else { '0' }).collect();
                    counts.insert(bits, json!(k));
                }
                let id = format!("prx-{}", self.jobs.lock().unwrap().len());
                self.jobs.lock().unwrap().insert(id.clone(), (0, json!([{"measurement_keys": keys, "counts": counts}])));
                Response::json(200, &json!({"id": id, "type": "circuit", "status": "waiting"}))
            }
            (Method::Get, p) if p.ends_with("/artifacts/measurement_counts") => {
                let id = &p["/jobs/".len()..p.len() - "/artifacts/measurement_counts".len()];
                Response::json(200, &self.jobs.lock().unwrap()[id].1)
            }
            (Method::Get, p) if p.starts_with("/jobs/") => {
                let id = &p["/jobs/".len()..];
                let mut jobs = self.jobs.lock().unwrap();
                let job = jobs.get_mut(id).unwrap();
                job.0 += 1;
                Response::json(200, &json!({"id": id, "status": if job.0 < 2 { "processing" } else { "completed" }, "new_field": "ignored"}))
            }
            _ => not_found(req),
        }
    }
}

// ---------------------------------------------------------------------------
// Native turns: the service's published API, in `fake_turns`
// ---------------------------------------------------------------------------

mod fake_turns;

/// The native-gate service with its simulator and processors.
struct TurnsService(fake_turns::Fake);

impl TurnsService {
    fn new() -> Self {
        TurnsService(fake_turns::Fake::new())
    }

    fn answer(&self, req: &Request) -> Response {
        self.0.answer(req)
    }
}

// ---------------------------------------------------------------------------
// Cloud tasks: OpenQASM 3 in a verbatim box on a processor's labelled qubits,
// results in the person's bucket, as one row a shot or as probabilities
// ---------------------------------------------------------------------------

struct TasksService {
    simulator: bool,
    /// A processor whose native gates are `prx`, `xx` and `rz`, every pair
    /// coupled, rather than `rx rz cz` on a line.
    xx: bool,
    /// Answer probabilities rather than one row a shot.
    probabilities: bool,
    /// The processor's qubit labels, in the order of its coupling map's
    /// indices, in a line. They move when it is recalibrated.
    labels: Mutex<Vec<u64>>,
    /// The session token a request must carry, sent and signed.
    token: Option<&'static str>,
    /// The service no longer describes the device.
    device_gone: std::sync::atomic::AtomicBool,
    tasks: Mutex<Vec<Task>>,
}

/// A task as the service keeps it: where its results go, what they are, and
/// how far it has got.
struct Task {
    bucket: String,
    directory: String,
    results: Value,
    asked: u32,
    cancelled: Option<u32>,
}

const TASK_LABELS: [u64; 5] = [4, 7, 9, 12, 15];
/// The region tasks are made in, which their ids name.
const TASK_REGION: &str = "us-west-1";

impl TasksService {
    fn new(simulator: bool, probabilities: bool) -> TasksService {
        TasksService { simulator, xx: false, probabilities, labels: Mutex::new(TASK_LABELS.to_vec()), token: None, device_gone: Default::default(), tasks: Mutex::new(Vec::new()) }
    }

    fn device(&self) -> Value {
        let caps = if self.simulator {
            json!({"paradigm": {"qubitCount": 34, "nativeGateSet": [], "connectivity": {}}, "service": {"deviceCost": {"price": "0.075", "unit": "minute"}, "shotsRange": [1, 100000], "updatedAt": "2026-11-05T00:00:00Z"}})
        } else if self.xx {
            json!({"paradigm": {"qubitCount": self.labels.lock().unwrap().len(), "nativeGateSet": ["prx", "xx", "rz"], "connectivity": {"fullyConnected": true, "connectivityGraph": {}}}, "service": {"deviceCost": {"price": "0.02350", "unit": "shot"}, "shotsRange": [1, 2000], "updatedAt": "2026-11-05T00:00:00Z"}})
        } else {
            let labels = self.labels.lock().unwrap();
            let mut graph = serde_json::Map::new();
            for (i, l) in labels.iter().enumerate() {
                let near: Vec<String> = [i.checked_sub(1), Some(i + 1)].into_iter().flatten().filter_map(|j| labels.get(j)).map(u64::to_string).collect();
                graph.insert(l.to_string(), json!(near));
            }
            json!({"paradigm": {"qubitCount": labels.len(), "nativeGateSet": ["rx", "rz", "cz"], "connectivity": {"fullyConnected": false, "connectivityGraph": graph}}, "service": {"deviceCost": {"price": "0.000425", "unit": "shot"}, "shotsRange": [10, 50000], "updatedAt": "2026-11-05T00:00:00Z"}})
        };
        json!({"deviceName": if self.simulator { "SIM" } else { "Line" }, "providerName": "Lab", "deviceStatus": "ONLINE", "deviceType": if self.simulator { "SIMULATOR" } else { "QPU" }, "deviceCapabilities": caps.to_string()})
    }

    /// A task's id, as the service writes one: its region, an account, its number.
    fn task_id(n: usize) -> String {
        format!("arn:aws:braket:{TASK_REGION}:000000000000:quantum-task/{n}") // allow-name: a task's id, as the service gives it
    }

    /// The bucket and the key a storage request names, in either addressing:
    /// the bucket in the host, or, for a name with a dot, first in the path.
    fn storage_object(url: &str) -> Option<(String, String)> {
        let rest = url.strip_prefix("https://")?;
        let (host, path) = rest.split_once('/')?;
        let domain = format!("s3.{TASK_REGION}.amazonaws.com"); // allow-name: the storage service's endpoint
        if host == domain {
            let (bucket, key) = path.split_once('/')?;
            return Some((bucket.to_string(), key.to_string()));
        }
        let bucket = host.strip_suffix(&format!(".{domain}"))?;
        (!bucket.contains('.')).then(|| (bucket.to_string(), path.to_string()))
    }

    fn answer(&self, req: &Request) -> Response {
        let auth = req.header_value("Authorization").unwrap_or_default();
        if !auth.starts_with("AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/") { // allow-name: the signature's literal
            return Response::json(403, &json!({"message": "unsigned"}));
        }
        // A temporary key's token is sent, and is among what is signed.
        if self.token.is_some() && (req.header_value("x-amz-security-token") != self.token || !auth.contains("x-amz-security-token")) {
            return Response::json(403, &json!({"message": "no token"}));
        }
        if let Some((bucket, key)) = Self::storage_object(&req.url) {
            if !auth.contains(&format!("/{TASK_REGION}/s3/aws4_request")) {
                return Response::json(403, &json!({"message": "signed for another region or service"}));
            }
            let tasks = self.tasks.lock().unwrap();
            return match tasks.iter().find(|t| t.bucket == bucket && format!("{}/results.json", t.directory) == key) {
                Some(t) => Response::json(200, &t.results),
                None => Response::json(404, &json!({"Code": "NoSuchKey", "bucket": bucket, "key": key})),
            };
        }
        let path = req.url.strip_prefix("https://tasks.test").unwrap_or("");
        let task_of = |p: &str| -> Option<usize> { p.rsplit("%2F").next()?.trim_end_matches("/cancel").parse().ok() };
        match req.method {
            Method::Get if path.starts_with("/device/") => {
                if self.device_gone.load(std::sync::atomic::Ordering::Relaxed) {
                    return Response::json(404, &json!({"message": "no such device"}));
                }
                Response::json(200, &self.device())
            }
            Method::Get if path.starts_with("/quantum-task/") => {
                let mut tasks = self.tasks.lock().unwrap();
                let Some(t) = task_of(path).and_then(|n| tasks.get_mut(n)) else { return not_found(req) };
                t.asked += 1;
                // Queued, then running, then done; a cancel ends it in two steps.
                let status = match t.cancelled {
                    Some(at) if t.asked <= at + 1 => "CANCELLING",
                    Some(_) => "CANCELLED",
                    None if t.asked == 1 => "QUEUED",
                    None if t.asked == 2 => "RUNNING",
                    None => "COMPLETED",
                };
                Response::json(200, &json!({"status": status, "outputS3Bucket": t.bucket, "outputS3Directory": t.directory}))
            }
            Method::Put if path.starts_with("/quantum-task/") && path.ends_with("/cancel") => {
                let mut tasks = self.tasks.lock().unwrap();
                let Some(t) = task_of(path).and_then(|n| tasks.get_mut(n)) else { return not_found(req) };
                if t.asked < 3 && t.cancelled.is_none() {
                    t.cancelled = Some(t.asked);
                }
                Response::json(200, &json!({"cancellationStatus": "CANCELLING"}))
            }
            Method::Post if path == "/quantum-task" => {
                let b = body_json(req);
                let action: Value = serde_json::from_str(b["action"].as_str().unwrap()).unwrap();
                let source = action["source"].as_str().unwrap();
                let shots = b["shots"].as_u64().unwrap();
                let labels = self.labels.lock().unwrap().clone();
                let index = |q: &str| -> usize {
                    if let Some(l) = q.strip_prefix('$') {
                        labels.iter().position(|&x| x == l.parse::<u64>().unwrap()).expect("a processor's own label")
                    } else {
                        q.trim_start_matches("q[").trim_end_matches(']').parse().unwrap()
                    }
                };
                let (mut width, mut body, mut measured) = (labels.len(), String::new(), Vec::new());
                for line in source.lines() {
                    let line = line.trim().trim_end_matches(';');
                    if let Some(w) = line.strip_prefix("qubit[").and_then(|w| w.strip_suffix("] q")) {
                        width = w.parse().unwrap();
                    } else if let Some((_, q)) = line.split_once(" = measure ") {
                        measured.push(index(q));
                    } else if line.is_empty() || line.starts_with("OPENQASM") || line.starts_with("bit[") || line.starts_with("#pragma") || line == "box {" || line == "}" || line.starts_with("i ") {
                        continue;
                    } else if let Some(rest) = line.strip_prefix("prx(").or(line.strip_prefix("xx(")) {
                        // The service's phased X rotation, prx(θ, φ) = RZ(φ)·RX(θ)·RZ(−φ),
                        // and its XX(θ) = exp(−iθ/2·X⊗X), each angle where the device takes it.
                        let (params, args) = rest.split_once(')').unwrap();
                        let angles: Vec<f64> = params.split(',').map(radians_of).collect();
                        let qubits: Vec<usize> = args.split(',').map(|q| index(q.trim())).collect();
                        if line.starts_with("prx(") {
                            let (theta, phi, q) = (angles[0], angles[1], qubits[0]);
                            assert!((0.0..=PI).contains(&theta) && (0.0..2.0 * PI).contains(&phi), "prx({theta}, {phi}) is outside the device's ranges");
                            body.push_str(&format!("rz({}) q[{q}];\nrx({}) q[{q}];\nrz({}) q[{q}];\n", angle_text(-phi), angle_text(theta), angle_text(phi)));
                        } else {
                            assert!((0.0..=PI / 2.0).contains(&angles[0]), "xx({}) is outside the device's range", angles[0]);
                            body.push_str(&format!("rxx({}) q[{}], q[{}];\n", angle_text(angles[0]), qubits[0], qubits[1]));
                        }
                    } else {
                        let (gate, args) = line.split_once(' ').unwrap();
                        let gate = if gate == "v" { "sx" } else { gate };
                        let qubits: Vec<String> = args.split(',').map(|q| format!("q[{}]", index(q.trim()))).collect();
                        body.push_str(&format!("{gate} {};\n", qubits.join(", ")));
                    }
                }
                let p = physical_distribution(width, &body);
                // The device's own numbers for the qubits measured: integers, as the
                // service's published result schema has them (gate-model task result v1).
                let named: Vec<u64> = measured.iter().map(|&i| if self.simulator { i as u64 } else { labels[i] }).collect();
                let mut tasks = self.tasks.lock().unwrap();
                let n = tasks.len();
                let id = Self::task_id(n);
                let mut results = json!({
                    "braketSchemaHeader": {"name": "braket.task_result.gate_model_task_result", "version": "1"}, // allow-name: the schema's name
                    "measuredQubits": named,
                    "taskMetadata": {"braketSchemaHeader": {"name": "braket.task_result.task_metadata", "version": "1"}, "id": id, "shots": shots, "deviceId": b["deviceArn"]}, // allow-name: the schema's name
                    "additionalMetadata": {"action": action},
                });
                if self.probabilities {
                    let mut probs = serde_json::Map::new();
                    for (x, &px) in p.iter().enumerate().filter(|&(_, &px)| px > 1e-12) {
                        let bits: String = measured.iter().map(|&q| if (x >> q) & 1 == 1 { '1' } else { '0' }).collect();
                        let e = probs.entry(bits).or_insert(json!(0.0));
                        *e = json!(e.as_f64().unwrap() + px);
                    }
                    results["measurementProbabilities"] = Value::Object(probs);
                } else {
                    let mut rows = Vec::new();
                    for (reg, k) in register_counts(&p, &measured, shots) {
                        let row: Vec<u64> = (0..measured.len()).map(|i| (reg >> i) & 1).collect();
                        rows.extend(std::iter::repeat_n(json!(row), k as usize));
                    }
                    results["measurements"] = json!(rows);
                }
                let prefix = b["outputS3KeyPrefix"].as_str().unwrap();
                tasks.push(Task { bucket: b["outputS3Bucket"].as_str().unwrap().to_string(), directory: format!("{prefix}/{n}"), results, asked: 0, cancelled: None });
                Response::json(201, &json!({"quantumTaskArn": id}))
            }
            _ => not_found(req),
        }
    }
}

// ---------------------------------------------------------------------------
// The tests
// ---------------------------------------------------------------------------

fn no_wait<P: Provider>(s: QpuSampler<P>) -> QpuSampler<P> {
    s.with_wait(Wait { interval: std::time::Duration::ZERO, polls: 10 }, |_| {})
}

/// Circuits whose counts must match their exact distribution.
fn circuits() -> Vec<Circuit> {
    let mut ghz = Circuit::new(4);
    ghz.h(0).cx(0, 1).cx(1, 2).cx(2, 3);
    let mut mixed = Circuit::new(3);
    mixed.h(0).t(0).h(1).cx(0, 2).s(2).h(2).cz(1, 2).h(1);
    vec![probe_circuit(3), ghz, Circuit::qft(3), mixed]
}

/// The classical fidelity of counts against the exact distribution.
fn fidelity(c: &Circuit, counts: &[u64]) -> f64 {
    let p = reference_distribution(c).unwrap();
    let total: u64 = counts.iter().sum();
    p.iter().zip(counts).map(|(a, &k)| (a * k as f64 / total as f64).sqrt()).sum::<f64>().powi(2)
}

fn exercise<P: Provider>(s: &QpuSampler<P>) {
    for c in circuits() {
        let (counts, record) = s.run(&c, 4000).unwrap_or_else(|e| panic!("{}: {e}", s.device().name));
        assert_eq!(counts.iter().sum::<u64>(), 4000);
        let f = fidelity(&c, &counts);
        assert!(f > 0.999, "{}: fidelity {f} on {c:?}", s.device().name);
        assert_eq!(record.shots, 4000);
    }
    let (probe, _) = s.run(&probe_circuit(3), 1000).unwrap();
    check_probe(&probe, 0.99).unwrap();
}

#[test]
fn sampler_v2_round_trips_through_its_contract() {
    let service = SamplerV2Service::new();
    let transport = Scripted::new(|r: &Request| service.answer(r));
    let client = SamplerV2::new(&transport, Secret::new("sampler-key/+="), "crn:test", "test_device").with_endpoints("https://sampler.test/api/v1", "https://iam.test/identity/token").with_clock(|| 1_000);
    let s = no_wait(QpuSampler::new(client).unwrap());
    assert_eq!(s.device().coupling.qubits(), 6);
    assert_eq!(s.device().provenance, Provenance::Silicon);
    exercise(&s);
    assert!(s.jobs().iter().all(|j| j.polls == 2), "{:?}", s.jobs());
    // The key goes to the token endpoint only, form-encoded, and the token is
    // minted once.
    let requests = transport.requests();
    let iam: Vec<&Request> = requests.iter().filter(|r| r.url.contains("iam.test")).collect();
    assert_eq!(iam.len(), 1);
    assert!(String::from_utf8_lossy(iam[0].body.as_ref().unwrap()).ends_with("apikey=sampler-key%2F%2B%3D"));
    for r in requests.iter().filter(|r| !r.url.contains("iam.test")) {
        let all = format!("{} {:?} {}", r.url, r.headers, String::from_utf8_lossy(r.body.as_deref().unwrap_or_default()));
        assert!(!all.contains("sampler-key"), "the key leaked into {}", r.url);
    }
}

#[test]
fn prx_circuits_round_trip_through_their_contract() {
    let service = PrxService { jobs: Mutex::new(HashMap::new()) };
    let transport = Scripted::new(|r: &Request| service.answer(r));
    let s = no_wait(QpuSampler::new(PrxCircuits::new(&transport, Secret::new("prx-token"), "test-qc").with_base("https://prx.test/api/v1")).unwrap());
    assert_eq!(s.device().coupling.qubits(), 5);
    exercise(&s);
    // A mock quantum computer is never silicon.
    let mock = PrxCircuits::new(&transport, Secret::new("prx-token"), "test-qc:mock").with_base("https://prx.test/api/v1");
    assert!(mock.device().is_err() || matches!(mock.device().unwrap().provenance, Provenance::Simulator(_)));
}

/// A star device: five qubits round a resonator, one moving its state into
/// it. Every two-qubit gate goes through the resonator as move, cz, move; the
/// service holds the resonator to its turns and runs each move with a phase
/// it leaves undefined, which only the move back cancels.
#[test]
fn prx_circuits_route_through_a_star_s_resonator() {
    let service = PrxService { jobs: Mutex::new(HashMap::new()) };
    let transport = Scripted::new(|r: &Request| service.answer(r));
    let s = no_wait(QpuSampler::new(PrxCircuits::new(&transport, Secret::new("prx-token"), "star-qc").with_base("https://prx.test/api/v1")).unwrap());
    let d = s.device();
    assert!(d.coupling.qubits() == 6 && d.coupling.is_resonator(5) && d.coupling.moves(2, 5) && !d.coupling.moves(0, 5));
    exercise(&s);
    let sent: Vec<Value> = transport.requests().iter().filter(|r| r.url.ends_with("/jobs/star-qc/circuit")).flat_map(|r| body_json(r)["circuits"][0]["instructions"].as_array().unwrap().clone()).collect();
    assert!(sent.iter().any(|i| i["name"] == "move" && i["locus"] == json!(["QB3", "CR1"])));
    assert!(sent.iter().filter(|i| i["name"] == "cz").all(|i| i["locus"][1] == "CR1"), "every cz is on the resonator");
}

#[test]
fn native_turns_round_trip_through_their_contract_exactly() {
    let service = TurnsService::new();
    let transport = Scripted::new(|r: &Request| service.answer(r));
    let s = no_wait(QpuSampler::new(NativeTurns::new(&transport, Secret::new("turns-key"), "simulator").with_base("https://turns.test/v0.4")).unwrap());
    assert!(matches!(&s.device().provenance, Provenance::Simulator(name) if name.contains("native-turns simulator")));
    exercise(&s);
    assert!(s.jobs().iter().all(|j| j.swaps == 0 && j.service_kwh == Some(0.0)));
    // Every phase on the wire is an exact binary fraction of a turn.
    for r in transport.requests().iter().filter(|r| r.method == Method::Post) {
        for g in body_json(r)["input"]["circuit"].as_array().unwrap() {
            let t = g.get("phase").or_else(|| g.get("angle")).unwrap().as_f64().unwrap();
            assert_eq!((t * 4294967296.0).fract(), 0.0, "{g}");
        }
    }
}

#[test]
fn a_device_certified_through_the_fabric_carries_its_provenance() {
    let service = TurnsService::new();
    let transport = Scripted::new(|r: &Request| service.answer(r));
    let s = no_wait(QpuSampler::new(NativeTurns::new(&transport, Secret::new("turns-key"), "simulator").with_base("https://turns.test/v0.4")).unwrap());
    let member = s.member(QPU_DEFAULT, 2000).unwrap();
    assert_eq!((member.class, member.kernels.len()), (FabricClass::Qpu, 2));
    for (entry, gs) in member.kernels.iter().zip(Gateset::ALL) {
        assert_eq!(entry.kernel, sample_kernel(gs));
        assert_eq!(entry.certificate.provenance, s.device().provenance);
        let Parity::Statistical { fidelity, shots } = entry.certificate.parity else { panic!("{:?}", entry.certificate.parity) };
        assert_eq!(shots, 2000);
        assert!(fidelity > 0.999, "{gs:?}: {fidelity}");
    }
    assert!(member.id.starts_with("remote/qpu-native-turns-simulator-"), "{}", member.id);
    // The report names every certificate by its hash, and every job.
    let r = wai_quantum_qpu::report(&s, &member);
    assert_eq!(r["certificates"].as_array().unwrap().len(), 2);
    for (c, entry) in r["certificates"].as_array().unwrap().iter().zip(&member.kernels) {
        let hash: String = entry.certificate.hash().iter().map(|b| format!("{b:02x}")).collect();
        assert_eq!(c["hash"], hash.as_str());
        assert_eq!(c["parity"]["statistical"]["shots"], 2000);
    }
    assert_eq!(r["jobs"].as_array().unwrap().len(), 2 * QPU_DEFAULT.cases);
}

#[test]
fn the_probe_catches_a_service_with_the_other_bit_order() {
    let service = SamplerV2Service { flip_bit_order: true, ..SamplerV2Service::new() };
    let transport = Scripted::new(|r: &Request| service.answer(r));
    let client = SamplerV2::new(&transport, Secret::new("k"), "crn:test", "test_device").with_endpoints("https://sampler.test/api/v1", "https://iam.test/identity/token").with_clock(|| 1_000);
    let s = no_wait(QpuSampler::new(client).unwrap());
    let (counts, _) = s.run(&probe_circuit(3), 500).unwrap();
    let e = check_probe(&counts, 0.99).unwrap_err();
    assert!(e.contains("the most on index 4"), "{e}");
}

#[test]
fn failed_and_unfinished_jobs_are_errors() {
    let service = SamplerV2Service { fail_jobs: true, ..SamplerV2Service::new() };
    let transport = Scripted::new(|r: &Request| service.answer(r));
    let client = SamplerV2::new(&transport, Secret::new("k"), "crn:test", "test_device").with_endpoints("https://sampler.test/api/v1", "https://iam.test/identity/token").with_clock(|| 1_000);
    let s = no_wait(QpuSampler::new(client).unwrap());
    assert!(matches!(s.run(&probe_circuit(2), 10), Err(QpuError::Job { status, .. }) if status == "Failed"));
    let service = SamplerV2Service::new();
    let transport = Scripted::new(|r: &Request| service.answer(r));
    let client = SamplerV2::new(&transport, Secret::new("k"), "crn:test", "test_device").with_endpoints("https://sampler.test/api/v1", "https://iam.test/identity/token").with_clock(|| 1_000);
    let s = QpuSampler::new(client).unwrap().with_wait(Wait { interval: std::time::Duration::ZERO, polls: 1 }, |_| {});
    assert!(matches!(s.run(&probe_circuit(2), 10), Err(QpuError::Timeout { .. })));
    // A circuit wider than the device has no layout.
    assert!(matches!(s.run(&Circuit::new(7), 10), Err(QpuError::Compile(_))));
}

#[test]
fn a_rejected_request_is_an_http_error_with_the_body() {
    let transport = Scripted::new(|_: &Request| Response::json(403, &json!({"detail": "this token is not allowed here"})));
    let client = PrxCircuits::new(&transport, Secret::new("x"), "qc").with_base("https://prx.test/api/v1");
    match client.device() {
        Err(QpuError::Http { status: 403, body }) => assert!(body.contains("not allowed")),
        other => panic!("{other:?}"),
    }
    assert!(matches!(client.status("j"), Err(QpuError::Http { .. })));
}

/// `curl` end to end against a local server: the headers arrive, the body
/// arrives on its own, and the status comes back.
#[test]
fn curl_carries_a_request_to_a_server_and_back() {
    use std::io::{BufRead, BufReader, Read, Write};
    if std::process::Command::new("curl").arg("--version").output().is_err() {
        println!("no curl on this host: the transport test runs where curl is");
        return;
    }
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    let server = std::thread::spawn(move || {
        let (stream, _) = listener.accept().unwrap();
        let mut reader = BufReader::new(stream.try_clone().unwrap());
        let mut head = Vec::new();
        loop {
            let mut line = String::new();
            reader.read_line(&mut line).unwrap();
            if line == "\r\n" || line.is_empty() {
                break;
            }
            head.push(line.trim_end().to_string());
        }
        let len: usize = head.iter().find_map(|h| h.to_ascii_lowercase().strip_prefix("content-length: ").map(|v| v.parse().unwrap())).unwrap_or(0);
        let mut body = vec![0u8; len];
        reader.read_exact(&mut body).unwrap();
        let reply = b"{\"id\":\"job-7\"}";
        let mut stream = stream;
        write!(stream, "HTTP/1.1 201 Created\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", reply.len()).unwrap();
        stream.write_all(reply).unwrap();
        (head, body)
    });
    let curl = wai_quantum_qpu::transport::Curl::new(20);
    let r = curl.send(&Request::post(format!("http://127.0.0.1:{port}/jobs"), b"{\"shots\":5}".to_vec()).header("Authorization", "Bearer t-1").header("Content-Type", "application/json")).unwrap();
    let (head, body) = server.join().unwrap();
    assert_eq!(r.status, 201);
    assert_eq!(r.body, b"{\"id\":\"job-7\"}");
    assert_eq!(head[0], "POST /jobs HTTP/1.1");
    assert!(head.iter().any(|h| h == "Authorization: Bearer t-1"), "{head:?}");
    assert_eq!(body, b"{\"shots\":5}");
}

// ---------------------------------------------------------------------------
// What each built-in kind states in its receipt
// ---------------------------------------------------------------------------

const BELL: &str = "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n";

/// A transport that lives as long as the test, for a provider a target owns.
fn kept<F: Fn(&Request) -> Response + 'static>(answer: F) -> &'static Scripted<F> {
    Box::leak(Box::new(Scripted::new(answer)))
}

/// A Bell pair run on `provider` as a person runs it, confirmed, and sealed
/// with the published demo key (so that anyone can repeat these receipts).
fn sealed_run(kind: Kind, provider: Box<dyn Provider>) -> Value {
    let t = Target::new("mine", kind, provider, None);
    let req = json!({ "qasm": BELL, "shots": 400, "provider": "mine" });
    let key = wai_quantum::ed25519_dalek::SigningKey::from_bytes(&[42; 32]);
    wai_quantum_qpu::job::run(&t, &req, true, &key, Wait { interval: std::time::Duration::ZERO, polls: 10 }, |_| {}).unwrap_or_else(|e| panic!("{e}"))
}

/// The receipt keeps every rule a verifier checks, its counts' source and
/// class are the ones given, and what it says was sent is the body that was
/// posted: its members, and each setting at its place in it. With
/// `WAI_QUANTUM_RECEIPT_CORPUS` set to a directory, the body is written there
/// as `scripted-<name>.json`, for the receipt corpus
/// (wai/quantum-conformance/receipts).
fn holds_to_what_was_sent(name: &str, body: &Value, posted: &Value, source: &str, class: &str) {
    let c = wai_quantum::engine::job::check(body);
    assert!(c.valid(), "{name}: {:?}\n{}", c.reasons, body["result"]);
    let r = &body["result"];
    assert_eq!((r["counts_source"].as_str(), r["accuracy"]["class"].as_str()), (Some(source), Some(class)), "{name}");
    assert!(r["counts"].as_object().unwrap().keys().all(|k| k == "00" || k == "11"), "{name}: {}", r["counts"]);
    let e = &r["execution"];
    assert_eq!(e["stated_by"], "client", "{name}");
    let mut fields: Vec<&str> = posted.as_object().unwrap().keys().map(String::as_str).collect();
    fields.sort_unstable();
    assert_eq!(e["sent"]["fields"], json!(fields), "{name}: the members stated are the members sent");
    for (k, v) in e["sent"]["settings"].as_object().unwrap() {
        if k.starts_with('/') {
            assert_eq!(posted.pointer(k), Some(v), "{name}: {k} is what was sent");
        }
    }
    if let Some(dir) = std::env::var_os("WAI_QUANTUM_RECEIPT_CORPUS") {
        let path = std::path::Path::new(&dir).join(format!("scripted-{name}.json"));
        std::fs::write(&path, format!("{}\n", serde_json::to_string_pretty(body).unwrap())).unwrap();
    }
}

fn posted(requests: Vec<Request>) -> Value {
    body_json(requests.iter().rev().find(|r| r.method == Method::Post && !r.url.contains("iam.test")).expect("a submission"))
}

/// A built-in kind under a contract, through a service that follows its own: the
/// provider's estimate is unknown, so a ceiling refuses it and naming it runs it;
/// its result carries the planner's reasoning and a classical method's counts
/// beside the device's, and keeps every rule of section 8.7.
#[test]
fn a_built_in_kind_runs_under_a_contract_and_a_classical_method_shadows_it() {
    use wai_quantum::ed25519_dalek::SigningKey;
    use wai_quantum_qpu::job::{self, JobError, Target};
    use wai_quantum_qpu::registry::Kind;
    let target = || {
        let service = PrxService { jobs: Mutex::new(HashMap::new()) };
        let t = kept(move |r: &Request| service.answer(r));
        Target::new("lab", Kind::PrxCircuits, Box::new(PrxCircuits::new(t, Secret::new("prx-token"), "test-qc").with_base("https://prx.test/api/v1")), None)
    };
    let key = SigningKey::from_bytes(&[42; 32]);
    let bell = "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n";
    let wait = Wait { interval: std::time::Duration::ZERO, polls: 10 };
    // Its client states no price: an unknown cost does not meet a ceiling.
    let capped = json!({ "qasm": bell, "provider": "lab", "shots": 400, "contract": { "providers": "allowed", "max_cost": { "amount": "100", "currency": "USD" } } });
    let e = job::plan(&target(), &capped).err().unwrap().to_string();
    assert!(e.ends_with("[J071]") && e.contains("does not meet a ceiling"), "{e}");
    // Named, with the person's confirmation for an unknown cost, it runs.
    let req = json!({ "qasm": bell, "provider": "lab", "shots": 400, "seed": 5, "contract": { "providers": "allowed", "exactness": "measured" } });
    assert!(matches!(job::run(&target(), &req, false, &key, wait, |_| {}), Err(JobError::Unconfirmed { .. })));
    let body = job::run(&target(), &req, true, &key, wait, |_| {}).unwrap();
    let c = wai_quantum::engine::job::check(&body);
    assert!(c.valid(), "{:?}\n{}", c.reasons, body["result"]);
    let r = &body["result"];
    assert_eq!((r["accuracy"]["class"].as_str(), r["planner"]["choice"].as_str()), (Some("measured"), Some("provider:lab")));
    let route = r["planner"]["routes"].as_array().unwrap().iter().find(|x| x["route"] == "provider:lab").unwrap();
    assert_eq!(route["price"]["leaves_out"], json!(["time", "amount", "energy", "mitigation", "decoder", "charged"]), "{route}");
    assert!(route["price"]["money"]["unknown"].is_string(), "{route}");
    let s = &r["shadow"];
    assert_eq!((s["ran"].as_bool(), s["accuracy"]["class"].as_str()), (Some(true), Some("bit-exact")), "{s}");
    let total: u64 = s["counts"].as_object().unwrap().values().filter_map(Value::as_u64).sum();
    assert_eq!(total, 400);
    // The service follows its contract and the circuit is a Bell pair: both answers
    // are the same distribution, within what the shots leave.
    let d = &s["distance"];
    assert!(d["total_variation"].as_f64().unwrap() < 0.15, "{d}");
    if let Some(dir) = std::env::var_os("WAI_QUANTUM_RECEIPT_CORPUS") {
        std::fs::write(std::path::Path::new(&dir).join("scripted-contract-prx-circuits.json"), format!("{}\n", serde_json::to_string_pretty(&body).unwrap())).unwrap();
    }
}

/// A price the service bills by the minute is estimated at the least a task can
/// come to: under a contract's ceiling it is not known to be within it, and the
/// route says so; named with no ceiling, its cost is the person's to confirm.
#[test]
fn a_cost_estimated_at_the_least_it_can_come_to_meets_no_ceiling() {
    let target = || {
        let service = TasksService::new(true, false);
        let t = kept(move |r: &Request| service.answer(r));
        let keys = Keys::new("AKIDEXAMPLE", Secret::new("wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY"), None);
        let client = CloudTasks::new(t, keys, "us-west-1", "arn:test:simulator", "bucket").unwrap().with_base("https://tasks.test");
        Target::new("tasks", Kind::CloudTasks, Box::new(client), None)
    };
    let bell = "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n";
    let capped = json!({ "qasm": bell, "provider": "tasks", "shots": 100, "contract": { "providers": "allowed", "max_cost": { "amount": "100", "currency": "USD" } } });
    let e = job::plan(&target(), &capped).err().unwrap().to_string();
    assert!(e.ends_with("[J071]") && e.contains("the least it can come to"), "{e}");
    // Routed with every route in view, it is listed with its price and why.
    let routed = json!({ "qasm": bell, "shots": 100, "contract": { "exactness": "simulated", "providers": "allowed", "max_cost": { "amount": "100", "currency": "USD" } } });
    let e = job::route(&[target()], &routed, 1 << 30).err().unwrap().to_string();
    assert!(e.ends_with("[J071]") && e.contains("provider:tasks: at least 0.003750 USD"), "{e}");
    // Named, with no ceiling: planned, its estimate shown, and its route met.
    let named = json!({ "qasm": bell, "provider": "tasks", "shots": 100, "contract": { "providers": "allowed" } });
    let p = job::plan(&target(), &named).unwrap();
    let route = p.json["planner"]["routes"].as_array().unwrap().iter().find(|r| r["route"] == "provider:tasks").unwrap().clone();
    assert_eq!((route["outcome"].as_str(), route["price"]["money"]["at_least"].as_bool(), route["price"]["money"]["unit"].as_str()), (Some("chosen"), Some(true), Some("minute")), "{route}");
}

/// Each built-in kind, through a service that follows its contract, states
/// what its client sent (read from the very body it posts), where its counts
/// come from, and its class by its device's provenance; and its receipt keeps
/// every rule of wai/extensions/quantum-providers.md section 8.4.
#[test]
fn every_built_in_kind_states_what_it_sent_and_its_receipt_keeps_every_rule() {
    let service = SamplerV2Service::new();
    let t = kept(move |r: &Request| service.answer(r));
    let client = SamplerV2::new(t, Secret::new("sampler-key"), "crn:test", "test_device").with_endpoints("https://sampler.test/api/v1", "https://iam.test/identity/token").with_clock(|| 1_000);
    let body = sealed_run(Kind::SamplerV2, Box::new(client));
    holds_to_what_was_sent("sampler-v2", &body, &posted(t.requests()), "shots", "measured");
    let e = &body["result"]["execution"];
    assert_eq!((e["twirling"].as_str(), e["decoupling"].as_str(), e["mitigation"].as_str()), (Some("off"), Some("off"), Some("default")));

    let service = PrxService { jobs: Mutex::new(HashMap::new()) };
    let t = kept(move |r: &Request| service.answer(r));
    let body = sealed_run(Kind::PrxCircuits, Box::new(PrxCircuits::new(t, Secret::new("prx-token"), "test-qc").with_base("https://prx.test/api/v1")));
    holds_to_what_was_sent("prx-circuits-pinned", &body, &posted(t.requests()), "vendor_counts", "measured");
    // Pinned to the calibration it was compiled against, decoupling sent off,
    // and the service's other modes sent as they are.
    let sent = &body["result"]["execution"]["sent"];
    assert_eq!(sent["fields"], json!(["calibration_set_id", "circuits", "dd_mode", "heralding_mode", "move_gate_frame_tracking", "move_gate_validation", "shots"]));
    assert_eq!(sent["settings"], json!({ "/dd_mode": "disabled", "/heralding_mode": "none", "/move_gate_frame_tracking": "full", "/move_gate_validation": "strict" }));
    assert_eq!(body["result"]["execution"]["decoupling"], "off");
    assert_eq!(posted(t.requests())["calibration_set_id"], body["result"]["calibration"]);

    // Under a device's noise model the simulator answers shot records; this one
    // flips no reading, so a Bell pair's are 00 and 11 alone.
    for (name, noise, source) in [("native-turns", None, "probabilities"), ("native-turns-noise", Some("zz-model"), "shots")] {
        let service = TurnsService(fake_turns::Fake { flip: 0.0, ..fake_turns::Fake::new() }.with_backend("qpu.zz-model", "retired", 12, &["gpi", "gpi2", "zz"]));
        let t = kept(move |r: &Request| service.answer(r));
        let mut client = NativeTurns::new(t, Secret::new("turns-key"), "simulator").with_base("https://turns.test/v0.4");
        if let Some(n) = noise {
            client = client.with_noise(n);
        }
        let body = sealed_run(Kind::NativeTurns, Box::new(client));
        holds_to_what_was_sent(name, &body, &posted(t.requests()), source, "simulated");
        assert_eq!(body["result"]["execution"]["sent"]["settings"].get("/noise/model").and_then(Value::as_str), noise, "{name}");
        assert_eq!(body["result"]["execution"]["sent"]["settings"].get("/noise/seed").is_some(), noise.is_some(), "{name}");
    }

    for (name, simulator, probabilities, source, class) in [
        ("cloud-tasks-processor", false, false, "shots", "measured"),
        ("cloud-tasks-processor-probabilities", false, true, "probabilities", "measured"),
        ("cloud-tasks-simulator", true, false, "shots", "simulated"),
    ] {
        let service = TasksService::new(simulator, probabilities);
        let t = kept(move |r: &Request| service.answer(r));
        let keys = Keys::new("AKIDEXAMPLE", Secret::new("wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY"), None);
        let arn = if simulator { "arn:test:simulator" } else { "arn:test:processor" };
        let client = CloudTasks::new(t, keys, "us-west-1", arn, "bucket").unwrap().with_base("https://tasks.test");
        let body = sealed_run(Kind::CloudTasks, Box::new(client));
        let sent = posted(t.requests());
        holds_to_what_was_sent(name, &body, &sent, source, class);
        // A verbatim box is stated exactly when the program sent has one.
        let action: Value = serde_json::from_str(sent["action"].as_str().unwrap()).unwrap();
        let boxed = action["source"].as_str().unwrap().lines().any(|l| l.starts_with("#pragma") && l.ends_with("verbatim"));
        let e = &body["result"]["execution"];
        assert_eq!((boxed, e["verbatim"].as_bool(), e["sent"]["settings"].get("verbatim_box").is_some()), (!simulator, Some(!simulator), !simulator), "{name}");
        // The compiled program's text is in the receipt, beside its hash.
        let c = &body["result"]["compiled"];
        assert_eq!(c["program"].as_str().unwrap(), blake3::hash(c["qasm3"].as_str().unwrap().as_bytes()).to_hex().as_str());
    }
}
/// A device that answers fewer shots than it was asked for (a post-selected readout
/// discards the shots that fail its test): its counts are sealed as the shots it kept,
/// and the result says how many (`accepted_shots`, wai/extensions/quantum-providers.md
/// section 8.1); the receipt keeps every rule. With `WAI_QUANTUM_RECEIPT_CORPUS` set,
/// written to the corpus as `scripted-sampler-v2-post-selected.json`.
#[test]
fn a_device_that_keeps_fewer_shots_than_asked_seals_how_many_it_kept() {
    let mut service = SamplerV2Service::new();
    service.discard = 13;
    let t = kept(move |r: &Request| service.answer(r));
    let client = SamplerV2::new(t, Secret::new("sampler-key"), "crn:test", "test_device").with_endpoints("https://sampler.test/api/v1", "https://iam.test/identity/token").with_clock(|| 1_000);
    let body = sealed_run(Kind::SamplerV2, Box::new(client));
    holds_to_what_was_sent("sampler-v2-post-selected", &body, &posted(t.requests()), "shots", "measured");
    let r = &body["result"];
    assert_eq!((r["shots"].as_u64(), r["accepted_shots"].as_u64()), (Some(400), Some(387)), "{r}");
    let sum: u64 = r["counts"].as_object().unwrap().values().map(|v| v.as_u64().unwrap()).sum();
    assert_eq!(sum, 387);
    // A run that kept every shot says nothing of it, as every result before.
    let service = SamplerV2Service::new();
    let t = kept(move |r: &Request| service.answer(r));
    let client = SamplerV2::new(t, Secret::new("sampler-key"), "crn:test", "test_device").with_endpoints("https://sampler.test/api/v1", "https://iam.test/identity/token").with_clock(|| 1_000);
    assert!(sealed_run(Kind::SamplerV2, Box::new(client))["result"].get("accepted_shots").is_none());
}

fn task_keys(token: Option<&str>) -> Keys {
    Keys::new("AKIDEXAMPLE", Secret::new("wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY"), token.map(Secret::new))
}

/// A processor whose native gates are `prx`, `xx` and `rz`: every circuit
/// compiled to `prx rz xx`, each entangling gate the fully entangling
/// `xx(pi/2)` in the verbatim box, and its counts the circuit's.
#[test]
fn cloud_tasks_drive_a_processor_of_prx_xx_and_rz_in_its_own_gates() {
    let mut service = TasksService::new(false, false);
    service.xx = true;
    *service.labels.lock().unwrap() = (0..5).collect();
    let transport = Scripted::new(|r: &Request| service.answer(r));
    let client = CloudTasks::new(&transport, task_keys(None), "us-west-1", "arn:test:device", "results-bucket").unwrap().with_base("https://tasks.test");
    let s = no_wait(QpuSampler::new(client).unwrap());
    assert_eq!(s.device().basis, Basis::PrxRzXx);
    exercise(&s);
    let sources: Vec<String> = transport
        .requests()
        .iter()
        .filter(|r| r.url.ends_with("/quantum-task"))
        .map(|r| serde_json::from_str::<Value>(body_json(r)["action"].as_str().unwrap()).unwrap()["source"].as_str().unwrap().to_string())
        .collect();
    assert!(sources.iter().any(|s| s.contains("xx(1*pi/2) $")), "{sources:?}");
    assert!(sources.iter().all(|s| !s.contains("cz ") && s.contains("#pragma braket verbatim")), "{sources:?}"); // allow-name: the service's pragma
}

#[test]
fn cloud_tasks_round_trip_through_their_contract() {
    // Each bucket's name addresses its results: in the host, or, with a dot, in the path.
    for (simulator, probabilities, token, bucket) in [(false, false, None, "results-bucket"), (false, true, Some("session-token"), "results.dotted-bucket"), (true, false, None, "results.dotted-bucket"), (true, true, Some("session-token"), "results-bucket")] {
        let mut service = TasksService::new(simulator, probabilities);
        service.token = token;
        let transport = Scripted::new(|r: &Request| service.answer(r));
        let client = CloudTasks::new(&transport, task_keys(token), "us-west-1", "arn:test:device", bucket).unwrap().with_base("https://tasks.test");
        let s = no_wait(QpuSampler::new(client).unwrap());
        assert_eq!(s.device().provenance == Provenance::Silicon, !simulator);
        exercise(&s);
        // Every task walked through its states, and was read where its bucket's name puts it.
        let reads: Vec<String> = transport.requests().iter().filter(|r| r.url.ends_with("results.json")).map(|r| r.url.clone()).collect();
        assert!(!reads.is_empty() && reads.iter().all(|u| if bucket.contains('.') { u.starts_with(&format!("https://s3.us-west-1.amazonaws.com/{bucket}/")) } else { u.starts_with(&format!("https://{bucket}.s3.us-west-1.amazonaws.com/")) }), "{reads:?}"); // allow-name: the storage service's endpoint
        // The keys are signed with, and nothing secret is in anything sent.
        for r in transport.requests() {
            let all = format!("{} {:?} {}", r.url, r.headers, String::from_utf8_lossy(r.body.as_deref().unwrap_or_default()));
            assert!(!all.contains("EXAMPLEKEY"), "the secret went out in {}", r.url);
        }
    }
}

/// A request a person sends and confirms, on a device that may then change.
const TWO_QUBITS: &str = "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nx q[0];\nc[0] = measure q[0];\nc[1] = measure q[1];\n";

/// The job sent by one process and collected by a later one, with `between`
/// done to the device while it waited: what it sealed.
fn collected_later(tag: &str, between: impl Fn(&TasksService)) -> Value {
    let service: &'static TasksService = Box::leak(Box::new(TasksService::new(false, false)));
    let transport = kept(move |r: &Request| service.answer(r));
    let target = || {
        let client = CloudTasks::new(transport, task_keys(None), "us-west-1", "arn:test:device", "bucket").unwrap().with_base("https://tasks.test");
        Target::new("tasks", Kind::CloudTasks, Box::new(client), None)
    };
    let dir = std::env::temp_dir().join(format!("wai-qpu-clients-{tag}-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    let ledger = Ledger::at(&dir);
    let req = json!({ "qasm": TWO_QUBITS, "shots": 100, "provider": "tasks" });
    let sent = job::start(&target(), &req, true, &ledger).unwrap_or_else(|e| panic!("{e}"));
    between(service);
    let mut back = ledger.get(sent.handle()).unwrap();
    let key = wai_quantum::ed25519_dalek::SigningKey::from_bytes(&[42; 32]);
    let done = job::follow(&target(), &mut back, &key, &ledger, Wait { interval: std::time::Duration::ZERO, polls: 10 }, |_| {}).unwrap_or_else(|e| panic!("{e}"));
    let _ = std::fs::remove_dir_all(&dir);
    let Progress::Done(body) = done else { panic!("not done: {}", done.job()) };
    body
}

#[test]
fn a_job_collected_later_is_read_through_the_qubits_it_was_sent_on() {
    let unmoved = collected_later("unmoved", |_| {})["result"]["counts"].clone();
    // A qubit added below the ones it ran on moves every index after it.
    let added = collected_later("added", |s| *s.labels.lock().unwrap() = vec![2, 4, 7, 9, 12, 15]);
    assert_eq!(added["result"]["counts"], unmoved, "read through the labels of the day it was collected");
    // A qubit it ran on is gone from the device's list, and it still ran there.
    let gone = collected_later("gone", |s| *s.labels.lock().unwrap() = vec![7, 9, 12, 15]);
    assert_eq!(gone["result"]["counts"], unmoved);
    // The device no longer described at all: the record says what is needed.
    let unread = collected_later("unread", |s| s.device_gone.store(true, std::sync::atomic::Ordering::Relaxed));
    assert_eq!(unread["result"]["counts"], unmoved);
}

#[test]
fn a_simulator_s_estimate_says_it_is_only_the_minimum() {
    let service = TasksService::new(true, false);
    let transport = kept(move |r: &Request| service.answer(r));
    let client = CloudTasks::new(transport, task_keys(None), "us-west-1", "arn:test:device", "bucket").unwrap().with_base("https://tasks.test");
    let t = Target::new("tasks", Kind::CloudTasks, Box::new(client), None);
    let p = job::plan(&t, &json!({ "qasm": TWO_QUBITS, "shots": 100, "provider": "tasks" })).unwrap_or_else(|e| panic!("{e}"));
    assert!(p.confirmation().contains("at least 0.003750 USD"), "{}", p.confirmation());
    assert_eq!(p.json["estimate"]["at_least"], true, "{}", p.json["estimate"]);
}

/// A job recorded by the last release, and collected by this one: as that
/// release wrote its record, with what it did not yet write taken out. Its
/// receipt keeps every rule, and says no one stated what was sent.
#[test]
fn a_job_recorded_by_the_last_release_is_sealed_with_a_receipt_this_one_accepts() {
    fn on(tag: &str, provider: Box<dyn Provider>, again: Box<dyn Provider>, kind: Kind, class: &str) {
        let dir = std::env::temp_dir().join(format!("wai-qpu-clients-recorded-{tag}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let ledger = Ledger::at(&dir);
        let req = json!({ "qasm": TWO_QUBITS, "shots": 100, "provider": "mine" });
        let sent = job::start(&Target::new("mine", kind, provider, None), &req, true, &ledger).unwrap_or_else(|e| panic!("{tag}: {e}"));
        let path = dir.join(format!("{}.json", sent.handle()));
        let mut record: Value = serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
        // What 0.4.1 wrote: wai/wai-quantum-qpu/src/job.rs at the release's commit.
        for (at, member) in [("sealing", "execution"), ("sealing", "approximate"), ("sealing/compiled", "qasm3"), ("sealing/compiled", "qubit_names"), ("", "qubit_names")] {
            let o = if at.is_empty() { record.as_object_mut() } else { record.pointer_mut(&format!("/{at}")).and_then(Value::as_object_mut) };
            o.unwrap().shift_remove(member);
        }
        std::fs::write(&path, record.to_string()).unwrap();
        let mut back = ledger.get(sent.handle()).unwrap();
        let key = wai_quantum::ed25519_dalek::SigningKey::from_bytes(&[42; 32]);
        let done = job::follow(&Target::new("mine", kind, again, None), &mut back, &key, &ledger, Wait { interval: std::time::Duration::ZERO, polls: 10 }, |_| {}).unwrap_or_else(|e| panic!("{tag}: {e}"));
        let _ = std::fs::remove_dir_all(&dir);
        let Progress::Done(body) = done else { panic!("{tag}: not done: {}", done.job()) };
        let c = wai_quantum::engine::job::check(&body);
        assert_eq!(c.reasons, Vec::<&str>::new(), "{tag}: {}", body["result"]);
        let r = &body["result"];
        assert_eq!(r["execution"]["stated_by"], "nobody", "{tag}");
        assert_eq!(r["accuracy"]["class"], class, "{tag}");
        assert_eq!(r["compiled"]["program"].as_str().unwrap(), blake3::hash(r["compiled"]["qasm3"].as_str().unwrap().as_bytes()).to_hex().as_str(), "{tag}");
    }
    let service: &'static SamplerV2Service = Box::leak(Box::new(SamplerV2Service::new()));
    let t = kept(move |r: &Request| service.answer(r));
    let client = || -> Box<dyn Provider> { Box::new(SamplerV2::new(t, Secret::new("sampler-key"), "crn:test", "test_device").with_endpoints("https://sampler.test/api/v1", "https://iam.test/identity/token").with_clock(|| 1_000)) };
    on("sampler", client(), client(), Kind::SamplerV2, "measured");
    let service: &'static PrxService = Box::leak(Box::new(PrxService { jobs: Mutex::new(HashMap::new()) }));
    let t = kept(move |r: &Request| service.answer(r));
    let client = || -> Box<dyn Provider> { Box::new(PrxCircuits::new(t, Secret::new("prx-token"), "test-qc").with_base("https://prx.test/api/v1")) };
    on("prx", client(), client(), Kind::PrxCircuits, "measured");
    let service: &'static TurnsService = Box::leak(Box::new(TurnsService::new()));
    let t = kept(move |r: &Request| service.answer(r));
    let client = || -> Box<dyn Provider> { Box::new(NativeTurns::new(t, Secret::new("turns-key"), "simulator").with_base("https://turns.test/v0.4")) };
    on("turns", client(), client(), Kind::NativeTurns, "simulated");
}

/// A program is sealed on indices, and the receipt says which of the device's
/// qubits each one was on the day it was sent, as the body sent names them.
#[test]
fn a_receipt_names_the_device_s_qubits_as_the_program_was_sent_on_them() {
    let service = TasksService::new(false, false);
    let t = kept(move |r: &Request| service.answer(r));
    let client = CloudTasks::new(t, task_keys(None), "us-west-1", "arn:test:processor", "bucket").unwrap().with_base("https://tasks.test");
    let body = sealed_run(Kind::CloudTasks, Box::new(client));
    let names = body["result"]["compiled"]["qubit_names"].as_object().unwrap_or_else(|| panic!("{}", body["result"]["compiled"]));
    let action: Value = serde_json::from_str(posted(t.requests())["action"].as_str().unwrap()).unwrap();
    let source = action["source"].as_str().unwrap().to_string();
    for q in body["result"]["compiled"]["final_layout"].as_array().unwrap() {
        let name = names[&q.to_string()].as_str().unwrap();
        assert!(source.contains(&format!("measure ${name};")), "{name} in {source}");
    }

    let service = PrxService { jobs: Mutex::new(HashMap::new()) };
    let t = kept(move |r: &Request| service.answer(r));
    let body = sealed_run(Kind::PrxCircuits, Box::new(PrxCircuits::new(t, Secret::new("prx-token"), "test-qc").with_base("https://prx.test/api/v1")));
    let names = body["result"]["compiled"]["qubit_names"].as_object().unwrap_or_else(|| panic!("{}", body["result"]["compiled"]));
    let sent = posted(t.requests()).to_string();
    for q in body["result"]["compiled"]["final_layout"].as_array().unwrap() {
        let name = names[&q.to_string()].as_str().unwrap();
        assert!(sent.contains(name), "{name} in {sent}");
    }
}

/// A task cancelled while it waits ends cancelled, through the service's two
/// steps, and its record says so.
#[test]
fn a_cloud_task_cancelled_while_it_waits_ends_cancelled() {
    let service: &'static TasksService = Box::leak(Box::new(TasksService::new(false, false)));
    let transport = kept(move |r: &Request| service.answer(r));
    let client = CloudTasks::new(transport, task_keys(None), "us-west-1", "arn:test:device", "results-bucket").unwrap().with_base("https://tasks.test");
    let t = Target::new("tasks", Kind::CloudTasks, Box::new(client), None);
    let dir = std::env::temp_dir().join(format!("wai-qpu-clients-cancel-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    let ledger = Ledger::at(&dir);
    let mut ticket = job::start(&t, &json!({ "qasm": TWO_QUBITS, "shots": 100, "provider": "tasks" }), true, &ledger).unwrap_or_else(|e| panic!("{e}"));
    let key = wai_quantum::ed25519_dalek::SigningKey::from_bytes(&[42; 32]);
    let ended = job::cancel(&t, &mut ticket, &key, &ledger).unwrap_or_else(|e| panic!("{e}"));
    assert_eq!(ended.job()["status"], "cancelled", "{}", ended.job());
    // The service's own word, asked again: cancelling, then cancelled.
    let id = TasksService::task_id(0);
    assert_eq!(t.provider.status(&id).unwrap(), wai_quantum_qpu::JobStatus::Ended { status: "cancelled".into(), reason: String::new() });
    let _ = std::fs::remove_dir_all(&dir);
}