test/wsclient.zig
Ref: Size: 40.2 KiB History
//! e2e fixture: a scripted WebSocket client standing in for the browser
//! (M-web Task 9). Speaks the hub's wire (RFC 6455 client side, masked;
//! one envelope byte per message) and maintains a REAL replica from the
//! frames it receives — the same replica.zig the CLI and the wasm core
//! run — so its `dumpexit` grid diffs honestly against `mux d dump`.
//! Driven by a line-oriented script on stdin, ptyclient's conventions:
//! decodeEscapes, expect deadlines in ms, exit codes 2/3/4.
//!
//! Verbs:
//! attach C R [NAME] send an attach quoting the replica's resume args,
//! to session NAME (default: the default session)
//! attachfresh C R [NAME] same, quoting (0,0)
//! send BYTES input frame (escapes decoded)
//! resize C R resize frame
//! expectgrid NEEDLE MS poll the replica's plain dump for NEEDLE
//! expectstate STATE MS wait for a control message naming STATE
//! expectrefused MS wait for an exit_status that arrived before any
//! grid — the browser's own refusal discriminator
//! expectups N MS wait until N `up` control messages have arrived
//! reattach C R MS [NAME] mux.js's ENV_CONTROL handler for MS ms: attach
//! once now and once per `up` that follows
//! settle QUIET MS drain frames until QUIET ms of silence
//! dumpexit write the replica grid (mux d dump format) to
//! --out and exit 0
//!
//! Usage: wsclient --port N --tile IDX --out FILE --err FILE
//! [--origin STR] < script
const std = @import("std");
const Grid = @import("term").grid.Grid;
const Replica = @import("term").replica.Replica;
const proto = @import("term").protocol;
// The script dialect this fixture and ptyclient both speak: the escape
// table and the exit codes (test/script.zig).
const script = @import("script");
const decodeEscapes = script.decodeEscapes;
const EXIT_USAGE = script.EXIT_USAGE;
const EXIT_TIMEOUT = script.EXIT_TIMEOUT;
const EXIT_DIED = script.EXIT_DIED;
var err_file: ?std.fs.File = null;
fn fatal(code: u8, comptime fmt: []const u8, args: anytype) noreturn {
var buf: [512]u8 = undefined;
const msg = std.fmt.bufPrint(&buf, "wsclient: " ++ fmt ++ "\n", args) catch "wsclient: error\n";
if (err_file) |f| f.writeAll(msg) catch {};
std.debug.print("{s}", .{msg});
std.process.exit(code);
}
// ---------------------------------------------------------------------------
// RFC 6455, client side. The server side is std's; this is the ~60-line
// mirror image: masked sends, unmasked receives.
/// One client→server message, ready to write: `h0` is the whole first
/// header byte (FIN | opcode — 0x82 binary, 0x8a pong), then the length
/// form, the 4-byte mask, and the payload XOR'd through it. Client→server
/// frames are always masked, which is the rest of this function.
fn maskedMessage(alloc: std.mem.Allocator, h0: u8, payload: []const u8, mask: [4]u8) ![]u8 {
var out: std.ArrayList(u8) = .empty;
errdefer out.deinit(alloc);
try out.append(alloc, h0);
if (payload.len <= 125) {
try out.append(alloc, 0x80 | @as(u8, @intCast(payload.len)));
} else if (payload.len <= 0xffff) {
try out.append(alloc, 0x80 | 126);
var b: [2]u8 = undefined;
std.mem.writeInt(u16, &b, @intCast(payload.len), .big);
try out.appendSlice(alloc, &b);
} else {
try out.append(alloc, 0x80 | 127);
var b: [8]u8 = undefined;
std.mem.writeInt(u64, &b, payload.len, .big);
try out.appendSlice(alloc, &b);
}
try out.appendSlice(alloc, &mask);
const start = out.items.len;
try out.appendSlice(alloc, payload);
for (out.items[start..], 0..) |*c, i| c.* ^= mask[i % 4];
return out.toOwnedSlice(alloc);
}
/// Server→client frames accumulate here off the socket; whole messages
/// pop out. Control frames (ping/close) are handled by the caller via
/// the opcode.
const WsReader = struct {
buf: std.ArrayList(u8) = .empty,
const Msg = struct { opcode: u4, payload: []const u8, consumed: usize };
/// Parse one whole frame from the front of the buffer, or null.
fn peek(self: *const WsReader) ?Msg {
const b = self.buf.items;
if (b.len < 2) return null;
const opcode: u4 = @truncate(b[0] & 0x0f);
const masked = b[1] & 0x80 != 0;
if (masked) return null; // server frames are never masked; treated as garbage by caller
var len: u64 = b[1] & 0x7f;
var off: usize = 2;
if (len == 126) {
if (b.len < 4) return null;
len = std.mem.readInt(u16, b[2..4], .big);
off = 4;
} else if (len == 127) {
if (b.len < 10) return null;
len = std.mem.readInt(u64, b[2..10], .big);
off = 10;
}
if (len > 64 * 1024 * 1024) return null; // absurd: caller dies on stall
if (b.len < off + len) return null;
return .{ .opcode = opcode, .payload = b[off .. off + @as(usize, @intCast(len))], .consumed = off + @as(usize, @intCast(len)) };
}
fn consume(self: *WsReader, n: usize) void {
const rest = self.buf.items[n..];
std.mem.copyForwards(u8, self.buf.items[0..rest.len], rest);
self.buf.shrinkRetainingCapacity(rest.len);
}
};
// ---------------------------------------------------------------------------
const Client = struct {
alloc: std.mem.Allocator,
sock: std.posix.fd_t,
reader: WsReader = .{},
rep: Replica,
/// Last control-message state seen (the tile chrome's vocabulary).
last_state: [16]u8 = @splat(0),
last_state_len: usize = 0,
/// The size the last script-level `attach` quoted. Every attach this
/// fixture ever sends quotes THIS, never the grid — see sendAttach.
att_cols: u16 = 0,
att_rows: u16 = 0,
/// The session the last script-level `attach` named, "" for the default.
/// Held for the same reason the size is: a re-attach this fixture sends
/// on its own (the resync path) has to land on the SAME session, or the
/// scenario would silently start diffing a different terminal.
att_name: [proto.session_name_max]u8 = @splat(0),
att_name_len: usize = 0,
/// mux.js's discriminator, mirrored so a scenario can assert on it:
/// an exit_status BEFORE any grid is the daemon refusing the attach;
/// after one it is the shell exiting. A fixture that could not tell
/// them apart would pass on either.
got_state: bool = false,
refused: bool = false,
/// COUNTED, not sampled: `last_state` is overwritten by whatever
/// arrives next, and one pump call drains every buffered message —
/// so a `reconnecting` → `up` pair narrating a re-dial can pass
/// through leaving `last_state` reading `up` exactly as it did
/// before. A scenario that must see the re-dial happen counts.
ups: usize = 0,
fn sendMessage(self: *Client, payload: []const u8) void {
self.sendRaw(0x82, payload); // FIN | binary
}
fn sendRaw(self: *Client, h0: u8, payload: []const u8) void {
var mask: [4]u8 = undefined;
std.crypto.random.bytes(&mask);
const msg = maskedMessage(self.alloc, h0, payload, mask) catch fatal(EXIT_USAGE, "oom", .{});
defer self.alloc.free(msg);
writeAll(self.sock, msg) catch fatal(EXIT_DIED, "hub hung up mid-send", .{});
}
fn sendFrame(self: *Client, t: u8, payload: []const u8) void {
var out: std.ArrayList(u8) = .empty;
defer out.deinit(self.alloc);
out.append(self.alloc, 0x00) catch fatal(EXIT_USAGE, "oom", .{});
out.append(self.alloc, t) catch fatal(EXIT_USAGE, "oom", .{});
var lenb: [4]u8 = undefined;
std.mem.writeInt(u32, &lenb, @intCast(payload.len), .little);
out.appendSlice(self.alloc, &lenb) catch fatal(EXIT_USAGE, "oom", .{});
out.appendSlice(self.alloc, payload) catch fatal(EXIT_USAGE, "oom", .{});
self.sendMessage(out.items);
}
/// The ONE place an attach frame is built, which is what keeps the
/// passivity contract honest. An unzoomed wall tile attaches at 0x0 —
/// no size claim at all, refused the grid by applySize and refused
/// claimGrid forever after — and every LATER attach has to quote the
/// same size the script chose. The resync path is the trap: re-attaching
/// at the grid the snapshot taught us (80x24) is an attach at a
/// differing size, which moves the shared grid and repaints every other
/// client, so a stand-in that did that would be pinning the opposite of
/// the contract. The browser has the same rule structurally: mux.js
/// routes every attach through sendAttach, which quotes the tile's
/// scripted size.
///
/// `fresh` quotes (0,0) instead of the replica's resume coordinates —
/// what a resync needs, since there the replica is the suspect part.
///
/// `name` is the session, "" for the default, and it goes on the wire
/// the way mux.js puts it there: bytes appended after the fixed 20,
/// nothing the replica ever sees. An empty name appends nothing, so the
/// unnamed attach is byte-for-byte the pre-M18 one.
fn sendAttach(self: *Client, cols: u16, rows: u16, fresh: bool, name: []const u8) void {
self.att_cols = cols;
self.att_rows = rows;
// copyForwards, not @memcpy: the resync re-attach passes THIS field
// back in as the argument, and @memcpy calls a source that aliases
// the destination undefined behaviour (it panics on it in Debug).
std.mem.copyForwards(u8, self.att_name[0..name.len], name);
self.att_name_len = name.len;
// Both facts are ABOUT the attach now in flight, not about the
// connection — mux.js clears its own `gotState` on every attach
// for the same reason. A tile that held a grid, re-attached, and
// was then refused must read as REFUSED; leaving either latched
// makes the refusal look like a shell exiting, or lets a stale
// one answer a later wait.
self.got_state = false;
self.refused = false;
const q = if (fresh) Replica.AttachArgs{ .have_seq = 0, .have_epoch = 0 } else self.rep.attachArgs();
var buf: [proto.attach_max_len]u8 = undefined;
const att = proto.encodeAttachNamed(&buf, cols, rows, q.have_seq, q.have_epoch, name);
self.sendFrame(@intFromEnum(proto.MsgType.attach), att);
}
/// Pump whatever is on the socket into the reader and apply every
/// whole message. Returns false when the hub hung up. `wait_ms` is
/// one poll's patience, not a deadline.
fn pump(self: *Client, wait_ms: i32) bool {
var fds = [_]std.posix.pollfd{
.{ .fd = self.sock, .events = std.posix.POLL.IN, .revents = 0 },
};
const n = std.posix.poll(&fds, wait_ms) catch return false;
if (n > 0 and fds[0].revents != 0) {
var buf: [64 * 1024]u8 = undefined;
const got = std.posix.read(self.sock, &buf) catch return false;
if (got == 0) return false;
self.reader.buf.appendSlice(self.alloc, buf[0..got]) catch return false;
}
while (self.reader.peek()) |msg| {
self.handle(msg);
self.reader.consume(msg.consumed);
}
return true;
}
fn handle(self: *Client, msg: WsReader.Msg) void {
switch (msg.opcode) {
0x8 => fatal(EXIT_DIED, "hub sent close", .{}),
// The hub pings an idle browser (webhub.ping_idle_ms) and ends
// the tile after three unanswered intervals, so a fixture that
// ignored pings would be a fixture that gets reaped. Answering
// is also what puts a masked PONG on the hub's inbound path —
// the one frame readSmallMessage swallows, and therefore the
// one headFrame has to name for itself.
0x9 => {
self.sendRaw(0x8a, msg.payload); // FIN | pong, payload echoed
return;
},
0x1, 0x2 => {},
else => return,
}
const data = msg.payload;
if (data.len < 1) return;
if (data[0] == 0x01) {
// {"state":"..."} — extracted textually; the vocabulary is
// closed and the producer is ours.
const json = data[1..];
const k = "\"state\":\"";
if (std.mem.indexOf(u8, json, k)) |i| {
const rest = json[i + k.len ..];
const end = std.mem.indexOfScalar(u8, rest, '"') orelse return;
const state = rest[0..end];
const n = @min(state.len, self.last_state.len);
@memcpy(self.last_state[0..n], state[0..n]);
self.last_state_len = n;
if (std.mem.eql(u8, state, "up")) self.ups += 1;
}
return;
}
if (data[0] != 0x00 or data.len < 1 + proto.frame_header_len) return;
const t = data[1];
const plen = std.mem.readInt(u32, data[2..6][0..4], .little);
if (data.len - 6 != plen) return;
const payload = data[6..];
const snapshot: u8 = @intFromEnum(proto.MsgType.snapshot);
const delta: u8 = @intFromEnum(proto.MsgType.delta);
if (t == @intFromEnum(proto.MsgType.exit_status)) {
if (!self.got_state) self.refused = true;
return;
}
if (t == snapshot or t == delta) {
// replica.zig's pinned subtlety, as mux.js mirrors it: a
// DELTA's arrival alone proves the attach was admitted,
// decodable or not.
if (t == delta) self.got_state = true;
const applied = self.rep.apply(@enumFromInt(t), payload) catch |err| switch (err) {
// A snapshot that blanked the grid before it failed: this
// replica holds nothing worth painting, so the stand-in stops
// reporting state, exactly as the browser resets its core.
error.SnapshotAborted => {
self.got_state = false;
return;
},
else => return,
};
if (applied == .painted) self.got_state = true;
if (applied == .resync) {
// Mirror the browser: a garbled delta re-attaches fresh,
// at the TILE's size — never at rep.grid, which is the
// authoritative grid this tile is not allowed to move —
// and to the SAME session the script named.
self.sendAttach(self.att_cols, self.att_rows, true, self.att_name[0..self.att_name_len]);
}
}
// Everything else (pty_mode, scrollback) is visible
// in --err via the state machinery when a scenario needs it.
}
fn stateIs(self: *const Client, want: []const u8) bool {
return std.mem.eql(u8, self.last_state[0..self.last_state_len], want);
}
};
fn writeAll(fd: std.posix.fd_t, data: []const u8) !void {
var i: usize = 0;
while (i < data.len) i += try std.posix.write(fd, data[i..]);
}
fn nowMs() i64 {
return std.time.milliTimestamp();
}
// ---------------------------------------------------------------------------
pub fn main() !void {
var gpa: std.heap.DebugAllocator(.{}) = .init;
defer _ = gpa.deinit();
const alloc = gpa.allocator();
const args = try std.process.argsAlloc(alloc);
defer std.process.argsFree(alloc, args);
var port: ?u16 = null;
var tile: usize = 0;
var out_path: ?[]const u8 = null;
var err_path: ?[]const u8 = null;
var origin: ?[]const u8 = null;
var i: usize = 1;
while (i < args.len) : (i += 1) {
const a = args[i];
// The value check is INSIDE each arm rather than in its condition:
// with `and i + 1 < args.len` up there, a flag in final position
// falls through to the else and reports "unknown arg --port",
// sending the operator to look for a typo in a flag that is
// spelled correctly. ptyclient's loop is the model.
if (std.mem.eql(u8, a, "--port")) {
i += 1;
if (i >= args.len) fatal(EXIT_USAGE, "--port needs a value", .{});
port = std.fmt.parseInt(u16, args[i], 10) catch
fatal(EXIT_USAGE, "--port: not a number: {s}", .{args[i]});
} else if (std.mem.eql(u8, a, "--tile")) {
i += 1;
if (i >= args.len) fatal(EXIT_USAGE, "--tile needs a value", .{});
tile = std.fmt.parseInt(usize, args[i], 10) catch
fatal(EXIT_USAGE, "--tile: not a number: {s}", .{args[i]});
} else if (std.mem.eql(u8, a, "--out")) {
i += 1;
if (i >= args.len) fatal(EXIT_USAGE, "--out needs a path", .{});
out_path = args[i];
} else if (std.mem.eql(u8, a, "--err")) {
i += 1;
if (i >= args.len) fatal(EXIT_USAGE, "--err needs a path", .{});
err_path = args[i];
} else if (std.mem.eql(u8, a, "--origin")) {
i += 1;
if (i >= args.len) fatal(EXIT_USAGE, "--origin needs a value", .{});
origin = args[i];
} else {
fatal(EXIT_USAGE, "unknown arg {s} (usage: wsclient --port N --tile IDX --out F --err F [--origin STR])", .{a});
}
}
const p = port orelse fatal(EXIT_USAGE, "--port required", .{});
const op = out_path orelse fatal(EXIT_USAGE, "--out required", .{});
const ep = err_path orelse fatal(EXIT_USAGE, "--err required", .{});
err_file = std.fs.cwd().createFile(ep, .{}) catch fatal(EXIT_USAGE, "cannot open --err", .{});
// --- TCP + upgrade ---
const addr = std.net.Address.parseIp("127.0.0.1", p) catch unreachable;
const stream = std.net.tcpConnectToAddress(addr) catch
fatal(EXIT_DIED, "cannot connect to 127.0.0.1:{d}", .{p});
const sock = stream.handle;
var key_raw: [16]u8 = undefined;
std.crypto.random.bytes(&key_raw);
var key_b64: [24]u8 = undefined;
_ = std.base64.standard.Encoder.encode(&key_b64, &key_raw);
const default_origin = try std.fmt.allocPrint(alloc, "http://127.0.0.1:{d}", .{p});
defer alloc.free(default_origin);
const req = try std.fmt.allocPrint(alloc, "GET /ws/{d} HTTP/1.1\r\n" ++
"host: 127.0.0.1:{d}\r\n" ++
"connection: upgrade\r\n" ++
"upgrade: websocket\r\n" ++
"sec-websocket-version: 13\r\n" ++
"sec-websocket-key: {s}\r\n" ++
"origin: {s}\r\n\r\n", .{ tile, p, key_b64, origin orelse default_origin });
defer alloc.free(req);
writeAll(sock, req) catch fatal(EXIT_DIED, "hub hung up during upgrade", .{});
// Read the response head. On anything but 101 print the status line
// and exit 4 — the wrong-Origin scenario asserts exactly this.
var head: std.ArrayList(u8) = .empty;
defer head.deinit(alloc);
while (std.mem.indexOf(u8, head.items, "\r\n\r\n") == null) {
if (head.items.len > 16 * 1024) fatal(EXIT_DIED, "oversize upgrade response", .{});
var b: [1024]u8 = undefined;
const n = std.posix.read(sock, &b) catch fatal(EXIT_DIED, "read failed during upgrade", .{});
if (n == 0) fatal(EXIT_DIED, "hub closed during upgrade", .{});
try head.appendSlice(alloc, b[0..n]);
}
const head_end = std.mem.indexOf(u8, head.items, "\r\n\r\n").? + 4;
const status_line = head.items[0..std.mem.indexOf(u8, head.items, "\r\n").?];
if (std.mem.indexOf(u8, status_line, "101") == null)
fatal(EXIT_DIED, "upgrade refused: {s}", .{status_line});
// The accept key must be OUR key's digest — a hub echoing a canned
// value would pass anything else this fixture checks.
var sha = std.crypto.hash.Sha1.init(.{});
sha.update(&key_b64);
sha.update("258EAFA5-E914-47DA-95CA-C5AB0DC85B11");
var digest: [20]u8 = undefined;
sha.final(&digest);
var accept_b64: [28]u8 = undefined;
_ = std.base64.standard.Encoder.encode(&accept_b64, &digest);
if (std.mem.indexOf(u8, head.items[0..head_end], &accept_b64) == null)
fatal(EXIT_DIED, "sec-websocket-accept mismatch", .{});
// --- replica + client ---
const g = try Grid.init(alloc, 80, 24);
defer g.deinit();
var cl = Client{ .alloc = alloc, .sock = sock, .rep = Replica.init(alloc, g) };
defer cl.reader.buf.deinit(alloc);
// Bytes past the head are the first WS frames.
try cl.reader.buf.appendSlice(alloc, head.items[head_end..]);
// One UNSOLICITED pong, which RFC 6455 explicitly allows as a
// one-way heartbeat. Waiting for the hub's own ping would mean
// waiting 30 seconds, so nothing would ever exercise the pong path
// in the e2e suite; sending one here means every hub scenario proves
// it. It is the frame std's readSmallMessage swallows before looping
// to the next one — so if the hub ever stops tossing pongs itself,
// the pump blocks on a frame that has not arrived, the tile stops
// answering, and these scenarios time out.
cl.sendRaw(0x8a, "hello");
// --- script loop ---
var stdin_buf: std.ArrayList(u8) = .empty;
defer stdin_buf.deinit(alloc);
var rbuf: [4096]u8 = undefined;
while (true) {
const n = std.posix.read(std.posix.STDIN_FILENO, &rbuf) catch break;
if (n == 0) break;
try stdin_buf.appendSlice(alloc, rbuf[0..n]);
}
var lines = std.mem.splitScalar(u8, stdin_buf.items, '\n');
while (lines.next()) |raw| {
const line = std.mem.trim(u8, raw, " \t\r");
if (line.len == 0 or line[0] == '#') continue;
const sp = std.mem.indexOfScalar(u8, line, ' ');
const verb = if (sp) |s| line[0..s] else line;
const rest = if (sp) |s| line[s + 1 ..] else "";
if (std.mem.eql(u8, verb, "attach") or std.mem.eql(u8, verb, "attachfresh")) {
var it = std.mem.tokenizeScalar(u8, rest, ' ');
const cols = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "attach C R", .{}), 10) catch fatal(EXIT_USAGE, "attach C R", .{});
const rows = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "attach C R", .{}), 10) catch fatal(EXIT_USAGE, "attach C R", .{});
// The optional third token is the session; absent means the
// default, the empty name. Still no fourth: a stray token means
// the operator meant something this verb does not do, and
// silently dropping it is how a scenario ends up asserting
// nothing (ptyclient's resize, same rule).
const name = it.next() orelse "";
if (name.len > 0 and !proto.validSessionName(name))
fatal(EXIT_USAGE, "attach C R [NAME]: bad session name {s}", .{name});
if (it.next() != null) fatal(EXIT_USAGE, "attach C R [NAME] takes at most three arguments", .{});
cl.sendAttach(cols, rows, std.mem.eql(u8, verb, "attachfresh"), name);
} else if (std.mem.eql(u8, verb, "send")) {
const bytes = decodeEscapes(alloc, rest) catch fatal(EXIT_USAGE, "bad escape in send", .{});
defer alloc.free(bytes);
cl.sendFrame(@intFromEnum(proto.MsgType.input), bytes);
} else if (std.mem.eql(u8, verb, "resize")) {
var it = std.mem.tokenizeScalar(u8, rest, ' ');
const cols = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "resize C R", .{}), 10) catch fatal(EXIT_USAGE, "resize C R", .{});
const rows = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "resize C R", .{}), 10) catch fatal(EXIT_USAGE, "resize C R", .{});
if (it.next() != null) fatal(EXIT_USAGE, "resize C R takes exactly two arguments", .{});
const sz = proto.encodeSize(cols, rows);
cl.sendFrame(@intFromEnum(proto.MsgType.resize), &sz);
} else if (std.mem.eql(u8, verb, "expectgrid")) {
const last = std.mem.lastIndexOfScalar(u8, rest, ' ') orelse fatal(EXIT_USAGE, "expectgrid NEEDLE MS", .{});
if (last == 0) fatal(EXIT_USAGE, "empty needle", .{});
const ms = std.fmt.parseInt(i64, rest[last + 1 ..], 10) catch fatal(EXIT_USAGE, "bad deadline", .{});
const needle = decodeEscapes(alloc, rest[0..last]) catch fatal(EXIT_USAGE, "bad escape", .{});
defer alloc.free(needle);
const deadline = nowMs() + ms;
while (true) {
const dump = cl.rep.grid.dumpPlain(alloc) catch fatal(EXIT_USAGE, "oom", .{});
const hit = std.mem.indexOf(u8, dump, needle) != null;
alloc.free(@constCast(dump));
if (hit) break;
if (nowMs() >= deadline) fatal(EXIT_TIMEOUT, "expectgrid '{s}' timed out", .{needle});
if (!cl.pump(50)) fatal(EXIT_DIED, "hub hung up during expectgrid", .{});
}
} else if (std.mem.eql(u8, verb, "expectstate")) {
const last = std.mem.lastIndexOfScalar(u8, rest, ' ') orelse fatal(EXIT_USAGE, "expectstate STATE MS", .{});
// A doubled space leaves an empty state, and the empty string
// is what last_state holds before any control message — so the
// wait would pass on having observed nothing. expectgrid
// refuses the same shape above.
if (last == 0) fatal(EXIT_USAGE, "empty state", .{});
const ms = std.fmt.parseInt(i64, rest[last + 1 ..], 10) catch fatal(EXIT_USAGE, "bad deadline", .{});
const want = rest[0..last];
const deadline = nowMs() + ms;
while (!cl.stateIs(want)) {
if (nowMs() >= deadline) fatal(EXIT_TIMEOUT, "expectstate '{s}' timed out (at '{s}')", .{ want, cl.last_state[0..cl.last_state_len] });
if (!cl.pump(50)) fatal(EXIT_DIED, "hub hung up during expectstate", .{});
}
} else if (std.mem.eql(u8, verb, "expectrefused")) {
const ms = std.fmt.parseInt(i64, rest, 10) catch fatal(EXIT_USAGE, "expectrefused MS", .{});
const deadline = nowMs() + ms;
while (!cl.refused) {
if (nowMs() >= deadline) fatal(EXIT_TIMEOUT, "expectrefused timed out (state '{s}')", .{cl.last_state[0..cl.last_state_len]});
if (!cl.pump(50)) fatal(EXIT_DIED, "hub hung up during expectrefused", .{});
}
} else if (std.mem.eql(u8, verb, "expectups")) {
var it = std.mem.tokenizeScalar(u8, rest, ' ');
const want = std.fmt.parseInt(usize, it.next() orelse fatal(EXIT_USAGE, "expectups N MS", .{}), 10) catch fatal(EXIT_USAGE, "expectups N MS", .{});
const ms = std.fmt.parseInt(i64, it.next() orelse fatal(EXIT_USAGE, "expectups N MS", .{}), 10) catch fatal(EXIT_USAGE, "expectups N MS", .{});
const deadline = nowMs() + ms;
while (cl.ups < want) {
if (nowMs() >= deadline) fatal(EXIT_TIMEOUT, "expectups {d} timed out at {d}", .{ want, cl.ups });
if (!cl.pump(50)) fatal(EXIT_DIED, "hub hung up during expectups", .{});
}
} else if (std.mem.eql(u8, verb, "reattach")) {
// The page's own reflex, which no other verb models: mux.js
// re-attaches on EVERY `up`, so a hub that redials in a tight
// loop gets a tight loop of attaches back. A scenario that
// scripted a fixed number of attaches would measure its own
// script instead of the hub's dial rate.
var it = std.mem.tokenizeScalar(u8, rest, ' ');
const cols = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "reattach C R MS [NAME]", .{}), 10) catch fatal(EXIT_USAGE, "reattach C R MS [NAME]", .{});
const rows = std.fmt.parseInt(u16, it.next() orelse fatal(EXIT_USAGE, "reattach C R MS [NAME]", .{}), 10) catch fatal(EXIT_USAGE, "reattach C R MS [NAME]", .{});
const ms = std.fmt.parseInt(i64, it.next() orelse fatal(EXIT_USAGE, "reattach C R MS [NAME]", .{}), 10) catch fatal(EXIT_USAGE, "reattach C R MS [NAME]", .{});
const name = it.next() orelse "";
if (name.len > 0 and !proto.validSessionName(name))
fatal(EXIT_USAGE, "reattach C R MS [NAME]: bad session name {s}", .{name});
if (it.next() != null) fatal(EXIT_USAGE, "reattach C R MS [NAME] takes at most four arguments", .{});
const deadline = nowMs() + ms;
var seen = cl.ups;
cl.sendAttach(cols, rows, true, name);
while (nowMs() < deadline) {
// A hub that hangs up ends the window rather than the run:
// the dial count outside is the assertion, and it is
// readable either way.
if (!cl.pump(50)) break;
if (cl.ups != seen) {
seen = cl.ups;
cl.sendAttach(cols, rows, true, name);
}
}
} else if (std.mem.eql(u8, verb, "settle")) {
var it = std.mem.tokenizeScalar(u8, rest, ' ');
const quiet = std.fmt.parseInt(i64, it.next() orelse fatal(EXIT_USAGE, "settle QUIET MS", .{}), 10) catch fatal(EXIT_USAGE, "settle QUIET MS", .{});
const ms = std.fmt.parseInt(i64, it.next() orelse fatal(EXIT_USAGE, "settle QUIET MS", .{}), 10) catch fatal(EXIT_USAGE, "settle QUIET MS", .{});
const deadline = nowMs() + ms;
var last_traffic = nowMs();
while (nowMs() - last_traffic < quiet) {
// Not `break`: a settle that gives up silently reports a
// quiesced session it never observed, and the scenario
// that follows then diffs a grid still in motion — a
// divergence blamed on the replica. ptyclient's settle
// exits 3 here for the same reason.
if (nowMs() >= deadline)
fatal(EXIT_TIMEOUT, "settle: never saw {d}ms of silence within {d}ms (state '{s}', seq {d})", .{ quiet, ms, cl.last_state[0..cl.last_state_len], cl.rep.last_seq });
const before = cl.rep.last_seq;
if (!cl.pump(50)) fatal(EXIT_DIED, "hub hung up during settle", .{});
if (cl.rep.last_seq != before) last_traffic = nowMs();
}
} else if (std.mem.eql(u8, verb, "dumpexit")) {
const dump = cl.rep.grid.dumpPlain(alloc) catch fatal(EXIT_USAGE, "oom", .{});
defer alloc.free(@constCast(dump));
const out = std.fs.cwd().createFile(op, .{}) catch fatal(EXIT_USAGE, "cannot open --out", .{});
defer out.close();
out.writeAll(dump) catch fatal(EXIT_USAGE, "write --out failed", .{});
// `mux d dump` ends its output with a newline; match it so the
// e2e diff compares grids, not file conventions.
if (dump.len == 0 or dump[dump.len - 1] != '\n')
out.writeAll("\n") catch fatal(EXIT_USAGE, "write --out failed", .{});
stream.close();
return;
} else {
fatal(EXIT_USAGE, "unknown verb {s}", .{verb});
}
}
fatal(EXIT_USAGE, "script ended without dumpexit", .{});
}
// ---------------------------------------------------------------------------
// Tests. The fixture itself holds a replica and parses no VT, exactly like
// the browser it stands in for; only the tests below reach for the engine,
// to build the daemon's own frames rather than hand-write bytes the wire
// would have to be trusted to match.
const Engine = @import("engine").Engine;
const delta_mod = @import("engine").delta;
test "masked message: header layout and mask application, all three length forms" {
const alloc = std.testing.allocator;
const mask = [4]u8{ 0xaa, 0xbb, 0xcc, 0xdd };
const small = try maskedMessage(alloc, 0x82, "hi", mask);
defer alloc.free(small);
try std.testing.expectEqual(@as(u8, 0x82), small[0]); // FIN | binary
try std.testing.expectEqual(@as(u8, 0x80 | 2), small[1]); // masked, len 2
try std.testing.expectEqualSlices(u8, &mask, small[2..6]);
try std.testing.expectEqual(@as(u8, 'h' ^ 0xaa), small[6]);
try std.testing.expectEqual(@as(u8, 'i' ^ 0xbb), small[7]);
const mid_payload = [_]u8{0x55} ** 300;
const mid = try maskedMessage(alloc, 0x82, &mid_payload, mask);
defer alloc.free(mid);
try std.testing.expectEqual(@as(u8, 0x80 | 126), mid[1]);
try std.testing.expectEqual(@as(u16, 300), std.mem.readInt(u16, mid[2..4], .big));
try std.testing.expectEqual(@as(u8, 0x55 ^ 0xcc), mid[4 + 4 + 2]); // idx 2 in payload → mask[2]
const big_payload = try alloc.alloc(u8, 70 * 1024);
defer alloc.free(big_payload);
@memset(big_payload, 1);
const big = try maskedMessage(alloc, 0x82, big_payload, mask);
defer alloc.free(big);
try std.testing.expectEqual(@as(u8, 0x80 | 127), big[1]);
try std.testing.expectEqual(@as(u64, 70 * 1024), std.mem.readInt(u64, big[2..10], .big));
// A pong is the same masked envelope with a different opcode — the
// fixture answers the hub's idle ping with one, and that is the only
// reason the opcode is a parameter at all.
const pong = try maskedMessage(alloc, 0x8a, "hello", mask);
defer alloc.free(pong);
try std.testing.expectEqual(@as(u8, 0x8a), pong[0]); // FIN | pong
try std.testing.expectEqual(@as(u8, 0x80 | 5), pong[1]); // masked, len 5
try std.testing.expectEqual(@as(u8, 'h' ^ 0xaa), pong[6]);
}
test "ws reader: split delivery reassembles; server frames arrive unmasked" {
const alloc = std.testing.allocator;
var r = WsReader{};
defer r.buf.deinit(alloc);
// A 130-byte binary message → 126-form header, unmasked.
var payload: [130]u8 = undefined;
for (&payload, 0..) |*c, i| c.* = @intCast(i & 0xff);
var hdr = [_]u8{ 0x82, 126, 0, 130 };
// Feed in three ragged slices; nothing pops until it is whole.
try r.buf.appendSlice(alloc, hdr[0..2]);
try std.testing.expect(r.peek() == null);
try r.buf.appendSlice(alloc, hdr[2..]);
try r.buf.appendSlice(alloc, payload[0..70]);
try std.testing.expect(r.peek() == null);
try r.buf.appendSlice(alloc, payload[70..]);
const msg = r.peek().?;
try std.testing.expectEqual(@as(u4, 2), msg.opcode);
try std.testing.expectEqualSlices(u8, &payload, msg.payload);
r.consume(msg.consumed);
try std.testing.expectEqual(@as(usize, 0), r.buf.items.len);
}
test "the resync re-attach quotes the TILE's size, never the grid it learned" {
// The passivity contract, pinned at the WS WRITE SEAM: a wall tile
// attaches at 0x0, learns the authoritative 80x24 grid from the
// snapshot, and then hits a garbled delta. The re-attach that follows
// must still claim nothing — quoting the learned grid would be an
// attach at a differing size, which claims the shared session and
// repaints every other client, i.e. the exact opposite of what a
// passive tile is for.
const alloc = std.testing.allocator;
const fds = try std.posix.pipe();
defer std.posix.close(fds[0]);
defer std.posix.close(fds[1]);
const g = try Grid.init(alloc, 80, 24);
defer g.deinit();
var cl = Client{ .alloc = alloc, .sock = fds[1], .rep = Replica.init(alloc, g) };
defer cl.reader.buf.deinit(alloc);
// Named, so the resync's re-attach is pinned to land on the same
// session as well as the same size: a tile that healed onto the default
// session would be diffing a different terminal from then on.
cl.sendAttach(1, 1, false, "b");
// The daemon's answer: a unicast snapshot carrying the true grid, built
// the way the daemon builds it so the fixture cannot drift from the wire.
const daemon_eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
defer daemon_eng.deinit();
const snap_body = try delta_mod.buildSnapshot(alloc, daemon_eng, .{
.seq = 7,
.history_rows = 0,
.cols = 80,
.rows = 24,
.epoch = 3,
});
defer alloc.free(snap_body);
var snap: std.ArrayList(u8) = .empty;
defer snap.deinit(alloc);
try snap.appendSlice(alloc, &[_]u8{ 0x00, @intFromEnum(proto.MsgType.snapshot), 0, 0, 0, 0 });
try snap.appendSlice(alloc, snap_body);
std.mem.writeInt(u32, snap.items[2..6], @intCast(snap_body.len), .little);
cl.handle(.{ .opcode = 0x2, .payload = snap.items, .consumed = snap.items.len });
try std.testing.expectEqual(@as(u16, 80), cl.rep.grid.cols);
// A delta whose header claims two rows and whose payload carries one:
// the row_count check refuses it and Replica reports .resync.
var body: std.ArrayList(u8) = .empty;
defer body.deinit(alloc);
try proto.appendDeltaHeader(&body, alloc, .{
.seq = 8,
.history_rows = 0,
.cursor_x = 0,
.cursor_y = 0,
.row_count = 2,
});
try proto.appendDeltaRow(&body, alloc, 0, "x");
var bad: std.ArrayList(u8) = .empty;
defer bad.deinit(alloc);
try bad.appendSlice(alloc, &[_]u8{ 0x00, @intFromEnum(proto.MsgType.delta), 0, 0, 0, 0 });
std.mem.writeInt(u32, bad.items[2..6], @intCast(body.items.len), .little);
try bad.appendSlice(alloc, body.items);
cl.handle(.{ .opcode = 0x2, .payload = bad.items, .consumed = bad.items.len });
// Both attaches, off the wire, unmasked the way the hub would.
var got: [1024]u8 = undefined;
const n = try std.posix.read(fds[0], &got);
var off: usize = 0;
var attaches: usize = 0;
var last: proto.AttachReq = undefined;
while (off + 6 <= n) {
const plen: usize = got[off + 1] & 0x7f;
const mask = got[off + 2 ..][0..4].*;
var msg: [64]u8 = undefined;
for (got[off + 6 ..][0..plen], 0..) |c, k| msg[k] = c ^ mask[k % 4];
if (msg[1] == @intFromEnum(proto.MsgType.attach)) {
last = try proto.decodeAttach(msg[6..plen]);
attaches += 1;
}
off += 6 + plen;
}
try std.testing.expectEqual(@as(usize, 2), attaches);
try std.testing.expectEqual(@as(u16, 1), last.cols);
try std.testing.expectEqual(@as(u16, 1), last.rows);
// ...and fresh: what we hold is what was garbled.
try std.testing.expectEqual(@as(u64, 0), last.have_seq);
try std.testing.expectEqual(@as(u64, 0), last.have_epoch);
// ...and still on the session the script named.
try std.testing.expectEqualStrings("b", last.name);
}
test "the dump this exits with is the daemon's own dump format" {
// dumpexit writes the replica's dumpPlain, and the e2e diff compares it
// against what `mux d dump` prints, so the two must not differ on
// formatting. The pin: build a real snapshot from a daemon-side engine,
// apply it to this fixture's replica, and compare the dumps.
//
// Through a trailing-space trim on the ENGINE side, and only that: a
// daemon dumps with ghostty's trimming off, while a client never holds a
// trailing space — the encoder stops a row at its last non-blank cell,
// and the VT formatter that fed the old wire trimmed in the same place.
// `test/e2e_lib.sh assert_ws_converged` strips it on both sides too.
const alloc = std.testing.allocator;
const daemon_eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
defer daemon_eng.deinit();
daemon_eng.feed("convergence \r\nby construction");
const fixture_grid = try Grid.init(alloc, 80, 24);
defer fixture_grid.deinit();
var rep = Replica.init(alloc, fixture_grid);
const payload = try delta_mod.buildSnapshot(alloc, daemon_eng, .{
.seq = 1,
.history_rows = 0,
.cols = 80,
.rows = 24,
.epoch = 1,
});
defer alloc.free(payload);
_ = try rep.apply(.snapshot, payload);
const raw = try daemon_eng.dumpPlain(alloc);
defer alloc.free(raw);
const a = try trimRowTails(alloc, raw);
defer alloc.free(a);
const b = try fixture_grid.dumpPlain(alloc);
defer alloc.free(b);
try std.testing.expectEqualStrings(a, b);
}
/// A copy of `text` with each row's trailing spaces removed.
fn trimRowTails(alloc: std.mem.Allocator, text: []const u8) ![]u8 {
var out: std.ArrayList(u8) = .empty;
errdefer out.deinit(alloc);
var it = std.mem.splitScalar(u8, text, '\n');
var first = true;
while (it.next()) |line| {
if (!first) try out.append(alloc, '\n');
first = false;
try out.appendSlice(alloc, std.mem.trimRight(u8, line, " "));
}
return out.toOwnedSlice(alloc);
}
// Forces semantic analysis of every pub decl under `zig build test`, so an
// unreferenced decl must at least compile (the silent-module-loss hazard,
// decisions.md). Pub decls only: std.meta.declarations sees nothing private.
test {
std.testing.refAllDeclsRecursive(@This());
}