tricu

io_driver.zig (69578B)

---

 const std = @import("std");
 const Arena = @import("arena.zig").Arena;
 const codecs = @import("codecs.zig");
 const source_text = @import("source_text.zig");
 const indexed = @import("indexed.zig");
 const reduce = @import("reduce.zig");
 const tree = @import("tree.zig");
 
 const c = @cImport({
     // Import declarations, not glibc's fortified C inline wrappers. With the
     // pinned Zig/glibc, optimized translate-c diagnoses the variadic open/openat
     // wrapper bodies even though this driver calls uv_fs_open, not libc open.
     // This is local to header translation: libuv and C callers retain their
     // normal build hardening, and Zig ReleaseSafe retains runtime checks.
     @cUndef("_FORTIFY_SOURCE");
     @cDefine("_FORTIFY_SOURCE", "0");
     @cInclude("uv.h");
 });
 
 // ---------------------------------------------------------------------------
 // Action tag constants (must match lib/io.tri and IODriver.hs)
 // ---------------------------------------------------------------------------
 
 pub const ActionTag = enum(u8) {
     pure = 0,
     bind = 1,
     putStr = 10,
     getLine = 11,
     readFile = 20,
     readSourceFile = 37,
     encodeSourceText = 38,
     readIndexedTree = 50,
     hashIndexedTree = 51,
     writeIndexedTreeIfAbsent = 52,
     writeFile = 21,
     writeBytes = 22,
     renameFile = 24,
     createDirectory = 25,
     fileExists = 27,
     sha256Hex = 28,
     writeBytesIfAbsent = 32,
     atomicWriteBytes = 33,
     ask = 30,
     local = 31,
     get = 40,
     put = 41,
     fork = 60,
     await = 61,
     yield = 62,
     sleep = 63,
 };
 
 const FileWrite = struct { path: u32, contents: u32 };
 
 pub const Action = union(ActionTag) {
     pure: u32,
     bind: struct { left: u32, k: u32 },
     putStr: u32,
     getLine,
     readFile: u32,
     readSourceFile: u32,
     encodeSourceText: u32,
     readIndexedTree: struct { path: u32, digest: u32 },
     hashIndexedTree: u32,
     writeIndexedTreeIfAbsent: struct { path: u32, digest: u32, tree: u32 },
     writeFile: FileWrite,
     writeBytes: FileWrite,
     renameFile: struct { old: u32, new: u32 },
     createDirectory: u32,
     fileExists: u32,
     sha256Hex: u32,
     writeBytesIfAbsent: FileWrite,
     atomicWriteBytes: FileWrite,
     ask,
     local: struct { f: u32, action: u32 },
     get,
     put: u32,
     fork: u32,
     await: u32,
     yield,
     sleep: u32,
 };
 
 // ---------------------------------------------------------------------------
 // Error codes (must match IODriver.hs)
 // ---------------------------------------------------------------------------
 
 const ERR_DOES_NOT_EXIST: u64 = 1;
 const ERR_PERMISSION: u64 = 2;
 const ERR_ALREADY_EXISTS: u64 = 3;
 const ERR_IO_OTHER: u64 = 4;
 const ERR_POLICY_DENY: u64 = 20;
 const ERR_INVALID_ACTION: u64 = 40;
 const ERR_INVALID_STRING: u64 = 41;
 
 // ---------------------------------------------------------------------------
 // Permissions
 // ---------------------------------------------------------------------------
 
 pub const IOPerms = struct {
     allow_read_all: bool = false,
     allow_write_all: bool = false,
 };
 
 // ---------------------------------------------------------------------------
 // IO sentinel detection
 // ---------------------------------------------------------------------------
 
 pub fn isIOSentinel(arena: *Arena, root: u32) !?u32 {
     const node = arena.get(try reduce.reduce(root, arena, 1_000_000)).*;
     if (node != .fork) return null;
 
     const sentinel = node.fork.left;
     const rest = node.fork.right;
 
     const sentinel_str = try codecs.toString(arena, sentinel);
     defer {
         if (sentinel_str) |s| {
             arena.allocator.free(s);
         }
     }
     if (sentinel_str == null) return null;
     if (!std.mem.eql(u8, sentinel_str.?, "tricuIO")) return null;
 
     const rest_node = arena.get(try reduce.reduce(rest, arena, 1_000_000)).*;
     if (rest_node != .fork) return null;
 
     const version_num = try codecs.toNumber(arena, rest_node.fork.left);
     if (version_num == null or version_num.? != 1) return null;
 
     return rest_node.fork.right;
 }
 
 // ---------------------------------------------------------------------------
 // Action decoding
 // ---------------------------------------------------------------------------
 
 pub fn decodeAction(arena: *Arena, root: u32) !?Action {
     const node = arena.get(root).*;
     if (node != .fork) return null;
 
     const tag_num = try codecs.toNumber(arena, node.fork.left);
     if (tag_num == null) return null;
 
     const tag: ActionTag = switch (tag_num.?) {
         0 => .pure,
         1 => .bind,
         10 => .putStr,
         11 => .getLine,
         20 => .readFile,
         37 => .readSourceFile,
         38 => .encodeSourceText,
         50 => .readIndexedTree,
         51 => .hashIndexedTree,
         52 => .writeIndexedTreeIfAbsent,
         21 => .writeFile,
         22 => .writeBytes,
         24 => .renameFile,
         25 => .createDirectory,
         27 => .fileExists,
         28 => .sha256Hex,
         32 => .writeBytesIfAbsent,
         33 => .atomicWriteBytes,
         30 => .ask,
         31 => .local,
         40 => .get,
         41 => .put,
         60 => .fork,
         61 => .await,
         62 => .yield,
         63 => .sleep,
         else => return null,
     };
 
     const payload = node.fork.right;
 
     return switch (tag) {
         .pure => Action{ .pure = payload },
         .bind => blk: {
             const payload_node = arena.get(try reduce.reduce(payload, arena, 1_000_000));
             if (payload_node.* != .fork) return null;
             break :blk Action{ .bind = .{ .left = payload_node.fork.left, .k = payload_node.fork.right } };
         },
         .putStr => Action{ .putStr = payload },
         .getLine => Action.getLine,
         .readFile => Action{ .readFile = payload },
         .readSourceFile => Action{ .readSourceFile = payload },
         .encodeSourceText => Action{ .encodeSourceText = payload },
         .hashIndexedTree => Action{ .hashIndexedTree = payload },
         .readIndexedTree, .writeIndexedTreeIfAbsent => blk: {
             const pair = arena.get(try reduce.reduce(payload, arena, 1_000_000)).*;
             if (pair != .fork) return null;
             if (tag == .readIndexedTree) break :blk .{ .readIndexedTree = .{ .path = pair.fork.left, .digest = pair.fork.right } };
             const rest = arena.get(try reduce.reduce(pair.fork.right, arena, 1_000_000)).*;
             if (rest != .fork) return null;
             break :blk .{ .writeIndexedTreeIfAbsent = .{ .path = pair.fork.left, .digest = rest.fork.left, .tree = rest.fork.right } };
         },
         .writeFile, .writeBytes, .writeBytesIfAbsent, .atomicWriteBytes, .renameFile => blk: {
             const pair = arena.get(try reduce.reduce(payload, arena, 1_000_000));
             if (pair.* != .fork) return null;
             break :blk switch (tag) {
                 .writeFile => Action{ .writeFile = .{ .path = pair.fork.left, .contents = pair.fork.right } },
                 .writeBytes => Action{ .writeBytes = .{ .path = pair.fork.left, .contents = pair.fork.right } },
                 .writeBytesIfAbsent => Action{ .writeBytesIfAbsent = .{ .path = pair.fork.left, .contents = pair.fork.right } },
                 .atomicWriteBytes => Action{ .atomicWriteBytes = .{ .path = pair.fork.left, .contents = pair.fork.right } },
                 .renameFile => Action{ .renameFile = .{ .old = pair.fork.left, .new = pair.fork.right } },
                 else => unreachable,
             };
         },
         .createDirectory => Action{ .createDirectory = payload },
         .fileExists => Action{ .fileExists = payload },
         .sha256Hex => Action{ .sha256Hex = payload },
         .ask => Action.ask,
         .local => blk: {
             const payload_node = arena.get(try reduce.reduce(payload, arena, 1_000_000));
             if (payload_node.* != .fork) return null;
             break :blk Action{ .local = .{ .f = payload_node.fork.left, .action = payload_node.fork.right } };
         },
         .get => Action.get,
         .put => Action{ .put = payload },
         .fork => Action{ .fork = payload },
         .await => Action{ .await = payload },
         .yield => Action.yield,
         .sleep => Action{ .sleep = payload },
     };
 }
 
 // ---------------------------------------------------------------------------
 // Response tree constructors
 // ---------------------------------------------------------------------------
 
 pub fn makePure(arena: *Arena, val: u32) !u32 {
     const tag = try codecs.ofNumber(arena, 0);
     return try arena.alloc(.{ .fork = .{ .left = tag, .right = val } });
 }
 
 pub fn makeOkResult(arena: *Arena, val: u32) !u32 {
     const ok_tag = try arena.alloc(.{ .stem = .{ .child = try arena.alloc(.leaf) } });
     const val_pair = try arena.alloc(.{ .fork = .{ .left = val, .right = try arena.alloc(.leaf) } });
     return try arena.alloc(.{ .fork = .{ .left = ok_tag, .right = val_pair } });
 }
 
 pub fn makeErrString(arena: *Arena, message: []const u8) !u32 {
     return makeErrValue(arena, try codecs.ofString(arena, message));
 }
 
 fn makeErrValue(arena: *Arena, text: u32) !u32 {
     const value = try arena.alloc(.{ .fork = .{ .left = text, .right = try arena.alloc(.leaf) } });
     return try arena.alloc(.{ .fork = .{ .left = try arena.alloc(.leaf), .right = value } });
 }
 
 fn sourceReadError(arena: *Arena, path: []const u8, reason: []const u8) !u32 {
     const message = try std.fmt.allocPrint(arena.allocator, "{s}: {s}", .{ path, reason });
     defer arena.allocator.free(message);
     return makeErrValue(arena, try source_text.fromUtf8(arena, message));
 }
 
 fn sourceReadResult(arena: *Arena, path: []const u8, bytes: []const u8) !u32 {
     const scalars = source_text.fromUtf8(arena, bytes) catch |err| switch (err) {
         error.InvalidUtf8 => return sourceReadError(arena, path, "invalid UTF-8 source"),
         else => return err,
     };
     if (std.mem.startsWith(u8, bytes, "\xef\xbb\xbf"))
         return sourceReadError(arena, path, "UTF-8 source BOM is not permitted");
     const raw = try codecs.ofBytes(arena, bytes);
     const pair = try arena.alloc(.{ .fork = .{ .left = raw, .right = scalars } });
     return makeOkResult(arena, pair);
 }
 
 pub fn makeErrResult(arena: *Arena, code: u64) !u32 {
     // The portable IO protocol carries text errors, matching the Haskell
     // driver; numeric values remain internal filesystem status codes.
     return makeErrString(arena, fsErrorMessage(code));
 }
 
 // A null callback makes libuv filesystem operations synchronous. Each
 // request is cleaned up; temp files live in the destination directory so
 // both link (no replace) and rename (replace) are atomic on that filesystem.
 const FsWriteResult = union(enum) { written: bool, failure: u64 };
 
 const FsReadResult = union(enum) { bytes: []u8, failure: u64 };
 
 fn readFs(loop: *c.uv_loop_t, gpa: std.mem.Allocator, path: [:0]const u8) !FsReadResult {
     var req: c.uv_fs_t = undefined;
     const fd = c.uv_fs_open(loop, &req, path.ptr, c.O_RDONLY, 0, null);
     c.uv_fs_req_cleanup(&req);
     if (fd < 0) return .{ .failure = mapUvErr(fd) };
     defer {
         _ = c.uv_fs_close(loop, &req, fd, null);
         c.uv_fs_req_cleanup(&req);
     }
     var bytes: std.ArrayList(u8) = .empty;
     defer bytes.deinit(gpa);
     var buffer: [4096]u8 = undefined;
     var uv_buffer = c.uv_buf_init(&buffer, buffer.len);
     while (true) {
         const count = c.uv_fs_read(loop, &req, fd, &uv_buffer, 1, -1, null);
         c.uv_fs_req_cleanup(&req);
         if (count < 0) return .{ .failure = mapUvErr(count) };
         if (count == 0) return .{ .bytes = try bytes.toOwnedSlice(gpa) };
         try bytes.appendSlice(gpa, buffer[0..@intCast(count)]);
     }
 }
 
 fn uvFileError(rc: c_int) u64 {
     return mapUvErr(rc);
 }
 
 fn syncFsPath(loop: *c.uv_loop_t, path: [*:0]const u8, directory: bool) ?u64 {
     var open_req: c.uv_fs_t = undefined;
     const fd = c.uv_fs_open(loop, &open_req, path, c.O_RDONLY | (if (directory) @as(c_int, c.O_DIRECTORY) else 0), 0, null);
     c.uv_fs_req_cleanup(&open_req);
     if (fd < 0) return uvFileError(fd);
     var sync_req: c.uv_fs_t = undefined;
     const synced = c.uv_fs_fsync(loop, &sync_req, @intCast(fd), null);
     c.uv_fs_req_cleanup(&sync_req);
     var close_req: c.uv_fs_t = undefined;
     const closed = c.uv_fs_close(loop, &close_req, @intCast(fd), null);
     c.uv_fs_req_cleanup(&close_req);
     if (synced < 0) return uvFileError(synced);
     if (closed < 0) return uvFileError(closed);
     return null;
 }
 
 fn syncFsParent(loop: *c.uv_loop_t, gpa: std.mem.Allocator, path: []const u8) !?u64 {
     const parent = std.fs.path.dirname(path) orelse ".";
     const parent_z = try gpa.dupeZ(u8, parent);
     defer gpa.free(parent_z);
     return syncFsPath(loop, parent_z, true);
 }
 
 fn writeAtomicFs(loop: *c.uv_loop_t, gpa: std.mem.Allocator, path: [:0]const u8, contents: []const u8, exclusive: bool) !FsWriteResult {
     const template = try std.fmt.allocPrintSentinel(gpa, "{s}.tmp.XXXXXX", .{path}, 0);
     defer gpa.free(template);
     var req: c.uv_fs_t = undefined;
     const fd = c.uv_fs_mkstemp(loop, &req, template.ptr, null);
     if (fd < 0) {
         c.uv_fs_req_cleanup(&req);
         return .{ .failure = uvFileError(fd) };
     }
     const tmp = gpa.dupeZ(u8, std.mem.span(req.path)) catch |err| {
         var close_req: c.uv_fs_t = undefined;
         _ = c.uv_fs_close(loop, &close_req, fd, null);
         c.uv_fs_req_cleanup(&close_req);
         var unlink_req: c.uv_fs_t = undefined;
         _ = c.uv_fs_unlink(loop, &unlink_req, req.path, null);
         c.uv_fs_req_cleanup(&unlink_req);
         c.uv_fs_req_cleanup(&req);
         return err;
     };
     c.uv_fs_req_cleanup(&req);
     defer gpa.free(tmp);
     defer {
         var unlink_req: c.uv_fs_t = undefined;
         _ = c.uv_fs_unlink(loop, &unlink_req, tmp.ptr, null);
         c.uv_fs_req_cleanup(&unlink_req);
     }
     var offset: usize = 0;
     var failure: ?u64 = null;
     while (offset < contents.len) {
         const remain = contents[offset..];
         var buf = c.uv_buf_init(@ptrCast(@constCast(remain.ptr)), @intCast(@min(remain.len, std.math.maxInt(c_uint))));
         var write_req: c.uv_fs_t = undefined;
         const n = c.uv_fs_write(loop, &write_req, fd, &buf, 1, @intCast(offset), null);
         c.uv_fs_req_cleanup(&write_req);
         if (n <= 0) {
             failure = if (n == 0) ERR_IO_OTHER else uvFileError(n);
             break;
         }
         offset += @intCast(n);
     }
     var close_req: c.uv_fs_t = undefined;
     const closed = c.uv_fs_close(loop, &close_req, fd, null);
     c.uv_fs_req_cleanup(&close_req);
     if (failure) |err| return .{ .failure = err };
     if (closed < 0) return .{ .failure = uvFileError(closed) };
     if (syncFsPath(loop, tmp.ptr, false)) |err| return .{ .failure = err };
     var publish_req: c.uv_fs_t = undefined;
     const published = if (exclusive)
         c.uv_fs_link(loop, &publish_req, tmp.ptr, path.ptr, null)
     else
         c.uv_fs_rename(loop, &publish_req, tmp.ptr, path.ptr, null);
     c.uv_fs_req_cleanup(&publish_req);
     if (published == c.UV_EEXIST and exclusive) {
         if (syncFsPath(loop, path.ptr, false)) |err| return .{ .failure = err };
         if (try syncFsParent(loop, gpa, path)) |err| return .{ .failure = err };
         return .{ .written = false };
     }
     if (published < 0) return .{ .failure = uvFileError(published) };
     if (try syncFsParent(loop, gpa, path)) |err| return .{ .failure = err };
     return .{ .written = true };
 }
 
 fn fsErrorMessage(err: u64) []const u8 {
     return switch (err) {
         ERR_DOES_NOT_EXIST => "does not exist",
         ERR_PERMISSION, ERR_POLICY_DENY => "permission denied",
         ERR_ALREADY_EXISTS => "already exists",
         ERR_INVALID_STRING => "invalid string",
         ERR_INVALID_ACTION => "invalid action",
         else => "io error",
     };
 }
 
 // ---------------------------------------------------------------------------
 // Frame stack and runtime
 // ---------------------------------------------------------------------------
 
 const Frame = union(enum) {
     bind: u32, // continuation k
     local: u32, // old env
 };
 
 const Runtime = struct {
     env: u32,
     state: u32,
 };
 
 // ---------------------------------------------------------------------------
 // Task
 // ---------------------------------------------------------------------------
 
 const Task = struct {
     id: u64,
     parent: ?*Task,
     frames: std.ArrayList(Frame),
     runtime: Runtime,
     current: u32,
     status: enum { runnable, blocked, completed },
     result: ?u32,
     waiting_for: ?u64,
 
     fn init(gpa: std.mem.Allocator, id: u64, parent: ?*Task, env: u32, state: u32, current: u32) !*Task {
         const task = try gpa.create(Task);
         task.* = .{
             .id = id,
             .parent = parent,
             .frames = std.ArrayList(Frame).empty,
             .runtime = .{ .env = env, .state = state },
             .current = current,
             .status = .runnable,
             .result = null,
             .waiting_for = null,
         };
         return task;
     }
 
     fn deinit(self: *Task, gpa: std.mem.Allocator) void {
         self.frames.deinit(gpa);
         gpa.destroy(self);
     }
 
     // finishValue processes a value through the frame stack.
     // Returns true if the task has completed (no more frames).
     fn finishValue(self: *Task, arena: *Arena, value: u32) !bool {
         if (self.frames.pop()) |frame| {
             switch (frame) {
                 .bind => |k| {
                     self.current = try arena.alloc(.{ .app = .{ .func = k, .arg = value } });
                     return false;
                 },
                 .local => |old_env| {
                     self.runtime.env = old_env;
                     self.current = try makePure(arena, value);
                     return false;
                 },
             }
         } else {
             self.current = value;
             return true;
         }
     }
 };
 
 // ---------------------------------------------------------------------------
 // Scheduler
 // ---------------------------------------------------------------------------
 
 const Scheduler = struct {
     gpa: std.mem.Allocator,
     loop: *c.uv_loop_t,
     arena: *Arena,
     tasks: std.ArrayList(*Task),
     runnable: std.ArrayList(*Task),
     next_id: u64,
     perms: IOPerms,
 
     fn init(gpa: std.mem.Allocator, loop: *c.uv_loop_t, arena: *Arena, perms: IOPerms) !Scheduler {
         const sched = Scheduler{
             .gpa = gpa,
             .loop = loop,
             .arena = arena,
             .tasks = std.ArrayList(*Task).empty,
             .runnable = std.ArrayList(*Task).empty,
             .next_id = 1,
             .perms = perms,
         };
         return sched;
     }
 
     fn deinit(self: *Scheduler) void {
         for (self.tasks.items) |task| {
             task.deinit(self.gpa);
         }
         self.tasks.deinit(self.gpa);
         self.runnable.deinit(self.gpa);
     }
 
     fn createTask(self: *Scheduler, parent: ?*Task, env: u32, state: u32, current: u32) !*Task {
         const id = self.next_id;
         self.next_id += 1;
         const task = try Task.init(self.gpa, id, parent, env, state, current);
         errdefer task.deinit(self.gpa);
         try self.tasks.append(self.gpa, task);
         return task;
     }
 
     fn run(self: *Scheduler) !void {
         while (true) {
             if (self.runnable.items.len > 0) {
                 const task = self.runnable.orderedRemove(0);
                 try self.stepTask(task);
             } else if (self.hasPendingHandles()) {
                 _ = c.uv_run(self.loop, c.UV_RUN_ONCE);
             } else {
                 break;
             }
         }
     }
 
     fn hasPendingHandles(self: *Scheduler) bool {
         return c.uv_loop_alive(self.loop) != 0;
     }
 
     fn completeTask(self: *Scheduler, task: *Task) !void {
         task.status = .completed;
         task.result = task.current;
         // Unblock any tasks waiting for this one
         for (self.tasks.items) |t| {
             if (t.status == .blocked and t.waiting_for == task.id) {
                 t.status = .runnable;
                 t.waiting_for = null;
                 t.current = try makePure(self.arena, task.result.?);
                 try self.runnable.append(self.gpa, t);
             }
         }
     }
 
     fn completeIO(self: *Scheduler, task: *Task, result: u32) !void {
         if (try task.finishValue(self.arena, result)) {
             try self.completeTask(task);
         } else {
             try self.runnable.append(self.gpa, task);
         }
     }
 
     fn fsReply(self: *Scheduler, task: *Task, err: u64) !void {
         try self.completeIO(task, try makeErrString(self.arena, fsErrorMessage(err)));
     }
 
     // Like the native CAS operations, source reads use synchronous libuv calls.
     // This avoids entering callback/reducer code while local trees are unrooted;
     // allocation and filesystem failures propagate normally to the host.
     // A source read occupies one scheduler turn (it is not a streaming action).
     fn readSource(self: *Scheduler, task: *Task, path_tree: u32) !void {
         const path = source_text.toUtf8(self.arena, path_tree) catch |err| switch (err) {
             error.InvalidScalarList => return self.fsReply(task, ERR_INVALID_STRING),
             else => return err,
         };
         defer self.arena.allocator.free(path);
         if (!self.perms.allow_read_all) return self.fsReply(task, ERR_POLICY_DENY);
         if (std.mem.indexOfScalar(u8, path, 0) != null) return self.fsReply(task, ERR_INVALID_STRING);
         const path_z = try self.gpa.dupeZ(u8, path);
         defer self.gpa.free(path_z);
         const read = try readFs(self.loop, self.gpa, path_z);
         switch (read) {
             .failure => |code| try self.fsReply(task, code),
             .bytes => |bytes| {
                 defer self.gpa.free(bytes);
                 try self.completeIO(task, try sourceReadResult(self.arena, path, bytes));
             },
         }
     }
 
     // IO protocol adapters: codec and filesystem helpers know nothing about
     // compiler manifests, cache aliases, path selection or publication order.
     fn indexedIO(self: *Scheduler, task: *Task, path_tree: u32, expected_tree: u32, value: ?u32) !void {
         const path = source_text.toUtf8(self.arena, path_tree) catch |err| switch (err) {
             error.InvalidScalarList => return self.fsReply(task, ERR_INVALID_STRING),
             else => return err,
         };
         defer self.arena.allocator.free(path);
         if (std.mem.indexOfScalar(u8, path, 0) != null) return self.fsReply(task, ERR_INVALID_STRING);
         const expected = source_text.toUtf8(self.arena, expected_tree) catch |err| switch (err) {
             error.InvalidScalarList => return self.completeIO(task, try makeErrString(self.arena, "invalid CAS digest")),
             else => return err,
         };
         defer self.arena.allocator.free(expected);
         if (expected.len != 64) return self.completeIO(task, try makeErrString(self.arena, "invalid CAS digest"));
         for (expected) |ch| if (!(ch >= '0' and ch <= '9' or ch >= 'a' and ch <= 'f'))
             return self.completeIO(task, try makeErrString(self.arena, "invalid CAS digest"));
         if (!self.perms.allow_read_all or (value != null and !self.perms.allow_write_all)) return self.fsReply(task, ERR_POLICY_DENY);
         const path_z = try self.gpa.dupeZ(u8, path);
         defer self.gpa.free(path_z);
         var encoded: ?[]u8 = null;
         defer if (encoded) |bytes| self.arena.allocator.free(bytes);
         var fresh = false;
         if (value) |tree_value| {
             encoded = try indexed.encode(self.arena, tree_value);
             if (!std.mem.eql(u8, &indexed.digest(encoded.?), expected))
                 return self.completeIO(task, try makeErrString(self.arena, "CAS digest mismatch"));
             switch (try writeAtomicFs(self.loop, self.gpa, path_z, encoded.?, true)) {
                 .failure => |code| return self.fsReply(task, code),
                 .written => |written| fresh = written,
             }
         }
         // Authenticate before decode; for publication compare the actual winner
         // byte-for-byte, including after an EEXIST race. Never overwrite it.
         switch (try readFs(self.loop, self.gpa, path_z)) {
             .failure => |code| return self.fsReply(task, code),
             .bytes => |bytes| {
                 defer self.gpa.free(bytes);
                 const result = if (encoded) |ours| blk: {
                     if (!std.mem.eql(u8, bytes, ours)) return self.completeIO(task, try makeErrString(self.arena, "CAS object exists with different bytes"));
                     break :blk if (fresh) try self.arena.alloc(.{ .stem = .{ .child = try self.arena.alloc(.leaf) } }) else try self.arena.alloc(.leaf);
                 } else blk: {
                     if (!std.mem.eql(u8, &indexed.digest(bytes), expected)) return self.completeIO(task, try makeErrString(self.arena, "CAS digest mismatch"));
                     break :blk indexed.load(self.arena, bytes, "root", true) catch |err| switch (err) {
                         error.OutOfMemory => return err,
                         else => return self.completeIO(task, try makeErrString(self.arena, "invalid indexed tree")),
                     };
                 };
                 try self.completeIO(task, try makeOkResult(self.arena, result));
             },
         }
     }
 
     fn stepTask(self: *Scheduler, task: *Task) !void {
         // Snapshot every scheduler-owned node for this turn. A turn may force
         // repeatedly; frames popped during it remain rooted until it completes.
         // Callbacks only allocate/link trees and never enter the reducer.
         const base = self.arena.roots.items.len;
         defer self.arena.roots.shrinkRetainingCapacity(base);
         for (self.tasks.items) |t| {
             try self.arena.roots.appendSlice(self.gpa, &.{ t.current, t.runtime.env, t.runtime.state });
             if (t.result) |result| try self.arena.roots.append(self.gpa, result);
             for (t.frames.items) |frame| switch (frame) {
                 .bind => |k| try self.arena.roots.append(self.gpa, k),
                 .local => |env| try self.arena.roots.append(self.gpa, env),
             };
         }
         const reduced = try reduce.reduce(task.current, self.arena, std.math.maxInt(u64));
 
         const decoded = try decodeAction(self.arena, reduced);
         if (decoded == null) {
             // Not a recognized action — if no frames, it's the final result.
             // Otherwise treat as invalid.
             if (task.frames.items.len == 0) {
                 task.current = reduced;
                 try self.completeTask(task);
                 return;
             }
             const invalid = self.arena.get(reduced).*;
             const tag = if (invalid == .fork) try codecs.toNumber(self.arena, invalid.fork.left) else null;
             var diagnostic: [96]u8 = undefined;
             const message = if (tag) |n| try std.fmt.bufPrint(&diagnostic, "invalid or unsupported IO action tag {d}", .{n}) else "invalid action";
             const err = try makeErrString(self.arena, message);
             if (try task.finishValue(self.arena, err)) {
                 try self.completeTask(task);
             } else {
                 try self.runnable.append(self.gpa, task);
             }
             return;
         }
 
         const action = decoded.?;
         switch (action) {
             .pure => |val| {
                 if (try task.finishValue(self.arena, val)) {
                     try self.completeTask(task);
                 } else {
                     try self.runnable.append(self.gpa, task);
                 }
             },
 
             .bind => |b| {
                 try task.frames.append(self.gpa, .{ .bind = b.k });
                 task.current = b.left;
                 try self.runnable.append(self.gpa, task);
             },
 
             .putStr => |str_tree| {
                 const str = try codecs.toString(self.arena, str_tree) orelse {
                     const err = try makeErrResult(self.arena, ERR_INVALID_STRING);
                     if (try task.finishValue(self.arena, err)) {
                         try self.completeTask(task);
                     } else {
                         try self.runnable.append(self.gpa, task);
                     }
                     return;
                 };
                 defer self.gpa.free(str);
                 var written: usize = 0;
                 while (written < str.len) {
                     const n = std.c.write(1, str[written..].ptr, str.len - written);
                     if (n <= 0) return error.WriteFailed;
                     written += @intCast(n);
                 }
                 const leaf = try self.arena.alloc(.leaf);
                 if (try task.finishValue(self.arena, leaf)) {
                     try self.completeTask(task);
                 } else {
                     try self.runnable.append(self.gpa, task);
                 }
             },
 
             .getLine => {
                 var buf: [4096]u8 = undefined;
                 var len: usize = 0;
                 while (len < buf.len) {
                     const n = std.c.read(0, buf[len..].ptr, 1);
                     if (n <= 0) break;
                     if (buf[len] == '\n') break;
                     len += 1;
                 }
                 const line = buf[0..len];
                 const str_tree = try codecs.ofString(self.arena, line);
                 if (try task.finishValue(self.arena, str_tree)) {
                     try self.completeTask(task);
                 } else {
                     try self.runnable.append(self.gpa, task);
                 }
             },
 
             .readSourceFile => |path_tree| try self.readSource(task, path_tree),
             .readIndexedTree => |read| try self.indexedIO(task, read.path, read.digest, null),
             .writeIndexedTreeIfAbsent => |write| try self.indexedIO(task, write.path, write.digest, write.tree),
             .hashIndexedTree => |value| {
                 const bytes = try indexed.encode(self.arena, value);
                 defer self.arena.allocator.free(bytes);
                 const hex = indexed.digest(bytes);
                 try self.completeIO(task, try makeOkResult(self.arena, try codecs.ofString(self.arena, &hex)));
             },
 
             .encodeSourceText => |text| {
                 const bytes = source_text.toUtf8(self.arena, text) catch |err| switch (err) {
                     error.InvalidScalarList => {
                         try self.completeIO(task, try makeErrString(self.arena, "invalid source scalar list"));
                         return;
                     },
                     else => return err,
                 };
                 defer self.arena.allocator.free(bytes);
                 try self.completeIO(task, try makeOkResult(self.arena, try codecs.ofBytes(self.arena, bytes)));
             },
 
             .readFile => |path_tree| {
                 const path = try codecs.toString(self.arena, path_tree) orelse {
                     const err = try makeErrResult(self.arena, ERR_INVALID_STRING);
                     if (try task.finishValue(self.arena, err)) {
                         try self.completeTask(task);
                     } else {
                         try self.runnable.append(self.gpa, task);
                     }
                     return;
                 };
 
                 if (!self.perms.allow_read_all) {
                     self.arena.allocator.free(path);
                     const err = try makeErrResult(self.arena, ERR_POLICY_DENY);
                     if (try task.finishValue(self.arena, err)) {
                         try self.completeTask(task);
                     } else {
                         try self.runnable.append(self.gpa, task);
                     }
                     return;
                 }
 
                 const ctx = try self.gpa.create(FileReadCtx);
                 ctx.* = .{
                     .scheduler = self,
                     .task = task,
                     .arena = self.arena,
                     .gpa = self.gpa,
                     .fd = -1,
                     .buf = std.ArrayList(u8).empty,
                     .path = try self.gpa.dupeZ(u8, path),
                     .req = undefined,
                     .read_buf = null,
                 };
                 self.arena.allocator.free(path);
                 ctx.req.data = ctx;
                 _ = c.uv_fs_open(self.loop, &ctx.req, ctx.path.ptr, c.O_RDONLY, 0, file_open_cb);
             },
 
             .writeFile => |wf| {
                 const path = try codecs.toString(self.arena, wf.path) orelse {
                     const err = try makeErrResult(self.arena, ERR_INVALID_STRING);
                     if (try task.finishValue(self.arena, err)) {
                         try self.completeTask(task);
                     } else {
                         try self.runnable.append(self.gpa, task);
                     }
                     return;
                 };
 
                 const contents = try codecs.toString(self.arena, wf.contents) orelse {
                     self.arena.allocator.free(path);
                     const err = try makeErrResult(self.arena, ERR_INVALID_STRING);
                     if (try task.finishValue(self.arena, err)) {
                         try self.completeTask(task);
                     } else {
                         try self.runnable.append(self.gpa, task);
                     }
                     return;
                 };
 
                 if (!self.perms.allow_write_all) {
                     self.arena.allocator.free(path);
                     self.arena.allocator.free(contents);
                     const err = try makeErrResult(self.arena, ERR_POLICY_DENY);
                     if (try task.finishValue(self.arena, err)) {
                         try self.completeTask(task);
                     } else {
                         try self.runnable.append(self.gpa, task);
                     }
                     return;
                 }
 
                 const ctx = try self.gpa.create(FileWriteCtx);
                 ctx.* = .{
                     .scheduler = self,
                     .task = task,
                     .arena = self.arena,
                     .gpa = self.gpa,
                     .fd = -1,
                     .path = try self.gpa.dupeZ(u8, path),
                     .contents = contents,
                     .offset = 0,
                     .req = undefined,
                 };
                 self.arena.allocator.free(path);
                 ctx.req.data = ctx;
                 const flags = c.O_WRONLY | c.O_CREAT | c.O_TRUNC;
                 _ = c.uv_fs_open(self.loop, &ctx.req, ctx.path.ptr, flags, 0o644, file_write_open_cb);
             },
 
             .writeBytes, .writeBytesIfAbsent, .atomicWriteBytes => |write| {
                 const path = try codecs.toString(self.arena, write.path) orelse {
                     try self.fsReply(task, ERR_INVALID_STRING);
                     return;
                 };
                 defer self.arena.allocator.free(path);
                 const contents = try codecs.toBytes(self.arena, write.contents) orelse {
                     try self.completeIO(task, try makeErrString(self.arena, "invalid bytes"));
                     return;
                 };
                 defer self.arena.allocator.free(contents);
                 if (!self.perms.allow_write_all) {
                     try self.fsReply(task, ERR_POLICY_DENY);
                     return;
                 }
                 const path_z = try self.gpa.dupeZ(u8, path);
                 defer self.gpa.free(path_z);
                 const exclusive = action == .writeBytesIfAbsent;
                 const published = try writeAtomicFs(self.loop, self.gpa, path_z, contents, exclusive);
                 switch (published) {
                     .failure => |code| try self.fsReply(task, code),
                     .written => |fresh| {
                         const value = if (action == .writeBytes) try self.arena.alloc(.leaf) else if (fresh) try self.arena.alloc(.{ .stem = .{ .child = try self.arena.alloc(.leaf) } }) else try self.arena.alloc(.leaf);
                         try self.completeIO(task, try makeOkResult(self.arena, value));
                     },
                 }
             },
 
             .createDirectory, .fileExists => |path_tree| {
                 const path = try codecs.toString(self.arena, path_tree) orelse {
                     try self.fsReply(task, ERR_INVALID_STRING);
                     return;
                 };
                 defer self.arena.allocator.free(path);
                 const is_create = action == .createDirectory;
                 if (is_create and !self.perms.allow_write_all or
                     !is_create and !self.perms.allow_read_all)
                 {
                     try self.fsReply(task, ERR_POLICY_DENY);
                     return;
                 }
                 const path_z = try self.gpa.dupeZ(u8, path);
                 defer self.gpa.free(path_z);
                 var req: c.uv_fs_t = undefined;
                 const rc = if (is_create) c.uv_fs_mkdir(self.loop, &req, path_z.ptr, 0o755, null) else c.uv_fs_stat(self.loop, &req, path_z.ptr, null);
                 c.uv_fs_req_cleanup(&req);
                 if (rc < 0 and (is_create and rc != c.UV_EEXIST or
                     !is_create and rc != c.UV_ENOENT))
                 {
                     try self.fsReply(task, uvFileError(rc));
                     return;
                 }
                 if (is_create and rc == c.UV_EEXIST) {
                     var stat_req: c.uv_fs_t = undefined;
                     const stat = c.uv_fs_stat(self.loop, &stat_req, path_z.ptr, null);
                     const is_dir = stat >= 0 and stat_req.statbuf.st_mode & c.S_IFMT == c.S_IFDIR;
                     c.uv_fs_req_cleanup(&stat_req);
                     if (!is_dir) {
                         try self.fsReply(task, ERR_ALREADY_EXISTS);
                         return;
                     }
                 }
                 if (is_create) {
                     if (syncFsPath(self.loop, path_z.ptr, true)) |err| {
                         try self.fsReply(task, err);
                         return;
                     }
                     if (try syncFsParent(self.loop, self.gpa, path_z)) |err| {
                         try self.fsReply(task, err);
                         return;
                     }
                 }
                 const value = if (is_create) try self.arena.alloc(.leaf) else if (rc == 0) try self.arena.alloc(.{ .stem = .{ .child = try self.arena.alloc(.leaf) } }) else try self.arena.alloc(.leaf);
                 try self.completeIO(task, try makeOkResult(self.arena, value));
             },
 
             .renameFile => |paths| {
                 const old = try codecs.toString(self.arena, paths.old) orelse {
                     try self.fsReply(task, ERR_INVALID_STRING);
                     return;
                 };
                 defer self.arena.allocator.free(old);
                 const new = try codecs.toString(self.arena, paths.new) orelse {
                     try self.fsReply(task, ERR_INVALID_STRING);
                     return;
                 };
                 defer self.arena.allocator.free(new);
                 if (!self.perms.allow_write_all) {
                     try self.fsReply(task, ERR_POLICY_DENY);
                     return;
                 }
                 const old_z = try self.gpa.dupeZ(u8, old);
                 defer self.gpa.free(old_z);
                 const new_z = try self.gpa.dupeZ(u8, new);
                 defer self.gpa.free(new_z);
                 var req: c.uv_fs_t = undefined;
                 const rc = c.uv_fs_rename(self.loop, &req, old_z.ptr, new_z.ptr, null);
                 c.uv_fs_req_cleanup(&req);
                 if (rc < 0) return self.fsReply(task, uvFileError(rc));
                 try self.completeIO(task, try makeOkResult(self.arena, try self.arena.alloc(.leaf)));
             },
 
             .sha256Hex => |bytes_tree| {
                 const bytes = try codecs.toBytes(self.arena, bytes_tree) orelse {
                     try self.completeIO(task, try makeErrString(self.arena, "invalid bytes"));
                     return;
                 };
                 defer self.arena.allocator.free(bytes);
                 var digest: [32]u8 = undefined;
                 std.crypto.hash.sha2.Sha256.hash(bytes, &digest, .{});
                 const hex = std.fmt.bytesToHex(digest, .lower);
                 try self.completeIO(task, try makeOkResult(self.arena, try codecs.ofString(self.arena, &hex)));
             },
 
             .ask => {
                 if (try task.finishValue(self.arena, task.runtime.env)) {
                     try self.completeTask(task);
                 } else {
                     try self.runnable.append(self.gpa, task);
                 }
             },
 
             .local => |loc| {
                 const new_env = try reduce.reduce(try self.arena.alloc(.{ .app = .{ .func = loc.f, .arg = task.runtime.env } }), self.arena, std.math.maxInt(u64));
                 try task.frames.append(self.gpa, .{ .local = task.runtime.env });
                 task.runtime.env = new_env;
                 task.current = loc.action;
                 try self.runnable.append(self.gpa, task);
             },
 
             .get => {
                 if (try task.finishValue(self.arena, task.runtime.state)) {
                     try self.completeTask(task);
                 } else {
                     try self.runnable.append(self.gpa, task);
                 }
             },
 
             .put => |new_state| {
                 task.runtime.state = new_state;
                 const leaf = try self.arena.alloc(.leaf);
                 if (try task.finishValue(self.arena, leaf)) {
                     try self.completeTask(task);
                 } else {
                     try self.runnable.append(self.gpa, task);
                 }
             },
 
             .fork => |child_action| {
                 const child = try self.createTask(task, task.runtime.env, task.runtime.state, child_action);
                 try self.runnable.append(self.gpa, child);
                 const handle = try codecs.ofNumber(self.arena, child.id);
                 if (try task.finishValue(self.arena, handle)) {
                     try self.completeTask(task);
                 } else {
                     try self.runnable.append(self.gpa, task);
                 }
             },
 
             .await => |handle_tree| {
                 const handle = try codecs.toNumber(self.arena, handle_tree) orelse {
                     const err = try makeErrResult(self.arena, ERR_INVALID_ACTION);
                     if (try task.finishValue(self.arena, err)) {
                         try self.completeTask(task);
                     } else {
                         try self.runnable.append(self.gpa, task);
                     }
                     return;
                 };
                 var found: ?*Task = null;
                 for (self.tasks.items) |t| {
                     if (t.id == handle) {
                         found = t;
                         break;
                     }
                 }
                 if (found == null) {
                     const err = try makeErrResult(self.arena, ERR_INVALID_ACTION);
                     if (try task.finishValue(self.arena, err)) {
                         try self.completeTask(task);
                     } else {
                         try self.runnable.append(self.gpa, task);
                     }
                     return;
                 }
                 if (found.?.status == .completed) {
                     const result = found.?.result.?;
                     if (try task.finishValue(self.arena, result)) {
                         try self.completeTask(task);
                     } else {
                         try self.runnable.append(self.gpa, task);
                     }
                 } else {
                     task.status = .blocked;
                     task.waiting_for = handle;
                     // Task remains out of runnable until child completes
                 }
             },
 
             .yield => {
                 const leaf = try self.arena.alloc(.leaf);
                 if (try task.finishValue(self.arena, leaf)) {
                     try self.completeTask(task);
                 } else {
                     try self.runnable.append(self.gpa, task);
                 }
             },
 
             .sleep => |ms_tree| {
                 const ms = try codecs.toNumber(self.arena, ms_tree) orelse 0;
                 const ctx = try self.gpa.create(SleepCtx);
                 ctx.* = .{
                     .scheduler = self,
                     .task = task,
                     .arena = self.arena,
                     .timer = undefined,
                 };
                 ctx.timer.data = ctx;
                 _ = c.uv_timer_init(self.loop, &ctx.timer);
                 _ = c.uv_timer_start(&ctx.timer, sleep_cb, @intCast(ms), 0);
             },
         }
     }
 };
 
 // ---------------------------------------------------------------------------
 // Async file read
 // ---------------------------------------------------------------------------
 
 const FileReadCtx = struct {
     scheduler: *Scheduler,
     task: *Task,
     arena: *Arena,
     gpa: std.mem.Allocator,
     fd: c_int,
     buf: std.ArrayList(u8),
     path: [:0]const u8,
     req: c.uv_fs_t,
     read_buf: ?[]u8,
 };
 
 fn mapUvErr(uv_err: c_int) u64 {
     return switch (uv_err) {
         c.UV_ENOENT => ERR_DOES_NOT_EXIST,
         c.UV_EACCES => ERR_PERMISSION,
         c.UV_EEXIST => ERR_ALREADY_EXISTS,
         else => ERR_IO_OTHER,
     };
 }
 
 fn file_open_cb(req: [*c]c.uv_fs_t) callconv(.c) void {
     const ctx = @as(*FileReadCtx, @ptrCast(@alignCast(req.*.data)));
     const result = req.*.result;
     c.uv_fs_req_cleanup(req);
     if (result < 0) {
         const err = makeErrResult(ctx.arena, mapUvErr(@intCast(result))) catch {
             ctx.gpa.destroy(ctx);
             return;
         };
         if (ctx.task.finishValue(ctx.arena, err) catch false) {
             ctx.scheduler.completeTask(ctx.task) catch {};
         } else {
             ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {};
         }
         ctx.buf.deinit(ctx.gpa);
         ctx.gpa.free(ctx.path);
         ctx.gpa.destroy(ctx);
         return;
     }
     ctx.fd = @intCast(result);
     const read_buf = ctx.gpa.alloc(u8, 4096) catch unreachable;
     ctx.read_buf = read_buf;
     var uv_buf = c.uv_buf_init(@ptrCast(read_buf.ptr), @intCast(read_buf.len));
     _ = c.uv_fs_read(ctx.scheduler.loop, req, ctx.fd, &uv_buf, 1, -1, file_read_cb);
 }
 
 fn file_read_cb(req: [*c]c.uv_fs_t) callconv(.c) void {
     const ctx = @as(*FileReadCtx, @ptrCast(@alignCast(req.*.data)));
     const nread = req.*.result;
     c.uv_fs_req_cleanup(req);
     if (nread < 0) {
         _ = c.uv_fs_close(ctx.scheduler.loop, req, ctx.fd, null);
         const err = makeErrResult(ctx.arena, mapUvErr(@intCast(nread))) catch {
             ctx.gpa.destroy(ctx);
             return;
         };
         if (ctx.task.finishValue(ctx.arena, err) catch false) {
             ctx.scheduler.completeTask(ctx.task) catch {};
         } else {
             ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {};
         }
         if (ctx.read_buf) |b| ctx.gpa.free(b);
         ctx.buf.deinit(ctx.gpa);
         ctx.gpa.free(ctx.path);
         ctx.gpa.destroy(ctx);
         return;
     }
     if (nread == 0) {
         // EOF
         _ = c.uv_fs_close(ctx.scheduler.loop, req, ctx.fd, null);
         const bytes_tree = codecs.ofBytes(ctx.arena, ctx.buf.items) catch {
             ctx.gpa.destroy(ctx);
             return;
         };
         const ok = makeOkResult(ctx.arena, bytes_tree) catch {
             ctx.gpa.destroy(ctx);
             return;
         };
         if (ctx.task.finishValue(ctx.arena, ok) catch false) {
             ctx.scheduler.completeTask(ctx.task) catch {};
         } else {
             ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {};
         }
         if (ctx.read_buf) |b| ctx.gpa.free(b);
         ctx.buf.deinit(ctx.gpa);
         ctx.gpa.free(ctx.path);
         ctx.gpa.destroy(ctx);
         return;
     }
     const data = ctx.read_buf.?[0..@intCast(nread)];
     ctx.buf.appendSlice(ctx.gpa, data) catch unreachable;
     ctx.gpa.free(ctx.read_buf.?);
     const read_buf = ctx.gpa.alloc(u8, 4096) catch unreachable;
     ctx.read_buf = read_buf;
     var uv_buf = c.uv_buf_init(@ptrCast(read_buf.ptr), @intCast(read_buf.len));
     _ = c.uv_fs_read(ctx.scheduler.loop, req, ctx.fd, &uv_buf, 1, -1, file_read_cb);
 }
 
 // ---------------------------------------------------------------------------
 // Async file write
 // ---------------------------------------------------------------------------
 
 const FileWriteCtx = struct {
     scheduler: *Scheduler,
     task: *Task,
     arena: *Arena,
     gpa: std.mem.Allocator,
     fd: c_int,
     path: [:0]const u8,
     contents: []const u8,
     offset: usize,
     req: c.uv_fs_t,
 };
 
 fn file_write_open_cb(req: [*c]c.uv_fs_t) callconv(.c) void {
     const ctx = @as(*FileWriteCtx, @ptrCast(@alignCast(req.*.data)));
     const result = req.*.result;
     c.uv_fs_req_cleanup(req);
     if (result < 0) {
         const err = makeErrResult(ctx.arena, mapUvErr(@intCast(result))) catch {
             ctx.gpa.destroy(ctx);
             return;
         };
         if (ctx.task.finishValue(ctx.arena, err) catch false) {
             ctx.scheduler.completeTask(ctx.task) catch {};
         } else {
             ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {};
         }
         ctx.gpa.free(ctx.path);
         ctx.gpa.free(ctx.contents);
         ctx.gpa.destroy(ctx);
         return;
     }
     ctx.fd = @intCast(result);
     if (ctx.contents.len == 0) {
         _ = c.uv_fs_close(ctx.scheduler.loop, req, ctx.fd, file_write_close_cb);
         return;
     }
     file_write_next(ctx, req);
 }
 
 fn file_write_next(ctx: *FileWriteCtx, req: [*c]c.uv_fs_t) void {
     const remaining = ctx.contents[ctx.offset..];
     var uv_buf = c.uv_buf_init(@ptrCast(@constCast(remaining.ptr)), @intCast(@min(remaining.len, std.math.maxInt(c_uint))));
     _ = c.uv_fs_write(ctx.scheduler.loop, req, ctx.fd, &uv_buf, 1, @intCast(ctx.offset), file_write_cb);
 }
 
 fn file_write_cb(req: [*c]c.uv_fs_t) callconv(.c) void {
     const ctx = @as(*FileWriteCtx, @ptrCast(@alignCast(req.*.data)));
     const nwrite = req.*.result;
     c.uv_fs_req_cleanup(req);
     if (nwrite <= 0) {
         _ = c.uv_fs_close(ctx.scheduler.loop, req, ctx.fd, null);
         const err = makeErrResult(ctx.arena, if (nwrite == 0) ERR_IO_OTHER else mapUvErr(@intCast(nwrite))) catch {
             ctx.gpa.destroy(ctx);
             return;
         };
         if (ctx.task.finishValue(ctx.arena, err) catch false) {
             ctx.scheduler.completeTask(ctx.task) catch {};
         } else {
             ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {};
         }
         ctx.gpa.free(ctx.path);
         ctx.gpa.free(ctx.contents);
         ctx.gpa.destroy(ctx);
         return;
     }
     ctx.offset += @intCast(nwrite);
     if (ctx.offset < ctx.contents.len) {
         file_write_next(ctx, req);
         return;
     }
     _ = c.uv_fs_close(ctx.scheduler.loop, req, ctx.fd, file_write_close_cb);
 }
 
 fn file_write_close_cb(req: [*c]c.uv_fs_t) callconv(.c) void {
     const ctx = @as(*FileWriteCtx, @ptrCast(@alignCast(req.*.data)));
     c.uv_fs_req_cleanup(req);
     const leaf = ctx.arena.alloc(.leaf) catch {
         ctx.gpa.destroy(ctx);
         return;
     };
     const ok = makeOkResult(ctx.arena, leaf) catch {
         ctx.gpa.destroy(ctx);
         return;
     };
     if (ctx.task.finishValue(ctx.arena, ok) catch false) {
         ctx.scheduler.completeTask(ctx.task) catch {};
     } else {
         ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {};
     }
     ctx.gpa.free(ctx.path);
     ctx.gpa.free(ctx.contents);
     ctx.gpa.destroy(ctx);
 }
 
 // ---------------------------------------------------------------------------
 // Async sleep
 // ---------------------------------------------------------------------------
 
 const SleepCtx = struct {
     scheduler: *Scheduler,
     task: *Task,
     arena: *Arena,
     timer: c.uv_timer_t,
 };
 
 fn sleep_cb(handle: [*c]c.uv_timer_t) callconv(.c) void {
     const ctx = @as(*SleepCtx, @ptrCast(@alignCast(handle.*.data)));
     defer ctx.scheduler.gpa.destroy(ctx);
     const leaf = ctx.arena.alloc(.leaf) catch {
         ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {};
         return;
     };
     if (ctx.task.finishValue(ctx.arena, leaf) catch false) {
         ctx.scheduler.completeTask(ctx.task) catch {};
     } else {
         ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {};
     }
 }
 
 // ---------------------------------------------------------------------------
 // Public entry point
 // ---------------------------------------------------------------------------
 
 pub fn runIO(gpa: std.mem.Allocator, arena: *Arena, program: u32, perms: IOPerms) !u32 {
     const base = arena.roots.items.len;
     defer arena.roots.shrinkRetainingCapacity(base);
     try arena.roots.append(gpa, program);
     const action_tree = try isIOSentinel(arena, program) orelse {
         return error.InvalidIOSentinel;
     };
 
     var loop: c.uv_loop_t = undefined;
     const rc = c.uv_loop_init(&loop);
     if (rc != 0) return error.LoopInitFailed;
     defer _ = c.uv_loop_close(&loop);
 
     var scheduler = try Scheduler.init(gpa, &loop, arena, perms);
     defer scheduler.deinit();
 
     const main_task = try scheduler.createTask(null, try arena.alloc(.leaf), try arena.alloc(.leaf), action_tree);
     try scheduler.runnable.append(gpa, main_task);
 
     try scheduler.run();
 
     // Return the main task's result
     return main_task.result orelse program;
 }
 
 test "scheduler safepoint roots include blocked tasks, frames, environment, state and results" {
     const testing = std.testing;
     var arena = Arena.init(testing.allocator);
     defer arena.deinit();
     var loop: c.uv_loop_t = undefined;
     try testing.expectEqual(@as(c_int, 0), c.uv_loop_init(&loop));
     defer _ = c.uv_loop_close(&loop);
     var scheduler = try Scheduler.init(testing.allocator, &loop, &arena, .{});
     defer scheduler.deinit();
     const leaf = try arena.alloc(.leaf);
     const external = try codecs.ofNumber(&arena, 99);
     try arena.roots.append(arena.allocator, external);
     const value = try codecs.ofNumber(&arena, 42);
     const action = try makePure(&arena, value);
     const task = try scheduler.createTask(null, leaf, leaf, action);
     const blocked = try scheduler.createTask(null, try codecs.ofNumber(&arena, 17), try codecs.ofNumber(&arena, 23), leaf);
     blocked.status = .blocked;
     blocked.result = try codecs.ofNumber(&arena, 31);
     const saved_env = try codecs.ofNumber(&arena, 47);
     const continuation = try codecs.ofNumber(&arena, 53);
     try blocked.frames.append(arena.allocator, .{ .local = saved_env });
     try blocked.frames.append(arena.allocator, .{ .bind = continuation });
     for (0..1000) |_| _ = try arena.alloc(.leaf);
     const before_collection = arena.len();
     arena.gc_enabled = true;
     try arena.setGcPolicy(1, 1);
     try scheduler.stepTask(task);
     try testing.expect(arena.collections > 0);
     try testing.expect(before_collection - arena.len() + arena.free.items.len >= 1000);
     try testing.expectEqual(@as(usize, 1), arena.roots.items.len);
     // Observations below are outside a scheduler turn; explicitly hold its
     // roots by stopping automatic collection, not by accidentally relying on
     // an allocation threshold.
     arena.gc_enabled = false;
     const ids = [_]u32{ external, task.result.?, blocked.runtime.env, blocked.runtime.state, blocked.result.?, saved_env, continuation };
     const expected = [_]u64{ 99, 42, 17, 23, 31, 47, 53 };
     for (ids, expected) |id, n| try testing.expectEqual(n, (try codecs.toNumber(&arena, id)).?);
 }
 
 test "IO sentinel forces delayed outer and version/action pair" {
     const testing = std.testing;
     var arena = Arena.init(testing.allocator);
     defer arena.deinit();
     const leaf = try arena.alloc(.leaf);
     const value = try codecs.ofNumber(&arena, 42);
     const action = try makePure(&arena, value);
     const version = try codecs.ofNumber(&arena, 1);
     const pair = try arena.alloc(.{ .fork = .{ .left = version, .right = action } });
     const k_pair = try arena.alloc(.{ .fork = .{ .left = leaf, .right = pair } });
     const delayed_pair = try arena.alloc(.{ .app = .{ .func = k_pair, .arg = leaf } });
     const marker = try codecs.ofString(&arena, "tricuIO");
     const sentinel = try arena.alloc(.{ .fork = .{ .left = marker, .right = delayed_pair } });
     const k_sentinel = try arena.alloc(.{ .fork = .{ .left = leaf, .right = sentinel } });
     const program = try arena.alloc(.{ .app = .{ .func = k_sentinel, .arg = leaf } });
     arena.gc_enabled = true;
     try arena.setGcPolicy(1, 1);
     const result = try runIO(arena.allocator, &arena, program, .{});
     arena.gc_enabled = false; // result is held locally, outside a registered scope
     try testing.expectEqual(@as(u64, 42), (try codecs.toNumber(&arena, result)).?);
     try testing.expectEqual(@as(usize, 0), arena.roots.items.len);
 }
 
 fn testAction(arena: *Arena, tag: u64, payload: u32, perms: IOPerms) !codecs.UnwrapResult {
     const action = try arena.alloc(.{ .fork = .{ .left = try codecs.ofNumber(arena, tag), .right = payload } });
     const rest = try arena.alloc(.{ .fork = .{ .left = try codecs.ofNumber(arena, 1), .right = action } });
     const program = try arena.alloc(.{ .fork = .{ .left = try codecs.ofString(arena, "tricuIO"), .right = rest } });
     arena.gc_enabled = true;
     try arena.setGcPolicy(1, 1);
     const result = try runIO(arena.allocator, arena, program, perms);
     arena.gc_enabled = false; // observations are local, outside the scheduler
     return (try codecs.unwrapResult(arena, result)).?;
 }
 
 fn expectErrorText(arena: *Arena, reply: codecs.UnwrapResult, text: []const u8) !void {
     try std.testing.expect(!reply.ok);
     const message = try source_text.toUtf8(arena, reply.value);
     defer arena.allocator.free(message);
     try std.testing.expect(std.mem.indexOf(u8, message, text) != null);
 }
 
 test "encodeSourceText is strict, lazy, permission-free and distinct from byte codecs" {
     var arena = Arena.init(std.testing.allocator);
     defer arena.deinit();
     const expected = "\u{feff}aé😀\x00\r\ne\u{301}";
     const list = try source_text.fromUtf8(&arena, expected);
     const leaf = try arena.alloc(.leaf);
     const constant = try arena.alloc(.{ .fork = .{ .left = leaf, .right = list } });
     const delayed = try arena.alloc(.{ .app = .{ .func = constant, .arg = leaf } });
     const result = try testAction(&arena, 38, delayed, .{});
     try std.testing.expect(result.ok);
     const bytes = (try codecs.toBytes(&arena, result.value)).?;
     defer arena.allocator.free(bytes);
     try std.testing.expectEqualStrings(expected, bytes);
     for ([_]u64{ 0xd800, 0x110000 }) |n| {
         const bad = try codecs.ofList(&arena, &.{try codecs.ofNumber(&arena, n)});
         try expectErrorText(&arena, try testAction(&arena, 38, bad, .{}), "invalid source scalar list");
     }
     // Noncanonical zero has the right numeric value, but is not a scalar tree.
     const zero = try arena.alloc(.leaf);
     const redundant = try arena.alloc(.{ .fork = .{ .left = zero, .right = zero } });
     const bad = try codecs.ofList(&arena, &.{redundant});
     try expectErrorText(&arena, try testAction(&arena, 38, bad, .{}), "invalid source scalar list");
 }
 
 test "readSourceFile preserves bytes and scalars, rejects malformed input, and enforces read permission" {
     var tmp = std.testing.tmpDir(.{});
     defer tmp.cleanup();
     const directory = try tmp.dir.realPathFileAlloc(std.testing.io, ".", std.testing.allocator);
     defer std.testing.allocator.free(directory);
     const path = try std.fmt.allocPrint(std.testing.allocator, "{s}/é😀.tri", .{directory});
     defer std.testing.allocator.free(path);
     var arena = Arena.init(std.testing.allocator);
     defer arena.deinit();
     var bytes: std.ArrayList(u8) = .empty;
     defer bytes.deinit(std.testing.allocator);
     try bytes.appendNTimes(std.testing.allocator, 'a', 4095);
     try bytes.appendSlice(std.testing.allocator, "😀\r\né\x00\u{feff}e\u{301}");
     for ([_][]const u8{ bytes.items, "" }) |contents| {
         try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "é😀.tri", .data = contents });
         const reply = try testAction(&arena, 37, try source_text.fromUtf8(&arena, path), .{ .allow_read_all = true });
         try std.testing.expect(reply.ok);
         const pair = arena.get(reply.value).fork;
         const raw = (try codecs.toBytes(&arena, pair.left)).?;
         defer arena.allocator.free(raw);
         const text = try source_text.toUtf8(&arena, pair.right);
         defer arena.allocator.free(text);
         try std.testing.expectEqualStrings(contents, raw);
         try std.testing.expectEqualStrings(contents, text);
     }
     try expectErrorText(&arena, try testAction(&arena, 37, try source_text.fromUtf8(&arena, path), .{}), "permission denied");
     for ([_][]const u8{ "\x80", "\xc0\xaf", "\xe2\x82", "\xed\xa0\x80", "\xf4\x90\x80\x80", "\xef\xbb\xbfok" }) |contents| {
         try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "é😀.tri", .data = contents });
         const reply = try testAction(&arena, 37, try source_text.fromUtf8(&arena, path), .{ .allow_read_all = true });
         try expectErrorText(&arena, reply, path);
         try expectErrorText(&arena, reply, if (std.mem.startsWith(u8, contents, "\xef\xbb\xbf")) "BOM is not permitted" else "invalid UTF-8 source");
     }
     // An embedded NUL must not turn a different path into the existing file.
     const nul_path = try std.fmt.allocPrint(std.testing.allocator, "{s}\x00ignored", .{path});
     defer std.testing.allocator.free(nul_path);
     try expectErrorText(&arena, try testAction(&arena, 37, try source_text.fromUtf8(&arena, nul_path), .{ .allow_read_all = true }), "invalid string");
     try tmp.dir.deleteFile(std.testing.io, "é😀.tri");
     try expectErrorText(&arena, try testAction(&arena, 37, try source_text.fromUtf8(&arena, path), .{ .allow_read_all = true }), "does not exist");
     // Ordinary readFile remains a byte transport: malformed UTF-8 and BOMs
     // are not interpreted. Use its existing byte-oriented path convention.
     const binary = "\xff\x00\xef\xbb\xbf\r\n";
     try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "é😀.tri", .data = binary });
     const reply = try testAction(&arena, 20, try codecs.ofString(&arena, path), .{ .allow_read_all = true });
     try std.testing.expect(reply.ok);
     const raw = (try codecs.toBytes(&arena, reply.value)).?;
     defer arena.allocator.free(raw);
     try std.testing.expectEqualStrings(binary, raw);
 }
 
 fn sourceIOAllocationProbe(allocator: std.mem.Allocator, path: []const u8) !void {
     var arena = Arena.init(allocator);
     defer arena.deinit();
     const reply = try testAction(&arena, 37, try source_text.fromUtf8(&arena, path), .{ .allow_read_all = true });
     try std.testing.expect(reply.ok);
     const scalars = arena.get(reply.value).fork.right;
     const encoded = try testAction(&arena, 38, scalars, .{});
     try std.testing.expect(encoded.ok);
     const raw = (try codecs.toBytes(&arena, encoded.value)).?;
     defer allocator.free(raw);
     try std.testing.expectEqualStrings("aé😀", raw);
 }
 
 test "source IO unwinds allocation failures without leaking tasks, buffers or requests" {
     var tmp = std.testing.tmpDir(.{});
     defer tmp.cleanup();
     try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "source.tri", .data = "aé😀" });
     const path = try tmp.dir.realPathFileAlloc(std.testing.io, "source.tri", std.testing.allocator);
     defer std.testing.allocator.free(path);
     try std.testing.checkAllAllocationFailures(std.testing.allocator, sourceIOAllocationProbe, .{path});
 }
 
 fn testIndexedAction(arena: *Arena, tag: u64, path: []const u8, hash: []const u8, value: u32, perms: IOPerms) !codecs.UnwrapResult {
     const path_tree = try source_text.fromUtf8(arena, path);
     const digest_tree = try codecs.ofString(arena, hash);
     const rest = if (tag == 52) try arena.alloc(.{ .fork = .{ .left = digest_tree, .right = value } }) else digest_tree;
     const args = try arena.alloc(.{ .fork = .{ .left = path_tree, .right = rest } });
     return testAction(arena, tag, args, perms);
 }
 
 test "indexed IO authenticates, enforces permissions before demand, and never replaces a different winner" {
     var tmp = std.testing.tmpDir(.{});
     defer tmp.cleanup();
     const directory = try tmp.dir.realPathFileAlloc(std.testing.io, ".", std.testing.allocator);
     defer std.testing.allocator.free(directory);
     const path = try std.fmt.allocPrint(std.testing.allocator, "{s}/é.arboricx", .{directory});
     defer std.testing.allocator.free(path);
     var arena = Arena.init(std.testing.allocator);
     defer arena.deinit();
     const leaf = try arena.alloc(.leaf);
     // Test helpers turn on collection; retain the tree across independent IO calls.
     try arena.roots.append(arena.allocator, leaf);
     const bytes = try indexed.encode(&arena, leaf);
     defer arena.allocator.free(bytes);
     const hash = indexed.digest(bytes);
     const perms = IOPerms{ .allow_read_all = true, .allow_write_all = true };
     const hashed = try testAction(&arena, 51, leaf, .{});
     try std.testing.expect(hashed.ok);
     const actual = (try codecs.toString(&arena, hashed.value)).?;
     defer arena.allocator.free(actual);
     try std.testing.expectEqualStrings(&hash, actual);
     for ([_]bool{ true, false }) |fresh| {
         const result = try testIndexedAction(&arena, 52, path, &hash, leaf, perms);
         try std.testing.expect(result.ok);
         try std.testing.expectEqual(fresh, (try codecs.toBool(&arena, result.value)).?);
     }
     const read = try testIndexedAction(&arena, 50, path, &hash, leaf, .{ .allow_read_all = true });
     try std.testing.expect(read.ok and arena.get(read.value).* == .leaf);
     // A denied write must not try to normalize a divergent supplied tree.
     const poison = try arena.alloc(.leaf);
     arena.get(poison).* = .{ .app = .{ .func = poison, .arg = leaf } };
     try arena.roots.append(arena.allocator, poison);
     for ([_]IOPerms{ .{}, .{ .allow_read_all = true }, .{ .allow_write_all = true } }) |denied| {
         try expectErrorText(&arena, try testIndexedAction(&arena, 52, path, &hash, poison, denied), "permission denied");
     }
     try expectErrorText(&arena, try testIndexedAction(&arena, 50, path, &hash, leaf, .{}), "permission denied");
     try expectErrorText(&arena, try testIndexedAction(&arena, 50, path, "ABC", leaf, perms), "invalid CAS digest");
     const wrong = "0" ** 64;
     try expectErrorText(&arena, try testIndexedAction(&arena, 50, path, wrong, leaf, perms), "CAS digest mismatch");
     try expectErrorText(&arena, try testIndexedAction(&arena, 52, path, wrong, leaf, perms), "CAS digest mismatch");
     try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "é.arboricx", .data = "corrupt" });
     try expectErrorText(&arena, try testIndexedAction(&arena, 52, path, &hash, leaf, perms), "different bytes");
     const still_corrupt = try readFsForTest(&arena, path);
     defer arena.allocator.free(still_corrupt);
     try std.testing.expectEqualStrings("corrupt", still_corrupt);
     const corrupt_hash = indexed.digest("corrupt");
     try expectErrorText(&arena, try testIndexedAction(&arena, 50, path, &corrupt_hash, leaf, perms), "invalid indexed tree");
 }
 
 fn readFsForTest(arena: *Arena, path: []const u8) ![]u8 {
     var loop: c.uv_loop_t = undefined;
     try std.testing.expectEqual(@as(c_int, 0), c.uv_loop_init(&loop));
     defer _ = c.uv_loop_close(&loop);
     const z = try arena.allocator.dupeZ(u8, path);
     defer arena.allocator.free(z);
     return switch (try readFs(&loop, arena.allocator, z)) {
         .bytes => |b| b,
         .failure => error.TestUnexpectedResult,
     };
 }
 
 fn indexedIOAllocationProbe(allocator: std.mem.Allocator, path: []const u8) !void {
     var arena = Arena.init(allocator);
     defer arena.deinit();
     const leaf = try arena.alloc(.leaf);
     try arena.roots.append(allocator, leaf);
     const hash = "6f93378feaa8bf5ab7cefc071175a654c871bfd53f165d7d96795fb2b38483f1";
     const perms = IOPerms{ .allow_read_all = true, .allow_write_all = true };
     const written = try testIndexedAction(&arena, 52, path, hash, leaf, perms);
     try std.testing.expect(written.ok);
     const read = try testIndexedAction(&arena, 50, path, hash, leaf, perms);
     try std.testing.expect(read.ok and arena.get(read.value).* == .leaf);
 }
 
 test "indexed publication and reads unwind every allocator failure" {
     var tmp = std.testing.tmpDir(.{});
     defer tmp.cleanup();
     const directory = try tmp.dir.realPathFileAlloc(std.testing.io, ".", std.testing.allocator);
     defer std.testing.allocator.free(directory);
     const path = try std.fmt.allocPrint(std.testing.allocator, "{s}/object", .{directory});
     defer std.testing.allocator.free(path);
     try std.testing.checkAllAllocationFailures(std.testing.allocator, indexedIOAllocationProbe, .{path});
 }