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

fix 7 issues

Browse Source
This commit is contained in:
jaberjaber23
2026-03-10 16:53:04 +03:00
parent 86b50070e8
commit edd0fed518
21 changed files with 760 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+15 -14
View File
@@ -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"
Cargo.toml
+2 -2
View File
@@ -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"
crates/openfang-api/src/routes.rs
+24 -12
View File
@@ -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<Arc<AppState>>) -> 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() {
crates/openfang-api/static/index_body.html
+15 -4
View File
@@ -3458,7 +3458,7 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"]</pre>
<div x-show="tab === 'budget'" x-data="{
budgetData: null, agentRanking: [], budgetLoading: true,
editMode: false,
editHourly: '', editDaily: '', editMonthly: '', editAlert: '',
editHourly: '', editDaily: '', editMonthly: '', editAlert: '', editTokenLimit: '',
saving: false,
async loadBudget() {
this.budgetLoading = true;
@@ -3477,6 +3477,7 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"]</pre>
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"]</pre>
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"]</pre>
</div>
</div>
</div>
<div class="text-xs text-dim mb-3" x-show="budgetData.alert_threshold > 0 && !editMode">
<div class="text-xs text-dim mb-1" x-show="budgetData.alert_threshold > 0 && !editMode">
Alert threshold: <span x-text="(budgetData.alert_threshold * 100).toFixed(0) + '%'"></span> of any limit
</div>
<div class="text-xs text-dim mb-3" x-show="!editMode">
Hourly token limit (per agent): <span x-text="fmtTokens(budgetData.default_max_llm_tokens_per_hour || 0)"></span>
</div>
<!-- Edit limits form -->
<div x-show="editMode" class="card" style="margin:12px 0;padding:12px;border:1px solid var(--accent);border-radius:6px">
@@ -3557,7 +3563,11 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"]</pre>
<input type="number" step="5" min="0" max="100" x-model="editAlert" class="input" style="width:100%;margin-top:2px" placeholder="80">
</div>
</div>
<div class="text-xs text-dim">Set to 0 for unlimited. Changes apply immediately (in-memory, not persisted to config.toml).</div>
<div style="margin-bottom:8px">
<label class="text-xs text-dim">Hourly Token Limit (per agent, 0 = use per-agent values)</label>
<input type="number" step="100000" min="0" x-model="editTokenLimit" class="input" style="width:260px;margin-top:2px" placeholder="0 = per-agent default">
</div>
<div class="text-xs text-dim">Set to 0 for unlimited/per-agent default. Changes apply immediately (in-memory, not persisted to config.toml).</div>
<button class="btn btn-sm mt-2" @click="editMode = false" style="margin-right:8px">Cancel</button>
</div>
@@ -3565,7 +3575,7 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"]</pre>
<h4 style="margin-top:16px;margin-bottom:8px">Top Spenders (Today)</h4>
<div class="table-wrap" x-show="agentRanking.length">
<table>
<thead><tr><th>Agent</th><th>Today</th><th>Hourly Limit</th><th>Daily Limit</th><th>Monthly Limit</th></tr></thead>
<thead><tr><th>Agent</th><th>Today</th><th>Hourly Limit</th><th>Daily Limit</th><th>Monthly Limit</th><th>Token Limit/hr</th></tr></thead>
<tbody>
<template x-for="a in agentRanking" :key="a.agent_id">
<tr>
@@ -3574,6 +3584,7 @@ args = ["-y", "@modelcontextprotocol/server-filesystem", "/path"]</pre>
<td class="text-dim" x-text="fmtUsd(a.hourly_limit)"></td>
<td class="text-dim" x-text="fmtUsd(a.daily_limit)"></td>
<td class="text-dim" x-text="fmtUsd(a.monthly_limit)"></td>
<td class="text-dim" x-text="fmtTokens(a.max_llm_tokens_per_hour || 0)"></td>
</tr>
</template>
</tbody>
crates/openfang-channels/src/bridge.rs
+35 -7
View File
@@ -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<dyn ChannelAdapter>,
@@ -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());
crates/openfang-channels/src/nostr.rs
+1 -1
View File
@@ -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,
},
crates/openfang-kernel/src/kernel.rs
+56 -5
View File
@@ -151,6 +151,10 @@ pub struct OpenFangKernel {
pub channel_adapters: dashmap::DashMap<String, Arc<dyn openfang_channels::types::ChannelAdapter>>,
/// Hot-reloadable default model override (set via config hot-reload, read at agent spawn).
pub default_model_override: std::sync::RwLock<Option<openfang_types::config::DefaultModelConfig>>,
/// 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<AgentId, Arc<tokio::sync::Mutex<()>>>,
/// Weak self-reference for trigger dispatch (set after Arc wrapping).
self_handle: OnceLock<Weak<OpenFangKernel>>,
}
@@ -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<Arc<dyn KernelHandle>>,
content_blocks: Option<Vec<openfang_types::message::ContentBlock>>,
) -> KernelResult<AgentLoopResult> {
// 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<String> {
// 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<Arc<dyn LlmDriver>> {
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.
crates/openfang-kernel/src/metering.rs
+3
View File
@@ -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.
crates/openfang-kernel/src/registry.rs
+4
View File
@@ -284,6 +284,7 @@ impl AgentRegistry {
hourly: Option<f64>,
daily: Option<f64>,
monthly: Option<f64>,
tokens_per_hour: Option<u64>,
) -> 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(())
}
crates/openfang-kernel/src/triggers.rs
+1 -1
View File
@@ -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) => {
crates/openfang-memory/src/session.rs
+1 -1
View File
@@ -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 => {}
crates/openfang-runtime/src/compactor.rs
+1 -1
View File
@@ -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)
);
}
crates/openfang-runtime/src/docker_sandbox.rs
+1 -1
View File
@@ -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");
crates/openfang-runtime/src/drivers/openai.rs
+283 -3
View File
@@ -166,6 +166,9 @@ struct OaiChoice {
struct OaiResponseMessage {
content: Option<String>,
tool_calls: Option<Vec<OaiToolCall>>,
/// Reasoning/thinking content returned by some models (DeepSeek-R1, Qwen3, etc.)
/// via LM Studio, Ollama, and other local inference servers.
reasoning_content: Option<String>,
}
#[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 <think>...</think> 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<String> = 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 <think>...</think> 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 `<think>...</think>` blocks from content text.
///
/// Some local LLMs (Qwen3, DeepSeek-R1) embed their reasoning directly in the
/// content field wrapped in `<think>` tags. This function separates the thinking
/// from the actual response text.
///
/// Returns `(cleaned_text, Option<thinking_text>)`.
fn extract_think_tags(text: &str) -> (String, Option<String>) {
let mut thinking_parts = Vec::new();
let mut cleaned = text.to_string();
// Extract all <think>...</think> blocks (greedy within each block)
while let Some(start) = cleaned.find("<think>") {
if let Some(end) = cleaned.find("</think>") {
let think_start = start + "<think>".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 <think>...</think> block
cleaned = format!(
"{}{}",
&cleaned[..start],
&cleaned[end + "</think>".len()..]
);
} else {
break;
}
} else {
// Unclosed <think> tag — treat everything after as thinking
let thought = cleaned[start + "<think>".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 = "<think>Let me reason about this...</think>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 = "<think>I need to think about this carefully.\n\nThe user wants to know about Rust.</think>";
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 = "<think>First thought</think>Middle text<think>Second thought</think>Final 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 text<think>unclosed 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());
}
}
crates/openfang-runtime/src/provider_health.rs
+1 -1
View File
@@ -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)))
}
}
crates/openfang-runtime/src/subprocess_sandbox.rs
+48 -1
View File
@@ -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}");
}
}
crates/openfang-runtime/src/tool_runner.rs
+1 -1
View File
@@ -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
crates/openfang-runtime/src/web_fetch.rs
+28 -2
View File
@@ -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<WebCache>) -> 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() {
crates/openfang-types/src/config.rs
+6
View File
@@ -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,
}
}
}
crates/openfang-types/src/tool.rs
+233 -12
View File
@@ -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<Vec<(String, serde_j
return Some(result);
}
// If all items are simple types, create a type array
// If all items are simple types, pick the first non-null type (best effort).
// Gemini rejects type arrays, so we can't emit ["string", "number"].
if types.len() == items.len() && types.len() > 1 {
let type_array: Vec<serde_json::Value> =
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);
}
}
crates/openfang-wire/src/peer.rs
+1 -1
View File
@@ -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