taria-ratatui 0.4.0

//! Integration tests for the taria transport side of `TariaLayer`: handshake,
//! snapshot streaming, agent input, per-input acks, and reconnects, all over a
//! real Unix domain socket.
//!
//! Unix only: they drive the Unix transport with `std`'s Unix sockets.
#![cfg(unix)]

use std::io::{self, BufRead, BufReader, Read, Write};
use std::net::Shutdown;
use std::ops::{Deref, DerefMut};
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::UnixStream;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};

use taria::wire::{AppToBridge, BridgeToApp, InputId, InputStatus};
use taria::{AgentInput, Node, PROTOCOL_VERSION, Role, Snapshot};
use taria_ratatui::TariaLayer;

const TIMEOUT: Duration = Duration::from_secs(5);

/// A [`TariaLayer`] bound in its own temp dir. Dropping the guard drops the
/// layer first (which unlinks the socket file) and then removes the dir
/// itself - also when the test panics - so runs leave nothing in /tmp.
struct TestLayer {
    layer: TariaLayer,
    _dir: tempfile::TempDir,
}

impl Deref for TestLayer {
    type Target = TariaLayer;

    fn deref(&self) -> &TariaLayer {
        &self.layer
    }
}

impl DerefMut for TestLayer {
    fn deref_mut(&mut self) -> &mut TariaLayer {
        &mut self.layer
    }
}

/// Bind a layer on a throwaway socket path cleaned up on drop.
fn bind_layer(label: &str) -> TestLayer {
    let dir = tempfile::Builder::new()
        .prefix("taria-it-")
        // The layer vets the socket dir: it must be private (0700).
        .permissions(std::fs::Permissions::from_mode(0o700))
        .tempdir()
        .expect("create test socket dir");
    let layer = TariaLayer::bind_at(label, dir.path().join(format!("{label}.sock"))).unwrap();
    TestLayer { layer, _dir: dir }
}

/// A test stand-in for the bridge: a line-framed client on the layer's socket.
struct Client {
    reader: BufReader<UnixStream>,
    writer: UnixStream,
}

impl Client {
    fn connect(layer: &TariaLayer) -> Client {
        let stream = UnixStream::connect(layer.socket_path()).unwrap();
        stream.set_read_timeout(Some(TIMEOUT)).unwrap();
        let writer = stream.try_clone().unwrap();
        Client {
            reader: BufReader::new(stream),
            writer,
        }
    }

    fn read_message(&mut self) -> AppToBridge {
        let mut line = String::new();
        let n = self.reader.read_line(&mut line).unwrap();
        assert!(n > 0, "server closed the connection unexpectedly");
        serde_json::from_str(&line).unwrap()
    }

    fn write_line(&mut self, line: &str) {
        self.writer.write_all(line.as_bytes()).unwrap();
        self.writer.write_all(b"\n").unwrap();
    }

    fn send(&mut self, msg: &BridgeToApp) {
        let line = serde_json::to_string(msg).unwrap();
        self.write_line(&line);
    }

    /// Send one input under `id`; every ack answering it carries that id.
    fn send_input(&mut self, id: u64, input: AgentInput) {
        self.send(&BridgeToApp::input(InputId::new(id), input));
    }
}

/// Write one input straight to a stream, for tests that hand the read half to
/// another thread and so cannot use [`Client`].
fn write_input(stream: &mut UnixStream, id: u64, input: AgentInput) {
    let mut line = serde_json::to_string(&BridgeToApp::input(InputId::new(id), input)).unwrap();
    line.push('\n');
    stream.write_all(line.as_bytes()).unwrap();
}

fn key(name: &str) -> AgentInput {
    AgentInput::key(name)
}

fn ack(id: u64, status: InputStatus) -> AppToBridge {
    AppToBridge::ack(InputId::new(id), status)
}

/// Poll `try_recv` until an input arrives or the timeout passes.
fn poll_try_recv(layer: &TariaLayer) -> Option<(InputId, AgentInput)> {
    let deadline = Instant::now() + TIMEOUT;
    while Instant::now() < deadline {
        if let Some(pair) = layer.try_recv() {
            return Some(pair);
        }
        thread::sleep(Duration::from_millis(5));
    }
    None
}

/// [`poll_try_recv`], keeping only the input, for a test that does not ack.
fn poll_input(layer: &TariaLayer) -> Option<AgentInput> {
    poll_try_recv(layer).map(|(_, input)| input)
}

fn publish_single_node(layer: &mut TariaLayer, id: &str) {
    let mut rec = layer.frame();
    rec.push(Node::new(id, Role::Text));
    rec.publish();
}

/// The longest dropping a layer may take while a bridge is connected, as its
/// doc states it: one `ACCEPT_POLL` nap (50 ms), one `WRITE_TIMEOUT` (5 s)
/// for the line being written when the shutdown came, and the
/// `SHUTDOWN_FLUSH` (250 ms). The constants are private to the crate, so they
/// are spelled out here.
const DROP_BOUND: Duration = Duration::from_millis(50 + 5_000 + 250);

/// Slack for a loaded machine. Far short of a second `WRITE_TIMEOUT`, which a
/// timeout on each wait for room, rather than on the line, used to add.
const DROP_MARGIN: Duration = Duration::from_millis(1_000);

/// Drop `layer` on another thread and return how long it took, failing at
/// [`DROP_BOUND`] plus [`DROP_MARGIN`] rather than waiting out a regression.
fn time_drop(layer: TariaLayer) -> Duration {
    let (tx, rx) = mpsc::channel();
    let started = Instant::now();
    thread::spawn(move || {
        drop(layer);
        let _ = tx.send(started.elapsed());
    });
    let limit = DROP_BOUND + DROP_MARGIN;
    rx.recv_timeout(limit)
        .unwrap_or_else(|_| panic!("dropping the layer took longer than {limit:?}"))
}

/// A tree whose snapshot line is about 900 KB: far more than the socket
/// buffers between the two ends hold, and short enough of `MAX_LINE_BYTES`
/// that nothing in it is cut.
fn publish_900_kb(layer: &mut TariaLayer) {
    layer.publish([Node::new("draft", Role::TextInput).value("d".repeat(900 * 1024))]);
}

#[test]
fn hello_then_snapshots_stream_to_client() {
    let mut layer = bind_layer("stream");
    let mut client = Client::connect(&layer);

    // First message is always the handshake.
    let hello = client.read_message();
    assert_eq!(
        hello,
        AppToBridge::hello("stream", PROTOCOL_VERSION).stamps(true)
    );

    // A publish after connect streams to the client.
    publish_single_node(&mut layer, "first");
    let AppToBridge::Snapshot(snapshot) = client.read_message() else {
        panic!("expected a snapshot after publish");
    };
    assert_eq!(snapshot.seq, 1);
    assert_eq!(snapshot.root.children[0].id.0, "first");

    // And so does the next one.
    publish_single_node(&mut layer, "second");
    let AppToBridge::Snapshot(snapshot) = client.read_message() else {
        panic!("expected a second snapshot");
    };
    assert_eq!(snapshot.seq, 2);
    assert_eq!(snapshot.root.children[0].id.0, "second");
}

