use crate::common::stable_hash64;use crate::node_import_cache::{    NodeImportCache, NodeImportCacheCleanup, NODE_IMPORT_CACHE_ASSET_ROOT_ENV,};use crate::runtime_support::{    NODE_COMPILE_CACHE_ENV, NODE_DISABLE_COMPILE_CACHE_ENV, NODE_FROZEN_TIME_ENV,    NODE_SANDBOX_ROOT_ENV,};use crate::signal::NodeSignalHandlerRegistration;use crate::v8_host::{V8RuntimeHost, V8SessionFrameReceiver, V8SessionHandle};use crate::v8_ipc::BinaryFrame;use crate::v8_runtime;use agentos_bridge::queue_tracker::{register_queue, TrackedLimit};use agentos_runtime::RuntimeContext;use agentos_v8_runtime::runtime_protocol::{RuntimeCommand, WarmSessionHint};use flume::{Receiver as EventReceiver, Sender as EventSender};use getrandom::getrandom;use serde::Deserialize;use serde::Serialize;use serde_json::{json, Value};use std::cmp::Reverse;use std::collections::{BTreeMap, BinaryHeap, HashMap, HashSet, VecDeque};use std::fmt;use std::fs::{self, File};use std::io::{BufRead, BufReader, BufWriter, Write};use std::os::fd::OwnedFd;use std::panic::{catch_unwind, AssertUnwindSafe};use std::path::{Path, PathBuf};use std::sync::{    atomic::{AtomicBool, AtomicU64, Ordering},    mpsc::{self, Receiver, SyncSender, TrySendError},    Arc, Condvar, Mutex, OnceLock,};use std::thread;use std::time::{Duration, Instant};use tokio::sync::Notify;use tokio::time; const NODE_ENTRYPOINT_ENV: &str = "AGENTOS_ENTRYPOINT";const NODE_BOOTSTRAP_ENV: &str = "AGENTOS_BOOTSTRAP_MODULE";const NODE_GUEST_ARGV_ENV: &str = "AGENTOS_GUEST_ARGV";const NODE_PREWARM_IMPORTS_ENV: &str = "AGENTOS_NODE_PREWARM_IMPORTS";const NODE_IMPORT_COMPILE_CACHE_NAMESPACE_VERSION: &str = "3";const NODE_IMPORT_CACHE_LOADER_PATH_ENV: &str = "AGENTOS_NODE_IMPORT_CACHE_LOADER_PATH";const NODE_IMPORT_CACHE_PATH_ENV: &str = "AGENTOS_NODE_IMPORT_CACHE_PATH";const NODE_KEEP_STDIN_OPEN_ENV: &str = "AGENTOS_KEEP_STDIN_OPEN";const NODE_GUEST_ENTRYPOINT_ENV: &str = "AGENTOS_GUEST_ENTRYPOINT";const NODE_GUEST_ENTRYPOINT_MODULE_MODE_ENV: &str = "AGENTOS_GUEST_ENTRYPOINT_MODULE_MODE";const NODE_GUEST_PATH_MAPPINGS_ENV: &str = "AGENTOS_GUEST_PATH_MAPPINGS";const NODE_VIRTUAL_PROCESS_EXEC_PATH_ENV: &str = "AGENTOS_VIRTUAL_PROCESS_EXEC_PATH";const NODE_VIRTUAL_PROCESS_PID_ENV: &str = "AGENTOS_VIRTUAL_PROCESS_PID";const NODE_VIRTUAL_PROCESS_PPID_ENV: &str = "AGENTOS_VIRTUAL_PROCESS_PPID";const NODE_VIRTUAL_PROCESS_UID_ENV: &str = "AGENTOS_VIRTUAL_PROCESS_UID";const NODE_VIRTUAL_PROCESS_GID_ENV: &str = "AGENTOS_VIRTUAL_PROCESS_GID";const NODE_PARENT_ALLOW_CHILD_PROCESS_ENV: &str = "AGENTOS_PARENT_NODE_ALLOW_CHILD_PROCESS";const NODE_PARENT_ALLOW_WORKER_ENV: &str = "AGENTOS_PARENT_NODE_ALLOW_WORKER";const NODE_EXTRA_FS_READ_PATHS_ENV: &str = "AGENTOS_EXTRA_FS_READ_PATHS";const NODE_EXTRA_FS_WRITE_PATHS_ENV: &str = "AGENTOS_EXTRA_FS_WRITE_PATHS";const NODE_ALLOWED_BUILTINS_ENV: &str = "AGENTOS_ALLOWED_NODE_BUILTINS";const NODE_LOOPBACK_EXEMPT_PORTS_ENV: &str = "AGENTOS_LOOPBACK_EXEMPT_PORTS";const NODE_SYNC_RPC_ENABLE_ENV: &str = "AGENTOS_NODE_SYNC_RPC_ENABLE";const NODE_SYNC_RPC_REQUEST_FD_ENV: &str = "AGENTOS_NODE_SYNC_RPC_REQUEST_FD";const NODE_SYNC_RPC_RESPONSE_FD_ENV: &str = "AGENTOS_NODE_SYNC_RPC_RESPONSE_FD";const NODE_SYNC_RPC_DATA_BYTES_ENV: &str = "AGENTOS_NODE_SYNC_RPC_DATA_BYTES";const NODE_SYNC_RPC_WAIT_TIMEOUT_MS_ENV: &str = "AGENTOS_NODE_SYNC_RPC_WAIT_TIMEOUT_MS";static NEXT_V8_SESSION_ID: AtomicU64 = AtomicU64::new(1);static JAVASCRIPT_TIMER_WHEEL: OnceLock<Arc<TimerWheel>> = OnceLock::new();static JAVASCRIPT_TIMER_WHEEL_INIT: Mutex<()> = Mutex::new(()); #[derive(Default)]struct JsStartPhaseStats {    calls: u64,    total_ns: u128,    max_ns: u128,} static JS_START_PHASES: OnceLock<Mutex<BTreeMap<String, JsStartPhaseStats>>> = OnceLock::new();static JS_EVENT_PHASES: OnceLock<Mutex<BTreeMap<String, JsStartPhaseStats>>> = OnceLock::new(); fn js_start_phases_enabled() -> bool {    std::env::var("AGENTOS_JS_START_PHASES").as_deref() == Ok("1")} fn js_event_phases_enabled() -> bool {    std::env::var("AGENTOS_JS_EVENT_PHASES").as_deref() == Ok("1")} fn record_js_start_phase(stage: &str, elapsed: Duration) {    if !js_start_phases_enabled() {        return;    }    record_js_phase_stats(        &JS_START_PHASES,        "AGENTOS_JS_START_PHASES_FILE",        stage,        elapsed,    );} fn record_js_event_phase(stage: &str, elapsed: Duration) {    if !js_event_phases_enabled() {        return;    }    record_js_phase_stats(        &JS_EVENT_PHASES,        "AGENTOS_JS_EVENT_PHASES_FILE",        stage,        elapsed,    );} fn record_js_phase_stats(    phases: &OnceLock<Mutex<BTreeMap<String, JsStartPhaseStats>>>,    path_env: &str,    stage: &str,    elapsed: Duration,) {    let phases = phases.get_or_init(|| Mutex::new(BTreeMap::new()));    let Ok(mut phases) = phases.lock() else {        return;    };    let stats = phases.entry(stage.to_string()).or_default();    stats.calls += 1;    let elapsed_ns = elapsed.as_nanos();    stats.total_ns += elapsed_ns;    stats.max_ns = stats.max_ns.max(elapsed_ns);     let Some(path) = std::env::var_os(path_env) else {        return;    };    let mut output = String::new();    for (stage, stats) in phases.iter() {        let total_us = stats.total_ns / 1_000;        let avg_us = if stats.calls == 0 {            0        } else {            total_us / u128::from(stats.calls)        };        let max_us = stats.max_ns / 1_000;        output.push_str(&format!(            "stage={stage} calls={} total_us={total_us} avg_us={avg_us} max_us={max_us}\n",            stats.calls        ));    }    let _ = fs::write(path, output);} const DEFAULT_V8_CPU_TIME_LIMIT_MS: u32 = 30_000;const DEFAULT_V8_WALL_CLOCK_LIMIT_MS: u32 = 0;const DEFAULT_NODE_IMPORT_CACHE_MATERIALIZE_TIMEOUT_MS: u64 = 30_000;const NODE_SYNC_RPC_DEFAULT_DATA_BYTES: usize = 4 * 1024 * 1024;const NODE_SYNC_RPC_DEFAULT_WAIT_TIMEOUT_MS: u64 = 30_000;const NODE_SYNC_RPC_RESPONSE_QUEUE_CAPACITY: usize = 1;const FORWARD_KERNEL_STDIN_RPC_ENV: &str = "AGENTOS_FORWARD_KERNEL_STDIN_RPC";// Defense-in-depth headroom: a transient burst of guest events (e.g. a chatty// tool/skill turn) should be absorbed by the buffer, so the producer only ever// hits backpressure under a genuinely stuck consumer rather than on every spike.const JAVASCRIPT_EVENT_CHANNEL_CAPACITY: usize = 512;const JAVASCRIPT_EVENT_PAYLOAD_LIMIT_BYTES: usize = 1024 * 1024;const JAVASCRIPT_CAPTURED_OUTPUT_LIMIT_BYTES: usize = 16 * 1024 * 1024;const KERNEL_STDIN_BUFFER_LIMIT_BYTES: usize = 16 * 1024 * 1024;const NODE_WARMUP_MARKER_VERSION: &str = "1";const NODE_WARMUP_SPECIFIERS: &[&str] = &[    "secure-exec:builtin/path",    "secure-exec:builtin/url",    "secure-exec:builtin/fs-promises",    "secure-exec:polyfill/path",]; #[derive(Debug, Default, Clone)]struct SyncBridgePhaseStats {    calls: u64,    total_us: u64,    max_us: u64,} static SYNC_BRIDGE_PHASES: OnceLock<Mutex<BTreeMap<String, SyncBridgePhaseStats>>> =    OnceLock::new();static SYNC_BRIDGE_REQUEST_ENQUEUED: OnceLock<Mutex<HashMap<u64, (String, Instant)>>> =    OnceLock::new(); fn sync_bridge_phases_enabled() -> bool {    std::env::var("AGENTOS_SYNC_BRIDGE_PHASES").as_deref() == Ok("1")} fn record_sync_bridge_phase(method: &str, stage: &str, elapsed: Duration) {    if !sync_bridge_phases_enabled() {        return;    }    let stats = SYNC_BRIDGE_PHASES.get_or_init(|| Mutex::new(BTreeMap::new()));    let Ok(mut stats) = stats.lock() else {        return;    };    let elapsed_us = elapsed.as_micros() as u64;    let key = format!("{method}:{stage}");    let entry = stats.entry(key).or_default();    entry.calls += 1;    entry.total_us = entry.total_us.wrapping_add(elapsed_us);    entry.max_us = entry.max_us.max(elapsed_us);     if let Ok(path) = std::env::var("AGENTOS_SYNC_BRIDGE_PHASES_FILE") {        let mut lines = String::new();        for (key, value) in stats.iter() {            let Some((method, stage)) = key.split_once(':') else {                continue;            };            let avg_us = value.total_us.checked_div(value.calls).unwrap_or(0);            lines.push_str(&format!(                "method={method} stage={stage} calls={} total_us={} avg_us={} max_us={}\n",                value.calls, value.total_us, avg_us, value.max_us            ));        }        let _ = fs::write(path, lines);    }} pub fn record_sync_bridge_request_enqueued(call_id: u64, method: &str) {    if !sync_bridge_phases_enabled() {        return;    }    let requests = SYNC_BRIDGE_REQUEST_ENQUEUED.get_or_init(|| Mutex::new(HashMap::new()));    let Ok(mut requests) = requests.lock() else {        return;    };    if requests.len() > 4096 {        requests.clear();    }    requests.insert(call_id, (method.to_owned(), Instant::now()));} pub fn record_sync_bridge_request_observed(call_id: u64, fallback_method: &str) {    if !sync_bridge_phases_enabled() {        return;    }    let Some(requests) = SYNC_BRIDGE_REQUEST_ENQUEUED.get() else {        return;    };    let Ok(mut requests) = requests.lock() else {        return;    };    let Some((method, started)) = requests.remove(&call_id) else {        return;    };    let method = if method.is_empty() {        fallback_method    } else {        method.as_str()    };    record_sync_bridge_phase(method, "request_service_observed", started.elapsed());}const CONTROLLED_STDERR_PREFIXES: &[&str] =    &[crate::node_import_cache::NODE_IMPORT_CACHE_METRICS_PREFIX];const RESERVED_NODE_ENV_KEYS: &[&str] = &[    NODE_BOOTSTRAP_ENV,    NODE_COMPILE_CACHE_ENV,    NODE_DISABLE_COMPILE_CACHE_ENV,    NODE_ENTRYPOINT_ENV,    NODE_EXTRA_FS_READ_PATHS_ENV,    NODE_EXTRA_FS_WRITE_PATHS_ENV,    NODE_SANDBOX_ROOT_ENV,    NODE_FROZEN_TIME_ENV,    NODE_GUEST_ENTRYPOINT_ENV,    NODE_GUEST_ENTRYPOINT_MODULE_MODE_ENV,    NODE_GUEST_ARGV_ENV,    NODE_GUEST_PATH_MAPPINGS_ENV,    NODE_VIRTUAL_PROCESS_EXEC_PATH_ENV,    NODE_VIRTUAL_PROCESS_PID_ENV,    NODE_VIRTUAL_PROCESS_PPID_ENV,    NODE_VIRTUAL_PROCESS_UID_ENV,    NODE_VIRTUAL_PROCESS_GID_ENV,    NODE_PARENT_ALLOW_CHILD_PROCESS_ENV,    NODE_PARENT_ALLOW_WORKER_ENV,    NODE_IMPORT_CACHE_ASSET_ROOT_ENV,    NODE_IMPORT_CACHE_LOADER_PATH_ENV,    NODE_IMPORT_CACHE_PATH_ENV,    NODE_KEEP_STDIN_OPEN_ENV,    NODE_ALLOWED_BUILTINS_ENV,    NODE_LOOPBACK_EXEMPT_PORTS_ENV,    NODE_SYNC_RPC_ENABLE_ENV,    NODE_SYNC_RPC_REQUEST_FD_ENV,    NODE_SYNC_RPC_RESPONSE_FD_ENV,    NODE_SYNC_RPC_DATA_BYTES_ENV,    NODE_SYNC_RPC_WAIT_TIMEOUT_MS_ENV,]; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]#[serde(tag = "type", rename_all = "snake_case")]enum NodeControlMessage {    NodeImportCacheMetrics {        metrics: serde_json::Value,    },    PythonExit {        #[serde(rename = "exitCode")]        exit_code: i32,    },    SignalState {        signal: u32,        registration: NodeSignalHandlerRegistration,    },} #[derive(Debug, Default)]struct LinePrefixFilter {    pending: Vec<u8>,} #[derive(Debug, Clone, PartialEq, Eq)]pub struct JavascriptSyncRpcRequest {    pub id: u64,    pub method: String,    pub args: Vec<Value>,    pub raw_bytes_args: HashMap<usize, Vec<u8>>,} #[derive(Debug, Deserialize)]struct JavascriptSyncRpcRequestWire {    id: u64,    method: String,    #[serde(default)]    args: Vec<Value>,} impl LinePrefixFilter {    fn filter_chunk(&mut self, chunk: &[u8], prefixes: &[&str]) -> Vec<u8> {        self.pending.extend_from_slice(chunk);        let mut filtered = Vec::new();         while let Some(newline_index) = self.pending.iter().position(|byte| *byte == b'\n') {            let line = self.pending.drain(..=newline_index).collect::<Vec<_>>();            if !has_control_prefix(&line, prefixes) {                filtered.extend_from_slice(&line);            }        }         filtered    }} fn has_control_prefix(line: &[u8], prefixes: &[&str]) -> bool {    let text = String::from_utf8_lossy(line);    let trimmed = text.trim_end_matches(['\r', '\n']);    prefixes.iter().any(|prefix| trimmed.starts_with(prefix))} #[cfg(test)]#[derive(Debug)]struct JavascriptSyncRpcResponseWriter {    sender: SyncSender<Vec<u8>>,    timeout: Duration,} #[cfg(test)]impl JavascriptSyncRpcResponseWriter {    fn new(writer: File, timeout: Duration) -> Self {        let (sender, receiver) = mpsc::sync_channel(NODE_SYNC_RPC_RESPONSE_QUEUE_CAPACITY);        spawn_javascript_sync_rpc_response_writer(writer, receiver);        Self { sender, timeout }    }     fn send(&self, payload: Vec<u8>) -> Result<(), JavascriptExecutionError> {        let started = Instant::now();        let mut payload = Some(payload);         loop {            match self                .sender                .try_send(payload.take().expect("payload should be present"))            {                Ok(()) => return Ok(()),                Err(TrySendError::Disconnected(_)) => {                    return Err(JavascriptExecutionError::RpcResponse(String::from(                        "JavaScript sync RPC response channel closed unexpectedly",                    )));                }                Err(TrySendError::Full(returned_payload)) => {                    if started.elapsed() >= self.timeout {                        return Err(JavascriptExecutionError::RpcResponse(format!(                            "timed out after {}ms while queueing JavaScript sync RPC response",                            self.timeout.as_millis()                        )));                    }                    payload = Some(returned_payload);                    thread::sleep(Duration::from_millis(5));                }            }        }    }} #[cfg(test)]impl Clone for JavascriptSyncRpcResponseWriter {    fn clone(&self) -> Self {        Self {            sender: self.sender.clone(),            timeout: self.timeout,        }    }} #[derive(Debug, Clone, Copy, PartialEq, Eq)]enum PendingSyncRpcState {    Pending(u64),    TimedOut(u64),} #[derive(Debug, Clone, Copy, PartialEq, Eq)]enum PendingSyncRpcResolution {    Pending,    TimedOut,    Missing,} #[derive(Debug, Clone, PartialEq, Eq)]pub struct CreateJavascriptContextRequest {    pub vm_id: String,    pub bootstrap_module: Option<String>,    pub compile_cache_root: Option<PathBuf>,} #[derive(Debug, Clone, PartialEq, Eq)]pub struct JavascriptContext {    pub context_id: String,    pub vm_id: String,    pub bootstrap_module: Option<String>,    pub compile_cache_dir: Option<PathBuf>,} /// Per-execution JavaScript runtime limits, carried as typed fields on the/// execution request rather than `AGENTOS_*` env vars. The sidecar populates/// these from the per-VM `VmLimits` (which originate 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 JavascriptExecutionLimits {    /// V8 heap cap in MB. `None`/`Some(0)` keeps the engine default heap.    pub v8_heap_limit_mb: Option<u32>,    /// Sync-RPC blocking-wait ceiling in ms. `None` keeps the engine default.    pub sync_rpc_wait_timeout_ms: Option<u64>,    /// Active JavaScript CPU-time budget in ms. `None` keeps the engine default;    /// `Some(0)` disables the CPU watchdog.    pub cpu_time_limit_ms: Option<u32>,    /// JavaScript wall-clock backstop in ms. `None` keeps the engine default;    /// `Some(0)` disables the wall-clock watchdog.    pub wall_clock_limit_ms: Option<u32>,    /// Timeout for materializing the per-VM Node import cache.    pub import_cache_materialize_timeout_ms: Option<u64>,    /// Maximum live JavaScript timers in this execution. `None` keeps the    /// engine default. The sidecar supplies the VM-scoped configured value.    pub max_timers: Option<usize>,    /// Maximum readiness identities delivered in one V8 turn. Sidecar VM    /// execution must supply `limits.reactor.workQuantum` here.    pub reactor_work_quantum: Option<usize>,    /// Per-call host bridge deadline. Sidecar VMs supply    /// `limits.reactor.operationDeadlineMs`; zero is invalid.    pub bridge_call_timeout_ms: Option<u64>,} /// Per-execution guest-runtime config carried as typed fields rather than/// `AGENTOS_*` env vars. The sidecar populates these from kernel state/// (`user_profile()`, `resource_limits()`) and `CreateVmConfig`; the runtime/// shim interpolates them into a `_processConfig` object the guest reads, so the/// guest's virtual identity no longer rides the ambient env channel. `None`/// keeps the guest-runtime default. See the env-vs-wire rule in/// `crates/sidecar/CLAUDE.md`.#[derive(Debug, Clone, Default, PartialEq, Eq)]pub struct GuestRuntimeConfig {    /// Virtual `process.pid`.    pub virtual_pid: Option<u64>,    /// Virtual `process.ppid`.    pub virtual_ppid: Option<u64>,    /// Virtual `process.uid` / `process.euid`.    pub virtual_uid: Option<u64>,    /// Virtual `process.gid` / `process.egid` / `process.groups`.    pub virtual_gid: Option<u64>,    /// Virtual `process.execPath`.    pub virtual_exec_path: Option<String>,    /// `os.cpus().length`.    pub os_cpu_count: Option<u64>,    /// `os.totalmem()` in bytes.    pub os_totalmem: Option<u64>,    /// `os.freemem()` in bytes.    pub os_freemem: Option<u64>,    /// `os.homedir()`.    pub os_homedir: Option<String>,    /// `os.hostname()`.    pub os_hostname: Option<String>,    /// `os.tmpdir()`.    pub os_tmpdir: Option<String>,    /// `os.type()`.    pub os_type: Option<String>,    /// `os.release()`.    pub os_release: Option<String>,    /// `os.version()`.    pub os_version: Option<String>,    /// `os.machine()`.    pub os_machine: Option<String>,    /// Default login shell.    pub os_shell: Option<String>,    /// `os.userInfo().username`.    pub os_user: Option<String>,

Lines 1–500 of 8917. Load more