a73x

633b38dd

test: wsclient — scripted WebSocket replica client for hub scenarios

a73x   2026-08-13 10:44

Commit message
test: wsclient — scripted WebSocket replica client for hub scenarios

M-web Task 9, ptyclient's sibling: speaks the hub's wire (RFC 6455
client side, ~60 lines of masked sends and unmasked reassembly) and
maintains a REAL replica — the same replica.zig the CLI and wasm core
run — so dumpexit writes the daemon's own dump format and the e2e diff
compares grids, not file conventions (a test pins the two engines
byte-identical through a snapshot round-trip). Verbs: attach/
attachfresh, send, resize, expectgrid, expectstate, settle, dumpexit;
--origin forges the wrong-Origin scenario's header; exit codes 2/3/4
per ptyclient. The upgrade validates the accept key against OUR nonce,
so a canned 101 cannot pass.

Hand smoke: attach 1x1 through a live daemon+hub, type via input
frames, dumpexit CONVERGED against muxd dump.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
build.zig
Old New
@@ -478,6 +478,20 @@ pub fn build(b: *std.Build) void {
478 // The page's three assets arrive as anonymous imports so @embedFile 478 // The page's three assets arrive as anonymous imports so @embedFile
479 // can name them; the wasm one is the artifact itself, which also 479 // can name them; the wasm one is the artifact itself, which also
480 // sequences the wasm build before the hub's. 480 // sequences the wasm build before the hub's.
481 const wsclient_mod = b.createModule(.{
482 .root_source_file = b.path("test/wsclient.zig"),
483 .target = target,
484 .optimize = optimize,
485 .link_libc = true,
486 });
487 wsclient_mod.addImport("engine", engine_mod);
488 wsclient_mod.addImport("replica", replica_mod);
489 wsclient_mod.addImport("protocol", protocol_mod);
490 const wsclient_exe = b.addExecutable(.{ .name = "wsclient", .root_module = wsclient_mod });
491 wsclient_exe.use_llvm = true;
492 wsclient_exe.use_lld = true;
493 b.installArtifact(wsclient_exe);
494
481 const webhub_main_mod = b.createModule(.{ 495 const webhub_main_mod = b.createModule(.{
482 .root_source_file = b.path("src/webhub_main.zig"), 496 .root_source_file = b.path("src/webhub_main.zig"),
483 .target = target, 497 .target = target,
@@ -509,7 +523,7 @@ pub fn build(b: *std.Build) void {
509 // absence here was a live hazard recorded in decisions.md — muxd's 523 // absence here was a live hazard recorded in decisions.md — muxd's
510 // entrypoint could grow tests that silently never ran, exactly as 524 // entrypoint could grow tests that silently never ran, exactly as
511 // mux_main.zig's five did before it was added. 525 // mux_main.zig's five did before it was added.
512 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, delta_mod, replica_mod, keymap_mod, webhub_mod, sockpath_mod, server_mod, client_mod, proxy_mod, mux_mod, quic_mod, quic_server_mod, exe_mod, testtmp_mod, quic_client_mod, predict_mod, rawmode_mod, delaypipe_mod, xdg_mod, spawn_mod, handoff_mod, paint_mod, render_mod, ptyclient_mod, webhub_main_mod}) |mod| { 526 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, delta_mod, replica_mod, keymap_mod, webhub_mod, sockpath_mod, server_mod, client_mod, proxy_mod, mux_mod, quic_mod, quic_server_mod, exe_mod, testtmp_mod, quic_client_mod, predict_mod, rawmode_mod, delaypipe_mod, xdg_mod, spawn_mod, handoff_mod, paint_mod, render_mod, ptyclient_mod, webhub_main_mod, wsclient_mod}) |mod| {
513 const t = b.addTest(.{ .root_module = mod }); 527 const t = b.addTest(.{ .root_module = mod });
514 t.use_llvm = true; 528 t.use_llvm = true;
515 t.use_lld = true; 529 t.use_lld = true;
test/wsclient.zig
Old New
@@ -0,0 +1,529 @@
1 //! e2e fixture: a scripted WebSocket client standing in for the browser
2 //! (M-web Task 9). Speaks the hub's wire (RFC 6455 client side, masked;
3 //! one envelope byte per message) and maintains a REAL replica from the
4 //! frames it receives — the same replica.zig the CLI and the wasm core
5 //! run — so its `dumpexit` grid diffs honestly against `muxd dump`.
6 //! Driven by a line-oriented script on stdin, ptyclient's conventions:
7 //! decodeEscapes, expect deadlines in ms, exit codes 2/3/4.
8 //!
9 //! Verbs:
10 //! attach C R send an attach quoting the replica's resume args
11 //! attachfresh C R same, quoting (0,0)
12 //! send BYTES input frame (escapes decoded)
13 //! resize C R resize frame
14 //! expectgrid NEEDLE MS poll the replica's plain dump for NEEDLE
15 //! expectstate STATE MS wait for a control message naming STATE
16 //! settle QUIET MS drain frames until QUIET ms of silence
17 //! dumpexit write the replica grid (muxd dump format) to
18 //! --out and exit 0
19 //!
20 //! Usage: wsclient --port N --tile IDX --out FILE --err FILE
21 //! [--origin STR] < script
22 const std = @import("std");
23 const Engine = @import("engine").Engine;
24 const Replica = @import("replica").Replica;
25 const proto = @import("protocol");
26
27 const EXIT_USAGE: u8 = 2;
28 const EXIT_TIMEOUT: u8 = 3;
29 const EXIT_DIED: u8 = 4;
30
31 var err_file: ?std.fs.File = null;
32
33 fn fatal(code: u8, comptime fmt: []const u8, args: anytype) noreturn {
34 var buf: [512]u8 = undefined;
35 const msg = std.fmt.bufPrint(&buf, "wsclient: " ++ fmt ++ "\n", args) catch "wsclient: error\n";
36 if (err_file) |f| f.writeAll(msg) catch {};
37 std.debug.print("{s}", .{msg});
38 std.process.exit(code);
39 }
40
41 /// C-style escapes, ptyclient's exact table (\xNN digit-by-digit so a
42 /// signed parse cannot sneak through).
43 fn decodeEscapes(alloc: std.mem.Allocator, s: []const u8) ![]u8 {
44 var out: std.ArrayList(u8) = .empty;
45 errdefer out.deinit(alloc);
46 var i: usize = 0;
47 while (i < s.len) : (i += 1) {
48 if (s[i] != '\\') {
49 try out.append(alloc, s[i]);
50 continue;
51 }
52 i += 1;
53 if (i >= s.len) return error.BadEscape;
54 switch (s[i]) {
55 'n' => try out.append(alloc, '\n'),
56 'r' => try out.append(alloc, '\r'),
57 't' => try out.append(alloc, '\t'),
58 '\\' => try out.append(alloc, '\\'),
59 'x' => {
60 if (i + 2 >= s.len) return error.BadEscape;
61 const hi = std.fmt.charToDigit(s[i + 1], 16) catch return error.BadEscape;
62 const lo = std.fmt.charToDigit(s[i + 2], 16) catch return error.BadEscape;
63 try out.append(alloc, hi * 16 + lo);
64 i += 2;
65 },
66 else => return error.BadEscape,
67 }
68 }
69 return out.toOwnedSlice(alloc);
70 }
71
72 // ---------------------------------------------------------------------------
73 // RFC 6455, client side. The server side is std's; this is the ~60-line
74 // mirror image: masked sends, unmasked receives.
75
76 /// One client→server message: FIN + binary opcode, 4-byte mask, payload
77 /// XOR'd. Layout returned ready to write.
78 fn maskedMessage(alloc: std.mem.Allocator, payload: []const u8, mask: [4]u8) ![]u8 {
79 var out: std.ArrayList(u8) = .empty;
80 errdefer out.deinit(alloc);
81 try out.append(alloc, 0x82); // FIN | binary
82 if (payload.len <= 125) {
83 try out.append(alloc, 0x80 | @as(u8, @intCast(payload.len)));
84 } else if (payload.len <= 0xffff) {
85 try out.append(alloc, 0x80 | 126);
86 var b: [2]u8 = undefined;
87 std.mem.writeInt(u16, &b, @intCast(payload.len), .big);
88 try out.appendSlice(alloc, &b);
89 } else {
90 try out.append(alloc, 0x80 | 127);
91 var b: [8]u8 = undefined;
92 std.mem.writeInt(u64, &b, payload.len, .big);
93 try out.appendSlice(alloc, &b);
94 }
95 try out.appendSlice(alloc, &mask);
96 const start = out.items.len;
97 try out.appendSlice(alloc, payload);
98 for (out.items[start..], 0..) |*c, i| c.* ^= mask[i % 4];
99 return out.toOwnedSlice(alloc);
100 }
101
102 /// Server→client frames accumulate here off the socket; whole messages
103 /// pop out. Control frames (ping/close) are handled by the caller via
104 /// the opcode.
105 const WsReader = struct {
106 buf: std.ArrayList(u8) = .empty,
107
108 const Msg = struct { opcode: u4, payload: []const u8, consumed: usize };
109
110 /// Parse one whole frame from the front of the buffer, or null.
111 fn peek(self: *const WsReader) ?Msg {
112 const b = self.buf.items;
113 if (b.len < 2) return null;
114 const opcode: u4 = @truncate(b[0] & 0x0f);
115 const masked = b[1] & 0x80 != 0;
116 if (masked) return null; // server frames are never masked; treated as garbage by caller
117 var len: u64 = b[1] & 0x7f;
118 var off: usize = 2;
119 if (len == 126) {
120 if (b.len < 4) return null;
121 len = std.mem.readInt(u16, b[2..4], .big);
122 off = 4;
123 } else if (len == 127) {
124 if (b.len < 10) return null;
125 len = std.mem.readInt(u64, b[2..10], .big);
126 off = 10;
127 }
128 if (len > 64 * 1024 * 1024) return null; // absurd: caller dies on stall
129 if (b.len < off + len) return null;
130 return .{ .opcode = opcode, .payload = b[off .. off + @as(usize, @intCast(len))], .consumed = off + @as(usize, @intCast(len)) };
131 }
132
133 fn consume(self: *WsReader, alloc: std.mem.Allocator, n: usize) void {
134 const rest = self.buf.items[n..];
135 std.mem.copyForwards(u8, self.buf.items[0..rest.len], rest);
136 self.buf.shrinkRetainingCapacity(rest.len);
137 _ = alloc;
138 }
139 };
140
141 // ---------------------------------------------------------------------------
142
143 const Client = struct {
144 alloc: std.mem.Allocator,
145 sock: std.posix.fd_t,
146 reader: WsReader = .{},
147 eng: *Engine,
148 rep: Replica,
149 /// Last control-message state seen (the tile chrome's vocabulary).
150 last_state: [16]u8 = @splat(0),
151 last_state_len: usize = 0,
152
153 fn sendMessage(self: *Client, payload: []const u8) void {
154 var mask: [4]u8 = undefined;
155 std.crypto.random.bytes(&mask);
156 const msg = maskedMessage(self.alloc, payload, mask) catch fatal(EXIT_USAGE, "oom", .{});
157 defer self.alloc.free(msg);
158 writeAll(self.sock, msg) catch fatal(EXIT_DIED, "hub hung up mid-send", .{});
159 }
160
161 fn sendFrame(self: *Client, t: u8, payload: []const u8) void {
162 var out: std.ArrayList(u8) = .empty;
163 defer out.deinit(self.alloc);
164 out.append(self.alloc, 0x00) catch fatal(EXIT_USAGE, "oom", .{});
165 out.append(self.alloc, t) catch fatal(EXIT_USAGE, "oom", .{});
166 var lenb: [4]u8 = undefined;
167 std.mem.writeInt(u32, &lenb, @intCast(payload.len), .little);
168 out.appendSlice(self.alloc, &lenb) catch fatal(EXIT_USAGE, "oom", .{});
169 out.appendSlice(self.alloc, payload) catch fatal(EXIT_USAGE, "oom", .{});
170 self.sendMessage(out.items);
171 }
172
173 /// Pump whatever is on the socket into the reader and apply every
174 /// whole message. Returns false when the hub hung up. `wait_ms` is
175 /// one poll's patience, not a deadline.
176 fn pump(self: *Client, wait_ms: i32) bool {
177 var fds = [_]std.posix.pollfd{
178 .{ .fd = self.sock, .events = std.posix.POLL.IN, .revents = 0 },
179 };
180 const n = std.posix.poll(&fds, wait_ms) catch return false;
181 if (n > 0 and fds[0].revents != 0) {
182 var buf: [64 * 1024]u8 = undefined;
183 const got = std.posix.read(self.sock, &buf) catch return false;
184 if (got == 0) return false;
185 self.reader.buf.appendSlice(self.alloc, buf[0..got]) catch return false;
186 }
187 while (self.reader.peek()) |msg| {
188 self.handle(msg);
189 self.reader.consume(self.alloc, msg.consumed);
190 }
191 return true;
192 }
193
194 fn handle(self: *Client, msg: WsReader.Msg) void {
195 switch (msg.opcode) {
196 0x8 => fatal(EXIT_DIED, "hub sent close", .{}),
197 0x9 => return, // ping: the hub never sends one; ignore rather than die
198 0x1, 0x2 => {},
199 else => return,
200 }
201 const data = msg.payload;
202 if (data.len < 1) return;
203 if (data[0] == 0x01) {
204 // {"state":"..."} — extracted textually; the vocabulary is
205 // closed and the producer is ours.
206 const json = data[1..];
207 const k = "\"state\":\"";
208 if (std.mem.indexOf(u8, json, k)) |i| {
209 const rest = json[i + k.len ..];
210 const end = std.mem.indexOfScalar(u8, rest, '"') orelse return;
211 const state = rest[0..end];
212 const n = @min(state.len, self.last_state.len);
213 @memcpy(self.last_state[0..n], state[0..n]);
214 self.last_state_len = n;
215 }
216 return;
217 }
218 if (data[0] != 0x00 or data.len < 1 + proto.frame_header_len) return;
219 const t = data[1];
220 const plen = std.mem.readInt(u32, data[2..6][0..4], .little);
221 if (data.len - 6 != plen) return;
222 const payload = data[6..];
223 const snapshot: u8 = @intFromEnum(proto.MsgType.snapshot);
224 const delta: u8 = @intFromEnum(proto.MsgType.delta);
225 if (t == snapshot or t == delta) {
226 const applied = self.rep.apply(@enumFromInt(t), payload) catch return;
227 if (applied == .resync) {
228 // Mirror the browser: a garbled delta re-attaches fresh.
229 var att = proto.encodeAttach(self.rep.grid.cols, self.rep.grid.rows, 0, 0);
230 self.sendFrame(@intFromEnum(proto.MsgType.attach), &att);
231 }
232 }
233 // Everything else (exit_status, pty_mode, scrollback) is visible
234 // in --err via the state machinery when a scenario needs it.
235 }
236
237 fn stateIs(self: *const Client, want: []const u8) bool {
238 return std.mem.eql(u8, self.last_state[0..self.last_state_len], want);
239 }
240 };
241
242 fn writeAll(fd: std.posix.fd_t, data: []const u8) !void {
243 var i: usize = 0;
244 while (i < data.len) i += try std.posix.write(fd, data[i..]);
245 }
246
247 fn nowMs() i64 {
248 return std.time.milliTimestamp();
249 }
250
251 // ---------------------------------------------------------------------------
252
253 pub fn main() !void {
254 var gpa: std.heap.DebugAllocator(.{}) = .init;
255 defer _ = gpa.deinit();
256 const alloc = gpa.allocator();
257
258 const args = try std.process.argsAlloc(alloc);
259 defer std.process.argsFree(alloc, args);
260
261 var port: ?u16 = null;
262 var tile: usize = 0;
263 var out_path: ?[]const u8 = null;
264 var err_path: ?[]const u8 = null;
265 var origin: ?[]const u8 = null;
266
267 var i: usize = 1;
268 while (i < args.len) : (i += 1) {
269 const a = args[i];
270 if (std.mem.eql(u8, a, "--port") and i + 1 < args.len) {
271 i += 1;
272 port = std.fmt.parseInt(u16, args[i], 10) catch fatal(EXIT_USAGE, "bad --port", .{});
273 } else if (std.mem.eql(u8, a, "--tile") and i + 1 < args.len) {
274 i += 1;
275 tile = std.fmt.parseInt(usize, args[i], 10) catch fatal(EXIT_USAGE, "bad --tile", .{});
276 } else if (std.mem.eql(u8, a, "--out") and i + 1 < args.len) {
277 i += 1;
278 out_path = args[i];
279 } else if (std.mem.eql(u8, a, "--err") and i + 1 < args.len) {
280 i += 1;
281 err_path = args[i];
282 } else if (std.mem.eql(u8, a, "--origin") and i + 1 < args.len) {
283 i += 1;
284 origin = args[i];
285 } else {
286 fatal(EXIT_USAGE, "unknown arg {s}", .{a});
287 }
288 }
289 const p = port orelse fatal(EXIT_USAGE, "--port required", .{});
290 const op = out_path orelse fatal(EXIT_USAGE, "--out required", .{});
291 const ep = err_path orelse fatal(EXIT_USAGE, "--err required", .{});
292 err_file = std.fs.cwd().createFile(ep, .{}) catch fatal(EXIT_USAGE, "cannot open --err", .{});
293
294 // --- TCP + upgrade ---
295 const addr = std.net.Address.parseIp("127.0.0.1", p) catch unreachable;
296 const stream = std.net.tcpConnectToAddress(addr) catch
297 fatal(EXIT_DIED, "cannot connect to 127.0.0.1:{d}", .{p});
298 const sock = stream.handle;
299
300 var key_raw: [16]u8 = undefined;
301 std.crypto.random.bytes(&key_raw);
302 var key_b64: [24]u8 = undefined;
303 _ = std.base64.standard.Encoder.encode(&key_b64, &key_raw);
304
305 const default_origin = try std.fmt.allocPrint(alloc, "http://127.0.0.1:{d}", .{p});
306 defer alloc.free(default_origin);
307 const req = try std.fmt.allocPrint(alloc,
308 "GET /ws/{d} HTTP/1.1\r\n" ++
309 "host: 127.0.0.1:{d}\r\n" ++
310 "connection: upgrade\r\n" ++
311 "upgrade: websocket\r\n" ++
312 "sec-websocket-version: 13\r\n" ++
313 "sec-websocket-key: {s}\r\n" ++
314 "origin: {s}\r\n\r\n", .{ tile, p, key_b64, origin orelse default_origin });
315 defer alloc.free(req);
316 writeAll(sock, req) catch fatal(EXIT_DIED, "hub hung up during upgrade", .{});
317
318 // Read the response head. On anything but 101 print the status line
319 // and exit 4 — the wrong-Origin scenario asserts exactly this.
320 var head: std.ArrayList(u8) = .empty;
321 defer head.deinit(alloc);
322 while (std.mem.indexOf(u8, head.items, "\r\n\r\n") == null) {
323 if (head.items.len > 16 * 1024) fatal(EXIT_DIED, "oversize upgrade response", .{});
324 var b: [1024]u8 = undefined;
325 const n = std.posix.read(sock, &b) catch fatal(EXIT_DIED, "read failed during upgrade", .{});
326 if (n == 0) fatal(EXIT_DIED, "hub closed during upgrade", .{});
327 try head.appendSlice(alloc, b[0..n]);
328 }
329 const head_end = std.mem.indexOf(u8, head.items, "\r\n\r\n").? + 4;
330 const status_line = head.items[0 .. std.mem.indexOf(u8, head.items, "\r\n").?];
331 if (std.mem.indexOf(u8, status_line, "101") == null)
332 fatal(EXIT_DIED, "upgrade refused: {s}", .{status_line});
333 // The accept key must be OUR key's digest — a hub echoing a canned
334 // value would pass anything else this fixture checks.
335 var sha = std.crypto.hash.Sha1.init(.{});
336 sha.update(&key_b64);
337 sha.update("258EAFA5-E914-47DA-95CA-C5AB0DC85B11");
338 var digest: [20]u8 = undefined;
339 sha.final(&digest);
340 var accept_b64: [28]u8 = undefined;
341 _ = std.base64.standard.Encoder.encode(&accept_b64, &digest);
342 if (std.mem.indexOf(u8, head.items[0..head_end], &accept_b64) == null)
343 fatal(EXIT_DIED, "sec-websocket-accept mismatch", .{});
344
345 // --- replica + client ---
346 var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
347 defer eng.deinit();
348 var cl = Client{ .alloc = alloc, .sock = sock, .eng = eng, .rep = Replica.init(alloc, eng) };
349 defer cl.reader.buf.deinit(alloc);
350 // Bytes past the head are the first WS frames.
351 try cl.reader.buf.appendSlice(alloc, head.items[head_end..]);
352
353 // --- script loop ---
354 var stdin_buf: std.ArrayList(u8) = .empty;
355 defer stdin_buf.deinit(alloc);
356 var rbuf: [4096]u8 = undefined;
357 while (true) {
358 const n = std.posix.read(std.posix.STDIN_FILENO, &rbuf) catch break;
359 if (n == 0) break;
360 try stdin_buf.appendSlice(alloc, rbuf[0..n]);
361 }
362
363 var lines = std.mem.splitScalar(u8, stdin_buf.items, '\n');
364 while (lines.next()) |raw| {
365 const line = std.mem.trim(u8, raw, " \t\r");
366 if (line.len == 0 or line[0] == '#') continue;
367 const sp = std.mem.indexOfScalar(u8, line, ' ');
368 const verb = if (sp) |s| line[0..s] else line;
369 const rest = if (sp) |s| line[s + 1 ..] else "";
370
371 if (std.mem.eql(u8, verb, "attach") or std.mem.eql(u8, verb, "attachfresh")) {
372 var it = std.mem.tokenizeScalar(u8, rest, ' ');
373 const cols = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "attach C R", .{}), 10) catch fatal(EXIT_USAGE, "attach C R", .{});
374 const rows = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "attach C R", .{}), 10) catch fatal(EXIT_USAGE, "attach C R", .{});
375 const fresh = std.mem.eql(u8, verb, "attachfresh");
376 const q = if (fresh) Replica.AttachArgs{ .have_seq = 0, .have_epoch = 0 } else cl.rep.attachArgs();
377 const att = proto.encodeAttach(cols, rows, q.have_seq, q.have_epoch);
378 cl.sendFrame(@intFromEnum(proto.MsgType.attach), &att);
379 } else if (std.mem.eql(u8, verb, "send")) {
380 const bytes = decodeEscapes(alloc, rest) catch fatal(EXIT_USAGE, "bad escape in send", .{});
381 defer alloc.free(bytes);
382 cl.sendFrame(@intFromEnum(proto.MsgType.input), bytes);
383 } else if (std.mem.eql(u8, verb, "resize")) {
384 var it = std.mem.tokenizeScalar(u8, rest, ' ');
385 const cols = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "resize C R", .{}), 10) catch fatal(EXIT_USAGE, "resize C R", .{});
386 const rows = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "resize C R", .{}), 10) catch fatal(EXIT_USAGE, "resize C R", .{});
387 const sz = proto.encodeSize(cols, rows);
388 cl.sendFrame(@intFromEnum(proto.MsgType.resize), &sz);
389 } else if (std.mem.eql(u8, verb, "expectgrid")) {
390 const last = std.mem.lastIndexOfScalar(u8, rest, ' ') orelse fatal(EXIT_USAGE, "expectgrid NEEDLE MS", .{});
391 if (last == 0) fatal(EXIT_USAGE, "empty needle", .{});
392 const ms = std.fmt.parseInt(i64, rest[last + 1 ..], 10) catch fatal(EXIT_USAGE, "bad deadline", .{});
393 const needle = decodeEscapes(alloc, rest[0..last]) catch fatal(EXIT_USAGE, "bad escape", .{});
394 defer alloc.free(needle);
395 const deadline = nowMs() + ms;
396 while (true) {
397 const dump = cl.rep.eng.dumpPlain(alloc) catch fatal(EXIT_USAGE, "oom", .{});
398 const hit = std.mem.indexOf(u8, dump, needle) != null;
399 alloc.free(@constCast(dump));
400 if (hit) break;
401 if (nowMs() >= deadline) fatal(EXIT_TIMEOUT, "expectgrid '{s}' timed out", .{needle});
402 if (!cl.pump(50)) fatal(EXIT_DIED, "hub hung up during expectgrid", .{});
403 }
404 } else if (std.mem.eql(u8, verb, "expectstate")) {
405 const last = std.mem.lastIndexOfScalar(u8, rest, ' ') orelse fatal(EXIT_USAGE, "expectstate STATE MS", .{});
406 const ms = std.fmt.parseInt(i64, rest[last + 1 ..], 10) catch fatal(EXIT_USAGE, "bad deadline", .{});
407 const want = rest[0..last];
408 const deadline = nowMs() + ms;
409 while (!cl.stateIs(want)) {
410 if (nowMs() >= deadline) fatal(EXIT_TIMEOUT, "expectstate '{s}' timed out (at '{s}')", .{ want, cl.last_state[0..cl.last_state_len] });
411 if (!cl.pump(50)) fatal(EXIT_DIED, "hub hung up during expectstate", .{});
412 }
413 } else if (std.mem.eql(u8, verb, "settle")) {
414 var it = std.mem.tokenizeScalar(u8, rest, ' ');
415 const quiet = std.fmt.parseInt(i64, it.next() orelse fatal(EXIT_USAGE, "settle QUIET MS", .{}), 10) catch fatal(EXIT_USAGE, "settle QUIET MS", .{});
416 const ms = std.fmt.parseInt(i64, it.next() orelse fatal(EXIT_USAGE, "settle QUIET MS", .{}), 10) catch fatal(EXIT_USAGE, "settle QUIET MS", .{});
417 const deadline = nowMs() + ms;
418 var last_traffic = nowMs();
419 while (nowMs() - last_traffic < quiet) {
420 if (nowMs() >= deadline) break;
421 const before = cl.rep.last_seq;
422 if (!cl.pump(50)) fatal(EXIT_DIED, "hub hung up during settle", .{});
423 if (cl.rep.last_seq != before) last_traffic = nowMs();
424 }
425 } else if (std.mem.eql(u8, verb, "dumpexit")) {
426 const dump = cl.rep.eng.dumpPlain(alloc) catch fatal(EXIT_USAGE, "oom", .{});
427 defer alloc.free(@constCast(dump));
428 const out = std.fs.cwd().createFile(op, .{}) catch fatal(EXIT_USAGE, "cannot open --out", .{});
429 defer out.close();
430 out.writeAll(dump) catch fatal(EXIT_USAGE, "write --out failed", .{});
431 // `muxd dump` ends its output with a newline; match it so the
432 // e2e diff compares grids, not file conventions.
433 if (dump.len == 0 or dump[dump.len - 1] != '\n')
434 out.writeAll("\n") catch fatal(EXIT_USAGE, "write --out failed", .{});
435 stream.close();
436 return;
437 } else {
438 fatal(EXIT_USAGE, "unknown verb {s}", .{verb});
439 }
440 }
441 fatal(EXIT_USAGE, "script ended without dumpexit", .{});
442 }
443
444 // ---------------------------------------------------------------------------
445
446 test "masked message: header layout and mask application, all three length forms" {
447 const alloc = std.testing.allocator;
448 const mask = [4]u8{ 0xaa, 0xbb, 0xcc, 0xdd };
449
450 const small = try maskedMessage(alloc, "hi", mask);
451 defer alloc.free(small);
452 try std.testing.expectEqual(@as(u8, 0x82), small[0]); // FIN | binary
453 try std.testing.expectEqual(@as(u8, 0x80 | 2), small[1]); // masked, len 2
454 try std.testing.expectEqualSlices(u8, &mask, small[2..6]);
455 try std.testing.expectEqual(@as(u8, 'h' ^ 0xaa), small[6]);
456 try std.testing.expectEqual(@as(u8, 'i' ^ 0xbb), small[7]);
457
458 const mid_payload = [_]u8{0x55} ** 300;
459 const mid = try maskedMessage(alloc, &mid_payload, mask);
460 defer alloc.free(mid);
461 try std.testing.expectEqual(@as(u8, 0x80 | 126), mid[1]);
462 try std.testing.expectEqual(@as(u16, 300), std.mem.readInt(u16, mid[2..4], .big));
463 try std.testing.expectEqual(@as(u8, 0x55 ^ 0xcc), mid[4 + 4 + 2]); // idx 2 in payload → mask[2]
464
465 const big_payload = try alloc.alloc(u8, 70 * 1024);
466 defer alloc.free(big_payload);
467 @memset(big_payload, 1);
468 const big = try maskedMessage(alloc, big_payload, mask);
469 defer alloc.free(big);
470 try std.testing.expectEqual(@as(u8, 0x80 | 127), big[1]);
471 try std.testing.expectEqual(@as(u64, 70 * 1024), std.mem.readInt(u64, big[2..10], .big));
472 }
473
474 test "ws reader: split delivery reassembles; server frames arrive unmasked" {
475 const alloc = std.testing.allocator;
476 var r = WsReader{};
477 defer r.buf.deinit(alloc);
478
479 // A 130-byte binary message → 126-form header, unmasked.
480 var payload: [130]u8 = undefined;
481 for (&payload, 0..) |*c, i| c.* = @intCast(i & 0xff);
482 var hdr = [_]u8{ 0x82, 126, 0, 130 };
483 // Feed in three ragged slices; nothing pops until it is whole.
484 try r.buf.appendSlice(alloc, hdr[0..2]);
485 try std.testing.expect(r.peek() == null);
486 try r.buf.appendSlice(alloc, hdr[2..]);
487 try r.buf.appendSlice(alloc, payload[0..70]);
488 try std.testing.expect(r.peek() == null);
489 try r.buf.appendSlice(alloc, payload[70..]);
490 const msg = r.peek().?;
491 try std.testing.expectEqual(@as(u4, 2), msg.opcode);
492 try std.testing.expectEqualSlices(u8, &payload, msg.payload);
493 r.consume(alloc, msg.consumed);
494 try std.testing.expectEqual(@as(usize, 0), r.buf.items.len);
495 }
496
497 test "the dump this exits with is the daemon's own dump format" {
498 // dumpexit writes Engine.dumpPlain — the SAME function muxd dump
499 // prints through — so the e2e diff cannot fail on formatting. The
500 // pin: feed both a daemon-side engine and this fixture's replica the
501 // same snapshot; byte-identical dumps.
502 const alloc = std.testing.allocator;
503 var daemon_eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
504 defer daemon_eng.deinit();
505 daemon_eng.feed("convergence\r\nby construction");
506
507 var fixture_eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
508 defer fixture_eng.deinit();
509 var rep = Replica.init(alloc, fixture_eng);
510 const state = try daemon_eng.dumpState(alloc);
511 defer alloc.free(state);
512 var payload = try alloc.alloc(u8, proto.snapshot_prefix_len + state.len);
513 defer alloc.free(payload);
514 proto.writeSnapshotPrefix(payload[0..proto.snapshot_prefix_len], .{
515 .seq = 1,
516 .history_rows = 0,
517 .cols = 80,
518 .rows = 24,
519 .epoch = 1,
520 });
521 @memcpy(payload[proto.snapshot_prefix_len..], state);
522 _ = try rep.apply(.snapshot, payload);
523
524 const a = try daemon_eng.dumpPlain(alloc);
525 defer alloc.free(a);
526 const b = try fixture_eng.dumpPlain(alloc);
527 defer alloc.free(b);
528 try std.testing.expectEqualStrings(a, b);
529 }