From 991aea85ee2036b22dea71aa773b5e7b28a77f08 Mon Sep 17 00:00:00 2001 From: Sky Moore Date: Sat, 21 Mar 2026 19:08:47 +0000 Subject: [PATCH 1/4] feat: use rmcp for mcp protocol instead of hand rolled Replace the custom JSON-RPC + stdio/SSE transport layer with the rmcp SDK (crate 'rmcp'). This gives us spec-compliant Streamable-HTTP transport, automatic Mcp-Session-Id tracking, SSE stream parsing, and content-type negotiation out of the box while deleting ~300 lines of hand-rolled plumbing. Key changes: - Add rmcp dependency with transport feature - Replace McpTransportHandle enum with rmcp RunningService - Replace manual JSON-RPC send_request/send_notification with rmcp client calls - Add custom HTTP headers support for authenticated remote MCP servers - Simplify tool discovery and invocation through rmcp's typed API --- Cargo.lock | 180 +++++- Cargo.toml | 3 + crates/openfang-api/src/routes.rs | 6 + crates/openfang-cli/src/main.rs | 3 +- crates/openfang-extensions/src/bundled.rs | 3 + crates/openfang-extensions/src/lib.rs | 3 + crates/openfang-extensions/src/registry.rs | 4 + crates/openfang-kernel/src/kernel.rs | 17 +- crates/openfang-runtime/Cargo.toml | 2 + crates/openfang-runtime/src/mcp.rs | 697 ++++++--------------- crates/openfang-types/src/config.rs | 11 + 11 files changed, 424 insertions(+), 505 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 763da87c..7c4172bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -384,6 +384,28 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "aws-lc-rs" +version = "1.16.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a054912289d18629dc78375ba2c3726a3afe3ff71b4edba9dedfca0e3446d1fc" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.39.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa7e52a4c5c547c741610a2c6f123f3881e409b714cd27e6798ef020c514f0a" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", +] + [[package]] name = "axum" version = "0.8.8" @@ -860,6 +882,15 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" +[[package]] +name = "cmake" +version = "0.1.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75443c44cd6b379beb8c5b45d85d0773baf31cce901fe7bb252f4eff3008ef7d" +dependencies = [ + "cc", +] + [[package]] name = "cobs" version = "0.3.0" @@ -1958,6 +1989,12 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "futf" version = "0.1.5" @@ -3608,6 +3645,18 @@ version = "1.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086" +[[package]] +name = "nix" +version = "0.31.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d6d0705320c1e6ba1d912b5e37cf18071b6c2e9b7fa8215a1e8a7651966f5d3" +dependencies = [ + "bitflags 2.11.0", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "no-std-compat" version = "0.4.1" @@ -4149,11 +4198,13 @@ dependencies = [ "dashmap", "futures", "hex", + "http", "openfang-memory", "openfang-skills", "openfang-types", "regex-lite", "reqwest 0.12.28", + "rmcp", "rusqlite", "serde", "serde_json", @@ -4896,6 +4947,20 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "process-wrap" +version = "9.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e842efad9119158434d193c6682e2ebee4b44d6ad801d7b349623b3f57cdf55" +dependencies = [ + "futures", + "indexmap 2.13.0", + "nix", + "tokio", + "tracing", + "windows 0.62.2", +] + [[package]] name = "prost" version = "0.13.5" @@ -5017,6 +5082,7 @@ version = "0.11.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098" dependencies = [ + "aws-lc-rs", "bytes", "getrandom 0.3.4", "lru-slab", @@ -5420,6 +5486,7 @@ dependencies = [ "log", "percent-encoding", "pin-project-lite", + "quinn", "rustls", "rustls-pki-types", "rustls-platform-verifier", @@ -5477,6 +5544,29 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rmcp" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba6b9d2f0efe2258b23767f1f9e0054cfbcac9c2d6f81a031214143096d7864f" +dependencies = [ + "async-trait", + "chrono", + "futures", + "http", + "pin-project-lite", + "process-wrap", + "reqwest 0.13.2", + "serde", + "serde_json", + "sse-stream", + "thiserror 2.0.18", + "tokio", + "tokio-stream", + "tokio-util", + "tracing", +] + [[package]] name = "rmp" version = "0.8.15" @@ -5573,6 +5663,7 @@ version = "0.23.37" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4" dependencies = [ + "aws-lc-rs", "log", "once_cell", "ring", @@ -5637,6 +5728,7 @@ version = "0.103.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d7df23109aa6c1567d1c575b9952556388da57401e4ace1d15f79eedad0d8f53" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", @@ -6243,6 +6335,19 @@ dependencies = [ "der", ] +[[package]] +name = "sse-stream" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb4dc4d33c68ec1f27d386b5610a351922656e1fdf5c05bbaad930cd1519479a" +dependencies = [ + "bytes", + "futures-util", + "http-body", + "http-body-util", + "pin-project-lite", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -6449,7 +6554,7 @@ dependencies = [ "tao-macros", "unicode-segmentation", "url", - "windows", + "windows 0.61.3", "windows-core 0.61.2", "windows-version", "x11-dl", @@ -6538,7 +6643,7 @@ dependencies = [ "webkit2gtk", "webview2-com", "window-vibrancy", - "windows", + "windows 0.61.3", ] [[package]] @@ -6800,7 +6905,7 @@ dependencies = [ "url", "webkit2gtk", "webview2-com", - "windows", + "windows 0.61.3", ] [[package]] @@ -6825,7 +6930,7 @@ dependencies = [ "url", "webkit2gtk", "webview2-com", - "windows", + "windows 0.61.3", "wry", ] @@ -6886,7 +6991,7 @@ checksum = "0b1e66e07de489fe43a46678dd0b8df65e0c973909df1b60ba33874e297ba9b9" dependencies = [ "quick-xml 0.37.5", "thiserror 2.0.18", - "windows", + "windows 0.61.3", "windows-version", ] @@ -8299,7 +8404,7 @@ checksum = "7130243a7a5b33c54a444e54842e6a9e133de08b5ad7b5861cd8ed9a6a5bc96a" dependencies = [ "webview2-com-macros", "webview2-com-sys", - "windows", + "windows 0.61.3", "windows-core 0.61.2", "windows-implement", "windows-interface", @@ -8323,7 +8428,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "381336cfffd772377d291702245447a5251a2ffa5bad679c99e61bc48bacbf9c" dependencies = [ "thiserror 2.0.18", - "windows", + "windows 0.61.3", "windows-core 0.61.2", ] @@ -8399,11 +8504,23 @@ version = "0.61.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893" dependencies = [ - "windows-collections", + "windows-collections 0.2.0", "windows-core 0.61.2", - "windows-future", + "windows-future 0.2.1", "windows-link 0.1.3", - "windows-numerics", + "windows-numerics 0.2.0", +] + +[[package]] +name = "windows" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580" +dependencies = [ + "windows-collections 0.3.2", + "windows-core 0.62.2", + "windows-future 0.3.2", + "windows-numerics 0.3.1", ] [[package]] @@ -8415,6 +8532,15 @@ dependencies = [ "windows-core 0.61.2", ] +[[package]] +name = "windows-collections" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610" +dependencies = [ + "windows-core 0.62.2", +] + [[package]] name = "windows-core" version = "0.61.2" @@ -8449,7 +8575,18 @@ checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" dependencies = [ "windows-core 0.61.2", "windows-link 0.1.3", - "windows-threading", + "windows-threading 0.1.0", +] + +[[package]] +name = "windows-future" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb" +dependencies = [ + "windows-core 0.62.2", + "windows-link 0.2.1", + "windows-threading 0.2.1", ] [[package]] @@ -8496,6 +8633,16 @@ dependencies = [ "windows-link 0.1.3", ] +[[package]] +name = "windows-numerics" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26" +dependencies = [ + "windows-core 0.62.2", + "windows-link 0.2.1", +] + [[package]] name = "windows-result" version = "0.3.4" @@ -8634,6 +8781,15 @@ dependencies = [ "windows-link 0.1.3", ] +[[package]] +name = "windows-threading" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37" +dependencies = [ + "windows-link 0.2.1", +] + [[package]] name = "windows-version" version = "0.1.7" @@ -8977,7 +9133,7 @@ dependencies = [ "webkit2gtk", "webkit2gtk-sys", "webview2-com", - "windows", + "windows 0.61.3", "windows-core 0.61.2", "windows-version", "x11-dl", diff --git a/Cargo.toml b/Cargo.toml index f25f99f7..70d3df5f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -130,6 +130,9 @@ html-escape = "0.2" # Lightweight regex regex-lite = "0.1" +# MCP SDK (official Rust implementation) +rmcp = { version = "1.2", default-features = false, features = ["client", "transport-child-process", "transport-streamable-http-client-reqwest", "reqwest"] } + # Socket options (SO_REUSEADDR) socket2 = "0.5" diff --git a/crates/openfang-api/src/routes.rs b/crates/openfang-api/src/routes.rs index c9f68b1f..03172769 100644 --- a/crates/openfang-api/src/routes.rs +++ b/crates/openfang-api/src/routes.rs @@ -4871,6 +4871,12 @@ pub async fn list_mcp_servers(State(state): State>) -> impl IntoRe "url": url, }) } + openfang_types::config::McpTransportEntry::Http { url } => { + serde_json::json!({ + "type": "http", + "url": url, + }) + } }; serde_json::json!({ "name": s.name, diff --git a/crates/openfang-cli/src/main.rs b/crates/openfang-cli/src/main.rs index 40148e2d..9dc9c1c7 100644 --- a/crates/openfang-cli/src/main.rs +++ b/crates/openfang-cli/src/main.rs @@ -2596,7 +2596,8 @@ decay_rate = 0.05 checks.push(serde_json::json!({"check": "mcp_server_config", "status": "warn", "name": server.name})); } } - openfang_types::config::McpTransportEntry::Sse { url } => { + openfang_types::config::McpTransportEntry::Sse { url } + | openfang_types::config::McpTransportEntry::Http { url } => { if url.is_empty() { if !json { ui::check_warn(&format!( diff --git a/crates/openfang-extensions/src/bundled.rs b/crates/openfang-extensions/src/bundled.rs index d4928ae5..75b955cf 100644 --- a/crates/openfang-extensions/src/bundled.rs +++ b/crates/openfang-extensions/src/bundled.rs @@ -146,6 +146,9 @@ mod tests { crate::McpTransportTemplate::Sse { .. } => { panic!("{} unexpectedly uses SSE transport", id); } + crate::McpTransportTemplate::Http { .. } => { + panic!("{} unexpectedly uses HTTP transport", id); + } } } } diff --git a/crates/openfang-extensions/src/lib.rs b/crates/openfang-extensions/src/lib.rs index cf6aca80..929dc6ff 100644 --- a/crates/openfang-extensions/src/lib.rs +++ b/crates/openfang-extensions/src/lib.rs @@ -88,6 +88,9 @@ pub enum McpTransportTemplate { Sse { url: String, }, + Http { + url: String, + }, } /// An environment variable required by an integration. diff --git a/crates/openfang-extensions/src/registry.rs b/crates/openfang-extensions/src/registry.rs index cd0fd7c4..282ad3dd 100644 --- a/crates/openfang-extensions/src/registry.rs +++ b/crates/openfang-extensions/src/registry.rs @@ -191,6 +191,9 @@ impl IntegrationRegistry { crate::McpTransportTemplate::Sse { url } => { McpTransportEntry::Sse { url: url.clone() } } + crate::McpTransportTemplate::Http { url } => { + McpTransportEntry::Http { url: url.clone() } + } }; let env: Vec = template .required_env @@ -202,6 +205,7 @@ impl IntegrationRegistry { transport, timeout_secs: 30, env, + headers: Vec::new(), }) }) .collect() diff --git a/crates/openfang-kernel/src/kernel.rs b/crates/openfang-kernel/src/kernel.rs index a3d1c084..42141f0d 100644 --- a/crates/openfang-kernel/src/kernel.rs +++ b/crates/openfang-kernel/src/kernel.rs @@ -4703,6 +4703,7 @@ impl OpenFangKernel { args: args.clone(), }, McpTransportEntry::Sse { url } => McpTransport::Sse { url: url.clone() }, + McpTransportEntry::Http { url } => McpTransport::Http { url: url.clone() }, }; // Resolve env vars from vault/dotenv before passing to MCP subprocess. @@ -4721,6 +4722,7 @@ impl OpenFangKernel { transport, timeout_secs: server_config.timeout_secs, env: server_config.env.clone(), + headers: server_config.headers.clone(), }; match McpConnection::connect(mcp_config).await { @@ -4822,6 +4824,7 @@ impl OpenFangKernel { args: args.clone(), }, McpTransportEntry::Sse { url } => McpTransport::Sse { url: url.clone() }, + McpTransportEntry::Http { url } => McpTransport::Http { url: url.clone() }, }; let mcp_config = McpServerConfig { @@ -4829,6 +4832,7 @@ impl OpenFangKernel { transport, timeout_secs: server_config.timeout_secs, env: server_config.env.clone(), + headers: server_config.headers.clone(), }; self.extension_health.register(&server_config.name); @@ -4940,6 +4944,7 @@ impl OpenFangKernel { args: args.clone(), }, McpTransportEntry::Sse { url } => McpTransport::Sse { url: url.clone() }, + McpTransportEntry::Http { url } => McpTransport::Http { url: url.clone() }, }; let mcp_config = McpServerConfig { @@ -4947,6 +4952,7 @@ impl OpenFangKernel { transport, timeout_secs: server_config.timeout_secs, env: server_config.env.clone(), + headers: server_config.headers.clone(), }; match McpConnection::connect(mcp_config).await { @@ -5032,7 +5038,16 @@ impl OpenFangKernel { agent_id: AgentId, skill_snapshot: Option<&openfang_skills::registry::SkillRegistry>, ) -> Vec { - let all_builtins = builtin_tool_definitions(); + let all_builtins = if self.config.browser.enabled { + builtin_tool_definitions() + } else { + // When built-in browser is disabled (replaced by an external + // browser MCP server such as CamoFox), filter out browser_* tools. + builtin_tool_definitions() + .into_iter() + .filter(|t| !t.name.starts_with("browser_")) + .collect() + }; // Look up agent entry for profile, skill/MCP allowlists, and declared tools let entry = self.registry.get(agent_id); diff --git a/crates/openfang-runtime/Cargo.toml b/crates/openfang-runtime/Cargo.toml index 065cde42..aff9a802 100644 --- a/crates/openfang-runtime/Cargo.toml +++ b/crates/openfang-runtime/Cargo.toml @@ -30,6 +30,8 @@ zeroize = { workspace = true } dashmap = { workspace = true } regex-lite = { workspace = true } rusqlite = { workspace = true } +rmcp = { workspace = true } +http = "1" tokio-tungstenite = "0.24" shlex = "1" diff --git a/crates/openfang-runtime/src/mcp.rs b/crates/openfang-runtime/src/mcp.rs index 89442423..c15bb86a 100644 --- a/crates/openfang-runtime/src/mcp.rs +++ b/crates/openfang-runtime/src/mcp.rs @@ -1,16 +1,20 @@ //! MCP (Model Context Protocol) client — connect to external MCP servers. //! -//! MCP uses JSON-RPC 2.0 over stdio or HTTP+SSE. This module lets OpenFang -//! agents use tools from any MCP server (100+ available: GitHub, filesystem, -//! databases, APIs, etc.). +//! Uses the official `rmcp` SDK for protocol handling. Supports: +//! - **stdio**: subprocess with JSON-RPC over stdin/stdout +//! - **sse**: deprecated HTTP+SSE transport (protocol version 2024-11-05) +//! - **http**: Streamable HTTP transport (protocol version 2025-03-26+) //! //! All MCP tools are namespaced with `mcp_{server}_{tool}` to prevent collisions. +use http::{HeaderName, HeaderValue}; use openfang_types::tool::ToolDefinition; +use rmcp::model::{CallToolRequestParams, ClientCapabilities, ClientInfo, Implementation}; +use rmcp::service::RunningService; +use rmcp::{RoleClient, ServiceExt}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; -use std::process::Stdio; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use std::sync::Arc; use tracing::{debug, info}; // --------------------------------------------------------------------------- @@ -30,6 +34,12 @@ pub struct McpServerConfig { /// Environment variables to pass through to the subprocess (sandboxed). #[serde(default)] pub env: Vec, + /// Extra HTTP headers to send with every SSE / Streamable-HTTP request. + /// Each entry is `"Header-Name: value"`. Useful for authentication + /// (`Authorization: Bearer `), API keys (`X-Api-Key: ...`), + /// or any custom headers required by a remote MCP server. + #[serde(default)] + pub headers: Vec, } fn default_timeout() -> u64 { @@ -46,8 +56,14 @@ pub enum McpTransport { #[serde(default)] args: Vec, }, - /// HTTP Server-Sent Events. + /// Deprecated HTTP+SSE transport (protocol version 2024-11-05). + /// Uses POST for sending and SSE for receiving. Sse { url: String }, + /// Streamable HTTP transport (MCP 2025-03-26+). + /// Single endpoint, client MUST send Accept: application/json, text/event-stream. + /// Server responds with either JSON or SSE stream. + /// Supports Mcp-Session-Id for session management. + Http { url: String }, } // --------------------------------------------------------------------------- @@ -64,59 +80,9 @@ pub struct McpConnection { /// Needed because `normalize_name` replaces hyphens with underscores, /// but the server expects the original name (e.g. "list-connections"). original_names: HashMap, - /// Transport handle for sending requests. - transport: McpTransportHandle, - /// Next JSON-RPC request ID. - next_id: u64, -} - -/// Transport handle — abstraction over stdio subprocess or HTTP. -enum McpTransportHandle { - Stdio { - child: Box, - stdin: tokio::process::ChildStdin, - stdout: BufReader, - }, - Sse { - client: reqwest::Client, - url: String, - }, -} - -/// JSON-RPC 2.0 request. -#[derive(Serialize)] -struct JsonRpcRequest { - jsonrpc: &'static str, - id: u64, - method: String, - #[serde(skip_serializing_if = "Option::is_none")] - params: Option, -} - -/// JSON-RPC 2.0 response. -#[derive(Deserialize)] -struct JsonRpcResponse { - #[allow(dead_code)] - jsonrpc: String, - #[allow(dead_code)] - id: Option, - result: Option, - error: Option, -} - -/// JSON-RPC 2.0 error object. -#[derive(Debug, Deserialize)] -pub struct JsonRpcError { - pub code: i64, - pub message: String, - #[allow(dead_code)] - pub data: Option, -} - -impl std::fmt::Display for JsonRpcError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "JSON-RPC error {}: {}", self.code, self.message) - } + /// The rmcp client handle — type-erased because the concrete type + /// depends on which transport was used (stdio vs HTTP). + client: RunningService, } // --------------------------------------------------------------------------- @@ -126,13 +92,17 @@ impl std::fmt::Display for JsonRpcError { impl McpConnection { /// Connect to an MCP server, perform handshake, and discover tools. pub async fn connect(config: McpServerConfig) -> Result { - let transport = match &config.transport { + let client_info = ClientInfo::new( + ClientCapabilities::default(), + Implementation::new("openfang", env!("CARGO_PKG_VERSION")), + ); + + let client = match &config.transport { McpTransport::Stdio { command, args } => { - Self::connect_stdio(command, args, &config.env).await? + Self::connect_stdio(command, args, &config.env, client_info).await? } - McpTransport::Sse { url } => { - // SSRF check: reject private/localhost URLs unless explicitly configured - Self::connect_sse(url).await? + McpTransport::Sse { url } | McpTransport::Http { url } => { + Self::connect_http(url, &config.headers, client_info).await? } }; @@ -140,13 +110,9 @@ impl McpConnection { config, tools: Vec::new(), original_names: HashMap::new(), - transport, - next_id: 1, + client, }; - // Initialize handshake - conn.initialize().await?; - // Discover tools conn.discover_tools().await?; @@ -159,76 +125,34 @@ impl McpConnection { Ok(conn) } - /// Send the MCP `initialize` handshake. - async fn initialize(&mut self) -> Result<(), String> { - let params = serde_json::json!({ - "protocolVersion": "2024-11-05", - "capabilities": {}, - "clientInfo": { - "name": "openfang", - "version": env!("CARGO_PKG_VERSION") - } - }); - - let response = self.send_request("initialize", Some(params)).await?; - - if let Some(result) = response { - debug!( - server = %self.config.name, - server_info = %result, - "MCP initialize response" - ); - } - - // Send initialized notification (no response expected) - self.send_notification("notifications/initialized", None) - .await?; - - Ok(()) - } - /// Discover available tools via `tools/list`. async fn discover_tools(&mut self) -> Result<(), String> { - let response = self.send_request("tools/list", None).await?; + let tools = self + .client + .list_all_tools() + .await + .map_err(|e| format!("Failed to list MCP tools: {e}"))?; - if let Some(result) = response { - if let Some(tools_array) = result.get("tools").and_then(|t| t.as_array()) { - let server_name = &self.config.name; - for tool in tools_array { - let raw_name = tool["name"].as_str().unwrap_or("unnamed"); - let description = tool["description"].as_str().unwrap_or(""); - let input_schema = tool - .get("inputSchema") - .cloned() - .and_then(|v| { - // Ensure input_schema is a JSON object. MCP servers may - // return it as a string, null, or omit it entirely. - match &v { - serde_json::Value::Object(_) => Some(v), - serde_json::Value::String(s) => { - serde_json::from_str::(s) - .ok() - .filter(|p| p.is_object()) - } - _ => None, - } - }) - .unwrap_or(serde_json::json!({"type": "object"})); + let server_name = &self.config.name; + for tool in &tools { + let raw_name = &tool.name; + let description = tool.description.as_deref().unwrap_or(""); - // Namespace: mcp_{server}_{tool} - let namespaced = format_mcp_tool_name(server_name, raw_name); + let input_schema = serde_json::to_value(&tool.input_schema) + .unwrap_or(serde_json::json!({"type": "object"})); - // Store original name so we can send it back to the server - self.original_names - .insert(namespaced.clone(), raw_name.to_string()); + // Namespace: mcp_{server}_{tool} + let namespaced = format_mcp_tool_name(server_name, raw_name); - self.tools.push(ToolDefinition { - name: namespaced, - description: format!("[MCP:{server_name}] {description}"), - input_schema, - }); - } - } + // Store original name so we can send it back to the server + self.original_names + .insert(namespaced.clone(), raw_name.to_string()); + + self.tools.push(ToolDefinition { + name: namespaced, + description: format!("[MCP:{server_name}] {description}"), + input_schema, + }); } Ok(()) @@ -243,40 +167,41 @@ impl McpConnection { arguments: &serde_json::Value, ) -> Result { // Look up the original tool name from the server (preserves hyphens etc.) - let raw_name = self + let raw_name: String = self .original_names .get(name) - .map(|s| s.as_str()) - .or_else(|| strip_mcp_prefix(&self.config.name, name)) - .unwrap_or(name); + .cloned() + .or_else(|| strip_mcp_prefix(&self.config.name, name).map(|s| s.to_string())) + .unwrap_or_else(|| name.to_string()); - let params = serde_json::json!({ - "name": raw_name, - "arguments": arguments, - }); + let args = arguments + .as_object() + .cloned() + .unwrap_or_default(); - let response = self.send_request("tools/call", Some(params)).await?; + debug!(tool = %raw_name, server = %self.config.name, "MCP tool call"); - match response { - Some(result) => { - // Extract text content from the response - if let Some(content) = result.get("content").and_then(|c| c.as_array()) { - let texts: Vec<&str> = content - .iter() - .filter_map(|item| { - if item["type"].as_str() == Some("text") { - item["text"].as_str() - } else { - None - } - }) - .collect(); - Ok(texts.join("\n")) - } else { - Ok(result.to_string()) - } - } - None => Err("No result from MCP tools/call".to_string()), + let params = CallToolRequestParams::new(raw_name).with_arguments(args); + + let result = self + .client + .call_tool(params) + .await + .map_err(|e| format!("MCP tool call failed: {e}"))?; + + // Extract text content from the response. + // `Content` is `Annotated` which Derefs to `RawContent`. + let texts: Vec<&str> = result + .content + .iter() + .filter_map(|item| item.as_text().map(|tc| tc.text.as_str())) + .collect(); + + if texts.is_empty() { + // Fallback: serialize the entire result + Ok(serde_json::to_string(&result).unwrap_or_default()) + } else { + Ok(texts.join("\n")) } } @@ -290,294 +215,125 @@ impl McpConnection { &self.config.name } - // --- Transport helpers --- - - async fn send_request( - &mut self, - method: &str, - params: Option, - ) -> Result, String> { - let id = self.next_id; - self.next_id += 1; - - let request = JsonRpcRequest { - jsonrpc: "2.0", - id, - method: method.to_string(), - params, - }; - - let request_json = serde_json::to_string(&request) - .map_err(|e| format!("Failed to serialize request: {e}"))?; - - debug!(method, id, "MCP request"); - - match &mut self.transport { - McpTransportHandle::Stdio { stdin, stdout, .. } => { - // Write request + newline - stdin - .write_all(request_json.as_bytes()) - .await - .map_err(|e| format!("Failed to write to MCP stdin: {e}"))?; - stdin - .write_all(b"\n") - .await - .map_err(|e| format!("Failed to write newline: {e}"))?; - stdin - .flush() - .await - .map_err(|e| format!("Failed to flush stdin: {e}"))?; - - // Read response lines until we find one matching our request ID. - // MCP servers may send notifications or log lines before the - // actual response. - let timeout_dur = tokio::time::Duration::from_secs(self.config.timeout_secs); - let deadline = tokio::time::Instant::now() + timeout_dur; - - loop { - let mut line = String::new(); - let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); - if remaining.is_zero() { - return Err("MCP request timed out".to_string()); - } - match tokio::time::timeout(remaining, stdout.read_line(&mut line)).await { - Ok(Ok(0)) => return Err("MCP server closed connection".to_string()), - Ok(Ok(_)) => {} - Ok(Err(e)) => return Err(format!("Failed to read MCP response: {e}")), - Err(_) => return Err("MCP request timed out".to_string()), - } - - let trimmed = line.trim(); - if trimmed.is_empty() { - continue; - } - - // Try to parse as JSON-RPC response - let parsed: Result = serde_json::from_str(trimmed); - match parsed { - Ok(response) if response.id == Some(id) => { - if let Some(err) = response.error { - return Err(format!("{err}")); - } - return Ok(response.result); - } - Ok(_other) => { - // Response for a different ID or notification — skip - debug!("MCP: skipping non-matching response line"); - continue; - } - Err(_) => { - // Not valid JSON-RPC — skip (could be a log line) - debug!("MCP: skipping non-JSON line from server"); - continue; - } - } - } - } - McpTransportHandle::Sse { client, url } => { - let response = client - .post(url.as_str()) - .json(&request) - .timeout(std::time::Duration::from_secs(self.config.timeout_secs)) - .send() - .await - .map_err(|e| format!("MCP SSE request failed: {e}"))?; - - if !response.status().is_success() { - return Err(format!("MCP SSE returned {}", response.status())); - } - - let body = response - .text() - .await - .map_err(|e| format!("Failed to read SSE response: {e}"))?; - - // Handle Streamable HTTP MCP responses that use SSE framing - // (e.g. "event: message\ndata: {...}\n\n"). Extract the JSON - // from the last `data:` line if the body looks like SSE. - let json_body = if body.trim_start().starts_with("event:") || body.trim_start().starts_with("data:") { - body.lines() - .filter_map(|line| line.strip_prefix("data: ").or_else(|| line.strip_prefix("data:"))) - .filter(|s| !s.is_empty()) - .last() - .unwrap_or(&body) - .to_string() - } else { - body - }; - - let rpc_response: JsonRpcResponse = serde_json::from_str(&json_body) - .map_err(|e| format!("Invalid MCP SSE JSON-RPC response: {e}"))?; - - if let Some(err) = rpc_response.error { - return Err(format!("{err}")); - } - - Ok(rpc_response.result) - } - } - } - - async fn send_notification( - &mut self, - method: &str, - params: Option, - ) -> Result<(), String> { - let notification = serde_json::json!({ - "jsonrpc": "2.0", - "method": method, - "params": params.unwrap_or(serde_json::json!({})), - }); - - let json = serde_json::to_string(¬ification) - .map_err(|e| format!("Failed to serialize notification: {e}"))?; - - match &mut self.transport { - McpTransportHandle::Stdio { stdin, .. } => { - stdin - .write_all(json.as_bytes()) - .await - .map_err(|e| format!("Write notification: {e}"))?; - stdin - .write_all(b"\n") - .await - .map_err(|e| format!("Write newline: {e}"))?; - stdin.flush().await.map_err(|e| format!("Flush: {e}"))?; - } - McpTransportHandle::Sse { client, url } => { - let _ = client.post(url.as_str()).json(¬ification).send().await; - } - } - - Ok(()) - } + // -- Transport constructors ----------------------------------------------- + /// Connect using stdio transport (subprocess). async fn connect_stdio( command: &str, args: &[String], env_whitelist: &[String], - ) -> Result { + client_info: ClientInfo, + ) -> Result, String> { + use rmcp::transport::{ConfigureCommandExt, TokioChildProcess}; + use tokio::process::Command; + // Validate command path (no path traversal) if command.contains("..") { return Err("MCP command path contains '..': rejected".to_string()); } - // On Windows, npm/npx install as .cmd batch wrappers. Detect and adapt. - let resolved_command: String = if cfg!(windows) { - // If the user already specified .cmd/.bat, use as-is - if command.ends_with(".cmd") || command.ends_with(".bat") { - command.to_string() - } else { - // Check if the .cmd variant exists on PATH - let cmd_variant = format!("{command}.cmd"); - let has_cmd = std::env::var("PATH") - .unwrap_or_default() - .split(';') - .any(|dir| std::path::Path::new(dir).join(&cmd_variant).exists()); - if has_cmd { - cmd_variant - } else { - command.to_string() + let cmd_str = command.to_string(); + let args_vec: Vec = args.to_vec(); + let env_list: Vec = env_whitelist.to_vec(); + + let transport = + TokioChildProcess::new(Command::new(&cmd_str).configure(move |cmd| { + for arg in &args_vec { + cmd.arg(arg); + } + // Sandbox: clear environment, only pass whitelisted vars + cmd.env_clear(); + for var_name in &env_list { + if let Ok(val) = std::env::var(var_name) { + cmd.env(var_name, val); + } + } + // Always pass PATH for binary resolution + if let Ok(path) = std::env::var("PATH") { + cmd.env("PATH", path); + } + // On Windows, npm/node need extra vars + if cfg!(windows) { + for var in &[ + "APPDATA", + "LOCALAPPDATA", + "USERPROFILE", + "SystemRoot", + "TEMP", + "TMP", + "HOME", + "HOMEDRIVE", + "HOMEPATH", + ] { + if let Ok(val) = std::env::var(var) { + cmd.env(var, val); + } + } + } + })) + .map_err(|e| format!("Failed to spawn MCP server '{cmd_str}': {e}"))?; + + let client = client_info + .serve(transport) + .await + .map_err(|e| format!("MCP stdio handshake failed: {e}"))?; + + Ok(client) + } + + /// Connect using Streamable HTTP transport (or SSE fallback via the same endpoint). + /// + /// The `rmcp` SDK's `StreamableHttpClientTransport` handles the full + /// Streamable HTTP protocol: Accept headers, Mcp-Session-Id tracking, + /// SSE stream parsing, and content-type negotiation. + async fn connect_http( + url: &str, + headers: &[String], + client_info: ClientInfo, + ) -> Result, String> { + use rmcp::transport::streamable_http_client::StreamableHttpClientTransportConfig; + use rmcp::transport::StreamableHttpClientTransport; + + Self::check_ssrf(url)?; + + // Parse custom headers (e.g., "Authorization: Bearer "). + let mut custom_headers: HashMap = HashMap::new(); + for header_str in headers { + if let Some((name, value)) = header_str.split_once(':') { + let name = name.trim(); + let value = value.trim(); + if let (Ok(hn), Ok(hv)) = ( + HeaderName::from_bytes(name.as_bytes()), + HeaderValue::from_str(value), + ) { + custom_headers.insert(hn, hv); } } - } else { - command.to_string() + } + + let config = StreamableHttpClientTransportConfig { + uri: Arc::from(url), + custom_headers, + ..Default::default() }; - let mut cmd = tokio::process::Command::new(&resolved_command); - cmd.args(args); - cmd.stdin(Stdio::piped()); - cmd.stdout(Stdio::piped()); - cmd.stderr(Stdio::piped()); + let transport = StreamableHttpClientTransport::from_config(config); - // Sandbox: clear environment, only pass whitelisted vars - cmd.env_clear(); - for var_name in env_whitelist { - if let Ok(val) = std::env::var(var_name) { - cmd.env(var_name, val); - } - } - // Always pass PATH for binary resolution - if let Ok(path) = std::env::var("PATH") { - cmd.env("PATH", path); - } - // On Windows, npm/node need APPDATA, USERPROFILE, LOCALAPPDATA, and SystemRoot - if cfg!(windows) { - for var in &[ - "APPDATA", - "LOCALAPPDATA", - "USERPROFILE", - "SystemRoot", - "TEMP", - "TMP", - "HOME", - "HOMEDRIVE", - "HOMEPATH", - ] { - if let Ok(val) = std::env::var(var) { - cmd.env(var, val); - } - } - } + let client = client_info + .serve(transport) + .await + .map_err(|e| format!("MCP HTTP connection failed: {e}"))?; - let mut child = cmd - .spawn() - .map_err(|e| format!("Failed to spawn MCP server '{resolved_command}': {e}"))?; - - // Log stderr in background for debugging MCP server issues - if let Some(stderr) = child.stderr.take() { - let cmd_name = resolved_command.clone(); - tokio::spawn(async move { - use tokio::io::AsyncBufReadExt; - let reader = tokio::io::BufReader::new(stderr); - let mut lines = reader.lines(); - while let Ok(Some(line)) = lines.next_line().await { - tracing::debug!(mcp_server = %cmd_name, "stderr: {line}"); - } - }); - } - - let stdin = child - .stdin - .take() - .ok_or("Failed to capture MCP server stdin")?; - let stdout = child - .stdout - .take() - .ok_or("Failed to capture MCP server stdout")?; - - Ok(McpTransportHandle::Stdio { - child: Box::new(child), - stdin, - stdout: BufReader::new(stdout), - }) + Ok(client) } - async fn connect_sse(url: &str) -> Result { - // Basic SSRF check: reject obviously private URLs + /// Basic SSRF check: reject obviously private/metadata URLs. + fn check_ssrf(url: &str) -> Result<(), String> { let lower = url.to_lowercase(); if lower.contains("169.254.169.254") || lower.contains("metadata.google") { - return Err("SSRF: MCP SSE URL targets metadata endpoint".to_string()); - } - - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(30)) - .build() - .map_err(|e| format!("Failed to create HTTP client: {e}"))?; - - Ok(McpTransportHandle::Sse { - client, - url: url.to_string(), - }) - } -} - -impl Drop for McpConnection { - fn drop(&mut self) { - if let McpTransportHandle::Stdio { ref mut child, .. } = self.transport { - // Best-effort kill of the subprocess - let _ = child.start_kill(); + return Err("SSRF: MCP URL targets metadata endpoint".to_string()); } + Ok(()) } } @@ -668,16 +424,12 @@ mod tests { #[test] fn test_hyphenated_tool_name_preserved() { - // Tool names with hyphens get normalized to underscores for namespacing, - // but original_names map preserves the original for call_tool dispatch. let namespaced = format_mcp_tool_name("sqlcl", "list-connections"); assert_eq!(namespaced, "mcp_sqlcl_list_connections"); - // Simulate what discover_tools does let mut original_names = HashMap::new(); original_names.insert(namespaced.clone(), "list-connections".to_string()); - // call_tool should resolve to original hyphenated name let raw = original_names .get(&namespaced) .map(|s| s.as_str()) @@ -696,25 +448,21 @@ mod tests { #[test] fn test_extract_mcp_server_from_known_with_hyphens() { - // Server "bocha-search" normalized to "bocha_search" in tool prefix let servers = vec!["bocha-search", "github"]; let tool = "mcp_bocha_search_bocha_web_search"; assert_eq!( extract_mcp_server_from_known(tool, &servers), Some("bocha-search") ); - // Simple server name still works assert_eq!( extract_mcp_server_from_known("mcp_github_create_issue", &servers), Some("github") ); - // Non-MCP tool returns None assert_eq!(extract_mcp_server_from_known("file_read", &servers), None); } #[test] fn test_extract_mcp_server_from_known_longest_match() { - // "my-api" and "my-api-v2" — should match the longer one let servers = vec!["my-api", "my-api-v2"]; assert_eq!( extract_mcp_server_from_known("mcp_my_api_v2_get_users", &servers), @@ -726,60 +474,6 @@ mod tests { ); } - #[test] - fn test_mcp_jsonrpc_initialize() { - // Verify the initialize request structure - let request = JsonRpcRequest { - jsonrpc: "2.0", - id: 1, - method: "initialize".to_string(), - params: Some(serde_json::json!({ - "protocolVersion": "2024-11-05", - "capabilities": {}, - "clientInfo": { - "name": "openfang", - "version": "0.1.0" - } - })), - }; - let json = serde_json::to_string(&request).unwrap(); - assert!(json.contains("initialize")); - assert!(json.contains("protocolVersion")); - assert!(json.contains("openfang")); - } - - #[test] - fn test_mcp_jsonrpc_tools_list() { - // Simulate a tools/list response - let response_json = r#"{ - "jsonrpc": "2.0", - "id": 2, - "result": { - "tools": [ - { - "name": "create_issue", - "description": "Create a GitHub issue", - "inputSchema": { - "type": "object", - "properties": { - "title": {"type": "string"}, - "body": {"type": "string"} - }, - "required": ["title"] - } - } - ] - } - }"#; - - let response: JsonRpcResponse = serde_json::from_str(response_json).unwrap(); - assert!(response.error.is_none()); - let result = response.result.unwrap(); - let tools = result["tools"].as_array().unwrap(); - assert_eq!(tools.len(), 1); - assert_eq!(tools[0]["name"].as_str().unwrap(), "create_issue"); - } - #[test] fn test_mcp_transport_config_serde() { let config = McpServerConfig { @@ -793,6 +487,7 @@ mod tests { }, timeout_secs: 30, env: vec!["GITHUB_PERSONAL_ACCESS_TOKEN".to_string()], + headers: vec![], }; let json = serde_json::to_string(&config).unwrap(); @@ -817,6 +512,7 @@ mod tests { }, timeout_secs: 60, env: vec![], + headers: vec![], }; let json = serde_json::to_string(&sse_config).unwrap(); let back: McpServerConfig = serde_json::from_str(&json).unwrap(); @@ -824,5 +520,24 @@ mod tests { McpTransport::Sse { url } => assert_eq!(url, "https://example.com/mcp"), _ => panic!("Expected SSE transport"), } + + // HTTP (Streamable HTTP) variant + let http_config = McpServerConfig { + name: "atlassian".to_string(), + transport: McpTransport::Http { + url: "https://mcp.atlassian.com/v1/mcp".to_string(), + }, + timeout_secs: 120, + env: vec![], + headers: vec!["Authorization: Bearer test-token-456".to_string()], + }; + let json = serde_json::to_string(&http_config).unwrap(); + let back: McpServerConfig = serde_json::from_str(&json).unwrap(); + match back.transport { + McpTransport::Http { url } => { + assert_eq!(url, "https://mcp.atlassian.com/v1/mcp") + } + _ => panic!("Expected Http transport"), + } } } diff --git a/crates/openfang-types/src/config.rs b/crates/openfang-types/src/config.rs index aaeae70e..6fbb3514 100644 --- a/crates/openfang-types/src/config.rs +++ b/crates/openfang-types/src/config.rs @@ -310,6 +310,10 @@ impl Default for WebFetchConfig { #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(default)] pub struct BrowserConfig { + /// Enable the built-in CDP browser tools (browser_navigate, browser_click, + /// etc.). Set to `false` when using an external browser MCP server such as + /// CamoFox, which replaces these tools with its own set. + pub enabled: bool, /// Run browser in headless mode (no visible window). pub headless: bool, /// Viewport width in pixels. @@ -329,6 +333,7 @@ pub struct BrowserConfig { impl Default for BrowserConfig { fn default() -> Self { Self { + enabled: true, headless: true, viewport_width: 1280, viewport_height: 720, @@ -1224,6 +1229,10 @@ pub struct McpServerConfigEntry { /// Environment variables to pass through (e.g., ["GITHUB_PERSONAL_ACCESS_TOKEN"]). #[serde(default)] pub env: Vec, + /// Extra HTTP headers for SSE / Streamable-HTTP transports. + /// Each entry is `"Header-Name: value"` (e.g., `"Authorization: Bearer "`). + #[serde(default)] + pub headers: Vec, } fn default_mcp_timeout() -> u64 { @@ -1242,6 +1251,8 @@ pub enum McpTransportEntry { }, /// HTTP Server-Sent Events. Sse { url: String }, + /// Streamable HTTP (MCP 2025-03-26+). + Http { url: String }, } /// A2A (Agent-to-Agent) protocol configuration. From e53e238e81cac75d2559e2ef3dcf66e9c163aa22 Mon Sep 17 00:00:00 2001 From: Sky Moore Date: Sat, 21 Mar 2026 19:23:58 +0000 Subject: [PATCH 2/4] fix: cargo fmt and update rustls-webpki to 0.103.10 (RUSTSEC-2026-0049) --- Cargo.lock | 4 +- crates/openfang-api/src/routes.rs | 6 +- crates/openfang-channels/src/whatsapp.rs | 9 ++- crates/openfang-memory/src/knowledge.rs | 12 +--- crates/openfang-runtime/src/agent_loop.rs | 34 +++++++-- .../openfang-runtime/src/drivers/anthropic.rs | 5 +- crates/openfang-runtime/src/mcp.rs | 70 +++++++++---------- 7 files changed, 77 insertions(+), 63 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7c4172bf..8495e00c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5724,9 +5724,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.9" +version = "0.103.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7df23109aa6c1567d1c575b9952556388da57401e4ace1d15f79eedad0d8f53" +checksum = "df33b2b81ac578cabaf06b89b0631153a3f416b0a886e8a7a1707fb51abbd1ef" dependencies = [ "aws-lc-rs", "ring", diff --git a/crates/openfang-api/src/routes.rs b/crates/openfang-api/src/routes.rs index 03172769..76441cd7 100644 --- a/crates/openfang-api/src/routes.rs +++ b/crates/openfang-api/src/routes.rs @@ -367,7 +367,11 @@ pub async fn send_message( // (not as a separate session message which the LLM may not process). let content_blocks = if !req.attachments.is_empty() { let image_blocks = resolve_attachments(&req.attachments); - if image_blocks.is_empty() { None } else { Some(image_blocks) } + if image_blocks.is_empty() { + None + } else { + Some(image_blocks) + } } else { None }; diff --git a/crates/openfang-channels/src/whatsapp.rs b/crates/openfang-channels/src/whatsapp.rs index 06156ab5..16f37b56 100644 --- a/crates/openfang-channels/src/whatsapp.rs +++ b/crates/openfang-channels/src/whatsapp.rs @@ -258,7 +258,8 @@ impl ChannelAdapter for WhatsAppAdapter { "https://graph.facebook.com/v21.0/{}/messages", self.phone_number_id ); - let resp = self.client + let resp = self + .client .post(&api_url) .bearer_auth(&*self.access_token) .json(&body) @@ -284,7 +285,8 @@ impl ChannelAdapter for WhatsAppAdapter { "https://graph.facebook.com/v21.0/{}/messages", self.phone_number_id ); - let resp = self.client + let resp = self + .client .post(&api_url) .bearer_auth(&*self.access_token) .json(&body) @@ -310,7 +312,8 @@ impl ChannelAdapter for WhatsAppAdapter { "https://graph.facebook.com/v21.0/{}/messages", self.phone_number_id ); - let resp = self.client + let resp = self + .client .post(&api_url) .bearer_auth(&*self.access_token) .json(&body) diff --git a/crates/openfang-memory/src/knowledge.rs b/crates/openfang-memory/src/knowledge.rs index 8c87b0ac..e4f5c773 100644 --- a/crates/openfang-memory/src/knowledge.rs +++ b/crates/openfang-memory/src/knowledge.rs @@ -100,11 +100,7 @@ impl KnowledgeStore { let mut idx = 1; if let Some(ref source) = pattern.source { - sql.push_str(&format!( - " AND (s.id = ?{} OR s.name = ?{})", - idx, - idx + 1 - )); + sql.push_str(&format!(" AND (s.id = ?{} OR s.name = ?{})", idx, idx + 1)); params.push(Box::new(source.clone())); params.push(Box::new(source.clone())); idx += 2; @@ -117,11 +113,7 @@ impl KnowledgeStore { idx += 1; } if let Some(ref target) = pattern.target { - sql.push_str(&format!( - " AND (t.id = ?{} OR t.name = ?{})", - idx, - idx + 1 - )); + sql.push_str(&format!(" AND (t.id = ?{} OR t.name = ?{})", idx, idx + 1)); params.push(Box::new(target.clone())); params.push(Box::new(target.clone())); idx += 2; diff --git a/crates/openfang-runtime/src/agent_loop.rs b/crates/openfang-runtime/src/agent_loop.rs index bd68752e..7a3b02c2 100644 --- a/crates/openfang-runtime/src/agent_loop.rs +++ b/crates/openfang-runtime/src/agent_loop.rs @@ -57,8 +57,15 @@ fn phantom_action_detected(text: &str) -> bool { let lower = text.to_lowercase(); let action_verbs = ["sent ", "posted ", "emailed ", "delivered ", "forwarded "]; let channel_refs = [ - "telegram", "whatsapp", "slack", "discord", "email", "channel", - "message sent", "successfully sent", "has been sent", + "telegram", + "whatsapp", + "slack", + "discord", + "email", + "channel", + "message sent", + "successfully sent", + "has been sent", ]; let has_action = action_verbs.iter().any(|v| lower.contains(v)); let has_channel = channel_refs.iter().any(|c| lower.contains(c)); @@ -272,7 +279,9 @@ pub async fn run_agent_loop( // The LLM already received them via llm_messages above. for msg in session.messages.iter_mut() { if let MessageContent::Blocks(blocks) = &mut msg.content { - let had_images = blocks.iter().any(|b| matches!(b, ContentBlock::Image { .. })); + let had_images = blocks + .iter() + .any(|b| matches!(b, ContentBlock::Image { .. })); if had_images { blocks.retain(|b| !matches!(b, ContentBlock::Image { .. })); if blocks.is_empty() { @@ -460,7 +469,10 @@ pub async fn run_agent_loop( // One-shot retry: if the LLM returns empty text with no tool use, // try once more before accepting the empty result. // Triggers on first call OR when input_tokens=0 (silently failed request). - if text.trim().is_empty() && response.tool_calls.is_empty() && !response.has_any_content() { + if text.trim().is_empty() + && response.tool_calls.is_empty() + && !response.has_any_content() + { let is_silent_failure = response.usage.input_tokens == 0 && response.usage.output_tokens == 0; if iteration == 0 || is_silent_failure { @@ -505,7 +517,10 @@ pub async fn run_agent_loop( // channel action (send, post, email, etc.) but never actually // called the corresponding tool, re-prompt once to force real // tool usage instead of hallucinated completion. - let text = if !any_tools_executed && iteration == 0 && phantom_action_detected(&text) { + let text = if !any_tools_executed + && iteration == 0 + && phantom_action_detected(&text) + { warn!(agent = %manifest.name, "Phantom action detected — re-prompting for real tool use"); messages.push(Message::assistant(text)); messages.push(Message::user( @@ -1419,7 +1434,9 @@ pub async fn run_agent_loop_streaming( // The LLM already received them via llm_messages above. for msg in session.messages.iter_mut() { if let MessageContent::Blocks(blocks) = &mut msg.content { - let had_images = blocks.iter().any(|b| matches!(b, ContentBlock::Image { .. })); + let had_images = blocks + .iter() + .any(|b| matches!(b, ContentBlock::Image { .. })); if had_images { blocks.retain(|b| !matches!(b, ContentBlock::Image { .. })); if blocks.is_empty() { @@ -1620,7 +1637,10 @@ pub async fn run_agent_loop_streaming( // One-shot retry: if the LLM returns empty text with no tool use, // try once more before accepting the empty result. // Triggers on first call OR when input_tokens=0 (silently failed request). - if text.trim().is_empty() && response.tool_calls.is_empty() && !response.has_any_content() { + if text.trim().is_empty() + && response.tool_calls.is_empty() + && !response.has_any_content() + { let is_silent_failure = response.usage.input_tokens == 0 && response.usage.output_tokens == 0; if iteration == 0 || is_silent_failure { diff --git a/crates/openfang-runtime/src/drivers/anthropic.rs b/crates/openfang-runtime/src/drivers/anthropic.rs index ad47d8b5..4dab8991 100644 --- a/crates/openfang-runtime/src/drivers/anthropic.rs +++ b/crates/openfang-runtime/src/drivers/anthropic.rs @@ -471,9 +471,8 @@ impl LlmDriver for AnthropicDriver { input_json, }) = blocks.get(block_idx) { - let input: serde_json::Value = - serde_json::from_str(input_json) - .unwrap_or_else(|_| serde_json::json!({})); + let input: serde_json::Value = serde_json::from_str(input_json) + .unwrap_or_else(|_| serde_json::json!({})); let _ = tx .send(StreamEvent::ToolUseEnd { id: id.clone(), diff --git a/crates/openfang-runtime/src/mcp.rs b/crates/openfang-runtime/src/mcp.rs index c15bb86a..b9f5f381 100644 --- a/crates/openfang-runtime/src/mcp.rs +++ b/crates/openfang-runtime/src/mcp.rs @@ -174,10 +174,7 @@ impl McpConnection { .or_else(|| strip_mcp_prefix(&self.config.name, name).map(|s| s.to_string())) .unwrap_or_else(|| name.to_string()); - let args = arguments - .as_object() - .cloned() - .unwrap_or_default(); + let args = arguments.as_object().cloned().unwrap_or_default(); debug!(tool = %raw_name, server = %self.config.name, "MCP tool call"); @@ -236,42 +233,41 @@ impl McpConnection { let args_vec: Vec = args.to_vec(); let env_list: Vec = env_whitelist.to_vec(); - let transport = - TokioChildProcess::new(Command::new(&cmd_str).configure(move |cmd| { - for arg in &args_vec { - cmd.arg(arg); + let transport = TokioChildProcess::new(Command::new(&cmd_str).configure(move |cmd| { + for arg in &args_vec { + cmd.arg(arg); + } + // Sandbox: clear environment, only pass whitelisted vars + cmd.env_clear(); + for var_name in &env_list { + if let Ok(val) = std::env::var(var_name) { + cmd.env(var_name, val); } - // Sandbox: clear environment, only pass whitelisted vars - cmd.env_clear(); - for var_name in &env_list { - if let Ok(val) = std::env::var(var_name) { - cmd.env(var_name, val); + } + // Always pass PATH for binary resolution + if let Ok(path) = std::env::var("PATH") { + cmd.env("PATH", path); + } + // On Windows, npm/node need extra vars + if cfg!(windows) { + for var in &[ + "APPDATA", + "LOCALAPPDATA", + "USERPROFILE", + "SystemRoot", + "TEMP", + "TMP", + "HOME", + "HOMEDRIVE", + "HOMEPATH", + ] { + if let Ok(val) = std::env::var(var) { + cmd.env(var, val); } } - // Always pass PATH for binary resolution - if let Ok(path) = std::env::var("PATH") { - cmd.env("PATH", path); - } - // On Windows, npm/node need extra vars - if cfg!(windows) { - for var in &[ - "APPDATA", - "LOCALAPPDATA", - "USERPROFILE", - "SystemRoot", - "TEMP", - "TMP", - "HOME", - "HOMEDRIVE", - "HOMEPATH", - ] { - if let Ok(val) = std::env::var(var) { - cmd.env(var, val); - } - } - } - })) - .map_err(|e| format!("Failed to spawn MCP server '{cmd_str}': {e}"))?; + } + })) + .map_err(|e| format!("Failed to spawn MCP server '{cmd_str}': {e}"))?; let client = client_info .serve(transport) From feecb6044266538a5f6c2a4873212c3a098f559e Mon Sep 17 00:00:00 2001 From: Sky Moore Date: Sat, 21 Mar 2026 19:33:41 +0000 Subject: [PATCH 3/4] fix: use specific capability-denial assertion to avoid OS error false positive on Linux CI --- crates/openfang-runtime/src/tool_runner.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/crates/openfang-runtime/src/tool_runner.rs b/crates/openfang-runtime/src/tool_runner.rs index ac8e3df2..a44f93ec 100644 --- a/crates/openfang-runtime/src/tool_runner.rs +++ b/crates/openfang-runtime/src/tool_runner.rs @@ -3651,10 +3651,13 @@ mod tests { None, // process_manager ) .await; - // Should NOT be "Permission denied" — it should normalize to file_write - // and pass the capability check. It will fail for other reasons (path validation). + // Should NOT be the capability-check denial — it should normalize to file_write + // and pass the capability check. It may fail for other reasons (path validation, + // OS-level errors), but not the agent capability gate. assert!( - !result.content.contains("Permission denied"), + !result + .content + .contains("does not have capability to use tool"), "fs-write should normalize to file_write and pass capability check, got: {}", result.content ); From 9e2853a5f83a60c9c0a8cf81758e0f532c1f089f Mon Sep 17 00:00:00 2001 From: Sky Moore Date: Fri, 27 Mar 2026 18:40:33 +0000 Subject: [PATCH 4/4] fix: resolve CI failures after rebase on upstream/main - Add missing budget_config field to AppState in all 3 test files - Fix redundant closures and unwrap_or_else in openfang-memory semantic.rs - Fix needless_borrow in openfang-api routes.rs (toml::from_str) - Update parse_researcher_hand test to match new max_iterations = 25 - Update tar to 0.4.45 to fix RUSTSEC-2026-0067 and RUSTSEC-2026-0068 - Apply cargo fmt fixes in ws.rs and feishu.rs --- Cargo.lock | 29 ++-- crates/openfang-api/src/routes.rs | 2 +- crates/openfang-api/src/ws.rs | 8 +- .../tests/api_integration_test.rs | 2 + .../tests/daemon_lifecycle_test.rs | 2 + crates/openfang-api/tests/load_test.rs | 1 + crates/openfang-channels/src/feishu.rs | 61 ++++++--- .../tests/bridge_integration_test.rs | 17 ++- crates/openfang-cli/src/main.rs | 3 +- crates/openfang-hands/src/bundled.rs | 124 ++++++++++++++---- crates/openfang-hands/src/registry.rs | 4 +- crates/openfang-kernel/src/heartbeat.rs | 51 +++++-- crates/openfang-kernel/src/kernel.rs | 22 +++- crates/openfang-memory/src/semantic.rs | 24 ++-- crates/openfang-memory/src/substrate.rs | 14 +- crates/openfang-runtime/src/agent_loop.rs | 9 +- .../openfang-runtime/src/drivers/anthropic.rs | 5 +- .../src/drivers/claude_code.rs | 10 +- crates/openfang-runtime/src/drivers/openai.rs | 14 +- crates/openfang-runtime/src/drivers/vertex.rs | 5 +- crates/openfang-skills/src/registry.rs | 20 ++- 21 files changed, 311 insertions(+), 116 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8495e00c..a0980879 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -912,7 +912,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -1819,7 +1819,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -4155,6 +4155,7 @@ dependencies = [ "async-trait", "chrono", "openfang-types", + "reqwest 0.12.28", "rmp-serde", "rusqlite", "serde", @@ -4318,6 +4319,15 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "openssl-src" +version = "300.5.5+3.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f1787d533e03597a7934fd0a765f0d28e94ecc5fb7789f8053b1e699a56f709" +dependencies = [ + "cc", +] + [[package]] name = "openssl-sys" version = "0.9.112" @@ -4326,6 +4336,7 @@ checksum = "57d55af3b3e226502be1526dfdba67ab0e9c96fc293004e79576b2b9edb0dbdb" dependencies = [ "cc", "libc", + "openssl-src", "pkg-config", "vcpkg", ] @@ -5654,7 +5665,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -5713,7 +5724,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -6573,9 +6584,9 @@ dependencies = [ [[package]] name = "tar" -version = "0.4.44" +version = "0.4.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d863878d212c87a19c1a610eb53bb01fe12951c0501cf5a0d65f724914a667a" +checksum = "22692a6476a21fa75fdfc11d452fda482af402c008cdbaf3476414e122040973" dependencies = [ "filetime", "libc", @@ -7005,7 +7016,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -7557,7 +7568,7 @@ checksum = "f2f6fb2847f6742cd76af783a2a2c49e9375d0a111c7bef6f71cd9e738c72d6e" dependencies = [ "memoffset", "tempfile", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -8454,7 +8465,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] diff --git a/crates/openfang-api/src/routes.rs b/crates/openfang-api/src/routes.rs index 76441cd7..882afaa1 100644 --- a/crates/openfang-api/src/routes.rs +++ b/crates/openfang-api/src/routes.rs @@ -3133,7 +3133,7 @@ pub async fn list_templates() -> impl IntoResponse { let manifest_content = std::fs::read_to_string(&manifest_path).ok(); let description = manifest_content .as_ref() - .and_then(|content| toml::from_str::(&content).ok()) + .and_then(|content| toml::from_str::(content).ok()) .map(|m| m.description) .unwrap_or_default(); diff --git a/crates/openfang-api/src/ws.rs b/crates/openfang-api/src/ws.rs index 4052fe16..353ffd77 100644 --- a/crates/openfang-api/src/ws.rs +++ b/crates/openfang-api/src/ws.rs @@ -1003,7 +1003,9 @@ fn map_stream_event(event: &StreamEvent, verbose: VerboseLevel) -> Option { + StreamEvent::ToolUseEnd { + id, name, input, .. + } if name == "canvas_present" => { let html = input.get("html").and_then(|v| v.as_str()).unwrap_or(""); let title = input .get("title") @@ -1017,7 +1019,9 @@ fn map_stream_event(event: &StreamEvent, verbose: VerboseLevel) -> Option match verbose { + StreamEvent::ToolUseEnd { + id, name, input, .. + } => match verbose { VerboseLevel::Off => None, VerboseLevel::On => { let input_preview: String = serde_json::to_string(input) diff --git a/crates/openfang-api/tests/api_integration_test.rs b/crates/openfang-api/tests/api_integration_test.rs index eada9df5..e3592abe 100644 --- a/crates/openfang-api/tests/api_integration_test.rs +++ b/crates/openfang-api/tests/api_integration_test.rs @@ -78,6 +78,7 @@ async fn start_test_server_with_provider( shutdown_notify: Arc::new(tokio::sync::Notify::new()), clawhub_cache: dashmap::DashMap::new(), provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(), + budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())), }); let app = Router::new() @@ -707,6 +708,7 @@ async fn start_test_server_with_auth(api_key: &str) -> TestServer { shutdown_notify: Arc::new(tokio::sync::Notify::new()), clawhub_cache: dashmap::DashMap::new(), provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(), + budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())), }); let api_key = state.kernel.config.api_key.trim().to_string(); diff --git a/crates/openfang-api/tests/daemon_lifecycle_test.rs b/crates/openfang-api/tests/daemon_lifecycle_test.rs index cbe27d41..aba689bf 100644 --- a/crates/openfang-api/tests/daemon_lifecycle_test.rs +++ b/crates/openfang-api/tests/daemon_lifecycle_test.rs @@ -115,6 +115,7 @@ async fn test_full_daemon_lifecycle() { shutdown_notify: Arc::new(tokio::sync::Notify::new()), clawhub_cache: dashmap::DashMap::new(), provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(), + budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())), }); let app = Router::new() @@ -240,6 +241,7 @@ async fn test_server_immediate_responsiveness() { shutdown_notify: Arc::new(tokio::sync::Notify::new()), clawhub_cache: dashmap::DashMap::new(), provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(), + budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())), }); let app = Router::new() diff --git a/crates/openfang-api/tests/load_test.rs b/crates/openfang-api/tests/load_test.rs index 89080db7..c6484354 100644 --- a/crates/openfang-api/tests/load_test.rs +++ b/crates/openfang-api/tests/load_test.rs @@ -59,6 +59,7 @@ async fn start_test_server() -> TestServer { shutdown_notify: Arc::new(tokio::sync::Notify::new()), clawhub_cache: dashmap::DashMap::new(), provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(), + budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())), }); let app = Router::new() diff --git a/crates/openfang-channels/src/feishu.rs b/crates/openfang-channels/src/feishu.rs index 67464ebc..4264da69 100644 --- a/crates/openfang-channels/src/feishu.rs +++ b/crates/openfang-channels/src/feishu.rs @@ -1792,10 +1792,8 @@ mod tests { #[test] fn test_feishu_websocket_adapter_creation() { - let adapter = FeishuAdapter::new_websocket( - "cli_abc123".to_string(), - "app-secret-456".to_string(), - ); + let adapter = + FeishuAdapter::new_websocket("cli_abc123".to_string(), "app-secret-456".to_string()); assert_eq!(adapter.name(), "feishu"); assert_eq!(adapter.connection_mode, FeishuConnectionMode::WebSocket); assert_eq!(adapter.webhook_port, 0); // not used in WS mode @@ -1803,8 +1801,7 @@ mod tests { #[test] fn test_connection_mode_default_is_webhook() { - let adapter = - FeishuAdapter::new("cli_abc123".to_string(), "secret".to_string(), 9000); + let adapter = FeishuAdapter::new("cli_abc123".to_string(), "secret".to_string(), 9000); assert_eq!(adapter.connection_mode, FeishuConnectionMode::Webhook); } @@ -1836,14 +1833,32 @@ mod tests { #[test] fn test_combine_payload_multi_part() { let headers_seq0 = vec![ - FeishuWsHeader { key: "sum".to_string(), value: "2".to_string() }, - FeishuWsHeader { key: "seq".to_string(), value: "0".to_string() }, - FeishuWsHeader { key: "message_id".to_string(), value: "msg1".to_string() }, + FeishuWsHeader { + key: "sum".to_string(), + value: "2".to_string(), + }, + FeishuWsHeader { + key: "seq".to_string(), + value: "0".to_string(), + }, + FeishuWsHeader { + key: "message_id".to_string(), + value: "msg1".to_string(), + }, ]; let headers_seq1 = vec![ - FeishuWsHeader { key: "sum".to_string(), value: "2".to_string() }, - FeishuWsHeader { key: "seq".to_string(), value: "1".to_string() }, - FeishuWsHeader { key: "message_id".to_string(), value: "msg1".to_string() }, + FeishuWsHeader { + key: "sum".to_string(), + value: "2".to_string(), + }, + FeishuWsHeader { + key: "seq".to_string(), + value: "1".to_string(), + }, + FeishuWsHeader { + key: "message_id".to_string(), + value: "msg1".to_string(), + }, ]; let mut parts = HashMap::new(); // First part — not yet complete @@ -1856,10 +1871,14 @@ mod tests { #[test] fn test_ws_header_lookup() { - let headers = vec![ - FeishuWsHeader { key: "service_id".to_string(), value: "svc_123".to_string() }, - ]; - assert_eq!(ws_header(&headers, "service_id"), Some("svc_123".to_string())); + let headers = vec![FeishuWsHeader { + key: "service_id".to_string(), + value: "svc_123".to_string(), + }]; + assert_eq!( + ws_header(&headers, "service_id"), + Some("svc_123".to_string()) + ); assert_eq!(ws_header(&headers, "missing"), None); } @@ -1881,8 +1900,14 @@ mod tests { service: 1, method: 2, headers: vec![ - FeishuWsHeader { key: "log_id".to_string(), value: "log_abc".to_string() }, - FeishuWsHeader { key: "type".to_string(), value: "event".to_string() }, + FeishuWsHeader { + key: "log_id".to_string(), + value: "log_abc".to_string(), + }, + FeishuWsHeader { + key: "type".to_string(), + value: "event".to_string(), + }, ], payload: Some(vec![]), payload_encoding: None, diff --git a/crates/openfang-channels/tests/bridge_integration_test.rs b/crates/openfang-channels/tests/bridge_integration_test.rs index e954d4df..85ea2072 100644 --- a/crates/openfang-channels/tests/bridge_integration_test.rs +++ b/crates/openfang-channels/tests/bridge_integration_test.rs @@ -495,7 +495,10 @@ async fn test_bridge_manager_lifecycle() { assert_eq!(sent.len(), 5, "Expected 5 responses, got {}", sent.len()); for (i, (_, text)) in sent.iter().enumerate() { - assert!(text.contains(&format!("message {i}")), "Expected 'message {i}' in: {text}"); + assert!( + text.contains(&format!("message {i}")), + "Expected 'message {i}' in: {text}" + ); } // Stop — should complete without hanging @@ -544,11 +547,19 @@ async fn test_bridge_multiple_adapters() { let tg_sent = tg_ref.get_sent(); assert_eq!(tg_sent.len(), 1); - assert!(tg_sent[0].1.contains("from telegram"), "Expected 'from telegram' in: {}", tg_sent[0].1); + assert!( + tg_sent[0].1.contains("from telegram"), + "Expected 'from telegram' in: {}", + tg_sent[0].1 + ); let dc_sent = dc_ref.get_sent(); assert_eq!(dc_sent.len(), 1); - assert!(dc_sent[0].1.contains("from discord"), "Expected 'from discord' in: {}", dc_sent[0].1); + assert!( + dc_sent[0].1.contains("from discord"), + "Expected 'from discord' in: {}", + dc_sent[0].1 + ); manager.stop().await; } diff --git a/crates/openfang-cli/src/main.rs b/crates/openfang-cli/src/main.rs index 9dc9c1c7..e7778cf4 100644 --- a/crates/openfang-cli/src/main.rs +++ b/crates/openfang-cli/src/main.rs @@ -886,7 +886,6 @@ fn write_stdout_safe(msg: &str) { } fn main() { - // Load ~/.openfang/.env into process environment (system env takes priority). dotenv::load_dotenv(); @@ -6949,4 +6948,4 @@ args = ["-y", "@modelcontextprotocol/server-github"] assert!(!is_openfang_path_line("# openfang config", dir)); assert!(!is_openfang_path_line("alias of=openfang", dir)); } -} \ No newline at end of file +} diff --git a/crates/openfang-hands/src/bundled.rs b/crates/openfang-hands/src/bundled.rs index 6ed6f3f0..330d56c5 100644 --- a/crates/openfang-hands/src/bundled.rs +++ b/crates/openfang-hands/src/bundled.rs @@ -166,7 +166,7 @@ mod tests { assert!(def.tools.contains(&"event_publish".to_string())); assert!(!def.settings.is_empty()); assert!(!def.dashboard.metrics.is_empty()); - assert_eq!(def.agent.max_iterations, Some(80)); + assert_eq!(def.agent.max_iterations, Some(25)); } #[test] @@ -315,38 +315,112 @@ mod tests { assert_eq!(def.category, crate::HandCategory::Security); assert!(def.skill_content.is_some()); // Required env vars - assert!(!def.requires.is_empty(), "infisical-sync must declare env var requirements"); + assert!( + !def.requires.is_empty(), + "infisical-sync must declare env var requirements" + ); let req_keys: Vec<&str> = def.requires.iter().map(|r| r.key.as_str()).collect(); - assert!(req_keys.contains(&"INFISICAL_URL"), "must require INFISICAL_URL"); - assert!(req_keys.contains(&"INFISICAL_CLIENT_ID"), "must require INFISICAL_CLIENT_ID"); - assert!(req_keys.contains(&"INFISICAL_CLIENT_SECRET"), "must require INFISICAL_CLIENT_SECRET"); + assert!( + req_keys.contains(&"INFISICAL_URL"), + "must require INFISICAL_URL" + ); + assert!( + req_keys.contains(&"INFISICAL_CLIENT_ID"), + "must require INFISICAL_CLIENT_ID" + ); + assert!( + req_keys.contains(&"INFISICAL_CLIENT_SECRET"), + "must require INFISICAL_CLIENT_SECRET" + ); // Einstein scheduling tools - assert!(def.tools.contains(&"schedule_create".to_string()), "must have schedule_create"); - assert!(def.tools.contains(&"schedule_list".to_string()), "must have schedule_list"); - assert!(def.tools.contains(&"schedule_delete".to_string()), "must have schedule_delete"); + assert!( + def.tools.contains(&"schedule_create".to_string()), + "must have schedule_create" + ); + assert!( + def.tools.contains(&"schedule_list".to_string()), + "must have schedule_list" + ); + assert!( + def.tools.contains(&"schedule_delete".to_string()), + "must have schedule_delete" + ); // Memory tools - assert!(def.tools.contains(&"memory_store".to_string()), "must have memory_store"); - assert!(def.tools.contains(&"memory_recall".to_string()), "must have memory_recall"); + assert!( + def.tools.contains(&"memory_store".to_string()), + "must have memory_store" + ); + assert!( + def.tools.contains(&"memory_recall".to_string()), + "must have memory_recall" + ); // Knowledge graph tools - assert!(def.tools.contains(&"knowledge_add_entity".to_string()), "must have knowledge_add_entity"); - assert!(def.tools.contains(&"knowledge_add_relation".to_string()), "must have knowledge_add_relation"); - assert!(def.tools.contains(&"knowledge_query".to_string()), "must have knowledge_query"); + assert!( + def.tools.contains(&"knowledge_add_entity".to_string()), + "must have knowledge_add_entity" + ); + assert!( + def.tools.contains(&"knowledge_add_relation".to_string()), + "must have knowledge_add_relation" + ); + assert!( + def.tools.contains(&"knowledge_query".to_string()), + "must have knowledge_query" + ); // Event bus - assert!(def.tools.contains(&"event_publish".to_string()), "must have event_publish"); + assert!( + def.tools.contains(&"event_publish".to_string()), + "must have event_publish" + ); // Infisical-specific tools - assert!(def.tools.contains(&"shell_exec".to_string()), "must have shell_exec"); - assert!(def.tools.contains(&"vault_set".to_string()), "must have vault_set"); - assert!(def.tools.contains(&"vault_get".to_string()), "must have vault_get"); - assert!(def.tools.contains(&"vault_list".to_string()), "must have vault_list"); - assert!(def.tools.contains(&"vault_delete".to_string()), "must have vault_delete"); + assert!( + def.tools.contains(&"shell_exec".to_string()), + "must have shell_exec" + ); + assert!( + def.tools.contains(&"vault_set".to_string()), + "must have vault_set" + ); + assert!( + def.tools.contains(&"vault_get".to_string()), + "must have vault_get" + ); + assert!( + def.tools.contains(&"vault_list".to_string()), + "must have vault_list" + ); + assert!( + def.tools.contains(&"vault_delete".to_string()), + "must have vault_delete" + ); // Dashboard - assert!(!def.dashboard.metrics.is_empty(), "must have dashboard metrics"); - let metric_keys: Vec<&str> = def.dashboard.metrics.iter().map(|m| m.memory_key.as_str()).collect(); - assert!(metric_keys.contains(&"infisical_sync_secrets_count"), "must have secrets_count metric"); - assert!(metric_keys.contains(&"infisical_sync_last_sync"), "must have last_sync metric"); + assert!( + !def.dashboard.metrics.is_empty(), + "must have dashboard metrics" + ); + let metric_keys: Vec<&str> = def + .dashboard + .metrics + .iter() + .map(|m| m.memory_key.as_str()) + .collect(); + assert!( + metric_keys.contains(&"infisical_sync_secrets_count"), + "must have secrets_count metric" + ); + assert!( + metric_keys.contains(&"infisical_sync_last_sync"), + "must have last_sync metric" + ); // Agent config - assert!(!def.agent.system_prompt.is_empty(), "must have system_prompt"); - assert!(def.agent.temperature < 0.2, "security hand should use low temperature"); + assert!( + !def.agent.system_prompt.is_empty(), + "must have system_prompt" + ); + assert!( + def.agent.temperature < 0.2, + "security hand should use low temperature" + ); } #[test] diff --git a/crates/openfang-hands/src/registry.rs b/crates/openfang-hands/src/registry.rs index f20be58e..856b82ba 100644 --- a/crates/openfang-hands/src/registry.rs +++ b/crates/openfang-hands/src/registry.rs @@ -382,9 +382,7 @@ impl HandRegistry { // Only non-optional requirements gate readiness. // Optional requirements (e.g. chromium for browser hand) are nice-to-have; // missing them results in "degraded" status but not "requirements not met". - let requirements_met = reqs - .iter() - .all(|(req, ok)| *ok || req.optional); + let requirements_met = reqs.iter().all(|(req, ok)| *ok || req.optional); // A hand is active if at least one instance is in Active status. let active = self diff --git a/crates/openfang-kernel/src/heartbeat.rs b/crates/openfang-kernel/src/heartbeat.rs index f8889c9f..ddfe0488 100644 --- a/crates/openfang-kernel/src/heartbeat.rs +++ b/crates/openfang-kernel/src/heartbeat.rs @@ -169,7 +169,8 @@ pub fn check_agents(registry: &AgentRegistry, config: &HeartbeatConfig) -> Vec, last_active: chrono::DateTime) -> AgentEntry { + fn make_entry( + name: &str, + state: AgentState, + created_at: chrono::DateTime, + last_active: chrono::DateTime, + ) -> AgentEntry { AgentEntry { id: AgentId::new(), name: name.to_string(), @@ -351,14 +357,22 @@ mod tests { // statuses because it was never genuinely active. let registry = crate::registry::AgentRegistry::new(); let five_min_ago = Utc::now() - Duration::seconds(300); - let idle_agent = make_entry("idle-agent", AgentState::Running, five_min_ago, five_min_ago); + let idle_agent = make_entry( + "idle-agent", + AgentState::Running, + five_min_ago, + five_min_ago, + ); registry.register(idle_agent).unwrap(); let config = HeartbeatConfig::default(); // timeout = 180s let statuses = check_agents(®istry, &config); // The idle agent should be skipped entirely - assert!(statuses.is_empty(), "idle agent should be skipped by heartbeat"); + assert!( + statuses.is_empty(), + "idle agent should be skipped by heartbeat" + ); } #[test] @@ -368,14 +382,22 @@ mod tests { let registry = crate::registry::AgentRegistry::new(); let ten_min_ago = Utc::now() - Duration::seconds(600); let five_min_ago = Utc::now() - Duration::seconds(300); - let active_agent = make_entry("active-agent", AgentState::Running, ten_min_ago, five_min_ago); + let active_agent = make_entry( + "active-agent", + AgentState::Running, + ten_min_ago, + five_min_ago, + ); registry.register(active_agent).unwrap(); let config = HeartbeatConfig::default(); // timeout = 180s, inactive = ~300s let statuses = check_agents(®istry, &config); assert_eq!(statuses.len(), 1); - assert!(statuses[0].unresponsive, "active agent past timeout should be unresponsive"); + assert!( + statuses[0].unresponsive, + "active agent past timeout should be unresponsive" + ); } #[test] @@ -391,7 +413,10 @@ mod tests { let statuses = check_agents(®istry, &config); assert_eq!(statuses.len(), 1); - assert!(!statuses[0].unresponsive, "recently active agent should not be unresponsive"); + assert!( + !statuses[0].unresponsive, + "recently active agent should not be unresponsive" + ); } #[test] @@ -400,14 +425,22 @@ mod tests { // even if it was never genuinely active. let registry = crate::registry::AgentRegistry::new(); let five_min_ago = Utc::now() - Duration::seconds(300); - let crashed_agent = make_entry("crashed-idle", AgentState::Crashed, five_min_ago, five_min_ago); + let crashed_agent = make_entry( + "crashed-idle", + AgentState::Crashed, + five_min_ago, + five_min_ago, + ); registry.register(crashed_agent).unwrap(); let config = HeartbeatConfig::default(); let statuses = check_agents(®istry, &config); assert_eq!(statuses.len(), 1); - assert!(statuses[0].unresponsive, "crashed agent should be marked unresponsive"); + assert!( + statuses[0].unresponsive, + "crashed agent should be marked unresponsive" + ); } #[test] diff --git a/crates/openfang-kernel/src/kernel.rs b/crates/openfang-kernel/src/kernel.rs index 42141f0d..5b1429b0 100644 --- a/crates/openfang-kernel/src/kernel.rs +++ b/crates/openfang-kernel/src/kernel.rs @@ -1835,7 +1835,10 @@ impl OpenFangKernel { granted_tools: tools.iter().map(|t| t.name.clone()).collect(), recalled_memories: vec![], skill_summary: Self::build_skill_summary_from(&skill_snapshot, &manifest.skills), - skill_prompt_context: Self::collect_prompt_context_from(&skill_snapshot, &manifest.skills), + skill_prompt_context: Self::collect_prompt_context_from( + &skill_snapshot, + &manifest.skills, + ), mcp_summary: if mcp_tool_count > 0 { self.build_mcp_summary(&manifest.mcp_servers) } else { @@ -2393,7 +2396,10 @@ impl OpenFangKernel { granted_tools: tools.iter().map(|t| t.name.clone()).collect(), recalled_memories: vec![], // Recalled in agent_loop, not here skill_summary: Self::build_skill_summary_from(&skill_snapshot, &manifest.skills), - skill_prompt_context: Self::collect_prompt_context_from(&skill_snapshot, &manifest.skills), + skill_prompt_context: Self::collect_prompt_context_from( + &skill_snapshot, + &manifest.skills, + ), mcp_summary: if mcp_tool_count > 0 { self.build_mcp_summary(&manifest.mcp_servers) } else { @@ -5180,10 +5186,18 @@ impl OpenFangKernel { .unwrap_or_default(); if !tool_allowlist.is_empty() { - all_tools.retain(|t| tool_allowlist.iter().any(|a| a.to_lowercase() == t.name.to_lowercase())); + all_tools.retain(|t| { + tool_allowlist + .iter() + .any(|a| a.to_lowercase() == t.name.to_lowercase()) + }); } if !tool_blocklist.is_empty() { - all_tools.retain(|t| !tool_blocklist.iter().any(|b| b.to_lowercase() == t.name.to_lowercase())); + all_tools.retain(|t| { + !tool_blocklist + .iter() + .any(|b| b.to_lowercase() == t.name.to_lowercase()) + }); } // Step 5: Remove shell_exec if exec_policy denies it. diff --git a/crates/openfang-memory/src/semantic.rs b/crates/openfang-memory/src/semantic.rs index 73181d6b..79d9ae4b 100644 --- a/crates/openfang-memory/src/semantic.rs +++ b/crates/openfang-memory/src/semantic.rs @@ -163,14 +163,7 @@ impl SemanticStore { } Err(e) => { warn!(error = %e, "HTTP memory store failed, falling back to SQLite"); - self.remember_sqlite( - agent_id, - content, - source, - scope, - metadata.clone(), - None, - ) + self.remember_sqlite(agent_id, content, source, scope, metadata.clone(), None) } } } @@ -446,17 +439,13 @@ impl SemanticStore { let created_at = r .created_at .map(|ms| { - chrono::DateTime::from_timestamp_millis(ms as i64) - .unwrap_or_else(|| Utc::now()) + chrono::DateTime::from_timestamp_millis(ms as i64).unwrap_or_else(Utc::now) }) - .unwrap_or_else(|| Utc::now()); + .unwrap_or_else(Utc::now); MemoryFragment { id: MemoryId::new(), - agent_id: filter - .as_ref() - .and_then(|f| f.agent_id) - .unwrap_or_else(AgentId::new), + agent_id: filter.as_ref().and_then(|f| f.agent_id).unwrap_or_default(), content: r.content, embedding: None, metadata: HashMap::new(), @@ -470,7 +459,10 @@ impl SemanticStore { }) .collect(); - debug!(count = fragments.len(), "Recalled memories via HTTP backend"); + debug!( + count = fragments.len(), + "Recalled memories via HTTP backend" + ); Ok(fragments) } } diff --git a/crates/openfang-memory/src/substrate.rs b/crates/openfang-memory/src/substrate.rs index 597e9138..03ecd78c 100644 --- a/crates/openfang-memory/src/substrate.rs +++ b/crates/openfang-memory/src/substrate.rs @@ -43,7 +43,11 @@ impl MemorySubstrate { /// When `memory_config.backend == "http"` and `http_url`/`http_token_env` are set, /// the semantic store routes `remember`/`recall` to the memory-api gateway. /// All other stores (KV, knowledge graph, sessions) remain local SQLite. - pub fn open(db_path: &Path, decay_rate: f32, memory_config: &MemoryConfig) -> OpenFangResult { + pub fn open( + db_path: &Path, + decay_rate: f32, + memory_config: &MemoryConfig, + ) -> OpenFangResult { let conn = Connection::open(db_path).map_err(|e| OpenFangError::Memory(e.to_string()))?; conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;") .map_err(|e| OpenFangError::Memory(e.to_string()))?; @@ -70,13 +74,17 @@ impl MemorySubstrate { ) -> SemanticStore { #[cfg(feature = "http-memory")] if memory_config.backend == "http" { - if let (Some(url), Some(token_env)) = (&memory_config.http_url, &memory_config.http_token_env) { + if let (Some(url), Some(token_env)) = + (&memory_config.http_url, &memory_config.http_token_env) + { match crate::http_client::MemoryApiClient::new(url, token_env) { Ok(client) => { // Best-effort health check on startup match client.health_check() { Ok(()) => info!(url = %url, "HTTP memory backend connected"), - Err(e) => warn!(url = %url, error = %e, "HTTP memory backend health check failed, will retry on use"), + Err(e) => { + warn!(url = %url, error = %e, "HTTP memory backend health check failed, will retry on use") + } } return SemanticStore::new_with_http(conn, client); } diff --git a/crates/openfang-runtime/src/agent_loop.rs b/crates/openfang-runtime/src/agent_loop.rs index 7a3b02c2..3d9601b0 100644 --- a/crates/openfang-runtime/src/agent_loop.rs +++ b/crates/openfang-runtime/src/agent_loop.rs @@ -398,7 +398,14 @@ pub async fn run_agent_loop( // Call LLM with retry, error classification, and circuit breaker let provider_name = manifest.model.provider.as_str(); - let mut response = call_with_retry(&*driver, request, Some(provider_name), None, &manifest.fallback_models).await?; + let mut response = call_with_retry( + &*driver, + request, + Some(provider_name), + None, + &manifest.fallback_models, + ) + .await?; total_usage.input_tokens += response.usage.input_tokens; total_usage.output_tokens += response.usage.output_tokens; diff --git a/crates/openfang-runtime/src/drivers/anthropic.rs b/crates/openfang-runtime/src/drivers/anthropic.rs index 4dab8991..6bac8b31 100644 --- a/crates/openfang-runtime/src/drivers/anthropic.rs +++ b/crates/openfang-runtime/src/drivers/anthropic.rs @@ -520,9 +520,8 @@ impl LlmDriver for AnthropicDriver { name, input_json, } => { - let input: serde_json::Value = - serde_json::from_str(&input_json) - .unwrap_or_else(|_| serde_json::json!({})); + let input: serde_json::Value = serde_json::from_str(&input_json) + .unwrap_or_else(|_| serde_json::json!({})); content.push(ContentBlock::ToolUse { id: id.clone(), name: name.clone(), diff --git a/crates/openfang-runtime/src/drivers/claude_code.rs b/crates/openfang-runtime/src/drivers/claude_code.rs index 20981e81..21e0fb6d 100644 --- a/crates/openfang-runtime/src/drivers/claude_code.rs +++ b/crates/openfang-runtime/src/drivers/claude_code.rs @@ -545,15 +545,11 @@ impl LlmDriver for ClaudeCodeDriver { .join("") }) .unwrap_or_default(); - let text_chunk = - if !chunk.is_empty() { chunk } else { nested }; + let text_chunk = if !chunk.is_empty() { chunk } else { nested }; if !text_chunk.is_empty() { full_text.push_str(&text_chunk); - let _ = tx - .send(StreamEvent::TextDelta { - text: text_chunk, - }) - .await; + let _ = + tx.send(StreamEvent::TextDelta { text: text_chunk }).await; } } "result" | "done" | "complete" => { diff --git a/crates/openfang-runtime/src/drivers/openai.rs b/crates/openfang-runtime/src/drivers/openai.rs index 3d2f46b1..8b927fa8 100644 --- a/crates/openfang-runtime/src/drivers/openai.rs +++ b/crates/openfang-runtime/src/drivers/openai.rs @@ -681,9 +681,8 @@ impl LlmDriver for OpenAIDriver { if let Some(calls) = choice.message.tool_calls { for call in calls { - let input: serde_json::Value = - serde_json::from_str(&call.function.arguments) - .unwrap_or_else(|_| serde_json::json!({})); + let input: serde_json::Value = serde_json::from_str(&call.function.arguments) + .unwrap_or_else(|_| serde_json::json!({})); content.push(ContentBlock::ToolUse { id: call.id.clone(), name: call.function.name.clone(), @@ -1150,7 +1149,10 @@ impl LlmDriver for OpenAIDriver { } // Reasoning/thinking content delta (DeepSeek-R1, Qwen3 via LM Studio/Ollama) - if let Some(reasoning) = delta["reasoning_content"].as_str().or_else(|| delta["reasoning"].as_str()) { + if let Some(reasoning) = delta["reasoning_content"] + .as_str() + .or_else(|| delta["reasoning"].as_str()) + { if !reasoning.is_empty() { reasoning_content.push_str(reasoning); let _ = tx @@ -1315,8 +1317,8 @@ impl LlmDriver for OpenAIDriver { } for (id, name, arguments) in &tool_accum { - let input: serde_json::Value = serde_json::from_str(arguments) - .unwrap_or_else(|_| serde_json::json!({})); + let input: serde_json::Value = + serde_json::from_str(arguments).unwrap_or_else(|_| serde_json::json!({})); content.push(ContentBlock::ToolUse { id: id.clone(), name: name.clone(), diff --git a/crates/openfang-runtime/src/drivers/vertex.rs b/crates/openfang-runtime/src/drivers/vertex.rs index 858e6316..9f048416 100644 --- a/crates/openfang-runtime/src/drivers/vertex.rs +++ b/crates/openfang-runtime/src/drivers/vertex.rs @@ -449,7 +449,10 @@ fn convert_response(resp: VertexResponse) -> Result { - content.push(ContentBlock::Text { text, provider_metadata: None }); + content.push(ContentBlock::Text { + text, + provider_metadata: None, + }); } VertexPart::FunctionCall { function_call } => { tool_calls.push(ToolCall { diff --git a/crates/openfang-skills/src/registry.rs b/crates/openfang-skills/src/registry.rs index 8100b9ef..d006ead4 100644 --- a/crates/openfang-skills/src/registry.rs +++ b/crates/openfang-skills/src/registry.rs @@ -636,7 +636,12 @@ input_schema = { type = "object" } registry.load_all().unwrap(); assert_eq!(registry.count(), 1); assert_eq!( - registry.get("shared-skill").unwrap().manifest.skill.description, + registry + .get("shared-skill") + .unwrap() + .manifest + .skill + .description, "Global version" ); @@ -645,9 +650,18 @@ input_schema = { type = "object" } snapshot.load_workspace_skills(ws_dir.path()).unwrap(); // The workspace version must override the global version - assert_eq!(snapshot.count(), 1, "Duplicate should be overwritten, not added"); assert_eq!( - snapshot.get("shared-skill").unwrap().manifest.skill.description, + snapshot.count(), + 1, + "Duplicate should be overwritten, not added" + ); + assert_eq!( + snapshot + .get("shared-skill") + .unwrap() + .manifest + .skill + .description, "Workspace override version", "Workspace skill must override global skill (#808)" );