From 410250803f68e0344b485cffb5bafd18524871b1 Mon Sep 17 00:00:00 2001 From: Graham King Date: Fri, 14 Aug 2026 15:53:37 -0400 Subject: [PATCH 1/3] feat(libsy-llm-client): Move retry logic from libsy to libsy-llm-client The HTTP work belong in `libsy-llm-client`. The PR moves LLM calling retry code from `crates/libsy/src/algorithms/fall_through.rs` to `crates/libsy-llm-client/src/run.rs`. `FallThrough` holds almost no state now. Previously retry was handled by the `FallThrough` algorithm wrapper. It would issue a CallModel, inspect the HTTP response, and then ask the algorithm for a different model on failure. Now that retry is outside libsy FallThrough returns to it's design goal of being a set of classifiers with "fall through" if an earlier one does not select a model. For the client to retry we need a list of ModelId to try. `Driver::call_model` now takes `Vec`, which are the models the algorithm wants us to try in order, the selected one first. Algorithms behave like this already, but the multiple-models is somewhat hidden by the interaction with FallThrough. This makes it explicit. Assisted-by: Codex:GPT 5.6 Sol medium Plan and final review: Claude:Opus 5 high Signed-off-by: Graham King --- crates/libsy-llm-client/README.md | 15 +- crates/libsy-llm-client/src/observability.rs | 1 - crates/libsy-llm-client/src/run.rs | 408 ++++++++++++++++-- .../libsy-llm-client/tests/observability.rs | 2 +- crates/libsy/README.md | 7 +- crates/libsy/src/algorithms/fall_through.rs | 359 ++------------- crates/libsy/src/algorithms/llm_class.rs | 39 +- crates/libsy/src/algorithms/passthrough.rs | 4 +- crates/libsy/src/algorithms/stage.rs | 8 +- crates/libsy/src/algorithms/util/affinity.rs | 13 - crates/libsy/src/algorithms/util/llm_judge.rs | 5 +- crates/libsy/src/core/algorithm.rs | 348 +++------------ crates/libsy/src/core/classifier.rs | 5 - crates/libsy/src/core/testing.rs | 17 +- crates/libsy/src/error.rs | 4 - crates/switchyard-py/src/libsy_bindings.rs | 24 +- crates/switchyard-server/tests/server.rs | 4 +- docs/operations/context_window.md | 21 +- .../stage_router_routing.md | 4 +- switchyard_rust/libsy.py | 3 + tests/test_libsy_minimal_bindings.py | 26 +- 21 files changed, 583 insertions(+), 734 deletions(-) diff --git a/crates/libsy-llm-client/README.md b/crates/libsy-llm-client/README.md index 207ea787b..d86dc7708 100644 --- a/crates/libsy-llm-client/README.md +++ b/crates/libsy-llm-client/README.md @@ -144,9 +144,9 @@ async fn stream( ### Routing an algorithm [`run`] takes a libsy algorithm and a [`ClientRouter`], and returns the final response plus -the trace of decisions the algorithm published. The router resolves each offloaded call to -the client for the target the algorithm selected; `ClientRouter::single` is the -single-provider case: +the trace of decisions the algorithm published. Each offloaded `CallModel` carries an ordered +`models` list. The router resolves and tries those candidates in order; `ClientRouter::single` +is the single-provider case: ```rust use std::sync::Arc; @@ -249,9 +249,12 @@ fn build_multi_format_client( transport failures are retried; streaming body failures are not replayed after the response has been returned. -Retries replay the same upstream request. A transport failure can therefore -duplicate a request that the provider processed but did not finish returning, -and the retry budget plus capped `Retry-After` delays determines total latency. +Retries replay the same upstream request to the same model. Each candidate's +`max_retries` budget is exhausted before candidate fallback advances to the next +model. The worst case is `candidates × (max_retries + 1)` upstream requests, and +total latency includes every candidate's capped `Retry-After` backoff. A transport +failure can duplicate a request that the provider processed but did not finish +returning. ## Errors diff --git a/crates/libsy-llm-client/src/observability.rs b/crates/libsy-llm-client/src/observability.rs index e20d1a885..c02475b4b 100644 --- a/crates/libsy-llm-client/src/observability.rs +++ b/crates/libsy-llm-client/src/observability.rs @@ -177,7 +177,6 @@ fn client_call_error_type(error: &LibsyError) -> Cow<'static, str> { LibsyError::AlgorithmError { .. } => Cow::Borrowed("algorithm_error"), LibsyError::Driver(_) => Cow::Borrowed("driver_error"), LibsyError::MissingFinalResponse => Cow::Borrowed("missing_final_response"), - LibsyError::AllTargetsExcluded => Cow::Borrowed("context_window_exceeded"), LibsyError::External { .. } => Cow::Borrowed("_OTHER"), } } diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 1bcf66244..7737d5bbc 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -8,8 +8,10 @@ //! consumer — it drives the stream with [`switchyard_libsy::drive`], hands each call to a //! [`RoutedLlmClient`], and returns the final response with the trace of decisions. //! -//! libsy owns the stream mechanics; what this module adds is the client call itself and the -//! `libsy.client_call` span around it. +//! libsy owns the stream mechanics; what this module adds is ordered candidate fallback and the +//! `libsy.client_call` span around each candidate. Each candidate exhausts its backend retry +//! budget before fallback advances, so the worst case is `candidates × (max_retries + 1)` +//! upstream attempts plus every candidate's backoff. use std::collections::HashMap; use std::sync::Arc; @@ -17,7 +19,9 @@ use std::time::{Duration, Instant}; use parking_lot::Mutex; use switchyard_libsy::{Algorithm, CallModel, LibsyError, Result, drive}; -use switchyard_protocol::{Decision, LlmClientError, ModelId, Request, Response, RoutedLlmClient}; +use switchyard_protocol::{ + Decision, LlmClientError, ModelId, Request, Response, RoutedLlmClient, RoutingFallbackReason, +}; use crate::observation::{LlmCallObservation, RunObservation, RunObserver}; use crate::{metrics, observability}; @@ -98,10 +102,60 @@ impl RoutedCallWindows { } } -/// Serve one offloaded call. A failed *model* call is forwarded to the algorithm via -/// `respond`; this errors only when the promise itself could not be fulfilled. `serve` makes -/// the one provider call a routed request performs, so it gets its own `libsy.client_call` -/// span. +/// Serve one offloaded call and fulfill its promise. +/// +/// Errors only when the promise itself could not be fulfilled; a call that failed on every +/// candidate is forwarded to the algorithm as an `Err`. +async fn serve( + clients: ClientRouter, + call: CallModel, + observer: Option, + // Output parameter because `drive` takes a function that returns a plain `Result<()>`. + routed_calls: Arc>, +) -> Result<()> { + let result = call_first_available(&clients, &call, &observer, &routed_calls).await; + call.respond(result) +} + +/// Try candidates in order until one succeeds or a failure stops fallback. +async fn call_first_available( + clients: &ClientRouter, + call: &CallModel, + observer: &Option, + routed_calls: &Arc>, +) -> Result { + for (index, target) in call.models.iter().enumerate() { + let request = request_for(&call.request, target); + match call_one( + clients, + target, + request, + call, + observer, + routed_calls, + index, + call.models.len(), + ) + .await + { + Ok(response) => return Ok(response), + Err(error) if index + 1 == call.models.len() => return Err(error), + Err(error) => match fallback_reason(&error) { + Some(reason) => tracing::info!( + from = %target, + to = %call.models[index + 1], + reason = reason.as_str(), + "model call failed; trying next candidate" + ), + None => return Err(error), + }, + } + } + Err(LibsyError::NoTargets) +} + +/// Call one candidate model and record its observation and span. +#[allow(clippy::too_many_arguments)] #[tracing::instrument( target = "libsy", name = "libsy.client_call", @@ -109,12 +163,14 @@ impl RoutedCallWindows { fields( algorithm = call.algorithm, switchyard.algorithm = call.algorithm, - selected_model = call.selected_model_id(), + switchyard.candidate = index + 1, + switchyard.candidate_count = count, + selected_model = %model_id, otel.kind = "client", - otel.name = %format_args!("chat {}", call.selected_model_id()), + otel.name = %format_args!("chat {model_id}"), openinference.span.kind = "LLM", gen_ai.operation.name = "chat", - gen_ai.request.model = call.selected_model_id(), + gen_ai.request.model = %model_id, gen_ai.request.stream = tracing::field::Empty, gen_ai.request.temperature = tracing::field::Empty, gen_ai.request.top_p = tracing::field::Empty, @@ -138,15 +194,20 @@ impl RoutedCallWindows { error = tracing::field::Empty, ) )] -async fn serve( - clients: ClientRouter, - call: CallModel, - observer: Option, - // Output parameter because `drive` takes a function that returns a plain `Result<()>`. - routed_calls: Arc>, -) -> Result<()> { +async fn call_one( + clients: &ClientRouter, + model_id: &ModelId, + request: Request, + call: &CallModel, + observer: &Option, + routed_calls: &Arc>, + // index is for span log + index: usize, + // count is for span log + count: usize, +) -> Result { let span = tracing::Span::current(); - observability::record_gen_ai_request(&span, &call.request.llm_request); + observability::record_gen_ai_request(&span, &request.llm_request); if let Some(session_id) = call .request .metadata @@ -155,25 +216,23 @@ async fn serve( { span.record("gen_ai.conversation.id", session_id); } - let target = ModelId::from(call.selected_model_id()); - let request = call.request.clone(); - let is_answer_call = call.decision.is_answer_call(); + let is_answer_call = call.is_answer_call; // Resolved before the clock starts: picking the client is Switchyard's work, not // the provider's, so it belongs in the routing overhead. - let client = clients.route(&target); + let client = clients.route(model_id); let started = Instant::now(); let result = match client { Ok(client) => client.call(request).await, Err(error) => Err(error), } - .map_err(|source| LibsyError::client_call(target.clone(), source)); + .map_err(|source| LibsyError::client_call(model_id.clone(), source)); let ended = Instant::now(); let duration = ended - started; let result = observability::observe_client_call(result); if let Some(observer) = observer { observer(RunObservation::LlmCall(LlmCallObservation { - selected_model: target, + selected_model: model_id.clone(), is_answer_call, is_success: result.is_ok(), duration, @@ -188,7 +247,33 @@ async fn serve( routed_calls.lock().record(started, ended); } - call.respond(result) + result +} + +/// Whether a failed candidate is worth routing around. +fn fallback_reason(error: &LibsyError) -> Option { + let LibsyError::ClientCall { source, .. } = error else { + return None; + }; + match source { + LlmClientError::ContextWindowExceeded { .. } => Some(RoutingFallbackReason::ContextWindow), + LlmClientError::Transport { .. } | LlmClientError::Timeout { .. } => { + Some(RoutingFallbackReason::Unavailable) + } + LlmClientError::UpstreamHttp { status, .. } + if matches!(*status, 403 | 408 | 429) || (500..=599).contains(status) => + { + Some(RoutingFallbackReason::Unavailable) + } + _ => None, + } +} + +/// Clone a request and stamp the candidate model that should receive it. +fn request_for(request: &Request, target: &ModelId) -> Request { + let mut request = request.clone(); + request.llm_request.model = Some(target.to_string()); + request } /// Resolves a routed call's selected model to the client that serves it. @@ -256,3 +341,276 @@ impl FromIterator<(ModelId, Arc)> for ClientRouter { Self::new(iter.into_iter().collect()) } } + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use super::*; + use async_trait::async_trait; + use futures::StreamExt; + use switchyard_libsy::Driver; + use switchyard_protocol::{ + LlmResponse, LlmResponseChunk, LlmResponseStreamEvent, completion_text, text_request, + text_response, + }; + use wiremock::matchers::method; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use crate::{Backend, HttpBackendConfig, ModelConfig, TranslatingLlmClient}; + + struct CandidateAlgorithm { + models: Vec, + } + + #[async_trait] + impl Algorithm for CandidateAlgorithm { + fn name(&self) -> &str { + "candidate_test" + } + + async fn route(self: Arc, driver: Driver, request: Request) -> Result { + driver.call_model(request, self.models.clone(), true).await + } + } + + #[derive(Clone, Copy)] + enum FirstOutcome { + ContextWindow, + Unauthorized, + StreamSuccess, + MidStreamError, + } + + struct CandidateClient { + calls: Mutex>, + first: FirstOutcome, + } + + #[async_trait] + impl RoutedLlmClient for CandidateClient { + async fn call(&self, request: Request) -> std::result::Result { + let model = request.model_id().unwrap_or_default(); + self.calls.lock().push(model.clone()); + if model == "weak" { + return match self.first { + FirstOutcome::ContextWindow => Err(LlmClientError::ContextWindowExceeded { + model, + message: "too long".to_string(), + }), + FirstOutcome::Unauthorized => Err(LlmClientError::UpstreamHttp { + status: 401, + body: "unauthorized".to_string(), + }), + FirstOutcome::StreamSuccess => Ok(stream_response(vec![ + LlmResponseChunk::TextDelta { + index: 0, + text: "streamed".to_string(), + }, + LlmResponseChunk::MessageStop { + reason: Some("stop".to_string()), + }, + ])), + FirstOutcome::MidStreamError => Ok(stream_response(vec![ + LlmResponseChunk::TextDelta { + index: 0, + text: "partial".to_string(), + }, + LlmResponseChunk::StreamError { + message: "stream failed".to_string(), + }, + ])), + }; + } + Ok(Response { + llm_response: LlmResponse::Agg(text_response(Some(model.to_string()), model)), + metadata: None, + }) + } + } + + fn stream_response(chunks: Vec) -> Response { + Response { + llm_response: LlmResponse::Stream( + futures::stream::iter( + chunks + .into_iter() + .map(|chunk| Ok(LlmResponseStreamEvent::from(chunk))), + ) + .boxed(), + ), + metadata: None, + } + } + + fn request() -> Request { + Request { + llm_request: text_request(Some("auto".to_string()), "hello".to_string()), + raw_request: None, + metadata: None, + } + } + + async fn run_candidates( + first: FirstOutcome, + ) -> (Arc, Result<(Vec, Response)>) { + let client = Arc::new(CandidateClient { + calls: Mutex::new(Vec::new()), + first, + }); + let algorithm = Arc::new(CandidateAlgorithm { + models: vec!["weak".into(), "strong".into()], + }); + let result = run( + algorithm, + ClientRouter::single(client.clone()), + request(), + None, + ) + .await; + (client, result) + } + + #[test] + fn fallback_only_accepts_context_and_unavailable_failures() { + let error = |source| LibsyError::client_call("target", source); + assert_eq!( + fallback_reason(&error(LlmClientError::ContextWindowExceeded { + model: "target".into(), + message: "too long".to_string(), + })), + Some(RoutingFallbackReason::ContextWindow) + ); + for status in [403, 408, 429, 500, 599] { + assert_eq!( + fallback_reason(&error(LlmClientError::UpstreamHttp { + status, + body: "failed".to_string(), + })), + Some(RoutingFallbackReason::Unavailable) + ); + } + for status in [400, 401, 404, 409, 499, 600] { + assert_eq!( + fallback_reason(&error(LlmClientError::UpstreamHttp { + status, + body: "failed".to_string(), + })), + None + ); + } + } + + #[tokio::test] + async fn candidate_failures_follow_the_fallback_policy() -> Result<()> { + // Context overflow is retryable across candidates. + let (client, result) = run_candidates(FirstOutcome::ContextWindow).await; + let (_, response) = result?; + assert_eq!( + &*client.calls.lock(), + &[ModelId::from("weak"), "strong".into()] + ); + assert_eq!( + response + .llm_response + .as_agg() + .map(|response| response.model.as_deref()), + Some(Some("strong")) + ); + + // Authentication failure is not retryable, so the second candidate is untouched. + let (client, result) = run_candidates(FirstOutcome::Unauthorized).await; + assert!(matches!( + result, + Err(LibsyError::ClientCall { + source: LlmClientError::UpstreamHttp { status: 401, .. }, + .. + }) + )); + assert_eq!(&*client.calls.lock(), &[ModelId::from("weak")]); + Ok(()) + } + + #[tokio::test] + async fn retry_budget_is_exhausted_before_falling_through() -> Result<()> { + let server = MockServer::start().await; + let calls = Arc::new(Mutex::new(Vec::new())); + let observed_calls = Arc::clone(&calls); + Mock::given(method("POST")) + .respond_with(move |request: &wiremock::Request| { + let body: serde_json::Value = + serde_json::from_slice(&request.body).unwrap_or(serde_json::Value::Null); + let model = body["model"].as_str().unwrap_or_default().to_string(); + observed_calls.lock().push(model.clone()); + if model == "weak" { + ResponseTemplate::new(503) + .insert_header("retry-after", "0") + .set_body_string("unavailable") + } else { + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "id": "answer", + "model": "strong", + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "ok"}, + "finish_reason": "stop" + }], + "usage": {} + })) + } + }) + .mount(&server) + .await; + + let backend = || { + Backend::OpenAiChat(HttpBackendConfig { + base_url: format!("{}/v1", server.uri()), + api_key: None, + forward_auth: false, + extra_headers: BTreeMap::new(), + extra_body: BTreeMap::new(), + max_retries: 2, + }) + }; + let client = Arc::new( + TranslatingLlmClient::new(&[ + ModelConfig::new("weak", backend(), None), + ModelConfig::new("strong", backend(), None), + ]) + .map_err(|error| LibsyError::external("building test client", error))?, + ); + let algorithm = Arc::new(CandidateAlgorithm { + models: vec!["weak".into(), "strong".into()], + }); + run(algorithm, ClientRouter::single(client), request(), None).await?; + + assert_eq!(&*calls.lock(), &["weak", "weak", "weak", "strong"]); + Ok(()) + } + + #[tokio::test] + async fn streams_are_outside_the_candidate_fallback_boundary() -> Result<()> { + // Receiving a stream handle is a successful call and ends candidate selection. + let (client, result) = run_candidates(FirstOutcome::StreamSuccess).await; + let (_, response) = result?; + let aggregate = response + .llm_response + .into_agg() + .await + .map_err(|source| LibsyError::client_call("weak", source))?; + assert_eq!(&*client.calls.lock(), &[ModelId::from("weak")]); + assert_eq!(completion_text(&aggregate), "streamed"); + + // A later in-stream failure surfaces during aggregation without trying another model. + let (client, result) = run_candidates(FirstOutcome::MidStreamError).await; + let (_, response) = result?; + let error = response + .llm_response + .into_agg() + .await + .expect_err("the stream should fail during aggregation"); + assert!(error.to_string().contains("stream failed")); + assert_eq!(&*client.calls.lock(), &[ModelId::from("weak")]); + Ok(()) + } +} diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index aefae5bc2..dd5afa35e 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -444,7 +444,7 @@ impl Algorithm for SingleCallAlgo { tracing::info!("picked '{target}'"); let decision = Decision::new(target.clone(), true); driver.decide(decision.clone()).await?; - driver.call_model(request, decision).await + driver.call_model(request, vec![target.into()], true).await } } diff --git a/crates/libsy/README.md b/crates/libsy/README.md index 895c69f82..08072689c 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -33,9 +33,10 @@ tokio = { version = "1", features = ["macros", "rt"] } A target is a bare model id naming a routing destination. An [`Algorithm`] selects targets and records [`Decision`](switchyard_protocol::Decision)s, offloading every model call to its caller: [`Algorithm::run_stream`] yields a [`Step`] stream whose -[`Step::CallModel`] items the host serves over its own transport. libsy makes no -network calls itself — `switchyard-llm-client`'s `run` is a ready-made consumer -that drives the stream and performs the calls over HTTP. +[`Step::CallModel`] items the host serves over its own transport. Each call carries +an ordered, non-empty list of candidate models; the host tries them until one +answers. libsy makes no network calls itself — `switchyard-llm-client`'s `run` is +a ready-made consumer that drives the stream and performs the calls over HTTP. The provider-neutral [`Request`], [`Response`], [`Usage`], and [`LlmResponse`] contracts come from `switchyard-protocol`. diff --git a/crates/libsy/src/algorithms/fall_through.rs b/crates/libsy/src/algorithms/fall_through.rs index 314f3440f..be385a70e 100644 --- a/crates/libsy/src/algorithms/fall_through.rs +++ b/crates/libsy/src/algorithms/fall_through.rs @@ -13,11 +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. -//! An unavailable target is skipped only for the current request. +//! The selected target is offered first, followed by every other configured target. The consumer +//! may fall through that ordered candidate list when a model call fails. -use std::collections::HashSet; use std::{ collections::HashMap, sync::{Arc, Once, Weak}, @@ -28,11 +26,11 @@ use async_trait::async_trait; use parking_lot::Mutex; use tokio::sync::Mutex as AsyncMutex; -use crate::core::algorithm::{self, Algorithm, Driver, RoutingIdentity, SessionEvictions}; +use crate::core::algorithm::{self, Algorithm, Driver}; use crate::core::classifier::{Classification, Classifier, Score}; use crate::core::processor::{Event, Processor}; use crate::{LibsyError, Result}; -use switchyard_protocol::{Decision, ModelId, Request, Response, RoutingFallbackReason}; +use switchyard_protocol::{Decision, ModelId, Request, Response}; struct SessionState { state: Arc>, @@ -99,7 +97,6 @@ pub struct FallThrough { targets: Vec, session_states: Option>>, cleanup_started: Once, - session_evictions: SessionEvictions, } impl FallThrough<()> { @@ -113,7 +110,6 @@ impl FallThrough<()> { targets, session_states: None, cleanup_started: Once::new(), - session_evictions: SessionEvictions::default(), } } } @@ -132,7 +128,6 @@ where targets, session_states: Some(Arc::new(Mutex::new(HashMap::new()))), cleanup_started: Once::new(), - session_evictions: SessionEvictions::default(), } } @@ -190,56 +185,29 @@ where // The request is threaded mutably through the whole fold: any component may rewrite // it, later components see the rewrite, and the final value reaches the model. let mut request = request; - // Targets this turn must not route to. Scratch state for one run: seeded from the - // session's overflow history, then grown by each route-level failure below. - let mut excluded = HashSet::new(); - // Processors may rewrite the request; overflow history stays with its inbound identity. - let identity = RoutingIdentity::from_request(&request); - algorithm::exclude_evicted( - &mut excluded, - &self.targets, - &self.session_evictions, - identity.as_ref(), - ); let session_state = self.session_state(&request); - let (target, decision, served, deciding) = match session_state { + let (target, served) = match session_state { Some(state) => { let mut state = state.lock().await; - self.route(&mut state, &excluded, &driver, &mut request) - .await? + self.route(&mut state, &driver, &mut request).await? } None => { let mut state = S::default(); - self.route(&mut state, &excluded, &driver, &mut request) - .await? + self.route(&mut state, &driver, &mut request).await? } }; // A classifier that already called a model — because deciding required one, and that // call also answers the turn — hands its response back here, so the turn is not paid - // for twice. There is no outbound call left to overflow, so the fallback is skipped. + // for twice. // Nothing reads it on the way out: streamed or buffered, it reaches the caller // untouched. match served { Some(response) => Ok(response), None => { - algorithm::call_model_with_fallback( - &mut excluded, - &driver, - &self.targets, - target, - decision, - request, - identity.as_ref(), - &self.session_evictions, - |request, target| { - for classifier in &self.classifiers { - classifier.target_unavailable(request, target); - } - }, - |from, to, reason| self.fallback_decision(deciding.as_ref(), from, to, reason), - ) - .await + driver + .call_model(request, self.candidates(&target), true) + .await } } } @@ -249,30 +217,18 @@ where if let Some(states) = &self.session_states { states.lock().remove(session); } - self.session_evictions.remove_session(session); } - /// The decision published when a route-level failure selects a different target. - fn fallback_decision( - &self, - deciding: &dyn Classifier, - from: &ModelId, - to: &ModelId, - reason: RoutingFallbackReason, - ) -> Decision { - let failure = match reason { - RoutingFallbackReason::ContextWindow => "exceeded its context window", - RoutingFallbackReason::Unavailable => "was unavailable", - }; - let message = with_routing_tier( - format!( - "{from} {failure}; fell back to {to} (fallback reason: {})", - reason.as_str(), - ), - deciding.routing_tier(to), - ); - tracing::info!("{message}"); - Decision::new(to.clone(), true) + /// The selected target first, then every other configured target as a fallback candidate. + fn candidates(&self, target: &ModelId) -> Vec { + std::iter::once(target.clone()) + .chain( + self.targets + .iter() + .filter(|candidate| *candidate != target) + .cloned(), + ) + .collect() } /// Returns this request's retained state without holding the registry lock. @@ -292,10 +248,9 @@ where async fn route( &self, state: &mut S, - excluded: &HashSet, driver: &Driver, request: &mut Request, - ) -> Result<(ModelId, Decision, Option, Arc>)> { + ) -> Result<(ModelId, Option)> { // 1. Processor chain accumulates request-side facts into the composition's state. for processor in &self.processors { processor.process(state, Event::Request(request)).await?; @@ -319,17 +274,10 @@ where }); }; - // 3. Resolve the target, log the choice, and publish the decision. When an excluded - // target sends the request elsewhere, the log describes where it actually went. - let target = algorithm::select_eligible_model(&self.targets, &score.target, excluded)?; - let message = if target == score.target { - (self.decision_reason)(&self.name, &score) - } else { - format!( - "{} exceeded its context window; fell back to {}", - score.target, target - ) - }; + // 3. Resolve the target, log the choice, and publish the decision. + algorithm::ensure_model_is_target(&self.targets, &score.target)?; + let target = score.target.clone(); + let message = (self.decision_reason)(&self.name, &score); let message = with_routing_tier(message, deciding.routing_tier(&target)); tracing::info!("{message}"); let decision: Decision = Decision::new(target.clone(), true); @@ -345,7 +293,7 @@ where processor.process(state, event).await?; } - Ok((target, decision, served, deciding)) + Ok((target, served)) } } @@ -417,12 +365,10 @@ mod tests { use super::*; use crate::algorithms::util::prompts; use crate::core::classifier::Classification; - use crate::{AffinityRouter, SystemPromptProcessor, TargetPrompts}; + use crate::{SystemPromptProcessor, TargetPrompts}; use crate::core::testing::{Serve, echo, reply, test_drive}; - use switchyard_protocol::{ - LlmClientError, LlmRequest, Message, Metadata, Role, completion_text, text_request, - }; + use switchyard_protocol::{LlmRequest, Message, Metadata, Role, completion_text, text_request}; #[derive(Debug, thiserror::Error)] #[error("{0}")] @@ -437,11 +383,11 @@ mod tests { /// Echoes the routed model name, capturing the request it was handed so a test can /// assert on what actually reached the model. fn capturing(into: Arc>>) -> impl Serve { - move |decision: Decision, request: Request| { + move |target: ModelId, request: Request| { let into = Arc::clone(&into); async move { *into.lock() = Some(request); - Ok(reply(decision.selected_model_id())) + Ok(reply(target)) } } } @@ -465,11 +411,11 @@ mod tests { impl PromptRecorder { fn serve(self: &Arc) -> impl Serve { let recorder = Arc::clone(self); - move |decision: Decision, request: Request| { + move |target: ModelId, request: Request| { let recorder = Arc::clone(&recorder); async move { *recorder.0.lock() = Some(RecordedCall { - target: decision.selected_model_id().to_string(), + target: target.to_string(), messages: request .llm_request .messages @@ -483,7 +429,7 @@ mod tests { .filter_map(|block| block.content.iter().find_map(text_of)) .collect(), }); - Ok(reply(decision.selected_model_id())) + Ok(reply(target)) } } } @@ -622,55 +568,6 @@ mod tests { run_turn(&Arc::new(router), serve).await } - /// Discards the recorded calls; the overflow tests that don't assert on them. - fn overflows(targets: &'static [&'static str]) -> impl Serve { - overflowing(targets, Arc::new(Mutex::new(Vec::new()))) - } - - /// Rejects the named `overflowing` targets with a context-window error and echoes for - /// the rest, so the retry path can be driven. Every call is recorded in `calls`. - fn overflowing( - overflowing: &'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().clone(); - calls.lock().push(model.to_string()); - if overflowing.contains(&model.as_str()) { - return Err(LlmClientError::ContextWindowExceeded { - model, - message: "prompt is too long".to_string(), - }); - } - Ok(reply(model)) - } - } - } - - /// Rejects the named `unavailable` targets with a 503 and echoes for the rest. - /// Every call is recorded in `calls`. - fn unavailable( - unavailable: &'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 unavailable.contains(&model.as_str()) { - return Err(LlmClientError::UpstreamHttp { - status: 503, - body: "unavailable".to_string(), - }); - } - Ok(reply(model)) - } - } - } - // --- tests ------------------------------------------------------------------------- #[tokio::test] @@ -760,187 +657,23 @@ mod tests { } #[tokio::test] - async fn a_target_that_overflowed_is_skipped_for_the_rest_of_the_session() -> 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 = overflowing(&["weak"], calls.clone()); - assert_eq!(run_turn(&router, serve).await?.0, "strong"); - } - assert_eq!(calls.lock().iter().filter(|m| *m == "weak").count(), 1); - Ok(()) - } - - #[tokio::test] - async fn a_different_session_starts_with_an_empty_eviction_set() -> 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)])), - ); - run_turn(&router, overflowing(&["weak"], calls.clone())).await?; - let mut other = request(); - other.metadata = Some(Metadata { - session_id: Some("session-2".to_string()), - ..Metadata::default() - }); - run_request(&router, other, overflowing(&["weak"], calls.clone())).await?; - assert_eq!(calls.lock().iter().filter(|m| *m == "weak").count(), 2); - Ok(()) - } - - #[tokio::test] - async fn second_turn_after_full_exhaustion_still_reaches_upstream() -> 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)])), - ); - let first = run_turn(&router, overflowing(&["weak", "strong"], calls.clone())).await; - assert!(first.is_err()); - calls.lock().clear(); - match run_turn(&router, overflowing(&["weak", "strong"], calls.clone())).await { - Err(LibsyError::ClientCall { .. }) => {} - Err(other) => panic!("turn 2 gave {other:?}, calls={:?}", calls.lock()), - Ok(_) => panic!("expected an error"), - } - Ok(()) - } + async fn selected_target_leads_the_ordered_candidate_list() -> Result<()> { + use futures::StreamExt; - #[tokio::test] - async fn an_overflowing_target_is_retried_on_one_that_fits() -> Result<()> { - let router = FallThrough::<()>::new(target_set(&["weak", "strong"])) - .with_classifier(fixed(vec![score("weak", 0.9)])); - let (model, _) = run_with(router, overflows(&["weak"])).await?; - assert_eq!(model, "strong"); - Ok(()) - } - - #[tokio::test] - async fn unavailable_target_clears_matching_affinity_before_the_next_turn() -> Result<()> { - let calls = Arc::new(Mutex::new(Vec::new())); - let affinity = Arc::new(AffinityRouter::new()); let router = Arc::new( - FallThrough::<()>::new(target_set(&["weak", "strong"])) - .with_processor(affinity.clone()) - .with_classifier(affinity) - .with_classifier(Arc::new(DefaultTarget::new("weak"))), + FallThrough::<()>::new(target_set(&["weak", "mid", "strong"])) + .with_classifier(fixed(vec![score("mid", 0.9)])), ); - for _ in 0..2 { - let (model, trace) = run_turn(&router, unavailable(&["weak"], calls.clone())).await?; - assert_eq!(model, "strong"); - assert_eq!( - trace - .last() - .map(|decision| decision.selected_model_id().as_str()), - Some("strong") - ); - } - assert_eq!(&*calls.lock(), &["weak", "strong", "weak", "strong"]); - Ok(()) - } - - #[tokio::test] - async fn fallback_decision_preserves_answer_call_semantics() -> Result<()> { - struct TieredClassifier; - - #[async_trait] - impl Classifier for TieredClassifier { - async fn score( - &self, - _state: &mut (), - _request: &mut Request, - _driver: Option<&Driver>, - ) -> Result<(Classification, Option)> { - Ok((Classification::Scores(vec![score("weak", 1.0)]), None)) + let stream = router.run_stream(request()); + tokio::pin!(stream); + while let Some(step) = stream.next().await { + if let crate::Step::CallModel(call) = step? { + assert_eq!(call.models, target_set(&["mid", "weak", "strong"])); + assert_eq!(call.request.llm_request.model.as_deref(), Some("mid")); + return Ok(()); } } - - let router = FallThrough::<()>::new(target_set(&["weak", "strong"])) - .with_classifier(Arc::new(TieredClassifier)); - let (_, trace) = run_with(router, unavailable(&["weak"], Arc::default())).await?; - - assert_eq!(trace.len(), 2); - assert_eq!(trace[0].selected_model_id(), "weak"); - assert!(trace[0].is_answer_call()); - - let fallback = &trace[1]; - assert_eq!(fallback.selected_model_id(), "strong"); - assert!(fallback.is_answer_call()); - Ok(()) - } - - #[tokio::test] - async fn overflowing_targets_are_retried_until_one_fits() -> Result<()> { - let router = FallThrough::<()>::new(target_set(&["weak", "mid", "strong"])) - .with_classifier(fixed(vec![score("weak", 0.9)])); - let (model, _) = run_with(router, overflows(&["weak", "mid"])).await?; - assert_eq!(model, "strong"); - Ok(()) - } - - #[tokio::test] - async fn exhausting_every_target_surfaces_the_client_overflow() -> Result<()> { - // Only the client error maps to a 400 upstream, so it must survive exhaustion. - let router = FallThrough::<()>::new(target_set(&["weak", "strong"])) - .with_classifier(fixed(vec![score("weak", 0.9)])); - match run_with(router, overflows(&["weak", "strong"])).await { - Ok(_) => panic!("expected an overflow error, got a response"), - Err(LibsyError::ClientCall { - source: LlmClientError::ContextWindowExceeded { .. }, - .. - }) => Ok(()), - Err(other) => panic!("expected ContextWindowExceeded, got {other:?}"), - } - } - - #[tokio::test] - async fn a_retried_request_runs_the_processors_once() -> Result<()> { - // Routing runs before the call loop, so an overflow must not replay processors. - struct CountingProcessor(Arc>>); - - #[async_trait] - impl Processor for CountingProcessor { - async fn process(&self, _state: &mut (), event: Event<'_>) -> Result<()> { - let kind = match event { - Event::Request(_) => "request", - Event::Decision { .. } => "decision", - _ => "other", - }; - self.0.lock().push(kind); - Ok(()) - } - } - - let seen = Arc::new(Mutex::new(Vec::new())); - let router = FallThrough::<()>::new(target_set(&["weak", "strong"])) - .with_classifier(fixed(vec![score("weak", 0.9)])) - .with_processor(Arc::new(CountingProcessor(seen.clone()))); - let (model, _) = run_with(router, overflows(&["weak"])).await?; - assert_eq!(model, "strong"); - assert_eq!(seen.lock().iter().filter(|e| **e == "request").count(), 1); - assert_eq!(seen.lock().iter().filter(|e| **e == "decision").count(), 1); - Ok(()) - } - - #[tokio::test] - async fn a_target_barred_by_overflow_history_reports_the_fallback() -> Result<()> { - // Headers and usage metrics read the decision, so it must describe the real call. - let router = Arc::new( - FallThrough::<()>::new(target_set(&["weak", "strong"])) - .with_classifier(fixed(vec![score("weak", 0.9)])), - ); - // The first turn teaches the session that "weak" overflows; the second is barred - // from it before calling, so routing redirects and must say where it went. - run_turn(&router, overflows(&["weak"])).await?; - let (text, trace) = run_turn(&router, overflows(&["weak"])).await?; - - assert_eq!(text, "strong"); - assert_eq!(trace[0].selected_model_id(), "strong"); - Ok(()) + Err(test_error("expected a CallModel step")) } #[tokio::test] diff --git a/crates/libsy/src/algorithms/llm_class.rs b/crates/libsy/src/algorithms/llm_class.rs index 7fec82dc3..7d8b2fb66 100644 --- a/crates/libsy/src/algorithms/llm_class.rs +++ b/crates/libsy/src/algorithms/llm_class.rs @@ -9,7 +9,7 @@ use std::sync::Arc; use async_trait::async_trait; use serde::{Deserialize, Deserializer}; use serde_json::Value; -use switchyard_protocol::{ContentBlock, Decision, Message, ModelId, Role}; +use switchyard_protocol::{ContentBlock, Message, ModelId, Role}; use super::fall_through::{DefaultTarget, FallThrough}; use super::util::DEFAULT_JUDGE_MAX_OUTPUT_TOKENS; @@ -515,15 +515,14 @@ impl Classifier for EscalationClassifier { // Call efficient model and buffer the response so the judge can read it. // - // If the efficient model exceeds its context window, fall through to capable: returning - // `(decisive(capable), None)` tells FallThrough::execute to call - // call_model_with_fallback with the capable target instead of surfacing the error. + // If the efficient model exceeds its context window, fall through to capable. This call + // deliberately has one candidate so the classifier sees the efficient model's error. tracing::info!( target = %self.efficient, "escalation classifier selected efficient tier" ); let efficient_response = match driver - .call_model(request.clone(), Decision::new(self.efficient.clone(), true)) + .call_model(request.clone(), vec![self.efficient.clone()], true) .await { Ok(r) => r, @@ -1020,15 +1019,15 @@ mod tests { fn serve(self: &Arc) -> impl Serve { let recorder = Arc::clone(self); - move |decision: Decision, request: Request| { + move |model: ModelId, request: Request| { let recorder = Arc::clone(&recorder); async move { - let model = decision.selected_model_id().to_string(); + let model = model.to_string(); recorder.calls.lock().push(model.clone()); recorder .call_roles .lock() - .push((model.clone(), decision.is_answer_call())); + .push((model.clone(), model != "judge")); let completion = if model == "judge" { recorder .judge_max_output_tokens @@ -1064,8 +1063,8 @@ mod tests { /// The judge times out; every other target answers normally. fn unreachable_judge() -> impl Serve { - |decision: Decision, request: Request| async move { - let model = decision.selected_model_id().to_string(); + |model: ModelId, request: Request| async move { + let model = model.to_string(); if model == "judge" { return Err(LlmClientError::Timeout { source: Box::new(std::io::Error::other("judge unreachable")), @@ -1739,8 +1738,6 @@ mod tests { use std::collections::VecDeque; - use switchyard_protocol::Decision; - /// A queue of replies, drained in order. struct Queue(Mutex>); @@ -1762,8 +1759,8 @@ mod tests { /// Serves the judge target from `judge` and every other target from `model`, each with /// its next queued reply. fn queued(model: Arc, judge: Arc) -> impl Serve { - move |decision: Decision, request: Request| { - let queue = if decision.selected_model_id() == "judge" { + move |target: ModelId, request: Request| { + let queue = if target == "judge" { Arc::clone(&judge) } else { Arc::clone(&model) @@ -1908,10 +1905,10 @@ mod tests { let router = escalation_router()?; // Efficient overflows, capable answers, and the judge must never be called. - let serve = |decision: Decision, _request: Request| async move { - match decision.selected_model_id().as_str() { + let serve = |target: ModelId, _request: Request| async move { + match target.as_str() { "efficient" => Err(LlmClientError::ContextWindowExceeded { - model: decision.selected_model_id().clone(), + model: target, message: "prompt is too long".to_string(), }), "judge" => panic!("the judge must not be consulted when efficient overflows"), @@ -1940,10 +1937,10 @@ mod tests { let calls = Arc::new(Mutex::new(Vec::new())); let serve = { let calls = Arc::clone(&calls); - move |decision: Decision, _request: Request| { + move |model: ModelId, _request: Request| { let calls = Arc::clone(&calls); async move { - let model = decision.selected_model_id().to_string(); + let model = model.to_string(); calls.lock().push(model.clone()); match model.as_str() { "efficient" => Ok(streamed_then_error(LlmClientError::Transport { @@ -1975,8 +1972,8 @@ mod tests { #[tokio::test] async fn escalation_classifier_preserves_non_transport_stream_errors() -> Result<()> { let router = escalation_router()?; - let serve = |decision: Decision, _request: Request| async move { - match decision.selected_model_id().as_str() { + let serve = |target: ModelId, _request: Request| async move { + match target.as_str() { "efficient" => Ok(streamed_then_error(LlmClientError::InvalidResponse { source: Box::new(std::io::Error::other("invalid stream event")), })), diff --git a/crates/libsy/src/algorithms/passthrough.rs b/crates/libsy/src/algorithms/passthrough.rs index f123c1d4f..3de5ccc9a 100644 --- a/crates/libsy/src/algorithms/passthrough.rs +++ b/crates/libsy/src/algorithms/passthrough.rs @@ -35,7 +35,9 @@ impl Algorithm for Passthrough { tracing::info!(target = %self.target, "passthrough selected target"); let decision: Decision = Decision::new(self.target.clone(), true); driver.decide(decision.clone()).await?; - driver.call_model(request, decision).await + driver + .call_model(request, vec![self.target.clone()], true) + .await } } diff --git a/crates/libsy/src/algorithms/stage.rs b/crates/libsy/src/algorithms/stage.rs index 948793586..f751f7939 100644 --- a/crates/libsy/src/algorithms/stage.rs +++ b/crates/libsy/src/algorithms/stage.rs @@ -219,7 +219,7 @@ mod tests { use crate::core::classifier::Score; use crate::core::state::StateValue; use crate::core::testing::{Serve, reply, test_drive}; - use switchyard_protocol::{Decision, Metadata, Response}; + use switchyard_protocol::{Metadata, Response}; /// A classifier that always picks `target`, standing in for a cascade member. struct Fixed(&'static str); @@ -358,10 +358,10 @@ mod tests { /// back so the fallback classifier has an answer without a real model. fn serve(self: &Arc) -> impl Serve { let recorder = Arc::clone(self); - move |decision: Decision, request: Request| { + move |target: ModelId, request: Request| { let recorder = Arc::clone(&recorder); async move { - let target = decision.selected_model_id().to_string(); + let target = target.to_string(); recorder.calls.lock().push(Call { target: target.clone(), messages: request @@ -370,7 +370,7 @@ mod tests { .iter() .filter_map(|message| message.text_content("|")) .collect(), - is_answer_call: decision.is_answer_call(), + is_answer_call: target != JUDGE, }); let completion = if target == JUDGE { let p_solve = *recorder.judge_p_solve.lock(); diff --git a/crates/libsy/src/algorithms/util/affinity.rs b/crates/libsy/src/algorithms/util/affinity.rs index d372fdef0..b95f91f15 100644 --- a/crates/libsy/src/algorithms/util/affinity.rs +++ b/crates/libsy/src/algorithms/util/affinity.rs @@ -179,19 +179,6 @@ impl Classifier for AffinityRouter where S: Send + 'static, { - fn target_unavailable(&self, request: &Request, target: &ModelId) { - let Some(key) = self.affinity_key(request) else { - return; - }; - let mut assignments = self.assignments.lock(); - if assignments - .get(&key) - .is_some_and(|assigned| assigned == target) - { - assignments.remove(&key); - } - } - async fn score( &self, _state: &mut S, diff --git a/crates/libsy/src/algorithms/util/llm_judge.rs b/crates/libsy/src/algorithms/util/llm_judge.rs index 29d7a49c5..4d5cfb564 100644 --- a/crates/libsy/src/algorithms/util/llm_judge.rs +++ b/crates/libsy/src/algorithms/util/llm_judge.rs @@ -22,7 +22,7 @@ use crate::core::algorithm::Driver; use crate::core::classifier::{Classification, Classifier}; use crate::core::state::State; use crate::{LibsyError, Result}; -use switchyard_protocol::{Decision, LlmClientError, Request, Response}; +use switchyard_protocol::{LlmClientError, Request, Response}; /// Builds the classifier-specific message view presented to a structured judge. pub(crate) trait ClassifierInput: Send + Sync { @@ -240,7 +240,8 @@ where let response = driver .call_model( self.judge.build_request(state, request), - Decision::new(self.target.to_string(), false), + vec![self.target.clone()], + false, ) .await .inspect_err(|error| report_fail_open(judge_model, error, libsy_error_reason(error))) diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index d75163f8c..88ca1ae46 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -5,18 +5,10 @@ //! algorithm implements, and the offload channel it uses to make model calls and //! publish [`Decision`]s. -use std::{ - collections::{HashMap, HashSet}, - future::Future, - panic::AssertUnwindSafe, - pin::Pin, - sync::Arc, - time::Instant, -}; +use std::{future::Future, panic::AssertUnwindSafe, pin::Pin, sync::Arc, time::Instant}; use async_trait::async_trait; use futures::{FutureExt, Stream, StreamExt}; -use parking_lot::Mutex; use tokio::sync::{mpsc, oneshot}; use tokio_stream::wrappers::ReceiverStream; use tracing::Instrument; @@ -28,9 +20,7 @@ use tracing::Instrument; /// [`switchyard_protocol::LlmResponseStreamEvent`] is its host/algorithm envelope; and /// [`switchyard_protocol::LlmResponse`] carries either a live /// [`switchyard_protocol::LlmResponseStream`] or the terminal aggregate. -use switchyard_protocol::{ - Decision, LlmClientError, ModelId, Request, Response, RoutingFallbackReason, -}; +use switchyard_protocol::{Decision, ModelId, Request, Response}; use crate::{DriverError, LibsyError, Result, observability}; @@ -47,32 +37,23 @@ pub type StepStream = Pin> + Send>>; /// you. A host that only wants the routing outcome can take the contents with /// [`into_parts`](Self::into_parts) and never respond; dropping the stream ends the run. /// -/// The selected model is available both from -/// [`decision.selected_model_id()`](Decision::selected_model_id) and from -/// `request.llm_request.model`. [`Driver::call_model`] stamps the decision's model onto the -/// request before publishing the call, so every consumer receives a request ready for the -/// selected target. +/// [`Driver::call_model`] stamps the first candidate model onto the request before publishing +/// the call. A consumer that falls through to a later candidate must re-stamp it. pub struct CallModel { /// The name of the algorithm that produced this call, so a host instrumenting the /// calls it serves can attribute its own spans to the algorithm behind them. pub algorithm: String, - /// The request to serve; its `model` is the selected model identified by `decision`. + /// The request to serve; its `model` is stamped with `models[0]`. pub request: Request, - /// The routing decision behind this call; `selected_model_id()` identifies the model to use. - pub decision: Decision, + /// Candidate models, tried in order until one answers. Never empty. + pub models: Vec, + /// True for an answer-generating call, false for classifier and judge calls. + pub is_answer_call: bool, // How to send the response back to the algorithm reply: oneshot::Sender>, } impl CallModel { - /// The selected model stamped onto this call's request by [`Driver::call_model`]. - pub fn selected_model_id(&self) -> &str { - let model_id = self.request.llm_request.model.as_deref(); - // Driver stamps the model before constructing CallModel. - debug_assert!(model_id.is_some()); - model_id.unwrap_or_default() - } - /// Fulfill the promise with the caller's model-call result. Pass `Err(..)` to /// propagate a failed model call back to the algorithm. Consumes the promise: it /// can only be fulfilled once. @@ -85,17 +66,17 @@ impl CallModel { /// Take the call's contents without answering it, dropping the promise — the routing /// outcome plus the request as the algorithm would have sent it, after any rewriting. /// - /// Should only be called if `decision.is_answer_call` is true as that is the final call. + /// Should only be called if `is_answer_call` is true as that is the final call. /// The algorithm's [`Driver::call_model`] will fail with [`DriverError::Abandoned`] and /// the run ends there. Taking a call the algorithm does not depend on (a judge or /// classifier call) may instead let it fail open and complete with degraded routing. /// /// An abandoned run is not recorded as a failed one. Dropping a [`CallModel`] without /// calling this still yields [`DriverError::ResponseDropped`], which is. - pub fn into_parts(self) -> (Request, Decision) { + pub fn into_parts(self) -> (Request, Vec) { let Self { request, - decision, + models, reply, .. } = self; @@ -103,7 +84,7 @@ impl CallModel { // telemetry can tell an abandoned run from a failed one. The receiver is already // gone if the algorithm stopped waiting, which is fine. let _ = reply.send(Err(DriverError::Abandoned.into())); - (request, decision) + (request, models) } } @@ -134,8 +115,9 @@ impl Driver { ) } - /// Offload a model call: publish it as a [`Step::CallModel`] and await the consumer's - /// [`Response`]. Errors if the stream is closed or the call failed. + /// Publish a model call and await the consumer's response. + /// + /// Errors if the stream is closed or the call failed. /// The await is wrapped in a `libsy.llm_call` span measuring *fulfillment* as /// the algorithm observes it (host queueing/serving included; a streamed /// response resolves when its stream handle arrives); latency, outcome, and @@ -147,7 +129,7 @@ impl Driver { skip_all, fields( algorithm = self.algorithm, - selected_model = %decision.selected_model_id(), + selected_model = %models.first().unwrap_or(&ModelId::from("NoTargets")), openinference.span.kind = "CHAIN", outcome = tracing::field::Empty, error = tracing::field::Empty, @@ -157,16 +139,23 @@ impl Driver { reasoning_tokens = tracing::field::Empty, ) )] - pub async fn call_model(&self, mut request: Request, decision: Decision) -> Result { - let selected_model_id = decision.selected_model_id().to_string(); - request.llm_request.model = Some(selected_model_id.clone()); - let is_answer_call = decision.is_answer_call(); + pub async fn call_model( + &self, + mut request: Request, + models: Vec, + is_answer_call: bool, + ) -> Result { + let Some(selected_model_id) = models.first().cloned() else { + return Err(LibsyError::NoTargets); + }; + request.llm_request.model = Some(selected_model_id.to_string()); let started = Instant::now(); let (reply, response) = oneshot::channel::>(); let call = CallModel { algorithm: self.algorithm.clone(), request, - decision, + models, + is_answer_call, reply, }; let result = async { @@ -182,7 +171,7 @@ impl Driver { let elapsed = started.elapsed(); observability::record_llm_call( &self.algorithm, - &selected_model_id, + selected_model_id.as_str(), is_answer_call, elapsed, &result, @@ -313,27 +302,8 @@ pub(crate) fn ensure_model_is_target(targets: &[ModelId], name: &ModelId) -> Res }) } -/// `name` itself, or the first target not in `excluded` when `name` has been barred. -/// Errors if `name` is unknown, or if every target is excluded. -pub(crate) fn select_eligible_model( - targets: &[ModelId], - name: &ModelId, - excluded: &HashSet, -) -> Result { - ensure_model_is_target(targets, name)?; - if !excluded.contains(name) { - return Ok(name.clone()); - } - targets - .iter() - .find(|target| !excluded.contains(*target)) - .cloned() - .ok_or(LibsyError::AllTargetsExcluded) -} - -/// Key for overflow history: a root request by its session, a child request by its session -/// and agent. Keying a child finer than its session keeps one child's overflow from evicting -/// a target for the parent or a sibling sharing the session. +/// Key for routing affinity: a root request by its session, a child request by its session +/// and agent. #[derive(Clone, Hash, PartialEq, Eq)] pub(crate) enum RoutingIdentity { /// Root request, keyed by session ID. @@ -360,155 +330,6 @@ impl RoutingIdentity { Some(Self::Session(session.to_string())) } } - - /// The session this identity belongs to; shared by a session's root and its children. - fn session(&self) -> &str { - match self { - Self::Session(session) | Self::Subagent { session, .. } => session, - } - } -} - -/// Bounds process-local overflow history. Dropping a live entry costs one rediscovered -/// overflow, so the victim choice does not need to be exact. -const MAX_EVICTION_IDENTITIES: usize = 1_024; - -/// Per-identity record of the targets that overflowed their context window. -/// -/// A conversation only grows, so a target that could not fit one turn will not fit a -/// later one; remembering it lets the next turn skip a call certain to fail. Requests -/// without a routing identity are not tracked — there is nothing to remember them by. -#[derive(Default)] -pub(crate) struct SessionEvictions { - by_identity: Mutex>>, -} - -impl SessionEvictions { - /// Forgets overflow history for a completed session, including every child of it. - pub(crate) fn remove_session(&self, session: &str) { - self.by_identity - .lock() - .retain(|identity, _| identity.session() != session); - } - - /// The targets `identity` has already overflowed; empty for an untracked request. - fn evicted_for(&self, identity: Option<&RoutingIdentity>) -> Vec { - let Some(identity) = identity else { - return Vec::new(); - }; - self.by_identity - .lock() - .get(identity) - .map(|targets| targets.iter().cloned().collect()) - .unwrap_or_default() - } - - /// Remembers that `target` overflowed for `identity`, tracking at most - /// [`MAX_EVICTION_IDENTITIES`] identities. - fn record(&self, identity: Option<&RoutingIdentity>, target: &ModelId) { - let Some(identity) = identity else { return }; - let mut histories = self.by_identity.lock(); - if histories.len() >= MAX_EVICTION_IDENTITIES - && !histories.contains_key(identity) - && let Some(oldest) = histories.keys().next().cloned() - { - histories.remove(&oldest); - } - histories - .entry(identity.clone()) - .or_default() - .insert(target.clone()); - } -} - -/// How many of `targets` this request is still allowed to reach. -fn eligible_targets(targets: &[ModelId], excluded: &HashSet) -> usize { - targets - .iter() - .filter(|target| !excluded.contains(*target)) - .count() -} - -/// Bars the targets `identity` has already overflowed from this request, so routing does -/// not select one that is certain to fail again. -pub(crate) fn exclude_evicted( - excluded: &mut HashSet, - targets: &[ModelId], - evictions: &SessionEvictions, - identity: Option<&RoutingIdentity>, -) { - for target in evictions.evicted_for(identity) { - // Never seed the pool empty: a later turn may be small enough to serve, and the - // caller should get the upstream's answer rather than a routing error. - if eligible_targets(targets, excluded) <= 1 { - break; - } - excluded.insert(target); - } -} - -/// Returns the failed target and routing fallback policy for a terminal client error. -fn classify_fallback(error: &LibsyError) -> Option<(&ModelId, RoutingFallbackReason)> { - let LibsyError::ClientCall { target, source } = error else { - return None; - }; - let reason = match source { - LlmClientError::ContextWindowExceeded { .. } => RoutingFallbackReason::ContextWindow, - LlmClientError::Transport { .. } | LlmClientError::Timeout { .. } => { - RoutingFallbackReason::Unavailable - } - LlmClientError::UpstreamHttp { status, .. } - if matches!(*status, 403 | 408 | 429) || (500..=599).contains(status) => - { - RoutingFallbackReason::Unavailable - } - _ => return None, - }; - Some((target, reason)) -} - -/// Calls `target`, falling back to the next eligible target after a route-level failure, -/// until a call succeeds or every target has been tried. -/// -/// Routing is deliberately not re-run: the fallback replaces the target in place, so the -/// caller's request-side work and retained state still see exactly one turn. -/// `fallback_decision` builds the [`Decision`] published for a `from -> to` hop. Context -/// overflows are recorded for `identity`; unavailable targets remain request-local. -#[allow(clippy::too_many_arguments)] -pub(crate) async fn call_model_with_fallback( - excluded: &mut HashSet, - driver: &Driver, - targets: &[ModelId], - mut target: ModelId, - mut decision: Decision, - request: Request, - identity: Option<&RoutingIdentity>, - evictions: &SessionEvictions, - target_unavailable: impl Fn(&Request, &ModelId), - fallback_decision: impl Fn(&ModelId, &ModelId, RoutingFallbackReason) -> Decision, -) -> Result { - loop { - let result = driver.call_model(request.clone(), decision.clone()).await; - let Err(error) = result else { return result }; - let Some((failed, reason)) = classify_fallback(&error) else { - return Err(error); - }; - // A target already excluded means the pool is spent; surface the client error - // so the caller still sees the concrete upstream failure. - if !excluded.insert(failed.clone()) { - return Err(error); - } - match reason { - RoutingFallbackReason::ContextWindow => evictions.record(identity, failed), - RoutingFallbackReason::Unavailable => target_unavailable(&request, failed), - } - let Ok(next) = select_eligible_model(targets, &target, excluded) else { - return Err(error); - }; - decision = fallback_decision(&target, &next, reason); - target = next; - driver.decide(decision.clone()).await?; - } } /// An optimization strategy. Implement [`route`](Self::route); @@ -588,6 +409,8 @@ pub trait Algorithm: Send + Sync + 'static { #[cfg(test)] mod tests { + use std::collections::HashMap; + use super::*; use crate::core::testing::{Serve, ServeResult, echo, reply, test_drive}; use futures::StreamExt; @@ -603,61 +426,6 @@ mod tests { LibsyError::external("test", TestError(message)) } - fn classified_client_error(source: LlmClientError) -> Option { - classify_fallback(&LibsyError::client_call("target", source)).map(|(_, reason)| reason) - } - - #[test] - fn route_fallback_only_accepts_context_and_unavailable_failures() { - assert_eq!( - classified_client_error(LlmClientError::ContextWindowExceeded { - model: ModelId::from("target"), - message: "too long".to_string(), - }), - Some(RoutingFallbackReason::ContextWindow) - ); - for source in [ - LlmClientError::Transport { - source: Box::new(std::io::Error::other("connection failed")), - }, - LlmClientError::Timeout { - source: Box::new(std::io::Error::other("request timed out")), - }, - ] { - assert_eq!( - classified_client_error(source), - Some(RoutingFallbackReason::Unavailable) - ); - } - for (status, expected) in [ - (400, None), - (401, None), - (403, Some(RoutingFallbackReason::Unavailable)), - (404, None), - (408, Some(RoutingFallbackReason::Unavailable)), - (409, None), - (429, Some(RoutingFallbackReason::Unavailable)), - (499, None), - (500, Some(RoutingFallbackReason::Unavailable)), - (599, Some(RoutingFallbackReason::Unavailable)), - (600, None), - ] { - assert_eq!( - classified_client_error(LlmClientError::UpstreamHttp { - status, - body: "failed".to_string(), - }), - expected - ); - } - assert_eq!( - classified_client_error(LlmClientError::InvalidResponse { - source: Box::new(std::io::Error::other("invalid response")), - }), - None - ); - } - /// Build a routed decision for orchestration tests. fn test_decision(selected_model_id: ModelId) -> Decision { Decision::new(selected_model_id, true) @@ -683,7 +451,7 @@ mod tests { .clone(); let decision = test_decision(target.clone()); driver.decide(decision.clone()).await?; - driver.call_model(request, decision).await + driver.call_model(request, vec![target], true).await } } @@ -713,12 +481,12 @@ mod tests { let first_driver = driver.clone(); let mut first = tokio::spawn(async move { first_driver - .call_model(request(), test_decision(ModelId::from("first"))) + .call_model(request(), vec![ModelId::from("first")], true) .await }); let second = tokio::spawn(async move { driver - .call_model(request(), test_decision(ModelId::from("second"))) + .call_model(request(), vec![ModelId::from("second")], true) .await }); @@ -728,7 +496,7 @@ mod tests { let Step::CallModel(call) = step else { return Err(test_error("expected a CallModel step")); }; - let selected_model = call.selected_model_id().to_string(); + let selected_model = call.models[0].to_string(); calls.insert(selected_model, call); } assert!( @@ -765,7 +533,7 @@ mod tests { let (driver, mut step_rx) = Driver::new("test"); let producer = tokio::spawn(async move { driver - .call_model(request(), test_decision(ModelId::from("dropped"))) + .call_model(request(), vec![ModelId::from("dropped")], true) .await }); let step = step_rx.recv().await.ok_or(DriverError::StreamClosed)??; @@ -803,19 +571,18 @@ mod tests { #[tokio::test] async fn into_parts_yields_the_selected_model_without_answering_it() -> Result<()> { let (driver, mut step_rx) = Driver::new("test"); - let decision = test_decision(ModelId::from("answer/model")); - let producer = tokio::spawn({ - let decision = decision.clone(); - async move { driver.call_model(request(), decision).await } + let producer = tokio::spawn(async move { + driver + .call_model(request(), vec![ModelId::from("answer/model")], true) + .await }); let step = step_rx.recv().await.ok_or(DriverError::StreamClosed)??; let Step::CallModel(call) = step else { return Err(test_error("expected a CallModel step")); }; - let (taken_request, taken_decision) = call.into_parts(); - assert_eq!(taken_decision.selected_model_id(), "answer/model"); - assert!(taken_decision.is_answer_call()); + let (taken_request, taken_models) = call.into_parts(); + assert_eq!(taken_models, vec![ModelId::from("answer/model")]); assert_eq!( taken_request.llm_request.model.as_deref(), Some("answer/model") @@ -859,7 +626,7 @@ mod tests { /// replaying `chunks` in order (as `Ok` items). fn streaming_orch(chunks: Vec) -> (Arc, impl Serve) { let algo = orch(target_set(&["stream/model"])); - let serve = move |_decision: Decision, _request: Request| { + let serve = move |_target: ModelId, _request: Request| { let chunks = chunks.clone(); async move { let stream = @@ -943,7 +710,7 @@ mod tests { match step? { Step::CallModel(call) => { saw_call = true; - assert_eq!(call.selected_model_id(), "offload/model"); + assert_eq!(call.models, vec![ModelId::from("offload/model")]); // Fulfilling the promise is the "real" model call the caller makes. call.respond(Ok(Response { llm_response: LlmResponse::Agg(text_response( @@ -1012,11 +779,11 @@ mod tests { for _ in 0..N { let algo = algo.clone(); let barrier = barrier.clone(); - let serve = move |decision: Decision, _request: Request| { + let serve = move |target: ModelId, _request: Request| { let barrier = barrier.clone(); async move { barrier.wait().await; - Ok(reply(decision.selected_model_id())) + Ok(reply(target)) } }; handles.push(tokio::spawn(async move { @@ -1301,10 +1068,8 @@ mod tests { } async fn route(self: Arc, driver: Driver, request: Request) -> Result { - let dec_w = test_decision(self.winner.clone().into()); - let dec_l = test_decision(self.loser.clone().into()); - let win = driver.call_model(request.clone(), dec_w); - let lose = driver.call_model(request, dec_l); + let win = driver.call_model(request.clone(), vec![self.winner.clone().into()], true); + let lose = driver.call_model(request, vec![self.loser.clone().into()], true); // First to resolve wins; `select!` drops the losing future (and its promise). tokio::select! { res = win => res, @@ -1322,10 +1087,10 @@ mod tests { winner: "winner".to_string(), loser: "loser".to_string(), }); - let serve = move |decision: Decision, _request: Request| { + let serve = move |target: ModelId, _request: Request| { let started = started.clone(); async move { - if decision.selected_model_id() == "loser" { + if target == "loser" { started.notify_one(); match loser_delay { Some(delay) => tokio::time::sleep(delay).await, @@ -1334,7 +1099,7 @@ mod tests { } else { started.notified().await; } - Ok(reply(decision.selected_model_id())) + Ok(reply(target)) } }; (algo, serve) @@ -1400,8 +1165,7 @@ mod tests { async fn route(self: Arc, driver: Driver, request: Request) -> Result { let offloads = futures::future::join_all((0..self.n).map(|i| { - let decision = test_decision(format!("m{i}").into()); - driver.call_model(request.clone(), decision) + driver.call_model(request.clone(), vec![format!("m{i}").into()], true) })); tokio::select! { _ = offloads => Err(test_error("offloads unexpectedly completed")), @@ -1420,7 +1184,7 @@ mod tests { // Serving enters each call; once all N are in flight it signals, then pends forever. let started = Arc::new(AtomicUsize::new(0)); - let serve = move |_decision: Decision, _request: Request| { + let serve = move |_target: ModelId, _request: Request| { let started = started.clone(); let all_started = all_started.clone(); async move { diff --git a/crates/libsy/src/core/classifier.rs b/crates/libsy/src/core/classifier.rs index 77191ae0e..a515a7c64 100644 --- a/crates/libsy/src/core/classifier.rs +++ b/crates/libsy/src/core/classifier.rs @@ -76,11 +76,6 @@ pub trait Classifier: Send + Sync { None } - /// Drops retained routing state when `target` was unavailable for `request`. - /// - /// Stateless classifiers do not need to implement this hook. - fn target_unavailable(&self, _request: &Request, _target: &ModelId) {} - /// Score the classifier's targets given the current state and request. /// /// When present, `driver` lets a classifier offload model calls. It is `None` diff --git a/crates/libsy/src/core/testing.rs b/crates/libsy/src/core/testing.rs index 1f03b74cc..5db31e5eb 100644 --- a/crates/libsy/src/core/testing.rs +++ b/crates/libsy/src/core/testing.rs @@ -17,7 +17,7 @@ use std::sync::Arc; use futures::future::BoxFuture; use switchyard_protocol::{ - Decision, LlmClientError, LlmResponse, Request, Response, text_response, + Decision, LlmClientError, LlmResponse, ModelId, Request, Response, text_response, }; use crate::core::algorithm::{Algorithm, CallModel}; @@ -29,16 +29,16 @@ pub(crate) type ServeResult = std::result::Result; /// Answers offloaded model calls. Returning `Err` propagates a failed *model* call back into /// the algorithm, which may route around it. pub(crate) trait Serve: Send + Sync + 'static { - fn serve(&self, decision: Decision, request: Request) -> BoxFuture<'static, ServeResult>; + fn serve(&self, target: ModelId, request: Request) -> BoxFuture<'static, ServeResult>; } impl Serve for F where - F: Fn(Decision, Request) -> Fut + Send + Sync + 'static, + F: Fn(ModelId, Request) -> Fut + Send + Sync + 'static, Fut: Future + Send + 'static, { - fn serve(&self, decision: Decision, request: Request) -> BoxFuture<'static, ServeResult> { - Box::pin(self(decision, request)) + fn serve(&self, target: ModelId, request: Request) -> BoxFuture<'static, ServeResult> { + Box::pin(self(target, request)) } } @@ -59,10 +59,9 @@ pub(crate) async fn test_drive( /// error-shape assertions match production. async fn fulfill(serve: Arc, call: CallModel) -> Result<()> { let request = call.request.clone(); - let decision = call.decision.clone(); - let target = decision.selected_model_id().to_string(); + let target = call.models[0].clone(); let result = serve - .serve(decision, request) + .serve(target.clone(), request) .await .map_err(|source| LibsyError::client_call(target, source)); call.respond(result) @@ -71,7 +70,7 @@ async fn fulfill(serve: Arc, call: CallModel) -> Result<()> { /// Answers with the selected model name as the completion — what most routing tests need, /// since they assert on *which* target was called. pub(crate) fn echo() -> impl Serve { - |decision: Decision, _request: Request| async move { Ok(reply(decision.selected_model_id())) } + |target: ModelId, _request: Request| async move { Ok(reply(target)) } } /// A buffered response whose completion text is `completion`. diff --git a/crates/libsy/src/error.rs b/crates/libsy/src/error.rs index aa5c63b4a..7cb47b06d 100644 --- a/crates/libsy/src/error.rs +++ b/crates/libsy/src/error.rs @@ -50,10 +50,6 @@ pub enum LibsyError { source: LlmClientError, }, - /// Every target overflowed its context window. - #[error("every target exceeded its context window")] - AllTargetsExcluded, - /// A user extension or other foreign operation failed. #[error("{operation} failed: {source}")] External { diff --git a/crates/switchyard-py/src/libsy_bindings.rs b/crates/switchyard-py/src/libsy_bindings.rs index 78a095260..094c4c6af 100644 --- a/crates/switchyard-py/src/libsy_bindings.rs +++ b/crates/switchyard-py/src/libsy_bindings.rs @@ -184,15 +184,21 @@ struct PyModelCall { inner: Option, algorithm: String, request: Py, + models: Vec, decision: Py, } impl PyModelCall { fn new(py: Python<'_>, call: CallModel) -> PyResult { let request = to_python(py, &call.request.llm_request)?; - let decision = Py::new(py, PyDecision::from(call.decision.clone()))?; + let selected = call.models[0].clone(); + let decision = Py::new( + py, + PyDecision::from(Decision::new(selected, call.is_answer_call)), + )?; Ok(Self { algorithm: call.algorithm.clone(), + models: call.models.iter().map(ToString::to_string).collect(), inner: Some(call), request, decision, @@ -220,6 +226,12 @@ impl PyModelCall { self.request.clone_ref(py) } + /// Candidate models in the order the host should try them. + #[getter] + fn models(&self) -> Vec { + self.models.clone() + } + /// The routing decision behind this call. #[getter] fn decision(&self, py: Python<'_>) -> Py { @@ -229,11 +241,9 @@ impl PyModelCall { /// Consume the answer call without serving it and return its rewritten request and decision. #[pyo3(name = "into_parts")] fn take_parts(&mut self, py: Python<'_>) -> PyResult<(Py, Py)> { - let (request, decision) = self.take()?.into_parts(); - Ok(( - to_python(py, &request.llm_request)?, - Py::new(py, PyDecision::from(decision))?, - )) + let decision = self.decision.clone_ref(py); + let (request, _models) = self.take()?.into_parts(); + Ok((to_python(py, &request.llm_request)?, decision)) } /// Fulfill this call with an aggregate normalized response dictionary. @@ -254,7 +264,7 @@ impl PyModelCall { return Err(PyTypeError::new_err("error must derive from BaseException")); } let call = self.take()?; - let target = call.decision.selected_model_id().clone(); + let target = call.models[0].clone(); let source = if error.is_instance_of::() { LlmClientError::ContextWindowExceeded { model: target.clone(), diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index b3cfb3791..87ed8ab63 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -2126,7 +2126,7 @@ async fn unavailable_target_fails_over_across_endpoints_and_stops_when_exhausted .headers .get("x-model-router-selected-model") .and_then(|value| value.to_str().ok()), - Some("model/strong") + Some("model/weak") ); let calls = upstream.calls.lock().await; assert_eq!( @@ -2150,7 +2150,7 @@ async fn unavailable_target_fails_over_across_endpoints_and_stops_when_exhausted .collect::, _>>()?; assert_eq!(records.len(), 3); assert!(records.iter().all(|record| { - record["model"] == "model/strong" && record.get("fallback_reason").is_none() + record["model"] == "model/weak" && record.get("fallback_reason").is_none() })); let previous_call_count = upstream.calls.lock().await.len(); diff --git a/docs/operations/context_window.md b/docs/operations/context_window.md index 4ade38066..cd7a370f5 100644 --- a/docs/operations/context_window.md +++ b/docs/operations/context_window.md @@ -1,14 +1,11 @@ # Context-Window Handling When an upstream rejects a request because the prompt exceeds the model's -context window, Switchyard drops that target and calls another target on the -same route, repeating until a call succeeds or every target has been tried. If -the request carries `x-switchyard-session-id` or a recognized coding-agent -session header such as `x-claude-code-session-id`, the target remains excluded -for the rest of that session. Without a session header, the fallback still -applies to the current request, but the overflow is not remembered. -After truncating or resetting context, clients should use a new session ID; -reusing the old ID preserves its target exclusions. +context window, Switchyard calls the remaining targets on the same route in +configured order, stopping when one answers or every target has been tried. +Fallback applies only to the current request. An overflow is not remembered +across turns, so the route may select and try the same target again on the next +request. ## What counts as an overflow @@ -50,12 +47,8 @@ picker = "efficient_first" confidence_threshold = 0.5 ``` -Response headers report where the request actually landed: - -```text -x-model-router-selected-model: openai/gpt-4o-mini -x-model-router-rationale: openai/gpt-4o exceeded its context window; fell back to openai/gpt-4o-mini -``` +Routing response headers describe the algorithm's selected model. They do not +change when the client falls through to a later candidate. ## When every target overflows diff --git a/docs/routing_algorithms/stage_router_routing.md b/docs/routing_algorithms/stage_router_routing.md index 7bf0da7bb..f4e4c451e 100644 --- a/docs/routing_algorithms/stage_router_routing.md +++ b/docs/routing_algorithms/stage_router_routing.md @@ -9,8 +9,8 @@ mechanical work. Which tier a turn defaults to depends on the picker you choose off that default. You configure it with a single knob, `confidence_threshold`, plus an optional LLM classifier. -If the selected target exceeds its context window, the router tries the next -eligible target until one succeeds or all configured targets have been tried. See +If the selected target exceeds its context window, the client tries the route's +remaining targets in configured order for that request. See [Context-Window Handling](../operations/context_window.md). ## How it works diff --git a/switchyard_rust/libsy.py b/switchyard_rust/libsy.py index 498eeca05..67e4e1888 100644 --- a/switchyard_rust/libsy.py +++ b/switchyard_rust/libsy.py @@ -58,6 +58,9 @@ def algorithm(self) -> str: ... @property def request(self) -> dict[str, object]: ... + @property + def models(self) -> list[str]: ... + @property def decision(self) -> Decision: ... diff --git a/tests/test_libsy_minimal_bindings.py b/tests/test_libsy_minimal_bindings.py index 15b7b5843..8e93cd6bb 100644 --- a/tests/test_libsy_minimal_bindings.py +++ b/tests/test_libsy_minimal_bindings.py @@ -60,14 +60,20 @@ async def run_algorithm( async for step in algorithm.run_stream(request or request_body(), headers=headers): match step: case Step.CallModel(call): - target = call.decision.selected_model_id - client = (clients or {})[target] - try: - response = await client.call(call.request) - except Exception as error: - call.fail(error) - else: - call.respond(response) + for index, target in enumerate(call.models): + candidate_request = {**call.request, "model": target} + client = (clients or {})[target] + try: + response = await client.call(candidate_request) + except ContextWindowExceededError as error: + if index + 1 == len(call.models): + call.fail(error) + except Exception as error: + call.fail(error) + break + else: + call.respond(response) + break case Step.Decision(decision): decisions.append(decision) case Step.Done(response): @@ -86,6 +92,7 @@ async def test_random_streams_complex_steps_and_accepts_a_dictionary_response() match step: case Step.CallModel(call): variants.append("call_model") + assert call.models == ["fast"] client_response = await client.call(call.request) call.respond(client_response) with pytest.raises(LibsyError, match="already been completed"): @@ -118,6 +125,7 @@ async def test_into_parts_supports_decision_only_routing() -> None: case Step.CallModel(call) if call.decision.is_answer_call: request, decision = call.into_parts() assert call.algorithm == "random" + assert call.models == ["fast"] assert request["messages"] == request_body()["messages"] assert decision.selected_model_id == "fast" assert decision.is_answer_call is True @@ -351,5 +359,5 @@ async def call(self, request: dict[str, Any]) -> dict[str, Any]: {"fast": OverflowClient(), "strong": EchoClient("strong")}, ) - assert [decision.selected_model_id for decision in decisions] == ["fast", "strong"] + assert [decision.selected_model_id for decision in decisions] == ["fast"] assert response["model"] == "strong" From 4b6255acfa26bbaa9b85da61c3a192afab6ddcaa Mon Sep 17 00:00:00 2001 From: Graham King Date: Fri, 14 Aug 2026 16:30:24 -0400 Subject: [PATCH 2/3] Feedback Signed-off-by: Graham King --- crates/libsy-llm-client/src/run.rs | 6 ++++ crates/libsy/src/core/algorithm.rs | 10 ++++-- crates/libsy/src/core/testing.rs | 2 +- crates/protocol/src/envelope.rs | 36 +++++++++++++++++++++ crates/protocol/src/metadata.rs | 4 ++- crates/switchyard-py/src/libsy_bindings.rs | 14 ++++++-- crates/switchyard-server/src/lib.rs | 37 ++++++++++------------ crates/switchyard-server/tests/server.rs | 7 ++-- docs/operations/context_window.md | 5 +-- 9 files changed, 89 insertions(+), 32 deletions(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 7737d5bbc..9ef133550 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -229,6 +229,10 @@ async fn call_one( let ended = Instant::now(); let duration = ended - started; + let result = result.map(|mut response| { + response.set_served_model(model_id); + response + }); let result = observability::observe_client_call(result); if let Some(observer) = observer { observer(RunObservation::LlmCall(LlmCallObservation { @@ -517,6 +521,7 @@ mod tests { .map(|response| response.model.as_deref()), Some(Some("strong")) ); + assert_eq!(response.served_model().map(ModelId::as_str), Some("strong")); // Authentication failure is not retryable, so the second candidate is untouched. let (client, result) = run_candidates(FirstOutcome::Unauthorized).await; @@ -593,6 +598,7 @@ mod tests { // Receiving a stream handle is a successful call and ends candidate selection. let (client, result) = run_candidates(FirstOutcome::StreamSuccess).await; let (_, response) = result?; + assert_eq!(response.served_model().map(ModelId::as_str), Some("weak")); let aggregate = response .llm_response .into_agg() diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 88ca1ae46..f46b4815e 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -43,7 +43,7 @@ pub struct CallModel { /// The name of the algorithm that produced this call, so a host instrumenting the /// calls it serves can attribute its own spans to the algorithm behind them. pub algorithm: String, - /// The request to serve; its `model` is stamped with `models[0]`. + /// The request to serve; its `model` is stamped with the first candidate. pub request: Request, /// Candidate models, tried in order until one answers. Never empty. pub models: Vec, @@ -129,7 +129,7 @@ impl Driver { skip_all, fields( algorithm = self.algorithm, - selected_model = %models.first().unwrap_or(&ModelId::from("NoTargets")), + selected_model = %models.first().map(ModelId::as_str).unwrap_or("NoTargets"), openinference.span.kind = "CHAIN", outcome = tracing::field::Empty, error = tracing::field::Empty, @@ -496,7 +496,11 @@ mod tests { let Step::CallModel(call) = step else { return Err(test_error("expected a CallModel step")); }; - let selected_model = call.models[0].to_string(); + let selected_model = call + .models + .first() + .ok_or_else(|| test_error("model call has no candidates"))? + .to_string(); calls.insert(selected_model, call); } assert!( diff --git a/crates/libsy/src/core/testing.rs b/crates/libsy/src/core/testing.rs index 5db31e5eb..deee05650 100644 --- a/crates/libsy/src/core/testing.rs +++ b/crates/libsy/src/core/testing.rs @@ -59,7 +59,7 @@ pub(crate) async fn test_drive( /// error-shape assertions match production. async fn fulfill(serve: Arc, call: CallModel) -> Result<()> { let request = call.request.clone(); - let target = call.models[0].clone(); + let target = call.models.first().cloned().ok_or(LibsyError::NoTargets)?; let result = serve .serve(target.clone(), request) .await diff --git a/crates/protocol/src/envelope.rs b/crates/protocol/src/envelope.rs index f4c4b257e..5c50e3341 100644 --- a/crates/protocol/src/envelope.rs +++ b/crates/protocol/src/envelope.rs @@ -49,4 +49,40 @@ impl Response { pub fn selected_model(&self) -> Option<&str> { self.llm_response.selected_model() } + + /// Returns the Switchyard target that successfully served this response. + /// + /// Unlike [`Self::selected_model`], this is available for streamed responses because the + /// client records the target when it receives the stream handle. + pub fn served_model(&self) -> Option<&ModelId> { + self.metadata.as_ref()?.served_model.as_ref() + } + + /// Records the Switchyard target that successfully served this response. + pub fn set_served_model(&mut self, model: &ModelId) { + self.metadata.get_or_insert_default().served_model = Some(model.clone()); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::text_response; + + #[test] + fn served_model_round_trips_through_response_metadata() { + let mut response = Response { + llm_response: LlmResponse::Agg(text_response(None, "answer")), + metadata: None, + }; + + assert_eq!(response.served_model(), None); + response.set_served_model(&ModelId::from("first")); + assert_eq!(response.served_model().map(ModelId::as_str), Some("first")); + response.set_served_model(&ModelId::from("fallback")); + assert_eq!( + response.served_model().map(ModelId::as_str), + Some("fallback") + ); + } } diff --git a/crates/protocol/src/metadata.rs b/crates/protocol/src/metadata.rs index a9e11ed49..59f13122f 100644 --- a/crates/protocol/src/metadata.rs +++ b/crates/protocol/src/metadata.rs @@ -9,7 +9,7 @@ use std::{collections::BTreeMap, str::FromStr as _}; -use crate::WireFormat; +use crate::{ModelId, WireFormat}; // Dotted paths addressing fields inside Codex's turn-metadata header JSON value. const CODEX_SESSION_ID_PATH: &str = "x-codex-turn-metadata.session_id"; @@ -185,6 +185,8 @@ pub struct Metadata { pub session_final: Option, /// External trace/request id for joining with the host's telemetry. pub correlation_id: Option, + /// Switchyard target that successfully served a response. + pub served_model: Option, /// Arbitrary host-defined key/value metadata. pub extra_metadata: Option>, /// HTTP headers to attach when forwarding the request/response, if any. diff --git a/crates/switchyard-py/src/libsy_bindings.rs b/crates/switchyard-py/src/libsy_bindings.rs index 094c4c6af..f52e29e36 100644 --- a/crates/switchyard-py/src/libsy_bindings.rs +++ b/crates/switchyard-py/src/libsy_bindings.rs @@ -191,7 +191,12 @@ struct PyModelCall { impl PyModelCall { fn new(py: Python<'_>, call: CallModel) -> PyResult { let request = to_python(py, &call.request.llm_request)?; - let selected = call.models[0].clone(); + let selected = call + .models + .first() + .cloned() + .ok_or(RustLibsyError::NoTargets) + .map_err(py_libsy_error)?; let decision = Py::new( py, PyDecision::from(Decision::new(selected, call.is_answer_call)), @@ -264,7 +269,12 @@ impl PyModelCall { return Err(PyTypeError::new_err("error must derive from BaseException")); } let call = self.take()?; - let target = call.models[0].clone(); + let target = call + .models + .first() + .cloned() + .ok_or(RustLibsyError::NoTargets) + .map_err(py_libsy_error)?; let source = if error.is_instance_of::() { LlmClientError::ContextWindowExceeded { model: target.clone(), diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index b1eb5acfb..fd02cd01b 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -37,7 +37,7 @@ use parking_lot::Mutex; use serde::Deserialize; use serde_json::{Value, json}; use switchyard_llm_client::{ClientRouter, RunObservation, RunObserver, TranslatingLlmClient}; -use switchyard_protocol::{Decision, LlmClientError, Metadata, ModelId, Request, Usage}; +use switchyard_protocol::{LlmClientError, Metadata, ModelId, Request, Usage}; use tokio::net::{TcpListener, TcpSocket}; use tokio::task; use tracing::{Instrument, Level}; @@ -757,22 +757,21 @@ async fn handle_llm_request( Ok(result) => result, Err(error) => return algorithm_error(error), }; - // Metrics, response body, and routing header all read the same decision, so - // the model they name can never disagree. An empty trace leaves the body with - // the id the upstream reported. let decision = trace.last(); - let response = if let Some(decision) = decision { + // The response carries the candidate that actually served it. Fall back to the routing + // decision for algorithms that return a response without an offloaded model call. + let served_model = response + .served_model() + .cloned() + .or_else(|| decision.map(|decision| decision.selected_model_id().clone())); + let response = if let Some(served_model) = served_model.as_ref() { let cache_eligible = cache_probe .as_ref() - .map(|probe| { - state - .stats - .prefix_eligibility(decision.selected_model_id(), probe) - }) + .map(|probe| state.stats.prefix_eligibility(served_model, probe)) .unwrap_or(0.0); usage_metrics::observe( response, - decision.selected_model_id(), + served_model.as_str(), started.0, state.stats, cache_eligible, @@ -782,13 +781,13 @@ async fn handle_llm_request( response }; - let served_model = decision.map(|decision| decision.selected_model_id().to_string()); - let mut response = match into_http_response(response, wire_format, served_model) { + let response_model = served_model.as_ref().map(ToString::to_string); + let mut response = match into_http_response(response, wire_format, response_model) { Ok(response) => response, Err(error) => return server_error(error.to_string()), }; - if let Some(decision) = decision { - attach_routing_headers(&mut response, decision); + if let Some(served_model) = served_model.as_ref() { + attach_routing_headers(&mut response, served_model.as_str()); } response } @@ -864,12 +863,8 @@ fn metadata_from_headers(headers: HeaderMap) -> Metadata { metadata } -fn attach_routing_headers(response: &mut Response, decision: &Decision) { - insert_routing_header( - response, - HEADER_SELECTED_MODEL, - decision.selected_model_id(), - ); +fn attach_routing_headers(response: &mut Response, served_model: &str) { + insert_routing_header(response, HEADER_SELECTED_MODEL, served_model); } fn insert_routing_header(response: &mut Response, name: &'static str, value: &str) { diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index 87ed8ab63..7cee8c07c 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -2126,8 +2126,9 @@ async fn unavailable_target_fails_over_across_endpoints_and_stops_when_exhausted .headers .get("x-model-router-selected-model") .and_then(|value| value.to_str().ok()), - Some("model/weak") + Some("model/strong") ); + assert_eq!(response.json()?["model"], "model/strong"); let calls = upstream.calls.lock().await; assert_eq!( calls[previous_call_count..] @@ -2142,6 +2143,8 @@ async fn unavailable_target_fails_over_across_endpoints_and_stops_when_exhausted // Fallback causes are logged rather than accumulated in the legacy stats counters. assert_eq!(stats["routing_fallbacks"]["unavailable"], 0); assert_eq!(stats["routing_fallbacks"]["context_window"], 0); + assert_eq!(stats["models"]["model/strong"]["calls"], 3); + assert_eq!(stats["models"]["model/weak"]["errors"], 3); let records = std::fs::read_to_string(&log_path)?; let records = records @@ -2150,7 +2153,7 @@ async fn unavailable_target_fails_over_across_endpoints_and_stops_when_exhausted .collect::, _>>()?; assert_eq!(records.len(), 3); assert!(records.iter().all(|record| { - record["model"] == "model/weak" && record.get("fallback_reason").is_none() + record["model"] == "model/strong" && record.get("fallback_reason").is_none() })); let previous_call_count = upstream.calls.lock().await.len(); diff --git a/docs/operations/context_window.md b/docs/operations/context_window.md index cd7a370f5..bd0e6b0b8 100644 --- a/docs/operations/context_window.md +++ b/docs/operations/context_window.md @@ -47,8 +47,9 @@ picker = "efficient_first" confidence_threshold = 0.5 ``` -Routing response headers describe the algorithm's selected model. They do not -change when the client falls through to a later candidate. +The response body, `x-model-router-selected-model` header, usage metrics, and +routing log name the candidate that actually served the request, including when +the client falls through from the algorithm's first choice. ## When every target overflows From b0a6a28da9420db9c28f26914ae5ee37d1208f44 Mon Sep 17 00:00:00 2001 From: Graham King Date: Fri, 14 Aug 2026 17:40:40 -0400 Subject: [PATCH 3/3] Fix merge Signed-off-by: Graham King --- crates/switchyard-server/tests/server.rs | 90 ------------------------ 1 file changed, 90 deletions(-) diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index 7cee8c07c..1cd87dca7 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -1998,96 +1998,6 @@ async fn routing_log_keeps_the_canonical_session_id_until_a_stream_drains() -> T Ok(()) } -// Overflow history is isolated per child, cleared with the session, and not retained when a -// child lacks an agent ID. -#[tokio::test] -async fn overflow_history_is_scoped_to_agent_and_session_lifetime() -> TestResult { - let upstream = MockUpstream::start().await?; - let state = fallback_state(&upstream.base_url)?; - let app = build_switchyard_router(state); - let child_a = [ - ("x-switchyard-session-id", "shared-session"), - ("x-switchyard-agent-id", "child-a"), - ("x-switchyard-is-subagent", "true"), - ]; - let root = [ - ("x-switchyard-session-id", "shared-session"), - ("x-switchyard-agent-id", "root"), - ("x-switchyard-is-subagent", "false"), - ]; - let child_b = [ - ("x-switchyard-session-id", "shared-session"), - ("x-switchyard-agent-id", "child-b"), - ("x-switchyard-is-subagent", "true"), - ]; - let child_without_agent_id = [ - ("x-switchyard-session-id", "shared-session"), - ("x-switchyard-is-subagent", "true"), - ]; - let final_root = [ - ("x-switchyard-session-id", "shared-session"), - ("x-switchyard-agent-id", "root"), - ("x-switchyard-is-subagent", "false"), - ("x-switchyard-session-final", "true"), - ]; - type Case<'a> = (&'a str, &'a [(&'a str, &'a str)], &'a [&'a str]); - let cases: [Case<'_>; 8] = [ - ( - "overflow", - child_a.as_slice(), - &["model/weak", "model/strong"], - ), - ("fits", child_a.as_slice(), &["model/strong"]), - ("fits", root.as_slice(), &["model/weak"]), - ("fits", child_b.as_slice(), &["model/weak"]), - ("fits", final_root.as_slice(), &["model/weak"]), - ("fits", child_a.as_slice(), &["model/weak"]), - ( - "overflow", - child_without_agent_id.as_slice(), - &["model/weak", "model/strong"], - ), - ( - "overflow", - child_without_agent_id.as_slice(), - &["model/weak", "model/strong"], - ), - ]; - - for (content, headers, expected_calls) in cases { - let previous_call_count = upstream.calls.lock().await.len(); - let response = send_with_headers( - &app, - "POST", - "/v1/chat/completions", - Some(json!({ - "model": ROUTE_MODEL, - "messages": [{"role": "user", "content": content}] - })), - headers, - ) - .await?; - assert_eq!(response.status, StatusCode::OK); - let expected_model = expected_calls.last().copied(); - assert_eq!( - response - .headers - .get("x-model-router-selected-model") - .and_then(|value| value.to_str().ok()), - expected_model - ); - let calls = upstream.calls.lock().await; - assert_eq!( - calls[previous_call_count..] - .iter() - .map(|call| call["model"].as_str().unwrap_or("")) - .collect::>(), - expected_calls - ); - } - Ok(()) -} - #[tokio::test] async fn unavailable_target_fails_over_across_endpoints_and_stops_when_exhausted() -> TestResult { let upstream = MockUpstream::start().await?;