From 09a7bf5b3d5fde89f8a845d5a47d6a525b28e2ae Mon Sep 17 00:00:00 2001 From: Guinness Chen Date: Wed, 8 Jul 2026 15:03:34 -0700 Subject: [PATCH] fix: retry failed OpenAI file uploads --- codex-rs/codex-api/src/files.rs | 108 +++++++++++++++++++----- codex-rs/core/src/mcp_openai_file.rs | 120 ++++++++++++++++++++++++--- 2 files changed, 196 insertions(+), 32 deletions(-) diff --git a/codex-rs/codex-api/src/files.rs b/codex-rs/codex-api/src/files.rs index 3cd3932d0681..d42ad8b03b47 100644 --- a/codex-rs/codex-api/src/files.rs +++ b/codex-rs/codex-api/src/files.rs @@ -1,3 +1,4 @@ +use std::error::Error; use std::time::Duration; use crate::AuthProvider; @@ -37,14 +38,18 @@ pub enum OpenAiFileError { size_bytes: u64, limit_bytes: u64, }, - #[error("failed to send OpenAI file request to {url}: {source}")] + #[error("failed to send OpenAI file {operation} request to {url} ({kind}): {detail}")] Request { + operation: &'static str, url: String, + kind: &'static str, + detail: String, #[source] source: reqwest::Error, }, - #[error("OpenAI file request to {url} failed with status {status}: {body}")] + #[error("OpenAI file {operation} request to {url} failed with status {status}: {body}")] UnexpectedStatus { + operation: &'static str, url: String, status: StatusCode, body: String, @@ -61,6 +66,21 @@ pub enum OpenAiFileError { UploadFailed { file_id: String, message: String }, } +impl OpenAiFileError { + pub fn is_retryable_upload_failure(&self) -> bool { + matches!( + self, + Self::Request { + operation: "upload", + .. + } | Self::UnexpectedStatus { + operation: "upload", + .. + } + ) + } +} + #[derive(Deserialize)] struct CreateFileResponse { file_id: String, @@ -105,22 +125,20 @@ pub async fn upload_openai_file( })) .send() .await - .map_err(|source| OpenAiFileError::Request { - url: create_url.clone(), - source, - })?; + .map_err(|source| request_error("create", &create_url, source))?; let create_status = create_response.status(); let create_body = create_response.text().await.unwrap_or_default(); if !create_status.is_success() { return Err(OpenAiFileError::UnexpectedStatus { - url: create_url, + operation: "create", + url: redacted_url(&create_url), status: create_status, body: create_body, }); } let create_payload: CreateFileResponse = serde_json::from_str(&create_body).map_err(|source| OpenAiFileError::Decode { - url: create_url.clone(), + url: redacted_url(&create_url), source, })?; @@ -132,15 +150,13 @@ pub async fn upload_openai_file( .body(reqwest::Body::wrap_stream(contents)) .send() .await - .map_err(|source| OpenAiFileError::Request { - url: create_payload.upload_url.clone(), - source, - })?; + .map_err(|source| request_error("upload", &create_payload.upload_url, source))?; let upload_status = upload_response.status(); let upload_body = upload_response.text().await.unwrap_or_default(); if !upload_status.is_success() { return Err(OpenAiFileError::UnexpectedStatus { - url: create_payload.upload_url.clone(), + operation: "upload", + url: redacted_url(&create_payload.upload_url), status: upload_status, body: upload_body, }); @@ -157,22 +173,20 @@ pub async fn upload_openai_file( .json(&serde_json::json!({})) .send() .await - .map_err(|source| OpenAiFileError::Request { - url: finalize_url.clone(), - source, - })?; + .map_err(|source| request_error("finalize", &finalize_url, source))?; let finalize_status = finalize_response.status(); let finalize_body = finalize_response.text().await.unwrap_or_default(); if !finalize_status.is_success() { return Err(OpenAiFileError::UnexpectedStatus { - url: finalize_url.clone(), + operation: "finalize", + url: redacted_url(&finalize_url), status: finalize_status, body: finalize_body, }); } let finalize_payload: DownloadLinkResponse = serde_json::from_str(&finalize_body).map_err(|source| OpenAiFileError::Decode { - url: finalize_url.clone(), + url: redacted_url(&finalize_url), source, })?; @@ -212,6 +226,54 @@ pub async fn upload_openai_file( } } +fn request_error(operation: &'static str, url: &str, source: reqwest::Error) -> OpenAiFileError { + let kind = if source.is_timeout() { + "timeout" + } else if source.is_connect() { + "connect" + } else if source.is_body() { + "body" + } else if source.is_request() { + "request" + } else { + "transport" + }; + let mut details = Vec::new(); + let mut cause = source.source(); + while let Some(current) = cause { + let message = current.to_string(); + if details.last() != Some(&message) { + details.push(message); + } + cause = current.source(); + } + let detail = if details.is_empty() { + kind.to_string() + } else { + details.join(": ") + }; + + OpenAiFileError::Request { + operation, + url: redacted_url(url), + kind, + detail, + source, + } +} + +fn redacted_url(url: &str) -> String { + let Ok(mut url) = reqwest::Url::parse(url) else { + return url + .split_once('?') + .map_or(url, |(base, _)| base) + .to_string(); + }; + url.set_query(None); + url.set_fragment(None); + url.to_string() +} + fn authorized_request( auth: &dyn AuthProvider, method: reqwest::Method, @@ -272,6 +334,14 @@ mod tests { format!("{}/backend-api", server.uri()) } + #[test] + fn redacted_url_removes_presigned_query() { + assert_eq!( + redacted_url("https://files.example/upload/raw?se=soon&sig=secret"), + "https://files.example/upload/raw" + ); + } + #[tokio::test] async fn upload_openai_file_returns_canonical_uri() { let server = MockServer::start().await; diff --git a/codex-rs/core/src/mcp_openai_file.rs b/codex-rs/core/src/mcp_openai_file.rs index 5044568cd62a..1ac30990b8a9 100644 --- a/codex-rs/core/src/mcp_openai_file.rs +++ b/codex-rs/core/src/mcp_openai_file.rs @@ -19,6 +19,8 @@ use codex_login::CodexAuth; use codex_utils_path_uri::PathUri; use serde_json::Value as JsonValue; +const OPENAI_FILE_UPLOAD_ATTEMPTS: usize = 2; + pub(crate) async fn rewrite_mcp_tool_arguments_for_openai_files( sess: &Session, turn_context: &TurnContext, @@ -149,25 +151,41 @@ async fn build_uploaded_argument_value( OPENAI_FILE_UPLOAD_LIMIT_BYTES, ))); } - let contents = fs - .read_file_stream(&path_uri, /*sandbox*/ None) - .await - .map_err(|error| contextualize_error(error.to_string()))?; let file_name = resolved_path .file_name() .and_then(|value| value.to_str()) .unwrap_or("file") .to_string(); let upload_auth = codex_model_provider::auth_provider_from_auth(auth); - let uploaded = upload_openai_file( - turn_context.config.chatgpt_base_url.trim_end_matches('/'), - upload_auth.as_ref(), - file_name, - metadata.size, - contents, - ) - .await - .map_err(|error| contextualize_error(error.to_string()))?; + let mut attempt = 0; + let uploaded = loop { + attempt += 1; + let contents = fs + .read_file_stream(&path_uri, /*sandbox*/ None) + .await + .map_err(|error| contextualize_error(error.to_string()))?; + match upload_openai_file( + turn_context.config.chatgpt_base_url.trim_end_matches('/'), + upload_auth.as_ref(), + file_name.clone(), + metadata.size, + contents, + ) + .await + { + Ok(uploaded) => break uploaded, + Err(error) + if attempt < OPENAI_FILE_UPLOAD_ATTEMPTS && error.is_retryable_upload_failure() => + { + tracing::warn!( + attempt, + error = %error, + "retrying OpenAI file upload with a fresh URL" + ); + } + Err(error) => return Err(contextualize_error(error.to_string())), + } + }; Ok(serde_json::json!({ "download_url": uploaded.download_url, "file_id": uploaded.file_id, @@ -306,6 +324,82 @@ mod tests { ); } + #[tokio::test] + async fn build_uploaded_argument_value_retries_upload_with_fresh_url() { + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::Request; + use wiremock::ResponseTemplate; + use wiremock::matchers::method; + use wiremock::matchers::path; + + let server = MockServer::start().await; + let create_attempts = Arc::new(AtomicUsize::new(0)); + let responder_attempts = Arc::clone(&create_attempts); + let server_uri = server.uri(); + Mock::given(method("POST")) + .and(path("/backend-api/files")) + .respond_with(move |_request: &Request| { + if responder_attempts.fetch_add(1, Ordering::SeqCst) == 0 { + return ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "file_id": "file_1", + "upload_url": "http://127.0.0.1:1/upload/file_1?sig=secret", + })); + } + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "file_id": "file_2", + "upload_url": format!("{server_uri}/upload/file_2"), + })) + }) + .expect(2) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path("/upload/file_2")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/backend-api/files/file_2/uploaded")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "status": "success", + "download_url": format!("{}/download/file_2", server.uri()), + "file_name": "file_report.csv", + "mime_type": "text/csv", + }))) + .expect(1) + .mount(&server) + .await; + + let (_, mut turn_context) = make_session_and_context().await; + let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); + let dir = tempdir().expect("temp dir"); + tokio::fs::write(dir.path().join("file_report.csv"), b"hello") + .await + .expect("write local file"); + set_primary_environment_cwd(&mut turn_context, dir.path()); + + let mut config = (*turn_context.config).clone(); + config.chatgpt_base_url = format!("{}/backend-api", server.uri()); + turn_context.config = Arc::new(config); + + let rewritten = build_uploaded_argument_value( + &turn_context, + Some(&auth), + "file", + /*index*/ None, + "file_report.csv", + ) + .await + .expect("second upload attempt should succeed"); + + assert_eq!(rewritten["file_id"], "file_2"); + assert_eq!(create_attempts.load(Ordering::SeqCst), 2); + } + #[tokio::test] async fn build_uploaded_argument_value_rejects_oversized_file_before_reading() { let (_, mut turn_context) = make_session_and_context().await;