mirror of
https://github.com/openai/codex.git
synced 2026-10-19 17:28:46 +00:00
Allow metadata updates without materializing threads (#38413)
## Why Thread stores may filter a metadata patch to a no-op. Requiring every successful update to return a `StoredThread` forces those implementations to materialize a thread even when the caller does not need one. ## What changed - Let `ThreadStore::update_thread_metadata` return `None` after a successful update that did not materialize a thread. - Preserve the materialized-thread contract of `LiveThread` and `ThreadManager` metadata updates by reading the thread when the store returns `None`. - Keep completion-only metadata updates from issuing that fallback read. ## Testing Cover materialized in-memory updates and verify that fallback reads occur only for callers that require the updated thread. GitOrigin-RevId: 3059efc9d5b69e0b46e35fb5b5aa9638086654f2
This commit is contained in:
committed by
copyberry
parent
d167a3604c
commit
42bb50d502
@@ -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.
|
||||
|
||||
@@ -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::<InMemoryThreadStore>()
|
||||
.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;
|
||||
|
||||
@@ -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<InMemoryThreadStoreState>,
|
||||
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<StoredThread>> {
|
||||
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(
|
||||
|
||||
@@ -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<StoredThread> {
|
||||
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.
|
||||
|
||||
@@ -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<StoredThread>> {
|
||||
Box::pin(async move {
|
||||
update_thread_metadata::update_thread_metadata(self, params)
|
||||
.await
|
||||
.map(Some)
|
||||
})
|
||||
}
|
||||
|
||||
fn move_thread_to_section(
|
||||
|
||||
@@ -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!(
|
||||
|
||||
@@ -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<StoredThread>>;
|
||||
|
||||
/// Moves a thread to, within, or out of a server-ordered section.
|
||||
fn move_thread_to_section(
|
||||
|
||||
Reference in New Issue
Block a user