From 94d101924e35a69a6286652e14853bfa674127f6 Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Tue, 10 Mar 2026 04:39:51 +0000 Subject: [PATCH] refactor: encapsulate leaked abstractions into owning modules (#778) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor: encapsulate leaked abstractions from main.rs and app.rs into owning modules Move module-specific initialization logic out of main.rs (1222→665 lines, -46%) and app.rs (944→780 lines, -17%) into their respective owning modules as public factory functions. This enforces separation of concerns so that adding a new DB backend, MCP transport, or channel doesn't require editing main.rs/app.rs. Key changes: - Tracing init functions → src/tracing_fmt.rs - DB connection factory (connect_with_handles + DatabaseHandles) → src/db/mod.rs - Secrets store factory (create_secrets_store) → src/secrets/mod.rs - MCP transport dispatch factory (create_client_from_config) → src/tools/mcp/factory.rs - Orchestrator setup (setup_orchestrator + OrchestratorSetup) → src/orchestrator/mod.rs - WASM channel setup (setup_wasm_channels) → src/channels/wasm/setup.rs - Worker entry points (run_worker, run_claude_bridge) → src/worker/mod.rs - Shared CLI secrets init (init_secrets_store) → src/cli/mod.rs - Tunnel startup (start_managed_tunnel) → src/tunnel/mod.rs - Onboard check (check_onboard_needed) → src/setup/mod.rs - ExtensionManager unified MCP: uses create_client_from_config via McpProcessManager, enabling stdio/Unix transports for hot-activated MCP servers - Deduplicated ~130 lines of secrets store init across cli/mcp.rs and cli/tool.rs - CLAUDE.md updated with module-owned initialization guideline [skip-regression-check] Co-Authored-By: Claude Opus 4.6 * refactor: address review feedback — deduplicate db factory, extract channel helper - connect_from_config() now delegates to connect_with_handles() to eliminate duplicated backend-matching logic (Copilot review feedback) - Extract register_channel() helper from setup_wasm_channels() loop body to improve readability (Gemini review feedback) [skip-regression-check] Co-Authored-By: Claude Opus 4.6 * style: fix rustfmt line wrapping in setup_wasm_channels Co-Authored-By: Claude Opus 4.6 * test: add integration test for module-owned initialization factories Exercises the full factory chain end-to-end to verify nothing was lost when initialization logic was moved from main.rs/app.rs into owning modules: - connect_with_handles returns Database + populated backend handles - connect_from_config delegates correctly (produces working Database) - secrets::create_secrets_store builds working store from DatabaseHandles - db::create_secrets_store standalone factory round-trips secrets - Both secrets factories produce compatible stores (cross-read works) - ExtensionManager constructs with McpProcessManager and is functional - DatabaseHandles default is empty All tests run without external services using libsql in-memory/tempfile. Co-Authored-By: Claude Opus 4.6 * fix: wire cli/mcp.rs and cli/tool.rs to shared init_secrets_store() Both files had inline implementations identical to cli::init_secrets_store(). Replace with delegation to complete the claimed deduplication. [skip-regression-check] Co-Authored-By: Claude Opus 4.6 * style: fix rustfmt line wrapping in integration test Co-Authored-By: Claude Opus 4.6 * fix(review): remove unused Config import and deduplicate Error Handling section - Remove `#[allow(unused_imports)]` and unused `use crate::config::Config` from cli/tool.rs (no longer needed after delegating to shared `cli::init_secrets_store()`) - Remove duplicate Error Handling subsection from CLAUDE.md Key Patterns (all four bullets already exist in Code Style section and review-discipline.md) Addresses Copilot review comments. [skip-regression-check] Co-Authored-By: Claude Opus 4.6 * fix(review): address remaining Copilot review comments - secrets/mod.rs: clarify docstring that None is a normal no-db condition - app.rs: add comment explaining the empty_handles fallback path - orchestrator/mod.rs: combine duplicated sandbox condition into single block - setup/mod.rs: document env var reads and thread-safety caveat [skip-regression-check] Co-Authored-By: Claude Opus 4.6 --------- Co-authored-by: Claude Opus 4.6 Co-authored-by: Henry Park --- CLAUDE.md | 39 +- src/app.rs | 219 ++-------- src/channels/wasm/mod.rs | 2 + src/channels/wasm/setup.rs | 324 ++++++++++++++ src/channels/web/server.rs | 3 + src/cli/mcp.rs | 14 +- src/cli/mod.rs | 41 +- src/cli/tool.rs | 14 +- src/db/mod.rs | 35 +- src/extensions/manager.rs | 28 +- src/main.rs | 608 ++------------------------- src/orchestrator/mod.rs | 112 +++++ src/secrets/mod.rs | 34 ++ src/setup/mod.rs | 32 ++ src/tools/builtin/extension_tools.rs | 1 + src/tools/mcp/factory.rs | 98 +++++ src/tools/mcp/mod.rs | 2 + src/tracing_fmt.rs | 19 + src/tunnel/mod.rs | 62 +++ src/worker/mod.rs | 58 +++ tests/module_init_integration.rs | 237 +++++++++++ 21 files changed, 1167 insertions(+), 815 deletions(-) create mode 100644 src/channels/wasm/setup.rs create mode 100644 src/tools/mcp/factory.rs create mode 100644 tests/module_init_integration.rs diff --git a/CLAUDE.md b/CLAUDE.md index dbdf6289..1b454e21 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -64,6 +64,13 @@ src/ │ ├── repl.rs # Simple REPL (for testing) │ ├── web/ # Web gateway (browser UI) — see src/channels/web/CLAUDE.md │ └── wasm/ # WASM channel runtime +│ ├── mod.rs +│ ├── bundled.rs # Bundled channel discovery +│ ├── capabilities.rs # Channel-specific capabilities (HTTP endpoint, emit rate) +│ ├── error.rs # WASM channel error types +│ ├── runtime.rs # WASM channel execution runtime +│ ├── setup.rs # WasmChannelSetup, setup_wasm_channels(), inject_channel_credentials() +│ └── wrapper.rs # Channel trait wrapper for WASM modules │ ├── cli/ # CLI subcommands (clap) │ ├── mod.rs # Cli struct, Command enum (run/onboard/config/tool/registry/mcp/memory/pairing/service/doctor/status/completion) @@ -76,7 +83,13 @@ src/ │ ├── hooks/ # Lifecycle hooks (6 points: BeforeInbound, BeforeToolCall, BeforeOutbound, OnSessionStart, OnSessionEnd, TransformResponse) │ -├── tunnel/ # Tunnel abstraction (cloudflare, ngrok, tailscale, custom, none) +├── tunnel/ # Tunnel abstraction for public internet exposure +│ ├── mod.rs # Tunnel trait, TunnelProviderConfig, create_tunnel(), start_managed_tunnel() +│ ├── cloudflare.rs # CloudflareTunnel (cloudflared binary) +│ ├── ngrok.rs # NgrokTunnel +│ ├── tailscale.rs # TailscaleTunnel (serve/funnel modes) +│ ├── custom.rs # CustomTunnel (arbitrary command with {host}/{port}) +│ └── none.rs # NoneTunnel (local-only, no exposure) │ ├── observability/ # Pluggable event/metric recording (noop, log, multi) │ @@ -105,8 +118,26 @@ src/ │ ├── rate_limiter.rs # Shared sliding-window rate limiter │ ├── builtin/ # Built-in tools (echo, time, json, http, web_fetch, file, shell, memory, message, job, routine, extension_tools, skill_tools, secrets_tools) │ ├── builder/ # Dynamic tool building -│ ├── mcp/ # Model Context Protocol client -│ └── wasm/ # Full WASM sandbox (wasmtime) — runtime, host functions, fuel metering, allowlist, credential injection +│ │ ├── core.rs # BuildRequirement, SoftwareType, Language +│ │ ├── templates.rs # Project scaffolding +│ │ ├── testing.rs # Test harness integration +│ │ └── validation.rs # WASM validation +│ ├── mcp/ # Model Context Protocol +│ │ ├── client.rs # MCP client over HTTP +│ │ ├── factory.rs # create_client_from_config() — transport dispatch factory +│ │ ├── protocol.rs # JSON-RPC types +│ │ └── session.rs # MCP session management (Mcp-Session-Id header, per-server state) +│ └── wasm/ # Full WASM sandbox (wasmtime) +│ ├── runtime.rs # Module compilation and caching +│ ├── wrapper.rs # Tool trait wrapper for WASM modules +│ ├── host.rs # Host functions (logging, time, workspace) +│ ├── limits.rs # Fuel metering and memory limiting +│ ├── allowlist.rs # Network endpoint allowlisting +│ ├── credential_injector.rs # Safe credential injection +│ ├── loader.rs # WASM tool discovery from filesystem +│ ├── rate_limiter.rs # Per-tool rate limiting +│ ├── error.rs # WASM-specific error types +│ └── storage.rs # Linear memory persistence │ ├── db/ # Dual-backend persistence (PostgreSQL + libSQL) — see src/db/CLAUDE.md │ @@ -144,6 +175,8 @@ Dual-backend: PostgreSQL + libSQL/Turso. **All new persistence features must sup When modifying a module with a spec, read the spec first. Code follows spec; spec is the tiebreaker. +**Module-owned initialization:** Module-specific initialization logic (database connection, transport creation, channel setup) must live in the owning module as a public factory function — not in `main.rs` or `app.rs`. These entry-point files orchestrate calls to module factories. Feature-flag branching (`#[cfg(feature = ...)]`) must be confined to the module that owns the abstraction. + | Module | Spec | |--------|------| | `src/agent/` | `src/agent/CLAUDE.md` | diff --git a/src/app.rs b/src/app.rs index 9fcb19f3..12f0e9bf 100644 --- a/src/app.rs +++ b/src/app.rs @@ -77,10 +77,7 @@ pub struct AppBuilder { llm_override: Option>, // Backend-specific handles needed by secrets store - #[cfg(feature = "postgres")] - pg_pool: Option, - #[cfg(feature = "libsql")] - libsql_db: Option>, + handles: Option, } impl AppBuilder { @@ -105,10 +102,7 @@ impl AppBuilder { db: None, secrets_store: None, llm_override: None, - #[cfg(feature = "postgres")] - pg_pool: None, - #[cfg(feature = "libsql")] - libsql_db: None, + handles: None, } } @@ -137,71 +131,10 @@ impl AppBuilder { return Ok(()); } - let db: Arc = match self.config.database.backend { - #[cfg(feature = "libsql")] - crate::config::DatabaseBackend::LibSql => { - use crate::db::Database as _; - use crate::db::libsql::LibSqlBackend; - use secrecy::ExposeSecret as _; - - let default_path = crate::config::default_libsql_path(); - let db_path = self - .config - .database - .libsql_path - .as_deref() - .unwrap_or(&default_path); - - let backend = if let Some(ref url) = self.config.database.libsql_url { - let token = - self.config - .database - .libsql_auth_token - .as_ref() - .ok_or_else(|| { - anyhow::anyhow!( - "LIBSQL_AUTH_TOKEN is required when LIBSQL_URL is set" - ) - })?; - LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()).await? - } else { - LibSqlBackend::new_local(db_path).await? - }; - backend.run_migrations().await?; - tracing::info!("libSQL database connected and migrations applied"); - - #[cfg(feature = "libsql")] - { - self.libsql_db = Some(backend.shared_db()); - } - - Arc::new(backend) as Arc - } - #[cfg(feature = "postgres")] - _ => { - use crate::db::Database as _; - let pg = crate::db::postgres::PgBackend::new(&self.config.database) - .await - .map_err(|e| anyhow::anyhow!("{}", e))?; - pg.run_migrations() - .await - .map_err(|e| anyhow::anyhow!("{}", e))?; - tracing::info!("PostgreSQL database connected and migrations applied"); - - #[cfg(feature = "postgres")] - { - self.pg_pool = Some(pg.pool()); - } - - Arc::new(pg) as Arc - } - #[cfg(not(feature = "postgres"))] - _ => { - anyhow::bail!( - "No database backend available. Enable 'postgres' or 'libsql' feature." - ); - } - }; + let (db, handles) = crate::db::connect_with_handles(&self.config.database) + .await + .map_err(|e| anyhow::anyhow!("{}", e))?; + self.handles = Some(handles); // Post-init: migrate disk config, reload config from DB, attach session, cleanup if let Err(e) = crate::bootstrap::migrate_disk_to_db(db.as_ref(), "default").await { @@ -251,10 +184,7 @@ impl AppBuilder { crate::config::inject_os_credentials(); // Consume unused handles - #[cfg(feature = "libsql")] - { - self.libsql_db.take(); - } + self.handles.take(); // Re-resolve only the LLM config with OS credentials. let store: Option<&(dyn crate::db::SettingsStore + Sync)> = @@ -278,35 +208,16 @@ impl AppBuilder { Ok(c) => Arc::new(c), Err(e) => { tracing::warn!("Failed to initialize secrets crypto: {}", e); - #[cfg(feature = "libsql")] - { - self.libsql_db.take(); - } + self.handles.take(); return Ok(()); } }; - let store: Option> = None; - - #[cfg(feature = "libsql")] - let store = store.or_else(|| { - self.libsql_db.take().map(|db| { - Arc::new(crate::secrets::LibSqlSecretsStore::new( - db, - Arc::clone(&crypto), - )) as Arc - }) - }); - - #[cfg(feature = "postgres")] - let store = store.or_else(|| { - self.pg_pool.as_ref().map(|pool| { - Arc::new(crate::secrets::PostgresSecretsStore::new( - pool.clone(), - Arc::clone(&crypto), - )) as Arc - }) - }); + // Fallback covers the no-database path where `init_database` returned + // early before populating `self.handles`. + let empty_handles = crate::db::DatabaseHandles::default(); + let handles = self.handles.as_ref().unwrap_or(&empty_handles); + let store = crate::secrets::create_secrets_store(crypto, handles); if let Some(ref secrets) = store { // Inject LLM API keys from encrypted storage @@ -472,9 +383,7 @@ impl AppBuilder { ), anyhow::Error, > { - use crate::tools::mcp::{ - McpClient, McpTransport, config::load_mcp_servers_from_db, is_authenticated, - }; + use crate::tools::mcp::config::load_mcp_servers_from_db; use crate::tools::wasm::{WasmToolLoader, load_dev_tools}; let mcp_session_manager = Arc::new(McpSessionManager::new()); @@ -578,95 +487,24 @@ impl AppBuilder { join_set.spawn(async move { let server_name = server.name.clone(); - let client: McpClient = match server.effective_transport() { - crate::tools::mcp::config::EffectiveTransport::Stdio { - command, - args, - env, - } => { - match pm - .spawn_stdio( - &server_name, - command, - args.to_vec(), - env.clone(), - ) - .await - { - Ok(transport) => McpClient::new_with_transport( - &server_name, - transport as Arc, - None, - secrets, - "default", - Some(server), - ), - Err(e) => { - tracing::warn!( - "Failed to spawn stdio MCP server '{}': {}", - server_name, - e - ); - return; - } - } - } - #[cfg(unix)] - crate::tools::mcp::config::EffectiveTransport::Unix { - socket_path, - } => { - match crate::tools::mcp::unix_transport::UnixMcpTransport::connect( - &server_name, - socket_path, - ) - .await - { - Ok(transport) => McpClient::new_with_transport( - &server_name, - Arc::new(transport) as Arc, - None, - secrets, - "default", - Some(server), - ), - Err(e) => { - tracing::warn!( - "Failed to connect to Unix MCP server '{}': {}", - server_name, - e - ); - return; - } - } - } - #[cfg(not(unix))] - crate::tools::mcp::config::EffectiveTransport::Unix { .. } => { + let client = match crate::tools::mcp::create_client_from_config( + server, + &mcp_sm, + &pm, + secrets, + "default", + ) + .await + { + Ok(c) => c, + Err(e) => { tracing::warn!( - "Unix socket transport is not supported on this platform (server '{}')", - server_name + "Failed to create MCP client for '{}': {}", + server_name, + e ); return; } - crate::tools::mcp::config::EffectiveTransport::Http => { - if let Some(ref secrets) = secrets { - let has_tokens = - is_authenticated(&server, secrets, "default") - .await; - - if has_tokens || server.requires_auth() { - McpClient::new_authenticated( - server, - Arc::clone(&mcp_sm), - Arc::clone(secrets), - "default", - ) - } else { - McpClient::new_with_config(server) - } - } else { - McpClient::new_with_config(server) - } - } }; match client.list_tools().await { @@ -767,6 +605,7 @@ impl AppBuilder { let extension_manager = { let manager = Arc::new(ExtensionManager::new( Arc::clone(&mcp_session_manager), + Arc::clone(&mcp_process_manager), ext_secrets, Arc::clone(tools), Some(Arc::clone(hooks)), diff --git a/src/channels/wasm/mod.rs b/src/channels/wasm/mod.rs index 29c7632b..0d4a6c3f 100644 --- a/src/channels/wasm/mod.rs +++ b/src/channels/wasm/mod.rs @@ -86,6 +86,7 @@ mod loader; mod router; mod runtime; mod schema; +pub mod setup; pub(crate) mod signature; #[allow(dead_code)] pub(crate) mod storage; @@ -105,4 +106,5 @@ pub use runtime::{PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeC pub use schema::{ ChannelCapabilitiesFile, ChannelConfig, SecretSetupSchema, SetupSchema, WebhookSchema, }; +pub use setup::{WasmChannelSetup, inject_channel_credentials, setup_wasm_channels}; pub use wrapper::{HttpResponse, SharedWasmChannel, WasmChannel}; diff --git a/src/channels/wasm/setup.rs b/src/channels/wasm/setup.rs new file mode 100644 index 00000000..ca202b3b --- /dev/null +++ b/src/channels/wasm/setup.rs @@ -0,0 +1,324 @@ +//! WASM channel setup and credential injection. +//! +//! Encapsulates the logic for loading WASM channels, registering their +//! webhook routes, and injecting credentials from the secrets store. + +use std::collections::HashSet; +use std::sync::Arc; + +use crate::channels::wasm::{ + LoadedChannel, RegisteredEndpoint, SharedWasmChannel, WasmChannel, WasmChannelLoader, + WasmChannelRouter, WasmChannelRuntime, WasmChannelRuntimeConfig, create_wasm_channel_router, +}; +use crate::config::Config; +use crate::db::Database; +use crate::extensions::ExtensionManager; +use crate::pairing::PairingStore; +use crate::secrets::SecretsStore; + +/// Result of WASM channel setup. +pub struct WasmChannelSetup { + pub channels: Vec<(String, Box)>, + pub channel_names: Vec, + pub webhook_routes: Option, + /// Runtime objects needed for hot-activation via ExtensionManager. + pub wasm_channel_runtime: Arc, + pub pairing_store: Arc, + pub wasm_channel_router: Arc, +} + +/// Load WASM channels and register their webhook routes. +pub async fn setup_wasm_channels( + config: &Config, + secrets_store: &Option>, + extension_manager: Option<&Arc>, + database: Option<&Arc>, +) -> Option { + let runtime = match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) { + Ok(r) => Arc::new(r), + Err(e) => { + tracing::warn!("Failed to initialize WASM channel runtime: {}", e); + return None; + } + }; + + let pairing_store = Arc::new(PairingStore::new()); + let settings_store: Option> = + database.map(|db| Arc::clone(db) as Arc); + let mut loader = WasmChannelLoader::new( + Arc::clone(&runtime), + Arc::clone(&pairing_store), + settings_store, + ); + if let Some(secrets) = secrets_store { + loader = loader.with_secrets_store(Arc::clone(secrets)); + } + + let results = match loader + .load_from_dir(&config.channels.wasm_channels_dir) + .await + { + Ok(r) => r, + Err(e) => { + tracing::warn!("Failed to scan WASM channels directory: {}", e); + return None; + } + }; + + let wasm_router = Arc::new(WasmChannelRouter::new()); + let mut channels: Vec<(String, Box)> = Vec::new(); + let mut channel_names: Vec = Vec::new(); + + for loaded in results.loaded { + let (name, channel) = register_channel(loaded, config, secrets_store, &wasm_router).await; + channel_names.push(name.clone()); + channels.push((name, channel)); + } + + for (path, err) in &results.errors { + tracing::warn!("Failed to load WASM channel {}: {}", path.display(), err); + } + + // Always create webhook routes (even with no channels loaded) so that + // channels hot-added at runtime can receive webhooks without a restart. + let webhook_routes = { + Some(create_wasm_channel_router( + Arc::clone(&wasm_router), + extension_manager.map(Arc::clone), + )) + }; + + Some(WasmChannelSetup { + channels, + channel_names, + webhook_routes, + wasm_channel_runtime: runtime, + pairing_store, + wasm_channel_router: wasm_router, + }) +} + +/// Process a single loaded WASM channel: retrieve secrets, inject config, +/// register with the router, and set up signing keys and credentials. +async fn register_channel( + loaded: LoadedChannel, + config: &Config, + secrets_store: &Option>, + wasm_router: &Arc, +) -> (String, Box) { + let channel_name = loaded.name().to_string(); + tracing::info!("Loaded WASM channel: {}", channel_name); + + let secret_name = loaded.webhook_secret_name(); + let sig_key_secret_name = loaded.signature_key_secret_name(); + let hmac_secret_name = loaded.hmac_secret_name(); + + let webhook_secret = if let Some(secrets) = secrets_store { + secrets + .get_decrypted("default", &secret_name) + .await + .ok() + .map(|s| s.expose().to_string()) + } else { + None + }; + + let secret_header = loaded.webhook_secret_header().map(|s| s.to_string()); + + let webhook_path = format!("/webhook/{}", channel_name); + let endpoints = vec![RegisteredEndpoint { + channel_name: channel_name.clone(), + path: webhook_path, + methods: vec!["POST".to_string()], + require_secret: webhook_secret.is_some(), + }]; + + let channel_arc = Arc::new(loaded.channel); + + // Inject runtime config (tunnel URL, webhook secret, owner_id). + { + let mut config_updates = std::collections::HashMap::new(); + + if let Some(ref tunnel_url) = config.tunnel.public_url { + config_updates.insert( + "tunnel_url".to_string(), + serde_json::Value::String(tunnel_url.clone()), + ); + } + + if let Some(ref secret) = webhook_secret { + config_updates.insert( + "webhook_secret".to_string(), + serde_json::Value::String(secret.clone()), + ); + } + + if let Some(&owner_id) = config + .channels + .wasm_channel_owner_ids + .get(channel_name.as_str()) + { + config_updates.insert("owner_id".to_string(), serde_json::json!(owner_id)); + } + + if !config_updates.is_empty() { + channel_arc.update_config(config_updates).await; + tracing::info!( + channel = %channel_name, + has_tunnel = config.tunnel.public_url.is_some(), + has_webhook_secret = webhook_secret.is_some(), + "Injected runtime config into channel" + ); + } + } + + tracing::info!( + channel = %channel_name, + has_webhook_secret = webhook_secret.is_some(), + secret_header = ?secret_header, + "Registering channel with router" + ); + + wasm_router + .register( + Arc::clone(&channel_arc), + endpoints, + webhook_secret.clone(), + secret_header, + ) + .await; + + // Register Ed25519 signature key if declared in capabilities. + if let Some(ref sig_key_name) = sig_key_secret_name + && let Some(secrets) = secrets_store + && let Ok(key_secret) = secrets.get_decrypted("default", sig_key_name).await + { + match wasm_router + .register_signature_key(&channel_name, key_secret.expose()) + .await + { + Ok(()) => { + tracing::info!(channel = %channel_name, "Registered Ed25519 signature key") + } + Err(e) => { + tracing::error!(channel = %channel_name, error = %e, "Invalid signature key in secrets store") + } + } + } + + // Register HMAC signing secret if declared in capabilities. + if let Some(ref hmac_secret_name) = hmac_secret_name + && let Some(secrets) = secrets_store + && let Ok(secret) = secrets.get_decrypted("default", hmac_secret_name).await + { + wasm_router + .register_hmac_secret(&channel_name, secret.expose()) + .await; + tracing::info!(channel = %channel_name, "Registered HMAC signing secret"); + } + + // Inject credentials from secrets store / environment. + if let Some(secrets) = secrets_store { + match inject_channel_credentials(&channel_arc, secrets.as_ref(), &channel_name).await { + Ok(count) => { + if count > 0 { + tracing::info!( + channel = %channel_name, + credentials_injected = count, + "Channel credentials injected" + ); + } + } + Err(e) => { + tracing::error!( + channel = %channel_name, + error = %e, + "Failed to inject channel credentials" + ); + } + } + } + + (channel_name, Box::new(SharedWasmChannel::new(channel_arc))) +} + +/// Inject credentials for a channel based on naming convention. +/// +/// Looks for secrets matching the pattern `{channel_name}_*` and injects them +/// as credential placeholders (e.g., `telegram_bot_token` -> `{TELEGRAM_BOT_TOKEN}`). +/// +/// Falls back to environment variables with the uppercase name if not found +/// in the secrets store (e.g., `TELEGRAM_BOT_TOKEN`). +pub async fn inject_channel_credentials( + channel: &Arc, + secrets: &dyn SecretsStore, + channel_name: &str, +) -> anyhow::Result { + let all_secrets = secrets + .list("default") + .await + .map_err(|e| anyhow::anyhow!("Failed to list secrets: {}", e))?; + + let prefix = format!("{}_", channel_name); + let mut count = 0; + let mut injected_placeholders = HashSet::new(); + + for secret_meta in all_secrets { + if !secret_meta.name.starts_with(&prefix) { + continue; + } + + let decrypted = match secrets.get_decrypted("default", &secret_meta.name).await { + Ok(d) => d, + Err(e) => { + tracing::warn!( + secret = %secret_meta.name, + error = %e, + "Failed to decrypt secret for channel credential injection" + ); + continue; + } + }; + + let placeholder = secret_meta.name.to_uppercase(); + + tracing::debug!( + channel = %channel_name, + secret = %secret_meta.name, + placeholder = %placeholder, + "Injecting credential" + ); + + channel + .set_credential(&placeholder, decrypted.expose().to_string()) + .await; + injected_placeholders.insert(placeholder); + count += 1; + } + + // Fall back to environment variables for required secrets not found in the store. + // This allows channels to work when configured via env vars (e.g., TELEGRAM_BOT_TOKEN) + // without requiring the setup wizard to have run. + let caps = channel.capabilities(); + if let Some(ref http_cap) = caps.tool_capabilities.http { + for cred_mapping in http_cap.credentials.values() { + let placeholder = cred_mapping.secret_name.to_uppercase(); + if injected_placeholders.contains(&placeholder) { + continue; + } + if let Ok(env_value) = std::env::var(&placeholder) + && !env_value.is_empty() + { + tracing::debug!( + channel = %channel_name, + placeholder = %placeholder, + "Injecting credential from environment variable" + ); + channel.set_credential(&placeholder, env_value).await; + count += 1; + } + } + } + + Ok(count) +} diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index f454363d..b69f8ac4 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -2609,6 +2609,7 @@ mod tests { let ext_mgr = Arc::new(ExtensionManager::new( mcp_sm, + Arc::new(crate::tools::mcp::process::McpProcessManager::new()), secrets, tool_registry, None, @@ -2658,6 +2659,7 @@ mod tests { let ext_mgr = Arc::new(ExtensionManager::new( mcp_sm, + Arc::new(crate::tools::mcp::process::McpProcessManager::new()), secrets.clone(), tool_registry, None, @@ -2763,6 +2765,7 @@ mod tests { let ext_mgr = Arc::new(ExtensionManager::new( mcp_sm, + Arc::new(crate::tools::mcp::process::McpProcessManager::new()), secrets.clone(), tool_registry, None, diff --git a/src/cli/mcp.rs b/src/cli/mcp.rs index b13bf598..5cc7b911 100644 --- a/src/cli/mcp.rs +++ b/src/cli/mcp.rs @@ -10,7 +10,7 @@ use clap::{Args, Subcommand}; use crate::config::Config; use crate::db::Database; -use crate::secrets::{SecretsCrypto, SecretsStore}; +use crate::secrets::SecretsStore; use crate::tools::mcp::{ McpClient, McpServerConfig, McpSessionManager, OAuthConfig, auth::{authorize_mcp_server, is_authenticated}, @@ -628,17 +628,7 @@ async fn save_servers( /// Initialize and return the secrets store. async fn get_secrets_store() -> anyhow::Result> { - let config = Config::from_env().await?; - - let master_key = config.secrets.master_key().ok_or_else(|| { - anyhow::anyhow!( - "SECRETS_MASTER_KEY not set. Run 'ironclaw onboard' first or set it in .env" - ) - })?; - - let crypto = Arc::new(SecretsCrypto::new(master_key.clone())?); - - Ok(crate::db::create_secrets_store(&config.database, crypto).await?) + crate::cli::init_secrets_store().await } #[cfg(test)] diff --git a/src/cli/mod.rs b/src/cli/mod.rs index 1e47ccfd..d7394d4e 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -28,8 +28,6 @@ pub use config::{ConfigCommand, run_config_command}; pub use doctor::run_doctor_command; pub use mcp::{McpCommand, run_mcp_command}; pub use memory::MemoryCommand; -#[cfg(feature = "postgres")] -pub use memory::run_memory_command; pub use memory::run_memory_command_with_db; pub use pairing::{PairingCommand, run_pairing_command, run_pairing_command_with_store}; pub use registry::{RegistryCommand, run_registry_command}; @@ -37,6 +35,8 @@ pub use service::{ServiceCommand, run_service_command}; pub use status::run_status_command; pub use tool::{ToolCommand, run_tool_command}; +use std::sync::Arc; + use clap::{ColorChoice, Parser, Subcommand}; #[derive(Parser, Debug)] @@ -225,6 +225,43 @@ impl Cli { } } +/// Initialize a secrets store from environment config. +/// +/// Shared helper for CLI subcommands (`mcp auth`, `tool auth`, etc.) that need +/// access to encrypted secrets without spinning up the full AppBuilder. +pub async fn init_secrets_store() +-> anyhow::Result> { + let config = crate::config::Config::from_env().await?; + let master_key = config.secrets.master_key().ok_or_else(|| { + anyhow::anyhow!( + "SECRETS_MASTER_KEY not set. Run 'ironclaw onboard' first or set it in .env" + ) + })?; + + let crypto = Arc::new(crate::secrets::SecretsCrypto::new(master_key.clone())?); + + Ok(crate::db::create_secrets_store(&config.database, crypto).await?) +} + +/// Run the Memory CLI subcommand. +pub async fn run_memory_command(mem_cmd: &MemoryCommand) -> anyhow::Result<()> { + let config = crate::config::Config::from_env() + .await + .map_err(|e| anyhow::anyhow!("{}", e))?; + + let session = crate::llm::create_session_manager(config.llm.session.clone()).await; + + let embeddings = config + .embeddings + .create_provider(&config.llm.nearai.base_url, session); + + let db: Arc = crate::db::connect_from_config(&config.database) + .await + .map_err(|e| anyhow::anyhow!("{}", e))?; + + run_memory_command_with_db(mem_cmd.clone(), db, embeddings).await +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/cli/tool.rs b/src/cli/tool.rs index 752f4263..ac5d1b37 100644 --- a/src/cli/tool.rs +++ b/src/cli/tool.rs @@ -10,8 +10,7 @@ use clap::Subcommand; use tokio::fs; use crate::bootstrap::ironclaw_base_dir; -use crate::config::Config; -use crate::secrets::{CreateSecretParams, SecretsCrypto, SecretsStore}; +use crate::secrets::{CreateSecretParams, SecretsStore}; use crate::tools::wasm::{CapabilitiesFile, compute_binary_hash}; /// Default tools directory. @@ -552,16 +551,7 @@ fn validate_tool_name(name: &str) -> anyhow::Result<()> { /// Initialize the secrets store from environment config. async fn init_secrets_store() -> anyhow::Result> { - let config = Config::from_env().await?; - let master_key = config.secrets.master_key().ok_or_else(|| { - anyhow::anyhow!( - "SECRETS_MASTER_KEY not set. Run 'ironclaw onboard' first or set it in .env" - ) - })?; - - let crypto = Arc::new(SecretsCrypto::new(master_key.clone())?); - - Ok(crate::db::create_secrets_store(&config.database, crypto).await?) + crate::cli::init_secrets_store().await } /// Configure authentication for a tool. diff --git a/src/db/mod.rs b/src/db/mod.rs index d7e11c12..9485d41c 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -51,6 +51,29 @@ use crate::workspace::{SearchConfig, SearchResult}; pub async fn connect_from_config( config: &crate::config::DatabaseConfig, ) -> Result, DatabaseError> { + let (db, _handles) = connect_with_handles(config).await?; + Ok(db) +} + +/// Backend-specific handles retained after database connection. +/// +/// These are needed by satellite stores (e.g., `SecretsStore`) that require +/// a backend-specific handle rather than the generic `Arc`. +#[derive(Default)] +pub struct DatabaseHandles { + #[cfg(feature = "postgres")] + pub pg_pool: Option, + #[cfg(feature = "libsql")] + pub libsql_db: Option>, +} + +/// Connect to the database, run migrations, and return both the generic +/// `Database` trait object and the backend-specific handles. +pub async fn connect_with_handles( + config: &crate::config::DatabaseConfig, +) -> Result<(Arc, DatabaseHandles), DatabaseError> { + let mut handles = DatabaseHandles::default(); + match config.backend { #[cfg(feature = "libsql")] crate::config::DatabaseBackend::LibSql => { @@ -74,7 +97,11 @@ pub async fn connect_from_config( .map_err(|e| DatabaseError::Pool(e.to_string()))? }; backend.run_migrations().await?; - Ok(Arc::new(backend)) + tracing::info!("libSQL database connected and migrations applied"); + + handles.libsql_db = Some(backend.shared_db()); + + Ok((Arc::new(backend) as Arc, handles)) } #[cfg(feature = "postgres")] _ => { @@ -82,7 +109,11 @@ pub async fn connect_from_config( .await .map_err(|e| DatabaseError::Pool(e.to_string()))?; pg.run_migrations().await?; - Ok(Arc::new(pg)) + tracing::info!("PostgreSQL database connected and migrations applied"); + + handles.pg_pool = Some(pg.pool()); + + Ok((Arc::new(pg) as Arc, handles)) } #[cfg(not(feature = "postgres"))] _ => Err(DatabaseError::Pool( diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index 3f51511e..5e74c344 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -73,6 +73,7 @@ pub struct ExtensionManager { // MCP infrastructure mcp_session_manager: Arc, + mcp_process_manager: Arc, /// Active MCP clients keyed by server name. mcp_clients: RwLock>>, @@ -116,6 +117,7 @@ impl ExtensionManager { #[allow(clippy::too_many_arguments)] pub fn new( mcp_session_manager: Arc, + mcp_process_manager: Arc, secrets: Arc, tool_registry: Arc, hooks: Option>, @@ -136,6 +138,7 @@ impl ExtensionManager { registry, discovery: OnlineDiscovery::new(), mcp_session_manager, + mcp_process_manager, mcp_clients: RwLock::new(HashMap::new()), wasm_tool_runtime, wasm_tools_dir, @@ -2467,18 +2470,15 @@ impl ExtensionManager { .await .map_err(|e| ExtensionError::NotInstalled(e.to_string()))?; - let has_tokens = is_authenticated(&server, &self.secrets, &self.user_id).await; - - let client = if has_tokens || server.requires_auth() { - McpClient::new_authenticated( - server.clone(), - Arc::clone(&self.mcp_session_manager), - Arc::clone(&self.secrets), - &self.user_id, - ) - } else { - McpClient::new_with_config(server.clone()) - }; + let client = crate::tools::mcp::create_client_from_config( + server.clone(), + &self.mcp_session_manager, + &self.mcp_process_manager, + Some(Arc::clone(&self.secrets)), + &self.user_id, + ) + .await + .map_err(|e| ExtensionError::ActivationFailed(e.to_string()))?; // Try to list and create tools let mcp_tools = client @@ -3736,6 +3736,7 @@ mod tests { tools_dir: std::path::PathBuf, ) -> crate::extensions::manager::ExtensionManager { use crate::secrets::{InMemorySecretsStore, SecretsCrypto}; + use crate::tools::mcp::process::McpProcessManager; use crate::tools::mcp::session::McpSessionManager; let key = secrecy::SecretString::from(crate::secrets::keychain::generate_master_key_hex()); @@ -3747,6 +3748,7 @@ mod tests { crate::extensions::manager::ExtensionManager::new( mcp, + Arc::new(McpProcessManager::new()), secrets, tools, None, // hooks @@ -3906,6 +3908,7 @@ mod tests { ) -> ExtensionManager { use crate::secrets::{InMemorySecretsStore, SecretsCrypto}; use crate::tools::ToolRegistry; + use crate::tools::mcp::process::McpProcessManager; use crate::tools::mcp::session::McpSessionManager; std::fs::create_dir_all(&tools_dir).ok(); @@ -3917,6 +3920,7 @@ mod tests { ExtensionManager::new( Arc::new(McpSessionManager::new()), + Arc::new(McpProcessManager::new()), Arc::new(InMemorySecretsStore::new(crypto)), Arc::new(ToolRegistry::new()), None, diff --git a/src/main.rs b/src/main.rs index f594bf41..15f8cdaa 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,7 +4,6 @@ use std::sync::Arc; use std::time::Duration; use clap::Parser; -use tracing_subscriber::EnvFilter; use ironclaw::{ agent::{Agent, AgentDeps}, @@ -12,10 +11,7 @@ use ironclaw::{ channels::{ ChannelManager, GatewayChannel, HttpChannel, ReplChannel, SignalChannel, WebhookServer, WebhookServerConfig, - wasm::{ - RegisteredEndpoint, SharedWasmChannel, WasmChannelLoader, WasmChannelRouter, - WasmChannelRuntime, WasmChannelRuntimeConfig, create_wasm_channel_router, - }, + wasm::{WasmChannelRouter, WasmChannelRuntime}, web::log_layer::LogBroadcaster, }, cli::{ @@ -25,26 +21,14 @@ use ironclaw::{ config::Config, hooks::bootstrap_hooks, llm::create_session_manager, - orchestrator::{ - ContainerJobConfig, ContainerJobManager, OrchestratorApi, ReaperConfig, SandboxReaper, - TokenStore, api::OrchestratorState, - }, + orchestrator::{ReaperConfig, SandboxReaper}, pairing::PairingStore, - secrets::SecretsStore, + tracing_fmt::{init_cli_tracing, init_worker_tracing}, }; #[cfg(any(feature = "postgres", feature = "libsql"))] use ironclaw::setup::{SetupConfig, SetupWizard}; -/// Initialize tracing for simple CLI commands (warn level, no fancy layers). -fn init_cli_tracing() { - tracing_subscriber::fmt() - .with_env_filter( - EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("warn")), - ) - .init(); -} - /// Synchronous entry point. Loads `.env` files before the Tokio runtime /// starts so that `std::env::set_var` is safe (no worker threads yet). fn main() -> anyhow::Result<()> { @@ -80,7 +64,7 @@ async fn async_main() -> anyhow::Result<()> { } Some(Command::Memory(mem_cmd)) => { init_cli_tracing(); - return run_memory_command(mem_cmd).await; + return ironclaw::cli::run_memory_command(mem_cmd).await; } Some(Command::Pairing(pairing_cmd)) => { init_cli_tracing(); @@ -108,7 +92,7 @@ async fn async_main() -> anyhow::Result<()> { max_iterations, }) => { init_worker_tracing(); - return run_worker(*job_id, orchestrator_url, *max_iterations).await; + return ironclaw::worker::run_worker(*job_id, orchestrator_url, *max_iterations).await; } Some(Command::ClaudeBridge { job_id, @@ -117,7 +101,13 @@ async fn async_main() -> anyhow::Result<()> { model, }) => { init_worker_tracing(); - return run_claude_bridge(*job_id, orchestrator_url, *max_turns, model).await; + return ironclaw::worker::run_claude_bridge( + *job_id, + orchestrator_url, + *max_turns, + model, + ) + .await; } Some(Command::Onboard { skip_auth, @@ -169,7 +159,7 @@ async fn async_main() -> anyhow::Result<()> { // Enhanced first-run detection #[cfg(any(feature = "postgres", feature = "libsql"))] if !cli.no_onboard - && let Some(reason) = check_onboard_needed() + && let Some(reason) = ironclaw::setup::check_onboard_needed() { println!("Onboarding needed: {}", reason); println!(); @@ -227,95 +217,21 @@ async fn async_main() -> anyhow::Result<()> { // ── Tunnel setup ─────────────────────────────────────────────────── - let (config, active_tunnel) = start_tunnel(config).await; + let (config, active_tunnel) = ironclaw::tunnel::start_managed_tunnel(config).await; // ── Orchestrator / container job manager ──────────────────────────── - // Proactive Docker detection - let docker_status = if config.sandbox.enabled { - let detection = ironclaw::sandbox::check_docker().await; - match detection.status { - ironclaw::sandbox::DockerStatus::Available => { - tracing::info!("Docker is available"); - } - ironclaw::sandbox::DockerStatus::NotInstalled => { - tracing::warn!( - "Docker is not installed -- sandbox disabled for this session. {}", - detection.platform.install_hint() - ); - } - ironclaw::sandbox::DockerStatus::NotRunning => { - tracing::warn!( - "Docker is installed but not running -- sandbox disabled for this session. {}", - detection.platform.start_hint() - ); - } - ironclaw::sandbox::DockerStatus::Disabled => {} - } - detection.status - } else { - ironclaw::sandbox::DockerStatus::Disabled - }; - - let job_event_tx: Option< - tokio::sync::broadcast::Sender<(uuid::Uuid, ironclaw::channels::web::types::SseEvent)>, - > = if config.sandbox.enabled && docker_status.is_ok() { - let (tx, _) = tokio::sync::broadcast::channel(256); - Some(tx) - } else { - None - }; - let prompt_queue = Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::< - uuid::Uuid, - std::collections::VecDeque, - >::new())); - - let container_job_manager: Option> = - if config.sandbox.enabled && docker_status.is_ok() { - let token_store = TokenStore::new(); - let job_config = ContainerJobConfig { - image: config.sandbox.image.clone(), - memory_limit_mb: config.sandbox.memory_limit_mb, - cpu_shares: config.sandbox.cpu_shares, - orchestrator_port: 50051, - claude_code_api_key: std::env::var("ANTHROPIC_API_KEY").ok(), - claude_code_oauth_token: ironclaw::config::ClaudeCodeConfig::extract_oauth_token(), - claude_code_model: config.claude_code.model.clone(), - claude_code_max_turns: config.claude_code.max_turns, - claude_code_memory_limit_mb: config.claude_code.memory_limit_mb, - claude_code_allowed_tools: config.claude_code.allowed_tools.clone(), - }; - let jm = Arc::new(ContainerJobManager::new(job_config, token_store.clone())); - - // Start the orchestrator internal API in the background - let orchestrator_state = OrchestratorState { - llm: components.llm.clone(), - job_manager: Arc::clone(&jm), - token_store, - job_event_tx: job_event_tx.clone(), - prompt_queue: Arc::clone(&prompt_queue), - store: components.db.clone(), - secrets_store: components.secrets_store.clone(), - user_id: "default".to_string(), - }; - - tokio::spawn(async move { - if let Err(e) = OrchestratorApi::start(orchestrator_state, 50051).await { - tracing::error!("Orchestrator API failed: {}", e); - } - }); - - if config.claude_code.enabled { - tracing::info!( - "Claude Code sandbox mode available (model: {}, max_turns: {})", - config.claude_code.model, - config.claude_code.max_turns - ); - } - Some(jm) - } else { - None - }; + let orch = ironclaw::orchestrator::setup_orchestrator( + &config, + &components.llm, + components.db.as_ref(), + components.secrets_store.as_ref(), + ) + .await; + let container_job_manager = orch.container_job_manager; + let job_event_tx = orch.job_event_tx; + let prompt_queue = orch.prompt_queue; + let docker_status = orch.docker_status; // ── Channel setup ────────────────────────────────────────────────── @@ -355,7 +271,7 @@ async fn async_main() -> anyhow::Result<()> { // Load WASM channels and register their webhook routes. if config.channels.wasm_channels_enabled && config.channels.wasm_channels_dir.exists() { - let wasm_result = setup_wasm_channels( + let wasm_result = ironclaw::channels::wasm::setup_wasm_channels( &config, &components.secrets_store, components.extension_manager.as_ref(), @@ -769,475 +685,3 @@ async fn async_main() -> anyhow::Result<()> { Ok(()) } - -// ── Helper functions ──────────────────────────────────────────────────── - -/// Initialize tracing for worker/bridge processes (info level). -fn init_worker_tracing() { - tracing_subscriber::fmt() - .with_env_filter( - EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("ironclaw=info")), - ) - .init(); -} - -/// Run the Memory CLI subcommand. -async fn run_memory_command(mem_cmd: &ironclaw::cli::MemoryCommand) -> anyhow::Result<()> { - let config = Config::from_env() - .await - .map_err(|e| anyhow::anyhow!("{}", e))?; - - let session = create_session_manager(config.llm.session.clone()).await; - - let embeddings = config - .embeddings - .create_provider(&config.llm.nearai.base_url, session); - - let db: Arc = ironclaw::db::connect_from_config(&config.database) - .await - .map_err(|e| anyhow::anyhow!("{}", e))?; - - ironclaw::cli::run_memory_command_with_db(mem_cmd.clone(), db, embeddings).await -} - -/// Run the Worker subcommand (inside Docker containers). -async fn run_worker( - job_id: uuid::Uuid, - orchestrator_url: &str, - max_iterations: u32, -) -> anyhow::Result<()> { - tracing::info!( - "Starting worker for job {} (orchestrator: {})", - job_id, - orchestrator_url - ); - - let config = ironclaw::worker::runtime::WorkerConfig { - job_id, - orchestrator_url: orchestrator_url.to_string(), - max_iterations, - timeout: std::time::Duration::from_secs(600), - }; - - let runtime = ironclaw::worker::WorkerRuntime::new(config) - .map_err(|e| anyhow::anyhow!("Worker init failed: {}", e))?; - - runtime - .run() - .await - .map_err(|e| anyhow::anyhow!("Worker failed: {}", e)) -} - -/// Run the Claude Code bridge subcommand (inside Docker containers). -async fn run_claude_bridge( - job_id: uuid::Uuid, - orchestrator_url: &str, - max_turns: u32, - model: &str, -) -> anyhow::Result<()> { - tracing::info!( - "Starting Claude Code bridge for job {} (orchestrator: {}, model: {})", - job_id, - orchestrator_url, - model - ); - - let config = ironclaw::worker::claude_bridge::ClaudeBridgeConfig { - job_id, - orchestrator_url: orchestrator_url.to_string(), - max_turns, - model: model.to_string(), - timeout: std::time::Duration::from_secs(1800), - allowed_tools: ironclaw::config::ClaudeCodeConfig::from_env().allowed_tools, - }; - - let runtime = ironclaw::worker::ClaudeBridgeRuntime::new(config) - .map_err(|e| anyhow::anyhow!("Claude bridge init failed: {}", e))?; - - runtime - .run() - .await - .map_err(|e| anyhow::anyhow!("Claude bridge failed: {}", e)) -} - -/// Start managed tunnel if configured and no static URL is already set. -async fn start_tunnel( - mut config: ironclaw::config::Config, -) -> ( - ironclaw::config::Config, - Option>, -) { - if config.tunnel.public_url.is_some() { - tracing::info!( - "Static tunnel URL in use: {}", - config.tunnel.public_url.as_deref().unwrap_or("?") - ); - return (config, None); - } - - let Some(ref provider_config) = config.tunnel.provider else { - return (config, None); - }; - - let gateway_port = config - .channels - .gateway - .as_ref() - .map(|g| g.port) - .unwrap_or(3000); - let gateway_host = config - .channels - .gateway - .as_ref() - .map(|g| g.host.as_str()) - .unwrap_or("127.0.0.1"); - - match ironclaw::tunnel::create_tunnel(provider_config) { - Ok(Some(tunnel)) => { - tracing::info!( - "Starting {} tunnel on {}:{}...", - tunnel.name(), - gateway_host, - gateway_port - ); - match tunnel.start(gateway_host, gateway_port).await { - Ok(url) => { - tracing::info!("Tunnel started: {}", url); - config.tunnel.public_url = Some(url); - (config, Some(tunnel)) - } - Err(e) => { - tracing::error!("Failed to start tunnel: {}", e); - (config, None) - } - } - } - Ok(None) => (config, None), - Err(e) => { - tracing::error!("Failed to create tunnel: {}", e); - (config, None) - } - } -} - -/// Result of WASM channel setup. -struct WasmChannelSetup { - channels: Vec<(String, Box)>, - channel_names: Vec, - webhook_routes: Option, - /// Runtime objects needed for hot-activation via ExtensionManager. - wasm_channel_runtime: Arc, - pairing_store: Arc, - wasm_channel_router: Arc, -} - -/// Load WASM channels and register their webhook routes. -async fn setup_wasm_channels( - config: &ironclaw::config::Config, - secrets_store: &Option>, - extension_manager: Option<&Arc>, - database: Option<&Arc>, -) -> Option { - let runtime = match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) { - Ok(r) => Arc::new(r), - Err(e) => { - tracing::warn!("Failed to initialize WASM channel runtime: {}", e); - return None; - } - }; - - let pairing_store = Arc::new(PairingStore::new()); - let settings_store: Option> = - database.map(|db| Arc::clone(db) as Arc); - let mut loader = WasmChannelLoader::new( - Arc::clone(&runtime), - Arc::clone(&pairing_store), - settings_store, - ); - if let Some(secrets) = secrets_store { - loader = loader.with_secrets_store(Arc::clone(secrets)); - } - - let results = match loader - .load_from_dir(&config.channels.wasm_channels_dir) - .await - { - Ok(r) => r, - Err(e) => { - tracing::warn!("Failed to scan WASM channels directory: {}", e); - return None; - } - }; - - let wasm_router = Arc::new(WasmChannelRouter::new()); - let mut channels: Vec<(String, Box)> = Vec::new(); - let mut channel_names: Vec = Vec::new(); - - for loaded in results.loaded { - let channel_name = loaded.name().to_string(); - channel_names.push(channel_name.clone()); - tracing::info!("Loaded WASM channel: {}", channel_name); - - let secret_name = loaded.webhook_secret_name(); - let sig_key_secret_name = loaded.signature_key_secret_name(); - let hmac_secret_name = loaded.hmac_secret_name(); - - let webhook_secret = if let Some(secrets) = secrets_store { - secrets - .get_decrypted("default", &secret_name) - .await - .ok() - .map(|s| s.expose().to_string()) - } else { - None - }; - - let secret_header = loaded.webhook_secret_header().map(|s| s.to_string()); - - let webhook_path = format!("/webhook/{}", channel_name); - let endpoints = vec![RegisteredEndpoint { - channel_name: channel_name.clone(), - path: webhook_path, - methods: vec!["POST".to_string()], - require_secret: webhook_secret.is_some(), - }]; - - let channel_arc = Arc::new(loaded.channel); - - { - let mut config_updates = std::collections::HashMap::new(); - - if let Some(ref tunnel_url) = config.tunnel.public_url { - config_updates.insert( - "tunnel_url".to_string(), - serde_json::Value::String(tunnel_url.clone()), - ); - } - - if let Some(ref secret) = webhook_secret { - config_updates.insert( - "webhook_secret".to_string(), - serde_json::Value::String(secret.clone()), - ); - } - - // Inject owner_id if configured for this channel. - if let Some(&owner_id) = config - .channels - .wasm_channel_owner_ids - .get(channel_name.as_str()) - { - config_updates.insert("owner_id".to_string(), serde_json::json!(owner_id)); - } - - if !config_updates.is_empty() { - channel_arc.update_config(config_updates).await; - tracing::info!( - channel = %channel_name, - has_tunnel = config.tunnel.public_url.is_some(), - has_webhook_secret = webhook_secret.is_some(), - "Injected runtime config into channel" - ); - } - } - - tracing::info!( - channel = %channel_name, - has_webhook_secret = webhook_secret.is_some(), - secret_header = ?secret_header, - "Registering channel with router" - ); - - wasm_router - .register( - Arc::clone(&channel_arc), - endpoints, - webhook_secret.clone(), - secret_header, - ) - .await; - - // Register Ed25519 signature key if declared in capabilities - if let Some(ref sig_key_name) = sig_key_secret_name - && let Some(secrets) = secrets_store - && let Ok(key_secret) = secrets.get_decrypted("default", sig_key_name).await - { - match wasm_router - .register_signature_key(&channel_name, key_secret.expose()) - .await - { - Ok(()) => { - tracing::info!(channel = %channel_name, "Registered Ed25519 signature key") - } - Err(e) => { - tracing::error!(channel = %channel_name, error = %e, "Invalid signature key in secrets store") - } - } - } - - // Register HMAC signing secret if declared in capabilities - if let Some(ref hmac_secret_name) = hmac_secret_name - && let Some(secrets) = secrets_store - && let Ok(secret) = secrets.get_decrypted("default", hmac_secret_name).await - { - wasm_router - .register_hmac_secret(&channel_name, secret.expose()) - .await; - tracing::info!(channel = %channel_name, "Registered HMAC signing secret"); - } - - if let Some(secrets) = secrets_store { - match inject_channel_credentials(&channel_arc, secrets.as_ref(), &channel_name).await { - Ok(count) => { - if count > 0 { - tracing::info!( - channel = %channel_name, - credentials_injected = count, - "Channel credentials injected" - ); - } - } - Err(e) => { - tracing::error!( - channel = %channel_name, - error = %e, - "Failed to inject channel credentials" - ); - } - } - } - - channels.push((channel_name, Box::new(SharedWasmChannel::new(channel_arc)))); - } - - for (path, err) in &results.errors { - tracing::warn!("Failed to load WASM channel {}: {}", path.display(), err); - } - - // Always create webhook routes (even with no channels loaded) so that - // channels hot-added at runtime can receive webhooks without a restart. - let webhook_routes = { - Some(create_wasm_channel_router( - Arc::clone(&wasm_router), - extension_manager.map(Arc::clone), - )) - }; - - Some(WasmChannelSetup { - channels, - channel_names, - webhook_routes, - wasm_channel_runtime: runtime, - pairing_store, - wasm_channel_router: wasm_router, - }) -} - -/// Check if onboarding is needed and return the reason. -#[cfg(any(feature = "postgres", feature = "libsql"))] -fn check_onboard_needed() -> Option<&'static str> { - let has_db = std::env::var("DATABASE_URL").is_ok() - || std::env::var("LIBSQL_PATH").is_ok() - || ironclaw::config::default_libsql_path().exists(); - - if !has_db { - return Some("Database not configured"); - } - - if std::env::var("ONBOARD_COMPLETED") - .map(|v| v == "true") - .unwrap_or(false) - { - return None; - } - - if std::env::var("NEARAI_API_KEY").is_err() { - let session_path = ironclaw::config::default_session_path(); - if !session_path.exists() { - return Some("First run"); - } - } - - None -} - -/// Inject credentials for a channel based on naming convention. -/// -/// Looks for secrets matching the pattern `{channel_name}_*` and injects them -/// as credential placeholders (e.g., `telegram_bot_token` -> `{TELEGRAM_BOT_TOKEN}`). -/// -/// Falls back to environment variables with the uppercase name if not found -/// in the secrets store (e.g., `TELEGRAM_BOT_TOKEN`). -async fn inject_channel_credentials( - channel: &Arc, - secrets: &dyn SecretsStore, - channel_name: &str, -) -> anyhow::Result { - let all_secrets = secrets - .list("default") - .await - .map_err(|e| anyhow::anyhow!("Failed to list secrets: {}", e))?; - - let prefix = format!("{}_", channel_name); - let mut count = 0; - let mut injected_placeholders = std::collections::HashSet::new(); - - for secret_meta in all_secrets { - if !secret_meta.name.starts_with(&prefix) { - continue; - } - - let decrypted = match secrets.get_decrypted("default", &secret_meta.name).await { - Ok(d) => d, - Err(e) => { - tracing::warn!( - secret = %secret_meta.name, - error = %e, - "Failed to decrypt secret for channel credential injection" - ); - continue; - } - }; - - let placeholder = secret_meta.name.to_uppercase(); - - tracing::debug!( - channel = %channel_name, - secret = %secret_meta.name, - placeholder = %placeholder, - "Injecting credential" - ); - - channel - .set_credential(&placeholder, decrypted.expose().to_string()) - .await; - injected_placeholders.insert(placeholder); - count += 1; - } - - // Fall back to environment variables for required secrets not found in the store. - // This allows channels to work when configured via env vars (e.g., TELEGRAM_BOT_TOKEN) - // without requiring the setup wizard to have run. - let caps = channel.capabilities(); - if let Some(ref http_cap) = caps.tool_capabilities.http { - for cred_mapping in http_cap.credentials.values() { - let placeholder = cred_mapping.secret_name.to_uppercase(); - if injected_placeholders.contains(&placeholder) { - continue; - } - if let Ok(env_value) = std::env::var(&placeholder) - && !env_value.is_empty() - { - tracing::debug!( - channel = %channel_name, - placeholder = %placeholder, - "Injecting credential from environment variable" - ); - channel.set_credential(&placeholder, env_value).await; - count += 1; - } - } - } - - Ok(count) -} diff --git a/src/orchestrator/mod.rs b/src/orchestrator/mod.rs index 084ae04e..5e750ddf 100644 --- a/src/orchestrator/mod.rs +++ b/src/orchestrator/mod.rs @@ -39,3 +39,115 @@ pub use job_manager::{ CompletionResult, ContainerHandle, ContainerJobConfig, ContainerJobManager, JobMode, }; pub use reaper::{ReaperConfig, SandboxReaper}; + +use std::collections::{HashMap, VecDeque}; +use std::sync::Arc; + +use tokio::sync::{Mutex, broadcast}; +use uuid::Uuid; + +use crate::channels::web::types::SseEvent; +use crate::db::Database; +use crate::llm::LlmProvider; +use crate::secrets::SecretsStore; + +/// Result of orchestrator setup, containing all handles needed by the agent. +pub struct OrchestratorSetup { + pub container_job_manager: Option>, + pub job_event_tx: Option>, + pub prompt_queue: Arc>>>, + pub docker_status: crate::sandbox::DockerStatus, +} + +/// Detect Docker availability, create the container job manager, and start +/// the orchestrator internal API in the background. +pub async fn setup_orchestrator( + config: &crate::config::Config, + llm: &Arc, + db: Option<&Arc>, + secrets_store: Option<&Arc>, +) -> OrchestratorSetup { + let prompt_queue = Arc::new(Mutex::new( + HashMap::>::new(), + )); + + let docker_status = if config.sandbox.enabled { + let detection = crate::sandbox::check_docker().await; + match detection.status { + crate::sandbox::DockerStatus::Available => { + tracing::info!("Docker is available"); + } + crate::sandbox::DockerStatus::NotInstalled => { + tracing::warn!( + "Docker is not installed -- sandbox disabled for this session. {}", + detection.platform.install_hint() + ); + } + crate::sandbox::DockerStatus::NotRunning => { + tracing::warn!( + "Docker is installed but not running -- sandbox disabled for this session. {}", + detection.platform.start_hint() + ); + } + crate::sandbox::DockerStatus::Disabled => {} + } + detection.status + } else { + crate::sandbox::DockerStatus::Disabled + }; + + let (job_event_tx, container_job_manager) = if config.sandbox.enabled && docker_status.is_ok() { + let (tx, _) = broadcast::channel(256); + let job_event_tx = Some(tx); + + let token_store = TokenStore::new(); + let job_config = ContainerJobConfig { + image: config.sandbox.image.clone(), + memory_limit_mb: config.sandbox.memory_limit_mb, + cpu_shares: config.sandbox.cpu_shares, + orchestrator_port: 50051, + claude_code_api_key: std::env::var("ANTHROPIC_API_KEY").ok(), + claude_code_oauth_token: crate::config::ClaudeCodeConfig::extract_oauth_token(), + claude_code_model: config.claude_code.model.clone(), + claude_code_max_turns: config.claude_code.max_turns, + claude_code_memory_limit_mb: config.claude_code.memory_limit_mb, + claude_code_allowed_tools: config.claude_code.allowed_tools.clone(), + }; + let jm = Arc::new(ContainerJobManager::new(job_config, token_store.clone())); + + let orchestrator_state = api::OrchestratorState { + llm: Arc::clone(llm), + job_manager: Arc::clone(&jm), + token_store, + job_event_tx: job_event_tx.clone(), + prompt_queue: Arc::clone(&prompt_queue), + store: db.cloned(), + secrets_store: secrets_store.cloned(), + user_id: "default".to_string(), + }; + + tokio::spawn(async move { + if let Err(e) = OrchestratorApi::start(orchestrator_state, 50051).await { + tracing::error!("Orchestrator API failed: {}", e); + } + }); + + if config.claude_code.enabled { + tracing::info!( + "Claude Code sandbox mode available (model: {}, max_turns: {})", + config.claude_code.model, + config.claude_code.max_turns + ); + } + (job_event_tx, Some(jm)) + } else { + (None, None) + }; + + OrchestratorSetup { + container_job_manager, + job_event_tx, + prompt_queue, + docker_status, + } +} diff --git a/src/secrets/mod.rs b/src/secrets/mod.rs index 323f17c9..9ebad715 100644 --- a/src/secrets/mod.rs +++ b/src/secrets/mod.rs @@ -75,3 +75,37 @@ pub use types::{ }; pub use store::in_memory::InMemorySecretsStore; + +/// Create a secrets store from a master key and database handles. +/// +/// Returns `None` if no matching backend handle is available (e.g. when +/// running without a database). This is a normal condition in no-db mode, +/// not an error — callers should treat `None` as "secrets unavailable". +pub fn create_secrets_store( + crypto: std::sync::Arc, + handles: &crate::db::DatabaseHandles, +) -> Option> { + let store: Option> = None; + + #[cfg(feature = "libsql")] + let store = store.or_else(|| { + handles.libsql_db.as_ref().map(|db| { + std::sync::Arc::new(LibSqlSecretsStore::new( + std::sync::Arc::clone(db), + std::sync::Arc::clone(&crypto), + )) as std::sync::Arc + }) + }); + + #[cfg(feature = "postgres")] + let store = store.or_else(|| { + handles.pg_pool.as_ref().map(|pool| { + std::sync::Arc::new(PostgresSecretsStore::new( + pool.clone(), + std::sync::Arc::clone(&crypto), + )) as std::sync::Arc + }) + }); + + store +} diff --git a/src/setup/mod.rs b/src/setup/mod.rs index a0ea82ce..bf8ca6e4 100644 --- a/src/setup/mod.rs +++ b/src/setup/mod.rs @@ -31,3 +31,35 @@ pub use prompts::{ }; #[cfg(any(feature = "postgres", feature = "libsql"))] pub use wizard::{SetupConfig, SetupWizard}; + +/// Check if onboarding is needed and return the reason. +/// +/// Reads environment variables (`DATABASE_URL`, `LIBSQL_PATH`, +/// `ONBOARD_COMPLETED`, `NEARAI_API_KEY`) and checks for the default +/// session file on disk. Not safe to call concurrently with `env::set_var`. +#[cfg(any(feature = "postgres", feature = "libsql"))] +pub fn check_onboard_needed() -> Option<&'static str> { + let has_db = std::env::var("DATABASE_URL").is_ok() + || std::env::var("LIBSQL_PATH").is_ok() + || crate::config::default_libsql_path().exists(); + + if !has_db { + return Some("Database not configured"); + } + + if std::env::var("ONBOARD_COMPLETED") + .map(|v| v == "true") + .unwrap_or(false) + { + return None; + } + + if std::env::var("NEARAI_API_KEY").is_err() { + let session_path = crate::config::default_session_path(); + if !session_path.exists() { + return Some("First run"); + } + } + + None +} diff --git a/src/tools/builtin/extension_tools.rs b/src/tools/builtin/extension_tools.rs index 00c79548..ce8a06a8 100644 --- a/src/tools/builtin/extension_tools.rs +++ b/src/tools/builtin/extension_tools.rs @@ -777,6 +777,7 @@ mod tests { Arc::new(ExtensionManager::new( Arc::new(McpSessionManager::new()), + Arc::new(crate::tools::mcp::process::McpProcessManager::new()), Arc::new(InMemorySecretsStore::new(crypto)), Arc::new(ToolRegistry::new()), None, diff --git a/src/tools/mcp/factory.rs b/src/tools/mcp/factory.rs new file mode 100644 index 00000000..b5acb3f9 --- /dev/null +++ b/src/tools/mcp/factory.rs @@ -0,0 +1,98 @@ +//! Factory for creating MCP clients from server configuration. +//! +//! Encapsulates the transport dispatch logic (stdio, Unix socket, HTTP) +//! so that callers don't need to match on `EffectiveTransport` themselves. + +use std::sync::Arc; + +use crate::secrets::SecretsStore; +use crate::tools::mcp::config::{EffectiveTransport, McpServerConfig}; +use crate::tools::mcp::{McpClient, McpProcessManager, McpSessionManager, McpTransport}; + +/// Error returned when MCP client creation fails. +#[derive(Debug, thiserror::Error)] +pub enum McpFactoryError { + #[error("Failed to spawn stdio MCP server '{name}': {reason}")] + StdioSpawn { name: String, reason: String }, + #[error("Failed to connect to Unix MCP server '{name}': {reason}")] + UnixConnect { name: String, reason: String }, + #[error("Unix socket transport is not supported on this platform (server '{name}')")] + UnixNotSupported { name: String }, +} + +/// Create an `McpClient` from a server configuration, dispatching on the +/// effective transport type. +pub async fn create_client_from_config( + server: McpServerConfig, + session_manager: &Arc, + process_manager: &Arc, + secrets: Option>, + user_id: &str, +) -> Result { + let server_name = server.name.clone(); + + match server.effective_transport() { + EffectiveTransport::Stdio { command, args, env } => { + let transport = process_manager + .spawn_stdio(&server_name, command, args.to_vec(), env.clone()) + .await + .map_err(|e| McpFactoryError::StdioSpawn { + name: server_name.clone(), + reason: e.to_string(), + })?; + + Ok(McpClient::new_with_transport( + &server_name, + transport as Arc, + None, + secrets, + user_id, + Some(server), + )) + } + #[cfg(unix)] + EffectiveTransport::Unix { socket_path } => { + let transport = crate::tools::mcp::unix_transport::UnixMcpTransport::connect( + &server_name, + socket_path, + ) + .await + .map_err(|e| McpFactoryError::UnixConnect { + name: server_name.clone(), + reason: e.to_string(), + })?; + + Ok(McpClient::new_with_transport( + &server_name, + Arc::new(transport) as Arc, + None, + secrets, + user_id, + Some(server), + )) + } + #[cfg(not(unix))] + EffectiveTransport::Unix { .. } => { + Err(McpFactoryError::UnixNotSupported { name: server_name }) + } + EffectiveTransport::Http => { + if let Some(ref secrets) = secrets { + let has_tokens = + crate::tools::mcp::is_authenticated(&server, secrets, user_id).await; + + if has_tokens || server.requires_auth() { + Ok(McpClient::new_authenticated( + server, + Arc::clone(session_manager), + Arc::clone(secrets), + user_id, + )) + } else { + Ok(McpClient::new_with_config(server)) + } + } else { + Ok(McpClient::new_with_config(server)) + } + } + } +} diff --git a/src/tools/mcp/mod.rs b/src/tools/mcp/mod.rs index 8ab107c9..49a4d7b0 100644 --- a/src/tools/mcp/mod.rs +++ b/src/tools/mcp/mod.rs @@ -31,6 +31,7 @@ pub mod auth; mod client; pub mod config; +pub mod factory; pub(crate) mod http_transport; pub(crate) mod process; mod protocol; @@ -43,6 +44,7 @@ pub(crate) mod unix_transport; pub use auth::{is_authenticated, refresh_access_token}; pub use client::McpClient; pub use config::{McpServerConfig, McpServersFile, OAuthConfig}; +pub use factory::{McpFactoryError, create_client_from_config}; pub use process::McpProcessManager; pub use protocol::{InitializeResult, McpRequest, McpResponse, McpTool}; pub use session::McpSessionManager; diff --git a/src/tracing_fmt.rs b/src/tracing_fmt.rs index f0d7073a..5a9f61e0 100644 --- a/src/tracing_fmt.rs +++ b/src/tracing_fmt.rs @@ -21,8 +21,27 @@ use std::io::{self, Write}; +use tracing_subscriber::EnvFilter; use tracing_subscriber::fmt::MakeWriter; +/// Initialize tracing for simple CLI commands (warn level, no fancy layers). +pub fn init_cli_tracing() { + tracing_subscriber::fmt() + .with_env_filter( + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("warn")), + ) + .init(); +} + +/// Initialize tracing for worker/bridge processes (info level). +pub fn init_worker_tracing() { + tracing_subscriber::fmt() + .with_env_filter( + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("ironclaw=info")), + ) + .init(); +} + /// Maximum bytes per tracing event written to the terminal. const TERMINAL_MAX_EVENT_BYTES: usize = 500; diff --git a/src/tunnel/mod.rs b/src/tunnel/mod.rs index 5551f2ed..38ad814b 100644 --- a/src/tunnel/mod.rs +++ b/src/tunnel/mod.rs @@ -180,6 +180,68 @@ pub fn create_tunnel(config: &TunnelProviderConfig) -> Result (crate::config::Config, Option>) { + if config.tunnel.public_url.is_some() { + tracing::info!( + "Static tunnel URL in use: {}", + config.tunnel.public_url.as_deref().unwrap_or("?") + ); + return (config, None); + } + + let Some(ref provider_config) = config.tunnel.provider else { + return (config, None); + }; + + let gateway_port = config + .channels + .gateway + .as_ref() + .map(|g| g.port) + .unwrap_or(3000); + let gateway_host = config + .channels + .gateway + .as_ref() + .map(|g| g.host.as_str()) + .unwrap_or("127.0.0.1"); + + match create_tunnel(provider_config) { + Ok(Some(tunnel)) => { + tracing::info!( + "Starting {} tunnel on {}:{}...", + tunnel.name(), + gateway_host, + gateway_port + ); + match tunnel.start(gateway_host, gateway_port).await { + Ok(url) => { + tracing::info!("Tunnel started: {}", url); + config.tunnel.public_url = Some(url); + (config, Some(tunnel)) + } + Err(e) => { + tracing::error!("Failed to start tunnel: {}", e); + (config, None) + } + } + } + Ok(None) => (config, None), + Err(e) => { + tracing::error!("Failed to create tunnel: {}", e); + (config, None) + } + } +} + // ── Tests ──────────────────────────────────────────────────────── #[cfg(test)] diff --git a/src/worker/mod.rs b/src/worker/mod.rs index 88dd7c56..dce75b3d 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -33,3 +33,61 @@ pub use api::WorkerHttpClient; pub use claude_bridge::ClaudeBridgeRuntime; pub use proxy_llm::ProxyLlmProvider; pub use runtime::WorkerRuntime; + +/// Run the Worker subcommand (inside Docker containers). +pub async fn run_worker( + job_id: uuid::Uuid, + orchestrator_url: &str, + max_iterations: u32, +) -> anyhow::Result<()> { + tracing::info!( + "Starting worker for job {} (orchestrator: {})", + job_id, + orchestrator_url + ); + + let config = runtime::WorkerConfig { + job_id, + orchestrator_url: orchestrator_url.to_string(), + max_iterations, + timeout: std::time::Duration::from_secs(600), + }; + + let rt = + WorkerRuntime::new(config).map_err(|e| anyhow::anyhow!("Worker init failed: {}", e))?; + + rt.run() + .await + .map_err(|e| anyhow::anyhow!("Worker failed: {}", e)) +} + +/// Run the Claude Code bridge subcommand (inside Docker containers). +pub async fn run_claude_bridge( + job_id: uuid::Uuid, + orchestrator_url: &str, + max_turns: u32, + model: &str, +) -> anyhow::Result<()> { + tracing::info!( + "Starting Claude Code bridge for job {} (orchestrator: {}, model: {})", + job_id, + orchestrator_url, + model + ); + + let config = claude_bridge::ClaudeBridgeConfig { + job_id, + orchestrator_url: orchestrator_url.to_string(), + max_turns, + model: model.to_string(), + timeout: std::time::Duration::from_secs(1800), + allowed_tools: crate::config::ClaudeCodeConfig::from_env().allowed_tools, + }; + + let rt = ClaudeBridgeRuntime::new(config) + .map_err(|e| anyhow::anyhow!("Claude bridge init failed: {}", e))?; + + rt.run() + .await + .map_err(|e| anyhow::anyhow!("Claude bridge failed: {}", e)) +} diff --git a/tests/module_init_integration.rs b/tests/module_init_integration.rs new file mode 100644 index 00000000..c75ccc6f --- /dev/null +++ b/tests/module_init_integration.rs @@ -0,0 +1,237 @@ +//! Integration test for module-owned initialization factories. +//! +//! Verifies that the refactored factory functions in `db`, `secrets`, +//! `orchestrator`, and `extensions` modules wire up correctly end-to-end, +//! ensuring nothing was lost when initialization logic was moved out of +//! `main.rs` and `app.rs` into owning modules. + +use std::sync::Arc; + +use ironclaw::db::DatabaseHandles; +use ironclaw::secrets::{CreateSecretParams, SecretsCrypto, SecretsStore}; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Build a libsql DatabaseConfig pointing at a temp file. +#[cfg(feature = "libsql")] +fn libsql_config(path: &std::path::Path) -> ironclaw::config::DatabaseConfig { + ironclaw::config::DatabaseConfig { + backend: ironclaw::config::DatabaseBackend::LibSql, + url: secrecy::SecretString::from(String::new()), + pool_size: 1, + ssl_mode: ironclaw::config::SslMode::Prefer, + libsql_path: Some(path.to_path_buf()), + libsql_url: None, + libsql_auth_token: None, + } +} + +/// Build a master-key crypto instance for tests. +fn test_crypto() -> Arc { + let key = secrecy::SecretString::from(ironclaw::secrets::keychain::generate_master_key_hex()); + Arc::new(SecretsCrypto::new(key).expect("test crypto")) +} + +// --------------------------------------------------------------------------- +// connect_with_handles: returns Database + populated handles +// --------------------------------------------------------------------------- + +#[cfg(feature = "libsql")] +#[tokio::test] +async fn connect_with_handles_returns_db_and_libsql_handle() { + let dir = tempfile::tempdir().expect("tempdir"); + let db_path = dir.path().join("test.db"); + let config = libsql_config(&db_path); + + let (db, handles) = ironclaw::db::connect_with_handles(&config) + .await + .expect("connect_with_handles"); + + // Database trait object works — run a trivial operation. + db.run_migrations().await.expect("migrations"); + + // Handle is populated. + assert!( + handles.libsql_db.is_some(), + "libsql handle should be Some after connect_with_handles" + ); +} + +// --------------------------------------------------------------------------- +// connect_from_config delegates to connect_with_handles +// --------------------------------------------------------------------------- + +#[cfg(feature = "libsql")] +#[tokio::test] +async fn connect_from_config_produces_working_db() { + let dir = tempfile::tempdir().expect("tempdir"); + let db_path = dir.path().join("test.db"); + let config = libsql_config(&db_path); + + // connect_from_config delegates to connect_with_handles internally. + let db = ironclaw::db::connect_from_config(&config) + .await + .expect("connect_from_config"); + + // Verify usable — migrations should be idempotent. + db.run_migrations().await.expect("migrations"); +} + +// --------------------------------------------------------------------------- +// secrets::create_secrets_store from DatabaseHandles +// --------------------------------------------------------------------------- + +#[cfg(feature = "libsql")] +#[tokio::test] +async fn secrets_store_from_handles_round_trips() { + let dir = tempfile::tempdir().expect("tempdir"); + let db_path = dir.path().join("test.db"); + let config = libsql_config(&db_path); + + let (_db, handles) = ironclaw::db::connect_with_handles(&config) + .await + .expect("connect"); + + let crypto = test_crypto(); + let store = ironclaw::secrets::create_secrets_store(crypto, &handles) + .expect("create_secrets_store should return Some for libsql"); + + // Round-trip a secret to prove the store works. + store + .create("test", CreateSecretParams::new("test_key", "test_value")) + .await + .expect("create secret"); + + let decrypted = store + .get_decrypted("test", "test_key") + .await + .expect("get_decrypted"); + assert_eq!(decrypted.expose(), "test_value"); +} + +// --------------------------------------------------------------------------- +// db::create_secrets_store (standalone CLI factory) +// --------------------------------------------------------------------------- + +#[cfg(feature = "libsql")] +#[tokio::test] +async fn db_create_secrets_store_standalone_round_trips() { + let dir = tempfile::tempdir().expect("tempdir"); + let db_path = dir.path().join("test.db"); + let config = libsql_config(&db_path); + let crypto = test_crypto(); + + let store = ironclaw::db::create_secrets_store(&config, crypto) + .await + .expect("db::create_secrets_store"); + + store + .create( + "test", + CreateSecretParams::new("standalone_key", "standalone_value"), + ) + .await + .expect("create secret"); + + let decrypted = store + .get_decrypted("test", "standalone_key") + .await + .expect("get_decrypted"); + assert_eq!(decrypted.expose(), "standalone_value"); +} + +// --------------------------------------------------------------------------- +// Both secrets factories produce equivalent stores +// --------------------------------------------------------------------------- + +#[cfg(feature = "libsql")] +#[tokio::test] +async fn both_secrets_factories_produce_compatible_stores() { + let dir = tempfile::tempdir().expect("tempdir"); + let db_path = dir.path().join("test.db"); + let config = libsql_config(&db_path); + let crypto = test_crypto(); + + // Factory 1: connect_with_handles + secrets::create_secrets_store + let (_db, handles) = ironclaw::db::connect_with_handles(&config) + .await + .expect("connect"); + let store_a = ironclaw::secrets::create_secrets_store(Arc::clone(&crypto), &handles) + .expect("store from handles"); + + // Factory 2: db::create_secrets_store (standalone) + let store_b = ironclaw::db::create_secrets_store(&config, crypto) + .await + .expect("standalone store"); + + // Write with factory 1, read with factory 2. + store_a + .create( + "test", + CreateSecretParams::new("cross_factory", "shared_secret"), + ) + .await + .expect("create via store_a"); + + let decrypted = store_b + .get_decrypted("test", "cross_factory") + .await + .expect("read via store_b"); + assert_eq!(decrypted.expose(), "shared_secret"); +} + +// --------------------------------------------------------------------------- +// ExtensionManager constructs with McpProcessManager +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn extension_manager_with_process_manager_constructs() { + use ironclaw::extensions::ExtensionManager; + use ironclaw::secrets::InMemorySecretsStore; + use ironclaw::tools::ToolRegistry; + use ironclaw::tools::mcp::McpProcessManager; + use ironclaw::tools::mcp::McpSessionManager; + + let crypto = test_crypto(); + let secrets: Arc = Arc::new(InMemorySecretsStore::new(crypto)); + let tools = Arc::new(ToolRegistry::new()); + let tools_dir = tempfile::tempdir().expect("tools_dir"); + let channels_dir = tempfile::tempdir().expect("channels_dir"); + + let manager = ExtensionManager::new( + Arc::new(McpSessionManager::new()), + Arc::new(McpProcessManager::new()), + secrets, + tools, + None, + None, + tools_dir.path().to_path_buf(), + channels_dir.path().to_path_buf(), + None, + "test".to_string(), + None, + Vec::new(), + ); + + // Verify the manager is functional — list returns Ok. + let result = manager.list(None, false).await; + assert!(result.is_ok(), "list should succeed on empty manager"); + assert!(result.unwrap().is_empty()); +} + +// --------------------------------------------------------------------------- +// DatabaseHandles: default is empty +// --------------------------------------------------------------------------- + +#[test] +fn database_handles_default_is_empty() { + let handles = DatabaseHandles::default(); + + #[cfg(feature = "postgres")] + assert!(handles.pg_pool.is_none()); + + #[cfg(feature = "libsql")] + assert!(handles.libsql_db.is_none()); +}