open-standards/joule-code/joule-code-rs/crates/joule-moq/tests/draft23_vectors.rs

//! The `draft-ietf-moq-transport-23` vector corpus (`tests/vectors/draft-23/`).
//!
//! Built here from typed values, deterministically, as `draft22_vectors.rs` builds -22's:
//! the bytes are what this crate's -23 encoder makes of the value (or, for a negative
//! vector and the few forms the encoder never makes, bytes written out by hand), and the
//! expected reading is the value itself in `joule_moq::draft23::json`'s view, or, for a
//! negative vector, the refusal it must meet with -23's consequence: since -23 (#1773) most
//! faults close the session with a code of their own, chosen by the fault and the plane it
//! arose on (`joule_moq::draft23::error`). Expectations are stated here, never read back
//! from the decoder.
//!
//! The corpus has every control message of -23 and every parameter each may carry, the
//! forms -23 changed (a varint Length, MAX_CONTROL_MESSAGE_SIZE, REQUEST_OK answering FETCH,
//! Error Details, NAMESPACE's parameters, PAUSED, SUBSCRIBE_TRACKS_ID, counted
//! FILL_PARAMETERS, fetch streams with Object ID Delta + 2, delta-coded End of Range, End of
//! Gone Range, End of Group and End of Track), and negative vectors: the planted faults, among
//! them each -22 form a -23 endpoint refuses.
//!
//! This test holds the crate to the corpus: the committed `vectors.json` is what this file
//! generates (`JOULE_MOQ_WRITE_VECTORS=1` rewrites it); each good vector decodes to its
//! reading and, where canonical, re-encodes to its bytes; each negative vector is refused
//! with its kind and consequence, by the whole-stream decoder and, for streams, by the
//! incremental one in pieces of every size tried. `vectors/draft-23/moqt23_check.py`
//! decodes the same corpus independently, from the draft's text.

use joule_moq::carriage::WaiReceiptProperty;
use joule_moq::draft22::kvp::{opt, prop, KeyValuePair, Properties, SetupOptions};
use joule_moq::draft22::params::{Range, RangeFilter};
use joule_moq::draft22::{varint, GroupOrder, Location, LocationFilter, Token, WireError};
use joule_moq::draft23::codes::{PublishDoneCode, RequestErrorCode};
use joule_moq::draft23::control::*;
use joule_moq::draft23::data::{
    encode_control_stream, encode_fetch_stream, encode_padding_stream, encode_subgroup_stream, stream_plane, stream_type, DatagramContent, FetchCodec, FetchHeader, FetchObject,
    Limits, ObjectDatagram, ObjectStatus, SubgroupCodec, SubgroupHeader, SubgroupId, SubgroupObject, END_OF_GROUP,
};
use joule_moq::draft23::params::param;
use joule_moq::draft23::{decode_uni_stream, json, Datagram, FetchEntry, Parameter, Parameters, Plane, RangeEnd, StreamHeader, UniEvent, UniStreamDecoder};
use joule_moq::track::{TrackName, TrackNamespace};
use serde_json::{json, Value};

const CORPUS: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/vectors/draft-23/vectors.json");

fn hx(b: &[u8]) -> String {
    joule_moq::draft22::json::hex(b)
}

fn ns(fields: &[&str]) -> TrackNamespace {
    TrackNamespace::from_strs(fields)
}

fn name(s: &str) -> TrackName {
    TrackName::from_str(s)
}

fn kv(ty: u64, v: u64) -> KeyValuePair {
    KeyValuePair::varint(ty, v)
}

fn kb(ty: u64, b: &[u8]) -> KeyValuePair {
    KeyValuePair::bytes(ty, b.to_vec())
}

fn props(mut ps: Vec<KeyValuePair>) -> Properties {
    ps.sort_by_key(|k| k.ty);
    Properties(ps)
}

fn with_immutable(mut outer: Vec<KeyValuePair>, inner: Vec<KeyValuePair>) -> Properties {
    let mut b = Vec::new();
    joule_moq::draft22::kvp::encode_pairs(&inner, &mut b).unwrap();
    outer.push(KeyValuePair::bytes(prop::IMMUTABLE_PROPERTIES, b));
    let mut p = Properties(outer);
    p.0.sort_by_key(|k| k.ty);
    p
}

fn rf(ty: u64, set_id: u8, property_type: Option<u64>, ranges: &[(u64, Option<u64>)]) -> Parameter {
    Parameter::RangeFilter { ty, filter: Some(RangeFilter { set_id, property_type, ranges: ranges.iter().map(|&(start, end)| Range { start, end }).collect() }) }
}

fn sorted(ps: Vec<Parameter>) -> Parameters {
    let mut v = ps;
    v.sort_by_key(|p| p.type_code());
    Parameters(v)
}

struct Corpus {
    vectors: Vec<Value>,
}

impl Corpus {
    fn push(&mut self, name: &str, kind: &str, bytes: &[u8], canonical: bool, context: Option<Value>, expect: Value) {
        let mut v = json!({"name": name, "kind": kind, "hex": hx(bytes), "canonical": canonical, "expect": expect});
        if let Some(c) = context {
            v["context"] = c;
        }
        self.vectors.push(v);
    }

    fn control(&mut self, name: &str, m: ControlMessage) {
        let b = m.encode().expect(name);
        self.push(name, "control", &b, true, None, json::control(&m));
    }

    fn stream(&mut self, name: &str, bytes: Vec<u8>, order: GroupOrder, events: Vec<UniEvent>) {
        let expect = Value::Array(events.iter().map(json::uni_event).collect());
        self.push(name, "uni-stream", &bytes, true, Some(json!({"group_order": order as u8})), expect);
    }

    fn datagram(&mut self, name: &str, d: Datagram, bytes: Vec<u8>, canonical: bool) {
        self.push(name, "datagram", &bytes, canonical, None, json::datagram(&d));
    }

    fn bad(&mut self, name: &str, kind: &str, bytes: &[u8], context: Option<Value>, error: &str, consequence: &str) {
        self.push(name, kind, bytes, false, context, json!({"error": error, "consequence": consequence}));
    }
}

// -23's codes (§12.2), as the corpus writes a consequence.
const PV: &str = "session:PROTOCOL_VIOLATION";
const MKV: &str = "session:MALFORMED_KEY_VALUE";
const ISH: &str = "session:INVALID_STREAM_HEADER";
const MO: &str = "session:MALFORMED_OBJECT";
const IOS: &str = "session:INVALID_OBJECT_STATUS";
const TO: &str = "session:TRUNCATED_OBJECT";
const MM: &str = "session:MALFORMED_MESSAGE";
const IF: &str = "session:INVALID_FIELD";
const ITN: &str = "session:INVALID_TRACK_NAMESPACE";
const UM: &str = "session:UNEXPECTED_MESSAGE";
const IP: &str = "session:INVALID_PARAMETER";
const DEO: &str = "session:DELTA_ENCODING_OVERFLOW";
const FLE: &str = "session:FIELD_LENGTH_EXCEEDED";
const MAT: &str = "session:MALFORMED_AUTH_TOKEN";
const MT: &str = "malformed-track";
const FILTER: &str = "request:INVALID_FILTER";

