Compare commits

...

1 Commits

Author SHA1 Message Date
Liang-Ting Jiang
ec9bbef348 Add OpenAI file download materialization and library upload 2026-04-23 15:00:00 -07:00
10 changed files with 971 additions and 77 deletions

View File

@@ -7,9 +7,11 @@ use codex_client::build_reqwest_client_with_custom_ca;
use reqwest::StatusCode;
use reqwest::header::CONTENT_LENGTH;
use serde::Deserialize;
use serde::de::DeserializeOwned;
use tokio::fs::File;
use tokio::time::Instant;
use tokio_util::io::ReaderStream;
use url::Url;
pub const OPENAI_FILE_URI_PREFIX: &str = "sediment://";
pub const OPENAI_FILE_UPLOAD_LIMIT_BYTES: u64 = 512 * 1024 * 1024;
@@ -18,16 +20,24 @@ const OPENAI_FILE_REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
const OPENAI_FILE_FINALIZE_TIMEOUT: Duration = Duration::from_secs(30);
const OPENAI_FILE_FINALIZE_RETRY_DELAY: Duration = Duration::from_millis(250);
const OPENAI_FILE_USE_CASE: &str = "codex";
const OPENAI_LIBRARY_FILE_USE_CASE: &str = "my_files";
const OPENAI_LIBRARY_ENTRY_SURFACE: &str = "codex_apps";
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct OpenAiFileUploadOptions {
pub store_in_library: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UploadedOpenAiFile {
pub file_id: String,
pub uri: String,
pub download_url: String,
pub download_url: Option<String>,
pub file_name: String,
pub file_size_bytes: u64,
pub mime_type: Option<String>,
pub path: PathBuf,
pub library_file_id: Option<String>,
}
#[derive(Debug, thiserror::Error)]
@@ -68,6 +78,12 @@ pub enum OpenAiFileError {
#[source]
source: serde_json::Error,
},
#[error("failed to resolve OpenAI file URL `{url}`: {source}")]
InvalidUrl {
url: String,
#[source]
source: url::ParseError,
},
#[error("OpenAI file upload for `{file_id}` is not ready yet")]
UploadNotReady { file_id: String },
#[error("OpenAI file upload for `{file_id}` failed: {message}")]
@@ -90,14 +106,114 @@ struct DownloadLinkResponse {
error_message: Option<String>,
}
#[derive(Deserialize)]
struct ProcessUploadStreamStatus {
event: Option<String>,
message: Option<String>,
#[serde(default)]
extra: Option<ProcessUploadStreamExtra>,
}
#[derive(Debug, Default, Deserialize)]
struct ProcessUploadStreamExtra {
#[serde(alias = "metadata_object_id", alias = "library_file_id")]
library_file_id: Option<String>,
#[serde(alias = "library_file_name")]
file_name: Option<String>,
mime_type: Option<String>,
}
pub fn openai_file_uri(file_id: &str) -> String {
format!("{OPENAI_FILE_URI_PREFIX}{file_id}")
}
fn openai_file_api_base_url(base_url: &str) -> String {
let base_url = base_url.trim_end_matches('/');
if let Some(root) = base_url.strip_suffix("/backend-api/codex") {
return format!("{root}/backend-api");
}
if let Some(root) = base_url.strip_suffix("/api/codex") {
return format!("{root}/backend-api");
}
if base_url.ends_with("/backend-api") {
return base_url.to_string();
}
format!("{base_url}/backend-api")
}
fn is_localhost_url(url: &Url) -> bool {
matches!(url.host_str(), Some("localhost" | "127.0.0.1" | "::1"))
}
fn local_dev_same_origin_api_base_url(base_url: &str) -> Option<String> {
let mut url = Url::parse(base_url).ok()?;
if !is_localhost_url(&url) {
return None;
}
url.set_path("/api");
url.set_query(None);
url.set_fragment(None);
Some(url.to_string().trim_end_matches('/').to_string())
}
fn local_dev_sa_server_api_base_url(base_url: &str) -> Option<String> {
let mut url = Url::parse(base_url).ok()?;
if !is_localhost_url(&url) {
return None;
}
if url.port_or_known_default() == Some(8000) {
return None;
}
url.set_port(Some(8000)).ok()?;
url.set_path("/api");
url.set_query(None);
url.set_fragment(None);
Some(url.to_string().trim_end_matches('/').to_string())
}
fn openai_file_api_base_url_candidates(base_url: &str) -> Vec<String> {
let mut candidates = vec![openai_file_api_base_url(base_url)];
if let Some(candidate) = local_dev_same_origin_api_base_url(base_url)
&& !candidates.contains(&candidate)
{
candidates.push(candidate);
}
if let Some(candidate) = local_dev_sa_server_api_base_url(base_url)
&& !candidates.contains(&candidate)
{
candidates.push(candidate);
}
candidates
}
pub async fn download_openai_file(
base_url: &str,
auth: &impl AuthProvider,
download_url: &str,
) -> Result<Vec<u8>, OpenAiFileError> {
let resolved_url = resolve_openai_file_download_url(base_url, download_url)?;
let request_builder = if should_attach_auth_to_openai_file_url(&resolved_url, base_url) {
authorized_request(auth, reqwest::Method::GET, resolved_url.as_str())
} else {
build_reqwest_client()
.request(reqwest::Method::GET, resolved_url.as_str())
.timeout(OPENAI_FILE_REQUEST_TIMEOUT)
};
let response = request_builder
.send()
.await
.map_err(|source| OpenAiFileError::Request {
url: resolved_url.to_string(),
source,
})?;
response_bytes(resolved_url.as_str(), response).await
}
pub async fn upload_local_file(
base_url: &str,
auth: &dyn AuthProvider,
path: &Path,
options: &OpenAiFileUploadOptions,
) -> Result<UploadedOpenAiFile, OpenAiFileError> {
let metadata = tokio::fs::metadata(path)
.await
@@ -128,33 +244,40 @@ pub async fn upload_local_file(
.and_then(|value| value.to_str())
.unwrap_or("file")
.to_string();
let create_url = format!("{}/files", base_url.trim_end_matches('/'));
let create_response = authorized_request(auth, reqwest::Method::POST, &create_url)
.json(&serde_json::json!({
"file_name": file_name,
"file_size": metadata.len(),
"use_case": OPENAI_FILE_USE_CASE,
}))
.send()
.await
.map_err(|source| OpenAiFileError::Request {
url: create_url.clone(),
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,
status: create_status,
body: create_body,
});
let mut create_request = serde_json::json!({
"file_name": file_name,
"file_size": metadata.len(),
"use_case": OPENAI_FILE_USE_CASE,
});
if options.store_in_library {
create_request["store_in_library"] = serde_json::json!(true);
}
let create_payload: CreateFileResponse =
serde_json::from_str(&create_body).map_err(|source| OpenAiFileError::Decode {
url: create_url.clone(),
source,
})?;
let mut last_not_found_error = None;
let mut api_base_url = openai_file_api_base_url(base_url);
let mut create_payload = None;
for candidate in openai_file_api_base_url_candidates(base_url) {
match create_file(auth, &candidate, &create_request).await {
Ok(payload) => {
api_base_url = candidate;
create_payload = Some(payload);
break;
}
Err(error @ OpenAiFileError::UnexpectedStatus { status, .. })
if status == StatusCode::NOT_FOUND =>
{
last_not_found_error = Some(error);
}
Err(error) => return Err(error),
}
}
let Some(create_payload) = create_payload else {
return Err(
last_not_found_error.unwrap_or(OpenAiFileError::UploadFailed {
file_id: "unknown".to_string(),
message: "file creation did not produce a response".to_string(),
}),
);
};
let upload_file = File::open(path)
.await
@@ -174,21 +297,31 @@ pub async fn upload_local_file(
url: create_payload.upload_url.clone(),
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(),
status: upload_status,
body: upload_body,
response_text(&create_payload.upload_url, upload_response).await?;
if options.store_in_library {
let processed =
process_upload_stream(auth, &api_base_url, &create_payload.file_id, &file_name).await?;
let library_file_id =
processed
.library_file_id
.ok_or_else(|| OpenAiFileError::UploadFailed {
file_id: create_payload.file_id.clone(),
message: "upload completed without creating a library_file_id".to_string(),
})?;
return Ok(UploadedOpenAiFile {
file_id: create_payload.file_id.clone(),
uri: openai_file_uri(&create_payload.file_id),
download_url: None,
file_name: processed.file_name.unwrap_or(file_name),
file_size_bytes: metadata.len(),
mime_type: processed.mime_type,
path: path.to_path_buf(),
library_file_id: Some(library_file_id),
});
}
let finalize_url = format!(
"{}/files/{}/uploaded",
base_url.trim_end_matches('/'),
create_payload.file_id,
);
let finalize_url = format!("{api_base_url}/files/{}/uploaded", create_payload.file_id);
let finalize_started_at = Instant::now();
loop {
let finalize_response = authorized_request(auth, reqwest::Method::POST, &finalize_url)
@@ -199,36 +332,25 @@ pub async fn upload_local_file(
url: finalize_url.clone(),
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(),
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(),
source,
})?;
response_json(&finalize_url, finalize_response).await?;
match finalize_payload.status.as_str() {
"success" => {
return Ok(UploadedOpenAiFile {
file_id: create_payload.file_id.clone(),
uri: openai_file_uri(&create_payload.file_id),
download_url: finalize_payload.download_url.ok_or_else(|| {
download_url: Some(finalize_payload.download_url.ok_or_else(|| {
OpenAiFileError::UploadFailed {
file_id: create_payload.file_id.clone(),
message: "missing download_url".to_string(),
}
})?,
})?),
file_name: finalize_payload.file_name.unwrap_or(file_name),
file_size_bytes: metadata.len(),
mime_type: finalize_payload.mime_type,
path: path.to_path_buf(),
library_file_id: None,
});
}
"retry" => {
@@ -251,6 +373,102 @@ pub async fn upload_local_file(
}
}
async fn create_file(
auth: &dyn AuthProvider,
api_base_url: &str,
create_request: &serde_json::Value,
) -> Result<CreateFileResponse, OpenAiFileError> {
let create_url = format!("{api_base_url}/files");
let create_response = authorized_request(auth, reqwest::Method::POST, &create_url)
.json(create_request)
.send()
.await
.map_err(|source| OpenAiFileError::Request {
url: create_url.clone(),
source,
})?;
response_json(&create_url, create_response).await
}
async fn process_upload_stream(
auth: &dyn AuthProvider,
api_base_url: &str,
file_id: &str,
file_name: &str,
) -> Result<ProcessUploadStreamExtra, OpenAiFileError> {
let process_url = format!("{api_base_url}/files/process_upload_stream");
let process_response = authorized_request(auth, reqwest::Method::POST, &process_url)
.json(&serde_json::json!({
"file_id": file_id,
"file_name": file_name,
"use_case": OPENAI_LIBRARY_FILE_USE_CASE,
"index_for_retrieval": false,
"entry_surface": OPENAI_LIBRARY_ENTRY_SURFACE,
"metadata": {
"store_in_library": true,
},
}))
.send()
.await
.map_err(|source| OpenAiFileError::Request {
url: process_url.clone(),
source,
})?;
let process_body = response_text(&process_url, process_response).await?;
let mut result = ProcessUploadStreamExtra::default();
for line in process_body.lines() {
let line = line.trim();
if line.is_empty() {
continue;
}
let ProcessUploadStreamStatus {
event,
message,
extra,
} = serde_json::from_str(line).map_err(|source| OpenAiFileError::Decode {
url: process_url.clone(),
source,
})?;
let extra = extra.unwrap_or_default();
if let Some(event) = event.as_deref()
&& is_process_upload_stream_error_event(event)
{
return Err(OpenAiFileError::UploadFailed {
file_id: file_id.to_string(),
message: message
.filter(|message| !message.is_empty())
.unwrap_or_else(|| format!("process_upload_stream returned {event}")),
});
}
if result.library_file_id.is_none() {
result.library_file_id = non_empty_string(extra.library_file_id);
}
if result.file_name.is_none() {
result.file_name = non_empty_string(extra.file_name);
}
if result.mime_type.is_none() {
result.mime_type = non_empty_string(extra.mime_type);
}
}
Ok(result)
}
fn non_empty_string(value: Option<String>) -> Option<String> {
value.filter(|value| !value.is_empty())
}
fn is_process_upload_stream_error_event(event: &str) -> bool {
let event_tail = event
.rsplit(|ch| ch == '.' || ch == '_')
.next()
.unwrap_or(event);
matches!(event_tail, "error" | "cancelled" | "unknown")
}
fn authorized_request(
auth: &dyn AuthProvider,
method: reqwest::Method,
@@ -266,6 +484,93 @@ fn authorized_request(
.headers(headers)
}
async fn response_text(url: &str, response: reqwest::Response) -> Result<String, OpenAiFileError> {
let status = response.status();
let body = response.text().await.unwrap_or_default();
if !status.is_success() {
return Err(OpenAiFileError::UnexpectedStatus {
url: url.to_string(),
status,
body,
});
}
Ok(body)
}
async fn response_json<T: DeserializeOwned>(
url: &str,
response: reqwest::Response,
) -> Result<T, OpenAiFileError> {
let body = response_text(url, response).await?;
serde_json::from_str(&body).map_err(|source| OpenAiFileError::Decode {
url: url.to_string(),
source,
})
}
async fn response_bytes(
url: &str,
response: reqwest::Response,
) -> Result<Vec<u8>, OpenAiFileError> {
let status = response.status();
if !status.is_success() {
let body = response.text().await.unwrap_or_default();
return Err(OpenAiFileError::UnexpectedStatus {
url: url.to_string(),
status,
body,
});
}
let bytes = response
.bytes()
.await
.map_err(|source| OpenAiFileError::Request {
url: url.to_string(),
source,
})?;
Ok(bytes.to_vec())
}
fn resolve_openai_file_download_url(
base_url: &str,
download_url: &str,
) -> Result<Url, OpenAiFileError> {
match Url::parse(download_url) {
Ok(url) => Ok(url),
Err(url::ParseError::RelativeUrlWithoutBase) => {
let normalized_base_url = if base_url.ends_with('/') {
base_url.to_string()
} else {
format!("{base_url}/")
};
let base =
Url::parse(&normalized_base_url).map_err(|source| OpenAiFileError::InvalidUrl {
url: normalized_base_url.clone(),
source,
})?;
base.join(download_url)
.map_err(|source| OpenAiFileError::InvalidUrl {
url: download_url.to_string(),
source,
})
}
Err(source) => Err(OpenAiFileError::InvalidUrl {
url: download_url.to_string(),
source,
}),
}
}
fn should_attach_auth_to_openai_file_url(download_url: &Url, base_url: &str) -> bool {
let Ok(base_url) = Url::parse(base_url) else {
return false;
};
match (download_url.host_str(), base_url.host_str()) {
(Some(download_host), Some(base_host)) => download_host.eq_ignore_ascii_case(base_host),
_ => false,
}
}
fn build_reqwest_client() -> reqwest::Client {
build_reqwest_client_with_custom_ca(reqwest::Client::builder()).unwrap_or_else(|error| {
tracing::warn!(error = %error, "failed to build OpenAI file upload client");
@@ -309,7 +614,45 @@ mod tests {
}
fn base_url_for(server: &MockServer) -> String {
format!("{}/backend-api", server.uri())
server.uri()
}
#[test]
fn openai_file_api_base_url_candidates_include_local_dev_fallbacks() {
assert_eq!(
openai_file_api_base_url_candidates("http://localhost:8061"),
vec![
"http://localhost:8061/backend-api".to_string(),
"http://localhost:8061/api".to_string(),
"http://localhost:8000/api".to_string(),
]
);
}
#[tokio::test]
async fn download_openai_file_resolves_relative_url_and_attaches_auth() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/files/download/file_123"))
.and(header("authorization", "Bearer token"))
.and(header("chatgpt-account-id", "account_id"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("content-type", "text/plain")
.set_body_bytes(b"hello".to_vec()),
)
.mount(&server)
.await;
let downloaded = download_openai_file(
&format!("{}/backend-api/codex", server.uri()),
&chatgpt_auth(),
"/files/download/file_123",
)
.await
.expect("download succeeds");
assert_eq!(downloaded, b"hello".to_vec());
}
#[tokio::test]
@@ -363,18 +706,98 @@ mod tests {
let path = dir.path().join("hello.txt");
tokio::fs::write(&path, b"hello").await.expect("write file");
let uploaded = upload_local_file(&base_url, &chatgpt_auth(), &path)
.await
.expect("upload succeeds");
let uploaded = upload_local_file(
&base_url,
&chatgpt_auth(),
&path,
&OpenAiFileUploadOptions::default(),
)
.await
.expect("upload succeeds");
assert_eq!(uploaded.file_id, "file_123");
assert_eq!(uploaded.uri, "sediment://file_123");
assert_eq!(
uploaded.download_url,
format!("{}/download/file_123", server.uri())
Some(format!("{}/download/file_123", server.uri()))
);
assert_eq!(uploaded.file_name, "hello.txt");
assert_eq!(uploaded.mime_type, Some("text/plain".to_string()));
assert_eq!(finalize_attempts.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn upload_local_file_stores_library_file_with_process_upload_stream() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/backend-api/files"))
.and(header("chatgpt-account-id", "account_id"))
.and(body_json(serde_json::json!({
"file_name": "hello.txt",
"file_size": 5,
"use_case": "codex",
"store_in_library": true,
})))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"file_id": "file_123", "upload_url": format!("{}/upload/file_123", server.uri())})),
)
.mount(&server)
.await;
Mock::given(method("PUT"))
.and(path("/upload/file_123"))
.and(header("content-length", "5"))
.respond_with(ResponseTemplate::new(200))
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/backend-api/files/process_upload_stream"))
.and(body_json(serde_json::json!({
"file_id": "file_123",
"file_name": "hello.txt",
"use_case": "my_files",
"index_for_retrieval": false,
"entry_surface": "codex_apps",
"metadata": {
"store_in_library": true,
},
})))
.respond_with(ResponseTemplate::new(200).set_body_bytes(
concat!(
"{\"file_id\":\"file_123\",\"event\":\"indexing.completed\",\"message\":\"\",",
"\"extra\":{\"metadata_object_id\":\"library_123\",",
"\"library_file_name\":\"hello.txt\",\"mime_type\":\"text/plain\"}}\n",
"{\"file_id\":\"file_123\",\"event\":\"completed\",",
"\"message\":\"Succeeded processing file file_123\",",
"\"progress\":100,\"extra\":null}\n",
)
.as_bytes()
.to_vec(),
))
.mount(&server)
.await;
let base_url = base_url_for(&server);
let dir = TempDir::new().expect("temp dir");
let path = dir.path().join("hello.txt");
tokio::fs::write(&path, b"hello").await.expect("write file");
let uploaded = upload_local_file(
&base_url,
&chatgpt_auth(),
&path,
&OpenAiFileUploadOptions {
store_in_library: true,
},
)
.await
.expect("upload succeeds");
assert_eq!(uploaded.file_id, "file_123");
assert_eq!(uploaded.uri, "sediment://file_123");
assert_eq!(uploaded.download_url, None);
assert_eq!(uploaded.file_name, "hello.txt");
assert_eq!(uploaded.mime_type, Some("text/plain".to_string()));
assert_eq!(uploaded.library_file_id, Some("library_123".to_string()));
}
}

View File

@@ -57,6 +57,8 @@ pub use crate::endpoint::ResponsesWebsocketClient;
pub use crate::endpoint::ResponsesWebsocketConnection;
pub use crate::endpoint::session_update_session_json;
pub use crate::error::ApiError;
pub use crate::files::OpenAiFileUploadOptions;
pub use crate::files::download_openai_file;
pub use crate::files::upload_local_file;
pub use crate::provider::Provider;
pub use crate::provider::RetryConfig;

View File

@@ -543,14 +543,25 @@ fn codex_apps_tools_cache_filters_disallowed_connectors() {
"calendar",
Some("Calendar"),
),
create_test_tool_with_connector(
CODEX_APPS_MCP_SERVER_NAME,
"allowed_openai_library_tool",
"connector_openai_library",
Some("Library"),
),
];
write_cached_codex_apps_tools(&cache_context, &tools);
let cached = read_cached_codex_apps_tools(&cache_context).expect("cache entry exists for user");
assert_eq!(cached.len(), 1);
assert_eq!(cached.len(), 2);
assert_eq!(cached[0].callable_name, "allowed_tool");
assert_eq!(cached[0].connector_id.as_deref(), Some("calendar"));
assert_eq!(cached[1].callable_name, "allowed_openai_library_tool");
assert_eq!(
cached[1].connector_id.as_deref(),
Some("connector_openai_library")
);
}
#[test]

View File

@@ -0,0 +1,341 @@
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use codex_api::download_openai_file;
use codex_login::CodexAuth;
use codex_model_provider::BearerAuthProvider;
use codex_protocol::mcp::CallToolResult;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Map as JsonMap;
use serde_json::Value as JsonValue;
use tracing::warn;
const CODEX_APPS_FILE_DOWNLOAD_ARTIFACTS_DIR: &str = ".tmp/codex_apps_downloads";
const CODEX_APPS_PROVIDER_BUILTIN: &str = "builtin";
const CODEX_APPS_META_PROVIDER_KEY: &str = "provider";
const CODEX_APPS_META_MATERIALIZE_FILE_DOWNLOAD_KEY: &str = "materialize_file_download";
#[derive(Debug, Deserialize, Serialize)]
struct CodexAppsFileDownloadPayload {
file_id: String,
#[serde(default)]
file_name: Option<String>,
file_uri: CodexAppsFileUri,
}
#[derive(Debug, Deserialize, Serialize)]
struct CodexAppsFileUri {
download_url: String,
#[serde(default)]
file_name: Option<String>,
}
fn codex_apps_download_base_url(turn_context: &TurnContext) -> &str {
turn_context.config.chatgpt_base_url.as_str()
}
fn should_materialize_codex_apps_file_download(
server: &str,
codex_apps_meta: Option<&JsonMap<String, JsonValue>>,
) -> bool {
if server != codex_mcp::CODEX_APPS_MCP_SERVER_NAME {
return false;
}
let Some(codex_apps_meta) = codex_apps_meta else {
return false;
};
codex_apps_meta
.get(CODEX_APPS_META_PROVIDER_KEY)
.and_then(JsonValue::as_str)
== Some(CODEX_APPS_PROVIDER_BUILTIN)
&& codex_apps_meta
.get(CODEX_APPS_META_MATERIALIZE_FILE_DOWNLOAD_KEY)
.and_then(JsonValue::as_bool)
== Some(true)
}
pub(crate) async fn maybe_materialize_codex_apps_file_download_result(
sess: &Session,
turn_context: &TurnContext,
server: &str,
codex_apps_meta: Option<&JsonMap<String, JsonValue>>,
result: CallToolResult,
) -> CallToolResult {
let auth = sess.services.auth_manager.auth().await;
maybe_materialize_codex_apps_file_download_result_with_auth(
turn_context,
&sess.conversation_id.to_string(),
auth.as_ref(),
server,
codex_apps_meta,
result,
)
.await
}
async fn maybe_materialize_codex_apps_file_download_result_with_auth(
turn_context: &TurnContext,
session_id: &str,
auth: Option<&CodexAuth>,
server: &str,
codex_apps_meta: Option<&JsonMap<String, JsonValue>>,
mut result: CallToolResult,
) -> CallToolResult {
if !should_materialize_codex_apps_file_download(server, codex_apps_meta)
|| result.is_error == Some(true)
{
return result;
}
let Some(payload) = extract_codex_apps_file_download_payload(&result) else {
return result;
};
let download_base_url = codex_apps_download_base_url(turn_context);
if result.structured_content.is_none()
&& let Ok(structured_content) = serde_json::to_value(&payload)
{
result.structured_content = Some(structured_content);
}
let Some(auth) = auth else {
warn!(
"skipping codex_apps file download materialization because ChatGPT auth is unavailable"
);
return result;
};
let token_data = match auth.get_token_data() {
Ok(token_data) => token_data,
Err(error) => {
warn!(error = %error, "failed to read ChatGPT auth for codex_apps file download materialization");
return result;
}
};
let auth_provider = BearerAuthProvider {
token: Some(token_data.access_token),
account_id: token_data.account_id,
is_fedramp_account: auth.is_fedramp_account(),
};
let downloaded = match download_openai_file(
download_base_url,
&auth_provider,
&payload.file_uri.download_url,
)
.await
{
Ok(downloaded) => downloaded,
Err(error) => {
warn!(
error = %error,
file_id = payload.file_id,
"failed to materialize codex_apps file download via app-server",
);
return result;
}
};
let artifact_path = codex_apps_file_download_artifact_path(
&turn_context.config.codex_home,
session_id,
&payload.file_id,
payload
.file_name
.as_deref()
.or(payload.file_uri.file_name.as_deref())
.unwrap_or("downloaded_file"),
);
if let Some(parent) = artifact_path.parent()
&& let Err(error) = tokio::fs::create_dir_all(parent.as_path()).await
{
warn!(
error = %error,
path = %parent.display(),
"failed to create codex_apps file download artifact directory",
);
return result;
}
if let Err(error) = tokio::fs::write(artifact_path.as_path(), &downloaded).await {
warn!(
error = %error,
path = %artifact_path.display(),
"failed to write codex_apps file download artifact",
);
return result;
}
let local_path = artifact_path.to_string_lossy().to_string();
if let Some(JsonValue::Object(map)) = result.structured_content.as_mut() {
map.insert(
"local_path".to_string(),
JsonValue::String(local_path.clone()),
);
}
result.content.push(serde_json::json!({
"type": "text",
"text": format!("Downloaded file to local path: {local_path}"),
}));
result
}
fn extract_codex_apps_file_download_payload(
result: &CallToolResult,
) -> Option<CodexAppsFileDownloadPayload> {
if let Some(structured_content) = result.structured_content.clone()
&& let Ok(payload) =
serde_json::from_value::<CodexAppsFileDownloadPayload>(structured_content)
{
return Some(payload);
}
result
.content
.iter()
.filter_map(|item| item.as_object())
.find_map(|item| {
let text = item.get("text")?.as_str()?;
serde_json::from_str::<CodexAppsFileDownloadPayload>(text).ok()
})
}
fn codex_apps_file_download_artifact_path(
codex_home: &codex_utils_absolute_path::AbsolutePathBuf,
session_id: &str,
file_id: &str,
file_name: &str,
) -> codex_utils_absolute_path::AbsolutePathBuf {
codex_home
.join(CODEX_APPS_FILE_DOWNLOAD_ARTIFACTS_DIR)
.join(sanitize_path_component(session_id, "session"))
.join(sanitize_path_component(file_id, "file"))
.join(sanitize_file_name(file_name))
}
fn sanitize_path_component(value: &str, fallback: &str) -> String {
let sanitized: String = value
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
ch
} else {
'_'
}
})
.collect();
if sanitized.is_empty() {
fallback.to_string()
} else {
sanitized
}
}
fn sanitize_file_name(value: &str) -> String {
let sanitized: String = value
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '.' {
ch
} else {
'_'
}
})
.collect();
if sanitized.is_empty() {
"downloaded_file".to_string()
} else {
sanitized
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::session::tests::make_session_and_context;
use codex_login::CodexAuth;
use pretty_assertions::assert_eq;
use std::sync::Arc;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::header;
use wiremock::matchers::method;
use wiremock::matchers::path;
fn download_materialization_meta() -> JsonMap<String, JsonValue> {
serde_json::json!({
"provider": "builtin",
"materialize_file_download": true,
})
.as_object()
.cloned()
.expect("_codex_apps metadata object")
}
#[tokio::test]
async fn codex_apps_file_download_materialization_adds_local_path_for_marked_tools() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/download/file_123"))
.and(header("authorization", "Bearer Access Token"))
.and(header("chatgpt-account-id", "account_id"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("content-type", "text/plain")
.set_body_bytes(b"downloaded contents".to_vec()),
)
.mount(&server)
.await;
let (_, mut turn_context) = make_session_and_context().await;
let mut config = (*turn_context.config).clone();
config.chatgpt_base_url = format!("{}/backend-api/codex", server.uri());
turn_context.config = Arc::new(config);
let original = CallToolResult {
content: vec![serde_json::json!({
"type": "text",
"text": "{\"file_id\":\"file_123\"}",
})],
structured_content: Some(serde_json::json!({
"file_id": "file_123",
"file_name": "testing-file.txt",
"file_uri": {
"download_url": format!("{}/download/file_123", server.uri()),
"file_id": "file_123",
"file_name": "testing-file.txt",
"mime_type": "text/plain",
}
})),
is_error: Some(false),
meta: None,
};
let result = maybe_materialize_codex_apps_file_download_result_with_auth(
&turn_context,
"session-1",
Some(&CodexAuth::create_dummy_chatgpt_auth_for_testing()),
codex_mcp::CODEX_APPS_MCP_SERVER_NAME,
Some(&download_materialization_meta()),
original,
)
.await;
let local_path = result
.structured_content
.as_ref()
.and_then(|value| value.get("local_path"))
.and_then(JsonValue::as_str)
.expect("local_path in structured content");
assert!(local_path.contains("codex_apps_downloads"));
let saved = tokio::fs::read(local_path)
.await
.expect("saved local file should exist");
assert_eq!(saved, b"downloaded contents".to_vec());
assert!(result.content.iter().any(|block| {
block.get("type").and_then(JsonValue::as_str) == Some("text")
&& block
.get("text")
.and_then(JsonValue::as_str)
.is_some_and(|text| text.contains("Downloaded file to local path:"))
}));
}
}

