//! Wire and transport conformance (bus-v1 §4 and §11 acceptance test 1): bounded
//! little-endian framing, partial reads/writes, strict JSON (duplicate keys at any depth,
//! invalid UTF-8, unknown fields), malformed ids/u64s, connection negotiation, and behavioral
//! parity between the in-memory and Unix-domain transports.
//!
//! This suite speaks the raw protocol by hand rather than going through the SDK: it sends
//! frames no correct client would ever construct, so it can check what the router does with a
//! hostile or merely buggy peer, not just what a well-behaved one gets back.

mod common;

use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

use common::{Via, code, env, obj, within};
use flybus::Limits;
use flybus::wire::{Envelope, Kind, MAX_ENVELOPE_BYTES, contract_digest, read_frame, write_frame};
use serde_json::{Map, Value, json};
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf};

// -------------------------------------------------------------------------------------------
// Low-level helpers: a hand-rolled connection independent of `common::Raw`, for the framing
// tests that need control over individual bytes and reads that `Raw`'s all-at-once
// `send_bytes` cannot express.

/// Forces every read through this wrapper to surface at most one byte, so a caller reading a
/// frame through it can only succeed by looping the way [`read_frame`] does — never by
/// getting lucky with one big read.
struct Trickle<R> {
    inner: R,
}

impl<R: AsyncRead + Unpin> AsyncRead for Trickle<R> {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        let this = self.get_mut();
        let mut one = [0u8; 1];
        let mut small = ReadBuf::new(&mut one);
        match Pin::new(&mut this.inner).poll_read(cx, &mut small) {
            Poll::Ready(Ok(())) => {
                let n = small.filled().len();
                if n > 0 {
                    buf.put_slice(&one[..n]);
                }
                Poll::Ready(Ok(()))
            }
            other => other,
        }
    }
}

async fn send_envelope(
    wr: &mut (impl AsyncWrite + Unpin),
    id: &str,
    op: &str,
    body: Map<String, Value>,
) {
    let env = Envelope::new(id.to_owned(), Kind::Command, op, body);
    write_frame(wr, &env.encode().unwrap()).await.unwrap();
}

/// Writes the same frame [`send_envelope`] would, one byte at a time, flushing and yielding
/// between every byte.
async fn send_envelope_byte_by_byte(
    wr: &mut (impl AsyncWrite + Unpin),
    id: &str,
    op: &str,
    body: Map<String, Value>,
) {
    let env = Envelope::new(id.to_owned(), Kind::Command, op, body);
    let bytes = env.encode().unwrap();
    let len = (bytes.len() as u32).to_le_bytes();
    for byte in len.iter().chain(bytes.iter()) {
        wr.write_all(std::slice::from_ref(byte)).await.unwrap();
        wr.flush().await.unwrap();
        tokio::task::yield_now().await;
    }
}

async fn recv_envelope(rd: &mut (impl AsyncRead + Unpin)) -> Envelope {
    let bytes = within("frame", read_frame(rd))
        .await
        .unwrap()
        .expect("stream open");
    Envelope::decode(&bytes).unwrap()
}

/// A hand-rolled `bus.hello`, sent as command `msg-1`. Returns the reply's `value` object.
async fn manual_hello(
    rd: &mut (impl AsyncRead + Unpin),
    wr: &mut (impl AsyncWrite + Unpin),
    id: &str,
    slow: bool,
) -> Map<String, Value> {
    let body = obj(
        json!({"clientId": id, "clientIncarnation": format!("inc-{id}"), "supportedMajors": [1]}),
    );
    if slow {
        send_envelope_byte_by_byte(wr, "msg-1", "bus.hello", body).await;
    } else {
        send_envelope(wr, "msg-1", "bus.hello", body).await;
    }
    let reply = recv_envelope(rd).await;
    assert_eq!(reply.kind, Kind::Reply);
    assert_eq!(reply.reply_to.as_deref(), Some("msg-1"));
    assert_eq!(
        reply.body["ok"],
        json!(true),
        "hello failed: {:?}",
        reply.body
    );
    reply.body["value"].as_object().unwrap().clone()
}

