//! A window closed from the operator's surface by a relay of its own process, at one cut across
//! the relays that carry a room, and the signed summaries drained (`src/window.rs`,
//! `src/outbox.rs`, `src/control.rs`). Each check is held against a fault planted for it:
//!
//! - **A window's counts by session, bound to its record.** The surface's `window` answers a
//!   sealed window's parts signed by the relay; the room's `record` over the window recomputes
//!   from them, so the operator closes the window from the parts. Planted: a part's count raised
//!   and signed by the relay's own key, a session's part left out and signed so, a part changed
//!   after it was signed, and the next window's parts set against this window's record: each
//!   refused.
//! - **Relays that carry a room close a window at one cut.** A core and two edges, each a viewer
//!   on it; the edges `close` before a group begins and the core after, so the group began
//!   between their instants. Joined, the cut puts it in the closing window at all three, and the
//!   core's and the first edge's statements of the link chain exactly in that window and the
//!   next. Planted: the second edge seals at its own cut, as an edge cutting at its own instant
//!   would, and the chain is caught broken, short in the first window and the contradiction of
//!   live-metering §11 in the next; and a cut below the core's own is refused, nothing sealed.
//! - **An edge rides through a window end.** Its viewer is sent no GOAWAY across two window ends,
//!   stays open, and receives every group once. Planted: the second edge ends its window by a
//!   restart, as edges did before, and the same check catches the GOAWAY its viewer is sent.
//! - **Summaries drained.** An acknowledged summary is freed and never handed out again; an
//!   acknowledgement past what was handed out frees nothing; past `summary.held_max` with no
//!   spill file every summary is held and a new session refused `summaries_unacknowledged`
//!   until the operator acknowledges; with a spill file the summaries go there, come back in
//!   order, and a restart hands over those not acknowledged, and none that were.
//! - **Summaries handed over on a file** (WAI decision D95). A relay configured as the live
//!   room's is, with `summary.out_path` and no operator's surface, under a bound of 2: five
//!   sessions, past the bound, each admitted, each summary on the file whole and verifying, none
//!   held. Planted: the same relay with a surface its operator never acknowledges on refuses the
//!   session past the bound `summaries_unacknowledged`, after telling the operator at half the
//!   bound and past it; and the same relay whose file cannot be written holds the summaries and
//!   refuses past the bound, never dropping one.

#![cfg(unix)]

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq::draft22::{Parameters, SessionError};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{UpstreamConfig, UpstreamTransport};
use joule_moq_relay::metering::HopRecord;
use joule_moq_relay::transport::{quic, TransportError};
use joule_moq_relay::window::{link_chain, Cut, SignedWindow};
use joule_moq_relay::RelayServer;
use serde_json::{json, Value};
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::time::timeout;

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

fn hexkey(k: &SigningKey) -> String {
    hex::encode(k.verifying_key().to_bytes())
}

/// 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-cut-{}-{what}.sock", std::process::id()))
}

/// A directory of this test's own, empty.
fn dir(what: &str) -> PathBuf {
    let d = std::env::temp_dir().join(format!("wai-relay-cut-{}-{what}", std::process::id()));
    let _ = std::fs::remove_dir_all(&d);
    std::fs::create_dir_all(&d).unwrap();
    d
}

/// 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")
}

/// A request the surface must answer `ok`.
async fn ok(path: &Path, req: Value) -> Value {
    let a = ask(path, req.clone()).await;
    assert_eq!(a["ok"], true, "{req} answered {a}");
    a
}

/// The sealed window's statement, of `room`, from the surface.
async fn statement(path: &Path, window: u64, room: &str) -> SignedWindow {
    let a = ok(path, json!({"op": "window", "window": window, "room": room})).await;
    serde_json::from_value(a["statement"].clone()).expect("a signed window")
}

/// The room's record over the sealed window, from the surface.
async fn record(path: &Path, room: &str, t0: u64, window: u64) -> HopRecord {
    let a = ok(path, json!({"op": "record", "room": room, "t0_unix_us": t0, "window": window})).await;
    let h = |k: &str| hex::decode(a[k].as_str().unwrap()).unwrap();
    HopRecord::decode(&h("record"), &h("ledger")).expect("the record reads")
}

