terminal/snapshot: add incremental decoder (#13569)

This adds a new `terminal.snapshot.Decoder` that allows for incremental
decoding of a snapshot stream. There are two methods: `ready` builds up
the entire terminal up to READY. Then `next` acts like a Zig iterator
and applies incremental history as it becomes available. In between
calls to `ready` and `next` the caller can do whatever.

The use case for this: with a 1MB ascii stream, the time to decode to
READY is ~40us on my machine, versus 1.5ms for the entire history. This
means that a terminal could be rendered and visible after 40us rather
than waiting for the full terminal. This isn't a large terminal, but
that READY time should be pretty standard since screens don't get that
big, but history is unbounded.

src/benchmark/TerminalSnapshot.zig

@@ -19,6 +19,10 @@
 //! * `decode` restores one complete snapshot per loop from bytes prepared
 //!   during setup. The restored terminal is destroyed inside the step, so
 //!   this measures the complete restore lifecycle.
+//! * `decode-ready` restores only the renderable prefix per loop using the
+//!   incremental decoder and stops at READY, skipping history. Compared
+//!   against `decode`, this is the time-to-interactive win of applying
+//!   history off the critical path.
 //! * `report` encodes once and prints total and per-record-tag sizes. It
 //!   is for inspecting the wire shape, not timing comparisons.
 //!
@@ -92,6 +96,9 @@ pub const Mode = enum {
     /// Restore one complete snapshot per loop, including its teardown.
     decode,
 
+    /// Restore one renderable READY prefix per loop, including teardown.
+    @"decode-ready",
+
     /// Print encoded sizes by record tag. Not a timing benchmark.
     report,
 };
@@ -122,6 +129,7 @@ pub fn benchmark(self: *TerminalSnapshot) Benchmark {
             .noop => stepNoop,
             .encode => stepEncode,
             .decode => stepDecode,
+            .@"decode-ready" => stepDecodeReady,
             .report => stepReport,
         },
         .setupFn = setup,
@@ -231,6 +239,28 @@ fn stepDecode(ptr: *anyopaque) Benchmark.Error!void {
     }
 }
 
+/// The renderable-prefix half of decode: everything through READY, leaving
+/// all history unread. The gap between this and `stepDecode` is the latency
+/// class the incremental decoder removes from time-to-interactive.
+fn stepDecodeReady(ptr: *anyopaque) Benchmark.Error!void {
+    const self: *TerminalSnapshot = @ptrCast(@alignCast(ptr));
+    const bytes = self.encoded.written();
+    for (0..self.opts.loops) |_| {
+        var reader: std.Io.Reader = .fixed(bytes);
+        var decoder: snapshot.Decoder = .init(&reader);
+        var decoded = decoder.ready(
+            self.alloc,
+            global.io(),
+            .{ .max_continuation_bytes = 1024 * 1024 },
+        ) catch |err| {
+            log.warn("snapshot READY decoding failed err={}", .{err});
+            return error.BenchmarkFailed;
+        };
+        std.mem.doNotOptimizeAway(&decoded);
+        decoded.deinit(self.alloc);
+    }
+}
+
 /// Print the encoded size grouped by record tag. This shares the encoder
 /// with encode mode but deliberately makes no timing claims.
 fn stepReport(ptr: *anyopaque) Benchmark.Error!void {
@@ -305,3 +335,16 @@ test "TerminalSnapshot decode round trip" {
     const bench = impl.benchmark();
     _ = try bench.run(.once);
 }
+
+test "TerminalSnapshot decode READY prefix" {
+    const testing = std.testing;
+    const impl: *TerminalSnapshot = try .create(testing.allocator, .{
+        .mode = .@"decode-ready",
+        .@"terminal-rows" = 4,
+        .@"terminal-cols" = 8,
+    });
+    defer impl.destroy(testing.allocator);
+
+    const bench = impl.benchmark();
+    _ = try bench.run(.once);
+}

src/terminal/snapshot/history.zig

@@ -193,6 +193,40 @@ pub const DecodeError = Decoder.InitError ||
         UnexpectedScreenKey,
     };
 
