a73x

074ffa6e

feat(muxa): agent client skeleton — status, capture, send over the unix socket

a73x   2026-08-13 17:51

Commit message
feat(muxa): agent client skeleton — status, capture, send over the unix socket

Every verb prints one JSON object on stdout and exits 0; failures print
{"error":...} and exit nonzero, so a driving agent parses one shape on
both paths. Attaches at 0x0 always: applySize refuses anything under 2,
so the slot makes no claim in claimGrid and the human's grid never moves
because an agent connected.

send acknowledges before it closes, and that is not belt-and-braces.
Attaching queues a snapshot, and the daemon flushes a client's pending
bytes BEFORE it reads that client, so write-and-close makes the flush hit
EPIPE, drops us, and discards the input frame still unread. Measured:
closing at once never landed the input, any delay or drain always did.
A status_reply is proof the daemon has read past the input frame, since
frames are served in stream order — the reply's arrival is its content.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
build.zig
Old New
@@ -416,6 +416,17 @@ pub fn build(b: *std.Build) void {
416 linkQuic(b, exe, quic); 416 linkQuic(b, exe, quic);
417 b.installArtifact(exe); 417 b.installArtifact(exe);
418 418
419 // The agent-facing client. Protocol and nothing else: it speaks frames
420 // over the unix socket and owns no terminal, which is the whole point —
421 // it attaches at 0x0 and never claims the grid.
422 const muxa_mod = b.createModule(.{
423 .root_source_file = b.path("src/muxa.zig"),
424 .target = target,
425 .optimize = optimize,
426 .link_libc = true,
427 });
428 muxa_mod.addImport("protocol", protocol_mod);
429
419 const mux_exe = b.addExecutable(.{ .name = "mux", .root_module = mux_mod }); 430 const mux_exe = b.addExecutable(.{ .name = "mux", .root_module = mux_mod });
420 mux_exe.use_llvm = true; 431 mux_exe.use_llvm = true;
421 mux_exe.use_lld = true; 432 mux_exe.use_lld = true;
@@ -423,6 +434,11 @@ pub fn build(b: *std.Build) void {
423 linkQuic(b, mux_exe, quic); 434 linkQuic(b, mux_exe, quic);
424 b.installArtifact(mux_exe); 435 b.installArtifact(mux_exe);
425 436
437 const muxa_exe = b.addExecutable(.{ .name = "muxa", .root_module = muxa_mod });
438 muxa_exe.use_llvm = true;
439 muxa_exe.use_lld = true;
440 b.installArtifact(muxa_exe);
441
426 const rawmode_exe = b.addExecutable(.{ .name = "rawmode", .root_module = rawmode_mod }); 442 const rawmode_exe = b.addExecutable(.{ .name = "rawmode", .root_module = rawmode_mod });
427 rawmode_exe.use_llvm = true; 443 rawmode_exe.use_llvm = true;
428 rawmode_exe.use_lld = true; 444 rawmode_exe.use_lld = true;
@@ -555,7 +571,7 @@ pub fn build(b: *std.Build) void {
555 // absence here was a live hazard recorded in decisions.md — muxd's 571 // absence here was a live hazard recorded in decisions.md — muxd's
556 // entrypoint could grow tests that silently never ran, exactly as 572 // entrypoint could grow tests that silently never ran, exactly as
557 // mux_main.zig's five did before it was added. 573 // mux_main.zig's five did before it was added.
558 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, delta_mod, cmd_mod, shellint_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| { 574 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, delta_mod, cmd_mod, shellint_mod, replica_mod, keymap_mod, webhub_mod, sockpath_mod, muxa_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| {
559 const t = b.addTest(.{ .root_module = mod }); 575 const t = b.addTest(.{ .root_module = mod });
560 t.use_llvm = true; 576 t.use_llvm = true;
561 t.use_lld = true; 577 t.use_lld = true;
src/muxa.zig
Old New
@@ -0,0 +1,339 @@
1 //! muxa: the agent-facing mux client. Every verb prints one JSON object on
2 //! stdout and exits 0 on success; failures print {"error": "..."} and exit
3 //! nonzero. Attaches at 0x0 always — an agent must never claim the grid
4 //! out from under the human's size (load-bearing spec rule).
5 const std = @import("std");
6 const proto = @import("protocol");
7
8 const usage =
9 \\usage: muxa <verb> [--sock PATH] [--settle MS] [--timeout MS] [--vt] [args]
10 \\verbs:
11 \\ status session snapshot as JSON
12 \\ capture current grid as text (--vt for styled)
13 \\ send BYTES raw bytes to the pty (C-style escapes: \n \r \t \e \xNN)
14 \\ run CMDLINE send CMDLINE + newline, await return, report exit/output
15 \\ await wait for the current/next command to return
16 \\
17 ;
18
19 const Opts = struct {
20 verb: enum { status, capture, send, run, @"await" },
21 sock: ?[]const u8 = null,
22 settle_ms: u32 = 0,
23 // Never 0 by default: the daemon reads a 0 timeout on await_req as "no
24 // bound at all" (documented on AwaitReq), so a muxa that defaulted to 0
25 // would turn every await into an unbounded wait.
26 timeout_ms: u32 = 30_000,
27 vt: bool = false,
28 arg: ?[]const u8 = null,
29 };
30
31 fn parseArgs(args: []const [:0]const u8) ?Opts {
32 if (args.len < 2) return null;
33 const verb = std.meta.stringToEnum(@FieldType(Opts, "verb"), args[1]) orelse return null;
34 var o: Opts = .{ .verb = verb };
35 var i: usize = 2;
36 while (i < args.len) : (i += 1) {
37 const a = args[i];
38 if (std.mem.eql(u8, a, "--sock")) {
39 i += 1;
40 if (i >= args.len) return null;
41 o.sock = args[i];
42 } else if (std.mem.eql(u8, a, "--settle")) {
43 i += 1;
44 if (i >= args.len) return null;
45 o.settle_ms = std.fmt.parseInt(u32, args[i], 10) catch return null;
46 } else if (std.mem.eql(u8, a, "--timeout")) {
47 i += 1;
48 if (i >= args.len) return null;
49 o.timeout_ms = std.fmt.parseInt(u32, args[i], 10) catch return null;
50 } else if (std.mem.eql(u8, a, "--vt")) {
51 o.vt = true;
52 } else if (o.arg == null and a.len > 0 and a[0] != '-') {
53 o.arg = a;
54 } else return null;
55 }
56 return o;
57 }
58
59 /// JSON string escape, the six mandatory escapes + control bytes as \u00XX.
60 fn jsonEscape(writer: anytype, s: []const u8) !void {
61 try writer.writeByte('"');
62 for (s) |b| switch (b) {
63 '"' => try writer.writeAll("\\\""),
64 '\\' => try writer.writeAll("\\\\"),
65 '\n' => try writer.writeAll("\\n"),
66 '\r' => try writer.writeAll("\\r"),
67 '\t' => try writer.writeAll("\\t"),
68 0x00...0x08, 0x0b, 0x0c, 0x0e...0x1f => try writer.print("\\u{x:0>4}", .{b}),
69 else => try writer.writeByte(b),
70 };
71 try writer.writeByte('"');
72 }
73
74 test "jsonEscape pins the escapes" {
75 var buf: [128]u8 = undefined;
76 var fbs = std.io.fixedBufferStream(&buf);
77 try jsonEscape(fbs.writer(), "a\"b\\c\nd\x1be");
78 try std.testing.expectEqualStrings("\"a\\\"b\\\\c\\nd\\u001be\"", fbs.getWritten());
79 }
80
81 /// Decode C-style escapes for `send`. Caller frees.
82 fn decodeEscapes(alloc: std.mem.Allocator, s: []const u8) ![]u8 {
83 var out: std.ArrayList(u8) = .empty;
84 errdefer out.deinit(alloc);
85 var i: usize = 0;
86 while (i < s.len) : (i += 1) {
87 if (s[i] != '\\' or i + 1 >= s.len) {
88 try out.append(alloc, s[i]);
89 continue;
90 }
91 i += 1;
92 switch (s[i]) {
93 'n' => try out.append(alloc, '\n'),
94 'r' => try out.append(alloc, '\r'),
95 't' => try out.append(alloc, '\t'),
96 'e' => try out.append(alloc, 0x1b),
97 '\\' => try out.append(alloc, '\\'),
98 'x' => {
99 if (i + 2 >= s.len) return error.BadEscape;
100 try out.append(alloc, try std.fmt.parseInt(u8, s[i + 1 .. i + 3], 16));
101 i += 2;
102 },
103 else => return error.BadEscape,
104 }
105 }
106 return out.toOwnedSlice(alloc);
107 }
108
109 test "decodeEscapes covers the sequences send needs" {
110 const alloc = std.testing.allocator;
111 const got = try decodeEscapes(alloc, "q\\n\\e[A\\x03");
112 defer alloc.free(got);
113 try std.testing.expectEqualSlices(u8, "q\n\x1b[A\x03", got);
114 try std.testing.expectError(error.BadEscape, decodeEscapes(alloc, "\\q"));
115 }
116
117 test "parseArgs verbs and flags" {
118 const a1 = [_][:0]const u8{ "muxa", "status" };
119 try std.testing.expectEqual(@FieldType(Opts, "verb").status, parseArgs(&a1).?.verb);
120 const a2 = [_][:0]const u8{ "muxa", "run", "--timeout", "5000", "make test" };
121 const o2 = parseArgs(&a2).?;
122 try std.testing.expectEqual(@as(u32, 5000), o2.timeout_ms);
123 try std.testing.expectEqualStrings("make test", o2.arg.?);
124 const a3 = [_][:0]const u8{ "muxa", "bogus" };
125 try std.testing.expectEqual(@as(?Opts, null), parseArgs(&a3));
126 }
127
128 const Conn = struct {
129 fd: std.posix.fd_t,
130
131 fn open(sock_path: []const u8) !Conn {
132 const s = try std.net.connectUnixSocket(sock_path);
133 return .{ .fd = s.handle };
134 }
135
136 fn close(self: *Conn) void {
137 std.posix.close(self.fd);
138 }
139
140 fn sendFrame(self: *Conn, t: proto.MsgType, payload: []const u8) !void {
141 try proto.writeFrame(self.fd, t, payload);
142 }
143
144 /// Read frames until one of type `want` arrives (snapshots, deltas and
145 /// pushes stream past an attached client; skip what we did not ask
146 /// for). Bounded by `deadline_ms` wall time via poll.
147 fn awaitFrame(
148 self: *Conn,
149 alloc: std.mem.Allocator,
150 want: proto.MsgType,
151 deadline_ms: i64,
152 ) !proto.Frame {
153 while (true) {
154 const now = std.time.milliTimestamp();
155 if (now >= deadline_ms) return error.Timeout;
156 var fds = [_]std.posix.pollfd{
157 .{ .fd = self.fd, .events = std.posix.POLL.IN, .revents = 0 },
158 };
159 const n = try std.posix.poll(&fds, @intCast(@min(deadline_ms - now, 250)));
160 if (n == 0) continue;
161 const frame = try proto.readFrame(alloc, self.fd) orelse return error.DaemonGone;
162 if (frame.type == want) return frame;
163 frame.deinit(alloc);
164 }
165 }
166 };
167
168 /// Every failure exit goes through here, so stdout carries one JSON object
169 /// whatever went wrong — a driving agent parses the same shape on both
170 /// paths instead of switching on exit code first.
171 fn fail(msg: []const u8, detail: []const u8) u8 {
172 var buf: [2048]u8 = undefined;
173 var fbs = std.io.fixedBufferStream(&buf);
174 writeError(fbs.writer(), msg, detail) catch {
175 // The message did not fit. Still JSON, still one line.
176 proto.writeAllFd(std.posix.STDOUT_FILENO, "{\"error\":\"failure too long to report\"}\n") catch {};
177 return 1;
178 };
179 proto.writeAllFd(std.posix.STDOUT_FILENO, fbs.getWritten()) catch {};
180 return 1;
181 }
182
183 fn writeError(writer: anytype, msg: []const u8, detail: []const u8) !void {
184 try writer.writeAll("{\"error\":");
185 try jsonEscape(writer, msg);
186 try writer.writeAll(",\"detail\":");
187 try jsonEscape(writer, detail);
188 try writer.writeAll("}\n");
189 }
190
191 fn defaultSockPath(alloc: std.mem.Allocator) ![]const u8 {
192 if (std.posix.getenv("XDG_RUNTIME_DIR")) |dir| {
193 return std.fmt.allocPrint(alloc, "{s}/muxd.sock", .{dir});
194 }
195 return std.fmt.allocPrint(alloc, "/tmp/muxd-{d}.sock", .{std.os.linux.getuid()});
196 }
197
198 pub fn main() !u8 {
199 var arena_state = std.heap.ArenaAllocator.init(std.heap.page_allocator);
200 defer arena_state.deinit();
201 const alloc = arena_state.allocator();
202
203 const args = try std.process.argsAlloc(alloc);
204 const o = parseArgs(args) orelse {
205 // Usage is diagnostic, so it goes to stderr: stdout stays strictly
206 // one JSON object per invocation, even on the argument-error path.
207 proto.writeAllFd(std.posix.STDERR_FILENO, usage) catch {};
208 return 2;
209 };
210
211 const sock_path = if (o.sock) |s| s else try defaultSockPath(alloc);
212
213 var conn = Conn.open(sock_path) catch |e| {
214 return fail("cannot connect to the daemon", @errorName(e));
215 };
216 defer conn.close();
217
218 const deadline = std.time.milliTimestamp() + o.timeout_ms;
219 return switch (o.verb) {
220 .status => verbStatus(alloc, &conn, deadline),
221 .capture => verbCapture(alloc, &conn, o.vt, deadline),
222 .send => verbSend(alloc, &conn, o.arg, deadline),
223 .run => fail("run: not implemented yet", ""),
224 .@"await" => fail("await: not implemented yet", ""),
225 };
226 }
227
228 fn verbStatus(alloc: std.mem.Allocator, conn: *Conn, deadline: i64) !u8 {
229 conn.sendFrame(.status_req, "") catch |e| return fail("status: send failed", @errorName(e));
230 const frame = conn.awaitFrame(alloc, .status_reply, deadline) catch |e|
231 return fail("status: no reply", @errorName(e));
232 defer frame.deinit(alloc);
233 const st = proto.decodeStatusReply(frame.payload) catch |e|
234 return fail("status: bad reply", @errorName(e));
235
236 var out: std.ArrayList(u8) = .empty;
237 defer out.deinit(alloc);
238 try printStatus(out.writer(alloc), st);
239 proto.writeAllFd(std.posix.STDOUT_FILENO, out.items) catch {};
240 return 0;
241 }
242
243 fn printStatus(writer: anytype, st: proto.StatusReply) !void {
244 try writer.print(
245 "{{\"cols\":{d},\"rows\":{d},\"cursor\":{{\"x\":{d},\"y\":{d}}}," ++
246 "\"history_rows\":{d},\"alt_screen\":{},\"icanon\":{},\"echo\":{},\"cmd\":{{\"phase\":",
247 .{ st.cols, st.rows, st.cursor_x, st.cursor_y, st.history_rows, st.alt_screen, st.mode.icanon, st.mode.echo },
248 );
249 try jsonEscape(writer, @tagName(st.cmd.phase));
250 try writer.writeAll(",\"mechanism\":");
251 try jsonEscape(writer, @tagName(st.cmd.mechanism));
252 try writer.writeAll(",\"exit_code\":");
253 // A command that has not returned has no exit code, and JSON null is
254 // the honest spelling — 0 would read as "succeeded".
255 if (st.cmd.exit_code) |c| {
256 try writer.print("{d}", .{c});
257 } else {
258 try writer.writeAll("null");
259 }
260 try writer.print(
261 ",\"start_row\":{d},\"end_row\":{d},\"seq\":{d}}}}}\n",
262 .{ st.cmd.start_row, st.cmd.end_row, st.cmd.seq },
263 );
264 }
265
266 test "printStatus spells a pending exit code as JSON null" {
267 var buf: [512]u8 = undefined;
268 var fbs = std.io.fixedBufferStream(&buf);
269 try printStatus(fbs.writer(), .{
270 .cols = 80,
271 .rows = 24,
272 .cursor_x = 1,
273 .cursor_y = 2,
274 .history_rows = 7,
275 .alt_screen = false,
276 .mode = .{ .icanon = true, .echo = true },
277 .cmd = .{ .phase = .running, .mechanism = .marks, .exit_code = null, .start_row = 3, .end_row = 4, .seq = 9 },
278 });
279 const got = fbs.getWritten();
280 try std.testing.expect(std.mem.indexOf(u8, got, "\"exit_code\":null") != null);
281 try std.testing.expect(std.mem.indexOf(u8, got, "\"phase\":\"running\"") != null);
282 try std.testing.expect(std.mem.indexOf(u8, got, "\"cursor\":{\"x\":1,\"y\":2}") != null);
283 }
284
285 fn verbCapture(alloc: std.mem.Allocator, conn: *Conn, vt: bool, deadline: i64) !u8 {
286 const payload = [_]u8{if (vt) 1 else 0};
287 conn.sendFrame(.debug_dump, &payload) catch |e| return fail("capture: send failed", @errorName(e));
288 const frame = conn.awaitFrame(alloc, .dump_reply, deadline) catch |e|
289 return fail("capture: no reply", @errorName(e));
290 defer frame.deinit(alloc);
291
292 var out: std.ArrayList(u8) = .empty;
293 defer out.deinit(alloc);
294 const writer = out.writer(alloc);
295 try writer.writeAll("{\"grid\":");
296 try jsonEscape(writer, frame.payload);
297 try writer.writeAll("}\n");
298 proto.writeAllFd(std.posix.STDOUT_FILENO, out.items) catch {};
299 return 0;
300 }
301
302 /// Join the session claiming NO grid. applySize refuses anything under 2,
303 /// so the slot stays 0x0 and makes no claim in claimGrid: the human's
304 /// terminal must never be resized because an agent connected.
305 fn attachZero(conn: *Conn) !void {
306 try conn.sendFrame(.attach, &proto.encodeAttach(0, 0, 0, 0));
307 }
308
309 fn verbSend(alloc: std.mem.Allocator, conn: *Conn, arg: ?[]const u8, deadline: i64) !u8 {
310 const spec = arg orelse return fail("send: needs BYTES", "");
311 const bytes = decodeEscapes(alloc, spec) catch |e| return fail("send: bad escape", @errorName(e));
312 defer alloc.free(bytes);
313
314 attachZero(conn) catch |e| return fail("send: attach failed", @errorName(e));
315 conn.sendFrame(.input, bytes) catch |e| return fail("send: input failed", @errorName(e));
316
317 // Write-and-close LOSES the input, and not as a rare race: attaching
318 // queues a snapshot, and the daemon flushes a client's pending bytes
319 // BEFORE it reads that client (server.zig's poll arm). Closing straight
320 // after the write means the flush hits EPIPE, the daemon drops us, and
321 // the input frame is discarded still unread. Measured: closing at once
322 // never lands, while any delay or drain always does.
323 //
324 // So the round trip is the acknowledgement. Frames are served in stream
325 // order, so a status_reply is proof the daemon has already read PAST the
326 // input frame and fed it to the pty; awaitFrame skips the snapshot and
327 // the pushes on the way, which is what keeps the socket drained enough
328 // for that flush to succeed. Nothing is done with the reply — its
329 // arrival is the whole content.
330 conn.sendFrame(.status_req, "") catch |e| return fail("send: ack request failed", @errorName(e));
331 const ack = conn.awaitFrame(alloc, .status_reply, deadline) catch |e|
332 return fail("send: daemon never acknowledged the input", @errorName(e));
333 ack.deinit(alloc);
334
335 conn.sendFrame(.detach, "") catch |e| return fail("send: detach failed", @errorName(e));
336
337 proto.writeAllFd(std.posix.STDOUT_FILENO, "{\"sent\":true}\n") catch {};
338 return 0;
339 }