593f8fa3

feat(muxa): QUIC transport — same verbs, remote daemons, one reconnect re-issue

a73x 2026-08-13 18:47

Commit message

feat(muxa): QUIC transport — same verbs, remote daemons, one reconnect re-issue

`--quic HOST[:PORT]` and `--key PATH` put every verb over a remote daemon
with the same frames and the same JSON. Conn becomes a union over the two
links; the verbs above it were not touched and cannot tell which one they
are on, which is the claim the flag makes.

The QUIC arm does NOT inherit the socket arm's blocking read. It delimits
whole frames out of the client's inbound buffer and leaves a partial tail
there, so a daemon that stops mid-frame costs the deadline it was given and
not a second more. The debt comment moves to the unix arm, which keeps it
deliberately: over a local socket a mid-frame stall means a daemon that has
stopped running, and a second partial-frame buffer would be paid for a case
that cannot happen there.

The await grace window becomes RTT-aware — max(2s, 4x the handshake),
capped at 30s — because the daemon's timeout window opens a flight after
this process starts counting and closes a flight before the reply lands. On
loopback and LAN the derivation never leaves the old 2s floor.

A lost connection mid-await is the one transport failure retried rather
than reported: reconnect once, re-attach at 0x0, re-issue the await with
the ORIGINAL since_seq, and continue the original deadline. since_seq is
what makes it safe — the request is a question about a watermark, so
re-asking it is the same question.

Measured, loopback: run/status/capture/send/await all valid JSON with marks
spans and output; an await in flight when the daemon is stopped fails in
4.0s of a 20s deadline with QuicHandshakeFailed, which only waitReady
inside the redial can produce; a refused port is a no in under 1ms; an
unbounded `--timeout 0` dial to a silent host still ends, at the 15s idle
timeout. The unix-socket smokes are unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

build.zig

