fix(gateway): decode compressed request bodies

crates/cli/Cargo.toml

@@ -66,6 +66,7 @@ tokio-tungstenite = { version = "0.27", default-features = false, features = ["c
 toml = "0.9"
 toml_edit = "0.23"
 uuid = { workspace = true, features = ["serde", "v7"] }
+zstd = "0.13"
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"

crates/cli/src/gateway/mod.rs

@@ -17,7 +17,9 @@ use std::sync::{Arc, Mutex};
 use async_stream::stream;
 use axum::body::{Body, Bytes};
 use axum::extract::State;
-use axum::http::{HeaderMap, HeaderName, HeaderValue, Method, Request, Response, StatusCode};
+use axum::http::{
+    HeaderMap, HeaderName, HeaderValue, Method, Request, Response, StatusCode, header,
+};
 use futures_util::StreamExt;
 use nemo_relay::api::llm::{
     LlmCallExecuteParams, LlmRequest, LlmStreamCallExecuteParams, llm_call_execute,
@@ -769,11 +771,15 @@ fn effective_dispatch_request(
         };
     };
 
+    let mut body_reencoded = false;
     let body_bytes = if request.content.is_null() {
         body_bytes.clone()
     } else {
         match serde_json::to_vec(&request.content) {
-            Ok(serialized) => Bytes::from(serialized),
+            Ok(serialized) => {
+                body_reencoded = true;
+                Bytes::from(serialized)
+            }
             Err(error) => {
                 eprintln!(
                     "nemo-relay CLI gateway: failed to serialize rewritten LLM request body; forwarding original request: {error}"
@@ -812,6 +818,9 @@ fn effective_dispatch_request(
         };
         headers.insert(name, value);
     }
+    if body_reencoded {
+        headers.remove(header::CONTENT_ENCODING);
+    }
     EffectiveUpstreamRequest {
         body_bytes,
         headers,

crates/cli/src/gateway/request.rs

@@ -3,10 +3,12 @@
 
 //! Gateway request validation, buffering, and normalized LLM start construction.
 
+use std::borrow::Cow;
 use std::error::Error;
+use std::io::Read;
 
 use axum::body::{Body, Bytes};
-use axum::http::{HeaderMap, Method, Request};
+use axum::http::{HeaderMap, Method, Request, header};
 use http_body_util::LengthLimitError;
 use nemo_relay::api::llm::LlmRequest;
 use serde_json::{Value, json};
@@ -46,7 +48,13 @@ pub(super) async fn prepare_gateway_request(
     let body_bytes = axum::body::to_bytes(body, config.max_passthrough_body_bytes)
         .await
         .map_err(passthrough_body_error)?;
-    let request_json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or(Value::Null);
+    let request_json = request_body_for_observability(
+        &body_bytes,
+        &parts.headers,
+        config.max_passthrough_body_bytes,
+    )
+    .and_then(|body| serde_json::from_slice::<Value>(&body).ok())
+    .unwrap_or(Value::Null);
     let path_and_query = parts
         .uri
         .path_and_query()
@@ -76,6 +84,61 @@ pub(super) async fn prepare_gateway_request(
     })
 }
 
+// Decodes the transport body only for Relay's managed request representation. The original bytes
+// and Content-Encoding header remain on PreparedGatewayRequest so unsupported or malformed
+// encodings still pass through unchanged. When the managed pipeline reserializes decoded JSON,
+// effective_dispatch_request removes Content-Encoding from the identity-encoded upstream body.
+fn request_body_for_observability<'a>(
+    body: &'a [u8],
+    headers: &HeaderMap,
+    max_decoded_bytes: usize,
+) -> Option<Cow<'a, [u8]>> {
+    let mut encodings = Vec::new();
+    for value in headers.get_all(header::CONTENT_ENCODING) {
+        for encoding in value.to_str().ok()?.split(',') {
+            let encoding = encoding.trim();
+            if encoding.is_empty() {
+                return None;
+            }
+            encodings.push(encoding.to_ascii_lowercase());
+        }
+    }
+    if encodings.is_empty() {
+        return Some(Cow::Borrowed(body));
+    }
+
+    let mut decoded = Cow::Borrowed(body);
+    for encoding in encodings.iter().rev() {
+        match encoding.as_str() {
+            "identity" => {}
+            "zstd" => decoded = Cow::Owned(decode_zstd(&decoded, max_decoded_bytes)?),
+            _ => return None,
+        }
+    }
+    (decoded.len() <= max_decoded_bytes).then_some(decoded)
+}
+
+fn decode_zstd(body: &[u8], max_decoded_bytes: usize) -> Option<Vec<u8>> {
+    let mut decoder = zstd::stream::read::Decoder::new(body).ok()?;
+    decoder
+        .window_log_max(zstd_window_log_max(max_decoded_bytes))
+        .ok()?;
+    let limit = u64::try_from(max_decoded_bytes)
+        .unwrap_or(u64::MAX)
+        .saturating_add(1);
+    let mut decoded = Vec::new();
+    decoder.take(limit).read_to_end(&mut decoded).ok()?;
+    (decoded.len() <= max_decoded_bytes).then_some(decoded)
+}
+
+pub(super) fn zstd_window_log_max(max_decoded_bytes: usize) -> u32 {
+    const ZSTD_WINDOW_LOG_MIN: u32 = 10;
+    const ZSTD_WINDOW_LOG_MAX: u32 = if usize::BITS == 32 { 30 } else { 31 };
+
+    let required_log = usize::BITS - max_decoded_bytes.saturating_sub(1).leading_zeros();
+    required_log.clamp(ZSTD_WINDOW_LOG_MIN, ZSTD_WINDOW_LOG_MAX)
+}
+
 fn passthrough_body_error(error: axum::Error) -> CliError {
     if error.source().is_some_and(|source| {
         source.is::<LengthLimitError>()

crates/cli/tests/coverage/shared/gateway_tests.rs

@@ -103,6 +103,145 @@ async fn prepared_gateway_request_consumes_private_client_proof() {
     );
 }
 
+#[tokio::test]
+async fn prepared_gateway_request_decodes_zstd_for_observability() {
+    let body = br#"{"model":"gpt-test","stream":true}"#;
+    let compressed = zstd::stream::encode_all(body.as_slice(), 0).unwrap();
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/v1/responses")
+        .header(header::CONTENT_ENCODING, "zstd")
+        .body(Body::from(compressed.clone()))
+        .unwrap();
+
+    let prepared = prepare_gateway_request(&GatewayConfig::default(), request, true)
+        .await
+        .unwrap();
+
+    assert_eq!(prepared.body_bytes.as_ref(), compressed);
+    assert_eq!(
+        prepared.request_json,
+        json!({
+            "model": "gpt-test",
+            "stream": true,
+        })
+    );
+    assert!(prepared.streaming);
+    assert_eq!(
+        prepared.headers.get(header::CONTENT_ENCODING).unwrap(),
+        "zstd"
+    );
+}
+
+#[tokio::test]
+async fn prepared_gateway_request_decodes_chained_zstd_for_observability() {
+    let body = br#"{"model":"gpt-test","stream":true}"#;
+    let compressed_once = zstd::stream::encode_all(body.as_slice(), 0).unwrap();
+    let compressed_twice = zstd::stream::encode_all(compressed_once.as_slice(), 0).unwrap();
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/v1/responses")
+        .header(header::CONTENT_ENCODING, "zstd, zstd")
+        .body(Body::from(compressed_twice.clone()))
+        .unwrap();
+
+    let prepared = prepare_gateway_request(&GatewayConfig::default(), request, true)
+        .await
+        .unwrap();
+
+    assert_eq!(prepared.body_bytes.as_ref(), compressed_twice);
+    assert_eq!(
+        prepared.request_json,
+        json!({
+            "model": "gpt-test",
+            "stream": true,
+        })
+    );
+    assert_eq!(
+        prepared.headers.get(header::CONTENT_ENCODING).unwrap(),
+        "zstd, zstd"
+    );
+}
+
+#[test]
+fn zstd_decoder_window_tracks_the_managed_body_limit() {
+    assert_eq!(zstd_window_log_max(0), 10);
+    assert_eq!(zstd_window_log_max(1 << 10), 10);
+    assert_eq!(zstd_window_log_max((1 << 10) + 1), 11);
+    assert_eq!(
+        zstd_window_log_max(usize::MAX),
+        if usize::BITS == 32 { 30 } else { 31 }
+    );
+}
+
+#[tokio::test]
+async fn request_observability_decode_is_bounded_and_encoding_aware() {
+    let oversized = vec![b'x'; 256];
+    let compressed = zstd::stream::encode_all(oversized.as_slice(), 0).unwrap();
+    let config = GatewayConfig {
+        max_passthrough_body_bytes: 32,
+        ..GatewayConfig::default()
+    };
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/v1/responses")
+        .header(header::CONTENT_ENCODING, "zstd")
+        .body(Body::from(compressed))
+        .unwrap();
+    let prepared = prepare_gateway_request(&config, request, true)
+        .await
+        .unwrap();
+    assert!(prepared.request_json.is_null());
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/v1/responses")
+        .header(header::CONTENT_ENCODING, "gzip")
+        .body(Body::from(r#"{"model":"opaque"}"#))
+        .unwrap();
+    let prepared = prepare_gateway_request(&config, request, true)
+        .await
+        .unwrap();
+    assert!(prepared.request_json.is_null());
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/v1/responses")
+        .header(header::CONTENT_ENCODING, "identity")
+        .body(Body::from(r#"{"model":"gpt-test"}"#))
+        .unwrap();
+    let prepared = prepare_gateway_request(&config, request, true)
+        .await
+        .unwrap();
+    assert_eq!(
+        prepared.request_json,
+        json!({
+            "model": "gpt-test",
+        })
+    );
+}
+
+#[tokio::test]
+async fn malformed_encoded_request_remains_a_raw_passthrough() {
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/v1/responses")
+        .header(header::CONTENT_ENCODING, "zstd")
+        .body(Body::from("not-a-zstd-frame"))
+        .unwrap();
+    let prepared = prepare_gateway_request(&GatewayConfig::default(), request, true)
+        .await
+        .unwrap();
+    let managed = build_llm_gateway_start(&prepared).request;
+
+    let (body, headers) =
+        effective_upstream_request(&prepared.body_bytes, &prepared.headers, Some(&managed));
+
+    assert!(managed.content.is_null());
+    assert_eq!(body, prepared.body_bytes);
+    assert_eq!(headers.get(header::CONTENT_ENCODING).unwrap(), "zstd");
+}
+
 #[test]
 fn selects_provider_routes() {
     assert_eq!(
@@ -296,6 +435,26 @@ fn effective_upstream_request_overlays_runtime_body_and_headers() {
     );
 }
 
+#[test]
+fn effective_upstream_request_removes_content_encoding_after_reencoding() {
+    let original_body = Bytes::from_static(b"compressed bytes");
+    let mut original_headers = HeaderMap::new();
+    original_headers.insert(header::CONTENT_ENCODING, HeaderValue::from_static("zstd"));
+    let request = LlmRequest {
+        headers: Map::from_iter([("content-encoding".to_string(), json!("zstd"))]),
+        content: json!({ "model": "rewritten" }),
+    };
+
+    let (body, headers) =
+        effective_upstream_request(&original_body, &original_headers, Some(&request));
+
+    assert_eq!(
+        serde_json::from_slice::<Value>(&body).unwrap(),
+        json!({ "model": "rewritten" })
+    );
+    assert!(!headers.contains_key(header::CONTENT_ENCODING));
+}
+
 #[test]
 fn effective_upstream_request_returns_original_without_runtime_request() {
     let original_body = Bytes::from_static(br#"{"model":"original"}"#);