From d5e7248958b9c9b82cd71f3b33cd9d3b16224100 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Tue, 3 Feb 2026 14:08:09 +0000 Subject: [PATCH 1/3] feat: clean codex-api part 1 (#10501) --- codex-rs/codex-api/src/auth.rs | 16 ++- codex-rs/codex-api/src/endpoint/compact.rs | 67 +++------- codex-rs/codex-api/src/endpoint/mod.rs | 2 +- codex-rs/codex-api/src/endpoint/models.rs | 50 +++---- codex-rs/codex-api/src/endpoint/responses.rs | 48 ++++--- .../src/endpoint/responses_websocket.rs | 18 +-- codex-rs/codex-api/src/endpoint/session.rs | 126 ++++++++++++++++++ codex-rs/codex-api/src/endpoint/streaming.rs | 95 ------------- 8 files changed, 208 insertions(+), 214 deletions(-) create mode 100644 codex-rs/codex-api/src/endpoint/session.rs delete mode 100644 codex-rs/codex-api/src/endpoint/streaming.rs diff --git a/codex-rs/codex-api/src/auth.rs b/codex-rs/codex-api/src/auth.rs index 6c26963cba..f649062db1 100644 --- a/codex-rs/codex-api/src/auth.rs +++ b/codex-rs/codex-api/src/auth.rs @@ -1,4 +1,6 @@ use codex_client::Request; +use http::HeaderMap; +use http::HeaderValue; /// Provides bearer and account identity information for API requests. /// @@ -12,16 +14,20 @@ pub trait AuthProvider: Send + Sync { } } -pub(crate) fn add_auth_headers(auth: &A, mut req: Request) -> Request { +pub(crate) fn add_auth_headers_to_header_map(auth: &A, headers: &mut HeaderMap) { if let Some(token) = auth.bearer_token() - && let Ok(header) = format!("Bearer {token}").parse() + && let Ok(header) = HeaderValue::from_str(&format!("Bearer {token}")) { - let _ = req.headers.insert(http::header::AUTHORIZATION, header); + let _ = headers.insert(http::header::AUTHORIZATION, header); } if let Some(account_id) = auth.account_id() - && let Ok(header) = account_id.parse() + && let Ok(header) = HeaderValue::from_str(&account_id) { - let _ = req.headers.insert("ChatGPT-Account-ID", header); + let _ = headers.insert("ChatGPT-Account-ID", header); } +} + +pub(crate) fn add_auth_headers(auth: &A, mut req: Request) -> Request { + add_auth_headers_to_header_map(auth, &mut req.headers); req } diff --git a/codex-rs/codex-api/src/endpoint/compact.rs b/codex-rs/codex-api/src/endpoint/compact.rs index b7166779a2..44a56a11a7 100644 --- a/codex-rs/codex-api/src/endpoint/compact.rs +++ b/codex-rs/codex-api/src/endpoint/compact.rs @@ -1,9 +1,8 @@ use crate::auth::AuthProvider; -use crate::auth::add_auth_headers; use crate::common::CompactionInput; +use crate::endpoint::session::EndpointSession; use crate::error::ApiError; use crate::provider::Provider; -use crate::telemetry::run_with_request_telemetry; use codex_client::HttpTransport; use codex_client::RequestTelemetry; use codex_protocol::models::ResponseItem; @@ -14,28 +13,23 @@ use serde_json::to_value; use std::sync::Arc; pub struct CompactClient { - transport: T, - provider: Provider, - auth: A, - request_telemetry: Option>, + session: EndpointSession, } impl CompactClient { pub fn new(transport: T, provider: Provider, auth: A) -> Self { Self { - transport, - provider, - auth, - request_telemetry: None, + session: EndpointSession::new(transport, provider, auth), } } - pub fn with_telemetry(mut self, request: Option>) -> Self { - self.request_telemetry = request; - self + pub fn with_telemetry(self, request: Option>) -> Self { + Self { + session: self.session.with_request_telemetry(request), + } } - fn path(&self) -> &'static str { + fn path() -> &'static str { "responses/compact" } @@ -44,21 +38,10 @@ impl CompactClient { body: serde_json::Value, extra_headers: HeaderMap, ) -> Result, ApiError> { - let path = self.path(); - let builder = || { - let mut req = self.provider.build_request(Method::POST, path); - req.headers.extend(extra_headers.clone()); - req.body = Some(body.clone()); - add_auth_headers(&self.auth, req) - }; - - let resp = run_with_request_telemetry( - self.provider.retry.to_policy(), - self.request_telemetry.clone(), - builder, - |req| self.transport.execute(req), - ) - .await?; + let resp = self + .session + .execute(Method::POST, Self::path(), extra_headers, Some(body)) + .await?; let parsed: CompactHistoryResponse = serde_json::from_slice(&resp.body).map_err(|e| ApiError::Stream(e.to_string()))?; Ok(parsed.output) @@ -83,14 +66,11 @@ struct CompactHistoryResponse { #[cfg(test)] mod tests { use super::*; - use crate::provider::RetryConfig; use async_trait::async_trait; use codex_client::Request; use codex_client::Response; use codex_client::StreamResponse; use codex_client::TransportError; - use http::HeaderMap; - use std::time::Duration; #[derive(Clone, Default)] struct DummyTransport; @@ -115,26 +95,11 @@ mod tests { } } - fn provider() -> Provider { - Provider { - name: "test".to_string(), - base_url: "https://example.com/v1".to_string(), - query_params: None, - headers: HeaderMap::new(), - retry: RetryConfig { - max_attempts: 1, - base_delay: Duration::from_millis(1), - retry_429: false, - retry_5xx: true, - retry_transport: true, - }, - stream_idle_timeout: Duration::from_secs(1), - } - } - #[test] fn path_is_responses_compact() { - let client = CompactClient::new(DummyTransport, provider(), DummyAuth); - assert_eq!(client.path(), "responses/compact"); + assert_eq!( + CompactClient::::path(), + "responses/compact" + ); } } diff --git a/codex-rs/codex-api/src/endpoint/mod.rs b/codex-rs/codex-api/src/endpoint/mod.rs index 2bb524b227..23579ffcf1 100644 --- a/codex-rs/codex-api/src/endpoint/mod.rs +++ b/codex-rs/codex-api/src/endpoint/mod.rs @@ -3,4 +3,4 @@ pub mod compact; pub mod models; pub mod responses; pub mod responses_websocket; -mod streaming; +mod session; diff --git a/codex-rs/codex-api/src/endpoint/models.rs b/codex-rs/codex-api/src/endpoint/models.rs index fa4dddd325..5d1c5fb12e 100644 --- a/codex-rs/codex-api/src/endpoint/models.rs +++ b/codex-rs/codex-api/src/endpoint/models.rs @@ -1,8 +1,7 @@ use crate::auth::AuthProvider; -use crate::auth::add_auth_headers; +use crate::endpoint::session::EndpointSession; use crate::error::ApiError; use crate::provider::Provider; -use crate::telemetry::run_with_request_telemetry; use codex_client::HttpTransport; use codex_client::RequestTelemetry; use codex_protocol::openai_models::ModelInfo; @@ -13,53 +12,42 @@ use http::header::ETAG; use std::sync::Arc; pub struct ModelsClient { - transport: T, - provider: Provider, - auth: A, - request_telemetry: Option>, + session: EndpointSession, } impl ModelsClient { pub fn new(transport: T, provider: Provider, auth: A) -> Self { Self { - transport, - provider, - auth, - request_telemetry: None, + session: EndpointSession::new(transport, provider, auth), } } - pub fn with_telemetry(mut self, request: Option>) -> Self { - self.request_telemetry = request; - self + pub fn with_telemetry(self, request: Option>) -> Self { + Self { + session: self.session.with_request_telemetry(request), + } } - fn path(&self) -> &'static str { + fn path() -> &'static str { "models" } + fn append_client_version_query(req: &mut codex_client::Request, client_version: &str) { + let separator = if req.url.contains('?') { '&' } else { '?' }; + req.url = format!("{}{}client_version={client_version}", req.url, separator); + } + pub async fn list_models( &self, client_version: &str, extra_headers: HeaderMap, ) -> Result<(Vec, Option), ApiError> { - let builder = || { - let mut req = self.provider.build_request(Method::GET, self.path()); - req.headers.extend(extra_headers.clone()); - - let separator = if req.url.contains('?') { '&' } else { '?' }; - req.url = format!("{}{}client_version={client_version}", req.url, separator); - - add_auth_headers(&self.auth, req) - }; - - let resp = run_with_request_telemetry( - self.provider.retry.to_policy(), - self.request_telemetry.clone(), - builder, - |req| self.transport.execute(req), - ) - .await?; + let resp = self + .session + .execute_with(Method::GET, Self::path(), extra_headers, None, |req| { + Self::append_client_version_query(req, client_version); + }) + .await?; let header_etag = resp .headers diff --git a/codex-rs/codex-api/src/endpoint/responses.rs b/codex-rs/codex-api/src/endpoint/responses.rs index 2555ab5e63..6a74ad69c3 100644 --- a/codex-rs/codex-api/src/endpoint/responses.rs +++ b/codex-rs/codex-api/src/endpoint/responses.rs @@ -3,7 +3,7 @@ use crate::common::Prompt as ApiPrompt; use crate::common::Reasoning; use crate::common::ResponseStream; use crate::common::TextControls; -use crate::endpoint::streaming::StreamingClient; +use crate::endpoint::session::EndpointSession; use crate::error::ApiError; use crate::provider::Provider; use crate::requests::ResponsesRequest; @@ -16,13 +16,16 @@ use codex_client::RequestCompression; use codex_client::RequestTelemetry; use codex_protocol::protocol::SessionSource; use http::HeaderMap; +use http::HeaderValue; +use http::Method; use serde_json::Value; use std::sync::Arc; use std::sync::OnceLock; use tracing::instrument; pub struct ResponsesClient { - streaming: StreamingClient, + session: EndpointSession, + sse_telemetry: Option>, } #[derive(Default)] @@ -42,7 +45,8 @@ pub struct ResponsesOptions { impl ResponsesClient { pub fn new(transport: T, provider: Provider, auth: A) -> Self { Self { - streaming: StreamingClient::new(transport, provider, auth), + session: EndpointSession::new(transport, provider, auth), + sse_telemetry: None, } } @@ -52,7 +56,8 @@ impl ResponsesClient { sse: Option>, ) -> Self { Self { - streaming: self.streaming.with_telemetry(request, sse), + session: self.session.with_request_telemetry(request), + sse_telemetry: sse, } } @@ -102,12 +107,12 @@ impl ResponsesClient { .store_override(store_override) .extra_headers(extra_headers) .compression(compression) - .build(self.streaming.provider())?; + .build(self.session.provider())?; self.stream_request(request, turn_state).await } - fn path(&self) -> &'static str { + fn path() -> &'static str { "responses" } @@ -118,20 +123,33 @@ impl ResponsesClient { compression: Compression, turn_state: Option>>, ) -> Result { - let compression = match compression { + let request_compression = match compression { Compression::None => RequestCompression::None, Compression::Zstd => RequestCompression::Zstd, }; - self.streaming - .stream( - self.path(), - body, + let stream_response = self + .session + .stream_with( + Method::POST, + Self::path(), extra_headers, - compression, - spawn_response_stream, - turn_state, + Some(body), + |req| { + req.headers.insert( + http::header::ACCEPT, + HeaderValue::from_static("text/event-stream"), + ); + req.compression = request_compression; + }, ) - .await + .await?; + + Ok(spawn_response_stream( + stream_response, + self.session.provider().stream_idle_timeout, + self.sse_telemetry.clone(), + turn_state, + )) } } diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index 2a7d8726f1..cac686dd02 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -1,4 +1,5 @@ use crate::auth::AuthProvider; +use crate::auth::add_auth_headers_to_header_map; use crate::common::ResponseEvent; use crate::common::ResponseStream; use crate::common::ResponsesWsRequest; @@ -11,7 +12,6 @@ use codex_client::TransportError; use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; -use http::HeaderValue; use serde_json::Value; use std::sync::Arc; use std::sync::OnceLock; @@ -134,7 +134,7 @@ impl ResponsesWebsocketClient { let mut headers = self.provider.headers.clone(); headers.extend(extra_headers); - apply_auth_headers(&mut headers, &self.auth); + add_auth_headers_to_header_map(&self.auth, &mut headers); let (stream, server_reasoning_included) = connect_websocket(ws_url, headers, turn_state).await?; @@ -147,20 +147,6 @@ impl ResponsesWebsocketClient { } } -// TODO (pakrym): share with /auth -fn apply_auth_headers(headers: &mut HeaderMap, auth: &impl AuthProvider) { - if let Some(token) = auth.bearer_token() - && let Ok(header) = HeaderValue::from_str(&format!("Bearer {token}")) - { - let _ = headers.insert(http::header::AUTHORIZATION, header); - } - if let Some(account_id) = auth.account_id() - && let Ok(header) = HeaderValue::from_str(&account_id) - { - let _ = headers.insert("ChatGPT-Account-ID", header); - } -} - async fn connect_websocket( url: Url, headers: HeaderMap, diff --git a/codex-rs/codex-api/src/endpoint/session.rs b/codex-rs/codex-api/src/endpoint/session.rs new file mode 100644 index 0000000000..a6cd7bfe37 --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/session.rs @@ -0,0 +1,126 @@ +use crate::auth::AuthProvider; +use crate::auth::add_auth_headers; +use crate::error::ApiError; +use crate::provider::Provider; +use crate::telemetry::run_with_request_telemetry; +use codex_client::HttpTransport; +use codex_client::Request; +use codex_client::RequestTelemetry; +use codex_client::Response; +use codex_client::StreamResponse; +use http::HeaderMap; +use http::Method; +use serde_json::Value; +use std::sync::Arc; + +pub(crate) struct EndpointSession { + transport: T, + provider: Provider, + auth: A, + request_telemetry: Option>, +} + +impl EndpointSession { + pub(crate) fn new(transport: T, provider: Provider, auth: A) -> Self { + Self { + transport, + provider, + auth, + request_telemetry: None, + } + } + + pub(crate) fn with_request_telemetry( + mut self, + request: Option>, + ) -> Self { + self.request_telemetry = request; + self + } + + pub(crate) fn provider(&self) -> &Provider { + &self.provider + } + + fn make_request( + &self, + method: &Method, + path: &str, + extra_headers: &HeaderMap, + body: Option<&Value>, + ) -> Request { + let mut req = self.provider.build_request(method.clone(), path); + req.headers.extend(extra_headers.clone()); + if let Some(body) = body { + req.body = Some(body.clone()); + } + add_auth_headers(&self.auth, req) + } + + pub(crate) async fn execute( + &self, + method: Method, + path: &str, + extra_headers: HeaderMap, + body: Option, + ) -> Result { + self.execute_with(method, path, extra_headers, body, |_| {}) + .await + } + + pub(crate) async fn execute_with( + &self, + method: Method, + path: &str, + extra_headers: HeaderMap, + body: Option, + configure: C, + ) -> Result + where + C: Fn(&mut Request), + { + let make_request = || { + let mut req = self.make_request(&method, path, &extra_headers, body.as_ref()); + configure(&mut req); + req + }; + + let response = run_with_request_telemetry( + self.provider.retry.to_policy(), + self.request_telemetry.clone(), + make_request, + |req| self.transport.execute(req), + ) + .await?; + + Ok(response) + } + + pub(crate) async fn stream_with( + &self, + method: Method, + path: &str, + extra_headers: HeaderMap, + body: Option, + configure: C, + ) -> Result + where + C: Fn(&mut Request), + { + let make_request = || { + let mut req = self.make_request(&method, path, &extra_headers, body.as_ref()); + configure(&mut req); + req + }; + + let stream = run_with_request_telemetry( + self.provider.retry.to_policy(), + self.request_telemetry.clone(), + make_request, + |req| self.transport.stream(req), + ) + .await?; + + Ok(stream) + } +} diff --git a/codex-rs/codex-api/src/endpoint/streaming.rs b/codex-rs/codex-api/src/endpoint/streaming.rs deleted file mode 100644 index 15d4c077a0..0000000000 --- a/codex-rs/codex-api/src/endpoint/streaming.rs +++ /dev/null @@ -1,95 +0,0 @@ -use crate::auth::AuthProvider; -use crate::auth::add_auth_headers; -use crate::common::ResponseStream; -use crate::error::ApiError; -use crate::provider::Provider; -use crate::telemetry::SseTelemetry; -use crate::telemetry::run_with_request_telemetry; -use codex_client::HttpTransport; -use codex_client::RequestCompression; -use codex_client::RequestTelemetry; -use codex_client::StreamResponse; -use http::HeaderMap; -use http::Method; -use serde_json::Value; -use std::sync::Arc; -use std::sync::OnceLock; -use std::time::Duration; - -pub(crate) struct StreamingClient { - transport: T, - provider: Provider, - auth: A, - request_telemetry: Option>, - sse_telemetry: Option>, -} - -type StreamSpawner = fn( - StreamResponse, - Duration, - Option>, - Option>>, -) -> ResponseStream; - -impl StreamingClient { - pub(crate) fn new(transport: T, provider: Provider, auth: A) -> Self { - Self { - transport, - provider, - auth, - request_telemetry: None, - sse_telemetry: None, - } - } - - pub(crate) fn with_telemetry( - mut self, - request: Option>, - sse: Option>, - ) -> Self { - self.request_telemetry = request; - self.sse_telemetry = sse; - self - } - - pub(crate) fn provider(&self) -> &Provider { - &self.provider - } - - pub(crate) async fn stream( - &self, - path: &str, - body: Value, - extra_headers: HeaderMap, - compression: RequestCompression, - spawner: StreamSpawner, - turn_state: Option>>, - ) -> Result { - 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.compression = compression; - add_auth_headers(&self.auth, req) - }; - - let stream_response = run_with_request_telemetry( - self.provider.retry.to_policy(), - self.request_telemetry.clone(), - builder, - |req| self.transport.stream(req), - ) - .await?; - - Ok(spawner( - stream_response, - self.provider.stream_idle_timeout, - self.sse_telemetry.clone(), - turn_state, - )) - } -} From 944541e93640a744f1a7aed2f214d2fa567f5cad Mon Sep 17 00:00:00 2001 From: gt-oai Date: Tue, 3 Feb 2026 14:58:33 +0000 Subject: [PATCH 2/3] Add more detail to 401 error (#10508) Add the error.message if it exists, the body otherwise. Truncate body to 1k characters. Print the cf-ray and the requestId. **Before:** Screenshot 2026-02-03 at 13 15 28 **After:** Screenshot 2026-02-03 at 13 15 38 --- codex-rs/core/src/api_bridge.rs | 26 ++++--- codex-rs/core/src/client.rs | 29 +++----- codex-rs/core/src/error.rs | 117 ++++++++++++++++++++++++++++++-- 3 files changed, 140 insertions(+), 32 deletions(-) diff --git a/codex-rs/core/src/api_bridge.rs b/codex-rs/core/src/api_bridge.rs index 86aeaedda0..f7aeb570bd 100644 --- a/codex-rs/core/src/api_bridge.rs +++ b/codex-rs/core/src/api_bridge.rs @@ -28,6 +28,7 @@ pub(crate) fn map_api_error(err: ApiError) -> CodexErr { status, body: message, url: None, + cf_ray: None, request_id: None, }), ApiError::InvalidRequest { message } => CodexErr::InvalidRequest(message), @@ -89,13 +90,14 @@ pub(crate) fn map_api_error(err: ApiError) -> CodexErr { CodexErr::RetryLimit(RetryLimitReachedError { status, - request_id: extract_request_id(headers.as_ref()), + request_id: extract_request_tracking_id(headers.as_ref()), }) } else { CodexErr::UnexpectedStatus(UnexpectedResponseError { status, body: body_text, url, + cf_ray: extract_header(headers.as_ref(), CF_RAY_HEADER), request_id: extract_request_id(headers.as_ref()), }) } @@ -115,6 +117,9 @@ pub(crate) fn map_api_error(err: ApiError) -> CodexErr { const MODEL_CAP_MODEL_HEADER: &str = "x-codex-model-cap-model"; const MODEL_CAP_RESET_AFTER_HEADER: &str = "x-codex-model-cap-reset-after-seconds"; +const REQUEST_ID_HEADER: &str = "x-request-id"; +const OAI_REQUEST_ID_HEADER: &str = "x-oai-request-id"; +const CF_RAY_HEADER: &str = "cf-ray"; #[cfg(test)] mod tests { @@ -149,15 +154,20 @@ mod tests { } } +fn extract_request_tracking_id(headers: Option<&HeaderMap>) -> Option { + extract_request_id(headers).or_else(|| extract_header(headers, CF_RAY_HEADER)) +} + fn extract_request_id(headers: Option<&HeaderMap>) -> Option { + extract_header(headers, REQUEST_ID_HEADER) + .or_else(|| extract_header(headers, OAI_REQUEST_ID_HEADER)) +} + +fn extract_header(headers: Option<&HeaderMap>, name: &str) -> Option { headers.and_then(|map| { - ["cf-ray", "x-request-id", "x-oai-request-id"] - .iter() - .find_map(|name| { - map.get(*name) - .and_then(|v| v.to_str().ok()) - .map(str::to_string) - }) + map.get(name) + .and_then(|value| value.to_str().ok()) + .map(str::to_string) }) } diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 953b11b8f8..fcc308f380 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -567,10 +567,10 @@ impl ModelClientSession { Ok(stream) => { return Ok(map_response_stream(stream, self.state.otel_manager.clone())); } - Err(ApiError::Transport(TransportError::Http { status, .. })) - if status == StatusCode::UNAUTHORIZED => - { - handle_unauthorized(status, &mut auth_recovery).await?; + Err(ApiError::Transport( + unauthorized_transport @ TransportError::Http { status, .. }, + )) if status == StatusCode::UNAUTHORIZED => { + handle_unauthorized(unauthorized_transport, &mut auth_recovery).await?; continue; } Err(err) => return Err(map_api_error(err)), @@ -606,10 +606,10 @@ impl ModelClientSession { .await { Ok(connection) => connection, - Err(ApiError::Transport(TransportError::Http { status, .. })) - if status == StatusCode::UNAUTHORIZED => - { - handle_unauthorized(status, &mut auth_recovery).await?; + Err(ApiError::Transport( + unauthorized_transport @ TransportError::Http { status, .. }, + )) if status == StatusCode::UNAUTHORIZED => { + handle_unauthorized(unauthorized_transport, &mut auth_recovery).await?; continue; } Err(err) => return Err(map_api_error(err)), @@ -780,7 +780,7 @@ where /// When refresh succeeds, the caller should retry the API call; otherwise /// the mapped `CodexErr` is returned to the caller. async fn handle_unauthorized( - status: StatusCode, + transport: TransportError, auth_recovery: &mut Option, ) -> Result<()> { if let Some(recovery) = auth_recovery @@ -793,16 +793,7 @@ async fn handle_unauthorized( }; } - Err(map_unauthorized_status(status)) -} - -fn map_unauthorized_status(status: StatusCode) -> CodexErr { - map_api_error(ApiError::Transport(TransportError::Http { - status, - url: None, - headers: None, - body: None, - })) + Err(map_api_error(ApiError::Transport(transport))) } struct ApiTelemetry { diff --git a/codex-rs/core/src/error.rs b/codex-rs/core/src/error.rs index 684d968162..889335824e 100644 --- a/codex-rs/core/src/error.rs +++ b/codex-rs/core/src/error.rs @@ -286,13 +286,42 @@ pub struct UnexpectedResponseError { pub status: StatusCode, pub body: String, pub url: Option, + pub cf_ray: Option, pub request_id: Option, } const CLOUDFLARE_BLOCKED_MESSAGE: &str = "Access blocked by Cloudflare. This usually happens when connecting from a restricted region"; +const UNEXPECTED_RESPONSE_BODY_MAX_BYTES: usize = 1000; impl UnexpectedResponseError { + fn display_body(&self) -> String { + if let Some(message) = self.extract_error_message() { + return message; + } + + let trimmed_body = self.body.trim(); + if trimmed_body.is_empty() { + return "Unknown error".to_string(); + } + + truncate_with_ellipsis(trimmed_body, UNEXPECTED_RESPONSE_BODY_MAX_BYTES) + } + + fn extract_error_message(&self) -> Option { + let json = serde_json::from_str::(&self.body).ok()?; + let message = json + .get("error") + .and_then(|error| error.get("message")) + .and_then(serde_json::Value::as_str)?; + let message = message.trim(); + if message.is_empty() { + None + } else { + Some(message.to_string()) + } + } + fn friendly_message(&self) -> Option { if self.status != StatusCode::FORBIDDEN { return None; @@ -307,6 +336,9 @@ impl UnexpectedResponseError { if let Some(url) = &self.url { message.push_str(&format!(", url: {url}")); } + if let Some(cf_ray) = &self.cf_ray { + message.push_str(&format!(", cf-ray: {cf_ray}")); + } if let Some(id) = &self.request_id { message.push_str(&format!(", request id: {id}")); } @@ -321,11 +353,14 @@ impl std::fmt::Display for UnexpectedResponseError { write!(f, "{friendly}") } else { let status = self.status; - let body = &self.body; + let body = self.display_body(); let mut message = format!("unexpected status {status}: {body}"); if let Some(url) = &self.url { message.push_str(&format!(", url: {url}")); } + if let Some(cf_ray) = &self.cf_ray { + message.push_str(&format!(", cf-ray: {cf_ray}")); + } if let Some(id) = &self.request_id { message.push_str(&format!(", request id: {id}")); } @@ -335,6 +370,21 @@ impl std::fmt::Display for UnexpectedResponseError { } impl std::error::Error for UnexpectedResponseError {} + +fn truncate_with_ellipsis(text: &str, max_bytes: usize) -> String { + if text.len() <= max_bytes { + return text.to_string(); + } + + let mut cut = max_bytes; + while !text.is_char_boundary(cut) { + cut = cut.saturating_sub(1); + } + let mut truncated = text[..cut].to_string(); + truncated.push_str("..."); + truncated +} + #[derive(Debug)] pub struct RetryLimitReachedError { pub status: StatusCode, @@ -952,15 +1002,14 @@ mod tests { body: "Cloudflare error: Sorry, you have been blocked" .to_string(), url: Some("http://example.com/blocked".to_string()), - request_id: Some("ray-id".to_string()), + cf_ray: Some("ray-id".to_string()), + request_id: None, }; let status = StatusCode::FORBIDDEN.to_string(); let url = "http://example.com/blocked"; assert_eq!( err.to_string(), - format!( - "{CLOUDFLARE_BLOCKED_MESSAGE} (status {status}), url: {url}, request id: ray-id" - ) + format!("{CLOUDFLARE_BLOCKED_MESSAGE} (status {status}), url: {url}, cf-ray: ray-id") ); } @@ -970,6 +1019,7 @@ mod tests { status: StatusCode::FORBIDDEN, body: "plain text error".to_string(), url: Some("http://example.com/plain".to_string()), + cf_ray: None, request_id: None, }; let status = StatusCode::FORBIDDEN.to_string(); @@ -980,6 +1030,63 @@ mod tests { ); } + #[test] + fn unexpected_status_prefers_error_message_when_present() { + let err = UnexpectedResponseError { + status: StatusCode::UNAUTHORIZED, + body: r#"{"error":{"message":"Workspace is not authorized in this region."},"status":401}"# + .to_string(), + url: Some("https://chatgpt.com/backend-api/codex/responses".to_string()), + cf_ray: None, + request_id: Some("req-123".to_string()), + }; + let status = StatusCode::UNAUTHORIZED.to_string(); + assert_eq!( + err.to_string(), + format!( + "unexpected status {status}: Workspace is not authorized in this region., url: https://chatgpt.com/backend-api/codex/responses, request id: req-123" + ) + ); + } + + #[test] + fn unexpected_status_truncates_long_body_with_ellipsis() { + let long_body = "x".repeat(UNEXPECTED_RESPONSE_BODY_MAX_BYTES + 10); + let err = UnexpectedResponseError { + status: StatusCode::BAD_GATEWAY, + body: long_body, + url: Some("http://example.com/long".to_string()), + cf_ray: None, + request_id: Some("req-long".to_string()), + }; + let status = StatusCode::BAD_GATEWAY.to_string(); + let expected_body = format!("{}...", "x".repeat(UNEXPECTED_RESPONSE_BODY_MAX_BYTES)); + assert_eq!( + err.to_string(), + format!( + "unexpected status {status}: {expected_body}, url: http://example.com/long, request id: req-long" + ) + ); + } + + #[test] + fn unexpected_status_includes_cf_ray_and_request_id() { + let err = UnexpectedResponseError { + status: StatusCode::UNAUTHORIZED, + body: "plain text error".to_string(), + url: Some("https://chatgpt.com/backend-api/codex/responses".to_string()), + cf_ray: Some("9c81f9f18f2fa49d-LHR".to_string()), + request_id: Some("req-xyz".to_string()), + }; + let status = StatusCode::UNAUTHORIZED.to_string(); + assert_eq!( + err.to_string(), + format!( + "unexpected status {status}: plain text error, url: https://chatgpt.com/backend-api/codex/responses, cf-ray: 9c81f9f18f2fa49d-LHR, request id: req-xyz" + ) + ); + } + #[test] fn usage_limit_reached_includes_hours_and_minutes() { let base = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); From ed778f9017e7df1412dc908fe3859ff316433655 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Tue, 3 Feb 2026 15:34:28 +0000 Subject: [PATCH 3/3] Avoid redundant transactional check before inserting dynamic tools (#10521) Summary - remove the extra transaction guard that checked for existing dynamic tools per thread before inserting new ones - insert each tool record with `ON CONFLICT(thread_id, position) DO NOTHING` to ignore duplicates instead of pre-querying - simplify execution to use the shared pool directly and avoid unneeded commits Testing - Not run (not requested) --- codex-rs/state/src/runtime.rs | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 3b37b6d424..9a750f1d5d 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -418,17 +418,7 @@ ON CONFLICT(id) DO UPDATE SET if tools.is_empty() { return Ok(()); } - let mut tx = self.pool.begin().await?; let thread_id = thread_id.to_string(); - let existing: Option = - sqlx::query_scalar("SELECT 1 FROM thread_dynamic_tools WHERE thread_id = ? LIMIT 1") - .bind(thread_id.as_str()) - .fetch_optional(&mut *tx) - .await?; - if existing.is_some() { - tx.commit().await?; - return Ok(()); - } for (idx, tool) in tools.iter().enumerate() { let position = i64::try_from(idx).unwrap_or(i64::MAX); let input_schema = serde_json::to_string(&tool.input_schema)?; @@ -441,6 +431,7 @@ INSERT INTO thread_dynamic_tools ( description, input_schema ) VALUES (?, ?, ?, ?, ?) +ON CONFLICT(thread_id, position) DO NOTHING "#, ) .bind(thread_id.as_str()) @@ -448,10 +439,9 @@ INSERT INTO thread_dynamic_tools ( .bind(tool.name.as_str()) .bind(tool.description.as_str()) .bind(input_schema) - .execute(&mut *tx) + .execute(self.pool.as_ref()) .await?; } - tx.commit().await?; Ok(()) }