path: root/src/detached/client.zig

//! THE FRONTEND SIDE of a detached session: a socket, a grid, and no core.
//!
//! A FRONTEND IS INPUT AND SCREEN, and that is the whole of it. Keystrokes,
//! mouse, window size go out; frames come back and get painted. This is the
//! shape the ESP32-P4 serial console has always had — a panel and a keypad on
//! the far end of a wire, performing no effects of its own — and after the
//! machine-local IO moved into the daemon it is the shape EVERY frontend has,
//! terminal and window alike.
//!
//! THIS SIDE DOES NOT OWN A `Pardes`. There is no `update`, no `postEvent` and
//! no `Host` in this file. The core lives in the detached process (server.zig),
//! which is also where every `Host.VTable` call now lands: forking a pane's
//! shell, writing a file, watching a path and reading the theme directory are
//! all done BY THE DAEMON, against the same machine's kernel it shares with
//! this frontend over an AF_UNIX socket. NO FORK HAPPENS IN A FRONTEND ANY
//! MORE. That is not a simplification of this file, it is the fix: a pane's
//! shell used to be a child of whichever frontend forked it, so leaving killed
//! the shells of a session whose entire promise is outliving frontends.
//!
//! `Client` is NOT a renderer either. It owns `grid` and `cursor`: the cells the
//! session is showing, kept current by applying frames as they arrive. Whoever
//! owns a terminal (or a window, or an ESP32-P4 panel) reads those and paints.
//! The split is deliberate — pardes already has terminal frontends, and a second
//! one living in here would be a second convention for a job that has one.
//!
//! THE ATTACH IS NOT A BLOCKING HANDSHAKE. `open` connects and writes the
//! `hello`; the `welcome` (or the `refuse`) arrives through the ordinary
//! `wait`/`next` loop like everything else. A frontend therefore has one loop
//! and one place it sleeps, instead of a startup path that can hang for two
//! seconds before its terminal is even in raw mode. `attached()` says whether
//! the session has greeted us; `refusal` says why it did not.
//!
//! WHAT ARRIVES. `next` hands back one decoded `wire.ServerMsg` at a time, and
//! there are exactly seven of them:
//!   * `welcome` and `frame` have already been applied to `slot`/`cols`/`rows`/
//!     `grid`/`cursor` by the time you see them. They are returned so a frontend
//!     knows the screen moved.
//!   * `set_clipboard`, `read_clipboard` and `open_link` are the only effects
//!     still on the wire, and they are here because each needs THIS HUMAN'S
//!     DISPLAY: a daemon nobody is looking at has no clipboard and no browser.
//!     The answer to `read_clipboard` is not a reply message: it is an ordinary
//!     `Event.paste` sent back through `send`, which is the same asynchronous
//!     shape `read_clipboard` already has in-process.
//!   * `refuse` is followed by the session closing the connection, and `quit`
//!     means the session itself has ended.
//! The switch in `next` is exhaustive over that set on purpose: putting a
//! machine-local effect back on the wire is a compile error here, and the test
//! at the bottom of this file says so in the other direction too.
//!
//! GEOMETRY. `cols`/`rows` are the SESSION's grid, which with several frontends
//! attached is the smallest common one and can be smaller than this frontend's
//! window. Tell the session about the window with `resize`; do not assume the
//! next frame will agree with it.
//!
//! BORROWED BYTES. Every slice in a returned `ServerMsg` points into this
//! client's receive buffer and is valid until the next call to `next` or
//! `wait`. A frontend that needs a payload for longer copies it — the same rule
//! the core's own `Event.output` bytes have.
const std = @import("std");
const libc = std.c;
const pardes = @import("../pardes.zig");
const server = @import("server.zig");
const wire = @import("wire.zig");

const read_chunk = 16 * 1024;

pub const Error = error{
    /// No `$XDG_RUNTIME_DIR` and no `$HOME`, or a name that is not one path
    /// component — there is no socket path to try.
    NoSessionPath,
    /// Nothing is listening there: the name is wrong, or that session ended.
    /// Its socket file, if it is still lying about, is unlinked by the sweep the
    /// next detached session runs.
    NoSession,
    /// The socket is there and this frontend will not talk to it: the directory
    /// or the socket is not a private one of ours (server.zig `vetted`). A
    /// planted socket at a derivable path collects every keystroke typed into
    /// the frontend that trusts it, so this is refused rather than reported as
    /// "no session" — the two need different answers from a human.
    NotPrivate,
    /// The session hung up, or the connection failed under us. Every read and
    /// write path funnels here: a frontend's answer to all of them is the same
    /// (report and exit), so distinguishing them would be a distinction nobody
    /// acts on.
    Closed,
    /// The session said something before it said `welcome`.
    Ungreeted,
};

