//! Grants that change while the relay runs, and a window closed with its sessions still open
//! (`src/control.rs`), on the operator's surface:
//!
//! - **A withdrawn grant ends its sessions alone.** Two rooms, each a publisher and a
//!   subscriber under a token of its own; the first room's token is withdrawn on the control
//!   socket: its two sessions are closed `UNAUTHORIZED`, `grant_revoked`, while the other room
//!   goes on delivering objects with no GOAWAY. A token added at run time is admitted, the
//!   withdrawn one refused, and a metered grant added at run time names the grant in the
//!   ledger of a session already open.
//! - **A token a request presented** ends its session the same way when it is withdrawn, and
//!   a change that leaves it alone leaves that session alone.
//! - **A metered grant withdrawn** ends the sessions under it, with no rule naming its token,
//!   and a `set` that maps the token to another grant moves its session there.
//! - **A change that narrows a rule** ends the sessions holding what the rule no longer covers,
//!   a subscription moved off its prefix or a publication whose right went, and none other: a
//!   session that loses a right it does not use goes on.
//! - **The operator's surface is local.** A `listen` address off loopback refuses the relay's
//!   start; the socket is made with mode 0600; a peer of another user, or off loopback, is
//!   refused; a connection from a non-loopback address of this host to the loopback listener
//!   is answered with a refusal; a request of another protocol is answered and ends its
//!   connection, its body unread.
//! - **A `set` states what it replaces**: one that leaves out `required` or `metered` does not
//!   read and changes nothing, and a reload leaves `required` as it stands.
//! - **An authority's revocation list** (WAI live-rooms §5): a grant revoked by its digest or
//!   by its viewer's tag is refused at its SETUP, `grant_revoked`, before its proof; a grant
//!   revoked while its session is open ends that session alone; and an authority's grant is
//!   metered from its own ceilings with no metered grant named. A grant whose funds ceiling and
//!   currency break live-rooms §4.3's rule is refused at its SETUP, `grant_invalid`.
//! - **A revocation refuses every use** (live-rooms §5): it ends a session that proved nothing,
//!   on a relay that requires no authorization; one a rule also admits, and one whose request
//!   presented the grant, whether or not a rule names it; a request that presents a revoked
//!   grant is refused, on a relay that requires no authorization too, its session standing and
//!   no later change ending it for that refused use; and the list holds once its authority is
//!   dropped. The records a session's requests may present are bounded.
//! - **An authority** added while the relay runs is honoured from then, and one dropped ends the
//!   sessions that proved its grants and no other authority's.
//! - **A session's grant in a room**: one that holds grants for two rooms is in each room's
//!   ledger under that room's grant, and in a third room's under neither, whether an authority
//!   signed them or a rule with `proof` and a metered grant name them.
//! - **A revocation stays** (live-rooms §5): through a reload, a `set` that does not repeat it,
//!   and a restart, which reads the file the relay kept it in.
//! - **A flush inside a group** (WAI live-metering §3.8): the window is cut after a group's
//!   first subgroup and before its second; the second, which arrives after the cut, is counted
//!   in the window its group began in, the next group in the next; the record the control
//!   socket signs over the first window, with every session still open, passes
//!   `wai/live-conformance/live_metering_verify.py`'s reading of a record and its ledger; and
//!   each session's windows sum to its whole summary.
//! - **A first flush**: a relay never flushed holds no second copy of its sessions' counts, and
//!   its first flush puts everything counted so far in window 0, the ended sessions' among it.
//! - **An interval** seals with nobody asking, and a relay told to hold a ceiling without one does
//!   not start.
//! - **A window with a DVR store** (§3.9): a group cut by a flush is stated where its in entries
//!   allow, sealed before or after the store wrote it, and every record through the verifier.
//! - **SIGHUP** reads the grants file again: a token it drops ends its session, one it adds
//!   is admitted.

// The control socket and SIGHUP, which every test here goes through, are Unix's. The relay off
// Unix serves `control.listen` alone.
#![cfg(unix)]

mod common;

use common::*;
use ed25519_dalek::{Signer, SigningKey};
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{Parameter, Parameters, RequestErrorCode, SessionError, Token};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, DvrConfig, DvrWindow, Grant, MeteredGrant, Revocation};
use joule_moq_relay::control::{admits, own_uid, Peer};
use joule_moq_relay::metering::HopRecord;
use joule_moq_relay::summary::SessionSummary;
use joule_moq_relay::transport::TransportError;
use joule_moq_relay::{RelayConfig, RelayServer};
use serde_json::{json, Value};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::time::timeout;

fn now_us() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
}

fn sha_hex(b: &[u8]) -> String {
    hex::encode(joule_moq_relay::auth::sha256(b))
}

/// SIGHUP goes to the whole process, so to every relay here that names a grants file: the tests
/// that name one take turns.
static GRANTS_FILE: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

/// A socket path of this test's own, short enough for every host's limit.
fn socket_path(what: &str) -> PathBuf {
    std::env::temp_dir().join(format!("wai-relay-ctl-{}-{what}.sock", std::process::id()))
}

/// One request on the control socket at `path`, and its answer.
async fn ask(path: &Path, req: Value) -> Value {
    timeout(STEP * 2, async {
        let s = tokio::net::UnixStream::connect(path).await.expect("the control socket");
        let (r, mut w) = s.into_split();
        w.write_all(format!("{req}\n").as_bytes()).await.unwrap();
        let mut line = String::new();
        BufReader::new(r).read_line(&mut line).await.unwrap();
        serde_json::from_str::<Value>(&line).unwrap_or_else(|e| panic!("an answer: {e}: {line:?}"))
    })
    .await
    .expect("an answer in time")
}

fn rule(prefix: &[&str], token: &[u8], publish: bool, subscribe: bool) -> Grant {
    Grant { prefix: prefix.iter().map(|s| s.to_string()).collect(), token_sha256: sha_hex(token), publish, subscribe, proof: false, unlimited: false, seek: false }
}

fn bearer(token: &[u8]) -> ClientOptions {
    ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: token.to_vec() }], ..ClientOptions::default() }
}

/// The session `c` is closed by the relay with `code`; its reason.
async fn closed_with(c: &Client, code: SessionError, what: &str) -> String {
    match timeout(STEP, c.conn.closed()).await.unwrap_or_else(|_| panic!("{what}: not closed in time")) {
        TransportError::Closed { code: got, reason, .. } => {
            assert_eq!(got, code.0, "{what}: {reason}");
            reason
        }
        other => panic!("{what}: {other:?}"),
    }
}

/// Every session closed and the summaries of `n` kept, each wait on its own bound. A session
/// leaves the relay's open sessions as its end begins, and its summary is kept once its scheduler
/// has stopped and its counts are closed (`session::cleanup`), which on a loaded host comes
/// later: so a test that reads the ended sessions' summaries, or a record made of them, waits for
/// the summaries themselves. `n` is every session the relay admitted (one refused at its SETUP
/// has none), and no more are kept.
async fn ended(relay: &joule_moq_relay::Relay, n: usize) {
    assert!(until(STEP, || relay.stats().sessions_open == 0).await, "every session closed: {:?}", relay.stats());
    assert!(until(STEP, || relay.summaries().len() >= n).await, "{n} sessions' summaries kept: {} so far", relay.summaries().len());
    assert_eq!(relay.summaries().len(), n, "a summary of each session admitted, and no other");
}

/// The media streams of `alias` read whole from `got`, by Group ID.
fn groups_of(got: &[Got], alias: u64) -> Vec<u64> {
    let mut g: Vec<u64> = got.iter().filter(|x| x.header.track_alias == alias && x.reset.is_none() && !x.objects.is_empty()).map(|x| x.header.group_id).collect();
    g.sort();
    g.dedup();
    g
}

// ── A withdrawn grant ends its sessions alone ───────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn withdrawing_one_grant_ends_only_its_sessions() {
    timeout(Duration::from_secs(180), withdraw_run()).await.expect("the test finishes in time");
}

async fn withdraw_run() {
    let (ta_tok, tb_tok, tc_tok) = (b"room-a-token".to_vec(), b"room-b-token".to_vec(), b"room-b-late-token".to_vec());
    let sock = socket_path("withdraw");
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.grants = vec![rule(&["roomA"], &ta_tok, true, true), rule(&["roomB"], &tb_tok, true, true)];
        c.control.socket = Some(sock.clone());
    })
    .await;
    let relay = server.relay().clone();
    let ta = Tracks::new(ns(&["roomA", "p"]), "cam", 10, 4, 128);
    let tb = Tracks::new(ns(&["roomB", "p"]), "cam", 10, 4, 128);
    let (key_a, key_b) = (SigningKey::from_bytes(&[0xa1; 32]), SigningKey::from_bytes(&[0xb2; 32]));

    let (pa, _ea) = connect(&server, Via::Quic, bearer(&ta_tok)).await;
    let (pb, _eb) = connect(&server, Via::Quic, bearer(&tb_tok)).await;
    let _pa = publish_tracks(&pa, &ta, None).await;
    let _pb = publish_tracks(&pb, &tb, None).await;
    let (sa, _esa) = connect(&server, Via::WebSocket, bearer(&ta_tok)).await;
    let (sb, _esb) = connect(&server, Via::Quic, bearer(&tb_tok)).await;
    let va = subscribe_both(&sa, &ta, Parameters::new()).await;
    let vb = subscribe_both(&sb, &tb, Parameters::new()).await;
    publish_group(&pa, &key_a, &ta, 0, &payload("a", 0, 400)).await;
    publish_group(&pb, &key_b, &tb, 0, &payload("b", 0, 400)).await;
    let got = collect_concurrent(&sa, STEP, Duration::from_millis(300), |g| groups_of(g, va.media_alias) == [0]).await;
    assert_eq!(groups_of(&got, va.media_alias), vec![0], "room A delivers before the change");
    let got = collect_concurrent(&sb, STEP, Duration::from_millis(300), |g| groups_of(g, vb.media_alias) == [0]).await;
    assert_eq!(groups_of(&got, vb.media_alias), vec![0], "room B delivers before the change");

    // The grants in force, as the socket shows them.
    let shown = ask(&sock, json!({"op": "show"})).await;
    assert_eq!(shown["ok"], true, "{shown}");
    assert_eq!(shown["auth"]["grants"].as_array().unwrap().len(), 2, "{shown}");

    // Room A's token is withdrawn: its two sessions end, with the reason, and only they.
    let a = ask(&sock, json!({"op": "withdraw", "token_sha256": sha_hex(&ta_tok)})).await;
    assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(2)), "{a}");
    for (c, what) in [(&pa, "room A's publisher"), (&sa, "room A's subscriber")] {
        let why = closed_with(c, SessionError::UNAUTHORIZED, what).await;
        assert!(why.starts_with("grant_revoked"), "{what}: {why}");
    }
    assert!(until(STEP, || relay.stats().sessions_open == 2).await, "room B's two sessions stand");

    // A metered grant added now names the grant of room B's sessions, already open.
    let grant_b = "b0".repeat(32);
    let m = ask(&sock, json!({"op": "meter", "grant": {"token_sha256": sha_hex(&tb_tok), "grant": grant_b, "joules_ceiling_micro": 7_000_000u64, "funds_ceiling": 900, "currency": "USD"}})).await;
    assert_eq!((m["ok"].as_bool(), m["ended"].as_u64()), (Some(true), Some(0)), "{m}");

    // Room B goes on delivering, with no GOAWAY.
    for g in 1..=4 {
        publish_group(&pb, &key_b, &tb, g, &payload("b", g, 400)).await;
    }
    let got = collect_concurrent(&sb, STEP, Duration::from_millis(300), |g| groups_of(g, vb.media_alias) == [1, 2, 3, 4]).await;
    assert_eq!(groups_of(&got, vb.media_alias), vec![1, 2, 3, 4], "room B delivers every group after the change");
    assert!(pb.goaway().borrow().is_none() && sb.goaway().borrow().is_none(), "no GOAWAY to room B");

    // The withdrawn token is refused; a token added at run time is admitted.
    let (x, _ex) = connect(&server, Via::Quic, bearer(&ta_tok)).await;
    let refused = x.subscribe(ta.ns.clone(), ta.media.clone(), Parameters::new()).await.expect_err("a withdrawn token");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
    let (late, _el) = connect(&server, Via::Quic, bearer(&tc_tok)).await;
    let refused = late.subscribe(tb.ns.clone(), tb.media.clone(), Parameters::new()).await.expect_err("a token not yet granted");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
    let g = ask(&sock, json!({"op": "grant", "grant": {"prefix": ["roomB"], "token_sha256": sha_hex(&tc_tok), "subscribe": true}})).await;
    assert_eq!(g["ok"], true, "{g}");
    let vl = subscribe_both(&late, &tb, Parameters::new()).await;
    publish_group(&pb, &key_b, &tb, 5, &payload("b", 5, 400)).await;
    let got = collect_concurrent(&late, STEP, Duration::from_millis(300), |g| groups_of(g, vl.media_alias).contains(&5)).await;
    assert!(groups_of(&got, vl.media_alias).contains(&5), "the token granted at run time is served");
    // A request that does not read, and a change that does not check, change nothing.
    let bad = ask(&sock, json!({"op": "grant", "grant": {"prefix": [], "token_sha256": "zz"}})).await;
    assert_eq!(bad["ok"], false, "{bad}");
    let bad = ask(&sock, json!({"op": "shutdown"})).await;
    assert_eq!(bad["ok"], false, "{bad}");
    assert_eq!(relay.auth_config().grants.len(), 2, "room B's rule and the late one");

    for c in [&pb, &sb, &x, &late] {
        c.close(0, "done");
    }
    // Room A's two sessions, room B's two, the withdrawn token's and the late one's.
    ended(&relay, 6).await;
    let summaries: Vec<SessionSummary> = relay.summaries().into_iter().map(|s| s.summary).collect();
    let metered: Vec<_> = summaries.iter().filter_map(|s| s.tally.room_grant.as_ref()).collect();
    assert_eq!(metered.len(), 2, "room B's publisher and subscriber ran under the grant metered at run time: {metered:?}");
    assert!(metered.iter().all(|g| g.grant == grant_b && g.joules_ceiling_micro == 7_000_000 && g.funds_ceiling == 900 && g.currency == "USD"));
    server.shutdown().await;
    assert!(!sock.exists(), "the socket is removed at shutdown");
}

