diff --git a/src/app.rs b/src/app.rs index dc97e33b..22982e5a 100644 --- a/src/app.rs +++ b/src/app.rs @@ -200,9 +200,13 @@ impl AppBuilder { self.session.attach_store(db.clone(), "default").await; - if let Err(e) = db.cleanup_stale_sandbox_jobs().await { - tracing::warn!("Failed to cleanup stale sandbox jobs: {}", e); - } + // Fire-and-forget housekeeping — no need to block startup. + let db_cleanup = db.clone(); + tokio::spawn(async move { + if let Err(e) = db_cleanup.cleanup_stale_sandbox_jobs().await { + tracing::warn!("Failed to cleanup stale sandbox jobs: {}", e); + } + }); self.db = Some(db); Ok(()) @@ -715,15 +719,18 @@ impl AppBuilder { } if embeddings.is_some() { - match ws.backfill_embeddings().await { - Ok(count) if count > 0 => { - tracing::info!("Backfilled embeddings for {} chunks", count); + let ws_bg = Arc::clone(ws); + tokio::spawn(async move { + match ws_bg.backfill_embeddings().await { + Ok(count) if count > 0 => { + tracing::info!("Backfilled embeddings for {} chunks", count); + } + Ok(_) => {} + Err(e) => { + tracing::warn!("Failed to backfill embeddings: {}", e); + } } - Ok(_) => {} - Err(e) => { - tracing::warn!("Failed to backfill embeddings: {}", e); - } - } + }); } } diff --git a/src/channels/wasm/router.rs b/src/channels/wasm/router.rs index cbf8b7b8..f5ab72da 100644 --- a/src/channels/wasm/router.rs +++ b/src/channels/wasm/router.rs @@ -488,7 +488,7 @@ mod tests { let prepared = Arc::new(PreparedChannelModule { name: name.to_string(), description: format!("Test channel: {}", name), - component_bytes: Vec::new(), + component: None, limits: ResourceLimits::default(), }); diff --git a/src/channels/wasm/runtime.rs b/src/channels/wasm/runtime.rs index 62f9d19c..4e7effb8 100644 --- a/src/channels/wasm/runtime.rs +++ b/src/channels/wasm/runtime.rs @@ -68,38 +68,51 @@ impl WasmChannelRuntimeConfig { } /// A compiled WASM channel component ready for instantiation. -#[derive(Debug)] +/// +/// Stores the pre-compiled `Component` directly so instantiation +/// doesn't require recompilation. pub struct PreparedChannelModule { /// Channel name. pub name: String, /// Channel description. pub description: String, - /// Compiled component bytes (public for testing, otherwise use component_bytes()). - pub(crate) component_bytes: Vec, + /// Pre-compiled component (cheaply cloneable via internal Arc). + pub(crate) component: Option, /// Resource limits for this channel. pub limits: ResourceLimits, } impl PreparedChannelModule { - /// Get the compiled component bytes. - pub fn component_bytes(&self) -> &[u8] { - &self.component_bytes + /// Get the pre-compiled component for instantiation. + pub fn component(&self) -> Option<&wasmtime::component::Component> { + self.component.as_ref() } /// Create a PreparedChannelModule for testing purposes. /// - /// Creates a module with no actual WASM bytes, suitable for testing + /// Creates a module with no actual WASM component, suitable for testing /// channel infrastructure without requiring a real WASM component. pub fn for_testing(name: impl Into, description: impl Into) -> Self { Self { name: name.into(), description: description.into(), - component_bytes: Vec::new(), + component: None, limits: ResourceLimits::default(), } } } +impl std::fmt::Debug for PreparedChannelModule { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PreparedChannelModule") + .field("name", &self.name) + .field("description", &self.description) + .field("has_component", &self.component.is_some()) + .field("limits", &self.limits) + .finish() + } +} + /// WASM channel runtime. /// /// Manages the Wasmtime engine and a cache of prepared channel modules. @@ -137,6 +150,13 @@ impl WasmChannelRuntime { // Disable debug info in production wasmtime_config.debug_info(false); + // Enable persistent compilation cache. Wasmtime serializes compiled native + // code to disk (~/.cache/wasmtime by default), so subsequent startups + // deserialize instead of recompiling — typically 10-50x faster. + if let Err(e) = wasmtime_config.cache_config_load_default() { + tracing::warn!("Failed to enable wasmtime compilation cache: {}", e); + } + let engine = Engine::new(&wasmtime_config).map_err(|e| { WasmChannelError::Config(format!("Failed to create Wasmtime engine: {}", e)) })?; @@ -183,13 +203,13 @@ impl WasmChannelRuntime { // Compile in blocking task (Wasmtime compilation is synchronous) let prepared = tokio::task::spawn_blocking(move || { // Validate and compile the component - let _component = wasmtime::component::Component::new(&engine, &wasm_bytes) + let component = wasmtime::component::Component::new(&engine, &wasm_bytes) .map_err(|e| WasmChannelError::Compilation(e.to_string()))?; Ok::<_, WasmChannelError>(PreparedChannelModule { name: name.clone(), description: desc, - component_bytes: wasm_bytes, + component: Some(component), limits: limits.unwrap_or(default_limits), }) }) diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs index e28599a1..f546da85 100644 --- a/src/channels/wasm/wrapper.rs +++ b/src/channels/wasm/wrapper.rs @@ -37,7 +37,7 @@ use tokio::sync::{RwLock, mpsc, oneshot}; use tokio_stream::wrappers::ReceiverStream; use uuid::Uuid; use wasmtime::Store; -use wasmtime::component::{Component, Linker}; +use wasmtime::component::Linker; use wasmtime_wasi::{ResourceTable, WasiCtx, WasiCtxBuilder, WasiView}; use crate::channels::wasm::capabilities::ChannelCapabilities; @@ -725,9 +725,13 @@ impl WasmChannel { ) -> Result { let engine = runtime.engine(); - // Compile the component (uses cached bytes) - let component = Component::new(engine, prepared.component_bytes()) - .map_err(|e| WasmChannelError::Compilation(e.to_string()))?; + // Use the pre-compiled component (no recompilation needed) + let component = prepared + .component() + .ok_or_else(|| { + WasmChannelError::Compilation("No compiled component available".to_string()) + })? + .clone(); // Create linker and add host functions let mut linker = Linker::new(engine); @@ -778,7 +782,7 @@ impl WasmChannel { /// Returns the channel configuration for HTTP endpoint registration. async fn call_on_start(&self) -> Result { // If no WASM bytes, return default config (for testing) - if self.prepared.component_bytes.is_empty() { + if self.prepared.component().is_none() { tracing::info!( channel = %self.name, "WASM channel on_start called (no WASM module, returning defaults)" @@ -918,7 +922,7 @@ impl WasmChannel { ); // If no WASM bytes, return 200 OK (for testing) - if self.prepared.component_bytes.is_empty() { + if self.prepared.component().is_none() { tracing::debug!( channel = %self.name, method = method, @@ -1018,7 +1022,7 @@ impl WasmChannel { /// Called periodically if polling is configured. pub async fn call_on_poll(&self) -> Result<(), WasmChannelError> { // If no WASM bytes, do nothing (for testing) - if self.prepared.component_bytes.is_empty() { + if self.prepared.component().is_none() { tracing::debug!( channel = %self.name, "WASM channel on_poll called (no WASM module)" @@ -1118,7 +1122,7 @@ impl WasmChannel { ); // If no WASM bytes, do nothing (for testing) - if self.prepared.component_bytes.is_empty() { + if self.prepared.component().is_none() { tracing::debug!( channel = %self.name, message_id = %message_id, @@ -1236,7 +1240,7 @@ impl WasmChannel { metadata: &serde_json::Value, ) -> Result<(), WasmChannelError> { // If no WASM bytes, do nothing (for testing) - if self.prepared.component_bytes.is_empty() { + if self.prepared.component().is_none() { return Ok(()); } @@ -1307,7 +1311,7 @@ impl WasmChannel { timeout: Duration, wit_update: wit_channel::StatusUpdate, ) -> Result<(), WasmChannelError> { - if prepared.component_bytes.is_empty() { + if prepared.component().is_none() { return Ok(()); } @@ -1627,7 +1631,7 @@ impl WasmChannel { workspace_store: &Arc, ) -> Result, WasmChannelError> { // Skip if no WASM bytes (testing mode) - if prepared.component_bytes.is_empty() { + if prepared.component().is_none() { tracing::debug!( channel = %channel_name, "WASM channel on_poll called (no WASM module)" @@ -2206,7 +2210,7 @@ mod tests { let prepared = Arc::new(PreparedChannelModule { name: "test".to_string(), description: "Test channel".to_string(), - component_bytes: Vec::new(), + component: None, limits: ResourceLimits::default(), }); @@ -2271,7 +2275,7 @@ mod tests { #[tokio::test] async fn test_execute_poll_no_wasm_returns_empty() { - // When there's no WASM module (empty component_bytes), execute_poll + // When there's no WASM module (None component), execute_poll // should return an empty vector of messages let config = WasmChannelRuntimeConfig::for_testing(); let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap()); @@ -2279,7 +2283,7 @@ mod tests { let prepared = Arc::new(PreparedChannelModule { name: "poll-test".to_string(), description: "Test channel".to_string(), - component_bytes: Vec::new(), // No WASM bytes + component: None, // No WASM module limits: ResourceLimits::default(), }); @@ -2381,7 +2385,7 @@ mod tests { let prepared = Arc::new(PreparedChannelModule { name: "poll-channel".to_string(), description: "Polling test channel".to_string(), - component_bytes: Vec::new(), + component: None, limits: ResourceLimits::default(), }); diff --git a/src/llm/session.rs b/src/llm/session.rs index d677eacd..f6f9a54b 100644 --- a/src/llm/session.rs +++ b/src/llm/session.rs @@ -157,14 +157,14 @@ impl SessionManager { } // Token exists, validate it by calling /v1/users/me - println!("Validating session..."); + tracing::debug!("Validating session..."); match self.validate_token().await { Ok(()) => { - println!("Session valid."); + tracing::debug!("Session valid"); Ok(()) } Err(e) => { - println!("Session expired or invalid: {}", e); + tracing::info!("Session expired or invalid: {}", e); self.initiate_login().await } } diff --git a/src/main.rs b/src/main.rs index d57d3841..c2ed03bf 100644 --- a/src/main.rs +++ b/src/main.rs @@ -487,9 +487,13 @@ async fn main() -> anyhow::Result<()> { session.attach_store(Arc::clone(db), "default").await; // Mark any jobs left in "running" or "creating" state as "interrupted". - if let Err(e) = db.cleanup_stale_sandbox_jobs().await { - tracing::warn!("Failed to cleanup stale sandbox jobs: {}", e); - } + // Fire-and-forget housekeeping — no need to block startup. + let db_cleanup = Arc::clone(db); + tokio::spawn(async move { + if let Err(e) = db_cleanup.cleanup_stale_sandbox_jobs().await { + tracing::warn!("Failed to cleanup stale sandbox jobs: {}", e); + } + }); } // Create secrets store early: needed for injecting LLM API keys from encrypted @@ -800,14 +804,20 @@ async fn main() -> anyhow::Result<()> { ); } - // Register memory tools if database is available - if let Some(ref db) = db { - let mut workspace = Workspace::new_with_db("default", Arc::clone(db)); + // Create workspace once, reused for memory tools and agent + let workspace: Option> = if let Some(ref db) = db { + let mut ws = Workspace::new_with_db("default", Arc::clone(db)); if let Some(ref emb) = embeddings { - workspace = workspace.with_embeddings(emb.clone()); + ws = ws.with_embeddings(emb.clone()); } - let workspace = Arc::new(workspace); - tools.register_memory_tools(workspace); + Some(Arc::new(ws)) + } else { + None + }; + + // Register memory tools if workspace is available + if let Some(ref ws) = workspace { + tools.register_memory_tools(Arc::clone(ws)); } // Register builder tool if enabled. @@ -1322,17 +1332,6 @@ async fn main() -> anyhow::Result<()> { None }; - // Create workspace for agent (shared with memory tools) - let workspace = if let Some(ref db_ref) = db { - let mut ws = Workspace::new_with_db("default", Arc::clone(db_ref)); - if let Some(ref emb) = embeddings { - ws = ws.with_embeddings(emb.clone()); - } - Some(Arc::new(ws)) - } else { - None - }; - // Seed workspace with core identity files on first boot if let Some(ref ws) = workspace { match ws.seed_if_empty().await { @@ -1343,17 +1342,20 @@ async fn main() -> anyhow::Result<()> { } } - // Backfill embeddings if we just enabled the provider + // Backfill embeddings in background (fire-and-forget housekeeping) if let (Some(ws), Some(_)) = (&workspace, &embeddings) { - match ws.backfill_embeddings().await { - Ok(count) if count > 0 => { - tracing::info!("Backfilled embeddings for {} chunks", count); + let ws_bg = Arc::clone(ws); + tokio::spawn(async move { + match ws_bg.backfill_embeddings().await { + Ok(count) if count > 0 => { + tracing::info!("Backfilled embeddings for {} chunks", count); + } + Ok(_) => {} + Err(e) => { + tracing::warn!("Failed to backfill embeddings: {}", e); + } } - Ok(_) => {} - Err(e) => { - tracing::warn!("Failed to backfill embeddings: {}", e); - } - } + }); } // Create context manager (shared between job tools and agent) diff --git a/src/tools/wasm/runtime.rs b/src/tools/wasm/runtime.rs index 9f6e8133..cf34f3c8 100644 --- a/src/tools/wasm/runtime.rs +++ b/src/tools/wasm/runtime.rs @@ -64,8 +64,8 @@ impl WasmRuntimeConfig { /// A compiled WASM component ready for instantiation. /// /// Contains the pre-compiled component plus cached metadata extracted -/// from the component during preparation. -#[derive(Debug)] +/// from the component during preparation. Stores the compiled `Component` +/// directly so instantiation doesn't require recompilation. pub struct PreparedModule { /// Tool name. pub name: String, @@ -73,16 +73,26 @@ pub struct PreparedModule { pub description: String, /// Parameter schema JSON (cached from component). pub schema: serde_json::Value, - /// Compiled component bytes (can be serialized for caching). - component_bytes: Vec, + /// Pre-compiled component (cheaply cloneable via internal Arc). + component: wasmtime::component::Component, /// Resource limits for this tool. pub limits: ResourceLimits, } impl PreparedModule { - /// Get the compiled component bytes. - pub fn component_bytes(&self) -> &[u8] { - &self.component_bytes + /// Get the pre-compiled component for instantiation. + pub fn component(&self) -> &wasmtime::component::Component { + &self.component + } +} + +impl std::fmt::Debug for PreparedModule { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PreparedModule") + .field("name", &self.name) + .field("description", &self.description) + .field("limits", &self.limits) + .finish() } } @@ -123,6 +133,13 @@ impl WasmToolRuntime { // Disable debug info in production for smaller modules wasmtime_config.debug_info(false); + // Enable persistent compilation cache. Wasmtime serializes compiled native + // code to disk (~/.cache/wasmtime by default), so subsequent startups + // deserialize instead of recompiling — typically 10-50x faster. + if let Err(e) = wasmtime_config.cache_config_load_default() { + tracing::warn!("Failed to enable wasmtime compilation cache: {}", e); + } + let engine = Engine::new(&wasmtime_config).map_err(|e| { WasmError::EngineCreationFailed(format!("Failed to create Wasmtime engine: {}", e)) })?; @@ -199,7 +216,7 @@ impl WasmToolRuntime { name: name.clone(), description, schema, - component_bytes: wasm_bytes, + component, limits: limits.unwrap_or(default_limits), }) }) diff --git a/src/tools/wasm/wrapper.rs b/src/tools/wasm/wrapper.rs index 5424fb79..8ec7aac4 100644 --- a/src/tools/wasm/wrapper.rs +++ b/src/tools/wasm/wrapper.rs @@ -13,7 +13,7 @@ use std::time::{Duration, Instant}; use async_trait::async_trait; use wasmtime::Store; -use wasmtime::component::{Component, Linker}; +use wasmtime::component::Linker; use wasmtime_wasi::{ResourceTable, WasiCtx, WasiCtxBuilder, WasiView}; use crate::context::JobContext; @@ -581,9 +581,8 @@ impl WasmToolWrapper { // Set up resource limiter store.limiter(|data| &mut data.limiter); - // Compile the component (uses cached bytes) - let component = Component::new(engine, self.prepared.component_bytes()) - .map_err(|e| WasmError::CompilationFailed(e.to_string()))?; + // Use the pre-compiled component (no recompilation needed) + let component = self.prepared.component().clone(); // Create linker with all host functions properly namespaced let mut linker = Linker::new(engine);