Record per-candidate upstream error bodies

apps/aether-gateway/src/execution_runtime/stream/execution.rs

@@ -83,11 +83,11 @@ use crate::execution_runtime::{
 use crate::execution_runtime::{MAX_STREAM_PREFETCH_BYTES, MAX_STREAM_PREFETCH_FRAMES};
 use crate::log_ids::short_request_id;
 use crate::orchestration::{
-    apply_local_execution_effect, build_local_error_flow_metadata, with_error_flow_report_context,
+    apply_local_execution_effect, build_local_error_flow_metadata, trace_upstream_response_body,
-    with_upstream_response_report_context, LocalAdaptiveRateLimitEffect,
+    with_error_flow_report_context, with_upstream_response_report_context,
-    LocalAdaptiveSuccessEffect, LocalAttemptFailureEffect, LocalExecutionEffect,
+    LocalAdaptiveRateLimitEffect, LocalAdaptiveSuccessEffect, LocalAttemptFailureEffect,
-    LocalExecutionEffectContext, LocalHealthFailureEffect, LocalHealthSuccessEffect,
+    LocalExecutionEffect, LocalExecutionEffectContext, LocalHealthFailureEffect,
-    LocalOAuthInvalidationEffect, LocalPoolErrorEffect,
+    LocalHealthSuccessEffect, LocalOAuthInvalidationEffect, LocalPoolErrorEffect,
 };
 use crate::request_candidate_runtime::{
     ensure_execution_request_candidate_slot, record_local_request_candidate_status,
@@ -254,14 +254,17 @@ fn with_stream_error_trace_context(
     report_context: Option<&Value>,
     status_code: u16,
     headers: &BTreeMap<String, String>,
+    body_json: Option<&Value>,
+    body_bytes: &[u8],
     response_text: Option<&str>,
     local_failover_analysis: crate::orchestration::LocalFailoverAnalysis,
 ) -> Option<Value> {
+    let body = trace_upstream_response_body(body_json, body_bytes);
     let upstream_context = with_upstream_response_report_context(
         report_context,
         status_code,
         Some(headers),
-        None,
+        body.as_ref(),
         None,
         None,
     );
@@ -1287,6 +1290,8 @@ async fn execute_stream_from_frame_stream(
                 report_context.as_ref(),
                 status_code,
                 &headers,
+                provider_body_json.as_ref(),
+                &provider_error_body,
                 error_response_text.as_deref(),
                 failover_analysis,
             );
@@ -1337,6 +1342,8 @@ async fn execute_stream_from_frame_stream(
                 report_context.as_ref(),
                 status_code,
                 &headers,
+                provider_body_json.as_ref(),
+                &provider_error_body,
                 error_response_text.as_deref(),
                 failover_analysis,
             );
@@ -1385,6 +1392,8 @@ async fn execute_stream_from_frame_stream(
             report_context.as_ref(),
             status_code,
             &headers,
+            provider_body_json.as_ref(),
+            &provider_error_body,
             error_response_text.as_deref(),
             failover_analysis,
         );

apps/aether-gateway/src/execution_runtime/stream/execution_failures.rs

@@ -20,9 +20,10 @@ use crate::execution_runtime::submission::{
 use crate::log_ids::short_request_id;
 use crate::orchestration::{
     apply_local_execution_effect, resolve_local_failover_analysis_for_attempt,
-    with_upstream_response_report_context, LocalAdaptiveRateLimitEffect, LocalAttemptFailureEffect,
+    trace_upstream_response_body, with_upstream_response_report_context,
-    LocalExecutionEffect, LocalExecutionEffectContext, LocalHealthFailureEffect,
+    LocalAdaptiveRateLimitEffect, LocalAttemptFailureEffect, LocalExecutionEffect,
-    LocalOAuthInvalidationEffect, LocalPoolErrorEffect,
+    LocalExecutionEffectContext, LocalHealthFailureEffect, LocalOAuthInvalidationEffect,
+    LocalPoolErrorEffect,
 };
 use crate::request_candidate_runtime::record_report_request_candidate_status;
 use crate::usage::submit_sync_report;
@@ -135,11 +136,12 @@ fn build_stream_failure_sync_payload(
     failure: StreamFailureReport,
 ) -> GatewaySyncReportRequest {
     let status_code = failure.status_code;
+    let body = trace_upstream_response_body(None, provider_buffered_body);
     let report_context = with_upstream_response_report_context(
         report_context.as_ref(),
         status_code,
         Some(&headers),
-        None,
+        body.as_ref(),
         None,
         None,
     )

apps/aether-gateway/src/execution_runtime/sync/execution.rs

@@ -53,11 +53,11 @@ use crate::execution_runtime::{
 };
 use crate::log_ids::short_request_id;
 use crate::orchestration::{
-    apply_local_execution_effect, build_local_error_flow_metadata, with_error_flow_report_context,
+    apply_local_execution_effect, build_local_error_flow_metadata, trace_upstream_response_body,
-    with_upstream_response_report_context, LocalAdaptiveRateLimitEffect,
+    with_error_flow_report_context, with_upstream_response_report_context,
-    LocalAdaptiveSuccessEffect, LocalAttemptFailureEffect, LocalExecutionEffect,
+    LocalAdaptiveRateLimitEffect, LocalAdaptiveSuccessEffect, LocalAttemptFailureEffect,
-    LocalExecutionEffectContext, LocalHealthFailureEffect, LocalHealthSuccessEffect,
+    LocalExecutionEffect, LocalExecutionEffectContext, LocalHealthFailureEffect,
-    LocalOAuthInvalidationEffect, LocalPoolErrorEffect,
+    LocalHealthSuccessEffect, LocalOAuthInvalidationEffect, LocalPoolErrorEffect,
 };
 use crate::request_candidate_runtime::{
     ensure_execution_request_candidate_slot, record_local_request_candidate_extra_data,
@@ -139,14 +139,17 @@ fn with_sync_error_trace_context(
     report_context: Option<&serde_json::Value>,
     status_code: u16,
     headers: &BTreeMap<String, String>,
+    body_json: Option<&serde_json::Value>,
+    body_bytes: &[u8],
     response_text: Option<&str>,
     local_failover_analysis: crate::orchestration::LocalFailoverAnalysis,
 ) -> Option<serde_json::Value> {
+    let body = trace_upstream_response_body(body_json, body_bytes);
     let upstream_context = with_upstream_response_report_context(
         report_context,
         status_code,
         Some(headers),
-        None,
+        body.as_ref(),
         None,
         None,
     );
@@ -1481,6 +1484,8 @@ async fn execute_execution_runtime_sync_impl(
             report_context.as_ref(),
             result.status_code,
             &headers,
+            body_json.as_ref(),
+            &body_bytes,
             local_failover_response_text.as_deref(),
             local_failover_analysis,
         );
@@ -1557,6 +1562,8 @@ async fn execute_execution_runtime_sync_impl(
             report_context.as_ref(),
             result.status_code,
             &headers,
+            body_json.as_ref(),
+            &body_bytes,
             local_failover_response_text.as_deref(),
             local_failover_analysis,
         );
@@ -1587,6 +1594,8 @@ async fn execute_execution_runtime_sync_impl(
                 report_context.as_ref(),
                 result.status_code,
                 &headers,
+                body_json.as_ref(),
+                &body_bytes,
                 local_failover_response_text.as_deref(),
                 local_failover_analysis,
             )

apps/aether-gateway/src/handlers/admin/observability/monitoring/tests/trace.rs

@@ -437,7 +437,7 @@ async fn admin_monitoring_trace_request_exposes_request_path_from_usage_audit()
 
 #[tokio::test]
 async fn admin_monitoring_trace_request_exposes_failed_candidate_upstream_response_boundary() {
-    let candidate = sample_candidate(
+    let mut candidate = sample_candidate(
         "cand-used",
         "request-1",
         0,
@@ -446,6 +446,17 @@ async fn admin_monitoring_trace_request_exposes_failed_candidate_upstream_respon
         Some(33),
         Some(302),
     );
+    candidate.extra_data = Some(json!({
+        "cache_1h": true,
+        "upstream_response": {
+            "status_code": 302,
+            "body": {
+                "error": {
+                    "message": "redirect blocked"
+                }
+            }
+        }
+    }));
 
     let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![candidate]));
     let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
@@ -511,7 +522,10 @@ async fn admin_monitoring_trace_request_exposes_failed_candidate_upstream_respon
         extra["upstream_response"]["headers"]["location"],
         json!("/")
     );
-    assert!(extra["upstream_response"]["body"].is_null());
+    assert_eq!(
+        extra["upstream_response"]["body"]["error"]["message"],
+        json!("redirect blocked")
+    );
     assert!(extra.get("client_response").is_none());
     assert!(extra.get("provider_response").is_none());
 }

apps/aether-gateway/src/orchestration/mod.rs

@@ -1,4 +1,5 @@
 use aether_contracts::ExecutionPlan;
+use base64::Engine as _;
 use serde_json::{json, Map, Value};
 
 use crate::AppState;
@@ -152,6 +153,65 @@ pub(crate) fn with_upstream_response_report_context(
     Some(Value::Object(object))
 }
 
+pub(crate) fn trace_upstream_response_body(
+    body_json: Option<&Value>,
+    body_bytes: &[u8],
+) -> Option<Value> {
+    if let Some(body_json) = body_json {
+        return Some(limit_trace_upstream_response_body_json(body_json));
+    }
+
+    if body_bytes.is_empty() {
+        return None;
+    }
+
+    if let Ok(text) = std::str::from_utf8(body_bytes) {
+        let text = text.trim();
+        if text.is_empty() {
+            return None;
+        }
+        if let Ok(json_body) = serde_json::from_str::<Value>(text) {
+            return Some(limit_trace_upstream_response_body_json(&json_body));
+        }
+        return Some(Value::String(limit_trace_upstream_response_text(text)));
+    }
+
+    Some(json!({
+        "encoding": "base64",
+        "data": base64::engine::general_purpose::STANDARD.encode(
+            &body_bytes[..body_bytes.len().min(crate::MAX_ERROR_BODY_BYTES)]
+        ),
+        "truncated": body_bytes.len() > crate::MAX_ERROR_BODY_BYTES,
+    }))
+}
+
+fn limit_trace_upstream_response_body_json(body_json: &Value) -> Value {
+    let Ok(serialized) = serde_json::to_vec(body_json) else {
+        return body_json.clone();
+    };
+    if serialized.len() <= crate::MAX_ERROR_BODY_BYTES {
+        return body_json.clone();
+    }
+    Value::String(limit_trace_upstream_response_text(
+        String::from_utf8_lossy(&serialized).as_ref(),
+    ))
+}
+
+fn limit_trace_upstream_response_text(text: &str) -> String {
+    let mut bytes = 0usize;
+    let mut out = String::new();
+    for ch in text.chars() {
+        let len = ch.len_utf8();
+        if bytes + len > crate::MAX_ERROR_BODY_BYTES {
+            out.push_str("...[truncated]");
+            return out;
+        }
+        bytes += len;
+        out.push(ch);
+    }
+    out
+}
+
 fn trace_headers_to_json(headers: &std::collections::BTreeMap<String, String>) -> Value {
     Value::Object(Map::from_iter(headers.iter().map(|(key, value)| {
         (

apps/aether-gateway/src/tests/ai_execute/stream/decision.rs

@@ -2339,6 +2339,20 @@ async fn gateway_retries_next_local_openai_chat_stream_candidate_after_retryable
         stored_candidates[0].error_message.as_deref(),
         Some("execution runtime stream returned retryable status 429")
     );
+    let failed_upstream_response = stored_candidates[0]
+        .extra_data
+        .as_ref()
+        .and_then(|value| value.get("upstream_response"))
+        .expect("failed stream candidate should keep its upstream response");
+    assert_eq!(failed_upstream_response["status_code"], json!(429));
+    assert_eq!(
+        failed_upstream_response["body"]["error"]["message"],
+        json!("rate limited")
+    );
+    assert_eq!(
+        failed_upstream_response["body"]["error"]["type"],
+        json!("rate_limit_error")
+    );
     assert_eq!(stored_candidates[1].candidate_index, 1);
     assert_eq!(stored_candidates[1].status, RequestCandidateStatus::Success);
     assert_eq!(stored_candidates[1].status_code, Some(200));

apps/aether-gateway/src/tests/ai_execute/sync/chat/failover.rs

@@ -1145,6 +1145,16 @@ async fn gateway_retries_next_local_openai_chat_sync_candidate_after_auth_failur
         stored_candidates[0].error_message.as_deref(),
         Some("invalid auth token")
     );
+    let failed_upstream_response = stored_candidates[0]
+        .extra_data
+        .as_ref()
+        .and_then(|value| value.get("upstream_response"))
+        .expect("failed candidate should keep its upstream response");
+    assert_eq!(failed_upstream_response["status_code"], json!(401));
+    assert_eq!(
+        failed_upstream_response["body"]["error"]["message"],
+        json!("invalid auth token")
+    );
     assert_eq!(stored_candidates[1].candidate_index, 1);
     assert_eq!(stored_candidates[1].status, RequestCandidateStatus::Success);
     assert_eq!(stored_candidates[1].status_code, Some(200));

crates/aether-admin/src/observability/monitoring.rs

@@ -501,7 +501,7 @@ fn build_admin_monitoring_trace_candidate_extra_data(
                 usage.response_body_ref.as_deref(),
                 usage.response_body_state,
             ) {
-                extra_object.insert("upstream_response".to_string(), response);
+                merge_admin_monitoring_trace_response(extra_object, "upstream_response", response);
             }
         }
 
@@ -558,6 +558,39 @@ fn admin_monitoring_trace_response_data(
     }))
 }
 
+fn merge_admin_monitoring_trace_response(
+    extra_object: &mut serde_json::Map<String, Value>,
+    key: &str,
+    response: Value,
+) {
+    let Some(response_object) = response.as_object() else {
+        extra_object.insert(key.to_string(), response);
+        return;
+    };
+    let Some(existing_object) = extra_object.get_mut(key).and_then(Value::as_object_mut) else {
+        extra_object.insert(key.to_string(), Value::Object(response_object.clone()));
+        return;
+    };
+
+    for (field, value) in response_object {
+        if admin_monitoring_trace_response_value_empty(value)
+            && existing_object
+                .get(field)
+                .is_some_and(|existing| !admin_monitoring_trace_response_value_empty(existing))
+        {
+            continue;
+        }
+        existing_object.insert(field.clone(), value.clone());
+    }
+}
+
+fn admin_monitoring_trace_response_value_empty(value: &Value) -> bool {
+    value.is_null()
+        || value.as_str().is_some_and(str::is_empty)
+        || value.as_array().is_some_and(Vec::is_empty)
+        || value.as_object().is_some_and(serde_json::Map::is_empty)
+}
+
 fn admin_monitoring_usage_is_error_node(usage: &StoredRequestUsageAudit) -> bool {
     !usage.status.eq_ignore_ascii_case("completed")
         || usage