openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-16 12:13:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into gt/default-val

Browse Source
This commit is contained in:
gt-oai
2026-02-03 15:47:40 +00:00
committed by GitHub
parent c312b807d2 ed778f9017
commit 1150449df2
12 changed files with 350 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
codex-rs/codex-api/src/auth.rs
View File

@@ -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<A: AuthProvider>(auth: &A, mut req: Request) -> Request {
pub(crate) fn add_auth_headers_to_header_map<A: AuthProvider>(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<A: AuthProvider>(auth: &A, mut req: Request) -> Request {
add_auth_headers_to_header_map(auth, &mut req.headers);
req
}

67
codex-rs/codex-api/src/endpoint/compact.rs
View File

@@ -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<T: HttpTransport, A: AuthProvider> {
transport: T,
provider: Provider,
auth: A,
request_telemetry: Option<Arc<dyn RequestTelemetry>>,
session: EndpointSession<T, A>,
}
impl<T: HttpTransport, A: AuthProvider> CompactClient<T, A> {
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<Arc<dyn RequestTelemetry>>) -> Self {
self.request_telemetry = request;
self
pub fn with_telemetry(self, request: Option<Arc<dyn RequestTelemetry>>) -> 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<T: HttpTransport, A: AuthProvider> CompactClient<T, A> {
body: serde_json::Value,
extra_headers: HeaderMap,
) -> Result<Vec<ResponseItem>, 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::<DummyTransport, DummyAuth>::path(),
"responses/compact"
);
}
}

2
codex-rs/codex-api/src/endpoint/mod.rs
View File

@@ -3,4 +3,4 @@ pub mod compact;
pub mod models;
pub mod responses;
pub mod responses_websocket;
mod streaming;
mod session;

50
codex-rs/codex-api/src/endpoint/models.rs
View File

@@ -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<T: HttpTransport, A: AuthProvider> {
transport: T,
provider: Provider,
auth: A,
request_telemetry: Option<Arc<dyn RequestTelemetry>>,
session: EndpointSession<T, A>,
}
impl<T: HttpTransport, A: AuthProvider> ModelsClient<T, A> {
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<Arc<dyn RequestTelemetry>>) -> Self {
self.request_telemetry = request;
self
pub fn with_telemetry(self, request: Option<Arc<dyn RequestTelemetry>>) -> 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<ModelInfo>, Option<String>), 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

48
codex-rs/codex-api/src/endpoint/responses.rs
View File

@@ -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<T: HttpTransport, A: AuthProvider> {
streaming: StreamingClient<T, A>,
session: EndpointSession<T, A>,
sse_telemetry: Option<Arc<dyn SseTelemetry>>,
}
#[derive(Default)]
@@ -42,7 +45,8 @@ pub struct ResponsesOptions {
impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
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<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
sse: Option<Arc<dyn SseTelemetry>>,
) -> Self {
Self {
streaming: self.streaming.with_telemetry(request, sse),
session: self.session.with_request_telemetry(request),
sse_telemetry: sse,
}
}
@@ -102,12 +107,12 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
.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<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
compression: Compression,
turn_state: Option<Arc<OnceLock<String>>>,
) -> Result<ResponseStream, ApiError> {
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,
))
}
}

18
codex-rs/codex-api/src/endpoint/responses_websocket.rs
View File

@@ -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<A: AuthProvider> ResponsesWebsocketClient<A> {
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<A: AuthProvider> ResponsesWebsocketClient<A> {
}
}
// 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,

126
codex-rs/codex-api/src/endpoint/session.rs Normal file
View File

@@ -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<T: HttpTransport, A: AuthProvider> {
transport: T,
provider: Provider,
auth: A,
request_telemetry: Option<Arc<dyn RequestTelemetry>>,
}
impl<T: HttpTransport, A: AuthProvider> EndpointSession<T, A> {
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<Arc<dyn RequestTelemetry>>,
) -> 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<Value>,
) -> Result<Response, ApiError> {
self.execute_with(method, path, extra_headers, body, |_| {})
.await
}
pub(crate) async fn execute_with<C>(
&self,
method: Method,
path: &str,
extra_headers: HeaderMap,
body: Option<Value>,
configure: C,
) -> Result<Response, ApiError>
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<C>(
&self,
method: Method,
path: &str,
extra_headers: HeaderMap,
body: Option<Value>,
configure: C,
) -> Result<StreamResponse, ApiError>
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)
}
}

95
codex-rs/codex-api/src/endpoint/streaming.rs
View File

@@ -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<T: HttpTransport, A: AuthProvider> {
transport: T,
provider: Provider,
auth: A,
request_telemetry: Option<Arc<dyn RequestTelemetry>>,
sse_telemetry: Option<Arc<dyn SseTelemetry>>,
}
type StreamSpawner = fn(
StreamResponse,
Duration,
Option<Arc<dyn SseTelemetry>>,
Option<Arc<OnceLock<String>>>,
) -> ResponseStream;
impl<T: HttpTransport, A: AuthProvider> StreamingClient<T, A> {
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<Arc<dyn RequestTelemetry>>,
sse: Option<Arc<dyn SseTelemetry>>,
) -> 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<Arc<OnceLock<String>>>,
) -> Result<ResponseStream, ApiError> {
let builder = || {
let mut req = self.provider.build_request(Method::POST, path);
req.headers.extend(extra_headers.clone());
req.headers.insert(
http::header::ACCEPT,
http::HeaderValue::from_static("text/event-stream"),
);
req.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,
))
}
}

26
codex-rs/core/src/api_bridge.rs
View File

@@ -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<String> {
extract_request_id(headers).or_else(|| extract_header(headers, CF_RAY_HEADER))
}
fn extract_request_id(headers: Option<&HeaderMap>) -> Option<String> {
extract_header(headers, REQUEST_ID_HEADER)
.or_else(|| extract_header(headers, OAI_REQUEST_ID_HEADER))
}
fn extract_header(headers: Option<&HeaderMap>, name: &str) -> Option<String> {
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)
})
}

29
codex-rs/core/src/client.rs
View File

@@ -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<UnauthorizedRecovery>,
) -> 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 {

117
codex-rs/core/src/error.rs
View File

@@ -286,13 +286,42 @@ pub struct UnexpectedResponseError {
pub status: StatusCode,
pub body: String,
pub url: Option<String>,
pub cf_ray: Option<String>,
pub request_id: Option<String>,
}
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<String> {
let json = serde_json::from_str::<serde_json::Value>(&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<String> {
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: "<html><body>Cloudflare error: Sorry, you have been blocked</body></html>"
.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();

14
codex-rs/state/src/runtime.rs
View File

@@ -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<i64> =
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(())
}