/// Frame a control message body by hand: type, a varint Length, the body.
fn frame(ty: u64, body: &[u8]) -> Vec<u8> {
    let mut out = varint::to_vec(ty);
    out.extend(varint::to_vec(body.len() as u64));
    out.extend_from_slice(body);
    out
}

fn v(x: u64) -> Vec<u8> {
    varint::to_vec(x)
}

fn cat(parts: &[&[u8]]) -> Vec<u8> {
    parts.concat()
}

fn wai_receipt(content: u8, joules: u64, class: &str, evidence: Vec<u8>, group: u8) -> WaiReceiptProperty {
    WaiReceiptProperty { content_hash: [content; 32], joules_micro: joules, acquisition: class.into(), acquisition_evidence: evidence, group_receipt: [group; 32] }
}

fn build() -> Vec<Value> {
    let mut c = Corpus { vectors: Vec::new() };

    // ── §8.1 the variable-length integer: -23's Table 4 is -22's ──
    for (h, val, shortest) in [("25", 37u64, true), ("8025", 37, false), ("bbbd", 15_293, true), ("ed7f3e7d", 226_442_877, true), ("ffffffffffffffffff", u64::MAX, true)] {
        let b: Vec<u8> = (0..h.len()).step_by(2).map(|i| u8::from_str_radix(&h[i..i + 2], 16).unwrap()).collect();
        c.push(&format!("varint {h}"), "varint", &b, shortest, None, json!({"value": val, "length": b.len()}));
    }

    // ── §9.1 SETUP, with MAX_CONTROL_MESSAGE_SIZE (§9.1.8) ──
    let mut client = SetupOptions::new();
    client
        .push(kb(opt::PATH, b"/live?room=1"))
        .push(kb(opt::AUTHORIZATION_TOKEN, &Token::Register { alias: 1, token_type: 0, value: b"tok".to_vec() }.to_bytes()))
        .push(kv(opt::MAX_AUTH_TOKEN_CACHE_SIZE, 1024))
        .push(kb(opt::AUTHORITY, b"relay.example:443"))
        .push(kv(opt::MAX_FILTER_RANGES, 4))
        .push(kb(opt::MOQT_IMPLEMENTATION, b"example-impl/1.0"))
        .push(kv(opt::MAX_REQUEST_UPDATES, 2))
        .push(max_control_message_size_option(1 << 20))
        .push(kb(0x9d, b"grease"))
        .push(kv(0x3a46, 2));
    let mut client_sorted = client.clone();
    client_sorted.0.sort_by_key(|p| p.ty);
    c.push("SETUP from a client", "control", &ControlMessage::Setup(Setup { options: client }).encode().unwrap(), true, None, json::control(&ControlMessage::Setup(Setup { options: client_sorted })));
    let mut server = SetupOptions::new();
    server.push(kb(opt::MOQT_IMPLEMENTATION, b"example-impl/1.0")).push(max_control_message_size_option(65535));
    let mut server_sorted = server.clone();
    server_sorted.0.sort_by_key(|p| p.ty);
    c.push("SETUP from a server, MAX_CONTROL_MESSAGE_SIZE at its least", "control", &ControlMessage::Setup(Setup { options: server }).encode().unwrap(), true, None, json::control(&ControlMessage::Setup(Setup { options: server_sorted })));
    c.control("SETUP with no options", ControlMessage::Setup(Setup::default()));

    // ── §9.2 GOAWAY ──
    c.control("GOAWAY with a new session URI", ControlMessage::Goaway(Goaway { new_session_uri: b"moqt://relay.example/next".to_vec(), timeout_ms: 5000 }));
    c.control("GOAWAY with no URI and no timeout", ControlMessage::Goaway(Goaway::default()));

    // ── §9.6 SUBSCRIBE, with every parameter it may carry ──
    let every = sorted(vec![
        Parameter::ObjectDeliveryTimeout(250),
        Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: b"jwt".to_vec() }),
        Parameter::RendezvousTimeout(1000),
        Parameter::SubgroupDeliveryTimeout(500),
        Parameter::Paused(false),
        Parameter::SubscriberPriority(10),
        Parameter::LocationFilter(LocationFilter::NextObject),
        Parameter::GroupOrder(GroupOrder::Descending),
        Parameter::FillParameters(sorted(vec![Parameter::LocationFilter(LocationFilter::RelativeStart { start_group: 1 }), Parameter::FillTimeout(100), rf(param::OBJECTID_FILTER, 0, None, &[(0, Some(4))])])),
        rf(param::SUBGROUP_FILTER, 0, None, &[(0, Some(2))]),
        rf(param::OBJECTID_FILTER, 0, None, &[(3, Some(5)), (10, None)]),
        rf(param::PRIORITY_FILTER, 1, None, &[(0, Some(127))]),
        rf(param::OBJECT_PROPERTY_FILTER, 0, Some(0x3a40), &[(0, Some(2))]),
        Parameter::NewGroupRequest(0),
        Parameter::IncludeProperties(true),
    ]);
    c.control("SUBSCRIBE with every parameter", ControlMessage::Subscribe(TrackRequest { request_id: 0, namespace: ns(&["example", "cam1"]), name: name("video"), params: every }));
    // The extension parameter relays negotiate with the Setup Option RELAY_ID (README, *An
    // extension parameter: RELAY_PATH*): the relays a request passed through, the sender last.
    c.control(
        "SUBSCRIBE with RELAY_PATH, the negotiated extension parameter",
        ControlMessage::Subscribe(TrackRequest { request_id: 4, namespace: ns(&["example", "cam1"]), name: name("video"), params: sorted(vec![Parameter::Paused(false), Parameter::RelayPath(vec![[0x11; 32], [0x22; 32]])]) }),
    );
    c.control("SUBSCRIBE_NAMESPACE with RELAY_PATH", ControlMessage::SubscribeNamespace(NamespaceRequest { request_id: 6, prefix: ns(&["example"]), params: sorted(vec![Parameter::RelayPath(vec![[0x33; 32]])]) }));
    c.control("SUBSCRIBE paused, an empty namespace", ControlMessage::Subscribe(TrackRequest { request_id: 2, namespace: TrackNamespace(vec![]), name: name("t"), params: sorted(vec![Parameter::Paused(true)]) }));
    // A Length and a Request ID each in a longer form than they need: the reading is the same.
    let sub = ControlMessage::Subscribe(TrackRequest { request_id: 5, namespace: ns(&["a"]), name: name("b"), params: Parameters::new() });
    let mut body = Vec::new();
    varint::put_with_len(&mut body, 5, 3).unwrap();
    body.extend_from_slice(&cat(&[&v(1), &v(1), b"a", &v(1), b"b", &v(0)]));
    let mut long = v(msg::SUBSCRIBE);
    varint::put_with_len(&mut long, body.len() as u64, 2).unwrap();
    long.extend_from_slice(&body);
    c.push("SUBSCRIBE whose Length and Request ID take more bytes than they need", "control", &long, false, None, json::control(&sub));

    // ── §9.7 SUBSCRIBE_OK ──
    c.control(
        "SUBSCRIBE_OK with track properties",
        ControlMessage::SubscribeOk(SubscribeOk {
            track_alias: 7,
            params: sorted(vec![Parameter::Expires(30_000), Parameter::LargestObject(Location::new(5, 2))]),
            track_properties: with_immutable(
                vec![kv(prop::OBJECT_DELIVERY_TIMEOUT, 200), kv(prop::MAX_CACHE_DURATION, 10_000), kv(prop::DEFAULT_PUBLISHER_PRIORITY, 64), kv(prop::DEFAULT_PUBLISHER_GROUP_ORDER, 1), kv(prop::DYNAMIC_GROUPS, 1), kb(0x3a45, b"x")],
                vec![kv(0x78, 5)],
            ),
        }),
    );
    // ── §9.10 PUBLISH_STATE_NOTIFY ──
    c.control(
        "PUBLISH_STATE_NOTIFY",
        ControlMessage::PublishStateNotify(sorted(vec![
            Parameter::LargestObject(Location::new(9, 0)),
            Parameter::Paused(true),
            Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(9, 0) }),
        ])),
    );
    // ── §9.8 PUBLISH, with FILL_PARAMETERS and SUBSCRIBE_TRACKS_ID ──
    c.control(
        "PUBLISH answering SUBSCRIBE_TRACKS",
        ControlMessage::Publish(Publish {
            request_id: 1,
            namespace: ns(&["example", "cam1"]),
            name: name("audio"),
            track_alias: 3,
            params: sorted(vec![
                Parameter::Expires(60_000),
                Parameter::LargestObject(Location::new(4, 9)),
                Parameter::Paused(false),
                Parameter::GroupOrder(GroupOrder::Ascending),
                Parameter::LocationFilter(LocationFilter::AbsoluteStartGroupEnd { start: Location::new(4, 0), end_group_delta: 2 }),
                Parameter::FillParameters(sorted(vec![Parameter::LocationFilter(LocationFilter::None)])),
                Parameter::SubscribeTracksId(10),
            ]),
            track_properties: props(vec![kv(prop::DYNAMIC_GROUPS, 0)]),
        }),
    );
    // ── §9.9 PUBLISH_DONE ──
    c.control("PUBLISH_DONE", ControlMessage::PublishDone(PublishDone { status: PublishDoneCode::TRACK_ENDED, stream_count: 3, reason: b"done".to_vec() }));
    c.control("PUBLISH_DONE: EXPIRED_AUTH_TOKEN", ControlMessage::PublishDone(PublishDone { status: PublishDoneCode::EXPIRED_AUTH_TOKEN, stream_count: u64::MAX, reason: vec![] }));
    // ── §9.11 FETCH, and its REQUEST_OK (§9.3) ──
    c.control(
        "FETCH",
        ControlMessage::Fetch(TrackRequest {
            request_id: 4,
            namespace: ns(&["example", "cam1"]),
            name: name("video"),
            params: sorted(vec![
                Parameter::AuthorizationToken(Token::UseAlias { alias: 1 }),
                Parameter::FillTimeout(0),
                Parameter::SubscriberPriority(200),
                Parameter::LocationFilter(LocationFilter::AbsoluteRange { start: Location::new(1, 0), end_group_delta: 3, end_object: 7 }),
                Parameter::GroupOrder(GroupOrder::Descending),
                Parameter::IncludeProperties(false),
            ]),
        }),
    );
    c.control(
        "REQUEST_OK answering FETCH, the range reaching the Largest Object",
        ControlMessage::RequestOk(RequestOk {
            params: sorted(vec![Parameter::Expires(5000), Parameter::LargestObject(Location::new(4, 7))]),
            track_properties: props(vec![kv(prop::DEFAULT_PUBLISHER_PRIORITY, 128)]),
        }),
    );
    // ── §9.12 TRACK_STATUS ──
    c.control(
        "TRACK_STATUS",
        ControlMessage::TrackStatus(TrackRequest {
            request_id: 6,
            namespace: ns(&["example"]),
            name: name("cam1"),
            params: sorted(vec![Parameter::AuthorizationToken(Token::Register { alias: 2, token_type: 0x9d, value: b"abc".to_vec() }), Parameter::IncludeProperties(true)]),
        }),
    );
    // ── §9.13, §9.14, §9.17 namespace requests ──
    c.control(
        "PUBLISH_NAMESPACE",
        ControlMessage::PublishNamespace(NamespaceRequest { request_id: 3, prefix: ns(&["example", "cam1"]), params: sorted(vec![Parameter::AuthorizationToken(Token::UseValue { token_type: 1, value: vec![0xde, 0xad] })]) }),
    );
    c.control("SUBSCRIBE_NAMESPACE", ControlMessage::SubscribeNamespace(NamespaceRequest { request_id: 8, prefix: ns(&["example"]), params: Parameters::new() }));
    c.control(
        "SUBSCRIBE_TRACKS with every parameter it may carry",
        ControlMessage::SubscribeTracks(NamespaceRequest {
            request_id: 10,
            prefix: TrackNamespace(vec![]),
            params: sorted(vec![
                Parameter::ObjectDeliveryTimeout(100),
                Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: b"t".to_vec() }),
                Parameter::SubgroupDeliveryTimeout(300),
                Parameter::Paused(true),
                Parameter::SubscriberPriority(7),
                Parameter::LocationFilter(LocationFilter::RelativeStart { start_group: 0 }),
                Parameter::GroupOrder(GroupOrder::Ascending),
                Parameter::FillParameters(sorted(vec![Parameter::SubscriberPriority(9)])),
                rf(param::SUBGROUP_FILTER, 0, None, &[(0, Some(0))]),
                rf(param::OBJECTID_FILTER, 0, None, &[(0, None)]),
                rf(param::PRIORITY_FILTER, 0, None, &[(0, Some(255))]),
                rf(param::OBJECT_PROPERTY_FILTER, 0, Some(0x3a40), &[(0, Some(0))]),
                rf(param::TRACK_PROPERTY_FILTER, 0, Some(prop::DYNAMIC_GROUPS), &[(1, Some(1))]),
                Parameter::IncludeProperties(false),
            ]),
        }),
    );
    // ── §9.15, §9.16, §9.18 ──
    c.control("NAMESPACE, its parameters (none defined)", ControlMessage::Namespace(Namespace { suffix: ns(&["42"]), params: Parameters::new() }));
    c.control("NAMESPACE_DONE", ControlMessage::NamespaceDone(ns(&["123"])));
    c.control("PUBLISH_SKIPPED", ControlMessage::PublishSkipped(PublishSkipped { suffix: ns(&["cam2"]), name: name("video") }));
    // ── §9.5 REQUEST_UPDATE ──
    c.control(
        "REQUEST_UPDATE for a subscription",
        ControlMessage::RequestUpdate(RequestUpdate {
            request_id: 12,
            params: sorted(vec![
                Parameter::Paused(true),
                Parameter::SubscriberPriority(5),
                Parameter::LocationFilter(LocationFilter::None),
                Parameter::FillParameters(sorted(vec![Parameter::GroupOrder(GroupOrder::Descending)])),
                Parameter::RangeFilter { ty: param::OBJECTID_FILTER, filter: None },
                Parameter::NewGroupRequest(12),
            ]),
        }),
    );
    c.control("REQUEST_UPDATE for a namespace subscription", ControlMessage::RequestUpdate(RequestUpdate { request_id: 14, params: sorted(vec![Parameter::TrackNamespacePrefix(ns(&["example", "b"]))]) }));
    // ── §9.3 REQUEST_OK, §9.4 REQUEST_ERROR with Error Details ──
    c.control(
        "REQUEST_OK answering TRACK_STATUS, with EXPIRES",
        ControlMessage::RequestOk(RequestOk {
            params: sorted(vec![Parameter::Expires(9000), Parameter::LargestObject(Location::new(3, 9))]),
            track_properties: props(vec![kv(prop::DEFAULT_PUBLISHER_GROUP_ORDER, 2)]),
        }),
    );
    c.control("REQUEST_OK answering PUBLISH", ControlMessage::RequestOk(RequestOk { params: sorted(vec![Parameter::Expires(1000)]), track_properties: Properties::new() }));
    c.control("REQUEST_ERROR with no Error Details", ControlMessage::RequestError(RequestError::new(RequestErrorCode::DOES_NOT_EXIST, 0, b"no such track".to_vec())));
    c.control("REQUEST_ERROR: UNAVAILABLE, with a Retry Interval", ControlMessage::RequestError(RequestError::new(RequestErrorCode::UNAVAILABLE, 2001, b"no stream credit".to_vec())));
    c.control(
        "REQUEST_ERROR: REDIRECT and its Error Detail",
        ControlMessage::RequestError(RequestError {
            code: RequestErrorCode::REDIRECT,
            retry_interval: 1,
            reason: vec![],
            details: vec![ErrorDetail::Redirect(Redirect { connect_uri: b"moqt://other.example/live".to_vec(), namespace: ns(&["example"]), name: name("cam1") })],
        }),
    );
    c.control(
        "REQUEST_ERROR: REDIRECT for a namespace request, an empty Track Name",
        ControlMessage::RequestError(RequestError {
            code: RequestErrorCode::REDIRECT,
            retry_interval: 0,
            reason: vec![],
            details: vec![ErrorDetail::Redirect(Redirect { connect_uri: vec![], namespace: ns(&["elsewhere"]), name: TrackName(vec![]) })],
        }),
    );
    c.control(
        "REQUEST_ERROR: UNSUPPORTED_EXTENSION naming two properties, and an unknown detail",
        ControlMessage::RequestError(RequestError {
            code: RequestErrorCode::UNSUPPORTED_EXTENSION,
            retry_interval: 0,
            reason: b"mandatory".to_vec(),
            details: vec![ErrorDetail::UnsupportedProperty(0x4002), ErrorDetail::UnsupportedProperty(0x4004), ErrorDetail::Other(kv(0x9d + 0x7f, 1))],
        }),
    );
    c.control("REQUEST_ERROR with a code this draft does not register", ControlMessage::RequestError(RequestError::new(RequestErrorCode(0x30), 30_001, b"was PREFIX_OVERLAP".to_vec())));

    // ── §6.3 a control stream: SETUP, then GOAWAY ──
    let setup = Setup {
        options: {
            let mut o = SetupOptions::new();
            o.push(kb(opt::MOQT_IMPLEMENTATION, b"example-impl/1.0"));
            o
        },
    };
    let goaway = ControlMessage::Goaway(Goaway { new_session_uri: vec![], timeout_ms: 1000 });
    c.stream(
        "control stream: SETUP then GOAWAY",
        encode_control_stream(&setup, std::slice::from_ref(&goaway)).unwrap(),
        GroupOrder::Ascending,
        vec![UniEvent::Header(StreamHeader::Control), UniEvent::Control(ControlMessage::Setup(setup.clone())), UniEvent::Control(goaway.clone())],
    );

    // ── §11.3 subgroup streams: -22's bytes ──
    let sg = |h: SubgroupHeader, objs: Vec<SubgroupObject>| -> (Vec<u8>, Vec<UniEvent>) {
        let b = encode_subgroup_stream(&h, &objs).unwrap();
        let mut ev = vec![UniEvent::Header(StreamHeader::Subgroup(h))];
        ev.extend(objs.into_iter().map(UniEvent::SubgroupObject));
        (b, ev)
    };
    let (b, ev) = sg(
        SubgroupHeader { track_alias: 2, group_id: 0, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: false, end_of_group: true, first_object: true },
        vec![SubgroupObject::normal(0, b"key".to_vec()), SubgroupObject::normal(1, b"delta".to_vec()), SubgroupObject { object_id: 2, properties: Properties::new(), status: ObjectStatus::EndOfGroup, payload: vec![] }],
    );
    c.stream("subgroup: Subgroup ID 0, a priority, End of Group", b, GroupOrder::Ascending, ev);
    let wai = wai_receipt(0x11, 0, "", vec![], 0x22);
    let (b, ev) = sg(
        SubgroupHeader { track_alias: 9, group_id: 7, subgroup_id: SubgroupId::FirstObjectId, publisher_priority: None, properties: true, end_of_group: false, first_object: false },
        vec![
            SubgroupObject { object_id: 5, properties: props(vec![kv(prop::PRIOR_OBJECT_ID_GAP, 2), kv(prop::OBJECT_DELIVERY_TIMEOUT, 40)]), status: ObjectStatus::Normal, payload: b"five".to_vec() },
            SubgroupObject { object_id: 6, properties: Properties::new(), status: ObjectStatus::Normal, payload: vec![] },
            SubgroupObject {
                object_id: 8,
                properties: with_immutable(vec![kv(0x3a40, 1), kv(0x3a40, 2)], vec![kb(0x3a41, &wai.to_bytes()), kv(0x3a42, 3), kv(0x3a44, 1)]),
                status: ObjectStatus::Normal,
                payload: b"WAI1".to_vec(),
            },
        ],
    );
    c.stream("subgroup: properties, an unknown one twice (a relay allows it)", b, GroupOrder::Ascending, ev);
    let (b, ev) = sg(
        SubgroupHeader { track_alias: 9, group_id: 8, subgroup_id: SubgroupId::Zero, publisher_priority: Some(1), properties: true, end_of_group: false, first_object: true },
        vec![SubgroupObject { object_id: 0, properties: props(vec![kb(0x9d + 0x7f * 128, b"g")]), status: ObjectStatus::Normal, payload: b"g".to_vec() }],
    );
    c.stream("subgroup: a grease value in the Mandatory Track Property range on an object, which is not one (#2037)", b, GroupOrder::Ascending, ev);
    let (b, ev) = sg(
        SubgroupHeader { track_alias: 1, group_id: 300, subgroup_id: SubgroupId::Explicit(3), publisher_priority: Some(0), properties: false, end_of_group: false, first_object: true },
        vec![SubgroupObject::normal(0, b"x".to_vec()), SubgroupObject { object_id: 1, properties: Properties::new(), status: ObjectStatus::EndOfTrack, payload: vec![] }],
    );
    c.stream("subgroup: an explicit Subgroup ID, End of Track", b, GroupOrder::Ascending, ev);

    // ── §11.4 fetch streams: -23's encoding ──
    let fo = |g: u64, o: u64, s: Option<u64>, p: u8, pr: Option<Properties>, pl: &[u8]| FetchEntry::Object(FetchObject { group_id: g, object_id: o, subgroup_id: s, publisher_priority: p, properties: pr, payload: pl.to_vec() });
    for (order, label, gs) in [(GroupOrder::Ascending, "Ascending", [1u64, 4, 6, 9]), (GroupOrder::Descending, "Descending", [9, 6, 4, 1])] {
        let entries = vec![
            fo(gs[0], 0, Some(0), 7, None, b"a"),
            fo(gs[0], 1, Some(0), 7, None, b"b"),
            fo(gs[0], 5, Some(1), 9, Some(props(vec![kv(prop::PRIOR_OBJECT_ID_GAP, 3)])), b""),
            FetchEntry::EndOfRange { kind: RangeEnd::Gone, location: Location::new(gs[0], 9) },
            FetchEntry::EndOfRange { kind: RangeEnd::NonExistent, location: Location::new(gs[1], 3) },
            fo(gs[1], 4, None, 9, None, b"d"),
            FetchEntry::EndOfGroup,
            FetchEntry::EndOfRange { kind: RangeEnd::Unknown, location: Location::new(gs[2], 0) },
            FetchEntry::EndOfRange { kind: RangeEnd::TimedOut, location: Location::new(gs[3], 2) },
            fo(gs[3], 3, Some(7), 9, None, b"e"),
            fo(gs[3], 4, Some(8), 2, None, b"f"),
            FetchEntry::EndOfTrack,
        ];
        let b = encode_fetch_stream(&FetchHeader { request_id: 4 }, order, &entries).unwrap();
        let mut ev = vec![UniEvent::Header(StreamHeader::Fetch(FetchHeader { request_id: 4 }))];
        ev.extend(entries.into_iter().map(UniEvent::FetchEntry));
        c.stream(&format!("fetch in {label} order, with each indicator"), b, order, ev);
    }
    let first_eor = vec![FetchEntry::EndOfRange { kind: RangeEnd::Gone, location: Location::new(2, 5) }, fo(2, 6, Some(0), 1, None, b"z"), fo(2, 9, Some(0), 1, None, b"y"), FetchEntry::EndOfGroup, fo(3, 0, Some(0), 1, None, b"w")];
    let b = encode_fetch_stream(&FetchHeader { request_id: 2 }, GroupOrder::Ascending, &first_eor).unwrap();
    let mut ev = vec![UniEvent::Header(StreamHeader::Fetch(FetchHeader { request_id: 2 }))];
    ev.extend(first_eor.into_iter().map(UniEvent::FetchEntry));
    c.stream("fetch beginning with End of Gone Range, an Object ID Delta, End of Group", b, GroupOrder::Ascending, ev);
    let mut fb = Vec::new();
    FetchHeader { request_id: 10 }.encode(&mut fb);
    c.stream("fetch with no objects", fb, GroupOrder::Ascending, vec![UniEvent::Header(StreamHeader::Fetch(FetchHeader { request_id: 10 }))]);

    // ── §11.5 padding ──
    c.stream("padding stream", encode_padding_stream(8), GroupOrder::Ascending, vec![UniEvent::Header(StreamHeader::Padding), UniEvent::Padding(8)]);

    // ── §11.2 datagrams: -22's bytes ──
    for (label, d) in [
        ("datagram: Object ID 0, default priority", ObjectDatagram { track_alias: 2, group_id: 5, object_id: 0, publisher_priority: None, properties: None, end_of_group: false, content: DatagramContent::Payload(b"opus".to_vec()) }),
        (
            "datagram: a priority, properties, End of Group",
            ObjectDatagram { track_alias: 2, group_id: 5, object_id: 9, publisher_priority: Some(3), properties: Some(props(vec![kv(prop::OBJECT_DELIVERY_TIMEOUT, 20)])), end_of_group: true, content: DatagramContent::Payload(b"last".to_vec()) },
        ),
        ("datagram: End of Track status", ObjectDatagram { track_alias: 2, group_id: 6, object_id: 0, publisher_priority: Some(3), properties: None, end_of_group: false, content: DatagramContent::Status(ObjectStatus::EndOfTrack) }),
    ] {
        let b = d.encode().unwrap();
        c.datagram(label, Datagram::Object(d), b, true);
    }
    c.datagram("padding datagram", Datagram::Padding(4), joule_moq::draft22::data::Datagram::padding(4), true);

    // ── the WAI_RECEIPT property's value (carriage), unchanged at -23 ──
    for (label, p) in [("WAI_RECEIPT: unmetered", wai_receipt(0x11, 0, "", vec![], 0x22)), ("WAI_RECEIPT: ModelBased", wai_receipt(0x77, 300, "ModelBased", vec![], 0x88))] {
        let pj = json!({"content_hash": hx(&p.content_hash), "joules_micro": p.joules_micro, "acquisition": p.acquisition, "acquisition_evidence": hx(&p.acquisition_evidence), "group_receipt": hx(&p.group_receipt)});
        c.push(label, "wai-receipt", &p.to_bytes(), true, None, pj);
    }
    let mut extra = wai_receipt(0x11, 0, "", vec![], 0x22).to_bytes();
    extra.push(0);
    c.bad("WAI_RECEIPT: a byte after the last field (a tampered property)", "wai-receipt", &extra, None, "invalid-property", "property_invalid");

    // ══ negative vectors: the planted faults ══
    // -22 forms a -23 endpoint meets.
    c.bad("-22's FETCH_OK (0x18), reserved at -23", "control", &frame(0x18, &[0, 1, 0, 0]), None, "unknown-message-type", UM);
    let sub_body = |params: &[u8], n: u64| cat(&[&v(0), &v(1), &v(1), b"a", &v(1), b"b", &v(n), params]);
    c.bad("SUBSCRIBE: -22's FORWARD (0x10)", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x10, 1], 1)), None, "unknown-parameter", IP);
    let pn_body = cat(&[&v(0), &v(1), &v(1), b"p", &v(0)]);
    c.bad("PUBLISH_NAMESPACE framed with -22's 16-bit Length", "control", &cat(&[&v(msg::PUBLISH_NAMESPACE), &(pn_body.len() as u16).to_be_bytes(), &pn_body]), None, "truncated", MM);
    c.bad("SUBSCRIBE: FILL_PARAMETERS length-prefixed as -22 writes it", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x23, 2, 0x0a, 0x64], 1)), None, "truncated", MM);
    // A -23 field left out.
    c.bad("PUBLISH_NAMESPACE without its Number of Parameters", "control", &frame(msg::PUBLISH_NAMESPACE, &cat(&[&v(0), &v(1), &v(1), b"p"])), None, "truncated", MM);
    c.bad("NAMESPACE without its Number of Parameters (-22's form)", "control", &frame(msg::NAMESPACE, &cat(&[&v(1), &v(2), b"42"])), None, "truncated", MM);
    c.bad("REQUEST_ERROR: REDIRECT without its Error Detail", "control", &frame(msg::REQUEST_ERROR, &cat(&[&v(0x34), &v(0), &v(0)])), None, "session-rule", MM);
    c.bad("REQUEST_ERROR: a REDIRECT Error Detail that does not read", "control", &frame(msg::REQUEST_ERROR, &cat(&[&v(0x34), &v(0), &v(0), &v(1), &v(2), &[0x05, 0x00]])), None, "session-rule", MKV);
    c.bad("SUBSCRIBE: an empty Track Name", "control", &frame(msg::SUBSCRIBE, &cat(&[&v(0), &v(1), &v(1), b"a", &v(0), &v(0)])), None, "session-rule", PV);
    // Framing and values.
    let mut huge = v(msg::GOAWAY);
    huge.extend(v(65536));
    c.bad("control: a Length past MAX_CONTROL_MESSAGE_SIZE", "control", &huge, None, "length-exceeds-limit", FLE);
    let good_sub = ControlMessage::Subscribe(TrackRequest { request_id: 0, namespace: ns(&["a"]), name: name("b"), params: sorted(vec![Parameter::Paused(true)]) }).encode().unwrap();
    c.bad("control: the Length runs past the bytes", "control", &good_sub[..good_sub.len() - 1], None, "truncated", MM);
    c.bad("control: a byte after the last field", "control", &frame(msg::REQUEST_UPDATE, &cat(&[&v(4), &v(0), &[0]])), None, "trailing-bytes", MM);
    c.bad("control: an unknown type", "control", &frame(0x3f, &[]), None, "unknown-message-type", UM);
    let size = |n: u64| cat(&[&v(v(n).len() as u64), &v(n)]);
    c.bad("SETUP: MAX_CONTROL_MESSAGE_SIZE below 65535", "control", &frame(msg::SETUP, &cat(&[&v(MAX_CONTROL_MESSAGE_SIZE), &size(65534)])), None, "invalid-value", MKV);
    c.bad("SETUP: MAX_CONTROL_MESSAGE_SIZE of 0, one byte", "control", &frame(msg::SETUP, &cat(&[&v(MAX_CONTROL_MESSAGE_SIZE), &[0x01, 0x00]])), None, "invalid-value", MKV);
    c.bad("SETUP: MAX_CONTROL_MESSAGE_SIZE twice", "control", &frame(msg::SETUP, &cat(&[&v(MAX_CONTROL_MESSAGE_SIZE), &size(70000), &v(0), &size(80000)])), None, "duplicate", PV);
    c.bad("SETUP: a token that does not read", "control", &frame(msg::SETUP, &[0x03, 1, 0x07]), None, "malformed-token", MAT);
    c.bad("SUBSCRIBE: PAUSED of 2", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x11, 2], 1)), None, "invalid-value", IP);
    c.bad("SUBSCRIBE: GROUP_ORDER of 0", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x22, 0], 1)), None, "invalid-value", IP);
    c.bad("SUBSCRIBE: a Location Filter Type of 6", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x21, 6], 1)), None, "invalid-value", IP);
    c.bad("SUBSCRIBE: PAUSED twice", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x11, 1, 0x00, 0], 2)), None, "duplicate", IP);
    // RELAY_PATH (0x3A51): 1 to 16 relay ids of 32 bytes, in the messages a request crosses a
    // link in.
    let path = |n: usize| cat(&[&v(param::RELAY_PATH), &v(n as u64), &vec![0x44u8; n]]);
    c.bad("SUBSCRIBE: a RELAY_PATH of 33 bytes", "control", &frame(msg::SUBSCRIBE, &sub_body(&path(33), 1)), None, "invalid-value", IP);
    c.bad("SUBSCRIBE: a RELAY_PATH of no relay", "control", &frame(msg::SUBSCRIBE, &sub_body(&path(0), 1)), None, "invalid-value", IP);
    c.bad("SUBSCRIBE: a RELAY_PATH of 17 relays", "control", &frame(msg::SUBSCRIBE, &sub_body(&path(32 * 17), 1)), None, "invalid-value", IP);
    c.bad("SUBSCRIBE_TRACKS: RELAY_PATH, which it does not list", "control", &frame(msg::SUBSCRIBE_TRACKS, &cat(&[&v(0), &v(0), &v(1), &path(32)])), None, "parameter-not-allowed", IP);
    c.bad("SUBSCRIBE: SUBSCRIBE_TRACKS_ID outside PUBLISH", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x36, 1], 1)), None, "parameter-not-allowed", IP);
    c.bad("SUBSCRIBE: FILL_PARAMETERS holding PAUSED", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x23, 1, 0x11, 1], 1)), None, "parameter-not-allowed", IP);
    c.bad(
        "SUBSCRIBE_TRACKS: RENDEZVOUS_TIMEOUT, a single track's",
        "control",
        &frame(msg::SUBSCRIBE_TRACKS, &cat(&[&v(0), &v(0), &v(1), &[0x04, 5]])),
        None,
        "parameter-not-allowed",
        IP,
    );
    c.bad("NAMESPACE: a parameter (none is defined)", "control", &frame(msg::NAMESPACE, &cat(&[&v(1), &v(1), b"x", &v(1), &[0x08, 5]])), None, "parameter-not-allowed", IP);
    let mut ov = vec![0x35, 1];
    ov.extend(v(u64::MAX));
    ov.push(1);
    c.bad("SUBSCRIBE: a parameter type past 2^64-1", "control", &frame(msg::SUBSCRIBE, &sub_body(&ov, 2)), None, "type-overflow", DEO);
    let mut eg = vec![0x21, 3];
    eg.extend(v(u64::MAX));
    eg.extend(v(0));
    eg.extend(v(1));
    c.bad("SUBSCRIBE: StartGroup + EndGroupDelta past 2^64-1", "control", &frame(msg::SUBSCRIBE, &sub_body(&eg, 1)), None, "overflow", DEO);
    c.bad("SUBSCRIBE: one OBJECTID_FILTER key twice", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x26, 2, 0, 1, 0x00, 2, 0, 4], 2)), None, "duplicate-filter", FILTER);
    c.bad("SUBSCRIBE: a PRIORITY_FILTER above 255", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x27, 4, 0, 0x81, 0x00, 0], 1)), None, "invalid-filter", FILTER);
    c.bad("SUBSCRIBE: a token of an unknown alias type", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x03, 2, 0x04, 0x00], 1)), None, "malformed-token", MAT);
    let mut many = v(33);
    for _ in 0..33 {
        many.extend(cat(&[&v(1), b"x"]));
    }
    c.bad("SUBSCRIBE_NAMESPACE: 33 namespace fields", "control", &frame(msg::SUBSCRIBE_NAMESPACE, &cat(&[&v(0), &many, &v(0)])), None, "too-many-namespace-fields", ITN);
    c.bad("SUBSCRIBE_NAMESPACE: an empty namespace field", "control", &frame(msg::SUBSCRIBE_NAMESPACE, &cat(&[&v(0), &v(1), &v(0), &v(0)])), None, "empty-namespace-field", ITN);
    let long_field = vec![b'n'; 4000];
    c.bad("SUBSCRIBE: a full track name past 4,096 bytes", "control", &frame(msg::SUBSCRIBE, &cat(&[&v(0), &v(1), &v(4000), &long_field, &v(200)])), None, "length-exceeds-limit", FLE);
    c.bad("GOAWAY: a URI of 8,193 bytes", "control", &frame(msg::GOAWAY, &v(8193)), None, "length-exceeds-limit", FLE);
    c.bad("SUBSCRIBE_OK: DEFAULT_PUBLISHER_PRIORITY of 256", "control", &frame(msg::SUBSCRIBE_OK, &cat(&[&v(1), &v(0), &v(prop::DEFAULT_PUBLISHER_PRIORITY), &v(256)])), None, "invalid-value", MKV);
    c.bad("SUBSCRIBE_OK: a Prior Group ID Gap among Track Properties (outside its scope)", "control", &frame(msg::SUBSCRIBE_OK, &[1, 0, 0x3c, 1]), None, "malformed-track", MT);
    c.bad("SUBSCRIBE_OK: MAX_CACHE_DURATION twice", "control", &frame(msg::SUBSCRIBE_OK, &[1, 0, 0x04, 10, 0x00, 20]), None, "malformed-track", MT);
    c.bad("SUBSCRIBE_OK: Immutable Properties inside Immutable Properties", "control", &frame(msg::SUBSCRIBE_OK, &[1, 0, 0x0b, 3, 0x0b, 1, 0x00]), None, "malformed-track", MT);

    // Streams and datagrams.
    let asc = Some(json!({"group_order": 1}));
    let setup_bytes = ControlMessage::Setup(setup.clone()).encode().unwrap();
    let g = goaway.encode().unwrap();
    c.bad("control stream: GOAWAY twice", "uni-stream", &cat(&[&setup_bytes, &g, &g]), asc.clone(), "unexpected-message", UM);
    let sub_msg = ControlMessage::Subscribe(TrackRequest { request_id: 0, namespace: ns(&["a"]), name: name("b"), params: Parameters::new() }).encode().unwrap();
    c.bad("control stream: SUBSCRIBE after SETUP", "uni-stream", &cat(&[&setup_bytes, &sub_msg]), asc.clone(), "unexpected-message", UM);
    c.bad("control stream: SETUP with -22's 16-bit Length", "uni-stream", &cat(&[&v(msg::SETUP), &[0x00, 0x00]]), asc.clone(), "unknown-message-type", UM);
    c.bad("stream: type 0x06", "uni-stream", &[0x06, 0], asc.clone(), "unknown-stream-type", ISH);
    c.bad("subgroup: SUBGROUP_ID_MODE 0b11", "uni-stream", &[0x16, 0, 0, 0, 0], asc.clone(), "invalid-flags", ISH);
    let sgh = |flags: u8| vec![flags, 1, 2, 7];
    c.bad("subgroup: Object Status 0x1", "uni-stream", &cat(&[&sgh(0x10), &[0, 0, 1]]), asc.clone(), "unknown-object-status", IOS);
    c.bad("subgroup: properties on End of Group", "uni-stream", &cat(&[&sgh(0x11), &[0, 2, 0x02, 5, 0, 3]]), asc.clone(), "properties-on-non-normal", IOS);
    c.bad("subgroup: a FIN inside an object", "uni-stream", &cat(&[&sgh(0x10), &[0, 5, b'a', b'b']]), asc.clone(), "truncated", TO);
    let mut big = sgh(0x10);
    big.extend(v(u64::MAX));
    big.extend([1, b'a', 0, 1, b'b']);
    c.bad("subgroup: an Object ID past 2^64-1", "uni-stream", &big, asc.clone(), "overflow", DEO);
    c.bad("subgroup: a Track-only property on an object (outside its scope)", "uni-stream", &cat(&[&sgh(0x11), &[0, 2, 0x0e, 5, 1, b'x']]), asc.clone(), "malformed-track", MT);
    c.bad("subgroup: OBJECT_DELIVERY_TIMEOUT twice (one instance allowed)", "uni-stream", &cat(&[&sgh(0x11), &[0, 4, 0x02, 5, 0x00, 6, 1, b'x']]), asc.clone(), "malformed-track", MT);
    let mut mand = sgh(0x11);
    mand.push(0);
    let mp = cat(&[&v(0x4000), &v(1)]);
    mand.extend(v(mp.len() as u64));
    mand.extend(mp);
    mand.extend([1, b'x']);
    c.bad("subgroup: a Mandatory Track Property on an object", "uni-stream", &mand, asc.clone(), "malformed-track", MT);
    let mut fh = Vec::new();
    FetchHeader { request_id: 0 }.encode(&mut fh);
    c.bad("fetch: a first object that refers back", "uni-stream", &cat(&[&fh, &[0x1d, 0, 0, 7, 1, b'a']]), asc.clone(), "no-prior-object", MO);
    c.bad("fetch: End of Group before any object", "uni-stream", &cat(&[&fh, &v(END_OF_GROUP)]), asc.clone(), "no-prior-object", MO);
    c.bad("fetch: Serialization Flags of 0x82", "uni-stream", &cat(&[&fh, &v(0x82), &[0, 0]]), asc.clone(), "invalid-flags", MO);
    c.bad("fetch: -22's reading of an Object ID Delta past 2^64-1 (+ 2)", "uni-stream", &cat(&[&fh, &[0x1c, 0], &v(u64::MAX - 1), &[7, 1, b'a', 0x04], &v(0), &[1, b'b']]), asc.clone(), "overflow", DEO);
    c.bad("fetch: an object with no Object ID Delta after End of Group", "uni-stream", &cat(&[&fh, &[0x1c, 0, 0, 7, 0], &v(END_OF_GROUP), &[0x08, 0, 0]]), asc.clone(), "overflow", DEO);
    c.bad("fetch: an End of Range below group 0 in Descending order", "uni-stream", &cat(&[&fh, &[0x1c, 1, 0, 7, 1, b'a'], &v(0x10c), &[2, 0]]), Some(json!({"group_order": 2})), "overflow", DEO);
    c.bad("fetch: a FIN inside an object", "uni-stream", &cat(&[&fh, &[0x1c, 1, 0, 7, 3, b'a']]), asc.clone(), "truncated", TO);
    c.bad("fetch: a Track-only property on an object", "uni-stream", &cat(&[&fh, &[0x3c, 1, 0, 7, 2, 0x30, 1, 1, b'a']]), asc.clone(), "malformed-track", MT);
    c.bad("datagram: STATUS and END_OF_GROUP", "datagram", &[0x22, 1, 2, 3, 0], None, "invalid-flags", ISH);
    c.bad("datagram: PROPERTIES with a length of 0", "datagram", &[0x0d, 1, 2, 0, b'x'], None, "empty-datagram-properties", MO);
    c.bad("datagram: an unknown status", "datagram", &[0x2c, 1, 2, 2], None, "unknown-object-status", IOS);
    c.bad("datagram: a byte after the status", "datagram", &[0x2c, 1, 2, 0, 0], None, "trailing-bytes", MO);
    c.bad("datagram: cut short", "datagram", &[0x00, 1], None, "truncated", MO);
    c.bad("datagram: a Track-only property on an object", "datagram", &[0x0d, 1, 2, 2, 0x22, 1, b'x'], None, "malformed-track", MT);
    let mp = cat(&[&v(0x4000), &v(1)]);
    c.bad("datagram: a Mandatory Track Property on an object", "datagram", &cat(&[&[0x0d, 1, 2], &v(mp.len() as u64), &mp, b"x"]), None, "malformed-track", MT);
    let _ = IF;
    c.vectors
}