pub const Client = struct {
    gpa: std.mem.Allocator,
    fd: c_int = -1,
    /// Which client slot the session gave this connection. Diagnostics only,
    /// and it exists so both sides print the same number.
    slot: u8 = 0,
    /// Set by a `refuse`, and the reason a frontend prints before exiting.
    refusal: ?wire.Refusal = null,
    /// The SESSION's grid, not this frontend's window. Zero until the welcome.
    cols: u16 = 0,
    rows: u16 = 0,
    /// `cols * rows` cells: what the session is showing right now.
    grid: std.ArrayListUnmanaged(pardes.Cell) = .empty,
    cursor: ?wire.Cursor = null,
    in: std.ArrayListUnmanaged(u8) = .empty,
    out: std.ArrayListUnmanaged(u8) = .empty,
    /// Bytes of `in` belonging to the message `next` returned last. Compacted at
    /// the top of the next call, which is exactly what makes that message's
    /// borrowed slices valid until then and no longer.
    held: usize = 0,

    /// Connect to the session called `name` and say hello, telling it this
    /// frontend's window. Does not wait: the greeting arrives through `next`.
    pub fn open(gpa: std.mem.Allocator, name: []const u8, cols: u16, rows: u16) (Error || wire.Error)!Client {
        var path_buf: [server.path_max]u8 = undefined;
        const path = server.sessionPath(&path_buf, name) orelse return error.NoSessionPath;
        // `vetted` is one predicate over two different failures, and a human is
        // owed different words for them: `NotPrivate` promises, in its own doc
        // above, that the socket IS there. So ask the cheap question first,
        // because the commonest failure of all is a mistyped session name —
        // until this, `Attach nosuchsession` put "attach: NotPrivate" on the
        // message row, which reads as an accusation rather than a typo. Spelled
        // with the `access` this file already uses in `resolve`.
        const F_OK: c_int = 0;
        if (libc.access(path, F_OK) != 0) return error.NoSession;
        // Both ends vet, and this is this end's half: the session refuses a
        // directory or a socket anyone else can reach before it binds, and until
        // this a frontend connected to whatever it found at the path it derived.
        // See server.zig `vetted` for what is asked and why it is asked there.
        if (!server.vetted(path)) return error.NotPrivate;
        var addr: libc.sockaddr.un = .{ .path = @splat(0) };
        @memcpy(addr.path[0 .. path.len + 1], path[0 .. path.len + 1]);
        const fd = libc.socket(libc.AF.UNIX, libc.SOCK.STREAM, 0);
        if (fd < 0) return error.NoSession;
        // CLOEXEC anyway, even though a frontend no longer forks pane shells:
        // `open_link` runs this display's browser, and a `xdg-open` inheriting
        // this socket would keep the session believing a frontend is attached
        // long after this process left.
        server.setCloexec(fd);
        if (comptime server.darwin) {
            // linux says MSG_NOSIGNAL per write and darwin says it once per
            // socket, which is what `nosignal` being 0 on darwin MEANS — so
            // this call is the whole of that platform's protection and the
            // comment on `nosignal` used to say `open` made it without `open`
            // making it. A session that ends mid-write must not take the
            // frontend down with SIGPIPE. server.zig `accept` is the mirror.
            const on: c_int = 1;
            _ = libc.setsockopt(fd, libc.SOL.SOCKET, libc.SO.NOSIGPIPE, &on, @sizeOf(c_int));
        }
        // Still blocking for the connect itself, which on AF_UNIX either lands
        // in the listener's backlog immediately or is refused; there is no
        // in-progress state to poll for.
        if (libc.connect(fd, @ptrCast(&addr), @sizeOf(@TypeOf(addr))) != 0) {
            _ = libc.close(fd);
            return error.NoSession;
        }
        server.setNonblock(fd);
        var c: Client = .{ .gpa = gpa, .fd = fd };
        errdefer c.deinit();
        try c.send(.{ .hello = .{
            .cols = @min(cols, wire.max_cols),
            .rows = @min(rows, wire.max_rows),
        } });
        return c;
    }

    /// Has the session greeted us? Until it has, `grid` is empty and nothing
    /// has been drawn.
    pub fn attached(c: *const Client) bool {
        return c.cols != 0;
    }

    pub fn deinit(c: *Client) void {
        if (c.fd >= 0) _ = libc.close(c.fd);
        c.fd = -1;
        c.grid.deinit(c.gpa);
        c.in.deinit(c.gpa);
        c.out.deinit(c.gpa);
    }

    /// Leave without ending the session. The `bye` is a courtesy — the session
    /// handles a frontend that simply dies, and proving it does is what that
    /// test is for — but it turns "the peer vanished" into "the peer left" in
    /// the session's log, which is worth seven bytes.
    pub fn detach(c: *Client) void {
        c.send(.bye) catch {};
        c.deinit();
    }

    /// One message on its way to the core. Everything a frontend collects goes
    /// through here, and after the IO moved into the daemon that is a short
    /// list: keys, the mouse, a window resize, and the paste that answers a
    /// `read_clipboard`. `ClientMsg.output`/`.eof` still exist on the wire but
    /// no frontend sends them any more — pty bytes are read by the process that
    /// forked the shell, which is the daemon.
    pub fn send(c: *Client, msg: wire.ClientMsg) (Error || wire.Error)!void {
        const want = wire.clientBound(msg);
        c.out.ensureUnusedCapacity(c.gpa, want) catch return error.Closed;
        const at = c.out.items.len;
        // Encoded straight into the queue's tail rather than through a scratch
        // buffer: a paste is four megabytes and copying it twice is two copies.
        c.out.items.len += want;
        const bytes = wire.encodeClient(c.out.items[at..], msg) catch |err| {
            // AND THE QUEUE GOES BACK. `want` bytes of it are uninitialised
            // right now, and leaving them there — which is what a bare `try`
            // did — puts that much stack-shaped garbage on the socket at the
            // next flush: the session decodes it as a message, refuses it, and
            // drops a frontend whose only mistake was a message this protocol
            // cannot carry (an `Event.command` past 64 KiB is `Overlong`, and a
            // caller that mis-sized the queue is `NoSpace`).
            c.out.items.len = at;
            return err;
        };
        c.out.items.len = at + bytes.len;
        return c.flush();
    }

    /// Tell the session this frontend's window changed. Not a promise about the
    /// next frame: with other frontends attached the session grid is the
    /// smallest common one.
    ///
    /// Clamped, like the hello in `open`: a window past `max_cols`/`max_rows`
    /// is a geometry the protocol cannot carry, and the decoder on the far end
    /// answers one with `BadValue` — which `apply` turns into `close(.protocol)`
    /// with no `refuse` behind it, so the frontend was told only that the
    /// session "hung up on the connect". A 4K display at a small font is
    /// already past `max_rows`, which made an ordinary big screen unable to
    /// attach at all. Asking for the largest grid the wire carries is what this
    /// file's GEOMETRY note already promises a frontend gets: the session is
    /// drawn at ITS size wherever the window is bigger, exactly as it is when
    /// another frontend is the smaller one.
    ///
    /// A ZERO is dropped rather than clamped, and that asymmetry is the whole
    /// point of `getCols` refusing zero in the first place: the session grid is
    /// the smallest common one, so a frontend that reported 1 would collapse
    /// every other frontend to a single cell. `TIOCGWINSZ` answers 0x0 while a
    /// terminal is being torn down and tty.zig forwards a `winsize` verbatim
    /// (its ATTACH path already refuses a zero one, its resize path did not),
    /// so this is reachable without a hostile peer — and it reached the session
    /// as the same mute `close(.protocol)` the oversize geometry did. Keeping
    /// the last real window is what a momentary zero means.
    pub fn resize(c: *Client, cols: u16, rows: u16) (Error || wire.Error)!void {
        if (cols == 0 or rows == 0) return;
        return c.send(.{ .event = .{ .resize = .{
            .cols = @min(cols, wire.max_cols),
            .rows = @min(rows, wire.max_rows),
        } } });
    }

    /// Wait up to `timeout_ms` for the session to say something, and push
    /// whatever we still owe it. Zero blocks. This is the frontend's one
    /// sleeping place FOR THE SOCKET; the terminal it draws on is polled by
    /// whoever owns that, which is why this takes a timeout rather than a second
    /// descriptor.
    pub fn wait(c: *Client, timeout_ms: u32) (Error || wire.Error)!void {
        if (c.fd < 0) return error.Closed;
        try c.flush();
        var fds: [1]libc.pollfd = .{.{
            .fd = c.fd,
            .events = if (c.out.items.len != 0) poll_in | poll_out else poll_in,
            .revents = 0,
        }};
        const timeout: c_int = if (timeout_ms == 0) -1 else @intCast(@min(timeout_ms, std.math.maxInt(c_int)));
        if (libc.poll(&fds, 1, timeout) <= 0) return; // a timeout, or EINTR
        if (fds[0].revents & poll_out != 0) try c.flush();
        // POLLIN wins over POLLHUP: a session that wrote a `quit` and then
        // closed has bytes worth reading.
        if (fds[0].revents & poll_in != 0) return c.fill();
        if (fds[0].revents & (poll_hup | poll_err | poll_nval) != 0) return error.Closed;
    }

    /// The next complete message, or null when the buffer holds only part of
    /// one. `welcome` and `frame` have already been applied to this client's
    /// own state; every slice in the result borrows the receive buffer until the
    /// next call here or to `wait`.
    pub fn next(c: *Client) (Error || wire.Error)!?wire.ServerMsg {
        // Retire the message returned last, now that its borrow window is over.
        if (c.held != 0) {
            if (c.held == c.in.items.len) {
                c.in.clearRetainingCapacity();
            } else {
                std.mem.copyForwards(u8, c.in.items, c.in.items[c.held..]);
                c.in.items.len -= c.held;
            }
            c.held = 0;
        }
        const found = (try wire.framed(c.in.items)) orelse return null;
        const msg = try wire.decodeServer(found.tag, found.payload);
        c.held = found.total;
        switch (msg) {
            .welcome => |v| {
                // Both sides check the version. This side checks it too because
                // a session speaking something else may not have recognised our
                // hello as one either, and a frontend must not paint a frame it
                // decoded by a layout the other end does not use.
                if (v.version != wire.version) return error.Ungreeted;
                c.slot = v.slot;
                try c.reshape(v.cols, v.rows);
            },
            .refuse => |why| c.refusal = why,
            .frame => |f| {
                // A frame before the greeting would be the session drawing for a
                // connection it never accepted.
                if (!c.attached()) return error.Ungreeted;
                // A geometry change and a late attach are the same case on this
                // side too: reshape, and require the FULL frame the session
                // promises for it. Applying a diff to a grid we just cleared
                // would leave every untouched cell blank.
                if (f.cols != c.cols or f.rows != c.rows) {
                    if (f.kind != .full) return error.BadValue;
                    try c.reshape(f.cols, f.rows);
                }
                try f.apply(c.grid.items);
                c.cursor = f.cursor;
            },
            // The session has ended. Left for the caller to act on, and the
            // descriptor stays open so `deinit` is the only place that closes.
            .quit => {},
            // THIS frontend is done and the session is not. Same shape as
            // `quit` and the same non-answer here — the caller leaves its loop
            // — but the opposite meaning about what survives, so a frontend
            // must not collapse the two: `quit` is the session ending and
            // `detach` is success. Guarded, because a peer that has not greeted
            // us has no standing to dismiss us either.
            .detach => if (!c.attached()) return error.Ungreeted,
            // The three display effects need this process's clipboard and this
            // process's browser, so they are the caller's to perform and there
            // is nothing for a `Client` to update. Listed rather than swept up
            // by an `else`: an `else` here would silently accept a
            // machine-local effect returning to the wire, and the point of the
            // rewrite is that it cannot.
            //
            // Guarded like `.frame`, and for a stronger reason than drawing:
            // these reach the human's clipboard and the human's browser. The
            // protocol version is checked in the `.welcome` arm above and
            // nowhere else, so a peer that simply never greets us has had its
            // version checked by nobody — and until it does, it does not get to
            // open a URI on this display or read this display's selection back
            // over the socket. `Ungreeted` is the same refusal a frame gets.
            .set_clipboard, .read_clipboard, .open_link => if (!c.attached()) return error.Ungreeted,
        }
        return msg;
    }

    // ---- internals --------------------------------------------------------

    fn reshape(c: *Client, cols: u16, rows: u16) (Error || wire.Error)!void {
        c.grid.resize(c.gpa, @as(usize, cols) * @as(usize, rows)) catch return error.Closed;
        // Unpainted, which a frontend draws as the terminal's own default cell.
        // The full frame that follows paints over it.
        @memset(c.grid.items, .{});
        c.cols = cols;
        c.rows = rows;
        c.cursor = null;
    }

    /// Take everything the kernel is holding, not one chunk of it. The server
    /// deliberately reads its clients ONE chunk per round, because it is
    /// dividing a loop between thirty-two of them; a frontend has exactly one
    /// peer, and reading one 16 KiB slice per poll would leave it a full frame
    /// behind on every large one — and the session drops a frontend whose queue
    /// it cannot drain (server.zig `out_backlog`).
    fn fill(c: *Client) (Error || wire.Error)!void {
        var buf: [read_chunk]u8 = undefined;
        while (true) {
            const got = libc.read(c.fd, &buf, buf.len);
            if (got == 0) return error.Closed;
            if (got < 0) return switch (libc.errno(got)) {
                .INTR => continue,
                // Nothing more is ready; what we have is what there was.
                .AGAIN => {},
                else => error.Closed,
            };
            c.in.appendSlice(c.gpa, buf[0..@intCast(got)]) catch return error.Closed;
        }
    }

    fn flush(c: *Client) Error!void {
        if (c.fd < 0) return error.Closed;
        var off: usize = 0;
        while (off < c.out.items.len) {
            const n = libc.send(c.fd, c.out.items.ptr + off, c.out.items.len - off, nosignal);
            if (n < 0) switch (libc.errno(n)) {
                .INTR => continue,
                // The session is not draining us. The rest waits for POLLOUT;
                // the queue is bounded in practice because a frontend's output
                // is keystrokes and pty chunks, never frames.
                .AGAIN => break,
                else => return error.Closed,
            };
            if (n == 0) break;
            off += @intCast(n);
        }
        if (off == 0) return;
        if (off == c.out.items.len) return c.out.clearRetainingCapacity();
        std.mem.copyForwards(u8, c.out.items, c.out.items[off..]);
        c.out.items.len -= off;
    }
};

