//! The job service's verify, over HTTP and as an MCP tool, on the corpus of
//! job receipts: the verdict the manifest gives, every rule of the provider
//! protocol's section 8.4 and not the seal alone; and the sign-off loop's routes,
//! whose grades and sign-offs that verify works out again.

use serde_json::{Value, json};
use std::io::{Read, Write};
use std::net::TcpStream;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};

mod common;

use common::Running;

/// One HTTP/1.1 exchange, the body as bytes: the status and the body.
fn http(port: u16, method: &str, path: &str, body: &[u8]) -> (u16, String) {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(60))).unwrap();
    write!(s, "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nAccept: application/json, text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len()).unwrap();
    s.write_all(body).unwrap();
    let mut out = String::new();
    s.read_to_string(&mut out).unwrap();
    let status = out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0);
    (status, out.split_once("\r\n\r\n").map(|(_, b)| b.to_string()).unwrap_or_default())
}

/// [`http`] of a body the service stops reading: written from another thread, so that the
/// answer is read whenever it comes, and a connection the service closes on the rest of the
/// body is no failure of the exchange.
fn http_past_its_limit(port: u16, path: &str, body: Vec<u8>) -> (u16, String) {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(120))).unwrap();
    let mut w = s.try_clone().unwrap();
    let head = format!("POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len());
    let writer = std::thread::spawn(move || {
        let _ = w.write_all(head.as_bytes()).and_then(|()| w.write_all(&body));
    });
    let mut out = Vec::new();
    let mut buf = [0u8; 1 << 16];
    while let Ok(n) = s.read(&mut buf) {
        if n == 0 {
            break;
        }
        out.extend_from_slice(&buf[..n]);
    }
    let _ = writer.join();
    let out = String::from_utf8_lossy(&out).into_owned();
    let status = out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0);
    (status, out.split_once("\r\n\r\n").map(|(_, b)| b.to_string()).unwrap_or_default())
}

/// The service over the store at `dir`, once it answers: its process and the port it
/// bound. It binds a port the system chooses and says which in its log, as
/// tests/contract.rs's services do: a port found free here and handed to it raced, the
/// listener that found it inherited, until its exec, by a child another test forked at that
/// moment, so a test's first connection could reach that copy and be reset before the
/// service had bound the port (tests/write.rs on CI, run 28969).
fn start(dir: &Path) -> (Running, u16) {
    let path = dir.join("service.log");
    let mut child = Running::new(Command::new(env!("CARGO_BIN_EXE_wai-quantum-service"))
        .arg("serve")
        .env("WAI_QSVC_ADDR", "127.0.0.1:0")
        .env("WAI_QSVC_STORE", dir.join("store"))
        .env("WAI_QSVC_KEY", dir.join("signing.key"))
        .env("WAI_QSVC_JOB_GB", "2")
        .env("WAI_QSVC_BUDGET_GB", "4")
        .env("WAI_QSVC_WORKERS", "1")
        .env("WAI_QSVC_ENERGY", "0")
        // A set's receipt counts, as a check does: the corpus holds more than a window's.
        .env("WAI_QSVC_CLIENT_BURST", "100000")
        .env("WAI_QSVC_CLIENT_VERIFIES", "100000")
        .stdout(Stdio::null())
        .stderr(std::fs::File::create(&path).expect("the service's log"))
        .spawn()
        .expect("the service starts"));
    let log = || std::fs::read_to_string(&path).unwrap_or_default();
    let t = Instant::now();
    let port = loop {
        if let Some(p) = log().lines().rev().find_map(|l| l.split(" on 127.0.0.1:").nth(1)?.split(',').next()?.trim().parse::<u16>().ok()) {
            break p;
        }
        if let Ok(Some(status)) = child.try_wait() {
            panic!("the service ended as it started ({status}): {}", log());
        }
        assert!(t.elapsed() < Duration::from_secs(60), "the service did not start in 60 s: {}", log());
        std::thread::sleep(Duration::from_millis(50));
    };
    while TcpStream::connect(("127.0.0.1", port)).is_err() || http(port, "GET", "/v1/health", b"").0 != 200 {
        if let Ok(Some(status)) = child.try_wait() {
            panic!("the service ended ({status}): {}", log());
        }
        assert!(t.elapsed() < Duration::from_secs(60), "the service did not answer: {}", log());
        std::thread::sleep(Duration::from_millis(50));
    }
    (child, port)
}