#[test]
fn latest_snapshot_is_replayed_on_connect() {
    let mut layer = bind_layer("replay");
    // Published before any client exists.
    publish_single_node(&mut layer, "pre");

    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));
    let AppToBridge::Snapshot(snapshot) = client.read_message() else {
        panic!("expected the latest snapshot right after hello");
    };
    assert_eq!(snapshot.root.children[0].id.0, "pre");
}

/// The replay above is what made an oversized tree fatal: a bridge drops a
/// connection whose line runs past `MAX_LINE_BYTES`, reconnects, and is
/// replayed the same latest snapshot, for as long as the app holds the tree.
/// So the line a fresh client is sent has to be one it can read, and the
/// connection has to go on working after it.
#[test]
fn an_oversized_tree_is_replayed_as_a_line_the_bridge_can_read() {
    use taria::wire::MAX_LINE_BYTES;

    let mut layer = bind_layer("oversized");
    let draft = "d".repeat(MAX_LINE_BYTES + 1);
    let frame = || {
        [Node::new("draft", Role::TextInput)
            .value(draft.clone())
            .focused(true)]
    };
    layer.publish(frame());

    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));
    let mut line = String::new();
    client.reader.read_line(&mut line).unwrap();
    let content = line
        .strip_suffix('\n')
        .expect("the snapshot is one whole line");
    assert!(
        content.len() <= MAX_LINE_BYTES,
        "{} bytes, over the {MAX_LINE_BYTES} a bridge reads",
        content.len()
    );
    let AppToBridge::Snapshot(snapshot) = serde_json::from_str(content).unwrap() else {
        panic!("expected the latest snapshot right after hello");
    };
    let value = snapshot.root.children[0].value.as_deref().unwrap();
    assert!(value.ends_with('\u{2026}'), "the cut is marked");
    assert!(draft.starts_with(value.trim_end_matches('\u{2026}')));
    assert_eq!(layer.stats().oversized_snapshots, 1);

    // Still a working connection: an input goes in and its ack comes back.
    client.send_input(1, key("q"));
    assert_eq!(poll_input(&layer), Some(key("q")));
    assert_eq!(client.read_message(), ack(1, InputStatus::Delivered));

    // Redrawn unchanged, the tree is not a new snapshot. The first redraw
    // after the dequeue sends the one already published again, stamped with
    // the input and under the same seq, without cutting it again; the rest
    // send nothing, so the next line after it is the next change, under the
    // very next seq.
    for _ in 0..3 {
        layer.publish(frame());
    }
    let AppToBridge::Snapshot(restamped) = client.read_message() else {
        panic!("expected the tree again, stamped");
    };
    assert_eq!(restamped.seq, 1, "a stamp moving is not a new snapshot");
    assert_eq!(restamped.after_input, Some(InputId::new(1)));
    assert_eq!(
        restamped.root, snapshot.root,
        "the tree is the one already cut"
    );
    assert_eq!(
        layer.stats().oversized_snapshots,
        1,
        "the resend is not cut again"
    );
    layer.publish([Node::new("draft", Role::TextInput)
        .value(format!("x{draft}"))
        .focused(true)]);
    let AppToBridge::Snapshot(snapshot) = client.read_message() else {
        panic!("expected the changed tree");
    };
    assert_eq!(snapshot.seq, 2, "the redraws published nothing");
    assert_eq!(snapshot.after_input, Some(InputId::new(1)));
    let value = snapshot.root.children[0].value.as_deref().unwrap();
    assert!(value.starts_with('x') && value.ends_with('\u{2026}'));
    assert_eq!(layer.stats().oversized_snapshots, 2);
}

