a73x

src/server/server_test_harness.zig

Ref:   Size: 33.8 KiB   History

const std = @import("std");
const Grid = @import("term").grid.Grid;
pub const Pty = @import("pty").Pty;
const proto = @import("term").protocol;
const replica_mod = @import("term").replica;
const quic = @import("quic");
const quic_server = @import("quic_server.zig");
const TmpDir = @import("testtmp").TmpDir;
/// Re-exported so the eight sibling test files spell the module name once,
/// here, and reach it as `h.dial` — the same connect-a-daemon primitive the
/// CLI client and `mux a` dial through, so these tests hold the socket the
/// way the product does rather than hand-rolling the pair of calls.
pub const dial = @import("dial");
/// The awaits below are wrappers on `Link.awaitFrame` rather than their own
/// poll+readFrame loops, so a test waiting on a daemon frame waits the way
/// the client and `mux a` do. Re-exported like `dial` above: a sibling
/// spelling its own `Sink` needs the type, not a second import edge.
pub const link = @import("link");
const srv_mod = @import("server.zig");
const Server = srv_mod.Server;

pub fn serverThread(srv: *Server, stop: *std.atomic.Value(bool)) void {
    while (!stop.load(.acquire)) {
        srv.pumpOnce(50) catch break;
    }
}

/// Pump the daemon until pred says the world arrived, or the deadline says
/// it never will. Returns whether pred fired, so a test asserts the
/// CONDITION and its failure names what didn't happen — not a guessed
/// round count. A false return is an assertable value: the deadline turns
/// a wedge into a legible failure instead of a silent hang (a wedged zig
/// test prints nothing).
pub fn pumpUntil(
    srv: *Server,
    deadline_ms: u64,
    ctx: anytype,
    comptime pred: fn (@TypeOf(ctx)) bool,
) !bool {
    var left = deadline_ms;
    while (true) {
        if (pred(ctx)) return true;
        if (left == 0) return false;
        try srv.pumpOnce(5);
        left -|= 5;
    }
}

/// One client slot on a daemon, as a `pumpUntil` context. Here rather than
/// in a test file because "the client is seated" and "the client has
/// offered an agent" are the two conditions most of the sibling files open
/// with, and a predicate copied per file is a predicate that drifts per file.
pub const ClientSlot = struct {
    srv: *Server,
    n: usize,

    pub fn seated(s: ClientSlot) bool {
        return s.srv.clients[s.n] != null;
    }

    /// False rather than a panic on an empty slot: a wait for the offer can
    /// start before the accept lands, and "not yet" is the honest answer.
    pub fn offering(s: ClientSlot) bool {
        const c = s.srv.clients[s.n] orelse return false;
        return c.agent_offer;
    }
};

/// Which of two client slots is the latest-active one. STRICTLY ahead, not
/// merely tied: the rule under test is "the most recently active client
/// answers", and equal activity would let a wait finish before the lead had
/// actually changed hands.
pub const Lead = struct {
    srv: *Server,
    ahead: usize,
    behind: usize,

    pub fn taken(self: Lead) bool {
        const a = self.srv.clients[self.ahead] orelse return false;
        const b = self.srv.clients[self.behind] orelse return false;
        return a.activity > b.activity;
    }
};

