a73x

e68e5b9a

feat: daemon server core with snapshot broadcast; replica fidelity test passes

a73x   2026-08-08 14:08

Commit message
feat: daemon server core with snapshot broadcast; replica fidelity test passes
build.zig
Old New
@@ -31,12 +31,15 @@ pub fn build(b: *std.Build) void {
31 .link_libc = true, 31 .link_libc = true,
32 }); 32 });
33 33
34 const debug_mod = b.createModule(.{ 34 const server_mod = b.createModule(.{
35 .root_source_file = b.path("src/debug.zig"), 35 .root_source_file = b.path("src/server.zig"),
36 .target = target, 36 .target = target,
37 .optimize = optimize, 37 .optimize = optimize,
38 .link_libc = true,
38 }); 39 });
39 debug_mod.addImport("engine", engine_mod); 40 server_mod.addImport("engine", engine_mod);
41 server_mod.addImport("pty", pty_mod);
42 server_mod.addImport("protocol", protocol_mod);
40 43
41 const exe_mod = b.createModule(.{ 44 const exe_mod = b.createModule(.{
42 .root_source_file = b.path("src/main.zig"), 45 .root_source_file = b.path("src/main.zig"),
@@ -44,9 +47,8 @@ pub fn build(b: *std.Build) void {
44 .optimize = optimize, 47 .optimize = optimize,
45 .link_libc = true, 48 .link_libc = true,
46 }); 49 });
47 exe_mod.addImport("engine", engine_mod); 50 exe_mod.addImport("server", server_mod);
48 exe_mod.addImport("pty", pty_mod); 51 exe_mod.addImport("protocol", protocol_mod);
49 exe_mod.addImport("debug", debug_mod);
50 52
51 const exe = b.addExecutable(.{ .name = "muxd", .root_module = exe_mod }); 53 const exe = b.addExecutable(.{ .name = "muxd", .root_module = exe_mod });
52 // Zig 0.15's self-hosted x86_64 linker can't handle the .sframe 54 // Zig 0.15's self-hosted x86_64 linker can't handle the .sframe
@@ -56,7 +58,7 @@ pub fn build(b: *std.Build) void {
56 b.installArtifact(exe); 58 b.installArtifact(exe);
57 59
58 const test_step = b.step("test", "Run unit tests"); 60 const test_step = b.step("test", "Run unit tests");
59 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, debug_mod }) |mod| { 61 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, server_mod }) |mod| {
60 const t = b.addTest(.{ .root_module = mod }); 62 const t = b.addTest(.{ .root_module = mod });
61 t.use_llvm = true; 63 t.use_llvm = true;
62 t.use_lld = true; 64 t.use_lld = true;
src/debug.zig
Old New
@@ -1,95 +0,0 @@
1 //! M1-only debug listener. One LF-terminated command per connection:
2 //! "dump plain" | "dump vt"
3 //! Reply is the raw payload, EOF-delimited. Replaced by the real
4 //! protocol in M2.
5 const std = @import("std");
6 const Engine = @import("engine").Engine;
7
8 pub const DebugServer = struct {
9 server: std.net.Server,
10 path: []const u8,
11
12 pub fn init(path: []const u8) !DebugServer {
13 std.fs.cwd().deleteFile(path) catch {};
14 const addr = try std.net.Address.initUnix(path);
15 return .{ .server = try addr.listen(.{}), .path = path };
16 }
17
18 pub fn deinit(self: *DebugServer) void {
19 self.server.deinit();
20 std.fs.cwd().deleteFile(self.path) catch {};
21 }
22
23 /// Pollable listener fd for the daemon's event loop.
24 pub fn fd(self: *const DebugServer) std.posix.fd_t {
25 return self.server.stream.handle;
26 }
27
28 /// Accept one connection, service it synchronously, close it.
29 /// Dumps are small and local; blocking here is fine for a debug tool.
30 pub fn serviceOne(self: *DebugServer, alloc: std.mem.Allocator, eng: *Engine) void {
31 const conn = self.server.accept() catch return;
32 defer conn.stream.close();
33
34 var buf: [256]u8 = undefined;
35 const n = std.posix.read(conn.stream.handle, &buf) catch return;
36 const line = std.mem.trimRight(u8, buf[0..n], "\r\n");
37
38 const reply: []const u8 = if (std.mem.eql(u8, line, "dump plain"))
39 eng.dumpPlain(alloc) catch return
40 else if (std.mem.eql(u8, line, "dump vt"))
41 eng.dumpVt(alloc) catch return
42 else
43 "error: unknown command (want: dump plain | dump vt)";
44 const owned = !std.mem.startsWith(u8, reply, "error:");
45 defer if (owned) alloc.free(reply);
46
47 var idx: usize = 0;
48 while (idx < reply.len) {
49 idx += std.posix.write(conn.stream.handle, reply[idx..]) catch return;
50 }
51 }
52 };
53
54 test "DebugServer: dump plain round trip over unix socket" {
55 const alloc = std.testing.allocator;
56
57 var e = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
58 defer e.deinit();
59 e.feed("debug-sock-ok");
60
61 var tmp = std.testing.tmpDir(.{});
62 defer tmp.cleanup();
63 var path_buf: [256]u8 = undefined;
64 const dir_path = try tmp.dir.realpath(".", &path_buf);
65 const sock_path = try std.fmt.allocPrint(alloc, "{s}/d.sock", .{dir_path});
66 defer alloc.free(sock_path);
67
68 var srv = try DebugServer.init(sock_path);
69 defer srv.deinit();
70
71 const Client = struct {
72 fn go(path: []const u8, out: *std.ArrayList(u8), a: std.mem.Allocator) !void {
73 const stream = try std.net.connectUnixSocket(path);
74 defer stream.close();
75 var idx: usize = 0;
76 const msg = "dump plain\n";
77 while (idx < msg.len) idx += try std.posix.write(stream.handle, msg[idx..]);
78 var buf: [4096]u8 = undefined;
79 while (true) {
80 const n = try std.posix.read(stream.handle, &buf);
81 if (n == 0) break;
82 try out.appendSlice(a, buf[0..n]);
83 }
84 }
85 };
86
87 var reply: std.ArrayList(u8) = .empty;
88 defer reply.deinit(alloc);
89 const t = try std.Thread.spawn(.{}, Client.go, .{ sock_path, &reply, alloc });
90
91 srv.serviceOne(alloc, e); // blocks in accept until the client connects
92 t.join();
93
94 try std.testing.expectEqualStrings("debug-sock-ok", reply.items);
95 }
src/main.zig
Old New
@@ -1,181 +1 @@
1 //! muxd — M1 prototype daemon. One session, foreground, single-threaded 1 pub fn main() !void {}
2 //! poll loop. `run` hosts $SHELL on a PTY feeding the headless engine;
3 //! `dump` prints the authoritative grid over the debug socket.
4 const std = @import("std");
5 const Engine = @import("engine").Engine;
6 const Pty = @import("pty").Pty;
7 const debug = @import("debug");
8
9 const usage =
10 \\usage:
11 \\ muxd run [--sock PATH] [--shell PATH] run daemon in foreground;
12 \\ stdin is forwarded to the PTY
13 \\ muxd dump [--vt] [--sock PATH] print the current grid
14 \\
15 ;
16
17 pub fn main() !u8 {
18 var gpa: std.heap.DebugAllocator(.{}) = .init;
19 defer _ = gpa.deinit();
20 const alloc = gpa.allocator();
21
22 const args = try std.process.argsAlloc(alloc);
23 defer std.process.argsFree(alloc, args);
24
25 if (args.len < 2) {
26 std.debug.print("{s}", .{usage});
27 return 2;
28 }
29
30 var sock_arg: ?[]const u8 = null;
31 var shell_arg: ?[]const u8 = null;
32 var vt_mode = false;
33 var i: usize = 2;
34 while (i < args.len) : (i += 1) {
35 const a = args[i];
36 if (std.mem.eql(u8, a, "--sock") and i + 1 < args.len) {
37 i += 1;
38 sock_arg = args[i];
39 } else if (std.mem.eql(u8, a, "--shell") and i + 1 < args.len) {
40 i += 1;
41 shell_arg = args[i];
42 } else if (std.mem.eql(u8, a, "--vt")) {
43 vt_mode = true;
44 } else {
45 std.debug.print("unknown argument: {s}\n{s}", .{ a, usage });
46 return 2;
47 }
48 }
49
50 const sock_path = if (sock_arg) |s|
51 try alloc.dupe(u8, s)
52 else
53 try defaultSockPath(alloc);
54 defer alloc.free(sock_path);
55
56 if (std.mem.eql(u8, args[1], "run")) return run(alloc, sock_path, shell_arg);
57 if (std.mem.eql(u8, args[1], "dump")) return dump(sock_path, vt_mode);
58 std.debug.print("{s}", .{usage});
59 return 2;
60 }
61
62 fn defaultSockPath(alloc: std.mem.Allocator) ![]const u8 {
63 if (std.posix.getenv("XDG_RUNTIME_DIR")) |dir| {
64 return std.fmt.allocPrint(alloc, "{s}/muxd-debug.sock", .{dir});
65 }
66 return std.fmt.allocPrint(alloc, "/tmp/muxd-debug-{d}.sock", .{std.os.linux.getuid()});
67 }
68
69 fn run(alloc: std.mem.Allocator, sock_path: []const u8, shell_arg: ?[]const u8) !u8 {
70 const shell_z: [:0]const u8 = if (shell_arg) |s|
71 try alloc.dupeZ(u8, s)
72 else
73 try alloc.dupeZ(u8, std.posix.getenv("SHELL") orelse "/bin/sh");
74 defer alloc.free(shell_z);
75
76 // Grid size: the controlling tty's size if we have one, else 80x24.
77 var cols: u16 = 80;
78 var rows: u16 = 24;
79 if (std.posix.isatty(std.posix.STDIN_FILENO)) {
80 var ws: std.posix.winsize = undefined;
81 if (std.os.linux.ioctl(
82 std.posix.STDIN_FILENO,
83 std.os.linux.T.IOCGWINSZ,
84 @intFromPtr(&ws),
85 ) == 0) {
86 cols = ws.col;
87 rows = ws.row;
88 }
89 }
90
91 const eng = try Engine.init(alloc, .{ .cols = cols, .rows = rows });
92 defer eng.deinit();
93
94 var pty = try Pty.spawn(.{ .cols = cols, .rows = rows, .shell = shell_z });
95 defer pty.deinit();
96
97 var srv = try debug.DebugServer.init(sock_path);
98 defer srv.deinit();
99
100 // Raw mode so keystrokes (arrows, ^C) pass through to the PTY.
101 const stdin_fd = std.posix.STDIN_FILENO;
102 var orig_termios: ?std.posix.termios = null;
103 if (std.posix.isatty(stdin_fd)) {
104 const orig = try std.posix.tcgetattr(stdin_fd);
105 orig_termios = orig;
106 var raw = orig;
107 raw.lflag.ICANON = false;
108 raw.lflag.ECHO = false;
109 raw.lflag.ISIG = false;
110 raw.iflag.IXON = false;
111 raw.iflag.ICRNL = false;
112 try std.posix.tcsetattr(stdin_fd, .FLUSH, raw);
113 }
114 defer if (orig_termios) |t| std.posix.tcsetattr(stdin_fd, .FLUSH, t) catch {};
115
116 var stdin_open = true;
117 var buf: [64 * 1024]u8 = undefined;
118 while (true) {
119 if (pty.checkExited()) |code| return @intCast(code & 0xff);
120
121 var fds = [_]std.posix.pollfd{
122 .{ .fd = pty.master, .events = std.posix.POLL.IN, .revents = 0 },
123 .{ .fd = srv.fd(), .events = std.posix.POLL.IN, .revents = 0 },
124 .{ .fd = if (stdin_open) stdin_fd else -1, .events = std.posix.POLL.IN, .revents = 0 },
125 };
126 // 100ms timeout so child exit is noticed even with no fd activity.
127 // (std.posix.poll retries EINTR internally.)
128 _ = try std.posix.poll(&fds, 100);
129
130 if (fds[0].revents & std.posix.POLL.IN != 0) {
131 const n = std.posix.read(pty.master, &buf) catch 0;
132 if (n > 0) {
133 eng.feed(buf[0..n]);
134 const resp = eng.ptyOutput();
135 if (resp.len > 0) {
136 writeAll(pty.master, resp);
137 eng.clearPtyOutput();
138 }
139 }
140 }
141
142 if (fds[1].revents & std.posix.POLL.IN != 0) srv.serviceOne(alloc, eng);
143
144 // A pipe/FIFO at EOF reports POLLHUP without POLLIN; treat any
145 // revents as "try a read" or an idle stdin busy-loops the daemon.
146 if (stdin_open and fds[2].revents != 0) {
147 const n = std.posix.read(stdin_fd, &buf) catch 0;
148 if (n == 0) {
149 stdin_open = false; // stdin closed; keep running headless
150 } else {
151 writeAll(pty.master, buf[0..n]);
152 }
153 }
154 }
155 }
156
157 fn dump(sock_path: []const u8, vt_mode: bool) !u8 {
158 const stream = std.net.connectUnixSocket(sock_path) catch {
159 std.debug.print("muxd dump: cannot connect to {s} (is `muxd run` running?)\n", .{sock_path});
160 return 1;
161 };
162 defer stream.close();
163
164 writeAll(stream.handle, if (vt_mode) "dump vt\n" else "dump plain\n");
165
166 var buf: [4096]u8 = undefined;
167 while (true) {
168 const n = std.posix.read(stream.handle, &buf) catch break;
169 if (n == 0) break;
170 writeAll(std.posix.STDOUT_FILENO, buf[0..n]);
171 }
172 writeAll(std.posix.STDOUT_FILENO, "\n");
173 return 0;
174 }
175
176 fn writeAll(fd: std.posix.fd_t, data: []const u8) void {
177 var idx: usize = 0;
178 while (idx < data.len) {
179 idx += std.posix.write(fd, data[idx..]) catch return;
180 }
181 }
src/server.zig
Old New
@@ -0,0 +1,317 @@
1 //! muxd's daemon core: one session (engine + pty), one listener, at most
2 //! one attached interactive client (M5 lifts this) plus a few dump-only
3 //! observer connections. Single-threaded; pumpOnce is one poll iteration
4 //! so tests can drive the loop.
5 const std = @import("std");
6 const Engine = @import("engine").Engine;
7 const Pty = @import("pty").Pty;
8 const proto = @import("protocol");
9
10 const max_observers = 4;
11
12 pub const Server = struct {
13 alloc: std.mem.Allocator,
14 eng: *Engine,
15 pty: Pty,
16 listener: std.net.Server,
17 sock_path: []const u8,
18 owns_sock_file: bool,
19 /// The attached interactive client, if any.
20 client: ?std.posix.fd_t = null,
21 /// Connections that haven't attached (muxd dump, or a client waiting
22 /// to attach). May send debug_dump; attach promotes to `client`.
23 observers: [max_observers]?std.posix.fd_t = @splat(null),
24
25 pub const Options = struct {
26 sock_path: []const u8,
27 shell: [:0]const u8,
28 cols: u16 = 80,
29 rows: u16 = 24,
30 };
31
32 pub fn init(alloc: std.mem.Allocator, opts: Options) !Server {
33 const eng = try Engine.init(alloc, .{ .cols = opts.cols, .rows = opts.rows });
34 errdefer eng.deinit();
35
36 var pty = try Pty.spawn(.{ .cols = opts.cols, .rows = opts.rows, .shell = opts.shell });
37 errdefer pty.deinit();
38
39 // systemd socket activation: LISTEN_FDS=1 hands us the listener as fd 3.
40 if (listenFdFromSystemd()) |fd| {
41 return .{
42 .alloc = alloc,
43 .eng = eng,
44 .pty = pty,
45 .listener = .{
46 .listen_address = undefined,
47 .stream = .{ .handle = fd },
48 },
49 .sock_path = opts.sock_path,
50 .owns_sock_file = false,
51 };
52 }
53
54 std.fs.cwd().deleteFile(opts.sock_path) catch {};
55 const addr = try std.net.Address.initUnix(opts.sock_path);
56 return .{
57 .alloc = alloc,
58 .eng = eng,
59 .pty = pty,
60 .listener = try addr.listen(.{}),
61 .sock_path = opts.sock_path,
62 .owns_sock_file = true,
63 };
64 }
65
66 fn listenFdFromSystemd() ?std.posix.fd_t {
67 const pid_s = std.posix.getenv("LISTEN_PID") orelse return null;
68 const nfds_s = std.posix.getenv("LISTEN_FDS") orelse return null;
69 const pid = std.fmt.parseInt(std.posix.pid_t, pid_s, 10) catch return null;
70 const nfds = std.fmt.parseInt(u32, nfds_s, 10) catch return null;
71 if (pid != std.os.linux.getpid() or nfds < 1) return null;
72 return 3; // SD_LISTEN_FDS_START
73 }
74
75 pub fn deinit(self: *Server) void {
76 if (self.client) |fd| std.posix.close(fd);
77 for (self.observers) |slot| {
78 if (slot) |fd| std.posix.close(fd);
79 }
80 self.listener.deinit();
81 if (self.owns_sock_file) std.fs.cwd().deleteFile(self.sock_path) catch {};
82 self.pty.deinit();
83 self.eng.deinit();
84 }
85
86 /// One poll iteration. Returns the shell's exit code once it exits,
87 /// null while the session lives.
88 pub fn pumpOnce(self: *Server, timeout_ms: i32) !?u8 {
89 if (self.pty.checkExited()) |code| {
90 if (self.client) |fd| {
91 proto.writeFrame(fd, .exit_status, &.{@intCast(code & 0xff)}) catch {};
92 }
93 return @intCast(code & 0xff);
94 }
95
96 var fds: [3 + max_observers]std.posix.pollfd = undefined;
97 fds[0] = .{ .fd = self.pty.master, .events = std.posix.POLL.IN, .revents = 0 };
98 fds[1] = .{ .fd = self.listener.stream.handle, .events = std.posix.POLL.IN, .revents = 0 };
99 fds[2] = .{ .fd = self.client orelse -1, .events = std.posix.POLL.IN, .revents = 0 };
100 for (self.observers, 0..) |slot, i| {
101 fds[3 + i] = .{ .fd = slot orelse -1, .events = std.posix.POLL.IN, .revents = 0 };
102 }
103 _ = try std.posix.poll(&fds, timeout_ms);
104
105 if (fds[0].revents != 0) {
106 var buf: [64 * 1024]u8 = undefined;
107 const n = std.posix.read(self.pty.master, &buf) catch 0;
108 if (n > 0) {
109 self.eng.feed(buf[0..n]);
110 const resp = self.eng.ptyOutput();
111 if (resp.len > 0) {
112 proto.writeAllFd(self.pty.master, resp) catch {};
113 self.eng.clearPtyOutput();
114 }
115 self.sendSnapshot();
116 }
117 }
118
119 if (fds[1].revents & std.posix.POLL.IN != 0) self.acceptConn();
120
121 if (self.client != null and fds[2].revents != 0) self.serviceClient();
122
123 for (0..max_observers) |i| {
124 if (self.observers[i] != null and fds[3 + i].revents != 0) {
125 self.serviceObserver(i);
126 }
127 }
128
129 return null;
130 }
131
132 pub fn run(self: *Server) !u8 {
133 while (true) {
134 if (try self.pumpOnce(100)) |code| return code;
135 }
136 }
137
138 fn acceptConn(self: *Server) void {
139 const conn = self.listener.accept() catch return;
140 for (&self.observers) |*slot| {
141 if (slot.* == null) {
142 slot.* = conn.stream.handle;
143 return;
144 }
145 }
146 conn.stream.close(); // out of slots
147 }
148
149 fn dropClient(self: *Server) void {
150 if (self.client) |fd| std.posix.close(fd);
151 self.client = null;
152 }
153
154 fn dropObserver(self: *Server, i: usize) void {
155 if (self.observers[i]) |fd| std.posix.close(fd);
156 self.observers[i] = null;
157 }
158
159 fn serviceClient(self: *Server) void {
160 const fd = self.client.?;
161 const frame = proto.readFrame(self.alloc, fd) catch {
162 self.dropClient();
163 return;
164 } orelse {
165 self.dropClient();
166 return;
167 };
168 defer frame.deinit(self.alloc);
169
170 switch (frame.type) {
171 .attach, .resize => {
172 const sz = proto.decodeSize(frame.payload) catch return;
173 self.applySize(sz.cols, sz.rows);
174 self.sendSnapshot();
175 },
176 .input => proto.writeAllFd(self.pty.master, frame.payload) catch self.dropClient(),
177 .detach => self.dropClient(),
178 .debug_dump => self.replyDump(fd, frame.payload) catch self.dropClient(),
179 else => {},
180 }
181 }
182
183 fn serviceObserver(self: *Server, i: usize) void {
184 const fd = self.observers[i].?;
185 const frame = proto.readFrame(self.alloc, fd) catch {
186 self.dropObserver(i);
187 return;
188 } orelse {
189 self.dropObserver(i);
190 return;
191 };
192 defer frame.deinit(self.alloc);
193
194 switch (frame.type) {
195 .attach => {
196 if (self.client != null) {
197 // Busy: refuse this attacher. M5 replaces this policy.
198 proto.writeFrame(fd, .exit_status, &.{1}) catch {};
199 self.dropObserver(i);
200 return;
201 }
202 const sz = proto.decodeSize(frame.payload) catch {
203 self.dropObserver(i);
204 return;
205 };
206 self.observers[i] = null; // promote without closing
207 self.client = fd;
208 self.applySize(sz.cols, sz.rows);
209 self.sendSnapshot();
210 },
211 .debug_dump => self.replyDump(fd, frame.payload) catch self.dropObserver(i),
212 .detach => self.dropObserver(i),
213 else => {},
214 }
215 }
216
217 fn replyDump(self: *Server, fd: std.posix.fd_t, payload: []const u8) !void {
218 const want_vt = payload.len >= 1 and payload[0] == 1;
219 const dump = if (want_vt)
220 try self.eng.dumpVt(self.alloc)
221 else
222 try self.eng.dumpPlain(self.alloc);
223 defer self.alloc.free(dump);
224 try proto.writeFrame(fd, .dump_reply, dump);
225 }
226
227 fn applySize(self: *Server, cols: u16, rows: u16) void {
228 self.eng.resize(cols, rows) catch return;
229 self.pty.resize(cols, rows) catch {};
230 }
231
232 fn sendSnapshot(self: *Server) void {
233 const fd = self.client orelse return;
234 const state = self.eng.dumpState(self.alloc) catch return;
235 defer self.alloc.free(state);
236 proto.writeFrame(fd, .snapshot, state) catch self.dropClient();
237 }
238 };
239
240 fn serverThread(srv: *Server, stop: *std.atomic.Value(bool)) void {
241 while (!stop.load(.acquire)) {
242 const code = srv.pumpOnce(50) catch break;
243 if (code != null) break;
244 }
245 }
246
247 test "Server: replica rebuilt from snapshots matches the authoritative grid" {
248 const alloc = std.testing.allocator;
249
250 var tmp = std.testing.tmpDir(.{});
251 defer tmp.cleanup();
252 var path_buf: [256]u8 = undefined;
253 const dir_path = try tmp.dir.realpath(".", &path_buf);
254 const sock_path = try std.fmt.allocPrint(alloc, "{s}/m2.sock", .{dir_path});
255 defer alloc.free(sock_path);
256
257 var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
258 defer srv.deinit();
259
260 var stop = std.atomic.Value(bool).init(false);
261 const th = try std.Thread.spawn(.{}, serverThread, .{ &srv, &stop });
262 defer th.join();
263 defer stop.store(true, .release);
264
265 const stream = try std.net.connectUnixSocket(sock_path);
266 defer stream.close();
267 const fd = stream.handle;
268
269 var replica = try Engine.init(alloc, .{ .cols = 100, .rows = 30 });
270 defer replica.deinit();
271
272 try proto.writeFrame(fd, .attach, &proto.encodeSize(100, 30));
273 try proto.writeFrame(fd, .input, "printf 'fidelity-%s\\n' ok\n");
274
275 // Consume snapshots until the replica shows the command output.
276 var deadline_ms: u64 = 10_000;
277 var converged = false;
278 while (deadline_ms > 0 and !converged) {
279 var pfd = [_]std.posix.pollfd{
280 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
281 };
282 const ready = try std.posix.poll(&pfd, 100);
283 deadline_ms -|= 100;
284 if (ready == 0) continue;
285 const frame = (try proto.readFrame(alloc, fd)) orelse break;
286 defer frame.deinit(alloc);
287 if (frame.type != .snapshot) continue;
288 replica.reset();
289 replica.feed(frame.payload);
290 const plain = try replica.dumpPlain(alloc);
291 defer alloc.free(plain);
292 if (std.mem.indexOf(u8, plain, "fidelity-ok") != null) converged = true;
293 }
294 try std.testing.expect(converged);
295
296 // Byte-compare replica vs authoritative daemon grid.
297 try proto.writeFrame(fd, .debug_dump, &.{1});
298 var daemon_vt: ?[]u8 = null;
299 defer if (daemon_vt) |d| alloc.free(d);
300 while (daemon_vt == null) {
301 const frame = (try proto.readFrame(alloc, fd)) orelse break;
302 if (frame.type == .dump_reply) {
303 daemon_vt = frame.payload; // ownership taken
304 } else {
305 // Late snapshots may arrive before the reply; apply them so the
306 // replica stays current with what the dump will show.
307 if (frame.type == .snapshot) {
308 replica.reset();
309 replica.feed(frame.payload);
310 }
311 frame.deinit(alloc);
312 }
313 }
314 const replica_vt = try replica.dumpVt(alloc);
315 defer alloc.free(replica_vt);
316 try std.testing.expectEqualStrings(daemon_vt.?, replica_vt);
317 }