#![forbid(unsafe_code)] //! Process-owned trusted runtime services. //! //! Production sidecars construct exactly one [`SidecarRuntime`] at their //! process entrypoint. Subsystems receive a clone of [`RuntimeContext`]; they //! never construct Tokio runtimes of their own. use std::cell::Cell; use std::fmt; use std::future::Future; use std::pin::Pin; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::{mpsc, Arc, Mutex, OnceLock}; use std::task::{Context, Poll}; use std::thread; use std::time::{Duration, Instant}; use accounting::{LimitError, Reservation, ResourceClass, ResourceLedger, ResourceLimit}; use fairness::{FairBudget, FairWorkBroker, FairnessConfig}; use metrics::{ ExecutorMetricClass, RuntimeMetrics, TelemetryFallback, TelemetryFallbackCode, TelemetrySeverity, TelemetrySubsystem, WatchdogMetric, }; pub mod accounting; pub mod capability; pub mod fairness; pub mod metrics; pub mod readiness; pub mod supervision; pub use supervision::{ TaskClass, TaskClassSnapshot, TaskOwner, TaskSpawnError, TaskSupervisor, TaskTerminalReason, TaskTerminalReport, }; const DEFAULT_MAX_BLOCKING_JOB_BYTES: usize = 64 * 1024 * 1024; const DEFAULT_BLOCKING_JOB_TIMEOUT_MS: u64 = 30_000; const DEFAULT_MAX_BLOCKING_JOBS: usize = 1_028; const DEFAULT_MAX_QUEUED_BLOCKING_JOBS: usize = 1024; const DEFAULT_MAX_PROCESS_CAPABILITIES: usize = 16_384; const DEFAULT_MAX_PROCESS_SOCKETS: usize = 8_192; const DEFAULT_MAX_PROCESS_CONNECTIONS: usize = 8_192; const DEFAULT_MAX_PROCESS_SOCKET_BUFFERED_BYTES: usize = 1024 * 1024 * 1024; const DEFAULT_MAX_PROCESS_DATAGRAMS: usize = 65_536; const DEFAULT_MAX_PROCESS_TIMERS: usize = 65_536; const DEFAULT_MAX_PROCESS_TASKS: usize = 65_536; const DEFAULT_MAX_PROCESS_READY_HANDLES: usize = 16_384; const DEFAULT_MAX_PROCESS_HANDLE_COMMANDS: usize = 65_536; const DEFAULT_MAX_PROCESS_HANDLE_COMMAND_BYTES: usize = 256 * 1024 * 1024; const DEFAULT_MAX_PROCESS_BRIDGE_CALLS: usize = 65_536; const DEFAULT_MAX_PROCESS_BRIDGE_REQUEST_BYTES: usize = 256 * 1024 * 1024; const DEFAULT_MAX_PROCESS_BRIDGE_RESPONSE_BYTES: usize = 256 * 1024 * 1024; const DEFAULT_MAX_PROCESS_ASYNC_COMPLETIONS: usize = 65_536; const DEFAULT_MAX_PROCESS_ASYNC_COMPLETION_BYTES: usize = 256 * 1024 * 1024; const DEFAULT_MAX_PROCESS_UDP_DATAGRAMS: usize = 65_536; const DEFAULT_MAX_PROCESS_UDP_BYTES: usize = 256 * 1024 * 1024; const DEFAULT_MAX_PROCESS_TLS_BYTES: usize = 256 * 1024 * 1024; const DEFAULT_MAX_PROCESS_HTTP2_CONNECTIONS: usize = 4_096; const DEFAULT_MAX_PROCESS_HTTP2_STREAMS: usize = 65_536; const DEFAULT_MAX_PROCESS_HTTP2_BYTES: usize = 512 * 1024 * 1024; const DEFAULT_MAX_PROCESS_HTTP2_HEADER_BYTES: usize = 128 * 1024 * 1024; const DEFAULT_MAX_PROCESS_HTTP2_DATA_BYTES: usize = 512 * 1024 * 1024; const DEFAULT_MAX_PROCESS_HTTP2_COMMANDS: usize = 65_536; const DEFAULT_MAX_PROCESS_HTTP2_COMMAND_BYTES: usize = 256 * 1024 * 1024; const DEFAULT_MAX_PROCESS_HTTP2_EVENTS: usize = 65_536; const DEFAULT_MAX_PROCESS_HTTP2_EVENT_BYTES: usize = 512 * 1024 * 1024; const DEFAULT_TASK_POLL_WATCHDOG_MS: u64 = 100; const DEFAULT_MAX_TERMINAL_TASK_REPORTS: usize = 4_096; const DEFAULT_VM_EXECUTOR_TEARDOWN_TIMEOUT_MS: u64 = 5_000; pub const DEFAULT_PROTOCOL_MAX_INGRESS_FRAMES: usize = 128; pub const DEFAULT_PROTOCOL_MAX_INGRESS_BYTES: usize = 64 * 1024 * 1024; pub const DEFAULT_PROTOCOL_MAX_CONTROL_FRAMES: usize = 1_024; pub const DEFAULT_PROTOCOL_MAX_CONTROL_BYTES: usize = 64 * 1024 * 1024; pub const DEFAULT_PROTOCOL_MAX_EGRESS_FRAMES: usize = 4_096; pub const DEFAULT_PROTOCOL_MAX_EGRESS_BYTES: usize = 256 * 1024 * 1024; pub const DEFAULT_PROTOCOL_MAX_PENDING_RESPONSES: usize = 10_000; pub const DEFAULT_PROTOCOL_MAX_PENDING_RESPONSE_BYTES: usize = 256 * 1024 * 1024; pub const DEFAULT_PROTOCOL_MAX_PROCESS_EVENTS: usize = 10_000; pub const DEFAULT_PROTOCOL_MAX_OUTBOUND_REQUESTS: usize = 10_000; pub const DEFAULT_PROTOCOL_MAX_COMPLETED_RESPONSES: usize = 10_000; const DEFAULT_FAIRNESS_VM_OPERATIONS: usize = 64; const DEFAULT_FAIRNESS_VM_BYTES: usize = 1024 * 1024; const DEFAULT_FAIRNESS_CAPABILITY_OPERATIONS: usize = 16; const DEFAULT_FAIRNESS_CAPABILITY_BYTES: usize = 256 * 1024; const DEFAULT_FAIRNESS_MAX_VMS: usize = 4_096; const DEFAULT_FAIRNESS_MAX_CAPABILITIES_PER_VM: usize = 16_384; thread_local! { static IS_AGENTOS_RUNTIME_WORKER: Cell = const { Cell::new(false) }; } /// Whether the current OS thread is one of the process runtime's fixed workers. /// /// Synchronous compatibility adapters use this to reject waits that would /// consume a trusted runtime worker and could deadlock the work they submitted. pub fn is_runtime_worker_thread() -> bool { IS_AGENTOS_RUNTIME_WORKER.with(Cell::get) } /// Process-owned bounds for the multiplexed sidecar transport. /// /// Ordinary frames and response/control frames have independent admission so /// an event or request backlog cannot consume the capacity needed to settle an /// already-registered bridge call. Byte ceilings cover retained decoded or /// encoded frames; the one active decoder is separately bounded by the wire /// codec's `max_frame_bytes` setting. #[derive(Clone, Debug, PartialEq, Eq)] pub struct RuntimeProtocolConfig { pub max_ingress_frames: usize, pub max_ingress_bytes: usize, pub max_control_frames: usize, pub max_control_bytes: usize, pub max_egress_frames: usize, pub max_egress_bytes: usize, pub max_pending_responses: usize, pub max_pending_response_bytes: usize, pub max_process_events: usize, pub max_outbound_requests: usize, pub max_completed_responses: usize, } impl Default for RuntimeProtocolConfig { fn default() -> Self { Self { max_ingress_frames: DEFAULT_PROTOCOL_MAX_INGRESS_FRAMES, max_ingress_bytes: DEFAULT_PROTOCOL_MAX_INGRESS_BYTES, max_control_frames: DEFAULT_PROTOCOL_MAX_CONTROL_FRAMES, max_control_bytes: DEFAULT_PROTOCOL_MAX_CONTROL_BYTES, max_egress_frames: DEFAULT_PROTOCOL_MAX_EGRESS_FRAMES, max_egress_bytes: DEFAULT_PROTOCOL_MAX_EGRESS_BYTES, max_pending_responses: DEFAULT_PROTOCOL_MAX_PENDING_RESPONSES, max_pending_response_bytes: DEFAULT_PROTOCOL_MAX_PENDING_RESPONSE_BYTES, max_process_events: DEFAULT_PROTOCOL_MAX_PROCESS_EVENTS, max_outbound_requests: DEFAULT_PROTOCOL_MAX_OUTBOUND_REQUESTS, max_completed_responses: DEFAULT_PROTOCOL_MAX_COMPLETED_RESPONSES, } } } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RuntimeFairnessConfig { pub vm_quantum_operations: usize, pub vm_quantum_bytes: usize, pub capability_quantum_operations: usize, pub capability_quantum_bytes: usize, pub max_vm_deficit_operations: usize, pub max_vm_deficit_bytes: usize, pub max_capability_deficit_operations: usize, pub max_capability_deficit_bytes: usize, pub max_vms: usize, pub max_capabilities_per_vm: usize, } impl Default for RuntimeFairnessConfig { fn default() -> Self { Self { vm_quantum_operations: DEFAULT_FAIRNESS_VM_OPERATIONS, vm_quantum_bytes: DEFAULT_FAIRNESS_VM_BYTES, capability_quantum_operations: DEFAULT_FAIRNESS_CAPABILITY_OPERATIONS, capability_quantum_bytes: DEFAULT_FAIRNESS_CAPABILITY_BYTES, max_vm_deficit_operations: DEFAULT_FAIRNESS_VM_OPERATIONS * 4, max_vm_deficit_bytes: DEFAULT_FAIRNESS_VM_BYTES * 4, max_capability_deficit_operations: DEFAULT_FAIRNESS_CAPABILITY_OPERATIONS * 4, max_capability_deficit_bytes: DEFAULT_FAIRNESS_CAPABILITY_BYTES * 4, max_vms: DEFAULT_FAIRNESS_MAX_VMS, max_capabilities_per_vm: DEFAULT_FAIRNESS_MAX_CAPABILITIES_PER_VM, } } } impl RuntimeFairnessConfig { fn scheduler_config(&self) -> FairnessConfig { FairnessConfig { vm_quantum: FairBudget::new(self.vm_quantum_operations, self.vm_quantum_bytes), capability_quantum: FairBudget::new( self.capability_quantum_operations, self.capability_quantum_bytes, ), max_vm_deficit: FairBudget::new( self.max_vm_deficit_operations, self.max_vm_deficit_bytes, ), max_capability_deficit: FairBudget::new( self.max_capability_deficit_operations, self.max_capability_deficit_bytes, ), max_vms: self.max_vms, max_capabilities_per_vm: self.max_capabilities_per_vm, } } } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RuntimeResourceConfig { pub max_capabilities: usize, pub max_ready_handles: usize, pub max_sockets: usize, pub max_connections: usize, pub max_socket_buffered_bytes: usize, pub max_datagrams: usize, pub max_timers: usize, pub max_tasks: usize, pub max_handle_commands: usize, pub max_handle_command_bytes: usize, pub max_bridge_calls: usize, pub max_bridge_request_bytes: usize, pub max_bridge_response_bytes: usize, pub max_async_completions: usize, pub max_async_completion_bytes: usize, pub max_udp_datagrams: usize, pub max_udp_bytes: usize, pub max_tls_bytes: usize, pub max_http2_connections: usize, pub max_http2_streams: usize, pub max_http2_buffered_bytes: usize, pub max_http2_header_bytes: usize, pub max_http2_data_bytes: usize, pub max_http2_commands: usize, pub max_http2_command_bytes: usize, pub max_http2_events: usize, pub max_http2_event_bytes: usize, } impl Default for RuntimeResourceConfig { fn default() -> Self { Self { max_capabilities: DEFAULT_MAX_PROCESS_CAPABILITIES, max_ready_handles: DEFAULT_MAX_PROCESS_READY_HANDLES, max_sockets: DEFAULT_MAX_PROCESS_SOCKETS, max_connections: DEFAULT_MAX_PROCESS_CONNECTIONS, max_socket_buffered_bytes: DEFAULT_MAX_PROCESS_SOCKET_BUFFERED_BYTES, max_datagrams: DEFAULT_MAX_PROCESS_DATAGRAMS, max_timers: DEFAULT_MAX_PROCESS_TIMERS, max_tasks: DEFAULT_MAX_PROCESS_TASKS, max_handle_commands: DEFAULT_MAX_PROCESS_HANDLE_COMMANDS, max_handle_command_bytes: DEFAULT_MAX_PROCESS_HANDLE_COMMAND_BYTES, max_bridge_calls: DEFAULT_MAX_PROCESS_BRIDGE_CALLS, max_bridge_request_bytes: DEFAULT_MAX_PROCESS_BRIDGE_REQUEST_BYTES, max_bridge_response_bytes: DEFAULT_MAX_PROCESS_BRIDGE_RESPONSE_BYTES, max_async_completions: DEFAULT_MAX_PROCESS_ASYNC_COMPLETIONS, max_async_completion_bytes: DEFAULT_MAX_PROCESS_ASYNC_COMPLETION_BYTES, max_udp_datagrams: DEFAULT_MAX_PROCESS_UDP_DATAGRAMS, max_udp_bytes: DEFAULT_MAX_PROCESS_UDP_BYTES, max_tls_bytes: DEFAULT_MAX_PROCESS_TLS_BYTES, max_http2_connections: DEFAULT_MAX_PROCESS_HTTP2_CONNECTIONS, max_http2_streams: DEFAULT_MAX_PROCESS_HTTP2_STREAMS, max_http2_buffered_bytes: DEFAULT_MAX_PROCESS_HTTP2_BYTES, max_http2_header_bytes: DEFAULT_MAX_PROCESS_HTTP2_HEADER_BYTES, max_http2_data_bytes: DEFAULT_MAX_PROCESS_HTTP2_DATA_BYTES, max_http2_commands: DEFAULT_MAX_PROCESS_HTTP2_COMMANDS, max_http2_command_bytes: DEFAULT_MAX_PROCESS_HTTP2_COMMAND_BYTES, max_http2_events: DEFAULT_MAX_PROCESS_HTTP2_EVENTS, max_http2_event_bytes: DEFAULT_MAX_PROCESS_HTTP2_EVENT_BYTES, } } } impl RuntimeResourceConfig { fn limits(&self) -> Vec<(ResourceClass, ResourceLimit)> { vec![ ( ResourceClass::Capabilities, ResourceLimit::new(self.max_capabilities, "runtime.resources.maxCapabilities"), ), ( ResourceClass::ReadyHandles, ResourceLimit::new(self.max_ready_handles, "runtime.resources.maxReadyHandles"), ), ( ResourceClass::Sockets, ResourceLimit::new(self.max_sockets, "runtime.resources.maxSockets"), ), ( ResourceClass::Connections, ResourceLimit::new(self.max_connections, "runtime.resources.maxConnections"), ), ( ResourceClass::BufferedBytes, ResourceLimit::new( self.max_socket_buffered_bytes, "runtime.resources.maxSocketBufferedBytes", ), ), ( ResourceClass::Datagrams, ResourceLimit::new(self.max_datagrams, "runtime.resources.maxDatagrams"), ), ( ResourceClass::Timers, ResourceLimit::new(self.max_timers, "runtime.resources.maxTimers"), ), ( ResourceClass::Tasks, ResourceLimit::new(self.max_tasks, "runtime.resources.maxTasks"), ), ( ResourceClass::HandleCommands, ResourceLimit::new( self.max_handle_commands, "runtime.resources.maxHandleCommands", ), ), ( ResourceClass::HandleCommandBytes, ResourceLimit::new( self.max_handle_command_bytes, "runtime.resources.maxHandleCommandBytes", ), ), ( ResourceClass::BridgeCalls, ResourceLimit::new(self.max_bridge_calls, "runtime.resources.maxBridgeCalls"), ), ( ResourceClass::BridgeRequestBytes, ResourceLimit::new( self.max_bridge_request_bytes, "runtime.resources.maxBridgeRequestBytes", ), ), ( ResourceClass::BridgeResponseBytes, ResourceLimit::new( self.max_bridge_response_bytes, "runtime.resources.maxBridgeResponseBytes", ), ), ( ResourceClass::AsyncCompletions, ResourceLimit::new( self.max_async_completions, "runtime.resources.maxAsyncCompletions", ), ), ( ResourceClass::AsyncCompletionBytes, ResourceLimit::new( self.max_async_completion_bytes, "runtime.resources.maxAsyncCompletionBytes", ), ), ( ResourceClass::UdpDatagrams, ResourceLimit::new(self.max_udp_datagrams, "runtime.resources.maxUdpDatagrams"), ), ( ResourceClass::UdpBytes, ResourceLimit::new(self.max_udp_bytes, "runtime.resources.maxUdpBytes"), ), ( ResourceClass::TlsBytes, ResourceLimit::new(self.max_tls_bytes, "runtime.resources.maxTlsBytes"), ), ( ResourceClass::Http2Connections, ResourceLimit::new( self.max_http2_connections, "runtime.resources.maxHttp2Connections", ), ), ( ResourceClass::Http2Streams, ResourceLimit::new(self.max_http2_streams, "runtime.resources.maxHttp2Streams"), ), ( ResourceClass::Http2BufferedBytes, ResourceLimit::new( self.max_http2_buffered_bytes, "runtime.resources.maxHttp2BufferedBytes", ), ), ( ResourceClass::Http2HeaderBytes, ResourceLimit::new( self.max_http2_header_bytes, "runtime.resources.maxHttp2HeaderBytes", ), ), ( ResourceClass::Http2DataBytes, ResourceLimit::new( self.max_http2_data_bytes, "runtime.resources.maxHttp2DataBytes", ), ), ( ResourceClass::Http2Commands, ResourceLimit::new( self.max_http2_commands, "runtime.resources.maxHttp2Commands", ), ), ( ResourceClass::Http2CommandBytes, ResourceLimit::new( self.max_http2_command_bytes, "runtime.resources.maxHttp2CommandBytes", ), ), ( ResourceClass::Http2Events, ResourceLimit::new(self.max_http2_events, "runtime.resources.maxHttp2Events"), ), ( ResourceClass::Http2EventBytes, ResourceLimit::new( self.max_http2_event_bytes, "runtime.resources.maxHttp2EventBytes", ), ), ] } } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RuntimeConfig { pub worker_threads: usize, pub max_active_vm_executors: usize, pub vm_executor_teardown_timeout_ms: u64, pub blocking_worker_threads: usize, pub max_blocking_jobs: usize, pub max_queued_blocking_jobs: usize, pub max_blocking_job_bytes: usize, pub blocking_job_timeout_ms: u64, pub task_poll_watchdog_ms: u64, pub max_terminal_task_reports: usize, pub protocol: RuntimeProtocolConfig, pub resources: RuntimeResourceConfig, pub fairness: RuntimeFairnessConfig, } impl Default for RuntimeConfig { fn default() -> Self { let available = thread::available_parallelism() .map(usize::from) .unwrap_or(1); Self { worker_threads: available.clamp(1, 4), max_active_vm_executors: available.max(1), vm_executor_teardown_timeout_ms: DEFAULT_VM_EXECUTOR_TEARDOWN_TIMEOUT_MS, blocking_worker_threads: available.clamp(1, 4), max_blocking_jobs: DEFAULT_MAX_BLOCKING_JOBS, max_queued_blocking_jobs: DEFAULT_MAX_QUEUED_BLOCKING_JOBS, max_blocking_job_bytes: DEFAULT_MAX_BLOCKING_JOB_BYTES, blocking_job_timeout_ms: DEFAULT_BLOCKING_JOB_TIMEOUT_MS, task_poll_watchdog_ms: DEFAULT_TASK_POLL_WATCHDOG_MS, max_terminal_task_reports: DEFAULT_MAX_TERMINAL_TASK_REPORTS, protocol: RuntimeProtocolConfig::default(), resources: RuntimeResourceConfig::default(), fairness: RuntimeFairnessConfig::default(), } } } impl RuntimeConfig { pub fn validate(&self) -> Result<(), RuntimeBuildError> { for (field, value) in [ ("runtime.workerThreads", self.worker_threads), ( "runtime.executor.maxActiveVms", self.max_active_vm_executors, ), ( "runtime.blocking.workerThreads", self.blocking_worker_threads, ), ("runtime.blocking.maxJobs", self.max_blocking_jobs), ( "runtime.blocking.maxQueuedJobs", self.max_queued_blocking_jobs, ), ( "runtime.blocking.maxQueuedBytes", self.max_blocking_job_bytes, ), ( "runtime.tasks.maxTerminalReports", self.max_terminal_task_reports, ), ( "runtime.protocol.maxIngressFrames", self.protocol.max_ingress_frames, ), ( "runtime.protocol.maxIngressBytes", self.protocol.max_ingress_bytes, ), ( "runtime.protocol.maxControlFrames", self.protocol.max_control_frames, ), ( "runtime.protocol.maxControlBytes", self.protocol.max_control_bytes, ), ( "runtime.protocol.maxEgressFrames", self.protocol.max_egress_frames, ), ( "runtime.protocol.maxEgressBytes", self.protocol.max_egress_bytes, ), ( "runtime.protocol.maxPendingResponses", self.protocol.max_pending_responses, ), ( "runtime.protocol.maxPendingResponseBytes", self.protocol.max_pending_response_bytes, ), ( "runtime.protocol.maxProcessEvents", self.protocol.max_process_events, ), ( "runtime.protocol.maxOutboundRequests", self.protocol.max_outbound_requests, ), ( "runtime.protocol.maxCompletedResponses", self.protocol.max_completed_responses, ), ( "runtime.resources.maxCapabilities", self.resources.max_capabilities, ), ( "runtime.resources.maxReadyHandles", self.resources.max_ready_handles, ), ("runtime.resources.maxSockets", self.resources.max_sockets), ( "runtime.resources.maxConnections", self.resources.max_connections, ), ( "runtime.resources.maxSocketBufferedBytes", self.resources.max_socket_buffered_bytes, ), ( "runtime.resources.maxDatagrams", self.resources.max_datagrams, ), ("runtime.resources.maxTimers", self.resources.max_timers), ("runtime.resources.maxTasks", self.resources.max_tasks), ( "runtime.resources.maxHandleCommands", self.resources.max_handle_commands, ), ( "runtime.resources.maxHandleCommandBytes", self.resources.max_handle_command_bytes, ), ( "runtime.resources.maxBridgeCalls", self.resources.max_bridge_calls, ), ( "runtime.resources.maxBridgeRequestBytes", self.resources.max_bridge_request_bytes, ), ( "runtime.resources.maxBridgeResponseBytes", self.resources.max_bridge_response_bytes, ), ( "runtime.resources.maxAsyncCompletions", self.resources.max_async_completions, ), ( "runtime.resources.maxAsyncCompletionBytes", self.resources.max_async_completion_bytes, ), ( "runtime.resources.maxUdpDatagrams", self.resources.max_udp_datagrams, ), ( "runtime.resources.maxUdpBytes", self.resources.max_udp_bytes, ), ( "runtime.resources.maxTlsBytes", self.resources.max_tls_bytes, ), ( "runtime.resources.maxHttp2Connections", self.resources.max_http2_connections, ), ( "runtime.resources.maxHttp2Streams", self.resources.max_http2_streams, ), ( "runtime.resources.maxHttp2BufferedBytes", self.resources.max_http2_buffered_bytes, ), ( "runtime.resources.maxHttp2HeaderBytes", self.resources.max_http2_header_bytes, ), ( "runtime.resources.maxHttp2DataBytes", self.resources.max_http2_data_bytes, ), ( "runtime.resources.maxHttp2Commands", self.resources.max_http2_commands, ), ( "runtime.resources.maxHttp2CommandBytes", self.resources.max_http2_command_bytes, ), ( "runtime.resources.maxHttp2Events", self.resources.max_http2_events, ), ( "runtime.resources.maxHttp2EventBytes", self.resources.max_http2_event_bytes, ), ] { if value == 0 { return Err(RuntimeBuildError(format!( "ERR_AGENTOS_RUNTIME_CONFIG: {field} must be greater than zero" ))); } } if self.task_poll_watchdog_ms == 0 { return Err(RuntimeBuildError(String::from( "ERR_AGENTOS_RUNTIME_CONFIG: runtime.watchdog.taskPollMs must be greater than zero", ))); } if self.vm_executor_teardown_timeout_ms == 0 { return Err(RuntimeBuildError(String::from( "ERR_AGENTOS_RUNTIME_CONFIG: runtime.executor.teardownTimeoutMs must be greater than zero", ))); } if self.blocking_job_timeout_ms == 0 { return Err(RuntimeBuildError(String::from( "ERR_AGENTOS_RUNTIME_CONFIG: runtime.blocking.jobTimeoutMs must be greater than zero", ))); } for (field, value) in [ ( "runtime.fairness.vmQuantumOperations", self.fairness.vm_quantum_operations, ), ( "runtime.fairness.vmQuantumBytes", self.fairness.vm_quantum_bytes, ), ( "runtime.fairness.capabilityQuantumOperations", self.fairness.capability_quantum_operations, ), ( "runtime.fairness.capabilityQuantumBytes", self.fairness.capability_quantum_bytes, ), ( "runtime.fairness.maxVmDeficitOperations", self.fairness.max_vm_deficit_operations, ), ( "runtime.fairness.maxVmDeficitBytes", self.fairness.max_vm_deficit_bytes, ), ( "runtime.fairness.maxCapabilityDeficitOperations", self.fairness.max_capability_deficit_operations, ), ( "runtime.fairness.maxCapabilityDeficitBytes", self.fairness.max_capability_deficit_bytes, ), ("runtime.fairness.maxVms", self.fairness.max_vms), ( "runtime.fairness.maxCapabilitiesPerVm", self.fairness.max_capabilities_per_vm, ), ] { if value == 0 { return Err(RuntimeBuildError(format!( "ERR_AGENTOS_RUNTIME_CONFIG: {field} must be greater than zero" ))); } } if self.fairness.max_vm_deficit_operations < self.fairness.vm_quantum_operations || self.fairness.max_vm_deficit_bytes < self.fairness.vm_quantum_bytes || self.fairness.max_capability_deficit_operations < self.fairness.capability_quantum_operations || self.fairness.max_capability_deficit_bytes < self.fairness.capability_quantum_bytes { return Err(RuntimeBuildError(String::from( "ERR_AGENTOS_RUNTIME_CONFIG: fairness deficits must be at least their quantum", ))); } if self.max_blocking_jobs < self.blocking_worker_threads { return Err(RuntimeBuildError(format!( "ERR_AGENTOS_RUNTIME_CONFIG: runtime.blocking.maxJobs ({}) must be >= runtime.blocking.workerThreads ({})", self.max_blocking_jobs, self.blocking_worker_threads ))); } if self.max_queued_blocking_jobs > self.max_blocking_jobs { return Err(RuntimeBuildError(format!( "ERR_AGENTOS_RUNTIME_CONFIG: runtime.blocking.maxQueuedJobs ({}) must be <= runtime.blocking.maxJobs ({})", self.max_queued_blocking_jobs, self.max_blocking_jobs ))); } Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RuntimeBuildError(String); impl fmt::Display for RuntimeBuildError { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter.write_str(&self.0) } } impl std::error::Error for RuntimeBuildError {} #[derive(Debug, Clone, PartialEq, Eq)] pub enum BlockingJobError { ResourceLimit(LimitError), Capacity { limit: usize }, ShuttingDown, WorkerDropped, TimedOut { timeout: Duration }, } impl fmt::Display for BlockingJobError { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::ResourceLimit(error) => error.fmt(formatter), Self::Capacity { limit } => write!( formatter, "ERR_AGENTOS_BLOCKING_JOB_LIMIT: blocking executor queue exceeded {limit} jobs; raise runtime.blocking.maxQueuedJobs" ), Self::ShuttingDown => formatter.write_str( "ERR_AGENTOS_BLOCKING_EXECUTOR_SHUTDOWN: blocking executor is shutting down", ), Self::WorkerDropped => formatter.write_str( "ERR_AGENTOS_BLOCKING_WORKER_DROPPED: blocking worker ended without a result", ), Self::TimedOut { timeout } => write!( formatter, "ERR_AGENTOS_BLOCKING_JOB_TIMEOUT: blocking job exceeded its {}ms deadline", timeout.as_millis() ), } } } impl std::error::Error for BlockingJobError {} type BlockingOperation = Box; struct BlockingJob { operation: BlockingOperation, _slot: Reservation, _bytes: Reservation, } struct BlockingExecutorState { metrics: RuntimeMetrics, queued: AtomicUsize, active: AtomicUsize, } struct BlockingExecutorInner { sender: Mutex>>, workers: Mutex>>, max_queued_jobs: usize, max_bytes: usize, state: Arc, } impl Drop for BlockingExecutorInner { fn drop(&mut self) { self.sender.get_mut().ok().and_then(Option::take); let workers = self .workers .get_mut() .map(std::mem::take) .unwrap_or_default(); for worker in workers { if worker.join().is_err() { eprintln!("ERR_AGENTOS_BLOCKING_WORKER_PANIC: blocking executor worker panicked"); } } } } #[derive(Clone)] pub struct BlockingExecutor { inner: Arc, resources: Arc, admission_open: Arc, admission_gate: Arc>, } impl fmt::Debug for BlockingExecutor { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter .debug_struct("BlockingExecutor") .field("worker_count", &self.worker_count()) .field("max_queued_jobs", &self.inner.max_queued_jobs) .field("max_bytes", &self.inner.max_bytes) .field("reserved_bytes", &self.reserved_bytes()) .finish() } } impl BlockingExecutor { fn new( config: &RuntimeConfig, resources: Arc, metrics: RuntimeMetrics, admission_open: Arc, admission_gate: Arc>, ) -> Result { let (sender, receiver) = mpsc::sync_channel::(config.max_queued_blocking_jobs); let receiver = Arc::new(Mutex::new(receiver)); let executor_state = Arc::new(BlockingExecutorState { metrics, queued: AtomicUsize::new(0), active: AtomicUsize::new(0), }); let mut workers = Vec::with_capacity(config.blocking_worker_threads); for index in 0..config.blocking_worker_threads { let receiver = Arc::clone(&receiver); let executor_state = Arc::clone(&executor_state); // AGENTOS_THREAD_SITE: blocking-executor-worker let worker = thread::Builder::new() .name(format!("agentos-blocking-{index}")) .spawn(move || loop { let job = match receiver.lock() { Ok(receiver) => receiver.recv(), Err(_) => { eprintln!( "ERR_AGENTOS_BLOCKING_QUEUE_POISONED: blocking job receiver lock poisoned" ); break; } }; match job { Ok(job) => { decrement_saturating(&executor_state.queued); executor_state.active.fetch_add(1, Ordering::Relaxed); executor_state.metrics.observe_executor( ExecutorMetricClass::Blocking, executor_state.active.load(Ordering::Relaxed), executor_state.queued.load(Ordering::Relaxed), ); let BlockingJob { operation, _slot, _bytes, } = job; if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { operation(_slot, _bytes) })) .is_err() { eprintln!( "ERR_AGENTOS_BLOCKING_JOB_PANIC: blocking job panicked" ); } decrement_saturating(&executor_state.active); executor_state.metrics.observe_executor( ExecutorMetricClass::Blocking, executor_state.active.load(Ordering::Relaxed), executor_state.queued.load(Ordering::Relaxed), ); } Err(_) => break, } }) .map_err(|error| { RuntimeBuildError(format!( "ERR_AGENTOS_BLOCKING_WORKER_START: failed to start blocking worker {index}: {error}" )) })?; workers.push(worker); } Ok(Self { inner: Arc::new(BlockingExecutorInner { sender: Mutex::new(Some(sender)), workers: Mutex::new(workers), max_queued_jobs: config.max_queued_blocking_jobs, max_bytes: config.max_blocking_job_bytes, state: executor_state, }), resources, admission_open, admission_gate, }) } pub fn worker_count(&self) -> usize { self.inner .workers .lock() .map(|workers| workers.len()) .unwrap_or(0) } pub fn reserved_bytes(&self) -> usize { self.resources.usage(ResourceClass::ExecutorBytes).used } /// Share the fixed workers and queue while charging admission to a child /// ledger (which atomically charges its process parent too). pub fn scoped( &self, resources: Arc, admission_open: Arc, admission_gate: Arc>, ) -> Self { Self { inner: Arc::clone(&self.inner), resources, admission_open, admission_gate, } } pub fn submit(&self, reserved_bytes: usize, operation: F) -> Result<(), BlockingJobError> where F: FnOnce() + Send + 'static, { let (admission, _slot, _bytes) = self.reserve(reserved_bytes)?; let job = BlockingJob { operation: Box::new(move |_slot, _bytes| operation()), _slot, _bytes, }; let result = self.try_enqueue(job); drop(admission); result } pub async fn run( &self, reserved_bytes: usize, operation: F, ) -> Result where T: Send + 'static, F: FnOnce() -> T + Send + 'static, { let (admission, _slot, _bytes) = self.reserve(reserved_bytes)?; let (result_tx, result_rx) = tokio::sync::oneshot::channel(); let job = BlockingJob { operation: Box::new(move |slot, bytes| { let result = operation(); // Publish completion only after accounting is reconciled. A // caller observing the result must never still see the job's // reservations charged. drop(slot); drop(bytes); if result_tx.send(result).is_err() { eprintln!( "ERR_AGENTOS_BLOCKING_RESULT_DROPPED: asynchronous caller stopped waiting" ); } }), _slot, _bytes, }; let enqueue_result = self.try_enqueue(job); drop(admission); enqueue_result?; result_rx.await.map_err(|_| BlockingJobError::WorkerDropped) } pub fn run_sync( &self, reserved_bytes: usize, timeout: Duration, operation: F, ) -> Result where T: Send + 'static, F: FnOnce() -> T + Send + 'static, { let (admission, _slot, _bytes) = self.reserve(reserved_bytes)?; let (result_tx, result_rx) = mpsc::sync_channel(1); let job = BlockingJob { operation: Box::new(move |slot, bytes| { let result = operation(); // Synchronous callers can run immediately after recv returns, // so release admission before making that result observable. drop(slot); drop(bytes); if result_tx.send(result).is_err() { eprintln!( "ERR_AGENTOS_BLOCKING_RESULT_DROPPED: synchronous caller stopped waiting" ); } }), _slot, _bytes, }; let enqueue_result = self.try_enqueue(job); drop(admission); enqueue_result?; result_rx .recv_timeout(timeout) .map_err(|error| match error { mpsc::RecvTimeoutError::Timeout => BlockingJobError::TimedOut { timeout }, mpsc::RecvTimeoutError::Disconnected => BlockingJobError::WorkerDropped, }) } fn reserve( &self, requested: usize, ) -> Result<(std::sync::MutexGuard<'_, ()>, Reservation, Reservation), BlockingJobError> { // Keep admission open through the queue insertion. Close uses the same // gate, so a stale clone cannot enqueue blocking work after close // returns and teardown has begun reconciling the VM ledger. let admission = self .admission_gate .lock() .map_err(|_| BlockingJobError::ShuttingDown)?; if !self.admission_open.load(Ordering::Acquire) { return Err(BlockingJobError::ShuttingDown); } let slot = self .resources .reserve(ResourceClass::ExecutorSlots, 1) .map_err(BlockingJobError::ResourceLimit)?; let bytes = self .resources .reserve(ResourceClass::ExecutorBytes, requested) .map_err(BlockingJobError::ResourceLimit)?; Ok((admission, slot, bytes)) } fn try_enqueue(&self, job: BlockingJob) -> Result<(), BlockingJobError> { let sender = self .inner .sender .lock() .map_err(|_| BlockingJobError::ShuttingDown)?; let sender = sender.as_ref().ok_or(BlockingJobError::ShuttingDown)?; self.record_enqueued(); if let Err(error) = sender.try_send(job) { self.record_enqueue_failed(); return Err(match error { mpsc::TrySendError::Full(_) => BlockingJobError::Capacity { limit: self.inner.max_queued_jobs, }, mpsc::TrySendError::Disconnected(_) => BlockingJobError::ShuttingDown, }); } Ok(()) } fn record_enqueued(&self) { self.inner.state.queued.fetch_add(1, Ordering::Relaxed); self.observe_executor(); } fn record_enqueue_failed(&self) { decrement_saturating(&self.inner.state.queued); self.observe_executor(); } fn observe_executor(&self) { self.inner.state.metrics.observe_executor( ExecutorMetricClass::Blocking, self.inner.state.active.load(Ordering::Relaxed), self.inner.state.queued.load(Ordering::Relaxed), ); } } fn decrement_saturating(counter: &AtomicUsize) { let mut current = counter.load(Ordering::Relaxed); while current != 0 { match counter.compare_exchange_weak( current, current - 1, Ordering::Relaxed, Ordering::Relaxed, ) { Ok(_) => return, Err(observed) => current = observed, } } } #[derive(Clone, Debug)] pub struct RuntimeContext { handle: tokio::runtime::Handle, blocking: BlockingExecutor, resources: Arc, tasks: TaskSupervisor, metrics: RuntimeMetrics, fairness: FairWorkBroker, terminal_failure: Arc>>, task_poll_watchdog: Duration, max_active_vm_executors: usize, vm_executor_teardown_timeout: Duration, blocking_job_timeout: Duration, admission_open: Arc, admission_closed: Arc, next_vm_generation: Arc, default_owner: TaskOwner, } impl RuntimeContext { pub fn handle(&self) -> &tokio::runtime::Handle { &self.handle } pub fn blocking(&self) -> &BlockingExecutor { &self.blocking } /// Aggregate process accountant shared by every VM and trusted subsystem. pub fn resources(&self) -> &Arc { &self.resources } pub fn tasks(&self) -> &TaskSupervisor { &self.tasks } pub fn metrics(&self) -> &RuntimeMetrics { &self.metrics } pub fn max_active_vm_executors(&self) -> usize { self.max_active_vm_executors } pub fn vm_executor_teardown_timeout(&self) -> Duration { self.vm_executor_teardown_timeout } pub fn blocking_job_timeout(&self) -> Duration { self.blocking_job_timeout } pub fn fairness(&self) -> &FairWorkBroker { &self.fairness } /// Allocate an identity in the same process-wide namespace as the shared /// fairness broker. Every sidecar facade using this runtime must draw from /// this counter; per-facade counters can collide after an earlier VM retires. pub fn allocate_vm_generation(&self) -> Result { self.next_vm_generation .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| { current.checked_add(1) }) .map(|previous| previous + 1) .map_err(|_| { RuntimeBuildError(String::from( "ERR_AGENTOS_VM_GENERATION_EXHAUSTED: process VM generation counter overflowed", )) }) } /// VM generation bound to this admission scope, when this is a VM-scoped /// context. Fairness keys use this generation rather than a capability's /// per-registry generation, which restarts for every VM. pub fn vm_generation(&self) -> Option { match &self.default_owner { TaskOwner::Vm { generation } => Some(*generation), _ => None, } } /// First failed or panicked task in this process/VM admission scope. /// This is a durable lifecycle latch, not a drainable telemetry queue. pub fn terminal_failure(&self) -> Option { self.terminal_failure .lock() .map(|failure| failure.clone()) .unwrap_or_else(|_| { eprintln!( "ERR_AGENTOS_TASK_FAILURE_LATCH_POISONED: scope={}", self.resources.scope() ); Some(TaskTerminalReport { class: TaskClass::Runtime, owner: self.default_owner.clone(), scope: self.resources.scope().to_owned(), reason: TaskTerminalReason::Panicked, }) }) } /// Permanently close task and blocking-job admission for this context /// scope and every clone of it. Existing admitted work retains ownership /// until its normal terminal path releases accounting. pub fn close_admission(&self) { // The supervisor owns the gate shared by task and blocking admission. // Closing through it linearizes this transition against both paths. self.tasks.close_admission(); self.admission_closed.notify_waiters(); } pub fn admission_is_open(&self) -> bool { self.admission_open.load(Ordering::Acquire) } /// Resolve when this admission scope begins teardown. /// /// The notification is armed before checking the atomic flag so a close /// between observation and await cannot be lost. Each VM-scoped context /// owns a distinct signal; closing one VM never cancels process or sibling /// VM work. pub async fn admission_closed(&self) { loop { let notified = self.admission_closed.notified(); if !self.admission_is_open() { return; } notified.await; } } /// Create a VM-scoped admission view without creating another runtime, /// scheduler, queue, or worker. pub fn scoped(&self, resources: Arc) -> Self { let admission_open = Arc::new(AtomicBool::new(true)); let admission_gate = Arc::new(Mutex::new(())); Self { handle: self.handle.clone(), blocking: self.blocking.scoped( Arc::clone(&resources), Arc::clone(&admission_open), Arc::clone(&admission_gate), ), resources: Arc::clone(&resources), tasks: self .tasks .scoped(resources, Arc::clone(&admission_open), admission_gate), metrics: self.metrics.clone(), fairness: self.fairness.clone(), terminal_failure: Arc::new(Mutex::new(None)), task_poll_watchdog: self.task_poll_watchdog, max_active_vm_executors: self.max_active_vm_executors, vm_executor_teardown_timeout: self.vm_executor_teardown_timeout, blocking_job_timeout: self.blocking_job_timeout, admission_open, admission_closed: Arc::new(tokio::sync::Notify::new()), next_vm_generation: Arc::clone(&self.next_vm_generation), default_owner: self.default_owner.clone(), } } pub fn scoped_for_vm(&self, resources: Arc, generation: u64) -> Self { let mut scoped = self.scoped(resources); scoped.default_owner = TaskOwner::Vm { generation }; scoped } pub fn spawn( &self, class: TaskClass, future: F, ) -> Result, TaskSpawnError> where F: Future + Send + 'static, F::Output: Send + 'static, { let failure_latch = Arc::clone(&self.terminal_failure); let handler = supervision::terminal_handler(move |report| { latch_terminal_failure(&failure_latch, report); }); let mut guard = self .tasks .admit(class, self.default_owner.clone(), Some(handler))?; let future = WatchdogFuture::new(future, class, self.task_poll_watchdog, self.metrics.clone()); Ok(self.handle.spawn(async move { let output = future.await; guard.complete(); output })) } pub fn spawn_result( &self, class: TaskClass, future: F, ) -> Result>, TaskSpawnError> where F: Future> + Send + 'static, T: Send + 'static, E: Send + 'static, { let failure_latch = Arc::clone(&self.terminal_failure); let handler = supervision::terminal_handler(move |report| { latch_terminal_failure(&failure_latch, report); }); let mut guard = self .tasks .admit(class, self.default_owner.clone(), Some(handler))?; let future = WatchdogFuture::new(future, class, self.task_poll_watchdog, self.metrics.clone()); Ok(self.handle.spawn(async move { let output = future.await; if output.is_ok() { guard.complete(); } else { guard.fail(); } output })) } /// Spawn work with an explicit lifecycle owner. The terminal hook runs for /// every exit reason after task accounting is reconciled and outside the /// supervisor lock, so it can fail/close the owner without lock inversion. pub fn spawn_owned( &self, class: TaskClass, owner: TaskOwner, on_terminal: H, future: F, ) -> Result, TaskSpawnError> where F: Future + Send + 'static, F::Output: Send + 'static, H: Fn(&TaskTerminalReport) + Send + Sync + 'static, { let failure_latch = Arc::clone(&self.terminal_failure); let handler = supervision::terminal_handler(move |report| { latch_terminal_failure(&failure_latch, report); on_terminal(report); }); let mut guard = self.tasks.admit(class, owner, Some(handler))?; let future = WatchdogFuture::new(future, class, self.task_poll_watchdog, self.metrics.clone()); Ok(self.handle.spawn(async move { let output = future.await; guard.complete(); output })) } pub fn spawn_owned_result( &self, class: TaskClass, owner: TaskOwner, on_terminal: H, future: F, ) -> Result>, TaskSpawnError> where F: Future> + Send + 'static, T: Send + 'static, E: Send + 'static, H: Fn(&TaskTerminalReport) + Send + Sync + 'static, { let failure_latch = Arc::clone(&self.terminal_failure); let handler = supervision::terminal_handler(move |report| { latch_terminal_failure(&failure_latch, report); on_terminal(report); }); let mut guard = self.tasks.admit(class, owner, Some(handler))?; let future = WatchdogFuture::new(future, class, self.task_poll_watchdog, self.metrics.clone()); Ok(self.handle.spawn(async move { let output = future.await; if output.is_ok() { guard.complete(); } else { guard.fail(); } output })) } } fn latch_terminal_failure(latch: &Mutex>, report: &TaskTerminalReport) { if !matches!( report.reason, TaskTerminalReason::Failed | TaskTerminalReason::Panicked ) { return; } let mut failure = latch.lock().unwrap_or_else(|poisoned| { eprintln!("ERR_AGENTOS_TASK_FAILURE_LATCH_POISONED: recovering terminal failure"); poisoned.into_inner() }); if failure.is_none() { *failure = Some(report.clone()); } } struct WatchdogFuture { inner: Pin>, class: TaskClass, threshold: Duration, metrics: RuntimeMetrics, reported: bool, } impl WatchdogFuture { fn new(future: F, class: TaskClass, threshold: Duration, metrics: RuntimeMetrics) -> Self { Self { inner: Box::pin(future), class, threshold, metrics, reported: false, } } } impl Future for WatchdogFuture { type Output = F::Output; fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll { let this = self.as_mut().get_mut(); let started = Instant::now(); let result = this.inner.as_mut().poll(context); let elapsed = started.elapsed(); if elapsed >= this.threshold { this.metrics .record_watchdog(WatchdogMetric::LongTaskPoll, elapsed); if !this.reported { this.reported = true; let message = format!( "task_class={:?} poll_ms={} threshold_ms={}", this.class, elapsed.as_millis(), this.threshold.as_millis() ); this.metrics.emit_stderr_fallback(TelemetryFallback { severity: TelemetrySeverity::Warning, code: TelemetryFallbackCode::RuntimeWorkerStall, subsystem: TelemetrySubsystem::Runtime, message: &message, }); } } result } } pub struct SidecarRuntime { config: RuntimeConfig, runtime: tokio::runtime::Runtime, context: RuntimeContext, } static PROCESS_RUNTIME: OnceLock> = OnceLock::new(); impl SidecarRuntime { fn build(config: RuntimeConfig) -> Result { config.validate()?; let mut resource_limits = config.resources.limits(); resource_limits.push(( ResourceClass::ExecutorSlots, ResourceLimit::new(config.max_blocking_jobs, "runtime.blocking.maxJobs"), )); resource_limits.push(( ResourceClass::ExecutorBytes, ResourceLimit::new( config.max_blocking_job_bytes, "runtime.blocking.maxQueuedBytes", ), )); let metrics = RuntimeMetrics::new(); let fairness = FairWorkBroker::new(config.fairness.scheduler_config(), metrics.clone()) .map_err(|error| { RuntimeBuildError(format!( "ERR_AGENTOS_RUNTIME_FAIRNESS_START: failed to build process fairness broker: {error}" )) })?; let resources = Arc::new(ResourceLedger::root_with_metrics( "sidecar-process", resource_limits, metrics.clone(), )); let admission_open = Arc::new(AtomicBool::new(true)); let admission_gate = Arc::new(Mutex::new(())); let blocking = BlockingExecutor::new( &config, Arc::clone(&resources), metrics.clone(), Arc::clone(&admission_open), Arc::clone(&admission_gate), )?; let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(config.worker_threads) .thread_name_fn(|| { static NEXT_WORKER: AtomicUsize = AtomicUsize::new(0); format!( "agentos-runtime-{}", NEXT_WORKER.fetch_add(1, Ordering::Relaxed) ) }) .on_thread_start(|| IS_AGENTOS_RUNTIME_WORKER.with(|marker| marker.set(true))) .on_thread_stop(|| IS_AGENTOS_RUNTIME_WORKER.with(|marker| marker.set(false))) .enable_all() .build() .map_err(|error| { RuntimeBuildError(format!( "ERR_AGENTOS_RUNTIME_START: failed to build process runtime: {error}" )) })?; let context = RuntimeContext { handle: runtime.handle().clone(), blocking, resources: Arc::clone(&resources), tasks: TaskSupervisor::new( resources, metrics.clone(), Arc::clone(&admission_open), admission_gate, config.max_terminal_task_reports, ), metrics, fairness, terminal_failure: Arc::new(Mutex::new(None)), task_poll_watchdog: Duration::from_millis(config.task_poll_watchdog_ms), max_active_vm_executors: config.max_active_vm_executors, vm_executor_teardown_timeout: Duration::from_millis( config.vm_executor_teardown_timeout_ms, ), blocking_job_timeout: Duration::from_millis(config.blocking_job_timeout_ms), admission_open, admission_closed: Arc::new(tokio::sync::Notify::new()), next_vm_generation: Arc::new(AtomicU64::new(0)), default_owner: TaskOwner::Process, }; Ok(Self { config, runtime, context, }) } /// Return the one Tokio runtime owned by this sidecar process. /// /// The first caller fixes the process topology. A later caller requesting a /// different topology receives a typed configuration error instead of /// silently creating a subsystem- or VM-local runtime. pub fn process(config: &RuntimeConfig) -> Result<&'static Self, RuntimeBuildError> { match PROCESS_RUNTIME.get_or_init(|| Self::build(config.clone())) { Ok(runtime) if &runtime.config == config => Ok(runtime), Ok(runtime) => Err(RuntimeBuildError(format!( "ERR_AGENTOS_RUNTIME_ALREADY_CONFIGURED: process runtime uses {:?}, requested {:?}", runtime.config, config ))), Err(error) => Err(error.clone()), } } /// Obtain the runtime already constructed by the process entrypoint. /// Subsystems must never silently select a default topology: doing so can /// race trusted configuration and make the first incidental caller own the /// process scheduler. pub fn process_context() -> Result { match PROCESS_RUNTIME.get() { Some(Ok(runtime)) => Ok(runtime.context()), Some(Err(error)) => Err(error.clone()), #[cfg(test)] None => Self::process(&RuntimeConfig::default()).map(Self::context), #[cfg(not(test))] None => Err(RuntimeBuildError(String::from( "ERR_AGENTOS_RUNTIME_NOT_INITIALIZED: the process entrypoint must construct SidecarRuntime before starting subsystems", ))), } } pub fn context(&self) -> RuntimeContext { self.context.clone() } pub fn block_on(&self, future: F) -> F::Output { self.runtime.block_on(future) } } #[cfg(test)] mod tests { use super::*; #[test] fn process_runtime_bounds_every_resource_class_by_default() { let runtime = SidecarRuntime::build(RuntimeConfig::default()).expect("build runtime"); for resource in ResourceClass::ALL { let usage = runtime.context().resources().usage(resource); assert_eq!(usage.used, 0, "{} starts charged", resource.name()); assert!( usage.limit.is_some_and(|limit| limit > 0), "{} has no positive process limit", resource.name() ); } } #[test] fn vm_generation_allocator_is_shared_by_scoped_contexts() { let runtime = SidecarRuntime::build(RuntimeConfig::default()).expect("build runtime"); let process = runtime.context(); let resources = Arc::new(ResourceLedger::root("vm-generation-test", [])); let scoped = process.scoped(Arc::clone(&resources)); let first = process .allocate_vm_generation() .expect("allocate process generation"); let second = scoped .allocate_vm_generation() .expect("allocate scoped generation"); assert_eq!(second, first + 1); } #[test] fn validates_nonzero_runtime_limits() { let error = RuntimeConfig { worker_threads: 0, ..RuntimeConfig::default() } .validate() .expect_err("zero worker count must be rejected"); assert!(error.to_string().contains("runtime.workerThreads")); let error = RuntimeConfig { max_terminal_task_reports: 0, ..RuntimeConfig::default() } .validate() .expect_err("zero terminal-report capacity must be rejected"); assert!(error .to_string() .contains("runtime.tasks.maxTerminalReports")); let error = RuntimeConfig { max_active_vm_executors: 0, ..RuntimeConfig::default() } .validate() .expect_err("zero VM executor capacity must be rejected"); assert!(error.to_string().contains("runtime.executor.maxActiveVms")); let error = RuntimeConfig { vm_executor_teardown_timeout_ms: 0, ..RuntimeConfig::default() } .validate() .expect_err("zero VM executor teardown timeout must be rejected"); assert!(error .to_string() .contains("runtime.executor.teardownTimeoutMs")); let error = RuntimeConfig { blocking_job_timeout_ms: 0, ..RuntimeConfig::default() } .validate() .expect_err("zero blocking-job timeout must be rejected"); assert!(error.to_string().contains("runtime.blocking.jobTimeoutMs")); let error = RuntimeConfig { protocol: RuntimeProtocolConfig { max_ingress_bytes: 0, ..RuntimeProtocolConfig::default() }, ..RuntimeConfig::default() } .validate() .expect_err("zero protocol byte capacity must be rejected"); assert!(error .to_string() .contains("runtime.protocol.maxIngressBytes")); } #[test] fn blocking_executor_enforces_byte_reservations() { let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 1, blocking_worker_threads: 1, max_queued_blocking_jobs: 1, max_blocking_job_bytes: 8, ..RuntimeConfig::default() }) .expect("build runtime"); let blocking = runtime.context().blocking().clone(); let error = runtime .block_on(blocking.run(9, || 1usize)) .expect_err("oversize blocking job must be rejected"); assert!(matches!(error, BlockingJobError::ResourceLimit(_))); assert_eq!(blocking.reserved_bytes(), 0); } #[test] fn blocking_executor_runs_on_fixed_workers_and_releases_bytes() { let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 1, blocking_worker_threads: 2, max_queued_blocking_jobs: 2, max_blocking_job_bytes: 32, ..RuntimeConfig::default() }) .expect("build runtime"); let blocking = runtime.context().blocking().clone(); let worker_name = runtime .block_on(blocking.run(4, || { thread::current().name().unwrap_or_default().to_owned() })) .expect("blocking job result"); assert!(worker_name.starts_with("agentos-blocking-")); assert_eq!(blocking.worker_count(), 2); assert_eq!(blocking.reserved_bytes(), 0); let metrics = runtime.context().metrics().snapshot(); assert_eq!( metrics.buffers[metrics::BufferMetricClass::Executor.index()].current, 0 ); assert_eq!( metrics.buffers[metrics::BufferMetricClass::Executor.index()].high_water, 4 ); assert!( metrics.executors[ExecutorMetricClass::Blocking.index()] .active .high_water >= 1 ); } #[test] fn blocking_executor_supports_bounded_synchronous_callers() { let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 1, blocking_worker_threads: 1, max_queued_blocking_jobs: 1, max_blocking_job_bytes: 32, ..RuntimeConfig::default() }) .expect("build runtime"); let blocking = runtime.context().blocking().clone(); let value = blocking .run_sync(4, Duration::from_secs(1), || 42usize) .expect("synchronous blocking job result"); assert_eq!(value, 42); assert_eq!(blocking.reserved_bytes(), 0); } #[test] fn task_supervisor_reports_every_terminal_reason_and_reconciles() { let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 1, blocking_worker_threads: 1, max_queued_blocking_jobs: 1, ..RuntimeConfig::default() }) .expect("build runtime"); let context = runtime.context(); runtime.block_on(async { context .spawn(TaskClass::Runtime, async {}) .expect("completed task admission") .await .expect("completed task join"); let failed = context .spawn_result(TaskClass::Runtime, async { Err::<(), _>("failed") }) .expect("failed task admission") .await .expect("failed task join"); assert_eq!(failed, Err("failed")); let panicked = context .spawn(TaskClass::Runtime, async { panic!("task panic fixture") }) .expect("panicked task admission") .await; assert!(panicked.expect_err("task must panic").is_panic()); let cancelled = context .spawn(TaskClass::Runtime, std::future::pending::<()>()) .expect("cancelled task admission"); cancelled.abort(); assert!(cancelled .await .expect_err("task must cancel") .is_cancelled()); }); let snapshot = context.tasks().snapshot(TaskClass::Runtime); assert_eq!(snapshot.active, 0); assert_eq!(snapshot.completed, 1); assert_eq!(snapshot.failed, 1); assert_eq!(snapshot.panicked, 1); assert_eq!(snapshot.cancelled, 1); assert_eq!(context.resources().usage(ResourceClass::Tasks).used, 0); let terminal_failure = context .terminal_failure() .expect("failed task must latch owner failure"); assert_eq!(terminal_failure.reason, TaskTerminalReason::Failed); assert_eq!(terminal_failure.owner, TaskOwner::Process); let metric = context.metrics().snapshot().task(TaskClass::Runtime); assert_eq!(metric.active, 0); assert_eq!(metric.terminal_count(TaskTerminalReason::Completed), 1); assert_eq!(metric.terminal_count(TaskTerminalReason::Failed), 1); assert_eq!(metric.terminal_count(TaskTerminalReason::Panicked), 1); assert_eq!(metric.terminal_count(TaskTerminalReason::Cancelled), 1); } #[test] fn long_task_poll_records_watchdog_once_without_dynamic_labels() { let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 1, blocking_worker_threads: 1, max_queued_blocking_jobs: 1, task_poll_watchdog_ms: 1, ..RuntimeConfig::default() }) .expect("build runtime"); let context = runtime.context(); runtime.block_on(async { context .spawn(TaskClass::Runtime, async { std::thread::sleep(Duration::from_millis(5)); }) .expect("task admission") .await .expect("task join"); }); let snapshot = context.metrics().snapshot(); let watchdog = snapshot.watchdogs[WatchdogMetric::LongTaskPoll.index()]; assert_eq!(watchdog.events, 1); assert!(watchdog.max_stall_micros >= 1_000); assert_eq!( snapshot.stderr_fallbacks[TelemetrySeverity::Warning.index()], 1 ); } #[test] fn vm_scoped_context_shares_workers_and_charges_both_ledgers() { let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 1, blocking_worker_threads: 1, max_queued_blocking_jobs: 1, max_blocking_job_bytes: 32, ..RuntimeConfig::default() }) .expect("build runtime"); let process = Arc::clone(runtime.context().resources()); let vm_ledger = Arc::new(ResourceLedger::child( "vm=1 generation=1", [ ( ResourceClass::Tasks, ResourceLimit::new(1, "limits.reactor.maxTasks"), ), ( ResourceClass::ExecutorSlots, ResourceLimit::new(1, "limits.blocking.maxJobs"), ), ( ResourceClass::ExecutorBytes, ResourceLimit::new(8, "limits.blocking.maxQueuedBytes"), ), ], Arc::clone(&process), )); let scoped = runtime.context().scoped(Arc::clone(&vm_ledger)); assert_eq!(scoped.blocking().worker_count(), 1); let value = runtime .block_on(scoped.blocking().run(8, || 42usize)) .expect("VM-scoped blocking job"); assert_eq!(value, 42); assert!(vm_ledger.is_zero()); assert_eq!( process.usage(ResourceClass::ExecutorBytes).used, 0, "child release must reconcile the process parent" ); runtime.block_on(async { scoped .spawn(TaskClass::Vm, async {}) .expect("VM task admission") .await .expect("VM task join"); }); assert!(vm_ledger.is_zero()); } fn assert_vm_generation_churn_reconciles(generation_count: u64, logical_vm_count: u64) { use crate::capability::{CapabilityBackend, CapabilityKind, CapabilityRegistry}; assert!(generation_count > 0); assert!(logical_vm_count > 0); let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 2, blocking_worker_threads: 2, max_queued_blocking_jobs: 4, max_blocking_jobs: 8, max_blocking_job_bytes: 1024, ..RuntimeConfig::default() }) .expect("build runtime"); let process = Arc::clone(runtime.context().resources()); runtime.block_on(async { for generation in 1..=generation_count { let vm_index = (generation - 1) % logical_vm_count; let resources = Arc::new(ResourceLedger::child( format!("vm=churn-{vm_index} generation={generation}"), [ ( ResourceClass::Capabilities, ResourceLimit::new(2, "limits.reactor.maxCapabilities"), ), ( ResourceClass::ReadyHandles, ResourceLimit::new(2, "limits.reactor.maxReadyHandles"), ), ( ResourceClass::Sockets, ResourceLimit::new(2, "limits.resources.maxSockets"), ), ( ResourceClass::Connections, ResourceLimit::new(2, "limits.resources.maxConnections"), ), ( ResourceClass::Tasks, ResourceLimit::new(2, "limits.reactor.maxTasks"), ), ( ResourceClass::ExecutorSlots, ResourceLimit::new(4, "limits.reactor.maxBlockingJobs"), ), ( ResourceClass::ExecutorBytes, ResourceLimit::new(64, "limits.reactor.maxBlockingBytes"), ), ], Arc::clone(&process), )); let context = runtime .context() .scoped_for_vm(Arc::clone(&resources), generation); let capabilities = CapabilityRegistry::new(generation, Arc::clone(&resources)); let lease = capabilities .reserve(CapabilityKind::TcpSocket) .expect("reserve churn socket before allocation") .commit(CapabilityBackend::Kernel { socket_id: generation, }) .expect("commit churn socket"); let task = context .spawn(TaskClass::Socket, async { tokio::task::yield_now().await }) .expect("admit churn task"); let blocking = context .blocking() .run(16, move || generation) .await .expect("fixed blocking worker result"); assert_eq!(blocking, generation); task.await.expect("churn task join"); let capability_id = lease.id(); let turn = context .fairness() .acquire(generation, capability_id, FairBudget::new(1, 64)) .await .expect("churn fairness turn"); turn.complete(FairBudget::new(1, 16), false) .expect("complete churn fairness turn"); drop(lease); capabilities .close_admission() .expect("close churn capability admission"); context.close_admission(); context.tasks().wait_empty().await; capabilities.wait_empty().await; context .fairness() .retire_capability(generation, capability_id) .expect("retire churn capability fairness state"); context .fairness() .retire_vm(generation) .expect("retire churn VM fairness state"); assert!( resources.is_zero(), "generation {generation} leaked accounting" ); assert!(resources.integrity_ok()); } }); assert!(process.is_zero(), "VM churn drifted process accounting"); assert!(process.integrity_ok()); assert_eq!(runtime.context().tasks().active_total(), 0); } #[test] fn vm_generation_churn_reconciles_tasks_capabilities_fairness_and_bytes() { // Fast default-CI safeguard: enough generations to cross logical VM // identities repeatedly while keeping this test deterministic. assert_vm_generation_churn_reconciles(256, 8); } #[test] #[ignore = "expensive: multi-VM runtime accounting soak; run explicitly with --ignored"] fn multi_vm_generation_soak_has_no_accounting_or_scheduler_drift() { // The same invariant as the default regression, at a scale intended to // expose cumulative reservation, tombstone, and task-census drift. assert_vm_generation_churn_reconciles(50_000, 64); } #[test] fn closing_vm_context_rejects_stale_task_and_blocking_clones() { let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 1, blocking_worker_threads: 1, max_queued_blocking_jobs: 1, ..RuntimeConfig::default() }) .expect("build runtime"); let vm_ledger = Arc::new(ResourceLedger::child( "vm-generation-77", [ ( ResourceClass::Tasks, ResourceLimit::new(4, "limits.reactor.maxTasks"), ), ( ResourceClass::ExecutorSlots, ResourceLimit::new(4, "limits.reactor.maxBlockingJobs"), ), ( ResourceClass::ExecutorBytes, ResourceLimit::new(64, "limits.reactor.maxBlockingBytes"), ), ], Arc::clone(runtime.context().resources()), )); let scoped = runtime.context().scoped_for_vm(Arc::clone(&vm_ledger), 77); let stale = scoped.clone(); scoped.close_admission(); let task_error = stale .spawn(TaskClass::Vm, async {}) .expect_err("closed VM must reject stale task clone"); assert!(matches!(task_error, TaskSpawnError::AdmissionClosed { .. })); let blocking_error = stale .blocking() .submit(1, || {}) .expect_err("closed VM must reject stale blocking clone"); assert_eq!(blocking_error, BlockingJobError::ShuttingDown); assert!(vm_ledger.is_zero()); } #[test] fn close_admission_linearizes_task_and_blocking_rejection() { let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 1, blocking_worker_threads: 1, max_queued_blocking_jobs: 1, ..RuntimeConfig::default() }) .expect("build runtime"); let vm_ledger = Arc::new(ResourceLedger::child( "vm-generation-88", [ ( ResourceClass::Tasks, ResourceLimit::new(2, "limits.reactor.maxTasks"), ), ( ResourceClass::ExecutorSlots, ResourceLimit::new(2, "limits.reactor.maxBlockingJobs"), ), ( ResourceClass::ExecutorBytes, ResourceLimit::new(8, "limits.reactor.maxBlockingBytes"), ), ], Arc::clone(runtime.context().resources()), )); let scoped = runtime.context().scoped_for_vm(Arc::clone(&vm_ledger), 88); // Blocking admission and task admission share this exact gate. Holding // it models an admission operation at its linearization point. let gate = Arc::clone(&scoped.blocking.admission_gate); let held = gate.lock().expect("admission gate"); let closing = scoped.clone(); let (started_tx, started_rx) = mpsc::sync_channel(1); let (closed_tx, closed_rx) = mpsc::sync_channel(1); let close_thread = thread::spawn(move || { started_tx.send(()).expect("publish close start"); closing.close_admission(); closed_tx.send(()).expect("publish close completion"); }); started_rx.recv().expect("close started"); assert_eq!( closed_rx.recv_timeout(Duration::from_millis(20)), Err(mpsc::RecvTimeoutError::Timeout), "close must wait for the shared admission linearization gate" ); drop(held); closed_rx .recv_timeout(Duration::from_secs(1)) .expect("close completed after admission released"); close_thread.join().expect("close thread"); let task_error = scoped .spawn(TaskClass::Vm, async {}) .expect_err("task admission after linearized close must fail"); assert!(matches!(task_error, TaskSpawnError::AdmissionClosed { .. })); let blocking_error = scoped .blocking() .submit(1, || {}) .expect_err("blocking admission after linearized close must fail"); assert_eq!(blocking_error, BlockingJobError::ShuttingDown); assert!(vm_ledger.is_zero()); } #[test] fn closing_vm_context_wakes_admitted_readiness_waiters() { let runtime = SidecarRuntime::build(RuntimeConfig { worker_threads: 1, ..RuntimeConfig::default() }) .expect("build runtime"); let resources = Arc::new(ResourceLedger::child( "vm-generation-88", [( ResourceClass::Tasks, ResourceLimit::new(2, "limits.reactor.maxTasks"), )], Arc::clone(runtime.context().resources()), )); let scoped = runtime.context().scoped_for_vm(resources, 88); let waiter_context = scoped.clone(); let (started_tx, started_rx) = std::sync::mpsc::sync_channel(1); let waiter = scoped .spawn(TaskClass::Socket, async move { started_tx.send(()).expect("signal waiter admission"); waiter_context.admission_closed().await; }) .expect("spawn readiness waiter"); started_rx .recv_timeout(Duration::from_secs(1)) .expect("waiter started"); scoped.close_admission(); runtime.context().handle().block_on(async { tokio::time::timeout(Duration::from_secs(1), waiter) .await .expect("close wakes waiter before teardown deadline") .expect("waiter joins") }); assert_eq!(scoped.tasks().active_scoped(), 0); } }