taria-ratatui 0.3.2

//! The [`TariaLayer`]: socket lifecycle, background threads, snapshot
//! publishing, and input acknowledgement for a ratatui app.

use std::collections::VecDeque;
use std::env;
use std::fs;
use std::io::{self, BufRead, BufReader, Read, Write};
use std::mem;
use std::net::Shutdown;
use std::os::unix::fs::DirBuilderExt;
use std::os::unix::fs::FileTypeExt;
use std::os::unix::fs::MetadataExt;
use std::os::unix::fs::OpenOptionsExt;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{self, Receiver, SyncSender, TrySendError};
use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

use taria::wire::{AppToBridge, BridgeToApp, InputId, InputStatus, MAX_LINE_BYTES};
use taria::{AgentInput, MAX_NODE_DEPTH, Node, PROTOCOL_VERSION, Role, Snapshot, TreeTooDeep};

use crate::FrameRecorder;
use crate::bound::{self, ROOT_BUDGET, SnapshotTooLarge};

/// How long the writer may spend on any one line before the connection is
/// dropped. Protects the app from a peer that stops reading, or reads too
/// slowly to take a whole line in that time.
///
/// The bound is on the line, not on each wait for room in the socket, which
/// is all a socket's own write timeout bounds (see [`write_all_before`]).
/// Bounded that way, a line far larger than the socket buffers could stall
/// for one timeout per wait: a bridge that had stopped reading held a 900 KB
/// snapshot for two timeouts, and one taking 64 KB every 3 s, which freed
/// room before any wait timed out, held it for forty seconds. [`Drop`] waits
/// for the writer, so either held the app's exit too. Now a bridge that
/// cannot take a whole line, a snapshot of up to [`MAX_LINE_BYTES`] included,
/// within this time is dropped, and reconnects to be sent the latest snapshot
/// afresh.
const WRITE_TIMEOUT: Duration = Duration::from_secs(5);

/// The most one `write` call is handed, so that the kernel waits for room in
/// the socket at most once inside it, and the write timeout, which bounds
/// each such wait rather than the call, bounds the call too. See
/// [`write_all_before`].
///
/// One buffer on Linux, and within the send low-water mark that macOS and the
/// BSDs wait for before they copy anything. A typical line is one or a few
/// calls, and a snapshot at [`MAX_LINE_BYTES`] about five hundred, measured at
/// under a millisecond more than one call for the lot.
const WRITE_CHUNK: usize = 2 * 1024;

/// The longest one wait for room in the socket runs before the writer looks
/// at the clock again. See [`write_all_before`].
///
/// The kernel keeps a socket's write timeout on a timer wheel that files a
/// long timeout coarsely and fires it late: on Linux at 1000 Hz, a 5 s
/// timeout fired up to 430 ms past its time when measured, more than the
/// whole [`SHUTDOWN_FLUSH`]. A timeout this short is filed to the
/// millisecond, so waiting in slices of it holds a deadline to within a few
/// milliseconds, and a writer stalled on a stuck bridge wakes 20 times a
/// second to find it still stuck.
const WRITE_WAIT: Duration = Duration::from_millis(50);

/// How long the writer may spend, in total, on the acks still queued when the
/// layer shuts down.
///
/// A common exit drains input, answers it and drops the layer in one frame:
/// a `q` acted on as quit, a `drain_acking` that returns
/// [`Ignored`](InputStatus::Ignored) on the way out. Nothing gives the writer
/// a turn between those calls, so every one of those answers can still be
/// queued, the [`Delivered`](InputStatus::Delivered) of the dequeue included,
/// and closing the connection without them leaves the agent waiting out a
/// timeout on an input the app did take. Measured before this flush existed,
/// such an app got no ack out at all in 16 runs of 20. So they are written
/// before the connection closes.
///
/// Bounded, because [`Drop`] waits for the writer and an exit must not hang on
/// a bridge that stopped reading; [`WRITE_TIMEOUT`] is far too long for that.
/// The bound is total, not per write, and a peer that reads too slowly loses
/// the acks that miss it, which costs the same timeout the flush exists to
/// spare it. A snapshot still waiting is not written at all: it describes a UI
/// that is about to be gone, and it can be large enough to spend the whole
/// budget on its own.
const SHUTDOWN_FLUSH: Duration = Duration::from_millis(250);

/// Capacity of the agent-input queue between the socket thread and the app.
/// When the app is not draining inputs, the newest input is dropped, counted
/// (see [`TariaLayer::dropped_inputs`]) and acked
/// [`Dropped`](InputStatus::Dropped), instead of blocking the socket thread
/// or queueing without bound.
const INPUT_QUEUE: usize = 256;

/// How many acks may wait for the writer before the oldest is dropped.
///
/// Every other queue in the layer is bounded already: inputs stop at
/// [`INPUT_QUEUE`], snapshots coalesce into a single latest. An ack queue that
/// is not would let a bridge which reads steadily but far slower than an agent
/// sends input, never slowly enough for [`WRITE_TIMEOUT`] to fire, grow the
/// app's memory until the process is killed.
///
/// The *oldest* ack is the one dropped, the opposite end from the input queue:
/// an ack answers one specific input, and the answers an agent is still
/// waiting on are the newest, while the oldest name inputs whose waiter has
/// long since timed out. Drops are counted (see [`TariaLayer::dropped_acks`]).
const ACK_QUEUE: usize = 1024;

/// How many delivered inputs stay answerable with [`TariaLayer::ack`].
///
/// An id is remembered from the moment the app dequeues it until its
/// connection ends, or until this many further inputs have been delivered on
/// that same connection. Remembering them is what lets [`TariaLayer::ack`]
/// tell an id the live bridge is waiting on from one an earlier bridge sent
/// (ids restart with each bridge process). Forgetting one only loses a late
/// ack, which the peer already has to survive; remembering them without bound
/// would be a leak in an app that runs for days.
const ACKABLE_INPUTS: usize = INPUT_QUEUE;

/// How long the listener naps between accept attempts.
///
/// The listener socket is non-blocking, so the thread is never parked in
/// `accept()` and shutdown is noticed without anything connecting: [`Drop`]
/// sets the flag and the thread returns within one nap. Waking a blocking
/// `accept()` by connecting to the socket path cannot do that, because the
/// path may by then hold a socket another process bound, whose listener would
/// take the wake-up while this one stayed parked forever.
const ACCEPT_POLL: Duration = Duration::from_millis(50);

/// How long to wait for the "is anything listening there?" probe that runs
/// before an existing socket file is replaced.
///
/// `connect` is the only liveness probe the standard library offers, and it
/// blocks with no timeout: a socket whose owner stopped accepting with a full
/// backlog would park the app inside `bind`, during startup, which is the one
/// thing taria promises never to do. So the probe runs on a thread of its own
/// and its answer is waited for only this long; no answer in time means
/// "cannot tell", which refuses the bind rather than unlinking a socket that
/// may still be serving.
const SOCKET_PROBE_TIMEOUT: Duration = Duration::from_millis(250);

/// The embeddable taria endpoint for a ratatui app.
///
/// Binding a layer opens a Unix domain socket and spawns a listener thread
/// that serves one bridge client at a time: on connect it sends an
/// [`AppToBridge::Hello`] handshake plus the latest snapshot, then streams
/// every newly published snapshot and forwards incoming [`AgentInput`]
/// messages to the app via [`try_recv`](Self::try_recv) /
/// [`recv_timeout`](Self::recv_timeout). Every forwarded input is answered
/// with an [`AppToBridge::Ack`], so an agent can tell an input the app acted
/// on from one it never saw. An input belongs to the connection it arrived
/// on: one whose bridge disconnected before the app dequeued it is discarded
/// rather than applied, because it was aimed at a session that is gone, and
/// so is one a client wrote while it waited its turn, before the handshake
/// that opens its session (see [`stale_inputs`](Self::stale_inputs)).
///
/// A layer may also be *disabled*: [`bind_or_disabled`](Self::bind_or_disabled)
/// hands back an inert layer rather than an error, so taria failing to bind
/// can never stop an app from starting. Every method stays callable on a
/// disabled layer, which is what lets an app keep one code path.
///
/// Dropping the layer shuts the threads down and removes the socket file, but
/// only while that file is still the one this layer bound: a second instance
/// that took the path over is serving a socket of its own there, and unlinking
/// it would leave that instance running and unreachable.
pub struct TariaLayer {
    app_label: String,
    socket_path: PathBuf,
    /// Everything that exists only once the socket is bound. `None` on a
    /// disabled layer, so no method can reach machinery that is not running.
    inner: Option<Inner>,
    /// Why binding failed, on a disabled layer.
    bind_error: Option<io::Error>,
}

/// The parts of a layer that exist only while its socket is bound.
struct Inner {
    shared: Arc<Shared>,
    input_rx: Receiver<QueuedInput>,
    listener: Option<JoinHandle<()>>,
    /// Device and inode of the socket file as bound, so [`Drop`] can tell it
    /// from a file that replaced it. `None` when it could not be read, which
    /// makes every comparison fail and leaves the path alone.
    socket_ident: Option<(u64, u64)>,
    seq: u64,
    /// The root of the newest snapshot published.
    last_root: Option<Node>,
    /// The tree the newest frame built, as assembled and before anything was
    /// cut from it, when a cut made it differ from `last_root`. `None` when
    /// that frame's tree is `last_root` itself, so a tree that fits is held
    /// once rather than twice, and the next frame is compared against
    /// `last_root` instead.
    ///
    /// Kept for a frame that was not published too: a tree that changed only
    /// past a cut is cut to the snapshot already sent, and remembering it is
    /// what lets the next frame, drawn unchanged, skip the cut.
    last_raw: Option<Node>,
    /// Snapshots published with a branch cut for depth, and the failure
    /// behind the newest cut. Owned here rather than in [`Shared`] because
    /// only the app thread publishes and only the app thread reads these.
    truncated: u64,
    last_truncation: Option<TreeTooDeep>,
    /// Snapshots published with something cut to fit one line, and what was
    /// cut from the newest of them. Owned here for the same reason.
    oversized: u64,
    last_oversize: Option<SnapshotTooLarge>,
}

impl Inner {
    /// Is `root` the tree the newest frame built, as assembled?
    ///
    /// Compared against what that frame kept: the tree it was cut from, when
    /// it had to be cut, or else the snapshot, which is then the same tree.
    fn is_newest_frame(&mut self, root: &mut Node) -> bool {
        match self.last_raw.as_mut().or(self.last_root.as_mut()) {
            Some(newest) => same_tree(root, newest),
            None => false,
        }
    }
}

impl Drop for Inner {
    /// Free the trees without recursing: `last_raw` is the tree as built,
    /// which can be deeper than the call stack.
    fn drop(&mut self) {
        for tree in [self.last_raw.take(), self.last_root.take()]
            .into_iter()
            .flatten()
        {
            drop_iteratively(tree);
        }
    }
}

/// One agent input on its way from a connection's reader thread to the app.
///
/// The generation tags the connection it arrived on, so an input queued by a
/// bridge that has since disconnected can be told apart from one the live
/// bridge is waiting on.
struct QueuedInput {
    generation: u64,
    id: InputId,
    input: AgentInput,
}

impl TariaLayer {
    /// Bind the taria socket for this app and start serving.
    ///
    /// The socket path is resolved by [`taria::socket::resolve_path`], in
    /// order of preference:
    ///
    /// 1. `$TARIA_SOCK` verbatim, if set and non-empty;
    /// 2. `$XDG_RUNTIME_DIR/taria/<app_label>.sock`;
    /// 3. `<temp_dir>/taria-<uid or user>/<app_label>.sock`.
    ///
    /// The parent directory is created with mode `0700` and then verified to
    /// be a private directory (a real directory, owned by the current user,
    /// with no group/other permission bits); binding is refused otherwise,
    /// because a directory another user controls would let them replace or
    /// redirect the socket.
    ///
    /// A socket file left at the path by a run that never got to clean up is
    /// removed before binding. Nothing else is: a file that is not a socket
    /// stays where it is, and so does a socket another instance is still
    /// serving. Both refuse the bind instead, with an error saying which it
    /// was; see [`bind_at`](Self::bind_at).
    ///
    /// The label becomes the socket's file name, so it must be one: a label
    /// carrying a path separator, or `.` or `..`, is refused here rather than
    /// resolved into a path this layer would go on to bind and unlink. An app
    /// that needs a path of its own passes it to
    /// [`bind_at`](Self::bind_at).
    ///
    /// Apps that would rather run without taria than not run at all should
    /// use [`bind_or_disabled`](Self::bind_or_disabled).
    pub fn bind(app_label: &str) -> io::Result<Self> {
        let socket_path = resolve_socket_path(app_label).map_err(invalid_label)?;
        Self::bind_at(app_label, socket_path)
    }

