diff --git a/Cargo.lock b/Cargo.lock index b1cbce1be..ef6ca08ab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2509,6 +2509,7 @@ dependencies = [ "async-stream", "base64", "futures", + "http", "pretty_assertions", "serde", "serde_json", diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 948838f48..c95a3291d 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -804,8 +804,10 @@ fn first_event_overflow( .normalized() .iter() .find_map(|chunk| match chunk { - LlmResponseChunk::StreamError { message } if backend.is_context_overflow(message) => { - Some(message.clone()) + LlmResponseChunk::StreamError { error } + if backend.is_context_overflow(&error.message) => + { + Some(error.message.clone()) } _ => None, }) diff --git a/crates/libsy-llm-client/src/observability.rs b/crates/libsy-llm-client/src/observability.rs index c55d46ad8..8fc8e9bef 100644 --- a/crates/libsy-llm-client/src/observability.rs +++ b/crates/libsy-llm-client/src/observability.rs @@ -300,8 +300,12 @@ impl ClientStreamObserver { record_client_error(&self.span, "response_translation", message); self.outcome = Outcome::Failed; } - LlmResponseChunk::StreamError { message } => { - record_client_error(&self.span, "502", message); + LlmResponseChunk::StreamError { error } => { + record_client_error( + &self.span, + error.effective_http_status().as_str(), + &error.message, + ); self.outcome = Outcome::Failed; } _ => {} diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 01e1c36fc..11ba8ecd2 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -347,10 +347,10 @@ async fn buffer_routing_stream( LlmResponseChunk::DecodeError { message } => { return Err(LlmClientError::ResponseTranslation(message.clone())); } - LlmResponseChunk::StreamError { message } => { + LlmResponseChunk::StreamError { error } => { return Err(LlmClientError::UpstreamHttp { status: StatusCode::BAD_GATEWAY, - body: message.clone(), + body: error.upstream_http_body(), }); } _ => {} @@ -1015,7 +1015,9 @@ mod tests { text: "partial".to_string(), }, LlmResponseChunk::StreamError { - message: "stream failed".to_string(), + error: Box::new(switchyard_protocol::StreamErrorDetails::new( + "stream failed", + )), }, ])), }; @@ -1942,6 +1944,61 @@ mod tests { } } + #[tokio::test] + async fn routing_buffer_keeps_502_fallback_and_structured_code_visibility() { + for embedded_status in [400, 429, 99] { + let chunks: LlmResponseStream = stream::iter([Ok(LlmResponseChunk::StreamError { + error: Box::new(switchyard_protocol::StreamErrorDetails { + status: Some(embedded_status), + error_type: Some("invalid_request_error".into()), + code: Some(json!("content_policy_violation")), + param: Some(json!("input")), + message: "blocked".into(), + }), + } + .into())]) + .boxed(); + + let source = match buffer_routing_stream(chunks).await { + Err(source) => source, + Ok(_) => panic!("routing buffer should fail"), + }; + let LlmClientError::UpstreamHttp { status, body } = &source else { + panic!("expected upstream HTTP error"); + }; + assert_eq!(*status, StatusCode::BAD_GATEWAY, "{embedded_status}"); + let body: serde_json::Value = + serde_json::from_str(body).expect("normalized error body"); + assert_eq!(body["error"]["code"], "content_policy_violation"); + + let error = LibsyError::client_call("target", source); + assert_eq!( + fallback_reason(&error), + Some(RoutingFallbackReason::Unavailable), + "{embedded_status}" + ); + } + + let policy = switchyard_protocol::StreamErrorDetails { + status: Some(400), + error_type: Some("invalid_request_error".into()), + code: Some(json!("content_policy_violation")), + param: None, + message: "blocked".into(), + }; + let error = LibsyError::client_call( + "target", + LlmClientError::UpstreamHttp { + status: policy.effective_http_status(), + body: policy.upstream_http_body(), + }, + ); + assert_eq!( + fallback_reason(&error), + Some(RoutingFallbackReason::Unavailable) + ); + } + #[tokio::test] async fn candidate_failures_follow_the_fallback_policy() -> Result<()> { // Context overflow is retryable across candidates. diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 8150564f9..998144856 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -45,8 +45,8 @@ use switchyard_protocol::{ RoutedLlmClient, ToolCall, ToolResult, Usage, WireFormat, }; use switchyard_protocol::{ - LlmClientError, LlmResponseChunk, LlmResponseStreamEvent, StopReason, text_request, - text_response, + LlmClientError, LlmResponseChunk, LlmResponseStreamEvent, StopReason, StreamErrorDetails, + text_request, text_response, }; #[derive(Debug, thiserror::Error)] @@ -1286,6 +1286,33 @@ async fn observed_run_reports_one_successful_routed_call() -> switchyard_libsy:: /// A streamed response keeps the client span available until terminal usage arrives. struct StreamingUsageClient; +struct StreamStatusClient { + status: u16, +} + +#[async_trait] +impl RoutedLlmClient for StreamStatusClient { + async fn call(&self, _request: Request) -> Result { + Ok(Response { + llm_response: LlmResponse::Stream( + futures::stream::iter([Ok(LlmResponseChunk::StreamError { + error: Box::new(StreamErrorDetails { + status: Some(self.status), + error_type: Some("rate_limit_error".into()), + code: Some(json!("quota")), + param: None, + message: "slow down".into(), + }), + } + .into())]) + .boxed(), + ), + metadata: None, + upstream_headers: http::HeaderMap::new(), + }) + } +} + #[async_trait] impl RoutedLlmClient for StreamingUsageClient { async fn call(&self, request: Request) -> Result { @@ -1398,6 +1425,35 @@ async fn streamed_usage_updates_the_client_call_span() -> switchyard_libsy::Resu Ok(()) } +#[tokio::test] +async fn streamed_error_observability_uses_embedded_or_fallback_status() +-> switchyard_libsy::Result<()> { + let _guard = serialize_test().lock().await; + for (suffix, status, expected) in [("429", 429, "429"), ("invalid", 99, "502")] { + let (store, _, _, _, _) = telemetry(); + let model = format!("stream-status-{suffix}"); + let (_, response) = run( + algo("stream-status-algo", &model), + Arc::new(StreamStatusClient { status }), + request_with_metadata("stream-status-session", "stream-status-corr"), + ) + .await?; + response + .llm_response + .into_agg() + .await + .expect_err("structured stream should fail"); + + let spans = store.spans(); + let client_span = find_span(&spans, "libsy.client_call", "selected_model", &model); + assert_eq!( + client_span.fields.get("error.type").map(String::as_str), + Some(expected) + ); + } + Ok(()) +} + #[tokio::test] async fn dropped_stream_records_cancelled_outcome() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; diff --git a/crates/libsy/src/algorithms/advisor_gate/tests.rs b/crates/libsy/src/algorithms/advisor_gate/tests.rs index 72d39615b..5ff3cb7d4 100644 --- a/crates/libsy/src/algorithms/advisor_gate/tests.rs +++ b/crates/libsy/src/algorithms/advisor_gate/tests.rs @@ -727,7 +727,9 @@ async fn mid_stream_error_propagates_while_buffering() { text: "partial".to_string(), }]), LlmResponseStreamEvent::new(vec![LlmResponseChunk::StreamError { - message: "upstream reset".to_string(), + error: Box::new(switchyard_protocol::StreamErrorDetails::new( + "upstream reset", + )), }]), ])) }; diff --git a/crates/libsy/src/algorithms/util/buffered_response.rs b/crates/libsy/src/algorithms/util/buffered_response.rs index 706a117d6..e72871a13 100644 --- a/crates/libsy/src/algorithms/util/buffered_response.rs +++ b/crates/libsy/src/algorithms/util/buffered_response.rs @@ -66,10 +66,10 @@ pub(crate) async fn buffer_response( LlmResponseChunk::DecodeError { message } => { Some(LlmClientError::ResponseTranslation(message.clone())) } - LlmResponseChunk::StreamError { message } => { + LlmResponseChunk::StreamError { error } => { Some(LlmClientError::UpstreamHttp { - status: http::StatusCode::BAD_GATEWAY, - body: message.clone(), + status: error.effective_http_status(), + body: error.upstream_http_body(), }) } chunk => { diff --git a/crates/libsy/src/algorithms/util/llm_judge.rs b/crates/libsy/src/algorithms/util/llm_judge.rs index 4798178e9..658728f45 100644 --- a/crates/libsy/src/algorithms/util/llm_judge.rs +++ b/crates/libsy/src/algorithms/util/llm_judge.rs @@ -616,7 +616,9 @@ mod tests { text: "{\"ok\":".to_string(), }, LlmResponseChunk::StreamError { - message: "upstream exploded".to_string(), + error: Box::new(switchyard_protocol::StreamErrorDetails::new( + "upstream exploded", + )), }, ]; assert_eq!(score_served_with(Ok(streamed(chunks))).await?, "no-verdict"); diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 119809278..7027aa812 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -834,7 +834,9 @@ mod tests { text: "partial".to_string(), }, LlmResponseChunk::StreamError { - message: "upstream exploded".to_string(), + error: Box::new(switchyard_protocol::StreamErrorDetails::new( + "upstream exploded", + )), }, ]); let (_, response) = test_drive(orch, request(), serve).await?; diff --git a/crates/protocol/src/stream.rs b/crates/protocol/src/stream.rs index c59a66f77..812dff1aa 100644 --- a/crates/protocol/src/stream.rs +++ b/crates/protocol/src/stream.rs @@ -10,7 +10,7 @@ use std::pin::Pin; use futures::{Stream, StreamExt}; use http::StatusCode; -use serde::{Deserialize, Serialize}; +use serde::{Deserialize, Deserializer, Serialize}; use serde_json::Value; use thiserror::Error; @@ -20,9 +20,8 @@ use crate::{ llm::{AggLlmResponse, ContentBlock, ResponseOutput, Role, StopReason, ToolCall, Usage}, }; -/// Status reported for an upstream error delivered inside a streaming body. The -/// upstream already sent a success status line before failing, so there is no real -/// code to propagate; 502 matches how a failed upstream call surfaces elsewhere. +/// Fallback status for an upstream error delivered inside a streaming body when the +/// event does not embed a valid HTTP status. const MID_STREAM_UPSTREAM_STATUS: StatusCode = StatusCode::BAD_GATEWAY; /// Why a translated event stream stopped early. @@ -283,9 +282,9 @@ fn push_checked_chunk( LlmResponseChunk::DecodeError { message } => { Err(LlmClientError::ResponseTranslation(message)) } - LlmResponseChunk::StreamError { message } => Err(LlmClientError::UpstreamHttp { - status: MID_STREAM_UPSTREAM_STATUS, - body: message, + LlmResponseChunk::StreamError { error } => Err(LlmClientError::UpstreamHttp { + status: error.effective_http_status(), + body: error.upstream_http_body(), }), chunk => { accumulator.push(chunk); @@ -294,6 +293,65 @@ fn push_checked_chunk( } } +/// Provider-neutral details from an error delivered inside a response stream. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct StreamErrorDetails { + /// HTTP-like status reported inside the stream event. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub status: Option, + /// Provider error category, represented as `type` on provider wires. + #[serde(rename = "type", default, skip_serializing_if = "Option::is_none")] + pub error_type: Option, + /// Provider error code, retained without narrowing its JSON type. + #[serde( + default, + skip_serializing_if = "Option::is_none", + deserialize_with = "deserialize_optional_json_value" + )] + pub code: Option, + /// Request parameter associated with the failure, retained as JSON. + #[serde( + default, + skip_serializing_if = "Option::is_none", + deserialize_with = "deserialize_optional_json_value" + )] + pub param: Option, + /// Human-readable failure description. + pub message: String, +} + +fn deserialize_optional_json_value<'de, D>(deserializer: D) -> Result, D::Error> +where + D: Deserializer<'de>, +{ + Value::deserialize(deserializer).map(Some) +} + +impl StreamErrorDetails { + /// Creates a message-only error synthesized inside Switchyard. + pub fn new(message: impl Into) -> Self { + Self { + status: None, + error_type: None, + code: None, + param: None, + message: message.into(), + } + } + + /// Returns the embedded valid status or the mid-stream fallback. + pub fn effective_http_status(&self) -> StatusCode { + self.status + .and_then(|status| StatusCode::from_u16(status).ok()) + .unwrap_or(MID_STREAM_UPSTREAM_STATUS) + } + + /// Builds a normalized error body without forwarding a raw provider body. + pub fn upstream_http_body(&self) -> String { + serde_json::json!({"error": self}).to_string() + } +} + /// One provider-neutral streaming response chunk. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub enum LlmResponseChunk { @@ -352,8 +410,8 @@ pub enum LlmResponseChunk { }, /// Reports an upstream failure delivered inside an otherwise successful stream. StreamError { - /// Human-readable upstream failure. - message: String, + /// Provider-neutral error details. + error: Box, }, } @@ -620,7 +678,7 @@ mod tests { crate::WireFormat::OpenAiChat, json!({"error": {"message": "provider failed"}}), vec![LlmResponseChunk::StreamError { - message: "provider failed".to_string(), + error: Box::new(StreamErrorDetails::new("provider failed")), }], ), )]))); @@ -635,6 +693,63 @@ mod tests { )); } + #[test] + fn structured_stream_error_round_trips_without_narrowing_json_fields() { + let chunk = LlmResponseChunk::StreamError { + error: Box::new(StreamErrorDetails { + status: Some(429), + error_type: Some("rate_limit_error".to_string()), + code: Some(json!({"bucket": 7})), + param: Some(Value::Null), + message: "slow down".to_string(), + }), + }; + let encoded = serde_json::to_value(&chunk).expect("serialize stream error"); + assert_eq!(encoded["StreamError"]["error"]["type"], "rate_limit_error"); + assert!(encoded["StreamError"]["error"].get("error_type").is_none()); + let decoded: LlmResponseChunk = + serde_json::from_value(encoded).expect("deserialize stream error"); + assert_eq!(decoded, chunk); + } + + #[test] + fn stream_error_status_uses_valid_embedded_status_or_bad_gateway() { + let mut error = StreamErrorDetails::new("failed"); + assert_eq!(error.effective_http_status(), StatusCode::BAD_GATEWAY); + error.status = Some(429); + assert_eq!(error.effective_http_status(), StatusCode::TOO_MANY_REQUESTS); + error.status = Some(99); + assert_eq!(error.effective_http_status(), StatusCode::BAD_GATEWAY); + } + + #[test] + fn aggregating_stream_errors_uses_embedded_or_fallback_status_and_keeps_code() { + for (status, expected) in [ + (Some(429), StatusCode::TOO_MANY_REQUESTS), + (Some(99), StatusCode::BAD_GATEWAY), + ] { + let response = LlmResponse::Stream(Box::pin(stream::once(async move { + Ok(LlmResponseChunk::StreamError { + error: Box::new(StreamErrorDetails { + status, + error_type: Some("invalid_request_error".into()), + code: Some(json!("content_policy_violation")), + param: Some(json!("input")), + message: "blocked".into(), + }), + } + .into()) + }))); + let error = block_on(response.into_agg()).expect_err("stream should fail"); + let LlmClientError::UpstreamHttp { status, body } = error else { + panic!("expected upstream HTTP error"); + }; + assert_eq!(status, expected); + let body: Value = serde_json::from_str(&body).expect("normalized error body"); + assert_eq!(body["error"]["code"], "content_policy_violation"); + } + } + #[test] fn assembles_tool_calls_by_index() { // id/name arrive once, arguments stream across deltas and parse as JSON. diff --git a/crates/switchyard-nemo-relay-plugin/src/runtime.rs b/crates/switchyard-nemo-relay-plugin/src/runtime.rs index 95794f0d8..b07b42a53 100644 --- a/crates/switchyard-nemo-relay-plugin/src/runtime.rs +++ b/crates/switchyard-nemo-relay-plugin/src/runtime.rs @@ -5,7 +5,6 @@ use std::pin::Pin; use std::sync::{Arc, Mutex}; use futures_util::{Stream, StreamExt}; -use http::StatusCode; use nemo_relay_plugin::{ DataSchema, Json, LlmRequest as RelayRequest, LogSeverity, MetricKind, MetricMeasurement, MetricValueType, PluginRuntime, @@ -602,9 +601,9 @@ fn normalized_stream_error(chunks: &[LlmResponseChunk]) -> Option { Some(LlmClientError::ResponseTranslation(message.clone())) } - LlmResponseChunk::StreamError { message } => Some(LlmClientError::UpstreamHttp { - status: StatusCode::BAD_GATEWAY, - body: message.clone(), + LlmResponseChunk::StreamError { error } => Some(LlmClientError::UpstreamHttp { + status: error.effective_http_status(), + body: error.upstream_http_body(), }), _ => None, }) @@ -1733,15 +1732,23 @@ mod tests { } #[tokio::test] - async fn normalized_stream_error_emits_failure_telemetry_once() { - assert_normalized_stream_failure( - LlmResponseChunk::StreamError { - message: "provider response body".into(), - }, - "upstream_http", - Some(502), - ) - .await; + async fn normalized_stream_error_uses_embedded_or_fallback_status() { + for (status, expected) in [(429, 429), (99, 502)] { + assert_normalized_stream_failure( + LlmResponseChunk::StreamError { + error: Box::new(switchyard_protocol::StreamErrorDetails { + status: Some(status), + error_type: Some("rate_limit_error".into()), + code: Some(json!("quota")), + param: None, + message: "provider response body".into(), + }), + }, + "upstream_http", + Some(expected), + ) + .await; + } } #[tokio::test] diff --git a/crates/switchyard-translation/Cargo.toml b/crates/switchyard-translation/Cargo.toml index bb121d931..d765ca205 100644 --- a/crates/switchyard-translation/Cargo.toml +++ b/crates/switchyard-translation/Cargo.toml @@ -17,6 +17,7 @@ keywords = ["llm", "translation", "openai", "anthropic"] publish = ["crates-io"] [dependencies] +http.workspace = true base64.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/crates/switchyard-translation/src/codecs/anthropic/stream.rs b/crates/switchyard-translation/src/codecs/anthropic/stream.rs index cc3955f9c..2b713f8f2 100644 --- a/crates/switchyard-translation/src/codecs/anthropic/stream.rs +++ b/crates/switchyard-translation/src/codecs/anthropic/stream.rs @@ -5,17 +5,36 @@ use serde_json::{Map, Value, json}; -use crate::LlmResponseChunk; use crate::codecs::stream::{ - StreamCodec, StreamTranslationState, record_source_identity, + StreamCodec, StreamTranslationState, record_source_identity, stream_error_details, target_message_id_or_source_message_id, target_model_or_source_model, }; use crate::format::{FormatId, WireFormat}; use crate::util::{desanitize_anthropic_tool_use_id, sanitize_anthropic_tool_use_id}; +use crate::{LlmResponseChunk, StreamErrorDetails}; /// Stream codec for Anthropic Messages events. pub struct AnthropicMessagesStreamCodec; +fn anthropic_error_event(error: StreamErrorDetails) -> Value { + let StreamErrorDetails { + error_type, + code, + message, + .. + } = error; + let mut payload = Map::new(); + payload.insert( + "type".into(), + json!(error_type.unwrap_or_else(|| "api_error".into())), + ); + payload.insert("message".into(), json!(message)); + if let Some(Value::String(code)) = code { + payload.insert("code".into(), json!(code)); + } + json!({"type": "error", "error": payload}) +} + impl StreamCodec for AnthropicMessagesStreamCodec { fn format(&self) -> FormatId { WireFormat::AnthropicMessages.into() @@ -149,13 +168,11 @@ fn decode_anthropic_stream( reason: state.stop_reason.clone(), }], Some("error") => vec![LlmResponseChunk::StreamError { - message: object - .get("error") - .and_then(Value::as_object) - .and_then(|error| error.get("message")) - .and_then(Value::as_str) - .unwrap_or("unknown Anthropic stream error") - .to_string(), + error: Box::new(stream_error_details( + event, + object.get("error").unwrap_or(event), + "unknown Anthropic stream error", + )), }], _ => Vec::new(), } @@ -238,11 +255,16 @@ fn encode_anthropic_stream( state.stop_reason = reason.or_else(|| state.stop_reason.clone()); Vec::new() } - LlmResponseChunk::StreamError { message } | LlmResponseChunk::DecodeError { message } => { + LlmResponseChunk::DecodeError { message } => { // An in-band error is terminal: emit the error, then nothing further. state.finished = true; // finish() adds no success events state.errored = true; // the entry guard drops any later chunk - vec![json!({"type": "error", "error": {"message": message}})] + vec![anthropic_error_event(StreamErrorDetails::new(message))] + } + LlmResponseChunk::StreamError { error } => { + state.finished = true; + state.errored = true; + vec![anthropic_error_event(*error)] } } } @@ -259,7 +281,9 @@ fn finish_anthropic_stream(state: &mut StreamTranslationState) -> Vec { return encode_anthropic_stream( state, LlmResponseChunk::StreamError { - message: "Tool call ended without a non-empty ID and name".to_string(), + error: Box::new(StreamErrorDetails::new( + "Tool call ended without a non-empty ID and name", + )), }, ); } diff --git a/crates/switchyard-translation/src/codecs/openai_chat/stream.rs b/crates/switchyard-translation/src/codecs/openai_chat/stream.rs index 7a902627b..05185e050 100644 --- a/crates/switchyard-translation/src/codecs/openai_chat/stream.rs +++ b/crates/switchyard-translation/src/codecs/openai_chat/stream.rs @@ -5,18 +5,45 @@ use serde_json::{Map, Value, json}; -use crate::LlmResponseChunk; use crate::codecs::common::{first_nonempty_string, reasoning_text_from_details}; use crate::codecs::stream::{ - StreamCodec, StreamTranslationState, record_source_identity, state_source_is, string_field, - target_model_or_source_model, + StreamCodec, StreamTranslationState, record_source_identity, state_source_is, + stream_error_details, string_field, target_model_or_source_model, }; use crate::format::{FormatId, WireFormat}; use crate::llm::Usage; +use crate::{LlmResponseChunk, StreamErrorDetails}; /// Stream codec for OpenAI Chat Completions chunks. pub struct OpenAiChatStreamCodec; +fn openai_chat_error_event(error: StreamErrorDetails) -> Value { + let StreamErrorDetails { + status, + error_type, + code, + param, + message, + } = error; + let mut payload = Map::new(); + payload.insert( + "type".into(), + json!(error_type.unwrap_or_else(|| "api_error".into())), + ); + payload.insert("message".into(), json!(message)); + if let Some(code) = code { + payload.insert("code".into(), code); + } + if let Some(param) = param { + payload.insert("param".into(), param); + } + let mut event = json!({"error": payload}); + if let Some(status) = status { + event["status"] = json!(status); + } + event +} + impl StreamCodec for OpenAiChatStreamCodec { fn format(&self) -> FormatId { WireFormat::OpenAiChat.into() @@ -58,11 +85,11 @@ fn decode_openai_chat_stream( // `MessageStart` and the error text would be dropped. if let Some(error) = object.get("error") { return vec![LlmResponseChunk::StreamError { - message: error - .get("message") - .and_then(Value::as_str) - .unwrap_or("unknown OpenAI stream error") - .to_string(), + error: Box::new(stream_error_details( + event, + error, + "unknown OpenAI stream error", + )), }]; } @@ -310,11 +337,16 @@ fn encode_openai_chat_stream( state.saw_backend_usage.then(|| openai_usage_value(state)), )] } - LlmResponseChunk::DecodeError { message } | LlmResponseChunk::StreamError { message } => { + LlmResponseChunk::DecodeError { message } => { // An in-band error is terminal: emit the error, then nothing further. state.finished = true; // finish() adds no success events state.errored = true; // the entry guard drops any later chunk - vec![json!({"error": {"message": message}})] + vec![openai_chat_error_event(StreamErrorDetails::new(message))] + } + LlmResponseChunk::StreamError { error } => { + state.finished = true; + state.errored = true; + vec![openai_chat_error_event(*error)] } } } diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index a179d4ee7..e758412e4 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -4,22 +4,49 @@ //! Streaming codec for OpenAI Responses API events. use serde::Serialize; -use serde_json::{Value, json}; +use serde_json::{Map, Value, json}; -use crate::LlmResponseChunk; use crate::codecs::common::{ collect_responses_reasoning_text, encrypted_reasoning_data, encrypted_reasoning_item_id, }; use crate::codecs::stream::{ - StreamCodec, StreamTranslationState, record_source_identity, + StreamCodec, StreamTranslationState, record_source_identity, stream_error_details, target_message_id_or_source_message_id, target_model_or_source_model, }; use crate::format::{FormatId, WireFormat}; use crate::llm::Usage; +use crate::{LlmResponseChunk, StreamErrorDetails}; /// Stream codec for OpenAI Responses API events. pub struct OpenAiResponsesStreamCodec; +fn responses_error_event(error: StreamErrorDetails) -> Value { + let StreamErrorDetails { + status, + error_type, + code, + param, + message, + } = error; + let mut payload = Map::new(); + payload.insert( + "type".into(), + json!(error_type.unwrap_or_else(|| "api_error".into())), + ); + payload.insert("message".into(), json!(message)); + if let Some(code) = code { + payload.insert("code".into(), code); + } + if let Some(param) = param { + payload.insert("param".into(), param); + } + let mut event = json!({"type": "error", "error": payload}); + if let Some(status) = status { + event["status"] = json!(status); + } + event +} + impl StreamCodec for OpenAiResponsesStreamCodec { fn format(&self) -> FormatId { WireFormat::OpenAiResponses.into() @@ -299,22 +326,27 @@ fn decode_responses_stream( Some("response.incomplete") => vec![LlmResponseChunk::MessageStop { reason: Some("max_tokens".to_string()), }], - Some("response.failed") => vec![LlmResponseChunk::StreamError { - message: event - .get("response") - .and_then(|response| response.get("error")) - .and_then(|error| error.get("message")) - .and_then(Value::as_str) - .unwrap_or("unknown Responses stream error") - .to_string(), - }], - Some("error") => vec![LlmResponseChunk::StreamError { - message: event - .get("message") - .and_then(Value::as_str) - .unwrap_or("unknown Responses stream error") - .to_string(), - }], + Some("response.failed") => { + let payload = event.pointer("/response/error").unwrap_or(event); + vec![LlmResponseChunk::StreamError { + error: Box::new(stream_error_details( + event, + payload, + "unknown Responses stream error", + )), + }] + } + Some("error") => { + let nested = event.get("error"); + let payload = nested.unwrap_or(event); + let mut error = stream_error_details(event, payload, "unknown Responses stream error"); + if nested.is_none() { + error.error_type = None; + } + vec![LlmResponseChunk::StreamError { + error: Box::new(error), + }] + } _ => Vec::new(), } } @@ -394,11 +426,16 @@ fn encode_responses_stream( state.stop_reason = reason.or_else(|| state.stop_reason.clone()); Vec::new() } - LlmResponseChunk::DecodeError { message } | LlmResponseChunk::StreamError { message } => { + LlmResponseChunk::DecodeError { message } => { // An in-band error is terminal: emit the error, then nothing further. state.finished = true; // finish() adds no success events state.errored = true; // the entry guard drops any later chunk - vec![json!({"type": "error", "message": message})] + vec![responses_error_event(StreamErrorDetails::new(message))] + } + LlmResponseChunk::StreamError { error } => { + state.finished = true; + state.errored = true; + vec![responses_error_event(*error)] } } } @@ -788,7 +825,9 @@ fn snapshot_suffix( ) -> Result, LlmResponseChunk> { let Some(suffix) = snapshot.strip_prefix(decoded.as_str()) else { return Err(LlmResponseChunk::StreamError { - message: "Responses snapshot conflicts with streamed content".to_string(), + error: Box::new(StreamErrorDetails::new( + "Responses snapshot conflicts with streamed content", + )), }); }; if suffix.is_empty() { diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 87da2d649..74f77bdc5 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -6,10 +6,10 @@ use std::collections::BTreeMap; use std::sync::Arc; +use http::StatusCode; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value, json}; -use crate::LlmResponseChunk; use crate::codecs::anthropic::AnthropicMessagesStreamCodec; use crate::codecs::openai_chat::OpenAiChatStreamCodec; use crate::codecs::responses::OpenAiResponsesStreamCodec; @@ -17,6 +17,36 @@ use crate::engine::{FormatRegistry, TranslationEngine}; use crate::error::{Result, TranslationError}; use crate::format::{FormatId, WireFormat}; use crate::llm::Usage; +use crate::{LlmResponseChunk, StreamErrorDetails}; + +pub(crate) fn stream_error_details( + envelope: &Value, + payload: &Value, + fallback_message: &str, +) -> StreamErrorDetails { + fn numeric_status(value: &Value) -> Option { + value + .get("status") + .and_then(Value::as_u64) + .and_then(|status| u16::try_from(status).ok()) + .filter(|status| StatusCode::from_u16(*status).is_ok()) + } + + StreamErrorDetails { + status: numeric_status(envelope).or_else(|| numeric_status(payload)), + error_type: payload + .get("type") + .and_then(Value::as_str) + .map(str::to_owned), + code: payload.get("code").cloned(), + param: payload.get("param").cloned(), + message: payload + .get("message") + .and_then(Value::as_str) + .unwrap_or(fallback_message) + .to_owned(), + } +} /// Mutable state accumulated while translating one streaming response. #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index 95ebb75e7..15a5e0b44 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -559,7 +559,7 @@ mod tests { fn encode_stream_stops_after_an_in_band_error() -> Result<(), BoxError> { for message in [ LlmResponseChunk::StreamError { - message: "boom".to_string(), + error: Box::new(crate::StreamErrorDetails::new("boom")), }, LlmResponseChunk::DecodeError { message: "boom".to_string(), @@ -645,7 +645,7 @@ mod tests { WireFormat::OpenAiResponses, json!({"type": "error", "message": "boom"}), vec![LlmResponseChunk::StreamError { - message: "boom".to_string(), + error: Box::new(crate::StreamErrorDetails::new("boom")), }], ); let chunks: LlmResponseStream = stream::iter([Ok(error)]) @@ -851,12 +851,12 @@ mod tests { let bytes = stream::once(async move { Ok::, LlmClientError>(sse) }); let events = decode_all(bytes, WireFormat::OpenAiResponses)?; - let Some(LlmResponseChunk::StreamError { message }) = + let Some(LlmResponseChunk::StreamError { error }) = events.first().and_then(|event| event.normalized().first()) else { return Err("expected response.failed to decode as StreamError".into()); }; - assert_eq!(message, "upstream failed"); + assert_eq!(error.message, "upstream failed"); let chunks: LlmResponseStream = stream::iter( events diff --git a/crates/switchyard-translation/src/lib.rs b/crates/switchyard-translation/src/lib.rs index 4d0e81da6..4420a4cbc 100644 --- a/crates/switchyard-translation/src/lib.rs +++ b/crates/switchyard-translation/src/lib.rs @@ -21,7 +21,7 @@ pub mod util; pub use switchyard_protocol::stream::{ LlmResponseChunk, LlmResponseStream, LlmResponseStreamEvent, LlmStreamError, - ProviderStreamEvent, + ProviderStreamEvent, StreamErrorDetails, }; pub use switchyard_protocol::{format, llm}; diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index e22511c59..6b9a78e51 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -11,7 +11,8 @@ use pretty_assertions::assert_eq; use serde_json::{Value, json}; use switchyard_protocol::{LlmResponseStreamEvent, ResponseAccumulator, StopReason}; use switchyard_translation::{ - LlmResponseChunk, StreamTranslationState, TranslationEngine, WireFormat, decode_stream_event, + LlmResponseChunk, StreamErrorDetails, StreamTranslationState, TranslationEngine, WireFormat, + decode_stream_event, }; use common::{REASONING_MODEL, text_and_encrypted_reasoning_details}; @@ -1840,23 +1841,159 @@ fn anthropic_empty_tool_input_survives_stream_translation() -> TestResult { Ok(()) } -// An OpenAI-shaped error frame carries no `choices`, so it must decode to a stream error -// instead of a bare message start that silently drops the upstream message. #[test] -fn openai_chat_error_frame_decodes_to_stream_error() -> TestResult { - let mut state = StreamTranslationState::new(WireFormat::OpenAiChat, WireFormat::OpenAiChat); - let event = json!({"error": {"message": "upstream exploded", "type": "server_error"}}); - - let chunks = decode_stream_event(&mut state, WireFormat::OpenAiChat, &event); +fn provider_stream_errors_decode_to_structured_details() -> TestResult { + let cases = [ + ( + WireFormat::OpenAiChat, + json!({"status":429,"error":{"type":"rate_limit_error","code":"quota","param":null,"message":"slow down"}}), + StreamErrorDetails { + status: Some(429), + error_type: Some("rate_limit_error".into()), + code: Some(json!("quota")), + param: Some(Value::Null), + message: "slow down".into(), + }, + ), + ( + WireFormat::OpenAiResponses, + json!({"type":"error","status":99,"error":{"status":429,"type":"rate_limit_error","code":"quota","message":"nested status wins"}}), + StreamErrorDetails { + status: Some(429), + error_type: Some("rate_limit_error".into()), + code: Some(json!("quota")), + param: None, + message: "nested status wins".into(), + }, + ), + ( + WireFormat::OpenAiResponses, + json!({"type":"error","status":"invalid","error":{"status":429,"type":"error","code":"nested_error","message":"nested semantic type"}}), + StreamErrorDetails { + status: Some(429), + error_type: Some("error".into()), + code: Some(json!("nested_error")), + param: None, + message: "nested semantic type".into(), + }, + ), + ( + WireFormat::OpenAiResponses, + json!({"type":"error","status":400,"error":{"type":"invalid_request_error","code":"previous_response_not_found","param":"previous_response_id","message":"Previous response was not found."}}), + StreamErrorDetails { + status: Some(400), + error_type: Some("invalid_request_error".into()), + code: Some(json!("previous_response_not_found")), + param: Some(json!("previous_response_id")), + message: "Previous response was not found.".into(), + }, + ), + ( + WireFormat::OpenAiResponses, + json!({"type":"response.failed","status":503,"response":{"error":{"type":"server_error","code":503,"param":{"region":"west"},"message":"engine unavailable"}}}), + StreamErrorDetails { + status: Some(503), + error_type: Some("server_error".into()), + code: Some(json!(503)), + param: Some(json!({"region":"west"})), + message: "engine unavailable".into(), + }, + ), + ( + WireFormat::OpenAiResponses, + json!({"type":"error","status":500,"code":"server_error","param":null,"message":"flat failure"}), + StreamErrorDetails { + status: Some(500), + error_type: None, + code: Some(json!("server_error")), + param: Some(Value::Null), + message: "flat failure".into(), + }, + ), + ( + WireFormat::AnthropicMessages, + json!({"type":"error","error":{"type":"overloaded_error","code":"capacity","message":"busy"}}), + StreamErrorDetails { + status: None, + error_type: Some("overloaded_error".into()), + code: Some(json!("capacity")), + param: None, + message: "busy".into(), + }, + ), + ]; - assert_eq!(chunks.len(), 1); - match &chunks[0] { - LlmResponseChunk::StreamError { message } => assert_eq!(message, "upstream exploded"), - other => return Err(format!("expected StreamError, got {other:?}").into()), + for (format, event, expected) in cases { + let mut state = StreamTranslationState::new(format, format); + assert_eq!( + decode_stream_event(&mut state, format, &event), + vec![LlmResponseChunk::StreamError { + error: Box::new(expected), + }], + "{format:?}: {event}" + ); } Ok(()) } +#[test] +fn structured_stream_errors_encode_across_provider_formats() -> TestResult { + let engine = TranslationEngine::default(); + let input = json!({"type":"error","status":400,"error":{"type":"invalid_request_error","code":"previous_response_not_found","param":"previous_response_id","message":"missing"}}); + let mut chat = StreamTranslationState::new(WireFormat::OpenAiResponses, WireFormat::OpenAiChat); + assert_eq!( + engine.translate_event( + &mut chat, + WireFormat::OpenAiResponses, + WireFormat::OpenAiChat, + &input + )?, + vec![ + json!({"status":400,"error":{"type":"invalid_request_error","code":"previous_response_not_found","param":"previous_response_id","message":"missing"}}) + ] + ); + let mut anthropic = + StreamTranslationState::new(WireFormat::OpenAiResponses, WireFormat::AnthropicMessages); + assert_eq!( + engine.translate_event( + &mut anthropic, + WireFormat::OpenAiResponses, + WireFormat::AnthropicMessages, + &input + )?, + vec![ + json!({"type":"error","error":{"type":"invalid_request_error","code":"previous_response_not_found","message":"missing"}}) + ] + ); + let input = json!({"status":503,"error":{"type":"server_error","code":null,"param":null,"message":"unavailable"}}); + let mut responses = + StreamTranslationState::new(WireFormat::OpenAiChat, WireFormat::OpenAiResponses); + assert_eq!( + engine.translate_event( + &mut responses, + WireFormat::OpenAiChat, + WireFormat::OpenAiResponses, + &input + )?, + vec![ + json!({"type":"error","status":503,"error":{"type":"server_error","code":null,"param":null,"message":"unavailable"},"sequence_number":0}) + ] + ); + let input = json!({"type":"error","code":500,"param":null,"message":"flat failure"}); + let mut anthropic_fallback = + StreamTranslationState::new(WireFormat::OpenAiResponses, WireFormat::AnthropicMessages); + assert_eq!( + engine.translate_event( + &mut anthropic_fallback, + WireFormat::OpenAiResponses, + WireFormat::AnthropicMessages, + &input + )?, + vec![json!({"type":"error","error":{"type":"api_error","message":"flat failure"}})] + ); + Ok(()) +} + // Verifies the streaming encoder matches the buffered one: both Responses usage detail objects // are present even when the upstream reports no cache or reasoning breakdown. #[test]