/// A daemon on a socket of its own, which is what nearly every test in this
/// folder opens with: a short-path temp directory (`testtmp`, because a unix
/// socket path caps at 108 bytes), a socket named inside it, and a `Server`
/// bound to that socket. It owns all three and takes them down in the order
/// they have to come down in.
///
/// The pieces are still reachable — `td.srv` for the daemon, `td.sock_path`
/// for a dial, `td.tmp` for the tests that put an rc file, a key or an agent
/// socket in the same directory.
pub const TestDaemon = struct {
    alloc: std.mem.Allocator,
    tmp: TmpDir,
    /// Owned here, not by the caller: `Server` KEEPS both of these slices
    /// (`opts.sock_path` for the unlink at teardown, `opts.shell` for the
    /// manifest and for the next session's spawn), so both have to outlive
    /// `srv.deinit()`. A test that allocated them itself would free them on
    /// its own defer, which runs first.
    sock_path: []u8,
    shell: [:0]u8,
    /// Valid once `start` has run. `open` leaves it undefined on purpose so
    /// that `td.srv` is the spelling everywhere, rather than `td.srv.?`.
    srv: Server,
    started: bool = false,
    stop: std.atomic.Value(bool) = .init(false),
    thread: ?std.Thread = null,

    /// `Server.Options` minus `sock_path`, which is this fixture's to name.
    /// Every other field is passed through rather than flattened: the tests
    /// that ask for shell integration, a non-default grid or a version string
    /// are asking for behaviour, not for convenience.
    pub const Opts = struct {
        shell: [:0]const u8,
        cols: u16 = 80,
        rows: u16 = 24,
        shell_integration: bool = false,
        extra_env: []const Pty.EnvPair = &.{},
        version: []const u8 = "",
    };

    /// The directory and the socket path, with no daemon on them yet. For the
    /// tests whose session shell is a script they have to write INTO that
    /// directory first: they `open`, write the script, then `start` with it.
    /// `init` is the same two steps for everyone else.
    pub fn open(alloc: std.mem.Allocator, tag: []const u8) !TestDaemon {
        var tmp = try TmpDir.make();
        errdefer tmp.cleanup();
        // After the move into the returned value would be wrong: `TmpDir.path`
        // points into the struct's own buffer, so the slice is only good at the
        // address it is read from. This is a copy, which travels.
        const sock_path = try std.fmt.allocPrint(alloc, "{s}/{s}.sock", .{ tmp.path(), tag });
        return .{
            .alloc = alloc,
            .tmp = tmp,
            .sock_path = sock_path,
            .shell = try alloc.dupeZ(u8, ""),
            .srv = undefined,
        };
    }

    pub fn start(self: *TestDaemon, opts: Opts) !void {
        self.alloc.free(self.shell);
        self.shell = try self.alloc.dupeZ(u8, opts.shell);
        self.srv = try Server.init(self.alloc, .{
            .sock_path = self.sock_path,
            .shell = self.shell,
            .cols = opts.cols,
            .rows = opts.rows,
            .shell_integration = opts.shell_integration,
            .extra_env = opts.extra_env,
            .version = opts.version,
        });
        self.started = true;
    }

    pub fn init(alloc: std.mem.Allocator, tag: []const u8, opts: Opts) !TestDaemon {
        var self = try open(alloc, tag);
        errdefer self.deinit();
        try self.start(opts);
        return self;
    }

    /// `Opts` minus `shell`, which the stubborn doors supply: naming a shell
    /// there is the one thing a stubborn-shell test must not do. One field
    /// today rather than a copy of the whole list — a test that needs a grid
    /// size or an env pair on a stubborn shell adds it here, in the one place
    /// both doors read.
    pub const StubbornOpts = struct { version: []const u8 = "" };

    /// The DEFAULT session's door. Writes the stubborn shell, starts the
    /// daemon on it, and does not return until that shell's traps are armed.
    ///
    /// The wait belongs to the door and not to the tests, because five of them
    /// needed it and the two that spelled it by hand both got it wrong: first
    /// by not waiting at all, then by waiting on the wrong session. A test
    /// that comes through here cannot make either mistake, because it never
    /// names a session or a slot.
    pub fn startStubborn(self: *TestDaemon, alloc: std.mem.Allocator, opts: StubbornOpts) !void {
        const script = try writeStubbornShell(alloc, &self.tmp);
        // Freed here on purpose: `start` dupes it into `self.shell`, which is
        // what the Server keeps.
        defer alloc.free(script);
        try self.start(.{ .shell = script, .version = opts.version });
        try awaitStubbornArmed(&self.srv, &self.tmp, "", stubborn_arm_ms);
    }

    /// A NAMED session's door: birth it by attaching, wait for the snapshot
    /// that says the daemon seated the client, then wait for THAT session's
    /// own shell to arm. A different shell with a different pid from the
    /// default session's, which is exactly what a hand-written wait got
    /// wrong.
    ///
    /// Pumps, through `awaitFrame`, so it is for the tests that drive
    /// `pumpOnce` themselves — which is every caller of the stubborn shell.
    pub fn attachStubborn(
        self: *TestDaemon,
        alloc: std.mem.Allocator,
        name: []const u8,
        cols: u16,
        rows: u16,
    ) !std.net.Stream {
        const c = try dial.dialAttachNamed(self.sock_path, cols, rows, name);
        errdefer c.close();
        (try awaitFrame(alloc, &self.srv, c.handle, .snapshot, 400) orelse
            return error.NoState).deinit(alloc);
        try awaitStubbornArmed(&self.srv, &self.tmp, name, stubborn_arm_ms);
        return c;
    }

    /// Run the daemon's pump on a thread, for the tests that talk to it over a
    /// real socket instead of driving `pumpOnce` themselves.
    pub fn threaded(self: *TestDaemon) !void {
        self.thread = try std.Thread.spawn(.{}, serverThread, .{ &self.srv, &self.stop });
    }

    /// Take the daemon down but KEEP the directory, for the tests whose
    /// question is what the teardown left on disk — a shim directory removed,
    /// a planted file not removed. Those have to ask after the server is gone
    /// and while its directory is still there, so they scope this and let the
    /// `deinit` behind it do the rest; it is idempotent for that reason.
    pub fn shutdown(self: *TestDaemon) void {
        // Raise the flag BEFORE joining. `serverThread` only leaves its loop
        // when it reads the flag, so a join that comes first waits on a thread
        // nobody has asked to stop, and the test hangs until the runner is
        // killed. Written as two statements in one place because as two
        // `defer`s it was a declaration-order puzzle re-solved at every test
        // that spawned a thread.
        if (self.thread) |t| {
            self.stop.store(true, .release);
            t.join();
            self.thread = null;
        }
        if (self.started) {
            self.srv.deinit();
            self.started = false;
        }
    }

    pub fn deinit(self: *TestDaemon) void {
        self.shutdown();
        self.alloc.free(self.shell);
        self.alloc.free(self.sock_path);
        self.tmp.cleanup();
    }
};