#[test]
fn inputs_reach_the_app_acked_delivered_and_malformed_lines_are_ignored() {
    let layer = bind_layer("input");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    assert_eq!(layer.try_recv(), None);

    // Garbage must be skipped without killing the connection.
    client.write_line("not json at all");
    client.write_line(r#"{"type":"unknown"}"#);

    let sent = key("q");
    client.send_input(1, sent.clone());
    assert_eq!(poll_input(&layer), Some(sent));
    // Nothing was written before the app dequeued it: delivered means the
    // event loop took it, so the ack is the first thing the client sees.
    assert_eq!(client.read_message(), ack(1, InputStatus::Delivered));

    let act = AgentInput::act(taria::NodeId("btn".into()), taria::Action::Activate, None);
    client.send_input(2, act.clone());
    assert_eq!(layer.recv_timeout(TIMEOUT), Some((InputId::new(2), act)));
    assert_eq!(client.read_message(), ack(2, InputStatus::Delivered));
}

/// An input whose `kind` this build cannot read is answered here rather than
/// handed to the app.
///
/// It parses as [`AgentInput::Unknown`], which keeps nothing but the fact
/// that the kind was unreadable; the id survives on the message around it,
/// which is what makes an ack possible at all. Before the fallback existed
/// the whole line failed, the reader skipped it, and the agent waited out its
/// timeout having learned nothing. `Ignored` is the truth, and the app never
/// sees a value it could not have acted on.
#[test]
fn an_input_kind_this_build_cannot_read_is_acked_ignored_without_reaching_the_app() {
    let layer = bind_layer("unknowninput");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    client.write_line(r#"{"type":"input","id":41,"input":{"kind":"paste","text":"hi"}}"#);
    assert_eq!(client.read_message(), ack(41, InputStatus::Ignored));
    assert_eq!(
        poll_input(&layer),
        None,
        "an unreadable input must not be queued for the app"
    );
    assert_eq!(layer.stats().unknown_inputs, 1);
    assert_eq!(
        layer.stats().dropped_inputs,
        0,
        "it was answered, not dropped for want of queue space"
    );

    // The connection is unharmed: a readable input right after it still works.
    let sent = key("q");
    client.send_input(42, sent.clone());
    assert_eq!(poll_input(&layer), Some(sent));
    assert_eq!(client.read_message(), ack(42, InputStatus::Delivered));
    assert_eq!(layer.stats().unknown_inputs, 1);
}

/// A timeout the clock cannot turn into a deadline must not abort the app:
/// `Instant::now() + Duration::MAX` panics, and a panic here takes down the
/// app that embedded the layer. Such a timeout asks to wait for as long as
/// the queue can deliver an input, so that is what it gets.
#[test]
fn recv_timeout_survives_a_timeout_no_clock_can_hold() {
    let layer = bind_layer("hugetimeout");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    // Queued before the wait starts, so the wait ends on the input rather
    // than on a deadline that does not exist.
    client.send_input(1, key("j"));
    assert_eq!(
        layer.recv_timeout(Duration::MAX),
        Some((InputId::new(1), key("j")))
    );
    assert_eq!(client.read_message(), ack(1, InputStatus::Delivered));
}

#[test]
fn ack_can_be_refined_to_ignored() {
    let layer = bind_layer("ackignored");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    client.send_input(9, key("q"));
    let (id, input) = poll_try_recv(&layer).expect("input should reach the app");
    assert_eq!((id, input), (InputId::new(9), key("q")));
    assert_eq!(client.read_message(), ack(9, InputStatus::Delivered));

    // The app looked at it and deliberately did nothing. Last ack wins, so
    // an agent waiting on an effect can stop waiting.
    layer.ack(id, InputStatus::Ignored);
    assert_eq!(client.read_message(), ack(9, InputStatus::Ignored));
}

/// An ack belongs to the bridge whose input it answers.
///
/// Ids are unique only within a bridge process, and the next bridge is a new
/// process whose ids start over, so an id the app kept across a disconnect can
/// already name a live input of that bridge's. Sending the late ack there
/// resolves a waiter that never saw the input, with a verdict from a session
/// that is gone. This is not the accepted "a late `Ignored` ack can be lost":
/// there the answer goes nowhere, here it goes to the wrong peer.
#[test]
fn a_late_ack_never_answers_the_next_bridge() {
    let layer = bind_layer("lateack");
    let mut first = Client::connect(&layer);
    assert!(matches!(first.read_message(), AppToBridge::Hello { .. }));

    first.send_input(5, key("q"));
    let (id, _) = poll_try_recv(&layer).expect("the input reaches the app");
    assert_eq!(id, InputId::new(5));
    assert_eq!(first.read_message(), ack(5, InputStatus::Delivered));

    // The bridge dies with the app still holding id 5, then a new one
    // connects. Its handshake is proof the layer finished with the first
    // connection, because only one is served at a time.
    drop(first);
    let mut second = Client::connect(&layer);
    assert!(matches!(second.read_message(), AppToBridge::Hello { .. }));

    // The app answers the id it kept, a frame late.
    layer.ack(InputId::new(5), InputStatus::Ignored);

    second
        .reader
        .get_ref()
        .set_read_timeout(Some(Duration::from_millis(250)))
        .unwrap();
    let mut line = String::new();
    let err = second
        .reader
        .read_line(&mut line)
        .expect_err("the new bridge must hear nothing about an input it never sent");
    assert!(
        matches!(
            err.kind(),
            io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
        ),
        "expected the read to time out, got {err:?} with {line:?}"
    );

    // And its own id 5 still works, which is what the guard must not break.
    second
        .reader
        .get_ref()
        .set_read_timeout(Some(TIMEOUT))
        .unwrap();
    second.send_input(5, key("j"));
    let (id, input) = poll_try_recv(&layer).expect("the new bridge's input");
    assert_eq!((id, input), (InputId::new(5), key("j")));
    assert_eq!(second.read_message(), ack(5, InputStatus::Delivered));
    layer.ack(InputId::new(5), InputStatus::Ignored);
    assert_eq!(second.read_message(), ack(5, InputStatus::Ignored));
}

#[test]
fn ack_precedes_the_snapshot_published_after_it() {
    let mut layer = bind_layer("ackorder");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    client.send_input(3, key("j"));
    assert_eq!(poll_input(&layer), Some(key("j")));
    // The app reacts to the input by publishing a new tree.
    publish_single_node(&mut layer, "moved");

    // Order matters: an agent seeing the snapshot first cannot tell whether
    // it already reflects the input.
    assert_eq!(client.read_message(), ack(3, InputStatus::Delivered));
    let AppToBridge::Snapshot(snapshot) = client.read_message() else {
        panic!("expected the snapshot after the ack");
    };
    assert_eq!(snapshot.root.children[0].id.0, "moved");
}

/// The loop an app writes when it needs each input's id for something other
/// than an immediate verdict: every queued input, in order, each with its id
/// and its `Delivered` ack.
#[test]
fn a_try_recv_loop_hands_over_every_queued_input_and_acks_each() {
    let layer = bind_layer("recvloop");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    for id in 0..3 {
        client.send_input(id, key(&format!("k{id}")));
    }

    // The inputs cross a socket, so loop until all three have landed.
    let mut got = Vec::new();
    let deadline = Instant::now() + TIMEOUT;
    while got.len() < 3 && Instant::now() < deadline {
        while let Some((id, input)) = layer.try_recv() {
            got.push((id.get(), input));
        }
        thread::sleep(Duration::from_millis(5));
    }
    assert_eq!(
        got,
        vec![(0, key("k0")), (1, key("k1")), (2, key("k2"))],
        "the loop must hand over every queued input, in order"
    );

    for id in 0..3 {
        assert_eq!(client.read_message(), ack(id, InputStatus::Delivered));
    }
}

/// `drain` sends what the handler returns only where it says something the
/// dequeue did not: an `Ignored` refines the `Delivered` ack, and a
/// `Delivered` sends no second copy of it.
#[test]
fn drain_sends_the_refinements_and_nothing_else() {
    let mut layer = bind_layer("drain");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    for id in 0..3 {
        client.send_input(id, key(&format!("k{id}")));
    }

    // The handler never sees an id: it answers from the input alone, which is
    // the whole point of the method.
    let mut got = Vec::new();
    let deadline = Instant::now() + TIMEOUT;
    while got.len() < 3 && Instant::now() < deadline {
        layer.drain(|input| {
            let status = if input == key("k1") {
                InputStatus::Ignored
            } else {
                InputStatus::Delivered
            };
            got.push(input);
            status
        });
        thread::sleep(Duration::from_millis(5));
    }
    assert_eq!(got, vec![key("k0"), key("k1"), key("k2")]);

    // Acks are flushed ahead of a snapshot published after them, so any
    // extra ack the drain queued would arrive before this tree does.
    publish_single_node(&mut layer, "after");

    assert_eq!(client.read_message(), ack(0, InputStatus::Delivered));
    assert_eq!(client.read_message(), ack(1, InputStatus::Delivered));
    assert_eq!(
        client.read_message(),
        ack(1, InputStatus::Ignored),
        "the refinement follows its own input's dequeue ack, before the next input's"
    );
    assert_eq!(client.read_message(), ack(2, InputStatus::Delivered));
    let AppToBridge::Snapshot(snapshot) = client.read_message() else {
        panic!("a returned Delivered must not send a second ack ahead of the snapshot");
    };
    assert_eq!(snapshot.root.children[0].id.0, "after");
}

#[test]
fn client_can_reconnect_after_disconnect() {
    let mut layer = bind_layer("reconnect");
    publish_single_node(&mut layer, "n");

    let mut first = Client::connect(&layer);
    assert!(matches!(first.read_message(), AppToBridge::Hello { .. }));
    assert!(matches!(first.read_message(), AppToBridge::Snapshot(_)));
    drop(first);

    // The listener must return to accepting after losing its client.
    let mut second = Client::connect(&layer);
    assert!(matches!(second.read_message(), AppToBridge::Hello { .. }));
    assert!(matches!(second.read_message(), AppToBridge::Snapshot(_)));

    // The fresh connection is fully functional in both directions.
    second.send_input(1, key("esc"));
    assert_eq!(
        layer.recv_timeout(TIMEOUT),
        Some((InputId::new(1), key("esc")))
    );
    assert_eq!(second.read_message(), ack(1, InputStatus::Delivered));
}

#[test]
fn oversized_line_disconnects_the_client() {
    let layer = bind_layer("linecap");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    // Stream a few MiB with no newline; the layer must cut the connection
    // (capped at 1 MiB per line) rather than buffer it all.
    client.writer.set_write_timeout(Some(TIMEOUT)).unwrap();
    let chunk = [b'a'; 64 * 1024];
    let mut disconnected_while_writing = false;
    for _ in 0..48 {
        // 48 * 64 KiB = 3 MiB
        if client.writer.write_all(&chunk).is_err() {
            disconnected_while_writing = true;
            break;
        }
    }
    if !disconnected_while_writing {
        // The server may still be draining; its close must reach us as EOF
        // (or a reset), never as a timeout with the connection still open.
        let mut line = String::new();
        match client.reader.read_line(&mut line) {
            Ok(0) => {}
            Ok(n) => panic!("expected disconnect, read {n} bytes"),
            Err(err)
                if err.kind() == io::ErrorKind::WouldBlock
                    || err.kind() == io::ErrorKind::TimedOut =>
            {
                panic!("server kept the connection open after an oversized line")
            }
            Err(_) => {} // connection reset: also a disconnect
        }
    }

    // The listener recovers: a fresh client is served again.
    let mut second = Client::connect(&layer);
    assert!(matches!(second.read_message(), AppToBridge::Hello { .. }));
}

#[test]
fn input_flood_is_bounded_acked_dropped_and_never_blocks_the_socket_thread() {
    const FLOOD: u64 = 2000;
    const QUEUE: u64 = 256;

    let mut layer = bind_layer("inputflood");
    assert_eq!(layer.stats().dropped_inputs, 0, "no drops before any flood");

    // Read on a second thread for the whole flood: every dropped input is
    // acked, and a client that only wrote would let those acks back up until
    // the layer's write timeout killed the connection.
    let stream = UnixStream::connect(layer.socket_path()).unwrap();
    stream.set_read_timeout(Some(TIMEOUT)).unwrap();
    let mut writer = stream.try_clone().unwrap();
    // The flood starts once the session has: input written before the hello
    // is discarded unread, which would leave this test counting a race
    // instead of the queue.
    let mut reader = BufReader::new(stream);
    let mut hello = String::new();
    reader.read_line(&mut hello).unwrap();
    assert!(
        matches!(
            serde_json::from_str::<AppToBridge>(&hello).unwrap(),
            AppToBridge::Hello { .. }
        ),
        "the handshake is always first"
    );
    let (alive_tx, alive_rx) = mpsc::channel();
    let collector = thread::spawn(move || {
        let mut messages = Vec::new();
        let mut line = String::new();
        loop {
            line.clear();
            match reader.read_line(&mut line) {
                Ok(0) => break, // the layer closed the connection
                Ok(_) => {
                    let msg = serde_json::from_str::<AppToBridge>(&line).unwrap();
                    if matches!(
                        &msg,
                        AppToBridge::Snapshot(snapshot) if snapshot.root.children[0].id.0 == "alive"
                    ) {
                        // Nobody is listening once the wait below gave up.
                        let _ = alive_tx.send(());
                    }
                    messages.push(msg);
                }
                Err(err) => panic!("client read failed: {err}"),
            }
        }
        messages
    });

    // Far more inputs than the queue holds, with the app not draining.
    for id in 0..FLOOD {
        write_input(&mut writer, id, key(&format!("k{id}")));
    }

    // The socket thread must stay live mid-flood: a publish still streams.
    // Waited for before the half-close below, not after it: the reader's EOF
    // ends the connection at once, and the writer sends nothing it has not
    // sent by then, so a snapshot still behind a backlog of drop acks would
    // be lost by design rather than by a blocked thread.
    publish_single_node(&mut layer, "alive");
    assert!(
        alive_rx.recv_timeout(TIMEOUT).is_ok(),
        "a publish mid-flood must still reach the client"
    );

    // Half-close the write side: the reader consumes the whole flood, sees
    // EOF, and tears the connection down, which the collector observes as
    // EOF. This is the barrier proving the flood was fully processed without
    // blocking.
    writer.shutdown(Shutdown::Write).unwrap();
    let messages = collector.join().expect("collector thread");

    // Only the queue capacity was retained; the overflow was dropped instead
    // of buffered. None of it reaches the app either: the connection that
    // sent it is gone, so every queued input is stale by now.
    let mut received = 0;
    while layer.try_recv().is_some() {
        received += 1;
    }
    assert_eq!(
        received, 0,
        "inputs queued on a dead connection must never be handed to the app"
    );
    assert_eq!(
        layer.stats().stale_inputs,
        QUEUE,
        "the queue held {QUEUE} inputs, and each discard must be counted once"
    );
    assert_eq!(
        layer.stats().dropped_inputs,
        FLOOD - QUEUE,
        "each dropped input must be counted exactly once"
    );

    // The drops reach the agent, which used to see nothing at all. Not
    // necessarily all of them: the connection ends the moment the reader
    // hits EOF, and acks still queued then die with it.
    let dropped: Vec<InputId> = messages
        .iter()
        .filter_map(|msg| match msg {
            AppToBridge::Ack {
                id,
                status: InputStatus::Dropped,
                ..
            } => Some(*id),
            _ => None,
        })
        .collect();
    assert!(!dropped.is_empty(), "dropped inputs must be acked dropped");
    assert!(
        dropped.len() as u64 <= FLOOD - QUEUE,
        "more drops acked ({}) than dropped",
        dropped.len()
    );
    assert!(
        dropped.iter().all(|id| id.get() < FLOOD),
        "a dropped ack must name an input the client actually sent"
    );
    // Nothing was acked delivered: the app drained only after the flood, by
    // which point the connection was gone.
    let other_acks = messages
        .iter()
        .filter(
            |msg| matches!(msg, AppToBridge::Ack { status, .. } if *status != InputStatus::Dropped),
        )
        .count();
    assert_eq!(other_acks, 0, "the only acks during a flood are drops");

    // And the layer still serves fresh clients and inputs afterwards.
    let mut second = Client::connect(&layer);
    assert!(matches!(second.read_message(), AppToBridge::Hello { .. }));
    assert!(matches!(second.read_message(), AppToBridge::Snapshot(_)));
    second.send_input(0, key("after"));
    assert_eq!(
        layer.recv_timeout(TIMEOUT),
        Some((InputId::new(0), key("after")))
    );
    // A stale ack from the dead connection must not surface here: ids start
    // over with each client.
    assert_eq!(second.read_message(), ack(0, InputStatus::Delivered));

    // Accepted inputs never bump either counter.
    assert_eq!(layer.stats().dropped_inputs, FLOOD - QUEUE);
    assert_eq!(layer.stats().stale_inputs, QUEUE);
}

/// The mirror image of the bridge dropping inputs it queued while no app was
/// connected: an input that arrived on a connection which then died targets a
/// bridge session that no longer exists. Applying it would let, say, a `key q`
/// sent just before the bridge restarted quit the app on the next frame, with
/// nobody left to be told.
#[test]
fn inputs_from_a_dead_connection_are_discarded_and_the_next_ones_are_not() {
    let layer = bind_layer("staleinput");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    client.send_input(1, key("q"));
    // Half-close: the reader consumes the input, then sees EOF and tears the
    // connection down. Reading our side to EOF is the barrier proving the
    // layer is done with this connection, input included.
    client.writer.shutdown(Shutdown::Write).unwrap();
    let mut line = String::new();
    loop {
        line.clear();
        match client.reader.read_line(&mut line) {
            Ok(0) => break,
            Ok(_) => {}
            Err(err) => panic!("client read failed: {err}"),
        }
    }

    // Discarding must not cut a wait short: an app pacing its event loop on
    // `recv_timeout` would otherwise spin through the rest of its budget the
    // moment a bridge disconnects.
    const WAIT: Duration = Duration::from_millis(150);
    let started = Instant::now();
    assert_eq!(
        layer.recv_timeout(WAIT),
        None,
        "an input from a dead connection must never reach the app"
    );
    let waited = started.elapsed();
    assert!(
        waited >= WAIT,
        "recv_timeout returned after {waited:?}, short of its {WAIT:?} timeout"
    );
    assert_eq!(layer.stats().stale_inputs, 1, "the discard must be counted");

    // Nothing was acked for it either: the peer that would read the ack is
    // the one that is gone. A fresh session then works normally, and its
    // inputs are live.
    let mut second = Client::connect(&layer);
    assert!(matches!(second.read_message(), AppToBridge::Hello { .. }));
    second.send_input(1, key("j"));
    assert_eq!(
        layer.recv_timeout(TIMEOUT),
        Some((InputId::new(1), key("j")))
    );
    assert_eq!(second.read_message(), ack(1, InputStatus::Delivered));
    assert_eq!(
        layer.stats().stale_inputs,
        1,
        "a live input is not a stale one"
    );
}

/// One bridge at a time, which is the design and not an accident: the accept
/// loop serves a client to completion before returning to `accept`, so a
/// second one waits in the listen backlog.
///
/// Worth pinning because two agents attached at once is the likeliest way a
/// user meets it, and from the second bridge's side it looks like a socket
/// that connected and then said nothing. What must hold is that the wait is
/// only a wait: the second client is served in full, handshake included, the
/// moment the first goes away.
#[test]
fn a_second_client_waits_until_the_first_is_gone() {
    let mut layer = bind_layer("oneatatime");
    let mut first = Client::connect(&layer);
    assert!(matches!(first.read_message(), AppToBridge::Hello { .. }));

    // The kernel completes this connection into the backlog, so the client
    // has a socket either way; whether it is being served is what differs.
    let mut second = Client::connect(&layer);
    second
        .reader
        .get_ref()
        .set_read_timeout(Some(Duration::from_millis(250)))
        .unwrap();

    // The first client is live throughout: it gets the publish, and the
    // second gets nothing, not even the handshake that opens every session.
    publish_single_node(&mut layer, "while-first");
    let AppToBridge::Snapshot(snapshot) = first.read_message() else {
        panic!("the served client must still receive publishes");
    };
    assert_eq!(snapshot.root.children[0].id.0, "while-first");

    let mut line = String::new();
    let err = second
        .reader
        .read_line(&mut line)
        .expect_err("a waiting client must receive nothing at all");
    assert!(
        matches!(
            err.kind(),
            io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
        ),
        "expected the read to time out, got {err:?} with {line:?}"
    );

    // The first client goes away, which is the only thing the second was
    // waiting for.
    drop(first);
    second
        .reader
        .get_ref()
        .set_read_timeout(Some(TIMEOUT))
        .unwrap();
    assert_eq!(
        second.read_message(),
        AppToBridge::hello("oneatatime", PROTOCOL_VERSION).stamps(true),
        "the waiting client must be served in full once its turn comes"
    );
    let AppToBridge::Snapshot(snapshot) = second.read_message() else {
        panic!("the newly served client must get the latest snapshot");
    };
    assert_eq!(
        snapshot.root.children[0].id.0, "while-first",
        "the tree published while it waited is the one it starts from"
    );

    // And it is a full session, not a leftover: its input reaches the app and
    // is acked on the connection it arrived on.
    second.send_input(1, key("j"));
    assert_eq!(
        layer.recv_timeout(TIMEOUT),
        Some((InputId::new(1), key("j")))
    );
    assert_eq!(second.read_message(), ack(1, InputStatus::Delivered));
}

/// What a client writes while it waits its turn is not its session's input.
///
/// It sits in the kernel's socket buffer until the first client leaves, and
/// used to be read then as the new session's first lines, so the app applied
/// keys a client sent before it had seen any tree at all. Nothing written
/// before the hello can have been written in reply to it, so all of it is
/// discarded, counted and never acked, and the session starts with what the
/// client writes after the hello.
///
/// The blind bytes here end partway through a line, cut inside its leading
/// whitespace, so the remainder the client writes once it is served is a
/// whole input on its own. Only skipping the rest of the cut line keeps that
/// one from the app.
///
/// A third client waits behind both, and is served the moment the second
/// leaves, with no gap in which the listener finds nobody waiting. It waited
/// longer than the second did, so its blind input goes the same way.
#[test]
fn input_a_waiting_client_wrote_before_its_hello_never_reaches_the_app() {
    let mut layer = bind_layer("prehello");
    publish_single_node(&mut layer, "tree");
    let mut first = Client::connect(&layer);
    assert!(matches!(first.read_message(), AppToBridge::Hello { .. }));
    assert!(matches!(first.read_message(), AppToBridge::Snapshot(_)));

    // Written blind while the first client is served: one whole input, and
    // the start of another.
    let mut second = Client::connect(&layer);
    second.send_input(1, key("q"));
    second.writer.write_all(b"    ").unwrap();
    // Queued behind the second, and just as blind.
    let mut third = Client::connect(&layer);
    third.send_input(1, key("d"));

    drop(first);
    assert_eq!(
        second.read_message(),
        AppToBridge::hello("prehello", PROTOCOL_VERSION).stamps(true)
    );
    assert!(matches!(second.read_message(), AppToBridge::Snapshot(_)));

    // The rest of the cut line, then an input written in reply to the hello.
    second.send_input(2, key("y"));
    second.send_input(3, key("j"));

    assert_eq!(
        layer.recv_timeout(TIMEOUT),
        Some((InputId::new(3), key("j"))),
        "neither the blind input nor the rest of the cut line may reach the app"
    );
    assert_eq!(
        second.read_message(),
        ack(3, InputStatus::Delivered),
        "nothing is acked for the blind input or the cut line: the first ack \
         of the session answers the first input written after its hello"
    );
    assert_eq!(layer.try_recv(), None, "nothing else was queued");
    assert_eq!(
        layer.stats().stale_inputs,
        1,
        "the whole blind input is counted once; the cut line is not counted"
    );
    assert_eq!(layer.stats().unknown_inputs, 0);
    assert_eq!(layer.stats().dropped_inputs, 0);

    // The second leaves, and the third, waiting since before the second was
    // served, is served at once.
    drop(second);
    assert_eq!(
        third.read_message(),
        AppToBridge::hello("prehello", PROTOCOL_VERSION).stamps(true)
    );
    assert!(matches!(third.read_message(), AppToBridge::Snapshot(_)));
    third.send_input(2, key("k"));
    assert_eq!(
        layer.recv_timeout(TIMEOUT),
        Some((InputId::new(2), key("k"))),
        "a client served straight after another is drained like it"
    );
    assert_eq!(third.read_message(), ack(2, InputStatus::Delivered));
    assert_eq!(layer.try_recv(), None);
    assert_eq!(
        layer.stats().stale_inputs,
        2,
        "the third's blind input is counted"
    );
}

/// The discard above is for a client that waited behind another, and only
/// for one. A client that finds the app idle is accepted within a poll of
/// connecting, and a script that writes the moment it connects, without
/// reading the hello first (`printf '...' | socat - UNIX:...`), waited behind
/// nobody: what it wrote is its session's input, and 0.3.0 applied it.
#[test]
fn a_lone_client_that_writes_before_reading_its_hello_keeps_its_input() {
    let layer = bind_layer("blindlone");
    let mut client = Client::connect(&layer);
    // Written at once, almost always before the listener's next poll accepts
    // the connection, so it is already waiting when the session starts.
    client.send_input(1, key("j"));

    assert_eq!(
        client.read_message(),
        AppToBridge::hello("blindlone", PROTOCOL_VERSION).stamps(true)
    );
    assert_eq!(
        poll_input(&layer),
        Some(key("j")),
        "the first connection of a process waited behind nobody"
    );
    assert_eq!(client.read_message(), ack(1, InputStatus::Delivered));
    assert_eq!(layer.stats().stale_inputs, 0, "nothing was discarded");
}

/// The same for a client that connects once an earlier session is over and
/// the listener has since found nobody waiting: it connected after that
/// session ended, so it never waited behind it.
#[test]
fn a_client_that_finds_the_app_idle_after_a_session_keeps_its_blind_input() {
    let layer = bind_layer("blindagain");
    let mut first = Client::connect(&layer);
    assert!(matches!(first.read_message(), AppToBridge::Hello { .. }));
    // Half-close, then read to EOF: the barrier proving the layer is done
    // with the first connection.
    first.writer.shutdown(Shutdown::Write).unwrap();
    let mut line = String::new();
    while first.reader.read_line(&mut line).unwrap() > 0 {
        line.clear();
    }
    // Five of the listener's polls, so it has looked for a waiting client
    // since and found none.
    thread::sleep(Duration::from_millis(250));

    let mut second = Client::connect(&layer);
    second.send_input(1, key("j"));
    assert!(matches!(second.read_message(), AppToBridge::Hello { .. }));
    assert_eq!(
        poll_input(&layer),
        Some(key("j")),
        "a client that connected to an idle app waited behind nobody"
    );
    assert_eq!(second.read_message(), ack(1, InputStatus::Delivered));
    assert_eq!(layer.stats().stale_inputs, 0, "nothing was discarded");
}

/// A common way out of an app answers an input and drops the layer in the
/// same frame: `q` handled as quit, an input refused on the way out. Nothing
/// gives the writer a turn in between, so both of that input's acks can still
/// be queued when the shutdown starts, and they used to be thrown away with
/// the connection, leaving the agent to time out on an input the app took.
///
/// A smoke test, and no proof: the writer may well send both acks before the
/// drop, and then this passes without any flush at all. The proof is the
/// `writer_loop` unit tests in `layer.rs`, which put the acks and the
/// shutdown in place before the writer runs.
#[test]
fn acks_queued_as_the_layer_drops_still_reach_the_bridge() {
    let layer = bind_layer("exitflush");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    client.send_input(1, key("q"));
    let mut handled = 0;
    let deadline = Instant::now() + TIMEOUT;
    while handled == 0 && Instant::now() < deadline {
        layer.drain(|_| {
            handled += 1;
            InputStatus::Ignored
        });
        if handled == 0 {
            thread::sleep(Duration::from_millis(1));
        }
    }
    assert_eq!(handled, 1, "the input reaches the app");
    // At once, with nothing in between to give the writer a turn.
    drop(layer);

    assert_eq!(client.read_message(), ack(1, InputStatus::Delivered));
    assert_eq!(client.read_message(), ack(1, InputStatus::Ignored));
    let mut line = String::new();
    assert_eq!(
        client.reader.read_line(&mut line).unwrap(),
        0,
        "the connection closes after the acks, with nothing else: {line:?}"
    );
}

/// `Drop` waits for the writer, and the writer may be partway through a
/// snapshot line when the shutdown comes. Against a bridge that stopped
/// reading, a line far larger than the socket buffers fills them and the
/// write stalls. The stall is bounded per line: a timeout per `write` call
/// let the first call return its partial count after one timeout and the
/// next wait out a second, so dropping took over ten seconds.
#[test]
fn drop_is_bounded_against_a_bridge_that_stopped_reading_mid_snapshot() {
    let TestLayer { mut layer, _dir } = bind_layer("stuckdrop");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    // From here on the client reads nothing.
    publish_900_kb(&mut layer);
    // Long enough for the writer to be stalled in that line.
    thread::sleep(Duration::from_millis(100));
    let took = time_drop(layer);
    assert!(took <= DROP_BOUND + DROP_MARGIN, "took {took:?}");
    // Held to here, so the bridge was stuck rather than gone throughout.
    drop(client);
}

/// The same bound against a bridge that does read, too slowly for the line:
/// 64 KB every 3 s. Each wait for room in the socket then ends in a little
/// progress before its timeout can fire, so a timeout on each wait never
/// fired at all, and dropping the layer waited for the whole line, over
/// thirty seconds.
#[test]
fn drop_is_bounded_against_a_bridge_that_reads_too_slowly() {
    let TestLayer { mut layer, _dir } = bind_layer("slowdrop");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));

    let stop = Arc::new(AtomicBool::new(false));
    let trickle = thread::spawn({
        let stop = Arc::clone(&stop);
        move || {
            let mut chunk = Vec::with_capacity(64 * 1024);
            while !stop.load(Ordering::SeqCst) {
                chunk.clear();
                match (&mut client.reader).take(64 * 1024).read_to_end(&mut chunk) {
                    Ok(0) | Err(_) => return,
                    Ok(_) => {}
                }
                for _ in 0..30 {
                    if stop.load(Ordering::SeqCst) {
                        return;
                    }
                    thread::sleep(Duration::from_millis(100));
                }
            }
        }
    });

    publish_900_kb(&mut layer);
    thread::sleep(Duration::from_millis(100));
    let took = time_drop(layer);
    assert!(took <= DROP_BOUND + DROP_MARGIN, "took {took:?}");
    stop.store(true, Ordering::SeqCst);
    trickle.join().expect("the slow reader");
}

