From bb8e9416ce9ea4ede1c996748ffb74600d6bb804 Mon Sep 17 00:00:00 2001 From: Ethon Date: Sat, 10 Oct 2026 23:35:43 +0800 Subject: [PATCH 1/2] fix(core): reject execution waits when a VM is disposed --- crates/client/src/agent_os.rs | 5 + crates/client/src/config.rs | 8 + crates/client/src/error.rs | 12 +- crates/client/src/language_execution.rs | 186 +++++++++- crates/client/src/process.rs | 18 +- crates/client/tests/lifecycle_e2e.rs | 92 ++++- docs/content/docs/processes.mdx | 14 + docs/content/docs/resource-limits.mdx | 9 + packages/core/package.json | 2 +- packages/core/src/agent-os.ts | 117 +++++-- packages/core/src/index.ts | 32 +- packages/core/src/options-schema.ts | 13 +- packages/core/tests/execution-abort.test.ts | 12 +- .../core/tests/execution-disposal.test.ts | 327 ++++++++++++++++++ packages/core/tests/options-schema.test.ts | 16 + secure-exec/src/runtime.ts | 1 + 16 files changed, 800 insertions(+), 64 deletions(-) create mode 100644 packages/core/tests/execution-disposal.test.ts diff --git a/crates/client/src/agent_os.rs b/crates/client/src/agent_os.rs index 54e98ded01..9da2daae21 100644 --- a/crates/client/src/agent_os.rs +++ b/crates/client/src/agent_os.rs @@ -130,6 +130,7 @@ pub struct AgentOs { } pub(crate) struct AgentOsInner { + pub(crate) execution_waits: crate::language_execution::ExecutionWaits, // Transport / connection / VM handle. pub(crate) transport: Arc, pub(crate) connection_id: String, @@ -251,6 +252,8 @@ impl AgentOs { } async fn create_owned(options: AgentOsConfig) -> Result { + let execution_waits = + crate::language_execution::ExecutionWaits::new(options.max_pending_execution_waits)?; let config = Arc::new(options); // 1. Resolve the sidecar handle (shared "default" pool unless configured otherwise) and @@ -661,6 +664,7 @@ impl AgentOs { let cron = Arc::new(CronManager::new(driver)); let inner = AgentOsInner { + execution_waits, transport, connection_id, session_id, @@ -960,6 +964,7 @@ impl AgentOs { return result.clone(); } self.inner.disposed.store(true, Ordering::SeqCst); + self.inner.execution_waits.dispose(); // The `/opt/agentos` projection staging dir is owned + cleaned up by the // sidecar on VM dispose, so the client no longer removes it here. diff --git a/crates/client/src/config.rs b/crates/client/src/config.rs index 566e5eb74a..1ccae125cd 100644 --- a/crates/client/src/config.rs +++ b/crates/client/src/config.rs @@ -78,6 +78,9 @@ pub struct AgentOsConfig { pub permissions: Option, /// Operator-tunable VM limits. Default: sidecar/kernel built-ins. pub limits: Option, + /// Maximum concurrent SDK execution waits per VM. Defaults to 4096. + /// This client-local bound does not change native execution limits. + pub max_pending_execution_waits: Option, /// Sidecar placement/config. Default: shared `default` pool. pub sidecar: Option, /// Absolute path to the `agentos-native-sidecar` binary, resolved from the npm @@ -178,6 +181,11 @@ impl AgentOsConfigBuilder { self } + pub fn max_pending_execution_waits(mut self, limit: usize) -> Self { + self.config.max_pending_execution_waits = Some(limit); + self + } + pub fn limits(mut self, limits: AgentOsLimits) -> Self { self.config.limits = Some(limits); self diff --git a/crates/client/src/error.rs b/crates/client/src/error.rs index 7c1b7b689c..00ff24f14a 100644 --- a/crates/client/src/error.rs +++ b/crates/client/src/error.rs @@ -29,6 +29,14 @@ pub struct ResourceLimitDetails { /// Typed error taxonomy for the client SDK. #[derive(thiserror::Error, Debug, Clone)] pub enum ClientError { + /// The owning VM began shutdown before its execution wait settled. + #[error("ERR_AGENTOS_VM_DISPOSED")] + VmDisposed, + + /// The VM's client-local execution wait capacity is exhausted. + #[error("SDK execution wait limit maxPendingExecutionWaits ({limit}) reached; raise AgentOsConfig.max_pending_execution_waits")] + ExecutionWaitLimit { limit: usize }, + /// A filesystem path was not absolute (did not start with `/`). /// /// The message text matches the TypeScript `AgentOs` exactly (capital "P"). These strings are @@ -269,7 +277,9 @@ impl ClientError { format!("{code}: {message}") } } - ClientError::PathNotAbsolute(_) + ClientError::VmDisposed + | ClientError::ExecutionWaitLimit { .. } + | ClientError::PathNotAbsolute(_) | ClientError::PathNotNormalized(_) | ClientError::PathReadOnly(_) | ClientError::InvalidConfig(_) diff --git a/crates/client/src/language_execution.rs b/crates/client/src/language_execution.rs index fd07459625..4b84c15c2f 100644 --- a/crates/client/src/language_execution.rs +++ b/crates/client/src/language_execution.rs @@ -1,15 +1,78 @@ //! First-class JavaScript, TypeScript, Python, and shared execution lifecycle. use std::collections::HashMap; +use std::future::Future; +use std::sync::atomic::{AtomicBool, Ordering}; use agentos_sidecar_client::wire; -use tokio::sync::{broadcast, watch}; +use tokio::sync::{broadcast, watch, Semaphore}; use crate::agent_os::AgentOs; use crate::agent_os::ProcessEntry; use crate::error::{ClientError, ClientResult}; use crate::process::{ProcessOutput, ProcessRegistryReservation, ProcessStream}; +/// VM-local wait admission and cancellation. Permits never queue callers. +pub(crate) struct ExecutionWaits { + permits: Semaphore, + disposed: watch::Sender, + warning_active: AtomicBool, + limit: usize, +} + +impl ExecutionWaits { + pub(crate) fn new(limit: Option) -> ClientResult { + let limit = limit.unwrap_or(4096); + if limit == 0 || limit > Semaphore::MAX_PERMITS || limit as u64 > 9_007_199_254_740_991 { + return Err(ClientError::InvalidConfig(String::from( + "maxPendingExecutionWaits must be a positive safe integer", + ))); + } + Ok(Self { + permits: Semaphore::new(limit), + disposed: watch::channel(false).0, + warning_active: AtomicBool::new(false), + limit, + }) + } + + pub(crate) fn dispose(&self) { + self.disposed.send_replace(true); + } + + pub(crate) async fn until_disposed>( + &self, + operation: impl Future>, + ) -> ClientResult { + let mut disposed = self.disposed.subscribe(); + if *disposed.borrow() { + return Err(ClientError::VmDisposed); + } + let threshold = self.limit - self.limit / 5; + if self.limit - self.permits.available_permits() < threshold { + self.warning_active.store(false, Ordering::Relaxed); + } + let permit = self + .permits + .try_acquire() + .map_err(|_| ClientError::ExecutionWaitLimit { limit: self.limit })?; + let usage = self.limit - self.permits.available_permits(); + if usage >= threshold && !self.warning_active.swap(true, Ordering::Relaxed) { + eprintln!("agentOS SDK execution waits near maxPendingExecutionWaits ({usage}/{}); raise AgentOsConfig.max_pending_execution_waits", self.limit); + } + let result = tokio::select! { + biased; + _ = disposed.changed() => Err(ClientError::VmDisposed), + result = operation => result.map_err(Into::into), + }; + drop(permit); + if self.limit - self.permits.available_permits() < threshold { + self.warning_active.store(false, Ordering::Relaxed); + } + result + } +} + #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct ContextDescriptor { pub context_id: String, @@ -357,8 +420,12 @@ impl AgentOs { } let mut events = self.transport().subscribe_wire_events(); let accepted = match self - .transport() - .request_wire(self.execution_ownership(), payload) + .inner() + .execution_waits + .until_disposed( + self.transport() + .request_wire(self.execution_ownership(), payload), + ) .await? { wire::ResponsePayload::ExecutionAcceptedResponse(response) => response, @@ -380,7 +447,13 @@ impl AgentOs { })?, )); } - wait_for_completion_event(&mut events, &accepted.operation_id).await?; + self.inner() + .execution_waits + .until_disposed(wait_for_completion_event( + &mut events, + &accepted.operation_id, + )) + .await?; Ok(ExecutionSubmission::Completed( self.wait_execution(&accepted.operation_id).await?, )) @@ -1099,26 +1172,31 @@ impl AgentOs { async fn wait_execution(&self, execution_id: &str) -> ClientResult { let mut events = self.transport().subscribe_wire_events(); let first = self - .transport() - .request_wire( + .inner() + .execution_waits + .until_disposed(self.transport().request_wire( self.execution_ownership(), wire::RequestPayload::WaitExecutionRequest(wire::WaitExecutionRequest { execution_id: execution_id.to_owned(), }), - ) + )) .await?; let response = match first { wire::ResponsePayload::RejectedResponse(rejected) if rejected.code == "execution_busy" => { - wait_for_completion_event(&mut events, execution_id).await?; - self.transport() - .request_wire( + self.inner() + .execution_waits + .until_disposed(wait_for_completion_event(&mut events, execution_id)) + .await?; + self.inner() + .execution_waits + .until_disposed(self.transport().request_wire( self.execution_ownership(), wire::RequestPayload::WaitExecutionRequest(wire::WaitExecutionRequest { execution_id: execution_id.to_owned(), }), - ) + )) .await? } response => response, @@ -1449,3 +1527,89 @@ fn background_submission( ))), } } + +// These tests exercise private wait admission without exposing a transport-test API. +// Public lifecycle behavior is covered by tests/lifecycle_e2e.rs. +#[cfg(test)] +mod execution_wait_tests { + use super::*; + use std::time::Duration; + + async fn pending() -> ClientResult<()> { + std::future::pending().await + } + + #[tokio::test] + async fn disposal_rejects_pending_waits_without_polling_new_operations() { + let waits = ExecutionWaits::new(None).unwrap(); + let sibling = ExecutionWaits::new(None).unwrap(); + let operation = waits.until_disposed(pending()); + tokio::pin!(operation); + assert!( + tokio::time::timeout(Duration::from_millis(1), &mut operation) + .await + .is_err() + ); + waits.dispose(); + assert!(matches!(operation.await, Err(ClientError::VmDisposed))); + let polled = AtomicBool::new(false); + let error = waits + .until_disposed(async { + polled.store(true, Ordering::SeqCst); + Ok::<(), ClientError>(()) + }) + .await + .unwrap_err(); + assert_eq!(error.to_string(), "ERR_AGENTOS_VM_DISPOSED"); + assert!(!polled.load(Ordering::SeqCst)); + assert!(sibling + .until_disposed(async { Ok::<_, ClientError>(42) }) + .await + .is_ok()); + } + + #[tokio::test] + async fn pending_wait_capacity_is_released_on_caller_cancellation() { + let waits = ExecutionWaits::new(Some(1)).unwrap(); + let mut operation = Box::pin(waits.until_disposed(pending())); + assert!( + tokio::time::timeout(Duration::from_millis(1), operation.as_mut()) + .await + .is_err() + ); + assert!(matches!( + waits + .until_disposed(async { Ok::<_, ClientError>(()) }) + .await, + Err(ClientError::ExecutionWaitLimit { limit: 1 }) + )); + drop(operation); + assert!(waits + .until_disposed(async { Ok::<_, ClientError>(()) }) + .await + .is_ok()); + waits.dispose(); + } + + #[tokio::test] + async fn completed_waits_release_capacity_and_keep_results() { + let waits = ExecutionWaits::new(Some(1)).unwrap(); + let result = waits + .until_disposed(async { Ok::<_, ClientError>(42) }) + .await + .unwrap(); + assert!(waits + .until_disposed(async { Ok::<_, ClientError>(()) }) + .await + .is_ok()); + waits.dispose(); + assert_eq!(result, 42); + } + + #[test] + fn validates_wait_capacity_before_vm_creation() { + assert!(ExecutionWaits::new(Some(0)).is_err()); + assert!(ExecutionWaits::new(Some(usize::MAX)).is_err()); + assert!(ExecutionWaits::new(None).is_ok()); + } +} diff --git a/crates/client/src/process.rs b/crates/client/src/process.rs index c74348960f..5557acb64c 100644 --- a/crates/client/src/process.rs +++ b/crates/client/src/process.rs @@ -981,14 +981,28 @@ impl AgentOs { /// Await a spawned process's exit code. Unknown-pid lookup errors (synchronously in TS; here the /// lookup error is returned before any awaiting begins). pub async fn wait_process(&self, pid: u32) -> std::result::Result { - let (rx, process_id) = self + let (rx, process_id, language_execution) = self .inner() .processes .read(&pid, |_, entry| { - (entry.exit_tx.subscribe(), entry.process_id.clone()) + ( + entry.exit_tx.subscribe(), + entry.process_id.clone(), + entry.execution_id.is_some(), + ) }) .ok_or(ClientError::ProcessNotFound(pid))?; + if language_execution { + if let Some(code) = rx.borrow().exit_code() { + return Ok(code); + } + return self + .inner() + .execution_waits + .until_disposed(wait_for_process_outcome(rx, &process_id)) + .await; + } wait_for_process_outcome(rx, &process_id).await } diff --git a/crates/client/tests/lifecycle_e2e.rs b/crates/client/tests/lifecycle_e2e.rs index 97156193b5..c6286209e2 100644 --- a/crates/client/tests/lifecycle_e2e.rs +++ b/crates/client/tests/lifecycle_e2e.rs @@ -1,5 +1,5 @@ //! Lifecycle e2e against a real `agentos-native-sidecar`: independent VMs, post-shutdown isolation, and -//! idempotent shutdown. No V8/WASM required. +//! idempotent shutdown, and cancellation of execution waits. mod common; @@ -35,4 +35,94 @@ async fn lifecycle_independent_vms_and_idempotent_shutdown() { // Shutdown is idempotent. b.shutdown().await.expect("shutdown B"); b.shutdown().await.expect("shutdown B again (idempotent)"); + + shutdown_rejects_pending_execution_without_affecting_shared_sibling().await; +} + +async fn shutdown_rejects_pending_execution_without_affecting_shared_sibling() { + use agentos_client::error::ClientError; + use std::time::Duration; + + if !common::require_sidecar( + "shutdown_rejects_pending_execution_without_affecting_shared_sibling", + ) { + return; + } + let a = common::new_vm_with_sidecar_pool("lifecycle-execution-disposal").await; + let b = agentos_client::AgentOs::create(agentos_client::config::AgentOsConfig { + max_pending_execution_waits: Some(1), + sidecar: Some(agentos_client::config::AgentOsSidecarConfig::Shared { + pool: Some("lifecycle-execution-disposal".into()), + }), + ..Default::default() + }) + .await + .expect("create bounded sibling VM"); + let execution = tokio::spawn({ + let a = a.clone(); + async move { + a.execute_javascript( + "import fs from 'node:fs'; fs.writeFileSync('/tmp/execution-started', 'yes'); await new Promise(() => {});", + Default::default(), + ).await + } + }); + let started = tokio::time::timeout(Duration::from_secs(10), async { + loop { + if a.read_file("/tmp/execution-started").await.is_ok() { + break; + } + assert!( + !execution.is_finished(), + "execution ended before reaching the completion wait" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + let shutdown = tokio::spawn({ + let a = a.clone(); + async move { a.shutdown().await } + }); + let outcome = tokio::time::timeout(Duration::from_secs(1), execution).await; + let cleanup = shutdown.await.expect("shutdown task"); + let sibling = b.execute_javascript("42", Default::default()).await; + let process = b + .spawn_javascript("await new Promise(() => {});", Default::default()) + .await + .expect("spawn language process"); + let mut waiting = Box::pin(b.wait_process(process.pid)); + assert!( + tokio::time::timeout(Duration::from_millis(1), waiting.as_mut()) + .await + .is_err() + ); + let over_capacity = b.execute_javascript("42", Default::default()).await; + let shutdown = b.shutdown(); + let (wait_result, shutdown_result) = tokio::join!(waiting, shutdown); + shutdown_result.expect("shutdown sibling"); + assert!( + matches!(wait_result, Err(ClientError::VmDisposed)), + "background wait result: {wait_result:?}" + ); + assert!( + matches!( + over_capacity, + Err(ClientError::ExecutionWaitLimit { limit: 1 }) + ), + "background wait must count toward SDK capacity: {over_capacity:?}" + ); + started.expect("guest reached its completion wait"); + assert!(matches!( + outcome + .expect("execution wait must settle promptly") + .expect("execution task"), + Err(ClientError::VmDisposed) + )); + cleanup.expect("native VM teardown"); + assert!(sibling.is_ok(), "sibling VM remains usable: {sibling:?}"); + assert!(matches!( + a.execute_javascript("42", Default::default()).await, + Err(ClientError::VmDisposed) + )); } diff --git a/docs/content/docs/processes.mdx b/docs/content/docs/processes.mdx index 723f75598f..04c7c063c1 100644 --- a/docs/content/docs/processes.mdx +++ b/docs/content/docs/processes.mdx @@ -47,3 +47,17 @@ notification was missed. Real-terminal replay also reconciles `wait_shell` after a missed exit event or an earlier observation failure. TypeScript Core's local prompt terminal does not provide replay. + +## Embedded API + +Starting VM disposal rejects pending execution admission, completion, and +result waits with `ERR_AGENTOS_VM_DISPOSED`. TypeScript Core uses `dispose`; +Rust Core uses `shutdown` and returns `ClientError::VmDisposed`. New execution +waits on that VM also fail. Other VMs on the shared sidecar keep running, and +results already returned to the caller remain usable. + +Wait rejection does not confirm that guest code has stopped. Await VM disposal +to confirm native cleanup; disposal can still report a teardown error. + +Read more in the +[embedded API quickstart](/agentos/docs/quickstart-embedded). diff --git a/docs/content/docs/resource-limits.mdx b/docs/content/docs/resource-limits.mdx index 9cf2ac0432..e695d72588 100644 --- a/docs/content/docs/resource-limits.mdx +++ b/docs/content/docs/resource-limits.mdx @@ -29,6 +29,15 @@ Every queue, buffer, timeout, and resource collection has a bounded default. Near-capacity conditions emit `vm.limitWarning`; exhausted limits fail with a typed error naming the limit and the field used to raise it. +Embedded SDK execution waits have a separate, client-local bound of 4,096 +concurrent waits per VM. Configure `maxPendingExecutionWaits` in TypeScript +`AgentOsOptions` or `max_pending_execution_waits` in Rust `AgentOsConfig`. +At 80% capacity, the SDK writes a warning to the host console or stderr. +Exhausting this bound rejects the call with `AgentOsExecutionWaitLimit` in +TypeScript or `ClientError::ExecutionWaitLimit` in Rust. The error names the +option used to raise the bound. Completed, cancelled, and disposed waits +release their capacity. This option does not change sidecar execution limits. + `limits.reactor.shutdownDeadlineMs` bounds the SQLite close wait and subsequent resource drain together. Process termination and filesystem flushing happen before this budget. If close exceeds the deadline, diff --git a/packages/core/package.json b/packages/core/package.json index 06d86ccf81..120b6ee8ee 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -54,7 +54,7 @@ "check-types": "tsc --noEmit && tsc -p tsconfig.type-tests.json", "build": "tsc && node scripts/stage-default-software.mjs", "test": "vitest run --exclude '**/*.nightly.test.ts' --reporter=verbose", - "test:unit": "vitest run tests/agentos-package.test.ts tests/allowed-node-builtins.test.ts tests/host-functions-zod.test.ts tests/host-functions.test.ts tests/host-function-command-exec.test.ts tests/host-function-permissions.test.ts tests/cron-manager.test.ts tests/cron-timer-driver.test.ts tests/exec-timeout.test.ts tests/execution-abort.test.ts tests/generated-protocol.test.ts tests/leak-agent-os-processes.test.ts tests/leak-rpc-client.test.ts tests/mount-descriptors.test.ts tests/mount-reconfigure.test.ts tests/options-schema.test.ts tests/output-replay.test.ts tests/public-api-exports.test.ts tests/root-filesystem-descriptors.test.ts tests/runtime-compat-mount.test.ts tests/sidecar-client.test.ts tests/sidecar-permission-descriptors.test.ts tests/software-install.test.ts tests/wasm-permission-tiers.test.ts --fileParallelism=false", + "test:unit": "vitest run tests/agentos-package.test.ts tests/allowed-node-builtins.test.ts tests/host-functions-zod.test.ts tests/host-functions.test.ts tests/host-function-command-exec.test.ts tests/host-function-permissions.test.ts tests/cron-manager.test.ts tests/cron-timer-driver.test.ts tests/exec-timeout.test.ts tests/execution-abort.test.ts tests/execution-disposal.test.ts tests/generated-protocol.test.ts tests/leak-agent-os-processes.test.ts tests/leak-rpc-client.test.ts tests/mount-descriptors.test.ts tests/mount-reconfigure.test.ts tests/options-schema.test.ts tests/output-replay.test.ts tests/public-api-exports.test.ts tests/root-filesystem-descriptors.test.ts tests/runtime-compat-mount.test.ts tests/sidecar-client.test.ts tests/sidecar-permission-descriptors.test.ts tests/software-install.test.ts tests/wasm-permission-tiers.test.ts --fileParallelism=false", "test:pr": "pnpm test:unit", "test:nightly": "vitest run tests/*.nightly.test.ts --reporter=verbose --passWithNoTests" }, diff --git a/packages/core/src/agent-os.ts b/packages/core/src/agent-os.ts index a9c4d782c2..3b9c9b23c9 100644 --- a/packages/core/src/agent-os.ts +++ b/packages/core/src/agent-os.ts @@ -12,13 +12,6 @@ import type { MountConfigJsonValue, NativeMountPluginDescriptor, } from "@rivet-dev/agentos-runtime-core/descriptors"; -import * as executionProtocol from "@rivet-dev/agentos-runtime-core/protocol"; -import type { LivePackageAcquisitionSource } from "@rivet-dev/agentos-runtime-core/request-payloads"; -import { SidecarRejectedError } from "@rivet-dev/agentos-runtime-core/sidecar-errors"; -import type { - CreateVmConfig, - VmUserConfig, -} from "@rivet-dev/agentos-runtime-core/vm-config"; import { type HostFunction, type HostFunctionCollections, @@ -28,6 +21,13 @@ import { resolveHostFunctions, } from "@rivet-dev/agentos-runtime-core/host-functions"; import { zodToJsonSchema } from "@rivet-dev/agentos-runtime-core/host-functions-zod"; +import * as executionProtocol from "@rivet-dev/agentos-runtime-core/protocol"; +import type { LivePackageAcquisitionSource } from "@rivet-dev/agentos-runtime-core/request-payloads"; +import { SidecarRejectedError } from "@rivet-dev/agentos-runtime-core/sidecar-errors"; +import type { + CreateVmConfig, + VmUserConfig, +} from "@rivet-dev/agentos-runtime-core/vm-config"; import type { CodeEvaluationResult, CodeExecutionResult, @@ -54,6 +54,12 @@ import type { TypeScriptFileExecutionOptions, } from "./language-execution.js"; import { parseAgentOsOptions } from "./options-schema.js"; +import { + type ReplayReadOptions, + replayPage, + replayPageLimits, + replayWirePageLimits, +} from "./output-replay.js"; import { buildProcessForest } from "./process-forest.js"; import type { ConnectTerminalOptions, @@ -75,12 +81,6 @@ import { resolveSandboxOptions, } from "./sandbox.js"; import { resolvePublishedSidecarBinary } from "./sidecar/binary.js"; -import { - replayPage, - replayPageLimits, - replayWirePageLimits, - type ReplayReadOptions, -} from "./output-replay.js"; import { findCargoBinary, resolveCargoBinary } from "./sidecar/cargo.js"; export type { @@ -649,6 +649,8 @@ export interface AgentOsOptions< * on a slow consumer or a runaway guest before the limit is actually hit. */ onLimitWarning?: LimitWarningHandler; + /** Maximum simultaneous SDK execution waits for this VM (default 4096). */ + maxPendingExecutionWaits?: number; } export interface AgentOsRuntimeAdmin { @@ -2589,6 +2591,17 @@ function mapExecutionCompletedEvent( }; } +export class AgentOsExecutionWaitLimit extends Error { + readonly limit: number; + constructor(limit: number) { + super( + `SDK execution wait limit maxPendingExecutionWaits (${limit}) reached; raise AgentOsOptions.maxPendingExecutionWaits`, + ); + this.name = "AgentOsExecutionWaitLimit"; + this.limit = limit; + } +} + export class AgentOs { #kernel: Kernel; readonly sidecar: AgentOsSidecar; @@ -2619,6 +2632,10 @@ export class AgentOs { } >(); private _languageProcessIds = new Map(); + private _disposed = false; + private readonly _pendingExecutionWaits = new Set<(error: Error) => void>(); + private _executionWaitWarningActive = false; + private readonly _maxPendingExecutionWaits: number; private _pendingProcessRegistrations = 0; private _executionOutputHandlers = new Map< string, @@ -2775,7 +2792,9 @@ export class AgentOs { sidecarSession: AuthenticatedSession, sidecarVm: CreatedVm, limitWarningHandler?: LimitWarningHandler, + maxPendingExecutionWaits = 4096, ) { + this._maxPendingExecutionWaits = maxPendingExecutionWaits; this.#kernel = kernel; this.sidecar = sidecar; this._softwareRoots = softwareRoots; @@ -3116,6 +3135,7 @@ export class AgentOs { vmAdmin.sidecarSession, vmAdmin.sidecarVm, options?.onLimitWarning, + options?.maxPendingExecutionWaits, ); vm._sidecarLease = sidecarLease; vm._hostFunctions = vmAdmin.hostFunctions; @@ -3214,10 +3234,12 @@ export class AgentOs { }); let response: Awaited>; try { - response = await this._sidecarClient.sendVmRequest( - this._sidecarSession, - this._sidecarVm, - payload, + response = await this._untilDisposed(() => + this._sidecarClient.sendVmRequest( + this._sidecarSession, + this._sidecarVm, + payload, + ), ); } catch (error) { unsubscribeOutput(); @@ -3278,7 +3300,7 @@ export class AgentOs { }; } try { - await completion; + await this._untilDisposed(() => completion); return await this._waitExecutionResult(response.response.operationId); } finally { cleanup(); @@ -3957,6 +3979,35 @@ export class AgentOs { return response.response.executions.map(mapExecutionDescriptor); } + private async _untilDisposed(operation: () => Promise): Promise { + if (this._disposed) throw new Error("ERR_AGENTOS_VM_DISPOSED"); + if (this._pendingExecutionWaits.size >= this._maxPendingExecutionWaits) { + throw new AgentOsExecutionWaitLimit(this._maxPendingExecutionWaits); + } + let rejectDisposed!: (error: Error) => void; + const disposed = new Promise((_, reject) => { + rejectDisposed = reject; + }); + this._pendingExecutionWaits.add(rejectDisposed); + const warningThreshold = Math.ceil(this._maxPendingExecutionWaits * 0.8); + if ( + this._pendingExecutionWaits.size >= warningThreshold && + !this._executionWaitWarningActive + ) { + this._executionWaitWarningActive = true; + console.warn( + `agentOS SDK execution waits near maxPendingExecutionWaits (${this._pendingExecutionWaits.size}/${this._maxPendingExecutionWaits}); raise AgentOsOptions.maxPendingExecutionWaits`, + ); + } + try { + return await Promise.race([operation(), disposed]); + } finally { + this._pendingExecutionWaits.delete(rejectDisposed); + if (this._pendingExecutionWaits.size < warningThreshold) + this._executionWaitWarningActive = false; + } + } + private async _waitExecutionResult( executionId: string, ): Promise { @@ -3970,10 +4021,12 @@ export class AgentOs { let response: Awaited>; try { try { - response = await this._sidecarClient.sendVmRequest( - this._sidecarSession, - this._sidecarVm, - { type: "wait_execution", request: { executionId } }, + response = await this._untilDisposed(() => + this._sidecarClient.sendVmRequest( + this._sidecarSession, + this._sidecarVm, + { type: "wait_execution", request: { executionId } }, + ), ); } catch (error) { if ( @@ -3982,11 +4035,13 @@ export class AgentOs { ) { throw error; } - await completion; - response = await this._sidecarClient.sendVmRequest( - this._sidecarSession, - this._sidecarVm, - { type: "wait_execution", request: { executionId } }, + await this._untilDisposed(() => completion); + response = await this._untilDisposed(() => + this._sidecarClient.sendVmRequest( + this._sidecarSession, + this._sidecarVm, + { type: "wait_execution", request: { executionId } }, + ), ); } } finally { @@ -5244,7 +5299,9 @@ export class AgentOs { // The execution can finish between the local state check and the // signal request. Reconcile its final state and preserve kill/signal's // documented already-exited no-op behavior. - await this._waitProcess(pid); + // During disposal, the completion drain and native teardown confirm + // cleanup. Public result waits have already been rejected. + if (!this._disposed) await this._waitProcess(pid); } return; } @@ -5674,6 +5731,10 @@ export class AgentOs { } async dispose(): Promise { + this._disposed = true; + for (const reject of this._pendingExecutionWaits) + reject(new Error("ERR_AGENTOS_VM_DISPOSED")); + this._pendingExecutionWaits.clear(); this._cronManager.dispose(); const errors: unknown[] = []; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 41e12f24ad..f8e6be3ce7 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -1,5 +1,18 @@ // @rivet-dev/agentos +export type { + HostFunction, + HostFunctionCollection, + HostFunctionCollections, + HostFunctionExample, + HostFunctionSchemas, + ResolvedHostFunctions, +} from "@rivet-dev/agentos-runtime-core/host-functions"; +export { + hostFunctionCommandName, + hostFunctionDescription, + resolveHostFunctions, +} from "@rivet-dev/agentos-runtime-core/host-functions"; export { SidecarProcessError, SidecarProcessExited, @@ -7,7 +20,11 @@ export { type SidecarRejectionDetail, SidecarSilenceTimeout, } from "@rivet-dev/agentos-runtime-core/sidecar-errors"; -export { AgentOs, AgentOsSidecar } from "./agent-os.js"; +export { + AgentOs, + AgentOsExecutionWaitLimit, + AgentOsSidecar, +} from "./agent-os.js"; export { isPackageDescriptor, OPT_AGENTOS_BIN, @@ -25,19 +42,6 @@ export { hostDirMount, nodeModulesMount, } from "./host-dir-mount.js"; -export type { - HostFunction, - HostFunctionCollection, - HostFunctionCollections, - HostFunctionExample, - HostFunctionSchemas, - ResolvedHostFunctions, -} from "@rivet-dev/agentos-runtime-core/host-functions"; -export { - hostFunctionCommandName, - hostFunctionDescription, - resolveHostFunctions, -} from "@rivet-dev/agentos-runtime-core/host-functions"; export type * from "./language-execution.js"; export { createSnapshotExport } from "./layers.js"; export { diff --git a/packages/core/src/options-schema.ts b/packages/core/src/options-schema.ts index 098dffe4cc..1637a7c198 100644 --- a/packages/core/src/options-schema.ts +++ b/packages/core/src/options-schema.ts @@ -1,14 +1,14 @@ +import type { + HostFunction, + HostFunctionCollection, + HostFunctionCollections, +} from "@rivet-dev/agentos-runtime-core/host-functions"; import { z } from "zod/v4"; import type { AgentOsOptions, LimitWarningHandler, NativeMountConfig, } from "./agent-os.js"; -import type { - HostFunction, - HostFunctionCollection, - HostFunctionCollections, -} from "@rivet-dev/agentos-runtime-core/host-functions"; const stringArray = z.array(z.string()); const nonNegativeInteger = z.number().int().nonnegative(); @@ -408,6 +408,9 @@ export const agentOsOptionFieldSchemas = { permissions: permissionsSchema.optional(), sidecar: sidecarConfigSchema.optional(), limits: agentOsLimitsSchema.optional(), + maxPendingExecutionWaits: positiveInteger + .max(Number.MAX_SAFE_INTEGER) + .optional(), onLimitWarning: z .custom((value) => typeof value === "function", { message: "Expected function", diff --git a/packages/core/tests/execution-abort.test.ts b/packages/core/tests/execution-abort.test.ts index 2c60413cab..bc7bf150f3 100644 --- a/packages/core/tests/execution-abort.test.ts +++ b/packages/core/tests/execution-abort.test.ts @@ -39,7 +39,17 @@ type ExecutionAgent = { }; function createExecutionAgent() { - const agent = Object.create(AgentOs.prototype) as ExecutionAgent; + const agent = Reflect.construct(AgentOs, [ + {}, + {}, + [], + [], + {}, + {}, + { onEvent: () => () => {} }, + {}, + {}, + ]) as ExecutionAgent; agent._processes = new Map(); agent._languageProcesses = new Map(); agent._languageProcessIds = new Map(); diff --git a/packages/core/tests/execution-disposal.test.ts b/packages/core/tests/execution-disposal.test.ts new file mode 100644 index 0000000000..17730af189 --- /dev/null +++ b/packages/core/tests/execution-disposal.test.ts @@ -0,0 +1,327 @@ +import { ExecutionOutcome } from "@rivet-dev/agentos-runtime-core/protocol"; +import type { LiveEventFrame } from "@rivet-dev/agentos-runtime-core/protocol-frames"; +import { SIDECAR_PROTOCOL_SCHEMA } from "@rivet-dev/agentos-runtime-core/protocol-schema"; +import { SidecarRejectedError } from "@rivet-dev/agentos-runtime-core/sidecar-errors"; +import { describe, expect, it, vi } from "vitest"; +import { AgentOs, AgentOsExecutionWaitLimit } from "../src/agent-os.js"; +import type { SidecarProcess } from "../src/sidecar/rpc-client.js"; + +type Response = Awaited>; +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: Error) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + return { promise, resolve, reject }; +} +function executionTransport() { + const listeners = new Set<(event: LiveEventFrame) => void>(); + const requests: Array<{ + vmId: string; + payload: Parameters[2]; + response: ReturnType>; + }> = []; + const client = { + onEvent(handler: (event: LiveEventFrame) => void) { + listeners.add(handler); + return () => { + listeners.delete(handler); + }; + }, + sendVmRequest: vi.fn((_session, vm, payload) => { + const response = deferred(); + requests.push({ vmId: vm.vmId, payload, response }); + return response.promise; + }), + }; + const complete = (vmId: string) => { + const event: LiveEventFrame = { + frame_type: "event", + schema: SIDECAR_PROTOCOL_SCHEMA, + ownership: { + scope: "vm", + connection_id: "connection", + session_id: "session", + vm_id: vmId, + }, + payload: { + type: "execution_completed", + event: { + executionId: "operation-1", + generation: 1n, + outcome: ExecutionOutcome.Succeeded, + exitCode: 0, + error: null, + }, + }, + }; + for (const handler of listeners) handler(event); + }; + const makeVm = ( + vmId: string, + dispose = async () => {}, + maxPendingExecutionWaits?: number, + ) => { + // Match the existing Core unit fixtures: run field initializers and public + // namespaces with a controlled transport, without starting guest code. + const vm = Reflect.construct(AgentOs, [ + { dispose }, + {}, + [], + [], + {}, + {}, + client, + { connectionId: "connection", sessionId: "session" }, + { vmId }, + undefined, + maxPendingExecutionWaits, + ]) as AgentOs; + Object.assign(vm, { _cronManager: { dispose() {} } }); + return vm; + }; + return { client, requests, complete, makeVm }; +} +const accepted: Response = { + type: "execution_accepted", + response: { operationId: "operation-1", execution: null }, +}; +const succeeded: Response = { + type: "execution_completed", + response: { + execution: null, + outcome: ExecutionOutcome.Succeeded, + exitCode: 0, + error: null, + stdout: null, + stderr: null, + stdoutTruncated: null, + stderrTruncated: null, + evaluationValue: null, + typeScriptCheckResult: null, + }, +}; +function busy(code = "execution_busy") { + return new SidecarRejectedError(2, { + code, + message: "execution is busy", + limit_name: null, + configured_limit: null, + current_usage: null, + requested: null, + unit: null, + scope: null, + vm_id: null, + session_generation: null, + capability_id: null, + operation: null, + configuration_path: null, + retryable: true, + errno: null, + }); +} +async function nextTurn() { + await new Promise((resolve) => setTimeout(resolve, 0)); +} +function observe(promise: Promise) { + let outcome: { value: T } | { error: unknown } | undefined; + const settled = promise.then( + (value) => { + outcome = { value }; + }, + (error) => { + outcome = { error }; + }, + ); + return { outcome: () => outcome, settled }; +} + +describe("execution waits during VM disposal", () => { + it.each([ + "admission", + "completion", + "result", + "busy completion", + "busy retry", + ])("rejects a pending %s wait before native teardown finishes", async (stage) => { + const transport = executionTransport(); + const teardown = deferred(); + const vm = transport.makeVm("a", () => teardown.promise); + const execution = observe(vm.process.exec("pending")); + if (stage !== "admission") { + transport.requests[0].response.resolve(accepted); + await nextTurn(); + } + if (["result", "busy completion", "busy retry"].includes(stage)) { + transport.complete("a"); + await expect.poll(() => transport.requests.length).toBe(2); + } + if (stage.startsWith("busy")) { + transport.requests[1].response.reject(busy()); + await nextTurn(); + if (stage === "busy retry") { + transport.complete("a"); + await expect.poll(() => transport.requests.length).toBe(3); + } + } + const disposal = vm.dispose(); + try { + await expect + .poll(execution.outcome) + .toEqual({ error: new Error("ERR_AGENTOS_VM_DISPOSED") }); + // A response that arrives after disposal cannot restart result polling. + if (stage === "admission") { + transport.requests[0].response.resolve(accepted); + await nextTurn(); + expect(transport.requests).toHaveLength(1); + } + } finally { + teardown.resolve(); + await disposal; + } + }); + + it("bounds pending execution waits and warns before reaching the configured limit", async () => { + const transport = executionTransport(); + const vm = transport.makeVm("a", async () => {}, 2); + const warning = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + const first = observe(vm.process.exec("first")); + const second = observe(vm.process.exec("second")); + await expect(vm.process.exec("over limit")).rejects.toBeInstanceOf( + AgentOsExecutionWaitLimit, + ); + expect(transport.requests).toHaveLength(2); + expect(warning).toHaveBeenCalledOnce(); + expect(warning.mock.calls[0][0]).toContain("maxPendingExecutionWaits"); + await vm.dispose(); + await expect + .poll(first.outcome) + .toEqual({ error: new Error("ERR_AGENTOS_VM_DISPOSED") }); + await expect + .poll(second.outcome) + .toEqual({ error: new Error("ERR_AGENTOS_VM_DISPOSED") }); + } finally { + warning.mockRestore(); + await vm.dispose(); + } + }); + + it("rejects a background process.wait result request during disposal", async () => { + const transport = executionTransport(); + const vm = transport.makeVm("a"); + const spawning = vm.javascript.spawn("pending"); + const descriptor = { + executionId: "operation-1", + pid: 123, + createdAtMs: 0n, + }; + transport.requests[0].response.resolve({ + type: "execution_accepted", + response: { + operationId: "operation-1", + execution: descriptor, + }, + } as Response); + const process = await spawning; + const waiting = observe(vm.process.wait(process.pid)); + expect(transport.requests[1].payload.type).toBe("wait_execution"); + const disposal = vm.dispose(); + try { + await expect + .poll(waiting.outcome) + .toEqual({ error: new Error("ERR_AGENTOS_VM_DISPOSED") }); + } finally { + const signal = transport.requests.find( + (request) => request.payload.type === "signal_execution", + ); + signal?.response.resolve({ + type: "execution_descriptor", + response: { execution: descriptor }, + } as Response); + transport.complete("a"); + await disposal; + } + }); + + it("finishes disposal when a background execution exited before its signal arrived", async () => { + const transport = executionTransport(); + const vm = transport.makeVm("a"); + const spawning = vm.javascript.spawn("pending"); + transport.requests[0].response.resolve({ + type: "execution_accepted", + response: { + operationId: "operation-1", + execution: { executionId: "operation-1", pid: 123, createdAtMs: 0n }, + }, + } as Response); + const process = await spawning; + const waiting = observe(vm.process.wait(process.pid)); + const disposal = observe(vm.dispose()); + try { + const signal = transport.requests.find( + (request) => request.payload.type === "signal_execution", + ); + expect(signal).toBeDefined(); + signal!.response.reject(busy("execution_not_running")); + await nextTurn(); + transport.complete("a"); + await expect + .poll(waiting.outcome) + .toEqual({ error: new Error("ERR_AGENTOS_VM_DISPOSED") }); + await expect.poll(disposal.outcome).toEqual({ value: undefined }); + } finally { + transport.complete("a"); + await disposal.settled; + } + }); + + it("does not admit an execution after disposal", async () => { + const { client, makeVm } = executionTransport(); + const vm = makeVm("a"); + await vm.dispose(); + const execution = observe(vm.javascript.execute("42")); + await expect + .poll(execution.outcome) + .toEqual({ error: new Error("ERR_AGENTOS_VM_DISPOSED") }); + expect(client.sendVmRequest).not.toHaveBeenCalled(); + }); + + it("preserves the native teardown error while rejecting pending calls", async () => { + const transport = executionTransport(); + const error = new Error("native teardown failed"); + const vm = transport.makeVm("a", async () => { + throw error; + }); + const execution = observe(vm.process.exec("pending")); + await expect(vm.dispose()).rejects.toBe(error); + await expect + .poll(execution.outcome) + .toEqual({ error: new Error("ERR_AGENTOS_VM_DISPOSED") }); + }); + + it("keeps sibling executions and completed results usable on the same transport", async () => { + const transport = executionTransport(); + const a = transport.makeVm("a"); + const b = transport.makeVm("b"); + const first = observe(a.process.exec("pending")); + const second = b.process.exec("complete"); + transport.requests[1].response.resolve(accepted); + await nextTurn(); + await a.dispose(); + await expect + .poll(first.outcome) + .toEqual({ error: new Error("ERR_AGENTOS_VM_DISPOSED") }); + transport.complete("b"); + await expect.poll(() => transport.requests.length).toBe(3); + expect(transport.requests[2].vmId).toBe("b"); + transport.requests[2].response.resolve(succeeded); + await expect(second).resolves.toMatchObject({ + outcome: "succeeded", + exitCode: 0, + }); + await b.dispose(); + }); +}); diff --git a/packages/core/tests/options-schema.test.ts b/packages/core/tests/options-schema.test.ts index da08a83489..4695e32465 100644 --- a/packages/core/tests/options-schema.test.ts +++ b/packages/core/tests/options-schema.test.ts @@ -6,6 +6,22 @@ import { } from "../src/sandbox.js"; describe("AgentOsOptions validation", () => { + test("bounds client-local execution wait capacity", () => { + expect( + agentOsOptionsSchema.safeParse({ maxPendingExecutionWaits: 2 }).success, + ).toBe(true); + for (const maxPendingExecutionWaits of [ + 0, + -1, + 1.5, + Number.MAX_SAFE_INTEGER + 1, + ]) { + expect( + agentOsOptionsSchema.safeParse({ maxPendingExecutionWaits }).success, + ).toBe(false); + } + }); + test("accepts a complete initial environment including an explicit empty map", () => { expect(agentOsOptionsSchema.safeParse({ environment: {} }).success).toBe( true, diff --git a/secure-exec/src/runtime.ts b/secure-exec/src/runtime.ts index b8eed1bc40..2ace824000 100644 --- a/secure-exec/src/runtime.ts +++ b/secure-exec/src/runtime.ts @@ -35,6 +35,7 @@ const VM_OPTION_KEYS = Object.keys({ sidecar: true, limits: true, onLimitWarning: true, + maxPendingExecutionWaits: true, } satisfies Record) as (keyof VmOptions)[]; /** From b6d3a1dad3ef66ef255e886d1c61eb892d1f47d1 Mon Sep 17 00:00:00 2001 From: Ethon Date: Sun, 11 Oct 2026 02:00:56 +0800 Subject: [PATCH 2/2] fix(core): release shared sidecar holds after failed disposal --- docs/content/docs/processes.mdx | 4 ++ packages/core/package.json | 2 +- packages/core/src/agent-os.ts | 12 ++-- .../shared-sidecar-clean-exit-script.mjs | 33 ++++++++++- .../tests/shared-sidecar-clean-exit.test.ts | 55 +++++++++++-------- 5 files changed, 76 insertions(+), 30 deletions(-) diff --git a/docs/content/docs/processes.mdx b/docs/content/docs/processes.mdx index 04c7c063c1..649d65a958 100644 --- a/docs/content/docs/processes.mdx +++ b/docs/content/docs/processes.mdx @@ -59,5 +59,9 @@ results already returned to the caller remain usable. Wait rejection does not confirm that guest code has stopped. Await VM disposal to confirm native cleanup; disposal can still report a teardown error. +A failed TypeScript `dispose()` still releases its shared sidecar lease. The +host process can exit when no other VM leases are active. Sibling VMs remain +usable; releasing a lease does not confirm native cleanup. + Read more in the [embedded API quickstart](/agentos/docs/quickstart-embedded). diff --git a/packages/core/package.json b/packages/core/package.json index 120b6ee8ee..418de4b11c 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -54,7 +54,7 @@ "check-types": "tsc --noEmit && tsc -p tsconfig.type-tests.json", "build": "tsc && node scripts/stage-default-software.mjs", "test": "vitest run --exclude '**/*.nightly.test.ts' --reporter=verbose", - "test:unit": "vitest run tests/agentos-package.test.ts tests/allowed-node-builtins.test.ts tests/host-functions-zod.test.ts tests/host-functions.test.ts tests/host-function-command-exec.test.ts tests/host-function-permissions.test.ts tests/cron-manager.test.ts tests/cron-timer-driver.test.ts tests/exec-timeout.test.ts tests/execution-abort.test.ts tests/execution-disposal.test.ts tests/generated-protocol.test.ts tests/leak-agent-os-processes.test.ts tests/leak-rpc-client.test.ts tests/mount-descriptors.test.ts tests/mount-reconfigure.test.ts tests/options-schema.test.ts tests/output-replay.test.ts tests/public-api-exports.test.ts tests/root-filesystem-descriptors.test.ts tests/runtime-compat-mount.test.ts tests/sidecar-client.test.ts tests/sidecar-permission-descriptors.test.ts tests/software-install.test.ts tests/wasm-permission-tiers.test.ts --fileParallelism=false", + "test:unit": "vitest run tests/agentos-package.test.ts tests/allowed-node-builtins.test.ts tests/host-functions-zod.test.ts tests/host-functions.test.ts tests/host-function-command-exec.test.ts tests/host-function-permissions.test.ts tests/cron-manager.test.ts tests/cron-timer-driver.test.ts tests/exec-timeout.test.ts tests/execution-abort.test.ts tests/execution-disposal.test.ts tests/generated-protocol.test.ts tests/leak-agent-os-processes.test.ts tests/leak-rpc-client.test.ts tests/mount-descriptors.test.ts tests/mount-reconfigure.test.ts tests/options-schema.test.ts tests/output-replay.test.ts tests/public-api-exports.test.ts tests/root-filesystem-descriptors.test.ts tests/runtime-compat-mount.test.ts tests/shared-sidecar-clean-exit.test.ts tests/sidecar-client.test.ts tests/sidecar-permission-descriptors.test.ts tests/software-install.test.ts tests/wasm-permission-tiers.test.ts --fileParallelism=false", "test:pr": "pnpm test:unit", "test:nightly": "vitest run tests/*.nightly.test.ts --reporter=verbose --passWithNoTests" }, diff --git a/packages/core/src/agent-os.ts b/packages/core/src/agent-os.ts index 3b9c9b23c9..b95b69966c 100644 --- a/packages/core/src/agent-os.ts +++ b/packages/core/src/agent-os.ts @@ -6281,11 +6281,13 @@ async function leaseAgentOsSidecarVm( disposed = true; state.activeLeases.delete(leaseRecord!); state.description.activeVmCount = state.activeLeases.size; - await client.dispose(); - // Release this lease's hold; the shared sidecar is unref'd only - // once the last hold (across all in-flight + active leases) drops, - // so a one-shot host process can then exit on its own. - releaseHold(); + try { + await client.dispose(); + } finally { + // Release this lease's hold even when native teardown rejects. + // Other active or in-flight leases keep the shared sidecar alive. + releaseHold(); + } }, }; diff --git a/packages/core/tests/fixtures/shared-sidecar-clean-exit-script.mjs b/packages/core/tests/fixtures/shared-sidecar-clean-exit-script.mjs index 4790aba917..4269b6cbf6 100644 --- a/packages/core/tests/fixtures/shared-sidecar-clean-exit-script.mjs +++ b/packages/core/tests/fixtures/shared-sidecar-clean-exit-script.mjs @@ -4,8 +4,10 @@ // A correct dispose() must let node exit on its own — the shared sidecar's // child process + stdio handles must not keep the event loop alive after the // last VM lease is released. Imports the built package entry, like a consumer. +import assert from "node:assert/strict"; import { resolve } from "node:path"; import { pathToFileURL } from "node:url"; +import { SidecarProcess } from "@rivet-dev/agentos-runtime-core/sidecar-client"; const entry = pathToFileURL( resolve(import.meta.dirname, "../../dist/index.js"), @@ -14,7 +16,36 @@ const { AgentOs } = await import(entry); const vm = await AgentOs.create(); await vm.writeFile("/clean-exit.txt", "ok"); -await vm.dispose(); +if (process.env.AGENTOS_TEST_NATIVE_DISPOSE_FAILURE === "1") { + const sibling = await AgentOs.create(); + const nativeError = Object.assign(new Error("injected native VM teardown failure"), { + code: "ERR_AGENTOS_VM_TEARDOWN_DEADLINE", + }); + const disposeVm = SidecarProcess.prototype.disposeVm; + SidecarProcess.prototype.disposeVm = async function (...args) { + // Finish real native cleanup before injecting the rejection. This case + // isolates SDK failure handling from native resource reconciliation. + await disposeVm.apply(this, args); + throw nativeError; + }; + try { + await assert.rejects(vm.dispose(), /injected native VM teardown failure/); + } finally { + SidecarProcess.prototype.disposeVm = disposeVm; + } + assert.equal(sibling.sidecar.describe().activeVmCount, 1); + const result = await sibling.javascript.execute( + "await new Promise(resolve => setTimeout(resolve, 5)); console.log('sibling-ok')", + { output: { capture: "all" }, timeoutMs: 10_000 }, + ); + assert.equal(result.outcome, "succeeded"); + assert.equal(result.stdout.trim(), "sibling-ok"); + await sibling.dispose(); + assert.equal(sibling.sidecar.describe().activeVmCount, 0); + console.log("NATIVE_FAILURE_PROPAGATED SIBLING_TIMER_OK"); +} else { + await vm.dispose(); +} console.log("SCRIPT_DONE"); // Intentionally NO process.exit(): the process must terminate on its own. diff --git a/packages/core/tests/shared-sidecar-clean-exit.test.ts b/packages/core/tests/shared-sidecar-clean-exit.test.ts index b574269717..15e3a16f10 100644 --- a/packages/core/tests/shared-sidecar-clean-exit.test.ts +++ b/packages/core/tests/shared-sidecar-clean-exit.test.ts @@ -29,29 +29,38 @@ if (distMissing) { * itself, with no `process.exit()` escape hatch. */ describe("shared sidecar clean exit", () => { - it.skipIf(distMissing)("a standalone create()+dispose() script exits on its own", () => { - const script = resolve( - import.meta.dirname, - "fixtures/shared-sidecar-clean-exit-script.mjs", - ); - const result = spawnSync(process.execPath, [script], { - cwd: resolve(import.meta.dirname, ".."), - encoding: "utf8", - timeout: 60_000, - }); + it.skipIf(distMissing).each([false, true])( + "a standalone script exits after native disposal failure=%s", + (nativeDisposeFailure) => { + const script = resolve( + import.meta.dirname, + "fixtures/shared-sidecar-clean-exit-script.mjs", + ); + const result = spawnSync(process.execPath, [script], { + cwd: resolve(import.meta.dirname, ".."), + encoding: "utf8", + env: { + ...process.env, + AGENTOS_TEST_NATIVE_DISPOSE_FAILURE: nativeDisposeFailure ? "1" : "0", + }, + timeout: 60_000, + }); - const diag = `exit=${result.status} signal=${result.signal}\nstdout: ${result.stdout ?? ""}\nstderr: ${(result.stderr ?? "").slice(-800)}`; + const diag = `exit=${result.status} signal=${result.signal}\nstdout: ${result.stdout ?? ""}\nstderr: ${(result.stderr ?? "").slice(-800)}`; - // The script logic should complete regardless of the exit behavior. - expect(result.stdout ?? "", `script never finished its work.\n${diag}`).toContain( - "SCRIPT_DONE", - ); - // The real assertion: the process terminated on its own (was not killed - // by the spawn timeout). A hang leaves signal === "SIGTERM". - expect( - result.signal, - `process did not exit on its own within 60s — the shared sidecar kept the event loop alive.\n${diag}`, - ).toBeNull(); - expect(result.status, `non-zero exit.\n${diag}`).toBe(0); - }, 90_000); + // The script logic should complete regardless of the exit behavior. + expect( + result.stdout ?? "", + `script never finished its work.\n${diag}`, + ).toContain("SCRIPT_DONE"); + // The real assertion: the process terminated on its own (was not killed + // by the spawn timeout). A hang leaves signal === "SIGTERM". + expect( + result.signal, + `process did not exit on its own within 60s — the shared sidecar kept the event loop alive.\n${diag}`, + ).toBeNull(); + expect(result.status, `non-zero exit.\n${diag}`).toBe(0); + }, + 90_000, + ); });