49996784

feat: latest-wins follows input activity, matching the cited tmux semantics

a73x 2026-08-08 14:08

Commit message

feat: latest-wins follows input activity, matching the cited tmux semantics

README.md

Old		New
@@ -39,7 +39,8 @@ frame-agnostic byte pump that contains no protocol knowledge at all.
39	WAN link (see the script header for the variables it needs).	39	WAN link (see the script header for the variables it needs).
40		40
41	Multiple mux clients may attach to one session; the grid follows the most	41	Multiple mux clients may attach to one session; the grid follows the most
42	recent attacher/resize (latest wins).	42	recently active client — typing, attaching, or resizing claims it (latest
		43	wins).
43		44
44	Detach, or kill the client outright — the session survives and `mux`	45	Detach, or kill the client outright — the session survives and `mux`
45	resumes it from a state snapshot, including TUI screens and the primary	46	resumes it from a state snapshot, including TUI screens and the primary

docs/decisions.md

Old		New
@@ -169,8 +169,23 @@
169	## 2026-08-07 (M5)	169	## 2026-08-07 (M5)
170		170
171	- Resize policy: latest wins. The authoritative grid follows the most	171	- Resize policy: latest wins. The authoritative grid follows the most
172	recent attach or resize event from any client (modern tmux	172	recently active client: attaching, resizing, or typing claims it
173	`window-size latest`). Rejected: smallest-wins (punishes the larger	173	(modern tmux `window-size latest`). Amended post-M6 (2026-08-07):
		174	until then the triggers were attach and resize only, and real two-client
		175	use showed the tmux comparison overpromised — tmux's "latest" follows
		176	the active client, so typing at a console it had not resized left our
		177	grid at the other client's size. Input is now a trigger, applying the
		178	size and broadcasting down the same discontinuity path a resize takes:
		179	one broadcast on the first keystroke after a switch, nothing on the
		180	keystrokes after that, since the sizes then already match. A size the
		181	daemon refuses (degenerate, or an engine resize that fails) is neither
		182	recorded as that client's size nor broadcast — the grid never moved, so
		183	there is nothing to repaint anyone for, and a client whose size was
		184	never accepted claims nothing when it types. Scroll
		185	fetches are deliberately excluded from "activity" — paging history from
		186	a small terminal must not yank the grid away from the client actually
		187	working; nor are stats/dump/detach, which are not somebody using the
		188	terminal. Rejected: smallest-wins (punishes the larger
174	screen for the smaller one's presence — the handoff calls the result	189	screen for the smaller one's presence — the handoff calls the result
175	"widely disliked"); per-client reflow (needs a per-client engine or	190	"widely disliked"); per-client reflow (needs a per-client engine or
176	reflow pass, contradicting the single-authoritative-grid architecture —	191	reflow pass, contradicting the single-authoritative-grid architecture —

src/server.zig

@@ @@ -1,7 +1,8 @@ @@
+//! muxd's daemon core: one session (engine + pty), one listener, up to
+//! max_clients attached interactive clients plus a few dump-only observer
+//! connections. Every state update is broadcast to all attached clients;
+//! the grid follows the most recently active client — typing, attaching or
+//! resizing claims it (latest wins).
+//! Single-threaded; pumpOnce is one poll iteration so tests can drive the
+//! loop.
+const std = @import("std");
@@ @@ -203,6 +204,13 @@ pub fn installSignalHandlers() void { @@
+/// still owes it.
+const ClientSlot = struct {
+    fd: std.posix.fd_t,
+    /// The grid size this client last asked for *and got*: written only
+    /// after an applySize that succeeded, so it is never some other
+    /// client's size. 0x0 until the first accepted attach/resize, which
+    /// claimGrid reads as "makes no claim". Latest-wins re-reads this on
+    /// every keystroke — see the `.input` arm.
+    cols: u16 = 0,
+    rows: u16 = 0,
+    /// Frames queued but not yet accepted by the kernel. Bounded by
+    /// Server.pending_cap; a peer that stops reading gets dropped, never
+    /// waited on — one slow WAN client must not stall the session.
@@ @@ -585,17 +593,38 @@ pub const Server = struct { @@
+                // resync path isn't worth carrying — snapshot it (which,
+                // like any resize, everyone else hears about too).
+                const req = proto.decodeAttach(frame.payload) catch return;
+                if (self.applySize(req.cols, req.rows)) self.recordSize(i);
+                self.resyncSnapshot();
+            },
+            .resize => {
+                // Latest wins: whoever resized last sets the grid, and the
+                // broadcast snapshot tells every other client about it.
+                const sz = proto.decodeSize(frame.payload) catch return;
+                if (self.applySize(sz.cols, sz.rows)) self.recordSize(i);
+                self.resyncSnapshot();
+            },
+            .input => {
+                // Latest wins follows *activity*, which includes typing:
+                // the console you are typing at claims the grid, exactly as
+                // attaching or resizing from it would. Costs one broadcast
+                // per switch between differently-sized clients — the first
+                // keystroke moves the grid, and every keystroke after that
+                // finds the sizes already equal and does nothing.
+                //
+                // The other client-initiated frames are deliberately not
+                // activity: .fetch_scrollback would let paging history from
+                // a small terminal yank the grid away from the client that
+                // is actually working, and .stats_req/.debug_dump/.detach
+                // are not somebody using the terminal at all.
+                //
+                // Claimed before the bytes go out, so the shell reacts to
+                // this keystroke at the size the typist is watching. The
+                // input is forwarded even if the broadcast dropped this
+                // client — the bytes are already ours, and dropClient on an
+                // empty slot is a no-op.
+                self.claimGrid(i);
+                proto.writeAllFd(self.pty.master, frame.payload) catch self.dropClient(i);
+            },
+            .stats_req => {
+                var buf: [stats_text_len]u8 = undefined;
+                const text = self.statsText(&buf) catch {
@@ @@ -660,11 +689,18 @@ pub const Server = struct { @@
+                self.observers[i] = null; // promote without closing
+                self.clients[slot] = .{ .fd = fd };
+                const size_changed = (sz.cols != self.colsNow() or sz.rows != self.rowsNow());
+                const applied = self.applySize(sz.cols, sz.rows);
+                // The slot's size is what latest-wins re-reads when this
+                // client types, so it is set here and kept current by the
+                // .resize/.attach arms — but only when the grid really went
+                // there. A refused attach must leave the slot at 0x0.
+                if (applied) self.recordSize(slot);
+                // Latest wins: a size change broadcasts, repainting every
+                // client at the new attacher's size. A same-size join is
+                // the joiner's business alone — see sendResync. So is a
+                // refused one: the size differs but the grid never moved,
+                // so there is nothing to repaint anyone else for.
+                self.sendResync(slot, sz.have_seq, sz.have_epoch, size_changed and applied);
+            },
+            .debug_dump => self.replyDumpObserver(fd, frame.payload) catch self.dropObserver(i),
+            .stats_req => self.replyStatsObserver(fd) catch self.dropObserver(i),
@@ @@ -692,12 +728,44 @@ pub const Server = struct { @@
+        try proto.writeFrame(fd, .dump_reply, dump);
+    /// Take the grid to `cols` x `rows`. Returns whether the grid is now
+    /// that size — false means the request was refused and nothing moved,
+    /// which is what keeps a refused size out of a client's slot.
+    fn applySize(self: *Server, cols: u16, rows: u16) bool {
+        // A degenerate size (0x0 pty, buggy client) would trip engine
+        // asserts; keep the current grid instead.
+        if (cols < 2 or rows < 2) return false;
+        self.eng.resize(cols, rows) catch return false;
+        self.pty.resize(cols, rows) catch {};
+        return true;
+    }
+    /// Record the grid as client `i`'s size. ONLY call this after an
+    /// applySize that returned true: the grid is then the size this client
+    /// asked for, and recording it is recording the client's own size. Call
+    /// it after a refusal and you record whatever size the grid happens to
+    /// hold — some other client's — which this client would then claim as
+    /// its own the moment it typed.
+    fn recordSize(self: *Server, i: usize) void {
+        if (self.clients[i] == null) return;
+        self.clients[i].?.cols = self.colsNow();
+        self.clients[i].?.rows = self.rowsNow();
+    }
+    /// Latest wins on activity: bring the grid to client `i`'s size if it
+    /// isn't there already, and tell everyone. Same discontinuity path a
+    /// resize takes, because that is exactly what this is — the difference
+    /// is only what triggered it.
+    fn claimGrid(self: *Server, i: usize) void {
+        if (self.clients[i] == null) return;
+        const slot = &self.clients[i].?;
+        if (slot.cols == self.colsNow() and slot.rows == self.rowsNow()) return;
+        // A client that has never had a size accepted is still 0x0 and
+        // claims nothing — this is what stops a degenerate attacher from
+        // broadcasting a repaint on every keystroke for a resize that
+        // cannot happen. Two compares and a refusal, once per keystroke.
+        if (!self.applySize(slot.cols, slot.rows)) return;
+        self.resyncSnapshot();
+    /// Counterfactual: what M2 would have sent for this one update as a
+    try std.testing.expectEqual(@as(u16, 30), @as(u16, @intCast(srv.eng.term.rows)));
+/// Test helper: read `fd` until a snapshot arrives whose prefix reports
+/// `cols` x `rows`, discarding everything else. Lenient about what precedes
+/// it because a client's own join snapshot and the shell's deltas share the
+/// stream with the broadcast under test.
+fn awaitSnapshotSize(
+    alloc: std.mem.Allocator,
+    fd: std.posix.fd_t,
+    cols: u16,
+    rows: u16,
+    timeout_ms: u64,
+) !bool {
+    var deadline_ms = timeout_ms;
+    while (deadline_ms > 0) {
+        var pfd = [_]std.posix.pollfd{
+            .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
+        };
+        const ready = try std.posix.poll(&pfd, 100);
+        deadline_ms -|= 100;
+        if (ready == 0) continue;
+        const frame = (try proto.readFrame(alloc, fd)) orelse return false;
+        defer frame.deinit(alloc);
+        if (frame.type != .snapshot) continue;
+        const p = try proto.readSnapshotPrefix(frame.payload);
+        if (p.cols == cols and p.rows == rows) return true;
+    }
+    return false;
+}
+test "Server: typing claims the grid for the typist (latest-wins on input)" {
+    const alloc = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var path_buf: [256]u8 = undefined;
+    const dir_path = try tmp.dir.realpath(".", &path_buf);
+    const sock_path = try std.fmt.allocPrint(alloc, "{s}/typing.sock", .{dir_path});
+    defer alloc.free(sock_path);
+    var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
+    defer srv.deinit();
+    var stop = std.atomic.Value(bool).init(false);
+    const th = try std.Thread.spawn(.{}, serverThread, .{ &srv, &stop });
+    defer th.join();
+    defer stop.store(true, .release);
+    // A joins first and, being the only client, sets the grid: attach-wins
+    // is unchanged by any of this.
+    const a = try std.net.connectUnixSocket(sock_path);
+    defer a.close();
+    try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(100, 30, 0, 0));
+    try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 100, 30, 10_000));
+    // B joins at a different size and takes the grid, which A is told about.
+    const b = try std.net.connectUnixSocket(sock_path);
+    defer b.close();
+    try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
+    try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 80, 24, 10_000));
+    try std.testing.expect(try awaitSnapshotSize(alloc, b.handle, 80, 24, 10_000));
+    // Quiet on both sides, so what follows is unambiguously the typing's
+    // doing and not a frame still in flight from the join.
+    _ = try drainHeld(alloc, a.handle, 10_000);
+    _ = try drainHeld(alloc, b.handle, 10_000);
+    // The point of the test: A types without resizing or re-attaching, and
+    // the grid moves back to A's size for everyone.
+    try proto.writeFrame(a.handle, .input, "echo from-a\n");
+    try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 100, 30, 10_000));
+    try std.testing.expect(try awaitSnapshotSize(alloc, b.handle, 100, 30, 10_000));
+    _ = try drainHeld(alloc, a.handle, 10_000);
+    _ = try drainHeld(alloc, b.handle, 10_000);
+    // ...and it switches back when B types, so this is "most recent", not
+    // "first" or "largest".
+    try proto.writeFrame(b.handle, .input, "echo from-b\n");
+    try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 80, 24, 10_000));
+    try std.testing.expect(try awaitSnapshotSize(alloc, b.handle, 80, 24, 10_000));
+    _ = try drainHeld(alloc, a.handle, 10_000);
+    _ = try drainHeld(alloc, b.handle, 10_000);
+    // A second keystroke from the same client finds the grid already its
+    // size and must claim nothing: a snapshot here would mean every
+    // keystroke of an ordinary session pays for a full repaint.
+    try proto.writeFrame(b.handle, .input, "echo again-from-b\n");
+    const first = try firstStateFrame(alloc, b.handle, 10_000);
+    try std.testing.expect(first != null);
+    try std.testing.expectEqual(proto.MsgType.delta, first.?.type);
+}
+/// Test helper: watch `fd` until `marker` shows up in a replica built from
+/// what it sends, and fail if any of it is a snapshot. The replica starts
+/// blank on purpose — a delta carries every row it changed, so the row the
+/// marker lands on arrives whole.
+fn awaitMarkerWithoutSnapshot(
+    alloc: std.mem.Allocator,
+    fd: std.posix.fd_t,
+    cols: u16,
+    rows: u16,
+    marker: []const u8,
+    timeout_ms: u64,
+) !bool {
+    var replica = try Engine.init(alloc, .{ .cols = cols, .rows = rows });
+    defer replica.deinit();
+    var deadline_ms = timeout_ms;
+    while (deadline_ms > 0) {
+        var pfd = [_]std.posix.pollfd{
+            .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
+        };
+        const ready = try std.posix.poll(&pfd, 100);
+        deadline_ms -|= 100;
+        if (ready == 0) continue;
+        const frame = (try proto.readFrame(alloc, fd)) orelse return false;
+        defer frame.deinit(alloc);
+        if (frame.type == .snapshot) return error.UnexpectedSnapshotBroadcast;
+        if (frame.type != .delta) continue;
+        try applyFrame(alloc, replica, frame);
+        const plain = try replica.dumpPlain(alloc);
+        defer alloc.free(plain);
+        if (std.mem.indexOf(u8, plain, marker) != null) return true;
+    }
+    return false;
+}
+test "Server: a size the grid refuses never becomes a claim" {
+    const alloc = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var path_buf: [256]u8 = undefined;
+    const dir_path = try tmp.dir.realpath(".", &path_buf);
+    const sock_path = try std.fmt.allocPrint(alloc, "{s}/refused.sock", .{dir_path});
+    defer alloc.free(sock_path);
+    var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
+    defer srv.deinit();
+    var stop = std.atomic.Value(bool).init(false);
+    const th = try std.Thread.spawn(.{}, serverThread, .{ &srv, &stop });
+    // Stopped and joined explicitly below so the last assertion can read
+    // server state without racing the daemon thread.
+    var joined = false;
+    defer if (!joined) {
+        stop.store(true, .release);
+        th.join();
+    };
+    const a = try std.net.connectUnixSocket(sock_path);
+    defer a.close();
+    try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
+    try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 80, 24, 10_000));
+    _ = try drainHeld(alloc, a.handle, 10_000);
+    // D asks for a size applySize will not have (1x1 trips engine asserts).
+    // It still joins and is still served — it is only its *size* that is
+    // refused — but the grid must not move, and A must not be repainted for
+    // a resize that never happened.
+    const d = try std.net.connectUnixSocket(sock_path);
+    defer d.close();
+    try proto.writeFrame(d.handle, .attach, &proto.encodeAttach(1, 1, 0, 0));
+    try std.testing.expect(try awaitSnapshotSize(alloc, d.handle, 80, 24, 10_000));
+    // D types. Its slot holds no accepted size, so this claims nothing: A
+    // sees the output as deltas and no snapshot at all. That the marker
+    // arrives at all is the other half — a refused size must not cost the
+    // client its keystrokes.
+    try proto.writeFrame(d.handle, .input, "echo typed-by-refused\n");
+    try std.testing.expect(try awaitMarkerWithoutSnapshot(
+        alloc,
+        a.handle,
+,
+,
+        "typed-by-refused",
+_000,
+    ));
+    // The reviewer's repro: a real client moves the grid, and then D types
+    // again. If D's slot had been stamped with the grid it found at attach
+    // (80x24), this keystroke would drag everyone back to a size neither
+    // client asked for.
+    try proto.writeFrame(a.handle, .resize, &proto.encodeSize(100, 30));
+    try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 100, 30, 10_000));
+    _ = try drainHeld(alloc, a.handle, 10_000);
+    _ = try drainHeld(alloc, d.handle, 10_000);
+    try proto.writeFrame(d.handle, .input, "echo typed-again\n");
+    try std.testing.expect(try awaitMarkerWithoutSnapshot(
+        alloc,
+        a.handle,
+,
+,
+        "typed-again",
+_000,
+    ));
+    stop.store(true, .release);
+    th.join();
+    joined = true;
+    // The grid is where the only client with an accepted size put it.
+    try std.testing.expectEqual(@as(u16, 100), @as(u16, @intCast(srv.eng.term.cols)));
+    try std.testing.expectEqual(@as(u16, 30), @as(u16, @intCast(srv.eng.term.rows)));
+}
+test "Server: scrollback fetch is per-client and independent" {
+    const alloc = std.testing.allocator;