From 3985eaf1d78a778ea34e04b6771423156cf6690d Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Wed, 10 Jun 2026 12:50:19 -0400 Subject: [PATCH] refactor(llm): introduce Codec trait seam + extract openai_compatible (#481) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What Introduces the `Codec` / `StreamDecoder` trait seam in `fabro-llm` and extracts the OpenAI Chat Completions wire logic behind it as the first conforming codec. Two commits: 1. **`codec/mod.rs`** — the pure translation contract (`encode` / `decode_response` / `stream_decoder`, plus defaulted `encode_count_tokens` / `decode_count_tokens` / `decode_error`) and its data types (`CodecCtx`, `CodecParams`, `EncodedRequest`, `RawEvent`). A codec knows *what the bytes say*; it owns no HTTP, auth, or base URL. 2. **`codec/openai_compatible/`** — the Chat Completions codec split into `wire` / `translate` / `request` / `response` / `stream`. `providers/openai_compatible.rs` shrinks from 1,608 → ~330 lines: a thin transport shell that keeps the public struct/builders/auth/`validate_request`, owns the streaming byte loop + SSE `data:` framing, and delegates all translation to the codec. The two hand-rolled stream unfolds collapse into one. This is the first step of a gateway refactor that separates codec (wire dialect) from transport/auth/route, so later work (Bedrock, OpenRouter) becomes mostly config rather than parallel adapters. ## Behavior No behavior change. The public adapter API (`OpenAiCompatibleAdapter::new` / `with_name` / `with_catalog` / …) is unchanged, and **all 126 wire snapshots pass without edits** — the parity proof that the extracted codec produces byte-identical output. The 29 in-module unit tests move into the codec submodules alongside the code they exercise. ## On the trait `openai_compatible` is the simplest dialect, so its `impl Codec` is just three methods — count-tokens and error mapping inherit the defaults. The contract is defined in full now (a scoped `dead_code` allow on `codec/mod.rs` covers the seams the anthropic/openai/gemini codecs will exercise in follow-up PRs) so those extractions only *override* methods, never extend the trait. Extracting a real codec refined two trait signatures vs. the initial sketch: the canonical `Request` lives in `CodecCtx` (decoders need it for tool-argument parsing and the stream model fallback), and the header-parsed `rate_limit` threads into `decode_response` / `stream_decoder`. `on_event` returns `Result` so dialect error events propagate as stream errors. ## Tests - `cargo nextest run -p fabro-llm` — 515 passed (incl. 126 wire snapshots, unmodified) - `cargo nextest run --workspace` — green - fabro-agent `parity_matrix` (the frozen `OpenAiCompatibleAdapter` contract) — green - `cargo +nightly fmt --check` / `clippy --all-targets -- -D warnings` — clean 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- lib/crates/fabro-llm/src/codec/mod.rs | 169 ++ .../src/codec/openai_compatible/mod.rs | 43 + .../src/codec/openai_compatible/request.rs | 253 +++ .../src/codec/openai_compatible/response.rs | 83 + .../src/codec/openai_compatible/stream.rs | 475 +++++ .../src/codec/openai_compatible/translate.rs | 411 +++++ .../src/codec/openai_compatible/wire.rs | 143 ++ lib/crates/fabro-llm/src/lib.rs | 1 + .../src/providers/openai_compatible.rs | 1571 ++--------------- 9 files changed, 1688 insertions(+), 1461 deletions(-) create mode 100644 lib/crates/fabro-llm/src/codec/mod.rs create mode 100644 lib/crates/fabro-llm/src/codec/openai_compatible/mod.rs create mode 100644 lib/crates/fabro-llm/src/codec/openai_compatible/request.rs create mode 100644 lib/crates/fabro-llm/src/codec/openai_compatible/response.rs create mode 100644 lib/crates/fabro-llm/src/codec/openai_compatible/stream.rs create mode 100644 lib/crates/fabro-llm/src/codec/openai_compatible/translate.rs create mode 100644 lib/crates/fabro-llm/src/codec/openai_compatible/wire.rs diff --git a/lib/crates/fabro-llm/src/codec/mod.rs b/lib/crates/fabro-llm/src/codec/mod.rs new file mode 100644 index 000000000..b8b1c2a94 --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/mod.rs @@ -0,0 +1,169 @@ +//! The codec seam: pure, sync translation between the canonical core +//! (`Request`/`Response`/`StreamEvent`) and a provider wire dialect. +//! +//! A codec knows *what the bytes say*. It does NOT know how they travel +//! (auth, base URL, retries, streaming transport) — that's the adapter/ +//! transport layer. Everything a codec varies on arrives as data in +//! [`CodecCtx`] / [`CodecParams`]; codecs hold no per-request state. +//! +//! The trait is intentionally complete (count-tokens + error mapping have +//! defaults) so the per-dialect codecs that follow only ever *override* +//! methods, never extend the contract. + +// The contract is defined in full now, but `openai_compatible` (the first +// codec) is the simplest dialect and does not exercise every seam: the +// `model`/`params` context, `RawEvent.event`, and the count-tokens methods +// are consumed by the anthropic/openai/gemini codecs and the transport +// consolidation in later PRs of this series. Scoped to this trait-definition +// file; the codec impls below are fully used. +#![allow( + dead_code, + reason = "Codec contract is defined in full ahead of the dialects (PRs 3-6) that exercise \ + the capability context, SSE event type, and count-tokens routes." +)] + +pub(crate) mod openai_compatible; + +use fabro_model::Model; + +use crate::error::{Error, error_from_status_code}; +use crate::providers::common::parse_error_body; +use crate::types::{RateLimitInfo, Request, Response, StreamEvent}; + +/// Per-request context. Borrowed — the codec reads what it needs and returns. +pub(crate) struct CodecCtx<'a> { + /// The canonical request being translated. Decoders read it too + /// (e.g. tool-argument parsing keys off the request's tool definitions; + /// the stream model fallback uses `request.model`). + pub request: &'a Request, + /// Identity stamped into `Response.provider`, and the `provider_options` + /// namespace key for the openai_compatible codec (kimi/zai/…). + pub provider_name: &'a str, + /// The model id to send on the wire — catalog `api_id`, resolved by the + /// route (today `api_id == id` everywhere). + pub deployment_id: &'a str, + /// Model row for capability lookups (prompt_cache, reasoning levels, + /// max_output). `None` when no catalog is injected. + pub model: Option<&'a Model>, + /// Per-route dialect data (model/version placement, …). Defaulted to + /// today's direct-route values; Bedrock/OpenRouter add variants later. + pub params: &'a CodecParams, +} + +/// Per-route dialect knobs, expressed as data so one codec can serve several +/// routes. Starts empty; grows by adding `#[serde(default)]` fields (a +/// non-breaking change) — e.g. PR 3 adds version-placement, #459 adds +/// model-placement for Bedrock. +#[derive(Debug, Default, Clone)] +pub(crate) struct CodecParams; + +/// What [`Codec::encode`] produces. The transport applies `endpoint` + +/// `headers` on top of the route's base URL and auth; the codec never touches +/// HTTP. +pub(crate) struct EncodedRequest { + /// Request body. + pub body: serde_json::Value, + /// Path appended to the route base URL, fully formed by the codec + /// (incl. model-in-path and `?alt=sse` for gemini). e.g. + /// `/chat/completions`. + pub endpoint: String, + /// Dialect headers as data (e.g. `anthropic-version`, beta headers). + /// NOT auth or `content-type` — those are the transport's job. Empty for + /// the openai_compatible codec. + pub headers: Vec<(String, String)>, +} + +/// One framed item off the byte stream, handed to a [`StreamDecoder`]. +pub(crate) struct RawEvent<'a> { + /// SSE `event:` type — `Some` for anthropic; `None` for the data-only + /// framing openai/gemini use. + pub event: Option<&'a str>, + /// The `data:` payload, or a bare JSON line. The sentinel `[DONE]` is + /// passed through verbatim for the decoder to recognize. + pub data: &'a str, +} + +/// Stateless translator for one wire dialect. +pub(crate) trait Codec: Send + Sync { + /// Canonical request (`ctx.request`) → wire request. `stream` selects the + /// streaming shape (`stream: true` in the body, gemini's + /// `:streamGenerateContent` endpoint). Fallible: attachment/parameter + /// encoding can reject. + fn encode(&self, ctx: &CodecCtx<'_>, stream: bool) -> Result; + + /// Wire response body → canonical `Response` (content parts, finish + /// reason, usage). Each dialect's finish-reason map and usage arithmetic + /// live here. Stamps `ctx.provider_name` into `Response.provider` and the + /// transport-parsed `rate_limit` into the response. + fn decode_response( + &self, + body: &str, + ctx: &CodecCtx<'_>, + rate_limit: Option, + ) -> Result; + + /// A fresh stateful decoder for one streaming response. `rate_limit` is the + /// transport-parsed header value to embed in the synthesized `Finish`. + fn stream_decoder( + &self, + ctx: &CodecCtx<'_>, + rate_limit: Option, + ) -> Box; + + /// The third route, if the dialect has one (`/messages/count_tokens`, + /// `/responses/input_tokens`, `:countTokens`). `None` = the dialect has no + /// such route. Whether a given *deployment* may use it is a separate + /// route-level gate (Kimi-over-anthropic) decided before this is called. + fn encode_count_tokens(&self, _ctx: &CodecCtx<'_>) -> Option> { + None + } + + /// Parse the token count out of a count-tokens response. Only called when + /// [`Codec::encode_count_tokens`] returned `Some`; the default guards the + /// invariant for codecs without a count route. + fn decode_count_tokens(&self, _body: &str) -> Result { + Err(Error::Configuration { + message: "codec has no count_tokens route".to_string(), + source: None, + }) + } + + /// Map a non-2xx response to an `Error`. `retry_after` is the + /// transport-parsed `retry-after` header value in seconds (header parsing + /// is the transport's job, like `rate_limit` on the decode methods). + /// Default = shared HTTP-status mapping (what openai_compatible uses); + /// anthropic/openai/gemini override to fold in dialect error bodies + /// (error.type, gRPC status, …). + fn decode_error( + &self, + status: u16, + body: &str, + ctx: &CodecCtx<'_>, + retry_after: Option, + ) -> Error { + let (message, code, raw) = parse_error_body(body, "type"); + error_from_status_code( + status, + message, + ctx.provider_name.to_string(), + code, + raw, + retry_after, + ) + } +} + +/// Stateful per-stream decoder, driven by the shared transport loop. +/// `'static` because it is boxed into the stream's unfold state. +pub(crate) trait StreamDecoder: Send + 'static { + /// One framed event → zero or more canonical `StreamEvent`s. Returns + /// `Err` for dialect error events (anthropic `error`, openai + /// `response.failed`), which the transport yields as a stream error. + fn on_event(&mut self, ev: RawEvent<'_>) -> Result, Error>; + + /// Byte-stream-end hook. Semantics are per-decoder, not shared: + /// anthropic — return nothing (`message_stop` already finished it); + /// openai_compatible — synthesize `Finish` iff content started (minimax); + /// gemini — synthesize `Finish` unconditionally if not yet finished. + fn finish(&mut self) -> Vec; +} diff --git a/lib/crates/fabro-llm/src/codec/openai_compatible/mod.rs b/lib/crates/fabro-llm/src/codec/openai_compatible/mod.rs new file mode 100644 index 000000000..ef45023d5 --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_compatible/mod.rs @@ -0,0 +1,43 @@ +//! The OpenAI Chat Completions (`/chat/completions`) codec. +//! +//! Serves every "OpenAI-compatible" route (kimi, zai, minimax, venice, +//! inception, ollama, litellm, …). Pure translation: no HTTP, auth, or base +//! URL — the adapter shell owns those. Count-tokens and error mapping use the +//! `Codec` trait defaults (this dialect has no count route and uses the shared +//! HTTP-status error mapping). + +mod request; +mod response; +mod stream; +mod translate; +mod wire; + +use crate::codec::{Codec, CodecCtx, EncodedRequest, StreamDecoder}; +use crate::error::Error; +use crate::types::{RateLimitInfo, Response}; + +/// Codec for the OpenAI Chat Completions wire dialect. +pub(crate) struct OpenAiCompatible; + +impl Codec for OpenAiCompatible { + fn encode(&self, ctx: &CodecCtx<'_>, stream: bool) -> Result { + Ok(request::encode(ctx, stream)) + } + + fn decode_response( + &self, + body: &str, + ctx: &CodecCtx<'_>, + rate_limit: Option, + ) -> Result { + response::decode_response(body, ctx, rate_limit) + } + + fn stream_decoder( + &self, + ctx: &CodecCtx<'_>, + rate_limit: Option, + ) -> Box { + Box::new(stream::StreamState::new(ctx, rate_limit)) + } +} diff --git a/lib/crates/fabro-llm/src/codec/openai_compatible/request.rs b/lib/crates/fabro-llm/src/codec/openai_compatible/request.rs new file mode 100644 index 000000000..6d360cd1c --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_compatible/request.rs @@ -0,0 +1,253 @@ +//! Request encoding: canonical `Request` → Chat Completions body. + +use super::translate; +use super::wire::ApiRequest; +use crate::codec::{CodecCtx, EncodedRequest}; + +/// Build the Chat Completions request for `ctx.request`. `stream` toggles the +/// `stream` body field. The body is assembled as a `serde_json::Value` so +/// `provider_options.` fields can be merged in before sending. +/// +/// Infallible for this dialect — the `Codec::encode` `Result` is wrapped by the +/// trait impl. +pub(super) fn encode(ctx: &CodecCtx<'_>, stream: bool) -> EncodedRequest { + let request = ctx.request; + let chat_messages = translate::translate_messages(&request.messages); + let tools = request + .tools + .as_ref() + .map(|t| translate::translate_tools(t)); + let tool_choice = request + .tool_choice + .as_ref() + .map(translate::translate_tool_choice); + let response_format = request + .response_format + .as_ref() + .map(translate::translate_response_format); + + let api_request = ApiRequest { + model: ctx.deployment_id.to_string(), + messages: chat_messages, + temperature: request.temperature, + max_tokens: request.max_tokens, + top_p: request.top_p, + stop: request.stop_sequences.clone(), + tools, + tool_choice, + response_format, + stream: stream.then_some(true), + }; + + let mut body = serde_json::to_value(&api_request).unwrap_or_default(); + merge_provider_options( + &mut body, + request.provider_options.as_ref(), + ctx.provider_name, + ); + + EncodedRequest { + body, + endpoint: "/chat/completions".to_string(), + headers: Vec::new(), + } +} + +/// Merge `provider_options.` fields into the serialized API +/// request body. +/// +/// The provider name is configurable (e.g. "groq", "together", "kimi"), +/// allowing each instance to have its own namespace in `provider_options`. +pub(super) fn merge_provider_options( + body: &mut serde_json::Value, + provider_options: Option<&serde_json::Value>, + provider_name: &str, +) { + let Some(opts) = provider_options.and_then(|opts| opts.get(provider_name)) else { + return; + }; + let Some(body_map) = body.as_object_mut() else { + return; + }; + let Some(opts_map) = opts.as_object() else { + return; + }; + + for (key, value) in opts_map { + body_map.insert(key.clone(), value.clone()); + } +} + +#[cfg(test)] +mod tests { + use super::super::wire::ApiRequest; + use super::*; + use crate::codec::CodecParams; + use crate::types::{Message, Request}; + + fn minimal_request() -> Request { + Request { + model: "llama-3.1-70b".to_string(), + messages: vec![Message::user("Hello")], + provider: None, + tools: None, + tool_choice: None, + response_format: None, + temperature: None, + top_p: None, + max_tokens: None, + stop_sequences: None, + reasoning_effort: None, + speed: None, + metadata: None, + provider_options: None, + } + } + + /// Encode `request` through the codec with `deployment_id == request.model` + /// (the no-catalog case) and return the body. + fn encode_body(request: &Request, provider_name: &str, stream: bool) -> serde_json::Value { + let params = CodecParams; + let deployment_id = request.model.clone(); + let ctx = CodecCtx { + request, + provider_name, + deployment_id: &deployment_id, + model: None, + params: ¶ms, + }; + encode(&ctx, stream).body + } + + #[test] + fn api_request_stream_field_serialization() { + let req = ApiRequest { + model: "test".into(), + messages: vec![], + temperature: None, + max_tokens: None, + top_p: None, + stop: None, + tools: None, + tool_choice: None, + response_format: None, + stream: Some(true), + }; + let json = serde_json::to_value(&req).unwrap(); + assert_eq!(json["stream"], true); + + let req_no_stream = ApiRequest { + model: "test".into(), + messages: vec![], + temperature: None, + max_tokens: None, + top_p: None, + stop: None, + tools: None, + tool_choice: None, + response_format: None, + stream: None, + }; + let json_no_stream = serde_json::to_value(&req_no_stream).unwrap(); + assert!(json_no_stream.get("stream").is_none()); + } + + #[test] + fn encode_uses_deployment_id_as_model() { + let request = minimal_request(); + let params = CodecParams; + let deployment_id = "acme/model-large".to_string(); + let ctx = CodecCtx { + request: &request, + provider_name: "acme", + deployment_id: &deployment_id, + model: None, + params: ¶ms, + }; + let body = encode(&ctx, false).body; + assert_eq!(body["model"], "acme/model-large"); + } + + #[test] + fn provider_options_none_produces_standard_body() { + let request = minimal_request(); + let body = encode_body(&request, "groq", false); + assert_eq!(body["model"], "llama-3.1-70b"); + assert!(body.get("stream").is_none()); + } + + #[test] + fn provider_options_matching_name_merged() { + let mut request = minimal_request(); + request.provider_options = Some(serde_json::json!({ + "groq": { + "frequency_penalty": 0.5, + "presence_penalty": 0.3 + } + })); + let body = encode_body(&request, "groq", false); + assert_eq!(body["frequency_penalty"], 0.5); + assert_eq!(body["presence_penalty"], 0.3); + } + + #[test] + fn provider_options_different_name_ignored() { + let mut request = minimal_request(); + request.provider_options = Some(serde_json::json!({ + "together": { + "repetition_penalty": 1.2 + } + })); + let body = encode_body(&request, "groq", false); + assert!(body.get("repetition_penalty").is_none()); + } + + #[test] + fn provider_options_uses_adapter_name() { + let mut request = minimal_request(); + request.provider_options = Some(serde_json::json!({ + "together": { + "repetition_penalty": 1.2 + } + })); + let body = encode_body(&request, "together", false); + assert_eq!(body["repetition_penalty"], 1.2); + } + + #[test] + fn provider_options_preserves_standard_fields() { + let mut request = minimal_request(); + request.temperature = Some(0.7); + request.max_tokens = Some(200); + request.provider_options = Some(serde_json::json!({ + "groq": { + "frequency_penalty": 0.5 + } + })); + let body = encode_body(&request, "groq", true); + assert_eq!(body["temperature"], 0.7); + assert_eq!(body["max_tokens"], 200); + assert_eq!(body["stream"], true); + assert_eq!(body["frequency_penalty"], 0.5); + } + + #[test] + fn provider_options_can_override_model() { + let mut request = minimal_request(); + request.provider_options = Some(serde_json::json!({ + "groq": { + "model": "custom-model" + } + })); + let body = encode_body(&request, "groq", false); + assert_eq!(body["model"], "custom-model"); + } + + #[test] + fn merge_provider_options_with_non_object_value() { + let mut body = serde_json::json!({"model": "test"}); + let opts = serde_json::json!({"groq": "not-an-object"}); + merge_provider_options(&mut body, Some(&opts), "groq"); + assert_eq!(body["model"], "test"); + } +} diff --git a/lib/crates/fabro-llm/src/codec/openai_compatible/response.rs b/lib/crates/fabro-llm/src/codec/openai_compatible/response.rs new file mode 100644 index 000000000..41f4d0e01 --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_compatible/response.rs @@ -0,0 +1,83 @@ +//! Response decoding: Chat Completions body → canonical `Response`. + +use super::translate::{self, map_finish_reason}; +use super::wire::ApiResponse; +use crate::codec::CodecCtx; +use crate::error::{Error, ProviderErrorDetail, ProviderErrorKind}; +use crate::types::{ + ContentPart, Message, RateLimitInfo, Response, Role, ThinkingData, TokenCounts, ToolCall, +}; + +pub(super) fn decode_response( + body: &str, + ctx: &CodecCtx<'_>, + rate_limit: Option, +) -> Result { + let api_resp: ApiResponse = serde_json::from_str(body) + .map_err(|e| Error::network(format!("failed to parse response: {e}"), e))?; + + let choice = api_resp.choices.first().ok_or_else(|| Error::Provider { + kind: ProviderErrorKind::Server, + detail: Box::new(ProviderErrorDetail::new( + "no choices in response", + ctx.provider_name, + )), + })?; + + let mut content_parts = Vec::new(); + if let Some(reasoning) = &choice.message.reasoning_content { + if !reasoning.is_empty() { + content_parts.push(ContentPart::Thinking(ThinkingData { + text: reasoning.clone(), + signature: None, + redacted: false, + })); + } + } + if let Some(text) = &choice.message.content { + if !text.is_empty() { + content_parts.push(ContentPart::text(text)); + } + } + if let Some(tool_calls) = &choice.message.tool_calls { + let custom_tool_names = translate::custom_tool_names(ctx.request); + for tc in tool_calls { + let arguments = translate::parse_tool_arguments( + &tc.function.name, + &tc.function.arguments, + &custom_tool_names, + ); + let mut tool_call = ToolCall::new(&tc.id, &tc.function.name, arguments); + tool_call.raw_arguments = Some(tc.function.arguments.clone()); + content_parts.push(ContentPart::ToolCall(tool_call)); + } + } + + let finish_reason = map_finish_reason(choice.finish_reason.as_deref()); + + let usage = api_resp + .usage + .as_ref() + .map_or_else(TokenCounts::default, |u| TokenCounts { + input_tokens: u.prompt_tokens, + output_tokens: u.completion_tokens, + ..TokenCounts::default() + }); + + Ok(Response { + id: api_resp.id, + model: api_resp.model, + provider: ctx.provider_name.to_string(), + message: Message { + role: Role::Assistant, + content: content_parts, + name: None, + tool_call_id: None, + }, + finish_reason, + usage, + raw: serde_json::from_str(body).ok(), + warnings: vec![], + rate_limit, + }) +} diff --git a/lib/crates/fabro-llm/src/codec/openai_compatible/stream.rs b/lib/crates/fabro-llm/src/codec/openai_compatible/stream.rs new file mode 100644 index 000000000..f99f74b4f --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_compatible/stream.rs @@ -0,0 +1,475 @@ +//! Streaming decoder: Chat Completions SSE chunks → canonical `StreamEvent`s. +//! +//! Byte reading and `data:` framing live in the transport; this decoder is fed +//! already-stripped payloads (including the `[DONE]` sentinel) via `on_event`. + +use super::translate::{map_finish_reason, parse_tool_arguments}; +use super::wire::{AccumulatedToolCall, StreamChunk}; +use crate::codec::{CodecCtx, RawEvent, StreamDecoder}; +use crate::error::Error; +use crate::types::{ + ContentPart, FinishReason, Message, RateLimitInfo, Response, Role, StreamEvent, ThinkingData, + TokenCounts, ToolCall, +}; + +/// Accumulated state while decoding the Chat Completions SSE stream. +pub(super) struct StreamState { + provider_name: String, + model: String, + response_id: String, + response_model: String, + accumulated_text: String, + accumulated_reasoning: String, + tool_calls: Vec, + usage: TokenCounts, + finish_reason: FinishReason, + text_started: bool, + custom_tool_names: Vec, + /// True after `finish_events()` has run (guards against duplicates). + finished: bool, + rate_limit: Option, +} + +impl StreamState { + pub(super) fn new(ctx: &CodecCtx<'_>, rate_limit: Option) -> Self { + Self { + provider_name: ctx.provider_name.to_string(), + model: ctx.request.model.clone(), + response_id: String::new(), + response_model: String::new(), + accumulated_text: String::new(), + accumulated_reasoning: String::new(), + tool_calls: Vec::new(), + usage: TokenCounts::default(), + finish_reason: FinishReason::Stop, + text_started: false, + custom_tool_names: super::translate::custom_tool_names(ctx.request), + finished: false, + rate_limit, + } + } + + /// Process a parsed SSE chunk and return events to emit, if any. + fn process_chunk(&mut self, chunk: &StreamChunk) -> Option> { + // Capture response metadata from the first chunk. + if let Some(id) = &chunk.id { + if self.response_id.is_empty() { + self.response_id.clone_from(id); + } + } + if let Some(model) = &chunk.model { + if self.response_model.is_empty() { + self.response_model.clone_from(model); + } + } + + // Capture usage if present (often in a dedicated chunk). + if let Some(usage) = &chunk.usage { + self.usage = TokenCounts { + input_tokens: usage.prompt_tokens, + output_tokens: usage.completion_tokens, + ..TokenCounts::default() + }; + } + + let choices = chunk.choices.as_ref()?; + let choice = choices.first()?; + + let mut events = Vec::new(); + + // Check for finish_reason. + if let Some(reason) = &choice.finish_reason { + self.finish_reason = map_finish_reason(Some(reason.as_str())); + } + + let delta = choice.delta.as_ref()?; + + // Accumulate reasoning/thinking content (Kimi, etc.). + if let Some(reasoning) = &delta.reasoning_content { + if !reasoning.is_empty() { + self.accumulated_reasoning.push_str(reasoning); + } + } + + // Handle text content delta. + if let Some(content) = &delta.content { + if !content.is_empty() { + if !self.text_started { + self.text_started = true; + events.push(StreamEvent::TextStart { text_id: None }); + } + self.accumulated_text.push_str(content); + events.push(StreamEvent::text_delta(content, None)); + } + } + + // Handle tool call deltas. + if let Some(tool_calls) = &delta.tool_calls { + for tc in tool_calls { + let index = tc.index; + + // Grow the accumulated tool calls vector if needed. + while self.tool_calls.len() <= index { + self.tool_calls.push(AccumulatedToolCall { + id: String::new(), + name: String::new(), + arguments: String::new(), + started: false, + }); + } + + let accumulated = &mut self.tool_calls[index]; + + // First chunk for this tool call carries id and name. + if let Some(id) = &tc.id { + accumulated.id.clone_from(id); + } + if let Some(func) = &tc.function { + if let Some(name) = &func.name { + accumulated.name.clone_from(name); + } + if let Some(args) = &func.arguments { + accumulated.arguments.push_str(args); + } + } + + let partial_tool_call = + ToolCall::new(&accumulated.id, &accumulated.name, serde_json::json!(null)); + + if accumulated.started { + events.push(StreamEvent::ToolCallDelta { + tool_call: partial_tool_call, + }); + } else { + accumulated.started = true; + events.push(StreamEvent::ToolCallStart { + tool_call: partial_tool_call, + }); + } + } + } + + if events.is_empty() { + None + } else { + Some(events) + } + } + + /// Generate the final events when `[DONE]` (or end-of-stream) is received. + fn finish_events(&mut self) -> Vec { + self.finished = true; + let mut events = Vec::new(); + + // End text segment if it was started. + if self.text_started { + events.push(StreamEvent::TextEnd { text_id: None }); + } + + let mut content_parts = Vec::new(); + + // Include reasoning/thinking content if present (Kimi, etc.). + if !self.accumulated_reasoning.is_empty() { + content_parts.push(ContentPart::Thinking(ThinkingData { + text: std::mem::take(&mut self.accumulated_reasoning), + signature: None, + redacted: false, + })); + } + + if !self.accumulated_text.is_empty() { + content_parts.push(ContentPart::text(&self.accumulated_text)); + } + + for accumulated in &self.tool_calls { + let arguments = parse_tool_arguments( + &accumulated.name, + &accumulated.arguments, + &self.custom_tool_names, + ); + let mut tool_call = ToolCall::new(&accumulated.id, &accumulated.name, arguments); + tool_call.raw_arguments = Some(accumulated.arguments.clone()); + + events.push(StreamEvent::ToolCallEnd { + tool_call: tool_call.clone(), + }); + content_parts.push(ContentPart::ToolCall(tool_call)); + } + + // Infer finish reason from tool calls if not explicitly set. + if !self.tool_calls.is_empty() && self.finish_reason == FinishReason::Stop { + self.finish_reason = FinishReason::ToolCalls; + } + + let response_model = if self.response_model.is_empty() { + self.model.clone() + } else { + self.response_model.clone() + }; + + let response = Response { + id: self.response_id.clone(), + model: response_model, + provider: self.provider_name.clone(), + message: Message { + role: Role::Assistant, + content: content_parts, + name: None, + tool_call_id: None, + }, + finish_reason: self.finish_reason.clone(), + usage: self.usage.clone(), + raw: None, + warnings: vec![], + rate_limit: self.rate_limit.clone(), + }; + + events.push(StreamEvent::finish( + self.finish_reason.clone(), + self.usage.clone(), + response, + )); + + events + } +} + +impl StreamDecoder for StreamState { + fn on_event(&mut self, ev: RawEvent<'_>) -> Result, Error> { + // Chat Completions uses data-only framing; the `event:` field is unused. + if ev.data == "[DONE]" { + return Ok(self.finish_events()); + } + + let chunk: StreamChunk = serde_json::from_str(ev.data) + .map_err(|e| Error::stream_error(format!("failed to parse SSE chunk: {e}"), e))?; + + Ok(self.process_chunk(&chunk).unwrap_or_default()) + } + + fn finish(&mut self) -> Vec { + // Stream ended without `[DONE]`. Some providers (e.g. Minimax) omit the + // sentinel; emit accumulated finish events if we have content and + // haven't already finished. + if !self.finished && (self.text_started || !self.tool_calls.is_empty()) { + self.finish_events() + } else { + Vec::new() + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::codec::CodecParams; + use crate::types::Request; + + /// Build a decoder through `StreamState::new` (with a minimal request) for + /// unit tests that drive `process_chunk` / `finish_events`. + fn test_state(provider: &str, model: &str) -> StreamState { + let request = Request { + model: model.to_string(), + messages: Vec::new(), + provider: None, + tools: None, + tool_choice: None, + response_format: None, + temperature: None, + top_p: None, + max_tokens: None, + stop_sequences: None, + reasoning_effort: None, + speed: None, + metadata: None, + provider_options: None, + }; + let params = CodecParams; + let ctx = CodecCtx { + request: &request, + provider_name: provider, + deployment_id: model, + model: None, + params: ¶ms, + }; + StreamState::new(&ctx, None) + } + + #[test] + fn stream_chunk_minimax_format() { + let json = r#"{"id":"abc","choices":[{"index":0,"delta":{"content":"hello","role":"assistant","name":"MiniMax AI","audio_content":""}}],"created":1772268546,"model":"MiniMax-M2.5","object":"chat.completion.chunk","usage":null,"input_sensitive":false,"output_sensitive":false}"#; + let chunk: StreamChunk = serde_json::from_str(json).unwrap(); + let choices = chunk.choices.unwrap(); + let delta = choices[0].delta.as_ref().unwrap(); + assert_eq!(delta.content.as_deref(), Some("hello")); + } + + #[test] + fn stream_chunk_text_delta_parsing() { + let json = r#"{"id":"chatcmpl-1","model":"gpt-4","choices":[{"delta":{"content":"Hello"},"finish_reason":null}]}"#; + let chunk: StreamChunk = serde_json::from_str(json).unwrap(); + assert_eq!(chunk.id.as_deref(), Some("chatcmpl-1")); + assert_eq!(chunk.model.as_deref(), Some("gpt-4")); + let choices = chunk.choices.unwrap(); + assert_eq!(choices.len(), 1); + let delta = choices[0].delta.as_ref().unwrap(); + assert_eq!(delta.content.as_deref(), Some("Hello")); + assert!(choices[0].finish_reason.is_none()); + } + + #[test] + fn stream_chunk_tool_call_parsing() { + let json = r#"{"id":"chatcmpl-1","model":"gpt-4","choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"get_weather","arguments":"{\"ci"}}]},"finish_reason":null}]}"#; + let chunk: StreamChunk = serde_json::from_str(json).unwrap(); + let choices = chunk.choices.unwrap(); + let delta = choices[0].delta.as_ref().unwrap(); + let tc = &delta.tool_calls.as_ref().unwrap()[0]; + assert_eq!(tc.index, 0); + assert_eq!(tc.id.as_deref(), Some("call_1")); + let func = tc.function.as_ref().unwrap(); + assert_eq!(func.name.as_deref(), Some("get_weather")); + assert_eq!(func.arguments.as_deref(), Some("{\"ci")); + } + + #[test] + fn stream_chunk_usage_parsing() { + let json = r#"{"id":"chatcmpl-1","model":"gpt-4","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":20,"total_tokens":30}}"#; + let chunk: StreamChunk = serde_json::from_str(json).unwrap(); + let usage = chunk.usage.unwrap(); + assert_eq!(usage.prompt_tokens, 10); + assert_eq!(usage.completion_tokens, 20); + } + + #[test] + fn stream_chunk_finish_reason_parsing() { + let json = r#"{"id":"chatcmpl-1","model":"gpt-4","choices":[{"delta":{},"finish_reason":"stop"}]}"#; + let chunk: StreamChunk = serde_json::from_str(json).unwrap(); + let choices = chunk.choices.unwrap(); + assert_eq!(choices[0].finish_reason.as_deref(), Some("stop")); + } + + #[test] + fn process_text_chunks() { + let mut state = test_state("test", "model"); + + let chunk1: StreamChunk = serde_json::from_str( + r#"{"id":"c1","model":"m1","choices":[{"delta":{"content":"Hello"},"finish_reason":null}]}"#, + ).unwrap(); + let events1 = state.process_chunk(&chunk1).unwrap(); + assert_eq!(events1.len(), 2); + assert!(matches!(events1[0], StreamEvent::TextStart { .. })); + assert!(matches!(events1[1], StreamEvent::TextDelta { .. })); + + let chunk2: StreamChunk = serde_json::from_str( + r#"{"id":"c1","model":"m1","choices":[{"delta":{"content":" world"},"finish_reason":null}]}"#, + ).unwrap(); + let events2 = state.process_chunk(&chunk2).unwrap(); + assert_eq!(events2.len(), 1); + assert!(matches!(events2[0], StreamEvent::TextDelta { .. })); + + assert_eq!(state.accumulated_text, "Hello world"); + } + + #[test] + fn process_tool_call_chunks() { + let mut state = test_state("test", "model"); + + let chunk1: StreamChunk = serde_json::from_str( + r#"{"id":"c1","model":"m1","choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"fn1","arguments":"{\"k"}}]},"finish_reason":null}]}"#, + ).unwrap(); + let events1 = state.process_chunk(&chunk1).unwrap(); + assert_eq!(events1.len(), 1); + assert!(matches!(events1[0], StreamEvent::ToolCallStart { .. })); + + let chunk2: StreamChunk = serde_json::from_str( + r#"{"id":"c1","model":"m1","choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ey\"}"}}]},"finish_reason":null}]}"#, + ).unwrap(); + let events2 = state.process_chunk(&chunk2).unwrap(); + assert_eq!(events2.len(), 1); + assert!(matches!(events2[0], StreamEvent::ToolCallDelta { .. })); + + assert_eq!(state.tool_calls[0].arguments, r#"{"key"}"#); + } + + #[test] + fn finish_events_text_only() { + let mut state = test_state("test-provider", "test-model"); + state.response_id = "resp-1".into(); + state.response_model = "gpt-4".into(); + state.accumulated_text = "Hello world".into(); + state.text_started = true; + state.usage = TokenCounts { + input_tokens: 5, + output_tokens: 10, + ..TokenCounts::default() + }; + + let events = state.finish_events(); + assert_eq!(events.len(), 2); + assert!(matches!(events[0], StreamEvent::TextEnd { .. })); + match &events[1] { + StreamEvent::Finish { + finish_reason, + usage, + response, + } => { + assert_eq!(*finish_reason, FinishReason::Stop); + assert_eq!(usage.input_tokens, 5); + assert_eq!(usage.output_tokens, 10); + assert_eq!(response.text(), "Hello world"); + assert_eq!(response.id, "resp-1"); + assert_eq!(response.model, "gpt-4"); + assert_eq!(response.provider, "test-provider"); + } + other => panic!("Expected Finish, got {other:?}"), + } + } + + #[test] + fn finish_events_with_tool_calls() { + let mut state = test_state("test", "model"); + state.response_id = "resp-1".into(); + state.tool_calls.push(AccumulatedToolCall { + id: "call_1".into(), + name: "get_weather".into(), + arguments: r#"{"city":"SF"}"#.into(), + started: true, + }); + + let events = state.finish_events(); + assert_eq!(events.len(), 2); + match &events[0] { + StreamEvent::ToolCallEnd { tool_call } => { + assert_eq!(tool_call.id, "call_1"); + assert_eq!(tool_call.name, "get_weather"); + assert_eq!(tool_call.raw_arguments.as_deref(), Some(r#"{"city":"SF"}"#)); + } + other => panic!("Expected ToolCallEnd, got {other:?}"), + } + match &events[1] { + StreamEvent::Finish { + finish_reason, + response, + .. + } => { + assert_eq!(*finish_reason, FinishReason::ToolCalls); + let calls = response.tool_calls(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].name, "get_weather"); + } + other => panic!("Expected Finish, got {other:?}"), + } + } + + #[test] + fn uses_request_model_as_fallback() { + let mut state = test_state("test", "fallback-model"); + let events = state.finish_events(); + match &events[0] { + StreamEvent::Finish { response, .. } => { + assert_eq!(response.model, "fallback-model"); + } + other => panic!("Expected Finish, got {other:?}"), + } + } +} diff --git a/lib/crates/fabro-llm/src/codec/openai_compatible/translate.rs b/lib/crates/fabro-llm/src/codec/openai_compatible/translate.rs new file mode 100644 index 000000000..d2e5e3422 --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_compatible/translate.rs @@ -0,0 +1,411 @@ +//! Pure mapping between canonical types and the Chat Completions wire shapes. + +use super::wire::{ChatFunction, ChatMessage, ChatToolCall}; +use crate::types::{ + ContentPart, FinishReason, Message, Request, ResponseFormat, ResponseFormatType, Role, + ToolChoice, ToolDefinition, +}; + +pub(super) fn map_finish_reason(reason: Option<&str>) -> FinishReason { + match reason { + Some("stop") | None => FinishReason::Stop, + Some("length") => FinishReason::Length, + Some("tool_calls") => FinishReason::ToolCalls, + Some("content_filter") => FinishReason::ContentFilter, + Some(other) => FinishReason::Other(other.to_string()), + } +} + +/// Build the content string from a message's parts, including fallback text +/// for unsupported content types (Audio, Document). +fn content_text_with_fallbacks(parts: &[ContentPart]) -> String { + let mut segments: Vec = Vec::new(); + for part in parts { + match part { + ContentPart::Text(text) => segments.push(text.clone()), + ContentPart::Audio(_) => { + segments.push("[Audio content not supported by this provider]".to_string()); + } + ContentPart::Document(doc) => { + let desc = doc.file_name.as_ref().map_or_else( + || "[Document content not supported by this provider]".to_string(), + |name| { + format!("[Document '{name}': content type not supported by this provider]") + }, + ); + segments.push(desc); + } + _ => {} + } + } + segments.join("") +} + +pub(super) fn translate_messages(messages: &[Message]) -> Vec { + messages + .iter() + .flat_map(|msg| { + // Tool messages must be split into one ChatMessage per ToolResult, + // each with its own tool_call_id. The Chat Completions API requires + // every tool_call_id from the assistant to have a matching tool message. + if msg.role == Role::Tool { + return msg + .content + .iter() + .filter_map(|part| { + if let ContentPart::ToolResult(tr) = part { + let output = tr + .content + .as_str() + .map_or_else(|| tr.content.to_string(), str::to_string); + Some(ChatMessage { + role: "tool".to_string(), + content: Some(output), + reasoning_content: None, + tool_call_id: Some(tr.tool_call_id.clone()), + tool_calls: None, + }) + } else { + None + } + }) + .collect::>(); + } + + let role = match msg.role { + Role::System | Role::Developer => "system", + Role::User => "user", + Role::Assistant => "assistant", + Role::Tool => unreachable!( + "Role::Tool is handled in the early-return branch above this match" + ), + }; + + let mut tool_calls: Vec = Vec::new(); + if msg.role == Role::Assistant { + for part in &msg.content { + if let ContentPart::ToolCall(tc) = part { + let arguments = tc + .raw_arguments + .clone() + .unwrap_or_else(|| tc.arguments.to_string()); + tool_calls.push(ChatToolCall { + id: tc.id.clone(), + kind: "function".to_string(), + function: ChatFunction { + name: tc.name.clone(), + arguments, + }, + }); + } + } + } + + let text = content_text_with_fallbacks(&msg.content); + let content = if text.is_empty() { None } else { Some(text) }; + let tool_calls = if tool_calls.is_empty() { + None + } else { + Some(tool_calls) + }; + + // Extract reasoning/thinking content for assistant messages. + let reasoning_content = if msg.role == Role::Assistant { + let reasoning: String = msg + .content + .iter() + .filter_map(|part| match part { + ContentPart::Thinking(t) if !t.redacted => Some(t.text.as_str()), + _ => None, + }) + .collect::>() + .join(""); + if reasoning.is_empty() { + None + } else { + Some(reasoning) + } + } else { + None + }; + + vec![ChatMessage { + role: role.to_string(), + content, + reasoning_content, + tool_call_id: msg.tool_call_id.clone(), + tool_calls, + }] + }) + .collect() +} + +pub(super) fn translate_tools(tools: &[ToolDefinition]) -> Vec { + tools + .iter() + .map(|t| { + serde_json::json!({ + "type": "function", + "function": { + "name": t.name, + "description": t.description, + "parameters": t.parameters, + } + }) + }) + .collect() +} + +pub(super) fn translate_tool_choice(choice: &ToolChoice) -> serde_json::Value { + match choice { + ToolChoice::Auto => serde_json::json!("auto"), + ToolChoice::None => serde_json::json!("none"), + ToolChoice::Required => serde_json::json!("required"), + ToolChoice::Named { tool_name } => { + serde_json::json!({"type": "function", "function": {"name": tool_name}}) + } + } +} + +pub(super) fn custom_tool_names(request: &Request) -> Vec { + request + .tools + .as_deref() + .unwrap_or_default() + .iter() + .filter(|tool| tool.is_custom()) + .map(|tool| tool.name.clone()) + .collect() +} + +pub(super) fn parse_tool_arguments( + tool_name: &str, + raw_arguments: &str, + custom_tool_names: &[String], +) -> serde_json::Value { + match serde_json::from_str(raw_arguments) { + Ok(arguments) => arguments, + Err(_) if custom_tool_names.iter().any(|name| name == tool_name) => { + serde_json::Value::String(raw_arguments.to_string()) + } + Err(_) => serde_json::json!({}), + } +} + +/// Translate unified `ResponseFormat` to Chat Completions `response_format`. +pub(super) fn translate_response_format(format: &ResponseFormat) -> serde_json::Value { + match format.kind { + ResponseFormatType::Text => serde_json::json!({"type": "text"}), + ResponseFormatType::JsonObject => serde_json::json!({"type": "json_object"}), + ResponseFormatType::JsonSchema => { + let mut json_schema = serde_json::json!({ + "name": "response", + "strict": format.strict, + }); + if let Some(schema) = &format.json_schema { + json_schema["schema"] = schema.clone(); + } + serde_json::json!({ + "type": "json_schema", + "json_schema": json_schema, + }) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::types::{AudioData, ContentPart, DocumentData, Message, Role, ToolCall}; + + #[test] + fn translate_assistant_message_with_tool_calls_only() { + let msg = Message { + role: Role::Assistant, + content: vec![ContentPart::ToolCall(ToolCall::new( + "call_1", + "get_weather", + serde_json::json!({"city": "SF"}), + ))], + name: None, + tool_call_id: None, + }; + let translated = translate_messages(&[msg]); + assert_eq!(translated.len(), 1); + assert_eq!(translated[0].role, "assistant"); + assert!(translated[0].content.is_none()); + let tool_calls = translated[0].tool_calls.as_ref().unwrap(); + assert_eq!(tool_calls.len(), 1); + assert_eq!(tool_calls[0].id, "call_1"); + assert_eq!(tool_calls[0].kind, "function"); + assert_eq!(tool_calls[0].function.name, "get_weather"); + assert_eq!(tool_calls[0].function.arguments, r#"{"city":"SF"}"#); + } + + #[test] + fn translate_assistant_message_with_text_and_tool_calls() { + let msg = Message { + role: Role::Assistant, + content: vec![ + ContentPart::text("Let me check the weather"), + ContentPart::ToolCall(ToolCall::new( + "call_2", + "get_weather", + serde_json::json!({"city": "NYC"}), + )), + ], + name: None, + tool_call_id: None, + }; + let translated = translate_messages(&[msg]); + assert_eq!( + translated[0].content.as_deref(), + Some("Let me check the weather") + ); + let tool_calls = translated[0].tool_calls.as_ref().unwrap(); + assert_eq!(tool_calls.len(), 1); + assert_eq!(tool_calls[0].function.name, "get_weather"); + } + + #[test] + fn translate_assistant_message_with_raw_arguments() { + let mut tc = ToolCall::new("call_3", "search", serde_json::json!({"q": "rust"})); + tc.raw_arguments = Some(r#"{"q": "rust"}"#.to_string()); + let msg = Message { + role: Role::Assistant, + content: vec![ContentPart::ToolCall(tc)], + name: None, + tool_call_id: None, + }; + let translated = translate_messages(&[msg]); + let tool_calls = translated[0].tool_calls.as_ref().unwrap(); + // Should prefer raw_arguments over serializing arguments + assert_eq!(tool_calls[0].function.arguments, r#"{"q": "rust"}"#); + } + + #[test] + fn translate_tool_message_has_tool_call_id() { + let msg = Message::tool_result( + "call_1", + serde_json::Value::String("72F and sunny".into()), + false, + ); + let translated = translate_messages(&[msg]); + assert_eq!(translated[0].role, "tool"); + assert_eq!(translated[0].tool_call_id.as_deref(), Some("call_1")); + assert!(translated[0].tool_calls.is_none()); + } + + #[test] + fn translate_user_message_has_no_tool_calls() { + let msg = Message::user("Hello"); + let translated = translate_messages(&[msg]); + assert_eq!(translated[0].role, "user"); + assert_eq!(translated[0].content.as_deref(), Some("Hello")); + assert!(translated[0].tool_calls.is_none()); + } + + #[test] + fn assistant_tool_calls_serialize_correctly() { + let msg = Message { + role: Role::Assistant, + content: vec![ContentPart::ToolCall(ToolCall::new( + "call_1", + "get_weather", + serde_json::json!({"city": "SF"}), + ))], + name: None, + tool_call_id: None, + }; + let translated = translate_messages(&[msg]); + let json = serde_json::to_value(&translated[0]).unwrap(); + assert!(json.get("content").is_none()); + assert!(json.get("tool_call_id").is_none()); + let tool_calls = json["tool_calls"].as_array().unwrap(); + assert_eq!(tool_calls.len(), 1); + assert_eq!(tool_calls[0]["type"], "function"); + assert_eq!(tool_calls[0]["id"], "call_1"); + assert_eq!(tool_calls[0]["function"]["name"], "get_weather"); + } + + #[test] + fn audio_content_produces_text_fallback() { + let msg = Message { + role: Role::User, + content: vec![ContentPart::Audio(AudioData { + url: Some("https://example.com/audio.wav".to_string()), + data: None, + media_type: None, + })], + name: None, + tool_call_id: None, + }; + let translated = translate_messages(&[msg]); + assert_eq!( + translated[0].content.as_deref(), + Some("[Audio content not supported by this provider]") + ); + } + + #[test] + fn document_content_produces_text_fallback_with_filename() { + let msg = Message { + role: Role::User, + content: vec![ContentPart::Document(DocumentData { + url: Some("https://example.com/doc.pdf".to_string()), + data: None, + media_type: None, + file_name: Some("report.pdf".to_string()), + })], + name: None, + tool_call_id: None, + }; + let translated = translate_messages(&[msg]); + assert_eq!( + translated[0].content.as_deref(), + Some("[Document 'report.pdf': content type not supported by this provider]") + ); + } + + #[test] + fn document_content_produces_text_fallback_without_filename() { + let msg = Message { + role: Role::User, + content: vec![ContentPart::Document(DocumentData { + url: None, + data: Some(vec![1, 2, 3]), + media_type: None, + file_name: None, + })], + name: None, + tool_call_id: None, + }; + let translated = translate_messages(&[msg]); + assert_eq!( + translated[0].content.as_deref(), + Some("[Document content not supported by this provider]") + ); + } + + #[test] + fn mixed_text_and_audio_content_concatenates() { + let msg = Message { + role: Role::User, + content: vec![ + ContentPart::text("Check this: "), + ContentPart::Audio(AudioData { + url: None, + data: Some(vec![1, 2]), + media_type: None, + }), + ], + name: None, + tool_call_id: None, + }; + let translated = translate_messages(&[msg]); + assert_eq!( + translated[0].content.as_deref(), + Some("Check this: [Audio content not supported by this provider]") + ); + } +} diff --git a/lib/crates/fabro-llm/src/codec/openai_compatible/wire.rs b/lib/crates/fabro-llm/src/codec/openai_compatible/wire.rs new file mode 100644 index 000000000..fd4750f0a --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_compatible/wire.rs @@ -0,0 +1,143 @@ +//! Serde types mirroring the OpenAI Chat Completions wire shapes. + +#[derive(serde::Serialize)] +pub(super) struct ApiRequest { + pub model: String, + pub messages: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub temperature: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_tokens: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub top_p: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub stop: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub tools: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_choice: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub response_format: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub stream: Option, +} + +#[derive(serde::Serialize)] +pub(super) struct ChatMessage { + pub role: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub content: Option, + /// Reasoning/thinking content echoed back for providers that require it + /// (Kimi). + #[serde(skip_serializing_if = "Option::is_none")] + pub reasoning_content: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_call_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_calls: Option>, +} + +#[derive(serde::Serialize)] +pub(super) struct ChatToolCall { + pub id: String, + #[serde(rename = "type")] + pub kind: String, + pub function: ChatFunction, +} + +#[derive(serde::Serialize)] +pub(super) struct ChatFunction { + pub name: String, + pub arguments: String, +} + +// --- Response types (non-streaming) --- + +#[derive(serde::Deserialize)] +pub(super) struct ApiResponse { + pub id: String, + pub model: String, + pub choices: Vec, + pub usage: Option, +} + +#[derive(serde::Deserialize)] +pub(super) struct ApiChoice { + pub message: ApiChoiceMessage, + pub finish_reason: Option, +} + +#[derive(serde::Deserialize)] +pub(super) struct ApiChoiceMessage { + pub content: Option, + pub reasoning_content: Option, + pub tool_calls: Option>, +} + +#[derive(serde::Deserialize)] +pub(super) struct ApiToolCall { + pub id: String, + pub function: ApiFunction, +} + +#[derive(serde::Deserialize)] +pub(super) struct ApiFunction { + pub name: String, + pub arguments: String, +} + +#[derive(serde::Deserialize)] +#[allow( + clippy::struct_field_names, + reason = "Field names mirror the provider API payload." +)] +pub(super) struct ApiUsage { + pub prompt_tokens: i64, + pub completion_tokens: i64, +} + +// --- Streaming response types --- + +#[derive(serde::Deserialize)] +pub(super) struct StreamChunk { + pub id: Option, + pub model: Option, + pub choices: Option>, + pub usage: Option, +} + +#[derive(serde::Deserialize)] +pub(super) struct StreamChoice { + pub delta: Option, + pub finish_reason: Option, +} + +#[derive(serde::Deserialize)] +pub(super) struct StreamDelta { + pub content: Option, + /// Reasoning/thinking content (used by Kimi and other reasoning models). + pub reasoning_content: Option, + pub tool_calls: Option>, +} + +#[derive(serde::Deserialize)] +pub(super) struct StreamToolCall { + pub index: usize, + pub id: Option, + pub function: Option, +} + +#[derive(serde::Deserialize)] +pub(super) struct StreamFunction { + pub name: Option, + pub arguments: Option, +} + +// --- Accumulated tool call state for streaming --- + +pub(super) struct AccumulatedToolCall { + pub id: String, + pub name: String, + pub arguments: String, + pub started: bool, +} diff --git a/lib/crates/fabro-llm/src/lib.rs b/lib/crates/fabro-llm/src/lib.rs index 63066b622..e1f1c9b01 100644 --- a/lib/crates/fabro-llm/src/lib.rs +++ b/lib/crates/fabro-llm/src/lib.rs @@ -1,5 +1,6 @@ pub mod adapter_registry; pub mod client; +mod codec; pub mod error; pub mod generate; pub mod middleware; diff --git a/lib/crates/fabro-llm/src/providers/openai_compatible.rs b/lib/crates/fabro-llm/src/providers/openai_compatible.rs index 3b784dc04..3ba873e11 100644 --- a/lib/crates/fabro-llm/src/providers/openai_compatible.rs +++ b/lib/crates/fabro-llm/src/providers/openai_compatible.rs @@ -1,21 +1,19 @@ +use std::collections::VecDeque; use std::sync::Arc; use fabro_model::Catalog; -use futures::{StreamExt, stream}; +use futures::stream; -use crate::error::{Error, ProviderErrorDetail, ProviderErrorKind, error_from_status_code}; +use crate::codec::openai_compatible::OpenAiCompatible; +use crate::codec::{Codec, CodecCtx, CodecParams, RawEvent, StreamDecoder}; +use crate::error::Error; use crate::provider::{ ProviderAdapter, StreamEventStream, validate_standard_speed, validate_tool_choice, }; use crate::providers::common::{ - api_model_id, parse_error_body, parse_rate_limit_headers, parse_retry_after, - send_and_read_response, -}; -use crate::types::{ - AdapterTimeout, ContentPart, FinishReason, Message, RateLimitInfo, Request, Response, - ResponseFormat, ResponseFormatType, Role, StreamEvent, ThinkingData, TokenCounts, ToolCall, - ToolChoice, ToolDefinition, + api_model_id, parse_rate_limit_headers, parse_retry_after, send_and_read_response, }; +use crate::types::{AdapterTimeout, Request, Response, StreamEvent}; /// `OpenAI`-compatible Chat Completions adapter (Section 7.10). /// @@ -24,6 +22,10 @@ use crate::types::{ /// /// Does NOT support reasoning tokens, built-in tools, or other Responses API /// features. Use the primary `OpenAiAdapter` for `OpenAI`'s own API. +/// +/// This is a thin transport shell over the `openai_compatible` codec: it owns +/// auth, base URL, and the streaming byte loop, and delegates all wire +/// translation to the codec. pub struct Adapter { pub(crate) http: super::http_api::HttpApi, provider_name: String, @@ -85,428 +87,61 @@ impl Adapter { } req } -} -// --- Request types (Chat Completions format) --- - -#[derive(serde::Serialize)] -struct ApiRequest { - model: String, - messages: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - temperature: Option, - #[serde(skip_serializing_if = "Option::is_none")] - max_tokens: Option, - #[serde(skip_serializing_if = "Option::is_none")] - top_p: Option, - #[serde(skip_serializing_if = "Option::is_none")] - stop: Option>, - #[serde(skip_serializing_if = "Option::is_none")] - tools: Option>, - #[serde(skip_serializing_if = "Option::is_none")] - tool_choice: Option, - #[serde(skip_serializing_if = "Option::is_none")] - response_format: Option, - #[serde(skip_serializing_if = "Option::is_none")] - stream: Option, -} - -#[derive(serde::Serialize)] -struct ChatMessage { - role: String, - #[serde(skip_serializing_if = "Option::is_none")] - content: Option, - /// Reasoning/thinking content echoed back for providers that require it - /// (Kimi). - #[serde(skip_serializing_if = "Option::is_none")] - reasoning_content: Option, - #[serde(skip_serializing_if = "Option::is_none")] - tool_call_id: Option, - #[serde(skip_serializing_if = "Option::is_none")] - tool_calls: Option>, -} - -#[derive(serde::Serialize)] -struct ChatToolCall { - id: String, - #[serde(rename = "type")] - kind: String, - function: ChatFunction, -} - -#[derive(serde::Serialize)] -struct ChatFunction { - name: String, - arguments: String, -} - -// --- Response types (non-streaming) --- - -#[derive(serde::Deserialize)] -struct ApiResponse { - id: String, - model: String, - choices: Vec, - usage: Option, -} - -#[derive(serde::Deserialize)] -struct ApiChoice { - message: ApiChoiceMessage, - finish_reason: Option, -} - -#[derive(serde::Deserialize)] -struct ApiChoiceMessage { - content: Option, - reasoning_content: Option, - tool_calls: Option>, -} - -#[derive(serde::Deserialize)] -struct ApiToolCall { - id: String, - function: ApiFunction, -} - -#[derive(serde::Deserialize)] -struct ApiFunction { - name: String, - arguments: String, -} - -#[derive(serde::Deserialize)] -#[allow( - clippy::struct_field_names, - reason = "Field names mirror the provider API payload." -)] -struct ApiUsage { - prompt_tokens: i64, - completion_tokens: i64, -} - -// --- Streaming response types --- - -#[derive(serde::Deserialize)] -struct StreamChunk { - id: Option, - model: Option, - choices: Option>, - usage: Option, -} - -#[derive(serde::Deserialize)] -struct StreamChoice { - delta: Option, - finish_reason: Option, -} - -#[derive(serde::Deserialize)] -struct StreamDelta { - content: Option, - /// Reasoning/thinking content (used by Kimi and other reasoning models). - reasoning_content: Option, - tool_calls: Option>, -} - -#[derive(serde::Deserialize)] -struct StreamToolCall { - index: usize, - id: Option, - function: Option, -} - -#[derive(serde::Deserialize)] -struct StreamFunction { - name: Option, - arguments: Option, -} - -// --- Accumulated tool call state for streaming --- - -struct AccumulatedToolCall { - id: String, - name: String, - arguments: String, - started: bool, -} - -fn map_finish_reason(reason: Option<&str>) -> FinishReason { - match reason { - Some("stop") | None => FinishReason::Stop, - Some("length") => FinishReason::Length, - Some("tool_calls") => FinishReason::ToolCalls, - Some("content_filter") => FinishReason::ContentFilter, - Some(other) => FinishReason::Other(other.to_string()), + /// Resolve the wire model id (catalog `api_id`, falling back to the + /// requested model). + fn deployment_id(&self, request: &Request) -> String { + api_model_id(self.catalog.as_deref(), &request.model) } -} -/// Build the content string from a message's parts, including fallback text -/// for unsupported content types (Audio, Document). -fn content_text_with_fallbacks(parts: &[ContentPart]) -> String { - let mut segments: Vec = Vec::new(); - for part in parts { - match part { - ContentPart::Text(text) => segments.push(text.clone()), - ContentPart::Audio(_) => { - segments.push("[Audio content not supported by this provider]".to_string()); - } - ContentPart::Document(doc) => { - let desc = doc.file_name.as_ref().map_or_else( - || "[Document content not supported by this provider]".to_string(), - |name| { - format!("[Document '{name}': content type not supported by this provider]") - }, - ); - segments.push(desc); - } - _ => {} + /// Build the borrowed codec context. `deployment_id` and `params` are + /// created by the caller so their borrows outlive the context. + fn codec_ctx<'a>( + &'a self, + request: &'a Request, + deployment_id: &'a str, + params: &'a CodecParams, + ) -> CodecCtx<'a> { + CodecCtx { + request, + provider_name: &self.provider_name, + deployment_id, + model: None, + params, } } - segments.join("") -} -fn translate_messages(messages: &[Message]) -> Vec { - messages - .iter() - .flat_map(|msg| { - // Tool messages must be split into one ChatMessage per ToolResult, - // each with its own tool_call_id. The Chat Completions API requires - // every tool_call_id from the assistant to have a matching tool message. - if msg.role == Role::Tool { - return msg - .content - .iter() - .filter_map(|part| { - if let ContentPart::ToolResult(tr) = part { - let output = tr - .content - .as_str() - .map_or_else(|| tr.content.to_string(), str::to_string); - Some(ChatMessage { - role: "tool".to_string(), - content: Some(output), - reasoning_content: None, - tool_call_id: Some(tr.tool_call_id.clone()), - tool_calls: None, - }) - } else { - None - } - }) - .collect::>(); - } - - let role = match msg.role { - Role::System | Role::Developer => "system", - Role::User => "user", - Role::Assistant => "assistant", - Role::Tool => unreachable!( - "Role::Tool is handled in the early-return branch above this match" - ), - }; - - let mut tool_calls: Vec = Vec::new(); - if msg.role == Role::Assistant { - for part in &msg.content { - if let ContentPart::ToolCall(tc) = part { - let arguments = tc - .raw_arguments - .clone() - .unwrap_or_else(|| tc.arguments.to_string()); - tool_calls.push(ChatToolCall { - id: tc.id.clone(), - kind: "function".to_string(), - function: ChatFunction { - name: tc.name.clone(), - arguments, - }, - }); - } - } - } - - let text = content_text_with_fallbacks(&msg.content); - let content = if text.is_empty() { None } else { Some(text) }; - let tool_calls = if tool_calls.is_empty() { - None - } else { - Some(tool_calls) - }; - - // Extract reasoning/thinking content for assistant messages. - let reasoning_content = if msg.role == Role::Assistant { - let reasoning: String = msg - .content - .iter() - .filter_map(|part| match part { - ContentPart::Thinking(t) if !t.redacted => Some(t.text.as_str()), - _ => None, - }) - .collect::>() - .join(""); - if reasoning.is_empty() { - None - } else { - Some(reasoning) - } - } else { - None - }; - - vec![ChatMessage { - role: role.to_string(), - content, - reasoning_content, - tool_call_id: msg.tool_call_id.clone(), - tool_calls, - }] - }) - .collect() -} - -fn translate_tools(tools: &[ToolDefinition]) -> Vec { - tools - .iter() - .map(|t| { - serde_json::json!({ - "type": "function", - "function": { - "name": t.name, - "description": t.description, - "parameters": t.parameters, - } - }) - }) - .collect() -} - -fn translate_tool_choice(choice: &ToolChoice) -> serde_json::Value { - match choice { - ToolChoice::Auto => serde_json::json!("auto"), - ToolChoice::None => serde_json::json!("none"), - ToolChoice::Required => serde_json::json!("required"), - ToolChoice::Named { tool_name } => { - serde_json::json!({"type": "function", "function": {"name": tool_name}}) + /// Encode `ctx.request` through the codec and assemble the HTTP request: + /// base URL + codec endpoint, default headers, auth, body, and dialect + /// headers. + fn encoded_request( + &self, + codec: &OpenAiCompatible, + ctx: &CodecCtx<'_>, + stream: bool, + ) -> Result { + let encoded = codec.encode(ctx, stream)?; + let url = format!("{}{}", self.http.base_url, encoded.endpoint); + let mut req = self.build_request(&url).json(&encoded.body); + for (key, value) in &encoded.headers { + req = req.header(key, value); } + Ok(req) } } -fn custom_tool_names(request: &Request) -> Vec { - request - .tools - .as_deref() - .unwrap_or_default() - .iter() - .filter(|tool| tool.is_custom()) - .map(|tool| tool.name.clone()) - .collect() -} - -fn parse_tool_arguments( - tool_name: &str, - raw_arguments: &str, - custom_tool_names: &[String], -) -> serde_json::Value { - match serde_json::from_str(raw_arguments) { - Ok(arguments) => arguments, - Err(_) if custom_tool_names.iter().any(|name| name == tool_name) => { - serde_json::Value::String(raw_arguments.to_string()) - } - Err(_) => serde_json::json!({}), - } -} - -/// Translate unified `ResponseFormat` to Chat Completions `response_format`. -fn translate_response_format(format: &ResponseFormat) -> serde_json::Value { - match format.kind { - ResponseFormatType::Text => serde_json::json!({"type": "text"}), - ResponseFormatType::JsonObject => serde_json::json!({"type": "json_object"}), - ResponseFormatType::JsonSchema => { - let mut json_schema = serde_json::json!({ - "name": "response", - "strict": format.strict, - }); - if let Some(schema) = &format.json_schema { - json_schema["schema"] = schema.clone(); - } - serde_json::json!({ - "type": "json_schema", - "json_schema": json_schema, - }) - } - } -} - -/// Build the API request body from a unified `Request`. -/// -/// Returns a `serde_json::Value` so that `provider_options.` -/// fields can be merged into the request before sending. -#[cfg(test)] -fn build_api_request( - request: &Request, - stream: Option, - provider_name: &str, -) -> serde_json::Value { - build_api_request_with_catalog(request, stream, provider_name, None) -} - -fn build_api_request_with_catalog( - request: &Request, - stream: Option, - provider_name: &str, - catalog: Option<&Catalog>, -) -> serde_json::Value { - let chat_messages = translate_messages(&request.messages); - let tools = request.tools.as_ref().map(|t| translate_tools(t)); - let tool_choice = request.tool_choice.as_ref().map(translate_tool_choice); - let response_format = request - .response_format - .as_ref() - .map(translate_response_format); - - let api_request = ApiRequest { - model: api_model_id(catalog, &request.model), - messages: chat_messages, - temperature: request.temperature, - max_tokens: request.max_tokens, - top_p: request.top_p, - stop: request.stop_sequences.clone(), - tools, - tool_choice, - response_format, - stream, - }; - - let mut body = serde_json::to_value(&api_request).unwrap_or_default(); - merge_provider_options(&mut body, request.provider_options.as_ref(), provider_name); - body -} - -/// Merge `provider_options.` fields into the serialized API -/// request body. -/// -/// The provider name is configurable (e.g. "groq", "together", -/// "openai-compatible"), allowing each instance to have its own namespace in -/// `provider_options`. -fn merge_provider_options( - body: &mut serde_json::Value, - provider_options: Option<&serde_json::Value>, - provider_name: &str, -) { - let Some(opts) = provider_options.and_then(|opts| opts.get(provider_name)) else { - return; - }; - let Some(body_map) = body.as_object_mut() else { - return; - }; - let Some(opts_map) = opts.as_object() else { - return; - }; - - for (key, value) in opts_map { - body_map.insert(key.clone(), value.clone()); - } +/// State driving the streaming byte loop: the codec's decoder plus the line +/// reader, with a small buffer that flattens batched events into individual +/// stream items. +struct StreamLoop { + decoder: Box, + line_reader: super::common::LineReader, + /// Events decoded but not yet yielded. + pending: VecDeque, + /// Byte stream exhausted. + done: bool, + /// `finish()` already drained. + finished_emitted: bool, } #[async_trait::async_trait] @@ -525,102 +160,32 @@ impl ProviderAdapter for Adapter { async fn complete(&self, request: &Request) -> Result { self.validate_request(request)?; - let api_body = build_api_request_with_catalog( - request, - None, - &self.provider_name, - self.catalog.as_deref(), - ); - let url = format!("{}/chat/completions", self.http.base_url); - let mut req = self.build_request(&url).json(&api_body); + let codec = OpenAiCompatible; + let deployment_id = self.deployment_id(request); + let params = CodecParams; + let ctx = self.codec_ctx(request, &deployment_id, ¶ms); + + let mut req = self.encoded_request(&codec, &ctx, false)?; if let Some(t) = self.http.request_timeout { req = req.timeout(t); } + let (body, headers) = send_and_read_response(req, &self.provider_name, "type").await?; - - let api_resp: ApiResponse = serde_json::from_str(&body) - .map_err(|e| Error::network(format!("failed to parse response: {e}"), e))?; - - let choice = api_resp.choices.first().ok_or_else(|| Error::Provider { - kind: ProviderErrorKind::Server, - detail: Box::new(ProviderErrorDetail::new( - "no choices in response", - &self.provider_name, - )), - })?; - - let mut content_parts = Vec::new(); - if let Some(reasoning) = &choice.message.reasoning_content { - if !reasoning.is_empty() { - content_parts.push(ContentPart::Thinking(ThinkingData { - text: reasoning.clone(), - signature: None, - redacted: false, - })); - } - } - if let Some(text) = &choice.message.content { - if !text.is_empty() { - content_parts.push(ContentPart::text(text)); - } - } - if let Some(tool_calls) = &choice.message.tool_calls { - let custom_tool_names = custom_tool_names(request); - for tc in tool_calls { - let arguments = parse_tool_arguments( - &tc.function.name, - &tc.function.arguments, - &custom_tool_names, - ); - let mut tool_call = ToolCall::new(&tc.id, &tc.function.name, arguments); - tool_call.raw_arguments = Some(tc.function.arguments.clone()); - content_parts.push(ContentPart::ToolCall(tool_call)); - } - } - - let finish_reason = map_finish_reason(choice.finish_reason.as_deref()); - - let usage = api_resp - .usage - .as_ref() - .map_or_else(TokenCounts::default, |u| TokenCounts { - input_tokens: u.prompt_tokens, - output_tokens: u.completion_tokens, - ..TokenCounts::default() - }); - - Ok(Response { - id: api_resp.id, - model: api_resp.model, - provider: self.provider_name.clone(), - message: Message { - role: Role::Assistant, - content: content_parts, - name: None, - tool_call_id: None, - }, - finish_reason, - usage, - raw: serde_json::from_str(&body).ok(), - warnings: vec![], - rate_limit: parse_rate_limit_headers(&headers), - }) + let rate_limit = parse_rate_limit_headers(&headers); + codec.decode_response(&body, &ctx, rate_limit) } async fn stream(&self, request: &Request) -> Result { self.validate_request(request)?; - let api_body = build_api_request_with_catalog( - request, - Some(true), - &self.provider_name, - self.catalog.as_deref(), - ); - let url = format!("{}/chat/completions", self.http.base_url); - let http_resp = self - .build_request(&url) - .json(&api_body) + let codec = OpenAiCompatible; + let deployment_id = self.deployment_id(request); + let params = CodecParams; + let ctx = self.codec_ctx(request, &deployment_id, ¶ms); + + let req = self.encoded_request(&codec, &ctx, true)?; + let http_resp = req .send() .await .map_err(|e| Error::network(e.to_string(), e))?; @@ -632,977 +197,61 @@ impl ProviderAdapter for Adapter { .text() .await .map_err(|e| Error::network(e.to_string(), e))?; - let (msg, code, raw) = parse_error_body(&body, "type"); - return Err(error_from_status_code( - status.as_u16(), - msg, - self.provider_name.clone(), - code, - raw, - retry_after, - )); + return Err(codec.decode_error(status.as_u16(), &body, &ctx, retry_after)); } - let provider_name = self.provider_name.clone(); - let model = request.model.clone(); let rate_limit = parse_rate_limit_headers(http_resp.headers()); let stream_read_timeout = self.http.stream_read_timeout; - let custom_tool_names = custom_tool_names(request); + let decoder = codec.stream_decoder(&ctx, rate_limit); + let line_reader = super::common::LineReader::new(http_resp, stream_read_timeout); - let stream = stream::unfold( - StreamState::new( - http_resp, - provider_name, - model, - rate_limit, - stream_read_timeout, - custom_tool_names, - ), + let out = stream::unfold( + StreamLoop { + decoder, + line_reader, + pending: VecDeque::new(), + done: false, + finished_emitted: false, + }, |mut state| async move { loop { - let line = match state.next_line().await { - Ok(Some(line)) => line, - Ok(None) => { - // Stream ended without [DONE]. Some providers - // (e.g. Minimax) omit the sentinel. Emit - // accumulated finish events if we have content - // and haven't already emitted them. - if !state.finished - && (state.text_started || !state.tool_calls.is_empty()) - { - let events = state.finish_events(); - return Some((Ok(events), state)); - } + if let Some(event) = state.pending.pop_front() { + return Some((Ok(event), state)); + } + + if state.done { + if state.finished_emitted { + return None; + } + state.finished_emitted = true; + state.pending = state.decoder.finish().into(); + if state.pending.is_empty() { return None; } - Err(e) => return Some((Err(e), state)), - }; - - let line = line.trim(); - if line.is_empty() || line.starts_with(':') { continue; } - let data = match line.strip_prefix("data:") { - Some(d) => d.trim(), - None => continue, - }; - - if data == "[DONE]" { - let events = state.finish_events(); - return Some((Ok(events), state)); - } - - let chunk: StreamChunk = match serde_json::from_str(data) { - Ok(c) => c, - Err(e) => { - return Some(( - Err(Error::stream_error( - format!("failed to parse SSE chunk: {e}"), - e, - )), - state, - )); + match state.line_reader.read_next_chunk("\n").await { + Ok(Some(line)) => { + let line = line.trim(); + if line.is_empty() || line.starts_with(':') { + continue; + } + let Some(data) = line.strip_prefix("data:").map(str::trim) else { + continue; + }; + match state.decoder.on_event(RawEvent { event: None, data }) { + Ok(events) => state.pending = events.into(), + Err(e) => return Some((Err(e), state)), + } } - }; - - if let Some(events) = state.process_chunk(&chunk) { - return Some((Ok(events), state)); + Ok(None) => state.done = true, + Err(e) => return Some((Err(e), state)), } } }, ); - // Flatten batched events into individual stream events. - let flat_stream = stream::unfold( - FlattenState { - inner: Box::pin(stream), - pending: Vec::new(), - }, - |mut flatten_state| async { - loop { - if let Some(event) = flatten_state.pending.pop() { - return Some((Ok(event), flatten_state)); - } - - match flatten_state.inner.next().await { - Some(Ok(mut events)) => { - // Reverse so we can pop from the end in order. - events.reverse(); - flatten_state.pending = events; - } - Some(Err(e)) => return Some((Err(e), flatten_state)), - None => return None, - } - } - }, - ); - - Ok(Box::pin(flat_stream)) - } -} - -/// State for flattening batched events into individual stream events. -struct FlattenState { - inner: std::pin::Pin, Error>> + Send>>, - pending: Vec, -} - -/// Accumulated state while processing the SSE stream. -struct StreamState { - line_reader: super::common::LineReader, - provider_name: String, - model: String, - response_id: String, - response_model: String, - accumulated_text: String, - accumulated_reasoning: String, - tool_calls: Vec, - usage: TokenCounts, - finish_reason: FinishReason, - text_started: bool, - done: bool, - custom_tool_names: Vec, - /// True after `finish_events()` has been called (guards against - /// duplicates). - finished: bool, - rate_limit: Option, -} - -impl StreamState { - fn new( - response: fabro_http::Response, - provider_name: String, - model: String, - rate_limit: Option, - stream_read_timeout: Option, - custom_tool_names: Vec, - ) -> Self { - Self { - line_reader: super::common::LineReader::new(response, stream_read_timeout), - provider_name, - model, - response_id: String::new(), - response_model: String::new(), - accumulated_text: String::new(), - accumulated_reasoning: String::new(), - tool_calls: Vec::new(), - usage: TokenCounts::default(), - finish_reason: FinishReason::Stop, - text_started: false, - done: false, - custom_tool_names, - finished: false, - rate_limit, - } - } - - /// Read the next complete line from the SSE byte stream. - async fn next_line(&mut self) -> Result, Error> { - if self.done { - return Ok(None); - } - if let Some(line) = self.line_reader.read_next_chunk("\n").await? { - Ok(Some(line)) - } else { - self.done = true; - Ok(None) - } - } - - /// Process a parsed SSE chunk and return events to emit, if any. - fn process_chunk(&mut self, chunk: &StreamChunk) -> Option> { - // Capture response metadata from the first chunk. - if let Some(id) = &chunk.id { - if self.response_id.is_empty() { - self.response_id.clone_from(id); - } - } - if let Some(model) = &chunk.model { - if self.response_model.is_empty() { - self.response_model.clone_from(model); - } - } - - // Capture usage if present (often in a dedicated chunk). - if let Some(usage) = &chunk.usage { - self.usage = TokenCounts { - input_tokens: usage.prompt_tokens, - output_tokens: usage.completion_tokens, - ..TokenCounts::default() - }; - } - - let choices = chunk.choices.as_ref()?; - let choice = choices.first()?; - - let mut events = Vec::new(); - - // Check for finish_reason. - if let Some(reason) = &choice.finish_reason { - self.finish_reason = map_finish_reason(Some(reason.as_str())); - } - - let delta = choice.delta.as_ref()?; - - // Accumulate reasoning/thinking content (Kimi, etc.). - if let Some(reasoning) = &delta.reasoning_content { - if !reasoning.is_empty() { - self.accumulated_reasoning.push_str(reasoning); - } - } - - // Handle text content delta. - if let Some(content) = &delta.content { - if !content.is_empty() { - if !self.text_started { - self.text_started = true; - events.push(StreamEvent::TextStart { text_id: None }); - } - self.accumulated_text.push_str(content); - events.push(StreamEvent::text_delta(content, None)); - } - } - - // Handle tool call deltas. - if let Some(tool_calls) = &delta.tool_calls { - for tc in tool_calls { - let index = tc.index; - - // Grow the accumulated tool calls vector if needed. - while self.tool_calls.len() <= index { - self.tool_calls.push(AccumulatedToolCall { - id: String::new(), - name: String::new(), - arguments: String::new(), - started: false, - }); - } - - let accumulated = &mut self.tool_calls[index]; - - // First chunk for this tool call carries id and name. - if let Some(id) = &tc.id { - accumulated.id.clone_from(id); - } - if let Some(func) = &tc.function { - if let Some(name) = &func.name { - accumulated.name.clone_from(name); - } - if let Some(args) = &func.arguments { - accumulated.arguments.push_str(args); - } - } - - let partial_tool_call = - ToolCall::new(&accumulated.id, &accumulated.name, serde_json::json!(null)); - - if accumulated.started { - events.push(StreamEvent::ToolCallDelta { - tool_call: partial_tool_call, - }); - } else { - accumulated.started = true; - events.push(StreamEvent::ToolCallStart { - tool_call: partial_tool_call, - }); - } - } - } - - if events.is_empty() { - None - } else { - Some(events) - } - } - - /// Generate the final events when `[DONE]` is received. - fn finish_events(&mut self) -> Vec { - self.finished = true; - let mut events = Vec::new(); - - // End text segment if it was started. - if self.text_started { - events.push(StreamEvent::TextEnd { text_id: None }); - } - - // End all tool calls with complete data. - let mut content_parts = Vec::new(); - - // Include reasoning/thinking content if present (Kimi, etc.). - if !self.accumulated_reasoning.is_empty() { - content_parts.push(ContentPart::Thinking(ThinkingData { - text: std::mem::take(&mut self.accumulated_reasoning), - signature: None, - redacted: false, - })); - } - - if !self.accumulated_text.is_empty() { - content_parts.push(ContentPart::text(&self.accumulated_text)); - } - - for accumulated in &self.tool_calls { - let arguments = parse_tool_arguments( - &accumulated.name, - &accumulated.arguments, - &self.custom_tool_names, - ); - let mut tool_call = ToolCall::new(&accumulated.id, &accumulated.name, arguments); - tool_call.raw_arguments = Some(accumulated.arguments.clone()); - - events.push(StreamEvent::ToolCallEnd { - tool_call: tool_call.clone(), - }); - content_parts.push(ContentPart::ToolCall(tool_call)); - } - - // Infer finish reason from tool calls if not explicitly set. - if !self.tool_calls.is_empty() && self.finish_reason == FinishReason::Stop { - self.finish_reason = FinishReason::ToolCalls; - } - - let response_model = if self.response_model.is_empty() { - self.model.clone() - } else { - self.response_model.clone() - }; - - let response = Response { - id: self.response_id.clone(), - model: response_model, - provider: self.provider_name.clone(), - message: Message { - role: Role::Assistant, - content: content_parts, - name: None, - tool_call_id: None, - }, - finish_reason: self.finish_reason.clone(), - usage: self.usage.clone(), - raw: None, - warnings: vec![], - rate_limit: self.rate_limit.clone(), - }; - - events.push(StreamEvent::finish( - self.finish_reason.clone(), - self.usage.clone(), - response, - )); - - events - } -} - -#[cfg(test)] -mod tests { - use fabro_model::catalog::LlmCatalogSettings; - - use super::*; - use crate::types::{AudioData, DocumentData}; - - #[test] - fn stream_chunk_minimax_format() { - let json = r#"{"id":"abc","choices":[{"index":0,"delta":{"content":"hello","role":"assistant","name":"MiniMax AI","audio_content":""}}],"created":1772268546,"model":"MiniMax-M2.5","object":"chat.completion.chunk","usage":null,"input_sensitive":false,"output_sensitive":false}"#; - let chunk: StreamChunk = serde_json::from_str(json).unwrap(); - let choices = chunk.choices.unwrap(); - let delta = choices[0].delta.as_ref().unwrap(); - assert_eq!(delta.content.as_deref(), Some("hello")); - } - - #[test] - fn stream_chunk_text_delta_parsing() { - let json = r#"{"id":"chatcmpl-1","model":"gpt-4","choices":[{"delta":{"content":"Hello"},"finish_reason":null}]}"#; - let chunk: StreamChunk = serde_json::from_str(json).unwrap(); - assert_eq!(chunk.id.as_deref(), Some("chatcmpl-1")); - assert_eq!(chunk.model.as_deref(), Some("gpt-4")); - let choices = chunk.choices.unwrap(); - assert_eq!(choices.len(), 1); - let delta = choices[0].delta.as_ref().unwrap(); - assert_eq!(delta.content.as_deref(), Some("Hello")); - assert!(choices[0].finish_reason.is_none()); - } - - #[test] - fn stream_chunk_tool_call_parsing() { - let json = r#"{"id":"chatcmpl-1","model":"gpt-4","choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"get_weather","arguments":"{\"ci"}}]},"finish_reason":null}]}"#; - let chunk: StreamChunk = serde_json::from_str(json).unwrap(); - let choices = chunk.choices.unwrap(); - let delta = choices[0].delta.as_ref().unwrap(); - let tc = &delta.tool_calls.as_ref().unwrap()[0]; - assert_eq!(tc.index, 0); - assert_eq!(tc.id.as_deref(), Some("call_1")); - let func = tc.function.as_ref().unwrap(); - assert_eq!(func.name.as_deref(), Some("get_weather")); - assert_eq!(func.arguments.as_deref(), Some("{\"ci")); - } - - #[test] - fn stream_chunk_usage_parsing() { - let json = r#"{"id":"chatcmpl-1","model":"gpt-4","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":20,"total_tokens":30}}"#; - let chunk: StreamChunk = serde_json::from_str(json).unwrap(); - let usage = chunk.usage.unwrap(); - assert_eq!(usage.prompt_tokens, 10); - assert_eq!(usage.completion_tokens, 20); - } - - #[test] - fn stream_chunk_finish_reason_parsing() { - let json = r#"{"id":"chatcmpl-1","model":"gpt-4","choices":[{"delta":{},"finish_reason":"stop"}]}"#; - let chunk: StreamChunk = serde_json::from_str(json).unwrap(); - let choices = chunk.choices.unwrap(); - assert_eq!(choices[0].finish_reason.as_deref(), Some("stop")); - } - - #[test] - fn stream_state_process_text_chunks() { - let http_resp = - fabro_http::Response::from(http::Response::builder().status(200).body("").unwrap()); - let mut state = StreamState::new( - http_resp, - "test".into(), - "model".into(), - None, - Some(std::time::Duration::from_secs(30)), - Vec::new(), - ); - - // First text chunk should emit TextStart + TextDelta. - let chunk1: StreamChunk = serde_json::from_str( - r#"{"id":"c1","model":"m1","choices":[{"delta":{"content":"Hello"},"finish_reason":null}]}"#, - ).unwrap(); - let events1 = state.process_chunk(&chunk1).unwrap(); - assert_eq!(events1.len(), 2); - assert!(matches!(events1[0], StreamEvent::TextStart { .. })); - assert!(matches!(events1[1], StreamEvent::TextDelta { .. })); - - // Second text chunk should emit only TextDelta (no second TextStart). - let chunk2: StreamChunk = serde_json::from_str( - r#"{"id":"c1","model":"m1","choices":[{"delta":{"content":" world"},"finish_reason":null}]}"#, - ).unwrap(); - let events2 = state.process_chunk(&chunk2).unwrap(); - assert_eq!(events2.len(), 1); - assert!(matches!(events2[0], StreamEvent::TextDelta { .. })); - - assert_eq!(state.accumulated_text, "Hello world"); - } - - #[test] - fn stream_state_process_tool_call_chunks() { - let http_resp = - fabro_http::Response::from(http::Response::builder().status(200).body("").unwrap()); - let mut state = StreamState::new( - http_resp, - "test".into(), - "model".into(), - None, - Some(std::time::Duration::from_secs(30)), - Vec::new(), - ); - - // First tool call chunk (has id and name) -> ToolCallStart. - let chunk1: StreamChunk = serde_json::from_str( - r#"{"id":"c1","model":"m1","choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"fn1","arguments":"{\"k"}}]},"finish_reason":null}]}"#, - ).unwrap(); - let events1 = state.process_chunk(&chunk1).unwrap(); - assert_eq!(events1.len(), 1); - assert!(matches!(events1[0], StreamEvent::ToolCallStart { .. })); - - // Subsequent chunk (more arguments) -> ToolCallDelta. - let chunk2: StreamChunk = serde_json::from_str( - r#"{"id":"c1","model":"m1","choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ey\"}"}}]},"finish_reason":null}]}"#, - ).unwrap(); - let events2 = state.process_chunk(&chunk2).unwrap(); - assert_eq!(events2.len(), 1); - assert!(matches!(events2[0], StreamEvent::ToolCallDelta { .. })); - - assert_eq!(state.tool_calls[0].arguments, r#"{"key"}"#); - } - - #[test] - fn stream_state_finish_events_text_only() { - let http_resp = - fabro_http::Response::from(http::Response::builder().status(200).body("").unwrap()); - let mut state = StreamState::new( - http_resp, - "test-provider".into(), - "test-model".into(), - None, - Some(std::time::Duration::from_secs(30)), - Vec::new(), - ); - state.response_id = "resp-1".into(); - state.response_model = "gpt-4".into(); - state.accumulated_text = "Hello world".into(); - state.text_started = true; - state.usage = TokenCounts { - input_tokens: 5, - output_tokens: 10, - ..TokenCounts::default() - }; - - let events = state.finish_events(); - // TextEnd + Finish - assert_eq!(events.len(), 2); - assert!(matches!(events[0], StreamEvent::TextEnd { .. })); - match &events[1] { - StreamEvent::Finish { - finish_reason, - usage, - response, - } => { - assert_eq!(*finish_reason, FinishReason::Stop); - assert_eq!(usage.input_tokens, 5); - assert_eq!(usage.output_tokens, 10); - assert_eq!(response.text(), "Hello world"); - assert_eq!(response.id, "resp-1"); - assert_eq!(response.model, "gpt-4"); - assert_eq!(response.provider, "test-provider"); - } - other => panic!("Expected Finish, got {other:?}"), - } - } - - #[test] - fn stream_state_finish_events_with_tool_calls() { - let http_resp = - fabro_http::Response::from(http::Response::builder().status(200).body("").unwrap()); - let mut state = StreamState::new( - http_resp, - "test".into(), - "model".into(), - None, - Some(std::time::Duration::from_secs(30)), - Vec::new(), - ); - state.response_id = "resp-1".into(); - state.tool_calls.push(AccumulatedToolCall { - id: "call_1".into(), - name: "get_weather".into(), - arguments: r#"{"city":"SF"}"#.into(), - started: true, - }); - - let events = state.finish_events(); - // ToolCallEnd + Finish (no TextEnd since text_started is false) - assert_eq!(events.len(), 2); - match &events[0] { - StreamEvent::ToolCallEnd { tool_call } => { - assert_eq!(tool_call.id, "call_1"); - assert_eq!(tool_call.name, "get_weather"); - assert_eq!(tool_call.raw_arguments.as_deref(), Some(r#"{"city":"SF"}"#)); - } - other => panic!("Expected ToolCallEnd, got {other:?}"), - } - match &events[1] { - StreamEvent::Finish { - finish_reason, - response, - .. - } => { - assert_eq!(*finish_reason, FinishReason::ToolCalls); - let calls = response.tool_calls(); - assert_eq!(calls.len(), 1); - assert_eq!(calls[0].name, "get_weather"); - } - other => panic!("Expected Finish, got {other:?}"), - } - } - - #[test] - fn stream_state_uses_request_model_as_fallback() { - let http_resp = - fabro_http::Response::from(http::Response::builder().status(200).body("").unwrap()); - let mut state = StreamState::new( - http_resp, - "test".into(), - "fallback-model".into(), - None, - Some(std::time::Duration::from_secs(30)), - Vec::new(), - ); - // response_model is empty, so finish_events should use the request model. - let events = state.finish_events(); - match &events[0] { - StreamEvent::Finish { response, .. } => { - assert_eq!(response.model, "fallback-model"); - } - other => panic!("Expected Finish, got {other:?}"), - } - } - - #[test] - fn api_request_stream_field_serialization() { - let req = ApiRequest { - model: "test".into(), - messages: vec![], - temperature: None, - max_tokens: None, - top_p: None, - stop: None, - tools: None, - tool_choice: None, - response_format: None, - stream: Some(true), - }; - let json = serde_json::to_value(&req).unwrap(); - assert_eq!(json["stream"], true); - - // When stream is None, it should be omitted. - let req_no_stream = ApiRequest { - model: "test".into(), - messages: vec![], - temperature: None, - max_tokens: None, - top_p: None, - stop: None, - tools: None, - tool_choice: None, - response_format: None, - stream: None, - }; - let json_no_stream = serde_json::to_value(&req_no_stream).unwrap(); - assert!(json_no_stream.get("stream").is_none()); - } - - #[test] - fn translate_assistant_message_with_tool_calls_only() { - let msg = Message { - role: Role::Assistant, - content: vec![ContentPart::ToolCall(ToolCall::new( - "call_1", - "get_weather", - serde_json::json!({"city": "SF"}), - ))], - name: None, - tool_call_id: None, - }; - let translated = translate_messages(&[msg]); - assert_eq!(translated.len(), 1); - assert_eq!(translated[0].role, "assistant"); - assert!(translated[0].content.is_none()); - let tool_calls = translated[0].tool_calls.as_ref().unwrap(); - assert_eq!(tool_calls.len(), 1); - assert_eq!(tool_calls[0].id, "call_1"); - assert_eq!(tool_calls[0].kind, "function"); - assert_eq!(tool_calls[0].function.name, "get_weather"); - assert_eq!(tool_calls[0].function.arguments, r#"{"city":"SF"}"#); - } - - #[test] - fn translate_assistant_message_with_text_and_tool_calls() { - let msg = Message { - role: Role::Assistant, - content: vec![ - ContentPart::text("Let me check the weather"), - ContentPart::ToolCall(ToolCall::new( - "call_2", - "get_weather", - serde_json::json!({"city": "NYC"}), - )), - ], - name: None, - tool_call_id: None, - }; - let translated = translate_messages(&[msg]); - assert_eq!( - translated[0].content.as_deref(), - Some("Let me check the weather") - ); - let tool_calls = translated[0].tool_calls.as_ref().unwrap(); - assert_eq!(tool_calls.len(), 1); - assert_eq!(tool_calls[0].function.name, "get_weather"); - } - - #[test] - fn translate_assistant_message_with_raw_arguments() { - let mut tc = ToolCall::new("call_3", "search", serde_json::json!({"q": "rust"})); - tc.raw_arguments = Some(r#"{"q": "rust"}"#.to_string()); - let msg = Message { - role: Role::Assistant, - content: vec![ContentPart::ToolCall(tc)], - name: None, - tool_call_id: None, - }; - let translated = translate_messages(&[msg]); - let tool_calls = translated[0].tool_calls.as_ref().unwrap(); - // Should prefer raw_arguments over serializing arguments - assert_eq!(tool_calls[0].function.arguments, r#"{"q": "rust"}"#); - } - - #[test] - fn translate_tool_message_has_tool_call_id() { - let msg = Message::tool_result( - "call_1", - serde_json::Value::String("72F and sunny".into()), - false, - ); - let translated = translate_messages(&[msg]); - assert_eq!(translated[0].role, "tool"); - assert_eq!(translated[0].tool_call_id.as_deref(), Some("call_1")); - assert!(translated[0].tool_calls.is_none()); - } - - #[test] - fn translate_user_message_has_no_tool_calls() { - let msg = Message::user("Hello"); - let translated = translate_messages(&[msg]); - assert_eq!(translated[0].role, "user"); - assert_eq!(translated[0].content.as_deref(), Some("Hello")); - assert!(translated[0].tool_calls.is_none()); - } - - #[test] - fn assistant_tool_calls_serialize_correctly() { - let msg = Message { - role: Role::Assistant, - content: vec![ContentPart::ToolCall(ToolCall::new( - "call_1", - "get_weather", - serde_json::json!({"city": "SF"}), - ))], - name: None, - tool_call_id: None, - }; - let translated = translate_messages(&[msg]); - let json = serde_json::to_value(&translated[0]).unwrap(); - assert!(json.get("content").is_none()); - assert!(json.get("tool_call_id").is_none()); - let tool_calls = json["tool_calls"].as_array().unwrap(); - assert_eq!(tool_calls.len(), 1); - assert_eq!(tool_calls[0]["type"], "function"); - assert_eq!(tool_calls[0]["id"], "call_1"); - assert_eq!(tool_calls[0]["function"]["name"], "get_weather"); - } - - fn minimal_request() -> Request { - Request { - model: "llama-3.1-70b".to_string(), - messages: vec![Message::user("Hello")], - provider: None, - tools: None, - tool_choice: None, - response_format: None, - temperature: None, - top_p: None, - max_tokens: None, - stop_sequences: None, - reasoning_effort: None, - speed: None, - metadata: None, - provider_options: None, - } - } - - #[test] - fn provider_options_none_produces_standard_body() { - let request = minimal_request(); - let body = build_api_request(&request, None, "groq"); - assert_eq!(body["model"], "llama-3.1-70b"); - assert!(body.get("stream").is_none()); - } - - #[test] - fn catalog_api_id_is_used_for_provider_request_body() { - let settings: LlmCatalogSettings = toml::from_str( - r#" -[providers.acme] -display_name = "Acme" -adapter = "openai_compatible" -agent_profile = "openai" -base_url = "https://api.acme.test/v1" - -[providers.acme.auth] -credentials = ["env:ACME_API_KEY"] - -[models."acme-large"] -provider = "acme" -api_id = "acme/model-large" -display_name = "Acme Large" -family = "acme" -default = true - -[models."acme-large".limits] -context_window = 128000 - -[models."acme-large".features] -tools = true -vision = false -reasoning = false -"#, - ) - .unwrap(); - let catalog = Catalog::from_builtin_with_overrides(&settings).unwrap(); - let mut request = minimal_request(); - request.model = "acme-large".to_string(); - - let body = build_api_request_with_catalog(&request, None, "acme", Some(&catalog)); - - assert_eq!(request.model, "acme-large"); - assert_eq!(body["model"], "acme/model-large"); - } - - #[test] - fn provider_options_matching_name_merged() { - let mut request = minimal_request(); - request.provider_options = Some(serde_json::json!({ - "groq": { - "frequency_penalty": 0.5, - "presence_penalty": 0.3 - } - })); - - let body = build_api_request(&request, None, "groq"); - assert_eq!(body["frequency_penalty"], 0.5); - assert_eq!(body["presence_penalty"], 0.3); - } - - #[test] - fn provider_options_different_name_ignored() { - let mut request = minimal_request(); - request.provider_options = Some(serde_json::json!({ - "together": { - "repetition_penalty": 1.2 - } - })); - - let body = build_api_request(&request, None, "groq"); - assert!(body.get("repetition_penalty").is_none()); - } - - #[test] - fn provider_options_uses_adapter_name() { - let mut request = minimal_request(); - request.provider_options = Some(serde_json::json!({ - "together": { - "repetition_penalty": 1.2 - } - })); - - let body = build_api_request(&request, None, "together"); - assert_eq!(body["repetition_penalty"], 1.2); - } - - #[test] - fn provider_options_preserves_standard_fields() { - let mut request = minimal_request(); - request.temperature = Some(0.7); - request.max_tokens = Some(200); - request.provider_options = Some(serde_json::json!({ - "groq": { - "frequency_penalty": 0.5 - } - })); - - let body = build_api_request(&request, Some(true), "groq"); - assert_eq!(body["temperature"], 0.7); - assert_eq!(body["max_tokens"], 200); - assert_eq!(body["stream"], true); - assert_eq!(body["frequency_penalty"], 0.5); - } - - #[test] - fn provider_options_can_override_model() { - let mut request = minimal_request(); - request.provider_options = Some(serde_json::json!({ - "groq": { - "model": "custom-model" - } - })); - - let body = build_api_request(&request, None, "groq"); - assert_eq!(body["model"], "custom-model"); - } - - #[test] - fn merge_provider_options_with_non_object_value() { - let mut body = serde_json::json!({"model": "test"}); - let opts = serde_json::json!({"groq": "not-an-object"}); - merge_provider_options(&mut body, Some(&opts), "groq"); - // Should not crash and body should be unchanged - assert_eq!(body["model"], "test"); - } - - #[test] - fn audio_content_produces_text_fallback() { - let msg = Message { - role: Role::User, - content: vec![ContentPart::Audio(AudioData { - url: Some("https://example.com/audio.wav".to_string()), - data: None, - media_type: None, - })], - name: None, - tool_call_id: None, - }; - let translated = translate_messages(&[msg]); - assert_eq!( - translated[0].content.as_deref(), - Some("[Audio content not supported by this provider]") - ); - } - - #[test] - fn document_content_produces_text_fallback_with_filename() { - let msg = Message { - role: Role::User, - content: vec![ContentPart::Document(DocumentData { - url: Some("https://example.com/doc.pdf".to_string()), - data: None, - media_type: None, - file_name: Some("report.pdf".to_string()), - })], - name: None, - tool_call_id: None, - }; - let translated = translate_messages(&[msg]); - assert_eq!( - translated[0].content.as_deref(), - Some("[Document 'report.pdf': content type not supported by this provider]") - ); - } - - #[test] - fn document_content_produces_text_fallback_without_filename() { - let msg = Message { - role: Role::User, - content: vec![ContentPart::Document(DocumentData { - url: None, - data: Some(vec![1, 2, 3]), - media_type: None, - file_name: None, - })], - name: None, - tool_call_id: None, - }; - let translated = translate_messages(&[msg]); - assert_eq!( - translated[0].content.as_deref(), - Some("[Document content not supported by this provider]") - ); - } - - #[test] - fn mixed_text_and_audio_content_concatenates() { - let msg = Message { - role: Role::User, - content: vec![ - ContentPart::text("Check this: "), - ContentPart::Audio(AudioData { - url: None, - data: Some(vec![1, 2]), - media_type: None, - }), - ], - name: None, - tool_call_id: None, - }; - let translated = translate_messages(&[msg]); - assert_eq!( - translated[0].content.as_deref(), - Some("Check this: [Audio content not supported by this provider]") - ); + Ok(Box::pin(out)) } }