+/// Errors possible while restoring one history PAGE into a native Screen.
+pub const DecodePageError = Allocator.Error ||
+    page.DecodeError ||
+    TerminalPageList.PageAllocation.FinalizeError;
+
+/// Restore the next history PAGE record and prepend it to the native Screen.
+///
+/// Returns the number of rows added above the screen's existing content.
+/// The page is decoded directly into a detached PageList-pooled allocation
+/// and prepended only after its record validates, so a failure leaves the
+/// screen unchanged. A limit failure from `finalize` occurs after the record
+/// bytes were fully consumed, leaving `source` aligned on the next record.
+pub fn decodePage(
+    source: *std.Io.Reader,
+    alloc: Allocator,
+    terminal_screen: *TerminalScreen,
+) DecodePageError!usize {
+    // PAGE exposes its exact capacity before decoding the payload, allowing
+    // the destination PageList to allocate the final backing memory once.
+    var decoder: page.Decoder = undefined;
+    try decoder.init(source);
+    var allocation = try terminal_screen.pages.allocatePage(
+        decoder.capacity(),
+    );
+    defer allocation.deinit();
+    try decoder.decode(allocation.page(), alloc);
+
+    const rows = allocation.page().size.rows;
+    const contains_prompt = hasSemanticPrompt(allocation.page());
+    try allocation.finalize(.prepend);
+    if (contains_prompt) terminal_screen.semantic_prompt.seen = true;
+    return rows;
+}
+
 /// A decoded HISTORY manifest ready to restore its following PAGE records.
 ///
 /// Keeping manifest decoding separate lets the full snapshot wrapper route a
@@ -209,13 +243,10 @@ pub const Decoder = struct {
             UnexpectedRecordTag,
         };
 
