diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 1a6a9a715f..3e93007bab 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -759,7 +759,8 @@ impl ThreadManager { /// /// Loaded threads route through `CodexThread`/`LiveThread`, so metadata changes stay ordered /// with live rollout writes. Cold threads go directly to the store, which owns unloaded JSONL - /// compatibility and SQLite metadata updates. + /// compatibility and SQLite metadata updates. This API always returns a materialized thread; + /// if the store reports a successful no-op without one, it performs a fallback read. pub async fn update_thread_metadata( &self, thread_id: ThreadId, @@ -777,7 +778,8 @@ impl ThreadManager { .await .map_err(|err| thread_store_metadata_update_error(thread_id, err)); } - self.state + let updated = self + .state .thread_store .update_thread_metadata(UpdateThreadMetadataParams { thread_id, @@ -790,7 +792,20 @@ impl ThreadManager { CodexErr::ThreadNotFound(thread_id) } err => thread_store_metadata_update_error(thread_id, err), - }) + })?; + match updated { + Some(thread) => Ok(thread), + None => self + .state + .thread_store + .read_thread(ReadThreadParams { + thread_id, + include_archived, + include_history: false, + }) + .await + .map_err(|err| thread_store_metadata_update_error(thread_id, err)), + } } /// Moves a persisted thread to, within, or out of a server-ordered section. diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 6589ae3d68..4e9023822b 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -1651,6 +1651,135 @@ async fn rollout_path_resume_and_fork_read_history_through_thread_store() { .expect("shutdown forked thread"); } +#[tokio::test] +async fn metadata_update_without_result_reads_only_when_the_caller_needs_the_thread() { + let temp_dir = tempdir().expect("tempdir"); + let mut config = test_config().await; + config.codex_home = temp_dir.path().join("codex-home").abs(); + config.cwd = config.codex_home.abs(); + config.experimental_thread_store = ThreadStoreConfig::InMemory { + id: format!("metadata-update-none-{}", uuid::Uuid::new_v4()), + }; + std::fs::create_dir_all(&config.codex_home).expect("create codex home"); + + let auth_manager = + AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); + let thread_store = thread_store_from_config(&config, /*state_db*/ None); + let in_memory_store = thread_store + .as_any() + .downcast_ref::() + .expect("configured in-memory store"); + let manager = ThreadManager::new( + &config, + auth_manager.clone(), + build_models_manager(&config, auth_manager), + crate::CodexAppsToolsCache::default(), + SessionSource::Exec, + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + empty_extension_registry(), + Arc::new(crate::test_support::EmptyUserInstructionsProvider), + /*analytics_events_client*/ None, + thread_store.clone(), + /*agent_graph_store*/ None, + TEST_INSTALLATION_ID.to_string(), + /*attestation_provider*/ None, + /*external_time_provider*/ None, + ); + let started = manager + .start_thread(StartThreadOptions::new(config)) + .await + .expect("start thread"); + started + .thread + .flush_rollout() + .await + .expect("flush initial metadata"); + manager + .update_thread_metadata( + started.thread_id, + ThreadMetadataPatch { + name: Some(Some("initial name".to_string())), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await + .expect("flush pending live metadata before measuring calls"); + in_memory_store.omit_metadata_update_result_for_testing(); + + let before_loaded_update = in_memory_store.calls().await; + let loaded = manager + .update_thread_metadata( + started.thread_id, + ThreadMetadataPatch { + name: Some(Some("loaded name".to_string())), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await + .expect("update loaded thread metadata"); + assert_eq!(loaded.name.as_deref(), Some("loaded name")); + let after_loaded_update = in_memory_store.calls().await; + assert_eq!( + after_loaded_update.update_thread_metadata, + before_loaded_update.update_thread_metadata + 1 + ); + assert_eq!( + after_loaded_update.read_thread, + before_loaded_update.read_thread + 1 + ); + + started + .thread + .append_rollout_items(&[RolloutItem::EventMsg(EventMsg::UserMessage( + UserMessageEvent { + message: "completion-only metadata".to_string(), + ..Default::default() + }, + ))]) + .await + .expect("append item with derived metadata"); + let after_completion_only_update = in_memory_store.calls().await; + assert_eq!( + after_completion_only_update.update_thread_metadata, + after_loaded_update.update_thread_metadata + 1 + ); + assert_eq!( + after_completion_only_update.read_thread, + after_loaded_update.read_thread + ); + + started + .thread + .shutdown_and_wait() + .await + .expect("shutdown loaded thread"); + let _ = manager.remove_thread(&started.thread_id).await; + let before_cold_update = in_memory_store.calls().await; + let cold = manager + .update_thread_metadata( + started.thread_id, + ThreadMetadataPatch { + name: Some(Some("cold name".to_string())), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await + .expect("update cold thread metadata"); + assert_eq!(cold.name.as_deref(), Some("cold name")); + let after_cold_update = in_memory_store.calls().await; + assert_eq!( + after_cold_update.update_thread_metadata, + before_cold_update.update_thread_metadata + 1 + ); + assert_eq!( + after_cold_update.read_thread, + before_cold_update.read_thread + 1 + ); +} + #[tokio::test] async fn new_uses_active_provider_for_model_refresh() { let server = MockServer::start().await; diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index bd4c678484..6993e55daf 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -5,6 +5,8 @@ use std::sync::Arc; use std::sync::Mutex; use std::sync::MutexGuard; use std::sync::OnceLock; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; use chrono::DateTime; use chrono::Utc; @@ -383,6 +385,33 @@ mod tests { ); } + #[tokio::test] + async fn metadata_update_returns_the_materialized_thread() { + let store = InMemoryThreadStore::default(); + let thread_id = ThreadId::default(); + store + .create_thread(create_thread_params(thread_id, ThreadHistoryMode::Legacy)) + .await + .expect("create thread"); + + let updated = ThreadStore::update_thread_metadata( + &store, + UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + name: Some(Some("renamed".to_string())), + ..Default::default() + }, + include_archived: false, + }, + ) + .await + .expect("update metadata"); + let updated = updated.expect("in-memory store returns updated thread"); + assert_eq!(updated.thread_id, thread_id); + assert_eq!(updated.name.as_deref(), Some("renamed")); + } + fn create_thread_params( thread_id: ThreadId, history_mode: ThreadHistoryMode, @@ -463,6 +492,7 @@ pub struct InMemoryThreadStoreCalls { #[derive(Default)] pub struct InMemoryThreadStore { state: tokio::sync::Mutex, + omit_metadata_update_result: AtomicBool, } #[derive(Default)] @@ -499,6 +529,12 @@ impl InMemoryThreadStore { self.state.lock().await.calls.clone() } + /// Makes metadata updates apply normally while returning no materialized thread. + pub fn omit_metadata_update_result_for_testing(&self) { + self.omit_metadata_update_result + .store(true, Ordering::Relaxed); + } + async fn create_thread(&self, params: CreateThreadParams) -> ThreadStoreResult<()> { reject_paginated_history_mode(params.history_mode)?; let mut state = self.state.lock().await; @@ -930,8 +966,11 @@ impl ThreadStore for InMemoryThreadStore { fn update_thread_metadata( &self, params: UpdateThreadMetadataParams, - ) -> ThreadStoreFuture<'_, StoredThread> { - Box::pin(InMemoryThreadStore::update_thread_metadata(self, params)) + ) -> ThreadStoreFuture<'_, Option> { + Box::pin(async move { + let updated = InMemoryThreadStore::update_thread_metadata(self, params).await?; + Ok((!self.omit_metadata_update_result.load(Ordering::Relaxed)).then_some(updated)) + }) } fn move_thread_to_section( diff --git a/codex-rs/thread-store/src/live_thread.rs b/codex-rs/thread-store/src/live_thread.rs index dc9afcf9e7..2f043800f2 100644 --- a/codex-rs/thread-store/src/live_thread.rs +++ b/codex-rs/thread-store/src/live_thread.rs @@ -336,19 +336,31 @@ impl LiveThread { Ok(()) } + /// Updates metadata while preserving this API's materialized-thread contract. + /// + /// Stores may successfully return no thread for a no-op update, so this reads the thread as a + /// fallback in that case. pub async fn update_metadata( &self, patch: ThreadMetadataPatch, include_archived: bool, ) -> ThreadStoreResult { self.flush_pending_metadata_update().await?; - self.thread_store + let updated = self + .thread_store .update_thread_metadata(UpdateThreadMetadataParams { thread_id: self.thread_id, patch, include_archived, }) - .await + .await?; + match updated { + Some(thread) => Ok(thread), + None => { + self.read_thread(include_archived, /*include_history*/ false) + .await + } + } } /// Returns the live local rollout path for legacy local-only callers. diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 0512e9cfc3..e5c12dd11f 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -529,8 +529,12 @@ impl ThreadStore for LocalThreadStore { fn update_thread_metadata( &self, params: UpdateThreadMetadataParams, - ) -> ThreadStoreFuture<'_, StoredThread> { - Box::pin(async move { update_thread_metadata::update_thread_metadata(self, params).await }) + ) -> ThreadStoreFuture<'_, Option> { + Box::pin(async move { + update_thread_metadata::update_thread_metadata(self, params) + .await + .map(Some) + }) } fn move_thread_to_section( diff --git a/codex-rs/thread-store/src/local/update_thread_metadata.rs b/codex-rs/thread-store/src/local/update_thread_metadata.rs index 8062edeb3d..12195734fc 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -798,7 +798,8 @@ mod tests { include_archived: false, }) .await - .expect("set thread name"); + .expect("set thread name") + .expect("local store returns updated thread"); assert_eq!(thread.name.as_deref(), Some("A sharper name")); let latest_name = codex_rollout::find_thread_name_by_id(home.path(), &thread_id) @@ -986,7 +987,8 @@ mod tests { include_archived: false, }) .await - .expect("set paginated thread name"); + .expect("set paginated thread name") + .expect("local store returns updated thread"); assert_eq!(thread.name.as_deref(), Some("Canonical paginated name")); let metadata = runtime @@ -1015,7 +1017,8 @@ mod tests { include_archived: false, }) .await - .expect("apply derived paginated metadata"); + .expect("apply derived paginated metadata") + .expect("local store returns updated thread"); assert_eq!(thread.name.as_deref(), Some("Canonical paginated name")); let session_index_path = home.path().join("session_index.jsonl"); @@ -1031,7 +1034,8 @@ mod tests { include_archived: false, }) .await - .expect("set paginated thread name with unavailable index"); + .expect("set paginated thread name with unavailable index") + .expect("local store returns updated thread"); assert_eq!(thread.name.as_deref(), Some("Updated SQLite name")); let err = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None) @@ -1079,7 +1083,8 @@ mod tests { include_archived: false, }) .await - .expect("set thread memory mode"); + .expect("set thread memory mode") + .expect("local store returns updated thread"); assert_eq!(thread.thread_id, thread_id); let appended = last_rollout_item(path.as_path()); @@ -1149,7 +1154,8 @@ mod tests { include_archived: false, }) .await - .expect("paginated metadata update"); + .expect("paginated metadata update") + .expect("local store returns updated thread"); let git_info = thread.git_info.expect("git info"); assert_eq!(git_info.commit_hash, None); @@ -1237,7 +1243,8 @@ mod tests { include_archived: false, }) .await - .expect("set git metadata"); + .expect("set git metadata") + .expect("local store returns updated thread"); assert_eq!( thread.git_info.expect("git info").branch.as_deref(), @@ -1296,7 +1303,8 @@ mod tests { include_archived: false, }) .await - .expect("set memory mode on external live thread"); + .expect("set memory mode on external live thread") + .expect("local store returns updated thread"); assert_eq!(thread.thread_id, thread_id); assert!(thread.rollout_path.is_some()); @@ -1334,7 +1342,8 @@ mod tests { include_archived: false, }) .await - .expect("set git metadata"); + .expect("set git metadata") + .expect("local store returns updated thread"); let git_info = thread.git_info.expect("git info should be present"); assert_eq!( @@ -1374,7 +1383,8 @@ mod tests { include_archived: false, }) .await - .expect("set permission profile"); + .expect("set permission profile") + .expect("local store returns updated thread"); assert_eq!(thread.permission_profile, PermissionProfile::Disabled); let thread = store @@ -1387,7 +1397,8 @@ mod tests { include_archived: false, }) .await - .expect("clear reasoning effort"); + .expect("clear reasoning effort") + .expect("local store returns updated thread"); assert_eq!(thread.reasoning_effort, None); let metadata = runtime @@ -1447,7 +1458,8 @@ mod tests { include_archived: false, }) .await - .expect("partially update git metadata"); + .expect("partially update git metadata") + .expect("local store returns updated thread"); let git_info = thread.git_info.expect("git info should be present"); assert_eq!( @@ -1507,7 +1519,8 @@ mod tests { include_archived: false, }) .await - .expect("clear git metadata"); + .expect("clear git metadata") + .expect("local store returns updated thread"); assert!(thread.git_info.is_none()); let appended = last_rollout_item(path.as_path()); @@ -1588,7 +1601,8 @@ mod tests { include_archived: false, }) .await - .expect("partially update after clear with missing sqlite row"); + .expect("partially update after clear with missing sqlite row") + .expect("local store returns updated thread"); let git_info = thread.git_info.expect("branch should be present"); assert_eq!(git_info.commit_hash, None); assert_eq!(git_info.branch.as_deref(), Some("feature")); @@ -1695,7 +1709,8 @@ mod tests { include_archived: false, }) .await - .expect("combined patch should apply"); + .expect("combined patch should apply") + .expect("local store returns updated thread"); assert_eq!(thread.name.as_deref(), Some("Combined metadata")); assert_eq!( @@ -1795,7 +1810,8 @@ mod tests { include_archived: false, }) .await - .expect("apply observed metadata"); + .expect("apply observed metadata") + .expect("local store returns updated thread"); assert_eq!(thread.name.as_deref(), Some("Derived first message")); } @@ -1839,7 +1855,8 @@ mod tests { include_archived: false, }) .await - .expect("apply later observed metadata"); + .expect("apply later observed metadata") + .expect("local store returns updated thread"); assert_eq!(thread.preview, "Hello from user"); assert_eq!( @@ -1927,7 +1944,8 @@ mod tests { include_archived: false, }) .await - .expect("update paginated thread without sqlite row"); + .expect("update paginated thread without sqlite row") + .expect("local store returns updated thread"); assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated); assert_eq!( @@ -1967,7 +1985,8 @@ mod tests { include_archived: true, }) .await - .expect("update archived thread without sqlite row"); + .expect("update archived thread without sqlite row") + .expect("local store returns updated thread"); assert!(thread.archived_at.is_some()); assert!( @@ -2096,7 +2115,8 @@ mod tests { include_archived: true, }) .await - .expect("set archived thread name"); + .expect("set archived thread name") + .expect("local store returns updated thread"); assert!(thread.archived_at.is_some()); assert!( @@ -2160,7 +2180,8 @@ mod tests { include_archived: true, }) .await - .expect("set archived thread name"); + .expect("set archived thread name") + .expect("local store returns updated thread"); assert!(thread.archived_at.is_some()); assert!( diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index d0927b5132..90b53b38ed 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -262,14 +262,18 @@ pub trait ThreadStore: Any + Send + Sync { }) } - /// Applies a literal metadata patch and returns the updated thread. + /// Applies a literal metadata patch and returns the updated thread when one was materialized. + /// + /// `None` means the update succeeded without materializing a thread, for example because the + /// implementation filtered the patch to a no-op. Callers that require a `StoredThread` must + /// perform a fallback read. /// /// Implementations should apply the supplied fields directly. Policy such as deciding whether /// an append-derived preview should be emitted belongs above the store. fn update_thread_metadata( &self, params: UpdateThreadMetadataParams, - ) -> ThreadStoreFuture<'_, StoredThread>; + ) -> ThreadStoreFuture<'_, Option>; /// Moves a thread to, within, or out of a server-ordered section. fn move_thread_to_section(