#[test]
fn drop_removes_the_socket_file() {
    let bound = bind_layer("cleanup");
    let path = bound.socket_path().to_path_buf();
    assert!(path.exists());
    // Split the guard so only the layer is dropped here: the socket file
    // removal being asserted must come from the layer, not the temp dir.
    let TestLayer { layer, _dir } = bound;
    drop(layer);
    assert!(!path.exists(), "socket file should be removed on drop");
}

/// The next message, which has to be a snapshot.
fn read_snapshot(client: &mut Client) -> Snapshot {
    match client.read_message() {
        AppToBridge::Snapshot(snapshot) => snapshot,
        other => panic!("expected a snapshot, got {other:?}"),
    }
}

/// Assert the layer writes nothing to `client` for `window`.
fn expect_silence(client: &mut Client, window: Duration) {
    client
        .reader
        .get_ref()
        .set_read_timeout(Some(window))
        .unwrap();
    let mut line = String::new();
    let read = client.reader.read_line(&mut line);
    assert!(
        matches!(
            &read,
            Err(err) if matches!(err.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut)
        ),
        "expected nothing to be written, got {read:?} with {line:?}"
    );
    client
        .reader
        .get_ref()
        .set_read_timeout(Some(TIMEOUT))
        .unwrap();
}

/// How long [`expect_silence`] listens for a line that must not come. A
/// write the layer was going to make is made within a few milliseconds.
const QUIET: Duration = Duration::from_millis(200);