    /// Like [`bind`](Self::bind), but at an explicit socket path, skipping
    /// resolution. Useful for tests and apps that manage their own runtime
    /// directories. The same parent-directory creation and privacy checks as
    /// [`bind`](Self::bind) apply: a relative path (including a bare
    /// filename) is resolved against the current directory first, so the
    /// parent directory that would hold the socket is always vetted.
    ///
    /// # What is already at the path
    ///
    /// Binding needs the path to be free, and only a dead socket is cleared
    /// to make it so:
    ///
    /// * nothing there: bound;
    /// * a socket nothing is listening on, left by a run that was killed
    ///   before it could clean up: unlinked, then bound;
    /// * a socket another instance is serving: refused with
    ///   [`AddrInUse`](io::ErrorKind::AddrInUse), because taking it over
    ///   would leave that instance running with no way for a bridge to reach
    ///   it. Run the second instance on a socket of its own instead;
    /// * anything that is not a socket: refused with
    ///   [`AlreadyExists`](io::ErrorKind::AlreadyExists), the file untouched.
    ///   The path can come verbatim from `$TARIA_SOCK`, and a typo there is
    ///   not a reason to delete a file the user meant to keep.
    pub fn bind_at(app_label: &str, socket_path: impl Into<PathBuf>) -> io::Result<Self> {
        // Absolutize before looking at the parent: a bare filename like
        // "app.sock" has an empty parent, which must not bypass the privacy
        // check by silently binding in an unvetted current directory.
        let socket_path = absolutize(socket_path.into())?;

        // Check the length here, before anything is created: over the limit,
        // `bind` below would fail with `InvalidInput: path must be shorter
        // than SUN_LEN`, which names neither the path, its length, the limit,
        // nor a way out. `SocketPathTooLong` names all four.
        taria::socket::check_path_len(&socket_path)
            .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err.to_string()))?;

        if let Some(parent) = socket_path.parent()
            && !parent.as_os_str().is_empty()
        {
            ensure_private_dir(parent)?;
        }
        // Free the path if a dead socket is holding it, and refuse the bind
        // if anything else is.
        clear_socket_path(&socket_path)?;

        let listener = UnixListener::bind(&socket_path)?;
        // Polled rather than parked in `accept()`, so shutdown needs no
        // connection to be noticed; see `ACCEPT_POLL`.
        if let Err(err) = listener.set_nonblocking(true) {
            drop(listener);
            let _ = fs::remove_file(&socket_path);
            return Err(err);
        }
        // Identify the socket just bound, before anyone else can replace it.
        let socket_ident = socket_ident(&socket_path);
        let shared = Arc::new(Shared {
            app_label: app_label.to_string(),
            state: Mutex::new(State::default()),
            cv: Condvar::new(),
            dropped_inputs: AtomicU64::new(0),
            stale_inputs: AtomicU64::new(0),
            unknown_inputs: AtomicU64::new(0),
            dropped_acks: AtomicU64::new(0),
        });
        let (input_tx, input_rx) = mpsc::sync_channel(INPUT_QUEUE);

        let thread_shared = Arc::clone(&shared);
        let handle = thread::Builder::new()
            .name("taria-listener".into())
            .spawn(move || accept_loop(listener, thread_shared, input_tx));
        let handle = match handle {
            Ok(handle) => handle,
            Err(err) => {
                let _ = fs::remove_file(&socket_path);
                return Err(err);
            }
        };

        Ok(Self {
            app_label: app_label.to_string(),
            socket_path,
            inner: Some(Inner {
                shared,
                input_rx,
                listener: Some(handle),
                socket_ident,
                seq: 0,
                last_root: None,
                last_raw: None,
                truncated: 0,
                last_truncation: None,
                oversized: 0,
                last_oversize: None,
            }),
            bind_error: None,
        })
    }

    /// Bind like [`bind`](Self::bind), or return a disabled layer when that
    /// fails, so an app never has to choose between starting and having
    /// taria.
    ///
    /// A disabled layer answers every method: publishing does nothing, no
    /// input ever arrives, and [`frame`](Self::frame) still hands back a
    /// recorder, so the render path needs no branch on whether taria came up.
    /// [`is_enabled`](Self::is_enabled) and [`bind_error`](Self::bind_error)
    /// report what happened.
    ///
    /// The layer deliberately prints nothing itself: once the app owns the
    /// alternate screen, a stray print garbles the display. Print
    /// [`bind_error`](Self::bind_error) at the moment the app chooses, which
    /// is usually before entering the alternate screen or after leaving it.
    pub fn bind_or_disabled(app_label: &str) -> Self {
        match resolve_socket_path(app_label) {
            Ok(socket_path) => Self::bind_or_disabled_at(app_label, socket_path),
            // A refused label resolves to no path at all, so the disabled
            // layer carries an empty one: the error names the label, which is
            // the thing that has to change, and an invented path would only
            // look like somewhere the socket might be.
            Err(err) => Self::disabled(app_label, PathBuf::new(), invalid_label(err)),
        }
    }

    /// [`bind_or_disabled`](Self::bind_or_disabled) at an explicit path, so
    /// the disabled branch can be exercised without mutating the process
    /// environment (which no test can do safely while other tests run).
    fn bind_or_disabled_at(app_label: &str, socket_path: PathBuf) -> Self {
        match Self::bind_at(app_label, socket_path.clone()) {
            Ok(layer) => layer,
            Err(err) => Self::disabled(app_label, socket_path, err),
        }
    }

    /// An inert layer that answers every method and serves nothing.
    fn disabled(app_label: &str, socket_path: PathBuf, err: io::Error) -> Self {
        Self {
            app_label: app_label.to_string(),
            socket_path,
            inner: None,
            bind_error: Some(err),
        }
    }

    /// Is this layer serving on a socket?
    ///
    /// False only for a disabled layer from
    /// [`bind_or_disabled`](Self::bind_or_disabled); see
    /// [`bind_error`](Self::bind_error) for why it is disabled.
    pub fn is_enabled(&self) -> bool {
        self.inner.is_some()
    }

    /// Why binding failed, on a disabled layer. `None` while the layer is
    /// serving.
    ///
    /// The layer never prints this itself, because a print while the
    /// alternate screen is up garbles the display. The app decides when it is
    /// safe to show and whether it is worth showing at all.
    pub fn bind_error(&self) -> Option<&io::Error> {
        self.bind_error.as_ref()
    }

    /// The path of the Unix socket this layer is serving on, or the path it
    /// tried to bind if it is disabled, so an app can print either.
    pub fn socket_path(&self) -> &Path {
        &self.socket_path
    }

    /// The label passed to [`bind`](Self::bind), used in the handshake and as
    /// the auto-generated root node's label.
    pub fn app_label(&self) -> &str {
        &self.app_label
    }

    /// Begin recording a new frame. Record nodes with
    /// [`FrameRecorder::push`] or [`sem`](crate::sem), then call
    /// [`FrameRecorder::publish`].
    ///
    /// Works on a disabled layer too, where publishing is simply a no-op, so
    /// the render path never branches on whether taria is up.
    pub fn frame(&mut self) -> FrameRecorder<'_> {
        FrameRecorder::new(self)
    }

    /// Non-blocking poll for the next agent input, if one has arrived.
    ///
    /// Acks the input [`Delivered`](InputStatus::Delivered) on its way out.
    /// Delivered means the app's event loop dequeued it, nothing more; an app
    /// that then decides to do nothing with it should say so with
    /// [`ack`](Self::ack), whose id comes from
    /// [`try_recv_with_id`](Self::try_recv_with_id).
    ///
    /// Inputs left over from a bridge connection that has since ended are
    /// discarded here rather than handed over, and input a client wrote before
    /// its handshake never reaches the queue at all; see
    /// [`stale_inputs`](Self::stale_inputs).
    ///
    /// [`AgentInput::Unknown`] never arrives here either. An input whose
    /// `kind` this build cannot read carries nothing to apply, so the layer
    /// acks it [`Ignored`](InputStatus::Ignored) where it is parsed and never
    /// queues it; see [`unknown_inputs`](Self::unknown_inputs).
    ///
    /// Always `None` on a disabled layer.
    pub fn try_recv(&self) -> Option<AgentInput> {
        self.try_recv_with_id().map(|(_, input)| input)
    }

    /// Wait up to `timeout` for the next agent input.
    ///
    /// Acks [`Delivered`](InputStatus::Delivered) and discards stale inputs
    /// exactly like [`try_recv`](Self::try_recv); a discard consumes none of
    /// the budget, the wait resumes for what is left of `timeout`. A disabled
    /// layer has nothing to wait for but still waits out the timeout, so an
    /// app that paces its loop on this call keeps its timing whether or not
    /// taria bound.
    ///
    /// A `timeout` so large that no clock can hold the deadline (near
    /// [`Duration::MAX`]) waits for an input for as long as the queue can
    /// deliver one, which is what such a deadline asks for anyway. On a
    /// disabled layer that is no time at all: its queue can never deliver, so
    /// the call returns at once rather than parking the app forever on a wait
    /// with no end.
    pub fn recv_timeout(&self, timeout: Duration) -> Option<AgentInput> {
        self.recv_timeout_with_id(timeout).map(|(_, input)| input)
    }

    /// [`try_recv`](Self::try_recv), keeping the [`InputId`] so the app can
    /// answer the input again later with [`ack`](Self::ack).
    ///
    /// The ack still happens here, so an app can never leave an input
    /// unacknowledged by forgetting to call something.
    pub fn try_recv_with_id(&self) -> Option<(InputId, AgentInput)> {
        let inner = self.inner.as_ref()?;
        loop {
            let queued = inner.input_rx.try_recv().ok()?;
            if inner.shared.deliver(queued.generation, queued.id) {
                return Some((queued.id, queued.input));
            }
        }
    }

    /// [`recv_timeout`](Self::recv_timeout), keeping the [`InputId`] so the
    /// app can answer the input again later with [`ack`](Self::ack).
    pub fn recv_timeout_with_id(&self, timeout: Duration) -> Option<(InputId, AgentInput)> {
        let Some(inner) = self.inner.as_ref() else {
            // No socket, so no input will ever arrive. Sleep the timeout out
            // anyway: a caller using this as its clock must not spin because
            // taria happened to be unavailable.
            //
            // Except for a timeout no clock can hold, which asks to wait
            // until an input arrives. A disabled layer's queue can never
            // deliver one, so that wait would have no end, and freezing
            // because taria could not bind is the very failure
            // `bind_or_disabled` exists to prevent: an app pacing its loop on
            // this call would run bound and hang unbound. Returning at once
            // is also what the enabled path does with a queue that can no
            // longer deliver, where `recv` on a channel whose senders are
            // gone returns immediately.
            if Instant::now().checked_add(timeout).is_some() {
                thread::sleep(timeout);
            }
            return None;
        };
        // Discarding a stale input must not cut the wait short: an app that
        // paces its event loop on this call would otherwise spin through the
        // rest of its budget the moment a bridge disconnects.
        //
        // `Instant + Duration` panics when the deadline is not representable,
        // and a library crate must not take the app down over an argument.
        // A deadline that far out is indistinguishable from none at all, so
        // it becomes a plain blocking wait instead.
        let deadline = Instant::now().checked_add(timeout);
        loop {
            let queued = match deadline {
                Some(deadline) => {
                    let remaining = deadline.saturating_duration_since(Instant::now());
                    inner.input_rx.recv_timeout(remaining).ok()?
                }
                None => inner.input_rx.recv().ok()?,
            };
            if inner.shared.deliver(queued.generation, queued.id) {
                return Some((queued.id, queued.input));
            }
        }
    }

    /// Hand every queued agent input to `f`, in arrival order.
    ///
    /// The drain loop an app otherwise writes by hand, once per place it
    /// polls for input, with nothing keeping the copies in step.
    pub fn drain(&self, mut f: impl FnMut(AgentInput)) {
        while let Some(input) = self.try_recv() {
            f(input);
        }
    }

    /// [`drain`](Self::drain), passing each input's [`InputId`] alongside it
    /// so the handler can [`ack`](Self::ack) whatever it chose to ignore.
    ///
    /// Reach for [`drain_acking`](Self::drain_acking) first: an app that only
    /// wants to refine an ack needs the id for nothing else, and every
    /// integration that wrote this loop by hand wrote the same four lines
    /// around it.
    pub fn drain_with_ids(&self, mut f: impl FnMut(InputId, AgentInput)) {
        while let Some((id, input)) = self.try_recv_with_id() {
            f(id, input);
        }
    }

    /// [`drain`](Self::drain) where the handler returns each input's status
    /// and the layer sends the ones that say something new.
    ///
    /// Answering an input the app looked at and deliberately did nothing with
    /// is the whole reason [`drain_with_ids`](Self::drain_with_ids) hands out
    /// an [`InputId`], and two independent apps wrote the same
    /// dequeue-compare-[`ack`](Self::ack) loop around it, each with its own
    /// two-variant verdict enum in front. Returning
    /// [`Ignored`](InputStatus::Ignored) says the same thing with no id in the
    /// app's hands and no enum of its own.
    ///
    /// Return [`Delivered`](InputStatus::Delivered) for an input the app
    /// acted on. It sends nothing: the dequeue already acked it, and
    /// re-sending would double this layer's ack traffic, which on a bridge
    /// reading slower than the app answers costs the refinements queued
    /// behind it (see [`dropped_acks`](Self::dropped_acks)).
    ///
    /// `drain_with_ids` stays for an app that needs the id for something
    /// else, such as an input it can only answer after a later frame.
    pub fn drain_acking(&self, mut f: impl FnMut(AgentInput) -> InputStatus) {
        while let Some((id, input)) = self.try_recv_with_id() {
            let status = f(input);
            if status != InputStatus::Delivered {
                self.ack(id, status);
            }
        }
    }

    /// Answer an input again, overriding the ack it already has.
    ///
    /// Last ack wins. An app that dequeued an input (acked
    /// [`Delivered`](InputStatus::Delivered) by that dequeue) and then
    /// deliberately did nothing with it can refine that to
    /// [`Ignored`](InputStatus::Ignored) so an agent waiting on an effect
    /// stops waiting. That is every input kind, not only an act a modal dialog
    /// blocks or naming a node this app does not know: a `set_value` carrying
    /// no value, an [`AgentInput::Text`] sent while nothing here is accepting
    /// typing, an [`AgentInput::Key`] this app could not lower, and an input
    /// kind this build has never heard of. [`InputStatus::Ignored`] draws the
    /// line, including why a key bound to nothing is still delivered.
    ///
    /// The ack travels the same path as snapshots, so it can never overtake
    /// the snapshot published after it. Does nothing on a disabled layer.
    ///
    /// An id whose bridge connection has since ended is answered nowhere.
    /// Ids are unique only within a bridge process and a fresh bridge counts
    /// from the start again, so an id an app kept across a disconnect can
    /// already name a live input of the next bridge's, and that bridge's
    /// waiter must not be handed a verdict from a session it never saw.
    /// Holding an id across a disconnect is not a mistake, so the drop is
    /// silent: the peer this ack was for is gone, and there is nobody left to
    /// tell. An id also stops being answerable once its connection has
    /// delivered a queue's worth of newer inputs past it, which costs at most
    /// a late ack the peer already has to survive.
    pub fn ack(&self, id: InputId, status: InputStatus) {
        if let Some(inner) = self.inner.as_ref() {
            inner.shared.queue_app_ack(id, status);
        }
    }

    /// How many agent inputs have been dropped so far because the input
    /// queue was full. Always 0 on a disabled layer.
    ///
    /// When an agent floods inputs faster than the app drains them, the layer
    /// drops the newest input rather than blocking the socket thread. The
    /// agent learns of each drop from its [`Dropped`](InputStatus::Dropped)
    /// ack; this counter is the app-facing tally of the same event, kept
    /// because the layer must not print while the app owns the alternate
    /// screen. It is monotonic and accumulates across reconnects for the
    /// lifetime of the layer.
    ///
    /// Apps should read it after restoring the terminal (for example right
    /// before exit, once the alternate screen has been left) and report a
    /// nonzero value on stderr or in a log.
    pub fn dropped_inputs(&self) -> u64 {
        self.inner.as_ref().map_or(0, |inner| {
            inner.shared.dropped_inputs.load(Ordering::Relaxed)
        })
    }

    /// How many agent inputs have been discarded so far, rather than handed to
    /// the app, because they were not aimed at the session that was live when
    /// they were read. Always 0 on a disabled layer.
    ///
    /// Two things put an input here. The first is a bridge connection that
    /// ended before the app dequeued what arrived on it. Such an input was
    /// aimed at a bridge session that no longer exists: applying it would act
    /// on an agent's intent minutes or milliseconds after the agent that
    /// formed it is gone, with nobody left to receive the result. A `key q`
    /// that arrives just before the bridge restarts would otherwise quit the
    /// app on the next frame. The bridge drops its own queue on reconnect for
    /// the mirror-image reason, so both sides agree that an input belongs to
    /// one connection.
    ///
    /// The second is input a client wrote while it waited behind another.
    /// Only one bridge is served at a time, so a client that connects while
    /// another is attached waits in the listen backlog, and whatever it writes
    /// meanwhile sits unread in the kernel's socket buffer until its turn
    /// comes. That client has seen no tree, so its inputs name nodes and keys
    /// it guessed at, and applying them once the first client leaves would
    /// act on a UI it never looked at. So a connection that was already
    /// waiting when the previous session ended has what it wrote read off and
    /// thrown away just before its hello goes out, and every complete line of
    /// that which is an input is counted here, a kind this build cannot read
    /// included. None of it is acked: the session it would be answered on had
    /// not begun. Clients waiting back to back are each treated so.
    ///
    /// A client that waited behind nobody keeps what it wrote before reading
    /// its hello: the first connection of a process, and one that connects
    /// after the listener last found nobody waiting. A script that connects
    /// to an idle app and writes at once is such a client, and its input is
    /// its session's, as it was in 0.3.0. A client that waits for the hello
    /// before writing is never affected either way, and taria-mcp is one,
    /// whose input tools refuse to send until the app's first snapshot has
    /// arrived.
    ///
    /// Monotonic across reconnects, and read like
    /// [`dropped_inputs`](Self::dropped_inputs): after the terminal is
    /// restored, never while the app owns the alternate screen.
    pub fn stale_inputs(&self) -> u64 {
        self.inner
            .as_ref()
            .map_or(0, |inner| inner.shared.stale_inputs.load(Ordering::Relaxed))
    }

    /// How many agent inputs have been answered without reaching the app
    /// because their `kind` is one this build cannot read. Always 0 on a
    /// disabled layer.
    ///
    /// An input like that parses as [`AgentInput::Unknown`], which carries
    /// neither the kind it arrived under nor what it asked for, so there is
    /// nothing for the app to apply. The layer acks it
    /// [`Ignored`](InputStatus::Ignored) itself and never queues it; see
    /// [`try_recv`](Self::try_recv), which is why an app never has to think
    /// about the variant.
    ///
    /// A nonzero value means the bridge on the other end is built against a
    /// newer taria than this app, and the agent is asking for things this app
    /// cannot do yet. Raising the app's taria dependency is the fix. Read it
    /// like [`dropped_inputs`](Self::dropped_inputs): after the terminal is
    /// restored, never while the app owns the alternate screen.
    pub fn unknown_inputs(&self) -> u64 {
        self.inner.as_ref().map_or(0, |inner| {
            inner.shared.unknown_inputs.load(Ordering::Relaxed)
        })
    }

    /// How many acks have been dropped so far because more answers were
    /// waiting for the bridge than the queue holds. Always 0 on a disabled
    /// layer.
    ///
    /// Acks queue for the writer thread, and a bridge that reads steadily but
    /// far slower than an agent sends input never stalls long enough for the
    /// write timeout to disconnect it. The queue is bounded so the app cannot
    /// be grown out of memory by such a peer; the oldest answers go first,
    /// because the ones an agent is still waiting on are the newest.
    ///
    /// A nonzero value means some agent requests were answered nowhere and
    /// their callers waited out a timeout instead. The slow end is the bridge,
    /// not the app. Monotonic across reconnects, and read like
    /// [`dropped_inputs`](Self::dropped_inputs): after the terminal is
    /// restored, never while the app owns the alternate screen.
    pub fn dropped_acks(&self) -> u64 {
        self.inner
            .as_ref()
            .map_or(0, |inner| inner.shared.dropped_acks.load(Ordering::Relaxed))
    }

    /// How many published snapshots have had a branch cut for being deeper
    /// than [`taria::MAX_NODE_DEPTH`]. Always 0 on a disabled layer.
    ///
    /// It counts snapshots, not frames. A frame whose tree, once cut, is the
    /// snapshot already published is not sent again and is not counted
    /// again, so a deep tree that sits still while the app redraws it sixty
    /// times a second counts once, and a deep tree that changes counts once
    /// per change. The number says how many trees the agent was handed cut,
    /// which is the thing worth knowing, rather than how often the app drew
    /// one.
    ///
    /// See [`publish`](Self::publish) for what the cut does and why.
    /// [`last_truncation`](Self::last_truncation) names the branch. Monotonic
    /// for the lifetime of the layer, and read like
    /// [`dropped_inputs`](Self::dropped_inputs): after the terminal is
    /// restored, never while the app owns the alternate screen.
    pub fn truncated_snapshots(&self) -> u64 {
        self.inner.as_ref().map_or(0, |inner| inner.truncated)
    }

    /// The depth failure behind the most recent truncation, or `None` if no
    /// published tree has been cut.
    ///
    /// [`TreeTooDeep`](taria::TreeTooDeep) carries the depth measured and the
    /// id of a node found at it. An app whose tree is generated from data has
    /// no other way to tell which branch ran away, which is the whole reason
    /// the counter alone is not enough here.
    pub fn last_truncation(&self) -> Option<TreeTooDeep> {
        self.inner.as_ref()?.last_truncation.clone()
    }

    /// How many published snapshots have had something cut to fit in one
    /// line of [`taria::wire::MAX_LINE_BYTES`]. Always 0 on a disabled layer.
    ///
    /// Counted like [`truncated_snapshots`](Self::truncated_snapshots):
    /// snapshots, not frames, so a tree that stays too large while the app
    /// redraws it counts once, and one that changes counts once per change.
    ///
    /// See [`publish`](Self::publish) for what is cut and why.
    /// [`last_oversize`](Self::last_oversize) says how large the line would
    /// have been and where the bytes were. Monotonic for the lifetime of the
    /// layer, and read like [`dropped_inputs`](Self::dropped_inputs): after
    /// the terminal is restored, never while the app owns the alternate
    /// screen.
    pub fn oversized_snapshots(&self) -> u64 {
        self.inner.as_ref().map_or(0, |inner| inner.oversized)
    }

    /// What was cut from the most recent oversized snapshot, or `None` if no
    /// published tree has had to be cut to fit.
    ///
    /// [`SnapshotTooLarge`] carries the size the line would have had and the
    /// id of the node holding the longest string. A tree that runs over has
    /// usually run away in one place, a draft, a log, a list holding every
    /// row, and the counter alone cannot say which.
    pub fn last_oversize(&self) -> Option<SnapshotTooLarge> {
        self.inner.as_ref()?.last_oversize.clone()
    }

    /// Publish `nodes` as a snapshot, without going through a
    /// [`FrameRecorder`].
    ///
    /// For apps that build their node list separately from drawing, where the
    /// recorder's frame, push-loop and publish triplet is three lines saying
    /// one thing. The nodes are wrapped in the same auto-generated root, so
    /// both paths produce the same tree.
    ///
    /// # Trees too deep to send
    ///
    /// A tree deeper than [`taria::MAX_NODE_DEPTH`], counting the root this
    /// method adds, is cut here: every node at the limit publishes without
    /// its children. The cut is per branch, so everything above it reaches
    /// the agent unchanged.
    ///
    /// Cutting rather than refusing, because the two alternatives are both
    /// worse than losing the deep part. Publishing the tree as built is what
    /// breaks the peer: a snapshot past the limit exceeds what a JSON parser
    /// will recurse into, the bridge skips the line exactly as it skips a
    /// truncated one, and the agent is left reading a stale tree or told no
    /// app has published yet, with nothing anywhere saying why. Skipping the
    /// publish is that same stale tree, chosen deliberately.
    ///
    /// The app is told instead of the agent, because the app is the only one
    /// that can fix it: see
    /// [`truncated_snapshots`](Self::truncated_snapshots) for the count and
    /// [`last_truncation`](Self::last_truncation) for the branch. The fix is
    /// almost always to publish what the widget draws rather than the data
    /// behind it, which for a deep tree view is the expanded path and the
    /// rows currently on screen.
    ///
    /// # Trees too large to send
    ///
    /// A snapshot travels as one line, and a bridge reads a line up to
    /// [`taria::wire::MAX_LINE_BYTES`] and no further: past that it cannot
    /// find where the next message starts, so it drops the connection.
    /// Dropping it fixes nothing. The bridge reconnects, is sent the latest
    /// snapshot, which is the same one, and drops that connection too, so an
    /// app holding one long value (a draft, a log tail, a transcript) leaves
    /// the agent reconnecting for as long as the value stays long, with
    /// nothing on either side saying why.
    ///
    /// So every tree is bounded here, after the depth cut, to fit that line
    /// at any `seq`. The bound cuts what costs the most first and stops as
    /// soon as the tree fits:
    ///
    /// 1. nodes that could never fit, one whose id, role and actions, with
    ///    its strings at the floor, cannot fit beside its ancestors, are
    ///    dropped with everything under them before any string is cut, so
    ///    that they cost their siblings nothing;
    /// 2. strings are cut: the longest values first, all down to one shared
    ///    length, so the biggest shrink first and shorter ones are untouched,
    ///    then values and labels together, the same way, when values alone
    ///    are not enough. A cut value ends in `…`, except that a
    ///    [`Role::Log`] or [`Role::Terminal`] value keeps its end and starts
    ///    with `…` instead, because the end is where those grow. A string of
    ///    256 bytes of JSON or less, the floor, is never cut;
    /// 3. every node keeps at most one shared number of children, the
    ///    largest that fits, so a list of fifty thousand rows loses its last
    ///    rows while a status bar beside it keeps its place;
    /// 4. shallowest first, the children that cap removed go back while they
    ///    fit, each parent keeping a run of its first children with no gap,
    ///    apart from a node that could never fit, and the first child too
    ///    large for the room left ends its parent's run;
    /// 5. if all else fails, a bare root, which always fits.
    ///
    /// The third and fourth stages plan with every string at the floor, but
    /// the strings of the nodes they keep are then cut only as far as the
    /// kept tree needs, and not at all when it fits as it is.
    ///
    /// Ids and custom action names are never changed, so whatever an agent
    /// can read it can still act on. Focus is decided after the bound, for
    /// the reason it is decided after the depth cut: the focused node can be
    /// one of the rows dropped.
    ///
    /// A frame whose tree is the one the frame before it built stops before
    /// any of this: it is compared and nothing else, neither measured nor
    /// cut, so a tree that sits still costs the same whether or not it fits.
    /// A tree that changed is measured exactly, by serializing it into a
    /// counter that keeps nothing, and one that does not fit is cut again on
    /// every frame it changes, so the bound is a guard, not a way to publish:
    /// see [`oversized_snapshots`](Self::oversized_snapshots) and
    /// [`last_oversize`](Self::last_oversize) for what was cut and where.
    /// The fix is the one for depth, to publish what the widget draws: the
    /// lines of a log the pane shows, the rows of a list on screen, the
    /// visible part of a draft.
    pub fn publish(&mut self, nodes: impl IntoIterator<Item = Node>) {
        self.publish_nodes(nodes.into_iter().collect());
    }

    /// Publish a frame's recorded top-level nodes as a new snapshot.
    ///
    /// Wraps the nodes in an auto-generated `app` root (focused only if no
    /// recorded node is), bumps `seq`, and hands the snapshot to the writer
    /// thread. Skipped when the tree is the one the previous frame built,
    /// before anything is cut or measured; when what is left once it is cut
    /// is the previous snapshot; and on a disabled layer. Never blocks the
    /// render path beyond a brief mutex hold.
    ///
    /// A tree over [`MAX_NODE_DEPTH`] is cut before any of that, and then a
    /// tree too large for one line; see [`publish`](Self::publish).
    pub(crate) fn publish_nodes(&mut self, nodes: Vec<Node>) {
        let Some(inner) = self.inner.as_mut() else {
            // Freed without recursing even here: the tree as built can be
            // deeper than the call stack, and a disabled layer must not be
            // the one thing that cannot take it.
            nodes.into_iter().for_each(drop_iteratively);
            return;
        };
        let mut root = Node::new("app", Role::App)
            .label(inner.shared.app_label.clone())
            .children(nodes);
        // Focus is decided on the tree as built, so that a tree no cut
        // touches is already the snapshot it goes out as, and the next frame
        // can be compared against that snapshot with no second copy kept.
        root.focused = !focused_below(&root);
        // A frame drawn unchanged stops here, before anything is cut or
        // measured. Everything below is a function of the tree alone: the
        // root's label is the app's, fixed when the layer was bound, the
        // depth limit and the budget are constants, and focus reads nothing
        // but the tree. So the same tree would be cut to the same snapshot,
        // the one already sent, and skipping the work changes nothing but
        // what it costs, which for a tree too large for the line is the
        // whole bound on every frame the app draws. The tree as built is
        // freed without recursing, here and wherever else the layer lets go
        // of one, because the depth cut below exists for trees deeper than
        // the call stack, and the drop glue `Node` derives recurses.
        if inner.is_newest_frame(&mut root) {
            drop_iteratively(root);
            return;
        }

        // The root added here counts towards the depth, so the check has to
        // run on the assembled tree rather than on the caller's nodes. A
        // tree over the limit goes out as a copy cut to it, and the tree as
        // built is kept to compare the next frame against. Focus is decided
        // again on the copy: cutting a branch can remove the only focused
        // node, and a snapshot has to carry focus somewhere.
        let too_deep = root.check_depth().err();
        let (mut tree, mut raw) = if too_deep.is_some() {
            let mut cut = copy_within_depth(&mut root);
            cut.focused = !focused_below(&cut);
            (cut, Some(root))
        } else {
            (root, None)
        };

        // Then the size, which has to come after the depth cut: measuring
        // serializes the tree, which recurses, and the cut is what makes
        // that safe. Focus was decided before the measure so that the size
        // the bound reports is that of the tree as it would otherwise have
        // gone out, and is decided again after a cut, because a dropped row
        // can be the focused one. The tree as built is copied only when it
        // is about to be cut, so a tree that fits is never copied here.
        let before = bound::measure_root(&tree);
        let too_large = if before.is_some_and(|bytes| bytes <= ROOT_BUDGET) {
            None
        } else {
            if raw.is_none() {
                raw = Some(tree.clone());
            }
            let bounded = bound::cut_to_fit(&mut tree, ROOT_BUDGET, before);
            tree.focused = !focused_below(&tree);
            Some(bounded)
        };
        if let Some(replaced) = mem::replace(&mut inner.last_raw, raw) {
            drop_iteratively(replaced);
        }

        // Compared again once cut: a tree that changed only past a cut, a
        // draft typed into beyond what fits, is cut to the snapshot already
        // sent.
        if inner.last_root.as_ref() == Some(&tree) {
            return;
        }
        inner.seq += 1;
        // Counted here, past the dedupe, so both count snapshots the agent
        // was sent rather than frames the app drew: an over-deep or
        // oversized tree that sits still on screen would otherwise count
        // sixty times a second.
        if let Some(too_deep) = too_deep {
            inner.truncated = inner.truncated.saturating_add(1);
            inner.last_truncation = Some(too_deep);
        }
        if let Some(bounded) = too_large {
            inner.oversized = inner.oversized.saturating_add(1);
            inner.last_oversize = Some(SnapshotTooLarge::new(bounded));
        }
        let snapshot = Snapshot::new(inner.seq, tree.clone());
        inner.last_root = Some(tree);

        let mut state = inner.shared.lock_state();
        state.latest = Some(snapshot);
        state.epoch += 1;
        drop(state);
        inner.shared.cv.notify_all();
    }

    /// The most recently published snapshot, if any. Test-only introspection.
    #[cfg(test)]
    pub(crate) fn latest_snapshot(&self) -> Option<Snapshot> {
        self.inner.as_ref()?.shared.lock_state().latest.clone()
    }
}