/// Bring a replica grid up to date with one daemon frame. The replay lives in
/// replica.zig — the production client's — so these tests replay through the
/// code the client ships and not a hand-rolled twin.
pub fn applyFrame(alloc: std.mem.Allocator, replica: *Grid, frame: proto.Frame) !void {
    if (frame.type != .snapshot and frame.type != .delta) return;
    var r = replica_mod.Replica.init(alloc, replica);
    // The old helper propagated a garbled delta as BadPayload; .resync is
    // that same event with the production name on it.
    if (try r.apply(frame.type, frame.payload) == .resync) return error.BadPayload;
}

/// A copy of `text` with each row's trailing spaces removed.
///
/// The one formatting difference between a daemon's `Engine.dumpPlain` and a
/// client's `Grid.dumpPlain`: ghostty dumps with trimming off, so a space a
/// program wrote at the end of a row survives it, while a client never holds
/// one — the encoder stops a row at its last non-blank cell, and the VT
/// formatter that fed the old wire trimmed trailing whitespace in the same
/// place. `test/e2e_lib.sh assert_ws_converged` strips it on both sides for
/// the same reason. A convergence pin compares the daemon's dump through
/// this, and nothing else about the two dumps may differ.
pub fn trimRowTails(alloc: std.mem.Allocator, text: []const u8) ![]u8 {
    var out: std.ArrayList(u8) = .empty;
    errdefer out.deinit(alloc);
    var it = std.mem.splitScalar(u8, text, '\n');
    var first = true;
    while (it.next()) |line| {
        if (!first) try out.append(alloc, '\n');
        first = false;
        try out.appendSlice(alloc, std.mem.trimRight(u8, line, " "));
    }
    return out.toOwnedSlice(alloc);
}

/// A connected pair of unix stream sockets, standing in for an attached client.
/// `daemon` goes in a client slot; `peer` is the client's end.
///
/// SOCKETS rather than pipes because the client send path is `send(2)`, which on
/// a pipe fd fails ENOTSOCK — and `std.posix.send` maps that errno to
/// `unreachable`, so it panics rather than returning an error.
pub const SockPair = struct { daemon: std.posix.fd_t, peer: std.posix.fd_t };

pub fn connectedPair() !SockPair {
    var fds: [2]std.posix.fd_t = undefined;
    const rc = std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &fds);
    if (rc != 0) return std.posix.unexpectedErrno(std.posix.errno(rc));
    return .{ .daemon = fds[0], .peer = fds[1] };
}

