const std = @import("std");const auto_classifier_context = @import("../permissions/auto_classifier_context.zig");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 types = @import("../shared/types.zig"); const Allocator = std.mem.Allocator;const schema_version: u64 = 7;const root_context_schema_version: u64 = 6;const permission_mode_schema_version: u64 = 5;const manager_epoch_schema_version: u64 = 4;const process_epoch_schema_version: u64 = 3;const legacy_schema_version: u64 = 2;const max_record_bytes: usize = 512 * 1024;const record_file = "control.json";const lock_file = "subagent-control.lock";const lock_deadline_ms: u64 = 2000; pub const Record = struct { child_id: []u8, generation: u64, parent_id: ?[]u8, mode: domain.Mode, configuration: domain.Configuration, state: domain.State, archived_from: ?domain.State = null, queue: []domain.QueuedMessage, events: []domain.Event, operations: []domain.OperationReceipt, next_event_sequence: u64, notification_cursor: u64, events_evicted_through: u64 = 0, queue_evicted: bool = false, 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, created_at_ms: i64, updated_at_ms: i64, pub fn deinit(self: *Record, alloc: Allocator) void { alloc.free(self.child_id); if (self.parent_id) |id| alloc.free(id); self.configuration.deinit(alloc); for (self.queue) |*message| message.deinit(alloc); alloc.free(self.queue); for (self.events) |*event| event.deinit(alloc); alloc.free(self.events); for (self.operations) |*operation| operation.deinit(alloc); alloc.free(self.operations); self.* = undefined; } /// Returns an owned copy; caller frees it with `deinit`. pub fn clone(self: Record, alloc: Allocator) !Record { const child_id = try alloc.dupe(u8, self.child_id); errdefer alloc.free(child_id); const parent_id = if (self.parent_id) |id| try alloc.dupe(u8, id) else null; errdefer if (parent_id) |id| alloc.free(id); var configuration = try self.configuration.clone(alloc); errdefer configuration.deinit(alloc); const queue = try cloneQueue(alloc, self.queue); errdefer freeQueue(alloc, queue); const events = try cloneEvents(alloc, self.events); errdefer freeEvents(alloc, events); return .{ .child_id = child_id, .generation = self.generation, .parent_id = parent_id, .mode = self.mode, .configuration = configuration, .state = self.state, .archived_from = self.archived_from, .queue = queue, .events = events, .operations = try cloneOperations(alloc, self.operations), .next_event_sequence = self.next_event_sequence, .notification_cursor = self.notification_cursor, .events_evicted_through = self.events_evicted_through, .queue_evicted = self.queue_evicted, .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, .created_at_ms = self.created_at_ms, .updated_at_ms = self.updated_at_ms, }; }}; pub const LoadError = error{ OutOfMemory, ControlNotFound, InvalidControlRecord, UnsupportedControlSchema, ControlRecordTooLarge, ControlPathUnsafe, PrivateStatePermissionsUnsupported, ControlStoreFailed,}; pub const SaveError = error{ OutOfMemory, ControlIdentityMismatch, ControlRecordTooLarge, ControlPathUnsafe, PrivateStatePermissionsUnsupported, ControlCommitIndeterminate, ControlStoreFailed,}; pub const LockError = error{ OutOfMemory, ControlLockBusy, ControlLockUnsupported, ControlPathUnsafe, PrivateStatePermissionsUnsupported, ControlStoreFailed,}; pub const Store = struct { capability: *session_child_store.SessionChildCapability, expected_child_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.ControlLockBusy, error.LockUnsupported => error.ControlLockUnsupported, error.SessionPathUnsafe => error.ControlPathUnsafe, error.PrivateStatePermissionsUnsupported => error.PrivateStatePermissionsUnsupported, else => error.ControlStoreFailed, }; } /// Returns an owned record, or null when no control record exists. 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.ControlPathUnsafe, error.PrivateStatePermissionsUnsupported => { return error.PrivateStatePermissionsUnsupported; }, else => return error.ControlStoreFailed, }; defer file.deinit(); const bytes = file.readToEnd(alloc, max_record_bytes) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.StreamTooLong => return error.ControlRecordTooLarge, else => return error.ControlStoreFailed, }; defer alloc.free(bytes); var record = try parseRecord(alloc, bytes); errdefer record.deinit(alloc); if (!std.mem.eql(u8, record.child_id, self.expected_child_id)) { return error.InvalidControlRecord; } return record; } /// Returns an owned record; caller frees it with `Record.deinit`. pub fn load(self: Store, alloc: Allocator) LoadError!Record { return (try self.loadOptional(alloc)) orelse error.ControlNotFound; } pub fn save(self: Store, alloc: Allocator, record: Record) SaveError!void { if (!std.mem.eql(u8, record.child_id, self.expected_child_id)) { return error.ControlIdentityMismatch; } var retained = record.clone(alloc) catch return error.OutOfMemory; defer retained.deinit(alloc); prepareForSave(alloc, &retained) catch |err| return switch (err) { error.OutOfMemory => error.OutOfMemory, error.ControlRecordTooLarge => error.ControlRecordTooLarge, }; validateRecordSemanticsForRecord(retained) catch return error.ControlRecordTooLarge; const bytes = renderRecord(alloc, retained) catch return error.OutOfMemory; defer alloc.free(bytes); std.debug.assert(bytes.len <= max_record_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.ControlPathUnsafe, error.PrivateStatePermissionsUnsupported => { return error.PrivateStatePermissionsUnsupported; }, error.SessionChildCommitIndeterminate => { return error.ControlCommitIndeterminate; }, else => return error.ControlStoreFailed, }; entry.deinit(alloc); }}; pub const PrepareError = error{ OutOfMemory, ControlRecordTooLarge }; /// Canonical-byte retention preparation. Active work is never selected for/// compaction; terminal persistent queue payloads, receipt epochs, and their/// now-unreferenced event prefix are reduced together until the record fits.pub fn prepareForSave(alloc: Allocator, record: *Record) PrepareError!void { return prepareForSaveLimit(alloc, record, max_record_bytes);} fn prepareForSaveLimit( alloc: Allocator, record: *Record, byte_limit: usize,) PrepareError!void { var terminal_queue_compacted = false; while (true) { const bytes = renderRecord(alloc, record.*) catch return error.OutOfMemory; const fits = bytes.len <= byte_limit; alloc.free(bytes); if (fits) return; if (!terminal_queue_compacted) { terminal_queue_compacted = true; if (try compactTerminalQueue(alloc, record)) { _ = try evictUnrequiredEventPrefix(alloc, record); continue; } } if (try evictOldestReceiptHorizon(alloc, record)) { _ = try evictUnrequiredEventPrefix(alloc, record); continue; } if (try evictUnrequiredEventPrefix(alloc, record)) continue; return error.ControlRecordTooLarge; }} fn compactTerminalQueue(alloc: Allocator, record: *Record) error{OutOfMemory}!bool { if (record.mode == .one_off) return false; var retained_count: usize = 0; for (record.queue) |message| { if (!isTerminalQueueStatus(message.status) or terminalNeedsRecovery(record.generation, record.events, message)) { retained_count += 1; } } if (retained_count == record.queue.len) return false; const retained = try alloc.alloc(domain.QueuedMessage, retained_count); var retained_index: usize = 0; for (record.queue) |*message| { if (isTerminalQueueStatus(message.status) and !terminalNeedsRecovery(record.generation, record.events, message.*)) { message.deinit(alloc); continue; } retained[retained_index] = message.*; retained_index += 1; } alloc.free(record.queue); record.queue = retained; record.queue_evicted = true; return true;} fn terminalNeedsRecovery( generation: u64, events: []const domain.Event, message: domain.QueuedMessage,) bool { var index = events.len; while (index != 0) { index -= 1; const event = events[index]; switch (event.kind) { .work_transition => |transition| { if (!std.mem.eql(u8, transition.work_item_id, message.id)) continue; return event.revision == generation; }, else => {}, } } return true;} fn evictOldestReceiptHorizon(alloc: Allocator, record: *Record) error{OutOfMemory}!bool { if (record.operations.len <= 1) return false; const evicted_index = oldestCommittedIssuanceReceiptIndex(record.operations); const evicted_identity = tool_result.parseBoundOperationId( record.operations[evicted_index].id, ); const retained = try alloc.alloc( domain.OperationReceipt, record.operations.len - 1, ); @memcpy(retained[0..evicted_index], record.operations[0..evicted_index]); @memcpy(retained[evicted_index..], record.operations[evicted_index + 1 ..]); record.operations[evicted_index].deinit(alloc); alloc.free(record.operations); record.operations = retained; if (evicted_identity) |identity| { if (identity.authority != .manager) { record.legacy_replay_closed = true; return true; } const next = identity.epoch +| 1; const source = identity.source; switch (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 oldestCommittedIssuanceReceiptIndex( operations: []const domain.OperationReceipt,) usize { var selected: ?usize = null; var selected_epoch: u64 = 0; for (operations, 0..) |operation, index| { const identity = tool_result.parseBoundOperationId(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 evictUnrequiredEventPrefix(alloc: Allocator, record: *Record) error{OutOfMemory}!bool { if (record.events.len == 0) return false; var first_required = record.next_event_sequence; for (record.operations) |operation| { first_required = @min(first_required, operation.event_sequence); } for (record.queue) |message| { for (record.events) |event| switch (event.kind) { .work_transition => |transition| { if (std.mem.eql(u8, transition.work_item_id, message.id)) { first_required = @min(first_required, event.sequence); break; } }, else => {}, }; } const first_retained_sequence = record.events[0].sequence; if (first_required <= first_retained_sequence) return false; const remove_u64 = @min( first_required - first_retained_sequence, @as(u64, @intCast(record.events.len)), ); const remove_count: usize = @intCast(remove_u64); if (remove_count == 0) return false; const retained = try alloc.alloc(domain.Event, record.events.len - remove_count); @memcpy(retained, record.events[remove_count..]); const evicted_through = record.events[remove_count - 1].sequence; for (record.events[0..remove_count]) |*event| event.deinit(alloc); record.events_evicted_through = evicted_through; record.notification_cursor = @max( record.notification_cursor, record.events_evicted_through, ); alloc.free(record.events); record.events = retained; return true;} fn isTerminalQueueStatus(status: domain.QueueStatus) bool { return status == .completed or status == .failed or status == .cancelled;} pub fn validateManagedRecord( alloc: Allocator, capability: *session_child_store.SessionChildCapability, expected_child_id: []const u8,) LoadError!void { var store = Store{ .capability = capability, .expected_child_id = expected_child_id, }; var record = (try store.loadOptional(alloc)) orelse return; record.deinit(alloc);} /// Returns an owned JSON record; caller frees it with `alloc.free`.fn renderRecord(alloc: Allocator, record: Record) ![]u8 { var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); const writer = &out.writer; try writer.print("{{\"schema_version\":{d},\"child_id\":", .{schema_version}); try writeJsonString(writer, record.child_id); try writer.print(",\"generation\":{d},\"parent_id\":", .{record.generation}); try writeOptionalString(writer, record.parent_id); try writer.writeAll(",\"mode\":"); try writeJsonString(writer, @tagName(record.mode)); try writer.writeAll(",\"configuration\":"); try renderConfiguration(writer, record.configuration); try writer.writeAll(",\"state\":"); try writeJsonString(writer, @tagName(record.state)); try writer.writeAll(",\"archived_from\":"); if (record.archived_from) |state| { try writeJsonString(writer, @tagName(state)); } else try writer.writeAll("null"); try writer.writeAll(",\"queue\":["); for (record.queue, 0..) |message, index| { if (index != 0) try writer.writeByte(','); try renderMessage(writer, message); } try writer.writeAll("],\"events\":["); for (record.events, 0..) |event, index| { if (index != 0) try writer.writeByte(','); try renderEvent(writer, event); } try writer.writeAll("],\"operations\":["); for (record.operations, 0..) |operation, index| { if (index != 0) try writer.writeByte(','); try renderOperation(writer, operation); } try writer.print( "],\"next_event_sequence\":{d},\"notification_cursor\":{d},\"events_evicted_through\":{d},\"queue_evicted\":{},\"legacy_replay_closed\":{},\"model_replay_floor\":{d},\"human_replay_floor\":{d},\"model_epoch_high\":{d},\"human_epoch_high\":{d},\"created_at_ms\":{d},\"updated_at_ms\":{d}}}", .{ record.next_event_sequence, record.notification_cursor, record.events_evicted_through, record.queue_evicted, record.legacy_replay_closed, record.model_replay_floor, record.human_replay_floor, record.model_epoch_high, record.human_epoch_high, record.created_at_ms, record.updated_at_ms, }, ); return out.toOwnedSlice();} fn renderConfiguration( writer: *std.Io.Writer, configuration: domain.Configuration,) !void { try writer.writeAll("{\"name\":"); try writeJsonString(writer, configuration.name); try writer.writeAll(",\"model\":"); try writeOptionalString(writer, configuration.model); try writer.writeAll(",\"effort\":"); if (configuration.effort) |effort| { try writeJsonString(writer, effort.label()); } else try writer.writeAll("null"); try writer.writeAll(",\"permission_mode\":"); try writeJsonString(writer, @tagName(configuration.permission_mode)); try writer.writeAll(",\"notifications\":"); try renderNotifications(writer, configuration.notifications); try writer.writeByte('}');} fn renderNotifications( writer: *std.Io.Writer, notifications: domain.NotificationPolicy,) !void { try writer.print( "{{\"terminal\":{{\"completed\":{},\"failed\":{},\"cancelled\":{}}},\"milestones\":[", .{ notifications.terminal.completed, notifications.terminal.failed, notifications.terminal.cancelled, }, ); for (notifications.milestones, 0..) |name, index| { if (index != 0) try writer.writeByte(','); try writeJsonString(writer, name); } try writer.writeAll("],\"report_interval_ms\":"); try writeOptionalU64(writer, notifications.report_interval_ms); try writer.writeAll(",\"report_duration_ms\":"); try writeOptionalU64(writer, notifications.report_duration_ms); try writer.writeAll(",\"stop_conditions\":["); for (notifications.stop_conditions, 0..) |condition, index| { if (index != 0) try writer.writeByte(','); try writeJsonString(writer, @tagName(condition));Lines 1–500 of 2780. Load more
const std = @import("std"); const auto_classifier_context = @import("../permissions/auto_classifier_context.zig"); 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 types = @import("../shared/types.zig"); const Allocator = std.mem.Allocator; const schema_version: u64 = 7; const root_context_schema_version: u64 = 6; const permission_mode_schema_version: u64 = 5; const manager_epoch_schema_version: u64 = 4; const process_epoch_schema_version: u64 = 3; const legacy_schema_version: u64 = 2; const max_record_bytes: usize = 512 * 1024; const record_file = "control.json"; const lock_file = "subagent-control.lock"; const lock_deadline_ms: u64 = 2000; pub const Record = struct { child_id: []u8, generation: u64, parent_id: ?[]u8, mode: domain.Mode, configuration: domain.Configuration, state: domain.State, archived_from: ?domain.State = null, queue: []domain.QueuedMessage, events: []domain.Event, operations: []domain.OperationReceipt, next_event_sequence: u64, notification_cursor: u64, events_evicted_through: u64 = 0, queue_evicted: bool = false, 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, created_at_ms: i64, updated_at_ms: i64, pub fn deinit(self: *Record, alloc: Allocator) void { alloc.free(self.child_id); if (self.parent_id) |id| alloc.free(id); self.configuration.deinit(alloc); for (self.queue) |*message| message.deinit(alloc); alloc.free(self.queue); for (self.events) |*event| event.deinit(alloc); alloc.free(self.events); for (self.operations) |*operation| operation.deinit(alloc); alloc.free(self.operations); self.* = undefined; } /// Returns an owned copy; caller frees it with `deinit`. pub fn clone(self: Record, alloc: Allocator) !Record { const child_id = try alloc.dupe(u8, self.child_id); errdefer alloc.free(child_id); const parent_id = if (self.parent_id) |id| try alloc.dupe(u8, id) else null; errdefer if (parent_id) |id| alloc.free(id); var configuration = try self.configuration.clone(alloc); errdefer configuration.deinit(alloc); const queue = try cloneQueue(alloc, self.queue); errdefer freeQueue(alloc, queue); const events = try cloneEvents(alloc, self.events); errdefer freeEvents(alloc, events); return .{ .child_id = child_id, .generation = self.generation, .parent_id = parent_id, .mode = self.mode, .configuration = configuration, .state = self.state, .archived_from = self.archived_from, .queue = queue, .events = events, .operations = try cloneOperations(alloc, self.operations), .next_event_sequence = self.next_event_sequence, .notification_cursor = self.notification_cursor, .events_evicted_through = self.events_evicted_through, .queue_evicted = self.queue_evicted, .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, .created_at_ms = self.created_at_ms, .updated_at_ms = self.updated_at_ms, }; } }; pub const LoadError = error{ OutOfMemory, ControlNotFound, InvalidControlRecord, UnsupportedControlSchema, ControlRecordTooLarge, ControlPathUnsafe, PrivateStatePermissionsUnsupported, ControlStoreFailed, }; pub const SaveError = error{ OutOfMemory, ControlIdentityMismatch, ControlRecordTooLarge, ControlPathUnsafe, PrivateStatePermissionsUnsupported, ControlCommitIndeterminate, ControlStoreFailed, }; pub const LockError = error{ OutOfMemory, ControlLockBusy, ControlLockUnsupported, ControlPathUnsafe, PrivateStatePermissionsUnsupported, ControlStoreFailed, }; pub const Store = struct { capability: *session_child_store.SessionChildCapability, expected_child_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.ControlLockBusy, error.LockUnsupported => error.ControlLockUnsupported, error.SessionPathUnsafe => error.ControlPathUnsafe, error.PrivateStatePermissionsUnsupported => error.PrivateStatePermissionsUnsupported, else => error.ControlStoreFailed, }; } /// Returns an owned record, or null when no control record exists. 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.ControlPathUnsafe, error.PrivateStatePermissionsUnsupported => { return error.PrivateStatePermissionsUnsupported; }, else => return error.ControlStoreFailed, }; defer file.deinit(); const bytes = file.readToEnd(alloc, max_record_bytes) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.StreamTooLong => return error.ControlRecordTooLarge, else => return error.ControlStoreFailed, }; defer alloc.free(bytes); var record = try parseRecord(alloc, bytes); errdefer record.deinit(alloc); if (!std.mem.eql(u8, record.child_id, self.expected_child_id)) { return error.InvalidControlRecord; } return record; } /// Returns an owned record; caller frees it with `Record.deinit`. pub fn load(self: Store, alloc: Allocator) LoadError!Record { return (try self.loadOptional(alloc)) orelse error.ControlNotFound; } pub fn save(self: Store, alloc: Allocator, record: Record) SaveError!void { if (!std.mem.eql(u8, record.child_id, self.expected_child_id)) { return error.ControlIdentityMismatch; } var retained = record.clone(alloc) catch return error.OutOfMemory; defer retained.deinit(alloc); prepareForSave(alloc, &retained) catch |err| return switch (err) { error.OutOfMemory => error.OutOfMemory, error.ControlRecordTooLarge => error.ControlRecordTooLarge, }; validateRecordSemanticsForRecord(retained) catch return error.ControlRecordTooLarge; const bytes = renderRecord(alloc, retained) catch return error.OutOfMemory; defer alloc.free(bytes); std.debug.assert(bytes.len <= max_record_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.ControlPathUnsafe, error.PrivateStatePermissionsUnsupported => { return error.PrivateStatePermissionsUnsupported; }, error.SessionChildCommitIndeterminate => { return error.ControlCommitIndeterminate; }, else => return error.ControlStoreFailed, }; entry.deinit(alloc); } }; pub const PrepareError = error{ OutOfMemory, ControlRecordTooLarge }; /// Canonical-byte retention preparation. Active work is never selected for /// compaction; terminal persistent queue payloads, receipt epochs, and their /// now-unreferenced event prefix are reduced together until the record fits. pub fn prepareForSave(alloc: Allocator, record: *Record) PrepareError!void { return prepareForSaveLimit(alloc, record, max_record_bytes); } fn prepareForSaveLimit( alloc: Allocator, record: *Record, byte_limit: usize, ) PrepareError!void { var terminal_queue_compacted = false; while (true) { const bytes = renderRecord(alloc, record.*) catch return error.OutOfMemory; const fits = bytes.len <= byte_limit; alloc.free(bytes); if (fits) return; if (!terminal_queue_compacted) { terminal_queue_compacted = true; if (try compactTerminalQueue(alloc, record)) { _ = try evictUnrequiredEventPrefix(alloc, record); continue; } } if (try evictOldestReceiptHorizon(alloc, record)) { _ = try evictUnrequiredEventPrefix(alloc, record); continue; } if (try evictUnrequiredEventPrefix(alloc, record)) continue; return error.ControlRecordTooLarge; } } fn compactTerminalQueue(alloc: Allocator, record: *Record) error{OutOfMemory}!bool { if (record.mode == .one_off) return false; var retained_count: usize = 0; for (record.queue) |message| { if (!isTerminalQueueStatus(message.status) or terminalNeedsRecovery(record.generation, record.events, message)) { retained_count += 1; } } if (retained_count == record.queue.len) return false; const retained = try alloc.alloc(domain.QueuedMessage, retained_count); var retained_index: usize = 0; for (record.queue) |*message| { if (isTerminalQueueStatus(message.status) and !terminalNeedsRecovery(record.generation, record.events, message.*)) { message.deinit(alloc); continue; } retained[retained_index] = message.*; retained_index += 1; } alloc.free(record.queue); record.queue = retained; record.queue_evicted = true; return true; } fn terminalNeedsRecovery( generation: u64, events: []const domain.Event, message: domain.QueuedMessage, ) bool { var index = events.len; while (index != 0) { index -= 1; const event = events[index]; switch (event.kind) { .work_transition => |transition| { if (!std.mem.eql(u8, transition.work_item_id, message.id)) continue; return event.revision == generation; }, else => {}, } } return true; } fn evictOldestReceiptHorizon(alloc: Allocator, record: *Record) error{OutOfMemory}!bool { if (record.operations.len <= 1) return false; const evicted_index = oldestCommittedIssuanceReceiptIndex(record.operations); const evicted_identity = tool_result.parseBoundOperationId( record.operations[evicted_index].id, ); const retained = try alloc.alloc( domain.OperationReceipt, record.operations.len - 1, ); @memcpy(retained[0..evicted_index], record.operations[0..evicted_index]); @memcpy(retained[evicted_index..], record.operations[evicted_index + 1 ..]); record.operations[evicted_index].deinit(alloc); alloc.free(record.operations); record.operations = retained; if (evicted_identity) |identity| { if (identity.authority != .manager) { record.legacy_replay_closed = true; return true; } const next = identity.epoch +| 1; const source = identity.source; switch (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 oldestCommittedIssuanceReceiptIndex( operations: []const domain.OperationReceipt, ) usize { var selected: ?usize = null; var selected_epoch: u64 = 0; for (operations, 0..) |operation, index| { const identity = tool_result.parseBoundOperationId(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 evictUnrequiredEventPrefix(alloc: Allocator, record: *Record) error{OutOfMemory}!bool { if (record.events.len == 0) return false; var first_required = record.next_event_sequence; for (record.operations) |operation| { first_required = @min(first_required, operation.event_sequence); } for (record.queue) |message| { for (record.events) |event| switch (event.kind) { .work_transition => |transition| { if (std.mem.eql(u8, transition.work_item_id, message.id)) { first_required = @min(first_required, event.sequence); break; } }, else => {}, }; } const first_retained_sequence = record.events[0].sequence; if (first_required <= first_retained_sequence) return false; const remove_u64 = @min( first_required - first_retained_sequence, @as(u64, @intCast(record.events.len)), ); const remove_count: usize = @intCast(remove_u64); if (remove_count == 0) return false; const retained = try alloc.alloc(domain.Event, record.events.len - remove_count); @memcpy(retained, record.events[remove_count..]); const evicted_through = record.events[remove_count - 1].sequence; for (record.events[0..remove_count]) |*event| event.deinit(alloc); record.events_evicted_through = evicted_through; record.notification_cursor = @max( record.notification_cursor, record.events_evicted_through, ); alloc.free(record.events); record.events = retained; return true; } fn isTerminalQueueStatus(status: domain.QueueStatus) bool { return status == .completed or status == .failed or status == .cancelled; } pub fn validateManagedRecord( alloc: Allocator, capability: *session_child_store.SessionChildCapability, expected_child_id: []const u8, ) LoadError!void { var store = Store{ .capability = capability, .expected_child_id = expected_child_id, }; var record = (try store.loadOptional(alloc)) orelse return; record.deinit(alloc); } /// Returns an owned JSON record; caller frees it with `alloc.free`. fn renderRecord(alloc: Allocator, record: Record) ![]u8 { var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); const writer = &out.writer; try writer.print("{{\"schema_version\":{d},\"child_id\":", .{schema_version}); try writeJsonString(writer, record.child_id); try writer.print(",\"generation\":{d},\"parent_id\":", .{record.generation}); try writeOptionalString(writer, record.parent_id); try writer.writeAll(",\"mode\":"); try writeJsonString(writer, @tagName(record.mode)); try writer.writeAll(",\"configuration\":"); try renderConfiguration(writer, record.configuration); try writer.writeAll(",\"state\":"); try writeJsonString(writer, @tagName(record.state)); try writer.writeAll(",\"archived_from\":"); if (record.archived_from) |state| { try writeJsonString(writer, @tagName(state)); } else try writer.writeAll("null"); try writer.writeAll(",\"queue\":["); for (record.queue, 0..) |message, index| { if (index != 0) try writer.writeByte(','); try renderMessage(writer, message); } try writer.writeAll("],\"events\":["); for (record.events, 0..) |event, index| { if (index != 0) try writer.writeByte(','); try renderEvent(writer, event); } try writer.writeAll("],\"operations\":["); for (record.operations, 0..) |operation, index| { if (index != 0) try writer.writeByte(','); try renderOperation(writer, operation); } try writer.print( "],\"next_event_sequence\":{d},\"notification_cursor\":{d},\"events_evicted_through\":{d},\"queue_evicted\":{},\"legacy_replay_closed\":{},\"model_replay_floor\":{d},\"human_replay_floor\":{d},\"model_epoch_high\":{d},\"human_epoch_high\":{d},\"created_at_ms\":{d},\"updated_at_ms\":{d}}}", .{ record.next_event_sequence, record.notification_cursor, record.events_evicted_through, record.queue_evicted, record.legacy_replay_closed, record.model_replay_floor, record.human_replay_floor, record.model_epoch_high, record.human_epoch_high, record.created_at_ms, record.updated_at_ms, }, ); return out.toOwnedSlice(); } fn renderConfiguration( writer: *std.Io.Writer, configuration: domain.Configuration, ) !void { try writer.writeAll("{\"name\":"); try writeJsonString(writer, configuration.name); try writer.writeAll(",\"model\":"); try writeOptionalString(writer, configuration.model); try writer.writeAll(",\"effort\":"); if (configuration.effort) |effort| { try writeJsonString(writer, effort.label()); } else try writer.writeAll("null"); try writer.writeAll(",\"permission_mode\":"); try writeJsonString(writer, @tagName(configuration.permission_mode)); try writer.writeAll(",\"notifications\":"); try renderNotifications(writer, configuration.notifications); try writer.writeByte('}'); } fn renderNotifications( writer: *std.Io.Writer, notifications: domain.NotificationPolicy, ) !void { try writer.print( "{{\"terminal\":{{\"completed\":{},\"failed\":{},\"cancelled\":{}}},\"milestones\":[", .{ notifications.terminal.completed, notifications.terminal.failed, notifications.terminal.cancelled, }, ); for (notifications.milestones, 0..) |name, index| { if (index != 0) try writer.writeByte(','); try writeJsonString(writer, name); } try writer.writeAll("],\"report_interval_ms\":"); try writeOptionalU64(writer, notifications.report_interval_ms); try writer.writeAll(",\"report_duration_ms\":"); try writeOptionalU64(writer, notifications.report_duration_ms); try writer.writeAll(",\"stop_conditions\":["); for (notifications.stop_conditions, 0..) |condition, index| { if (index != 0) try writer.writeByte(','); try writeJsonString(writer, @tagName(condition));