fn corpus_doc(vectors: Vec<Value>) -> Value {
    json!({
        "about": "draft-ietf-moq-transport-23 wire vectors: varints, control messages, unidirectional streams (control, subgroup, fetch, padding), datagrams, and the value of the WAI_RECEIPT object property of this crate's carriage. Each vector gives its bytes in hex and either its reading, in the JSON view of joule_moq::draft23::json described in the crate README, or the refusal it must meet with -23's consequence (the code -23 gives the fault on its plane: a control message, a data stream, a datagram). A uni-stream vector is the stream's bytes up to its FIN; its context gives the Group Order of the request a fetch stream answers (1 Ascending, 2 Descending). A vector marked canonical is the form this crate's encoder makes. Generated by tests/draft23_vectors.rs; regenerate with JOULE_MOQ_WRITE_VECTORS=1.",
        "draft": "draft-ietf-moq-transport-23",
        "vectors": vectors,
    })
}

fn unhex(s: &str) -> Vec<u8> {
    (0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap()).collect()
}

fn order_of(v: &Value) -> GroupOrder {
    match v["context"]["group_order"].as_u64() {
        Some(2) => GroupOrder::Descending,
        _ => GroupOrder::Ascending,
    }
}

fn decode(v: &Value) -> Result<(Value, Option<Vec<u8>>), Value> {
    let b = unhex(v["hex"].as_str().unwrap());
    match v["kind"].as_str().unwrap() {
        "varint" => {
            let (val, n) = varint::get(&b, "varint").map_err(|e| json::error(&e, Plane::Control))?;
            Ok((json!({"value": val, "length": n}), Some(varint::to_vec(val))))
        }
        "control" => {
            let m = ControlMessage::decode_exact(&b).map_err(|e| json::error(&e, Plane::Control))?;
            Ok((json::control(&m), Some(m.encode().unwrap())))
        }
        "uni-stream" => {
            let order = order_of(v);
            let ev = decode_uni_stream(&b, order).map_err(|e| json::error(&e, stream_plane(&b)))?;
            Ok((Value::Array(ev.iter().map(json::uni_event).collect()), reencode(&ev, order)))
        }
        "datagram" => {
            let d = Datagram::decode(&b).map_err(|e| json::error(&e, Plane::Datagram))?;
            Ok((json::datagram(&d), Some(d.encode().unwrap())))
        }
        "wai-receipt" => match WaiReceiptProperty::from_bytes(&b) {
            Ok(p) => Ok((
                json!({"content_hash": hx(&p.content_hash), "joules_micro": p.joules_micro, "acquisition": p.acquisition, "acquisition_evidence": hx(&p.acquisition_evidence), "group_receipt": hx(&p.group_receipt)}),
                Some(p.to_bytes()),
            )),
            Err(_) => Err(json!({"error": "invalid-property", "consequence": "property_invalid"})),
        },
        k => panic!("unknown kind {k}"),
    }
}

