diff --git a/Cargo.lock b/Cargo.lock index ddd59b65..7f1e049f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3792,7 +3792,7 @@ dependencies = [ [[package]] name = "openfang-api" -version = "0.3.41" +version = "0.3.42" dependencies = [ "async-trait", "axum", @@ -3829,7 +3829,7 @@ dependencies = [ [[package]] name = "openfang-channels" -version = "0.3.41" +version = "0.3.42" dependencies = [ "async-trait", "axum", @@ -3861,7 +3861,7 @@ dependencies = [ [[package]] name = "openfang-cli" -version = "0.3.41" +version = "0.3.42" dependencies = [ "clap", "clap_complete", @@ -3888,7 +3888,7 @@ dependencies = [ [[package]] name = "openfang-desktop" -version = "0.3.41" +version = "0.3.42" dependencies = [ "axum", "open", @@ -3914,7 +3914,7 @@ dependencies = [ [[package]] name = "openfang-extensions" -version = "0.3.41" +version = "0.3.42" dependencies = [ "aes-gcm", "argon2", @@ -3942,7 +3942,7 @@ dependencies = [ [[package]] name = "openfang-hands" -version = "0.3.41" +version = "0.3.42" dependencies = [ "chrono", "dashmap", @@ -3959,7 +3959,7 @@ dependencies = [ [[package]] name = "openfang-kernel" -version = "0.3.41" +version = "0.3.42" dependencies = [ "async-trait", "chrono", @@ -3996,7 +3996,7 @@ dependencies = [ [[package]] name = "openfang-memory" -version = "0.3.41" +version = "0.3.42" dependencies = [ "async-trait", "chrono", @@ -4015,7 +4015,7 @@ dependencies = [ [[package]] name = "openfang-migrate" -version = "0.3.41" +version = "0.3.42" dependencies = [ "chrono", "dirs 6.0.0", @@ -4034,7 +4034,7 @@ dependencies = [ [[package]] name = "openfang-runtime" -version = "0.3.41" +version = "0.3.42" dependencies = [ "anyhow", "async-trait", @@ -4068,7 +4068,7 @@ dependencies = [ [[package]] name = "openfang-skills" -version = "0.3.41" +version = "0.3.42" dependencies = [ "chrono", "hex", @@ -4091,7 +4091,7 @@ dependencies = [ [[package]] name = "openfang-types" -version = "0.3.41" +version = "0.3.42" dependencies = [ "async-trait", "chrono", @@ -4110,7 +4110,7 @@ dependencies = [ [[package]] name = "openfang-wire" -version = "0.3.41" +version = "0.3.42" dependencies = [ "async-trait", "chrono", @@ -7009,6 +7009,7 @@ dependencies = [ "futures-util", "http", "http-body", + "http-body-util", "iri-string", "pin-project-lite", "tokio", @@ -8772,7 +8773,7 @@ checksum = "b9cc00251562a284751c9973bace760d86c0276c471b4be569fe6b068ee97a56" [[package]] name = "xtask" -version = "0.3.41" +version = "0.3.42" [[package]] name = "yoke" diff --git a/Cargo.toml b/Cargo.toml index f4e2cdc9..d1b6bb89 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,7 +18,7 @@ members = [ ] [workspace.package] -version = "0.3.42" +version = "0.3.43" edition = "2021" license = "Apache-2.0 OR MIT" repository = "https://github.com/RightNow-AI/openfang" @@ -62,7 +62,7 @@ clap = { version = "4", features = ["derive"] } clap_complete = "4" # HTTP client (for LLM drivers) -reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "multipart", "rustls-tls"] } +reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "multipart", "rustls-tls", "gzip", "deflate", "brotli"] } # Async trait async-trait = "0.1" diff --git a/crates/openfang-api/src/routes.rs b/crates/openfang-api/src/routes.rs index 5a904828..b1a0a266 100644 --- a/crates/openfang-api/src/routes.rs +++ b/crates/openfang-api/src/routes.rs @@ -4750,6 +4750,9 @@ pub async fn update_budget( if let Some(v) = body["alert_threshold"].as_f64() { (*config_ptr).budget.alert_threshold = v.clamp(0.0, 1.0); } + if let Some(v) = body["default_max_llm_tokens_per_hour"].as_u64() { + (*config_ptr).budget.default_max_llm_tokens_per_hour = v; + } } let status = state @@ -4790,6 +4793,10 @@ pub async fn agent_budget_status( let daily = usage_store.query_daily(agent_id).unwrap_or(0.0); let monthly = usage_store.query_monthly(agent_id).unwrap_or(0.0); + // Token usage from scheduler + let token_usage = state.kernel.scheduler.get_usage(agent_id); + let tokens_used = token_usage.map(|(t, _)| t).unwrap_or(0); + ( StatusCode::OK, Json(serde_json::json!({ @@ -4810,6 +4817,11 @@ pub async fn agent_budget_status( "limit": quota.max_cost_per_month_usd, "pct": if quota.max_cost_per_month_usd > 0.0 { monthly / quota.max_cost_per_month_usd } else { 0.0 }, }, + "tokens": { + "used": tokens_used, + "limit": quota.max_llm_tokens_per_hour, + "pct": if quota.max_llm_tokens_per_hour > 0 { tokens_used as f64 / quota.max_llm_tokens_per_hour as f64 } else { 0.0 }, + }, })), ) } @@ -4832,6 +4844,7 @@ pub async fn agent_budget_ranking(State(state): State>) -> impl In "hourly_limit": entry.manifest.resources.max_cost_per_hour_usd, "daily_limit": entry.manifest.resources.max_cost_per_day_usd, "monthly_limit": entry.manifest.resources.max_cost_per_month_usd, + "max_llm_tokens_per_hour": entry.manifest.resources.max_llm_tokens_per_hour, })) } else { None @@ -4861,18 +4874,19 @@ pub async fn update_agent_budget( let hourly = body["max_cost_per_hour_usd"].as_f64(); let daily = body["max_cost_per_day_usd"].as_f64(); let monthly = body["max_cost_per_month_usd"].as_f64(); + let tokens = body["max_llm_tokens_per_hour"].as_u64(); - if hourly.is_none() && daily.is_none() && monthly.is_none() { + if hourly.is_none() && daily.is_none() && monthly.is_none() && tokens.is_none() { return ( StatusCode::BAD_REQUEST, - Json(serde_json::json!({"error": "Provide at least one of: max_cost_per_hour_usd, max_cost_per_day_usd, max_cost_per_month_usd"})), + Json(serde_json::json!({"error": "Provide at least one of: max_cost_per_hour_usd, max_cost_per_day_usd, max_cost_per_month_usd, max_llm_tokens_per_hour"})), ); } match state .kernel .registry - .update_resources(agent_id, hourly, daily, monthly) + .update_resources(agent_id, hourly, daily, monthly, tokens) { Ok(()) => { // Persist updated entry @@ -6867,15 +6881,13 @@ pub async fn delete_provider_key( .model_catalog .read() .unwrap_or_else(|e| e.into_inner()); - match catalog.get_provider(&name) { - Some(p) => p.api_key_env.clone(), - None => { - return ( - StatusCode::NOT_FOUND, - Json(serde_json::json!({"error": format!("Unknown provider '{}'", name)})), - ); - } - } + catalog + .get_provider(&name) + .map(|p| p.api_key_env.clone()) + .unwrap_or_else(|| { + // Custom/unknown provider — derive env var from convention + format!("{}_API_KEY", name.to_uppercase().replace('-', "_")) + }) }; if env_var.is_empty() { diff --git a/crates/openfang-api/static/index_body.html b/crates/openfang-api/static/index_body.html index 6b406fd3..a57e8e87 100644 --- a/crates/openfang-api/static/index_body.html +++ b/crates/openfang-api/static/index_body.html @@ -3458,7 +3458,7 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"]
this.editDaily = this.budgetData.daily_limit || 0; this.editMonthly = this.budgetData.monthly_limit || 0; this.editAlert = ((this.budgetData.alert_threshold || 0.8) * 100).toFixed(0); + this.editTokenLimit = this.budgetData.default_max_llm_tokens_per_hour || 0; this.editMode = true; }, async saveBudget() { @@ -3488,12 +3489,14 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"] if (+this.editMonthly !== this.budgetData.monthly_limit) body.max_monthly_usd = +this.editMonthly; let alertVal = (+this.editAlert) / 100; if (Math.abs(alertVal - this.budgetData.alert_threshold) > 0.001) body.alert_threshold = alertVal; + if (+this.editTokenLimit !== (this.budgetData.default_max_llm_tokens_per_hour || 0)) body.default_max_llm_tokens_per_hour = +this.editTokenLimit; await OpenFangAPI.put('/api/budget', body); this.editMode = false; await this.loadBudget(); } catch(e) { OpenFangToast.error('Failed to save: ' + (e.message || e)); } this.saving = false; }, + fmtTokens(v) { return v > 0 ? (v >= 1000000 ? (v/1000000).toFixed(1)+'M' : v >= 1000 ? (v/1000).toFixed(0)+'K' : v) : 'per-agent'; }, pctColor(pct) { return pct >= 0.8 ? '#ef4444' : pct >= 0.5 ? '#eab308' : '#22c55e'; }, fmtUsd(v) { return v > 0 ? '$' + v.toFixed(4) : 'unlimited'; } }" x-init="loadBudget()"> @@ -3533,9 +3536,12 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"]
-
+
Alert threshold: of any limit
+
+ Hourly token limit (per agent): +
@@ -3557,7 +3563,11 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"]
-
Set to 0 for unlimited. Changes apply immediately (in-memory, not persisted to config.toml).
+
+ + +
+
Set to 0 for unlimited/per-agent default. Changes apply immediately (in-memory, not persisted to config.toml).
@@ -3565,7 +3575,7 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"]

Top Spenders (Today)

- + diff --git a/crates/openfang-channels/src/bridge.rs b/crates/openfang-channels/src/bridge.rs index 3527dbd0..330d9883 100644 --- a/crates/openfang-channels/src/bridge.rs +++ b/crates/openfang-channels/src/bridge.rs @@ -288,6 +288,15 @@ impl BridgeManager { } /// Start an adapter: subscribe to its message stream and spawn a dispatch task. + /// + /// Each incoming message is dispatched as a concurrent task so that slow LLM + /// calls (10-30s) don't block subsequent messages. This prevents voice/media + /// messages sent in quick succession from appearing "lost" — all messages + /// begin processing immediately. Per-agent serialization (to prevent session + /// corruption) is handled by the kernel's `agent_msg_locks`. + /// + /// A semaphore limits concurrent dispatch tasks to prevent unbounded memory + /// growth under burst traffic. pub async fn start_adapter( &mut self, adapter: Arc, @@ -299,6 +308,10 @@ impl BridgeManager { let adapter_clone = adapter.clone(); let mut shutdown = self.shutdown_rx.clone(); + // Limit concurrent dispatch tasks to prevent unbounded growth. + // 32 is generous — most setups have 1-5 concurrent users. + let semaphore = Arc::new(tokio::sync::Semaphore::new(32)); + let task = tokio::spawn(async move { let mut stream = std::pin::pin!(stream); loop { @@ -306,13 +319,28 @@ impl BridgeManager { msg = stream.next() => { match msg { Some(message) => { - dispatch_message( - &message, - &handle, - &router, - adapter_clone.as_ref(), - &rate_limiter, - ).await; + // Spawn each dispatch as a concurrent task so the stream + // loop is never blocked by slow LLM calls. The kernel's + // per-agent lock ensures session integrity. + let handle = handle.clone(); + let router = router.clone(); + let adapter = adapter_clone.clone(); + let rate_limiter = rate_limiter.clone(); + let sem = semaphore.clone(); + tokio::spawn(async move { + // Acquire semaphore permit (blocks if 32 tasks are in flight). + let _permit = match sem.acquire().await { + Ok(p) => p, + Err(_) => return, // semaphore closed — shutting down + }; + dispatch_message( + &message, + &handle, + &router, + adapter.as_ref(), + &rate_limiter, + ).await; + }); } None => { info!("Channel adapter {} stream ended", adapter_clone.name()); diff --git a/crates/openfang-channels/src/nostr.rs b/crates/openfang-channels/src/nostr.rs index 6d9adb9e..18a43275 100644 --- a/crates/openfang-channels/src/nostr.rs +++ b/crates/openfang-channels/src/nostr.rs @@ -339,7 +339,7 @@ impl ChannelAdapter for NostrAdapter { platform_id: sender_pubkey.clone(), display_name: format!( "{}...", - &sender_pubkey[..8.min(sender_pubkey.len())] + openfang_types::truncate_str(&sender_pubkey, 8) ), openfang_user: None, }, diff --git a/crates/openfang-kernel/src/kernel.rs b/crates/openfang-kernel/src/kernel.rs index 1dde28da..0e5b8356 100644 --- a/crates/openfang-kernel/src/kernel.rs +++ b/crates/openfang-kernel/src/kernel.rs @@ -151,6 +151,10 @@ pub struct OpenFangKernel { pub channel_adapters: dashmap::DashMap>, /// Hot-reloadable default model override (set via config hot-reload, read at agent spawn). pub default_model_override: std::sync::RwLock>, + /// Per-agent message locks — serializes LLM calls for the same agent to prevent + /// session corruption when multiple messages arrive concurrently (e.g. rapid voice + /// messages via Telegram). Different agents can still run in parallel. + agent_msg_locks: dashmap::DashMap>>, /// Weak self-reference for trigger dispatch (set after Arc wrapping). self_handle: OnceLock>, } @@ -984,6 +988,7 @@ impl OpenFangKernel { whatsapp_gateway_pid: Arc::new(std::sync::Mutex::new(None)), channel_adapters: dashmap::DashMap::new(), default_model_override: std::sync::RwLock::new(None), + agent_msg_locks: dashmap::DashMap::new(), self_handle: OnceLock::new(), }; @@ -1400,6 +1405,10 @@ impl OpenFangKernel { /// When `content_blocks` is `Some`, the LLM agent loop receives structured /// multimodal content (text + images) instead of just a text string. This /// enables vision models to process images sent from channels like Telegram. + /// + /// Per-agent locking ensures that concurrent messages for the same agent + /// are serialized (preventing session corruption), while messages for + /// different agents run in parallel. pub async fn send_message_with_handle_and_blocks( &self, agent_id: AgentId, @@ -1407,6 +1416,17 @@ impl OpenFangKernel { kernel_handle: Option>, content_blocks: Option>, ) -> KernelResult { + // Acquire per-agent lock to serialize concurrent messages for the same agent. + // This prevents session corruption when multiple messages arrive in quick + // succession (e.g. rapid voice messages via Telegram). Messages for different + // agents are not blocked — each agent has its own independent lock. + let lock = self + .agent_msg_locks + .entry(agent_id) + .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))) + .clone(); + let _guard = lock.lock().await; + // Enforce quota before running the agent loop self.scheduler .check_quota(agent_id) @@ -4005,6 +4025,28 @@ impl OpenFangKernel { /// without requiring a daemon restart. Uses the hot-reloaded default model /// override when available. /// If fallback models are configured, wraps the primary in a `FallbackDriver`. + /// Look up a provider's base URL, checking runtime catalog first, then boot-time config. + /// + /// Custom providers added at runtime via the dashboard (`set_provider_url`) are + /// stored in the model catalog but NOT in `self.config.provider_urls` (which is + /// the boot-time snapshot). This helper checks both sources so that custom + /// providers work immediately without a daemon restart. + fn lookup_provider_url(&self, provider: &str) -> Option { + // 1. Boot-time config (from config.toml [provider_urls]) + if let Some(url) = self.config.provider_urls.get(provider) { + return Some(url.clone()); + } + // 2. Model catalog (updated at runtime by set_provider_url / apply_url_overrides) + if let Ok(catalog) = self.model_catalog.read() { + if let Some(p) = catalog.get_provider(provider) { + if !p.base_url.is_empty() { + return Some(p.base_url.clone()); + } + } + } + None + } + fn resolve_driver(&self, manifest: &AgentManifest) -> KernelResult> { let agent_provider = &manifest.model.provider; @@ -4053,17 +4095,20 @@ impl OpenFangKernel { std::env::var(&env_var).ok() }; - // Don't inherit default provider's base_url when switching providers + // Don't inherit default provider's base_url when switching providers. + // Uses lookup_provider_url() which checks both boot-time config AND the + // runtime model catalog, so custom providers added via the dashboard + // (which only update the catalog, not self.config) are found (#494). let base_url = if has_custom_url { manifest.model.base_url.clone() } else if agent_provider == default_provider { effective_default .base_url .clone() - .or_else(|| self.config.provider_urls.get(agent_provider.as_str()).cloned()) + .or_else(|| self.lookup_provider_url(agent_provider)) } else { - // Check provider_urls before falling back to hardcoded defaults - self.config.provider_urls.get(agent_provider.as_str()).cloned() + // Check provider_urls + catalog before falling back to hardcoded defaults + self.lookup_provider_url(agent_provider) }; let driver_config = DriverConfig { @@ -4114,7 +4159,7 @@ impl OpenFangKernel { base_url: fb .base_url .clone() - .or_else(|| self.config.provider_urls.get(&fb.provider).cloned()), + .or_else(|| self.lookup_provider_url(&fb.provider)), }; match drivers::create_driver(&config) { Ok(d) => chain.push((d, fb.model.clone())), @@ -4858,6 +4903,12 @@ fn apply_budget_defaults( if budget.max_monthly_usd > 0.0 && resources.max_cost_per_month_usd == 0.0 { resources.max_cost_per_month_usd = budget.max_monthly_usd; } + // Override per-agent hourly token limit when the global default is set. + // This lets users raise (or lower) the token budget for all agents at once + // via config.toml [budget] default_max_llm_tokens_per_hour = 10000000 + if budget.default_max_llm_tokens_per_hour > 0 { + resources.max_llm_tokens_per_hour = budget.default_max_llm_tokens_per_hour; + } } /// Infer provider from a model name when catalog lookup fails. diff --git a/crates/openfang-kernel/src/metering.rs b/crates/openfang-kernel/src/metering.rs index a42ca616..82fa30a1 100644 --- a/crates/openfang-kernel/src/metering.rs +++ b/crates/openfang-kernel/src/metering.rs @@ -128,6 +128,7 @@ impl MeteringEngine { 0.0 }, alert_threshold: budget.alert_threshold, + default_max_llm_tokens_per_hour: budget.default_max_llm_tokens_per_hour, } } @@ -224,6 +225,8 @@ pub struct BudgetStatus { pub monthly_limit: f64, pub monthly_pct: f64, pub alert_threshold: f64, + /// Global default token limit per agent per hour (0 = use per-agent values). + pub default_max_llm_tokens_per_hour: u64, } /// Returns (input_per_million, output_per_million) pricing for a model. diff --git a/crates/openfang-kernel/src/registry.rs b/crates/openfang-kernel/src/registry.rs index a21e13fa..f1f18dbf 100644 --- a/crates/openfang-kernel/src/registry.rs +++ b/crates/openfang-kernel/src/registry.rs @@ -284,6 +284,7 @@ impl AgentRegistry { hourly: Option, daily: Option, monthly: Option, + tokens_per_hour: Option, ) -> OpenFangResult<()> { let mut entry = self .agents @@ -298,6 +299,9 @@ impl AgentRegistry { if let Some(v) = monthly { entry.manifest.resources.max_cost_per_month_usd = v; } + if let Some(v) = tokens_per_hour { + entry.manifest.resources.max_llm_tokens_per_hour = v; + } entry.last_active = chrono::Utc::now(); Ok(()) } diff --git a/crates/openfang-kernel/src/triggers.rs b/crates/openfang-kernel/src/triggers.rs index baf10982..7803a1ad 100644 --- a/crates/openfang-kernel/src/triggers.rs +++ b/crates/openfang-kernel/src/triggers.rs @@ -278,7 +278,7 @@ fn describe_event(event: &Event) -> String { tr.tool_id, if tr.success { "succeeded" } else { "failed" }, tr.execution_time_ms, - &tr.content[..tr.content.len().min(200)] + openfang_types::truncate_str(&tr.content, 200) ) } EventPayload::MemoryUpdate(delta) => { diff --git a/crates/openfang-memory/src/session.rs b/crates/openfang-memory/src/session.rs index b109c263..12dcec29 100644 --- a/crates/openfang-memory/src/session.rs +++ b/crates/openfang-memory/src/session.rs @@ -587,7 +587,7 @@ impl SessionStore { ContentBlock::Thinking { thinking } => { text_parts.push(format!( "[thinking: {}]", - &thinking[..thinking.len().min(200)] + openfang_types::truncate_str(thinking, 200) )); } ContentBlock::Unknown => {} diff --git a/crates/openfang-runtime/src/compactor.rs b/crates/openfang-runtime/src/compactor.rs index 6fd7056e..2af9da2b 100644 --- a/crates/openfang-runtime/src/compactor.rs +++ b/crates/openfang-runtime/src/compactor.rs @@ -1227,7 +1227,7 @@ mod tests { assert!( text.contains("truncated from"), "Oversized message should be truncated, got: {}", - &text[..text.len().min(200)] + crate::str_utils::safe_truncate_str(&text, 200) ); } diff --git a/crates/openfang-runtime/src/docker_sandbox.rs b/crates/openfang-runtime/src/docker_sandbox.rs index bdb73ccd..512da074 100644 --- a/crates/openfang-runtime/src/docker_sandbox.rs +++ b/crates/openfang-runtime/src/docker_sandbox.rs @@ -100,7 +100,7 @@ pub async fn create_sandbox( let container_name = sanitize_container_name(&format!( "{}-{}", config.container_prefix, - &agent_id[..agent_id.len().min(8)] + crate::str_utils::safe_truncate_str(agent_id, 8) ))?; let mut cmd = tokio::process::Command::new("docker"); diff --git a/crates/openfang-runtime/src/drivers/openai.rs b/crates/openfang-runtime/src/drivers/openai.rs index 0ccbfe20..ed568570 100644 --- a/crates/openfang-runtime/src/drivers/openai.rs +++ b/crates/openfang-runtime/src/drivers/openai.rs @@ -166,6 +166,9 @@ struct OaiChoice { struct OaiResponseMessage { content: Option, tool_calls: Option>, + /// Reasoning/thinking content returned by some models (DeepSeek-R1, Qwen3, etc.) + /// via LM Studio, Ollama, and other local inference servers. + reasoning_content: Option, } #[derive(Debug, Deserialize)] @@ -489,12 +492,52 @@ impl LlmDriver for OpenAIDriver { let mut content = Vec::new(); let mut tool_calls = Vec::new(); + // Capture reasoning_content from models that use a separate field + // (DeepSeek-R1, Qwen3, etc. via LM Studio/Ollama) + if let Some(ref reasoning) = choice.message.reasoning_content { + if !reasoning.is_empty() { + debug!(len = reasoning.len(), "Captured reasoning_content from response"); + content.push(ContentBlock::Thinking { + thinking: reasoning.clone(), + }); + } + } + if let Some(text) = choice.message.content { if !text.is_empty() { - content.push(ContentBlock::Text { text }); + // Extract ... blocks that some local models + // embed directly in the content field. + let (cleaned, thinking) = extract_think_tags(&text); + if let Some(think_text) = thinking { + // Only add if we didn't already get reasoning_content + if choice.message.reasoning_content.is_none() { + content.push(ContentBlock::Thinking { + thinking: think_text, + }); + } + } + if !cleaned.is_empty() { + content.push(ContentBlock::Text { text: cleaned }); + } } } + // If we have reasoning but no text content and no tool calls, + // synthesize a brief text block so the agent loop doesn't treat + // this as an empty response. + let has_text = content.iter().any(|b| matches!(b, ContentBlock::Text { .. })); + let has_thinking = content.iter().any(|b| matches!(b, ContentBlock::Thinking { .. })); + if has_thinking && !has_text && choice.message.tool_calls.is_none() { + // Extract the last sentence or line from the thinking as a response + let thinking_text = content.iter().find_map(|b| match b { + ContentBlock::Thinking { thinking } => Some(thinking.as_str()), + _ => None, + }).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 }); + } + if let Some(calls) = choice.message.tool_calls { for call in calls { let input: serde_json::Value = @@ -525,7 +568,7 @@ impl LlmDriver for OpenAIDriver { } }; - let usage = oai_response + let mut usage = oai_response .usage .map(|u| TokenUsage { input_tokens: u.prompt_tokens, @@ -533,6 +576,15 @@ impl LlmDriver for OpenAIDriver { }) .unwrap_or_default(); + // Guard: if the model returned content but usage is missing/zero + // (common with local LLMs like LM Studio, Ollama), set a synthetic + // non-zero output_tokens so the agent loop doesn't misclassify + // this as a "silent failure" and loop unnecessarily. + if !content.is_empty() && usage.input_tokens == 0 && usage.output_tokens == 0 { + debug!("Response has content but no usage stats — setting synthetic output_tokens=1"); + usage.output_tokens = 1; + } + return Ok(CompletionResponse { content, stop_reason, @@ -835,6 +887,7 @@ impl LlmDriver for OpenAIDriver { // Parse the SSE stream let mut buffer = String::new(); let mut text_content = String::new(); + let mut reasoning_content = String::new(); // Track tool calls: index -> (id, name, arguments) let mut tool_accum: Vec<(String, String, String)> = Vec::new(); let mut finish_reason: Option = None; @@ -902,6 +955,18 @@ impl LlmDriver for OpenAIDriver { } } + // Reasoning/thinking content delta (DeepSeek-R1, Qwen3 via LM Studio/Ollama) + if let Some(reasoning) = delta["reasoning_content"].as_str() { + if !reasoning.is_empty() { + reasoning_content.push_str(reasoning); + let _ = tx + .send(StreamEvent::ThinkingDelta { + text: reasoning.to_string(), + }) + .await; + } + } + // Tool call deltas if let Some(calls) = delta["tool_calls"].as_array() { for call in calls { @@ -954,6 +1019,7 @@ impl LlmDriver for OpenAIDriver { // Log stream summary for diagnostics let is_empty_stream = text_content.is_empty() + && reasoning_content.is_empty() && tool_accum.is_empty() && usage.input_tokens == 0 && usage.output_tokens == 0; @@ -970,6 +1036,7 @@ impl LlmDriver for OpenAIDriver { chunks = chunk_count, sse_lines = sse_line_count, text_len = text_content.len(), + reasoning_len = reasoning_content.len(), tool_count = tool_accum.len(), finish = ?finish_reason, input_tokens = usage.input_tokens, @@ -983,8 +1050,42 @@ impl LlmDriver for OpenAIDriver { let mut content = Vec::new(); let mut tool_calls = Vec::new(); + // Add reasoning/thinking content if present + if !reasoning_content.is_empty() { + content.push(ContentBlock::Thinking { + thinking: reasoning_content.clone(), + }); + } + if !text_content.is_empty() { - content.push(ContentBlock::Text { text: text_content }); + // Extract ... blocks from streamed text content + let (cleaned, thinking) = extract_think_tags(&text_content); + if let Some(think_text) = thinking { + // Only add if we didn't already get reasoning_content + if reasoning_content.is_empty() { + content.push(ContentBlock::Thinking { + thinking: think_text, + }); + } + } + if !cleaned.is_empty() { + content.push(ContentBlock::Text { text: cleaned }); + } + } + + // If we have reasoning but no text content and no tool calls, + // synthesize a brief text block so the agent loop doesn't treat + // this as an empty response. + let has_text = content.iter().any(|b| matches!(b, ContentBlock::Text { .. })); + let has_thinking = content.iter().any(|b| matches!(b, ContentBlock::Thinking { .. })); + if has_thinking && !has_text && tool_accum.is_empty() { + let thinking_text = content.iter().find_map(|b| match b { + ContentBlock::Thinking { thinking } => Some(thinking.as_str()), + _ => None, + }).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 }); } for (id, name, arguments) in &tool_accum { @@ -1022,6 +1123,15 @@ impl LlmDriver for OpenAIDriver { } }; + // Guard: if the model returned content but usage is missing/zero + // (common with local LLMs like LM Studio, Ollama), set a synthetic + // non-zero output_tokens so the agent loop doesn't misclassify + // this as a "silent failure" and loop unnecessarily. + if !content.is_empty() && usage.input_tokens == 0 && usage.output_tokens == 0 { + debug!("Stream has content but no usage stats — setting synthetic output_tokens=1"); + usage.output_tokens = 1; + } + let _ = tx .send(StreamEvent::ContentComplete { stop_reason, usage }) .await; @@ -1041,6 +1151,85 @@ impl LlmDriver for OpenAIDriver { } } +/// Extract `...` blocks from content text. +/// +/// Some local LLMs (Qwen3, DeepSeek-R1) embed their reasoning directly in the +/// content field wrapped in `` tags. This function separates the thinking +/// from the actual response text. +/// +/// Returns `(cleaned_text, Option)`. +fn extract_think_tags(text: &str) -> (String, Option) { + let mut thinking_parts = Vec::new(); + let mut cleaned = text.to_string(); + + // Extract all ... blocks (greedy within each block) + while let Some(start) = cleaned.find("") { + if let Some(end) = cleaned.find("") { + let think_start = start + "".len(); + if think_start <= end { + let thought = cleaned[think_start..end].trim().to_string(); + if !thought.is_empty() { + thinking_parts.push(thought); + } + // Remove the entire ... block + cleaned = format!( + "{}{}", + &cleaned[..start], + &cleaned[end + "".len()..] + ); + } else { + break; + } + } else { + // Unclosed tag — treat everything after as thinking + let thought = cleaned[start + "".len()..].trim().to_string(); + if !thought.is_empty() { + thinking_parts.push(thought); + } + cleaned = cleaned[..start].to_string(); + break; + } + } + + let cleaned = cleaned.trim().to_string(); + if thinking_parts.is_empty() { + (cleaned, None) + } else { + (cleaned, Some(thinking_parts.join("\n\n"))) + } +} + +/// Extract a usable summary from thinking-only output. +/// +/// When a local model returns only thinking/reasoning with no actual response text, +/// we extract the last meaningful paragraph as a synthesized response rather than +/// showing "empty response" to the user. +fn extract_thinking_summary(thinking: &str) -> String { + let trimmed = thinking.trim(); + if trimmed.is_empty() { + return "[The model produced reasoning but no final answer. Try rephrasing your question.]".to_string(); + } + + // Take the last non-empty paragraph (models usually conclude with their answer) + let paragraphs: Vec<&str> = trimmed + .split("\n\n") + .map(|p| p.trim()) + .filter(|p| !p.is_empty()) + .collect(); + + if let Some(last) = paragraphs.last() { + // If the last paragraph is reasonably short, use it directly + if last.len() <= 2000 { + last.to_string() + } else { + // Take the last 2000 chars + last[last.len() - 2000..].to_string() + } + } else { + "[The model produced reasoning but no final answer. Try rephrasing your question.]".to_string() + } +} + /// Parse Groq's `tool_use_failed` error and extract the tool call from `failed_generation`. /// Extract the max_tokens limit from an API error message. /// Looks for patterns like: `must be less than or equal to \`8192\`` @@ -1265,4 +1454,95 @@ mod tests { fn test_extract_max_tokens_limit_no_match() { assert_eq!(extract_max_tokens_limit("some random error"), None); } + + // ----- extract_think_tags tests ----- + + #[test] + fn test_extract_think_tags_no_tags() { + let (cleaned, thinking) = extract_think_tags("Hello world"); + assert_eq!(cleaned, "Hello world"); + assert!(thinking.is_none()); + } + + #[test] + fn test_extract_think_tags_with_thinking() { + let input = "Let me reason about this...The answer is 42."; + let (cleaned, thinking) = extract_think_tags(input); + assert_eq!(cleaned, "The answer is 42."); + assert_eq!(thinking.unwrap(), "Let me reason about this..."); + } + + #[test] + fn test_extract_think_tags_only_thinking() { + let input = "I need to think about this carefully.\n\nThe user wants to know about Rust."; + let (cleaned, thinking) = extract_think_tags(input); + assert_eq!(cleaned, ""); + assert!(thinking.is_some()); + assert!(thinking.unwrap().contains("think about this carefully")); + } + + #[test] + fn test_extract_think_tags_multiple_blocks() { + let input = "First thoughtMiddle textSecond thoughtFinal text"; + let (cleaned, thinking) = extract_think_tags(input); + assert_eq!(cleaned, "Middle textFinal text"); + let t = thinking.unwrap(); + assert!(t.contains("First thought")); + assert!(t.contains("Second thought")); + } + + #[test] + fn test_extract_think_tags_unclosed() { + let input = "Some textunclosed thinking content"; + let (cleaned, thinking) = extract_think_tags(input); + assert_eq!(cleaned, "Some text"); + assert_eq!(thinking.unwrap(), "unclosed thinking content"); + } + + // ----- extract_thinking_summary tests ----- + + #[test] + fn test_extract_thinking_summary_empty() { + let summary = extract_thinking_summary(""); + assert!(summary.contains("no final answer")); + } + + #[test] + fn test_extract_thinking_summary_single_paragraph() { + let summary = extract_thinking_summary("The answer is 42."); + assert_eq!(summary, "The answer is 42."); + } + + #[test] + fn test_extract_thinking_summary_multiple_paragraphs() { + let input = "First I need to consider X.\n\nThen I should check Y.\n\nThe answer is 42."; + let summary = extract_thinking_summary(input); + assert_eq!(summary, "The answer is 42."); + } + + // ----- reasoning_content deserialization test ----- + + #[test] + fn test_oai_response_message_with_reasoning_content() { + let json = r#"{"content": null, "reasoning_content": "Let me think...", "tool_calls": null}"#; + let msg: OaiResponseMessage = serde_json::from_str(json).unwrap(); + assert!(msg.content.is_none()); + assert_eq!(msg.reasoning_content.as_deref(), Some("Let me think...")); + } + + #[test] + fn test_oai_response_message_without_reasoning_content() { + let json = r#"{"content": "Hello", "tool_calls": null}"#; + let msg: OaiResponseMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.content.as_deref(), Some("Hello")); + assert!(msg.reasoning_content.is_none()); + } + + #[test] + fn test_oai_response_message_null_content_null_reasoning() { + let json = r#"{"content": null, "tool_calls": null}"#; + let msg: OaiResponseMessage = serde_json::from_str(json).unwrap(); + assert!(msg.content.is_none()); + assert!(msg.reasoning_content.is_none()); + } } diff --git a/crates/openfang-runtime/src/provider_health.rs b/crates/openfang-runtime/src/provider_health.rs index b9990bbe..1b17260c 100644 --- a/crates/openfang-runtime/src/provider_health.rs +++ b/crates/openfang-runtime/src/provider_health.rs @@ -262,7 +262,7 @@ pub async fn probe_model( } else { let status = resp.status().as_u16(); let body = resp.text().await.unwrap_or_default(); - Err(format!("HTTP {status}: {}", &body[..body.len().min(200)])) + Err(format!("HTTP {status}: {}", crate::str_utils::safe_truncate_str(&body, 200))) } } diff --git a/crates/openfang-runtime/src/subprocess_sandbox.rs b/crates/openfang-runtime/src/subprocess_sandbox.rs index 0a9b6384..edec65b3 100644 --- a/crates/openfang-runtime/src/subprocess_sandbox.rs +++ b/crates/openfang-runtime/src/subprocess_sandbox.rs @@ -207,7 +207,7 @@ pub fn validate_command_allowlist(command: &str, policy: &ExecPolicy) -> Result< } ExecSecurityMode::Full => { tracing::warn!( - command = &command[..command.len().min(100)], + command = crate::str_utils::safe_truncate_str(command, 100), "Shell exec in full mode — no restrictions" ); Ok(()) @@ -855,4 +855,51 @@ mod tests { assert!(validate_command_allowlist("echo ${HOME}", &policy).is_err()); assert!(validate_command_allowlist("echo hello\ncurl bad", &policy).is_err()); } + + // ── CJK / multi-byte safety tests (issue #490) ────────────────────── + + #[test] + fn test_full_mode_cjk_command_no_panic() { + // CJK characters are 3 bytes each. A command string with CJK chars + // must not panic when we truncate it for tracing in Full mode. + let policy = ExecPolicy { + mode: ExecSecurityMode::Full, + ..ExecPolicy::default() + }; + // 50 CJK chars = 150 bytes — truncation at byte 100 would land + // mid-char without safe_truncate_str. + let cjk_command: String = "\u{4e16}".repeat(50); + assert!(validate_command_allowlist(&cjk_command, &policy).is_ok()); + } + + #[test] + fn test_full_mode_mixed_cjk_ascii_no_panic() { + let policy = ExecPolicy { + mode: ExecSecurityMode::Full, + ..ExecPolicy::default() + }; + // "echo " (5 bytes) + 40 CJK chars (120 bytes) = 125 bytes total. + // Byte 100 falls inside a 3-byte CJK char. + let mut cmd = String::from("echo "); + cmd.extend(std::iter::repeat_n('\u{4f60}', 40)); + assert!(validate_command_allowlist(&cmd, &policy).is_ok()); + } + + #[test] + fn test_allowlist_cjk_unlisted_no_panic() { + let policy = ExecPolicy::default(); + // CJK command not in allowlist — should return Err, not panic + let cjk_cmd: String = "\u{597d}".repeat(50); + assert!(validate_command_allowlist(&cjk_cmd, &policy).is_err()); + } + + #[test] + fn test_extract_all_commands_cjk_separators() { + // Ensure extract_all_commands handles CJK content between separators + // without panicking (separators are ASCII, but content is CJK) + let cmd = "\u{4f60}\u{597d}"; + let cmds = extract_all_commands(cmd); + assert_eq!(cmds.len(), 1); + assert_eq!(cmds[0], "\u{4f60}\u{597d}"); + } } diff --git a/crates/openfang-runtime/src/tool_runner.rs b/crates/openfang-runtime/src/tool_runner.rs index cf05a801..bff189b3 100644 --- a/crates/openfang-runtime/src/tool_runner.rs +++ b/crates/openfang-runtime/src/tool_runner.rs @@ -3144,7 +3144,7 @@ async fn tool_canvas_present( let _ = tokio::fs::create_dir_all(&output_dir).await; let timestamp = chrono::Utc::now().format("%Y%m%d_%H%M%S"); - let filename = format!("canvas_{timestamp}_{}.html", &canvas_id[..8]); + let filename = format!("canvas_{timestamp}_{}.html", crate::str_utils::safe_truncate_str(&canvas_id, 8)); let filepath = output_dir.join(&filename); // Write the full HTML document diff --git a/crates/openfang-runtime/src/web_fetch.rs b/crates/openfang-runtime/src/web_fetch.rs index b76ea08c..bd916cef 100644 --- a/crates/openfang-runtime/src/web_fetch.rs +++ b/crates/openfang-runtime/src/web_fetch.rs @@ -4,6 +4,7 @@ //! Pipeline: SSRF check → cache lookup → HTTP GET → detect HTML → //! html_to_markdown() → truncate → wrap_external_content() → cache → return +use crate::str_utils::safe_truncate_str; use crate::web_cache::WebCache; use crate::web_content::{html_to_markdown, wrap_external_content}; use openfang_types::config::WebFetchConfig; @@ -23,6 +24,9 @@ impl WebFetchEngine { pub fn new(config: WebFetchConfig, cache: Arc) -> Self { let client = reqwest::Client::builder() .timeout(std::time::Duration::from_secs(config.timeout_secs)) + .gzip(true) + .deflate(true) + .brotli(true) .build() .unwrap_or_default(); Self { @@ -132,11 +136,11 @@ impl WebFetchEngine { resp_body }; - // Step 5: Truncate + // Step 5: Truncate (char-boundary-safe to avoid panics on multi-byte UTF-8) let truncated = if processed.len() > self.config.max_chars { format!( "{}... [truncated, {} total chars]", - &processed[..self.config.max_chars], + safe_truncate_str(&processed, self.config.max_chars), processed.len() ) } else { @@ -277,6 +281,28 @@ fn extract_host(url: &str) -> String { #[cfg(test)] mod tests { use super::*; + use crate::str_utils::safe_truncate_str; + + #[test] + fn test_truncate_multibyte_no_panic() { + // Simulate a gzip-decoded response containing multi-byte UTF-8 + // (Chinese, Japanese, emoji — common on international finance sites). + // Old code: &s[..max] panics when max lands inside a multi-byte char. + let content = "\u{4f60}\u{597d}\u{4e16}\u{754c}!"; // "你好世界!" = 13 bytes + // Truncate at byte 7 — lands inside the 3rd Chinese char (bytes 6..9). + // safe_truncate_str walks back to byte 6, returning "你好". + let truncated = safe_truncate_str(content, 7); + assert_eq!(truncated, "\u{4f60}\u{597d}"); + assert!(truncated.len() <= 7); + } + + #[test] + fn test_truncate_emoji_no_panic() { + let content = "\u{1f4b0}\u{1f4c8}\u{1f4b9}"; // 💰📈💹 = 12 bytes + // Truncate at byte 5 — lands inside the 2nd emoji (bytes 4..8). + let truncated = safe_truncate_str(content, 5); + assert_eq!(truncated, "\u{1f4b0}"); // 4 bytes + } #[test] fn test_ssrf_blocks_localhost() { diff --git a/crates/openfang-types/src/config.rs b/crates/openfang-types/src/config.rs index 99aca824..fd799292 100644 --- a/crates/openfang-types/src/config.rs +++ b/crates/openfang-types/src/config.rs @@ -1108,6 +1108,11 @@ pub struct BudgetConfig { pub max_monthly_usd: f64, /// Alert threshold as a fraction (0.0 - 1.0). Trigger warnings at this % of any limit. pub alert_threshold: f64, + /// Default per-agent hourly token limit override. When set (> 0), agents that + /// still have the built-in default (1,000,000) or a lower per-agent value will + /// be overridden to this value. Set to 0 to keep each agent's own limit. + /// Use this to globally raise or lower the token budget for all agents. + pub default_max_llm_tokens_per_hour: u64, } impl Default for BudgetConfig { @@ -1117,6 +1122,7 @@ impl Default for BudgetConfig { max_daily_usd: 0.0, max_monthly_usd: 0.0, alert_threshold: 0.8, + default_max_llm_tokens_per_hour: 0, } } } diff --git a/crates/openfang-types/src/tool.rs b/crates/openfang-types/src/tool.rs index 507c7ebe..a977a5b4 100644 --- a/crates/openfang-types/src/tool.rs +++ b/crates/openfang-types/src/tool.rs @@ -83,20 +83,73 @@ fn normalize_schema_recursive(schema: &serde_json::Value) -> serde_json::Value { // Strip fields unsupported by Gemini and most non-Anthropic providers if matches!( key.as_str(), - "$schema" | "$defs" | "$ref" | "additionalProperties" | "default" - | "$id" | "$comment" | "examples" | "title" + "$schema" + | "$defs" + | "$ref" + | "additionalProperties" + | "default" + | "$id" + | "$comment" + | "examples" + | "title" + | "const" + | "format" ) { continue; } - // Convert anyOf to flat type + enum when possible - if key == "anyOf" { + // Convert anyOf/oneOf to flat type + enum when possible + if key == "anyOf" || key == "oneOf" { if let Some(converted) = try_flatten_any_of(value) { for (k, v) in converted { result.insert(k, v); } continue; } + // Can't flatten — strip entirely rather than leave unsupported keyword + continue; + } + + // Flatten type arrays like ["string", "null"] to single type + nullable + if key == "type" { + if let Some(arr) = value.as_array() { + let types: Vec<&str> = arr.iter().filter_map(|v| v.as_str()).collect(); + let has_null = types.contains(&"null"); + let non_null: Vec<&&str> = + types.iter().filter(|&&t| t != "null").collect(); + if has_null && non_null.len() == 1 { + // ["string", "null"] → type: "string", nullable: true + result.insert( + "type".to_string(), + serde_json::Value::String(non_null[0].to_string()), + ); + result.insert("nullable".to_string(), serde_json::Value::Bool(true)); + continue; + } else if non_null.len() == 1 { + // ["string"] → type: "string" + result.insert( + "type".to_string(), + serde_json::Value::String(non_null[0].to_string()), + ); + continue; + } else if !non_null.is_empty() { + // Multiple non-null types — pick first (best effort) + result.insert( + "type".to_string(), + serde_json::Value::String(non_null[0].to_string()), + ); + if has_null { + result.insert( + "nullable".to_string(), + serde_json::Value::Bool(true), + ); + } + continue; + } + } + // Scalar type string — pass through + result.insert(key.clone(), value.clone()); + continue; } // Recurse into properties @@ -215,17 +268,20 @@ fn try_flatten_any_of(any_of: &serde_json::Value) -> Option 1 { - let type_array: Vec = - types.into_iter().map(serde_json::Value::String).collect(); - return Some(vec![( + let mut result = vec![( "type".to_string(), - serde_json::Value::Array(type_array), - )]); + serde_json::Value::String(types[0].clone()), + )]; + if has_null { + result.push(("nullable".to_string(), serde_json::Value::Bool(true))); + } + return Some(result); } - // Can't flatten — leave as-is + // Can't flatten — caller will strip the key entirely None } @@ -299,7 +355,9 @@ mod tests { }); let result = normalize_schema_for_provider(&schema, "groq"); let value_prop = &result["properties"]["value"]; - assert!(value_prop["type"].is_array()); + // Gemini rejects type arrays — should flatten to first type + assert_eq!(value_prop["type"], "string"); + assert!(value_prop.get("anyOf").is_none()); } #[test] @@ -432,4 +490,167 @@ mod tests { assert!(result.get("$defs").is_none()); assert_eq!(result["properties"]["x"]["type"], "string"); } + + // --- Issue #488 tests --- + + #[test] + fn test_normalize_strips_const() { + let schema = serde_json::json!({ + "type": "object", + "properties": { + "version": { "type": "string", "const": "v1" } + } + }); + let result = normalize_schema_for_provider(&schema, "gemini"); + assert!(result["properties"]["version"].get("const").is_none()); + assert_eq!(result["properties"]["version"]["type"], "string"); + } + + #[test] + fn test_normalize_strips_format() { + let schema = serde_json::json!({ + "type": "object", + "properties": { + "created_at": { "type": "string", "format": "date-time" }, + "email": { "type": "string", "format": "email" } + } + }); + let result = normalize_schema_for_provider(&schema, "gemini"); + assert!(result["properties"]["created_at"].get("format").is_none()); + assert!(result["properties"]["email"].get("format").is_none()); + assert_eq!(result["properties"]["created_at"]["type"], "string"); + assert_eq!(result["properties"]["email"]["type"], "string"); + } + + #[test] + fn test_normalize_flattens_oneof_nullable() { + let schema = serde_json::json!({ + "type": "object", + "properties": { + "value": { + "oneOf": [ + { "type": "string" }, + { "type": "null" } + ] + } + } + }); + let result = normalize_schema_for_provider(&schema, "gemini"); + let value_prop = &result["properties"]["value"]; + assert_eq!(value_prop["type"], "string"); + assert_eq!(value_prop["nullable"], true); + assert!(value_prop.get("oneOf").is_none()); + } + + #[test] + fn test_normalize_strips_oneof_complex() { + // Complex oneOf that can't be flattened — should be stripped entirely + let schema = serde_json::json!({ + "type": "object", + "properties": { + "data": { + "oneOf": [ + { "type": "object", "properties": { "a": { "type": "string" } } }, + { "type": "object", "properties": { "b": { "type": "number" } } } + ] + } + } + }); + let result = normalize_schema_for_provider(&schema, "gemini"); + let data_prop = &result["properties"]["data"]; + assert!(data_prop.get("oneOf").is_none()); + } + + #[test] + fn test_normalize_flattens_type_array_nullable() { + let schema = serde_json::json!({ + "type": "object", + "properties": { + "name": { "type": ["string", "null"] } + } + }); + let result = normalize_schema_for_provider(&schema, "gemini"); + let name_prop = &result["properties"]["name"]; + assert_eq!(name_prop["type"], "string"); + assert_eq!(name_prop["nullable"], true); + } + + #[test] + fn test_normalize_flattens_type_array_multi() { + let schema = serde_json::json!({ + "type": "object", + "properties": { + "value": { "type": ["string", "number", "null"] } + } + }); + let result = normalize_schema_for_provider(&schema, "gemini"); + let value_prop = &result["properties"]["value"]; + // Should pick first non-null type + assert_eq!(value_prop["type"], "string"); + assert_eq!(value_prop["nullable"], true); + } + + #[test] + fn test_normalize_flattens_type_array_single() { + // Single-element type array + let schema = serde_json::json!({ + "type": "object", + "properties": { + "x": { "type": ["integer"] } + } + }); + let result = normalize_schema_for_provider(&schema, "gemini"); + assert_eq!(result["properties"]["x"]["type"], "integer"); + assert!(result["properties"]["x"].get("nullable").is_none()); + } + + #[test] + fn test_normalize_strips_anyof_complex() { + // Complex anyOf that can't be flattened — should be stripped entirely + let schema = serde_json::json!({ + "type": "object", + "properties": { + "payload": { + "anyOf": [ + { "type": "object", "properties": { "url": { "type": "string" } } }, + { "type": "array", "items": { "type": "string" } } + ] + } + } + }); + let result = normalize_schema_for_provider(&schema, "gemini"); + let payload_prop = &result["properties"]["payload"]; + assert!(payload_prop.get("anyOf").is_none()); + } + + #[test] + fn test_normalize_combined_issue_488() { + // Real-world schema combining multiple #488 issues + let schema = serde_json::json!({ + "type": "object", + "properties": { + "api_version": { "type": "string", "const": "v2", "format": "semver" }, + "timestamp": { "type": "string", "format": "date-time" }, + "label": { + "oneOf": [ + { "type": "string" }, + { "type": "null" } + ] + }, + "tags": { "type": ["string", "null"] } + } + }); + let result = normalize_schema_for_provider(&schema, "gemini"); + // const and format stripped + assert!(result["properties"]["api_version"].get("const").is_none()); + assert!(result["properties"]["api_version"].get("format").is_none()); + assert!(result["properties"]["timestamp"].get("format").is_none()); + // oneOf flattened + assert_eq!(result["properties"]["label"]["type"], "string"); + assert_eq!(result["properties"]["label"]["nullable"], true); + assert!(result["properties"]["label"].get("oneOf").is_none()); + // type array flattened + assert_eq!(result["properties"]["tags"]["type"], "string"); + assert_eq!(result["properties"]["tags"]["nullable"], true); + } } diff --git a/crates/openfang-wire/src/peer.rs b/crates/openfang-wire/src/peer.rs index ea98d266..64229d2e 100644 --- a/crates/openfang-wire/src/peer.rs +++ b/crates/openfang-wire/src/peer.rs @@ -55,7 +55,7 @@ impl NonceTracker { // Check for replay if self.seen.contains_key(nonce) { - return Err(format!("Nonce replay detected: {}", &nonce[..nonce.len().min(16)])); + return Err(format!("Nonce replay detected: {}", openfang_types::truncate_str(nonce, 16))); } // Record the nonce
AgentTodayHourly LimitDaily LimitMonthly Limit
AgentTodayHourly LimitDaily LimitMonthly LimitToken Limit/hr