// The socket primitives are server.zig's, which is the file that owns this
// transport's conventions and both ends of it — see its `setNonblock`,
// `nosignal` and `poll_*`. There was a third copy of all of them here.
const nosignal = server.nosignal;
const poll_in = server.poll_in;
const poll_out = server.poll_out;
const poll_hup = server.poll_hup;
const poll_err = server.poll_err;
const poll_nval = server.poll_nval;

/// Which session an attach meant, answered before a frontend opens its window.
///
/// `--attach=<name>` is taken at its word beyond one `access` on the socket
/// file, because the connect is the real authority on whether anything is
/// listening — and a typo is the one failure worth catching earlier, since the
/// alternative is a full-screen flash on the way to a one-line message. Bare
/// `--attach`, and the bare `Attach` word, is THE session: bare `--detach`
/// names itself by its own pid, and nobody can be expected to read a pid out
/// of `$XDG_RUNTIME_DIR`. With exactly one listening, that is the one meant;
/// with none or several this says WHICH case it is instead of picking one.
///
/// Both the directory and the filename convention come off ONE probe through
/// `server.sessionPath` rather than being re-derived here, for the reason that
/// function exists at all: the side that binds and the side that looks must
/// never be able to disagree about where a session lives.
///
/// It lives in this file, rather than in either shell, because BOTH frontends
/// now ask the same question — a terminal for `--attach` and the SDL window
/// for the `Attach` word — and a second copy of a directory scan is exactly
/// how the two of them would start disagreeing.
pub const Resolved = union(enum) {
    /// The session to open. Borrows `requested` when it was named, and `buf`
    /// when it had to be scanned for.
    name: []const u8,
    /// No socket of that name, or no session at all. The caller knows which it
    /// asked for, so it owns the wording.
    none,
    /// Several are listening, and choosing between them is not ours to do.
    ambiguous: usize,
};

pub fn resolve(buf: *[server.path_max]u8, requested: []const u8) Resolved {
    if (requested.len != 0) {
        var one_buf: [server.path_max]u8 = undefined;
        const one = server.sessionPath(&one_buf, requested) orelse return .none;
        const F_OK: c_int = 0;
        if (libc.access(one, F_OK) != 0) return .none;
        return .{ .name = requested };
    }
    const probe_name = "0";
    var probe_buf: [server.path_max]u8 = undefined;
    const probe = server.sessionPath(&probe_buf, probe_name) orelse return .none;
    const base = std.fs.path.basename(probe);
    const cut = std.mem.lastIndexOf(u8, base, probe_name).?;
    const prefix = base[0..cut];
    const suffix = base[cut + probe_name.len ..];
    var dir_buf: [server.path_max:0]u8 = undefined;
    const dir = std.fs.path.dirname(probe) orelse "";
    if (dir.len == 0 or dir.len >= dir_buf.len) return .none;
    @memcpy(dir_buf[0..dir.len], dir);
    dir_buf[dir.len] = 0;
    const d = libc.opendir(dir_buf[0..dir.len :0]) orelse return .none;
    defer _ = libc.closedir(d);
    var found: usize = 0;
    var len: usize = 0;
    while (libc.readdir(d)) |ent| {
        const entry = std.mem.sliceTo(&ent.name, 0);
        if (entry.len <= prefix.len + suffix.len) continue;
        if (!std.mem.startsWith(u8, entry, prefix) or !std.mem.endsWith(u8, entry, suffix)) continue;
        const name = entry[prefix.len .. entry.len - suffix.len];
        if (name.len > buf.len) continue;
        found += 1;
        @memcpy(buf[0..name.len], name);
        len = name.len;
    }
    // A socket file whose session is gone still counts here: the sweep that
    // unlinks corpses runs when the NEXT session binds (server.zig `sweep`),
    // and probing every candidate with a connect would put a phantom frontend
    // into a live session's slot table just to count it. One stale file
    // therefore fails at `open` with "no such session", which is the truth.
    if (found != 1) return if (found == 0) .none else .{ .ambiguous = found };
    return .{ .name = buf[0..len] };
}

