diff --git a/Cargo.lock b/Cargo.lock index fd4d9b6bf..89b712d9d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2392,6 +2392,7 @@ dependencies = [ "opentelemetry_sdk", "parking_lot", "prometheus", + "reqwest", "rustls", "serde", "serde_json", diff --git a/crates/libsy-llm-client/src/backend.rs b/crates/libsy-llm-client/src/backend.rs index fab446095..a526712fc 100644 --- a/crates/libsy-llm-client/src/backend.rs +++ b/crates/libsy-llm-client/src/backend.rs @@ -391,34 +391,6 @@ mod tests { ); } - #[test] - fn count_tokens_url_joins_every_base_url_shape() { - assert_eq!( - Backend::Anthropic(config("https://host")).count_tokens_url(), - "https://host/v1/messages/count_tokens" - ); - assert_eq!( - Backend::Anthropic(config("https://host/v1")).count_tokens_url(), - "https://host/v1/messages/count_tokens" - ); - assert_eq!( - Backend::Anthropic(config("https://host/v1/messages")).count_tokens_url(), - "https://host/v1/messages/count_tokens" - ); - // Trailing slash is trimmed before the join. - assert_eq!( - Backend::Anthropic(config("https://host/v1/")).count_tokens_url(), - "https://host/v1/messages/count_tokens" - ); - } - - #[test] - fn only_anthropic_backend_is_anthropic() { - assert!(Backend::Anthropic(config("x")).is_anthropic()); - assert!(!Backend::OpenAiChat(config("x")).is_anthropic()); - assert!(!Backend::OpenAiResponses(config("x")).is_anthropic()); - } - #[test] fn wire_format_matches_variant() { assert_eq!( diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 4a517936f..9f47828aa 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -82,6 +82,34 @@ impl ModelConfig { } } +/// A model-bearing provider operation outside the normal completion endpoint. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum AuxiliaryOperation { + /// Anthropic Messages input-token counting. + AnthropicCountTokens, + /// OpenAI Responses input-token counting. + ResponsesInputTokens, + /// OpenAI Responses compaction. + ResponsesCompact, +} + +impl AuxiliaryOperation { + const fn wire_format(self) -> WireFormat { + match self { + Self::AnthropicCountTokens => WireFormat::AnthropicMessages, + Self::ResponsesInputTokens | Self::ResponsesCompact => WireFormat::OpenAiResponses, + } + } + + fn url(self, backend: &Backend) -> String { + match self { + Self::AnthropicCountTokens => backend.count_tokens_url(), + Self::ResponsesInputTokens => format!("{}/input_tokens", backend.url()), + Self::ResponsesCompact => format!("{}/compact", backend.url()), + } + } +} + /// A client that dispatches neutral-IR requests to per-model HTTP backends. /// /// Construct it with a list of [`ModelConfig`]s — one per model, each naming a @@ -146,22 +174,27 @@ impl TranslatingLlmClient { }) } - /// Whether `model` has an Anthropic backend that supports token counting. - pub fn supports_count_tokens(&self, model: &ModelId) -> bool { - self.backend_for(model, WireFormat::AnthropicMessages) - .is_some() + /// Whether `model` has a backend for `operation`. + pub fn supports_auxiliary(&self, model: &ModelId, operation: AuxiliaryOperation) -> bool { + self.backend_for(model, operation.wire_format()).is_some() } - /// Counts input tokens with `model`'s Anthropic backend. + /// Calls a model-bearing auxiliary provider operation. /// - /// Returns an error when the model has no Anthropic backend or the upstream + /// Returns an error when the model has no compatible backend or the upstream /// request fails or returns invalid JSON. - pub async fn count_tokens(&self, model: &ModelId, request: Request) -> Result { - let backend = self - .backend_for(model, WireFormat::AnthropicMessages) - .ok_or_else(|| LlmClientError::Configuration { - message: format!("model {model} has no Anthropic backend for count_tokens"), - })?; + pub async fn call_auxiliary( + &self, + model: &ModelId, + request: Request, + operation: AuxiliaryOperation, + ) -> Result { + let wire_format = operation.wire_format(); + let backend = + self.backend_for(model, wire_format) + .ok_or_else(|| LlmClientError::Configuration { + message: format!("model {model} has no backend for {operation:?}"), + })?; let Request { mut llm_request, metadata, @@ -171,20 +204,17 @@ impl TranslatingLlmClient { let http_response = self .send_encoded( backend, - WireFormat::AnthropicMessages, + wire_format, llm_request, metadata.as_ref(), model, - UpstreamEndpoint::CountTokens, + UpstreamEndpoint::Auxiliary(operation), ) .await?; - let body = match http_response { - EncodedResponse::Buffered { body, .. } => body, - EncodedResponse::Streaming(_) => { - return Err(LlmClientError::InvalidRequest { - message: "count_tokens does not support streaming requests".to_string(), - }); - } + let EncodedResponse::Buffered { body, .. } = http_response else { + return Err(LlmClientError::InvalidRequest { + message: "auxiliary endpoints do not support streaming".to_string(), + }); }; serde_json::from_slice(&body).map_err(|error| LlmClientError::InvalidResponse { source: Box::new(error), @@ -200,8 +230,7 @@ impl TranslatingLlmClient { /// overflow via the backend's provider rules. Shared by /// [`call_rewrite_model`](Self::call_rewrite_model) (which POSTs to the /// backend's completion URL and decodes a response) and - /// [`count_tokens`](Self::count_tokens) (which POSTs to the `count_tokens` - /// URL and returns the raw JSON). + /// the model-bearing auxiliary operations, which return raw JSON. async fn send_encoded( &self, backend: &Backend, @@ -562,14 +591,14 @@ impl RoutedLlmClient for TranslatingLlmClient { #[derive(Clone, Copy)] enum UpstreamEndpoint { Completion, - CountTokens, + Auxiliary(AuxiliaryOperation), } impl UpstreamEndpoint { fn url(self, backend: &Backend) -> String { match self { UpstreamEndpoint::Completion => backend.url(), - UpstreamEndpoint::CountTokens => backend.count_tokens_url(), + UpstreamEndpoint::Auxiliary(operation) => operation.url(backend), } } diff --git a/crates/libsy-llm-client/src/lib.rs b/crates/libsy-llm-client/src/lib.rs index 059675bbe..36175dd53 100644 --- a/crates/libsy-llm-client/src/lib.rs +++ b/crates/libsy-llm-client/src/lib.rs @@ -26,7 +26,7 @@ pub mod raw; pub mod run; pub use backend::{Backend, DEFAULT_MAX_RETRIES, HttpBackendConfig}; -pub use client::{ModelConfig, TranslatingLlmClient}; +pub use client::{AuxiliaryOperation, ModelConfig, TranslatingLlmClient}; pub use error::{LlmClientError, Result}; pub use observation::{LlmCallObservation, RunObservation, RunObserver}; pub use raw::RawResponse; diff --git a/crates/switchyard-runner/src/config.rs b/crates/switchyard-runner/src/config.rs index d4887fab1..4b211ae81 100644 --- a/crates/switchyard-runner/src/config.rs +++ b/crates/switchyard-runner/src/config.rs @@ -12,13 +12,13 @@ use serde::de::DeserializeOwned; use serde::{Deserialize, Deserializer}; use serde_json::Value; use switchyard_llm_client::{ - Backend, ClientRouter, DEFAULT_MAX_RETRIES, HttpBackendConfig, ModelConfig, + AuxiliaryOperation, Backend, ClientRouter, DEFAULT_MAX_RETRIES, HttpBackendConfig, ModelConfig, TranslatingLlmClient, }; use switchyard_protocol::{ModelId, RoutedLlmClient, WireFormat}; use crate::{ - AlgorithmSpec, CallerAuthKind, CountTokensTarget, DecisionTarget, ModelCapabilities, Route, + AlgorithmSpec, AuxiliaryTarget, CallerAuthKind, DecisionTarget, ModelCapabilities, Route, Runner, RunnerError, }; @@ -54,6 +54,7 @@ pub(crate) fn runner_from_toml(source: &str) -> RunnerResult { #[serde(deny_unknown_fields)] pub(crate) struct DeploymentConfig { schema_version: u32, + fallback_client: Option, #[serde(default)] llm_clients: BTreeMap, targets: BTreeMap, @@ -165,6 +166,7 @@ impl DeploymentConfig { let clients = self.build_clients()?; let targets = self.build_targets(); + let fallback_base_url = self.fallback_base_url()?; let mut routes = Vec::with_capacity(self.routes.len()); for (route_name, config) in &self.routes { validate_value("route name", route_name)?; @@ -188,7 +190,10 @@ impl DeploymentConfig { .map_err(|error| RunnerError::configuration_source(error.to_string(), error))?; let (route_clients, caller_auth) = self.build_route_clients(route_name, config, &clients)?; - let count_tokens_target = self.build_count_tokens_target(config, &clients); + let anthropic_auxiliary_target = + self.build_anthropic_auxiliary_target(config, &clients); + let responses_auxiliary_target = + self.build_responses_auxiliary_target(config, &clients); let decision_targets = config .routing_target_names() .into_iter() @@ -199,12 +204,14 @@ impl DeploymentConfig { route_clients, caller_auth, capabilities, - count_tokens_target, + anthropic_auxiliary_target, + responses_auxiliary_target, decision_targets, ); routes.push((config.id.clone(), route)); } - Ok(Runner::new(routes)) + let runner = Runner::new(routes).with_fallback_url(fallback_base_url); + Ok(runner) } fn build_clients(&self) -> RunnerResult>> { @@ -291,27 +298,60 @@ impl DeploymentConfig { Ok((ClientRouter::new(by_model), caller_auth)) } - fn build_count_tokens_target( + fn fallback_base_url(&self) -> RunnerResult> { + let Some(name) = &self.fallback_client else { + return Ok(None); + }; + let config = self.llm_clients.get(name).ok_or_else(|| { + RunnerError::configuration(format!( + "fallback_client references unknown llm client {name}" + )) + })?; + Ok(Some(config.base_url.as_str().to_string())) + } + + fn build_anthropic_auxiliary_target( &self, route: &RouteConfig, clients: &BTreeMap>, - ) -> Option { + ) -> Option { route .routing_target_names() .into_iter() .enumerate() .filter_map(|(index, name)| { - let target = self.targets.get(name)?; - let client = clients.get(&target.llm_client)?; - client.supports_count_tokens(&target.id).then_some(( - count_tokens_priority(name, &target.id), - index, - target, - client, - )) + let target = self.build_auxiliary_target( + name, + clients, + AuxiliaryOperation::AnthropicCountTokens, + )?; + Some((count_tokens_priority(name, &target.model), index, target)) }) - .min_by_key(|(priority, index, _, _)| (*priority, *index)) - .map(|(_, _, target, client)| CountTokensTarget { + .min_by_key(|(priority, index, _)| (*priority, *index)) + .map(|(_, _, target)| target) + } + + fn build_responses_auxiliary_target( + &self, + route: &RouteConfig, + clients: &BTreeMap>, + ) -> Option { + route.routing_target_names().into_iter().find_map(|name| { + self.build_auxiliary_target(name, clients, AuxiliaryOperation::ResponsesInputTokens) + }) + } + + fn build_auxiliary_target( + &self, + name: &str, + clients: &BTreeMap>, + operation: AuxiliaryOperation, + ) -> Option { + let target = self.targets.get(name)?; + let client = clients.get(&target.llm_client)?; + client + .supports_auxiliary(&target.id, operation) + .then(|| AuxiliaryTarget { model: target.id.clone(), client: client.clone(), }) diff --git a/crates/switchyard-runner/src/failure.rs b/crates/switchyard-runner/src/failure.rs index 4e2901b7a..8113ef25b 100644 --- a/crates/switchyard-runner/src/failure.rs +++ b/crates/switchyard-runner/src/failure.rs @@ -100,7 +100,7 @@ impl RunnerError { ), Self::UnknownRouteModel(_) | Self::IncompatibleCallerFormat(_) - | Self::CountTokensUnsupported => summary( + | Self::AuxiliaryUnsupported => summary( RouteErrorKind::InvalidRequest, RouteErrorPhase::BeforeResponse, None, @@ -329,7 +329,7 @@ mod tests { RouteErrorKind::Configuration )); - let unsupported = RunnerError::CountTokensUnsupported; + let unsupported = RunnerError::AuxiliaryUnsupported; assert!(matches!( unsupported.execution_error_summary().kind, RouteErrorKind::InvalidRequest diff --git a/crates/switchyard-runner/src/lib.rs b/crates/switchyard-runner/src/lib.rs index 75543c2fe..efb0441e4 100644 --- a/crates/switchyard-runner/src/lib.rs +++ b/crates/switchyard-runner/src/lib.rs @@ -15,6 +15,6 @@ pub use algorithm::{ }; pub use failure::{RouteErrorKind, RouteErrorPhase, RouteErrorSummary, stream_error_summary}; pub use route::{ - CallerAuthKind, CountTokensTarget, ModelCapabilities, Route, RunOutput, RunnerError, + AuxiliaryTarget, CallerAuthKind, ModelCapabilities, Route, RunOutput, RunnerError, }; pub use runner::{DecisionDescription, DecisionTarget, ModelInfo, Runner}; diff --git a/crates/switchyard-runner/src/route.rs b/crates/switchyard-runner/src/route.rs index dc4db13ee..43022a9b5 100644 --- a/crates/switchyard-runner/src/route.rs +++ b/crates/switchyard-runner/src/route.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use libsy::{Algorithm, CallModel, LibsyError, RoutingOutcome, drive}; use serde_json::Value; -use switchyard_llm_client::{ClientRouter, RunObserver, TranslatingLlmClient}; +use switchyard_llm_client::{AuxiliaryOperation, ClientRouter, RunObserver, TranslatingLlmClient}; use switchyard_protocol::{LlmClientError, ModelId, Request, Response, WireFormat}; use thiserror::Error; @@ -56,9 +56,9 @@ impl CallerAuthKind { } } -/// Exact upstream model used for Anthropic token counting. +/// Exact upstream model and client used for an auxiliary provider operation. #[derive(Clone)] -pub struct CountTokensTarget { +pub struct AuxiliaryTarget { pub model: ModelId, pub client: Arc, } @@ -76,8 +76,8 @@ pub enum RunnerError { UnknownRouteModel(String), #[error("caller format is incompatible with {} credentials", .0.as_str())] IncompatibleCallerFormat(CallerAuthKind), - #[error("route has no Anthropic target for token counting")] - CountTokensUnsupported, + #[error("route has no compatible target for the auxiliary operation")] + AuxiliaryUnsupported, #[error(transparent)] Algorithm(#[from] LibsyError), #[error(transparent)] @@ -112,7 +112,8 @@ pub struct Route { clients: ClientRouter, caller_auth: Option, capabilities: ModelCapabilities, - count_tokens_target: Option, + anthropic_auxiliary_target: Option, + responses_auxiliary_target: Option, decision_targets: Vec, } @@ -129,7 +130,8 @@ impl Route { clients: ClientRouter, caller_auth: Option, capabilities: ModelCapabilities, - count_tokens_target: Option, + anthropic_auxiliary_target: Option, + responses_auxiliary_target: Option, decision_targets: Vec, ) -> Self { Self { @@ -137,7 +139,8 @@ impl Route { clients, caller_auth, capabilities, - count_tokens_target, + anthropic_auxiliary_target, + responses_auxiliary_target, decision_targets, } } @@ -203,15 +206,23 @@ impl Route { .map_err(Into::into) } - /// Counts tokens using the configured Anthropic-capable target. - pub async fn count_tokens(&self, request: Request) -> Result { - let target = self - .count_tokens_target - .as_ref() - .ok_or(RunnerError::CountTokensUnsupported)?; + /// Executes a model-bearing provider operation through a compatible target. + pub async fn call_auxiliary( + &self, + request: Request, + operation: AuxiliaryOperation, + ) -> Result { + let target = match operation { + AuxiliaryOperation::AnthropicCountTokens => &self.anthropic_auxiliary_target, + AuxiliaryOperation::ResponsesInputTokens | AuxiliaryOperation::ResponsesCompact => { + &self.responses_auxiliary_target + } + } + .as_ref() + .ok_or(RunnerError::AuxiliaryUnsupported)?; target .client - .count_tokens(&target.model, request) + .call_auxiliary(&target.model, request, operation) .await .map_err(Into::into) } diff --git a/crates/switchyard-runner/src/runner.rs b/crates/switchyard-runner/src/runner.rs index 93dce5e76..bcd175344 100644 --- a/crates/switchyard-runner/src/runner.rs +++ b/crates/switchyard-runner/src/runner.rs @@ -16,6 +16,7 @@ use crate::{ModelCapabilities, Route, RunnerError}; /// Immutable named route table. pub struct Runner { routes: Vec<(ModelId, Route)>, + fallback_base_url: Option, } /// Borrowed model metadata returned while listing routes. @@ -57,7 +58,15 @@ impl Runner { /// Builds a runner from named routes in caller-provided order. /// Pre-condition: There must be at least one route. pub fn new(routes: Vec<(ModelId, Route)>) -> Self { - Self { routes } + Self { + routes, + fallback_base_url: None, + } + } + + pub(crate) fn with_fallback_url(mut self, fallback_base_url: Option) -> Self { + self.fallback_base_url = fallback_base_url; + self } /// Returns the route registered for a model. @@ -77,6 +86,11 @@ impl Runner { }) } + /// Returns the validated API root used for unmatched HTTP requests. + pub fn fallback_base_url(&self) -> Option<&str> { + self.fallback_base_url.as_deref() + } + /// Resolves an outcome to configured target names and non-secret client settings. pub fn describe_decision( &self, diff --git a/crates/switchyard-runner/tests/route.rs b/crates/switchyard-runner/tests/route.rs index 71ce57c7e..b6b1d1827 100644 --- a/crates/switchyard-runner/tests/route.rs +++ b/crates/switchyard-runner/tests/route.rs @@ -52,6 +52,7 @@ fn plugin_route(client: Arc) -> Route { None, ModelCapabilities::default(), None, + None, Vec::new(), ) } diff --git a/crates/switchyard-server/Cargo.toml b/crates/switchyard-server/Cargo.toml index 0c4b47e24..29cea432f 100644 --- a/crates/switchyard-server/Cargo.toml +++ b/crates/switchyard-server/Cargo.toml @@ -32,6 +32,7 @@ opentelemetry-prometheus = "0.32" opentelemetry_sdk = { version = "0.32", default-features = false, features = ["metrics", "trace"] } parking_lot.workspace = true prometheus = "0.14" +reqwest.workspace = true serde.workspace = true switchyard-llm-client.workspace = true switchyard-protocol.workspace = true diff --git a/crates/switchyard-server/README.md b/crates/switchyard-server/README.md index 2ff1a7a5a..3f59fd0af 100644 --- a/crates/switchyard-server/README.md +++ b/crates/switchyard-server/README.md @@ -130,6 +130,9 @@ documented in [Stage-Router Routing](../../docs/routing_algorithms/stage_router_ | `POST` | `/v1/responses` | OpenAI Responses | | `POST` | `/v1/decision` | Resolve selected and fallback targets without a post-routing answer call | | `POST` | `/v1/messages/count_tokens` | Token count from a route's Anthropic target | +| `POST` | `/v1/responses/input_tokens` | Token count from a route's OpenAI Responses target | +| `POST` | `/v1/responses/compact` | Compaction through a route's OpenAI Responses target | +| `ANY` | Any unmatched path | Raw forward through the optional `fallback_client` | | `GET` | `/v1/models` | Routes served by this deployment | | `GET` | `/v1/stats` | Per-model usage plus curated algorithm stats | | `POST` | `/v1/stats/reset` | Clear accumulated stats | @@ -140,6 +143,16 @@ Requests name a route by its `id`, so `POST /v1/chat/completions` with `"model": routes through the `[routes.general]` entry above. Any of the three request formats can address any route, and the server translates between them. +Set the top-level `fallback_client` to an entry under `[llm_clients]` to proxy any otherwise +unmatched method and path through that client. The fallback client does not need a target. Requests +and responses are forwarded without translation, including paths, query strings, bodies, and model +identifiers. The caller's end-to-end headers, including authorization, are forwarded; hop-by-hop +headers are removed, and the client's configured API key, extra headers, format, and retry policy +are not applied. Without `fallback_client`, unmatched paths return `404`. + +The three model-bearing auxiliary endpoints above remain routed operations: they resolve the route +name to a compatible target, rewrite the upstream model ID, and use that target's configured client. + `POST /v1/decision` accepts `{"input_format": "openai_chat", "request": {...}}`, where the nested request names the route in `model`. It executes required classifier or judge calls, then returns the selected target and ordered fallbacks with their model, format, base URL, and diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 77150399c..ae7671b61 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -23,6 +23,7 @@ use std::path::PathBuf; use std::sync::Arc; use std::time::{Duration, Instant}; +use axum::body::Body; use axum::extract::rejection::{JsonRejection, QueryRejection}; use axum::extract::{DefaultBodyLimit, Query, Request as HttpRequest, State}; use axum::http::header::CONTENT_TYPE; @@ -36,7 +37,7 @@ use libsy::{Algorithm, LibsyError, RoutingOutcome}; use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; -use switchyard_llm_client::{ClientRouter, RunObservation, RunObserver}; +use switchyard_llm_client::{AuxiliaryOperation, ClientRouter, RunObservation, RunObserver}; use switchyard_protocol::{LlmClientError, Metadata, ModelId, Request, Usage}; use switchyard_runner::{ CallerAuthKind, DecisionTarget, ModelCapabilities, Route, RunOutput, Runner, RunnerError, @@ -136,6 +137,7 @@ struct DecisionLlmClientResponse { #[derive(Clone)] pub struct ServerState { runner: Arc, + fallback_http: reqwest::Client, metrics: prometheus::Registry, stats: StatsAccumulator, routing_log: Option, @@ -190,6 +192,7 @@ impl ServerState { None, ModelCapabilities::default(), None, + None, Vec::new(), ), ) @@ -202,12 +205,17 @@ impl ServerState { /// Creates HTTP-server state around an already configured runner. pub fn from_runner(runner: Runner) -> ServerResult { let metrics = metrics::registry().map_err(ServerError::new)?; + let fallback_http = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(|error| ServerError::new(error.to_string()))?; let stats = StatsAccumulator::new( metrics.clone(), runner.models().map(|model| model.algorithm), ); Ok(Self { runner: Arc::new(runner), + fallback_http, metrics, stats, routing_log: None, @@ -479,6 +487,11 @@ pub fn build_switchyard_router(state: ServerState) -> Router { .route("/v1/responses", post(openai_responses)) .route("/v1/decision", post(decision)) .route("/v1/messages/count_tokens", post(anthropic_count_tokens)) + .route( + "/v1/responses/input_tokens", + post(openai_responses_input_tokens), + ) + .route("/v1/responses/compact", post(openai_responses_compact)) .route("/v1/models", get(models)) .route("/v1/stats", get(get_stats)) .route("/v1/stats/reset", post(reset_stats)) @@ -488,7 +501,7 @@ pub fn build_switchyard_router(state: ServerState) -> Router { router = router.route("/v1/routing/session-stats", get(get_session_stats)); } router - .fallback(not_found) + .fallback(proxy_unmatched) .layer(DefaultBodyLimit::max(DEFAULT_MAX_REQUEST_BODY_BYTES)) // `layer` only wraps routes registered before it, so this stays last. .layer(axum::middleware::from_fn(stamp_request_start)) @@ -539,6 +552,82 @@ async fn openai_responses( handle_endpoint(state, started, headers, body, WireFormat::OpenAiResponses).await } +// Forwards unmatched requests unchanged to the configured API root. +async fn proxy_unmatched(State(state): State, request: HttpRequest) -> Response { + let Some(base_url) = state.runner.fallback_base_url() else { + return not_found().await; + }; + let (mut parts, body) = request.into_parts(); + let path_and_query = parts + .uri + .path_and_query() + .map_or_else(|| parts.uri.path().to_string(), ToString::to_string); + strip_hop_by_hop_headers(&mut parts.headers); + parts.headers.remove(axum::http::header::HOST); + let body = reqwest::Body::wrap_stream(body.into_data_stream()); + match state + .fallback_http + .request(parts.method, fallback_url(base_url, &path_and_query)) + .headers(parts.headers) + .body(body) + .send() + .await + { + Ok(response) => { + let response: http::Response = response.into(); + let (mut parts, body) = response.into_parts(); + strip_hop_by_hop_headers(&mut parts.headers); + Response::from_parts(parts, Body::new(body)) + } + Err(error) => error_response( + StatusCode::BAD_GATEWAY, + error.to_string(), + "upstream_error", + "upstream_error", + ), + } +} + +fn fallback_url(base_url: &str, path_and_query: &str) -> String { + let base_url = base_url.trim_end_matches('/'); + let root = [ + "/v1/chat/completions", + "/v1/responses", + "/v1/messages", + "/v1", + ] + .iter() + .find_map(|suffix| base_url.strip_suffix(suffix)) + .unwrap_or(base_url); + format!("{root}{path_and_query}") +} + +fn strip_hop_by_hop_headers(headers: &mut HeaderMap) { + let connection_headers = headers + .get_all(axum::http::header::CONNECTION) + .iter() + .filter_map(|value| value.to_str().ok()) + .flat_map(|value| value.split(',')) + .filter_map(|name| HeaderName::from_bytes(name.trim().as_bytes()).ok()) + .collect::>(); + for name in connection_headers { + headers.remove(name); + } + for name in [ + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "proxy-connection", + "te", + "trailer", + "transfer-encoding", + "upgrade", + ] { + headers.remove(name); + } +} + /// One provider request submitted for a routing decision without an answer-model call. #[derive(Deserialize)] #[serde(deny_unknown_fields)] @@ -634,22 +723,86 @@ async fn anthropic_count_tokens( Ok(resolved) => resolved, Err(response) => return anthropic_error_response(response), }; - anthropic_error_response(match route.count_tokens(request).await { - Ok(payload) => (StatusCode::OK, Json(payload)).into_response(), - Err(RunnerError::CountTokensUnsupported) => error_response( - StatusCode::BAD_REQUEST, - "route has no Anthropic target for token counting", - "invalid_request_error", - "count_tokens_unsupported", - ), - Err(RunnerError::Client(error)) => count_tokens_error(error), - Err(error) => server_error(error.to_string()), - }) + anthropic_error_response( + match route + .call_auxiliary(request, AuxiliaryOperation::AnthropicCountTokens) + .await + { + Ok(payload) => (StatusCode::OK, Json(payload)).into_response(), + Err(RunnerError::AuxiliaryUnsupported) => error_response( + StatusCode::BAD_REQUEST, + "route has no Anthropic target for token counting", + "invalid_request_error", + "count_tokens_unsupported", + ), + Err(RunnerError::Client(error)) => client_error(&error), + Err(error) => server_error(error.to_string()), + }, + ) +} + +async fn openai_responses_input_tokens( + State(state): State, + headers: HeaderMap, + body: std::result::Result, JsonRejection>, +) -> Response { + openai_responses_auxiliary( + state, + headers, + body, + AuxiliaryOperation::ResponsesInputTokens, + ) + .await +} + +async fn openai_responses_compact( + State(state): State, + headers: HeaderMap, + body: std::result::Result, JsonRejection>, +) -> Response { + openai_responses_auxiliary(state, headers, body, AuxiliaryOperation::ResponsesCompact).await } -/// Maps a token-count client failure with the same policy as a routed client call. -fn count_tokens_error(error: LlmClientError) -> Response { - client_error(&error) +// Resolves route aliases and configured credentials before calling a model-bearing Responses API. +async fn openai_responses_auxiliary( + state: ServerState, + headers: HeaderMap, + body: std::result::Result, JsonRejection>, + operation: AuxiliaryOperation, +) -> Response { + let body = match llm_json_body(body) { + Ok(body) => body, + Err((status, message)) => { + return render_error_response( + invalid_body_error(status, message), + WireFormat::OpenAiResponses, + ); + } + }; + let (route, request) = match resolve_route( + &state, + metadata_from_headers(headers), + body, + WireFormat::OpenAiResponses, + ) { + Ok(resolved) => resolved, + Err(response) => return render_error_response(response, WireFormat::OpenAiResponses), + }; + let result = route.call_auxiliary(request, operation).await; + render_error_response( + match result { + Ok(payload) => (StatusCode::OK, Json(payload)).into_response(), + Err(RunnerError::AuxiliaryUnsupported) => error_response( + StatusCode::BAD_REQUEST, + "route has no OpenAI Responses target", + "invalid_request_error", + "responses_auxiliary_unsupported", + ), + Err(RunnerError::Client(error)) => client_error(&error), + Err(error) => server_error(error.to_string()), + }, + WireFormat::OpenAiResponses, + ) } async fn handle_endpoint( @@ -1450,6 +1603,9 @@ fn endpoint_listing(has_routing_log: bool) -> String { " POST /v1/messages Anthropic Messages", " POST /v1/responses OpenAI Responses", " POST /v1/messages/count_tokens", + " POST /v1/responses/input_tokens", + " POST /v1/responses/compact", + " ANY unmatched paths optional fallback client", " GET /v1/models configured routes", " GET /v1/stats routing stats", " POST /v1/stats/reset", diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index 32290ea36..1da34ccc7 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -11,7 +11,7 @@ use std::sync::Arc; use axum::body::{Body, Bytes}; use axum::extract::{DefaultBodyLimit, State}; -use axum::http::{HeaderMap, Request as HttpRequest, StatusCode}; +use axum::http::{HeaderMap, HeaderValue, Request as HttpRequest, StatusCode, Uri}; use axum::response::sse::{Event, Sse}; use axum::response::{IntoResponse, Response as HttpResponse}; use axum::routing::post; @@ -58,6 +58,12 @@ impl MockUpstream { ) .route("/capture", post(upstream_redirect_capture)) .route("/v1/messages/count_tokens", post(upstream_count_tokens)) + .route( + "/v1/responses/input_tokens", + post(upstream_responses_auxiliary), + ) + .route("/v1/responses/compact", post(upstream_responses_auxiliary)) + .route("/future/provider/endpoint", post(upstream_fallback)) .layer(DefaultBodyLimit::disable()) .with_state(Arc::clone(&calls)); let listener = TcpListener::bind("127.0.0.1:0").await?; @@ -430,6 +436,51 @@ async fn upstream_count_tokens( Json(json!({"input_tokens": 7})).into_response() } +async fn upstream_responses_auxiliary( + State(calls): State>>>, + uri: Uri, + headers: HeaderMap, + Json(body): Json, +) -> HttpResponse { + calls.lock().await.push(json!({ + "path": uri.path(), + "body": body, + "configured_header": headers.get("x-configured-client").and_then(|value| value.to_str().ok()) + })); + if uri.path().ends_with("/input_tokens") { + Json(json!({"input_tokens": 11})).into_response() + } else { + Json(json!({"id": "resp_compacted", "object": "response", "output": []})).into_response() + } +} + +async fn upstream_fallback( + State(calls): State>>>, + headers: HeaderMap, + Json(body): Json, +) -> HttpResponse { + calls.lock().await.push(json!({ + "body": body, + "authorization": headers.get("authorization").and_then(|value| value.to_str().ok()), + "end_to_end": headers.get("x-end-to-end").and_then(|value| value.to_str().ok()), + "configured_secret": headers.contains_key("x-configured-secret"), + "connection": headers.contains_key("connection"), + "connection_nominated": headers.contains_key("x-remove-me") + })); + let mut response = Json(json!({"input_tokens": 7})).into_response(); + response + .headers_mut() + .insert("connection", HeaderValue::from_static("x-upstream-hop")); + response + .headers_mut() + .insert("x-upstream-hop", HeaderValue::from_static("remove")); + response.headers_mut().insert( + "x-end-to-end-response", + HeaderValue::from_static("preserve"), + ); + response +} + fn random_state(base_url: &str, routes: &[(&str, &[&str])]) -> TestResult { random_state_with_retries(base_url, routes, 0) } @@ -1713,7 +1764,7 @@ response_format_type = "json_object" } #[tokio::test] -async fn count_tokens_forwards_to_configured_anthropic_target() -> TestResult { +async fn model_bearing_auxiliary_endpoints_use_configured_route_targets() -> TestResult { let upstream = MockUpstream::start().await?; let state = load_test_config(&format!( r#" @@ -1723,6 +1774,15 @@ schema_version = 1 format = "anthropic_messages" base_url = "{base_url}" +[llm_clients.responses] +format = "openai_responses" +base_url = "{base_url}" +extra_headers = {{ "x-configured-client" = "responses" }} + +[targets.responses] +id = "real/responses-model" +llm_client = "responses" + [targets.strong] id = "real/opus" llm_client = "claude" @@ -1734,13 +1794,13 @@ llm_client = "claude" [routes.random] id = "switchyard/random" type = "random" -targets = ["other", "strong"] +targets = ["responses", "other", "strong"] "#, base_url = upstream.base_url ))?; let app = build_switchyard_router(state); - let response = send( + let count_tokens = send( &app, "POST", "/v1/messages/count_tokens", @@ -1750,13 +1810,151 @@ targets = ["other", "strong"] })), ) .await?; - assert_eq!(response.status, StatusCode::OK); - assert_eq!(response.json()?["input_tokens"], 7); + assert_eq!(count_tokens.status, StatusCode::OK); + assert_eq!(count_tokens.json()?["input_tokens"], 7); + + let input_tokens = send( + &app, + "POST", + "/v1/responses/input_tokens", + Some(json!({"model": "switchyard/random", "input": "count me"})), + ) + .await?; + assert_eq!(input_tokens.status, StatusCode::OK); + assert_eq!(input_tokens.json()?["input_tokens"], 11); + + let compact = send( + &app, + "POST", + "/v1/responses/compact", + Some(json!({"model": "switchyard/random", "input": "compact me"})), + ) + .await?; + assert_eq!(compact.status, StatusCode::OK); + assert_eq!(compact.json()?["id"], "resp_compacted"); let calls = upstream.calls.lock().await; - assert_eq!(calls.len(), 1); - // The inbound route name is rewritten to the real upstream model. + assert_eq!(calls.len(), 3); assert_eq!(calls[0]["model"], "real/opus"); + assert_eq!( + calls[1], + json!({ + "path": "/v1/responses/input_tokens", + "body": {"model": "real/responses-model", "input": "count me"}, + "configured_header": "responses" + }) + ); + assert_eq!( + calls[2], + json!({ + "path": "/v1/responses/compact", + "body": {"model": "real/responses-model", "input": "compact me"}, + "configured_header": "responses" + }) + ); + drop(calls); + + let unsupported = random_state(&upstream.base_url, &[(ROUTE_MODEL, &["model/weak"])])?; + let unsupported = build_switchyard_router(unsupported); + for (path, body) in [ + ( + "/v1/messages/count_tokens", + json!({"model": ROUTE_MODEL, "messages": [{"role": "user", "content": "hi"}]}), + ), + ( + "/v1/responses/input_tokens", + json!({"model": ROUTE_MODEL, "input": "count me"}), + ), + ( + "/v1/responses/compact", + json!({"model": ROUTE_MODEL, "input": "compact me"}), + ), + ] { + let response = send(&unsupported, "POST", path, Some(body)).await?; + assert_eq!(response.status, StatusCode::BAD_REQUEST, "{path}"); + } + Ok(()) +} + +#[tokio::test] +async fn fallback_client_forwards_unmatched_requests_and_is_optional() -> TestResult { + let upstream = MockUpstream::start().await?; + let state = load_test_config(&format!( + r#" +schema_version = 1 +fallback_client = "fallback" + +[llm_clients.fallback] +format = "openai_responses" +base_url = "{base_url}" +extra_headers = {{ "x-configured-secret" = "must-not-forward" }} + +[llm_clients.routed] +format = "openai_chat" +base_url = "{base_url}" + +[targets.weak] +id = "model/weak" +llm_client = "routed" + +[routes.random] +id = "switchyard/random" +type = "passthrough" +target = "weak" +"#, + base_url = upstream.base_url + ))?; + let app = build_switchyard_router(state); + + let response = send_with_headers( + &app, + "POST", + "/future/provider/endpoint?mode=raw", + Some(json!({ + "model": "provider/model", + "provider_field": {"nested": true} + })), + &[ + ("authorization", "Bearer caller-key"), + ("connection", "keep-alive, x-remove-me"), + ("keep-alive", "timeout=5"), + ("x-remove-me", "remove"), + ("x-end-to-end", "preserve"), + ], + ) + .await?; + assert_eq!(response.status, StatusCode::OK); + assert_eq!(response.json()?["input_tokens"], 7); + assert!(!response.headers.contains_key("connection")); + assert!(!response.headers.contains_key("x-upstream-hop")); + assert_eq!( + response + .headers + .get("x-end-to-end-response") + .and_then(|value| value.to_str().ok()), + Some("preserve") + ); + let calls = upstream.calls.lock().await; + assert_eq!( + calls.as_slice(), + &[json!({ + "body": { + "model": "provider/model", + "provider_field": {"nested": true} + }, + "authorization": "Bearer caller-key", + "end_to_end": "preserve", + "configured_secret": false, + "connection": false, + "connection_nominated": false + })] + ); + drop(calls); + + let state = random_state(&upstream.base_url, &[(ROUTE_MODEL, &["model/weak"])])?; + let app = build_switchyard_router(state); + let response = send(&app, "POST", "/future/provider/endpoint", None).await?; + assert_eq!(response.status, StatusCode::NOT_FOUND); Ok(()) } @@ -1896,55 +2094,6 @@ target = "openai" Ok(()) } -#[tokio::test] -async fn count_tokens_without_anthropic_target_returns_bad_request() -> TestResult { - let upstream = MockUpstream::start().await?; - let state = load_test_config(&format!( - r#" -schema_version = 1 - -[llm_clients.upstream] -format = "openai_chat" -base_url = "{base_url}" - -[targets.weak] -id = "model/weak" -llm_client = "upstream" - -[routes.random] -id = "switchyard/random" -type = "random" -targets = ["weak"] -"#, - base_url = upstream.base_url - ))?; - let app = build_switchyard_router(state); - - let response = send( - &app, - "POST", - "/v1/messages/count_tokens", - Some(json!({ - "model": "switchyard/random", - "messages": [{"role": "user", "content": "hi"}] - })), - ) - .await?; - assert_eq!(response.status, StatusCode::BAD_REQUEST); - // This route has no Anthropic-format target. - assert_eq!( - response.json()?, - json!({ - "type": "error", - "error": { - "type": "invalid_request_error", - "message": "route has no Anthropic target for token counting" - } - }) - ); - Ok(()) -} - #[tokio::test] async fn routes_dispatch_and_discovery_endpoints_are_stable() -> TestResult { let (upstream, app) = test_app(&[ @@ -3069,55 +3218,6 @@ async fn advisor_route_routing_log_records_classifier_tier() -> TestResult { Ok(()) } -#[tokio::test] -async fn advisor_route_count_tokens_uses_executor() -> TestResult { - let upstream = MockUpstream::start().await?; - let state = load_test_config(&format!( - r#" -schema_version = 1 - -[llm_clients.claude] -format = "anthropic_messages" -base_url = "{base_url}" - -[targets.executor] -id = "model/executor" -llm_client = "claude" - -[targets.advisor] -id = "model/advisor" -llm_client = "claude" - -[routes.gated] -id = "switchyard/advisor" -type = "advisor" -executor_target = "executor" -advisor_target = "advisor" -"#, - base_url = upstream.base_url - ))?; - let app = build_switchyard_router(state); - - let response = send( - &app, - "POST", - "/v1/messages/count_tokens", - Some(json!({ - "model": "switchyard/advisor", - "messages": [{"role": "user", "content": "hi"}] - })), - ) - .await?; - assert_eq!(response.status, StatusCode::OK); - assert_eq!(response.json()?["input_tokens"], 7); - // The executor is the route's only completion target, so it backs - // count_tokens; the judge-only advisor never does. - let calls = upstream.calls.lock().await; - assert_eq!(calls.len(), 1); - assert_eq!(calls[0]["model"], "model/executor"); - Ok(()) -} - /// An advisor deployment whose reviewer client never retries, so a down /// advisor hits fail-open after a single attempt (the documented deployment /// posture for the advisor tier). diff --git a/docs/reference/toml_schema.md b/docs/reference/toml_schema.md index f4f2f7742..966edd978 100644 --- a/docs/reference/toml_schema.md +++ b/docs/reference/toml_schema.md @@ -32,6 +32,13 @@ target = "strong" `schema_version` must be `1`. Table names under `llm_clients`, `targets`, and `routes` are local references; clients send the route's `id` as the model name. +The optional top-level `fallback_client` names an entry under `[llm_clients]`. When set, any HTTP +method and path not implemented by Switchyard is forwarded through that client without translating +the path, query string, body, response, or model identifier. The fallback client does not need a +target. Only its `base_url` is used: caller end-to-end headers are forwarded, hop-by-hop headers are +removed, and configured API keys, extra headers, format, and retries are not applied. When omitted, +unmatched paths return `404`. + `schema_version`, `[targets]`, and `[routes]` must all be present, even when a route reaches no upstream. A file without a `[targets]` table is rejected with `missing field targets`; an empty `[targets]` table satisfies it.