diff --git a/AGENTS.md b/AGENTS.md index 297fb0fe15..6e21be78f6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -23,6 +23,7 @@ In the codex-rs folder where the rust code lives: - When making a change that adds or changes an API, ensure that the documentation in the `docs/` folder is up to date if applicable. - Prefer private modules and explicitly exported public crate API. - If you change `ConfigToml` or nested config types, run `just write-config-schema` to update `codex-rs/core/config.schema.json`. +- When working with MCP tool calls, prefer using `codex-rs/codex-mcp/src/mcp_connection_manager.rs` to handle mutation of tools and tool calls. Aim to minimize the footprint of changes and leverage existing abstractions rather than plumbing code through multiple levels of function calls. - If you change Rust dependencies (`Cargo.toml` or `Cargo.lock`), run `just bazel-lock-update` from the repo root to refresh `MODULE.bazel.lock`, and include that lockfile update in the same change. - After dependency changes, run `just bazel-lock-check` from the repo root so lockfile drift is caught diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 93db88947b..894dc096fd 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1414,6 +1414,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-test", diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 4e3bc70cb3..aae240e3f9 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -199,6 +199,7 @@ use codex_core::SteerInputError; use codex_core::ThreadConfigSnapshot; use codex_core::ThreadManager; use codex_core::ThreadSortKey as CoreThreadSortKey; +use codex_core::append_thread_name; use codex_core::config::Config; use codex_core::config::ConfigOverrides; use codex_core::config::NetworkProxyAuditMetadata; @@ -288,11 +289,13 @@ use codex_protocol::protocol::ReviewTarget as CoreReviewTarget; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionMetaLine; +use codex_protocol::protocol::ThreadNameUpdatedEvent; use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::protocol::W3cTraceContext; use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS; use codex_protocol::user_input::UserInput as CoreInputItem; use codex_rmcp_client::perform_oauth_login_return_url; +use codex_rollout::append_rollout_item_to_path; use codex_rollout::state_db::StateDbHandle; use codex_rollout::state_db::get_state_db; use codex_rollout::state_db::reconcile_rollout; @@ -2674,11 +2677,11 @@ impl CodexMessageProcessor { return; } - let thread_exists = + let rollout_path = match find_thread_path_by_id_str(&self.config.codex_home, &thread_id.to_string()).await { - Ok(Some(_)) => true, - Ok(None) => false, + Ok(Some(path)) => Some(path), + Ok(None) => None, Err(err) => { self.send_invalid_request_error( request_id, @@ -2689,19 +2692,39 @@ impl CodexMessageProcessor { } }; - if !thread_exists { + let Some(rollout_path) = rollout_path else { self.send_invalid_request_error(request_id, format!("thread not found: {thread_id}")) .await; return; - } + }; - if let Err(err) = - codex_core::append_thread_name(&self.config.codex_home, thread_id, &name).await - { + let msg = EventMsg::ThreadNameUpdated(ThreadNameUpdatedEvent { + thread_id, + thread_name: Some(name.clone()), + }); + let item = RolloutItem::EventMsg(msg); + if let Err(err) = append_rollout_item_to_path(rollout_path.as_path(), &item).await { self.send_internal_error(request_id, format!("failed to set thread name: {err}")) .await; return; } + if let Err(err) = append_thread_name(&self.config.codex_home, thread_id, &name).await { + self.send_internal_error(request_id, format!("failed to index thread name: {err}")) + .await; + return; + } + + let state_db_ctx = open_state_db_for_direct_thread_lookup(&self.config).await; + reconcile_rollout( + state_db_ctx.as_deref(), + rollout_path.as_path(), + self.config.model_provider_id.as_str(), + /*builder*/ None, + &[], + /*archived_only*/ None, + /*new_thread_memory_mode*/ None, + ) + .await; self.outgoing .send_response(request_id, ThreadSetNameResponse {}) @@ -3445,13 +3468,7 @@ impl CodexMessageProcessor { threads.push((conversation_id, thread)); } - let names = match find_thread_names_by_ids(&self.config.codex_home, &thread_ids).await { - Ok(names) => names, - Err(err) => { - warn!("Failed to read thread names: {err}"); - HashMap::new() - } - }; + let names = thread_titles_by_ids(&self.config, &thread_ids).await; let statuses = self .thread_watch_manager @@ -3461,7 +3478,9 @@ impl CodexMessageProcessor { let data = threads .into_iter() .map(|(conversation_id, mut thread)| { - thread.name = names.get(&conversation_id).cloned(); + if let Some(title) = names.get(&conversation_id).cloned() { + set_thread_name_from_title(&mut thread, title); + } if let Some(status) = statuses.get(&thread.id) { thread.status = status.clone(); } @@ -4265,13 +4284,8 @@ impl CodexMessageProcessor { } async fn attach_thread_name(&self, thread_id: ThreadId, thread: &mut Thread) { - match find_thread_name_by_id(&self.config.codex_home, &thread_id).await { - Ok(name) => { - thread.name = name; - } - Err(err) => { - warn!("Failed to read thread name for {thread_id}: {err}"); - } + if let Some(title) = title_from_state_db(&self.config, thread_id).await { + set_thread_name_from_title(thread, title); } } @@ -8029,7 +8043,7 @@ async fn handle_thread_listener_command( async fn handle_pending_thread_resume_request( conversation_id: ThreadId, conversation: &Arc, - codex_home: &Path, + _codex_home: &Path, thread_state_manager: &ThreadStateManager, thread_state: &Arc>, thread_watch_manager: &ThreadWatchManager, @@ -8087,11 +8101,6 @@ async fn handle_pending_thread_resume_request( has_live_in_progress_turn, ); - match find_thread_name_by_id(codex_home, &conversation_id).await { - Ok(thread_name) => thread.name = thread_name, - Err(err) => warn!("Failed to read thread name for {conversation_id}: {err}"), - } - let ThreadConfigSnapshot { model, model_provider_id, @@ -8653,7 +8662,7 @@ async fn read_summary_from_state_db_by_thread_id( config: &Config, thread_id: ThreadId, ) -> Option { - let state_db_ctx = get_state_db(config).await; + let state_db_ctx = open_state_db_for_direct_thread_lookup(config).await; read_summary_from_state_db_context_by_thread_id(state_db_ctx.as_ref(), thread_id).await } @@ -8670,6 +8679,71 @@ async fn read_summary_from_state_db_context_by_thread_id( Some(summary_from_thread_metadata(&metadata)) } +async fn title_from_state_db(config: &Config, thread_id: ThreadId) -> Option { + if let Some(state_db_ctx) = open_state_db_for_direct_thread_lookup(config).await + && let Some(metadata) = state_db_ctx.get_thread(thread_id).await.ok().flatten() + && let Some(title) = distinct_title(&metadata) + { + return Some(title); + } + find_thread_name_by_id(&config.codex_home, &thread_id) + .await + .ok() + .flatten() +} + +async fn thread_titles_by_ids( + config: &Config, + thread_ids: &HashSet, +) -> HashMap { + let mut names = HashMap::with_capacity(thread_ids.len()); + if let Some(state_db_ctx) = open_state_db_for_direct_thread_lookup(config).await { + for &thread_id in thread_ids { + let Ok(Some(metadata)) = state_db_ctx.get_thread(thread_id).await else { + continue; + }; + if let Some(title) = distinct_title(&metadata) { + names.insert(thread_id, title); + } + } + } + if names.len() < thread_ids.len() + && let Ok(legacy_names) = find_thread_names_by_ids(&config.codex_home, thread_ids).await + { + for (thread_id, title) in legacy_names { + names.entry(thread_id).or_insert(title); + } + } + names +} + +async fn open_state_db_for_direct_thread_lookup(config: &Config) -> Option { + StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone()) + .await + .ok() +} + +fn non_empty_title(metadata: &ThreadMetadata) -> Option { + let title = metadata.title.trim(); + (!title.is_empty()).then(|| title.to_string()) +} + +fn distinct_title(metadata: &ThreadMetadata) -> Option { + let title = non_empty_title(metadata)?; + if metadata.first_user_message.as_deref().map(str::trim) == Some(title.as_str()) { + None + } else { + Some(title) + } +} + +fn set_thread_name_from_title(thread: &mut Thread, title: String) { + if title.trim().is_empty() || thread.preview.trim() == title.trim() { + return; + } + thread.name = Some(title); +} + async fn summary_from_thread_list_item( it: codex_core::ThreadItem, fallback_provider: &str, @@ -8968,6 +9042,14 @@ async fn load_thread_summary_for_rollout( } else if let Some(summary) = read_summary_from_state_db_by_thread_id(config, thread_id).await { merge_mutable_thread_metadata(&mut thread, summary_to_thread(summary)); } + let title = if let Some(metadata) = persisted_metadata { + non_empty_title(metadata) + } else { + title_from_state_db(config, thread_id).await + }; + if let Some(title) = title { + set_thread_name_from_title(&mut thread, title); + } Ok(thread) } diff --git a/codex-rs/app-server/tests/suite/v2/thread_name_websocket.rs b/codex-rs/app-server/tests/suite/v2/thread_name_websocket.rs index 4ec7344d6c..b41e2f1d18 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_name_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_name_websocket.rs @@ -21,7 +21,14 @@ use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadSetNameParams; use codex_app_server_protocol::ThreadSetNameResponse; +use codex_core::find_thread_name_by_id; +use codex_core::find_thread_path_by_id_str; +use codex_protocol::ThreadId; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::RolloutLine; use pretty_assertions::assert_eq; +use std::path::Path; use tempfile::TempDir; use tokio::time::Duration; use tokio::time::timeout; @@ -77,6 +84,11 @@ async fn thread_name_updated_broadcasts_for_loaded_threads() -> Result<()> { let ws2_notification = read_notification_for_method(&mut ws2, "thread/name/updated").await?; assert_thread_name_updated(ws2_notification, &conversation_id, renamed)?; + assert_legacy_thread_name(codex_home.path(), &conversation_id, renamed).await?; + assert_eq!( + thread_name_update_rollout_count(codex_home.path(), &conversation_id).await?, + 1 + ); assert_no_message(&mut ws1, Duration::from_millis(250)).await?; assert_no_message(&mut ws2, Duration::from_millis(250)).await?; @@ -128,6 +140,11 @@ async fn thread_name_updated_broadcasts_for_not_loaded_threads() -> Result<()> { let ws2_notification = read_notification_for_method(&mut ws2, "thread/name/updated").await?; assert_thread_name_updated(ws2_notification, &conversation_id, renamed)?; + assert_legacy_thread_name(codex_home.path(), &conversation_id, renamed).await?; + assert_eq!( + thread_name_update_rollout_count(codex_home.path(), &conversation_id).await?, + 1 + ); assert_no_message(&mut ws1, Duration::from_millis(250)).await?; assert_no_message(&mut ws2, Duration::from_millis(250)).await?; @@ -174,3 +191,38 @@ fn assert_thread_name_updated( assert_eq!(notification.thread_name.as_deref(), Some(thread_name)); Ok(()) } + +async fn assert_legacy_thread_name( + codex_home: &Path, + conversation_id: &str, + expected_name: &str, +) -> Result<()> { + let thread_id = ThreadId::from_string(conversation_id)?; + assert_eq!( + find_thread_name_by_id(codex_home, &thread_id) + .await? + .as_deref(), + Some(expected_name) + ); + Ok(()) +} + +async fn thread_name_update_rollout_count( + codex_home: &Path, + conversation_id: &str, +) -> Result { + let rollout_path = find_thread_path_by_id_str(codex_home, conversation_id) + .await? + .context("rollout path")?; + let contents = tokio::fs::read_to_string(rollout_path).await?; + Ok(contents + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .filter(|line| { + matches!( + line.item, + RolloutItem::EventMsg(EventMsg::ThreadNameUpdated(_)) + ) + }) + .count()) +} diff --git a/codex-rs/codex-api/Cargo.toml b/codex-rs/codex-api/Cargo.toml index 7bafd15328..862d262b90 100644 --- a/codex-rs/codex-api/Cargo.toml +++ b/codex-rs/codex-api/Cargo.toml @@ -14,22 +14,24 @@ codex-protocol = { workspace = true } codex-utils-rustls-provider = { workspace = true } futures = { workspace = true } http = { workspace = true } +reqwest = { workspace = true, features = ["json", "stream"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } thiserror = { workspace = true } -tokio = { workspace = true, features = ["macros", "net", "rt", "sync", "time"] } +tokio = { workspace = true, features = ["fs", "macros", "net", "rt", "sync", "time"] } tokio-tungstenite = { workspace = true } tungstenite = { workspace = true } tracing = { workspace = true } eventsource-stream = { workspace = true } regex-lite = { workspace = true } -tokio-util = { workspace = true, features = ["codec"] } +tokio-util = { workspace = true, features = ["codec", "io"] } url = { workspace = true } [dev-dependencies] anyhow = { workspace = true } assert_matches = { workspace = true } pretty_assertions = { workspace = true } +tempfile = { workspace = true } tokio-test = { workspace = true } wiremock = { workspace = true } reqwest = { workspace = true } diff --git a/codex-rs/codex-api/src/files.rs b/codex-rs/codex-api/src/files.rs new file mode 100644 index 0000000000..6fad5b62f5 --- /dev/null +++ b/codex-rs/codex-api/src/files.rs @@ -0,0 +1,370 @@ +use std::path::Path; +use std::path::PathBuf; +use std::time::Duration; + +use crate::AuthProvider; +use codex_client::build_reqwest_client_with_custom_ca; +use reqwest::StatusCode; +use reqwest::header::CONTENT_LENGTH; +use serde::Deserialize; +use tokio::fs::File; +use tokio::time::Instant; +use tokio_util::io::ReaderStream; + +pub const OPENAI_FILE_URI_PREFIX: &str = "sediment://"; +pub const OPENAI_FILE_UPLOAD_LIMIT_BYTES: u64 = 512 * 1024 * 1024; + +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"; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UploadedOpenAiFile { + pub file_id: String, + pub uri: String, + pub download_url: String, + pub file_name: String, + pub file_size_bytes: u64, + pub mime_type: Option, + pub path: PathBuf, +} + +#[derive(Debug, thiserror::Error)] +pub enum OpenAiFileError { + #[error("path `{path}` does not exist")] + MissingPath { path: PathBuf }, + #[error("path `{path}` is not a file")] + NotAFile { path: PathBuf }, + #[error("path `{path}` cannot be read: {source}")] + ReadFile { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error( + "file `{path}` is too large: {size_bytes} bytes exceeds the limit of {limit_bytes} bytes" + )] + FileTooLarge { + path: PathBuf, + size_bytes: u64, + limit_bytes: u64, + }, + #[error("failed to send OpenAI file request to {url}: {source}")] + Request { + url: String, + #[source] + source: reqwest::Error, + }, + #[error("OpenAI file request to {url} failed with status {status}: {body}")] + UnexpectedStatus { + url: String, + status: StatusCode, + body: String, + }, + #[error("failed to parse OpenAI file response from {url}: {source}")] + Decode { + url: String, + #[source] + source: serde_json::Error, + }, + #[error("OpenAI file upload for `{file_id}` is not ready yet")] + UploadNotReady { file_id: String }, + #[error("OpenAI file upload for `{file_id}` failed: {message}")] + UploadFailed { file_id: String, message: String }, +} + +#[derive(Deserialize)] +struct CreateFileResponse { + file_id: String, + upload_url: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "snake_case")] +struct DownloadLinkResponse { + status: String, + download_url: Option, + file_name: Option, + mime_type: Option, + error_message: Option, +} + +pub fn openai_file_uri(file_id: &str) -> String { + format!("{OPENAI_FILE_URI_PREFIX}{file_id}") +} + +pub async fn upload_local_file( + base_url: &str, + auth: &impl AuthProvider, + path: &Path, +) -> Result { + let metadata = tokio::fs::metadata(path) + .await + .map_err(|source| match source.kind() { + std::io::ErrorKind::NotFound => OpenAiFileError::MissingPath { + path: path.to_path_buf(), + }, + _ => OpenAiFileError::ReadFile { + path: path.to_path_buf(), + source, + }, + })?; + if !metadata.is_file() { + return Err(OpenAiFileError::NotAFile { + path: path.to_path_buf(), + }); + } + if metadata.len() > OPENAI_FILE_UPLOAD_LIMIT_BYTES { + return Err(OpenAiFileError::FileTooLarge { + path: path.to_path_buf(), + size_bytes: metadata.len(), + limit_bytes: OPENAI_FILE_UPLOAD_LIMIT_BYTES, + }); + } + + let file_name = path + .file_name() + .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 create_payload: CreateFileResponse = + serde_json::from_str(&create_body).map_err(|source| OpenAiFileError::Decode { + url: create_url.clone(), + source, + })?; + + let upload_file = File::open(path) + .await + .map_err(|source| OpenAiFileError::ReadFile { + path: path.to_path_buf(), + source, + })?; + let upload_response = build_reqwest_client() + .put(&create_payload.upload_url) + .timeout(OPENAI_FILE_REQUEST_TIMEOUT) + .header("x-ms-blob-type", "BlockBlob") + .header(CONTENT_LENGTH, metadata.len()) + .body(reqwest::Body::wrap_stream(ReaderStream::new(upload_file))) + .send() + .await + .map_err(|source| OpenAiFileError::Request { + 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, + }); + } + + let finalize_url = format!( + "{}/files/{}/uploaded", + base_url.trim_end_matches('/'), + create_payload.file_id, + ); + let finalize_started_at = Instant::now(); + loop { + let finalize_response = authorized_request(auth, reqwest::Method::POST, &finalize_url) + .json(&serde_json::json!({})) + .send() + .await + .map_err(|source| OpenAiFileError::Request { + 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, + })?; + + 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(|| { + 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(), + }); + } + "retry" => { + if finalize_started_at.elapsed() >= OPENAI_FILE_FINALIZE_TIMEOUT { + return Err(OpenAiFileError::UploadNotReady { + file_id: create_payload.file_id, + }); + } + tokio::time::sleep(OPENAI_FILE_FINALIZE_RETRY_DELAY).await; + } + _ => { + return Err(OpenAiFileError::UploadFailed { + file_id: create_payload.file_id, + message: finalize_payload + .error_message + .unwrap_or_else(|| "upload finalization returned an error".to_string()), + }); + } + } + } +} + +fn authorized_request( + auth: &impl AuthProvider, + method: reqwest::Method, + url: &str, +) -> reqwest::RequestBuilder { + let client = build_reqwest_client(); + let mut request = client + .request(method, url) + .timeout(OPENAI_FILE_REQUEST_TIMEOUT); + if let Some(token) = auth.bearer_token() { + request = request.bearer_auth(token); + } + if let Some(account_id) = auth.account_id() { + request = request.header("chatgpt-account-id", account_id); + } + request +} + +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"); + reqwest::Client::new() + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::CoreAuthProvider; + use pretty_assertions::assert_eq; + use std::sync::Arc; + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + use tempfile::TempDir; + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::Request; + use wiremock::ResponseTemplate; + use wiremock::matchers::body_json; + use wiremock::matchers::header; + use wiremock::matchers::method; + use wiremock::matchers::path; + + fn chatgpt_auth() -> CoreAuthProvider { + CoreAuthProvider::for_test(Some("token"), Some("account_id")) + } + + fn base_url_for(server: &MockServer) -> String { + format!("{}/backend-api", server.uri()) + } + + #[tokio::test] + async fn upload_local_file_returns_canonical_uri() { + 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", + }))) + .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; + let finalize_attempts = Arc::new(AtomicUsize::new(0)); + let finalize_attempts_responder = Arc::clone(&finalize_attempts); + let download_url = format!("{}/download/file_123", server.uri()); + Mock::given(method("POST")) + .and(path("/backend-api/files/file_123/uploaded")) + .respond_with(move |_request: &Request| { + if finalize_attempts_responder.fetch_add(1, Ordering::SeqCst) == 0 { + return ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "status": "retry" + })); + } + + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "status": "success", + "download_url": download_url, + "file_name": "hello.txt", + "mime_type": "text/plain", + "file_size_bytes": 5 + })) + }) + .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) + .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()) + ); + 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); + } +} diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index f2e415e96a..ac26d3cdba 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -3,6 +3,7 @@ pub(crate) mod auth; pub(crate) mod common; pub(crate) mod endpoint; pub(crate) mod error; +pub(crate) mod files; pub(crate) mod provider; pub(crate) mod rate_limits; pub(crate) mod requests; @@ -52,6 +53,7 @@ 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::upload_local_file; pub use crate::provider::Provider; pub use crate::provider::RetryConfig; pub use crate::provider::is_azure_responses_wire_base_url; diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index 8912c71df6..153ddcf0b4 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -38,4 +38,5 @@ pub use mcp_connection_manager::McpConnectionManager; pub use mcp_connection_manager::SandboxState; pub use mcp_connection_manager::ToolInfo; pub use mcp_connection_manager::codex_apps_tools_cache_key; +pub use mcp_connection_manager::declared_openai_file_input_param_names; pub use mcp_connection_manager::filter_non_codex_apps_mcp_tools_only; diff --git a/codex-rs/codex-mcp/src/mcp_connection_manager.rs b/codex-rs/codex-mcp/src/mcp_connection_manager.rs index f0f292fcb8..1bb8f7cf8d 100644 --- a/codex-rs/codex-mcp/src/mcp_connection_manager.rs +++ b/codex-rs/codex-mcp/src/mcp_connection_manager.rs @@ -74,6 +74,8 @@ use rmcp::model::Tool; use serde::Deserialize; use serde::Serialize; +use serde_json::Map; +use serde_json::Value as JsonValue; use sha1::Digest; use sha1::Sha1; use tokio::sync::Mutex; @@ -197,6 +199,89 @@ pub struct ToolInfo { pub connector_description: Option, } +const META_OPENAI_FILE_PARAMS: &str = "openai/fileParams"; + +pub fn declared_openai_file_input_param_names( + meta: Option<&Map>, +) -> Vec { + let Some(meta) = meta else { + return Vec::new(); + }; + + meta.get(META_OPENAI_FILE_PARAMS) + .and_then(JsonValue::as_array) + .into_iter() + .flatten() + .filter_map(JsonValue::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_string) + .collect() +} + +/// Returns the model-visible view of a tool while preserving the raw metadata +/// used by execution. Keep cache entries raw and call this at manager return +/// boundaries. +fn tool_with_model_visible_input_schema(tool: &Tool) -> Tool { + let file_params = declared_openai_file_input_param_names(tool.meta.as_deref()); + if file_params.is_empty() { + return tool.clone(); + } + + let mut tool = tool.clone(); + let mut input_schema = JsonValue::Object(tool.input_schema.as_ref().clone()); + mask_input_schema_for_file_path_params(&mut input_schema, &file_params); + if let JsonValue::Object(input_schema) = input_schema { + tool.input_schema = Arc::new(input_schema); + } + tool +} + +fn mask_input_schema_for_file_path_params(input_schema: &mut JsonValue, file_params: &[String]) { + let Some(properties) = input_schema + .as_object_mut() + .and_then(|schema| schema.get_mut("properties")) + .and_then(JsonValue::as_object_mut) + else { + return; + }; + + for field_name in file_params { + let Some(property_schema) = properties.get_mut(field_name) else { + continue; + }; + mask_input_property_schema(property_schema); + } +} + +fn mask_input_property_schema(schema: &mut JsonValue) { + let Some(object) = schema.as_object_mut() else { + return; + }; + + let mut description = object + .get("description") + .and_then(JsonValue::as_str) + .map(str::to_string) + .unwrap_or_default(); + let guidance = "This parameter expects an absolute local file path. If you want to upload a file, provide the absolute path to that file here."; + if description.is_empty() { + description = guidance.to_string(); + } else if !description.contains(guidance) { + description = format!("{description} {guidance}"); + } + + let is_array = object.get("type").and_then(JsonValue::as_str) == Some("array") + || object.get("items").is_some(); + object.clear(); + object.insert("description".to_string(), JsonValue::String(description)); + if is_array { + object.insert("type".to_string(), JsonValue::String("array".to_string())); + object.insert("items".to_string(), serde_json::json!({ "type": "string" })); + } else { + object.insert("type".to_string(), JsonValue::String("string".to_string())); + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct CodexAppsToolsCacheKey { account_id: Option, @@ -534,6 +619,10 @@ impl AsyncManagedClient { let annotate_tools = |tools: Vec| { let mut tools = tools; for tool in &mut tools { + if tool.server_name == CODEX_APPS_MCP_SERVER_NAME { + tool.tool = tool_with_model_visible_input_schema(&tool.tool); + } + let plugin_names = match tool.connector_id.as_deref() { Some(connector_id) => self .tool_plugin_provenance @@ -914,10 +1003,13 @@ impl McpConnectionManager { list_start.elapsed(), &[("cache", "miss")], ); - Ok(qualify_tools(filter_tools( - tools, - &managed_client.tool_filter, - ))) + let tools = filter_tools(tools, &managed_client.tool_filter) + .into_iter() + .map(|mut tool| { + tool.tool = tool_with_model_visible_input_schema(&tool.tool); + tool + }); + Ok(qualify_tools(tools)) } /// Returns a single map that contains all resources. Each key is the @@ -1419,6 +1511,12 @@ async fn start_server_task( .await .map_err(StartupOutcomeError::from)?; + let server_supports_sandbox_state_capability = initialize_result + .capabilities + .experimental + .as_ref() + .and_then(|exp| exp.get(MCP_SANDBOX_STATE_CAPABILITY)) + .is_some(); let list_start = Instant::now(); let fetch_start = Instant::now(); let tools = list_tools_for_client_uncached( @@ -1448,12 +1546,6 @@ async fn start_server_task( } let tools = filter_tools(tools, &tool_filter); - let server_supports_sandbox_state_capability = initialize_result - .capabilities - .experimental - .as_ref() - .and_then(|exp| exp.get(MCP_SANDBOX_STATE_CAPABILITY)) - .is_some(); let managed = ManagedClient { client: Arc::clone(&client), tools, diff --git a/codex-rs/codex-mcp/src/mcp_connection_manager_tests.rs b/codex-rs/codex-mcp/src/mcp_connection_manager_tests.rs index 76b16c959d..f299a9e040 100644 --- a/codex-rs/codex-mcp/src/mcp_connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/mcp_connection_manager_tests.rs @@ -3,6 +3,7 @@ use codex_protocol::protocol::GranularApprovalConfig; use codex_protocol::protocol::McpAuthStatus; use pretty_assertions::assert_eq; use rmcp::model::JsonObject; +use rmcp::model::Meta; use rmcp::model::NumberOrString; use std::collections::HashSet; use std::sync::Arc; @@ -63,6 +64,86 @@ fn create_codex_apps_tools_cache_context( } } +#[test] +fn declared_openai_file_fields_treat_names_literally() { + let meta = serde_json::json!({ + "openai/fileParams": ["file", "input_file", "attachments"] + }); + let meta = meta.as_object().expect("meta object"); + + assert_eq!( + declared_openai_file_input_param_names(Some(meta)), + vec![ + "file".to_string(), + "input_file".to_string(), + "attachments".to_string(), + ] + ); +} + +#[test] +fn tool_with_model_visible_input_schema_masks_file_params() { + let mut tool = create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "upload").tool; + tool.input_schema = Arc::new( + serde_json::json!({ + "type": "object", + "properties": { + "file": { + "type": "object", + "description": "Original file payload." + }, + "files": { + "type": "array", + "items": {"type": "object"} + } + } + }) + .as_object() + .expect("object") + .clone(), + ); + tool.meta = Some(Meta( + serde_json::json!({ + "openai/fileParams": ["file", "files"] + }) + .as_object() + .expect("object") + .clone(), + )); + + let tool = tool_with_model_visible_input_schema(&tool); + + assert_eq!( + *tool.input_schema, + serde_json::json!({ + "type": "object", + "properties": { + "file": { + "type": "string", + "description": "Original file payload. This parameter expects an absolute local file path. If you want to upload a file, provide the absolute path to that file here." + }, + "files": { + "type": "array", + "items": {"type": "string"}, + "description": "This parameter expects an absolute local file path. If you want to upload a file, provide the absolute path to that file here." + } + } + }) + .as_object() + .expect("object") + .clone() + ); +} + +#[test] +fn tool_with_model_visible_input_schema_leaves_tools_without_file_params_unchanged() { + let original_tool = create_test_tool("custom", "upload").tool; + + let tool = tool_with_model_visible_input_schema(&original_tool); + + assert_eq!(tool, original_tool); +} + #[test] fn elicitation_granular_policy_defaults_to_prompting() { assert!(!elicitation_is_rejected_by_policy( diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 54b9d524b9..6375f2a83a 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -33,7 +33,7 @@ use crate::realtime_conversation::handle_close as handle_realtime_conversation_c use crate::realtime_conversation::handle_start as handle_realtime_conversation_start; use crate::realtime_conversation::handle_text as handle_realtime_conversation_text; use crate::render_skills_section; -use crate::rollout::session_index; +use crate::rollout::find_thread_name_by_id; use crate::session_prefix::format_subagent_notification_message; use crate::skills_load_input_from_config; use crate::stream_events_utils::HandleOutputCtx; @@ -1096,6 +1096,26 @@ fn local_time_context() -> (String, String) { } } +async fn thread_title_from_state_db( + state_db: Option<&state_db::StateDbHandle>, + codex_home: &Path, + conversation_id: ThreadId, +) -> Option { + if let Some(metadata) = state_db + && let Some(metadata) = metadata.get_thread(conversation_id).await.ok().flatten() + { + let title = metadata.title.trim(); + if !title.is_empty() && metadata.first_user_message.as_deref().map(str::trim) != Some(title) + { + return Some(title.to_string()); + } + } + find_thread_name_by_id(codex_home, &conversation_id) + .await + .ok() + .flatten() +} + #[derive(Clone)] pub(crate) struct SessionConfiguration { /// Provider identifier ("openai", "openrouter", ...). @@ -1882,19 +1902,12 @@ impl Session { tx }; let thread_name = - match session_index::find_thread_name_by_id(&config.codex_home, &conversation_id) + thread_title_from_state_db(state_db_ctx.as_ref(), &config.codex_home, conversation_id) .instrument(info_span!( "session_init.thread_name_lookup", otel.name = "session_init.thread_name_lookup", )) - .await - { - Ok(name) => name, - Err(err) => { - warn!("Failed to read session index for thread name: {err}"); - None - } - }; + .await; session_configuration.thread_name = thread_name.clone(); let state = SessionState::new(session_configuration.clone()); let managed_network_requirements_enabled = config.managed_network_requirements_enabled(); @@ -4848,7 +4861,6 @@ mod handlers { use crate::review_prompts::resolve_review_request; use crate::rollout::RolloutRecorder; - use crate::rollout::session_index; use crate::tasks::CompactTask; use crate::tasks::UndoTask; use crate::tasks::UserShellCommandMode; @@ -5528,13 +5540,25 @@ mod handlers { .await; } - /// Persists the thread name in the session index, updates in-memory state, and emits - /// a `ThreadNameUpdated` event on success. - /// - /// This appends the name to `CODEX_HOME/sessions_index.jsonl` via `session_index::append_thread_name` for the - /// current `thread_id`, then updates `SessionConfiguration::thread_name`. - /// - /// Returns an error event if the name is empty or session persistence is disabled. + async fn persist_thread_name_update( + sess: &Arc, + event: ThreadNameUpdatedEvent, + ) -> anyhow::Result { + let msg = EventMsg::ThreadNameUpdated(event); + let item = RolloutItem::EventMsg(msg.clone()); + let recorder = { + let guard = sess.services.rollout.lock().await; + guard.clone() + } + .ok_or_else(|| anyhow::anyhow!("Session persistence is disabled; cannot rename thread."))?; + recorder.persist().await?; + recorder.record_items(std::slice::from_ref(&item)).await?; + recorder.flush().await?; + Ok(msg) + } + + /// Persists the thread name in the rollout and state database, updates in-memory state, and + /// emits a `ThreadNameUpdated` event on success. pub async fn set_thread_name(sess: &Arc, sub_id: String, name: String) { let Some(name) = crate::util::normalize_thread_name(&name) else { let event = Event { @@ -5548,47 +5572,33 @@ mod handlers { return; }; - let persistence_enabled = { - let rollout = sess.services.rollout.lock().await; - rollout.is_some() - }; - if !persistence_enabled { - let event = Event { - id: sub_id, - msg: EventMsg::Error(ErrorEvent { - message: "Session persistence is disabled; cannot rename thread.".to_string(), - codex_error_info: Some(CodexErrorInfo::Other), - }), - }; - sess.send_event_raw(event).await; - return; + let updated = ThreadNameUpdatedEvent { + thread_id: sess.conversation_id, + thread_name: Some(name.clone()), }; - if let Err(e) = sess.try_ensure_rollout_materialized().await { - let event = Event { - id: sub_id, - msg: EventMsg::Error(ErrorEvent { - message: format!("Failed to set thread name: {e}"), - codex_error_info: Some(CodexErrorInfo::Other), - }), - }; - sess.send_event_raw(event).await; - return; - } + let msg = match persist_thread_name_update(sess, updated).await { + Ok(msg) => msg, + Err(err) => { + warn!("Failed to persist thread name update to rollout: {err}"); + let event = Event { + id: sub_id, + msg: EventMsg::Error(ErrorEvent { + message: err.to_string(), + codex_error_info: Some(CodexErrorInfo::Other), + }), + }; + sess.send_event_raw(event).await; + return; + } + }; - let codex_home = sess.codex_home().await; - if let Err(e) = - session_index::append_thread_name(&codex_home, sess.conversation_id, &name).await + if let Some(state_db) = sess.services.state_db.as_deref() + && let Err(err) = state_db + .update_thread_title(sess.conversation_id, &name) + .await { - let event = Event { - id: sub_id, - msg: EventMsg::Error(ErrorEvent { - message: format!("Failed to set thread name: {e}"), - codex_error_info: Some(CodexErrorInfo::Other), - }), - }; - sess.send_event_raw(event).await; - return; + warn!("Failed to update thread title in state db: {err}"); } { @@ -5596,14 +5606,14 @@ mod handlers { state.session_configuration.thread_name = Some(name.clone()); } - sess.send_event_raw(Event { - id: sub_id, - msg: EventMsg::ThreadNameUpdated(ThreadNameUpdatedEvent { - thread_id: sess.conversation_id, - thread_name: Some(name), - }), - }) - .await; + let codex_home = sess.codex_home().await; + if let Err(err) = + crate::rollout::append_thread_name(&codex_home, sess.conversation_id, &name).await + { + warn!("Failed to update legacy thread name index: {err}"); + } + + sess.deliver_event_raw(Event { id: sub_id, msg }).await; } pub async fn shutdown(sess: &Arc, sub_id: String) -> bool { diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index efae8ae918..0d7b161191 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -56,6 +56,7 @@ mod original_image_detail; pub use codex_mcp::MCP_SANDBOX_STATE_CAPABILITY; pub use codex_mcp::MCP_SANDBOX_STATE_METHOD; pub use codex_mcp::SandboxState; +mod mcp_openai_file; mod mcp_tool_call; mod memories; pub(crate) mod mention_syntax; @@ -165,10 +166,10 @@ pub use rollout::append_thread_name; pub use rollout::find_archived_thread_path_by_id_str; #[deprecated(note = "use find_thread_path_by_id_str")] pub use rollout::find_conversation_path_by_id_str; +pub use rollout::find_thread_meta_by_name_str; pub use rollout::find_thread_name_by_id; pub use rollout::find_thread_names_by_ids; pub use rollout::find_thread_path_by_id_str; -pub use rollout::find_thread_path_by_name_str; pub use rollout::parse_cursor; pub use rollout::read_head_for_summary; pub use rollout::read_session_meta_line; diff --git a/codex-rs/core/src/mcp_openai_file.rs b/codex-rs/core/src/mcp_openai_file.rs new file mode 100644 index 0000000000..587dd4b770 --- /dev/null +++ b/codex-rs/core/src/mcp_openai_file.rs @@ -0,0 +1,473 @@ +//! Bridges Apps SDK-style `openai/fileParams` metadata into Codex's MCP flow. +//! +//! Strategy: +//! - Inspect `_meta["openai/fileParams"]` to discover which tool arguments are +//! file inputs. +//! - At tool execution time, upload those local files to OpenAI file storage +//! and rewrite only the declared arguments into the provided-file payload +//! shape expected by the downstream Apps tool. +//! +//! Model-visible schema masking is owned by `codex-mcp` alongside MCP tool +//! inventory, so this module only handles the execution-time argument rewrite. + +use crate::codex::Session; +use crate::codex::TurnContext; +use codex_api::CoreAuthProvider; +use codex_api::upload_local_file; +use codex_login::CodexAuth; +use serde_json::Value as JsonValue; + +pub(crate) async fn rewrite_mcp_tool_arguments_for_openai_files( + sess: &Session, + turn_context: &TurnContext, + arguments_value: Option, + openai_file_input_params: Option<&[String]>, +) -> Result, String> { + let Some(openai_file_input_params) = openai_file_input_params else { + return Ok(arguments_value); + }; + + let Some(arguments_value) = arguments_value else { + return Ok(None); + }; + let Some(arguments) = arguments_value.as_object() else { + return Ok(Some(arguments_value)); + }; + let auth = sess.services.auth_manager.auth().await; + let mut rewritten_arguments = arguments.clone(); + + for field_name in openai_file_input_params { + 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? + else { + continue; + }; + rewritten_arguments.insert(field_name.clone(), uploaded_value); + } + + if rewritten_arguments == *arguments { + return Ok(Some(arguments_value)); + } + + Ok(Some(JsonValue::Object(rewritten_arguments))) +} + +async fn rewrite_argument_value_for_openai_files( + turn_context: &TurnContext, + auth: Option<&CodexAuth>, + field_name: &str, + value: &JsonValue, +) -> Result, String> { + match value { + JsonValue::String(path_or_file_ref) => { + let rewritten = build_uploaded_local_argument_value( + turn_context, + auth, + field_name, + /*index*/ None, + path_or_file_ref, + ) + .await?; + Ok(Some(rewritten)) + } + JsonValue::Array(values) => { + let mut rewritten_values = Vec::with_capacity(values.len()); + for (index, item) in values.iter().enumerate() { + let Some(path_or_file_ref) = item.as_str() else { + return Ok(None); + }; + let rewritten = build_uploaded_local_argument_value( + turn_context, + auth, + field_name, + Some(index), + path_or_file_ref, + ) + .await?; + rewritten_values.push(rewritten); + } + Ok(Some(JsonValue::Array(rewritten_values))) + } + _ => Ok(None), + } +} + +async fn build_uploaded_local_argument_value( + turn_context: &TurnContext, + auth: Option<&CodexAuth>, + field_name: &str, + index: Option, + file_path: &str, +) -> Result { + let resolved_path = turn_context.resolve_path(Some(file_path.to_string())); + let Some(auth) = auth else { + return Err( + "ChatGPT auth is required to upload local files for Codex Apps tools".to_string(), + ); + }; + let token_data = auth + .get_token_data() + .map_err(|error| format!("failed to read ChatGPT auth for file upload: {error}"))?; + let upload_auth = CoreAuthProvider { + token: Some(token_data.access_token), + account_id: token_data.account_id, + }; + let uploaded = upload_local_file( + turn_context.config.chatgpt_base_url.trim_end_matches('/'), + &upload_auth, + &resolved_path, + ) + .await + .map_err(|error| match index { + Some(index) => { + format!("failed to upload `{file_path}` for `{field_name}[{index}]`: {error}") + } + 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, + })) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::codex::make_session_and_context; + use codex_utils_absolute_path::AbsolutePathBuf; + use pretty_assertions::assert_eq; + use std::sync::Arc; + use tempfile::tempdir; + + #[tokio::test] + async fn openai_file_argument_rewrite_requires_declared_file_params() { + let (session, turn_context) = make_session_and_context().await; + let arguments = Some(serde_json::json!({ + "file": "/tmp/codex-smoke-file.txt" + })); + + let rewritten = rewrite_mcp_tool_arguments_for_openai_files( + &session, + &Arc::new(turn_context), + arguments.clone(), + /*openai_file_input_params*/ None, + ) + .await + .expect("rewrite should succeed"); + + assert_eq!(rewritten, arguments); + } + + #[tokio::test] + async fn build_uploaded_local_argument_value_uploads_local_file_path() { + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::ResponseTemplate; + use wiremock::matchers::body_json; + use wiremock::matchers::header; + use wiremock::matchers::method; + use wiremock::matchers::path; + + 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": "file_report.csv", + "file_size": 5, + "use_case": "codex", + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "file_id": "file_123", + "upload_url": format!("{}/upload/file_123", server.uri()), + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path("/upload/file_123")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/backend-api/files/file_123/uploaded")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "status": "success", + "download_url": format!("{}/download/file_123", server.uri()), + "file_name": "file_report.csv", + "mime_type": "text/csv", + "file_size_bytes": 5, + }))) + .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"); + let local_path = dir.path().join("file_report.csv"); + tokio::fs::write(&local_path, b"hello") + .await + .expect("write local file"); + turn_context.cwd = AbsolutePathBuf::try_from(dir.path()).expect("absolute 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_local_argument_value( + &turn_context, + Some(&auth), + "file", + /*index*/ None, + "file_report.csv", + ) + .await + .expect("rewrite should upload the local file"); + + assert_eq!( + rewritten, + serde_json::json!({ + "download_url": format!("{}/download/file_123", server.uri()), + "file_id": "file_123", + "mime_type": "text/csv", + "file_name": "file_report.csv", + "uri": "sediment://file_123", + "file_size_bytes": 5, + }) + ); + } + + #[tokio::test] + async fn rewrite_argument_value_for_openai_files_rewrites_scalar_path() { + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::ResponseTemplate; + use wiremock::matchers::body_json; + use wiremock::matchers::header; + use wiremock::matchers::method; + use wiremock::matchers::path; + + 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": "file_report.csv", + "file_size": 5, + "use_case": "codex", + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "file_id": "file_123", + "upload_url": format!("{}/upload/file_123", server.uri()), + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path("/upload/file_123")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/backend-api/files/file_123/uploaded")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "status": "success", + "download_url": format!("{}/download/file_123", server.uri()), + "file_name": "file_report.csv", + "mime_type": "text/csv", + "file_size_bytes": 5, + }))) + .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"); + let local_path = dir.path().join("file_report.csv"); + tokio::fs::write(&local_path, b"hello") + .await + .expect("write local file"); + turn_context.cwd = AbsolutePathBuf::try_from(dir.path()).expect("absolute 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 = rewrite_argument_value_for_openai_files( + &turn_context, + Some(&auth), + "file", + &serde_json::json!("file_report.csv"), + ) + .await + .expect("rewrite should succeed"); + + assert_eq!( + rewritten, + Some(serde_json::json!({ + "download_url": format!("{}/download/file_123", server.uri()), + "file_id": "file_123", + "mime_type": "text/csv", + "file_name": "file_report.csv", + "uri": "sediment://file_123", + "file_size_bytes": 5, + })) + ); + } + + #[tokio::test] + async fn rewrite_argument_value_for_openai_files_rewrites_array_paths() { + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::ResponseTemplate; + use wiremock::matchers::body_json; + use wiremock::matchers::header; + use wiremock::matchers::method; + use wiremock::matchers::path; + + 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": "one.csv", + "file_size": 3, + "use_case": "codex", + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "file_id": "file_1", + "upload_url": format!("{}/upload/file_1", server.uri()), + }))) + .expect(1) + .mount(&server) + .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": "two.csv", + "file_size": 3, + "use_case": "codex", + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "file_id": "file_2", + "upload_url": format!("{}/upload/file_2", server.uri()), + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path("/upload/file_1")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .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_1/uploaded")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "status": "success", + "download_url": format!("{}/download/file_1", server.uri()), + "file_name": "one.csv", + "mime_type": "text/csv", + "file_size_bytes": 3, + }))) + .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": "two.csv", + "mime_type": "text/csv", + "file_size_bytes": 3, + }))) + .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("one.csv"), b"one") + .await + .expect("write first local file"); + tokio::fs::write(dir.path().join("two.csv"), b"two") + .await + .expect("write second local file"); + turn_context.cwd = AbsolutePathBuf::try_from(dir.path()).expect("absolute 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 = rewrite_argument_value_for_openai_files( + &turn_context, + Some(&auth), + "files", + &serde_json::json!(["one.csv", "two.csv"]), + ) + .await + .expect("rewrite should succeed"); + + assert_eq!( + rewritten, + Some(serde_json::json!([ + { + "download_url": format!("{}/download/file_1", server.uri()), + "file_id": "file_1", + "mime_type": "text/csv", + "file_name": "one.csv", + "uri": "sediment://file_1", + "file_size_bytes": 3, + }, + { + "download_url": format!("{}/download/file_2", server.uri()), + "file_id": "file_2", + "mime_type": "text/csv", + "file_name": "two.csv", + "uri": "sediment://file_2", + "file_size_bytes": 3, + } + ])) + ); + } + + #[tokio::test] + async fn rewrite_mcp_tool_arguments_for_openai_files_surfaces_upload_failures() { + let (mut session, turn_context) = make_session_and_context().await; + session.services.auth_manager = crate::test_support::auth_manager_from_auth( + CodexAuth::create_dummy_chatgpt_auth_for_testing(), + ); + let error = rewrite_mcp_tool_arguments_for_openai_files( + &session, + &turn_context, + Some(serde_json::json!({ + "file": "/definitely/missing/file.csv", + })), + Some(&["file".to_string()]), + ) + .await + .expect_err("missing file should fail"); + + assert!(error.contains("failed to upload")); + assert!(error.contains("file")); + } +} diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index 35b15eaa19..5f8b489428 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -26,6 +26,7 @@ use crate::guardian::guardian_approval_request_to_json; use crate::guardian::guardian_rejection_message; use crate::guardian::review_approval_request; use crate::guardian::routes_approval_to_guardian; +use crate::mcp_openai_file::rewrite_mcp_tool_arguments_for_openai_files; use crate::mcp_tool_approval_templates::RenderedMcpToolApprovalParam; use crate::mcp_tool_approval_templates::render_mcp_tool_approval_template; use codex_analytics::AppInvocation; @@ -34,6 +35,7 @@ use codex_analytics::build_track_events_context; use codex_config::types::AppToolApproval; use codex_features::Feature; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; +use codex_mcp::declared_openai_file_input_param_names; use codex_otel::sanitize_metric_tag_value; use codex_protocol::mcp::CallToolResult; use codex_protocol::openai_models::InputModality; @@ -178,14 +180,16 @@ pub(crate) async fn handle_mcp_tool_call( let start = Instant::now(); let result = async { - sess.call_tool( + execute_mcp_tool_call( + sess.as_ref(), + turn_context.as_ref(), &server, &tool_name, arguments_value.clone(), + metadata.as_ref(), request_meta.clone(), ) .await - .map_err(|e| format!("tool call error: {e:?}")) } .instrument(mcp_tool_call_span( sess.as_ref(), @@ -200,13 +204,6 @@ pub(crate) async fn handle_mcp_tool_call( }, )) .await; - let result = sanitize_mcp_tool_result_for_model( - turn_context - .model_info - .input_modalities - .contains(&InputModality::Image), - result, - ); if let Err(error) = &result { tracing::warn!("MCP tool call error: {error:?}"); } @@ -294,11 +291,17 @@ pub(crate) async fn handle_mcp_tool_call( maybe_mark_thread_memory_mode_polluted(sess.as_ref(), turn_context.as_ref()).await; let start = Instant::now(); - // Perform the tool call. let result = async { - sess.call_tool(&server, &tool_name, arguments_value.clone(), request_meta) - .await - .map_err(|e| format!("tool call error: {e:?}")) + execute_mcp_tool_call( + sess.as_ref(), + turn_context.as_ref(), + &server, + &tool_name, + arguments_value.clone(), + metadata.as_ref(), + request_meta, + ) + .await } .instrument(mcp_tool_call_span( sess.as_ref(), @@ -313,13 +316,6 @@ pub(crate) async fn handle_mcp_tool_call( }, )) .await; - let result = sanitize_mcp_tool_result_for_model( - turn_context - .model_info - .input_modalities - .contains(&InputModality::Image), - result, - ); if let Err(error) = &result { tracing::warn!("MCP tool call error: {error:?}"); } @@ -453,6 +449,35 @@ fn record_server_fields(span: &Span, url: Option<&str>) { } } +async fn execute_mcp_tool_call( + sess: &Session, + turn_context: &TurnContext, + server: &str, + tool_name: &str, + arguments_value: Option, + metadata: Option<&McpToolApprovalMetadata>, + request_meta: Option, +) -> Result { + 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()), + ) + .await?; + let result = sess + .call_tool(server, tool_name, rewritten_arguments, request_meta) + .await + .map_err(|e| format!("tool call error: {e:?}"))?; + sanitize_mcp_tool_result_for_model( + turn_context + .model_info + .input_modalities + .contains(&InputModality::Image), + Ok(result), + ) +} + async fn maybe_mark_thread_memory_mode_polluted(sess: &Session, turn_context: &TurnContext) { if !turn_context .config @@ -566,6 +591,7 @@ pub(crate) struct McpToolApprovalMetadata { tool_title: Option, tool_description: Option, codex_apps_meta: Option>, + openai_file_input_params: Option>, } const MCP_TOOL_CODEX_APPS_META_KEY: &str = "_codex_apps"; @@ -980,7 +1006,6 @@ pub(crate) async fn lookup_mcp_tool_metadata( .await .list_all_tools() .await; - let tool_info = tools .into_values() .find(|tool_info| tool_info.server_name == server && tool_info.tool.name == tool_name)?; @@ -1022,6 +1047,10 @@ pub(crate) async fn lookup_mcp_tool_metadata( .and_then(|meta| meta.get(MCP_TOOL_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()), }) } diff --git a/codex-rs/core/src/mcp_tool_call_tests.rs b/codex-rs/core/src/mcp_tool_call_tests.rs index 9e89e0d4b2..6815c46421 100644 --- a/codex-rs/core/src/mcp_tool_call_tests.rs +++ b/codex-rs/core/src/mcp_tool_call_tests.rs @@ -59,6 +59,7 @@ fn approval_metadata( tool_title: tool_title.map(str::to_string), tool_description: tool_description.map(str::to_string), codex_apps_meta: None, + openai_file_input_params: None, } } @@ -597,6 +598,7 @@ async fn codex_apps_tool_call_request_meta_includes_turn_metadata_and_codex_apps .cloned() .expect("_codex_apps metadata should be an object"), ), + openai_file_input_params: None, }; assert_eq!( @@ -744,6 +746,7 @@ fn guardian_mcp_review_request_includes_annotations_when_present() { tool_title: None, tool_description: None, codex_apps_meta: None, + openai_file_input_params: None, }; let request = build_guardian_mcp_tool_review_request("call-1", &invocation, Some(&metadata)); @@ -1254,6 +1257,7 @@ async fn approve_mode_skips_when_annotations_do_not_require_approval() { tool_title: Some("Read Only Tool".to_string()), tool_description: None, codex_apps_meta: None, + openai_file_input_params: None, }; let decision = maybe_request_mcp_tool_approval( @@ -1321,6 +1325,7 @@ async fn guardian_mode_skips_auto_when_annotations_do_not_require_approval() { tool_title: Some("Read Only Tool".to_string()), tool_description: None, codex_apps_meta: None, + openai_file_input_params: None, }; let decision = maybe_request_mcp_tool_approval( @@ -1391,6 +1396,7 @@ async fn guardian_mode_mcp_denial_returns_rationale_message() { tool_title: Some("Dangerous Tool".to_string()), tool_description: Some("Reads calendar data.".to_string()), codex_apps_meta: None, + openai_file_input_params: None, }; let decision = maybe_request_mcp_tool_approval( @@ -1441,6 +1447,7 @@ async fn prompt_mode_waits_for_approval_when_annotations_do_not_require_approval tool_title: Some("Read Only Tool".to_string()), tool_description: None, codex_apps_meta: None, + openai_file_input_params: None, }; let mut approval_task = { @@ -1517,6 +1524,7 @@ async fn approve_mode_blocks_when_arc_returns_interrupt_for_model() { tool_title: Some("Dangerous Tool".to_string()), tool_description: Some("Performs a risky action.".to_string()), codex_apps_meta: None, + openai_file_input_params: None, }; let decision = maybe_request_mcp_tool_approval( @@ -1586,6 +1594,7 @@ async fn custom_approve_mode_blocks_when_arc_returns_interrupt_for_model() { tool_title: Some("Dangerous Tool".to_string()), tool_description: Some("Performs a risky action.".to_string()), codex_apps_meta: None, + openai_file_input_params: None, }; let decision = maybe_request_mcp_tool_approval( @@ -1655,6 +1664,7 @@ async fn approve_mode_blocks_when_arc_returns_interrupt_without_annotations() { tool_title: Some("Dangerous Tool".to_string()), tool_description: Some("Performs a risky action.".to_string()), codex_apps_meta: None, + openai_file_input_params: None, }; let decision = maybe_request_mcp_tool_approval( @@ -1732,6 +1742,7 @@ async fn never_policy_approves_mcp_tool_calls_for_all_approval_modes() { tool_title: Some("Dangerous Tool".to_string()), tool_description: Some("Performs a risky action.".to_string()), codex_apps_meta: None, + openai_file_input_params: None, }; for approval_mode in [ @@ -1833,6 +1844,7 @@ async fn approve_mode_routes_arc_ask_user_to_guardian_when_guardian_reviewer_is_ tool_title: Some("Dangerous Tool".to_string()), tool_description: Some("Performs a risky action.".to_string()), codex_apps_meta: None, + openai_file_input_params: None, }; let decision = maybe_request_mcp_tool_approval( diff --git a/codex-rs/core/src/memories/mod.rs b/codex-rs/core/src/memories/mod.rs index ac49c640ca..194c27c907 100644 --- a/codex-rs/core/src/memories/mod.rs +++ b/codex-rs/core/src/memories/mod.rs @@ -25,6 +25,7 @@ pub(crate) use control::clear_memory_root_contents; pub(crate) use start::start_memories_startup_task; mod artifacts { + pub(super) const EXTENSIONS_SUBDIR: &str = "memories_extensions"; pub(super) const ROLLOUT_SUMMARIES_SUBDIR: &str = "rollout_summaries"; pub(super) const RAW_MEMORIES_FILENAME: &str = "raw_memories.md"; } @@ -106,6 +107,10 @@ fn rollout_summaries_dir(root: &Path) -> PathBuf { root.join(artifacts::ROLLOUT_SUMMARIES_SUBDIR) } +fn memory_extensions_root(root: &Path) -> PathBuf { + root.with_file_name(artifacts::EXTENSIONS_SUBDIR) +} + fn raw_memories_file(root: &Path) -> PathBuf { root.join(artifacts::RAW_MEMORIES_FILENAME) } diff --git a/codex-rs/core/src/memories/prompts.rs b/codex-rs/core/src/memories/prompts.rs index 56cb2b7b60..079ccd5c6a 100644 --- a/codex-rs/core/src/memories/prompts.rs +++ b/codex-rs/core/src/memories/prompts.rs @@ -1,3 +1,4 @@ +use crate::memories::memory_extensions_root; use crate::memories::memory_root; use crate::memories::phase_one; use crate::memories::storage::rollout_summary_file_stem_from_parts; @@ -31,6 +32,18 @@ static MEMORY_TOOL_DEVELOPER_INSTRUCTIONS_TEMPLATE: LazyLock