diff --git a/Cargo.lock b/Cargo.lock index 1a44e43..c5047ad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -487,6 +487,23 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "futures-sink" version = "0.3.32" @@ -506,8 +523,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-core", + "futures-io", + "futures-macro", "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -1050,6 +1070,8 @@ version = "0.1.0" dependencies = [ "anyhow", "async-trait", + "axum", + "base64", "chrono", "dashmap", "dirs", @@ -1061,6 +1083,7 @@ dependencies = [ "tempfile", "thiserror", "tokio", + "tokio-stream", "toml", "tracing", "uuid", @@ -1411,12 +1434,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-native-tls", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", ] @@ -2267,6 +2292,19 @@ dependencies = [ "wasmparser", ] +[[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 = "wasmparser" version = "0.244.0" diff --git a/Cargo.toml b/Cargo.toml index 6e7adfc..9fb4f60 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,7 @@ serde_json = "1" toml = "0.8" # HTTP -reqwest = { version = "0.12", features = ["json", "multipart"] } +reqwest = { version = "0.12", features = ["json", "multipart", "stream"] } axum = { version = "0.8", features = ["ws"] } axum-extra = { version = "0.10", features = ["typed-header"] } tower = "0.5" diff --git a/mofa-engine-core/Cargo.toml b/mofa-engine-core/Cargo.toml index 0dcc507..47d44c0 100644 --- a/mofa-engine-core/Cargo.toml +++ b/mofa-engine-core/Cargo.toml @@ -13,6 +13,8 @@ serde = { workspace = true } serde_json = { workspace = true } toml = { workspace = true } reqwest = { workspace = true } +tokio-stream = { workspace = true } +base64 = "0.22" thiserror = { workspace = true } anyhow = { workspace = true } tracing = { workspace = true } @@ -22,3 +24,6 @@ sysinfo = { workspace = true } dashmap = { workspace = true } dirs = { workspace = true } tempfile = { workspace = true } + +[dev-dependencies] +axum = { workspace = true } diff --git a/mofa-engine-core/src/backends.rs b/mofa-engine-core/src/backends.rs index 7783c49..c9ae65d 100644 --- a/mofa-engine-core/src/backends.rs +++ b/mofa-engine-core/src/backends.rs @@ -3,7 +3,9 @@ pub mod local_tts; pub mod ollama; pub mod openai_compat; +pub mod video_task; pub use local_tts::LocalTtsProvider; pub use ollama::OllamaProvider; pub use openai_compat::OpenAiCompatProvider; +pub use video_task::VideoTaskProvider; diff --git a/mofa-engine-core/src/backends/local_tts.rs b/mofa-engine-core/src/backends/local_tts.rs index 4959866..f9c7e00 100644 --- a/mofa-engine-core/src/backends/local_tts.rs +++ b/mofa-engine-core/src/backends/local_tts.rs @@ -205,6 +205,8 @@ impl LocalTtsProvider { tokens_used: None, fallback_used: false, routing_reason: None, + reasoning: None, + files: Vec::new(), }) } } @@ -337,6 +339,7 @@ mod tests { messages: vec![Message { role: "user".into(), content: text.into(), + images: Vec::new(), }], input_file: None, params: serde_json::Value::Null, diff --git a/mofa-engine-core/src/backends/ollama.rs b/mofa-engine-core/src/backends/ollama.rs index a2d30b1..cfecc04 100644 --- a/mofa-engine-core/src/backends/ollama.rs +++ b/mofa-engine-core/src/backends/ollama.rs @@ -98,6 +98,9 @@ struct OllamaChatRequest { struct OllamaMessage { role: String, content: String, + /// Bare base64 image payloads (Ollama chat API format). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + images: Vec, } #[derive(Debug, Deserialize)] @@ -211,10 +214,12 @@ impl Provider for OllamaProvider { messages: vec![OllamaMessage { role: "user".into(), content: " ".into(), + images: Vec::new(), }], stream: false, keep_alive: Some("5m".into()), }; + // (health probe: no images) let url = format!("{}/api/chat", self.base_url); let resp = self .client @@ -251,6 +256,7 @@ impl Provider for OllamaProvider { messages: vec![OllamaMessage { role: "user".into(), content: " ".into(), + images: Vec::new(), }], stream: false, keep_alive: Some("0".into()), @@ -298,6 +304,16 @@ impl Provider for OllamaProvider { .map(|m| OllamaMessage { role: m.role.clone(), content: m.content.clone(), + // Ollama wants bare base64; kernel messages carry data URLs. + images: m + .images + .iter() + .map(|url| { + url.split_once(",") + .map(|(_, payload)| payload.to_string()) + .unwrap_or_else(|| url.clone()) + }) + .collect(), }) .collect(); @@ -354,6 +370,8 @@ impl Provider for OllamaProvider { tokens_used: chat_resp.eval_count, fallback_used: false, routing_reason: None, + reasoning: None, + files: Vec::new(), }) } } diff --git a/mofa-engine-core/src/backends/openai_compat.rs b/mofa-engine-core/src/backends/openai_compat.rs index b5ea556..5ae0e22 100644 --- a/mofa-engine-core/src/backends/openai_compat.rs +++ b/mofa-engine-core/src/backends/openai_compat.rs @@ -6,7 +6,7 @@ use async_trait::async_trait; use mofa_kernel::{ BackendFeature, BackendHealth, Capability, CostTier, EngineError, InferenceRequest, InferenceResponse, LifecycleResult, ModelAvailability, ModelCard, ModelResidency, Provider, - ProviderKind, canonical_model_id, model_id_name, + ProviderKind, StreamDelta, StreamSink, canonical_model_id, model_id_name, }; use reqwest::Client; use serde::{Deserialize, Serialize}; @@ -44,6 +44,14 @@ impl OpenAiCompatProvider { Self::with_output_dir(name, base_url, api_key, models, cost_tier, None) } + /// Replace the built-in HTTP client. The default client follows system + /// proxy settings (cloud providers may need one); callers talking to a + /// loopback endpoint can inject a `no_proxy()` client instead. + pub fn with_http_client(mut self, client: Client) -> Self { + self.client = client; + self + } + /// Create a provider, writing TTS artifacts into `output_dir` (or the system /// temp dir when `None`) so they land where the artifact sweeper looks. pub fn with_output_dir( @@ -83,12 +91,107 @@ struct ChatCompletionRequest { messages: Vec, #[serde(skip_serializing_if = "Option::is_none")] max_tokens: Option, + #[serde(skip_serializing_if = "Option::is_none")] + temperature: Option, + #[serde(skip_serializing_if = "Option::is_none")] + stream: Option, + #[serde(skip_serializing_if = "Option::is_none")] + stream_options: Option, + /// Hybrid reasoning models (e.g. Qwen3) gate their thinking mode with + /// this body flag. + #[serde(skip_serializing_if = "Option::is_none")] + enable_thinking: Option, +} + +/// `stream_options` payload: ask OpenAI-compatible providers to attach a +/// final usage frame to the SSE stream. +#[derive(Debug, Serialize)] +struct StreamOptions { + include_usage: bool, +} + +/// POST {base}/images/generations — the OpenAI image API shape. +#[derive(Debug, Serialize)] +struct ImageGenRequest { + model: String, + prompt: String, + #[serde(skip_serializing_if = "Option::is_none")] + n: Option, + #[serde(skip_serializing_if = "Option::is_none")] + size: Option, + #[serde(skip_serializing_if = "Option::is_none")] + response_format: Option, +} + +#[derive(Debug, Deserialize)] +struct ImageGenResponse { + data: Option>, +} + +#[derive(Debug, Deserialize)] +struct ImageGenItem { + #[serde(default)] + b64_json: Option, + #[serde(default)] + url: Option, } -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Deserialize)] struct ChatMessage { role: String, content: String, + /// Attached images (data/http URLs) for vision input. + #[serde(default)] + images: Vec, + /// DeepSeek-R1 style reasoning trace, on response messages only. + #[serde(default, skip_serializing_if = "Option::is_none")] + reasoning_content: Option, +} + +/// Serialize chat messages with OpenAI vision semantics: plain `content` +/// string when there are no images, multipart content parts otherwise. +impl Serialize for ChatMessage { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + use serde::ser::SerializeMap; + let mut map = serializer.serialize_map(Some(2))?; + map.serialize_entry("role", &self.role)?; + if self.images.is_empty() { + map.serialize_entry("content", &self.content)?; + } else { + let mut parts: Vec = Vec::with_capacity(self.images.len() + 1); + if !self.content.is_empty() { + parts.push(serde_json::json!({ "type": "text", "text": self.content })); + } + for url in &self.images { + parts.push(serde_json::json!({ "type": "image_url", "image_url": { "url": url } })); + } + map.serialize_entry("content", &parts)?; + } + map.end() + } +} + +/// One SSE frame of a streaming chat completion. +#[derive(Debug, Deserialize)] +struct ChatStreamChunk { + choices: Option>, + usage: Option, +} + +#[derive(Debug, Deserialize)] +struct ChatStreamChoice { + delta: Option, +} + +#[derive(Debug, Deserialize)] +struct ChatStreamDelta { + #[serde(default)] + content: Option, + #[serde(default)] + reasoning_content: Option, } #[derive(Debug, Deserialize)] @@ -221,15 +324,41 @@ impl Provider for OpenAiCompatProvider { } match capability { - Capability::Chat => self.invoke_chat(model_name, request, start).await, + Capability::Chat | Capability::Vlm => { + self.invoke_chat(model_name, request, start).await + } Capability::Tts => self.invoke_tts(model_name, request, start).await, Capability::Asr => self.invoke_asr(model_name, request, start).await, + Capability::ImageGen => self.invoke_image_gen(model_name, request, start).await, other => Err(EngineError::UnsupportedOperation(format!( "provider '{}' does not support {other}", self.name ))), } } + + /// Stream chat completions token-by-token. Reasoning models surface their + /// thinking trace as `Thinking` deltas ahead of the answer text. Other + /// capabilities keep the default single-delta compatibility path. + async fn stream( + &self, + model_id: &str, + request: &InferenceRequest, + sink: StreamSink, + ) -> Result { + let capability = request.capability.unwrap_or(Capability::Chat); + if !matches!(capability, Capability::Chat | Capability::Vlm) { + let response = self.invoke(model_id, request).await?; + if let Some(text) = &response.text + && !text.is_empty() + { + let _ = sink.send(StreamDelta::Text(text.clone())).await; + } + return Ok(response); + } + let model_name = model_id_name(model_id); + self.stream_chat(model_name, request, sink).await + } } impl OpenAiCompatProvider { @@ -248,8 +377,14 @@ impl OpenAiCompatProvider { messages: vec![ChatMessage { role: "user".into(), content: "hi".into(), + images: Vec::new(), + reasoning_content: None, }], max_tokens: Some(1), + temperature: None, + stream: None, + stream_options: None, + enable_thinking: None, }; let url = format!("{}/chat/completions", self.base_url); @@ -276,6 +411,8 @@ impl OpenAiCompatProvider { .map(|m| ChatMessage { role: m.role.clone(), content: m.content.clone(), + images: m.images.clone(), + reasoning_content: None, }) .collect(); @@ -289,10 +426,20 @@ impl OpenAiCompatProvider { .and_then(|v| v.as_u64()) .map(|v| v as u32); + let temperature = request.params.get("temperature").and_then(|v| v.as_f64()); + let enable_thinking = request + .params + .get("enable_thinking") + .and_then(|v| v.as_bool()); + let body = ChatCompletionRequest { model: model_name.to_string(), messages, max_tokens, + temperature, + stream: None, + stream_options: None, + enable_thinking, }; let url = format!("{}/chat/completions", self.base_url); @@ -323,12 +470,15 @@ impl OpenAiCompatProvider { detail: format!("parse error: {e}"), })?; - let text = chat + let answer = chat .choices .as_ref() .and_then(|c| c.first()) - .and_then(|c| c.message.as_ref()) - .map(|m| m.content.clone()); + .and_then(|c| c.message.as_ref()); + let text = answer.map(|m| m.content.clone()); + let reasoning = answer + .and_then(|m| m.reasoning_content.clone()) + .filter(|r| !r.is_empty()); let tokens = chat.usage.and_then(|u| u.total_tokens); let duration_ms = start.elapsed().as_millis() as u64; @@ -343,6 +493,151 @@ impl OpenAiCompatProvider { tokens_used: tokens, fallback_used: false, routing_reason: None, + reasoning, + files: Vec::new(), + }) + } + + /// Generate images via the OpenAI images API. b64 payloads are written + /// straight to the artifacts dir; url payloads are downloaded first. + /// Every artifact path lands in `files` (the first also mirrors into + /// `file` for single-image callers). + async fn invoke_image_gen( + &self, + model_name: &str, + request: &InferenceRequest, + start: std::time::Instant, + ) -> Result { + let prompt = request + .messages + .first() + .map(|m| m.content.clone()) + .or_else(|| { + request + .params + .get("prompt") + .and_then(|v| v.as_str()) + .map(String::from) + }) + .ok_or_else(|| { + EngineError::InvalidRequest("image generation requires a prompt".into()) + })?; + + let n = request + .params + .get("n") + .and_then(|v| v.as_u64()) + .map(|v| v as u32); + let size = request + .params + .get("size") + .and_then(|v| v.as_str()) + .map(String::from); + + let body = ImageGenRequest { + model: model_name.to_string(), + prompt, + n, + size, + response_format: Some("b64_json".into()), + }; + + let url = format!("{}/images/generations", self.base_url); + let resp = self + .client + .post(&url) + .bearer_auth(&self.api_key) + .json(&body) + .send() + .await + .map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: e.to_string(), + })?; + + if !resp.status().is_success() { + let status = resp.status(); + let text = resp.text().await.unwrap_or_default(); + return Err(EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("image gen HTTP {status}: {text}"), + }); + } + + let payload: ImageGenResponse = + resp.json().await.map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("image gen parse error: {e}"), + })?; + + let items = payload.data.unwrap_or_default(); + if items.is_empty() { + return Err(EngineError::ProviderError { + provider: self.name.clone(), + detail: "image gen returned no artifacts".into(), + }); + } + + let mut files: Vec = Vec::with_capacity(items.len()); + for (index, item) in items.into_iter().enumerate() { + let bytes = if let Some(b64) = item.b64_json { + use base64::Engine as _; + base64::engine::general_purpose::STANDARD + .decode(b64.as_bytes()) + .map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("invalid b64 image: {e}"), + })? + } else if let Some(image_url) = item.url { + self.client + .get(&image_url) + .send() + .await + .map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("image download failed: {e}"), + })? + .bytes() + .await + .map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("image download read failed: {e}"), + })? + .to_vec() + } else { + continue; + }; + let path = self.output_dir.join(format!( + "mofa_img_{:02}_{}.png", + index, + uuid::Uuid::new_v4() + )); + tokio::fs::write(&path, &bytes) + .await + .map_err(|e| EngineError::Internal(format!("write error: {e}")))?; + files.push(path.to_string_lossy().to_string()); + } + + if files.is_empty() { + return Err(EngineError::ProviderError { + provider: self.name.clone(), + detail: "image gen artifacts had neither b64_json nor url".into(), + }); + } + + let duration_ms = start.elapsed().as_millis() as u64; + Ok(InferenceResponse { + text: None, + file: Some(files[0].clone()), + model_used: model_name.to_string(), + provider: self.name.clone(), + duration_ms, + request_id: request.request_id.clone(), + tokens_used: None, + fallback_used: false, + routing_reason: None, + reasoning: None, + files, }) } @@ -425,6 +720,8 @@ impl OpenAiCompatProvider { tokens_used: None, fallback_used: false, routing_reason: None, + reasoning: None, + files: Vec::new(), }) } @@ -500,6 +797,8 @@ impl OpenAiCompatProvider { tokens_used: None, fallback_used: false, routing_reason: None, + reasoning: None, + files: Vec::new(), }) } } @@ -508,6 +807,290 @@ impl OpenAiCompatProvider { mod tests { use super::*; + /// Vision input: kernel messages carrying image URLs serialize as + /// multipart content parts (text + image_url) on the provider wire. + #[tokio::test] + async fn chat_request_serializes_images_as_multipart_content() { + let (seen_tx, mut seen_rx) = tokio::sync::mpsc::channel::(1); + let seen_for_task = std::sync::Arc::new(seen_tx); + + let app = axum::Router::new().route( + "/v1/chat/completions", + axum::routing::post( + move |axum::Json(body): axum::Json| async move { + let _ = seen_for_task.send(body).await; + axum::Json(serde_json::json!({ + "choices": [{ "message": { "role": "assistant", "content": "这是一只猫" } }], + "usage": { "total_tokens": 9 } + })) + }, + ), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let provider = OpenAiCompatProvider::new( + "mock-openai", + format!("http://{addr}/v1"), + "sk-test", + vec![ModelDef { + name: "vlm".into(), + capability: "vlm".into(), + context_window: Some(8192), + memory_mb: None, + }], + CostTier::Low, + ) + .with_http_client(reqwest::Client::builder().no_proxy().build().unwrap()); + + let request = InferenceRequest { + capability: Some(Capability::Vlm), + model: Some("mock-openai/vlm".into()), + app_id: None, + session_id: None, + fallback_policy: Default::default(), + messages: vec![mofa_kernel::Message { + role: "user".into(), + content: "图里是什么？".into(), + images: vec!["data:image/png;base64,QUJD".into()], + }], + input_file: None, + params: serde_json::json!({}), + hint_next: None, + request_id: "req-vlm".into(), + }; + + let resp = provider + .invoke("mock-openai/vlm", &request) + .await + .expect("invoke succeeds"); + assert_eq!(resp.text.as_deref(), Some("这是一只猫")); + + let sent = seen_rx.recv().await.expect("provider saw request body"); + let content = &sent["messages"][0]["content"]; + assert!(content.is_array(), "content must be multipart: {content}"); + assert_eq!(content[0]["type"], "text"); + assert_eq!(content[0]["text"], "图里是什么？"); + assert_eq!(content[1]["type"], "image_url"); + assert_eq!(content[1]["image_url"]["url"], "data:image/png;base64,QUJD"); + } + + /// No images → plain string content (byte-compatible with before). + #[test] + fn chat_message_without_images_serializes_plain_content() { + let msg = ChatMessage { + role: "user".into(), + content: "hi".into(), + images: Vec::new(), + reasoning_content: None, + }; + let json = serde_json::to_value(&msg).unwrap(); + assert_eq!(json["content"], "hi"); + assert!(json.get("images").is_none()); + } + /// Image generation against a mock images endpoint: b64 artifacts are + /// written to the output dir, `files` lists them all, and the request + /// body carries prompt/n/size. + #[tokio::test] + async fn image_gen_writes_artifacts_and_forwards_params() { + use base64::Engine as _; + let png_1px_red = base64::engine::general_purpose::STANDARD.encode([ + 0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, 0x00, 0x00, 0x00, 0x0D, 0x49, 0x48, + 0x44, 0x52, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, 0x08, 0x02, 0x00, 0x00, + 0x00, 0x90, 0x77, 0x53, 0xDE, + ]); + + let (seen_tx, mut seen_rx) = tokio::sync::mpsc::channel::(1); + let seen_for_task = std::sync::Arc::new(seen_tx); + + let app = axum::Router::new().route( + "/v1/images/generations", + axum::routing::post( + move |axum::Json(body): axum::Json| async move { + let _ = seen_for_task.send(body).await; + axum::Json(serde_json::json!({ + "created": 1, + "data": [ + { "b64_json": png_1px_red }, + { "b64_json": png_1px_red }, + ] + })) + }, + ), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let output_dir = tempfile::tempdir().unwrap(); + let provider = OpenAiCompatProvider::with_output_dir( + "mock-openai", + format!("http://{addr}/v1"), + "sk-test", + vec![ModelDef { + name: "gpt-image".into(), + capability: "image_gen".into(), + context_window: None, + memory_mb: None, + }], + CostTier::High, + Some(output_dir.path().to_string_lossy().to_string()), + ) + .with_http_client(reqwest::Client::builder().no_proxy().build().unwrap()); + + let request = InferenceRequest { + capability: Some(Capability::ImageGen), + model: Some("mock-openai/gpt-image".into()), + app_id: None, + session_id: None, + fallback_policy: Default::default(), + messages: vec![mofa_kernel::Message { + role: "user".into(), + content: "一只橘猫".into(), + images: Vec::new(), + }], + input_file: None, + params: serde_json::json!({"n": 2, "size": "1024x1024"}), + hint_next: None, + request_id: "req-img".into(), + }; + + let resp = provider + .invoke("mock-openai/gpt-image", &request) + .await + .expect("image gen succeeds"); + + assert_eq!(resp.files.len(), 2); + assert_eq!(resp.file.as_deref(), Some(resp.files[0].as_str())); + for path in &resp.files { + let bytes = std::fs::read(path).expect("artifact on disk"); + // PNG magic survived the round-trip. + assert_eq!(&bytes[..4], &[0x89, 0x50, 0x4E, 0x47]); + } + + let sent = seen_rx.recv().await.expect("provider saw request body"); + assert_eq!(sent["prompt"], "一只橘猫"); + assert_eq!(sent["n"], 2); + assert_eq!(sent["size"], "1024x1024"); + assert_eq!(sent["response_format"], "b64_json"); + assert_eq!(resp.model_used, "gpt-image"); + } + /// Stream an OpenAI-style SSE chat completion (reasoning model) against a + /// mock provider endpoint, and verify delta kinds, order, aggregation, + /// and request passthrough (stream flag, temperature, max_tokens). + #[tokio::test] + async fn stream_chat_parses_reasoning_and_text_deltas() { + use axum::body::Body; + use axum::http::header; + use axum::response::Response; + use tokio_stream::wrappers::ReceiverStream; + + let (seen_tx, mut seen_rx) = tokio::sync::mpsc::channel::(1); + let seen = std::sync::Arc::new(seen_tx); + let seen_for_task = seen.clone(); + + let app = axum::Router::new().route( + "/v1/chat/completions", + axum::routing::post( + move |axum::Json(body): axum::Json| async move { + let _ = seen_for_task.send(body).await; + let (tx, rx) = tokio::sync::mpsc::channel::, std::io::Error>>(8); + tokio::spawn(async move { + // Deliberately split the first frame across two + // channel items: the parser must buffer partial lines. + let frames: Vec = vec![ + r#"data: {"choices":[{"delta":{"reasoning_content":"思考"}}]}"#.into(), + "\n\n".into(), + r#"data: {"choices":[{"delta":{"content":"答"}}]}"#.into(), + "\n\n".into(), + r#"data: {"choices":[{"delta":{"content":"案"}}]}"#.into(), + "\n\n".into(), + r#"data: {"choices":[],"usage":{"total_tokens":7}}"#.into(), + "\n\n".into(), + "data: [DONE]\n\n".into(), + ]; + for frame in frames { + let mid = frame.len() / 2; + let (a, b) = frame.split_at(mid); + let _ = tx.send(Ok(a.as_bytes().to_vec())).await; + let _ = tx.send(Ok(b.as_bytes().to_vec())).await; + } + }); + let body = Body::from_stream(ReceiverStream::new(rx)); + Response::builder() + .status(200) + .header(header::CONTENT_TYPE, "text/event-stream") + .body(body) + .unwrap() + }, + ), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + // The default client follows system proxy settings, which cannot + // reach the loopback mock; talk to it directly. + let direct_client = Client::builder().no_proxy().build().expect("test client"); + let provider = OpenAiCompatProvider::new( + "mock-openai", + &format!("http://{addr}/v1"), + "sk-test", + vec![ModelDef { + name: "reasoner".into(), + capability: "chat".into(), + context_window: Some(8192), + memory_mb: None, + }], + CostTier::Low, + ) + .with_http_client(direct_client); + + let request = InferenceRequest { + capability: Some(Capability::Chat), + model: Some("mock-openai/reasoner".into()), + app_id: None, + session_id: None, + fallback_policy: Default::default(), + messages: vec![mofa_kernel::Message { + role: "user".into(), + content: "hi".into(), + images: Vec::new(), + }], + input_file: None, + params: serde_json::json!({"temperature": 0.3, "max_tokens": 64, "enable_thinking": true}), + hint_next: None, + request_id: "req-t".into(), + }; + + let (sink_tx, mut sink_rx) = tokio::sync::mpsc::channel::(16); + let resp = provider + .stream("mock-openai/reasoner", &request, sink_tx) + .await + .expect("stream succeeds"); + + let mut kinds = Vec::new(); + while let Some(delta) = sink_rx.recv().await { + kinds.push(if delta.is_thinking() { + "thinking" + } else { + "text" + }); + } + assert_eq!(kinds, vec!["thinking", "text", "text"]); + assert_eq!(resp.text.as_deref(), Some("答案")); + assert_eq!(resp.reasoning.as_deref(), Some("思考")); + assert_eq!(resp.tokens_used, Some(7)); + + let sent = seen_rx.recv().await.expect("provider saw request body"); + assert_eq!(sent["stream"], true); + assert_eq!(sent["stream_options"]["include_usage"], true); + assert_eq!(sent["temperature"], 0.3); + assert_eq!(sent["max_tokens"], 64); + assert_eq!(sent["enable_thinking"], true); + assert_eq!(resp.model_used, "reasoner"); + } #[tokio::test] async fn discover_returns_configured_models() { let provider = OpenAiCompatProvider::new( @@ -547,3 +1130,153 @@ mod tests { assert_eq!(p.name(), "x"); } } + +// ==================== Streaming ==================== + +impl OpenAiCompatProvider { + /// Stream a chat completion: POST with `stream: true`, parse the SSE + /// frames, forward `reasoning_content` deltas as `Thinking` and `content` + /// deltas as `Text`, and return the aggregate response. + async fn stream_chat( + &self, + model_name: &str, + request: &InferenceRequest, + sink: StreamSink, + ) -> Result { + let start = std::time::Instant::now(); + let messages: Vec = request + .messages + .iter() + .map(|m| ChatMessage { + role: m.role.clone(), + content: m.content.clone(), + images: m.images.clone(), + reasoning_content: None, + }) + .collect(); + if messages.is_empty() { + return Err(EngineError::InvalidRequest("no messages provided".into())); + } + + let max_tokens = request + .params + .get("max_tokens") + .and_then(|v| v.as_u64()) + .map(|v| v as u32); + let temperature = request.params.get("temperature").and_then(|v| v.as_f64()); + let enable_thinking = request + .params + .get("enable_thinking") + .and_then(|v| v.as_bool()); + + let body = ChatCompletionRequest { + model: model_name.to_string(), + messages, + max_tokens, + temperature, + stream: Some(true), + stream_options: Some(StreamOptions { + include_usage: true, + }), + enable_thinking, + }; + + let url = format!("{}/chat/completions", self.base_url); + let resp = self + .client + .post(&url) + .bearer_auth(&self.api_key) + .json(&body) + .send() + .await + .map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: e.to_string(), + })?; + + if !resp.status().is_success() { + let status = resp.status(); + let text = resp.text().await.unwrap_or_default(); + return Err(EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("HTTP {status}: {text}"), + }); + } + + let mut text_out = String::new(); + let mut reasoning_out = String::new(); + let mut tokens_used: Option = None; + // SSE frames can split across network chunks; buffer partial lines + // (splitting at `\n` is UTF-8 safe). + let mut line_buf: Vec = Vec::new(); + let mut upstream = resp.bytes_stream(); + while let Some(item) = tokio_stream::StreamExt::next(&mut upstream).await { + let bytes = item.map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: e.to_string(), + })?; + line_buf.extend_from_slice(&bytes); + while let Some(pos) = line_buf.iter().position(|&b| b == b'\n') { + let line: Vec = line_buf.drain(..=pos).collect(); + consume_sse_line( + &String::from_utf8_lossy(&line), + &sink, + &mut text_out, + &mut reasoning_out, + &mut tokens_used, + ) + .await; + } + } + + Ok(InferenceResponse { + text: (!text_out.is_empty()).then_some(text_out), + file: None, + model_used: model_name.to_string(), + provider: self.name.clone(), + duration_ms: start.elapsed().as_millis() as u64, + request_id: request.request_id.clone(), + tokens_used, + fallback_used: false, + routing_reason: None, + reasoning: (!reasoning_out.is_empty()).then_some(reasoning_out), + files: Vec::new(), + }) + } +} + +/// Parse one SSE line from an OpenAI-compatible stream and push its deltas. +async fn consume_sse_line( + line: &str, + sink: &StreamSink, + text_out: &mut String, + reasoning_out: &mut String, + tokens_used: &mut Option, +) { + let line = line.trim_end_matches(['\n', '\r']); + let Some(data) = line.strip_prefix("data: ") else { + return; // comments, keep-alive pings, `event:` lines + }; + let data = data.trim(); + if data.is_empty() || data == "[DONE]" { + return; + } + let Ok(chunk) = serde_json::from_str::(data) else { + return; // non-JSON keep-alive payload + }; + if let Some(choice) = chunk.choices.as_ref().and_then(|c| c.first()) { + if let Some(delta) = &choice.delta { + if let Some(reasoning) = delta.reasoning_content.as_ref().filter(|r| !r.is_empty()) { + reasoning_out.push_str(reasoning); + let _ = sink.send(StreamDelta::Thinking(reasoning.clone())).await; + } + if let Some(content) = delta.content.as_ref().filter(|c| !c.is_empty()) { + text_out.push_str(content); + let _ = sink.send(StreamDelta::Text(content.clone())).await; + } + } + } + if let Some(usage) = chunk.usage { + *tokens_used = usage.total_tokens; + } +} diff --git a/mofa-engine-core/src/backends/video_task.rs b/mofa-engine-core/src/backends/video_task.rs new file mode 100644 index 0000000..aa8cf80 --- /dev/null +++ b/mofa-engine-core/src/backends/video_task.rs @@ -0,0 +1,460 @@ +//! Task-based video generation provider (TOOL-02): the create → poll → +//! download pattern used by DashScope video-synthesis (Seedance) and +//! similar vendor APIs. One `invoke` covers the whole lifecycle with +//! bounded polling; the artifact lands in `files`. + +use async_trait::async_trait; +use serde::Deserialize; +use serde_json::{Value, json}; +use std::time::{Duration, Instant}; + +use mofa_kernel::{ + BackendFeature, BackendHealth, Capability, CostTier, EngineError, InferenceRequest, + InferenceResponse, LifecycleResult, ModelAvailability, ModelCard, ModelResidency, Provider, + ProviderKind, +}; + +use crate::config::ModelDef; + +/// Polling bounds: every 2s, up to 10 minutes (short clips finish in +/// tens of seconds; the cap keeps a hung task from blocking forever). +const POLL_INTERVAL: Duration = Duration::from_secs(2); +const POLL_DEADLINE: Duration = Duration::from_secs(600); + +pub struct VideoTaskProvider { + name: String, + base_url: String, + api_key: String, + models: Vec, + cost_tier: CostTier, + output_dir: std::path::PathBuf, + client: reqwest::Client, +} + +impl VideoTaskProvider { + pub fn new( + name: impl Into, + base_url: impl Into, + api_key: impl Into, + models: Vec, + cost_tier: CostTier, + output_dir: Option, + ) -> Self { + let client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(60)) + .build() + .expect("failed to build HTTP client"); + Self { + name: name.into(), + base_url: base_url.into(), + api_key: api_key.into(), + models, + cost_tier, + output_dir: output_dir + .filter(|s| !s.is_empty()) + .map(std::path::PathBuf::from) + .unwrap_or_else(std::env::temp_dir), + client, + } + } + + /// Replace the built-in HTTP client (tests inject a no-proxy client + /// for loopback mocks). + pub fn with_http_client(mut self, client: reqwest::Client) -> Self { + self.client = client; + self + } + + async fn create_task( + &self, + model_name: &str, + prompt: &str, + params: &Value, + ) -> Result { + let mut body = json!({ + "model": model_name, + "input": { + "prompt": prompt, + }, + "parameters": {}, + }); + if let Some(size) = params.get("size").and_then(Value::as_str) { + // DashScope accepts WxH like "1280x720". + body["input"]["prompt"] = json!(format!("{prompt}")); + let _ = size; // size maps below for vendors that take it explicitly + } + if let Some(duration) = params.get("duration").and_then(Value::as_u64) { + body["parameters"]["duration"] = json!(duration); + } + if let Some(size) = params.get("size").and_then(Value::as_str) { + body["parameters"]["size"] = json!(size); + } + + let url = format!("{}/video-synthesis", self.base_url); + let resp = self + .client + .post(&url) + .bearer_auth(&self.api_key) + .header("X-DashScope-Async", "enable") + .json(&body) + .send() + .await + .map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("create task failed: {e}"), + })?; + if !resp.status().is_success() { + let status = resp.status(); + let text = resp.text().await.unwrap_or_default(); + return Err(EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("create task HTTP {status}: {text}"), + }); + } + let payload: Value = resp.json().await.map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("create task parse error: {e}"), + })?; + payload + .pointer("/output/task_id") + .and_then(Value::as_str) + .map(str::to_string) + .ok_or_else(|| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("create task returned no task_id: {payload}"), + }) + } + + /// Poll until the task leaves PENDING/RUNNING. Returns the final + /// status payload. + async fn poll_task(&self, task_id: &str) -> Result { + let url = format!("{}/tasks/{task_id}", self.base_url); + let deadline = Instant::now() + POLL_DEADLINE; + loop { + if Instant::now() >= deadline { + return Err(EngineError::Timeout(format!( + "video task '{task_id}' did not finish within {}s", + POLL_DEADLINE.as_secs() + ))); + } + let resp = self + .client + .get(&url) + .bearer_auth(&self.api_key) + .send() + .await + .map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("poll task failed: {e}"), + })?; + if !resp.status().is_success() { + let status = resp.status(); + let text = resp.text().await.unwrap_or_default(); + return Err(EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("poll task HTTP {status}: {text}"), + }); + } + let payload: Value = resp.json().await.map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("poll task parse error: {e}"), + })?; + match payload + .pointer("/output/task_status") + .and_then(Value::as_str) + .unwrap_or("UNKNOWN") + { + "SUCCEEDED" => return Ok(payload), + "FAILED" | "CANCELED" | "UNKNOWN" => { + let message = payload + .pointer("/output/message") + .and_then(Value::as_str) + .unwrap_or("no failure detail"); + return Err(EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("video task '{task_id}' ended unsuccessfully: {message}"), + }); + } + _ => tokio::time::sleep(POLL_INTERVAL).await, + } + } + } + + async fn download_video(&self, url: &str) -> Result, EngineError> { + self.client + .get(url) + .send() + .await + .map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("video download failed: {e}"), + })? + .bytes() + .await + .map(|b| b.to_vec()) + .map_err(|e| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("video download read failed: {e}"), + }) + } +} + +#[derive(Debug, Deserialize)] +struct TaskStatusResponse { + output: TaskOutput, +} + +#[derive(Debug, Deserialize)] +struct TaskOutput { + task_id: String, + #[serde(default)] + task_status: String, + #[serde(default)] + video_url: Option, + #[serde(default)] + metrics: Option, +} + +#[async_trait] +impl Provider for VideoTaskProvider { + fn name(&self) -> &str { + &self.name + } + + fn kind(&self) -> ProviderKind { + ProviderKind::VideoTask + } + + fn features(&self) -> Vec { + vec![] + } + + async fn discover(&self) -> Result, EngineError> { + let cards = self + .models + .iter() + .filter_map(|m| { + let cap = Capability::from_str_loose(&m.capability)?; + let mut card = + ModelCard::new(self.name.clone(), m.name.clone(), cap, self.cost_tier); + card.id = format!("{}/{}", self.name, m.name); + card.availability = ModelAvailability::Configured; + card.residency = ModelResidency::Remote; + card.context_window = m.context_window.unwrap_or(0); + card.memory_estimate_bytes = 0; + card.execution.max_concurrency = 4; + card.refresh_status(); + Some(card) + }) + .collect(); + Ok(cards) + } + + async fn health(&self) -> Result { + if self.api_key.is_empty() { + return Ok(BackendHealth::Unavailable); + } + Ok(BackendHealth::Healthy) + } + + async fn load(&self, model_id: &str) -> Result { + Ok(LifecycleResult { + model_id: model_id.into(), + residency: ModelResidency::Remote, + memory_bytes: Some(0), + changed: false, + }) + } + + async fn unload(&self, model_id: &str) -> Result { + Ok(LifecycleResult { + model_id: model_id.into(), + residency: ModelResidency::Remote, + memory_bytes: Some(0), + changed: false, + }) + } + + async fn invoke( + &self, + model_id: &str, + request: &InferenceRequest, + ) -> Result { + let start = Instant::now(); + let model_name = mofa_kernel::model_id_name(model_id); + let prompt = request + .messages + .first() + .map(|m| m.content.clone()) + .ok_or_else(|| { + EngineError::InvalidRequest("video generation requires a prompt".into()) + })?; + + let task_id = self + .create_task(model_name, &prompt, &request.params) + .await?; + let final_status = self.poll_task(&task_id).await?; + let video_url = final_status + .pointer("/output/video_url") + .and_then(Value::as_str) + .ok_or_else(|| EngineError::ProviderError { + provider: self.name.clone(), + detail: format!("succeeded task '{task_id}' carried no video_url: {final_status}"), + })?; + let bytes = self.download_video(video_url).await?; + + let path = self + .output_dir + .join(format!("mofa_video_{}.mp4", uuid::Uuid::new_v4())); + tokio::fs::write(&path, &bytes) + .await + .map_err(|e| EngineError::Internal(format!("write error: {e}")))?; + let file = path.to_string_lossy().to_string(); + + Ok(InferenceResponse { + text: None, + file: Some(file.clone()), + model_used: model_name.to_string(), + provider: self.name.clone(), + duration_ms: start.elapsed().as_millis() as u64, + request_id: request.request_id.clone(), + tokens_used: None, + fallback_used: false, + routing_reason: None, + reasoning: None, + files: vec![file], + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::ModelDef; + + #[test] + fn task_status_response_parses_dashscope_shape() { + let raw = r#"{ + "output": { + "task_id": "t-1", + "task_status": "SUCCEEDED", + "video_url": "https://cdn.example/v.mp4" + } + }"#; + let parsed: TaskStatusResponse = serde_json::from_str(raw).unwrap(); + assert_eq!(parsed.output.task_id, "t-1"); + assert_eq!(parsed.output.task_status, "SUCCEEDED"); + assert_eq!( + parsed.output.video_url.as_deref(), + Some("https://cdn.example/v.mp4") + ); + } + + #[tokio::test] + async fn full_lifecycle_against_mock_endpoints() { + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + let polls = Arc::new(AtomicUsize::new(0)); + let polls_for_task = polls.clone(); + // addr is known only after binding; use a shared slot the poll + // handler reads at request time. + let addr_slot = Arc::new(std::sync::Mutex::new(String::new())); + let addr_for_poll = addr_slot.clone(); + + let app = axum::Router::new() + .route( + "/v1/video-synthesis", + axum::routing::post(|axum::Json(body): axum::Json| async move { + assert!(body["input"]["prompt"].as_str().is_some()); + let task = serde_json::json!({ + "output": { "task_id": "task-42", "task_status": "PENDING" } + }); + axum::Json(task) + }), + ) + .route( + "/v1/tasks/{task_id}", + axum::routing::get( + move |axum::extract::Path(task_id): axum::extract::Path| async move { + let n = polls_for_task.fetch_add(1, Ordering::SeqCst); + // PENDING → RUNNING → SUCCEEDED + let status = match n { + 0 => "PENDING", + 1 => "RUNNING", + _ => "SUCCEEDED", + }; + // A real http URL the provider can download. + let host = addr_for_poll.lock().unwrap().clone(); + let video_url = format!("http://{host}/video"); + let body = if status == "SUCCEEDED" { + serde_json::json!({ + "output": { + "task_id": task_id, + "task_status": status, + "video_url": video_url, + } + }) + } else { + serde_json::json!({ + "output": { "task_id": task_id, "task_status": status } + }) + }; + axum::Json(body) + }, + ), + ) + .route("/video", axum::routing::get(|| async { "VIDEOBYTES" })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + *addr_slot.lock().unwrap() = addr.to_string(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let output_dir = tempfile::tempdir().unwrap(); + let provider = VideoTaskProvider::new( + "mock-video", + format!("http://{addr}/v1"), + "sk-test", + vec![ModelDef { + name: "seedance".into(), + capability: "video_gen".into(), + context_window: None, + memory_mb: None, + }], + CostTier::High, + Some(output_dir.path().to_string_lossy().to_string()), + ) + .with_http_client(reqwest::Client::builder().no_proxy().build().unwrap()); + + let request = InferenceRequest { + capability: Some(Capability::VideoGen), + model: Some("mock-video/seedance".into()), + app_id: None, + session_id: None, + fallback_policy: Default::default(), + messages: vec![mofa_kernel::Message { + role: "user".into(), + content: "一只橘猫追激光笔".into(), + images: Vec::new(), + }], + input_file: None, + params: serde_json::json!({"size": "1280x720", "duration": 5}), + hint_next: None, + request_id: "req-video".into(), + }; + + let resp = provider + .invoke("mock-video/seedance", &request) + .await + .expect("video lifecycle succeeds"); + assert!(resp.file.as_deref().unwrap().ends_with(".mp4")); + let bytes = std::fs::read(resp.file.as_deref().unwrap()).unwrap(); + assert_eq!(bytes, b"VIDEOBYTES"); + assert_eq!(resp.model_used, "seedance"); + // At least PENDING + RUNNING + SUCCEEDED polls happened. + assert!(polls.load(Ordering::SeqCst) >= 3); + } +} diff --git a/mofa-engine-core/src/config.rs b/mofa-engine-core/src/config.rs index f39927c..3e4d750 100644 --- a/mofa-engine-core/src/config.rs +++ b/mofa-engine-core/src/config.rs @@ -346,6 +346,22 @@ pub struct ModelDef { } impl EngineConfig { + /// The config file the loader would use: explicit path > `./config.toml` + /// > `~/.config/mofa-engine/config.toml`. `None` when only env + /// auto-detection applies (nothing to persist into). + pub fn resolved_path() -> Option { + if let Ok(explicit) = std::env::var("MOFA_ENGINE_CONFIG") { + if !explicit.is_empty() { + return Some(std::path::PathBuf::from(explicit)); + } + } + let cwd = std::path::PathBuf::from("config.toml"); + if cwd.exists() { + return Some(cwd); + } + dirs::config_dir().map(|d| d.join("mofa-engine").join("config.toml")) + } + /// Load configuration from the first available source. /// /// Priority: explicit path > `./config.toml` > `~/.config/mofa-engine/config.toml` > env auto-detect. @@ -683,6 +699,7 @@ impl ProviderConfig { "ollama" => Ok(ProviderKind::Ollama), "openai_compatible" => Ok(ProviderKind::OpenAiCompatible), "local_tts" => Ok(ProviderKind::LocalTts), + "video_task" => Ok(ProviderKind::VideoTask), other => Err(EngineError::Config(format!( "unknown provider kind '{}' for provider '{}'", other, self.name diff --git a/mofa-engine-core/src/engine.rs b/mofa-engine-core/src/engine.rs index e92221e..72a0cd6 100644 --- a/mofa-engine-core/src/engine.rs +++ b/mofa-engine-core/src/engine.rs @@ -13,7 +13,7 @@ use dashmap::DashMap; use mofa_kernel::{ BackendHealth, BackendStatus, Capability, CostTier, EngineError, EngineEvent, EngineStatus, FallbackPolicy, InferenceRequest, InferenceResponse, ModelCard, ModelResidency, Provider, - ProviderHealth, ProviderKind, StreamChunk, model_id_name, + ProviderHealth, ProviderKind, StreamChunk, StreamDelta, model_id_name, }; use serde::{Deserialize, Serialize}; use tokio::sync::{Mutex as AsyncMutex, Semaphore, broadcast, mpsc}; @@ -74,6 +74,8 @@ pub struct MemoryReport { pub struct Engine { /// Named providers. providers: Vec, + /// Providers registered after startup (provider-config API). + runtime_providers: std::sync::RwLock>, /// Cached model cards. models: DashMap, /// Latest backend health by provider. @@ -140,6 +142,86 @@ impl Engine { } /// Create and initialize a new engine from validated configuration. + /// Build a provider from its config entry. Shared by startup assembly + /// and the runtime provider-config API. + pub fn build_provider( + pc: &crate::config::ProviderConfig, + artifacts_dir: Option<&str>, + ) -> Result, EngineError> { + let kind = pc.provider_kind()?; + let cost_tier = CostTier::from_str_loose(&pc.cost_tier); + match kind { + ProviderKind::Ollama => Ok(Arc::new(OllamaProvider::new(&pc.name, &pc.base_url))), + ProviderKind::OpenAiCompatible => Ok(Arc::new(OpenAiCompatProvider::with_output_dir( + &pc.name, + &pc.base_url, + pc.api_key.clone().unwrap_or_default(), + pc.models.clone(), + cost_tier, + artifacts_dir.map(str::to_string), + ))), + ProviderKind::VideoTask => Ok(Arc::new(crate::backends::VideoTaskProvider::new( + &pc.name, + &pc.base_url, + pc.api_key.clone().unwrap_or_default(), + pc.models.clone(), + cost_tier, + artifacts_dir.map(str::to_string), + ))), + ProviderKind::LocalTts => { + let command = pc.command.clone().ok_or_else(|| { + EngineError::Config(format!( + "provider '{}' (local_tts) requires a command", + pc.name + )) + })?; + Ok(Arc::new(LocalTtsProvider::new( + &pc.name, + command, + pc.args.clone(), + pc.output_format.clone(), + pc.output_dir.clone(), + pc.models.clone(), + ))) + } + _ => Err(EngineError::Config(format!( + "provider '{}' uses unsupported provider kind", + pc.name + ))), + } + } + + /// Register a provider at runtime (provider-config API). The provider + /// joins routing immediately; callers persist the config entry + /// separately so it survives restarts. + pub fn add_provider_config( + &self, + pc: &crate::config::ProviderConfig, + ) -> Result<(), EngineError> { + if !pc.enabled { + return Err(EngineError::InvalidRequest(format!( + "provider '{}' is disabled", + pc.name + ))); + } + let kind = pc.provider_kind()?; + let provider = Self::build_provider(pc, None)?; + let mut runtime = self + .runtime_providers + .write() + .expect("runtime providers lock"); + // Replace an existing runtime entry with the same name. + runtime.retain(|registered| registered.name != pc.name); + runtime.push(RegisteredProvider { + name: pc.name.clone(), + kind, + priority: pc.priority, + provider, + }); + drop(runtime); + Ok(()) + } + pub async fn try_new(config: EngineConfig) -> Result, EngineError> { config.validate()?; @@ -157,40 +239,8 @@ impl Engine { } let kind = pc.provider_kind()?; - let cost_tier = CostTier::from_str_loose(&pc.cost_tier); - let provider: Arc = match kind { - ProviderKind::Ollama => Arc::new(OllamaProvider::new(&pc.name, &pc.base_url)), - ProviderKind::OpenAiCompatible => Arc::new(OpenAiCompatProvider::with_output_dir( - &pc.name, - &pc.base_url, - pc.api_key.clone().unwrap_or_default(), - pc.models.clone(), - cost_tier, - config.artifacts.dir.clone(), - )), - ProviderKind::LocalTts => { - let command = pc.command.clone().ok_or_else(|| { - EngineError::Config(format!( - "provider '{}' (local_tts) requires a command", - pc.name - )) - })?; - Arc::new(LocalTtsProvider::new( - &pc.name, - command, - pc.args.clone(), - pc.output_format.clone(), - pc.output_dir.clone(), - pc.models.clone(), - )) - } - _ => { - return Err(EngineError::Config(format!( - "provider '{}' uses unsupported provider kind", - pc.name - ))); - } - }; + let provider: Arc = + Self::build_provider(pc, config.artifacts.dir.as_deref())?; providers.push(RegisteredProvider { name: pc.name.clone(), @@ -215,6 +265,7 @@ impl Engine { let engine = Arc::new(Self { providers, + runtime_providers: std::sync::RwLock::new(Vec::new()), models: DashMap::new(), backend_health: DashMap::new(), memory, @@ -257,11 +308,10 @@ impl Engine { async fn discover_all(&self) { let discovery_timeout = self.timeouts.discovery(); let handles = self - .providers - .iter() - .map(|registered| { - let name = registered.name.clone(); - let provider = Arc::clone(®istered.provider); + .registered_snapshot() + .into_iter() + .map(|(name, _kind, _priority, provider)| { + let name = name.clone(); tokio::spawn(async move { let result = tokio::time::timeout(discovery_timeout, provider.discover()) .await @@ -737,14 +787,18 @@ impl Engine { }) .await; - // Forward the provider's text deltas into the output stream as they + // Forward the provider's output deltas into the output stream as they // arrive. The provider drops its sink when it returns, ending the // forwarder, so all deltas are flushed before the terminal chunk. - let (delta_tx, mut delta_rx) = mpsc::channel::(64); + let (delta_tx, mut delta_rx) = mpsc::channel::(64); let forward_out = out.clone(); let forwarder = tokio::spawn(async move { while let Some(delta) = delta_rx.recv().await { - if forward_out.send(StreamChunk::Text { delta }).await.is_err() { + let chunk = match delta { + StreamDelta::Text(delta) => StreamChunk::Text { delta }, + StreamDelta::Thinking(delta) => StreamChunk::Thinking { delta }, + }; + if forward_out.send(chunk).await.is_err() { break; } } @@ -1320,29 +1374,53 @@ impl Engine { .clone() } + /// All registered providers: startup config plus runtime additions. + fn registered_snapshot(&self) -> Vec<(String, ProviderKind, u8, Arc)> { + let mut out: Vec<(String, ProviderKind, u8, Arc)> = self + .providers + .iter() + .map(|r| (r.name.clone(), r.kind, r.priority, Arc::clone(&r.provider))) + .collect(); + if let Ok(runtime) = self.runtime_providers.read() { + for r in runtime.iter() { + out.push((r.name.clone(), r.kind, r.priority, Arc::clone(&r.provider))); + } + } + out + } + fn find_provider(&self, name: &str) -> Option> { - self.providers + if let Some(found) = self + .providers + .iter() + .find(|registered| registered.name == name) + { + return Some(Arc::clone(&found.provider)); + } + self.runtime_providers + .read() + .ok()? .iter() .find(|registered| registered.name == name) .map(|registered| Arc::clone(®istered.provider)) } fn routing_providers(&self) -> Vec { - self.providers - .iter() - .map(|registered| { + self.registered_snapshot() + .into_iter() + .map(|(name, kind, priority, _)| { let health = self .backend_health - .get(®istered.name) + .get(&name) .map(|h| *h) .unwrap_or(BackendHealth::Unknown); + let circuit_open = self.circuit_breakers.state(&name) == CircuitState::Open; RoutingProvider { - name: registered.name.clone(), - kind: registered.kind, - priority: registered.priority, + name, + kind, + priority, health, - circuit_open: self.circuit_breakers.state(®istered.name) - == CircuitState::Open, + circuit_open, } }) .collect() @@ -1811,6 +1889,8 @@ mod tests { tokens_used: Some(1), fallback_used: false, routing_reason: None, + reasoning: None, + files: Vec::new(), }) } } @@ -1836,6 +1916,7 @@ mod tests { .collect(); let engine = Arc::new(Engine { providers: registered, + runtime_providers: std::sync::RwLock::new(Vec::new()), models: DashMap::new(), backend_health: DashMap::new(), memory: MemoryManager::new(Some(budget_mb)), @@ -1883,6 +1964,7 @@ mod tests { messages: vec![Message { role: "user".into(), content: "hello".into(), + images: Vec::new(), }], input_file: None, params: serde_json::Value::Null, @@ -2553,11 +2635,13 @@ mod tests { chunks } - /// A provider that emits several real text deltas, to prove the engine + /// A provider that emits several real output deltas, to prove the engine /// forwards incremental output in order (not just the compatibility path). struct MultiDeltaProvider { name: String, deltas: Vec, + /// Reasoning deltas, streamed ahead of the answer text. + thinking: Vec, } #[async_trait] @@ -2611,6 +2695,8 @@ mod tests { tokens_used: Some(self.deltas.len() as u32), fallback_used: false, routing_reason: None, + reasoning: (!self.thinking.is_empty()).then(|| self.thinking.concat()), + files: Vec::new(), }) } async fn stream( @@ -2619,8 +2705,13 @@ mod tests { request: &InferenceRequest, sink: mofa_kernel::StreamSink, ) -> Result { + for d in &self.thinking { + let _ = sink + .send(mofa_kernel::StreamDelta::Thinking(d.clone())) + .await; + } for d in &self.deltas { - let _ = sink.send(d.clone()).await; + let _ = sink.send(mofa_kernel::StreamDelta::Text(d.clone())).await; } self.invoke(model_id, request).await } @@ -2654,11 +2745,112 @@ mod tests { assert_eq!(text, "ok"); } + #[tokio::test] + async fn runtime_provider_registration_serves_requests() { + use crate::config::{EngineConfig, ModelDef, ProviderConfig}; + + let app = axum::Router::new().route( + "/v1/chat/completions", + axum::routing::post(|| async { + axum::Json(serde_json::json!({ + "choices": [{ "message": { "role": "assistant", "content": "来自运行时注册的回复" } }], + })) + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let engine = build_engine(vec![], 1000, 0); + engine.refresh_resources().await; + assert_eq!(engine.capabilities().await.len(), 0); + + let pc = ProviderConfig { + name: "runtime-mock".into(), + kind: "openai_compatible".into(), + base_url: format!("http://{addr}/v1"), + api_key: Some("sk-test".into()), + enabled: true, + priority: 5, + cost_tier: "low".into(), + models: vec![ModelDef { + name: "rt-model".into(), + capability: "chat".into(), + context_window: Some(4096), + memory_mb: None, + }], + command: None, + args: Vec::new(), + output_format: None, + output_dir: None, + }; + engine + .add_provider_config(&pc) + .expect("runtime registration"); + engine.refresh_resources().await; + + // The runtime model is discoverable and routable. + let cards = engine.capabilities().await; + assert!( + cards.iter().any(|c| c.id == "runtime-mock/rt-model"), + "cards: {cards:?}" + ); + + let resp = engine + .invoke(mofa_kernel::InferenceRequest { + capability: Some(mofa_kernel::Capability::Chat), + model: Some("runtime-mock/rt-model".into()), + messages: vec![mofa_kernel::Message { + role: "user".into(), + content: "hi".into(), + images: Vec::new(), + }], + params: serde_json::json!({}), + request_id: "req-rt".into(), + app_id: None, + session_id: None, + fallback_policy: Default::default(), + input_file: None, + hint_next: None, + }) + .await + .expect("invoke routes to runtime provider"); + assert_eq!(resp.text.as_deref(), Some("来自运行时注册的回复")); + } + + #[tokio::test] + async fn stream_emits_thinking_deltas_before_text() { + let provider = Arc::new(MultiDeltaProvider { + name: "ollama".into(), + deltas: vec!["答案".into()], + thinking: vec!["先想".into(), "再想".into()], + }); + let engine = build_engine(vec![provider], 1000, 0); + engine.refresh_resources().await; + + let chunks = collect_stream(engine.invoke_stream(chat_request(None))).await; + // Thinking deltas arrive ahead of the answer text, both preserved. + let kinds: Vec<&str> = chunks + .iter() + .filter_map(|c| match c { + StreamChunk::Thinking { .. } => Some("thinking"), + StreamChunk::Text { .. } => Some("text"), + _ => None, + }) + .collect(); + assert_eq!(kinds, vec!["thinking", "thinking", "text"]); + match chunks.last() { + Some(StreamChunk::Completed { .. }) => {} + other => panic!("expected Completed, got {other:?}"), + } + } + #[tokio::test] async fn stream_forwards_incremental_deltas_in_order() { let provider = Arc::new(MultiDeltaProvider { name: "ollama".into(), deltas: vec!["Hello".into(), ", ".into(), "world".into()], + thinking: vec![], }); let engine = build_engine(vec![provider], 1000, 0); engine.refresh_resources().await; diff --git a/mofa-engine-core/src/router.rs b/mofa-engine-core/src/router.rs index a478e7b..cb9e6d2 100644 --- a/mofa-engine-core/src/router.rs +++ b/mofa-engine-core/src/router.rs @@ -300,6 +300,7 @@ mod tests { messages: vec![Message { role: "user".into(), content: "hi".into(), + images: Vec::new(), }], input_file: None, params: serde_json::Value::Null, diff --git a/mofa-engine-sdk/src/server.rs b/mofa-engine-sdk/src/server.rs index c211d02..560388c 100644 --- a/mofa-engine-sdk/src/server.rs +++ b/mofa-engine-sdk/src/server.rs @@ -34,6 +34,8 @@ const REQUEST_ID_HEADER: &str = "x-request-id"; /// Maximum accepted request body size (16 MiB) — bounds base64/JSON payloads. const MAX_BODY_BYTES: usize = 16 * 1024 * 1024; +use mofa_engine_core::config::EngineConfig; + /// Shared application state. #[derive(Clone)] struct AppState { @@ -41,6 +43,9 @@ struct AppState { started_at: std::time::Instant, /// Optional bearer token; when set, `/v1` routes require it. api_token: Option>, + /// Where the loaded config lives; runtime provider additions append + /// here so they survive restarts. + config_path: Option, } /// Start the HTTP server. @@ -65,6 +70,7 @@ pub async fn start_server( engine, started_at: std::time::Instant::now(), api_token, + config_path: EngineConfig::resolved_path(), }; let app = build_app(state); @@ -98,6 +104,10 @@ fn build_app(state: AppState) -> Router { .route("/v1/subscriptions/{id}", delete(unsubscribe_handler)) .route("/v1/events", get(events_handler)) .route("/v1/discovery/refresh", post(refresh_handler)) + .route( + "/v1/config/providers", + get(list_providers_handler).post(add_provider_handler), + ) .route("/v1/models/load", post(load_model_handler)) .route("/v1/models/unload", post(unload_model_handler)) .route_layer(middleware::from_fn_with_state( @@ -229,6 +239,106 @@ async fn refresh_handler(State(state): State) -> Json) -> Json { + let cards = state.engine.capabilities().await; + let mut by_provider: std::collections::BTreeMap> = + std::collections::BTreeMap::new(); + for card in cards { + let provider = card + .id + .split_once('/') + .map(|(p, _)| p.to_string()) + .unwrap_or_else(|| card.provider.clone()); + by_provider.entry(provider).or_default().push(card.id); + } + let providers: Vec = by_provider + .into_iter() + .map(|(name, models)| serde_json::json!({ "name": name, "models": models })) + .collect(); + Json(serde_json::json!({ "providers": providers })) +} + +/// POST /v1/config/providers — register a provider at runtime (BYOK setup). +/// Persists the entry into the config file so it survives restarts. +async fn add_provider_handler( + State(state): State, + Json(pc): Json, +) -> Response { + // Minimal validation: resolvable kind, non-empty name/key for cloud kinds. + if pc.name.trim().is_empty() { + return ( + StatusCode::BAD_REQUEST, + Json(mofa_kernel::ErrorInfo { + code: mofa_kernel::ErrorCode::InvalidRequest, + message: "provider name must not be empty".into(), + retryable: false, + source: None, + }), + ) + .into_response(); + } + if let Err(err) = pc.provider_kind() { + return (StatusCode::BAD_REQUEST, Json(err.info())).into_response(); + } + if pc.kind == "openai_compatible" && pc.api_key.as_deref().unwrap_or("").trim().is_empty() { + return ( + StatusCode::BAD_REQUEST, + Json(mofa_kernel::ErrorInfo { + code: mofa_kernel::ErrorCode::InvalidRequest, + message: "openai_compatible providers require an api_key".into(), + retryable: false, + source: None, + }), + ) + .into_response(); + } + // Persist first: a failing append must not leave runtime/config diverged. + if let Some(path) = &state.config_path { + let entry = format!( + "\n[[providers]]\nname = {:?}\nkind = {:?}\nbase_url = {:?}\napi_key = {:?}\npriority = {}\ncost_tier = {:?}\n", + pc.name, + pc.kind, + pc.base_url, + pc.api_key.clone().unwrap_or_default(), + pc.priority, + pc.cost_tier, + ); + if let Err(e) = std::fs::OpenOptions::new() + .append(true) + .create(true) + .open(path) + .and_then(|mut file| std::io::Write::write_all(&mut file, entry.as_bytes())) + { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(mofa_kernel::ErrorInfo { + code: mofa_kernel::ErrorCode::Config, + message: format!("failed to persist provider config: {e}"), + retryable: false, + source: None, + }), + ) + .into_response(); + } + } + match state.engine.add_provider_config(&pc) { + Ok(()) => { + state.engine.refresh_resources().await; + ( + StatusCode::CREATED, + Json(serde_json::json!({ + "name": pc.name, + "persisted": state.config_path.is_some(), + })), + ) + .into_response() + } + Err(err) => (StatusCode::BAD_REQUEST, Json(err.info())).into_response(), + } +} + /// Prometheus text-exposition metrics. Public (no auth) so scrapers can reach it. async fn metrics_handler(State(state): State) -> impl IntoResponse { let body = state.engine.metrics_prometheus(); diff --git a/mofa-kernel/src/traits.rs b/mofa-kernel/src/traits.rs index 6767793..581b366 100644 --- a/mofa-kernel/src/traits.rs +++ b/mofa-kernel/src/traits.rs @@ -5,7 +5,7 @@ use async_trait::async_trait; use crate::error::EngineError; use crate::types::{ BackendFeature, BackendHealth, InferenceRequest, InferenceResponse, LifecycleResult, ModelCard, - ProviderKind, StreamSink, + ProviderKind, StreamDelta, StreamSink, }; /// A model provider backend. @@ -62,7 +62,7 @@ pub trait Provider: Send + Sync { if let Some(text) = &response.text && !text.is_empty() { - let _ = sink.send(text.clone()).await; + let _ = sink.send(StreamDelta::Text(text.clone())).await; } Ok(response) } diff --git a/mofa-kernel/src/types.rs b/mofa-kernel/src/types.rs index 3eabccb..6c5f479 100644 --- a/mofa-kernel/src/types.rs +++ b/mofa-kernel/src/types.rs @@ -68,6 +68,9 @@ pub enum ProviderKind { OpenAiCompatible, /// Local process-adapter backend (e.g. an MLX/Kokoro or Piper TTS CLI). LocalTts, + /// Task-based video generation (create → poll → download), e.g. + /// DashScope video-synthesis (Seedance). + VideoTask, } impl ProviderKind { @@ -341,6 +344,10 @@ pub struct Message { pub role: String, /// Message content. pub content: String, + /// Attached images (data URLs or http URLs) for vision input. Present + /// only on user turns; providers map them to their multimodal format. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub images: Vec, } /// Named-model fallback behavior. @@ -411,6 +418,14 @@ pub struct InferenceResponse { pub fallback_used: bool, /// Machine-readable routing reason. pub routing_reason: Option, + /// Reasoning/thinking output (e.g. DeepSeek-R1 style `reasoning_content`), + /// when the serving model exposes a separate thinking trace. + #[serde(default)] + pub reasoning: Option, + /// All artifact paths produced by one request (image gen with n>1). + /// `file` mirrors the first entry for single-artifact callers. + #[serde(default)] + pub files: Vec, } /// A single event in a streaming inference response. @@ -436,6 +451,12 @@ pub enum StreamChunk { /// Text delta appended to the output so far. delta: String, }, + /// An incremental piece of reasoning/thinking output, kept separate from + /// the answer text (e.g. DeepSeek-R1 style `reasoning_content`). + Thinking { + /// Reasoning delta appended to the thinking trace so far. + delta: String, + }, /// Terminal success event with aggregate metadata and any file output. Completed { /// Wall-clock duration in milliseconds. @@ -453,11 +474,35 @@ pub enum StreamChunk { Error(ErrorInfo), } -/// Channel a provider uses to emit incremental text deltas while streaming. +/// An incremental piece of provider output pushed to a [`StreamSink`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum StreamDelta { + /// Answer text delta. + Text(String), + /// Reasoning/thinking delta, rendered separately from the answer. + Thinking(String), +} + +impl StreamDelta { + /// The payload string, regardless of kind. + pub fn into_payload(self) -> String { + match self { + Self::Text(s) | Self::Thinking(s) => s, + } + } + + /// Whether this delta carries reasoning rather than answer text. + pub fn is_thinking(&self) -> bool { + matches!(self, Self::Thinking(_)) + } +} + +/// Channel a provider uses to emit incremental output deltas while streaming. /// /// The engine owns the surrounding envelope (`Started`/`Completed`/`Error`); -/// providers push only text deltas here and return the final aggregate response. -pub type StreamSink = tokio::sync::mpsc::Sender; +/// providers push only output deltas here and return the final aggregate +/// response. +pub type StreamSink = tokio::sync::mpsc::Sender; /// Result of a lifecycle operation. #[derive(Debug, Clone, Serialize, Deserialize)]