diff --git a/crates/capsule-cli/Cargo.lock b/crates/capsule-cli/Cargo.lock index 6213825..5ca96fa 100644 --- a/crates/capsule-cli/Cargo.lock +++ b/crates/capsule-cli/Cargo.lock @@ -322,7 +322,7 @@ dependencies = [ [[package]] name = "capsule-core" -version = "0.8.8" +version = "0.8.9" dependencies = [ "anyhow", "blake3", @@ -351,7 +351,7 @@ dependencies = [ [[package]] name = "capsule-run" -version = "0.8.8" +version = "0.8.9" dependencies = [ "capsule-core", "clap", diff --git a/crates/capsule-cli/Cargo.toml b/crates/capsule-cli/Cargo.toml index 14d3a9e..e608503 100644 --- a/crates/capsule-cli/Cargo.toml +++ b/crates/capsule-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "capsule-run" -version = "0.8.8" +version = "0.8.9" edition = "2024" description = "Secure WASM runtime to isolate and manage AI agent tasks" license = "Apache-2.0" @@ -16,7 +16,7 @@ path = "src/main.rs" [dependencies] clap = { version = "4.5.53", features = ["derive"] } -capsule-core = { version= "0.8.8", path = "../capsule-core" } +capsule-core = { version= "0.8.9", path = "../capsule-core" } tokio = { version = "1.48.0", features = ["rt", "rt-multi-thread", "macros", "io-util", "sync"] } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/crates/capsule-cli/npm/package.json b/crates/capsule-cli/npm/package.json index 4a76980..b8b68ff 100644 --- a/crates/capsule-cli/npm/package.json +++ b/crates/capsule-cli/npm/package.json @@ -1,6 +1,6 @@ { "name": "@capsule-run/cli", - "version": "0.8.8", + "version": "0.8.9", "description": "Secure WASM runtime to isolate and manage AI agent tasks", "bin": { "capsule": "./bin/capsule.js" @@ -29,9 +29,9 @@ "node": ">=18" }, "optionalDependencies": { - "@capsule-run/cli-darwin-arm64": "0.8.8", - "@capsule-run/cli-darwin-x64": "0.8.8", - "@capsule-run/cli-linux-x64": "0.8.8", - "@capsule-run/cli-win32-x64": "0.8.8" + "@capsule-run/cli-darwin-arm64": "0.8.9", + "@capsule-run/cli-darwin-x64": "0.8.9", + "@capsule-run/cli-linux-x64": "0.8.9", + "@capsule-run/cli-win32-x64": "0.8.9" } } diff --git a/crates/capsule-cli/pyproject.toml b/crates/capsule-cli/pyproject.toml index 91b9c0e..aade098 100644 --- a/crates/capsule-cli/pyproject.toml +++ b/crates/capsule-cli/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "maturin" [project] name = "capsule-run" -version = "0.8.8" +version = "0.8.9" description = "Secure WASM runtime to isolate and manage AI agent tasks" readme = "docs/README-pypi.md" license = {text = "Apache-2.0"} diff --git a/crates/capsule-cli/src/commands/build.rs b/crates/capsule-cli/src/commands/build.rs index 3d88bde..48f32d9 100644 --- a/crates/capsule-cli/src/commands/build.rs +++ b/crates/capsule-cli/src/commands/build.rs @@ -11,7 +11,7 @@ use capsule_core::wasm::compiler::python::{PythonWasmCompiler, PythonWasmCompile use capsule_core::wasm::runtime::Runtime; use capsule_core::wasm::runtime::RuntimeConfig; use capsule_core::wasm::runtime::WasmRuntimeError; -use capsule_core::wasm::utilities::task_reporter::TaskReporter; +use capsule_core::wasm::utilities::task_reporter::{LogLevel, TaskReporter}; pub enum BuildError { IoError(String), @@ -103,7 +103,7 @@ pub fn compile_to_wasm(file_path: &Path, export: bool) -> Result, export: bool) -> Result { let manifest = Manifest::new()?; - let mut reporter = TaskReporter::new(true); + let mut reporter = TaskReporter::new(LogLevel::Verbose); let file_path: PathBuf = match file_path { Some(path) => path.to_path_buf(), @@ -118,7 +118,7 @@ pub async fn execute(file_path: Option<&Path>, export: bool) -> Result, json: bool, verbose: bool, -) -> Result { - execute_with_runtime(wasm_path, args, mounts, json, verbose, None).await -} - -pub async fn execute_with_runtime( - wasm_path: &Path, - args: Vec, - mounts: Vec, - json: bool, - verbose: bool, shared_runtime: Option>, ) -> Result { let ext = wasm_path.extension().and_then(|e| e.to_str()).unwrap_or(""); @@ -78,7 +68,15 @@ pub async fn execute_with_runtime( ))); } - let mut reporter = TaskReporter::new(!json); + let log_level = if json { + LogLevel::Silent + } else if verbose { + LogLevel::Verbose + } else { + LogLevel::Normal + }; + + let mut reporter = TaskReporter::new(log_level.clone()); let wasm_path_abs = wasm_path .canonicalize() @@ -102,11 +100,16 @@ pub async fn execute_with_runtime( Some(r) => r, None => { reporter.start_progress("Initializing runtime"); + let capsule_toml = Manifest::new().map(|m| m.capsule_toml).unwrap_or_default(); - let runtime_config = RuntimeConfig { cache_dir, verbose }; - let r = Runtime::new(runtime_config, capsule_toml)?; + let runtime_config = RuntimeConfig { + cache_dir, + log_level, + }; + let runtime = Runtime::new(runtime_config, capsule_toml)?; + reporter.finish_progress(Some("Runtime ready")); - r + runtime } }; diff --git a/crates/capsule-cli/src/commands/run.rs b/crates/capsule-cli/src/commands/run.rs index e8233e3..88e93fd 100644 --- a/crates/capsule-cli/src/commands/run.rs +++ b/crates/capsule-cli/src/commands/run.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::fmt; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -9,7 +10,8 @@ use capsule_core::wasm::commands::run::RunInstance; use capsule_core::wasm::execution_policy::{Compute, ExecutionPolicy}; use capsule_core::wasm::runtime::{Runtime, RuntimeConfig, WasmRuntimeError}; use capsule_core::wasm::utilities::task_config::TaskConfig; -use capsule_core::wasm::utilities::task_reporter::TaskReporter; +use capsule_core::wasm::utilities::task_reporter::{LogLevel, TaskReporter}; +use tokio::sync::Mutex; use crate::build::{BuildError, TaskRegistry, compile_to_wasm}; use crate::commands::shared::load_env_variables; @@ -63,35 +65,88 @@ fn extract_main_execution_policy( Some(task_config.to_execution_policy(capsule_toml)) } -pub async fn execute( - file_path: Option<&Path>, - args: Vec, - mounts: Vec, - json: bool, - verbose: bool, -) -> Result { - execute_with_runtime(file_path, args, mounts, json, verbose, None).await +async fn resolve_wasm( + file_path: &Path, + reporter: &mut TaskReporter, + wasm_cache: Option>>>, +) -> Result<(PathBuf, Option, PathBuf), RunError> { + if let Some(cache) = wasm_cache { + let cache_key = file_path.to_string_lossy().to_string(); + + if let Some(wasm_path) = cache.lock().await.get(&cache_key).cloned() { + let cache_dir = std::env::current_dir().unwrap_or_default().join(".capsule"); + return Ok((wasm_path, None, cache_dir)); + } + + reporter.start_progress("Preparing environment"); + + let compile_result = tokio::task::spawn_blocking({ + let file_path = file_path.to_path_buf(); + move || compile_to_wasm(&file_path, false) + }) + .await + .map_err(|e| RunError::IoError(e.to_string()))??; + + reporter.finish_progress(Some("Environment ready")); + + cache + .lock() + .await + .insert(cache_key, compile_result.wasm_path.clone()); + + Ok(( + compile_result.wasm_path, + compile_result.task_registry, + compile_result.cache_dir, + )) + } else { + reporter.start_progress("Preparing environment"); + + let compile_result = tokio::task::spawn_blocking({ + let file_path = file_path.to_path_buf(); + move || compile_to_wasm(&file_path, false) + }) + .await + .map_err(|e| RunError::IoError(e.to_string()))??; + + reporter.finish_progress(Some("Environment ready")); + + Ok(( + compile_result.wasm_path, + compile_result.task_registry, + compile_result.cache_dir, + )) + } } -pub async fn execute_with_runtime( +pub async fn execute( file_path: Option<&Path>, args: Vec, mounts: Vec, json: bool, verbose: bool, shared_runtime: Option>, + wasm_cache: Option>>>, ) -> Result { let manifest = Manifest::new()?; - let mut reporter = TaskReporter::new(!json); + + let log_level = if json { + LogLevel::Silent + } else if verbose { + LogLevel::Verbose + } else { + LogLevel::Normal + }; + + let mut reporter = TaskReporter::new(log_level.clone()); let file_path: PathBuf = match file_path { Some(path) => path.to_path_buf(), None => PathBuf::from(manifest.get_entrypoint()?), }; - reporter.start_progress("Preparing environment"); - let compile_result = compile_to_wasm(&file_path, false)?; - reporter.finish_progress(Some("Environment ready")); + let (wasm_path, task_registry, cache_dir) = + resolve_wasm(&file_path, &mut reporter, wasm_cache).await?; let project_root = std::env::current_dir().unwrap_or_else(|_| { file_path @@ -103,9 +158,8 @@ pub async fn execute_with_runtime( load_env_variables(&project_root).map_err(RunError::IoError)?; - let mut execution_policy = - extract_main_execution_policy(compile_result.task_registry, &manifest.capsule_toml) - .unwrap_or_else(|| ExecutionPolicy::default().compute(Some(Compute::Custom(u64::MAX)))); + let mut execution_policy = extract_main_execution_policy(task_registry, &manifest.capsule_toml) + .unwrap_or_else(|| ExecutionPolicy::default().compute(Some(Compute::Custom(u64::MAX)))); execution_policy.mounts.extend(mounts); @@ -113,16 +167,20 @@ pub async fn execute_with_runtime( Some(r) => r, None => { reporter.start_progress("Initializing runtime"); + let runtime_config = RuntimeConfig { - cache_dir: compile_result.cache_dir, - verbose, + cache_dir, + log_level, }; - Runtime::new(runtime_config, manifest.capsule_toml)? + let runtime = Runtime::new(runtime_config, manifest.capsule_toml)?; + + reporter.finish_progress(Some("Runtime ready")); + runtime } }; let create_instance_command = CreateInstance::new(execution_policy.clone(), args.clone()) - .wasm_path(compile_result.wasm_path) + .wasm_path(wasm_path) .project_root(project_root); let (store, instance, task_id) = runtime.execute(create_instance_command).await?; diff --git a/crates/capsule-cli/src/commands/worker.rs b/crates/capsule-cli/src/commands/worker.rs index cc03a22..49e9f1a 100644 --- a/crates/capsule-cli/src/commands/worker.rs +++ b/crates/capsule-cli/src/commands/worker.rs @@ -1,12 +1,14 @@ +use std::collections::HashMap; use std::fmt; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use capsule_core::config::manifest::Manifest; use capsule_core::wasm::runtime::{Runtime, RuntimeConfig}; +use capsule_core::wasm::utilities::task_reporter::LogLevel; use serde::{Deserialize, Serialize}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::sync::mpsc; +use tokio::sync::{Mutex, mpsc}; use crate::commands::exec::ExecError; use crate::commands::run::RunError; @@ -66,12 +68,14 @@ pub async fn execute() -> Result<(), WorkerError> { let capsule_toml = Manifest::new().map(|m| m.capsule_toml).unwrap_or_default(); let runtime_config = RuntimeConfig { cache_dir, - verbose: false, + log_level: LogLevel::Silent, }; let runtime = Runtime::new(runtime_config, capsule_toml) .map_err(|e| WorkerError::RuntimeError(e.to_string()))?; + let wasm_cache: Arc>> = Arc::new(Mutex::new(HashMap::new())); + let stdin = tokio::io::stdin(); let mut lines = BufReader::new(stdin).lines(); @@ -112,9 +116,17 @@ pub async fn execute() -> Result<(), WorkerError> { }; let runtime = Arc::clone(&runtime); + let wasm_cache = Arc::clone(&wasm_cache); tokio::spawn(async move { - let result = dispatch(request.file, request.args, request.mounts, runtime).await; + let result = dispatch( + request.file, + request.args, + request.mounts, + runtime, + wasm_cache, + ) + .await; let response = WorkerResponse { id: request.id, @@ -141,6 +153,7 @@ async fn dispatch( args: Vec, mounts: Vec, runtime: Arc, + wasm_cache: Arc>>, ) -> Result { use std::ffi::OsStr; @@ -151,23 +164,19 @@ async fn dispatch( .to_lowercase(); match ext.as_str() { - "wasm" | "cwasm" => super::exec::execute_with_runtime( - Path::new(&file), - args, - mounts, - true, - false, - Some(runtime), - ) - .await - .map_err(|e: ExecError| e.to_string()), - _ => super::run::execute_with_runtime( + "wasm" | "cwasm" => { + super::exec::execute(Path::new(&file), args, mounts, true, false, Some(runtime)) + .await + .map_err(|e: ExecError| e.to_string()) + } + _ => super::run::execute( Some(Path::new(&file)), args, mounts, true, false, Some(runtime), + Some(wasm_cache), ) .await .map_err(|e: RunError| e.to_string()), diff --git a/crates/capsule-cli/src/main.rs b/crates/capsule-cli/src/main.rs index 943905e..3360085 100644 --- a/crates/capsule-cli/src/main.rs +++ b/crates/capsule-cli/src/main.rs @@ -62,7 +62,7 @@ async fn main() -> Result<(), CliError> { None => args, }; let file_path = file.as_deref().map(Path::new); - let result = run::execute(file_path, args, mount, json, verbose).await?; + let result = run::execute(file_path, args, mount, json, verbose, None, None).await?; if json { println!("{}", result); @@ -93,7 +93,7 @@ async fn main() -> Result<(), CliError> { Some(ref path) => load_args_file(path).map_err(CliError::ExecError)?, None => args, }; - let result = exec::execute(Path::new(&file), args, mount, json, verbose).await?; + let result = exec::execute(Path::new(&file), args, mount, json, verbose, None).await?; if json { println!("{}", result); diff --git a/crates/capsule-core/Cargo.lock b/crates/capsule-core/Cargo.lock index 929ea83..69ff3cd 100644 --- a/crates/capsule-core/Cargo.lock +++ b/crates/capsule-core/Cargo.lock @@ -302,7 +302,7 @@ dependencies = [ [[package]] name = "capsule-core" -version = "0.8.8" +version = "0.8.9" dependencies = [ "anyhow", "blake3", diff --git a/crates/capsule-core/Cargo.toml b/crates/capsule-core/Cargo.toml index edb2d99..603a7c6 100644 --- a/crates/capsule-core/Cargo.toml +++ b/crates/capsule-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "capsule-core" -version = "0.8.8" +version = "0.8.9" edition = "2024" description = "Core library for Capsule - WASM runtime for AI agent isolation" license = "Apache-2.0" diff --git a/crates/capsule-core/src/wasm/commands/create.rs b/crates/capsule-core/src/wasm/commands/create.rs index ed735e0..dff1b6d 100644 --- a/crates/capsule-core/src/wasm/commands/create.rs +++ b/crates/capsule-core/src/wasm/commands/create.rs @@ -159,6 +159,7 @@ impl RuntimeCommand for CreateInstance { limits, runtime: Some(Arc::clone(&runtime)), policy: self.policy.clone(), + wasm_path: self.wasm_path.clone(), peak_memory_bytes: 0, host_requests: Vec::new(), }; @@ -169,7 +170,7 @@ impl RuntimeCommand for CreateInstance { store.limiter(|state| state); - let component = match runtime.get_component().await { + let component = match runtime.get_component(&self.wasm_path).await { Some(c) => c, None => { let file_name = self.wasm_path.file_name().unwrap_or_default(); @@ -178,10 +179,13 @@ impl RuntimeCommand for CreateInstance { .join("wasm") .join(std::path::Path::new(file_name).with_extension("cwasm")); - let c = load_or_compile_component(&runtime.engine, &self.wasm_path, &cwasm_path)?; + let component = + load_or_compile_component(&runtime.engine, &self.wasm_path, &cwasm_path)?; - runtime.set_component(c.clone()).await; - c + runtime + .set_component(self.wasm_path.clone(), component.clone()) + .await; + component } }; diff --git a/crates/capsule-core/src/wasm/runtime.rs b/crates/capsule-core/src/wasm/runtime.rs index e525aa7..aa5661c 100644 --- a/crates/capsule-core/src/wasm/runtime.rs +++ b/crates/capsule-core/src/wasm/runtime.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::fmt; use std::path::PathBuf; use std::sync::Arc; @@ -9,7 +10,7 @@ use wasmtime::{Config, Engine}; use crate::config::log::{Log, LogError}; use crate::config::manifest::CapsuleToml; use crate::wasm::utilities::cache::precompile_component; -use crate::wasm::utilities::task_reporter::TaskReporter; +use crate::wasm::utilities::task_reporter::{LogLevel, TaskReporter}; pub enum WasmRuntimeError { WasmtimeError(wasmtime::Error), @@ -61,14 +62,14 @@ pub trait RuntimeCommand { pub struct RuntimeConfig { pub cache_dir: PathBuf, - pub verbose: bool, + pub log_level: LogLevel, } impl Default for RuntimeConfig { fn default() -> Self { Self { cache_dir: PathBuf::from(".capsule"), - verbose: false, + log_level: LogLevel::Normal, } } } @@ -80,9 +81,9 @@ pub struct Runtime { #[allow(dead_code)] pub(crate) cache_dir: PathBuf, - pub verbose: bool, + pub log_level: LogLevel, - component: RwLock>, + component: RwLock>, pub task_reporter: Arc>, pub capsule_toml: CapsuleToml, } @@ -113,14 +114,14 @@ impl Runtime { engine_config.async_support(true); engine_config.consume_fuel(true); - let task_reporter = Arc::new(Mutex::new(TaskReporter::new(config.verbose))); + let task_reporter = Arc::new(Mutex::new(TaskReporter::new(config.log_level.clone()))); Ok(Arc::new(Self { engine: Engine::new(&engine_config)?, log, cache_dir: config.cache_dir, - verbose: config.verbose, - component: RwLock::new(None), + log_level: config.log_level, + component: RwLock::new(HashMap::new()), task_reporter, capsule_toml, })) @@ -133,12 +134,12 @@ impl Runtime { command.execute(Arc::clone(self)).await } - pub async fn get_component(&self) -> Option { - self.component.read().await.clone() + pub async fn get_component(&self, wasm_path: &PathBuf) -> Option { + self.component.read().await.get(wasm_path).cloned() } - pub async fn set_component(&self, component: Component) { - *self.component.write().await = Some(component); + pub async fn set_component(&self, wasm_path: PathBuf, component: Component) { + self.component.write().await.insert(wasm_path, component); } pub fn precompile( diff --git a/crates/capsule-core/src/wasm/state.rs b/crates/capsule-core/src/wasm/state.rs index 19914ee..815272e 100644 --- a/crates/capsule-core/src/wasm/state.rs +++ b/crates/capsule-core/src/wasm/state.rs @@ -1,3 +1,4 @@ +use std::path::PathBuf; use std::sync::Arc; use anyhow::Result; @@ -36,6 +37,7 @@ pub struct State { pub limits: StoreLimits, pub runtime: Option>, pub policy: ExecutionPolicy, + pub wasm_path: PathBuf, pub peak_memory_bytes: u64, pub host_requests: Vec, } @@ -112,7 +114,9 @@ impl Host for State { let mut last_error: Option = None; for attempt in 0..=max_retries { - let create_cmd = CreateInstance::new(policy.clone(), vec![]).task_name(&name); + let create_cmd = CreateInstance::new(policy.clone(), vec![]) + .task_name(&name) + .wasm_path(self.wasm_path.clone()); let (store, instance, task_id) = match runtime.execute(create_cmd).await { Ok(result) => result, diff --git a/crates/capsule-core/src/wasm/utilities/task_reporter.rs b/crates/capsule-core/src/wasm/utilities/task_reporter.rs index f40b1da..a8ee76e 100644 --- a/crates/capsule-core/src/wasm/utilities/task_reporter.rs +++ b/crates/capsule-core/src/wasm/utilities/task_reporter.rs @@ -6,17 +6,24 @@ use chrono::Utc; use colored::Colorize; use indicatif::{ProgressBar, ProgressStyle}; +#[derive(Debug, PartialEq, Eq, Clone)] +pub enum LogLevel { + Verbose, + Normal, + Silent, +} + pub struct TaskReporter { - verbose: bool, + log_level: LogLevel, start_time: Instant, spinner: Option, is_active: Arc, } impl TaskReporter { - pub fn new(verbose: bool) -> Self { + pub fn new(log_level: LogLevel) -> Self { Self { - verbose, + log_level, start_time: Instant::now(), spinner: None, is_active: Arc::new(AtomicBool::new(false)), @@ -37,9 +44,13 @@ impl TaskReporter { } pub fn task_running(&mut self, task_name: &str, _task_id: &str) { + if self.log_level == LogLevel::Silent { + return; + } + self.finish_spinner(); self.start_time = Instant::now(); - if self.verbose { + if self.log_level == LogLevel::Verbose { eprintln!( "{} {} [{}] task execution started", Self::timestamp(), @@ -50,8 +61,12 @@ impl TaskReporter { } pub fn task_completed(&mut self, task_name: &str) { + if self.log_level == LogLevel::Silent { + return; + } + self.finish_spinner(); - if self.verbose { + if self.log_level == LogLevel::Verbose { let elapsed = self.start_time.elapsed(); let time_str = self.format_duration(elapsed); eprintln!( @@ -65,8 +80,12 @@ impl TaskReporter { } pub fn task_completed_with_time(&mut self, task_name: &str, elapsed: Duration) { + if self.log_level == LogLevel::Silent { + return; + } + self.finish_spinner(); - if self.verbose { + if self.log_level == LogLevel::Verbose { let time_str = self.format_duration(elapsed); eprintln!( "{} {} [{}] Completed in {}", @@ -79,8 +98,12 @@ impl TaskReporter { } pub fn task_failed(&mut self, task_name: &str, error: &str) { + if self.log_level == LogLevel::Silent { + return; + } + self.finish_spinner(); - if self.verbose { + if self.log_level == LogLevel::Verbose { eprintln!( "{} {} [{}] {}", Self::timestamp(), @@ -94,8 +117,12 @@ impl TaskReporter { } pub fn task_timeout(&mut self, task_name: &str) { + if self.log_level == LogLevel::Silent { + return; + } + self.finish_spinner(); - if self.verbose { + if self.log_level == LogLevel::Verbose { eprintln!( "{} {} [{}] Timed out", Self::timestamp(), @@ -108,44 +135,42 @@ impl TaskReporter { } pub fn start_progress(&mut self, message: &str) { - if self.verbose { - self.start_time = Instant::now(); - self.spinner = Some(self.create_spinner(message)); - self.is_active.store(true, Ordering::SeqCst); + if self.log_level == LogLevel::Silent { + return; } - } - pub fn update_progress(&mut self, message: &str) { - if let Some(spinner) = &self.spinner { - spinner.set_message(message.to_string()); - } + self.start_time = Instant::now(); + self.spinner = Some(self.create_spinner(message)); + self.is_active.store(true, Ordering::SeqCst); } pub fn finish_progress(&mut self, completion_message: Option<&str>) { + if self.log_level == LogLevel::Silent { + return; + } + self.finish_spinner(); let elapsed = self.start_time.elapsed(); let time_str = self.format_duration(elapsed); - if let Some(msg) = completion_message - && self.verbose - { + if let Some(msg) = completion_message { eprintln!("{} {} ({})", "✓".green(), msg.green(), time_str); } } - pub fn info(&self, message: &str) { - if self.verbose { - eprintln!("{}", message); - } - } - pub fn success(&self, message: &str) { - if self.verbose { - eprintln!("{}", message.green()); + if self.log_level == LogLevel::Silent { + return; } + + eprintln!("{}", message.green()); } pub fn error(&self, message: &str) { + if self.log_level == LogLevel::Silent { + return; + } + eprintln!("{}", message.red()); } @@ -195,43 +220,43 @@ mod tests { #[test] fn test_new() { - let reporter = TaskReporter::new(true); - assert!(reporter.verbose); + let reporter = TaskReporter::new(LogLevel::Verbose); + assert_eq!(reporter.log_level, LogLevel::Verbose); assert!(reporter.spinner.is_none()); assert!(!reporter.is_active.load(Ordering::SeqCst)); } #[test] fn test_format_duration_seconds() { - let reporter = TaskReporter::new(false); + let reporter = TaskReporter::new(LogLevel::Normal); let duration = Duration::from_secs_f64(45.67); assert_eq!(reporter.format_duration(duration), "45.67s"); } #[test] fn test_format_duration_minutes() { - let reporter = TaskReporter::new(false); + let reporter = TaskReporter::new(LogLevel::Normal); let duration = Duration::from_secs(125); assert_eq!(reporter.format_duration(duration), "2m 5s"); } #[test] fn test_format_duration_hours() { - let reporter = TaskReporter::new(false); + let reporter = TaskReporter::new(LogLevel::Normal); let duration = Duration::from_secs(3665); assert_eq!(reporter.format_duration(duration), "1h 1m 5s"); } #[test] fn test_task_running_sets_active() { - let mut reporter = TaskReporter::new(true); + let mut reporter = TaskReporter::new(LogLevel::Normal); reporter.task_running("test_task", "task_123"); assert!(!reporter.is_active.load(Ordering::SeqCst)); } #[test] fn test_finish_spinner_clears_active() { - let mut reporter = TaskReporter::new(true); + let mut reporter = TaskReporter::new(LogLevel::Normal); reporter.start_progress("test"); reporter.finish_spinner(); assert!(!reporter.is_active.load(Ordering::SeqCst)); diff --git a/crates/capsule-sdk/javascript/package-lock.json b/crates/capsule-sdk/javascript/package-lock.json index c7d41f5..9b7c4c0 100644 --- a/crates/capsule-sdk/javascript/package-lock.json +++ b/crates/capsule-sdk/javascript/package-lock.json @@ -1,12 +1,12 @@ { "name": "@capsule-run/sdk", - "version": "0.8.8", + "version": "0.8.9", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@capsule-run/sdk", - "version": "0.8.8", + "version": "0.8.9", "license": "Apache-2.0", "dependencies": { "@bytecodealliance/jco": "^1.0.0", diff --git a/crates/capsule-sdk/javascript/package.json b/crates/capsule-sdk/javascript/package.json index c6c24fb..0e86f47 100644 --- a/crates/capsule-sdk/javascript/package.json +++ b/crates/capsule-sdk/javascript/package.json @@ -1,6 +1,6 @@ { "name": "@capsule-run/sdk", - "version": "0.8.8", + "version": "0.8.9", "description": "Capsule JavaScript SDK - run AI agent tasks in secure WASM sandboxes", "type": "module", "main": "./dist/index.js", diff --git a/crates/capsule-sdk/javascript/src/polyfills/fs.ts b/crates/capsule-sdk/javascript/src/polyfills/fs.ts index 263a5b2..a073bc2 100644 --- a/crates/capsule-sdk/javascript/src/polyfills/fs.ts +++ b/crates/capsule-sdk/javascript/src/polyfills/fs.ts @@ -78,7 +78,7 @@ function normalizePath(path: string): string { else if (p.startsWith('/')) p = p.slice(1); if (p.length > 1) p = p.replace(/\/+$/, ''); - return p; + return p || '.'; } /** diff --git a/crates/capsule-sdk/javascript/src/run.ts b/crates/capsule-sdk/javascript/src/run.ts index 06791a2..8e6e649 100644 --- a/crates/capsule-sdk/javascript/src/run.ts +++ b/crates/capsule-sdk/javascript/src/run.ts @@ -11,7 +11,6 @@ import { existsSync, writeFileSync, unlinkSync } from 'fs'; import { tmpdir } from 'os'; import { join } from 'path'; import { randomUUID } from 'crypto'; -import { createInterface } from 'readline'; import { HostRequest } from './task'; export interface RunnerOptions { @@ -36,9 +35,8 @@ export interface RunnerResult { }; } -const WASM_EXTENSIONS = new Set(['.wasm']); -const ARGS_FILE_THRESHOLD = 8 * 1024; // 8KB - +const WASM_EXTENSIONS = new Set(['.wasm', '.cwasm']); +const ARGS_FILE_THRESHOLD = 8 * 1024; interface PendingRequest { resolve: (result: RunnerResult) => void; @@ -47,6 +45,7 @@ interface PendingRequest { const workerRegistry = new Map(); const pending = new Map(); +const workerStdout = new Map(); function workerKey(capsulePath: string, cwd: string): string { return `${capsulePath}|${cwd}`; @@ -63,6 +62,7 @@ function getWorker(capsulePath: string, cwd: string): ChildProcess { if (existing) { existing.kill(); workerRegistry.delete(key); + workerStdout.delete(key); } const command = getCapsuleCommand(capsulePath); @@ -70,33 +70,48 @@ function getWorker(capsulePath: string, cwd: string): ChildProcess { let child: ChildProcess; if (process.platform === 'win32') { const comspec = process.env.comspec || 'cmd.exe'; - child = spawn(comspec, ['/d', '/s', '/c', command, 'worker'], { cwd, stdio: ['pipe', 'pipe', 'inherit'] }); + child = spawn(comspec, ['/d', '/s', '/k', command, 'worker'], { cwd, stdio: ['pipe', 'pipe', 'inherit'] }); } else { child = spawn(command, ['worker'], { cwd, stdio: ['pipe', 'pipe', 'inherit'] }); } - const rl = createInterface({ input: child.stdout! }); - rl.on('line', (line) => { - let response: { id: string; output?: unknown; error?: string }; - try { - response = JSON.parse(line); - } catch { - return; - } + workerStdout.set(key, child.stdout); + + let lineBuffer = ''; + child.stdout!.on('data', (chunk: Buffer) => { + lineBuffer += chunk.toString(); + let nl: number; + + while ((nl = lineBuffer.indexOf('\n')) !== -1) { + const line = lineBuffer.slice(0, nl); + lineBuffer = lineBuffer.slice(nl + 1); - const request = pending.get(response.id); - if (!request) return; - pending.delete(response.id); + let response: { id: string; output?: unknown; error?: string }; + + try { + response = JSON.parse(line); + } catch { + continue; + } - if (response.error) { - request.reject(new Error(response.error)); - } else { - request.resolve(response.output as RunnerResult); + const request = pending.get(response.id); + + if (!request) continue; + pending.delete(response.id); + + if (pending.size === 0) workerStdout.get(key)?.unref?.(); + + if (response.error) { + request.reject(new Error(response.error)); + } else { + request.resolve(response.output as RunnerResult); + } } }); child.on('exit', () => { workerRegistry.delete(key); + workerStdout.delete(key); for (const [id, req] of pending) { req.reject(new Error('Capsule worker process exited unexpectedly')); pending.delete(id); @@ -105,6 +120,7 @@ function getWorker(capsulePath: string, cwd: string): ChildProcess { child.on('error', (err) => { workerRegistry.delete(key); + workerStdout.delete(key); const message = (err as NodeJS.ErrnoException).code === 'ENOENT' ? `Capsule CLI not found. Use 'npm install -g @capsule-run/cli' to install it.` : err.message; @@ -117,6 +133,9 @@ function getWorker(capsulePath: string, cwd: string): ChildProcess { workerRegistry.set(key, child); process.once('exit', () => child.kill()); + child.unref(); + (child.stdout as any).unref(); + return child; } @@ -138,17 +157,20 @@ function writeArgsFile(args: string[]): string { function runViaWorker(options: RunnerOptions): Promise { const { file, args = [], mounts = [], cwd, capsulePath = 'capsule' } = options; const resolvedCwd = cwd || process.cwd(); + const key = workerKey(capsulePath, resolvedCwd); const id = randomUUID(); return new Promise((resolve, reject) => { const worker = getWorker(capsulePath, resolvedCwd); - pending.set(id, { resolve, reject }); + workerStdout.get(key)?.ref?.(); + const request = JSON.stringify({ id, file, args, mounts }) + '\n'; worker.stdin!.write(request, (err) => { if (err) { pending.delete(id); + if (pending.size === 0) workerStdout.get(key)?.unref?.(); reject(new Error(`Failed to send task to worker: ${err.message}`)); } }); diff --git a/crates/capsule-sdk/python/pyproject.toml b/crates/capsule-sdk/python/pyproject.toml index 65ec2cf..2fcafa8 100644 --- a/crates/capsule-sdk/python/pyproject.toml +++ b/crates/capsule-sdk/python/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "capsule" -version = "0.8.8" +version = "0.8.9" description = "Capsule Python SDK - run AI agent tasks in secure WASM sandboxes" readme = "README.md" requires-python = ">=3.10" diff --git a/crates/capsule-sdk/python/src/capsule/worker.py b/crates/capsule-sdk/python/src/capsule/worker.py index d86d876..fbdcdd2 100644 --- a/crates/capsule-sdk/python/src/capsule/worker.py +++ b/crates/capsule-sdk/python/src/capsule/worker.py @@ -39,28 +39,47 @@ async def _start(self) -> None: async def _reader_loop(self) -> None: assert self._process and self._process.stdout - while True: - try: - line = await self._process.stdout.readline() - except Exception: - break - - if not line: - break - try: - response = json.loads(line.decode("utf-8")) - req_id = response.get("id") - future = self._pending.pop(req_id, None) - if future and not future.done(): - future.set_result(line.decode("utf-8")) - except Exception: - continue - - for future in self._pending.values(): - if not future.done(): - future.set_exception(RuntimeError("capsule worker process exited unexpectedly")) - - self._pending.clear() + try: + while True: + try: + line = await self._process.stdout.readline() + except Exception: + break + + if not line: + break + try: + response = json.loads(line.decode("utf-8")) + req_id = response.get("id") + future = self._pending.pop(req_id, None) + if future and not future.done(): + future.set_result(line.decode("utf-8")) + except Exception: + continue + finally: + for future in self._pending.values(): + if not future.done(): + future.set_exception(RuntimeError("capsule worker process exited unexpectedly")) + self._pending.clear() + + proc = self._process + if proc is not None and proc.returncode is None: + try: + if proc.stdin: + proc.stdin.close() + except Exception: + pass + try: + proc.terminate() + except Exception: + pass + try: + await asyncio.shield(asyncio.wait_for(proc.wait(), timeout=3.0)) + except Exception: + try: + proc.kill() + except Exception: + pass async def _ensure_running(self) -> None: async with self._lock: @@ -165,12 +184,12 @@ async def run_with_worker( ) raise FileNotFoundError(f"File not found: {resolved_file}") - client = await _get_client(capsule_path, cwd) + client = await _get_client(capsule_path, cwd or os.getcwd()) try: raw = await client.send(resolved_file, args, mounts) except RuntimeError: - key = (capsule_path, cwd) + key = (capsule_path, cwd or os.getcwd()) _clients.pop(key, None) raise @@ -190,5 +209,20 @@ async def close_all() -> None: """Close all active worker processes. Call on application shutdown.""" for client in list(_clients.values()): await client.close() - _clients.clear() + + +import atexit + +def _close_all_sync() -> None: + """Close all workers synchronously (for atexit hook).""" + for client in _clients.values(): + proc = client._process + if proc and proc.returncode is None: + try: + proc.terminate() + except Exception: + pass + + +atexit.register(_close_all_sync) diff --git a/examples/javascript/dialogue-evaluator/main.ts b/examples/javascript/dialogue-evaluator/main.ts index f55b838..2f5e236 100644 --- a/examples/javascript/dialogue-evaluator/main.ts +++ b/examples/javascript/dialogue-evaluator/main.ts @@ -25,7 +25,8 @@ const insertDialogueLine = task({ const evaluateDialogueLine = task({ name: "Evaluate Dialogue Line", - envVariables: ["OPENAI_API_KEY", "OPENAI_BASE_URL", "OPENAI_MODEL"] + envVariables: ["OPENAI_API_KEY", "OPENAI_BASE_URL", "OPENAI_MODEL"], + allowedHosts: ["*"], }, async (prompt: string): Promise => { const openai = new OpenAI({ diff --git a/examples/javascript/dialogue-evaluator/tsconfig.json b/examples/javascript/dialogue-evaluator/tsconfig.json index 4f687a5..115f2d3 100644 --- a/examples/javascript/dialogue-evaluator/tsconfig.json +++ b/examples/javascript/dialogue-evaluator/tsconfig.json @@ -1,8 +1,8 @@ { "compilerOptions": { "target": "ES2020", - "module": "NodeNext", - "moduleResolution": "nodenext", + "module": "ESNext", + "moduleResolution": "Node", "esModuleInterop": true, "skipLibCheck": true, "strict": true,