// -------------------------------------------------------------------------------------------
// Bounded little-endian framing (bus-v1 §4: "Reject ... zero/oversize frames. Read length
// before allocating.")

async fn zero_length_frame_closes_the_connection(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw_hello("zerolen").await;
    raw.send_prefix(0).await;
    let body = raw
        .closing()
        .await
        .expect("a zero-length frame must close the connection with a notice");
    assert_eq!(body["code"], json!("INVALID_ENVELOPE"));
}

/// The length prefix is checked, and the frame refused, before any body bytes are read. We
/// announce an oversize frame and never send its body: an implementation that read the length
/// after allocating (or tried to read the body anyway) would hang here instead of refusing
/// promptly, and `closing`'s bounded wait turns that into a loud failure rather than a stall.
async fn oversize_length_prefix_is_rejected_before_reading_body(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw_hello("oversize").await;
    raw.send_prefix((MAX_ENVELOPE_BYTES as u32) + 1).await;
    let body = raw
        .closing()
        .await
        .expect("an oversize frame must be refused, not hung waiting for a body that never comes");
    assert_eq!(body["code"], json!("INVALID_ENVELOPE"));
}

/// A frame at exactly the envelope ceiling decodes and dispatches normally; the same shape one
/// byte past it is refused before any JSON parsing.
async fn frame_at_the_size_ceiling_is_accepted_one_byte_over_is_not(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw_hello("ceiling").await;
    // Pad an otherwise-ordinary rpc.call to an exact byte count: measure the unpadded shape
    // once, then fill the gap with a string needing no escaping, so each character costs
    // exactly one byte and the target length is hit without any search.
    let shape = |pad: usize| {
        json!({
            "protocol": "flybus", "major": 1, "minor": 0,
            "id": "msg-9", "replyTo": null, "kind": "command", "op": "rpc.call",
            "body": {
                "callId": "call-1", "target": "no.such.service", "expectedIncarnation": null,
                "method": "M", "payload": {"pad": "a".repeat(pad)},
            },
            "attachments": [],
        })
    };
    let base_len = serde_json::to_vec(&shape(0)).unwrap().len();
    let exact = serde_json::to_vec(&shape(MAX_ENVELOPE_BYTES - base_len)).unwrap();
    assert_eq!(exact.len(), MAX_ENVELOPE_BYTES);
    raw.send_bytes(&exact).await;
    let reply = raw.reply("msg-9").await;
    // Decoded and dispatched fine: an ordinary domain reply (no such service), not a refusal.
    assert_eq!(code(&reply), "NO_SERVICE");

    let mut over_raw = e.raw_hello("overceiling").await;
    let over = serde_json::to_vec(&shape(MAX_ENVELOPE_BYTES - base_len + 1)).unwrap();
    assert_eq!(over.len(), MAX_ENVELOPE_BYTES + 1);
    over_raw.send_bytes(&over).await;
    let body = over_raw
        .closing()
        .await
        .expect("a frame one byte over the ceiling must be refused, not parsed");
    assert_eq!(body["code"], json!("INVALID_ENVELOPE"));
}