/// An upstream entry for `server` over its WebSocket listener, the port it bound itself.
fn upstream_to(server: &RelayServer, prefixes: &[&[&str]]) -> UpstreamConfig {
    UpstreamConfig {
        name: "core".into(),
        url: format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path),
        transport: UpstreamTransport::WebSocket,
        prefixes: prefixes.iter().map(|p| p.iter().map(|f| f.to_string()).collect()).collect(),
        ca_path: None,
        cert_sha256: None,
        native_roots: false,
        token_sha256: None,
        token_path: None,
        relay_key: Some(hex::encode(server.relay().relay_id())),
    }
}

/// Media and receipts objects forwarded by every session of `server` open now.
fn forwarded(server: &RelayServer) -> u64 {
    server.relay().open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum()
}

/// The viewer rode through: it was sent no GOAWAY and its session is open.
fn rode_through(v: &Client) -> Result<(), String> {
    if let Some(g) = v.goaway().borrow().clone() {
        return Err(format!("the viewer was sent GOAWAY (new session URI {:?})", String::from_utf8_lossy(&g.new_session_uri)));
    }
    Ok(())
}

// ── A window's counts by session, bound to its record ──────────────────────

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

async fn parts_run() {
    let t0 = now_us() - 1_000_000;
    let room = [0x61; 32];
    let r = hex::encode(room);
    let pk = SigningKey::from_bytes(&[0x62; 32]);
    let d = dir("parts");
    let seed = [0x63u8; 32];
    std::fs::write(d.join("relay.key"), hex::encode(seed)).unwrap();
    let sock = socket_path("parts");
    let server = relay(|c| {
        c.control.socket = Some(sock.clone());
        c.control.flush_grace_ms = 200;
        c.summary.key_path = Some(d.join("relay.key"));
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
    })
    .await;
    let t = Tracks::new(ns(&[&r, &hexkey(&pk), "video"]), "0", 10, 11, 64);
    let media = joule_moq::draft22::names::render(&t.ns, Some(&t.media));
    let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _published = publish_tracks(&p, &t, None).await;
    let (v, _ev) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let s = subscribe_both(&v, &t, Parameters::new()).await;
    let mut got = Vec::new();
    for (window, groups) in [(0u64, 0..3u64), (1, 3..5)] {
        for g in groups.clone() {
            publish_group(&p, &pk, &t, g, &payload("w", g, 300)).await;
        }
        got.extend(collect(&v, STEP, |x| x.len() as u64 >= 2 * (groups.end - groups.start)).await);
        assert!(until(STEP, || forwarded(&server) == 2 * groups.end).await, "every object counted out");
        let f = ok(&sock, json!({"op": "flush"})).await;
        assert_eq!(f["window"], window);
        assert_eq!(f["cut"][&media], groups.end, "the cut: the group after the last that began here: {f}");
    }
    assert_eq!(verify(&got, &s, &pk).0, 5, "every group delivered");

    // The operator closes window 0 from the surface alone: its parts, and the room's record.
    let w0 = statement(&sock, 0, &r).await;
    let rec0 = record(&sock, &r, t0, 0).await;
    w0.check_record(&rec0, t0).expect("the record recomputes from the window's parts");
    assert_eq!(w0.statement.relay_key, hex::encode(server.relay().relay_id()));
    // What the lifecycle is written from: who published which media, who watched which layer.
    let publisher = w0.statement.parts.iter().find(|x| x.tally.publisher.objects_received > 0).expect("the publisher's part");
    assert_eq!(publisher.tally.publisher.tracks[&media].objects_received, 3, "its media in the window");
    let viewer = w0.statement.parts.iter().find(|x| !x.tally.subscriber.layer_spans.is_empty()).expect("the viewer's part");
    assert_eq!(viewer.tally.subscriber.tracks[&media].objects_forwarded, 3);
    assert!(viewer.tally.subscriber.layer_spans.iter().all(|x| x.class == 0 && x.to_unix_us <= w0.statement.to_unix_us), "its layer and its time in the window");
    assert_eq!(w0.statement.parts.len(), 2, "both open, nobody left");

    // Planted, each refused.
    let key = SigningKey::from_bytes(&seed);
    let mut caught = 0;
    let mut raised = w0.statement.clone();
    let vi = raised.parts.iter().position(|x| !x.tally.subscriber.layer_spans.is_empty()).unwrap();
    raised.parts[vi].tally.subscriber.tracks.get_mut(&media).unwrap().objects_forwarded += 1;
    let e = SignedWindow::sign(raised, &key).check_record(&rec0, t0).expect_err("a count raised, signed by the relay's own key");
    assert!(e.starts_with("parts_mismatch"), "{e}");
    caught += 1;
    let mut fewer = w0.statement.clone();
    fewer.parts.remove(vi);
    let e = SignedWindow::sign(fewer, &key).check_record(&rec0, t0).expect_err("the viewer's part left out, signed so");
    assert!(e.starts_with("parts_mismatch"), "{e}");
    caught += 1;
    let mut changed = w0.clone();
    changed.statement.parts[vi].tally.subscriber.layer_spans[0].to_unix_us -= 1;
    let e = changed.check_record(&rec0, t0).expect_err("a part changed after it was signed");
    assert!(e.starts_with("bad_signature"), "{e}");
    caught += 1;
    let w1 = statement(&sock, 1, &r).await;
    let rec1 = record(&sock, &r, t0, 1).await;
    w1.check_record(&rec1, t0).expect("window 1's record recomputes from its parts");
    let e = w1.check_record(&rec0, t0).expect_err("window 1's parts against window 0's record");
    assert!(e.starts_with("parts_mismatch"), "{e}");
    caught += 1;
    eprintln!("planted faults caught (parts): {caught} of 4");
    assert_eq!(caught, 4);

    // Unsealed, and not kept: refused.
    let a = ask(&sock, json!({"op": "window", "window": 9})).await;
    assert_eq!(a["ok"], false, "{a}");
    p.close(0, "done");
    v.close(0, "done");
    server.shutdown().await;
    let _ = std::fs::remove_dir_all(&d);
}