// ── The operator's surface is local ─────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_control_request_from_outside_is_refused() {
    timeout(Duration::from_secs(120), outside_run()).await.expect("the test finishes in time");
}

async fn outside_run() {
    // Off loopback, the relay does not start.
    for a in ["0.0.0.0:0", "[::]:0", "192.0.2.7:4490"] {
        let mut c = RelayConfig::loopback();
        c.control.listen = Some(a.parse().unwrap());
        let e = RelayServer::start(c).await.err().unwrap_or_else(|| panic!("a control listener on {a} was accepted"));
        assert!(e.contains("loopback"), "{a}: {e}");
    }
    // A peer is served only on the relay's own user's socket, or from loopback.
    let me = own_uid();
    assert!(admits(&Peer::Unix { uid: me }, me));
    assert!(!admits(&Peer::Unix { uid: me.wrapping_add(1).max(1) }, me), "another user's peer");
    assert!(admits(&Peer::Tcp("127.0.0.1:5000".parse().unwrap()), me) && admits(&Peer::Tcp("[::1]:5000".parse().unwrap()), me));
    for a in ["192.0.2.7:5000", "10.0.0.1:5000", "[2001:db8::7]:5000"] {
        assert!(!admits(&Peer::Tcp(a.parse().unwrap()), me), "{a}");
    }

    let sock = socket_path("outside");
    let server = relay(|c| {
        c.control.socket = Some(sock.clone());
        c.control.listen = Some("127.0.0.1:0".parse().unwrap());
    })
    .await;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let mode = std::fs::metadata(&sock).expect("the socket").permissions().mode() & 0o777;
        assert_eq!(mode, 0o600, "the socket is its owner's alone");
    }
    // From loopback, served.
    let addr = server.control_addr.expect("the loopback listener");
    let answer = |s: tokio::net::TcpStream| async move {
        let (r, mut w) = s.into_split();
        let _ = w.write_all(b"{\"op\":\"show\"}\n").await;
        let mut line = String::new();
        let _ = timeout(STEP, BufReader::new(r).read_line(&mut line)).await;
        serde_json::from_str::<Value>(&line).unwrap_or(Value::Null)
    };
    let a = answer(tokio::net::TcpStream::connect(addr).await.unwrap()).await;
    assert_eq!(a["ok"], true, "{a}");
    // A request of another protocol is answered and ends its connection, its body unread.
    let body = json!({"op": "grant", "grant": {"prefix": ["x"], "token_sha256": "ab".repeat(32), "subscribe": true}}).to_string();
    let mut http = tokio::net::TcpStream::connect(addr).await.unwrap();
    http.write_all(format!("POST / HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: text/plain\r\nContent-Length: {}\r\n\r\n{body}\n", body.len() + 1).as_bytes()).await.unwrap();
    let mut said = String::new();
    let _ = timeout(STEP, http.read_to_string(&mut said)).await.expect("the connection ends");
    assert!(server.relay().auth_config().grants.is_empty(), "a body after a line of another protocol was obeyed: {said}");
    // From an address of this host that is not loopback, to the loopback listener: refused.
    // Where the host has no such address, or will not route it to loopback, this part is
    // said to be skipped.
    let outward = std::net::UdpSocket::bind("0.0.0.0:0").and_then(|u| u.connect("192.0.2.1:9").and_then(|_| u.local_addr())).ok().map(|a| a.ip()).filter(|ip| !ip.is_loopback() && !ip.is_unspecified());
    match outward {
        Some(ip) => {
            let s = tokio::net::TcpSocket::new_v4().unwrap();
            match s.bind((ip, 0).into()) {
                Ok(()) => match timeout(STEP, s.connect(addr)).await {
                    Ok(Ok(stream)) => {
                        assert_eq!(stream.local_addr().unwrap().ip(), ip);
                        let a = answer(stream).await;
                        assert_eq!(a["ok"], false, "a peer at {ip} was served: {a}");
                        assert!(a["error"].as_str().unwrap_or("").contains("refused"), "{a}");
                    }
                    other => eprintln!("SKIPPED: a connection from {ip} to {addr}: {other:?}"),
                },
                Err(e) => eprintln!("SKIPPED: binding {ip}: {e}"),
            }
        }
        None => eprintln!("SKIPPED: this host has no address off loopback"),
    }
    server.shutdown().await;
}

/// As root, a peer of another user in earnest: it drops to that user, connects to the socket and
/// asks, and the relay refuses it by what the kernel states of it, though the socket's mode
/// would let it in. No other user can be two users on one host, so elsewhere this says it was
/// skipped, and `src/control.rs`'s test of the socket holds the rule.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_peer_of_another_user_is_refused_on_the_socket() {
    timeout(Duration::from_secs(60), other_user_run()).await.expect("the test finishes in time");
}

async fn other_user_run() {
    if own_uid() != 0 {
        eprintln!("SKIPPED: only root connects as another user");
        return;
    }
    use std::os::unix::fs::PermissionsExt;
    let sock = socket_path("other-user");
    let server = relay(|c| c.control.socket = Some(sock.clone())).await;
    std::fs::set_permissions(&sock, std::fs::Permissions::from_mode(0o666)).unwrap();
    let req = format!("{}\n", json!({"op": "grant", "grant": {"prefix": ["x"], "token_sha256": "ab".repeat(32), "subscribe": true}}));
    // SAFETY: an all-zero sockaddr_un is one with an empty path.
    let mut addr: libc::sockaddr_un = unsafe { std::mem::zeroed() };
    addr.sun_family = libc::AF_UNIX as _;
    for (to, from) in addr.sun_path.iter_mut().zip(sock.as_os_str().as_encoded_bytes()) {
        *to = *from as libc::c_char;
    }
    let status = tokio::task::spawn_blocking(move || {
        // SAFETY: the child of a threaded process calls only async-signal-safe functions, on what
        // was made before the fork, and leaves by _exit.
        unsafe {
            let pid = libc::fork();
            assert!(pid >= 0, "fork");
            if pid == 0 {
                let mut answer = [0u8; 256];
                let s = libc::socket(libc::AF_UNIX, libc::SOCK_STREAM, 0);
                let connected = libc::setuid(65534) == 0 && s >= 0 && libc::connect(s, (&raw const addr).cast(), std::mem::size_of::<libc::sockaddr_un>() as libc::socklen_t) == 0;
                // A refusal reads no request, so this write may find the socket closed: the answer
                // is there to read all the same.
                let n = if connected {
                    libc::write(s, req.as_ptr().cast(), req.len());
                    libc::read(s, answer.as_mut_ptr().cast(), answer.len())
                } else {
                    -1
                };
                let refused = n > 0 && answer[..n as usize].windows(7).any(|w| w == b"refused");
                libc::_exit(match (n > 0, refused) {
                    (true, true) => 0,
                    (true, false) => 1,
                    _ => 2,
                });
            }
            let mut status = 0;
            assert_eq!(libc::waitpid(pid, &mut status, 0), pid);
            status
        }
    })
    .await
    .unwrap();
    assert!(libc::WIFEXITED(status), "the peer ended by signal: {status}");
    assert_eq!(libc::WEXITSTATUS(status), 0, "0 refused, 1 served, 2 no answer");
    assert!(server.relay().auth_config().grants.is_empty(), "its request was obeyed");
    server.shutdown().await;
}

// ── A token a request presented ─────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_token_a_request_presented_ends_its_session_when_withdrawn() {
    timeout(Duration::from_secs(120), presented_run()).await.expect("the test finishes in time");
}

async fn presented_run() {
    let (publisher_tok, request_tok, other_tok) = (b"publisher".to_vec(), b"in-a-request".to_vec(), b"another".to_vec());
    let sock = socket_path("presented");
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.grants = vec![rule(&["room"], &publisher_tok, true, false), rule(&["room"], &request_tok, false, true)];
        c.control.socket = Some(sock.clone());
    })
    .await;
    let relay = server.relay().clone();
    let t = Tracks::new(ns(&["room", "p"]), "cam", 10, 4, 128);
    let key = SigningKey::from_bytes(&[0xd4; 32]);
    let (p, _ep) = connect(&server, Via::Quic, bearer(&publisher_tok)).await;
    let _pp = publish_tracks(&p, &t, None).await;
    // No token in its SETUP: its subscriptions stand on the token each request carries.
    let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let with_token = Parameters::new().with(Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: request_tok.clone() }));
    let v = subscribe_both_with(&s, &t, with_token.clone(), with_token).await;
    publish_group(&p, &key, &t, 0, &payload("r", 0, 300)).await;
    let got = collect_concurrent(&s, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [0]).await;
    assert_eq!(groups_of(&got, v.media_alias), vec![0], "served on the token of its requests");

    // A change that leaves its token alone leaves it alone.
    let a = ask(&sock, json!({"op": "grant", "grant": {"prefix": ["elsewhere"], "token_sha256": sha_hex(&other_tok), "subscribe": true}})).await;
    assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(0)), "{a}");
    publish_group(&p, &key, &t, 1, &payload("r", 1, 300)).await;
    let got = collect_concurrent(&s, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [1]).await;
    assert_eq!(groups_of(&got, v.media_alias), vec![1], "it goes on");

    // Its token withdrawn, it ends, with the reason; the publisher, under another, does not.
    let a = ask(&sock, json!({"op": "withdraw", "token_sha256": sha_hex(&request_tok)})).await;
    assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(1)), "{a}");
    let why = closed_with(&s, SessionError::UNAUTHORIZED, "the session of a request's token").await;
    assert!(why.starts_with("grant_revoked") && why.contains("a request presented"), "{why}");
    assert!(until(STEP, || relay.stats().sessions_open == 1).await, "the publisher stands");
    p.close(0, "done");
    server.shutdown().await;
}

// ── A change that narrows a rule ────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_change_that_narrows_a_rule_ends_what_it_no_longer_covers() {
    timeout(Duration::from_secs(180), narrow_run()).await.expect("the test finishes in time");
}