// Input arrives through frontend queues, so attached clients bound socket waits to half a frame.
pub const poll_ms: u32 = 8;

/// One round of waiting for the greeting, and how many of them. The COUNT is
/// derived from `server.greet_deadline_default_ms` rather than written down
/// again, so the two ends of the handshake give each other the same grace out
/// of one number instead of two that can drift apart.
///
/// Rounds rather than a clock, and that is sound rather than lazy: the only
/// message a session may legally send before its `welcome` is a `refuse`.
/// Everything else is refused by `next` as `Ungreeted` — a frame because it
/// would be drawing for a connection that was never accepted, and the three
/// display effects because a peer whose version nobody has checked does not get
/// to open a URI or read a clipboard. So a peer that is silent costs one whole
/// timeout per round, which makes the round count a real wall-clock bound; and
/// a peer that is NOISY but ungreeted fails on its first message rather than
/// spending the budget.
const greet_round_ms: u32 = 250;
const greet_rounds: u32 = server.greet_deadline_default_ms / greet_round_ms;

/// What came of trying to attach. A VALUE and not an error union, because four
/// of the six outcomes are ordinary answers a human needs different words for,
/// and because the two frontends report them by different mechanisms — a
/// terminal prints a sentence and exits, a window logs and unwinds. `resolve`
/// above returns a union for the same reason.
pub const Attempt = union(enum) {
    /// Resolved, connected, AND greeted. The caller owns it.
    greeted: Client,
    /// No socket of that name, or nothing detached at all. The caller knows
    /// which it asked for, so the wording is the caller's.
    no_session,
    /// Several are listening and choosing is not ours to do.
    ambiguous: usize,
    /// The session said no, and said why.
    refused: wire.Refusal,
    /// It accepted and then never greeted us inside the deadline.
    silent,
    /// Everything else, already reported by the errno it came from.
    lost: anyerror,
};

/// Wait for welcome before the caller replaces its session; connect alone can
/// still lead to a version refusal. Every non-greeted result owns no resources.
pub fn attempt(gpa: std.mem.Allocator, requested: []const u8, cols: u16, rows: u16) Attempt {
    var buf: [server.path_max]u8 = undefined;
    const name = switch (resolve(&buf, requested)) {
        .name => |n| n,
        .none => return .no_session,
        .ambiguous => |n| return .{ .ambiguous = n },
    };
    var c = Client.open(gpa, name, cols, rows) catch |err| return switch (err) {
        // `open`'s own two ways of saying "there is nothing there" collapse
        // into the one the caller has wording for.
        error.NoSession, error.NoSessionPath => .no_session,
        else => .{ .lost = err },
    };
    var rounds: u32 = 0;
    while (rounds < greet_rounds) : (rounds += 1) {
        c.wait(greet_round_ms) catch |err| return give(&c, err);
        // Drain whatever landed. `next` is what applies the welcome, so the
        // loop below is not discarding anything it needs — the state it wants
        // is in `c` afterwards.
        while (true) {
            const msg = c.next() catch |err| return give(&c, err);
            if (msg == null) break;
        }
        if (c.attached()) return .{ .greeted = c };
        if (c.refusal) |why| {
            c.deinit();
            return .{ .refused = why };
        }
    }
    c.deinit();
    return .silent;
}

/// Close, and say what the failure actually was. A session that refuses us
/// writes its reason and closes in the same pass (server.zig `refuseFd`), so a
/// read error here is usually the far side hanging up on a refusal we have
/// already decoded — reporting that as a lost connection would throw away the
/// one sentence worth telling the human.
fn give(c: *Client, err: anyerror) Attempt {
    const why = c.refusal;
    c.deinit();
    if (why) |w| return .{ .refused = w };
    return .{ .lost = err };
}

// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------
//
// One real core, one real unix socket, real frontends. The harness runs the
// server's vtable in the order `Pardes.pump` runs it, so what these exercise is
// the transport as the core actually drives it rather than a mock of it.

const testing = std.testing;

extern "c" fn setenv(name: [*:0]const u8, value: [*:0]const u8, overwrite: c_int) c_int;
extern "c" fn unsetenv(name: [*:0]const u8) c_int;

