From 55acaaffaca40350d22ceed7148d19220e1305f3 Mon Sep 17 00:00:00 2001 From: Channing Conger Date: Tue, 25 Nov 2025 11:53:38 -0800 Subject: [PATCH] Request compression. Add new model_provider flag for compression to enable request compression. We support zstd and gzip, server also supports brotli You can test this against the sign in with chatgpt flow by adding the following profile: ``` [profiles.compressed] name = "compressed" model_provider = "openai-zstd" [model_providers.openai-zstd] name = "OpenAI (ChatGPT, zstd)" wire_api = "responses" request_compression = "zstd" requires_openai_auth = true ``` This will zstd compress your request before sending it to the server. --- codex-rs/Cargo.lock | 45 ++++ codex-rs/Cargo.toml | 1 + codex-rs/codex-api/Cargo.toml | 1 + codex-rs/codex-api/src/endpoint/compact.rs | 5 +- codex-rs/codex-api/src/endpoint/models.rs | 2 + codex-rs/codex-api/src/endpoint/streaming.rs | 41 +++- codex-rs/codex-api/src/provider.rs | 7 + codex-rs/codex-api/src/requests/chat.rs | 2 + codex-rs/codex-api/src/requests/responses.rs | 2 + codex-rs/codex-api/tests/clients.rs | 2 + .../codex-api/tests/models_integration.rs | 2 + codex-rs/codex-api/tests/sse_end_to_end.rs | 2 + codex-rs/codex-client/src/default_client.rs | 7 + codex-rs/codex-client/src/lib.rs | 1 + codex-rs/codex-client/src/request.rs | 10 +- codex-rs/codex-client/src/transport.rs | 14 +- codex-rs/core/Cargo.toml | 1 + codex-rs/core/src/config/mod.rs | 1 + codex-rs/core/src/default_client.rs | 1 - codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/model_provider_info.rs | 25 +++ codex-rs/core/src/models_manager/manager.rs | 2 + .../core/tests/chat_completions_payload.rs | 2 + codex-rs/core/tests/chat_completions_sse.rs | 2 + codex-rs/core/tests/common/Cargo.toml | 3 + codex-rs/core/tests/common/lib.rs | 4 + codex-rs/core/tests/common/request.rs | 59 ++++++ codex-rs/core/tests/common/responses.rs | 10 +- codex-rs/core/tests/common/test_codex.rs | 1 + codex-rs/core/tests/responses_headers.rs | 194 ++++++++++++++++++ codex-rs/core/tests/suite/client.rs | 16 +- codex-rs/core/tests/suite/compact.rs | 64 +++--- .../core/tests/suite/compact_resume_fork.rs | 20 +- codex-rs/core/tests/suite/json_result.rs | 3 +- codex-rs/core/tests/suite/review.rs | 7 +- .../suite/stream_error_allows_next_turn.rs | 8 +- .../core/tests/suite/stream_no_completed.rs | 2 + 37 files changed, 491 insertions(+), 79 deletions(-) create mode 100644 codex-rs/core/tests/common/request.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 90acb1e772..fc8985497e 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -819,6 +819,8 @@ version = "1.2.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "deec109607ca693028562ed836a5f1c4b8bd77755c4e132fc5ce11b0b6211ae7" dependencies = [ + "jobserver", + "libc", "shlex", ] @@ -985,6 +987,7 @@ dependencies = [ "tokio-util", "tracing", "wiremock", + "zstd", ] [[package]] @@ -1348,6 +1351,7 @@ dependencies = [ "which", "wildmatch", "wiremock", + "zstd", ] [[package]] @@ -2109,16 +2113,19 @@ dependencies = [ "codex-protocol", "codex-utils-absolute-path", "codex-utils-cargo-bin", + "http 1.3.1", "notify", "pretty_assertions", "regex-lite", "reqwest", + "serde", "serde_json", "shlex", "tempfile", "tokio", "walkdir", "wiremock", + "zstd", ] [[package]] @@ -3924,6 +3931,16 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8eaf4bc02d17cbdd7ff4c7438cafcdf7fb9a4613313ad11b4f8fefe7d3fa0130" +[[package]] +name = "jobserver" +version = "0.1.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33" +dependencies = [ + "getrandom 0.3.3", + "libc", +] + [[package]] name = "js-sys" version = "0.3.77" @@ -8809,6 +8826,34 @@ dependencies = [ "syn 2.0.104", ] +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.0.16+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" +dependencies = [ + "cc", + "pkg-config", +] + [[package]] name = "zune-core" version = "0.4.12" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 645d5aa210..98241a8a2e 100644 --- a/codex-rs/Cargo.toml +++ b/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 = {} diff --git a/codex-rs/codex-api/Cargo.toml b/codex-rs/codex-api/Cargo.toml index e9fc78878b..3d14848f91 100644 --- a/codex-rs/codex-api/Cargo.toml +++ b/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 } diff --git a/codex-rs/codex-api/src/endpoint/compact.rs b/codex-rs/codex-api/src/endpoint/compact.rs index 2b02ebd0f0..0ed9568493 100644 --- a/codex-rs/codex-api/src/endpoint/compact.rs +++ b/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 CompactClient { 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::RequestCompression; use crate::provider::RetryConfig; use async_trait::async_trait; use codex_client::Request; @@ -135,6 +137,7 @@ mod tests { retry_5xx: true, retry_transport: true, }, + request_compression: RequestCompression::None, stream_idle_timeout: Duration::from_secs(1), } } diff --git a/codex-rs/codex-api/src/endpoint/models.rs b/codex-rs/codex-api/src/endpoint/models.rs index 74e109bf2a..c7be369099 100644 --- a/codex-rs/codex-api/src/endpoint/models.rs +++ b/codex-rs/codex-api/src/endpoint/models.rs @@ -82,6 +82,7 @@ impl ModelsClient { #[cfg(test)] mod tests { use super::*; + use crate::provider::RequestCompression; use crate::provider::RetryConfig; use crate::provider::WireApi; use async_trait::async_trait; @@ -158,6 +159,7 @@ mod tests { retry_5xx: true, retry_transport: true, }, + request_compression: RequestCompression::None, stream_idle_timeout: Duration::from_secs(1), } } diff --git a/codex-rs/codex-api/src/endpoint/streaming.rs b/codex-rs/codex-api/src/endpoint/streaming.rs index 156d4084bc..ea7e4c85f3 100644 --- a/codex-rs/codex-api/src/endpoint/streaming.rs +++ b/codex-rs/codex-api/src/endpoint/streaming.rs @@ -3,16 +3,24 @@ use crate::auth::add_auth_headers; use crate::common::ResponseStream; use crate::error::ApiError; use crate::provider::Provider; +use crate::provider::RequestCompression; use crate::telemetry::SseTelemetry; use crate::telemetry::run_with_request_telemetry; +use bytes::Bytes; +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 http::header::ACCEPT; +use http::header::CONTENT_ENCODING; +use http::header::CONTENT_TYPE; use serde_json::Value; use std::sync::Arc; use std::time::Duration; +use zstd::stream::encode_all; pub(crate) struct StreamingClient { transport: T, @@ -54,14 +62,24 @@ impl StreamingClient { extra_headers: HeaderMap, spawner: fn(StreamResponse, Duration, Option>) -> ResponseStream, ) -> Result { + let content_encoding = + matches!(self.provider.request_compression, RequestCompression::Zstd); + let encoded_body = + encode_body(&body, self.provider.request_compression).map_err(ApiError::Stream)?; + 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.body = Some(body.clone()); + req.headers + .insert(ACCEPT, HeaderValue::from_static("text/event-stream")); + req.headers + .entry(CONTENT_TYPE) + .or_insert_with(|| HeaderValue::from_static("application/json")); + if content_encoding { + req.headers + .insert(CONTENT_ENCODING, HeaderValue::from_static("zstd")); + } + req.body = Some(encoded_body.clone()); add_auth_headers(&self.auth, req) }; @@ -80,3 +98,16 @@ impl StreamingClient { )) } } + +fn encode_body(body: &Value, compression: RequestCompression) -> Result { + match compression { + RequestCompression::None => Ok(Body::Json(body.clone())), + RequestCompression::Zstd => { + let json = serde_json::to_vec(body) + .map_err(|err| format!("failed to encode request body as json: {err}"))?; + let compressed = encode_all(json.as_slice(), 0) + .map_err(|err| format!("failed to compress request body: {err}"))?; + Ok(Body::Bytes(Bytes::from(compressed))) + } + } +} diff --git a/codex-rs/codex-api/src/provider.rs b/codex-rs/codex-api/src/provider.rs index 8bd5fc9093..1faf8893ed 100644 --- a/codex-rs/codex-api/src/provider.rs +++ b/codex-rs/codex-api/src/provider.rs @@ -41,6 +41,12 @@ impl RetryConfig { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RequestCompression { + None, + Zstd, +} + /// HTTP endpoint configuration used to talk to a concrete API deployment. /// /// Encapsulates base URL, default headers, query params, retry policy, and @@ -53,6 +59,7 @@ pub struct Provider { pub wire: WireApi, pub headers: HeaderMap, pub retry: RetryConfig, + pub request_compression: RequestCompression, pub stream_idle_timeout: Duration, } diff --git a/codex-rs/codex-api/src/requests/chat.rs b/codex-rs/codex-api/src/requests/chat.rs index 60f450ca0d..b22ce2a3f9 100644 --- a/codex-rs/codex-api/src/requests/chat.rs +++ b/codex-rs/codex-api/src/requests/chat.rs @@ -351,6 +351,7 @@ fn push_tool_call_message(messages: &mut Vec, tool_call: Value, reasoning #[cfg(test)] mod tests { use super::*; + use crate::provider::RequestCompression; use crate::provider::RetryConfig; use crate::provider::WireApi; use codex_protocol::models::FunctionCallOutputPayload; @@ -374,6 +375,7 @@ mod tests { retry_5xx: true, retry_transport: true, }, + request_compression: RequestCompression::None, stream_idle_timeout: Duration::from_secs(1), } } diff --git a/codex-rs/codex-api/src/requests/responses.rs b/codex-rs/codex-api/src/requests/responses.rs index 543b79bbe9..28e5f600ab 100644 --- a/codex-rs/codex-api/src/requests/responses.rs +++ b/codex-rs/codex-api/src/requests/responses.rs @@ -172,6 +172,7 @@ fn attach_item_ids(payload_json: &mut Value, original_items: &[ResponseItem]) { #[cfg(test)] mod tests { use super::*; + use crate::provider::RequestCompression; use crate::provider::RetryConfig; use crate::provider::WireApi; use codex_protocol::protocol::SubAgentSource; @@ -193,6 +194,7 @@ mod tests { retry_5xx: true, retry_transport: true, }, + request_compression: RequestCompression::None, stream_idle_timeout: Duration::from_secs(5), } } diff --git a/codex-rs/codex-api/tests/clients.rs b/codex-rs/codex-api/tests/clients.rs index 3dafaf74fa..1b46edd06f 100644 --- a/codex-rs/codex-api/tests/clients.rs +++ b/codex-rs/codex-api/tests/clients.rs @@ -11,6 +11,7 @@ use codex_api::Provider; use codex_api::ResponsesClient; use codex_api::ResponsesOptions; use codex_api::WireApi; +use codex_api::provider::RequestCompression; use codex_client::HttpTransport; use codex_client::Request; use codex_client::Response; @@ -132,6 +133,7 @@ fn provider(name: &str, wire: WireApi) -> Provider { retry_5xx: false, retry_transport: true, }, + request_compression: RequestCompression::None, stream_idle_timeout: Duration::from_millis(10), } } diff --git a/codex-rs/codex-api/tests/models_integration.rs b/codex-rs/codex-api/tests/models_integration.rs index 8ed6ce0d6d..f9abf537a7 100644 --- a/codex-rs/codex-api/tests/models_integration.rs +++ b/codex-rs/codex-api/tests/models_integration.rs @@ -1,6 +1,7 @@ use codex_api::AuthProvider; use codex_api::ModelsClient; use codex_api::provider::Provider; +use codex_api::provider::RequestCompression; use codex_api::provider::RetryConfig; use codex_api::provider::WireApi; use codex_client::ReqwestTransport; @@ -42,6 +43,7 @@ fn provider(base_url: &str) -> Provider { retry_5xx: true, retry_transport: true, }, + request_compression: RequestCompression::None, stream_idle_timeout: std::time::Duration::from_secs(1), } } diff --git a/codex-rs/codex-api/tests/sse_end_to_end.rs b/codex-rs/codex-api/tests/sse_end_to_end.rs index b91cf3a5d8..abfcadf384 100644 --- a/codex-rs/codex-api/tests/sse_end_to_end.rs +++ b/codex-rs/codex-api/tests/sse_end_to_end.rs @@ -9,6 +9,7 @@ use codex_api::Provider; use codex_api::ResponseEvent; use codex_api::ResponsesClient; use codex_api::WireApi; +use codex_api::provider::RequestCompression; use codex_client::HttpTransport; use codex_client::Request; use codex_client::Response; @@ -74,6 +75,7 @@ fn provider(name: &str, wire: WireApi) -> Provider { retry_5xx: false, retry_transport: true, }, + request_compression: RequestCompression::None, stream_idle_timeout: Duration::from_millis(50), } } diff --git a/codex-rs/codex-client/src/default_client.rs b/codex-rs/codex-client/src/default_client.rs index efb4d5aec4..1ab612e842 100644 --- a/codex-rs/codex-client/src/default_client.rs +++ b/codex-rs/codex-client/src/default_client.rs @@ -104,6 +104,13 @@ impl CodexRequestBuilder { self.map(|builder| builder.json(value)) } + pub fn body(self, body: T) -> Self + where + T: Into, + { + self.map(|builder| builder.body(body)) + } + pub async fn send(self) -> Result { let headers = trace_headers(); diff --git a/codex-rs/codex-client/src/lib.rs b/codex-rs/codex-client/src/lib.rs index 66d1083c07..f0742a6f7e 100644 --- a/codex-rs/codex-client/src/lib.rs +++ b/codex-rs/codex-client/src/lib.rs @@ -10,6 +10,7 @@ pub use crate::default_client::CodexHttpClient; pub use crate::default_client::CodexRequestBuilder; pub use crate::error::StreamError; pub use crate::error::TransportError; +pub use crate::request::Body; pub use crate::request::Request; pub use crate::request::Response; pub use crate::retry::RetryOn; diff --git a/codex-rs/codex-client/src/request.rs b/codex-rs/codex-client/src/request.rs index f3d205de99..70974250d4 100644 --- a/codex-rs/codex-client/src/request.rs +++ b/codex-rs/codex-client/src/request.rs @@ -5,12 +5,18 @@ use serde::Serialize; use serde_json::Value; use std::time::Duration; +#[derive(Debug, Clone)] +pub enum Body { + Json(Value), + Bytes(Bytes), +} + #[derive(Debug, Clone)] pub struct Request { pub method: Method, pub url: String, pub headers: HeaderMap, - pub body: Option, + pub body: Option, pub timeout: Option, } @@ -26,7 +32,7 @@ impl Request { } pub fn with_json(mut self, body: &T) -> Self { - self.body = serde_json::to_value(body).ok(); + self.body = serde_json::to_value(body).ok().map(Body::Json); self } } diff --git a/codex-rs/codex-client/src/transport.rs b/codex-rs/codex-client/src/transport.rs index abe6e29ee5..4039f3c6b8 100644 --- a/codex-rs/codex-client/src/transport.rs +++ b/codex-rs/codex-client/src/transport.rs @@ -1,6 +1,7 @@ use crate::default_client::CodexHttpClient; use crate::default_client::CodexRequestBuilder; use crate::error::TransportError; +use crate::request::Body; use crate::request::Request; use crate::request::Response; use async_trait::async_trait; @@ -52,7 +53,10 @@ impl ReqwestTransport { builder = builder.timeout(timeout); } if let Some(body) = req.body { - builder = builder.json(&body); + builder = match body { + Body::Json(value) => builder.json(&value), + Body::Bytes(bytes) => builder.body(bytes), + }; } Ok(builder) } @@ -101,10 +105,10 @@ impl HttpTransport for ReqwestTransport { async fn stream(&self, req: Request) -> Result { if enabled!(Level::TRACE) { trace!( - "{} to {}: {}", - req.method, - req.url, - req.body.as_ref().unwrap_or_default() + method = %req.method, + url = %req.url, + body = ?req.body, + "Sending streaming request" ); } diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 51fab19dec..01c294b451 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -89,6 +89,7 @@ url = { workspace = true } uuid = { workspace = true, features = ["serde", "v4", "v5"] } which = { workspace = true } wildmatch = { workspace = true } +zstd = { workspace = true } [features] deterministic_process_ids = [] diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 8eac13fd2e..51a1bd07f7 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -3090,6 +3090,7 @@ model_verbosity = "high" base_url: Some("https://api.openai.com/v1".to_string()), env_key: Some("OPENAI_API_KEY".to_string()), wire_api: crate::WireApi::Chat, + request_compression: crate::RequestCompression::None, env_key_instructions: None, experimental_bearer_token: None, query_params: None, diff --git a/codex-rs/core/src/default_client.rs b/codex-rs/core/src/default_client.rs index 3cd882489d..bcf06882fd 100644 --- a/codex-rs/core/src/default_client.rs +++ b/codex-rs/core/src/default_client.rs @@ -24,7 +24,6 @@ use std::sync::OnceLock; pub static USER_AGENT_SUFFIX: LazyLock>> = LazyLock::new(|| Mutex::new(None)); pub const DEFAULT_ORIGINATOR: &str = "codex_cli_rs"; pub const CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR: &str = "CODEX_INTERNAL_ORIGINATOR_OVERRIDE"; - #[derive(Debug, Clone)] pub struct Originator { pub value: String, diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 8794484083..54b1a62e9b 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -56,6 +56,7 @@ pub use model_provider_info::DEFAULT_OLLAMA_PORT; pub use model_provider_info::LMSTUDIO_OSS_PROVIDER_ID; pub use model_provider_info::ModelProviderInfo; pub use model_provider_info::OLLAMA_OSS_PROVIDER_ID; +pub use model_provider_info::RequestCompression; pub use model_provider_info::WireApi; pub use model_provider_info::built_in_model_providers; pub use model_provider_info::create_oss_provider_with_base_url; diff --git a/codex-rs/core/src/model_provider_info.rs b/codex-rs/core/src/model_provider_info.rs index 9617392237..1271dbd587 100644 --- a/codex-rs/core/src/model_provider_info.rs +++ b/codex-rs/core/src/model_provider_info.rs @@ -47,6 +47,15 @@ pub enum WireApi { Chat, } +/// Body compression to use for outbound requests. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum RequestCompression { + #[default] + None, + Zstd, +} + /// Serializable representation of a provider definition. #[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] pub struct ModelProviderInfo { @@ -70,6 +79,10 @@ pub struct ModelProviderInfo { #[serde(default)] pub wire_api: WireApi, + /// Compression applied to request bodies. + #[serde(default)] + pub request_compression: RequestCompression, + /// Optional query parameters to append to the base URL. pub query_params: Option>, @@ -160,6 +173,10 @@ impl ModelProviderInfo { }, headers, retry, + request_compression: match self.request_compression { + RequestCompression::None => codex_api::provider::RequestCompression::None, + RequestCompression::Zstd => codex_api::provider::RequestCompression::Zstd, + }, stream_idle_timeout: self.stream_idle_timeout(), }) } @@ -225,6 +242,7 @@ impl ModelProviderInfo { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, query_params: None, http_headers: Some( [("version".to_string(), env!("CARGO_PKG_VERSION").to_string())] @@ -313,6 +331,7 @@ pub fn create_oss_provider_with_base_url(base_url: &str, wire_api: WireApi) -> M env_key_instructions: None, experimental_bearer_token: None, wire_api, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, @@ -341,6 +360,7 @@ base_url = "http://localhost:11434/v1" env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Chat, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, @@ -369,6 +389,7 @@ query_params = { api-version = "2025-04-01-preview" } env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Chat, + request_compression: RequestCompression::None, query_params: Some(maplit::hashmap! { "api-version".to_string() => "2025-04-01-preview".to_string(), }), @@ -400,6 +421,7 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" } env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Chat, + request_compression: RequestCompression::None, query_params: None, http_headers: Some(maplit::hashmap! { "X-Example-Header".to_string() => "example-value".to_string(), @@ -435,6 +457,7 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" } env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, @@ -457,6 +480,7 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" } env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, @@ -481,6 +505,7 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" } env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, diff --git a/codex-rs/core/src/models_manager/manager.rs b/codex-rs/core/src/models_manager/manager.rs index 060f4a5c27..da96a4147a 100644 --- a/codex-rs/core/src/models_manager/manager.rs +++ b/codex-rs/core/src/models_manager/manager.rs @@ -325,6 +325,7 @@ mod tests { use super::cache::ModelsCache; use super::*; use crate::CodexAuth; + use crate::RequestCompression; use crate::auth::AuthCredentialsStoreMode; use crate::config::ConfigBuilder; use crate::features::Feature; @@ -379,6 +380,7 @@ mod tests { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, diff --git a/codex-rs/core/tests/chat_completions_payload.rs b/codex-rs/core/tests/chat_completions_payload.rs index 8af5df2169..b9c97dd3a1 100644 --- a/codex-rs/core/tests/chat_completions_payload.rs +++ b/codex-rs/core/tests/chat_completions_payload.rs @@ -10,6 +10,7 @@ use codex_core::LocalShellStatus; use codex_core::ModelClient; use codex_core::ModelProviderInfo; use codex_core::Prompt; +use codex_core::RequestCompression; use codex_core::ResponseItem; use codex_core::WireApi; use codex_core::models_manager::manager::ModelsManager; @@ -52,6 +53,7 @@ async fn run_request(input: Vec) -> Value { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Chat, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, diff --git a/codex-rs/core/tests/chat_completions_sse.rs b/codex-rs/core/tests/chat_completions_sse.rs index 4f05838279..3b6cd3159c 100644 --- a/codex-rs/core/tests/chat_completions_sse.rs +++ b/codex-rs/core/tests/chat_completions_sse.rs @@ -8,6 +8,7 @@ use codex_core::ContentItem; use codex_core::ModelClient; use codex_core::ModelProviderInfo; use codex_core::Prompt; +use codex_core::RequestCompression; use codex_core::ResponseEvent; use codex_core::ResponseItem; use codex_core::WireApi; @@ -51,6 +52,7 @@ async fn run_stream_with_bytes(sse_body: &[u8]) -> Vec { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Chat, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, diff --git a/codex-rs/core/tests/common/Cargo.toml b/codex-rs/core/tests/common/Cargo.toml index c61a095686..a2b1231c97 100644 --- a/codex-rs/core/tests/common/Cargo.toml +++ b/codex-rs/core/tests/common/Cargo.toml @@ -15,14 +15,17 @@ codex-core = { workspace = true, features = ["test-support"] } codex-protocol = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-cargo-bin = { workspace = true } +http = { workspace = true } notify = { workspace = true } regex-lite = { workspace = true } serde_json = { workspace = true } +serde = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["time"] } walkdir = { workspace = true } wiremock = { workspace = true } shlex = { workspace = true } +zstd = { workspace = true } [dev-dependencies] pretty_assertions = { workspace = true } diff --git a/codex-rs/core/tests/common/lib.rs b/codex-rs/core/tests/common/lib.rs index 9568ec2786..8d8517085c 100644 --- a/codex-rs/core/tests/common/lib.rs +++ b/codex-rs/core/tests/common/lib.rs @@ -11,11 +11,15 @@ use regex_lite::Regex; use std::path::PathBuf; pub mod process; +pub mod request; pub mod responses; pub mod streaming_sse; pub mod test_codex; pub mod test_codex_exec; +pub use request::RequestBodyExt; +pub use request::body_contains; + #[track_caller] pub fn assert_regex_match<'s>(pattern: &str, actual: &'s str) -> regex_lite::Captures<'s> { let regex = Regex::new(pattern).unwrap_or_else(|err| { diff --git a/codex-rs/core/tests/common/request.rs b/codex-rs/core/tests/common/request.rs new file mode 100644 index 0000000000..ce93c4a247 --- /dev/null +++ b/codex-rs/core/tests/common/request.rs @@ -0,0 +1,59 @@ +use http::header::CONTENT_ENCODING; +use serde::de::DeserializeOwned; +use wiremock::Match; + +pub fn decoded_body_bytes(request: &wiremock::Request) -> Vec { + if is_zstd_encoded(request) { + zstd::decode_all(request.body.as_slice()).unwrap_or_else(|err| { + panic!("failed to decode zstd-encoded request body: {err}"); + }) + } else { + request.body.clone() + } +} + +pub fn decoded_body_string(request: &wiremock::Request) -> String { + String::from_utf8_lossy(&decoded_body_bytes(request)).into_owned() +} + +pub trait RequestBodyExt { + fn json_body(&self) -> T; + fn text_body(&self) -> String; +} + +impl RequestBodyExt for wiremock::Request { + fn json_body(&self) -> T { + serde_json::from_slice(&decoded_body_bytes(self)).unwrap_or_else(|err| { + panic!("failed to decode request body as JSON: {err}"); + }) + } + + fn text_body(&self) -> String { + decoded_body_string(self) + } +} + +pub fn body_contains(needle: impl Into) -> impl Match { + BodyContains { + needle: needle.into(), + } +} + +struct BodyContains { + needle: String, +} + +impl Match for BodyContains { + fn matches(&self, request: &wiremock::Request) -> bool { + decoded_body_string(request).contains(self.needle.as_str()) + } +} + +fn is_zstd_encoded(request: &wiremock::Request) -> bool { + request + .headers + .get(CONTENT_ENCODING) + .and_then(|value| value.to_str().ok()) + .map(|value| value.eq_ignore_ascii_case("zstd")) + .unwrap_or(false) +} diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index 3934771409..8d7901d9d0 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/codex-rs/core/tests/common/responses.rs @@ -15,6 +15,7 @@ use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path_regex; +use crate::RequestBodyExt; use crate::test_codex::ApplyPatchModelOutput; #[derive(Debug, Clone)] @@ -67,7 +68,7 @@ pub struct ResponsesRequest(wiremock::Request); impl ResponsesRequest { pub fn body_json(&self) -> Value { - self.0.body_json().unwrap() + self.0.json_body() } /// Returns all `input_text` spans from `message` inputs for the provided role. @@ -83,7 +84,7 @@ impl ResponsesRequest { } pub fn input(&self) -> Vec { - self.0.body_json::().unwrap()["input"] + self.body_json()["input"] .as_array() .expect("input array not found in request") .clone() @@ -721,10 +722,7 @@ pub async fn get_responses_request_bodies(server: &MockServer) -> Vec { get_responses_requests(server) .await .into_iter() - .map(|req| { - req.body_json::() - .expect("request body to be valid JSON") - }) + .map(|req| req.json_body::()) .collect() } diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index a081a6bf12..be57142e13 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -22,6 +22,7 @@ use serde_json::Value; use tempfile::TempDir; use wiremock::MockServer; +use crate::RequestBodyExt; use crate::load_default_config_for_test; use crate::responses::get_responses_request_bodies; use crate::responses::start_mock_server; diff --git a/codex-rs/core/tests/responses_headers.rs b/codex-rs/core/tests/responses_headers.rs index 3b0ffd2983..28f79217a0 100644 --- a/codex-rs/core/tests/responses_headers.rs +++ b/codex-rs/core/tests/responses_headers.rs @@ -7,6 +7,7 @@ use codex_core::ContentItem; use codex_core::ModelClient; use codex_core::ModelProviderInfo; use codex_core::Prompt; +use codex_core::RequestCompression; use codex_core::ResponseEvent; use codex_core::ResponseItem; use codex_core::WireApi; @@ -46,6 +47,7 @@ async fn responses_stream_includes_subagent_header_on_review() { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::Zstd, query_params: None, http_headers: None, env_http_headers: None, @@ -140,6 +142,7 @@ async fn responses_stream_includes_subagent_header_on_other() { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::Zstd, query_params: None, http_headers: None, env_http_headers: None, @@ -230,6 +233,7 @@ async fn responses_respects_model_family_overrides_from_config() { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, @@ -317,3 +321,193 @@ async fn responses_respects_model_family_overrides_from_config() { Some("detailed") ); } + +#[tokio::test] +async fn responses_request_body_is_zstd_encoded() { + core_test_support::skip_if_no_network!(); + + let server = responses::start_mock_server().await; + let response_body = responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_completed("resp-1"), + ]); + + let request_recorder = responses::mount_sse_once(&server, response_body).await; + + let provider = ModelProviderInfo { + name: "mock".into(), + base_url: Some(format!("{}/v1", server.uri())), + env_key: None, + env_key_instructions: None, + experimental_bearer_token: None, + wire_api: WireApi::Responses, + request_compression: RequestCompression::Zstd, + query_params: None, + http_headers: None, + env_http_headers: None, + request_max_retries: Some(0), + stream_max_retries: Some(0), + stream_idle_timeout_ms: Some(5_000), + requires_openai_auth: false, + }; + + let codex_home = TempDir::new().expect("failed to create TempDir"); + let mut config = load_default_config_for_test(&codex_home); + config.model_provider_id = provider.name.clone(); + config.model_provider = provider.clone(); + let effort = config.model_reasoning_effort; + let summary = config.model_reasoning_summary; + let model = ModelsManager::get_model_offline(config.model.as_deref()); + config.model = Some(model.clone()); + let config = Arc::new(config); + + let conversation_id = ConversationId::new(); + let session_source = SessionSource::Exec; + let model_family = ModelsManager::construct_model_family_offline(model.as_str(), &config); + let otel_manager = OtelManager::new( + conversation_id, + model.as_str(), + model_family.slug.as_str(), + None, + Some("test@test.com".to_string()), + Some(AuthMode::ChatGPT), + false, + "test".to_string(), + session_source.clone(), + ); + + let client = ModelClient::new( + Arc::clone(&config), + None, + model_family, + otel_manager, + provider, + effort, + summary, + conversation_id, + session_source, + ); + assert_eq!( + client.get_provider().request_compression, + RequestCompression::Zstd + ); + + let mut prompt = Prompt::default(); + prompt.input = vec![ResponseItem::Message { + id: None, + role: "user".into(), + content: vec![ContentItem::InputText { + text: "hello".into(), + }], + }]; + + let mut stream = client.stream(&prompt).await.expect("stream failed"); + while let Some(event) = stream.next().await { + if matches!(event, Ok(ResponseEvent::Completed { .. })) { + break; + } + } + + let request = request_recorder.single_request(); + assert_eq!(request.header("content-encoding").as_deref(), Some("zstd")); + assert_eq!( + request.header("content-type").as_deref(), + Some("application/json") + ); + let request_body = request.body_json(); + assert_eq!(request_body["stream"].as_bool(), Some(true)); + assert_eq!( + request_body["input"][0]["content"][0]["text"].as_str(), + Some("hello") + ); +} + +#[tokio::test] +async fn responses_request_body_is_uncompressed_when_disabled() { + core_test_support::skip_if_no_network!(); + + let server = responses::start_mock_server().await; + let response_body = responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_completed("resp-1"), + ]); + + let request_recorder = responses::mount_sse_once(&server, response_body).await; + + let provider = ModelProviderInfo { + name: "mock".into(), + base_url: Some(format!("{}/v1", server.uri())), + env_key: None, + env_key_instructions: None, + experimental_bearer_token: None, + wire_api: WireApi::Responses, + request_compression: RequestCompression::None, + query_params: None, + http_headers: None, + env_http_headers: None, + request_max_retries: Some(0), + stream_max_retries: Some(0), + stream_idle_timeout_ms: Some(5_000), + requires_openai_auth: false, + }; + + let codex_home = TempDir::new().expect("failed to create TempDir"); + let mut config = load_default_config_for_test(&codex_home); + config.model_provider_id = provider.name.clone(); + config.model_provider = provider.clone(); + let effort = config.model_reasoning_effort; + let summary = config.model_reasoning_summary; + let model = ModelsManager::get_model_offline(config.model.as_deref()); + config.model = Some(model.clone()); + let config = Arc::new(config); + + let conversation_id = ConversationId::new(); + let session_source = SessionSource::Exec; + let model_family = ModelsManager::construct_model_family_offline(model.as_str(), &config); + let otel_manager = OtelManager::new( + conversation_id, + model.as_str(), + model_family.slug.as_str(), + None, + Some("test@test.com".to_string()), + Some(AuthMode::ChatGPT), + false, + "test".to_string(), + session_source.clone(), + ); + + let client = ModelClient::new( + Arc::clone(&config), + None, + model_family, + otel_manager, + provider, + effort, + summary, + conversation_id, + session_source, + ); + + let mut prompt = Prompt::default(); + prompt.input = vec![ResponseItem::Message { + id: None, + role: "user".into(), + content: vec![ContentItem::InputText { + text: "hello".into(), + }], + }]; + + let mut stream = client.stream(&prompt).await.expect("stream failed"); + while let Some(event) = stream.next().await { + if matches!(event, Ok(ResponseEvent::Completed { .. })) { + break; + } + } + + let request = request_recorder.single_request(); + assert_eq!(request.header("content-encoding"), None); + assert_eq!( + request.header("content-type").as_deref(), + Some("application/json") + ); +} diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 5ce6e9f2ff..bcae6f46de 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -9,6 +9,7 @@ use codex_core::ModelClient; use codex_core::ModelProviderInfo; use codex_core::NewConversation; use codex_core::Prompt; +use codex_core::RequestCompression; use codex_core::ResponseEvent; use codex_core::ResponseItem; use codex_core::WireApi; @@ -29,6 +30,8 @@ use codex_protocol::models::ReasoningItemReasoningSummary; use codex_protocol::models::WebSearchAction; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::user_input::UserInput; +use core_test_support::RequestBodyExt; +use core_test_support::body_contains; use core_test_support::load_default_config_for_test; use core_test_support::load_sse_fixture_with_id; use core_test_support::responses::ev_completed_with_tokens; @@ -51,7 +54,6 @@ use uuid::Uuid; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; -use wiremock::matchers::body_string_contains; use wiremock::matchers::header_regex; use wiremock::matchers::method; use wiremock::matchers::path; @@ -507,7 +509,7 @@ async fn chatgpt_auth_sends_correct_request() { let request_authorization = request.headers.get("authorization").unwrap(); let request_originator = request.headers.get("originator").unwrap(); let request_chatgpt_account_id = request.headers.get("chatgpt-account-id").unwrap(); - let request_body = request.body_json::().unwrap(); + let request_body = request.json_body::(); assert_eq!( request_conversation_id.to_str().unwrap(), @@ -1125,6 +1127,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, @@ -1495,7 +1498,7 @@ async fn context_window_error_sets_total_tokens_to_model_window() -> anyhow::Res mount_sse_once_match( &server, - body_string_contains("trigger context window"), + body_contains("trigger context window"), sse_failed( "resp_context_window", "context_length_exceeded", @@ -1506,7 +1509,7 @@ async fn context_window_error_sets_total_tokens_to_model_window() -> anyhow::Res mount_sse_once_match( &server, - body_string_contains("seed turn"), + body_contains("seed turn"), sse_completed("resp_seed"), ) .await; @@ -1623,6 +1626,7 @@ async fn azure_overrides_assign_properties_used_for_responses_url() { )])), env_key_instructions: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, http_headers: Some(std::collections::HashMap::from([( "Custom-Header".to_string(), "Value".to_string(), @@ -1706,6 +1710,7 @@ async fn env_var_overrides_loaded_auth() { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, http_headers: Some(std::collections::HashMap::from([( "Custom-Header".to_string(), "Value".to_string(), @@ -1882,8 +1887,7 @@ async fn history_dedupes_streamed_and_final_messages_across_turns() { ]); let r3_input_array = requests[2] - .body_json::() - .unwrap() + .json_body::() .get("input") .and_then(|v| v.as_array()) .cloned() diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index 09b5fb18b8..4b88b916ae 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -17,6 +17,7 @@ use codex_core::protocol::SandboxPolicy; use codex_core::protocol::WarningEvent; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::user_input::UserInput; +use core_test_support::RequestBodyExt; use core_test_support::load_default_config_for_test; use core_test_support::responses::ev_local_shell_call; use core_test_support::responses::ev_reasoning_item; @@ -132,7 +133,6 @@ async fn summarize_context_three_requests_and_instructions() { // SSE 3: minimal completed; we only need to capture the request body. let sse3 = sse(vec![ev_completed("r3")]); - // Mount the three expected requests in sequence so the assertions below can // inspect them without relying on specific prompt markers. let request_log = mount_sse_sequence(&server, vec![sse1, sse2, sse3]).await; @@ -361,7 +361,8 @@ async fn manual_compact_uses_custom_prompt() { let requests = get_responses_requests(&server).await; let body = requests .iter() - .find_map(|req| req.body_json::().ok()) + .map(core_test_support::RequestBodyExt::json_body::) + .next() .expect("summary request body"); let input = body @@ -591,9 +592,7 @@ async fn multiple_auto_compact_per_task_runs_after_token_limit_hit() { // collect the requests payloads from the model let requests_payloads = get_responses_requests(&server).await; - let body = requests_payloads[0] - .body_json::() - .unwrap(); + let body = requests_payloads[0].json_body::(); let input = body.get("input").and_then(|v| v.as_array()).unwrap(); fn normalize_inputs(values: &[serde_json::Value]) -> Vec { @@ -634,9 +633,7 @@ async fn multiple_auto_compact_per_task_runs_after_token_limit_hit() { prefixed_third_summary.as_str(), ]; for (i, expected_summary) in compaction_indices.into_iter().zip(expected_summaries) { - let body = requests_payloads.clone()[i] - .body_json::() - .unwrap(); + let body = requests_payloads.clone()[i].json_body::(); let input = body.get("input").and_then(|v| v.as_array()).unwrap(); let input = normalize_inputs(input); assert_eq!(input.len(), 3); @@ -999,7 +996,7 @@ async fn multiple_auto_compact_per_task_runs_after_token_limit_hit() { ]); for (i, request) in requests_payloads.iter().enumerate() { - let body = request.body_json::().unwrap(); + let body = request.json_body::(); let input = body.get("input").and_then(|v| v.as_array()).unwrap(); let expected_input = expected_requests_inputs[i].as_array().unwrap(); assert_eq!(normalize_inputs(input), normalize_inputs(expected_input)); @@ -1038,30 +1035,30 @@ async fn auto_compact_runs_after_token_limit_hit() { let prefixed_auto_summary = AUTO_SUMMARY_TEXT; let first_matcher = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); + let body = req.text_body(); body.contains(FIRST_AUTO_MSG) && !body.contains(SECOND_AUTO_MSG) - && !body_contains_text(body, SUMMARIZATION_PROMPT) + && !body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) }; mount_sse_once_match(&server, first_matcher, sse1).await; let second_matcher = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); + let body = req.text_body(); body.contains(SECOND_AUTO_MSG) && body.contains(FIRST_AUTO_MSG) - && !body_contains_text(body, SUMMARIZATION_PROMPT) + && !body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) }; mount_sse_once_match(&server, second_matcher, sse2).await; let third_matcher = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - body_contains_text(body, SUMMARIZATION_PROMPT) + let body = req.text_body(); + body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) }; mount_sse_once_match(&server, third_matcher, sse3).await; - let fourth_matcher = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - body.contains(POST_AUTO_USER_MSG) && !body_contains_text(body, SUMMARIZATION_PROMPT) + let body = req.text_body(); + body.contains(POST_AUTO_USER_MSG) + && !body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) }; mount_sse_once_match(&server, fourth_matcher, sse4).await; @@ -1126,10 +1123,7 @@ async fn auto_compact_runs_after_token_limit_hit() { requests.len() ); let is_auto_compact = |req: &wiremock::Request| { - body_contains_text( - std::str::from_utf8(&req.body).unwrap_or(""), - SUMMARIZATION_PROMPT, - ) + body_contains_text(req.text_body().as_str(), SUMMARIZATION_PROMPT) }; let auto_compact_count = requests.iter().filter(|req| is_auto_compact(req)).count(); assert_eq!( @@ -1151,20 +1145,16 @@ async fn auto_compact_runs_after_token_limit_hit() { .enumerate() .rev() .find_map(|(idx, req)| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - (body.contains(POST_AUTO_USER_MSG) && !body_contains_text(body, SUMMARIZATION_PROMPT)) + let body = req.text_body(); + (body.contains(POST_AUTO_USER_MSG) && !body_contains_text(&body, SUMMARIZATION_PROMPT)) .then_some(idx) }) .expect("follow-up request missing"); assert_eq!(follow_up_index, 3, "follow-up request should be last"); - let body_first = requests[0].body_json::().unwrap(); - let body_auto = requests[auto_compact_index] - .body_json::() - .unwrap(); - let body_follow_up = requests[follow_up_index] - .body_json::() - .unwrap(); + let body_first = requests[0].json_body::(); + let body_auto = requests[auto_compact_index].json_body::(); + let body_follow_up = requests[follow_up_index].json_body::(); let instructions = body_auto .get("instructions") .and_then(|v| v.as_str()) @@ -1375,24 +1365,24 @@ async fn auto_compact_persists_rollout_entries() { ]); let first_matcher = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); + let body = req.text_body(); body.contains(FIRST_AUTO_MSG) && !body.contains(SECOND_AUTO_MSG) - && !body_contains_text(body, SUMMARIZATION_PROMPT) + && !body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) }; mount_sse_once_match(&server, first_matcher, sse1).await; let second_matcher = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); + let body = req.text_body(); body.contains(SECOND_AUTO_MSG) && body.contains(FIRST_AUTO_MSG) - && !body_contains_text(body, SUMMARIZATION_PROMPT) + && !body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) }; mount_sse_once_match(&server, second_matcher, sse2).await; let third_matcher = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - body_contains_text(body, SUMMARIZATION_PROMPT) + let body = req.text_body(); + body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) }; mount_sse_once_match(&server, third_matcher, sse3).await; diff --git a/codex-rs/core/tests/suite/compact_resume_fork.rs b/codex-rs/core/tests/suite/compact_resume_fork.rs index 3e38c89b33..4edb1c83dd 100644 --- a/codex-rs/core/tests/suite/compact_resume_fork.rs +++ b/codex-rs/core/tests/suite/compact_resume_fork.rs @@ -23,6 +23,7 @@ use codex_core::protocol::Op; use codex_core::protocol::WarningEvent; use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR; use codex_protocol::user_input::UserInput; +use core_test_support::RequestBodyExt; use core_test_support::load_default_config_for_test; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; @@ -796,8 +797,9 @@ async fn mount_initial_flow(server: &MockServer) { let sse5 = sse(vec![ev_completed("r5")]); let match_first = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); + let body = req.text_body(); body.contains("\"text\":\"hello world\"") + && !body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) && !body.contains(&format!("\"text\":\"{SUMMARY_TEXT}\"")) && !body.contains("\"text\":\"AFTER_COMPACT\"") && !body.contains("\"text\":\"AFTER_RESUME\"") @@ -806,13 +808,13 @@ async fn mount_initial_flow(server: &MockServer) { mount_sse_once_match(server, match_first, sse1).await; let match_compact = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - body_contains_text(body, SUMMARIZATION_PROMPT) || body.contains(&json_fragment(FIRST_REPLY)) + let body = req.text_body(); + body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) }; mount_sse_once_match(server, match_compact, sse2).await; let match_after_compact = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); + let body = req.text_body(); body.contains("\"text\":\"AFTER_COMPACT\"") && !body.contains("\"text\":\"AFTER_RESUME\"") && !body.contains("\"text\":\"AFTER_FORK\"") @@ -820,13 +822,13 @@ async fn mount_initial_flow(server: &MockServer) { mount_sse_once_match(server, match_after_compact, sse3).await; let match_after_resume = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); + let body = req.text_body(); body.contains("\"text\":\"AFTER_RESUME\"") }; mount_sse_once_match(server, match_after_resume, sse4).await; let match_after_fork = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); + let body = req.text_body(); body.contains("\"text\":\"AFTER_FORK\"") }; mount_sse_once_match(server, match_after_fork, sse5).await; @@ -840,13 +842,13 @@ async fn mount_second_compact_flow(server: &MockServer) { let sse7 = sse(vec![ev_completed("r7")]); let match_second_compact = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - body.contains("AFTER_FORK") + let body = req.text_body(); + body_contains_text(body.as_str(), SUMMARIZATION_PROMPT) && body.contains("AFTER_FORK") }; mount_sse_once_match(server, match_second_compact, sse6).await; let match_after_second_resume = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); + let body = req.text_body(); body.contains(&format!("\"text\":\"{AFTER_SECOND_RESUME}\"")) }; mount_sse_once_match(server, match_after_second_resume, sse7).await; diff --git a/codex-rs/core/tests/suite/json_result.rs b/codex-rs/core/tests/suite/json_result.rs index 1b9949102e..7731b8b7e0 100644 --- a/codex-rs/core/tests/suite/json_result.rs +++ b/codex-rs/core/tests/suite/json_result.rs @@ -6,6 +6,7 @@ use codex_core::protocol::Op; use codex_core::protocol::SandboxPolicy; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::user_input::UserInput; +use core_test_support::RequestBodyExt; use core_test_support::responses; use core_test_support::skip_if_no_network; use core_test_support::test_codex::TestCodex; @@ -54,7 +55,7 @@ async fn codex_returns_json_result(model: String) -> anyhow::Result<()> { let expected_schema: serde_json::Value = serde_json::from_str(SCHEMA)?; let match_json_text_param = move |req: &wiremock::Request| { - let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap_or_default(); + let body: serde_json::Value = req.json_body(); let Some(text) = body.get("text") else { return false; }; diff --git a/codex-rs/core/tests/suite/review.rs b/codex-rs/core/tests/suite/review.rs index b35213f7e6..d09caafe35 100644 --- a/codex-rs/core/tests/suite/review.rs +++ b/codex-rs/core/tests/suite/review.rs @@ -21,6 +21,7 @@ use codex_core::protocol::RolloutItem; use codex_core::protocol::RolloutLine; use codex_core::review_format::render_review_output_text; use codex_protocol::user_input::UserInput; +use core_test_support::RequestBodyExt; use core_test_support::load_default_config_for_test; use core_test_support::load_sse_fixture_with_id_from_str; use core_test_support::responses::get_responses_requests; @@ -430,7 +431,7 @@ async fn review_uses_custom_review_model_from_config() { let request = requests .first() .expect("expected POST request to /responses"); - let body = request.body_json::().unwrap(); + let body = request.json_body::(); assert_eq!(body["model"].as_str().unwrap(), "gpt-5.1"); server.verify().await; @@ -551,7 +552,7 @@ async fn review_input_isolated_from_parent_history() { let request = requests .first() .expect("expected POST request to /responses"); - let body = request.body_json::().unwrap(); + let body = request.json_body::(); let input = body["input"].as_array().expect("input array"); assert!( input.len() >= 2, @@ -676,7 +677,7 @@ async fn review_history_surfaces_in_parent_session() { // Critically, no messages from the review thread should appear. let requests = get_responses_requests(&server).await; assert_eq!(requests.len(), 2); - let body = requests[1].body_json::().unwrap(); + let body = requests[1].json_body::(); let input = body["input"].as_array().expect("input array"); // Must include the followup as the last item for this turn diff --git a/codex-rs/core/tests/suite/stream_error_allows_next_turn.rs b/codex-rs/core/tests/suite/stream_error_allows_next_turn.rs index e7a6091264..6bdcacc3b5 100644 --- a/codex-rs/core/tests/suite/stream_error_allows_next_turn.rs +++ b/codex-rs/core/tests/suite/stream_error_allows_next_turn.rs @@ -1,8 +1,10 @@ use codex_core::ModelProviderInfo; +use codex_core::RequestCompression; use codex_core::WireApi; use codex_core::protocol::EventMsg; use codex_core::protocol::Op; use codex_protocol::user_input::UserInput; +use core_test_support::body_contains; use core_test_support::load_sse_fixture_with_id; use core_test_support::skip_if_no_network; use core_test_support::test_codex::TestCodex; @@ -11,7 +13,6 @@ use core_test_support::wait_for_event; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; -use wiremock::matchers::body_string_contains; use wiremock::matchers::method; use wiremock::matchers::path; @@ -38,7 +39,7 @@ async fn continue_after_stream_error() { // so the failing request should only occur once. Mock::given(method("POST")) .and(path("/v1/responses")) - .and(body_string_contains("first message")) + .and(body_contains("first message")) .respond_with(fail) .up_to_n_times(2) .mount(&server) @@ -50,7 +51,7 @@ async fn continue_after_stream_error() { Mock::given(method("POST")) .and(path("/v1/responses")) - .and(body_string_contains("follow up")) + .and(body_contains("follow up")) .respond_with(ok) .expect(1) .mount(&server) @@ -66,6 +67,7 @@ async fn continue_after_stream_error() { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None, diff --git a/codex-rs/core/tests/suite/stream_no_completed.rs b/codex-rs/core/tests/suite/stream_no_completed.rs index a203658c2f..12b04bd6a1 100644 --- a/codex-rs/core/tests/suite/stream_no_completed.rs +++ b/codex-rs/core/tests/suite/stream_no_completed.rs @@ -2,6 +2,7 @@ //! delivering a `response.completed` event. use codex_core::ModelProviderInfo; +use codex_core::RequestCompression; use codex_core::WireApi; use codex_core::protocol::EventMsg; use codex_core::protocol::Op; @@ -73,6 +74,7 @@ async fn retries_on_early_close() { env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Responses, + request_compression: RequestCompression::None, query_params: None, http_headers: None, env_http_headers: None,