tiny.acp.ReaderStorage

1/// A caller uses this type to read lines from a pipe or a file within a memory budget fixed in2/// advance, as the client does with the agent's output. The struct splits incoming bytes into lines3/// inside one buffer that `init` allocates once. The caller moves the bytes: it polls for a line,4/// and on `need_input` it reads into `writable` and passes the count to `commit`, or calls `finish`5/// at the end of input. After `activate`, nothing the reader does allocates. A line past the limit6/// leaves the reader terminal. The order of calls is `init`, `activate`, the steady calls, then7/// `deinit`, and each call asserts the phase it needs. The struct keeps no allocator: `init` and8/// `deinit` each take one. The package exports the type as `acp.ReaderStorage`.9pub const Storage = struct {10    /// The reader's current lifecycle step.11    phase: alloc_phase.capacity.Phase,12    /// The line limit and buffer size worked out by `init`.13    capacity: capacity_mod.Capacity,14    /// The buffer, the buffer size long. The slice is empty after `deinit`.15    bytes: []u8,16    /// The index of the first byte not yet returned as a line. The index is 0 at the start and17    /// after each round of moving the unread bytes to the front of the buffer (compaction).18    start: usize = 0,19    /// The index one past the last byte read in.20    end: usize = 0,21    /// Set by `finish` when no more input will come.22    eof: bool = false,23    /// Whether the reader has stopped for good, set when a line runs past the limit, after which24    /// every `poll` fails with `error.ReaderMessageCapacityExceeded`.25    terminal: bool = false,26    /// The longest line seen, capped at the limit, in bytes.27    high_water_message_bytes: usize = 0,28    /// The number of lines rejected for running past the limit, which stops at the largest `u64`.29    rejected_message_count: u64 = 0,3031    /// The limits type, `reader.Limits`.32    pub const Limits: type = capacity_mod.Limits;33    /// The capacity type, `reader.Capacity`.34    pub const Capacity: type = capacity_mod.Capacity;35    /// The error set of a running reader, `Exhaustion`.36    pub const Exhaustion: type = @import("storage.zig").Exhaustion;37    /// The errors `init` returns: `error.OutOfMemory` from the allocator and38    /// `error.CapacityOverflow` from working out the buffer size. Another program in the repository39    /// reuses it as its own `init` error set.40    pub const InitError = std.mem.Allocator.Error || capacity_mod.DeriveError;4142    /// The repository's compile-time capacity checks read this declaration to hold the reader to43    /// the memory it states. The declaration provides a compile-time record of the reader's memory,44    /// with the id `acp.reader_storage`. It declares one region of the line limit plus one byte,45    /// sized by the caller's limit and held for the reader's steady life. The record lists what the46    /// region leaves out, in three groups. The first group covers threaded I/O, process pipes and47    /// kernel buffers. The second group covers parsed JSON, protocol results, updates and48    /// permission records. The third group covers outgoing messages and the capture of reply text.49    /// At overload, a line one byte past the limit ends the reading before any JSON parsing or call50    /// to the caller's callbacks for updates and permission decisions (observer). Splitting lines51    /// and compaction allocate nothing after activation, and a test witnesses it. Ten promises52    /// define the claim, each named by the test that witnesses it. The function `activate` seals53    /// the reader and `deinit` tears it down. A compile-time check at the end of the file checks54    /// the type's shape against the declaration. Each such test names its promise's key in a55    /// `@stardustClaim` block.56    pub const claim: alloc_phase.capacity.Declaration = .{57        .source = .{58            .id = "acp.reader_storage",59            .kind = .phase_static,60            .limit_source = .caller,61            .storage = .{62                .covered = &.{63                    .{64                        .id = "one_reusable_acp_newline_framing_and_message_region",65                        .lifetime = .steady,66                        .detail = "one reusable ACP newline-framing and message region",67                    },68                },69                .excluded = &.{70                    "threaded I/O, process pipes, and kernel buffering",71                    "parsed JSON, protocol results, updates, and permission ownership",72                    "outbound serialization and prompt capture",73                },74            },75            .capacity = .{76                .inputs = &.{77                    alloc_phase.capacity.bindInput(Limits, "message_bytes", "message_bytes"),78                },79                .type_selectors = &.{},80                .nodes = &.{81                    .{ .input = 0 },82                    .{ .constant = 1 },83                    .{ .add = .{ .left = 0, .right = 1 } },84                },85                .assertions = &.{.{86                    .scope = .closure_total,87                    .measure = .retained,88                    .relation = .exact,89                    .expression = 2,90                }},91            },92            .overload = .{93                .kind = .terminal,94                .detail = "max plus one byte ends framing before JSON parsing or observer publication",95            },96            .risks = .{97                .transitive = .{98                    .status = .witnessed,99                    .detail = "fragmented framing and compaction allocate nothing after activation",100                },101                .foreign = .{102                    .status = .excluded,103                    .detail = "process transport and kernel buffering are outside reader storage",104                },105            },106            .obligations = &.{107                .{ .key = "acp_reader_capacity", .role = .capacity_model },108                .{ .key = "acp_reader_acquisition", .role = .custom },109                .{ .key = "acp_reader_oom", .role = .custom },110                .{ .key = "acp_reader_boundary", .role = .overload },111                .{ .key = "acp_reader_fragmented", .role = .custom },112                .{ .key = "acp_reader_sealed", .role = .transitive_risk },113                .{ .key = "acp_reader_client_terminal_overload", .role = .overload },114                .{ .key = "acp_reader_client_terminal_foreign_risk", .role = .foreign_risk },115                .{ .key = "acp_reader_admission", .role = .custom },116                .{ .key = "acp_reader_root", .role = .custom },117            },118        },119        .bindings = .{120            .owner = @This(),121            .seal = .{122                .family = alloc_phase.capacity.selector(@This().activate),123                .premise = .{124                    .class = .checked_semantic_fact,125                    .authority = .checker,126                },127            },128            .teardown = .{129                .family = alloc_phase.capacity.selector(@This().deinit),130                .premise = .{131                    .class = .checked_semantic_fact,132                    .authority = .checker,133                },134            },135        },136    };137138    /// A caller uses this function to set up the reader, before the caller starts reading. The139    /// function allocates the buffer, the line limit plus one byte, in one allocation. The call140    /// returns the reader in its initialization phase, and the caller calls `activate` before the141    /// first poll. The function fails with `error.CapacityOverflow` before it allocates when the142    /// limit is too large to add one to, and with `error.OutOfMemory` when the allocation fails.143    /// The function leaves nothing allocated when it fails. The caller frees the buffer with144    /// `deinit` and the same allocator.145    pub fn init(allocator: std.mem.Allocator, limits: Limits) InitError!Storage {146        const capacity = try Capacity.derive(limits);147        const bytes = try allocator.alloc(u8, capacity.storage_bytes);148        const storage = Storage{149            .phase = .initialization,150            .capacity = capacity,151            .bytes = bytes,152        };153        std.debug.assert(storage.bytes.len == storage.capacity.storage_bytes);154        std.debug.assert(storage.capacity.storage_bytes > storage.capacity.message_bytes);155        return storage;156    }157158    /// A caller uses this function to mark the end of setup, after which the reader allocates159    /// nothing. The call moves the reader from its initialization phase to its steady phase. The160    /// caller calls it once, after `init` and before the first poll.161    pub fn activate(self: *Storage) void {162        std.debug.assert(self.phase == .initialization);163        std.debug.assert(self.bytes.len == self.capacity.storage_bytes);164        self.phase = .steady;165    }166167    /// A caller's read loop calls this function first on each turn to take the next line. The168    /// function returns the next whole line without its newline and moves past it. With no whole169    /// line buffered and more than the limit waiting, the call fails with170    /// `error.ReaderMessageCapacityExceeded` and the reader turns terminal. After `finish`, the171    /// function returns the remaining bytes as a last line, then `end`. Otherwise the function172    /// returns `need_input`. Once the reader is terminal, every call fails with the same error. A173    /// line of exactly the limit passes, and one byte more fails. The returned line stays valid174    /// until the next call to `writable` or `deinit`. The call requires the steady phase, and175    /// allocates nothing.176    pub fn poll(self: *Storage) Storage.Exhaustion!Poll {177        std.debug.assert(self.phase == .steady);178        std.debug.assert(self.start <= self.end);179        std.debug.assert(self.end <= self.bytes.len);180        if (self.terminal) return error.ReaderMessageCapacityExceeded;181        const buffered = self.bytes[self.start..self.end];182        if (std.mem.indexOfScalar(u8, buffered, '\n')) |newline| {183            self.observe(newline);184            const line = buffered[0..newline];185            self.start += newline + 1;186            return .{ .line = line };187        }188        self.observe(@min(buffered.len, self.capacity.message_bytes));189        if (buffered.len > self.capacity.message_bytes) return self.reject();190        if (self.eof) {191            if (buffered.len == 0) return .end;192            self.start = self.end;193            return .{ .line = buffered };194        }195        return .need_input;196    }197198    /// A caller uses this function to get the free space the caller's next read goes into. The199    /// function returns the free space at the end of the buffer. The call first moves the unread200    /// bytes to the front of the buffer, and lines returned earlier stop being valid. The caller201    /// calls it after `poll` returned `need_input`, and never after `finish` or once the reader is202    /// terminal. The space the function returns then holds at least one byte.203    pub fn writable(self: *Storage) []u8 {204        std.debug.assert(self.phase == .steady);205        std.debug.assert(!self.eof);206        std.debug.assert(!self.terminal);207        std.debug.assert(self.start <= self.end);208        std.debug.assert(self.end <= self.bytes.len);209        if (self.start != 0) {210            const remaining = self.end - self.start;211            std.mem.copyForwards(u8, self.bytes[0..remaining], self.bytes[self.start..self.end]);212            self.start = 0;213            self.end = remaining;214        }215        std.debug.assert(self.end < self.bytes.len);216        return self.bytes[self.end..];217    }218219    /// A caller uses this function to hand the bytes that arrived from its read to the reader. The220    /// function marks `count` bytes at the start of the space from `writable` as read in. The value221    /// `count` is above zero and at most the size of that space. A read that returns zero bytes222    /// means the end of input, and the caller calls `finish` for it.223    pub fn commit(self: *Storage, count: usize) void {224        std.debug.assert(self.phase == .steady);225        std.debug.assert(!self.eof);226        std.debug.assert(!self.terminal);227        std.debug.assert(count > 0);228        std.debug.assert(self.start <= self.end);229        std.debug.assert(count <= self.bytes.len - self.end);230        self.end += count;231        std.debug.assert(self.end <= self.bytes.len);232    }233234    /// A caller uses this function to tell the reader the input has ended. The function marks the235    /// end of input, so `poll` returns the remaining bytes as a last line and then `end`. The236    /// caller calls it once, and never once the reader is terminal.237    pub fn finish(self: *Storage) void {238        std.debug.assert(self.phase == .steady);239        std.debug.assert(!self.eof);240        std.debug.assert(!self.terminal);241        std.debug.assert(self.start <= self.end);242        self.eof = true;243    }244245    /// A caller uses this function to read the reader's sizes and counters at any time. The246    /// function returns a snapshot of the phase, the sizes and the counters. The call allocates247    /// nothing and changes nothing.248    pub fn status(self: *const Storage) Status {249        return .{250            .phase = self.phase,251            .message_bytes = self.capacity.message_bytes,252            .storage_bytes = self.capacity.storage_bytes,253            .buffered_bytes = self.end - self.start,254            .high_water_message_bytes = self.high_water_message_bytes,255            .rejected_message_count = self.rejected_message_count,256            .terminal = self.terminal,257        };258    }259260    /// A caller uses this function to free the buffer when the caller is done reading. The function261    /// frees the buffer with the allocator given to `init` and moves the reader to its teardown262    /// phase. The call is allowed before `activate`, as the client does when its setup fails. The263    /// caller calls it once.264    pub fn deinit(self: *Storage, allocator: std.mem.Allocator) void {265        std.debug.assert(self.phase != .teardown);266        std.debug.assert(self.bytes.len == self.capacity.storage_bytes);267        self.phase = .teardown;268        allocator.free(self.bytes);269        self.bytes = &.{};270        self.start = 0;271        self.end = 0;272    }273274    fn observe(self: *Storage, message_bytes: usize) void {275        std.debug.assert(message_bytes <= self.capacity.message_bytes);276        self.high_water_message_bytes = @max(self.high_water_message_bytes, message_bytes);277    }278279    fn reject(self: *Storage) Storage.Exhaustion {280        std.debug.assert(!self.terminal);281        std.debug.assert(self.end - self.start > self.capacity.message_bytes);282        self.terminal = true;283        self.rejected_message_count +|= 1;284        return error.ReaderMessageCapacityExceeded;285    }286};

1pub const ReaderStorage = reader.Storage;