View File

@@ -21,6 +21,7 @@ pub use codex_thread::CodexThread;
pub use codex_thread::CodexThreadTurnContextOverrides;
pub use codex_thread::ThreadConfigSnapshot;
mod agent;
mod codex_apps_file_download;
mod codex_delegate;
mod command_canonicalization;
mod commit_attribution;

View File

@@ -12,6 +12,7 @@
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use codex_api::OpenAiFileUploadOptions;
use codex_api::upload_local_file;
use codex_login::CodexAuth;
use codex_model_provider::BearerAuthProvider;
@@ -22,6 +23,7 @@ pub(crate) async fn rewrite_mcp_tool_arguments_for_openai_files(
turn_context: &TurnContext,
arguments_value: Option<JsonValue>,
openai_file_input_params: Option<&[String]>,
upload_options: Option<&OpenAiFileUploadOptions>,
) -> Result<Option<JsonValue>, String> {
let Some(openai_file_input_params) = openai_file_input_params else {
return Ok(arguments_value);
@@ -40,9 +42,14 @@ pub(crate) async fn rewrite_mcp_tool_arguments_for_openai_files(
let Some(value) = arguments.get(field_name) else {
continue;
};
let Some(uploaded_value) =
rewrite_argument_value_for_openai_files(turn_context, auth.as_ref(), field_name, value)
.await?
let Some(uploaded_value) = rewrite_argument_value_for_openai_files(
turn_context,
auth.as_ref(),
field_name,
value,
upload_options,
)
.await?
else {
continue;
};
@@ -61,6 +68,7 @@ async fn rewrite_argument_value_for_openai_files(
auth: Option<&CodexAuth>,
field_name: &str,
value: &JsonValue,
upload_options: Option<&OpenAiFileUploadOptions>,
) -> Result<Option<JsonValue>, String> {
match value {
JsonValue::String(path_or_file_ref) => {
@@ -70,6 +78,7 @@ async fn rewrite_argument_value_for_openai_files(
field_name,
/*index*/ None,
path_or_file_ref,
upload_options,
)
.await?;
Ok(Some(rewritten))
@@ -86,6 +95,7 @@ async fn rewrite_argument_value_for_openai_files(
field_name,
Some(index),
path_or_file_ref,
upload_options,
)
.await?;
rewritten_values.push(rewritten);
@@ -102,6 +112,7 @@ async fn build_uploaded_local_argument_value(
field_name: &str,
index: Option<usize>,
file_path: &str,
upload_options: Option<&OpenAiFileUploadOptions>,
) -> Result<JsonValue, String> {
let resolved_path = turn_context.resolve_path(Some(file_path.to_string()));
let Some(auth) = auth else {
@@ -117,10 +128,12 @@ async fn build_uploaded_local_argument_value(
account_id: token_data.account_id,
is_fedramp_account: auth.is_fedramp_account(),
};
let default_upload_options = OpenAiFileUploadOptions::default();
let uploaded = upload_local_file(
turn_context.config.chatgpt_base_url.trim_end_matches('/'),
&upload_auth,
&resolved_path,
upload_options.unwrap_or(&default_upload_options),
)
.await
.map_err(|error| match index {
@@ -129,14 +142,31 @@ async fn build_uploaded_local_argument_value(
}
None => format!("failed to upload `{file_path}` for `{field_name}`: {error}"),
})?;
Ok(serde_json::json!({
"download_url": uploaded.download_url,
"file_id": uploaded.file_id,
"mime_type": uploaded.mime_type,
"file_name": uploaded.file_name,
"uri": uploaded.uri,
"file_size_bytes": uploaded.file_size_bytes,
}))
let mut uploaded_value = serde_json::Map::from_iter([
("file_id".to_string(), serde_json::json!(uploaded.file_id)),
(
"file_name".to_string(),
serde_json::json!(uploaded.file_name),
),
("uri".to_string(), serde_json::json!(uploaded.uri)),
(
"file_size_bytes".to_string(),
serde_json::json!(uploaded.file_size_bytes),
),
]);
if let Some(download_url) = uploaded.download_url {
uploaded_value.insert("download_url".to_string(), serde_json::json!(download_url));
}
if let Some(mime_type) = uploaded.mime_type {
uploaded_value.insert("mime_type".to_string(), serde_json::json!(mime_type));
}
if let Some(library_file_id) = uploaded.library_file_id {
uploaded_value.insert(
"library_file_id".to_string(),
serde_json::json!(library_file_id),
);
}
Ok(JsonValue::Object(uploaded_value))
}
#[cfg(test)]
@@ -160,6 +190,7 @@ mod tests {
&Arc::new(turn_context),
arguments.clone(),
/*openai_file_input_params*/ None,
/*upload_options*/ None,
)
.await
.expect("rewrite should succeed");
@@ -231,6 +262,7 @@ mod tests {
"file",
/*index*/ None,
"file_report.csv",
/*upload_options*/ None,
)
.await
.expect("rewrite should upload the local file");
@@ -310,6 +342,7 @@ mod tests {
Some(&auth),
"file",
&serde_json::json!("file_report.csv"),
/*upload_options*/ None,
)
.await
.expect("rewrite should succeed");
@@ -424,6 +457,7 @@ mod tests {
Some(&auth),
"files",
&serde_json::json!(["one.csv", "two.csv"]),
/*upload_options*/ None,
)
.await
.expect("rewrite should succeed");
@@ -464,6 +498,7 @@ mod tests {
"file": "/definitely/missing/file.csv",
})),
Some(&["file".to_string()]),
/*upload_options*/ None,
)
.await
.expect_err("missing file should fail");

View File

@@ -12,6 +12,7 @@ use tracing::error;
use crate::arc_monitor::ArcMonitorOutcome;
use crate::arc_monitor::monitor_action;
use crate::codex_apps_file_download::maybe_materialize_codex_apps_file_download_result;
use crate::config::Config;
use crate::config::edit::ConfigEdit;
use crate::config::edit::ConfigEditsBuilder;
@@ -36,6 +37,7 @@ use crate::tools::sandboxing::PermissionRequestPayload;
use codex_analytics::AppInvocation;
use codex_analytics::InvocationType;
use codex_analytics::build_track_events_context;
use codex_api::OpenAiFileUploadOptions;
use codex_config::types::AppToolApproval;
use codex_features::Feature;
use codex_hooks::PermissionRequestDecision;
@@ -477,6 +479,14 @@ async fn execute_mcp_tool_call(
rewritten_arguments: Option<JsonValue>,
request_meta: Option<JsonValue>,
) -> Result<CallToolResult, String> {
let rewritten_arguments = rewrite_mcp_tool_arguments_for_openai_files(
sess,
turn_context,
arguments_value,
metadata.and_then(|metadata| metadata.openai_file_input_params.as_deref()),
metadata.and_then(|metadata| metadata.openai_file_upload_options.as_ref()),
)
.await?;
let request_meta =
with_mcp_tool_call_thread_id_meta(request_meta, &sess.conversation_id.to_string());
let request_meta =
@@ -487,6 +497,14 @@ async fn execute_mcp_tool_call(
.call_tool(server, tool_name, rewritten_arguments, request_meta)
.await
.map_err(|e| format!("tool call error: {e:?}"))?;
let result = maybe_materialize_codex_apps_file_download_result(
sess,
turn_context,
server,
metadata.and_then(|metadata| metadata.codex_apps_meta.as_ref()),
result,
)
.await;
sanitize_mcp_tool_result_for_model(
turn_context
.model_info
@@ -658,12 +676,37 @@ pub(crate) struct McpToolApprovalMetadata {
mcp_app_resource_uri: Option<String>,
codex_apps_meta: Option<serde_json::Map<String, serde_json::Value>>,
openai_file_input_params: Option<Vec<String>>,
openai_file_upload_options: Option<OpenAiFileUploadOptions>,
}
const MCP_TOOL_CODEX_APPS_META_KEY: &str = "_codex_apps";
const CODEX_APPS_META_KEY: &str = "_codex_apps";
const MCP_TOOL_OPENAI_OUTPUT_TEMPLATE_META_KEY: &str = "openai/outputTemplate";
const MCP_TOOL_UI_RESOURCE_URI_META_KEY: &str = "ui/resourceUri";
const MCP_TOOL_THREAD_ID_META_KEY: &str = "threadId";
const MCP_TOOL_OPENAI_FILE_UPLOAD_CONFIG_KEY: &str = "openai/fileUploadConfig";
#[derive(Debug, Clone, Deserialize)]
struct RawOpenAiFileUploadConfig {
#[serde(default)]
store_in_library: bool,
}
fn parse_openai_file_upload_options(
meta: Option<&serde_json::Map<String, serde_json::Value>>,
) -> Option<OpenAiFileUploadOptions> {
let raw = meta?
.get(MCP_TOOL_OPENAI_FILE_UPLOAD_CONFIG_KEY)
.cloned()
.and_then(|value| serde_json::from_value::<RawOpenAiFileUploadConfig>(value).ok())?;
if !raw.store_in_library {
return None;
}
Some(OpenAiFileUploadOptions {
store_in_library: true,
})
}
fn custom_mcp_tool_approval_mode(
turn_context: &TurnContext,
@@ -710,7 +753,7 @@ fn build_mcp_tool_call_request_meta(
metadata.and_then(|metadata| metadata.codex_apps_meta.clone())
{
request_meta.insert(
MCP_TOOL_CODEX_APPS_META_KEY.to_string(),
CODEX_APPS_META_KEY.to_string(),
serde_json::Value::Object(codex_apps_meta),
);
}
@@ -1182,13 +1225,16 @@ pub(crate) async fn lookup_mcp_tool_metadata(
.tool
.meta
.as_ref()
.and_then(|meta| meta.get(MCP_TOOL_CODEX_APPS_META_KEY))
.and_then(|meta| meta.get(CODEX_APPS_META_KEY))
.and_then(serde_json::Value::as_object)
.cloned(),
openai_file_input_params: Some(declared_openai_file_input_param_names(
tool_info.tool.meta.as_deref(),
))
.filter(|params| !params.is_empty()),
openai_file_upload_options: parse_openai_file_upload_options(
tool_info.tool.meta.as_deref(),
),
})
}

View File

@@ -65,6 +65,7 @@ fn approval_metadata(
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
}
}
@@ -182,6 +183,24 @@ fn mcp_app_resource_uri_reads_known_tool_meta_keys() {
);
}
#[test]
fn parse_openai_file_upload_options_reads_store_in_library_config() {
let meta = serde_json::json!({
"openai/fileUploadConfig": {
"store_in_library": true,
}
});
let parsed = parse_openai_file_upload_options(meta.as_object());
assert_eq!(
parsed,
Some(codex_api::OpenAiFileUploadOptions {
store_in_library: true,
})
);
}
#[test]
fn approval_required_when_read_only_false_and_destructive() {
let annotations = annotations(Some(false), Some(true), /*open_world*/ None);
@@ -709,6 +728,7 @@ async fn codex_apps_tool_call_request_meta_includes_turn_metadata_and_codex_apps
.expect("_codex_apps metadata should be an object"),
),
openai_file_input_params: None,
openai_file_upload_options: None,
};
assert_eq!(
@@ -719,7 +739,7 @@ async fn codex_apps_tool_call_request_meta_includes_turn_metadata_and_codex_apps
),
Some(serde_json::json!({
crate::X_CODEX_TURN_METADATA_HEADER: expected_turn_metadata,
MCP_TOOL_CODEX_APPS_META_KEY: {
"_codex_apps": {
"resource_uri": "connector://calendar/tools/calendar_create_event",
"contains_mcp_source": true,
"connector_id": "calendar",
@@ -887,6 +907,7 @@ fn guardian_mcp_review_request_includes_annotations_when_present() {
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
let request = build_guardian_mcp_tool_review_request("call-1", &invocation, Some(&metadata));
@@ -1451,6 +1472,7 @@ async fn approve_mode_skips_when_annotations_do_not_require_approval() {
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
let decision = maybe_request_mcp_tool_approval(
@@ -1524,6 +1546,7 @@ async fn guardian_mode_skips_auto_when_annotations_do_not_require_approval() {
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
let decision = maybe_request_mcp_tool_approval(
@@ -1797,6 +1820,7 @@ async fn guardian_mode_mcp_denial_returns_rationale_message() {
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
let decision = maybe_request_mcp_tool_approval(
@@ -1850,6 +1874,7 @@ async fn prompt_mode_waits_for_approval_when_annotations_do_not_require_approval
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
let mut approval_task = {
@@ -1929,6 +1954,7 @@ async fn approve_mode_blocks_when_arc_returns_interrupt_for_model() {
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
let decision = maybe_request_mcp_tool_approval(
@@ -2001,6 +2027,7 @@ async fn custom_approve_mode_blocks_when_arc_returns_interrupt_for_model() {
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
let decision = maybe_request_mcp_tool_approval(
@@ -2073,6 +2100,7 @@ async fn approve_mode_blocks_when_arc_returns_interrupt_without_annotations() {
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
let decision = maybe_request_mcp_tool_approval(
@@ -2153,6 +2181,7 @@ async fn full_access_mode_skips_arc_monitor_for_all_approval_modes() {
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
for approval_mode in [
@@ -2260,6 +2289,7 @@ async fn approve_mode_routes_arc_ask_user_to_guardian_when_guardian_reviewer_is_
mcp_app_resource_uri: None,
codex_apps_meta: None,
openai_file_input_params: None,
openai_file_upload_options: None,
};
let decision = maybe_request_mcp_tool_approval(

View File

@@ -96,7 +96,7 @@ async fn codex_apps_file_params_upload_local_paths_before_mcp_tool_call() -> Res
let apps_server = AppsTestServer::mount(&server).await?;
Mock::given(method("POST"))
.and(path("/files"))
.and(path("/backend-api/files"))
.and(header("chatgpt-account-id", "account_id"))
.and(body_json(json!({
"file_name": "report.txt",
@@ -118,7 +118,7 @@ async fn codex_apps_file_params_upload_local_paths_before_mcp_tool_call() -> Res
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/files/file_123/uploaded"))
.and(path("/backend-api/files/file_123/uploaded"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"status": "success",
"download_url": format!("{}/download/file_123", server.uri()),

View File

@@ -11,6 +11,7 @@ const DISALLOWED_CONNECTOR_IDS: &[&str] = &[
];
const FIRST_PARTY_CHAT_DISALLOWED_CONNECTOR_IDS: &[&str] =
&["connector_0f9c9d4592e54d0a9a12b3f44a1e2010"];
const ALLOWED_OPENAI_CONNECTOR_IDS: &[&str] = &["connector_openai_library"];
const DISALLOWED_CONNECTOR_PREFIX: &str = "connector_openai_";
pub fn is_connector_id_allowed(connector_id: &str) -> bool {
@@ -24,6 +25,10 @@ fn is_connector_id_allowed_for_originator(connector_id: &str, originator_value:
DISALLOWED_CONNECTOR_IDS
};
if ALLOWED_OPENAI_CONNECTOR_IDS.contains(&connector_id) {
return true;
}
!connector_id.starts_with(DISALLOWED_CONNECTOR_PREFIX)
&& !disallowed_connector_ids.contains(&connector_id)
}