async fn narrow_run() {
    let (publisher, moved, kept, unused, unpublished) = (b"publisher".as_slice(), b"moved-away".as_slice(), b"kept".as_slice(), b"loses-a-right-unused".as_slice(), b"loses-publish".as_slice());
    let waiting = b"waits-for-its-track".as_slice();
    let sock = socket_path("narrow");
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.grants = vec![rule(&["roomA", "p"], publisher, true, false), rule(&["roomC"], publisher, true, false), rule(&["roomA"], moved, false, true), rule(&["roomA"], kept, false, true), rule(&["roomA"], unused, true, true), rule(&["roomB"], unpublished, true, false), rule(&["roomC"], waiting, false, true)];
        c.control.socket = Some(sock.clone());
        c.limits.max_rendezvous_ms = 30_000;
    })
    .await;
    let relay = server.relay().clone();
    let ta = Tracks::new(ns(&["roomA", "p"]), "cam", 10, 4, 128);
    let tb = Tracks::new(ns(&["roomB", "p"]), "cam", 20, 5, 128);
    let key = SigningKey::from_bytes(&[0xe5; 32]);
    let (p, _ep) = connect(&server, Via::Quic, bearer(publisher)).await;
    let _pp = publish_tracks(&p, &ta, None).await;
    let (q, _eq) = connect(&server, Via::Quic, bearer(unpublished)).await;
    let _pq = publish_tracks(&q, &tb, None).await;
    let (s_moved, _e1) = connect(&server, Via::Quic, bearer(moved)).await;
    let (s_kept, _e2) = connect(&server, Via::Quic, bearer(kept)).await;
    let (s_unused, _e3) = connect(&server, Via::Quic, bearer(unused)).await;
    let v_moved = subscribe_both(&s_moved, &ta, Parameters::new()).await;
    let v_kept = subscribe_both(&s_kept, &ta, Parameters::new()).await;
    let v_unused = subscribe_both(&s_unused, &ta, Parameters::new()).await;
    publish_group(&p, &key, &ta, 0, &payload("n", 0, 300)).await;
    for (c, v) in [(&s_moved, &v_moved), (&s_kept, &v_kept), (&s_unused, &v_unused)] {
        let got = collect_concurrent(c, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [0]).await;
        assert_eq!(groups_of(&got, v.media_alias), vec![0], "delivered before the change");
    }

    // A subscription admitted now, waiting for a track nobody publishes yet, holds nothing.
    let tc = Tracks::new(ns(&["roomC", "p"]), "cam", 30, 6, 128);
    let s_waiting = std::sync::Arc::new(connect(&server, Via::Quic, bearer(waiting)).await);
    let pending = {
        let (c, t) = (s_waiting.clone(), tc.clone());
        tokio::spawn(async move { c.0.subscribe(t.ns.clone(), t.media.clone(), Parameters::new().with(Parameter::RendezvousTimeout(30_000))).await.map(|_| ()) })
    };
    tokio::time::sleep(Duration::from_secs(1)).await;

    // Every token stays named. One moves to another prefix, one loses the right to publish that
    // its session uses, one the right its session does not use, one the prefix its session is
    // still waiting on.
    let next = [rule(&["roomA", "p"], publisher, true, false), rule(&["roomC"], publisher, true, false), rule(&["roomZ"], moved, false, true), rule(&["roomA"], kept, false, true), rule(&["roomA"], unused, false, true), rule(&["roomB"], unpublished, false, true), rule(&["roomZ"], waiting, false, true)];
    let a = ask(&sock, json!({"op": "set", "auth": {"required": true, "grants": next, "authorities": []}, "metered": []})).await;
    assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(2)), "{a}");
    for (c, what) in [(&s_moved, "a subscription its rule no longer covers"), (&q, "a publication whose right went")] {
        let why = closed_with(c, SessionError::UNAUTHORIZED, what).await;
        assert!(why.starts_with("grant_revoked") && why.contains("no longer cover"), "{what}: {why}");
    }
    assert!(until(STEP, || relay.stats().sessions_open == 4).await, "the others stand, the waiting session among them");
    // Its track appears: the session that waited meets the grants in force as it takes its place.
    // On a host so loaded that the request came after the change, it was refused as it arrived.
    if pending.is_finished() {
        eprintln!("SKIPPED: the waiting request reached the relay after the change: {:?}", pending.await);
        s_waiting.0.close(0, "done");
    } else {
        let _pc = publish_tracks(&p, &tc, None).await;
        let why = closed_with(&s_waiting.0, SessionError::UNAUTHORIZED, "a subscription placed after the change").await;
        assert!(why.starts_with("grant_revoked") && why.contains("no longer cover"), "{why}");
        let _ = timeout(STEP, pending).await.expect("the request ends");
    }
    publish_group(&p, &key, &ta, 1, &payload("n", 1, 300)).await;
    for (c, v) in [(&s_kept, &v_kept), (&s_unused, &v_unused)] {
        let got = collect_concurrent(c, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [1]).await;
        assert_eq!(groups_of(&got, v.media_alias), vec![1], "the rest goes on delivering");
    }
    for c in [&p, &s_kept, &s_unused] {
        c.close(0, "done");
    }
    server.shutdown().await;
}

// ── A set states what it replaces ───────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_set_states_what_it_replaces() {
    let _turn = GRANTS_FILE.lock().await;
    timeout(Duration::from_secs(120), set_run()).await.expect("the test finishes in time");
}

async fn set_run() {
    let tok = b"a-token".to_vec();
    let sock = socket_path("set");
    let file = std::env::temp_dir().join(format!("wai-relay-set-{}.toml", std::process::id()));
    std::fs::write(&file, grants_file(&[])).unwrap();
    let server = relay(|c| {
        c.control.socket = Some(sock.clone());
        c.control.grants_path = Some(file.clone());
    })
    .await;
    let relay = server.relay().clone();
    let t = Tracks::new(ns(&["live", "s"]), "cam", 10, 4, 128);
    let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _pp = publish_tracks(&p, &t, None).await;
    let anonymous = || async {
        let (c, e) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let r = c.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.map(|_| ());
        c.close(0, "done");
        drop(e);
        r
    };
    anonymous().await.expect("open, as the configuration has it");
    assert!(until(STEP, || relay.stats().sessions_open == 1).await, "the publisher alone");

    // Stated, it closes the relay; the publisher, with no token, holds what it no longer may.
    let metered = json!({"token_sha256": sha_hex(&tok), "grant": "c0".repeat(32)});
    let rules = [rule(&["live"], &tok, true, true)];
    let a = ask(&sock, json!({"op": "set", "auth": {"required": true, "grants": rules, "authorities": []}, "metered": [metered]})).await;
    assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(1)), "{a}");
    let (p2, _e2) = connect(&server, Via::Quic, bearer(&tok)).await;
    let _pp2 = publish_tracks(&p2, &t, None).await;
    assert!(anonymous().await.is_err(), "required, as the set stated");

    // Left out, each changes nothing: the relay stays closed, and the metered grant stands.
    let a = ask(&sock, json!({"op": "set", "auth": {"grants": rules, "authorities": []}, "metered": []})).await;
    assert_eq!(a["ok"], false, "a set with no required: {a}");
    let a = ask(&sock, json!({"op": "set", "auth": {"required": true, "grants": rules, "authorities": []}})).await;
    assert_eq!(a["ok"], false, "a set with no metered: {a}");
    assert!(relay.auth_config().required && relay.metered_grants().len() == 1);
    assert!(anonymous().await.is_err(), "still required");

    // A reload takes the file's lists, and leaves whether authorization is required as it stands.
    std::fs::write(&file, grants_file(&[(tok.as_slice(), true, true)])).unwrap();
    let a = ask(&sock, json!({"op": "reload"})).await;
    assert_eq!(a["ok"], true, "{a}");
    assert!(relay.auth_config().required, "the configuration's false is not taken back by a reload");
    assert!(anonymous().await.is_err(), "still required after the reload");
    p2.close(0, "done");
    let _ = std::fs::remove_file(&file);
    server.shutdown().await;
}

// ── A metered grant withdrawn ───────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn withdrawing_a_metered_grant_ends_the_sessions_under_it() {
    timeout(Duration::from_secs(120), metered_run()).await.expect("the test finishes in time");
}

async fn metered_run() {
    let (one, two, three) = (b"metered-one".as_slice(), b"metered-two".as_slice(), b"metered-three".as_slice());
    let grant = |n: u8| hex::encode([n; 32]);
    let metered = |t: &[u8], n: u8| MeteredGrant { token_sha256: sha_hex(t), grant: grant(n), joules_ceiling_micro: u64::MAX, funds_ceiling: u64::MAX, currency: String::new() };
    let sock = socket_path("metered");
    let server = relay(|c| {
        c.metering.grants = vec![metered(one, 0xa1), metered(two, 0xa2), metered(three, 0xa3)];
        c.control.socket = Some(sock.clone());
    })
    .await;
    let relay = server.relay().clone();
    let (s1, _e1) = connect(&server, Via::Quic, bearer(one)).await;
    let (s2, _e2) = connect(&server, Via::Quic, bearer(two)).await;
    let (s3, _e3) = connect(&server, Via::Quic, bearer(three)).await;
    assert!(until(STEP, || relay.stats().sessions_open == 3).await);

    // No rule names the token: its metered grant is all that does, and the session ends with it.
    let a = ask(&sock, json!({"op": "withdraw", "token_sha256": sha_hex(one)})).await;
    assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(1)), "{a}");
    let why = closed_with(&s1, SessionError::UNAUTHORIZED, "the session under the withdrawn metered grant").await;
    assert!(why.starts_with("grant_revoked") && why.contains("its SETUP presented"), "{why}");

    // A set that leaves a token out ends its session, and one that maps a token to another grant
    // moves its session there.
    let a = ask(&sock, json!({"op": "set", "auth": {"required": false, "grants": [], "authorities": []}, "metered": [metered(three, 0xb3)]})).await;
    assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(1)), "{a}");
    let why = closed_with(&s2, SessionError::UNAUTHORIZED, "the session of the token the set left out").await;
    assert!(why.starts_with("grant_revoked"), "{why}");
    assert!(until(STEP, || relay.stats().sessions_open == 1).await, "the session whose token stays mapped goes on");
    assert_eq!(relay.open_tallies()[0].1.room_grant.as_ref().map(|g| g.grant.clone()), Some(grant(0xb3)));
    s3.close(0, "done");
    ended(&relay, 3).await;
    let mut named: Vec<String> = relay.summaries().into_iter().filter_map(|s| s.summary.tally.room_grant.map(|g| g.grant)).collect();
    named.sort();
    assert_eq!(named, vec![grant(0xa1), grant(0xa2), grant(0xb3)], "each session's summary names what it ran under");
    server.shutdown().await;
}

// ── An authority's revocation list ──────────────────────────────────────────

fn now_s() -> u64 {
    now_us() / 1_000_000
}

/// A grant laid out as live-rooms §4.1 lays it out, signed by `authority`: `publish` the media
/// bits, a policy that permits the chat kind when `events`, its `viewer`, its ceilings.
#[allow(clippy::too_many_arguments)]
fn room_grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, publish: u8, events: bool, viewer: [u8; 16], joules: u64, funds: u64, currency: &[u8; 3]) -> Vec<u8> {
    let now = now_s();
    let mut g = b"WRG1\x01".to_vec();
    g.extend_from_slice(&room);
    g.extend_from_slice(&principal.verifying_key().to_bytes());
    g.extend_from_slice(&[0; 40]);
    g.extend_from_slice(&(now - 60).to_le_bytes());
    g.extend_from_slice(&(now + 600).to_le_bytes());
    g.extend_from_slice(&[0, publish, 1, 0, 0, 0]);
    g.extend_from_slice(&[0; 16]); // once
    g.extend_from_slice(&viewer);
    g.extend_from_slice(b"WEP1\x01");
    let mut kinds = [0u8; 32];
    if events {
        kinds[0] = 1 << 3;
    }
    g.extend_from_slice(&kinds);
    g.extend_from_slice(&[0; 28]);
    g.extend_from_slice(&joules.to_le_bytes());
    g.extend_from_slice(&funds.to_le_bytes());
    g.extend_from_slice(currency);
    g.extend_from_slice(&[0, 0, 0, 0]);
    let mut m = joule_moq::grant_proof::GRANT_DOMAIN.to_vec();
    m.extend_from_slice(&g);
    g.extend_from_slice(&authority.sign(&m).to_bytes());
    g
}

fn digest_hex(g: &[u8]) -> String {
    hex::encode(joule_moq::grant_proof::grant_digest_and_principal(g).unwrap().0)
}

