const std = @import("std");const builtin = @import("builtin");const approval_persistence = @import("approval_persistence.zig");const approval_registry = @import("approval_registry.zig");const authority = @import("authority.zig");const auto_classifier_context = @import("../permissions/auto_classifier_context.zig");const communication = @import("communication.zig");const communication_manager = @import("communication_manager.zig");const create_store = @import("create_store.zig");const communication_store = @import("communication_store.zig");const control_store = @import("control_store.zig");const domain = @import("domain.zig");const execution = @import("execution.zig");const manager_mod = @import("manager.zig");const tool_result = @import("tool_result.zig");const debug_trace = @import("../shared/debug_trace.zig");const io_mod = @import("../shared/io.zig");const session = @import("../session/session.zig");const session_child_store = @import("../session/session_child_store.zig");const session_codec = @import("../session/session_codec.zig");const session_store = @import("../session/session_store.zig");const mcp_access = @import("../mcp/access_policy.zig");const mode_registry = @import("../modes/mode_registry.zig");const permissions = @import("../permissions/permissions.zig");const session_permission_state = @import("../permissions/session_permission_state.zig");const tool_dispatch = @import("../tooling/tool_dispatch.zig");const tool_set_contract = @import("../tooling/tool_set.zig");const types = @import("../shared/types.zig"); const Allocator = std.mem.Allocator;const inspect_wait_external_poll_ms: i64 = 100; const TargetAuthorizationTestHook = struct { context: ?*anyopaque = null, run_fn: *const fn (?*anyopaque) void,}; const TestHooks = struct { var after_target_authorization: ?TargetAuthorizationTestHook = null;}; fn runAfterTargetAuthorizationTestHook() void { if (comptime builtin.is_test) { const hook = TestHooks.after_target_authorization orelse return; hook.run_fn(hook.context); }} pub const RecoveryState = enum(u8) { pending, scheduled, running, deferred, complete,}; const RecoveryTrigger = enum { automatic, explicit,}; const RecoveryStartDecision = enum { schedule, start, wait, no_effect,}; const RecoveryFinishOutcome = enum { fully_reconciled, incomplete, failed,}; fn decideRecoveryStart( state: RecoveryState, trigger: RecoveryTrigger,) RecoveryStartDecision { return switch (state) { .pending => switch (trigger) { .automatic => .schedule, .explicit => .start, }, .scheduled, .running => switch (trigger) { .automatic => .no_effect, .explicit => .wait, }, .deferred => switch (trigger) { .automatic => .no_effect, .explicit => .start, }, .complete => .no_effect, };} fn recoveryStateAfterFinish(outcome: RecoveryFinishOutcome) RecoveryState { return switch (outcome) { .fully_reconciled => .complete, .incomplete, .failed => .deferred, };} pub const BackgroundRecoveryError = error{ThreadSpawnFailed}; pub const Defaults = struct { model: []const u8, effort: types.ReasoningEffort, fast_mode: bool = false, conversation_language: session.ConversationLanguage,}; pub const ChildRunner = struct { context: ?*anyopaque = null, run_fn: *const fn ( ?*anyopaque, *execution.TurnContext, domain.QueuedMessage, domain.AdmissionSnapshot, *std.atomic.Value(bool), ) execution.ServiceError!execution.RunOutcome = unavailableChildRun,}; fn unavailableChildRun( _: ?*anyopaque, _: *execution.TurnContext, _: domain.QueuedMessage, _: domain.AdmissionSnapshot, _: *std.atomic.Value(bool),) execution.ServiceError!execution.RunOutcome { return error.ProviderFailed;} pub const ExecuteOptions = struct { caller_id: []const u8, invocation_id: []const u8, parent_permission_mode: types.PermissionMode = .yolo, root_user_intent_context: []const u8 = "", root_user_messages: []const []const u8 = &.{}, root_user_evidence_complete: bool = false, defaults: Defaults, max_result_bytes: usize, timestamp_ms: i64, relationship_approval_id: ?[]const u8 = null, identity_epoch: u64 = 0,}; pub const MessageSendOptions = struct { caller_id: []const u8, invocation_id: []const u8, child_id: []const u8, content: []const u8, timestamp_ms: i64, identity_epoch: u64 = 0,}; pub const HumanCommandOptions = struct { invocation_id: []const u8, defaults: Defaults, expected_generation: ?u64 = null, timestamp_ms: i64, identity_epoch: u64 = 0,}; pub const ApprovalResolveOptions = struct { request_id: []const u8, child_id: []const u8, decision: types.ToolPermissionDecision, feedback: ?[]const u8 = null, timestamp_ms: i64,}; fn buildHumanQueuedRootUserContext( alloc: Allocator, command: domain.Command,) !?[]u8 { const current_request = humanQueuedRootUserMessage(command); return if (current_request) |request| try auto_classifier_context.buildCanonicalRootUserContext( alloc, request, &.{}, ) else null;} fn humanQueuedRootUserMessage(command: domain.Command) ?[]const u8 { return switch (command) { .create => |create| create.prompt, .message => |message| switch (message) { .send => |send| send.content, .milestone => null, }, .inspect, .relationship, .configure, .lifecycle => null, };} const ModelCommandOutcome = union(enum) { result: manager_mod.Result, relationship_approval: domain.RelationshipCommand, adapter_failure: struct { child_id: ?[]const u8, code: []const u8, retryable: bool = false, }, fn deinit(self: *ModelCommandOutcome, alloc: Allocator) void { switch (self.*) { .result => |*result| result.deinit(alloc), .relationship_approval, .adapter_failure => {}, } self.* = undefined; }}; pub const Runtime = struct { alloc: Allocator, sessions: *session_store.Store, root_id: []u8, host_authority: authority.HostResolver, child_runner: ChildRunner, manager: manager_mod.Manager, durable_approvals: approval_persistence.DurableRegistry, approvals: approval_registry.Registry, authority_resolver: authority.Resolver, owner: execution.Owner, recovery_mutex: std.Io.Mutex = .init, recovery_condition: std.Io.Condition = .init, recovery_thread: ?std.Thread = null, recovery_state: std.atomic.Value(RecoveryState) = .init(.pending), pub fn create( alloc: Allocator, sessions: *session_store.Store, root_id: []const u8, host_authority: authority.HostResolver, child_runner: ChildRunner, ) !*Runtime { try domain.validateId(root_id); const runtime = try alloc.create(Runtime); errdefer alloc.destroy(runtime); const owned_root = try alloc.dupe(u8, root_id); errdefer alloc.free(owned_root); runtime.* = .{ .alloc = alloc, .sessions = sessions, .root_id = owned_root, .host_authority = host_authority, .child_runner = child_runner, .manager = .{ .sessions = sessions }, .durable_approvals = .{ .alloc = alloc, .sessions = sessions }, .approvals = undefined, .authority_resolver = undefined, .owner = undefined, }; runtime.approvals = .{ .alloc = alloc, .persistence = runtime.durable_approvals.interface(), }; runtime.authority_resolver = .{ .sessions = sessions, .host = host_authority, }; runtime.owner = runtime.ownerValue(); return runtime; } pub fn deinit(self: *Runtime) void { if (self.recovery_thread) |thread| thread.join(); self.recovery_thread = null; const had_owned_work = self.owner.started_any; self.owner.deinit(); if (had_owned_work) { var recovery_owner = self.ownerValue(); _ = recovery_owner.recoverTree( self.root_id, io_mod.milliTimestamp(), ) catch |err| { debug_trace.logf( "subagent", "host exit recovery failed root_id={s} outcome={s}", .{ self.root_id, @errorName(err) }, ); }; recovery_owner.deinit(); } self.approvals.deinit(); self.alloc.free(self.root_id); const alloc = self.alloc; self.* = undefined; alloc.destroy(self); } /// Refreshes borrowed host pointers after an embedding runtime moves. /// Callers must do this before any child work starts. pub fn rebind( self: *Runtime, sessions: *session_store.Store, child_runner_context: ?*anyopaque, host_authority: authority.HostResolver, ) void { std.debug.assert(self.owner.slots.items.len == 0); std.debug.assert(self.recovery_thread == null); self.sessions = sessions; self.manager.sessions = sessions; self.durable_approvals.sessions = sessions; self.authority_resolver.sessions = sessions; self.authority_resolver.host = self.host_authority; self.owner.sessions = sessions; self.child_runner.context = child_runner_context; self.host_authority = host_authority; self.authority_resolver.host = host_authority; } pub fn issueOperationIdentity( self: *Runtime, alloc: Allocator, invocation_id: []const u8, source: domain.OperationIdentitySource, ) !u64 { return issueManagerOperationIdentity( alloc, self.sessions, self.root_id, self.manager.options.child_store, invocation_id, source, ); } fn ownerValue(self: *Runtime) execution.Owner { return .{ .alloc = self.alloc, .sessions = self.sessions, .manager = &self.manager, .services = .{ .context = self, .capture_fn = captureAdmission, .run_fn = runChild, }, .live_authority = &self.authority_resolver, .approval_registry = &self.approvals, .notification_clock = .{ .now_fn = notificationNow, }, .notification_poller = .{ .context = self, .poll_fn = pollNotification, }, }; } fn notificationNow(_: ?*anyopaque) i64 { return io_mod.milliTimestamp(); } fn pollNotification( raw: ?*anyopaque, alloc: Allocator, child_id: []const u8, work_id: []const u8, now_ms: i64, ) communication_manager.Error!communication_manager.PollOutcome { const self: *Runtime = @ptrCast(@alignCast(raw orelse return error.StoreUnavailable)); var delivery_manager = communication_manager.Manager{ .sessions = self.sessions, .child_store_options = self.manager.options.child_store, }; return delivery_manager.poll(alloc, child_id, work_id, now_ms); } pub fn execute( self: *Runtime, alloc: Allocator, command: *domain.Command, options: ExecuteOptions, ) ![]u8 { if (command.* == .inspect) { return self.executeModelInspection(alloc, command.*, options); } if (!try self.admitModelCommand( alloc, command, options.caller_id, options.parent_permission_mode, )) { return boundedFailureAlloc( alloc, options.invocation_id, commandTarget(command.*), "permission_escalation", false, options.max_result_bytes, ); } const identity_epoch = if (options.identity_epoch != 0) options.identity_epoch else try self.issueOperationIdentity(alloc, options.invocation_id, .model); const operation_id = try tool_result.boundOperationIdAlloc( alloc, options.invocation_id, .model, identity_epoch, ); defer alloc.free(operation_id); const identity_admitted = try self.operationIdentityOutstanding( alloc, operation_id, ); self.recoverIfNeeded(options.timestamp_ms); var outcome = try self.executeModelMutation( alloc, command, options, operation_id, identity_epoch, identity_admitted, ); defer outcome.deinit(alloc); self.finishModelOutcome(alloc, operation_id, &outcome); return switch (outcome) { .result => |result| self.encodeResult( alloc, operation_id, result, options.max_result_bytes, null, ), .relationship_approval => |relationship| encodeRelationshipApprovalIntent( alloc, operation_id, relationship, options.max_result_bytes, ), .adapter_failure => |failure| boundedFailureAlloc( alloc, operation_id, failure.child_id, failure.code, failure.retryable, options.max_result_bytes, ), }; } fn executeModelInspection( self: *Runtime, alloc: Allocator, command: domain.Command, options: ExecuteOptions, ) ![]u8 { std.debug.assert(command == .inspect); const operation_id = try tool_result.boundOperationIdAlloc( alloc, options.invocation_id, .model, 0, ); defer alloc.free(operation_id); self.recoverIfNeeded(options.timestamp_ms); const target_id = command.inspect.id; const wait = command.inspect.wait; const deadline = if (wait) |requested| std.Io.Clock.Timestamp.fromNow(io_mod.getIo(), .{ .clock = .awake, .raw = .fromMilliseconds(@intCast(requested.timeout_ms)), }) else null; while (true) { var waiter = execution.ChildWaiter{ .child_id = target_id }; if (wait != null) try self.owner.registerChildWaiter(&waiter); defer if (waiter.registered) self.owner.unregisterChildWaiter(&waiter); if (!std.mem.eql(u8, options.caller_id, self.root_id) and !try self.isAttached(self.root_id, options.caller_id)) { return tool_result.failureAlloc( alloc, operation_id, null, "rejected", "caller_unavailable", false, null, ); } if (!std.mem.eql(u8, target_id, options.caller_id) and !try self.isAttached(options.caller_id, target_id)) { return tool_result.failureAlloc( alloc, operation_id, target_id, "rejected",Lines 1–500 of 7251. Load more
const std = @import("std"); const builtin = @import("builtin"); const approval_persistence = @import("approval_persistence.zig"); const approval_registry = @import("approval_registry.zig"); const authority = @import("authority.zig"); const auto_classifier_context = @import("../permissions/auto_classifier_context.zig"); const communication = @import("communication.zig"); const communication_manager = @import("communication_manager.zig"); const create_store = @import("create_store.zig"); const communication_store = @import("communication_store.zig"); const control_store = @import("control_store.zig"); const domain = @import("domain.zig"); const execution = @import("execution.zig"); const manager_mod = @import("manager.zig"); const tool_result = @import("tool_result.zig"); const debug_trace = @import("../shared/debug_trace.zig"); const io_mod = @import("../shared/io.zig"); const session = @import("../session/session.zig"); const session_child_store = @import("../session/session_child_store.zig"); const session_codec = @import("../session/session_codec.zig"); const session_store = @import("../session/session_store.zig"); const mcp_access = @import("../mcp/access_policy.zig"); const mode_registry = @import("../modes/mode_registry.zig"); const permissions = @import("../permissions/permissions.zig"); const session_permission_state = @import("../permissions/session_permission_state.zig"); const tool_dispatch = @import("../tooling/tool_dispatch.zig"); const tool_set_contract = @import("../tooling/tool_set.zig"); const types = @import("../shared/types.zig"); const Allocator = std.mem.Allocator; const inspect_wait_external_poll_ms: i64 = 100; const TargetAuthorizationTestHook = struct { context: ?*anyopaque = null, run_fn: *const fn (?*anyopaque) void, }; const TestHooks = struct { var after_target_authorization: ?TargetAuthorizationTestHook = null; }; fn runAfterTargetAuthorizationTestHook() void { if (comptime builtin.is_test) { const hook = TestHooks.after_target_authorization orelse return; hook.run_fn(hook.context); } } pub const RecoveryState = enum(u8) { pending, scheduled, running, deferred, complete, }; const RecoveryTrigger = enum { automatic, explicit, }; const RecoveryStartDecision = enum { schedule, start, wait, no_effect, }; const RecoveryFinishOutcome = enum { fully_reconciled, incomplete, failed, }; fn decideRecoveryStart( state: RecoveryState, trigger: RecoveryTrigger, ) RecoveryStartDecision { return switch (state) { .pending => switch (trigger) { .automatic => .schedule, .explicit => .start, }, .scheduled, .running => switch (trigger) { .automatic => .no_effect, .explicit => .wait, }, .deferred => switch (trigger) { .automatic => .no_effect, .explicit => .start, }, .complete => .no_effect, }; } fn recoveryStateAfterFinish(outcome: RecoveryFinishOutcome) RecoveryState { return switch (outcome) { .fully_reconciled => .complete, .incomplete, .failed => .deferred, }; } pub const BackgroundRecoveryError = error{ThreadSpawnFailed}; pub const Defaults = struct { model: []const u8, effort: types.ReasoningEffort, fast_mode: bool = false, conversation_language: session.ConversationLanguage, }; pub const ChildRunner = struct { context: ?*anyopaque = null, run_fn: *const fn ( ?*anyopaque, *execution.TurnContext, domain.QueuedMessage, domain.AdmissionSnapshot, *std.atomic.Value(bool), ) execution.ServiceError!execution.RunOutcome = unavailableChildRun, }; fn unavailableChildRun( _: ?*anyopaque, _: *execution.TurnContext, _: domain.QueuedMessage, _: domain.AdmissionSnapshot, _: *std.atomic.Value(bool), ) execution.ServiceError!execution.RunOutcome { return error.ProviderFailed; } pub const ExecuteOptions = struct { caller_id: []const u8, invocation_id: []const u8, parent_permission_mode: types.PermissionMode = .yolo, root_user_intent_context: []const u8 = "", root_user_messages: []const []const u8 = &.{}, root_user_evidence_complete: bool = false, defaults: Defaults, max_result_bytes: usize, timestamp_ms: i64, relationship_approval_id: ?[]const u8 = null, identity_epoch: u64 = 0, }; pub const MessageSendOptions = struct { caller_id: []const u8, invocation_id: []const u8, child_id: []const u8, content: []const u8, timestamp_ms: i64, identity_epoch: u64 = 0, }; pub const HumanCommandOptions = struct { invocation_id: []const u8, defaults: Defaults, expected_generation: ?u64 = null, timestamp_ms: i64, identity_epoch: u64 = 0, }; pub const ApprovalResolveOptions = struct { request_id: []const u8, child_id: []const u8, decision: types.ToolPermissionDecision, feedback: ?[]const u8 = null, timestamp_ms: i64, }; fn buildHumanQueuedRootUserContext( alloc: Allocator, command: domain.Command, ) !?[]u8 { const current_request = humanQueuedRootUserMessage(command); return if (current_request) |request| try auto_classifier_context.buildCanonicalRootUserContext( alloc, request, &.{}, ) else null; } fn humanQueuedRootUserMessage(command: domain.Command) ?[]const u8 { return switch (command) { .create => |create| create.prompt, .message => |message| switch (message) { .send => |send| send.content, .milestone => null, }, .inspect, .relationship, .configure, .lifecycle => null, }; } const ModelCommandOutcome = union(enum) { result: manager_mod.Result, relationship_approval: domain.RelationshipCommand, adapter_failure: struct { child_id: ?[]const u8, code: []const u8, retryable: bool = false, }, fn deinit(self: *ModelCommandOutcome, alloc: Allocator) void { switch (self.*) { .result => |*result| result.deinit(alloc), .relationship_approval, .adapter_failure => {}, } self.* = undefined; } }; pub const Runtime = struct { alloc: Allocator, sessions: *session_store.Store, root_id: []u8, host_authority: authority.HostResolver, child_runner: ChildRunner, manager: manager_mod.Manager, durable_approvals: approval_persistence.DurableRegistry, approvals: approval_registry.Registry, authority_resolver: authority.Resolver, owner: execution.Owner, recovery_mutex: std.Io.Mutex = .init, recovery_condition: std.Io.Condition = .init, recovery_thread: ?std.Thread = null, recovery_state: std.atomic.Value(RecoveryState) = .init(.pending), pub fn create( alloc: Allocator, sessions: *session_store.Store, root_id: []const u8, host_authority: authority.HostResolver, child_runner: ChildRunner, ) !*Runtime { try domain.validateId(root_id); const runtime = try alloc.create(Runtime); errdefer alloc.destroy(runtime); const owned_root = try alloc.dupe(u8, root_id); errdefer alloc.free(owned_root); runtime.* = .{ .alloc = alloc, .sessions = sessions, .root_id = owned_root, .host_authority = host_authority, .child_runner = child_runner, .manager = .{ .sessions = sessions }, .durable_approvals = .{ .alloc = alloc, .sessions = sessions }, .approvals = undefined, .authority_resolver = undefined, .owner = undefined, }; runtime.approvals = .{ .alloc = alloc, .persistence = runtime.durable_approvals.interface(), }; runtime.authority_resolver = .{ .sessions = sessions, .host = host_authority, }; runtime.owner = runtime.ownerValue(); return runtime; } pub fn deinit(self: *Runtime) void { if (self.recovery_thread) |thread| thread.join(); self.recovery_thread = null; const had_owned_work = self.owner.started_any; self.owner.deinit(); if (had_owned_work) { var recovery_owner = self.ownerValue(); _ = recovery_owner.recoverTree( self.root_id, io_mod.milliTimestamp(), ) catch |err| { debug_trace.logf( "subagent", "host exit recovery failed root_id={s} outcome={s}", .{ self.root_id, @errorName(err) }, ); }; recovery_owner.deinit(); } self.approvals.deinit(); self.alloc.free(self.root_id); const alloc = self.alloc; self.* = undefined; alloc.destroy(self); } /// Refreshes borrowed host pointers after an embedding runtime moves. /// Callers must do this before any child work starts. pub fn rebind( self: *Runtime, sessions: *session_store.Store, child_runner_context: ?*anyopaque, host_authority: authority.HostResolver, ) void { std.debug.assert(self.owner.slots.items.len == 0); std.debug.assert(self.recovery_thread == null); self.sessions = sessions; self.manager.sessions = sessions; self.durable_approvals.sessions = sessions; self.authority_resolver.sessions = sessions; self.authority_resolver.host = self.host_authority; self.owner.sessions = sessions; self.child_runner.context = child_runner_context; self.host_authority = host_authority; self.authority_resolver.host = host_authority; } pub fn issueOperationIdentity( self: *Runtime, alloc: Allocator, invocation_id: []const u8, source: domain.OperationIdentitySource, ) !u64 { return issueManagerOperationIdentity( alloc, self.sessions, self.root_id, self.manager.options.child_store, invocation_id, source, ); } fn ownerValue(self: *Runtime) execution.Owner { return .{ .alloc = self.alloc, .sessions = self.sessions, .manager = &self.manager, .services = .{ .context = self, .capture_fn = captureAdmission, .run_fn = runChild, }, .live_authority = &self.authority_resolver, .approval_registry = &self.approvals, .notification_clock = .{ .now_fn = notificationNow, }, .notification_poller = .{ .context = self, .poll_fn = pollNotification, }, }; } fn notificationNow(_: ?*anyopaque) i64 { return io_mod.milliTimestamp(); } fn pollNotification( raw: ?*anyopaque, alloc: Allocator, child_id: []const u8, work_id: []const u8, now_ms: i64, ) communication_manager.Error!communication_manager.PollOutcome { const self: *Runtime = @ptrCast(@alignCast(raw orelse return error.StoreUnavailable)); var delivery_manager = communication_manager.Manager{ .sessions = self.sessions, .child_store_options = self.manager.options.child_store, }; return delivery_manager.poll(alloc, child_id, work_id, now_ms); } pub fn execute( self: *Runtime, alloc: Allocator, command: *domain.Command, options: ExecuteOptions, ) ![]u8 { if (command.* == .inspect) { return self.executeModelInspection(alloc, command.*, options); } if (!try self.admitModelCommand( alloc, command, options.caller_id, options.parent_permission_mode, )) { return boundedFailureAlloc( alloc, options.invocation_id, commandTarget(command.*), "permission_escalation", false, options.max_result_bytes, ); } const identity_epoch = if (options.identity_epoch != 0) options.identity_epoch else try self.issueOperationIdentity(alloc, options.invocation_id, .model); const operation_id = try tool_result.boundOperationIdAlloc( alloc, options.invocation_id, .model, identity_epoch, ); defer alloc.free(operation_id); const identity_admitted = try self.operationIdentityOutstanding( alloc, operation_id, ); self.recoverIfNeeded(options.timestamp_ms); var outcome = try self.executeModelMutation( alloc, command, options, operation_id, identity_epoch, identity_admitted, ); defer outcome.deinit(alloc); self.finishModelOutcome(alloc, operation_id, &outcome); return switch (outcome) { .result => |result| self.encodeResult( alloc, operation_id, result, options.max_result_bytes, null, ), .relationship_approval => |relationship| encodeRelationshipApprovalIntent( alloc, operation_id, relationship, options.max_result_bytes, ), .adapter_failure => |failure| boundedFailureAlloc( alloc, operation_id, failure.child_id, failure.code, failure.retryable, options.max_result_bytes, ), }; } fn executeModelInspection( self: *Runtime, alloc: Allocator, command: domain.Command, options: ExecuteOptions, ) ![]u8 { std.debug.assert(command == .inspect); const operation_id = try tool_result.boundOperationIdAlloc( alloc, options.invocation_id, .model, 0, ); defer alloc.free(operation_id); self.recoverIfNeeded(options.timestamp_ms); const target_id = command.inspect.id; const wait = command.inspect.wait; const deadline = if (wait) |requested| std.Io.Clock.Timestamp.fromNow(io_mod.getIo(), .{ .clock = .awake, .raw = .fromMilliseconds(@intCast(requested.timeout_ms)), }) else null; while (true) { var waiter = execution.ChildWaiter{ .child_id = target_id }; if (wait != null) try self.owner.registerChildWaiter(&waiter); defer if (waiter.registered) self.owner.unregisterChildWaiter(&waiter); if (!std.mem.eql(u8, options.caller_id, self.root_id) and !try self.isAttached(self.root_id, options.caller_id)) { return tool_result.failureAlloc( alloc, operation_id, null, "rejected", "caller_unavailable", false, null, ); } if (!std.mem.eql(u8, target_id, options.caller_id) and !try self.isAttached(options.caller_id, target_id)) { return tool_result.failureAlloc( alloc, operation_id, target_id, "rejected",