fn reencode(ev: &[UniEvent], order: GroupOrder) -> Option<Vec<u8>> {
    let mut out = Vec::new();
    let mut sub: Option<SubgroupCodec> = None;
    let mut fetch: Option<FetchCodec> = None;
    for e in ev {
        match e {
            UniEvent::Header(StreamHeader::Control) => {}
            UniEvent::Header(StreamHeader::Padding) => out.extend(varint::to_vec(stream_type::PADDING)),
            UniEvent::Header(StreamHeader::Subgroup(h)) => {
                h.encode(&mut out);
                sub = Some(SubgroupCodec::new(*h));
            }
            UniEvent::Header(StreamHeader::Fetch(h)) => {
                h.encode(&mut out);
                fetch = Some(FetchCodec::new(order));
            }
            UniEvent::Control(m) => out.extend(m.encode().ok()?),
            UniEvent::SubgroupObject(o) => sub.as_mut()?.encode_object(o, &mut out).ok()?,
            UniEvent::FetchEntry(f) => fetch.as_mut()?.encode(f, &mut out).ok()?,
            UniEvent::Padding(n) => out.resize(out.len() + n, 0),
        }
    }
    Some(out)
}

fn decode_incrementally(b: &[u8], order: GroupOrder, step: usize) -> Result<Vec<UniEvent>, WireError> {
    let mut d = UniStreamDecoder::new(order, Limits { max_payload: u64::MAX, max_properties: u64::MAX });
    let mut out = Vec::new();
    for chunk in b.chunks(step.max(1)) {
        d.push(chunk);
        while let Some(e) = d.next()? {
            out.push(e);
        }
    }
    d.finish()?;
    Ok(out)
}