// ── Relays that carry a room close a window at one cut ─────────────────────

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

async fn cuts_run() {
    let room = [0x71; 32];
    let r = hex::encode(room);
    let pk = SigningKey::from_bytes(&[0x72; 32]);
    let (sc, s1, s2) = (socket_path("core"), socket_path("edge1"), socket_path("edge2"));
    let quiet = |c: &mut joule_moq_relay::RelayConfig| {
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
        c.control.flush_grace_ms = 300;
        c.limits.upstream_reconnect_ms = 100;
        c.limits.goaway_timeout_ms = 1_000;
    };
    let core = relay(|c| {
        quiet(c);
        c.control.socket = Some(sc.clone());
    })
    .await;
    let edge = |sock: PathBuf| {
        let up = upstream_to(&core, &[&[&r]]);
        relay(move |c| {
            quiet(c);
            c.control.socket = Some(sock);
            c.upstream = vec![up];
        })
    };
    let (e1, e2) = (edge(s1.clone()).await, edge(s2.clone()).await);
    for e in [&e1, &e2] {
        assert!(until(STEP, || e.relay().links().iter().any(|l| l.session().is_some())).await, "the edge's link to the core opens");
    }
    let t = Tracks::new(ns(&[&r, &hexkey(&pk), "video"]), "0", 10, 11, 64);
    let media = joule_moq::draft22::names::render(&t.ns, Some(&t.media));
    let (p, _ep) = connect(&core, Via::WebSocket, ClientOptions::default()).await;
    let _published = publish_tracks(&p, &t, None).await;
    let (v1, _) = connect(&e1, Via::WebSocket, ClientOptions::default()).await;
    let (v2, _) = connect(&e2, Via::WebSocket, ClientOptions::default()).await;
    let (sub1, sub2) = (subscribe_both(&v1, &t, Parameters::new()).await, subscribe_both(&v2, &t, Parameters::new()).await);
    let (mut got1, mut got2) = (Vec::new(), Vec::new());
    let sessions = |s: &RelayServer| s.relay().stats().sessions_open;
    // Groups `gs` published, received at both viewers, and counted out at every relay.
    let groups = |gs: std::ops::Range<u64>| {
        let (p, pk, t) = (&p, &pk, &t);
        async move {
            for g in gs {
                publish_group(p, pk, t, g, &payload("c", g, 400)).await;
            }
        }
    };
    let all = 2u64; // a media and a receipts object a group
    let received = async |v: &Client, got: &mut Vec<Got>, n: u64| {
        let more = collect(v, STEP, |x| (got.len() + x.len()) as u64 >= all * n).await;
        got.extend(more);
    };
    let counted = async |n: u64| {
        // At each edge its viewer's; at the core its two links'.
        assert!(until(STEP, || forwarded(&e1) == all * n && forwarded(&e2) == all * n && forwarded(&core) == 2 * all * n).await, "every object counted out at every relay");
    };
    groups(0..4).await;
    received(&v1, &mut got1, 4).await;
    received(&v2, &mut got2, 4).await;
    counted(4).await;
    let before = (sessions(&core), sessions(&e1), sessions(&e2));

    // The edges close first; group 4 begins after their instants and before the core's.
    let c1 = ok(&s1, json!({"op": "close"})).await;
    let c2 = ok(&s2, json!({"op": "close"})).await;
    groups(4..5).await;
    received(&v1, &mut got1, 5).await;
    received(&v2, &mut got2, 5).await;
    let cc = ok(&sc, json!({"op": "close"})).await;
    let cut_of = |a: &Value| serde_json::from_value::<Cut>(a["cut"].clone()).expect("a cut");
    let (own1, own2, own_core) = (cut_of(&c1), cut_of(&c2), cut_of(&cc));
    assert_eq!((own1.of(&media), own2.of(&media), own_core.of(&media)), (Some(4), Some(4), Some(5)), "each relay's own cut: the group after the last that began there");
    let joined = Cut::join([&own_core, &own1, &own2]);
    assert_eq!(joined.of(&media), Some(5), "the join: group 4 in the closing window, everywhere");
    // Group 5 begins after every instant: it waits for the cut at all three.
    groups(5..6).await;
    received(&v1, &mut got1, 6).await;
    received(&v2, &mut got2, 6).await;
    counted(6).await;

    let mut caught = 0;
    // Planted: a cut below the core's own, refused with nothing sealed.
    let early = ask(&sc, json!({"op": "seal", "window": cc["window"], "cut": own1})).await;
    assert_eq!(early["ok"], false, "{early}");
    assert!(early["error"].as_str().unwrap().starts_with("cut_too_early"), "{early}");
    assert!(core.relay().window(cc["window"].as_u64().unwrap()).is_none(), "nothing sealed");
    caught += 1;

    // The core and the first edge seal at the join; the second edge, planted, at its own cut,
    // where an edge cutting at its own instant would.
    let seal = |path: PathBuf, w: Value, cut: Cut| async move { ok(&path, json!({"op": "seal", "window": w, "cut": cut})).await };
    let (a, b, c) = tokio::join!(seal(sc.clone(), cc["window"].clone(), joined.clone()), seal(s1.clone(), c1["window"].clone(), joined.clone()), seal(s2.clone(), c2["window"].clone(), own2.clone()));
    assert_eq!(a["cut_sha256"], b["cut_sha256"], "the core and the first edge sealed at one cut");
    assert_ne!(a["cut_sha256"], c["cut_sha256"]);
    let window = |a: &Value| a["window"].as_u64().unwrap();
    let (wc, w1, w2) = (window(&a), window(&b), window(&c));
    let (stc, st1, st2) = (statement(&sc, wc, &r).await, statement(&s1, w1, &r).await, statement(&s2, w2, &r).await);
    let flows = link_chain(&stc, &st1, true).expect("the core and the first edge chain over their link in the window");
    assert_eq!((flows[&media].out, flows[&media].inn), (5, 5), "groups 0 to 4, out at the core and in at the edge");
    let e = link_chain(&stc, &st2, true).expect_err("the second edge cut one group early");
    assert!(e.starts_with("hop_chain_mismatch"), "{e}");
    caught += 1;

    // The next window, every relay at one cut.
    groups(6..8).await;
    received(&v1, &mut got1, 8).await;
    received(&v2, &mut got2, 8).await;
    counted(8).await;
    let closes = [ok(&s1, json!({"op": "close"})).await, ok(&s2, json!({"op": "close"})).await, ok(&sc, json!({"op": "close"})).await];
    let cuts: Vec<Cut> = closes.iter().map(cut_of).collect();
    let joined = Cut::join(&cuts);
    assert_eq!(joined.of(&media), Some(8));
    let (a, b, c) = tokio::join!(seal(s1.clone(), closes[0]["window"].clone(), joined.clone()), seal(s2.clone(), closes[1]["window"].clone(), joined.clone()), seal(sc.clone(), closes[2]["window"].clone(), joined.clone()));
    let (st1, st2, stc) = (statement(&s1, window(&a), &r).await, statement(&s2, window(&b), &r).await, statement(&sc, window(&c), &r).await);
    let flows = link_chain(&stc, &st1, true).expect("and in the next");
    assert_eq!((flows[&media].out, flows[&media].inn), (3, 3), "groups 5 to 7");
    let e = link_chain(&stc, &st2, false).expect_err("the group the second edge cut early, counted in this window");
    assert!(e.starts_with("hop_chain_mismatch") && e.contains("4 objects in over the link, 3 sent"), "{e}");
    caught += 1;

    // The first edge rode through both window ends: its viewer was sent nothing, is open, and
    // has every group once, verified; no session anywhere ended.
    rode_through(&v1).expect("the first edge's viewer rode through");
    assert_eq!(verify(&got1, &sub1, &pk).0, 8, "every group once, at the first edge");
    assert_eq!(verify(&got2, &sub2, &pk).0, 8, "and at the second");
    assert_eq!((sessions(&core), sessions(&e1), sessions(&e2)), before, "nobody left, at the core or at an edge");
    assert!(!e1.relay().going_away());

    // Planted: the second edge ends its window as edges did, by a restart: its viewer is sent
    // GOAWAY, and the check catches it.
    let stopping = tokio::spawn(e2.shutdown());
    assert!(until(STEP, || v2.goaway().borrow().is_some()).await, "the restarting edge sends GOAWAY");
    let e = rode_through(&v2).expect_err("a GOAWAY at a window's end");
    assert!(e.contains("GOAWAY"), "{e}");
    caught += 1;
    rode_through(&v1).expect("the first edge's viewer still");
    eprintln!("planted faults caught (cuts and ride-through): {caught} of 4");
    assert_eq!(caught, 4);

    for c in [&p, &v1, &v2] {
        c.close(0, "done");
    }
    let _ = stopping.await;
    e1.shutdown().await;
    core.shutdown().await;
    drop(sub1);
    drop(sub2);
}

