Enable optional request compression

codex-rs/Cargo.toml

@@ -235,6 +235,7 @@ wildmatch = "2.6.1"
 
 wiremock = "0.6"
 zeroize = "1.8.2"
+zstd = "0.13"
 
 [workspace.lints]
 rust = {}

codex-rs/codex-api/Cargo.toml

@@ -19,6 +19,7 @@ tracing = { workspace = true }
 eventsource-stream = { workspace = true }
 regex-lite = { workspace = true }
 tokio-util = { workspace = true, features = ["codec"] }
+zstd = { workspace = true }
 
 [dev-dependencies]
 anyhow = { workspace = true }

codex-rs/codex-api/src/endpoint/chat.rs

@@ -6,7 +6,10 @@ use crate::common::ResponseStream;
 use crate::endpoint::streaming::StreamingClient;
 use crate::error::ApiError;
 use crate::provider::Provider;
+use crate::provider::RequestCompression;
 use crate::provider::WireApi;
+use crate::requests::body::encode_body;
+use crate::requests::body::insert_compression_headers;
 use crate::sse::chat::spawn_chat_stream;
 use crate::telemetry::SseTelemetry;
 use codex_client::HttpTransport;
@@ -45,8 +48,13 @@ impl<T: HttpTransport, A: AuthProvider> ChatClient<T, A> {
         }
     }
 