/// bus-v1 §4 specifies "u32 little-endian" explicitly. The prefix is built by hand here,
/// independent of the library's own `to_le_bytes` call, for a frame long enough (over 255
/// bytes) that a big-endian misreading would produce a huge, unmistakably wrong length. A
/// byte-order regression shows up as a bounded timeout below, not a silent pass.
async fn length_prefix_is_little_endian(via: Via) {
    let e = env(via).await;
    let (mut rd, mut wr) = tokio::io::split(e.transport().await);
    manual_hello(&mut rd, &mut wr, "byteorder", false).await;

    let body = obj(json!({"name": "a".repeat(190), "retained": "none"}));
    let frame_env = Envelope::new("msg-2".into(), Kind::Command, "topic.declare", body);
    let bytes = frame_env.encode().unwrap();
    assert!(
        bytes.len() > 255,
        "the padding must force a non-trivial high byte in the length"
    );
    let len = bytes.len() as u32;
    let prefix = [
        (len & 0xFF) as u8,
        ((len >> 8) & 0xFF) as u8,
        ((len >> 16) & 0xFF) as u8,
        ((len >> 24) & 0xFF) as u8,
    ];
    assert_eq!(
        prefix,
        len.to_le_bytes(),
        "sanity: the hand-built prefix matches the standard LE encoding"
    );
    let mut frame = prefix.to_vec();
    frame.extend_from_slice(&bytes);
    wr.write_all(&frame).await.unwrap();
    wr.flush().await.unwrap();

    let reply = tokio::time::timeout(Duration::from_secs(5), recv_envelope(&mut rd))
        .await
        .expect("the router must read the length as little-endian, not hang reinterpreting it");
    assert_eq!(reply.body["ok"], json!(true), "{:?}", reply.body);
    assert_eq!(reply.body["value"]["declared"], json!(true));
}

// -------------------------------------------------------------------------------------------
// Partial reads and writes (bus-v1 §4: "Handle partial reads/writes.")

/// A stream that ends mid-frame is a clean disconnect, not a panic and not a phantom command:
/// the router just stops reading and tears the connection down, the same as for a client that
/// vanishes between frames.
async fn truncated_frame_disconnects_cleanly(via: Via) {
    let e = env(via).await;
    {
        let mut raw = e.raw_hello("truncated").await;
        // Announce a 64-byte body, then drop the connection before sending any of it.
        raw.send_prefix(64).await;
    }
    e.settle("the truncated connection is gone", |s| s.connections == 0)
        .await;
}

/// A frame written to the router one byte at a time, with a yield between every byte, decodes
/// exactly as if it had arrived in one write.
async fn a_frame_written_one_byte_at_a_time_still_decodes(via: Via) {
    let e = env(via).await;
    let (mut rd, mut wr) = tokio::io::split(e.transport().await);
    let value = manual_hello(&mut rd, &mut wr, "trickle-writer", true).await;
    assert_eq!(value["selectedMajor"], json!(1));
    assert_eq!(value["contractDigest"], json!(contract_digest()));
}

/// The same guarantee on the reading side: a reply read one byte at a time through
/// [`read_frame`] — the exact function both the router and the client SDK use — reassembles
/// correctly.
async fn a_reply_read_one_byte_at_a_time_still_decodes(via: Via) {
    let e = env(via).await;
    let (rd, mut wr) = tokio::io::split(e.transport().await);
    let mut trickle = Trickle { inner: rd };
    let value = manual_hello(&mut trickle, &mut wr, "trickle-reader", false).await;
    assert_eq!(value["selectedMajor"], json!(1));
}

// -------------------------------------------------------------------------------------------
// Strict JSON: duplicate keys at any depth, invalid UTF-8.

