a73x

224be598

feat(muxa): run and await — exit codes over marks, spans from scrollback, session death is an answer

a73x   2026-08-13 18:13

Commit message
feat(muxa): run and await — exit codes over marks, spans from scrollback, session death is an answer

run takes the session's return watermark BEFORE it sends the command line,
never after: the seq of the last return is what the new command must beat,
and a command fast enough to return between the two would have moved the
watermark past a value we never recorded, leaving the await waiting for a
return that had already happened.

The transcript is fetched only in the marks regime and only on a return.
pgid and settle answer WHEN a command ended, never WHERE, so a span from
them would be a guess dressed as output. Even in the marks regime the rows
are best-effort by construction — the alt screen and scrollback pruning can
invalidate them between the reply and the fetch — so every failure there
degrades to no `output` key rather than to a failed run. The exit code is
the answer; the transcript is the bonus.

An exit_status frame mid-wait no longer surfaces as DaemonGone. It is the
session's last word and it carries the only copy of the code, so awaitFrame
captures it and returns error.SessionExited; run and await print
{"reason":"session_ended","exit_code":N} and exit 0, because the command
ending the session IS what happened to it. status, capture and send have no
answer to give and still fail, but structurally — {"error":"session ended",
"exit_code":N} — instead of as a transport error.

Only a timeout exits nonzero (3). A command that returned 1 is a successful
question with the failure in exit_code, and an agent that switched on muxa's
own exit code would conflate the two.

The client's await deadline outlasts the daemon's by a grace window. The
daemon starts its timeout_ms when it READS the request, already later than
the instant this process began counting, so waiting exactly timeout_ms lost
that race every time: measured with the window at 0, `run 'sleep 5'
--timeout 700` reports {"error":"run: no reply"} and exit 1 instead of
{"reason":"timeout"} and exit 3. `--timeout 0` keeps its documented meaning
of no bound at all, here as on the wire, rather than a deadline already past.