fn viewer_tag_hex(viewer: [u8; 16]) -> String {
    let mut c = ring::digest::Context::new(&ring::digest::SHA256);
    c.update(b"wai:live-viewer\x01");
    c.update(&viewer);
    hex::encode(c.finish().as_ref())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_authoritys_revoked_grant_is_refused_at_its_proof() {
    timeout(Duration::from_secs(180), revocation_run()).await.expect("the test finishes in time");
}

async fn revocation_run() {
    let authority = SigningKey::from_bytes(&[0x61; 32]);
    let ak = hex::encode(authority.verifying_key().to_bytes());
    let room = [0x7e; 32];
    let r = hex::encode(room);
    let keys: Vec<SigningKey> = (0..5u8).map(|i| SigningKey::from_bytes(&[0x70 + i; 32])).collect();
    let none = [0u8; 16];
    let alice = room_grant(&authority, room, &keys[0], 0b011, true, none, u64::MAX, u64::MAX, &[0; 3]);
    let by_digest = room_grant(&authority, room, &keys[1], 0, false, none, u64::MAX, u64::MAX, &[0; 3]);
    let by_viewer = room_grant(&authority, room, &keys[2], 0, false, [0x5c; 16], u64::MAX, u64::MAX, &[0; 3]);
    let later = room_grant(&authority, room, &keys[3], 0, false, [0x5d; 16], u64::MAX, u64::MAX, &[0; 3]);
    let metered = room_grant(&authority, room, &keys[4], 0, false, [0x5e; 16], 5_000_000, 500, b"EUR");
    let sock = socket_path("revoke");
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
        c.auth.revocations = vec![Revocation { authority: ak.clone(), revoked: vec![digest_hex(&by_digest), viewer_tag_hex([0x5c; 16])] }];
        c.control.socket = Some(sock.clone());
    })
    .await;
    let relay = server.relay().clone();
    let target = format!("//{}{}", server.websocket_addr.unwrap(), relay.config().path);
    let with = |g: &Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g.clone(), k.clone())], target: target.clone(), ..ClientOptions::default() };

    // Revoked by its digest, or by its viewer's tag: refused at SETUP, before its proof.
    for (g, k, what) in [(&by_digest, &keys[1], "revoked by its digest"), (&by_viewer, &keys[2], "revoked by its viewer's tag")] {
        let (c, _) = connect(&server, Via::WebSocket, with(g, k)).await;
        let why = closed_with(&c, SessionError::UNAUTHORIZED, what).await;
        assert!(why.contains("grant_revoked"), "{what}: {why}");
    }

    // Alice publishes; two viewers under grants not revoked subscribe.
    let ta = Tracks::new(ns(&[&r, &hex::encode(keys[0].verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
    let (a, _) = connect(&server, Via::WebSocket, with(&alice, &keys[0])).await;
    let _pa = publish_tracks(&a, &ta, None).await;
    let (v3, _) = connect(&server, Via::WebSocket, with(&later, &keys[3])).await;
    let (v4, _) = connect(&server, Via::WebSocket, with(&metered, &keys[4])).await;
    let s3 = subscribe_both(&v3, &ta, Parameters::new()).await;
    let s4 = subscribe_both(&v4, &ta, Parameters::new()).await;
    publish_group(&a, &keys[0], &ta, 0, &payload("a", 0, 300)).await;
    for (v, s) in [(&v3, &s3), (&v4, &s4)] {
        let got = collect_concurrent(v, STEP, Duration::from_millis(300), |g| groups_of(g, s.media_alias) == [0]).await;
        assert_eq!(groups_of(&got, s.media_alias), vec![0]);
    }

    // The authority revokes the third viewer's grant while its session is open: that session
    // ends, and the room goes on for the others.
    let x = ask(&sock, json!({"op": "revoke", "authority": ak, "revoked": [digest_hex(&later)]})).await;
    assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
    let why = closed_with(&v3, SessionError::UNAUTHORIZED, "a grant revoked while it ran").await;
    assert!(why.contains("grant_revoked") && why.contains("live-rooms §5"), "{why}");
    publish_group(&a, &keys[0], &ta, 1, &payload("a", 1, 300)).await;
    let got = collect_concurrent(&v4, STEP, Duration::from_millis(300), |g| groups_of(g, s4.media_alias) == [1]).await;
    assert_eq!(groups_of(&got, s4.media_alias), vec![1], "the room goes on");
    let (again, _) = connect(&server, Via::WebSocket, with(&later, &keys[3])).await;
    let why = closed_with(&again, SessionError::UNAUTHORIZED, "the revoked grant presented again").await;
    assert!(why.contains("grant_revoked"), "{why}");

    // The metered viewer's ledger entry is its grant's own: its digest and its ceilings.
    v4.close(0, "done");
    a.close(0, "done");
    // Alice's and the two viewers' (the three refused at SETUP have none).
    ended(&relay, 3).await;
    let s = relay.summaries().into_iter().map(|s| s.summary).find(|s| s.tally.room_grant.as_ref().is_some_and(|g| g.grant == digest_hex(&metered))).expect("the viewer's summary names its grant");
    let g = s.tally.room_grant.unwrap();
    assert_eq!((g.joules_ceiling_micro, g.funds_ceiling, g.currency.as_str()), (5_000_000, 500, "EUR"));
    let rec = relay.hop_record(room, now_us() - 600_000_000);
    let e = rec.ledger.iter().find(|e| hex::encode(e.grant) == digest_hex(&metered)).expect("the ledger names the grant");
    assert_eq!((e.joules_ceiling, e.funds_ceiling, &e.currency), (5_000_000, 500, b"EUR"));
    let ea = rec.ledger.iter().find(|e| hex::encode(e.grant) == digest_hex(&alice)).expect("alice's grant");
    assert_eq!((ea.joules_ceiling, ea.funds_ceiling, ea.currency), (u64::MAX, u64::MAX, [0; 3]), "no ceiling: none");
    conformance_reads(&rec.to_bytes(), &rec.ledger_bytes(), &room, 1);
    server.shutdown().await;
}

/// live-rooms §4.3 step 5: a grant whose funds ceiling and currency disagree is `grant_invalid`,
/// whoever signed it, and so never reaches a ledger, whose rule (live-metering §3.6) it would
/// break for every grant of the record.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_grant_whose_funds_and_currency_disagree_is_refused() {
    timeout(Duration::from_secs(120), invalid_run()).await.expect("the test finishes in time");
}

async fn invalid_run() {
    let authority = SigningKey::from_bytes(&[0x63; 32]);
    let ak = hex::encode(authority.verifying_key().to_bytes());
    let room = [0x7d; 32];
    let viewer = SigningKey::from_bytes(&[0x75; 32]);
    for required in [true, false] {
        let server = relay(|c| {
            c.auth.required = required;
            c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
        })
        .await;
        let relay = server.relay().clone();
        let target = format!("//{}{}", server.websocket_addr.unwrap(), relay.config().path);
        let with = |g: &Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g.clone(), k.clone())], target: target.clone(), ..ClientOptions::default() };
        for (funds, currency, what) in [(500, [0u8; 3], "funds with no currency"), (500, *b"eur", "a currency not in capitals"), (500, [0xff, b'U', b'R'], "a currency not text"), (u64::MAX, *b"EUR", "a currency with no funds")] {
            let g = room_grant(&authority, room, &viewer, 0, false, [0; 16], 5_000_000, funds, &currency);
            let (c, _) = connect(&server, Via::WebSocket, with(&g, &viewer)).await;
            let why = closed_with(&c, SessionError::UNAUTHORIZED, what).await;
            assert!(why.contains("grant_invalid"), "{what}: {why}");
        }
        // A grant that keeps the rule is admitted.
        let good = room_grant(&authority, room, &viewer, 0, false, [0; 16], 5_000_000, 500, b"EUR");
        let (c, _) = connect(&server, Via::WebSocket, with(&good, &viewer)).await;
        assert!(until(STEP, || relay.stats().sessions_open == 1).await, "the session stands");
        c.close(0, "done");
        assert!(until(STEP, || relay.stats().sessions_open == 0).await);
        server.shutdown().await;
    }
}

// ── A revocation refuses every use ──────────────────────────────────────────

/// A viewer's grant of `authority` for `room`, with no ceiling.
fn viewer_grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey) -> Vec<u8> {
    room_grant(authority, room, principal, 0, false, [0; 16], u64::MAX, u64::MAX, &[0; 3])
}

/// A session on `server` presenting `grants`, each with its proof.
fn proving(server: &RelayServer, grants: &[(&Vec<u8>, &SigningKey)]) -> ClientOptions {
    let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    ClientOptions { grants: grants.iter().map(|(g, k)| ((*g).clone(), (*k).clone())).collect(), target, ..ClientOptions::default() }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_revocation_ends_its_session_whatever_admitted_it() {
    timeout(Duration::from_secs(240), every_use_run()).await.expect("the test finishes in time");
}

async fn every_use_run() {
    let authority = SigningKey::from_bytes(&[0x65; 32]);
    let ak = hex::encode(authority.verifying_key().to_bytes());
    let room = [0x7b; 32];
    let r = hex::encode(room);
    let (k_gone, k_stands, k_asked) = (SigningKey::from_bytes(&[0x78; 32]), SigningKey::from_bytes(&[0x79; 32]), SigningKey::from_bytes(&[0x7a; 32]));
    let gone = viewer_grant(&authority, room, &k_gone);
    let stands = viewer_grant(&authority, room, &k_stands);
    let asked = viewer_grant(&authority, room, &k_asked);
    let revoke = |g: &[u8]| json!({"op": "revoke", "authority": ak, "revoked": [digest_hex(g)]});

    // No authorization required: the relay reads no proof, so the session proved nothing, and the
    // list is all that stands against the grant.
    let sock = socket_path("use-open");
    let server = relay(|c| {
        c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
        c.control.socket = Some(sock.clone());
    })
    .await;
    let open = server.relay().clone();
    let t = Tracks::new(ns(&[&r, "p"]), "cam", 10, 4, 128);
    let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _pp = publish_tracks(&p, &t, None).await;
    let (a, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
    let (b, _) = connect(&server, Via::WebSocket, proving(&server, &[(&stands, &k_stands)])).await;
    // A session whose requests alone present a grant no rule names: the relay serves it on no
    // grant at all, and the revocation reaches it all the same.
    let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let with_token = Parameters::new().with(Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: asked.clone() }));
    let _vs = subscribe_both_with(&s, &t, with_token.clone(), with_token.clone()).await;
    assert!(until(STEP, || open.stats().sessions_open == 4).await);
    let x = ask(&sock, revoke(&gone)).await;
    assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
    let why = closed_with(&a, SessionError::UNAUTHORIZED, "a session of an open relay that presented the grant").await;
    assert!(why.starts_with("grant_revoked") && why.contains("live-rooms §5"), "{why}");
    assert!(until(STEP, || open.stats().sessions_open == 3).await, "the other grant's session stands");
    let (again, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
    let why = closed_with(&again, SessionError::UNAUTHORIZED, "the revoked grant presented again").await;
    assert!(why.contains("grant_revoked"), "{why}");
    let x = ask(&sock, revoke(&asked)).await;
    assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
    let why = closed_with(&s, SessionError::UNAUTHORIZED, "a session of an open relay whose request presented the grant").await;
    assert!(why.starts_with("grant_revoked") && why.contains("live-rooms §5"), "{why}");
    // A request that presents it now is refused, as a SETUP is. Its session stands: a request
    // without the grant is served, and a later change does not end it for the refused one.
    let (late, _el) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let refused = late.subscribe(t.ns.clone(), t.media.clone(), with_token).await.expect_err("a revoked grant in a request, on an open relay");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()) && refused.contains("grant_revoked"), "{refused}");
    let _vl = subscribe_both(&late, &t, Parameters::new()).await;
    let x = ask(&sock, json!({"op": "grant", "grant": {"prefix": ["elsewhere"], "token_sha256": sha_hex(b"unrelated"), "subscribe": true}})).await;
    assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(0)), "{x}");
    for c in [&p, &b, &late] {
        c.close(0, "done");
    }
    server.shutdown().await;

    // A rule names each grant beside its authority, and goes on naming it after the revocation.
    let sock = socket_path("use-rule");
    let by_proof = |g: &[u8]| Grant { proof: true, ..rule(&[&r], g, false, true) };
    let publisher = b"a-publisher".as_slice();
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
        c.auth.grants = vec![by_proof(&gone), by_proof(&stands), rule(&[&r], &asked, false, true), rule(&[&r], publisher, true, false)];
        c.control.socket = Some(sock.clone());
    })
    .await;
    let relay = server.relay().clone();
    let t = Tracks::new(ns(&[&r, "p"]), "cam", 10, 4, 128);
    let key = SigningKey::from_bytes(&[0xd5; 32]);
    let (p, _ep) = connect(&server, Via::Quic, bearer(publisher)).await;
    let _pp = publish_tracks(&p, &t, None).await;
    let (a, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
    let (b, _) = connect(&server, Via::WebSocket, proving(&server, &[(&stands, &k_stands)])).await;
    let va = subscribe_both(&a, &t, Parameters::new()).await;
    let vb = subscribe_both(&b, &t, Parameters::new()).await;
    // A session that presents the third grant in its requests alone, under a rule that asks for
    // no proof.
    let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let with_token = Parameters::new().with(Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: asked.clone() }));
    let vs = subscribe_both_with(&s, &t, with_token.clone(), with_token.clone()).await;
    publish_group(&p, &key, &t, 0, &payload("u", 0, 300)).await;
    for (c, v) in [(&a, &va), (&b, &vb), (&s, &vs)] {
        let got = collect_concurrent(c, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [0]).await;
        assert_eq!(groups_of(&got, v.media_alias), vec![0], "delivered before the revocation");
    }

    let x = ask(&sock, revoke(&gone)).await;
    assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
    let why = closed_with(&a, SessionError::UNAUTHORIZED, "a session a rule admitted").await;
    assert!(why.starts_with("grant_revoked") && why.contains("live-rooms §5"), "{why}");
    let (again, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
    let why = closed_with(&again, SessionError::UNAUTHORIZED, "the grant presented again under its rule").await;
    assert!(why.contains("grant_revoked"), "{why}");

    // The grant its requests presented: the session ends, and a request presenting it is refused.
    let x = ask(&sock, revoke(&asked)).await;
    assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
    let why = closed_with(&s, SessionError::UNAUTHORIZED, "a session that presented the grant in a request").await;
    assert!(why.starts_with("grant_revoked") && why.contains("live-rooms §5"), "{why}");
    let (late, _el) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let refused = late.subscribe(t.ns.clone(), t.media.clone(), with_token).await.expect_err("a revoked grant in a request");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");

    publish_group(&p, &key, &t, 1, &payload("u", 1, 300)).await;
    let got = collect_concurrent(&b, STEP, Duration::from_millis(300), |g| groups_of(g, vb.media_alias) == [1]).await;
    assert_eq!(groups_of(&got, vb.media_alias), vec![1], "the room goes on");
    late.close(0, "done");
    assert!(until(STEP, || relay.stats().sessions_open == 2).await, "the publisher and the grant that stands");

    // Its authority dropped, the rule still names the grant, and the list still holds it.
    let x = ask(&sock, json!({"op": "drop_authority", "key": ak})).await;
    assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(0)), "{x}");
    let (again, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
    let why = closed_with(&again, SessionError::UNAUTHORIZED, "the grant after its authority went").await;
    assert!(why.contains("grant_revoked"), "{why}");
    for c in [&p, &b] {
        c.close(0, "done");
    }
    server.shutdown().await;
}