/// bus-v1 §4: "Reject duplicate JSON keys" — the same rule at the top level, nested inside
/// `body`, and nested inside an array element within it.
async fn duplicate_json_keys_are_rejected_at_every_depth(via: Via) {
    let e = env(via).await;
    let cases: [&[u8]; 4] = [
        br#"{"protocol":"flybus","protocol":"flybus","major":1,"minor":0,"id":"msg-1","replyTo":null,"kind":"command","op":"bus.hello","body":{"clientId":"x","clientIncarnation":"y","supportedMajors":[1]},"attachments":[]}"#,
        br#"{"protocol":"flybus","major":1,"minor":0,"id":"msg-1","replyTo":null,"kind":"command","op":"bus.hello","body":{"clientId":"x","clientId":"z","clientIncarnation":"y","supportedMajors":[1]},"attachments":[]}"#,
        br#"{"protocol":"flybus","major":1,"minor":0,"id":"msg-1","replyTo":null,"kind":"command","op":"rpc.call","body":{"callId":"call-1","target":"a.b","expectedIncarnation":null,"method":"M","payload":{"n":{"d":1,"d":2}}},"attachments":[]}"#,
        br#"{"protocol":"flybus","major":1,"minor":0,"id":"msg-1","replyTo":null,"kind":"command","op":"rpc.call","body":{"callId":"call-1","target":"a.b","expectedIncarnation":null,"method":"M","payload":{"items":[{"x":1,"x":2}]}},"attachments":[]}"#,
    ];
    for (i, case) in cases.iter().enumerate() {
        let mut raw = e.raw().await;
        raw.send_bytes(case).await;
        let body = raw.closing().await;
        assert!(
            body.is_some(),
            "case {i}: duplicate keys must close the connection"
        );
        assert_eq!(body.unwrap()["code"], json!("INVALID_ENVELOPE"), "case {i}");
    }
}

/// Invalid UTF-8 anywhere in the frame is refused before JSON parsing starts, whether it falls
/// inside a string value, trails a complete object, or leads the buffer.
async fn invalid_utf8_is_rejected(via: Via) {
    let e = env(via).await;
    let cases: [&[u8]; 3] = [
        b"{\"protocol\":\"flybus\",\"major\":1,\"minor\":0,\"id\":\"msg-1\",\"replyTo\":null,\"kind\":\"command\",\"op\":\"bus.hello\",\"body\":{\"clientId\":\"\xff\",\"clientIncarnation\":\"y\",\"supportedMajors\":[1]},\"attachments\":[]}",
        b"{\"protocol\":\"flybus\",\"major\":1,\"minor\":0,\"id\":\"msg-1\",\"replyTo\":null,\"kind\":\"command\",\"op\":\"bus.hello\",\"body\":{\"clientId\":\"x\",\"clientIncarnation\":\"y\",\"supportedMajors\":[1]},\"attachments\":[]}\xff",
        b"\xc0\x80{\"protocol\":\"flybus\"}",
    ];
    for (i, case) in cases.iter().enumerate() {
        let mut raw = e.raw().await;
        raw.send_bytes(case).await;
        let body = raw.closing().await;
        assert!(
            body.is_some(),
            "case {i}: invalid UTF-8 must close the connection"
        );
        assert_eq!(body.unwrap()["code"], json!("INVALID_ENVELOPE"), "case {i}");
    }
}

// -------------------------------------------------------------------------------------------
// Unknown envelope fields: a transport-level refusal at the envelope's own level, an ordinary
// domain error inside an operation's body.

async fn unknown_top_level_field_closes_the_connection(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw().await;
    let v = json!({
        "protocol": "flybus", "major": 1, "minor": 0,
        "id": "msg-1", "replyTo": null, "kind": "command", "op": "bus.hello",
        "body": {"clientId": "x", "clientIncarnation": "y", "supportedMajors": [1]},
        "attachments": [],
        "extra": true,
    });
    raw.send_bytes(&serde_json::to_vec(&v).unwrap()).await;
    let body = raw
        .closing()
        .await
        .expect("an unknown envelope field must close the connection");
    assert_eq!(body["code"], json!("INVALID_ENVELOPE"));
}

/// A field the *operation* does not recognize is a normal `ok:false` reply, not a transport
/// violation: only the envelope's own shape is the wire's concern, so the connection stays
/// open and keeps working afterward.
async fn unknown_body_field_is_a_domain_error_not_a_disconnect(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw_hello("unknownfield").await;
    let bogus = json!({
        "callId": "call-1", "target": "no.such.service", "expectedIncarnation": null,
        "method": "M", "payload": {}, "bogus": true,
    });
    let reply = raw.call("rpc.call", bogus).await;
    assert_eq!(code(&reply), "INVALID_ENVELOPE");
    let ok = raw
        .call(
            "topic.declare",
            json!({"name": "still.alive", "retained": "none"}),
        )
        .await;
    assert_eq!(code(&ok), "OK");
}

