const std = @import("std");const domain = @import("domain.zig");const io_mod = @import("../shared/io.zig");const session_child_store = @import("../session/session_child_store.zig");const tool_result = @import("tool_result.zig"); const Allocator = std.mem.Allocator;const schema_version: u64 = 3;const process_epoch_schema_version: u64 = 2;const legacy_schema_version: u64 = 1;const max_record_bytes: usize = 512 * 1024;const max_outstanding_operations: usize = 256;const record_file = "create-operations.json";const lock_file = "subagent-control.lock";const lock_deadline_ms: u64 = 2000; pub const Entry = struct { operation_id: []u8, request_fingerprint: [32]u8, child_id: []u8, fn deinit(self: *Entry, alloc: Allocator) void { alloc.free(self.operation_id); alloc.free(self.child_id); self.* = undefined; }}; pub const OutstandingOperation = struct { operation_id: []u8, fn deinit(self: *OutstandingOperation, alloc: Allocator) void { alloc.free(self.operation_id); self.* = undefined; }}; pub const IdentityResolution = enum { receipt, stable_failure, aborted_before_effect, pending_approval, retryable_failure, commit_indeterminate,}; pub const IdentityFinalization = enum { retire, retain,}; pub fn identityFinalization( resolution: IdentityResolution,) IdentityFinalization { return switch (resolution) { .receipt, .stable_failure, .aborted_before_effect, => .retire, .pending_approval, .retryable_failure, .commit_indeterminate, => .retain, };} pub const Record = struct { root_id: []u8, generation: u64, entries: []Entry, outstanding_operations: []OutstandingOperation, identity_epoch_high: u64 = 0, legacy_replay_closed: bool = false, model_replay_floor: u64 = 0, human_replay_floor: u64 = 0, model_epoch_high: u64 = 0, human_epoch_high: u64 = 0, pub fn init(alloc: Allocator, root_id: []const u8) !Record { const owned_root = try alloc.dupe(u8, root_id); errdefer alloc.free(owned_root); const entries = try alloc.alloc(Entry, 0); errdefer alloc.free(entries); return .{ .root_id = owned_root, .generation = 0, .entries = entries, .outstanding_operations = try alloc.alloc(OutstandingOperation, 0), }; } pub fn deinit(self: *Record, alloc: Allocator) void { alloc.free(self.root_id); for (self.entries) |*entry| entry.deinit(alloc); alloc.free(self.entries); for (self.outstanding_operations) |*operation| operation.deinit(alloc); alloc.free(self.outstanding_operations); self.* = undefined; } pub fn find(self: Record, operation_id: []const u8) ?*const Entry { for (self.entries) |*entry| { if (std.mem.eql(u8, entry.operation_id, operation_id)) return entry; } return null; } pub fn classify(self: Record, operation_id: []const u8) IdentityStatus { if (self.find(operation_id) != null) return .retained; const identity = tool_result.parseBoundOperationId(operation_id) orelse return if (self.legacy_replay_closed) .expired else .absent; if (identity.authority == .process_local) return .expired; if (self.hasOutstanding(operation_id)) return .absent; return .expired; } pub fn outstandingEpochForInvocation( self: Record, invocation_id: []const u8, source: domain.OperationIdentitySource, ) ?u64 { for (self.outstanding_operations) |operation| { if (!tool_result.boundOperationMatchesInvocation( operation.operation_id, invocation_id, source, )) continue; return tool_result.parseBoundOperationId(operation.operation_id).?.epoch; } return null; } pub fn hasOutstanding(self: Record, operation_id: []const u8) bool { for (self.outstanding_operations) |operation| { if (std.mem.eql(u8, operation.operation_id, operation_id)) return true; } return false; } pub fn reserveIdentity( self: *Record, alloc: Allocator, invocation_id: []const u8, source: domain.OperationIdentitySource, ) !u64 { if (self.outstandingEpochForInvocation(invocation_id, source)) |epoch| { return epoch; } if (self.outstanding_operations.len == max_outstanding_operations) { return error.TooManyOutstandingOperations; } const epoch = try std.math.add(u64, self.identity_epoch_high, 1); const operation_id = try tool_result.boundOperationIdAlloc( alloc, invocation_id, source, epoch, ); errdefer alloc.free(operation_id); const replacement = try alloc.alloc( OutstandingOperation, self.outstanding_operations.len + 1, ); @memcpy( replacement[0..self.outstanding_operations.len], self.outstanding_operations, ); replacement[self.outstanding_operations.len] = .{ .operation_id = operation_id, }; alloc.free(self.outstanding_operations); self.outstanding_operations = replacement; self.identity_epoch_high = epoch; self.generation = try std.math.add(u64, self.generation, 1); return epoch; } pub fn completeIdentity( self: *Record, alloc: Allocator, operation_id: []const u8, ) !bool { var remove_index: ?usize = null; for (self.outstanding_operations, 0..) |operation, index| { if (std.mem.eql(u8, operation.operation_id, operation_id)) { remove_index = index; break; } } const index = remove_index orelse return false; const replacement = try alloc.alloc( OutstandingOperation, self.outstanding_operations.len - 1, ); @memcpy(replacement[0..index], self.outstanding_operations[0..index]); @memcpy( replacement[index..], self.outstanding_operations[index + 1 ..], ); self.outstanding_operations[index].deinit(alloc); alloc.free(self.outstanding_operations); self.outstanding_operations = replacement; self.generation = try std.math.add(u64, self.generation, 1); return true; } pub fn finalizeIdentity( self: *Record, alloc: Allocator, operation_id: []const u8, finalization: IdentityFinalization, ) !bool { return switch (finalization) { .retain => false, .retire => self.completeIdentity(alloc, operation_id), }; } pub fn append( self: *Record, alloc: Allocator, operation_id: []const u8, request_fingerprint: [32]u8, child_id: []const u8, ) !void { const next_generation = try std.math.add(u64, self.generation, 1); const owned_operation = try alloc.dupe(u8, operation_id); errdefer alloc.free(owned_operation); const owned_child = try alloc.dupe(u8, child_id); errdefer alloc.free(owned_child); const replacement = try alloc.alloc(Entry, self.entries.len + 1); @memcpy(replacement[0..self.entries.len], self.entries); replacement[self.entries.len] = .{ .operation_id = owned_operation, .request_fingerprint = request_fingerprint, .child_id = owned_child, }; alloc.free(self.entries); self.entries = replacement; self.generation = next_generation; if (tool_result.parseBoundOperationId(operation_id)) |identity| { self.legacy_replay_closed = true; if (identity.authority != .manager) return; switch (identity.source) { .model => self.model_epoch_high = @max(self.model_epoch_high, identity.epoch), .human => self.human_epoch_high = @max(self.human_epoch_high, identity.epoch), } } } fn clone(self: Record, alloc: Allocator) !Record { const root_id = try alloc.dupe(u8, self.root_id); errdefer alloc.free(root_id); const entries = try alloc.alloc(Entry, self.entries.len); var initialized: usize = 0; errdefer { for (entries[0..initialized]) |*entry| entry.deinit(alloc); alloc.free(entries); } for (self.entries) |entry| { const operation_id = try alloc.dupe(u8, entry.operation_id); errdefer alloc.free(operation_id); entries[initialized] = .{ .operation_id = operation_id, .request_fingerprint = entry.request_fingerprint, .child_id = try alloc.dupe(u8, entry.child_id), }; initialized += 1; } const outstanding = try alloc.alloc( OutstandingOperation, self.outstanding_operations.len, ); var outstanding_initialized: usize = 0; errdefer { for (outstanding[0..outstanding_initialized]) |*operation| { operation.deinit(alloc); } alloc.free(outstanding); } for (self.outstanding_operations, outstanding) |operation, *copy| { copy.* = .{ .operation_id = try alloc.dupe(u8, operation.operation_id), }; outstanding_initialized += 1; } return .{ .root_id = root_id, .generation = self.generation, .entries = entries, .outstanding_operations = outstanding, .identity_epoch_high = self.identity_epoch_high, .legacy_replay_closed = self.legacy_replay_closed, .model_replay_floor = self.model_replay_floor, .human_replay_floor = self.human_replay_floor, .model_epoch_high = self.model_epoch_high, .human_epoch_high = self.human_epoch_high, }; }}; pub const IdentityStatus = enum { absent, retained, expired }; pub const LoadError = error{ OutOfMemory, RecordNotFound, InvalidRecord, UnsupportedSchema, RecordTooLarge, PathUnsafe, PrivateStatePermissionsUnsupported, StoreFailed,}; pub const SaveError = error{ OutOfMemory, IdentityMismatch, RecordTooLarge, PathUnsafe, PrivateStatePermissionsUnsupported, CommitIndeterminate, StoreFailed,}; pub const LockError = error{ OutOfMemory, LockBusy, LockUnsupported, PathUnsafe, PrivateStatePermissionsUnsupported, StoreFailed,}; pub const Store = struct { capability: *session_child_store.SessionChildCapability, expected_root_id: []const u8, pub fn acquireLock(self: Store) LockError!io_mod.TimedAdvisoryLock { return self.capability.acquireTimedAdvisoryLock( .subagent_control, lock_file, lock_deadline_ms, ) catch |err| switch (err) { error.OutOfMemory => error.OutOfMemory, error.LockBusy => error.LockBusy, error.LockUnsupported => error.LockUnsupported, error.SessionPathUnsafe => error.PathUnsafe, error.PrivateStatePermissionsUnsupported => error.PrivateStatePermissionsUnsupported, else => error.StoreFailed, }; } pub fn loadOptional(self: Store, alloc: Allocator) LoadError!?Record { var file = self.capability.openFileReadOnly( alloc, .subagent_control, record_file, ) catch |err| switch (err) { error.FileNotFound => return null, error.OutOfMemory => return error.OutOfMemory, error.SessionPathUnsafe => return error.PathUnsafe, error.PrivateStatePermissionsUnsupported => { return error.PrivateStatePermissionsUnsupported; }, else => return error.StoreFailed, }; defer file.deinit(); const bytes = file.readToEnd(alloc, max_record_bytes) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.StreamTooLong => return error.RecordTooLarge, else => return error.StoreFailed, }; defer alloc.free(bytes); var record = try parseRecord(alloc, bytes); errdefer record.deinit(alloc); if (!std.mem.eql(u8, record.root_id, self.expected_root_id)) { return error.InvalidRecord; } return record; } pub fn save(self: Store, alloc: Allocator, record: Record) SaveError!void { if (!std.mem.eql(u8, record.root_id, self.expected_root_id)) { return error.IdentityMismatch; } var retained = record.clone(alloc) catch return error.OutOfMemory; defer retained.deinit(alloc); const bytes = prepareEncoded(alloc, &retained) catch |err| return switch (err) { error.OutOfMemory => error.OutOfMemory, error.RecordTooLarge => error.RecordTooLarge, }; defer alloc.free(bytes); var entry = self.capability.atomicReplace( alloc, .subagent_control, record_file, bytes, ) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.SessionPathUnsafe => return error.PathUnsafe, error.PrivateStatePermissionsUnsupported => { return error.PrivateStatePermissionsUnsupported; }, error.SessionChildCommitIndeterminate => return error.CommitIndeterminate, else => return error.StoreFailed, }; entry.deinit(alloc); }}; fn prepareEncoded(alloc: Allocator, record: *Record) error{ OutOfMemory, RecordTooLarge }![]u8 { return prepareEncodedLimit(alloc, record, max_record_bytes);} fn prepareEncodedLimit( alloc: Allocator, record: *Record, byte_limit: usize,) error{ OutOfMemory, RecordTooLarge }![]u8 { while (true) { const bytes = renderRecord(alloc, record.*) catch return error.OutOfMemory; if (bytes.len <= byte_limit) return bytes; alloc.free(bytes); if (!try evictOldestHorizon(alloc, record)) return error.RecordTooLarge; }} fn evictOldestHorizon(alloc: Allocator, record: *Record) error{OutOfMemory}!bool { if (record.entries.len <= 1) return false; const evicted_index = oldestCommittedIssuanceIndex(record.entries); const evicted_identity = tool_result.parseBoundOperationId( record.entries[evicted_index].operation_id, ); const retained = try alloc.alloc(Entry, record.entries.len - 1); @memcpy(retained[0..evicted_index], record.entries[0..evicted_index]); @memcpy(retained[evicted_index..], record.entries[evicted_index + 1 ..]); record.entries[evicted_index].deinit(alloc); alloc.free(record.entries); record.entries = retained; if (evicted_identity) |identity| { if (identity.authority != .manager) { record.legacy_replay_closed = true; return true; } const next = identity.epoch +| 1; switch (identity.source) { .model => record.model_replay_floor = @max(record.model_replay_floor, next), .human => record.human_replay_floor = @max(record.human_replay_floor, next), } } else { record.legacy_replay_closed = true; } return true;} fn oldestCommittedIssuanceIndex(entries: []const Entry) usize { var selected: ?usize = null; var selected_epoch: u64 = 0; for (entries, 0..) |entry, index| { const identity = tool_result.parseBoundOperationId(entry.operation_id); if (identity == null or identity.?.authority == .process_local) return index; if (selected == null or identity.?.epoch < selected_epoch) { selected = index; selected_epoch = identity.?.epoch; } } return selected.?;} fn renderRecord(alloc: Allocator, record: Record) ![]u8 { var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); try out.writer.print("{{\"schema_version\":{d},\"root_id\":", .{schema_version}); try std.json.Stringify.value(record.root_id, .{}, &out.writer); try out.writer.print(",\"generation\":{d},\"entries\":[", .{record.generation}); for (record.entries, 0..) |entry, index| { if (index != 0) try out.writer.writeByte(','); try out.writer.writeAll("{\"operation_id\":"); try std.json.Stringify.value(entry.operation_id, .{}, &out.writer); try out.writer.writeAll(",\"request_fingerprint\":"); const fingerprint = std.fmt.bytesToHex(entry.request_fingerprint, .lower); try std.json.Stringify.value(fingerprint[0..], .{}, &out.writer); try out.writer.writeAll(",\"child_id\":"); try std.json.Stringify.value(entry.child_id, .{}, &out.writer); try out.writer.writeByte('}'); } try out.writer.writeAll("],\"outstanding_operations\":["); for (record.outstanding_operations, 0..) |operation, index| { if (index != 0) try out.writer.writeByte(','); try std.json.Stringify.value(operation.operation_id, .{}, &out.writer); } try out.writer.print( "],\"identity_epoch_high\":{d},\"legacy_replay_closed\":{},\"model_replay_floor\":{d},\"human_replay_floor\":{d},\"model_epoch_high\":{d},\"human_epoch_high\":{d}}}", .{ record.identity_epoch_high, record.legacy_replay_closed, record.model_replay_floor, record.human_replay_floor, record.model_epoch_high, record.human_epoch_high,Lines 1–500 of 1157. Load more
const std = @import("std"); const domain = @import("domain.zig"); const io_mod = @import("../shared/io.zig"); const session_child_store = @import("../session/session_child_store.zig"); const tool_result = @import("tool_result.zig"); const Allocator = std.mem.Allocator; const schema_version: u64 = 3; const process_epoch_schema_version: u64 = 2; const legacy_schema_version: u64 = 1; const max_record_bytes: usize = 512 * 1024; const max_outstanding_operations: usize = 256; const record_file = "create-operations.json"; const lock_file = "subagent-control.lock"; const lock_deadline_ms: u64 = 2000; pub const Entry = struct { operation_id: []u8, request_fingerprint: [32]u8, child_id: []u8, fn deinit(self: *Entry, alloc: Allocator) void { alloc.free(self.operation_id); alloc.free(self.child_id); self.* = undefined; } }; pub const OutstandingOperation = struct { operation_id: []u8, fn deinit(self: *OutstandingOperation, alloc: Allocator) void { alloc.free(self.operation_id); self.* = undefined; } }; pub const IdentityResolution = enum { receipt, stable_failure, aborted_before_effect, pending_approval, retryable_failure, commit_indeterminate, }; pub const IdentityFinalization = enum { retire, retain, }; pub fn identityFinalization( resolution: IdentityResolution, ) IdentityFinalization { return switch (resolution) { .receipt, .stable_failure, .aborted_before_effect, => .retire, .pending_approval, .retryable_failure, .commit_indeterminate, => .retain, }; } pub const Record = struct { root_id: []u8, generation: u64, entries: []Entry, outstanding_operations: []OutstandingOperation, identity_epoch_high: u64 = 0, legacy_replay_closed: bool = false, model_replay_floor: u64 = 0, human_replay_floor: u64 = 0, model_epoch_high: u64 = 0, human_epoch_high: u64 = 0, pub fn init(alloc: Allocator, root_id: []const u8) !Record { const owned_root = try alloc.dupe(u8, root_id); errdefer alloc.free(owned_root); const entries = try alloc.alloc(Entry, 0); errdefer alloc.free(entries); return .{ .root_id = owned_root, .generation = 0, .entries = entries, .outstanding_operations = try alloc.alloc(OutstandingOperation, 0), }; } pub fn deinit(self: *Record, alloc: Allocator) void { alloc.free(self.root_id); for (self.entries) |*entry| entry.deinit(alloc); alloc.free(self.entries); for (self.outstanding_operations) |*operation| operation.deinit(alloc); alloc.free(self.outstanding_operations); self.* = undefined; } pub fn find(self: Record, operation_id: []const u8) ?*const Entry { for (self.entries) |*entry| { if (std.mem.eql(u8, entry.operation_id, operation_id)) return entry; } return null; } pub fn classify(self: Record, operation_id: []const u8) IdentityStatus { if (self.find(operation_id) != null) return .retained; const identity = tool_result.parseBoundOperationId(operation_id) orelse return if (self.legacy_replay_closed) .expired else .absent; if (identity.authority == .process_local) return .expired; if (self.hasOutstanding(operation_id)) return .absent; return .expired; } pub fn outstandingEpochForInvocation( self: Record, invocation_id: []const u8, source: domain.OperationIdentitySource, ) ?u64 { for (self.outstanding_operations) |operation| { if (!tool_result.boundOperationMatchesInvocation( operation.operation_id, invocation_id, source, )) continue; return tool_result.parseBoundOperationId(operation.operation_id).?.epoch; } return null; } pub fn hasOutstanding(self: Record, operation_id: []const u8) bool { for (self.outstanding_operations) |operation| { if (std.mem.eql(u8, operation.operation_id, operation_id)) return true; } return false; } pub fn reserveIdentity( self: *Record, alloc: Allocator, invocation_id: []const u8, source: domain.OperationIdentitySource, ) !u64 { if (self.outstandingEpochForInvocation(invocation_id, source)) |epoch| { return epoch; } if (self.outstanding_operations.len == max_outstanding_operations) { return error.TooManyOutstandingOperations; } const epoch = try std.math.add(u64, self.identity_epoch_high, 1); const operation_id = try tool_result.boundOperationIdAlloc( alloc, invocation_id, source, epoch, ); errdefer alloc.free(operation_id); const replacement = try alloc.alloc( OutstandingOperation, self.outstanding_operations.len + 1, ); @memcpy( replacement[0..self.outstanding_operations.len], self.outstanding_operations, ); replacement[self.outstanding_operations.len] = .{ .operation_id = operation_id, }; alloc.free(self.outstanding_operations); self.outstanding_operations = replacement; self.identity_epoch_high = epoch; self.generation = try std.math.add(u64, self.generation, 1); return epoch; } pub fn completeIdentity( self: *Record, alloc: Allocator, operation_id: []const u8, ) !bool { var remove_index: ?usize = null; for (self.outstanding_operations, 0..) |operation, index| { if (std.mem.eql(u8, operation.operation_id, operation_id)) { remove_index = index; break; } } const index = remove_index orelse return false; const replacement = try alloc.alloc( OutstandingOperation, self.outstanding_operations.len - 1, ); @memcpy(replacement[0..index], self.outstanding_operations[0..index]); @memcpy( replacement[index..], self.outstanding_operations[index + 1 ..], ); self.outstanding_operations[index].deinit(alloc); alloc.free(self.outstanding_operations); self.outstanding_operations = replacement; self.generation = try std.math.add(u64, self.generation, 1); return true; } pub fn finalizeIdentity( self: *Record, alloc: Allocator, operation_id: []const u8, finalization: IdentityFinalization, ) !bool { return switch (finalization) { .retain => false, .retire => self.completeIdentity(alloc, operation_id), }; } pub fn append( self: *Record, alloc: Allocator, operation_id: []const u8, request_fingerprint: [32]u8, child_id: []const u8, ) !void { const next_generation = try std.math.add(u64, self.generation, 1); const owned_operation = try alloc.dupe(u8, operation_id); errdefer alloc.free(owned_operation); const owned_child = try alloc.dupe(u8, child_id); errdefer alloc.free(owned_child); const replacement = try alloc.alloc(Entry, self.entries.len + 1); @memcpy(replacement[0..self.entries.len], self.entries); replacement[self.entries.len] = .{ .operation_id = owned_operation, .request_fingerprint = request_fingerprint, .child_id = owned_child, }; alloc.free(self.entries); self.entries = replacement; self.generation = next_generation; if (tool_result.parseBoundOperationId(operation_id)) |identity| { self.legacy_replay_closed = true; if (identity.authority != .manager) return; switch (identity.source) { .model => self.model_epoch_high = @max(self.model_epoch_high, identity.epoch), .human => self.human_epoch_high = @max(self.human_epoch_high, identity.epoch), } } } fn clone(self: Record, alloc: Allocator) !Record { const root_id = try alloc.dupe(u8, self.root_id); errdefer alloc.free(root_id); const entries = try alloc.alloc(Entry, self.entries.len); var initialized: usize = 0; errdefer { for (entries[0..initialized]) |*entry| entry.deinit(alloc); alloc.free(entries); } for (self.entries) |entry| { const operation_id = try alloc.dupe(u8, entry.operation_id); errdefer alloc.free(operation_id); entries[initialized] = .{ .operation_id = operation_id, .request_fingerprint = entry.request_fingerprint, .child_id = try alloc.dupe(u8, entry.child_id), }; initialized += 1; } const outstanding = try alloc.alloc( OutstandingOperation, self.outstanding_operations.len, ); var outstanding_initialized: usize = 0; errdefer { for (outstanding[0..outstanding_initialized]) |*operation| { operation.deinit(alloc); } alloc.free(outstanding); } for (self.outstanding_operations, outstanding) |operation, *copy| { copy.* = .{ .operation_id = try alloc.dupe(u8, operation.operation_id), }; outstanding_initialized += 1; } return .{ .root_id = root_id, .generation = self.generation, .entries = entries, .outstanding_operations = outstanding, .identity_epoch_high = self.identity_epoch_high, .legacy_replay_closed = self.legacy_replay_closed, .model_replay_floor = self.model_replay_floor, .human_replay_floor = self.human_replay_floor, .model_epoch_high = self.model_epoch_high, .human_epoch_high = self.human_epoch_high, }; } }; pub const IdentityStatus = enum { absent, retained, expired }; pub const LoadError = error{ OutOfMemory, RecordNotFound, InvalidRecord, UnsupportedSchema, RecordTooLarge, PathUnsafe, PrivateStatePermissionsUnsupported, StoreFailed, }; pub const SaveError = error{ OutOfMemory, IdentityMismatch, RecordTooLarge, PathUnsafe, PrivateStatePermissionsUnsupported, CommitIndeterminate, StoreFailed, }; pub const LockError = error{ OutOfMemory, LockBusy, LockUnsupported, PathUnsafe, PrivateStatePermissionsUnsupported, StoreFailed, }; pub const Store = struct { capability: *session_child_store.SessionChildCapability, expected_root_id: []const u8, pub fn acquireLock(self: Store) LockError!io_mod.TimedAdvisoryLock { return self.capability.acquireTimedAdvisoryLock( .subagent_control, lock_file, lock_deadline_ms, ) catch |err| switch (err) { error.OutOfMemory => error.OutOfMemory, error.LockBusy => error.LockBusy, error.LockUnsupported => error.LockUnsupported, error.SessionPathUnsafe => error.PathUnsafe, error.PrivateStatePermissionsUnsupported => error.PrivateStatePermissionsUnsupported, else => error.StoreFailed, }; } pub fn loadOptional(self: Store, alloc: Allocator) LoadError!?Record { var file = self.capability.openFileReadOnly( alloc, .subagent_control, record_file, ) catch |err| switch (err) { error.FileNotFound => return null, error.OutOfMemory => return error.OutOfMemory, error.SessionPathUnsafe => return error.PathUnsafe, error.PrivateStatePermissionsUnsupported => { return error.PrivateStatePermissionsUnsupported; }, else => return error.StoreFailed, }; defer file.deinit(); const bytes = file.readToEnd(alloc, max_record_bytes) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.StreamTooLong => return error.RecordTooLarge, else => return error.StoreFailed, }; defer alloc.free(bytes); var record = try parseRecord(alloc, bytes); errdefer record.deinit(alloc); if (!std.mem.eql(u8, record.root_id, self.expected_root_id)) { return error.InvalidRecord; } return record; } pub fn save(self: Store, alloc: Allocator, record: Record) SaveError!void { if (!std.mem.eql(u8, record.root_id, self.expected_root_id)) { return error.IdentityMismatch; } var retained = record.clone(alloc) catch return error.OutOfMemory; defer retained.deinit(alloc); const bytes = prepareEncoded(alloc, &retained) catch |err| return switch (err) { error.OutOfMemory => error.OutOfMemory, error.RecordTooLarge => error.RecordTooLarge, }; defer alloc.free(bytes); var entry = self.capability.atomicReplace( alloc, .subagent_control, record_file, bytes, ) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.SessionPathUnsafe => return error.PathUnsafe, error.PrivateStatePermissionsUnsupported => { return error.PrivateStatePermissionsUnsupported; }, error.SessionChildCommitIndeterminate => return error.CommitIndeterminate, else => return error.StoreFailed, }; entry.deinit(alloc); } }; fn prepareEncoded(alloc: Allocator, record: *Record) error{ OutOfMemory, RecordTooLarge }![]u8 { return prepareEncodedLimit(alloc, record, max_record_bytes); } fn prepareEncodedLimit( alloc: Allocator, record: *Record, byte_limit: usize, ) error{ OutOfMemory, RecordTooLarge }![]u8 { while (true) { const bytes = renderRecord(alloc, record.*) catch return error.OutOfMemory; if (bytes.len <= byte_limit) return bytes; alloc.free(bytes); if (!try evictOldestHorizon(alloc, record)) return error.RecordTooLarge; } } fn evictOldestHorizon(alloc: Allocator, record: *Record) error{OutOfMemory}!bool { if (record.entries.len <= 1) return false; const evicted_index = oldestCommittedIssuanceIndex(record.entries); const evicted_identity = tool_result.parseBoundOperationId( record.entries[evicted_index].operation_id, ); const retained = try alloc.alloc(Entry, record.entries.len - 1); @memcpy(retained[0..evicted_index], record.entries[0..evicted_index]); @memcpy(retained[evicted_index..], record.entries[evicted_index + 1 ..]); record.entries[evicted_index].deinit(alloc); alloc.free(record.entries); record.entries = retained; if (evicted_identity) |identity| { if (identity.authority != .manager) { record.legacy_replay_closed = true; return true; } const next = identity.epoch +| 1; switch (identity.source) { .model => record.model_replay_floor = @max(record.model_replay_floor, next), .human => record.human_replay_floor = @max(record.human_replay_floor, next), } } else { record.legacy_replay_closed = true; } return true; } fn oldestCommittedIssuanceIndex(entries: []const Entry) usize { var selected: ?usize = null; var selected_epoch: u64 = 0; for (entries, 0..) |entry, index| { const identity = tool_result.parseBoundOperationId(entry.operation_id); if (identity == null or identity.?.authority == .process_local) return index; if (selected == null or identity.?.epoch < selected_epoch) { selected = index; selected_epoch = identity.?.epoch; } } return selected.?; } fn renderRecord(alloc: Allocator, record: Record) ![]u8 { var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); try out.writer.print("{{\"schema_version\":{d},\"root_id\":", .{schema_version}); try std.json.Stringify.value(record.root_id, .{}, &out.writer); try out.writer.print(",\"generation\":{d},\"entries\":[", .{record.generation}); for (record.entries, 0..) |entry, index| { if (index != 0) try out.writer.writeByte(','); try out.writer.writeAll("{\"operation_id\":"); try std.json.Stringify.value(entry.operation_id, .{}, &out.writer); try out.writer.writeAll(",\"request_fingerprint\":"); const fingerprint = std.fmt.bytesToHex(entry.request_fingerprint, .lower); try std.json.Stringify.value(fingerprint[0..], .{}, &out.writer); try out.writer.writeAll(",\"child_id\":"); try std.json.Stringify.value(entry.child_id, .{}, &out.writer); try out.writer.writeByte('}'); } try out.writer.writeAll("],\"outstanding_operations\":["); for (record.outstanding_operations, 0..) |operation, index| { if (index != 0) try out.writer.writeByte(','); try std.json.Stringify.value(operation.operation_id, .{}, &out.writer); } try out.writer.print( "],\"identity_epoch_high\":{d},\"legacy_replay_closed\":{},\"model_replay_floor\":{d},\"human_replay_floor\":{d},\"model_epoch_high\":{d},\"human_epoch_high\":{d}}}", .{ record.identity_epoch_high, record.legacy_replay_closed, record.model_replay_floor, record.human_replay_floor, record.model_epoch_high, record.human_epoch_high,