fn corpus() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../quantum-conformance/receipts")
}

fn sorted(v: &Value) -> Vec<String> {
    let mut r: Vec<String> = v.as_array().map(|a| a.iter().filter_map(Value::as_str).map(str::to_string).collect()).unwrap_or_default();
    r.sort();
    r
}

#[test]
fn the_service_gives_every_receipt_the_verdict_the_corpus_gives_it() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-verify-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (mut child, port) = start(&dir);
    let manifest: Value = serde_json::from_str(&std::fs::read_to_string(corpus().join("manifest.json")).unwrap()).unwrap();
    let mut wrong = Vec::new();
    for e in manifest["entries"].as_array().unwrap() {
        let name = e["file"].as_str().unwrap();
        let want = sorted(&e["reasons"]);
        let bytes = std::fs::read(corpus().join(name)).unwrap();
        // Over HTTP: its text, as it is. A body that is no job at all has its
        // own refusal, 400, as it always has.
        let (code, answer) = http(port, "POST", "/v1/verify", &bytes);
        let alone = want.len() == 1 && ["unreadable", "not_a_job", "no_receipt"].contains(&want[0].as_str());
        let said: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
        if !(alone && code == 400) && !(code == 200 && sorted(&said["reasons"]) == want && said["valid"].as_bool() == Some(want.is_empty())) {
            wrong.push(format!("{name}: over HTTP, the manifest {want:?}, the service {code} {answer}"));
        }
        // As an MCP tool, given the job's text: every receipt whose bytes a string can hold
        // (a tool's argument is a string, UTF-8 by its form).
        let Ok(text) = String::from_utf8(bytes.clone()) else { continue };
        let call = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "verify_quantum_receipt", "arguments": { "job": text } } });
        let (_, answer) = http(port, "POST", "/mcp", call.to_string().as_bytes());
        let said: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
        let s = &said["result"]["structuredContent"];
        let _ = alone;
        if sorted(&s["reasons"]) != want || s["valid"].as_bool() != Some(want.is_empty()) || s["given"] != "text" {
            wrong.push(format!("{name}: as a tool, the manifest {want:?}, the service {s}"));
        }
        // As an MCP tool, given the job as members (the job as an object, and the call's
        // arguments as the job's own members), each written into the call as the corpus
        // writes it: judged as the message writes it, so a member named twice and a whole
        // number past a double are seen. A job that is no JSON value makes a message that
        // is none, which is the protocol's error.
        for (form, arguments) in [("an object", format!("{{\"job\": {text}}}")), ("its members", text.clone())] {
            // A call's arguments are an object: a body that is none is given as its text, or as `job`.
            if form == "its members" && !text.trim_start().starts_with('{') {
                continue;
            }
            let call = format!("{{\"jsonrpc\": \"2.0\", \"id\": 1, \"method\": \"tools/call\", \"params\": {{\"name\": \"verify_quantum_receipt\", \"arguments\": {arguments}}}}}");
            let (_, answer) = http(port, "POST", "/mcp", call.as_bytes());
            let said: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
            let s = &said["result"]["structuredContent"];
            let no_message = !s.is_object() && want == ["unreadable"];
            if !no_message && (sorted(&s["reasons"]) != want || s["valid"].as_bool() != Some(want.is_empty()) || s["given"] != "members") {
                wrong.push(format!("{name}: as a tool, given {form}, the manifest {want:?}, the service {s} ({})", answer.chars().take(300).collect::<String>()));
            }
            if name.contains("read-duplicate") && s["reasons"] != json!(["unreadable"]) {
                wrong.push(format!("{name}: as a tool, given {form}, a member named twice was not seen: {s}"));
            }
        }
    }
    let _ = child.kill();
    let _ = child.wait();
    let _ = std::fs::remove_dir_all(&dir);
    assert!(wrong.is_empty(), "{} wrong:\n{}", wrong.len(), wrong.join("\n"));
}