/// A session on a socket of its own under `.zig-cache/tmp`, so a test never
/// collides with a real session in `$XDG_RUNTIME_DIR` and never depends on that
/// variable being set at all. It is process-wide, so it is saved and restored.
const Harness = struct {
    tmp: std.testing.TmpDir,
    saved: ?[:0]const u8,
    saved_buf: [4096:0]u8 = undefined,
    core: *pardes.Pardes,
    session: server.Session,
    arena: std.heap.ArenaAllocator,
    name_buf: [32]u8 = undefined,
    name: []const u8 = &.{},

    fn init(h: *Harness, cols: u16, rows: u16) !void {
        h.tmp = std.testing.tmpDir(.{});
        errdefer h.tmp.cleanup();
        h.saved = if (libc.getenv("XDG_RUNTIME_DIR")) |v|
            try std.fmt.bufPrintSentinel(&h.saved_buf, "{s}", .{std.mem.span(v)}, 0)
        else
            null;
        errdefer h.restoreEnv();
        var dir_buf: [4096:0]u8 = undefined;
        const dir = try std.fmt.bufPrintSentinel(&dir_buf, ".zig-cache/tmp/{s}", .{h.tmp.sub_path}, 0);
        // `ensureSocketDir` refuses anything with a bit granted to group or
        // other, which is the whole point of it; a tmpDir arrives 0755.
        try testing.expectEqual(@as(c_int, 0), libc.chmod(dir, 0o700));
        _ = setenv("XDG_RUNTIME_DIR", dir.ptr, 1);
        h.core = try pardes.Pardes.init(testing.allocator, .{ .tty_only = true, .cols = cols, .rows = rows });
        errdefer h.core.deinit();
        h.arena = .init(testing.allocator);
        errdefer h.arena.deinit();
        // `io` is not optional on `Session`: the daemon does the file watching
        // and the theme scan itself now, and both of those take a `std.Io`.
        h.session = .{
            .gpa = testing.allocator,
            .worker_gpa = testing.allocator,
            .io = std.testing.io,
            .core = h.core,
            .cols = cols,
            .rows = rows,
        };
        try h.session.initAsync();
        errdefer h.session.deinit();
        h.name = "s";
        try testing.expect(h.session.listen(h.name));
        // The pre-loop drain tty.zig has, for its reason: the startup spawns are
        // already queued and a session must not open its socket with panes that
        // have not been created. These now fork IN THIS PROCESS, because the
        // daemon is the thing that owns pane shells — which is exactly the
        // property under test, and the reason no frontend below is ever asked
        // to fork anything.
        h.core.host = h.session.host();
        while (h.core.nextEffect()) |e| h.core.perform(e);
    }

    fn restoreEnv(h: *Harness) void {
        if (h.saved) |v| {
            _ = setenv("XDG_RUNTIME_DIR", v.ptr, 1);
        } else _ = unsetenv("XDG_RUNTIME_DIR");
    }

    fn deinit(h: *Harness) void {
        h.session.deinit();
        h.core.deinit();
        h.arena.deinit();
        h.restoreEnv();
        h.tmp.cleanup();
    }

    /// `Pardes.pump`, with the one substitution a single-threaded test needs: a
    /// bounded wait, so a frontend that says nothing cannot hang the suite
    /// where the real session would sleep until it spoke. The queued-input
    /// drain `pump` does between the two is absent because this host has no
    /// queue: it calls `update` directly (borrowed bytes), and `postEvent` is
    /// for hosts whose worker threads post from off the loop.
    fn pump(h: *Harness) !void {
        const host = h.session.host();
        host.vtable.wait_input.?(host.ctx, 20);
        while (h.core.nextEffect()) |e| h.core.perform(e);
        if (h.core.quit) return;
        _ = h.arena.reset(.retain_capacity);
        const surface = try h.core.render(h.arena.allocator());
        host.vtable.present.?(host.ctx, surface);
    }

    /// The round budget every `pumpUntil*` below shares. A round moves at most
    /// one socket buffer, because nothing here is concurrent: the session
    /// flushes until `EAGAIN`, and only then does the client read. That buffer
    /// is 8 KiB on darwin (`net.local.stream.sendspace`) against linux's 208
    /// KiB, which is the same asymmetry the slow-frontend test at the bottom of
    /// this file already had to say out loud — and a full frame of the largest
    /// grid the protocol carries is near a megabyte, so 64 rounds is a linux-
    /// only number. High enough for that frame on the smaller buffer, and still
    /// a bound: a broken transport fails a test rather than hanging the suite.
    const rounds = 256;

    /// Pump until this client has the message we are waiting for.
    fn pumpUntil(h: *Harness, c: *Client, comptime want: std.meta.Tag(wire.ServerMsg)) !wire.ServerMsg {
        for (0..rounds) |_| {
            try h.pump();
            try c.wait(5);
            while (try c.next()) |msg| if (std.meta.activeTag(msg) == want) return msg;
        }
        return error.NeverArrived;
    }

    /// Pump until this client is sent a frame that CHANGES something. The first
    /// frame after an input is not always the one carrying it — an effect the
    /// input queued (a `Look` on a directory emits a spawn) lands a frame
    /// later, and the frame in between legitimately says nothing.
    fn pumpUntilChange(h: *Harness, c: *Client) !wire.Frame {
        for (0..rounds) |_| {
            const msg = try h.pumpUntil(c, .frame);
            if (msg.frame.nruns > 0) return msg.frame;
        }
        return error.NothingChanged;
    }

    /// Pump until this client's grid is this shape, draining everything that
    /// arrives. A test waits for the STATE rather than for the n-th message
    /// because a resize is announced when the session settles it, which may be
    /// one empty frame after the pump that caused it.
    fn pumpUntilGrid(h: *Harness, c: *Client, cols: u16, rows: u16) !void {
        for (0..rounds) |_| {
            try h.pump();
            try c.wait(5);
            while (try c.next()) |_| {}
            if (c.cols == cols and c.rows == rows) return;
        }
        return error.NeverResized;
    }

    /// Pump until two frontends are showing the same screen, draining both on
    /// every pass. One pump sends every attached frontend a frame, so a test
    /// that drains only one of them is comparing two different instants — and
    /// what a shared session promises is that they CONVERGE, which is what this
    /// waits for.
    fn pumpUntilSameScreen(h: *Harness, a: *Client, b: *Client) !void {
        for (0..rounds) |_| {
            try h.pump();
            try a.wait(5);
            try b.wait(5);
            while (try a.next()) |_| {}
            while (try b.next()) |_| {}
            if (sameScreen(a.grid.items, b.grid.items)) return;
        }
        return error.NeverConverged;
    }

    /// Pump until THIS frontend is showing the core's own screen, draining
    /// everything that arrives on every pass. `pumpUntilSameScreen`'s reason,
    /// with one frontend instead of two, and a second reason of its own:
    /// `pumpUntil` returns the instant it decodes the message it was waiting
    /// for and leaves the rest of that burst in the socket, so a client can sit
    /// one whole frame behind for the rest of a test. Comparing once against
    /// that grid compares the core's NOW with the frontend's THEN — which is
    /// nothing at all until the pane shell's prompt lands in the frame nobody
    /// read, and then it is a cell that differs.
    ///
    /// Checked BEFORE the first pump, so a test already in sync spends nothing
    /// and no extra frame is manufactured to make one appear. Out of rounds it
    /// hands the last comparison to `expectSameScreen`, which names the cell:
    /// a transport that really does drop one must fail as a wrong screen, not
    /// as a timeout.
    fn pumpUntilShowsCore(h: *Harness, c: *Client) !void {
        for (0..rounds) |_| {
            _ = h.arena.reset(.retain_capacity);
            if (sameScreen((try h.core.render(h.arena.allocator())).cells, c.grid.items)) return;
            try h.pump();
            try c.wait(5);
            while (try c.next()) |_| {}
        }
        _ = h.arena.reset(.retain_capacity);
        return expectSameScreen((try h.core.render(h.arena.allocator())).cells, c.grid.items);
    }

    /// Attach a frontend and get it greeted: `open` writes the hello into the
    /// listener's backlog, one pump accepts and answers it. Nothing blocks,
    /// which is the whole reason the handshake is not a blocking call.
    fn attach(h: *Harness, cols: u16, rows: u16) !Client {
        var c = try Client.open(testing.allocator, h.name, cols, rows);
        errdefer c.deinit();
        _ = try h.pumpUntil(&c, .welcome);
        return c;
    }
};

test "detached session: a frontend attaches, is greeted, and is sent the screen" {
    var h: Harness = undefined;
    try h.init(60, 16);
    defer h.deinit();

    var c = try h.attach(60, 16);
    defer c.deinit();
    try testing.expectEqual(@as(u8, 0), c.slot);
    try testing.expectEqual(@as(u16, 60), c.cols);
    try testing.expectEqual(@as(u16, 16), c.rows);

    const frame = (try h.pumpUntil(&c, .frame)).frame;
    // The first frame a frontend gets must be full: it has nothing to diff
    // against.
    try testing.expectEqual(wire.FrameKind.full, frame.kind);
    try testing.expectEqual(@as(usize, 60 * 16), c.grid.items.len);
    // ...and it must be the core's own frame, cell for cell. This is the whole
    // claim of the transport — asserted once the frontend has read everything
    // the session sent, which `pumpUntil` above deliberately did not do.
    try h.pumpUntilShowsCore(&c);
}

test "detached session: input from a frontend reaches the core and comes back as a diff" {
    var h: Harness = undefined;
    try h.init(60, 16);
    defer h.deinit();
    var c = try h.attach(60, 16);
    defer c.deinit();
    _ = try h.pumpUntil(&c, .frame);

    // A builtin line is the cheapest input with a guaranteed visible effect,
    // and it travels the same `Event` path a keystroke does.
    try c.send(.{ .event = .{ .command = "Look /" } });
    const frame = try h.pumpUntilChange(&c);
    // The change arrived as a DIFF: this frontend was already in sync, so
    // nothing it already had was re-sent.
    try testing.expectEqual(wire.FrameKind.diff, frame.kind);
    try h.pumpUntilShowsCore(&c);
}