Carried from the skeleton's review: the default socket path moves to
sockpath, whose charter is a socket path's identity — muxa thereby also
gains the max_sun_path refusal-by-name that module's doc claims all three
binaries perform; a failed connect names the path it tried, this binary's
likeliest field failure; `--` ends the flags, so `muxa send -- '-n foo\n'`
reaches the pty; a dangling backslash is error.BadEscape like every other
escape we cannot read, rather than the one typo that reaches the pty as a
literal; and awaitFrame now records its blocking debt — poll bounds the
WAIT, not the read, so a peer stalling mid-frame outlives the deadline,
which is harmless over a unix socket and real once QUIC is under it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
build.zig
Old New
@@ -426,6 +426,7 @@ pub fn build(b: *std.Build) void {
426 .link_libc = true, 426 .link_libc = true,
427 }); 427 });
428 muxa_mod.addImport("protocol", protocol_mod); 428 muxa_mod.addImport("protocol", protocol_mod);
429 muxa_mod.addImport("sockpath", sockpath_mod);
429 430
430 const mux_exe = b.addExecutable(.{ .name = "mux", .root_module = mux_mod }); 431 const mux_exe = b.addExecutable(.{ .name = "mux", .root_module = mux_mod });
431 mux_exe.use_llvm = true; 432 mux_exe.use_llvm = true;
src/main.zig
Old New
@@ -293,7 +293,7 @@ pub fn main() !u8 {
293 const sock_path = if (o.sock) |s| 293 const sock_path = if (o.sock) |s|
294 try alloc.dupe(u8, s) 294 try alloc.dupe(u8, s)
295 else 295 else
296 try defaultSockPath(alloc); 296 try sockpath.defaultSockPath(alloc);
297 defer alloc.free(sock_path); 297 defer alloc.free(sock_path);
298 298
299 // The sun_path bound (sockpath.max_sun_path). Checked here, once, 299 // The sun_path bound (sockpath.max_sun_path). Checked here, once,
@@ -487,13 +487,6 @@ fn run(alloc: std.mem.Allocator, o: Opts, sock_path: []const u8) !u8 {
487 return try srv.run(); 487 return try srv.run();
488 } 488 }
489 489
490 pub fn defaultSockPath(alloc: std.mem.Allocator) ![]const u8 {
491 if (std.posix.getenv("XDG_RUNTIME_DIR")) |dir| {
492 return std.fmt.allocPrint(alloc, "{s}/muxd.sock", .{dir});
493 }
494 return std.fmt.allocPrint(alloc, "/tmp/muxd-{d}.sock", .{std.os.linux.getuid()});
495 }
496
497 /// Ask once, print the first reply of the type asked for, exit. `dump` and 490 /// Ask once, print the first reply of the type asked for, exit. `dump` and
498 /// `stats` are this same round-trip and differed only in the verb they 491 /// `stats` are this same round-trip and differed only in the verb they
499 /// name, the frame they send and the frame they wait for. 492 /// name, the frame they send and the frame they wait for.
src/muxa.zig
Old New
@@ -4,6 +4,7 @@
4 //! out from under the human's size (load-bearing spec rule). 4 //! out from under the human's size (load-bearing spec rule).
5 const std = @import("std"); 5 const std = @import("std");
6 const proto = @import("protocol"); 6 const proto = @import("protocol");
7 const sockpath = @import("sockpath");
7 8
8 const usage = 9 const usage =
9 \\usage: muxa <verb> [--sock PATH] [--settle MS] [--timeout MS] [--vt] [args] 10 \\usage: muxa <verb> [--sock PATH] [--settle MS] [--timeout MS] [--vt] [args]
@@ -33,9 +34,19 @@ fn parseArgs(args: []const [:0]const u8) ?Opts {
33 const verb = std.meta.stringToEnum(@FieldType(Opts, "verb"), args[1]) orelse return null; 34 const verb = std.meta.stringToEnum(@FieldType(Opts, "verb"), args[1]) orelse return null;
34 var o: Opts = .{ .verb = verb }; 35 var o: Opts = .{ .verb = verb };
35 var i: usize = 2; 36 var i: usize = 2;
37 // Everything after a bare `--` is the positional argument, whatever it
38 // looks like. Agents send byte-strings for their own reasons, and
39 // `muxa send -- '-n foo\n'` must reach the pty rather than be read as
40 // a flag this binary does not have.
41 var end_of_flags = false;
36 while (i < args.len) : (i += 1) { 42 while (i < args.len) : (i += 1) {
37 const a = args[i]; 43 const a = args[i];
38 if (std.mem.eql(u8, a, "--sock")) { 44 if (end_of_flags) {
45 if (o.arg != null) return null;
46 o.arg = a;
47 } else if (std.mem.eql(u8, a, "--")) {
48 end_of_flags = true;
49 } else if (std.mem.eql(u8, a, "--sock")) {
39 i += 1; 50 i += 1;
40 if (i >= args.len) return null; 51 if (i >= args.len) return null;
41 o.sock = args[i]; 52 o.sock = args[i];
@@ -84,10 +95,15 @@ fn decodeEscapes(alloc: std.mem.Allocator, s: []const u8) ![]u8 {
84 errdefer out.deinit(alloc); 95 errdefer out.deinit(alloc);
85 var i: usize = 0; 96 var i: usize = 0;
86 while (i < s.len) : (i += 1) { 97 while (i < s.len) : (i += 1) {
87 if (s[i] != '\\' or i + 1 >= s.len) { 98 if (s[i] != '\\') {
88 try out.append(alloc, s[i]); 99 try out.append(alloc, s[i]);
89 continue; 100 continue;
90 } 101 }
102 // A backslash with nothing after it is an unfinished escape, and it
103 // is refused like any other one we cannot read (\q). Passing it
104 // through as a literal would be the single case where a typo in an
105 // escape reaches the pty instead of being reported.
106 if (i + 1 >= s.len) return error.BadEscape;
91 i += 1; 107 i += 1;
92 switch (s[i]) { 108 switch (s[i]) {
93 'n' => try out.append(alloc, '\n'), 109 'n' => try out.append(alloc, '\n'),
@@ -112,6 +128,9 @@ test "decodeEscapes covers the sequences send needs" {
112 defer alloc.free(got); 128 defer alloc.free(got);
113 try std.testing.expectEqualSlices(u8, "q\n\x1b[A\x03", got); 129 try std.testing.expectEqualSlices(u8, "q\n\x1b[A\x03", got);
114 try std.testing.expectError(error.BadEscape, decodeEscapes(alloc, "\\q")); 130 try std.testing.expectError(error.BadEscape, decodeEscapes(alloc, "\\q"));
131 // A dangling backslash is an escape the caller did not finish writing,
132 // and it is refused rather than passed through as a literal.
133 try std.testing.expectError(error.BadEscape, decodeEscapes(alloc, "ok\\"));
115 } 134 }
116 135
117 test "parseArgs verbs and flags" { 136 test "parseArgs verbs and flags" {
@@ -125,8 +144,34 @@ test "parseArgs verbs and flags" {
125 try std.testing.expectEqual(@as(?Opts, null), parseArgs(&a3)); 144 try std.testing.expectEqual(@as(?Opts, null), parseArgs(&a3));
126 } 145 }
127 146
147 test "parseArgs: -- hands the rest to the verb, flags and all" {
148 // Without the end-of-flags marker this is an unknown flag and the whole
149 // invocation is refused — the exact shape an agent sends when a key
150 // sequence starts with a dash.
151 const dashed = [_][:0]const u8{ "muxa", "send", "-n foo" };
152 try std.testing.expectEqual(@as(?Opts, null), parseArgs(&dashed));
153
154 const a = [_][:0]const u8{ "muxa", "send", "--settle", "50", "--", "-n foo" };
155 const o = parseArgs(&a).?;
156 try std.testing.expectEqual(@as(u32, 50), o.settle_ms);
157 try std.testing.expectEqualStrings("-n foo", o.arg.?);
158
159 // Past the marker, a flag spelling is just text — and a second
160 // positional is still one too many.
161 const flagish = [_][:0]const u8{ "muxa", "run", "--", "--timeout" };
162 try std.testing.expectEqualStrings("--timeout", parseArgs(&flagish).?.arg.?);
163 const two = [_][:0]const u8{ "muxa", "run", "--", "a", "b" };
164 try std.testing.expectEqual(@as(?Opts, null), parseArgs(&two));
165 }
166
128 const Conn = struct { 167 const Conn = struct {
129 fd: std.posix.fd_t, 168 fd: std.posix.fd_t,
169 /// The code from the `exit_status` frame that ended a wait, set the
170 /// moment awaitFrame returns error.SessionExited. The frame is the
171 /// session's last word and carries the only copy of the code, so it is
172 /// captured here rather than thrown away with the frame; callers read
173 /// it to turn the error into an answer.
174 session_exit: ?u8 = null,
130 175
131 fn open(sock_path: []const u8) !Conn { 176 fn open(sock_path: []const u8) !Conn {
132 const s = try std.net.connectUnixSocket(sock_path); 177 const s = try std.net.connectUnixSocket(sock_path);
@@ -144,6 +189,18 @@ const Conn = struct {
144 /// Read frames until one of type `want` arrives (snapshots, deltas and 189 /// Read frames until one of type `want` arrives (snapshots, deltas and
145 /// pushes stream past an attached client; skip what we did not ask 190 /// pushes stream past an attached client; skip what we did not ask
146 /// for). Bounded by `deadline_ms` wall time via poll. 191 /// for). Bounded by `deadline_ms` wall time via poll.
192 ///
193 /// `exit_status` is the one skipped frame that ends the wait instead:
194 /// the reply we are waiting for is never coming, and the reason is an
195 /// answer — the session ran its last command — not a transport
196 /// failure. Callers get error.SessionExited plus `session_exit`.
197 ///
198 /// Debt: only the WAIT is deadline-bounded, not the read. Once poll
199 /// says a frame has begun, readFrame's readExact blocks until the whole
200 /// payload lands, so a peer that stalls mid-frame outlives the
201 /// deadline. Harmless over a local socket where the daemon writes whole
202 /// frames at once; it becomes real when a network is under this (QUIC,
203 /// Task 10) and wants a nonblocking fd with a partial-frame buffer.
147 fn awaitFrame( 204 fn awaitFrame(
148 self: *Conn, 205 self: *Conn,
149 alloc: std.mem.Allocator, 206 alloc: std.mem.Allocator,
@@ -160,11 +217,46 @@ const Conn = struct {
160 if (n == 0) continue; 217 if (n == 0) continue;
161 const frame = try proto.readFrame(alloc, self.fd) orelse return error.DaemonGone; 218 const frame = try proto.readFrame(alloc, self.fd) orelse return error.DaemonGone;
162 if (frame.type == want) return frame; 219 if (frame.type == want) return frame;
163 frame.deinit(alloc); 220 defer frame.deinit(alloc);
221 if (frame.type == .exit_status) {
222 // A daemon that spelled the frame without a code still ends
223 // the session; null is the honest code, not 0.
224 self.session_exit = if (frame.payload.len >= 1) frame.payload[0] else null;
225 return error.SessionExited;
226 }
164 } 227 }
165 } 228 }
166 }; 229 };
167 230
231 test "awaitFrame ends a wait on exit_status, keeping the code" {
232 const alloc = std.testing.allocator;
233 // A pipe stands in for the daemon: awaitFrame polls and reads an fd and
234 // asks nothing else of it.
235 const pipe = try std.posix.pipe();
236 defer std.posix.close(pipe[0]);
237 defer std.posix.close(pipe[1]);
238
239 var conn = Conn{ .fd = pipe[0] };
240 // A push to skip on the way, then the session's last word. The reply
241 // this wait asked for is never coming, and the code is the answer.
242 try proto.writeFrame(pipe[1], .pty_mode, &[_]u8{0});
243 try proto.writeFrame(pipe[1], .exit_status, &[_]u8{5});
244 try std.testing.expectError(
245 error.SessionExited,
246 conn.awaitFrame(alloc, .status_reply, std.time.milliTimestamp() + 2000),
247 );
248 try std.testing.expectEqual(@as(?u8, 5), conn.session_exit);
249
250 // ...and it is spelled as a session ending, not as a command's code.
251 var out: std.ArrayList(u8) = .empty;
252 defer out.deinit(alloc);
253 try printSessionEnded(out.writer(alloc), conn.session_exit, 42);
254 try std.testing.expectEqualStrings(
255 "{\"reason\":\"session_ended\",\"exit_code\":5,\"duration_ms\":42}\n",
256 out.items,
257 );
258 }
259
168 /// Every failure exit goes through here, so stdout carries one JSON object 260 /// 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 261 /// whatever went wrong — a driving agent parses the same shape on both
170 /// paths instead of switching on exit code first. 262 /// paths instead of switching on exit code first.
@@ -188,11 +280,28 @@ fn writeError(writer: anytype, msg: []const u8, detail: []const u8) !void {
188 try writer.writeAll("}\n"); 280 try writer.writeAll("}\n");
189 } 281 }
190 282
191 fn defaultSockPath(alloc: std.mem.Allocator) ![]const u8 { 283 /// The wall-clock instant a round trip gives up at. `--timeout 0` means
192 if (std.posix.getenv("XDG_RUNTIME_DIR")) |dir| { 284 /// "no bound at all" everywhere else in this protocol (AwaitReq spells it
193 return std.fmt.allocPrint(alloc, "{s}/muxd.sock", .{dir}); 285 /// out), so it means that here too — the alternative reading, a deadline
194 } 286 /// already in the past, would make `--timeout 0` fail instantly instead of
195 return std.fmt.allocPrint(alloc, "/tmp/muxd-{d}.sock", .{std.os.linux.getuid()}); 287 /// waiting forever, which is the opposite of what it asks for.
288 fn deadlineFor(timeout_ms: u32) i64 {
289 if (timeout_ms == 0) return std.math.maxInt(i64);
290 return std.time.milliTimestamp() + timeout_ms;
291 }
292
293 /// The session ended under us: JSON, like every other outcome, but on the
294 /// failure path — the verb that asked (status, capture, send) has no answer
295 /// to give. `run` and `await` do have one and print it themselves.
296 fn failSessionEnded(code: ?u8) u8 {
297 var buf: [128]u8 = undefined;
298 var fbs = std.io.fixedBufferStream(&buf);
299 const writer = fbs.writer();
300 writer.writeAll("{\"error\":\"session ended\",\"exit_code\":") catch return 1;
301 writeExitCode(writer, code) catch return 1;
302 writer.writeAll("}\n") catch return 1;
303 proto.writeAllFd(std.posix.STDOUT_FILENO, fbs.getWritten()) catch {};
304 return 1;
196 } 305 }
197 306
198 pub fn main() !u8 { 307 pub fn main() !u8 {
@@ -208,27 +317,48 @@ pub fn main() !u8 {
208 return 2; 317 return 2;
209 }; 318 };
210 319
211 const sock_path = if (o.sock) |s| s else try defaultSockPath(alloc); 320 const sock_path = if (o.sock) |s| s else try sockpath.defaultSockPath(alloc);
321
322 // Refused by name, before connecting: connect would bounce a too-long
323 // path off the kernel with a generic error, and the path is the whole
324 // story. Every binary owes this check in its own words (sockpath).
325 if (sock_path.len > sockpath.max_sun_path) {
326 var buf: [64]u8 = undefined;
327 const detail = std.fmt.bufPrint(
328 &buf,
329 "{d} bytes, max {d}",
330 .{ sock_path.len, sockpath.max_sun_path },
331 ) catch "too long";
332 return fail("socket path too long", detail);
333 }
212 334
213 var conn = Conn.open(sock_path) catch |e| { 335 var conn = Conn.open(sock_path) catch |e| {
214 return fail("cannot connect to the daemon", @errorName(e)); 336 // The path goes in the detail: a muxa pointed at the wrong socket
337 // is this binary's likeliest field failure, and an agent reading
338 // "FileNotFound" alone cannot tell which path it was that missed.
339 var buf: [256]u8 = undefined;
340 const detail = std.fmt.bufPrint(&buf, "{s}: {s}", .{ sock_path, @errorName(e) }) catch
341 @errorName(e);
342 return fail("cannot connect to the daemon", detail);
215 }; 343 };
216 defer conn.close(); 344 defer conn.close();
217 345
218 const deadline = std.time.milliTimestamp() + o.timeout_ms; 346 const deadline = deadlineFor(o.timeout_ms);
219 return switch (o.verb) { 347 return switch (o.verb) {
220 .status => verbStatus(alloc, &conn, deadline), 348 .status => verbStatus(alloc, &conn, deadline),
221 .capture => verbCapture(alloc, &conn, o.vt, deadline), 349 .capture => verbCapture(alloc, &conn, o.vt, deadline),
222 .send => verbSend(alloc, &conn, o.arg, deadline), 350 .send => verbSend(alloc, &conn, o.arg, deadline),
223 .run => fail("run: not implemented yet", ""), 351 .run => verbRun(alloc, &conn, o, deadline),
224 .@"await" => fail("await: not implemented yet", ""), 352 .@"await" => verbAwait(alloc, &conn, o, deadline),
225 }; 353 };
226 } 354 }
227 355
228 fn verbStatus(alloc: std.mem.Allocator, conn: *Conn, deadline: i64) !u8 { 356 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)); 357 conn.sendFrame(.status_req, "") catch |e| return fail("status: send failed", @errorName(e));
230 const frame = conn.awaitFrame(alloc, .status_reply, deadline) catch |e| 358 const frame = conn.awaitFrame(alloc, .status_reply, deadline) catch |e| switch (e) {
231 return fail("status: no reply", @errorName(e)); 359 error.SessionExited => return failSessionEnded(conn.session_exit),
360 else => return fail("status: no reply", @errorName(e)),
361 };
232 defer frame.deinit(alloc); 362 defer frame.deinit(alloc);
233 const st = proto.decodeStatusReply(frame.payload) catch |e| 363 const st = proto.decodeStatusReply(frame.payload) catch |e|
234 return fail("status: bad reply", @errorName(e)); 364 return fail("status: bad reply", @errorName(e));
@@ -240,6 +370,17 @@ fn verbStatus(alloc: std.mem.Allocator, conn: *Conn, deadline: i64) !u8 {
240 return 0; 370 return 0;
241 } 371 }
242 372
373 /// A command that has not returned — or one whose mechanism cannot know a
374 /// code — has no exit code, and JSON null is the honest spelling: 0 would
375 /// read as "succeeded".
376 fn writeExitCode(writer: anytype, code: ?u8) !void {
377 if (code) |c| {
378 try writer.print("{d}", .{c});
379 } else {
380 try writer.writeAll("null");
381 }
382 }
383
243 fn printStatus(writer: anytype, st: proto.StatusReply) !void { 384 fn printStatus(writer: anytype, st: proto.StatusReply) !void {
244 try writer.print( 385 try writer.print(
245 "{{\"cols\":{d},\"rows\":{d},\"cursor\":{{\"x\":{d},\"y\":{d}}}," ++ 386 "{{\"cols\":{d},\"rows\":{d},\"cursor\":{{\"x\":{d},\"y\":{d}}}," ++
@@ -250,13 +391,7 @@ fn printStatus(writer: anytype, st: proto.StatusReply) !void {
250 try writer.writeAll(",\"mechanism\":"); 391 try writer.writeAll(",\"mechanism\":");
251 try jsonEscape(writer, @tagName(st.cmd.mechanism)); 392 try jsonEscape(writer, @tagName(st.cmd.mechanism));
252 try writer.writeAll(",\"exit_code\":"); 393 try writer.writeAll(",\"exit_code\":");
253 // A command that has not returned has no exit code, and JSON null is 394 try writeExitCode(writer, st.cmd.exit_code);
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( 395 try writer.print(
261 ",\"start_row\":{d},\"end_row\":{d},\"seq\":{d}}}}}\n", 396 ",\"start_row\":{d},\"end_row\":{d},\"seq\":{d}}}}}\n",
262 .{ st.cmd.start_row, st.cmd.end_row, st.cmd.seq }, 397 .{ st.cmd.start_row, st.cmd.end_row, st.cmd.seq },
@@ -285,8 +420,10 @@ test "printStatus spells a pending exit code as JSON null" {
285 fn verbCapture(alloc: std.mem.Allocator, conn: *Conn, vt: bool, deadline: i64) !u8 { 420 fn verbCapture(alloc: std.mem.Allocator, conn: *Conn, vt: bool, deadline: i64) !u8 {
286 const payload = [_]u8{if (vt) 1 else 0}; 421 const payload = [_]u8{if (vt) 1 else 0};
287 conn.sendFrame(.debug_dump, &payload) catch |e| return fail("capture: send failed", @errorName(e)); 422 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| 423 const frame = conn.awaitFrame(alloc, .dump_reply, deadline) catch |e| switch (e) {
289 return fail("capture: no reply", @errorName(e)); 424 error.SessionExited => return failSessionEnded(conn.session_exit),
425 else => return fail("capture: no reply", @errorName(e)),
426 };
290 defer frame.deinit(alloc); 427 defer frame.deinit(alloc);
291 428
292 var out: std.ArrayList(u8) = .empty; 429 var out: std.ArrayList(u8) = .empty;
@@ -328,8 +465,14 @@ fn verbSend(alloc: std.mem.Allocator, conn: *Conn, arg: ?[]const u8, deadline: i
328 // for that flush to succeed. Nothing is done with the reply — its 465 // for that flush to succeed. Nothing is done with the reply — its
329 // arrival is the whole content. 466 // arrival is the whole content.
330 conn.sendFrame(.status_req, "") catch |e| return fail("send: ack request failed", @errorName(e)); 467 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| 468 const ack = conn.awaitFrame(alloc, .status_reply, deadline) catch |e| switch (e) {
332 return fail("send: daemon never acknowledged the input", @errorName(e)); 469 // The bytes we sent ended the session (`exit\n`). Reported as the
470 // session's death rather than as "sent", because this verb's answer
471 // is about the send and there is no longer a session to have sent
472 // to — an agent that wants the death to be an ANSWER runs `run`.
473 error.SessionExited => return failSessionEnded(conn.session_exit),
474 else => return fail("send: daemon never acknowledged the input", @errorName(e)),
475 };
333 ack.deinit(alloc); 476 ack.deinit(alloc);
334 477
335 conn.sendFrame(.detach, "") catch |e| return fail("send: detach failed", @errorName(e)); 478 conn.sendFrame(.detach, "") catch |e| return fail("send: detach failed", @errorName(e));
@@ -337,3 +480,294 @@ fn verbSend(alloc: std.mem.Allocator, conn: *Conn, arg: ?[]const u8, deadline: i
337 proto.writeAllFd(std.posix.STDOUT_FILENO, "{\"sent\":true}\n") catch {}; 480 proto.writeAllFd(std.posix.STDOUT_FILENO, "{\"sent\":true}\n") catch {};
338 return 0; 481 return 0;
339 } 482 }
483
484 /// How much longer than the daemon this client is willing to wait.
485 ///
486 /// Load-bearing: the daemon starts its own `timeout_ms` window when it
487 /// READS the await_req, which is already later than the instant this
488 /// process started counting. Waiting exactly `timeout_ms` here would lose
489 /// that race every single time, and every timeout would surface as
490 /// `{"error":"await: no reply"}` instead of the structured
491 /// `{"reason":"timeout"}` with exit 3 that the agent is meant to read.
492 const await_grace_ms = 2_000;
493
494 /// The span fetch gets its own window rather than the tail of the run's: a
495 /// command that returned in the last millisecond of `--timeout` still has a
496 /// transcript worth having, and this round trip is a local read that either
497 /// answers promptly or is not coming.
498 const span_fetch_ms = 2_000;
499
500 /// Ask to be told when the session next comes to rest, and wait for it.
501 fn doAwait(
502 alloc: std.mem.Allocator,
503 conn: *Conn,
504 o: Opts,
505 since_seq: u64,
506 deadline: i64,
507 ) !proto.AwaitReply {
508 try conn.sendFrame(.await_req, &proto.encodeAwaitReq(.{
509 .since_seq = since_seq,
510 .settle_ms = o.settle_ms,
511 .timeout_ms = o.timeout_ms,
512 }));
513 const frame = try conn.awaitFrame(alloc, .await_reply, deadline);
514 defer frame.deinit(alloc);
515 return try proto.decodeAwaitReply(frame.payload);
516 }
517
518 /// The session's RETURN WATERMARK: the seq of the last command return, 0 if
519 /// none. Handed straight to `since_seq`, where it means "only a return
520 /// newer than this may answer me".
521 fn currentSeq(alloc: std.mem.Allocator, conn: *Conn, deadline: i64) !u64 {
522 try conn.sendFrame(.status_req, "");
523 const frame = try conn.awaitFrame(alloc, .status_reply, deadline);
524 defer frame.deinit(alloc);
525 const s = try proto.decodeStatusReply(frame.payload);
526 return s.cmd.seq;
527 }
528
529 /// Strip the styling out of scrollback rows: an agent reading `output`
530 /// wants what the command printed, not how it was coloured.
531 ///
532 /// CSI (ESC [ … final byte) and OSC (ESC ] … BEL or ST) go, as does any
533 /// other two-byte escape; text and newlines stay. Deliberately not a VT
534 /// parser — these rows come from our own formatter, which emits SGR and
535 /// nothing more exotic. Caller frees.
536 fn stripSgr(alloc: std.mem.Allocator, s: []const u8) ![]u8 {
537 var out: std.ArrayList(u8) = .empty;
538 errdefer out.deinit(alloc);
539 var i: usize = 0;
540 while (i < s.len) {
541 if (s[i] != 0x1b or i + 1 >= s.len) {
542 try out.append(alloc, s[i]);
543 i += 1;
544 continue;
545 }
546 switch (s[i + 1]) {
547 '[' => {
548 i += 2;
549 // Parameter and intermediate bytes, then one final byte in
550 // 0x40..0x7e that ends the sequence.
551 while (i < s.len and (s[i] < 0x40 or s[i] > 0x7e)) i += 1;
552 if (i < s.len) i += 1;
553 },
554 ']' => {
555 i += 2;
556 while (i < s.len) : (i += 1) {
557 if (s[i] == 0x07) {
558 i += 1;
559 break;
560 }
561 if (s[i] == 0x1b and i + 1 < s.len and s[i + 1] == '\\') {
562 i += 2;
563 break;
564 }
565 }
566 },
567 // ESC 7, ESC M and friends: two bytes, both dropped.
568 else => i += 2,
569 }
570 }
571 return out.toOwnedSlice(alloc);
572 }
573
574 test "stripSgr leaves text, drops SGR and OSC" {
575 const alloc = std.testing.allocator;
576 const got = try stripSgr(alloc, "\x1b[0m\x1b[1;31mred\x1b[0m ok\n\x1b]0;title\x07plain");
577 defer alloc.free(got);
578 try std.testing.expectEqualStrings("red ok\nplain", got);
579 }
580
581 /// The rows a command occupied, as plain text. `end_row` is the row the D
582 /// mark landed on — the prompt redraw — so the span is [start_row, end_row)
583 /// and an end at or before the start is simply no output.
584 ///
585 /// Rows are absolute screen rows and best-effort by construction (see
586 /// MarkEvent.row): the alt screen and scrollback pruning can invalidate
587 /// them between the reply and this fetch. Every failure mode here is
588 /// therefore a null output, never a failed run — the exit code is the
589 /// answer, and the transcript is the bonus.
590 fn fetchSpan(
591 alloc: std.mem.Allocator,
592 conn: *Conn,
593 start_row: u32,
594 end_row: u32,
595 deadline: i64,
596 ) !?[]u8 {
597 if (end_row <= start_row) return null;
598 const count: u16 = @intCast(@min(end_row - start_row, std.math.maxInt(u16)));
599 try conn.sendFrame(.fetch_scrollback, &proto.encodeScrollbackReq(start_row, count));
600 const frame = try conn.awaitFrame(alloc, .scrollback_chunk, deadline);
601 defer frame.deinit(alloc);
602 // The chunk leads with the request it answers; the rows follow.
603 if (frame.payload.len <= 6) return null;
604 return try stripSgr(alloc, frame.payload[6..]);
605 }
606
607 /// One JSON object: what ended the wait, what the session's command state
608 /// was when it ended, and how long we waited. `output` is present only when
609 /// there is a transcript to give — an absent key and an empty string are
610 /// different answers.
611 fn printAwaitReply(
612 writer: anytype,
613 r: proto.AwaitReply,
614 output: ?[]const u8,
615 duration_ms: i64,
616 ) !void {
617 try writer.writeAll("{\"reason\":");
618 try jsonEscape(writer, @tagName(r.reason));
619 try writer.writeAll(",\"phase\":");
620 try jsonEscape(writer, @tagName(r.state.phase));
621 try writer.writeAll(",\"mechanism\":");
622 try jsonEscape(writer, @tagName(r.state.mechanism));
623 try writer.writeAll(",\"exit_code\":");
624 try writeExitCode(writer, r.state.exit_code);
625 try writer.print(
626 ",\"start_row\":{d},\"end_row\":{d},\"duration_ms\":{d}",
627 .{ r.state.start_row, r.state.end_row, duration_ms },
628 );
629 if (output) |text| {
630 try writer.writeAll(",\"output\":");
631 try jsonEscape(writer, text);
632 }
633 try writer.writeAll("}\n");
634 }
635
636 test "printAwaitReply omits output when there is none and spells a missing code null" {
637 const alloc = std.testing.allocator;
638 const r: proto.AwaitReply = .{
639 .state = .{
640 .phase = .returned,
641 .mechanism = .settle,
642 .exit_code = null,
643 .start_row = 3,
644 .end_row = 9,
645 .seq = 12,
646 },
647 .reason = .settled,
648 };
649
650 var bare: std.ArrayList(u8) = .empty;
651 defer bare.deinit(alloc);
652 try printAwaitReply(bare.writer(alloc), r, null, 250);
653 try std.testing.expectEqualStrings(
654 "{\"reason\":\"settled\",\"phase\":\"returned\",\"mechanism\":\"settle\"," ++
655 "\"exit_code\":null,\"start_row\":3,\"end_row\":9,\"duration_ms\":250}\n",
656 bare.items,
657 );
658
659 var with: std.ArrayList(u8) = .empty;
660 defer with.deinit(alloc);
661 try printAwaitReply(with.writer(alloc), r, "a\nb", 250);
662 try std.testing.expect(std.mem.indexOf(u8, with.items, "\"output\":\"a\\nb\"") != null);
663 }
664
665 /// The session ran its last command. An ANSWER for `run` and `await` — the
666 /// command is over and this is how — so it prints on stdout and exits 0,
667 /// unlike the other verbs, which have nothing to report and fail.
668 fn printSessionEnded(writer: anytype, code: ?u8, duration_ms: i64) !void {
669 try writer.writeAll("{\"reason\":\"session_ended\",\"exit_code\":");
670 try writeExitCode(writer, code);
671 try writer.print(",\"duration_ms\":{d}}}\n", .{duration_ms});
672 }
673
674 /// Print an await outcome and choose the exit code for it. A timeout is the
675 /// only nonzero one: the agent asked a question and got "still running",
676 /// which is a distinct thing to branch on, while `returned` and `settled`
677 /// are both answers — including a command that returned nonzero, whose
678 /// failure is in `exit_code`, not in muxa's.
679 fn reportAwait(
680 alloc: std.mem.Allocator,
681 r: proto.AwaitReply,
682 output: ?[]const u8,
683 duration_ms: i64,
684 ) !u8 {
685 var out: std.ArrayList(u8) = .empty;
686 defer out.deinit(alloc);
687 try printAwaitReply(out.writer(alloc), r, output, duration_ms);
688 proto.writeAllFd(std.posix.STDOUT_FILENO, out.items) catch {};
689 return if (r.reason == .timeout) 3 else 0;
690 }
691
692 fn reportSessionEnded(alloc: std.mem.Allocator, code: ?u8, duration_ms: i64) !u8 {
693 var out: std.ArrayList(u8) = .empty;
694 defer out.deinit(alloc);
695 try printSessionEnded(out.writer(alloc), code, duration_ms);
696 proto.writeAllFd(std.posix.STDOUT_FILENO, out.items) catch {};
697 return 0;
698 }
699
700 fn verbAwait(alloc: std.mem.Allocator, conn: *Conn, o: Opts, deadline: i64) !u8 {
701 const started = std.time.milliTimestamp();
702 attachZero(conn) catch |e| return fail("await: attach failed", @errorName(e));
703 const since = currentSeq(alloc, conn, deadline) catch |e| switch (e) {
704 error.SessionExited => return reportSessionEnded(alloc, conn.session_exit, elapsed(started)),
705 else => return fail("await: status failed", @errorName(e)),
706 };
707 const r = doAwait(alloc, conn, o, since, awaitDeadline(o)) catch |e| switch (e) {
708 error.SessionExited => return reportSessionEnded(alloc, conn.session_exit, elapsed(started)),
709 else => return fail("await: no reply", @errorName(e)),
710 };
711 return reportAwait(alloc, r, null, elapsed(started));
712 }
713
714 fn verbRun(alloc: std.mem.Allocator, conn: *Conn, o: Opts, deadline: i64) !u8 {
715 const cmdline = o.arg orelse return fail("run: needs CMDLINE", "");
716 const started = std.time.milliTimestamp();
717
718 attachZero(conn) catch |e| return fail("run: attach failed", @errorName(e));
719
720 // BEFORE the input, not after: the watermark has to be the one this
721 // command must beat. Read afterwards, a command fast enough to return
722 // between the two would have already moved the seq past a value we
723 // never recorded, and the await would sit waiting for a return that
724 // had happened.
725 const since = currentSeq(alloc, conn, deadline) catch |e| switch (e) {
726 error.SessionExited => return reportSessionEnded(alloc, conn.session_exit, elapsed(started)),
727 else => return fail("run: status failed", @errorName(e)),
728 };
729
730 // The cmdline goes to the pty verbatim — escapes are `send`'s business
731 // — plus the newline that submits it. No ack round-trip is needed the
732 // way `send` needs one: the await_req that follows is itself the read
733 // that proves the daemon got past this frame, and this process stays
734 // connected until the reply lands.
735 const line = std.fmt.allocPrint(alloc, "{s}\n", .{cmdline}) catch |e|
736 return fail("run: cannot build the command line", @errorName(e));
737 defer alloc.free(line);
738 conn.sendFrame(.input, line) catch |e| return fail("run: input failed", @errorName(e));
739
740 const r = doAwait(alloc, conn, o, since, awaitDeadline(o)) catch |e| switch (e) {
741 error.SessionExited => return reportSessionEnded(alloc, conn.session_exit, elapsed(started)),
742 else => return fail("run: no reply", @errorName(e)),
743 };
744
745 // Only the marks regime knows where the command's rows are; pgid and
746 // settle answer WHEN, never WHERE, and a span from them would be a
747 // guess dressed as a transcript.
748 var output: ?[]u8 = null;
749 defer if (output) |text| alloc.free(text);
750 if (r.state.mechanism == .marks and r.reason == .returned) {
751 output = fetchSpan(
752 alloc,
753 conn,
754 r.state.start_row,
755 r.state.end_row,
756 @max(deadline, deadlineFor(span_fetch_ms)),
757 ) catch null;
758 }
759
760 return reportAwait(alloc, r, output, elapsed(started));
761 }
762
763 fn elapsed(started: i64) i64 {
764 return std.time.milliTimestamp() - started;
765 }
766
767 /// This client's deadline for the await itself — the daemon's own bound
768 /// plus the grace window (see await_grace_ms). An unbounded request stays
769 /// unbounded here too.
770 fn awaitDeadline(o: Opts) i64 {
771 if (o.timeout_ms == 0) return std.math.maxInt(i64);
772 return std.time.milliTimestamp() + o.timeout_ms + await_grace_ms;
773 }
src/sockpath.zig
Old New
@@ -15,6 +15,17 @@ const std = @import("std");
15 /// the kernel's and belongs in one place, the wording is theirs. 15 /// the kernel's and belongs in one place, the wording is theirs.
16 pub const max_sun_path = 107; 16 pub const max_sun_path = 107;
17 17
18 /// Where a binary looks when nobody named a socket. The default path is
19 /// part of a socket path's identity too: it is what makes two binaries
20 /// started with no `--sock` land on the SAME daemon, so it lives here with
21 /// the bound rather than once per binary.
22 pub fn defaultSockPath(alloc: std.mem.Allocator) ![]const u8 {
23 if (std.posix.getenv("XDG_RUNTIME_DIR")) |dir| {
24 return std.fmt.allocPrint(alloc, "{s}/muxd.sock", .{dir});
25 }
26 return std.fmt.allocPrint(alloc, "/tmp/muxd-{d}.sock", .{std.os.linux.getuid()});
27 }
28
18 /// A socket file's identity at the moment it was bound, so teardown can 29 /// A socket file's identity at the moment it was bound, so teardown can
19 /// tell our socket from one that replaced it. 30 /// tell our socket from one that replaced it.
20 /// 31 ///