Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion crates/native-sidecar/src/stdio.rs
Original file line number Diff line number Diff line change
Expand Up @@ -747,10 +747,22 @@ fn run_with_optional_control(
extensions: Vec<Box<dyn Extension>>,
control_fd: Option<OwnedFd>,
) -> Result<(), Box<dyn Error>> {
let config = NativeSidecarConfig {
let mut config = NativeSidecarConfig {
compile_cache_root: Some(default_compile_cache_root()),
..NativeSidecarConfig::default()
};
// Operator overrides must land before `SidecarRuntime::process` fixes the
// process topology: the first caller's config is the one the whole process
// keeps, and a later differing config is a typed error, not a re-configure.
config.runtime.apply_env_overrides()?;
// The admitted ceiling is fixed for the life of the process and shared by
// every VM it hosts, so make the effective value observable at startup
// rather than only when an execution is rejected for exceeding it.
tracing::info!(
max_active_guest_executions = config.runtime.max_active_guest_executions,
env = agentos_runtime::MAX_ACTIVE_GUEST_EXECUTIONS_ENV,
"guest execution admission ceiling"
);
let runtime = agentos_runtime::SidecarRuntime::process(&config.runtime)?;
let runtime_context = runtime.context();
// Initialize the embedded V8 runtime + platform now, on the long-lived main
Expand Down
13 changes: 13 additions & 0 deletions crates/native-sidecar/tests/fixtures/limits-inventory.json
Original file line number Diff line number Diff line change
Expand Up @@ -1658,6 +1658,19 @@
"rationale": "Default for the configured process-wide protocol queue bound.",
"wired": "RuntimeConfig.protocol.max_process_events"
},
{
"name": "DEFAULT_MAX_ACTIVE_GUEST_EXECUTIONS",
"path": "crates/runtime/src/lib.rs",
"class": "policy",
"rationale": "Default process-wide cap on concurrently running guest executions; each slot owns one OS thread and one V8 isolate.",
"wired": "RuntimeConfig.max_active_guest_executions"
},
{
"name": "MAX_ACTIVE_GUEST_EXECUTIONS_CEILING",
"path": "crates/runtime/src/lib.rs",
"class": "invariant",
"rationale": "Hard ceiling for the AGENTOS_MAX_ACTIVE_GUEST_EXECUTIONS operator override; admission stays bounded whatever an operator requests."
},
{
"name": "DEFAULT_VM_EXECUTOR_TEARDOWN_TIMEOUT_MS",
"path": "crates/runtime/src/lib.rs",
Expand Down
167 changes: 155 additions & 12 deletions crates/runtime/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,38 @@ 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;

/// Process-wide ceiling on concurrently running guest executions.
///
/// Each admitted execution owns one OS thread and one V8 isolate (thread-affine,
/// so it can never be multiplexed onto a shared pool) capped at
/// `DEFAULT_HEAP_LIMIT_MB`. The binding constraint is therefore threads and
/// memory, NOT CPU: an agent parked on a network read, or a shell blocked in
/// `waitpid`, burns no CPU and still holds its slot for the whole life of the
/// guest process. Deriving this from `available_parallelism()` made the ceiling
/// depend on the host's core count and silently rejected ordinary workloads —
/// a shell plus the command it waits on already needs two slots.
///
/// A fixed default keeps the admitted concurrency identical on every host.
/// Raise it with `AGENTOS_MAX_ACTIVE_GUEST_EXECUTIONS`.
pub const DEFAULT_MAX_ACTIVE_GUEST_EXECUTIONS: usize = 64;

/// Hard ceiling for `AGENTOS_MAX_ACTIVE_GUEST_EXECUTIONS`.
///
/// Admission stays bounded regardless of what an operator asks for: at this
/// ceiling the process still reserves 1024 OS threads and isolates, which is
/// past the point where a host is thread- and memory-bound. Requests above it
/// are a typed configuration error, never a silent clamp.
pub const MAX_ACTIVE_GUEST_EXECUTIONS_CEILING: usize = 1_024;

/// Operator override for [`DEFAULT_MAX_ACTIVE_GUEST_EXECUTIONS`].
///
/// Read once, by the process entrypoint, before any VM exists. The value is
/// process topology (see [`SidecarRuntime::process`]) and is deliberately not a
/// client wire field: one sidecar process is shared by every VM and connection,
/// so no single tenant may rewrite it for its neighbours.
pub const MAX_ACTIVE_GUEST_EXECUTIONS_ENV: &str = "AGENTOS_MAX_ACTIVE_GUEST_EXECUTIONS";

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;
Expand Down Expand Up @@ -417,7 +449,10 @@ impl RuntimeResourceConfig {
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RuntimeConfig {
pub worker_threads: usize,
pub max_active_vm_executors: usize,
/// Process-wide cap on concurrently running guest executions (JavaScript,
/// TypeScript, Python, and WASM alike — every live guest process holds one).
/// See [`DEFAULT_MAX_ACTIVE_GUEST_EXECUTIONS`].
pub max_active_guest_executions: usize,
pub vm_executor_teardown_timeout_ms: u64,
pub blocking_worker_threads: usize,
pub max_blocking_jobs: usize,
Expand All @@ -438,7 +473,7 @@ impl Default for RuntimeConfig {
.unwrap_or(1);
Self {
worker_threads: available.clamp(1, 4),
max_active_vm_executors: available.max(1),
max_active_guest_executions: DEFAULT_MAX_ACTIVE_GUEST_EXECUTIONS,
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,
Expand All @@ -455,12 +490,46 @@ impl Default for RuntimeConfig {
}

impl RuntimeConfig {
/// Apply operator overrides from the process environment.
///
/// Call this from the process entrypoint, before [`SidecarRuntime::process`]
/// fixes the topology. A present-but-unusable value is a hard, typed error
/// naming the variable and its bounds: an operator who asked for a specific
/// admission ceiling must never silently get a different one.
pub fn apply_env_overrides(&mut self) -> Result<(), RuntimeBuildError> {
self.apply_env_overrides_from(|key| std::env::var(key).ok())
}

/// Testable core of [`apply_env_overrides`]. `read` resolves a variable
/// name to its value, mirroring `std::env::var(..).ok()`.
pub fn apply_env_overrides_from(
&mut self,
read: impl Fn(&str) -> Option<String>,
) -> Result<(), RuntimeBuildError> {
let Some(raw) = read(MAX_ACTIVE_GUEST_EXECUTIONS_ENV) else {
return Ok(());
};
let value = raw.trim();
let parsed: usize = value.parse().map_err(|_| {
RuntimeBuildError(format!(
"ERR_AGENTOS_RUNTIME_CONFIG: {MAX_ACTIVE_GUEST_EXECUTIONS_ENV} must be an integer between 1 and {MAX_ACTIVE_GUEST_EXECUTIONS_CEILING}, got {value:?}"
))
})?;
if parsed == 0 || parsed > MAX_ACTIVE_GUEST_EXECUTIONS_CEILING {
return Err(RuntimeBuildError(format!(
"ERR_AGENTOS_RUNTIME_CONFIG: {MAX_ACTIVE_GUEST_EXECUTIONS_ENV} must be between 1 and {MAX_ACTIVE_GUEST_EXECUTIONS_CEILING}, got {parsed}"
)));
}
self.max_active_guest_executions = parsed;
Ok(())
}

pub fn validate(&self) -> Result<(), RuntimeBuildError> {
for (field, value) in [
("runtime.workerThreads", self.worker_threads),
(
"runtime.executor.maxActiveVms",
self.max_active_vm_executors,
"runtime.executor.maxActiveGuestExecutions",
self.max_active_guest_executions,
),
(
"runtime.blocking.workerThreads",
Expand Down Expand Up @@ -1114,7 +1183,7 @@ pub struct RuntimeContext {
fairness: FairWorkBroker,
terminal_failure: Arc<Mutex<Option<TaskTerminalReport>>>,
task_poll_watchdog: Duration,
max_active_vm_executors: usize,
max_active_guest_executions: usize,
vm_executor_teardown_timeout: Duration,
blocking_job_timeout: Duration,
admission_open: Arc<AtomicBool>,
Expand Down Expand Up @@ -1145,8 +1214,8 @@ impl RuntimeContext {
&self.metrics
}

pub fn max_active_vm_executors(&self) -> usize {
self.max_active_vm_executors
pub fn max_active_guest_executions(&self) -> usize {
self.max_active_guest_executions
}

pub fn vm_executor_teardown_timeout(&self) -> Duration {
Expand Down Expand Up @@ -1257,7 +1326,7 @@ impl RuntimeContext {
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,
max_active_guest_executions: self.max_active_guest_executions,
vm_executor_teardown_timeout: self.vm_executor_teardown_timeout,
blocking_job_timeout: self.blocking_job_timeout,
admission_open,
Expand Down Expand Up @@ -1535,7 +1604,7 @@ impl SidecarRuntime {
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,
max_active_guest_executions: config.max_active_guest_executions,
vm_executor_teardown_timeout: Duration::from_millis(
config.vm_executor_teardown_timeout_ms,
),
Expand Down Expand Up @@ -1598,6 +1667,78 @@ impl SidecarRuntime {
mod tests {
use super::*;

fn env_override(value: Option<&str>) -> Result<RuntimeConfig, RuntimeBuildError> {
let mut config = RuntimeConfig::default();
let value = value.map(str::to_owned);
config.apply_env_overrides_from(|key| {
(key == MAX_ACTIVE_GUEST_EXECUTIONS_ENV)
.then(|| value.clone())
.flatten()
})?;
Ok(config)
}

#[test]
fn default_guest_execution_ceiling_does_not_depend_on_host_cpu_count() {
// A CPU-derived ceiling rejected ordinary workloads on small hosts and
// made admitted concurrency differ per machine. Guest executions are
// bounded by threads and memory, not by cores.
assert_eq!(
RuntimeConfig::default().max_active_guest_executions,
DEFAULT_MAX_ACTIVE_GUEST_EXECUTIONS
);
assert!(
DEFAULT_MAX_ACTIVE_GUEST_EXECUTIONS >= 2,
"a shell and the command it waits on already need two slots"
);
}

#[test]
fn absent_guest_execution_override_keeps_the_default() {
let config = env_override(None).expect("absent override is not an error");
assert_eq!(
config.max_active_guest_executions,
DEFAULT_MAX_ACTIVE_GUEST_EXECUTIONS
);
}

#[test]
fn guest_execution_override_applies_and_stays_valid() {
let config = env_override(Some(" 128 ")).expect("surrounding whitespace is accepted");
assert_eq!(config.max_active_guest_executions, 128);
config.validate().expect("override must stay valid");
}

#[test]
fn unusable_guest_execution_override_is_a_typed_error() {
// An operator who asked for a specific ceiling must never silently get a
// different one: no clamping, no falling back to the default.
for value in ["", "many", "0", "-1", "1.5"] {
let error = env_override(Some(value)).expect_err("unusable override must be rejected");
assert!(
error.to_string().contains(MAX_ACTIVE_GUEST_EXECUTIONS_ENV),
"error must name the variable: {error}"
);
}

let above_ceiling = MAX_ACTIVE_GUEST_EXECUTIONS_CEILING + 1;
let error = env_override(Some(&above_ceiling.to_string()))
.expect_err("a request above the hard ceiling must be rejected");
assert!(
error
.to_string()
.contains(&MAX_ACTIVE_GUEST_EXECUTIONS_CEILING.to_string()),
"error must name the ceiling: {error}"
);

let at_ceiling = env_override(Some(&MAX_ACTIVE_GUEST_EXECUTIONS_CEILING.to_string()))
.expect("the ceiling itself is admissible");
assert_eq!(
at_ceiling.max_active_guest_executions,
MAX_ACTIVE_GUEST_EXECUTIONS_CEILING
);
}

#[test]
fn process_runtime_bounds_every_resource_class_by_default() {
let runtime = SidecarRuntime::build(RuntimeConfig::default()).expect("build runtime");
Expand Down Expand Up @@ -1650,12 +1791,14 @@ mod tests {
.contains("runtime.tasks.maxTerminalReports"));

let error = RuntimeConfig {
max_active_vm_executors: 0,
max_active_guest_executions: 0,
..RuntimeConfig::default()
}
.validate()
.expect_err("zero VM executor capacity must be rejected");
assert!(error.to_string().contains("runtime.executor.maxActiveVms"));
.expect_err("zero guest-execution capacity must be rejected");
assert!(error
.to_string()
.contains("runtime.executor.maxActiveGuestExecutions"));

let error = RuntimeConfig {
vm_executor_teardown_timeout_ms: 0,
Expand Down
6 changes: 3 additions & 3 deletions crates/v8-runtime/src/embedded_runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ impl EmbeddedV8Runtime {
// without immediately evicting each other.
let snapshot_cache = Arc::new(SnapshotCache::new(8));
let call_id_router: CallIdRouter = Arc::new(BridgeCallRegistry::with_default_limit());
let configured_max_concurrency = runtime.max_active_vm_executors();
let configured_max_concurrency = runtime.max_active_guest_executions();
let executor_teardown_timeout = runtime.vm_executor_teardown_timeout();
let session_mgr = Arc::new(Mutex::new(SessionManager::new(
max_concurrency.unwrap_or(configured_max_concurrency),
Expand Down Expand Up @@ -682,7 +682,7 @@ pub fn spawn_embedded_runtime_ipc(
let shutdown_stream = host_stream.try_clone()?;
let alive = Arc::new(AtomicBool::new(true));
let alive_for_thread = Arc::clone(&alive);
let max_concurrency = max_concurrency.unwrap_or_else(|| runtime.max_active_vm_executors());
let max_concurrency = max_concurrency.unwrap_or_else(|| runtime.max_active_guest_executions());

// AGENTOS_THREAD_SITE: embedded-v8-dispatch
let join_handle = thread::Builder::new()
Expand Down Expand Up @@ -1175,7 +1175,7 @@ mod tests {
.lock()
.expect("embedded runtime codec test lock poisoned");
let mut config = agentos_runtime::RuntimeConfig {
max_active_vm_executors: 2,
max_active_guest_executions: 2,
vm_executor_teardown_timeout_ms: 31,
..agentos_runtime::RuntimeConfig::default()
};
Expand Down
30 changes: 20 additions & 10 deletions crates/v8-runtime/src/session.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1062,12 +1062,15 @@ impl SessionSlotPermit {
metrics: RuntimeMetrics,
) -> Result<Self, String> {
let (lock, _) = &**control;
let mut active = lock
.lock()
.map_err(|_| String::from("ERR_AGENTOS_VM_EXECUTOR_POISONED: slot lock poisoned"))?;
let mut active = lock.lock().map_err(|_| {
String::from("ERR_AGENTOS_GUEST_EXECUTION_POISONED: slot lock poisoned")
})?;
if *active >= maximum {
return Err(format!(
"ERR_AGENTOS_VM_EXECUTOR_LIMIT: active V8 executors reached limit of {maximum}; raise runtime.executor.maxActiveVms"
"ERR_AGENTOS_GUEST_EXECUTION_LIMIT: concurrently running guest executions reached the process limit of {maximum}; \
every live guest process (JavaScript, TypeScript, Python, or WASM command) holds one slot for its whole lifetime, \
so a parent and the child it waits on need two. Raise runtime.executor.maxActiveGuestExecutions by setting {} on the sidecar process.",
agentos_runtime::MAX_ACTIVE_GUEST_EXECUTIONS_ENV
));
}
*active += 1;
Expand All @@ -1090,10 +1093,12 @@ impl Drop for SessionSlotPermit {
cvar.notify_all();
}
Ok(_) => eprintln!(
"ERR_AGENTOS_VM_EXECUTOR_ACCOUNTING_UNDERFLOW: executor permit released at zero"
"ERR_AGENTOS_GUEST_EXECUTION_ACCOUNTING_UNDERFLOW: executor permit released at zero"
),
Err(_) => {
eprintln!("ERR_AGENTOS_VM_EXECUTOR_POISONED: executor permit could not be released")
eprintln!(
"ERR_AGENTOS_GUEST_EXECUTION_POISONED: executor permit could not be released"
)
}
}
}
Expand Down Expand Up @@ -4005,7 +4010,7 @@ mod tests {
return;
}
let mut config = agentos_runtime::RuntimeConfig {
max_active_vm_executors: 3,
max_active_guest_executions: 3,
vm_executor_teardown_timeout_ms: 23,
..agentos_runtime::RuntimeConfig::default()
};
Expand All @@ -4016,7 +4021,7 @@ mod tests {
let (event_tx, _event_rx) = crossbeam_channel::unbounded();
let router: CallIdRouter = Arc::new(BridgeCallRegistry::with_default_limit());
let mut manager = SessionManager::new(
runtime.max_active_vm_executors(),
runtime.max_active_guest_executions(),
event_tx,
router,
Arc::new(SnapshotCache::new(1)),
Expand Down Expand Up @@ -4169,7 +4174,12 @@ mod tests {
let error = mgr
.create_session("s3".into(), None, None, None)
.expect_err("third executor must be rejected before thread creation");
assert!(error.contains("ERR_AGENTOS_VM_EXECUTOR_LIMIT"));
assert!(error.contains("ERR_AGENTOS_GUEST_EXECUTION_LIMIT"));
// The limit error must name a knob an operator can actually set.
assert!(
error.contains(agentos_runtime::MAX_ACTIVE_GUEST_EXECUTIONS_ENV),
"limit error must say how to raise it: {error}"
);

// Allow threads to acquire slots
std::thread::sleep(std::time::Duration::from_millis(300));
Expand Down Expand Up @@ -4251,7 +4261,7 @@ mod tests {
let error = mgr
.create_session("two-phase".into(), None, None, None)
.expect_err("old generation must retain its executor permit");
assert!(error.contains("ERR_AGENTOS_VM_EXECUTOR_LIMIT"));
assert!(error.contains("ERR_AGENTOS_GUEST_EXECUTION_LIMIT"));
first_shutdown.finish();

mgr.create_session("two-phase".into(), None, None, None)
Expand Down
Loading
Loading