/// A session's requests may present records no rule names up to the relay's bound, each counted
/// once; a request past it is refused, and one that presents none is still answered.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_records_a_sessions_requests_present_are_bounded() {
    timeout(Duration::from_secs(120), bound_run()).await.expect("the test finishes in time");
}

async fn bound_run() {
    let unauthorized = RequestErrorCode::UNAUTHORIZED.to_string();
    let server = relay(|_| {}).await;
    let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let record = |n: u8| {
        let mut g = b"WRG1\x01".to_vec();
        g.resize(joule_moq_relay::relay::PRESENTED_BYTES / 4, n);
        g
    };
    // No such track: what the request is answered is all this test reads.
    let answer = async |n: Option<u8>| {
        let params = n.map_or_else(Parameters::new, |n| Parameters::new().with(Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: record(n) })));
        s.subscribe(ns(&["room", "p"]), name("none"), params).await.expect_err("nobody publishes it")
    };
    for n in 0..4 {
        let why = answer(Some(n)).await;
        assert!(!why.contains(&unauthorized), "within the bound: {why}");
    }
    let why = answer(Some(4)).await;
    assert!(why.contains(&unauthorized) && why.contains("past the grants"), "past it: {why}");
    let why = answer(Some(0)).await;
    assert!(!why.contains(&unauthorized), "one it holds takes no more room: {why}");
    let why = answer(None).await;
    assert!(!why.contains(&unauthorized), "a request presenting none: {why}");
    s.close(0, "done");
    server.shutdown().await;
}

// ── An authority added and dropped ──────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_authority_added_is_honoured_and_one_dropped_ends_its_sessions_alone() {
    timeout(Duration::from_secs(180), authorities_run()).await.expect("the test finishes in time");
}

async fn authorities_run() {
    let (first, second, added) = (SigningKey::from_bytes(&[0x66; 32]), SigningKey::from_bytes(&[0x67; 32]), SigningKey::from_bytes(&[0x68; 32]));
    let key_of = |k: &SigningKey| hex::encode(k.verifying_key().to_bytes());
    let (room_1, room_2, room_3) = ([0x71; 32], [0x72; 32], [0x73; 32]);
    let (viewer, performer) = (SigningKey::from_bytes(&[0x7d; 32]), SigningKey::from_bytes(&[0x7e; 32]));
    let (g1, g2, g3) = (viewer_grant(&first, room_1, &viewer), viewer_grant(&second, room_2, &viewer), viewer_grant(&added, room_3, &viewer));
    let g3_publisher = room_grant(&added, room_3, &performer, 0b011, true, [0; 16], u64::MAX, u64::MAX, &[0; 3]);
    let sock = socket_path("authorities");
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: key_of(&first), rooms: vec![] }, Authority { key: key_of(&second), rooms: vec![] }];
        c.control.socket = Some(sock.clone());
    })
    .await;
    let relay = server.relay().clone();
    let (s1, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g1, &viewer)])).await;
    let (s2, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g2, &viewer)])).await;
    assert!(until(STEP, || relay.stats().sessions_open == 2).await);

    // Not honoured yet: its grant gives nothing.
    let t = Tracks::new(ns(&[&hex::encode(room_3), &key_of(&performer), "video"]), "0", 10, 5, 128);
    let (early, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g3, &viewer)])).await;
    let refused = early.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.expect_err("an authority not honoured yet");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
    early.close(0, "done");
    assert!(until(STEP, || relay.stats().sessions_open == 2).await);

    // Added at run time: its grants are read, proved and served.
    let x = ask(&sock, json!({"op": "authority", "authority": {"key": key_of(&added)}})).await;
    assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(0)), "{x}");
    let shown = ask(&sock, json!({"op": "show"})).await;
    assert_eq!(shown["auth"]["authorities"].as_array().map(Vec::len), Some(3), "{shown}");
    let (p, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g3_publisher, &performer)])).await;
    let _pp = publish_tracks(&p, &t, None).await;
    let (s3, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g3, &viewer)])).await;
    let v3 = subscribe_both(&s3, &t, Parameters::new()).await;
    publish_group(&p, &performer, &t, 0, &payload("a", 0, 300)).await;
    let got = collect_concurrent(&s3, STEP, Duration::from_millis(300), |g| groups_of(g, v3.media_alias) == [0]).await;
    assert_eq!(groups_of(&got, v3.media_alias), vec![0], "the authority added at run time is honoured");

    // Dropped: the sessions that proved its grants end, though they hold nothing, and only they.
    let x = ask(&sock, json!({"op": "drop_authority", "key": key_of(&first)})).await;
    assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
    let why = closed_with(&s1, SessionError::UNAUTHORIZED, "the session of the authority dropped").await;
    assert!(why.starts_with("grant_revoked") && why.contains("its SETUP presented"), "{why}");
    assert!(until(STEP, || relay.stats().sessions_open == 3).await, "the other authorities' sessions stand");
    publish_group(&p, &performer, &t, 1, &payload("a", 1, 300)).await;
    let got = collect_concurrent(&s3, STEP, Duration::from_millis(300), |g| groups_of(g, v3.media_alias) == [1]).await;
    assert_eq!(groups_of(&got, v3.media_alias), vec![1], "another authority's room goes on");

    // It is no authority now: dropped again it is not found, and its grant gives nothing.
    let x = ask(&sock, json!({"op": "drop_authority", "key": key_of(&first)})).await;
    assert_eq!(x["ok"], false, "{x}");
    let (late, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g1, &viewer)])).await;
    let room_1_track = Tracks::new(ns(&[&hex::encode(room_1), &key_of(&performer), "video"]), "0", 20, 6, 128);
    let refused = late.subscribe(room_1_track.ns.clone(), room_1_track.media.clone(), Parameters::new()).await.expect_err("an authority dropped");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
    for c in [&s2, &p, &s3, &late] {
        c.close(0, "done");
    }
    server.shutdown().await;
}

// ── A session's grant in a room ─────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_rooms_ledger_names_the_grant_of_that_room() {
    timeout(Duration::from_secs(180), two_rooms_run()).await.expect("the test finishes in time");
}

async fn two_rooms_run() {
    let t0 = now_us() - 1_000_000;
    let authority = SigningKey::from_bytes(&[0x69; 32]);
    let ak = hex::encode(authority.verifying_key().to_bytes());
    let (room_a, room_b, room_c) = ([0x5a; 32], [0x5b; 32], [0x5c; 32]);
    let (viewer, performer, other) = (SigningKey::from_bytes(&[0x6a; 32]), SigningKey::from_bytes(&[0x6b; 32]), SigningKey::from_bytes(&[0x6c; 32]));
    let in_a = room_grant(&authority, room_a, &viewer, 0, false, [0; 16], 1_000_000, u64::MAX, &[0; 3]);
    let in_b = room_grant(&authority, room_b, &viewer, 0, false, [0; 16], 2_000_000, 300, b"EUR");
    let publisher = room_grant(&authority, room_b, &performer, 0b011, true, [0; 16], u64::MAX, u64::MAX, &[0; 3]);
    let room_c_token = b"room-c".as_slice();
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
        c.auth.grants = vec![rule(&[&hex::encode(room_c)], room_c_token, true, true)];
    })
    .await;
    let relay = server.relay().clone();
    let tb = Tracks::new(ns(&[&hex::encode(room_b), &hex::encode(performer.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
    let tc = Tracks::new(ns(&[&hex::encode(room_c), &hex::encode(other.verifying_key().to_bytes()), "video"]), "0", 20, 6, 128);
    let (p, _) = connect(&server, Via::WebSocket, proving(&server, &[(&publisher, &performer)])).await;
    let _pp = publish_tracks(&p, &tb, None).await;
    let (q, _eq) = connect(&server, Via::Quic, bearer(room_c_token)).await;
    let _pq = publish_tracks(&q, &tc, None).await;
    // Grants for rooms A and B, A's first, and a token for room C; it subscribes in B and C.
    let holds = ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: room_c_token.to_vec() }], ..proving(&server, &[(&in_a, &viewer), (&in_b, &viewer)]) };
    let (v, _) = connect(&server, Via::WebSocket, holds).await;
    let sb = subscribe_both(&v, &tb, Parameters::new()).await;
    let sc = subscribe_both(&v, &tc, Parameters::new()).await;
    publish_group(&p, &performer, &tb, 0, &payload("b", 0, 300)).await;
    publish_group(&q, &other, &tc, 0, &payload("c", 0, 300)).await;
    let got = collect_concurrent(&v, STEP, Duration::from_millis(300), |g| groups_of(g, sb.media_alias) == [0] && groups_of(g, sc.media_alias) == [0]).await;
    assert_eq!((groups_of(&got, sb.media_alias), groups_of(&got, sc.media_alias)), (vec![0], vec![0]));
    let forwarded = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum::<u64>();
    assert!(until(STEP, || forwarded(&relay) == 4).await, "both rooms' objects and receipts counted out");

    let grants_of = |rec: &HopRecord| rec.ledger.iter().map(|e| hex::encode(e.grant)).collect::<Vec<_>>();
    let holds_to_its_room = |what: &str, b: &HopRecord, c: &HopRecord| {
        let names = grants_of(b);
        assert!(names.contains(&digest_hex(&in_b)) && names.contains(&digest_hex(&publisher)), "{what}: room B's grants: {names:?}");
        assert!(!names.contains(&digest_hex(&in_a)), "{what}: room A's grant in room B's ledger: {names:?}");
        let e = b.ledger.iter().find(|e| hex::encode(e.grant) == digest_hex(&in_b)).expect("the viewer's grant");
        assert_eq!((e.joules_ceiling, e.funds_ceiling, &e.currency, e.sessions), (2_000_000, 300, b"EUR", 1), "{what}: its ceilings");
        assert!(e.bytes_out > 0 && e.subscriber_us.iter().sum::<u64>() > 0, "{what}: its bytes and its time count against it");
        assert_eq!(grants_of(c), vec!["00".repeat(32)], "{what}: no grant of room C");
        assert_eq!(c.ledger[0].sessions, 2, "{what}: its publisher and the viewer");
    };

    // Over a window of the sessions, still open, and again over the sessions ended.
    let w = relay.flush_and_seal(Duration::from_millis(200)).await;
    let (b, c) = (relay.window_record(room_b, t0, w.index).unwrap(), relay.window_record(room_c, t0, w.index).unwrap());
    holds_to_its_room("a window", &b, &c);
    conformance_reads(&b.to_bytes(), &b.ledger_bytes(), &room_b, 1);
    conformance_reads(&c.to_bytes(), &c.ledger_bytes(), &room_c, 1);
    assert!(relay.window_record(room_a, t0, w.index).unwrap().ledger.is_empty(), "nothing of room A was held");
    for x in [&p, &q, &v] {
        x.close(0, "done");
    }
    ended(&relay, 3).await;
    holds_to_its_room("the sessions ended", &relay.hop_record(room_b, t0), &relay.hop_record(room_c, t0));
    let s = relay.summaries().into_iter().map(|s| s.summary).find(|s| s.tally.room_grants.len() == 2).expect("the viewer's summary names each grant");
    assert_eq!(s.tally.room_grants[&hex::encode(room_a)].grant, digest_hex(&in_a));
    assert_eq!(s.tally.room_grants[&hex::encode(room_b)].grant, digest_hex(&in_b));
    server.shutdown().await;
}

