diff --git a/model_gateway/src/routers/openai/chat.rs b/model_gateway/src/routers/openai/chat.rs index 3ce09c0246..7520742398 100644 --- a/model_gateway/src/routers/openai/chat.rs +++ b/model_gateway/src/routers/openai/chat.rs @@ -30,7 +30,7 @@ use crate::{ }; /// Shared context passed to chat routing functions. -pub(super) struct RouterContext<'a> { +pub(super) struct ChatRouterContext<'a> { pub worker_registry: &'a WorkerRegistry, pub provider_registry: &'a ProviderRegistry, pub shared_components: &'a Arc, @@ -40,7 +40,7 @@ pub(super) struct RouterContext<'a> { /// Route a chat completion request to the appropriate upstream worker. pub(super) async fn route_chat( - deps: &RouterContext<'_>, + deps: &ChatRouterContext<'_>, headers: Option<&HeaderMap>, body: &ChatCompletionRequest, model_id: Option<&str>, diff --git a/model_gateway/src/routers/openai/responses/mod.rs b/model_gateway/src/routers/openai/responses/mod.rs index a60c1b1e10..e9c6255093 100644 --- a/model_gateway/src/routers/openai/responses/mod.rs +++ b/model_gateway/src/routers/openai/responses/mod.rs @@ -13,6 +13,7 @@ mod accumulator; mod common; pub(crate) mod history; mod non_streaming; +pub(crate) mod route; mod streaming; mod utils; diff --git a/model_gateway/src/routers/openai/responses/route.rs b/model_gateway/src/routers/openai/responses/route.rs new file mode 100644 index 0000000000..f740782b3b --- /dev/null +++ b/model_gateway/src/routers/openai/responses/route.rs @@ -0,0 +1,195 @@ +//! Responses API routing orchestration. +//! +//! Mirrors the delegation pattern in `chat.rs`: the `RouterTrait` method in +//! `router.rs` packs borrowed references into [`ResponsesRouterContext`] and +//! delegates to [`route_responses`]. + +use std::{sync::Arc, time::Instant}; + +use axum::{http::HeaderMap, response::Response}; +use openai_protocol::responses::{ResponseInput, ResponseInputOutputItem, ResponsesRequest}; +use serde_json::to_value; + +use super::{ + super::{ + context::{ + ComponentRefs, PayloadState, RequestContext, ResponsesComponents, WorkerSelection, + }, + provider::ProviderRegistry, + router::resolve_provider, + }, + handle_non_streaming_response, handle_streaming_response, +}; +use crate::{ + core::{Endpoint, ProviderType, WorkerRegistry}, + observability::metrics::{bool_to_static_str, metrics_labels, Metrics}, + routers::{ + error, + worker_selection::{SelectWorkerRequest, WorkerSelector}, + }, +}; + +/// Shared context passed to responses routing functions. +pub(in crate::routers::openai) struct ResponsesRouterContext<'a> { + pub worker_registry: &'a WorkerRegistry, + pub provider_registry: &'a ProviderRegistry, + pub responses_components: &'a Arc, + pub client: &'a reqwest::Client, +} + +/// Route a responses API request to the appropriate upstream worker. +pub(in crate::routers::openai) async fn route_responses( + deps: &ResponsesRouterContext<'_>, + headers: Option<&HeaderMap>, + body: &ResponsesRequest, + model_id: Option<&str>, +) -> Response { + let start = Instant::now(); + let model = model_id.unwrap_or(body.model.as_str()); + let streaming = body.stream.unwrap_or(false); + + Metrics::record_router_request( + metrics_labels::ROUTER_OPENAI, + metrics_labels::BACKEND_EXTERNAL, + metrics_labels::CONNECTION_HTTP, + model, + metrics_labels::ENDPOINT_RESPONSES, + bool_to_static_str(streaming), + ); + + let worker = match WorkerSelector::new(deps.worker_registry, deps.client) + .select_worker(&SelectWorkerRequest { + model_id: model, + headers, + provider: Some(ProviderType::OpenAI), + ..Default::default() + }) + .await + { + Ok(w) => w, + Err(response) => { + Metrics::record_router_error( + metrics_labels::ROUTER_OPENAI, + metrics_labels::BACKEND_EXTERNAL, + metrics_labels::CONNECTION_HTTP, + model, + metrics_labels::ENDPOINT_RESPONSES, + metrics_labels::ERROR_NO_WORKERS, + ); + return response; + } + }; + + // Validate mutual exclusivity of conversation and previous_response_id + // Treat empty strings as unset to match other metadata paths + let has_conversation = body.conversation.as_ref().is_some_and(|s| !s.is_empty()); + let has_previous_response = body + .previous_response_id + .as_ref() + .is_some_and(|s| !s.is_empty()); + if has_conversation && has_previous_response { + Metrics::record_router_error( + metrics_labels::ROUTER_OPENAI, + metrics_labels::BACKEND_EXTERNAL, + metrics_labels::CONNECTION_HTTP, + model, + metrics_labels::ENDPOINT_RESPONSES, + metrics_labels::ERROR_VALIDATION, + ); + return error::bad_request( + "invalid_request", + "Cannot specify both 'conversation' and 'previous_response_id'".to_string(), + ); + } + + let mut request_body = body.clone(); + if let Some(model) = model_id { + request_body.model = model.to_string(); + } + request_body.conversation = None; + + let original_previous_response_id = match super::history::load_input_history( + deps.responses_components, + body, + &mut request_body, + model, + ) + .await + { + Ok(id) => id, + Err(response) => return response, + }; + + request_body.store = Some(false); + if let ResponseInput::Items(ref mut items) = request_body.input { + items.retain(|item| !matches!(item, ResponseInputOutputItem::Reasoning { .. })); + } + + let mut payload = match to_value(&request_body) { + Ok(v) => v, + Err(e) => { + Metrics::record_router_error( + metrics_labels::ROUTER_OPENAI, + metrics_labels::BACKEND_EXTERNAL, + metrics_labels::CONNECTION_HTTP, + model, + metrics_labels::ENDPOINT_RESPONSES, + metrics_labels::ERROR_VALIDATION, + ); + return error::bad_request( + "invalid_request", + format!("Failed to serialize request: {e}"), + ); + } + }; + + let provider = resolve_provider(deps.provider_registry, worker.as_ref(), model); + if let Err(e) = provider.transform_request(&mut payload, Endpoint::Responses) { + Metrics::record_router_error( + metrics_labels::ROUTER_OPENAI, + metrics_labels::BACKEND_EXTERNAL, + metrics_labels::CONNECTION_HTTP, + model, + metrics_labels::ENDPOINT_RESPONSES, + metrics_labels::ERROR_VALIDATION, + ); + return error::bad_request("invalid_request", format!("Provider transform error: {e}")); + } + + let mut ctx = RequestContext::for_responses( + Arc::new(body.clone()), + headers.cloned(), + model_id.map(String::from), + ComponentRefs::Responses(Arc::clone(deps.responses_components)), + ); + + ctx.state.worker = Some(WorkerSelection { + worker: Arc::clone(&worker), + provider: Arc::clone(&provider), + }); + + ctx.state.payload = Some(PayloadState { + json: payload, + url: format!("{}/v1/responses", worker.url()), + previous_response_id: original_previous_response_id, + }); + + let response = if ctx.is_streaming() { + handle_streaming_response(ctx).await + } else { + handle_non_streaming_response(ctx).await + }; + + if response.status().is_success() { + Metrics::record_router_duration( + metrics_labels::ROUTER_OPENAI, + metrics_labels::BACKEND_EXTERNAL, + metrics_labels::CONNECTION_HTTP, + model, + metrics_labels::ENDPOINT_RESPONSES, + start.elapsed(), + ); + } + + response +} diff --git a/model_gateway/src/routers/openai/router.rs b/model_gateway/src/routers/openai/router.rs index 90cecbd0df..40779d3997 100644 --- a/model_gateway/src/routers/openai/router.rs +++ b/model_gateway/src/routers/openai/router.rs @@ -1,7 +1,6 @@ use std::{ any::Any, sync::{atomic::AtomicBool, Arc}, - time::Instant, }; use axum::{body::Body, extract::Request, http::HeaderMap, response::Response}; @@ -11,27 +10,22 @@ use openai_protocol::{ RealtimeClientSecretCreateRequest, RealtimeSessionCreateRequest, RealtimeTranscriptionSessionCreateRequest, }, - responses::{ResponseInput, ResponseInputOutputItem, ResponsesGetParams, ResponsesRequest}, + responses::{ResponsesGetParams, ResponsesRequest}, }; -use serde_json::to_value; use super::{ - chat::{self, RouterContext}, - context::{ - ComponentRefs, PayloadState, RequestContext, ResponsesComponents, SharedComponents, - WorkerSelection, - }, + chat::{self, ChatRouterContext}, + context::{ResponsesComponents, SharedComponents}, health, provider::ProviderRegistry, - responses::{handle_non_streaming_response, handle_streaming_response}, + responses::route::{self as responses_route, ResponsesRouterContext}, }; use crate::{ app_context::AppContext, config::types::RetryConfig, - core::{Endpoint, ProviderType, Worker, WorkerRegistry}, - observability::metrics::{bool_to_static_str, metrics_labels, Metrics}, + core::{ProviderType, Worker, WorkerRegistry}, + observability::metrics::{metrics_labels, Metrics}, routers::{ - error, header_utils::extract_auth_header, openai::realtime::{rest::forward_realtime_rest, ws::handle_realtime_ws, RealtimeRegistry}, worker_selection::{SelectWorkerRequest, WorkerSelector}, @@ -130,10 +124,6 @@ impl OpenAIRouter { }) .await } - - fn responses_components(&self) -> Arc { - Arc::clone(&self.responses_components) - } } #[async_trait::async_trait] @@ -156,7 +146,7 @@ impl crate::routers::RouterTrait for OpenAIRouter { body: &ChatCompletionRequest, model_id: Option<&str>, ) -> Response { - let deps = RouterContext { + let deps = ChatRouterContext { worker_registry: &self.worker_registry, provider_registry: &self.provider_registry, shared_components: &self.shared_components, @@ -172,146 +162,13 @@ impl crate::routers::RouterTrait for OpenAIRouter { body: &ResponsesRequest, model_id: Option<&str>, ) -> Response { - let start = Instant::now(); - let model = model_id.unwrap_or(body.model.as_str()); - let streaming = body.stream.unwrap_or(false); - - Metrics::record_router_request( - metrics_labels::ROUTER_OPENAI, - metrics_labels::BACKEND_EXTERNAL, - metrics_labels::CONNECTION_HTTP, - model, - metrics_labels::ENDPOINT_RESPONSES, - bool_to_static_str(streaming), - ); - - let worker = match self.select_worker(model, headers).await { - Ok(w) => w, - Err(response) => { - Metrics::record_router_error( - metrics_labels::ROUTER_OPENAI, - metrics_labels::BACKEND_EXTERNAL, - metrics_labels::CONNECTION_HTTP, - model, - metrics_labels::ENDPOINT_RESPONSES, - metrics_labels::ERROR_NO_WORKERS, - ); - return response; - } - }; - - // Validate mutual exclusivity of conversation and previous_response_id - // Treat empty strings as unset to match other metadata paths - let has_conversation = body.conversation.as_ref().is_some_and(|s| !s.is_empty()); - let has_previous_response = body - .previous_response_id - .as_ref() - .is_some_and(|s| !s.is_empty()); - if has_conversation && has_previous_response { - Metrics::record_router_error( - metrics_labels::ROUTER_OPENAI, - metrics_labels::BACKEND_EXTERNAL, - metrics_labels::CONNECTION_HTTP, - model, - metrics_labels::ENDPOINT_RESPONSES, - metrics_labels::ERROR_VALIDATION, - ); - return error::bad_request( - "invalid_request", - "Cannot specify both 'conversation' and 'previous_response_id'".to_string(), - ); - } - - let mut request_body = body.clone(); - if let Some(model) = model_id { - request_body.model = model.to_string(); - } - request_body.conversation = None; - - let original_previous_response_id = match super::responses::history::load_input_history( - &self.responses_components, - body, - &mut request_body, - model, - ) - .await - { - Ok(id) => id, - Err(response) => return response, - }; - - request_body.store = Some(false); - if let ResponseInput::Items(ref mut items) = request_body.input { - items.retain(|item| !matches!(item, ResponseInputOutputItem::Reasoning { .. })); - } - - let mut payload = match to_value(&request_body) { - Ok(v) => v, - Err(e) => { - Metrics::record_router_error( - metrics_labels::ROUTER_OPENAI, - metrics_labels::BACKEND_EXTERNAL, - metrics_labels::CONNECTION_HTTP, - model, - metrics_labels::ENDPOINT_RESPONSES, - metrics_labels::ERROR_VALIDATION, - ); - return error::bad_request( - "invalid_request", - format!("Failed to serialize request: {e}"), - ); - } - }; - - let provider = resolve_provider(&self.provider_registry, worker.as_ref(), model); - if let Err(e) = provider.transform_request(&mut payload, Endpoint::Responses) { - Metrics::record_router_error( - metrics_labels::ROUTER_OPENAI, - metrics_labels::BACKEND_EXTERNAL, - metrics_labels::CONNECTION_HTTP, - model, - metrics_labels::ENDPOINT_RESPONSES, - metrics_labels::ERROR_VALIDATION, - ); - return error::bad_request("invalid_request", format!("Provider transform error: {e}")); - } - - let mut ctx = RequestContext::for_responses( - Arc::new(body.clone()), - headers.cloned(), - model_id.map(String::from), - ComponentRefs::Responses(self.responses_components()), - ); - - ctx.state.worker = Some(WorkerSelection { - worker: Arc::clone(&worker), - provider: Arc::clone(&provider), - }); - - ctx.state.payload = Some(PayloadState { - json: payload, - url: format!("{}/v1/responses", worker.url()), - previous_response_id: original_previous_response_id, - }); - - let response = if ctx.is_streaming() { - handle_streaming_response(ctx).await - } else { - handle_non_streaming_response(ctx).await + let deps = ResponsesRouterContext { + worker_registry: &self.worker_registry, + provider_registry: &self.provider_registry, + responses_components: &self.responses_components, + client: &self.responses_components.shared.client, }; - - if response.status().is_success() { - Metrics::record_router_duration( - metrics_labels::ROUTER_OPENAI, - metrics_labels::BACKEND_EXTERNAL, - metrics_labels::CONNECTION_HTTP, - model, - metrics_labels::ENDPOINT_RESPONSES, - start.elapsed(), - ); - } - - response + responses_route::route_responses(&deps, headers, body, model_id).await } async fn get_response(