// ── Summaries drained ───────────────────────────────────────────────────────

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

/// A session opened over the WebSocket binding and closed, waited for until its summary is kept.
async fn one_session(server: &RelayServer, kept: usize) {
    let (c, _) = connect(server, Via::WebSocket, ClientOptions::default()).await;
    c.close(0, "done");
    assert!(until(STEP, || server.relay().summaries_held().0 >= kept).await, "the session's summary kept");
}

/// The sessions of the summaries a drain answered, in order.
fn drained(a: &Value) -> Vec<(u64, u64)> {
    a["summaries"].as_array().unwrap().iter().map(|x| (x["seq"].as_u64().unwrap(), x["summary"]["summary"]["session"].as_u64().unwrap())).collect()
}

async fn drain_run() {
    let mut caught = 0;
    // No spill file: a bound of 2.
    let sock = socket_path("drain");
    let server = relay(|c| {
        c.control.socket = Some(sock.clone());
        c.summary.held_max = 2;
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
    })
    .await;
    for n in 1..=3 {
        one_session(&server, n).await;
    }
    // Past the bound: every summary held, none dropped, and a new session refused.
    let held = server.relay().clone();
    assert_eq!(held.summaries().len(), 3, "the third, past the bound, held");
    assert!(held.refusing_for_summaries());
    // Refused before the relay's SETUP is sent: what is held is the close and its reason.
    let ep = quic::client_endpoint(loopback(), vec![server.identity.chain[0].clone()], vec![joule_moq::draft22::ALPN.to_vec()], None).unwrap();
    let conn: joule_moq_relay::transport::SharedConnection = std::sync::Arc::new(quic::connect(&ep, server.quic_addr.unwrap(), "localhost").await.expect("a QUIC session"));
    let client = Client::connect(conn.clone(), ClientOptions::default()).await;
    match timeout(STEP, conn.closed()).await {
        Ok(TransportError::Closed { code, reason, .. }) => {
            assert_eq!(code, SessionError::INTERNAL_ERROR.0, "{reason}");
            assert!(reason.starts_with(joule_moq_relay::outbox::SUMMARIES_UNACKNOWLEDGED), "{reason}");
        }
        other => panic!("a session past the bound was admitted: {other:?} (connect: {:?})", client.as_ref().map(|_| "Ok")),
    }
    drop(client);
    caught += 1;
    // Planted: an acknowledgement of what was never handed out frees nothing.
    let a = ask(&sock, json!({"op": "ack", "through": 3})).await;
    assert_eq!(a["ok"], false, "{a}");
    assert_eq!(held.summaries_held().0, 3);
    caught += 1;
    let first = ok(&sock, json!({"op": "summaries", "limit": 2})).await;
    let seen = drained(&first);
    assert_eq!(seen.iter().map(|x| x.0).collect::<Vec<_>>(), vec![1, 2]);
    let a = ask(&sock, json!({"op": "ack", "through": 3})).await;
    assert_eq!(a["ok"], false, "3 was not handed out: {a}");
    caught += 1;
    let a = ok(&sock, json!({"op": "ack", "through": 1})).await;
    assert_eq!((a["freed"].as_u64(), a["held"].as_u64()), (Some(1), Some(2)));
    // The acknowledged summary is freed, and never handed out again.
    let rest = drained(&ok(&sock, json!({"op": "summaries"})).await);
    assert_eq!(rest.iter().map(|x| x.0).collect::<Vec<_>>(), vec![2, 3], "1 is gone");
    assert!(!held.summaries().iter().any(|s| s.summary.session == seen[0].1), "freed");
    assert!(!held.refusing_for_summaries(), "back at the bound: sessions admitted");
    one_session(&server, 3).await;
    let a = ok(&sock, json!({"op": "summaries", "after": 3})).await;
    assert_eq!(drained(&a).iter().map(|x| x.0).collect::<Vec<_>>(), vec![4], "the session admitted since");
    let a = ok(&sock, json!({"op": "ack", "through": 4})).await;
    assert_eq!(a["held"].as_u64(), Some(0));
    server.shutdown().await;

    // A spill file: past a bound of 1, the summaries go there and come back in order.
    let d = dir("spill");
    let spill = d.join("summaries.jsonl");
    let sock = socket_path("spill");
    let start = |sp: PathBuf, so: PathBuf| {
        relay(move |c| {
            c.control.socket = Some(so);
            c.summary.held_max = 1;
            c.summary.spill_path = Some(sp);
            c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
        })
    };
    let server = start(spill.clone(), sock.clone()).await;
    for n in 1..=3 {
        one_session(&server, n).await;
    }
    assert_eq!(server.relay().summaries_held(), (3, 2, 0, 0), "one in memory, two spilled");
    assert!(!server.relay().refusing_for_summaries(), "spilled, so every session admitted");
    assert_eq!(std::fs::read_to_string(&spill).unwrap().lines().count(), 2);
    let all = drained(&ok(&sock, json!({"op": "summaries"})).await);
    assert_eq!(all.iter().map(|x| x.0).collect::<Vec<_>>(), vec![1, 2, 3], "read back in order");
    let ids: Vec<u64> = all.iter().map(|x| x.1).collect();
    // Two acknowledged, the one in memory and the first spilled; then the relay restarts.
    ok(&sock, json!({"op": "ack", "through": 2})).await;
    server.shutdown().await;
    let server = start(spill.clone(), sock.clone()).await;
    let back = drained(&ok(&sock, json!({"op": "summaries"})).await);
    assert_eq!(back.iter().map(|x| x.1).collect::<Vec<_>>(), vec![ids[2]], "a restart hands over the one not acknowledged, and none that was");
    ok(&sock, json!({"op": "ack", "through": back[0].0})).await;
    assert_eq!(std::fs::metadata(&spill).unwrap().len(), 0, "every spilled summary acknowledged: the file emptied");
    server.shutdown().await;
    eprintln!("planted faults caught (summaries): {caught} of 3");
    assert_eq!(caught, 3);
    let _ = std::fs::remove_dir_all(&d);
}