/// A relay that holds no authority and names each grant by a rule with `proof` and a metered
/// grant: a session that proved one grant for each of two rooms runs in each under that room's,
/// the record of each stating its room, and a token that is no record names the rooms its grants
/// do not.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_rooms_ledger_names_the_grant_of_that_room_with_no_authority() {
    timeout(Duration::from_secs(180), rooms_by_rule_run()).await.expect("the test finishes in time");
}

async fn rooms_by_rule_run() {
    let t0 = now_us() - 1_000_000;
    let authority = SigningKey::from_bytes(&[0x6d; 32]);
    let (room_a, room_b, room_c) = ([0x5d; 32], [0x5e; 32], [0x5f; 32]);
    let (rb, rc) = (hex::encode(room_b), hex::encode(room_c));
    let (viewer, performer, other) = (SigningKey::from_bytes(&[0x6e; 32]), SigningKey::from_bytes(&[0x6f; 32]), SigningKey::from_bytes(&[0x70; 32]));
    let in_a = viewer_grant(&authority, room_a, &viewer);
    let in_b = viewer_grant(&authority, room_b, &viewer);
    let (publisher_token, room_c_token) = (b"room-b-publisher".as_slice(), b"room-c".as_slice());
    let of_c = "c7".repeat(32);
    let by_proof = |room: &[u8; 32], g: &[u8]| Grant { proof: true, ..rule(&[&hex::encode(room)], g, false, true) };
    let maps = |t: &[u8], grant: &str, joules: u64| MeteredGrant { token_sha256: sha_hex(t), grant: grant.to_string(), joules_ceiling_micro: joules, funds_ceiling: u64::MAX, currency: String::new() };
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.grants = vec![by_proof(&room_a, &in_a), by_proof(&room_b, &in_b), rule(&[&rb], publisher_token, true, false), rule(&[&rc], room_c_token, true, true)];
        c.metering.grants = vec![maps(&in_a, &digest_hex(&in_a), 1_000_000), maps(&in_b, &digest_hex(&in_b), 2_000_000), maps(room_c_token, &of_c, u64::MAX)];
    })
    .await;
    let relay = server.relay().clone();
    let tb = Tracks::new(ns(&[&rb, &hex::encode(performer.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
    let tc = Tracks::new(ns(&[&rc, &hex::encode(other.verifying_key().to_bytes()), "video"]), "0", 20, 6, 128);
    let (p, _ep) = connect(&server, Via::Quic, bearer(publisher_token)).await;
    let _pp = publish_tracks(&p, &tb, None).await;
    let (q, _eq) = connect(&server, Via::Quic, bearer(room_c_token)).await;
    let _pq = publish_tracks(&q, &tc, None).await;
    let holds = ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: room_c_token.to_vec() }], ..proving(&server, &[(&in_a, &viewer), (&in_b, &viewer)]) };
    let (v, _) = connect(&server, Via::WebSocket, holds).await;
    let sb = subscribe_both(&v, &tb, Parameters::new()).await;
    let sc = subscribe_both(&v, &tc, Parameters::new()).await;
    publish_group(&p, &performer, &tb, 0, &payload("b", 0, 300)).await;
    publish_group(&q, &other, &tc, 0, &payload("c", 0, 300)).await;
    let got = collect_concurrent(&v, STEP, Duration::from_millis(300), |g| groups_of(g, sb.media_alias) == [0] && groups_of(g, sc.media_alias) == [0]).await;
    assert_eq!((groups_of(&got, sb.media_alias), groups_of(&got, sc.media_alias)), (vec![0], vec![0]));
    let forwarded = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum::<u64>();
    assert!(until(STEP, || forwarded(&relay) == 4).await, "both rooms' objects and receipts counted out");

    let grants_of = |rec: &HopRecord| rec.ledger.iter().map(|e| hex::encode(e.grant)).collect::<Vec<_>>();
    let holds_to_its_room = |what: &str, b: &HopRecord, c: &HopRecord| {
        let names = grants_of(b);
        assert!(names.contains(&digest_hex(&in_b)), "{what}: room B's grant: {names:?}");
        assert!(!names.contains(&digest_hex(&in_a)) && !names.contains(&of_c), "{what}: another room's grant in room B's ledger: {names:?}");
        let e = b.ledger.iter().find(|e| hex::encode(e.grant) == digest_hex(&in_b)).expect("the viewer's grant");
        assert_eq!((e.joules_ceiling, e.sessions), (2_000_000, 1), "{what}: its ceiling");
        assert!(e.bytes_out > 0 && e.subscriber_us.iter().sum::<u64>() > 0, "{what}: its bytes and its time count against it");
        assert_eq!(grants_of(c), vec![of_c.clone()], "{what}: the token's grant, in the room no record names");
        assert_eq!(c.ledger[0].sessions, 2, "{what}: its publisher and the viewer");
    };

    // Over a window of the sessions, still open, and again over the sessions ended.
    let w = relay.flush_and_seal(Duration::from_millis(200)).await;
    let (b, c) = (relay.window_record(room_b, t0, w.index).unwrap(), relay.window_record(room_c, t0, w.index).unwrap());
    holds_to_its_room("a window", &b, &c);
    conformance_reads(&b.to_bytes(), &b.ledger_bytes(), &room_b, 1);
    conformance_reads(&c.to_bytes(), &c.ledger_bytes(), &room_c, 1);
    assert!(relay.window_record(room_a, t0, w.index).unwrap().ledger.is_empty(), "nothing of room A was held");
    for x in [&p, &q, &v] {
        x.close(0, "done");
    }
    ended(&relay, 3).await;
    holds_to_its_room("the sessions ended", &relay.hop_record(room_b, t0), &relay.hop_record(room_c, t0));
    let s = relay.summaries().into_iter().map(|s| s.summary).find(|s| s.tally.room_grants.len() == 2).expect("the viewer's summary names each grant");
    assert_eq!(s.tally.room_grants[&hex::encode(room_a)].grant, digest_hex(&in_a));
    assert_eq!(s.tally.room_grants[&hex::encode(room_b)].grant, digest_hex(&in_b));
    assert_eq!(s.tally.room_grant.map(|g| g.grant), Some(of_c.clone()), "the grant of no room");
    server.shutdown().await;
}

// ── A revocation stays ──────────────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_revocation_outlives_a_reload_a_set_and_a_restart() {
    let _turn = GRANTS_FILE.lock().await;
    timeout(Duration::from_secs(180), stays_run()).await.expect("the test finishes in time");
}

async fn stays_run() {
    let authority = SigningKey::from_bytes(&[0x64; 32]);
    let ak = hex::encode(authority.verifying_key().to_bytes());
    let room = [0x7c; 32];
    let (k_gone, k_stands) = (SigningKey::from_bytes(&[0x76; 32]), SigningKey::from_bytes(&[0x77; 32]));
    let gone = room_grant(&authority, room, &k_gone, 0, false, [0; 16], u64::MAX, u64::MAX, &[0; 3]);
    let stands = room_grant(&authority, room, &k_stands, 0, false, [0; 16], u64::MAX, u64::MAX, &[0; 3]);
    let sock = socket_path("stays");
    let dir = PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join(format!("stays-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (grants, kept) = (dir.join("grants.toml"), dir.join("revoked.toml"));
    std::fs::write(&grants, grants_file(&[])).unwrap();
    let config = |c: &mut RelayConfig| {
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
        c.control.socket = Some(sock.clone());
        c.control.grants_path = Some(grants.clone());
        c.control.revocations_path = Some(kept.clone());
    };
    // The revoked grant closed at its SETUP, and the other standing.
    let holds = |server: joule_moq_relay::RelayServer, when: &'static str| {
        let (gone, stands, k_gone, k_stands) = (gone.clone(), stands.clone(), k_gone.clone(), k_stands.clone());
        async move {
            let relay = server.relay().clone();
            let target = format!("//{}{}", server.websocket_addr.unwrap(), relay.config().path);
            let with = |g: &Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g.clone(), k.clone())], target: target.clone(), ..ClientOptions::default() };
            let (c, _) = connect(&server, Via::WebSocket, with(&gone, &k_gone)).await;
            let why = closed_with(&c, SessionError::UNAUTHORIZED, when).await;
            assert!(why.contains("grant_revoked"), "{when}: {why}");
            let (c, _) = connect(&server, Via::WebSocket, with(&stands, &k_stands)).await;
            assert!(until(STEP, || relay.stats().sessions_open == 1).await, "{when}: the other grant stands");
            c.close(0, "done");
            assert!(until(STEP, || relay.stats().sessions_open == 0).await, "{when}");
            server
        }
    };

    let server = relay(config).await;
    let x = ask(&sock, json!({"op": "revoke", "authority": ak, "revoked": [digest_hex(&gone)]})).await;
    assert_eq!(x["ok"], true, "{x}");
    let server = holds(server, "revoked").await;

    // A reload of a file that does not name it, and a set that does not repeat it.
    let x = ask(&sock, json!({"op": "reload"})).await;
    assert_eq!(x["ok"], true, "{x}");
    assert_eq!(server.relay().auth_config().revocations.len(), 1, "the list is in force after a reload");
    let server = holds(server, "after a reload").await;
    let x = ask(&sock, json!({"op": "set", "auth": {"required": true, "grants": [], "authorities": [{"key": ak}]}, "metered": []})).await;
    assert_eq!(x["ok"], true, "{x}");
    let server = holds(server, "after a set that does not repeat it").await;

    // A restart reads the file it kept it in.
    server.shutdown().await;
    let server = relay(config).await;
    let server = holds(server, "after a restart").await;
    server.shutdown().await;
    // A file that does not read stops the relay: it never starts without what it was told to keep.
    std::fs::write(&kept, "[[revocations]]\nauthority = ").unwrap();
    let mut c = RelayConfig::loopback();
    config(&mut c);
    assert!(RelayServer::start(c).await.is_err(), "a kept list that does not read");
    std::fs::remove_dir_all(&dir).unwrap();
}

// ── A flush inside a group ──────────────────────────────────────────────────

/// A Python 3 that runs the conformance verifier, if this host has one. Under CI, and wherever
/// `JOULE_MOQ_RELAY_VERIFY_REQUIRE` is set, one is needed, and a skip is never a pass.
fn python() -> Option<&'static str> {
    let found = ["python3", "python3.14", "python3.13", "python3.12", "python3.11", "python3.10", "python3.9", "python3.8"]
        .into_iter()
        .find(|p| std::process::Command::new(p).args(["-c", "import sys; sys.exit(0 if sys.version_info >= (3, 8) else 1)"]).output().is_ok_and(|o| o.status.success()));
    if found.is_none() {
        for required in ["JOULE_MOQ_RELAY_VERIFY_REQUIRE", "CI"] {
            assert!(std::env::var_os(required).is_none(), "{required} is set, and this host has no Python 3.8+ for live_metering_verify.py");
        }
    }
    found
}

/// Each reading its own files: the tests of this file run at once.
static READINGS: AtomicU64 = AtomicU64::new(0);