/// Text that does not read as a job's, a body that is no job, and a result with no receipt
/// each get a verdict with that reason alone (§8.4), as every surface gives it, not a refusal
/// of the request. As a tool, given the job's members, the host's unsealed ones among them, a
/// wasmQ job's unsealed status is noted where it differs from what is sealed.
#[test]
fn every_body_gets_a_verdict_and_the_tool_reads_the_body_it_is_given() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-verdicts-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (mut child, port) = start(&dir);
    let job = std::fs::read(corpus().join("jobs/cli-wasmq-fault.json")).unwrap();
    let mut bom = vec![0xef, 0xbb, 0xbf];
    bom.extend_from_slice(&job);
    let mut problems = Vec::new();
    for (what, body, want) in [
        ("a byte order mark", bom, "unreadable"),
        ("not JSON", b"not json".to_vec(), "unreadable"),
        ("not UTF-8", vec![b'{', 0xff, b'}'], "unreadable"),
        ("no job", br#"{"request": {}}"#.to_vec(), "not_a_job"),
        ("no receipt", br#"{"request": {}, "result": {"counts": {}}}"#.to_vec(), "no_receipt"),
    ] {
        let (code, answer) = http(port, "POST", "/v1/verify", &body);
        let said: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
        if code != 200 || said["reasons"] != json!([want]) || said["valid"] != false {
            problems.push(format!("{what}: {code} {answer}"));
        }
    }
    let parsed: Value = serde_json::from_slice(&job).unwrap();
    let args = json!({ "status": "done", "method": parsed["method"], "request": parsed["request"], "result": parsed["result"] });
    let call = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "verify_quantum_receipt", "arguments": args } });
    let (_, answer) = http(port, "POST", "/mcp", call.to_string().as_bytes());
    let said: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
    let s = &said["result"]["structuredContent"];
    if s["valid"] != true || s["wasmq"]["notes"] != json!([r#"the host's unsealed status says "done"; the sealed result says "fault""#]) {
        problems.push(format!("the tool, given the body: {s}"));
    }
    let _ = child.kill();
    let _ = child.wait();
    let _ = std::fs::remove_dir_all(&dir);
    assert!(problems.is_empty(), "{}", problems.join("\n"));
}