// -------------------------------------------------------------------------------------------
// Malformed ids and u64s.

/// Every envelope-level scalar the wire validates before dispatch: a malformed `id`, an
/// unparseable `kind`, and an `op` outside its `[a-z.]` alphabet. Each is refused at decode,
/// before hello state or operation semantics are consulted at all.
async fn malformed_envelope_scalars_are_rejected(via: Via) {
    let e = env(via).await;
    let hello_body = json!({"clientId": "x", "clientIncarnation": "y", "supportedMajors": [1]});
    let base = |id: &str, kind: &str, op: &str| {
        json!({
            "protocol": "flybus", "major": 1, "minor": 0,
            "id": id, "replyTo": null, "kind": kind, "op": op,
            "body": hello_body, "attachments": [],
        })
    };
    let cases = [
        (base("", "command", "bus.hello"), "empty id"),
        (base("MSG-1", "command", "bus.hello"), "uppercase id"),
        (
            base("-x", "command", "bus.hello"),
            "id starting with a separator",
        ),
        (
            base(&"a".repeat(65), "command", "bus.hello"),
            "id over 64 characters",
        ),
        (base("msg-1", "bogus", "bus.hello"), "unknown kind"),
        (base("msg-1", "command", "Bus.Hello"), "uppercase op"),
        (base("msg-1", "command", "bus.hello1"), "digit in op"),
    ];
    for (v, what) in cases {
        let mut raw = e.raw().await;
        raw.send_bytes(&serde_json::to_vec(&v).unwrap()).await;
        let body = raw.closing().await;
        assert!(body.is_some(), "{what}: must close the connection");
        assert_eq!(body.unwrap()["code"], json!("INVALID_ENVELOPE"), "{what}");
    }
}

async fn non_null_reply_to_on_a_command_is_rejected(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw_hello("replyto").await;
    let v = json!({
        "protocol": "flybus", "major": 1, "minor": 0,
        "id": "msg-2", "replyTo": "msg-1", "kind": "command", "op": "topic.declare",
        "body": {"name": "a.b", "retained": "none"}, "attachments": [],
    });
    raw.send_bytes(&serde_json::to_vec(&v).unwrap()).await;
    let body = raw
        .closing()
        .await
        .expect("a non-null replyTo on a command must close the connection");
    assert_eq!(body["code"], json!("INVALID_ENVELOPE"));
}

/// `msg-<U64>` is canonical, not merely `Id`-shaped: no leading zero, no missing digits, no
/// other prefix.
async fn non_canonical_command_ids_are_rejected(via: Via) {
    let e = env(via).await;
    for bad in ["msg-01", "msg-abc", "notmsg-1", "msg--1", "msg-"] {
        let mut raw = e.raw().await;
        let v = json!({
            "protocol": "flybus", "major": 1, "minor": 0,
            "id": bad, "replyTo": null, "kind": "command", "op": "bus.hello",
            "body": {"clientId": "x", "clientIncarnation": "y", "supportedMajors": [1]},
            "attachments": [],
        });
        raw.send_bytes(&serde_json::to_vec(&v).unwrap()).await;
        let body = raw.closing().await;
        assert!(
            body.is_some(),
            "{bad}: a non-canonical command id must close the connection"
        );
        assert_eq!(body.unwrap()["code"], json!("INVALID_ENVELOPE"), "{bad}");
    }
}

async fn non_increasing_command_ids_are_rejected(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw_hello("nonincreasing").await; // hello already spent msg-1
    let v = json!({
        "protocol": "flybus", "major": 1, "minor": 0,
        "id": "msg-1", "replyTo": null, "kind": "command", "op": "topic.declare",
        "body": {"name": "a.b", "retained": "none"}, "attachments": [],
    });
    raw.send_bytes(&serde_json::to_vec(&v).unwrap()).await;
    let body = raw
        .closing()
        .await
        .expect("a repeated command id must close the connection");
    assert_eq!(body["code"], json!("INVALID_ENVELOPE"));
}

