f07a56c8

feat: the wire carries cells; the replica is a grid and every painter reads it

a73x 2026-09-04 18:04

Commit message

feat: the wire carries cells; the replica is a grid and every painter reads it

The daemon encodes a viewport row as a CellRow and the client copies it into
a grid, so no client parses VT any more. Frame numbers move with the payload
change: snapshot 0x95, delta 0x96, scrollback_chunk 0x97, with 0x81, 0x85 and
0x87 retired and never reused, so a binary from either side of the break drops
the other's frames instead of parsing cells as VT or the reverse.

The snapshot cursor rides in the body rather than in the prefix, so the
prefix keeps the layout its golden pin and every reader of it already have.
composeDelta and its rule 4 exemption are gone with the VT payloads.

This commit covers protocol.zig, delta.zig, the daemon's three payload sites
and the server test harness; the painters and the two browser clients follow.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CsWfuJFQbTfGtKZLS5qw4q

src/engine/delta.zig

Old		New
@@ -72,12 +72,12 @@ pub const DeltaTracker = struct {
72	self.reset_seq = self.seq;	72	self.reset_seq = self.seq;
73	self.cursor = eng.cursorPos();	73	self.cursor = eng.cursorPos();
74	self.history_rows = eng.historyRows();	74	self.history_rows = eng.historyRows();
75	// Claim no rows until every row is stamped: a dump that fails partway	75	// Claim no rows until every row is stamped: an encode that fails partway
76	// leaves stale seqs in the tail of `row_seqs`, and any above a client's	76	// leaves stale seqs in the tail of `row_seqs`, and any above a client's
77	// `have_seq` puts that row in every delta from here on.	77	// `have_seq` puts that row in every delta from here on.
78	self.rows = 0;	78	self.rows = 0;
79	for (0..rows) \|y\| {	79	for (0..rows) \|y\| {
80	const bytes = try eng.dumpVtRow(alloc, @intCast(y));	80	const bytes = try eng.encodeViewportRow(alloc, @intCast(y));
81	defer alloc.free(bytes);	81	defer alloc.free(bytes);
82	self.row_hashes[y] = Wyhash.hash(0, bytes);	82	self.row_hashes[y] = Wyhash.hash(0, bytes);
83	self.row_seqs[y] = self.seq;	83	self.row_seqs[y] = self.seq;
@@ -98,8 +98,9 @@ pub const DeltaTracker = struct {
98	/// `eng`, or whether the client has to be resynced from scratch.	98	/// `eng`, or whether the client has to be resynced from scratch.
99	///	99	///
100	/// Geometry is load-bearing, not decorative: row_hashes is indexed by the	100	/// Geometry is load-bearing, not decorative: row_hashes is indexed by the
101	/// tracker's own row count, and dumpVtRow asserts against the engine's. A	101	/// tracker's own row count, and encodeViewportRow asserts against the
102	/// tracker left stale by a failed rebuild resyncs here instead of running	102	/// engine's. A tracker left stale by a failed rebuild resyncs here
		103	/// instead of running
103	/// off the end of the grid. A tracker that was never built has no rows to	104	/// off the end of the grid. A tracker that was never built has no rows to
104	/// diff at all, and an alt-screen flip replaces the whole grid, so no row	105	/// diff at all, and an alt-screen flip replaces the whole grid, so no row
105	/// seq from before it means anything.	106	/// seq from before it means anything.
@@ -114,7 +115,7 @@ pub const DeltaTracker = struct {
114		115
115	/// Diff current engine state against the tracked state. Advances seq	116	/// Diff current engine state against the tracked state. Advances seq
116	/// and tracked rows when anything changed. Allocates only the per-row	117	/// and tracked rows when anything changed. Allocates only the per-row
117	/// dumps it hashes.	118	/// encodings it hashes.
118	pub fn update(self: DeltaTracker, alloc: std.mem.Allocator, eng: Engine) !Update {	119	pub fn update(self: DeltaTracker, alloc: std.mem.Allocator, eng: Engine) !Update {
119	if (!self.continuous(eng)) return .discontinuity;	120	if (!self.continuous(eng)) return .discontinuity;
120		121
@@ -124,7 +125,7 @@ pub const DeltaTracker = struct {
124	const next_seq = self.seq + 1;	125	const next_seq = self.seq + 1;
125	var changed: usize = 0;	126	var changed: usize = 0;
126	for (0..self.rows) \|y\| {	127	for (0..self.rows) \|y\| {
127	const bytes = try eng.dumpVtRow(alloc, @intCast(y));	128	const bytes = try eng.encodeViewportRow(alloc, @intCast(y));
128	defer alloc.free(bytes);	129	defer alloc.free(bytes);
129	const hash = Wyhash.hash(0, bytes);	130	const hash = Wyhash.hash(0, bytes);
130	self.scratch_hashes[y] = hash;	131	self.scratch_hashes[y] = hash;
@@ -177,8 +178,8 @@ pub const DeltaTracker = struct {
177		178
178	/// Build a delta payload of all rows changed after `since`. The header's	179	/// Build a delta payload of all rows changed after `since`. The header's
179	/// `row_count` and the appended rows MUST agree, so both come from the same	180	/// `row_count` and the appended rows MUST agree, so both come from the same
180	/// predicate with nothing mutating in between. Changed rows are dumped twice	181	/// predicate with nothing mutating in between. Changed rows are encoded
181	/// — once to hash, once to serialize — which is 1-3 rows in steady state.	182	/// twice — once to hash, once to send — which is 1-3 rows in steady state.
182	pub fn buildDeltaSince(self: DeltaTracker, alloc: std.mem.Allocator, eng: Engine, since: u64) ![]u8 {	183	pub fn buildDeltaSince(self: DeltaTracker, alloc: std.mem.Allocator, eng: Engine, since: u64) ![]u8 {
183	var rows_changed: u16 = 0;	184	var rows_changed: u16 = 0;
184	for (self.row_seqs) \|s\| {	185	for (self.row_seqs) \|s\| {
@@ -195,7 +196,7 @@ pub const DeltaTracker = struct {
195	});	196	});
196	for (self.row_seqs, 0..) \|s, y\| {	197	for (self.row_seqs, 0..) \|s, y\| {
197	if (s <= since) continue;	198	if (s <= since) continue;
198	const bytes = try eng.dumpVtRow(alloc, @intCast(y));	199	const bytes = try eng.encodeViewportRow(alloc, @intCast(y));
199	defer alloc.free(bytes);	200	defer alloc.free(bytes);
200	try proto.appendDeltaRow(&payload, alloc, @intCast(y), bytes);	201	try proto.appendDeltaRow(&payload, alloc, @intCast(y), bytes);
201	}	202	}
@@ -203,6 +204,42 @@ pub const DeltaTracker = struct {
203	}	204	}
204	};	205	};
205		206
		207	/// The snapshot payload: the prefix, the cursor, then every viewport row as
		208	/// cells. The cursor rides in the body rather than in the prefix so the
		209	/// prefix keeps the layout its golden pin and its readers already have.
		210	pub fn buildSnapshot(alloc: std.mem.Allocator, eng: *Engine, prefix: proto.SnapshotPrefix) ![]u8 {
		211	var out: std.ArrayList(u8) = .empty;
		212	errdefer out.deinit(alloc);
		213	var pbuf: [proto.snapshot_prefix_len]u8 = undefined;
		214	proto.writeSnapshotPrefix(&pbuf, prefix);
		215	try out.appendSlice(alloc, &pbuf);
		216	var cbuf: [proto.snapshot_cursor_len]u8 = undefined;
		217	const cur = eng.cursorPos();
		218	proto.writeSnapshotCursor(&cbuf, cur.x, cur.y);
		219	try out.appendSlice(alloc, &cbuf);
		220	var y: u16 = 0;
		221	while (y < eng.term.rows) : (y += 1) {
		222	const row = try eng.encodeViewportRow(alloc, y);
		223	defer alloc.free(row);
		224	try out.appendSlice(alloc, row);
		225	}
		226	return out.toOwnedSlice(alloc);
		227	}
		228
		229	// ---------------------------------------------------------------------------
		230	// Tests. A delta payload is rows of cells now, so the assertions decode it the
		231	// way a client does — through grid.decodeRow — rather than searching bytes.
		232
		233	const grid = @import("grid.zig");
		234
		235	/// Apply every row of a delta payload into `g` and return the plain text, so
		236	/// a test can say what the far side would be SHOWING after the frame.
		237	fn deltaText(alloc: std.mem.Allocator, g: *grid.Grid, payload: []const u8) ![]const u8 {
		238	var it = proto.deltaRowIterator(payload);
		239	while (try it.next()) \|row\| try g.applyRow(row.row, row.bytes);
		240	return g.dumpPlain(alloc);
		241	}
		242
206	test "DeltaTracker: alt-screen flip is a discontinuity and rows follow the active screen" {	243	test "DeltaTracker: alt-screen flip is a discontinuity and rows follow the active screen" {
207	const alloc = std.testing.allocator;	244	const alloc = std.testing.allocator;
208		245
@@ -235,9 +272,11 @@ test "DeltaTracker: alt-screen flip is a discontinuity and rows follow the activ
235	}	272	}
236	const payload = try tracker.buildDeltaSince(alloc, eng, tracker.seq - 1);	273	const payload = try tracker.buildDeltaSince(alloc, eng, tracker.seq - 1);
237	defer alloc.free(payload);	274	defer alloc.free(payload);
238	const composed = try proto.composeDelta(alloc, payload);	275	const g = try grid.Grid.init(alloc, 80, 24);
239	defer alloc.free(composed.bytes);	276	defer g.deinit();
240	try std.testing.expect(std.mem.indexOf(u8, composed.bytes, "alt content") != null);	277	const text = try deltaText(alloc, g, payload);
		278	defer alloc.free(text);
		279	try std.testing.expect(std.mem.indexOf(u8, text, "alt content") != null);
241	}	280	}
242		281
243	test "DeltaTracker: a resize behind the tracker's back resyncs instead of over-reading" {	282	test "DeltaTracker: a resize behind the tracker's back resyncs instead of over-reading" {
@@ -282,10 +321,12 @@ test "DeltaTracker: blind output is answerable on reattach without rendering a r
282	try std.testing.expect(tracker.canServe(have));	321	try std.testing.expect(tracker.canServe(have));
283	const payload = try tracker.buildDeltaSince(alloc, eng, have);	322	const payload = try tracker.buildDeltaSince(alloc, eng, have);
284	defer alloc.free(payload);	323	defer alloc.free(payload);
285	const composed = try proto.composeDelta(alloc, payload);	324	const g = try grid.Grid.init(alloc, 80, 24);
286	defer alloc.free(composed.bytes);	325	defer g.deinit();
		326	const text = try deltaText(alloc, g, payload);
		327	defer alloc.free(text);
287	try std.testing.expect(	328	try std.testing.expect(
288	std.mem.indexOf(u8, composed.bytes, "printed with nobody watching") != null,	329	std.mem.indexOf(u8, text, "printed with nobody watching") != null,
289	);	330	);
290		331
291	// The HEADER, not only the rows: `buildDeltaSince` serialises the tracker's	332	// The HEADER, not only the rows: `buildDeltaSince` serialises the tracker's
@@ -375,9 +416,11 @@ test "DeltaTracker: a row that reverts after a blind stretch is still sent" {
375	}	416	}
376	const payload = try tracker.buildDeltaSince(alloc, eng, tracker.seq - 1);	417	const payload = try tracker.buildDeltaSince(alloc, eng, tracker.seq - 1);
377	defer alloc.free(payload);	418	defer alloc.free(payload);
378	const composed = try proto.composeDelta(alloc, payload);	419	const g = try grid.Grid.init(alloc, 80, 24);
379	defer alloc.free(composed.bytes);	420	defer g.deinit();
380	try std.testing.expect(std.mem.indexOf(u8, composed.bytes, "AAA") != null);	421	const text = try deltaText(alloc, g, payload);
		422	defer alloc.free(text);
		423	try std.testing.expect(std.mem.indexOf(u8, text, "AAA") != null);
381	}	424	}
382		425
383	test "DeltaTracker: a screen switch with nobody watching is still a discontinuity" {	426	test "DeltaTracker: a screen switch with nobody watching is still a discontinuity" {
@@ -400,6 +443,40 @@ test "DeltaTracker: a screen switch with nobody watching is still a discontinuit
400	}	443	}
401	}	444	}
402		445
		446	test "buildSnapshot: prefix, cursor and one CellRow per viewport row" {
		447	const alloc = std.testing.allocator;
		448	const eng = try Engine.init(alloc, .{ .cols = 4, .rows = 2 });
		449	defer eng.deinit();
		450	eng.feed("a\r\nb");
		451
		452	const payload = try buildSnapshot(alloc, eng, .{
		453	.seq = 5,
		454	.history_rows = 1,
		455	.cols = 4,
		456	.rows = 2,
		457	.epoch = 0x99,
		458	});
		459	defer alloc.free(payload);
		460
		461	const prefix = try proto.readSnapshotPrefix(payload);
		462	try std.testing.expectEqual(@as(u64, 5), prefix.seq);
		463	try std.testing.expectEqual(@as(u16, 4), prefix.cols);
		464	try std.testing.expectEqual(@as(u16, 2), prefix.rows);
		465	// The cursor is the engine's own, not a value the caller passed in: a
		466	// snapshot that carried the prefix's idea of it would park the caret
		467	// wherever the last resize left it.
		468	const cur = try proto.readSnapshotCursor(payload);
		469	const eng_cur = eng.cursorPos();
		470	try std.testing.expectEqual(eng_cur.x, cur.x);
		471	try std.testing.expectEqual(eng_cur.y, cur.y);
		472
		473	const body = payload[proto.snapshot_prefix_len + proto.snapshot_cursor_len ..];
		474	const rows = try grid.decodeRows(alloc, body, 2, 4);
		475	defer grid.freeRows(alloc, rows);
		476	try std.testing.expectEqualStrings("a", rows[0].textOf(rows[0].cells[0]));
		477	try std.testing.expectEqualStrings("b", rows[1].textOf(rows[1].cells[0]));
		478	}
		479
403	// Forces semantic analysis of every pub decl under `zig build test`, so an	480	// Forces semantic analysis of every pub decl under `zig build test`, so an
404	// unreferenced decl must at least compile (the silent-module-loss hazard,	481	// unreferenced decl must at least compile (the silent-module-loss hazard,
405	// decisions.md). Pub decls only: std.meta.declarations sees nothing private.	482	// decisions.md). Pub decls only: std.meta.declarations sees nothing private.

src/engine/protocol.zig

Old		New
@@ -2,7 +2,6 @@
2	//! semantics for deltas. Frame = 1 byte MsgType, u32 LE payload length, payload.	2	//! semantics for deltas. Frame = 1 byte MsgType, u32 LE payload length, payload.
3	//! Hand-rolled deliberately — payloads are row-keyed blobs and fixed-width	3	//! Hand-rolled deliberately — payloads are row-keyed blobs and fixed-width
4	//! little-endian integers, which readInt/writeInt cover without a dependency.	4	//! little-endian integers, which readInt/writeInt cover without a dependency.
5	// folder rule 4 exemption: a delta row IS painted bytes on the wire — composeDelta stamps CUP and EL around a row the far side replays.
6	const std = @import("std");	5	const std = @import("std");
7		6
8	pub const MsgType = enum(u8) {	7	pub const MsgType = enum(u8) {
@@ -26,15 +25,20 @@ pub const MsgType = enum(u8) {
26	end_req = 0x11, // payload: u8 flags (bit0 force) ++ optional session-name tail (empty = default session)	25	end_req = 0x11, // payload: u8 flags (bit0 force) ++ optional session-name tail (empty = default session)
27	debug_dump = 0x7f, // payload: 1 byte: 0 = plain, 1 = vt ++ optional session-name tail (empty = default session)	26	debug_dump = 0x7f, // payload: 1 byte: 0 = plain, 1 = vt ++ optional session-name tail (empty = default session)
28	// daemon -> client	27	// daemon -> client
29	snapshot = 0x81, // payload: SnapshotPrefix ++ full-state vt dump	28	// The three replay frames were renumbered in 2026-09 when their payloads
		29	// stopped being VT and became cells. The old numbers 0x81 snapshot,
		30	// 0x85 scrollback_chunk and 0x87 delta are retired and never reused, so a
		31	// binary from either side of the change drops the other's frames as
		32	// unknown instead of parsing cells as VT or the reverse.
		33	snapshot = 0x95, // payload: SnapshotPrefix ++ u16 LE cursor_x ++ u16 LE cursor_y ++ rows x CellRow
30	exit_status = 0x82, // payload: 1 byte exit code	34	exit_status = 0x82, // payload: 1 byte exit code
31	// Retired when attach became a JOIN rather than a takeover; no daemon	35	// Retired when attach became a JOIN rather than a takeover; no daemon
32	// sends it. The clients still handle it (`wallview`, `interact`)	36	// sends it. The clients still handle it (`wallview`, `interact`)
33	// because a new client may attach to an old daemon that does.	37	// because a new client may attach to an old daemon that does.
34	taken_over = 0x84, // payload: empty; a newer client attached, you're out	38	taken_over = 0x84, // payload: empty; a newer client attached, you're out
35	scrollback_chunk = 0x85, // payload: u32 LE start, u16 LE count ++ vt rows	39	scrollback_chunk = 0x97, // payload: u32 LE start, u16 LE count ++ count x CellRow
36	stats_reply = 0x86, // payload: human-readable stats text	40	stats_reply = 0x86, // payload: human-readable stats text
37	delta = 0x87, // payload: see DeltaHeader + rows	41	delta = 0x96, // payload: DeltaHeader ++ row_count x (u16 LE row ++ u32 LE len ++ CellRow)
38	pty_mode = 0x88, // payload: 1 byte flags: bit0 icanon, bit1 echo	42	pty_mode = 0x88, // payload: 1 byte flags: bit0 icanon, bit1 echo
39	endpoint_reply = 0x89, // payload: u16 LE port; 0 = no listener could be produced (reason in daemon log)	43	endpoint_reply = 0x89, // payload: u16 LE port; 0 = no listener could be produced (reason in daemon log)
40	cmd_state = 0x8a, // payload: CmdState (see encodeCmdState); pushed on marks-regime transitions	44	cmd_state = 0x8a, // payload: CmdState (see encodeCmdState); pushed on marks-regime transitions
@@ -1197,6 +1201,26 @@ pub fn readSnapshotPrefix(payload: []const u8) !SnapshotPrefix {
1197	};	1201	};
1198	}	1202	}
1199		1203
		1204	/// The cursor rides in the snapshot BODY rather than in the prefix so that
		1205	/// the prefix keeps its golden layout and every reader of it is unchanged.
		1206	pub const snapshot_cursor_len = 4;
		1207
		1208	pub fn writeSnapshotCursor(buf: *[snapshot_cursor_len]u8, x: u16, y: u16) void {
		1209	std.mem.writeInt(u16, buf[0..2], x, .little);
		1210	std.mem.writeInt(u16, buf[2..4], y, .little);
		1211	}
		1212
		1213	pub const SnapshotCursor = struct { x: u16, y: u16 };
		1214
		1215	pub fn readSnapshotCursor(payload: []const u8) !SnapshotCursor {
		1216	if (payload.len < snapshot_prefix_len + snapshot_cursor_len) return error.BadPayload;
		1217	const b = payload[snapshot_prefix_len..][0..snapshot_cursor_len];
		1218	return .{
		1219	.x = std.mem.readInt(u16, b[0..2], .little),
		1220	.y = std.mem.readInt(u16, b[2..4], .little),
		1221	};
		1222	}
		1223
1200	/// No epoch: a delta only arrives on the connection its snapshot opened.	1224	/// No epoch: a delta only arrives on the connection its snapshot opened.
1201	pub const DeltaHeader = struct {	1225	pub const DeltaHeader = struct {
1202	seq: u64,	1226	seq: u64,
@@ -1555,38 +1579,6 @@ pub const CellRowReader = struct {
1555	}	1579	}
1556	};	1580	};
1557		1581
1558	pub const ComposedDelta = struct { header: DeltaHeader, bytes: []u8 };
1559
1560	/// Turn a delta payload into the VT byte string that applies it: per row, CUP
1561	/// to the row start, EL(2), the row's styled content, then a CUP to the delta's
1562	/// cursor. The composed bytes assume the receiver has no scroll region or origin
1563	/// mode — true of a freshly reset replica and of the client's full-screen paint.
1564	/// The header's `row_count` is authoritative.
1565	pub fn composeDelta(alloc: std.mem.Allocator, payload: []const u8) !ComposedDelta {
1566	const hdr = try readDeltaHeader(payload);
1567	var out: std.ArrayList(u8) = .empty;
1568	errdefer out.deinit(alloc);
1569
1570	var seen: usize = 0;
1571	var it = deltaRowIterator(payload);
1572	while (try it.next()) \|row\| {
1573	var buf: [16]u8 = undefined;
1574	const cup = try std.fmt.bufPrint(&buf, "\x1b[{d};1H\x1b[2K", .{@as(u32, row.row) + 1});
1575	try out.appendSlice(alloc, cup);
1576	try out.appendSlice(alloc, row.bytes);
1577	seen += 1;
1578	}
1579	if (seen != hdr.row_count) return error.BadPayload;
1580
1581	var cbuf: [16]u8 = undefined;
1582	const cur = try std.fmt.bufPrint(&cbuf, "\x1b[{d};{d}H", .{
1583	@as(u32, hdr.cursor_y) + 1,
1584	@as(u32, hdr.cursor_x) + 1,
1585	});
1586	try out.appendSlice(alloc, cur);
1587	return .{ .header = hdr, .bytes = try out.toOwnedSlice(alloc) };
1588	}
1589
1590	test "the frame header layout is these bytes, and the decoders read them back" {	1582	test "the frame header layout is these bytes, and the decoders read them back" {
1591	const alloc = std.testing.allocator;	1583	const alloc = std.testing.allocator;
1592	const golden = [_]u8{ 0x02, 3, 0, 0, 0, 'a', 'b', 'c' };	1584	const golden = [_]u8{ 0x02, 3, 0, 0, 0, 'a', 'b', 'c' };
@@ -1912,28 +1904,6 @@ test "delta build/iterate round trip" {
1912	try std.testing.expectEqual(@as(?DeltaRow, null), try it.next());	1904	try std.testing.expectEqual(@as(?DeltaRow, null), try it.next());
1913	}	1905	}
1914		1906
1915	test "composeDelta produces CUP+EL row paints and final cursor restore" {
1916	const alloc = std.testing.allocator;
1917	var payload: std.ArrayList(u8) = .empty;
1918	defer payload.deinit(alloc);
1919	try appendDeltaHeader(&payload, alloc, .{
1920	.seq = 1,
1921	.history_rows = 0,
1922	.cursor_x = 4,
1923	.cursor_y = 2,
1924	.row_count = 1,
1925	});
1926	try appendDeltaRow(&payload, alloc, 9, "\x1b[0mrow-ten");
1927
1928	const composed = try composeDelta(alloc, payload.items);
1929	defer alloc.free(composed.bytes);
1930	// Row 9 (0-based) paints at line 10; EL(2) clears the old content.
1931	try std.testing.expect(std.mem.indexOf(u8, composed.bytes, "\x1b[10;1H\x1b[2K\x1b[0mrow-ten") != null);
1932	// Ends with the cursor restore (1-based 3;5).
1933	try std.testing.expect(std.mem.endsWith(u8, composed.bytes, "\x1b[3;5H"));
1934	try std.testing.expectEqual(@as(u64, 1), composed.header.seq);
1935	}
1936
1937	test "delta row with a bogus length is rejected, not trusted" {	1907	test "delta row with a bogus length is rejected, not trusted" {
1938	const alloc = std.testing.allocator;	1908	const alloc = std.testing.allocator;
1939	var payload: std.ArrayList(u8) = .empty;	1909	var payload: std.ArrayList(u8) = .empty;
@@ -1960,40 +1930,6 @@ test "delta payload too short for a header iterates empty" {
1960	try std.testing.expectEqual(@as(?DeltaRow, null), try it.next());	1930	try std.testing.expectEqual(@as(?DeltaRow, null), try it.next());
1961	}	1931	}
1962		1932
1963	test "composeDelta rejects a row_count the payload does not back up" {
1964	const alloc = std.testing.allocator;
1965	var payload: std.ArrayList(u8) = .empty;
1966	defer payload.deinit(alloc);
1967	try appendDeltaHeader(&payload, alloc, .{
1968	.seq = 1,
1969	.history_rows = 0,
1970	.cursor_x = 0,
1971	.cursor_y = 0,
1972	.row_count = 2, // claims two rows...
1973	});
1974	try appendDeltaRow(&payload, alloc, 0, "only-one"); // ...carries one.
1975
1976	try std.testing.expectError(error.BadPayload, composeDelta(alloc, payload.items));
1977	}
1978
1979	test "empty delta composes to a bare cursor home" {
1980	const alloc = std.testing.allocator;
1981	var payload: std.ArrayList(u8) = .empty;
1982	defer payload.deinit(alloc);
1983	try appendDeltaHeader(&payload, alloc, .{
1984	.seq = 3,
1985	.history_rows = 0,
1986	.cursor_x = 0,
1987	.cursor_y = 0,
1988	.row_count = 0,
1989	});
1990	try std.testing.expectEqual(@as(usize, delta_header_len), payload.items.len);
1991
1992	const composed = try composeDelta(alloc, payload.items);
1993	defer alloc.free(composed.bytes);
1994	try std.testing.expectEqualStrings("\x1b[1;1H", composed.bytes);
1995	}
1996
1997	test "encodeAttach golden bytes" {	1933	test "encodeAttach golden bytes" {
1998	try std.testing.expectEqualSlices(u8, &[_]u8{	1934	try std.testing.expectEqualSlices(u8, &[_]u8{
1999	0x78, 0x00, // cols 120	1935	0x78, 0x00, // cols 120
@@ -2061,6 +1997,33 @@ test "snapshot prefix round trip and golden bytes" {
2061	try std.testing.expectEqual(p, q);	1997	try std.testing.expectEqual(p, q);
2062	}	1998	}
2063		1999
		2000	test "snapshot cursor golden bytes, read back from behind the prefix" {
		2001	// The cursor sits immediately after the prefix, so a reader offsets by
		2002	// both lengths. Pinned as literal bytes rather than as a round trip: a
		2003	// peer on the other side of an upgrade decodes them with its own copy of
		2004	// this rule, and encode and decode drifting together would pass a round
		2005	// trip and paint the cursor in the wrong place.
		2006	var payload: [snapshot_prefix_len + snapshot_cursor_len]u8 = undefined;
		2007	writeSnapshotPrefix(payload[0..snapshot_prefix_len], .{
		2008	.seq = 1,
		2009	.history_rows = 0,
		2010	.cols = 80,
		2011	.rows = 24,
		2012	.epoch = 1,
		2013	});
		2014	writeSnapshotCursor(payload[snapshot_prefix_len..][0..snapshot_cursor_len], 0x0102, 0x0304);
		2015	try std.testing.expectEqualSlices(u8, &.{ 0x02, 0x01, 0x04, 0x03 }, payload[snapshot_prefix_len..]);
		2016	const cur = try readSnapshotCursor(&payload);
		2017	try std.testing.expectEqual(@as(u16, 0x0102), cur.x);
		2018	try std.testing.expectEqual(@as(u16, 0x0304), cur.y);
		2019	// A payload that holds the prefix but not the cursor is refused: it
		2020	// would otherwise read four bytes of the first row as a position.
		2021	try std.testing.expectError(
		2022	error.BadPayload,
		2023	readSnapshotCursor(payload[0 .. snapshot_prefix_len + 3]),
		2024	);
		2025	}
		2026
2064	test "snapshot prefix rejects short payloads" {	2027	test "snapshot prefix rejects short payloads" {
2065	try std.testing.expectError(error.BadPayload, readSnapshotPrefix(&[_]u8{0} ** 15));	2028	try std.testing.expectError(error.BadPayload, readSnapshotPrefix(&[_]u8{0} ** 15));
2066	// A v1 prefix (16 bytes, no epoch) is short now: honouring it would mean	2029	// A v1 prefix (16 bytes, no epoch) is short now: honouring it would mean

src/engine/replica.zig

Old		New
@@ -1,24 +1,26 @@
1	//! The replay core: what a mux client does to mirror a daemon's session.	1	//! The replay core: what a mux client does to mirror a daemon's session.
2	//! Applies snapshot and delta frames to a local engine, tracks the resume	2	//! Applies snapshot and delta frames into a local grid, tracks the resume
3	//! coordinates, and follows the authoritative grid — ONE implementation for the	3	//! coordinates, and follows the authoritative grid — ONE implementation for the
4	//! CLI client, the wasm core and the server's test fixtures.	4	//! CLI client, the wasm core and the server's test fixtures.
5	//!	5	//!
6	//! Deliberately platform-free: no posix, no fds, no clocks, since this must	6	//! No VT parser on this side: the frames carry cells, and grid.zig copies them
7	//! compile for wasm32-freestanding. The Replica BORROWS its engine; the caller	7	//! in. Deliberately platform-free too — no posix, no fds, no clocks, since this
8	//! owns that lifetime.	8	//! must compile for wasm32-freestanding. The Replica BORROWS its grid; the
		9	//! caller owns that lifetime.
9		10
10	const std = @import("std");	11	const std = @import("std");
11	const Engine = @import("engine.zig").Engine;	12	const grid_mod = @import("grid.zig");
		13	const Grid = grid_mod.Grid;
12	const proto = @import("protocol.zig");	14	const proto = @import("protocol.zig");
13		15
14	pub const Replica = struct {	16	pub const Replica = struct {
15	alloc: std.mem.Allocator,	17	alloc: std.mem.Allocator,
16	/// Borrowed. The engine the frames are replayed into.	18	/// Borrowed. The grid the frames are copied into.
17	eng: *Engine,	19	grid: *Grid,
18	/// The authoritative grid size, learned from snapshot prefixes. Under	20	/// The authoritative grid size, learned from snapshot prefixes. Under
19	/// latest-wins another client's attach or resize can make it differ	21	/// latest-wins another client's attach or resize can make it differ
20	/// from any local tty; the replica follows the grid, not the tty.	22	/// from any local tty; the replica follows the grid, not the tty.
21	grid: proto.Size,	23	grid_size: proto.Size,
22	/// The daemon instance we are talking to, learned from its snapshots,	24	/// The daemon instance we are talking to, learned from its snapshots,
23	/// and quoted back on reconnect so the daemon can tell whether the seq	25	/// and quoted back on reconnect so the daemon can tell whether the seq
24	/// we hold is one of its own (a restarted daemon counts from zero over	26	/// we hold is one of its own (a restarted daemon counts from zero over
@@ -36,55 +38,81 @@ pub const Replica = struct {
36	state_since_attach: bool = false,	38	state_since_attach: bool = false,
37		39
38	pub const Applied = enum {	40	pub const Applied = enum {
39	/// The frame landed; the engine reflects it.	41	/// The frame landed; the grid reflects it.
40	painted,	42	painted,
41	/// A delta that could not be trusted; the engine was not touched. The	43	/// A delta that could not be trusted; the grid was not touched. The
42	/// caller re-attaches with `have_seq=0`, since quoting a seq invites the	44	/// caller re-attaches with `have_seq=0`, since quoting a seq invites the
43	/// delta that cannot fix us. NOT the reconnect path: the transport is	45	/// delta that cannot fix us. NOT the reconnect path: the transport is
44	/// alive and the replica is what is suspect.	46	/// alive and the replica is what is suspect.
45	resync,	47	resync,
46	};	48	};
47		49
48	pub fn init(alloc: std.mem.Allocator, eng: *Engine) Replica {	50	pub fn init(alloc: std.mem.Allocator, g: *Grid) Replica {
49	return .{	51	return .{
50	.alloc = alloc,	52	.alloc = alloc,
51	.eng = eng,	53	.grid = g,
52	.grid = .{ .cols = eng.term.cols, .rows = eng.term.rows },	54	.grid_size = .{ .cols = g.cols, .rows = g.rows },
53	};	55	};
54	}	56	}
55		57
56	/// Consume one replay frame; only `.snapshot` and `.delta` are replay	58	/// Consume one replay frame; only `.snapshot` and `.delta` are replay
57	/// frames.	59	/// frames.
58	///	60	///
59	/// `.snapshot`: a short prefix is `error.BadPayload` with nothing consumed	61	/// `.snapshot`: a payload too short for the prefix or the cursor is
60	/// and `state_since_attach` untouched, because a short snapshot proves	62	/// `error.BadPayload` with nothing consumed and `state_since_attach`
61	/// nothing. A good one adopts seq/epoch/history, resizes when the grid	63	/// untouched, because a short snapshot proves nothing. A row that will
62	/// moved, then reset+feed. `.delta`: arrival alone sets	64	/// not decode is `error.BadPayload` as well rather than `.resync`: a
63	/// `state_since_attach`; a rejected payload is `.resync`.	65	/// snapshot IS the resync, so asking for another cannot fix it, and the
		66	/// caller ends the tile with the error instead of looping.
		67	///
		68	/// `.delta`: arrival alone sets `state_since_attach`; a rejected payload
		69	/// is `.resync`, and the rows that did decode before the bad one stay —
		70	/// the resync's snapshot rewrites every row anyway.
64	pub fn apply(self: *Replica, t: proto.MsgType, payload: []const u8) !Applied {	71	pub fn apply(self: *Replica, t: proto.MsgType, payload: []const u8) !Applied {
65	switch (t) {	72	switch (t) {
66	.snapshot => {	73	.snapshot => {
67	const prefix = try proto.readSnapshotPrefix(payload);	74	const prefix = try proto.readSnapshotPrefix(payload);
		75	const cur = try proto.readSnapshotCursor(payload);
		76	// Refused BEFORE the resize, so a width no row could ever
		77	// decode is reported against the prefix that named it rather
		78	// than against the first row, and so the grid is never left
		79	// at a size nothing can be decoded into.
		80	if (prefix.cols > proto.max_cols) return error.BadPayload;
68	self.state_since_attach = true;	81	self.state_since_attach = true;
69	self.session_epoch = prefix.epoch;	82	self.session_epoch = prefix.epoch;
70	self.last_seq = prefix.seq;	83	self.last_seq = prefix.seq;
71	self.history_rows = prefix.history_rows;	84	self.history_rows = prefix.history_rows;
72	if (prefix.cols != self.grid.cols or prefix.rows != self.grid.rows) {	85	if (prefix.cols != self.grid_size.cols or prefix.rows != self.grid_size.rows) {
73	try self.eng.resize(prefix.cols, prefix.rows);	86	try self.grid.resize(prefix.cols, prefix.rows);
74	self.grid = .{ .cols = prefix.cols, .rows = prefix.rows };	87	self.grid_size = .{ .cols = prefix.cols, .rows = prefix.rows };
		88	}
		89	// Cleared, then every row written: a snapshot is the whole
		90	// screen, and a row the payload happens not to reach must not
		91	// keep what the previous grid had there.
		92	self.grid.clear();
		93	var rest = payload[proto.snapshot_prefix_len + proto.snapshot_cursor_len ..];
		94	var y: u16 = 0;
		95	while (y < prefix.rows) : (y += 1) {
		96	rest = try grid_mod.decodeRow(self.alloc, &self.grid.lines[y], rest, self.grid.cols);
75	}	97	}
76	self.eng.reset();	98	self.grid.cursor = .{ .x = cur.x, .y = cur.y };
77	self.eng.feed(payload[proto.snapshot_prefix_len..]);
78	return .painted;	99	return .painted;
79	},	100	},
80	.delta => {	101	.delta => {
81	self.state_since_attach = true;	102	self.state_since_attach = true;
82	const composed = proto.composeDelta(self.alloc, payload) catch	103	const hdr = proto.readDeltaHeader(payload) catch return .resync;
83	return .resync;	104	var it = proto.deltaRowIterator(payload);
84	defer self.alloc.free(composed.bytes);	105	var seen: usize = 0;
85	self.history_rows = composed.header.history_rows;	106	while (it.next() catch return .resync) \|row\| {
86	self.last_seq = composed.header.seq;	107	self.grid.applyRow(row.row, row.bytes) catch return .resync;
87	self.eng.feed(composed.bytes);	108	seen += 1;
		109	}
		110	// The header's row_count is authoritative: a payload carrying
		111	// fewer rows than it claims is a truncation, not a short frame.
		112	if (seen != hdr.row_count) return .resync;
		113	self.history_rows = hdr.history_rows;
		114	self.last_seq = hdr.seq;
		115	self.grid.cursor = .{ .x = hdr.cursor_x, .y = hdr.cursor_y };
88	return .painted;	116	return .painted;
89	},	117	},
90	else => unreachable, // not a replay frame; callers dispatch	118	else => unreachable, // not a replay frame; callers dispatch
@@ -110,30 +138,61 @@ pub const Replica = struct {
110	// Tests. The wire layouts these build are golden-pinned in protocol.zig, so	138	// Tests. The wire layouts these build are golden-pinned in protocol.zig, so
111	// constructing real payloads here is mechanical, not speculative.	139	// constructing real payloads here is mechanical, not speculative.
112		140
		141	/// A snapshot payload: prefix, cursor, then one CellRow per row of `rows`,
		142	/// each row being the plain text to put at column 0.
113	fn testSnapshot(	143	fn testSnapshot(
114	alloc: std.mem.Allocator,	144	alloc: std.mem.Allocator,
115	p: proto.SnapshotPrefix,	145	p: proto.SnapshotPrefix,
116	state: []const u8,	146	cursor: proto.SnapshotCursor,
		147	rows: []const []const u8,
117	) ![]u8 {	148	) ![]u8 {
118	var payload = try alloc.alloc(u8, proto.snapshot_prefix_len + state.len);	149	var out: std.ArrayList(u8) = .empty;
119	proto.writeSnapshotPrefix(payload[0..proto.snapshot_prefix_len], p);	150	errdefer out.deinit(alloc);
120	@memcpy(payload[proto.snapshot_prefix_len..], state);	151	var pbuf: [proto.snapshot_prefix_len]u8 = undefined;
121	return payload;	152	proto.writeSnapshotPrefix(&pbuf, p);
		153	try out.appendSlice(alloc, &pbuf);
		154	var cbuf: [proto.snapshot_cursor_len]u8 = undefined;
		155	proto.writeSnapshotCursor(&cbuf, cursor.x, cursor.y);
		156	try out.appendSlice(alloc, &cbuf);
		157	for (rows) \|text\| try appendTextRow(&out, alloc, text);
		158	return out.toOwnedSlice(alloc);
122	}	159	}
123		160
124	test "snapshot replay: prefix consumed, state fed, epoch and seq adopted" {	161	/// One CellRow of default-styled narrow ASCII cells.
125	const alloc = std.testing.allocator;	162	fn appendTextRow(list: *std.ArrayList(u8), alloc: std.mem.Allocator, text: []const u8) !void {
126	var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	163	var w = try proto.CellRowWriter.begin(list, alloc);
127	defer eng.deinit();	164	errdefer w.deinit();
128	var r = Replica.init(alloc, eng);	165	for (text) \|ch\| try w.cell(.{}, .narrow, &[_]u8{ch});
		166	w.finish();
		167	}
		168
		169	/// The CellRow bytes alone, for a delta row.
		170	fn testRow(alloc: std.mem.Allocator, text: []const u8) ![]u8 {
		171	var out: std.ArrayList(u8) = .empty;
		172	errdefer out.deinit(alloc);
		173	try appendTextRow(&out, alloc, text);
		174	return out.toOwnedSlice(alloc);
		175	}
129		176
		177	test "snapshot replay: prefix consumed, rows copied in, epoch and seq adopted" {
		178	const alloc = std.testing.allocator;
		179	const g = try Grid.init(alloc, 80, 24);
		180	defer g.deinit();
		181	var r = Replica.init(alloc, g);
		182
		183	// Two rows authored, not one: a decoder that stopped after the first row
		184	// would still make the assertion about "hi" pass.
		185	var rows: [24][]const u8 = undefined;
		186	for (&rows) \|row\| row. = "";
		187	rows[0] = "hi";
		188	rows[1] = "there";
130	const payload = try testSnapshot(alloc, .{	189	const payload = try testSnapshot(alloc, .{
131	.seq = 7,	190	.seq = 7,
132	.history_rows = 3,	191	.history_rows = 3,
133	.cols = 80,	192	.cols = 80,
134	.rows = 24,	193	.rows = 24,
135	.epoch = 0xABCD,	194	.epoch = 0xABCD,
136	}, "hi");	195	}, .{ .x = 2, .y = 1 }, &rows);
137	defer alloc.free(payload);	196	defer alloc.free(payload);
138		197
139	try std.testing.expectEqual(Replica.Applied.painted, try r.apply(.snapshot, payload));	198	try std.testing.expectEqual(Replica.Applied.painted, try r.apply(.snapshot, payload));
@@ -141,90 +200,202 @@ test "snapshot replay: prefix consumed, state fed, epoch and seq adopted" {
141	try std.testing.expectEqual(@as(u64, 0xABCD), r.session_epoch);	200	try std.testing.expectEqual(@as(u64, 0xABCD), r.session_epoch);
142	try std.testing.expectEqual(@as(u32, 3), r.history_rows);	201	try std.testing.expectEqual(@as(u32, 3), r.history_rows);
143	try std.testing.expect(r.state_since_attach);	202	try std.testing.expect(r.state_since_attach);
		203	// The cursor rides in the body and is adopted with the rows.
		204	try std.testing.expectEqual(@as(u16, 2), g.cursor.x);
		205	try std.testing.expectEqual(@as(u16, 1), g.cursor.y);
		206
		207	const dump = try g.dumpPlain(alloc);
		208	defer alloc.free(dump);
		209	try std.testing.expectEqualStrings("hi\nthere", dump);
		210	}
		211
		212	test "snapshot replay: a row the previous grid held is cleared, not kept" {
		213	// A snapshot is the whole screen. Without the clear, a shorter payload's
		214	// unwritten rows would keep the old session's text under the new one.
		215	const alloc = std.testing.allocator;
		216	const g = try Grid.init(alloc, 8, 3);
		217	defer g.deinit();
		218	var r = Replica.init(alloc, g);
		219
		220	const first = try testSnapshot(alloc, .{
		221	.seq = 1,
		222	.history_rows = 0,
		223	.cols = 8,
		224	.rows = 3,
		225	.epoch = 1,
		226	}, .{ .x = 0, .y = 0 }, &.{ "aaa", "bbb", "ccc" });
		227	defer alloc.free(first);
		228	_ = try r.apply(.snapshot, first);
144		229
145	const dump = try eng.dumpPlain(alloc);	230	const second = try testSnapshot(alloc, .{
		231	.seq = 2,
		232	.history_rows = 0,
		233	.cols = 8,
		234	.rows = 3,
		235	.epoch = 1,
		236	}, .{ .x = 0, .y = 0 }, &.{ "z", "", "" });
		237	defer alloc.free(second);
		238	_ = try r.apply(.snapshot, second);
		239
		240	const dump = try g.dumpPlain(alloc);
146	defer alloc.free(dump);	241	defer alloc.free(dump);
147	try std.testing.expect(std.mem.startsWith(u8, dump, "hi"));	242	try std.testing.expectEqualStrings("z", dump);
148	}	243	}
149		244
150	test "snapshot at a new grid size resizes the replica engine first" {	245	test "snapshot at a new grid size resizes the replica grid first" {
151	const alloc = std.testing.allocator;	246	const alloc = std.testing.allocator;
152	var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	247	const g = try Grid.init(alloc, 80, 24);
153	defer eng.deinit();	248	defer g.deinit();
154	var r = Replica.init(alloc, eng);	249	var r = Replica.init(alloc, g);
155		250
		251	var rows: [30][]const u8 = undefined;
		252	for (&rows) \|row\| row. = "";
		253	rows[0] = "wide";
156	const payload = try testSnapshot(alloc, .{	254	const payload = try testSnapshot(alloc, .{
157	.seq = 1,	255	.seq = 1,
158	.history_rows = 0,	256	.history_rows = 0,
159	.cols = 100,	257	.cols = 100,
160	.rows = 30,	258	.rows = 30,
161	.epoch = 1,	259	.epoch = 1,
162	}, "wide");	260	}, .{ .x = 4, .y = 0 }, &rows);
163	defer alloc.free(payload);	261	defer alloc.free(payload);
164		262
165	try std.testing.expectEqual(Replica.Applied.painted, try r.apply(.snapshot, payload));	263	try std.testing.expectEqual(Replica.Applied.painted, try r.apply(.snapshot, payload));
166	try std.testing.expectEqual(@as(u16, 100), r.grid.cols);	264	try std.testing.expectEqual(@as(u16, 100), r.grid_size.cols);
167	try std.testing.expectEqual(@as(u16, 30), r.grid.rows);	265	try std.testing.expectEqual(@as(u16, 30), r.grid_size.rows);
168	try std.testing.expectEqual(@as(u16, 100), eng.term.cols);	266	try std.testing.expectEqual(@as(u16, 100), g.cols);
169	try std.testing.expectEqual(@as(u16, 30), eng.term.rows);	267	try std.testing.expectEqual(@as(u16, 30), g.rows);
170	}	268	}
171		269
172	test "short snapshot proves nothing: BadPayload, state_since_attach untouched" {	270	test "short snapshot proves nothing: BadPayload, state_since_attach untouched" {
173	const alloc = std.testing.allocator;	271	const alloc = std.testing.allocator;
174	var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	272	const g = try Grid.init(alloc, 80, 24);
175	defer eng.deinit();	273	defer g.deinit();
176	var r = Replica.init(alloc, eng);	274	var r = Replica.init(alloc, g);
177		275
178	const short = [_]u8{0} ** (proto.snapshot_prefix_len - 1);	276	const short = [_]u8{0} ** (proto.snapshot_prefix_len - 1);
179	try std.testing.expectError(error.BadPayload, r.apply(.snapshot, &short));	277	try std.testing.expectError(error.BadPayload, r.apply(.snapshot, &short));
		278	// A whole prefix and no cursor is short too: the four bytes behind it
		279	// would otherwise be read out of the first row.
		280	const no_cursor = [_]u8{0} ** proto.snapshot_prefix_len;
		281	try std.testing.expectError(error.BadPayload, r.apply(.snapshot, &no_cursor));
180	try std.testing.expect(!r.state_since_attach);	282	try std.testing.expect(!r.state_since_attach);
181	try std.testing.expectEqual(@as(u64, 0), r.last_seq);	283	try std.testing.expectEqual(@as(u64, 0), r.last_seq);
182	}	284	}
183		285
184	test "delta replay: composed rows land, last_seq advances, history follows" {	286	test "replica: a snapshot whose rows do not decode is BadPayload, not resync" {
		287	// A snapshot IS the resync, so answering one with "please resync" is a
		288	// loop. The caller ends the tile on the error instead.
		289	const alloc = std.testing.allocator;
		290	const g = try Grid.init(alloc, 80, 24);
		291	defer g.deinit();
		292	var r = Replica.init(alloc, g);
		293
		294	// A body of VT bytes: what a daemon on the far side of the renumbering
		295	// would have sent, and what this decoder must refuse rather than paint.
		296	var payload: std.ArrayList(u8) = .empty;
		297	defer payload.deinit(alloc);
		298	var pbuf: [proto.snapshot_prefix_len]u8 = undefined;
		299	proto.writeSnapshotPrefix(&pbuf, .{
		300	.seq = 1,
		301	.history_rows = 0,
		302	.cols = 80,
		303	.rows = 24,
		304	.epoch = 1,
		305	});
		306	try payload.appendSlice(alloc, &pbuf);
		307	var cbuf: [proto.snapshot_cursor_len]u8 = undefined;
		308	proto.writeSnapshotCursor(&cbuf, 0, 0);
		309	try payload.appendSlice(alloc, &cbuf);
		310	try payload.appendSlice(alloc, "\x1b[1mVT");
		311
		312	try std.testing.expectError(error.BadPayload, r.apply(.snapshot, payload.items));
		313	}
		314
		315	test "replica: a snapshot claiming more columns than a row can hold is refused at the prefix" {
		316	const alloc = std.testing.allocator;
		317	const g = try Grid.init(alloc, 80, 24);
		318	defer g.deinit();
		319	var r = Replica.init(alloc, g);
		320
		321	var payload: std.ArrayList(u8) = .empty;
		322	defer payload.deinit(alloc);
		323	var pbuf: [proto.snapshot_prefix_len]u8 = undefined;
		324	proto.writeSnapshotPrefix(&pbuf, .{
		325	.seq = 1,
		326	.history_rows = 0,
		327	.cols = proto.max_cols + 1,
		328	.rows = 1,
		329	.epoch = 1,
		330	});
		331	try payload.appendSlice(alloc, &pbuf);
		332	var cbuf: [proto.snapshot_cursor_len]u8 = undefined;
		333	proto.writeSnapshotCursor(&cbuf, 0, 0);
		334	try payload.appendSlice(alloc, &cbuf);
		335	try appendTextRow(&payload, alloc, "a");
		336
		337	try std.testing.expectError(error.BadPayload, r.apply(.snapshot, payload.items));
		338	// The grid is still the one it was, at the size it was: nothing resized
		339	// to a width no row could ever be decoded into.
		340	try std.testing.expectEqual(@as(u16, 80), g.cols);
		341	try std.testing.expectEqual(@as(u16, 24), g.rows);
		342	}
		343
		344	test "delta replay: rows land, last_seq advances, history and cursor follow" {
185	const alloc = std.testing.allocator;	345	const alloc = std.testing.allocator;
186	var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	346	const g = try Grid.init(alloc, 80, 24);
187	defer eng.deinit();	347	defer g.deinit();
188	var r = Replica.init(alloc, eng);	348	var r = Replica.init(alloc, g);
189		349
		350	var rows: [24][]const u8 = undefined;
		351	for (&rows) \|row\| row. = "";
		352	rows[0] = "hi";
190	const snap = try testSnapshot(alloc, .{	353	const snap = try testSnapshot(alloc, .{
191	.seq = 7,	354	.seq = 7,
192	.history_rows = 0,	355	.history_rows = 0,
193	.cols = 80,	356	.cols = 80,
194	.rows = 24,	357	.rows = 24,
195	.epoch = 0xABCD,	358	.epoch = 0xABCD,
196	}, "hi");	359	}, .{ .x = 0, .y = 0 }, &rows);
197	defer alloc.free(snap);	360	defer alloc.free(snap);
198	_ = try r.apply(.snapshot, snap);	361	_ = try r.apply(.snapshot, snap);
199		362
		363	// Two rows, and neither of them row 0: a delta that applied its rows at
		364	// the wrong index would still show the text somewhere.
200	var delta: std.ArrayList(u8) = .empty;	365	var delta: std.ArrayList(u8) = .empty;
201	defer delta.deinit(alloc);	366	defer delta.deinit(alloc);
202	try proto.appendDeltaHeader(&delta, alloc, .{	367	try proto.appendDeltaHeader(&delta, alloc, .{
203	.seq = 8,	368	.seq = 8,
204	.history_rows = 2,	369	.history_rows = 2,
205	.cursor_x = 2,	370	.cursor_x = 2,
206	.cursor_y = 0,	371	.cursor_y = 3,
207	.row_count = 1,	372	.row_count = 2,
208	});	373	});
209	try proto.appendDeltaRow(&delta, alloc, 0, "yo");	374	const row_two = try testRow(alloc, "yo");
		375	defer alloc.free(row_two);
		376	const row_three = try testRow(alloc, "ok");
		377	defer alloc.free(row_three);
		378	try proto.appendDeltaRow(&delta, alloc, 2, row_two);
		379	try proto.appendDeltaRow(&delta, alloc, 3, row_three);
210		380
211	try std.testing.expectEqual(Replica.Applied.painted, try r.apply(.delta, delta.items));	381	try std.testing.expectEqual(Replica.Applied.painted, try r.apply(.delta, delta.items));
212	try std.testing.expectEqual(@as(u64, 8), r.last_seq);	382	try std.testing.expectEqual(@as(u64, 8), r.last_seq);
213	try std.testing.expectEqual(@as(u32, 2), r.history_rows);	383	try std.testing.expectEqual(@as(u32, 2), r.history_rows);
		384	try std.testing.expectEqual(@as(u16, 2), g.cursor.x);
		385	try std.testing.expectEqual(@as(u16, 3), g.cursor.y);
214		386
215	const dump = try eng.dumpPlain(alloc);	387	const dump = try g.dumpPlain(alloc);
216	defer alloc.free(dump);	388	defer alloc.free(dump);
217	try std.testing.expect(std.mem.startsWith(u8, dump, "yo"));	389	try std.testing.expectEqualStrings("hi\n\nyo\nok", dump);
218	}	390	}
219		391
220	test "delta decode failure reports .resync — and its arrival still proves admission" {	392	test "delta decode failure reports .resync — and its arrival still proves admission" {
221	const alloc = std.testing.allocator;	393	const alloc = std.testing.allocator;
222	var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	394	const g = try Grid.init(alloc, 80, 24);
223	defer eng.deinit();	395	defer g.deinit();
224	var r = Replica.init(alloc, eng);	396	var r = Replica.init(alloc, g);
225		397
226	// Header claims two rows, payload carries one: composeDelta rejects it	398	// Header claims two rows, payload carries one: the row_count check.
227	// (the row_count check protocol.zig pins).
228	var delta: std.ArrayList(u8) = .empty;	399	var delta: std.ArrayList(u8) = .empty;
229	defer delta.deinit(alloc);	400	defer delta.deinit(alloc);
230	try proto.appendDeltaHeader(&delta, alloc, .{	401	try proto.appendDeltaHeader(&delta, alloc, .{
@@ -234,7 +405,9 @@ test "delta decode failure reports .resync — and its arrival still proves admi
234	.cursor_y = 0,	405	.cursor_y = 0,
235	.row_count = 2,	406	.row_count = 2,
236	});	407	});
237	try proto.appendDeltaRow(&delta, alloc, 0, "x");	408	const row = try testRow(alloc, "x");
		409	defer alloc.free(row);
		410	try proto.appendDeltaRow(&delta, alloc, 0, row);
238		411
239	try std.testing.expectEqual(Replica.Applied.resync, try r.apply(.delta, delta.items));	412	try std.testing.expectEqual(Replica.Applied.resync, try r.apply(.delta, delta.items));
240	// The subtlety the CLI relies on: a delta's ARRIVAL alone proves the	413	// The subtlety the CLI relies on: a delta's ARRIVAL alone proves the
@@ -244,23 +417,65 @@ test "delta decode failure reports .resync — and its arrival still proves admi
244	try std.testing.expectEqual(@as(u64, 0), r.last_seq);	417	try std.testing.expectEqual(@as(u64, 0), r.last_seq);
245	}	418	}
246		419
		420	test "replica: a delta whose row is wider than the grid is resync and the grid is untouched" {
		421	const alloc = std.testing.allocator;
		422	const g = try Grid.init(alloc, 4, 2);
		423	defer g.deinit();
		424	var r = Replica.init(alloc, g);
		425
		426	const seed = try testSnapshot(alloc, .{
		427	.seq = 1,
		428	.history_rows = 0,
		429	.cols = 4,
		430	.rows = 2,
		431	.epoch = 1,
		432	}, .{ .x = 0, .y = 0 }, &.{ "keep", "me" });
		433	defer alloc.free(seed);
		434	_ = try r.apply(.snapshot, seed);
		435
		436	var delta: std.ArrayList(u8) = .empty;
		437	defer delta.deinit(alloc);
		438	try proto.appendDeltaHeader(&delta, alloc, .{
		439	.seq = 2,
		440	.history_rows = 5,
		441	.cursor_x = 1,
		442	.cursor_y = 1,
		443	.row_count = 1,
		444	});
		445	// Five cells into a four-column grid.
		446	const too_wide = try testRow(alloc, "abcde");
		447	defer alloc.free(too_wide);
		448	try proto.appendDeltaRow(&delta, alloc, 0, too_wide);
		449
		450	try std.testing.expectEqual(Replica.Applied.resync, try r.apply(.delta, delta.items));
		451	// Neither the row nor the resume coordinates moved: `decodeRow` validates
		452	// before it writes, so a refused row leaves what was there.
		453	const dump = try g.dumpPlain(alloc);
		454	defer alloc.free(dump);
		455	try std.testing.expectEqualStrings("keep\nme", dump);
		456	try std.testing.expectEqual(@as(u64, 1), r.last_seq);
		457	try std.testing.expectEqual(@as(u32, 0), r.history_rows);
		458	}
		459
247	test "attach args: first attach quotes (0,0); after a snapshot, (last_seq, epoch)" {	460	test "attach args: first attach quotes (0,0); after a snapshot, (last_seq, epoch)" {
248	const alloc = std.testing.allocator;	461	const alloc = std.testing.allocator;
249	var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	462	const g = try Grid.init(alloc, 80, 24);
250	defer eng.deinit();	463	defer g.deinit();
251	var r = Replica.init(alloc, eng);	464	var r = Replica.init(alloc, g);
252		465
253	const fresh = r.attachArgs();	466	const fresh = r.attachArgs();
254	try std.testing.expectEqual(@as(u64, 0), fresh.have_seq);	467	try std.testing.expectEqual(@as(u64, 0), fresh.have_seq);
255	try std.testing.expectEqual(@as(u64, 0), fresh.have_epoch);	468	try std.testing.expectEqual(@as(u64, 0), fresh.have_epoch);
256		469
		470	var rows: [24][]const u8 = undefined;
		471	for (&rows) \|row\| row. = "";
257	const snap = try testSnapshot(alloc, .{	472	const snap = try testSnapshot(alloc, .{
258	.seq = 42,	473	.seq = 42,
259	.history_rows = 0,	474	.history_rows = 0,
260	.cols = 80,	475	.cols = 80,
261	.rows = 24,	476	.rows = 24,
262	.epoch = 0xFEED,	477	.epoch = 0xFEED,
263	}, "");	478	}, .{ .x = 0, .y = 0 }, &rows);
264	defer alloc.free(snap);	479	defer alloc.free(snap);
265	_ = try r.apply(.snapshot, snap);	480	_ = try r.apply(.snapshot, snap);
266		481
@@ -271,9 +486,9 @@ test "attach args: first attach quotes (0,0); after a snapshot, (last_seq, epoch
271		486
272	test "scrollStart: rows count up from the live viewport top, saturating at row 0" {	487	test "scrollStart: rows count up from the live viewport top, saturating at row 0" {
273	const alloc = std.testing.allocator;	488	const alloc = std.testing.allocator;
274	var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	489	const g = try Grid.init(alloc, 80, 24);
275	defer eng.deinit();	490	defer g.deinit();
276	var r = Replica.init(alloc, eng);	491	var r = Replica.init(alloc, g);
277	r.history_rows = 100;	492	r.history_rows = 100;
278		493
279	// A screenful at a time, as the scroll keys ask...	494	// A screenful at a time, as the scroll keys ask...

src/server/server.zig

Old		New
@@ -8,7 +8,8 @@ const std = @import("std");
8	const Engine = @import("term").engine.Engine;	8	const Engine = @import("term").engine.Engine;
9	const Pty = @import("pty").Pty;	9	const Pty = @import("pty").Pty;
10	const proto = @import("term").protocol;	10	const proto = @import("term").protocol;
11	const DeltaTracker = @import("term").delta.DeltaTracker;	11	const delta_mod = @import("term").delta;
		12	const DeltaTracker = delta_mod.DeltaTracker;
12	const cmdmod = @import("cmd.zig");	13	const cmdmod = @import("cmd.zig");
13	const shellint = @import("shellint.zig");	14	const shellint = @import("shellint.zig");
14	const sockpath = @import("sockpath");	15	const sockpath = @import("sockpath");
@@ -1931,12 +1932,15 @@ pub const Server = struct {
1931	// disturbs another's live stream.	1932	// disturbs another's live stream.
1932	const si = self.clients[i].?.session orelse return;	1933	const si = self.clients[i].?.session orelse return;
1933	const req = proto.decodeScrollbackReq(frame.payload) catch return;	1934	const req = proto.decodeScrollbackReq(frame.payload) catch return;
1934	const rows = self.ses(si).eng.dumpScrollback(self.alloc, req.start, req.count) catch return;	1935	const got = self.ses(si).eng.encodeScrollback(self.alloc, req.start, req.count) catch return;
1935	defer self.alloc.free(rows);	1936	defer self.alloc.free(got.bytes);
1936	const payload = self.alloc.alloc(u8, 6 + rows.len) catch return;	1937	const payload = self.alloc.alloc(u8, 6 + got.bytes.len) catch return;
1937	defer self.alloc.free(payload);	1938	defer self.alloc.free(payload);
1938	@memcpy(payload[0..6], &proto.encodeScrollbackReq(req.start, req.count));	1939	// The echoed header is what the encoder CLAMPED to, not what was
1939	@memcpy(payload[6..], rows);	1940	// asked for: the client decodes exactly `count` rows out of the body,
		1941	// and a request past the end of history answers with fewer.
		1942	@memcpy(payload[0..6], &proto.encodeScrollbackReq(got.first, got.count));
		1943	@memcpy(payload[6..], got.bytes);
1940	_ = self.queueFrame(i, .scrollback_chunk, payload);	1944	_ = self.queueFrame(i, .scrollback_chunk, payload);
1941	}	1945	}
1942		1946
@@ -2413,9 +2417,12 @@ pub const Server = struct {
2413	/// serialization per update purely to measure the saving; fine for a	2417	/// serialization per update purely to measure the saving; fine for a
2414	/// prototype.	2418	/// prototype.
2415	fn accrueSnapshotEquiv(self: *Server, si: usize) void {	2419	fn accrueSnapshotEquiv(self: *Server, si: usize) void {
2416	if (self.ses(si).eng.dumpState(self.alloc)) \|state\| {	2420	// Built and thrown away: the stat means "what a full snapshot would
2417	self.stats.snapshot_equiv_bytes += proto.snapshot_prefix_len + state.len;	2421	// have cost", so it has to measure the payload this daemon would
2418	self.alloc.free(state);	2422	// actually have sent rather than any other serialization of the grid.
		2423	if (self.buildSnapshotPayload(si)) \|payload\| {
		2424	self.stats.snapshot_equiv_bytes += payload.len;
		2425	self.alloc.free(payload);
2419	} else \|_\| {}	2426	} else \|_\| {}
2420	}	2427	}
2421		2428
@@ -2691,24 +2698,19 @@ pub const Server = struct {
2691	};	2698	};
2692	}	2699	}
2693		2700
2694	/// The snapshot payload for the grid as it stands: the fixed prefix ++	2701	/// The snapshot payload for the grid as it stands: the fixed prefix, the
2695	/// full-state dump. Caller owns the result. Reads tracker.seq, so it	2702	/// cursor, then every viewport row as cells. Caller owns the result.
2696	/// must be called after the rebuild that stamps it.	2703	/// Reads tracker.seq, so it must be called after the rebuild that stamps
		2704	/// it.
2697	fn buildSnapshotPayload(self: *Server, si: usize) ![]u8 {	2705	fn buildSnapshotPayload(self: *Server, si: usize) ![]u8 {
2698	const s = self.ses(si);	2706	const s = self.ses(si);
2699	const state = try s.eng.dumpState(self.alloc);	2707	return delta_mod.buildSnapshot(self.alloc, s.eng, .{
2700	defer self.alloc.free(state);
2701	const payload = try self.alloc.alloc(u8, proto.snapshot_prefix_len + state.len);
2702	errdefer self.alloc.free(payload);
2703	proto.writeSnapshotPrefix(payload[0..proto.snapshot_prefix_len], .{
2704	.seq = s.tracker.seq,	2708	.seq = s.tracker.seq,
2705	.history_rows = s.eng.historyRows(),	2709	.history_rows = s.eng.historyRows(),
2706	.cols = self.colsNow(si),	2710	.cols = self.colsNow(si),
2707	.rows = self.rowsNow(si),	2711	.rows = self.rowsNow(si),
2708	.epoch = s.epoch,	2712	.epoch = s.epoch,
2709	});	2713	});
2710	@memcpy(payload[proto.snapshot_prefix_len..], state);
2711	return payload;
2712	}	2714	}
2713		2715
2714	/// One call because a rebuild is the ONLY thing that narrows the servable	2716	/// One call because a rebuild is the ONLY thing that narrows the servable

src/server/server_test_attach.zig

Old		New
@@ -1,5 +1,6 @@
1	const std = @import("std");	1	const std = @import("std");
2	const Engine = @import("term").engine.Engine;	2	const grid_mod = @import("term").grid;
		3	const Grid = @import("term").grid.Grid;
3	const proto = @import("term").protocol;	4	const proto = @import("term").protocol;
4	const TmpDir = @import("testtmp").TmpDir;	5	const TmpDir = @import("testtmp").TmpDir;
5	const h = @import("server_test_harness.zig");	6	const h = @import("server_test_harness.zig");
@@ -23,7 +24,7 @@ const serverThread = h.serverThread;
23	/// session keeps broadcasting underneath it.	24	/// session keeps broadcasting underneath it.
24	const ApplyEach = struct {	25	const ApplyEach = struct {
25	alloc: std.mem.Allocator,	26	alloc: std.mem.Allocator,
26	replica: *Engine,	27	replica: *Grid,
27		28
28	fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {	29	fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
29	const self: *@This() = @ptrCast(@alignCast(ctx.?));	30	const self: *@This() = @ptrCast(@alignCast(ctx.?));
@@ -162,7 +163,7 @@ test "Server: a full client table refuses the next attach instead of displacing
162		163
163	// The client that attached first is still attached and still fed.	164	// The client that attached first is still attached and still fed.
164	try proto.writeFrame(streams[0].handle, .input, "echo still-here\n");	165	try proto.writeFrame(streams[0].handle, .input, "echo still-here\n");
165	var replica = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	166	var replica = try Grid.init(alloc, 80, 24);
166	defer replica.deinit();	167	defer replica.deinit();
167	try std.testing.expect(try awaitReplicaText(alloc, streams[0].handle, 10_000, .{	168	try std.testing.expect(try awaitReplicaText(alloc, streams[0].handle, 10_000, .{
168	.replica = replica,	169	.replica = replica,
@@ -186,12 +187,12 @@ test "Server: two clients converge on one session" {
186	// Input through A must reach both replicas.	187	// Input through A must reach both replicas.
187	try proto.writeFrame(a.handle, .input, "echo both-see-this\n");	188	try proto.writeFrame(a.handle, .input, "echo both-see-this\n");
188		189
189	var replica_a = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	190	var replica_a = try Grid.init(alloc, 80, 24);
190	defer replica_a.deinit();	191	defer replica_a.deinit();
191	var replica_b = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	192	var replica_b = try Grid.init(alloc, 80, 24);
192	defer replica_b.deinit();	193	defer replica_b.deinit();
193		194
194	for ([_]struct { fd: std.posix.fd_t, rep: *Engine }{	195	for ([_]struct { fd: std.posix.fd_t, rep: *Grid }{
195	.{ .fd = a.handle, .rep = replica_a },	196	.{ .fd = a.handle, .rep = replica_a },
196	.{ .fd = b.handle, .rep = replica_b },	197	.{ .fd = b.handle, .rep = replica_b },
197	}) \|side\| {	198	}) \|side\| {
@@ -201,35 +202,40 @@ test "Server: two clients converge on one session" {
201	}));	202	}));
202	}	203	}
203		204
204	// Byte-level convergence: both replicas match the daemon exactly.	205	// Cell-level convergence: both replicas show what the daemon shows. The
205	try proto.writeFrame(a.handle, .debug_dump, &.{1});	206	// daemon's own dump is asked for as PLAIN and compared through
206	var daemon_vt: ?[]u8 = null;	207	// `h.trimRowTails`, because trailing spaces are the one formatting
207	defer if (daemon_vt) \|d\| alloc.free(d);	208	// difference between the two dumps and the VT formatter that fed the old
		209	// wire trimmed them too.
		210	try proto.writeFrame(a.handle, .debug_dump, &.{0});
		211	var daemon_plain: ?[]u8 = null;
		212	defer if (daemon_plain) \|d\| alloc.free(d);
208	var feed_a: ApplyEach = .{ .alloc = alloc, .replica = replica_a };	213	var feed_a: ApplyEach = .{ .alloc = alloc, .replica = replica_a };
209	if (try h.awaitFrameOnSink(alloc, a.handle, .dump_reply, 5000, feed_a.sink())) \|frame\| {	214	if (try h.awaitFrameOnSink(alloc, a.handle, .dump_reply, 5000, feed_a.sink())) \|frame\| {
210	daemon_vt = frame.payload; // ownership taken	215	daemon_plain = try h.trimRowTails(alloc, frame.payload);
		216	frame.deinit(alloc);
211	}	217	}
212	try std.testing.expect(daemon_vt != null);	218	try std.testing.expect(daemon_plain != null);
213		219
214	const va = try replica_a.dumpVt(alloc);	220	const va = try replica_a.dumpPlain(alloc);
215	defer alloc.free(va);	221	defer alloc.free(va);
216	try std.testing.expectEqualStrings(daemon_vt.?, va);	222	try std.testing.expectEqualStrings(daemon_plain.?, va);
217		223
218	// B saw the same broadcasts but may still have some in its socket	224	// B saw the same broadcasts but may still have some in its socket
219	// buffer: drain until it agrees with the dump A already fetched.	225	// buffer: drain until it agrees with the dump A already fetched.
220	const Converge = struct {	226	const Converge = struct {
221	alloc: std.mem.Allocator,	227	alloc: std.mem.Allocator,
222	replica: *Engine,	228	replica: *Grid,
223	want: []const u8,	229	want: []const u8,
224	fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {	230	fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
225	const self: *@This() = @ptrCast(@alignCast(ctx.?));	231	const self: *@This() = @ptrCast(@alignCast(ctx.?));
226	try applyFrame(self.alloc, self.replica, frame);	232	try applyFrame(self.alloc, self.replica, frame);
227	const vt = try self.replica.dumpVt(self.alloc);	233	const plain = try self.replica.dumpPlain(self.alloc);
228	defer self.alloc.free(vt);	234	defer self.alloc.free(plain);
229	if (std.mem.eql(u8, self.want, vt)) return error.Converged;	235	if (std.mem.eql(u8, self.want, plain)) return error.Converged;
230	}	236	}
231	};	237	};
232	var conv: Converge = .{ .alloc = alloc, .replica = replica_b, .want = daemon_vt.? };	238	var conv: Converge = .{ .alloc = alloc, .replica = replica_b, .want = daemon_plain.? };
233	_ = h.awaitFrameOnSink(alloc, b.handle, h.never_from_daemon, 5000, .{	239	_ = h.awaitFrameOnSink(alloc, b.handle, h.never_from_daemon, 5000, .{
234	.ctx = &conv,	240	.ctx = &conv,
235	.on = Converge.on,	241	.on = Converge.on,
@@ -239,9 +245,9 @@ test "Server: two clients converge on one session" {
239	};	245	};
240	// Dumped again rather than kept from the sink: the sink BORROWS its	246	// Dumped again rather than kept from the sink: the sink BORROWS its
241	// frame, and the comparison it made is the one this reproduces.	247	// frame, and the comparison it made is the one this reproduces.
242	const vb = try replica_b.dumpVt(alloc);	248	const vb = try replica_b.dumpPlain(alloc);
243	defer alloc.free(vb);	249	defer alloc.free(vb);
244	try std.testing.expectEqualStrings(daemon_vt.?, vb);	250	try std.testing.expectEqualStrings(daemon_plain.?, vb);
245	}	251	}
246		252
247	test "Server: a same-size join snapshots the joiner only" {	253	test "Server: a same-size join snapshots the joiner only" {
@@ -255,7 +261,7 @@ test "Server: a same-size join snapshots the joiner only" {
255	const a = try dial.dialAttach(td.sock_path, 80, 24);	261	const a = try dial.dialAttach(td.sock_path, 80, 24);
256	defer a.close();	262	defer a.close();
257		263
258	var replica_a = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	264	var replica_a = try Grid.init(alloc, 80, 24);
259	defer replica_a.deinit();	265	defer replica_a.deinit();
260		266
261	// Drain A's own join snapshot and everything the shell's startup	267	// Drain A's own join snapshot and everything the shell's startup
@@ -308,7 +314,7 @@ test "Server: a same-size join snapshots the joiner only" {
308	}));	314	}));
309		315
310	// ...and B, the joiner, sees the same input.	316	// ...and B, the joiner, sees the same input.
311	var replica_b = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	317	var replica_b = try Grid.init(alloc, 80, 24);
312	defer replica_b.deinit();	318	defer replica_b.deinit();
313	try std.testing.expect(try awaitReplicaText(alloc, b.handle, 10_000, .{	319	try std.testing.expect(try awaitReplicaText(alloc, b.handle, 10_000, .{
314	.replica = replica_b,	320	.replica = replica_b,
@@ -482,7 +488,7 @@ test "Server: a size the grid refuses never becomes a claim" {
482	// The replica starts blank on purpose — a delta carries every row it	488	// The replica starts blank on purpose — a delta carries every row it
483	// changed, so the row the marker lands on arrives whole, and a	489	// changed, so the row the marker lands on arrives whole, and a
484	// snapshot is the failure rather than the way the text gets here.	490	// snapshot is the failure rather than the way the text gets here.
485	var seen = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	491	var seen = try Grid.init(alloc, 80, 24);
486	defer seen.deinit();	492	defer seen.deinit();
487	try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{	493	try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{
488	.replica = seen,	494	.replica = seen,
@@ -504,7 +510,7 @@ test "Server: a size the grid refuses never becomes a claim" {
504	{	510	{
505	// Blank again, at the size the grid has moved to; a snapshot here	511	// Blank again, at the size the grid has moved to; a snapshot here
506	// would mean D's keystroke claimed a size it was refused.	512	// would mean D's keystroke claimed a size it was refused.
507	var seen = try Engine.init(alloc, .{ .cols = 100, .rows = 30 });	513	var seen = try Grid.init(alloc, 100, 30);
508	defer seen.deinit();	514	defer seen.deinit();
509	try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{	515	try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{
510	.replica = seen,	516	.replica = seen,
@@ -670,12 +676,26 @@ test "Server: scrollback fetch is per-client and independent" {
670	chunk = frame.payload; // ownership taken	676	chunk = frame.payload; // ownership taken
671	}	677	}
672	try std.testing.expect(chunk != null);	678	try std.testing.expect(chunk != null);
673	try std.testing.expect(std.mem.indexOf(u8, chunk.?[6..], "seq 1 100") != null);	679	// Decoded rather than searched: a chunk is CellRows now, and a byte
		680	// search would pass on a run header that happened to spell the text.
		681	const chunk_count = std.mem.readInt(u16, chunk.?[4..6], .little);
		682	const hist_rows = try grid_mod.decodeRows(alloc, chunk.?[6..], chunk_count, 100);
		683	defer grid_mod.freeRows(alloc, hist_rows);
		684	var seen_seq_line = false;
		685	for (hist_rows) \|*r\| {
		686	var line: std.ArrayList(u8) = .empty;
		687	defer line.deinit(alloc);
		688	for (r.cells) \|c\| {
		689	if (c.text_len == 0) try line.append(alloc, ' ') else try line.appendSlice(alloc, r.textOf(c));
		690	}
		691	if (std.mem.indexOf(u8, line.items, "seq 1 100") != null) seen_seq_line = true;
		692	}
		693	try std.testing.expect(seen_seq_line);
674		694
675	// ...while B keeps streaming live updates, undisturbed: B's fetch was	695	// ...while B keeps streaming live updates, undisturbed: B's fetch was
676	// never asked for, so B must never see a scrollback_chunk.	696	// never asked for, so B must never see a scrollback_chunk.
677	try proto.writeFrame(b.handle, .input, "echo b-still-live\n");	697	try proto.writeFrame(b.handle, .input, "echo b-still-live\n");
678	var replica_b = try Engine.init(alloc, .{ .cols = 100, .rows = 30 });	698	var replica_b = try Grid.init(alloc, 100, 30);
679	defer replica_b.deinit();	699	defer replica_b.deinit();
680	// The replica starts empty, so only a snapshot or the deltas after it	700	// The replica starts empty, so only a snapshot or the deltas after it
681	// can produce the echoed text — and neither of A's answers may appear.	701	// can produce the echoed text — and neither of A's answers may appear.
@@ -698,7 +718,7 @@ test "Server: replica rebuilt from snapshots matches the authoritative grid" {
698	defer stream.close();	718	defer stream.close();
699	const fd = stream.handle;	719	const fd = stream.handle;
700		720
701	var replica = try Engine.init(alloc, .{ .cols = 100, .rows = 30 });	721	var replica = try Grid.init(alloc, 100, 30);
702	defer replica.deinit();	722	defer replica.deinit();
703		723
704	try proto.writeFrame(fd, .input, "printf 'fidelity-%s\\n' ok\n");	724	try proto.writeFrame(fd, .input, "printf 'fidelity-%s\\n' ok\n");
@@ -710,20 +730,24 @@ test "Server: replica rebuilt from snapshots matches the authoritative grid" {
710	.needle = "fidelity-ok",	730	.needle = "fidelity-ok",
711	}));	731	}));
712		732
713	// Byte-compare replica vs authoritative daemon grid.	733	// Compare replica against the authoritative daemon grid. Plain, through
714	try proto.writeFrame(fd, .debug_dump, &.{1});	734	// `h.trimRowTails`: trailing spaces are the one formatting difference
715	var daemon_vt: ?[]u8 = null;	735	// between a daemon's dump and a client's, and the old VT wire trimmed
716	defer if (daemon_vt) \|d\| alloc.free(d);	736	// them as well.
		737	try proto.writeFrame(fd, .debug_dump, &.{0});
		738	var daemon_plain: ?[]u8 = null;
		739	defer if (daemon_plain) \|d\| alloc.free(d);
717	// Late updates may arrive before the reply; the sink applies them so	740	// Late updates may arrive before the reply; the sink applies them so
718	// the replica stays current with what the dump will show.	741	// the replica stays current with what the dump will show.
719	var feed: ApplyEach = .{ .alloc = alloc, .replica = replica };	742	var feed: ApplyEach = .{ .alloc = alloc, .replica = replica };
720	if (try h.awaitFrameOnSink(alloc, fd, .dump_reply, 10_000, feed.sink())) \|frame\| {	743	if (try h.awaitFrameOnSink(alloc, fd, .dump_reply, 10_000, feed.sink())) \|frame\| {
721	daemon_vt = frame.payload; // ownership taken	744	daemon_plain = try h.trimRowTails(alloc, frame.payload);
		745	frame.deinit(alloc);
722	}	746	}
723	try std.testing.expect(daemon_vt != null);	747	try std.testing.expect(daemon_plain != null);
724	const replica_vt = try replica.dumpVt(alloc);	748	const replica_plain = try replica.dumpPlain(alloc);
725	defer alloc.free(replica_vt);	749	defer alloc.free(replica_plain);
726	try std.testing.expectEqualStrings(daemon_vt.?, replica_vt);	750	try std.testing.expectEqualStrings(daemon_plain.?, replica_plain);
727	}	751	}
728		752
729	test "Server: typing produces deltas, not snapshots; stats track both" {	753	test "Server: typing produces deltas, not snapshots; stats track both" {

src/server/server_test_deliver.zig

Old		New
@@ -1,4 +1,5 @@
1	const std = @import("std");	1	const std = @import("std");
		2	const delta_mod = @import("term").delta;
2	const proto = @import("term").protocol;	3	const proto = @import("term").protocol;
3	const h = @import("server_test_harness.zig");	4	const h = @import("server_test_harness.zig");
4	const dial = h.dial;	5	const dial = h.dial;
@@ -87,9 +88,19 @@ test "Server: a client exceeding the pending cap is dropped" {
87	// Size the cap to a few snapshots' worth of backlog. Below one frame it	88	// Size the cap to a few snapshots' worth of backlog. Below one frame it
88	// would drop on the very first send, which would prove nothing about	89	// would drop on the very first send, which would prove nothing about
89	// bounding a backlog; the production default is 8 MiB.	90	// bounding a backlog; the production default is 8 MiB.
90	const state = try td.srv.sessions.table[0].?.eng.dumpState(alloc);	91	// Measured off the payload the daemon actually sends, not off any other
91	defer alloc.free(state);	92	// serialization of the grid: a cap sized from a bigger encoding is a cap
92	const frame_len = 5 + proto.snapshot_prefix_len + state.len;	93	// of far more than four frames, and the loop bound below would be what
		94	// ended the test instead of the cap.
		95	const snap = try delta_mod.buildSnapshot(alloc, td.srv.sessions.table[0].?.eng, .{
		96	.seq = 0,
		97	.history_rows = 0,
		98	.cols = 80,
		99	.rows = 24,
		100	.epoch = 0,
		101	});
		102	defer alloc.free(snap);
		103	const frame_len = 5 + snap.len;
93	td.srv.pending_cap = 4 * frame_len;	104	td.srv.pending_cap = 4 * frame_len;
94		105
95	var rounds: usize = 0;	106	var rounds: usize = 0;
@@ -419,10 +430,20 @@ test "Server: broadcast stats count every send but the counterfactual once" {
419	// The counterfactual is per event: what a snapshot-only daemon would	430	// The counterfactual is per event: what a snapshot-only daemon would
420	// have sent for this one update is one snapshot, however many clients	431	// have sent for this one update is one snapshot, however many clients
421	// received it.	432	// received it.
422	const state = try td.srv.sessions.table[0].?.eng.dumpState(alloc);	433	// Rebuilt here the way the daemon builds it, so the stat is checked
423	defer alloc.free(state);	434	// against the bytes a snapshot would really have cost rather than
		435	// against a second serialization nobody sends.
		436	const s = td.srv.sessions.table[0].?;
		437	const snap = try delta_mod.buildSnapshot(alloc, s.eng, .{
		438	.seq = s.tracker.seq,
		439	.history_rows = s.eng.historyRows(),
		440	.cols = 80,
		441	.rows = 24,
		442	.epoch = s.epoch,
		443	});
		444	defer alloc.free(snap);
424	try std.testing.expectEqual(	445	try std.testing.expectEqual(
425	@as(u64, @intCast(proto.snapshot_prefix_len + state.len)),	446	@as(u64, @intCast(snap.len)),
426	td.srv.stats.snapshot_equiv_bytes,	447	td.srv.stats.snapshot_equiv_bytes,
427	);	448	);
428	try std.testing.expectEqual(@as(u64, 0), td.srv.stats.snapshots);	449	try std.testing.expectEqual(@as(u64, 0), td.srv.stats.snapshots);

src/server/server_test_harness.zig

Old		New
@@ -1,5 +1,6 @@
1	const std = @import("std");	1	const std = @import("std");
2	const Engine = @import("term").engine.Engine;	2	const Engine = @import("term").engine.Engine;
		3	const Grid = @import("term").grid.Grid;
3	pub const Pty = @import("pty").Pty;	4	pub const Pty = @import("pty").Pty;
4	const proto = @import("term").protocol;	5	const proto = @import("term").protocol;
5	const replica_mod = @import("term").replica;	6	const replica_mod = @import("term").replica;
@@ -247,10 +248,10 @@ pub const TestDaemon = struct {
247	}	248	}
248	};	249	};
249		250
250	/// Bring a replica engine up to date with one daemon frame. The replay lives in	251	/// Bring a replica grid up to date with one daemon frame. The replay lives in
251	/// replica.zig — the production client's — so these tests replay through the	252	/// replica.zig — the production client's — so these tests replay through the
252	/// code the client ships and not a hand-rolled twin.	253	/// code the client ships and not a hand-rolled twin.
253	pub fn applyFrame(alloc: std.mem.Allocator, replica: *Engine, frame: proto.Frame) !void {	254	pub fn applyFrame(alloc: std.mem.Allocator, replica: *Grid, frame: proto.Frame) !void {
254	if (frame.type != .snapshot and frame.type != .delta) return;	255	if (frame.type != .snapshot and frame.type != .delta) return;
255	var r = replica_mod.Replica.init(alloc, replica);	256	var r = replica_mod.Replica.init(alloc, replica);
256	// The old helper propagated a garbled delta as BadPayload; .resync is	257	// The old helper propagated a garbled delta as BadPayload; .resync is
@@ -258,6 +259,29 @@ pub fn applyFrame(alloc: std.mem.Allocator, replica: *Engine, frame: proto.Frame
258	if (try r.apply(frame.type, frame.payload) == .resync) return error.BadPayload;	259	if (try r.apply(frame.type, frame.payload) == .resync) return error.BadPayload;
259	}	260	}
260		261
		262	/// A copy of `text` with each row's trailing spaces removed.
		263	///
		264	/// The one formatting difference between a daemon's `Engine.dumpPlain` and a
		265	/// client's `Grid.dumpPlain`: ghostty dumps with trimming off, so a space a
		266	/// program wrote at the end of a row survives it, while a client never holds
		267	/// one — the encoder stops a row at its last non-blank cell, and the VT
		268	/// formatter that fed the old wire trimmed trailing whitespace in the same
		269	/// place. `test/e2e_lib.sh assert_ws_converged` strips it on both sides for
		270	/// the same reason. A convergence pin compares the daemon's dump through
		271	/// this, and nothing else about the two dumps may differ.
		272	pub fn trimRowTails(alloc: std.mem.Allocator, text: []const u8) ![]u8 {
		273	var out: std.ArrayList(u8) = .empty;
		274	errdefer out.deinit(alloc);
		275	var it = std.mem.splitScalar(u8, text, '\n');
		276	var first = true;
		277	while (it.next()) \|line\| {
		278	if (!first) try out.append(alloc, '\n');
		279	first = false;
		280	try out.appendSlice(alloc, std.mem.trimRight(u8, line, " "));
		281	}
		282	return out.toOwnedSlice(alloc);
		283	}
		284
261	/// A connected pair of unix stream sockets, standing in for an attached client.	285	/// A connected pair of unix stream sockets, standing in for an attached client.
262	/// `daemon` goes in a client slot; `peer` is the client's end.	286	/// `daemon` goes in a client slot; `peer` is the client's end.
263	///	287	///
@@ -480,7 +504,7 @@ pub const never_from_daemon: proto.MsgType = .attach;
480	/// What `awaitReplicaText` is waiting for on one connection.	504	/// What `awaitReplicaText` is waiting for on one connection.
481	pub const ReplicaWait = struct {	505	pub const ReplicaWait = struct {
482	/// Replayed through `applyFrame`, so through the production replica.	506	/// Replayed through `applyFrame`, so through the production replica.
483	replica: *Engine,	507	replica: *Grid,
484	/// The text the grid must show. Found in `dumpPlain`, not in the frame	508	/// The text the grid must show. Found in `dumpPlain`, not in the frame
485	/// bytes: a row can arrive spread over several deltas, and the grid is	509	/// bytes: a row can arrive spread over several deltas, and the grid is
486	/// what a user would see.	510	/// what a user would see.

src/server/server_test_quic.zig

Old		New
@@ -1,5 +1,5 @@
1	const std = @import("std");	1	const std = @import("std");
2	const Engine = @import("term").engine.Engine;	2	const Grid = @import("term").grid.Grid;
3	const proto = @import("term").protocol;	3	const proto = @import("term").protocol;
4	const quic = @import("quic");	4	const quic = @import("quic");
5	const quic_server = @import("quic_server.zig");	5	const quic_server = @import("quic_server.zig");
@@ -305,7 +305,7 @@ test "Server: one QUIC client leaving does not disturb the other" {
305	try proto.appendFrame(&bbuf, alloc, .input, "echo quic-two-ok\n");	305	try proto.appendFrame(&bbuf, alloc, .input, "echo quic-two-ok\n");
306	b.out = bbuf.items;	306	b.out = bbuf.items;
307	b.drain();	307	b.drain();
308	var replica = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	308	var replica = try Grid.init(alloc, 80, 24);
309	defer replica.deinit();	309	defer replica.deinit();
310	// The predicate has to be the ASSERTION, not a weaker relative of it:	310	// The predicate has to be the ASSERTION, not a weaker relative of it:
311	// `echoed()` counts every byte B has ever taken, and B took a snapshot	311	// `echoed()` counts every byte B has ever taken, and B took a snapshot

src/server/server_test_session.zig

Old		New
@@ -1,6 +1,6 @@
1	const std = @import("std");	1	const std = @import("std");
2	const builtin = @import("builtin");	2	const builtin = @import("builtin");
3	const Engine = @import("term").engine.Engine;	3	const Grid = @import("term").grid.Grid;
4	const proto = @import("term").protocol;	4	const proto = @import("term").protocol;
5	const quic = @import("quic");	5	const quic = @import("quic");
6	const quic_server = @import("quic_server.zig");	6	const quic_server = @import("quic_server.zig");
@@ -529,7 +529,7 @@ fn pumpUntilReplicaSees(
529	alloc: std.mem.Allocator,	529	alloc: std.mem.Allocator,
530	srv: *Server,	530	srv: *Server,
531	fd: std.posix.fd_t,	531	fd: std.posix.fd_t,
532	rep: *Engine,	532	rep: *Grid,
533	needle: []const u8,	533	needle: []const u8,
534	iters: usize,	534	iters: usize,
535	) !bool {	535	) !bool {
@@ -561,9 +561,9 @@ test "Server: two named sessions hold two shells with independent content" {
561	try proto.writeFrame(ca.handle, .input, "MARKER-ALPHA\n");	561	try proto.writeFrame(ca.handle, .input, "MARKER-ALPHA\n");
562	try proto.writeFrame(cb.handle, .input, "MARKER-BETA\n");	562	try proto.writeFrame(cb.handle, .input, "MARKER-BETA\n");
563		563
564	var rep_a = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	564	var rep_a = try Grid.init(alloc, 80, 24);
565	defer rep_a.deinit();	565	defer rep_a.deinit();
566	var rep_b = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	566	var rep_b = try Grid.init(alloc, 80, 24);
567	defer rep_b.deinit();	567	defer rep_b.deinit();
568		568
569	// Liveness first: each client converges on its own marker.	569	// Liveness first: each client converges on its own marker.
@@ -648,9 +648,9 @@ test "Server: every session shell is told the socket it lives on and its own nam
648	try proto.writeFrame(c0.handle, .input, probe);	648	try proto.writeFrame(c0.handle, .input, probe);
649	try proto.writeFrame(ca.handle, .input, probe);	649	try proto.writeFrame(ca.handle, .input, probe);
650		650
651	var rep_0 = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	651	var rep_0 = try Grid.init(alloc, 80, 24);
652	defer rep_0.deinit();	652	defer rep_0.deinit();
653	var rep_a = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	653	var rep_a = try Grid.init(alloc, 80, 24);
654	defer rep_a.deinit();	654	defer rep_a.deinit();
655		655
656	// The default session's name on the wire is empty; what its shell is	656	// The default session's name on the wire is empty; what its shell is
@@ -865,7 +865,7 @@ test "Server: one session's shell exiting drops only its clients; the daemon car
865		865
866	// b's liveness established BEFORE a dies, so its survival below is a	866	// b's liveness established BEFORE a dies, so its survival below is a
867	// comparison and not a hope.	867	// comparison and not a hope.
868	var rep_b = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	868	var rep_b = try Grid.init(alloc, 80, 24);
869	defer rep_b.deinit();	869	defer rep_b.deinit();
870	try proto.writeFrame(cb.handle, .input, "pre\n");	870	try proto.writeFrame(cb.handle, .input, "pre\n");
871	try std.testing.expect(	871	try std.testing.expect(
@@ -956,7 +956,7 @@ fn probeEmptiedDaemon(alloc: std.mem.Allocator, sock_path: []const u8) !void {
956	// bash and the script's startup is longer; the same race is open on	956	// bash and the script's startup is longer; the same race is open on
957	// Linux and merely lost less often. One round trip through the script	957	// Linux and merely lost less often. One round trip through the script
958	// closes it.	958	// closes it.
959	var rep = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	959	var rep = try Grid.init(alloc, 80, 24);
960	defer rep.deinit();	960	defer rep.deinit();
961	try proto.writeFrame(c1.handle, .input, "alive\n");	961	try proto.writeFrame(c1.handle, .input, "alive\n");
962	if (!try h.awaitReplicaText(alloc, c1.handle, 8000, .{	962	if (!try h.awaitReplicaText(alloc, c1.handle, 8000, .{
@@ -1096,9 +1096,9 @@ test "Server: dump names a session; an unknown name answers in words" {
1096	try proto.writeFrame(ca.handle, .input, "MARKER-ALPHA\n");	1096	try proto.writeFrame(ca.handle, .input, "MARKER-ALPHA\n");
1097	try proto.writeFrame(cb.handle, .input, "MARKER-BETA\n");	1097	try proto.writeFrame(cb.handle, .input, "MARKER-BETA\n");
1098		1098
1099	var rep_a = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	1099	var rep_a = try Grid.init(alloc, 80, 24);
1100	defer rep_a.deinit();	1100	defer rep_a.deinit();
1101	var rep_b = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	1101	var rep_b = try Grid.init(alloc, 80, 24);
1102	defer rep_b.deinit();	1102	defer rep_b.deinit();
1103	try std.testing.expect(	1103	try std.testing.expect(
1104	try pumpUntilReplicaSees(alloc, &td.srv, ca.handle, rep_a, "MARKER-ALPHA", 400),	1104	try pumpUntilReplicaSees(alloc, &td.srv, ca.handle, rep_a, "MARKER-ALPHA", 400),