// ── Summaries handed over on a file ─────────────────────────────────────────────────────────

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

/// A session opened over the WebSocket binding and closed, waited for until the relay has its
/// summary, kept or handed over.
async fn ended(server: &RelayServer, n: u64) {
    let (c, _) = connect(server, Via::WebSocket, ClientOptions::default()).await;
    c.close(0, "done");
    let r = server.relay().clone();
    assert!(until(STEP, || r.summaries_handed_over().0.unwrap_or(0) + r.summaries_held().0 as u64 >= n).await, "the session's summary handed over or kept");
}

/// A session past the bound: refused before the relay's SETUP, `summaries_unacknowledged`; or
/// admitted. `true` where it was refused so.
async fn refused_for_summaries(server: &RelayServer) -> bool {
    let ep = quic::client_endpoint(loopback(), vec![server.identity.chain[0].clone()], vec![joule_moq::draft22::ALPN.to_vec()], None).unwrap();
    let conn: joule_moq_relay::transport::SharedConnection = std::sync::Arc::new(quic::connect(&ep, server.quic_addr.unwrap(), "localhost").await.expect("a QUIC session"));
    let client = Client::connect(conn.clone(), ClientOptions::default()).await;
    let refused = match &client {
        Err(_) => matches!(timeout(STEP, conn.closed()).await, Ok(TransportError::Closed { code, reason, .. }) if code == SessionError::INTERNAL_ERROR.0 && reason.starts_with(joule_moq_relay::outbox::SUMMARIES_UNACKNOWLEDGED)),
        Ok(c) => {
            c.close(0, "admitted");
            false
        }
    };
    drop(client);
    refused
}