test "detached Restore keeps attached frontends and follows queued frames with a full replacement" {
    var h: Harness = undefined;
    try h.init(60, 16);
    defer h.deinit();
    _ = try h.core.setTestFile("saved body\n");
    var c = try h.attach(60, 16);
    defer c.deinit();
    _ = try h.pumpUntil(&c, .frame);
    try h.pumpUntilShowsCore(&c);

    const before = h.core;
    const fd = h.session.clients[c.slot].fd;
    try testing.expectError(error.BadDumpMagic, h.session.restore(
        ".{ .magic = \"not-a-pardes-dump\", .theme = \"dark\", .screen = .{ .cols = 60, .rows = 16 } }",
    ));
    try testing.expectEqual(before, h.session.core);
    try testing.expectEqual(fd, h.session.clients[c.slot].fd);
    try testing.expect(h.session.clients[c.slot].attached);

    try h.core.dumpState();
    const saved = try testing.allocator.dupe(u8, h.core.dump_out.?);
    defer testing.allocator.free(saved);
    while (h.core.nextEffect()) |_| {}
    _ = try h.core.setTestFile("changed after dump\n");
    try h.session.restore(saved);
    h.core = h.session.core;
    try testing.expect(h.core != before);
    try testing.expectEqual(fd, h.session.clients[c.slot].fd);
    try testing.expect(h.session.clients[c.slot].attached);
    try testing.expectEqualStrings("saved body\n", h.core.panes[0].?.file.?.content);
    try testing.expect(h.session.clients[c.slot].need_full);
    var wake = [_]libc.pollfd{.{ .fd = h.session.mailbox.wake[0], .events = poll_in, .revents = 0 }};
    try testing.expectEqual(@as(c_int, 1), libc.poll(&wake, wake.len, 0));
    for (0..Harness.rounds) |_| {
        if ((try h.pumpUntil(&c, .frame)).frame.kind == .full) break;
    } else return error.NoFullReplacement;
    try h.pumpUntilShowsCore(&c);
}

test "detached selection pipe runs off the loop and returns through the attached frontend" {
    var h: Harness = undefined;
    try h.init(60, 16);
    defer h.deinit();
    const pane = try h.core.setTestFile("one\ntwo\n");
    pane.cur_row = 0;
    pane.cur_col = 2;
    pane.vsel = .{ .active = true, .row = 0, .col = 0, .explicit = true };
    var client = try h.attach(60, 16);
    defer client.deinit();
    _ = try h.pumpUntil(&client, .frame);

    try client.send(.{ .event = .{ .key = .{ .cp = '|' } } });
    try client.send(.{ .event = .{ .key = .{ .cp = 't', .text = "tr a-z A-Z" } } });
    try client.send(.{ .event = .{ .key = .{ .cp = pardes.Key.enter } } });
    for (0..Harness.rounds) |_| {
        try h.pump();
        try client.wait(5);
        while (try client.next()) |_| {}
        if (std.mem.eql(u8, pane.file.?.content, "ONE\ntwo\n")) break;
    } else return error.PipeDidNotComplete;
    try testing.expect(h.core.pipe_wait == null);
    try testing.expectEqual(@as(usize, 0), h.session.pipe_tasks.len);
    try h.pumpUntilShowsCore(&client);
}

test "detached session: two frontends share one screen at the smallest common grid" {
    var h: Harness = undefined;
    try h.init(80, 24);
    defer h.deinit();
    var a = try h.attach(80, 24);
    defer a.deinit();
    _ = try h.pumpUntil(&a, .frame);

    // A second, smaller frontend. tmux's rule: the session shrinks to what both
    // can show, because two people looking at different screens is the point of
    // a shared session lost.
    var b = try h.attach(50, 12);
    defer b.deinit();
    try testing.expectEqual(@as(u8, 1), b.slot);
    try testing.expectEqual(@as(u16, 50), b.cols);
    try testing.expectEqual(@as(u16, 12), b.rows);

    // Both are now on the 50x12 grid — and getting there IS the proof that the
    // reshape was sent as a FULL frame: `next` refuses a diff whose geometry
    // does not match the grid it holds, so a client whose shape moved can only
    // have been reshaped by a full one.
    try h.pumpUntilGrid(&a, 50, 12);
    try h.pumpUntilGrid(&b, 50, 12);
    try testing.expectEqual(@as(usize, 50 * 12), a.grid.items.len);
    // CONVERGENCE, not a snapshot. `pumpUntilGrid(&b, ...)` pumped the session
    // while draining only `b`, and the daemon owns the pane shells now: a
    // prompt arriving on a pty moves the screen between pumps, so `a` can be
    // holding an undrained frame and comparing the two grids here would compare
    // two instants. What a shared session promises is that they agree, which is
    // what this waits for.
    try h.pumpUntilSameScreen(&a, &b);

    // ...and input from EITHER moves that one screen. Both are drained on every
    // pump before they are compared: one pump sends every attached frontend a
    // frame, so a test that drains only one is comparing two instants.
    try b.send(.{ .event = .{ .command = "Look /" } });
    _ = try h.pumpUntilChange(&a);
    try h.pumpUntilSameScreen(&a, &b);
}

test "detached session: a window past the protocol attaches at the largest grid it carries" {
    var h: Harness = undefined;
    try h.init(80, 24);
    defer h.deinit();

    // A 4K display at a small font is already past `max_rows`, and until the
    // clamp in `open` that hello was a geometry the session's decoder refused:
    // `apply` answered `BadValue` with `close(.protocol)` and no `refuse`
    // behind it, so a frontend on a big screen was told the session "hung up on
    // the connect" — at a session with every slot free. It attaches now, at the
    // biggest grid the wire has.
    var c = try h.attach(wire.max_cols + 400, wire.max_rows + 70);
    defer c.deinit();
    try testing.expectEqual(wire.max_cols, c.cols);
    try testing.expectEqual(wire.max_rows, c.rows);

    // ...and the full frame that follows is the 65536-cell one whose single run
    // does not fit a u16 count (wire.zig `run_max`), which is the half of this
    // the clamp alone would have made universal rather than fixed.
    const frame = (try h.pumpUntil(&c, .frame)).frame;
    try testing.expectEqual(wire.FrameKind.full, frame.kind);
    try testing.expectEqual(@as(usize, @as(usize, wire.max_cols) * wire.max_rows), c.grid.items.len);
    // TWO runs and not one, which is the assertion that fails first if the
    // encoder's bound goes: this grid's full frame is 65536 painted cells and
    // a run counts to 65535.
    try testing.expectEqual(@as(u32, 2), frame.nruns);
    try h.pumpUntilShowsCore(&c);
}

test "detached session: a frontend that dies takes nothing with it" {
    var h: Harness = undefined;
    try h.init(60, 16);
    defer h.deinit();
    var a = try h.attach(60, 16);
    defer a.deinit();
    var b = try h.attach(60, 16);
    _ = try h.pumpUntil(&a, .frame);
    _ = try h.pumpUntil(&b, .frame);
    try testing.expect(h.session.clients[0].attached);
    try testing.expect(h.session.clients[1].attached);

    // Not a `bye`: the socket goes away under the session's feet, which is what
    // a frontend crashing or being killed looks like from here.
    b.deinit();
    _ = try h.pumpUntil(&a, .frame);
    try testing.expect(!h.session.clients[1].attached);
    try testing.expectEqual(@as(c_int, -1), h.session.clients[1].fd);
    // The survivor is still served and the core is still running.
    try testing.expect(h.session.clients[0].attached);
    try testing.expect(!h.core.quit);
    try a.send(.{ .event = .{ .command = "Look /" } });
    try testing.expect((try h.pumpUntilChange(&a)).nruns > 0);

    // The freed slot takes the next frontend, and the screen comes with it.
    var d = try h.attach(60, 16);
    defer d.deinit();
    try testing.expectEqual(@as(u8, 1), d.slot);
    try testing.expectEqual(wire.FrameKind.full, (try h.pumpUntil(&d, .frame)).frame.kind);
    // ...and it is the SAME screen the survivor is looking at. Waited for
    // rather than snapshotted, for the reason above: `pumpUntil(&d, ...)`
    // drained only `d`, and pane output means the screen does not stand still
    // between pumps.
    try h.pumpUntilSameScreen(&a, &d);
}

