diff --git a/build.zig b/build.zig
index 241e316fbe..8882e95dd4 100644
--- a/build.zig
+++ b/build.zig
@@ -96,6 +96,7 @@ pub fn build(b: *std.Build) void {
     @import("zig/build/kernel_symbols.zig").tests(b, test_step);
     @import("zig/build/checkpoint_security.zig").tests(b, test_step);
     @import("zig/build/quant_geometry_host.zig").steps(b, test_step);
+    @import("zig/build/stage.zig").steps(b, test_step);
     if (b.graph.host.result.os.tag == .macos) {
         const step5_mods = metalEngineModules(b, b.graph.host, .Debug, draft_ids);
         const empty = b.addWriteFiles().add("step5-compile-only.metallib", "");
diff --git a/zig/build/stage.zig b/zig/build/stage.zig
new file mode 100644
index 0000000000..f3b31592f2
--- /dev/null
+++ b/zig/build/stage.zig
@@ -0,0 +1,22 @@
+//! The pipeline stage link (zig/src/fabric/stage): its module, host tests, and compile checks for Linux and macOS.
+const std = @import("std");
+
+pub fn module(b: *std.Build, target: std.Build.ResolvedTarget, optimize: std.builtin.OptimizeMode) *std.Build.Module {
+    return b.createModule(.{ .root_source_file = b.path("zig/src/fabric/stage/stage.zig"), .target = target, .optimize = optimize, .link_libc = true });
+}
+
+pub fn steps(b: *std.Build, all: *std.Build.Step) void {
+    const host = b.graph.host;
+    if (host.result.os.tag != .linux and host.result.os.tag != .macos) return;
+    const run = b.addRunArtifact(b.addTest(.{ .root_module = module(b, host, .Debug) }));
+    run.has_side_effects = true; // loopback sockets
+    const step = b.step("test-stage", "Pipeline stage link: rings of two and three ranks on loopback, refusals and failures");
+    step.dependOn(&run.step);
+    all.dependOn(step);
+    const compile = b.step("test-stage-compile", "Compile the stage link's tests for Linux and macOS on aarch64");
+    for ([_]std.Target.Os.Tag{ .linux, .macos }) |os| {
+        const target = b.resolveTargetQuery(.{ .cpu_arch = .aarch64, .os_tag = os });
+        compile.dependOn(&b.addTest(.{ .root_module = module(b, target, .Debug) }).step);
+    }
+    all.dependOn(compile);
+}
diff --git a/zig/src/fabric/stage/config.zig b/zig/src/fabric/stage/config.zig
new file mode 100644
index 0000000000..a538d4ac84
--- /dev/null
+++ b/zig/src/fabric/stage/config.zig
@@ -0,0 +1,128 @@
+//! A stage rank's ring settings from TF_STAGE_RANK, _WORLD, _PEERS, _SECRET_FILE and _TIMEOUT_S.
+const std = @import("std");
+const ring = @import("ring.zig");
+const secret = @import("secret.zig");
+
+/// The most ranks a ring takes.
+pub const most_ranks = 64;
+
+pub const Error = secret.Error || error{ MissingStageSetting, BadStageRank, BadStageWorld, BadStagePeers, BadStageTimeout };
+
+/// Parsed settings; `peers` points into `slots` and into the TF_STAGE_PEERS text.
+pub const Settings = struct {
+    rank: u32,
+    world: u32,
+    timeout_ns: u64,
+    key: secret.Key = undefined,
+    slots: [most_ranks]ring.Peer = undefined,
+    n: usize = 0,
+
+    pub fn peers(s: *const Settings) []const ring.Peer {
+        return s.slots[0..s.n];
+    }
+
+    /// The ring's options for these settings and the checkpoint's and layout's `fingerprint`.
+    pub fn options(s: *const Settings, fingerprint: [32]u8) ring.Options {
+        return .{ .rank = s.rank, .world = s.world, .peers = s.peers(), .fingerprint = fingerprint, .key = s.key, .timeout_ns = s.timeout_ns };
+    }
+};
+
+/// The settings named by `get` (an environment lookup), refused by name when one is missing or malformed.
+pub fn parse(get: anytype, s: *Settings) Error!void {
+    const rank = get.lookup("TF_STAGE_RANK") orelse return error.MissingStageSetting;
+    const world = get.lookup("TF_STAGE_WORLD") orelse return error.MissingStageSetting;
+    const list = get.lookup("TF_STAGE_PEERS") orelse return error.MissingStageSetting;
+    const path = get.lookup("TF_STAGE_SECRET_FILE") orelse return error.MissingStageSetting;
+    s.* = .{ .rank = 0, .world = 0, .timeout_ns = 60 * std.time.ns_per_s };
+    s.world = std.fmt.parseInt(u32, world, 10) catch return error.BadStageWorld;
+    if (s.world < 2 or s.world > most_ranks) return error.BadStageWorld;
+    s.rank = std.fmt.parseInt(u32, rank, 10) catch return error.BadStageRank;
+    if (s.rank >= s.world) return error.BadStageRank;
+    if (get.lookup("TF_STAGE_TIMEOUT_S")) |v| {
+        const secs = std.fmt.parseInt(u32, v, 10) catch return error.BadStageTimeout;
+        if (secs == 0) return error.BadStageTimeout;
+        s.timeout_ns = @as(u64, secs) * std.time.ns_per_s;
+    }
+    var it = std.mem.splitScalar(u8, list, ',');
+    while (it.next()) |entry| {
+        if (s.n == most_ranks) return error.BadStagePeers;
+        const colon = std.mem.lastIndexOfScalar(u8, entry, ':') orelse return error.BadStagePeers;
+        const host = entry[0..colon];
+        const port = std.fmt.parseInt(u16, entry[colon + 1 ..], 10) catch return error.BadStagePeers;
+        if (host.len == 0 or port == 0) return error.BadStagePeers;
+        s.slots[s.n] = .{ .host = host, .port = port };
+        s.n += 1;
+    }
+    if (s.n != s.world) return error.BadStagePeers;
+    var name: [4096]u8 = undefined;
+    if (path.len >= name.len) return error.SecretUnreadable;
+    @memcpy(name[0..path.len], path);
+    name[path.len] = 0;
+    s.key = try secret.fromFile(name[0..path.len :0]);
+}
+
+/// The process environment, for `parse`.
+pub const Env = struct {
+    pub fn lookup(_: Env, name: [:0]const u8) ?[]const u8 {
+        const v = std.c.getenv(name) orelse return null;
+        return std.mem.span(v);
+    }
+};
+
+const Fake = struct {
+    rank: ?[]const u8 = "1",
+    world: ?[]const u8 = "3",
+    peers: ?[]const u8 = "192.0.2.1:7100,stage-b.example:7101,127.0.0.1:7102",
+    timeout: ?[]const u8 = null,
+    secret_file: ?[]const u8 = "/nonexistent/stage-secret",
+    pub fn lookup(f: Fake, name: [:0]const u8) ?[]const u8 {
+        if (std.mem.eql(u8, name, "TF_STAGE_RANK")) return f.rank;
+        if (std.mem.eql(u8, name, "TF_STAGE_WORLD")) return f.world;
+        if (std.mem.eql(u8, name, "TF_STAGE_PEERS")) return f.peers;
+        if (std.mem.eql(u8, name, "TF_STAGE_TIMEOUT_S")) return f.timeout;
+        if (std.mem.eql(u8, name, "TF_STAGE_SECRET_FILE")) return f.secret_file;
+        return null;
+    }
+};
+
+/// An owner-only secret file for one test, at `path`.
+fn writeSecret(path: [:0]const u8) !void {
+    _ = std.c.unlink(path);
+    const fd = std.c.open(path, .{ .ACCMODE = .WRONLY, .CREAT = true, .EXCL = true }, @as(std.c.mode_t, 0o600));
+    if (fd < 0) return error.SecretUnreadable;
+    defer _ = std.c.close(fd);
+    _ = std.c.write(fd, "a stage ring's test secret\n", 27);
+}
+
+test "stage config: a three-rank ring's settings parse, with its timeout and key" {
+    var buf: [64]u8 = undefined;
+    const path = try std.fmt.bufPrintSentinel(&buf, "/tmp/tf-stage-config-{d}", .{std.c.getpid()}, 0);
+    try writeSecret(path);
+    defer _ = std.c.unlink(path);
+    var s: Settings = undefined;
+    try parse(Fake{ .timeout = "5", .secret_file = path }, &s);
+    try std.testing.expectEqual(try secret.fromBytes("a stage ring's test secret"), s.key);
+    try std.testing.expectEqual(@as(u32, 1), s.rank);
+    try std.testing.expectEqual(@as(u32, 3), s.world);
+    try std.testing.expectEqual(5 * std.time.ns_per_s, s.timeout_ns);
+    try std.testing.expectEqualStrings("stage-b.example", s.peers()[1].host);
+    try std.testing.expectEqual(@as(u16, 7102), s.peers()[2].port);
+    try parse(Fake{ .secret_file = path }, &s);
+    try std.testing.expectEqual(60 * std.time.ns_per_s, s.timeout_ns);
+}
+
+test "stage config: a missing or malformed setting is refused by name" {
+    var s: Settings = undefined;
+    try std.testing.expectError(error.MissingStageSetting, parse(Fake{ .peers = null }, &s));
+    try std.testing.expectError(error.MissingStageSetting, parse(Fake{ .secret_file = null }, &s));
+    try std.testing.expectError(error.SecretUnreadable, parse(Fake{}, &s));
+    try std.testing.expectError(error.BadStageWorld, parse(Fake{ .world = "1" }, &s));
+    try std.testing.expectError(error.BadStageWorld, parse(Fake{ .world = "65" }, &s));
+    try std.testing.expectError(error.BadStageRank, parse(Fake{ .rank = "3" }, &s));
+    try std.testing.expectError(error.BadStageRank, parse(Fake{ .rank = "-1" }, &s));
+    try std.testing.expectError(error.BadStageTimeout, parse(Fake{ .timeout = "0" }, &s));
+    try std.testing.expectError(error.BadStagePeers, parse(Fake{ .peers = "a:1,b:2" }, &s));
+    try std.testing.expectError(error.BadStagePeers, parse(Fake{ .peers = "a:1,b:0,c:3" }, &s));
+    try std.testing.expectError(error.BadStagePeers, parse(Fake{ .peers = "a:1,:2,c:3" }, &s));
+    try std.testing.expectError(error.BadStagePeers, parse(Fake{ .peers = "a:1,b,c:3" }, &s));
+}
diff --git a/zig/src/fabric/stage/frame.zig b/zig/src/fabric/stage/frame.zig
new file mode 100644
index 0000000000..06b2ae0e6e
--- /dev/null
+++ b/zig/src/fabric/stage/frame.zig
@@ -0,0 +1,59 @@
+//! A stage link's frame header: 32 little-endian bytes (magic, version, kind, round, rows, payload bytes).
+const std = @import("std");
+
+pub const magic: u32 = 0x5453_4654; // "TFST" in its little-endian bytes (54 46 53 54)
+pub const version: u16 = 1;
+pub const size = 32;
+
+/// What a frame carries: the link's hello, a round's plan, its hidden rows, or a clean stop.
+pub const Kind = enum(u8) { hello = 1, plan = 2, rows = 3, stop = 4 };
+
+pub const Header = struct {
+    kind: Kind,
+    /// The round this frame belongs to; each link's rounds run 1, 2, 3, ... in order.
+    round: u64,
+    /// The rows a `rows` frame carries (0 for the others).
+    rows: u32,
+    /// Payload bytes after the header.
+    bytes: u64,
+};
+
+pub const Error = error{ BadMagic, BadVersion, BadKind, TooLarge };
+
+pub fn encode(h: Header) [size]u8 {
+    var out: [size]u8 = @splat(0);
+    std.mem.writeInt(u32, out[0..4], magic, .little);
+    std.mem.writeInt(u16, out[4..6], version, .little);
+    out[6] = @backingInt(h.kind);
+    std.mem.writeInt(u64, out[8..16], h.round, .little);
+    std.mem.writeInt(u32, out[16..20], h.rows, .little);
+    std.mem.writeInt(u64, out[24..32], h.bytes, .little);
+    return out;
+}
+
+/// A header from `raw`, refused unless it is this link's magic and version, a known kind, and at most `cap` bytes.
+pub fn decode(raw: *const [size]u8, cap: u64) Error!Header {
+    if (std.mem.readInt(u32, raw[0..4], .little) != magic) return error.BadMagic;
+    if (std.mem.readInt(u16, raw[4..6], .little) != version) return error.BadVersion;
+    const kind = std.enums.fromInt(Kind, raw[6]) orelse return error.BadKind;
+    const bytes = std.mem.readInt(u64, raw[24..32], .little);
+    if (bytes > cap) return error.TooLarge;
+    return .{ .kind = kind, .round = std.mem.readInt(u64, raw[8..16], .little), .rows = std.mem.readInt(u32, raw[16..20], .little), .bytes = bytes };
+}
+
+test "stage frame: a header round-trips with TFST on the wire, and a wrong magic, version, kind or size is refused" {
+    const h: Header = .{ .kind = .rows, .round = 0x0102_0304_0506_0708, .rows = 4096, .bytes = 80 << 20 };
+    const raw = encode(h);
+    try std.testing.expectEqualStrings("TFST", raw[0..4]);
+    try std.testing.expectEqual(h, try decode(&raw, 80 << 20));
+    try std.testing.expectError(error.TooLarge, decode(&raw, (80 << 20) - 1));
+    var bad = raw;
+    bad[0] ^= 1;
+    try std.testing.expectError(error.BadMagic, decode(&bad, 1 << 30));
+    bad = raw;
+    bad[4] = 2;
+    try std.testing.expectError(error.BadVersion, decode(&bad, 1 << 30));
+    bad = raw;
+    bad[6] = 9;
+    try std.testing.expectError(error.BadKind, decode(&bad, 1 << 30));
+}
diff --git a/zig/src/fabric/stage/net.zig b/zig/src/fabric/stage/net.zig
new file mode 100644
index 0000000000..9ce2c9b8d5
--- /dev/null
+++ b/zig/src/fabric/stage/net.zig
@@ -0,0 +1,201 @@
+//! TCP for the stage link on Linux and macOS: listen on one address, connect and accept with deadlines, whole sends.
+const std = @import("std");
+const c = std.c;
+const posix = std.posix;
+
+pub const Fd = c.fd_t;
+
+pub const Error = error{ SocketFailed, AddressUnresolved, AddressInUse, WildcardAddress, Timeout, Closed };
+
+/// MSG_NOSIGNAL where the platform has it; macOS sets SO_NOSIGPIPE on the socket instead.
+const nosignal: u32 = if (@hasDecl(posix.MSG, "NOSIGNAL")) posix.MSG.NOSIGNAL else 0;
+
+pub fn nowNs() u64 {
+    var ts: c.timespec = undefined;
+    _ = c.clock_gettime(.MONOTONIC, &ts);
+    return @as(u64, @intCast(ts.sec)) * std.time.ns_per_s + @as(u64, @intCast(ts.nsec));
+}
+
+pub fn sleepNs(ns: u64) void {
+    var ts: c.timespec = .{ .sec = @intCast(ns / std.time.ns_per_s), .nsec = @intCast(ns % std.time.ns_per_s) };
+    while (c.nanosleep(&ts, &ts) != 0) {}
+}
+
+/// No Nagle (a decode row is one small frame), keepalive (a vanished host fails a read), no SIGPIPE on a dead peer.
+fn tune(fd: Fd) void {
+    const one: c_int = 1;
+    _ = c.setsockopt(fd, posix.IPPROTO.TCP, posix.TCP.NODELAY, &one, @sizeOf(c_int));
+    _ = c.setsockopt(fd, posix.SOL.SOCKET, posix.SO.KEEPALIVE, &one, @sizeOf(c_int));
+    if (@hasDecl(posix.SO, "NOSIGPIPE")) _ = c.setsockopt(fd, posix.SOL.SOCKET, posix.SO.NOSIGPIPE, &one, @sizeOf(c_int));
+    const idle: c_int = 30;
+    const intvl: c_int = 10;
+    const cnt: c_int = 6;
+    _ = c.setsockopt(fd, posix.IPPROTO.TCP, (if (@hasDecl(posix.TCP, "KEEPIDLE")) posix.TCP.KEEPIDLE else posix.TCP.KEEPALIVE), &idle, @sizeOf(c_int));
+    _ = c.setsockopt(fd, posix.IPPROTO.TCP, posix.TCP.KEEPINTVL, &intvl, @sizeOf(c_int));
+    _ = c.setsockopt(fd, posix.IPPROTO.TCP, posix.TCP.KEEPCNT, &cnt, @sizeOf(c_int));
+}
+
+/// A link never blocks in send or recv: macOS ignores MSG_DONTWAIT on a blocking socket, so a stalled peer hung a send.
+fn nonBlocking(fd: Fd) bool {
+    const flags = c.fcntl(fd, posix.F.GETFL);
+    return flags >= 0 and c.fcntl(fd, posix.F.SETFL, flags | @as(c_int, @bitCast(posix.O{ .NONBLOCK = true }))) >= 0;
+}
+
+fn closeOnExec(fd: Fd) void {
+    _ = c.fcntl(fd, posix.F.SETFD, @as(c_int, posix.FD_CLOEXEC));
+}
+
+/// The IPv4 address `host` names, in network order.
+fn resolve(host: []const u8) Error!u32 {
+    var name: [256]u8 = undefined;
+    if (host.len >= name.len) return error.AddressUnresolved;
+    @memcpy(name[0..host.len], host);
+    name[host.len] = 0;
+    const hints: c.addrinfo = .{ .flags = .{}, .family = posix.AF.INET, .socktype = posix.SOCK.STREAM, .protocol = 0, .addrlen = 0, .addr = null, .canonname = null, .next = null };
+    var res: ?*c.addrinfo = null;
+    if (@backingInt(c.getaddrinfo(name[0..host.len :0], null, &hints, &res)) != 0) return error.AddressUnresolved;
+    defer c.freeaddrinfo(res.?);
+    const in: *const posix.sockaddr.in = @ptrCast(@alignCast(res.?.addr.?));
+    return in.addr;
+}
+
+/// A listening socket on `host`'s address only (never every interface), at `port` (0: the kernel picks one).
+pub fn listen(host: []const u8, at: u16) Error!Fd {
+    const ip = try resolve(host);
+    if (ip == 0) return error.WildcardAddress; // 0.0.0.0 would listen on every interface
+    const fd = c.socket(posix.AF.INET, posix.SOCK.STREAM, 0);
+    if (fd < 0) return error.SocketFailed;
+    closeOnExec(fd);
+    errdefer _ = c.close(fd);
+    const one: c_int = 1;
+    _ = c.setsockopt(fd, posix.SOL.SOCKET, posix.SO.REUSEADDR, &one, @sizeOf(c_int));
+    var addr: posix.sockaddr.in = .{ .port = std.mem.nativeToBig(u16, at), .addr = ip };
+    if (c.bind(fd, @ptrCast(&addr), @sizeOf(posix.sockaddr.in)) != 0) return error.AddressInUse;
+    if (c.listen(fd, 8) != 0) return error.SocketFailed;
+    return fd;
+}
+
+/// The port a listening socket holds.
+pub fn port(fd: Fd) Error!u16 {
+    var addr: posix.sockaddr.in = undefined;
+    var len: posix.socklen_t = @sizeOf(posix.sockaddr.in);
+    if (c.getsockname(fd, @ptrCast(&addr), &len) != 0) return error.SocketFailed;
+    return std.mem.bigToNative(u16, addr.port);
+}
+
+/// One connection before `deadline` (monotonic ns).
+pub fn accept(fd: Fd, deadline: u64) Error!Fd {
+    while (true) {
+        try wait(fd, posix.POLL.IN, deadline);
+        const conn = c.accept(fd, null, null);
+        if (conn >= 0) {
+            closeOnExec(conn);
+            if (!nonBlocking(conn)) {
+                _ = c.close(conn);
+                return error.SocketFailed;
+            }
+            tune(conn);
+            return conn;
+        }
+    }
+}
+
+/// Connects to `host:port`, retrying every 50 ms (the peer may not listen yet) until `deadline`.
+pub fn connect(host: []const u8, to: u16, deadline: u64) Error!Fd {
+    const ip = try resolve(host);
+    var addr: posix.sockaddr.in = .{ .port = std.mem.nativeToBig(u16, to), .addr = ip };
+    while (true) {
+        const fd = c.socket(posix.AF.INET, posix.SOCK.STREAM, 0);
+        if (fd < 0) return error.SocketFailed;
+        closeOnExec(fd);
+        if (connectBefore(fd, @ptrCast(&addr), @sizeOf(posix.sockaddr.in), deadline)) {
+            tune(fd);
+            return fd;
+        }
+        _ = c.close(fd);
+        if (nowNs() >= deadline) return error.Timeout;
+        sleepNs(50 * std.time.ns_per_ms);
+    }
+}
+
+/// Connects without blocking; the socket stays non-blocking for `sendAll` and `recvExact`.
+fn connectBefore(fd: Fd, address: *const posix.sockaddr, len: posix.socklen_t, deadline: u64) bool {
+    if (!nonBlocking(fd)) return false;
+    const rc = c.connect(fd, address, len);
+    if (rc == 0) return true;
+    if (posix.errno(rc) != .INPROGRESS) return false;
+    wait(fd, posix.POLL.OUT, deadline) catch return false;
+    var err: c_int = 0;
+    var size: posix.socklen_t = @sizeOf(c_int);
+    return c.getsockopt(fd, posix.SOL.SOCKET, posix.SO.ERROR, @ptrCast(&err), &size) == 0 and err == 0;
+}
+
+/// Until `fd` is ready for `events`, or `error.Timeout` at `deadline`; a hung-up peer counts as ready (the read fails).
+fn wait(fd: Fd, events: i16, deadline: u64) Error!void {
+    while (true) {
+        const now = nowNs();
+        if (now >= deadline) return error.Timeout;
+        var p = [_]c.pollfd{.{ .fd = fd, .events = events, .revents = 0 }};
+        const ms: c_int = @intCast(@min((deadline - now) / std.time.ns_per_ms + 1, 1000));
+        if (c.poll(&p, 1, ms) > 0) return;
+    }
+}
+
+/// Every byte of `bytes` before `deadline`, or `error.Closed` when the peer is gone.
+pub fn sendAll(fd: Fd, bytes: []const u8, deadline: u64) Error!void {
+    var at: usize = 0;
+    while (at < bytes.len) {
+        try wait(fd, posix.POLL.OUT, deadline);
+        const n = c.send(fd, bytes[at..].ptr, bytes.len - at, nosignal | posix.MSG.DONTWAIT);
+        if (n < 0 and (posix.errno(n) == .INTR or posix.errno(n) == .AGAIN)) continue;
+        if (n <= 0) return error.Closed;
+        at += @intCast(n);
+    }
+}
+
+/// Fills `out` before `deadline`, or `error.Closed` at the end of the stream.
+pub fn recvExact(fd: Fd, out: []u8, deadline: u64) Error!void {
+    var at: usize = 0;
+    while (at < out.len) {
+        try wait(fd, posix.POLL.IN, deadline);
+        const n = c.recv(fd, out[at..].ptr, out.len - at, posix.MSG.DONTWAIT);
+        if (n < 0 and (posix.errno(n) == .INTR or posix.errno(n) == .AGAIN)) continue;
+        if (n <= 0) return error.Closed;
+        at += @intCast(n);
+    }
+}
+
+/// Ends the stream both ways, the descriptor kept: the peer sees `error.Closed`, and `close` still frees it.
+pub fn hangUp(fd: Fd) void {
+    _ = c.shutdown(fd, posix.SHUT.RDWR);
+}
+
+pub fn close(fd: Fd) void {
+    hangUp(fd);
+    _ = c.close(fd);
+}
+
+test "stage net: a listener on loopback trades bytes, times out and sees the hang-up" {
+    const l = try listen("127.0.0.1", 0);
+    defer close(l);
+    const deadline = nowNs() + 5 * std.time.ns_per_s;
+    const a = try connect("127.0.0.1", try port(l), deadline);
+    const b = try accept(l, deadline);
+    defer close(b);
+    try sendAll(a, "hello", deadline);
+    var got: [5]u8 = undefined;
+    try recvExact(b, &got, deadline);
+    try std.testing.expectEqualStrings("hello", &got);
+    try std.testing.expectError(error.Timeout, recvExact(b, &got, nowNs() + 20 * std.time.ns_per_ms));
+    close(a);
+    try std.testing.expectError(error.Closed, recvExact(b, &got, deadline));
+}
+
+test "stage net: an unknown host is refused, and a connect with no listener times out" {
+    try std.testing.expectError(error.AddressUnresolved, listen("no-such-host.invalid", 0));
+    try std.testing.expectError(error.WildcardAddress, listen("0.0.0.0", 0));
+    const l = try listen("127.0.0.1", 0);
+    const free = try port(l);
+    close(l);
+    try std.testing.expectError(error.Timeout, connect("127.0.0.1", free, nowNs() + 120 * std.time.ns_per_ms));
+}
diff --git a/zig/src/fabric/stage/ring.zig b/zig/src/fabric/stage/ring.zig
new file mode 100644
index 0000000000..100b1f84f1
--- /dev/null
+++ b/zig/src/fabric/stage/ring.zig
@@ -0,0 +1,163 @@
+//! A ring of pipeline stages over TCP: each rank sends to the next and receives from the one before, in round order.
+const std = @import("std");
+const net = @import("net.zig");
+const frame = @import("frame.zig");
+const secret = @import("secret.zig");
+
+pub const Peer = struct { host: []const u8, port: u16 };
+
+pub const Options = struct {
+    rank: u32,
+    world: u32,
+    /// Every rank's address, in rank order; a rank listens on its own entry only.
+    peers: []const Peer,
+    /// The checkpoint's and layout's fingerprint: a neighbour with another one is refused.
+    fingerprint: [32]u8,
+    /// The ring's shared key (secret.zig): proven both ways after the hello, before any plan or rows frame.
+    key: secret.Key,
+    /// How long the ring may take to form, and each later send or receive by default.
+    timeout_ns: u64 = 60 * std.time.ns_per_s,
+    /// The largest payload a frame may carry (a prompt chunk's rows, a plan).
+    max_bytes: u64 = 256 << 20,
+};
+
+pub const Error = net.Error || frame.Error || secret.Error || error{ BadRing, NotNeighbour, OtherWorld, OtherFingerprint, BadSecret, OutOfOrder, WrongKind, Stopped, Broken };
+
+const hello_bytes = 40; // rank, world, fingerprint
+
+pub const Ring = struct {
+    rank: u32,
+    world: u32,
+    max_bytes: u64,
+    timeout_ns: u64,
+    next: net.Fd,
+    prev: net.Fd,
+    /// The last round sent to the next rank and received from the one before.
+    sent: u64 = 0,
+    received: u64 = 0,
+    /// Set by any failed send or receive: the stream may be mid-frame, so every later call is `error.Broken`.
+    broken: bool = false,
+
+    /// This rank's listener, on its own address only; `join` takes it.
+    pub fn bind(o: Options) Error!net.Fd {
+        try check(o);
+        return net.listen(o.peers[o.rank].host, o.peers[o.rank].port);
+    }
+
+    /// Connects to the next rank, accepts the one before; both ends check the ring and the key (closes `listener`).
+    pub fn join(o: Options, listener: net.Fd) Error!Ring {
+        defer net.close(listener);
+        try check(o);
+        const deadline = net.nowNs() + o.timeout_ns;
+        const nxt = (o.rank + 1) % o.world;
+        const next = try net.connect(o.peers[nxt].host, o.peers[nxt].port, deadline);
+        errdefer net.close(next);
+        const prev = try net.accept(listener, deadline);
+        errdefer net.close(prev);
+        var r: Ring = .{ .rank = o.rank, .world = o.world, .max_bytes = o.max_bytes, .timeout_ns = o.timeout_ns, .next = next, .prev = prev };
+        var mine: [hello_bytes]u8 = undefined;
+        std.mem.writeInt(u32, mine[0..4], o.rank, .little);
+        std.mem.writeInt(u32, mine[4..8], o.world, .little);
+        mine[8..40].* = o.fingerprint;
+        try r.put(.hello, 0, 0, &.{&mine}, deadline);
+        var theirs: [hello_bytes]u8 = undefined;
+        _ = try r.take(.hello, 0, &theirs, deadline);
+        if (std.mem.readInt(u32, theirs[4..8], .little) != o.world) return error.OtherWorld;
+        if (std.mem.readInt(u32, theirs[0..4], .little) != (o.rank + o.world - 1) % o.world) return error.NotNeighbour;
+        if (!std.mem.eql(u8, theirs[8..40], &o.fingerprint)) return error.OtherFingerprint;
+        try prove(o, next, prev, deadline);
+        return r;
+    }
+
+    pub fn open(o: Options) Error!Ring {
+        return join(o, try bind(o));
+    }
+
+    /// Round `round`'s frame of `kind` to the next rank: a header, then `parts` back to back.
+    pub fn send(r: *Ring, kind: frame.Kind, round: u64, rows: u32, parts: []const []const u8) Error!void {
+        if (r.broken) return error.Broken;
+        errdefer r.broken = true;
+        if (kind == .hello or (kind != .stop and round != r.sent + 1 and round != r.sent)) return error.OutOfOrder;
+        try r.put(kind, round, rows, parts, net.nowNs() + r.timeout_ns);
+        r.sent = round;
+    }
+
+    /// Round `round`'s frame of `kind` from the rank before into `out`; its header, the payload at `out[0..bytes]`.
+    pub fn recv(r: *Ring, kind: frame.Kind, round: u64, out: []u8) Error!frame.Header {
+        if (r.broken) return error.Broken;
+        errdefer r.broken = true;
+        const h = try r.take(kind, round, out, net.nowNs() + r.timeout_ns);
+        r.received = round;
+        return h;
+    }
+
+    /// A clean stop to the next rank: it sees `error.Stopped` where it waited for a frame; on a broken link, a hang-up.
+    pub fn stop(r: *Ring) void {
+        if (!r.broken) r.put(.stop, r.sent, 0, &.{}, net.nowNs() + r.timeout_ns) catch {
+            r.broken = true;
+        };
+        if (r.broken) r.hangUp(); // a stop header could complete a half-sent frame
+    }
+
+    /// Both links ended, the descriptors kept for `close`.
+    fn hangUp(r: *Ring) void {
+        net.hangUp(r.next);
+        net.hangUp(r.prev);
+    }
+
+    /// Both links closed: a neighbour waiting on either sees `error.Closed`, so a failed rank ends the ring.
+    pub fn close(r: *Ring) void {
+        net.close(r.next);
+        net.close(r.prev);
+    }
+
+    fn put(r: *Ring, kind: frame.Kind, round: u64, rows: u32, parts: []const []const u8, deadline: u64) Error!void {
+        var bytes: u64 = 0;
+        for (parts) |p| bytes += p.len;
+        if (bytes > r.max_bytes) return error.TooLarge;
+        const head = frame.encode(.{ .kind = kind, .round = round, .rows = rows, .bytes = bytes });
+        try net.sendAll(r.next, &head, deadline);
+        for (parts) |p| try net.sendAll(r.next, p, deadline);
+    }
+
+    fn take(r: *Ring, kind: frame.Kind, round: u64, out: []u8, deadline: u64) Error!frame.Header {
+        var raw: [frame.size]u8 = undefined;
+        try net.recvExact(r.prev, &raw, deadline);
+        const h = try frame.decode(&raw, @min(r.max_bytes, out.len));
+        if (h.kind == .stop and kind != .stop) return error.Stopped;
+        if (h.kind != kind) return error.WrongKind;
+        if (h.round != round) return error.OutOfOrder;
+        try net.recvExact(r.prev, out[0..@intCast(h.bytes)], deadline);
+        return h;
+    }
+};
+
+/// Each link's mutual proof of the key: nonces both ways, then each end's HMAC over them; the key never crosses.
+fn prove(o: Options, next: net.Fd, prev: net.Fd, deadline: u64) Error!void {
+    const mine_n = try secret.nonce();
+    const mine_p = try secret.nonce();
+    try net.sendAll(next, &mine_n, deadline);
+    try net.sendAll(prev, &mine_p, deadline);
+    var their_n: secret.Nonce = undefined;
+    var their_p: secret.Nonce = undefined;
+    try net.recvExact(next, &their_n, deadline);
+    try net.recvExact(prev, &their_p, deadline);
+    const out_n = secret.proof(&o.key, .connector, o.rank, o.world, &o.fingerprint, &their_n, &mine_n);
+    const out_p = secret.proof(&o.key, .acceptor, o.rank, o.world, &o.fingerprint, &their_p, &mine_p);
+    try net.sendAll(next, &out_n, deadline);
+    try net.sendAll(prev, &out_p, deadline);
+    var got: secret.Proof = undefined;
+    try net.recvExact(prev, &got, deadline);
+    const before = (o.rank + o.world - 1) % o.world;
+    if (!secret.same(&got, &secret.proof(&o.key, .connector, before, o.world, &o.fingerprint, &mine_p, &their_p))) return error.BadSecret;
+    try net.recvExact(next, &got, deadline);
+    if (!secret.same(&got, &secret.proof(&o.key, .acceptor, (o.rank + 1) % o.world, o.world, &o.fingerprint, &mine_n, &their_n))) return error.BadSecret;
+}
+
+fn check(o: Options) Error!void {
+    if (o.world < 2 or o.rank >= o.world or o.peers.len != o.world) return error.BadRing;
+}
+
+test {
+    _ = @import("ring_test.zig");
+}
diff --git a/zig/src/fabric/stage/ring_test.zig b/zig/src/fabric/stage/ring_test.zig
new file mode 100644
index 0000000000..b5f2a80931
--- /dev/null
+++ b/zig/src/fabric/stage/ring_test.zig
@@ -0,0 +1,243 @@
+//! Rings of 2 and 3 ranks on loopback threads: rows go round exact and in order, and every refusal fires as planted.
+const std = @import("std");
+const ring = @import("ring.zig");
+const net = @import("net.zig");
+const t = std.testing;
+
+const most = 3;
+const fp_a: [32]u8 = @splat(0xA5);
+const key_a: [32]u8 = @splat(0x3C);
+
+/// Every rank's listener first (kernel-picked ports), then their options.
+const Setup = struct {
+    peers: [most]ring.Peer,
+    listeners: [most]net.Fd,
+    opts: [most]ring.Options,
+
+    fn init(s: *Setup, world: u32) !void {
+        for (0..world) |i| {
+            s.listeners[i] = try net.listen("127.0.0.1", 0);
+            s.peers[i] = .{ .host = "127.0.0.1", .port = try net.port(s.listeners[i]) };
+        }
+        for (0..world) |i| s.opts[i] = .{ .rank = @intCast(i), .world = world, .peers = s.peers[0..world], .fingerprint = fp_a, .key = key_a, .timeout_ns = 5 * std.time.ns_per_s };
+    }
+};
+
+const Joined = struct { r: ?ring.Ring = null, err: ?anyerror = null };
+
+fn isErr(got: ?anyerror, want: anyerror) bool {
+    return if (got) |e| e == want else false;
+}
+
+fn joinOne(o: ring.Options, l: net.Fd, out: *Joined) void {
+    out.r = ring.Ring.join(o, l) catch |e| {
+        out.err = e;
+        return;
+    };
+}
+
+/// Joins `world` ranks at once; each slot holds its ring or its error.
+fn joinAll(s: *Setup, world: u32, out: []Joined) !void {
+    var th: [most]std.Thread = undefined;
+    for (0..world) |i| th[i] = try std.Thread.spawn(.{}, joinOne, .{ s.opts[i], s.listeners[i], &out[i] });
+    for (th[0..world]) |x| x.join();
+}
+
+/// One round on rank `rank` of the ring: rank 0 sends `rows` and receives what the last rank sends back.
+fn stage(r: *ring.Ring, round: u64, rows: []const u8, out: []u8, delay_ms: u64) !void {
+    if (r.rank == 0) {
+        try r.send(.plan, round, 0, &.{std.mem.asBytes(&round)});
+        try r.send(.rows, round, @intCast(rows.len / 8), &.{ rows[0 .. rows.len / 2], rows[rows.len / 2 ..] });
+        _ = try r.recv(.rows, round, out);
+        return;
+    }
+    var plan: [8]u8 = undefined;
+    _ = try r.recv(.plan, round, &plan);
+    const h = try r.recv(.rows, round, out);
+    if (delay_ms > 0) net.sleepNs(delay_ms * std.time.ns_per_ms);
+    for (out[0..@intCast(h.bytes)]) |*b| b.* +%= 1; // a stage's work: every byte plus one
+    if (r.rank + 1 < r.world) try r.send(.plan, round, 0, &.{&plan});
+    try r.send(.rows, round, h.rows, &.{out[0..@intCast(h.bytes)]});
+}
+
+fn follower(r: *ring.Ring, rounds: u64, bytes: usize, delay_ms: u64, err: *?anyerror) void {
+    var buf: [4096]u8 = undefined;
+    var round: u64 = 1;
+    while (round <= rounds) : (round += 1) stage(r, round, buf[0..bytes], &buf, delay_ms) catch |e| {
+        err.* = e;
+        r.close();
+        return;
+    };
+}
+
+fn roundTrip(world: u32, delay_ms: u64) !void {
+    var s: Setup = undefined;
+    try s.init(world);
+    var j: [most]Joined = @splat(.{});
+    try joinAll(&s, world, &j);
+    for (j[0..world]) |x| try t.expect(x.err == null);
+    defer for (j[0..world]) |*x| x.r.?.close();
+    var errs: [most]?anyerror = @splat(null);
+    var th: [most]std.Thread = undefined;
+    for (1..world) |i| th[i] = try std.Thread.spawn(.{}, follower, .{ &j[i].r.?, 3, 0, delay_ms, &errs[i] });
+    var rows: [4096]u8 = undefined;
+    var back: [4096]u8 = undefined;
+    for (1..4) |round| {
+        for (&rows, 0..) |*b, i| b.* = @truncate(i * 7 + round);
+        try stage(&j[0].r.?, round, &rows, &back, 0);
+        for (rows, back) |a, b| try t.expectEqual(a +% @as(u8, @intCast(world - 1)), b);
+    }
+    for (th[1..world]) |x| x.join();
+    for (errs[1..world]) |e| try t.expect(e == null);
+}
+
+test "stage ring: two and three ranks carry a plan and split rows round in order, every byte exact" {
+    try roundTrip(2, 0);
+    try roundTrip(3, 0);
+}
+
+test "stage ring: a slow middle stage only delays the ring, the bytes are the same" {
+    try roundTrip(3, 30);
+}
+
+test "stage ring: another fingerprint or another world is refused at the hello" {
+    var s: Setup = undefined;
+    try s.init(2);
+    s.opts[1].fingerprint[0] ^= 1;
+    var j: [most]Joined = @splat(.{});
+    try joinAll(&s, 2, &j);
+    for (j[0..2]) |*x| if (x.r) |*r| r.close();
+    try t.expect(isErr(j[0].err, error.OtherFingerprint) or isErr(j[1].err, error.OtherFingerprint));
+    try s.init(2);
+    s.opts[0].world = 3;
+    s.opts[0].peers = &.{ s.peers[0], s.peers[1], s.peers[1] };
+    j = @splat(.{});
+    try joinAll(&s, 2, &j);
+    for (j[0..2]) |*x| if (x.r) |*r| r.close();
+    try t.expect(j[0].err != null or j[1].err != null);
+}
+
+/// Two joined ranks on loopback.
+fn pair(j: *[most]Joined) !void {
+    var s: Setup = undefined;
+    try s.init(2);
+    j.* = @splat(.{});
+    try joinAll(&s, 2, j);
+    for (j[0..2]) |x| try t.expect(x.err == null);
+}
+
+fn closePair(j: *[most]Joined) void {
+    j[0].r.?.close();
+    j[1].r.?.close();
+}
+
+test "stage ring: a rank without the key is refused by both its neighbours, in a ring of two and of three" {
+    for ([_]u32{ 2, 3 }) |world| {
+        var s: Setup = undefined;
+        try s.init(world);
+        s.opts[1].key[0] ^= 1;
+        var j: [most]Joined = @splat(.{});
+        try joinAll(&s, world, &j);
+        for (j[0..world]) |*x| if (x.r) |*r| r.close();
+        try t.expect(isErr(j[0].err, error.BadSecret) or isErr(j[0].err, error.Closed));
+        try t.expect(isErr(j[1].err, error.BadSecret) or isErr(j[1].err, error.Closed));
+        try t.expect(isErr(j[0].err, error.BadSecret) or isErr(j[1].err, error.BadSecret));
+        if (world == 3) try t.expect(isErr(j[2].err, error.BadSecret) or isErr(j[2].err, error.Closed));
+    }
+}
+
+test "stage ring: a stranger that connects and talks first is refused before any secret byte is sent" {
+    var s: Setup = undefined;
+    try s.init(2);
+    const stranger = try net.connect("127.0.0.1", s.peers[1].port, net.nowNs() + std.time.ns_per_s);
+    defer net.close(stranger);
+    try net.sendAll(stranger, "GET / HTTP/1.1\r\nHost: stage-ring\r\n\r\nxxxxxxxx", net.nowNs() + std.time.ns_per_s);
+    defer net.close(s.listeners[0]); // rank 0's backlog takes rank 1's connect, so rank 1 reaches its accept
+    var o = s.opts[1];
+    o.timeout_ns = 500 * std.time.ns_per_ms;
+    try t.expectError(error.BadMagic, ring.Ring.join(o, s.listeners[1]));
+}
+
+test "stage ring: a frame out of round or of another kind is refused, and the link is broken after it" {
+    var buf: [16]u8 = undefined;
+    var j: [most]Joined = undefined;
+    try pair(&j);
+    try t.expectError(error.OutOfOrder, j[0].r.?.send(.rows, 5, 0, &.{}));
+    try t.expectError(error.Broken, j[0].r.?.send(.rows, 1, 0, &.{}));
+    closePair(&j);
+    try pair(&j);
+    try j[0].r.?.send(.rows, 1, 0, &.{"abc"});
+    try t.expectError(error.OutOfOrder, j[1].r.?.recv(.rows, 2, &buf));
+    try t.expectError(error.Broken, j[1].r.?.recv(.rows, 1, &buf));
+    closePair(&j);
+    try pair(&j);
+    try j[0].r.?.send(.plan, 1, 0, &.{"x"});
+    try t.expectError(error.WrongKind, j[1].r.?.recv(.rows, 1, &buf));
+    closePair(&j);
+}
+
+test "stage ring: a stop ends the next rank's wait" {
+    var buf: [16]u8 = undefined;
+    var j: [most]Joined = undefined;
+    try pair(&j);
+    defer closePair(&j);
+    try j[0].r.?.send(.rows, 1, 2, &.{"ab"});
+    _ = try j[1].r.?.recv(.rows, 1, &buf);
+    j[0].r.?.stop();
+    try t.expectError(error.Stopped, j[1].r.?.recv(.rows, 2, &buf));
+}
+
+test "stage ring: a stop after a send that failed mid-payload hangs up, so the next rank never sees a whole frame" {
+    const big = 64 << 20; // past loopback's socket buffers, so the send times out partway
+    const rows = try t.allocator.alloc(u8, big);
+    defer t.allocator.free(rows);
+    @memset(rows, 0x5A);
+    const out = try t.allocator.alloc(u8, big);
+    defer t.allocator.free(out);
+    var j: [most]Joined = undefined;
+    try pair(&j);
+    defer closePair(&j);
+    j[0].r.?.timeout_ns = 200 * std.time.ns_per_ms;
+    try t.expectError(error.Timeout, j[0].r.?.send(.rows, 1, 1, &.{rows}));
+    j[0].r.?.stop();
+    try t.expectError(error.Closed, j[1].r.?.recv(.rows, 1, out));
+}
+
+test "stage ring: a frame past the receiver's buffer is refused before its payload is read" {
+    var s: Setup = undefined;
+    try s.init(2);
+    var j: [most]Joined = @splat(.{});
+    try joinAll(&s, 2, &j);
+    defer j[0].r.?.close();
+    defer j[1].r.?.close();
+    try j[0].r.?.send(.rows, 1, 1, &.{"0123456789"});
+    var small: [4]u8 = undefined;
+    try t.expectError(error.TooLarge, j[1].r.?.recv(.rows, 1, &small));
+}
+
+test "stage ring: a rank that fails closes its links, and both neighbours of a three-ring see it" {
+    var s: Setup = undefined;
+    try s.init(3);
+    var j: [most]Joined = @splat(.{});
+    try joinAll(&s, 3, &j);
+    defer j[0].r.?.close();
+    defer j[2].r.?.close();
+    j[1].r.?.close(); // rank 1 dies
+    var buf: [16]u8 = undefined;
+    try t.expectError(error.Closed, j[2].r.?.recv(.rows, 1, &buf));
+    var closed = false;
+    for (0..50) |_| {
+        j[0].r.?.send(.rows, 1, 0, &.{"xxxxxxxxxxxxxxxx"}) catch |e| {
+            closed = isErr(e, error.Closed);
+            break;
+        };
+    }
+    try t.expect(closed);
+}
+
+test "stage ring: a ring needs two ranks, a rank inside it and one address a rank" {
+    const p = [_]ring.Peer{.{ .host = "127.0.0.1", .port = 1 }};
+    try t.expectError(error.BadRing, ring.Ring.bind(.{ .rank = 0, .world = 1, .peers = &p, .fingerprint = fp_a, .key = key_a }));
+    try t.expectError(error.BadRing, ring.Ring.bind(.{ .rank = 2, .world = 2, .peers = &.{ p[0], p[0] }, .fingerprint = fp_a, .key = key_a }));
+    try t.expectError(error.BadRing, ring.Ring.bind(.{ .rank = 0, .world = 2, .peers = &p, .fingerprint = fp_a, .key = key_a }));
+}
diff --git a/zig/src/fabric/stage/secret.zig b/zig/src/fabric/stage/secret.zig
new file mode 100644
index 0000000000..87541c9e51
--- /dev/null
+++ b/zig/src/fabric/stage/secret.zig
@@ -0,0 +1,143 @@
+//! The ring's shared secret: a key from an owner-only file, and each link's mutual HMAC proof over fresh nonces.
+const std = @import("std");
+const c = std.c;
+const builtin = @import("builtin");
+const Hmac = std.crypto.auth.hmac.sha2.HmacSha256;
+const Sha256 = std.crypto.hash.sha2.Sha256;
+
+pub const Key = [32]u8;
+pub const Nonce = [32]u8;
+pub const Proof = [Hmac.mac_length]u8;
+
+/// The fewest secret bytes a key file holds.
+pub const least_bytes = 16;
+
+pub const Error = error{ SecretUnreadable, SecretNotPrivate, SecretTooShort, SecretTooLong, NoRandom };
+
+/// Which end of a link proves: the rank that connected, or the one that accepted.
+pub const Role = enum(u8) { connector = 'C', acceptor = 'A' };
+
+/// The key for the secret in `path`: its SHA-256, trailing whitespace dropped; owner-only and at most 4 KiB.
+pub fn fromFile(path: [:0]const u8) Error!Key {
+    const fd = c.open(path, .{ .ACCMODE = .RDONLY, .CLOEXEC = true });
+    if (fd < 0) return error.SecretUnreadable;
+    defer _ = c.close(fd);
+    if (try mode(fd) & 0o077 != 0) return error.SecretNotPrivate;
+    var buf: [4096]u8 = undefined;
+    var n: usize = 0;
+    while (n < buf.len) {
+        const got = c.read(fd, buf[n..].ptr, buf.len - n);
+        if (got < 0) return error.SecretUnreadable;
+        if (got == 0) break;
+        n += @intCast(got);
+    }
+    defer std.crypto.secureZero(u8, &buf);
+    if (n == buf.len) {
+        var more: [1]u8 = undefined;
+        if (c.read(fd, &more, 1) != 0) return error.SecretTooLong;
+    }
+    return fromBytes(buf[0..n]);
+}
+
+/// `fd`'s permission bits: statx on Linux (its libc has no fstat), fstat elsewhere.
+fn mode(fd: c.fd_t) Error!u32 {
+    if (builtin.os.tag == .linux) {
+        const linux = std.os.linux;
+        var sx: linux.Statx = undefined;
+        if (c.statx(fd, "", 0x1000, .{ .MODE = true }, &sx) != 0) return error.SecretUnreadable; // AT_EMPTY_PATH
+        return sx.mode;
+    }
+    var st: c.Stat = undefined;
+    if (c.fstat(fd, &st) != 0) return error.SecretUnreadable;
+    return @intCast(st.mode);
+}
+
+/// The key for secret `bytes` (trailing whitespace dropped).
+pub fn fromBytes(bytes: []const u8) Error!Key {
+    const s = std.mem.trimEnd(u8, bytes, " \t\r\n");
+    if (s.len < least_bytes) return error.SecretTooShort;
+    var key: Key = undefined;
+    Sha256.hash(s, &key, .{});
+    return key;
+}
+
+/// 32 bytes from the kernel's random source (/dev/urandom on Linux and macOS alike).
+pub fn nonce() Error!Nonce {
+    var n: Nonce = undefined;
+    const fd = c.open("/dev/urandom", .{ .ACCMODE = .RDONLY, .CLOEXEC = true });
+    if (fd < 0) return error.NoRandom;
+    defer _ = c.close(fd);
+    var at: usize = 0;
+    while (at < n.len) {
+        const got = c.read(fd, n[at..].ptr, n.len - at);
+        if (got <= 0) return error.NoRandom;
+        at += @intCast(got);
+    }
+    return n;
+}
+
+/// The proof the `role` end of a link (rank `rank` of `world`) sends: its peer's nonce, then its own.
+pub fn proof(key: *const Key, role: Role, rank: u32, world: u32, fingerprint: *const [32]u8, theirs: *const Nonce, mine: *const Nonce) Proof {
+    var h = Hmac.init(key);
+    h.update("TFST link v1");
+    h.update(&.{@backingInt(role)});
+    var ints: [8]u8 = undefined;
+    std.mem.writeInt(u32, ints[0..4], rank, .little);
+    std.mem.writeInt(u32, ints[4..8], world, .little);
+    h.update(&ints);
+    h.update(fingerprint);
+    h.update(theirs);
+    h.update(mine);
+    var out: Proof = undefined;
+    h.final(&out);
+    return out;
+}
+
+/// Whether `got` is the proof `want`, compared in constant time.
+pub fn same(got: *const Proof, want: *const Proof) bool {
+    return std.crypto.timing_safe.eql(Proof, got.*, want.*);
+}
+
+test "stage secret: proofs bind the key, the role, the rank, the fingerprint and both nonces" {
+    const k = try fromBytes("correct horse battery staple\n");
+    const other = try fromBytes("correct horse battery staplf");
+    const fp: [32]u8 = @splat(7);
+    const a = try nonce();
+    const b = try nonce();
+    const p = proof(&k, .connector, 1, 3, &fp, &a, &b);
+    try std.testing.expect(same(&p, &proof(&k, .connector, 1, 3, &fp, &a, &b)));
+    try std.testing.expect(!same(&p, &proof(&other, .connector, 1, 3, &fp, &a, &b)));
+    try std.testing.expect(!same(&p, &proof(&k, .acceptor, 1, 3, &fp, &a, &b)));
+    try std.testing.expect(!same(&p, &proof(&k, .connector, 2, 3, &fp, &a, &b)));
+    try std.testing.expect(!same(&p, &proof(&k, .connector, 1, 3, &fp, &b, &a)));
+    var fp2 = fp;
+    fp2[0] ^= 1;
+    try std.testing.expect(!same(&p, &proof(&k, .connector, 1, 3, &fp2, &a, &b)));
+    try std.testing.expect(!std.mem.eql(u8, &a, &b));
+}
+
+test "stage secret: a key file must be owner-only and hold 16 bytes to 4 KiB" {
+    try std.testing.expectError(error.SecretTooShort, fromBytes("short secret \n"));
+    try std.testing.expectError(error.SecretUnreadable, fromFile("/nonexistent/stage-secret"));
+    var dir_buf: [64]u8 = undefined;
+    const path = try std.fmt.bufPrintSentinel(&dir_buf, "/tmp/tf-stage-secret-{d}", .{c.getpid()}, 0);
+    defer _ = c.unlink(path);
+    for ([_]struct { mode: c.mode_t, ok: bool }{ .{ .mode = 0o600, .ok = true }, .{ .mode = 0o644, .ok = false }, .{ .mode = 0o640, .ok = false } }) |case| {
+        _ = c.unlink(path);
+        const fd = c.open(path, .{ .ACCMODE = .WRONLY, .CREAT = true, .EXCL = true }, case.mode);
+        try std.testing.expect(fd >= 0);
+        _ = c.fchmod(fd, case.mode); // the umask may have narrowed it
+        _ = c.write(fd, "0123456789abcdef0123\n", 21);
+        _ = c.close(fd);
+        if (case.ok) {
+            try std.testing.expectEqual(try fromBytes("0123456789abcdef0123"), try fromFile(path));
+        } else try std.testing.expectError(error.SecretNotPrivate, fromFile(path));
+    }
+    _ = c.unlink(path);
+    const fd = c.open(path, .{ .ACCMODE = .WRONLY, .CREAT = true, .EXCL = true }, @as(c.mode_t, 0o600));
+    try std.testing.expect(fd >= 0);
+    const big: [4097]u8 = @splat('k');
+    _ = c.write(fd, &big, big.len);
+    _ = c.close(fd);
+    try std.testing.expectError(error.SecretTooLong, fromFile(path));
+}
diff --git a/zig/src/fabric/stage/stage.zig b/zig/src/fabric/stage/stage.zig
new file mode 100644
index 0000000000..f30db6501d
--- /dev/null
+++ b/zig/src/fabric/stage/stage.zig
@@ -0,0 +1,18 @@
+//! The pipeline stage link: a TCP ring of ranks carrying each round's plan and hidden rows, on Linux and macOS.
+pub const net = @import("net.zig");
+pub const frame = @import("frame.zig");
+pub const ring = @import("ring.zig");
+pub const config = @import("config.zig");
+pub const secret = @import("secret.zig");
+
+pub const Ring = ring.Ring;
+pub const Options = ring.Options;
+pub const Peer = ring.Peer;
+
+test {
+    _ = net;
+    _ = frame;
+    _ = ring;
+    _ = config;
+    _ = secret;
+}