/// Test helper: the first snapshot-or-delta frame to arrive, reduced to what
/// the resync tests assert on. `epoch` is 0 for a delta — only snapshots
/// carry the session epoch.
pub const StateFrame = struct { type: proto.MsgType, seq: u64, epoch: u64 };

/// Either state frame answers, so the wait names `.snapshot` as its `want`
/// and catches `.delta` in the sink. The sink copies the header's ints out
/// and ends the wait with `error.StateFrameSeen`, which is this function's
/// success rather than a failure — a Sink borrows its frame, so nothing may
/// be kept but the decoded numbers.
pub fn firstStateFrame(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64) !?StateFrame {
    const Catch = struct {
        got: StateFrame = undefined,
        fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
            if (frame.type != .delta) return;
            const self: *@This() = @ptrCast(@alignCast(ctx.?));
            const hdr = try proto.readDeltaHeader(frame.payload);
            self.got = .{ .type = .delta, .seq = hdr.seq, .epoch = 0 };
            return error.StateFrameSeen;
        }
    };
    var caught: Catch = .{};
    // Not `l.close()` on any path: the caller owns this fd and closes it.
    var l: link.Link = .{ .fd = fd };
    const frame = l.awaitFrame(alloc, .snapshot, @intCast(timeout_ms), .{
        .ctx = &caught,
        .on = Catch.on,
    }) catch |err| switch (err) {
        error.StateFrameSeen => return caught.got,
        // A peer that went away before any state frame reads the same as one
        // that never sent one, which is what every caller asserts on.
        error.Closed => return null,
        else => return err,
    } orelse return null;
    defer frame.deinit(alloc);
    const p = try proto.readSnapshotPrefix(frame.payload);
    return .{ .type = .snapshot, .seq = p.seq, .epoch = p.epoch };
}

// ---------------------------------------------------------------------------
// QUIC integration tests: that a QUIC client still gets the shell's exit code,
// and that one client leaving does not take the others with it.
// ---------------------------------------------------------------------------

/// Bring a Server up with a QUIC listener bound to an ephemeral loopback
/// port, and hand back the address a client should dial.
pub fn quicTestServer(srv: *Server, key: quic.Key) !struct { l: *quic_server.Listener, addr: std.net.Address } {
    const bind = try std.net.Address.parseIp("127.0.0.1", 0);
    const l = try quic_server.Listener.init(srv.alloc, bind, key, srv.quicHandler(), 5000);
    srv.attachQuic(l);
    var actual: std.posix.sockaddr.storage = undefined;
    var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
    try std.posix.getsockname(l.pollFd(), @ptrCast(&actual), &len);
    return .{ .l = l, .addr = std.net.Address.initPosix(@ptrCast(@alignCast(&actual))) };
}

/// Single-threaded: the daemon's own pump is what services the listener, which
/// is the integration under test.
///
/// `pumpUntil` with a body, and it keeps the body rather than becoming a
/// `pumpUntil` call: every round has to give each QUIC peer a turn to read
/// and acknowledge, or the daemon's egress never drains and `done` can
/// never come true.
pub fn quicPump(
    srv: *Server,
    clients: []*quic_server.TestPeer,
    budget_ms: u64,
    ctx: anytype,
    done: *const fn (@TypeOf(ctx)) bool,
) !void {
    var waited: u64 = 0;
    while (waited < budget_ms) {
        if (done(ctx)) return;
        try srv.pumpOnce(5);
        for (clients) |cl| {
            var pfd = [_]std.posix.pollfd{.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 }};
            _ = std.posix.poll(&pfd, 1) catch 0;
            cl.drain();
        }
        waited += 6;
    }
}

/// Find a frame of `want` in a client's received bytes, returning its
/// payload. The client is handed a byte stream, so this does the same
/// framing walk a real client would.
pub fn findFrame(bytes: []const u8, want: proto.MsgType) ?[]const u8 {
    var off: usize = 0;
    while (off + 5 <= bytes.len) {
        const len = std.mem.readInt(u32, bytes[off + 1 ..][0..4], .little);
        if (off + 5 + len > bytes.len) return null;
        if (@as(proto.MsgType, @enumFromInt(bytes[off])) == want) {
            return bytes[off + 5 ..][0..len];
        }
        off += 5 + len;
    }
    return null;
}

