admin/openfang
1
0
Fork 0
mirror of https://github.com/RightNow-AI/openfang.git synced 2026-07-29 22:22:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: resolve 7 more bugs (#825, #828, #856, #770, #774, #851/#808, #785)

Browse Source 
- #825: Doctor now surfaces blocked workspace skills count in injection scan
- #828: Skill install detects Git URLs (https://, git@) and clones before install
- #856: Custom model names preserved — user-defined models take priority over builtins
- #770: Dashboard WS streaming now triggers Alpine.js reactivity via splice()
- #774: tool_use.input always normalized to JSON object (fixes Anthropic API errors)
- #851/#808: Global skills loaded for all agents; workspace skills properly override globals
- #785: Gemini streaming SSE parser handles \r\n line endings (fixes empty response loop)

All 2,186 tests passing. Live tested with daemon.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
jaberjaber23
2026-03-27 04:04:17 +03:00
co-authored by Claude Opus 4.6
parent e7b9143423
commit 9b7496947b
11 changed files with 811 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -3906,6 +3906,7 @@ dependencies = [
"reqwest 0.12.28",
"serde",
"serde_json",
"tempfile",
"tokio",
"toml 0.8.2",
"tracing",
crates/openfang-api/static/js/api.js
+18 -7
View File
@@ -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();
};
crates/openfang-api/static/js/pages/chat.js
+28 -8
View File
@@ -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 = '<details><summary>Reasoning...</summary>\n\n' + phaseMsg._reasoning + '</details>';
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;
crates/openfang-cli/Cargo.toml
+1
View File
@@ -31,3 +31,4 @@ openfang-runtime = { path = "../openfang-runtime" }
uuid = { workspace = true }
ratatui = { workspace = true }
colored = { workspace = true }
tempfile = { workspace = true }
crates/openfang-cli/src/main.rs
+86 -2
View File
@@ -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...");
crates/openfang-kernel/src/kernel.rs
+120 -54
View File
@@ -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::<Vec<_>>(),
"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::<Vec<_>>(),
"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<ToolDefinition> {
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<ToolDefinition> {
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(&registry, 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(&registry, 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))
crates/openfang-runtime/src/drivers/anthropic.rs
+84 -2
View File
@@ -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::<serde_json::Value>(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");
}
}
}
crates/openfang-runtime/src/drivers/gemini.rs
+156 -11
View File
@@ -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<String>)> = Vec::new();
let mut finish_reason: Option<String> = 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::<GeminiResponse>(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();
crates/openfang-runtime/src/drivers/openai.rs
+6 -4
View File
@@ -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;
}
crates/openfang-runtime/src/model_catalog.rs
+142 -4
View File
@@ -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");
}
}
crates/openfang-skills/src/registry.rs
+169
View File
@@ -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"
);
}
}