-    pub async fn stream_request(&self, request: ChatRequest) -> Result<ResponseStream, ApiError> {
-        self.stream(request.body, request.headers).await
+    pub async fn stream_request(
+        &self,
+        request: ChatRequest,
+        request_compression: RequestCompression,
+    ) -> Result<ResponseStream, ApiError> {
+        self.stream(request.body, request.headers, request_compression)
+            .await
     }
 
     pub async fn stream_prompt(
@@ -55,6 +63,7 @@ impl<T: HttpTransport, A: AuthProvider> ChatClient<T, A> {
         prompt: &ApiPrompt,
         conversation_id: Option<String>,
         session_source: Option<SessionSource>,
+        request_compression: RequestCompression,
     ) -> Result<ResponseStream, ApiError> {
         use crate::requests::ChatRequestBuilder;
 
@@ -64,7 +73,7 @@ impl<T: HttpTransport, A: AuthProvider> ChatClient<T, A> {
                 .session_source(session_source)
                 .build(self.streaming.provider())?;
 
-        self.stream_request(request).await
+        self.stream_request(request, request_compression).await
     }
 
     fn path(&self) -> &'static str {
@@ -78,9 +87,13 @@ impl<T: HttpTransport, A: AuthProvider> ChatClient<T, A> {
         &self,
         body: Value,
         extra_headers: HeaderMap,
+        request_compression: RequestCompression,
     ) -> Result<ResponseStream, ApiError> {
+        let mut headers = extra_headers;
+        insert_compression_headers(&mut headers, request_compression);
+        let encoded_body = encode_body(&body, request_compression)?;
         self.streaming
-            .stream(self.path(), body, extra_headers, spawn_chat_stream)
+            .stream(self.path(), encoded_body, headers, spawn_chat_stream)
             .await
     }
 }

codex-rs/codex-api/src/endpoint/compact.rs

@@ -5,6 +5,7 @@ use crate::error::ApiError;
 use crate::provider::Provider;
 use crate::provider::WireApi;
 use crate::telemetry::run_with_request_telemetry;
+use codex_client::Body;
 use codex_client::HttpTransport;
 use codex_client::RequestTelemetry;
 use codex_protocol::models::ResponseItem;
@@ -54,7 +55,7 @@ impl<T: HttpTransport, A: AuthProvider> CompactClient<T, A> {
         let builder = || {
             let mut req = self.provider.build_request(Method::POST, path);
             req.headers.extend(extra_headers.clone());
-            req.body = Some(body.clone());
+            req.body = Some(Body::Json(body.clone()));
             add_auth_headers(&self.auth, req)
         };
 
@@ -89,6 +90,7 @@ struct CompactHistoryResponse {
 #[cfg(test)]
 mod tests {
     use super::*;
+
     use crate::provider::RetryConfig;
     use async_trait::async_trait;
     use codex_client::Request;

codex-rs/codex-api/src/endpoint/responses.rs

@@ -6,6 +6,7 @@ use crate::common::TextControls;
 use crate::endpoint::streaming::StreamingClient;
 use crate::error::ApiError;
 use crate::provider::Provider;
+use crate::provider::RequestCompression;
 use crate::provider::WireApi;
 use crate::requests::ResponsesRequest;
 use crate::requests::ResponsesRequestBuilder;
@@ -15,7 +16,6 @@ use codex_client::HttpTransport;
 use codex_client::RequestTelemetry;
 use codex_protocol::protocol::SessionSource;
 use http::HeaderMap;
-use serde_json::Value;
 use std::sync::Arc;
 use tracing::instrument;
 
@@ -33,6 +33,7 @@ pub struct ResponsesOptions {
     pub conversation_id: Option<String>,
     pub session_source: Option<SessionSource>,
     pub extra_headers: HeaderMap,
+    pub request_compression: RequestCompression,
 }
 
 impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
@@ -56,7 +57,7 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
         &self,
         request: ResponsesRequest,
     ) -> Result<ResponseStream, ApiError> {
-        self.stream(request.body, request.headers).await
+        self.stream(request).await
     }
 
     #[instrument(level = "trace", skip_all, err)]
@@ -75,6 +76,7 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
             conversation_id,
             session_source,
             extra_headers,
+            request_compression,
         } = options;
 
         let request = ResponsesRequestBuilder::new(model, &prompt.instructions, &prompt.input)
@@ -88,6 +90,7 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
             .session_source(session_source)
             .store_override(store_override)
             .extra_headers(extra_headers)
+            .request_compression(request_compression)
             .build(self.streaming.provider())?;
 
         self.stream_request(request).await
@@ -100,13 +103,14 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
         }
     }
 
-    pub async fn stream(
-        &self,
-        body: Value,
-        extra_headers: HeaderMap,
-    ) -> Result<ResponseStream, ApiError> {
+    pub async fn stream(&self, request: ResponsesRequest) -> Result<ResponseStream, ApiError> {
         self.streaming
-            .stream(self.path(), body, extra_headers, spawn_response_stream)
+            .stream(
+                self.path(),
+                request.body,
+                request.headers,
+                spawn_response_stream,
+            )
             .await
     }
 }

codex-rs/codex-api/src/endpoint/streaming.rs

@@ -5,12 +5,15 @@ use crate::error::ApiError;
 use crate::provider::Provider;
 use crate::telemetry::SseTelemetry;
 use crate::telemetry::run_with_request_telemetry;
+use codex_client::Body;
 use codex_client::HttpTransport;
 use codex_client::RequestTelemetry;
 use codex_client::StreamResponse;
 use http::HeaderMap;
+use http::HeaderValue;
 use http::Method;
-use serde_json::Value;
+use http::header::ACCEPT;
+use http::header::CONTENT_TYPE;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -50,17 +53,18 @@ impl<T: HttpTransport, A: AuthProvider> StreamingClient<T, A> {
     pub(crate) async fn stream(
         &self,
         path: &str,
-        body: Value,
+        body: Body,
         extra_headers: HeaderMap,
         spawner: fn(StreamResponse, Duration, Option<Arc<dyn SseTelemetry>>) -> ResponseStream,
     ) -> Result<ResponseStream, ApiError> {
         let builder = || {
             let mut req = self.provider.build_request(Method::POST, path);
             req.headers.extend(extra_headers.clone());
-            req.headers.insert(
-                http::header::ACCEPT,
-                http::HeaderValue::from_static("text/event-stream"),
-            );
+            req.headers
+                .insert(ACCEPT, HeaderValue::from_static("text/event-stream"));
+            req.headers
+                .entry(CONTENT_TYPE)
+                .or_insert_with(|| HeaderValue::from_static("application/json"));
             req.body = Some(body.clone());
             add_auth_headers(&self.auth, req)
         };

codex-rs/codex-api/src/provider.rs

@@ -41,6 +41,13 @@ impl RetryConfig {
     }
 }
 
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum RequestCompression {
+    #[default]
+    None,
+    Zstd,
+}
+
 /// HTTP endpoint configuration used to talk to a concrete API deployment.
 ///
 /// Encapsulates base URL, default headers, query params, retry policy, and

codex-rs/codex-api/src/requests/body.rs

@@ -0,0 +1,40 @@
+use crate::error::ApiError;
+use crate::provider::RequestCompression;
+use bytes::Bytes;
+use codex_client::Body;
+use http::HeaderMap;
+use http::HeaderValue;
+use http::header::CONTENT_ENCODING;
+use serde_json::Value;
+use std::time::Instant;
+use tracing::info;
+use zstd::stream::encode_all;
+
+pub(crate) fn encode_body(body: &Value, compression: RequestCompression) -> Result<Body, ApiError> {
+    match compression {
+        RequestCompression::None => Ok(Body::Json(body.clone())),
+        RequestCompression::Zstd => {
+            let json = serde_json::to_vec(body).map_err(|err| {
+                ApiError::Stream(format!("failed to encode request body as json: {err}"))
+            })?;
+            let started_at = Instant::now();
+            let compressed = encode_all(json.as_slice(), 0).map_err(|err| {
+                ApiError::Stream(format!("failed to compress request body: {err}"))
+            })?;
+            let elapsed = started_at.elapsed();
+            info!(
+                input_bytes = json.len(),
+                output_bytes = compressed.len(),
+                elapsed_ms = elapsed.as_millis(),
+                "compressed request body"
+            );
+            Ok(Body::Bytes(Bytes::from(compressed)))
+        }
+    }
+}
+
+pub(crate) fn insert_compression_headers(headers: &mut HeaderMap, compression: RequestCompression) {
+    if matches!(compression, RequestCompression::Zstd) {
+        headers.insert(CONTENT_ENCODING, HeaderValue::from_static("zstd"));
+    }
+}

codex-rs/codex-api/src/requests/chat.rs

@@ -351,6 +351,7 @@ fn push_tool_call_message(messages: &mut Vec<Value>, tool_call: Value, reasoning
 #[cfg(test)]
 mod tests {
     use super::*;
+
     use crate::provider::RetryConfig;
     use crate::provider::WireApi;
     use codex_protocol::models::FunctionCallOutputPayload;

codex-rs/codex-api/src/requests/mod.rs

@@ -1,3 +1,4 @@
+pub(crate) mod body;
 pub mod chat;
 pub(crate) mod headers;
 pub mod responses;