const std = @import("std");const command_output_content = @import("../tooling/command_output_content.zig");const io_mod = @import("../shared/io.zig");const types = @import("../shared/types.zig");const artifact_digest = @import("artifact_digest.zig");const session_child_store = @import("session_child_store.zig"); const Allocator = std.mem.Allocator;const Sha256 = std.crypto.hash.sha2.Sha256;const Stream = command_output_content.Stream;const replay_magic = "FXRPLY01";const frame_header_bytes: usize = 9;const max_frame_payload_bytes: usize = 1024 * 1024; const DecodedFrameHeader = struct {    stream: Stream,    payload_len: usize,}; fn decodeFrameHeader(header: []const u8) !DecodedFrameHeader {    std.debug.assert(header.len == frame_header_bytes);    const stream: Stream = switch (header[0]) {        0 => .stdout,        1 => .stderr,        else => return error.InvalidReplayStream,    };    const payload_len_u64 = std.mem.readInt(u64, header[1..][0..8], .little);    const payload_len = std.math.cast(usize, payload_len_u64) orelse        return error.ReplayFrameTooLarge;    if (payload_len == 0) return error.EmptyReplayFrame;    if (payload_len > max_frame_payload_bytes) return error.ReplayFrameTooLarge;    return .{ .stream = stream, .payload_len = payload_len };} const OpenSpool = struct {    file: session_child_store.ManagedFile,    handle: []u8,    framed_bytes: usize,}; const OwnedDescriptor = struct {    handle: []u8,    framed_bytes: usize,     fn view(self: OwnedDescriptor) types.CommandOutputReplayDescriptor {        return .{            .handle = self.handle,            .framed_bytes = self.framed_bytes,        };    }}; const CaptureState = union(enum) {    buffered: std.ArrayList(u8),    streaming: OpenSpool,    sealed_file: OwnedDescriptor,    unavailable,    retained: ?OwnedDescriptor,    consumed,}; /// Arena-lived single-owner handoff. The pointer may be copied with a tool/// result, but state transitions consume storage at most once.pub const Capture = struct {    inline_limit: usize,    comparison_limit: usize,    capability: ?*session_child_store.SessionChildCapability,    state: CaptureState = .{ .buffered = .empty },    had_output: bool = false,    sealed: bool = false,    content_hasher: Sha256 = Sha256.init(.{}),    digest_started: bool = false,     pub fn create(        alloc: Allocator,        inline_limit: usize,        capability: ?*session_child_store.SessionChildCapability,    ) !*Capture {        const capture = try alloc.create(Capture);        capture.* = .{            .inline_limit = inline_limit,            .comparison_limit = inline_limit,            .capability = capability,        };        return capture;    }     /// Records bytes only after the downstream callback accepted them.    /// Storage failures degrade replay without changing command execution.    pub fn appendAccepted(        self: *Capture,        alloc: Allocator,        stream: Stream,        bytes: []const u8,    ) void {        if (bytes.len == 0 or self.sealed) return;        self.had_output = true;        self.appendAcceptedFallible(alloc, stream, bytes) catch |err| {            self.makeUnavailable(alloc, err);            return;        };        if (!self.digest_started) {            self.content_hasher.update(replay_magic);            self.digest_started = true;        }        var header: [frame_header_bytes]u8 = undefined;        header[0] = @intFromEnum(stream);        std.mem.writeInt(u64, header[1..], @intCast(bytes.len), .little);        self.content_hasher.update(&header);        self.content_hasher.update(bytes);    }     fn appendAcceptedFallible(        self: *Capture,        alloc: Allocator,        stream: Stream,        bytes: []const u8,    ) !void {        var header: [frame_header_bytes]u8 = undefined;        header[0] = @intFromEnum(stream);        const payload_len = std.math.cast(u64, bytes.len) orelse            return error.ReplayFrameTooLarge;        std.mem.writeInt(u64, header[1..], payload_len, .little);         switch (self.state) {            .buffered => |*buffered| {                const base_len = if (buffered.items.len == 0)                    replay_magic.len                else                    buffered.items.len;                const with_header = try std.math.add(usize, base_len, header.len);                const required = try std.math.add(usize, with_header, bytes.len);                if (required <= self.inline_limit) {                    try buffered.ensureTotalCapacityPrecise(alloc, required);                    if (buffered.items.len == 0) buffered.appendSliceAssumeCapacity(replay_magic);                    buffered.appendSliceAssumeCapacity(&header);                    buffered.appendSliceAssumeCapacity(bytes);                    return;                }                try self.promoteAndAppend(alloc, buffered, &header, bytes);            },            .streaming => |*spool| {                try spool.file.writeAll(&header);                try spool.file.writeAll(bytes);                spool.framed_bytes = try std.math.add(                    usize,                    spool.framed_bytes,                    try std.math.add(usize, header.len, bytes.len),                );            },            .unavailable, .retained, .consumed, .sealed_file => {},        }    }     fn promoteAndAppend(        self: *Capture,        alloc: Allocator,        buffered: *std.ArrayList(u8),        header: []const u8,        bytes: []const u8,    ) !void {        const capability = self.capability orelse return error.ReplayStoreUnavailable;        var spool = try createSpool(alloc, capability);        var transferred = false;        errdefer if (!transferred) {            spool.file.deinit();            self.deleteSpool(spool.handle);            alloc.free(spool.handle);        };        if (buffered.items.len > 0) {            try spool.file.writeAll(buffered.items);            spool.framed_bytes = buffered.items.len;        } else {            try spool.file.writeAll(replay_magic);            spool.framed_bytes = replay_magic.len;        }        try spool.file.writeAll(header);        try spool.file.writeAll(bytes);        spool.framed_bytes = try std.math.add(            usize,            spool.framed_bytes,            try std.math.add(usize, header.len, bytes.len),        );        buffered.deinit(alloc);        self.state = .{ .streaming = spool };        transferred = true;    }     /// Closes and synchronizes any spool before the capture leaves tool runtime.    pub fn seal(self: *Capture, alloc: Allocator) void {        if (self.sealed) return;        self.sealed = true;        switch (self.state) {            .streaming => |*spool| {                spool.file.sync() catch |err| {                    self.makeUnavailable(alloc, err);                    return;                };                var digest: [Sha256.digest_length]u8 = undefined;                var hasher = self.content_hasher;                hasher.final(&digest);                self.contentAddressSpool(                    alloc,                    spool,                    digest,                ) catch |err| {                    self.makeUnavailable(alloc, err);                    return;                };                const descriptor = OwnedDescriptor{                    .handle = spool.handle,                    .framed_bytes = spool.framed_bytes,                };                spool.file.deinit();                self.state = .{ .sealed_file = descriptor };            },            else => {},        }    }     pub fn hasOutput(self: *const Capture) bool {        return self.had_output;    }     pub fn setComparisonLimit(self: *Capture, limit: usize) void {        self.comparison_limit = @max(self.inline_limit, limit);    }     pub fn comparisonLimit(self: *const Capture) usize {        return self.comparison_limit;    }     /// Returns a canonical comparison view only when accepted payload stays    /// within the route's bounded ordinary-result limit. Inline capture still    /// never exceeds `inline_limit`; a route may compare a larger private spool    /// when its executor can return more bytes in an ordinary exact result.    pub fn canonicalizeForComparison(        self: *Capture,        alloc: Allocator,    ) !?command_output_content.CanonicalOutput {        self.seal(alloc);        if (!self.had_output) {            var empty: command_output_content.CanonicalOutput = .{};            try empty.finish(alloc);            return empty;        }        return switch (self.state) {            .buffered => |*buffered| try canonicalizeFramedBytes(                alloc,                buffered.items,                self.comparison_limit,            ),            .sealed_file => |descriptor| try self.canonicalizeSpool(                alloc,                descriptor,                self.comparison_limit,            ),            else => null,        };    }     fn canonicalizeSpool(        self: *Capture,        alloc: Allocator,        descriptor: OwnedDescriptor,        max_payload_bytes: usize,    ) !command_output_content.CanonicalOutput {        const capability = self.capability orelse return error.ReplayStoreUnavailable;        var reader = try Reader.open(alloc, capability, descriptor.view());        defer reader.deinit();        var output: command_output_content.CanonicalOutput = .{};        errdefer output.deinit(alloc);        var payload_bytes: usize = 0;        while (try reader.next(alloc)) |frame| {            defer alloc.free(frame.payload);            payload_bytes = try std.math.add(usize, payload_bytes, frame.payload.len);            if (payload_bytes > max_payload_bytes) return error.ReplayComparisonTooLarge;            try output.append(alloc, frame.stream, frame.payload);        }        try output.finish(alloc);        return output;    }     fn canonicalizeFramedBytes(        alloc: Allocator,        bytes: []const u8,        max_payload_bytes: usize,    ) !command_output_content.CanonicalOutput {        if (bytes.len < replay_magic.len or            !std.mem.eql(u8, bytes[0..replay_magic.len], replay_magic))        {            return error.InvalidReplayHeader;        }        var output: command_output_content.CanonicalOutput = .{};        errdefer output.deinit(alloc);        var offset = replay_magic.len;        var payload_bytes: usize = 0;        while (offset < bytes.len) {            if (bytes.len - offset < frame_header_bytes) {                return error.TruncatedReplayFrame;            }            const header = try decodeFrameHeader(                bytes[offset..][0..frame_header_bytes],            );            const payload_start = try std.math.add(usize, offset, frame_header_bytes);            const next_offset = try std.math.add(                usize,                payload_start,                header.payload_len,            );            if (next_offset > bytes.len) return error.TruncatedReplayFrame;            payload_bytes = try std.math.add(                usize,                payload_bytes,                header.payload_len,            );            if (payload_bytes > max_payload_bytes) return error.ReplayComparisonTooLarge;            try output.append(                alloc,                header.stream,                bytes[payload_start..next_offset],            );            offset = next_offset;        }        try output.finish(alloc);        return output;    }     /// Stages a durable descriptor in result memory. The capture remains the    /// cleanup owner until the caller publishes that memory successfully.    pub fn retain(        self: *Capture,        alloc: Allocator,    ) ?types.CommandOutputReplay {        self.seal(alloc);        if (!self.had_output) {            self.abort(alloc);            return null;        }        return switch (self.state) {            .buffered => |*buffered| blk: {                const descriptor = self.materializeInline(alloc, buffered) catch |err| {                    self.makeUnavailable(alloc, err);                    self.state = .{ .retained = null };                    break :blk .unavailable;                };                self.state = .{ .retained = descriptor };                break :blk .{ .available = descriptor.view() };            },            .sealed_file => |descriptor| blk: {                self.state = .{ .retained = descriptor };                break :blk .{ .available = descriptor.view() };            },            .unavailable => blk: {                self.state = .{ .retained = null };                break :blk .unavailable;            },            .retained => |descriptor| if (descriptor) |value|                .{ .available = value.view() }            else                .unavailable,            .streaming => unreachable,            .consumed => null,        };    }     fn materializeInline(        self: *Capture,        alloc: Allocator,        buffered: *std.ArrayList(u8),    ) !OwnedDescriptor {        const capability = self.capability orelse return error.ReplayStoreUnavailable;        var spool = try createSpool(alloc, capability);        var transferred = false;        errdefer if (!transferred) {            spool.file.deinit();            self.deleteSpool(spool.handle);            alloc.free(spool.handle);        };        const framed_bytes = buffered.items.len;        try spool.file.writeAll(buffered.items);        try spool.file.sync();        var digest: [Sha256.digest_length]u8 = undefined;        Sha256.hash(buffered.items, &digest, .{});        try self.contentAddressSpool(alloc, &spool, digest);        spool.file.deinit();        buffered.deinit(alloc);        transferred = true;        return .{            .handle = spool.handle,            .framed_bytes = framed_bytes,        };    }     fn contentAddressSpool(        self: *Capture,        alloc: Allocator,        spool: *OpenSpool,        digest: [Sha256.digest_length]u8,    ) !void {        const capability = self.capability orelse            return error.ReplayStoreUnavailable;        const target_handle = try contentAddressedHandle(            alloc,            spool.handle,            digest,        );        capability.rename(            .command_artifacts,            spool.handle,            target_handle,        ) catch |err| {            if (capability.hasIndeterminateEntry(                .command_artifacts,                target_handle,            )) {                alloc.free(spool.handle);                spool.handle = target_handle;            } else {                alloc.free(target_handle);            }            return err;        };        alloc.free(spool.handle);        spool.handle = target_handle;    }     /// Deletes a capture before result memory is staged. Safe to invoke repeatedly.    pub fn abort(self: *Capture, alloc: Allocator) void {        switch (self.state) {            .buffered => |*buffered| buffered.deinit(alloc),            .streaming => |*spool| {                spool.file.deinit();                self.deleteSpool(spool.handle);                alloc.free(spool.handle);            },            .sealed_file => |descriptor| {                self.deleteSpool(descriptor.handle);                alloc.free(descriptor.handle);            },            .unavailable => {},            .retained => {},            .consumed => return,        }        self.state = .consumed;    }     /// Deletes a capture even when `retain` already staged its descriptor.    /// Callers use this to roll back a failed result-memory publication.    pub fn discard(self: *Capture, alloc: Allocator) void {        switch (self.state) {            .retained => |descriptor| {                if (descriptor) |value| {                    self.deleteSpool(value.handle);                    alloc.free(value.handle);                }                self.state = .consumed;            },            else => self.abort(alloc),        }    }     /// Releases retained descriptor storage after publication or external    /// rollback without deleting the session-owned sidecar.    pub fn releaseRetained(self: *Capture, alloc: Allocator) void {        switch (self.state) {            .retained => |descriptor| {                if (descriptor) |value| alloc.free(value.handle);                self.state = .consumed;            },            .consumed => {},            else => self.abort(alloc),        }    }     fn makeUnavailable(self: *Capture, alloc: Allocator, err: anyerror) void {        switch (self.state) {            .buffered => |*buffered| buffered.deinit(alloc),            .streaming => |*spool| {                spool.file.deinit();                self.deleteSpool(spool.handle);                alloc.free(spool.handle);            },            .sealed_file => |descriptor| {                self.deleteSpool(descriptor.handle);                alloc.free(descriptor.handle);            },            .unavailable, .retained, .consumed => {},        }        self.state = .unavailable;        @import("../shared/debug_trace.zig").logf(            "session",            "command replay capture unavailable err={s}",            .{@errorName(err)},        );    }     fn deleteSpool(self: *Capture, handle: []const u8) void {        const capability = self.capability orelse return;        capability.delete(.command_artifacts, handle) catch |err| {

Lines 1–500 of 971. Load more