use crate::common::{encode_json_string, frozen_time_ms}; use crate::javascript::{ CreateJavascriptContextRequest, GuestRuntimeConfig, JavascriptExecution, JavascriptExecutionEngine, JavascriptExecutionError, JavascriptExecutionEvent, JavascriptExecutionLimits, JavascriptSyncRpcRequest, StartJavascriptExecutionRequest, }; use crate::node_import_cache::{NodeImportCache, NODE_IMPORT_CACHE_ASSET_ROOT_ENV}; use crate::runtime_support::{ env_flag_enabled, file_fingerprint, resolve_execution_path, warmup_marker_path, NODE_DISABLE_COMPILE_CACHE_ENV, NODE_FROZEN_TIME_ENV, }; use crate::v8_runtime; use agentos_runtime::RuntimeContext; use base64::Engine as _; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; use std::collections::BTreeMap; use std::fmt; use std::fs; use std::io::{Read, Seek, SeekFrom}; use std::os::unix::fs::MetadataExt; use std::path::{Component, Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use tokio::sync::Notify; const NODE_ALLOW_PROCESS_BINDINGS_ENV: &str = "AGENTOS_ALLOW_PROCESS_BINDINGS"; const NODE_GUEST_PATH_MAPPINGS_ENV: &str = "AGENTOS_GUEST_PATH_MAPPINGS"; const NODE_SYNC_RPC_DATA_BYTES_ENV: &str = "AGENTOS_NODE_SYNC_RPC_DATA_BYTES"; const PYODIDE_INDEX_URL_ENV: &str = "AGENTOS_PYODIDE_INDEX_URL"; const PYODIDE_PACKAGE_BASE_URL_ENV: &str = "AGENTOS_PYODIDE_PACKAGE_BASE_URL"; const PYODIDE_PACKAGE_CACHE_DIR_ENV: &str = "AGENTOS_PYODIDE_PACKAGE_CACHE_DIR"; const PYODIDE_GUEST_ROOT: &str = "/__agentos_pyodide"; const PYODIDE_CACHE_GUEST_ROOT: &str = "/__agentos_pyodide_cache"; const PYTHON_CODE_ENV: &str = "AGENTOS_PYTHON_CODE"; const PYTHON_FILE_ENV: &str = "AGENTOS_PYTHON_FILE"; const PYTHON_PREWARM_ONLY_ENV: &str = "AGENTOS_PYTHON_PREWARM_ONLY"; const PYTHON_WARMUP_DEBUG_ENV: &str = "AGENTOS_PYTHON_WARMUP_DEBUG"; const PYTHON_WARMUP_METRICS_PREFIX: &str = "__AGENTOS_PYTHON_WARMUP_METRICS__:"; const PYTHON_WARMUP_MARKER_VERSION: &str = "2"; const DEFAULT_PYTHON_OUTPUT_BUFFER_MAX_BYTES: usize = 1024 * 1024; const DEFAULT_PYTHON_EXECUTION_TIMEOUT_MS: u64 = 5 * 60 * 1000; const DEFAULT_PYTHON_MAX_OLD_SPACE_MB: usize = 0; const DEFAULT_PYTHON_VFS_RPC_TIMEOUT_MS: u64 = 30_000; const PYTHON_SYNC_RPC_DATA_BYTES: usize = 20 * 1024 * 1024; const PYTHON_SYNC_RPC_WAIT_TIMEOUT_MS: u64 = 120_000; const PYTHON_PREWARM_TIMEOUT: Duration = Duration::from_secs(120); #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum PythonVfsRpcMethod { Read, Write, Stat, Lstat, ReadDir, Mkdir, Unlink, Rmdir, Rename, Symlink, ReadLink, Setattr, HttpRequest, DnsLookup, SubprocessRun, SocketConnect, SocketSend, SocketRecv, SocketClose, UdpCreate, UdpSendto, UdpRecvfrom, } impl PythonVfsRpcMethod { fn from_wire(value: &str) -> Option { match value { "fsRead" => Some(Self::Read), "fsWrite" => Some(Self::Write), "fsStat" => Some(Self::Stat), "fsLstat" => Some(Self::Lstat), "fsReaddir" => Some(Self::ReadDir), "fsMkdir" => Some(Self::Mkdir), "fsUnlink" => Some(Self::Unlink), "fsRmdir" => Some(Self::Rmdir), "fsRename" => Some(Self::Rename), "fsSymlink" => Some(Self::Symlink), "fsReadlink" => Some(Self::ReadLink), "fsSetattr" => Some(Self::Setattr), "httpRequest" => Some(Self::HttpRequest), "dnsLookup" => Some(Self::DnsLookup), "subprocessRun" => Some(Self::SubprocessRun), "socketConnect" => Some(Self::SocketConnect), "socketSend" => Some(Self::SocketSend), "socketRecv" => Some(Self::SocketRecv), "socketClose" => Some(Self::SocketClose), "udpCreate" => Some(Self::UdpCreate), "udpSendto" => Some(Self::UdpSendto), "udpRecvfrom" => Some(Self::UdpRecvfrom), _ => None, } } } #[derive(Debug, Clone, PartialEq, Eq)] pub struct PythonVfsRpcRequest { pub id: u64, pub method: PythonVfsRpcMethod, pub path: String, /// Second path for `Rename` (the destination); `None` for other methods. pub destination: Option, /// Symlink target (the path the link points at), for `Symlink`. pub target: Option, /// `Setattr` metadata fields (each applied only when present). pub mode: Option, pub uid: Option, pub gid: Option, pub atime_ms: Option, pub mtime_ms: Option, pub content_base64: Option, pub recursive: bool, pub url: Option, pub http_method: Option, pub headers: BTreeMap, pub body_base64: Option, pub hostname: Option, pub family: Option, /// Port for socket connect/sendto. pub port: Option, /// Socket handle for send/recv/close/sendto/recvfrom. pub socket_id: Option, pub command: Option, pub args: Vec, /// Explicit child argv[0], kept separate from the executable path. pub argv0: Option, pub cwd: Option, pub env: BTreeMap, pub shell: bool, pub max_buffer: Option, /// Maximum time one socket receive operation may wait for shared /// readiness. `Some(0)` preserves nonblocking socket behavior. pub timeout_ms: Option, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct PythonVfsRpcStat { pub mode: u32, pub size: u64, pub is_directory: bool, pub is_symbolic_link: bool, } #[derive(Debug, Clone, PartialEq, Eq)] pub enum PythonVfsRpcResponsePayload { Empty, Read { content_base64: String, }, Stat { stat: PythonVfsRpcStat, }, ReadDir { entries: Vec, }, Http { status: u16, reason: String, url: String, headers: BTreeMap>, body_base64: String, }, DnsLookup { addresses: Vec, }, SubprocessRun { exit_code: i32, stdout: String, stderr: String, max_buffer_exceeded: bool, }, SocketCreated { socket_id: u64, }, SocketSent { bytes_sent: usize, }, SocketReceived { data_base64: String, closed: bool, timed_out: bool, }, UdpReceived { data_base64: String, host: String, port: u16, timed_out: bool, }, SymlinkTarget { target: String, }, } #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] struct PythonVfsBridgeRequestWire { method: String, #[serde(default)] path: String, #[serde(default)] destination: Option, #[serde(default)] target: Option, // JS numbers cross the bridge as f64; accept that and narrow below. #[serde(default)] mode: Option, #[serde(default)] uid: Option, #[serde(default)] gid: Option, #[serde(default, rename = "atimeMs")] atime_ms: Option, #[serde(default, rename = "mtimeMs")] mtime_ms: Option, #[serde(default)] content_base64: Option, #[serde(default)] recursive: bool, #[serde(default)] url: Option, #[serde(default, rename = "httpMethod")] http_method: Option, #[serde(default)] headers: BTreeMap, #[serde(default, rename = "bodyBase64")] body_base64: Option, #[serde(default)] hostname: Option, #[serde(default)] family: Option, #[serde(default)] port: Option, #[serde(default, rename = "socketId")] socket_id: Option, #[serde(default)] command: Option, #[serde(default)] args: Vec, #[serde(default)] argv0: Option, #[serde(default)] cwd: Option, #[serde(default)] env: BTreeMap, #[serde(default)] shell: bool, #[serde(default, rename = "maxBuffer")] max_buffer: Option, #[serde(default, rename = "timeoutMs")] timeout_ms: Option, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "camelCase")] struct PythonGuestPathMappingWire { guest_path: String, host_path: String, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct CreatePythonContextRequest { pub vm_id: String, pub pyodide_dist_path: PathBuf, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct PythonContext { pub context_id: String, pub vm_id: String, pub pyodide_dist_path: PathBuf, } /// Per-execution Python runtime limits, carried as typed fields rather than /// `AGENTOS_*` env vars. Populated by the sidecar from the per-VM `VmLimits` /// (originating from `CreateVmConfig` on the BARE wire); `None` selects the /// engine default. See the env-vs-wire rule in `crates/sidecar/CLAUDE.md`. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct PythonExecutionLimits { /// Captured-output buffer cap in bytes. `None` keeps the engine default. pub output_buffer_max_bytes: Option, /// Execution wall-clock cap in ms. `None` keeps the engine default; /// `Some(0)` disables the timeout. pub execution_timeout_ms: Option, /// Pyodide V8 old-space cap in MB (`0` keeps the V8 default). `None` keeps /// the engine default. pub max_old_space_mb: Option, /// VFS sync-RPC wait ceiling in ms. `None` keeps the engine default. pub vfs_rpc_timeout_ms: Option, /// VM readiness work bound forwarded unchanged to the Pyodide V8 runner. pub reactor_work_quantum: Option, /// Per-call host bridge deadline forwarded unchanged to the Pyodide V8 runner. pub bridge_call_timeout_ms: Option, /// Maximum host-direct descriptors retained for managed Pyodide assets. /// `None` keeps the execution engine's bounded fallback. The native sidecar /// always supplies the VM kernel's configured descriptor limit. pub max_open_fds: Option, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct StartPythonExecutionRequest { pub vm_id: String, pub context_id: String, pub code: String, pub file_path: Option, pub env: BTreeMap, pub cwd: PathBuf, /// Per-execution runtime limits (see [`PythonExecutionLimits`]). pub limits: PythonExecutionLimits, /// Per-execution guest-runtime config, forwarded to the Pyodide runner's JS /// execution (see [`GuestRuntimeConfig`]). pub guest_runtime: GuestRuntimeConfig, } #[derive(Debug, Clone, PartialEq, Eq)] pub enum PythonExecutionEvent { Stdout(Vec), Stderr(Vec), JavascriptSyncRpcRequest(JavascriptSyncRpcRequest), VfsRpcRequest(Box), Exited(i32), } #[derive(Debug, Clone, PartialEq, Eq)] pub struct PythonExecutionResult { pub execution_id: String, pub exit_code: i32, pub stdout: Vec, pub stderr: Vec, } #[derive(Debug)] pub enum PythonExecutionError { MissingContext(String), InvalidLimit(String), VmMismatch { expected: String, found: String, }, /// Guest Python is unavailable because this build was compiled without the /// bundled Pyodide runtime assets (the published crate excludes them). RuntimeUnavailable, PrepareRuntime(std::io::Error), PrepareWarmPath(std::io::Error), WarmupFailed { exit_code: i32, stderr: String, }, Spawn(std::io::Error), StdinClosed, Stdin(std::io::Error), Kill(std::io::Error), Control(std::io::Error), TimedOut(Duration), PendingVfsRpcRequest(u64), RpcResponse(String), OutputBufferExceeded { stream: &'static str, limit: usize, }, EventChannelClosed, } impl fmt::Display for PythonExecutionError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::MissingContext(context_id) => { write!(f, "unknown guest Python context: {context_id}") } Self::InvalidLimit(message) => write!(f, "invalid Python limit: {message}"), Self::VmMismatch { expected, found } => { write!( f, "guest Python context belongs to vm {expected}, not {found}" ) } Self::RuntimeUnavailable => write!( f, "guest Python execution is unavailable: this build of agentos-execution \ was compiled without the bundled Pyodide runtime assets" ), Self::PrepareRuntime(err) => { write!(f, "failed to prepare guest Python runtime assets: {err}") } Self::PrepareWarmPath(err) => { write!(f, "failed to prepare guest Python warm path: {err}") } Self::WarmupFailed { exit_code, stderr } => { if stderr.trim().is_empty() { write!(f, "guest Python warmup exited with status {exit_code}") } else { write!( f, "guest Python warmup exited with status {exit_code}: {}", stderr.trim() ) } } Self::Spawn(err) => write!(f, "failed to start guest Python runtime: {err}"), Self::StdinClosed => f.write_str("guest Python stdin is already closed"), Self::Stdin(err) => write!(f, "failed to write guest stdin: {err}"), Self::Kill(err) => write!(f, "failed to kill guest Python runtime: {err}"), Self::Control(err) => write!(f, "failed to control guest Python runtime: {err}"), Self::TimedOut(timeout) => write!( f, "guest Python runtime timed out after {}ms", timeout.as_millis() ), Self::PendingVfsRpcRequest(id) => { write!( f, "guest Python execution requires servicing pending VFS RPC request {id}" ) } Self::RpcResponse(message) => { write!( f, "failed to reply to guest Python VFS RPC request: {message}" ) } Self::OutputBufferExceeded { stream, limit } => { write!( f, "guest Python {stream} exceeded the captured output limit of {limit} bytes" ) } Self::EventChannelClosed => { f.write_str("guest Python event channel closed unexpectedly") } } } } impl std::error::Error for PythonExecutionError {} /// Returns an error when this build was compiled without the bundled Pyodide /// runtime assets (the published crate excludes them; see `build.rs`). In the /// workspace build the in-tree assets are present and this is a no-op. fn ensure_pyodide_available() -> Result<(), PythonExecutionError> { #[cfg(agentos_pyodide_unavailable)] { return Err(PythonExecutionError::RuntimeUnavailable); } #[cfg(not(agentos_pyodide_unavailable))] { Ok(()) } } #[derive(Debug)] pub struct PythonExecution { runtime: RuntimeContext, execution_id: String, child_pid: u32, inner: JavascriptExecution, pyodide_dist_path: PathBuf, managed_host_files: PythonManagedHostFiles, pending_vfs_rpc: Arc>>, v8_session: crate::v8_host::V8SessionHandle, output_buffer_max_bytes: usize, execution_timeout: Option, vfs_rpc_timeout: Duration, } /// Cloneable response lane for a pending Python VFS RPC. /// /// Socket operations complete on the shared Tokio runtime after the sidecar /// dispatcher has returned to its event loop. Keeping only the pending-RPC /// state and V8 session handle here lets those tasks reply directly without /// parking the dispatcher or borrowing the process table across an await. #[derive(Debug, Clone)] pub struct PythonVfsRpcResponder { pending_vfs_rpc: Arc>>, v8_session: crate::v8_host::V8SessionHandle, } #[derive(Debug)] struct PendingVfsRpc { state: PendingVfsRpcState, timeout_abort: Option, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum PendingVfsRpcState { Pending(u64), TimedOut(u64), } #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum PendingVfsRpcResolution { Pending, TimedOut, Missing, } impl PythonExecution { pub fn vfs_rpc_responder(&self) -> PythonVfsRpcResponder { PythonVfsRpcResponder { pending_vfs_rpc: Arc::clone(&self.pending_vfs_rpc), v8_session: self.v8_session.clone(), } } pub fn execution_id(&self) -> &str { &self.execution_id } pub fn child_pid(&self) -> u32 { self.child_pid } pub fn uses_shared_v8_runtime(&self) -> bool { self.inner.uses_shared_v8_runtime() } pub fn start_prepared(&mut self) -> Result<(), PythonExecutionError> { self.inner.start_prepared().map_err(map_javascript_error) } #[doc(hidden)] pub fn is_prepared_for_start(&self) -> bool { self.inner.is_prepared_for_start() } pub fn write_stdin(&mut self, chunk: &[u8]) -> Result<(), PythonExecutionError> { self.inner .write_kernel_stdin_only(chunk) .map_err(map_javascript_error) } pub fn close_stdin(&mut self) -> Result<(), PythonExecutionError> { self.inner.close_kernel_stdin_only(); Ok(()) } pub fn cancel(&mut self) -> Result<(), PythonExecutionError> { self.kill() } pub fn kill(&mut self) -> Result<(), PythonExecutionError> { self.close_stdin()?; self.inner.terminate().map_err(map_javascript_error) } pub fn pause(&self) -> Result<(), PythonExecutionError> { self.inner.pause().map_err(map_javascript_error) } pub fn resume(&self) -> Result<(), PythonExecutionError> { self.inner.resume().map_err(map_javascript_error) } pub fn respond_vfs_rpc_success( &mut self, id: u64, payload: PythonVfsRpcResponsePayload, ) -> Result<(), PythonExecutionError> { self.vfs_rpc_responder().respond_success(id, payload) } pub fn respond_vfs_rpc_error( &mut self, id: u64, code: impl Into, message: impl Into, ) -> Result<(), PythonExecutionError> { self.vfs_rpc_responder().respond_error(id, code, message) } pub fn respond_javascript_sync_rpc_success( &mut self, id: u64, result: Value, ) -> Result<(), PythonExecutionError> { self.inner .respond_sync_rpc_success(id, result) .map_err(map_javascript_error) } pub fn claim_javascript_sync_rpc_response( &mut self, id: u64, ) -> Result { self.inner .claim_sync_rpc_response(id) .map_err(map_javascript_error) } pub fn respond_claimed_javascript_sync_rpc_success( &mut self, id: u64, result: Value, ) -> Result<(), PythonExecutionError> { self.inner .respond_claimed_sync_rpc_success(id, result) .map_err(map_javascript_error) } pub fn respond_javascript_sync_rpc_error( &mut self, id: u64, code: impl Into, message: impl Into, ) -> Result<(), PythonExecutionError> { self.inner .respond_sync_rpc_error(id, code, message) .map_err(map_javascript_error) } pub fn respond_claimed_javascript_sync_rpc_error( &mut self, id: u64, code: impl Into, message: impl Into, ) -> Result<(), PythonExecutionError> { self.inner .respond_claimed_sync_rpc_error(id, code, message) .map_err(map_javascript_error) } pub async fn poll_event( &mut self, timeout: Duration, ) -> Result, PythonExecutionError> { self.poll_event_until(Some(timeout)).await } pub fn try_poll_event(&mut self) -> Result, PythonExecutionError> { loop { let Some(event) = self.inner.try_poll_event().map_err(map_javascript_error)? else { return Ok(None); }; if let Some(event) = self.translate_javascript_event(event)? { return Ok(Some(event)); } } } async fn poll_event_until( &mut self, timeout: Option, ) -> Result, PythonExecutionError> { let started = Instant::now(); loop { let remaining = timeout.map(|timeout| { if timeout.is_zero() { Duration::ZERO } else { timeout.saturating_sub(started.elapsed()) } }); match self .inner .poll_event_until(remaining) .await .map_err(map_javascript_error)? { Some(event) => { if let Some(event) = self.translate_javascript_event(event)? { return Ok(Some(event)); } } None => return Ok(None), } } } /// Service a module-resolution JS sync RPC host-directly via the underlying /// JS execution's translator. For consumers driving `poll_event_blocking` /// manually without a kernel/service loop. pub fn try_service_standalone_module_sync_rpc( &mut self, request: &JavascriptSyncRpcRequest, ) -> Result { self.inner .try_service_standalone_module_sync_rpc(request) .map_err(map_javascript_error) } /// Service the local stdin bridge for consumers that drive a standalone /// Python execution without the sidecar's sync-RPC dispatcher. #[doc(hidden)] pub fn try_service_standalone_stdin_sync_rpc( &mut self, request: &JavascriptSyncRpcRequest, ) -> Result { self.inner .handle_kernel_stdin_sync_rpc(request) .map_err(map_javascript_error) } pub fn poll_event_blocking( &mut self, timeout: Duration, ) -> Result, PythonExecutionError> { let deadline = Instant::now() + timeout; loop { let remaining = deadline.saturating_duration_since(Instant::now()); match self .inner .poll_event_blocking(remaining) .map_err(map_javascript_error)? { Some(event) => { if let Some(event) = self.translate_javascript_event(event)? { return Ok(Some(event)); } } None => { if Instant::now() >= deadline { return Ok(None); } } } } } fn next_event_blocking(&mut self) -> Result { loop { let event = self .inner .next_event_blocking() .map_err(map_javascript_error)?; if let Some(event) = self.translate_javascript_event(event)? { return Ok(event); } } } pub fn wait( mut self, timeout: Option, ) -> Result { self.close_stdin()?; let mut stdout = PythonOutputBuffer::new(self.output_buffer_max_bytes); let mut stderr = PythonOutputBuffer::new(self.output_buffer_max_bytes); let started = Instant::now(); let timeout = match (timeout, self.execution_timeout) { (Some(requested), Some(configured)) => Some(requested.min(configured)), (Some(requested), None) => Some(requested), (None, Some(configured)) => Some(configured), (None, None) => None, }; loop { let poll_timeout = python_wait_remaining(timeout, started); let event = match poll_timeout { Some(timeout) => self.poll_event_blocking(timeout)?, None => Some(self.next_event_blocking()?), }; match event { Some(PythonExecutionEvent::Stdout(chunk)) => stdout.extend(&chunk), Some(PythonExecutionEvent::Stderr(chunk)) => stderr.extend(&chunk), Some(PythonExecutionEvent::JavascriptSyncRpcRequest(request)) => { // Module-resolution sync RPCs are serviced host-directly via // the JS execution's own translator (the standalone Python // wait loop runs without a kernel/service loop). if self .inner .try_service_standalone_module_sync_rpc(&request) .map_err(map_javascript_error)? { continue; } if let Some((code, message)) = python_javascript_sync_rpc_error(&request) { self.inner .respond_sync_rpc_error(request.id, code, message) .map_err(map_javascript_error)?; continue; } return Err(PythonExecutionError::RpcResponse(format!( "guest Python execution requires servicing pending JavaScript sync RPC request {} {} {:?}", request.id, request.method, request.args ))); } Some(PythonExecutionEvent::VfsRpcRequest(request)) => { return Err(PythonExecutionError::PendingVfsRpcRequest(request.id)); } Some(PythonExecutionEvent::Exited(exit_code)) => { return Ok(PythonExecutionResult { execution_id: self.execution_id.clone(), exit_code, stdout: stdout.into_inner(), stderr: stderr.into_inner(), }); } None => {} } if let Some(limit) = timeout { if started.elapsed() >= limit { self.kill()?; return Err(PythonExecutionError::TimedOut(limit)); } } } } fn translate_javascript_event( &mut self, event: JavascriptExecutionEvent, ) -> Result, PythonExecutionError> { match event { JavascriptExecutionEvent::Stdout(chunk) => { Ok(Some(PythonExecutionEvent::Stdout(chunk))) } JavascriptExecutionEvent::Stderr(chunk) => { Ok(Some(PythonExecutionEvent::Stderr(chunk))) } JavascriptExecutionEvent::Exited(code) => Ok(Some(PythonExecutionEvent::Exited(code))), JavascriptExecutionEvent::SignalState { .. } => Ok(None), JavascriptExecutionEvent::SyncRpcRequest(request) => { if request.method == "_pythonRpc" { let request = parse_python_bridge_sync_rpc_request(&request)?; set_pending_vfs_rpc_state(&self.pending_vfs_rpc, request.id)?; spawn_python_vfs_rpc_timeout( &self.runtime, request.id, self.vfs_rpc_timeout, self.pending_vfs_rpc.clone(), self.v8_session.clone(), )?; Ok(Some(PythonExecutionEvent::VfsRpcRequest(Box::new(request)))) } else { if self.try_service_standalone_module_sync_rpc(&request)? { return Ok(None); } if let Some(action) = python_javascript_sync_rpc_action( &self.pyodide_dist_path, &mut self.managed_host_files, &request, )? { respond_python_javascript_sync_rpc_action( &mut self.inner, request.id, action, )?; Ok(None) } else { Ok(Some(PythonExecutionEvent::JavascriptSyncRpcRequest( request, ))) } } } } } } fn python_wait_remaining(timeout: Option, started: Instant) -> Option { timeout.map(|limit| limit.saturating_sub(started.elapsed())) } impl PythonVfsRpcResponder { pub fn respond_success( &self, id: u64, payload: PythonVfsRpcResponsePayload, ) -> Result<(), PythonExecutionError> { match clear_pending_vfs_rpc(&self.pending_vfs_rpc, id)? { PendingVfsRpcResolution::Pending => {} PendingVfsRpcResolution::TimedOut | PendingVfsRpcResolution::Missing => { return Err(PythonExecutionError::RpcResponse(format!( "VFS RPC request {id} is no longer pending" ))); } } let result = match payload { PythonVfsRpcResponsePayload::Empty => json!({}), PythonVfsRpcResponsePayload::Read { content_base64 } => { json!({ "contentBase64": content_base64 }) } PythonVfsRpcResponsePayload::Stat { stat } => json!({ "stat": { "mode": stat.mode, "size": stat.size, "isDirectory": stat.is_directory, "isSymbolicLink": stat.is_symbolic_link, } }), PythonVfsRpcResponsePayload::ReadDir { entries } => { json!({ "entries": entries }) } PythonVfsRpcResponsePayload::Http { status, reason, url, headers, body_base64, } => json!({ "status": status, "reason": reason, "url": url, "headers": headers, "bodyBase64": body_base64, }), PythonVfsRpcResponsePayload::DnsLookup { addresses } => { json!({ "addresses": addresses }) } PythonVfsRpcResponsePayload::SubprocessRun { exit_code, stdout, stderr, max_buffer_exceeded, } => json!({ "exitCode": exit_code, "stdout": stdout, "stderr": stderr, "maxBufferExceeded": max_buffer_exceeded, }), PythonVfsRpcResponsePayload::SocketCreated { socket_id } => json!({ "socketId": socket_id, }), PythonVfsRpcResponsePayload::SocketSent { bytes_sent } => json!({ "bytesSent": bytes_sent, }), PythonVfsRpcResponsePayload::SocketReceived { data_base64, closed, timed_out, } => json!({ "dataBase64": data_base64, "closed": closed, "timedOut": timed_out, }), PythonVfsRpcResponsePayload::UdpReceived { data_base64, host, port, timed_out, } => json!({ "dataBase64": data_base64, "host": host, "port": port, "timedOut": timed_out, }), PythonVfsRpcResponsePayload::SymlinkTarget { target } => json!({ "target": target, }), }; let payload = v8_runtime::json_to_cbor_payload(&result) .map_err(|error| PythonExecutionError::RpcResponse(error.to_string()))?; self.v8_session .send_bridge_response(id, 0, payload) .map_err(|error| PythonExecutionError::RpcResponse(error.to_string())) } pub fn respond_error( &self, id: u64, code: impl Into, message: impl Into, ) -> Result<(), PythonExecutionError> { match clear_pending_vfs_rpc(&self.pending_vfs_rpc, id)? { PendingVfsRpcResolution::Pending => {} PendingVfsRpcResolution::TimedOut | PendingVfsRpcResolution::Missing => { return Err(PythonExecutionError::RpcResponse(format!( "VFS RPC request {id} is no longer pending" ))); } } let error = format!("{}: {}", code.into(), message.into()); self.v8_session .send_bridge_response(id, 1, error.into_bytes()) .map_err(|error| PythonExecutionError::RpcResponse(error.to_string())) } } fn clear_pending_vfs_rpc( pending_vfs_rpc: &Arc>>, id: u64, ) -> Result { let mut pending = pending_vfs_rpc .lock() .map_err(|_| PythonExecutionError::EventChannelClosed)?; let resolution = match pending.as_ref().map(|rpc| rpc.state) { Some(PendingVfsRpcState::Pending(current)) if current == id => { PendingVfsRpcResolution::Pending } Some(PendingVfsRpcState::TimedOut(current)) if current == id => { PendingVfsRpcResolution::TimedOut } _ => return Ok(PendingVfsRpcResolution::Missing), }; if let Some(rpc) = pending.take() { if let Some(timeout_abort) = rpc.timeout_abort { timeout_abort.abort(); } } Ok(resolution) } fn cancel_pending_vfs_rpc(pending_vfs_rpc: &Arc>>) { let pending = pending_vfs_rpc .lock() .map(|mut pending| pending.take()) .unwrap_or_else(|poisoned| { eprintln!("ERR_AGENTOS_PYTHON_VFS_RPC_STATE_POISONED: cancelling pending timeout"); poisoned.into_inner().take() }); if let Some(timeout_abort) = pending.and_then(|rpc| rpc.timeout_abort) { timeout_abort.abort(); } } impl Drop for PythonExecution { fn drop(&mut self) { cancel_pending_vfs_rpc(&self.pending_vfs_rpc); if let Err(error) = self.close_stdin() { eprintln!("ERR_AGENTOS_PYTHON_STDIN_CLOSE: {error}"); } if let Err(error) = self.inner.terminate() { eprintln!("ERR_AGENTOS_PYTHON_TERMINATE: {error}"); } } } #[derive(Debug)] pub struct PythonExecutionEngine { runtime: Option, next_context_id: usize, next_execution_id: usize, contexts: BTreeMap, import_caches: BTreeMap, javascript_context_ids: BTreeMap, javascript_engine: JavascriptExecutionEngine, } impl Default for PythonExecutionEngine { fn default() -> Self { let runtime = default_python_test_runtime_context(); let javascript_engine = runtime .as_ref() .map_or_else(JavascriptExecutionEngine::default, |runtime| { JavascriptExecutionEngine::new(runtime.clone()) }); Self { runtime, next_context_id: 0, next_execution_id: 0, contexts: BTreeMap::new(), import_caches: BTreeMap::new(), javascript_context_ids: BTreeMap::new(), javascript_engine, } } } #[cfg(test)] fn default_python_test_runtime_context() -> Option { agentos_runtime::SidecarRuntime::process(&agentos_runtime::RuntimeConfig::default()) .ok() .map(agentos_runtime::SidecarRuntime::context) } #[cfg(not(test))] fn default_python_test_runtime_context() -> Option { None } impl PythonExecutionEngine { pub fn new(runtime: RuntimeContext) -> Self { Self { runtime: Some(runtime.clone()), next_context_id: 0, next_execution_id: 0, contexts: BTreeMap::new(), import_caches: BTreeMap::new(), javascript_context_ids: BTreeMap::new(), javascript_engine: JavascriptExecutionEngine::new(runtime), } } pub fn set_runtime_context(&mut self, runtime: RuntimeContext) { self.javascript_engine.set_runtime_context(runtime.clone()); self.runtime = Some(runtime); } fn runtime_context(&self) -> Result<&RuntimeContext, PythonExecutionError> { self.runtime.as_ref().ok_or_else(|| { PythonExecutionError::Spawn(std::io::Error::other( "ERR_AGENTOS_RUNTIME_NOT_INJECTED: PythonExecutionEngine requires a process RuntimeContext; construct it with PythonExecutionEngine::new(runtime)", )) }) } pub fn set_event_notify(&mut self, notify: Option>) { self.javascript_engine.set_event_notify(notify); } pub fn bundled_pyodide_dist_path_for_vm( &mut self, vm_id: &str, ) -> Result { ensure_pyodide_available()?; let runtime = self.runtime_context()?.clone(); let import_cache = self.import_caches.entry(vm_id.to_owned()).or_default(); import_cache .ensure_materialized_with_runtime(&runtime) .map_err(PythonExecutionError::PrepareRuntime)?; Ok(import_cache.pyodide_dist_path().to_path_buf()) } pub async fn bundled_pyodide_dist_path_for_vm_async( &mut self, vm_id: &str, runtime: &RuntimeContext, ) -> Result { ensure_pyodide_available()?; let import_cache = self.import_caches.entry(vm_id.to_owned()).or_default(); import_cache .ensure_materialized_with_timeout_and_runtime_async(runtime, PYTHON_PREWARM_TIMEOUT) .await .map_err(PythonExecutionError::PrepareRuntime)?; Ok(import_cache.pyodide_dist_path().to_path_buf()) } pub fn create_context(&mut self, request: CreatePythonContextRequest) -> PythonContext { self.next_context_id += 1; self.import_caches.entry(request.vm_id.clone()).or_default(); let javascript_context = self.javascript_engine .create_context(CreateJavascriptContextRequest { vm_id: request.vm_id.clone(), bootstrap_module: None, compile_cache_root: None, }); let context = PythonContext { context_id: format!("python-ctx-{}", self.next_context_id), vm_id: request.vm_id, pyodide_dist_path: request.pyodide_dist_path, }; self.javascript_context_ids .insert(context.context_id.clone(), javascript_context.context_id); self.contexts .insert(context.context_id.clone(), context.clone()); context } /// Dispose the Python context and its private JavaScript bridge context. /// Live executions retain their own bridge session and are unaffected. pub fn dispose_context(&mut self, context_id: &str) -> bool { let removed = self.contexts.remove(context_id).is_some(); if let Some(javascript_context_id) = self.javascript_context_ids.remove(context_id) { self.javascript_engine .dispose_context(&javascript_context_id); } removed } #[doc(hidden)] pub fn context_count_for_test(&self) -> usize { self.contexts.len() } #[doc(hidden)] pub fn javascript_context_count_for_test(&self) -> usize { self.javascript_engine.context_count_for_test() } pub fn start_execution( &mut self, request: StartPythonExecutionRequest, ) -> Result { let runtime = self.runtime_context()?.clone(); self.create_execution_with_runtime(request, runtime, false) } pub fn prepare_execution( &mut self, request: StartPythonExecutionRequest, ) -> Result { let runtime = self.runtime_context()?.clone(); self.create_execution_with_runtime(request, runtime, true) } pub fn start_execution_with_runtime( &mut self, request: StartPythonExecutionRequest, runtime: RuntimeContext, ) -> Result { self.create_execution_with_runtime(request, runtime, false) } fn create_execution_with_runtime( &mut self, request: StartPythonExecutionRequest, runtime: RuntimeContext, defer_execute: bool, ) -> Result { ensure_pyodide_available()?; let context = self .contexts .get(&request.context_id) .cloned() .ok_or_else(|| PythonExecutionError::MissingContext(request.context_id.clone()))?; if context.vm_id != request.vm_id { return Err(PythonExecutionError::VmMismatch { expected: context.vm_id, found: request.vm_id, }); } let frozen_time_ms = frozen_time_ms(); let javascript_context_id = self .javascript_context_ids .get(&context.context_id) .cloned() .ok_or_else(|| PythonExecutionError::MissingContext(context.context_id.clone()))?; let warmup_metrics = { let import_cache = self.import_caches.entry(context.vm_id.clone()).or_default(); import_cache .ensure_materialized_with_runtime(&runtime) .map_err(PythonExecutionError::PrepareRuntime)?; prewarm_python_path( import_cache, &mut self.javascript_engine, &javascript_context_id, &context, &request, frozen_time_ms, &runtime, )? }; self.finish_start_execution( request, runtime, &context, javascript_context_id, frozen_time_ms, warmup_metrics, defer_execute, ) } /// Start Python from an async sidecar dispatch without parking a trusted /// Tokio worker during import-cache materialization or V8 prewarm. pub async fn start_execution_with_runtime_async( &mut self, request: StartPythonExecutionRequest, runtime: RuntimeContext, ) -> Result { ensure_pyodide_available()?; let context = self .contexts .get(&request.context_id) .cloned() .ok_or_else(|| PythonExecutionError::MissingContext(request.context_id.clone()))?; if context.vm_id != request.vm_id { return Err(PythonExecutionError::VmMismatch { expected: context.vm_id, found: request.vm_id, }); } let frozen_time_ms = frozen_time_ms(); let javascript_context = self.javascript_engine .create_context(CreateJavascriptContextRequest { vm_id: request.vm_id.clone(), bootstrap_module: None, compile_cache_root: None, }); let javascript_context_id = javascript_context.context_id.clone(); self.javascript_context_ids .insert(context.context_id.clone(), javascript_context_id.clone()); let warmup_metrics = { let import_cache = self.import_caches.entry(context.vm_id.clone()).or_default(); import_cache .ensure_materialized_with_timeout_and_runtime_async( &runtime, PYTHON_PREWARM_TIMEOUT, ) .await .map_err(PythonExecutionError::PrepareRuntime)?; prewarm_python_path_async( import_cache, &mut self.javascript_engine, &javascript_context_id, &context, &request, frozen_time_ms, &runtime, ) .await? }; self.finish_start_execution( request, runtime, &context, javascript_context_id, frozen_time_ms, warmup_metrics, false, ) } #[allow(clippy::too_many_arguments)] fn finish_start_execution( &mut self, request: StartPythonExecutionRequest, runtime: RuntimeContext, context: &PythonContext, javascript_context_id: String, frozen_time_ms: u128, warmup_metrics: Option>, defer_execute: bool, ) -> Result { self.next_execution_id += 1; let execution_id = format!("exec-{}", self.next_execution_id); let import_cache = self .import_caches .get(&context.vm_id) .expect("vm import cache should exist after materialization"); let pyodide_dist_path = resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd); let javascript_execution = start_python_javascript_execution( &mut self.javascript_engine, &runtime, import_cache, &javascript_context_id, context, &request, PythonJavascriptExecutionOptions { frozen_time_ms, prewarm_only: false, warmup_metrics: warmup_metrics.as_deref(), defer_execute, }, )?; let pending_vfs_rpc = Arc::new(Mutex::new(None)); let vfs_rpc_timeout = python_vfs_rpc_timeout(&request); Ok(PythonExecution { runtime, execution_id, child_pid: javascript_execution.child_pid(), v8_session: javascript_execution.v8_session_handle(), inner: javascript_execution, pyodide_dist_path, managed_host_files: PythonManagedHostFiles::new(python_managed_host_file_limit( &request, )), pending_vfs_rpc, output_buffer_max_bytes: python_output_buffer_max_bytes(&request), execution_timeout: python_execution_timeout(&request), vfs_rpc_timeout, }) } pub fn dispose_vm(&mut self, vm_id: &str) { self.contexts.retain(|_, context| context.vm_id != vm_id); self.javascript_context_ids .retain(|python_context_id, _| self.contexts.contains_key(python_context_id)); self.import_caches.remove(vm_id); self.javascript_engine.dispose_vm(vm_id); } } fn set_pending_vfs_rpc_state( pending_vfs_rpc: &Arc>>, id: u64, ) -> Result<(), PythonExecutionError> { let mut pending = pending_vfs_rpc .lock() .map_err(|_| PythonExecutionError::EventChannelClosed)?; if let Some(PendingVfsRpc { state: PendingVfsRpcState::Pending(current), .. }) = pending.as_ref() { return Err(PythonExecutionError::PendingVfsRpcRequest(*current)); } if let Some(previous) = pending.take() { if let Some(timeout_abort) = previous.timeout_abort { timeout_abort.abort(); } } *pending = Some(PendingVfsRpc { state: PendingVfsRpcState::Pending(id), timeout_abort: None, }); Ok(()) } fn map_javascript_error(error: JavascriptExecutionError) -> PythonExecutionError { match error { JavascriptExecutionError::EmptyArgv => PythonExecutionError::Spawn(std::io::Error::new( std::io::ErrorKind::InvalidInput, "guest Python bootstrap requires a JavaScript entrypoint", )), JavascriptExecutionError::InvalidLimit(message) => { PythonExecutionError::InvalidLimit(message) } JavascriptExecutionError::MissingContext(context_id) => { PythonExecutionError::MissingContext(context_id) } JavascriptExecutionError::VmMismatch { expected, found } => { PythonExecutionError::VmMismatch { expected, found } } JavascriptExecutionError::PrepareImportCache(error) => { PythonExecutionError::PrepareRuntime(error) } JavascriptExecutionError::Spawn(error) => PythonExecutionError::Spawn(error), JavascriptExecutionError::PendingSyncRpcRequest(id) => { PythonExecutionError::PendingVfsRpcRequest(id) } JavascriptExecutionError::ExpiredSyncRpcRequest(id) => { PythonExecutionError::RpcResponse(format!("VFS RPC request {id} is no longer pending")) } JavascriptExecutionError::RpcResponse(message) => { PythonExecutionError::RpcResponse(message) } JavascriptExecutionError::Terminate(error) => PythonExecutionError::Kill(error), JavascriptExecutionError::Control(error) => PythonExecutionError::Control(error), JavascriptExecutionError::StdinClosed => PythonExecutionError::StdinClosed, JavascriptExecutionError::Stdin(error) => PythonExecutionError::Stdin(error), JavascriptExecutionError::OutputBufferExceeded { stream, limit } => { PythonExecutionError::OutputBufferExceeded { stream, limit } } JavascriptExecutionError::EventChannelClosed => PythonExecutionError::EventChannelClosed, } } struct PythonJavascriptExecutionOptions<'a> { frozen_time_ms: u128, prewarm_only: bool, warmup_metrics: Option<&'a [u8]>, defer_execute: bool, } fn start_python_javascript_execution( javascript_engine: &mut JavascriptExecutionEngine, runtime: &RuntimeContext, import_cache: &NodeImportCache, javascript_context_id: &str, context: &PythonContext, request: &StartPythonExecutionRequest, options: PythonJavascriptExecutionOptions<'_>, ) -> Result { let internal_env = build_python_internal_env( import_cache, context, request, options.frozen_time_ms, options.prewarm_only, ); let inline_code = build_python_runner_module_source(import_cache, &internal_env, options.warmup_metrics)?; let mut env = request.env.clone(); env.extend(internal_env); // The Pyodide runner is itself a V8 execution. Its heap cap (the Python // `maxOldSpaceMb` knob) and sync-RPC wait ceiling ride the typed runner // limits, not env — the JS engine reads them from `limits`, not `AGENTOS_*`. let max_old_space_mb = python_max_old_space_mb(request); let runner_limits = python_runner_javascript_limits(&request.limits, max_old_space_mb); let javascript_request = StartJavascriptExecutionRequest { vm_id: request.vm_id.clone(), context_id: javascript_context_id.to_owned(), argv: vec![import_cache.python_runner_path().display().to_string()], argv0: None, env, cwd: request.cwd.clone(), limits: runner_limits, // Forward the guest-runtime identity so the runner's shim sets // process.* from typed config rather than env. guest_runtime: request.guest_runtime.clone(), wasm_module_bytes: None, inline_code: Some(inline_code), }; if options.defer_execute { javascript_engine.prepare_execution_with_runtime(javascript_request, runtime.clone()) } else { javascript_engine.start_execution_with_runtime(javascript_request, runtime.clone()) } .map_err(map_javascript_error) } fn python_runner_javascript_limits( limits: &PythonExecutionLimits, max_old_space_mb: usize, ) -> JavascriptExecutionLimits { JavascriptExecutionLimits { v8_heap_limit_mb: (max_old_space_mb > 0).then_some(max_old_space_mb as u32), sync_rpc_wait_timeout_ms: Some(PYTHON_SYNC_RPC_WAIT_TIMEOUT_MS), reactor_work_quantum: limits.reactor_work_quantum, bridge_call_timeout_ms: limits.bridge_call_timeout_ms, ..JavascriptExecutionLimits::default() } } fn build_python_internal_env( import_cache: &NodeImportCache, context: &PythonContext, request: &StartPythonExecutionRequest, frozen_time_ms: u128, prewarm_only: bool, ) -> BTreeMap { let mut internal_env = request .env .iter() .filter(|(key, _)| key.starts_with("AGENTOS_")) .map(|(key, value)| (key.clone(), value.clone())) .collect::>(); let pyodide_dist_path = resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd); add_python_guest_path_mapping(&mut internal_env, &pyodide_dist_path); internal_env.insert( PYODIDE_INDEX_URL_ENV.to_string(), String::from(PYODIDE_GUEST_ROOT), ); internal_env.insert( PYODIDE_PACKAGE_BASE_URL_ENV.to_string(), request .env .get(PYODIDE_PACKAGE_BASE_URL_ENV) .cloned() .unwrap_or_else(|| String::from(PYODIDE_GUEST_ROOT)), ); internal_env.insert( PYODIDE_PACKAGE_CACHE_DIR_ENV.to_string(), String::from(PYODIDE_CACHE_GUEST_ROOT), ); internal_env.insert( NODE_IMPORT_CACHE_ASSET_ROOT_ENV.to_string(), import_cache.asset_root().display().to_string(), ); internal_env.insert( NODE_ALLOW_PROCESS_BINDINGS_ENV.to_string(), String::from("1"), ); internal_env.insert( NODE_SYNC_RPC_DATA_BYTES_ENV.to_string(), PYTHON_SYNC_RPC_DATA_BYTES.to_string(), ); internal_env.insert( NODE_DISABLE_COMPILE_CACHE_ENV.to_string(), String::from("1"), ); // The runner's V8 heap cap and sync-RPC wait timeout are carried as typed // `JavascriptExecutionLimits` on the runner request (see the launch site), // not as `AGENTOS_V8_HEAP_LIMIT_MB` / `AGENTOS_NODE_SYNC_RPC_WAIT_TIMEOUT_MS` // env knobs, which the JS engine no longer reads. internal_env.insert(PYTHON_CODE_ENV.to_string(), request.code.clone()); internal_env.insert(NODE_FROZEN_TIME_ENV.to_string(), frozen_time_ms.to_string()); if prewarm_only { internal_env.insert(PYTHON_PREWARM_ONLY_ENV.to_string(), String::from("1")); } else { internal_env.insert(String::from("AGENTOS_KEEP_STDIN_OPEN"), String::from("1")); internal_env.insert( String::from("AGENTOS_FORWARD_KERNEL_STDIN_RPC"), String::from("1"), ); internal_env.remove(PYTHON_PREWARM_ONLY_ENV); } if let Some(file_path) = &request.file_path { internal_env.insert(PYTHON_FILE_ENV.to_string(), file_path.display().to_string()); } else { internal_env.remove(PYTHON_FILE_ENV); } internal_env } fn add_python_guest_path_mapping( internal_env: &mut BTreeMap, pyodide_dist_path: &Path, ) { let pyodide_cache_path = pyodide_cache_path(pyodide_dist_path); let mut mappings = internal_env .get(NODE_GUEST_PATH_MAPPINGS_ENV) .and_then(|value| serde_json::from_str::>(value).ok()) .unwrap_or_default(); mappings.retain(|mapping| { mapping.guest_path != PYODIDE_GUEST_ROOT && mapping.guest_path != PYODIDE_CACHE_GUEST_ROOT }); mappings.push(PythonGuestPathMappingWire { guest_path: String::from(PYODIDE_GUEST_ROOT), host_path: pyodide_dist_path.display().to_string(), }); mappings.push(PythonGuestPathMappingWire { guest_path: String::from(PYODIDE_CACHE_GUEST_ROOT), host_path: pyodide_cache_path.display().to_string(), }); let serialized = serde_json::to_string(&mappings).unwrap_or_else(|_| String::from("[]")); internal_env.insert(String::from(NODE_GUEST_PATH_MAPPINGS_ENV), serialized); } fn pyodide_cache_path(pyodide_dist_path: &Path) -> PathBuf { let base = pyodide_dist_path .parent() .and_then(|parent| { if parent.file_name().is_some_and(|name| name == "assets") { parent.parent() } else { Some(parent) } }) .unwrap_or(pyodide_dist_path); base.join("pyodide-package-cache") } fn build_python_runner_module_source( import_cache: &NodeImportCache, internal_env: &BTreeMap, warmup_metrics: Option<&[u8]>, ) -> Result { let runner_source = fs::read_to_string(import_cache.python_runner_path()) .map_err(PythonExecutionError::PrepareRuntime)?; let runner_source = format!("import * as __agentOSConstantsBinding from 'node:constants';\n{runner_source}"); let bootstrap = build_python_runner_bootstrap(internal_env, warmup_metrics); Ok(insert_python_runner_bootstrap(&runner_source, &bootstrap)) } fn build_python_runner_bootstrap( internal_env: &BTreeMap, warmup_metrics: Option<&[u8]>, ) -> String { let internal_env_json = serde_json::to_string(internal_env).unwrap_or_else(|_| String::from("{}")); let warmup_metrics_json = warmup_metrics.map(|bytes| { serde_json::to_string(&String::from_utf8_lossy(bytes).to_string()) .unwrap_or_else(|_| String::from("\"\"")) }); match warmup_metrics_json { Some(warmup_metrics_json) => format!( "globalThis.__agentOSPythonInternalEnv = {internal_env_json};\n\ if (typeof process !== 'undefined') {{\n process.env = {{ ...(process.env || {{}}), ...globalThis.__agentOSPythonInternalEnv }};\n}}\n\ if (typeof process?.stderr?.write === 'function') {{\n process.stderr.write({warmup_metrics_json});\n}}\n" ), None => format!( "globalThis.__agentOSPythonInternalEnv = {internal_env_json};\n\ if (typeof process !== 'undefined') {{\n process.env = {{ ...(process.env || {{}}), ...globalThis.__agentOSPythonInternalEnv }};\n}}\n" ), } } fn insert_python_runner_bootstrap(source: &str, bootstrap: &str) -> String { let mut insert_at = 0usize; let mut saw_import = false; for line in source.split_inclusive('\n') { let trimmed = line.trim_start(); if trimmed.starts_with("import ") || (saw_import && trimmed.is_empty()) { insert_at += line.len(); saw_import = saw_import || trimmed.starts_with("import "); continue; } break; } format!( "{}{}{}", &source[..insert_at], bootstrap, &source[insert_at..] ) } fn parse_python_bridge_sync_rpc_request( request: &JavascriptSyncRpcRequest, ) -> Result { if request.method != "_pythonRpc" { return Err(PythonExecutionError::RpcResponse(format!( "unexpected JavaScript sync RPC method for guest Python runtime: {}", request.method ))); } let payload = request.args.first().ok_or_else(|| { PythonExecutionError::RpcResponse(String::from( "guest Python bridge call did not include a request payload", )) })?; let wire: PythonVfsBridgeRequestWire = serde_json::from_value(payload.clone()).map_err(|error| { PythonExecutionError::RpcResponse(format!( "invalid guest Python bridge request payload: {error}" )) })?; let method = PythonVfsRpcMethod::from_wire(&wire.method).ok_or_else(|| { PythonExecutionError::RpcResponse(format!( "unsupported agentos python rpc method {} for {}", wire.method, request.id )) })?; Ok(PythonVfsRpcRequest { id: request.id, method, path: wire.path, destination: wire.destination, target: wire.target, mode: wire.mode.map(|value| value as u32), uid: wire.uid.map(|value| value as u32), gid: wire.gid.map(|value| value as u32), atime_ms: wire.atime_ms.map(|value| value as u64), mtime_ms: wire.mtime_ms.map(|value| value as u64), content_base64: wire.content_base64, recursive: wire.recursive, url: wire.url, http_method: wire.http_method, headers: wire.headers, body_base64: wire.body_base64, hostname: wire.hostname, family: wire.family, port: wire.port, socket_id: wire.socket_id, command: wire.command, args: wire.args, argv0: wire.argv0, cwd: wire.cwd, env: wire.env, shell: wire.shell, max_buffer: wire.max_buffer, timeout_ms: wire.timeout_ms, }) } #[derive(Debug)] struct PythonOutputBuffer { bytes: Vec, max_bytes: usize, } impl PythonOutputBuffer { fn new(max_bytes: usize) -> Self { Self { bytes: Vec::new(), max_bytes, } } fn extend(&mut self, chunk: &[u8]) { if self.bytes.len() >= self.max_bytes { return; } let remaining = self.max_bytes - self.bytes.len(); let take = remaining.min(chunk.len()); self.bytes.extend_from_slice(&chunk[..take]); } fn into_inner(self) -> Vec { self.bytes } } fn python_output_buffer_max_bytes(request: &StartPythonExecutionRequest) -> usize { request .limits .output_buffer_max_bytes .unwrap_or(DEFAULT_PYTHON_OUTPUT_BUFFER_MAX_BYTES) } fn python_execution_timeout(request: &StartPythonExecutionRequest) -> Option { match request.limits.execution_timeout_ms { // `Some(0)` explicitly disables the timeout. Some(0) => None, Some(value) => Some(Duration::from_millis(value)), None => Some(Duration::from_millis(DEFAULT_PYTHON_EXECUTION_TIMEOUT_MS)), } } fn python_max_old_space_mb(request: &StartPythonExecutionRequest) -> usize { request .limits .max_old_space_mb .filter(|value| *value > 0) .unwrap_or(DEFAULT_PYTHON_MAX_OLD_SPACE_MB) } fn python_vfs_rpc_timeout(request: &StartPythonExecutionRequest) -> Duration { Duration::from_millis( request .limits .vfs_rpc_timeout_ms .filter(|value| *value > 0) .unwrap_or(DEFAULT_PYTHON_VFS_RPC_TIMEOUT_MS), ) } fn spawn_python_vfs_rpc_timeout( runtime: &RuntimeContext, id: u64, timeout: Duration, pending: Arc>>, v8_session: crate::v8_host::V8SessionHandle, ) -> Result<(), PythonExecutionError> { let cancellation = runtime.clone(); let pending_for_task = Arc::clone(&pending); let handle = runtime .spawn(agentos_runtime::TaskClass::Timer, async move { tokio::select! { _ = tokio::time::sleep(timeout) => {} _ = cancellation.admission_closed() => { let mut guard = pending_for_task.lock().unwrap_or_else(|poisoned| { eprintln!( "ERR_AGENTOS_PYTHON_VFS_RPC_STATE_POISONED: recovering request {id} during runtime shutdown" ); poisoned.into_inner() }); if guard.as_ref().map(|rpc| rpc.state) == Some(PendingVfsRpcState::Pending(id)) { *guard = None; } return; } } let mut guard = pending_for_task.lock().unwrap_or_else(|poisoned| { eprintln!( "ERR_AGENTOS_PYTHON_VFS_RPC_STATE_POISONED: recovering request {id} while delivering its timeout" ); poisoned.into_inner() }); let should_timeout = if guard.as_ref().map(|rpc| rpc.state) == Some(PendingVfsRpcState::Pending(id)) { *guard = Some(PendingVfsRpc { state: PendingVfsRpcState::TimedOut(id), timeout_abort: None, }); true } else { false }; drop(guard); if !should_timeout { return; } if let Err(error) = v8_session.send_bridge_response( id, 1, format!( "ERR_AGENTOS_PYTHON_VFS_RPC_TIMEOUT: guest Python VFS RPC request {id} timed out after {}ms", timeout.as_millis() ) .into_bytes(), ) { eprintln!( "ERR_AGENTOS_PYTHON_VFS_RPC_TIMEOUT_DELIVERY: could not deliver timeout for request {id}: {error}" ); } }) .map_err(|error| { PythonExecutionError::RpcResponse(format!( "could not arm Python VFS RPC timeout for request {id}: {error}" )) })?; let timeout_abort = handle.abort_handle(); let mut guard = pending .lock() .map_err(|_| PythonExecutionError::EventChannelClosed)?; if let Some(rpc) = guard.as_mut() { if rpc.state == PendingVfsRpcState::Pending(id) { rpc.timeout_abort = Some(timeout_abort); return Ok(()); } } timeout_abort.abort(); Ok(()) } fn resolved_pyodide_dist_path(path: &Path, cwd: &Path) -> PathBuf { resolve_execution_path(path, cwd) } struct PythonPrewarmOutput { stdout: Vec, stderr: Vec, sync_rpc_log: Vec, managed_host_files: PythonManagedHostFiles, } impl PythonPrewarmOutput { fn new(max_open_fds: usize) -> Self { Self { stdout: Vec::new(), stderr: Vec::new(), sync_rpc_log: Vec::new(), managed_host_files: PythonManagedHostFiles::new(max_open_fds), } } } fn handle_python_prewarm_event( prewarm_execution: &mut JavascriptExecution, context: &PythonContext, request: &StartPythonExecutionRequest, event: Option, output: &mut PythonPrewarmOutput, ) -> Result, PythonExecutionError> { match event { Some(JavascriptExecutionEvent::Stdout(chunk)) => output.stdout.extend(chunk), Some(JavascriptExecutionEvent::Stderr(chunk)) => output.stderr.extend(chunk), Some(JavascriptExecutionEvent::Exited(exit_code)) => { return Ok(Some(PythonExecutionResult { execution_id: String::from("python-prewarm"), exit_code, stdout: std::mem::take(&mut output.stdout), stderr: std::mem::take(&mut output.stderr), })); } Some(JavascriptExecutionEvent::SignalState { .. }) => {} Some(JavascriptExecutionEvent::SyncRpcRequest(sync_request)) => { output.sync_rpc_log.push(format!( "{} {} {:?}", sync_request.id, sync_request.method, sync_request.args )); // The Python runner module imports node builtins and the pyodide // ESM entry; module-resolution sync RPCs are serviced host-directly // because this prewarm has no kernel/service-loop consumer. if prewarm_execution .try_service_standalone_module_sync_rpc(&sync_request) .map_err(map_javascript_error)? { output .sync_rpc_log .push(format!("responded {} (module)", sync_request.id)); return Ok(None); } let pyodide_dist_path = resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd); if let Some(action) = python_javascript_sync_rpc_action( &pyodide_dist_path, &mut output.managed_host_files, &sync_request, )? { respond_python_javascript_sync_rpc_action( prewarm_execution, sync_request.id, action, )?; output .sync_rpc_log .push(format!("responded {}", sync_request.id)); return Ok(None); } if let Some((code, message)) = python_javascript_sync_rpc_error(&sync_request) { prewarm_execution .respond_sync_rpc_error(sync_request.id, code, message) .map_err(map_javascript_error)?; output .sync_rpc_log .push(format!("errored {}", sync_request.id)); return Ok(None); } if sync_request.method == "_pythonRpc" { let request = parse_python_bridge_sync_rpc_request(&sync_request)?; return Err(PythonExecutionError::WarmupFailed { exit_code: 1, stderr: format!( "unexpected Python prewarm VFS RPC request {} {} {:?}", request.id, request.path, request.method ), }); } return Err(PythonExecutionError::WarmupFailed { exit_code: 1, stderr: format!( "unexpected Python prewarm JavaScript sync RPC request {} {} {:?}", sync_request.id, sync_request.method, sync_request.args ), }); } None => { return Err(PythonExecutionError::WarmupFailed { exit_code: 1, stderr: format!( "python prewarm timed out after {}s\nstdout:\n{}\nstderr:\n{}\nsync rpc:\n{}", PYTHON_PREWARM_TIMEOUT.as_secs(), String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr), output.sync_rpc_log.join("\n"), ), }); } } Ok(None) } fn prewarm_python_path( import_cache: &NodeImportCache, javascript_engine: &mut JavascriptExecutionEngine, javascript_context_id: &str, context: &PythonContext, request: &StartPythonExecutionRequest, frozen_time_ms: u128, runtime: &RuntimeContext, ) -> Result>, PythonExecutionError> { let debug_enabled = python_warmup_metrics_enabled(request); let marker_contents = warmup_marker_contents(import_cache, context, request); let marker_path = warmup_marker_path( import_cache.prewarm_marker_dir(), "python-runner-prewarm", PYTHON_WARMUP_MARKER_VERSION, &marker_contents, ); let marker_exists = marker_path.exists(); let started = Instant::now(); let mut prewarm_execution = start_python_javascript_execution( javascript_engine, runtime, import_cache, javascript_context_id, context, request, PythonJavascriptExecutionOptions { frozen_time_ms, prewarm_only: true, warmup_metrics: None, defer_execute: false, }, )?; let mut output = PythonPrewarmOutput::new(python_managed_host_file_limit(request)); let result = loop { let event = prewarm_execution .poll_event_blocking(PYTHON_PREWARM_TIMEOUT) .map_err(map_javascript_error)?; if let Some(result) = handle_python_prewarm_event( &mut prewarm_execution, context, request, event, &mut output, )? { break result; } }; let duration_ms = started.elapsed().as_secs_f64() * 1000.0; if result.exit_code != 0 { return Err(PythonExecutionError::WarmupFailed { exit_code: result.exit_code, stderr: String::from_utf8_lossy(&result.stderr).into_owned(), }); } if marker_exists { return Ok(warmup_metrics_line( debug_enabled, false, "cached", 0.0, import_cache, context, request, )); } fs::write(&marker_path, marker_contents).map_err(PythonExecutionError::PrepareWarmPath)?; Ok(warmup_metrics_line( debug_enabled, true, "executed", duration_ms, import_cache, context, request, )) } async fn prewarm_python_path_async( import_cache: &NodeImportCache, javascript_engine: &mut JavascriptExecutionEngine, javascript_context_id: &str, context: &PythonContext, request: &StartPythonExecutionRequest, frozen_time_ms: u128, runtime: &RuntimeContext, ) -> Result>, PythonExecutionError> { let debug_enabled = python_warmup_metrics_enabled(request); let marker_contents = warmup_marker_contents(import_cache, context, request); let marker_path = warmup_marker_path( import_cache.prewarm_marker_dir(), "python-runner-prewarm", PYTHON_WARMUP_MARKER_VERSION, &marker_contents, ); let marker_exists = marker_path.exists(); let started = Instant::now(); let mut prewarm_execution = start_python_javascript_execution( javascript_engine, runtime, import_cache, javascript_context_id, context, request, PythonJavascriptExecutionOptions { frozen_time_ms, prewarm_only: true, warmup_metrics: None, defer_execute: false, }, )?; let mut output = PythonPrewarmOutput::new(python_managed_host_file_limit(request)); let result = loop { let event = prewarm_execution .poll_event(PYTHON_PREWARM_TIMEOUT) .await .map_err(map_javascript_error)?; if let Some(result) = handle_python_prewarm_event( &mut prewarm_execution, context, request, event, &mut output, )? { break result; } }; let duration_ms = started.elapsed().as_secs_f64() * 1000.0; if result.exit_code != 0 { return Err(PythonExecutionError::WarmupFailed { exit_code: result.exit_code, stderr: String::from_utf8_lossy(&result.stderr).into_owned(), }); } if marker_exists { return Ok(warmup_metrics_line( debug_enabled, false, "cached", 0.0, import_cache, context, request, )); } fs::write(&marker_path, marker_contents).map_err(PythonExecutionError::PrepareWarmPath)?; Ok(warmup_metrics_line( debug_enabled, true, "executed", duration_ms, import_cache, context, request, )) } #[derive(Debug)] enum PythonJavascriptSyncRpcAction { Success(Value), RawSuccess(Vec), Error { code: &'static str, message: String }, } #[derive(Debug)] struct PythonManagedHostFiles { next_fd: u64, max_files: usize, files: BTreeMap, } // Standalone execution callers do not have a VM kernel from which to source // policy. Keep their fallback bounded; production sidecar launches always pass // `vm.kernel.resource_limits().max_open_fds` in `PythonExecutionLimits`. const DEFAULT_PYTHON_MANAGED_HOST_FILE_LIMIT: usize = 256; impl Default for PythonManagedHostFiles { fn default() -> Self { Self::new(DEFAULT_PYTHON_MANAGED_HOST_FILE_LIMIT) } } impl PythonManagedHostFiles { fn new(max_files: usize) -> Self { Self { // Keep host-direct descriptors disjoint from ordinary guest and // stdio descriptors. They never cross the JavaScript execution. next_fd: 0x4000_0000, max_files, files: BTreeMap::new(), } } } fn python_managed_host_file_limit(request: &StartPythonExecutionRequest) -> usize { request .limits .max_open_fds .unwrap_or(DEFAULT_PYTHON_MANAGED_HOST_FILE_LIMIT) } fn python_javascript_sync_rpc_action( pyodide_dist_path: &Path, managed_host_files: &mut PythonManagedHostFiles, request: &JavascriptSyncRpcRequest, ) -> Result, PythonExecutionError> { if matches!(request.method.as_str(), "fs.readSync" | "_fsReadRaw") { let Some(fd) = request.args.first().and_then(Value::as_u64) else { return Ok(None); }; let Some(file) = managed_host_files.files.get_mut(&fd) else { return Ok(None); }; let length = request .args .get(1) .and_then(Value::as_u64) .and_then(|length| usize::try_from(length).ok()) .ok_or_else(|| { PythonExecutionError::RpcResponse(String::from( "managed fs.readSync length must fit within usize", )) })?; if length > PYTHON_SYNC_RPC_DATA_BYTES { return Err(PythonExecutionError::RpcResponse(format!( "managed fs.readSync length {length} exceeds {PYTHON_SYNC_RPC_DATA_BYTES} bytes" ))); } if let Some(position) = request.args.get(2).and_then(Value::as_u64) { file.seek(SeekFrom::Start(position)) .map_err(PythonExecutionError::PrepareRuntime)?; } let mut bytes = vec![0; length]; let bytes_read = file .read(&mut bytes) .map_err(PythonExecutionError::PrepareRuntime)?; bytes.truncate(bytes_read); return Ok(Some(if request.raw_bytes_args.contains_key(&usize::MAX) { PythonJavascriptSyncRpcAction::RawSuccess(bytes) } else { PythonJavascriptSyncRpcAction::Success(Value::String(v8_runtime::base64_encode_pub( &bytes, ))) })); } if request.method == "fs.closeSync" { let Some(fd) = request.args.first().and_then(Value::as_u64) else { return Ok(None); }; if managed_host_files.files.remove(&fd).is_none() { return Ok(None); } return Ok(Some(PythonJavascriptSyncRpcAction::Success(Value::Null))); } if request.method == "fs.fstatSync" { let Some(fd) = request.args.first().and_then(Value::as_u64) else { return Ok(None); }; let Some(file) = managed_host_files.files.get(&fd) else { return Ok(None); }; let metadata = file .metadata() .map_err(PythonExecutionError::PrepareRuntime)?; return Ok(Some(PythonJavascriptSyncRpcAction::Success( python_host_stat_value(&metadata), ))); } let Some(path) = request.args.first().and_then(Value::as_str) else { return Ok(None); }; let path_kind = python_managed_path_kind(pyodide_dist_path, path); let Some(host_path) = path_kind.host_path() else { return Ok(None); }; Ok(Some(match request.method.as_str() { "fs.openSync" => { let flags = request.args.get(1).unwrap_or(&Value::Null); let read_only = matches!(flags.as_str(), Some("r")) || flags.as_u64().is_some_and(|flags| flags == 0); if !read_only { PythonJavascriptSyncRpcAction::Error { code: "EROFS", message: format!( "EROFS: managed Python runtime assets are read-only, open '{path}'" ), } } else if managed_host_files.files.len() >= managed_host_files.max_files { PythonJavascriptSyncRpcAction::Error { code: "EMFILE", message: format!( "EMFILE: managed Python host descriptor limit {} reached (limits.resources.maxOpenFds); raise limits.resources.maxOpenFds", managed_host_files.max_files ), } } else { match fs::File::open(&host_path) { Ok(file) => { let fd = managed_host_files.next_fd; managed_host_files.next_fd = managed_host_files.next_fd.checked_add(1).ok_or_else(|| { PythonExecutionError::RpcResponse(String::from( "managed Python host descriptor ids exhausted", )) })?; managed_host_files.files.insert(fd, file); PythonJavascriptSyncRpcAction::Success(json!(fd)) } Err(error) => { return python_sync_rpc_fs_action_error(path, "open", error).map(Some); } } } } "fs.promises.readFile" | "fs.readFileSync" => { let bytes = match fs::read(&host_path) { Ok(bytes) => bytes, Err(error) => { return python_sync_rpc_fs_action_error(path, "open", error).map(Some); } }; let encoding = python_prewarm_sync_rpc_encoding(&request.args); match encoding.as_deref() { Some("utf8") | Some("utf-8") => PythonJavascriptSyncRpcAction::Success( Value::String(String::from_utf8_lossy(&bytes).into_owned()), ), _ => PythonJavascriptSyncRpcAction::Success(json!({ "__agentOSType": "bytes", "base64": v8_runtime::base64_encode_pub(&bytes), })), } } "fs.statSync" | "fs.promises.stat" => match fs::metadata(&host_path) { Ok(metadata) => { PythonJavascriptSyncRpcAction::Success(python_host_stat_value(&metadata)) } Err(error) => return python_sync_rpc_fs_action_error(path, "stat", error).map(Some), }, "fs.lstatSync" | "fs.promises.lstat" => match fs::symlink_metadata(&host_path) { Ok(metadata) => { PythonJavascriptSyncRpcAction::Success(python_host_stat_value(&metadata)) } Err(error) => return python_sync_rpc_fs_action_error(path, "lstat", error).map(Some), }, "fs.existsSync" => PythonJavascriptSyncRpcAction::Success(Value::Bool(host_path.exists())), "fs.accessSync" | "fs.promises.access" => match fs::metadata(&host_path) { Ok(_) => PythonJavascriptSyncRpcAction::Success(Value::Null), Err(error) => return python_sync_rpc_fs_action_error(path, "access", error).map(Some), }, "fs.readdirSync" | "fs.promises.readdir" => match fs::read_dir(&host_path) { Ok(entries) => PythonJavascriptSyncRpcAction::Success(python_readdir_value( entries .filter_map(|entry| entry.ok()) .filter_map(|entry| entry.file_name().into_string().ok()) .collect(), )), Err(error) => return python_sync_rpc_fs_action_error(path, "scandir", error).map(Some), }, "fs.mkdirSync" | "fs.promises.mkdir" => { let recursive = python_sync_rpc_recursive_flag(&request.args); if recursive { fs::create_dir_all(&host_path).map_err(PythonExecutionError::PrepareRuntime)?; } else { match fs::create_dir(&host_path) { Ok(()) => {} Err(error) => { return python_sync_rpc_fs_action_error(path, "mkdir", error).map(Some); } } } PythonJavascriptSyncRpcAction::Success(Value::Null) } "fs.writeFileSync" | "fs.promises.writeFile" => { let contents = python_sync_rpc_bytes_arg(&request.args, 1)?; if let Some(parent) = host_path.parent() { fs::create_dir_all(parent).map_err(PythonExecutionError::PrepareRuntime)?; } fs::write(&host_path, contents).map_err(PythonExecutionError::PrepareRuntime)?; PythonJavascriptSyncRpcAction::Success(Value::Null) } "fs.realpathSync" | "fs.realpathSync.native" => match fs::canonicalize(&host_path) { Ok(canonical) => PythonJavascriptSyncRpcAction::Success(Value::String( path_kind.render_path(pyodide_dist_path, &canonical, path), )), Err(error) => { return python_sync_rpc_fs_action_error(path, "realpath", error).map(Some); } }, _ => return Ok(None), })) } fn python_sync_rpc_fs_action_error( path: &str, syscall: &str, error: std::io::Error, ) -> Result { let action = match error.kind() { std::io::ErrorKind::NotFound => PythonJavascriptSyncRpcAction::Error { code: "ENOENT", message: format!("ENOENT: no such file or directory, {syscall} '{path}'"), }, std::io::ErrorKind::AlreadyExists => PythonJavascriptSyncRpcAction::Error { code: "EEXIST", message: format!("EEXIST: file already exists, {syscall} '{path}'"), }, std::io::ErrorKind::PermissionDenied => PythonJavascriptSyncRpcAction::Error { code: "EACCES", message: format!("EACCES: permission denied, {syscall} '{path}'"), }, _ => { return Err(PythonExecutionError::PrepareRuntime(std::io::Error::new( error.kind(), error.to_string(), ))); } }; Ok(action) } fn respond_python_javascript_sync_rpc_action( execution: &mut JavascriptExecution, id: u64, action: PythonJavascriptSyncRpcAction, ) -> Result<(), PythonExecutionError> { match action { PythonJavascriptSyncRpcAction::Success(value) => execution .respond_sync_rpc_success(id, value) .map_err(map_javascript_error), PythonJavascriptSyncRpcAction::RawSuccess(bytes) => execution .respond_sync_rpc_raw_success(id, bytes) .map_err(map_javascript_error), PythonJavascriptSyncRpcAction::Error { code, message } => execution .respond_sync_rpc_error(id, code, message) .map_err(map_javascript_error), } } #[derive(Debug, Clone)] enum PythonManagedPathKind { GuestPyodide, GuestCache, HostManaged, Unmanaged, } impl PythonManagedPathKind { fn render_path(&self, pyodide_dist_path: &Path, canonical: &Path, original: &str) -> String { match self { Self::GuestPyodide | Self::GuestCache => { python_host_path_to_guest(pyodide_dist_path, canonical) .unwrap_or_else(|| original.to_owned()) } Self::HostManaged => canonical.display().to_string(), Self::Unmanaged => original.to_owned(), } } } fn python_managed_path_kind(pyodide_dist_path: &Path, path: &str) -> PythonManagedResolvedPath { let cache_path = pyodide_cache_path(pyodide_dist_path); if let Some(normalized) = strip_guest_managed_root(path, PYODIDE_GUEST_ROOT) { let root = canonicalize_existing_or_self(pyodide_dist_path); let relative = normalize_relative_guest_suffix(normalized); let host_path = if relative.as_os_str().is_empty() { root.clone() } else { root.join(relative) }; if confined_managed_path(&host_path, &root) { return PythonManagedResolvedPath { kind: PythonManagedPathKind::GuestPyodide, host_path: Some(host_path), }; } return PythonManagedResolvedPath { kind: PythonManagedPathKind::Unmanaged, host_path: None, }; } if let Some(normalized) = strip_guest_managed_root(path, PYODIDE_CACHE_GUEST_ROOT) { let root = canonicalize_existing_or_self(&cache_path); let relative = normalize_relative_guest_suffix(normalized); let host_path = if relative.as_os_str().is_empty() { root.clone() } else { root.join(relative) }; if confined_managed_path(&host_path, &root) { return PythonManagedResolvedPath { kind: PythonManagedPathKind::GuestCache, host_path: Some(host_path), }; } return PythonManagedResolvedPath { kind: PythonManagedPathKind::Unmanaged, host_path: None, }; } let candidate = PathBuf::from(path); let pyodide_root = canonicalize_existing_or_self(pyodide_dist_path); let cache_root = canonicalize_existing_or_self(&cache_path); if candidate.is_absolute() && !path_has_parent_or_prefix_component(&candidate) && confined_managed_path(&candidate, &pyodide_root) { return PythonManagedResolvedPath { kind: PythonManagedPathKind::HostManaged, host_path: Some(candidate), }; } if candidate.is_absolute() && !path_has_parent_or_prefix_component(&candidate) && confined_managed_path(&candidate, &cache_root) { return PythonManagedResolvedPath { kind: PythonManagedPathKind::HostManaged, host_path: Some(candidate), }; } PythonManagedResolvedPath { kind: PythonManagedPathKind::Unmanaged, host_path: None, } } #[derive(Debug, Clone)] struct PythonManagedResolvedPath { kind: PythonManagedPathKind, host_path: Option, } impl PythonManagedResolvedPath { fn host_path(&self) -> Option { self.host_path.clone() } fn render_path(&self, pyodide_dist_path: &Path, canonical: &Path, original: &str) -> String { self.kind .render_path(pyodide_dist_path, canonical, original) } } fn strip_guest_managed_root<'a>(path: &'a str, root: &str) -> Option<&'a str> { if path == root { return Some(""); } path.strip_prefix(root)?.strip_prefix('/') } fn normalize_relative_guest_suffix(suffix: &str) -> PathBuf { let mut normalized = PathBuf::new(); for segment in suffix.trim_start_matches('/').split('/') { if segment.is_empty() || segment == "." { continue; } if segment == ".." { normalized.pop(); } else { normalized.push(segment); } } normalized } fn path_has_parent_or_prefix_component(path: &Path) -> bool { path.components() .any(|component| matches!(component, Component::ParentDir | Component::Prefix(_))) } fn canonicalize_existing_or_self(path: &Path) -> PathBuf { fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()) } fn confined_managed_path(path: &Path, root: &Path) -> bool { let canonical_root = canonicalize_existing_or_self(root); let Some(canonical_path) = canonicalize_managed_candidate(path) else { return false; }; canonical_path == canonical_root || canonical_path.starts_with(canonical_root) } fn canonicalize_managed_candidate(path: &Path) -> Option { let mut missing_components = Vec::new(); let mut current = path; loop { match fs::canonicalize(current) { Ok(mut canonical) => { for component in missing_components.iter().rev() { canonical.push(component); } return Some(canonical); } Err(_) => { let file_name = current.file_name()?.to_owned(); if Path::new(&file_name) .components() .any(|component| !matches!(component, Component::Normal(_))) { return None; } missing_components.push(file_name); current = current.parent()?; } } } } fn python_host_path_to_guest(pyodide_dist_path: &Path, host_path: &Path) -> Option { if let Ok(relative) = host_path.strip_prefix(pyodide_dist_path) { let suffix = relative.to_string_lossy().replace('\\', "/"); return Some(if suffix.is_empty() { String::from(PYODIDE_GUEST_ROOT) } else { format!("{PYODIDE_GUEST_ROOT}/{suffix}") }); } let cache_path = pyodide_cache_path(pyodide_dist_path); let relative = host_path.strip_prefix(cache_path).ok()?; let suffix = relative.to_string_lossy().replace('\\', "/"); Some(if suffix.is_empty() { String::from(PYODIDE_CACHE_GUEST_ROOT) } else { format!("{PYODIDE_CACHE_GUEST_ROOT}/{suffix}") }) } fn python_host_stat_value(metadata: &fs::Metadata) -> Value { json!({ "mode": metadata.mode(), "size": metadata.size(), "blocks": metadata.blocks(), "dev": metadata.dev(), "rdev": metadata.rdev(), "isDirectory": metadata.is_dir(), "isSymbolicLink": metadata.file_type().is_symlink(), "atimeMs": metadata.atime() * 1000 + (metadata.atime_nsec() / 1_000_000), "mtimeMs": metadata.mtime() * 1000 + (metadata.mtime_nsec() / 1_000_000), "ctimeMs": metadata.ctime() * 1000 + (metadata.ctime_nsec() / 1_000_000), "birthtimeMs": metadata.ctime() * 1000 + (metadata.ctime_nsec() / 1_000_000), "ino": metadata.ino(), "nlink": metadata.nlink(), "uid": metadata.uid(), "gid": metadata.gid(), }) } fn python_readdir_value(entries: Vec) -> Value { json!(entries .into_iter() .filter(|entry| entry != "." && entry != "..") .collect::>()) } fn python_sync_rpc_recursive_flag(args: &[Value]) -> bool { args.get(1) .and_then(|value| { value .as_bool() .or_else(|| value.get("recursive").and_then(Value::as_bool)) }) .unwrap_or(false) } fn python_sync_rpc_bytes_arg( args: &[Value], index: usize, ) -> Result, PythonExecutionError> { let Some(value) = args.get(index) else { return Err(PythonExecutionError::RpcResponse(format!( "sync RPC argument {index} is required" ))); }; if let Some(text) = value.as_str() { return Ok(text.as_bytes().to_vec()); } let Some(base64_value) = value .get("__agentOSType") .and_then(Value::as_str) .filter(|kind| *kind == "bytes") .and_then(|_| value.get("base64")) .and_then(Value::as_str) else { return Err(PythonExecutionError::RpcResponse(format!( "sync RPC argument {index} must be a string or encoded bytes payload" ))); }; base64::engine::general_purpose::STANDARD .decode(base64_value) .map_err(|error| { PythonExecutionError::RpcResponse(format!( "sync RPC argument {index} contains invalid base64: {error}" )) }) } fn python_prewarm_sync_rpc_encoding(args: &[Value]) -> Option { args.get(1).and_then(|value| { value.as_str().map(str::to_owned).or_else(|| { value .get("encoding") .and_then(Value::as_str) .map(str::to_owned) }) }) } fn python_javascript_sync_rpc_error( request: &JavascriptSyncRpcRequest, ) -> Option<(&'static str, String)> { if matches!( request.method.as_str(), "net.connect" | "net.createConnection" | "dns.lookup" | "dns.resolve" | "dns.resolve4" | "dns.resolve6" | "dns.reverse" | "dgram.send" | "http.request" | "https.request" | "tls.connect" ) { return Some(( "ERR_ACCESS_DENIED", String::from( "network access is not available during standalone guest Python execution", ), )); } None } fn warmup_marker_contents( import_cache: &NodeImportCache, context: &PythonContext, request: &StartPythonExecutionRequest, ) -> String { let pyodide_dist_path = resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd); let compile_cache_dir = import_cache.shared_compile_cache_dir(); [ env!("CARGO_PKG_NAME").to_string(), env!("CARGO_PKG_VERSION").to_string(), PYTHON_WARMUP_MARKER_VERSION.to_string(), String::from("secure-exec-v8"), python_max_old_space_mb(request).to_string(), compile_cache_dir.display().to_string(), pyodide_dist_path.display().to_string(), file_fingerprint(&pyodide_dist_path.join("pyodide.mjs")), file_fingerprint(&pyodide_dist_path.join("pyodide-lock.json")), file_fingerprint(&pyodide_dist_path.join("pyodide.asm.js")), file_fingerprint(&pyodide_dist_path.join("pyodide.asm.wasm")), file_fingerprint(&pyodide_dist_path.join("python_stdlib.zip")), ] .join("\n") } fn python_warmup_metrics_enabled(request: &StartPythonExecutionRequest) -> bool { env_flag_enabled(&request.env, PYTHON_WARMUP_DEBUG_ENV) } fn warmup_metrics_line( debug_enabled: bool, executed: bool, reason: &str, duration_ms: f64, import_cache: &NodeImportCache, context: &PythonContext, request: &StartPythonExecutionRequest, ) -> Option> { if !debug_enabled { return None; } let compile_cache_dir = import_cache.shared_compile_cache_dir(); let pyodide_dist_path = resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd); Some( format!( "{PYTHON_WARMUP_METRICS_PREFIX}{{\"phase\":\"prewarm\",\"executed\":{},\"reason\":{},\"durationMs\":{duration_ms:.3},\"heapLimitMb\":{},\"compileCacheDir\":{},\"pyodideDistPath\":{}}}\n", if executed { "true" } else { "false" }, encode_json_string(reason), python_max_old_space_mb(request), encode_json_string(&compile_cache_dir.display().to_string()), encode_json_string(&pyodide_dist_path.display().to_string()), ) .into_bytes(), ) } #[cfg(test)] mod tests { use super::{ clear_pending_vfs_rpc, python_javascript_sync_rpc_action, python_managed_path_kind, python_runner_javascript_limits, python_wait_remaining, CreatePythonContextRequest, JavascriptSyncRpcRequest, PendingVfsRpc, PendingVfsRpcResolution, PendingVfsRpcState, PythonExecutionEngine, PythonExecutionLimits, PythonJavascriptSyncRpcAction, PythonManagedHostFiles, PythonManagedPathKind, PYODIDE_CACHE_GUEST_ROOT, PYODIDE_GUEST_ROOT, }; use std::collections::HashMap; use std::fs; #[cfg(unix)] use std::os::unix::fs::symlink; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use tempfile::tempdir; #[test] fn python_runner_forwards_vm_reactor_limits_to_javascript() { let limits = PythonExecutionLimits { reactor_work_quantum: Some(23), bridge_call_timeout_ms: Some(54_321), ..PythonExecutionLimits::default() }; let javascript = python_runner_javascript_limits(&limits, 256); assert_eq!(javascript.v8_heap_limit_mb, Some(256)); assert_eq!(javascript.reactor_work_quantum, Some(23)); assert_eq!(javascript.bridge_call_timeout_ms, Some(54_321)); } #[test] fn dispose_context_reclaims_python_and_nested_javascript_metadata() { let mut engine = PythonExecutionEngine::default(); let baseline = ( engine.context_count_for_test(), engine.javascript_context_count_for_test(), ); let temp = tempdir().expect("create Pyodide fixture root"); let context = engine.create_context(CreatePythonContextRequest { vm_id: String::from("vm-python-context-dispose"), pyodide_dist_path: temp.path().to_path_buf(), }); assert_eq!(engine.context_count_for_test(), baseline.0 + 1); assert_eq!(engine.javascript_context_count_for_test(), baseline.1 + 1); assert!(engine.dispose_context(&context.context_id)); assert_eq!( ( engine.context_count_for_test(), engine.javascript_context_count_for_test(), ), baseline ); } #[test] fn idle_wait_uses_readiness_instead_of_turn_polling() { let started = Instant::now(); assert_eq!(python_wait_remaining(None, started), None); let remaining = python_wait_remaining(Some(Duration::from_secs(1)), started) .expect("finite wait keeps one deadline"); assert!(remaining > Duration::from_millis(900)); assert!(remaining <= Duration::from_secs(1)); } #[test] fn stale_python_vfs_completion_has_no_pending_waiter() { let pending = Arc::new(Mutex::new(None)); assert_eq!( clear_pending_vfs_rpc(&pending, 41).expect("inspect pending request"), PendingVfsRpcResolution::Missing ); } #[test] fn timed_out_python_vfs_completion_is_consumed_as_stale() { let pending = Arc::new(Mutex::new(Some(PendingVfsRpc { state: PendingVfsRpcState::TimedOut(42), timeout_abort: None, }))); assert_eq!( clear_pending_vfs_rpc(&pending, 42).expect("clear timed-out request"), PendingVfsRpcResolution::TimedOut ); assert!(pending.lock().expect("pending request lock").is_none()); } #[test] fn python_managed_asset_descriptor_reads_use_raw_bounded_responses() { let temp = tempdir().expect("create temp dir"); let pyodide = temp.path().join("pyodide"); fs::create_dir_all(&pyodide).expect("create pyodide root"); fs::write(pyodide.join("python_stdlib.zip"), b"stdlib-bytes").expect("write managed asset"); let mut files = PythonManagedHostFiles::default(); let open = JavascriptSyncRpcRequest { id: 1, method: String::from("fs.openSync"), args: vec![ serde_json::Value::String(format!("{PYODIDE_GUEST_ROOT}/python_stdlib.zip")), serde_json::json!(0), serde_json::Value::Null, ], raw_bytes_args: HashMap::new(), }; let fd = match python_javascript_sync_rpc_action(&pyodide, &mut files, &open) .expect("route managed open") .expect("managed open action") { PythonJavascriptSyncRpcAction::Success(value) => { value.as_u64().expect("managed descriptor") } other => panic!("unexpected managed open action: {other:?}"), }; let read = JavascriptSyncRpcRequest { id: 2, method: String::from("fs.readSync"), args: vec![ serde_json::json!(fd), serde_json::json!(64), serde_json::Value::Null, ], raw_bytes_args: HashMap::from([(usize::MAX, Vec::new())]), }; match python_javascript_sync_rpc_action(&pyodide, &mut files, &read) .expect("route managed read") .expect("managed read action") { PythonJavascriptSyncRpcAction::RawSuccess(bytes) => { assert_eq!(bytes, b"stdlib-bytes") } other => panic!("unexpected managed read action: {other:?}"), } let close = JavascriptSyncRpcRequest { id: 3, method: String::from("fs.closeSync"), args: vec![serde_json::json!(fd)], raw_bytes_args: HashMap::new(), }; assert!(matches!( python_javascript_sync_rpc_action(&pyodide, &mut files, &close) .expect("route managed close"), Some(PythonJavascriptSyncRpcAction::Success( serde_json::Value::Null )) )); assert!(files.files.is_empty()); } #[test] fn python_managed_asset_descriptors_enforce_limit_and_reuse_capacity_after_close() { let temp = tempdir().expect("create temp dir"); let pyodide = temp.path().join("pyodide"); fs::create_dir_all(&pyodide).expect("create pyodide root"); fs::write(pyodide.join("python_stdlib.zip"), b"stdlib-bytes").expect("write managed asset"); let mut files = PythonManagedHostFiles::new(2); let open = |id| JavascriptSyncRpcRequest { id, method: String::from("fs.openSync"), args: vec![ serde_json::Value::String(format!("{PYODIDE_GUEST_ROOT}/python_stdlib.zip")), serde_json::json!(0), serde_json::Value::Null, ], raw_bytes_args: HashMap::new(), }; let opened_fd = |action: PythonJavascriptSyncRpcAction| match action { PythonJavascriptSyncRpcAction::Success(value) => { value.as_u64().expect("managed descriptor") } other => panic!("unexpected managed open action: {other:?}"), }; let first = opened_fd( python_javascript_sync_rpc_action(&pyodide, &mut files, &open(1)) .expect("route first open") .expect("first open action"), ); let _second = opened_fd( python_javascript_sync_rpc_action(&pyodide, &mut files, &open(2)) .expect("route second open") .expect("second open action"), ); assert_eq!(files.files.len(), 2); assert!(matches!( python_javascript_sync_rpc_action(&pyodide, &mut files, &open(3)) .expect("route saturated open"), Some(PythonJavascriptSyncRpcAction::Error { code: "EMFILE", message }) if message.contains("limits.resources.maxOpenFds") )); assert_eq!(files.files.len(), 2); let close = JavascriptSyncRpcRequest { id: 4, method: String::from("fs.closeSync"), args: vec![serde_json::json!(first)], raw_bytes_args: HashMap::new(), }; assert!(matches!( python_javascript_sync_rpc_action(&pyodide, &mut files, &close).expect("route close"), Some(PythonJavascriptSyncRpcAction::Success( serde_json::Value::Null )) )); let _replacement = opened_fd( python_javascript_sync_rpc_action(&pyodide, &mut files, &open(5)) .expect("route replacement open") .expect("replacement open action"), ); assert_eq!(files.files.len(), 2); } #[test] fn python_managed_guest_paths_normalize_dot_dot_inside_root() { let temp = tempdir().expect("create temp dir"); let pyodide = temp.path().join("pyodide"); fs::create_dir_all(pyodide.join("lib")).expect("create pyodide lib"); let resolved = python_managed_path_kind( &pyodide, &format!("{PYODIDE_GUEST_ROOT}/lib/../pyodide.mjs"), ); assert!(matches!(resolved.kind, PythonManagedPathKind::GuestPyodide)); assert_eq!( resolved.host_path().expect("host path"), pyodide.join("pyodide.mjs") ); } #[test] fn python_managed_guest_paths_clamp_dot_dot_escape_to_root() { let temp = tempdir().expect("create temp dir"); let pyodide = temp.path().join("pyodide"); fs::create_dir_all(&pyodide).expect("create pyodide root"); let resolved = python_managed_path_kind(&pyodide, &format!("{PYODIDE_GUEST_ROOT}/../../outside.txt")); assert!(matches!(resolved.kind, PythonManagedPathKind::GuestPyodide)); assert_eq!( resolved.host_path().expect("host path"), pyodide.join("outside.txt") ); } #[cfg(unix)] #[test] fn python_managed_guest_paths_reject_symlink_escape() { let temp = tempdir().expect("create temp dir"); let pyodide = temp.path().join("pyodide"); let outside = temp.path().join("outside"); fs::create_dir_all(&pyodide).expect("create pyodide root"); fs::create_dir_all(&outside).expect("create outside dir"); symlink(&outside, pyodide.join("escape")).expect("create escape symlink"); let resolved = python_managed_path_kind(&pyodide, &format!("{PYODIDE_GUEST_ROOT}/escape/file.txt")); assert!(matches!(resolved.kind, PythonManagedPathKind::Unmanaged)); assert!(resolved.host_path().is_none()); } #[cfg(unix)] #[test] fn python_managed_guest_paths_reject_symlink_escape_to_missing_descendant() { let temp = tempdir().expect("create temp dir"); let pyodide = temp.path().join("pyodide"); let outside = temp.path().join("outside"); fs::create_dir_all(&pyodide).expect("create pyodide root"); fs::create_dir_all(&outside).expect("create outside dir"); symlink(&outside, pyodide.join("escape")).expect("create escape symlink"); let resolved = python_managed_path_kind( &pyodide, &format!("{PYODIDE_GUEST_ROOT}/escape/missing/file.txt"), ); assert!(matches!(resolved.kind, PythonManagedPathKind::Unmanaged)); assert!(resolved.host_path().is_none()); } #[test] fn python_managed_host_paths_accept_canonical_root_descendants() { let temp = tempdir().expect("create temp dir"); let pyodide = temp.path().join("pyodide"); fs::create_dir_all(pyodide.join("pkg")).expect("create pyodide package dir"); let candidate = pyodide.join("pkg/module.py"); let resolved = python_managed_path_kind(&pyodide, &candidate.display().to_string()); assert!(matches!(resolved.kind, PythonManagedPathKind::HostManaged)); assert_eq!(resolved.host_path().expect("host path"), candidate); } #[test] fn python_managed_host_paths_reject_unresolved_dot_dot_escape() { let temp = tempdir().expect("create temp dir"); let pyodide = temp.path().join("pyodide"); fs::create_dir_all(&pyodide).expect("create pyodide root"); let candidate = pyodide.join("missing/../../outside.txt"); let resolved = python_managed_path_kind(&pyodide, &candidate.display().to_string()); assert!(matches!(resolved.kind, PythonManagedPathKind::Unmanaged)); assert!(resolved.host_path().is_none()); } #[test] fn python_managed_cache_guest_paths_resolve_inside_cache_root() { let temp = tempdir().expect("create temp dir"); let pyodide = temp.path().join("pyodide"); fs::create_dir_all(&pyodide).expect("create pyodide root"); let resolved = python_managed_path_kind( &pyodide, &format!("{PYODIDE_CACHE_GUEST_ROOT}/wheels/pkg.whl"), ); let host_path = resolved.host_path().expect("host path"); assert!(matches!(resolved.kind, PythonManagedPathKind::GuestCache)); assert!(host_path.ends_with("pyodide-package-cache/wheels/pkg.whl")); } }