From 45e06b9badbafa31b1587f3cbe328827b7381d9a Mon Sep 17 00:00:00 2001 From: jaberjaber23 Date: Thu, 5 Mar 2026 21:35:37 +0300 Subject: [PATCH] channel resilience --- Cargo.lock | 28 +-- Cargo.toml | 2 +- crates/openfang-api/Cargo.toml | 2 +- crates/openfang-api/src/channel_bridge.rs | 29 +++ crates/openfang-api/src/routes.rs | 127 ++++++++++- crates/openfang-api/static/index_body.html | 30 +++ crates/openfang-api/static/js/pages/agents.js | 48 +++- crates/openfang-kernel/src/registry.rs | 15 ++ crates/openfang-runtime/src/agent_loop.rs | 210 ++++++++++++++++++ 9 files changed, 471 insertions(+), 20 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8652fc3d..fb58a377 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3866,7 +3866,7 @@ dependencies = [ [[package]] name = "openfang-api" -version = "0.3.20" +version = "0.3.22" dependencies = [ "async-trait", "axum", @@ -3903,7 +3903,7 @@ dependencies = [ [[package]] name = "openfang-channels" -version = "0.3.20" +version = "0.3.22" dependencies = [ "async-trait", "axum", @@ -3934,7 +3934,7 @@ dependencies = [ [[package]] name = "openfang-cli" -version = "0.3.20" +version = "0.3.22" dependencies = [ "clap", "clap_complete", @@ -3961,7 +3961,7 @@ dependencies = [ [[package]] name = "openfang-desktop" -version = "0.3.20" +version = "0.3.22" dependencies = [ "axum", "open", @@ -3987,7 +3987,7 @@ dependencies = [ [[package]] name = "openfang-extensions" -version = "0.3.20" +version = "0.3.22" dependencies = [ "aes-gcm", "argon2", @@ -4015,7 +4015,7 @@ dependencies = [ [[package]] name = "openfang-hands" -version = "0.3.20" +version = "0.3.22" dependencies = [ "chrono", "dashmap", @@ -4032,7 +4032,7 @@ dependencies = [ [[package]] name = "openfang-kernel" -version = "0.3.20" +version = "0.3.22" dependencies = [ "async-trait", "chrono", @@ -4068,7 +4068,7 @@ dependencies = [ [[package]] name = "openfang-memory" -version = "0.3.20" +version = "0.3.22" dependencies = [ "async-trait", "chrono", @@ -4087,7 +4087,7 @@ dependencies = [ [[package]] name = "openfang-migrate" -version = "0.3.20" +version = "0.3.22" dependencies = [ "chrono", "dirs 6.0.0", @@ -4106,7 +4106,7 @@ dependencies = [ [[package]] name = "openfang-runtime" -version = "0.3.20" +version = "0.3.22" dependencies = [ "anyhow", "async-trait", @@ -4138,7 +4138,7 @@ dependencies = [ [[package]] name = "openfang-skills" -version = "0.3.20" +version = "0.3.22" dependencies = [ "chrono", "hex", @@ -4161,7 +4161,7 @@ dependencies = [ [[package]] name = "openfang-types" -version = "0.3.20" +version = "0.3.22" dependencies = [ "async-trait", "chrono", @@ -4180,7 +4180,7 @@ dependencies = [ [[package]] name = "openfang-wire" -version = "0.3.20" +version = "0.3.22" dependencies = [ "async-trait", "chrono", @@ -8802,7 +8802,7 @@ checksum = "b9cc00251562a284751c9973bace760d86c0276c471b4be569fe6b068ee97a56" [[package]] name = "xtask" -version = "0.3.20" +version = "0.3.22" [[package]] name = "yoke" diff --git a/Cargo.toml b/Cargo.toml index fe1451da..939bcbef 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,7 +18,7 @@ members = [ ] [workspace.package] -version = "0.3.22" +version = "0.3.23" edition = "2021" license = "Apache-2.0 OR MIT" repository = "https://github.com/RightNow-AI/openfang" diff --git a/crates/openfang-api/Cargo.toml b/crates/openfang-api/Cargo.toml index fbcda731..28b080f1 100644 --- a/crates/openfang-api/Cargo.toml +++ b/crates/openfang-api/Cargo.toml @@ -34,9 +34,9 @@ tokio-stream = { workspace = true } subtle = { workspace = true } base64 = { workspace = true } socket2 = { workspace = true } +reqwest = { workspace = true } [dev-dependencies] tokio-test = { workspace = true } -reqwest = { workspace = true } tempfile = { workspace = true } uuid = { workspace = true } diff --git a/crates/openfang-api/src/channel_bridge.rs b/crates/openfang-api/src/channel_bridge.rs index 9285d060..b7218067 100644 --- a/crates/openfang-api/src/channel_bridge.rs +++ b/crates/openfang-api/src/channel_bridge.rs @@ -1621,6 +1621,35 @@ pub async fn reload_channels_from_disk( *guard = None; } + // Re-read secrets.env so new API tokens are available in std::env + let secrets_path = state.kernel.config.home_dir.join("secrets.env"); + if secrets_path.exists() { + if let Ok(content) = std::fs::read_to_string(&secrets_path) { + for line in content.lines() { + let trimmed = line.trim(); + if trimmed.is_empty() || trimmed.starts_with('#') { + continue; + } + if let Some(eq_pos) = trimmed.find('=') { + let key = trimmed[..eq_pos].trim(); + let mut value = trimmed[eq_pos + 1..].trim().to_string(); + if !key.is_empty() { + // Strip matching quotes + if ((value.starts_with('"') && value.ends_with('"')) + || (value.starts_with('\'') && value.ends_with('\''))) + && value.len() >= 2 + { + value = value[1..value.len() - 1].to_string(); + } + // Always overwrite — the file is the source of truth after dashboard edits + std::env::set_var(key, &value); + } + } + } + info!("Reloaded secrets.env for channel hot-reload"); + } + } + // Re-read config from disk let config_path = state.kernel.config.home_dir.join("config.toml"); let fresh_config = openfang_kernel::config::load_config(Some(&config_path)); diff --git a/crates/openfang-api/src/routes.rs b/crates/openfang-api/src/routes.rs index b6f7dde8..89353f39 100644 --- a/crates/openfang-api/src/routes.rs +++ b/crates/openfang-api/src/routes.rs @@ -1002,6 +1002,7 @@ pub async fn get_agent( "skills_mode": if entry.manifest.skills.is_empty() { "all" } else { "allowlist" }, "mcp_servers": entry.manifest.mcp_servers, "mcp_servers_mode": if entry.manifest.mcp_servers.is_empty() { "all" } else { "allowlist" }, + "fallback_models": entry.manifest.fallback_models, })), ) } @@ -2161,8 +2162,15 @@ pub async fn remove_channel( } } -/// POST /api/channels/{name}/test — Basic connectivity check for a channel. -pub async fn test_channel(Path(name): Path) -> impl IntoResponse { +/// POST /api/channels/{name}/test — Connectivity check + optional live test message. +/// +/// Accepts an optional JSON body with `channel_id` (for Discord/Slack) or `chat_id` +/// (for Telegram). When provided, sends a real test message to verify the bot can +/// post to that channel. +pub async fn test_channel( + Path(name): Path, + raw_body: axum::body::Bytes, +) -> impl IntoResponse { let meta = match find_channel_meta(&name) { Some(m) => m, None => { @@ -2195,15 +2203,112 @@ pub async fn test_channel(Path(name): Path) -> impl IntoResponse { ); } + // If a target channel/chat ID is provided, send a real test message + let body: serde_json::Value = if raw_body.is_empty() { + serde_json::Value::Null + } else { + serde_json::from_slice(&raw_body).unwrap_or(serde_json::Value::Null) + }; + let target = body + .get("channel_id") + .or_else(|| body.get("chat_id")) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + if let Some(target_id) = target { + match send_channel_test_message(&name, &target_id).await { + Ok(()) => { + return ( + StatusCode::OK, + Json(serde_json::json!({ + "status": "ok", + "message": format!("Test message sent to {} channel {}.", meta.display_name, target_id) + })), + ); + } + Err(e) => { + return ( + StatusCode::OK, + Json(serde_json::json!({ + "status": "error", + "message": format!("Credentials valid but failed to send test message: {e}") + })), + ); + } + } + } + ( StatusCode::OK, Json(serde_json::json!({ "status": "ok", - "message": format!("All required credentials for {} are set.", meta.display_name) + "message": format!("All required credentials for {} are set. Provide channel_id or chat_id to send a test message.", meta.display_name) })), ) } +/// Send a real test message to a specific channel/chat on the given platform. +async fn send_channel_test_message(channel_name: &str, target_id: &str) -> Result<(), String> { + let client = reqwest::Client::new(); + let test_msg = "OpenFang test message — your channel is connected!"; + + match channel_name { + "discord" => { + let token = std::env::var("DISCORD_BOT_TOKEN") + .map_err(|_| "DISCORD_BOT_TOKEN not set".to_string())?; + let url = format!("https://discord.com/api/v10/channels/{target_id}/messages"); + let resp = client + .post(&url) + .header("Authorization", format!("Bot {token}")) + .json(&serde_json::json!({ "content": test_msg })) + .send() + .await + .map_err(|e| format!("HTTP request failed: {e}"))?; + if !resp.status().is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(format!("Discord API error: {body}")); + } + } + "telegram" => { + let token = std::env::var("TELEGRAM_BOT_TOKEN") + .map_err(|_| "TELEGRAM_BOT_TOKEN not set".to_string())?; + let url = format!("https://api.telegram.org/bot{token}/sendMessage"); + let resp = client + .post(&url) + .json(&serde_json::json!({ "chat_id": target_id, "text": test_msg })) + .send() + .await + .map_err(|e| format!("HTTP request failed: {e}"))?; + if !resp.status().is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(format!("Telegram API error: {body}")); + } + } + "slack" => { + let token = std::env::var("SLACK_BOT_TOKEN") + .map_err(|_| "SLACK_BOT_TOKEN not set".to_string())?; + let url = "https://slack.com/api/chat.postMessage"; + let resp = client + .post(url) + .header("Authorization", format!("Bearer {token}")) + .json(&serde_json::json!({ "channel": target_id, "text": test_msg })) + .send() + .await + .map_err(|e| format!("HTTP request failed: {e}"))?; + if !resp.status().is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(format!("Slack API error: {body}")); + } + } + _ => { + return Err(format!( + "Live test messaging not supported for {channel_name}. Credentials are valid." + )); + } + } + Ok(()) +} + /// POST /api/channels/reload — Manually trigger a channel hot-reload from disk config. pub async fn reload_channels(State(state): State>) -> impl IntoResponse { match crate::channel_bridge::reload_channels_from_disk(&state).await { @@ -7612,6 +7717,7 @@ pub struct PatchAgentConfigRequest { pub provider: Option, pub api_key_env: Option, pub base_url: Option, + pub fallback_models: Option>, } /// PATCH /api/agents/{id}/config — Hot-update agent name, description, system prompt, and identity. @@ -7812,6 +7918,21 @@ pub async fn patch_agent_config( } } + // Update fallback model chain + if let Some(fallbacks) = req.fallback_models { + if state + .kernel + .registry + .update_fallback_models(agent_id, fallbacks) + .is_err() + { + return ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({"error": "Agent not found"})), + ); + } + } + // Persist updated manifest to database so changes survive restart if let Some(entry) = state.kernel.registry.get(agent_id) { if let Err(e) = state.kernel.memory.save_agent(&entry) { diff --git a/crates/openfang-api/static/index_body.html b/crates/openfang-api/static/index_body.html index 49dc964c..6aeae4b8 100644 --- a/crates/openfang-api/static/index_body.html +++ b/crates/openfang-api/static/index_body.html @@ -912,6 +912,36 @@
Created
+ + +
+ Fallbacks +
+ + + + +
+
diff --git a/crates/openfang-api/static/js/pages/agents.js b/crates/openfang-api/static/js/pages/agents.js index 777c9d52..2313ee30 100644 --- a/crates/openfang-api/static/js/pages/agents.js +++ b/crates/openfang-api/static/js/pages/agents.js @@ -63,6 +63,9 @@ function agentsPage() { editingModel: false, newModelValue: '', modelSaving: false, + // -- Fallback chain -- + editingFallback: false, + newFallbackValue: '', // -- Templates state -- tplTemplates: [], @@ -316,12 +319,15 @@ function agentsPage() { OpenFangAPI.wsDisconnect(); }, - showDetail(agent) { + async showDetail(agent) { this.detailAgent = agent; + this.detailAgent._fallbacks = []; this.detailTab = 'info'; this.agentFiles = []; this.editingFile = null; this.fileContent = ''; + this.editingFallback = false; + this.newFallbackValue = ''; this.configForm = { name: agent.name || '', system_prompt: agent.system_prompt || '', @@ -331,6 +337,11 @@ function agentsPage() { vibe: (agent.identity && agent.identity.vibe) || '' }; this.showDetailModal = true; + // Fetch full agent detail to get fallback_models + try { + var full = await OpenFangAPI.get('/api/agents/' + agent.id); + this.detailAgent._fallbacks = full.fallback_models || []; + } catch(e) { /* ignore */ } }, killAgent(agent) { @@ -601,6 +612,41 @@ function agentsPage() { this.modelSaving = false; }, + // ── Fallback model chain ── + async addFallback() { + if (!this.detailAgent || !this.newFallbackValue.trim()) return; + var parts = this.newFallbackValue.trim().split('/'); + var provider = parts.length > 1 ? parts[0] : this.detailAgent.model_provider; + var model = parts.length > 1 ? parts.slice(1).join('/') : parts[0]; + if (!this.detailAgent._fallbacks) this.detailAgent._fallbacks = []; + this.detailAgent._fallbacks.push({ provider: provider, model: model }); + try { + await OpenFangAPI.patch('/api/agents/' + this.detailAgent.id + '/config', { + fallback_models: this.detailAgent._fallbacks + }); + OpenFangToast.success('Fallback added: ' + provider + '/' + model); + } catch(e) { + OpenFangToast.error('Failed to save fallbacks: ' + e.message); + this.detailAgent._fallbacks.pop(); + } + this.editingFallback = false; + this.newFallbackValue = ''; + }, + + async removeFallback(idx) { + if (!this.detailAgent || !this.detailAgent._fallbacks) return; + var removed = this.detailAgent._fallbacks.splice(idx, 1); + try { + await OpenFangAPI.patch('/api/agents/' + this.detailAgent.id + '/config', { + fallback_models: this.detailAgent._fallbacks + }); + OpenFangToast.success('Fallback removed'); + } catch(e) { + OpenFangToast.error('Failed to save fallbacks: ' + e.message); + this.detailAgent._fallbacks.splice(idx, 0, removed[0]); + } + }, + // ── Tool filters ── async loadToolFilters() { if (!this.detailAgent) return; diff --git a/crates/openfang-kernel/src/registry.rs b/crates/openfang-kernel/src/registry.rs index 427623a1..a21e13fa 100644 --- a/crates/openfang-kernel/src/registry.rs +++ b/crates/openfang-kernel/src/registry.rs @@ -177,6 +177,21 @@ impl AgentRegistry { Ok(()) } + /// Update an agent's fallback model chain. + pub fn update_fallback_models( + &self, + id: AgentId, + fallback_models: Vec, + ) -> OpenFangResult<()> { + let mut entry = self + .agents + .get_mut(&id) + .ok_or_else(|| OpenFangError::AgentNotFound(id.to_string()))?; + entry.manifest.fallback_models = fallback_models; + entry.last_active = chrono::Utc::now(); + Ok(()) + } + /// Update an agent's skill allowlist. pub fn update_skills(&self, id: AgentId, skills: Vec) -> OpenFangResult<()> { let mut entry = self diff --git a/crates/openfang-runtime/src/agent_loop.rs b/crates/openfang-runtime/src/agent_loop.rs index 144712e8..11fa8ea1 100644 --- a/crates/openfang-runtime/src/agent_loop.rs +++ b/crates/openfang-runtime/src/agent_loop.rs @@ -1835,6 +1835,136 @@ fn recover_text_tool_calls(text: &str, available_tools: &[ToolDefinition]) -> Ve }); } + // Pattern 3: TOOL_NAME{JSON} (Qwen / DeepSeek variant) + search_from = 0; + while let Some(start) = text[search_from..].find("") { + let abs_start = search_from + start; + let after_tag = abs_start + "".len(); + + let Some(close_offset) = text[after_tag..].find("") else { + search_from = after_tag; + continue; + }; + let inner = &text[after_tag..after_tag + close_offset]; + search_from = after_tag + close_offset + "".len(); + + let Some(brace_pos) = inner.find('{') else { + continue; + }; + let tool_name = inner[..brace_pos].trim(); + let json_body = inner[brace_pos..].trim(); + + if tool_name.is_empty() || !tool_names.contains(&tool_name) { + continue; + } + + let input: serde_json::Value = match serde_json::from_str(json_body) { + Ok(v) => v, + Err(_) => continue, + }; + + if calls + .iter() + .any(|c| c.name == tool_name && c.input == input) + { + continue; + } + + info!( + tool = tool_name, + "Recovered text-based tool call ( variant) → synthetic ToolUse" + ); + calls.push(ToolCall { + id: format!("recovered_{}", uuid::Uuid::new_v4()), + name: tool_name.to_string(), + input, + }); + } + + // Pattern 4: Markdown code blocks containing tool_name {JSON} + // Matches: ```\nexec {"command":"ls"}\n``` or ```bash\nexec {"command":"ls"}\n``` + { + let mut in_block = false; + let mut block_content = String::new(); + for line in text.lines() { + let trimmed = line.trim(); + if trimmed.starts_with("```") { + if in_block { + // End of block — try to extract tool call from content + let content = block_content.trim(); + if let Some(brace_pos) = content.find('{') { + let potential_tool = content[..brace_pos].trim(); + if tool_names.contains(&potential_tool) { + if let Ok(input) = serde_json::from_str::( + content[brace_pos..].trim(), + ) { + if !calls + .iter() + .any(|c| c.name == potential_tool && c.input == input) + { + info!( + tool = potential_tool, + "Recovered tool call from markdown code block" + ); + calls.push(ToolCall { + id: format!("recovered_{}", uuid::Uuid::new_v4()), + name: potential_tool.to_string(), + input, + }); + } + } + } + } + block_content.clear(); + in_block = false; + } else { + in_block = true; + block_content.clear(); + } + } else if in_block { + if !block_content.is_empty() { + block_content.push('\n'); + } + block_content.push_str(trimmed); + } + } + } + + // Pattern 5: Backtick-wrapped tool call: `tool_name {"key":"value"}` + { + let parts: Vec<&str> = text.split('`').collect(); + // Every odd-indexed element is inside backticks + for chunk in parts.iter().skip(1).step_by(2) { + let trimmed = chunk.trim(); + if let Some(brace_pos) = trimmed.find('{') { + let potential_tool = trimmed[..brace_pos].trim(); + if !potential_tool.is_empty() + && !potential_tool.contains(' ') + && tool_names.contains(&potential_tool) + { + if let Ok(input) = + serde_json::from_str::(trimmed[brace_pos..].trim()) + { + if !calls + .iter() + .any(|c| c.name == potential_tool && c.input == input) + { + info!( + tool = potential_tool, + "Recovered tool call from backtick-wrapped text" + ); + calls.push(ToolCall { + id: format!("recovered_{}", uuid::Uuid::new_v4()), + name: potential_tool.to_string(), + input, + }); + } + } + } + } + } + } + calls } @@ -2692,6 +2822,86 @@ mod tests { assert_eq!(calls[1].name, "web_fetch"); } + #[test] + fn test_recover_tool_tag_variant() { + let tools = vec![ToolDefinition { + name: "exec".into(), + description: "Execute".into(), + input_schema: serde_json::json!({}), + }]; + let text = r#"I'll run that for you. exec{"command":"ls -la"}"#; + let calls = recover_text_tool_calls(text, &tools); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].name, "exec"); + assert_eq!(calls[0].input["command"], "ls -la"); + } + + #[test] + fn test_recover_markdown_code_block() { + let tools = vec![ToolDefinition { + name: "exec".into(), + description: "Execute".into(), + input_schema: serde_json::json!({}), + }]; + let text = "I'll execute that command:\n```\nexec {\"command\": \"ls -la\"}\n```"; + let calls = recover_text_tool_calls(text, &tools); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].name, "exec"); + assert_eq!(calls[0].input["command"], "ls -la"); + } + + #[test] + fn test_recover_markdown_code_block_with_lang() { + let tools = vec![ToolDefinition { + name: "web_search".into(), + description: "Search".into(), + input_schema: serde_json::json!({}), + }]; + let text = "```json\nweb_search {\"query\": \"rust\"}\n```"; + let calls = recover_text_tool_calls(text, &tools); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].name, "web_search"); + } + + #[test] + fn test_recover_backtick_wrapped() { + let tools = vec![ToolDefinition { + name: "exec".into(), + description: "Execute".into(), + input_schema: serde_json::json!({}), + }]; + let text = r#"Let me run `exec {"command":"pwd"}` for you."#; + let calls = recover_text_tool_calls(text, &tools); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].name, "exec"); + assert_eq!(calls[0].input["command"], "pwd"); + } + + #[test] + fn test_recover_backtick_ignores_unknown_tool() { + let tools = vec![ToolDefinition { + name: "exec".into(), + description: "Execute".into(), + input_schema: serde_json::json!({}), + }]; + let text = r#"Try `unknown_tool {"key":"val"}` instead."#; + let calls = recover_text_tool_calls(text, &tools); + assert!(calls.is_empty()); + } + + #[test] + fn test_recover_no_duplicates_across_patterns() { + let tools = vec![ToolDefinition { + name: "exec".into(), + description: "Execute".into(), + input_schema: serde_json::json!({}), + }]; + // Same call in both function tag and tool tag — should only appear once + let text = r#"{"command":"ls"} exec{"command":"ls"}"#; + let calls = recover_text_tool_calls(text, &tools); + assert_eq!(calls.len(), 1); + } + // --- End-to-end integration test: text-as-tool-call recovery through agent loop --- /// Mock driver that simulates a Groq/Llama model outputting tool calls as text.