// ---------------------------------------------------------------------------
// endpoint_req: the lazy QUIC bind. Two tests on purpose — `Listener` keeps a
// process-global "one at a time" latch, so every binding test must give it back
// and fewer tests is fewer places that can fail to.
// ---------------------------------------------------------------------------

/// The daemon here has no thread of its own: nothing arrives on `fd` unless
/// this loop pumps it, so the wait is `iters` short `Link.awaitFrame` waits
/// with a pump between them rather than one long one. 2 ms rather than 1 so
/// that each turn always reaches its poll — `awaitFrame` re-reads the clock
/// before polling, and a 1 ms deadline can already be spent by then.
pub fn awaitFrame(
    alloc: std.mem.Allocator,
    srv: *Server,
    fd: std.posix.fd_t,
    want: proto.MsgType,
    iters: usize,
) !?proto.Frame {
    return awaitFrameSink(alloc, srv, fd, want, iters, .{});
}

/// `awaitFrame` for the callers that assert on what arrived BEFORE the
/// match — a delta where only a snapshot may cross, a frame after a
/// goodbye. Same borrow rule as `awaitFrameOnSink`.
pub fn awaitFrameSink(
    alloc: std.mem.Allocator,
    srv: *Server,
    fd: std.posix.fd_t,
    want: proto.MsgType,
    iters: usize,
    sink: link.Sink,
) !?proto.Frame {
    // Not `l.close()` on any path: the caller owns this fd and closes it.
    var l: link.Link = .{ .fd = fd };
    var i: usize = 0;
    while (i < iters) : (i += 1) {
        try srv.pumpOnce(5);
        // Anything else on the way (a snapshot for the attached half of this
        // test) is not what was asked for; the default sink drops it.
        const got = l.awaitFrame(alloc, want, 2, sink) catch |err| switch (err) {
            error.Closed => return null,
            else => return err,
        };
        if (got) |frame| return frame;
    }
    return null;
}

/// Threaded twin of `awaitFrame`: polls `fd` until a frame of type `want`
/// arrives or `timeout_ms` of wall clock passes. For the tests whose daemon
/// runs on `serverThread` — `awaitFrame` pumps the server itself, so a test
/// that has a thread doing that would be double-pumping.
///
/// The budget is real elapsed time, read off `milliTimestamp`. The loops this
/// replaces spelled theirs as `deadline_ms -|= 100` once per `poll`, which
/// charges the full 100 ms even when the poll returned instantly because a
/// frame was already waiting: a socket with traffic on it spent a "5000 ms"
/// budget in fifty near-instant turns. Those budgets were frame counts wearing
/// a millisecond's name, and a slow box shortened them further rather than
/// giving them more time.
///
/// Non-matching frames are dropped, the same as `awaitFrame` drops them, so
/// this is only right for "the first `want` to arrive"; a caller that asserts
/// on what came before it wants `awaitFrameOnSink`. Null means the budget ran
/// out or the peer closed; the caller is the one that knows which of those is
/// a failure.
pub fn awaitFrameOn(
    alloc: std.mem.Allocator,
    fd: std.posix.fd_t,
    want: proto.MsgType,
    timeout_ms: i64,
) !?proto.Frame {
    return awaitFrameOnSink(alloc, fd, want, timeout_ms, .{});
}

/// `awaitFrameOn` for the callers that DO assert on what came before the
/// match — a count of scrollback frames, the size of a snapshot that should
/// not have arrived. The sink BORROWS each non-matching frame (see
/// `link.Sink`), so anything kept has to be copied out of it, and an error
/// out of the sink ends the wait with that error.
pub fn awaitFrameOnSink(
    alloc: std.mem.Allocator,
    fd: std.posix.fd_t,
    want: proto.MsgType,
    timeout_ms: i64,
    sink: link.Sink,
) !?proto.Frame {
    // Not `l.close()` on any path: the caller owns this fd and closes it.
    var l: link.Link = .{ .fd = fd };
    return l.awaitFrame(alloc, want, @intCast(@max(timeout_ms, 0)), sink) catch |err| switch (err) {
        // The budget running out and the peer going away were both null in
        // the hand-rolled loop this replaces, and every caller reads null as
        // "no such frame arrived".
        error.Closed => null,
        else => err,
    };
}

