lib/acp/src/reader/storage.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3 const capacity_mod = @import("capacity.zig");
4
5 /// A caller uses this error to tell an overlong line apart from other failures. The type defines
6 /// the one error the reader returns once it runs: `ReaderMessageCapacityExceeded`, for a line past
7 /// the limit. The package exports it as `acp.ReaderExhaustion`. The client returns it from `start`
8 /// and the prompt calls.
9 pub const Exhaustion = error{ReaderMessageCapacityExceeded};
10
11 /// A caller's read loop switches on this result to handle a line, read more input, or stop. The
12 /// union holds the result of one poll: a whole line, a request for more input, or the end of input.
13 /// The client's read loop and another program's server loop both switch on it.
14 pub const Poll = union(enum) {
15 /// Carries one line without its newline, as a slice of the reader's buffer. The slice also
16 /// holds the last bytes of the input when it ends without a newline. The reader splits at `\n`
17 /// alone and keeps a carriage return before it. The slice remains valid until the next call to
18 /// `writable`, which moves the unread bytes, or to `deinit`.
19 line: []const u8,
20 /// Signals that no whole line is buffered yet. The caller reads more input into `writable` and
21 /// passes the count to `commit`, or calls `finish` at the end of input.
22 need_input,
23 /// Signals that the input has ended and every byte has been returned. Later polls return it
24 /// again.
25 end,
26 };
27
28 /// A caller uses this snapshot to report how the reader is doing, or to explain why it stopped. The
29 /// struct records a snapshot of the reader's lifecycle step, sizes and counters. Both
30 /// `Storage.status` and `Client.readerStatus` return it. The package exports the type as
31 /// `acp.ReaderStatus`. The lifecycle step is initialization after `init`, steady after `activate`,
32 /// and teardown after `deinit`.
33 pub const Status = struct {
34 /// The reader's lifecycle step (phase) when the snapshot was taken.
35 phase: alloc_phase.capacity.Phase,
36 /// The line limit, in bytes.
37 message_bytes: usize,
38 /// The buffer size, the line limit plus one byte.
39 storage_bytes: usize,
40 /// The bytes read in but not yet returned as lines.
41 buffered_bytes: usize,
42 /// Records the longest line seen, capped at the limit (high-water mark), in bytes. The value
43 /// equals the limit once a line has run past it. A partial line still waiting for its newline
44 /// counts toward this size too, up to the limit.
45 high_water_message_bytes: usize,
46 /// Records the number of lines rejected for running past the limit. The count stops rising at
47 /// the largest `u64`. The first rejection leaves the reader stopped for good (terminal), so in
48 /// normal use the count is 0 or 1.
49 rejected_message_count: u64,
50 /// Reports true once a line has run past the limit, and every later poll fails.
51 terminal: bool,
52 };
53
54 /// A caller uses this type to read lines from a pipe or a file within a memory budget fixed in
55 /// advance, as the client does with the agent's output. The struct splits incoming bytes into lines
56 /// inside one buffer that `init` allocates once. The caller moves the bytes: it polls for a line,
57 /// and on `need_input` it reads into `writable` and passes the count to `commit`, or calls `finish`
58 /// at the end of input. After `activate`, nothing the reader does allocates. A line past the limit
59 /// leaves the reader terminal. The order of calls is `init`, `activate`, the steady calls, then
60 /// `deinit`, and each call asserts the phase it needs. The struct keeps no allocator: `init` and
61 /// `deinit` each take one. The package exports the type as `acp.ReaderStorage`.
62 pub const Storage = struct {
63 /// The reader's current lifecycle step.
64 phase: alloc_phase.capacity.Phase,
65 /// The line limit and buffer size worked out by `init`.
66 capacity: capacity_mod.Capacity,
67 /// The buffer, the buffer size long. The slice is empty after `deinit`.
68 bytes: []u8,
69 /// The index of the first byte not yet returned as a line. The index is 0 at the start and
70 /// after each round of moving the unread bytes to the front of the buffer (compaction).
71 start: usize = 0,
72 /// The index one past the last byte read in.
73 end: usize = 0,
74 /// Set by `finish` when no more input will come.
75 eof: bool = false,
76 /// Whether the reader has stopped for good, set when a line runs past the limit, after which
77 /// every `poll` fails with `error.ReaderMessageCapacityExceeded`.
78 terminal: bool = false,
79 /// The longest line seen, capped at the limit, in bytes.
80 high_water_message_bytes: usize = 0,
81 /// The number of lines rejected for running past the limit, which stops at the largest `u64`.
82 rejected_message_count: u64 = 0,
83
84 /// The limits type, `reader.Limits`.
85 pub const Limits: type = capacity_mod.Limits;
86 /// The capacity type, `reader.Capacity`.
87 pub const Capacity: type = capacity_mod.Capacity;
88 /// The error set of a running reader, `Exhaustion`.
89 pub const Exhaustion: type = @import("storage.zig").Exhaustion;
90 /// The errors `init` returns: `error.OutOfMemory` from the allocator and
91 /// `error.CapacityOverflow` from working out the buffer size. Another program in the repository
92 /// reuses it as its own `init` error set.
93 pub const InitError = std.mem.Allocator.Error || capacity_mod.DeriveError;
94
95 /// The repository's compile-time capacity checks read this declaration to hold the reader to
96 /// the memory it states. The declaration provides a compile-time record of the reader's memory,
97 /// with the id `acp.reader_storage`. It declares one region of the line limit plus one byte,
98 /// sized by the caller's limit and held for the reader's steady life. The record lists what the
99 /// region leaves out, in three groups. The first group covers threaded I/O, process pipes and
100 /// kernel buffers. The second group covers parsed JSON, protocol results, updates and
101 /// permission records. The third group covers outgoing messages and the capture of reply text.
102 /// At overload, a line one byte past the limit ends the reading before any JSON parsing or call
103 /// to the caller's callbacks for updates and permission decisions (observer). Splitting lines
104 /// and compaction allocate nothing after activation, and a test witnesses it. Ten promises
105 /// define the claim, each named by the test that witnesses it. The function `activate` seals
106 /// the reader and `deinit` tears it down. A compile-time check at the end of the file checks
107 /// the type's shape against the declaration. Each such test names its promise's key in a
108 /// `@stardustClaim` block.
109 pub const claim: alloc_phase.capacity.Declaration = .{
110 .source = .{
111 .id = "acp.reader_storage",
112 .kind = .phase_static,
113 .limit_source = .caller,
114 .storage = .{
115 .covered = &.{
116 .{
117 .id = "one_reusable_acp_newline_framing_and_message_region",
118 .lifetime = .steady,
119 .detail = "one reusable ACP newline-framing and message region",
120 },
121 },
122 .excluded = &.{
123 "threaded I/O, process pipes, and kernel buffering",
124 "parsed JSON, protocol results, updates, and permission ownership",
125 "outbound serialization and prompt capture",
126 },
127 },
128 .capacity = .{
129 .inputs = &.{
130 alloc_phase.capacity.bindInput(Limits, "message_bytes", "message_bytes"),
131 },
132 .type_selectors = &.{},
133 .nodes = &.{
134 .{ .input = 0 },
135 .{ .constant = 1 },
136 .{ .add = .{ .left = 0, .right = 1 } },
137 },
138 .assertions = &.{.{
139 .scope = .closure_total,
140 .measure = .retained,
141 .relation = .exact,
142 .expression = 2,
143 }},
144 },
145 .overload = .{
146 .kind = .terminal,
147 .detail = "max plus one byte ends framing before JSON parsing or observer publication",
148 },
149 .risks = .{
150 .transitive = .{
151 .status = .witnessed,
152 .detail = "fragmented framing and compaction allocate nothing after activation",
153 },
154 .foreign = .{
155 .status = .excluded,
156 .detail = "process transport and kernel buffering are outside reader storage",
157 },
158 },
159 .obligations = &.{
160 .{ .key = "acp_reader_capacity", .role = .capacity_model },
161 .{ .key = "acp_reader_acquisition", .role = .custom },
162 .{ .key = "acp_reader_oom", .role = .custom },
163 .{ .key = "acp_reader_boundary", .role = .overload },
164 .{ .key = "acp_reader_fragmented", .role = .custom },
165 .{ .key = "acp_reader_sealed", .role = .transitive_risk },
166 .{ .key = "acp_reader_client_terminal_overload", .role = .overload },
167 .{ .key = "acp_reader_client_terminal_foreign_risk", .role = .foreign_risk },
168 .{ .key = "acp_reader_admission", .role = .custom },
169 .{ .key = "acp_reader_root", .role = .custom },
170 },
171 },
172 .bindings = .{
173 .owner = @This(),
174 .seal = .{
175 .family = alloc_phase.capacity.selector(@This().activate),
176 .premise = .{
177 .class = .checked_semantic_fact,
178 .authority = .checker,
179 },
180 },
181 .teardown = .{
182 .family = alloc_phase.capacity.selector(@This().deinit),
183 .premise = .{
184 .class = .checked_semantic_fact,
185 .authority = .checker,
186 },
187 },
188 },
189 };
190
191 /// A caller uses this function to set up the reader, before the caller starts reading. The
192 /// function allocates the buffer, the line limit plus one byte, in one allocation. The call
193 /// returns the reader in its initialization phase, and the caller calls `activate` before the
194 /// first poll. The function fails with `error.CapacityOverflow` before it allocates when the
195 /// limit is too large to add one to, and with `error.OutOfMemory` when the allocation fails.
196 /// The function leaves nothing allocated when it fails. The caller frees the buffer with
197 /// `deinit` and the same allocator.
198 pub fn init(allocator: std.mem.Allocator, limits: Limits) InitError!Storage {
199 const capacity = try Capacity.derive(limits);
200 const bytes = try allocator.alloc(u8, capacity.storage_bytes);
201 const storage = Storage{
202 .phase = .initialization,
203 .capacity = capacity,
204 .bytes = bytes,
205 };
206 std.debug.assert(storage.bytes.len == storage.capacity.storage_bytes);
207 std.debug.assert(storage.capacity.storage_bytes > storage.capacity.message_bytes);
208 return storage;
209 }
210
211 /// A caller uses this function to mark the end of setup, after which the reader allocates
212 /// nothing. The call moves the reader from its initialization phase to its steady phase. The
213 /// caller calls it once, after `init` and before the first poll.
214 pub fn activate(self: *Storage) void {
215 std.debug.assert(self.phase == .initialization);
216 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
217 self.phase = .steady;
218 }
219
220 /// A caller's read loop calls this function first on each turn to take the next line. The
221 /// function returns the next whole line without its newline and moves past it. With no whole
222 /// line buffered and more than the limit waiting, the call fails with
223 /// `error.ReaderMessageCapacityExceeded` and the reader turns terminal. After `finish`, the
224 /// function returns the remaining bytes as a last line, then `end`. Otherwise the function
225 /// returns `need_input`. Once the reader is terminal, every call fails with the same error. A
226 /// line of exactly the limit passes, and one byte more fails. The returned line stays valid
227 /// until the next call to `writable` or `deinit`. The call requires the steady phase, and
228 /// allocates nothing.
229 pub fn poll(self: *Storage) Storage.Exhaustion!Poll {
230 std.debug.assert(self.phase == .steady);
231 std.debug.assert(self.start <= self.end);
232 std.debug.assert(self.end <= self.bytes.len);
233 if (self.terminal) return error.ReaderMessageCapacityExceeded;
234 const buffered = self.bytes[self.start..self.end];
235 if (std.mem.indexOfScalar(u8, buffered, '\n')) |newline| {
236 self.observe(newline);
237 const line = buffered[0..newline];
238 self.start += newline + 1;
239 return .{ .line = line };
240 }
241 self.observe(@min(buffered.len, self.capacity.message_bytes));
242 if (buffered.len > self.capacity.message_bytes) return self.reject();
243 if (self.eof) {
244 if (buffered.len == 0) return .end;
245 self.start = self.end;
246 return .{ .line = buffered };
247 }
248 return .need_input;
249 }
250
251 /// A caller uses this function to get the free space the caller's next read goes into. The
252 /// function returns the free space at the end of the buffer. The call first moves the unread
253 /// bytes to the front of the buffer, and lines returned earlier stop being valid. The caller
254 /// calls it after `poll` returned `need_input`, and never after `finish` or once the reader is
255 /// terminal. The space the function returns then holds at least one byte.
256 pub fn writable(self: *Storage) []u8 {
257 std.debug.assert(self.phase == .steady);
258 std.debug.assert(!self.eof);
259 std.debug.assert(!self.terminal);
260 std.debug.assert(self.start <= self.end);
261 std.debug.assert(self.end <= self.bytes.len);
262 if (self.start != 0) {
263 const remaining = self.end - self.start;
264 std.mem.copyForwards(u8, self.bytes[0..remaining], self.bytes[self.start..self.end]);
265 self.start = 0;
266 self.end = remaining;
267 }
268 std.debug.assert(self.end < self.bytes.len);
269 return self.bytes[self.end..];
270 }
271
272 /// A caller uses this function to hand the bytes that arrived from its read to the reader. The
273 /// function marks `count` bytes at the start of the space from `writable` as read in. The value
274 /// `count` is above zero and at most the size of that space. A read that returns zero bytes
275 /// means the end of input, and the caller calls `finish` for it.
276 pub fn commit(self: *Storage, count: usize) void {
277 std.debug.assert(self.phase == .steady);
278 std.debug.assert(!self.eof);
279 std.debug.assert(!self.terminal);
280 std.debug.assert(count > 0);
281 std.debug.assert(self.start <= self.end);
282 std.debug.assert(count <= self.bytes.len - self.end);
283 self.end += count;
284 std.debug.assert(self.end <= self.bytes.len);
285 }
286
287 /// A caller uses this function to tell the reader the input has ended. The function marks the
288 /// end of input, so `poll` returns the remaining bytes as a last line and then `end`. The
289 /// caller calls it once, and never once the reader is terminal.
290 pub fn finish(self: *Storage) void {
291 std.debug.assert(self.phase == .steady);
292 std.debug.assert(!self.eof);
293 std.debug.assert(!self.terminal);
294 std.debug.assert(self.start <= self.end);
295 self.eof = true;
296 }
297
298 /// A caller uses this function to read the reader's sizes and counters at any time. The
299 /// function returns a snapshot of the phase, the sizes and the counters. The call allocates
300 /// nothing and changes nothing.
301 pub fn status(self: *const Storage) Status {
302 return .{
303 .phase = self.phase,
304 .message_bytes = self.capacity.message_bytes,
305 .storage_bytes = self.capacity.storage_bytes,
306 .buffered_bytes = self.end - self.start,
307 .high_water_message_bytes = self.high_water_message_bytes,
308 .rejected_message_count = self.rejected_message_count,
309 .terminal = self.terminal,
310 };
311 }
312
313 /// A caller uses this function to free the buffer when the caller is done reading. The function
314 /// frees the buffer with the allocator given to `init` and moves the reader to its teardown
315 /// phase. The call is allowed before `activate`, as the client does when its setup fails. The
316 /// caller calls it once.
317 pub fn deinit(self: *Storage, allocator: std.mem.Allocator) void {
318 std.debug.assert(self.phase != .teardown);
319 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
320 self.phase = .teardown;
321 allocator.free(self.bytes);
322 self.bytes = &.{};
323 self.start = 0;
324 self.end = 0;
325 }
326
327 fn observe(self: *Storage, message_bytes: usize) void {
328 std.debug.assert(message_bytes <= self.capacity.message_bytes);
329 self.high_water_message_bytes = @max(self.high_water_message_bytes, message_bytes);
330 }
331
332 fn reject(self: *Storage) Storage.Exhaustion {
333 std.debug.assert(!self.terminal);
334 std.debug.assert(self.end - self.start > self.capacity.message_bytes);
335 self.terminal = true;
336 self.rejected_message_count +|= 1;
337 return error.ReaderMessageCapacityExceeded;
338 }
339 };
340
341 fn appendInput(storage: *Storage, input: []const u8) !void {
342 const writable = storage.writable();
343 if (input.len > writable.len) return error.TestInputTooLarge;
344 @memcpy(writable[0..input.len], input);
345 storage.commit(input.len);
346 }
347
348 fn checkInitFailures(allocator: std.mem.Allocator) !void {
349 var storage = try Storage.init(allocator, .{ .message_bytes = 9 });
350 storage.deinit(allocator);
351 }
352
353 test "ACP reader storage acquires one exact region" {
354 comptime {
355 @stardustClaim(
356 @import("alloc_phase").capacity.witness(Storage, "acp_reader_acquisition"),
357 null,
358 null,
359 null,
360 null,
361 null,
362 null,
363 );
364 }
365
366 var counting = std.testing.FailingAllocator.init(std.testing.allocator, .{});
367 const message_bytes = 128 * 1024;
368 var storage = try Storage.init(counting.allocator(), .{ .message_bytes = message_bytes });
369 defer storage.deinit(counting.allocator());
370
371 try std.testing.expectEqual(@as(usize, 1), counting.alloc_index);
372 try std.testing.expectEqual(@as(usize, message_bytes + 1), counting.allocated_bytes);
373 storage.activate();
374 try std.testing.expectEqual(
375 @intFromPtr(storage.bytes.ptr),
376 @intFromPtr(storage.writable().ptr),
377 );
378 }
379
380 test "ACP reader storage retries after every allocation failure" {
381 comptime {
382 @stardustClaim(
383 @import("alloc_phase").capacity.witness(Storage, "acp_reader_oom"),
384 null,
385 null,
386 null,
387 null,
388 null,
389 null,
390 );
391 }
392
393 try std.testing.checkAllAllocationFailures(std.testing.allocator, checkInitFailures, .{});
394 }
395
396 test "ACP reader accepts the exact limit and rejects max plus one" {
397 comptime {
398 @stardustClaim(
399 @import("alloc_phase").capacity.witness(Storage, "acp_reader_boundary"),
400 null,
401 null,
402 null,
403 null,
404 null,
405 null,
406 );
407 }
408
409 var storage = try Storage.init(std.testing.allocator, .{ .message_bytes = 5 });
410 defer storage.deinit(std.testing.allocator);
411 storage.activate();
412
413 try appendInput(&storage, "abcde\n");
414 const exact = (try storage.poll()).line;
415 try std.testing.expectEqualStrings("abcde", exact);
416 try appendInput(&storage, "abcdef");
417 try std.testing.expectError(error.ReaderMessageCapacityExceeded, storage.poll());
418 try std.testing.expectError(error.ReaderMessageCapacityExceeded, storage.poll());
419 try std.testing.expectEqual(Status{
420 .phase = .steady,
421 .message_bytes = 5,
422 .storage_bytes = 6,
423 .buffered_bytes = 6,
424 .high_water_message_bytes = 5,
425 .rejected_message_count = 1,
426 .terminal = true,
427 }, storage.status());
428 }
429
430 test "ACP reader rejection telemetry saturates" {
431 var storage = try Storage.init(std.testing.allocator, .{ .message_bytes = 0 });
432 defer storage.deinit(std.testing.allocator);
433 storage.activate();
434 storage.rejected_message_count = std.math.maxInt(u64);
435
436 try appendInput(&storage, "x");
437 try std.testing.expectError(error.ReaderMessageCapacityExceeded, storage.poll());
438 try std.testing.expectEqual(std.math.maxInt(u64), storage.status().rejected_message_count);
439 }
440
441 test "ACP reader frames fragmented and final messages" {
442 comptime {
443 @stardustClaim(
444 @import("alloc_phase").capacity.witness(Storage, "acp_reader_fragmented"),
445 null,
446 null,
447 null,
448 null,
449 null,
450 null,
451 );
452 }
453
454 var storage = try Storage.init(std.testing.allocator, .{ .message_bytes = 5 });
455 defer storage.deinit(std.testing.allocator);
456 storage.activate();
457
458 try appendInput(&storage, "ab");
459 try std.testing.expectEqual(Poll.need_input, try storage.poll());
460 try appendInput(&storage, "c\nx");
461 try std.testing.expectEqualStrings("abc", (try storage.poll()).line);
462 try std.testing.expectEqual(Poll.need_input, try storage.poll());
463 try appendInput(&storage, "y\n");
464 try std.testing.expectEqualStrings("xy", (try storage.poll()).line);
465 try appendInput(&storage, "final");
466 try std.testing.expectEqual(Poll.need_input, try storage.poll());
467 storage.finish();
468 try std.testing.expectEqualStrings("final", (try storage.poll()).line);
469 try std.testing.expectEqual(Poll.end, try storage.poll());
470 }
471
472 test "ACP reader framing remains allocation-free after storage seals" {
473 comptime {
474 @stardustClaim(
475 @import("alloc_phase").capacity.witness(Storage, "acp_reader_sealed"),
476 null,
477 null,
478 null,
479 null,
480 null,
481 null,
482 );
483 }
484
485 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
486 var maybe_storage: ?Storage = null;
487 defer {
488 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
489 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
490 if (maybe_storage) |*storage| {
491 if (storage.phase != .teardown) storage.deinit(phase_allocator.teardownAllocator());
492 }
493 phase_allocator.deinit();
494 }
495
496 maybe_storage = try Storage.init(
497 phase_allocator.initializationAllocator(),
498 .{ .message_bytes = 5 },
499 );
500 phase_allocator.seal();
501 const storage = &maybe_storage.?;
502 storage.activate();
503 try appendInput(storage, "one\n");
504 try std.testing.expectEqualStrings("one", (try storage.poll()).line);
505 try appendInput(storage, "two");
506 storage.finish();
507 try std.testing.expectEqualStrings("two", (try storage.poll()).line);
508 try std.testing.expectEqual(Poll.end, try storage.poll());
509 }
510
511 comptime {
512 alloc_phase.capacity.requireAllocatorRejectingOwnerShape(Storage);
513 }