impl Drop for TariaLayer {
    /// Stop the threads and unlink the socket. A disabled layer has neither,
    /// so it touches no filesystem at all.
    ///
    /// Acks still queued for a connected bridge are written before its
    /// connection is closed, so an app that answers an input and exits in the
    /// same frame does not leave the agent waiting on that answer; a snapshot
    /// still waiting is not. See `SHUTDOWN_FLUSH` for why the one and not the
    /// other, and for the bound on it.
    ///
    /// How long this takes depends only on what the threads are doing, never
    /// on what is at the socket path. The listener polls the shutdown flag
    /// rather than waiting to be woken by a connection, so it returns whether
    /// the socket file was replaced by a second instance, removed, or is
    /// being served by some other process. With no bridge connected that is
    /// at most one `ACCEPT_POLL` nap. With one, the line being written when
    /// the shutdown came is finished or given up within `WRITE_TIMEOUT` of
    /// starting, however large it is and however the bridge reads, no other
    /// line but the queued acks is started after it, and those get
    /// `SHUTDOWN_FLUSH`. So dropping never takes longer than `ACCEPT_POLL` +
    /// `WRITE_TIMEOUT` + `SHUTDOWN_FLUSH`, about 5.3 s, and takes longer than
    /// one `ACCEPT_POLL` only against a bridge too slow to take the line it
    /// is being sent.
    ///
    /// The trees the layer still holds are freed without recursing, because
    /// the one the newest frame built can be deeper than the call stack.
    fn drop(&mut self) {
        let Some(inner) = self.inner.as_mut() else {
            return;
        };
        {
            let mut state = inner.shared.lock_state();
            state.shutdown = true;
            inner.shared.cv.notify_all();
        }
        if let Some(handle) = inner.listener.take() {
            let _ = handle.join();
        }
        remove_own_socket(&self.socket_path, inner.socket_ident);
    }
}

/// Unlink the socket file, but only while it is still the one this layer
/// bound.
///
/// A second instance that started later has a socket of its own at the same
/// path, and unlinking that on the way out would leave it running with no way
/// for a bridge to reach it (the mirror of the takeover `clear_socket_path`
/// refuses). Device and inode are what separate the file this layer created
/// from a file that merely has the same name.
fn remove_own_socket(path: &Path, bound: Option<(u64, u64)>) {
    if bound.is_some() && socket_ident(path) == bound {
        let _ = fs::remove_file(path);
    }
}

/// Identify the file at `path` by device and inode, so a later look can tell
/// the same file from a different one that took its place. `None` when it
/// cannot be read, which fails every comparison and so leaves the path alone.
fn socket_ident(path: &Path) -> Option<(u64, u64)> {
    let meta = fs::symlink_metadata(path).ok()?;
    Some((meta.dev(), meta.ino()))
}

/// State shared between the app thread and the connection threads.
struct Shared {
    app_label: String,
    state: Mutex<State>,
    cv: Condvar,
    /// Agent inputs dropped because the input queue was full. Written by
    /// reader threads, read via [`TariaLayer::dropped_inputs`].
    dropped_inputs: AtomicU64,
    /// Agent inputs discarded rather than delivered: at dequeue because the
    /// connection they arrived on had ended, and before the handshake because
    /// the client wrote them while it waited its turn. Written by the app
    /// thread for the first and by the listener thread for the second, both
    /// with an atomic add, so neither can lose the other's count; read via
    /// [`TariaLayer::stale_inputs`].
    stale_inputs: AtomicU64,
    /// Agent inputs answered [`Ignored`](InputStatus::Ignored) by the reader
    /// without reaching the app, because their `kind` did not parse. Written
    /// by reader threads, read via [`TariaLayer::unknown_inputs`].
    unknown_inputs: AtomicU64,
    /// Acks dropped because [`ACK_QUEUE`] answers were already waiting for a
    /// bridge that is not reading them fast enough. Written by whichever
    /// thread queues an ack, read via [`TariaLayer::dropped_acks`].
    dropped_acks: AtomicU64,
}

#[derive(Default)]
struct State {
    /// Latest published snapshot; sent to newly connected clients and, via
    /// `epoch` bumps, streamed to the current client.
    latest: Option<Snapshot>,
    /// Bumped on every publish so the writer knows something new exists.
    epoch: u64,
    /// Acks waiting to go out, oldest first. Queued for the writer rather
    /// than written where they are produced, so that one thread owns the
    /// stream and an ack can never overtake the snapshot published after it.
    /// Bounded at [`ACK_QUEUE`]; see [`Shared::push_ack`].
    acks: VecDeque<(InputId, InputStatus)>,
    /// Input ids handed to the app on `delivered_generation`, oldest first
    /// and capped at [`ACKABLE_INPUTS`]. [`TariaLayer::ack`] answers only an
    /// id in here, which is how a late ack for a bridge that is gone is told
    /// from one the live bridge is waiting on.
    delivered: VecDeque<InputId>,
    /// The generation `delivered` belongs to. Compared with `generation`
    /// rather than cleared wherever a connection ends, so no path that
    /// retires a generation can forget to clear it.
    delivered_generation: u64,
    /// Which bridge connection is live. Bumped when a connection starts and
    /// again when it ends, so the value between connections matches nothing.
    /// Every queued input carries the generation it arrived on, which is what
    /// lets the app tell an input the live bridge is waiting on from one that
    /// outlived its sender.
    generation: u64,
    shutdown: bool,
}

impl Shared {
    /// Lock the state, recovering from poisoning (a panicking connection
    /// thread must not take the app down with it).
    fn lock_state(&self) -> MutexGuard<'_, State> {
        self.state.lock().unwrap_or_else(PoisonError::into_inner)
    }

    fn is_shutdown(&self) -> bool {
        self.lock_state().shutdown
    }

    /// Queue an ack for the writer and wake it.
    ///
    /// For acks the layer produces itself, about an input the app never saw:
    /// a [`Dropped`](InputStatus::Dropped) for one the queue had no room for,
    /// an [`Ignored`](InputStatus::Ignored) for one this build cannot read.
    /// The app's own acks go through [`queue_app_ack`](Self::queue_app_ack),
    /// which first checks the input still belongs to the live connection.
    fn queue_ack(&self, id: InputId, status: InputStatus) {
        let mut state = self.lock_state();
        self.push_ack(&mut state, id, status);
        drop(state);
        self.cv.notify_all();
    }

    /// Queue an ack the app asked for, if the input it answers came from the
    /// connection that is live now.
    ///
    /// See [`TariaLayer::ack`] for why an id from a connection that has ended
    /// is dropped in silence rather than sent or reported.
    fn queue_app_ack(&self, id: InputId, status: InputStatus) {
        let mut state = self.lock_state();
        if !state.was_delivered(id) {
            return;
        }
        self.push_ack(&mut state, id, status);
        drop(state);
        self.cv.notify_all();
    }

    /// Append an ack, dropping the oldest one waiting when the queue is full.
    ///
    /// The caller holds the lock; waking the writer is the caller's job too,
    /// because [`deliver`](Self::deliver) has more to do under the same lock.
    /// See [`ACK_QUEUE`] for the bound and why it drops from this end.
    fn push_ack(&self, state: &mut State, id: InputId, status: InputStatus) {
        if state.acks.len() >= ACK_QUEUE {
            state.acks.pop_front();
            self.dropped_acks.fetch_add(1, Ordering::Relaxed);
        }
        state.acks.push_back((id, status));
    }

    /// Retire the generation of the connection being served, so anything
    /// still queued from it is discarded rather than delivered.
    ///
    /// [`reader_loop`] does this as it exits, which covers every connection
    /// that reached the point of having a reader; [`serve_client`] calls it
    /// for the ones that failed before that.
    fn retire_generation(&self) {
        self.lock_state().generation += 1;
    }

    /// Forget every queued ack. An ack answers the client that sent the
    /// input, so one outliving its connection would reach a client that never
    /// sent it. Ids do not repeat for the lifetime of a bridge process (see
    /// [`InputId`]), so the next client would not mistake it for its own, but
    /// there is nothing useful to do with it either.
    fn clear_acks(&self) {
        self.lock_state().acks.clear();
    }

    /// Answer an input the app just dequeued, or refuse it as stale.
    ///
    /// Returns true after queueing the [`Delivered`](InputStatus::Delivered)
    /// ack for an input whose connection is still live. Returns false for an
    /// input from a retired generation: the caller must discard it instead of
    /// acting on it, because it was aimed at a bridge session that no longer
    /// exists. A discard is counted (see [`TariaLayer::stale_inputs`]) and
    /// deliberately not acked, since the peer that would read the ack is the
    /// one that is gone.
    fn deliver(&self, generation: u64, id: InputId) -> bool {
        let mut state = self.lock_state();
        if state.generation != generation {
            drop(state);
            // An atomic add, not a load and store: the listener thread counts
            // its pre-handshake discards into the same counter (see
            // `start_connection`), possibly at this very moment.
            self.stale_inputs.fetch_add(1, Ordering::Relaxed);
            return false;
        }
        // Remembered before the ack so that an app which refines this ack
        // straight away, inside the same drain closure, is answering an id
        // the layer already knows it handed over.
        state.record_delivered(generation, id);
        self.push_ack(&mut state, id, InputStatus::Delivered);
        drop(state);
        self.cv.notify_all();
        true
    }
}

impl State {
    /// Remember `id` as one the app may answer again with
    /// [`TariaLayer::ack`], for as long as its connection lives.
    fn record_delivered(&mut self, generation: u64, id: InputId) {
        if self.delivered_generation != generation {
            self.delivered.clear();
            self.delivered_generation = generation;
        }
        if self.delivered.len() >= ACKABLE_INPUTS {
            self.delivered.pop_front();
        }
        self.delivered.push_back(id);
    }

    /// Was `id` handed to the app by the connection that is live now?
    ///
    /// False between connections and for anything the previous one delivered,
    /// which is what keeps a late ack from answering a fresh bridge's input
    /// that happens to carry the same id.
    fn was_delivered(&self, id: InputId) -> bool {
        self.delivered_generation == self.generation && self.delivered.contains(&id)
    }
}

/// Is any node below `root` focused?
///
/// Walks with a stack of sibling iterators rather than by recursing, for the
/// reason [`Node::check_depth`] does: it runs on the tree as built, before
/// the depth cut, and a recursive walk would overflow on a tree deeper than
/// the call stack. The stack holds one iterator per level, so a wide tree
/// costs no more memory than a narrow one.
fn focused_below(root: &Node) -> bool {
    let mut pending = vec![root.children.iter()];
    while let Some(siblings) = pending.last_mut() {
        match siblings.next() {
            Some(node) if node.focused => return true,
            Some(node) => pending.push(node.children.iter()),
            None => {
                pending.pop();
            }
        }
    }
    false
}

