From 951e8d0feb96a8d422e8781bae476e392c75b718 Mon Sep 17 00:00:00 2001 From: jaberjaber23 Date: Thu, 12 Mar 2026 01:22:45 +0300 Subject: [PATCH] fix 11 issues --- crates/openfang-api/src/channel_bridge.rs | 4 +- crates/openfang-api/src/openai_compat.rs | 2 +- crates/openfang-api/src/routes.rs | 43 +- crates/openfang-api/src/ws.rs | 2 +- crates/openfang-api/static/js/pages/chat.js | 27 +- crates/openfang-channels/src/bridge.rs | 123 ++++- crates/openfang-channels/src/telegram.rs | 404 ++++++++++++++-- crates/openfang-channels/src/types.rs | 7 + .../openfang-hands/bundled/browser/HAND.toml | 3 +- crates/openfang-hands/src/lib.rs | 7 + crates/openfang-hands/src/registry.rs | 125 +++++ crates/openfang-kernel/src/kernel.rs | 330 +++++++++---- crates/openfang-kernel/src/triggers.rs | 221 +++++++++ crates/openfang-memory/src/session.rs | 2 +- crates/openfang-runtime/src/agent_loop.rs | 18 +- crates/openfang-runtime/src/compactor.rs | 10 +- .../openfang-runtime/src/drivers/anthropic.rs | 6 +- .../src/drivers/claude_code.rs | 45 +- .../openfang-runtime/src/drivers/fallback.rs | 1 + crates/openfang-runtime/src/drivers/gemini.rs | 443 +++++++++++++++--- crates/openfang-runtime/src/drivers/mod.rs | 11 +- crates/openfang-runtime/src/drivers/openai.rs | 15 +- crates/openfang-runtime/src/kernel_handle.rs | 24 +- crates/openfang-runtime/src/llm_driver.rs | 19 +- crates/openfang-runtime/src/session_repair.rs | 7 +- crates/openfang-runtime/src/tool_runner.rs | 77 ++- crates/openfang-types/src/message.rs | 12 +- 27 files changed, 1733 insertions(+), 255 deletions(-) diff --git a/crates/openfang-api/src/channel_bridge.rs b/crates/openfang-api/src/channel_bridge.rs index 42b8587c..1477f85b 100644 --- a/crates/openfang-api/src/channel_bridge.rs +++ b/crates/openfang-api/src/channel_bridge.rs @@ -84,7 +84,7 @@ impl ChannelBridgeHandle for KernelBridgeAdapter { let text: String = blocks .iter() .filter_map(|b| match b { - openfang_types::message::ContentBlock::Text { text } => Some(text.as_str()), + openfang_types::message::ContentBlock::Text { text, .. } => Some(text.as_str()), _ => None, }) .collect::>() @@ -684,7 +684,7 @@ impl ChannelBridgeHandle for KernelBridgeAdapter { )); } self.kernel - .set_agent_model(agent_id, model) + .set_agent_model(agent_id, model, None) .map_err(|e| format!("{e}"))?; // Read back resolved model+provider from registry let entry = self diff --git a/crates/openfang-api/src/openai_compat.rs b/crates/openfang-api/src/openai_compat.rs index a0e806c6..1d95076e 100644 --- a/crates/openfang-api/src/openai_compat.rs +++ b/crates/openfang-api/src/openai_compat.rs @@ -203,7 +203,7 @@ fn convert_messages(oai_messages: &[OaiMessage]) -> Vec { .iter() .filter_map(|part| match part { OaiContentPart::Text { text } => { - Some(ContentBlock::Text { text: text.clone() }) + Some(ContentBlock::Text { text: text.clone(), provider_metadata: None }) } OaiContentPart::ImageUrl { image_url } => { // Parse data URI: data:{media_type};base64,{data} diff --git a/crates/openfang-api/src/routes.rs b/crates/openfang-api/src/routes.rs index dfb3e4a1..c2e063f4 100644 --- a/crates/openfang-api/src/routes.rs +++ b/crates/openfang-api/src/routes.rs @@ -453,7 +453,7 @@ pub async fn get_agent_session( let mut texts = Vec::new(); for b in blocks { match b { - openfang_types::message::ContentBlock::Text { text } => { + openfang_types::message::ContentBlock::Text { text, .. } => { texts.push(text.clone()); } openfang_types::message::ContentBlock::Image { @@ -3489,7 +3489,10 @@ pub async fn list_hands(State(state): State>) -> impl IntoResponse .hand_registry .check_requirements(&d.id) .unwrap_or_default(); - let all_satisfied = reqs.iter().all(|(_, ok)| *ok); + let readiness = state.kernel.hand_registry.readiness(&d.id); + let requirements_met = readiness.as_ref().map(|r| r.requirements_met).unwrap_or(false); + let active = readiness.as_ref().map(|r| r.active).unwrap_or(false); + let degraded = readiness.as_ref().map(|r| r.degraded).unwrap_or(false); serde_json::json!({ "id": d.id, "name": d.name, @@ -3497,11 +3500,14 @@ pub async fn list_hands(State(state): State>) -> impl IntoResponse "category": d.category, "icon": d.icon, "tools": d.tools, - "requirements_met": all_satisfied, + "requirements_met": requirements_met, + "active": active, + "degraded": degraded, "requirements": reqs.iter().map(|(r, ok)| serde_json::json!({ "key": r.key, "label": r.label, "satisfied": ok, + "optional": r.optional, })).collect::>(), "dashboard_metrics": d.dashboard.metrics.len(), "has_settings": !d.settings.is_empty(), @@ -3546,7 +3552,10 @@ pub async fn get_hand( .hand_registry .check_requirements(&hand_id) .unwrap_or_default(); - let all_satisfied = reqs.iter().all(|(_, ok)| *ok); + let readiness = state.kernel.hand_registry.readiness(&hand_id); + let requirements_met = readiness.as_ref().map(|r| r.requirements_met).unwrap_or(false); + let active = readiness.as_ref().map(|r| r.active).unwrap_or(false); + let degraded = readiness.as_ref().map(|r| r.degraded).unwrap_or(false); let settings_status = state .kernel .hand_registry @@ -3561,7 +3570,9 @@ pub async fn get_hand( "category": def.category, "icon": def.icon, "tools": def.tools, - "requirements_met": all_satisfied, + "requirements_met": requirements_met, + "active": active, + "degraded": degraded, "requirements": reqs.iter().map(|(r, ok)| { let mut req_json = serde_json::json!({ "key": r.key, @@ -3569,6 +3580,7 @@ pub async fn get_hand( "type": format!("{:?}", r.requirement_type), "check_value": r.check_value, "satisfied": ok, + "optional": r.optional, }); if let Some(ref desc) = r.description { req_json["description"] = serde_json::json!(desc); @@ -3617,12 +3629,17 @@ pub async fn check_hand_deps( .hand_registry .check_requirements(&hand_id) .unwrap_or_default(); - let all_satisfied = reqs.iter().all(|(_, ok)| *ok); + let readiness = state.kernel.hand_registry.readiness(&hand_id); + let requirements_met = readiness.as_ref().map(|r| r.requirements_met).unwrap_or(false); + let active = readiness.as_ref().map(|r| r.active).unwrap_or(false); + let degraded = readiness.as_ref().map(|r| r.degraded).unwrap_or(false); ( StatusCode::OK, Json(serde_json::json!({ "hand_id": def.id, - "requirements_met": all_satisfied, + "requirements_met": requirements_met, + "active": active, + "degraded": degraded, "server_platform": server_platform(), "requirements": reqs.iter().map(|(r, ok)| { let mut req_json = serde_json::json!({ @@ -3631,6 +3648,7 @@ pub async fn check_hand_deps( "type": format!("{:?}", r.requirement_type), "check_value": r.check_value, "satisfied": ok, + "optional": r.optional, }); if let Some(ref desc) = r.description { req_json["description"] = serde_json::json!(desc); @@ -5179,7 +5197,8 @@ pub async fn patch_agent( } } if let Some(model) = body.get("model").and_then(|v| v.as_str()) { - if let Err(e) = state.kernel.set_agent_model(agent_id, model) { + let explicit_provider = body.get("provider").and_then(|v| v.as_str()); + if let Err(e) = state.kernel.set_agent_model(agent_id, model, explicit_provider) { return ( StatusCode::BAD_REQUEST, Json(serde_json::json!({"error": format!("{e}")})), @@ -6426,7 +6445,8 @@ pub async fn set_model( ) } }; - match state.kernel.set_agent_model(agent_id, model) { + let explicit_provider = body["provider"].as_str(); + match state.kernel.set_agent_model(agent_id, model, explicit_provider) { Ok(()) => { // Return the resolved model+provider so frontend stays in sync. // The model name may have been normalized (provider prefix stripped), @@ -6984,6 +7004,7 @@ pub async fn test_provider( } else { Some(base_url) }, + skip_permissions: true, }; match openfang_runtime::drivers::create_driver(&driver_config) { @@ -8279,7 +8300,7 @@ pub async fn patch_agent_config( } } else { // Provider is empty string — resolve from catalog - if let Err(e) = state.kernel.set_agent_model(agent_id, new_model) { + if let Err(e) = state.kernel.set_agent_model(agent_id, new_model, None) { return ( StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": format!("{e}")})), @@ -8288,7 +8309,7 @@ pub async fn patch_agent_config( } } else { // No provider field at all — resolve from catalog - if let Err(e) = state.kernel.set_agent_model(agent_id, new_model) { + if let Err(e) = state.kernel.set_agent_model(agent_id, new_model, None) { return ( StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": format!("{e}")})), diff --git a/crates/openfang-api/src/ws.rs b/crates/openfang-api/src/ws.rs index 7bfe0a9d..bb3752c6 100644 --- a/crates/openfang-api/src/ws.rs +++ b/crates/openfang-api/src/ws.rs @@ -809,7 +809,7 @@ async fn handle_command( serde_json::json!({"type": "error", "content": "Agent not found"}) } } else { - match state.kernel.set_agent_model(agent_id, args) { + match state.kernel.set_agent_model(agent_id, args, None) { Ok(()) => { if let Some(entry) = state.kernel.registry.get(agent_id) { let model = &entry.manifest.model.model; diff --git a/crates/openfang-api/static/js/pages/chat.js b/crates/openfang-api/static/js/pages/chat.js index c5555dbb..9a748f3c 100644 --- a/crates/openfang-api/static/js/pages/chat.js +++ b/crates/openfang-api/static/js/pages/chat.js @@ -650,17 +650,32 @@ function chatPage() { // Show tool/phase progress so the user sees the agent is working var phaseMsg = this.messages.length ? this.messages[this.messages.length - 1] : null; if (phaseMsg && (phaseMsg.thinking || phaseMsg.streaming)) { - var detail = data.detail || data.phase || 'Working...'; - // Context warning: show prominently + // Skip phases that have no user-meaningful display text — "streaming" + // and "done" are lifecycle signals, not status to show in the chat bubble. + if (data.phase === 'streaming' || data.phase === 'done') { + break; + } + // Context warning: show prominently as a separate system message if (data.phase === 'context_warning') { - this.messages.push({ id: ++msgId, role: 'system', text: detail, meta: '', tools: [] }); + var cwDetail = data.detail || 'Context limit reached.'; + this.messages.push({ id: ++msgId, role: 'system', text: cwDetail, meta: '', tools: [] }); } else if (data.phase === 'thinking' && this.thinkingMode === 'stream') { // Stream reasoning tokens to a collapsible panel if (!phaseMsg._reasoning) phaseMsg._reasoning = ''; - phaseMsg._reasoning += (detail || '') + '\n'; + phaseMsg._reasoning += (data.detail || '') + '\n'; phaseMsg.text = '
Reasoning...\n\n' + phaseMsg._reasoning + '
'; - } else { - phaseMsg.text = detail; + } else if (phaseMsg.thinking) { + // Only update text on messages still in thinking state (not yet + // receiving streamed content) to avoid overwriting accumulated text. + var phaseDetail; + if (data.phase === 'tool_use') { + phaseDetail = 'Using ' + (data.detail || 'tool') + '...'; + } else if (data.phase === 'thinking') { + phaseDetail = 'Thinking...'; + } else { + phaseDetail = data.detail || 'Working...'; + } + phaseMsg.text = phaseDetail; } } this.scrollToBottom(); diff --git a/crates/openfang-channels/src/bridge.rs b/crates/openfang-channels/src/bridge.rs index 742a5b3b..3a8c3d68 100644 --- a/crates/openfang-channels/src/bridge.rs +++ b/crates/openfang-channels/src/bridge.rs @@ -41,7 +41,7 @@ pub trait ChannelBridgeHandle: Send + Sync { let text: String = blocks .iter() .filter_map(|b| match b { - ContentBlock::Text { text } => Some(text.as_str()), + ContentBlock::Text { text, .. } => Some(text.as_str()), _ => None, }) .collect::>() @@ -568,6 +568,9 @@ async fn dispatch_message( ChannelContent::Location { lat, lon } => { format!("[User shared location: {lat}, {lon}]") } + ChannelContent::FileData { ref filename, .. } => { + format!("[User sent a local file: {filename}]") + } }; // Check if it's a slash command embedded in text (e.g. "/agents") @@ -792,6 +795,40 @@ async fn dispatch_message( } } +/// Detect image format from the first few magic bytes. +/// +/// Returns `Some("image/...")` for JPEG, PNG, GIF, and WebP. +fn detect_image_magic(bytes: &[u8]) -> Option { + if bytes.len() >= 3 && bytes[..3] == [0xFF, 0xD8, 0xFF] { + return Some("image/jpeg".to_string()); + } + if bytes.len() >= 4 && bytes[..4] == [0x89, 0x50, 0x4E, 0x47] { + return Some("image/png".to_string()); + } + if bytes.len() >= 4 && bytes[..4] == [0x47, 0x49, 0x46, 0x38] { + return Some("image/gif".to_string()); + } + if bytes.len() >= 12 && bytes[..4] == [0x52, 0x49, 0x46, 0x46] && bytes[8..12] == [0x57, 0x45, 0x42, 0x50] + { + return Some("image/webp".to_string()); + } + None +} + +/// Guess image media type from the URL file extension. +fn media_type_from_url(url: &str) -> String { + if url.contains(".png") { + "image/png".to_string() + } else if url.contains(".gif") { + "image/gif".to_string() + } else if url.contains(".webp") { + "image/webp".to_string() + } else { + // JPEG is the most common image format — safe default + "image/jpeg".to_string() + } +} + /// Download an image from a URL and build content blocks for multimodal LLM input. /// /// Returns a `Vec` containing an image block (base64-encoded) and @@ -810,28 +847,20 @@ async fn download_image_to_blocks(url: &str, caption: Option<&str>) -> Vec b, @@ -839,10 +868,19 @@ async fn download_image_to_blocks(url: &str, caption: Option<&str>) -> Vec MAX_IMAGE_BYTES { warn!( "Image too large ({} bytes), skipping vision — sending as text", @@ -852,7 +890,7 @@ async fn download_image_to_blocks(url: &str, caption: Option<&str>) -> Vec format!("[Image too large for vision ({} KB)]\nCaption: {c}", bytes.len() / 1024), None => format!("[Image too large for vision ({} KB)]", bytes.len() / 1024), }; - return vec![ContentBlock::Text { text: desc }]; + return vec![ContentBlock::Text { text: desc, provider_metadata: None }]; } let data = base64::engine::general_purpose::STANDARD.encode(&bytes); @@ -864,6 +902,7 @@ async fn download_image_to_blocks(url: &str, caption: Option<&str>) -> Vec>>, shutdown_tx: Arc>, shutdown_rx: watch::Receiver, } @@ -63,6 +66,7 @@ impl TelegramAdapter { allowed_users, poll_interval, api_base_url, + bot_username: Arc::new(tokio::sync::RwLock::new(None)), shutdown_tx: Arc::new(shutdown_tx), shutdown_rx, } @@ -189,6 +193,44 @@ impl TelegramAdapter { Ok(()) } + /// Call `sendDocument` with multipart upload for local file data. + /// + /// Used by the proactive `channel_send` tool when `file_path` is provided. + /// Uploads raw bytes as a multipart form instead of passing a URL. + async fn api_send_document_upload( + &self, + chat_id: i64, + data: Vec, + filename: &str, + mime_type: &str, + thread_id: Option, + ) -> Result<(), Box> { + let url = format!( + "{}/bot{}/sendDocument", + self.api_base_url, + self.token.as_str() + ); + + let file_part = reqwest::multipart::Part::bytes(data) + .file_name(filename.to_string()) + .mime_str(mime_type)?; + + let mut form = reqwest::multipart::Form::new() + .text("chat_id", chat_id.to_string()) + .part("document", file_part); + + if let Some(tid) = thread_id { + form = form.text("message_thread_id", tid.to_string()); + } + + let resp = self.client.post(&url).multipart(form).send().await?; + if !resp.status().is_success() { + let body_text = resp.text().await.unwrap_or_default(); + warn!("Telegram sendDocument upload failed: {body_text}"); + } + Ok(()) + } + /// Call `sendVoice` on the Telegram API. async fn api_send_voice( &self, @@ -330,6 +372,14 @@ impl TelegramAdapter { self.api_send_document(chat_id, &url, &filename, thread_id) .await?; } + ChannelContent::FileData { + data, + filename, + mime_type, + } => { + self.api_send_document_upload(chat_id, data, &filename, &mime_type, thread_id) + .await?; + } ChannelContent::Voice { url, .. } => { self.api_send_voice(chat_id, &url, thread_id).await?; } @@ -361,8 +411,12 @@ impl ChannelAdapter for TelegramAdapter { &self, ) -> Result + Send>>, Box> { - // Validate token first (fail fast) + // Validate token first (fail fast) and store bot username for mention detection let bot_name = self.validate_token().await?; + { + let mut username = self.bot_username.write().await; + *username = Some(bot_name.clone()); + } info!("Telegram bot @{bot_name} connected"); // Clear any existing webhook to avoid 409 Conflict during getUpdates polling. @@ -393,6 +447,7 @@ impl ChannelAdapter for TelegramAdapter { let allowed_users = self.allowed_users.clone(); let poll_interval = self.poll_interval; let api_base_url = self.api_base_url.clone(); + let bot_username = self.bot_username.clone(); let mut shutdown = self.shutdown_rx.clone(); tokio::spawn(async move { @@ -505,7 +560,8 @@ impl ChannelAdapter for TelegramAdapter { } // Parse the message - let msg = match parse_telegram_update(update, &allowed_users, token.as_str(), &client, &api_base_url).await { + let bot_uname = bot_username.read().await.clone(); + let msg = match parse_telegram_update(update, &allowed_users, token.as_str(), &client, &api_base_url, bot_uname.as_deref()).await { Some(m) => m, None => continue, // filtered out or unparseable }; @@ -613,12 +669,51 @@ async fn parse_telegram_update( token: &str, client: &reqwest::Client, api_base_url: &str, + bot_username: Option<&str>, ) -> Option { - let message = update - .get("message") - .or_else(|| update.get("edited_message"))?; - let from = message.get("from")?; - let user_id = from["id"].as_i64()?; + let update_id = update["update_id"].as_i64().unwrap_or(0); + let message = match update.get("message").or_else(|| update.get("edited_message")) { + Some(m) => m, + None => { + debug!("Telegram: dropping update {update_id} — no message or edited_message field"); + return None; + } + }; + + // Extract sender info: prefer `from` (user), fall back to `sender_chat` (channel/group) + let (user_id, display_name) = if let Some(from) = message.get("from") { + let uid = match from["id"].as_i64() { + Some(id) => id, + None => { + debug!("Telegram: dropping update {update_id} — from.id is not an integer"); + return None; + } + }; + let first_name = from["first_name"].as_str().unwrap_or("Unknown"); + let last_name = from["last_name"].as_str().unwrap_or(""); + let name = if last_name.is_empty() { + first_name.to_string() + } else { + format!("{first_name} {last_name}") + }; + (uid, name) + } else if let Some(sender_chat) = message.get("sender_chat") { + // Messages sent on behalf of a channel or group have `sender_chat` instead of `from`. + let uid = match sender_chat["id"].as_i64() { + Some(id) => id, + None => { + debug!("Telegram: dropping update {update_id} — sender_chat.id is not an integer"); + return None; + } + }; + let title = sender_chat["title"] + .as_str() + .unwrap_or("Unknown Channel"); + (uid, title.to_string()) + } else { + debug!("Telegram: dropping update {update_id} — no from or sender_chat field"); + return None; + }; // Security: check allowed_users (compare as strings for consistency) let user_id_str = user_id.to_string(); @@ -627,13 +722,12 @@ async fn parse_telegram_update( return None; } - let chat_id = message["chat"]["id"].as_i64()?; - let first_name = from["first_name"].as_str().unwrap_or("Unknown"); - let last_name = from["last_name"].as_str().unwrap_or(""); - let display_name = if last_name.is_empty() { - first_name.to_string() - } else { - format!("{first_name} {last_name}") + let chat_id = match message["chat"]["id"].as_i64() { + Some(id) => id, + None => { + debug!("Telegram: dropping update {update_id} — chat.id is not an integer"); + return None; + } }; let chat_type = message["chat"]["type"].as_str().unwrap_or("private"); @@ -710,6 +804,7 @@ async fn parse_telegram_update( ChannelContent::Location { lat, lon } } else { // Unsupported message type (stickers, polls, etc.) + debug!("Telegram: dropping update {update_id} — unsupported message type (no text/photo/document/voice/location)"); return None; }; @@ -719,6 +814,17 @@ async fn parse_telegram_update( .as_i64() .map(|tid| tid.to_string()); + // Detect @mention of the bot in entities / caption_entities for MentionOnly group policy. + let mut metadata = HashMap::new(); + if is_group { + if let Some(bot_uname) = bot_username { + let was_mentioned = check_mention_entities(message, bot_uname); + if was_mentioned { + metadata.insert("was_mentioned".to_string(), serde_json::json!(true)); + } + } + } + Some(ChannelMessage { channel: ChannelType::Telegram, platform_message_id: message_id.to_string(), @@ -732,10 +838,45 @@ async fn parse_telegram_update( timestamp, is_group, thread_id, - metadata: HashMap::new(), + metadata, }) } +/// Check whether the bot was @mentioned in a Telegram message. +/// +/// Inspects both `entities` (for text messages) and `caption_entities` (for media +/// with captions) for entity type `"mention"` whose text matches `@bot_username`. +fn check_mention_entities(message: &serde_json::Value, bot_username: &str) -> bool { + let bot_mention = format!("@{}", bot_username.to_lowercase()); + + // Check both entities (text messages) and caption_entities (photo/document captions) + for entities_key in &["entities", "caption_entities"] { + if let Some(entities) = message[entities_key].as_array() { + // Get the text that the entities refer to + let text = if *entities_key == "entities" { + message["text"].as_str().unwrap_or("") + } else { + message["caption"].as_str().unwrap_or("") + }; + + for entity in entities { + if entity["type"].as_str() != Some("mention") { + continue; + } + let offset = entity["offset"].as_i64().unwrap_or(0) as usize; + let length = entity["length"].as_i64().unwrap_or(0) as usize; + if offset + length <= text.len() { + let mention_text = &text[offset..offset + length]; + if mention_text.to_lowercase() == bot_mention { + return true; + } + } + } + } + } + false +} + /// Calculate exponential backoff capped at MAX_BACKOFF. pub fn calculate_backoff(current: Duration) -> Duration { (current * 2).min(MAX_BACKOFF) @@ -827,7 +968,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); assert_eq!(msg.channel, ChannelType::Telegram); assert_eq!(msg.sender.display_name, "Alice Smith"); assert_eq!(msg.sender.platform_id, "111222333"); @@ -859,7 +1000,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); match &msg.content { ChannelContent::Command { name, args } => { assert_eq!(name, "agent"); @@ -891,17 +1032,17 @@ mod tests { let client = test_client(); // Empty allowed_users = allow all - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await; + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await; assert!(msg.is_some()); // Non-matching allowed_users = filter out let blocked: Vec = vec!["111".to_string(), "222".to_string()]; - let msg = parse_telegram_update(&update, &blocked, "fake:token", &client, DEFAULT_API_URL).await; + let msg = parse_telegram_update(&update, &blocked, "fake:token", &client, DEFAULT_API_URL, None).await; assert!(msg.is_none()); // Matching allowed_users = allow let allowed: Vec = vec!["999".to_string()]; - let msg = parse_telegram_update(&update, &allowed, "fake:token", &client, DEFAULT_API_URL).await; + let msg = parse_telegram_update(&update, &allowed, "fake:token", &client, DEFAULT_API_URL, None).await; assert!(msg.is_some()); } @@ -927,7 +1068,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); assert_eq!(msg.channel, ChannelType::Telegram); assert_eq!(msg.sender.display_name, "Alice Smith"); assert!(matches!(msg.content, ChannelContent::Text(ref t) if t == "Edited message!")); @@ -963,7 +1104,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); match &msg.content { ChannelContent::Command { name, args } => { assert_eq!(name, "agents"); @@ -987,7 +1128,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); assert!(matches!(msg.content, ChannelContent::Location { .. })); } @@ -1011,7 +1152,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); // With a fake token, getFile will fail, so we get a text fallback match &msg.content { ChannelContent::Text(t) => { @@ -1046,7 +1187,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); match &msg.content { ChannelContent::Text(t) => { assert!(t.contains("Document received")); @@ -1077,7 +1218,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); match &msg.content { ChannelContent::Text(t) => { assert!(t.contains("Voice message")); @@ -1106,7 +1247,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); assert_eq!(msg.thread_id, Some("42".to_string())); assert!(msg.is_group); } @@ -1126,7 +1267,7 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); assert_eq!(msg.thread_id, None); assert!(!msg.is_group); } @@ -1148,10 +1289,215 @@ mod tests { }); let client = test_client(); - let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL).await.unwrap(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); assert_eq!(msg.thread_id, Some("99".to_string())); } + #[tokio::test] + async fn test_parse_sender_chat_fallback() { + // Messages sent on behalf of a channel have `sender_chat` instead of `from`. + let update = serde_json::json!({ + "update_id": 500, + "message": { + "message_id": 80, + "sender_chat": { + "id": -1001999888777_i64, + "title": "My Channel", + "type": "channel" + }, + "chat": { "id": -1001234567890_i64, "type": "supergroup" }, + "date": 1700000000, + "text": "Forwarded from channel" + } + }); + + let client = test_client(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await.unwrap(); + assert_eq!(msg.sender.display_name, "My Channel"); + assert_eq!(msg.sender.platform_id, "-1001234567890"); + assert!(matches!(msg.content, ChannelContent::Text(ref t) if t == "Forwarded from channel")); + } + + #[tokio::test] + async fn test_parse_no_from_no_sender_chat_drops() { + // Updates with neither `from` nor `sender_chat` should be dropped with debug logging. + let update = serde_json::json!({ + "update_id": 501, + "message": { + "message_id": 81, + "chat": { "id": 123, "type": "private" }, + "date": 1700000000, + "text": "orphan" + } + }); + + let client = test_client(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, None).await; + assert!(msg.is_none()); + } + + #[tokio::test] + async fn test_was_mentioned_in_group() { + // Bot @mentioned in a group message should set metadata["was_mentioned"]. + let update = serde_json::json!({ + "update_id": 600, + "message": { + "message_id": 90, + "from": { "id": 123, "first_name": "Alice" }, + "chat": { "id": -1001234567890_i64, "type": "supergroup" }, + "date": 1700000000, + "text": "Hey @testbot what do you think?", + "entities": [{ + "type": "mention", + "offset": 4, + "length": 8 + }] + } + }); + + let client = test_client(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, Some("testbot")).await.unwrap(); + assert!(msg.is_group); + assert_eq!(msg.metadata.get("was_mentioned").and_then(|v| v.as_bool()), Some(true)); + } + + #[tokio::test] + async fn test_not_mentioned_in_group() { + // Group message without a mention should NOT have was_mentioned. + let update = serde_json::json!({ + "update_id": 601, + "message": { + "message_id": 91, + "from": { "id": 123, "first_name": "Alice" }, + "chat": { "id": -1001234567890_i64, "type": "supergroup" }, + "date": 1700000000, + "text": "Just chatting" + } + }); + + let client = test_client(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, Some("testbot")).await.unwrap(); + assert!(msg.is_group); + assert!(!msg.metadata.contains_key("was_mentioned")); + } + + #[tokio::test] + async fn test_mentioned_different_bot_not_set() { + // @mention of a different bot should NOT set was_mentioned. + let update = serde_json::json!({ + "update_id": 602, + "message": { + "message_id": 92, + "from": { "id": 123, "first_name": "Alice" }, + "chat": { "id": -1001234567890_i64, "type": "supergroup" }, + "date": 1700000000, + "text": "Hey @otherbot what do you think?", + "entities": [{ + "type": "mention", + "offset": 4, + "length": 9 + }] + } + }); + + let client = test_client(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, Some("testbot")).await.unwrap(); + assert!(msg.is_group); + assert!(!msg.metadata.contains_key("was_mentioned")); + } + + #[tokio::test] + async fn test_mention_in_caption_entities() { + // Bot mentioned in a photo caption should set was_mentioned. + let update = serde_json::json!({ + "update_id": 603, + "message": { + "message_id": 93, + "from": { "id": 123, "first_name": "Alice" }, + "chat": { "id": -1001234567890_i64, "type": "supergroup" }, + "date": 1700000000, + "photo": [ + { "file_id": "photo_id", "file_unique_id": "x", "width": 800, "height": 600 } + ], + "caption": "Look @testbot", + "caption_entities": [{ + "type": "mention", + "offset": 5, + "length": 8 + }] + } + }); + + let client = test_client(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, Some("testbot")).await.unwrap(); + assert!(msg.is_group); + assert_eq!(msg.metadata.get("was_mentioned").and_then(|v| v.as_bool()), Some(true)); + } + + #[tokio::test] + async fn test_mention_case_insensitive() { + // Mention detection should be case-insensitive. + let update = serde_json::json!({ + "update_id": 604, + "message": { + "message_id": 94, + "from": { "id": 123, "first_name": "Alice" }, + "chat": { "id": -1001234567890_i64, "type": "supergroup" }, + "date": 1700000000, + "text": "Hey @TestBot help", + "entities": [{ + "type": "mention", + "offset": 4, + "length": 8 + }] + } + }); + + let client = test_client(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, Some("testbot")).await.unwrap(); + assert_eq!(msg.metadata.get("was_mentioned").and_then(|v| v.as_bool()), Some(true)); + } + + #[tokio::test] + async fn test_private_chat_no_mention_check() { + // Private chats should NOT populate was_mentioned even with entities. + let update = serde_json::json!({ + "update_id": 605, + "message": { + "message_id": 95, + "from": { "id": 123, "first_name": "Alice" }, + "chat": { "id": 123, "type": "private" }, + "date": 1700000000, + "text": "Hey @testbot", + "entities": [{ + "type": "mention", + "offset": 4, + "length": 8 + }] + } + }); + + let client = test_client(); + let msg = parse_telegram_update(&update, &[], "fake:token", &client, DEFAULT_API_URL, Some("testbot")).await.unwrap(); + assert!(!msg.is_group); + // In private chats, mention detection is skipped — no metadata set + assert!(!msg.metadata.contains_key("was_mentioned")); + } + + #[test] + fn test_check_mention_entities_direct() { + let message = serde_json::json!({ + "text": "Hello @mybot world", + "entities": [{ + "type": "mention", + "offset": 6, + "length": 6 + }] + }); + assert!(check_mention_entities(&message, "mybot")); + assert!(!check_mention_entities(&message, "otherbot")); + } + #[test] fn test_sanitize_telegram_html_basic() { // Allowed tags preserved, unknown tags escaped diff --git a/crates/openfang-channels/src/types.rs b/crates/openfang-channels/src/types.rs index bfd3fe1b..34e63727 100644 --- a/crates/openfang-channels/src/types.rs +++ b/crates/openfang-channels/src/types.rs @@ -49,6 +49,13 @@ pub enum ChannelContent { url: String, filename: String, }, + /// Local file data (bytes read from disk). Used by the proactive `channel_send` + /// tool when `file_path` is provided instead of `file_url`. + FileData { + data: Vec, + filename: String, + mime_type: String, + }, Voice { url: String, duration_seconds: u32, diff --git a/crates/openfang-hands/bundled/browser/HAND.toml b/crates/openfang-hands/bundled/browser/HAND.toml index ce8cdc6c..08ee0b6c 100644 --- a/crates/openfang-hands/bundled/browser/HAND.toml +++ b/crates/openfang-hands/bundled/browser/HAND.toml @@ -35,7 +35,8 @@ key = "chromium" label = "Chromium or Google Chrome must be installed" requirement_type = "binary" check_value = "chromium" -description = "A Chromium-based browser is required. Google Chrome, Chromium, or any Chromium derivative will work. You can also set the CHROME_PATH environment variable to point to your browser binary." +optional = true +description = "A Chromium-based browser is recommended. Playwright can install its own bundled browser if none is found. Google Chrome, Chromium, or any Chromium derivative will also work. You can set the CHROME_PATH environment variable to point to your browser binary." [requires.install] macos = "brew install --cask google-chrome" diff --git a/crates/openfang-hands/src/lib.rs b/crates/openfang-hands/src/lib.rs index 1d734a83..f2d3e339 100644 --- a/crates/openfang-hands/src/lib.rs +++ b/crates/openfang-hands/src/lib.rs @@ -117,6 +117,13 @@ pub struct HandRequirement { /// Human-readable description of why this is needed. #[serde(default)] pub description: Option, + /// Whether this requirement is optional (non-critical). + /// + /// Optional requirements do not block activation. When an active hand has + /// unmet optional requirements it is reported as "degraded" rather than + /// "requirements not met". + #[serde(default)] + pub optional: bool, /// Platform-specific installation instructions. #[serde(default)] pub install: Option, diff --git a/crates/openfang-hands/src/registry.rs b/crates/openfang-hands/src/registry.rs index 45ba3f1c..36fa0010 100644 --- a/crates/openfang-hands/src/registry.rs +++ b/crates/openfang-hands/src/registry.rs @@ -352,6 +352,46 @@ impl HandRegistry { entry.updated_at = chrono::Utc::now(); Ok(()) } + + /// Compute readiness for a hand, cross-referencing requirements with + /// active instance state. + /// + /// Returns `None` if the hand definition does not exist. + pub fn readiness(&self, hand_id: &str) -> Option { + let reqs = self.check_requirements(hand_id).ok()?; + + let requirements_met = reqs.iter().all(|(_, ok)| *ok); + + // A hand is active if at least one instance is in Active status. + let active = self.instances.iter().any(|entry| { + entry.hand_id == hand_id && entry.status == HandStatus::Active + }); + + // Degraded: active, but at least one non-optional requirement is unmet + // OR any optional requirement is unmet. In practice, the most useful + // definition is: active + any requirement unsatisfied. + let degraded = active && reqs.iter().any(|(_, ok)| !ok); + + Some(HandReadiness { + requirements_met, + active, + degraded, + }) + } +} + +/// Readiness snapshot for a hand definition — combines requirement checks +/// with runtime activation state so the API can report unambiguous status. +#[derive(Debug, Clone, Serialize)] +pub struct HandReadiness { + /// Whether all declared requirements are currently satisfied. + pub requirements_met: bool, + /// Whether the hand currently has a running (Active-status) instance. + pub active: bool, + /// Whether the hand is active but some requirements are unmet. + /// This means the hand is running in a degraded mode — some features + /// may not work (e.g. browser hand without chromium). + pub degraded: bool, } impl Default for HandRegistry { @@ -638,6 +678,7 @@ mod tests { requirement_type: RequirementType::EnvVar, check_value: "OPENFANG_TEST_HAND_REQ".to_string(), description: None, + optional: false, install: None, }; assert!(check_requirement(&req)); @@ -648,9 +689,93 @@ mod tests { requirement_type: RequirementType::EnvVar, check_value: "OPENFANG_NONEXISTENT_VAR_12345".to_string(), description: None, + optional: false, install: None, }; assert!(!check_requirement(&req_missing)); std::env::remove_var("OPENFANG_TEST_HAND_REQ"); } + + #[test] + fn readiness_nonexistent_hand() { + let reg = HandRegistry::new(); + assert!(reg.readiness("nonexistent").is_none()); + } + + #[test] + fn readiness_inactive_hand() { + let reg = HandRegistry::new(); + reg.load_bundled(); + + // Lead hand has no requirements, so requirements_met = true + let r = reg.readiness("lead").unwrap(); + assert!(r.requirements_met); + assert!(!r.active); + assert!(!r.degraded); + } + + #[test] + fn readiness_active_hand_all_met() { + let reg = HandRegistry::new(); + reg.load_bundled(); + + // Lead hand has no requirements — activate it + let instance = reg.activate("lead", HashMap::new()).unwrap(); + let r = reg.readiness("lead").unwrap(); + assert!(r.requirements_met); + assert!(r.active); + assert!(!r.degraded); // all met, so not degraded + + reg.deactivate(instance.instance_id).unwrap(); + } + + #[test] + fn readiness_active_hand_degraded() { + let reg = HandRegistry::new(); + reg.load_bundled(); + + // Browser hand requires python3 + chromium. Activate it — if either + // requirement is unmet on this machine, it will show as degraded. + let instance = reg.activate("browser", HashMap::new()).unwrap(); + let r = reg.readiness("browser").unwrap(); + assert!(r.active); + + // If any requirement is not satisfied, degraded should be true + if !r.requirements_met { + assert!(r.degraded); + } else { + assert!(!r.degraded); + } + + reg.deactivate(instance.instance_id).unwrap(); + } + + #[test] + fn readiness_paused_hand_not_active() { + let reg = HandRegistry::new(); + reg.load_bundled(); + + let instance = reg.activate("lead", HashMap::new()).unwrap(); + reg.pause(instance.instance_id).unwrap(); + + let r = reg.readiness("lead").unwrap(); + assert!(!r.active); // Paused is not Active + assert!(!r.degraded); + + reg.deactivate(instance.instance_id).unwrap(); + } + + #[test] + fn optional_field_defaults_false() { + let req = HandRequirement { + key: "test".to_string(), + label: "test".to_string(), + requirement_type: RequirementType::Binary, + check_value: "test".to_string(), + description: None, + optional: false, + install: None, + }; + assert!(!req.optional); + } } diff --git a/crates/openfang-kernel/src/kernel.rs b/crates/openfang-kernel/src/kernel.rs index 8f6dbc6b..0b1b1cbe 100644 --- a/crates/openfang-kernel/src/kernel.rs +++ b/crates/openfang-kernel/src/kernel.rs @@ -569,6 +569,7 @@ impl OpenFangKernel { .base_url .clone() .or_else(|| config.provider_urls.get(&config.default_model.provider).cloned()), + skip_permissions: true, }; // Primary driver failure is non-fatal: the dashboard should remain accessible // even if the LLM provider is misconfigured. Users can fix config via dashboard. @@ -589,6 +590,7 @@ impl OpenFangKernel { provider: provider.to_string(), api_key: std::env::var(env_var).ok(), base_url: config.provider_urls.get(provider).cloned(), + skip_permissions: true, }; match drivers::create_driver(&auto_config) { Ok(d) => { @@ -633,6 +635,7 @@ impl OpenFangKernel { .base_url .clone() .or_else(|| config.provider_urls.get(&fb.provider).cloned()), + skip_permissions: true, }; match drivers::create_driver(&fb_config) { Ok(d) => { @@ -2624,20 +2627,51 @@ impl OpenFangKernel { } /// Switch an agent's model. - pub fn set_agent_model(&self, agent_id: AgentId, model: &str) -> KernelResult<()> { - // Resolve provider from model catalog so switching models also switches provider - let resolved_provider = self - .model_catalog - .read() - .ok() - .and_then(|catalog| { - catalog - .find_model(model) - .map(|entry| entry.provider.clone()) - }); + /// + /// When `explicit_provider` is `Some`, that provider name is used as-is + /// (respecting the user's custom configuration). When `None`, the provider + /// is auto-detected from the model catalog or inferred from the model name, + /// but only if the agent does NOT have a custom `base_url` configured. + /// Agents with a custom `base_url` keep their current provider unless + /// overridden explicitly — this prevents custom setups (e.g. Tencent, + /// Azure, or other third-party endpoints) from being misidentified. + pub fn set_agent_model( + &self, + agent_id: AgentId, + model: &str, + explicit_provider: Option<&str>, + ) -> KernelResult<()> { + let provider = if let Some(ep) = explicit_provider { + // User explicitly set the provider — use it as-is + Some(ep.to_string()) + } else { + // Check whether the agent has a custom base_url, which indicates + // a user-configured provider endpoint. In that case, preserve the + // current provider name instead of overriding it with auto-detection. + let has_custom_url = self + .registry + .get(agent_id) + .map(|e| e.manifest.model.base_url.is_some()) + .unwrap_or(false); - // If catalog lookup failed, try to infer provider from model name prefix - let provider = resolved_provider.or_else(|| infer_provider_from_model(model)); + if has_custom_url { + // Keep the current provider — don't let auto-detection override + // a deliberately configured custom endpoint. + None + } else { + // No custom base_url: safe to auto-detect from catalog / model name + let resolved_provider = self + .model_catalog + .read() + .ok() + .and_then(|catalog| { + catalog + .find_model(model) + .map(|entry| entry.provider.clone()) + }); + resolved_provider.or_else(|| infer_provider_from_model(model)) + } + }; // Strip the provider prefix from the model name (e.g. "openrouter/deepseek/deepseek-chat" → "deepseek/deepseek-chat") let normalized_model = if let Some(ref prov) = provider { @@ -2902,7 +2936,6 @@ impl OpenFangKernel { agent_id: AgentId, ) -> KernelResult { use openfang_runtime::compactor::generate_context_report; - use openfang_runtime::tool_runner::builtin_tool_definitions; let entry = self.registry.get(agent_id).ok_or_else(|| { KernelError::OpenFang(OpenFangError::AgentNotFound(agent_id.to_string())) @@ -2921,7 +2954,8 @@ impl OpenFangKernel { }); let system_prompt = &entry.manifest.model.system_prompt; - let tools = builtin_tool_definitions(); + // Use the agent's actual filtered tools instead of all builtins + let tools = self.available_tools(agent_id); // Use 200K default or the model's known context window let context_window = if session.context_window_tokens > 0 { session.context_window_tokens @@ -3063,6 +3097,18 @@ impl OpenFangKernel { } else { None }, + // Redundant safety: tool_allowlist mirrors capabilities.tools above. + // available_tools() already filters by capabilities.tools, but this + // provides defense-in-depth. + tool_allowlist: def.tools.clone(), + tool_blocklist: Vec::new(), + // Custom profile avoids ToolProfile-based expansion overriding the + // explicit tool list. + profile: if !def.tools.is_empty() { + Some(ToolProfile::Custom) + } else { + None + }, ..Default::default() }; @@ -3102,9 +3148,14 @@ impl OpenFangKernel { ); } - // If an agent with this hand's name already exists, remove it first + // If an agent with this hand's name already exists, remove it first. + // Save triggers before kill so they can be restored under the new ID + // (issue #519 — triggers were lost on agent restart). let existing = self.registry.list().into_iter().find(|e| e.name == def.agent.name); let old_agent_id = existing.as_ref().map(|e| e.id); + let saved_triggers = old_agent_id + .map(|id| self.triggers.take_agent_triggers(id)) + .unwrap_or_default(); if let Some(old) = existing { info!(agent = %old.name, id = %old.id, "Removing existing hand agent for reactivation"); let _ = self.kill_agent(old.id); @@ -3113,6 +3164,19 @@ impl OpenFangKernel { // Spawn the agent let agent_id = self.spawn_agent(manifest)?; + // Restore triggers from the old agent under the new agent ID (#519). + if !saved_triggers.is_empty() { + let restored = self.triggers.restore_triggers(agent_id, saved_triggers); + if restored > 0 { + info!( + old_agent = %old_agent_id.unwrap(), + new_agent = %agent_id, + restored, + "Reassigned triggers after hand reactivation" + ); + } + } + // Migrate cron jobs from old agent to new agent so they survive restarts. // Without this, persisted cron jobs would reference the stale old UUID // and fail silently (issue #461). @@ -3469,11 +3533,12 @@ impl OpenFangKernel { match self.activate_hand(&hand_id, config) { Ok(inst) => { info!(hand = %hand_id, instance = %inst.instance_id, "Hand restored"); - // Reassign cron jobs from the pre-restart agent ID to the - // newly spawned agent so scheduled tasks survive daemon - // restarts (issue #402). activate_hand only handles - // reassignment when an existing agent is found in the live - // registry, which is empty on a fresh boot. + // Reassign cron jobs and triggers from the pre-restart + // agent ID to the newly spawned agent so scheduled tasks + // and event triggers survive daemon restarts (issues + // #402, #519). activate_hand only handles reassignment + // when an existing agent is found in the live registry, + // which is empty on a fresh boot. if let (Some(old_id), Some(new_id)) = (old_agent_id, inst.agent_id) { if old_id != new_id { let migrated = @@ -3490,6 +3555,20 @@ impl OpenFangKernel { warn!("Failed to persist cron jobs after hand restore: {e}"); } } + // Reassign triggers (#519). Currently a no-op on + // cold boot (triggers are in-memory only), but + // correct if trigger persistence is added later. + let t_migrated = + self.triggers.reassign_agent_triggers(old_id, new_id); + if t_migrated > 0 { + info!( + hand = %hand_id, + old_agent = %old_id, + new_agent = %new_id, + migrated = t_migrated, + "Reassigned triggers after restart" + ); + } } } } @@ -4131,6 +4210,7 @@ impl OpenFangKernel { provider: agent_provider.clone(), api_key, base_url, + skip_permissions: true, }; match drivers::create_driver(&driver_config) { @@ -4176,6 +4256,7 @@ impl OpenFangKernel { .base_url .clone() .or_else(|| self.lookup_provider_url(&fb.provider)), + skip_permissions: true, }; match drivers::create_driver(&config) { Ok(d) => chain.push((d, fb.model.clone())), @@ -4506,11 +4587,18 @@ impl OpenFangKernel { } } - /// Get the list of tools available to an agent based on its capabilities. + /// Get the list of tools available to an agent based on its manifest. + /// + /// The agent's declared tools (`capabilities.tools`) are the primary filter. + /// Only tools listed there are sent to the LLM, saving tokens and preventing + /// the model from calling tools the agent isn't designed to use. + /// + /// If `capabilities.tools` is empty (or contains `"*"`), all tools are + /// available (backwards compatible). fn available_tools(&self, agent_id: AgentId) -> Vec { let all_builtins = builtin_tool_definitions(); - // Look up agent entry for profile, skill/MCP allowlists, and capabilities + // Look up agent entry for profile, skill/MCP allowlists, and declared tools let entry = self.registry.get(agent_id); let (skill_allowlist, mcp_allowlist, tool_profile) = entry .as_ref() @@ -4523,31 +4611,51 @@ impl OpenFangKernel { }) .unwrap_or_default(); - // Filter builtin tools by ToolProfile (if set and not Full). - // This is the primary token-saving mechanism: a chat agent with ToolProfile::Minimal - // gets 2 tools instead of 46+, saving ~15-20K tokens of tool definitions. + // Extract the agent's declared tool list from capabilities.tools. + // This is the primary mechanism: only send declared tools to the LLM. + let declared_tools: Vec = entry + .as_ref() + .map(|e| e.manifest.capabilities.tools.clone()) + .unwrap_or_default(); + + // Check if the agent has unrestricted tool access: + // - capabilities.tools is empty (not specified → all tools) + // - capabilities.tools contains "*" (explicit wildcard) + let tools_unrestricted = declared_tools.is_empty() + || declared_tools.iter().any(|t| t == "*"); + + // Step 1: Filter builtin tools. + // Priority: declared tools > ToolProfile > all builtins. let has_tool_all = entry.as_ref().is_some_and(|_| { let caps = self.capabilities.list(agent_id); caps.iter().any(|c| matches!(c, Capability::ToolAll)) }); - let mut all_tools = match &tool_profile { - Some(profile) if *profile != ToolProfile::Full && *profile != ToolProfile::Custom => { - let allowed = profile.tools(); - all_builtins - .into_iter() - .filter(|t| allowed.iter().any(|a| a == "*" || a == &t.name)) - .collect() + let mut all_tools: Vec = if !tools_unrestricted { + // Agent declares specific tools — only include matching builtins + all_builtins + .into_iter() + .filter(|t| declared_tools.iter().any(|d| d == &t.name)) + .collect() + } else { + // No specific tools declared — fall back to profile or all builtins + match &tool_profile { + Some(profile) + if *profile != ToolProfile::Full && *profile != ToolProfile::Custom => + { + let allowed = profile.tools(); + all_builtins + .into_iter() + .filter(|t| allowed.iter().any(|a| a == "*" || a == &t.name)) + .collect() + } + _ if has_tool_all => all_builtins, + _ => all_builtins, } - _ if has_tool_all => all_builtins, - _ => all_builtins, }; - // Track tool names added by skills/MCP — these already passed their own - // allowlist filters and should bypass the per-agent capability check. - let mut extension_tool_names = std::collections::HashSet::new(); - - // Add skill-provided tools (filtered by agent's skill allowlist) + // Step 2: Add skill-provided tools (filtered by agent's skill allowlist, + // then by declared tools). let skill_tools = { let registry = self .skill_registry @@ -4560,7 +4668,12 @@ impl OpenFangKernel { } }; for skill_tool in skill_tools { - extension_tool_names.insert(skill_tool.name.clone()); + // If agent declares specific tools, only include matching skill tools + if !tools_unrestricted + && !declared_tools.iter().any(|d| d == &skill_tool.name) + { + continue; + } all_tools.push(ToolDefinition { name: skill_tool.name.clone(), description: skill_tool.description.clone(), @@ -4568,20 +4681,17 @@ impl OpenFangKernel { }); } - // Add MCP tools (filtered by agent's MCP server allowlist) + // Step 3: Add MCP tools (filtered by agent's MCP server allowlist, + // then by declared tools). if let Ok(mcp_tools) = self.mcp_tools.lock() { - if mcp_allowlist.is_empty() { - for t in mcp_tools.iter() { - extension_tool_names.insert(t.name.clone()); - } - all_tools.extend(mcp_tools.iter().cloned()); + let mcp_candidates: Vec = if mcp_allowlist.is_empty() { + mcp_tools.iter().cloned().collect() } else { - // Normalize allowlist names for matching let normalized: Vec = mcp_allowlist .iter() .map(|s| openfang_runtime::mcp::normalize_name(s)) .collect(); - let filtered: Vec<_> = mcp_tools + mcp_tools .iter() .filter(|t| { openfang_runtime::mcp::extract_mcp_server(&t.name) @@ -4589,15 +4699,19 @@ impl OpenFangKernel { .unwrap_or(false) }) .cloned() - .collect(); - for t in &filtered { - extension_tool_names.insert(t.name.clone()); + .collect() + }; + for t in mcp_candidates { + // If agent declares specific tools, only include matching MCP tools + if !tools_unrestricted && !declared_tools.iter().any(|d| d == &t.name) { + continue; } - all_tools.extend(filtered); + all_tools.push(t); } } - // Apply per-agent tool allowlist/blocklist (manifest-level filtering) + // Step 4: Apply per-agent tool_allowlist/tool_blocklist overrides. + // These are separate from capabilities.tools and act as additional filters. let (tool_allowlist, tool_blocklist) = entry .as_ref() .map(|e| (e.manifest.tool_allowlist.clone(), e.manifest.tool_blocklist.clone())) @@ -4610,8 +4724,7 @@ impl OpenFangKernel { all_tools.retain(|t| !tool_blocklist.iter().any(|b| b == &t.name)); } - // Remove shell_exec from tool list if exec_policy won't allow it, - // so the LLM doesn't try to call a tool that will be blocked. + // Step 5: Remove shell_exec if exec_policy denies it. let exec_blocks_shell = entry.as_ref().is_some_and(|e| { e.manifest .exec_policy @@ -4622,26 +4735,7 @@ impl OpenFangKernel { all_tools.retain(|t| t.name != "shell_exec"); } - let caps = self.capabilities.list(agent_id); - - // If agent has ToolAll, return all tools - if caps.iter().any(|c| matches!(c, Capability::ToolAll)) { - return all_tools; - } - - // Filter to tools the agent has capability for. - // MCP and skill tools bypass this check — they already passed their - // own allowlist filters above (mcp_servers / skills on the manifest). all_tools - .into_iter() - .filter(|tool| { - extension_tool_names.contains(&tool.name) - || caps.iter().any(|c| match c { - Capability::ToolInvoke(name) => name == &tool.name || name == "*", - _ => false, - }) - }) - .collect() } /// Collect prompt context from prompt-only skills for system prompt injection. @@ -5535,6 +5629,7 @@ impl KernelHandle for OpenFangKernel { channel: &str, recipient: &str, message: &str, + thread_id: Option<&str>, ) -> Result { let adapter = self .channel_adapters @@ -5558,10 +5653,19 @@ impl KernelHandle for OpenFangKernel { openfang_user: None, }; - adapter - .send(&user, openfang_channels::types::ChannelContent::Text(message.to_string())) - .await - .map_err(|e| format!("Channel send failed: {e}"))?; + let content = openfang_channels::types::ChannelContent::Text(message.to_string()); + + if let Some(tid) = thread_id { + adapter + .send_in_thread(&user, content, tid) + .await + .map_err(|e| format!("Channel send failed: {e}"))?; + } else { + adapter + .send(&user, content) + .await + .map_err(|e| format!("Channel send failed: {e}"))?; + } Ok(format!("Message sent to {} via {}", recipient, channel)) } @@ -5574,6 +5678,7 @@ impl KernelHandle for OpenFangKernel { media_url: &str, caption: Option<&str>, filename: Option<&str>, + thread_id: Option<&str>, ) -> Result { let adapter = self .channel_adapters @@ -5611,14 +5716,73 @@ impl KernelHandle for OpenFangKernel { } }; - adapter - .send(&user, content) - .await - .map_err(|e| format!("Channel media send failed: {e}"))?; + if let Some(tid) = thread_id { + adapter + .send_in_thread(&user, content, tid) + .await + .map_err(|e| format!("Channel media send failed: {e}"))?; + } else { + adapter + .send(&user, content) + .await + .map_err(|e| format!("Channel media send failed: {e}"))?; + } Ok(format!("{} sent to {} via {}", media_type, recipient, channel)) } + async fn send_channel_file_data( + &self, + channel: &str, + recipient: &str, + data: Vec, + filename: &str, + mime_type: &str, + thread_id: Option<&str>, + ) -> Result { + let adapter = self + .channel_adapters + .get(channel) + .ok_or_else(|| { + let available: Vec = self + .channel_adapters + .iter() + .map(|e| e.key().clone()) + .collect(); + format!( + "Channel '{}' not found. Available channels: {:?}", + channel, available + ) + })? + .clone(); + + let user = openfang_channels::types::ChannelUser { + platform_id: recipient.to_string(), + display_name: recipient.to_string(), + openfang_user: None, + }; + + let content = openfang_channels::types::ChannelContent::FileData { + data, + filename: filename.to_string(), + mime_type: mime_type.to_string(), + }; + + if let Some(tid) = thread_id { + adapter + .send_in_thread(&user, content, tid) + .await + .map_err(|e| format!("Channel file send failed: {e}"))?; + } else { + adapter + .send(&user, content) + .await + .map_err(|e| format!("Channel file send failed: {e}"))?; + } + + Ok(format!("File '{}' sent to {} via {}", filename, recipient, channel)) + } + async fn spawn_agent_checked( &self, manifest_toml: &str, diff --git a/crates/openfang-kernel/src/triggers.rs b/crates/openfang-kernel/src/triggers.rs index 7803a1ad..32eaa93b 100644 --- a/crates/openfang-kernel/src/triggers.rs +++ b/crates/openfang-kernel/src/triggers.rs @@ -143,6 +143,109 @@ impl TriggerEngine { } } + /// Take all triggers for an agent, removing them from the engine. + /// + /// Returns the extracted triggers so they can be restored under a + /// different agent ID via [`restore_triggers`]. This is used during + /// hand reactivation: triggers must be saved before `kill_agent` + /// destroys them, then restored with the new agent ID after spawn. + pub fn take_agent_triggers(&self, agent_id: AgentId) -> Vec { + let trigger_ids = self + .agent_triggers + .remove(&agent_id) + .map(|(_, ids)| ids) + .unwrap_or_default(); + let mut taken = Vec::with_capacity(trigger_ids.len()); + for id in trigger_ids { + if let Some((_, t)) = self.triggers.remove(&id) { + taken.push(t); + } + } + if !taken.is_empty() { + info!( + agent = %agent_id, + count = taken.len(), + "Took triggers for agent (pending reassignment)" + ); + } + taken + } + + /// Restore previously taken triggers under a new agent ID. + /// + /// Each trigger keeps its original pattern, prompt template, fire count, + /// and max_fires, but is re-keyed to `new_agent_id`. New trigger IDs are + /// generated so there are no stale references. + /// + /// Returns the number of triggers restored. + pub fn restore_triggers(&self, new_agent_id: AgentId, triggers: Vec) -> usize { + let count = triggers.len(); + for old in triggers { + let new_id = TriggerId::new(); + let trigger = Trigger { + id: new_id, + agent_id: new_agent_id, + pattern: old.pattern, + prompt_template: old.prompt_template, + enabled: old.enabled, + created_at: old.created_at, + fire_count: old.fire_count, + max_fires: old.max_fires, + }; + self.triggers.insert(new_id, trigger); + self.agent_triggers + .entry(new_agent_id) + .or_default() + .push(new_id); + } + if count > 0 { + info!( + agent = %new_agent_id, + count, + "Restored triggers under new agent" + ); + } + count + } + + /// Reassign all triggers from one agent to another in place. + /// + /// Used during cold boot when the old agent ID (from persisted state) no + /// longer exists and a new agent was spawned. Updates the `agent_id` field + /// on each trigger and moves the index entry. + /// + /// Returns the number of triggers reassigned. + pub fn reassign_agent_triggers( + &self, + old_agent_id: AgentId, + new_agent_id: AgentId, + ) -> usize { + let trigger_ids = self + .agent_triggers + .remove(&old_agent_id) + .map(|(_, ids)| ids) + .unwrap_or_default(); + let count = trigger_ids.len(); + for id in &trigger_ids { + if let Some(mut t) = self.triggers.get_mut(id) { + t.agent_id = new_agent_id; + } + } + if !trigger_ids.is_empty() { + self.agent_triggers + .entry(new_agent_id) + .or_default() + .extend(trigger_ids); + info!( + old_agent = %old_agent_id, + new_agent = %new_agent_id, + count, + "Reassigned triggers to new agent" + ); + } + count + } + /// Enable or disable a trigger. Returns true if the trigger was found. pub fn set_enabled(&self, trigger_id: TriggerId, enabled: bool) -> bool { if let Some(mut t) = self.triggers.get_mut(&trigger_id) { @@ -508,4 +611,122 @@ mod tests { ); assert_eq!(engine.evaluate(&event).len(), 1); } + + // -- reassign_agent_triggers (#519) ------------------------------------ + + #[test] + fn test_reassign_agent_triggers_basic() { + let engine = TriggerEngine::new(); + let old_agent = AgentId::new(); + let new_agent = AgentId::new(); + engine.register(old_agent, TriggerPattern::All, "a".to_string(), 0); + engine.register(old_agent, TriggerPattern::System, "b".to_string(), 0); + + let count = engine.reassign_agent_triggers(old_agent, new_agent); + assert_eq!(count, 2); + assert_eq!(engine.list_agent_triggers(old_agent).len(), 0); + assert_eq!(engine.list_agent_triggers(new_agent).len(), 2); + + // Verify triggers actually fire for the new agent + let event = Event::new( + AgentId::new(), + EventTarget::Broadcast, + EventPayload::System(SystemEvent::HealthCheck { + status: "ok".to_string(), + }), + ); + let matches = engine.evaluate(&event); + assert_eq!(matches.len(), 2); + assert!(matches.iter().all(|(id, _)| *id == new_agent)); + } + + #[test] + fn test_reassign_agent_triggers_no_match_returns_zero() { + let engine = TriggerEngine::new(); + let agent_a = AgentId::new(); + engine.register(agent_a, TriggerPattern::All, "a".to_string(), 0); + + let count = engine.reassign_agent_triggers(AgentId::new(), AgentId::new()); + assert_eq!(count, 0); + // Original triggers untouched + assert_eq!(engine.list_agent_triggers(agent_a).len(), 1); + } + + #[test] + fn test_reassign_does_not_touch_other_agents() { + let engine = TriggerEngine::new(); + let agent_a = AgentId::new(); + let agent_b = AgentId::new(); + let agent_c = AgentId::new(); + engine.register(agent_a, TriggerPattern::All, "a".to_string(), 0); + engine.register(agent_b, TriggerPattern::System, "b".to_string(), 0); + + let count = engine.reassign_agent_triggers(agent_a, agent_c); + assert_eq!(count, 1); + // agent_b untouched + assert_eq!(engine.list_agent_triggers(agent_b).len(), 1); + assert_eq!(engine.list_agent_triggers(agent_c).len(), 1); + } + + // -- take / restore triggers (#519) ------------------------------------ + + #[test] + fn test_take_and_restore_triggers() { + let engine = TriggerEngine::new(); + let old_agent = AgentId::new(); + let new_agent = AgentId::new(); + engine.register( + old_agent, + TriggerPattern::ContentMatch { + substring: "deploy".to_string(), + }, + "Deploy alert: {{event}}".to_string(), + 5, + ); + engine.register(old_agent, TriggerPattern::Lifecycle, "lc".to_string(), 0); + + // Take triggers — engine should be empty for old agent + let taken = engine.take_agent_triggers(old_agent); + assert_eq!(taken.len(), 2); + assert_eq!(engine.list_agent_triggers(old_agent).len(), 0); + assert_eq!(engine.list_all().len(), 0); + + // Restore under new agent + let restored = engine.restore_triggers(new_agent, taken); + assert_eq!(restored, 2); + assert_eq!(engine.list_agent_triggers(new_agent).len(), 2); + + // Verify patterns and max_fires are preserved + let triggers = engine.list_agent_triggers(new_agent); + let has_content_match = triggers.iter().any(|t| { + matches!(&t.pattern, TriggerPattern::ContentMatch { substring } if substring == "deploy") + && t.max_fires == 5 + }); + assert!(has_content_match, "ContentMatch trigger with max_fires=5 should be preserved"); + } + + #[test] + fn test_take_empty_returns_empty() { + let engine = TriggerEngine::new(); + let taken = engine.take_agent_triggers(AgentId::new()); + assert!(taken.is_empty()); + } + + #[test] + fn test_restore_preserves_enabled_state() { + let engine = TriggerEngine::new(); + let old_agent = AgentId::new(); + let new_agent = AgentId::new(); + let tid = engine.register(old_agent, TriggerPattern::All, "a".to_string(), 0); + engine.set_enabled(tid, false); + + let taken = engine.take_agent_triggers(old_agent); + assert_eq!(taken.len(), 1); + assert!(!taken[0].enabled); + + engine.restore_triggers(new_agent, taken); + let restored = engine.list_agent_triggers(new_agent); + assert_eq!(restored.len(), 1); + assert!(!restored[0].enabled, "Disabled state should survive take/restore"); + } } diff --git a/crates/openfang-memory/src/session.rs b/crates/openfang-memory/src/session.rs index b04149f5..74862c37 100644 --- a/crates/openfang-memory/src/session.rs +++ b/crates/openfang-memory/src/session.rs @@ -557,7 +557,7 @@ impl SessionStore { MessageContent::Blocks(blocks) => { for block in blocks { match block { - ContentBlock::Text { text } => { + ContentBlock::Text { text, .. } => { text_parts.push(text.clone()); } ContentBlock::ToolUse { id, name, input, .. } => { diff --git a/crates/openfang-runtime/src/agent_loop.rs b/crates/openfang-runtime/src/agent_loop.rs index 1e73f7f9..8967c754 100644 --- a/crates/openfang-runtime/src/agent_loop.rs +++ b/crates/openfang-runtime/src/agent_loop.rs @@ -730,6 +730,7 @@ pub async fn run_agent_loop( wanted to do and that it requires their approval.]", denial_count ), + provider_metadata: None, }); } @@ -747,6 +748,7 @@ pub async fn run_agent_loop( alternatives instead of making up data.]", non_denial_errors ), + provider_metadata: None, }); } @@ -1291,9 +1293,15 @@ pub async fn run_agent_loop_streaming( thinking: None, }; - // Notify phase: Streaming (streaming variant always streams) + // Notify phase: on first iteration emit Streaming; on subsequent + // iterations (after tool execution) emit Thinking so the UI shows + // "Thinking..." instead of overwriting streamed text with "streaming". if let Some(cb) = on_phase { - cb(LoopPhase::Streaming); + if iteration == 0 { + cb(LoopPhase::Streaming); + } else { + cb(LoopPhase::Thinking); + } } // Stream LLM call with retry, error classification, and circuit breaker @@ -1707,6 +1715,7 @@ pub async fn run_agent_loop_streaming( wanted to do and that it requires their approval.]", denial_count ), + provider_metadata: None, }); } @@ -1724,6 +1733,7 @@ pub async fn run_agent_loop_streaming( alternatives instead of making up data.]", non_denial_errors ), + provider_metadata: None, }); } @@ -2633,6 +2643,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: "Hello from the agent!".to_string(), + provider_metadata: None, }], stop_reason: StopReason::EndTurn, tool_calls: vec![], @@ -2885,6 +2896,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: "Recovered after retry!".to_string(), + provider_metadata: None, }], stop_reason: StopReason::EndTurn, tool_calls: vec![], @@ -3754,6 +3766,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: r#"Let me search for that. {"query":"rust async"}"#.to_string(), + provider_metadata: None, }], stop_reason: StopReason::EndTurn, tool_calls: vec![], // BUG: no tool_calls! @@ -3767,6 +3780,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: "Based on the search results, Rust async is great!".to_string(), + provider_metadata: None, }], stop_reason: StopReason::EndTurn, tool_calls: vec![], diff --git a/crates/openfang-runtime/src/compactor.rs b/crates/openfang-runtime/src/compactor.rs index 50fe9248..85570546 100644 --- a/crates/openfang-runtime/src/compactor.rs +++ b/crates/openfang-runtime/src/compactor.rs @@ -356,7 +356,7 @@ fn build_conversation_text(messages: &[Message], config: &CompactionConfig) -> S MessageContent::Blocks(blocks) => { for block in blocks { match block { - ContentBlock::Text { text } => { + ContentBlock::Text { text, .. } => { if !text.is_empty() { if oversized && text.len() > config.max_chunk_chars / 4 { let limit = config.max_chunk_chars / 4; @@ -448,6 +448,7 @@ async fn summarize_messages( role: Role::User, content: MessageContent::Blocks(vec![ContentBlock::Text { text: summarize_prompt, + provider_metadata: None, }]), }], tools: vec![], @@ -563,7 +564,7 @@ async fn summarize_in_chunks( model: model.to_string(), messages: vec![Message { role: Role::User, - content: MessageContent::Blocks(vec![ContentBlock::Text { text: merge_prompt }]), + content: MessageContent::Blocks(vec![ContentBlock::Text { text: merge_prompt, provider_metadata: None }]), }], tools: vec![], max_tokens: config.max_summary_tokens, @@ -766,6 +767,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: "Summary of conversation".to_string(), + provider_metadata: None, }], stop_reason: openfang_types::message::StopReason::EndTurn, tool_calls: vec![], @@ -827,6 +829,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: "Summary with tools".to_string(), + provider_metadata: None, }], stop_reason: openfang_types::message::StopReason::EndTurn, tool_calls: vec![], @@ -919,6 +922,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: "Summary: discussed topics 0 through 79".to_string(), + provider_metadata: None, }], stop_reason: openfang_types::message::StopReason::EndTurn, tool_calls: vec![], @@ -1114,6 +1118,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: format!("Chunk summary {n}"), + provider_metadata: None, }], stop_reason: openfang_types::message::StopReason::EndTurn, tool_calls: vec![], @@ -1180,6 +1185,7 @@ mod tests { content: MessageContent::Blocks(vec![ ContentBlock::Text { text: "Let me search".to_string(), + provider_metadata: None, }, ContentBlock::ToolUse { id: "tu-1".to_string(), diff --git a/crates/openfang-runtime/src/drivers/anthropic.rs b/crates/openfang-runtime/src/drivers/anthropic.rs index 6f1179ac..857774e2 100644 --- a/crates/openfang-runtime/src/drivers/anthropic.rs +++ b/crates/openfang-runtime/src/drivers/anthropic.rs @@ -507,7 +507,7 @@ impl LlmDriver for AnthropicDriver { for block in blocks { match block { ContentBlockAccum::Text(text) => { - content.push(ContentBlock::Text { text }); + content.push(ContentBlock::Text { text, provider_metadata: None }); } ContentBlockAccum::Thinking(thinking) => { content.push(ContentBlock::Thinking { thinking }); @@ -563,7 +563,7 @@ fn convert_message(msg: &Message) -> ApiMessage { let api_blocks: Vec = blocks .iter() .filter_map(|block| match block { - ContentBlock::Text { text } => { + ContentBlock::Text { text, .. } => { Some(ApiContentBlock::Text { text: text.clone() }) } ContentBlock::Image { media_type, data } => Some(ApiContentBlock::Image { @@ -610,7 +610,7 @@ fn convert_response(api: ApiResponse) -> CompletionResponse { for block in api.content { match block { ResponseContentBlock::Text { text } => { - content.push(ContentBlock::Text { text }); + content.push(ContentBlock::Text { text, provider_metadata: None }); } ResponseContentBlock::ToolUse { id, name, input } => { content.push(ContentBlock::ToolUse { diff --git a/crates/openfang-runtime/src/drivers/claude_code.rs b/crates/openfang-runtime/src/drivers/claude_code.rs index 547ce343..1cdfe3b4 100644 --- a/crates/openfang-runtime/src/drivers/claude_code.rs +++ b/crates/openfang-runtime/src/drivers/claude_code.rs @@ -47,17 +47,29 @@ const SENSITIVE_SUFFIXES: &[&str] = &["_SECRET", "_TOKEN", "_PASSWORD"]; /// LLM driver that delegates to the Claude Code CLI. pub struct ClaudeCodeDriver { cli_path: String, + skip_permissions: bool, } impl ClaudeCodeDriver { /// Create a new Claude Code driver. /// /// `cli_path` overrides the CLI binary path; defaults to `"claude"` on PATH. - pub fn new(cli_path: Option) -> Self { + /// `skip_permissions` adds `--dangerously-skip-permissions` to the spawned + /// command so that the CLI runs non-interactively (required for daemon mode). + pub fn new(cli_path: Option, skip_permissions: bool) -> Self { + if skip_permissions { + warn!( + "Claude Code driver: --dangerously-skip-permissions enabled. \ + The CLI will not prompt for tool approvals. \ + OpenFang's own capability/RBAC system enforces access control." + ); + } + Self { cli_path: cli_path .filter(|s| !s.is_empty()) .unwrap_or_else(|| "claude".to_string()), + skip_permissions, } } @@ -193,6 +205,10 @@ impl LlmDriver for ClaudeCodeDriver { .arg("--output-format") .arg("json"); + if self.skip_permissions { + cmd.arg("--dangerously-skip-permissions"); + } + if let Some(ref model) = model_flag { cmd.arg("--model").arg(model); } @@ -202,7 +218,7 @@ impl LlmDriver for ClaudeCodeDriver { cmd.stdout(std::process::Stdio::piped()); cmd.stderr(std::process::Stdio::piped()); - debug!(cli = %self.cli_path, "Spawning Claude Code CLI"); + debug!(cli = %self.cli_path, skip_permissions = self.skip_permissions, "Spawning Claude Code CLI"); let output = cmd .output() @@ -255,7 +271,7 @@ impl LlmDriver for ClaudeCodeDriver { .unwrap_or_default(); let usage = parsed.usage.unwrap_or_default(); return Ok(CompletionResponse { - content: vec![ContentBlock::Text { text: text.clone() }], + content: vec![ContentBlock::Text { text: text.clone(), provider_metadata: None }], stop_reason: StopReason::EndTurn, tool_calls: Vec::new(), usage: TokenUsage { @@ -268,7 +284,7 @@ impl LlmDriver for ClaudeCodeDriver { // Fallback: treat entire stdout as plain text let text = stdout.trim().to_string(); Ok(CompletionResponse { - content: vec![ContentBlock::Text { text }], + content: vec![ContentBlock::Text { text, provider_metadata: None }], stop_reason: StopReason::EndTurn, tool_calls: Vec::new(), usage: TokenUsage { @@ -293,6 +309,10 @@ impl LlmDriver for ClaudeCodeDriver { .arg("stream-json") .arg("--verbose"); + if self.skip_permissions { + cmd.arg("--dangerously-skip-permissions"); + } + if let Some(ref model) = model_flag { cmd.arg("--model").arg(model); } @@ -302,7 +322,7 @@ impl LlmDriver for ClaudeCodeDriver { cmd.stdout(std::process::Stdio::piped()); cmd.stderr(std::process::Stdio::piped()); - debug!(cli = %self.cli_path, "Spawning Claude Code CLI (streaming)"); + debug!(cli = %self.cli_path, skip_permissions = self.skip_permissions, "Spawning Claude Code CLI (streaming)"); let mut child = cmd .spawn() @@ -404,7 +424,7 @@ impl LlmDriver for ClaudeCodeDriver { .await; Ok(CompletionResponse { - content: vec![ContentBlock::Text { text: full_text }], + content: vec![ContentBlock::Text { text: full_text, provider_metadata: None }], stop_reason: StopReason::EndTurn, tool_calls: Vec::new(), usage: final_usage, @@ -495,22 +515,29 @@ mod tests { #[test] fn test_new_defaults_to_claude() { - let driver = ClaudeCodeDriver::new(None); + let driver = ClaudeCodeDriver::new(None, true); assert_eq!(driver.cli_path, "claude"); + assert!(driver.skip_permissions); } #[test] fn test_new_with_custom_path() { - let driver = ClaudeCodeDriver::new(Some("/usr/local/bin/claude".to_string())); + let driver = ClaudeCodeDriver::new(Some("/usr/local/bin/claude".to_string()), true); assert_eq!(driver.cli_path, "/usr/local/bin/claude"); } #[test] fn test_new_with_empty_path() { - let driver = ClaudeCodeDriver::new(Some(String::new())); + let driver = ClaudeCodeDriver::new(Some(String::new()), true); assert_eq!(driver.cli_path, "claude"); } + #[test] + fn test_skip_permissions_disabled() { + let driver = ClaudeCodeDriver::new(None, false); + assert!(!driver.skip_permissions); + } + #[test] fn test_sensitive_env_list_coverage() { // Ensure all major provider keys are in the strip list diff --git a/crates/openfang-runtime/src/drivers/fallback.rs b/crates/openfang-runtime/src/drivers/fallback.rs index c6094f06..c72ca1bf 100644 --- a/crates/openfang-runtime/src/drivers/fallback.rs +++ b/crates/openfang-runtime/src/drivers/fallback.rs @@ -140,6 +140,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: "OK".to_string(), + provider_metadata: None, }], stop_reason: StopReason::EndTurn, tool_calls: vec![], diff --git a/crates/openfang-runtime/src/drivers/gemini.rs b/crates/openfang-runtime/src/drivers/gemini.rs index 03585513..f70a9efe 100644 --- a/crates/openfang-runtime/src/drivers/gemini.rs +++ b/crates/openfang-runtime/src/drivers/gemini.rs @@ -64,19 +64,41 @@ struct GeminiContent { } /// A part within a content entry. +/// +/// Gemini 2.5+/3.x thinking models include `thoughtSignature` at the **part +/// level** (sibling of `functionCall`/`text`, not nested inside them). When +/// echoing model turns back in the conversation history, the signature must be +/// preserved on every part that originally carried one — otherwise the API +/// returns `INVALID_ARGUMENT: Function call is missing a thought_signature`. #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(untagged)] enum GeminiPart { + /// Text part, optionally carrying a thought signature. Text { text: String, + /// Part-level thought signature (Gemini 2.5+/3.x thinking models). + #[serde( + rename = "thoughtSignature", + default, + skip_serializing_if = "Option::is_none" + )] + thought_signature: Option, }, InlineData { #[serde(rename = "inlineData")] inline_data: GeminiInlineData, }, + /// Function call part with an optional part-level thought signature. FunctionCall { #[serde(rename = "functionCall")] function_call: GeminiFunctionCallData, + /// Part-level thought signature (Gemini 2.5+/3.x thinking models). + #[serde( + rename = "thoughtSignature", + default, + skip_serializing_if = "Option::is_none" + )] + thought_signature: Option, }, FunctionResponse { #[serde(rename = "functionResponse")] @@ -95,9 +117,6 @@ struct GeminiInlineData { struct GeminiFunctionCallData { name: String, args: serde_json::Value, - /// Gemini 2.5+ thinking models return this on functionCall parts. - #[serde(rename = "thoughtSignature", default, skip_serializing_if = "Option::is_none")] - thought_signature: Option, } #[derive(Debug, Serialize, Deserialize, Clone)] @@ -216,13 +235,30 @@ fn convert_messages( }; let parts = match &msg.content { - MessageContent::Text(text) => vec![GeminiPart::Text { text: text.clone() }], + MessageContent::Text(text) => vec![GeminiPart::Text { + text: text.clone(), + thought_signature: None, + }], MessageContent::Blocks(blocks) => { let mut parts = Vec::new(); for block in blocks { match block { - ContentBlock::Text { text } => { - parts.push(GeminiPart::Text { text: text.clone() }); + ContentBlock::Text { + text, + provider_metadata, + } => { + // Echo back thought_signature from provider_metadata + // if present — required by Gemini 3.x thinking models + // on ALL parts, including text parts. + let thought_signature = provider_metadata + .as_ref() + .and_then(|m| m.get("thought_signature")) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + parts.push(GeminiPart::Text { + text: text.clone(), + thought_signature, + }); } ContentBlock::ToolUse { name, @@ -231,7 +267,9 @@ fn convert_messages( .. } => { // Echo back thought_signature from provider_metadata - // if present — required by Gemini 2.5+ thinking models. + // if present — required by Gemini 2.5+/3.x thinking models. + // The signature lives at the part level (sibling of + // functionCall), not inside the functionCall object. let thought_signature = provider_metadata .as_ref() .and_then(|m| m.get("thought_signature")) @@ -241,8 +279,8 @@ fn convert_messages( function_call: GeminiFunctionCallData { name: name.clone(), args: input.clone(), - thought_signature, }, + thought_signature, }); } ContentBlock::Image { media_type, data } => { @@ -304,7 +342,10 @@ fn extract_system(messages: &[Message], system: &Option) -> Option Result { for part in gemini_content.parts { match part { - GeminiPart::Text { text } => { + GeminiPart::Text { + text, + thought_signature, + } => { if !text.is_empty() { - content.push(ContentBlock::Text { text }); - } - } - GeminiPart::FunctionCall { function_call } => { - let id = format!("call_{}", uuid::Uuid::new_v4().simple()); - // Preserve thought_signature in provider_metadata so it - // gets echoed back on the next request (Gemini 2.5+ requirement). - let provider_metadata = - function_call.thought_signature.as_ref().map(|sig| { + // Preserve thought_signature in provider_metadata so + // it gets echoed back on the next request. Gemini + // 3.x thinking models include thoughtSignature on + // ALL parts (text + functionCall). + let provider_metadata = thought_signature.map(|sig| { serde_json::json!({ "thought_signature": sig }) }); + content.push(ContentBlock::Text { + text, + provider_metadata, + }); + } + } + GeminiPart::FunctionCall { + function_call, + thought_signature, + } => { + let id = format!("call_{}", uuid::Uuid::new_v4().simple()); + // Preserve thought_signature in provider_metadata so it + // gets echoed back on the next request (Gemini 2.5+/3.x). + // The signature lives at the part level, not inside + // functionCall. + let provider_metadata = thought_signature.map(|sig| { + serde_json::json!({ "thought_signature": sig }) + }); content.push(ContentBlock::ToolUse { id: id.clone(), name: function_call.name.clone(), @@ -577,6 +635,8 @@ impl LlmDriver for GeminiDriver { // Parse SSE stream let mut buffer = String::new(); let mut text_content = String::new(); + // Thought signature for accumulated text content (last one wins) + let mut text_thought_sig: Option = None; // Track function calls: (name, args_json, thought_signature) let mut fn_calls: Vec<(String, serde_json::Value, Option)> = Vec::new(); let mut finish_reason: Option = None; @@ -621,15 +681,26 @@ impl LlmDriver for GeminiDriver { if let Some(ref content) = candidate.content { for part in &content.parts { match part { - GeminiPart::Text { text } => { + GeminiPart::Text { + text, + thought_signature, + } => { if !text.is_empty() { text_content.push_str(text); let _ = tx .send(StreamEvent::TextDelta { text: text.clone() }) .await; } + // Capture thought signature for text parts + // (last one wins across streamed chunks). + if thought_signature.is_some() { + text_thought_sig = thought_signature.clone(); + } } - GeminiPart::FunctionCall { function_call } => { + GeminiPart::FunctionCall { + function_call, + thought_signature, + } => { let id = format!("call_{}", uuid::Uuid::new_v4().simple()); let _ = tx .send(StreamEvent::ToolUseStart { @@ -652,7 +723,7 @@ impl LlmDriver for GeminiDriver { fn_calls.push(( function_call.name.clone(), function_call.args.clone(), - function_call.thought_signature.clone(), + thought_signature.clone(), )); } GeminiPart::InlineData { .. } @@ -669,7 +740,13 @@ impl LlmDriver for GeminiDriver { let mut tool_calls = Vec::new(); if !text_content.is_empty() { - content.push(ContentBlock::Text { text: text_content }); + let provider_metadata = text_thought_sig.map(|sig| { + serde_json::json!({ "thought_signature": sig }) + }); + content.push(ContentBlock::Text { + text: text_content, + provider_metadata, + }); } for (name, args, thought_sig) in fn_calls { @@ -744,12 +821,14 @@ mod tests { role: Some("user".to_string()), parts: vec![GeminiPart::Text { text: "Hello".to_string(), + thought_signature: None, }], }], system_instruction: Some(GeminiContent { role: None, parts: vec![GeminiPart::Text { text: "You are helpful.".to_string(), + thought_signature: None, }], }), tools: vec![], @@ -843,7 +922,7 @@ mod tests { let sys = sys_instruction.unwrap(); assert!(sys.role.is_none()); match &sys.parts[0] { - GeminiPart::Text { text } => assert_eq!(text, "Be helpful."), + GeminiPart::Text { text, .. } => assert_eq!(text, "Be helpful."), _ => panic!("Expected text part"), } } @@ -908,6 +987,7 @@ mod tests { role: Some("model".to_string()), parts: vec![GeminiPart::Text { text: "Hello!".to_string(), + thought_signature: None, }], }), finish_reason: Some("STOP".to_string()), @@ -946,6 +1026,7 @@ mod tests { role: Some("model".to_string()), parts: vec![GeminiPart::Text { text: "Truncated...".to_string(), + thought_signature: None, }], }), finish_reason: Some("MAX_TOKENS".to_string()), @@ -976,7 +1057,7 @@ mod tests { let result = extract_system(&messages, &system); assert!(result.is_some()); match &result.unwrap().parts[0] { - GeminiPart::Text { text } => assert_eq!(text, "Be concise."), + GeminiPart::Text { text, .. } => assert_eq!(text, "Be concise."), _ => panic!("Expected text"), } } @@ -993,7 +1074,7 @@ mod tests { let result = extract_system(&messages, &None); assert!(result.is_some()); match &result.unwrap().parts[0] { - GeminiPart::Text { text } => assert_eq!(text, "System prompt here."), + GeminiPart::Text { text, .. } => assert_eq!(text, "System prompt here."), _ => panic!("Expected text"), } } @@ -1016,11 +1097,15 @@ mod tests { assert_eq!(json["maxOutputTokens"], 2048); } - // --- Issue #501: thought_signature round-trip tests --- + // --- Issue #501/#506: thought_signature round-trip tests --- + // + // Gemini 2.5+/3.x thinking models include `thoughtSignature` at the PART + // level (sibling of `functionCall`/`text`). All signatures must be echoed + // back exactly as received in follow-up requests. #[test] - fn test_thought_signature_captured_from_response() { - // Gemini 2.5+ thinking models return thoughtSignature on functionCall parts. + fn test_thought_signature_captured_from_function_call_response() { + // Gemini 3.x: thoughtSignature at the part level (sibling of functionCall). let json = serde_json::json!({ "candidates": [{ "content": { @@ -1028,9 +1113,9 @@ mod tests { "parts": [{ "functionCall": { "name": "web_search", - "args": {"query": "rust lang"}, - "thoughtSignature": "abc123signature" - } + "args": {"query": "rust lang"} + }, + "thoughtSignature": "abc123signature" }] }, "finishReason": "STOP" @@ -1061,9 +1146,45 @@ mod tests { } #[test] - fn test_thought_signature_echoed_in_request() { + fn test_thought_signature_captured_from_text_response() { + // Gemini 3.x: thoughtSignature on text parts too. + let json = serde_json::json!({ + "candidates": [{ + "content": { + "role": "model", + "parts": [{ + "text": "Let me think about this...", + "thoughtSignature": "text_sig_456" + }] + }, + "finishReason": "STOP" + }], + "usageMetadata": { + "promptTokenCount": 10, + "candidatesTokenCount": 8 + } + }); + + let resp: GeminiResponse = serde_json::from_value(json).unwrap(); + let completion = convert_response(resp).unwrap(); + + match &completion.content[0] { + ContentBlock::Text { + text, + provider_metadata, + } => { + assert_eq!(text, "Let me think about this..."); + let meta = provider_metadata.as_ref().expect("provider_metadata should be set for text with thoughtSignature"); + assert_eq!(meta["thought_signature"], "text_sig_456"); + } + _ => panic!("Expected Text content block"), + } + } + + #[test] + fn test_thought_signature_echoed_in_request_function_call() { // When a ToolUse block carries provider_metadata with thought_signature, - // convert_messages should echo it back in the GeminiFunctionCallData. + // convert_messages should echo it back at the part level. let messages = vec![ Message::user("Search for rust"), Message { @@ -1090,16 +1211,20 @@ mod tests { let (contents, _) = convert_messages(&messages, &None); - // The assistant's turn (index 1) should have a FunctionCall with the thought_signature + // The assistant's turn (index 1) should have a FunctionCall with the + // thought_signature at the part level. let assistant_turn = &contents[1]; assert_eq!(assistant_turn.role.as_deref(), Some("model")); let fc_part = &assistant_turn.parts[0]; match fc_part { - GeminiPart::FunctionCall { function_call } => { + GeminiPart::FunctionCall { + function_call, + thought_signature, + } => { assert_eq!(function_call.name, "web_search"); assert_eq!( - function_call.thought_signature.as_deref(), + thought_signature.as_deref(), Some("sig_xyz789") ); } @@ -1107,6 +1232,42 @@ mod tests { } } + #[test] + fn test_thought_signature_echoed_in_request_text() { + // When a Text block carries provider_metadata with thought_signature, + // convert_messages should echo it back on the text part. + let messages = vec![ + Message::user("Hello"), + Message { + role: Role::Assistant, + content: MessageContent::Blocks(vec![ContentBlock::Text { + text: "Let me think...".to_string(), + provider_metadata: Some(serde_json::json!({ + "thought_signature": "text_sig_abc" + })), + }]), + }, + ]; + + let (contents, _) = convert_messages(&messages, &None); + let assistant_turn = &contents[1]; + assert_eq!(assistant_turn.role.as_deref(), Some("model")); + + match &assistant_turn.parts[0] { + GeminiPart::Text { + text, + thought_signature, + } => { + assert_eq!(text, "Let me think..."); + assert_eq!( + thought_signature.as_deref(), + Some("text_sig_abc") + ); + } + _ => panic!("Expected Text part"), + } + } + #[test] fn test_thought_signature_none_when_absent() { // When there's no thought_signature, provider_metadata should be None @@ -1148,7 +1309,7 @@ mod tests { #[test] fn test_thought_signature_not_echoed_without_metadata() { // ToolUse blocks without provider_metadata should produce - // GeminiFunctionCallData with thought_signature: None + // GeminiPart::FunctionCall with thought_signature: None let messages = vec![ Message::user("Hello"), Message { @@ -1166,9 +1327,11 @@ mod tests { let assistant_turn = &contents[1]; match &assistant_turn.parts[0] { - GeminiPart::FunctionCall { function_call } => { + GeminiPart::FunctionCall { + thought_signature, .. + } => { assert!( - function_call.thought_signature.is_none(), + thought_signature.is_none(), "thought_signature should be None when no provider_metadata" ); } @@ -1178,26 +1341,32 @@ mod tests { #[test] fn test_thought_signature_serialization_round_trip() { - // Verify the GeminiFunctionCallData serializes thoughtSignature correctly - let data = GeminiFunctionCallData { - name: "web_search".to_string(), - args: serde_json::json!({"query": "test"}), + // Verify the part-level thoughtSignature serializes correctly + let part = GeminiPart::FunctionCall { + function_call: GeminiFunctionCallData { + name: "web_search".to_string(), + args: serde_json::json!({"query": "test"}), + }, thought_signature: Some("my_sig_abc".to_string()), }; - - let part = GeminiPart::FunctionCall { - function_call: data, - }; let json = serde_json::to_value(&part).unwrap(); - assert_eq!(json["functionCall"]["thoughtSignature"], "my_sig_abc"); + // thoughtSignature should be at the part level (sibling of functionCall), + // NOT nested inside functionCall. + assert_eq!(json["thoughtSignature"], "my_sig_abc"); assert_eq!(json["functionCall"]["name"], "web_search"); + assert!( + json["functionCall"].get("thoughtSignature").is_none(), + "thoughtSignature must NOT be nested inside functionCall" + ); // Verify it can round-trip through deserialization let deserialized: GeminiPart = serde_json::from_value(json).unwrap(); match deserialized { - GeminiPart::FunctionCall { function_call } => { + GeminiPart::FunctionCall { + thought_signature, .. + } => { assert_eq!( - function_call.thought_signature.as_deref(), + thought_signature.as_deref(), Some("my_sig_abc") ); } @@ -1208,25 +1377,63 @@ mod tests { #[test] fn test_thought_signature_omitted_when_none() { // When thought_signature is None, the JSON should not contain the field - let data = GeminiFunctionCallData { - name: "read_file".to_string(), - args: serde_json::json!({}), - thought_signature: None, - }; - let part = GeminiPart::FunctionCall { - function_call: data, + function_call: GeminiFunctionCallData { + name: "read_file".to_string(), + args: serde_json::json!({}), + }, + thought_signature: None, }; let json = serde_json::to_value(&part).unwrap(); assert!( - json["functionCall"].get("thoughtSignature").is_none(), + json.get("thoughtSignature").is_none(), "thoughtSignature should be omitted when None" ); } + #[test] + fn test_thought_signature_omitted_on_text_when_none() { + let part = GeminiPart::Text { + text: "Hello".to_string(), + thought_signature: None, + }; + let json = serde_json::to_value(&part).unwrap(); + assert!( + json.get("thoughtSignature").is_none(), + "thoughtSignature should be omitted on text parts when None" + ); + assert_eq!(json["text"], "Hello"); + } + + #[test] + fn test_text_thought_signature_round_trip() { + // Verify text part thought signature serializes at part level + let part = GeminiPart::Text { + text: "Thinking...".to_string(), + thought_signature: Some("text_sig_xyz".to_string()), + }; + let json = serde_json::to_value(&part).unwrap(); + assert_eq!(json["text"], "Thinking..."); + assert_eq!(json["thoughtSignature"], "text_sig_xyz"); + + // Round-trip + let deserialized: GeminiPart = serde_json::from_value(json).unwrap(); + match deserialized { + GeminiPart::Text { + text, + thought_signature, + } => { + assert_eq!(text, "Thinking..."); + assert_eq!(thought_signature.as_deref(), Some("text_sig_xyz")); + } + _ => panic!("Expected Text"), + } + } + #[test] fn test_multiple_function_calls_with_mixed_signatures() { - // Response with multiple function calls, some with signatures, some without + // Response with multiple function calls, some with signatures, some without. + // Gemini 3.x: thoughtSignature at part level. let json = serde_json::json!({ "candidates": [{ "content": { @@ -1235,9 +1442,9 @@ mod tests { { "functionCall": { "name": "web_search", - "args": {"query": "rust"}, - "thoughtSignature": "sig_1" - } + "args": {"query": "rust"} + }, + "thoughtSignature": "sig_1" }, { "functionCall": { @@ -1286,4 +1493,114 @@ mod tests { _ => panic!("Expected ToolUse"), } } + + #[test] + fn test_gemini_3x_text_and_function_call_both_have_signatures() { + // Gemini 3.x thinking models include thoughtSignature on ALL parts: + // text parts AND functionCall parts. Both must round-trip. + let json = serde_json::json!({ + "candidates": [{ + "content": { + "role": "model", + "parts": [ + { + "text": "I'll search for that.", + "thoughtSignature": "text_sig_aaa" + }, + { + "functionCall": { + "name": "web_search", + "args": {"query": "rust"} + }, + "thoughtSignature": "fc_sig_bbb" + } + ] + }, + "finishReason": "STOP" + }], + "usageMetadata": { + "promptTokenCount": 20, + "candidatesTokenCount": 15 + } + }); + + let resp: GeminiResponse = serde_json::from_value(json).unwrap(); + let completion = convert_response(resp).unwrap(); + + // Text part should have its signature + match &completion.content[0] { + ContentBlock::Text { + text, + provider_metadata, + } => { + assert_eq!(text, "I'll search for that."); + let meta = provider_metadata.as_ref().unwrap(); + assert_eq!(meta["thought_signature"], "text_sig_aaa"); + } + _ => panic!("Expected Text"), + } + + // Function call part should have its signature + match &completion.content[1] { + ContentBlock::ToolUse { + name, + provider_metadata, + .. + } => { + assert_eq!(name, "web_search"); + let meta = provider_metadata.as_ref().unwrap(); + assert_eq!(meta["thought_signature"], "fc_sig_bbb"); + } + _ => panic!("Expected ToolUse"), + } + + // Now convert back to Gemini format and verify signatures are echoed + let messages = vec![ + Message::user("Search for rust"), + Message { + role: Role::Assistant, + content: MessageContent::Blocks(completion.content), + }, + ]; + let (contents, _) = convert_messages(&messages, &None); + let model_turn = &contents[1]; + + // Verify text part echoes back its signature + match &model_turn.parts[0] { + GeminiPart::Text { + thought_signature, .. + } => { + assert_eq!( + thought_signature.as_deref(), + Some("text_sig_aaa"), + "Text part must echo back its thoughtSignature" + ); + } + _ => panic!("Expected Text part"), + } + + // Verify function call part echoes back its signature + match &model_turn.parts[1] { + GeminiPart::FunctionCall { + thought_signature, .. + } => { + assert_eq!( + thought_signature.as_deref(), + Some("fc_sig_bbb"), + "FunctionCall part must echo back its thoughtSignature" + ); + } + _ => panic!("Expected FunctionCall part"), + } + + // Verify the serialized JSON has signatures at the part level + let serialized = serde_json::to_value(&model_turn.parts[0]).unwrap(); + assert_eq!(serialized["thoughtSignature"], "text_sig_aaa"); + let serialized = serde_json::to_value(&model_turn.parts[1]).unwrap(); + assert_eq!(serialized["thoughtSignature"], "fc_sig_bbb"); + assert!( + serialized["functionCall"].get("thoughtSignature").is_none(), + "thoughtSignature must be at part level, NOT inside functionCall" + ); + } } diff --git a/crates/openfang-runtime/src/drivers/mod.rs b/crates/openfang-runtime/src/drivers/mod.rs index fdd62aaf..a6fd2f65 100644 --- a/crates/openfang-runtime/src/drivers/mod.rs +++ b/crates/openfang-runtime/src/drivers/mod.rs @@ -303,7 +303,10 @@ pub fn create_driver(config: &DriverConfig) -> Result, LlmErr // Claude Code CLI — subprocess-based, no API key needed if provider == "claude-code" { let cli_path = config.base_url.clone(); - return Ok(Arc::new(claude_code::ClaudeCodeDriver::new(cli_path))); + return Ok(Arc::new(claude_code::ClaudeCodeDriver::new( + cli_path, + config.skip_permissions, + ))); } // GitHub Copilot — wraps OpenAI-compatible driver with automatic token exchange. @@ -522,6 +525,7 @@ mod tests { provider: "my-custom-llm".to_string(), api_key: Some("test".to_string()), base_url: Some("http://localhost:9999/v1".to_string()), + skip_permissions: true, }; let driver = create_driver(&config); assert!(driver.is_ok()); @@ -533,6 +537,7 @@ mod tests { provider: "nonexistent".to_string(), api_key: None, base_url: None, + skip_permissions: true, }; let driver = create_driver(&config); assert!(driver.is_err()); @@ -633,6 +638,7 @@ mod tests { provider: "nvidia".to_string(), api_key: None, // not explicitly passed base_url: Some("https://integrate.api.nvidia.com/v1".to_string()), + skip_permissions: true, }; let driver = create_driver(&config); assert!(driver.is_ok(), "Custom provider with env var convention should succeed"); @@ -646,6 +652,7 @@ mod tests { provider: "nvidia".to_string(), api_key: None, base_url: None, + skip_permissions: true, }; let driver = create_driver(&config); assert!(driver.is_err()); @@ -660,6 +667,7 @@ mod tests { provider: "nvidia".to_string(), api_key: None, base_url: None, + skip_permissions: true, }; let result = create_driver(&config); assert!(result.is_err()); @@ -683,6 +691,7 @@ mod tests { provider: "my-custom-provider".to_string(), api_key: Some("explicit-key".to_string()), base_url: Some("https://api.example.com/v1".to_string()), + skip_permissions: true, }; let driver = create_driver(&config); assert!(driver.is_ok()); diff --git a/crates/openfang-runtime/src/drivers/openai.rs b/crates/openfang-runtime/src/drivers/openai.rs index 2bc74b6f..15a5a665 100644 --- a/crates/openfang-runtime/src/drivers/openai.rs +++ b/crates/openfang-runtime/src/drivers/openai.rs @@ -268,7 +268,7 @@ impl LlmDriver for OpenAIDriver { reasoning_content: None, }); } - ContentBlock::Text { text } => { + ContentBlock::Text { text, .. } => { parts.push(OaiContentPart::Text { text: text.clone() }); } ContentBlock::Image { media_type, data } => { @@ -297,7 +297,7 @@ impl LlmDriver for OpenAIDriver { let mut tool_calls = Vec::new(); for block in blocks { match block { - ContentBlock::Text { text } => text_parts.push(text.clone()), + ContentBlock::Text { text, .. } => text_parts.push(text.clone()), ContentBlock::ToolUse { id, name, input, .. } => { tool_calls.push(OaiToolCall { id: id.clone(), @@ -563,7 +563,7 @@ impl LlmDriver for OpenAIDriver { } } if !cleaned.is_empty() { - content.push(ContentBlock::Text { text: cleaned }); + content.push(ContentBlock::Text { text: cleaned, provider_metadata: None }); } } } @@ -581,7 +581,7 @@ impl LlmDriver for OpenAIDriver { }).unwrap_or(""); let summary = extract_thinking_summary(thinking_text); debug!(summary_len = summary.len(), "Synthesizing text from thinking-only response"); - content.push(ContentBlock::Text { text: summary }); + content.push(ContentBlock::Text { text: summary, provider_metadata: None }); } if let Some(calls) = choice.message.tool_calls { @@ -720,7 +720,7 @@ impl LlmDriver for OpenAIDriver { let mut tool_calls_out = Vec::new(); for block in blocks { match block { - ContentBlock::Text { text } => text_parts.push(text.clone()), + ContentBlock::Text { text, .. } => text_parts.push(text.clone()), ContentBlock::ToolUse { id, name, input, .. } => { tool_calls_out.push(OaiToolCall { id: id.clone(), @@ -1169,7 +1169,7 @@ impl LlmDriver for OpenAIDriver { } } if !cleaned.is_empty() { - content.push(ContentBlock::Text { text: cleaned }); + content.push(ContentBlock::Text { text: cleaned, provider_metadata: None }); } } @@ -1185,7 +1185,7 @@ impl LlmDriver for OpenAIDriver { }).unwrap_or(""); let summary = extract_thinking_summary(thinking_text); debug!(summary_len = summary.len(), "Synthesizing text from thinking-only stream response"); - content.push(ContentBlock::Text { text: summary }); + content.push(ContentBlock::Text { text: summary, provider_metadata: None }); } for (id, name, arguments) in &tool_accum { @@ -1410,6 +1410,7 @@ fn parse_groq_failed_tool_call(body: &str) -> Option { return Some(CompletionResponse { content: vec![ContentBlock::Text { text: failed.to_string(), + provider_metadata: None, }], tool_calls: vec![], stop_reason: StopReason::EndTurn, diff --git a/crates/openfang-runtime/src/kernel_handle.rs b/crates/openfang-runtime/src/kernel_handle.rs index 1bb2e76e..35da921d 100644 --- a/crates/openfang-runtime/src/kernel_handle.rs +++ b/crates/openfang-runtime/src/kernel_handle.rs @@ -183,19 +183,22 @@ pub trait KernelHandle: Send + Sync { } /// Send a message to a user on a named channel adapter (e.g., "email", "telegram"). + /// When `thread_id` is provided, the message is sent as a thread reply. /// Returns a confirmation string on success. async fn send_channel_message( &self, channel: &str, recipient: &str, message: &str, + thread_id: Option<&str>, ) -> Result { - let _ = (channel, recipient, message); + let _ = (channel, recipient, message, thread_id); Err("Channel send not available".to_string()) } /// Send media content (image/file) to a user on a named channel adapter. /// `media_type` is "image" or "file", `media_url` is the URL, `caption` is optional text. + /// When `thread_id` is provided, the media is sent as a thread reply. async fn send_channel_media( &self, channel: &str, @@ -204,11 +207,28 @@ pub trait KernelHandle: Send + Sync { media_url: &str, caption: Option<&str>, filename: Option<&str>, + thread_id: Option<&str>, ) -> Result { - let _ = (channel, recipient, media_type, media_url, caption, filename); + let _ = (channel, recipient, media_type, media_url, caption, filename, thread_id); Err("Channel media send not available".to_string()) } + /// Send a local file (raw bytes) to a user on a named channel adapter. + /// Used by the `channel_send` tool when `file_path` is provided. + /// When `thread_id` is provided, the file is sent as a thread reply. + async fn send_channel_file_data( + &self, + channel: &str, + recipient: &str, + data: Vec, + filename: &str, + mime_type: &str, + thread_id: Option<&str>, + ) -> Result { + let _ = (channel, recipient, data, filename, mime_type, thread_id); + Err("Channel file data send not available".to_string()) + } + /// Spawn an agent with capability inheritance enforcement. /// `parent_caps` are the parent's granted capabilities. The kernel MUST verify /// that every capability in the child manifest is covered by `parent_caps`. diff --git a/crates/openfang-runtime/src/llm_driver.rs b/crates/openfang-runtime/src/llm_driver.rs index f4999f86..9178be9d 100644 --- a/crates/openfang-runtime/src/llm_driver.rs +++ b/crates/openfang-runtime/src/llm_driver.rs @@ -86,7 +86,7 @@ impl CompletionResponse { self.content .iter() .filter_map(|block| match block { - ContentBlock::Text { text } => Some(text.as_str()), + ContentBlock::Text { text, .. } => Some(text.as_str()), ContentBlock::Thinking { .. } => None, _ => None, }) @@ -167,6 +167,19 @@ pub struct DriverConfig { pub api_key: Option, /// Base URL override. pub base_url: Option, + /// Skip interactive permission prompts (Claude Code provider only). + /// + /// When `true`, adds `--dangerously-skip-permissions` to the spawned + /// `claude` CLI. Defaults to `true` because OpenFang runs as a daemon + /// with no interactive terminal, so permission prompts would block + /// indefinitely. OpenFang's own capability / RBAC layer already + /// restricts what agents can do, making this safe. + #[serde(default = "default_skip_permissions")] + pub skip_permissions: bool, +} + +fn default_skip_permissions() -> bool { + true } /// SECURITY: Custom Debug impl redacts the API key. @@ -176,6 +189,7 @@ impl std::fmt::Debug for DriverConfig { .field("provider", &self.provider) .field("api_key", &self.api_key.as_ref().map(|_| "")) .field("base_url", &self.base_url) + .field("skip_permissions", &self.skip_permissions) .finish() } } @@ -190,9 +204,11 @@ mod tests { content: vec![ ContentBlock::Text { text: "Hello ".to_string(), + provider_metadata: None, }, ContentBlock::Text { text: "world!".to_string(), + provider_metadata: None, }, ], stop_reason: StopReason::EndTurn, @@ -255,6 +271,7 @@ mod tests { Ok(CompletionResponse { content: vec![ContentBlock::Text { text: "Hello!".to_string(), + provider_metadata: None, }], stop_reason: StopReason::EndTurn, tool_calls: vec![], diff --git a/crates/openfang-runtime/src/session_repair.rs b/crates/openfang-runtime/src/session_repair.rs index 15aaf43f..3854f076 100644 --- a/crates/openfang-runtime/src/session_repair.rs +++ b/crates/openfang-runtime/src/session_repair.rs @@ -492,7 +492,7 @@ fn is_empty_or_blank_content(content: &MessageContent) -> bool { MessageContent::Blocks(blocks) => { blocks.is_empty() || blocks.iter().all(|b| match b { - ContentBlock::Text { text } => text.trim().is_empty(), + ContentBlock::Text { text, .. } => text.trim().is_empty(), ContentBlock::Unknown => true, _ => false, }) @@ -630,7 +630,7 @@ pub fn prune_heartbeat_turns(messages: &mut Vec, keep_recent: usize) { } MessageContent::Blocks(blocks) => { blocks.len() == 1 - && matches!(&blocks[0], ContentBlock::Text { text } if { + && matches!(&blocks[0], ContentBlock::Text { text, .. } if { let t = text.trim(); t == "NO_REPLY" || t == "[no reply needed]" }) @@ -675,7 +675,7 @@ fn merge_content(dst: &mut MessageContent, src: MessageContent) { /// Convert MessageContent to a Vec. fn content_to_blocks(content: MessageContent) -> Vec { match content { - MessageContent::Text(s) => vec![ContentBlock::Text { text: s }], + MessageContent::Text(s) => vec![ContentBlock::Text { text: s, provider_metadata: None }], MessageContent::Blocks(blocks) => blocks, } } @@ -1017,6 +1017,7 @@ mod tests { role: Role::Assistant, content: MessageContent::Blocks(vec![ContentBlock::Text { text: String::new(), + provider_metadata: None, }]), }, Message::user("Never mind"), diff --git a/crates/openfang-runtime/src/tool_runner.rs b/crates/openfang-runtime/src/tool_runner.rs index bff189b3..20613278 100644 --- a/crates/openfang-runtime/src/tool_runner.rs +++ b/crates/openfang-runtime/src/tool_runner.rs @@ -330,7 +330,7 @@ pub async fn execute_tool( "cron_cancel" => tool_cron_cancel(input, kernel).await, // Channel send tool (proactive outbound messaging) - "channel_send" => tool_channel_send(input, kernel).await, + "channel_send" => tool_channel_send(input, kernel, workspace_root).await, // Persistent process tools "process_start" => tool_process_start(input, process_manager, caller_agent_id).await, @@ -1024,7 +1024,7 @@ pub fn builtin_tool_definitions() -> Vec { // --- Channel send tool (proactive outbound messaging) --- ToolDefinition { name: "channel_send".to_string(), - description: "Send a message or media to a user on a configured channel (email, telegram, slack, etc). For email: recipient is the email address; optionally set subject. For media: set image_url or file_url to send an image or file instead of (or alongside) text.".to_string(), + description: "Send a message or media to a user on a configured channel (email, telegram, slack, etc). For email: recipient is the email address; optionally set subject. For media: set image_url, file_url, or file_path to send an image or file instead of (or alongside) text. Use thread_id to reply in a specific thread/topic.".to_string(), input_schema: serde_json::json!({ "type": "object", "properties": { @@ -1034,7 +1034,9 @@ pub fn builtin_tool_definitions() -> Vec { "message": { "type": "string", "description": "The message body to send (required for text, optional caption for media)" }, "image_url": { "type": "string", "description": "URL of an image to send (supported on Telegram, Discord, Slack)" }, "file_url": { "type": "string", "description": "URL of a file to send as attachment" }, - "filename": { "type": "string", "description": "Filename for file attachments (defaults to 'file')" } + "file_path": { "type": "string", "description": "Local file path to send as attachment (reads from disk; use instead of file_url for local files)" }, + "filename": { "type": "string", "description": "Filename for file attachments (defaults to the basename of file_path, or 'file')" }, + "thread_id": { "type": "string", "description": "Thread/topic ID to reply in (e.g., Telegram message_thread_id, Slack thread_ts)" } }, "required": ["channel", "recipient"] }), @@ -2175,6 +2177,7 @@ async fn tool_cron_cancel( async fn tool_channel_send( input: &serde_json::Value, kernel: Option<&Arc>, + workspace_root: Option<&Path>, ) -> Result { let kh = require_kernel(kernel)?; @@ -2192,14 +2195,17 @@ async fn tool_channel_send( return Err("Recipient cannot be empty".to_string()); } - // Check for media content (image_url or file_url) + let thread_id = input["thread_id"].as_str().filter(|s| !s.is_empty()); + + // Check for media content (image_url, file_url, or file_path) let image_url = input["image_url"].as_str().filter(|s| !s.is_empty()); let file_url = input["file_url"].as_str().filter(|s| !s.is_empty()); + let file_path = input["file_path"].as_str().filter(|s| !s.is_empty()); if let Some(url) = image_url { let caption = input["message"].as_str().filter(|s| !s.is_empty()); return kh - .send_channel_media(&channel, recipient, "image", url, caption, None) + .send_channel_media(&channel, recipient, "image", url, caption, None, thread_id) .await; } @@ -2207,7 +2213,64 @@ async fn tool_channel_send( let caption = input["message"].as_str().filter(|s| !s.is_empty()); let filename = input["filename"].as_str(); return kh - .send_channel_media(&channel, recipient, "file", url, caption, filename) + .send_channel_media(&channel, recipient, "file", url, caption, filename, thread_id) + .await; + } + + // Local file attachment: read from disk and send as FileData + if let Some(raw_path) = file_path { + let resolved = resolve_file_path(raw_path, workspace_root)?; + let data = tokio::fs::read(&resolved) + .await + .map_err(|e| format!("Failed to read file '{}': {e}", resolved.display()))?; + + // Derive filename from the path if not explicitly provided + let filename = input["filename"] + .as_str() + .filter(|s| !s.is_empty()) + .map(|s| s.to_string()) + .unwrap_or_else(|| { + resolved + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("file") + .to_string() + }); + + // Determine MIME type from extension + let ext = resolved + .extension() + .and_then(|e| e.to_str()) + .unwrap_or("") + .to_lowercase(); + let mime_type = match ext.as_str() { + "png" => "image/png", + "jpg" | "jpeg" => "image/jpeg", + "gif" => "image/gif", + "webp" => "image/webp", + "svg" => "image/svg+xml", + "pdf" => "application/pdf", + "txt" => "text/plain", + "csv" => "text/csv", + "json" => "application/json", + "xml" => "application/xml", + "zip" => "application/zip", + "gz" | "gzip" => "application/gzip", + "tar" => "application/x-tar", + "mp3" => "audio/mpeg", + "wav" => "audio/wav", + "mp4" => "video/mp4", + "doc" => "application/msword", + "docx" => "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + "xls" => "application/vnd.ms-excel", + "xlsx" => "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + _ => "application/octet-stream", + }; + + return kh + .send_channel_file_data( + &channel, recipient, data, &filename, mime_type, thread_id, + ) .await; } @@ -2238,7 +2301,7 @@ async fn tool_channel_send( message.to_string() }; - kh.send_channel_message(&channel, recipient, &final_message) + kh.send_channel_message(&channel, recipient, &final_message, thread_id) .await } diff --git a/crates/openfang-types/src/message.rs b/crates/openfang-types/src/message.rs index c01b9a0a..99be5957 100644 --- a/crates/openfang-types/src/message.rs +++ b/crates/openfang-types/src/message.rs @@ -42,6 +42,11 @@ pub enum ContentBlock { Text { /// The text content. text: String, + /// Provider-specific metadata (e.g. Gemini `thoughtSignature`). + /// Opaque to the core — drivers read/write this to round-trip + /// fields the provider requires on subsequent requests. + #[serde(default, skip_serializing_if = "Option::is_none")] + provider_metadata: Option, }, /// An inline base64-encoded image. #[serde(rename = "image")] @@ -134,7 +139,7 @@ impl MessageContent { MessageContent::Blocks(blocks) => blocks .iter() .map(|b| match b { - ContentBlock::Text { text } => text.len(), + ContentBlock::Text { text, .. } => text.len(), ContentBlock::ToolResult { content, .. } => content.len(), ContentBlock::Thinking { thinking } => thinking.len(), ContentBlock::ToolUse { .. } @@ -152,7 +157,7 @@ impl MessageContent { MessageContent::Blocks(blocks) => blocks .iter() .filter_map(|b| match b { - ContentBlock::Text { text } => Some(text.as_str()), + ContentBlock::Text { text, .. } => Some(text.as_str()), _ => None, }) .collect::>() @@ -310,6 +315,7 @@ mod tests { let blocks = vec![ ContentBlock::Text { text: "What is in this image?".to_string(), + provider_metadata: None, }, ContentBlock::Image { media_type: "image/jpeg".to_string(), @@ -321,7 +327,7 @@ mod tests { match msg.content { MessageContent::Blocks(ref b) => { assert_eq!(b.len(), 2); - assert!(matches!(&b[0], ContentBlock::Text { text } if text == "What is in this image?")); + assert!(matches!(&b[0], ContentBlock::Text { text, .. } if text == "What is in this image?")); assert!(matches!(&b[1], ContentBlock::Image { media_type, .. } if media_type == "image/jpeg")); } _ => panic!("Expected blocks content"),