/// A frame type the daemon never SENDS. Naming it as `want` is how a wait
/// whose answer comes out of the SINK says so: nothing can match, so every
/// frame reaches the sink and the wait ends on the sink's error or on the
/// deadline. `.attach` is a client's first word on the wire — the daemon
/// reads them and writes none.
pub const never_from_daemon: proto.MsgType = .attach;

/// What `awaitReplicaText` is waiting for on one connection.
pub const ReplicaWait = struct {
    /// Replayed through `applyFrame`, so through the production replica.
    replica: *Grid,
    /// The text the grid must show. Found in `dumpPlain`, not in the frame
    /// bytes: a row can arrive spread over several deltas, and the grid is
    /// what a user would see.
    needle: []const u8,
    /// Frame types whose mere ARRIVAL fails the wait — a snapshot where the
    /// tracker owed a delta, a scrollback chunk on a client that asked for
    /// none. `error.RejectedFrameArrived` rather than a per-caller error
    /// name, because the type is in the trace either way.
    reject: []const proto.MsgType = &.{},
};

/// The sink behind `awaitReplicaText`, public because the daemon is
/// sometimes threaded and sometimes pumped by the test, and only the wait
/// around this differs between those two.
pub const ReplicaFeed = struct {
    alloc: std.mem.Allocator,
    w: ReplicaWait,

    pub fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
        const self: *@This() = @ptrCast(@alignCast(ctx.?));
        for (self.w.reject) |bad| {
            if (frame.type == bad) {
                // Unsolicited position metadata accompanies every grid. Only
                // a correlated copy reply would leak another client's read.
                if (bad == .selection_reply and (try proto.decodeSelectionReply(frame.payload)).id == 0) continue;
                return error.RejectedFrameArrived;
            }
        }
        // applyFrame ignores everything that is not state, so the stream's
        // replies and marks pass through untouched.
        try applyFrame(self.alloc, self.w.replica, frame);
        const plain = try self.w.replica.dumpPlain(self.alloc);
        defer self.alloc.free(plain);
        if (std.mem.indexOf(u8, plain, self.w.needle) != null) return error.TextArrived;
    }

    pub fn sink(self: *ReplicaFeed) link.Sink {
        return .{ .ctx = self, .on = ReplicaFeed.on };
    }
};

/// Replay one connection's frames into a replica until its grid shows the
/// text, or the budget runs out. The loop this replaces was written out
/// eight times in server_test_attach.zig alone, each copy re-deciding what
/// a poll timeout meant.
pub fn awaitReplicaText(
    alloc: std.mem.Allocator,
    fd: std.posix.fd_t,
    budget_ms: i64,
    w: ReplicaWait,
) !bool {
    var feed: ReplicaFeed = .{ .alloc = alloc, .w = w };
    _ = awaitFrameOnSink(alloc, fd, never_from_daemon, budget_ms, feed.sink()) catch |err| switch (err) {
        error.TextArrived => return true,
        else => return err,
    };
    return false;
}

/// True when the peer closed inside the budget. The CLOSE is the answer
/// here, not a frame — `Link.awaitFrame`'s `error.Closed`, which the
/// helpers above fold into a null because for them a gone peer and a quiet
/// one are the same non-answer. For a test asserting that the daemon hung
/// up, they are the opposite of each other. Anything that arrives first
/// goes to `sink`, which is where a caller says whether a frame after the
/// goodbye is a failure.
pub fn awaitClosed(alloc: std.mem.Allocator, fd: std.posix.fd_t, budget_ms: i64, sink: link.Sink) !bool {
    var l: link.Link = .{ .fd = fd };
    _ = l.awaitFrame(alloc, never_from_daemon, @intCast(@max(budget_ms, 0)), sink) catch |err| switch (err) {
        error.Closed => return true,
        else => return err,
    };
    return false;
}

/// `awaitClosed` for a daemon with no thread of its own: nothing reaches
/// the fd, close included, unless this loop pumps it.
pub fn pumpUntilClosed(
    alloc: std.mem.Allocator,
    srv: *Server,
    fd: std.posix.fd_t,
    deadline_ms: u64,
    sink: link.Sink,
) !bool {
    var left = deadline_ms;
    while (true) {
        if (try awaitClosed(alloc, fd, 2, sink)) return true;
        if (left == 0) return false;
        try srv.pumpOnce(5);
        left -|= 5;
    }
}