/// Are `a` and `b` the same tree?
///
/// What `==` on [`Node`] answers, without its recursion, because one side is
/// the tree as built and can be deeper than the call stack. Each pair of
/// nodes is compared with the children set aside, which still compares every
/// field a node has, the ones a later taria adds included, and the children
/// are then compared as pairs of their own. The `&mut` is only for setting
/// the children aside; both trees come back as they were.
fn same_tree(a: &mut Node, b: &mut Node) -> bool {
    if !same_node(a, b) {
        return false;
    }
    let mut pending = vec![a.children.iter_mut().zip(b.children.iter_mut())];
    while let Some(pairs) = pending.last_mut() {
        match pairs.next() {
            Some((a, b)) => {
                if !same_node(a, b) {
                    return false;
                }
                pending.push(a.children.iter_mut().zip(b.children.iter_mut()));
            }
            None => {
                pending.pop();
            }
        }
    }
    true
}

/// Are `a` and `b` equal apart from their children, and do they have as
/// many?
fn same_node(a: &mut Node, b: &mut Node) -> bool {
    if a.children.len() != b.children.len() {
        return false;
    }
    let a_children = mem::take(&mut a.children);
    let b_children = mem::take(&mut b.children);
    let same = *a == *b;
    a.children = a_children;
    b.children = b_children;
    same
}

/// Free a tree one node at a time.
///
/// The drop glue [`Node`] derives frees a node's children before the node,
/// by recursing, so a tree deeper than the call stack aborts the process
/// when it is dropped: not a panic that unwinds, a stack overflow. Every
/// tree this crate lets go of before the depth cut goes through here
/// instead. Each node is freed with its children moved out to a list of
/// their own, so a chain holds one node there at a time. A child with no
/// children of its own is freed where it stands instead, which recurses no
/// further, so a frame of fifty thousand rows drawn unchanged frees them
/// at the cost of the derived glue rather than queueing every one.
pub(crate) fn drop_iteratively(root: Node) {
    let mut pending = vec![root];
    while let Some(mut node) = pending.pop() {
        pending.extend(
            node.children
                .drain(..)
                .filter(|child| !child.children.is_empty()),
        );
    }
}

/// `node` without its children, copied without copying them.
fn shallow_copy(node: &mut Node) -> Node {
    let children = mem::take(&mut node.children);
    let copy = node.clone();
    node.children = children;
    copy
}

/// A copy of `root` cut to [`MAX_NODE_DEPTH`]: every node at the limit is
/// copied without its children, and nothing below it is copied at all.
///
/// A copy, because the tree as built is kept to compare the next frame
/// against; cut while it is copied, so the part cut away costs nothing. Built
/// with a stack rather than by recursing, because the tree it reads is the
/// one too deep for that. Only the branches that run over lose anything; the
/// rest of the tree keeps every node it had.
fn copy_within_depth(root: &mut Node) -> Node {
    // A copy under construction, beside the children of its original still
    // to copy into it. The stack's height is the level its top is at,
    // counting the root as 1.
    let mut stack = vec![(shallow_copy(root), root.children.iter_mut())];
    while let Some((_, children)) = stack.last_mut() {
        let next = children.next();
        let level = stack.len() + 1;
        match next {
            Some(child) if level >= MAX_NODE_DEPTH => {
                let copy = shallow_copy(child);
                if let Some((parent, _)) = stack.last_mut() {
                    parent.children.push(copy);
                }
            }
            Some(child) => {
                let copy = shallow_copy(child);
                stack.push((copy, child.children.iter_mut()));
            }
            // The root stays on the stack once its last child is in, so the
            // loop ends with it there.
            None if stack.len() == 1 => break,
            None => {
                if let Some((done, _)) = stack.pop()
                    && let Some((parent, _)) = stack.last_mut()
                {
                    parent.children.push(done);
                }
            }
        }
    }
    let copied = stack.pop().map(|(copy, _)| copy);
    drop(stack);
    // The loop never empties the stack, so the fallback is never taken; it
    // is the root alone, which is still a tree within the limit.
    copied.unwrap_or_else(|| shallow_copy(root))
}

/// Accept loop: serves one bridge client at a time until shutdown.
///
/// The listener is non-blocking, so this thread is never parked in `accept()`
/// and the shutdown flag is checked on its own schedule; see [`ACCEPT_POLL`]
/// for why exit cannot be left to depend on a connection arriving.
///
/// It also tracks whether the next connection it accepts waited behind the
/// last one, which is what decides if [`start_connection`] discards what that
/// client wrote before its hello. A connection accepted after a session
/// ended, before any accept has found the backlog empty, was waiting in the
/// backlog while that session ran; a client that connects in the instant
/// between the session's end and the next accept is taken for one too. One
/// accepted after an accept found nothing connected after the session had
/// ended, so it waited behind nobody, and neither did the first connection of
/// the process.
fn accept_loop(listener: UnixListener, shared: Arc<Shared>, input_tx: SyncSender<QueuedInput>) {
    let mut waited = false;
    loop {
        if shared.is_shutdown() {
            return;
        }
        match listener.accept() {
            Ok((stream, _addr)) => {
                if shared.is_shutdown() {
                    return;
                }
                // Both halves of a session are read and written as blocking
                // streams, and an accepted socket inherits the listener's
                // non-blocking flag on some platforms. Say it outright rather
                // than rely on which one this is.
                if stream.set_nonblocking(false).is_err() {
                    continue;
                }
                // Errors just drop this client; the loop keeps accepting.
                let _ = serve_client(stream, &shared, &input_tx, waited);
                // Whoever is in the backlog now connected while that client
                // was served, so it waited behind it, however that ended.
                waited = true;
            }
            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
                // Nobody is waiting, so whoever connects next waited behind
                // nobody.
                waited = false;
                // Nap instead of spinning, then back to the shutdown check at
                // the top.
                thread::sleep(ACCEPT_POLL);
            }
            Err(_) => {
                if shared.is_shutdown() {
                    return;
                }
                // Avoid a hot loop on a persistently failing listener.
                thread::sleep(ACCEPT_POLL);
            }
        }
    }
}

/// Serve one connected bridge: handshake, then reader thread + writer loop
/// until the client disconnects or the layer shuts down.
///
/// `waited` says the client was in the listen backlog when the previous
/// session ended; see [`accept_loop`] and [`start_connection`].
fn serve_client(
    stream: UnixStream,
    shared: &Arc<Shared>,
    input_tx: &SyncSender<QueuedInput>,
    waited: bool,
) -> io::Result<()> {
    // SO_SNDTIMEO is shared across the duplicated fds; only writes block long
    // enough to need it. Every line sets it again before each of its writes,
    // from its own deadline (see `write_all_before`), so after the first line
    // it stays at `WRITE_WAIT` or less, and this value governs no write: it
    // only means the socket never starts out with no timeout at all.
    stream.set_write_timeout(Some(WRITE_TIMEOUT))?;

    // Capture the snapshot and epoch atomically so the writer loop neither
    // misses nor duplicates a publish that races the handshake. Acks left
    // over from the previous connection go here: they name ids this client
    // never used. The generation taken here tags every input this connection
    // queues, so an input that outlives it is discarded rather than applied
    // to the app on behalf of a bridge that is gone.
    let (initial_snapshot, initial_epoch, generation) = {
        let mut state = shared.lock_state();
        state.acks.clear();
        state.generation += 1;
        (state.latest.clone(), state.epoch, state.generation)
    };
    let alive = Arc::new(AtomicBool::new(true));
    let (mut write_stream, reader) = match start_connection(
        stream,
        shared,
        input_tx,
        initial_snapshot,
        &alive,
        generation,
        waited,
    ) {
        Ok(started) => started,
        Err(err) => {
            // The reader thread normally retires this generation as it exits,
            // and here it never started (or never got the handshake). Retire
            // it by hand so the invariant holds through a failed connection
            // too: between connections the counter matches no queued input.
            shared.retire_generation();
            shared.clear_acks();
            return Err(err);
        }
    };

    writer_loop(&mut write_stream, shared, &alive, initial_epoch);

    // Unblock the reader (it is parked in a blocking read) and reap it before
    // going back to accept the next client.
    let _ = write_stream.shutdown(Shutdown::Both);
    let _ = reader.join();
    // Whatever is still queued can no longer be delivered, and means nothing
    // to the next client. Queued inputs need no sweep here: the reader
    // retired their generation, so the app discards them as it dequeues.
    shared.clear_acks();
    Ok(())
}

/// Discard what a client that waited its turn wrote before its handshake,
/// send the handshake (and the snapshot a new client is owed) and start the
/// connection's reader thread, so every way of failing to get a connection
/// off the ground funnels through one `Err` for the caller to clean up after.
/// On success the writer's handle on the socket comes back beside the reader
/// thread.
///
/// # Input written before the hello
///
/// The accept loop serves one client to completion before it accepts the
/// next, so a client that connects meanwhile sits in the listen backlog with
/// a socket that works, and whatever it writes there waits in the kernel's
/// buffer: a probe measured 180 KB accepted before a write would block, all
/// of it readable once the connection is finally served. Read as the new
/// session's first lines, that is input the app applies on behalf of a client
/// that wrote it blind, to a tree it never saw, and the hello is the boundary
/// that tells the two apart: nothing written before it can have been written
/// in reply to it. See [`TariaLayer::stale_inputs`], which counts what is
/// thrown away here.
///
/// Only a client that `waited` is drained: one the accept loop took from the
/// backlog after a session ended, before it last found the backlog empty
/// (see [`accept_loop`]). A client that connects to an idle app is accepted
/// within one [`ACCEPT_POLL`], and one that writes at once without reading
/// the hello, a script piping a line into the socket, waited behind nobody.
/// What it wrote is input for the session it is about to get, as it was in
/// 0.3.0, and discarding it would lose it unacked, with nothing to tell the
/// client. The protocol leaves the choice to the app; this keeps the discard
/// to the client it exists for.
///
/// For a client that waited, the stream is read without blocking, right
/// before the hello is written and not a step earlier, and everything already
/// there is discarded. What arrives between this read and the hello's write
/// slips through, and only a client writing blind can land in that gap. A
/// client that waits for the hello loses nothing either way.
///
/// The bytes may stop partway through a line, whose remainder then arrives
/// as the session's first bytes. That remainder belongs to a line written
/// blind, and its own tail can parse as a whole input (a line with leading
/// whitespace, cut inside it, is one), so the reader is told to skip through
/// the first newline it reads before it treats anything as input.
///
/// More than [`MAX_LINE_BYTES`] waiting is an error, like a line that long:
/// past that it is a flood, not a client that wrote a little too early, and
/// reading it all off would let a client that never stops writing hold the
/// listener here forever.
///
/// The latest snapshot is skipped, not written, once the layer is shutting
/// down, for the reason the writer skips one (see [`writer_loop`]).
fn start_connection(
    stream: UnixStream,
    shared: &Arc<Shared>,
    input_tx: &SyncSender<QueuedInput>,
    initial_snapshot: Option<Snapshot>,
    alive: &Arc<AtomicBool>,
    generation: u64,
    waited: bool,
) -> io::Result<(UnixStream, JoinHandle<()>)> {
    let mut write_stream = stream.try_clone()?;
    let starts_mid_line = if waited {
        discard_unsolicited(&stream, shared)?
    } else {
        false
    };

    write_line(
        &mut write_stream,
        &AppToBridge::hello(shared.app_label.clone(), PROTOCOL_VERSION),
    )?;
    if let Some(snapshot) = initial_snapshot
        && !shared.is_shutdown()
    {
        write_line(&mut write_stream, &AppToBridge::Snapshot(snapshot))?;
    }
    let reader = thread::Builder::new().name("taria-reader".into()).spawn({
        let alive = Arc::clone(alive);
        let shared = Arc::clone(shared);
        let input_tx = input_tx.clone();
        move || reader_loop(stream, input_tx, alive, shared, generation, starts_mid_line)
    })?;
    Ok((write_stream, reader))
}

/// Read off and count what a client that waited its turn wrote before its
/// hello, for [`start_connection`]. Returns whether it stopped partway through
/// a line, whose remainder the reader is then told to skip.
fn discard_unsolicited(stream: &UnixStream, shared: &Shared) -> io::Result<bool> {
    // The flag belongs to the open socket, not to this handle, so the
    // `try_clone`d writer goes non-blocking with it. It is put back before
    // anything is written: the hello and every later write rely on blocking
    // until their deadline.
    stream.set_nonblocking(true)?;
    let mut unsolicited = Vec::new();
    let read = read_unsolicited(stream, &mut unsolicited);
    // Counted whatever the read ended in: an input read off here was
    // discarded either way, including when the next byte overflows the cap.
    shared
        .stale_inputs
        .fetch_add(count_input_lines(&unsolicited), Ordering::Relaxed);
    stream.set_nonblocking(false)?;
    read?;
    Ok(unsolicited.last().is_some_and(|&byte| byte != b'\n'))
}

/// Append everything `reader` has ready to `pending`, stopping when a read
/// would block or the peer has closed its end.
///
/// For the non-blocking pass [`start_connection`] makes before the hello.
/// More than [`MAX_LINE_BYTES`] in all is an
/// [`InvalidData`](io::ErrorKind::InvalidData) error, with `pending` left
/// holding exactly the first [`MAX_LINE_BYTES`] bytes: the read that ran
/// over is kept up to the cap, so every complete line before it can still be
/// counted as an input thrown away. Any other error ends the read the same
/// way: the connection is broken, and `pending` holds what came before.
fn read_unsolicited(mut reader: impl Read, pending: &mut Vec<u8>) -> io::Result<()> {
    let mut chunk = [0u8; 8 * 1024];
    loop {
        match reader.read(&mut chunk) {
            Ok(0) => return Ok(()),
            Ok(n) => {
                let Some(read) = chunk.get(..n) else {
                    return Err(io::Error::other(
                        "a read reported more bytes than it was given",
                    ));
                };
                let room = MAX_LINE_BYTES.saturating_sub(pending.len());
                if n > room {
                    // Up to the cap, so a complete line that fits before it
                    // is counted even though the read it came in ran over.
                    pending.extend_from_slice(read.get(..room).unwrap_or_default());
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidData,
                        format!(
                            "the client wrote more than {MAX_LINE_BYTES} bytes before its \
                             handshake"
                        ),
                    ));
                }
                pending.extend_from_slice(read);
            }
            Err(err) if err.kind() == io::ErrorKind::Interrupted => {}
            Err(err) if err.kind() == io::ErrorKind::WouldBlock => return Ok(()),
            Err(err) => return Err(err),
        }
    }
}

/// How many complete lines of `bytes` are agent inputs, of any kind.
///
/// A line that does not end in `\n` is not complete, and neither is it
/// counted: whether it would have parsed is unknowable. Lines that do not
/// parse at all are passed over in silence, as the reader passes them over on
/// a live connection. An input of a kind this build cannot read
/// ([`AgentInput::Unknown`]) counts like any other, because it was discarded
/// like any other; on a live connection it would have been answered, but
/// before the hello there is nobody to answer.
fn count_input_lines(bytes: &[u8]) -> u64 {
    let inputs = bytes
        .split_inclusive(|&byte| byte == b'\n')
        .filter(|line| line.ends_with(b"\n"))
        .filter(|line| {
            matches!(
                serde_json::from_slice::<BridgeToApp>(line),
                Ok(BridgeToApp::Input { .. })
            )
        })
        .count();
    u64::try_from(inputs).unwrap_or(u64::MAX)
}

/// Reader half of a connection: parse `BridgeToApp` lines and forward agent
/// inputs to the app. Malformed lines are ignored. A line longer than
/// [`MAX_LINE_BYTES`] marks the connection broken (the loop exits and the
/// client is disconnected) instead of buffering it. When the input queue is
/// full the newest input is dropped, acked [`Dropped`](InputStatus::Dropped)
/// so the agent learns of it, and counted for the app (see
/// [`TariaLayer::dropped_inputs`]); a flood of inputs can never block this
/// thread.
///
/// Every forwarded input carries `generation`, which this loop retires when
/// the connection ends so anything still queued from it is discarded instead
/// of delivered.
///
/// `starts_mid_line` says the bytes the client wrote before its hello, which
/// [`start_connection`] discarded, stopped partway through a line. Everything
/// up to and including the first newline is that line's remainder, and is
/// skipped without being parsed, counted or acked; see [`start_connection`].
/// It is still a line, so a remainder over [`MAX_LINE_BYTES`] breaks the
/// connection like any other.
fn reader_loop(
    stream: UnixStream,
    input_tx: SyncSender<QueuedInput>,
    alive: Arc<AtomicBool>,
    shared: Arc<Shared>,
    generation: u64,
    mut starts_mid_line: bool,
) {
    let mut reader = BufReader::new(stream);
    let mut buf = Vec::new();
    loop {
        buf.clear();
        // Read at most one byte past the cap: a line that fits ends in `\n`
        // within the limit; anything longer is an oversized frame.
        let read = (&mut reader)
            .take((MAX_LINE_BYTES + 1) as u64)
            .read_until(b'\n', &mut buf);
        let n = match read {
            Ok(n) => n,
            Err(_) => break,
        };
        if n == 0 {
            break; // EOF: the client closed the connection.
        }
        if buf.last() != Some(&b'\n') {
            // Oversized line (or EOF mid-line): treat the connection as
            // broken rather than accumulating an unbounded buffer.
            break;
        }
        if std::mem::take(&mut starts_mid_line) {
            // The rest of a line the client began before its hello. Its tail
            // can parse on its own, and it was still written blind.
            continue;
        }
        let Ok(msg) = serde_json::from_slice::<BridgeToApp>(&buf) else {
            continue;
        };
        let BridgeToApp::Input { id, input, .. } = msg else {
            // A message variant added to the protocol after this adapter was
            // written. Skipped like a line that failed to parse, and for the
            // same reason: it asks for something this build cannot do. It
            // cannot be acked either, because an ack answers an input id and
            // this message is not an input.
            continue;
        };
        if matches!(input, AgentInput::Unknown) {
            // An input whose `kind` this build cannot read, which the parse
            // kept only because the id lives on the message around it.
            // Answered here rather than handed to the app: the variant
            // carries nothing to act on, so every app would need an arm for a
            // value that is unactionable by construction. Delivering it would
            // also ack `Delivered` at the dequeue, a claim the app took the
            // input, and leave the true answer to whether each app author
            // remembers to refine it. `Ignored` is both true and already
            // known here, so the agent learns in one round trip.
            shared.unknown_inputs.fetch_add(1, Ordering::Relaxed);
            shared.queue_ack(id, InputStatus::Ignored);
            continue;
        }
        let queued = QueuedInput {
            generation,
            id,
            input,
        };
        match input_tx.try_send(queued) {
            Ok(()) => {}
            Err(TrySendError::Full(_)) => {
                shared.dropped_inputs.fetch_add(1, Ordering::Relaxed);
                shared.queue_ack(id, InputStatus::Dropped);
            }
            Err(TrySendError::Disconnected(_)) => break,
        }
    }
    let mut state = shared.lock_state();
    // Retire this connection's generation before anyone can observe the
    // disconnect: an input still queued from it must already be stale by the
    // time the peer is gone, because no ack could reach that peer any more.
    // Only one connection is served at a time and `serve_client` joins this
    // thread before accepting the next, so this always retires the generation
    // that just ended.
    state.generation += 1;
    alive.store(false, Ordering::SeqCst);
    // Notify while holding the lock so the writer cannot miss the wakeup
    // between checking `alive` and parking on the condvar.
    shared.cv.notify_all();
    drop(state);
}