/// Drain until one input has been handed to `f`, failing at the timeout.
fn drain_one(layer: &TariaLayer, mut f: impl FnMut(AgentInput) -> InputStatus) -> AgentInput {
    let deadline = Instant::now() + TIMEOUT;
    let mut got = None;
    while got.is_none() && Instant::now() < deadline {
        layer.drain(|input| {
            got = Some(input.clone());
            f(input)
        });
        thread::sleep(Duration::from_millis(5));
    }
    got.expect("an input should reach the drain")
}

/// The stamp is the newest input handed to the app, by any of the three
/// ways of taking one, and the first snapshot published after it carries it.
#[test]
fn every_way_of_taking_an_input_stamps_the_next_publish() {
    let mut layer = bind_layer("stampapis");
    let mut client = Client::connect(&layer);
    assert_eq!(
        client.read_message(),
        AppToBridge::hello("stampapis", PROTOCOL_VERSION).stamps(true),
        "the hello says this peer stamps"
    );

    publish_single_node(&mut layer, "before");
    let snapshot = read_snapshot(&mut client);
    assert_eq!(snapshot.after_input, None, "nothing has been dequeued yet");

    client.send_input(11, key("a"));
    assert_eq!(poll_try_recv(&layer), Some((InputId::new(11), key("a"))));
    publish_single_node(&mut layer, "after-try-recv");
    assert_eq!(client.read_message(), ack(11, InputStatus::Delivered));
    let snapshot = read_snapshot(&mut client);
    assert_eq!(
        (snapshot.seq, snapshot.after_input),
        (2, Some(InputId::new(11)))
    );

    client.send_input(12, key("b"));
    assert_eq!(
        layer.recv_timeout(TIMEOUT),
        Some((InputId::new(12), key("b")))
    );
    publish_single_node(&mut layer, "after-recv-timeout");
    assert_eq!(client.read_message(), ack(12, InputStatus::Delivered));
    let snapshot = read_snapshot(&mut client);
    assert_eq!(
        (snapshot.seq, snapshot.after_input),
        (3, Some(InputId::new(12)))
    );

    client.send_input(13, key("c"));
    assert_eq!(drain_one(&layer, |_| InputStatus::Delivered), key("c"));
    publish_single_node(&mut layer, "after-drain");
    assert_eq!(client.read_message(), ack(13, InputStatus::Delivered));
    let snapshot = read_snapshot(&mut client);
    assert_eq!(
        (snapshot.seq, snapshot.after_input),
        (4, Some(InputId::new(13)))
    );
    assert_eq!(snapshot.root.children[0].id.0, "after-drain");
}