/// One answer to every form of a job (the provider protocol's section 8.4): an estimate or
/// an algorithm's answer with no receipt, or that is no job, is that reason alone over HTTP
/// and from the tool, given its text, the job as an object or its members; a forged
/// estimate sealed again is run again and is not valid however it is given; a genuine job
/// is valid given as an object; a job past 1 MiB is read, and one past what verify reads is
/// `too_large`, a verdict like any other.
#[test]
fn one_answer_to_every_form_of_a_job() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-one-answer-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (mut child, port) = start(&dir);
    let estimate = wai_quantum::engine::estimate::FORMAT;
    let algorithms = wai_quantum::engine::schema::ALGORITHMS;
    let key = wai_quantum::ed25519_dalek::SigningKey::from_bytes(&[9; 32]);
    let angles = "OPENQASM 3.0;\nqubit[4] q;\nh q;\nrz(0.3) q[0];\nccx q[0], q[1], q[2];\nry(1.1) q[1];\nt q[2];\nctrl(3) @ x q[0], q[1], q[2], q[3];\n";
    let made = wai_quantum::engine::estimate::sealed(&json!({ "qasm": angles, "budget": 1e-3, "max_slowdown": 64 }), &key).unwrap();
    let mut forged = made.clone();
    forged["result"]["estimate"]["physical_qubits"] = json!(1);
    forged["result"].as_object_mut().unwrap().remove("receipt");
    wai_quantum::engine::receipt::seal_into(&key, &forged["request"].clone(), &mut forged["result"]);
    let tool = |args: Value| -> Value {
        let call = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "verify_quantum_receipt", "arguments": args } });
        let (_, answer) = http(port, "POST", "/mcp", call.to_string().as_bytes());
        let said: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
        said["result"]["structuredContent"].clone()
    };
    let mut problems = Vec::new();
    let mut each = |what: &str, body: &Value, ok: &dyn Fn(&Value) -> bool| {
        let (code, answer) = http(port, "POST", "/v1/verify", body.to_string().as_bytes());
        let over_http: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
        let mut members = body.as_object().cloned().unwrap_or_default();
        members.remove("job");
        for (form, said) in [("POST /v1/verify", if code == 200 { over_http } else { json!({ "status": code, "body": answer }) }), ("the tool, its text", tool(json!({ "job": body.to_string() }))), ("the tool, the job as an object", tool(json!({ "job": body }))), ("the tool, its members", tool(Value::Object(members)))] {
            if !ok(&said) {
                problems.push(format!("{what}, {form}: {said}"));
            }
        }
    };
    let alone = |want: &'static str| move |v: &Value| v["valid"] == false && v["reasons"] == json!([want]);
    each("an estimate with no receipt", &json!({ "request": { "qasm": "OPENQASM 3.0;\nqubit q;\nh q;\n" }, "result": { "format": estimate } }), &alone("no_receipt"));
    each("an estimate with no receipt and no program", &json!({ "request": {}, "result": { "format": estimate } }), &alone("no_receipt"));
    each("an estimate that is no job", &json!({ "request": 5, "result": { "format": estimate } }), &alone("not_a_job"));
    each("an algorithm's answer with no receipt", &json!({ "request": { "qasm": "OPENQASM 3.0;\nqubit q;\n" }, "result": { "schema": algorithms, "call": "gradient" } }), &alone("no_receipt"));
    each("an algorithm's answer that is no job", &json!({ "request": 5, "result": { "schema": algorithms } }), &alone("not_a_job"));
    each("a genuine estimate", &made, &|v: &Value| v["valid"] == true && v["reproduced"] == true);
    each("a forged estimate, sealed again", &forged, &|v: &Value| v["valid"] == false && v["reproduced"] == false && v["signed"] == true);
    let job: Value = serde_json::from_slice(&std::fs::read(corpus().join("jobs/cli-wasmq-run.json")).unwrap()).unwrap();
    each("a genuine wasmQ job", &job, &|v: &Value| v["valid"] == true);
    // A job of 1.5 MiB (a member nothing seals holds the rest) is read and judged.
    let mut big = job.clone();
    big["pad"] = json!("x".repeat(3 << 19));
    each("a genuine job of 1.5 MiB", &big, &|v: &Value| v["valid"] == true);
    // A request no host runs again, sealed: an estimate's verdict says so, as an algorithm's
    // answer's does.
    let reseal = |mut body: Value| {
        body["result"].as_object_mut().unwrap().remove("receipt");
        wai_quantum::engine::receipt::seal_into(&key, &body["request"].clone(), &mut body["result"]);
        body
    };
    let unrunnable = reseal(json!({ "request": {}, "result": made["result"].clone() }));
    each("an estimate whose request no estimate runs", &unrunnable, &|v: &Value| v["valid"] == false && v["reproduced"] == false && v["rerun_error"].is_string());
    // An algorithm's answer whose program nests past the reader's bound (a flat sum of 2000
    // terms) gets its verdict, and the service stays up.
    let gradient = json!({ "qasm": "OPENQASM 3.0;\ninput float t;\nqubit q;\nrx(t) q;\n", "params": { "t": 0.4 }, "hamiltonian": ["Z0"] });
    let answered = wai_quantum::engine::variational::seal(&gradient, wai_quantum::engine::variational::call("gradient", &gradient).unwrap(), &key);
    each("a genuine gradient", &answered, &|v: &Value| v["valid"] == true && v["rerun"] == true);
    let mut deep = answered.clone();
    // Nested 1000 deep in brackets, past the reader's bound (a flat sum is read, at any length).
    deep["request"]["qasm"] = json!(format!("OPENQASM 3.0;\ninput float t;\nqubit q;\nint x = {}1{};\nrx(t) q;\n", "(".repeat(1000), ")".repeat(1000)));
    each("a gradient of a program past the nesting bound", &reseal(deep), &|v: &Value| v["valid"] == false && v["rerun"] == false && v["rerun_error"].as_str().is_some_and(|e| e.contains("Q006")));
    let up = http(port, "GET", "/v1/health", b"").0 == 200;
    // Past what verify reads: too_large, with what it reads, on every surface. Past the bytes:
    let at_most = json!({ "bytes": 4 << 20, "values": 1 << 19 });
    let too_large = |v: &Value| v["valid"] == false && v["reasons"] == json!(["too_large"]) && v["read_at_most"] == json!({ "bytes": 4 << 20, "values": 1 << 19 });
    let padded = |n: usize| {
        let mut j = job.clone();
        j["pad"] = json!("");
        let rest = j.to_string().len();
        j["pad"] = json!("x".repeat(n - rest));
        j
    };
    each("a job of 4 MiB", &padded(4 << 20), &|v: &Value| v["valid"] == true);
    each("a job of 4 MiB and a byte", &padded((4 << 20) + 1), &too_large);
    // a job of quotes, 3 MiB of text, which a string of it doubles again, within them:
    let mut quoted = job.clone();
    quoted["pad"] = json!("\"".repeat(3 << 19));
    each("a job of 3 MiB of escaped quotes", &quoted, &|v: &Value| v["valid"] == true);
    // past the values, in a few bytes each:
    let mut many = job.clone();
    many["pad"] = json!(vec![0; 1 << 19]);
    each("a job of 2^19 values more", &many, &too_large);
    if !up {
        problems.push("the service did not stay up".into());
    }
    // and past what /mcp reads of a message: its verdict by its id, too_large.
    let mut huge = job.clone();
    huge["pad"] = json!("x".repeat(30 << 20));
    let (code, answer) = http_past_its_limit(port, "/v1/verify", huge.to_string().into_bytes());
    let said: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
    if code != 200 || said["reasons"] != json!(["too_large"]) || said["read_at_most"] != at_most {
        problems.push(format!("a job of 30 MiB: {code} {}", &answer[..answer.len().min(300)]));
    }
    let call = json!({ "jsonrpc": "2.0", "id": 77, "method": "tools/call", "params": { "name": "verify_quantum_receipt", "arguments": { "job": huge } } });
    let (code, answer) = http_past_its_limit(port, "/mcp", call.to_string().into_bytes());
    let said: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
    if code != 200 || said["id"] != 77 || !too_large(&said["result"]["structuredContent"]) {
        problems.push(format!("a call of 30 MiB: {code} {}", &answer[..answer.len().min(300)]));
    }
    // A job nested past what a reader takes, as an object or as members: unreadable, by its id.
    for d in [127usize, 128, 140, 1000] {
        let raw = format!("{{\"request\":{},\"result\":{},\"pad\":{}{}}}", job["request"], job["result"], "[".repeat(d - 1), "]".repeat(d - 1));
        for (form, args) in [("as an object", format!("{{\"job\":{raw}}}")), ("as members", raw.clone()), ("as text", json!({ "job": raw }).to_string())] {
            let call = format!("{{\"jsonrpc\":\"2.0\",\"id\":5,\"method\":\"tools/call\",\"params\":{{\"name\":\"verify_quantum_receipt\",\"arguments\":{args}}}}}");
            let (code, answer) = http(port, "POST", "/mcp", call.as_bytes());
            let said: Value = serde_json::from_str(&answer).unwrap_or(Value::Null);
            let s = &said["result"]["structuredContent"];
            let want = if d <= 127 { s["valid"] == true } else { s["reasons"] == json!(["unreadable"]) };
            if code != 200 || said["id"] != 5 || !want {
                problems.push(format!("a job {d} deep, {form}: {code} {}", &answer[..answer.len().min(300)]));
            }
        }
    }
    let _ = child.kill();
    let _ = child.wait();
    let _ = std::fs::remove_dir_all(&dir);
    assert!(problems.is_empty(), "{}", problems.join("\n"));
}