/// Writer half of a connection: send queued acks, then each newly published
/// snapshot, as lines. Acks go first in every pass, so an ack always reaches
/// the client before the snapshot published after it; an agent that saw the
/// snapshot first could not tell whether it reflects its input yet.
///
/// Returns when the layer shuts down, the reader reports the client gone, or
/// a write fails. A shutdown first writes every ack still queued, the rest of
/// the batch in hand included, within [`SHUTDOWN_FLUSH`] in all, and skips the
/// snapshot that would have followed them; a client already gone gets
/// nothing, since nobody is left to read it.
///
/// The shutdown flag is checked before every line, the snapshot's included,
/// so a shutdown that comes while a line is being written waits for that line
/// only, and [`WRITE_TIMEOUT`] bounds the whole of it. That is the bound
/// [`Drop`] promises, because it waits for this thread.
///
/// Every ack that flush can write belongs to this connection. The queue is
/// emptied under the lock as each connection starts, in the same step that
/// moves the generation on, so nothing queued for an earlier one survives
/// into it. While this connection's reader runs, `alive` is true and the
/// generation is this connection's, and the reader changes both together,
/// under the lock, as it exits, so an `alive` read under the lock vouches
/// for every ack taken in the same hold. Every shutdown check takes the acks
/// in the same hold that read `alive`.
fn writer_loop(stream: &mut UnixStream, shared: &Shared, alive: &AtomicBool, mut last_epoch: u64) {
    loop {
        let (acks, snapshot) = {
            let mut state = shared.lock_state();
            loop {
                if !alive.load(Ordering::SeqCst) {
                    return;
                }
                if state.shutdown {
                    let queued = std::mem::take(&mut state.acks);
                    drop(state);
                    flush_acks(stream, queued);
                    return;
                }
                if !state.acks.is_empty() || state.epoch != last_epoch {
                    let acks = std::mem::take(&mut state.acks);
                    let snapshot = if state.epoch == last_epoch {
                        None
                    } else {
                        last_epoch = state.epoch;
                        state.latest.clone()
                    };
                    break (acks, snapshot);
                }
                state = shared
                    .cv
                    .wait(state)
                    .unwrap_or_else(PoisonError::into_inner);
            }
        };
        let mut acks = acks.into_iter();
        while let Some((id, status)) = acks.next() {
            // Rechecked per ack rather than once per pass: a full batch
            // against a peer that stopped reading is a whole queue of write
            // timeouts long, and `Drop` waits for this thread, so an exit
            // during the batch would sit through every one of them. The acks
            // not written yet go out under the flush's bound instead, this
            // one first and anything queued since the batch was taken last.
            let unwritten = std::iter::once((id, status)).chain(&mut acks);
            if stop_writing(stream, shared, alive, unwritten) {
                return;
            }
            if write_line(stream, &AppToBridge::ack(id, status)).is_err() {
                return;
            }
        }
        let Some(snapshot) = snapshot else {
            continue;
        };
        // Checked again here, or a shutdown that came while the batch's last
        // ack was being written would be noticed only after the snapshot:
        // the one line a shutdown skips, and the longest.
        if stop_writing(stream, shared, alive, std::iter::empty()) {
            return;
        }
        if write_line(stream, &AppToBridge::Snapshot(snapshot)).is_err() {
            return;
        }
    }
}

/// The check [`writer_loop`] makes before each line of a batch: is the
/// client gone, or is the layer shutting down?
///
/// Returns true when the writer must stop: at once when the client is gone,
/// and when the layer is shutting down, once `unwritten` and then every ack
/// queued since the batch was taken have been flushed. `unwritten` is
/// consumed only in that case, so the caller can go on with it otherwise. The
/// queued acks are taken in the same hold of the lock that read `alive`,
/// which is what vouches for them; see [`writer_loop`].
fn stop_writing(
    stream: &mut UnixStream,
    shared: &Shared,
    alive: &AtomicBool,
    unwritten: impl IntoIterator<Item = (InputId, InputStatus)>,
) -> bool {
    let mut state = shared.lock_state();
    if !alive.load(Ordering::SeqCst) {
        return true;
    }
    if !state.shutdown {
        return false;
    }
    let queued = std::mem::take(&mut state.acks);
    drop(state);
    flush_acks(stream, unwritten.into_iter().chain(queued));
    true
}

/// Serialize one message as a newline-terminated JSON frame and write it,
/// the whole line within [`WRITE_TIMEOUT`] of starting or not at all.
///
/// The deadline is on the line, not on each `write` it takes; see
/// [`WRITE_TIMEOUT`] for why. A line that misses it may be written in part,
/// and every caller then drops the connection, which is the only thing that
/// can follow a partial line.
fn write_line(stream: &mut UnixStream, msg: &AppToBridge) -> io::Result<()> {
    let mut line = serde_json::to_vec(msg).map_err(io::Error::other)?;
    line.push(b'\n');
    let deadline = Instant::now()
        .checked_add(WRITE_TIMEOUT)
        .ok_or_else(|| io::Error::other("the write deadline is past what the clock can hold"))?;
    write_all_before(stream, &line, deadline)
}

/// Write the acks a shutdown found queued, oldest first, giving up once
/// [`SHUTDOWN_FLUSH`] has passed or a write fails.
///
/// Best effort by design: the caller returns either way, and an ack that
/// misses the deadline costs its waiter the timeout it would have waited out
/// with no flush at all. The connection is closed straight after, so the
/// write timeout this leaves on the socket is never used again.
fn flush_acks(stream: &mut UnixStream, acks: impl IntoIterator<Item = (InputId, InputStatus)>) {
    let Some(deadline) = Instant::now().checked_add(SHUTDOWN_FLUSH) else {
        return;
    };
    // One buffer and one deadline for the lot, rather than a line at a time,
    // so the bound is on the flush and not on each of its writes.
    let mut lines = Vec::new();
    for (id, status) in acks {
        let Ok(line) = serde_json::to_vec(&AppToBridge::ack(id, status)) else {
            break;
        };
        lines.extend_from_slice(&line);
        lines.push(b'\n');
    }
    let _ = write_all_before(stream, &lines, deadline);
}

/// `write_all`, finished by `deadline` or failed with
/// [`TimedOut`](io::ErrorKind::TimedOut).
///
/// A socket's write timeout does not bound a `write` call, only each wait for
/// room inside one, and a call waits again for every piece the kernel splits
/// it into (Linux allocates a buffer of at most a few tens of KB at a time,
/// and waits for room before each). So a large line, against a bridge that
/// frees a little room before every wait can time out, never times out at
/// all: a 900 KB snapshot to a bridge taking 64 KB every 3 s held the writer
/// for forty seconds with a 5 s timeout. Setting the timeout to what is left
/// of a deadline before each call does not help on its own, because a single
/// call runs past the deadline.
///
/// So no call is handed more than [`WRITE_CHUNK`] bytes, few enough that the
/// kernel waits for room at most once inside it, and that wait is given what
/// is left of the deadline, or [`WRITE_WAIT`] if that is less, set before
/// every call. A wait that times out short of the deadline is simply waited
/// again. A timeout of zero means none at all to the socket and is refused,
/// which is why nothing left ends the loop rather than reaching it.
fn write_all_before(
    stream: &mut UnixStream,
    mut bytes: &[u8],
    deadline: Instant,
) -> io::Result<()> {
    while !bytes.is_empty() {
        let left = deadline.saturating_duration_since(Instant::now());
        if left.is_zero() {
            return Err(io::ErrorKind::TimedOut.into());
        }
        stream.set_write_timeout(Some(left.min(WRITE_WAIT)))?;
        let chunk = bytes.get(..WRITE_CHUNK).unwrap_or(bytes);
        match stream.write(chunk) {
            Ok(0) => return Err(io::ErrorKind::WriteZero.into()),
            Ok(n) => bytes = bytes.get(n..).unwrap_or_default(),
            Err(err) if err.kind() == io::ErrorKind::Interrupted => {}
            // The socket is blocking, so this is the wait's timeout firing,
            // and the clock at the top of the loop says whether that was the
            // deadline.
            Err(err)
                if matches!(
                    err.kind(),
                    io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
                ) => {}
            Err(err) => return Err(err),
        }
    }
    Ok(())
}

/// Make `path` absolute, resolving a relative path (including a bare
/// filename) against the current directory. Ensures the socket path always
/// has a real parent directory for [`ensure_private_dir`] to vet.
fn absolutize(path: PathBuf) -> io::Result<PathBuf> {
    if path.is_absolute() {
        Ok(path)
    } else {
        Ok(env::current_dir()?.join(path))
    }
}

/// Make `path` free for `bind`, or say why it cannot be.
///
/// Binding needs the path to be absent, so a socket left behind by a run that
/// was killed before it could clean up has to be unlinked first. That unlink
/// has to be earned. The path can come verbatim from `$TARIA_SOCK`, and a
/// private directory passes the vetting whatever else is in it (`~/.ssh` is
/// 0700 and user-owned), so removing whatever happens to be there would let a
/// typo delete a file the user meant to keep. A socket another instance is
/// still serving must survive too: unlinking it and binding a fresh one in its
/// place leaves that instance running, believing it is reachable, while every
/// bridge connects to the newcomer instead.
///
/// So: only a socket is ever removed, and only once a connection to it has
/// been refused.
fn clear_socket_path(path: &Path) -> io::Result<()> {
    // symlink_metadata so a symlink is seen as itself: following one would
    // decide about a socket somewhere else and unlink the link.
    let meta = match fs::symlink_metadata(path) {
        Ok(meta) => meta,
        Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(()),
        Err(err) => {
            return Err(io::Error::new(
                err.kind(),
                format!("cannot inspect {}: {err}", path.display()),
            ));
        }
    };
    if !meta.file_type().is_socket() {
        return Err(io::Error::new(
            io::ErrorKind::AlreadyExists,
            format!(
                "{} already exists and is not a socket (it is {}); refusing to delete it to \
                 bind there. Point $TARIA_SOCK at a path taria may create.",
                path.display(),
                describe_file_type(&meta)
            ),
        ));
    }
    match probe_socket(path) {
        // Nothing is listening: the socket outlived whatever bound it.
        SocketProbe::Stale => match fs::remove_file(path) {
            Ok(()) => Ok(()),
            // Someone else cleaned it up in the meantime, which is the state
            // this call wanted anyway.
            Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
            Err(err) => Err(io::Error::new(
                err.kind(),
                format!("cannot remove the stale socket {}: {err}", path.display()),
            )),
        },
        SocketProbe::Live => Err(io::Error::new(
            io::ErrorKind::AddrInUse,
            format!(
                "another instance is already serving the taria socket {}; refusing to take it \
                 over, because that would leave it running with no way for a bridge to reach \
                 it. To run a second instance alongside it, give it a socket of its own: set \
                 TARIA_SOCK to a different path.",
                path.display()
            ),
        )),
        SocketProbe::Unknown(why) => Err(io::Error::new(
            io::ErrorKind::AddrInUse,
            format!(
                "cannot tell whether another instance is serving the taria socket {} ({why}); \
                 refusing to remove a socket that may still be in use. Set TARIA_SOCK to a \
                 different path to bind elsewhere.",
                path.display()
            ),
        )),
    }
}

/// What is at a socket path that is not a socket, for the error that refuses
/// to delete it.
fn describe_file_type(meta: &fs::Metadata) -> &'static str {
    let kind = meta.file_type();
    if kind.is_symlink() {
        "a symlink"
    } else if kind.is_dir() {
        "a directory"
    } else if kind.is_file() {
        "a regular file"
    } else {
        "another kind of file"
    }
}

/// The answer to "is anything serving the socket at this path?".
enum SocketProbe {
    /// The connection was accepted: something is listening.
    Live,
    /// The connection was refused, or the file went away: nothing is.
    Stale,
    /// The probe could not answer, carrying the reason for the error that
    /// reports it.
    Unknown(String),
}

/// Try to connect to the socket at `path`, briefly, to find out whether
/// anything is listening on it.
///
/// The connect runs on a thread of its own because it blocks with no timeout;
/// see [`SOCKET_PROBE_TIMEOUT`]. An answer that does not arrive in time is
/// [`Unknown`](SocketProbe::Unknown), never a guess: the whole point is to
/// avoid unlinking a socket that is still being served.
fn probe_socket(path: &Path) -> SocketProbe {
    let (tx, rx) = mpsc::channel();
    let probe_path = path.to_path_buf();
    let spawned = thread::Builder::new()
        .name("taria-socket-probe".into())
        .spawn(move || {
            let outcome = UnixStream::connect(&probe_path).map(|stream| {
                // Nothing is ever sent on it. A peer serving this path sees a
                // connection that opens and closes, which is what it does
                // with any client that goes away.
                let _ = stream.shutdown(Shutdown::Both);
            });
            // The receiver is gone once the wait above times out; that is the
            // `Unknown` case, and there is nothing to report to.
            let _ = tx.send(outcome);
        });
    if let Err(err) = spawned {
        return SocketProbe::Unknown(format!("the probe thread could not start: {err}"));
    }
    match rx.recv_timeout(SOCKET_PROBE_TIMEOUT) {
        Ok(Ok(())) => SocketProbe::Live,
        // Refused is what a socket with no listener answers; a socket that
        // vanished between the look and the connect leaves the path free too.
        Ok(Err(err))
            if matches!(
                err.kind(),
                io::ErrorKind::ConnectionRefused | io::ErrorKind::NotFound
            ) =>
        {
            SocketProbe::Stale
        }
        Ok(Err(err)) => SocketProbe::Unknown(format!("connecting to it failed: {err}")),
        Err(_) => SocketProbe::Unknown(format!(
            "connecting to it did not finish within {SOCKET_PROBE_TIMEOUT:?}"
        )),
    }
}

/// Create (if needed) and vet the directory that will hold the socket.
///
/// The socket's parent directory decides who can replace or redirect the
/// socket: if it is a symlink, owned by another user, or accessible to
/// group/others, a local attacker can swap the socket for their own and
/// impersonate the app (or intercept the bridge). The directory is created
/// with mode `0700`, then verified via `symlink_metadata`: it must be a
/// real directory (not a symlink), owned by the current user, with no
/// group/other permission bits. Any violation is an error.
fn ensure_private_dir(dir: &Path) -> io::Result<()> {
    let mut builder = fs::DirBuilder::new();
    builder.recursive(true);
    builder.mode(0o700);
    // An existing directory is fine here; whether it is *acceptable* is
    // decided by the checks below, which also surface real create failures.
    let _ = builder.create(dir);

    // symlink_metadata so a planted symlink is seen as itself, not followed.
    let meta = fs::symlink_metadata(dir).map_err(|err| {
        io::Error::new(
            err.kind(),
            format!("socket directory {} is unusable: {err}", dir.display()),
        )
    })?;
    if !meta.file_type().is_dir() {
        return Err(io::Error::other(format!(
            "socket directory {} is not a real directory (it may be a symlink planted by \
             another user to hijack the socket); refusing to use it",
            dir.display()
        )));
    }
    let uid = current_uid()?;
    if meta.uid() != uid {
        return Err(io::Error::other(format!(
            "socket directory {} is owned by uid {} instead of the current user (uid {}); \
             its owner could replace the socket to hijack the connection; refusing to use it",
            dir.display(),
            meta.uid(),
            uid
        )));
    }
    if meta.mode() & 0o077 != 0 {
        return Err(io::Error::other(format!(
            "socket directory {} is group/world-accessible (mode {:03o}); other users could \
             replace the socket to hijack the connection; re-create it with mode 0700",
            dir.display(),
            meta.mode() & 0o777
        )));
    }
    Ok(())
}

/// The current effective uid, without `unsafe` or extra dependencies.
///
/// On Linux, `/proc/self` is owned by this process's effective uid. On other
/// Unixes, fall back to creating a probe file in the system temp dir: a file
/// this process creates is owned by its effective uid.
fn current_uid() -> io::Result<u32> {
    if let Ok(meta) = fs::metadata("/proc/self") {
        return Ok(meta.uid());
    }
    let probe = env::temp_dir().join(format!("taria-uid-probe-{}", std::process::id()));
    let _ = fs::remove_file(&probe);
    let file = fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .mode(0o600)
        .open(&probe)?;
    let uid = file.metadata()?.uid();
    drop(file);
    let _ = fs::remove_file(&probe);
    Ok(uid)
}

/// Resolve the default socket path for `app_label` from the environment.
///
/// The rule itself lives in [`taria::socket::resolve_path`], shared with the
/// bridge so the two sides cannot look for the socket in different places,
/// including its refusal of a label that is not a plain file name. Reading the
/// environment stays here, where the process actually is.
fn resolve_socket_path(app_label: &str) -> Result<PathBuf, taria::socket::InvalidAppLabel> {
    taria::socket::resolve_path(
        env::var_os("TARIA_SOCK"),
        env::var_os("XDG_RUNTIME_DIR"),
        &env::temp_dir(),
        &user_identity(),
        app_label,
    )
}

/// A refused app label as the `io::Error` every bind path already reports.
///
/// `InvalidInput` because the label is one: the same kind
/// [`bind_at`](TariaLayer::bind_at) gives an over-long path, and the message
/// is the label error's own, which names the label and what a label may be.
fn invalid_label(err: taria::socket::InvalidAppLabel) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidInput, err.to_string())
}

/// Uid where available (via `/proc/self` on Linux), else `$USER`/`$LOGNAME`,
/// else a fixed fallback. Only used to namespace the temp-dir fallback path.
fn user_identity() -> String {
    if let Ok(meta) = fs::metadata("/proc/self") {
        return meta.uid().to_string();
    }
    env::var("USER")
        .or_else(|_| env::var("LOGNAME"))
        .unwrap_or_else(|_| "default".to_string())
}

#[cfg(test)]
mod tests {
    use std::os::unix::fs::PermissionsExt;

    use super::*;
    use crate::test_util::bind_test_layer;

