diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index ef56b27064..820d370650 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -6,7 +6,6 @@ use std::borrow::Cow; use std::collections::{BTreeMap, BTreeSet, HashMap}; -use std::future::ready; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, SystemTime}; @@ -563,13 +562,21 @@ impl TranslatingLlmClient { if status.is_success() { if streaming { let upstream_headers = response.headers().clone(); - let chunks = match prepare_response_stream(response, backend, model).await { + let chunks = match prepare_response_stream(response, backend, metadata, model).await + { Ok(chunks) => chunks, Err(error) => { - metrics::record_upstream_attempt(None); + // An in-band provider rejection can follow HTTP 200; record its status. + let provider_status = match &error { + LlmClientError::UpstreamHttp { status, .. } => Some(*status), + _ => None, + }; + metrics::record_upstream_attempt( + provider_status.as_ref().map(StatusCode::as_u16), + ); return Err(AttemptFailure { error, - status: Some(status), + status: Some(provider_status.unwrap_or(status)), retry_after: None, }); } @@ -913,9 +920,11 @@ fn stream_with_deadline( .boxed() } +/// Check initial stream events before returning them, while fallback is still possible. async fn prepare_response_stream( response: reqwest::Response, backend: &Backend, + metadata: Option<&Metadata>, model: &ModelId, ) -> Result { let bytes = response.bytes_stream().map(|chunk| { @@ -924,42 +933,73 @@ async fn prepare_response_stream( .map_err(convert_reqwest_error) }); let mut chunks = decode_stream(bytes, backend.wire_format())?; - match chunks.next().await { - None => Ok(stream::empty().boxed()), - // Nothing has reached the caller, so transport failures can still be retried. - Some(Err(error @ (LlmClientError::Transport { .. } | LlmClientError::Timeout { .. }))) => { - Err(error) - } - Some(first) => { - // An in-band context overflow skips retries and advances to another candidate. - if let Some(message) = first_event_overflow(&first, backend) { - return Err(LlmClientError::ContextWindowExceeded { - model: model.clone(), - message, - }); + let mut prefix = Vec::with_capacity(2); + while let Some(next) = chunks.next().await { + match next { + // Nothing has reached the caller, so transport failures can still be retried. + Err(error @ (LlmClientError::Transport { .. } | LlmClientError::Timeout { .. })) => { + return Err(error); + } + Ok(event) => { + if let Some(error) = pre_answer_error(&event, backend, metadata, model) { + return Err(error); + } + // OpenAI streams often send a role-only prelude before the answer. + let hold_role = prefix.is_empty() + && matches!(backend, Backend::OpenAiChat(_)) + && !event.normalized().is_empty() + && event + .normalized() + .iter() + .all(|chunk| matches!(chunk, LlmResponseChunk::MessageStart { .. })); + prefix.push(Ok(event)); + if !hold_role { + break; + } + } + Err(error) => { + prefix.push(Err(error)); + break; } - Ok(stream::once(ready(first)).chain(chunks).boxed()) } } + Ok(stream::iter(prefix).chain(chunks).boxed()) } -// The overflow message when a stream's first event is an in-band provider rejection -// of the whole request, rather than the start of a response. -fn first_event_overflow( - first: &Result, +/// Classify a provider rejection before any stream event reaches the caller. +fn pre_answer_error( + event: &LlmResponseStreamEvent, backend: &Backend, -) -> Option { - first - .as_ref() - .ok()? - .normalized() - .iter() - .find_map(|chunk| match chunk { - LlmResponseChunk::StreamError { message } if backend.is_context_overflow(message) => { - Some(message.clone()) - } - _ => None, - }) + metadata: Option<&Metadata>, + model: &ModelId, +) -> Option { + let message = event.normalized().iter().find_map(|chunk| match chunk { + LlmResponseChunk::StreamError { message } => Some(message), + _ => None, + })?; + if backend.is_context_overflow(message) { + return Some(LlmClientError::ContextWindowExceeded { + model: model.clone(), + message: message.clone(), + }); + } + if !matches!(backend, Backend::OpenAiChat(_)) { + return None; + } + let code = event + .preservation()? + .raw() + .get("error")? + .get("code")? + .as_u64() + .and_then(|code| u16::try_from(code).ok())?; + if !metrics::is_retryable_http_status(code) { + return None; + } + Some(LlmClientError::UpstreamHttp { + status: StatusCode::from_u16(code).ok()?, + body: redact_forwarded_headers(message.clone(), metadata, backend.is_forwarding_auth()), + }) } // Uses Retry-After when supplied, capped so an upstream cannot stall a request indefinitely. @@ -2862,6 +2902,48 @@ mod tests { Ok(()) } + #[tokio::test] + async fn retry_exhaustion_preserves_stream_error_message_without_forwarded_credentials() + -> Result<()> { + let server = MockServer::start().await; + let sse = "data: {\"error\":{\"code\":503,\"message\":\"provider overloaded for Bearer client-key\"}}\n\n"; + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .and(wiremock::matchers::header( + "authorization", + "Bearer client-key", + )) + .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream")) + .expect(2) + .mount(&server) + .await; + + let mut backend_config = forwarding_config(&format!("{}/v1", server.uri())); + backend_config.max_retries = 1; + let client = TranslatingLlmClient::new(&[ModelConfig::new( + "gpt", + Backend::OpenAiChat(backend_config), + None, + )])?; + let mut headers = HeaderMap::new(); + headers.insert( + "authorization", + http::HeaderValue::from_static("Bearer client-key"), + ); + let mut request = request_with_headers("gpt", headers); + request.llm_request.stream = true; + + let Err(error) = client.call_rewrite_model(request, None).await else { + panic!("expected an upstream stream error"); + }; + let LlmClientError::UpstreamHttp { status, body } = error else { + panic!("expected the upstream stream error, got {error:?}"); + }; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(body, "provider overloaded for [REDACTED]"); + Ok(()) + } + #[tokio::test] async fn streaming_openai_chat_preserves_usage_opt_out() -> std::result::Result<(), Box> { diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index d660f8c987..1a9e05eefe 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -2928,10 +2928,9 @@ mod tests { .map_err(|source| LibsyError::client_call("strong", source))?; assert_eq!(completion_text(&aggregate), "ok"); - // Only a classified overflow re-routes; any other in-band error event still - // streams through to the consumer without trying another candidate. + // A non-retryable in-band error still reaches the caller. let (_server, calls, result) = run_streaming_candidates( - "data: {\"error\":{\"code\":500,\"message\":\"engine exploded\",\"object\":\"error\"},\"model\":\"weak\"}\n\ndata: [DONE]\n\n", + "data: {\"error\":{\"code\":401,\"message\":\"invalid key\",\"object\":\"error\"},\"model\":\"weak\"}\n\ndata: [DONE]\n\n", ) .await; let (_, response) = result?; @@ -2941,7 +2940,7 @@ mod tests { .into_agg() .await .expect_err("the in-band error should surface to the consumer"); - assert!(error.to_string().contains("engine exploded")); + assert!(error.to_string().contains("invalid key")); // Once content has streamed, a retry would repeat delivered output, so a later // overflow event keeps today's semantics and surfaces during aggregation. @@ -2975,4 +2974,78 @@ mod tests { assert_eq!(completion_text(&aggregate), ""); Ok(()) } + + /// A role-only prelude leaves fallback available for a retryable provider error. + #[tokio::test] + async fn pre_answer_retryable_stream_errors_fall_back_to_the_next_candidate() -> Result<()> { + for weak_sse in [ + "data: {\"error\":{\"code\":503,\"message\":\"provider overloaded\"}}\n\n", + "data: {\"error\":{\"code\":429,\"message\":\"rate limited\"}}\n\n", + "data: {\"id\":\"turn\",\"model\":\"weak\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\ndata: {\"error\":{\"code\":503,\"message\":\"provider overloaded\"}}\n\n", + ] { + let (_server, calls, result) = run_streaming_candidates(weak_sse).await; + let (_, response) = result?; + assert_eq!(&*calls.lock(), &["weak", "strong"]); + assert_eq!(response.served_model().map(ModelId::as_str), Some("strong")); + let aggregate = response + .llm_response + .into_agg() + .await + .map_err(|source| LibsyError::client_call("strong", source))?; + assert_eq!(completion_text(&aggregate), "ok"); + } + Ok(()) + } + + /// Once answer content arrives, later errors must not replay the request on another model. + #[tokio::test] + async fn stream_fallback_stops_at_answer_content_and_non_retryable_errors() -> Result<()> { + for weak_sse in [ + "data: {\"error\":{\"code\":400,\"message\":\"invalid request\"}}\n\n", + "data: {\"error\":{\"code\":\"busy\",\"message\":\"not an HTTP status\"}}\n\n", + "data: {\"id\":\"turn\",\"model\":\"weak\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"partial\"},\"finish_reason\":null}]}\n\ndata: {\"error\":{\"code\":503,\"message\":\"provider overloaded\"}}\n\n", + ] { + let (_server, calls, result) = run_streaming_candidates(weak_sse).await; + let (_, response) = result?; + assert_eq!(&*calls.lock(), &["weak"]); + response + .llm_response + .into_agg() + .await + .expect_err("the stream error should reach the caller"); + } + + let (_server, calls, result) = run_streaming_candidates(STRONG_SSE).await; + let (_, response) = result?; + assert_eq!(&*calls.lock(), &["weak"]); + let aggregate = response + .llm_response + .into_agg() + .await + .map_err(|source| LibsyError::client_call("weak", source))?; + assert_eq!(completion_text(&aggregate), "ok"); + + let (_server, calls, result) = run_streaming_candidates( + "data: {\"id\":\"turn\",\"model\":\"weak\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\ndata: {\"id\":\"turn\",\"model\":\"weak\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n", + ) + .await; + let (_, response) = result?; + assert_eq!(&*calls.lock(), &["weak"]); + let LlmResponse::Stream(stream) = response.llm_response else { + panic!("expected a stream"); + }; + let events = stream.collect::>().await; + assert_eq!(events.len(), 2); + assert!(matches!( + events[0].as_ref().expect("role event").normalized(), + [LlmResponseChunk::MessageStart { .. }] + )); + assert!(events[1] + .as_ref() + .expect("answer event") + .normalized() + .iter() + .any(|chunk| matches!(chunk, LlmResponseChunk::TextDelta { text, .. } if text == "ok"))); + Ok(()) + } } diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 9ca3687699..e099f58358 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -10,7 +10,7 @@ //! and model name. Counters are cumulative across flushes; the helpers take the //! latest (max) matching data point. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::fmt; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, OnceLock}; @@ -39,7 +39,10 @@ use switchyard_libsy::{ LlmClassifierConfig, LlmTaskClassifier, PickerMode, RoutingOutcome, RuntimeModels, StageRouter, StageRouterConfig, Step, TaskClassifierConfig, }; -use switchyard_llm_client::{ClientRouter, RunObservation, RunObserver}; +use switchyard_llm_client::{ + Backend, ClientRouter, HttpBackendConfig, ModelConfig, RawResponse, RunObservation, + RunObserver, TranslatingLlmClient, +}; use switchyard_protocol::{Category, ModelId}; use switchyard_protocol::{ ContentBlock, LlmRequest, LlmResponse, Message, Metadata, Request, Response, Role, @@ -49,6 +52,8 @@ use switchyard_protocol::{ LlmClientError, LlmResponseChunk, LlmResponseStreamEvent, StopReason, text_request, text_response, }; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; #[derive(Debug, thiserror::Error)] #[error("{0}")] @@ -976,6 +981,165 @@ async fn affinity_keeps_the_algorithm_selection_after_client_fallback() Ok(()) } +/// A pre-answer stream rejection records its provider status like an HTTP rejection. +#[tokio::test] +async fn pre_answer_stream_error_records_provider_status() +-> Result<(), Box> { + let _guard = serialize_test().lock().await; + let (store, exporter, provider, _, _) = telemetry(); + let before = flushed_metrics(exporter, provider); + let attempts_503_before = u64_counter_value( + &before, + "switchyard.upstream_attempts", + &[("outcome", "retryable_error"), ("code", "503")], + ) + .unwrap_or_default(); + let attempts_200_before = u64_counter_value( + &before, + "switchyard.upstream_attempts", + &[("outcome", "ok"), ("code", "200")], + ) + .unwrap_or_default(); + let transport_before = u64_counter_value( + &before, + "switchyard.upstream_attempts", + &[("outcome", "retryable_error"), ("code", "none")], + ) + .unwrap_or_default(); + + for (model, http_status, body, expected_503, expected_200) in [ + ( + "obs-in-band-503", + 200, + "data: {\"error\":{\"code\":503,\"message\":\"provider overloaded\"}}\n\n", + 1, + 0, + ), + ( + "obs-http-503", + 503, + "{\"error\":{\"message\":\"provider overloaded\"}}", + 2, + 0, + ), + ( + "obs-valid-stream-200", + 200, + "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\ndata: [DONE]\n\n", + 2, + 1, + ), + ] { + let server = MockServer::start().await; + let content_type = if http_status == 503 { + "application/json" + } else { + "text/event-stream" + }; + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(ResponseTemplate::new(http_status).set_body_raw(body, content_type)) + .expect(1) + .mount(&server) + .await; + let client = TranslatingLlmClient::new(&[ModelConfig::new( + model, + Backend::OpenAiChat(HttpBackendConfig { + base_url: format!("{}/v1", server.uri()), + api_key: None, + forward_auth: false, + extra_headers: BTreeMap::new(), + extra_body: BTreeMap::new(), + omit_body_fields: BTreeSet::new(), + reasoning_effort: None, + max_retries: 0, + failure_cooldown: Duration::ZERO, + timeout: None, + }), + None, + )])?; + let response = client + .call_rewrite_model_raw( + json!({ + "model": model, + "messages": [{"role": "user", "content": "hello"}], + "stream": true + }), + None, + None, + WireFormat::OpenAiChat, + ) + .await; + let expected_status = if model == "obs-valid-stream-200" { + match response? { + RawResponse::Stream(mut stream) => { + while let Some(event) = stream.next().await { + event?; + } + } + RawResponse::Buffered(_) => panic!("expected a stream"), + } + "200" + } else { + match response { + Err(LlmClientError::UpstreamHttp { status, .. }) => { + assert_eq!(status.as_u16(), 503); + } + _ => panic!("expected a provider rejection"), + } + "503" + }; + let attempt = find_span(&store.spans(), "libsy.upstream_attempt", "model", model); + assert_eq!( + attempt.fields.get("status_code").map(String::as_str), + Some(expected_status), + "{model}" + ); + assert_eq!( + attempt.fields.get("outcome").map(String::as_str), + Some(if expected_status == "200" { + "ok" + } else { + "error" + }), + "{model}" + ); + + let snapshots = flushed_metrics(exporter, provider); + assert_eq!( + u64_counter_value( + &snapshots, + "switchyard.upstream_attempts", + &[("outcome", "retryable_error"), ("code", "503")], + ) + .unwrap_or_default(), + attempts_503_before + expected_503, + "{model}" + ); + assert_eq!( + u64_counter_value( + &snapshots, + "switchyard.upstream_attempts", + &[("outcome", "ok"), ("code", "200")], + ) + .unwrap_or_default(), + attempts_200_before + expected_200, + "{model}" + ); + assert_eq!( + u64_counter_value( + &snapshots, + "switchyard.upstream_attempts", + &[("outcome", "retryable_error"), ("code", "none")], + ) + .unwrap_or_default(), + transport_before, + "{model}" + ); + } + Ok(()) +} + #[tokio::test] async fn successful_run_records_metrics_spans_and_outcome_metadata() -> switchyard_libsy::Result<()> { diff --git a/docs/internal/metrics_reference.md b/docs/internal/metrics_reference.md index 6dea9bdc6e..c5a69f8c52 100644 --- a/docs/internal/metrics_reference.md +++ b/docs/internal/metrics_reference.md @@ -99,13 +99,15 @@ In addition, `switchyard_client_responses_total` records `outcome="client_discon | Metric | Type | Meaning | |---|---|---| | `switchyard_client_responses_total{outcome}` | counter | HTTP responses returned to clients on the LLM-serving routes (`/v1/chat/completions`, `/v1/messages`, `/v1/responses`). The denominator for the **router-served** error rate. | -| `switchyard_upstream_attempts_total{outcome,code}` | counter | Individual upstream call attempts. One client request can produce N attempts via retry. The denominator for the **direct-to-endpoint** baseline error rate. The `code` label carries the raw upstream HTTP status for plotting the error-code distribution (see below). | +| `switchyard_upstream_attempts_total{outcome,code}` | counter | Individual upstream call attempts. One client request can produce N attempts via retry. The denominator for the **direct-to-endpoint** baseline error rate. The `code` label carries the upstream status, including a retryable status reported before answer content in an OpenAI Chat stream (see below). | | `switchyard_router_retry_recovered_total` | counter | Upstream operations that succeeded after at least one retry. | ### The `code` label on `switchyard_upstream_attempts_total` -`code` is the raw upstream HTTP status as a string: `"200"`, `"429"`, -`"500"`, `"504"`, etc. Two special values: +`code` is normally the upstream HTTP status as a string: `"200"`, `"429"`, +`"500"`, `"504"`, etc. If an OpenAI Chat stream returns HTTP 200 but reports a +retryable error before answer content, `code` is the status in that error frame. +Two special values: * `code="none"`: a non-HTTP failure (network error, connection reset, pre-status timeout). The attempt never received a status line, so there