/// Below the envelope's own `id`, an operation's own ids (`callId`) get the same canonical-U64
/// treatment, but as an ordinary domain reply: the connection is unharmed by a bad one.
async fn malformed_call_id_is_a_domain_error(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw_hello("badcallid").await;
    for bad in ["call-01", "call-abc", "callx-1", ""] {
        let reply =
            raw.call("rpc.call", json!({"callId": bad, "target": "a.b", "expectedIncarnation": null, "method": "M", "payload": {}})).await;
        assert_eq!(code(&reply), "INVALID_ENVELOPE", "{bad:?}");
    }
}

/// U64-string fields (`ipc-v1.md` §2: `"0"` or `[1-9][0-9]*`, at most `u64::MAX`) reject a
/// leading zero, a non-digit, an empty string, whitespace, a decimal point and an overflow —
/// each as a domain error the connection survives.
async fn malformed_u64_fields_are_a_domain_error(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw_hello("badu64").await;
    for bad in ["01", "-1", "abc", "", "18446744073709551616", " 1", "1.0"] {
        let reply = raw
            .call(
                "artifact.allocate",
                json!({"byteLength": bad, "contentType": "application/octet-stream"}),
            )
            .await;
        assert_eq!(code(&reply), "INVALID_ENVELOPE", "{bad:?}");
    }
    let ok = raw
        .call(
            "artifact.allocate",
            json!({"byteLength": "1", "contentType": "application/octet-stream"}),
        )
        .await;
    assert_eq!(code(&ok), "OK");
}

// -------------------------------------------------------------------------------------------
// Negotiation (bus-v1 §4, "Connection negotiation").

async fn first_command_must_be_hello(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw().await;
    raw.command(
        "service.register",
        json!({"name": "x.y", "maxQueued": 1, "maxInFlight": 1}),
        json!([]),
    )
    .await;
    let body = raw
        .closing()
        .await
        .expect("a non-hello first command must close the connection");
    assert_eq!(body["code"], json!("INVALID_ENVELOPE"));
}

async fn hello_refuses_an_unsupported_major(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw().await;
    let reply = raw.call("bus.hello", json!({"clientId": "futuristic", "clientIncarnation": "inc-1", "supportedMajors": [2, 3]})).await;
    assert_eq!(code(&reply), "VERSION_MISMATCH");
    assert!(
        raw.recv().await.is_none(),
        "the connection must close right after refusing the major"
    );
}

async fn hello_refuses_attachments(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw().await;
    let attachments = json!([{
        "name": "a",
        "ref": {
            "storeId": "s", "artifactId": "a-1", "generation": "1",
            "byteLength": "1", "contentType": "x", "digest": null,
        },
        "ownerId": "o",
    }]);
    let reply = raw
        .call_with(
            "bus.hello",
            json!({"clientId": "attacher", "clientIncarnation": "inc-1", "supportedMajors": [1]}),
            attachments,
        )
        .await;
    assert_eq!(code(&reply), "INVALID_ENVELOPE");
}

async fn a_second_hello_on_the_same_connection_is_rejected(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw_hello("rehello").await;
    let v = json!({
        "protocol": "flybus", "major": 1, "minor": 0,
        "id": "msg-2", "replyTo": null, "kind": "command", "op": "bus.hello",
        "body": {"clientId": "rehello", "clientIncarnation": "inc-2", "supportedMajors": [1]},
        "attachments": [],
    });
    raw.send_bytes(&serde_json::to_vec(&v).unwrap()).await;
    let body = raw
        .closing()
        .await
        .expect("a second bus.hello must close the connection");
    assert_eq!(body["code"], json!("INVALID_ENVELOPE"));
}

