feat(protocol): add structured stream errors

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,
         })

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;
             }
             _ => {}

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.

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<Response, LlmClientError> {
+        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<Response, LlmClientError> {
@@ -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;

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",
+                )),
             }]),
         ]))
     };

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 => {

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");

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?;