fix(server): preserve safe headers on terminal upstream errors

crates/libsy-llm-client/src/client.rs

@@ -501,6 +501,7 @@ impl TranslatingLlmClient {
         }
 
         let retry_after = retry_after_delay(response.headers());
+        let headers = Box::new(safe_upstream_error_headers(response.headers()));
         let body = match response.text().await {
             Ok(body) => body,
             Err(error) => {
@@ -521,7 +522,11 @@ impl TranslatingLlmClient {
                     message: body,
                 }
             } else {
-                LlmClientError::UpstreamHttp { status, body }
+                LlmClientError::UpstreamHttp {
+                    status,
+                    body,
+                    headers,
+                }
             };
         Err(AttemptFailure {
             error,
@@ -610,6 +615,7 @@ impl TranslatingLlmClient {
                                         metadata.as_ref(),
                                         backend.is_forwarding_auth(),
                                     ),
+                                    headers: Box::default(),
                                 }
                             }
                             error => LlmClientError::ResponseTranslation(error.to_string()),
@@ -852,6 +858,21 @@ fn first_event_overflow(
         })
 }
 
+/// Retain end-to-end retry and correlation headers from a failed upstream response.
+fn safe_upstream_error_headers(headers: &HeaderMap) -> HeaderMap {
+    let mut safe = HeaderMap::new();
+    for (name, value) in headers {
+        let name_str = name.as_str();
+        let allowed = matches!(name_str, "retry-after" | "request-id" | "x-request-id")
+            || name_str.starts_with("x-ratelimit-")
+            || name_str.starts_with("anthropic-ratelimit-");
+        if allowed && !is_non_forwardable_header(name_str, headers) {
+            safe.append(name.clone(), value.clone());
+        }
+    }
+    safe
+}
+
 // Uses Retry-After when supplied, capped so an upstream cannot stall a request indefinitely.
 fn retry_after_delay(headers: &HeaderMap) -> Option<Duration> {
     let value = headers.get(RETRY_AFTER)?.to_str().ok()?;
@@ -2433,7 +2454,8 @@ mod tests {
             error,
             LlmClientError::UpstreamHttp {
                 status: StatusCode::UNAUTHORIZED,
-                body
+                body,
+                ..
             } if body == "invalid key"
         ));
         assert_eq!(calls.load(Ordering::SeqCst), 1);
@@ -2496,6 +2518,8 @@ mod tests {
                 let attempt = observed_calls.fetch_add(1, Ordering::SeqCst) + 1;
                 ResponseTemplate::new(500)
                     .insert_header("retry-after", "0")
+                    .insert_header("x-request-id", format!("attempt-{attempt}").as_str())
+                    .insert_header("set-cookie", "session=secret")
                     .set_body_string(format!("attempt {attempt}"))
             })
             .mount(&server)
@@ -2510,17 +2534,51 @@ mod tests {
             panic!("expected retry exhaustion");
         };
 
-        assert!(matches!(
-            error,
-            LlmClientError::UpstreamHttp {
-                status: StatusCode::INTERNAL_SERVER_ERROR,
-                body
-            } if body == "attempt 3"
-        ));
+        let LlmClientError::UpstreamHttp {
+            status,
+            body,
+            headers,
+        } = error
+        else {
+            panic!("expected final upstream error");
+        };
+        assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
+        assert_eq!(body, "attempt 3");
+        assert_eq!(headers.get("x-request-id"), Some(&"attempt-3".parse()?));
+        assert!(!headers.contains_key("set-cookie"));
         assert_eq!(calls.load(Ordering::SeqCst), 3);
         Ok(())
     }
 
+    /// Connection-nominated headers must not survive the upstream error allowlist.
+    #[test]
+    fn upstream_error_headers_exclude_connection_nominations() {
+        let mut headers = HeaderMap::new();
+        headers.append(
+            "connection",
+            "x-request-id, x-ratelimit-reset-requests"
+                .parse()
+                .expect("header value"),
+        );
+        headers.append("connection", "retry-after".parse().expect("header value"));
+        headers.insert("x-request-id", "hop-id".parse().expect("header value"));
+        headers.insert(
+            "x-ratelimit-reset-requests",
+            "15s".parse().expect("header value"),
+        );
+        headers.insert("retry-after", "15".parse().expect("header value"));
+        headers.insert("request-id", "end-to-end-id".parse().expect("header value"));
+
+        let safe = safe_upstream_error_headers(&headers);
+        assert!(!safe.contains_key("x-request-id"));
+        assert!(!safe.contains_key("x-ratelimit-reset-requests"));
+        assert!(!safe.contains_key("retry-after"));
+        assert_eq!(
+            safe.get("request-id"),
+            Some(&"end-to-end-id".parse().expect("header value"))
+        );
+    }
+
     #[tokio::test]
     async fn deadline_expires_before_send_or_stream_poll()
     -> std::result::Result<(), Box<dyn Error + Sync + Send + 'static>> {