async fn hello_reports_the_contract_digest_and_valid_limits(via: Via) {
    let e = env(via).await;
    let mut raw = e.raw().await;
    let reply = raw.hello("reporter").await.unwrap();
    assert_eq!(reply["selectedMajor"], json!(1));
    assert_eq!(reply["selectedMinor"], json!(0));
    assert_eq!(reply["contractDigest"], json!(contract_digest()));
    assert!(reply["routerId"].as_str().unwrap().starts_with("router-"));
    assert!(reply["connectionId"].as_str().unwrap().starts_with("conn-"));
    let limits = Limits::from_json(&reply["limits"])
        .expect("the router's own limits object must round-trip");
    assert_eq!(limits, Limits::default());
}

both_transports!(
    zero_length_frame_closes_the_connection,
    oversize_length_prefix_is_rejected_before_reading_body,
    frame_at_the_size_ceiling_is_accepted_one_byte_over_is_not,
    length_prefix_is_little_endian,
    truncated_frame_disconnects_cleanly,
    a_frame_written_one_byte_at_a_time_still_decodes,
    a_reply_read_one_byte_at_a_time_still_decodes,
    duplicate_json_keys_are_rejected_at_every_depth,
    invalid_utf8_is_rejected,
    unknown_top_level_field_closes_the_connection,
    unknown_body_field_is_a_domain_error_not_a_disconnect,
    malformed_envelope_scalars_are_rejected,
    non_null_reply_to_on_a_command_is_rejected,
    non_canonical_command_ids_are_rejected,
    non_increasing_command_ids_are_rejected,
    malformed_call_id_is_a_domain_error,
    malformed_u64_fields_are_a_domain_error,
    first_command_must_be_hello,
    hello_refuses_an_unsupported_major,
    hello_refuses_attachments,
    a_second_hello_on_the_same_connection_is_rejected,
    hello_reports_the_contract_digest_and_valid_limits,
);

// -------------------------------------------------------------------------------------------
// bus-v1 §11 acceptance test 1: "In-memory transport must pass the same tests as Unix
// sockets." Every test above already runs on both (that is what `both_transports!` is for);
// this one drives the identical raw script over both side by side in a single test, so a
// transport-specific quirk in one implementation cannot hide behind "it still passes its own
// copy of the suite."

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn memory_and_unix_negotiate_the_identical_contract() {
    let mem = env(Via::Memory).await;
    let unix = env(Via::Unix).await;
    let mut raw_mem = mem.raw().await;
    let mut raw_unix = unix.raw().await;

    let hm = raw_mem.hello("parity").await.unwrap();
    let hu = raw_unix.hello("parity").await.unwrap();
    assert_eq!(hm["selectedMajor"], hu["selectedMajor"]);
    assert_eq!(hm["selectedMinor"], hu["selectedMinor"]);
    assert_eq!(hm["contractDigest"], hu["contractDigest"]);
    assert_eq!(hm["limits"], hu["limits"]);

    let cm = raw_mem
        .call(
            "topic.declare",
            json!({"name": "parity.topic", "retained": "latest"}),
        )
        .await
        .unwrap();
    let cu = raw_unix
        .call(
            "topic.declare",
            json!({"name": "parity.topic", "retained": "latest"}),
        )
        .await
        .unwrap();
    assert_eq!(cm["declared"], cu["declared"]);
    assert!(cm["topicIncarnation"].as_str().unwrap().starts_with("top-"));
    assert!(cu["topicIncarnation"].as_str().unwrap().starts_with("top-"));

    // The identical malformed frame gets the identical transport-level refusal on both.
    let mut bad_mem = mem.raw_hello("parity-bad").await;
    let mut bad_unix = unix.raw_hello("parity-bad").await;
    let bad_frame = br#"{"a":1,"a":2}"#;
    bad_mem.send_bytes(bad_frame).await;
    bad_unix.send_bytes(bad_frame).await;
    let bm = bad_mem.closing().await.unwrap();
    let bu = bad_unix.closing().await.unwrap();
    assert_eq!(bm["code"], bu["code"]);
}