diff --git a/Cargo.lock b/Cargo.lock index 225aae1d8..33077d064 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1808,7 +1808,6 @@ dependencies = [ "serde_json", "switchyard-protocol", "switchyard-translation", - "thiserror", "tokio", "wiremock", ] @@ -1821,6 +1820,7 @@ dependencies = [ "futures", "serde", "serde_json", + "thiserror", ] [[package]] diff --git a/crates/libsy-llm-client/Cargo.toml b/crates/libsy-llm-client/Cargo.toml index 67c338931..62a8c012f 100644 --- a/crates/libsy-llm-client/Cargo.toml +++ b/crates/libsy-llm-client/Cargo.toml @@ -18,7 +18,6 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus async-trait = "0.1" futures-util = "0.3" serde_json = "1" -thiserror = "2" [dev-dependencies] futures = "0.3" diff --git a/crates/libsy-llm-client/README.md b/crates/libsy-llm-client/README.md index e769123f6..0465dad1e 100644 --- a/crates/libsy-llm-client/README.md +++ b/crates/libsy-llm-client/README.md @@ -88,9 +88,9 @@ async fn ask(client: &TranslatingLlmClient) -> switchyard_llm_client::Result Ok(completion_text(&agg)), - LlmResponse::Stream(_) => Err(LlmClientError::Stream( - "expected a buffered response".to_string(), - )), + LlmResponse::Stream(_) => Err(LlmClientError::InvalidResponse { + source: "expected a buffered response".into(), + }), } } ``` @@ -101,10 +101,12 @@ Set `stream` on the IR request and drive the returned chunk stream: ```rust use futures_util::StreamExt; -use switchyard_llm_client::{LlmClientError, TranslatingLlmClient}; +use switchyard_llm_client::TranslatingLlmClient; use switchyard_protocol::{text_request, Context, LlmResponse, LlmResponseChunk, Request}; -async fn stream(client: &TranslatingLlmClient) -> switchyard_llm_client::Result<()> { +async fn stream( + client: &TranslatingLlmClient, +) -> Result<(), Box> { let mut llm_request = text_request(None, "Count to five."); llm_request.stream = true; let request = Request { llm_request, raw_request: None, metadata: None }; @@ -118,7 +120,7 @@ async fn stream(client: &TranslatingLlmClient) -> switchyard_llm_client::Result< match item { Ok(LlmResponseChunk::TextDelta { text, .. }) => print!("{text}"), Ok(_) => {} // usage, tool-call deltas, message start/stop - Err(error) => return Err(LlmClientError::Stream(error.to_string())), + Err(error) => return Err(error), } } } @@ -174,14 +176,17 @@ fn build_multi_format_client( | Variant | When | |---------|------| -| `MissingModel` | no `model_name` arg and no `request.llm_request.model` | -| `UnknownModel(model)` | model has no backends configured | -| `UnknownModelFormat { model, format }` | model has no backend for that format | -| `Translation(msg)` | encode/decode failed in the translation engine | -| `Transport(msg)` | connect/timeout/transport failure | +| `InvalidRequest { message }` | the request does not identify a model | +| `Configuration { message }` | the model or requested wire format has no configured backend | +| `RequestTranslation(msg)` | decoding the inbound request failed in the translation engine | +| `RequestEncoding(msg)` | re-encoding an already-decoded request to the wire format failed (internal fault) | +| `ResponseTranslation(msg)` | response decoding or encoding failed in the translation engine | +| `Timeout { source }` | request or response body read exceeded its timeout | +| `Transport { source }` | non-timeout connection or transport failure | | `ContextWindowExceeded { model, message }` | upstream 400 detected as a context overflow (checked before `UpstreamHttp`, so callers can evict-and-retry) | | `UpstreamHttp { status, body }` | any other non-2xx upstream response | -| `Stream(msg)` | mid-stream read / malformed frame | +| `InvalidResponse { source }` | the upstream response could not be decoded | +| `Other(source)` | a client-specific failure outside the shared categories | [`switchyard_protocol::Request`]: ../libsy-protocol [`switchyard_protocol::Response`]: ../libsy-protocol @@ -190,4 +195,4 @@ fn build_multi_format_client( [`HttpBackendConfig`]: src/backend.rs [`ModelConfig`]: src/client.rs [`TranslatingLlmClient::call_rewrite_model`]: src/client.rs -[`LlmClientError`]: src/error.rs +[`LlmClientError`]: ../protocol/src/client.rs diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 73b20d4dd..bf73cb7d4 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -84,9 +84,12 @@ impl TranslatingLlmClient { /// Builds a client over the given [`ModelConfig`]s, with a fresh shared HTTP /// client and the built-in translation codecs. pub fn new(model_configs: &[ModelConfig]) -> Result { - let client = reqwest::Client::builder().build().map_err(|error| { - LlmClientError::Transport(format!("failed to build HTTP client: {error}")) - })?; + let client = + reqwest::Client::builder() + .build() + .map_err(|error| LlmClientError::Transport { + source: Box::new(error), + })?; let model_to_config = model_configs .iter() .map(|config| (config.model_name.clone(), config.clone())) @@ -120,10 +123,8 @@ impl TranslatingLlmClient { /// /// Resolution: `model_name` wins over `request.llm_request.model`; the /// resolved name is both the outer map key and the model id written into the - /// request before translation. Errors with [`LlmClientError::MissingModel`] - /// when neither is set, [`LlmClientError::UnknownModel`] when the model has - /// no backends, and [`LlmClientError::UnknownModelFormat`] when the model has - /// no backend for `format`. + /// request before translation. Missing models are invalid requests; unknown + /// models or wire formats are configuration errors. pub async fn call_rewrite_model( &self, _ctx: Context, @@ -141,27 +142,30 @@ impl TranslatingLlmClient { let model = model_name .map(str::to_string) .or_else(|| llm_request.model.clone()) - .ok_or(LlmClientError::MissingModel)?; + .ok_or_else(|| LlmClientError::InvalidRequest { + message: "no model given".to_string(), + })?; let orig_format = metadata.as_ref().and_then(|m| m.wire_format); let wire_format = orig_format.unwrap_or( self.model_to_config .get(&model) .map(|config| config.default_backend.wire_format()) - .ok_or(LlmClientError::UnknownModel(model.clone()))?, + .ok_or_else(|| LlmClientError::Configuration { + message: format!("no backend configured for model {model:?}"), + })?, ); let backend = self.backend_for(&model, wire_format) - .ok_or(LlmClientError::UnknownModelFormat { - model: model.clone(), - format: wire_format, + .ok_or_else(|| LlmClientError::Configuration { + message: format!("model {model:?} has no backend for format {wire_format}"), })?; // The resolved name is the upstream model id (per the crate contract). llm_request.model = Some(model.clone()); let mut body = encode_request(&llm_request, wire_format) - .map_err(|error| LlmClientError::Translation(error.to_string()))?; + .map_err(|error| LlmClientError::RequestEncoding(error.to_string()))?; // `encode_request` round-trips a preserved same-format body verbatim, // which keeps the caller's original `model`; force the resolved model so // the upstream always sees the target id. @@ -173,16 +177,18 @@ impl TranslatingLlmClient { let builder = apply_extra_headers(builder, backend); let builder = backend.apply_auth(builder); - let http_response = builder - .send() - .await - .map_err(|error| LlmClientError::Transport(error.to_string()))?; + let http_response = builder.send().await.map_err(convert_reqwest_error)?; let status = http_response.status(); if !status.is_success() { - let body = http_response - .text() - .await - .unwrap_or_else(|error| format!("")); + let body = match http_response.text().await { + Ok(body) => body, + Err(error) if error.is_timeout() => { + return Err(LlmClientError::Timeout { + source: Box::new(error), + }) + } + Err(error) => format!(""), + }; if status == reqwest::StatusCode::BAD_REQUEST && backend.is_context_overflow(&body) { return Err(LlmClientError::ContextWindowExceeded { model, @@ -200,20 +206,28 @@ impl TranslatingLlmClient { // transport-agnostic and lives in `switchyard-translation`. let bytes = http_response.bytes_stream().map(|chunk| { chunk.map(|bytes| bytes.to_vec()).map_err(|error| { - Box::new(LlmClientError::Stream(format!( - "stream read failed: {error}" - ))) as Box + let error = if error.is_timeout() { + LlmClientError::Timeout { + source: Box::new(error), + } + } else { + LlmClientError::Transport { + source: Box::new(error), + } + }; + Box::new(error) as Box }) }); let chunks = decode_stream(bytes, wire_format) - .map_err(|error| LlmClientError::Stream(error.to_string()))?; + .map_err(|source| LlmClientError::InvalidResponse { source })?; LlmResponse::Stream(chunks) } else { - let body = http_response.json::().await.map_err(|error| { - LlmClientError::Transport(format!("invalid upstream JSON: {error}")) - })?; + let body = http_response + .json::() + .await + .map_err(convert_reqwest_error)?; let agg = decode_aggregated_response(&body, wire_format) - .map_err(|error| LlmClientError::Translation(error.to_string()))?; + .map_err(|error| LlmClientError::ResponseTranslation(error.to_string()))?; LlmResponse::Agg(agg) }; @@ -245,7 +259,7 @@ impl TranslatingLlmClient { wire_format: WireFormat, ) -> Result { let llm_request = decode_request(wire_format, &raw_http_request) - .map_err(|error| LlmClientError::Translation(error.to_string()))?; + .map_err(|error| LlmClientError::RequestTranslation(error.to_string()))?; // The model the client asked for; restamped onto the response so it never // leaks the upstream id. let requested_model = llm_request.model.clone(); @@ -270,12 +284,12 @@ impl TranslatingLlmClient { LlmResponse::Agg(agg) => { let body = encode_aggregated_response(&agg, wire_format, requested_model.as_deref()) - .map_err(|error| LlmClientError::Translation(error.to_string()))?; + .map_err(|error| LlmClientError::ResponseTranslation(error.to_string()))?; Ok(RawResponse::Buffered(body)) } LlmResponse::Stream(chunks) => { let events = encode_stream(chunks, wire_format, requested_model) - .map_err(|error| LlmClientError::Stream(error.to_string()))?; + .map_err(|source| LlmClientError::InvalidResponse { source })?; Ok(RawResponse::Stream(events)) } } @@ -289,11 +303,31 @@ impl RoutedLlmClient for TranslatingLlmClient { ctx: Context, request: Request, decision: std::sync::Arc, - ) -> std::result::Result> { + ) -> Result { let model_name = Some(decision.selected_model()); - self.call_rewrite_model(ctx, request, model_name) - .await - .map_err(|error| Box::new(error) as Box) + self.call_rewrite_model(ctx, request, model_name).await + } +} + +fn convert_reqwest_error(error: reqwest::Error) -> LlmClientError { + // `Response::json` labels body collection and JSON parsing failures as + // decode errors; only the latter is a translation failure. + if error.is_timeout() { + LlmClientError::Timeout { + source: Box::new(error), + } + } else if error.is_body() { + LlmClientError::Transport { + source: Box::new(error), + } + } else if error.is_decode() + && std::error::Error::source(&error).is_some_and(|source| source.is::()) + { + LlmClientError::ResponseTranslation(format!("invalid upstream JSON: {error}")) + } else { + LlmClientError::Transport { + source: Box::new(error), + } } } @@ -340,6 +374,8 @@ fn is_reserved_header(name: &str) -> bool { mod tests { use std::collections::BTreeMap; use std::error::Error; + use std::io::{Read, Write}; + use std::thread::JoinHandle; use serde_json::json; use switchyard_protocol::{completion_text, text_request, LlmRequest}; @@ -366,6 +402,26 @@ mod tests { )] } + fn truncated_response_server() -> std::io::Result<(String, JoinHandle>)> { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let address = listener.local_addr()?; + let handle = std::thread::spawn(move || { + let (mut stream, _) = listener.accept()?; + let mut request = [0_u8; 1024]; + if stream.read(&mut request)? == 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "client closed before sending a request", + )); + } + stream.write_all( + b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\ + Content-Length: 100\r\nConnection: close\r\n\r\n{}", + ) + }); + Ok((format!("http://{address}/v1"), handle)) + } + fn request_for(model: Option<&str>, stream: bool) -> Request { let mut llm_request = text_request(model.map(str::to_string), "hi"); llm_request.stream = stream; @@ -403,7 +459,10 @@ mod tests { else { panic!("expected an error"); }; - assert!(matches!(error, LlmClientError::MissingModel)); + assert!(matches!( + error, + LlmClientError::InvalidRequest { message } if message == "no model given" + )); Ok(()) } @@ -417,7 +476,11 @@ mod tests { else { panic!("expected an error"); }; - assert!(matches!(error, LlmClientError::UnknownModel(model) if model == "gpt")); + assert!(matches!( + error, + LlmClientError::Configuration { message } + if message.contains("gpt") + )); Ok(()) } @@ -438,8 +501,9 @@ mod tests { }; assert!(matches!( error, - LlmClientError::UnknownModelFormat { model, format } - if model == "gpt" && format == WireFormat::AnthropicMessages + LlmClientError::Configuration { message } + if message.contains("gpt") + && message.contains(&WireFormat::AnthropicMessages.to_string()) )); Ok(()) } @@ -470,7 +534,11 @@ mod tests { else { panic!("expected an error"); }; - assert!(matches!(error, LlmClientError::UnknownModel(model) if model == "b")); + assert!(matches!( + error, + LlmClientError::Configuration { message } + if message.contains("\"b\"") + )); Ok(()) } @@ -504,6 +572,58 @@ mod tests { Ok(()) } + #[tokio::test] + async fn invalid_json_is_a_response_translation_error( + ) -> std::result::Result<(), Box> { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_raw("not json", "application/json")) + .mount(&server) + .await; + + let client = TranslatingLlmClient::new(&chat_map(&format!("{}/v1", server.uri())))?; + let Err(error) = client + .call_rewrite_model(Context::default(), request_for(Some("gpt"), false), None) + .await + else { + panic!("expected invalid JSON to fail"); + }; + + assert!(matches!( + error, + LlmClientError::ResponseTranslation(message) + if message.contains("invalid upstream JSON") + )); + Ok(()) + } + + #[tokio::test] + async fn response_body_io_failure_is_a_transport_error( + ) -> std::result::Result<(), Box> { + let (base_url, server) = truncated_response_server()?; + let client = TranslatingLlmClient::new(&chat_map(&base_url))?; + let result = client + .call_rewrite_model(Context::default(), request_for(Some("gpt"), false), None) + .await; + server + .join() + .map_err(|_| std::io::Error::other("response server thread panicked"))??; + + let Err(error) = result else { + panic!("expected the truncated response body to fail"); + }; + let LlmClientError::Transport { source } = error else { + panic!("expected a transport error"); + }; + let Some(source) = source.downcast_ref::() else { + panic!("expected the reqwest transport source"); + }; + assert!(source.is_decode()); + assert!(!std::error::Error::source(&source) + .is_some_and(|source| source.is::())); + Ok(()) + } + #[tokio::test] async fn rewrites_model_to_resolved_upstream_id( ) -> std::result::Result<(), Box> { @@ -581,6 +701,39 @@ mod tests { Ok(()) } + #[tokio::test] + async fn routed_llm_client_exposes_timeout_variant( + ) -> std::result::Result<(), Box> { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with( + ResponseTemplate::new(200).set_delay(std::time::Duration::from_millis(100)), + ) + .mount(&server) + .await; + + let mut client = TranslatingLlmClient::new(&chat_map(&format!("{}/v1", server.uri())))?; + client.client = reqwest::Client::builder() + .timeout(std::time::Duration::from_millis(10)) + .build()?; + let decision: std::sync::Arc = std::sync::Arc::new(FixedDecision("gpt")); + + let Err(error) = client + .call(Context::default(), request_for(None, false), decision) + .await + else { + panic!("expected a timeout"); + }; + let LlmClientError::Timeout { source } = error else { + panic!("expected the protocol timeout variant"); + }; + let Some(source) = source.downcast_ref::() else { + panic!("expected the reqwest timeout source"); + }; + assert!(source.is_timeout()); + Ok(()) + } + #[tokio::test] async fn context_overflow_400_is_mapped( ) -> std::result::Result<(), Box> { @@ -701,6 +854,30 @@ mod tests { Ok(()) } + #[tokio::test] + async fn invalid_raw_request_is_a_request_translation_error( + ) -> std::result::Result<(), Box> { + let client = TranslatingLlmClient::new(&[])?; + let Err(error) = client + .call_rewrite_model_raw( + Context::default(), + json!("invalid"), + None, + Some("gpt"), + WireFormat::OpenAiChat, + ) + .await + else { + panic!("expected request translation to fail"); + }; + + assert!(matches!( + error, + LlmClientError::RequestTranslation(message) if !message.is_empty() + )); + Ok(()) + } + // Raw path, buffered: decode an OpenAI Chat body -> call -> encode back to OpenAI // Chat JSON, with the client-facing `model` restamped over the upstream id. #[tokio::test] diff --git a/crates/libsy-llm-client/src/error.rs b/crates/libsy-llm-client/src/error.rs index d02c0523c..64d524302 100644 --- a/crates/libsy-llm-client/src/error.rs +++ b/crates/libsy-llm-client/src/error.rs @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! Error type for the LLM client, plus shared context-window-overflow detection. +//! Canonical client error re-export and shared context-window-overflow detection. //! //! The overflow detection is ported from //! `switchyard-components/src/backends/context_overflow.rs`; that helper is @@ -9,67 +9,12 @@ //! so the small, self-contained logic is vendored here. use serde_json::Value; -use switchyard_translation::WireFormat; -use thiserror::Error; + +pub use switchyard_protocol::LlmClientError; /// Result alias for LLM client operations. pub type Result = std::result::Result; -/// Failures surfaced while resolving, translating, sending, or decoding a call. -#[derive(Debug, Error)] -pub enum LlmClientError { - /// The resolved model name has no entry in the client's backend map. - #[error("no backend configured for model {0:?}")] - UnknownModel(String), - - /// The model exists but has no backend configured for the requested format. - #[error("model {model:?} has no backend for format {format}")] - UnknownModelFormat { - /// Model that was resolved. - model: String, - /// Wire format requested for the call. - format: WireFormat, - }, - - /// Neither an explicit `model_name` nor a model on the request was given. - #[error("no model given: pass model_name or set request.llm_request.model")] - MissingModel, - - /// Request encoding or response decoding failed in the translation engine. - #[error("translation failed: {0}")] - Translation(String), - - /// The HTTP request could not be sent (connect/timeout/transport failure). - #[error("upstream transport error: {0}")] - Transport(String), - - /// An upstream 400 whose body is detected as a context-window overflow. - /// - /// Kept distinct from [`LlmClientError::UpstreamHttp`] and checked first so a - /// caller can implement evict-and-retry by matching this variant instead of - /// sniffing error strings. - #[error("context window exceeded for model {model}: {message}")] - ContextWindowExceeded { - /// Model that overflowed, as sent upstream. - model: String, - /// Raw upstream error body. - message: String, - }, - - /// A non-2xx upstream response that is not a context-window overflow. - #[error("upstream returned HTTP {status}: {body}")] - UpstreamHttp { - /// Upstream HTTP status code. - status: u16, - /// Raw upstream error body. - body: String, - }, - - /// A streamed response failed mid-flight (read error or malformed frame). - #[error("stream error: {0}")] - Stream(String), -} - /// Detects a context-overflow body using a provider-supplied structured check /// and a substring phrase list. /// diff --git a/crates/libsy/README.md b/crates/libsy/README.md index 19ec68e8a..f426adb89 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -85,7 +85,7 @@ struct MyClient { /* http client, base url, key */ } #[async_trait::async_trait] impl RoutedLlmClient for MyClient { async fn call(&self, ctx: Context, request: Request, decision: Arc) - -> Result> { + -> Result { let model = decision.selected_model(); // the routed target — map it to a provider id // request.llm_request.model is the agent's original name (not a call target) // ... POST to your endpoint, read the completion ... @@ -261,10 +261,13 @@ All libsy-owned APIs return [`switchyard_libsy::Result`], whose error is algorithm task failures, incomplete runs, and client-call failures without inspecting error strings. -`RoutedLlmClient` is owned by `switchyard-protocol` and still returns a boxed error. -When [`Algorithm::run`] serves a target, libsy preserves that error as the source of -`LibsyError::ClientCall`. Custom algorithms, classifiers, and processors that need to -surface their own error types can preserve them with +`RoutedLlmClient` is owned by `switchyard-protocol` and returns +`LlmClientError`. Callers can distinguish invalid requests, configuration +failures, request/response translation failures, request encoding failures, transport +failures, timeouts, context-window overflows, upstream HTTP responses, invalid responses, +and uncategorized client-specific failures. When [`Algorithm::run`] serves a target, +libsy preserves that error as the source of `LibsyError::ClientCall`. +Custom algorithms, classifiers, and processors that need to surface their own errors can use `LibsyError::external("operation", error)`. ## Observability diff --git a/crates/libsy/examples/ensemble.rs b/crates/libsy/examples/ensemble.rs index 6a0a68bf1..5705865fd 100644 --- a/crates/libsy/examples/ensemble.rs +++ b/crates/libsy/examples/ensemble.rs @@ -491,11 +491,11 @@ mod tests { _ctx: Context, request: Request, decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { let name = decision.selected_model().to_string(); self.calls .lock() - .map_err(|_| "lock poisoned")? + .map_err(|_| switchyard_protocol::LlmClientError::Other("lock poisoned".into()))? .push(name.clone()); let completion = if name == self.judge_model { judge_pick(&prompt_text(&request.llm_request), &self.prefer) @@ -741,9 +741,10 @@ mod tests { _ctx: Context, _request: Request, _decision: Arc, - ) -> std::result::Result> - { - Err("upstream down".into()) + ) -> std::result::Result { + Err(switchyard_protocol::LlmClientError::Other( + "upstream down".into(), + )) } } let client = Arc::new(FailingClient) as Arc; @@ -809,8 +810,7 @@ mod tests { _ctx: Context, request: Request, decision: Arc, - ) -> std::result::Result> - { + ) -> std::result::Result { let name = decision.selected_model().to_string(); let completion = if name == self.judge_model { // Judge runs after the barrier releases; it must not wait. diff --git a/crates/libsy/examples/research_agent.rs b/crates/libsy/examples/research_agent.rs index 40a6a87f9..7cf53ea9c 100644 --- a/crates/libsy/examples/research_agent.rs +++ b/crates/libsy/examples/research_agent.rs @@ -35,7 +35,7 @@ impl RoutedLlmClient for StubClient { _ctx: Context, _request: Request, decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { // The model to call is the routed decision's selection, not the inbound name. let model = decision.selected_model().to_string(); println!(" -> model call: {model}"); diff --git a/crates/libsy/src/algorithms/fall_through.rs b/crates/libsy/src/algorithms/fall_through.rs index b163e2141..780d23117 100644 --- a/crates/libsy/src/algorithms/fall_through.rs +++ b/crates/libsy/src/algorithms/fall_through.rs @@ -174,7 +174,7 @@ mod tests { _ctx: Context, _request: Request, decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { Ok(Response { llm_response: LlmResponse::Agg(text_response( None, diff --git a/crates/libsy/src/algorithms/llm_class.rs b/crates/libsy/src/algorithms/llm_class.rs index fccb9d601..6f668311b 100644 --- a/crates/libsy/src/algorithms/llm_class.rs +++ b/crates/libsy/src/algorithms/llm_class.rs @@ -238,14 +238,17 @@ mod tests { _ctx: Context, request: Request, decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { let name = decision.selected_model().to_string(); let completion = if name == self.classifier_model { self.score.clone() } else { format!("answer from {name}") }; - self.seen.lock().map_err(|_| "lock poisoned")?.push(request); + self.seen + .lock() + .map_err(|_| switchyard_protocol::LlmClientError::Other("lock poisoned".into()))? + .push(request); Ok(Response { llm_response: LlmResponse::Agg(text_response(None, completion)), metadata: None, diff --git a/crates/libsy/src/algorithms/rand.rs b/crates/libsy/src/algorithms/rand.rs index 2b2566578..8399b5b86 100644 --- a/crates/libsy/src/algorithms/rand.rs +++ b/crates/libsy/src/algorithms/rand.rs @@ -108,7 +108,7 @@ mod tests { _ctx: Context, _request: Request, decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { Ok(Response { llm_response: LlmResponse::Agg(text_response(None, decision.selected_model())), metadata: None, diff --git a/crates/libsy/src/algorithms/subagent_override.rs b/crates/libsy/src/algorithms/subagent_override.rs index 96336baf0..b64c6b868 100644 --- a/crates/libsy/src/algorithms/subagent_override.rs +++ b/crates/libsy/src/algorithms/subagent_override.rs @@ -110,7 +110,7 @@ mod tests { _ctx: Context, _request: Request, decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { Ok(Response { llm_response: LlmResponse::Agg(text_response(None, decision.selected_model())), metadata: None, diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 11af2748c..fdfb153cc 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -16,7 +16,7 @@ use tracing::Instrument; /// [`LlmResponseChunk`] is one streaming event; [`LlmResponse`] is the streamed response /// (a live [`LlmResponseStream`] or the terminal aggregate). pub use switchyard_protocol::{ - AggLlmResponse, Context, Decision, LlmRequest, LlmResponse, LlmResponseChunk, + AggLlmResponse, Context, Decision, LlmClientError, LlmRequest, LlmResponse, LlmResponseChunk, LlmResponseStream, Metadata, Request, Response, RoutedLlmClient, Signals, Usage, }; @@ -501,7 +501,7 @@ mod tests { _ctx: Context, _request: Request, decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { // Echo back the model the algorithm routed to (the decision's selection). Ok(Response { llm_response: LlmResponse::Agg(text_response( @@ -610,7 +610,7 @@ mod tests { _ctx: Context, _request: Request, _decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { let stream = futures::stream::iter(self.chunks.clone().into_iter().map(Ok)).boxed(); Ok(Response { llm_response: LlmResponse::Stream(stream), @@ -836,8 +836,7 @@ mod tests { _ctx: Context, _request: Request, decision: Arc, - ) -> std::result::Result> - { + ) -> std::result::Result { self.barrier.wait().await; Ok(Response { llm_response: LlmResponse::Agg(text_response( @@ -1138,7 +1137,7 @@ mod tests { _ctx: Context, _request: Request, decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { self.started.notify_one(); match self.delay { Some(delay) => tokio::time::sleep(delay).await, @@ -1167,7 +1166,7 @@ mod tests { _ctx: Context, _request: Request, decision: Arc, - ) -> std::result::Result> { + ) -> std::result::Result { self.gate.notified().await; Ok(Response { llm_response: LlmResponse::Agg(text_response( @@ -1293,8 +1292,7 @@ mod tests { _ctx: Context, _request: Request, _decision: Arc, - ) -> std::result::Result> - { + ) -> std::result::Result { if self.started.fetch_add(1, Ordering::SeqCst) + 1 == self.n { self.all_started.notify_one(); } diff --git a/crates/libsy/src/error.rs b/crates/libsy/src/error.rs index 6282319b8..86b8f8877 100644 --- a/crates/libsy/src/error.rs +++ b/crates/libsy/src/error.rs @@ -5,6 +5,7 @@ use std::{error::Error as StdError, time::Duration}; +use switchyard_protocol::LlmClientError; use thiserror::Error; /// Result type returned by libsy APIs. @@ -59,9 +60,9 @@ pub enum LibsyError { ClientCall { /// Target whose client failed. target: String, - /// Error supplied by the protocol-owned client trait. + /// Typed error supplied by the protocol-owned client trait. #[source] - source: Box, + source: LlmClientError, }, /// A user extension or other foreign operation failed. @@ -77,7 +78,7 @@ pub enum LibsyError { impl LibsyError { /// Wrap an error returned by the protocol-owned client trait. - pub fn client_call(target: impl Into, source: Box) -> Self { + pub fn client_call(target: impl Into, source: LlmClientError) -> Self { Self::ClientCall { target: target.into(), source, @@ -148,8 +149,10 @@ mod tests { #[test] fn client_call_preserves_target_and_source() { - let error = - LibsyError::client_call("strong", Box::new(std::io::Error::other("upstream down"))); + let error = LibsyError::client_call( + "strong", + LlmClientError::Other(Box::new(std::io::Error::other("upstream down"))), + ); match &error { LibsyError::ClientCall { target, source } => { assert_eq!(target, "strong"); diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs index 15f30da6e..956ca7150 100644 --- a/crates/libsy/tests/observability.rs +++ b/crates/libsy/tests/observability.rs @@ -11,7 +11,6 @@ //! latest (max) matching data point. use std::collections::BTreeMap; -use std::error::Error; use std::fmt; use std::sync::{Arc, Mutex, MutexGuard, OnceLock}; @@ -32,8 +31,6 @@ use switchyard_libsy::{ }; use switchyard_protocol::text_request; -type BoxErr = Box; - #[derive(Debug, thiserror::Error)] #[error("{0}")] struct TestError(&'static str); @@ -281,7 +278,7 @@ impl RoutedLlmClient for UsageClient { _ctx: Context, _request: Request, decision: Arc, - ) -> Result { + ) -> Result { Ok(Response { llm_response: LlmResponse::Agg(AggLlmResponse { model: Some(decision.selected_model().to_string()), diff --git a/crates/protocol/Cargo.toml b/crates/protocol/Cargo.toml index 9e4b891eb..26ab9aade 100644 --- a/crates/protocol/Cargo.toml +++ b/crates/protocol/Cargo.toml @@ -21,3 +21,4 @@ async-trait = "0.1" futures = "0.3" serde = { version = "1", features = ["derive"] } serde_json = "1" +thiserror = "2" diff --git a/crates/protocol/src/client.rs b/crates/protocol/src/client.rs index e66924067..47753c979 100644 --- a/crates/protocol/src/client.rs +++ b/crates/protocol/src/client.rs @@ -10,13 +10,91 @@ //! in libsy's orchestration crate — so a client crate that depends only on the //! protocol can serve routed calls without pulling in the orchestrator. -use std::error::Error; use std::sync::Arc; use async_trait::async_trait; +use thiserror::Error; use crate::{Context, Request, Response}; +/// A boxed client-specific error preserved as the source of a routed call failure. +pub type BoxError = Box; + +/// Failures a routed LLM client can surface to its caller. +/// +/// The variants classify failures that routing hosts commonly need to handle, +/// while boxed sources preserve implementation-specific detail. `Other` is the +/// escape hatch for failures that do not fit a shared category. +#[non_exhaustive] +#[derive(Debug, Error)] +pub enum LlmClientError { + /// The request cannot be served as supplied. + #[error("invalid request: {message}")] + InvalidRequest { message: String }, + + /// Decoding the inbound request failed in the translation engine. + #[error("request translation failed: {0}")] + RequestTranslation(String), + + /// Encoding the request for the upstream failed in the translation engine. + #[error("outbound request encoding failed: {0}")] + RequestEncoding(String), + + /// Decoding or encoding the response failed in the translation engine. + #[error("response translation failed: {0}")] + ResponseTranslation(String), + + /// The client is not configured to serve the selected target. + #[error("client configuration error: {message}")] + Configuration { message: String }, + + /// The upstream could not be reached or the request could not be sent. + #[error("upstream transport error: {source}")] + Transport { + /// Client-specific transport failure. + #[source] + source: BoxError, + }, + + /// The upstream request exceeded its timeout. + #[error("upstream request timed out: {source}")] + Timeout { + /// Client-specific timeout failure. + #[source] + source: BoxError, + }, + + /// The upstream rejected the request because it exceeds the model's context window. + #[error("context window exceeded for model {model}: {message}")] + ContextWindowExceeded { + /// Model whose context window was exceeded. + model: String, + /// Upstream error message. + message: String, + }, + + /// The upstream returned a non-success HTTP response. + #[error("upstream returned HTTP {status}: {body}")] + UpstreamHttp { + /// Upstream HTTP status code. + status: u16, + /// Raw upstream error body. + body: String, + }, + + /// The upstream returned a response the client could not decode. + #[error("invalid upstream response: {source}")] + InvalidResponse { + /// Client-specific decoding or validation failure. + #[source] + source: BoxError, + }, + + /// A client-specific failure. Prefer adding variants than using this. + #[error(transparent)] + Other(#[from] BoxError), +} + /// A decision/trace object produced by an algorithm. /// /// Carried as a trait object (not a generic parameter) so a stream consumer can @@ -49,5 +127,5 @@ pub trait RoutedLlmClient: Send + Sync { ctx: Context, request: Request, decision: Arc, - ) -> Result>; + ) -> Result; } diff --git a/crates/switchyard-py/src/libsy_bindings.rs b/crates/switchyard-py/src/libsy_bindings.rs index 8eedb7195..6a9918c2d 100644 --- a/crates/switchyard-py/src/libsy_bindings.rs +++ b/crates/switchyard-py/src/libsy_bindings.rs @@ -3,7 +3,6 @@ //! Minimal Python API for running Rust-owned libsy algorithms. -use std::error::Error; use std::sync::Arc; use async_trait::async_trait; @@ -12,15 +11,13 @@ use pyo3::prelude::*; use serde_json::{json, Value}; use switchyard_libsy::algorithms::{Noop, Random, SubagentOverride}; use switchyard_libsy::{ - AggLlmResponse, Algorithm, Context, Decision, LlmResponse, LlmTarget, LlmTargetSet, Metadata, - Request, Response, RoutedLlmClient, + AggLlmResponse, Algorithm, Context, Decision, LlmClientError, LlmResponse, LlmTarget, + LlmTargetSet, Metadata, Request, Response, RoutedLlmClient, }; use crate::errors::py_libsy_error; use crate::py_serde::{from_python, to_python}; -type BoxError = Box; - /// Adapts a Python object with `async call(request)` to libsy. struct PythonLlmClient { inner: Py, @@ -33,18 +30,18 @@ impl RoutedLlmClient for PythonLlmClient { _ctx: Context, request: Request, _decision: Arc, - ) -> Result { + ) -> Result { let metadata = request.metadata; let future = Python::attach(|py| { let request = to_python(py, &request.llm_request)?; let awaitable = self.inner.bind(py).call_method1("call", (request,))?; pyo3_async_runtimes::tokio::into_future(awaitable) }) - .map_err(boxed_python_error)?; + .map_err(other_python_error)?; - let response = future.await.map_err(boxed_python_error)?; + let response = future.await.map_err(other_python_error)?; let aggregate = Python::attach(|py| from_python::(response.bind(py))) - .map_err(boxed_python_error)?; + .map_err(invalid_python_response)?; Ok(Response { llm_response: LlmResponse::Agg(aggregate), metadata, @@ -196,8 +193,14 @@ fn subagent_override_algorithm( )))) } -fn boxed_python_error(error: PyErr) -> BoxError { - std::io::Error::other(error.to_string()).into() +fn other_python_error(error: PyErr) -> LlmClientError { + LlmClientError::Other(Box::new(std::io::Error::other(error.to_string()))) +} + +fn invalid_python_response(error: PyErr) -> LlmClientError { + LlmClientError::InvalidResponse { + source: Box::new(std::io::Error::other(error.to_string())), + } } pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index a466fec72..e9935adc3 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -21,9 +21,8 @@ use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; use axum::{Json, Router}; use axum_server::tls_rustls::RustlsConfig; -use libsy::{Algorithm, Context, Decision, LibsyError, Metadata, Request}; +use libsy::{Algorithm, Context, Decision, LibsyError, LlmClientError, Metadata, Request}; use serde_json::{json, Value}; -use switchyard_llm_client::LlmClientError; use tokio::net::{TcpListener, TcpSocket}; use switchyard_translation::{decode_request, WireFormat}; @@ -368,28 +367,20 @@ fn algorithm_error(error: LibsyError) -> Response { let LibsyError::ClientCall { source, .. } = &error else { return server_error(error.to_string()); }; - let Some(error) = source.downcast_ref::() else { - return server_error(error.to_string()); - }; - match error { - LlmClientError::UnknownModel(model) => error_response( - StatusCode::BAD_GATEWAY, - format!("No upstream backend configured for model {model}"), - "upstream_error", - "upstream_model_not_found", - ), - LlmClientError::UnknownModelFormat { model, format } => error_response( - StatusCode::BAD_GATEWAY, - format!("No upstream backend configured for model {model} and format {format}"), - "upstream_error", - "upstream_format_not_found", - ), - LlmClientError::MissingModel => error_response( + match source { + LlmClientError::InvalidRequest { message } + | LlmClientError::RequestTranslation(message) => error_response( StatusCode::BAD_REQUEST, - error.to_string(), + message, "invalid_request_error", "invalid_request_error", ), + LlmClientError::Configuration { message } => error_response( + StatusCode::BAD_GATEWAY, + message, + "upstream_error", + "upstream_configuration_error", + ), LlmClientError::ContextWindowExceeded { message, .. } => error_response( StatusCode::BAD_REQUEST, message, @@ -402,14 +393,28 @@ fn algorithm_error(error: LibsyError) -> Response { "upstream_error", "upstream_error", ), - LlmClientError::Translation(message) - | LlmClientError::Transport(message) - | LlmClientError::Stream(message) => error_response( + LlmClientError::Transport { source } | LlmClientError::InvalidResponse { source } => { + error_response( + StatusCode::BAD_GATEWAY, + source.to_string(), + "upstream_error", + "upstream_error", + ) + } + LlmClientError::ResponseTranslation(message) => error_response( StatusCode::BAD_GATEWAY, message, "upstream_error", "upstream_error", ), + LlmClientError::Timeout { source } => error_response( + StatusCode::GATEWAY_TIMEOUT, + source.to_string(), + "upstream_error", + "upstream_timeout", + ), + LlmClientError::RequestEncoding(message) => server_error(message), + _ => server_error(error.to_string()), } }