tricu

io_v2_file_host.zig (51825B)

---

 //! Synchronous native compiler/file profile, shared by the CLI and private
 //! integration probes. Explicit capabilities; no standard streams or broad OS parity.
 const std = @import("std");
 const Protocol = @import("io_v2_protocol.zig");
 const Core = @import("io_v2_core.zig");
 const F = @import("io_v2_files.zig");
 const Paths = @import("io_v2_paths.zig");
 const UTF8 = @import("io_v2_utf8.zig");
 const Stage = @import("io_v2_stage.zig");
 const Digest = @import("io_v2_digest.zig").Digest;
 const Snapshot = @import("io_v2_snapshot.zig");
 const Indexed = @import("indexed.zig");
 const C = @import("codecs.zig");
 const Text = @import("source_text.zig");
 pub const Host = struct {
     // Fixed-size host diagnostics remain inspectable even when a reducer/allocator
     // failure prevents producing a tree Result. No allocation or dangling node IDs.
     algorithm_detail: [512]u8 = undefined,
     algorithm_detail_len: usize = 0,
     utf8_detail: [96]u8 = undefined,
     utf8_detail_len: usize = 0,
     snapshot_cleanup_failed: bool = false,
     stage_operation: bool = false,
     stage_status: Stage.Status = .{},
     published: usize = 0,
     opened: usize = 0,
     closed: usize = 0,
     close_errors: [16]F.Status = undefined,
     close_error_count: usize = 0,
     close_errors_omitted: usize = 0,
     fn recordCloseError(self: *Host, status: F.Status) void {
         if (self.close_error_count < self.close_errors.len) {
             self.close_errors[self.close_error_count] = status;
             self.close_error_count += 1;
         } else self.close_errors_omitted += 1;
     }
     pub const Backend = union(enum) { placeholder, file: *F.File, digest: *Digest, stage: *Stage.Stage };
     pub const standard = [3]Backend{ .placeholder, .placeholder, .placeholder };
     fn fileBackend(file: *F.File) Backend {
         return .{ .file = file };
     }
     pub fn close(self: *Host, registry: anytype, backend: Backend) !void {
         var status: F.Status = .{};
         return self.closeWithStatus(registry, backend, &status);
     }
     fn closeWithStatus(self: *Host, registry: anytype, backend: Backend, status: *F.Status) !void {
         defer self.closed += 1;
         switch (backend) {
             .placeholder => unreachable,
             .digest => |digest| registry.allocator.destroy(digest),
             .file => |file| {
                 defer registry.allocator.destroy(file);
                 file.close(status) catch |err| {
                     self.recordCloseError(status.*);
                     return err;
                 };
             },
             .stage => |stage| {
                 defer registry.allocator.destroy(stage);
                 defer stage.deinit();
                 var released: Stage.Status = .{};
                 stage.release(&released) catch |err| {
                     status.* = .{ .code = released.code, .cleanup_code = released.secondary };
                     self.recordCloseError(status.*);
                     return err;
                 };
             },
         }
     }
     const Op = enum { open_reader, open_writer, open_appender, open_update, read, write, close, seek, position, size, resize, sync, stat, lstat, create_directory, remove_directory, remove_file, rename_path, sync_directory, open_digest, update_digest, finish_digest, open_stage, seal_stage, publish_stage, encode_utf8, decode_utf8, hash_indexed, write_indexed, verify_indexed, read_indexed, capabilities, inspect };
     const entries = .{
         .{ "runtime", "capabilities", Op.capabilities }, .{ "runtime", "inspect", Op.inspect },
         .{ "indexed-tree", "read", Op.read_indexed },    .{ "indexed-tree", "hash", Op.hash_indexed },
         .{ "indexed-tree", "write", Op.write_indexed },  .{ "indexed-tree", "verify", Op.verify_indexed },
         .{ "utf8", "encode", Op.encode_utf8 },           .{ "utf8", "decode", Op.decode_utf8 },
         .{ "stage", "open", Op.open_stage },             .{ "stage", "seal", Op.seal_stage },
         .{ "stage", "publish", Op.publish_stage },       .{ "digest", "open", Op.open_digest },
         .{ "digest", "update", Op.update_digest },       .{ "digest", "finish", Op.finish_digest },
         .{ "path", "stat", Op.stat },                    .{ "path", "lstat", Op.lstat },
         .{ "path", "remove-file", Op.remove_file },      .{ "path", "rename", Op.rename_path },
         .{ "directory", "create", Op.create_directory }, .{ "directory", "remove", Op.remove_directory },
         .{ "directory", "sync", Op.sync_directory },     .{ "file", "open-reader", Op.open_reader },
         .{ "file", "open-writer", Op.open_writer },      .{ "file", "open-appender", Op.open_appender },
         .{ "file", "open-update", Op.open_update },      .{ "stream", "read", Op.read },
         .{ "stream", "write", Op.write },                .{ "resource", "close", Op.close },
         .{ "file", "seek", Op.seek },                    .{ "file", "position", Op.position },
         .{ "file", "size", Op.size },                    .{ "file", "resize", Op.resize },
         .{ "file", "sync", Op.sync },
     };
     fn resolve(group: []const u8, operation: []const u8) ?Op {
         inline for (entries) |entry| if (std.mem.eql(u8, group, entry[0]) and std.mem.eql(u8, operation, entry[1])) return entry[2];
         return null;
     }
     pub fn request(self: *Host, engine: anytype, scope: u64, req: anytype) !u32 {
         if (req.revision != 1) return Core.unsupported(engine, req);
         const op = resolve(req.group.slice(), req.operation.slice()) orelse return Core.unsupported(engine, req);
         var name_buffer: [129]u8 = undefined;
         const name = try std.fmt.bufPrint(&name_buffer, "{s}/{s}", .{ req.group.slice(), req.operation.slice() });
         var status: F.Status = .{};
         self.stage_operation = false;
         self.snapshot_cleanup_failed = false;
         self.stage_status = .{};
         const value = self.dispatch(engine, scope, op, req.arguments, &status) catch |err| {
             if (err == error.InvalidAlgorithm) return engine.result(false, try engine.list(&.{
                 try engine.text("unsupported"),                                                       try engine.text(name),
                 try Text.fromUtf8(engine.arena, self.algorithm_detail[0..self.algorithm_detail_len]), try engine.text("none"),
             }));
             if (self.stage_operation) status.code = self.stage_status.code;
             const category = switch (err) {
                 error.InvalidArgument, error.EmptySnapshotDirectory => "invalid-argument",
                 error.InvalidData, error.SnapshotTruncated, error.SnapshotInvalid => "invalid-data",
                 error.SnapshotBound => "invalid-bound",
                 error.SnapshotDigestMismatch => "integrity-failed",
                 error.SnapshotCleanup => "cleanup-failed",
                 error.InvalidState => "invalid-state",
                 error.InvalidDestination => "invalid-destination",
                 error.InvalidBound, error.EncodedLimit => "invalid-bound",
                 error.DigestMismatch, error.FrameMismatch => "integrity-failed",
                 error.UnsupportedFileType, error.UnsupportedMode => "unsupported",
                 error.WrongAccess => "wrong-kind",
                 error.NotDirectory => "not-directory",
                 error.Closed, error.ClosedHandle => "closed-handle",
                 error.StaleHandle => "stale-handle",
                 error.OutOfScope => "out-of-scope",
                 error.Busy => "busy",
                 error.ClosedScope => "closed-scope",
                 error.BorrowedResource => "borrowed-resource",
                 error.InternalResource => "internal-resource",
                 error.HostFailure => F.errorCategory(status.code),
                 else => return err,
             };
             const detail = if (err == error.EmptySnapshotDirectory) "empty snapshot directory" else if (err == error.SnapshotBound) "invalid snapshot size limit" else if (err == error.SnapshotTruncated) "incomplete indexed frame" else if (err == error.SnapshotInvalid) "invalid indexed frame" else if (err == error.SnapshotDigestMismatch) "CAS digest mismatch" else if (err == error.SnapshotCleanup) "snapshot release failed" else if (err == error.EncodedLimit) "encoded tree exceeds limit" else if (err == error.DigestMismatch) "CAS digest mismatch" else if (err == error.FrameMismatch) "indexed frame differs from canonical tree" else if (err == error.InvalidData) self.utf8_detail[0..self.utf8_detail_len] else if (err == error.HostFailure) F.errorText(status.code) else if (err == error.UnsupportedFileType) "path-based files support regular files only" else if (err == error.UnsupportedMode) "operation unavailable in this file mode" else if (err == error.NotDirectory) "path is not a directory" else if (err == error.InvalidDestination) "destination must name a file outside the private staging namespace" else if (err == error.InvalidState) (if (self.stage_operation) "stage is not in the required phase" else "digest already finished") else "";
             // Mutating opens can create/truncate before a later failure. Never
             // claim absence of effects; this is not transactional rollback.
             const progress = if (err == error.SnapshotTruncated or err == error.SnapshotInvalid or err == error.SnapshotDigestMismatch) "unknown" else if (err == error.FrameMismatch) "unknown" else if (self.stage_operation) switch (self.stage_status.progress) {
                 .none => "none",
                 .not_published => "not-published",
                 .unknown => "unknown",
                 .published => "published",
             } else if (err == error.HostFailure and op != .open_reader and op != .size and op != .position and op != .stat and op != .lstat) "unknown" else "none";
             var detail_buffer: [1024]u8 = undefined;
             const reported = if (self.snapshot_cleanup_failed) try std.fmt.bufPrint(&detail_buffer, "{s}; cleanup: snapshot release failed", .{detail}) else detail;
             return engine.result(false, try engine.list(&.{ try engine.text(category), try engine.text(name), try engine.text(reported), try engine.text(progress) }));
         };
         return engine.result(true, value);
     }
     fn decodeText(engine: anytype, input: u32) ![]u8 {
         var output: std.ArrayList(u8) = .empty;
         errdefer output.deinit(engine.arena.allocator);
         var rest = input;
         const reader = engine.reader();
         while (try reader.node(rest) != .leaf) {
             const cell = try reader.pair(rest) orelse return error.InvalidArgument;
             const scalar = try reader.nat(cell.first, 21) orelse return error.InvalidArgument;
             if (scalar > 0x10ffff or (scalar >= 0xd800 and scalar <= 0xdfff)) return error.InvalidArgument;
             var encoded: [4]u8 = undefined;
             const count = try std.unicode.utf8Encode(@intCast(scalar), &encoded);
             try output.appendSlice(engine.arena.allocator, encoded[0..count]);
             rest = cell.second;
         }
         return output.toOwnedSlice(engine.arena.allocator);
     }
     fn decodeChunk(engine: anytype, input: u32, bytes: *[4096]u8) !usize {
         var count: usize = 0;
         var rest = input;
         const reader = engine.reader();
         while (try reader.node(rest) != .leaf) {
             // Match the protocol decoder (invalid argument), not the lower-level
             // byte-IO bound check. Observe excess spine, never excess head/tail.
             if (count == bytes.len) return error.InvalidArgument;
             const cell = try reader.pair(rest) orelse return error.InvalidArgument;
             bytes[count] = @intCast(try reader.nat(cell.first, 8) orelse return error.InvalidArgument);
             count += 1;
             rest = cell.second;
         }
         return count;
     }
     fn readIndexed(self: *Host, engine: anytype, scope: u64, arguments: u32, status: *F.Status) !u32 {
         const r = engine.reader();
         var fields: [5]u32 = undefined;
         var rest = arguments;
         for (&fields) |*field| {
             const cell = try r.pair(rest) orelse return error.InvalidArgument;
             field.* = cell.first;
             rest = cell.second;
         }
         if (try r.node(rest) != .leaf) return error.InvalidArgument;
         const token = try engine.token(fields[0], "io.v2.resource") orelse return error.InvalidArgument;
         const directory = try decodeText(engine, fields[1]);
         defer engine.arena.allocator.free(directory);
         const path = try engine.arena.allocator.dupeZ(u8, directory);
         defer engine.arena.allocator.free(path);
         const limit = try r.nat(fields[2], 63) orelse return error.InvalidArgument;
         const size = try r.nat(fields[3], 63) orelse return error.InvalidArgument;
         var digest: [64]u8 = undefined;
         rest = fields[4];
         for (&digest) |*byte| {
             const cell = try r.pair(rest) orelse return error.InvalidArgument;
             const scalar = try r.nat(cell.first, 7) orelse return error.InvalidArgument;
             if (!((scalar >= '0' and scalar <= '9') or (scalar >= 'a' and scalar <= 'f'))) return error.InvalidArgument;
             byte.* = @intCast(scalar);
             rest = cell.second;
         }
         if (try r.node(rest) != .leaf) return error.InvalidArgument;
         if (size > limit) return error.SnapshotBound;
         if (directory.len == 0) return error.EmptySnapshotDirectory;
         const registry = engine.scheduler.registry;
         const backend = (try registry.lookup(scope, token)).backend;
         if (backend != .file or (backend.file.mode != .reader and backend.file.mode != .update)) return error.WrongAccess;
         const lease = try registry.beginLease(scope, token, .exclusive, false);
         defer registry.endLease(lease.token) catch unreachable;
         const temporary = blk: {
             const admission = try registry.beginAcquire(scope);
             errdefer registry.abortAcquire(admission) catch unreachable;
             const file = try registry.allocator.create(F.File);
             errdefer registry.allocator.destroy(file);
             file.* = Snapshot.create(registry.allocator, status, path) catch |err| {
                 if (status.cleanup_code != 0) self.recordCloseError(.{ .code = status.cleanup_code });
                 return err;
             };
             errdefer {
                 var cleanup: F.Status = .{};
                 file.close(&cleanup) catch self.recordCloseError(cleanup);
             }
             break :blk try registry.publish(admission, .{ .file = file }, .{ .internal = true });
         };
         self.opened += 1;
         const file = (try registry.lookup(scope, temporary)).backend.file;
         const result = blk: {
             Snapshot.authenticate(file, backend.file, status, size, &digest) catch |err| break :blk @as(anyerror!u32, err);
             var snapshot = Snapshot.Snapshot{ .file = file, .status = status, .length = size };
             break :blk Indexed.loadSnapshot(engine.arena, &snapshot) catch |err| switch (err) {
                 error.InvalidData, error.ExportNotFound => error.SnapshotInvalid,
                 else => err,
             };
         };
         // All host work is synchronous. Retire the internal descriptor even on
         // ordinary Result failure or OOM; cleanup must not replace a primary fault.
         const owned = (try registry.claimClose(scope, temporary, true)).?;
         defer registry.finishClose(temporary) catch unreachable;
         var cleanup: F.Status = .{};
         self.closeWithStatus(registry, owned.backend, &cleanup) catch {
             if (result) |_| return error.SnapshotCleanup else |primary| {
                 self.snapshot_cleanup_failed = true;
                 return primary;
             }
         };
         return result;
     }
     fn indexed(self: *Host, engine: anytype, scope: u64, op: Op, arguments: u32, status: *F.Status) !u32 {
         const r = engine.reader();
         const first = try r.pair(arguments) orelse return error.InvalidArgument;
         if (op == .hash_indexed) {
             const limit = try r.nat(first.first, 63) orelse return error.InvalidArgument;
             var plan = try Indexed.prepare(engine.arena, first.second);
             defer plan.deinit();
             if (plan.encodedLength() > limit) return error.EncodedLimit;
             const hash = try plan.hash();
             return engine.list(&.{ try C.ofNumber(engine.arena, plan.encodedLength()), try engine.text(&hash) });
         }
         const token = try engine.token(first.first, "io.v2.resource") orelse return error.InvalidArgument;
         const second = try r.pair(first.second) orelse return error.InvalidArgument;
         const limit = try r.nat(second.first, 63) orelse return error.InvalidArgument;
         const third = try r.pair(second.second) orelse return error.InvalidArgument;
         const tree_field = if (op == .write_indexed) try r.pair(third.second) orelse return error.InvalidArgument else third;
         if (try r.node(tree_field.second) != .leaf) return error.InvalidArgument;
         var expected: [64]u8 = undefined;
         if (op == .write_indexed) {
             var rest = third.first;
             for (&expected) |*byte| {
                 const cell = try r.pair(rest) orelse return error.InvalidArgument;
                 const scalar = try r.nat(cell.first, 7) orelse return error.InvalidArgument;
                 if (!((scalar >= '0' and scalar <= '9') or (scalar >= 'a' and scalar <= 'f'))) return error.InvalidArgument;
                 byte.* = @intCast(scalar);
                 rest = cell.second;
             }
             if (try r.node(rest) != .leaf) return error.InvalidArgument;
         }
         const registry = engine.scheduler.registry;
         const backend = (try registry.lookup(scope, token)).backend;
         if (op == .verify_indexed) {
             if (backend != .file or (backend.file.mode != .reader and backend.file.mode != .update)) return error.WrongAccess;
         } else switch (backend) {
             .file => |file| if (file.mode == .reader) return error.WrongAccess,
             .stage => {},
             else => return error.WrongAccess,
         }
         const lease = try registry.beginLease(scope, token, .exclusive, false);
         defer registry.endLease(lease.token) catch unreachable;
         var plan = try Indexed.prepare(engine.arena, tree_field.first);
         defer plan.deinit();
         if (plan.encodedLength() > limit) return error.EncodedLimit;
         const hash = try plan.hash();
         if (op == .write_indexed and !std.mem.eql(u8, &hash, &expected)) return error.DigestMismatch;
         var sink = IndexedSink{ .host = self, .backend = backend, .status = status, .verify = op == .verify_indexed };
         try plan.emit(&sink);
         return engine.list(&.{ try C.ofNumber(engine.arena, plan.encodedLength()), try engine.text(&hash) });
     }
     const IndexedSink = struct {
         host: *Host,
         backend: Backend,
         status: *F.Status,
         verify: bool,
         pub fn writeAll(self: *@This(), bytes: []const u8) !void {
             var offset: usize = 0;
             while (offset < bytes.len) {
                 const count = if (self.verify) blk: {
                     var actual: [4096]u8 = undefined;
                     const n = try self.backend.file.read(self.status, actual[0 .. bytes.len - offset]);
                     if (n == 0 or !std.mem.eql(u8, actual[0..n], bytes[offset..][0..n])) return error.FrameMismatch;
                     break :blk n;
                 } else switch (self.backend) {
                     .file => |file| try file.write(self.status, bytes[offset..]),
                     .stage => |stage| blk: {
                         self.host.stage_operation = true;
                         break :blk try stage.write(&self.host.stage_status, bytes[offset..]);
                     },
                     else => unreachable,
                 };
                 std.debug.assert(count > 0 and count <= bytes.len - offset);
                 offset += count;
             }
         }
     };
     const core_names = [_][]const u8{ "pure", "bind", "fail", "scope", "current-scope", "spawn", "join", "discard", "cancel", "cancelled" };
     fn strings(engine: anytype, values: []const []const u8) !u32 {
         var ids: std.ArrayList(u32) = .empty;
         defer ids.deinit(engine.arena.allocator);
         for (values) |value| try ids.append(engine.arena.allocator, try engine.text(value));
         return engine.list(ids.items);
     }
     fn row(engine: anytype, key: []const u8, value: u32) !u32 {
         return engine.list(&.{ try engine.text(key), value });
     }
     fn capabilities(engine: anytype) !u32 {
         var groups: std.ArrayList(u32) = .empty;
         defer groups.deinit(engine.arena.allocator);
         inline for (.{ "runtime", "resource", "stream", "file", "path", "directory", "stage", "digest", "indexed-tree", "utf8" }) |group| {
             var operations: std.ArrayList(u32) = .empty;
             defer operations.deinit(engine.arena.allocator);
             inline for (entries) |entry| {
                 if (comptime std.mem.eql(u8, group, entry[0])) try operations.append(engine.arena.allocator, try engine.text(entry[1]));
             }
             try groups.append(engine.arena.allocator, try engine.list(&.{ try engine.text(group), try C.ofNumber(engine.arena, 1), try engine.list(operations.items) }));
         }
         const limits = try engine.list(&.{
             try row(engine, "chunk-bytes", try C.ofNumber(engine.arena, 4096)),
             try row(engine, "transfer-handles", try C.ofNumber(engine.arena, 256)),
             try row(engine, "file-offset-max", try C.ofNumber(engine.arena, F.limit)),
             try row(engine, "indexed-sections", try C.ofNumber(engine.arena, 64)),
             try row(engine, "indexed-manifest-bytes", try C.ofNumber(engine.arena, 65536)),
             try row(engine, "utf8-encode-scalars", try C.ofNumber(engine.arena, 1024)),
         });
         return engine.list(&.{
             try row(engine, "schema", try C.ofNumber(engine.arena, 1)),
             try row(engine, "io-version", try C.ofNumber(engine.arena, 2)),
             try row(engine, "backend", try engine.text("zig-native")),
             try row(engine, "authority", try engine.text("os-process")),
             try row(engine, "core", try strings(engine, &core_names)),
             try row(engine, "groups", try engine.list(groups.items)),
             try row(engine, "limits", limits),
             try row(engine, "digest-algorithms", try strings(engine, &.{"sha256"})),
             try row(engine, "restrictions", try strings(engine, &.{
                 "action-boundary-cancellation",    "no-hard-deadlines",              "synchronous-host-calls",
                 "path-files-regular-only",         "standard-streams-unsupported",   "directory-cursors-unsupported",
                 "file-single-cursor-lease",        "append-no-position-or-resize",   "stage-same-filesystem",
                 "stage-no-copy-fallback",          "stage-private-namespace",        "publish-new-requires-hard-links",
                 "no-crash-recovery",               "indexed-authenticated-snapshot", "indexed-caller-frame-and-disk-limit",
                 "indexed-dag-resident",            "indexed-writer-preflight",       "indexed-budget-not-heap-quota",
                 "sync-host-filesystem-guarantees", "utf8-strict-scalars",
             })),
         });
     }
     fn dispatch(self: *Host, engine: anytype, scope: u64, op: Op, arguments: u32, status: *F.Status) !u32 {
         const r = engine.reader();
         const registry = engine.scheduler.registry;
         if (op == .capabilities) {
             if (try r.node(arguments) != .leaf) return error.InvalidArgument;
             return capabilities(engine);
         }
         if (op == .inspect) {
             const action = Protocol.decode(engine.arena, arguments, r.fuel) catch |err| switch (err) {
                 error.InvalidComputationShape, error.UnknownComputationConstructor, error.InvalidFacilityHeader => return error.InvalidArgument,
                 else => return err,
             };
             return switch (action) {
                 .core => |value| engine.list(&.{ try engine.text("core"), try engine.text(core_names[@intFromEnum(value.code)]) }),
                 .request => |value| engine.list(&.{ try engine.text("facility"), try engine.text(value.group.slice()), try C.ofNumber(engine.arena, value.revision), try engine.text(value.operation.slice()), try engine.text(if (value.revision == 1 and resolve(value.group.slice(), value.operation.slice()) != null) "available" else "unsupported") }),
             };
         }
         if (op == .read_indexed) return self.readIndexed(engine, scope, arguments, status);
         if (op == .hash_indexed or op == .write_indexed or op == .verify_indexed) return self.indexed(engine, scope, op, arguments, status);
         if (op == .encode_utf8) {
             var bytes: [4096]u8 = undefined;
             var scalars: usize = 0;
             var used: usize = 0;
             var rest = arguments;
             while (try r.node(rest) != .leaf) {
                 if (scalars == 1024) return error.InvalidArgument;
                 const cell = try r.pair(rest) orelse return error.InvalidArgument;
                 const scalar = try r.nat(cell.first, 21) orelse return error.InvalidArgument;
                 if (scalar > 0x10ffff or (scalar >= 0xd800 and scalar <= 0xdfff)) return error.InvalidArgument;
                 used += try std.unicode.utf8Encode(@intCast(scalar), bytes[used..][0..4]);
                 scalars += 1;
                 rest = cell.second;
             }
             return C.ofBytes(engine.arena, bytes[0..used]);
         }
         if (op == .decode_utf8) {
             const final = try r.pair(arguments) orelse return error.InvalidArgument;
             const input = try r.pair(final.second) orelse return error.InvalidArgument;
             if (try r.node(input.second) != .leaf) return error.InvalidArgument;
             const ended = try r.nat(final.first, 1) orelse return error.InvalidArgument;
             var bytes: [4096]u8 = undefined;
             const count = try decodeChunk(engine, input.first, &bytes);
             return switch (UTF8.scan(bytes[0..count], ended == 1)) {
                 .too_large => unreachable, // argument decoder already bounds it
                 .fault => |fault| blk: {
                     const why: []const u8 = switch (fault.kind) {
                         .leading => "invalid UTF-8 leading byte",
                         .sequence => "invalid UTF-8 sequence",
                         .incomplete => "incomplete UTF-8 sequence",
                     };
                     const detail = try std.fmt.bufPrint(&self.utf8_detail, "{s} at byte {d}", .{ why, fault.offset });
                     self.utf8_detail_len = detail.len;
                     break :blk error.InvalidData;
                 },
                 .prefix => |prefix| try engine.list(&.{ try Text.fromUtf8(engine.arena, bytes[0..prefix]), try C.ofBytes(engine.arena, bytes[prefix..count]) }),
             };
         }
         switch (op) {
             .stat, .lstat, .create_directory, .remove_directory, .remove_file, .rename_path, .sync_directory => {
                 const paths = if (op == .rename_path) try r.pair(arguments) orelse return error.InvalidArgument else null;
                 const raw = try decodeText(engine, if (paths) |p| p.first else arguments);
                 defer engine.arena.allocator.free(raw);
                 const path = try engine.arena.allocator.dupeZ(u8, raw);
                 defer engine.arena.allocator.free(path);
                 if (op == .stat or op == .lstat) {
                     const info = try Paths.stat(status, path, op == .stat);
                     return switch (info) {
                         .regular => |size| try engine.list(&.{ try engine.text("file"), try C.ofNumber(engine.arena, size) }),
                         .directory => try engine.list(&.{try engine.text("directory")}),
                         .symlink => try engine.list(&.{try engine.text("symlink")}),
                         .other => try engine.list(&.{try engine.text("other")}),
                     };
                 }
                 if (op == .sync_directory) {
                     Paths.syncDirectory(status, path) catch |err| {
                         if (status.cleanup_code != 0) self.recordCloseError(.{ .code = status.cleanup_code });
                         return err;
                     };
                     return engine.leaf;
                 }
                 if (op == .rename_path) {
                     const other_raw = try decodeText(engine, paths.?.second);
                     defer engine.arena.allocator.free(other_raw);
                     const other = try engine.arena.allocator.dupeZ(u8, other_raw);
                     defer engine.arena.allocator.free(other);
                     try Paths.mutate(status, .rename, path, other);
                 } else {
                     try Paths.mutate(status, switch (op) {
                         .create_directory => .create_directory,
                         .remove_directory => .remove_directory,
                         .remove_file => .remove_file,
                         else => unreachable,
                     }, path, null);
                 }
                 return engine.leaf;
             },
             else => {},
         }
         switch (op) {
             .open_reader, .open_writer, .open_appender, .open_update, .open_digest, .open_stage => {
                 const pair = try r.pair(arguments) orelse return error.InvalidArgument;
                 const owner = try engine.token(pair.first, "io.v2.scope") orelse return error.InvalidArgument;
                 if (owner.run != registry.identity or !registry.within(scope, owner.key)) return error.InvalidArgument;
                 const text = try decodeText(engine, pair.second);
                 defer engine.arena.allocator.free(text);
                 const path = try engine.arena.allocator.dupeZ(u8, text);
                 defer engine.arena.allocator.free(path);
                 if (op == .open_digest) {
                     if (!std.mem.eql(u8, text, "sha256")) {
                         const view = try std.unicode.Utf8View.init(text);
                         var it = view.iterator();
                         var scalars: usize = 0;
                         while (scalars < 128 and it.nextCodepoint() != null) : (scalars += 1) {}
                         self.algorithm_detail_len = it.i;
                         @memcpy(self.algorithm_detail[0..it.i], text[0..it.i]);
                         return error.InvalidAlgorithm;
                     }
                     // End acquisition rollback BEFORE encoding the source token.
                     // Once published, only scope teardown owns this state, even
                     // when reply allocation fails immediately afterwards.
                     const token = acquired: {
                         const admission = try registry.beginAcquire(owner.key);
                         errdefer registry.abortAcquire(admission) catch unreachable;
                         const digest = try registry.allocator.create(Digest);
                         errdefer registry.allocator.destroy(digest);
                         digest.* = .{};
                         break :acquired try registry.publish(admission, .{ .digest = digest }, .{});
                     };
                     self.opened += 1;
                     return engine.marker("io.v2.resource", token.key);
                 }
                 if (op == .open_stage) {
                     self.stage_operation = true;
                     const token = acquired: {
                         const admission = try registry.beginAcquire(owner.key);
                         errdefer registry.abortAcquire(admission) catch unreachable;
                         const stage = try registry.allocator.create(Stage.Stage);
                         errdefer registry.allocator.destroy(stage);
                         stage.* = try Stage.Stage.init(registry.allocator, &self.stage_status, path);
                         errdefer {
                             var released: Stage.Status = .{};
                             stage.release(&released) catch self.recordCloseError(.{ .code = released.code, .cleanup_code = released.secondary });
                             stage.deinit();
                         }
                         break :acquired try registry.publish(admission, .{ .stage = stage }, .{});
                     };
                     self.opened += 1;
                     return engine.marker("io.v2.resource", token.key);
                 }
                 const mode: F.Mode = switch (op) {
                     .open_reader => .reader,
                     .open_writer => .writer,
                     .open_appender => .appender,
                     .open_update => .update,
                     else => unreachable,
                 };
                 const token = F.acquireWrapped(registry, owner.key, status, path, mode, fileBackend) catch |err| {
                     if (status.cleanup_code != 0) self.recordCloseError(.{ .code = status.cleanup_code });
                     return err;
                 };
                 self.opened += 1;
                 return engine.marker("io.v2.resource", token.key);
             },
             else => {},
         }
         const paired = op == .read or op == .write or op == .update_digest or op == .seek or op == .resize or op == .publish_stage;
         const pair = if (paired) try r.pair(arguments) orelse return error.InvalidArgument else null;
         const token = try engine.token(if (pair) |p| p.first else arguments, "io.v2.resource") orelse return error.InvalidArgument;
         var number: u64 = 0;
         var bytes: [4096]u8 = undefined;
         var count: usize = 0;
         switch (op) {
             .read, .seek, .resize => {
                 number = try r.nat(pair.?.second, if (op == .read) 16 else 63) orelse return error.InvalidArgument;
                 if (op == .read and (number == 0 or number > 4096)) return error.InvalidBound;
             },
             .write, .update_digest => {
                 count = try decodeChunk(engine, pair.?.second, &bytes);
                 if (count == 0) return error.InvalidBound;
             },
             else => {},
         }
         var destination: ?[:0]u8 = null;
         defer if (destination) |path| engine.arena.allocator.free(path);
         if (op == .publish_stage) {
             const fields = try r.pair(pair.?.second) orelse return error.InvalidArgument;
             number = try r.nat(fields.first, 1) orelse return error.InvalidArgument;
             const raw = try decodeText(engine, fields.second);
             defer engine.arena.allocator.free(raw);
             destination = try engine.arena.allocator.dupeZ(u8, raw);
         }
         if (op == .close) {
             const resource = try registry.claimClose(scope, token, false) orelse return error.Busy;
             defer registry.finishClose(token) catch unreachable;
             try self.closeWithStatus(registry, resource.backend, status);
             return engine.leaf;
         }
         const backend = (try registry.lookup(scope, token)).backend;
         if (op == .seal_stage or op == .publish_stage or (op == .write and backend == .stage)) {
             if (backend != .stage) return error.WrongAccess;
             const lease = try registry.beginLease(scope, token, .exclusive, false);
             defer registry.endLease(lease.token) catch unreachable;
             self.stage_operation = true;
             if (op == .write) return C.ofNumber(engine.arena, try backend.stage.write(&self.stage_status, bytes[0..count]));
             if (op == .seal_stage) {
                 try backend.stage.seal(&self.stage_status);
                 return engine.leaf;
             }
             const published = try backend.stage.publish(&self.stage_status, number == 1, destination.?);
             if (published == .published) self.published += 1;
             return engine.text(if (published == .published) "published" else "exists");
         }
         if (op == .update_digest or op == .finish_digest) {
             if (backend != .digest) return error.WrongAccess;
             const lease = try registry.beginLease(scope, token, .exclusive, false);
             defer registry.endLease(lease.token) catch unreachable;
             if (op == .update_digest) {
                 try backend.digest.update(bytes[0..count]);
                 return engine.leaf;
             }
             const hex = try backend.digest.finish();
             return engine.text(&hex);
         }
         if (backend != .file) return error.WrongAccess;
         const file = backend.file;
         const lease = try registry.beginLease(scope, token, .exclusive, false);
         defer registry.endLease(lease.token) catch unreachable;
         return switch (op) {
             .read => blk: {
                 const n = try file.read(status, bytes[0..@intCast(number)]);
                 const label = try engine.text(if (n == 0) "eof" else "data");
                 break :blk try engine.arena.alloc(.{ .fork = .{ .left = label, .right = if (n == 0) engine.leaf else try C.ofBytes(engine.arena, bytes[0..n]) } });
             },
             .write => try C.ofNumber(engine.arena, try file.write(status, bytes[0..count])),
             .seek => blk: {
                 try file.seek(number);
                 break :blk engine.leaf;
             },
             .resize => blk: {
                 try file.resize(status, number);
                 break :blk engine.leaf;
             },
             .position => try C.ofNumber(engine.arena, try file.position()),
             .size => try C.ofNumber(engine.arena, try file.size(status)),
             .sync => blk: {
                 try file.sync(status);
                 break :blk engine.leaf;
             },
             else => unreachable,
         };
     }
 };
 
 fn testProgram(a: *@import("arena.zig").Arena, path: []const u8, terminal: ?u64) !u32 {
     return testOpenProgram(a, path, terminal, "file", "open-reader");
 }
 fn testOpenProgram(a: *@import("arena.zig").Arena, path: []const u8, terminal: ?u64, group: []const u8, operation: []const u8) !u32 {
     const leaf = try a.alloc(.leaf);
     const scope = try C.ofList(a, &.{ try C.ofString(a, "io.v2.scope"), try C.ofNumber(a, 1), leaf });
     const args = try a.alloc(.{ .fork = .{ .left = scope, .right = try Text.fromUtf8(a, path) } });
     const named = try a.alloc(.{ .fork = .{ .left = try C.ofString(a, group), .right = try a.alloc(.{ .fork = .{ .left = try C.ofNumber(a, 1), .right = try a.alloc(.{ .fork = .{ .left = try C.ofString(a, operation), .right = args } }) } }) } });
     const request = try a.alloc(.{ .fork = .{ .left = try C.ofNumber(a, 10), .right = named } });
     if (terminal) |tag| {
         const next = try a.alloc(.{ .fork = .{ .left = try C.ofNumber(a, tag), .right = leaf } });
         const constant = try a.alloc(.{ .fork = .{ .left = leaf, .right = next } });
         return a.alloc(.{ .fork = .{ .left = try C.ofNumber(a, 1), .right = try a.alloc(.{ .fork = .{ .left = request, .right = constant } }) } });
     }
     return request;
 }
 fn failureProbe(allocator: std.mem.Allocator, path: []const u8) !void {
     var a = @import("arena.zig").Arena.init(allocator);
     defer a.deinit();
     const program = try testProgram(&a, path, null);
     a.gc_enabled = true;
     try a.setGcPolicy(1, 1);
     var host: Host = .{};
     defer std.debug.assert(host.opened == host.closed);
     _ = try Core.runWithHost(Host, &host, &a, program, 1, 10000);
     try std.testing.expectEqual(@as(usize, 1), host.opened);
 }
 test "physical resources settle on success, fail, cancellation, protocol fault and allocation failure" {
     const testing = std.testing;
     var tmp = testing.tmpDir(.{});
     defer tmp.cleanup();
     try tmp.dir.writeFile(testing.io, .{ .sub_path = "file", .data = "abc" });
     const path = try tmp.dir.realPathFileAlloc(testing.io, "file", testing.allocator);
     defer testing.allocator.free(path);
     for ([_]u64{ 0, 2, 9, 11 }) |tag| {
         var a = @import("arena.zig").Arena.init(testing.allocator);
         defer a.deinit();
         const program = try testProgram(&a, path, tag);
         var host: Host = .{};
         const outcome = try Core.runWithHost(Host, &host, &a, program, 1, 10000);
         try testing.expectEqual(@as(usize, 1), host.opened);
         try testing.expectEqual(host.opened, host.closed);
         try testing.expectEqual(@as(usize, 0), host.close_error_count);
         const expected: @TypeOf(outcome.kind) = if (tag == 0) .completed else if (tag == 9) .cancelled else .failed;
         try testing.expectEqual(expected, outcome.kind);
     }
     try testing.checkAllAllocationFailures(testing.allocator, failureProbe, .{path});
 }
 
 const CloseFailureHost = struct {
     pub const Backend = Host.Backend;
     pub const standard = Host.standard;
     inner: Host = .{},
     pub fn request(self: *@This(), engine: anytype, scope: u64, req: anytype) !u32 {
         return self.inner.request(engine, scope, req);
     }
     pub fn close(self: *@This(), registry: anytype, backend: Backend) !void {
         // Simulate a close that has released its descriptor but reports failure.
         // No external FD is closed/reused, and every real descriptor is released.
         var original = backend.file.*;
         var status: F.Status = .{};
         try original.close(&status);
         backend.file.fd = -1;
         try self.inner.close(registry, backend);
     }
 };
 test "physical close failures defeat success but retain primary failure and cancellation" {
     const testing = std.testing;
     var tmp = testing.tmpDir(.{});
     defer tmp.cleanup();
     try tmp.dir.writeFile(testing.io, .{ .sub_path = "file", .data = "abc" });
     const path = try tmp.dir.realPathFileAlloc(testing.io, "file", testing.allocator);
     defer testing.allocator.free(path);
     for ([_]u64{ 0, 2, 9 }) |tag| {
         var a = @import("arena.zig").Arena.init(testing.allocator);
         defer a.deinit();
         const program = try testProgram(&a, path, tag);
         var host: CloseFailureHost = .{};
         const outcome = try Core.runWithHost(CloseFailureHost, &host, &a, program, 1, 10000);
         try testing.expectEqual(@as(usize, 1), host.inner.closed);
         try testing.expectEqual(@as(usize, 1), host.inner.close_error_count);
         const expected: @TypeOf(outcome.kind) = if (tag == 9) .cancelled else .failed;
         try testing.expectEqual(expected, outcome.kind);
         if (tag != 0) {
             try testing.expect(a.get(outcome.value).* == .leaf);
             try testing.expectEqual(@as(usize, 1), outcome.cleanup.len);
         }
     }
 }
 
 fn digestAllocationProbe(allocator: std.mem.Allocator) !void {
     var a = @import("arena.zig").Arena.init(allocator);
     defer a.deinit();
     const program = try testOpenProgram(&a, "sha256", null, "digest", "open");
     a.gc_enabled = true;
     try a.setGcPolicy(1, 1);
     var host: Host = .{};
     defer std.debug.assert(host.opened == host.closed);
     _ = try Core.runWithHost(Host, &host, &a, program, 1, 10000);
     try std.testing.expectEqual(@as(usize, 1), host.opened);
 }
 test "digest acquisition and reply allocation failures retire every admitted state" {
     try std.testing.checkAllAllocationFailures(std.testing.allocator, digestAllocationProbe, .{});
 }
 
 fn stageAllocationProbe(allocator: std.mem.Allocator, directory: []const u8, dir: std.Io.Dir) !void {
     var a = @import("arena.zig").Arena.init(allocator);
     defer a.deinit();
     const program = try testOpenProgram(&a, directory, null, "stage", "open");
     a.gc_enabled = true;
     try a.setGcPolicy(1, 1);
     var host: Host = .{};
     defer std.debug.assert(host.opened == host.closed);
     const outcome = Core.runWithHost(Host, &host, &a, program, 1, 10000);
     var iterator = dir.iterate();
     try std.testing.expect((try iterator.next(std.testing.io)) == null);
     _ = try outcome;
     try std.testing.expectEqual(@as(usize, 1), host.opened);
 }
 test "stage acquisition and reply allocation failures remove every private name" {
     const testing = std.testing;
     var tmp = testing.tmpDir(.{ .iterate = true });
     defer tmp.cleanup();
     const directory = try tmp.dir.realPathFileAlloc(testing.io, ".", testing.allocator);
     defer testing.allocator.free(directory);
     try testing.checkAllAllocationFailures(testing.allocator, stageAllocationProbe, .{ directory, tmp.dir });
 }
 
 fn testRequest(a: *@import("arena.zig").Arena, group: []const u8, operation: []const u8, args: u32) !u32 {
     const payload = try a.alloc(.{ .fork = .{ .left = try C.ofString(a, group), .right = try a.alloc(.{ .fork = .{ .left = try C.ofNumber(a, 1), .right = try a.alloc(.{ .fork = .{ .left = try C.ofString(a, operation), .right = args } }) } }) } });
     return a.alloc(.{ .fork = .{ .left = try C.ofNumber(a, 10), .right = payload } });
 }
 fn then(a: *@import("arena.zig").Arena, left: u32, right: u32) !u32 {
     const constant = try a.alloc(.{ .fork = .{ .left = try a.alloc(.leaf), .right = right } });
     return a.alloc(.{ .fork = .{ .left = try C.ofNumber(a, 1), .right = try a.alloc(.{ .fork = .{ .left = left, .right = constant } }) } });
 }
 fn publicationAllocationProbe(allocator: std.mem.Allocator, directory: []const u8, destination: []const u8, dir: std.Io.Dir) !void {
     var a = @import("arena.zig").Arena.init(allocator);
     defer a.deinit();
     const opened = try testOpenProgram(&a, directory, null, "stage", "open");
     const token = try C.ofList(&a, &.{ try C.ofString(&a, "io.v2.resource"), try C.ofNumber(&a, 1), try C.ofNumber(&a, 5) });
     const sealed = try testRequest(&a, "stage", "seal", token);
     const args = try a.alloc(.{ .fork = .{ .left = token, .right = try a.alloc(.{ .fork = .{ .left = try C.ofNumber(&a, 1), .right = try Text.fromUtf8(&a, destination) } }) } });
     const publish = try testRequest(&a, "stage", "publish", args);
     const program = try then(&a, opened, try then(&a, sealed, publish));
     a.gc_enabled = true;
     try a.setGcPolicy(1, 1);
     var host: Host = .{};
     const result = Core.runWithHost(Host, &host, &a, program, 1, 10000);
     try std.testing.expectEqual(host.opened, host.closed);
     var iterator = dir.iterate();
     var visible: usize = 0;
     while (try iterator.next(std.testing.io)) |entry| {
         try std.testing.expectEqualStrings("out", entry.name); // never a private name
         visible += 1;
     }
     try std.testing.expectEqual(host.published, visible); // OOM cannot roll back visibility
     if (visible != 0) try dir.deleteFile(std.testing.io, "out");
     _ = try result;
     try std.testing.expectEqual(@as(usize, 1), host.published);
 }
 test "allocation failures before and after publication clean private names without rollback" {
     const testing = std.testing;
     var tmp = testing.tmpDir(.{ .iterate = true });
     defer tmp.cleanup();
     const directory = try tmp.dir.realPathFileAlloc(testing.io, ".", testing.allocator);
     defer testing.allocator.free(directory);
     const destination = try std.fmt.allocPrint(testing.allocator, "{s}/out", .{directory});
     defer testing.allocator.free(destination);
     try testing.checkAllAllocationFailures(testing.allocator, publicationAllocationProbe, .{ directory, destination, tmp.dir });
 }
 
 fn indexedWriteAllocationProbe(allocator: std.mem.Allocator, path: []const u8, dir: std.Io.Dir) !void {
     defer dir.deleteFile(std.testing.io, "out") catch {};
     var a = @import("arena.zig").Arena.init(allocator);
     defer a.deinit();
     const opened = try testOpenProgram(&a, path, null, "file", "open-writer");
     const token = try C.ofList(&a, &.{ try C.ofString(&a, "io.v2.resource"), try C.ofNumber(&a, 1), try C.ofNumber(&a, 5) });
     const args = try C.ofList(&a, &.{ token, try C.ofNumber(&a, 431), try C.ofString(&a, "6f93378feaa8bf5ab7cefc071175a654c871bfd53f165d7d96795fb2b38483f1"), try a.alloc(.leaf) });
     const program = try then(&a, opened, try testRequest(&a, "indexed-tree", "write", args));
     a.gc_enabled = true;
     try a.setGcPolicy(1, 1);
     var host: Host = .{};
     const outcome = Core.runWithHost(Host, &host, &a, program, 1, 10000);
     try std.testing.expectEqual(host.opened, host.closed);
     if (host.opened != 0) {
         const bytes = try dir.readFileAlloc(std.testing.io, "out", std.testing.allocator, .limited(1024));
         defer std.testing.allocator.free(bytes);
         // Preflight never writes; a failure encoding the reply cannot undo the
         // completed write. There are no allocations inside native emission.
         try std.testing.expect(bytes.len == 0 or bytes.len == 431);
         if (bytes.len != 0) try std.testing.expectEqualStrings("6f93378feaa8bf5ab7cefc071175a654c871bfd53f165d7d96795fb2b38483f1", &Indexed.digest(bytes));
         if (outcome) |_| try std.testing.expectEqual(@as(usize, 431), bytes.len) else |_| {}
     }
     _ = try outcome;
     try std.testing.expectEqual(@as(usize, 1), host.opened);
 }
 test "indexed writer preflight and reply OOM settle leases without rolling back writes" {
     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}/out", .{directory});
     defer std.testing.allocator.free(path);
     try std.testing.checkAllAllocationFailures(std.testing.allocator, indexedWriteAllocationProbe, .{ path, tmp.dir });
 }
 
 fn snapshotAllocationProbe(allocator: std.mem.Allocator, path: []const u8, directory: []const u8, dir: std.Io.Dir) !void {
     var a = @import("arena.zig").Arena.init(allocator);
     defer a.deinit();
     const opened = try testOpenProgram(&a, path, null, "file", "open-reader");
     const token = try C.ofList(&a, &.{ try C.ofString(&a, "io.v2.resource"), try C.ofNumber(&a, 1), try C.ofNumber(&a, 5) });
     const args = try C.ofList(&a, &.{ token, try Text.fromUtf8(&a, directory), try C.ofNumber(&a, 431), try C.ofNumber(&a, 431), try C.ofString(&a, "6f93378feaa8bf5ab7cefc071175a654c871bfd53f165d7d96795fb2b38483f1") });
     const program = try then(&a, opened, try testRequest(&a, "indexed-tree", "read", args));
     a.gc_enabled = true;
     try a.setGcPolicy(1, 1);
     var host: Host = .{};
     const outcome = Core.runWithHost(Host, &host, &a, program, 1, 10000);
     try std.testing.expectEqual(host.opened, host.closed);
     var iterator = dir.iterate();
     while (try iterator.next(std.testing.io)) |entry| try std.testing.expectEqualStrings("in", entry.name);
     _ = try outcome;
     try std.testing.expectEqual(@as(usize, 2), host.opened);
 }
 test "snapshot acquisition decode and reply OOM close every file without private names" {
     var tmp = std.testing.tmpDir(.{ .iterate = true });
     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}/in", .{directory});
     defer std.testing.allocator.free(path);
     var arena = @import("arena.zig").Arena.init(std.testing.allocator);
     defer arena.deinit();
     const bytes = try Indexed.encode(&arena, try arena.alloc(.leaf));
     defer std.testing.allocator.free(bytes);
     try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "in", .data = bytes });
     try std.testing.checkAllAllocationFailures(std.testing.allocator, snapshotAllocationProbe, .{ path, directory, tmp.dir });
 }