/// The liveness half: "no mark arrived" is worthless against a shell that
/// never started.
pub fn awaitGridText(
    alloc: std.mem.Allocator,
    srv: *Server,
    needle: []const u8,
    budget_ms: i64,
) !bool {
    const deadline = std.time.milliTimestamp() + budget_ms;
    while (std.time.milliTimestamp() < deadline) {
        try srv.pumpOnce(5);
        const grid = try srv.sessions.table[0].?.eng.dumpPlain(alloc);
        defer alloc.free(grid);
        if (std.mem.indexOf(u8, grid, needle) != null) return true;
    }
    return false;
}

// ---------------------------------------------------------------------------
// Named sessions: attach-or-create, and every broadcast staying home. The
// default session is created by init in slot 0, and a session only leaves when
// its shell exits. Every boundary assertion has a LIVENESS half first, or a
// leak test against a silent session passes vacuously.
// ---------------------------------------------------------------------------

/// Attach `fd` to session `name` at the given size, holding nothing —
/// the way a fresh client spells it.
pub fn attachNamed(fd: std.posix.fd_t, cols: u16, rows: u16, name: []const u8) !void {
    var buf: [proto.attach_max_len]u8 = undefined;
    try proto.writeFrame(fd, .attach, proto.encodeAttachNamed(&buf, cols, rows, 0, 0, name));
}

/// The shell EXITS instead of sleeping, so a session dies through `reap` rather
/// than `Server.deinit` — the tests above reach only deinit. The second `read`
/// makes the death triggerable on its own, since the slots must be seen FULL
/// first.
pub fn writeDyingGapShell(alloc: std.mem.Allocator, tmp: *TmpDir) ![:0]u8 {
    try tmp.dir.writeFile(.{
        .sub_path = "gapdie.sh",
        .data =
        \\#!/bin/sh
        \\read -r go
        \\printf 'gap-open\n'
        \\printf '\033]52;c;c2Vjb25k\007\007'
        \\printf 'after-osc'
        \\read -r die
        \\exit 0
        \\
        ,
        .flags = .{ .mode = 0o755 },
    });
    return std.fmt.allocPrintSentinel(alloc, "{s}/gapdie.sh", .{tmp.path()}, 0);
}

/// A shell that IGNORES both signals `Pty.requestExit` has and never reads
/// stdin again, so closing the master cannot end it either — SIGKILL is the
/// only thing left that can. What a bounded end has to survive.
fn writeStubbornShell(alloc: std.mem.Allocator, tmp: *TmpDir) ![:0]u8 {
    // Blocked in `open(2)` on a fifo nobody writes: no child to orphan, no spin.
    // A `sleep` loop leaves a `sleep` behind for up to its full second (measured:
    // 791 ms past the shell's SIGKILL), and looping on `read` off the closed
    // master burns a whole core on EOF (measured: 99 ticks/s).
    //
    // The loop stays as the fallback so a box that cannot mkfifo still gets a
    // stubborn shell: a fixture that quietly exited here would let the
    // bounded-end gates pass for the wrong reason.
    //
    // The file it touches on line three is how a caller knows the traps are
    // ARMED, and `awaitStubbornArmed` is the wait. Nothing else announces it:
    // this shell prints nothing by design, so a test that ended it the
    // instant it was spawned was racing the shell's own startup, and the
    // SIGTERM `requestExit` sends landed while TERM was still fatal. The
    // name carries `$$` because a test may run several of these at once and
    // a shared marker would be answered by whichever armed first.
    const body = try std.fmt.allocPrint(alloc,
        \\#!/bin/sh
        \\trap '' TERM HUP
        \\: > "{[d]s}/stubborn-$$.armed"
        \\if mkfifo "{[d]s}/stubborn.fifo" 2>/dev/null; then
        \\  read x < "{[d]s}/stubborn.fifo"
        \\fi
        \\while :; do sleep 1; done
        \\
    , .{ .d = tmp.path() });
    defer alloc.free(body);
    try tmp.dir.writeFile(.{
        .sub_path = "stubborn.sh",
        .data = body,
        .flags = .{ .mode = 0o755 },
    });
    return std.fmt.allocPrintSentinel(alloc, "{s}/stubborn.sh", .{tmp.path()}, 0);
}