Old		New
@@ -416,9 +416,12 @@ pub fn build(b: *std.Build) void {
416	linkQuic(b, exe, quic);	416	linkQuic(b, exe, quic);
417	b.installArtifact(exe);	417	b.installArtifact(exe);
418		418
419	// The agent-facing client. Protocol and nothing else: it speaks frames	419	// The agent-facing client. It speaks frames and owns no terminal, which
420	// over the unix socket and owns no terminal, which is the whole point —	420	// is the whole point — it attaches at 0x0 and never claims the grid.
421	// it attaches at 0x0 and never claims the grid.	421	// The transport modules are the CLI client's, minus everything that
		422	// renders: `quic_client` for the remote arm and `xdg` for the one
		423	// key-resolution rule all three binaries obey. Deliberately still no
		424	// engine and no replica — muxa has nothing to draw.
422	const muxa_mod = b.createModule(.{	425	const muxa_mod = b.createModule(.{
423	.root_source_file = b.path("src/muxa.zig"),	426	.root_source_file = b.path("src/muxa.zig"),
424	.target = target,	427	.target = target,
@@ -427,6 +430,8 @@ pub fn build(b: *std.Build) void {
427	});	430	});
428	muxa_mod.addImport("protocol", protocol_mod);	431	muxa_mod.addImport("protocol", protocol_mod);
429	muxa_mod.addImport("sockpath", sockpath_mod);	432	muxa_mod.addImport("sockpath", sockpath_mod);
		433	muxa_mod.addImport("quic_client", quic_client_mod);
		434	muxa_mod.addImport("xdg", xdg_mod);
430		435
431	const mux_exe = b.addExecutable(.{ .name = "mux", .root_module = mux_mod });	436	const mux_exe = b.addExecutable(.{ .name = "mux", .root_module = mux_mod });
432	mux_exe.use_llvm = true;	437	mux_exe.use_llvm = true;
@@ -438,6 +443,8 @@ pub fn build(b: *std.Build) void {
438	const muxa_exe = b.addExecutable(.{ .name = "muxa", .root_module = muxa_mod });	443	const muxa_exe = b.addExecutable(.{ .name = "muxa", .root_module = muxa_mod });
439	muxa_exe.use_llvm = true;	444	muxa_exe.use_llvm = true;
440	muxa_exe.use_lld = true;	445	muxa_exe.use_lld = true;
		446	// The agent client dials remote daemons now, so it carries the stack too.
		447	linkQuic(b, muxa_exe, quic);
441	b.installArtifact(muxa_exe);	448	b.installArtifact(muxa_exe);
442		449
443	const rawmode_exe = b.addExecutable(.{ .name = "rawmode", .root_module = rawmode_mod });	450	const rawmode_exe = b.addExecutable(.{ .name = "rawmode", .root_module = rawmode_mod });
@@ -585,7 +592,7 @@ pub fn build(b: *std.Build) void {
585	if (mod == server_mod or mod == quic_mod or mod == quic_server_mod or	592	if (mod == server_mod or mod == quic_mod or mod == quic_server_mod or
586	mod == exe_mod or mod == client_mod or mod == mux_mod or	593	mod == exe_mod or mod == client_mod or mod == mux_mod or
587	mod == quic_client_mod or mod == webhub_mod or	594	mod == quic_client_mod or mod == webhub_mod or
588	mod == webhub_main_mod) linkQuic(b, t, quic);	595	mod == muxa_mod or mod == webhub_main_mod) linkQuic(b, t, quic);
589	test_step.dependOn(&b.addRunArtifact(t).step);	596	test_step.dependOn(&b.addRunArtifact(t).step);
590	}	597	}
591		598

src/muxa.zig

@@ @@ -5,9 +5,12 @@ @@
+const std = @import("std");
+const proto = @import("protocol");
+const sockpath = @import("sockpath");
+const quic_client = @import("quic_client");
+const xdg = @import("xdg");
+const usage =
+    \\usage: muxa <verb> [--sock PATH | --quic HOST[:PORT] [--key PATH]]
+    \\                   [--settle MS] [--timeout MS] [--vt] [args]
+    \\verbs:
+    \\  status              session snapshot as JSON
+    \\  capture             current grid as text (--vt for styled)
@@ @@ -20,6 +23,15 @@ const usage = @@
+const Opts = struct {
+    verb: enum { status, capture, send, run, @"await" },
+    sock: ?[]const u8 = null,
+    /// `HOST[:PORT]` of a remote daemon's QUIC listener. The verbs are
+    /// identical over it — same frames, same JSON — which is the whole
+    /// claim: an agent driving a session over a WAN types one more flag.
+    quic: ?[]const u8 = null,
+    /// `--key PATH`, the highest-priority spelling of the QUIC key. Null
+    /// does NOT mean "no key": `$MUX_KEY_FILE` and the XDG default are
+    /// still to be tried, and neither is parse's to look at (xdg.pickKey
+    /// and xdg.resolveKeyPath own that order, as they do for muxd and mux).
+    key: ?[]const u8 = null,
+    settle_ms: u32 = 0,
+    // Never 0 by default: the daemon reads a 0 timeout on await_req as "no
+    // bound at all" (documented on AwaitReq), so a muxa that defaulted to 0
@@ @@ -50,6 +62,14 @@ fn parseArgs(args: []const [:0]const u8) ?Opts { @@
+            i += 1;
+            if (i >= args.len) return null;
+            o.sock = args[i];
+        } else if (std.mem.eql(u8, a, "--quic")) {
+            i += 1;
+            if (i >= args.len) return null;
+            o.quic = args[i];
+        } else if (std.mem.eql(u8, a, "--key")) {
+            i += 1;
+            if (i >= args.len) return null;
+            o.key = args[i];
+        } else if (std.mem.eql(u8, a, "--settle")) {
+            i += 1;
+            if (i >= args.len) return null;
@@ @@ -64,6 +84,15 @@ fn parseArgs(args: []const [:0]const u8) ?Opts { @@
+            o.arg = a;
+        } else return null;
+    // Name ONE transport. A `--sock` silently ignored beside a `--quic`
+    // would send an agent's frames somewhere other than the socket it
+    // named, and the two answers differ — this is the mistake `mux`
+    // refuses as `.conflict` for the same reason.
+    if (o.quic != null and o.sock != null) return null;
+    // A key with nothing to authenticate to, refused exactly where muxd
+    // refuses it: there is no reading of `--key` without `--quic` that
+    // makes it sensible, and the unix socket has no key at all.
+    if (o.key != null and o.quic == null) return null;
+    return o;
+    try std.testing.expectEqual(@as(?Opts, null), parseArgs(&two));
+test "parseArgs: --quic and --key, and the pairs that make no sense" {
+    const q = [_][:0]const u8{ "muxa", "status", "--quic", "10.0.0.2:4433" };
+    const oq = parseArgs(&q).?;
+    try std.testing.expectEqualStrings("10.0.0.2:4433", oq.quic.?);
+    // Not naming a key is not an error here: MUX_KEY_FILE and the XDG
+    // default are still to be tried, and parse may look at neither.
+    try std.testing.expectEqual(@as(?[]const u8, null), oq.key);
+    const k = [_][:0]const u8{ "muxa", "run", "--quic", "box:4433", "--key", "/k", "make test" };
+    const ok = 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.?);
+    // A flag at the end of argv with no value is refused, like every other
+    // value-taking flag this parser has.
+    const dangling_q = [_][:0]const u8{ "muxa", "status", "--quic" };
+    try std.testing.expectEqual(@as(?Opts, null), parseArgs(&dangling_q));
+    const dangling_k = [_][:0]const u8{ "muxa", "status", "--quic", "b:1", "--key" };
+    try std.testing.expectEqual(@as(?Opts, null), parseArgs(&dangling_k));
+    // Two transports named at once: which one an agent's frames went to
+    // would be this parser's private business, and it is not entitled to
+    // one — the same refusal `mux` spells as `.conflict`.
+    const both = [_][:0]const u8{ "muxa", "status", "--sock", "/tmp/s", "--quic", "b:1" };
+    try std.testing.expectEqual(@as(?Opts, null), parseArgs(&both));
+    // A key with nothing to authenticate to, refused exactly where muxd
+    // refuses it.
+    const lonely_key = [_][:0]const u8{ "muxa", "status", "--key", "/k" };
+    try std.testing.expectEqual(@as(?Opts, null), parseArgs(&lonely_key));
+    // Neither named is the ordinary local case and stays silent.
+    const neither = [_][:0]const u8{ "muxa", "status" };
+    try std.testing.expectEqual(@as(?[]const u8, null), parseArgs(&neither).?.quic);
+}
+/// A live QUIC connection plus everything a REDIAL of it needs. The dial
+/// coordinates are kept rather than re-derived because the reconnect below
+/// happens mid-verb, long after argv and the key file have been read: a
+/// second resolution could pick a different key (the file having been
+/// rotated under us) and would then fail the handshake for a reason that
+/// has nothing to do with why the first connection died.
+const Quic = struct {
+    cl: *quic_client.Client,
+    addr: std.net.Address,
+    key: quic_client.Key,
+    idle_ms: u32,
+    /// Wall-clock milliseconds the FIRST handshake took, which is this
+    /// client's only measurement of how far away the daemon is. `graceMs`
+    /// turns it into the await grace window; see there.
+    connect_ms: i64,
+};
+const Conn = struct {
+    /// Which transport carries the frames. The verbs above this line are
+    /// written once and know nothing about the difference — that is the
+    /// claim `--quic` makes, and this union is where it is kept.
+    link: union(enum) {
+        fd: std.posix.fd_t,
+        quic: Quic,
+    },
+    /// The allocator the transport itself works with: the QUIC arm's frame
+    /// staging and its redials. Distinct from the `alloc` awaitFrame takes,
+    /// which owns the frame handed BACK to the caller — one process, one
+    /// arena, so they are the same allocator today and separate in the
+    /// signature because they answer to different owners.
+    alloc: std.mem.Allocator,
+    /// The code from the `exit_status` frame that ended a wait, set the
+    /// moment awaitFrame returns error.SessionExited. The 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; callers read
+    /// it to turn the error into an answer.
+    session_exit: ?u8 = null,
+    /// One reconnect per process, spent or not. See `awaitReissuing`.
+    reconnected: bool = false,
+    fn open(alloc: std.mem.Allocator, sock_path: []const u8) !Conn {
+        const s = try std.net.connectUnixSocket(sock_path);
+        return .{ .link = .{ .fd = s.handle }, .alloc = alloc };
+    }
+    /// Dial a daemon's QUIC listener and wait out the handshake before
+    /// returning. The wait is not optional and not the caller's: `connect`
+    /// only creates state — the first flight has not been answered — and a
+    /// `send` on a connection with no stream yet accepts zero bytes and
+    /// says so by returning 0, which would surface as a frame that
+    /// silently never left. Same reason client.zig's quicTransport waits.
+    fn openQuic(
+        alloc: std.mem.Allocator,
+        addr: std.net.Address,
+        key: quic_client.Key,
+        idle_ms: u32,
+        deadline_ms: i64,
+    ) !Conn {
+        const started = std.time.milliTimestamp();
+        const cl = try quic_client.Client.connect(alloc, addr, key, idle_ms);
+        errdefer cl.deinit();
+        try waitReady(cl, deadline_ms);
+        return .{
+            .link = .{ .quic = .{
+                .cl = cl,
+                .addr = addr,
+                .key = key,
+                .idle_ms = idle_ms,
+                .connect_ms = elapsed(started),
+            } },
+            .alloc = alloc,
+        };
+    fn close(self: *Conn) void {
+        switch (self.link) {
+            .fd => |fd| std.posix.close(fd),
+            .quic => self.link.quic.cl.deinit(),
+        }
+    }
+    /// How much longer than the daemon this client waits for an await,
+    /// which over a network is a function of how far away the daemon is.
+    ///
+    /// The unix arm keeps the flat 2s (see await_grace_ms). The QUIC arm
+    /// adds nothing until four round trips of its own handshake exceed
+    /// that, which on a LAN or loopback is never and on a 200ms link is
+    /// most of a second: the daemon's timeout window opens when it READS
+    /// the request, a whole flight after this process started counting,
+    /// and closes a flight before the reply lands. Four, not two, because
+    /// the request and the reply are not the only flights in the trip —
+    /// the daemon may be settling a command when the timeout fires.
+    ///
+    /// Capped, because `connect_ms` is bounded only by the handshake wait:
+    /// a connection that took fifteen seconds to come up would otherwise
+    /// buy a minute of grace, and past this cap we are no longer waiting
+    /// for the daemon's answer but for a network that has already shown it
+    /// cannot carry one.
+    fn graceMs(self: *const Conn) i64 {
+        return switch (self.link) {
+            .fd => await_grace_ms,
+            .quic => |q| @min(grace_cap_ms, @max(await_grace_ms, 4 * q.connect_ms)),
+        };
+    fn sendFrame(self: *Conn, t: proto.MsgType, payload: []const u8) !void {
+        switch (self.link) {
+            .fd => |fd| try proto.writeFrame(fd, t, payload),
+            .quic => try self.sendFrameQuic(t, payload),
+        }
+    }
+    /// The frame's wire bytes into the egress ring, all of them.
+    ///
+    /// `send` takes what fits and reports how much (a bounded ring: the
+    /// caller holds the backlog), so a short take is not a failure and not
+    /// ignorable either — the tail is offered again once acks have made
+    /// room. muxa's frames are a handful of bytes against a 256KB ring, so
+    /// this loop is expected never to turn twice; it is here because the
+    /// alternative to looping is a frame that leaves half-written, which
+    /// the peer reads as a corrupt stream rather than as an error.
+    fn sendFrameQuic(self: *Conn, t: proto.MsgType, payload: []const u8) !void {
+        var buf: std.ArrayList(u8) = .empty;
+        defer buf.deinit(self.alloc);
+        try proto.appendFrame(&buf, self.alloc, t, payload);
+        const q = &self.link.quic;
+        const deadline = std.time.milliTimestamp() + send_flush_ms;
+        var off: usize = 0;
+        while (off < buf.items.len) {
+            if (q.cl.dead) return error.ConnectionLost;
+            off += q.cl.send(buf.items[off..]);
+            if (off == buf.items.len) return;
+            if (std.time.milliTimestamp() >= deadline) return error.SendStalled;
+            // The ring is full: only the peer's acks can empty it, and they
+            // arrive through pump. Polling first keeps this from spinning.
+            var fds = [_]std.posix.pollfd{
+                .{ .fd = q.cl.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
+            };
+            _ = std.posix.poll(&fds, q.cl.timeoutMs(50)) catch return error.ConnectionLost;
+            q.cl.pump();
+        }
+    /// Read frames until one of type `want` arrives (snapshots, deltas and
@@ @@ -194,28 +382,44 @@ const Conn = struct { @@
+    /// the reply we are waiting for is never coming, and the reason is an
+    /// answer — the session ran its last command — not a transport
+    /// failure. Callers get error.SessionExited plus `session_exit`.
+    fn awaitFrame(
+        self: *Conn,
+        alloc: std.mem.Allocator,
+        want: proto.MsgType,
+        deadline_ms: i64,
+    ) !proto.Frame {
+        return switch (self.link) {
+            .fd => self.awaitFrameFd(alloc, want, deadline_ms),
+            .quic => self.awaitFrameQuic(alloc, want, deadline_ms),
+        };
+    }
+    /// Debt, deliberately retained on THIS arm: only the wait is
+    /// deadline-bounded, not the read. Once poll says a frame has begun,
+    /// readFrame's readExact blocks until the whole payload lands, so a
+    /// peer that stalls mid-frame outlives the deadline. That is harmless
+    /// over a local socket, where the daemon writes whole frames at once
+    /// and a stall means a daemon that has stopped running rather than a
+    /// path that has stopped delivering — and buying it off would mean a
+    /// second partial-frame buffer for a case that cannot happen here.
+    /// The QUIC arm below, where a network IS under the transport, does
+    /// not have the luxury and does not take it.
+    fn awaitFrameFd(
+        self: *Conn,
+        alloc: std.mem.Allocator,
+        want: proto.MsgType,
+        deadline_ms: i64,
+    ) !proto.Frame {
+        const fd = self.link.fd;
+        while (true) {
+            const now = std.time.milliTimestamp();
+            if (now >= deadline_ms) return error.Timeout;
+            var fds = [_]std.posix.pollfd{
+                .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
+            };
+            const n = try std.posix.poll(&fds, @intCast(@min(deadline_ms - now, 250)));
+            if (n == 0) continue;
+            const frame = try proto.readFrame(alloc, fd) orelse return error.DaemonGone;
+            if (frame.type == want) return frame;
+            defer frame.deinit(alloc);
+            if (frame.type == .exit_status) {
@@ @@ -226,8 +430,260 @@ const Conn = struct { @@
+    /// The same wait with a network under it, and the difference is that
+    /// NOTHING here blocks on the transport: a datagram carries whatever
+    /// arrived, whole frames or a third of one, so the frames are
+    /// delimited out of the client's inbound buffer and a partial tail
+    /// simply stays there until the rest lands. A daemon that stops
+    /// mid-frame costs this loop the deadline it was given and not a
+    /// second more.
+    ///
+    /// Every buffered frame is taken before the next poll — a datagram
+    /// routinely carries several, and the reply may be the second — and
+    /// `dead` is checked only once the buffer is empty, so bytes that
+    /// arrived before the connection died are still delivered.
+    fn awaitFrameQuic(
+        self: *Conn,
+        alloc: std.mem.Allocator,
+        want: proto.MsgType,
+        deadline_ms: i64,
+    ) !proto.Frame {
+        const q = &self.link.quic;
+        while (true) {
+            q.cl.pump();
+            while (try frameFrom(alloc, q.cl.inbound())) |got| {
+                q.cl.consume(got.consumed);
+                if (got.frame.type == want) return got.frame;
+                defer got.frame.deinit(alloc);
+                if (got.frame.type == .exit_status) {
+                    self.session_exit = if (got.frame.payload.len >= 1) got.frame.payload[0] else null;
+                    return error.SessionExited;
+                }
+            }
+            // Not `DaemonGone`: over a network the difference between "the
+            // daemon exited" and "the path to it went away" is not ours to
+            // claim, and the reconnect above only fires on this one.
+            if (q.cl.dead) return error.ConnectionLost;
+            const now = std.time.milliTimestamp();
+            if (now >= deadline_ms) return error.Timeout;
+            var fds = [_]std.posix.pollfd{
+                .{ .fd = q.cl.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
+            };
+            // Through timeoutMs, so ngtcp2's own timers — loss detection
+            // and, on a quiet await, the keepalive that keeps the idle
+            // timeout from firing under us — are serviced on schedule
+            // rather than whenever the daemon happens to say something.
+            const cap: i32 = @intCast(@min(deadline_ms - now, 250));
+            _ = std.posix.poll(&fds, q.cl.timeoutMs(cap)) catch return error.ConnectionLost;
+        }
+    }
+    /// Redial the same coordinates and hand the connection over. The old
+    /// client is torn down only once the new one is up, so a redial that
+    /// fails leaves this Conn holding a live (if dead-ended) client rather
+    /// than a freed one — `close` runs either way.
+    fn reconnect(self: *Conn, deadline_ms: i64) !void {
+        const q = &self.link.quic;
+        const cl = try quic_client.Client.connect(self.alloc, q.addr, q.key, q.idle_ms);
+        errdefer cl.deinit();
+        try waitReady(cl, deadline_ms);
+        q.cl.deinit();
+        q.cl = cl;
+        self.reconnected = true;
+    }
+};
+/// One frame delimited out of `buf`, and how many bytes of it that took.
+/// Null while the tail is still partial — a header that has not all
+/// arrived, or a payload still in flight — which is the ordinary state of
+/// a byte stream and never an error.
+///
+/// Takes a plain slice rather than the client, so the delimiting can be
+/// exercised against a canned buffer with no connection anywhere. The
+/// shape is the daemon's `pushInbound` walk, which delimits the same
+/// frames off the same wire from the other end.
+fn frameFrom(
+    alloc: std.mem.Allocator,
+    buf: []const u8,
+) !?struct { frame: proto.Frame, consumed: usize } {
+    if (buf.len < proto.frame_header_len) return null;
+    const len = std.mem.readInt(u32, buf[1..5], .little);
+    // A length no frame can legitimately carry: the stream is not what we
+    // think it is, and reading on would allocate against a number the peer
+    // chose. Same bound and same verdict as the daemon's walk.
+    if (len > proto.max_payload) return error.FrameTooLarge;
+    if (buf.len < proto.frame_header_len + len) return null;
+    const payload = try alloc.alloc(u8, len);
+    errdefer alloc.free(payload);
+    @memcpy(payload, buf[proto.frame_header_len..][0..len]);
+    return .{
+        .frame = .{ .type = @enumFromInt(buf[0]), .payload = payload },
+        .consumed = proto.frame_header_len + len,
+    };
+}
+test "frameFrom: a partial tail is not a frame and not an error" {
+    const alloc = std.testing.allocator;
+    // Nothing, and less than a header: the two shapes a datagram that
+    // carried the start of a frame leaves behind.
+    try std.testing.expect(try frameFrom(alloc, "") == null);
+    try std.testing.expect(try frameFrom(alloc, &[_]u8{ 0x0a, 1, 0 }) == null);
+    // A whole header whose payload is still in flight. This is the case a
+    // blocking read would have sat on: the length is known, the bytes are
+    // not here, and the answer is to wait rather than to read.
+    const partial = [_]u8{ @intFromEnum(proto.MsgType.input), 4, 0, 0, 0, 'a', 'b' };
+    try std.testing.expect(try frameFrom(alloc, &partial) == null);
+    // The same bytes, completed.
+    const whole = [_]u8{ @intFromEnum(proto.MsgType.input), 4, 0, 0, 0, 'a', 'b', 'c', 'd' };
+    const got = (try frameFrom(alloc, &whole)).?;
+    defer got.frame.deinit(alloc);
+    try std.testing.expectEqual(proto.MsgType.input, got.frame.type);
+    try std.testing.expectEqualStrings("abcd", got.frame.payload);
+    try std.testing.expectEqual(@as(usize, 9), got.consumed);
+}
+test "frameFrom: two frames in one buffer, walked by consumed" {
+    const alloc = std.testing.allocator;
+    // What a single datagram routinely carries: the push we skip and the
+    // reply we asked for. A walk that stopped after one would leave the
+    // answer sitting in the buffer while the deadline ran out.
+    var buf: std.ArrayList(u8) = .empty;
+    defer buf.deinit(alloc);
+    try proto.appendFrame(&buf, alloc, .pty_mode, &[_]u8{0});
+    try proto.appendFrame(&buf, alloc, .status_reply, "xy");
+    const first = (try frameFrom(alloc, buf.items)).?;
+    defer first.frame.deinit(alloc);
+    try std.testing.expectEqual(proto.MsgType.pty_mode, first.frame.type);
+    const second = (try frameFrom(alloc, buf.items[first.consumed..])).?;
+    defer second.frame.deinit(alloc);
+    try std.testing.expectEqual(proto.MsgType.status_reply, second.frame.type);
+    try std.testing.expectEqualStrings("xy", second.frame.payload);
+    try std.testing.expectEqual(buf.items.len, first.consumed + second.consumed);
+    // An empty payload is a frame like any other — `status_req` and
+    // `detach` are nothing else — and must not read as "nothing yet".
+    var empty: std.ArrayList(u8) = .empty;
+    defer empty.deinit(alloc);
+    try proto.appendFrame(&empty, alloc, .detach, "");
+    const none = (try frameFrom(alloc, empty.items)).?;
+    defer none.frame.deinit(alloc);
+    try std.testing.expectEqual(@as(usize, proto.frame_header_len), none.consumed);
+}
+test "frameFrom: a length no frame can carry is refused, not allocated" {
+    const alloc = std.testing.allocator;
+    // The peer chose this number. Reading on would mean allocating against
+    // it; the daemon's own walk refuses the same bound the same way.
+    var hdr: [proto.frame_header_len]u8 = undefined;
+    hdr[0] = @intFromEnum(proto.MsgType.input);
+    std.mem.writeInt(u32, hdr[1..5], proto.max_payload + 1, .little);
+    try std.testing.expectError(error.FrameTooLarge, frameFrom(alloc, &hdr));
+}
+test "graceMs: flat over a socket, RTT-derived over QUIC, and capped" {
+    const alloc = std.testing.allocator;
+    const local = Conn{ .link = .{ .fd = -1 }, .alloc = alloc };
+    try std.testing.expectEqual(@as(i64, 2_000), local.graceMs());
+    // The derivation is 4x the handshake, and it only ever WIDENS the
+    // window: a loopback or LAN daemon keeps the flat 2s.
+    // No client: the window is a function of the measurement, not of the
+    // connection, and nothing here may touch one.
+    var far = Conn{
+        .link = .{ .quic = .{
+            .cl = undefined,
+            .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.link.quic.connect_ms = 300;
+    try std.testing.expectEqual(@as(i64, 2_000), far.graceMs());
+    far.link.quic.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.link.quic.connect_ms = 60_000;
+    try std.testing.expectEqual(@as(i64, 30_000), far.graceMs());
+}
+/// Drive a fresh connection until it can carry bytes, or give up.
+///
+/// A refused port ends this early — quic_client turns the ICMP unreachable
+/// into `dead` — so the common mistake (no daemon on that port) costs
+/// milliseconds. A blackholed one produces no error at all, and there the
+/// deadline is the only thing that ends the wait; even an unbounded one
+/// (`--timeout 0`) terminates, because the connection's own idle timeout
+/// kills it after `idle_ms`.
+fn waitReady(cl: *quic_client.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;
+    // 127.0.0.1:1, where nothing listens: the refusal is REAL — an ICMP
+    // unreachable comes back and quic_client acts on it — which is what
+    // lets this exercise the whole redial path (dial, handshake wait,
+    // verdict) in a couple of loopback round trips instead of a timeout.
+    const addr = try std.net.Address.parseIp("127.0.0.1", 1);
+    const key: quic_client.Key = .{ .bytes = [_]u8{7} ** quic_client.key_len };
+    // The dial that stands in for the connection this client had before
+    // the network went away. It dies for the same reason the redial will,
+    // which is fine: what is under test is what `reconnect` DOES, and it
+    // does the same thing to a connection that died at second 30.
+    const deadline = std.time.milliTimestamp() + 2_000;
+    var conn = Conn{
+        .link = .{ .quic = .{
+            .cl = try quic_client.Client.connect(alloc, addr, key, 1_000),
+            .addr = addr,
+            .key = key,
+            .idle_ms = 1_000,
+            .connect_ms = 0,
+        } },
+        .alloc = alloc,
+    };
+    defer conn.close();
+    const t0 = std.time.milliTimestamp();
+    try std.testing.expectError(error.QuicHandshakeFailed, conn.reconnect(deadline));
+    // Fast, because the port refused rather than went quiet. A redial that
+    // swallowed the refusal would spend the whole 2s here — and in the
+    // field it would spend the agent's remaining deadline.
+    try std.testing.expect(std.time.milliTimestamp() - t0 < 1_000);
+    // A redial that failed is not a reconnect spent — but it is also not a
+    // Conn 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.reconnected);
+}
+test "awaitFrame ends a wait on exit_status, keeping the code" {
+    const alloc = std.testing.allocator;
+    // A pipe stands in for the daemon: awaitFrame polls and reads an fd and
+    defer std.posix.close(pipe[0]);
+    defer std.posix.close(pipe[1]);
+    var conn = Conn{ .link = .{ .fd = pipe[0] }, .alloc = alloc };
+    // 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});
@@ @@ -349,6 +805,22 @@ pub fn main() !u8 { @@
+        return 2;
+    };
+    // Started BEFORE the connect, not after: over QUIC the handshake is
+    // part of the round trip the caller bounded, and a `--timeout` that
+    // began counting only once the connection was up would promise
+    // something different on the two transports. Over a unix socket the
+    // connect is a syscall, so this moves the instant by microseconds.
+    const deadline = deadlineFor(o.timeout_ms);
+    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 try sockpath.defaultSockPath(alloc);
+    // Refused by name, before connecting: connect would bounce a too-long
@@ @@ -364,7 +836,7 @@ pub fn main() !u8 { @@
+        return fail("socket path too long", detail);
+    var conn = Conn.open(alloc, sock_path) catch |e| {
+        // The path goes in the detail: a muxa pointed at the wrong socket
+        // is this binary's likeliest field failure, and an agent reading
+        // "FileNotFound" alone cannot tell which path it was that missed.
@@ @@ -375,16 +847,144 @@ pub fn main() !u8 { @@
+    };
+    defer conn.close();
+    return dispatch(alloc, &conn, o, deadline);
+}
+/// The verbs, once. Both transports arrive here with a Conn and nothing
+/// else that distinguishes them, which is the property `--quic` is selling.
+fn dispatch(alloc: std.mem.Allocator, conn: *Conn, o: Opts, deadline: i64) !u8 {
+    return switch (o.verb) {
+        .status => verbStatus(alloc, conn, deadline),
+        .capture => verbCapture(alloc, conn, o.vt, deadline),
+        .send => verbSend(alloc, conn, o.arg, deadline),
+        .run => verbRun(alloc, conn, o, deadline),
+        .@"await" => verbAwait(alloc, conn, o, deadline),
+    };
+/// A QUIC transport, or the exit code standing in for the reason there is
+/// not one. Every refusal here goes through `fail`, so a dial that never
+/// happened prints the same one-JSON-object-on-stdout shape as a verb that
+/// ran — an agent parses one thing whatever went wrong.
+const Opened = union(enum) { conn: Conn, exit: u8 };
+fn openQuicConn(
+    alloc: std.mem.Allocator,
+    o: Opts,
+    host_port: []const u8,
+    deadline: i64,
+) Opened {
+    // `--key`, then `$MUX_KEY_FILE`, then the XDG default if it exists.
+    // The order is not spelled here on purpose: xdg owns it, muxd and mux
+    // read it from the same two functions, and a third copy is how two
+    // binaries end up authenticating with different keys.
+    const res = xdg.resolveKeyPath(alloc, xdg.pickKey(o.key, std.posix.getenv("MUX_KEY_FILE"))) catch |e|
+        return .{ .exit = fail("quic: cannot resolve a key path", @errorName(e)) };
+    const key_path = switch (res) {
+        .given, .default => |p| p,
+        // The path is the detail because it is the actionable half: the
+        // agent (or the human reading its log) needs to know which file
+        // `muxd keygen` was supposed to have written.
+        .missing => |p| return .{ .exit = fail(
+            "quic: no key: pass --key, set MUX_KEY_FILE, or run `muxd keygen`",
+            p,
+        ) },
+    };
+    const key = quic_client.Key.load(key_path) catch |e| {
+        // The daemon's words for a key the daemon would also refuse —
+        // including the group/other-readable refusal, which this binary
+        // gets for free by loading the key the same way.
+        var buf: [quic_client.key_refusal_len]u8 = undefined;
+        return .{ .exit = fail("quic: unusable key", quic_client.keyRefusalBody(&buf, e, key_path)) };
+    };
+    const addr = parseQuicAddr(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 = Conn.openQuic(alloc, addr, key, quic_client.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 };
+}
+/// `HOST[:PORT]`, with an omitted port meaning `quic_client.default_port`.
+/// A name is resolved rather than refused: unlike muxd's `--quic`, which
+/// names an address to BIND, this one names a box to reach, and a box is
+/// normally spelled with a name.
+///
+/// The grammar is the CLI client's — `mux quic://HOST:PORT` accepts these
+/// same spellings, brackets and all, and the two must not diverge: an
+/// agent and a human pointing at the same daemon type the same thing. It
+/// is a second copy of client.zig's `parseQuicAddr` and knowingly so:
+/// folding both into the shared `quic` module is the right home for it and
+/// is a change to the CLI client, which this one is not.
+fn parseQuicAddr(alloc: std.mem.Allocator, host_port: []const u8) !std.net.Address {
+    // `[::1]` — bracketed and portless: the brackets say where the address
+    // stops, so the port can default.
+    if (host_port.len >= 2 and host_port[0] == '[' and host_port[host_port.len - 1] == ']')
+        return resolveHost(alloc, host_port[1 .. host_port.len - 1], quic_client.default_port);
+    const colon = std.mem.lastIndexOfScalar(u8, host_port, ':') orelse
+        return resolveHost(alloc, host_port, quic_client.default_port);
+    var host = host_port[0..colon];
+    const port_s = host_port[colon + 1 ..];
+    if (host.len >= 2 and host[0] == '[' and host[host.len - 1] == ']') {
+        host = host[1 .. host.len - 1];
+    } else if (std.mem.indexOfScalar(u8, host, ':') != null) {
+        // An unbracketed IPv6 literal would have its last group taken as a
+        // port. Refused rather than guessed at.
+        return error.MalformedAddress;
+    }
+    const port = std.fmt.parseInt(u16, port_s, 10) catch return error.MalformedAddress;
+    return resolveHost(alloc, host, port);
+}
+fn resolveHost(alloc: std.mem.Allocator, host: []const u8, port: u16) !std.net.Address {
+    if (host.len == 0) return error.MalformedAddress;
+    if (std.net.Address.parseIp(host, port)) |addr| return addr else |_| {}
+    const list = try std.net.getAddressList(alloc, host, port);
+    defer list.deinit();
+    if (list.addrs.len == 0) return error.UnknownHostName;
+    return list.addrs[0];
+}
+test "parseQuicAddr: literals, brackets, and the spellings that are refused" {
+    const alloc = std.testing.allocator;
+    // Literals only here: a name would send this test to a resolver, and
+    // what it answered would depend on the machine running it.
+    try std.testing.expectEqual(
+        @as(u16, 4433),
+        (try parseQuicAddr(alloc, "127.0.0.1:4433")).getPort(),
+    );
+    // An omitted port means mux's own. Spelled out rather than written
+    // `quic_client.default_port`, because comparing the parse's answer
+    // against the constant the parse reads would hold for any value and
+    // say nothing about the port — and this is the number the daemon at
+    // the other end has to agree on.
+    try std.testing.expectEqual(@as(u16, 4433), (try parseQuicAddr(alloc, "10.0.0.2")).getPort());
+    const six = try parseQuicAddr(alloc, "[::1]:9999");
+    try std.testing.expectEqual(@as(u16, 9999), six.getPort());
+    try std.testing.expect(six.any.family == std.posix.AF.INET6);
+    // Bracketed and portless: the brackets say where the address stops, so
+    // the port can default.
+    try std.testing.expectEqual(@as(u16, 4433), (try parseQuicAddr(alloc, "[::1]")).getPort());
+    // An unbracketed IPv6 literal would have its last group read as a
+    // port. Refused rather than guessed at — the same refusal muxd's
+    // splitHostPort makes about its bind address.
+    try std.testing.expectError(error.MalformedAddress, parseQuicAddr(alloc, "fe80::1:4433"));
+    try std.testing.expectError(error.MalformedAddress, parseQuicAddr(alloc, "127.0.0.1:"));
+    try std.testing.expectError(error.MalformedAddress, parseQuicAddr(alloc, "127.0.0.1:99999"));
+    try std.testing.expectError(error.MalformedAddress, parseQuicAddr(alloc, ""));
+    try std.testing.expectError(error.MalformedAddress, parseQuicAddr(alloc, ":4433"));
+}
+fn verbStatus(alloc: std.mem.Allocator, conn: *Conn, deadline: i64) !u8 {
+    conn.sendFrame(.status_req, "") catch |e| return fail("status: send failed", @errorName(e));
+    const frame = conn.awaitFrame(alloc, .status_reply, deadline) catch |e| switch (e) {
+/// `{"reason":"timeout"}` with exit 3 that the agent is meant to read.
+const await_grace_ms = 2_000;
+/// The ceiling on the QUIC arm's derived grace (Conn.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 `sendFrameQuic` will keep offering a frame's tail to a full
+/// egress ring before giving up on it. Reaching this means the peer has
+/// stopped acknowledging 256KB of backlog, which is a dead connection
+/// wearing a different hat — but a bound is what keeps it from being an
+/// unbounded wait inside a call that has no deadline of its own.
+const send_flush_ms = 5_000;
+/// 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
@@ @@ -547,6 +1159,52 @@ fn doAwait( @@
+    return try proto.decodeAwaitReply(frame.payload);
+/// The await, plus the ONE reconnect this client is willing to spend on it.
+///
+/// A wait is the only round trip long enough for a network to die under —
+/// a status round trip is over in a millisecond, a `run` on a build is not
+/// — and losing it costs an agent the whole command it was watching, so
+/// this is the one place a transport failure is retried rather than
+/// reported. What makes the retry safe rather than a second command is
+/// `since_seq`: the request is a question about a watermark ("tell me
+/// about a return newer than this"), so re-asking it after a reconnect is
+/// the SAME question and the daemon answers it identically whether or not
+/// it saw the first one. The server's own tests pin that idempotency.
+///
+/// Three things are deliberately not reset:
+///
+///   * the deadline, which is the caller's whole bound and continues
+///     across the reconnect — a redial that ate four seconds has spent
+///     four seconds of the wait, not bought a fresh one;
+///   * `since_seq`, for the reason above — re-reading the watermark from
+///     the new connection would move it past a return that had happened
+///     while we were disconnected, and the await would then sit waiting
+///     for one that already went by;
+///   * the attach, which IS re-sent, at 0x0 like every other attach this
+///     binary makes: the daemon dropped our old client slot with the
+///     connection and would have no session to answer about otherwise.
+///
+/// Once, and once per process rather than per await: a loop here would be
+/// a client that hides a daemon that is gone, and the agent driving it
+/// asked a question that deserves an answer within the deadline it named.
+fn awaitReissuing(
+    alloc: std.mem.Allocator,
+    conn: *Conn,
+    o: Opts,
+    since_seq: u64,
+    deadline: i64,
+) !proto.AwaitReply {
+    return doAwait(alloc, conn, o, since_seq, deadline) catch |e| switch (e) {
+        error.ConnectionLost => {
+            if (conn.link != .quic or conn.reconnected) return e;
+            try conn.reconnect(deadline);
+            try attachZero(conn);
+            return doAwait(alloc, conn, o, since_seq, deadline);
+        },
+        else => e,
+    };
+}
+/// The session's RETURN WATERMARK: the seq of the last command return, 0 if
+/// none. Handed straight to `since_seq`, where it means "only a return
+/// newer than this may answer me".
+        error.SessionExited => return reportSessionEnded(alloc, conn.session_exit, elapsed(started)),
+        else => return fail("await: status failed", @errorName(e)),
+    };
+    const r = awaitReissuing(alloc, conn, o, since, awaitDeadline(o, conn)) catch |e| switch (e) {
+        error.SessionExited => return reportSessionEnded(alloc, conn.session_exit, elapsed(started)),
+        else => return fail("await: no reply", @errorName(e)),
+    };
+    defer alloc.free(line);
+    conn.sendFrame(.input, line) catch |e| return fail("run: input failed", @errorName(e));
+    const r = awaitReissuing(alloc, conn, o, since, awaitDeadline(o, conn)) catch |e| switch (e) {
+        error.SessionExited => return reportSessionEnded(alloc, conn.session_exit, elapsed(started)),
+        else => return fail("run: no reply", @errorName(e)),
+    };
@@ @@ -797,9 +1455,10 @@ fn elapsed(started: i64) i64 { @@
+/// This client's deadline for the await itself — the daemon's own bound
+/// plus the grace window (see await_grace_ms and Conn.graceMs, which is
+/// where the transport gets to widen it). An unbounded request stays
+/// unbounded here too.
+fn awaitDeadline(o: Opts, conn: *const Conn) i64 {
+    if (o.timeout_ms == 0) return std.math.maxInt(i64);
+    return std.time.milliTimestamp() + o.timeout_ms + conn.graceMs();