-    pub const RestoreError = Allocator.Error ||
-        page.DecodeError ||
-        TerminalPageList.PageAllocation.FinalizeError ||
-        error{
-            /// The Screen already contains complete pages before its active page.
-            ExistingHistory,
-        };
+    pub const RestoreError = DecodePageError || error{
+        /// The Screen already contains complete pages before its active page.
+        ExistingHistory,
+    };
 
     /// Decode and finish the self-contained HISTORY manifest.
     pub fn init(self: *Decoder, source: *std.Io.Reader) InitError!void {
@@ -245,20 +276,7 @@ pub const Decoder = struct {
 
         // Native row totals remain derived from the actual PAGE dimensions.
         for (0..self.header.page_count) |_| {
-            // PAGE exposes its exact capacity before decoding the payload,
-            // allowing the destination PageList to allocate the final backing
-            // memory once.
-            var decoder: page.Decoder = undefined;
-            try decoder.init(self.source);
-            var allocation = try terminal_screen.pages.allocatePage(
-                decoder.capacity(),
-            );
-            defer allocation.deinit();
-            try decoder.decode(allocation.page(), alloc);
-
-            const contains_prompt = hasSemanticPrompt(allocation.page());
-            try allocation.finalize(.prepend);
-            if (contains_prompt) terminal_screen.semantic_prompt.seen = true;
+            _ = try decodePage(self.source, alloc, terminal_screen);
         }
 
         terminal_screen.pages.assertIntegrity();

src/terminal/snapshot/main.zig

@@ -10,13 +10,11 @@
 //! To do that, it sends the active terminal state followed by a READY record,
 //! then complete history.
 //!
-//! READY denotes that enough authenticated state is present to render the
-//! terminal and reconstruct its unfinished standard Stream state. The current
-//! synchronous decoder still returns only after FINISH. A caller moves the
-//! Terminal into final storage, replays CONTINUATION once, and only then applies
-//! PTY bytes belonging after the snapshot cut.
+//! READY denotes that enough state has been sent down to render the
+//! terminal and reconstruct its unfinished VT Stream state.
 //!
-//! After READY, we send history pages (scrollback).
+//! After READY, we send history pages (scrollback). Finally, the
+//! snapshot ends with a FINISH payload.
 //!
 //! ## Snapshot Format
 //!
@@ -132,6 +130,27 @@
 //! Use `snapshot.decodeExact` for a bounded file or buffer that must contain
 //! only one snapshot. It preserves the stricter end-of-file check, which may
 //! block when used with a live stream.
+//!
+//! `snapshot.Decoder` decodes the same stream incrementally so the terminal
+//! becomes usable at READY, before history has arrived. Each `next` call
+//! applies one history page to the by-then live terminal and returning null
+//! validates FINISH:
+//!
+//! ```zig
+//! var decoder: snapshot.Decoder = .init(&reader);
+//! var decoded = try decoder.ready(alloc, io, .{
+//!     .max_continuation_bytes = 1024 * 1024,
+//! });
+//! defer decoded.deinit(alloc);
+//!
+//! var terminal = decoded.toOwned();
+//! defer terminal.deinit(alloc);
+//!
+//! // Render, replay the continuation, process input...
+//! while (try decoder.next(alloc, &terminal)) |progress| {
+//!     _ = progress; // Scrollback grew.
+//! }
+//! ```
 
 pub const checkpoint = @import("checkpoint.zig");
 pub const continuation = @import("continuation.zig");
@@ -153,6 +172,7 @@ pub const Continuation = codec.Continuation;
 pub const EncodeOptions = codec.EncodeOptions;
 pub const DecodeOptions = codec.DecodeOptions;
 pub const Decoded = codec.Decoded;
+pub const Decoder = codec.Decoder;
 pub const encode = codec.encode;
 pub const decode = codec.decode;
 pub const decodeExact = codec.decodeExact;

src/terminal/snapshot/page.zig

@@ -277,6 +277,34 @@ pub const Decoder = struct {
     }
 };
 
+/// Errors possible while discarding one PAGE record without decoding it.
+pub const DiscardError = record.Reader.InitError ||
+    record.Reader.FinishError ||
+    std.Io.Reader.Error ||
+    error{
+        /// The next record is valid but is not a PAGE.
+        UnexpectedRecordTag,
+    };
+
+/// Consume exactly one complete PAGE record, validating its framing and
+/// CRC32C while discarding the payload without structural validation.
+///
+/// The payload bytes still stream through `source`, so an enclosing running
+/// digest (such as the FINISH checkpoint) continues to cover them. This lets
+/// a decoder stay aligned with the record sequence, and keep authenticating
+/// it, while dropping page content it can no longer apply.
+pub fn discard(source: *std.Io.Reader) DiscardError!void {
+    var record_reader: record.Reader = undefined;
+    try record_reader.init(source);
+    if (record_reader.header.tag != .page) {
+        return error.UnexpectedRecordTag;
+    }
+    try record_reader.payloadReader().discardAll(
+        record_reader.header.payload_len,
+    );
+    try record_reader.finish();
+}
+
 /// Encode a PAGE payload directly from a native page.
 fn encodePayload(
     page: *const TerminalPage,
@@ -1421,6 +1449,76 @@ test "decode reuses duplicate hyperlinks" {
     );
 }
 
+test "discard consumes exactly one PAGE record and keeps digest coverage" {
+    const testing = std.testing;
+
+    // Discard validates framing only, so an arbitrary payload keeps this test
+    // focused on record consumption rather than page structure.
+    var encoded: std.Io.Writer.Allocating = .init(testing.allocator);
+    defer encoded.deinit();
+    var stream: record.Writer = .init(testing.allocator, &encoded.writer);
+    defer stream.deinit();
+    {
+        const payload = stream.begin(.page);
+        errdefer stream.cancel();
+        try payload.writeAll("undecodable page payload");
+        try stream.finish();
+    }
+    const record_len = encoded.written().len;
+    try encoded.writer.writeAll("next");
+
+    // Discarded payload bytes must still update an enclosing running digest,
+    // which is what lets a snapshot FINISH checkpoint authenticate records
+    // whose content was dropped.
+    var source: std.Io.Reader = .fixed(encoded.written());
+    var hashed: record.StreamReader = .init(&source);
+    try discard(hashed.reader());
+    var expected_digest: record.PrefixDigest = undefined;
+    std.crypto.hash.Blake3.hash(
+        encoded.written()[0..record_len],
+        &expected_digest,
+        .{},
+    );
+    try testing.expectEqual(expected_digest, hashed.prefixDigest());
+    try testing.expectEqualStrings("next", try source.take(4));
+
+    // Only PAGE records may be discarded; the tag remains strict.
+    var wrong_tag: std.Io.Writer.Allocating = .init(testing.allocator);
+    defer wrong_tag.deinit();
+    var wrong_tag_stream: record.Writer = .init(
+        testing.allocator,
+        &wrong_tag.writer,
+    );
+    defer wrong_tag_stream.deinit();
+    {
+        const payload = wrong_tag_stream.begin(.history);
+        errdefer wrong_tag_stream.cancel();
+        try payload.writeAll("payload");
+        try wrong_tag_stream.finish();
+    }
+    var wrong_tag_source: std.Io.Reader = .fixed(wrong_tag.written());
+    try testing.expectError(
+        error.UnexpectedRecordTag,
+        discard(&wrong_tag_source),
+    );
+
+    // Discarded bytes are still covered by the record CRC.
+    const corrupted = try testing.allocator.dupe(
+        u8,
+        encoded.written()[0..record_len],
+    );
+    defer testing.allocator.free(corrupted);
+    corrupted[corrupted.len - 1] ^= 1;
+    var corrupted_source: std.Io.Reader = .fixed(corrupted);
+    try testing.expectError(error.InvalidChecksum, discard(&corrupted_source));
+
+    // A truncated payload is detected before `finish`.
+    var truncated_source: std.Io.Reader = .fixed(
+        encoded.written()[0 .. record_len - 1],
+    );
+    try testing.expectError(error.EndOfStream, discard(&truncated_source));
+}
+
 test "decode ignores empty hyperlink strings" {
     const header: Header = .{
         .columns = 1,