Skip to content

cli: run the pull tests' fake hub on std.Io.net so Windows builds them #2284

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ashhart merged 3 commits into from
Nov 7, 2026
Merged

cli: run the pull tests' fake hub on std.Io.net so Windows builds them #2284

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5487287
cli: run the pull tests' fake hub on std.Io.net so Windows builds them
ElderOrb Nov 5, 2026
afbfdca
ci: drop the workflow's byte-order mark and note the Windows pull tests
ElderOrb Nov 5, 2026
35f233a
changelog: #2284 under Unreleased
ashhart Nov 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/workflows/no-gpu.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,6 +176,9 @@ jobs:
run: zig build -Dfixed-cpu test-golden
- name: zig build fuzz-cortheon
run: zig build -Dfixed-cpu fuzz-cortheon
# The pull client's framing and path checks: its fake hub runs on std.Io.net, so Windows builds and runs them.
- name: zig build test-pull-security
run: zig build -Dfixed-cpu test-pull-security
- name: Save the build cache
if: ${{ !cancelled() && github.event_name != 'pull_request' && github.ref == 'refs/heads/main' }}
uses: actions/cache/save@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
Expand Down
2 changes: 2 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,8 @@ Each release's page on GitHub has its notes and measurements. See [the 1.0.0 rel
- Qwen3.8-27B and the other dense Qwen3.5 models on CUDA: streams run on KV pool pages of 256 positions by default, the engine's own stream too, so `--parallel` is now a cap rather than a reservation: a request can wait for pages instead of every stream holding a full-context window. `TF_QWEN_KV_PAGES=0` keeps a window each, as before. Qwen3.6-35B-A3B and the other MoE checkpoints stay on windows by default, with pages opt-in (`TF_QWEN_KV_PAGES=256`). Paged prompt attention stages a chunk's page ids in shared memory, a decode thread looks its page up once a page, and a paged prompt-cache save or restore copies one 2D block a page. Replies are the same tokens on pages and windows (#2011). Thanks to @jschmied.
- Qwen3.5 and Qwen3.6 on CUDA: KV pool pages stay opt-in (`TF_QWEN_KV_PAGES=256`), dense and MoE alike; unset keeps a window each, as before, and `TF_QWEN_KV_PAGES_ALL` unset still leaves the engine's own stream a window. With pages set, `--parallel` is a cap rather than a reservation: a request can wait for pages instead of every stream holding a full-context window. Paged prompt attention stages a chunk's page ids in shared memory, a decode thread looks its page up once a page, and a paged prompt-cache save or restore copies one 2D block a page. Replies are the same tokens on pages and windows (#2011). Thanks to @jschmied.
- Qwen3.5 and Qwen3.6 on CUDA: KV pool pages stay opt-in (`TF_QWEN_KV_PAGES=256`), dense and MoE alike; unset keeps a window each, as before, and `TF_QWEN_KV_PAGES_ALL` unset still leaves the engine's own stream a window. With pages set, `--parallel` is a cap rather than a reservation: a request can wait for pages instead of every stream holding a full-context window. Paged prompt attention stages a chunk's page ids in shared memory, a decode thread looks its page up once a page, and a paged prompt-cache save or restore copies one 2D block a page. Replies are the same tokens on pages and windows (#2011, part of #1943). Thanks to @jschmied.
- Tests: the pull client's security tests build and run on Windows, on a portable fake hub, and CI runs them there
(#2284). Thanks to @ElderOrb.

- Metal: a request's finished stats list its prompt pass widths (`tensorfold.prefill_widths`, one entry per completed
target pass) on every Metal text family except Flash Next, whose engine reports none. Telemetry only (#2293).
Expand Down
165 changes: 93 additions & 72 deletions zig/src/cli/pull_test.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,12 @@ const exlr8_pull = @import("pull_exlr8.zig");

/// A fake hub on 127.0.0.1: revision and tree APIs, ranged resolve endpoints, a body cut once, a repo ignoring ranges.
const FakeHub = struct {
server_fd: std.posix.socket_t,
server: std.Io.net.Server,
io: std.Io,
threaded: *std.Io.Threaded,
stop: std.atomic.Value(bool) = .init(false),
live: std.ArrayList(std.Io.net.Stream) = .empty,
live_lock: std.Io.Mutex = .init,
thread: ?std.Thread = null,
weight_requests: std.atomic.Value(u32) = .init(0),
flaky_requests: std.atomic.Value(u32) = .init(0),
Expand All @@ -33,20 +37,13 @@ const FakeHub = struct {
const nested_body: []const u8 = &nested_body_storage;

fn open() !FakeHub {
const posix = std.posix;
const rc = posix.system.socket(posix.AF.INET, posix.SOCK.STREAM, 0);
if (posix.errno(rc) != .SUCCESS) return error.SocketFailed;
const fd: posix.socket_t = @intCast(rc);
errdefer _ = posix.system.close(fd);
const one: c_int = 1;
try posix.setsockopt(fd, posix.SOL.SOCKET, posix.SO.REUSEADDR, std.mem.asBytes(&one));
var addr: posix.sockaddr.in = .{ .family = posix.AF.INET, .port = std.mem.nativeToBig(u16, 0), .addr = std.mem.nativeToBig(u32, 0x7f000001) };
if (posix.errno(posix.system.bind(fd, @ptrCast(&addr), @sizeOf(posix.sockaddr.in))) != .SUCCESS) return error.BindFailed;
if (posix.errno(posix.system.listen(fd, 32)) != .SUCCESS) return error.ListenFailed;
var bound: posix.sockaddr.in = undefined;
var len: posix.socklen_t = @sizeOf(posix.sockaddr.in);
if (posix.errno(posix.system.getsockname(fd, @ptrCast(&bound), &len)) != .SUCCESS) return error.BindFailed;
const port = std.mem.bigToNative(u16, bound.port);
const threaded = try std.testing.allocator.create(std.Io.Threaded);
threaded.* = std.Io.Threaded.init(std.testing.allocator, .{});
errdefer std.testing.allocator.destroy(threaded);
const io = threaded.io();
const loopback: std.Io.net.IpAddress = .{ .ip4 = .loopback(0) };
var server = try loopback.listen(io, .{});
errdefer server.deinit(io);
var hash = std.crypto.hash.sha2.Sha256.init(.{});
hash.update(weights_body);
var digest: [32]u8 = undefined;
Expand All @@ -56,15 +53,17 @@ const FakeHub = struct {
var nested_digest: [32]u8 = undefined;
nested_hash.final(&nested_digest);
return .{
.server_fd = fd,
.server = server,
.io = io,
.threaded = threaded,
.weights = weights_body,
.config = config_body,
.nested = nested_body,
.weights_sha_hex = try std.fmt.allocPrint(std.testing.allocator, "{s}", .{std.fmt.bytesToHex(digest, .lower)}),
.nested_sha_hex = try std.fmt.allocPrint(std.testing.allocator, "{s}", .{std.fmt.bytesToHex(nested_digest, .lower)}),
.config_git_hex = gitHex(config_body),
.flaky_git_hex = gitHex(flaky_body),
.port = port,
.port = server.socket.address.getPort(),
};
}

Expand All @@ -85,57 +84,60 @@ const FakeHub = struct {
var n: usize = 0;
defer for (conns[0..n]) |t| t.join();
while (!fake.stop.load(.acquire)) {
var fds = [_]std.posix.pollfd{.{ .fd = fake.server_fd, .events = std.posix.POLL.IN, .revents = 0 }};
const ready = std.posix.poll(&fds, 50) catch return;
if (ready == 0) continue;
const rc = std.posix.system.accept(fake.server_fd, null, null);
if (std.posix.errno(rc) != .SUCCESS) return;
const conn_fd: std.posix.socket_t = @intCast(rc);
const conn = fake.server.accept(fake.io) catch return;
if (fake.stop.load(.acquire)) {
conn.close(fake.io);
return;
}
if (n == conns.len) {
_ = std.posix.system.close(conn_fd);
conn.close(fake.io);
continue;
}
conns[n] = std.Thread.spawn(.{}, handle, .{ fake, conn_fd }) catch {
_ = std.posix.system.close(conn_fd);
fake.live_lock.lockUncancelable(fake.io);
fake.live.append(std.testing.allocator, conn) catch {};
fake.live_lock.unlock(fake.io);
conns[n] = std.Thread.spawn(.{}, handle, .{ fake, conn }) catch {
fake.forget(conn);
conn.close(fake.io);
continue;
};
n += 1;
}
}

fn handle(fake: *FakeHub, fd: std.posix.socket_t) void {
defer _ = std.posix.system.close(fd);
fn handle(fake: *FakeHub, conn: std.Io.net.Stream) void {
defer {
fake.forget(conn);
conn.close(fake.io);
}
var buf: [8192]u8 = undefined;
var end: usize = 0;
var read_buf: [4096]u8 = undefined;
var reader = conn.reader(fake.io, &read_buf);
while (true) {
const head_end = std.mem.indexOf(u8, buf[0..end], "\r\n\r\n") orelse {
if (end == buf.len) return;
var fds = [_]std.posix.pollfd{.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 }};
const ready = std.posix.poll(&fds, 50) catch return;
if (ready == 0) {
if (fake.stop.load(.acquire)) return;
continue;
}
const n = std.posix.read(fd, buf[end..]) catch return;
var vecs = [_][]u8{buf[end..]};
const n = reader.interface.readVec(&vecs) catch return;
if (n == 0) return;
end += n;
continue;
};
if (!fake.respondOne(fd, buf[0..head_end])) return;
if (!fake.respondOne(conn, buf[0..head_end])) return;
const rest = end - (head_end + 4);
std.mem.copyForwards(u8, buf[0..rest], buf[head_end + 4 .. end]);
end = rest;
}
}

/// Answers one request; false closes the connection.
fn respondOne(fake: *FakeHub, fd: std.posix.socket_t, head: []const u8) bool {
fn respondOne(fake: *FakeHub, conn: std.Io.net.Stream, head: []const u8) bool {
var tree_buf: [256]u8 = undefined;
const line_end = std.mem.indexOf(u8, head, "\r\n") orelse return false;
var parts = std.mem.tokenizeScalar(u8, head[0..line_end], ' ');
const method = parts.next() orelse return false;
if (std.mem.eql(u8, method, "HEAD")) {
sendAll(fd, "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: keep-alive\r\n\r\n"); // no redirect: the hub's link is the storage link
fake.sendAll(conn, "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: keep-alive\r\n\r\n"); // no redirect: the hub's link is the storage link
return true;
}
if (!std.mem.eql(u8, method, "GET")) return false;
Expand All @@ -157,86 +159,86 @@ const FakeHub = struct {
}
}
if (std.mem.eql(u8, path, "/api/models/Org/Flash/revision/main")) {
respond(fd, "{\"sha\": \"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1\"}");
fake.respond(conn, "{\"sha\": \"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1\"}");
} else if (std.mem.eql(u8, path, "/api/models/Org/Whole/revision/main")) {
respond(fd, "{\"sha\": \"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb2\"}");
fake.respond(conn, "{\"sha\": \"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb2\"}");
} else if (std.mem.eql(u8, path, "/api/models/Org/Draft/revision/main")) {
respond(fd, "{\"sha\": \"ccccccccccccccccccccccccccccccccccccccc3\"}");
fake.respond(conn, "{\"sha\": \"ccccccccccccccccccccccccccccccccccccccc3\"}");
} else if (std.mem.eql(u8, path, "/api/models/Org/Draft/tree/ccccccccccccccccccccccccccccccccccccccc3")) {
respond(fd, "[{\"type\": \"file\", \"path\": \"config.json\", \"size\": 27}]");
fake.respond(conn, "[{\"type\": \"file\", \"path\": \"config.json\", \"size\": 27}]");
} else if (std.mem.eql(u8, path, "/Org/Draft/resolve/ccccccccccccccccccccccccccccccccccccccc3/config.json")) {
respond(fd, "{\"model_type\": \"gemma3n\"}");
fake.respond(conn, "{\"model_type\": \"gemma3n\"}");
} else if (std.mem.eql(u8, path, "/api/models/Org/DFlash/revision/main")) {
respond(fd, "{\"sha\": \"ddddddddddddddddddddddddddddddddddddddd4\"}");
fake.respond(conn, "{\"sha\": \"ddddddddddddddddddddddddddddddddddddddd4\"}");
} else if (std.mem.eql(u8, path, "/api/models/Org/DFlash/tree/ddddddddddddddddddddddddddddddddddddddd4")) {
respond(fd, std.fmt.bufPrint(&tree_buf, "[{{\"type\": \"file\", \"path\": \"config.json\", \"size\": {d}}}]", .{drafter_config.len}) catch return false);
fake.respond(conn, std.fmt.bufPrint(&tree_buf, "[{{\"type\": \"file\", \"path\": \"config.json\", \"size\": {d}}}]", .{drafter_config.len}) catch return false);
} else if (std.mem.eql(u8, path, "/Org/DFlash/resolve/ddddddddddddddddddddddddddddddddddddddd4/config.json")) {
respond(fd, drafter_config);
fake.respond(conn, drafter_config);
} else if (std.mem.eql(u8, path, "/api/models/Org/Flash/tree/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1") or std.mem.eql(u8, path, "/api/models/Org/Whole/tree/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb2")) {
// The hub's tree sends LFS oids as bare hex.
const tree = std.fmt.allocPrint(std.testing.allocator, "[{{\"type\": \"file\", \"oid\": \"{s}\", \"path\": \"config.json\", \"size\": {d}}}, {{\"type\": \"file\", \"oid\": \"1111111111111111111111111111111111111111\", \"path\": \"weights.safetensors\", \"size\": {d}, \"lfs\": {{\"oid\": \"{s}\", \"size\": {d}}}}}, {{\"type\": \"file\", \"oid\": \"{s}\", \"path\": \"flaky.json\", \"size\": {d}}}, {{\"type\": \"file\", \"oid\": \"3333333333333333333333333333333333333333\", \"path\": \"tensors/model-00001-of-00001.safetensors\", \"size\": {d}, \"lfs\": {{\"oid\": \"{s}\", \"size\": {d}}}}}]", .{ &fake.config_git_hex, fake.config.len, fake.weights.len, fake.weights_sha_hex, fake.weights.len, &fake.flaky_git_hex, flaky_body.len, fake.nested.len, fake.nested_sha_hex, fake.nested.len }) catch return false;
defer std.testing.allocator.free(tree);
respond(fd, tree);
fake.respond(conn, tree);
} else if (std.mem.eql(u8, path, config_path) or std.mem.eql(u8, path, "/Org/Whole/resolve/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb2/config.json")) {
respond(fd, fake.config);
fake.respond(conn, fake.config);
} else if (std.mem.endsWith(u8, path, "/flaky.json")) {
if (fake.flaky_requests.fetchAdd(1, .monotonic) > 0) {
respond(fd, flaky_body);
fake.respond(conn, flaky_body);
return true;
}
var head_buf: [128]u8 = undefined; // the first answer stops halfway and drops the connection
sendAll(fd, std.fmt.bufPrint(&head_buf, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {d}\r\n\r\n", .{flaky_body.len}) catch return false);
sendAll(fd, flaky_body[0 .. flaky_body.len / 2]);
fake.sendAll(conn, std.fmt.bufPrint(&head_buf, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {d}\r\n\r\n", .{flaky_body.len}) catch return false);
fake.sendAll(conn, flaky_body[0 .. flaky_body.len / 2]);
return false;
} else if (std.mem.eql(u8, path, "/Org/Whole/resolve/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb2/weights.safetensors")) {
_ = fake.weight_requests.fetchAdd(1, .monotonic);
whole(fd, fake.weights);
fake.whole(conn, fake.weights);
return false;
} else if (std.mem.eql(u8, path, nested_path) or std.mem.eql(u8, path, "/Org/Whole/resolve/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb2/tensors/model-00001-of-00001.safetensors")) {
whole(fd, fake.nested);
fake.whole(conn, fake.nested);
return false;
} else if (std.mem.eql(u8, path, weights_path)) {
_ = fake.weight_requests.fetchAdd(1, .monotonic);
const from = range_start orelse {
whole(fd, fake.weights);
fake.whole(conn, fake.weights);
return true;
};
fake.range_starts.store(from, .monotonic);
if (from >= fake.weights.len) {
respond(fd, "");
fake.respond(conn, "");
return true;
}
const stop = if (range_end) |last| @min(last + 1, fake.weights.len) else fake.weights.len;
const body = fake.weights[from..stop];
var head_buf: [256]u8 = undefined;
const head_text = std.fmt.bufPrint(&head_buf, "HTTP/1.1 206 Partial Content\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nConnection: keep-alive\r\n\r\n", .{ body.len, from, stop - 1, fake.weights.len }) catch return false;
sendAll(fd, head_text);
sendAll(fd, body);
fake.sendAll(conn, head_text);
fake.sendAll(conn, body);
} else {
respond(fd, "{}");
fake.respond(conn, "{}");
}
return true;
}

fn whole(fd: std.posix.socket_t, body: []const u8) void {
fn whole(fake: *FakeHub, conn: std.Io.net.Stream, body: []const u8) void {
var head_buf: [128]u8 = undefined;
const head_text = std.fmt.bufPrint(&head_buf, "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nConnection: keep-alive\r\n\r\n", .{body.len}) catch return;
sendAll(fd, head_text);
sendAll(fd, body);
fake.sendAll(conn, head_text);
fake.sendAll(conn, body);
}

fn respond(fd: std.posix.socket_t, body: []const u8) void {
fn respond(fake: *FakeHub, conn: std.Io.net.Stream, body: []const u8) void {
var head_buf: [128]u8 = undefined;
const head_text = std.fmt.bufPrint(&head_buf, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {d}\r\nConnection: keep-alive\r\n\r\n", .{body.len}) catch return;
sendAll(fd, head_text);
sendAll(fd, body);
fake.sendAll(conn, head_text);
fake.sendAll(conn, body);
}

fn sendAll(fd: std.posix.socket_t, bytes: []const u8) void {
var sent: usize = 0;
while (sent < bytes.len) {
const n = std.posix.system.write(fd, bytes[sent..].ptr, bytes.len - sent);
if (std.posix.errno(n) != .SUCCESS) return;
sent += @intCast(n);
}
fn sendAll(fake: *FakeHub, conn: std.Io.net.Stream, bytes: []const u8) void {
var buf: [4096]u8 = undefined;
var w = conn.writer(fake.io, &buf);
w.interface.writeAll(bytes) catch return;
w.interface.flush() catch return;
}

fn start(fake: *FakeHub) void {
Expand All @@ -247,10 +249,29 @@ const FakeHub = struct {
return std.fmt.allocPrint(a, "http://127.0.0.1:{d}", .{fake.port});
}

fn forget(fake: *FakeHub, conn: std.Io.net.Stream) void {
fake.live_lock.lockUncancelable(fake.io);
defer fake.live_lock.unlock(fake.io);
for (fake.live.items, 0..) |c, i| {
if (c.socket.handle == conn.socket.handle) {
_ = fake.live.swapRemove(i);
return;
}
}
}

fn stopServer(fake: *FakeHub) void {
fake.stop.store(true, .release);
fake.live_lock.lockUncancelable(fake.io);
for (fake.live.items) |conn| conn.shutdown(fake.io, .both) catch {};
fake.live_lock.unlock(fake.io);
// one connection of our own unblocks accept, and is closed on arrival
if (fake.server.socket.address.connect(fake.io, .{ .mode = .stream }) catch null) |wake| wake.close(fake.io);
if (fake.thread) |t| t.join();
_ = std.posix.system.close(fake.server_fd);
fake.server.deinit(fake.io);
fake.live.deinit(std.testing.allocator);
fake.threaded.deinit();
std.testing.allocator.destroy(fake.threaded);
std.testing.allocator.free(fake.weights_sha_hex);
std.testing.allocator.free(fake.nested_sha_hex);
}
Expand Down