diff --git a/crates/libsy-llm-client/src/backend.rs b/crates/libsy-llm-client/src/backend.rs index 95de23411..e275d02f0 100644 --- a/crates/libsy-llm-client/src/backend.rs +++ b/crates/libsy-llm-client/src/backend.rs @@ -3,7 +3,7 @@ //! Per-provider backend configuration: wire format, upstream URL, and auth. -use std::{collections::BTreeMap, fmt}; +use std::{collections::BTreeMap, fmt, time::Duration}; use reqwest::RequestBuilder; use serde_json::Value; @@ -35,6 +35,19 @@ const ANTHROPIC_OVERFLOW_PHRASES: &[&str] = &[ "context length", ]; +// Phrases marking a request the model cannot serve at all, whatever its size: +// multimodal content sent to a text-only deployment, or a server started without +// its multimodal projector. The wording comes from the serving stack rather than +// a provider error envelope, and no provider assigns it a structured `error.code`, +// so one phrase list covers every backend variant. +const CAPABILITY_REJECT_PHRASES: &[&str] = &[ + "mmproj", + "image input", + "does not support image", + "does not support multimodal", + "no multimodal support", +]; + /// Shared HTTP configuration for one upstream backend. #[derive(Clone)] pub struct HttpBackendConfig { @@ -48,6 +61,8 @@ pub struct HttpBackendConfig { pub extra_body: BTreeMap, /// Additional attempts after the initial upstream request. pub max_retries: u32, + /// Per-attempt request timeout in seconds. `None` leaves the request unbounded. + pub timeout_secs: Option, } impl fmt::Debug for HttpBackendConfig { @@ -58,6 +73,7 @@ impl fmt::Debug for HttpBackendConfig { .field("extra_headers", &self.extra_headers) .field("extra_body_keys", &self.extra_body.keys()) .field("max_retries", &self.max_retries) + .field("timeout_secs", &self.timeout_secs) .finish() } } @@ -145,6 +161,23 @@ impl Backend { self.config().max_retries } + /// Per-attempt request timeout, when a usable one is configured. + /// + /// Applies to each attempt rather than the call as a whole, so a call that + /// exhausts its retry budget can take up to `(max_retries + 1)` times this. + /// + /// Validated here rather than trusted from the field: [`HttpBackendConfig`] is + /// public and constructible directly, so the server's config check is not the only + /// way a value can arrive. `Duration::from_secs_f64` panics on negative, NaN, and + /// infinite input, and a malformed timeout must not be able to abort a request — an + /// unusable value is therefore treated as no timeout, matching an omitted field. + pub fn timeout(&self) -> Option { + self.config() + .timeout_secs + .filter(|seconds| seconds.is_finite() && *seconds > 0.0) + .map(Duration::from_secs_f64) + } + /// Whether this backend speaks the Anthropic Messages wire format — the only /// one with a `count_tokens` endpoint. pub fn is_anthropic(&self) -> bool { @@ -176,6 +209,16 @@ impl Backend { Backend::Anthropic(_) => is_overflow_body(body, |_| false, ANTHROPIC_OVERFLOW_PHRASES), } } + + /// Whether an upstream 400 `body` says the model cannot serve this request at + /// all — as opposed to it merely being too large. + /// + /// Provider-independent: the rejection is emitted by the serving stack, so the + /// same phrase list applies to every backend variant and there is no structured + /// check to short-circuit on. + pub(crate) fn is_capability_reject(&self, body: &str) -> bool { + is_overflow_body(body, |_| false, CAPABILITY_REJECT_PHRASES) + } } // Accept either a root `/v1` URL or an already-specific OpenAI endpoint URL. @@ -209,6 +252,7 @@ mod tests { extra_headers: BTreeMap::new(), extra_body: BTreeMap::new(), max_retries: 0, + timeout_secs: None, } } @@ -275,6 +319,28 @@ mod tests { ); } + #[test] + fn an_unusable_timeout_is_treated_as_no_timeout() { + // `HttpBackendConfig` is public, so the server's validation is not the only way + // a value arrives. None of these may panic a request. + for seconds in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY, -1.0, 0.0] { + let mut inner = config("x"); + inner.timeout_secs = Some(seconds); + assert_eq!( + Backend::OpenAiChat(inner).timeout(), + None, + "{seconds} must not configure a timeout" + ); + } + + let mut inner = config("x"); + inner.timeout_secs = Some(1.5); + assert_eq!( + Backend::OpenAiChat(inner).timeout(), + Some(Duration::from_millis(1_500)) + ); + } + #[test] fn only_anthropic_backend_is_anthropic() { assert!(Backend::Anthropic(config("x")).is_anthropic()); @@ -327,4 +393,36 @@ mod tests { ); assert!(!backend.is_context_overflow(r#"{"error":{"message":"overloaded"}}"#)); } + + #[test] + fn detects_capability_reject_across_backends() { + // Provider-independent: the serving stack emits it, so every variant matches. + for backend in [ + Backend::OpenAiChat(config("x")), + Backend::OpenAiResponses(config("x")), + Backend::Anthropic(config("x")), + ] { + assert!(backend.is_capability_reject( + r#"{"error":{"message":"image input is not supported by this model"}}"# + )); + assert!(backend.is_capability_reject( + r#"{"error":{"message":"server was started without an mmproj file"}}"# + )); + // Plain-text bodies from proxies still classify. + assert!(backend.is_capability_reject("this model does not support image content")); + } + } + + #[test] + fn capability_reject_and_overflow_do_not_overlap() { + let backend = Backend::OpenAiChat(config("x")); + // An overflow is not a capability reject: a smaller request can still succeed. + let overflow = r#"{"error":{"code":"context_length_exceeded","message":"too long"}}"#; + assert!(backend.is_context_overflow(overflow)); + assert!(!backend.is_capability_reject(overflow)); + // And unrelated failures are neither. + let rate_limit = r#"{"error":{"message":"rate limit exceeded"}}"#; + assert!(!backend.is_context_overflow(rate_limit)); + assert!(!backend.is_capability_reject(rate_limit)); + } } diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 08ff14497..beb04be54 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -48,6 +48,29 @@ const RESERVED_HEADERS: &[&str] = &[ "accept-encoding", ]; +// Body phrases marking a 4xx as a rejection of a request *parameter* rather than of +// the request's content. A server that refuses an unknown top-level field names it +// one of these ways; the reasoning-control knobs are listed explicitly because they +// are the fields operators most often configure per target. Matched against the raw +// body, which covers both JSON error envelopes and the plain-text bodies some +// proxies return. +const PARAM_REJECT_PHRASES: &[&str] = &[ + "chat_template_kwargs", + "enable_thinking", + "unexpected keyword", + "unknown parameter", + "unknown field", + "unrecognized", + "extra_forbidden", + "additional properties", + "extra fields not permitted", + "invalid parameter", +]; + +// Statuses that can mean "request malformed / unknown parameter". A 429 or 5xx is a +// transient fault and must never be treated as a parameter reject. +const PARAM_REJECT_STATUSES: &[u16] = &[400, 422]; + const INITIAL_RETRY_DELAY: Duration = Duration::from_millis(250); const MAX_RETRY_BACKOFF: Duration = Duration::from_secs(2); const MAX_RETRY_AFTER: Duration = Duration::from_secs(60); @@ -205,7 +228,10 @@ impl TranslatingLlmClient { strip_anthropic_incompatible_fields(&mut body); strip_unsigned_thinking_blocks(&mut body); } - merge_extra_body(&mut body, backend.extra_body()); + let injected_extra_body = merge_extra_body(&mut body, backend.extra_body()); + // Set once the injected defaults have been dropped, so the strip is attempted + // at most once per call. + let mut stripped_extra_body = false; if matches!(backend, Backend::Anthropic(_)) { enable_anthropic_prompt_caching(&mut body); } @@ -248,6 +274,24 @@ impl TranslatingLlmClient { return Ok(response); } Err(failure) => { + // The upstream refused a parameter, and this client injected some: + // drop exactly those and try again. This does not consume the retry + // budget — that budget exists for transient faults, and a request + // without an optional default is a different request, not a repeat + // of a failed one. Guarded so it happens at most once. + if !stripped_extra_body + && !injected_extra_body.is_empty() + && failure.is_param_reject() + { + stripped_extra_body = true; + remove_body_keys(&mut body, &injected_extra_body); + span.record("outcome", "error"); + if let Some(status) = failure.status { + span.record("status_code", status); + } + span.record("will_retry", true); + continue; + } let will_retry = attempt < max_retries && failure.is_retryable(); span.record("outcome", "error"); if let Some(status) = failure.status { @@ -279,7 +323,13 @@ impl TranslatingLlmClient { model: &str, streaming: bool, ) -> std::result::Result { - let builder = self.client.post(url).json(body); + let mut builder = self.client.post(url).json(body); + // Per attempt, not per call: the retry loop re-enters here, so each try gets + // its own budget. Without this the shared client has no timeout at all and a + // hung upstream holds the request open indefinitely. + if let Some(timeout) = backend.timeout() { + builder = builder.timeout(timeout); + } let builder = forward_metadata_headers(builder, metadata); let builder = apply_extra_headers(builder, backend); let builder = backend.apply_auth(builder); @@ -333,18 +383,27 @@ impl TranslatingLlmClient { } }; metrics::record_upstream_attempt(Some(status.as_u16())); - let error = - if status == reqwest::StatusCode::BAD_REQUEST && backend.is_context_overflow(&body) { - LlmClientError::ContextWindowExceeded { - model: model.to_string(), - message: body, - } - } else { - LlmClientError::UpstreamHttp { - status: status.as_u16(), - body, - } - }; + let error = if status == reqwest::StatusCode::BAD_REQUEST + && backend.is_context_overflow(&body) + { + LlmClientError::ContextWindowExceeded { + model: model.to_string(), + message: body, + } + // Checked after overflow: an oversized request is recoverable on the same + // target, a capability reject never is. + } else if status == reqwest::StatusCode::BAD_REQUEST && backend.is_capability_reject(&body) + { + LlmClientError::CapabilityRejected { + model: model.to_string(), + message: body, + } + } else { + LlmClientError::UpstreamHttp { + status: status.as_u16(), + body, + } + }; Err(AttemptFailure { error, status: Some(status.as_u16()), @@ -553,6 +612,26 @@ struct AttemptFailure { } impl AttemptFailure { + /// Whether this failure is the upstream refusing a request *parameter*, so the + /// call may be retried once without the fields the client injected. + /// + /// Deliberately narrow: only a 400/422 whose body names a parameter qualifies. + /// A rate limit or server error must surface as-is — stripping a field and + /// re-firing on a 429 would double the load and mask the real failure. Content + /// rejections (context overflow) carry their own variant and never match here. + fn is_param_reject(&self) -> bool { + let LlmClientError::UpstreamHttp { status, body } = &self.error else { + return false; + }; + if !PARAM_REJECT_STATUSES.contains(status) { + return false; + } + let lowered = body.to_ascii_lowercase(); + PARAM_REJECT_PHRASES + .iter() + .any(|phrase| lowered.contains(phrase)) + } + fn is_retryable(&self) -> bool { match &self.error { LlmClientError::Transport { .. } | LlmClientError::Timeout { .. } => true, @@ -726,12 +805,29 @@ fn is_unsigned_thinking_block(block: &Value) -> bool { } // Applies target defaults without overriding fields supplied by the caller. -fn merge_extra_body(body: &mut Value, extra_body: &BTreeMap) { +// Returns the keys actually injected, so a parameter reject can drop exactly those +// and leave the caller's own fields alone. +fn merge_extra_body(body: &mut Value, extra_body: &BTreeMap) -> Vec { let Value::Object(object) = body else { - return; + return Vec::new(); }; + let mut injected = Vec::new(); for (key, value) in extra_body { - object.entry(key.clone()).or_insert_with(|| value.clone()); + if !object.contains_key(key) { + object.insert(key.clone(), value.clone()); + injected.push(key.clone()); + } + } + injected +} + +// Removes previously injected default fields from an encoded body. +fn remove_body_keys(body: &mut Value, keys: &[String]) { + let Value::Object(object) = body else { + return; + }; + for key in keys { + object.remove(key); } } @@ -818,12 +914,14 @@ mod tests { extra_headers: BTreeMap::new(), extra_body: BTreeMap::new(), max_retries: 0, + timeout_secs: None, } } fn config_with_retries(base_url: &str, max_retries: u32) -> HttpBackendConfig { HttpBackendConfig { max_retries, + timeout_secs: None, ..config(base_url) } } @@ -1598,6 +1696,207 @@ mod tests { Ok(()) } + #[tokio::test] + async fn a_configured_timeout_bounds_each_attempt() + -> std::result::Result<(), Box> { + let server = MockServer::start().await; + // Answers far later than the configured timeout, so only the timeout can end it. + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with( + ResponseTemplate::new(200) + .set_delay(Duration::from_secs(30)) + .set_body_json(json!({"id": "1", "model": "gpt", "choices": [], "usage": {}})), + ) + .mount(&server) + .await; + + let mut backend = config(&format!("{}/v1", server.uri())); + backend.timeout_secs = Some(0.05); + backend.max_retries = 0; + let client = TranslatingLlmClient::new(&[ModelConfig::new( + "gpt", + Backend::OpenAiChat(backend), + None, + )])?; + + let error = client + .call_rewrite_model_raw( + json!({"model": "gpt", "messages": [{"role": "user", "content": "hi"}]}), + None, + Some("gpt"), + WireFormat::OpenAiChat, + ) + .await + .err() + .ok_or("expected the call to time out")?; + + assert!( + matches!(error, LlmClientError::Timeout { .. }), + "expected a timeout, got {error:?}" + ); + Ok(()) + } + + #[test] + fn param_reject_classification_is_narrow() { + let failure = |status: u16, body: &str| AttemptFailure { + error: LlmClientError::UpstreamHttp { + status, + body: body.to_string(), + }, + status: Some(status), + retry_after: None, + }; + + // A 400/422 naming a parameter is the whole of the class. + for status in [400, 422] { + assert!( + failure( + status, + r#"{"error":{"message":"unknown parameter: enable_thinking"}}"# + ) + .is_param_reject(), + "HTTP {status} naming a parameter should qualify" + ); + } + assert!(failure(400, "extra fields not permitted").is_param_reject()); + + // Transient faults must never be treated as a parameter reject, whatever + // their body says — stripping and re-firing would double the load. + for status in [429, 500, 503] { + assert!( + !failure(status, "unknown parameter").is_param_reject(), + "HTTP {status} is transient and must not qualify" + ); + } + // An unrelated 400 is not one either. + assert!(!failure(400, r#"{"error":{"message":"invalid api key"}}"#).is_param_reject()); + + // A content rejection carries its own variant and never matches. + let overflow = AttemptFailure { + error: LlmClientError::ContextWindowExceeded { + model: "gpt".to_string(), + message: "unknown parameter".to_string(), + }, + status: Some(400), + retry_after: None, + }; + assert!(!overflow.is_param_reject()); + } + + #[test] + fn merge_extra_body_reports_only_the_keys_it_injected() { + let mut body = json!({"model": "gpt", "max_tokens": 7}); + let extra = BTreeMap::from([ + ("max_tokens".to_string(), json!(999)), + ("enable_thinking".to_string(), json!(false)), + ]); + + let injected = merge_extra_body(&mut body, &extra); + + // The caller's own `max_tokens` is untouched and not reported. + assert_eq!(injected, vec!["enable_thinking".to_string()]); + assert_eq!(body.get("max_tokens"), Some(&json!(7))); + + // Stripping removes the injected default and leaves the caller's field. + remove_body_keys(&mut body, &injected); + assert!(body.get("enable_thinking").is_none()); + assert_eq!(body.get("max_tokens"), Some(&json!(7))); + } + + #[tokio::test] + async fn a_param_reject_retries_once_without_the_injected_defaults() + -> std::result::Result<(), Box> { + let server = MockServer::start().await; + // The upstream refuses the injected knob... + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .and(wiremock::matchers::body_partial_json( + json!({"enable_thinking": false}), + )) + .respond_with(ResponseTemplate::new(400).set_body_json(json!({ + "error": {"message": "unknown parameter: enable_thinking"} + }))) + .mount(&server) + .await; + // ...and serves the same request once the knob is gone. + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "id": "1", + "model": "gpt", + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "ok"}, + "finish_reason": "stop" + }], + "usage": {} + }))) + .mount(&server) + .await; + + let extra_body = BTreeMap::from([("enable_thinking".to_string(), json!(false))]); + let client = TranslatingLlmClient::new(&chat_map_with_extra_body( + &format!("{}/v1", server.uri()), + extra_body, + ))?; + + // Succeeds despite max_retries = 0: the strip is not a budgeted retry. + client + .call_rewrite_model_raw( + json!({ + "model": "client-facing", + "messages": [{"role": "user", "content": "hi"}] + }), + None, + Some("gpt"), + WireFormat::OpenAiChat, + ) + .await?; + Ok(()) + } + + #[tokio::test] + async fn a_persistent_param_reject_surfaces_rather_than_looping() + -> std::result::Result<(), Box> { + let server = MockServer::start().await; + // Rejects every request, knob or no knob. + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(ResponseTemplate::new(400).set_body_json(json!({ + "error": {"message": "unknown parameter: enable_thinking"} + }))) + .mount(&server) + .await; + + let extra_body = BTreeMap::from([("enable_thinking".to_string(), json!(false))]); + let client = TranslatingLlmClient::new(&chat_map_with_extra_body( + &format!("{}/v1", server.uri()), + extra_body, + ))?; + + let result = client + .call_rewrite_model_raw( + json!({ + "model": "client-facing", + "messages": [{"role": "user", "content": "hi"}] + }), + None, + Some("gpt"), + WireFormat::OpenAiChat, + ) + .await; + + // The strip is attempted at most once, so the second rejection is returned. + let error = result.err().ok_or("expected the call to fail")?; + assert!( + matches!(error, LlmClientError::UpstreamHttp { status: 400, .. }), + "expected the upstream 400 to surface, got {error:?}" + ); + Ok(()) + } + #[test] fn retryable_error_classes_are_explicit() { let transport = AttemptFailure { diff --git a/crates/libsy-llm-client/src/observability.rs b/crates/libsy-llm-client/src/observability.rs index 58270cc03..97287dabb 100644 --- a/crates/libsy-llm-client/src/observability.rs +++ b/crates/libsy-llm-client/src/observability.rs @@ -193,6 +193,7 @@ fn llm_client_error_type(error: &LlmClientError) -> Cow<'static, str> { LlmClientError::Transport { .. } => Cow::Borrowed("transport"), LlmClientError::Timeout { .. } => Cow::Borrowed("timeout"), LlmClientError::ContextWindowExceeded { .. } => Cow::Borrowed("context_window_exceeded"), + LlmClientError::CapabilityRejected { .. } => Cow::Borrowed("capability_rejected"), LlmClientError::UpstreamHttp { status, .. } => Cow::Owned(status.to_string()), LlmClientError::InvalidResponse { .. } => Cow::Borrowed("invalid_response"), LlmClientError::Ffi { .. } => Cow::Borrowed("ffi"), diff --git a/crates/libsy/src/algorithms/fall_through.rs b/crates/libsy/src/algorithms/fall_through.rs index 2f172560a..c3ae4123e 100644 --- a/crates/libsy/src/algorithms/fall_through.rs +++ b/crates/libsy/src/algorithms/fall_through.rs @@ -13,8 +13,9 @@ //! private state value across turns with the same session ID. Requests without a session ID use //! unretained per-run state. //! -//! Every composition retains one thing regardless: a target that overflows its context window is -//! remembered for the rest of its session and skipped on later turns. +//! Every composition retains one thing regardless: a target that overflows its context window, +//! or that rejects the request as one it cannot serve at all, is remembered for the rest of its +//! session and skipped on later turns. //! An unavailable target is skipped only for the current request. use std::collections::HashSet; @@ -264,6 +265,7 @@ where ) -> Decision { let failure = match reason { RoutingFallbackReason::ContextWindow => "exceeded its context window", + RoutingFallbackReason::Capability => "cannot serve the request", RoutingFallbackReason::Unavailable => "was unavailable", }; Decision::new( @@ -801,6 +803,75 @@ mod tests { Ok(()) } + /// Rejects the named `rejecting` targets as unable to serve the request at all and + /// echoes for the rest. The capability counterpart of `overflowing`. + fn capability_rejecting( + rejecting: &'static [&'static str], + calls: Arc>>, + ) -> impl Serve { + move |decision: Decision, _request: Request| { + let calls = Arc::clone(&calls); + async move { + let model = decision.selected_model_id().to_string(); + calls.lock().push(model.clone()); + if rejecting.contains(&model.as_str()) { + return Err(LlmClientError::CapabilityRejected { + model, + message: "image input is not supported".to_string(), + }); + } + Ok(reply(model)) + } + } + } + + #[tokio::test] + async fn a_capability_reject_falls_forward_and_evicts_like_an_overflow() -> Result<()> { + let calls = Arc::new(Mutex::new(Vec::new())); + let router = Arc::new( + FallThrough::<()>::new(target_set(&["weak", "strong"])) + .with_classifier(fixed(vec![score("weak", 0.9)])), + ); + for _ in 0..3 { + let serve = capability_rejecting(&["weak"], calls.clone()); + assert_eq!(run_turn(&router, serve).await?.0, "strong"); + } + // Evicted after the first rejection: a capability reject is a property of the + // target, so later turns must not re-probe it. + assert_eq!(calls.lock().iter().filter(|m| *m == "weak").count(), 1); + Ok(()) + } + + #[tokio::test] + async fn a_capability_reject_surfaces_once_the_pool_is_spent() -> Result<()> { + let calls = Arc::new(Mutex::new(Vec::new())); + let router = Arc::new( + FallThrough::<()>::new(target_set(&["weak", "strong"])) + .with_classifier(fixed(vec![score("weak", 0.9)])), + ); + // No target can serve it, so the caller sees the upstream rejection itself + // rather than an internal routing failure. + let serve = capability_rejecting(&["weak", "strong"], calls); + let error = + run_turn(&router, serve) + .await + .err() + .ok_or_else(|| LibsyError::AlgorithmError { + message: "expected the call to fail, but it succeeded".to_string(), + })?; + assert!( + matches!( + error, + LibsyError::ClientCall { + source: LlmClientError::CapabilityRejected { .. }, + .. + } + ), + "expected CapabilityRejected, got {error:?}" + ); + Ok(()) + } + #[tokio::test] async fn a_different_session_starts_with_an_empty_eviction_set() -> Result<()> { let calls = Arc::new(Mutex::new(Vec::new())); diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 34a33fee1..636bfd2d4 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -464,6 +464,7 @@ fn classify_fallback(error: &LibsyError) -> Option<(&str, RoutingFallbackReason) }; let reason = match source { LlmClientError::ContextWindowExceeded { .. } => RoutingFallbackReason::ContextWindow, + LlmClientError::CapabilityRejected { .. } => RoutingFallbackReason::Capability, LlmClientError::Transport { .. } | LlmClientError::Timeout { .. } => { RoutingFallbackReason::Unavailable } @@ -509,7 +510,13 @@ pub(crate) async fn call_model_with_fallback( return Err(error); } match reason { - RoutingFallbackReason::ContextWindow => evictions.record(identity, failed), + // Both are permanent properties of the target for this conversation: an + // overflow recurs as history grows, and a capability reject never clears. + // Recording the eviction keeps the target out of later turns for the same + // identity, rather than re-probing it every turn. + RoutingFallbackReason::ContextWindow | RoutingFallbackReason::Capability => { + evictions.record(identity, failed) + } RoutingFallbackReason::Unavailable => target_unavailable(&request, failed), } let Ok(next) = targets.resolve_target(&target.semantic_name, excluded) else { diff --git a/crates/protocol/src/client.rs b/crates/protocol/src/client.rs index cf137e84f..596fa471e 100644 --- a/crates/protocol/src/client.rs +++ b/crates/protocol/src/client.rs @@ -77,6 +77,19 @@ pub enum LlmClientError { message: String, }, + /// The upstream rejected the request because the model cannot serve it at all — + /// multimodal content sent to a text-only deployment, a server built without its + /// multimodal projector, and similar. Unlike [`Self::ContextWindowExceeded`] this is + /// not a function of request size, so shrinking the request cannot recover it; a + /// routing host's only remedy is a different target. + #[error("model {model} cannot serve this request: {message}")] + CapabilityRejected { + /// Model that rejected the request. + model: String, + /// Upstream error message. + message: String, + }, + /// The upstream returned a non-success HTTP response. #[error("upstream returned HTTP {status}: {body}")] UpstreamHttp { @@ -113,6 +126,9 @@ pub enum LlmClientError { pub enum RoutingFallbackReason { /// The selected target rejected the request because its context window was too small. ContextWindow, + /// The selected target rejected the request as one it cannot serve at all, such as an + /// unsupported modality or tool. Unlike an overflow, no smaller request would succeed. + Capability, /// The selected target was unavailable after its client retries finished. Unavailable, } @@ -122,6 +138,7 @@ impl RoutingFallbackReason { pub const fn as_str(self) -> &'static str { match self { Self::ContextWindow => "context_window", + Self::Capability => "capability", Self::Unavailable => "unavailable", } } diff --git a/crates/switchyard-server/src/config.rs b/crates/switchyard-server/src/config.rs index b644d4a04..e4fd032de 100644 --- a/crates/switchyard-server/src/config.rs +++ b/crates/switchyard-server/src/config.rs @@ -244,6 +244,9 @@ struct LlmClientConfig { extra_headers: BTreeMap, #[serde(default = "default_max_retries")] max_retries: u32, + /// Per-attempt request timeout in seconds. Omit to leave requests unbounded. + #[serde(default)] + timeout_secs: Option, } #[derive(Debug, Deserialize)] @@ -773,6 +776,13 @@ fn build_backend( "llm client {client_name} base_url must not be empty" ))); } + if let Some(timeout_secs) = config.timeout_secs + && (!timeout_secs.is_finite() || timeout_secs <= 0.0) + { + return Err(ServerError::new(format!( + "llm client {client_name} timeout_secs must be finite and positive, got {timeout_secs}" + ))); + } if config.max_retries > MAX_CONFIGURED_RETRIES { return Err(ServerError::new(format!( "llm client {client_name} max_retries must be at most {MAX_CONFIGURED_RETRIES}" @@ -806,6 +816,7 @@ fn build_backend( extra_headers: config.extra_headers.clone(), extra_body: extra_body.clone(), max_retries: config.max_retries, + timeout_secs: config.timeout_secs, }; Ok(match config.format { ClientFormat::OpenAiChat => Backend::OpenAiChat(http), diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 1f69e0b02..44221d4d4 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -861,6 +861,14 @@ fn client_error(error: &LlmClientError) -> Response { "invalid_request_error", "context_length_exceeded", ), + // The pool is spent: every eligible target refused the request as unservable, + // which is a client-side problem with the request, not an upstream fault. + LlmClientError::CapabilityRejected { message, .. } => error_response( + StatusCode::BAD_REQUEST, + message, + "invalid_request_error", + "unsupported_content", + ), LlmClientError::UpstreamHttp { status, body } => error_response( StatusCode::from_u16(*status).unwrap_or(StatusCode::BAD_GATEWAY), upstream_error_message(body), diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index 0e3272cb0..3f67e5a0d 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -214,6 +214,7 @@ fn random_state(base_url: &str, routes: &[(&str, &[&str])]) -> TestResult