/// An input that changes nothing still gets a tree stamped at it, or a
/// bridge could not tell it from an input the app is still working on. The
/// tree is the snapshot already sent, under its own seq, so a bridge that
/// reads no stamp sees the same snapshot twice; and without a new dequeue a
/// redraw still sends nothing.
#[test]
fn an_unchanged_frame_after_a_dequeue_is_sent_again_under_the_same_seq() {
    let mut layer = bind_layer("restamp");
    publish_single_node(&mut layer, "tree");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));
    let first = read_snapshot(&mut client);
    assert_eq!((first.seq, first.after_input), (1, None));

    // Nothing dequeued: the redraws are skipped as they always were.
    for _ in 0..3 {
        publish_single_node(&mut layer, "tree");
    }
    expect_silence(&mut client, QUIET);

    client.send_input(5, key("x"));
    assert_eq!(poll_input(&layer), Some(key("x")));
    assert_eq!(client.read_message(), ack(5, InputStatus::Delivered));
    publish_single_node(&mut layer, "tree");
    let again = read_snapshot(&mut client);
    assert_eq!(again.seq, 1, "a stamp moving is not a new snapshot");
    assert_eq!(again.after_input, Some(InputId::new(5)));
    assert_eq!(again.root, first.root);

    // No new dequeue since: skipped again.
    for _ in 0..3 {
        publish_single_node(&mut layer, "tree");
    }
    expect_silence(&mut client, QUIET);

    // And a change is the next snapshot, still stamped at the newest input.
    publish_single_node(&mut layer, "changed");
    let changed = read_snapshot(&mut client);
    assert_eq!(
        (changed.seq, changed.after_input),
        (2, Some(InputId::new(5)))
    );
}