/// Wait until the stubborn shell behind the session `wire_name` spells has
/// installed the traps that make it stubborn, or fail saying it never did.
///
/// Private, and reached only through `TestDaemon.startStubborn` and
/// `TestDaemon.attachStubborn`, which is the whole point: each door knows
/// which session it just created, so no test picks the name and none can pick
/// the wrong one. Both spellings of getting it wrong by hand have already
/// happened — no wait at all, then a wait on the default session while the
/// test ended a named one. The second was the more dangerous, because a "nag"
/// that dies to the first pre-trap SIGTERM satisfies `expect(!alive(pid))`
/// without the SIGKILL deadline that test is named for ever being reached.
/// `""` is the default session here, since `sessions.find` resolves it.
///
/// The wait has to happen at all because `Pty.requestExit` closes the master
/// (SIGHUP) and
/// sends SIGTERM; the script ignores both ONCE its `trap` line has run, and
/// dies to either before that. So a shell spawned and ended within the same
/// millisecond was a coin toss, and the two systems do not toss the same
/// coin: Linux's /bin/sh is dash and arms in 2 ms or less, macOS's is bash
/// 3.2 and took 12 to 21 ms across five spawns (measured 2026-09-04 on macOS
/// 26 under `make check`, whose parallel doc and format steps are load that
/// `zig build test` does not have).
///
/// What that bought was a silent wrong answer rather than a failure: the
/// upgrade test's session was reaped 20 ms after its accepted end, so
/// `validateUpgrade` found no session mid-hangup and ACCEPTED an upgrade the
/// daemon must refuse — the one thing that test exists to prevent, on the
/// only OS where the race was reliably lost.
///
/// It polls a file rather than the pump, because the shell writes that file
/// on its own and a daemon on a thread would make pumping here a race.
/// Long enough that no loaded machine can trip it, short enough to be an
/// answer: the arming itself took 21 ms at worst across the measurements
/// above.
const stubborn_arm_ms: i64 = 5000;

fn awaitStubbornArmed(srv: *Server, tmp: *TmpDir, wire_name: []const u8, budget_ms: i64) !void {
    const si = srv.sessions.find(wire_name) orelse return error.NoSuchSessionToArm;
    const s = srv.sessions.table[si] orelse return error.NoSuchSessionToArm;
    var name_buf: [64]u8 = undefined;
    const name = try std.fmt.bufPrint(&name_buf, "stubborn-{d}.armed", .{s.pty.child});
    const deadline = std.time.milliTimestamp() + budget_ms;
    while (std.time.milliTimestamp() < deadline) {
        if (tmp.dir.access(name, .{})) |_| return else |_| {}
        std.Thread.sleep(2 * std.time.ns_per_ms);
    }
    return error.StubbornShellNeverArmed;
}

// ---------------------------------------------------------------------------
// The harness's own primitive. `pumpUntil` is what the sibling files assert
// their daemon-side conditions through, so its two outcomes are pinned here
// rather than inferred from a suite that happens to pass.
// ---------------------------------------------------------------------------

const Preds = struct {
    fn hasAnyClient(srv: *Server) bool {
        for (srv.clients) |slot| if (slot != null) return true;
        return false;
    }
    fn never(srv: *Server) bool {
        _ = srv;
        return false;
    }
};

test "pumpUntil: a condition the daemon reaches returns true, one it never does returns false" {
    const alloc = std.testing.allocator;
    var td = try TestDaemon.init(alloc, "pumpuntil", .{ .shell = "/bin/cat" });
    defer td.deinit();

    // Nothing has dialled yet, so the predicate must be false BEFORE the
    // dial — otherwise the true below would prove nothing about pumping.
    try std.testing.expect(!Preds.hasAnyClient(&td.srv));
    const c = try dial.dialAttachNamed(td.sock_path, 80, 24, "");
    defer c.close();
    try std.testing.expect(try pumpUntil(&td.srv, 2000, &td.srv, Preds.hasAnyClient));

    // The deadline is the whole point: a predicate that can never fire has to
    // come back as an assertable false, in about the time asked for, rather
    // than wedging the runner. Only the lower bound is asserted — a loaded
    // box makes the upper one flaky, and overshooting is not the failure
    // this guards against.
    const t0 = std.time.milliTimestamp();
    try std.testing.expect(!try pumpUntil(&td.srv, 200, &td.srv, Preds.never));
    try std.testing.expect(std.time.milliTimestamp() - t0 >= 200);
}