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,