/// Only an input the app was handed stamps a tree. One the queue had no room
/// for, one of a kind this build cannot read, one whose connection ended
/// before the app took it and one a waiting client wrote before its hello
/// were never dequeued, and a stamp naming one would tell a bridge the app
/// had it in hand.
#[test]
fn inputs_the_app_never_took_never_stamp() {
    const QUEUE: u64 = 256;
    let mut layer = bind_layer("nostamp");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));
    publish_single_node(&mut layer, "tree");
    assert_eq!(read_snapshot(&mut client).after_input, None);

    // A kind this build cannot read is answered by the reader and never
    // queued, so a redraw has nothing to stamp.
    client.write_line(r#"{"type":"input","id":41,"input":{"kind":"paste","text":"hi"}}"#);
    assert_eq!(client.read_message(), ack(41, InputStatus::Ignored));
    assert_eq!(layer.try_recv(), None);
    publish_single_node(&mut layer, "tree");
    expect_silence(&mut client, QUIET);

    // A full queue drops the newest input. Every queued one is then taken,
    // and the stamp is the last of those, not the one dropped after them.
    for id in 1..=QUEUE + 1 {
        client.send_input(id, key("k"));
    }
    assert_eq!(client.read_message(), ack(QUEUE + 1, InputStatus::Dropped));
    let mut taken = 0;
    while layer.try_recv().is_some() {
        taken += 1;
    }
    assert_eq!(taken, QUEUE);
    publish_single_node(&mut layer, "tree");
    for id in 1..=QUEUE {
        assert_eq!(client.read_message(), ack(id, InputStatus::Delivered));
    }
    let restamped = read_snapshot(&mut client);
    assert_eq!(restamped.after_input, Some(InputId::new(QUEUE)));

    // An input still queued when its connection ends is discarded, and a
    // client that waited its turn has what it wrote before its hello thrown
    // away. Neither stamps the next connection's tree.
    let mut waiting = Client::connect(&layer);
    waiting.send_input(1, key("blind"));
    client.send_input(QUEUE + 2, key("late"));
    drop(client);
    assert!(matches!(waiting.read_message(), AppToBridge::Hello { .. }));
    assert_eq!(
        read_snapshot(&mut waiting).after_input,
        None,
        "a new connection's first tree is unstamped"
    );
    assert_eq!(layer.try_recv(), None, "both were discarded");
    assert_eq!(layer.stats().stale_inputs, 2);
    publish_single_node(&mut layer, "next");
    let next = read_snapshot(&mut waiting);
    assert_eq!(next.root.children[0].id.0, "next");
    assert_eq!(next.after_input, None);
    assert_eq!(layer.stats().dropped_inputs, 1);
    assert_eq!(layer.stats().unknown_inputs, 1);
}

