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<Self> {        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<String>,    /// Symlink target (the path the link points at), for `Symlink`.    pub target: Option<String>,    /// `Setattr` metadata fields (each applied only when present).    pub mode: Option<u32>,    pub uid: Option<u32>,    pub gid: Option<u32>,    pub atime_ms: Option<u64>,    pub mtime_ms: Option<u64>,    pub content_base64: Option<String>,    pub recursive: bool,    pub url: Option<String>,    pub http_method: Option<String>,    pub headers: BTreeMap<String, String>,    pub body_base64: Option<String>,    pub hostname: Option<String>,    pub family: Option<u8>,    /// Port for socket connect/sendto.    pub port: Option<u16>,    /// Socket handle for send/recv/close/sendto/recvfrom.    pub socket_id: Option<u64>,    pub command: Option<String>,    pub args: Vec<String>,    /// Explicit child argv[0], kept separate from the executable path.    pub argv0: Option<String>,    pub cwd: Option<String>,    pub env: BTreeMap<String, String>,    pub shell: bool,    pub max_buffer: Option<usize>,    /// Maximum time one socket receive operation may wait for shared    /// readiness. `Some(0)` preserves nonblocking socket behavior.    pub timeout_ms: Option<u64>,} #[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<String>,    },    Http {        status: u16,        reason: String,        url: String,        headers: BTreeMap<String, Vec<String>>,        body_base64: String,    },    DnsLookup {        addresses: Vec<String>,    },    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<String>,    #[serde(default)]    target: Option<String>,    // JS numbers cross the bridge as f64; accept that and narrow below.    #[serde(default)]    mode: Option<f64>,    #[serde(default)]    uid: Option<f64>,    #[serde(default)]    gid: Option<f64>,    #[serde(default, rename = "atimeMs")]    atime_ms: Option<f64>,    #[serde(default, rename = "mtimeMs")]    mtime_ms: Option<f64>,    #[serde(default)]    content_base64: Option<String>,    #[serde(default)]    recursive: bool,    #[serde(default)]    url: Option<String>,    #[serde(default, rename = "httpMethod")]    http_method: Option<String>,    #[serde(default)]    headers: BTreeMap<String, String>,    #[serde(default, rename = "bodyBase64")]    body_base64: Option<String>,    #[serde(default)]    hostname: Option<String>,    #[serde(default)]    family: Option<u8>,    #[serde(default)]    port: Option<u16>,    #[serde(default, rename = "socketId")]    socket_id: Option<u64>,    #[serde(default)]    command: Option<String>,    #[serde(default)]    args: Vec<String>,    #[serde(default)]    argv0: Option<String>,    #[serde(default)]    cwd: Option<String>,    #[serde(default)]    env: BTreeMap<String, String>,    #[serde(default)]    shell: bool,    #[serde(default, rename = "maxBuffer")]    max_buffer: Option<usize>,    #[serde(default, rename = "timeoutMs")]    timeout_ms: Option<u64>,} #[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<usize>,    /// Execution wall-clock cap in ms. `None` keeps the engine default;    /// `Some(0)` disables the timeout.    pub execution_timeout_ms: Option<u64>,    /// Pyodide V8 old-space cap in MB (`0` keeps the V8 default). `None` keeps    /// the engine default.    pub max_old_space_mb: Option<usize>,    /// VFS sync-RPC wait ceiling in ms. `None` keeps the engine default.    pub vfs_rpc_timeout_ms: Option<u64>,    /// VM readiness work bound forwarded unchanged to the Pyodide V8 runner.    pub reactor_work_quantum: Option<usize>,    /// Per-call host bridge deadline forwarded unchanged to the Pyodide V8 runner.    pub bridge_call_timeout_ms: Option<u64>,    /// 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<usize>,} #[derive(Debug, Clone, PartialEq, Eq)]pub struct StartPythonExecutionRequest {    pub vm_id: String,    pub context_id: String,    pub code: String,    pub file_path: Option<PathBuf>,    pub env: BTreeMap<String, String>,    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<u8>),    Stderr(Vec<u8>),    JavascriptSyncRpcRequest(JavascriptSyncRpcRequest),    VfsRpcRequest(Box<PythonVfsRpcRequest>),    Exited(i32),} #[derive(Debug, Clone, PartialEq, Eq)]pub struct PythonExecutionResult {    pub execution_id: String,    pub exit_code: i32,    pub stdout: Vec<u8>,    pub stderr: Vec<u8>,} #[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<Mutex<Option<PendingVfsRpc>>>,    v8_session: crate::v8_host::V8SessionHandle,    output_buffer_max_bytes: usize,    execution_timeout: Option<Duration>,    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<Mutex<Option<PendingVfsRpc>>>,    v8_session: crate::v8_host::V8SessionHandle,} #[derive(Debug)]struct PendingVfsRpc {    state: PendingVfsRpcState,    timeout_abort: Option<tokio::task::AbortHandle>,} #[derive(Debug, Clone, Copy, PartialEq, Eq)]enum PendingVfsRpcState {    Pending(u64),    TimedOut(u64),} #[derive(Debug, Clone, Copy, PartialEq, Eq)]enum PendingVfsRpcResolution {    Pending,    TimedOut,    Missing,

Lines 1–500 of 3155. Load more