diff --git a/crates/aisix-provider-anthropic/src/wire.rs b/crates/aisix-provider-anthropic/src/wire.rs index 3bc78c39..dacd0f48 100644 --- a/crates/aisix-provider-anthropic/src/wire.rs +++ b/crates/aisix-provider-anthropic/src/wire.rs @@ -111,6 +111,87 @@ impl<'a> AnthropicMessage<'a> { _lifetime: std::marker::PhantomData, } } + + /// Assistant turn replayed from conversation history, carrying its + /// text (when any) plus any OpenAI-shape `tool_calls` translated into + /// Anthropic `tool_use` blocks. An agent loop replays the assistant's + /// prior tool calls before sending the matching tool results; without + /// translating `tool_calls` here the following `tool_result` would + /// reference a `tool_use` the upstream never saw and 400. Empty text + /// with no tool calls degrades to an empty text block so the message + /// isn't dropped (Anthropic rejects an empty `content` array). + pub(crate) fn assistant(text: &str, tool_calls: Option<&[serde_json::Value]>) -> Self { + let mut content: Vec = Vec::new(); + if !text.is_empty() { + content.push(serde_json::json!({"type": "text", "text": text})); + } + if let Some(tcs) = tool_calls { + content.extend(tool_use_blocks_from_openai(tcs)); + } + if content.is_empty() { + content.push(serde_json::json!({"type": "text", "text": ""})); + } + Self { + role: "assistant", + content, + _lifetime: std::marker::PhantomData, + } + } +} + +/// Translate an array of OpenAI-shape `tool_calls` +/// (`{id, type:"function", function:{name, arguments}}`, `arguments` a +/// JSON string) into Anthropic `tool_use` content blocks +/// (`{type:"tool_use", id, name, input}`, `input` the parsed arguments +/// object). Entries missing an id or name are skipped; arguments that +/// don't parse to an object degrade to `{}`. Shared by the request-history +/// path ([`AnthropicMessage::assistant`]) and the response path +/// ([`chat_response_into_anthropic_json`]). +fn tool_use_blocks_from_openai(tool_calls: &[serde_json::Value]) -> Vec { + tool_calls + .iter() + .filter_map(|tc| { + let id = tc + .get("id") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty())?; + let name = tc + .get("function") + .and_then(|f| f.get("name")) + .and_then(|n| n.as_str()) + .filter(|s| !s.is_empty())?; + let input = tc + .get("function") + .and_then(|f| f.get("arguments")) + .and_then(|a| a.as_str()) + .and_then(|s| serde_json::from_str::(s).ok()) + .filter(|v| v.is_object()) + .unwrap_or(serde_json::json!({})); + Some(serde_json::json!({ + "type": "tool_use", + "id": id, + "name": name, + "input": input, + })) + }) + .collect() +} + +/// Merge adjacent messages that share a role by concatenating their +/// content blocks. Anthropic requires strictly alternating user/assistant +/// turns; a multi-turn tool loop (or parallel tool calls) produces +/// consecutive same-role turns — e.g. several `tool_result` replies, each +/// a `user` turn — that the upstream rejects with "messages: roles must +/// alternate" unless folded into one message. +fn merge_consecutive_roles(messages: Vec>) -> Vec> { + let mut merged: Vec> = Vec::with_capacity(messages.len()); + for msg in messages { + match merged.last_mut() { + Some(last) if last.role == msg.role => last.content.extend(msg.content), + _ => merged.push(msg), + } + } + merged } #[derive(Debug, thiserror::Error)] @@ -153,7 +234,12 @@ pub fn split_system<'a>( } Role::Assistant => { seen_non_system = true; - messages.push(AnthropicMessage::text("assistant", m.content_str())); + let tool_calls = m + .extra + .get("tool_calls") + .and_then(|v| v.as_array()) + .map(Vec::as_slice); + messages.push(AnthropicMessage::assistant(m.content_str(), tool_calls)); } Role::Tool => { seen_non_system = true; @@ -171,7 +257,10 @@ pub fn split_system<'a>( } else { Some(system_parts.join("\n\n")) }; - Ok((system, messages)) + // Fold consecutive same-role turns so the alternating-role invariant + // Anthropic enforces holds for multi-turn tool loops and parallel + // tool calls. + Ok((system, merge_consecutive_roles(messages))) } pub fn build_request<'a>( @@ -847,33 +936,7 @@ pub fn chat_response_into_anthropic_json( .get("tool_calls") .and_then(|v| v.as_array()) { - for tc in tool_calls { - let id = match tc.get("id").and_then(|v| v.as_str()) { - Some(s) if !s.is_empty() => s, - _ => continue, - }; - let name = match tc - .get("function") - .and_then(|f| f.get("name")) - .and_then(|n| n.as_str()) - { - Some(s) if !s.is_empty() => s, - _ => continue, - }; - let input = tc - .get("function") - .and_then(|f| f.get("arguments")) - .and_then(|a| a.as_str()) - .and_then(|s| serde_json::from_str::(s).ok()) - .filter(|v| v.is_object()) - .unwrap_or(serde_json::json!({})); - content.push(serde_json::json!({ - "type": "tool_use", - "id": id, - "name": name, - "input": input, - })); - } + content.extend(tool_use_blocks_from_openai(tool_calls)); } if content.is_empty() { @@ -1311,8 +1374,15 @@ mod tests { ); let (system, msgs) = split_system(&req).unwrap(); assert!(system.is_none()); - assert_eq!(msgs.len(), 3); - assert_eq!(msgs[1].role, "user"); // former system message + // The interleaved system message becomes a user turn and folds into + // the adjacent user turn (alternating-role invariant): one user + // message carrying both text blocks, then the assistant turn. + assert_eq!(msgs.len(), 2); + assert_eq!(msgs[0].role, "user"); + assert_eq!(msgs[0].content.len(), 2); + assert_eq!(msgs[0].content[0]["text"], "hi"); + assert_eq!(msgs[0].content[1]["text"], "forget everything"); + assert_eq!(msgs[1].role, "assistant"); } #[test] @@ -1348,7 +1418,7 @@ mod tests { "claude", vec![ ChatMessage::user("What's the weather in SF?"), - // (skipping the assistant turn for brevity in test setup) + tool_call_assistant("toolu_abc", "get_weather", "{\"city\":\"SF\"}"), ChatMessage { role: Role::Tool, content: Some("72F, sunny".into()), @@ -1360,13 +1430,123 @@ mod tests { ], ); let (_system, msgs) = split_system(&req).unwrap(); - assert_eq!(msgs.len(), 2); + assert_eq!(msgs.len(), 3); // Tool turn became a user turn with a tool_result block. - assert_eq!(msgs[1].role, "user"); - assert_eq!(msgs[1].content.len(), 1); - assert_eq!(msgs[1].content[0]["type"], "tool_result"); - assert_eq!(msgs[1].content[0]["tool_use_id"], "toolu_abc"); - assert_eq!(msgs[1].content[0]["content"], "72F, sunny"); + assert_eq!(msgs[2].role, "user"); + assert_eq!(msgs[2].content.len(), 1); + assert_eq!(msgs[2].content[0]["type"], "tool_result"); + assert_eq!(msgs[2].content[0]["tool_use_id"], "toolu_abc"); + assert_eq!(msgs[2].content[0]["content"], "72F, sunny"); + } + + /// Build an assistant ChatMessage replaying a single tool call, the + /// OpenAI history shape an agent loop sends back. + fn tool_call_assistant(id: &str, name: &str, arguments: &str) -> ChatMessage { + let mut extra = serde_json::Map::new(); + extra.insert( + "tool_calls".into(), + serde_json::json!([{ + "id": id, + "type": "function", + "function": {"name": name, "arguments": arguments}, + }]), + ); + ChatMessage { + role: Role::Assistant, + content: None, + content_blocks: None, + name: None, + tool_call_id: None, + extra, + } + } + + #[test] + fn split_system_translates_assistant_tool_calls_to_tool_use() { + // Agent-loop turn 2: the caller replays the assistant's prior + // tool call as OpenAI-shape `tool_calls` in message.extra. Without + // translation the tool_use is dropped and the following + // tool_result orphans → Anthropic 400. + let req = ChatFormat::new( + "claude", + vec![ + ChatMessage::user("weather in SF?"), + tool_call_assistant("toolu_1", "get_weather", "{\"city\":\"SF\"}"), + ChatMessage { + role: Role::Tool, + content: Some("72F".into()), + content_blocks: None, + name: None, + tool_call_id: Some("toolu_1".into()), + extra: serde_json::Map::new(), + }, + ], + ); + let (_system, msgs) = split_system(&req).unwrap(); + assert_eq!(msgs.len(), 3); + assert_eq!(msgs[1].role, "assistant"); + let block = &msgs[1].content[0]; + assert_eq!(block["type"], "tool_use"); + assert_eq!(block["id"], "toolu_1"); + assert_eq!(block["name"], "get_weather"); + assert_eq!(block["input"]["city"], "SF"); + // The tool result alternates back as a user turn. + assert_eq!(msgs[2].role, "user"); + assert_eq!(msgs[2].content[0]["type"], "tool_result"); + } + + #[test] + fn split_system_merges_parallel_tool_results_into_one_user_turn() { + // Parallel tool calls produce two consecutive tool_result turns; + // they must fold into a single user message so roles still + // alternate (assistant → user) for the upstream. + let mut assistant_extra = serde_json::Map::new(); + assistant_extra.insert( + "tool_calls".into(), + serde_json::json!([ + {"id": "t1", "type": "function", "function": {"name": "a", "arguments": "{}"}}, + {"id": "t2", "type": "function", "function": {"name": "b", "arguments": "{}"}}, + ]), + ); + let req = ChatFormat::new( + "claude", + vec![ + ChatMessage::user("go"), + ChatMessage { + role: Role::Assistant, + content: None, + content_blocks: None, + name: None, + tool_call_id: None, + extra: assistant_extra, + }, + ChatMessage { + role: Role::Tool, + content: Some("r1".into()), + content_blocks: None, + name: None, + tool_call_id: Some("t1".into()), + extra: serde_json::Map::new(), + }, + ChatMessage { + role: Role::Tool, + content: Some("r2".into()), + content_blocks: None, + name: None, + tool_call_id: Some("t2".into()), + extra: serde_json::Map::new(), + }, + ], + ); + let (_system, msgs) = split_system(&req).unwrap(); + // user, assistant(2 tool_use), user(2 tool_result) + assert_eq!(msgs.len(), 3); + assert_eq!(msgs[1].role, "assistant"); + assert_eq!(msgs[1].content.len(), 2); + assert_eq!(msgs[2].role, "user"); + assert_eq!(msgs[2].content.len(), 2); + assert_eq!(msgs[2].content[0]["tool_use_id"], "t1"); + assert_eq!(msgs[2].content[1]["tool_use_id"], "t2"); } #[test] diff --git a/crates/aisix-proxy/src/lib.rs b/crates/aisix-proxy/src/lib.rs index 38484c5a..484e5e30 100644 --- a/crates/aisix-proxy/src/lib.rs +++ b/crates/aisix-proxy/src/lib.rs @@ -50,6 +50,7 @@ mod render; mod request_id; mod rerank; mod responses; +mod responses_bridge; mod routing; mod state; mod stream_timeout; diff --git a/crates/aisix-proxy/src/responses.rs b/crates/aisix-proxy/src/responses.rs index 33a4b230..78a6728a 100644 --- a/crates/aisix-proxy/src/responses.rs +++ b/crates/aisix-proxy/src/responses.rs @@ -20,7 +20,9 @@ use axum::response::{IntoResponse, Response}; use axum::Json; use futures::StreamExt; use serde_json::Value; +use std::sync::Arc; use std::time::{Duration, Instant}; +use uuid::Uuid; use crate::attempt::{ attempt_error_from_proxy, ms_since, AttemptInfo, AttemptRecord, RoutingTelemetry, @@ -83,7 +85,8 @@ impl From for ResponsesDispatchError { } /// Subset of the OpenAI Responses-API `usage` block the gateway -/// surfaces for telemetry. Other fields (`total_tokens`, +/// surfaces for telemetry (plus the two Anthropic cache counters carried +/// only on the #825 cross-provider bridge path). Other fields (`total_tokens`, /// `output_tokens_details.audio_tokens`, etc.) are intentionally /// dropped here — cp-api's `dpmgr_usage_events` table records only /// the ones below. @@ -99,6 +102,15 @@ struct ResponseUsage { /// OpenAI prompt-cache hit count, subset of `prompt_tokens`, /// surfaced via `usage.input_tokens_details.cached_tokens`. cached_prompt_tokens: u32, + /// Anthropic `cache_creation_input_tokens` (cache write). Always 0 on + /// the verbatim OpenAI path; carried for the cross-provider bridge + /// path (#825) so an Anthropic-backed /v1/responses call bills cache + /// writes the same way /v1/messages does. + cache_creation_tokens: u32, + /// Anthropic `cache_read_input_tokens` (cache read). Always 0 on the + /// verbatim OpenAI path (OpenAI surfaces cache hits via + /// `cached_prompt_tokens` instead). + cache_read_tokens: u32, } pub async fn responses( @@ -289,13 +301,16 @@ async fn dispatch( api_key_id: &auth.entry.id, team_id: auth.key().team_id.as_deref(), }; - let resolved_chain = state.guardrail_index.resolve(&guardrail_ctx); + // Arc so the chain can be cloned into the cross-provider streaming + // response body (which outlives this handler) for end-of-stream output + // guardrails (#825), mirroring /v1/messages. + let resolved_chain = Arc::new(state.guardrail_index.resolve(&guardrail_ctx)); if !resolved_chain.is_empty() { let chat = responses_input_to_chat(&model_name, body); if let aisix_guardrails::GuardrailVerdict::Block { reason, guardrail_name, - } = aisix_guardrails::Guardrail::check_input(&resolved_chain, &chat).await + } = aisix_guardrails::Guardrail::check_input(resolved_chain.as_ref(), &chat).await { // Per #153 the matched-pattern detail stays in ops logs only; the // wire envelope names only the guardrail that fired (#519 B.4b) @@ -321,9 +336,11 @@ async fn dispatch( crate::quota::ModelRateLimit::from_model(&model_name, &model_entry.id, &model_entry.value); let _reservation = crate::quota::enforce(state, auth, Some(&model_rl)).await?; - // Resolve the attempt list (routing-aware). /v1/responses is - // OpenAI-only, so we attempt the group's OpenAI targets in order; a - // direct model resolves to itself (#471). + // Resolve the attempt list (routing-aware). A Model Group walks its + // targets in order; a direct model resolves to itself (#471). OpenAI + // targets take the verbatim Responses passthrough; every other provider + // is bridged through ChatFormat (#825), so a group can mix and fail over + // across both kinds. let attempt_models = crate::routing::resolve_attempt_models( &state.routing, &state.runtime_status, @@ -341,24 +358,13 @@ async fn dispatch( let is_routing_request = model_entry.value.routing.is_some(); let mut routing = RoutingTelemetry::default(); - let not_openai = || { - ProxyError::InvalidRequest(format!( - "model `{model_name}` is not an OpenAI provider; /v1/responses requires OpenAI" - )) - }; - - // Walk the OpenAI targets, failing over on a retryable failure. - // Streaming and non-streaming share this loop: `responses_to_target` - // branches internally and, for streaming, only returns Ok once the - // first chunk has arrived under `stream_timeout` (#554) — so the 200 is - // committed to exactly one target and a slow first chunk fails over. + // Walk the targets, failing over on a retryable failure. Streaming and + // non-streaming share this loop: the per-target dispatch branches + // internally and, for streaming, only returns Ok once the first chunk + // has arrived under `stream_timeout` (#554) — so the 200 is committed to + // exactly one target and a slow first chunk fails over. let mut last_err: Option = None; - let mut any_openai = false; for target in &attempt_models { - if target.model.provider.as_deref() != Some("openai") { - continue; - } - any_openai = true; let (idx, kind) = routing.begin_attempt(&target.model.display_name); let target_model = if is_routing_request { target.model.display_name.clone() @@ -366,30 +372,49 @@ async fn dispatch( String::new() }; let attempt_started = Instant::now(); - match responses_to_target( - state, - &snapshot, - body, - &target.model, - &target.id, - request_id, - &resolved_chain, - started, - &model_name, - &auth.entry.id, - client, - // Winning-attempt classification (#655) for the streaming - // path's end-of-stream UsageEvent (#808). The non-streaming - // and buffered paths emit from the handler and ignore it. - AttemptInfo { - index: idx, - kind: kind.to_string(), - model: target_model.clone(), - ..Default::default() - }, - ) - .await - { + // Winning-attempt classification (#655) for the streaming path's + // end-of-stream UsageEvent. The non-streaming / buffered paths emit + // from the handler and ignore it. + let attempt = AttemptInfo { + index: idx, + kind: kind.to_string(), + model: target_model.clone(), + ..Default::default() + }; + let result = if target.model.provider.as_deref() == Some("openai") { + responses_to_target( + state, + &snapshot, + body, + &target.model, + &target.id, + request_id, + resolved_chain.as_ref(), + started, + &model_name, + &auth.entry.id, + client, + attempt, + ) + .await + } else { + responses_cross_provider_to_target( + state, + &snapshot, + body, + &target.model, + &target.id, + request_id, + resolved_chain.clone(), + started, + &model_name, + &auth.entry.id, + client, + attempt, + ) + .await + }; + match result { Ok(mut success) => { routing.attempts.push(AttemptRecord { index: idx, @@ -432,9 +457,6 @@ async fn dispatch( } } - if !any_openai { - return Err(not_openai().into()); - } Err(ResponsesDispatchError { err: last_err.unwrap_or(ProxyError::ProviderUnavailable), routing, @@ -942,6 +964,276 @@ async fn responses_to_target( } } +/// Dispatch one non-OpenAI target by bridging the Responses-API request +/// through the gateway's canonical [`ChatFormat`] and the provider +/// [`Bridge`](aisix_gateway::Bridge), then re-encoding the response back +/// into the Responses-API shape (#825). This is what lets clients like +/// `codex` — which speak only the OpenAI Responses API — reach an +/// Anthropic (or any other) backend. Mirrors `messages::cross_provider_dispatch`. +#[allow(clippy::too_many_arguments)] +async fn responses_cross_provider_to_target( + state: &ProxyState, + snapshot: &aisix_core::AisixSnapshot, + body: &Value, + model: &aisix_core::Model, + model_id: &str, + request_id: &str, + chain: Arc, + started: Instant, + requested_model: &str, + api_key_id: &str, + client_ctx: &ClientContext, + attempt: AttemptInfo, +) -> Result { + use aisix_gateway::{Bridge, BridgeContext}; + + let provider = model + .provider + .as_deref() + .ok_or_else(|| { + ProxyError::InvalidRequest(format!("model `{requested_model}` has no provider prefix")) + })? + .to_string(); + let pk_entry = crate::dispatch::resolve_provider_key(snapshot, model)?; + let bridge: Arc = crate::dispatch::resolve_bridge(&state.hub, &pk_entry.value) + .ok_or(ProxyError::ProviderUnavailable)?; + + // Faithful Responses → ChatFormat transform; `chat.model` stays the + // operator-facing name so the bridge re-resolves the upstream id via + // `ctx.model.upstream_model()` exactly like chat.rs. + let chat = crate::responses_bridge::responses_request_to_chat(requested_model, body); + + let is_stream = chat.is_streaming(); + let model_arc = Arc::new(model.clone()); + let pk_arc = Arc::new(pk_entry.value.clone()); + let mut ctx = BridgeContext::new(request_id, model_arc, pk_arc); + let connect_deadline = if is_stream { + model.stream_timeout_effective() + } else { + model.request_timeout() + }; + if let Some(d) = connect_deadline { + ctx = ctx.with_deadline(d); + } + let provider_label = provider.to_ascii_lowercase(); + + if is_stream { + let upstream = bridge.chat_stream(&chat, &ctx).await.map_err(|err| { + if let Some((ttl, reason)) = + crate::cooldown::decide_cooldown(&err, model.cooldown.as_ref()) + { + state.runtime_status.mark_cooldown(model_id, ttl, reason); + } + ProxyError::Bridge(err) + })?; + // #554: peek the first chunk so a slow/erroring first token fails + // over before the 200 is committed (when a stream budget is set); + // the wrapper keeps enforcing the per-chunk read timeout either way. + let stream_budget = model.stream_timeout_effective(); + let upstream = crate::stream_timeout::with_read_timeout(upstream, stream_budget); + let upstream: aisix_gateway::ChatChunkStream = if stream_budget.is_some() { + let mut upstream = upstream; + let first_chunk = match upstream.next().await { + Some(Ok(chunk)) => chunk, + Some(Err(err)) => { + if let Some((ttl, reason)) = + crate::cooldown::decide_cooldown(&err, model.cooldown.as_ref()) + { + state.runtime_status.mark_cooldown(model_id, ttl, reason); + } + return Err(ProxyError::Bridge(err)); + } + None => { + let err = aisix_gateway::BridgeError::StreamAborted; + if let Some((ttl, reason)) = + crate::cooldown::decide_cooldown(&err, model.cooldown.as_ref()) + { + state.runtime_status.mark_cooldown(model_id, ttl, reason); + } + return Err(ProxyError::Bridge(err)); + } + }; + Box::pin( + futures::stream::once(std::future::ready(Ok::<_, aisix_gateway::BridgeError>( + first_chunk, + ))) + .chain(upstream), + ) + } else { + upstream + }; + // Health tracks the concrete resolved target, not the (possibly + // group-alias) requested model — matching the non-streaming branch. + state.health.record_success(&model.display_name); + state.runtime_status.mark_healthy(model_id); + + let response_id = format!("resp_{}", Uuid::new_v4().simple()); + let created_at = chrono::Utc::now().timestamp(); + let encoder = crate::responses_bridge::ResponsesSseEncoder::new( + response_id, + requested_model, + created_at, + ); + // Only an output-hook guardrail needs the streamed response text. When + // attached, the bridge buffers the SSE and scans before releasing it + // (#719 secure default); cap the buffer the same way the verbatim path + // does so a huge response can't OOM the gateway. + let output_guardrail = (!chain.is_empty() + && aisix_guardrails::Guardrail::runs_on_output(chain.as_ref())) + .then(|| chain.clone()); + let max_buffer_bytes = + match aisix_guardrails::Guardrail::stream_output_policy(chain.as_ref()) { + aisix_guardrails::StreamOutputPolicy::BufferFull { + max_buffer_bytes, .. + } => max_buffer_bytes, + _ => aisix_guardrails::DEFAULT_STREAM_OUTPUT_BUFFER_BYTES, + }; + + let state_c = state.clone(); + let request_id_c = request_id.to_string(); + let model_id_c = model_id.to_string(); + let requested_model_c = requested_model.to_string(); + let api_key_id_c = api_key_id.to_string(); + let client_c = client_ctx.clone(); + let attempt_c = attempt.clone(); + let sse_body = crate::responses_bridge::build_responses_bridge_stream( + upstream, + encoder, + started, + output_guardrail, + max_buffer_bytes, + requested_model.to_string(), + move |comp| { + let usage = ResponseUsage { + prompt_tokens: comp.prompt_tokens, + completion_tokens: comp.completion_tokens, + reasoning_tokens: comp.reasoning_tokens, + cached_prompt_tokens: comp.cached_prompt_tokens, + cache_creation_tokens: comp.cache_creation_tokens, + cache_read_tokens: comp.cache_read_tokens, + }; + // A clean stream is a committed 200; an output-guardrail block + // (or fail-closed overflow) bills the upstream tokens but is + // recorded as a 422 marked guardrail_blocked, matching the + // non-streaming path so the Blocked tab + ledger see it. + let status = if comp.guardrail_blocked { 422 } else { 200 }; + emit_usage_event( + &state_c, + &request_id_c, + &model_id_c, + &requested_model_c, + &api_key_id_c, + status, + started.elapsed(), + &usage, + &client_c, + attempt_c, + comp.guardrail_blocked, + ); + }, + ); + let mut response = axum::response::Response::new(sse_body); + response.headers_mut().insert( + axum::http::header::CONTENT_TYPE, + HeaderValue::from_static("text/event-stream"), + ); + response.headers_mut().insert( + axum::http::header::CACHE_CONTROL, + HeaderValue::from_static("no-cache"), + ); + if let Ok(hv) = HeaderValue::from_str(request_id) { + response + .headers_mut() + .insert(HeaderName::from_static("x-aisix-request-id"), hv); + } + return Ok(ResponseDispatchSuccess { + response, + provider: provider_label, + usage: None, + model_id: model_id.to_string(), + routing: RoutingTelemetry::default(), + guardrail_blocked: false, + usage_handled_by_stream: true, + }); + } + + // Non-streaming. + let resp = bridge.chat(&chat, &ctx).await.map_err(|err| { + if let Some((ttl, reason)) = crate::cooldown::decide_cooldown(&err, model.cooldown.as_ref()) + { + state.runtime_status.mark_cooldown(model_id, ttl, reason); + } + ProxyError::Bridge(err) + })?; + state.health.record_success(&model.display_name); + state.runtime_status.mark_healthy(model_id); + + let usage = ResponseUsage { + prompt_tokens: resp.usage.prompt_tokens, + completion_tokens: resp.usage.completion_tokens, + reasoning_tokens: resp.usage.reasoning_tokens, + cached_prompt_tokens: resp.usage.cached_prompt_tokens, + cache_creation_tokens: resp.usage.cache_creation_tokens, + cache_read_tokens: resp.usage.cache_read_tokens, + }; + + // #719: run output guardrails on the bridged response before re-encoding + // it as Responses JSON — the assistant text + tool calls are + // client-visible output, scanned the same way /v1/chat/completions does. + if aisix_guardrails::Guardrail::runs_on_output(chain.as_ref()) { + if let aisix_guardrails::GuardrailVerdict::Block { + reason, + guardrail_name, + } = aisix_guardrails::Guardrail::check_output(chain.as_ref(), &resp).await + { + tracing::warn!( + guardrail_hook = "output", + model = %requested_model, + reason = %reason, + "guardrail blocked /v1/responses (cross-provider) response", + ); + // #543: the upstream already billed — return the 422 body but + // carry the billed usage (marked guardrail_blocked) so the + // ledger doesn't underreport spend. + return Ok(ResponseDispatchSuccess { + response: ProxyError::ContentFiltered(crate::error::guardrail_block_message( + "response", + guardrail_name.as_deref(), + )) + .into_response(), + provider: provider_label, + usage: Some(usage), + model_id: model_id.to_string(), + routing: RoutingTelemetry::default(), + guardrail_blocked: true, + usage_handled_by_stream: false, + }); + } + } + + let created_at = chrono::Utc::now().timestamp(); + let json_body = crate::responses_bridge::chat_response_to_responses_json( + &resp, + requested_model, + created_at, + ); + let mut response = Json(json_body).into_response(); + if let Ok(hv) = HeaderValue::from_str(request_id) { + response + .headers_mut() + .insert(HeaderName::from_static("x-aisix-request-id"), hv); + } + Ok(ResponseDispatchSuccess { + response, + provider: provider_label, + usage: Some(usage), + model_id: model_id.to_string(), + routing: RoutingTelemetry::default(), + guardrail_blocked: false, + usage_handled_by_stream: false, + }) +} + /// Pull the usage counters out of a Responses-API non-streaming /// response body. Returns `None` only when: /// - The `usage` block is missing entirely, OR @@ -980,6 +1272,9 @@ fn extract_response_usage(body: &Value) -> Option { completion_tokens, reasoning_tokens, cached_prompt_tokens, + // OpenAI verbatim path: no Anthropic-style cache counters. + cache_creation_tokens: 0, + cache_read_tokens: 0, }) } @@ -1267,10 +1562,12 @@ fn apply_passthrough_headers( /// `embeddings::emit_usage_event` (#402) for the fields that matter /// to /v1/responses, with one extension: `reasoning_tokens` is /// surfaced for o1/o3/GPT-5 class models. `inbound_protocol` is -/// `"openai"` — Responses API is OpenAI-only. +/// `"openai"` — the Responses API is OpenAI-shaped on the wire even when +/// the resolved model is bridged to a non-OpenAI provider (#825). /// /// Other fields left at `UsageEvent::default()`: -/// - cache_creation_tokens / cache_read_tokens — Anthropic-only +/// - cache_creation_tokens / cache_read_tokens — populated only on the +/// #825 cross-provider bridge path (Anthropic backends); 0 otherwise /// - provider_request_id / provider_model_version / finish_reason /// — not yet plumbed for non-chat handlers (follow-up) /// - cost_usd — cp-api computes server-side from pricing catalog @@ -1304,6 +1601,10 @@ fn emit_usage_event( completion_tokens: usage.completion_tokens, cached_prompt_tokens: usage.cached_prompt_tokens, reasoning_tokens: usage.reasoning_tokens, + // Anthropic cache counters (#825 cross-provider path); 0 on the + // verbatim OpenAI path. + cache_creation_tokens: usage.cache_creation_tokens, + cache_read_tokens: usage.cache_read_tokens, latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32, status_code, inbound_protocol: "openai".to_string(), @@ -1441,6 +1742,7 @@ mod tests { use aisix_core::snapshot::SnapshotHandle; use aisix_core::{AisixSnapshot, ApiKey, Model, ProxyConfig}; use aisix_gateway::Hub; + use aisix_provider_anthropic::AnthropicBridge; use aisix_provider_openai::OpenAiBridge; use axum::body::to_bytes; use axum::http::{Request, StatusCode}; @@ -1485,10 +1787,11 @@ mod tests { ResourceEntry::new(OPENAI_PK_ID, pk, 1) } - fn anthropic_pk() -> ResourceEntry { - let pk: aisix_core::ProviderKey = - serde_json::from_str(r#"{"display_name":"anthropic-up","secret":"sk-ant-test","provider":"anthropic","adapter":"anthropic"}"#) - .unwrap(); + fn anthropic_pk_at(api_base: &str) -> ResourceEntry { + let json = format!( + r#"{{"display_name":"anthropic-up","secret":"sk-ant-test","api_base":"{api_base}","provider":"anthropic","adapter":"anthropic"}}"# + ); + let pk: aisix_core::ProviderKey = serde_json::from_str(&json).unwrap(); ResourceEntry::new(ANTHROPIC_PK_ID, pk, 1) } @@ -1498,9 +1801,9 @@ mod tests { snap } - fn new_snap_anthropic() -> AisixSnapshot { + fn new_snap_anthropic_at(api_base: &str) -> AisixSnapshot { let snap = AisixSnapshot::new(); - snap.provider_keys.insert(anthropic_pk()); + snap.provider_keys.insert(anthropic_pk_at(api_base)); snap } @@ -1516,6 +1819,10 @@ mod tests { fn build_app(snap: AisixSnapshot) -> axum::Router { let hub = Arc::new(Hub::new()); hub.register_specialized("openai", Arc::new(OpenAiBridge::new())); + // #825: the cross-provider /v1/responses path bridges non-OpenAI + // targets through the provider Bridge; register Anthropic so those + // tests resolve a bridge. + hub.register_specialized("anthropic", Arc::new(AnthropicBridge::new())); let handle = SnapshotHandle::new(snap); crate::build_router(crate::ProxyState::new(handle, hub, &cfg()).without_cache()) } @@ -1956,6 +2263,170 @@ mod tests { ); } + /// Anthropic Messages streaming SSE carrying a single text delta. + fn anthropic_text_sse(text: &str) -> String { + format!( + "event: message_start\n\ + data: {{\"type\":\"message_start\",\"message\":{{\"id\":\"msg_g\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"claude-3-haiku-20240307\",\"content\":[],\"usage\":{{\"input_tokens\":5,\"output_tokens\":0}}}}}}\n\n\ + event: content_block_start\n\ + data: {{\"type\":\"content_block_start\",\"index\":0,\"content_block\":{{\"type\":\"text\",\"text\":\"\"}}}}\n\n\ + event: content_block_delta\n\ + data: {{\"type\":\"content_block_delta\",\"index\":0,\"delta\":{{\"type\":\"text_delta\",\"text\":{text}}}}}\n\n\ + event: content_block_stop\n\ + data: {{\"type\":\"content_block_stop\",\"index\":0}}\n\n\ + event: message_delta\n\ + data: {{\"type\":\"message_delta\",\"delta\":{{\"stop_reason\":\"end_turn\"}},\"usage\":{{\"output_tokens\":3}}}}\n\n\ + event: message_stop\n\ + data: {{\"type\":\"message_stop\"}}\n\n", + text = serde_json::to_string(text).unwrap(), + ) + } + + /// #825 + #719: the cross-provider (bridged) streaming path must enforce + /// output guardrails too — else `stream:true` against a non-OpenAI model + /// bypasses the block. The bridge buffers the encoded SSE and, on a + /// block, emits only a terminal `error` event; no output_text delta with + /// the blocked literal reaches the client. + #[tokio::test] + async fn output_guardrail_blocks_streaming_cross_provider_response() { + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/messages")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(anthropic_text_sse("sure: BLOCKME here")), + ) + .mount(&upstream) + .await; + + let snap = new_snap_anthropic_at(&upstream.uri()); + snap.models.insert(anthropic_model("claude-resp")); + snap.apikeys.insert(apikey_entry(&["*"])); + snap.guardrails.insert(keyword_output_guardrail("BLOCKME")); + let app = build_app(snap); + + let resp = app + .oneshot(make_req( + serde_json::json!({"model":"claude-resp","input":"hi","stream":true}), + )) + .await + .unwrap(); + // The SSE 200 is committed by the first-chunk failover peek; the block + // surfaces as an in-band terminal error event. + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), 65536).await.unwrap(); + let body = String::from_utf8_lossy(&bytes); + assert!( + body.contains("content_filter"), + "missing block error: {body}" + ); + assert!( + !body.contains("BLOCKME"), + "blocked content leaked in stream: {body}" + ); + assert!( + !body.contains("response.output_text.delta"), + "held-back deltas leaked: {body}" + ); + } + + /// #825 companion: a clean bridged streaming response with an output + /// guardrail is scanned then released in full. + #[tokio::test] + async fn output_guardrail_allows_clean_streaming_cross_provider_response() { + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/messages")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(anthropic_text_sse("a clean answer")), + ) + .mount(&upstream) + .await; + + let snap = new_snap_anthropic_at(&upstream.uri()); + snap.models.insert(anthropic_model("claude-resp")); + snap.apikeys.insert(apikey_entry(&["*"])); + snap.guardrails.insert(keyword_output_guardrail("BLOCKME")); + let app = build_app(snap); + + let resp = app + .oneshot(make_req( + serde_json::json!({"model":"claude-resp","input":"hi","stream":true}), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), 65536).await.unwrap(); + let body = String::from_utf8_lossy(&bytes); + assert!( + body.contains("a clean answer"), + "clean body withheld: {body}" + ); + assert!(body.contains("response.completed")); + } + + /// #825: a blocked cross-provider STREAM still bills the upstream tokens + /// but the emitted UsageEvent is marked guardrail_blocked (status 422) — + /// matching the non-streaming path — so the dashboard's Blocked tab and + /// the budget ledger see it rather than recording it as clean usage. + #[tokio::test] + async fn streaming_cross_provider_block_emits_guardrail_blocked_usage_event() { + use aisix_obs::UsageSink; + + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/messages")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(anthropic_text_sse("sure: BLOCKME")), + ) + .mount(&upstream) + .await; + + let snap = new_snap_anthropic_at(&upstream.uri()); + snap.models.insert(anthropic_model("claude-resp")); + snap.apikeys.insert(apikey_entry(&["*"])); + snap.guardrails.insert(keyword_output_guardrail("BLOCKME")); + + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + let hub = Arc::new(Hub::new()); + hub.register_specialized("openai", Arc::new(OpenAiBridge::new())); + hub.register_specialized("anthropic", Arc::new(AnthropicBridge::new())); + let handle = SnapshotHandle::new(snap); + let state = crate::ProxyState::new(handle, hub, &cfg()) + .without_cache() + .with_usage_sink(UsageSink::new(tx)); + let app = crate::build_router(state); + + let resp = app + .oneshot(make_req( + serde_json::json!({"model":"claude-resp","input":"hi","stream":true}), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + // Drain the body so the stream's Drop guard fires the usage event. + let bytes = to_bytes(resp.into_body(), 65536).await.unwrap(); + assert!(!String::from_utf8_lossy(&bytes).contains("BLOCKME")); + + let event = tokio::time::timeout(std::time::Duration::from_millis(1000), rx.recv()) + .await + .expect("usage event must be emitted") + .expect("usage_sink sender dropped"); + assert!( + event.guardrail_blocked, + "a blocked stream must mark guardrail_blocked" + ); + assert_eq!(event.status_code, 422); + // The upstream-billed tokens are still recorded. + assert_eq!(event.prompt_tokens, 5); + assert_eq!(event.completion_tokens, 3); + } + /// #719 (audit HIGH-1): the streaming hold-back buffer is capped so a /// huge (or malicious) upstream response can't OOM the gateway. A /// response exceeding the BufferFull cap fails closed (422) rather than @@ -2310,9 +2781,29 @@ mod tests { assert_eq!(resp.status(), StatusCode::NOT_FOUND); } + /// #825: an Anthropic-backed model is no longer rejected on + /// /v1/responses — the request is bridged through ChatFormat to the + /// Anthropic Messages upstream and the reply is re-encoded into the + /// Responses-API shape. This is the codex-against-Anthropic path. #[tokio::test] - async fn non_openai_model_returns_400() { - let snap = new_snap_anthropic(); + async fn non_openai_model_bridges_to_responses_shape() { + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/messages")) + .and(header("x-api-key", "sk-ant-test")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "id": "msg_xprov", + "type": "message", + "role": "assistant", + "model": "claude-3-haiku-20240307", + "content": [{"type": "text", "text": "Hi from Claude"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 9, "output_tokens": 4} + }))) + .mount(&upstream) + .await; + + let snap = new_snap_anthropic_at(&upstream.uri()); snap.models.insert(anthropic_model("claude-haiku")); snap.apikeys.insert(apikey_entry(&["*"])); let app = build_app(snap); @@ -2324,7 +2815,73 @@ mod tests { }))) .await .unwrap(); - assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body["object"], "response"); + assert_eq!(body["status"], "completed"); + // Operator-facing model name echoed, not the upstream id. + assert_eq!(body["model"], "claude-haiku"); + assert_eq!(body["output"][0]["type"], "message"); + assert_eq!(body["output"][0]["content"][0]["type"], "output_text"); + assert_eq!(body["output"][0]["content"][0]["text"], "Hi from Claude"); + assert_eq!(body["usage"]["input_tokens"], 9); + assert_eq!(body["usage"]["output_tokens"], 4); + } + + /// #825 streaming: a streamed Anthropic-backed /v1/responses call emits + /// the canonical Responses SSE event sequence ending in + /// `response.completed` (the exact codex-tui path). + #[tokio::test] + async fn non_openai_streaming_bridges_to_responses_sse() { + let sse = concat!( + "event: message_start\n", + "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_s\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"claude-3-haiku-20240307\",\"content\":[],\"usage\":{\"input_tokens\":6,\"output_tokens\":0}}}\n\n", + "event: content_block_start\n", + "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n", + "event: content_block_delta\n", + "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"Hi\"}}\n\n", + "event: content_block_stop\n", + "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n", + "event: message_delta\n", + "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":3}}\n\n", + "event: message_stop\n", + "data: {\"type\":\"message_stop\"}\n\n", + ); + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/messages")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse), + ) + .mount(&upstream) + .await; + + let snap = new_snap_anthropic_at(&upstream.uri()); + snap.models.insert(anthropic_model("claude-haiku")); + snap.apikeys.insert(apikey_entry(&["*"])); + let app = build_app(snap); + + let resp = app + .oneshot(make_req(serde_json::json!({ + "model": "claude-haiku", + "input": "hi", + "stream": true + }))) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let text = String::from_utf8(bytes.to_vec()).unwrap(); + assert!( + text.contains("event: response.created"), + "missing created: {text}" + ); + assert!(text.contains("event: response.output_text.delta")); + assert!(text.contains("\"delta\":\"Hi\"")); + assert!(text.contains("event: response.completed")); } #[tokio::test] diff --git a/crates/aisix-proxy/src/responses_bridge.rs b/crates/aisix-proxy/src/responses_bridge.rs new file mode 100644 index 00000000..ce7d8b17 --- /dev/null +++ b/crates/aisix-proxy/src/responses_bridge.rs @@ -0,0 +1,1525 @@ +//! Cross-provider translation for `POST /v1/responses` (#825). +//! +//! The Responses API is OpenAI-specific, but clients such as the `codex` +//! CLI point it at non-OpenAI models. For an OpenAI upstream the handler +//! forwards the body verbatim (see [`crate::responses`]); for any other +//! provider this module translates the request into the gateway's +//! canonical [`ChatFormat`], so it can be dispatched through the same +//! provider [`Bridge`](aisix_gateway::Bridge) `/v1/chat/completions` uses, +//! and re-encodes the bridge's response back into the Responses API shape +//! — non-streaming JSON and streaming SSE. This mirrors the cross-provider +//! path of `/v1/messages` (`messages::cross_provider_dispatch`). +//! +//! Only the Responses fields that map cleanly onto chat completions are +//! carried (`instructions`, `input`, `tools`, `tool_choice`, +//! `temperature`, `top_p`, `max_output_tokens`, `stream`). OpenAI-only +//! knobs (`reasoning`, `store`, `previous_response_id`, `text`, …) are +//! dropped rather than forwarded — the downstream provider bridges flatten +//! unknown `extra` fields onto the upstream wire, where an OpenAI-only key +//! would 400 (e.g. Anthropic). Reasoning/thinking has no canonical bridge +//! mapping today, so it is intentionally not translated. + +use std::sync::Arc; +use std::time::Instant; + +use aisix_gateway::{ + ChatChunk, ChatChunkStream, ChatFormat, ChatMessage, ChatResponse, FinishReason, Role, + UsageStats, +}; +use serde_json::{json, Map, Value}; +use uuid::Uuid; + +/// Translate a `/v1/responses` request body into the gateway's canonical +/// [`ChatFormat`]. Unlike `responses::responses_input_to_chat` (which is a +/// lossy, text-only projection used solely for input-guardrail scanning), +/// this is the faithful transform actually dispatched upstream: it carries +/// roles, tool calls, tool results, tools, and sampling params. +pub fn responses_request_to_chat(model: &str, body: &Value) -> ChatFormat { + let mut messages: Vec = Vec::new(); + + // Top-level `instructions` is the Responses-API system prompt. + if let Some(instructions) = body.get("instructions").and_then(|v| v.as_str()) { + if !instructions.is_empty() { + messages.push(ChatMessage::system(instructions.to_string())); + } + } + + match body.get("input") { + Some(Value::String(text)) => { + if !text.is_empty() { + messages.push(ChatMessage::user(text.clone())); + } + } + Some(Value::Array(items)) => { + for item in items { + append_input_item(&mut messages, item); + } + } + _ => {} + } + + let mut chat = ChatFormat::new(model, messages); + chat.temperature = body + .get("temperature") + .and_then(|v| v.as_f64()) + .map(|f| f as f32); + chat.top_p = body.get("top_p").and_then(|v| v.as_f64()).map(|f| f as f32); + // Responses calls the cap `max_output_tokens`; tolerate `max_tokens` + // too for clients that send the chat-style name. A value that doesn't + // fit u32 is dropped (left unset) rather than silently wrapped to a + // small/zero cap. + chat.max_tokens = body + .get("max_output_tokens") + .or_else(|| body.get("max_tokens")) + .and_then(|v| v.as_u64()) + .and_then(|n| u32::try_from(n).ok()); + chat.stream = body.get("stream").and_then(|v| v.as_bool()); + + // Tools/tool_choice ride `extra` in OpenAI chat shape; every provider + // bridge translates that shape to its own (Anthropic, Gemini, …), so + // emitting it here is all that's needed. + if let Some(tools) = body.get("tools").and_then(responses_tools_to_chat) { + chat.extra.insert("tools".to_string(), tools); + } + if let Some(tc) = body + .get("tool_choice") + .and_then(responses_tool_choice_to_chat) + { + chat.extra.insert("tool_choice".to_string(), tc); + } + chat +} + +/// Append one Responses-API `input` array element as chat message(s). +fn append_input_item(messages: &mut Vec, item: &Value) { + // A bare-string element is user text. + if let Some(text) = item.as_str() { + if !text.is_empty() { + messages.push(ChatMessage::user(text.to_string())); + } + return; + } + + match item.get("type").and_then(|t| t.as_str()) { + // A prior assistant tool call replayed for the agent loop. + Some("function_call") => { + let call_id = item + .get("call_id") + .or_else(|| item.get("id")) + .and_then(|v| v.as_str()) + .unwrap_or_default(); + let name = item + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + let arguments = item.get("arguments").and_then(|v| v.as_str()).unwrap_or(""); + push_tool_call( + messages, + json!({ + "id": call_id, + "type": "function", + "function": {"name": name, "arguments": arguments}, + }), + ); + } + // The tool result fed back by the caller → a `tool` role message. + Some("function_call_output") => { + let call_id = item + .get("call_id") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + let output = item + .get("output") + .map(responses_content_text) + .unwrap_or_default(); + messages.push(ChatMessage { + role: Role::Tool, + content: Some(output), + content_blocks: None, + name: None, + tool_call_id: Some(call_id.to_string()), + extra: Map::new(), + }); + } + // Reasoning items can't be replayed across providers — drop them. + Some("reasoning") => {} + // A `message` item (or an untyped `{role, content}` element). + _ => { + let role = item.get("role").and_then(|v| v.as_str()).unwrap_or("user"); + let text = item + .get("content") + .map(responses_content_text) + .unwrap_or_default(); + if text.is_empty() { + return; + } + messages.push(match role { + "assistant" => ChatMessage::assistant(text), + "system" | "developer" => ChatMessage::system(text), + _ => ChatMessage::user(text), + }); + } + } +} + +/// Append an OpenAI-shape tool call, folding it into the immediately +/// preceding assistant tool-call message when contiguous so parallel +/// `function_call` items land in one assistant turn (one `tool_calls` +/// array) — the standard OpenAI history shape every bridge expects. +fn push_tool_call(messages: &mut Vec, tc: Value) { + if let Some(last) = messages.last_mut() { + if matches!(last.role, Role::Assistant) && last.content.is_none() { + if let Some(Value::Array(arr)) = last.extra.get_mut("tool_calls") { + arr.push(tc); + return; + } + } + } + let mut extra = Map::new(); + extra.insert("tool_calls".to_string(), Value::Array(vec![tc])); + messages.push(ChatMessage { + role: Role::Assistant, + content: None, + content_blocks: None, + name: None, + tool_call_id: None, + extra, + }); +} + +/// Plain text of a Responses-API content slot: a bare string, or the +/// concatenation of the `text` of an array of typed parts +/// (`input_text` / `output_text` / `text`). Non-text parts are skipped. +fn responses_content_text(v: &Value) -> String { + match v { + Value::String(s) => s.clone(), + Value::Array(parts) => parts + .iter() + .filter_map(|p| p.get("text").and_then(|t| t.as_str())) + .collect::>() + .join(""), + _ => String::new(), + } +} + +/// Translate Responses-API `tools` (flat function shape `{type:"function", +/// name, description, parameters}`) into OpenAI chat tools (`{type: +/// "function", function:{name, description, parameters}}`). Non-function +/// (hosted) tools have no chat equivalent and are dropped. Returns `None` +/// when nothing translates so the field stays absent from the wire. +fn responses_tools_to_chat(tools: &Value) -> Option { + let arr = tools.as_array()?; + let out: Vec = arr + .iter() + .filter_map(|t| { + if t.get("type").and_then(|v| v.as_str()) != Some("function") { + return None; + } + let name = t.get("name").and_then(|v| v.as_str())?; + let mut func = Map::new(); + func.insert("name".to_string(), json!(name)); + if let Some(d) = t.get("description") { + func.insert("description".to_string(), d.clone()); + } + if let Some(p) = t.get("parameters") { + func.insert("parameters".to_string(), p.clone()); + } + Some(json!({"type": "function", "function": Value::Object(func)})) + }) + .collect(); + (!out.is_empty()).then_some(Value::Array(out)) +} + +/// Translate Responses-API `tool_choice` to OpenAI chat shape: +/// `"auto"|"none"|"required"` pass through; `{type:"function", name}` → +/// `{type:"function", function:{name}}`. Hosted-tool choices have no chat +/// equivalent and drop to `None`. +fn responses_tool_choice_to_chat(tc: &Value) -> Option { + match tc { + Value::String(s) => Some(Value::String(s.clone())), + Value::Object(o) => { + if o.get("type").and_then(|v| v.as_str()) == Some("function") { + let name = o.get("name").and_then(|v| v.as_str())?; + Some(json!({"type": "function", "function": {"name": name}})) + } else { + None + } + } + _ => None, + } +} + +/// Build the non-streaming Responses-API response object from a bridge +/// [`ChatResponse`]. `requested_model` is echoed back (not the upstream +/// id). `created_at` is a unix timestamp stamped by the caller. +pub fn chat_response_to_responses_json( + resp: &ChatResponse, + requested_model: &str, + created_at: i64, +) -> Value { + let (status, incomplete_reason) = responses_status(&resp.finish_reason); + let output = build_output_items( + resp.message.content.as_deref(), + resp.message + .extra + .get("tool_calls") + .and_then(|v| v.as_array()), + ); + + let mut obj = json!({ + "id": format!("resp_{}", Uuid::new_v4().simple()), + "object": "response", + "created_at": created_at, + "status": status, + "model": requested_model, + "output": output, + "usage": responses_usage_json(&resp.usage), + }); + if let Some(reason) = incomplete_reason { + obj["incomplete_details"] = json!({"reason": reason}); + } + obj +} + +/// Map an internal finish reason to a Responses-API `status` plus optional +/// `incomplete_details.reason`. +fn responses_status(fr: &FinishReason) -> (&'static str, Option<&'static str>) { + match fr { + FinishReason::Length => ("incomplete", Some("max_output_tokens")), + FinishReason::ContentFilter => ("incomplete", Some("content_filter")), + _ => ("completed", None), + } +} + +/// Assemble the `output` array: a `message` item carrying the assistant +/// text (when any), followed by one `function_call` item per tool call. +fn build_output_items(text: Option<&str>, tool_calls: Option<&Vec>) -> Vec { + let mut output: Vec = Vec::new(); + if let Some(text) = text.filter(|s| !s.is_empty()) { + output.push(json!({ + "type": "message", + "id": format!("msg_{}", Uuid::new_v4().simple()), + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": text, "annotations": []}], + })); + } + if let Some(tool_calls) = tool_calls { + for tc in tool_calls { + let call_id = tc.get("id").and_then(|v| v.as_str()).unwrap_or_default(); + let name = tc + .get("function") + .and_then(|f| f.get("name")) + .and_then(|n| n.as_str()) + .unwrap_or_default(); + let arguments = tc + .get("function") + .and_then(|f| f.get("arguments")) + .and_then(|a| a.as_str()) + .unwrap_or(""); + output.push(json!({ + "type": "function_call", + "id": format!("fc_{}", Uuid::new_v4().simple()), + "call_id": call_id, + "name": name, + "arguments": arguments, + "status": "completed", + })); + } + } + output +} + +/// Render usage in the Responses-API shape. `cached_tokens` takes whichever +/// of the OpenAI-normalized hit count or the Anthropic cache-read count is +/// present (the other is 0). +fn responses_usage_json(u: &UsageStats) -> Value { + let total = if u.total_tokens > 0 { + u.total_tokens + } else { + u.prompt_tokens.saturating_add(u.completion_tokens) + }; + json!({ + "input_tokens": u.prompt_tokens, + "input_tokens_details": {"cached_tokens": u.cached_prompt_tokens.max(u.cache_read_tokens)}, + "output_tokens": u.completion_tokens, + "output_tokens_details": {"reasoning_tokens": u.reasoning_tokens}, + "total_tokens": total, + }) +} + +// ───────────────────────────────────────────────────────────────────── +// Streaming SSE encoder — internal ChatChunk stream → Responses-API +// SSE events. +// +// Event order for a text response: +// response.created → response.in_progress +// → response.output_item.added (message) +// → response.content_part.added (output_text) +// → response.output_text.delta ×N +// → response.output_text.done → response.content_part.done +// → response.output_item.done (message) +// → response.completed +// +// Tool calls add, per call: +// response.output_item.added (function_call) +// → response.function_call_arguments.delta ×N +// → response.function_call_arguments.done +// → response.output_item.done (function_call) +// +// `response.completed` carries the final output + usage. When an +// OpenAI-compatible upstream sends its usage frame AFTER the finish chunk +// (`stream_options.include_usage`), the completed event is withheld until +// the usage arrives (or `force_finish`), so token counts aren't zeroed. +// +// Reference: https://platform.openai.com/docs/api-reference/responses-streaming +// ───────────────────────────────────────────────────────────────────── + +/// One Responses-API SSE event, written as `event: {type}\ndata: {json}\n\n`. +#[derive(Debug, Clone)] +pub struct ResponsesSseEvent { + pub event_type: &'static str, + pub data: Value, +} + +impl ResponsesSseEvent { + pub fn to_sse_string(&self) -> String { + format!( + "event: {}\ndata: {}\n\n", + self.event_type, + serde_json::to_string(&self.data).expect("serde_json::Value always serializes"), + ) + } +} + +/// Per-tool-call streaming state. +#[derive(Debug)] +struct ToolCallState { + item_id: String, + call_id: String, + name: String, + output_index: u32, + arguments: String, + item_added: bool, +} + +/// State machine re-encoding a `ChatChunk` stream as Responses-API SSE. +#[derive(Debug)] +pub struct ResponsesSseEncoder { + response_id: String, + model_display_name: String, + created_at: i64, + sequence_number: u64, + sent_created: bool, + finished: bool, + /// Next output-item index to assign (shared by the message + tool items). + next_output_index: u32, + // Text message item. + text_item_id: Option, + text_output_index: u32, + text_accum: String, + /// Set once the per-item `*.done` events have been emitted, so + /// `close_items` is idempotent across the finish chunk + `force_finish`. + items_closed: bool, + // Tool-call items keyed by the OpenAI delta index. + tool_calls: std::collections::BTreeMap, + /// Withheld terminal status + incomplete reason while waiting on a + /// trailing usage frame. + pending_status: Option<&'static str>, + pending_reason: Option<&'static str>, + // Accumulated usage (max semantics, robust to double-emit). + usage_seen: bool, + prompt_tokens: u32, + completion_tokens: u32, + total_tokens: u32, + reasoning_tokens: u32, + cached_prompt_tokens: u32, + cache_creation_tokens: u32, + cache_read_tokens: u32, +} + +impl ResponsesSseEncoder { + pub fn new( + response_id: impl Into, + model_display_name: impl Into, + created_at: i64, + ) -> Self { + Self { + response_id: response_id.into(), + model_display_name: model_display_name.into(), + created_at, + sequence_number: 0, + sent_created: false, + finished: false, + next_output_index: 0, + text_item_id: None, + text_output_index: 0, + text_accum: String::new(), + items_closed: false, + tool_calls: std::collections::BTreeMap::new(), + pending_status: None, + pending_reason: None, + usage_seen: false, + prompt_tokens: 0, + completion_tokens: 0, + total_tokens: 0, + reasoning_tokens: 0, + cached_prompt_tokens: 0, + cache_creation_tokens: 0, + cache_read_tokens: 0, + } + } + + /// Build one event, stamping `type` + `sequence_number`. + fn event(&mut self, event_type: &'static str, mut data: Value) -> ResponsesSseEvent { + let seq = self.sequence_number; + self.sequence_number += 1; + if let Value::Object(map) = &mut data { + map.insert("type".to_string(), json!(event_type)); + map.insert("sequence_number".to_string(), json!(seq)); + } + ResponsesSseEvent { event_type, data } + } + + fn accumulate_usage(&mut self, chunk: &ChatChunk) { + if let Some(u) = chunk.usage.as_ref() { + self.usage_seen = true; + self.prompt_tokens = self.prompt_tokens.max(u.prompt_tokens); + self.completion_tokens = self.completion_tokens.max(u.completion_tokens); + self.total_tokens = self.total_tokens.max(u.total_tokens); + self.reasoning_tokens = self.reasoning_tokens.max(u.reasoning_tokens); + self.cached_prompt_tokens = self.cached_prompt_tokens.max(u.cached_prompt_tokens); + self.cache_creation_tokens = self.cache_creation_tokens.max(u.cache_creation_tokens); + self.cache_read_tokens = self.cache_read_tokens.max(u.cache_read_tokens); + } + } + + fn usage_value(&self) -> Value { + // `responses_usage_json` keeps a provider-supplied `total_tokens` + // when present and only falls back to prompt+completion when it's 0. + responses_usage_json(&UsageStats { + prompt_tokens: self.prompt_tokens, + completion_tokens: self.completion_tokens, + total_tokens: self.total_tokens, + cached_prompt_tokens: self.cached_prompt_tokens, + reasoning_tokens: self.reasoning_tokens, + cache_creation_tokens: self.cache_creation_tokens, + cache_read_tokens: self.cache_read_tokens, + ..Default::default() + }) + } + + /// The assembled assistant output for an end-of-stream output guardrail + /// scan: the full accumulated text plus the fully-reassembled tool calls + /// in canonical OpenAI `{id, type, function:{name, arguments}}` shape (so + /// an argument literal split across chunks is scanned as one string, not + /// as disjoint fragments). + pub fn assembled_assistant_message(&self) -> (String, Vec) { + let mut tool_calls: Vec<(u32, Value)> = self + .tool_calls + .values() + .map(|tc| { + ( + tc.output_index, + json!({ + "id": tc.call_id, + "type": "function", + "function": {"name": tc.name, "arguments": tc.arguments}, + }), + ) + }) + .collect(); + tool_calls.sort_by_key(|(idx, _)| *idx); + ( + self.text_accum.clone(), + tool_calls.into_iter().map(|(_, v)| v).collect(), + ) + } + + /// The bare response object embedded in lifecycle events. + fn response_object(&self, status: &str, with_output: bool, with_usage: bool) -> Value { + let mut obj = json!({ + "id": self.response_id, + "object": "response", + "created_at": self.created_at, + "status": status, + "model": self.model_display_name, + "output": if with_output { Value::Array(self.final_output_items()) } else { json!([]) }, + }); + if with_usage { + obj["usage"] = self.usage_value(); + } + obj + } + + /// Rebuild the completed `output` array from accumulated state. + fn final_output_items(&self) -> Vec { + let mut items: Vec<(u32, Value)> = Vec::new(); + if let Some(id) = self.text_item_id.as_ref() { + items.push(( + self.text_output_index, + json!({ + "type": "message", + "id": id, + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": self.text_accum, "annotations": []}], + }), + )); + } + for tc in self.tool_calls.values() { + items.push(( + tc.output_index, + json!({ + "type": "function_call", + "id": tc.item_id, + "call_id": tc.call_id, + "name": tc.name, + "arguments": tc.arguments, + "status": "completed", + }), + )); + } + items.sort_by_key(|(idx, _)| *idx); + items.into_iter().map(|(_, v)| v).collect() + } + + /// Translate one chunk into the SSE events to emit (possibly empty). + pub fn next_events(&mut self, chunk: &ChatChunk) -> Vec { + if self.finished { + return Vec::new(); + } + self.accumulate_usage(chunk); + + // Terminal status withheld for a trailing usage frame: release it + // once usage lands. Post-finish chunks carry no renderable content. + if let Some(status) = self.pending_status { + if self.usage_seen { + self.pending_status = None; + let reason = self.pending_reason.take(); + return vec![self.completed_event(status, reason)]; + } + return Vec::new(); + } + + let has_content = chunk + .delta + .content + .as_deref() + .is_some_and(|s| !s.is_empty()); + let has_tools = chunk + .delta + .tool_calls + .as_ref() + .is_some_and(|v| !v.is_empty()); + let has_finish = chunk.finish_reason.is_some(); + + let mut events = Vec::new(); + + if !self.sent_created && (has_content || has_tools || has_finish) { + self.sent_created = true; + events.push(self.event( + "response.created", + json!({"response": self.response_object("in_progress", false, false)}), + )); + events.push(self.event( + "response.in_progress", + json!({"response": self.response_object("in_progress", false, false)}), + )); + } + + // ── Text content ── + if has_content { + let delta = chunk.delta.content.clone().unwrap_or_default(); + if self.text_item_id.is_none() { + let item_id = format!("msg_{}", Uuid::new_v4().simple()); + let output_index = self.next_output_index; + self.next_output_index += 1; + self.text_item_id = Some(item_id.clone()); + self.text_output_index = output_index; + events.push(self.event( + "response.output_item.added", + json!({ + "output_index": output_index, + "item": {"type": "message", "id": item_id, "status": "in_progress", "role": "assistant", "content": []}, + }), + )); + events.push(self.event( + "response.content_part.added", + json!({ + "item_id": item_id, + "output_index": output_index, + "content_index": 0, + "part": {"type": "output_text", "text": "", "annotations": []}, + }), + )); + } + let item_id = self.text_item_id.clone().unwrap_or_default(); + let output_index = self.text_output_index; + self.text_accum.push_str(&delta); + events.push(self.event( + "response.output_text.delta", + json!({ + "item_id": item_id, + "output_index": output_index, + "content_index": 0, + "delta": delta, + }), + )); + } + + // ── Tool calls ── + if let Some(tool_calls) = chunk.delta.tool_calls.as_ref() { + for tc in tool_calls { + let oai_index = tc.get("index").and_then(|v| v.as_u64()).unwrap_or(0); + let id = tc.get("id").and_then(|v| v.as_str()).unwrap_or(""); + let name = tc + .get("function") + .and_then(|f| f.get("name")) + .and_then(|n| n.as_str()) + .unwrap_or(""); + let arguments = tc + .get("function") + .and_then(|f| f.get("arguments")) + .and_then(|a| a.as_str()) + .unwrap_or(""); + + if !self.tool_calls.contains_key(&oai_index) { + let output_index = self.next_output_index; + self.next_output_index += 1; + self.tool_calls.insert( + oai_index, + ToolCallState { + item_id: format!("fc_{}", Uuid::new_v4().simple()), + call_id: String::new(), + name: String::new(), + output_index, + arguments: String::new(), + item_added: false, + }, + ); + } + let state = self.tool_calls.get_mut(&oai_index).expect("just inserted"); + if !id.is_empty() { + state.call_id = id.to_string(); + } + if !name.is_empty() { + state.name = name.to_string(); + } + + // Emit output_item.added once the call id + name are known. + if !state.item_added && !state.call_id.is_empty() && !state.name.is_empty() { + state.item_added = true; + let (item_id, call_id, name, output_index) = ( + state.item_id.clone(), + state.call_id.clone(), + state.name.clone(), + state.output_index, + ); + events.push(self.event( + "response.output_item.added", + json!({ + "output_index": output_index, + "item": {"type": "function_call", "id": item_id, "call_id": call_id, "name": name, "arguments": "", "status": "in_progress"}, + }), + )); + } + + if !arguments.is_empty() { + let state = self.tool_calls.get_mut(&oai_index).expect("present"); + state.arguments.push_str(arguments); + if state.item_added { + let (item_id, output_index) = (state.item_id.clone(), state.output_index); + events.push(self.event( + "response.function_call_arguments.delta", + json!({ + "item_id": item_id, + "output_index": output_index, + "delta": arguments, + }), + )); + } + } + } + } + + // ── Finish ── + if let Some(fr) = chunk.finish_reason.as_ref() { + events.extend(self.close_items()); + let (status, reason) = responses_status(fr); + if self.usage_seen { + events.push(self.completed_event(status, reason)); + } else { + // Hold response.completed until the trailing usage frame. + self.pending_status = Some(status); + self.pending_reason = reason; + } + } + + events + } + + /// Emit the per-item `*.done` closing events for the open text + tool + /// items. Idempotent: a no-op after the first call, so the finish chunk + /// and a later `force_finish` (when the completed event was withheld for + /// usage) don't double-emit the done events. + fn close_items(&mut self) -> Vec { + if self.items_closed { + return Vec::new(); + } + self.items_closed = true; + let mut events = Vec::new(); + if self.text_item_id.is_some() { + let item_id = self.text_item_id.clone().unwrap_or_default(); + let output_index = self.text_output_index; + let text = self.text_accum.clone(); + events.push(self.event( + "response.output_text.done", + json!({ + "item_id": item_id, + "output_index": output_index, + "content_index": 0, + "text": text, + }), + )); + events.push(self.event( + "response.content_part.done", + json!({ + "item_id": item_id, + "output_index": output_index, + "content_index": 0, + "part": {"type": "output_text", "text": text, "annotations": []}, + }), + )); + events.push(self.event( + "response.output_item.done", + json!({ + "output_index": output_index, + "item": {"type": "message", "id": item_id, "status": "completed", "role": "assistant", "content": [{"type": "output_text", "text": text, "annotations": []}]}, + }), + )); + } + let pending: Vec = self + .tool_calls + .iter() + .filter(|(_, s)| s.item_added) + .map(|(k, _)| *k) + .collect(); + for k in pending { + let (item_id, call_id, name, arguments, output_index) = { + let s = self.tool_calls.get(&k).expect("present"); + ( + s.item_id.clone(), + s.call_id.clone(), + s.name.clone(), + s.arguments.clone(), + s.output_index, + ) + }; + events.push(self.event( + "response.function_call_arguments.done", + json!({ + "item_id": item_id, + "output_index": output_index, + "arguments": arguments, + }), + )); + events.push(self.event( + "response.output_item.done", + json!({ + "output_index": output_index, + "item": {"type": "function_call", "id": item_id, "call_id": call_id, "name": name, "arguments": arguments, "status": "completed"}, + }), + )); + } + events + } + + fn completed_event(&mut self, status: &str, reason: Option<&'static str>) -> ResponsesSseEvent { + self.finished = true; + let event_type = if status == "completed" { + "response.completed" + } else { + "response.incomplete" + }; + let mut response = self.response_object(status, true, true); + if let Some(reason) = reason { + response["incomplete_details"] = json!({"reason": reason}); + } + self.event(event_type, json!({"response": response})) + } + + pub fn is_finished(&self) -> bool { + self.finished + } + + /// Flush a clean close when the upstream stream ended without a finish + /// chunk, or while the completed event was withheld for usage. + pub fn force_finish(&mut self) -> Vec { + if self.finished { + return Vec::new(); + } + let mut events = Vec::new(); + // No renderable signal ever arrived → synthesize the preamble so + // the client still gets a well-formed (empty) response. + if !self.sent_created { + self.sent_created = true; + events.push(self.event( + "response.created", + json!({"response": self.response_object("in_progress", false, false)}), + )); + events.push(self.event( + "response.in_progress", + json!({"response": self.response_object("in_progress", false, false)}), + )); + } + let status = self.pending_status.take().unwrap_or("completed"); + let reason = self.pending_reason.take(); + events.extend(self.close_items()); + events.push(self.completed_event(status, reason)); + events + } +} + +/// End-of-stream telemetry captured by [`build_responses_bridge_stream`]. +#[derive(Default, Debug)] +pub struct ResponsesStreamCompletion { + pub prompt_tokens: u32, + pub completion_tokens: u32, + pub reasoning_tokens: u32, + pub cached_prompt_tokens: u32, + pub cache_creation_tokens: u32, + pub cache_read_tokens: u32, + pub finish_reason: String, + pub ttft_ms: u32, + /// Set when an output guardrail blocked the streamed response (a content + /// block or a fail-closed buffer overflow). The upstream still billed, so + /// the usage event carries the tokens but is marked blocked — matching + /// the non-streaming path so the dashboard's Blocked tab + budget ledger + /// see it. + pub guardrail_blocked: bool, +} + +struct CompleteOnDrop { + slot: Option<(F, ResponsesStreamCompletion)>, +} + +impl CompleteOnDrop { + fn comp(&mut self) -> &mut ResponsesStreamCompletion { + &mut self + .slot + .as_mut() + .expect("stream completion guard accessed after drop") + .1 + } +} + +impl Drop for CompleteOnDrop { + fn drop(&mut self) { + if let Some((f, comp)) = self.slot.take() { + f(comp); + } + } +} + +/// Wrap a bridge [`ChatChunkStream`] as a Responses-API SSE body, encoding +/// each chunk via [`ResponsesSseEncoder`]. An end-of-stream telemetry +/// callback fires from a Drop guard (so it runs on normal end and on client +/// disconnect). +/// +/// When `output_guardrail` is `Some`, the encoded SSE is **held back** and +/// released only after the assembled assistant output passes the scan — +/// mirroring the verbatim `/v1/responses` path's secure BufferFull default +/// (#719), so a configured output block can't be bypassed by streaming a +/// non-OpenAI model. The scan reads the fully-reassembled text + tool calls +/// (not raw deltas), and the buffer is capped — an output guardrail must +/// never release content it couldn't fully buffer to scan, so an overflow +/// fails closed. With no output guardrail the bytes forward live. +pub fn build_responses_bridge_stream( + upstream: ChatChunkStream, + encoder: ResponsesSseEncoder, + started: Instant, + output_guardrail: Option>, + max_buffer_bytes: usize, + model_label: String, + on_complete: impl FnOnce(ResponsesStreamCompletion) + Send + 'static, +) -> axum::body::Body { + use futures::StreamExt; + + let mut encoder = encoder; + let stream = async_stream::stream! { + let mut guard = CompleteOnDrop { slot: Some((on_complete, ResponsesStreamCompletion::default())) }; + let mut upstream = upstream; + let mut first_chunk_seen = false; + let buffering = output_guardrail.is_some(); + // Held SSE events when an output guardrail is attached; empty (and + // unused) on the live-forward path. + let mut held: Vec = Vec::new(); + let mut held_bytes = 0usize; + let mut overflowed = false; + while let Some(item) = upstream.next().await { + match item { + Ok(chunk) => { + if !first_chunk_seen + && (chunk.delta.content.is_some() || chunk.delta.tool_calls.is_some()) + { + first_chunk_seen = true; + guard.comp().ttft_ms = + started.elapsed().as_millis().min(u32::MAX as u128) as u32; + } + { + let comp = guard.comp(); + if let Some(fr) = chunk.finish_reason.as_ref() { + comp.finish_reason = finish_reason_label(fr); + } + if let Some(u) = chunk.usage.as_ref() { + comp.prompt_tokens = comp.prompt_tokens.max(u.prompt_tokens); + comp.completion_tokens = comp.completion_tokens.max(u.completion_tokens); + comp.reasoning_tokens = comp.reasoning_tokens.max(u.reasoning_tokens); + comp.cached_prompt_tokens = comp.cached_prompt_tokens.max(u.cached_prompt_tokens); + comp.cache_creation_tokens = comp.cache_creation_tokens.max(u.cache_creation_tokens); + comp.cache_read_tokens = comp.cache_read_tokens.max(u.cache_read_tokens); + } + } + for ev in encoder.next_events(&chunk) { + let b = bytes::Bytes::from(ev.to_sse_string()); + if buffering { + held_bytes += b.len(); + if held_bytes > max_buffer_bytes { + overflowed = true; + break; + } + held.push(b); + } else { + yield Ok::<_, std::io::Error>(b); + } + } + if overflowed || encoder.is_finished() { + break; + } + } + Err(e) => { + let frame = format!( + "event: error\ndata: {{\"type\":\"error\",\"code\":\"{}\",\"message\":{}}}\n\n", + e.error_type(), + serde_json::to_string(&e.to_string()).unwrap_or_else(|_| "\"error\"".into()), + ); + yield Ok(bytes::Bytes::from(frame)); + return; + } + } + } + if !encoder.is_finished() { + for ev in encoder.force_finish() { + let b = bytes::Bytes::from(ev.to_sse_string()); + if buffering { + held_bytes += b.len(); + if held_bytes > max_buffer_bytes { + overflowed = true; + break; + } + held.push(b); + } else { + yield Ok(b); + } + } + } + + // Live-forward path: nothing held, nothing to scan. + let Some(chain) = output_guardrail.as_ref() else { return; }; + + // Buffer overflow: an output guardrail must not release content it + // couldn't fully buffer to scan — fail closed (#719). + if overflowed { + tracing::warn!( + guardrail_hook = "output", + model = %model_label, + max_buffer_bytes, + "streaming /v1/responses (cross-provider) output exceeded buffer cap; failing closed", + ); + guard.comp().guardrail_blocked = true; + yield Ok(bytes::Bytes::from(guardrail_error_frame(None))); + return; + } + + // End-of-stream output guardrail (#719): scan the fully-reassembled + // assistant output (canonical tool calls, so a literal split across + // argument deltas can't slip through), then release or block. + let (text, tool_calls) = encoder.assembled_assistant_message(); + if !text.is_empty() || !tool_calls.is_empty() { + let mut message = aisix_gateway::ChatMessage::assistant(text); + if !tool_calls.is_empty() { + message.extra.insert("tool_calls".to_string(), Value::Array(tool_calls)); + } + let synth = ChatResponse { + id: String::new(), + model: model_label.clone(), + message, + finish_reason: FinishReason::Stop, + usage: UsageStats::new(0, 0), + }; + if let aisix_guardrails::GuardrailVerdict::Block { reason, guardrail_name } = + aisix_guardrails::Guardrail::check_output(chain.as_ref(), &synth).await + { + tracing::warn!( + guardrail_hook = "output", + model = %model_label, + reason = %reason, + "guardrail blocked streaming /v1/responses (cross-provider) response", + ); + guard.comp().guardrail_blocked = true; + yield Ok(bytes::Bytes::from(guardrail_error_frame(guardrail_name.as_deref()))); + return; + } + } + // Passed: release the held events verbatim. + for b in held { + yield Ok(b); + } + }; + axum::body::Body::from_stream(stream) +} + +/// Responses-API SSE `error` frame for an output-guardrail block. Carries the +/// firing guardrail's name (#519 B.4b) but never the matched-pattern detail. +fn guardrail_error_frame(guardrail_name: Option<&str>) -> String { + format!( + "event: error\ndata: {}\n\n", + json!({ + "type": "error", + "code": "content_filter", + "message": crate::error::guardrail_block_message("response", guardrail_name), + }) + ) +} + +fn finish_reason_label(reason: &FinishReason) -> String { + match reason { + FinishReason::Stop => "stop".into(), + FinishReason::Length => "length".into(), + FinishReason::ContentFilter => "content_filter".into(), + FinishReason::ToolCalls => "tool_calls".into(), + FinishReason::Other(s) => s.clone(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aisix_gateway::{ChatDelta, Role}; + + // ── Request translation ────────────────────────────────────── + + #[test] + fn instructions_become_system_and_input_string_becomes_user() { + let body = json!({ + "model": "opus-4.7", + "instructions": "be terse", + "input": "hi", + }); + let chat = responses_request_to_chat("opus-4.7", &body); + assert_eq!(chat.messages.len(), 2); + assert!(matches!(chat.messages[0].role, Role::System)); + assert_eq!(chat.messages[0].content_str(), "be terse"); + assert!(matches!(chat.messages[1].role, Role::User)); + assert_eq!(chat.messages[1].content_str(), "hi"); + } + + #[test] + fn input_array_messages_preserve_roles_and_text_parts() { + let body = json!({ + "model": "m", + "input": [ + {"role": "user", "content": [{"type": "input_text", "text": "part1"}, {"type": "input_text", "text": "part2"}]}, + {"role": "assistant", "content": "ok"}, + ], + }); + let chat = responses_request_to_chat("m", &body); + assert_eq!(chat.messages.len(), 2); + assert!(matches!(chat.messages[0].role, Role::User)); + assert_eq!(chat.messages[0].content_str(), "part1part2"); + assert!(matches!(chat.messages[1].role, Role::Assistant)); + } + + #[test] + fn function_call_and_output_become_assistant_tool_calls_and_tool_turn() { + // The codex agent-loop history shape. + let body = json!({ + "model": "m", + "input": [ + {"role": "user", "content": "run ls"}, + {"type": "function_call", "call_id": "call_1", "name": "shell", "arguments": "{\"cmd\":\"ls\"}"}, + {"type": "function_call_output", "call_id": "call_1", "output": "a.txt"}, + ], + }); + let chat = responses_request_to_chat("m", &body); + assert_eq!(chat.messages.len(), 3); + assert!(matches!(chat.messages[1].role, Role::Assistant)); + let tcs = chat.messages[1] + .extra + .get("tool_calls") + .unwrap() + .as_array() + .unwrap(); + assert_eq!(tcs.len(), 1); + assert_eq!(tcs[0]["id"], "call_1"); + assert_eq!(tcs[0]["function"]["name"], "shell"); + assert!(matches!(chat.messages[2].role, Role::Tool)); + assert_eq!(chat.messages[2].tool_call_id.as_deref(), Some("call_1")); + assert_eq!(chat.messages[2].content_str(), "a.txt"); + } + + #[test] + fn parallel_function_calls_fold_into_one_assistant_message() { + let body = json!({ + "model": "m", + "input": [ + {"type": "function_call", "call_id": "c1", "name": "a", "arguments": "{}"}, + {"type": "function_call", "call_id": "c2", "name": "b", "arguments": "{}"}, + ], + }); + let chat = responses_request_to_chat("m", &body); + assert_eq!(chat.messages.len(), 1); + let tcs = chat.messages[0] + .extra + .get("tool_calls") + .unwrap() + .as_array() + .unwrap(); + assert_eq!(tcs.len(), 2); + } + + #[test] + fn tools_and_params_translate_to_chat_shape() { + let body = json!({ + "model": "m", + "input": "hi", + "max_output_tokens": 256, + "temperature": 0.5, + "stream": true, + "tools": [{"type": "function", "name": "get_weather", "description": "d", "parameters": {"type": "object"}}], + "tool_choice": {"type": "function", "name": "get_weather"}, + // OpenAI-only knobs must NOT leak into chat.extra (they'd 400 Anthropic). + "reasoning": {"effort": "high"}, + "store": false, + }); + let chat = responses_request_to_chat("m", &body); + assert_eq!(chat.max_tokens, Some(256)); + assert_eq!(chat.temperature, Some(0.5)); + assert_eq!(chat.stream, Some(true)); + let tools = chat.extra.get("tools").unwrap().as_array().unwrap(); + assert_eq!(tools[0]["type"], "function"); + assert_eq!(tools[0]["function"]["name"], "get_weather"); + assert_eq!( + chat.extra.get("tool_choice").unwrap()["function"]["name"], + "get_weather" + ); + assert!(!chat.extra.contains_key("reasoning")); + assert!(!chat.extra.contains_key("store")); + } + + #[test] + fn out_of_range_max_output_tokens_is_ignored_not_truncated() { + // A value above u32::MAX must not wrap to a small/zero cap. + let body = json!({"model": "m", "input": "hi", "max_output_tokens": 10_000_000_000u64}); + let chat = responses_request_to_chat("m", &body); + assert_eq!(chat.max_tokens, None); + } + + // ── Non-streaming response translation ─────────────────────── + + fn chat_response_with( + text: Option<&str>, + tool_calls: Option, + fr: FinishReason, + ) -> ChatResponse { + let mut extra = Map::new(); + if let Some(tc) = tool_calls { + extra.insert("tool_calls".into(), tc); + } + ChatResponse { + id: "id".into(), + model: "m".into(), + message: ChatMessage { + role: Role::Assistant, + content: text.map(|s| s.to_string()), + content_blocks: None, + name: None, + tool_call_id: None, + extra, + }, + finish_reason: fr, + usage: UsageStats::new(11, 7), + } + } + + #[test] + fn non_streaming_text_response_builds_message_output_and_usage() { + let resp = chat_response_with(Some("hello"), None, FinishReason::Stop); + let out = chat_response_to_responses_json(&resp, "opus-4.7", 100); + assert_eq!(out["object"], "response"); + assert_eq!(out["status"], "completed"); + assert_eq!(out["model"], "opus-4.7"); + let item = &out["output"][0]; + assert_eq!(item["type"], "message"); + assert_eq!(item["content"][0]["type"], "output_text"); + assert_eq!(item["content"][0]["text"], "hello"); + assert_eq!(out["usage"]["input_tokens"], 11); + assert_eq!(out["usage"]["output_tokens"], 7); + assert_eq!(out["usage"]["total_tokens"], 18); + } + + #[test] + fn non_streaming_tool_call_response_builds_function_call_item() { + let tcs = json!([{"id": "call_9", "type": "function", "function": {"name": "shell", "arguments": "{\"cmd\":\"ls\"}"}}]); + let resp = chat_response_with(None, Some(tcs), FinishReason::ToolCalls); + let out = chat_response_to_responses_json(&resp, "m", 1); + let item = &out["output"][0]; + assert_eq!(item["type"], "function_call"); + assert_eq!(item["call_id"], "call_9"); + assert_eq!(item["name"], "shell"); + assert_eq!(item["arguments"], "{\"cmd\":\"ls\"}"); + } + + #[test] + fn length_finish_maps_to_incomplete_status() { + let resp = chat_response_with(Some("x"), None, FinishReason::Length); + let out = chat_response_to_responses_json(&resp, "m", 1); + assert_eq!(out["status"], "incomplete"); + assert_eq!(out["incomplete_details"]["reason"], "max_output_tokens"); + } + + // ── Streaming encoder ──────────────────────────────────────── + + fn content_chunk(text: &str) -> ChatChunk { + ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta { + content: Some(text.into()), + ..Default::default() + }, + finish_reason: None, + usage: None, + } + } + + fn types_of(events: &[ResponsesSseEvent]) -> Vec<&'static str> { + events.iter().map(|e| e.event_type).collect() + } + + #[test] + fn streaming_text_emits_canonical_event_sequence() { + let mut enc = ResponsesSseEncoder::new("resp_1", "opus-4.7", 0); + let mut all: Vec = Vec::new(); + all.extend(enc.next_events(&content_chunk("Hel"))); + all.extend(enc.next_events(&content_chunk("lo"))); + // Finish chunk carrying usage (Anthropic attaches it here). + all.extend(enc.next_events(&ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta::default(), + finish_reason: Some(FinishReason::Stop), + usage: Some(UsageStats::new(5, 2)), + })); + let types = types_of(&all); + assert_eq!( + types, + vec![ + "response.created", + "response.in_progress", + "response.output_item.added", + "response.content_part.added", + "response.output_text.delta", + "response.output_text.delta", + "response.output_text.done", + "response.content_part.done", + "response.output_item.done", + "response.completed", + ] + ); + assert!(enc.is_finished()); + let completed = all.last().unwrap(); + assert_eq!( + completed.data["response"]["output"][0]["content"][0]["text"], + "Hello" + ); + assert_eq!(completed.data["response"]["usage"]["input_tokens"], 5); + assert_eq!(completed.data["response"]["usage"]["output_tokens"], 2); + // sequence_number is monotonic from 0. + assert_eq!(all[0].data["sequence_number"], 0); + assert_eq!(all[1].data["sequence_number"], 1); + } + + #[test] + fn streaming_completed_withheld_until_trailing_usage_frame() { + // OpenAI-compat upstreams send usage AFTER the finish chunk. + let mut enc = ResponsesSseEncoder::new("resp_1", "m", 0); + let _ = enc.next_events(&content_chunk("hi")); + // Finish without usage → close items but NOT completed yet. + let at_finish = enc.next_events(&ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta::default(), + finish_reason: Some(FinishReason::Stop), + usage: None, + }); + assert!(!types_of(&at_finish).contains(&"response.completed")); + assert!(!enc.is_finished()); + // Trailing usage frame releases completed. + let usage_frame = enc.next_events(&ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta::default(), + finish_reason: None, + usage: Some(UsageStats::new(3, 4)), + }); + assert_eq!(types_of(&usage_frame), vec!["response.completed"]); + assert_eq!(usage_frame[0].data["response"]["usage"]["output_tokens"], 4); + } + + #[test] + fn streaming_tool_call_emits_function_call_events() { + let mut enc = ResponsesSseEncoder::new("resp_1", "m", 0); + let chunk = ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta { + tool_calls: Some(vec![json!({ + "index": 0, "id": "call_1", "type": "function", + "function": {"name": "shell", "arguments": "{\"cmd\""}, + })]), + ..Default::default() + }, + finish_reason: None, + usage: None, + }; + let mut all = enc.next_events(&chunk); + all.extend(enc.next_events(&ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta { + tool_calls: Some(vec![ + json!({"index": 0, "function": {"arguments": ":\"ls\"}"}}), + ]), + ..Default::default() + }, + finish_reason: None, + usage: None, + })); + all.extend(enc.next_events(&ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta::default(), + finish_reason: Some(FinishReason::ToolCalls), + usage: Some(UsageStats::new(4, 6)), + })); + let types = types_of(&all); + assert!(types.contains(&"response.output_item.added")); + assert!(types.contains(&"response.function_call_arguments.delta")); + assert!(types.contains(&"response.function_call_arguments.done")); + assert_eq!(*types.last().unwrap(), "response.completed"); + let completed = all.last().unwrap(); + let item = &completed.data["response"]["output"][0]; + assert_eq!(item["type"], "function_call"); + assert_eq!(item["call_id"], "call_1"); + assert_eq!(item["arguments"], "{\"cmd\":\"ls\"}"); + } + + #[test] + fn force_finish_on_empty_stream_emits_well_formed_completed() { + let mut enc = ResponsesSseEncoder::new("resp_1", "m", 0); + let events = enc.force_finish(); + let types = types_of(&events); + assert_eq!( + types, + vec![ + "response.created", + "response.in_progress", + "response.completed" + ] + ); + assert!(enc.is_finished()); + } + + #[test] + fn tool_call_finish_without_usage_then_force_finish_does_not_double_close() { + // Finish chunk lacks usage → done events emitted, completed withheld. + // force_finish must NOT re-emit the per-item done events. + let mut enc = ResponsesSseEncoder::new("resp_1", "m", 0); + let _ = enc.next_events(&ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta { + tool_calls: Some(vec![json!({ + "index": 0, "id": "call_1", "type": "function", + "function": {"name": "shell", "arguments": "{}"}, + })]), + ..Default::default() + }, + finish_reason: None, + usage: None, + }); + let at_finish = enc.next_events(&ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta::default(), + finish_reason: Some(FinishReason::ToolCalls), + usage: None, + }); + assert_eq!( + types_of(&at_finish) + .iter() + .filter(|t| **t == "response.output_item.done") + .count(), + 1 + ); + assert!(!enc.is_finished()); + let tail = enc.force_finish(); + // The trailing close emits only response.completed, not a second + // round of done events. + assert_eq!(types_of(&tail), vec!["response.completed"]); + } + + #[test] + fn streaming_preserves_provider_total_tokens() { + let mut enc = ResponsesSseEncoder::new("resp_1", "m", 0); + let _ = enc.next_events(&content_chunk("hi")); + let done = enc.next_events(&ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta::default(), + finish_reason: Some(FinishReason::Stop), + // Provider reports an authoritative total that differs from + // prompt+completion (e.g. it counts tool/system overhead). + usage: Some(UsageStats { + prompt_tokens: 5, + completion_tokens: 2, + total_tokens: 11, + ..Default::default() + }), + }); + let completed = done.last().unwrap(); + assert_eq!(completed.data["response"]["usage"]["total_tokens"], 11); + } + + #[test] + fn streaming_length_finish_emits_incomplete_with_reason() { + let mut enc = ResponsesSseEncoder::new("resp_1", "m", 0); + let _ = enc.next_events(&content_chunk("partial")); + let done = enc.next_events(&ChatChunk { + id: "c".into(), + model: "m".into(), + delta: ChatDelta::default(), + finish_reason: Some(FinishReason::Length), + usage: Some(UsageStats::new(3, 9)), + }); + let completed = done.last().unwrap(); + assert_eq!(completed.event_type, "response.incomplete"); + assert_eq!(completed.data["response"]["status"], "incomplete"); + assert_eq!( + completed.data["response"]["incomplete_details"]["reason"], + "max_output_tokens" + ); + } +} diff --git a/docs/integration/responses.md b/docs/integration/responses.md index ff76bc7e..b055b4f0 100644 --- a/docs/integration/responses.md +++ b/docs/integration/responses.md @@ -1,32 +1,50 @@ --- title: Responses API -description: Learn how AISIX AI Gateway handles the OpenAI Responses API and its current provider boundary. +description: Learn how AISIX AI Gateway handles the OpenAI Responses API across OpenAI and non-OpenAI providers. sidebar_position: 25 --- AISIX AI Gateway exposes `POST /v1/responses` as a proxy for the OpenAI Responses API. -Use this endpoint only when you specifically want the Responses API surface rather than chat completions. +Use this endpoint when your application (or a tool such as the OpenAI Codex CLI) speaks the Responses API surface rather than chat completions. It works regardless of which provider backs the resolved model. -## Current Provider Boundary +## Provider Support -This endpoint is currently available only for models whose configured provider is `openai`. +`/v1/responses` works for **any** configured provider: -If the resolved model points to any non-OpenAI provider, the gateway returns `400`. +- **OpenAI models** — the request body is forwarded verbatim to the upstream's own `/v1/responses` endpoint (a thin proxy), so every Responses feature the upstream supports passes through unchanged. +- **Non-OpenAI models** (Anthropic, Gemini, DeepSeek, …) — the gateway **bridges** the request: it translates the Responses payload into its internal chat format, dispatches through the same provider adapter `/v1/chat/completions` uses, and re-encodes the reply back into the Responses shape (non-streaming JSON and streaming SSE alike). This is what lets a Responses-only client such as Codex point at an Anthropic model. -This is a stricter provider boundary than `/v1/chat/completions`. +The bridge is the same machinery `/v1/chat/completions` and `/v1/messages` use for cross-provider translation, so behavior (tool calling, failover within a Model Group, usage accounting) stays consistent across surfaces. ## Gateway Behavior -For supported models, the gateway: +For every request the gateway authenticates and authorizes the caller key, resolves the model alias, runs the configured input guardrails, then: -1. authenticates and authorizes the caller key -2. verifies the model is an OpenAI provider -3. rewrites `model` to the upstream provider model id -4. forwards the request body to the upstream `/v1/responses` endpoint -5. returns JSON or streaming SSE depending on the request +**OpenAI provider (verbatim passthrough)** -The gateway is acting as a thin proxy here rather than a cross-provider compatibility layer. +1. rewrites `model` to the upstream provider model id +2. forwards the body to the upstream `/v1/responses` endpoint +3. returns JSON or streaming SSE depending on the request + +**Non-OpenAI provider (cross-provider bridge)** + +1. translates the Responses request (`instructions`, `input`, `tools`, `tool_choice`, `temperature`, `top_p`, `max_output_tokens`, `stream`) into the internal chat format +2. dispatches through the provider adapter (e.g. Anthropic Messages) +3. re-encodes the response into the Responses shape — a `message` output item for assistant text and a `function_call` output item per tool call, plus the streaming event sequence (`response.created` → `response.output_item.added` → `response.output_text.delta` / `response.function_call_arguments.delta` → `response.completed`) + +Multi-turn agent loops work across providers: `function_call` and `function_call_output` items in `input` are translated into the upstream's tool-use / tool-result turns. + +### What the bridge does not carry + +Responses fields that have no cross-provider equivalent are dropped rather than forwarded (forwarding an OpenAI-only field would make a provider like Anthropic reject the request): + +- `reasoning` (effort/summary) — extended-thinking is not mapped to a backend today +- `store`, `previous_response_id` — the gateway is stateless; replay the full `input` each turn +- hosted tools (`web_search`, `file_search`, `code_interpreter`, …) — only `type: "function"` tools translate +- `text`/`metadata`/`service_tier` and other OpenAI-only knobs + +These limitations apply only to the non-OpenAI bridge path; OpenAI models forward the body verbatim. ## Usage Accounting @@ -46,16 +64,35 @@ curl -sS -X POST http://127.0.0.1:3000/v1/responses \ }' ``` +### Point Codex at a non-OpenAI model + +```bash title="Codex (or any Responses client) against an Anthropic-backed alias" +curl -sS -X POST http://127.0.0.1:3000/v1/responses \ + -H "Authorization: Bearer YOUR_CALLER_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "model": "claude-prod", + "input": "Say hello from AISIX.", + "stream": true + }' +``` + +The gateway bridges this to the Anthropic Messages API and streams back the Responses SSE event sequence. + ## When To Use Responses Instead Of Chat Completions -- use `/v1/responses` when your application is already standardized on that OpenAI API surface -- use `/v1/chat/completions` when you want the broadest current compatibility across provider-backed models +- use `/v1/responses` when your application or tool is standardized on that OpenAI API surface (for example the Codex CLI) +- use `/v1/chat/completions` when you want the broadest feature coverage; the Responses bridge carries the common path (text, tool calls, streaming) but drops OpenAI-only knobs listed above ## Troubleshooting -### The same alias works for chat completions but not for responses +### Tool calls aren't replayed correctly across turns + +Send the full conversation in `input` each turn (the gateway is stateless): include the assistant's prior `function_call` items and the matching `function_call_output` items. The gateway translates them into the backend's tool-use / tool-result turns. + +### `reasoning` has no effect on a non-OpenAI model -That usually means the alias resolves to a non-OpenAI provider. +Extended-thinking config isn't bridged today (see [What the bridge does not carry](#what-the-bridge-does-not-carry)). The request still succeeds; the field is ignored. ## Related Pages diff --git a/tests/e2e/src/cases/responses-cross-provider-e2e.test.ts b/tests/e2e/src/cases/responses-cross-provider-e2e.test.ts new file mode 100644 index 00000000..145ccda4 --- /dev/null +++ b/tests/e2e/src/cases/responses-cross-provider-e2e.test.ts @@ -0,0 +1,388 @@ +import { createHash } from "node:crypto"; +import { createServer, type Server } from "node:http"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + AdminClient, + EtcdClient, + pickFreePort, + spawnApp, + startOpenAiUpstream, + waitConfigPropagation, + type OpenAiUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// E2E for AISIX-Cloud#825: the OpenAI Responses API (`POST /v1/responses`) +// must work against a non-OpenAI backend. The `codex` CLI speaks only the +// Responses API; pointing it at an Anthropic model (e.g. `opus-4.7`) used to +// return a hard 400 ("model ... is not an OpenAI provider; /v1/responses +// requires OpenAI"). The gateway now bridges the Responses request through +// the canonical ChatFormat to the Anthropic Messages upstream and re-encodes +// the reply back into the Responses-API shape — non-streaming JSON and +// streaming SSE — and translates the agent-loop tool history so multi-turn +// tool calls round-trip. + +const CALLER_PLAINTEXT = "sk-issue-825-xprovider"; +const CALLER_KEY_HASH = createHash("sha256") + .update(CALLER_PLAINTEXT) + .digest("hex"); + +// Anthropic Messages non-streaming reply (text). +const ANTHROPIC_NONSTREAM = { + id: "msg_xprov_ns", + type: "message", + role: "assistant", + model: "claude-3-haiku-20240307", + content: [{ type: "text", text: "Hello from Claude" }], + stop_reason: "end_turn", + usage: { input_tokens: 11, output_tokens: 7 }, +}; + +const STREAM_INPUT_TOKENS = 12; +const STREAM_OUTPUT_TOKENS = 5; + +// Anthropic Messages streaming wire shape (data-only; the mock writes the +// `data:` line). The DP parses by the JSON `type`. +const ANTHROPIC_STREAM_EVENTS = [ + JSON.stringify({ + type: "message_start", + message: { + id: "msg_xprov_s", + type: "message", + role: "assistant", + model: "claude-3-haiku-20240307", + content: [], + usage: { input_tokens: STREAM_INPUT_TOKENS, output_tokens: 0 }, + }, + }), + JSON.stringify({ + type: "content_block_start", + index: 0, + content_block: { type: "text", text: "" }, + }), + JSON.stringify({ + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "Hi from Claude" }, + }), + JSON.stringify({ type: "content_block_stop", index: 0 }), + JSON.stringify({ + type: "message_delta", + delta: { stop_reason: "end_turn" }, + usage: { output_tokens: STREAM_OUTPUT_TOKENS }, + }), + JSON.stringify({ type: "message_stop" }), +]; + +interface OtlpReceiver { + url: string; + spanAttrs: Array>; + close(): Promise; +} + +async function startOtlpReceiver(): Promise { + const spanAttrs: Array> = []; + const server: Server = createServer((req, res) => { + let raw = ""; + req.on("data", (c: Buffer) => (raw += c.toString("utf8"))); + req.on("end", () => { + try { + const body = JSON.parse(raw); + for (const rs of body.resourceSpans ?? []) { + for (const ss of rs.scopeSpans ?? []) { + for (const span of ss.spans ?? []) { + const attrs: Record = {}; + for (const a of span.attributes ?? []) { + const v = a.value ?? {}; + attrs[a.key] = + v.stringValue ?? String(v.intValue ?? v.boolValue ?? ""); + } + spanAttrs.push(attrs); + } + } + } + } catch (err) { + // Surface malformed OTLP so a decode bug is obvious rather than only + // showing up later as a missing-span assertion failure. + console.error("OTLP receiver failed to parse body:", err); + } + res.statusCode = 200; + res.end("{}"); + }); + }); + const port = await pickFreePort(); + await new Promise((resolve) => server.listen(port, "127.0.0.1", resolve)); + return { + url: `http://127.0.0.1:${port}/v1/traces`, + spanAttrs, + async close() { + await new Promise((resolve, reject) => { + server.close((err) => (err ? reject(err) : resolve())); + }); + }, + }; +} + +async function collectUsageSpans( + recv: OtlpReceiver, + requestId: string, + timeoutMs = 10_000, +): Promise>> { + const matches = () => + recv.spanAttrs.filter( + (a) => + a["aisix.request_id"] === requestId && + a["gen_ai.usage.input_tokens"] !== undefined, + ); + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (matches().length > 0) break; + await new Promise((r) => setTimeout(r, 50)); + } + if (matches().length === 0) { + throw new Error(`no usage span for request_id=${requestId}`); + } + await new Promise((r) => setTimeout(r, 300)); + return matches(); +} + +function post(app: SpawnedApp, body: unknown): Promise { + return fetch(`${app.proxyUrl}/v1/responses`, { + method: "POST", + headers: { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + "user-agent": "codex_cli_rs/0.5.0", + }, + body: JSON.stringify(body), + }); +} + +describe("/v1/responses cross-provider → Anthropic (#825)", () => { + let app: SpawnedApp | undefined; + let nsUpstream: OpenAiUpstream | undefined; + let streamUpstream: OpenAiUpstream | undefined; + let otlp: OtlpReceiver | undefined; + let etcdReachable = false; + + beforeAll(async () => { + etcdReachable = await new EtcdClient().ping(); + if (!etcdReachable) return; + + nsUpstream = await startOpenAiUpstream({ nonStreamBody: ANTHROPIC_NONSTREAM }); + streamUpstream = await startOpenAiUpstream({ + streamEvents: ANTHROPIC_STREAM_EVENTS, + eventDelayMs: 2, + }); + app = await spawnApp(); + const admin = new AdminClient(app.adminUrl, app.adminKey); + otlp = await startOtlpReceiver(); + await admin.createObservabilityExporter({ + name: "issue825-otlp", + enabled: true, + kind: "otlp_http", + endpoint: otlp.url, + }); + + // Two Anthropic-backed models, one per mock (non-streaming vs streaming); + // api_base is the bare host — the bridge composes `/v1/messages`. + const nsPk = await admin.createProviderKey({ + display_name: "issue825-ns-pk", + provider: "anthropic", + adapter: "anthropic", + secret: "sk-ant-mock", + api_base: nsUpstream.baseUrl, + }); + await admin.createModel({ + display_name: "opus-4.7", + provider: "anthropic", + model_name: "claude-3-haiku-20240307", + provider_key_id: nsPk.id, + }); + const streamPk = await admin.createProviderKey({ + display_name: "issue825-stream-pk", + provider: "anthropic", + adapter: "anthropic", + secret: "sk-ant-mock", + api_base: streamUpstream.baseUrl, + }); + await admin.createModel({ + display_name: "opus-4.7-stream", + provider: "anthropic", + model_name: "claude-3-haiku-20240307", + provider_key_id: streamPk.id, + }); + await admin.createApiKey({ + key_hash: CALLER_KEY_HASH, + allowed_models: ["opus-4.7", "opus-4.7-stream"], + }); + }); + + afterAll(async () => { + await app?.exit(); + await nsUpstream?.close(); + await streamUpstream?.close(); + await otlp?.close(); + }); + + test("non-streaming: Anthropic reply is re-encoded into the Responses shape + usage event", async (ctx) => { + if (!etcdReachable || !app || !nsUpstream || !otlp) { + ctx.skip(); + return; + } + await waitConfigPropagation(async () => { + try { + const r = await post(app!, { model: "opus-4.7", input: "ready" }); + return r.status === 200 && (await r.json()).object === "response"; + } catch { + return false; + } + }); + + const res = await post(app, { model: "opus-4.7", input: "hi" }); + expect(res.status).toBe(200); + const requestId = res.headers.get("x-aisix-request-id"); + expect(requestId).toBeTruthy(); + + const body = await res.json(); + expect(body.object).toBe("response"); + expect(body.status).toBe("completed"); + // Operator-facing model name, not the upstream id. + expect(body.model).toBe("opus-4.7"); + expect(body.output[0].type).toBe("message"); + expect(body.output[0].content[0].type).toBe("output_text"); + expect(body.output[0].content[0].text).toBe("Hello from Claude"); + expect(body.usage.input_tokens).toBe(11); + expect(body.usage.output_tokens).toBe(7); + + // The gateway spoke the Anthropic Messages protocol upstream. + const last = nsUpstream.receivedRequests.at(-1); + expect(last?.path).toBe("/v1/messages"); + + const spans = await collectUsageSpans(otlp, requestId!); + expect(spans).toHaveLength(1); + expect(spans[0]["gen_ai.usage.input_tokens"]).toBe("11"); + expect(spans[0]["gen_ai.usage.output_tokens"]).toBe("7"); + expect(spans[0]["http.response.status_code"]).toBe("200"); + }); + + test("streaming: codex-style streamed call yields Responses SSE events + usage event", async (ctx) => { + if (!etcdReachable || !app || !streamUpstream || !otlp) { + ctx.skip(); + return; + } + await waitConfigPropagation(async () => { + try { + const r = await post(app!, { + model: "opus-4.7-stream", + input: "ready", + stream: true, + }); + const t = await r.text(); + return r.status === 200 && t.includes("response.completed"); + } catch { + return false; + } + }); + + const res = await post(app, { + model: "opus-4.7-stream", + input: "hi", + stream: true, + }); + expect(res.status).toBe(200); + const requestId = res.headers.get("x-aisix-request-id"); + const body = await res.text(); + // Canonical Responses streaming event set, ending in response.completed. + expect(body).toContain("event: response.created"); + expect(body).toContain("event: response.output_item.added"); + expect(body).toContain("event: response.output_text.delta"); + expect(body).toContain('"delta":"Hi from Claude"'); + expect(body).toContain("event: response.completed"); + + const spans = await collectUsageSpans(otlp, requestId!); + expect(spans).toHaveLength(1); + expect(spans[0]["gen_ai.usage.input_tokens"]).toBe(String(STREAM_INPUT_TOKENS)); + expect(spans[0]["gen_ai.usage.output_tokens"]).toBe(String(STREAM_OUTPUT_TOKENS)); + }); + + test("multi-turn tool loop: function_call history is sent to Anthropic with alternating roles + tool_use/tool_result", async (ctx) => { + if (!etcdReachable || !app || !nsUpstream) { + ctx.skip(); + return; + } + await waitConfigPropagation(async () => { + try { + const r = await post(app!, { model: "opus-4.7", input: "ready" }); + return r.status === 200; + } catch { + return false; + } + }); + + const baseline = nsUpstream.receivedRequests.length; + const res = await post(app, { + model: "opus-4.7", + // The codex agent-loop history shape: a user turn, the assistant's + // prior tool call, and the tool result fed back. + input: [ + { role: "user", content: "run ls" }, + { + type: "function_call", + call_id: "call_1", + name: "shell", + arguments: '{"cmd":"ls"}', + }, + { type: "function_call_output", call_id: "call_1", output: "a.txt" }, + ], + tools: [ + { + type: "function", + name: "shell", + description: "run a shell command", + parameters: { + type: "object", + properties: { cmd: { type: "string" } }, + required: ["cmd"], + }, + }, + ], + }); + expect(res.status).toBe(200); + + // Inspect what the gateway actually sent to the Anthropic upstream. + const sent = nsUpstream.receivedRequests + .slice(baseline) + .find((r) => r.path === "/v1/messages"); + expect(sent).toBeDefined(); + const anthropicReq = JSON.parse(sent!.body); + + // Roles strictly alternate user → assistant → user. + const roles = anthropicReq.messages.map((m: { role: string }) => m.role); + expect(roles).toEqual(["user", "assistant", "user"]); + + // The assistant turn carries the tool_use translated from function_call. + const toolUse = anthropicReq.messages[1].content.find( + (b: { type: string }) => b.type === "tool_use", + ); + expect(toolUse).toBeDefined(); + expect(toolUse.id).toBe("call_1"); + expect(toolUse.name).toBe("shell"); + expect(toolUse.input).toEqual({ cmd: "ls" }); + + // The tool result alternates back as a user tool_result block. + const toolResult = anthropicReq.messages[2].content.find( + (b: { type: string }) => b.type === "tool_result", + ); + expect(toolResult).toBeDefined(); + expect(toolResult.tool_use_id).toBe("call_1"); + + // Tools were translated to the Anthropic input_schema shape. + expect(anthropicReq.tools[0].name).toBe("shell"); + expect(anthropicReq.tools[0].input_schema.type).toBe("object"); + + // No OpenAI-only Responses knobs leaked onto the Anthropic wire. + expect(anthropicReq.reasoning).toBeUndefined(); + expect(anthropicReq.store).toBeUndefined(); + }); +}); diff --git a/tests/e2e/src/cases/responses-endpoint-e2e.test.ts b/tests/e2e/src/cases/responses-endpoint-e2e.test.ts index 0f9f4b05..87b72679 100644 --- a/tests/e2e/src/cases/responses-endpoint-e2e.test.ts +++ b/tests/e2e/src/cases/responses-endpoint-e2e.test.ts @@ -19,27 +19,24 @@ import { // Two user journeys pinned, both derived from the gateway's own // published contract in `docs/api-proxy.md` §4.6: // -// > Native OpenAI Responses API. OpenAI Models only — non-OpenAI -// > providers return 400. -// -// 1. Happy path — POST /v1/responses with an OpenAI-provider +// 1. Happy path (OpenAI) — POST /v1/responses with an OpenAI-provider // Model. Gateway dispatches to upstream's /v1/responses // (NOT /v1/chat/completions), caller receives the upstream's // Responses-shape body byte-for-byte, with the configured // Model's display name translated to upstream model_name. // -// 2. Provider mismatch — POST /v1/responses with an Anthropic- -// provider Model. Gateway must return 400 per the published -// contract; upstream must NOT be hit (the entire point of -// the restriction is OpenAI-Responses-shape doesn't translate -// to Anthropic Messages today). +// 2. Cross-provider (#825) — POST /v1/responses with a non-OpenAI +// Model. The gateway no longer rejects with 400; it bridges the +// Responses request through the canonical ChatFormat to the +// provider's native endpoint and re-encodes the reply into the +// Responses shape. (Anthropic streaming/tool coverage lives in +// responses-cross-provider-e2e; here we pin the OpenAI-compatible +// deepseek bridge as the representative non-OpenAI case.) // // References: // - OpenAI Responses API spec // // - Gateway's own /v1/responses contract: `docs/api-proxy.md` §4.6 -// - OpenAI error envelope spec -// const CALLER_PLAINTEXT = "sk-resp-e2e-caller"; const CALLER_KEY_HASH = createHash("sha256") @@ -218,132 +215,99 @@ describe("responses endpoint e2e: /v1/responses dispatch + provider mismatch", ( expect(sentBody.input).toBe("Say hello"); }); - // All three non-OpenAI providers per docs §6 (anthropic, gemini, - // deepseek). Per docs §4.6, /v1/responses on any of them must - // return 400. Parametrizing across all three catches a regression - // that special-cased one provider but mis-handled others — gemini - // and deepseek's bridges *do* speak OpenAI wire shape upstream, so - // a regression that "just dispatched anyway" would actually return - // 200 from the upstream-compat layer, billing the caller and - // silently violating the published contract. - const NON_OPENAI_PROVIDERS = [ - { - provider: "anthropic" as const, - modelName: "claude-3-5-haiku-20241022", - secret: "sk-ant-mock", - // Anthropic's documented endpoint is `https://api.anthropic.com/v1/messages` - // → api_base is the bare host. - apiBaseSuffix: "" as const, - }, - { - provider: "google" as const, - modelName: "gemini-2.0-flash", - secret: "sk-mock", - apiBaseSuffix: "/v1" as const, - }, - { - provider: "deepseek" as const, - modelName: "deepseek-chat", - secret: "sk-mock", - apiBaseSuffix: "/v1" as const, - }, - ]; - - for (const tc of NON_OPENAI_PROVIDERS) { - test(`non-OpenAI provider (${tc.provider}): caller sees 400 invalid_request_error, upstream untouched (per docs §4.6)`, async (ctx) => { - if (!etcdReachable || !app || !admin) { - ctx.skip(); - return; - } + // #825: a non-OpenAI provider on /v1/responses is now bridged, not + // rejected. deepseek is the representative case — its bridge speaks the + // OpenAI chat wire shape upstream, so the default mock body (a + // chat.completion) round-trips cleanly. The gateway must translate the + // Responses request into a chat completion, hit the provider's native + // /chat/completions endpoint (NOT /v1/responses), and re-encode the + // reply back into the Responses shape. (Richer Anthropic streaming/tool + // coverage lives in responses-cross-provider-e2e.) + test("non-OpenAI provider (deepseek): /v1/responses is bridged to a Responses-shape 200", async (ctx) => { + if (!etcdReachable || !app || !admin) { + ctx.skip(); + return; + } - const upstream = await startOpenAiUpstream(); - upstreams.push(upstream); + const upstream = await startOpenAiUpstream({ + nonStreamBody: { + id: "chatcmpl-ds-01", + object: "chat.completion", + created: Math.floor(Date.now() / 1000), + model: "deepseek-chat", + choices: [ + { + index: 0, + message: { role: "assistant", content: "bridged reply" }, + finish_reason: "stop", + }, + ], + usage: { prompt_tokens: 8, completion_tokens: 4, total_tokens: 12 }, + }, + }); + upstreams.push(upstream); - const pk = await admin.createProviderKey({ - display_name: `resp-${tc.provider}-pk`, - secret: tc.secret, - api_base: `${upstream.baseUrl}${tc.apiBaseSuffix}`, - }); - const modelDisplayName = `resp-${tc.provider}`; - await admin.createModel({ - display_name: modelDisplayName, - provider: tc.provider, - model_name: tc.modelName, - provider_key_id: pk.id, - }); + const pk = await admin.createProviderKey({ + display_name: "resp-deepseek-pk", + provider: "deepseek", + adapter: "openai", + secret: "sk-mock", + api_base: `${upstream.baseUrl}/v1`, + }); + await admin.createModel({ + display_name: "resp-deepseek", + provider: "deepseek", + model_name: "deepseek-chat", + provider_key_id: pk.id, + }); - const headers = { - authorization: `Bearer ${CALLER_PLAINTEXT}`, - "content-type": "application/json", - }; + const headers = { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + }; - // Readiness gate: poll until the gateway returns the - // documented 400 with `error.type: "invalid_request_error"` - // per docs §2 status→type table. A 404 here would be the - // snapshot-lag "model not found" case (model_not_found is - // mapped to 404, NOT 400, per the docs), so probing on - // 400 + invalid_request_error specifically gates on the - // gateway resolving the model AND refusing per §4.6. - await waitConfigPropagation(async () => { - try { - const r = await fetch(`${app!.proxyUrl}/v1/responses`, { - method: "POST", - headers, - body: JSON.stringify({ - model: modelDisplayName, - input: "ready-probe", - }), - }); - if (r.status !== 400) { - await r.text(); - return false; - } - const j = (await r.json()) as { - error?: { type?: unknown }; - }; - return j.error?.type === "invalid_request_error"; - } catch { + // Readiness gate: poll until the bridged path returns a Responses-shape + // 200 (no longer the pre-#825 400). + await waitConfigPropagation(async () => { + try { + const r = await fetch(`${app!.proxyUrl}/v1/responses`, { + method: "POST", + headers, + body: JSON.stringify({ model: "resp-deepseek", input: "ready-probe" }), + }); + if (r.status !== 200) { + await r.text(); return false; } - }); - - const upstreamHitsBefore = upstream.receivedRequests.length; - - const res = await fetch(`${app.proxyUrl}/v1/responses`, { - method: "POST", - headers, - body: JSON.stringify({ - model: modelDisplayName, - input: "Say hello", - }), - }); + const j = (await r.json()) as { object?: unknown }; + return j.object === "response"; + } catch { + return false; + } + }); - // Per docs §4.6: non-OpenAI providers return 400. Status - // family 5xx would mean the gateway crashed (it should - // refuse cleanly, not panic). - expect(res.status).toBe(400); + const baseline = upstream.receivedRequests.length; + const res = await fetch(`${app.proxyUrl}/v1/responses`, { + method: "POST", + headers, + body: JSON.stringify({ model: "resp-deepseek", input: "Say hello" }), + }); - const body = (await res.json()) as { - error?: { type?: unknown; message?: unknown }; - }; - // Per docs §2 status→type table: 400 → invalid_request_error. - // Pinning the exact value catches a regression where the - // gateway's refusal vocabulary drifts from the published - // contract (e.g. emits "service_unavailable" or - // "model_not_found"). Same convention body-edges-e2e and - // error-envelope-normalization-e2e use. - expect(body.error?.type).toBe("invalid_request_error"); - expect(typeof body.error?.message).toBe("string"); - expect((body.error?.message as string).length).toBeGreaterThan(0); + expect(res.status).toBe(200); + const body = (await res.json()) as { + object?: unknown; + status?: unknown; + output?: Array<{ type?: unknown; content?: Array<{ text?: unknown }> }>; + }; + expect(body.object).toBe("response"); + expect(body.status).toBe("completed"); + expect(body.output?.[0]?.type).toBe("message"); + expect(body.output?.[0]?.content?.[0]?.text).toBe("bridged reply"); - // Hard contract: upstream must never be hit when the gateway - // refuses for provider mismatch — otherwise the gateway is - // billing the caller's quota on a request it claims to reject. - // Critical for gemini and deepseek specifically, whose bridges - // DO speak OpenAI wire shape upstream — a regression that - // dispatched anyway would silently 200 from the upstream- - // compat layer. - expect(upstream.receivedRequests.length).toBe(upstreamHitsBefore); - }); - } + // The bridge dispatched to the provider's chat endpoint, not + // /v1/responses (the verbatim path is OpenAI-only). + const calls = upstream.receivedRequests.slice(baseline); + expect(calls.some((r) => r.path.endsWith("/chat/completions"))).toBe(true); + expect(calls.some((r) => r.path === "/v1/responses")).toBe(false); + }); }); diff --git a/tests/e2e/src/cases/tools-cross-provider-e2e.test.ts b/tests/e2e/src/cases/tools-cross-provider-e2e.test.ts index 97423b5f..1dee4b2c 100644 --- a/tests/e2e/src/cases/tools-cross-provider-e2e.test.ts +++ b/tests/e2e/src/cases/tools-cross-provider-e2e.test.ts @@ -271,10 +271,21 @@ describe("tools cross-provider e2e: OpenAI tools → Anthropic upstream tool_use model: "tools-xprov", messages: [ { role: "user", content: "What's the weather in SF?" }, - // (real callers would also include the assistant's - // tool_calls turn here; we skip it because Anthropic only - // requires the tool_use_id to be referenced in the next - // user-side tool_result.) + // The assistant's tool_calls turn is replayed too: the bridge + // translates it to an Anthropic `tool_use` block so the + // following `tool_result` references a tool_use the upstream saw + // (and roles still alternate user → assistant → user). + { + role: "assistant", + content: null, + tool_calls: [ + { + id: "toolu_xprov_01", + type: "function", + function: { name: "get_weather", arguments: '{"location":"SF"}' }, + }, + ], + }, { role: "tool", tool_call_id: "toolu_xprov_01",