diff --git a/Cargo.lock b/Cargo.lock index 24b181cb..98010892 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3906,6 +3906,7 @@ dependencies = [ "reqwest 0.12.28", "serde", "serde_json", + "tempfile", "tokio", "toml 0.8.2", "tracing", diff --git a/crates/openfang-api/static/js/api.js b/crates/openfang-api/static/js/api.js index 08d6a44f..926cc403 100644 --- a/crates/openfang-api/static/js/api.js +++ b/crates/openfang-api/static/js/api.js @@ -224,9 +224,12 @@ var OpenFangAPI = (function() { try { var url = WS_BASE + '/api/agents/' + agentId + '/ws'; if (_authToken) url += '?token=' + encodeURIComponent(_authToken); - _ws = new WebSocket(url); + var socket = new WebSocket(url); + _ws = socket; - _ws.onopen = function() { + socket.onopen = function() { + // Guard: ignore if this socket was superseded by a newer connection + if (_ws !== socket) return; _wsConnected = true; _reconnectAttempts = 0; setConnectionState('connected'); @@ -237,14 +240,20 @@ var OpenFangAPI = (function() { if (_wsCallbacks.onOpen) _wsCallbacks.onOpen(); }; - _ws.onmessage = function(e) { + socket.onmessage = function(e) { try { var data = JSON.parse(e.data); - if (_wsCallbacks.onMessage) _wsCallbacks.onMessage(data); - } catch(err) { /* ignore parse errors */ } + } catch(parseErr) { + return; // Ignore malformed JSON frames + } + // Dispatch outside try/catch so handler errors are not swallowed + if (_wsCallbacks.onMessage) _wsCallbacks.onMessage(data); }; - _ws.onclose = function(e) { + socket.onclose = function(e) { + // Guard: only update state if this is still the active socket. + // A superseded socket closing must not null-out the new connection. + if (_ws !== socket) return; _wsConnected = false; _ws = null; if (_wsAgentId && _reconnectAttempts < MAX_RECONNECT && e.code !== 1000) { @@ -265,7 +274,9 @@ var OpenFangAPI = (function() { if (_wsCallbacks.onClose) _wsCallbacks.onClose(); }; - _ws.onerror = function() { + socket.onerror = function() { + // Guard: ignore errors from superseded sockets + if (_ws !== socket) return; _wsConnected = false; if (_wsCallbacks.onError) _wsCallbacks.onError(); }; diff --git a/crates/openfang-api/static/js/pages/chat.js b/crates/openfang-api/static/js/pages/chat.js index 20890cde..04ec40ac 100644 --- a/crates/openfang-api/static/js/pages/chat.js +++ b/crates/openfang-api/static/js/pages/chat.js @@ -622,8 +622,12 @@ function chatPage() { this.scrollToBottom(); this._resetTypingTimeout(); } else if (data.level) { - var lastThink = this.messages[this.messages.length - 1]; - if (lastThink && lastThink.thinking) lastThink.text = 'Thinking (' + data.level + ')...'; + var thinkIdx = this.messages.length - 1; + var lastThink = thinkIdx >= 0 ? this.messages[thinkIdx] : null; + if (lastThink && lastThink.thinking) { + lastThink.text = 'Thinking (' + data.level + ')...'; + this.messages.splice(thinkIdx, 1, lastThink); + } } break; @@ -636,9 +640,11 @@ function chatPage() { } this._resetTypingTimeout(); } else if (data.state === 'tool') { - var typingMsg = this.messages.length ? this.messages[this.messages.length - 1] : null; + var toolTypIdx = this.messages.length - 1; + var typingMsg = toolTypIdx >= 0 ? this.messages[toolTypIdx] : null; if (typingMsg && (typingMsg.thinking || typingMsg.streaming)) { typingMsg.text = 'Using ' + (data.tool || 'tool') + '...'; + this.messages.splice(toolTypIdx, 1, typingMsg); } this._resetTypingTimeout(); } else if (data.state === 'stop') { @@ -648,7 +654,8 @@ function chatPage() { case 'phase': // Show tool/phase progress so the user sees the agent is working - var phaseMsg = this.messages.length ? this.messages[this.messages.length - 1] : null; + var phaseIdx = this.messages.length - 1; + var phaseMsg = phaseIdx >= 0 ? this.messages[phaseIdx] : null; if (phaseMsg && (phaseMsg.thinking || phaseMsg.streaming)) { // Skip phases that have no user-meaningful display text — "streaming" // and "done" are lifecycle signals, not status to show in the chat bubble. @@ -664,6 +671,7 @@ function chatPage() { if (!phaseMsg._reasoning) phaseMsg._reasoning = ''; phaseMsg._reasoning += (data.detail || '') + '\n'; phaseMsg.text = '
Reasoning...\n\n' + phaseMsg._reasoning + '
'; + this.messages.splice(phaseIdx, 1, phaseMsg); } else if (phaseMsg.thinking) { // Only update text on messages still in thinking state (not yet // receiving streamed content) to avoid overwriting accumulated text. @@ -676,13 +684,15 @@ function chatPage() { phaseDetail = data.detail || 'Working...'; } phaseMsg.text = phaseDetail; + this.messages.splice(phaseIdx, 1, phaseMsg); } } this.scrollToBottom(); break; case 'text_delta': - var last = this.messages.length ? this.messages[this.messages.length - 1] : null; + var lastIdx = this.messages.length - 1; + var last = lastIdx >= 0 ? this.messages[lastIdx] : null; if (last && last.streaming) { if (last.thinking) { last.text = ''; last.thinking = false; } // If we already detected a text-based tool call, skip further text @@ -711,6 +721,10 @@ function chatPage() { } } this.tokenCount = Math.round(last.text.length / 4); + // Force Alpine reactivity: splice-in-place so x-for re-renders + // this item. Direct property mutation on array elements may not + // trigger DOM updates from async WebSocket callbacks. + this.messages.splice(lastIdx, 1, last); } else { this.messages.push({ id: ++msgId, role: 'agent', text: data.content, meta: '', streaming: true, tools: [] }); } @@ -718,17 +732,20 @@ function chatPage() { break; case 'tool_start': - var lastMsg = this.messages.length ? this.messages[this.messages.length - 1] : null; + var tsIdx = this.messages.length - 1; + var lastMsg = tsIdx >= 0 ? this.messages[tsIdx] : null; if (lastMsg && lastMsg.streaming) { if (!lastMsg.tools) lastMsg.tools = []; lastMsg.tools.push({ id: data.tool + '-' + Date.now(), name: data.tool, running: true, expanded: true, input: '', result: '', is_error: false }); + this.messages.splice(tsIdx, 1, lastMsg); } this.scrollToBottom(); break; case 'tool_end': // Tool call parsed by LLM — update tool card with input params - var lastMsg2 = this.messages.length ? this.messages[this.messages.length - 1] : null; + var teIdx = this.messages.length - 1; + var lastMsg2 = teIdx >= 0 ? this.messages[teIdx] : null; if (lastMsg2 && lastMsg2.tools) { for (var ti = lastMsg2.tools.length - 1; ti >= 0; ti--) { if (lastMsg2.tools[ti].name === data.tool && lastMsg2.tools[ti].running) { @@ -736,12 +753,14 @@ function chatPage() { break; } } + this.messages.splice(teIdx, 1, lastMsg2); } break; case 'tool_result': // Tool execution completed — update tool card with result - var lastMsg3 = this.messages.length ? this.messages[this.messages.length - 1] : null; + var trIdx = this.messages.length - 1; + var lastMsg3 = trIdx >= 0 ? this.messages[trIdx] : null; if (lastMsg3 && lastMsg3.tools) { for (var ri = lastMsg3.tools.length - 1; ri >= 0; ri--) { if (lastMsg3.tools[ri].name === data.tool && lastMsg3.tools[ri].running) { @@ -770,6 +789,7 @@ function chatPage() { break; } } + this.messages.splice(trIdx, 1, lastMsg3); } this.scrollToBottom(); break; diff --git a/crates/openfang-cli/Cargo.toml b/crates/openfang-cli/Cargo.toml index 057e28a9..073fea5b 100644 --- a/crates/openfang-cli/Cargo.toml +++ b/crates/openfang-cli/Cargo.toml @@ -31,3 +31,4 @@ openfang-runtime = { path = "../openfang-runtime" } uuid = { workspace = true } ratatui = { workspace = true } colored = { workspace = true } +tempfile = { workspace = true } diff --git a/crates/openfang-cli/src/main.rs b/crates/openfang-cli/src/main.rs index 1fc38b7a..2c37d65b 100644 --- a/crates/openfang-cli/src/main.rs +++ b/crates/openfang-cli/src/main.rs @@ -2690,8 +2690,15 @@ decay_rate = 0.05 } } } - if injection_warnings > 0 { - checks.push(serde_json::json!({"check": "skill_injection_scan", "status": "warn", "warnings": injection_warnings})); + let blocked = skill_reg.blocked_count(); + if injection_warnings > 0 || blocked > 0 { + let total_warnings = injection_warnings + blocked; + if blocked > 0 && !json { + ui::check_warn(&format!( + "{blocked} workspace skill(s) were blocked for critical prompt injection" + )); + } + checks.push(serde_json::json!({"check": "skill_injection_scan", "status": "warn", "warnings": total_warnings, "blocked": blocked})); } else { if !json { ui::check_ok("All skills pass prompt injection scan"); @@ -3538,6 +3545,83 @@ fn cmd_skill_install(source: &str) { "Installed skill: {} v{}", manifest.skill.name, manifest.skill.version ); + } else if source.starts_with("https://") + || source.starts_with("http://") + || source.starts_with("git@") + { + // Git URL install — clone to temp dir then install from there + ui::step(&format!("Cloning skill from {source}...")); + let tmp_dir = tempfile::tempdir().unwrap_or_else(|e| { + eprintln!("Failed to create temp directory: {e}"); + std::process::exit(1); + }); + let clone_path = tmp_dir.path().join("skill"); + let status = std::process::Command::new("git") + .args([ + "clone", + "--depth", + "1", + source, + clone_path.to_str().unwrap(), + ]) + .status(); + match status { + Ok(s) if s.success() => {} + Ok(_) => { + eprintln!("Failed to clone repository: {source}"); + std::process::exit(1); + } + Err(e) => { + eprintln!("Failed to run git: {e}"); + ui::hint("Make sure git is installed and available on your PATH."); + std::process::exit(1); + } + } + + // Reuse the local directory install logic on the cloned repo + let manifest_path = clone_path.join("skill.toml"); + if !manifest_path.exists() { + if openfang_skills::openclaw_compat::detect_openclaw_skill(&clone_path) { + println!("Detected OpenClaw skill format. Converting..."); + match openfang_skills::openclaw_compat::convert_openclaw_skill(&clone_path) { + Ok(manifest) => { + let dest = skills_dir.join(&manifest.skill.name); + copy_dir_recursive(&clone_path, &dest); + if let Err(e) = openfang_skills::openclaw_compat::write_openfang_manifest( + &dest, &manifest, + ) { + eprintln!("Failed to write manifest: {e}"); + std::process::exit(1); + } + println!("Installed OpenClaw skill: {}", manifest.skill.name); + } + Err(e) => { + eprintln!("Failed to convert OpenClaw skill: {e}"); + std::process::exit(1); + } + } + return; + } + eprintln!("No skill.toml found in cloned repository: {source}"); + std::process::exit(1); + } + + let toml_str = std::fs::read_to_string(&manifest_path).unwrap_or_else(|e| { + eprintln!("Error reading skill.toml: {e}"); + std::process::exit(1); + }); + let manifest: openfang_skills::SkillManifest = + toml::from_str(&toml_str).unwrap_or_else(|e| { + eprintln!("Error parsing skill.toml: {e}"); + std::process::exit(1); + }); + + let dest = skills_dir.join(&manifest.skill.name); + copy_dir_recursive(&clone_path, &dest); + println!( + "Installed skill: {} v{}", + manifest.skill.name, manifest.skill.version + ); } else { // Remote install from FangHub println!("Installing {source} from FangHub..."); diff --git a/crates/openfang-kernel/src/kernel.rs b/crates/openfang-kernel/src/kernel.rs index 699028a8..18108afe 100644 --- a/crates/openfang-kernel/src/kernel.rs +++ b/crates/openfang-kernel/src/kernel.rs @@ -1751,8 +1751,6 @@ impl OpenFangKernel { by_messages || by_tokens || by_quota }; - let tools = self.available_tools(agent_id); - let tools = entry.mode.filter_tools(tools); let driver = self.resolve_driver(&entry.manifest)?; // Look up model's actual context window from the catalog @@ -1777,6 +1775,31 @@ impl OpenFangKernel { } } + // Build workspace-aware skill snapshot BEFORE tool list and prompt building. + // Loading order: bundled → global (~/.openfang/skills) → workspace skills. + // Each layer overrides duplicates from the previous layer. (#851, #808) + let skill_snapshot = { + let mut snapshot = self + .skill_registry + .read() + .unwrap_or_else(|e| e.into_inner()) + .snapshot(); + if let Some(ref workspace) = manifest.workspace { + let ws_skills = workspace.join("skills"); + if ws_skills.exists() { + if let Err(e) = snapshot.load_workspace_skills(&ws_skills) { + warn!(agent_id = %agent_id, "Failed to load workspace skills (streaming): {e}"); + } + } + } + snapshot + }; + + // Use the workspace-aware snapshot for tool resolution so both global + // and workspace skill tools are visible to the LLM. + let tools = self.available_tools_with_registry(agent_id, Some(&skill_snapshot)); + let tools = entry.mode.filter_tools(tools); + // Build the structured system prompt via prompt_builder { let mcp_tool_count = self.mcp_tools.lock().map(|t| t.len()).unwrap_or(0); @@ -1807,8 +1830,8 @@ impl OpenFangKernel { base_system_prompt: manifest.model.system_prompt.clone(), granted_tools: tools.iter().map(|t| t.name.clone()).collect(), recalled_memories: vec![], - skill_summary: self.build_skill_summary(&manifest.skills), - skill_prompt_context: self.collect_prompt_context(&manifest.skills), + skill_summary: Self::build_skill_summary_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 { @@ -1918,21 +1941,8 @@ impl OpenFangKernel { } let messages_before = session.messages.len(); - let mut skill_snapshot = kernel_clone - .skill_registry - .read() - .unwrap_or_else(|e| e.into_inner()) - .snapshot(); - - // Load workspace-scoped skills (override global skills with same name) - if let Some(ref workspace) = manifest.workspace { - let ws_skills = workspace.join("skills"); - if ws_skills.exists() { - if let Err(e) = skill_snapshot.load_workspace_skills(&ws_skills) { - warn!(agent_id = %agent_id, "Failed to load workspace skills (streaming): {e}"); - } - } - } + // skill_snapshot was built before the spawn and moved into this + // closure — it already contains bundled + global + workspace skills. // Create a phase callback that emits PhaseChange events to WS/SSE clients let phase_tx = tx.clone(); @@ -2293,17 +2303,6 @@ impl OpenFangKernel { let messages_before = session.messages.len(); - let tools = self.available_tools(agent_id); - let tools = entry.mode.filter_tools(tools); - - info!( - agent = %entry.name, - agent_id = %agent_id, - tool_count = tools.len(), - tool_names = ?tools.iter().map(|t| t.name.as_str()).collect::>(), - "Tools selected for LLM request" - ); - // Apply model routing if configured (disabled in Stable mode) let mut manifest = entry.manifest.clone(); @@ -2321,6 +2320,39 @@ impl OpenFangKernel { } } + // Build workspace-aware skill snapshot BEFORE tool list and prompt building. + // Loading order: bundled → global (~/.openfang/skills) → workspace skills. + // Each layer overrides duplicates from the previous layer. (#851, #808) + let skill_snapshot = { + let mut snapshot = self + .skill_registry + .read() + .unwrap_or_else(|e| e.into_inner()) + .snapshot(); + if let Some(ref workspace) = manifest.workspace { + let ws_skills = workspace.join("skills"); + if ws_skills.exists() { + if let Err(e) = snapshot.load_workspace_skills(&ws_skills) { + warn!(agent_id = %agent_id, "Failed to load workspace skills: {e}"); + } + } + } + snapshot + }; + + // Use the workspace-aware snapshot for tool resolution so both global + // and workspace skill tools are visible to the LLM. + let tools = self.available_tools_with_registry(agent_id, Some(&skill_snapshot)); + let tools = entry.mode.filter_tools(tools); + + info!( + agent = %entry.name, + agent_id = %agent_id, + tool_count = tools.len(), + tool_names = ?tools.iter().map(|t| t.name.as_str()).collect::>(), + "Tools selected for LLM request" + ); + // Build the structured system prompt via prompt_builder { let mcp_tool_count = self.mcp_tools.lock().map(|t| t.len()).unwrap_or(0); @@ -2351,8 +2383,8 @@ impl OpenFangKernel { base_system_prompt: manifest.model.system_prompt.clone(), 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(&manifest.skills), - skill_prompt_context: self.collect_prompt_context(&manifest.skills), + skill_summary: Self::build_skill_summary_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 { @@ -2485,22 +2517,8 @@ impl OpenFangKernel { .map(|m| m.context_window as usize) }); - // Snapshot skill registry before async call (RwLockReadGuard is !Send) - let mut skill_snapshot = self - .skill_registry - .read() - .unwrap_or_else(|e| e.into_inner()) - .snapshot(); - - // Load workspace-scoped skills (override global skills with same name) - if let Some(ref workspace) = manifest.workspace { - let ws_skills = workspace.join("skills"); - if ws_skills.exists() { - if let Err(e) = skill_snapshot.load_workspace_skills(&ws_skills) { - warn!(agent_id = %agent_id, "Failed to load workspace skills: {e}"); - } - } - } + // skill_snapshot was already built above (before tool list and prompt) + // with bundled + global + workspace skills. Reuse it for the agent loop. // Build link context from user message (auto-extract URLs for the agent) let message_with_links = if let Some(link_ctx) = @@ -5100,6 +5118,21 @@ impl OpenFangKernel { /// If `capabilities.tools` is empty (or contains `"*"`), all tools are /// available (backwards compatible). fn available_tools(&self, agent_id: AgentId) -> Vec { + self.available_tools_with_registry(agent_id, None) + } + + /// Build the list of tools available to an agent, optionally using a + /// workspace-aware skill registry snapshot instead of the global registry. + /// + /// When `skill_snapshot` is `Some`, skill-provided tools are read from that + /// snapshot (which already includes global + workspace skills with correct + /// override priority). When `None`, falls back to `self.skill_registry` + /// (global-only, for diagnostic/non-agent callers). + fn available_tools_with_registry( + &self, + agent_id: AgentId, + skill_snapshot: Option<&openfang_skills::registry::SkillRegistry>, + ) -> Vec { let all_builtins = builtin_tool_definitions(); // Look up agent entry for profile, skill/MCP allowlists, and declared tools @@ -5160,7 +5193,15 @@ impl OpenFangKernel { // Step 2: Add skill-provided tools (filtered by agent's skill allowlist, // then by declared tools). - let skill_tools = { + // When a workspace-aware snapshot is provided, use it so that workspace + // skill overrides are reflected in the tool list sent to the LLM. + let skill_tools = if let Some(snapshot) = skill_snapshot { + if skill_allowlist.is_empty() { + snapshot.all_tool_definitions() + } else { + snapshot.tool_definitions_for_skills(&skill_allowlist) + } + } else { let registry = self .skill_registry .read() @@ -5272,11 +5313,24 @@ impl OpenFangKernel { /// Build a compact skill summary for the system prompt so the agent knows /// what extra capabilities are installed. + /// + /// Falls back to the global registry. Prefer `build_skill_summary_from` + /// with a workspace-aware snapshot for agent execution paths. + #[allow(dead_code)] fn build_skill_summary(&self, skill_allowlist: &[String]) -> String { let registry = self .skill_registry .read() .unwrap_or_else(|e| e.into_inner()); + Self::build_skill_summary_from(®istry, skill_allowlist) + } + + /// Build a compact skill summary using the provided registry (which may + /// include workspace skill overrides). + fn build_skill_summary_from( + registry: &openfang_skills::registry::SkillRegistry, + skill_allowlist: &[String], + ) -> String { let skills: Vec<_> = registry .list() .into_iter() @@ -5381,14 +5435,26 @@ impl OpenFangKernel { // inject_user_personalization() — logic moved to prompt_builder::build_user_section() + /// Collect prompt context from the global skill registry. + /// + /// Falls back to the global registry. Prefer `collect_prompt_context_from` + /// with a workspace-aware snapshot for agent execution paths. pub fn collect_prompt_context(&self, skill_allowlist: &[String]) -> String { - let mut context_parts = Vec::new(); - for skill in self + let registry = self .skill_registry .read() - .unwrap_or_else(|e| e.into_inner()) - .list() - { + .unwrap_or_else(|e| e.into_inner()); + Self::collect_prompt_context_from(®istry, skill_allowlist) + } + + /// Collect prompt context using the provided registry (which may include + /// workspace skill overrides). + fn collect_prompt_context_from( + registry: &openfang_skills::registry::SkillRegistry, + skill_allowlist: &[String], + ) -> String { + let mut context_parts = Vec::new(); + for skill in registry.list() { if skill.enabled && (skill_allowlist.is_empty() || skill_allowlist.contains(&skill.manifest.skill.name)) diff --git a/crates/openfang-runtime/src/drivers/anthropic.rs b/crates/openfang-runtime/src/drivers/anthropic.rs index 344b4d46..ad47d8b5 100644 --- a/crates/openfang-runtime/src/drivers/anthropic.rs +++ b/crates/openfang-runtime/src/drivers/anthropic.rs @@ -522,7 +522,8 @@ impl LlmDriver for AnthropicDriver { input_json, } => { let input: serde_json::Value = - serde_json::from_str(&input_json).unwrap_or_default(); + serde_json::from_str(&input_json) + .unwrap_or_else(|_| serde_json::json!({})); content.push(ContentBlock::ToolUse { id: id.clone(), name: name.clone(), @@ -553,6 +554,28 @@ impl LlmDriver for AnthropicDriver { } } +/// Ensure a `serde_json::Value` is a JSON object (dictionary). +/// +/// The Anthropic API requires `tool_use.input` to be a JSON object, never a +/// string, null, or other scalar. This helper handles: +/// - `Value::Object` → returned as-is +/// - `Value::String` → attempt to parse as JSON; if the result is an object, use it +/// - anything else (Null, Number, Bool, Array) → empty object `{}` +fn ensure_object(v: &serde_json::Value) -> serde_json::Value { + match v { + serde_json::Value::Object(_) => v.clone(), + serde_json::Value::String(s) => { + // The input may have been double-serialized (stored as a JSON string). + // Try to parse it back into a Value. + match serde_json::from_str::(s) { + Ok(parsed) if parsed.is_object() => parsed, + _ => serde_json::json!({}), + } + } + _ => serde_json::json!({}), + } +} + /// Convert an OpenFang Message to an Anthropic API message. fn convert_message(msg: &Message) -> ApiMessage { let role = match msg.role { @@ -582,7 +605,7 @@ fn convert_message(msg: &Message) -> ApiMessage { } => Some(ApiContentBlock::ToolUse { id: id.clone(), name: name.clone(), - input: input.clone(), + input: ensure_object(input), }), ContentBlock::ToolResult { tool_use_id, @@ -692,4 +715,63 @@ mod tests { assert_eq!(response.tool_calls[0].name, "web_search"); assert_eq!(response.usage.total(), 150); } + + #[test] + fn test_ensure_object_from_object() { + let obj = serde_json::json!({"key": "value"}); + assert_eq!(ensure_object(&obj), obj); + } + + #[test] + fn test_ensure_object_from_string() { + // Simulates double-serialized input (stored as JSON string) + let stringified = serde_json::Value::String(r#"{"query": "rust"}"#.to_string()); + let result = ensure_object(&stringified); + assert_eq!(result, serde_json::json!({"query": "rust"})); + } + + #[test] + fn test_ensure_object_from_null() { + let null = serde_json::Value::Null; + assert_eq!(ensure_object(&null), serde_json::json!({})); + } + + #[test] + fn test_ensure_object_from_non_object_string() { + // A string that parses to a non-object JSON value + let s = serde_json::Value::String("42".to_string()); + assert_eq!(ensure_object(&s), serde_json::json!({})); + } + + #[test] + fn test_ensure_object_from_invalid_json_string() { + let s = serde_json::Value::String("not json at all".to_string()); + assert_eq!(ensure_object(&s), serde_json::json!({})); + } + + #[test] + fn test_convert_message_normalizes_tool_input() { + // Simulate a ToolUse block with a stringified JSON input (legacy session data) + let msg = Message { + role: Role::Assistant, + content: MessageContent::Blocks(vec![ContentBlock::ToolUse { + id: "tu-1".to_string(), + name: "web_search".to_string(), + input: serde_json::Value::String(r#"{"query": "test"}"#.to_string()), + provider_metadata: None, + }]), + }; + let api_msg = convert_message(&msg); + if let ApiContent::Blocks(blocks) = api_msg.content { + match &blocks[0] { + ApiContentBlock::ToolUse { input, .. } => { + assert!(input.is_object(), "input should be an object, got: {input}"); + assert_eq!(input["query"], "test"); + } + _ => panic!("Expected ToolUse block"), + } + } else { + panic!("Expected Blocks content"); + } + } } diff --git a/crates/openfang-runtime/src/drivers/gemini.rs b/crates/openfang-runtime/src/drivers/gemini.rs index cd37cc1b..4cc26803 100644 --- a/crates/openfang-runtime/src/drivers/gemini.rs +++ b/crates/openfang-runtime/src/drivers/gemini.rs @@ -653,7 +653,20 @@ impl LlmDriver for GeminiDriver { return Err(LlmError::Api { status, message }); } - // Parse SSE stream + // Parse SSE stream — process line-by-line like the OpenAI driver. + // + // Gemini's `streamGenerateContent?alt=sse` endpoint sends + // standard SSE: each event is a `data: {...}` line followed by + // a blank line. The previous implementation looked for `\n\n` + // as the event delimiter, but many HTTP responses use `\r\n` + // line endings, so the actual delimiter is `\r\n\r\n` which + // `find("\n\n")` never matches — causing the entire stream to + // be silently buffered without extracting any events (zero + // TextDelta emissions → empty response → infinite retry loop). + // + // Fix: process the buffer one line at a time (splitting on + // `\n` and stripping trailing `\r`). A `data:` line is + // parsed immediately. Empty/blank lines are simply skipped. let mut buffer = String::new(); let mut text_content = String::new(); // Thought signature for accumulated text content (last one wins) @@ -662,22 +675,32 @@ impl LlmDriver for GeminiDriver { let mut fn_calls: Vec<(String, serde_json::Value, Option)> = Vec::new(); let mut finish_reason: Option = None; let mut usage = TokenUsage::default(); + let mut chunk_count: u32 = 0; + let mut sse_line_count: u32 = 0; let mut byte_stream = resp.bytes_stream(); while let Some(chunk_result) = byte_stream.next().await { let chunk = chunk_result.map_err(|e| LlmError::Http(e.to_string()))?; + chunk_count += 1; buffer.push_str(&String::from_utf8_lossy(&chunk)); - // Process complete SSE events (delimited by \n\n or \r\n\r\n) - while let Some(pos) = buffer.find("\n\n") { - let event_text = buffer[..pos].to_string(); - buffer = buffer[pos + 2..].to_string(); + // Process complete lines (handle both \r\n and \n endings) + while let Some(pos) = buffer.find('\n') { + let line = buffer[..pos].trim_end().to_string(); + buffer = buffer[pos + 1..].to_string(); - // Extract the data line (handle both "data: " and "data:" formats) - let data = event_text - .lines() - .find_map(|line| line.strip_prefix("data:").map(|d| d.trim_start())) - .unwrap_or(""); + // Skip empty lines and SSE comments + if line.is_empty() || line.starts_with(':') { + continue; + } + + sse_line_count += 1; + + // Extract the data payload (handle both "data: " and "data:" formats) + let data = match line.strip_prefix("data:") { + Some(d) => d.trim_start(), + None => continue, + }; if data.is_empty() { continue; @@ -685,7 +708,14 @@ impl LlmDriver for GeminiDriver { let json: GeminiResponse = match serde_json::from_str(data) { Ok(v) => v, - Err(_) => continue, + Err(e) => { + debug!( + error = %e, + data_preview = &data[..data.len().min(200)], + "Failed to parse Gemini SSE data line" + ); + continue; + } }; // Extract usage from each chunk (last one wins) @@ -768,6 +798,121 @@ impl LlmDriver for GeminiDriver { } } + // Process any remaining data left in the buffer after the stream + // ends (e.g. final chunk not terminated by a newline). + let remaining = buffer.trim(); + if !remaining.is_empty() { + if let Some(data) = remaining.strip_prefix("data:") { + let data = data.trim(); + if !data.is_empty() { + if let Ok(json) = serde_json::from_str::(data) { + if let Some(ref u) = json.usage_metadata { + usage.input_tokens = u.prompt_token_count; + usage.output_tokens = u.candidates_token_count; + } + for candidate in &json.candidates { + if let Some(fr) = &candidate.finish_reason { + finish_reason = Some(fr.clone()); + } + if let Some(ref content) = candidate.content { + for part in &content.parts { + match part { + GeminiPart::Text { + text, + thought_signature, + } => { + if !text.is_empty() { + text_content.push_str(text); + let _ = tx + .send(StreamEvent::TextDelta { + text: text.clone(), + }) + .await; + } + if thought_signature.is_some() { + text_thought_sig = thought_signature.clone(); + } + } + GeminiPart::FunctionCall { + function_call, + thought_signature, + } => { + let id = format!( + "call_{}", + uuid::Uuid::new_v4().simple() + ); + let _ = tx + .send(StreamEvent::ToolUseStart { + id: id.clone(), + name: function_call.name.clone(), + }) + .await; + let args_str = + serde_json::to_string(&function_call.args) + .unwrap_or_default(); + let _ = tx + .send(StreamEvent::ToolInputDelta { + text: args_str, + }) + .await; + let _ = tx + .send(StreamEvent::ToolUseEnd { + id, + name: function_call.name.clone(), + input: function_call.args.clone(), + }) + .await; + fn_calls.push(( + function_call.name.clone(), + function_call.args.clone(), + thought_signature.clone(), + )); + } + GeminiPart::Thought { ref text, .. } => { + if !text.is_empty() { + let _ = tx + .send(StreamEvent::ThinkingDelta { + text: text.clone(), + }) + .await; + } + } + _ => {} + } + } + } + } + } + } + } + } + + // Log stream summary for diagnostics (mirrors OpenAI driver) + let is_empty_stream = text_content.is_empty() + && fn_calls.is_empty() + && usage.input_tokens == 0 + && usage.output_tokens == 0; + if is_empty_stream { + warn!( + chunks = chunk_count, + sse_lines = sse_line_count, + finish = ?finish_reason, + buffer_remaining = buffer.len(), + "Gemini SSE stream returned empty: 0 content, 0 tokens — likely a silently failed request" + ); + } else { + debug!( + chunks = chunk_count, + sse_lines = sse_line_count, + text_len = text_content.len(), + tool_count = fn_calls.len(), + finish = ?finish_reason, + input_tokens = usage.input_tokens, + output_tokens = usage.output_tokens, + "Gemini SSE stream completed" + ); + } + // Build final response let mut content = Vec::new(); let mut tool_calls = Vec::new(); diff --git a/crates/openfang-runtime/src/drivers/openai.rs b/crates/openfang-runtime/src/drivers/openai.rs index b6dd5d21..1a406b80 100644 --- a/crates/openfang-runtime/src/drivers/openai.rs +++ b/crates/openfang-runtime/src/drivers/openai.rs @@ -683,7 +683,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_default(); + 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(), @@ -1312,7 +1313,8 @@ impl LlmDriver for OpenAIDriver { } for (id, name, arguments) in &tool_accum { - let input: serde_json::Value = serde_json::from_str(arguments).unwrap_or_default(); + 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(), @@ -1322,14 +1324,14 @@ impl LlmDriver for OpenAIDriver { tool_calls.push(ToolCall { id: id.clone(), name: name.clone(), - input, + input: input.clone(), }); let _ = tx .send(StreamEvent::ToolUseEnd { id: id.clone(), name: name.clone(), - input: serde_json::from_str(arguments).unwrap_or_default(), + input, }) .await; } diff --git a/crates/openfang-runtime/src/model_catalog.rs b/crates/openfang-runtime/src/model_catalog.rs index 0cbe4d1f..45adbb87 100644 --- a/crates/openfang-runtime/src/model_catalog.rs +++ b/crates/openfang-runtime/src/model_catalog.rs @@ -108,13 +108,57 @@ impl ModelCatalog { } /// Find a model by its canonical ID, display name, or alias. + /// + /// When multiple models match case-insensitively (e.g. a builtin `qwen3-30b-a3b` + /// and a custom `Qwen3-30B-A3B`), user-defined entries (Custom or Local tier) + /// take priority. This ensures models from `custom_models.json` or dynamically + /// discovered local models are not shadowed by builtins that happen to share the + /// same lowercased name (#856). pub fn find_model(&self, id_or_alias: &str) -> Option<&ModelCatalogEntry> { let lower = id_or_alias.to_lowercase(); - // Direct ID match first - if let Some(entry) = self.models.iter().find(|m| m.id.to_lowercase() == lower) { + + // Single scan: prefer user-defined models (Custom/Local tier) over builtins. + // + // Priority order: + // 1. User-defined entry with exact-case ID match + // 2. User-defined entry with case-insensitive ID match + // 3. Builtin entry with exact-case ID match + // 4. Builtin entry with case-insensitive ID match + // + // This ensures that custom models from custom_models.json and dynamically + // discovered local models are never shadowed by builtins that share the same + // lowercased name, regardless of how the caller cased the search term. + let mut user_ci: Option<&ModelCatalogEntry> = None; + let mut builtin_exact: Option<&ModelCatalogEntry> = None; + let mut builtin_ci: Option<&ModelCatalogEntry> = None; + + for m in &self.models { + if m.id.to_lowercase() != lower { + continue; + } + let is_user_defined = m.tier == ModelTier::Custom || m.tier == ModelTier::Local; + let is_exact = m.id == id_or_alias; + + match (is_user_defined, is_exact) { + (true, true) => return Some(m), // Best possible: user-defined + exact + (true, false) if user_ci.is_none() => user_ci = Some(m), + (false, true) if builtin_exact.is_none() => builtin_exact = Some(m), + (false, false) if builtin_ci.is_none() => builtin_ci = Some(m), + _ => {} + } + } + + if let Some(entry) = user_ci { return Some(entry); } - // Display-name match for dashboard/UI payloads that send labels. + if let Some(entry) = builtin_exact { + return Some(entry); + } + if let Some(entry) = builtin_ci { + return Some(entry); + } + + // 3. Display-name match for dashboard/UI payloads that send labels. if let Some(entry) = self .models .iter() @@ -122,7 +166,7 @@ impl ModelCatalog { { return Some(entry); } - // Alias resolution + // 4. Alias resolution if let Some(canonical) = self.aliases.get(&lower) { return self.models.iter().find(|m| m.id == *canonical); } @@ -291,6 +335,10 @@ impl ModelCatalog { /// /// Returns `true` if the model was added, `false` if a model with the same /// ID **and** provider already exists (case-insensitive). + /// + /// The entry's tier is forced to [`ModelTier::Custom`] so that user-defined + /// models are always preferred over builtins with the same lowercased name + /// (see `find_model` priority logic, #856). pub fn add_custom_model(&mut self, entry: ModelCatalogEntry) -> bool { let lower_id = entry.id.to_lowercase(); let lower_provider = entry.provider.to_lowercase(); @@ -302,6 +350,8 @@ impl ModelCatalog { return false; } let provider = entry.provider.clone(); + let mut entry = entry; + entry.tier = ModelTier::Custom; self.models.push(entry); // Update provider model count @@ -4206,4 +4256,92 @@ mod tests { assert!(entry.supports_tools); assert!(entry.supports_vision); } + + /// Regression test for #856: custom models with case-sensitive names must not be + /// shadowed by builtin models that share the same lowercased ID. + /// + /// When a user deploys `Qwen3-30B-A3B` via vLLM and adds it to custom_models.json, + /// find_model should return the custom entry (provider "vllm"), not the builtin + /// entry (provider "qwen") whose id is `qwen3-30b-a3b`. + #[test] + fn test_custom_model_not_shadowed_by_builtin_856() { + let mut catalog = ModelCatalog::new(); + + // Verify the builtin exists first + let builtin = catalog.find_model("qwen3-30b-a3b").unwrap(); + assert_eq!(builtin.provider, "qwen"); + + // Add a custom model with a case-sensitive name on a different provider + let added = catalog.add_custom_model(ModelCatalogEntry { + id: "Qwen3-30B-A3B".into(), + display_name: "Qwen3 30B (Local vLLM)".into(), + provider: "vllm".into(), + tier: ModelTier::Balanced, // user might not set tier explicitly + context_window: 32_768, + max_output_tokens: 4_096, + input_cost_per_m: 0.0, + output_cost_per_m: 0.0, + supports_tools: true, + supports_vision: false, + supports_streaming: true, + aliases: vec![], + }); + assert!(added, "custom model should be added (different provider)"); + + // add_custom_model should force Custom tier + let custom = catalog + .list_models() + .iter() + .find(|m| m.id == "Qwen3-30B-A3B") + .unwrap(); + assert_eq!(custom.tier, ModelTier::Custom); + + // Exact-case lookup must find the custom entry, not the builtin + let found = catalog.find_model("Qwen3-30B-A3B").unwrap(); + assert_eq!(found.id, "Qwen3-30B-A3B"); + assert_eq!(found.provider, "vllm"); + assert_eq!(found.tier, ModelTier::Custom); + + // Lowercase lookup: builtin "qwen3-30b-a3b" (tier Fast) gets an exact-case match, + // but the custom "Qwen3-30B-A3B" (tier Custom) gets a case-insensitive match. + // With our fix, the user-defined entry (Custom tier) wins even when there's + // an exact-case builtin match, because user-defined entries always take priority. + let lower_found = catalog.find_model("qwen3-30b-a3b").unwrap(); + assert_eq!(lower_found.provider, "vllm"); + assert_eq!(lower_found.tier, ModelTier::Custom); + } + + /// Verify that find_model's exact-case match takes priority over case-insensitive. + #[test] + fn test_find_model_exact_case_priority() { + let catalog = ModelCatalog::new(); + // MiniMax-M2.5 exists as a builtin with exact case "MiniMax-M2.5" + let entry = catalog.find_model("MiniMax-M2.5").unwrap(); + assert_eq!(entry.id, "MiniMax-M2.5"); + assert_eq!(entry.provider, "minimax"); + + // Case-insensitive still works + let lower = catalog.find_model("minimax-m2.5").unwrap(); + assert_eq!(lower.id, "MiniMax-M2.5"); + } + + /// Verify that dynamically discovered local models (Local tier) are preferred + /// over builtins in case-insensitive lookups. + #[test] + fn test_discovered_local_model_preferred() { + let mut catalog = ModelCatalog::new(); + + // merge_discovered_models adds models with Local tier + catalog.merge_discovered_models("ollama", &["Custom-Model-7B".to_string()]); + + // Verify it was added + let found = catalog.find_model("Custom-Model-7B").unwrap(); + assert_eq!(found.tier, ModelTier::Local); + assert_eq!(found.provider, "ollama"); + + // Case-insensitive lookup should prefer the Local-tier entry + let lower = catalog.find_model("custom-model-7b").unwrap(); + assert_eq!(lower.tier, ModelTier::Local); + assert_eq!(lower.provider, "ollama"); + } } diff --git a/crates/openfang-skills/src/registry.rs b/crates/openfang-skills/src/registry.rs index 5f3a9d9e..8100b9ef 100644 --- a/crates/openfang-skills/src/registry.rs +++ b/crates/openfang-skills/src/registry.rs @@ -17,6 +17,8 @@ pub struct SkillRegistry { skills_dir: PathBuf, /// When true, no new skills can be loaded (Stable mode). frozen: bool, + /// Number of workspace skills blocked for critical prompt injection. + blocked_skills_count: usize, } impl SkillRegistry { @@ -26,6 +28,7 @@ impl SkillRegistry { skills: HashMap::new(), skills_dir, frozen: false, + blocked_skills_count: 0, } } @@ -38,6 +41,7 @@ impl SkillRegistry { skills: self.skills.clone(), skills_dir: self.skills_dir.clone(), frozen: self.frozen, + blocked_skills_count: self.blocked_skills_count, } } @@ -53,6 +57,11 @@ impl SkillRegistry { self.frozen } + /// Return the number of workspace skills blocked for critical prompt injection. + pub fn blocked_count(&self) -> usize { + self.blocked_skills_count + } + /// Load all bundled skills (compile-time embedded SKILL.md files). /// /// Called before `load_all()` so that user-installed skills with the same name @@ -331,6 +340,7 @@ impl SkillRegistry { skill = %converted.manifest.skill.name, "BLOCKED workspace skill: critical prompt injection patterns" ); + self.blocked_skills_count += 1; continue; } @@ -549,4 +559,163 @@ input_schema = {{ type = "object" }} // Verify that skill.toml was written assert!(skill_dir.join("skill.toml").exists()); } + + /// #851: Global skills should be visible via snapshot even without workspace skills. + #[test] + fn test_snapshot_includes_global_skills() { + let global_dir = TempDir::new().unwrap(); + create_test_skill(global_dir.path(), "global-skill"); + + let mut registry = SkillRegistry::new(global_dir.path().to_path_buf()); + registry.load_all().unwrap(); + assert_eq!(registry.count(), 1); + + // Take a snapshot (simulates what the kernel does before agent execution) + let snapshot = registry.snapshot(); + assert_eq!(snapshot.count(), 1, "Snapshot must include global skills"); + assert!( + snapshot.get("global-skill").is_some(), + "Global skill must be accessible in snapshot" + ); + } + + /// #808: Workspace skills must override global skills with the same name. + #[test] + fn test_workspace_skill_overrides_global() { + let global_dir = TempDir::new().unwrap(); + let ws_dir = TempDir::new().unwrap(); + + // Create a global skill with one description + let global_skill_dir = global_dir.path().join("shared-skill"); + std::fs::create_dir_all(&global_skill_dir).unwrap(); + std::fs::write( + global_skill_dir.join("skill.toml"), + r#" +[skill] +name = "shared-skill" +version = "1.0.0" +description = "Global version" + +[runtime] +type = "python" +entry = "main.py" + +[[tools.provided]] +name = "shared_tool" +description = "Global tool" +input_schema = { type = "object" } +"#, + ) + .unwrap(); + + // Create a workspace skill with the same name but different description + let ws_skill_dir = ws_dir.path().join("shared-skill"); + std::fs::create_dir_all(&ws_skill_dir).unwrap(); + std::fs::write( + ws_skill_dir.join("skill.toml"), + r#" +[skill] +name = "shared-skill" +version = "2.0.0" +description = "Workspace override version" + +[runtime] +type = "python" +entry = "main.py" + +[[tools.provided]] +name = "shared_tool" +description = "Workspace tool" +input_schema = { type = "object" } +"#, + ) + .unwrap(); + + // Load global skills + let mut registry = SkillRegistry::new(global_dir.path().to_path_buf()); + registry.load_all().unwrap(); + assert_eq!(registry.count(), 1); + assert_eq!( + registry.get("shared-skill").unwrap().manifest.skill.description, + "Global version" + ); + + // Take a snapshot and load workspace skills (simulates kernel agent path) + let mut snapshot = registry.snapshot(); + 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, + "Workspace override version", + "Workspace skill must override global skill (#808)" + ); + assert_eq!( + snapshot.get("shared-skill").unwrap().manifest.skill.version, + "2.0.0" + ); + + // Tool definitions should come from workspace version + let tools = snapshot.all_tool_definitions(); + assert_eq!(tools.len(), 1); + assert_eq!(tools[0].description, "Workspace tool"); + } + + /// #851 + #808: Snapshot with both global and workspace skills, where workspace + /// overrides one global skill but a second global skill remains. + #[test] + fn test_snapshot_global_plus_workspace_merge() { + let global_dir = TempDir::new().unwrap(); + let ws_dir = TempDir::new().unwrap(); + + // Two global skills + create_test_skill(global_dir.path(), "alpha"); + create_test_skill(global_dir.path(), "beta"); + + // Workspace overrides only "alpha" + let ws_alpha = ws_dir.path().join("alpha"); + std::fs::create_dir_all(&ws_alpha).unwrap(); + std::fs::write( + ws_alpha.join("skill.toml"), + r#" +[skill] +name = "alpha" +version = "9.0.0" +description = "Workspace alpha" + +[runtime] +type = "python" +entry = "main.py" + +[[tools.provided]] +name = "alpha_tool" +description = "Workspace alpha tool" +input_schema = { type = "object" } +"#, + ) + .unwrap(); + + let mut registry = SkillRegistry::new(global_dir.path().to_path_buf()); + registry.load_all().unwrap(); + assert_eq!(registry.count(), 2); + + let mut snapshot = registry.snapshot(); + snapshot.load_workspace_skills(ws_dir.path()).unwrap(); + + // Both skills present + assert_eq!(snapshot.count(), 2); + // "alpha" is overridden + assert_eq!( + snapshot.get("alpha").unwrap().manifest.skill.version, + "9.0.0", + "Workspace should override alpha" + ); + // "beta" retains global version + assert_eq!( + snapshot.get("beta").unwrap().manifest.skill.version, + "0.1.0", + "Global beta should remain unchanged" + ); + } }