docs/superpowers/plans/2026-08-31-connection-primitives.md
Ref: Size: 46.4 KiB History
# Connection Primitives Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** One owner each for "a live connection to a daemon" (`src/link.zig`), "wait for one frame of type X" (`Link.awaitFrame`), "bind a unix socket I own" (`src/serve.zig`), and "pump a test daemon to a condition" (`harness.pumpUntil`) — deleting ~10 await-loop copies, 3 divergent binder sites, and 13 spin loops.
**Architecture:** The fd|pipe|quic union is promoted out of `client.Transport` into a shared `link` row; `Transport` and muxa's `AgentConnection` become policy wrappers around it. `serve.zig` owns bind+teardown (guarded unlink for all binders); accept loops stay with their owners. No wire change anywhere.
**Tech Stack:** Zig 0.15.2 (vendored: `deps/zig/zig`, the Makefile points at it — system zig will NOT build this). Linux only.
**Spec:** `docs/superpowers/specs/2026-08-31-connection-primitives-design.md`
## Global Constraints
- Build/test ONLY through the Makefile: `make check` after every task (fmt + unit tests + comment-claim refs), `make ci` at delivery. Capture `$?` before piping (`make test | tail` reports tail's exit).
- **No wire change.** Every byte on every socket identical before and after. No new frame types, no encoder edits.
- `proxy.zig` and the QUIC modules must NEVER import `link` or `term` — the byte-blind invariant.
- Comments say *why*, plainly; `zig build check` gates that cited symbols resolve. When moving code, move its load-bearing comments with it.
- A unit test that writes to fd 1 wedges `zig build test` silently. Tests capture output into pipes if they must.
- Where a thing can be plural, the default test fixture is plural; N=1 is an extra case, not the baseline.
- Commit per step with `--fixup`/`--squash` where amending; autosquash before delivery so history tells the feature's story. The commit-msg hook stamps `Patch:` trailers.
- Any hand-run rig exports isolated `XDG_STATE_HOME` and `XDG_RUNTIME_DIR`.
- The plan's code blocks are the reference; if the vendored Zig rejects a detail (e.g. a comptime fn-param spelling), fix the spelling, keep the contract, and note it in the commit body.
---
### Task 1: `src/link.zig` — the Link union and its mechanics
**Files:**
- Create: `src/link.zig`
- Modify: `build.zig` (module table around line 169, test-rows list around line 720)
**Interfaces:**
- Consumes: `term.protocol` (`writeFrame`, `readFrame`, `takeFrame`, `appendFrame`, `Frame`, `MsgType`), `quic.Client` (`pollFd`, `timeoutMs`, `pump`, `send`, `deinit`, fields `in`, `dead`).
- Produces (later tasks rely on these exact names):
- `pub const Link = union(enum) { fd, pipe, quic }` with `Pipe = { child: std.process.Child, r, w: std.posix.fd_t }` and `Quic = { cl: *quic.Client, qout: std.ArrayList(u8), alloc: std.mem.Allocator }`
- `pub const Incoming = union(enum) { frame: proto.Frame, incomplete, closed }`
- `pollFd(self: *const Link) std.posix.fd_t`
- `timeoutMs(self: *Link, cap_ms: i32) i32`
- `service(self: *Link) void`
- `sendFrame(self: *Link, t: proto.MsgType, payload: []const u8) !void`
- `flushWithin(self: *Link, deadline_ms: u32) !void` (errors: `error.ConnectionLost`, `error.SendStalled`)
- `readFrame(self: *Link, alloc: std.mem.Allocator) !Incoming`
- `close(self: *Link) void` (idempotent)
- [ ] **Step 1: Write `src/link.zig` with its tests included (Zig keeps tests in-file)**
The file. Module header states the contract; bodies below are moved from
`client.zig:694-766` (write/service/timeout/flush/read) with the switch arms
preserved byte-for-byte where they exist today:
```zig
//! The live connection to a daemon, however it was reached: a unix-socket fd,
//! the stdio of a `--via`/handoff child, or a QUIC client. Mechanics only —
//! send a frame, read a frame, wait, close. Policy stays with the owners:
//! redial and backoff are `client.Transport`'s and muxa's, attach semantics
//! are dial's and the callers'. This row exists because the fd|pipe|quic
//! union used to live twice (client.Transport, muxa.AgentConnection), each
//! with its own send, await and close.
const std = @import("std");
const proto = @import("term").protocol;
const quic = @import("quic");
/// Three outcomes, not two: QUIC's socket goes readable for acks and half
/// frames, so `null` cannot keep the socket path's meaning of "peer gone"
/// without making every partial frame a reconnect. (Moved from client.zig.)
pub const Incoming = union(enum) {
frame: proto.Frame,
incomplete,
closed,
};
/// What a non-matching frame does to an `awaitFrame` wait. `on == null` is
/// drop: the frame is freed and the wait continues — the observer-verb
/// policy dial.ask always had. A non-null `on` BORROWS the frame for the
/// duration of the call and must copy anything it keeps; awaitFrame frees
/// the frame when `on` returns. An error out of `on` ends the wait with
/// that error — some "other" frames are answers, not noise (muxa's
/// exit_status), and only the caller knows which.
pub const Sink = struct {
ctx: ?*anyopaque = null,
on: ?*const fn (ctx: ?*anyopaque, frame: proto.Frame) anyerror!void = null,
};
pub const Link = union(enum) {
/// A unix socket: one fd, read and write. -1 once closed — close() is
/// idempotent because a re-dial releases the dead link on entry and an
/// abort then closes the same value again through a defer; a second
/// close(2) on a stale fd is EBADF, which std.posix maps to unreachable.
fd: std.posix.fd_t,
/// `--via`, and the ssh half of a handoff: the child whose stdio IS the
/// transport. `r` is its stdout, `w` its stdin.
pipe: Pipe,
/// The connection that IS the transport; bytes go through the stream
/// layer, so there is nothing to write(2) to.
quic: Quic,
pub const Pipe = struct {
child: std.process.Child,
r: std.posix.fd_t,
w: std.posix.fd_t,
};
pub const Quic = struct {
cl: *quic.Client,
/// Bytes the QUIC ring would not take yet. Frames are appended whole
/// and handed over a prefix at a time, so a short accept can never
/// split one on the wire — the remainder is offered again next pass.
/// Lives here and not in the wrappers because the staging is a
/// property of the quic link, and it used to exist twice
/// (Transport.qout, muxa.sendFrameQuic's stack buffer).
qout: std.ArrayList(u8) = .empty,
alloc: std.mem.Allocator,
};
pub fn pollFd(self: *const Link) std.posix.fd_t {
return switch (self.*) {
.fd => |fd| fd,
.pipe => |p| p.r,
.quic => |q| q.cl.pollFd(),
};
}
/// Folds ngtcp2's next deadline in, so retransmits and idle timeouts
/// happen on time without a second timer.
pub fn timeoutMs(self: *Link, cap_ms: i32) i32 {
return switch (self.*) {
.quic => |*q| q.cl.timeoutMs(cap_ms),
.fd, .pipe => cap_ms,
};
}
/// Unconditional: a QUIC connection's timers are the only thing that
/// notices a peer which stopped answering.
pub fn service(self: *Link) void {
switch (self.*) {
.quic => |*q| {
q.cl.pump();
self.flushQuic();
},
.fd, .pipe => {},
}
}
/// Queue-and-offer, never blocking: fd and pipe write through; quic
/// appends whole and flushes what the ring takes. A caller that must
/// KNOW the bytes left (muxa's verbs) follows with flushWithin.
pub fn sendFrame(self: *Link, t: proto.MsgType, payload: []const u8) !void {
switch (self.*) {
.quic => |*q| {
try proto.appendFrame(&q.qout, q.alloc, t, payload);
self.flushQuic();
},
.fd => |fd| return proto.writeFrame(fd, t, payload),
.pipe => |p| return proto.writeFrame(p.w, t, payload),
}
}
/// Offer the outbound queue to the ring again. After every write and on
/// every service pass, because the room to accept comes from
/// acknowledgements, which arrive on their own schedule.
fn flushQuic(self: *Link) void {
const q = switch (self.*) {
.quic => |*q| q,
.fd, .pipe => return,
};
if (q.qout.items.len == 0) return;
const n = q.cl.send(q.qout.items);
if (n == 0) return;
q.qout.replaceRangeAssumeCapacity(0, n, &.{});
}
/// Drive the staged bytes out or say why not, within the deadline. A
/// no-op for fd/pipe (their sendFrame already either took the bytes or
/// failed). The QUIC arm is muxa's old sendFrameQuic loop: the ring is
/// full, only the peer's acks empty it, and they arrive through pump —
/// polling first keeps this from spinning.
pub fn flushWithin(self: *Link, deadline_ms: u32) !void {
const q = switch (self.*) {
.quic => |*q| q,
.fd, .pipe => return,
};
const end = std.time.milliTimestamp() + deadline_ms;
while (q.qout.items.len != 0) {
if (q.cl.dead) return error.ConnectionLost;
self.flushQuic();
if (q.qout.items.len == 0) return;
if (std.time.milliTimestamp() >= end) return error.SendStalled;
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();
}
}
/// The next whole frame, if there is one. (Moved from
/// client.Transport.readFrame; see Incoming for why a missing frame is
/// not automatically a dead transport.)
pub fn readFrame(self: *Link, alloc: std.mem.Allocator) !Incoming {
switch (self.*) {
.quic => |*q| {
// Death is checked after the pump, so bytes that arrived in
// the same pass as the close are still delivered before the
// tear.
const got = proto.takeFrame(alloc, &q.cl.in) catch |err| switch (err) {
error.OutOfMemory => return err, // not a transport event
else => return .closed,
};
if (got) |frame| return .{ .frame = frame };
return if (q.cl.dead) .closed else .incomplete;
},
.fd => |fd| {
const frame = (proto.readFrame(alloc, fd) catch |err| switch (err) {
error.OutOfMemory => return err,
else => return .closed,
}) orelse return .closed;
return .{ .frame = frame };
},
.pipe => |p| {
const frame = (proto.readFrame(alloc, p.r) catch |err| switch (err) {
error.OutOfMemory => return err,
else => return .closed,
}) orelse return .closed;
return .{ .frame = frame };
},
}
}
/// Wait for one frame of type `want`. THE primitive this row exists
/// for; every hand-rolled poll+readFrame loop in the tree is a copy of
/// this. Returns null when the deadline runs out (a null deadline waits
/// forever), error.Closed when the peer is gone — callers own the
/// wording for both (dial.ask maps Closed to its "no answer" null; muxa
/// maps it to DaemonGone/ConnectionLost). Non-matching frames go to the
/// sink (see Sink for ownership).
pub fn awaitFrame(
self: *Link,
alloc: std.mem.Allocator,
want: proto.MsgType,
deadline_ms: ?u32,
sink: Sink,
) !?proto.Frame {
const end: ?i64 = if (deadline_ms) |ms| std.time.milliTimestamp() + ms else null;
while (true) {
// Deliver what is already in hand before waiting: a QUIC frame
// may be whole in cl.in from an earlier pump, and poll would
// never fire for bytes already in userspace.
switch (try self.readFrame(alloc)) {
.frame => |frame| {
if (frame.type == want) return frame;
if (sink.on) |on| {
defer frame.deinit(alloc);
try on(sink.ctx, frame);
} else frame.deinit(alloc);
continue;
},
.closed => return error.Closed,
.incomplete => {},
}
// Nothing whole in hand: wait. The wait is capped at 250ms even
// with no caller deadline, because a QUIC link's timers (loss
// detection, keepalive) need servicing on schedule rather than
// whenever the daemon happens to say something; a plain fd with
// no deadline may block indefinitely, which is what a caller
// wants when a stopped daemon should be a visible hang.
var cap: i32 = undefined;
if (end) |e| {
const left = e - std.time.milliTimestamp();
if (left <= 0) return null;
cap = @intCast(@min(left, 250));
} else cap = switch (self.*) {
.quic => 250,
.fd, .pipe => -1,
};
var fds = [_]std.posix.pollfd{
.{ .fd = self.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
};
_ = std.posix.poll(&fds, self.timeoutMs(cap)) catch return error.Closed;
self.service();
// The fd arms block in readFrame on a partial frame even past
// the deadline — the price of frames over a stream, same as
// dial.askOn always paid; a malformed frame is an error rather
// than an absent reply.
}
}
/// Idempotent, and it has to be: a re-dial releases the dead link on
/// entry, and an abort then closes the same value again through the
/// pump's defer. (The pipe-kill ordering is client.Transport.close's,
/// moved: stdin first so the command sees EOF and can wind down its
/// remote end, then TERM. kill() waitpid()s internally — no zombie.)
pub fn close(self: *Link) void {
switch (self.*) {
.fd => |fd| {
if (fd == -1) return; // already released
std.posix.close(fd);
},
.pipe => |*p| {
if (p.child.stdin) |*in| {
in.close();
p.child.stdin = null;
}
_ = p.child.kill() catch {};
},
.quic => |*q| {
q.qout.deinit(q.alloc);
// Closes the UDP socket with it, so pollFd's value must not
// be closed again.
q.cl.deinit();
},
}
self.* = .{ .fd = -1 };
}
};
```
Tests, in the same file (the socketpair shapes are dial.zig's; a fake peer
thread plays the daemon). Note the sink test drives a PLURAL stream — two
noise frames before the answer — per the plural-fixture rule:
```zig
const testing = std.testing;
fn mkPair() ![2]std.posix.fd_t {
var pair: [2]i32 = undefined;
try testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
return .{ pair[0], pair[1] };
}
test "fd link: sendFrame puts one frame on the wire, readFrame takes it back" {
const alloc = testing.allocator;
const pair = try mkPair();
var a: Link = .{ .fd = pair[0] };
var b: Link = .{ .fd = pair[1] };
defer a.close();
defer b.close();
try a.sendFrame(.input, "hi");
const inc = try b.readFrame(alloc);
const frame = inc.frame;
defer frame.deinit(alloc);
try testing.expectEqual(proto.MsgType.input, frame.type);
try testing.expectEqualStrings("hi", frame.payload);
}
test "close is idempotent on every arm that can be closed twice" {
const pair = try mkPair();
std.posix.close(pair[1]);
var l: Link = .{ .fd = pair[0] };
l.close();
l.close(); // second close must be a no-op, not EBADF-unreachable
}
test "awaitFrame: null sink drops noise frames and returns the match" {
const alloc = testing.allocator;
const pair = try mkPair();
var l: Link = .{ .fd = pair[0] };
defer l.close();
// Two noise frames BEFORE the answer: the wait must survive a stream,
// not a single-frame fixture.
try proto.writeFrame(pair[1], .pty_mode, &.{0});
try proto.writeFrame(pair[1], .delta, "x");
try proto.writeFrame(pair[1], .stats_reply, "ok");
std.posix.close(pair[1]);
const got = (try l.awaitFrame(alloc, .stats_reply, 1000, .{})).?;
defer got.deinit(alloc);
try testing.expectEqualStrings("ok", got.payload);
}
test "awaitFrame: the sink sees every non-match, in order, and may end the wait" {
const alloc = testing.allocator;
const Seen = struct {
types: [8]proto.MsgType = undefined,
n: usize = 0,
fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
const self: *@This() = @ptrCast(@alignCast(ctx.?));
self.types[self.n] = frame.type;
self.n += 1;
if (frame.type == .exit_status) return error.SessionExited;
}
};
// Leg one: the sink observes and the wait completes.
{
const pair = try mkPair();
var l: Link = .{ .fd = pair[0] };
defer l.close();
try proto.writeFrame(pair[1], .snapshot, "");
try proto.writeFrame(pair[1], .stats_reply, "ok");
std.posix.close(pair[1]);
var seen: Seen = .{};
const got = (try l.awaitFrame(alloc, .stats_reply, 1000, .{ .ctx = &seen, .on = Seen.on })).?;
defer got.deinit(alloc);
try testing.expectEqual(@as(usize, 1), seen.n);
try testing.expectEqual(proto.MsgType.snapshot, seen.types[0]);
}
// Leg two: the sink's error IS the outcome — an "other" frame that is
// an answer, muxa's whole reason for a second copy.
{
const pair = try mkPair();
var l: Link = .{ .fd = pair[0] };
defer l.close();
try proto.writeFrame(pair[1], .exit_status, &.{7});
std.posix.close(pair[1]);
var seen: Seen = .{};
try testing.expectError(
error.SessionExited,
l.awaitFrame(alloc, .stats_reply, 1000, .{ .ctx = &seen, .on = Seen.on }),
);
}
}
test "awaitFrame: silence consumes the deadline and returns null; a closed peer is error.Closed" {
const alloc = testing.allocator;
{
const pair = try mkPair();
defer std.posix.close(pair[1]);
var l: Link = .{ .fd = pair[0] };
defer l.close();
const t0 = std.time.milliTimestamp();
try testing.expect((try l.awaitFrame(alloc, .stats_reply, 100, .{})) == null);
try testing.expect(std.time.milliTimestamp() - t0 >= 100);
}
{
const pair = try mkPair();
var l: Link = .{ .fd = pair[0] };
defer l.close();
std.posix.close(pair[1]);
try testing.expectError(error.Closed, l.awaitFrame(alloc, .stats_reply, 1000, .{}));
}
}
test "awaitFrame: no deadline waits out a late reply" {
const alloc = testing.allocator;
const pair = try mkPair();
var l: Link = .{ .fd = pair[0] };
defer l.close();
const Late = struct {
fn run(fd: std.posix.fd_t) void {
std.Thread.sleep(150 * std.time.ns_per_ms);
proto.writeFrame(fd, .stats_reply, "late") catch {};
std.posix.close(fd);
}
};
const t = try std.Thread.spawn(.{}, Late.run, .{pair[1]});
defer t.join();
const got = (try l.awaitFrame(alloc, .stats_reply, null, .{})).?;
defer got.deinit(alloc);
try testing.expectEqualStrings("late", got.payload);
}
```
- [ ] **Step 2: Wire the row in `build.zig` and verify the tests fail-then-pass**
In the module table (after the `dial` row at build.zig:169), mirroring dial's
comment style:
```zig
// The live connection itself — fd, pipe or QUIC — and the one wait-for-a-
// frame loop. `term` for frames, `quic` for the third arm; policy stays
// with the rows that import this one.
.{ .name = "link", .path = "src/link.zig", .link_libc = true, .imports = &.{ "term", "quic" }, .quic_tests = true },
```
Add `"link"` to the test-rows list near build.zig:720 (the block naming
`script, cliflags, testtmp, spawn, dial, quic, ...`), in whichever sublist
`quic` sits in (link links libc through quic). If the build refuses a flag
combination, copy the `client` row's flags — it has the same dependency shape.
Run: `make check`
Expected: PASS, with the new link tests listed as run. If `zig build test`
prints nothing for minutes, a test wrote to fd 1 — fix that first.
- [ ] **Step 3: Commit**
```bash
git add src/link.zig build.zig
git commit -m "link: the fd|pipe|quic union and the one await loop, as a row"
```
---
### Task 2: `dial.ask` rides the Link; `askOn` dies
**Files:**
- Modify: `src/dial.zig` (delete lines 98–139: `askOn` and its doc comment; rewrite `ask`'s body; the `askOn` socketpair tests at the bottom become `awaitFrame`-through-`ask`-shaped or move their pin to link.zig where Task 1 already covers it)
- Modify: `build.zig` (`dial` row gains `"link"` in imports)
**Interfaces:**
- Consumes: `Link.awaitFrame`, `Link.sendFrame`, `Link.close` from Task 1.
- Produces: `dial.ask` with its EXACT current signature and contract —
`ask(alloc, sock_path, req: proto.MsgType, payload: []const u8, want: proto.MsgType, deadline_ms: ?u32) !?proto.Frame`,
still minting `error.NoDaemon` and `error.RequestNotSent`, still returning
null for both "deadline ran out" and "peer closed without answering".
- [ ] **Step 1: Rewrite `ask` over the Link**
```zig
const link_mod = @import("link");
pub fn ask(
alloc: std.mem.Allocator,
sock_path: []const u8,
req: proto.MsgType,
payload: []const u8,
want: proto.MsgType,
deadline_ms: ?u32,
) !?proto.Frame {
const s = dial(sock_path) catch return error.NoDaemon;
var l: link_mod.Link = .{ .fd = s.handle };
defer l.close();
// A request that could not be delivered is its own answer: the caller
// reports a daemon that never heard the question differently from one
// that heard it and said nothing.
l.sendFrame(req, payload) catch return error.RequestNotSent;
// A peer that closed without answering is the same "no answer" as a
// deadline that ran out — dial.ask's contract predates the Link and
// keeps it; callers that need the distinction hold a Link themselves.
return l.awaitFrame(alloc, want, deadline_ms, .{}) catch |e| switch (e) {
error.Closed => null,
else => e,
};
}
```
Delete `askOn` (dial.zig:98–139). Keep `dial`, `dialAttach`,
`dialAttachNamed`, `detach` untouched.
- [ ] **Step 2: Re-point dial's askOn-specific tests**
The tests "ask: a path nothing is bound at is error.NoDaemon" and the
attach-helper tests stay verbatim. The two socketpair tests that reached the
private `askOn` ("a deadline gives up on silence…") pin behavior Task 1's
link tests now pin (the deadline-consumed case and the
late-reply-without-deadline case are both in link.zig); delete them here
after confirming both link tests exist by symbol.
- [ ] **Step 3: Wire and gate**
`build.zig` dial row becomes:
```zig
.{ .name = "dial", .path = "src/dial.zig", .link_libc = true, .imports = &.{ "term", "link", "quic" } },
```
(`link` pulls `quic`; if the build graph complains about transitive libc,
mirror what the `client` row does.)
Run: `make check` — expected PASS.
Then the behavior gate against a REAL daemon, because dial.ask serves the
observer verbs: `E2E_ONLY=03 make e2e` (the side-connection group), capture
`$?` before any pipe. Expected PASS.
- [ ] **Step 4: Commit**
```bash
git add src/dial.zig build.zig
git commit -m "dial.ask asks through the Link; askOn's loop retires into awaitFrame"
```
---
### Task 3: `client.Transport` wraps a Link
**Files:**
- Modify: `src/client/client.zig` — fields at 342–380; `pipeTransport`/`quicTransport` constructors; delete the bodies of `writeFrame`/`service`/`timeoutMs`/`flushQuic`/`readFrame` (694–766) and the per-arm half of `close` (767–808); `awaitFrames` (1174–1231); `pollFd`.
- Modify: `build.zig` (`client` row gains `"link"`).
**Interfaces:**
- Consumes: everything Task 1 produces.
- Produces: `Transport`'s public surface UNCHANGED — `open`, `openHandoff`, `openQuicEndpoint`, `pollFd`, `errFd`, `drainErr`, `adopt`, `writeFrame`, `service`, `timeoutMs`, `flushQuic`, `readFrame`, `close` keep their signatures, so wallview/wall_pump/interact/webhub compile untouched.
- [ ] **Step 1: Swap the fields**
`Transport.link` becomes `link_mod.Link` (import as `link_mod`, the same
shadow-avoidance dance the file already does for `dial`/`dialer`). Delete
`qout` (it lives in `Link.Quic` now) and delete `Conn`/`conn` — `pollFd()`
returns `self.link.pollFd()`; grep for every `conn.r`/`conn.w` use in
client.zig, wallview.zig, wall_pump.zig first (`grep -rn "\.conn\." src/client src/tui`)
and re-point each through the Link. If a use needs the write fd specifically,
add nothing to Link — `sendFrame` is the only writer by design; a use that
wants raw fds is a policy leak to flag in the PR, not to enable.
`Transport.close` keeps ONLY its own policy — the `err_fd` close and the
release check — then delegates:
```zig
pub fn close(self: *Transport) void {
if (self.link == .fd and self.link.fd == -1) return; // already released
if (self.err_fd >= 0) {
// Ours, not the child's: openHandoff took stderr off the Child so
// kill would leave it alone, so nothing else will close it.
std.posix.close(self.err_fd);
self.err_fd = -1;
}
self.link.close();
}
```
`writeFrame`/`service`/`timeoutMs`/`flushQuic`/`readFrame` become one-line
delegations (keep the public names; wall code calls them). The constructors
build Links: `pipeTransport` returns `.{ .link = .{ .pipe = .{ .child = child, .r = child.stdout.?.handle, .w = child.stdin.?.handle } } }`;
`quicTransport` returns `.{ .link = .{ .quic = .{ .cl = cl, .alloc = alloc } } }`.
`Incoming` becomes a re-export: `pub const Incoming = link_mod.Incoming;`
(callers spell `client.Incoming` today and must keep compiling).
- [ ] **Step 2: `awaitFrames` (client.zig:1174) rides `awaitFrame`**
Read its body first; it waits during birth/attach. Reshape it as a call to
`self.link.awaitFrame` with a sink that captures what the current loop's
non-match arms do (the BirthFake test at client.zig:2850 pins the order —
pty_mode before snapshot — and must still pass unchanged).
- [ ] **Step 3: Gate**
Run: `make check` — PASS. Then `make e2e` (the wall, handoff, QUIC groups
all cross this seam) — capture `$?`, expected PASS. The named risks: the
close-idempotence dance and the pipe-kill ordering moved in Task 1; if e2e
group 04 (handoff) fails, diff `Link.close`'s pipe arm against the old
`Transport.close` before touching anything else.
- [ ] **Step 4: Commit**
```bash
git add src/client/client.zig build.zig
git commit -m "Transport wraps the Link: policy stays, mechanics move"
```
---
### Task 4: muxa's `AgentConnection` wraps a Link
**Files:**
- Modify: `src/cli/muxa.zig` — the link union inside `AgentConnection` (~286), `open`/`openQuic`/`close`, `sendFrame` (deadline plumbing), delete `sendFrameQuic` (386–417), `awaitFrameFd` (425–457), `awaitFrameQuic` (458–507), `waitReady` (549–566, if the client's copy is reachable; otherwise keep — check callers), keep `reconnect` but re-point it at the Link's quic arm.
- Modify: `build.zig` (`agent` row gains `"link"`). muxa still does NOT import `client`.
**Interfaces:**
- Consumes: `Link.sendFrame` + `flushWithin`, `Link.awaitFrame` with a Sink, `Link.close`.
- Produces: `mux a`'s verbs behave identically — every JSON field, every exit code, every `mechanism` string.
- [ ] **Step 1: The union becomes a Link plus muxa's policy**
```zig
const AgentConnection = struct {
link: link_mod.Link,
/// 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,
connect_ms: i64,
reconnected: bool = false,
} = null,
alloc: std.mem.Allocator,
saw_snapshot: bool = false,
session_exit: ?u8 = null,
reconnect_failure: ?[]const u8 = null,
...
```
`sendFrame` becomes: `self.link.sendFrame(t, payload)` then, on the quic arm,
`self.link.flushWithin(bounded)` where `bounded` is today's
`@min(deadline_ms - now, send_flush_ms)` math converted to relative u32 at
this boundary; the BrokenPipe/ConnectionReset/ConnectionLost →
`refusalPending` classification stays wrapped around the call exactly as at
muxa.zig:358–370.
- [ ] **Step 2: The three await fns become one sink**
```zig
fn onOther(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
const self: *AgentConnection = @ptrCast(@alignCast(ctx.?));
// Skip unrelated snapshots and deltas while waiting. 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.
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; 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;
const got = self.link.awaitFrame(self.alloc, want, @intCast(left), .{
.ctx = self,
.on = onOther,
}) catch |e| switch (e) {
// Wording is per-arm on purpose: a closed unix socket is a dead
// daemon; 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,
},
else => return e,
};
return got orelse error.Timeout;
}
```
Delete `awaitFrameFd`, `awaitFrameQuic`, `sendFrameQuic`. `reconnect` builds a
fresh `quic.Client` from `self.redial.?` and swaps it into `self.link.quic.cl`
(deinit the old one first, exactly the current ordering at muxa.zig:504–510).
`graceMs` reads `self.redial.?.connect_ms`.
- [ ] **Step 3: Gate**
Run: `make check` then `make agent` — capture `$?`, both PASS. The agent
suite runs real verbs against a real daemon over both transports; it is the
pin for "every JSON field identical". Then
`grep -c "awaitFrameFd\|awaitFrameQuic\|sendFrameQuic" src/cli/muxa.zig`
must print 0.
- [ ] **Step 4: Commit**
```bash
git add src/cli/muxa.zig build.zig
git commit -m "muxa rides the Link: the second transport retires"
```
---
### Task 5: `src/serve.zig` — bind and teardown get one owner
**Files:**
- Create: `src/serve.zig`
- Modify: `build.zig` (new row; `daemon` and `client` rows gain `"serve"`)
- Modify: `src/server/server.zig` (init at 505–540: claim+listen+PathId → `serve.bind`; deinit at 815–817: guarded unlink → `Bound.close`; `initFromManifest` at ~628: `serve.adopt`)
- Modify: `src/server/server_agent.zig` (bind at 150–176, `AgentSock.release` at 22–34)
- Modify: `src/client/askpass.zig` (bind at 145–162, the unlink in `retire` at ~206)
- Test: additions in `src/serve.zig` itself, `src/server/server_test_agent.zig`, askpass's tests in `src/client/askpass.zig`
**Interfaces:**
- Consumes: `sockpath.claim`, `sockpath.PathId`.
- Produces:
- `pub const Policy = enum { refuse_live, clobber_own };`
- `pub fn bind(path: []const u8, opts: BindOpts) !Bound` where `BindOpts = struct { policy: Policy, backlog: u31 = 128, cloexec: bool = false }`
- `pub fn adopt(fd: std.posix.fd_t, path: []const u8) !Bound`
- `pub const Bound = struct { fd: std.posix.fd_t, path_id: sockpath.PathId, pub fn close(self: *Bound, path: []const u8) void; }`
- [ ] **Step 1: Write `src/serve.zig` with tests**
```zig
//! The server side of a unix socket path: the right to bind it and the duty
//! to unlink it, written once. Three binders used to answer this trio
//! independently — the daemon socket, the per-session agent sockets, the
//! askpass socket — and only the first carried the guarded unlink that
//! sockpath's incident record paid for. Accept loops are NOT here: the
//! daemon's slot-table accept, askpass's credential check and the hub's
//! thread-per-conn differ for reasons, and a shared loop would be a shape
//! they do not fit.
const std = @import("std");
const sockpath = @import("sockpath");
pub const Policy = enum {
/// Never steal a path that answers: sockpath.claim's refusal, the
/// daemon-socket rule.
refuse_live,
/// The name embeds our identity (a pid, a session name in our own
/// directory), so a leftover file there is ours — a previous us that
/// died without unlinking — and is deleted before the bind.
clobber_own,
};
pub const BindOpts = struct {
policy: Policy,
backlog: u31 = 128,
/// FALSE is load-bearing: `mux d upgrade` execs the candidate over the
/// running daemon, and the daemon listener and every agent socket must
/// survive that exec. askpass passes true — its process never execs
/// over itself and its children must not inherit the prompt socket.
cloexec: bool = false,
};
pub const Bound = struct {
fd: std.posix.fd_t,
path_id: sockpath.PathId,
/// Close, then unlink only if the path still names OUR socket: a newer
/// owner may have replaced the file, and deleting that one would steal
/// its clients. The stat comes after the close because a successor only
/// claims once nothing is listening — that narrows the race to the
/// stat→unlink gap, the floor Linux gives for deleting by name.
pub fn close(self: *Bound, path: []const u8) void {
if (self.fd == -1) return;
std.posix.close(self.fd);
self.fd = -1;
if (self.path_id.stillAt(path)) {
std.fs.cwd().deleteFile(path) catch {};
}
}
};
/// Bind, listen, and remember which inode is ours. Overlong paths are
/// initUnix's refusal (kernel truth, not a re-stated bound) — `mux d`'s
/// parse-time `sockpath.tooLong` with its own stderr wording remains the
/// one binder-side pre-check, and it lives with `mux d`.
pub fn bind(path: []const u8, opts: BindOpts) !Bound {
switch (opts.policy) {
.refuse_live => try sockpath.claim(path),
.clobber_own => std.fs.cwd().deleteFile(path) catch {},
}
const addr = try std.net.Address.initUnix(path);
const sock_flags: u32 = std.posix.SOCK.STREAM |
(if (opts.cloexec) std.posix.SOCK.CLOEXEC else 0);
const fd = try std.posix.socket(std.posix.AF.UNIX, sock_flags, 0);
errdefer std.posix.close(fd);
try std.posix.bind(fd, &addr.any, addr.getOsSockLen());
try std.posix.listen(fd, opts.backlog);
return .{ .fd = fd, .path_id = try sockpath.PathId.of(path) };
}
/// A listener that already exists — the upgrade manifest's adopted fd. The
/// PathId is re-stamped from the file as found, which is the manifest rule:
/// a watermark belongs to the space that minted it.
pub fn adopt(fd: std.posix.fd_t, path: []const u8) !Bound {
return .{ .fd = fd, .path_id = try sockpath.PathId.of(path) };
}
```
Tests in-file (`test_imports = &.{"testtmp"}` gives `TmpDir`):
```zig
const TmpDir = @import("testtmp").TmpDir;
const testing = std.testing;
test "refuse_live refuses a path a live listener owns; clobber_own takes its own leftover" {
var tmp = try TmpDir.make();
defer tmp.cleanup();
var buf: [280]u8 = undefined;
const path = try std.fmt.bufPrint(&buf, "{s}/serve.sock", .{tmp.path()});
var first = try bind(path, .{ .policy = .refuse_live });
try testing.expectError(error.AddressInUse, bind(path, .{ .policy = .refuse_live }));
// Kill the listener but leave the file: the dead-us case.
std.posix.close(first.fd);
first.fd = -1;
var second = try bind(path, .{ .policy = .clobber_own });
second.close(path);
}
test "close unlinks our socket but never a successor's" {
var tmp = try TmpDir.make();
defer tmp.cleanup();
var buf: [280]u8 = undefined;
const path = try std.fmt.bufPrint(&buf, "{s}/succ.sock", .{tmp.path()});
var old = try bind(path, .{ .policy = .clobber_own });
// A successor replaces the FILE while old's listener lives on — deleting
// a unix socket's path does not touch the listening fd, which is the
// incident shape sockpath.zig:213 records: two owners, one name, and
// the displaced one's teardown must not delete by that name.
var succ = try bind(path, .{ .policy = .clobber_own });
old.close(path); // guard fires: the inode at path is succ's — no unlink
_ = try std.fs.cwd().statFile(path); // successor's file survived
succ.close(path); // ours: unlinked
try testing.expectError(error.FileNotFound, std.fs.cwd().statFile(path));
}
```
(The second test's shape: prove `stillAt` is false for the displaced owner,
prove the successor's file survives the displaced owner's teardown, prove the
successor's own close does unlink. If `error.AddressInUse` is not the exact
error `claim` surfaces, read `sockpath.claim` and pin the error it actually
returns — the CONTRACT is "the second refuse_live bind fails while the first
listener lives".)
- [ ] **Step 2: Wire the row**
```zig
.{ .name = "serve", .path = "src/serve.zig", .imports = &.{"sockpath"}, .test_imports = &.{"testtmp"} },
```
Add `"serve"` to the daemon and client rows' imports and to the test-rows
list near build.zig:720. Run `make check` — the new tests pass.
- [ ] **Step 3: Convert the three binders, one commit each**
(a) **Daemon socket.** `Server.init` (server.zig:505–540): replace
`sockpath.claim` + `addr.listen` + `PathId.of` with one `serve.bind(opts.sock_path, .{ .policy = .refuse_live })`;
keep `self.listener: std.net.Server` by constructing it from the Bound's fd
(`.{ .listen_address = addr, .stream = .{ .handle = bound.fd } }`) or store
the Bound and re-point `acceptConn` — whichever touches fewer lines; the
deinit block at 815–817 becomes `self.bound.close(self.sock_path)` with the
moved comment. `initFromManifest` uses `serve.adopt`. Gate:
`E2E_ONLY=03 make e2e` plus the upgrade suite
(`server_test_upgrade` runs under `make test`) — the adopted-listener path
must still not re-claim.
(b) **Agent sockets.** `server_agent.zig:150–176` becomes
`serve.bind(path, .{ .policy = .clobber_own, .backlog = 8 })` (keep the
backlog-8 comment: the only dialler is one session's ssh clients).
`AgentSock` carries the `Bound`; `release` becomes `bound.close(path)` + free.
New test in `server_test_agent.zig`: two sessions' agent sockets (plural),
end one session, assert the OTHER session's socket file still exists and
still answers — then assert the ended one's file is gone (ask the fs, not
the daemon).
(c) **askpass.** `askpass.zig:153–162` becomes
`serve.bind(path, .{ .policy = .clobber_own, .cloexec = true })`; `retire`'s
unlink becomes the guarded close via the stored Bound — BUT retire
deliberately does not close the fd (detached pumps may be inside `declined`);
split: `retire` calls a new `Bound.unlinkIfOurs(path)` (add it: the guard
without the close, two lines, doc comment saying retire is why it exists) and
`stop` keeps closing the fd. New test beside the existing Listener tests: a
successor Listener binds the same pid-named path (simulate pid reuse by
spelling the same path), the old Listener's retire must not unlink the
successor's socket.
Run after each: `make check`. After all three: `make test | tail -30` with
`$?` captured, then `E2E_ONLY=07 make e2e` if a group covers askpass
(check `ls test/e2e_*` for the askpass/handoff group; run the one that names
it, else note "covered by unit suite only" in the commit body).
- [ ] **Step 4: Commits** (one per binder, plus the module)
```bash
git commit -m "serve: bind and the guarded unlink get one owner"
git commit -m "daemon socket binds through serve"
git commit -m "agent sockets bind through serve, and gain the successor guard"
git commit -m "askpass binds through serve; retire keeps its no-close shape"
```
---
### Task 6: harness `pumpUntil` and the await-copy sweep
**Files:**
- Modify: `src/server/server_test_harness.zig` (add `pumpUntil`; `awaitFrame`/`awaitFrameOn`/`firstStateFrame` become Link wrappers)
- Modify: `src/server/server_test_attach.zig`, `server_test_session.zig`, `server_test_modes.zig`, `server_test_clipboard.zig`, `server_test_deliver.zig`, `server_test_await.zig`, `server_test_agent.zig`, `server_test_upgrade.zig`, `server_test_quic.zig` (spin loops → `pumpUntil`; inline poll+readFrame loops → `awaitFrameOn`/sinks; private awaits deleted)
- Modify: `build.zig` (`daemon` row's `test_imports` gains `"link"`)
**Interfaces:**
- Consumes: `Link.awaitFrame` + `Sink`.
- Produces (test-only):
- `pub fn pumpUntil(srv: *Server, deadline_ms: u64, ctx: anytype, comptime pred: fn (@TypeOf(ctx)) bool) !bool`
- `awaitFrame`/`awaitFrameOn`/`firstStateFrame`/`awaitGridText` keep their exact current signatures (eight sibling files call them).
- [ ] **Step 1: `pumpUntil`**
```zig
/// Pump the daemon until pred says the world arrived, or the deadline says
/// it never will. Returns whether pred fired, so a test asserts the
/// CONDITION and its failure names what didn't happen — not a guessed
/// round count. A false return is an assertable value: the deadline turns
/// a wedge into a legible failure instead of a silent hang (a wedged zig
/// test prints nothing).
pub fn pumpUntil(
srv: *Server,
deadline_ms: u64,
ctx: anytype,
comptime pred: fn (@TypeOf(ctx)) bool,
) !bool {
var left = deadline_ms;
while (true) {
if (pred(ctx)) return true;
if (left == 0) return false;
try srv.pumpOnce(5);
left -|= 5;
}
}
```
With a self-test in the harness file: a `TestDaemon`, one dial, and
`pumpUntil(td.srv, 2000, td.srv, hasAnyClient)` flips true; a predicate that
can never fire returns false in ~deadline time.
- [ ] **Step 2: The harness awaits become wrappers**
`awaitFrameOn` keeps its signature and becomes
`var l: link_mod.Link = .{ .fd = fd }; defer l.* = .{ .fd = -1 };` — do NOT
`close` (the caller owns the fd) — then `l.awaitFrame(alloc, want, timeout, .{})`
mapping `error.Closed` to the function's current null/erroring behavior
(read its body first and preserve it exactly; the eight siblings' green run
is the proof). `firstStateFrame` uses a sink that decodes
`readSnapshotPrefix`/`readDeltaHeader` into its `StateFrame` — copying the
ints, per the Sink borrow rule. `awaitFrame` (the byte-scanning one at 264 —
read it; it may scan a buffer, not an fd) is converted only if it is
genuinely the same loop; if it scans captured bytes, it stays and the plan's
count table loses those lines — note it in the commit.
- [ ] **Step 3: The sweep, one test file per commit**
For each file in the Files list: replace `while (spun < N …) pumpOnce`
loops with `try expect(try h.pumpUntil(td.srv, <N*step rounded up to a wall
figure>, ctx, pred))`; replace inline poll+readFrame switch loops with
`h.awaitFrameOn` + a sink where the loop classified frames (the
history-counting loop at server_test_attach.zig:61–78 becomes a sink
capturing `history_rows`); delete `awaitSelectionReply` (clipboard:17),
`awaitSnapshotSize` (attach:400), `awaitMarkerWithoutSnapshot` (attach:478)
in favor of sinks at their call sites. EVERY replacement must watch its
assertion fire once: before committing each file, break the condition (flip
an expected value), run that file's test, see the new failure message name
the condition, restore. Run `make test | tail -30` (capture `$?`) after each
file.
- [ ] **Step 4: Commit per file, then the gate**
```bash
git commit -m "harness: pumpUntil and Link-backed awaits"
git commit -m "server_test_attach rides the harness awaits" # ...and so on per file
```
Final: `grep -rn "spun < \|rounds < " src/server/` prints nothing;
`grep -rln "posix.poll" src/server/server_test_*.zig` shrinks to files whose
polls are not frame-awaits (inspect each survivor and say why in the commit).
---
### Task 7: Delivery gate
**Files:**
- Modify: `docs/superpowers/specs/2026-08-31-connection-primitives-design.md` (replace the estimate table with the real diffstat)
- Modify: `CLAUDE.md` (the layout table: `link` and `serve` join the shared row's list)
**Interfaces:** none new.
- [ ] **Step 1: Success-criteria greps from the spec**
Run each; every miss is unfinished work, not a tolerable gap:
- `grep -rln "posix.poll" src/ --include="*.zig" | xargs grep -ln readFrame` → only `src/link.zig` (plus the wall's event loop files; name them in the delivery note).
- `grep -rn "union(enum)" src/cli/muxa.zig src/client/client.zig | grep -i "fd\|quic"` → no second link union.
- `grep -rn "addr.listen\|posix.listen" src/server/server.zig src/server/server_agent.zig src/client/askpass.zig` → nothing (all through serve).
- `grep -rn "while (spun" src/server/` → nothing.
- `grep -rn "@import(\"link\")\|@import(\"term\")" src/proxy.zig src/quic.zig` → nothing.
- [ ] **Step 2: The full gate**
```bash
make ci; echo "ci=$?"
make xversion-build && make xversion; echo "xver=$?"
```
Both 0. The xversion old side is `../mux-xver-old`; build its prefix first
if stale. No wire change means xversion is a formality — but run it, because
"should be a formality" is an assumption and the gate is the assertion.
- [ ] **Step 3: Re-measure and record**
`git diff --stat <base>..HEAD` and `wc -l src/*.zig src/*/*.zig | tail -1`;
replace the spec's estimate table with the real numbers. Update CLAUDE.md's
module table row for `src/`: add `link` and `serve` to the shared list with
one clause each.
- [ ] **Step 4: Autosquash and finish**
```bash
git rebase -i --autosquash <base> # NOTE: -i is unsupported in this harness —
# use: GIT_SEQUENCE_EDITOR=true git rebase --autosquash <base>
```
Only if no sibling agent holds the old tip (the no-history-rewrite rule).
Then a demo: `try.sh`-style script that starts an isolated daemon
(`XDG_STATE_HOME`/`XDG_RUNTIME_DIR` exported to a scratch dir), runs
`mux d endpoint`, one `mux a status`, and a `mux d stop` — the primitives
exercised end to end by the real binary.
```bash
git commit -m "docs: link and serve join the table; spec table becomes the diffstat"
```