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

fix 11 issues

Browse Source
This commit is contained in:
jaberjaber23
2026-03-12 01:22:45 +03:00
parent 98f8d1ca79
commit 951e8d0feb
27 changed files with 1733 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/openfang-api/src/channel_bridge.rs
+2 -2
View File
@@ -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::<Vec<_>>()
@@ -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
crates/openfang-api/src/openai_compat.rs
+1 -1
View File
@@ -203,7 +203,7 @@ fn convert_messages(oai_messages: &[OaiMessage]) -> Vec<Message> {
.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}
crates/openfang-api/src/routes.rs
+32 -11
View File
@@ -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<Arc<AppState>>) -> 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<Arc<AppState>>) -> 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::<Vec<_>>(),
"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}")})),
crates/openfang-api/src/ws.rs
+1 -1
View File
@@ -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;
crates/openfang-api/static/js/pages/chat.js
+21 -6
View File
@@ -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 = '<details><summary>Reasoning...</summary>\n\n' + phaseMsg._reasoning + '</details>';
} 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();
crates/openfang-channels/src/bridge.rs
+106 -17
View File
@@ -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::<Vec<_>>()
@@ -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<String> {
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<ContentBlock>` containing an image block (base64-encoded) and
@@ -810,28 +847,20 @@ async fn download_image_to_blocks(url: &str, caption: Option<&str>) -> Vec<Conte
warn!("Failed to download image from channel: {e}");
return vec![ContentBlock::Text {
text: format!("[Image download failed: {e}]"),
provider_metadata: None,
}];
}
};
// Detect media type from Content-Type header, fall back to URL extension
let content_type = resp
// Detect media type from Content-Type header — but only trust it if it's
// actually an image/* type. Many APIs (Telegram, S3 pre-signed URLs) return
// `application/octet-stream` for all files, which breaks vision.
let header_type = resp
.headers()
.get("content-type")
.and_then(|v| v.to_str().ok())
.map(|ct| ct.split(';').next().unwrap_or(ct).trim().to_string());
let media_type = content_type.unwrap_or_else(|| {
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 {
"image/jpeg".to_string()
}
});
.map(|ct| ct.split(';').next().unwrap_or(ct).trim().to_string())
.filter(|ct| ct.starts_with("image/"));
let bytes = match resp.bytes().await {
Ok(b) => b,
@@ -839,10 +868,19 @@ async fn download_image_to_blocks(url: &str, caption: Option<&str>) -> Vec<Conte
warn!("Failed to read image bytes: {e}");
return vec![ContentBlock::Text {
text: format!("[Image read failed: {e}]"),
provider_metadata: None,
}];
}
};
// Three-tier media type detection:
// 1. Trusted Content-Type header (only if image/*)
// 2. Magic byte sniffing (most reliable for binary data)
// 3. URL extension fallback
let media_type = header_type.unwrap_or_else(|| {
detect_image_magic(&bytes).unwrap_or_else(|| media_type_from_url(url))
});
if bytes.len() > 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<Conte
Some(c) => 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<Conte
if !cap.is_empty() {
blocks.push(ContentBlock::Text {
text: cap.to_string(),
provider_metadata: None,
});
}
}
@@ -1433,6 +1472,7 @@ mod tests {
let blocks = vec![
ContentBlock::Text {
text: "What is in this photo?".to_string(),
provider_metadata: None,
},
ContentBlock::Image {
media_type: "image/jpeg".to_string(),
@@ -1468,4 +1508,53 @@ mod tests {
.unwrap();
assert_eq!(result, "Echo: ");
}
#[test]
fn test_detect_image_magic_jpeg() {
let bytes = [0xFF, 0xD8, 0xFF, 0xE0, 0x00, 0x10];
assert_eq!(detect_image_magic(&bytes), Some("image/jpeg".to_string()));
}
#[test]
fn test_detect_image_magic_png() {
let bytes = [0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A];
assert_eq!(detect_image_magic(&bytes), Some("image/png".to_string()));
}
#[test]
fn test_detect_image_magic_gif() {
let bytes = [0x47, 0x49, 0x46, 0x38, 0x39, 0x61];
assert_eq!(detect_image_magic(&bytes), Some("image/gif".to_string()));
}
#[test]
fn test_detect_image_magic_webp() {
let bytes = [
0x52, 0x49, 0x46, 0x46, // RIFF
0x00, 0x00, 0x00, 0x00, // size (don't care)
0x57, 0x45, 0x42, 0x50, // WEBP
];
assert_eq!(detect_image_magic(&bytes), Some("image/webp".to_string()));
}
#[test]
fn test_detect_image_magic_unknown() {
let bytes = [0x00, 0x01, 0x02, 0x03];
assert_eq!(detect_image_magic(&bytes), None);
}
#[test]
fn test_detect_image_magic_empty() {
assert_eq!(detect_image_magic(&[]), None);
}
#[test]
fn test_media_type_from_url() {
assert_eq!(media_type_from_url("https://example.com/photo.png"), "image/png");
assert_eq!(media_type_from_url("https://example.com/anim.gif"), "image/gif");
assert_eq!(media_type_from_url("https://example.com/img.webp"), "image/webp");
assert_eq!(media_type_from_url("https://example.com/photo.jpg"), "image/jpeg");
// No extension — defaults to JPEG
assert_eq!(media_type_from_url("https://api.telegram.org/file/bot123/photos/file_42"), "image/jpeg");
}
}
crates/openfang-channels/src/telegram.rs
+375 -29
View File
@@ -36,6 +36,9 @@ pub struct TelegramAdapter {
poll_interval: Duration,
/// Base URL for Telegram Bot API (supports proxies/mirrors).
api_base_url: String,
/// Bot username (without @), populated from `getMe` during `start()`.
/// Used for @mention detection in group messages.
bot_username: Arc<tokio::sync::RwLock<Option<String>>>,
shutdown_tx: Arc<watch::Sender<bool>>,
shutdown_rx: watch::Receiver<bool>,
}
@@ -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<u8>,
filename: &str,
mime_type: &str,
thread_id: Option<i64>,
) -> Result<(), Box<dyn std::error::Error>> {
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<Pin<Box<dyn Stream<Item = ChannelMessage> + Send>>, Box<dyn std::error::Error>>
{
// 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<ChannelMessage> {
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<String> = 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<String> = 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
crates/openfang-channels/src/types.rs
+7
View File
@@ -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<u8>,
filename: String,
mime_type: String,
},
Voice {
url: String,
duration_seconds: u32,
crates/openfang-hands/bundled/browser/HAND.toml
+2 -1
View File
@@ -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"
crates/openfang-hands/src/lib.rs
+7
View File
@@ -117,6 +117,13 @@ pub struct HandRequirement {
/// Human-readable description of why this is needed.
#[serde(default)]
pub description: Option<String>,
/// 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<HandInstallInfo>,
crates/openfang-hands/src/registry.rs
+125
View File
@@ -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<HandReadiness> {
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);
}
}
crates/openfang-kernel/src/kernel.rs
+247 -83
View File
@@ -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<openfang_runtime::compactor::ContextReport> {
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<ToolDefinition> {
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<String> = 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<ToolDefinition> = 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<ToolDefinition> = if mcp_allowlist.is_empty() {
mcp_tools.iter().cloned().collect()
} else {
// Normalize allowlist names for matching
let normalized: Vec<String> = 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<String, String> {
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<String, String> {
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<u8>,
filename: &str,
mime_type: &str,
thread_id: Option<&str>,
) -> Result<String, String> {
let adapter = self
.channel_adapters
.get(channel)
.ok_or_else(|| {
let available: Vec<String> = 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,
crates/openfang-kernel/src/triggers.rs
+221
View File
@@ -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<Trigger> {
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<Trigger>) -> 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");
}
}
crates/openfang-memory/src/session.rs
+1 -1
View File
@@ -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, .. } => {
crates/openfang-runtime/src/agent_loop.rs
+16 -2
View File
@@ -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. <function=web_search>{"query":"rust async"}</function>"#.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![],
crates/openfang-runtime/src/compactor.rs
+8 -2
View File
@@ -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(),
crates/openfang-runtime/src/drivers/anthropic.rs
+3 -3
View File
@@ -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<ApiContentBlock> = 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 {
crates/openfang-runtime/src/drivers/claude_code.rs
+36 -9
View File
@@ -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<String>) -> 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<String>, 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
crates/openfang-runtime/src/drivers/fallback.rs
+1
View File
@@ -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![],
crates/openfang-runtime/src/drivers/gemini.rs
+380 -63
View File
@@ -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<String>,
},
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<String>,
},
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<String>,
}
#[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<String>) -> Option<Gemin
Some(GeminiContent {
role: None, // systemInstruction doesn't use a role
parts: vec![GeminiPart::Text { text }],
parts: vec![GeminiPart::Text {
text,
thought_signature: None,
}],
})
}
@@ -349,19 +390,36 @@ fn convert_response(resp: GeminiResponse) -> Result<CompletionResponse, LlmError
Some(gemini_content) => {
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<String> = None;
// Track function calls: (name, args_json, thought_signature)
let mut fn_calls: Vec<(String, serde_json::Value, Option<String>)> = Vec::new();
let mut finish_reason: Option<String> = 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"
);
}
}
crates/openfang-runtime/src/drivers/mod.rs
+10 -1
View File
@@ -303,7 +303,10 @@ pub fn create_driver(config: &DriverConfig) -> Result<Arc<dyn LlmDriver>, 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());
crates/openfang-runtime/src/drivers/openai.rs
+8 -7
View File
@@ -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<CompletionResponse> {
return Some(CompletionResponse {
content: vec![ContentBlock::Text {
text: failed.to_string(),
provider_metadata: None,
}],
tool_calls: vec![],
stop_reason: StopReason::EndTurn,
crates/openfang-runtime/src/kernel_handle.rs
+22 -2
View File
@@ -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<String, String> {
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<String, String> {
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<u8>,
filename: &str,
mime_type: &str,
thread_id: Option<&str>,
) -> Result<String, String> {
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`.
crates/openfang-runtime/src/llm_driver.rs
+18 -1
View File
@@ -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<String>,
/// Base URL override.
pub base_url: Option<String>,
/// 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(|_| "<redacted>"))
.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![],
crates/openfang-runtime/src/session_repair.rs
+4 -3
View File
@@ -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<Message>, 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<ContentBlock>.
fn content_to_blocks(content: MessageContent) -> Vec<ContentBlock> {
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"),
crates/openfang-runtime/src/tool_runner.rs
+70 -7
View File
@@ -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<ToolDefinition> {
// --- 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<ToolDefinition> {
"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<dyn KernelHandle>>,
workspace_root: Option<&Path>,
) -> Result<String, String> {
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
}
crates/openfang-types/src/message.rs
+9 -3
View File
@@ -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<serde_json::Value>,
},
/// 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::<Vec<_>>()
@@ -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"),