    /// Unique, absent scratch path under the system temp dir.
    fn scratch_path(name: &str) -> PathBuf {
        let path = env::temp_dir().join(format!("taria-dirtest-{}-{name}", std::process::id()));
        let _ = fs::remove_dir_all(&path);
        let _ = fs::remove_file(&path);
        path
    }

    /// A private (0700) temp dir that removes itself when the test ends, so
    /// nothing is left in the system temp dir even on panic.
    fn private_temp_dir(prefix: &str) -> tempfile::TempDir {
        tempfile::Builder::new()
            .prefix(prefix)
            .permissions(fs::Permissions::from_mode(0o700))
            .tempdir()
            .expect("create test dir")
    }

    /// The shared state a connection's threads work on, without a socket, for
    /// tests that drive those parts directly.
    fn test_shared(app_label: &str) -> Arc<Shared> {
        Arc::new(Shared {
            app_label: app_label.to_string(),
            state: Mutex::new(State::default()),
            cv: Condvar::new(),
            dropped_inputs: AtomicU64::new(0),
            stale_inputs: AtomicU64::new(0),
            unknown_inputs: AtomicU64::new(0),
            dropped_acks: AtomicU64::new(0),
        })
    }

    #[test]
    fn ensure_private_dir_accepts_fresh_0700_dir() {
        let dir = scratch_path("fresh");
        ensure_private_dir(&dir).expect("fresh private dir should be accepted");
        let meta = fs::symlink_metadata(&dir).unwrap();
        assert!(meta.file_type().is_dir());
        assert_eq!(meta.mode() & 0o077, 0, "mode: {:03o}", meta.mode() & 0o777);
        // And it stays acceptable on a second call (existing private dir).
        ensure_private_dir(&dir).expect("existing private dir should be accepted");
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn ensure_private_dir_rejects_symlinked_dir() {
        let target = scratch_path("symlink-target");
        fs::create_dir_all(&target).unwrap();
        fs::set_permissions(&target, fs::Permissions::from_mode(0o700)).unwrap();
        let link = scratch_path("symlink-link");
        std::os::unix::fs::symlink(&target, &link).unwrap();

        let err = ensure_private_dir(&link).expect_err("symlinked dir must be rejected");
        assert!(err.to_string().contains("symlink"), "err: {err}");

        let _ = fs::remove_file(&link);
        let _ = fs::remove_dir_all(&target);
    }

    #[test]
    fn ensure_private_dir_rejects_world_accessible_dir() {
        let dir = scratch_path("world-writable");
        fs::create_dir_all(&dir).unwrap();
        fs::set_permissions(&dir, fs::Permissions::from_mode(0o777)).unwrap();

        let err = ensure_private_dir(&dir).expect_err("0777 dir must be rejected");
        assert!(err.to_string().contains("group/world"), "err: {err}");

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn absolutize_joins_relative_paths_onto_cwd() {
        let cwd = env::current_dir().unwrap();

        let bare = absolutize(PathBuf::from("bare.sock")).unwrap();
        assert!(bare.is_absolute());
        assert_eq!(bare, cwd.join("bare.sock"));
        // The parent is now the (vetable) cwd, not the empty path.
        assert_eq!(bare.parent(), Some(cwd.as_path()));

        let nested = absolutize(PathBuf::from("sub/dir/app.sock")).unwrap();
        assert_eq!(nested, cwd.join("sub/dir/app.sock"));

        let absolute = absolutize(PathBuf::from("/already/abs.sock")).unwrap();
        assert_eq!(absolute, PathBuf::from("/already/abs.sock"));
    }

    /// A bare filename must not bypass the parent-directory privacy check:
    /// it resolves against the cwd, and binding succeeds only when the cwd
    /// itself passes [`ensure_private_dir`]. Tested without touching
    /// `set_current_dir` (racy across parallel tests) by comparing against
    /// vetting the cwd directly.
    #[test]
    fn bind_at_bare_filename_vets_cwd() {
        let cwd = env::current_dir().unwrap();
        let name = format!("taria-bare-vet-{}.sock", std::process::id());
        let result = TariaLayer::bind_at("bare-demo", &name);

        match ensure_private_dir(&cwd) {
            Ok(()) => {
                let layer = result.expect("private cwd: bare filename should bind in it");
                assert_eq!(layer.socket_path(), cwd.join(&name));
                drop(layer); // removes the socket file
            }
            Err(_) => {
                assert!(
                    result.is_err(),
                    "bare filename in an unvetted cwd must be refused, not bound"
                );
                assert!(
                    !cwd.join(&name).exists(),
                    "no socket may be created in an unvetted cwd"
                );
            }
        }
    }

    /// Over the kernel's limit, `bind` alone would say only
    /// `InvalidInput: path must be shorter than SUN_LEN`. The named error has
    /// to survive the trip through `io::Error`.
    #[test]
    fn bind_at_refuses_an_over_long_socket_path() {
        let dir = private_temp_dir("taria-longpath-");
        let path = dir.path().join(format!("{}.sock", "x".repeat(120)));

        let Err(err) = TariaLayer::bind_at("toolong", &path) else {
            panic!("over-long path must be refused, not bound");
        };
        let message = err.to_string();
        assert!(
            message.contains(&format!("{} byte", taria::socket::MAX_SOCKET_PATH_BYTES)),
            "err: {message}"
        );
        assert!(message.contains("TARIA_SOCK"), "err: {message}");
        assert!(!path.exists(), "no socket file may be left behind");
    }

    /// The label names the socket file, and this layer binds and unlinks what
    /// the resolution returns. A label that is a path instead of a file name
    /// must therefore be refused before any of that, on both entry points:
    /// `bind` cannot resolve one, and `bind_or_disabled`, which never fails,
    /// has to come back disabled rather than pointed somewhere else.
    #[test]
    fn a_label_that_is_not_a_file_name_binds_nothing() {
        for label in ["/etc/cron.d/evil", "../../../tmp/pwn", "sub/dir"] {
            let Err(err) = TariaLayer::bind(label) else {
                panic!("bind accepted the label {label:?}");
            };
            assert_eq!(err.kind(), io::ErrorKind::InvalidInput, "label: {label}");
            assert!(err.to_string().contains(label), "err: {err}");

            let layer = TariaLayer::bind_or_disabled(label);
            assert!(!layer.is_enabled(), "label: {label}");
            assert_eq!(layer.app_label(), label, "the label is reported as given");
            assert_eq!(
                layer.socket_path(),
                Path::new(""),
                "a refused label resolves to no path"
            );
            let message = layer
                .bind_error()
                .expect("a disabled layer says why")
                .to_string();
            assert!(message.contains(label), "err: {message}");
        }
    }

    /// The whole point of a disabled layer: every method still answers, and
    /// none of them does anything.
    #[test]
    fn disabled_layer_is_inert() {
        let dir = private_temp_dir("taria-disabled-");
        // A regular file where the socket's parent directory should be: the
        // bind cannot succeed, and nothing about that is timing-dependent.
        let blocker = dir.path().join("not-a-directory");
        fs::write(&blocker, b"").expect("create blocker file");
        let socket_path = blocker.join("app.sock");

        let mut layer = TariaLayer::bind_or_disabled_at("disabled", socket_path.clone());

        assert!(!layer.is_enabled());
        assert!(layer.bind_error().is_some(), "a disabled layer says why");
        assert_eq!(layer.socket_path(), socket_path);
        assert_eq!(layer.app_label(), "disabled");
        assert_eq!(layer.dropped_inputs(), 0);
        assert_eq!(layer.stale_inputs(), 0);
        assert_eq!(layer.unknown_inputs(), 0);
        assert_eq!(layer.truncated_snapshots(), 0);
        assert_eq!(layer.last_truncation(), None);
        assert_eq!(layer.oversized_snapshots(), 0);
        assert_eq!(layer.last_oversize(), None);

        assert_eq!(layer.try_recv(), None);
        assert_eq!(layer.try_recv_with_id(), None);
        assert_eq!(layer.recv_timeout(Duration::from_millis(1)), None);
        assert_eq!(layer.recv_timeout_with_id(Duration::from_millis(1)), None);
        layer.ack(0, InputStatus::Ignored);

        let mut drained = 0;
        layer.drain(|_| drained += 1);
        layer.drain_with_ids(|_, _| drained += 1);
        layer.drain_acking(|_| {
            drained += 1;
            InputStatus::Ignored
        });
        assert_eq!(drained, 0, "a disabled layer has nothing to drain");

        // The render path works unchanged; it just publishes nowhere.
        let mut rec = layer.frame();
        rec.push(Node::new("pane", Role::Pane));
        rec.publish();
        layer.publish([Node::new("pane", Role::Pane)]);
        assert_eq!(layer.latest_snapshot(), None);

        drop(layer);
        assert!(!socket_path.exists(), "a disabled layer creates no socket");
        assert!(blocker.exists(), "dropping it must unlink nothing");
    }

    /// A chain of `depth` nodes, one child each: `n1` down to `n{depth}`.
    fn chain(depth: usize) -> Node {
        let mut node = Node::new(format!("n{depth}"), Role::TreeItem);
        for level in (1..depth).rev() {
            node = Node::new(format!("n{level}"), Role::TreeItem).child(node);
        }
        node
    }

    /// The auto root counts towards the depth, so a chain of
    /// `MAX_NODE_DEPTH - 1` is exactly at the limit and nothing is cut.
    /// Checked on its own, because it is the boundary the cut must not creep
    /// past into trees that were always fine.
    #[test]
    fn a_tree_at_the_depth_limit_is_published_untouched() {
        let mut layer = bind_test_layer("taria-layer-", "atlimit");
        layer.publish([chain(MAX_NODE_DEPTH - 1)]);

        let root = layer
            .latest_snapshot()
            .expect("a tree at the limit publishes")
            .root;
        assert_eq!(root.depth(), MAX_NODE_DEPTH);
        assert!(root.check_depth().is_ok());
        assert_eq!(layer.truncated_snapshots(), 0);
        assert_eq!(layer.last_truncation(), None);
    }

    /// The failure this enforcement exists for: past the limit a snapshot
    /// does not arrive at the peer at all, and nothing anywhere says so. So
    /// the tree still goes out, cut to fit, rather than being held back (the
    /// agent would keep reading a stale tree, which is the same failure) or
    /// sent whole (which is the failure itself).
    #[test]
    fn a_tree_over_the_depth_limit_is_cut_rather_than_dropped() {
        let mut layer = bind_test_layer("taria-layer-", "deep");
        assert!(layer.latest_snapshot().is_none(), "nothing published yet");

        // One node over the limit once the auto root is counted.
        layer.publish([chain(MAX_NODE_DEPTH)]);
        let snapshot = layer
            .latest_snapshot()
            .expect("a tree over the limit is published cut, never withheld");
        assert_eq!(snapshot.seq, 1);
        assert_eq!(snapshot.root.depth(), MAX_NODE_DEPTH);
        assert!(snapshot.root.check_depth().is_ok());

        // Everything above the cut is untouched: only the deepest node lost
        // its children.
        let mut node = &snapshot.root;
        for level in 1..MAX_NODE_DEPTH {
            assert_eq!(node.children.len(), 1, "level {level} lost a sibling");
            node = &node.children[0];
        }
        assert_eq!(node.id.0, format!("n{}", MAX_NODE_DEPTH - 1));
        assert!(node.children.is_empty(), "the node at the limit is a leaf");

        // And the app can find out, with the branch named.
        assert_eq!(layer.truncated_snapshots(), 1);
        let too_deep = layer.last_truncation().expect("the cut is recorded");
        assert_eq!(too_deep.depth(), MAX_NODE_DEPTH + 1);
        assert_eq!(too_deep.deepest().0, format!("n{MAX_NODE_DEPTH}"));
    }

    /// Focus is decided after the cut, not before: a snapshot has to carry
    /// focus somewhere, and the only focused node can be in the part removed.
    #[test]
    fn a_cut_that_removes_the_only_focused_node_leaves_focus_on_the_root() {
        let mut layer = bind_test_layer("taria-layer-", "deepfocus");
        let mut deep = Node::new("leaf", Role::TreeItem).focused(true);
        for level in (1..MAX_NODE_DEPTH).rev() {
            deep = Node::new(format!("n{level}"), Role::TreeItem).child(deep);
        }
        layer.publish([deep]);

        let root = layer.latest_snapshot().expect("published").root;
        assert_eq!(layer.truncated_snapshots(), 1);
        assert!(
            root.focused,
            "the focused node was cut, so the root has to carry focus"
        );
        assert!(!subtree_has_focus_below_root(&root));
    }

    /// Focus anywhere under the root, which is what the root's own focus flag
    /// is the negation of.
    fn subtree_has_focus_below_root(root: &Node) -> bool {
        focused_below(root)
    }

    /// The tree as built is compared, searched for focus and copied before
    /// the depth cut, so none of the three may recurse: the tree the cut
    /// exists for can be deeper than the call stack.
    #[test]
    fn a_tree_deeper_than_the_call_stack_is_compared_and_cut_without_recursing() {
        const DEEP: usize = 100_000;
        let mut deep = chain(DEEP);
        let mut same = chain(DEEP);
        // Different only at the very bottom, so the comparison has to walk
        // the whole way down to see it.
        let mut other = Node::new(format!("n{DEEP}"), Role::TreeItem).label("changed");
        for level in (1..DEEP).rev() {
            other = Node::new(format!("n{level}"), Role::TreeItem).child(other);
        }

        assert!(!focused_below(&deep));
        assert!(same_tree(&mut deep, &mut same));
        assert!(!same_tree(&mut deep, &mut other));

        let cut = copy_within_depth(&mut deep);
        assert_eq!(cut.depth(), MAX_NODE_DEPTH);
        assert!(cut.check_depth().is_ok());
        assert_eq!(deep.depth(), DEEP, "the tree as built is left whole");
        for tree in [deep, same, other] {
            drop_iteratively(tree);
        }
    }

    /// Run `test` on a thread with a stack far smaller than a recursive drop
    /// of a 100k-deep tree needs, so a recursion anywhere in it overflows
    /// however large the test thread's own stack happens to be. A stack
    /// overflow aborts the process rather than failing the test, so this is
    /// what makes such a regression fail every time instead of now and then.
    fn on_a_small_stack(test: impl FnOnce() + Send + 'static) {
        let outcome = thread::Builder::new()
            .name("small-stack".into())
            .stack_size(512 * 1024)
            .spawn(test)
            .expect("spawn a test thread")
            .join();
        if let Err(panic) = outcome {
            std::panic::resume_unwind(panic);
        }
    }

    /// The derived drop glue of [`Node`] recurses, so every tree the layer
    /// lets go of before the depth cut is freed one node at a time: the
    /// frame drawn unchanged, the tree as built that a changed frame
    /// replaces, the one the layer holds when it is dropped, and whatever a
    /// disabled layer or an unpublished frame is handed. Publishing a tree
    /// deeper than the call stack succeeded before this, and the next
    /// publish aborted the process.
    #[test]
    fn a_tree_deeper_than_the_call_stack_is_freed_without_recursing() {
        const DEEP: usize = 100_000;
        on_a_small_stack(|| {
            let mut layer = bind_test_layer("taria-layer-", "deepfree");
            layer.publish([chain(DEEP)]);
            // Drawn again unchanged: the frame's own tree is freed.
            layer.publish([chain(DEEP)]);
            assert_eq!(layer.latest_snapshot().expect("published").seq, 1);
            // A different one, which the depth cut makes the snapshot already
            // sent: the tree as built that it replaces is freed all the same.
            layer.publish([chain(DEEP + 1)]);
            assert_eq!(layer.latest_snapshot().expect("published").seq, 1);
            // And one that is a new snapshot.
            layer.publish([chain(DEEP), Node::new("new", Role::Text)]);
            assert_eq!(layer.latest_snapshot().expect("published").seq, 2);
            assert_eq!(layer.truncated_snapshots(), 2);
            // A frame recorded and never published is freed with it.
            layer.frame().push(chain(DEEP));
            // And the layer frees the one it still holds.
            drop(layer);

            let mut disabled =
                TariaLayer::disabled("deepfree", PathBuf::new(), io::Error::other("test"));
            disabled.publish([chain(DEEP)]);
        });
    }

    /// Only the branches that run over are cut. A wide tree with one runaway
    /// branch keeps everything else, which is what makes cutting worth more
    /// than refusing the publish.
    #[test]
    fn only_the_branch_that_runs_over_is_cut() {
        let mut layer = bind_test_layer("taria-layer-", "deepwide");
        layer.publish([
            Node::new("shallow", Role::Text).label("kept"),
            chain(MAX_NODE_DEPTH * 4),
        ]);

        let root = layer.latest_snapshot().expect("published").root;
        assert_eq!(root.depth(), MAX_NODE_DEPTH);
        assert_eq!(root.children.len(), 2);
        assert_eq!(root.children[0].id.0, "shallow");
        assert_eq!(root.children[0].label.as_deref(), Some("kept"));
        assert_eq!(layer.truncated_snapshots(), 1);
    }

    /// The counter counts snapshots the agent was sent, not frames the app
    /// drew. A deep tree redrawn unchanged is the snapshot already published
    /// once it is cut, so it is not sent again and must not count again: at
    /// sixty frames a second it would otherwise add sixty a second.
    #[test]
    fn the_same_over_deep_tree_published_twice_counts_once() {
        let mut layer = bind_test_layer("taria-layer-", "deeptwice");
        layer.publish([chain(MAX_NODE_DEPTH * 2)]);
        layer.publish([chain(MAX_NODE_DEPTH * 2)]);
        assert_eq!(layer.latest_snapshot().expect("published").seq, 1);
        assert_eq!(layer.truncated_snapshots(), 1);

        // A deep tree that changed is a new snapshot, cut again.
        layer.publish([chain(MAX_NODE_DEPTH * 2), Node::new("new", Role::Text)]);
        assert_eq!(layer.latest_snapshot().expect("published").seq, 2);
        assert_eq!(layer.truncated_snapshots(), 2);
    }

    /// One focused text input holding `value`, as a draft would be published.
    fn draft(value: String) -> Node {
        Node::new("draft", Role::TextInput)
            .value(value)
            .focused(true)
    }

    /// The root the layer builds around [`draft`], for sizing values against
    /// and for the line it would have been sent as.
    fn root_around_draft(label: &str, value: String) -> Node {
        Node::new("app", Role::App).label(label).child(draft(value))
    }

    fn line_len(snapshot: Snapshot) -> usize {
        serde_json::to_vec(&AppToBridge::Snapshot(snapshot))
            .expect("serialize a snapshot")
            .len()
    }

    /// The size boundary, through the layer: a root exactly at the budget is
    /// published as built and counted nowhere; one byte more is cut, counted,
    /// and still fits the line.
    #[test]
    fn a_tree_at_the_size_budget_is_published_untouched_and_one_byte_over_is_cut() {
        let mut layer = bind_test_layer("taria-layer-", "atsize");
        let base = serde_json::to_vec(&root_around_draft("atsize", String::new()))
            .expect("serialize a root")
            .len();
        let at = "x".repeat(ROOT_BUDGET - base);

        layer.publish([draft(at.clone())]);
        let snapshot = layer
            .latest_snapshot()
            .expect("a tree at the budget publishes");
        assert_eq!(snapshot.root, root_around_draft("atsize", at.clone()));
        assert!(line_len(snapshot) <= MAX_LINE_BYTES);
        assert_eq!(layer.oversized_snapshots(), 0);
        assert_eq!(layer.last_oversize(), None);

        let over = format!("{at}x");
        layer.publish([draft(over.clone())]);
        let snapshot = layer
            .latest_snapshot()
            .expect("a tree over is published cut");
        assert_eq!(snapshot.seq, 2);
        let value = snapshot.root.children[0].value.clone().expect("a value");
        assert!(value.ends_with('\u{2026}'), "the cut is marked");
        assert!(over.starts_with(value.trim_end_matches('\u{2026}')));
        assert!(snapshot.root.children[0].focused, "focus is untouched");
        assert!(line_len(snapshot) <= MAX_LINE_BYTES);

        assert_eq!(layer.oversized_snapshots(), 1);
        let too_large = layer.last_oversize().expect("the cut is recorded");
        // The line it would have been, counted at the largest seq the way
        // the bound counts it: one byte over the root budget is one byte over
        // the line.
        let unbounded = Snapshot::new(u64::MAX, root_around_draft("atsize", over));
        assert_eq!(too_large.bytes(), line_len(unbounded));
        assert_eq!(too_large.bytes(), MAX_LINE_BYTES + 1);
        assert_eq!(too_large.limit(), MAX_LINE_BYTES);
        assert_eq!(too_large.strings_trimmed(), 1);
        assert_eq!(too_large.nodes_dropped(), 0);
        assert_eq!(too_large.largest().0, "draft");
        let message = too_large.to_string();
        assert!(message.contains("`draft`"), "{message}");
        assert!(message.contains(&MAX_LINE_BYTES.to_string()), "{message}");
    }

    /// Counted like the depth cut: an oversized draft that sits still while
    /// the app redraws it is one snapshot, and an edit to it is another.
    #[test]
    fn the_same_oversized_tree_published_twice_counts_once() {
        let mut layer = bind_test_layer("taria-layer-", "bigtwice");
        let big = "y".repeat(MAX_LINE_BYTES * 2);
        layer.publish([draft(big.clone())]);
        layer.publish([draft(big.clone())]);
        assert_eq!(layer.latest_snapshot().expect("published").seq, 1);
        assert_eq!(layer.oversized_snapshots(), 1);

        // An edit past the cut changes nothing the agent would be sent, so
        // it is not a new snapshot either.
        layer.publish([draft(format!("{big}z"))]);
        assert_eq!(layer.latest_snapshot().expect("published").seq, 1);
        assert_eq!(layer.oversized_snapshots(), 1);

        // One the agent would see is.
        layer.publish([draft(format!("z{big}"))]);
        assert_eq!(layer.latest_snapshot().expect("published").seq, 2);
        assert_eq!(layer.oversized_snapshots(), 2);
    }

    /// Focus is decided after the bound, as it is after the depth cut: the
    /// rows dropped to fit can include the focused one, and a snapshot has
    /// to carry focus somewhere.
    #[test]
    fn a_size_bound_that_drops_the_only_focused_node_leaves_focus_on_the_root() {
        let mut layer = bind_test_layer("taria-layer-", "bigfocus");
        // The focused row is the last, and a list keeps a run of its first
        // rows, so it is the first row to go.
        let rows = (0..60_000).map(|n| {
            Node::new(format!("row-{n:05}"), Role::ListItem)
                .label(format!("Task {n:05}"))
                .focused(n == 59_999)
        });
        layer.publish([Node::new("list", Role::List).children(rows)]);

        let root = layer.latest_snapshot().expect("published").root;
        assert_eq!(layer.oversized_snapshots(), 1);
        let too_large = layer.last_oversize().expect("recorded");
        assert!(too_large.nodes_dropped() > 0, "the last row went");
        assert!(
            root.focused,
            "the focused row was dropped, so the root has to carry focus"
        );
        assert!(!subtree_has_focus_below_root(&root));
    }

    /// Through the layer, the line a node that could never fit leaves: the
    /// draft beside it goes out whole, in the line it would have had alone
    /// plus the chrome. Dropped only after every string was cut to the
    /// floor, it left a line of 590 bytes with 256 of them the draft.
    #[test]
    fn a_node_that_could_never_fit_leaves_the_draft_beside_it_whole() {
        let mut layer = bind_test_layer("taria-layer-", "unfit");
        let value = "d".repeat(300_000);
        let chrome = [
            Node::new("status", Role::Status).value("1 draft"),
            Node::new("quit", Role::Button)
                .label("Quit")
                .action(taria::Action::Activate),
        ];
        let mut nodes = vec![
            Node::new("I".repeat(2_000_000), Role::Text),
            draft(value.clone()),
        ];
        nodes.extend(chrome.clone());
        layer.publish(nodes);

        let snapshot = layer.latest_snapshot().expect("published");
        let ids: Vec<&str> = snapshot
            .root
            .children
            .iter()
            .map(|child| child.id.0.as_str())
            .collect();
        assert_eq!(ids, ["draft", "status", "quit"]);
        assert_eq!(snapshot.root.children[0], draft(value.clone()));
        assert_eq!(snapshot.root.children[1..], chrome);

        let alone = line_len(Snapshot::new(1, root_around_draft("unfit", value)));
        let line = line_len(snapshot);
        assert!(line > alone && line <= MAX_LINE_BYTES, "{line} of {alone}");
        let too_large = layer.last_oversize().expect("recorded");
        assert_eq!(too_large.strings_trimmed(), 0);
        assert_eq!(too_large.nodes_dropped(), 1);
    }

    /// The cost the comparison before the bound exists for. Deduplicating
    /// after the bound meant a tree too large for the line paid the whole
    /// bound on every frame it was drawn, changed or not, which for a few
    /// megabytes is a hundred milliseconds a frame. Unchanged, a frame is now
    /// compared and nothing else.
    #[test]
    fn an_unchanged_oversized_tree_is_bounded_once_however_often_it_is_drawn() {
        let mut layer = bind_test_layer("taria-layer-", "bigstill");
        let rows: Vec<Node> = (0..5000)
            .map(|n| Node::new(format!("row-{n:05}"), Role::ListItem).label(format!("Task {n:05}")))
            .collect();
        let frame = |text: &str| {
            [
                draft(text.to_string()),
                Node::new("list", Role::List).children(rows.clone()),
            ]
        };
        let big = "d".repeat(MAX_LINE_BYTES * 3);

        let start = bound::runs();
        let nodes = frame(&big);
        let started = Instant::now();
        layer.publish(nodes);
        let first = started.elapsed();
        let once = bound::runs();
        assert_eq!(once.measured - start.measured, 1);
        assert_eq!(once.cut - start.cut, 1);

        let mut redrawn = Duration::ZERO;
        for _ in 1..100 {
            // Built outside the clock: what is timed is the layer's work.
            let nodes = frame(&big);
            let started = Instant::now();
            layer.publish(nodes);
            redrawn += started.elapsed();
        }
        assert_eq!(
            bound::runs(),
            once,
            "an unchanged frame is neither measured nor cut"
        );
        assert_eq!(layer.latest_snapshot().expect("published").seq, 1);
        assert_eq!(layer.oversized_snapshots(), 1);
        let per_frame = redrawn / 99;
        eprintln!("first frame {first:?}, each unchanged frame after it {per_frame:?}");
        assert!(
            per_frame * 4 < first,
            "an unchanged frame took {per_frame:?}, against {first:?} for the one bounded"
        );

        // An edit past the cut is cut once and sends nothing, and the frames
        // that redraw it unchanged after that are not cut again.
        let edited = format!("{big}z");
        for _ in 0..10 {
            layer.publish(frame(&edited));
        }
        let after = bound::runs();
        assert_eq!(after.cut - once.cut, 1);
        assert_eq!(layer.latest_snapshot().expect("published").seq, 1);
        assert_eq!(layer.oversized_snapshots(), 1);
    }

    /// The comparison costs a tree that fits nothing extra: it is measured
    /// once, never cut, and held once, as the snapshot it went out as. Only a
    /// tree that had to be cut keeps the tree it was cut from as well.
    #[test]
    fn a_tree_that_fits_is_measured_once_and_held_once() {
        let mut layer = bind_test_layer("taria-layer-", "fitstill");
        let frame = || {
            [Node::new("list", Role::List)
                .children((0..100).map(|n| Node::new(format!("row-{n}"), Role::ListItem)))]
        };
        let start = bound::runs();
        for _ in 0..10 {
            layer.publish(frame());
        }
        let runs = bound::runs();
        assert_eq!(runs.measured - start.measured, 1);
        assert_eq!(runs.cut, start.cut);
        let inner = layer.inner.as_ref().expect("enabled");
        assert!(
            inner.last_raw.is_none(),
            "no second copy of a tree that fits"
        );
        assert_eq!(layer.latest_snapshot().expect("published").seq, 1);

        layer.publish([draft("y".repeat(MAX_LINE_BYTES * 2))]);
        let inner = layer.inner.as_ref().expect("enabled");
        assert!(
            inner.last_raw.is_some(),
            "a cut tree keeps what it was cut from"
        );
    }

    /// A deep tree whose runaway branch is the last child at every level: 40
    /// levels, cut to 32 for depth and then to the line for size. Any one
    /// cap on children that keeps the branch keeps everything, so a cap
    /// alone kept the first level's leaves and a few percent of the line.
    /// What the cap dropped is offered back, shallowest first, so the line
    /// is used.
    #[test]
    fn a_deep_tree_cut_for_depth_and_then_for_size_still_fills_the_line() {
        let mut layer = bind_test_layer("taria-layer-", "deepfill");
        let leaves = |level: usize| {
            (0..200).map(move |n| {
                Node::new(format!("leaf-{level}-{n}"), Role::Text).label("x".repeat(250))
            })
        };
        // 39 links under the root, so 40 levels counting it.
        let mut branch = Node::new("n39", Role::TreeItem).children(leaves(39));
        for level in (1..39).rev() {
            branch = Node::new(format!("n{level}"), Role::TreeItem)
                .children(leaves(level))
                .child(branch);
        }
        layer.publish([branch]);

        let snapshot = layer.latest_snapshot().expect("published");
        assert_eq!(layer.truncated_snapshots(), 1);
        assert_eq!(layer.oversized_snapshots(), 1);
        assert_eq!(
            snapshot.root.children[0].children.len(),
            201,
            "the first level is kept whole, branch and all"
        );
        let len = line_len(snapshot);
        assert!(len <= MAX_LINE_BYTES);
        assert!(
            len * 10 >= MAX_LINE_BYTES * 9,
            "{len} of {MAX_LINE_BYTES} bytes used"
        );
    }

    #[test]
    fn publish_wraps_nodes_like_the_recorder_does() {
        let mut layer = bind_test_layer("taria-layer-", "publish");
        layer.publish([
            Node::new("a", Role::Text),
            Node::new("b", Role::Button).focused(true),
        ]);

        let root = layer.latest_snapshot().expect("published").root;
        assert_eq!(root.id.0, "app");
        assert_eq!(root.label.as_deref(), Some("publish"));
        assert!(!root.focused, "a focused node unfocuses the auto root");
        let ids: Vec<&str> = root.children.iter().map(|c| c.id.0.as_str()).collect();
        assert_eq!(ids, ["a", "b"]);
    }

    /// The ordering [`writer_loop`] promises, exercised where it actually
    /// bites: an ack and a newer snapshot pending in the *same* pass. Driving
    /// the loop directly over a socket pair is the only way to make that
    /// overlap certain; through a live connection the writer wakes on the
    /// first of the two and usually drains it before the second exists, so
    /// the socket tests only ever see the two in separate passes.
    #[test]
    fn writer_sends_pending_acks_before_a_pending_snapshot() {
        let (server, client) = UnixStream::pair().expect("socket pair");
        client
            .set_read_timeout(Some(Duration::from_secs(5)))
            .expect("read timeout");

        let shared = test_shared("ackorder");
        {
            // Both are waiting before the loop runs, so it has to choose an
            // order: the app acked an input and published a new tree in
            // response to it.
            let mut state = shared.lock_state();
            state.acks.push_back((7, InputStatus::Delivered));
            state.latest = Some(Snapshot::new(1, Node::new("app", Role::App)));
            state.epoch = 1;
        }

        let alive = Arc::new(AtomicBool::new(true));
        let writer = thread::spawn({
            let shared = Arc::clone(&shared);
            let alive = Arc::clone(&alive);
            move || {
                let mut server = server;
                writer_loop(&mut server, &shared, &alive, 0);
            }
        });

        let mut reader = BufReader::new(client);
        let mut read_message = || {
            let mut line = String::new();
            let n = reader.read_line(&mut line).expect("read a message");
            assert!(n > 0, "writer closed the stream");
            serde_json::from_str::<AppToBridge>(&line).expect("parse a message")
        };
        assert_eq!(
            read_message(),
            AppToBridge::ack(7, InputStatus::Delivered),
            "an agent that saw the snapshot first could not tell whether it \
             already reflects its input"
        );
        let AppToBridge::Snapshot(snapshot) = read_message() else {
            panic!("expected the snapshot after the ack");
        };
        assert_eq!(snapshot.seq, 1);

        shared.lock_state().shutdown = true;
        shared.cv.notify_all();
        writer.join().expect("writer thread");
    }

    /// Write junk into `stream` until the peer's side is full, so the next
    /// write blocks: a bridge that has stopped reading. Returns how many
    /// bytes it took, for a test that wants to read past them later.
    fn fill_until_blocked(stream: &UnixStream) -> usize {
        stream.set_nonblocking(true).expect("non-blocking");
        let chunk = [b'x'; 4096];
        let mut filled = 0;
        loop {
            match (&*stream).write(&chunk) {
                Ok(n) => filled += n,
                Err(err) if err.kind() == io::ErrorKind::WouldBlock => break,
                Err(err) => panic!("filling the socket failed: {err}"),
            }
        }
        stream.set_nonblocking(false).expect("blocking again");
        filled
    }

    /// Run [`writer_loop`] on its own thread and report when it returns, so
    /// a writer that never does fails the test instead of hanging the suite.
    /// The stream is dropped as the loop returns, which is the EOF a test
    /// reads to know nothing else was written.
    fn spawn_writer(server: UnixStream, shared: &Arc<Shared>) -> Receiver<()> {
        let (tx, rx) = mpsc::channel();
        thread::spawn({
            let shared = Arc::clone(shared);
            move || {
                let mut server = server;
                let alive = AtomicBool::new(true);
                writer_loop(&mut server, &shared, &alive, 0);
                drop(server);
                let _ = tx.send(());
            }
        });
        rx
    }

    /// The exit the flush exists for: the app dequeued an input, refined its
    /// ack and dropped the layer, all before the writer ran again. Both acks
    /// are still queued, and so is the snapshot the app published last.
    #[test]
    fn a_shutdown_writes_the_queued_acks_and_not_the_snapshot() {
        let (server, client) = UnixStream::pair().expect("socket pair");
        client
            .set_read_timeout(Some(Duration::from_secs(5)))
            .expect("read timeout");

        let shared = test_shared("exitflush");
        {
            let mut state = shared.lock_state();
            state.acks.push_back((4, InputStatus::Delivered));
            state.acks.push_back((4, InputStatus::Ignored));
            state.latest = Some(Snapshot::new(1, Node::new("app", Role::App)));
            state.epoch = 1;
            state.shutdown = true;
        }
        let done = spawn_writer(server, &shared);

        let mut reader = BufReader::new(client);
        let mut lines = Vec::new();
        loop {
            let mut line = String::new();
            if reader.read_line(&mut line).expect("read a message") == 0 {
                break;
            }
            lines.push(serde_json::from_str::<AppToBridge>(&line).expect("parse a message"));
        }
        assert_eq!(
            lines,
            [
                AppToBridge::ack(4, InputStatus::Delivered),
                AppToBridge::ack(4, InputStatus::Ignored),
            ],
            "every queued ack goes out, in order, and the snapshot behind them does not"
        );
        assert!(
            done.recv_timeout(Duration::from_secs(5)).is_ok(),
            "the writer returns once the acks are out"
        );
        assert!(shared.lock_state().acks.is_empty(), "the queue was taken");
    }

    /// The other place the writer notices a shutdown: partway through a
    /// batch, blocked on a slow bridge. What it has not written of that batch
    /// goes out under the flush's bound, followed by anything queued since,
    /// and the snapshot that was due after the batch is skipped.
    #[test]
    fn a_shutdown_mid_batch_writes_the_rest_of_it_and_what_queued_since() {
        let (server, client) = UnixStream::pair().expect("socket pair");
        client
            .set_read_timeout(Some(Duration::from_secs(5)))
            .expect("read timeout");
        let junk = fill_until_blocked(&server);

        let shared = test_shared("midbatch");
        {
            let mut state = shared.lock_state();
            state.acks.push_back((1, InputStatus::Delivered));
            state.acks.push_back((2, InputStatus::Delivered));
            state.latest = Some(Snapshot::new(1, Node::new("app", Role::App)));
            state.epoch = 1;
        }
        let done = spawn_writer(server, &shared);

        // The writer has the batch once the queue is empty. From there it is
        // blocked writing the first ack into the full socket, or about to be,
        // and either way the next time it looks at the shutdown flag is
        // inside that batch, before one of its acks.
        let deadline = Instant::now() + Duration::from_secs(5);
        while !shared.lock_state().acks.is_empty() {
            assert!(Instant::now() < deadline, "the writer never took the batch");
            thread::sleep(Duration::from_millis(1));
        }
        {
            let mut state = shared.lock_state();
            state.acks.push_back((3, InputStatus::Ignored));
            state.shutdown = true;
            shared.cv.notify_all();
        }

        // Reading frees the socket, which lets the blocked write finish and
        // the writer reach its next check.
        let mut reader = BufReader::new(client);
        let mut skipped = vec![0; junk];
        reader.read_exact(&mut skipped).expect("read past the junk");
        let mut lines = Vec::new();
        loop {
            let mut line = String::new();
            if reader.read_line(&mut line).expect("read a message") == 0 {
                break;
            }
            lines.push(serde_json::from_str::<AppToBridge>(&line).expect("parse a message"));
        }
        assert_eq!(
            lines,
            [
                AppToBridge::ack(1, InputStatus::Delivered),
                AppToBridge::ack(2, InputStatus::Delivered),
                AppToBridge::ack(3, InputStatus::Ignored),
            ],
            "the batch in hand first, then what was queued behind it, and no snapshot"
        );
        assert!(done.recv_timeout(Duration::from_secs(5)).is_ok());
    }

    /// The gap the per-ack check leaves: a shutdown that comes while the
    /// batch's last ack is being written. That ack's check has passed, and
    /// there is no ack after it whose check could notice, so the writer used
    /// to go straight on to the snapshot, the one line a shutdown skips and
    /// the one that can hold `Drop` for a whole `WRITE_TIMEOUT`.
    #[test]
    fn a_shutdown_while_the_last_ack_is_written_never_writes_the_snapshot() {
        let (server, client) = UnixStream::pair().expect("socket pair");
        client
            .set_read_timeout(Some(Duration::from_secs(5)))
            .expect("read timeout");
        let junk = fill_until_blocked(&server);

        let shared = test_shared("lastack");
        {
            let mut state = shared.lock_state();
            state.acks.push_back((1, InputStatus::Delivered));
            state.latest = Some(Snapshot::new(1, Node::new("app", Role::App)));
            state.epoch = 1;
        }
        let done = spawn_writer(server, &shared);

        let deadline = Instant::now() + Duration::from_secs(5);
        while !shared.lock_state().acks.is_empty() {
            assert!(Instant::now() < deadline, "the writer never took the batch");
            thread::sleep(Duration::from_millis(1));
        }
        // Time for the writer to pass the check before its only ack and
        // stall in that ack's write, against the full socket. Should it still
        // be short of the check, that check sees the shutdown instead, and
        // what is asserted below holds all the same.
        thread::sleep(Duration::from_millis(100));
        {
            let mut state = shared.lock_state();
            state.shutdown = true;
            shared.cv.notify_all();
        }

        // Reading frees the socket, which lets the stalled ack finish.
        let mut reader = BufReader::new(client);
        let mut skipped = vec![0; junk];
        reader.read_exact(&mut skipped).expect("read past the junk");
        let mut lines = Vec::new();
        loop {
            let mut line = String::new();
            if reader.read_line(&mut line).expect("read a message") == 0 {
                break;
            }
            lines.push(serde_json::from_str::<AppToBridge>(&line).expect("parse a message"));
        }
        assert_eq!(
            lines,
            [AppToBridge::ack(1, InputStatus::Delivered)],
            "the ack in hand goes out, and the snapshot behind it does not"
        );
        assert!(done.recv_timeout(Duration::from_secs(5)).is_ok());
    }

    /// The flush is bounded in total, not per ack: against a bridge that has
    /// stopped reading, a queue of acks each given the time left would still
    /// be one bound long, and a queue each given `WRITE_TIMEOUT` would hold
    /// the app's exit for minutes.
    #[test]
    fn the_shutdown_flush_gives_up_on_a_bridge_that_stopped_reading() {
        let (server, client) = UnixStream::pair().expect("socket pair");
        fill_until_blocked(&server);
        // As on a live connection, where every write already has this
        // timeout: the flush has to beat it, not lean on it.
        server
            .set_write_timeout(Some(WRITE_TIMEOUT))
            .expect("write timeout");

        let shared = test_shared("stuckflush");
        {
            let mut state = shared.lock_state();
            for id in 0..20 {
                state.acks.push_back((id, InputStatus::Delivered));
            }
            state.shutdown = true;
        }
        let done = spawn_writer(server, &shared);

        // Well short of one `WRITE_TIMEOUT`, so a flush that leaned on it
        // fails here rather than passing slowly.
        let margin = Duration::from_secs(2);
        done.recv_timeout(SHUTDOWN_FLUSH + margin)
            .expect("the flush must give up on a stuck bridge within its bound");
        // Held to here, so the bridge was stuck rather than gone throughout.
        drop(client);
    }

    /// A reader that hands out a scripted sequence of results, for the read
    /// a socket can only produce by timing: an interrupted call, then a
    /// would-block once everything waiting has been read.
    struct Scripted(VecDeque<io::Result<&'static [u8]>>);

    impl Read for Scripted {
        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
            let bytes = self
                .0
                .pop_front()
                .expect("read again after it would have blocked")?;
            buf[..bytes.len()].copy_from_slice(bytes);
            Ok(bytes.len())
        }
    }

    /// What the pre-handshake pass reads and what it counts: every complete
    /// line that is an input, of any kind, and nothing else.
    #[test]
    fn input_written_before_the_hello_is_read_off_and_counted_by_the_line() {
        let script = [
            Ok(&br#"{"type":"input","id":1,"input":{"kind":"key","key":"q"}}"#[..]),
            Err(io::ErrorKind::Interrupted.into()),
            Ok(&b"\nnot json\n"[..]),
            // A kind this build cannot read is still an input, discarded
            // like the rest.
            Ok(&br#"{"type":"input","id":2,"input":{"kind":"paste"}}"#[..]),
            Ok(&b"\n{\"type\":\"bye\"}\n"[..]),
            // Cut off: the reader skips its remainder rather than this
            // counting it.
            Ok(&br#"{"type":"input","id":3,"#[..]),
            Err(io::ErrorKind::WouldBlock.into()),
        ];
        let mut pending = Vec::new();
        read_unsolicited(Scripted(script.into()), &mut pending)
            .expect("an interrupted read is retried, and a would-block ends the pass");
        assert!(
            pending.ends_with(br#""id":3,"#),
            "everything ready was read"
        );
        assert_eq!(count_input_lines(&pending), 2);

        // A peer that closed its end is done too.
        let mut pending = Vec::new();
        read_unsolicited(&b"{}\n"[..], &mut pending).expect("EOF ends the pass");
        assert_eq!(pending, b"{}\n");

        // Up to a line's worth waiting is fine; past it is an oversized line.
        let line = vec![b' '; MAX_LINE_BYTES];
        read_unsolicited(line.as_slice(), &mut Vec::new()).expect("exactly the cap is accepted");
        let flood = vec![b'\n'; MAX_LINE_BYTES + 1];
        let mut pending = Vec::new();
        let err = read_unsolicited(flood.as_slice(), &mut pending)
            .expect_err("more than a line's worth before the hello is refused");
        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
        assert!(
            pending.len() <= MAX_LINE_BYTES,
            "and nothing past the cap is kept"
        );
    }

    /// The read that runs over the cap is kept up to it, so every complete
    /// line before the cap is counted. Dropping that whole read undercounted
    /// by as many lines as it held whenever the reads did not land on the cap
    /// exactly, which a socket's reads need not.
    #[test]
    fn every_complete_line_before_the_cap_is_counted() {
        let line = b"{\"type\":\"input\",\"id\":1,\"input\":{\"kind\":\"key\",\"key\":\"q\"}}\n";
        let fit = MAX_LINE_BYTES / line.len();
        assert_ne!(
            MAX_LINE_BYTES % line.len(),
            0,
            "the cap falls inside a line"
        );
        // Past the cap by more than one read's worth.
        let flood = line.repeat(fit + 400);
        // A short first read, so that no later one ends on the cap.
        let (first, rest) = flood.split_at(100);
        let mut pending = Vec::new();
        let err = read_unsolicited(first.chain(rest), &mut pending)
            .expect_err("more than a line's worth before the hello is refused");
        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
        assert_eq!(pending.len(), MAX_LINE_BYTES, "kept up to the cap");
        assert_eq!(
            count_input_lines(&pending),
            u64::try_from(fit).expect("fits"),
            "every input that ends before the cap is counted, and the one it cuts is not"
        );
    }

    /// Exit must not depend on the wake-up connection succeeding: with the
    /// socket file gone, nothing can return the listener from `accept()`, and
    /// a join would hang the app forever.
    #[test]
    fn drop_returns_even_when_the_listener_cannot_be_woken() {
        let layer = bind_test_layer("taria-drophang-", "unwakeable");
        fs::remove_file(layer.socket_path()).expect("remove the socket file");

        // Drop off-thread so a regression fails this test instead of hanging
        // the whole suite.
        let (tx, rx) = mpsc::channel();
        thread::spawn(move || {
            drop(layer);
            let _ = tx.send(());
        });
        assert!(
            rx.recv_timeout(Duration::from_secs(5)).is_ok(),
            "drop must not join a listener that can no longer be woken"
        );
    }

    /// The case the wake-up-by-connecting exit never covered: not the socket
    /// being gone, but a *different* socket at the same path. Running the app
    /// twice does exactly that, and the old exit connected to the path, which
    /// landed on the second instance's listener, counted as having woken this
    /// one, and then joined a thread parked in `accept()` on a socket nothing
    /// could reach. The app hung after restoring the terminal.
    ///
    /// Dropping must also leave that socket alone: it is the other instance's
    /// way of being reached.
    #[test]
    fn drop_neither_hangs_nor_unlinks_when_the_socket_was_replaced() {
        let dir = private_temp_dir("taria-replaced-");
        let path = dir.path().join("replaced.sock");
        let layer = TariaLayer::bind_at("replaced", &path).expect("bind the first layer");

        // What a second instance used to do, done by hand because binding
        // over a live socket is now refused.
        fs::remove_file(&path).expect("remove this layer's socket");
        let usurper = UnixListener::bind(&path).expect("bind another listener at the path");

        // Off-thread so a regression fails this test instead of hanging the
        // whole suite.
        let (tx, rx) = mpsc::channel();
        thread::spawn(move || {
            drop(layer);
            let _ = tx.send(());
        });
        assert!(
            rx.recv_timeout(Duration::from_secs(5)).is_ok(),
            "drop must not wait on a listener no connection can reach any more"
        );
        assert!(
            path.exists(),
            "the socket at the path belongs to another listener; drop must leave it"
        );
        assert!(
            UnixStream::connect(&path).is_ok(),
            "and it must still be served"
        );
        drop(usurper);
    }

    /// `$TARIA_SOCK` is used verbatim, and a private directory passes the
    /// vetting whatever else is inside it, so a typo in it (`~/.ssh/config`
    /// is 0700 and user-owned) used to mean startup silently deleted a file
    /// the user meant to keep.
    #[test]
    fn binding_never_deletes_something_that_is_not_a_socket() {
        let dir = private_temp_dir("taria-notasocket-");
        let path = dir.path().join("config");
        let content = b"Host example\n  User me\n";
        fs::write(&path, content).expect("create the file that must survive");

        let Err(err) = TariaLayer::bind_at("notasocket", &path) else {
            panic!("binding over a regular file must be refused");
        };
        assert_eq!(err.kind(), io::ErrorKind::AlreadyExists);
        let message = err.to_string();
        assert!(message.contains("not a socket"), "err: {message}");
        assert!(message.contains("a regular file"), "err: {message}");
        assert_eq!(
            fs::read(&path).expect("the file is still there").as_slice(),
            content,
            "the file at the socket path must be untouched"
        );

        // A symlink is judged as itself, not as whatever it points at.
        let link = dir.path().join("config-link");
        std::os::unix::fs::symlink(&path, &link).expect("create the symlink");
        let Err(err) = TariaLayer::bind_at("notasocket", &link) else {
            panic!("binding over a symlink must be refused");
        };
        assert!(err.to_string().contains("a symlink"), "err: {err}");
        assert!(link.exists(), "the symlink must be left in place");
    }

    /// The case the pre-bind unlink exists for: a socket left behind by a run
    /// that was killed before `Drop` could run. Nothing is listening on it,
    /// and refusing it would leave the app unable to start until somebody
    /// deleted the file by hand.
    #[test]
    fn a_socket_nothing_is_listening_on_is_replaced() {
        let dir = private_temp_dir("taria-stalesock-");
        let path = dir.path().join("stale.sock");
        // Closing a listener does not unlink its socket, which is exactly
        // what SIGKILL leaves behind.
        drop(UnixListener::bind(&path).expect("bind a listener to abandon"));
        assert!(path.exists(), "the abandoned socket file stays behind");

        let layer = TariaLayer::bind_at("stale", &path).expect("a dead socket is replaced");
        assert!(layer.is_enabled());
        assert!(
            UnixStream::connect(&path).is_ok(),
            "the socket bound over the stale one must serve"
        );
    }

    /// Two instances of the same app, which is an ordinary thing to do while
    /// testing a TUI. The second used to unlink the first's socket and bind
    /// its own there, leaving the first running, reporting itself enabled,
    /// and reachable by nobody.
    #[test]
    fn binding_over_a_live_socket_is_refused_and_leaves_it_serving() {
        let dir = private_temp_dir("taria-twice-");
        let path = dir.path().join("twice.sock");
        let first = TariaLayer::bind_at("twice", &path).expect("the first instance binds");

        let Err(err) = TariaLayer::bind_at("twice", &path) else {
            panic!("a second instance must not take over the first's socket");
        };
        assert_eq!(err.kind(), io::ErrorKind::AddrInUse);
        let message = err.to_string();
        assert!(message.contains("already serving"), "err: {message}");
        assert!(
            message.contains("TARIA_SOCK"),
            "the error must say how to run a second instance: {message}"
        );

        // `bind_or_disabled` never fails, so it has to come back disabled
        // rather than pointed at a socket it does not own.
        let second = TariaLayer::bind_or_disabled_at("twice", path.clone());
        assert!(!second.is_enabled(), "the second instance must not be live");
        assert_eq!(
            second.bind_error().map(io::Error::kind),
            Some(io::ErrorKind::AddrInUse),
            "and it must say why"
        );
        drop(second);

        // The first instance is untouched throughout: still bound, still
        // serving, and still the owner of the socket file.
        assert!(first.is_enabled());
        assert!(
            UnixStream::connect(&path).is_ok(),
            "the first instance must still be reachable"
        );
        drop(first);
        assert!(!path.exists(), "the first instance still owns its socket");
    }

    /// The one queue that could grow without bound: inputs are capped and
    /// snapshots coalesce into one latest, but a bridge that reads steadily
    /// and slowly never stalls long enough for the write timeout to
    /// disconnect it while answers pile up behind it.
    #[test]
    fn the_ack_queue_is_bounded_and_drops_the_oldest_answers() {
        let shared = test_shared("ackbound");
        let overflow = 10;
        for id in 0..(ACK_QUEUE + overflow) as InputId {
            shared.queue_ack(id, InputStatus::Dropped);
        }

        let state = shared.lock_state();
        assert_eq!(state.acks.len(), ACK_QUEUE, "the queue must stay bounded");
        assert_eq!(
            state.acks.front().map(|(id, _)| *id),
            Some(overflow as InputId),
            "the oldest answers are the ones given up; a waiter on those has \
             long since timed out"
        );
        assert_eq!(
            state.acks.back().map(|(id, _)| *id),
            Some((ACK_QUEUE + overflow - 1) as InputId),
            "the newest answer is the one an agent is still waiting on"
        );
        drop(state);
        assert_eq!(
            shared.dropped_acks.load(Ordering::Relaxed),
            overflow as u64,
            "each dropped answer must be counted once"
        );
    }

    /// An id names an input only within the bridge process that sent it, and
    /// a fresh bridge counts from the start again. So an ack the app makes
    /// after a reconnect must not be sent: it would resolve the new bridge's
    /// waiter on its own live id with a verdict from a session that never saw
    /// it.
    #[test]
    fn an_app_ack_is_dropped_unless_its_input_belongs_to_the_live_connection() {
        let shared = test_shared("appack");
        shared.lock_state().generation = 7;
        assert!(shared.deliver(7, 5), "the live connection's input delivers");
        shared.lock_state().acks.clear();

        shared.queue_app_ack(5, InputStatus::Ignored);
        assert_eq!(
            shared.lock_state().acks.pop_front(),
            Some((5, InputStatus::Ignored)),
            "refining the ack of a live input is the whole point of `ack`"
        );

        // The connection ends and the next one starts; its ids start over.
        shared.retire_generation();
        shared.lock_state().generation += 1;
        shared.queue_app_ack(5, InputStatus::Ignored);
        assert!(
            shared.lock_state().acks.is_empty(),
            "an id from a bridge that is gone must answer nobody"
        );

        // And an id falls out of reach once a queue's worth of newer inputs
        // has been delivered past it, which costs a late ack rather than a
        // wrong one.
        // One delivery past what the layer remembers, so exactly the oldest
        // id falls out.
        let generation = shared.lock_state().generation;
        for id in 100..=(100 + ACKABLE_INPUTS as InputId) {
            assert!(shared.deliver(generation, id));
        }
        shared.lock_state().acks.clear();
        shared.queue_app_ack(100, InputStatus::Ignored);
        assert!(shared.lock_state().acks.is_empty(), "id 100 was forgotten");
        shared.queue_app_ack(101, InputStatus::Ignored);
        assert_eq!(
            shared.lock_state().acks.pop_front(),
            Some((101, InputStatus::Ignored)),
            "everything still remembered stays answerable"
        );
    }

    /// A disabled layer must not park an app forever. The enabled path turns
    /// a deadline no clock can hold into a wait for as long as the queue can
    /// deliver; a disabled layer's queue can never deliver, so that wait has
    /// no end, and an app pacing its loop on this call would run bound and
    /// freeze unbound.
    #[test]
    fn a_disabled_layer_does_not_block_forever_on_an_unholdable_timeout() {
        let dir = private_temp_dir("taria-disabledwait-");
        let blocker = dir.path().join("not-a-directory");
        fs::write(&blocker, b"").expect("create blocker file");
        let layer = TariaLayer::bind_or_disabled_at("disabledwait", blocker.join("app.sock"));
        assert!(!layer.is_enabled());

        let started = Instant::now();
        assert_eq!(layer.recv_timeout(Duration::MAX), None);
        assert_eq!(layer.recv_timeout_with_id(Duration::MAX), None);
        let waited = started.elapsed();
        assert!(
            waited < Duration::from_secs(1),
            "a disabled layer waited {waited:?} on a timeout it can never serve"
        );

        // A timeout the clock can hold is still waited out, so an app that
        // paces its loop on this call keeps its timing.
        let started = Instant::now();
        assert_eq!(layer.recv_timeout(Duration::from_millis(50)), None);
        assert!(started.elapsed() >= Duration::from_millis(50));
    }

    #[test]
    fn focus_detection_reaches_every_level_below_the_root() {
        let unfocused = Node::new("a", Role::Pane)
            .focused(true)
            .child(Node::new("b", Role::Text));
        assert!(
            !focused_below(&unfocused),
            "the root's own flag is not below it"
        );

        let deep_focus = Node::new("a", Role::Pane)
            .child(Node::new("b", Role::List).child(Node::new("c", Role::ListItem).focused(true)));
        assert!(focused_below(&deep_focus));
    }
}