src/cli/muxa.zig
Ref: Size: 68.9 KiB History
//! `mux a`: the agent-facing mode. Every verb prints one JSON object on stdout;
//! failures print `{"error": "..."}` and exit nonzero. Attachments always use
//! 0x0 so automation cannot replace the human client's terminal dimensions.
//!
//! Exit codes are 0 for a completed request, 1 for a JSON error, 2 for invalid
//! arguments, 3 for timeout, and 4 when no JSON object could be written. A
//! session command's status appears in the `exit_code` field and does not become
//! this process's exit code.
const std = @import("std");
const proto = @import("term").protocol;
const grid = @import("term").grid;
const sockpath = @import("sockpath");
const quic = @import("quic");
const xdg = @import("xdg");
const cliflags = @import("cliflags");
const dial = @import("dial");
const link_mod = @import("link");
const build_options = @import("build_options");
const usage =
\\usage: mux a <verb> [--sock PATH | --quic HOST[:PORT] [--key PATH]]
\\ [--settle MS] [--timeout MS] [--vt] [--session NAME] [args]
\\ mux a --help | --version
\\NAME must already exist: `mux a` attaches at 0x0 and never creates a session
\\(`capture` is the exception that stays quiet: it answers in the grid).
\\verbs:
\\ status session snapshot as JSON
\\ capture current grid as text (--vt for styled)
\\ send BYTES raw bytes to the pty (C-style escapes: \n \r \t \e \xNN)
\\ run CMDLINE send CMDLINE + newline, await return, report exit/output
\\ await wait for the current/next command to return
\\
;
const AgentVerb = enum { status, capture, send, run, await };
const AgentArguments = struct {
sock: ?[]const u8 = null,
/// Remote daemon QUIC endpoint. Every verb uses the same frames and JSON
/// shape over local and QUIC transports.
quic: ?[]const u8 = null,
/// Explicit QUIC key path. Null still allows `MUX_KEY_FILE` and the XDG
/// default to be resolved after parsing.
key: ?[]const u8 = null,
settle: u32 = 0,
// The daemon interprets zero as an unbounded await, so the client default
// must remain nonzero.
timeout: u32 = 30_000,
vt: bool = false,
/// Validate only an explicitly supplied name. `sessionName` converts null
/// to the wire's empty-string encoding for the default session.
session: ?proto.SessionName = null,
/// Null until `positional` meets a verb; a returned AgentArguments has one.
_verb: ?AgentVerb = null,
_arg: ?[]const u8 = null,
pub fn sessionName(o: AgentArguments) []const u8 {
// Use the same name for attachment and every subsequent request so the
// daemon's attached-tail equality check cannot see a mismatch.
return if (o.session) |n| n.name else "";
}
/// Parse the first positional word as a verb and the next as its optional
/// argument. Reject unknown verbs and additional positional arguments.
pub fn positional(self: *AgentArguments, word: []const u8) bool {
if (self._verb == null) {
self._verb = std.meta.stringToEnum(AgentVerb, word) orelse return false;
return true;
}
if (self._arg != null) return false;
self._arg = word;
return true;
}
};
comptime {
cliflags.assertDocumented(AgentArguments, usage, &.{});
}
fn parseArgs(args: []const [:0]const u8) cliflags.ParseError!AgentArguments {
var o: AgentArguments = .{};
try cliflags.parseStrict(AgentArguments, &o, args[1..]);
if (o._verb == null) return error.Usage;
// Require one transport. Silently preferring QUIC over an explicit socket
// would send requests to a different daemon than the caller named.
if (o.quic != null and o.sock != null) return error.Usage;
// A key without QUIC has no transport to authenticate and is always invalid.
if (o.key != null and o.quic == null) return error.Usage;
return o;
}
/// JSON string escape, the six mandatory escapes + control bytes as \u00XX.
fn jsonEscape(writer: anytype, s: []const u8) !void {
try writer.writeByte('"');
for (s) |b| switch (b) {
'"' => try writer.writeAll("\\\""),
'\\' => try writer.writeAll("\\\\"),
'\n' => try writer.writeAll("\\n"),
'\r' => try writer.writeAll("\\r"),
'\t' => try writer.writeAll("\\t"),
0x00...0x08, 0x0b, 0x0c, 0x0e...0x1f => try writer.print("\\u{x:0>4}", .{b}),
else => try writer.writeByte(b),
};
try writer.writeByte('"');
}
test "jsonEscape pins the escapes" {
var buf: [128]u8 = undefined;
var fbs = std.io.fixedBufferStream(&buf);
try jsonEscape(fbs.writer(), "a\"b\\c\nd\x1be");
try std.testing.expectEqualStrings("\"a\\\"b\\\\c\\nd\\u001be\"", fbs.getWritten());
}
/// Decode C-style escapes for `send`. Caller frees.
fn decodeEscapes(alloc: std.mem.Allocator, s: []const u8) ![]u8 {
var out: std.ArrayList(u8) = .empty;
errdefer out.deinit(alloc);
var i: usize = 0;
while (i < s.len) : (i += 1) {
if (s[i] != '\\') {
try out.append(alloc, s[i]);
continue;
}
// A trailing backslash is an incomplete escape and is rejected like an
// unknown escape such as `\q`.
if (i + 1 >= s.len) return error.BadEscape;
i += 1;
switch (s[i]) {
'n' => try out.append(alloc, '\n'),
'r' => try out.append(alloc, '\r'),
't' => try out.append(alloc, '\t'),
'e' => try out.append(alloc, 0x1b),
'\\' => try out.append(alloc, '\\'),
'x' => {
if (i + 2 >= s.len) return error.BadEscape;
try out.append(alloc, try std.fmt.parseInt(u8, s[i + 1 .. i + 3], 16));
i += 2;
},
else => return error.BadEscape,
}
}
return out.toOwnedSlice(alloc);
}
test "decodeEscapes covers the sequences send needs" {
const alloc = std.testing.allocator;
const got = try decodeEscapes(alloc, "q\\n\\e[A\\x03");
defer alloc.free(got);
try std.testing.expectEqualSlices(u8, "q\n\x1b[A\x03", got);
try std.testing.expectError(error.BadEscape, decodeEscapes(alloc, "\\q"));
// Reject a dangling backslash rather than passing it through literally.
try std.testing.expectError(error.BadEscape, decodeEscapes(alloc, "ok\\"));
}
test "parseArgs verbs and flags" {
const a1 = [_][:0]const u8{ "a", "status" };
try std.testing.expectEqual(AgentVerb.status, (try parseArgs(&a1))._verb.?);
const a2 = [_][:0]const u8{ "a", "run", "--timeout", "5000", "make test" };
const o2 = try parseArgs(&a2);
try std.testing.expectEqual(@as(u32, 5000), o2.timeout);
try std.testing.expectEqualStrings("make test", o2._arg.?);
const a3 = [_][:0]const u8{ "a", "bogus" };
try std.testing.expectError(error.Usage, parseArgs(&a3));
// The verb is positional, so ordinary flags may precede it.
const early = [_][:0]const u8{ "a", "--vt", "capture" };
const oe = try parseArgs(&early);
try std.testing.expectEqual(AgentVerb.capture, oe._verb.?);
try std.testing.expect(oe.vt);
// Only the first positional word is a verb; a later verb-shaped word is the
// command argument.
const shadow = [_][:0]const u8{ "a", "send", "status" };
const os = try parseArgs(&shadow);
try std.testing.expectEqual(AgentVerb.send, os._verb.?);
try std.testing.expectEqualStrings("status", os._arg.?);
// An invocation containing only flags still lacks a required verb.
const verbless = [_][:0]const u8{ "a", "--vt" };
try std.testing.expectError(error.Usage, parseArgs(&verbless));
}
test "parseArgs: -- hands the rest to the verb, flags and all" {
// Without `--`, a dash-prefixed key sequence is an unknown flag.
const dashed = [_][:0]const u8{ "a", "send", "-n foo" };
try std.testing.expectError(error.Usage, parseArgs(&dashed));
const a = [_][:0]const u8{ "a", "send", "--settle", "50", "--", "-n foo" };
const o = try parseArgs(&a);
try std.testing.expectEqual(@as(u32, 50), o.settle);
try std.testing.expectEqualStrings("-n foo", o._arg.?);
// After `--`, flag-shaped words are text, but a third positional argument
// is still invalid.
const flagish = [_][:0]const u8{ "a", "run", "--", "--timeout" };
try std.testing.expectEqualStrings("--timeout", (try parseArgs(&flagish))._arg.?);
const two = [_][:0]const u8{ "a", "run", "--", "a", "b" };
try std.testing.expectError(error.Usage, parseArgs(&two));
// After `--`, literal `--help` reaches the session instead of opening usage.
const help_payload = [_][:0]const u8{ "a", "send", "--", "--help" };
try std.testing.expectEqualStrings("--help", (try parseArgs(&help_payload))._arg.?);
}
test "mux a: --help and --version are answered wherever they can be typed" {
// Help is recognized before or after the verb by the same parser pass.
try std.testing.expectError(error.Help, parseArgs(&[_][:0]const u8{ "a", "--help" }));
try std.testing.expectError(error.Help, parseArgs(&[_][:0]const u8{ "a", "-h" }));
try std.testing.expectError(error.Help, parseArgs(&[_][:0]const u8{ "a", "status", "--help" }));
try std.testing.expectError(error.Help, parseArgs(&[_][:0]const u8{ "a", "run", "--timeout", "--help" }));
try std.testing.expectError(error.Version, parseArgs(&[_][:0]const u8{ "a", "--version" }));
try std.testing.expectError(error.Version, parseArgs(&[_][:0]const u8{ "a", "status", "--version" }));
// Neither is a verb, so a word that is neither and is no verb either is
// still the usage error it always was.
try std.testing.expectError(error.Usage, parseArgs(&[_][:0]const u8{ "a", "--wat" }));
try std.testing.expectError(error.Usage, parseArgs(&[_][:0]const u8{"a"}));
}
test "mux a: --session rides every verb; a bad name is usage, not wire bytes" {
const a = [_][:0]const u8{ "a", "status", "--session", "b" };
const o = try parseArgs(&a);
try std.testing.expectEqualStrings("b", o.sessionName());
// No --session named: the wire's own default spelling, empty.
const bare = [_][:0]const u8{ "a", "status" };
try std.testing.expectEqualStrings("", (try parseArgs(&bare)).sessionName());
// Reject unaddressable session names before sending any protocol frame.
const bad = [_][:0]const u8{ "a", "status", "--session", "has space" };
try std.testing.expectError(error.Usage, parseArgs(&bad));
// Empty is the wire's default, not a name anyone can mean by typing it.
const empty = [_][:0]const u8{ "a", "status", "--session", "" };
try std.testing.expectError(error.Usage, parseArgs(&empty));
}
test "parseArgs: --quic and --key, and the pairs that make no sense" {
const q = [_][:0]const u8{ "a", "status", "--quic", "10.0.0.2:4433" };
const oq = try parseArgs(&q);
try std.testing.expectEqualStrings("10.0.0.2:4433", oq.quic.?);
// A missing explicit key is valid because runtime resolution may use
// `MUX_KEY_FILE` or the XDG default.
try std.testing.expectEqual(@as(?[]const u8, null), oq.key);
const k = [_][:0]const u8{ "a", "run", "--quic", "box:4433", "--key", "/k", "make test" };
const ok = try parseArgs(&k);
try std.testing.expectEqualStrings("box:4433", ok.quic.?);
try std.testing.expectEqualStrings("/k", ok.key.?);
try std.testing.expectEqualStrings("make test", ok._arg.?);
// The shared parser detects missing values and this mode returns usage
// without attempting a connection.
const dangling_q = [_][:0]const u8{ "a", "status", "--quic" };
try std.testing.expectError(error.Usage, parseArgs(&dangling_q));
// Reject two named transports rather than choosing one implicitly.
const both = [_][:0]const u8{ "a", "status", "--sock", "/tmp/s", "--quic", "b:1" };
try std.testing.expectError(error.Usage, parseArgs(&both));
// Reject a key without a QUIC transport.
const lonely_key = [_][:0]const u8{ "a", "status", "--key", "/k" };
try std.testing.expectError(error.Usage, parseArgs(&lonely_key));
// Neither named is the ordinary local case and stays silent.
const neither = [_][:0]const u8{ "a", "status" };
try std.testing.expectEqual(@as(?[]const u8, null), (try parseArgs(&neither)).quic);
}
const AgentConnection = struct {
/// Verbs are transport-blind; `--quic` chooses which arm of the Link.
/// The mechanics — send, await, close — are the Link's; everything else
/// on this struct is muxa's policy.
link: link_mod.Link,
/// The endpoint state required for one reconnect, and the marker for a
/// QUIC connection: `open` leaves it null. Reconnect policy is muxa's,
/// not the Link's — reuse the original address and key so a mid-command
/// redial cannot select rotated credentials or a different resolved
/// address.
redial: ?struct {
addr: std.net.Address,
key: quic.Key,
idle_ms: u32,
/// Duration of the initial handshake, used by `graceMs` as a rough
/// network latency measurement.
connect_ms: i64,
/// Whether the single permitted reconnect has been used.
reconnected: bool = false,
} = null,
/// Allocator for QUIC frame staging and reconnect state. Frame-returning
/// methods accept their result allocator separately.
alloc: std.mem.Allocator,
/// Whether the current attachment has received a snapshot. The daemon uses
/// the same exit frame for a rejected attach and an ended session; a valid
/// attach always sends a snapshot first.
saw_snapshot: bool = false,
/// The code from the `exit_status` frame that ended a wait. That frame is
/// the session's last word and carries the only copy of the code, so it is
/// captured here rather than thrown away with the frame.
session_exit: ?u8 = null,
/// Static error name from a failed reconnect, retained so the final JSON can
/// report more than the initial `ConnectionLost` condition.
reconnect_failure: ?[]const u8 = null,
fn open(alloc: std.mem.Allocator, sock_path: []const u8) !AgentConnection {
const s = try dial.dial(sock_path);
return .{ .link = .{ .fd = s.handle }, .alloc = alloc };
}
/// A `send` before the stream exists takes zero bytes, so the frame
/// would silently never leave.
fn openQuic(
alloc: std.mem.Allocator,
addr: std.net.Address,
key: quic.Key,
idle_ms: u32,
deadline_ms: i64,
) !AgentConnection {
const started = std.time.milliTimestamp();
const cl = try quic.Client.connect(alloc, addr, key, idle_ms);
errdefer cl.deinit();
try waitReady(cl, deadline_ms);
return .{
.link = .{ .quic = .{ .cl = cl, .alloc = alloc } },
.redial = .{
.addr = addr,
.key = key,
.idle_ms = idle_ms,
.connect_ms = elapsed(started),
},
.alloc = alloc,
};
}
fn close(self: *AgentConnection) void {
self.link.close();
}
/// QUIC widens the grace: the daemon's window opens a flight after
/// ours. The cap bounds a slow handshake. A socket has no handshake to
/// measure, and no `redial` either, so it keeps the flat window.
fn graceMs(self: *const AgentConnection) i64 {
const r = self.redial orelse return await_grace_ms;
return @min(grace_cap_ms, @max(await_grace_ms, 4 * r.connect_ms));
}
/// The verbs' one way out. A write that died of a closed peer is the
/// interesting case: the daemon may have refused the attach and gone,
/// in which case the refusal is the answer the agent wants and the
/// write error is only how we found out.
fn sendFrame(self: *AgentConnection, t: proto.MsgType, payload: []const u8, deadline_ms: i64) !void {
self.writeThrough(t, payload, deadline_ms) catch |e| {
switch (e) {
error.BrokenPipe,
error.ConnectionResetByPeer,
error.ConnectionLost,
=> try self.refusalPending(),
else => {},
}
return e;
};
}
/// Drain a pending attach rejection after a write races with the daemon's
/// exit frame and close. Requesting an otherwise unused frame type lets the
/// snapshot/exit classification terminate the wait.
fn refusalPending(self: *AgentConnection) error{AttachRefused}!void {
const frame = self.awaitFrame(.attach, std.time.milliTimestamp() + refusal_drain_ms) catch |e| {
if (e == error.AttachRefused) return error.AttachRefused;
return;
};
frame.deinit(self.alloc);
}
/// Hand the frame to the Link and, on QUIC, stay until the bytes are
/// gone or the wait is spent. A socket write either takes the frame or
/// fails, so there is nothing to drive out there.
fn writeThrough(
self: *AgentConnection,
t: proto.MsgType,
payload: []const u8,
deadline_ms: i64,
) !void {
try self.link.sendFrame(t, payload);
if (self.link != .quic) return;
// The verb's deadline takes precedence over `send_flush_ms`, so a
// verb asked for a 100ms answer cannot spend five seconds sending;
// the flush cap is what bounds `--timeout 0`. Relative at this
// boundary because that is what the Link takes, and floored at zero
// rather than clamped away: a deadline already spent still offers
// the bytes once before `SendStalled`, which is what the loop this
// replaced did with its first send.
const left = deadline_ms - std.time.milliTimestamp();
const bounded: u32 = if (left <= 0) 0 else @intCast(@min(left, send_flush_ms));
return self.link.flushWithin(bounded);
}
/// Skip unrelated snapshots and deltas while waiting for `want`. An exit
/// frame ends the wait as a session outcome rather than a transport error:
/// the daemon uses the same exit frame for a rejected attach and an ended
/// session, and a valid attach always sends a snapshot first.
fn onOther(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
const self: *AgentConnection = @ptrCast(@alignCast(ctx.?));
if (frame.type == .snapshot) self.saw_snapshot = true;
if (frame.type == .exit_status) {
if (!self.saw_snapshot) return error.AttachRefused;
// A missing status byte still ends the session, but its code is
// unknown rather than zero.
self.session_exit = if (frame.payload.len >= 1) frame.payload[0] else null;
return error.SessionExited;
}
}
fn awaitFrame(self: *AgentConnection, want: proto.MsgType, deadline_ms: i64) !proto.Frame {
const left = deadline_ms - std.time.milliTimestamp();
if (left <= 0) return error.Timeout;
// `--timeout 0` spells its deadline as maxInt(i64); the Link takes a
// relative u32, so the wait is capped at what that can hold. Seven
// weeks is unbounded for anything a verb waits on, and the cast
// would panic without it.
const window: u32 = @intCast(@min(left, std.math.maxInt(u32)));
const got = self.link.awaitFrame(self.alloc, want, window, .{
.ctx = self,
.on = onOther,
}) catch |e| switch (e) {
// Wording is per-arm on purpose: a closed unix socket is a dead
// daemon, while a dead QUIC connection cannot distinguish daemon
// exit from path failure, so it reports `ConnectionLost` and
// allows the one reconnect.
error.Closed => return switch (self.link) {
.quic => error.ConnectionLost,
else => error.DaemonGone,
},
// Everything else is itself, `FrameTooLarge` included: a daemon
// that got the framing wrong is a different fact from a daemon
// that went away, and the verb's detail says which.
else => return e,
};
return got orelse error.Timeout;
}
/// `connect_ms` deliberately keeps the FIRST handshake's measurement:
/// what a reader wants is the distance to the daemon, not the cost of
/// a redial made while the path was still coming back. Nothing reads
/// it after this point anyway.
fn reconnect(self: *AgentConnection, deadline_ms: i64) !void {
const r = &self.redial.?;
const q = &self.link.quic;
const cl = try quic.Client.connect(self.alloc, r.addr, r.key, r.idle_ms);
errdefer cl.deinit();
try waitReady(cl, deadline_ms);
q.cl.deinit();
q.cl = cl;
// Whatever the torn connection would not take is dropped with it. It
// is the tail of a frame the daemon never finished reading, and
// prefixing it onto the new stream would make the re-sent attach
// unparseable. (The staging used to be a stack buffer inside the
// send, so this was implicit.)
q.qout.clearRetainingCapacity();
r.reconnected = true;
}
};
test "graceMs: flat over a socket, RTT-derived over QUIC, and capped" {
const alloc = std.testing.allocator;
const local = AgentConnection{ .link = .{ .fd = -1 }, .alloc = alloc };
try std.testing.expectEqual(@as(i64, 2_000), local.graceMs());
// Grace is four times the handshake duration but never below two seconds.
// The calculation depends only on the stored measurement, so no live client
// is required.
var far = AgentConnection{
.link = .{ .quic = .{ .cl = undefined, .alloc = alloc } },
.redial = .{ .addr = undefined, .key = undefined, .idle_ms = 0, .connect_ms = 1 },
.alloc = alloc,
};
try std.testing.expectEqual(@as(i64, 2_000), far.graceMs());
// A 300ms handshake — a real intercontinental link — buys 1.2s, which
// is still under the floor, so the first number that moves it is a
// handshake past half a second.
far.redial.?.connect_ms = 300;
try std.testing.expectEqual(@as(i64, 2_000), far.graceMs());
far.redial.?.connect_ms = 900;
try std.testing.expectEqual(@as(i64, 3_600), far.graceMs());
// And it stops widening: past the cap we are no longer waiting on a
// daemon, we are waiting on a network that has already failed to carry
// an answer.
far.redial.?.connect_ms = 60_000;
try std.testing.expectEqual(@as(i64, 30_000), far.graceMs());
}
/// Drive a fresh QUIC connection until it can carry bytes. Immediate network
/// errors return early; a blackholed endpoint is bounded by the caller deadline
/// or the connection's idle timeout.
fn waitReady(cl: *quic.Client, deadline_ms: i64) !void {
while (true) {
cl.pump();
if (cl.isReady()) return;
if (cl.dead) return error.QuicHandshakeFailed;
const now = std.time.milliTimestamp();
if (now >= deadline_ms) return error.Timeout;
var fds = [_]std.posix.pollfd{
.{ .fd = cl.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
};
const cap: i32 = @intCast(@min(deadline_ms - now, 50));
_ = std.posix.poll(&fds, cl.timeoutMs(cap)) catch return error.QuicHandshakeFailed;
}
}
test "reconnect: redials the same coordinates, and a dead port is a fast no" {
const alloc = std.testing.allocator;
// Loopback port 1 returns an immediate ICMP error, exercising dial,
// handshake wait, and failure reporting without waiting for a timeout.
const addr = try std.net.Address.parseIp("127.0.0.1", 1);
const key: quic.Key = .{ .bytes = [_]u8{7} ** quic.key_len };
// Create the initial connection state, then exercise the same reconnect
// path used when an established connection later fails.
const deadline = std.time.milliTimestamp() + 2_000;
var conn = AgentConnection{
.link = .{ .quic = .{
.cl = try quic.Client.connect(alloc, addr, key, 1_000),
.alloc = alloc,
} },
.redial = .{ .addr = addr, .key = key, .idle_ms = 1_000, .connect_ms = 0 },
.alloc = alloc,
};
defer conn.close();
const t0 = std.time.milliTimestamp();
if (conn.reconnect(deadline)) |_| {
// Nothing listens there; a redial that reported success would mean
// the handshake wait had stopped being a wait for a handshake.
return error.TestUnexpectedResult;
} else |redial| {
try std.testing.expectEqual(error.QuicHandshakeFailed, redial);
// The bookkeeping awaitReissuing does, done here with the real
// error the real redial produced, so the sentence below is the one
// an agent gets rather than one this test made up.
conn.reconnect_failure = @errorName(redial);
}
// An immediate port rejection should return well before the two-second
// deadline.
try std.testing.expect(std.time.milliTimestamp() - t0 < 1_000);
// Preserve both the initial connection loss and the reconnect failure in
// the final diagnostic.
var buf: [128]u8 = undefined;
try std.testing.expectEqualStrings(
"connection lost; reconnect failed: QuicHandshakeFailed",
waitFailDetail(&buf, &conn, error.ConnectionLost),
);
// A redial that failed is not a reconnect spent — but it is also not a
// AgentConnection holding a freed client: the old one is torn down only once a
// new one is up, so the close above is safe on this path.
try std.testing.expect(!conn.redial.?.reconnected);
}
test "waitFailDetail: only a lost connection gets a sentence; the rest keep their names" {
const alloc = std.testing.allocator;
var buf: [128]u8 = undefined;
// Every other failure is untouched — the socket arm's reports must
// read exactly as they did before there was a QUIC arm.
const local = AgentConnection{ .link = .{ .fd = -1 }, .alloc = alloc };
try std.testing.expectEqualStrings("Timeout", waitFailDetail(&buf, &local, error.Timeout));
try std.testing.expectEqualStrings("DaemonGone", waitFailDetail(&buf, &local, error.DaemonGone));
// A tear with the one reconnect still unspent (nothing tried yet).
var far = AgentConnection{
.link = .{ .quic = .{ .cl = undefined, .alloc = alloc } },
.redial = .{ .addr = undefined, .key = undefined, .idle_ms = 0, .connect_ms = 0 },
.alloc = alloc,
};
try std.testing.expectEqualStrings("connection lost", waitFailDetail(&buf, &far, error.ConnectionLost));
// A tear AFTER a reconnect that worked: the second one inside a single
// wait, which is a different thing to be told than the first — the
// client did reconnect, and the path tore again anyway.
far.redial.?.reconnected = true;
try std.testing.expectEqualStrings(
"connection lost again, after the one reconnect",
waitFailDetail(&buf, &far, error.ConnectionLost),
);
// A buffer too small to hold the composed line drops the reason rather
// than the finding: the detail is the agent's only account of this.
far.reconnect_failure = "QuicHandshakeFailed";
var tiny: [8]u8 = undefined;
try std.testing.expectEqualStrings(
"connection lost; reconnect failed",
waitFailDetail(&tiny, &far, error.ConnectionLost),
);
}
test "awaitFrame ends a wait on exit_status, keeping the code" {
const alloc = std.testing.allocator;
// A pipe is sufficient because `awaitFrame` only polls and reads this test
// transport.
const pipe = try std.posix.pipe();
defer std.posix.close(pipe[0]);
defer std.posix.close(pipe[1]);
var conn = AgentConnection{ .link = .{ .fd = pipe[0] }, .alloc = alloc };
// A preceding snapshot proves attachment succeeded, so the exit frame means
// the session ended rather than the attach being rejected.
try proto.writeFrame(pipe[1], .snapshot, "");
// A push to skip on the way, then the session's last word. The reply
// this wait asked for is never coming, and the code is the answer.
try proto.writeFrame(pipe[1], .pty_mode, &[_]u8{0});
try proto.writeFrame(pipe[1], .exit_status, &[_]u8{5});
try std.testing.expectError(
error.SessionExited,
conn.awaitFrame(.status_reply, std.time.milliTimestamp() + 2000),
);
try std.testing.expectEqual(@as(?u8, 5), conn.session_exit);
// ...and it is spelled as a session ending, not as a command's code.
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
try printSessionEnded(out.writer(alloc), conn.session_exit, 42);
try std.testing.expectEqualStrings(
"{\"reason\":\"session_ended\",\"exit_code\":5,\"duration_ms\":42}\n",
out.items,
);
}
test "an exit_status before any snapshot is a refused attach, not a session that ended" {
const alloc = std.testing.allocator;
// A rejected 0x0 attach is encoded as `exit_status 1` plus close, identical
// to a real shell exit except that no snapshot precedes it.
const pipe = try std.posix.pipe();
defer std.posix.close(pipe[0]);
defer std.posix.close(pipe[1]);
var conn = AgentConnection{ .link = .{ .fd = pipe[0] }, .alloc = alloc };
try proto.writeFrame(pipe[1], .exit_status, &[_]u8{1});
try std.testing.expectError(
error.AttachRefused,
conn.awaitFrame(.status_reply, std.time.milliTimestamp() + 2000),
);
}
test "a re-attach forgets the snapshot it saw, so a refused reconnect is not an ending" {
const alloc = std.testing.allocator;
// A socketpair, not a pipe: this conn has to WRITE the attach as well as
// read down the ONE fd it holds, which is the shape under test. Through
// `std.c` because `std.posix` has no socketpair on 0.15.2.
var sp: [2]std.posix.fd_t = undefined;
try std.testing.expectEqual(
@as(c_int, 0),
std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &sp),
);
defer std.posix.close(sp[0]);
defer std.posix.close(sp[1]);
var conn = AgentConnection{ .link = .{ .fd = sp[0] }, .alloc = alloc };
// The first attach is served: a snapshot arrives and is skipped past on
// the way to a reply that never comes.
try proto.writeFrame(sp[1], .snapshot, "");
try std.testing.expectError(
error.Timeout,
conn.awaitFrame(.status_reply, std.time.milliTimestamp() + 50),
);
try std.testing.expect(conn.saw_snapshot);
// Reject the reconnect attachment. Resetting `saw_snapshot` prevents state
// from the old connection from misclassifying this as a session exit.
try attachZero(&conn, "s", std.time.milliTimestamp() + 2000);
try proto.writeFrame(sp[1], .exit_status, &[_]u8{1});
try std.testing.expectError(
error.AttachRefused,
conn.awaitFrame(.status_reply, std.time.milliTimestamp() + 2000),
);
}
test "a refusal that closes the socket before the input write is still reported as the refusal" {
const alloc = std.testing.allocator;
// Reproduce the race where the daemon's rejection frame and close arrive
// before the next write, causing BrokenPipe unless the pending frame is
// drained and classified.
var sp: [2]std.posix.fd_t = undefined;
try std.testing.expectEqual(
@as(c_int, 0),
std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &sp),
);
defer std.posix.close(sp[0]);
var conn = AgentConnection{ .link = .{ .fd = sp[0] }, .alloc = alloc };
const deadline = std.time.milliTimestamp() + 2000;
// The field ordering: the attach is served, the refusal comes back,
// and the close beats the input write that follows it.
try attachZero(&conn, "nosuch", deadline);
try proto.writeFrame(sp[1], .exit_status, &[_]u8{1});
std.posix.close(sp[1]);
try std.testing.expectError(error.AttachRefused, conn.sendFrame(.input, "x", deadline));
// And the verb's answer is the refusal's JSON, not the write's.
var sp2: [2]std.posix.fd_t = undefined;
try std.testing.expectEqual(
@as(c_int, 0),
std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &sp2),
);
defer std.posix.close(sp2[0]);
try proto.writeFrame(sp2[1], .exit_status, &[_]u8{1});
std.posix.close(sp2[1]);
const saved = try std.posix.dup(std.posix.STDOUT_FILENO);
defer {
std.posix.dup2(saved, std.posix.STDOUT_FILENO) catch {};
std.posix.close(saved);
}
const cap = try std.posix.pipe();
defer std.posix.close(cap[0]);
try std.posix.dup2(cap[1], std.posix.STDOUT_FILENO);
std.posix.close(cap[1]);
var conn2 = AgentConnection{ .link = .{ .fd = sp2[0] }, .alloc = alloc };
const code = try verbSend(alloc, &conn2, "x", "nosuch", std.time.milliTimestamp() + 2000);
try std.posix.dup2(saved, std.posix.STDOUT_FILENO);
var buf: [1024]u8 = undefined;
const n = try std.posix.read(cap[0], &buf);
try std.testing.expectEqual(@as(u8, 1), code);
const want = "{\"error\":\"attach refused\",\"detail\":";
try std.testing.expectEqualStrings(want, buf[0..@min(n, want.len)]);
}
test "the refused-attach detail names the session and both of the refusal's producers" {
var buf: [768]u8 = undefined;
// The name is in the detail because it is the one thing the agent got
// wrong; the `{"error":"attach refused","detail":` wrapper around it is
// `fail`'s, pinned through a real verb by the capture test above.
const named = refusedDetail(&buf, "nosuch", .attach);
try std.testing.expect(std.mem.indexOf(u8, named, "so nosuch must already exist") != null);
// Mention both possible causes because the same frame also represents a
// full client table.
try std.testing.expect(std.mem.indexOf(u8, named, "room for one more client") != null);
// The empty name is the wire's default spelling, not a session called
// "": a detail reading `so must already exist` would send an agent
// looking for a name it never typed.
const dflt = refusedDetail(&buf, "", .attach);
try std.testing.expect(std.mem.indexOf(u8, dflt, "so the default session must already exist") != null);
}
test "a refusal a full client table cannot have caused does not blame one" {
var buf: [768]u8 = undefined;
// Status never attaches, so this exit frame means the session lookup failed;
// observer-slot exhaustion closes without a frame.
const q = refusedDetail(&buf, "nosuch", .query);
try std.testing.expect(std.mem.indexOf(u8, q, "so nosuch must already exist") != null);
try std.testing.expect(std.mem.indexOf(u8, q, "max_clients") == null);
try std.testing.expect(std.mem.indexOf(u8, q, "room for one more client") == null);
// The verb never attached, so the sentence must not open by calling
// this an attach either.
try std.testing.expect(std.mem.indexOf(u8, q, "refused this attach") == null);
}
/// Exit code used when the required JSON object could not be written to stdout.
const write_failed_code: u8 = 4;
/// A write that fails must not exit 0: an agent checks the status,
/// then has no object.
fn emit(json: []const u8, ok: u8) u8 {
return emitTo(std.posix.STDOUT_FILENO, json, ok);
}
/// `emit` against a named fd, which is the whole reason it is split out:
/// stdout is not something a test can break without breaking the runner.
fn emitTo(fd: std.posix.fd_t, json: []const u8, ok: u8) u8 {
proto.writeAllFd(fd, json) catch |e| {
// Best-effort by construction: whatever took stdout away has very
// often taken stderr with it, and the exit code is the half that
// survives either way. This line is for the human reading the log.
var buf: [128]u8 = undefined;
const msg = std.fmt.bufPrint(&buf, "mux a: cannot write the reply: {s}\n", .{@errorName(e)}) catch
"mux a: cannot write the reply\n";
proto.writeAllFd(std.posix.STDERR_FILENO, msg) catch {};
return write_failed_code;
};
return ok;
}
test "an unwritable stdout is a distinct exit code, never a silent 0" {
// The diagnostic goes to stderr, and a test that let it through would
// print a line that reads like a failure on every green run. Swapped
// for /dev/null and put back.
const saved = try std.posix.dup(std.posix.STDERR_FILENO);
defer {
std.posix.dup2(saved, std.posix.STDERR_FILENO) catch {};
std.posix.close(saved);
}
const devnull = try std.posix.open("/dev/null", .{ .ACCMODE = .WRONLY }, 0);
defer std.posix.close(devnull);
try std.posix.dup2(devnull, std.posix.STDERR_FILENO);
// A pipe whose reader is gone: the next write is EPIPE, which is the
// field case — an agent harness that stopped reading this mode's stdout.
const gone = try std.posix.pipe();
std.posix.close(gone[0]);
defer std.posix.close(gone[1]);
try std.testing.expectEqual(write_failed_code, emitTo(gone[1], "{\"sent\":true}\n", 0));
// And the code the caller asked for is passed through untouched when
// the object does land — including the nonzero ones, which must not be
// confused with the write having failed.
const live = try std.posix.pipe();
defer std.posix.close(live[0]);
defer std.posix.close(live[1]);
try std.testing.expectEqual(@as(u8, 3), emitTo(live[1], "{\"reason\":\"timeout\"}\n", 3));
var buf: [64]u8 = undefined;
const n = try std.posix.read(live[0], &buf);
try std.testing.expectEqualStrings("{\"reason\":\"timeout\"}\n", buf[0..n]);
}
/// Emit the common JSON error shape used by every runtime failure.
fn fail(msg: []const u8, detail: []const u8) u8 {
var buf: [2048]u8 = undefined;
var fbs = std.io.fixedBufferStream(&buf);
writeError(fbs.writer(), msg, detail) catch {
// The message did not fit. Still JSON, still one line.
return emit("{\"error\":\"failure too long to report\"}\n", 1);
};
return emit(fbs.getWritten(), 1);
}
/// `fail` for a verb prefix known only at runtime — every failure in
/// the shared await/run pipeline. A message too long to prefix falls
/// back to the unprefixed one rather than losing the failure.
fn failAs(who: []const u8, msg: []const u8, detail: []const u8) u8 {
var buf: [256]u8 = undefined;
const joined = std.fmt.bufPrint(&buf, "{s}: {s}", .{ who, msg }) catch msg;
return fail(joined, detail);
}
fn writeError(writer: anytype, msg: []const u8, detail: []const u8) !void {
try writer.writeAll("{\"error\":");
try jsonEscape(writer, msg);
try writer.writeAll(",\"detail\":");
try jsonEscape(writer, detail);
try writer.writeAll("}\n");
}
/// `--timeout 0` means no bound here as everywhere else (AwaitReq): a
/// past deadline would fail instantly instead of waiting forever.
fn deadlineFor(timeout_ms: u32) i64 {
if (timeout_ms == 0) return std.math.maxInt(i64);
return std.time.milliTimestamp() + timeout_ms;
}
/// The session ended under us: JSON, but on the failure path — the
/// verb that asked (status, capture, send) has no answer to give.
/// `run` and `await` do have one and print it themselves.
fn failSessionEnded(code: ?u8) u8 {
var buf: [192]u8 = undefined;
var fbs = std.io.fixedBufferStream(&buf);
writeSessionEndedError(fbs.writer(), code) catch return 1;
return emit(fbs.getWritten(), 1);
}
/// `detail` is here because every other failure has one: an agent reading
/// `.detail` on any exit-1 must never meet a missing key. `exit_code` is the
/// machine field; the detail says the same in the other failures' prose.
fn writeSessionEndedError(writer: anytype, code: ?u8) !void {
try writer.writeAll("{\"error\":\"session ended\",\"detail\":");
if (code) |c| {
var buf: [40]u8 = undefined;
try jsonEscape(writer, try std.fmt.bufPrint(&buf, "shell exited with {d}", .{c}));
} else {
try jsonEscape(writer, "shell exited without reporting a code");
}
try writer.writeAll(",\"exit_code\":");
try writeExitCode(writer, code);
try writer.writeAll("}\n");
}
/// Kind of request rejected by the daemon. Status and capture are queries rather
/// than attachments, which changes the diagnostic for an exit frame received
/// before any snapshot.
const RefusedRequest = enum { attach, query };
/// Report an attach rejection instead of describing a session that never ran as
/// having ended.
fn failAttachRefused(name: []const u8, ask: RefusedRequest) u8 {
var buf: [768]u8 = undefined;
return fail("attach refused", refusedDetail(&buf, name, ask));
}
/// Map attach rejection from both send and receive paths to the same JSON error.
fn failSend(e: anyerror, name: []const u8, ask: RefusedRequest, who: []const u8, msg: []const u8) u8 {
if (e == error.AttachRefused) return failAttachRefused(name, ask);
return failAs(who, msg, @errorName(e));
}
/// Format the ambiguous pre-snapshot exit: either the session does not exist or
/// the daemon has no free client slot. Do not claim one cause when the frame
/// cannot distinguish them.
fn refusedDetail(buf: *[768]u8, name: []const u8, ask: RefusedRequest) []const u8 {
const why, const need = switch (ask) {
.attach => .{
"the daemon refused this attach: `mux a` joins at 0x0 and never creates, so ",
" must already exist and the daemon must have room for one more client (max_clients)",
},
.query => .{
"the daemon refused this query: `mux a` never creates a session, so ",
" must already exist",
},
};
// Display an empty wire name as the default session rather than as missing
// text in the diagnostic.
const shown = if (name.len == 0) "the default session" else name;
return std.fmt.bufPrint(buf, "{s}{s}{s}", .{ why, shown, need }) catch
std.fmt.bufPrint(buf, "{s}the session asked for{s}", .{ why, need }) catch why;
}
test "the session-ended failure keeps the error+detail shape every failure has" {
var buf: [192]u8 = undefined;
var fbs = std.io.fixedBufferStream(&buf);
try writeSessionEndedError(fbs.writer(), 5);
try std.testing.expectEqualStrings(
"{\"error\":\"session ended\",\"detail\":\"shell exited with 5\",\"exit_code\":5}\n",
fbs.getWritten(),
);
// No code is still a detail, never a missing key.
var none = std.io.fixedBufferStream(&buf);
try writeSessionEndedError(none.writer(), null);
try std.testing.expect(std.mem.indexOf(u8, none.getWritten(), "\"detail\":\"shell exited without") != null);
try std.testing.expect(std.mem.indexOf(u8, none.getWritten(), "\"exit_code\":null") != null);
}
/// Run agent mode using the argv slice supplied by the top-level dispatcher.
pub fn main(args: []const [:0]const u8) !u8 {
var arena_state = std.heap.ArenaAllocator.init(std.heap.page_allocator);
defer arena_state.deinit();
const alloc = arena_state.allocator();
// Successful and runtime responses use one JSON object on stdout. Shared
// argument errors retain the CLI help/version stream behavior.
const o = parseArgs(args) catch |e| return cliflags.exitFor(e, usage, "mux", build_options.version);
// Start the deadline before connecting so QUIC handshake time is included
// and timeout semantics match local connections.
const deadline = deadlineFor(o.timeout);
if (o.quic) |host_port| {
var conn = switch (openQuicConn(alloc, o, host_port, deadline)) {
.conn => |c| c,
.exit => |code| return code,
};
defer conn.close();
return dispatch(alloc, &conn, o, deadline);
}
const sock_path = if (o.sock) |s| s else sockpath.defaultSockPath(alloc) catch |err| switch (err) {
// Runtime path resolution failures use the same JSON shape as other
// agent-mode failures. The `error` is this mode's own word, but the
// DETAIL is `sockpath`'s one sentence: `mux a` had its own copy,
// and on a Mac whose /tmp/mux-<uid> is the thing at fault that copy
// sent the reader off to set a variable that was already correct.
error.NoRuntimeDir => return fail(
"no default socket path",
sockpath.no_runtime_dir_reason,
),
else => |e| return e,
};
var conn = AgentConnection.open(alloc, sock_path) catch |e| {
// Include the socket path so a client can identify which endpoint failed.
var buf: [256]u8 = undefined;
const detail = std.fmt.bufPrint(&buf, "{s}: {s}", .{ sock_path, @errorName(e) }) catch
@errorName(e);
return fail("cannot connect to the daemon", detail);
};
defer conn.close();
return dispatch(alloc, &conn, o, deadline);
}
/// Dispatch a parsed agent command over a transport-independent connection.
fn dispatch(alloc: std.mem.Allocator, conn: *AgentConnection, o: AgentArguments, deadline: i64) !u8 {
return switch (o._verb.?) {
.status => verbStatus(alloc, conn, o.sessionName(), deadline),
.capture => verbCapture(alloc, conn, o.vt, o.sessionName(), deadline),
.send => verbSend(alloc, conn, o._arg, o.sessionName(), deadline),
// Validate `run`'s required command line before sharing the await
// pipeline, where null specifically identifies the `await` command.
.run => if (o._arg) |cmdline|
awaitVerb(alloc, conn, o, deadline, cmdline)
else
fail("run: needs CMDLINE", ""),
.await => awaitVerb(alloc, conn, o, deadline, null),
};
}
/// Result of opening a QUIC connection: either the connection or an exit code
/// after a JSON error has already been emitted.
const QuicOpenResult = union(enum) { conn: AgentConnection, exit: u8 };
fn openQuicConn(
alloc: std.mem.Allocator,
o: AgentArguments,
host_port: []const u8,
deadline: i64,
) QuicOpenResult {
// Reuse XDG helpers for `--key`, `MUX_KEY_FILE`, and default-path precedence
// so every client selects the same credential.
const res = xdg.resolveKeyPath(alloc, xdg.pickKey(o.key, std.posix.getenv(xdg.key_env))) catch |e|
return .{ .exit = fail("quic: cannot resolve a key path", @errorName(e)) };
// `key_path` is read here and nowhere after: `Key.load` copies the bytes
// it needs and the connection keeps no path. Freeing on the way out is a
// no-op under this arena, and keeps the site correct under any allocator.
defer res.deinit(alloc);
const key_path = switch (res) {
.given, .default => |p| p,
// Include the missing path so callers know which credential to create.
.missing => |p| return .{ .exit = fail(
"quic: no key: pass --key, set MUX_KEY_FILE, or run `mux d keygen`",
p,
) },
};
const key = quic.Key.load(key_path) catch |e| {
// Reuse the daemon's key validation text, including permission checks.
var buf: [quic.key_refusal_len]u8 = undefined;
return .{ .exit = fail("quic: unusable key", quic.keyRefusalBody(&buf, e, key_path)) };
};
const addr = quic.parseAddr(alloc, host_port) catch |e| {
var buf: [512]u8 = undefined;
const detail = std.fmt.bufPrint(&buf, "{s}: {s}", .{ host_port, @errorName(e) }) catch
@errorName(e);
return .{ .exit = fail("quic: cannot read HOST:PORT", detail) };
};
const conn = AgentConnection.openQuic(alloc, addr, key, quic.default_idle_ms, deadline) catch |e| {
var buf: [512]u8 = undefined;
const detail = std.fmt.bufPrint(&buf, "{s}: {s}", .{ host_port, @errorName(e) }) catch
@errorName(e);
return .{ .exit = fail("cannot connect to the daemon", detail) };
};
return .{ .conn = conn };
}
fn verbStatus(alloc: std.mem.Allocator, conn: *AgentConnection, session: []const u8, deadline: i64) !u8 {
// status_req's WHOLE payload is the name — this connection never
// attaches (see attachZero's callers; `status` is not one of them), so
// there is no slot for the daemon to fall back to and the tail is the
// only word this ask gets to say.
conn.sendFrame(.status_req, session, deadline) catch |e|
return failSend(e, session, .query, "status", "send failed");
const frame = conn.awaitFrame(.status_reply, deadline) catch |e| switch (e) {
error.AttachRefused => return failAttachRefused(session, .query),
error.SessionExited => return failSessionEnded(conn.session_exit),
else => return fail("status: no reply", @errorName(e)),
};
defer frame.deinit(alloc);
const st = proto.decodeStatusReply(frame.payload) catch |e|
return fail("status: bad reply", @errorName(e));
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
try printStatus(out.writer(alloc), st);
return emit(out.items, 0);
}
/// A command that has not returned, or whose mechanism cannot know a
/// code, has no exit code: 0 would read as "succeeded".
fn writeExitCode(writer: anytype, code: ?u8) !void {
if (code) |c| {
try writer.print("{d}", .{c});
} else {
try writer.writeAll("null");
}
}
/// The five `CmdState` fields `status` and `await`/`run` both publish. A bare
/// fragment — no braces, no commas — because the two verbs nest it differently:
/// `status` puts it inside a `"cmd"` object, `await` inlines it at the top.
fn writeCmdFields(writer: anytype, st: proto.CmdState) !void {
try writer.writeAll("\"phase\":");
try jsonEscape(writer, @tagName(st.phase));
try writer.writeAll(",\"mechanism\":");
try jsonEscape(writer, @tagName(st.mechanism));
try writer.writeAll(",\"exit_code\":");
try writeExitCode(writer, st.exit_code);
try writer.print(",\"start_row\":{d},\"end_row\":{d}", .{ st.start_row, st.end_row });
}
fn printStatus(writer: anytype, st: proto.StatusReply) !void {
try writer.print(
"{{\"cols\":{d},\"rows\":{d},\"cursor\":{{\"x\":{d},\"y\":{d}}}," ++
"\"history_rows\":{d},\"alt_screen\":{},\"icanon\":{},\"echo\":{},\"cmd\":{{",
.{ st.cols, st.rows, st.cursor_x, st.cursor_y, st.history_rows, st.alt_screen, st.mode.icanon, st.mode.echo },
);
try writeCmdFields(writer, st.cmd);
// The watermark, and only `status` carries it: this is the number an
// agent feeds back as `since_seq`, which is why `await` does not print
// one (see proto.CmdState.seq).
try writer.print(",\"seq\":{d}}}}}\n", .{st.cmd.seq});
}
test "printStatus spells a pending exit code as JSON null" {
var buf: [512]u8 = undefined;
var fbs = std.io.fixedBufferStream(&buf);
try printStatus(fbs.writer(), .{
.cols = 80,
.rows = 24,
.cursor_x = 1,
.cursor_y = 2,
.history_rows = 7,
.alt_screen = false,
.mode = .{ .icanon = true, .echo = true },
.cmd = .{ .phase = .running, .mechanism = .marks, .exit_code = null, .start_row = 3, .end_row = 4, .seq = 9 },
});
// The whole object, byte for byte, not a handful of substrings: this is
// this mode's published contract with an agent's JSON parser, and the fields
// it shares with `await` are written by a helper both verbs call — a
// pin on the parts cannot see a comma or a nesting level move.
try std.testing.expectEqualStrings(
"{\"cols\":80,\"rows\":24,\"cursor\":{\"x\":1,\"y\":2},\"history_rows\":7," ++
"\"alt_screen\":false,\"icanon\":true,\"echo\":true," ++
"\"cmd\":{\"phase\":\"running\",\"mechanism\":\"marks\",\"exit_code\":null," ++
"\"start_row\":3,\"end_row\":4,\"seq\":9}}\n",
fbs.getWritten(),
);
}
fn verbCapture(alloc: std.mem.Allocator, conn: *AgentConnection, vt: bool, session: []const u8, deadline: i64) !u8 {
// vt byte ++ session-name tail, the same shape the daemon's own `dump` sends
// — and built by the same encoder, so it cannot drift from it.
var buf: [proto.debug_dump_max_len]u8 = undefined;
const payload = proto.encodeDebugDumpNamed(&buf, vt, session);
conn.sendFrame(.debug_dump, payload, deadline) catch |e|
return failSend(e, session, .query, "capture", "send failed");
const frame = conn.awaitFrame(.dump_reply, deadline) catch |e| switch (e) {
error.AttachRefused => return failAttachRefused(session, .query),
error.SessionExited => return failSessionEnded(conn.session_exit),
else => return fail("capture: no reply", @errorName(e)),
};
defer frame.deinit(alloc);
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
const writer = out.writer(alloc);
try writer.writeAll("{\"grid\":");
try jsonEscape(writer, frame.payload);
try writer.writeAll("}\n");
return emit(out.items, 0);
}
/// Join without claiming terminal dimensions. The 0x0 size cannot create a
/// session and does not resize an existing human client's grid.
fn attachZero(conn: *AgentConnection, name: []const u8, deadline: i64) !void {
// Clear snapshot state for every attachment, including reconnect, so an
// attach rejection cannot inherit success from the previous connection.
conn.saw_snapshot = false;
var buf: [proto.attach_max_len]u8 = undefined;
try conn.sendFrame(.attach, proto.encodeAttachNamed(&buf, 0, 0, 0, 0, name), deadline);
}
fn verbSend(alloc: std.mem.Allocator, conn: *AgentConnection, arg: ?[]const u8, session: []const u8, deadline: i64) !u8 {
const spec = arg orelse return fail("send: needs BYTES", "");
const bytes = decodeEscapes(alloc, spec) catch |e| return fail("send: bad escape", @errorName(e));
defer alloc.free(bytes);
attachZero(conn, session, deadline) catch |e|
return failSend(e, session, .attach, "send", "attach failed");
conn.sendFrame(.input, bytes, deadline) catch |e|
return failSend(e, session, .attach, "send", "input failed");
// Write-and-close LOSES the input, and not rarely: the daemon flushes a
// client's pending bytes BEFORE it reads that client, so closing straight
// after the write means the flush hits EPIPE and the input frame is
// discarded unread. The round trip IS the acknowledgement — frames are
// served in stream order, so a `status_reply` proves the daemon read past
// the input. The same `session` as the attach, not "": the daemon answers
// only a tail that names the slot's own session.
conn.sendFrame(.status_req, session, deadline) catch |e|
return failSend(e, session, .attach, "send", "ack request failed");
const ack = conn.awaitFrame(.status_reply, deadline) catch |e| switch (e) {
// The bytes we sent ended the session (`exit\n`). Reported as the
// session's death rather than as "sent", because this verb's answer
// is about the send and there is no longer a session to have sent
// to — an agent that wants the death to be an ANSWER runs `run`.
error.AttachRefused => return failAttachRefused(session, .attach),
error.SessionExited => return failSessionEnded(conn.session_exit),
else => return fail("send: daemon never acknowledged the input", @errorName(e)),
};
ack.deinit(alloc);
conn.sendFrame(.detach, "", deadline) catch |e| return fail("send: detach failed", @errorName(e));
return emit("{\"sent\":true}\n", 0);
}
/// Additional client-side wait beyond the daemon timeout. The daemon starts its
/// timer only after reading `await_req`, so equal deadlines would make the
/// client expire before receiving the daemon's timeout response.
const await_grace_ms = 2_000;
/// The ceiling on the QUIC arm's derived grace (AgentConnection.graceMs), and the
/// reason it has one is that `connect_ms` has no bound of its own worth
/// multiplying by four.
const grace_cap_ms = 30_000;
/// How long `writeThrough` keeps offering a frame's tail to a full egress ring.
/// Reaching it means the peer stopped acknowledging 256KB of backlog — a dead
/// connection in a different hat — but the bound is what keeps this call, which
/// has no deadline of its own, from waiting forever.
const send_flush_ms = 5_000;
/// How long a write that died of a closed peer will look for the refusal
/// the peer left behind. The socket is already closed, so the drain ends at
/// EOF long before this — the bound is for the half-closed case, where
/// nothing else would end the wait.
const refusal_drain_ms = 100;
/// The span fetch gets its own window rather than the tail of the run's: a
/// command that returned in the last millisecond of `--timeout` still has a
/// transcript worth having, and this round trip is a local read that either
/// answers promptly or is not coming.
const span_fetch_ms = 2_000;
/// No run deadline: `--timeout 0` would make the fetch unbounded.
fn spanFetchDeadline() i64 {
return deadlineFor(span_fetch_ms);
}
test "the span fetch is bounded even when the run it follows was not" {
// `--timeout 0` is the case that matters: the run's deadline is then
// "never", and a fetch that inherited it would outlive the answer.
try std.testing.expectEqual(std.math.maxInt(i64), deadlineFor(0));
const before = std.time.milliTimestamp();
const span = spanFetchDeadline();
try std.testing.expect(span >= before);
try std.testing.expect(span <= std.time.milliTimestamp() + span_fetch_ms);
}
/// Ask to be told when the session next comes to rest, and wait for it.
fn doAwait(
alloc: std.mem.Allocator,
conn: *AgentConnection,
o: AgentArguments,
since_seq: u64,
deadline: i64,
) !proto.AwaitReply {
var buf: [proto.await_req_max_len]u8 = undefined;
const payload = proto.encodeAwaitReqNamed(&buf, .{
.since_seq = since_seq,
.settle_ms = o.settle,
.timeout_ms = o.timeout,
}, o.sessionName());
try conn.sendFrame(.await_req, payload, deadline);
const frame = try conn.awaitFrame(.await_reply, deadline);
defer frame.deinit(alloc);
return try proto.decodeAwaitReply(frame.payload);
}
/// At-most-once: the re-issue re-sends the attach and the request,
/// never `run`'s input.
fn awaitReissuing(
alloc: std.mem.Allocator,
conn: *AgentConnection,
o: AgentArguments,
since_seq: u64,
deadline: i64,
) !proto.AwaitReply {
return doAwait(alloc, conn, o, since_seq, deadline) catch |e| switch (e) {
// `SendStalled` never redials: the peer is still there, it has
// just stopped acking, so a redial would be a second guess about
// a connection that never said it was gone.
error.ConnectionLost => {
// Once per process, not per await: a loop here is a client
// that hides a daemon that is gone.
if (conn.redial == null or conn.redial.?.reconnected) return e;
// The deadline continues across the redial — four seconds
// spent redialling are four seconds of the caller's wait, not
// a fresh bound.
conn.reconnect(deadline) catch |redial| {
conn.reconnect_failure = @errorName(redial);
return e;
};
// The attach is part of the reconnect, not a separate step: a
// daemon that lost our connection lost the client slot with
// it, so an await_req arriving unattached asks about nothing.
// A failure here is still the reconnect failing.
attachZero(conn, o.sessionName(), deadline) catch |reattach| {
// A refused re-attach is the daemon's answer, not the tear
// that got us here: `ConnectionLost` would send an agent
// to check the network for a session that is gone.
if (reattach == error.AttachRefused) return reattach;
conn.reconnect_failure = @errorName(reattach);
return e;
};
// The SAME `since_seq`, re-read from nothing: the request is a
// question about a watermark, so re-asking it is idempotent,
// while a watermark taken from the new connection would sit
// past a return that happened while we were disconnected.
return doAwait(alloc, conn, o, since_seq, deadline);
},
else => e,
};
}
/// What a wait that ended without a reply says past the verb's own "no reply".
/// Every error but one is its own name, because `ConnectionLost` is the only one
/// whose name is half the story: the redial failed, the redial was already
/// spent, or nothing tried to redial.
fn waitFailDetail(buf: []u8, conn: *const AgentConnection, e: anyerror) []const u8 {
if (e != error.ConnectionLost) return @errorName(e);
if (conn.reconnect_failure) |why| {
return std.fmt.bufPrint(buf, "connection lost; reconnect failed: {s}", .{why}) catch
"connection lost; reconnect failed";
}
const spent = if (conn.redial) |r| r.reconnected else false;
if (spent) return "connection lost again, after the one reconnect";
return "connection lost";
}
/// Sequence number of the session's most recent command return, or zero when no
/// command has returned. It becomes `since_seq` for the subsequent wait.
fn currentSeq(alloc: std.mem.Allocator, conn: *AgentConnection, session: []const u8, deadline: i64) !u64 {
// Same session as the attach that precedes this call — the attached-
// tail equality rule (server.zig) demands it.
try conn.sendFrame(.status_req, session, deadline);
const frame = try conn.awaitFrame(.status_reply, deadline);
defer frame.deinit(alloc);
const s = try proto.decodeStatusReply(frame.payload);
return s.cmd.seq;
}
/// A `scrollback_chunk` payload as the text the command printed: an agent
/// reading `output` wants what was on the screen, not how it was coloured.
/// The payload leads with the start and count it answers, and the rows after
/// that are `CellRow`s, so this is a decode. It used to be an SGR stripper
/// over VT rows, and when the wire started carrying cells that stripper
/// passed the row headers straight through — a NUL and a 0x80 ahead of every
/// line of `output`, which `make agent` caught on 2026-09-04.
///
/// The COUNT is the chunk's own, not the request's: the daemon clamps a span
/// that runs past what it still holds, and reading more rows than it sent
/// would decode whatever followed in the frame.
fn spanText(alloc: std.mem.Allocator, payload: []const u8) !?[]u8 {
if (payload.len <= 6) return null;
const count = std.mem.readInt(u16, payload[4..6], .little);
if (count == 0) return null;
const rows = try grid.decodeRows(alloc, payload[6..], count, null);
defer grid.freeRows(alloc, rows);
return try grid.dumpRowsPlain(alloc, rows);
}
test "spanText decodes the chunk's rows and joins them as text" {
const alloc = std.testing.allocator;
var payload: std.ArrayList(u8) = .empty;
defer payload.deinit(alloc);
try payload.appendSlice(alloc, &.{ 3, 0, 0, 0, 2, 0 }); // start 3, count 2
for ([_][]const u8{ "out-42", "second" }) |line| {
var w = try proto.CellRowWriter.begin(&payload, alloc);
defer w.deinit();
for (line) |ch| try w.cell(.{}, .narrow, &.{ch});
w.finish();
}
const got = (try spanText(alloc, payload.items)).?;
defer alloc.free(got);
try std.testing.expectEqualStrings("out-42\nsecond", got);
}
/// Rows go stale between reply and fetch, so failure here is a null
/// output, not a failed run.
fn fetchSpan(
alloc: std.mem.Allocator,
conn: *AgentConnection,
start_row: u32,
end_row: u32,
deadline: i64,
) !?[]u8 {
if (end_row <= start_row) return null;
const count: u16 = @intCast(@min(end_row - start_row, std.math.maxInt(u16)));
try conn.sendFrame(.fetch_scrollback, &proto.encodeScrollbackReq(start_row, count), deadline);
const frame = try conn.awaitFrame(.scrollback_chunk, deadline);
defer frame.deinit(alloc);
return try spanText(alloc, frame.payload);
}
/// `output` is absent when there is no transcript: an absent key and
/// an empty string differ.
fn printAwaitReply(
writer: anytype,
r: proto.AwaitReply,
output: ?[]const u8,
duration_ms: i64,
) !void {
try writer.writeAll("{\"reason\":");
try jsonEscape(writer, @tagName(r.reason));
try writer.writeAll(",");
try writeCmdFields(writer, r.state);
try writer.print(",\"duration_ms\":{d}", .{duration_ms});
if (output) |text| {
try writer.writeAll(",\"output\":");
try jsonEscape(writer, text);
}
try writer.writeAll("}\n");
}
test "printAwaitReply omits output when there is none and spells a missing code null" {
const alloc = std.testing.allocator;
const r: proto.AwaitReply = .{
.state = .{
.phase = .returned,
.mechanism = .settle,
.exit_code = null,
.start_row = 3,
.end_row = 9,
.seq = 12,
},
.reason = .settled,
};
var bare: std.ArrayList(u8) = .empty;
defer bare.deinit(alloc);
try printAwaitReply(bare.writer(alloc), r, null, 250);
try std.testing.expectEqualStrings(
"{\"reason\":\"settled\",\"phase\":\"returned\",\"mechanism\":\"settle\"," ++
"\"exit_code\":null,\"start_row\":3,\"end_row\":9,\"duration_ms\":250}\n",
bare.items,
);
var with: std.ArrayList(u8) = .empty;
defer with.deinit(alloc);
try printAwaitReply(with.writer(alloc), r, "a\nb", 250);
try std.testing.expect(std.mem.indexOf(u8, with.items, "\"output\":\"a\\nb\"") != null);
}
/// Emit the successful `run` or `await` result after the session command ends,
/// including a nonzero session exit code as data.
fn printSessionEnded(writer: anytype, code: ?u8, duration_ms: i64) !void {
try writer.writeAll("{\"reason\":\"session_ended\",\"exit_code\":");
try writeExitCode(writer, code);
try writer.print(",\"duration_ms\":{d}}}\n", .{duration_ms});
}
/// Timeout is the only nonzero code: a command returning nonzero
/// failed in `exit_code`, not here.
fn reportAwait(
alloc: std.mem.Allocator,
r: proto.AwaitReply,
output: ?[]const u8,
duration_ms: i64,
) !u8 {
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
try printAwaitReply(out.writer(alloc), r, output, duration_ms);
return emit(out.items, if (r.reason == .timeout) 3 else 0);
}
fn reportSessionEnded(alloc: std.mem.Allocator, code: ?u8, duration_ms: i64) !u8 {
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
try printSessionEnded(out.writer(alloc), code, duration_ms);
return emit(out.items, 0);
}
/// `run` is `await` with a command line put in: `cmdline` non-null is
/// the whole difference.
fn awaitVerb(
alloc: std.mem.Allocator,
conn: *AgentConnection,
o: AgentArguments,
deadline: i64,
cmdline: ?[]const u8,
) !u8 {
// Every error string this function can print names the verb the user
// typed, because "attach failed" from the wrong verb sends an agent
// looking in the wrong place.
const who = if (cmdline == null) "await" else "run";
const started = std.time.milliTimestamp();
attachZero(conn, o.sessionName(), deadline) catch |e|
return failSend(e, o.sessionName(), .attach, who, "attach failed");
// Read the watermark before sending input. A fast command could otherwise
// return between those operations and leave the wait targeting a later
// sequence.
const since = currentSeq(alloc, conn, o.sessionName(), deadline) catch |e| switch (e) {
error.AttachRefused => return failAttachRefused(o.sessionName(), .attach),
error.SessionExited => return reportSessionEnded(alloc, conn.session_exit, elapsed(started)),
else => return failAs(who, "status failed", @errorName(e)),
};
if (cmdline) |cmd| {
// The cmdline goes to the pty verbatim, plus the newline that submits
// it. No ack round trip: the `await_req` that follows is itself the read
// proving the daemon got past this frame.
const line = std.fmt.allocPrint(alloc, "{s}\n", .{cmd}) catch |e|
return failAs(who, "cannot build the command line", @errorName(e));
defer alloc.free(line);
conn.sendFrame(.input, line, deadline) catch |e|
return failSend(e, o.sessionName(), .attach, who, "input failed");
}
const r = awaitReissuing(alloc, conn, o, since, awaitDeadline(o, conn)) catch |e| switch (e) {
error.AttachRefused => return failAttachRefused(o.sessionName(), .attach),
error.SessionExited => return reportSessionEnded(alloc, conn.session_exit, elapsed(started)),
else => {
var detail: [128]u8 = undefined;
return failAs(who, "no reply", waitFailDetail(&detail, conn, e));
},
};
// Only the marks regime knows where the command's rows are; pgid and
// settle answer WHEN, never WHERE, and a span from them would be a
// guess dressed as a transcript. `await` never fetches one at all: it
// did not start the command, so the span it would name is not its own.
var output: ?[]u8 = null;
defer if (output) |text| alloc.free(text);
if (cmdline != null and r.state.mechanism == .marks and r.reason == .returned) {
output = fetchSpan(
alloc,
conn,
r.state.start_row,
r.state.end_row,
spanFetchDeadline(),
) catch null;
}
return reportAwait(alloc, r, output, elapsed(started));
}
fn elapsed(started: i64) i64 {
return std.time.milliTimestamp() - started;
}
/// This client's deadline: the daemon's own bound plus the grace window
/// (await_grace_ms, widened per transport by `AgentConnection.graceMs`). Unbounded
/// stays unbounded.
fn awaitDeadline(o: AgentArguments, conn: *const AgentConnection) i64 {
if (o.timeout == 0) return std.math.maxInt(i64);
return std.time.milliTimestamp() + o.timeout + conn.graceMs();
}
// Ensure every public declaration is semantically analyzed during tests;
// `std.meta.declarations` does not include private declarations.
test {
std.testing.refAllDeclsRecursive(@This());
}