#[test]
fn the_draft23_corpus_is_current_and_holds() {
    let doc = corpus_doc(build());
    let text = serde_json::to_string_pretty(&doc).unwrap() + "\n";
    if std::env::var("JOULE_MOQ_WRITE_VECTORS").as_deref() == Ok("1") {
        std::fs::create_dir_all(std::path::Path::new(CORPUS).parent().unwrap()).unwrap();
        std::fs::write(CORPUS, &text).unwrap();
    }
    let committed = std::fs::read_to_string(CORPUS).unwrap_or_else(|e| panic!("read {CORPUS}: {e}"));
    assert!(committed == text, "{CORPUS} is not what tests/draft23_vectors.rs generates; regenerate it with JOULE_MOQ_WRITE_VECTORS=1");

    let mut failures = Vec::new();
    let (mut good, mut bad) = (0, 0);
    for v in doc["vectors"].as_array().unwrap() {
        let name = v["name"].as_str().unwrap();
        let expect = &v["expect"];
        let b = unhex(v["hex"].as_str().unwrap());
        match (expect.get("error"), decode(v)) {
            (None, Ok((got, again))) => {
                good += 1;
                if &got != expect {
                    failures.push(format!("{name}: read as {got}"));
                }
                if v["canonical"] == true && again.as_deref() != Some(&b[..]) {
                    failures.push(format!("{name}: re-encodes to other bytes"));
                }
            }
            (None, Err(e)) => failures.push(format!("{name}: refused ({e}); the corpus reads it")),
            (Some(_), Ok((got, _))) => failures.push(format!("{name}: read as {got}; the corpus refuses it")),
            (Some(_), Err(e)) => {
                bad += 1;
                if &e != expect {
                    failures.push(format!("{name}: refused as {e}, the corpus says {expect}"));
                }
            }
        }
        if v["kind"] == "uni-stream" {
            for step in [1, 2, 7, b.len().max(1)] {
                let inc = decode_incrementally(&b, order_of(v), step);
                match (expect.get("error"), inc) {
                    (None, Ok(ev)) => {
                        let mut merged: Vec<UniEvent> = Vec::new();
                        for e in ev {
                            match (merged.last_mut(), e) {
                                (Some(UniEvent::Padding(n)), UniEvent::Padding(k)) => *n += k,
                                (_, e) => merged.push(e),
                            }
                        }
                        let got = Value::Array(merged.iter().map(json::uni_event).collect());
                        if &got != expect {
                            failures.push(format!("{name}: in pieces of {step}, read as {got}"));
                        }
                    }
                    (None, Err(e)) => failures.push(format!("{name}: in pieces of {step}, refused ({e})")),
                    (Some(_), Ok(_)) => failures.push(format!("{name}: in pieces of {step}, accepted")),
                    (Some(_), Err(e)) => {
                        // The incremental decoder refuses too, with the same consequence.
                        let got = json::error(&e, stream_plane(&b));
                        if got["consequence"] != expect["consequence"] {
                            failures.push(format!("{name}: in pieces of {step}, refused as {got}, the corpus says {expect}"));
                        }
                    }
                }
            }
        }
    }
    assert!(failures.is_empty(), "{} vector(s) disagree:\n{}", failures.len(), failures.join("\n"));
    assert!(good > 0 && bad > 0, "the corpus holds both kinds");
}