/// The record and its ledger read as this crate reads them, with §3.6's rules, and as
/// `live_metering_verify.py` reads them: its layout, class, rules, room and signature, the
/// ledger's entries and root against the record's, and the size bound of its version with
/// `media` media tracks.
fn conformance_reads(record: &[u8], ledger: &[u8], room: &[u8; 32], media: u64) {
    HopRecord::decode(record, ledger).expect("the record keeps live-metering §3.6");
    let Some(py) = python() else {
        eprintln!("SKIPPED: no Python 3.8+ here to run live_metering_verify.py");
        return;
    };
    let dir = PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join("live_control");
    std::fs::create_dir_all(&dir).unwrap();
    let n = READINGS.fetch_add(1, Ordering::Relaxed);
    let (rp, lp) = (dir.join(format!("window-{}-{n}.wmh", std::process::id())), dir.join(format!("window-{}-{n}.wml", std::process::id())));
    std::fs::write(&rp, record).unwrap();
    std::fs::write(&lp, ledger).unwrap();
    let verifiers = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../../wai/live-conformance").canonicalize().expect("wai/live-conformance");
    let script = r#"
import sys
sys.path.insert(0, sys.argv[1])
import live_metering_verify as V
rec, led, room, m = open(sys.argv[2], "rb").read(), open(sys.argv[3], "rb").read(), bytes.fromhex(sys.argv[4]), int(sys.argv[5])
if len(rec) > V.bound_of(rec, m):
    sys.exit("record_too_large")
try:
    x = V.read_hop(rec, room)
    entries = V.read_ledger(led)
except V.Refuse as e:
    sys.exit("refused: %s" % e)
if len(entries) != x["grants"] or V.ledger_root(entries) != x["ledger"]:
    sys.exit("ledger_mismatch")
print("read: version %d, %d entries, %d grants, span %d..%d" % (x["version"], len(x["tracks"]), x["grants"], x["start"], x["end"]))
"#;
    let out = std::process::Command::new(py).arg("-c").arg(script).arg(&verifiers).arg(&rp).arg(&lp).arg(hex::encode(room)).arg(media.to_string()).output().expect("python runs");
    let said = format!("{}{}", String::from_utf8_lossy(&out.stdout), String::from_utf8_lossy(&out.stderr));
    let _ = (std::fs::remove_file(&rp), std::fs::remove_file(&lp));
    assert!(out.status.success(), "live_metering_verify.py refused the record: {said}");
    eprintln!("live_metering_verify.py: {}", said.trim());
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_flush_inside_a_group_cuts_at_its_boundary() {
    timeout(Duration::from_secs(180), flush_run()).await.expect("the test finishes in time");
}

/// Subgroup `sub` of group `g`, objects `first..first + n`, each `len` bytes.
async fn send(c: &Client, alias: u64, g: u64, sub: u64, first: u64, n: u64, last: bool) {
    let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Explicit(sub), publisher_priority: Some(128), properties: false, end_of_group: last, first_object: true };
    let objects: Vec<SubgroupObject> = (first..first + n).map(|o| SubgroupObject::normal(o, payload("v", g * 100 + o, 500))).collect();
    c.send_subgroup(h, &objects).await.expect("a subgroup");
}

/// The objects of `alias` read, counted.
fn objects_of(got: &[Got], alias: u64) -> usize {
    got.iter().filter(|x| x.header.track_alias == alias).map(|x| x.objects.len()).sum()
}

async fn flush_run() {
    let t0 = now_us() - 1_000_000;
    let room = [0x3c; 32];
    let r = hex::encode(room);
    let p = hex::encode([0x4d; 32]);
    let sock = socket_path("flush");
    let server = relay(|c| {
        c.control.socket = Some(sock.clone());
        c.control.flush_grace_ms = 200;
    })
    .await;
    let relay = server.relay().clone();
    let track_ns = ns(&[&r, &p, "video"]);
    let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _held = publisher.publish(track_ns.clone(), name("0"), 10, Parameters::new(), track_props(128, None)).await.expect("PUBLISH");
    let (subscriber, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let (ok, _stx, _srx) = subscriber.subscribe(track_ns.clone(), name("0"), Parameters::new()).await.expect("SUBSCRIBE");
    let alias = ok.track_alias;

    // Group 0's first subgroup, before the cut.
    send(&publisher, 10, 0, 0, 0, 3, false).await;
    let got = collect_concurrent(&subscriber, STEP, Duration::from_millis(200), |g| objects_of(g, alias) == 3).await;
    assert_eq!(objects_of(&got, alias), 3);

    // The cut, inside group 0. Its seal waits for what follows to be read and counted, however
    // long a loaded host takes, as a grace would on a host that keeps up.
    assert_eq!(relay.flush().0, 0);
    // Group 0's second subgroup arrives after the cut; group 1 begins after it.
    send(&publisher, 10, 0, 1, 3, 3, true).await;
    send(&publisher, 10, 1, 0, 0, 3, true).await;
    let got = collect_concurrent(&subscriber, STEP, Duration::from_millis(200), |g| objects_of(g, alias) == 6).await;
    assert_eq!(objects_of(&got, alias), 6, "group 0's second subgroup and group 1 delivered");
    let forwarded = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum::<u64>();
    assert!(until(STEP, || forwarded(&relay) == 9).await, "the nine objects counted out");
    assert_eq!(relay.seal(0), vec![0]);

    // Window 0 holds group 0 whole, in and out, with both sessions open.
    let w0 = relay.window(0).expect("window 0 is sealed");
    let in_of = |w: &joule_moq_relay::relay::Window| w.parts.iter().flat_map(|s| s.tally.publisher.tracks.values()).map(|t| t.objects_received).sum::<u64>();
    let out_of = |w: &joule_moq_relay::relay::Window| w.parts.iter().flat_map(|s| s.tally.subscriber.tracks.values()).map(|t| t.objects_forwarded).sum::<u64>();
    assert_eq!((in_of(&w0), out_of(&w0)), (6, 6), "group 0 counted whole in the window it began in");
    assert_eq!(relay.stats().sessions_open, 2, "nobody left");

    // The record over window 0, signed on the control socket, its sessions still open.
    let a = ask(&sock, json!({"op": "record", "room": r, "t0_unix_us": t0, "window": 0})).await;
    assert_eq!(a["ok"], true, "{a}");
    let (bytes, ledger) = (hex::decode(a["record"].as_str().unwrap()).unwrap(), hex::decode(a["ledger"].as_str().unwrap()).unwrap());
    // Read first as WAI's second implementation reads a record, then as this crate does.
    conformance_reads(&bytes, &ledger, &room, 1);
    let rec = HopRecord::decode(&bytes, &ledger).expect("the record reads");
    rec.verify().expect("signed by the relay");
    assert_eq!(rec.hop, relay.relay_id());
    assert_eq!((rec.start, rec.end), (w0.from_unix_us - t0, w0.to_unix_us - t0), "the window's span on the room clock");
    let (inn, out): (Vec<&joule_moq_relay::metering::Entry>, Vec<&joule_moq_relay::metering::Entry>) = rec.entries.iter().partition(|e| e.dir == 1);
    assert_eq!((inn.len(), out.len(), inn[0].objects, out[0].objects, inn[0].kind), (1, 1, 6, 6, 1));
    assert_eq!(rec.sessions, 1, "the subscriber, open");
    assert!(rec.subscriber_us[0] > 0 && rec.subscriber_us[1..].iter().all(|&x| x == 0));
    assert_eq!(rec.ledger.len(), 1);
    assert_eq!((rec.ledger[0].sessions, rec.ledger[0].subscriber_us), (2, rec.subscriber_us));
    // A window not sealed is refused.
    let a = ask(&sock, json!({"op": "record", "room": r, "t0_unix_us": t0, "window": 7})).await;
    assert_eq!(a["ok"], false, "{a}");

    // Both sessions end; window 1 is closed and sealed; each session's windows sum to it.
    publisher.close(0, "done");
    subscriber.close(0, "done");
    assert!(until(STEP, || relay.summaries().len() == 2).await);
    // The surface's flush closes window 1, after its grace, its sessions ended.
    let flushed = ask(&sock, json!({"op": "flush"})).await;
    assert_eq!((flushed["ok"].as_bool(), flushed["window"].as_u64(), flushed["sessions"].as_u64()), (Some(true), Some(1), Some(2)), "{flushed}");
    let w1 = relay.window(1).expect("window 1");
    assert_eq!((in_of(&w1), out_of(&w1)), (3, 3), "group 1 in the window it began in");
    for whole in relay.summaries().into_iter().map(|s| s.summary) {
        let parts: Vec<&SessionSummary> = w0.parts.iter().chain(w1.parts.iter()).filter(|p| p.session == whole.session).collect();
        assert_eq!(parts.len(), 2, "session {}: one part a window", whole.session);
        let t = &whole.tally;
        let sum = |f: &dyn Fn(&SessionSummary) -> u64| parts.iter().map(|p| f(p)).sum::<u64>();
        assert_eq!(sum(&|p| p.tally.subscriber.objects_forwarded), t.subscriber.objects_forwarded);
        assert_eq!(sum(&|p| p.tally.subscriber.object_bytes_forwarded), t.subscriber.object_bytes_forwarded);
        assert_eq!(sum(&|p| p.tally.subscriber.wire_bytes_sent), t.subscriber.wire_bytes_sent);
        assert_eq!(sum(&|p| p.tally.subscriber.undelivered_at_close), t.subscriber.undelivered_at_close);
        assert_eq!(sum(&|p| p.tally.publisher.objects_received), t.publisher.objects_received);
        assert_eq!(sum(&|p| p.tally.publisher.object_bytes_received), t.publisher.object_bytes_received);
        for (n, x) in &t.subscriber.tracks {
            let of = |f: &dyn Fn(&joule_moq_relay::summary::TrackTally) -> u64| parts.iter().filter_map(|p| p.tally.subscriber.tracks.get(n)).map(f).sum::<u64>();
            assert_eq!((of(&|y| y.objects_forwarded), of(&|y| y.object_bytes_forwarded), of(&|y| y.dropped_delivery_timeout), of(&|y| y.undelivered_at_close)), (x.objects_forwarded, x.object_bytes_forwarded, x.dropped_delivery_timeout, x.undelivered_at_close), "{n}");
        }
        for (n, x) in &t.publisher.tracks {
            let of = |f: &dyn Fn(&joule_moq_relay::summary::PublisherTrackTally) -> u64| parts.iter().filter_map(|p| p.tally.publisher.tracks.get(n)).map(f).sum::<u64>();
            assert_eq!((of(&|y| y.objects_received), of(&|y| y.object_bytes_received)), (x.objects_received, x.object_bytes_received), "{n}");
        }
        let us = |s: &[joule_moq_relay::summary::LayerSpan]| s.iter().map(|x| x.to_unix_us - x.from_unix_us).sum::<u64>();
        assert_eq!(sum(&|p| us(&p.tally.subscriber.layer_spans)), us(&t.subscriber.layer_spans), "time on the room's tracks");
    }
    let total = |f: &dyn Fn(&HopRecord) -> u64| f(&relay.window_record(room, t0, 0).unwrap()) + f(&relay.window_record(room, t0, 1).unwrap());
    let whole = relay.hop_record(room, t0);
    assert_eq!(total(&|x| x.subscriber_us.iter().sum::<u64>()), whole.subscriber_us.iter().sum::<u64>(), "the windows' records sum to the whole's");
    assert_eq!(total(&|x| x.entries.iter().map(|e| e.objects).sum::<u64>()), whole.entries.iter().map(|e| e.objects).sum::<u64>());
    assert!(relay.window_record(room, t0, 0).unwrap().end <= relay.window_record(room, t0, 1).unwrap().start, "the windows' spans do not overlap");
    server.shutdown().await;
}

// ── A first flush ───────────────────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_relay_holds_no_second_copy_of_its_counts_until_its_first_flush() {
    timeout(Duration::from_secs(120), first_flush_run()).await.expect("the test finishes in time");
}