/// [`http`], or nothing when the service does not answer.
fn try_http(port: u16, method: &str, path: &str, body: &[u8]) -> Option<(u16, String)> {
    let mut s = TcpStream::connect(("127.0.0.1", port)).ok()?;
    s.set_read_timeout(Some(Duration::from_secs(60))).ok()?;
    write!(s, "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len()).ok()?;
    s.write_all(body).ok()?;
    let mut out = Vec::new();
    s.read_to_end(&mut out).ok()?;
    let out = String::from_utf8_lossy(&out).into_owned();
    let status = out.split_whitespace().nth(1)?.parse().ok()?;
    Some((status, out.split_once("\r\n\r\n").map(|(_, b)| b.to_string()).unwrap_or_default()))
}

/// A program nested past the reader's bound (Q006) is refused on every route that reads one,
/// a flat sum of 2000 terms (one level, however long) is read, and the service stays up: the
/// sum, 4 KB, used to overflow a worker's stack and end the service, on the routes that read
/// a program on an async worker.
#[test]
fn a_program_past_the_nesting_bound_is_refused_and_the_service_stays_up() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-deep-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (mut child, mut port) = start(&dir);
    let head = "OPENQASM 3.0;\ninclude \"stdgates.inc\";\nqubit[2] q;\nbit[2] c;\n";
    let programs = [
        ("a sum of 2000 terms", format!("{head}int x = 1{};\nh q[0];\nc = measure q;\n", "+1".repeat(1999))),
        ("1000 brackets", format!("{head}rz({}0.5{}) q[0];\nc = measure q;\n", "(".repeat(1000), ")".repeat(1000))),
        ("1000 blocks", format!("{head}{}h q[0];{}\nc = measure q;\n", "{ ".repeat(1000), " }".repeat(1000))),
    ];
    let mut wrong = Vec::new();
    for (what, p) in &programs {
        let bodies = [
            ("/v1/jobs", json!({ "qasm": p, "shots": 1, "seed": 1 })),
            ("/v1/run", json!({ "qasm": p, "shots": 1, "seed": 1 })),
            ("/v1/estimate", json!({ "qasm": p })),
            ("/v1/gradient", json!({ "qasm": p, "observables": ["Z0"] })),
            ("/v1/critique", json!({ "qasm": p })),
            ("/v1/compile", json!({ "qasm": p })),
            ("/v1/verify", json!({ "request": { "qasm": p }, "result": { "format": "wai.quantum.estimate/1" } })),
        ];
        for (path, body) in bodies {
            let (code, answer) = try_http(port, "POST", path, body.to_string().as_bytes()).unwrap_or((0, String::new()));
            let up = try_http(port, "GET", "/v1/health", b"").is_some_and(|(code, _)| code == 200);
            eprintln!("{what} at {path}: {code} {}", answer.chars().take(160).collect::<String>());
            if !up {
                wrong.push(format!("{what} at {path}: the service ended ({code} {answer})"));
                child.stop();
                (child, port) = start(&dir);
                continue;
            }
            // Read at once, a nested one is refused by its code; queued, the job fails by it.
            // The sum is flat, and read.
            let nested = *what != "a sum of 2000 terms";
            if nested && (200..300).contains(&code) && !answer.contains("Q006") && path != "/v1/jobs" && path != "/v1/verify" {
                wrong.push(format!("{what} at {path}: {code} without Q006: {}", answer.chars().take(300).collect::<String>()));
            }
            if !nested && (answer.contains("Q006") || code >= 500) {
                wrong.push(format!("{what} at {path}: {code}: {}", answer.chars().take(300).collect::<String>()));
            }
        }
    }
    let _ = child.kill();
    let _ = child.wait();
    let _ = std::fs::remove_dir_all(&dir);
    assert!(wrong.is_empty(), "{} wrong:\n{}", wrong.len(), wrong.join("\n"));
}

