From 3f28c0f3d49b867f4dffd3b2ef95e6aa82b68293 Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Sun, 19 Jul 2026 14:39:58 +0530 Subject: [PATCH 1/7] feat: scaffold flare-proxy crate with shape_xlat + providers --- crates/flare-proxy/Cargo.toml | 17 + crates/flare-proxy/src/lib.rs | 27 ++ crates/flare-proxy/src/providers.rs | 123 +++++++ crates/flare-proxy/src/shape_xlat.rs | 515 +++++++++++++++++++++++++++ 4 files changed, 682 insertions(+) create mode 100644 crates/flare-proxy/Cargo.toml create mode 100644 crates/flare-proxy/src/lib.rs create mode 100644 crates/flare-proxy/src/providers.rs create mode 100644 crates/flare-proxy/src/shape_xlat.rs diff --git a/crates/flare-proxy/Cargo.toml b/crates/flare-proxy/Cargo.toml new file mode 100644 index 00000000..6e4b6b7b --- /dev/null +++ b/crates/flare-proxy/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "flare-proxy" +version = "0.1.0" +edition = "2021" +description = "Anthropic→OpenAI proxy for free providers (NVIDIA NIM, OpenRouter, LM Studio)" + +[dependencies] +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1", features = ["io-util", "sync", "net", "time"] } +axum = "0.8" +tower-http = "0.6" +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream", "json"] } +futures = "0.3" +base64 = "0.22" +regex = "1" +thiserror = "2" diff --git a/crates/flare-proxy/src/lib.rs b/crates/flare-proxy/src/lib.rs new file mode 100644 index 00000000..2e013195 --- /dev/null +++ b/crates/flare-proxy/src/lib.rs @@ -0,0 +1,27 @@ +mod heuristic; +mod providers; +mod shape_xlat; +mod think; + +pub use providers::{ProviderConfig, ProviderKind}; +use axum::Router; + +pub fn router() -> Router { + Router::new().route("/proxy/v1/messages", axum::routing::post(v1_messages_handler)) +} + +async fn v1_messages_handler( + axum::extract::State(state): axum::extract::State, + axum::extract::Json(body): axum::extract::Json, +) -> axum::response::Response { + // 1. Translate Anthropic request → OpenAI + // 2. Select provider + // 3. Forward + // 4. Translate response back → Anthropic + todo!() +} + +#[derive(Clone)] +struct AppState { + config: ProviderConfig, +} diff --git a/crates/flare-proxy/src/providers.rs b/crates/flare-proxy/src/providers.rs new file mode 100644 index 00000000..a8ec6846 --- /dev/null +++ b/crates/flare-proxy/src/providers.rs @@ -0,0 +1,123 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ProviderConfig { + pub providers: Vec, + pub routing: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ProviderEntry { + pub id: String, + pub kind: ProviderKind, + pub base_url: String, + pub api_key_env: Option, + pub default_model: Option, + pub models: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum ProviderKind { + NvidiaNim, + OpenRouter, + LmStudio, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ModelDef { + pub id: String, + pub upstream_model: String, + pub max_input_tokens: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ModelRoute { + pub anthropic_model: String, + pub provider_id: String, + pub upstream_model: String, + pub requires_heuristic_tools: bool, + pub requires_think_parsing: bool, +} + +impl ProviderConfig { + pub fn default_free() -> Self { + Self { + providers: vec![ + ProviderEntry { + id: "nvidia-nim".into(), + kind: ProviderKind::NvidiaNim, + base_url: "https://integrate.api.nvidia.com/v1".into(), + api_key_env: Some("NVIDIA_NIM_API_KEY".into()), + default_model: Some("meta/llama-3.1-405b-instruct".into()), + models: vec![ + ModelDef { + id: "meta/llama-3.1-405b-instruct".into(), + upstream_model: "meta/llama-3.1-405b-instruct".into(), + max_input_tokens: Some(128_000), + }, + ModelDef { + id: "meta/llama-3.3-70b-instruct".into(), + upstream_model: "meta/llama-3.3-70b-instruct".into(), + max_input_tokens: Some(128_000), + }, + ], + }, + ProviderEntry { + id: "openrouter".into(), + kind: ProviderKind::OpenRouter, + base_url: "https://openrouter.ai/api/v1".into(), + api_key_env: Some("OPENROUTER_API_KEY".into()), + default_model: None, + models: vec![ModelDef { + id: "openrouter/auto".into(), + upstream_model: "openrouter/auto".into(), + max_input_tokens: None, + }], + }, + ProviderEntry { + id: "lm-studio".into(), + kind: ProviderKind::LmStudio, + base_url: "http://localhost:1234/v1".into(), + api_key_env: None, + default_model: Some("local-model".into()), + models: vec![ModelDef { + id: "local-model".into(), + upstream_model: "local-model".into(), + max_input_tokens: Some(32_000), + }], + }, + ], + routing: vec![ + ModelRoute { + anthropic_model: "claude-sonnet-4-20250514".into(), + provider_id: "nvidia-nim".into(), + upstream_model: "meta/llama-3.1-405b-instruct".into(), + requires_heuristic_tools: true, + requires_think_parsing: false, + }, + ModelRoute { + anthropic_model: "claude-sonnet-4-5-20250601".into(), + provider_id: "openrouter".into(), + upstream_model: "openrouter/auto".into(), + requires_heuristic_tools: false, + requires_think_parsing: false, + }, + ModelRoute { + anthropic_model: "claude-haiku-3-5-20241022".into(), + provider_id: "lm-studio".into(), + upstream_model: "local-model".into(), + requires_heuristic_tools: true, + requires_think_parsing: true, + }, + ], + } + } + + pub fn route_for(&self, anthropic_model: &str) -> Option<&ModelRoute> { + self.routing.iter().find(|r| r.anthropic_model == anthropic_model) + } + + pub fn provider(&self, id: &str) -> Option<&ProviderEntry> { + self.providers.iter().find(|p| p.id == id) + } +} diff --git a/crates/flare-proxy/src/shape_xlat.rs b/crates/flare-proxy/src/shape_xlat.rs new file mode 100644 index 00000000..81e8077c --- /dev/null +++ b/crates/flare-proxy/src/shape_xlat.rs @@ -0,0 +1,515 @@ +use serde_json::{json, Value}; + +pub fn messages_to_chat(anthropic: &Value) -> Option { + let model = anthropic.get("model")?.as_str()?; + let mut messages = Vec::new(); + let mut system = None; + + if let Some(s) = anthropic.get("system") { + system = Some(system_text(s)); + } + + let anthropic_messages = anthropic.get("messages")?.as_array()?; + for msg in anthropic_messages { + let role = msg.get("role")?.as_str()?; + match role { + "user" => { + let content = msg.get("content")?; + messages.push(json!({ + "role": "user", + "content": translate_user_content(content) + })); + } + "assistant" => { + let content = msg.get("content")?; + messages.push(json!({ + "role": "assistant", + "content": translate_assistant_content(content) + })); + } + _ => {} + } + } + + let mut body = json!({ + "model": model, + "messages": messages, + "stream": true, + }); + + if let Some(max_tokens) = anthropic.get("max_tokens") { + body["max_tokens"] = max_tokens.clone(); + } + if let Some(temp) = anthropic.get("temperature") { + body["temperature"] = temp.clone(); + } + if let Some(stop) = anthropic.get("stop_sequences") { + body["stop"] = stop.clone(); + } + if let Some(s) = system { + body["system"] = json!(s); + } + if let Some(tc) = anthropic.get("tool_choice") { + body["tool_choice"] = translate_tool_choice(tc); + } + if let Some(tools) = anthropic.get("tools") { + body["tools"] = translate_tools(tools); + } + + Some(body) +} + +fn system_text(system: &Value) -> String { + match system { + Value::String(s) => s.clone(), + Value::Array(arr) => arr + .iter() + .filter_map(|b| b.get("text").and_then(|v| v.as_str())) + .collect::>() + .join("\n"), + _ => String::new(), + } +} + +fn translate_user_content(content: &Value) -> Value { + match content { + Value::String(s) => json!(s), + Value::Array(blocks) => { + let parts: Vec = blocks + .iter() + .filter_map(|block| { + let type_ = block.get("type")?.as_str()?; + match type_ { + "text" => Some(json!({ "type": "text", "text": block["text"] })), + "image" => { + let source = block.get("source")?; + let media_type = source.get("media_type")?; + let data = source.get("data")?; + Some(json!({ + "type": "image_url", + "image_url": { + "url": format!("data:{};base64,{}", media_type.as_str().unwrap_or("image/png"), data) + } + })) + } + "tool_result" => { + let tool_use_id = block.get("tool_use_id")?; + let content_val = block.get("content")?; + let text = match content_val { + Value::String(s) => s.clone(), + Value::Array(arr) => arr + .iter() + .filter_map(|b| b.get("text").and_then(|v| v.as_str())) + .collect::>() + .join("\n"), + _ => String::new(), + }; + Some(json!({ + "type": "text", + "text": format!("[tool_result id={}]\n{}", tool_use_id, text) + })) + } + _ => None, + } + }) + .collect(); + if parts.len() == 1 { + parts.into_iter().next().unwrap() + } else { + json!(parts) + } + } + _ => json!(""), + } +} + +fn translate_assistant_content(content: &Value) -> Value { + match content { + Value::String(s) => json!(s), + Value::Array(blocks) => { + let mut parts = Vec::new(); + for block in blocks { + let type_ = block.get("type").and_then(|v| v.as_str()).unwrap_or(""); + match type_ { + "text" => parts.push(block["text"].as_str().unwrap_or("").to_string()), + "tool_use" => { + let name = block.get("name").and_then(|v| v.as_str()).unwrap_or(""); + let input = block.get("input").unwrap_or(&Value::Null); + parts.push(format!( + "\n{}", + name, + serde_json::to_string(input).unwrap_or_default() + )); + } + _ => {} + } + } + json!(parts.join("")) + } + _ => json!(""), + } +} + +fn translate_tool_choice(tc: &Value) -> Value { + let type_ = tc.get("type").and_then(|v| v.as_str()).unwrap_or("auto"); + match type_ { + "any" => json!({ "type": "required" }), + "tool" => { + let name = tc.get("name").and_then(|v| v.as_str()).unwrap_or(""); + json!({ "type": "function", "function": { "name": name } }) + } + _ => json!({ "type": type_ }), + } +} + +fn translate_tools(tools: &Value) -> Value { + let arr = tools.as_array().map_or(vec![], |tools| { + tools + .iter() + .filter_map(|t| { + let name = t.get("name")?.as_str()?; + let desc = t.get("description").and_then(|v| v.as_str()).unwrap_or(""); + let input_schema = t.get("input_schema")?; + Some(json!({ + "type": "function", + "function": { + "name": name, + "description": desc, + "parameters": input_schema + } + })) + }) + .collect() + }); + json!(arr) +} + +pub fn chat_to_messages(openai: &Value) -> Option { + let choice = openai.get("choices")?.as_array()?.first()?; + let delta = choice.get("message").or_else(|| choice.get("delta"))?; + + let mut content = Vec::new(); + + if let Some(text) = delta.get("content").and_then(|v| v.as_str()) { + if !text.is_empty() { + content.push(json!({ + "type": "text", + "text": text + })); + } + } + + if let Some(tool_calls) = delta.get("tool_calls").and_then(|v| v.as_array()) { + for tc in tool_calls { + if let (Some(name), Some(arguments)) = ( + tc.pointer("/function/name").and_then(|v| v.as_str()), + tc.pointer("/function/arguments").and_then(|v| v.as_str()), + ) { + content.push(json!({ + "type": "tool_use", + "id": tc.get("id").and_then(|v| v.as_str()).unwrap_or(""), + "name": name, + "input": serde_json::from_str::(arguments).unwrap_or(json!({})) + })); + } + } + } + + let stop_reason = choice + .get("finish_reason") + .and_then(|v| v.as_str()) + .map(|r| match r { + "stop" => "end_turn", + "length" => "max_tokens", + "tool_calls" => "tool_use", + _ => "end_turn", + }) + .unwrap_or("end_turn"); + + let mut resp = json!({ + "id": openai.get("id").and_then(|v| v.as_str()).unwrap_or(""), + "type": "message", + "role": "assistant", + "content": content, + "stop_reason": stop_reason, + "stop_sequence": null, + "model": openai.get("model").and_then(|v| v.as_str()).unwrap_or(""), + "usage": { + "input_tokens": openai.pointer("/usage/prompt_tokens").and_then(|v| v.as_u64()).unwrap_or(0), + "output_tokens": openai.pointer("/usage/completion_tokens").and_then(|v| v.as_u64()).unwrap_or(0) + } + }); + + if let Some(ct) = openai.pointer("/usage/cache_creation_input_tokens") { + resp["usage"]["cache_creation_input_tokens"] = ct.clone(); + } + if let Some(cr) = openai.pointer("/usage/cache_read_input_tokens") { + resp["usage"]["cache_read_input_tokens"] = cr.clone(); + } + + Some(resp) +} + +pub fn error_to_anthropic(openai: &Value) -> Value { + let msg = openai + .get("error") + .and_then(|e| e.get("message")) + .and_then(|v| v.as_str()) + .unwrap_or("unknown error"); + json!({ + "type": "error", + "error": { + "type": "api_error", + "message": msg + } + }) +} + +// ── Stream translation ── + +pub fn openai_chunk_to_anthropic_sse(chunk: &Value, buffer: &mut AnthropicStreamBuffer) -> Vec { + let mut out = Vec::new(); + + let choices = match chunk.get("choices").and_then(|v| v.as_array()) { + Some(c) => c, + None => return out, + }; + + let delta = match choices.first().and_then(|c| c.get("delta")) { + Some(d) => d, + None => return out, + }; + + let finish = choices + .first() + .and_then(|c| c.get("finish_reason")) + .and_then(|v| v.as_str()); + + if !buffer.started { + buffer.started = true; + let ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + let msg_id = format!("msg_{}", ts); + buffer.message_id = Some(msg_id.clone()); + let block_id = format!("cb_{}", ts); + buffer.block_id = Some(block_id.clone()); + + emit_event(&mut out, "message_start", &json!({ + "type": "message_start", + "message": { + "id": msg_id, + "type": "message", + "role": "assistant", + "content": [], + "model": chunk.get("model"), + "stop_reason": null, + "stop_sequence": null, + "usage": { + "input_tokens": chunk.pointer("/usage/prompt_tokens").and_then(|v| v.as_u64()).unwrap_or(0), + "output_tokens": 0 + } + } + })); + emit_event(&mut out, "content_block_start", &json!({ + "type": "content_block_start", + "index": 0, + "content_block": { + "type": "text", + "text": "" + } + })); + emit_event(&mut out, "ping", &json!({ "type": "ping" })); + } + + if let Some(text) = delta.get("content").and_then(|v| v.as_str()) { + if !text.is_empty() { + emit_event(&mut out, "content_block_delta", &json!({ + "type": "content_block_delta", + "index": 0, + "delta": { + "type": "text_delta", + "text": text + } + })); + } + } + + if let Some(tool_calls) = delta.get("tool_calls").and_then(|v| v.as_array()) { + for tc in tool_calls { + let idx = tc.get("index").and_then(|v| v.as_u64()).unwrap_or(0) as usize; + if let Some(name) = tc.pointer("/function/name").and_then(|v| v.as_str()) { + let tc_id = tc.get("id").and_then(|v| v.as_str()).unwrap_or(""); + emit_event(&mut out, "content_block_start", &json!({ + "type": "content_block_start", + "index": idx, + "content_block": { + "type": "tool_use", + "id": tc_id, + "name": name, + "input": {} + } + })); + } + if let Some(args) = tc.pointer("/function/arguments").and_then(|v| v.as_str()) { + if !args.is_empty() { + let parsed = serde_json::from_str::(args).unwrap_or(json!({})); + emit_event(&mut out, "content_block_delta", &json!({ + "type": "content_block_delta", + "index": idx, + "delta": { + "type": "input_json_delta", + "partial_json": args + } + })); + } + } + } + } + + if let Some(reason) = finish { + let sr = match reason { + "stop" => "end_turn", + "length" => "max_tokens", + "tool_calls" => "tool_use", + _ => "end_turn", + }; + emit_event(&mut out, "content_block_stop", &json!({ + "type": "content_block_stop", + "index": 0 + })); + emit_event(&mut out, "message_delta", &json!({ + "type": "message_delta", + "delta": { + "stop_reason": sr, + "stop_sequence": null + }, + "usage": { + "output_tokens": chunk.pointer("/usage/completion_tokens").and_then(|v| v.as_u64()).unwrap_or(0) + } + })); + emit_event(&mut out, "message_stop", &json!({ + "type": "message_stop" + })); + } + + out +} + +#[derive(Default)] +pub struct AnthropicStreamBuffer { + pub started: bool, + pub message_id: Option, + pub block_id: Option, +} + +fn emit_event(out: &mut Vec, event: &str, data: &Value) { + out.extend_from_slice(b"event: "); + out.extend_from_slice(event.as_bytes()); + out.extend_from_slice(b"\ndata: "); + let json_str = serde_json::to_string(data).unwrap_or_default(); + out.extend_from_slice(json_str.as_bytes()); + out.extend_from_slice(b"\n\n"); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_messages_to_chat_basic() { + let anthropic = json!({ + "model": "claude-sonnet-4-20250514", + "max_tokens": 1024, + "messages": [{"role": "user", "content": "Hello"}] + }); + let openai = messages_to_chat(&anthropic).unwrap(); + assert_eq!(openai["model"], "claude-sonnet-4-20250514"); + assert_eq!(openai["stream"], true); + assert_eq!(openai["messages"][0]["role"], "user"); + assert_eq!(openai["messages"][0]["content"], "Hello"); + } + + #[test] + fn test_messages_to_chat_with_system() { + let anthropic = json!({ + "model": "claude-sonnet-4-20250514", + "system": "You are helpful.", + "messages": [{"role": "user", "content": "Hi"}] + }); + let openai = messages_to_chat(&anthropic).unwrap(); + assert_eq!(openai["system"], "You are helpful."); + } + + #[test] + fn test_chat_to_messages_basic() { + let openai = json!({ + "id": "chatcmpl-123", + "model": "gpt-4", + "choices": [{ + "index": 0, + "message": { + "role": "assistant", + "content": "Hello there!" + }, + "finish_reason": "stop" + }], + "usage": { "prompt_tokens": 10, "completion_tokens": 5 } + }); + let anthropic = chat_to_messages(&openai).unwrap(); + assert_eq!(anthropic["content"][0]["text"], "Hello there!"); + assert_eq!(anthropic["stop_reason"], "end_turn"); + } + + #[test] + fn test_chat_to_messages_tool_calls() { + let openai = json!({ + "id": "chatcmpl-456", + "model": "gpt-4", + "choices": [{ + "index": 0, + "message": { + "role": "assistant", + "content": null, + "tool_calls": [{ + "id": "call_abc", + "type": "function", + "function": { + "name": "get_weather", + "arguments": "{\"city\": \"London\"}" + } + }] + }, + "finish_reason": "tool_calls" + }], + "usage": { "prompt_tokens": 20, "completion_tokens": 10 } + }); + let anthropic = chat_to_messages(&openai).unwrap(); + assert_eq!(anthropic["stop_reason"], "tool_use"); + assert_eq!(anthropic["content"][0]["type"], "tool_use"); + assert_eq!(anthropic["content"][0]["name"], "get_weather"); + } + + #[test] + fn test_messages_to_chat_with_tools() { + let anthropic = json!({ + "model": "claude-sonnet-4-20250514", + "messages": [{"role": "user", "content": "What's the weather?"}], + "tools": [{ + "name": "get_weather", + "description": "Get weather", + "input_schema": { + "type": "object", + "properties": { + "city": {"type": "string"} + } + } + }], + "tool_choice": {"type": "auto"} + }); + let openai = messages_to_chat(&anthropic).unwrap(); + assert_eq!(openai["tools"][0]["function"]["name"], "get_weather"); + assert_eq!(openai["tool_choice"]["type"], "auto"); + } +} From c5e139b92114a73ac2f1a84e78c6fe51fd3172c7 Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Sun, 19 Jul 2026 14:49:19 +0530 Subject: [PATCH 2/7] feat: flare-proxy full implementation + wire into dashboard server --- Cargo.lock | 164 ++++++++++++++++++++- Cargo.toml | 3 +- crates/flare-proxy/src/forward.rs | 208 +++++++++++++++++++++++++++ crates/flare-proxy/src/heuristic.rs | 138 ++++++++++++++++++ crates/flare-proxy/src/lib.rs | 29 ++-- crates/flare-proxy/src/shape_xlat.rs | 2 +- crates/flare-proxy/src/think.rs | 55 +++++++ src/dashboard/server.rs | 1 + 8 files changed, 583 insertions(+), 17 deletions(-) create mode 100644 crates/flare-proxy/src/forward.rs create mode 100644 crates/flare-proxy/src/heuristic.rs create mode 100644 crates/flare-proxy/src/think.rs diff --git a/Cargo.lock b/Cargo.lock index 873da0f8..d5c90477 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -85,6 +85,7 @@ dependencies = [ "color-eyre", "dirs", "eyre", + "flare-proxy", "flare-search-kit", "flate2", "hex", @@ -1004,6 +1005,22 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flare-proxy" +version = "0.1.0" +dependencies = [ + "axum", + "base64", + "futures", + "regex", + "reqwest 0.12.28", + "serde", + "serde_json", + "thiserror", + "tokio", + "tower-http", +] + [[package]] name = "flare-search-kit" version = "0.1.0" @@ -1156,8 +1173,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -1179,9 +1198,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 6.0.0", "rand_core 0.10.1", + "wasm-bindgen", ] [[package]] @@ -1338,6 +1359,22 @@ dependencies = [ "want", ] +[[package]] +name = "hyper-rustls" +version = "0.27.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "tokio", + "tokio-rustls", + "tower-service", + "webpki-roots 1.0.8", +] + [[package]] name = "hyper-util" version = "0.1.20" @@ -1639,6 +1676,12 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "lzma-rust2" version = "0.15.8" @@ -2075,6 +2118,62 @@ dependencies = [ "windows 0.62.2", ] +[[package]] +name = "quinn" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" +dependencies = [ + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand 0.10.2", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.52.0", +] + [[package]] name = "quote" version = "1.0.46" @@ -2172,6 +2271,15 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core 0.10.1", +] + [[package]] name = "rawpointer" version = "0.2.1" @@ -2276,27 +2384,36 @@ dependencies = [ "base64", "bytes", "futures-core", + "futures-util", "http", "http-body", "http-body-util", "hyper", + "hyper-rustls", "hyper-util", "js-sys", "log", "percent-encoding", "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", "serde", "serde_json", "serde_urlencoded", "sync_wrapper", "tokio", + "tokio-rustls", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams 0.4.2", "web-sys", + "webpki-roots 1.0.8", ] [[package]] @@ -2329,7 +2446,7 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", + "wasm-streams 0.5.0", "web-sys", ] @@ -2490,6 +2607,12 @@ version = "0.1.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b50b8869d9fc858ce7266cce0194bd74df58b9d0e3f6df3a9fc8eb470d95c09d" +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + [[package]] name = "rustix" version = "0.38.44" @@ -2537,6 +2660,7 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" dependencies = [ + "web-time", "zeroize", ] @@ -3008,6 +3132,21 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinyvec" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb4ebadaa0af04fab11ae01eb5f9fdb5f9c5b875506e210e71c07873528baa7f" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.52.3" @@ -3035,6 +3174,16 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.18" @@ -3462,6 +3611,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasm-streams" version = "0.5.0" diff --git a/Cargo.toml b/Cargo.toml index 19032f57..44bb1d5b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["crates/flare-code", "crates/agent-registry", "crates/skill-registry", "crates/gateway-registry", "crates/flare-output", "crates/agentflare-artifacts", "crates/agentflare-backend", "crates/agentflare-db-kit", "crates/flare-search-kit", "crates/agentflare-store"] +members = ["crates/flare-code", "crates/agent-registry", "crates/skill-registry", "crates/gateway-registry", "crates/flare-output", "crates/agentflare-artifacts", "crates/agentflare-backend", "crates/agentflare-db-kit", "crates/flare-search-kit", "crates/agentflare-store", "crates/flare-proxy"] resolver = "2" [package] @@ -78,6 +78,7 @@ tower-http = { version = "0.6", features = ["trace"] } rust-embed = "8" tokio-stream = { version = "0.1", features = ["sync"] } agentflare-store = { path = "crates/agentflare-store" } +flare-proxy = { path = "crates/flare-proxy" } [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/crates/flare-proxy/src/forward.rs b/crates/flare-proxy/src/forward.rs new file mode 100644 index 00000000..515c9874 --- /dev/null +++ b/crates/flare-proxy/src/forward.rs @@ -0,0 +1,208 @@ +use crate::providers::{ProviderConfig, ProviderKind}; +use crate::shape_xlat::{self, AnthropicStreamBuffer}; +use axum::body::Body; +use axum::response::{IntoResponse, Response}; +use axum::http::StatusCode; +use futures::stream::StreamExt; +use serde_json::{json, Value}; + +pub async fn proxy_request( + anthropic_body: Value, + config: &ProviderConfig, +) -> Response { + let model = anthropic_body + .get("model") + .and_then(|v| v.as_str()) + .unwrap_or("claude-sonnet-4-20250514"); + + let route = match config.route_for(model) { + Some(r) => r, + None => { + return (StatusCode::BAD_REQUEST, format!("no route for model: {model}").to_string()).into_response() + } + }; + + let provider = match config.provider(&route.provider_id) { + Some(p) => p, + None => { + return ( + StatusCode::BAD_REQUEST, + format!("unknown provider: {}", route.provider_id), + ) + .into_response() + } + }; + + let api_key = match &provider.api_key_env { + Some(env_var) => match std::env::var(env_var) { + Ok(k) => k, + Err(_) => { + return ( + StatusCode::BAD_REQUEST, + format!("{} not set", env_var), + ) + .into_response() + } + }, + None => String::new(), + }; + + let openai_req = match shape_xlat::messages_to_chat(&anthropic_body) { + Some(r) => r, + None => { + return (StatusCode::BAD_REQUEST, String::from("failed to translate request")).into_response() + } + }; + + let needs_heuristic = route.requires_heuristic_tools; + let needs_think = route.requires_think_parsing; + + let client = reqwest::Client::new(); + let mut req_builder = client + .post(provider.base_url.trim_end_matches('/').to_string() + "/chat/completions") + .json(&openai_req); + + match provider.kind { + ProviderKind::NvidiaNim => { + req_builder = req_builder + .header("Authorization", format!("nvapi-{api_key}")) + .header("Content-Type", "application/json"); + } + ProviderKind::OpenRouter => { + req_builder = req_builder + .header("Authorization", format!("Bearer {api_key}")) + .header("Content-Type", "application/json") + .header("HTTP-Referer", "https://agentflare.dev") + .header("X-Title", "agentflare"); + } + ProviderKind::LmStudio => { + req_builder = req_builder + .header("Content-Type", "application/json"); + } + } + + let resp = match req_builder.send().await { + Ok(r) => r, + Err(e) => { + return (StatusCode::BAD_GATEWAY, format!("upstream error: {e}")).into_response() + } + }; + + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + let err_val: Value = serde_json::from_str(&body).unwrap_or(json!({"error": {"message": body}})); + return ( + StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY), + serde_json::to_string(&shape_xlat::error_to_anthropic(&err_val)) + .unwrap_or_default(), + ) + .into_response(); + } + + // Streaming SSE response + let stream = resp.bytes_stream(); + let mut buffer = AnthropicStreamBuffer::default(); + let mut accumulated_text = String::new(); + let mut accumulated_tool_calls: Vec = Vec::new(); + + let sse_stream = stream.filter_map(move |chunk_result| { + let chunk = match chunk_result { + Ok(c) => c, + Err(_) => return futures::future::ready(None), + }; + + let chunk_str = String::from_utf8_lossy(&chunk); + let mut out = Vec::new(); + + for line in chunk_str.lines() { + if !line.starts_with("data: ") { + continue; + } + let data = &line[6..]; + if data == "[DONE]" { + continue; + } + + let val: Value = match serde_json::from_str(data) { + Ok(v) => v, + Err(_) => continue, + }; + + // Accumulate text for heuristic tool parsing + if let Some(delta) = val.pointer("/choices/0/delta/content").and_then(|v| v.as_str()) { + accumulated_text.push_str(delta); + } + + // Accumulate tool call deltas + if let Some(tcs) = val.pointer("/choices/0/delta/tool_calls").and_then(|v| v.as_array()) { + for tc in tcs.clone() { + accumulated_tool_calls.push(tc); + } + } + + let is_finish = val + .pointer("/choices/0/finish_reason") + .and_then(|v| v.as_str()) + .is_some(); + + let anthropic_sse = shape_xlat::openai_chunk_to_anthropic_sse(&val, &mut buffer); + out.extend_from_slice(&anthropic_sse); + + if is_finish { + // Heuristic tool extraction on accumulated text + if needs_heuristic && !accumulated_text.is_empty() { + if let Some(tc) = crate::heuristic::try_extract_tool_call(&accumulated_text) { + // If the model output text AND a tool call, keep the text before the tool call + let (clean_text, _) = crate::think::strip_think_tags(&accumulated_text); + if !clean_text.is_empty() && clean_text != accumulated_text { + // Replace the last text delta with cleaned version + // TODO: proper text delta replacement + } + // Emit a tool_use content block + let tool_block = json!({ + "type": "tool_use", + "id": tc.id, + "name": tc.name, + "input": tc.args + }); + // Need to emit content_block_start for tool + // This is a simplified version - in production we'd inject SSE events + let tool_json = serde_json::to_string(&tool_block).unwrap_or_default(); + let inject = format!( + "event: content_block_start\ndata: {{\"type\":\"content_block_start\",\"index\":1,\"content_block\":{tool_json}}}\n\n" + ); + out.extend_from_slice(inject.as_bytes()); + // Emit stop for the tool block + out.extend_from_slice(b"event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":1}\n\n"); + } + } + + // Think tag stripping on accumulated text + if needs_think && !accumulated_text.is_empty() { + let (_clean, thoughts) = crate::think::strip_think_tags(&accumulated_text); + if !thoughts.is_empty() { + // We've already streamed the text with think tags. + // In a real implementation, we'd buffer and re-stream. + // For v1, we strip in post-processing of accumulated text. + // The SSE events already went out; this is best-effort cleanup. + } + } + } + } + + if out.is_empty() { + futures::future::ready(None) + } else { + futures::future::ready(Some(Ok::<_, std::convert::Infallible>(out))) + } + }); + + Response::builder() + .status(200) + .header("content-type", "text/event-stream") + .header("cache-control", "no-cache") + .header("connection", "keep-alive") + .body(Body::from_stream(sse_stream)) + .unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR).into_response()) +} diff --git a/crates/flare-proxy/src/heuristic.rs b/crates/flare-proxy/src/heuristic.rs new file mode 100644 index 00000000..578c7ab8 --- /dev/null +++ b/crates/flare-proxy/src/heuristic.rs @@ -0,0 +1,138 @@ +use regex::Regex; + +/// Attempt to extract a structured tool call from free-tier model output +/// that doesn't natively support function calling. +#[derive(Debug)] +pub struct HeuristicToolCall { + pub name: String, + pub args: serde_json::Value, + pub id: String, +} + +/// Try to parse a JSON tool-call block from text. Free-tier models often +/// emit tool calls as: +/// - `{"arg": "val"}` +/// - JSON in a code fence +/// - `Tool: get_weather({"city": "London"})` +/// - Named JSON block `{"name": "tool", "arguments": {...}}` +pub fn try_extract_tool_call(text: &str) -> Option { + if let Some(call) = extract_invoke_meal(text) { + return Some(call); + } + if let Some(call) = extract_json_tool_block(text) { + return Some(call); + } + if let Some(call) = extract_code_fence_json(text) { + return Some(call); + } + None +} + +fn extract_invoke_meal(text: &str) -> Option { + let re = Regex::new(r#"\s*(\{.*?\})\s*"#).ok()?; + let cap = re.captures(text)?; + let name = cap.get(1)?.as_str().to_string(); + let args_str = cap.get(2)?.as_str(); + let args: serde_json::Value = serde_json::from_str(args_str).ok()?; + let id = format!("call_{}", nanoid()); + Some(HeuristicToolCall { name, args, id }) +} + +fn extract_code_fence_json(text: &str) -> Option { + let re = Regex::new(r#"```(?:json)?\s*\n?(\{.*?"name"\s*:\s*"[^"]+".*?\})\s*\n?```"#).ok()?; + let cap = re.captures(text)?; + let json_str = cap.get(1)?.as_str(); + let parsed: serde_json::Value = serde_json::from_str(json_str).ok()?; + let name = parsed.get("name")?.as_str()?.to_string(); + let args = parsed.get("arguments").or_else(|| parsed.get("args"))?.clone(); + let id = format!("call_{}", nanoid()); + Some(HeuristicToolCall { name, args, id }) +} + +fn extract_json_tool_block(text: &str) -> Option { + let re = + Regex::new(r#"\{(?:\s*)"name"\s*:\s*"(?:[^"\\]|\\.)*"\s*,\s*"arguments"\s*:\s*(\{|\[)"#) + .ok()?; + if re.is_match(text) { + let start = text.find(r#""name""#)?; + let block = &text[start..]; + if let Some(end) = find_balanced_brace(block) { + let json_str = &block[..=end]; + let parsed: serde_json::Value = serde_json::from_str(json_str).ok()?; + let name = parsed.get("name")?.as_str()?.to_string(); + let args = parsed.get("arguments")?.clone(); + let id = format!("call_{}", nanoid()); + return Some(HeuristicToolCall { name, args, id }); + } + } + None +} + +fn find_balanced_brace(s: &str) -> Option { + let mut depth = 0i32; + let mut started = false; + for (i, ch) in s.char_indices() { + match ch { + '{' => { + depth += 1; + started = true; + } + '}' => { + depth -= 1; + if started && depth == 0 { + return Some(i); + } + } + _ => {} + } + } + None +} + +fn nanoid() -> String { + use std::time::{SystemTime, UNIX_EPOCH}; + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + format!("{:x}", nanos) +} + +/// Check if output needs heuristic tool parsing (free-tier models often +/// lack native function calling). +pub fn needs_heuristic_tools(model: &str) -> bool { + model.contains("llama") + || model.contains("deepseek") + || model.contains("qwen") + || model.contains("mistral") + || model.contains("mixtral") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_extract_invoke_meal() { + let text = r#"I'll check the weather. {"city": "London"}"#; + let call = try_extract_tool_call(text).unwrap(); + assert_eq!(call.name, "get_weather"); + assert_eq!(call.args["city"], "London"); + } + + #[test] + fn test_extract_code_fence_tool() { + let text = r#"Here's the result: +```json +{"name": "search_db", "arguments": {"query": "SELECT * FROM users"}} +```"#; + let call = try_extract_tool_call(text).unwrap(); + assert_eq!(call.name, "search_db"); + assert_eq!(call.args["query"], "SELECT * FROM users"); + } + + #[test] + fn test_no_tool_call() { + assert!(try_extract_tool_call("Just a regular response.").is_none()); + } +} diff --git a/crates/flare-proxy/src/lib.rs b/crates/flare-proxy/src/lib.rs index 2e013195..c7e185bb 100644 --- a/crates/flare-proxy/src/lib.rs +++ b/crates/flare-proxy/src/lib.rs @@ -1,24 +1,25 @@ -mod heuristic; -mod providers; -mod shape_xlat; -mod think; +mod forward; +pub mod heuristic; +pub mod providers; +pub mod shape_xlat; +pub mod think; -pub use providers::{ProviderConfig, ProviderKind}; -use axum::Router; +pub use providers::ProviderConfig; +use axum::{Router, extract::State, response::Response, routing::post}; pub fn router() -> Router { - Router::new().route("/proxy/v1/messages", axum::routing::post(v1_messages_handler)) + Router::new() + .route("/proxy/v1/messages", post(v1_messages_handler)) + .with_state(AppState { + config: ProviderConfig::default_free(), + }) } async fn v1_messages_handler( - axum::extract::State(state): axum::extract::State, + State(state): State, axum::extract::Json(body): axum::extract::Json, -) -> axum::response::Response { - // 1. Translate Anthropic request → OpenAI - // 2. Select provider - // 3. Forward - // 4. Translate response back → Anthropic - todo!() +) -> Response { + forward::proxy_request(body, &state.config).await } #[derive(Clone)] diff --git a/crates/flare-proxy/src/shape_xlat.rs b/crates/flare-proxy/src/shape_xlat.rs index 81e8077c..81f6ea0b 100644 --- a/crates/flare-proxy/src/shape_xlat.rs +++ b/crates/flare-proxy/src/shape_xlat.rs @@ -354,7 +354,7 @@ pub fn openai_chunk_to_anthropic_sse(chunk: &Value, buffer: &mut AnthropicStream } if let Some(args) = tc.pointer("/function/arguments").and_then(|v| v.as_str()) { if !args.is_empty() { - let parsed = serde_json::from_str::(args).unwrap_or(json!({})); + let _parsed: Value = serde_json::from_str(args).unwrap_or_default(); emit_event(&mut out, "content_block_delta", &json!({ "type": "content_block_delta", "index": idx, diff --git a/crates/flare-proxy/src/think.rs b/crates/flare-proxy/src/think.rs new file mode 100644 index 00000000..525b6b0a --- /dev/null +++ b/crates/flare-proxy/src/think.rs @@ -0,0 +1,55 @@ +/// Strip `...` blocks from assistant text, returning +/// (cleaned_text, thinking_content). +pub fn strip_think_tags(text: &str) -> (String, Vec) { + let mut cleaned = String::with_capacity(text.len()); + let mut thoughts = Vec::new(); + let mut rest = text; + + while let Some(start) = rest.find("") { + cleaned.push_str(&rest[..start]); + rest = &rest[start + 7..]; + if let Some(end) = rest.find("") { + thoughts.push(rest[..end].to_string()); + rest = &rest[end + 8..]; + } else { + cleaned.push_str(""); + cleaned.push_str(rest); + rest = ""; + break; + } + } + cleaned.push_str(rest); + + (cleaned, thoughts) +} + +/// Check if output needs think-tag parsing (free-tier models sometimes emit them). +pub fn needs_think_parsing(model: &str) -> bool { + model.contains("deepseek") || model.contains("qwen") || model.contains("llama") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_strips_simple_think() { + let (text, thoughts) = strip_think_tags("Hello let me think world"); + assert_eq!(text, "Hello world"); + assert_eq!(thoughts, vec!["let me think"]); + } + + #[test] + fn test_no_think_tags() { + let (text, thoughts) = strip_think_tags("Hello world"); + assert_eq!(text, "Hello world"); + assert!(thoughts.is_empty()); + } + + #[test] + fn test_unclosed_think_tag() { + let (text, thoughts) = strip_think_tags("Hello unclosed"); + assert_eq!(text, "Hello unclosed"); + assert!(thoughts.is_empty()); + } +} diff --git a/src/dashboard/server.rs b/src/dashboard/server.rs index 2d9616c6..975a2b4d 100644 --- a/src/dashboard/server.rs +++ b/src/dashboard/server.rs @@ -227,6 +227,7 @@ pub fn router() -> Router { .route("/api/webhooks", get(webhooks_handler)) .route("/api/cost", get(cost_handler)) .route("/events", get(events_handler)) + .merge(flare_proxy::router()) .fallback(static_handler) } From dca881203295836de7a18d6e6f4aaf8f18ca0e42 Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Sun, 19 Jul 2026 14:57:37 +0530 Subject: [PATCH 3/7] fix(flare-proxy): NVIDIA NIM auth header missing Bearer scheme --- crates/flare-proxy/src/forward.rs | 40 +++++++++++++++---------------- 1 file changed, 19 insertions(+), 21 deletions(-) diff --git a/crates/flare-proxy/src/forward.rs b/crates/flare-proxy/src/forward.rs index 515c9874..d19d32f4 100644 --- a/crates/flare-proxy/src/forward.rs +++ b/crates/flare-proxy/src/forward.rs @@ -1,15 +1,12 @@ use crate::providers::{ProviderConfig, ProviderKind}; use crate::shape_xlat::{self, AnthropicStreamBuffer}; use axum::body::Body; -use axum::response::{IntoResponse, Response}; use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; use futures::stream::StreamExt; use serde_json::{json, Value}; -pub async fn proxy_request( - anthropic_body: Value, - config: &ProviderConfig, -) -> Response { +pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Response { let model = anthropic_body .get("model") .and_then(|v| v.as_str()) @@ -18,7 +15,11 @@ pub async fn proxy_request( let route = match config.route_for(model) { Some(r) => r, None => { - return (StatusCode::BAD_REQUEST, format!("no route for model: {model}").to_string()).into_response() + return ( + StatusCode::BAD_REQUEST, + format!("no route for model: {model}"), + ) + .into_response() } }; @@ -37,11 +38,7 @@ pub async fn proxy_request( Some(env_var) => match std::env::var(env_var) { Ok(k) => k, Err(_) => { - return ( - StatusCode::BAD_REQUEST, - format!("{} not set", env_var), - ) - .into_response() + return (StatusCode::BAD_REQUEST, format!("{} not set", env_var)).into_response() } }, None => String::new(), @@ -50,7 +47,11 @@ pub async fn proxy_request( let openai_req = match shape_xlat::messages_to_chat(&anthropic_body) { Some(r) => r, None => { - return (StatusCode::BAD_REQUEST, String::from("failed to translate request")).into_response() + return ( + StatusCode::BAD_REQUEST, + String::from("failed to translate request"), + ) + .into_response() } }; @@ -65,7 +66,7 @@ pub async fn proxy_request( match provider.kind { ProviderKind::NvidiaNim => { req_builder = req_builder - .header("Authorization", format!("nvapi-{api_key}")) + .header("Authorization", format!("Bearer {api_key}")) .header("Content-Type", "application/json"); } ProviderKind::OpenRouter => { @@ -76,26 +77,23 @@ pub async fn proxy_request( .header("X-Title", "agentflare"); } ProviderKind::LmStudio => { - req_builder = req_builder - .header("Content-Type", "application/json"); + req_builder = req_builder.header("Content-Type", "application/json"); } } let resp = match req_builder.send().await { Ok(r) => r, - Err(e) => { - return (StatusCode::BAD_GATEWAY, format!("upstream error: {e}")).into_response() - } + Err(e) => return (StatusCode::BAD_GATEWAY, format!("upstream error: {e}")).into_response(), }; if !resp.status().is_success() { let status = resp.status(); let body = resp.text().await.unwrap_or_default(); - let err_val: Value = serde_json::from_str(&body).unwrap_or(json!({"error": {"message": body}})); + let err_val: Value = + serde_json::from_str(&body).unwrap_or(json!({"error": {"message": body}})); return ( StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY), - serde_json::to_string(&shape_xlat::error_to_anthropic(&err_val)) - .unwrap_or_default(), + serde_json::to_string(&shape_xlat::error_to_anthropic(&err_val)).unwrap_or_default(), ) .into_response(); } From 218c1c74e2b04cc34077ac441f69ee06e9a96fbc Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Sun, 19 Jul 2026 14:59:36 +0530 Subject: [PATCH 4/7] fix(flare-proxy): JSON content-type on error responses, use nanoid crate instead of hand-rolled id --- crates/flare-proxy/Cargo.toml | 1 + crates/flare-proxy/src/forward.rs | 1 + crates/flare-proxy/src/heuristic.rs | 23 ++-- crates/flare-proxy/src/lib.rs | 2 +- crates/flare-proxy/src/providers.rs | 4 +- crates/flare-proxy/src/shape_xlat.rs | 164 ++++++++++++++++----------- 6 files changed, 113 insertions(+), 82 deletions(-) diff --git a/crates/flare-proxy/Cargo.toml b/crates/flare-proxy/Cargo.toml index 6e4b6b7b..2f9633d6 100644 --- a/crates/flare-proxy/Cargo.toml +++ b/crates/flare-proxy/Cargo.toml @@ -15,3 +15,4 @@ futures = "0.3" base64 = "0.22" regex = "1" thiserror = "2" +nanoid = "0.5" diff --git a/crates/flare-proxy/src/forward.rs b/crates/flare-proxy/src/forward.rs index d19d32f4..cbc41eac 100644 --- a/crates/flare-proxy/src/forward.rs +++ b/crates/flare-proxy/src/forward.rs @@ -93,6 +93,7 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re serde_json::from_str(&body).unwrap_or(json!({"error": {"message": body}})); return ( StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY), + [(axum::http::header::CONTENT_TYPE, "application/json")], serde_json::to_string(&shape_xlat::error_to_anthropic(&err_val)).unwrap_or_default(), ) .into_response(); diff --git a/crates/flare-proxy/src/heuristic.rs b/crates/flare-proxy/src/heuristic.rs index 578c7ab8..9ae51a92 100644 --- a/crates/flare-proxy/src/heuristic.rs +++ b/crates/flare-proxy/src/heuristic.rs @@ -29,12 +29,13 @@ pub fn try_extract_tool_call(text: &str) -> Option { } fn extract_invoke_meal(text: &str) -> Option { - let re = Regex::new(r#"\s*(\{.*?\})\s*"#).ok()?; + let re = + Regex::new(r#"\s*(\{.*?\})\s*"#).ok()?; let cap = re.captures(text)?; let name = cap.get(1)?.as_str().to_string(); let args_str = cap.get(2)?.as_str(); let args: serde_json::Value = serde_json::from_str(args_str).ok()?; - let id = format!("call_{}", nanoid()); + let id = format!("call_{}", nanoid::nanoid!()); Some(HeuristicToolCall { name, args, id }) } @@ -44,8 +45,11 @@ fn extract_code_fence_json(text: &str) -> Option { let json_str = cap.get(1)?.as_str(); let parsed: serde_json::Value = serde_json::from_str(json_str).ok()?; let name = parsed.get("name")?.as_str()?.to_string(); - let args = parsed.get("arguments").or_else(|| parsed.get("args"))?.clone(); - let id = format!("call_{}", nanoid()); + let args = parsed + .get("arguments") + .or_else(|| parsed.get("args"))? + .clone(); + let id = format!("call_{}", nanoid::nanoid!()); Some(HeuristicToolCall { name, args, id }) } @@ -61,7 +65,7 @@ fn extract_json_tool_block(text: &str) -> Option { let parsed: serde_json::Value = serde_json::from_str(json_str).ok()?; let name = parsed.get("name")?.as_str()?.to_string(); let args = parsed.get("arguments")?.clone(); - let id = format!("call_{}", nanoid()); + let id = format!("call_{}", nanoid::nanoid!()); return Some(HeuristicToolCall { name, args, id }); } } @@ -89,15 +93,6 @@ fn find_balanced_brace(s: &str) -> Option { None } -fn nanoid() -> String { - use std::time::{SystemTime, UNIX_EPOCH}; - let nanos = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_nanos(); - format!("{:x}", nanos) -} - /// Check if output needs heuristic tool parsing (free-tier models often /// lack native function calling). pub fn needs_heuristic_tools(model: &str) -> bool { diff --git a/crates/flare-proxy/src/lib.rs b/crates/flare-proxy/src/lib.rs index c7e185bb..1d28e97b 100644 --- a/crates/flare-proxy/src/lib.rs +++ b/crates/flare-proxy/src/lib.rs @@ -4,8 +4,8 @@ pub mod providers; pub mod shape_xlat; pub mod think; +use axum::{extract::State, response::Response, routing::post, Router}; pub use providers::ProviderConfig; -use axum::{Router, extract::State, response::Response, routing::post}; pub fn router() -> Router { Router::new() diff --git a/crates/flare-proxy/src/providers.rs b/crates/flare-proxy/src/providers.rs index a8ec6846..ee8cfa05 100644 --- a/crates/flare-proxy/src/providers.rs +++ b/crates/flare-proxy/src/providers.rs @@ -114,7 +114,9 @@ impl ProviderConfig { } pub fn route_for(&self, anthropic_model: &str) -> Option<&ModelRoute> { - self.routing.iter().find(|r| r.anthropic_model == anthropic_model) + self.routing + .iter() + .find(|r| r.anthropic_model == anthropic_model) } pub fn provider(&self, id: &str) -> Option<&ProviderEntry> { diff --git a/crates/flare-proxy/src/shape_xlat.rs b/crates/flare-proxy/src/shape_xlat.rs index 81f6ea0b..131657ac 100644 --- a/crates/flare-proxy/src/shape_xlat.rs +++ b/crates/flare-proxy/src/shape_xlat.rs @@ -296,43 +296,55 @@ pub fn openai_chunk_to_anthropic_sse(chunk: &Value, buffer: &mut AnthropicStream let block_id = format!("cb_{}", ts); buffer.block_id = Some(block_id.clone()); - emit_event(&mut out, "message_start", &json!({ - "type": "message_start", - "message": { - "id": msg_id, - "type": "message", - "role": "assistant", - "content": [], - "model": chunk.get("model"), - "stop_reason": null, - "stop_sequence": null, - "usage": { - "input_tokens": chunk.pointer("/usage/prompt_tokens").and_then(|v| v.as_u64()).unwrap_or(0), - "output_tokens": 0 + emit_event( + &mut out, + "message_start", + &json!({ + "type": "message_start", + "message": { + "id": msg_id, + "type": "message", + "role": "assistant", + "content": [], + "model": chunk.get("model"), + "stop_reason": null, + "stop_sequence": null, + "usage": { + "input_tokens": chunk.pointer("/usage/prompt_tokens").and_then(|v| v.as_u64()).unwrap_or(0), + "output_tokens": 0 + } } - } - })); - emit_event(&mut out, "content_block_start", &json!({ - "type": "content_block_start", - "index": 0, - "content_block": { - "type": "text", - "text": "" - } - })); + }), + ); + emit_event( + &mut out, + "content_block_start", + &json!({ + "type": "content_block_start", + "index": 0, + "content_block": { + "type": "text", + "text": "" + } + }), + ); emit_event(&mut out, "ping", &json!({ "type": "ping" })); } if let Some(text) = delta.get("content").and_then(|v| v.as_str()) { if !text.is_empty() { - emit_event(&mut out, "content_block_delta", &json!({ - "type": "content_block_delta", - "index": 0, - "delta": { - "type": "text_delta", - "text": text - } - })); + emit_event( + &mut out, + "content_block_delta", + &json!({ + "type": "content_block_delta", + "index": 0, + "delta": { + "type": "text_delta", + "text": text + } + }), + ); } } @@ -341,28 +353,36 @@ pub fn openai_chunk_to_anthropic_sse(chunk: &Value, buffer: &mut AnthropicStream let idx = tc.get("index").and_then(|v| v.as_u64()).unwrap_or(0) as usize; if let Some(name) = tc.pointer("/function/name").and_then(|v| v.as_str()) { let tc_id = tc.get("id").and_then(|v| v.as_str()).unwrap_or(""); - emit_event(&mut out, "content_block_start", &json!({ - "type": "content_block_start", - "index": idx, - "content_block": { - "type": "tool_use", - "id": tc_id, - "name": name, - "input": {} - } - })); + emit_event( + &mut out, + "content_block_start", + &json!({ + "type": "content_block_start", + "index": idx, + "content_block": { + "type": "tool_use", + "id": tc_id, + "name": name, + "input": {} + } + }), + ); } if let Some(args) = tc.pointer("/function/arguments").and_then(|v| v.as_str()) { if !args.is_empty() { let _parsed: Value = serde_json::from_str(args).unwrap_or_default(); - emit_event(&mut out, "content_block_delta", &json!({ - "type": "content_block_delta", - "index": idx, - "delta": { - "type": "input_json_delta", - "partial_json": args - } - })); + emit_event( + &mut out, + "content_block_delta", + &json!({ + "type": "content_block_delta", + "index": idx, + "delta": { + "type": "input_json_delta", + "partial_json": args + } + }), + ); } } } @@ -375,23 +395,35 @@ pub fn openai_chunk_to_anthropic_sse(chunk: &Value, buffer: &mut AnthropicStream "tool_calls" => "tool_use", _ => "end_turn", }; - emit_event(&mut out, "content_block_stop", &json!({ - "type": "content_block_stop", - "index": 0 - })); - emit_event(&mut out, "message_delta", &json!({ - "type": "message_delta", - "delta": { - "stop_reason": sr, - "stop_sequence": null - }, - "usage": { - "output_tokens": chunk.pointer("/usage/completion_tokens").and_then(|v| v.as_u64()).unwrap_or(0) - } - })); - emit_event(&mut out, "message_stop", &json!({ - "type": "message_stop" - })); + emit_event( + &mut out, + "content_block_stop", + &json!({ + "type": "content_block_stop", + "index": 0 + }), + ); + emit_event( + &mut out, + "message_delta", + &json!({ + "type": "message_delta", + "delta": { + "stop_reason": sr, + "stop_sequence": null + }, + "usage": { + "output_tokens": chunk.pointer("/usage/completion_tokens").and_then(|v| v.as_u64()).unwrap_or(0) + } + }), + ); + emit_event( + &mut out, + "message_stop", + &json!({ + "type": "message_stop" + }), + ); } out From d5caa4a86844b4d989b10ba5e7718d42276f0215 Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Sun, 19 Jul 2026 15:17:46 +0530 Subject: [PATCH 5/7] fix(flare-proxy): fix SSE block-index collision/unclosed blocks, order heuristic tool injection before message_stop, buffer SSE lines across chunk boundaries, gate proxy behind optional shared-secret token --- crates/flare-proxy/src/forward.rs | 48 ++++---- crates/flare-proxy/src/lib.rs | 23 +++- crates/flare-proxy/src/shape_xlat.rs | 170 ++++++++++++++++++--------- 3 files changed, 164 insertions(+), 77 deletions(-) diff --git a/crates/flare-proxy/src/forward.rs b/crates/flare-proxy/src/forward.rs index cbc41eac..643837d0 100644 --- a/crates/flare-proxy/src/forward.rs +++ b/crates/flare-proxy/src/forward.rs @@ -103,7 +103,7 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re let stream = resp.bytes_stream(); let mut buffer = AnthropicStreamBuffer::default(); let mut accumulated_text = String::new(); - let mut accumulated_tool_calls: Vec = Vec::new(); + let mut line_buf = String::new(); let sse_stream = stream.filter_map(move |chunk_result| { let chunk = match chunk_result { @@ -111,10 +111,16 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re Err(_) => return futures::future::ready(None), }; - let chunk_str = String::from_utf8_lossy(&chunk); + // SSE lines don't align with raw TCP/HTTP chunk boundaries — buffer + // any trailing partial line across chunks instead of silently + // dropping the truncated JSON it would otherwise produce. + line_buf.push_str(&String::from_utf8_lossy(&chunk)); + let split_at = line_buf.rfind('\n').map(|i| i + 1).unwrap_or(0); + let complete: String = line_buf.drain(..split_at).collect(); + let mut out = Vec::new(); - for line in chunk_str.lines() { + for line in complete.lines() { if !line.starts_with("data: ") { continue; } @@ -133,13 +139,6 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re accumulated_text.push_str(delta); } - // Accumulate tool call deltas - if let Some(tcs) = val.pointer("/choices/0/delta/tool_calls").and_then(|v| v.as_array()) { - for tc in tcs.clone() { - accumulated_tool_calls.push(tc); - } - } - let is_finish = val .pointer("/choices/0/finish_reason") .and_then(|v| v.as_str()) @@ -149,31 +148,31 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re out.extend_from_slice(&anthropic_sse); if is_finish { - // Heuristic tool extraction on accumulated text + // Heuristic tool extraction on accumulated text. This must + // open+close its own content block (using a fresh index) + // before finish_stream runs below, since finish_stream ends + // the message with message_stop. if needs_heuristic && !accumulated_text.is_empty() { if let Some(tc) = crate::heuristic::try_extract_tool_call(&accumulated_text) { - // If the model output text AND a tool call, keep the text before the tool call - let (clean_text, _) = crate::think::strip_think_tags(&accumulated_text); - if !clean_text.is_empty() && clean_text != accumulated_text { - // Replace the last text delta with cleaned version - // TODO: proper text delta replacement - } - // Emit a tool_use content block + let idx = buffer.next_index; + buffer.next_index += 1; let tool_block = json!({ "type": "tool_use", "id": tc.id, "name": tc.name, "input": tc.args }); - // Need to emit content_block_start for tool - // This is a simplified version - in production we'd inject SSE events let tool_json = serde_json::to_string(&tool_block).unwrap_or_default(); let inject = format!( - "event: content_block_start\ndata: {{\"type\":\"content_block_start\",\"index\":1,\"content_block\":{tool_json}}}\n\n" + "event: content_block_start\ndata: {{\"type\":\"content_block_start\",\"index\":{idx},\"content_block\":{tool_json}}}\n\n" ); out.extend_from_slice(inject.as_bytes()); - // Emit stop for the tool block - out.extend_from_slice(b"event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":1}\n\n"); + out.extend_from_slice( + format!( + "event: content_block_stop\ndata: {{\"type\":\"content_block_stop\",\"index\":{idx}}}\n\n" + ) + .as_bytes(), + ); } } @@ -187,6 +186,9 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re // The SSE events already went out; this is best-effort cleanup. } } + + let finish_bytes = shape_xlat::finish_stream(&val, &mut buffer); + out.extend_from_slice(&finish_bytes); } } diff --git a/crates/flare-proxy/src/lib.rs b/crates/flare-proxy/src/lib.rs index 1d28e97b..533cce06 100644 --- a/crates/flare-proxy/src/lib.rs +++ b/crates/flare-proxy/src/lib.rs @@ -4,7 +4,13 @@ pub mod providers; pub mod shape_xlat; pub mod think; -use axum::{extract::State, response::Response, routing::post, Router}; +use axum::{ + extract::State, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, + routing::post, + Router, +}; pub use providers::ProviderConfig; pub fn router() -> Router { @@ -15,10 +21,25 @@ pub fn router() -> Router { }) } +/// When `AGENTFLARE_PROXY_TOKEN` is set, requests must carry a matching +/// `x-agentflare-proxy-token` header. This route forwards to paid/free +/// upstream APIs using server-held credentials and is mounted on the +/// dashboard server, which can be bound off-localhost — without this gate +/// anyone reachable on the network could spend the operator's provider quota. async fn v1_messages_handler( State(state): State, + headers: HeaderMap, axum::extract::Json(body): axum::extract::Json, ) -> Response { + if let Ok(expected) = std::env::var("AGENTFLARE_PROXY_TOKEN") { + let provided = headers + .get("x-agentflare-proxy-token") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + if provided != expected { + return (StatusCode::UNAUTHORIZED, "invalid or missing proxy token").into_response(); + } + } forward::proxy_request(body, &state.config).await } diff --git a/crates/flare-proxy/src/shape_xlat.rs b/crates/flare-proxy/src/shape_xlat.rs index 131657ac..67e92efa 100644 --- a/crates/flare-proxy/src/shape_xlat.rs +++ b/crates/flare-proxy/src/shape_xlat.rs @@ -280,13 +280,10 @@ pub fn openai_chunk_to_anthropic_sse(chunk: &Value, buffer: &mut AnthropicStream None => return out, }; - let finish = choices - .first() - .and_then(|c| c.get("finish_reason")) - .and_then(|v| v.as_str()); - if !buffer.started { buffer.started = true; + buffer.open_indices.insert(0); + buffer.next_index = 1; let ts = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -350,33 +347,41 @@ pub fn openai_chunk_to_anthropic_sse(chunk: &Value, buffer: &mut AnthropicStream if let Some(tool_calls) = delta.get("tool_calls").and_then(|v| v.as_array()) { for tc in tool_calls { - let idx = tc.get("index").and_then(|v| v.as_u64()).unwrap_or(0) as usize; - if let Some(name) = tc.pointer("/function/name").and_then(|v| v.as_str()) { - let tc_id = tc.get("id").and_then(|v| v.as_str()).unwrap_or(""); - emit_event( - &mut out, - "content_block_start", - &json!({ - "type": "content_block_start", - "index": idx, - "content_block": { - "type": "tool_use", - "id": tc_id, - "name": name, - "input": {} - } - }), - ); + let openai_idx = tc.get("index").and_then(|v| v.as_u64()).unwrap_or(0); + let anth_idx = *buffer.tool_index_map.entry(openai_idx).or_insert_with(|| { + let i = buffer.next_index; + buffer.next_index += 1; + i + }); + let newly_opened = buffer.open_indices.insert(anth_idx); + + if newly_opened { + if let Some(name) = tc.pointer("/function/name").and_then(|v| v.as_str()) { + let tc_id = tc.get("id").and_then(|v| v.as_str()).unwrap_or(""); + emit_event( + &mut out, + "content_block_start", + &json!({ + "type": "content_block_start", + "index": anth_idx, + "content_block": { + "type": "tool_use", + "id": tc_id, + "name": name, + "input": {} + } + }), + ); + } } if let Some(args) = tc.pointer("/function/arguments").and_then(|v| v.as_str()) { if !args.is_empty() { - let _parsed: Value = serde_json::from_str(args).unwrap_or_default(); emit_event( &mut out, "content_block_delta", &json!({ "type": "content_block_delta", - "index": idx, + "index": anth_idx, "delta": { "type": "input_json_delta", "partial_json": args @@ -388,44 +393,60 @@ pub fn openai_chunk_to_anthropic_sse(chunk: &Value, buffer: &mut AnthropicStream } } - if let Some(reason) = finish { - let sr = match reason { - "stop" => "end_turn", - "length" => "max_tokens", - "tool_calls" => "tool_use", - _ => "end_turn", - }; + out +} + +/// Close every open content block and emit message_delta/message_stop. +/// Callers doing extra out-of-band block injection (e.g. heuristic tool-call +/// extraction) must do so — and register/close their own indices — before +/// calling this, since it ends the message. +pub fn finish_stream(chunk: &Value, buffer: &mut AnthropicStreamBuffer) -> Vec { + let mut out = Vec::new(); + + let finish_reason = chunk + .pointer("/choices/0/finish_reason") + .and_then(|v| v.as_str()) + .unwrap_or("stop"); + + for idx in std::mem::take(&mut buffer.open_indices) { emit_event( &mut out, "content_block_stop", &json!({ "type": "content_block_stop", - "index": 0 - }), - ); - emit_event( - &mut out, - "message_delta", - &json!({ - "type": "message_delta", - "delta": { - "stop_reason": sr, - "stop_sequence": null - }, - "usage": { - "output_tokens": chunk.pointer("/usage/completion_tokens").and_then(|v| v.as_u64()).unwrap_or(0) - } - }), - ); - emit_event( - &mut out, - "message_stop", - &json!({ - "type": "message_stop" + "index": idx }), ); } + let sr = match finish_reason { + "stop" => "end_turn", + "length" => "max_tokens", + "tool_calls" => "tool_use", + _ => "end_turn", + }; + emit_event( + &mut out, + "message_delta", + &json!({ + "type": "message_delta", + "delta": { + "stop_reason": sr, + "stop_sequence": null + }, + "usage": { + "output_tokens": chunk.pointer("/usage/completion_tokens").and_then(|v| v.as_u64()).unwrap_or(0) + } + }), + ); + emit_event( + &mut out, + "message_stop", + &json!({ + "type": "message_stop" + }), + ); + out } @@ -434,6 +455,9 @@ pub struct AnthropicStreamBuffer { pub started: bool, pub message_id: Option, pub block_id: Option, + pub next_index: usize, + pub open_indices: std::collections::BTreeSet, + pub tool_index_map: std::collections::HashMap, } fn emit_event(out: &mut Vec, event: &str, data: &Value) { @@ -544,4 +568,44 @@ mod tests { assert_eq!(openai["tools"][0]["function"]["name"], "get_weather"); assert_eq!(openai["tool_choice"]["type"], "auto"); } + + #[test] + fn test_stream_native_tool_call_does_not_collide_with_text_index() { + let mut buffer = AnthropicStreamBuffer::default(); + + let start = json!({"choices": [{"delta": {}, "index": 0}]}); + openai_chunk_to_anthropic_sse(&start, &mut buffer); + + // Native tool_calls whose own `index` is 0 (as most providers emit + // for the first tool call) must not reuse content-block index 0, + // which the eagerly-opened text block already claims. + let tool_delta = json!({ + "choices": [{ + "delta": { "tool_calls": [{ "index": 0, "id": "call_1", "function": { "name": "get_weather" } }] } + }] + }); + let bytes = openai_chunk_to_anthropic_sse(&tool_delta, &mut buffer); + let text = String::from_utf8(bytes).unwrap(); + assert!( + text.contains("\"index\":1"), + "tool block should get a fresh index, got: {text}" + ); + assert!(buffer.open_indices.contains(&0)); + assert!(buffer.open_indices.contains(&1)); + + let finish = json!({"choices": [{"finish_reason": "tool_calls"}]}); + let out = String::from_utf8(finish_stream(&finish, &mut buffer)).unwrap(); + assert_eq!( + out.matches("event: content_block_stop").count(), + 2, + "expected both blocks closed, got: {out}" + ); + assert!(buffer.open_indices.is_empty()); + let stop_pos = out.find("message_stop").unwrap(); + let last_block_stop_pos = out.rfind("content_block_stop").unwrap(); + assert!( + last_block_stop_pos < stop_pos, + "content_block_stop must precede message_stop" + ); + } } From 607c97729a85fb2819f53a4bae84f0560a5690f6 Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Sun, 19 Jul 2026 16:01:12 +0530 Subject: [PATCH 6/7] fix(flare-proxy): add license metadata, sync Cargo.lock nanoid entry cargo-deny licenses failed: flare-proxy had no license field, unlike every other workspace crate. clippy --locked failed: Cargo.lock was never regenerated after nanoid was added as a dependency. --- Cargo.lock | 1 + crates/flare-proxy/Cargo.toml | 3 +++ 2 files changed, 4 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index d5c90477..d852bc17 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1012,6 +1012,7 @@ dependencies = [ "axum", "base64", "futures", + "nanoid", "regex", "reqwest 0.12.28", "serde", diff --git a/crates/flare-proxy/Cargo.toml b/crates/flare-proxy/Cargo.toml index 2f9633d6..2015e8b7 100644 --- a/crates/flare-proxy/Cargo.toml +++ b/crates/flare-proxy/Cargo.toml @@ -3,6 +3,9 @@ name = "flare-proxy" version = "0.1.0" edition = "2021" description = "Anthropic→OpenAI proxy for free providers (NVIDIA NIM, OpenRouter, LM Studio)" +license = "Apache-2.0" +repository = "https://github.com/getappz/agentflare" +publish = false [dependencies] serde = { version = "1", features = ["derive"] } From 4950368fcc9a1f6de7199353301209f10de52bfc Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Sun, 19 Jul 2026 16:26:38 +0530 Subject: [PATCH 7/7] fix(flare-proxy): address CodeRabbit review findings - shape_xlat: prepend system prompt as a system message instead of setting body[\system\], which the OpenAI chat-completions schema does not have; system prompts were being silently dropped on every request (critical - breaks the proxy for real agent traffic). - shape_xlat: translate_user_content stringified image data and tool_use_id via Value's Display impl, embedding JSON quotes into base64 data URLs and tool_result markers; extract as str instead. - heuristic: extract_json_tool_block sliced from the \name\ substring instead of the enclosing brace, so find_balanced_brace matched the nested arguments object and the extractor always returned None; slice from the regex match start instead. find_balanced_brace now also tracks string state so braces inside string values do not miscount. - heuristic/think: add (?s) so .*? spans newlines, matching pretty-printed JSON tool calls; make model-family detection case-insensitive; think.rs now also matches mistral/mixtral. - lib: reject a configured-but-empty AGENTFLARE_PROXY_TOKEN instead of silently disabling the auth gate. - forward: reuse one reqwest::Client with a timeout via AppState instead of building a new client per request; buffer raw bytes across SSE chunk boundaries before UTF-8 decoding instead of lossy-decoding each raw chunk, which could mangle a multibyte sequence split across a chunk boundary. Think-tag suppression during streaming is left as a known limitation (flagged Heavy lift by review): raw deltas are already streamed before the accumulated-text pass runs, so it cannot retroactively strip tags from what the client already received. Fixing that requires buffering deltas and delaying emission, tracked separately. --- crates/flare-proxy/src/forward.rs | 46 +++++++++---------- crates/flare-proxy/src/heuristic.rs | 66 +++++++++++++++++++++------- crates/flare-proxy/src/lib.rs | 16 ++++++- crates/flare-proxy/src/shape_xlat.rs | 53 +++++++++++++++++----- crates/flare-proxy/src/think.rs | 7 ++- 5 files changed, 137 insertions(+), 51 deletions(-) diff --git a/crates/flare-proxy/src/forward.rs b/crates/flare-proxy/src/forward.rs index 643837d0..ff403bd0 100644 --- a/crates/flare-proxy/src/forward.rs +++ b/crates/flare-proxy/src/forward.rs @@ -6,7 +6,11 @@ use axum::response::{IntoResponse, Response}; use futures::stream::StreamExt; use serde_json::{json, Value}; -pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Response { +pub async fn proxy_request( + anthropic_body: Value, + config: &ProviderConfig, + client: &reqwest::Client, +) -> Response { let model = anthropic_body .get("model") .and_then(|v| v.as_str()) @@ -58,7 +62,6 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re let needs_heuristic = route.requires_heuristic_tools; let needs_think = route.requires_think_parsing; - let client = reqwest::Client::new(); let mut req_builder = client .post(provider.base_url.trim_end_matches('/').to_string() + "/chat/completions") .json(&openai_req); @@ -103,7 +106,7 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re let stream = resp.bytes_stream(); let mut buffer = AnthropicStreamBuffer::default(); let mut accumulated_text = String::new(); - let mut line_buf = String::new(); + let mut line_buf: Vec = Vec::new(); let sse_stream = stream.filter_map(move |chunk_result| { let chunk = match chunk_result { @@ -111,12 +114,14 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re Err(_) => return futures::future::ready(None), }; - // SSE lines don't align with raw TCP/HTTP chunk boundaries — buffer - // any trailing partial line across chunks instead of silently - // dropping the truncated JSON it would otherwise produce. - line_buf.push_str(&String::from_utf8_lossy(&chunk)); - let split_at = line_buf.rfind('\n').map(|i| i + 1).unwrap_or(0); - let complete: String = line_buf.drain(..split_at).collect(); + line_buf.extend_from_slice(&chunk); + let split_at = line_buf + .iter() + .rposition(|&b| b == b'\n') + .map(|i| i + 1) + .unwrap_or(0); + let complete_bytes: Vec = line_buf.drain(..split_at).collect(); + let complete = String::from_utf8_lossy(&complete_bytes).into_owned(); let mut out = Vec::new(); @@ -134,7 +139,6 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re Err(_) => continue, }; - // Accumulate text for heuristic tool parsing if let Some(delta) = val.pointer("/choices/0/delta/content").and_then(|v| v.as_str()) { accumulated_text.push_str(delta); } @@ -148,10 +152,6 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re out.extend_from_slice(&anthropic_sse); if is_finish { - // Heuristic tool extraction on accumulated text. This must - // open+close its own content block (using a fresh index) - // before finish_stream runs below, since finish_stream ends - // the message with message_stop. if needs_heuristic && !accumulated_text.is_empty() { if let Some(tc) = crate::heuristic::try_extract_tool_call(&accumulated_text) { let idx = buffer.next_index; @@ -176,15 +176,17 @@ pub async fn proxy_request(anthropic_body: Value, config: &ProviderConfig) -> Re } } - // Think tag stripping on accumulated text + // Think tag stripping on accumulated text. NOTE: the raw + // deltas above are already streamed out via + // openai_chunk_to_anthropic_sse before this point runs, so + // this pass over accumulated_text cannot retroactively + // remove think-tag content from what the client already + // received. Properly suppressing think tags requires + // buffering deltas and delaying emission, which is a larger + // change tracked separately; this block intentionally does + // not claim to do that suppression. if needs_think && !accumulated_text.is_empty() { - let (_clean, thoughts) = crate::think::strip_think_tags(&accumulated_text); - if !thoughts.is_empty() { - // We've already streamed the text with think tags. - // In a real implementation, we'd buffer and re-stream. - // For v1, we strip in post-processing of accumulated text. - // The SSE events already went out; this is best-effort cleanup. - } + let _ = crate::think::strip_think_tags(&accumulated_text); } let finish_bytes = shape_xlat::finish_stream(&val, &mut buffer); diff --git a/crates/flare-proxy/src/heuristic.rs b/crates/flare-proxy/src/heuristic.rs index 9ae51a92..24d30571 100644 --- a/crates/flare-proxy/src/heuristic.rs +++ b/crates/flare-proxy/src/heuristic.rs @@ -1,7 +1,7 @@ use regex::Regex; /// Attempt to extract a structured tool call from free-tier model output -/// that doesn't natively support function calling. +/// that does not natively support function calling. #[derive(Debug)] pub struct HeuristicToolCall { pub name: String, @@ -30,7 +30,7 @@ pub fn try_extract_tool_call(text: &str) -> Option { fn extract_invoke_meal(text: &str) -> Option { let re = - Regex::new(r#"\s*(\{.*?\})\s*"#).ok()?; + Regex::new(r#"(?s)\s*(\{.*?\})\s*"#).ok()?; let cap = re.captures(text)?; let name = cap.get(1)?.as_str().to_string(); let args_str = cap.get(2)?.as_str(); @@ -40,7 +40,8 @@ fn extract_invoke_meal(text: &str) -> Option { } fn extract_code_fence_json(text: &str) -> Option { - let re = Regex::new(r#"```(?:json)?\s*\n?(\{.*?"name"\s*:\s*"[^"]+".*?\})\s*\n?```"#).ok()?; + let re = + Regex::new(r#"(?s)```(?:json)?\s*\n?(\{.*?"name"\s*:\s*"[^"]+".*?\})\s*\n?```"#).ok()?; let cap = re.captures(text)?; let json_str = cap.get(1)?.as_str(); let parsed: serde_json::Value = serde_json::from_str(json_str).ok()?; @@ -57,26 +58,35 @@ fn extract_json_tool_block(text: &str) -> Option { let re = Regex::new(r#"\{(?:\s*)"name"\s*:\s*"(?:[^"\\]|\\.)*"\s*,\s*"arguments"\s*:\s*(\{|\[)"#) .ok()?; - if re.is_match(text) { - let start = text.find(r#""name""#)?; - let block = &text[start..]; - if let Some(end) = find_balanced_brace(block) { - let json_str = &block[..=end]; - let parsed: serde_json::Value = serde_json::from_str(json_str).ok()?; - let name = parsed.get("name")?.as_str()?.to_string(); - let args = parsed.get("arguments")?.clone(); - let id = format!("call_{}", nanoid::nanoid!()); - return Some(HeuristicToolCall { name, args, id }); - } - } - None + let m = re.find(text)?; + let block = &text[m.start()..]; + let end = find_balanced_brace(block)?; + let json_str = &block[..=end]; + let parsed: serde_json::Value = serde_json::from_str(json_str).ok()?; + let name = parsed.get("name")?.as_str()?.to_string(); + let args = parsed.get("arguments")?.clone(); + let id = format!("call_{}", nanoid::nanoid!()); + Some(HeuristicToolCall { name, args, id }) } fn find_balanced_brace(s: &str) -> Option { let mut depth = 0i32; let mut started = false; + let mut in_string = false; + let mut escaped = false; for (i, ch) in s.char_indices() { + if in_string { + if escaped { + escaped = false; + } else if ch == '\\' { + escaped = true; + } else if ch == '"' { + in_string = false; + } + continue; + } match ch { + '"' => in_string = true, '{' => { depth += 1; started = true; @@ -96,6 +106,7 @@ fn find_balanced_brace(s: &str) -> Option { /// Check if output needs heuristic tool parsing (free-tier models often /// lack native function calling). pub fn needs_heuristic_tools(model: &str) -> bool { + let model = model.to_lowercase(); model.contains("llama") || model.contains("deepseek") || model.contains("qwen") @@ -126,8 +137,31 @@ mod tests { assert_eq!(call.args["query"], "SELECT * FROM users"); } + #[test] + fn test_extract_code_fence_tool_multiline_json() { + let text = "Here's the result:\n```json\n{\n \"name\": \"search_db\",\n \"arguments\": {\n \"query\": \"SELECT * FROM users\"\n }\n}\n```"; + let call = try_extract_tool_call(text).unwrap(); + assert_eq!(call.name, "search_db"); + assert_eq!(call.args["query"], "SELECT * FROM users"); + } + + #[test] + fn test_extract_bare_json_tool_block() { + let text = + r#"I'll use a tool: {"name": "get_weather", "arguments": {"city": "London"}} done."#; + let call = try_extract_tool_call(text).unwrap(); + assert_eq!(call.name, "get_weather"); + assert_eq!(call.args["city"], "London"); + } + #[test] fn test_no_tool_call() { assert!(try_extract_tool_call("Just a regular response.").is_none()); } + + #[test] + fn test_needs_heuristic_tools_case_insensitive() { + assert!(needs_heuristic_tools("Meta-Llama-3-70B")); + assert!(needs_heuristic_tools("DeepSeek-V3")); + } } diff --git a/crates/flare-proxy/src/lib.rs b/crates/flare-proxy/src/lib.rs index 533cce06..dadb07c3 100644 --- a/crates/flare-proxy/src/lib.rs +++ b/crates/flare-proxy/src/lib.rs @@ -12,12 +12,18 @@ use axum::{ Router, }; pub use providers::ProviderConfig; +use std::time::Duration; pub fn router() -> Router { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(120)) + .build() + .unwrap_or_default(); Router::new() .route("/proxy/v1/messages", post(v1_messages_handler)) .with_state(AppState { config: ProviderConfig::default_free(), + client, }) } @@ -32,6 +38,13 @@ async fn v1_messages_handler( axum::extract::Json(body): axum::extract::Json, ) -> Response { if let Ok(expected) = std::env::var("AGENTFLARE_PROXY_TOKEN") { + if expected.is_empty() { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "AGENTFLARE_PROXY_TOKEN is set but empty", + ) + .into_response(); + } let provided = headers .get("x-agentflare-proxy-token") .and_then(|v| v.to_str().ok()) @@ -40,10 +53,11 @@ async fn v1_messages_handler( return (StatusCode::UNAUTHORIZED, "invalid or missing proxy token").into_response(); } } - forward::proxy_request(body, &state.config).await + forward::proxy_request(body, &state.config, &state.client).await } #[derive(Clone)] struct AppState { config: ProviderConfig, + client: reqwest::Client, } diff --git a/crates/flare-proxy/src/shape_xlat.rs b/crates/flare-proxy/src/shape_xlat.rs index 67e92efa..1161208f 100644 --- a/crates/flare-proxy/src/shape_xlat.rs +++ b/crates/flare-proxy/src/shape_xlat.rs @@ -3,10 +3,12 @@ use serde_json::{json, Value}; pub fn messages_to_chat(anthropic: &Value) -> Option { let model = anthropic.get("model")?.as_str()?; let mut messages = Vec::new(); - let mut system = None; if let Some(s) = anthropic.get("system") { - system = Some(system_text(s)); + messages.push(json!({ + "role": "system", + "content": system_text(s) + })); } let anthropic_messages = anthropic.get("messages")?.as_array()?; @@ -46,9 +48,6 @@ pub fn messages_to_chat(anthropic: &Value) -> Option { if let Some(stop) = anthropic.get("stop_sequences") { body["stop"] = stop.clone(); } - if let Some(s) = system { - body["system"] = json!(s); - } if let Some(tc) = anthropic.get("tool_choice") { body["tool_choice"] = translate_tool_choice(tc); } @@ -80,20 +79,24 @@ fn translate_user_content(content: &Value) -> Value { .filter_map(|block| { let type_ = block.get("type")?.as_str()?; match type_ { - "text" => Some(json!({ "type": "text", "text": block["text"] })), + "text" => { + let text = block.get("text")?.as_str()?; + Some(json!({ "type": "text", "text": text })) + } "image" => { let source = block.get("source")?; - let media_type = source.get("media_type")?; - let data = source.get("data")?; + let media_type = + source.get("media_type")?.as_str().unwrap_or("image/png"); + let data = source.get("data")?.as_str()?; Some(json!({ "type": "image_url", "image_url": { - "url": format!("data:{};base64,{}", media_type.as_str().unwrap_or("image/png"), data) + "url": format!("data:{};base64,{}", media_type, data) } })) } "tool_result" => { - let tool_use_id = block.get("tool_use_id")?; + let tool_use_id = block.get("tool_use_id")?.as_str()?; let content_val = block.get("content")?; let text = match content_val { Value::String(s) => s.clone(), @@ -473,6 +476,31 @@ fn emit_event(out: &mut Vec, event: &str, data: &Value) { mod tests { use super::*; + #[test] + fn test_translate_user_content_image_and_tool_result_not_quoted() { + let anthropic = json!({ + "model": "claude-sonnet-4-20250514", + "messages": [{ + "role": "user", + "content": [ + { + "type": "image", + "source": { "media_type": "image/png", "data": "abc123" } + }, + { + "type": "tool_result", + "tool_use_id": "toolu_01", + "content": "result text" + } + ] + }] + }); + let openai = messages_to_chat(&anthropic).unwrap(); + let parts = openai["messages"][0]["content"].as_array().unwrap(); + assert_eq!(parts[0]["image_url"]["url"], "data:image/png;base64,abc123"); + assert_eq!(parts[1]["text"], "[tool_result id=toolu_01]\nresult text"); + } + #[test] fn test_messages_to_chat_basic() { let anthropic = json!({ @@ -495,7 +523,10 @@ mod tests { "messages": [{"role": "user", "content": "Hi"}] }); let openai = messages_to_chat(&anthropic).unwrap(); - assert_eq!(openai["system"], "You are helpful."); + assert_eq!(openai["messages"][0]["role"], "system"); + assert_eq!(openai["messages"][0]["content"], "You are helpful."); + assert_eq!(openai["messages"][1]["role"], "user"); + assert_eq!(openai["messages"][1]["content"], "Hi"); } #[test] diff --git a/crates/flare-proxy/src/think.rs b/crates/flare-proxy/src/think.rs index 525b6b0a..66af3308 100644 --- a/crates/flare-proxy/src/think.rs +++ b/crates/flare-proxy/src/think.rs @@ -25,7 +25,12 @@ pub fn strip_think_tags(text: &str) -> (String, Vec) { /// Check if output needs think-tag parsing (free-tier models sometimes emit them). pub fn needs_think_parsing(model: &str) -> bool { - model.contains("deepseek") || model.contains("qwen") || model.contains("llama") + let model = model.to_lowercase(); + model.contains("deepseek") + || model.contains("qwen") + || model.contains("llama") + || model.contains("mistral") + || model.contains("mixtral") } #[cfg(test)]