async fn first_flush_run() {
    let t0 = now_us() - 1_000_000;
    let room = [0x3e; 32];
    let server = relay(|_| {}).await;
    let relay = server.relay().clone();
    let track_ns = ns(&[&hex::encode(room), &hex::encode([0x4f; 32]), "video"]);
    let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _held = publisher.publish(track_ns.clone(), name("0"), 10, Parameters::new(), track_props(128, None)).await.expect("PUBLISH");
    let (left, _el) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let (ok_left, _ltx, _lrx) = left.subscribe(track_ns.clone(), name("0"), Parameters::new()).await.expect("SUBSCRIBE");
    let (stays, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let (ok_stays, _stx, _srx) = stays.subscribe(track_ns.clone(), name("0"), Parameters::new()).await.expect("SUBSCRIBE");
    send(&publisher, 10, 0, 0, 0, 3, true).await;
    for (c, alias) in [(&left, ok_left.track_alias), (&stays, ok_stays.track_alias)] {
        let got = collect_concurrent(c, STEP, Duration::from_millis(200), |g| objects_of(g, alias) == 3).await;
        assert_eq!(objects_of(&got, alias), 3);
    }
    let counted_out = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum::<u64>();
    assert!(until(STEP, || counted_out(&relay) == 6).await, "both subscribers' objects counted out");
    left.close(0, "done");
    assert!(until(STEP, || relay.summaries().len() == 1).await);

    // Never flushed: every session's counts are its own whole, held once.
    assert_eq!(relay.ended_parts(), 0, "an ended session's counts are its summary");
    assert!(relay.open_tallies().iter().all(|(_, t)| t.parts.by_window.is_empty()), "an open session's are its tally");
    assert!(relay.window(0).is_none(), "nothing seals by itself");

    // The first flush: window 0 holds what the relay counted before it, ended and open.
    let w0 = relay.flush_and_seal(Duration::from_millis(100)).await;
    assert_eq!((w0.index, w0.parts.len()), (0, 3), "the session ended and the two open");
    let ended = relay.summaries().remove(0).summary;
    let part = w0.parts.iter().find(|p| p.session == ended.session).expect("the ended session's part");
    assert_eq!(part.tally, ended.tally, "its whole, the only window it was in");
    assert_eq!(part.close_code, None, "a part states no close");
    let count = |w: &joule_moq_relay::relay::Window| (w.parts.iter().map(|p| p.tally.publisher.objects_received).sum::<u64>(), w.parts.iter().map(|p| p.tally.subscriber.objects_forwarded).sum::<u64>());
    assert_eq!(count(&w0), (3, 6), "the publisher's three objects, sent to two subscribers");
    let rec = relay.window_record(room, t0, 0).expect("the record over window 0");
    conformance_reads(&rec.to_bytes(), &rec.ledger_bytes(), &room, 1);
    assert_eq!(rec.sessions, 2, "the subscriber ended and the one open");
    assert_eq!(relay.ended_parts(), 0, "sealed");

    // From then on the relay counts by window.
    send(&publisher, 10, 1, 0, 0, 3, true).await;
    let got = collect_concurrent(&stays, STEP, Duration::from_millis(200), |g| objects_of(g, ok_stays.track_alias) == 3).await;
    assert_eq!(objects_of(&got, ok_stays.track_alias), 3);
    assert!(until(STEP, || counted_out(&relay) == 6).await, "the three of the group after, on the three before");
    assert!(relay.open_tallies().iter().any(|(_, t)| !t.parts.by_window.is_empty()), "kept by window now");
    publisher.close(0, "done");
    stays.close(0, "done");
    assert!(until(STEP, || relay.summaries().len() == 3).await);
    assert_eq!(relay.ended_parts(), 2, "the two sessions that ended since, held for the window's seal");
    let w1 = relay.flush_and_seal(Duration::from_millis(100)).await;
    assert_eq!(count(&w1), (3, 3));
    for whole in relay.summaries().into_iter().map(|s| s.summary) {
        let parts: Vec<&SessionSummary> = w0.parts.iter().chain(w1.parts.iter()).filter(|p| p.session == whole.session).collect();
        let sum = |f: &dyn Fn(&SessionSummary) -> u64| parts.iter().map(|p| f(p)).sum::<u64>();
        assert_eq!(sum(&|p| p.tally.subscriber.objects_forwarded), whole.tally.subscriber.objects_forwarded, "session {}", whole.session);
        assert_eq!(sum(&|p| p.tally.publisher.objects_received), whole.tally.publisher.objects_received, "session {}", whole.session);
        let us = |s: &[joule_moq_relay::summary::LayerSpan]| s.iter().map(|x| x.to_unix_us - x.from_unix_us).sum::<u64>();
        assert_eq!(sum(&|p| us(&p.tally.subscriber.layer_spans)), us(&whole.tally.subscriber.layer_spans), "session {}: time on the room's tracks", whole.session);
    }
    server.shutdown().await;
}

// ── An interval ─────────────────────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_relay_given_an_interval_seals_with_nobody_asking() {
    timeout(Duration::from_secs(120), interval_run()).await.expect("the test finishes in time");
}

async fn interval_run() {
    // A ceiling is held at a seal, so a relay told to hold one needs an interval to seal by.
    let mut c = RelayConfig::loopback();
    c.control.hold_joule_ceilings = true;
    let e = RelayServer::start(c).await.err().expect("a relay that would hold no ceiling started");
    assert!(e.contains("flush_interval_ms"), "{e}");

    let room = [0x3f; 32];
    let server = relay(|c| {
        c.control.flush_interval_ms = Some(100);
        c.control.flush_grace_ms = 50;
    })
    .await;
    let relay = server.relay().clone();
    let track_ns = ns(&[&hex::encode(room), &hex::encode([0x50; 32]), "video"]);
    let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _held = publisher.publish(track_ns.clone(), name("0"), 10, Parameters::new(), track_props(128, None)).await.expect("PUBLISH");
    send(&publisher, 10, 0, 0, 0, 3, true).await;
    let sealed = |relay: &joule_moq_relay::Relay| (0..256).filter_map(|w| relay.window(w)).flat_map(|w| w.parts).map(|p| p.tally.publisher.objects_received).sum::<u64>();
    assert!(until(STEP, || sealed(&relay) == 3).await, "the objects stand in a sealed window, and no one flushed");
    assert_eq!(relay.stats().sessions_open, 1, "nobody left");
    publisher.close(0, "done");
    server.shutdown().await;
}

// ── A window with a DVR store ───────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_windows_store_figures_stay_within_its_in_entries() {
    timeout(Duration::from_secs(180), dvr_window_run()).await.expect("the test finishes in time");
}

async fn dvr_window_run() {
    let t0 = now_us() - 1_000_000;
    let room = [0x3d; 32];
    let r = hex::encode(room);
    // In memory where the host has it (`store_dir`): `written` waits for the store, not a disk.
    let dir = store_dir("dvr-window");
    let store = dir.clone();
    let server = relay(move |c| c.dvr = Some(DvrConfig { path: store, window: vec![DvrWindow { prefix: vec!["*".into()], max_age_s: 600, max_bytes: 1 << 30 }], room_keys: vec![], takedown_keys: vec![] })).await;
    let relay = server.relay().clone();
    let track_ns = ns(&[&r, &hex::encode([0x4e; 32]), "video"]);
    let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _held = publisher.publish(track_ns.clone(), name("0"), 10, Parameters::new(), track_props(128, None)).await.expect("PUBLISH");
    let written = |n: u64| {
        let relay = relay.clone();
        async move { assert!(until(STEP, || relay.dvr().unwrap().figures(room, 0, now_us()).objects_stored == n).await, "{n} objects written") }
    };
    let came_in = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.publisher.objects_received).sum::<u64>();
    // The figures of the record over window `w`, read as the verifier reads it: objects in, and
    // objects stored.
    let read = |w: u64| {
        let rec = relay.window_record(room, t0, w).expect("the window's record");
        conformance_reads(&rec.to_bytes(), &rec.ledger_bytes(), &room, 1);
        let inn: u64 = rec.entries.iter().filter(|e| e.dir == 1).map(|e| e.objects).sum();
        let d = rec.dvr.expect("version 2");
        assert_eq!(d.bytes_stored, d.objects_stored * 500, "window {w}: a payload's bytes");
        (inn, d.objects_stored)
    };

    // A group cut by the flush, its rest stored after the cut and before the seal: the window
    // that counted it, in which it began, states it whole.
    send(&publisher, 10, 0, 0, 0, 3, false).await;
    written(3).await;
    relay.flush();
    send(&publisher, 10, 0, 1, 3, 3, true).await;
    written(6).await;
    assert_eq!(relay.seal(0), vec![0]);
    assert_eq!(read(0), (6, 6), "the group whole in the window it began in");
    // Nothing more: the next window states nothing, and owes nothing.
    relay.flush();
    assert_eq!(relay.seal(1), vec![1]);
    assert_eq!(read(1), (0, 0));

    // A group sealed before it is whole, its later subgroup first: nothing of it is stored yet,
    // and the window that counted three of its objects states none.
    send(&publisher, 10, 1, 1, 3, 3, false).await;
    assert!(until(STEP, || came_in(&relay) == 9).await);
    relay.flush();
    assert_eq!(relay.seal(2), vec![2]);
    assert_eq!(read(2), (3, 0), "nothing written yet");
    // Its rest arrives, and the store writes all six: the window that counted three states three.
    send(&publisher, 10, 1, 0, 0, 3, true).await;
    written(12).await;
    relay.flush();
    assert_eq!(relay.seal(3), vec![3]);
    assert_eq!(read(3), (3, 3), "at most what its in entries count");
    // A window with no objects in states none, whatever the store has yet to state.
    relay.flush();
    assert_eq!(relay.seal(4), vec![4]);
    assert_eq!(read(4), (0, 0));
    // The next with objects in goes on stating it, as far as its own in entries go.
    send(&publisher, 10, 2, 0, 0, 3, true).await;
    written(15).await;
    relay.flush();
    assert_eq!(relay.seal(5), vec![5]);
    assert_eq!(read(5), (3, 3));
    // A window whose own group is not whole yet states what earlier windows could not.
    send(&publisher, 10, 3, 1, 3, 3, false).await;
    assert!(until(STEP, || came_in(&relay) == 18).await);
    relay.flush();
    assert_eq!(relay.seal(6), vec![6]);
    assert_eq!(read(6), (3, 3), "what the store wrote and no window could state");

    // A record over the sessions that ended, its publisher still open: no in entries, so none of
    // the store's writes in its span are among them (§3.9).
    let (viewer, _ev) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let (ok, _vtx, _vrx) = viewer.subscribe(track_ns.clone(), name("0"), Parameters::new()).await.expect("SUBSCRIBE");
    send(&publisher, 10, 4, 0, 0, 3, true).await;
    written(18).await;
    let got = collect_concurrent(&viewer, STEP, Duration::from_millis(200), |g| objects_of(g, ok.track_alias) == 3).await;
    assert_eq!(objects_of(&got, ok.track_alias), 3);
    viewer.close(0, "done");
    assert!(until(STEP, || relay.summaries().len() == 1).await);
    let rec = relay.hop_record(room, t0);
    conformance_reads(&rec.to_bytes(), &rec.ledger_bytes(), &room, 1);
    assert_eq!(rec.dvr.expect("version 2").objects_stored, 0, "none of its in entries count them");

    publisher.close(0, "done");
    server.shutdown().await;
    let _ = std::fs::remove_dir_all(&dir);
}

// ── SIGHUP ──────────────────────────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sighup_reads_the_grants_file_again() {
    let _turn = GRANTS_FILE.lock().await;
    timeout(Duration::from_secs(180), sighup_run()).await.expect("the test finishes in time");
}

fn grants_file(rules: &[(&[u8], bool, bool)]) -> String {
    rules.iter().map(|(t, p, s)| format!("[[grants]]\nprefix = [\"live\"]\ntoken_sha256 = \"{}\"\npublish = {p}\nsubscribe = {s}\n\n", sha_hex(t))).collect()
}

async fn sighup_run() {
    let (z, x, y) = (b"publisher-token".as_slice(), b"first-viewer-token".as_slice(), b"second-viewer-token".as_slice());
    let file = std::env::temp_dir().join(format!("wai-relay-grants-{}.toml", std::process::id()));
    std::fs::write(&file, grants_file(&[(z, true, false), (x, false, true)])).unwrap();
    let server = relay(|c| {
        c.auth.required = true;
        c.control.grants_path = Some(file.clone());
    })
    .await;
    let relay = server.relay().clone();
    let t = Tracks::new(ns(&["live", "s"]), "cam", 10, 4, 128);
    let key = SigningKey::from_bytes(&[0xc3; 32]);
    let (p, _ep) = connect(&server, Via::Quic, bearer(z)).await;
    let _pp = publish_tracks(&p, &t, None).await;
    let (vx, _ex) = connect(&server, Via::Quic, bearer(x)).await;
    let _sx = subscribe_both(&vx, &t, Parameters::new()).await;
    let (vy, _ey) = connect(&server, Via::Quic, bearer(y)).await;
    assert!(vy.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.is_err(), "not in the file yet");

    std::fs::write(&file, grants_file(&[(z, true, false), (y, false, true)])).unwrap();
    // SAFETY: kill with this process's own id and a signal the relay handles.
    assert_eq!(unsafe { libc::kill(libc::getpid(), libc::SIGHUP) }, 0);
    assert!(until(STEP, || relay.auth_config().grants.iter().any(|g| g.token_sha256 == sha_hex(y))).await, "the file read again");
    let why = closed_with(&vx, SessionError::UNAUTHORIZED, "the viewer whose token the file dropped").await;
    assert!(why.starts_with("grant_revoked"), "{why}");
    let sy = subscribe_both(&vy, &t, Parameters::new()).await;
    publish_group(&p, &key, &t, 0, &payload("s", 0, 300)).await;
    let got = collect_concurrent(&vy, STEP, Duration::from_millis(300), |g| groups_of(g, sy.media_alias) == [0]).await;
    assert_eq!(groups_of(&got, sy.media_alias), vec![0], "the token the file added is served, by the publisher it kept");
    let (vx2, _ex2) = connect(&server, Via::Quic, bearer(x)).await;
    assert!(vx2.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.is_err(), "the dropped token refused");
    // A file that does not read changes nothing.
    std::fs::write(&file, "[[grants]]\nprefix = 7\n").unwrap();
    assert!(relay.reload_grants().is_err());
    assert!(relay.auth_config().grants.iter().any(|g| g.token_sha256 == sha_hex(y)));
    for c in [&p, &vy, &vx2] {
        c.close(0, "done");
    }
    let _ = std::fs::remove_file(&file);
    server.shutdown().await;
}