/// An energy-landscape job over HTTP (wai/extensions/quantum-landscape.md): each operation
/// answered on the extension's reference, sealed with the service's key, and verified by the
/// service's own verify with its form held (§5.4); a program the reader refuses is refused
/// with its line, and a request that names a provider is refused (J002), as the service
/// reaches none.
#[test]
fn an_energy_landscape_job_is_sealed_and_verified_by_its_form() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-landscape-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (mut child, port) = start(&dir);
    let program = "vars 4\ncost 1\nagree 0 1\nagree 0 2\nagree 1 2\noppose 2 3\nagree 3 0\nsolve restarts=2 steps=9\n";
    for (op, request) in [
        ("run", json!({ "program": program, "seed": "7" })),
        ("replay", json!({ "program": program, "seeds": 3 })),
        ("sweep", json!({ "program": program, "seeds": 2, "temperatures": [0, 1.5] })),
        ("compare", json!({ "program": program, "seeds": 2 })),
    ] {
        let (code, body) = http(port, "POST", &format!("/v1/landscape/{op}"), request.to_string().as_bytes());
        assert_eq!(code, 200, "{op}: {body}");
        let job: Value = serde_json::from_str(&body).unwrap();
        assert_eq!(job["result"]["schema"], "wai.quantum.landscape/1", "{op}");
        let c = wai_quantum::engine::job::check(&job);
        assert!(c.valid(), "{op}: {:?} {:?}", c.reasons, c.landscape);
        let (code, said) = http(port, "POST", "/v1/verify", body.as_bytes());
        let said: Value = serde_json::from_str(&said).unwrap();
        assert!(code == 200 && said["valid"] == true, "{op}: {said}");
        assert_eq!(said["landscape"]["op"], op, "{said}");
        // The same answer from another member, as a host would never seal it: its form is broken.
        let mut forged = job.clone();
        forged["result"]["member"]["parity"] = json!("bit-exact");
        let (_, said) = http(port, "POST", "/v1/verify", forged.to_string().as_bytes());
        let said: Value = serde_json::from_str(&said).unwrap();
        assert_eq!(said["valid"], false, "{said}");
    }
    let (code, body) = http(port, "POST", "/v1/landscape/run", json!({ "program": "vars 2\noppos 0 1\nsolve\n" }).to_string().as_bytes());
    assert!(code == 422 && body.contains("Line 2: "), "{code} {body}");
    let (code, body) = http(port, "POST", "/v1/landscape/run", json!({ "program": program, "provider": "lab" }).to_string().as_bytes());
    assert!(code == 422 && body.contains("J002"), "{code} {body}");
    let (code, _) = http(port, "POST", "/v1/landscape/examples", b"{}");
    assert_eq!(code, 404);
    let _ = child.kill();
    let _ = child.wait();
    let _ = std::fs::remove_dir_all(&dir);
}