test "detached session: a frontend speaking another protocol is refused, loudly" {
    var h: Harness = undefined;
    try h.init(60, 16);
    defer h.deinit();

    // A raw socket rather than a `Client`, because the whole point is a peer
    // that does not agree with `wire.version` — and `open` would already have
    // sent a perfectly good hello.
    const fd = try rawConnect(&h);
    defer _ = libc.close(fd);
    var buf: [64]u8 = undefined;
    const hello = try wire.encodeClient(&buf, .{
        .hello = .{ .version = wire.version + 1, .cols = 60, .rows = 16 },
    });
    try testing.expectEqual(@as(isize, @intCast(hello.len)), libc.send(fd, hello.ptr, hello.len, nosignal));

    // Two pumps: one to accept the connection, one to read the hello and
    // answer it.
    try h.pump();
    try h.pump();
    var got: [64]u8 = undefined;
    const n = libc.read(fd, &got, got.len);
    try testing.expect(n > 0);
    const f = (try wire.framed(got[0..@intCast(n)])).?;
    try testing.expectEqual(wire.Refusal.version, (try wire.decodeServer(f.tag, f.payload)).refuse);
    // Refused means refused: the slot went back and no frame was ever sent.
    for (&h.session.clients) |*slot| try testing.expect(!slot.attached);
    try testing.expect(!h.core.quit);
}

test "detached session: a peer that sends garbage is dropped, not obeyed" {
    var h: Harness = undefined;
    try h.init(60, 16);
    defer h.deinit();
    var c = try h.attach(60, 16);
    defer c.deinit();
    _ = try h.pumpUntil(&c, .frame);

    // A well-formed frame around a tag this protocol has never defined. The
    // session must close the connection rather than guess at it.
    const junk = [_]u8{ 0xfe, 0x00, 0x00, 0x00, 0x00 };
    try testing.expectEqual(@as(isize, junk.len), libc.send(c.fd, &junk, junk.len, nosignal));
    for (0..8) |_| {
        try h.pump();
        if (!h.session.clients[0].attached) break;
    }
    try testing.expect(!h.session.clients[0].attached);
    // The session is untouched: a hostile frontend costs a slot, not a session.
    try testing.expect(!h.core.quit);
}

test "detached session: the session outlives every frontend and keeps its grid" {
    var h: Harness = undefined;
    try h.init(72, 20);
    defer h.deinit();
    {
        var c = try h.attach(40, 10);
        defer c.detach();
        _ = try h.pumpUntil(&c, .frame);
        try testing.expectEqual(@as(u16, 40), h.session.cols);
    }
    // Nobody attached. The grid stays where the last frontend left it rather
    // than collapsing: a detached session is one nobody is looking at, not one
    // of no size.
    try h.pump();
    try testing.expectEqual(@as(u16, 40), h.session.cols);
    try testing.expectEqual(@as(u16, 10), h.session.rows);
    try testing.expect(!h.core.quit);
    // ...and the next frontend takes the grid over: with one attachment the
    // smallest common grid IS that frontend's, so the session follows it up to
    // 90x30 rather than pinning the departed one's 40x10 forever.
    var again = try h.attach(90, 30);
    defer again.deinit();
    try testing.expectEqual(@as(u16, 90), again.cols);
    try testing.expectEqual(@as(u16, 30), again.rows);
    try h.pumpUntilGrid(&again, 90, 30);
    try testing.expectEqual(@as(usize, 90 * 30), again.grid.items.len);
}

test "detached session: the seam's own routing rules, per surviving effect" {
    var h: Harness = undefined;
    try h.init(60, 16);
    defer h.deinit();
    var a = try h.attach(60, 16);
    defer a.deinit();
    var b = try h.attach(60, 16);
    defer b.deinit();
    _ = try h.pumpUntil(&a, .frame);
    _ = try h.pumpUntil(&b, .frame);

    const host = h.session.host();
    // BROADCAST: the yank register is a fact about the session, so every
    // display it is being watched on gets it.
    host.vtable.set_clipboard.?(host.ctx, "yank");
    try expectBoth(&h, &a, &b, .set_clipboard);

    // Ask one frontend: two clipboard answers would paste twice.
    h.session.origin = 1;
    host.vtable.read_clipboard.?(host.ctx);
    try expectOnly(&h, &b, &a, .read_clipboard);
    // `open_link` follows the same origin: the browser that opens is the one on
    // the display of the human who clicked, not the oldest attachment's.
    host.vtable.open_link.?(host.ctx, "https://x");
    try expectOnly(&h, &b, &a, .open_link);
    // ...and with no origin it falls back to the primary, which is what a link
    // opened by something other than a keystroke gets.
    h.session.origin = 0;
    host.vtable.open_link.?(host.ctx, "https://y");
    try expectOnly(&h, &a, &b, .open_link);
}

test "detached session: a frontend is never asked to fork, write, or watch" {
    // THE INVARIANT OF THE WHOLE DETACHED DESIGN, pinned as a property of the
    // protocol rather than of one code path: a frontend is input and screen, so
    // the set of messages that can reach it is exactly the eight below. If a
    // machine-local effect is ever put back on the wire, a frontend becomes the
    // process that owns a pane's shell again — and a pane whose shell belongs to
    // a frontend dies when that frontend leaves, which is the bug this replaced.
    //
    // Adding a name here is meant to be an ARGUMENT, not a formality. The bar is
    // the one `quit` and `detach` clear and `spawn` cannot: it needs this
    // human's screen, keyboard, clipboard or browser, or it is the session
    // telling this frontend about its own membership. Anything that touches a
    // disk or a process table fails that bar by construction.
    const allowed = [_][]const u8{
        // Session control: who this connection is, and whether it is still one.
        // `detach` is this frontend leaving and `quit` is the session ending —
        // opposite meanings, same shape, and neither is an effect.
        "welcome",       "refuse",         "frame",     "quit", "detach",
        // The three that need THIS human's display and cannot be done by a
        // daemon nobody is looking at.
        "set_clipboard", "read_clipboard", "open_link",
    };

    // One: the union a frontend decodes into has no other variant. Named
    // rather than counted, so re-adding `spawn` fails with the name in it.
    const fields = @typeInfo(wire.ServerMsg).@"union".fields;
    inline for (fields) |f| {
        for (allowed) |ok| {
            if (std.mem.eql(u8, f.name, ok)) break;
        } else {
            std.debug.print("ServerMsg.{s} is not an effect a frontend may perform\n", .{f.name});
            return error.MachineLocalEffectOnTheWire;
        }
    }
    try testing.expectEqual(allowed.len, fields.len);

    // Two: and no TAG BYTE outside them decodes either — the check above is
    // about this build's union, this one is about the bytes on the socket. Every
    // other byte must be `BadTag`, including the eight that used to be defined:
    // a session built before this change cannot talk a frontend into forking.
    var accepted: usize = 0;
    for (0..256) |i| {
        const tag: u8 = @intCast(i);
        // Payloads are deliberately empty: what is asked is whether the TAG is
        // known, and every known tag fails later (`Truncated`) or succeeds, but
        // never with `BadTag`.
        if (wire.decodeServer(tag, &.{})) |_| accepted += 1 else |err| switch (err) {
            error.BadTag => continue,
            else => accepted += 1,
        }
        const named = for (allowed) |ok| {
            const want = std.meta.stringToEnum(wire.ServerTag, ok).?;
            if (@intFromEnum(want) == tag) break true;
        } else false;
        if (!named) {
            std.debug.print("tag 0x{x:0>2} is decodable by a frontend and is not one of the seven\n", .{tag});
            return error.MachineLocalEffectOnTheWire;
        }
    }
    try testing.expectEqual(allowed.len, accepted);
}