async fn handed_over_run() {
    let mut caught = 0;
    // The live room's relay: an out_path, no surface. A bound of 2, and five sessions.
    let d = dir("handed");
    let out = d.join("sessions.jsonl");
    let o = out.clone();
    let server = relay(move |c| {
        c.summary.out_path = Some(o);
        c.summary.held_max = 2;
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
    })
    .await;
    assert!(server.relay().config().summaries_handed_over());
    for n in 1..=5 {
        ended(&server, n).await;
        assert!(!server.relay().refusing_for_summaries(), "session {n}: none refused");
    }
    assert!(!refused_for_summaries(&server).await, "a sixth session, past the bound, admitted");
    assert!(until(STEP, || server.relay().summaries_handed_over().0 == Some(6)).await);
    assert_eq!(server.relay().summaries_held().0, 0, "none held");
    assert_eq!(server.relay().summaries_handed_over().1, 0, "nothing to tell: nothing held");
    let lines: Vec<String> = std::fs::read_to_string(&out).unwrap().lines().map(str::to_string).collect();
    assert_eq!(lines.len(), 6, "every summary on the file");
    let mut sessions = std::collections::BTreeSet::new();
    for l in &lines {
        let s: joule_moq_relay::SignedSessionSummary = serde_json::from_str(l).expect("a whole summary a line");
        s.verify().expect("signed by the relay");
        sessions.insert(s.summary.session);
    }
    assert_eq!(sessions.len(), 6, "one line for each session");
    server.shutdown().await;

    // Planted: a surface the operator never acknowledges on. Told at half the bound and past
    // it, and the session past the bound refused.
    let sock = socket_path("handed");
    let (o, so) = (d.join("surfaced.jsonl"), sock.clone());
    let server = relay(move |c| {
        c.summary.out_path = Some(o);
        c.control.socket = Some(so);
        c.summary.held_max = 2;
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
    })
    .await;
    assert!(!server.relay().config().summaries_handed_over());
    for n in 1..=3 {
        ended(&server, n).await;
    }
    assert_eq!(server.relay().summaries_handed_over(), (None, 2), "held, and the operator told at half the bound and past it");
    if refused_for_summaries(&server).await {
        caught += 1;
    }
    server.shutdown().await;

    // Planted: a file that cannot be written. Held, and the session past the bound refused.
    let o = d.join("no-such-dir").join("sessions.jsonl");
    let server = relay(move |c| {
        c.summary.out_path = Some(o);
        c.summary.held_max = 2;
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
    })
    .await;
    for n in 1..=3 {
        ended(&server, n).await;
    }
    assert_eq!((server.relay().summaries_held().0, server.relay().summaries_handed_over().0), (3, Some(0)), "none handed over, every one held");
    if refused_for_summaries(&server).await {
        caught += 1;
    }
    server.shutdown().await;
    eprintln!("planted faults caught (summaries handed over): {caught} of 2");
    assert_eq!(caught, 2);
    let _ = std::fs::remove_dir_all(&d);
    let _ = std::fs::remove_file(&sock);
}