/// The sign-off loop over HTTP, made in the service's reading child: a check that names
/// each case's layer, a grade sealed with the service's key that its own verify makes again,
/// a task that would publish its answer refused, a job another key sealed attested against a
/// trust list that names that key and the attestation verified, refused against a list that
/// does not, and a grade whose wide case is past the service's bounds refused at that case.
#[test]
fn the_sign_off_loop_answers_over_http_and_its_grade_and_sign_off_verify() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-loop-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (mut child, port) = start(&dir);
    let post = |path: &str, body: &Value| {
        let (code, text) = http(port, "POST", path, body.to_string().as_bytes());
        (code, serde_json::from_str::<Value>(&text).unwrap_or_else(|_| json!(text)))
    };
    let pair = "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n";
    let fair = json!({ "distribution": { "00": 0.5, "11": 0.5 } });
    let (code, checked) = post("/v1/check", &json!({ "qasm": pair, "intent": fair }));
    assert_eq!((code, checked["passed"].as_bool(), checked["grade"]["cases"][0]["layer"].as_str()), (200, Some(true), Some("exact_simulation")), "{checked}");
    let (code, graded) = post("/v1/grade", &json!({ "grade": { "task": { "title": "A Bell pair", "intent": fair }, "program": pair } }));
    assert_eq!((code, graded["result"]["passed"].as_bool(), graded["status"].as_str()), (200, Some(true), Some("done")), "{graded}");
    let (code, said) = post("/v1/verify", &graded);
    assert_eq!((code, said["valid"].as_bool(), said["answer"]["again"].as_str()), (200, Some(true), Some("reproduced")), "{said}");
    let (code, leaky) = post("/v1/grade", &json!({ "grade": { "task": { "intent": fair, "reference": pair }, "program": pair } }));
    assert_eq!((code, leaky["code"].as_str()), (422, Some("J001")), "{leaky}");
    // A job another key sealed, attested here.
    let other = wai_quantum::ed25519_dalek::SigningKey::from_bytes(&[23; 32]);
    let ran = wai_quantum::engine::job::run(&json!({ "qasm": pair, "shots": 64, "seed": 5 }), 1 << 30, &other, json!({ "unmetered": "a test" })).unwrap();
    let listing = |k: &str| json!({ "format": "wai.quantum.receipt-keys/1", "algorithm": "ed25519", "keys": [{ "key": k }] });
    let (code, attested) = post("/v1/attest", &json!({ "job": ran.to_string(), "intent": { "outcomes": ["00", "11"] }, "trust": listing(ran["result"]["receipt"]["key"].as_str().unwrap()) }));
    assert_eq!((code, attested["result"]["verdict"]["holds"].as_bool(), attested["attested"]["held_to"]["trust"].as_bool()), (200, Some(true), Some(true)), "{attested}");
    let (code, said) = post("/v1/verify", &json!({ "request": attested["request"], "result": attested["result"] }));
    assert_eq!((code, said["valid"].as_bool(), said["answer"]["again"].as_str()), (200, Some(true), Some("reproduced")), "{said}");
    let (code, untrusted) = post("/v1/attest", &json!({ "job": ran.to_string(), "trust": listing(&"00".repeat(32)) }));
    assert!(code == 422 && untrusted["code"] == "J001" && untrusted["error"].as_str().unwrap().contains("key_untrusted"), "{code} {untrusted}");
    // A generator's wide case, compared with a reference past what the service checks.
    let mut generator = String::from("OPENQASM 3.0;\ninput int n;\nqubit[n] q;\nh q;\n");
    for _ in 0..200 {
        generator += "rz(0.3) q;\n";
    }
    let reference = wai_quantum::engine::front::at_inputs(&generator, &wai_quantum::qasm::Inputs::from([("n".to_string(), 22)])).unwrap();
    let task = json!({ "cases": [{ "name": "n=2", "inputs": { "n": 2 }, "intent": fair }, { "name": "n=22", "inputs": { "n": 22 }, "intent": { "equivalent": reference } }] });
    let (code, wide) = post("/v1/grade", &json!({ "grade": { "task": task, "program": generator } }));
    assert!(code == 422 && wide["code"] == "J031" && wide["error"].as_str().unwrap().starts_with("case 1:"), "{code} {wide}");
    let _ = child.kill();
    let _ = child.wait();
    let _ = std::fs::remove_dir_all(&dir);
}