diff --git a/lib/crates/fabro-llm/src/codec/anthropic_messages/encode.rs b/lib/crates/fabro-llm/src/codec/anthropic_messages/encode.rs index d59bd8c3d..3a2347ef3 100644 --- a/lib/crates/fabro-llm/src/codec/anthropic_messages/encode.rs +++ b/lib/crates/fabro-llm/src/codec/anthropic_messages/encode.rs @@ -610,7 +610,8 @@ reasoning = true fn direct_params() -> CodecParams { CodecParams { anthropic_version: AnthropicVersion::Header("2023-06-01"), - anthropic_beta: true, + anthropic_beta: true, + ..CodecParams::default() } } diff --git a/lib/crates/fabro-llm/src/codec/mod.rs b/lib/crates/fabro-llm/src/codec/mod.rs index 4e3959311..c8f43709c 100644 --- a/lib/crates/fabro-llm/src/codec/mod.rs +++ b/lib/crates/fabro-llm/src/codec/mod.rs @@ -12,6 +12,7 @@ pub(crate) mod anthropic_messages; pub(crate) mod openai_compatible; +pub(crate) mod openai_responses; use fabro_model::Model; @@ -42,7 +43,8 @@ pub(crate) struct CodecCtx<'a> { /// Per-route dialect knobs, expressed as data so one codec can serve several /// routes. The default is inert ("nothing special"); a route that needs a /// dialect quirk sets the relevant field. Grows as codecs need it — #459 adds -/// `ModelPlacement` for Bedrock. +/// `ModelPlacement` for Bedrock. Inert for codecs that don't read a given +/// field. #[derive(Debug, Default, Clone)] pub(crate) struct CodecParams { /// Where/whether to place the Anthropic API version. Direct Anthropic uses @@ -52,6 +54,12 @@ pub(crate) struct CodecParams { /// Whether to emit Anthropic beta headers (prompt-caching / fast-mode / /// 1M-context). True on the direct route, false for Kimi-over-anthropic. pub anthropic_beta: bool, + /// Codex-endpoint dialect for the openai_responses codec: omit the + /// sampling params (`temperature`/`top_p`/`max_output_tokens`) the Codex + /// endpoint rejects and always send `instructions` (empty string when the + /// request has none). The transport-side half of codex mode (forced + /// streaming) is route config, not codec data. + pub openai_codex: bool, } /// Placement of the Anthropic API version on the wire. @@ -83,8 +91,9 @@ pub(crate) struct EncodedRequest { /// 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. + /// SSE `event:` type — `Some` when the framing carries one (anthropic, + /// openai responses); `None` for the data-only framing + /// openai_compatible/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. diff --git a/lib/crates/fabro-llm/src/codec/openai_responses/decode.rs b/lib/crates/fabro-llm/src/codec/openai_responses/decode.rs new file mode 100644 index 000000000..417dba9e7 --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_responses/decode.rs @@ -0,0 +1,360 @@ +//! Response decoding: OpenAI Responses API body → canonical `Response`. + +use serde::Deserialize; + +use super::wire::{ApiResponse, ApiUsage, InputTokensResponse}; +use crate::codec::CodecCtx; +use crate::error::Error; +use crate::types::{ + ContentPart, FinishReason, Message, RateLimitInfo, Response, Role, TokenCounts, ToolCall, +}; + +pub(super) fn token_counts_from_api_usage(usage: Option<&ApiUsage>) -> TokenCounts { + usage.map_or_else(TokenCounts::default, |u| { + let cached_tokens = u + .input_tokens_details + .as_ref() + .and_then(|d| d.cached_tokens) + .unwrap_or(0); + let reasoning_tokens = u + .output_tokens_details + .as_ref() + .and_then(|d| d.reasoning_tokens) + .unwrap_or(0); + TokenCounts { + input_tokens: u.input_tokens.saturating_sub(cached_tokens), + output_tokens: u.output_tokens.saturating_sub(reasoning_tokens), + reasoning_tokens, + cache_read_tokens: cached_tokens, + ..TokenCounts::default() + } + }) +} + +/// Map the Responses API status to a `FinishReason`. +pub(super) fn map_finish_reason(status: Option<&str>, has_tool_calls: bool) -> FinishReason { + if has_tool_calls { + return FinishReason::ToolCalls; + } + match status { + Some("completed") | None => FinishReason::Stop, + Some("incomplete") => FinishReason::Length, + Some("failed") => FinishReason::Error, + Some(other) => FinishReason::Other(other.to_string()), + } +} + +/// Build a `ToolCall` from a `function_call` / `custom_tool_call` output item. +/// The call-id/item-id round-trip rules live here, shared by the blocking and +/// streaming decode paths. +pub(super) fn tool_call_from_item(item: &serde_json::Value, custom: bool) -> ToolCall { + let item_id = item + .get("id") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let call_id = item + .get("call_id") + .and_then(serde_json::Value::as_str) + .unwrap_or(item_id); + let name = item + .get("name") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + + let mut tc = if custom { + let raw_input = item + .get("input") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let mut tc = ToolCall::new(call_id, name, serde_json::json!(raw_input)); + tc.tool_type = "custom".to_string(); + tc.raw_arguments = Some(raw_input.to_string()); + tc + } else { + let args_str = item + .get("arguments") + .and_then(serde_json::Value::as_str) + .unwrap_or("{}"); + let arguments = serde_json::from_str(args_str).unwrap_or_else(|_| serde_json::json!({})); + let mut tc = ToolCall::new(call_id, name, arguments); + tc.raw_arguments = Some(args_str.to_string()); + tc + }; + // Preserve item-level ID (fc_xxx) for Responses API round-trip + if !item_id.is_empty() { + tc.provider_metadata = Some(serde_json::json!({"id": item_id})); + } + tc +} + +/// Parse output items from the Responses API into content parts. +pub(super) fn parse_output(output: Vec) -> (Vec, bool) { + let mut parts = Vec::new(); + let mut has_tool_calls = false; + + for item in output { + let item_type = item + .get("type") + .and_then(serde_json::Value::as_str) + .unwrap_or_default() + .to_string(); + match item_type.as_str() { + "message" => { + // Preserve the full message item for Responses API round-tripping. + // The item's `id` and `status` fields are required so that reasoning + // items preceding it can find their "required following item." + let mut texts = Vec::new(); + if let Some(content) = item.get("content").and_then(|c| c.as_array()) { + for block in content { + if block.get("type").and_then(serde_json::Value::as_str) + == Some("output_text") + { + if let Some(text) = + block.get("text").and_then(serde_json::Value::as_str) + { + texts.push(ContentPart::text(text)); + } + } + } + } + parts.push(ContentPart::Other { + kind: ContentPart::OPENAI_MESSAGE.to_string(), + data: item, + }); + parts.extend(texts); + } + "reasoning" => { + parts.push(ContentPart::Other { + kind: ContentPart::OPENAI_REASONING.to_string(), + data: item, + }); + } + "function_call" | "custom_tool_call" => { + let tc = tool_call_from_item(&item, item_type == "custom_tool_call"); + // Skip tool calls with empty names (e.g. model-internal items) + if tc.name.is_empty() { + continue; + } + has_tool_calls = true; + parts.push(ContentPart::ToolCall(tc)); + } + _ => {} + } + } + + (parts, has_tool_calls) +} + +pub(super) fn decode_response( + body: &str, + ctx: &CodecCtx<'_>, + rate_limit: Option, +) -> Result { + let raw: serde_json::Value = serde_json::from_str(body) + .map_err(|e| Error::network(format!("failed to parse OpenAI response: {e}"), e))?; + let api_resp = ApiResponse::deserialize(&raw) + .map_err(|e| Error::network(format!("failed to parse OpenAI response: {e}"), e))?; + + let (content_parts, has_tool_calls) = parse_output(api_resp.output); + let finish_reason = map_finish_reason(api_resp.status.as_deref(), has_tool_calls); + + let usage = token_counts_from_api_usage(api_resp.usage.as_ref()); + + Ok(Response { + id: api_resp.id, + model: api_resp.model.unwrap_or_else(|| ctx.request.model.clone()), + provider: ctx.provider_name.to_string(), + message: Message { + role: Role::Assistant, + content: content_parts, + name: None, + tool_call_id: None, + }, + finish_reason, + usage, + raw: Some(raw), + warnings: vec![], + rate_limit, + }) +} + +pub(super) fn decode_count_tokens(body: &str) -> Result { + let response: InputTokensResponse = + serde_json::from_str(body).map_err(|e| Error::Configuration { + message: format!("failed to parse OpenAI input token response: {e}"), + source: None, + })?; + + if response.object != "response.input_tokens" { + return Err(Error::Configuration { + message: format!( + "failed to parse OpenAI input token response: unexpected object '{}'", + response.object + ), + source: None, + }); + } + + Ok(response.input_tokens) +} + +#[cfg(test)] +mod tests { + use super::super::encode; + use super::*; + + #[test] + fn parse_output_preserves_both_ids_on_function_call() { + let output = vec![serde_json::json!({ + "type": "function_call", + "id": "fc_abc123", + "call_id": "call_xyz789", + "name": "get_weather", + "arguments": "{\"location\":\"NYC\"}" + })]; + let (parts, has_tool_calls) = parse_output(output); + assert!(has_tool_calls); + assert_eq!(parts.len(), 1); + match &parts[0] { + ContentPart::ToolCall(tc) => { + // call_id is used as the ToolCall.id (links to tool results) + assert_eq!(tc.id, "call_xyz789"); + // item-level id (fc_xxx) is preserved in provider_metadata + let meta = tc + .provider_metadata + .as_ref() + .expect("provider_metadata should be set"); + assert_eq!(meta["id"], "fc_abc123"); + } + other => panic!("expected ToolCall, got {other:?}"), + } + } + + #[test] + fn parse_output_preserves_custom_tool_call_raw_input() { + let patch = "*** Begin Patch\n*** Add File: hello.txt\n+hello\n*** End Patch\n"; + let output = vec![serde_json::json!({ + "type": "custom_tool_call", + "id": "ctc_abc123", + "call_id": "call_xyz789", + "name": "apply_patch", + "input": patch, + })]; + + let (parts, has_tool_calls) = parse_output(output); + + assert!(has_tool_calls); + assert_eq!(parts.len(), 1); + match &parts[0] { + ContentPart::ToolCall(tc) => { + assert_eq!(tc.id, "call_xyz789"); + assert_eq!(tc.name, "apply_patch"); + assert_eq!(tc.tool_type, "custom"); + assert_eq!(tc.arguments, serde_json::json!(patch)); + assert_eq!(tc.raw_arguments.as_deref(), Some(patch)); + let meta = tc + .provider_metadata + .as_ref() + .expect("provider metadata should preserve item id"); + assert_eq!(meta["id"], "ctc_abc123"); + } + other => panic!("expected ToolCall, got {other:?}"), + } + } + + #[test] + fn parse_output_preserves_reasoning_items() { + let output = vec![ + serde_json::json!({ + "type": "reasoning", + "id": "rs_abc123", + "summary": [{"type": "summary_text", "text": "Thinking..."}] + }), + serde_json::json!({ + "type": "function_call", + "id": "fc_def456", + "call_id": "call_789", + "name": "search", + "arguments": "{}" + }), + ]; + let (parts, has_tool_calls) = parse_output(output); + assert!(has_tool_calls); + assert_eq!(parts.len(), 2); + // First part is the reasoning item + match &parts[0] { + ContentPart::Other { kind, data } => { + assert_eq!(kind, ContentPart::OPENAI_REASONING); + assert_eq!(data["type"], "reasoning"); + assert_eq!(data["id"], "rs_abc123"); + } + other => panic!("expected Other, got {other:?}"), + } + // Second part is the function call + assert!(matches!(&parts[1], ContentPart::ToolCall(_))); + } + + #[test] + fn parse_output_preserves_message_items() { + let output = vec![ + serde_json::json!({ + "type": "reasoning", + "id": "rs_abc", + "summary": [] + }), + serde_json::json!({ + "type": "message", + "id": "msg_xyz", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "Hello"}] + }), + serde_json::json!({ + "type": "function_call", + "id": "fc_123", + "call_id": "call_456", + "name": "search", + "arguments": "{}" + }), + ]; + let (parts, has_tool_calls) = parse_output(output); + assert!(has_tool_calls); + // reasoning + openai_message + text + function_call + assert_eq!(parts.len(), 4); + assert!( + matches!(&parts[0], ContentPart::Other { kind, .. } if kind == ContentPart::OPENAI_REASONING) + ); + assert!( + matches!(&parts[1], ContentPart::Other { kind, data } if kind == ContentPart::OPENAI_MESSAGE && data["id"] == "msg_xyz") + ); + assert!(matches!(&parts[2], ContentPart::Text(t) if t == "Hello")); + assert!(matches!(&parts[3], ContentPart::ToolCall(_))); + } + + #[test] + fn parse_output_round_trips_function_call_ids() { + // Simulate a response from the Responses API + let output = vec![serde_json::json!({ + "type": "function_call", + "id": "fc_item1", + "call_id": "call_001", + "name": "search", + "arguments": "{\"q\":\"test\"}" + })]; + let (parts, _) = parse_output(output); + + // Now translate back to input format + let msg = Message { + role: Role::Assistant, + content: parts, + name: None, + tool_call_id: None, + }; + let (_, input) = encode::translate_input(&[msg]); + let fc = &input[0]; + + // The round-tripped function call should have correct IDs + assert_eq!(fc["id"], "fc_item1"); + assert_eq!(fc["call_id"], "call_001"); + } +} diff --git a/lib/crates/fabro-llm/src/codec/openai_responses/encode.rs b/lib/crates/fabro-llm/src/codec/openai_responses/encode.rs new file mode 100644 index 000000000..8ede38ec3 --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_responses/encode.rs @@ -0,0 +1,949 @@ +//! Request encoding: canonical request → OpenAI Responses API body. +//! +//! Pure and sync. File-backed image attachments are resolved to inline data by +//! `attachments::resolve` in the adapter *before* encode runs, so the content +//! translation here never touches the filesystem. + +use std::collections::HashSet; + +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; + +use super::wire::ApiRequest; +use crate::codec::{CodecCtx, EncodedRequest}; +use crate::types::{ + ContentPart, Message, ResponseFormat, ResponseFormatType, Role, ToolChoice, ToolDefinition, +}; + +// --- Public entry points ----------------------------------------------------- + +pub(super) fn encode(ctx: &CodecCtx<'_>, stream: bool) -> EncodedRequest { + EncodedRequest { + body: build_body(ctx, stream), + endpoint: "/responses".to_string(), + headers: Vec::new(), + } +} + +pub(super) fn encode_count_tokens(ctx: &CodecCtx<'_>) -> EncodedRequest { + EncodedRequest { + body: filter_input_tokens_request_body(build_body(ctx, false)), + endpoint: "/responses/input_tokens".to_string(), + headers: Vec::new(), + } +} + +/// Serialize the API request and merge any `provider_options.openai` keys into +/// the body (overrides win, matching the long-standing contract). +fn build_body(ctx: &CodecCtx<'_>, stream: bool) -> serde_json::Value { + let api_request = build_api_request(ctx, stream); + let mut body = serde_json::to_value(&api_request).unwrap_or_else(|_| serde_json::json!({})); + + if let Some(openai_opts) = ctx + .request + .provider_options + .as_ref() + .and_then(|opts| opts.get("openai")) + { + if let (Some(base), Some(overrides)) = (body.as_object_mut(), openai_opts.as_object()) { + for (key, value) in overrides { + base.insert(key.clone(), value.clone()); + } + } + } + + body +} + +/// Build an `ApiRequest` from the canonical request. +/// +/// When the route is in codex mode (`ctx.params.openai_codex`), unsupported +/// fields (`temperature`, `max_output_tokens`, `top_p`) are omitted and empty +/// instructions are sent as `""` (required by the Codex endpoint). +fn build_api_request(ctx: &CodecCtx<'_>, stream: bool) -> ApiRequest { + let request = ctx.request; + let codex_mode = ctx.params.openai_codex; + + let (instructions, input) = translate_input(&request.messages); + let api_tools = request.tools.as_ref().map(|t| translate_tools(t)); + let tool_choice = request.tool_choice.as_ref().map(translate_tool_choice); + let reasoning = request + .reasoning_effort + .as_ref() + .map(|effort| serde_json::json!({"effort": <&'static str>::from(*effort)})); + let text = request + .response_format + .as_ref() + .and_then(translate_response_format); + + let include = vec!["reasoning.encrypted_content".to_string()]; + + let instructions = if codex_mode { + Some(instructions.unwrap_or_default()) + } else { + instructions + }; + + ApiRequest { + model: ctx.deployment_id.to_string(), + input, + instructions, + temperature: if codex_mode { + None + } else { + request.temperature + }, + max_output_tokens: if codex_mode { None } else { request.max_tokens }, + top_p: if codex_mode { None } else { request.top_p }, + tools: api_tools, + tool_choice, + reasoning, + text, + stop: request.stop_sequences.clone(), + metadata: request.metadata.clone(), + // store: false means output items are not persisted server-side. + // Request encrypted reasoning content on every turn so reasoning items + // from models that emit them by default can round-trip statelessly. + store: false, + include, + stream, + } +} + +/// Project a full request body down to the fields the +/// `/responses/input_tokens` endpoint accepts. +fn filter_input_tokens_request_body(mut body: serde_json::Value) -> serde_json::Value { + const ALLOWED_FIELDS: &[&str] = &[ + "conversation", + "input", + "instructions", + "model", + "parallel_tool_calls", + "previous_response_id", + "reasoning", + "text", + "tool_choice", + "tools", + "truncation", + ]; + + let Some(obj) = body.as_object_mut() else { + return serde_json::json!({}); + }; + obj.retain(|key, _| ALLOWED_FIELDS.contains(&key.as_str())); + body +} + +// --- Content / message / tool translation ------------------------------------ + +/// Translate unified messages to Responses API `input` array format. Sync: +/// file-backed image attachments are already resolved to inline data upstream. +pub(super) fn translate_input(messages: &[Message]) -> (Option, Vec) { + let mut instructions_parts: Vec = Vec::new(); + let mut input: Vec = Vec::new(); + let mut custom_call_ids: HashSet = HashSet::new(); + + for msg in messages { + match msg.role { + Role::System | Role::Developer => { + instructions_parts.push(msg.text()); + } + Role::User => { + let mut content = Vec::new(); + for part in &msg.content { + let maybe_content = match part { + ContentPart::Text(text) => { + Some(serde_json::json!({"type": "input_text", "text": text})) + } + ContentPart::Image(img) => match &img.url { + Some(url) => { + Some(serde_json::json!({"type": "input_image", "image_url": url})) + } + None => img.data.as_ref().map(|data| { + let mime = img.media_type.as_deref().unwrap_or("image/png"); + let b64 = BASE64_STANDARD.encode(data); + serde_json::json!({ + "type": "input_image", + "image_url": format!("data:{mime};base64,{b64}"), + }) + }), + }, + ContentPart::Audio(_) => Some( + serde_json::json!({"type": "input_text", "text": "[Audio content not supported by this provider]"}), + ), + 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]"), + ); + Some(serde_json::json!({"type": "input_text", "text": desc})) + } + _ => None, + }; + if let Some(content_part) = maybe_content { + content.push(content_part); + } + } + if !content.is_empty() { + input.push(serde_json::json!({ + "type": "message", + "role": "user", + "content": content, + })); + } + } + Role::Assistant => { + // If we have a preserved opaque message item (with id/status), use + // it instead of constructing a new message from Text parts. This is + // required so that reasoning items can find their "required following + // item" during Responses API round-tripping. + let has_opaque_message = msg.content.iter().any(|p| { + matches!(p, ContentPart::Other { kind, .. } if kind == ContentPart::OPENAI_MESSAGE) + }); + for part in &msg.content { + match part { + ContentPart::Text(text) if !has_opaque_message => { + input.push(serde_json::json!({ + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": text}], + })); + } + ContentPart::ToolCall(tc) if !tc.name.is_empty() => { + // Use the item-level ID (fc_xxx) for the `id` field; + // fall back to tc.id if no provider_metadata was stored. + let item_id = tc + .provider_metadata + .as_ref() + .and_then(|m| m.get("id")) + .and_then(serde_json::Value::as_str) + .unwrap_or(&tc.id); + if tc.tool_type == "custom" { + custom_call_ids.insert(tc.id.clone()); + let raw_input = tc.raw_arguments.as_ref().map_or_else( + || { + tc.arguments.as_str().map_or_else( + || tc.arguments.to_string(), + str::to_string, + ) + }, + Clone::clone, + ); + input.push(serde_json::json!({ + "type": "custom_tool_call", + "id": item_id, + "call_id": tc.id, + "name": tc.name, + "input": raw_input, + })); + } else { + let args = tc + .raw_arguments + .as_ref() + .map_or_else(|| tc.arguments.to_string(), Clone::clone); + input.push(serde_json::json!({ + "type": "function_call", + "id": item_id, + "call_id": tc.id, + "name": tc.name, + "arguments": args, + })); + } + } + ContentPart::Other { data, .. } if part.is_opaque_openai() => { + input.push(data.clone()); + } + _ => {} + } + } + } + Role::Tool => { + for part in &msg.content { + if let ContentPart::ToolResult(tr) = part { + let output = tr + .content + .as_str() + .map_or_else(|| tr.content.to_string(), str::to_string); + let is_custom = custom_call_ids.contains(&tr.tool_call_id) + || msg.name.as_deref() == Some("apply_patch"); + let mut item = if is_custom { + serde_json::json!({ + "type": "custom_tool_call_output", + "call_id": tr.tool_call_id, + "output": output, + }) + } else { + serde_json::json!({ + "type": "function_call_output", + "call_id": tr.tool_call_id, + "output": output, + }) + }; + if tr.is_error && !is_custom { + item["status"] = serde_json::json!("incomplete"); + } + input.push(item); + } + } + } + } + } + + let instructions = if instructions_parts.is_empty() { + None + } else { + Some(instructions_parts.join("\n")) + }; + + (instructions, input) +} + +/// Translate unified tool definitions to Responses API tool format. +pub(super) fn translate_tools(tools: &[ToolDefinition]) -> Vec { + tools + .iter() + .map(|t| { + if t.is_custom() { + serde_json::json!({ + "type": "custom", + "name": t.name, + "description": t.description, + "format": t.custom_format().cloned().unwrap_or_else(|| serde_json::json!({})), + }) + } else { + serde_json::json!({ + "type": "function", + "name": t.name, + "description": t.description, + "parameters": t.parameters, + }) + } + }) + .collect() +} + +/// Translate unified `ToolChoice` to Responses API format. +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", "name": tool_name}) + } + } +} + +/// Translate unified `ResponseFormat` to Responses API `text` field. +/// +/// The Responses API uses `"text": {"format": {...}}` for structured output. +fn translate_response_format(format: &ResponseFormat) -> Option { + match format.kind { + ResponseFormatType::Text => None, + ResponseFormatType::JsonObject => { + Some(serde_json::json!({"format": {"type": "json_object"}})) + } + ResponseFormatType::JsonSchema => { + let mut schema_obj = serde_json::json!({ + "type": "json_schema", + "name": "response", + "strict": format.strict, + }); + if let Some(schema) = &format.json_schema { + schema_obj["schema"] = schema.clone(); + } + Some(serde_json::json!({"format": schema_obj})) + } + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::*; + use crate::codec::CodecParams; + use crate::types::{AudioData, DocumentData, ReasoningEffort, Request, ToolCall, ToolResult}; + + fn minimal_request() -> Request { + Request { + model: "gpt-4o".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` (no catalog: the wire model id is the request model) + /// and return the merged body, mirroring the adapter's encode path. + fn encode_body(request: &Request, stream: bool, codex: bool) -> serde_json::Value { + let params = CodecParams { + openai_codex: codex, + ..CodecParams::default() + }; + let ctx = CodecCtx { + request, + provider_name: "openai", + deployment_id: &request.model, + model: None, + params: ¶ms, + }; + encode(&ctx, stream).body + } + + #[test] + fn build_request_body_includes_metadata() { + let mut metadata = HashMap::new(); + metadata.insert("user_id".to_string(), "u123".to_string()); + metadata.insert("session".to_string(), "s456".to_string()); + + let mut request = minimal_request(); + request.metadata = Some(metadata); + + let body = encode_body(&request, false, false); + let meta = body.get("metadata").expect("metadata should be present"); + assert_eq!(meta["user_id"], "u123"); + assert_eq!(meta["session"], "s456"); + } + + #[test] + fn build_request_body_omits_metadata_when_none() { + let request = minimal_request(); + let body = encode_body(&request, false, false); + assert!(body.get("metadata").is_none()); + } + + #[test] + fn build_request_body_merges_provider_options_openai() { + let mut request = minimal_request(); + request.provider_options = Some(serde_json::json!({ + "openai": { + "store": true, + "previous_response_id": "resp_abc123" + } + })); + + let body = encode_body(&request, false, false); + assert_eq!(body["store"], true); + assert_eq!(body["previous_response_id"], "resp_abc123"); + } + + #[test] + fn build_request_body_provider_options_override_fields() { + let mut request = minimal_request(); + request.temperature = Some(0.5); + request.provider_options = Some(serde_json::json!({ + "openai": { + "temperature": 0.9 + } + })); + + let body = encode_body(&request, false, false); + // provider_options should override the base field + assert_eq!(body["temperature"], 0.9); + } + + #[test] + fn build_request_body_ignores_non_openai_provider_options() { + let mut request = minimal_request(); + request.provider_options = Some(serde_json::json!({ + "anthropic": { + "thinking": {"type": "enabled", "budget_tokens": 10000} + } + })); + + let body = encode_body(&request, false, false); + // anthropic options should not leak into the OpenAI request + assert!(body.get("thinking").is_none()); + } + + #[test] + fn build_request_body_no_provider_options() { + let request = minimal_request(); + let body = encode_body(&request, false, false); + assert_eq!(body["model"], "gpt-4o"); + // stream field is omitted when false (skip_serializing_if) + assert!(body.get("stream").is_none()); + } + + #[test] + fn filter_input_tokens_request_body_keeps_only_count_fields() { + let mut metadata = HashMap::new(); + metadata.insert("trace".to_string(), "abc".to_string()); + + let mut request = minimal_request(); + request.tools = Some(vec![ToolDefinition::function( + "search", + "Search files", + serde_json::json!({"type": "object"}), + )]); + request.reasoning_effort = Some(ReasoningEffort::Low); + request.response_format = Some(ResponseFormat { + kind: ResponseFormatType::JsonSchema, + json_schema: Some(serde_json::json!({"type": "object"})), + strict: true, + }); + request.temperature = Some(0.2); + request.top_p = Some(0.9); + request.max_tokens = Some(32); + request.stop_sequences = Some(vec!["END".to_string()]); + request.metadata = Some(metadata); + + let body = encode_body(&request, true, false); + let filtered = filter_input_tokens_request_body(body); + + assert_eq!( + filtered, + serde_json::json!({ + "input": [{"type": "message", "content": [{"text": "Hello", "type": "input_text"}], "role": "user"}], + "model": "gpt-4o", + "reasoning": {"effort": "low"}, + "text": {"format": {"name": "response", "schema": {"type": "object"}, "strict": true, "type": "json_schema"}}, + "tools": [{"description": "Search files", "name": "search", "parameters": {"type": "object"}, "type": "function"}] + }) + ); + assert!(filtered.get("store").is_none()); + assert!(filtered.get("include").is_none()); + assert!(filtered.get("stream").is_none()); + assert!(filtered.get("max_output_tokens").is_none()); + assert!(filtered.get("metadata").is_none()); + assert!(filtered.get("temperature").is_none()); + assert!(filtered.get("top_p").is_none()); + assert!(filtered.get("stop").is_none()); + } + + #[test] + fn filter_input_tokens_request_body_preserves_codex_serialization() { + let body = encode_body(&minimal_request(), false, true); + let filtered = filter_input_tokens_request_body(body); + + assert_eq!(filtered["instructions"], ""); + assert!(filtered.get("input").is_some()); + assert!(filtered.get("model").is_some()); + assert!(filtered.get("max_output_tokens").is_none()); + assert!(filtered.get("include").is_none()); + } + + #[test] + fn count_tokens_endpoint_carries_filtered_body() { + let request = minimal_request(); + let params = CodecParams::default(); + let ctx = CodecCtx { + request: &request, + provider_name: "openai", + deployment_id: &request.model, + model: None, + params: ¶ms, + }; + + let encoded = encode_count_tokens(&ctx); + assert_eq!(encoded.endpoint, "/responses/input_tokens"); + assert!(encoded.body.get("store").is_none()); + assert!(encoded.body.get("include").is_none()); + assert_eq!(encoded.body["model"], "gpt-4o"); + } + + #[test] + fn build_request_body_includes_encrypted_reasoning_for_stateless_requests() { + let request = minimal_request(); + + let body = encode_body(&request, false, false); + + assert_eq!( + body["include"], + serde_json::json!(["reasoning.encrypted_content"]) + ); + } + + #[test] + fn build_request_body_emits_custom_apply_patch_tool() { + let mut request = minimal_request(); + request.tools = Some(vec![ + ToolDefinition::custom( + "apply_patch", + "Use the `apply_patch` tool to edit files. This is a FREEFORM tool, so do not wrap the patch in JSON.", + serde_json::json!({ + "type": "grammar", + "syntax": "lark", + "definition": "start: begin_patch hunk+ end_patch", + }), + ), + ToolDefinition::function( + "read_file", + "Read file", + serde_json::json!({ + "type": "object", + "properties": {"file_path": {"type": "string"}}, + "required": ["file_path"], + }), + ), + ]); + + let body = encode_body(&request, false, false); + let tools = body["tools"].as_array().expect("tools should be present"); + let apply_patch = tools + .iter() + .find(|tool| tool["name"] == "apply_patch") + .expect("apply_patch tool should be present"); + let read_file = tools + .iter() + .find(|tool| tool["name"] == "read_file") + .expect("read_file tool should be present"); + + assert_eq!(apply_patch["type"], "custom"); + assert_eq!(apply_patch["format"]["type"], "grammar"); + assert_eq!(apply_patch["format"]["syntax"], "lark"); + assert!(apply_patch.get("parameters").is_none()); + assert_eq!(read_file["type"], "function"); + assert_eq!(read_file["parameters"]["type"], "object"); + } + + #[test] + fn build_request_body_stream_flag() { + let request = minimal_request(); + let body = encode_body(&request, true, false); + assert!(body["stream"].as_bool().unwrap_or(false)); + } + + #[test] + fn build_request_body_metadata_and_provider_options_together() { + let mut metadata = HashMap::new(); + metadata.insert("trace_id".to_string(), "t789".to_string()); + + let mut request = minimal_request(); + request.metadata = Some(metadata); + request.provider_options = Some(serde_json::json!({ + "openai": { + "store": true + } + })); + + let body = encode_body(&request, false, false); + assert_eq!(body["metadata"]["trace_id"], "t789"); + assert_eq!(body["store"], true); + } + + #[test] + fn build_request_body_includes_stop_sequences() { + let mut request = minimal_request(); + request.stop_sequences = Some(vec!["END".to_string(), "STOP".to_string()]); + + let body = encode_body(&request, false, false); + let stop = body.get("stop").expect("stop should be present"); + let arr = stop.as_array().expect("stop should be an array"); + assert_eq!(arr.len(), 2); + assert_eq!(arr[0], "END"); + assert_eq!(arr[1], "STOP"); + } + + #[test] + fn build_request_body_omits_stop_when_none() { + let request = minimal_request(); + let body = encode_body(&request, false, false); + assert!(body.get("stop").is_none()); + } + + #[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 (_, input) = translate_input(&[msg]); + let content = input[0]["content"] + .as_array() + .expect("content should be array"); + assert_eq!(content[0]["type"], "input_text"); + assert_eq!( + content[0]["text"], + "[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 (_, input) = translate_input(&[msg]); + let content = input[0]["content"] + .as_array() + .expect("content should be array"); + assert_eq!(content[0]["type"], "input_text"); + assert_eq!( + content[0]["text"], + "[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 (_, input) = translate_input(&[msg]); + let content = input[0]["content"] + .as_array() + .expect("content should be array"); + assert_eq!(content[0]["type"], "input_text"); + assert_eq!( + content[0]["text"], + "[Document content not supported by this provider]" + ); + } + + #[test] + fn translate_input_uses_item_id_for_id_field() { + let mut tc = ToolCall::new( + "call_xyz789", + "get_weather", + serde_json::json!({"location": "NYC"}), + ); + tc.provider_metadata = Some(serde_json::json!({"id": "fc_abc123"})); + + let msg = Message { + role: Role::Assistant, + content: vec![ContentPart::ToolCall(tc)], + name: None, + tool_call_id: None, + }; + let (_, input) = translate_input(&[msg]); + let fc = &input[0]; + assert_eq!(fc["type"], "function_call"); + // id field uses the fc_ prefixed item ID + assert_eq!(fc["id"], "fc_abc123"); + // call_id field uses the call_ prefixed call ID + assert_eq!(fc["call_id"], "call_xyz789"); + } + + #[test] + fn translate_input_falls_back_to_tc_id_without_metadata() { + let tc = ToolCall::new("call_xyz789", "get_weather", serde_json::json!({})); + + let msg = Message { + role: Role::Assistant, + content: vec![ContentPart::ToolCall(tc)], + name: None, + tool_call_id: None, + }; + let (_, input) = translate_input(&[msg]); + let fc = &input[0]; + // Without provider_metadata, both fields use tc.id + assert_eq!(fc["id"], "call_xyz789"); + assert_eq!(fc["call_id"], "call_xyz789"); + } + + #[test] + fn reasoning_items_round_trip_through_translate_input() { + let reasoning = serde_json::json!({ + "type": "reasoning", + "id": "rs_abc123", + "summary": [{"type": "summary_text", "text": "Thinking..."}] + }); + let mut tc = ToolCall::new("call_789", "search", serde_json::json!({})); + tc.provider_metadata = Some(serde_json::json!({"id": "fc_def456"})); + + let msg = Message { + role: Role::Assistant, + content: vec![ + ContentPart::Other { + kind: ContentPart::OPENAI_REASONING.to_string(), + data: reasoning, + }, + ContentPart::ToolCall(tc), + ], + name: None, + tool_call_id: None, + }; + let (_, input) = translate_input(&[msg]); + assert_eq!(input.len(), 2); + // Reasoning item is emitted first + assert_eq!(input[0]["type"], "reasoning"); + assert_eq!(input[0]["id"], "rs_abc123"); + // Function call follows + assert_eq!(input[1]["type"], "function_call"); + assert_eq!(input[1]["id"], "fc_def456"); + assert_eq!(input[1]["call_id"], "call_789"); + } + + #[test] + fn reasoning_message_function_call_round_trip() { + // Simulates an assistant turn with reasoning + text + tool call. + // The opaque message item (with id/status) must be used instead of + // constructing a new one from Text, so the reasoning item can find + // its "required following item." + let reasoning = serde_json::json!({ + "type": "reasoning", + "id": "rs_xyz789", + "summary": [{"type": "summary_text", "text": "Let me check..."}] + }); + let opaque_message = serde_json::json!({ + "type": "message", + "id": "msg_abc123", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "Checking now."}] + }); + let mut tc = ToolCall::new("call_001", "shell", serde_json::json!({"cmd": "ls"})); + tc.provider_metadata = Some(serde_json::json!({"id": "fc_def456"})); + + let msg = Message { + role: Role::Assistant, + content: vec![ + ContentPart::Other { + kind: ContentPart::OPENAI_REASONING.to_string(), + data: reasoning, + }, + ContentPart::Other { + kind: ContentPart::OPENAI_MESSAGE.to_string(), + data: opaque_message, + }, + ContentPart::text("Checking now."), + ContentPart::ToolCall(tc), + ], + name: None, + tool_call_id: None, + }; + let (_, input) = translate_input(&[msg]); + assert_eq!(input.len(), 3); + // Reasoning first + assert_eq!(input[0]["type"], "reasoning"); + assert_eq!(input[0]["id"], "rs_xyz789"); + // Opaque message with id/status (not a reconstructed one) + assert_eq!(input[1]["type"], "message"); + assert_eq!(input[1]["id"], "msg_abc123"); + assert_eq!(input[1]["status"], "completed"); + // Function call last + assert_eq!(input[2]["type"], "function_call"); + assert_eq!(input[2]["id"], "fc_def456"); + } + + #[test] + fn text_without_opaque_message_still_constructs_message() { + // For non-OpenAI turns or turns without preserved message items, + // Text parts should still produce a constructed message. + let msg = Message { + role: Role::Assistant, + content: vec![ContentPart::text("Hello")], + name: None, + tool_call_id: None, + }; + let (_, input) = translate_input(&[msg]); + assert_eq!(input.len(), 1); + assert_eq!(input[0]["type"], "message"); + assert_eq!(input[0]["role"], "assistant"); + // No id field on constructed messages + assert!(input[0].get("id").is_none()); + } + + #[test] + fn custom_tool_call_history_round_trips_through_translate_input() { + let patch = "*** Begin Patch\n*** Delete File: stale.txt\n*** End Patch\n"; + let mut tc = ToolCall::new("call_001", "apply_patch", serde_json::json!(patch)); + tc.tool_type = "custom".to_string(); + tc.raw_arguments = Some(patch.to_string()); + tc.provider_metadata = Some(serde_json::json!({"id": "ctc_def456"})); + + let msg = Message { + role: Role::Assistant, + content: vec![ContentPart::ToolCall(tc)], + name: None, + tool_call_id: None, + }; + + let (_, input) = translate_input(&[msg]); + + assert_eq!(input.len(), 1); + assert_eq!(input[0]["type"], "custom_tool_call"); + assert_eq!(input[0]["id"], "ctc_def456"); + assert_eq!(input[0]["call_id"], "call_001"); + assert_eq!(input[0]["name"], "apply_patch"); + assert_eq!(input[0]["input"], patch); + } + + #[test] + fn custom_tool_result_history_round_trips_through_translate_input() { + let msg = Message { + role: Role::Tool, + content: vec![ContentPart::ToolResult(ToolResult::success( + "call_001", + serde_json::json!("Success. Updated the following files:\nA hello.txt\n"), + ))], + name: Some("apply_patch".to_string()), + tool_call_id: Some("call_001".to_string()), + }; + + let (_, input) = translate_input(&[msg]); + + assert_eq!(input.len(), 1); + assert_eq!(input[0]["type"], "custom_tool_call_output"); + assert_eq!(input[0]["call_id"], "call_001"); + assert_eq!( + input[0]["output"], + "Success. Updated the following files:\nA hello.txt\n" + ); + } + + #[test] + fn custom_tool_result_history_uses_prior_custom_call_without_tool_message_name() { + let patch = "*** Begin Patch\n*** Add File: hello.txt\n+hello\n*** End Patch\n"; + let mut tc = ToolCall::new("call_001", "apply_patch", serde_json::json!(patch)); + tc.tool_type = "custom".to_string(); + tc.raw_arguments = Some(patch.to_string()); + tc.provider_metadata = Some(serde_json::json!({"id": "ctc_def456"})); + + let assistant_msg = Message { + role: Role::Assistant, + content: vec![ContentPart::ToolCall(tc)], + name: None, + tool_call_id: None, + }; + let tool_msg = Message::tool_result( + "call_001", + serde_json::json!("Success. Updated the following files:\nA hello.txt\n"), + false, + ); + + let (_, input) = translate_input(&[assistant_msg, tool_msg]); + + assert_eq!(input.len(), 2); + assert_eq!(input[1]["type"], "custom_tool_call_output"); + assert_eq!(input[1]["call_id"], "call_001"); + assert_eq!( + input[1]["output"], + "Success. Updated the following files:\nA hello.txt\n" + ); + } +} diff --git a/lib/crates/fabro-llm/src/codec/openai_responses/mod.rs b/lib/crates/fabro-llm/src/codec/openai_responses/mod.rs new file mode 100644 index 000000000..75491bf1e --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_responses/mod.rs @@ -0,0 +1,55 @@ +//! The OpenAI Responses (`/responses`) codec. +//! +//! Serves OpenAI direct today, in two route flavors that share this codec: +//! the standard route and the Codex route (`CodecParams::openai_codex`, which +//! omits sampling params encode-side; its forced streaming lives in the +//! adapter's route config). Pure translation: no HTTP, auth, or base URL — +//! the adapter shell owns those. +//! +//! HTTP error bodies use the shared `decode_error` default (openai uses the +//! standard `error_from_status_code` + `parse_error_body` path); streaming +//! `error` / `response.failed` events are mapped inside the decoder +//! (`on_event` → `Err`). + +mod decode; +mod encode; +mod stream; +mod wire; + +use crate::codec::{Codec, CodecCtx, EncodedRequest, StreamDecoder}; +use crate::error::Error; +use crate::types::{RateLimitInfo, Response}; + +/// Codec for the OpenAI Responses wire dialect. +pub(crate) struct OpenAiResponses; + +impl Codec for OpenAiResponses { + fn encode(&self, ctx: &CodecCtx<'_>, stream: bool) -> Result { + Ok(encode::encode(ctx, stream)) + } + + fn decode_response( + &self, + body: &str, + ctx: &CodecCtx<'_>, + rate_limit: Option, + ) -> Result { + decode::decode_response(body, ctx, rate_limit) + } + + fn stream_decoder( + &self, + ctx: &CodecCtx<'_>, + rate_limit: Option, + ) -> Box { + Box::new(stream::SseAccumulator::new(ctx, rate_limit)) + } + + fn encode_count_tokens(&self, ctx: &CodecCtx<'_>) -> Option> { + Some(Ok(encode::encode_count_tokens(ctx))) + } + + fn decode_count_tokens(&self, body: &str) -> Result { + decode::decode_count_tokens(body) + } +} diff --git a/lib/crates/fabro-llm/src/codec/openai_responses/stream.rs b/lib/crates/fabro-llm/src/codec/openai_responses/stream.rs new file mode 100644 index 000000000..f2e17fc7d --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_responses/stream.rs @@ -0,0 +1,868 @@ +//! Streaming decoder: OpenAI Responses SSE events → canonical `StreamEvent`s. +//! +//! Byte reading and SSE block framing live in the transport; this decoder is +//! fed framed `RawEvent`s. The event type is resolved from the SSE `event:` +//! line or the JSON `type` field. The Responses API finishes via +//! `response.completed` / `response.incomplete`; byte-stream end synthesizes +//! nothing, so `finish()` returns an empty list. + +use serde::Deserialize; + +use super::decode::{map_finish_reason, token_counts_from_api_usage, tool_call_from_item}; +use super::wire::ApiUsage; +use crate::codec::{CodecCtx, RawEvent, StreamDecoder}; +use crate::error::{Error, ProviderErrorDetail, ProviderErrorKind}; +use crate::types::{ + ContentPart, FinishReason, Message, RateLimitInfo, Response, Role, StreamEvent, TokenCounts, + ToolCall, +}; + +/// Map an OpenAI stream `error` / `response.failed` payload to a provider +/// error, classifying on `code` falling back to `type`. +fn provider_error_from_openai_error_json(error: &serde_json::Value, provider: &str) -> Error { + let classifier = error + .get("code") + .and_then(serde_json::Value::as_str) + .filter(|code| !code.is_empty()) + .or_else(|| { + error + .get("type") + .and_then(serde_json::Value::as_str) + .filter(|error_type| !error_type.is_empty()) + }); + let message = error + .get("message") + .and_then(serde_json::Value::as_str) + .filter(|message| !message.is_empty()) + .map_or_else(|| "OpenAI stream error".to_string(), str::to_string); + + let kind = match classifier { + Some("insufficient_quota" | "billing_hard_limit_reached") => { + ProviderErrorKind::QuotaExceeded + } + Some("rate_limit_error" | "rate_limit_exceeded" | "too_many_requests") => { + ProviderErrorKind::RateLimit + } + Some("authentication_error" | "invalid_api_key" | "invalid_authentication") => { + ProviderErrorKind::Authentication + } + Some( + "access_denied" | "account_deactivated" | "permission_denied" | "permission_error", + ) => ProviderErrorKind::AccessDenied, + Some("content_filter" | "content_policy_violation") => ProviderErrorKind::ContentFilter, + Some("context_length_exceeded") => ProviderErrorKind::ContextLength, + Some("server_error" | "internal_error" | "service_unavailable" | "engine_overloaded") => { + ProviderErrorKind::Server + } + Some(code) if code.ends_with("_not_found") => ProviderErrorKind::NotFound, + Some(code) + if code.starts_with("invalid_") + || code.starts_with("unsupported_") + || code.ends_with("_too_large") + || code.ends_with("_too_long") => + { + ProviderErrorKind::InvalidRequest + } + Some(_) | None => ProviderErrorKind::Server, + }; + + Error::Provider { + kind, + detail: Box::new(ProviderErrorDetail { + message, + provider: provider.to_string(), + status_code: None, + error_code: classifier.map(str::to_string), + retry_after: None, + raw: Some(error.clone()), + }), + } +} + +/// Accumulated state across SSE events during streaming. +pub(super) struct SseAccumulator { + /// Requested model, used as the fallback when the response omits one. + model: String, + /// Configured provider name stamped into responses and error details. + provider: String, + response_id: String, + response_model: String, + accumulated_text: String, + tool_calls: Vec, + /// Raw reasoning output items to preserve for round-tripping. + reasoning_items: Vec, + /// Raw message output items to preserve for round-tripping. + message_items: Vec, + usage: TokenCounts, + finish_reason: FinishReason, + emitted_start: bool, + emitted_text_start: bool, + emitted_reasoning_start: bool, + rate_limit: Option, +} + +impl SseAccumulator { + pub(super) fn new(ctx: &CodecCtx<'_>, rate_limit: Option) -> Self { + Self { + model: ctx.request.model.clone(), + provider: ctx.provider_name.to_string(), + response_id: String::new(), + response_model: String::new(), + accumulated_text: String::new(), + tool_calls: Vec::new(), + reasoning_items: Vec::new(), + message_items: Vec::new(), + usage: TokenCounts::default(), + finish_reason: FinishReason::Stop, + emitted_start: false, + emitted_text_start: false, + emitted_reasoning_start: false, + rate_limit, + } + } + + /// Process a single SSE event and return the corresponding + /// `StreamEvent`(s). + fn process_sse_event( + &mut self, + event_type: Option<&str>, + data: &str, + ) -> Result, Error> { + let mut events = Vec::new(); + + if !self.emitted_start { + self.emitted_start = true; + events.push(StreamEvent::StreamStart); + } + + let json: serde_json::Value = match serde_json::from_str(data) { + Ok(v) => v, + Err(_) => return Ok(events), + }; + + // Resolve event type from the `event:` SSE line or from the JSON `type` + // field. + let resolved_type = event_type + .or_else(|| json.get("type").and_then(serde_json::Value::as_str)) + .unwrap_or_default(); + + match resolved_type { + "error" => { + let error = json.get("error").unwrap_or(&json); + return Err(provider_error_from_openai_error_json(error, &self.provider)); + } + "response.created" => self.handle_response_created(&json), + "response.output_text.delta" => self.handle_text_delta(&json, &mut events), + "response.function_call_arguments.delta" => { + self.handle_tool_call_delta(&json, &mut events, "function"); + } + "response.custom_tool_call_input.delta" => { + self.handle_tool_call_delta(&json, &mut events, "custom"); + } + "response.output_item.done" => self.handle_output_item_done(&json, &mut events), + "response.completed" | "response.incomplete" => { + self.handle_response_completed(&json, &mut events); + } + "response.failed" => { + let error = json + .get("response") + .and_then(|response| response.get("error")) + .unwrap_or(&json); + return Err(provider_error_from_openai_error_json(error, &self.provider)); + } + "response.reasoning_summary_text.delta" | "response.reasoning_text.delta" => { + if let Some(delta) = json.get("delta").and_then(serde_json::Value::as_str) { + if !self.emitted_reasoning_start { + self.emitted_reasoning_start = true; + events.push(StreamEvent::ReasoningStart); + } + events.push(StreamEvent::ReasoningDelta { + delta: delta.to_string(), + }); + } + } + // response.reasoning_summary_part.added and other unrecognized + // events are no-ops + _ => {} + } + + Ok(events) + } + + /// Handle `response.created` by extracting the response ID and model. + fn handle_response_created(&mut self, json: &serde_json::Value) { + if let Some(id) = json + .get("response") + .and_then(|r| r.get("id")) + .and_then(serde_json::Value::as_str) + { + self.response_id = id.to_string(); + } + if let Some(model) = json + .get("response") + .and_then(|r| r.get("model")) + .and_then(serde_json::Value::as_str) + { + self.response_model = model.to_string(); + } + } + + /// Handle `response.output_text.delta` by accumulating text and emitting + /// events. + fn handle_text_delta(&mut self, json: &serde_json::Value, events: &mut Vec) { + if let Some(delta) = json.get("delta").and_then(serde_json::Value::as_str) { + if !self.emitted_text_start { + self.emitted_text_start = true; + events.push(StreamEvent::TextStart { text_id: None }); + } + self.accumulated_text.push_str(delta); + events.push(StreamEvent::text_delta(delta, None)); + } + } + + /// Handle `response.function_call_arguments.delta` / + /// `response.custom_tool_call_input.delta` by accumulating args and + /// emitting events. + fn handle_tool_call_delta( + &mut self, + json: &serde_json::Value, + events: &mut Vec, + tool_type: &str, + ) { + let Some(delta) = json.get("delta").and_then(serde_json::Value::as_str) else { + return; + }; + + let call_id = json + .get("call_id") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let item_id = json + .get("item_id") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let lookup_id = if call_id.is_empty() { item_id } else { call_id }; + + let idx = if let Some(idx) = self.tool_calls.iter().position(|tc| tc.id == lookup_id) { + let tc = &mut self.tool_calls[idx]; + if let Some(raw) = &mut tc.raw_arguments { + raw.push_str(delta); + } + // Custom tool input is its raw string; keep `arguments` in sync as + // it accumulates. + if tool_type == "custom" { + if let serde_json::Value::String(args) = &mut tc.arguments { + args.push_str(delta); + } + } + idx + } else { + let name = json + .get("name") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let mut tc = ToolCall::new( + lookup_id, + name, + if tool_type == "custom" { + serde_json::json!(delta) + } else { + serde_json::json!({}) + }, + ); + tc.tool_type = tool_type.to_string(); + tc.raw_arguments = Some(delta.to_string()); + // Preserve item-level ID (fc_xxx) for Responses API round-trip + if !item_id.is_empty() && item_id != lookup_id { + tc.provider_metadata = Some(serde_json::json!({"id": item_id})); + } + events.push(StreamEvent::ToolCallStart { + tool_call: tc.clone(), + }); + self.tool_calls.push(tc); + self.tool_calls.len() - 1 + }; + + // The delta event carries the call identity, the arguments + // accumulated so far, and this chunk in `raw_arguments`. + let current = &self.tool_calls[idx]; + let mut tool_call = ToolCall::new(&*current.id, &*current.name, current.arguments.clone()); + tool_call.tool_type = tool_type.to_string(); + tool_call.raw_arguments = Some(delta.to_string()); + tool_call + .provider_metadata + .clone_from(¤t.provider_metadata); + + events.push(StreamEvent::ToolCallDelta { tool_call }); + } + + /// Handle `response.output_item.done` for text and function call items. + fn handle_output_item_done(&mut self, json: &serde_json::Value, events: &mut Vec) { + let item = json.get("item").unwrap_or(json); + let item_type = item.get("type").and_then(serde_json::Value::as_str); + + match item_type { + Some("reasoning") => { + if self.emitted_reasoning_start { + self.emitted_reasoning_start = false; + events.push(StreamEvent::ReasoningEnd); + } + self.reasoning_items.push(item.clone()); + } + Some("message") => { + if self.emitted_text_start { + events.push(StreamEvent::TextEnd { text_id: None }); + self.emitted_text_start = false; + } + self.message_items.push(item.clone()); + } + Some(t @ ("function_call" | "custom_tool_call")) => { + let tc = tool_call_from_item(item, t == "custom_tool_call"); + + if let Some(existing) = self.tool_calls.iter_mut().find(|c| c.id == tc.id) { + existing.name.clone_from(&tc.name); + existing.tool_type.clone_from(&tc.tool_type); + existing.arguments = tc.arguments.clone(); + existing.raw_arguments.clone_from(&tc.raw_arguments); + existing.provider_metadata.clone_from(&tc.provider_metadata); + } else { + self.tool_calls.push(tc.clone()); + } + + events.push(StreamEvent::ToolCallEnd { tool_call: tc }); + } + _ => {} + } + } + + /// Handle `response.completed` / `response.incomplete` by extracting usage + /// and building the final response. + fn handle_response_completed( + &mut self, + json: &serde_json::Value, + events: &mut Vec, + ) { + let response_data = json.get("response").unwrap_or(json); + + if let Some(usage_data) = response_data.get("usage") { + if let Ok(u) = ApiUsage::deserialize(usage_data) { + self.usage = token_counts_from_api_usage(Some(&u)); + } + } + + if let Some(id) = response_data.get("id").and_then(serde_json::Value::as_str) { + self.response_id = id.to_string(); + } + if let Some(model) = response_data + .get("model") + .and_then(serde_json::Value::as_str) + { + self.response_model = model.to_string(); + } + + let status = response_data + .get("status") + .and_then(serde_json::Value::as_str); + let has_tool_calls = !self.tool_calls.is_empty(); + self.finish_reason = map_finish_reason(status, has_tool_calls); + + let mut content_parts = Vec::new(); + // Reasoning items must precede function calls for Responses API + // round-trip + for item in std::mem::take(&mut self.reasoning_items) { + content_parts.push(ContentPart::Other { + kind: ContentPart::OPENAI_REASONING.to_string(), + data: item, + }); + } + // Preserve full message output items for Responses API round-tripping + for item in std::mem::take(&mut self.message_items) { + content_parts.push(ContentPart::Other { + kind: ContentPart::OPENAI_MESSAGE.to_string(), + data: item, + }); + } + if !self.accumulated_text.is_empty() { + content_parts.push(ContentPart::text(std::mem::take( + &mut self.accumulated_text, + ))); + } + for tc in std::mem::take(&mut self.tool_calls) { + // Skip tool calls with empty names (e.g. model-internal items) + if tc.name.is_empty() { + continue; + } + content_parts.push(ContentPart::ToolCall(tc)); + } + + let model = if self.response_model.is_empty() { + self.model.clone() + } else { + self.response_model.clone() + }; + + let response = Response { + id: self.response_id.clone(), + model, + provider: self.provider.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: Some(response_data.clone()), + warnings: vec![], + rate_limit: self.rate_limit.clone(), + }; + + events.push(StreamEvent::finish( + self.finish_reason.clone(), + self.usage.clone(), + response, + )); + } +} + +impl StreamDecoder for SseAccumulator { + fn on_event(&mut self, ev: RawEvent<'_>) -> Result, Error> { + self.process_sse_event(ev.event, ev.data) + } + + fn finish(&mut self) -> Vec { + // The Responses API finishes via `response.completed`/`.incomplete`; + // nothing is synthesized at byte-stream end. + Vec::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Build an accumulator without threading a `CodecCtx`/`Request`: the test + /// module sees the private fields, so the few that matter are set + /// directly. `emitted_start` is true so event assertions don't see the + /// initial `StreamStart`. + fn empty_accumulator() -> SseAccumulator { + SseAccumulator { + model: String::new(), + provider: "openai".to_string(), + response_id: String::new(), + response_model: String::new(), + accumulated_text: String::new(), + tool_calls: Vec::new(), + reasoning_items: Vec::new(), + message_items: Vec::new(), + usage: TokenCounts::default(), + finish_reason: FinishReason::Stop, + emitted_start: true, + emitted_text_start: false, + emitted_reasoning_start: false, + rate_limit: None, + } + } + + fn on_event( + acc: &mut SseAccumulator, + event: Option<&str>, + data: &str, + ) -> Result, Error> { + acc.on_event(RawEvent { event, data }) + } + + #[test] + fn token_counts_disjoint_with_cache_and_reasoning() { + let mut acc = empty_accumulator(); + let body = serde_json::json!({ + "response": { + "id": "resp_test", + "model": "gpt-5", + "output": [], + "status": "completed", + "usage": { + "input_tokens": 200, + "input_tokens_details": { "cached_tokens": 180 }, + "output_tokens": 500, + "output_tokens_details": { "reasoning_tokens": 300 }, + "total_tokens": 700 + } + } + }); + let mut events = Vec::new(); + + acc.handle_response_completed(&body, &mut events); + + assert_eq!(acc.usage.input_tokens, 20); + assert_eq!(acc.usage.cache_read_tokens, 180); + assert_eq!(acc.usage.output_tokens, 200); + assert_eq!(acc.usage.reasoning_tokens, 300); + assert_eq!(acc.usage.cache_write_tokens, 0); + assert_eq!(acc.usage.total_tokens(), 700); + } + + #[test] + fn custom_tool_call_streaming_delta_accumulates_raw_input() { + let mut acc = empty_accumulator(); + let first = r#"{ + "type": "response.custom_tool_call_input.delta", + "item_id": "ctc_abc", + "call_id": "call_001", + "delta": "*** Begin" + }"#; + let second = r#"{ + "type": "response.custom_tool_call_input.delta", + "item_id": "ctc_abc", + "call_id": "call_001", + "delta": " Patch\n" + }"#; + + let first_events = on_event( + &mut acc, + Some("response.custom_tool_call_input.delta"), + first, + ) + .expect("first custom delta should parse"); + let second_events = on_event( + &mut acc, + Some("response.custom_tool_call_input.delta"), + second, + ) + .expect("second custom delta should parse"); + + assert!(matches!( + first_events.iter().find(|event| matches!(event, StreamEvent::ToolCallStart { .. })), + Some(StreamEvent::ToolCallStart { tool_call }) + if tool_call.id == "call_001" && tool_call.tool_type == "custom" + )); + assert!(matches!( + second_events.last(), + Some(StreamEvent::ToolCallDelta { tool_call }) + if tool_call.raw_arguments.as_deref() == Some(" Patch\n") + && tool_call.tool_type == "custom" + )); + assert_eq!( + acc.tool_calls[0].raw_arguments.as_deref(), + Some("*** Begin Patch\n") + ); + } + + #[test] + fn custom_tool_call_output_item_done_emits_tool_call_end() { + let mut acc = empty_accumulator(); + let patch = "*** Begin Patch\n*** Add File: hello.txt\n+hello\n*** End Patch\n"; + let data = serde_json::json!({ + "type": "response.output_item.done", + "item": { + "type": "custom_tool_call", + "id": "ctc_abc", + "call_id": "call_001", + "name": "apply_patch", + "input": patch, + } + }); + + let events = on_event( + &mut acc, + Some("response.output_item.done"), + &data.to_string(), + ) + .expect("custom output item should parse"); + + assert!(matches!( + events.last(), + Some(StreamEvent::ToolCallEnd { tool_call }) + if tool_call.id == "call_001" + && tool_call.name == "apply_patch" + && tool_call.tool_type == "custom" + && tool_call.raw_arguments.as_deref() == Some(patch) + )); + } + + #[test] + fn error_event_with_insufficient_quota_returns_provider_error() { + let mut acc = empty_accumulator(); + let data = r#"{ + "type": "error", + "error": { + "type": "insufficient_quota", + "code": "insufficient_quota", + "message": "You exceeded your current quota.", + "param": null + } + }"#; + + let err = on_event(&mut acc, Some("error"), data) + .expect_err("error event should fail the stream"); + + match err { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::QuotaExceeded); + assert!(detail.message.contains("exceeded your current quota")); + assert_eq!(detail.error_code.as_deref(), Some("insufficient_quota")); + assert!(detail.raw.is_some()); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn error_event_classifies_on_type_when_code_absent() { + let mut acc = empty_accumulator(); + let data = r#"{ + "type": "error", + "error": { + "type": "insufficient_quota", + "message": "You exceeded your current quota." + } + }"#; + + let err = on_event(&mut acc, Some("error"), data) + .expect_err("error event should fail the stream"); + + match err { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::QuotaExceeded); + assert_eq!(detail.error_code.as_deref(), Some("insufficient_quota")); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn response_failed_event_with_server_error_returns_provider_error() { + let mut acc = empty_accumulator(); + let data = r#"{ + "type": "response.failed", + "response": { + "status": "failed", + "error": { + "type": "server_error", + "code": "server_error", + "message": "The server had an error while processing your request." + } + } + }"#; + + let err = on_event(&mut acc, Some("response.failed"), data) + .expect_err("response.failed should fail the stream"); + + match err { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::Server); + assert!(detail.message.contains("server had an error")); + assert_eq!(detail.error_code.as_deref(), Some("server_error")); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn response_incomplete_preserves_partial_text() { + let mut acc = empty_accumulator(); + + on_event( + &mut acc, + Some("response.created"), + r#"{"type":"response.created","response":{"id":"resp_123","model":"gpt-5.4"}}"#, + ) + .expect("created event should parse"); + on_event( + &mut acc, + Some("response.output_text.delta"), + r#"{"type":"response.output_text.delta","delta":"Hel"}"#, + ) + .expect("first delta should parse"); + on_event( + &mut acc, + Some("response.output_text.delta"), + r#"{"type":"response.output_text.delta","delta":"lo"}"#, + ) + .expect("second delta should parse"); + + let events = on_event( + &mut acc, + Some("response.incomplete"), + r#"{ + "type": "response.incomplete", + "response": { + "id": "resp_123", + "model": "gpt-5.4", + "status": "incomplete" + } + }"#, + ) + .expect("incomplete response should finish normally"); + + let finish = events + .last() + .expect("incomplete response should emit finish"); + match finish { + StreamEvent::Finish { + finish_reason, + response, + .. + } => { + assert_eq!(finish_reason.clone(), FinishReason::Length); + assert_eq!(response.text(), "Hello"); + } + other => panic!("expected finish event, got {other:?}"), + } + } + + #[test] + fn error_event_with_invalid_api_key_returns_authentication_error() { + let mut acc = empty_accumulator(); + let data = r#"{ + "type": "error", + "error": { + "type": "invalid_api_key", + "code": "invalid_api_key", + "message": "Incorrect API key provided." + } + }"#; + + let err = on_event(&mut acc, Some("error"), data) + .expect_err("error event should fail the stream"); + + match err { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::Authentication); + assert_eq!(detail.error_code.as_deref(), Some("invalid_api_key")); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn error_event_with_rate_limit_error_returns_rate_limit() { + let mut acc = empty_accumulator(); + let data = r#"{ + "type": "error", + "error": { + "type": "rate_limit_error", + "message": "Too many requests." + } + }"#; + + let err = on_event(&mut acc, Some("error"), data) + .expect_err("error event should fail the stream"); + + match err { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::RateLimit); + assert_eq!(detail.error_code.as_deref(), Some("rate_limit_error")); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn error_event_with_unknown_invalid_prefix_returns_invalid_request() { + let mut acc = empty_accumulator(); + let data = r#"{ + "type": "error", + "error": { + "type": "invalid_prompt", + "code": "invalid_prompt", + "message": "Prompt is invalid." + } + }"#; + + let err = on_event(&mut acc, Some("error"), data) + .expect_err("error event should fail the stream"); + + match err { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::InvalidRequest); + assert_eq!(detail.error_code.as_deref(), Some("invalid_prompt")); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn error_event_with_unknown_code_falls_back_to_server_with_message() { + let mut acc = empty_accumulator(); + let data = r#"{ + "type": "error", + "error": { + "type": "unexpected_stream_failure", + "code": "unexpected_stream_failure", + "message": "Unexpected stream failure." + } + }"#; + + let err = on_event(&mut acc, Some("error"), data) + .expect_err("error event should fail the stream"); + + match err { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::Server); + assert_eq!(detail.message, "Unexpected stream failure."); + assert_eq!( + detail.error_code.as_deref(), + Some("unexpected_stream_failure") + ); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn reasoning_summary_delta_emits_reasoning_events() { + let mut acc = empty_accumulator(); + let data = r#"{"type":"response.reasoning_summary_text.delta","delta":"Let me think"}"#; + let events = on_event( + &mut acc, + Some("response.reasoning_summary_text.delta"), + data, + ) + .expect("reasoning summary delta should parse"); + assert_eq!(events.len(), 2); + assert!(matches!(events[0], StreamEvent::ReasoningStart)); + assert!( + matches!(events[1], StreamEvent::ReasoningDelta { ref delta } if delta == "Let me think") + ); + } + + #[test] + fn reasoning_text_delta_emits_reasoning_events() { + let mut acc = empty_accumulator(); + + // First delta: should emit ReasoningStart + ReasoningDelta + let data1 = r#"{"type":"response.reasoning_text.delta","delta":"Step 1"}"#; + let events1 = on_event(&mut acc, Some("response.reasoning_text.delta"), data1) + .expect("first reasoning delta should parse"); + assert_eq!(events1.len(), 2); + assert!(matches!(events1[0], StreamEvent::ReasoningStart)); + assert!( + matches!(events1[1], StreamEvent::ReasoningDelta { ref delta } if delta == "Step 1") + ); + + // Second delta: should NOT emit duplicate ReasoningStart + let data2 = r#"{"type":"response.reasoning_text.delta","delta":"Step 2"}"#; + let events2 = on_event(&mut acc, Some("response.reasoning_text.delta"), data2) + .expect("second reasoning delta should parse"); + assert_eq!(events2.len(), 1); + assert!( + matches!(events2[0], StreamEvent::ReasoningDelta { ref delta } if delta == "Step 2") + ); + } + + #[test] + fn reasoning_end_emitted_on_item_done() { + let mut acc = empty_accumulator(); + acc.emitted_reasoning_start = true; + + let data = r#"{"item":{"type":"reasoning","id":"rs_abc","summary":[]}}"#; + let events = on_event(&mut acc, Some("response.output_item.done"), data) + .expect("output item done should parse"); + assert_eq!(events.len(), 1); + assert!(matches!(events[0], StreamEvent::ReasoningEnd)); + assert!(!acc.emitted_reasoning_start); + assert_eq!(acc.reasoning_items.len(), 1); + } +} diff --git a/lib/crates/fabro-llm/src/codec/openai_responses/wire.rs b/lib/crates/fabro-llm/src/codec/openai_responses/wire.rs new file mode 100644 index 000000000..f1c1e82e6 --- /dev/null +++ b/lib/crates/fabro-llm/src/codec/openai_responses/wire.rs @@ -0,0 +1,67 @@ +//! Serde types mirroring the OpenAI Responses API wire shapes. + +#[derive(serde::Serialize)] +pub(super) struct ApiRequest { + pub model: String, + pub input: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub instructions: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub temperature: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_output_tokens: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub top_p: 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 reasoning: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub text: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub stop: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub metadata: Option>, + pub store: bool, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub include: Vec, + #[serde(skip_serializing_if = "std::ops::Not::not")] + pub stream: bool, +} + +// --- Response types --- + +#[derive(serde::Deserialize)] +pub(super) struct ApiResponse { + pub id: String, + pub model: Option, + pub output: Vec, + pub status: Option, + pub usage: Option, +} + +#[derive(serde::Deserialize)] +pub(super) struct InputTokensResponse { + pub input_tokens: i64, + pub object: String, +} + +#[derive(serde::Deserialize)] +pub(super) struct ApiUsage { + pub input_tokens: i64, + pub output_tokens: i64, + pub output_tokens_details: Option, + pub input_tokens_details: Option, +} + +#[derive(serde::Deserialize)] +pub(super) struct OutputTokenDetails { + pub reasoning_tokens: Option, +} + +#[derive(serde::Deserialize)] +pub(super) struct InputTokenDetails { + pub cached_tokens: Option, +} diff --git a/lib/crates/fabro-llm/src/providers/anthropic.rs b/lib/crates/fabro-llm/src/providers/anthropic.rs index f5d4d79a1..beb2847ec 100644 --- a/lib/crates/fabro-llm/src/providers/anthropic.rs +++ b/lib/crates/fabro-llm/src/providers/anthropic.rs @@ -94,7 +94,8 @@ impl Adapter { auth: AuthScheme::ApiKey, codec_params: CodecParams { anthropic_version: AnthropicVersion::Header("2023-06-01"), - anthropic_beta: true, + anthropic_beta: true, + ..CodecParams::default() }, supports_count_tokens: true, force_streaming: false, diff --git a/lib/crates/fabro-llm/src/providers/openai.rs b/lib/crates/fabro-llm/src/providers/openai.rs index c9753b856..24e02a6ac 100644 --- a/lib/crates/fabro-llm/src/providers/openai.rs +++ b/lib/crates/fabro-llm/src/providers/openai.rs @@ -1,30 +1,31 @@ -use std::collections::HashMap; use std::sync::Arc; -use base64::Engine; -use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use fabro_model::Catalog; -use futures::{StreamExt, stream}; +use futures::stream; -use crate::error::{Error, ProviderErrorDetail, ProviderErrorKind, error_from_status_code}; +use crate::attachments::{self, AttachmentPolicy}; +use crate::codec::openai_responses::OpenAiResponses; +use crate::codec::{Codec, CodecCtx, CodecParams, EncodedRequest, RawEvent, StreamDecoder}; +use crate::error::Error; use crate::provider::{ ProviderAdapter, StreamEventStream, validate_standard_speed, validate_tool_choice, }; use crate::providers::common::{ - self as common, parse_error_body, parse_rate_limit_headers, parse_retry_after, - send_and_read_response, send_and_read_response_with_operation, + self as common, parse_rate_limit_headers, parse_retry_after, send_and_read_response, + send_and_read_response_with_operation, }; use crate::token_count::{InputTokenCount, InputTokenCountMethod}; -use crate::types::{ - AdapterTimeout, ContentPart, FinishReason, Message, RateLimitInfo, Request, Response, - ResponseFormat, ResponseFormatType, Role, StreamEvent, TokenCounts, ToolCall, ToolChoice, - ToolDefinition, -}; +use crate::types::{AdapterTimeout, Request, Response, StreamEvent}; const DEFAULT_BASE_URL: &str = "https://api.openai.com/v1"; /// Provider adapter for the `OpenAI` Responses API (`/v1/responses`). /// +/// A thin transport shell over the `openai_responses` codec: it owns auth +/// (bearer + org/project headers), base URL, the streaming byte loop, and the +/// route configuration for codex mode. All wire translation lives in the +/// codec. +/// /// Per spec Section 2.7, this adapter uses the Responses API (not Chat /// Completions) to properly surface reasoning tokens, built-in tools, and /// server-side state. @@ -108,10 +109,54 @@ impl Adapter { } } - /// Build a `fabro_http::RequestBuilder` with default headers, org/project - /// headers, and auth. - fn build_request(&self, url: &str) -> fabro_http::RequestBuilder { - let mut req = self.http.client.post(url); + /// Per-route dialect knobs for the codec. + /// + /// OpenAI has a single auth scheme (bearer + org/project headers), so the + /// only route variation is codex mode: its encode-side half (param + /// omission) rides on `CodecParams`; its transport-side half (forced + /// streaming) is checked directly off `codex_mode` in `complete`. + fn codec_params(&self) -> CodecParams { + CodecParams { + openai_codex: self.codex_mode, + ..CodecParams::default() + } + } + + /// 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: common::catalog_model(self.catalog.as_deref(), &request.model), + params, + } + } + + /// Build the canonical request for the codec, resolving file-backed + /// attachments to inline data first. Borrowed when nothing needs loading. + async fn resolve_request<'a>(&self, request: &'a Request) -> std::borrow::Cow<'a, Request> { + // OpenAI loads images inline; audio and documents render as text + // placeholders in the codec, so they are not loaded here. + let policy = AttachmentPolicy { + images: true, + documents: false, + audio: false, + }; + attachments::resolve(request, policy).await + } + + /// Apply the base URL, auth (bearer + org/project headers), and + /// codec-emitted headers to an encoded request. + fn build_http_request(&self, encoded: &EncodedRequest) -> fabro_http::RequestBuilder { + let url = format!("{}{}", self.http.base_url, encoded.endpoint); + let mut req = self.http.client.post(&url); // Apply default_headers first so adapter-specific headers can override for (key, value) in &self.http.default_headers { req = req.header(key, value); @@ -125,7 +170,10 @@ impl Adapter { if let Some(project_id) = &self.project_id { req = req.header("OpenAI-Project", project_id); } - req + for (key, value) in &encoded.headers { + req = req.header(key, value); + } + req.json(&encoded.body) } /// Complete a request by streaming and collecting the final response. @@ -147,655 +195,14 @@ impl Adapter { } } -// --- Request types (Responses API format) --- - -#[derive(serde::Serialize)] -struct ApiRequest { - model: String, - input: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - instructions: Option, - #[serde(skip_serializing_if = "Option::is_none")] - temperature: Option, - #[serde(skip_serializing_if = "Option::is_none")] - max_output_tokens: Option, - #[serde(skip_serializing_if = "Option::is_none")] - top_p: 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")] - reasoning: Option, - #[serde(skip_serializing_if = "Option::is_none")] - text: Option, - #[serde(skip_serializing_if = "Option::is_none")] - stop: Option>, - #[serde(skip_serializing_if = "Option::is_none")] - metadata: Option>, - store: bool, - #[serde(skip_serializing_if = "Vec::is_empty")] - include: Vec, - #[serde(skip_serializing_if = "std::ops::Not::not")] - stream: bool, -} - -// --- Response types (Responses API format) --- - -#[derive(serde::Deserialize)] -struct ApiResponse { - id: String, - model: Option, - output: Vec, - status: Option, - usage: Option, -} - -#[derive(serde::Deserialize)] -struct InputTokensResponse { - input_tokens: i64, - object: String, -} - -#[derive(serde::Deserialize)] -struct ApiUsage { - input_tokens: i64, - output_tokens: i64, - output_tokens_details: Option, - input_tokens_details: Option, -} - -#[derive(serde::Deserialize)] -struct OutputTokenDetails { - reasoning_tokens: Option, -} - -#[derive(serde::Deserialize)] -struct InputTokenDetails { - cached_tokens: Option, -} - -fn token_counts_from_api_usage(usage: Option<&ApiUsage>) -> TokenCounts { - usage.map_or_else(TokenCounts::default, |u| { - let cached_tokens = u - .input_tokens_details - .as_ref() - .and_then(|d| d.cached_tokens) - .unwrap_or(0); - let reasoning_tokens = u - .output_tokens_details - .as_ref() - .and_then(|d| d.reasoning_tokens) - .unwrap_or(0); - TokenCounts { - input_tokens: u.input_tokens.saturating_sub(cached_tokens), - output_tokens: u.output_tokens.saturating_sub(reasoning_tokens), - reasoning_tokens, - cache_read_tokens: cached_tokens, - ..TokenCounts::default() - } - }) -} - -/// Map the Responses API status to a `FinishReason`. -fn map_finish_reason(status: Option<&str>, has_tool_calls: bool) -> FinishReason { - if has_tool_calls { - return FinishReason::ToolCalls; - } - match status { - Some("completed") | None => FinishReason::Stop, - Some("incomplete") => FinishReason::Length, - Some("failed") => FinishReason::Error, - Some(other) => FinishReason::Other(other.to_string()), - } -} - -fn provider_error_from_openai_error_json(error: &serde_json::Value, provider: &str) -> Error { - let classifier = error - .get("code") - .and_then(serde_json::Value::as_str) - .filter(|code| !code.is_empty()) - .or_else(|| { - error - .get("type") - .and_then(serde_json::Value::as_str) - .filter(|error_type| !error_type.is_empty()) - }); - let message = error - .get("message") - .and_then(serde_json::Value::as_str) - .filter(|message| !message.is_empty()) - .map_or_else(|| "OpenAI stream error".to_string(), str::to_string); - - let kind = match classifier { - Some("insufficient_quota" | "billing_hard_limit_reached") => { - ProviderErrorKind::QuotaExceeded - } - Some("rate_limit_error" | "rate_limit_exceeded" | "too_many_requests") => { - ProviderErrorKind::RateLimit - } - Some("authentication_error" | "invalid_api_key" | "invalid_authentication") => { - ProviderErrorKind::Authentication - } - Some( - "access_denied" | "account_deactivated" | "permission_denied" | "permission_error", - ) => ProviderErrorKind::AccessDenied, - Some("content_filter" | "content_policy_violation") => ProviderErrorKind::ContentFilter, - Some("context_length_exceeded") => ProviderErrorKind::ContextLength, - Some("server_error" | "internal_error" | "service_unavailable" | "engine_overloaded") => { - ProviderErrorKind::Server - } - Some(code) if code.ends_with("_not_found") => ProviderErrorKind::NotFound, - Some(code) - if code.starts_with("invalid_") - || code.starts_with("unsupported_") - || code.ends_with("_too_large") - || code.ends_with("_too_long") => - { - ProviderErrorKind::InvalidRequest - } - Some(_) | None => ProviderErrorKind::Server, - }; - - Error::Provider { - kind, - detail: Box::new(ProviderErrorDetail { - message, - provider: provider.to_string(), - status_code: None, - error_code: classifier.map(str::to_string), - retry_after: None, - raw: Some(error.clone()), - }), - } -} - -/// Translate unified messages to Responses API `input` array format. -async fn translate_input(messages: &[Message]) -> (Option, Vec) { - let mut instructions_parts: Vec = Vec::new(); - let mut input: Vec = Vec::new(); - let mut tool_call_types: HashMap = HashMap::new(); - - for msg in messages { - match msg.role { - Role::System | Role::Developer => { - instructions_parts.push(msg.text()); - } - Role::User => { - let mut content = Vec::new(); - for part in &msg.content { - let maybe_content = match part { - ContentPart::Text(text) => { - Some(serde_json::json!({"type": "input_text", "text": text})) - } - ContentPart::Image(img) => match &img.url { - Some(url) => { - if common::is_file_path(url) { - match common::load_file_as_base64(url).await { - Ok((b64, mime)) => Some(serde_json::json!({ - "type": "input_image", - "image_url": format!("data:{mime};base64,{b64}"), - })), - Err(_) => None, - } - } else { - Some( - serde_json::json!({"type": "input_image", "image_url": url}), - ) - } - } - None => img.data.as_ref().map(|data| { - let mime = img.media_type.as_deref().unwrap_or("image/png"); - let b64 = BASE64_STANDARD.encode(data); - serde_json::json!({ - "type": "input_image", - "image_url": format!("data:{mime};base64,{b64}"), - }) - }), - }, - ContentPart::Audio(_) => Some( - serde_json::json!({"type": "input_text", "text": "[Audio content not supported by this provider]"}), - ), - 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]"), - ); - Some(serde_json::json!({"type": "input_text", "text": desc})) - } - _ => None, - }; - if let Some(content_part) = maybe_content { - content.push(content_part); - } - } - if !content.is_empty() { - input.push(serde_json::json!({ - "type": "message", - "role": "user", - "content": content, - })); - } - } - Role::Assistant => { - // If we have a preserved opaque message item (with id/status), use - // it instead of constructing a new message from Text parts. This is - // required so that reasoning items can find their "required following - // item" during Responses API round-tripping. - let has_opaque_message = msg.content.iter().any(|p| { - matches!(p, ContentPart::Other { kind, .. } if kind == ContentPart::OPENAI_MESSAGE) - }); - for part in &msg.content { - match part { - ContentPart::Text(text) if !has_opaque_message => { - input.push(serde_json::json!({ - "type": "message", - "role": "assistant", - "content": [{"type": "output_text", "text": text}], - })); - } - ContentPart::ToolCall(tc) if !tc.name.is_empty() => { - // Use the item-level ID (fc_xxx) for the `id` field; - // fall back to tc.id if no provider_metadata was stored. - let item_id = tc - .provider_metadata - .as_ref() - .and_then(|m| m.get("id")) - .and_then(serde_json::Value::as_str) - .unwrap_or(&tc.id); - tool_call_types - .insert(tc.id.clone(), (tc.tool_type.clone(), tc.name.clone())); - if tc.tool_type == "custom" { - let raw_input = tc.raw_arguments.as_ref().map_or_else( - || { - tc.arguments.as_str().map_or_else( - || tc.arguments.to_string(), - str::to_string, - ) - }, - Clone::clone, - ); - input.push(serde_json::json!({ - "type": "custom_tool_call", - "id": item_id, - "call_id": tc.id, - "name": tc.name, - "input": raw_input, - })); - } else { - let args = tc - .raw_arguments - .as_ref() - .map_or_else(|| tc.arguments.to_string(), Clone::clone); - input.push(serde_json::json!({ - "type": "function_call", - "id": item_id, - "call_id": tc.id, - "name": tc.name, - "arguments": args, - })); - } - } - ContentPart::Other { data, .. } if part.is_opaque_openai() => { - input.push(data.clone()); - } - _ => {} - } - } - } - Role::Tool => { - for part in &msg.content { - if let ContentPart::ToolResult(tr) = part { - let output = tr - .content - .as_str() - .map_or_else(|| tr.content.to_string(), str::to_string); - let is_custom = tool_call_types - .get(&tr.tool_call_id) - .is_some_and(|(tool_type, _)| tool_type == "custom") - || msg.name.as_deref() == Some("apply_patch"); - let mut item = if is_custom { - serde_json::json!({ - "type": "custom_tool_call_output", - "call_id": tr.tool_call_id, - "output": output, - }) - } else { - serde_json::json!({ - "type": "function_call_output", - "call_id": tr.tool_call_id, - "output": output, - }) - }; - if tr.is_error && !is_custom { - item["status"] = serde_json::json!("incomplete"); - } - input.push(item); - } - } - } - } - } - - let instructions = if instructions_parts.is_empty() { - None - } else { - Some(instructions_parts.join("\n")) - }; - - (instructions, input) -} - -/// Translate unified tool definitions to Responses API tool format. -fn translate_tools(tools: &[ToolDefinition]) -> Vec { - tools - .iter() - .map(|t| { - if t.is_custom() { - serde_json::json!({ - "type": "custom", - "name": t.name, - "description": t.description, - "format": t.custom_format().cloned().unwrap_or_else(|| serde_json::json!({})), - }) - } else { - serde_json::json!({ - "type": "function", - "name": t.name, - "description": t.description, - "parameters": t.parameters, - }) - } - }) - .collect() -} - -/// Translate unified `ToolChoice` to Responses API format. -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", "name": tool_name}) - } - } -} - -/// Translate unified `ResponseFormat` to Responses API `text` field. -/// -/// The Responses API uses `"text": {"format": {...}}` for structured output. -fn translate_response_format(format: &ResponseFormat) -> Option { - match format.kind { - ResponseFormatType::Text => None, - ResponseFormatType::JsonObject => { - Some(serde_json::json!({"format": {"type": "json_object"}})) - } - ResponseFormatType::JsonSchema => { - let mut schema_obj = serde_json::json!({ - "type": "json_schema", - "name": "response", - "strict": format.strict, - }); - if let Some(schema) = &format.json_schema { - schema_obj["schema"] = schema.clone(); - } - Some(serde_json::json!({"format": schema_obj})) - } - } -} - -/// Build an `ApiRequest` from a unified `Request`. -/// -/// When `codex_mode` is true, unsupported fields (`temperature`, -/// `max_output_tokens`, `top_p`) are omitted and empty instructions are sent as -/// `""` (required by the Codex endpoint). -async fn build_api_request( - request: &Request, - stream: bool, - codex_mode: bool, - catalog: Option<&Catalog>, -) -> ApiRequest { - let (instructions, input) = translate_input(&request.messages).await; - let api_tools = request.tools.as_ref().map(|t| translate_tools(t)); - let tool_choice = request.tool_choice.as_ref().map(translate_tool_choice); - let reasoning = request - .reasoning_effort - .as_ref() - .map(|effort| serde_json::json!({"effort": <&'static str>::from(*effort)})); - let text = request - .response_format - .as_ref() - .and_then(translate_response_format); - - let include = vec!["reasoning.encrypted_content".to_string()]; - - let instructions = if codex_mode { - Some(instructions.unwrap_or_default()) - } else { - instructions - }; - - ApiRequest { - model: common::api_model_id(catalog, &request.model), - input, - instructions, - temperature: if codex_mode { - None - } else { - request.temperature - }, - max_output_tokens: if codex_mode { None } else { request.max_tokens }, - top_p: if codex_mode { None } else { request.top_p }, - tools: api_tools, - tool_choice, - reasoning, - text, - stop: request.stop_sequences.clone(), - metadata: request.metadata.clone(), - // store: false means output items are not persisted server-side. - // Request encrypted reasoning content on every turn so reasoning items - // from models that emit them by default can round-trip statelessly. - store: false, - include, - stream, - } -} - -/// Serialize an `ApiRequest` to JSON and merge any `provider_options.openai` -/// keys into it. -#[cfg(test)] -async fn build_request_body( - request: &Request, - stream: bool, - codex_mode: bool, -) -> serde_json::Value { - build_request_body_with_catalog(request, stream, codex_mode, None).await -} - -async fn build_request_body_with_catalog( - request: &Request, - stream: bool, - codex_mode: bool, - catalog: Option<&Catalog>, -) -> serde_json::Value { - let api_request = build_api_request(request, stream, codex_mode, catalog).await; - let mut body = serde_json::to_value(&api_request).unwrap_or_else(|_| serde_json::json!({})); - - if let Some(openai_opts) = request - .provider_options - .as_ref() - .and_then(|opts| opts.get("openai")) - { - if let (Some(base), Some(overrides)) = (body.as_object_mut(), openai_opts.as_object()) { - for (key, value) in overrides { - base.insert(key.clone(), value.clone()); - } - } - } - - body -} - -fn filter_input_tokens_request_body(body: &serde_json::Value) -> serde_json::Value { - const ALLOWED_FIELDS: &[&str] = &[ - "conversation", - "input", - "instructions", - "model", - "parallel_tool_calls", - "previous_response_id", - "reasoning", - "text", - "tool_choice", - "tools", - "truncation", - ]; - - let Some(source) = body.as_object() else { - return serde_json::json!({}); - }; - - let mut filtered = serde_json::Map::new(); - for field in ALLOWED_FIELDS { - if let Some(value) = source.get(*field) { - filtered.insert((*field).to_string(), value.clone()); - } - } - serde_json::Value::Object(filtered) -} - -/// Parse output items from the Responses API into content parts. -fn parse_output(output: &[serde_json::Value]) -> (Vec, bool) { - let mut parts = Vec::new(); - let mut has_tool_calls = false; - - for item in output { - let item_type = item.get("type").and_then(serde_json::Value::as_str); - match item_type { - Some("message") => { - // Preserve the full message item for Responses API round-tripping. - // The item's `id` and `status` fields are required so that reasoning - // items preceding it can find their "required following item." - parts.push(ContentPart::Other { - kind: ContentPart::OPENAI_MESSAGE.to_string(), - data: item.clone(), - }); - if let Some(content) = item.get("content").and_then(|c| c.as_array()) { - for block in content { - if block.get("type").and_then(serde_json::Value::as_str) - == Some("output_text") - { - if let Some(text) = - block.get("text").and_then(serde_json::Value::as_str) - { - parts.push(ContentPart::text(text)); - } - } - } - } - } - Some("reasoning") => { - parts.push(ContentPart::Other { - kind: ContentPart::OPENAI_REASONING.to_string(), - data: item.clone(), - }); - } - Some("function_call") => { - let item_id = item - .get("id") - .and_then(serde_json::Value::as_str) - .unwrap_or(""); - let call_id = item - .get("call_id") - .and_then(serde_json::Value::as_str) - .unwrap_or(item_id) - .to_string(); - let name = item - .get("name") - .and_then(serde_json::Value::as_str) - .unwrap_or("") - .to_string(); - // Skip function calls with empty names (e.g. model-internal items) - if name.is_empty() { - continue; - } - has_tool_calls = true; - let args_str = item - .get("arguments") - .and_then(serde_json::Value::as_str) - .unwrap_or("{}"); - let arguments = - serde_json::from_str(args_str).unwrap_or_else(|_| serde_json::json!({})); - let mut tc = ToolCall::new(call_id, name, arguments); - tc.raw_arguments = Some(args_str.to_string()); - // Preserve item-level ID (fc_xxx) for Responses API round-trip - if !item_id.is_empty() { - tc.provider_metadata = Some(serde_json::json!({"id": item_id})); - } - parts.push(ContentPart::ToolCall(tc)); - } - Some("custom_tool_call") => { - let item_id = item - .get("id") - .and_then(serde_json::Value::as_str) - .unwrap_or(""); - let call_id = item - .get("call_id") - .and_then(serde_json::Value::as_str) - .unwrap_or(item_id) - .to_string(); - let name = item - .get("name") - .and_then(serde_json::Value::as_str) - .unwrap_or("") - .to_string(); - if name.is_empty() { - continue; - } - has_tool_calls = true; - let raw_input = item - .get("input") - .and_then(serde_json::Value::as_str) - .unwrap_or(""); - let mut tc = ToolCall::new(call_id, name, serde_json::json!(raw_input)); - tc.tool_type = "custom".to_string(); - tc.raw_arguments = Some(raw_input.to_string()); - if !item_id.is_empty() { - tc.provider_metadata = Some(serde_json::json!({"id": item_id})); - } - parts.push(ContentPart::ToolCall(tc)); - } - _ => {} - } - } - - (parts, has_tool_calls) -} - -// --- SSE streaming support --- - -/// Mutable state carried through SSE stream processing. -struct SseStreamState { - line_reader: super::common::LineReader, - model: String, - /// Configured provider name stamped into responses and error details. - provider: String, - response_id: String, - response_model: String, - accumulated_text: String, - tool_calls: Vec, - /// Raw reasoning output items to preserve for round-tripping. - reasoning_items: Vec, - /// Raw message output items to preserve for round-tripping. - message_items: Vec, - usage: TokenCounts, - finish_reason: FinishReason, - emitted_start: bool, - emitted_text_start: bool, - emitted_reasoning_start: bool, - raw_response: Option, - rate_limit: Option, +/// State driving the streaming byte loop: the codec's decoder plus the line +/// reader, with a buffer that flattens batched events into individual items. +struct StreamLoop { + decoder: Box, + line_reader: super::common::LineReader, + pending: std::collections::VecDeque, + done: bool, + finished_emitted: bool, } /// Parse a single SSE message block into an (`event_type`, `data`) pair. @@ -831,415 +238,6 @@ fn parse_sse_message(message_block: &str) -> Option<(Option, String)> { } } -/// Process the next chunk(s) from the byte stream and return `StreamEvent`s. -async fn process_next_sse_events(state: &mut SseStreamState) -> Result, Error> { - loop { - match state.line_reader.read_next_chunk("\n\n").await? { - Some(message_block) => { - if let Some((event_type, data)) = parse_sse_message(&message_block) { - let events = process_sse_event(state, event_type.as_deref(), &data)?; - if !events.is_empty() { - return Ok(events); - } - } - // No data or unhandled event type; keep reading. - } - None => return Ok(vec![]), - } - } -} - -/// Process a single SSE event and return the corresponding `StreamEvent`(s). -fn process_sse_event( - state: &mut SseStreamState, - event_type: Option<&str>, - data: &str, -) -> Result, Error> { - let mut events = Vec::new(); - - if !state.emitted_start { - state.emitted_start = true; - events.push(StreamEvent::StreamStart); - } - - let json: serde_json::Value = match serde_json::from_str(data) { - Ok(v) => v, - Err(_) => return Ok(events), - }; - - // Resolve event type from the `event:` SSE line or from the JSON `type` field. - let resolved_type = event_type - .map(str::to_string) - .or_else(|| { - json.get("type") - .and_then(serde_json::Value::as_str) - .map(str::to_string) - }) - .unwrap_or_default(); - - match resolved_type.as_str() { - "error" => { - let error = json.get("error").unwrap_or(&json); - return Err(provider_error_from_openai_error_json( - error, - &state.provider, - )); - } - "response.created" => handle_response_created(state, &json), - "response.output_text.delta" => handle_text_delta(state, &json, &mut events), - "response.function_call_arguments.delta" => { - handle_tool_call_delta(state, &json, &mut events, "function"); - } - "response.custom_tool_call_input.delta" => { - handle_tool_call_delta(state, &json, &mut events, "custom"); - } - "response.output_item.done" => handle_output_item_done(state, &json, &mut events), - "response.completed" | "response.incomplete" => { - handle_response_completed(state, &json, &mut events); - } - "response.failed" => { - let error = json - .get("response") - .and_then(|response| response.get("error")) - .unwrap_or(&json); - return Err(provider_error_from_openai_error_json( - error, - &state.provider, - )); - } - "response.reasoning_summary_text.delta" | "response.reasoning_text.delta" => { - if let Some(delta) = json.get("delta").and_then(serde_json::Value::as_str) { - if !state.emitted_reasoning_start { - state.emitted_reasoning_start = true; - events.push(StreamEvent::ReasoningStart); - } - events.push(StreamEvent::ReasoningDelta { - delta: delta.to_string(), - }); - } - } - // response.reasoning_summary_part.added and other unrecognized events are no-ops - _ => {} - } - - Ok(events) -} - -/// Handle `response.created` by extracting the response ID and model. -fn handle_response_created(state: &mut SseStreamState, json: &serde_json::Value) { - if let Some(id) = json - .get("response") - .and_then(|r| r.get("id")) - .and_then(serde_json::Value::as_str) - { - state.response_id = id.to_string(); - } - if let Some(model) = json - .get("response") - .and_then(|r| r.get("model")) - .and_then(serde_json::Value::as_str) - { - state.response_model = model.to_string(); - } -} - -/// Handle `response.output_text.delta` by accumulating text and emitting -/// events. -fn handle_text_delta( - state: &mut SseStreamState, - json: &serde_json::Value, - events: &mut Vec, -) { - if let Some(delta) = json.get("delta").and_then(serde_json::Value::as_str) { - if !state.emitted_text_start { - state.emitted_text_start = true; - events.push(StreamEvent::TextStart { text_id: None }); - } - state.accumulated_text.push_str(delta); - events.push(StreamEvent::text_delta(delta, None)); - } -} - -/// Handle `response.function_call_arguments.delta` by accumulating args and -/// emitting events. -fn handle_tool_call_delta( - state: &mut SseStreamState, - json: &serde_json::Value, - events: &mut Vec, - tool_type: &str, -) { - let Some(delta) = json.get("delta").and_then(serde_json::Value::as_str) else { - return; - }; - - let call_id = json - .get("call_id") - .and_then(serde_json::Value::as_str) - .unwrap_or("") - .to_string(); - let item_id = json - .get("item_id") - .and_then(serde_json::Value::as_str) - .unwrap_or("") - .to_string(); - let lookup_id = if call_id.is_empty() { - &item_id - } else { - &call_id - }; - - let tc_index = state.tool_calls.iter().position(|tc| tc.id == *lookup_id); - - if let Some(idx) = tc_index { - if let Some(ref mut raw) = state.tool_calls[idx].raw_arguments { - raw.push_str(delta); - if tool_type == "custom" { - state.tool_calls[idx].arguments = serde_json::json!(raw.clone()); - } - } - } else { - let name = json - .get("name") - .and_then(serde_json::Value::as_str) - .unwrap_or("") - .to_string(); - let mut tc = ToolCall::new( - lookup_id, - name, - if tool_type == "custom" { - serde_json::json!(delta) - } else { - serde_json::json!({}) - }, - ); - tc.tool_type = tool_type.to_string(); - tc.raw_arguments = Some(delta.to_string()); - // Preserve item-level ID (fc_xxx) for Responses API round-trip - if !item_id.is_empty() && item_id != *lookup_id { - tc.provider_metadata = Some(serde_json::json!({"id": item_id})); - } - state.tool_calls.push(tc.clone()); - events.push(StreamEvent::ToolCallStart { tool_call: tc }); - } - - let current_tc = state - .tool_calls - .iter() - .find(|tc| tc.id == *lookup_id) - .cloned() - .unwrap_or_else(|| ToolCall::new("", "", serde_json::json!({}))); - - events.push(StreamEvent::ToolCallDelta { - tool_call: ToolCall { - tool_type: tool_type.to_string(), - raw_arguments: Some(delta.to_string()), - ..current_tc - }, - }); -} - -/// Handle `response.output_item.done` for text and function call items. -fn handle_output_item_done( - state: &mut SseStreamState, - json: &serde_json::Value, - events: &mut Vec, -) { - let item_type = json - .get("item") - .and_then(|i| i.get("type")) - .and_then(serde_json::Value::as_str); - - match item_type { - Some("reasoning") => { - if state.emitted_reasoning_start { - state.emitted_reasoning_start = false; - events.push(StreamEvent::ReasoningEnd); - } - let item = json.get("item").unwrap_or(json); - state.reasoning_items.push(item.clone()); - } - Some("message") => { - if state.emitted_text_start { - events.push(StreamEvent::TextEnd { text_id: None }); - state.emitted_text_start = false; - } - let item = json.get("item").unwrap_or(json); - state.message_items.push(item.clone()); - } - Some("function_call") => { - let item = json.get("item").unwrap_or(json); - let item_id = item - .get("id") - .and_then(serde_json::Value::as_str) - .unwrap_or(""); - let call_id = item - .get("call_id") - .and_then(serde_json::Value::as_str) - .unwrap_or(item_id) - .to_string(); - let name = item - .get("name") - .and_then(serde_json::Value::as_str) - .unwrap_or("") - .to_string(); - let args_str = item - .get("arguments") - .and_then(serde_json::Value::as_str) - .unwrap_or("{}"); - let arguments = - serde_json::from_str(args_str).unwrap_or_else(|_| serde_json::json!({})); - - let mut tc = ToolCall::new(&call_id, &name, arguments); - tc.raw_arguments = Some(args_str.to_string()); - // Preserve item-level ID (fc_xxx) for Responses API round-trip - if !item_id.is_empty() { - tc.provider_metadata = Some(serde_json::json!({"id": item_id})); - } - - if let Some(existing) = state.tool_calls.iter_mut().find(|t| t.id == call_id) { - existing.name.clone_from(&name); - existing.arguments = tc.arguments.clone(); - existing.raw_arguments.clone_from(&tc.raw_arguments); - existing.provider_metadata.clone_from(&tc.provider_metadata); - } else { - state.tool_calls.push(tc.clone()); - } - - events.push(StreamEvent::ToolCallEnd { tool_call: tc }); - } - Some("custom_tool_call") => { - let item = json.get("item").unwrap_or(json); - let item_id = item - .get("id") - .and_then(serde_json::Value::as_str) - .unwrap_or(""); - let call_id = item - .get("call_id") - .and_then(serde_json::Value::as_str) - .unwrap_or(item_id) - .to_string(); - let name = item - .get("name") - .and_then(serde_json::Value::as_str) - .unwrap_or("") - .to_string(); - let raw_input = item - .get("input") - .and_then(serde_json::Value::as_str) - .unwrap_or(""); - - let mut tc = ToolCall::new(&call_id, &name, serde_json::json!(raw_input)); - tc.tool_type = "custom".to_string(); - tc.raw_arguments = Some(raw_input.to_string()); - if !item_id.is_empty() { - tc.provider_metadata = Some(serde_json::json!({"id": item_id})); - } - - if let Some(existing) = state.tool_calls.iter_mut().find(|t| t.id == call_id) { - existing.name.clone_from(&name); - existing.tool_type = "custom".to_string(); - existing.arguments = tc.arguments.clone(); - existing.raw_arguments.clone_from(&tc.raw_arguments); - existing.provider_metadata.clone_from(&tc.provider_metadata); - } else { - state.tool_calls.push(tc.clone()); - } - - events.push(StreamEvent::ToolCallEnd { tool_call: tc }); - } - _ => {} - } -} - -/// Handle `response.completed` by extracting usage and building the final -/// response. -fn handle_response_completed( - state: &mut SseStreamState, - json: &serde_json::Value, - events: &mut Vec, -) { - let response_data = json.get("response").unwrap_or(json); - - if let Some(usage_data) = response_data.get("usage") { - if let Ok(u) = serde_json::from_value::(usage_data.clone()) { - state.usage = token_counts_from_api_usage(Some(&u)); - } - } - - if let Some(id) = response_data.get("id").and_then(serde_json::Value::as_str) { - state.response_id = id.to_string(); - } - if let Some(model) = response_data - .get("model") - .and_then(serde_json::Value::as_str) - { - state.response_model = model.to_string(); - } - - let status = response_data - .get("status") - .and_then(serde_json::Value::as_str); - let has_tool_calls = !state.tool_calls.is_empty(); - state.finish_reason = map_finish_reason(status, has_tool_calls); - - state.raw_response = Some(response_data.clone()); - - let mut content_parts = Vec::new(); - // Reasoning items must precede function calls for Responses API round-trip - for item in std::mem::take(&mut state.reasoning_items) { - content_parts.push(ContentPart::Other { - kind: ContentPart::OPENAI_REASONING.to_string(), - data: item, - }); - } - // Preserve full message output items for Responses API round-tripping - for item in std::mem::take(&mut state.message_items) { - content_parts.push(ContentPart::Other { - kind: ContentPart::OPENAI_MESSAGE.to_string(), - data: item, - }); - } - if !state.accumulated_text.is_empty() { - content_parts.push(ContentPart::text(&state.accumulated_text)); - } - for tc in &state.tool_calls { - // Skip tool calls with empty names (e.g. model-internal items) - if tc.name.is_empty() { - continue; - } - content_parts.push(ContentPart::ToolCall(tc.clone())); - } - - let model = if state.response_model.is_empty() { - state.model.clone() - } else { - state.response_model.clone() - }; - - let response = Response { - id: state.response_id.clone(), - model, - provider: state.provider.clone(), - message: Message { - role: Role::Assistant, - content: content_parts, - name: None, - tool_call_id: None, - }, - finish_reason: state.finish_reason.clone(), - usage: state.usage.clone(), - raw: state.raw_response.clone(), - warnings: vec![], - rate_limit: state.rate_limit.clone(), - }; - - events.push(StreamEvent::finish( - state.finish_reason.clone(), - state.usage.clone(), - response, - )); -} - #[async_trait::async_trait] impl ProviderAdapter for Adapter { fn name(&self) -> &str { @@ -1259,17 +257,18 @@ impl ProviderAdapter for Adapter { request: &Request, ) -> Result, Error> { self.validate_request(request)?; - let request_body = build_request_body_with_catalog( - request, - false, - self.codex_mode, - self.catalog.as_deref(), - ) - .await; - let request_body = filter_input_tokens_request_body(&request_body); - let url = format!("{}/responses/input_tokens", self.http.base_url); - let mut req = self.build_request(&url).json(&request_body); + let resolved = self.resolve_request(request).await; + let codec = OpenAiResponses; + let deployment_id = common::api_model_id(self.catalog.as_deref(), &resolved.model); + let params = self.codec_params(); + let ctx = self.codec_ctx(&resolved, &deployment_id, ¶ms); + + let Some(encoded) = codec.encode_count_tokens(&ctx).transpose()? else { + return Ok(None); + }; + + let mut req = self.build_http_request(&encoded); if let Some(t) = self.http.request_timeout { req = req.timeout(t); } @@ -1280,89 +279,54 @@ impl ProviderAdapter for Adapter { "input_token_count", ) .await?; - let response: InputTokensResponse = - serde_json::from_str(&body).map_err(|e| Error::Configuration { - message: format!("failed to parse OpenAI input token response: {e}"), - source: None, - })?; - - if response.object != "response.input_tokens" { - return Err(Error::Configuration { - message: format!( - "failed to parse OpenAI input token response: unexpected object '{}'", - response.object - ), - source: None, - }); - } + let input_tokens = codec.decode_count_tokens(&body)?; Ok(Some(InputTokenCount { - input_tokens: response.input_tokens, - method: InputTokenCountMethod::ProviderApi, - provider: self.provider_name.clone(), - model: request.model.clone(), - warnings: vec![], + input_tokens, + method: InputTokenCountMethod::ProviderApi, + provider: self.provider_name.clone(), + model: request.model.clone(), + warnings: vec![], })) } async fn complete(&self, request: &Request) -> Result { self.validate_request(request)?; - // Codex endpoint requires streaming; collect the stream into a response. + // Codex endpoint requires streaming; collect the stream into a + // response. if self.codex_mode { return self.complete_via_stream(request).await; } - let request_body = - build_request_body_with_catalog(request, false, false, self.catalog.as_deref()).await; - let url = format!("{}/responses", self.http.base_url); + let resolved = self.resolve_request(request).await; + let codec = OpenAiResponses; + let deployment_id = common::api_model_id(self.catalog.as_deref(), &resolved.model); + let params = self.codec_params(); + let ctx = self.codec_ctx(&resolved, &deployment_id, ¶ms); - let mut req = self.build_request(&url).json(&request_body); + let encoded = codec.encode(&ctx, false)?; + let mut req = self.build_http_request(&encoded); 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 OpenAI response: {e}"), e))?; - - let (content_parts, has_tool_calls) = parse_output(&api_resp.output); - let finish_reason = map_finish_reason(api_resp.status.as_deref(), has_tool_calls); - - let usage = token_counts_from_api_usage(api_resp.usage.as_ref()); - - Ok(Response { - id: api_resp.id, - model: api_resp.model.unwrap_or_else(|| request.model.clone()), - 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 request_body = build_request_body_with_catalog( - request, - true, - self.codex_mode, - self.catalog.as_deref(), - ) - .await; - let url = format!("{}/responses", self.http.base_url); + let resolved = self.resolve_request(request).await; + let codec = OpenAiResponses; + let deployment_id = common::api_model_id(self.catalog.as_deref(), &resolved.model); + let params = self.codec_params(); + let ctx = self.codec_ctx(&resolved, &deployment_id, ¶ms); + + let encoded = codec.encode(&ctx, true)?; let http_resp = self - .build_request(&url) - .json(&request_body) + .build_http_request(&encoded) .send() .await .map_err(|e| Error::network(e.to_string(), e))?; @@ -1374,52 +338,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 model = request.model.clone(); let rate_limit = parse_rate_limit_headers(http_resp.headers()); let stream_read_timeout = self.http.stream_read_timeout; + let decoder = codec.stream_decoder(&ctx, rate_limit); - let state = SseStreamState { - line_reader: super::common::LineReader::new(http_resp, stream_read_timeout), - model, - provider: self.provider_name.clone(), - response_id: String::new(), - response_model: String::new(), - accumulated_text: String::new(), - tool_calls: Vec::new(), - reasoning_items: Vec::new(), - message_items: Vec::new(), - usage: TokenCounts::default(), - finish_reason: FinishReason::Stop, - emitted_start: false, - emitted_text_start: false, - emitted_reasoning_start: false, - raw_response: None, - rate_limit, - }; + let out = stream::unfold( + StreamLoop { + decoder, + line_reader: super::common::LineReader::new(http_resp, stream_read_timeout), + pending: std::collections::VecDeque::new(), + done: false, + finished_emitted: false, + }, + |mut state| async move { + loop { + if let Some(event) = state.pending.pop_front() { + return Some((Ok(event), state)); + } - let stream = stream::unfold(state, |mut state| async move { - let events = process_next_sse_events(&mut state).await; - let items: Vec> = match events { - Ok(events) if events.is_empty() => return None, - Ok(events) => events.into_iter().map(Ok).collect(), - Err(e) => vec![Err(e)], - }; - Some((stream::iter(items), state)) - }) - .flatten(); + if state.done { + if state.finished_emitted { + return None; + } + state.finished_emitted = true; + let events = state.decoder.finish(); + if events.is_empty() { + return None; + } + state.pending.extend(events); + continue; + } - Ok(Box::pin(stream)) + match state.line_reader.read_next_chunk("\n\n").await { + Ok(Some(block)) => { + let Some((event_type, data)) = parse_sse_message(&block) else { + continue; + }; + match state.decoder.on_event(RawEvent { + event: event_type.as_deref(), + data: &data, + }) { + Ok(events) => state.pending.extend(events), + Err(e) => return Some((Err(e), state)), + } + } + Ok(None) => state.done = true, + Err(e) => return Some((Err(e), state)), + } + } + }, + ); + + Ok(Box::pin(out)) } } @@ -1436,8 +409,7 @@ mod tests { use super::*; use crate::error::ProviderErrorKind; - use crate::providers::common::LineReader; - use crate::types::{AudioData, DocumentData, ReasoningEffort, ToolResult}; + use crate::types::Message; fn minimal_request() -> Request { Request { @@ -1519,137 +491,39 @@ mod tests { } } - #[tokio::test] - async fn build_request_body_includes_metadata() { - let mut metadata = HashMap::new(); - metadata.insert("user_id".to_string(), "u123".to_string()); - metadata.insert("session".to_string(), "s456".to_string()); - - let mut request = minimal_request(); - request.metadata = Some(metadata); - - let body = build_request_body(&request, false, false).await; - let meta = body.get("metadata").expect("metadata should be present"); - assert_eq!(meta["user_id"], "u123"); - assert_eq!(meta["session"], "s456"); + #[test] + fn adapter_with_org_id_sets_field() { + let adapter = Adapter::new("sk-test").with_org_id("org-123"); + assert_eq!(adapter.org_id.as_deref(), Some("org-123")); } - #[tokio::test] - async fn build_request_body_omits_metadata_when_none() { - let request = minimal_request(); - let body = build_request_body(&request, false, false).await; - assert!(body.get("metadata").is_none()); + #[test] + fn adapter_with_project_id_sets_field() { + let adapter = Adapter::new("sk-test").with_project_id("proj-456"); + assert_eq!(adapter.project_id.as_deref(), Some("proj-456")); } - #[tokio::test] - async fn build_request_body_merges_provider_options_openai() { - let mut request = minimal_request(); - request.provider_options = Some(serde_json::json!({ - "openai": { - "store": true, - "previous_response_id": "resp_abc123" - } - })); - - let body = build_request_body(&request, false, false).await; - assert_eq!(body["store"], true); - assert_eq!(body["previous_response_id"], "resp_abc123"); - } - - #[tokio::test] - async fn build_request_body_provider_options_override_fields() { - let mut request = minimal_request(); - request.temperature = Some(0.5); - request.provider_options = Some(serde_json::json!({ - "openai": { - "temperature": 0.9 - } - })); - - let body = build_request_body(&request, false, false).await; - // provider_options should override the base field - assert_eq!(body["temperature"], 0.9); - } - - #[tokio::test] - async fn build_request_body_ignores_non_openai_provider_options() { - let mut request = minimal_request(); - request.provider_options = Some(serde_json::json!({ - "anthropic": { - "thinking": {"type": "enabled", "budget_tokens": 10000} - } - })); - - let body = build_request_body(&request, false, false).await; - // anthropic options should not leak into the OpenAI request - assert!(body.get("thinking").is_none()); - } - - #[tokio::test] - async fn build_request_body_no_provider_options() { - let request = minimal_request(); - let body = build_request_body(&request, false, false).await; - assert_eq!(body["model"], "gpt-4o"); - // stream field is omitted when false (skip_serializing_if) - assert!(body.get("stream").is_none()); - } - - #[tokio::test] - async fn filter_input_tokens_request_body_keeps_only_count_fields() { - let mut metadata = HashMap::new(); - metadata.insert("trace".to_string(), "abc".to_string()); - - let mut request = minimal_request(); - request.tools = Some(vec![ToolDefinition::function( - "search", - "Search files", - serde_json::json!({"type": "object"}), - )]); - request.reasoning_effort = Some(ReasoningEffort::Low); - request.response_format = Some(ResponseFormat { - kind: ResponseFormatType::JsonSchema, - json_schema: Some(serde_json::json!({"type": "object"})), - strict: true, - }); - request.temperature = Some(0.2); - request.top_p = Some(0.9); - request.max_tokens = Some(32); - request.stop_sequences = Some(vec!["END".to_string()]); - request.metadata = Some(metadata); - - let body = build_request_body(&request, true, false).await; - let filtered = filter_input_tokens_request_body(&body); - + #[test] + fn adapter_with_default_headers_sets_field() { + let mut headers = HashMap::new(); + headers.insert("X-Custom".to_string(), "value".to_string()); + let adapter = Adapter::new("sk-test").with_default_headers(headers); assert_eq!( - filtered, - serde_json::json!({ - "input": [{"type": "message", "content": [{"text": "Hello", "type": "input_text"}], "role": "user"}], - "model": "gpt-4o", - "reasoning": {"effort": "low"}, - "text": {"format": {"name": "response", "schema": {"type": "object"}, "strict": true, "type": "json_schema"}}, - "tools": [{"description": "Search files", "name": "search", "parameters": {"type": "object"}, "type": "function"}] - }) + adapter + .http + .default_headers + .get("X-Custom") + .map(String::as_str), + Some("value") ); - assert!(filtered.get("store").is_none()); - assert!(filtered.get("include").is_none()); - assert!(filtered.get("stream").is_none()); - assert!(filtered.get("max_output_tokens").is_none()); - assert!(filtered.get("metadata").is_none()); - assert!(filtered.get("temperature").is_none()); - assert!(filtered.get("top_p").is_none()); - assert!(filtered.get("stop").is_none()); } - #[tokio::test] - async fn filter_input_tokens_request_body_preserves_codex_serialization() { - let body = build_request_body(&minimal_request(), false, true).await; - let filtered = filter_input_tokens_request_body(&body); - - assert_eq!(filtered["instructions"], ""); - assert!(filtered.get("input").is_some()); - assert!(filtered.get("model").is_some()); - assert!(filtered.get("max_output_tokens").is_none()); - assert!(filtered.get("include").is_none()); + #[test] + fn adapter_defaults_have_no_org_project_or_headers() { + let adapter = Adapter::new("sk-test"); + assert!(adapter.org_id.is_none()); + assert!(adapter.project_id.is_none()); + assert!(adapter.http.default_headers.is_empty()); } #[tokio::test] @@ -1749,950 +623,6 @@ mod tests { assert!(matches!(err, Error::Configuration { .. })); } - #[tokio::test] - async fn build_request_body_includes_encrypted_reasoning_for_stateless_requests() { - let request = minimal_request(); - - let body = build_request_body(&request, false, false).await; - - assert_eq!( - body["include"], - serde_json::json!(["reasoning.encrypted_content"]) - ); - } - - #[tokio::test] - async fn build_request_body_emits_custom_apply_patch_tool() { - let mut request = minimal_request(); - request.tools = Some(vec![ - ToolDefinition::custom( - "apply_patch", - "Use the `apply_patch` tool to edit files. This is a FREEFORM tool, so do not wrap the patch in JSON.", - serde_json::json!({ - "type": "grammar", - "syntax": "lark", - "definition": "start: begin_patch hunk+ end_patch", - }), - ), - ToolDefinition::function( - "read_file", - "Read file", - serde_json::json!({ - "type": "object", - "properties": {"file_path": {"type": "string"}}, - "required": ["file_path"], - }), - ), - ]); - - let body = build_request_body(&request, false, false).await; - let tools = body["tools"].as_array().expect("tools should be present"); - let apply_patch = tools - .iter() - .find(|tool| tool["name"] == "apply_patch") - .expect("apply_patch tool should be present"); - let read_file = tools - .iter() - .find(|tool| tool["name"] == "read_file") - .expect("read_file tool should be present"); - - assert_eq!(apply_patch["type"], "custom"); - assert_eq!(apply_patch["format"]["type"], "grammar"); - assert_eq!(apply_patch["format"]["syntax"], "lark"); - assert!(apply_patch.get("parameters").is_none()); - assert_eq!(read_file["type"], "function"); - assert_eq!(read_file["parameters"]["type"], "object"); - } - - #[tokio::test] - async fn build_request_body_stream_flag() { - let request = minimal_request(); - let body = build_request_body(&request, true, false).await; - assert!(body["stream"].as_bool().unwrap_or(false)); - } - - #[tokio::test] - async fn build_request_body_metadata_and_provider_options_together() { - let mut metadata = HashMap::new(); - metadata.insert("trace_id".to_string(), "t789".to_string()); - - let mut request = minimal_request(); - request.metadata = Some(metadata); - request.provider_options = Some(serde_json::json!({ - "openai": { - "store": true - } - })); - - let body = build_request_body(&request, false, false).await; - assert_eq!(body["metadata"]["trace_id"], "t789"); - assert_eq!(body["store"], true); - } - - #[test] - fn adapter_with_org_id_sets_field() { - let adapter = Adapter::new("sk-test").with_org_id("org-123"); - assert_eq!(adapter.org_id.as_deref(), Some("org-123")); - } - - #[test] - fn adapter_with_project_id_sets_field() { - let adapter = Adapter::new("sk-test").with_project_id("proj-456"); - assert_eq!(adapter.project_id.as_deref(), Some("proj-456")); - } - - #[test] - fn adapter_with_default_headers_sets_field() { - let mut headers = HashMap::new(); - headers.insert("X-Custom".to_string(), "value".to_string()); - let adapter = Adapter::new("sk-test").with_default_headers(headers); - assert_eq!( - adapter - .http - .default_headers - .get("X-Custom") - .map(String::as_str), - Some("value") - ); - } - - #[test] - fn adapter_defaults_have_no_org_project_or_headers() { - let adapter = Adapter::new("sk-test"); - assert!(adapter.org_id.is_none()); - assert!(adapter.project_id.is_none()); - assert!(adapter.http.default_headers.is_empty()); - } - #[tokio::test] - async 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 (_, input) = translate_input(&[msg]).await; - let content = input[0]["content"] - .as_array() - .expect("content should be array"); - assert_eq!(content[0]["type"], "input_text"); - assert_eq!( - content[0]["text"], - "[Audio content not supported by this provider]" - ); - } - - #[tokio::test] - async 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 (_, input) = translate_input(&[msg]).await; - let content = input[0]["content"] - .as_array() - .expect("content should be array"); - assert_eq!(content[0]["type"], "input_text"); - assert_eq!( - content[0]["text"], - "[Document 'report.pdf': content type not supported by this provider]" - ); - } - - #[tokio::test] - async 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 (_, input) = translate_input(&[msg]).await; - let content = input[0]["content"] - .as_array() - .expect("content should be array"); - assert_eq!(content[0]["type"], "input_text"); - assert_eq!( - content[0]["text"], - "[Document content not supported by this provider]" - ); - } - - #[test] - fn parse_output_preserves_both_ids_on_function_call() { - let output = vec![serde_json::json!({ - "type": "function_call", - "id": "fc_abc123", - "call_id": "call_xyz789", - "name": "get_weather", - "arguments": "{\"location\":\"NYC\"}" - })]; - let (parts, has_tool_calls) = parse_output(&output); - assert!(has_tool_calls); - assert_eq!(parts.len(), 1); - match &parts[0] { - ContentPart::ToolCall(tc) => { - // call_id is used as the ToolCall.id (links to tool results) - assert_eq!(tc.id, "call_xyz789"); - // item-level id (fc_xxx) is preserved in provider_metadata - let meta = tc - .provider_metadata - .as_ref() - .expect("provider_metadata should be set"); - assert_eq!(meta["id"], "fc_abc123"); - } - other => panic!("expected ToolCall, got {other:?}"), - } - } - - #[test] - fn parse_output_preserves_custom_tool_call_raw_input() { - let patch = "*** Begin Patch\n*** Add File: hello.txt\n+hello\n*** End Patch\n"; - let output = vec![serde_json::json!({ - "type": "custom_tool_call", - "id": "ctc_abc123", - "call_id": "call_xyz789", - "name": "apply_patch", - "input": patch, - })]; - - let (parts, has_tool_calls) = parse_output(&output); - - assert!(has_tool_calls); - assert_eq!(parts.len(), 1); - match &parts[0] { - ContentPart::ToolCall(tc) => { - assert_eq!(tc.id, "call_xyz789"); - assert_eq!(tc.name, "apply_patch"); - assert_eq!(tc.tool_type, "custom"); - assert_eq!(tc.arguments, serde_json::json!(patch)); - assert_eq!(tc.raw_arguments.as_deref(), Some(patch)); - let meta = tc - .provider_metadata - .as_ref() - .expect("provider metadata should preserve item id"); - assert_eq!(meta["id"], "ctc_abc123"); - } - other => panic!("expected ToolCall, got {other:?}"), - } - } - - #[tokio::test] - async fn translate_input_uses_item_id_for_id_field() { - let mut tc = ToolCall::new( - "call_xyz789", - "get_weather", - serde_json::json!({"location": "NYC"}), - ); - tc.provider_metadata = Some(serde_json::json!({"id": "fc_abc123"})); - - let msg = Message { - role: Role::Assistant, - content: vec![ContentPart::ToolCall(tc)], - name: None, - tool_call_id: None, - }; - let (_, input) = translate_input(&[msg]).await; - let fc = &input[0]; - assert_eq!(fc["type"], "function_call"); - // id field uses the fc_ prefixed item ID - assert_eq!(fc["id"], "fc_abc123"); - // call_id field uses the call_ prefixed call ID - assert_eq!(fc["call_id"], "call_xyz789"); - } - - #[tokio::test] - async fn translate_input_falls_back_to_tc_id_without_metadata() { - let tc = ToolCall::new("call_xyz789", "get_weather", serde_json::json!({})); - - let msg = Message { - role: Role::Assistant, - content: vec![ContentPart::ToolCall(tc)], - name: None, - tool_call_id: None, - }; - let (_, input) = translate_input(&[msg]).await; - let fc = &input[0]; - // Without provider_metadata, both fields use tc.id - assert_eq!(fc["id"], "call_xyz789"); - assert_eq!(fc["call_id"], "call_xyz789"); - } - - #[test] - fn parse_output_preserves_reasoning_items() { - let output = vec![ - serde_json::json!({ - "type": "reasoning", - "id": "rs_abc123", - "summary": [{"type": "summary_text", "text": "Thinking..."}] - }), - serde_json::json!({ - "type": "function_call", - "id": "fc_def456", - "call_id": "call_789", - "name": "search", - "arguments": "{}" - }), - ]; - let (parts, has_tool_calls) = parse_output(&output); - assert!(has_tool_calls); - assert_eq!(parts.len(), 2); - // First part is the reasoning item - match &parts[0] { - ContentPart::Other { kind, data } => { - assert_eq!(kind, ContentPart::OPENAI_REASONING); - assert_eq!(data["type"], "reasoning"); - assert_eq!(data["id"], "rs_abc123"); - } - other => panic!("expected Other, got {other:?}"), - } - // Second part is the function call - assert!(matches!(&parts[1], ContentPart::ToolCall(_))); - } - - #[test] - fn parse_output_preserves_message_items() { - let output = vec![ - serde_json::json!({ - "type": "reasoning", - "id": "rs_abc", - "summary": [] - }), - serde_json::json!({ - "type": "message", - "id": "msg_xyz", - "status": "completed", - "role": "assistant", - "content": [{"type": "output_text", "text": "Hello"}] - }), - serde_json::json!({ - "type": "function_call", - "id": "fc_123", - "call_id": "call_456", - "name": "search", - "arguments": "{}" - }), - ]; - let (parts, has_tool_calls) = parse_output(&output); - assert!(has_tool_calls); - // reasoning + openai_message + text + function_call - assert_eq!(parts.len(), 4); - assert!( - matches!(&parts[0], ContentPart::Other { kind, .. } if kind == ContentPart::OPENAI_REASONING) - ); - assert!( - matches!(&parts[1], ContentPart::Other { kind, data } if kind == ContentPart::OPENAI_MESSAGE && data["id"] == "msg_xyz") - ); - assert!(matches!(&parts[2], ContentPart::Text(t) if t == "Hello")); - assert!(matches!(&parts[3], ContentPart::ToolCall(_))); - } - - #[tokio::test] - async fn reasoning_items_round_trip_through_translate_input() { - let reasoning = serde_json::json!({ - "type": "reasoning", - "id": "rs_abc123", - "summary": [{"type": "summary_text", "text": "Thinking..."}] - }); - let mut tc = ToolCall::new("call_789", "search", serde_json::json!({})); - tc.provider_metadata = Some(serde_json::json!({"id": "fc_def456"})); - - let msg = Message { - role: Role::Assistant, - content: vec![ - ContentPart::Other { - kind: ContentPart::OPENAI_REASONING.to_string(), - data: reasoning, - }, - ContentPart::ToolCall(tc), - ], - name: None, - tool_call_id: None, - }; - let (_, input) = translate_input(&[msg]).await; - assert_eq!(input.len(), 2); - // Reasoning item is emitted first - assert_eq!(input[0]["type"], "reasoning"); - assert_eq!(input[0]["id"], "rs_abc123"); - // Function call follows - assert_eq!(input[1]["type"], "function_call"); - assert_eq!(input[1]["id"], "fc_def456"); - assert_eq!(input[1]["call_id"], "call_789"); - } - - #[tokio::test] - async fn reasoning_message_function_call_round_trip() { - // Simulates an assistant turn with reasoning + text + tool call. - // The opaque message item (with id/status) must be used instead of - // constructing a new one from Text, so the reasoning item can find - // its "required following item." - let reasoning = serde_json::json!({ - "type": "reasoning", - "id": "rs_xyz789", - "summary": [{"type": "summary_text", "text": "Let me check..."}] - }); - let opaque_message = serde_json::json!({ - "type": "message", - "id": "msg_abc123", - "status": "completed", - "role": "assistant", - "content": [{"type": "output_text", "text": "Checking now."}] - }); - let mut tc = ToolCall::new("call_001", "shell", serde_json::json!({"cmd": "ls"})); - tc.provider_metadata = Some(serde_json::json!({"id": "fc_def456"})); - - let msg = Message { - role: Role::Assistant, - content: vec![ - ContentPart::Other { - kind: ContentPart::OPENAI_REASONING.to_string(), - data: reasoning, - }, - ContentPart::Other { - kind: ContentPart::OPENAI_MESSAGE.to_string(), - data: opaque_message, - }, - ContentPart::text("Checking now."), - ContentPart::ToolCall(tc), - ], - name: None, - tool_call_id: None, - }; - let (_, input) = translate_input(&[msg]).await; - assert_eq!(input.len(), 3); - // Reasoning first - assert_eq!(input[0]["type"], "reasoning"); - assert_eq!(input[0]["id"], "rs_xyz789"); - // Opaque message with id/status (not a reconstructed one) - assert_eq!(input[1]["type"], "message"); - assert_eq!(input[1]["id"], "msg_abc123"); - assert_eq!(input[1]["status"], "completed"); - // Function call last - assert_eq!(input[2]["type"], "function_call"); - assert_eq!(input[2]["id"], "fc_def456"); - } - - #[tokio::test] - async fn text_without_opaque_message_still_constructs_message() { - // For non-OpenAI turns or turns without preserved message items, - // Text parts should still produce a constructed message. - let msg = Message { - role: Role::Assistant, - content: vec![ContentPart::text("Hello")], - name: None, - tool_call_id: None, - }; - let (_, input) = translate_input(&[msg]).await; - assert_eq!(input.len(), 1); - assert_eq!(input[0]["type"], "message"); - assert_eq!(input[0]["role"], "assistant"); - // No id field on constructed messages - assert!(input[0].get("id").is_none()); - } - - #[tokio::test] - async fn parse_output_round_trips_function_call_ids() { - // Simulate a response from the Responses API - let output = vec![serde_json::json!({ - "type": "function_call", - "id": "fc_item1", - "call_id": "call_001", - "name": "search", - "arguments": "{\"q\":\"test\"}" - })]; - let (parts, _) = parse_output(&output); - - // Now translate back to input format - let msg = Message { - role: Role::Assistant, - content: parts, - name: None, - tool_call_id: None, - }; - let (_, input) = translate_input(&[msg]).await; - let fc = &input[0]; - - // The round-tripped function call should have correct IDs - assert_eq!(fc["id"], "fc_item1"); - assert_eq!(fc["call_id"], "call_001"); - } - - #[tokio::test] - async fn custom_tool_call_history_round_trips_through_translate_input() { - let patch = "*** Begin Patch\n*** Delete File: stale.txt\n*** End Patch\n"; - let mut tc = ToolCall::new("call_001", "apply_patch", serde_json::json!(patch)); - tc.tool_type = "custom".to_string(); - tc.raw_arguments = Some(patch.to_string()); - tc.provider_metadata = Some(serde_json::json!({"id": "ctc_def456"})); - - let msg = Message { - role: Role::Assistant, - content: vec![ContentPart::ToolCall(tc)], - name: None, - tool_call_id: None, - }; - - let (_, input) = translate_input(&[msg]).await; - - assert_eq!(input.len(), 1); - assert_eq!(input[0]["type"], "custom_tool_call"); - assert_eq!(input[0]["id"], "ctc_def456"); - assert_eq!(input[0]["call_id"], "call_001"); - assert_eq!(input[0]["name"], "apply_patch"); - assert_eq!(input[0]["input"], patch); - } - - #[tokio::test] - async fn custom_tool_result_history_round_trips_through_translate_input() { - let msg = Message { - role: Role::Tool, - content: vec![ContentPart::ToolResult(ToolResult::success( - "call_001", - serde_json::json!("Success. Updated the following files:\nA hello.txt\n"), - ))], - name: Some("apply_patch".to_string()), - tool_call_id: Some("call_001".to_string()), - }; - - let (_, input) = translate_input(&[msg]).await; - - assert_eq!(input.len(), 1); - assert_eq!(input[0]["type"], "custom_tool_call_output"); - assert_eq!(input[0]["call_id"], "call_001"); - assert_eq!( - input[0]["output"], - "Success. Updated the following files:\nA hello.txt\n" - ); - } - - #[tokio::test] - async fn custom_tool_result_history_uses_prior_custom_call_without_tool_message_name() { - let patch = "*** Begin Patch\n*** Add File: hello.txt\n+hello\n*** End Patch\n"; - let mut tc = ToolCall::new("call_001", "apply_patch", serde_json::json!(patch)); - tc.tool_type = "custom".to_string(); - tc.raw_arguments = Some(patch.to_string()); - tc.provider_metadata = Some(serde_json::json!({"id": "ctc_def456"})); - - let assistant_msg = Message { - role: Role::Assistant, - content: vec![ContentPart::ToolCall(tc)], - name: None, - tool_call_id: None, - }; - let tool_msg = Message::tool_result( - "call_001", - serde_json::json!("Success. Updated the following files:\nA hello.txt\n"), - false, - ); - - let (_, input) = translate_input(&[assistant_msg, tool_msg]).await; - - assert_eq!(input.len(), 2); - assert_eq!(input[1]["type"], "custom_tool_call_output"); - assert_eq!(input[1]["call_id"], "call_001"); - assert_eq!( - input[1]["output"], - "Success. Updated the following files:\nA hello.txt\n" - ); - } - - #[tokio::test] - async fn build_request_body_includes_stop_sequences() { - let mut request = minimal_request(); - request.stop_sequences = Some(vec!["END".to_string(), "STOP".to_string()]); - - let body = build_request_body(&request, false, false).await; - let stop = body.get("stop").expect("stop should be present"); - let arr = stop.as_array().expect("stop should be an array"); - assert_eq!(arr.len(), 2); - assert_eq!(arr[0], "END"); - assert_eq!(arr[1], "STOP"); - } - - #[tokio::test] - async fn build_request_body_omits_stop_when_none() { - let request = minimal_request(); - let body = build_request_body(&request, false, false).await; - assert!(body.get("stop").is_none()); - } - - fn empty_sse_state() -> SseStreamState { - let http_resp = http::Response::builder().status(200).body("").unwrap(); - let response = fabro_http::Response::from(http_resp); - SseStreamState { - line_reader: LineReader::new(response, None), - model: String::new(), - provider: "openai".to_string(), - response_id: String::new(), - response_model: String::new(), - accumulated_text: String::new(), - tool_calls: Vec::new(), - reasoning_items: Vec::new(), - message_items: Vec::new(), - usage: TokenCounts::default(), - finish_reason: FinishReason::Stop, - emitted_start: true, - emitted_text_start: false, - emitted_reasoning_start: false, - raw_response: None, - rate_limit: None, - } - } - - #[test] - fn token_counts_disjoint_with_cache_and_reasoning() { - let mut state = empty_sse_state(); - let body = serde_json::json!({ - "response": { - "id": "resp_test", - "model": "gpt-5", - "output": [], - "status": "completed", - "usage": { - "input_tokens": 200, - "input_tokens_details": { "cached_tokens": 180 }, - "output_tokens": 500, - "output_tokens_details": { "reasoning_tokens": 300 }, - "total_tokens": 700 - } - } - }); - let mut events = Vec::new(); - - handle_response_completed(&mut state, &body, &mut events); - - assert_eq!(state.usage.input_tokens, 20); - assert_eq!(state.usage.cache_read_tokens, 180); - assert_eq!(state.usage.output_tokens, 200); - assert_eq!(state.usage.reasoning_tokens, 300); - assert_eq!(state.usage.cache_write_tokens, 0); - assert_eq!(state.usage.total_tokens(), 700); - } - - #[test] - fn custom_tool_call_streaming_delta_accumulates_raw_input() { - let mut state = empty_sse_state(); - let first = r#"{ - "type": "response.custom_tool_call_input.delta", - "item_id": "ctc_abc", - "call_id": "call_001", - "delta": "*** Begin" - }"#; - let second = r#"{ - "type": "response.custom_tool_call_input.delta", - "item_id": "ctc_abc", - "call_id": "call_001", - "delta": " Patch\n" - }"#; - - let first_events = process_sse_event( - &mut state, - Some("response.custom_tool_call_input.delta"), - first, - ) - .expect("first custom delta should parse"); - let second_events = process_sse_event( - &mut state, - Some("response.custom_tool_call_input.delta"), - second, - ) - .expect("second custom delta should parse"); - - assert!(matches!( - first_events.iter().find(|event| matches!(event, StreamEvent::ToolCallStart { .. })), - Some(StreamEvent::ToolCallStart { tool_call }) - if tool_call.id == "call_001" && tool_call.tool_type == "custom" - )); - assert!(matches!( - second_events.last(), - Some(StreamEvent::ToolCallDelta { tool_call }) - if tool_call.raw_arguments.as_deref() == Some(" Patch\n") - && tool_call.tool_type == "custom" - )); - assert_eq!( - state.tool_calls[0].raw_arguments.as_deref(), - Some("*** Begin Patch\n") - ); - } - - #[test] - fn custom_tool_call_output_item_done_emits_tool_call_end() { - let mut state = empty_sse_state(); - let patch = "*** Begin Patch\n*** Add File: hello.txt\n+hello\n*** End Patch\n"; - let data = serde_json::json!({ - "type": "response.output_item.done", - "item": { - "type": "custom_tool_call", - "id": "ctc_abc", - "call_id": "call_001", - "name": "apply_patch", - "input": patch, - } - }); - - let events = process_sse_event( - &mut state, - Some("response.output_item.done"), - &data.to_string(), - ) - .expect("custom output item should parse"); - - assert!(matches!( - events.last(), - Some(StreamEvent::ToolCallEnd { tool_call }) - if tool_call.id == "call_001" - && tool_call.name == "apply_patch" - && tool_call.tool_type == "custom" - && tool_call.raw_arguments.as_deref() == Some(patch) - )); - } - - #[test] - fn error_event_with_insufficient_quota_returns_provider_error() { - let mut state = empty_sse_state(); - let data = r#"{ - "type": "error", - "error": { - "type": "insufficient_quota", - "code": "insufficient_quota", - "message": "You exceeded your current quota.", - "param": null - } - }"#; - - let err = process_sse_event(&mut state, Some("error"), data) - .expect_err("error event should fail the stream"); - - match err { - Error::Provider { kind, detail } => { - assert_eq!(kind, ProviderErrorKind::QuotaExceeded); - assert!(detail.message.contains("exceeded your current quota")); - assert_eq!(detail.error_code.as_deref(), Some("insufficient_quota")); - assert!(detail.raw.is_some()); - } - other => panic!("expected provider error, got {other:?}"), - } - } - - #[test] - fn error_event_classifies_on_type_when_code_absent() { - let mut state = empty_sse_state(); - let data = r#"{ - "type": "error", - "error": { - "type": "insufficient_quota", - "message": "You exceeded your current quota." - } - }"#; - - let err = process_sse_event(&mut state, Some("error"), data) - .expect_err("error event should fail the stream"); - - match err { - Error::Provider { kind, detail } => { - assert_eq!(kind, ProviderErrorKind::QuotaExceeded); - assert_eq!(detail.error_code.as_deref(), Some("insufficient_quota")); - } - other => panic!("expected provider error, got {other:?}"), - } - } - - #[test] - fn response_failed_event_with_server_error_returns_provider_error() { - let mut state = empty_sse_state(); - let data = r#"{ - "type": "response.failed", - "response": { - "status": "failed", - "error": { - "type": "server_error", - "code": "server_error", - "message": "The server had an error while processing your request." - } - } - }"#; - - let err = process_sse_event(&mut state, Some("response.failed"), data) - .expect_err("response.failed should fail the stream"); - - match err { - Error::Provider { kind, detail } => { - assert_eq!(kind, ProviderErrorKind::Server); - assert!(detail.message.contains("server had an error")); - assert_eq!(detail.error_code.as_deref(), Some("server_error")); - } - other => panic!("expected provider error, got {other:?}"), - } - } - - #[test] - fn response_incomplete_preserves_partial_text() { - let mut state = empty_sse_state(); - - process_sse_event( - &mut state, - Some("response.created"), - r#"{"type":"response.created","response":{"id":"resp_123","model":"gpt-5.4"}}"#, - ) - .expect("created event should parse"); - process_sse_event( - &mut state, - Some("response.output_text.delta"), - r#"{"type":"response.output_text.delta","delta":"Hel"}"#, - ) - .expect("first delta should parse"); - process_sse_event( - &mut state, - Some("response.output_text.delta"), - r#"{"type":"response.output_text.delta","delta":"lo"}"#, - ) - .expect("second delta should parse"); - - let events = process_sse_event( - &mut state, - Some("response.incomplete"), - r#"{ - "type": "response.incomplete", - "response": { - "id": "resp_123", - "model": "gpt-5.4", - "status": "incomplete" - } - }"#, - ) - .expect("incomplete response should finish normally"); - - let finish = events - .last() - .expect("incomplete response should emit finish"); - match finish { - StreamEvent::Finish { - finish_reason, - response, - .. - } => { - assert_eq!(finish_reason.clone(), FinishReason::Length); - assert_eq!(response.text(), "Hello"); - } - other => panic!("expected finish event, got {other:?}"), - } - } - - #[test] - fn error_event_with_invalid_api_key_returns_authentication_error() { - let mut state = empty_sse_state(); - let data = r#"{ - "type": "error", - "error": { - "type": "invalid_api_key", - "code": "invalid_api_key", - "message": "Incorrect API key provided." - } - }"#; - - let err = process_sse_event(&mut state, Some("error"), data) - .expect_err("error event should fail the stream"); - - match err { - Error::Provider { kind, detail } => { - assert_eq!(kind, ProviderErrorKind::Authentication); - assert_eq!(detail.error_code.as_deref(), Some("invalid_api_key")); - } - other => panic!("expected provider error, got {other:?}"), - } - } - - #[test] - fn error_event_with_rate_limit_error_returns_rate_limit() { - let mut state = empty_sse_state(); - let data = r#"{ - "type": "error", - "error": { - "type": "rate_limit_error", - "message": "Too many requests." - } - }"#; - - let err = process_sse_event(&mut state, Some("error"), data) - .expect_err("error event should fail the stream"); - - match err { - Error::Provider { kind, detail } => { - assert_eq!(kind, ProviderErrorKind::RateLimit); - assert_eq!(detail.error_code.as_deref(), Some("rate_limit_error")); - } - other => panic!("expected provider error, got {other:?}"), - } - } - - #[test] - fn error_event_with_unknown_invalid_prefix_returns_invalid_request() { - let mut state = empty_sse_state(); - let data = r#"{ - "type": "error", - "error": { - "type": "invalid_prompt", - "code": "invalid_prompt", - "message": "Prompt is invalid." - } - }"#; - - let err = process_sse_event(&mut state, Some("error"), data) - .expect_err("error event should fail the stream"); - - match err { - Error::Provider { kind, detail } => { - assert_eq!(kind, ProviderErrorKind::InvalidRequest); - assert_eq!(detail.error_code.as_deref(), Some("invalid_prompt")); - } - other => panic!("expected provider error, got {other:?}"), - } - } - - #[test] - fn error_event_with_unknown_code_falls_back_to_server_with_message() { - let mut state = empty_sse_state(); - let data = r#"{ - "type": "error", - "error": { - "type": "unexpected_stream_failure", - "code": "unexpected_stream_failure", - "message": "Unexpected stream failure." - } - }"#; - - let err = process_sse_event(&mut state, Some("error"), data) - .expect_err("error event should fail the stream"); - - match err { - Error::Provider { kind, detail } => { - assert_eq!(kind, ProviderErrorKind::Server); - assert_eq!(detail.message, "Unexpected stream failure."); - assert_eq!( - detail.error_code.as_deref(), - Some("unexpected_stream_failure") - ); - } - other => panic!("expected provider error, got {other:?}"), - } - } - #[tokio::test] async fn codex_complete_via_stream_propagates_stream_errors() { let server = MockServer::start(); @@ -2726,59 +656,4 @@ data: {"type":"error","error":{"type":"insufficient_quota","code":"insufficient_ other => panic!("expected provider error, got {other:?}"), } } - - #[test] - fn reasoning_summary_delta_emits_reasoning_events() { - let mut state = empty_sse_state(); - let data = r#"{"type":"response.reasoning_summary_text.delta","delta":"Let me think"}"#; - let events = process_sse_event( - &mut state, - Some("response.reasoning_summary_text.delta"), - data, - ) - .expect("reasoning summary delta should parse"); - assert_eq!(events.len(), 2); - assert!(matches!(events[0], StreamEvent::ReasoningStart)); - assert!( - matches!(events[1], StreamEvent::ReasoningDelta { ref delta } if delta == "Let me think") - ); - } - - #[test] - fn reasoning_text_delta_emits_reasoning_events() { - let mut state = empty_sse_state(); - - // First delta: should emit ReasoningStart + ReasoningDelta - let data1 = r#"{"type":"response.reasoning_text.delta","delta":"Step 1"}"#; - let events1 = process_sse_event(&mut state, Some("response.reasoning_text.delta"), data1) - .expect("first reasoning delta should parse"); - assert_eq!(events1.len(), 2); - assert!(matches!(events1[0], StreamEvent::ReasoningStart)); - assert!( - matches!(events1[1], StreamEvent::ReasoningDelta { ref delta } if delta == "Step 1") - ); - - // Second delta: should NOT emit duplicate ReasoningStart - let data2 = r#"{"type":"response.reasoning_text.delta","delta":"Step 2"}"#; - let events2 = process_sse_event(&mut state, Some("response.reasoning_text.delta"), data2) - .expect("second reasoning delta should parse"); - assert_eq!(events2.len(), 1); - assert!( - matches!(events2[0], StreamEvent::ReasoningDelta { ref delta } if delta == "Step 2") - ); - } - - #[test] - fn reasoning_end_emitted_on_item_done() { - let mut state = empty_sse_state(); - state.emitted_reasoning_start = true; - - let data = r#"{"item":{"type":"reasoning","id":"rs_abc","summary":[]}}"#; - let events = process_sse_event(&mut state, Some("response.output_item.done"), data) - .expect("output item done should parse"); - assert_eq!(events.len(), 1); - assert!(matches!(events[0], StreamEvent::ReasoningEnd)); - assert!(!state.emitted_reasoning_start); - assert_eq!(state.reasoning_items.len(), 1); - } } diff --git a/lib/crates/fabro-llm/tests/it/wire/snapshots/it__wire__openai_responses__count_tokens_wire_shape.snap b/lib/crates/fabro-llm/tests/it/wire/snapshots/it__wire__openai_responses__count_tokens_wire_shape.snap index 4d98746a8..8524cc347 100644 --- a/lib/crates/fabro-llm/tests/it/wire/snapshots/it__wire__openai_responses__count_tokens_wire_shape.snap +++ b/lib/crates/fabro-llm/tests/it/wire/snapshots/it__wire__openai_responses__count_tokens_wire_shape.snap @@ -28,6 +28,7 @@ expression: rendered ] ], "body": { + "model": "gpt-test", "input": [ { "type": "message", @@ -41,7 +42,6 @@ expression: rendered } ], "instructions": "Be concise", - "model": "gpt-test", "tools": [ { "type": "function",