@@ -2624,6 +2682,7 @@ mod tests {
                 error: LlmClientError::UpstreamHttp {
                     status,
                     body: String::new(),
+                    headers: Box::default(),
                 },
                 status: Some(status),
                 retry_after: None,
@@ -2641,6 +2700,7 @@ mod tests {
                 error: LlmClientError::UpstreamHttp {
                     status,
                     body: String::new(),
+                    headers: Box::default(),
                 },
                 status: Some(status),
                 retry_after: None,
@@ -2768,6 +2828,7 @@ mod tests {
         let LlmClientError::UpstreamHttp {
             status,
             body: actual,
+            ..
         } = error
         else {
             panic!("expected the upstream HTTP error, got {error:?}");

crates/libsy-llm-client/src/run.rs

@@ -355,7 +355,7 @@ async fn call_one(
     result
 }
 
-// Preserve signed provider events while checking the complete routing response.
+/// Preserve signed provider events while checking the complete routing response.
 async fn buffer_routing_stream(
     mut chunks: LlmResponseStream,
 ) -> std::result::Result<LlmResponseStream, LlmClientError> {
@@ -370,6 +370,7 @@ async fn buffer_routing_stream(
                     return Err(LlmClientError::UpstreamHttp {
                         status: StatusCode::BAD_GATEWAY,
                         body: message.clone(),
+                        headers: Box::default(),
                     });
                 }
                 _ => {}
@@ -392,7 +393,7 @@ fn fallback_reason(error: &LibsyError) -> Option<RoutingFallbackReason> {
         }
         // A policy denial can be specific to one provider. Preserve its HTTP
         // error, but allow another candidate to serve the request.
-        LlmClientError::UpstreamHttp { status, body }
+        LlmClientError::UpstreamHttp { status, body, .. }
             if *status == StatusCode::BAD_REQUEST
                 && serde_json::from_str::<serde_json::Value>(body).is_ok_and(|value| {
                     value["error"]["code"].as_str() == Some("content_policy_violation")
@@ -1016,10 +1017,12 @@ mod tests {
                     FirstOutcome::ContentPolicy => Err(LlmClientError::UpstreamHttp {
                         status: StatusCode::BAD_REQUEST,
                         body: r#"{"error":{"code":"content_policy_violation","message":"request blocked by content policy","type":"invalid_request_error"},"metadata":{"documentation_section":"context window"}}"#.to_string(),
+                        headers: Box::default(),
                     }),
                     FirstOutcome::Unauthorized => Err(LlmClientError::UpstreamHttp {
                         status: StatusCode::UNAUTHORIZED,
                         body: "unauthorized".to_string(),
+                        headers: Box::default(),
                     }),
                     FirstOutcome::StreamSuccess => Ok(stream_response(vec![
                         LlmResponseChunk::TextDelta {
@@ -1924,6 +1927,7 @@ mod tests {
                 fallback_reason(&error(LlmClientError::UpstreamHttp {
                     status,
                     body: "failed".to_string(),
+                    headers: Box::default(),
                 })),
                 Some(RoutingFallbackReason::Unavailable)
             );
@@ -1944,6 +1948,7 @@ mod tests {
                 fallback_reason(&error(LlmClientError::UpstreamHttp {
                     status: StatusCode::BAD_REQUEST,
                     body: body.to_string(),
+                    headers: Box::default(),
                 })),
                 expected,
                 "{body}"
@@ -1961,6 +1966,7 @@ mod tests {
                 fallback_reason(&error(LlmClientError::UpstreamHttp {
                     status,
                     body: "failed".to_string(),
+                    headers: Box::default(),
                 })),
                 None
             );

crates/libsy-llm-client/tests/observability.rs

@@ -449,6 +449,7 @@ impl RoutedLlmClient for JudgeClient {
             JudgeOutcome::CallFailure => Err(LlmClientError::UpstreamHttp {
                 status: http::StatusCode::INTERNAL_SERVER_ERROR,
                 body: format!(r#"{{"error":{{"message":"server error: {LEAKED_CONTENT}"}}}}"#),
+                headers: Box::default(),
             }),
             JudgeOutcome::Reply(text) => Ok(Response {
                 llm_response: LlmResponse::Agg(text_response(None, *text)),
@@ -1575,7 +1576,7 @@ async fn upstream_body_is_redacted_from_the_client_call_span() -> switchyard_lib
     assert!(matches!(
         result,
         Err(LibsyError::ClientCall {
-            source: LlmClientError::UpstreamHttp { status, body },
+            source: LlmClientError::UpstreamHttp { status, body, .. },
             ..
         }) if status == http::StatusCode::INTERNAL_SERVER_ERROR && body.contains(LEAKED_CONTENT)
     ));

crates/libsy/src/algorithms/advisor_gate/tests.rs

@@ -718,6 +718,7 @@ async fn fail_open_logs_redact_the_upstream_error_body() {
                 Err(LlmClientError::UpstreamHttp {
                     status: http::StatusCode::INTERNAL_SERVER_ERROR,
                     body: format!("{MARKER}: validation failed"),
+                    headers: Box::default(),
                 })
             } else {
                 Ok(reply("done"))

crates/libsy/src/algorithms/util/buffered_response.rs

@@ -70,6 +70,7 @@ pub(crate) async fn buffer_response(
                             Some(LlmClientError::UpstreamHttp {
                                 status: http::StatusCode::BAD_GATEWAY,
                                 body: message.clone(),
+                                headers: Box::default(),
                             })
                         }
                         chunk => {

crates/libsy/src/algorithms/util/llm_judge.rs

@@ -684,13 +684,15 @@ mod tests {
                 LlmClientError::UpstreamHttp {
                     status: StatusCode::INTERNAL_SERVER_ERROR,
                     body: "server error".to_string(),
+                    headers: Box::default(),
                 },
                 "upstream_5xx",
             ),
             (
                 LlmClientError::UpstreamHttp {
                     status: StatusCode::FOUND,
                     body: "redirect".to_string(),
+                    headers: Box::default(),
                 },
                 "upstream_non_5xx",
             ),

crates/libsy/src/algorithms/util/robustness.rs

@@ -90,6 +90,7 @@ mod tests {
             source: LlmClientError::UpstreamHttp {
                 status: http::StatusCode::BAD_REQUEST,
                 body: format!(r#"{{"error":{{"message":"bad request: {SECRET}"}}}}"#),
+                headers: Box::default(),
             },
         };
 

crates/protocol/src/client.rs

@@ -99,6 +99,8 @@ pub enum LlmClientError {
         status: http::StatusCode,
         /// Raw upstream error body.
         body: String,
+        /// Safe retry and correlation headers from the final failed attempt.
+        headers: Box<http::HeaderMap>,
     },
 
     /// The upstream returned a response the client could not decode.

crates/protocol/src/stream.rs

@@ -275,6 +275,7 @@ impl AggLlmResponse {
     }
 }
 
+/// Accumulate a chunk unless it reports a provider or decoding failure.
 fn push_checked_chunk(
     accumulator: &mut ResponseAccumulator,
     chunk: LlmResponseChunk,
@@ -286,6 +287,7 @@ fn push_checked_chunk(
         LlmResponseChunk::StreamError { message } => Err(LlmClientError::UpstreamHttp {
             status: MID_STREAM_UPSTREAM_STATUS,
             body: message,
+            headers: Box::default(),
         }),
         chunk => {
             accumulator.push(chunk);

crates/switchyard-nemo-relay-plugin/src/runtime.rs

@@ -605,6 +605,7 @@ fn normalized_stream_error(chunks: &[LlmResponseChunk]) -> Option<LlmClientError
         LlmResponseChunk::StreamError { message } => Some(LlmClientError::UpstreamHttp {
             status: StatusCode::BAD_GATEWAY,
             body: message.clone(),
+            headers: Box::default(),
         }),
         _ => None,
     })

crates/switchyard-runner/src/failure.rs

@@ -188,6 +188,7 @@ mod tests {
             source: LlmClientError::UpstreamHttp {
                 status: reqwest::StatusCode::SERVICE_UNAVAILABLE,
                 body: format!("upstream response: {SECRET}"),
+                headers: Box::default(),
             },
         });