test "detached session: the client table is a refusal, not a queue" {
    // A small grid on purpose: these thirty-two peers never read, and a full
    // frame of 80x24 each would push them into the backlog rule that the next
    // test is about. 20x5 keeps every queue in the kernel's own buffer.
    var h: Harness = undefined;
    try h.init(20, 5);
    defer h.deinit();

    var fds: [server.max_clients]c_int = @splat(-1);
    defer for (fds) |fd| if (fd >= 0) {
        _ = libc.close(fd);
    };
    var buf: [64]u8 = undefined;
    const hello = try wire.encodeClient(&buf, .{ .hello = .{ .cols = 20, .rows = 5 } });
    for (&fds) |*fd| {
        fd.* = try rawConnect(&h);
        try testing.expectEqual(@as(isize, @intCast(hello.len)), libc.send(fd.*, hello.ptr, hello.len, nosignal));
    }
    // The listener's backlog is `max_clients` deep, so this takes a few rounds:
    // `accept` drains what is there each time it is woken.
    for (0..16) |_| {
        try h.pump();
        var attached: usize = 0;
        for (&h.session.clients) |*slot| if (slot.attached) {
            attached += 1;
        };
        if (attached == server.max_clients) break;
    }
    for (&h.session.clients) |*slot| try testing.expect(slot.attached);

    // The thirty-third is TOLD it does not fit, and told promptly: the
    // alternative — leaving it in the backlog — makes a level-triggered poll
    // report the listener ready forever and spins the core.
    const extra = try rawConnect(&h);
    defer _ = libc.close(extra);
    try testing.expectEqual(@as(isize, @intCast(hello.len)), libc.send(extra, hello.ptr, hello.len, nosignal));
    var got: [64]u8 = undefined;
    const refusal = for (0..8) |_| {
        try h.pump();
        const n = libc.read(extra, &got, got.len);
        if (n > 0) break got[0..@intCast(n)];
    } else return error.NeverRefused;
    const f = (try wire.framed(refusal)).?;
    try testing.expectEqual(wire.Refusal.full, (try wire.decodeServer(f.tag, f.payload)).refuse);
    // ...and the thirty-two it does serve are untouched.
    for (&h.session.clients) |*slot| try testing.expect(slot.attached);
    try testing.expect(!h.core.quit);
}

test "detached session: a frontend that stops reading is dropped, not waited for" {
    var h: Harness = undefined;
    try h.init(40, 10);
    defer h.deinit();
    var good = try h.attach(40, 10);
    defer good.deinit();

    // A peer that says hello and then never reads a byte again — a frontend
    // stopped in a debugger, or one whose terminal is blocked.
    const mute = try rawConnect(&h);
    defer _ = libc.close(mute);
    var buf: [64]u8 = undefined;
    const hello = try wire.encodeClient(&buf, .{ .hello = .{ .cols = 40, .rows = 10 } });
    try testing.expectEqual(@as(isize, @intCast(hello.len)), libc.send(mute, hello.ptr, hello.len, nosignal));
    for (0..8) |_| {
        try h.pump();
        if (h.session.clients[1].attached) break;
    }
    try testing.expect(h.session.clients[1].attached);

    // Clipboard updates broadcast to every frontend.
    const text = try testing.allocator.alloc(u8, 256 * 1024);
    defer testing.allocator.free(text);
    @memset(text, 'y');
    const host = h.session.host();
    for (0..24) |_| {
        if (!h.session.clients[1].attached) break;
        host.vtable.set_clipboard.?(host.ctx, text);
        // Read `good` back to EMPTY before the next mirror, rather than
        // pumping once and taking whatever one write fitted. One pump moves at
        // most one socket buffer, and that buffer is 8 KiB here
        // (`net.local.stream.sendspace` on Darwin) against Linux's 208 KiB —
        // so a single pump per 256 KiB mirror leaves the READING frontend
        // falling behind by a quarter megabyte a round and closes it for a
        // backlog it never caused. The claim under test is that a frontend
        // that reads survives, so it has to actually finish reading.
        for (0..256) |_| {
            try h.pump();
            try good.wait(5);
            while (try good.next()) |_| {}
            if (h.session.clients[0].out.items.len == 0) break;
        }
    }
    // Dropped rather than queued without bound, and rather than the core
    // blocking on it.
    try testing.expect(!h.session.clients[1].attached);
    try testing.expectEqual(@as(c_int, -1), h.session.clients[1].fd);
    // The frontend that WAS reading is still attached and still being drawn
    // for, which is the whole claim: one slow peer costs its own slot.
    try testing.expect(h.session.clients[0].attached);
    try testing.expect(!h.core.quit);
    try good.send(.{ .event = .{ .command = "Msg still here" } });
    try testing.expect((try h.pumpUntilChange(&good)).nruns > 0);
}

/// A connected socket with nothing said on it yet, for the tests whose peer is
/// deliberately not a `Client`: one that speaks another protocol, thirty-two
/// that fill the table, one that never reads.
fn rawConnect(h: *Harness) !c_int {
    var path_buf: [server.path_max]u8 = undefined;
    const path = server.sessionPath(&path_buf, h.name).?;
    var addr: libc.sockaddr.un = .{ .path = @splat(0) };
    @memcpy(addr.path[0 .. path.len + 1], path[0 .. path.len + 1]);
    const fd = libc.socket(libc.AF.UNIX, libc.SOCK.STREAM, 0);
    if (fd < 0) return error.NoSocket;
    errdefer _ = libc.close(fd);
    if (libc.connect(fd, @ptrCast(&addr), @sizeOf(@TypeOf(addr))) != 0) return error.NoSession;
    return fd;
}

/// `want` reached `to` and nothing reached `other`.
fn expectOnly(
    h: *Harness,
    to: *Client,
    other: *Client,
    comptime want: std.meta.Tag(wire.ServerMsg),
) !void {
    _ = try h.pumpUntil(to, want);
    try other.wait(5);
    while (try other.next()) |msg| if (std.meta.activeTag(msg) == want) {
        std.debug.print("{t} reached a frontend it was not routed to\n", .{want});
        return error.Misrouted;
    };
}

fn expectBoth(
    h: *Harness,
    a: *Client,
    b: *Client,
    comptime want: std.meta.Tag(wire.ServerMsg),
) !void {
    _ = try h.pumpUntil(a, want);
    _ = try h.pumpUntil(b, want);
}

/// Are these two grids showing the same thing? `visuallyEqual` and not
/// `std.meta.eql`, because `Cell.text` past `len` is scratch the decoder does
/// not invent — pardes.zig says in as many words that it must never
/// manufacture a difference.
fn sameScreen(want: []const pardes.Cell, have: []const pardes.Cell) bool {
    if (want.len != have.len) return false;
    for (want, have) |*x, *y| if (!x.visuallyEqual(y)) return false;
    return true;
}

fn expectSameScreen(want: []const pardes.Cell, have: []const pardes.Cell) !void {
    try testing.expectEqual(want.len, have.len);
    for (want, have, 0..) |*x, *y, i| if (!x.visuallyEqual(y)) {
        std.debug.print("cell {d} differs\n", .{i});
        return error.CellMismatch;
    };
}