/// Stamps belong to the connection they were taken on. A new bridge is a new
/// process whose ids start over, so the tree it is greeted with goes out
/// unstamped, and its first input stamps a tree even when it reuses the id a
/// previous bridge's last input had.
#[test]
fn a_new_connection_starts_unstamped() {
    let mut layer = bind_layer("stampconn");
    let mut first = Client::connect(&layer);
    assert!(matches!(first.read_message(), AppToBridge::Hello { .. }));
    first.send_input(3, key("a"));
    assert_eq!(poll_input(&layer), Some(key("a")));
    assert_eq!(first.read_message(), ack(3, InputStatus::Delivered));
    publish_single_node(&mut layer, "one");
    let stamped = read_snapshot(&mut first);
    assert_eq!(
        (stamped.seq, stamped.after_input),
        (1, Some(InputId::new(3)))
    );
    drop(first);

    let mut second = Client::connect(&layer);
    assert!(matches!(second.read_message(), AppToBridge::Hello { .. }));
    let greeted = read_snapshot(&mut second);
    assert_eq!((greeted.seq, greeted.after_input), (1, None));
    assert_eq!(greeted.root, stamped.root);
    publish_single_node(&mut layer, "one");
    expect_silence(&mut second, QUIET);

    second.send_input(3, key("b"));
    assert_eq!(poll_input(&layer), Some(key("b")));
    assert_eq!(second.read_message(), ack(3, InputStatus::Delivered));
    publish_single_node(&mut layer, "one");
    let restamped = read_snapshot(&mut second);
    assert_eq!(
        (restamped.seq, restamped.after_input),
        (1, Some(InputId::new(3))),
        "the same id on a new connection is a new input"
    );
}

/// An `Ignored` refinement still reaches the bridge before the tree
/// published after it, the stamped resend included: a bridge that read the
/// tree first would settle on "did not change" for an input the app refused.
#[test]
fn an_ignored_ack_precedes_the_stamped_snapshot() {
    let mut layer = bind_layer("stampignored");
    let mut client = Client::connect(&layer);
    assert!(matches!(client.read_message(), AppToBridge::Hello { .. }));
    publish_single_node(&mut layer, "tree");
    assert_eq!(read_snapshot(&mut client).seq, 1);

    client.send_input(9, key("x"));
    assert_eq!(drain_one(&layer, |_| InputStatus::Ignored), key("x"));
    publish_single_node(&mut layer, "tree");

    assert_eq!(client.read_message(), ack(9, InputStatus::Delivered));
    assert_eq!(client.read_message(), ack(9, InputStatus::Ignored));
    let snapshot = read_snapshot(&mut client);
    assert_eq!(
        (snapshot.seq, snapshot.after_input),
        (1, Some(InputId::new(9)))
    );
}