agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[codex] Migrate thread turns list to thread store (#19280)

Browse Source 
- migrate `thread/turns/list` to ThreadStore. Uses ThreadStore for most
data now but merges in the in-memory state from thread manager
- keep v2 `thread/list` pathless-store friendly by converting
`StoredThread` directly to API `Thread`
- add regression coverage for pathless store history/listing
This commit is contained in:
Tom
2026-04-30 14:16:42 -07:00
committed by GitHub
parent 9121132c8f
commit 127be0612c
12 changed files with 713 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+223 -163
View File
@@ -268,7 +268,6 @@ use codex_core::exec::ExecParams;
use codex_core::exec_env::create_env;
use codex_core::find_archived_thread_path_by_id_str;
use codex_core::find_thread_name_by_id;
use codex_core::find_thread_names_by_ids;
use codex_core::find_thread_path_by_id_str;
use codex_core::path_utils;
use codex_core::plugins::PluginInstallError as CorePluginInstallError;
@@ -377,7 +376,6 @@ use codex_state::ThreadMetadata;
use codex_state::ThreadMetadataBuilder;
use codex_state::log_db::LogDbLayer;
use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams;
#[cfg(debug_assertions)]
use codex_thread_store::InMemoryThreadStore;
use codex_thread_store::ListThreadsParams as StoreListThreadsParams;
use codex_thread_store::LocalThreadStore;
@@ -699,7 +697,6 @@ fn thread_store_from_config(config: &Config) -> Arc<dyn ThreadStore> {
match &config.experimental_thread_store {
ThreadStoreConfig::Local => Arc::new(configured_local_thread_store(config)),
ThreadStoreConfig::Remote { endpoint } => Arc::new(RemoteThreadStore::new(endpoint)),
#[cfg(debug_assertions)]
ThreadStoreConfig::InMemory { id } => InMemoryThreadStore::for_id(id),
}
}
@@ -3760,7 +3757,7 @@ impl CodexMessageProcessor {
ThreadSortKey::UpdatedAt => StoreThreadSortKey::UpdatedAt,
};
let sort_direction = sort_direction.unwrap_or(SortDirection::Desc);
let (summaries, next_cursor) = self
let (stored_threads, next_cursor) = self
.list_threads_common(
requested_page_size,
cursor,
@@ -3776,24 +3773,23 @@ impl CodexMessageProcessor {
},
)
.await?;
let backwards_cursor = summaries.first().and_then(|summary| {
thread_backwards_cursor_for_sort_key(summary, store_sort_key, sort_direction)
let backwards_cursor = stored_threads.first().and_then(|thread| {
thread_backwards_cursor_for_sort_key(thread, store_sort_key, sort_direction)
});
let mut threads = Vec::with_capacity(summaries.len());
let mut thread_ids = HashSet::with_capacity(summaries.len());
let mut status_ids = Vec::with_capacity(summaries.len());
let mut threads = Vec::with_capacity(stored_threads.len());
let mut status_ids = Vec::with_capacity(stored_threads.len());
let fallback_provider = self.config.model_provider_id.clone();
for summary in summaries {
let conversation_id = summary.conversation_id;
thread_ids.insert(conversation_id);
let thread = summary_to_thread(summary, &self.config.cwd);
for stored_thread in stored_threads {
let (thread, _) = thread_from_stored_thread(
stored_thread,
fallback_provider.as_str(),
&self.config.cwd,
);
status_ids.push(thread.id.clone());
threads.push((conversation_id, thread));
threads.push(thread);
}
let names = thread_titles_by_ids(&self.config, &thread_ids).await;
let statuses = self
.thread_watch_manager
.loaded_statuses_for_threads(status_ids)
@@ -3801,10 +3797,7 @@ impl CodexMessageProcessor {
let data: Vec<_> = threads
.into_iter()
.map(|(conversation_id, mut thread)| {
if let Some(title) = names.get(&conversation_id).cloned() {
set_thread_name_from_title(&mut thread, title);
}
.map(|mut thread| {
if let Some(status) = statuses.get(&thread.id) {
thread.status = status.clone();
}
@@ -3904,7 +3897,7 @@ impl CodexMessageProcessor {
thread_id: ThreadId,
include_turns: bool,
) -> Result<Thread, ThreadReadViewError> {
let loaded_thread = self.load_live_thread_for_read(thread_id).await;
let loaded_thread = self.thread_manager.get_thread(thread_id).await.ok();
let mut thread = if let Some(thread) = self
.load_persisted_thread_for_read(thread_id, include_turns)
.await?
@@ -3940,10 +3933,6 @@ impl CodexMessageProcessor {
Ok(thread)
}
async fn load_live_thread_for_read(&self, thread_id: ThreadId) -> Option<Arc<CodexThread>> {
self.thread_manager.get_thread(thread_id).await.ok()
}
async fn load_persisted_thread_for_read(
&self,
thread_id: ThreadId,
@@ -4071,87 +4060,37 @@ impl CodexMessageProcessor {
let thread_uuid = ThreadId::from_string(&thread_id)
.map_err(|err| invalid_request(format!("invalid thread id: {err}")))?;
let state_db_ctx = get_state_db(&self.config).await;
let mut rollout_path = self
.resolve_rollout_path(thread_uuid, state_db_ctx.as_ref())
.await;
if rollout_path.is_none() {
rollout_path =
match find_thread_path_by_id_str(&self.config.codex_home, &thread_uuid.to_string())
.await
{
Ok(Some(path)) => Some(path),
Ok(None) => match find_archived_thread_path_by_id_str(
&self.config.codex_home,
&thread_uuid.to_string(),
)
.await
{
Ok(path) => path,
Err(err) => {
return Err(invalid_request(format!(
"failed to locate archived thread id {thread_uuid}: {err}"
)));
}
},
Err(err) => {
return Err(invalid_request(format!(
"failed to locate thread id {thread_uuid}: {err}"
)));
}
};
}
if rollout_path.is_none() {
match self.thread_manager.get_thread(thread_uuid).await {
Ok(thread) => {
rollout_path = thread.rollout_path();
if rollout_path.is_none() {
return Err(invalid_request(
"ephemeral threads do not support thread/turns/list",
));
}
}
Err(_) => return Err(invalid_request(format!("thread not loaded: {thread_uuid}"))),
}
}
let Some(rollout_path) = rollout_path.as_ref() else {
return Err(internal_error(format!(
"failed to locate rollout for thread {thread_uuid}"
)));
};
let items = match read_rollout_items_from_rollout(rollout_path).await {
Ok(items) => items,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
return Err(invalid_request(format!(
"thread {thread_uuid} is not materialized yet; thread/turns/list is unavailable before first user message"
)));
}
Err(err) => {
return Err(internal_error(format!(
"failed to load rollout `{}` for thread {thread_uuid}: {err}",
rollout_path.display()
)));
}
};
let items = self
.load_thread_turns_list_history(thread_uuid)
.await
.map_err(thread_read_view_error)?;
// This API optimizes network transfer by letting clients page through a
// thread's turns incrementally, but it still replays the entire rollout on
// every request. Rollback and compaction events can change earlier turns, so
// the server has to rebuild the full turn list until turn metadata is indexed
// separately.
let has_live_in_progress_turn = match self.thread_manager.get_thread(thread_uuid).await {
Ok(thread) => matches!(thread.agent_status().await, AgentStatus::Running),
Err(_) => false,
let loaded_thread = self.thread_manager.get_thread(thread_uuid).await.ok();
let has_live_running_thread = match loaded_thread.as_ref() {
Some(thread) => matches!(thread.agent_status().await, AgentStatus::Running),
None => false,
};
let turns = reconstruct_thread_turns_from_rollout_items(
let active_turn = if loaded_thread.is_some() {
// Persisted history may not yet include the currently running turn. The
// app-server listener has already projected live turn events into ThreadState,
// so merge that in-memory snapshot before paginating.
let thread_state = self.thread_state_manager.thread_state(thread_uuid).await;
let state = thread_state.lock().await;
state.active_turn_snapshot()
} else {
None
};
let turns = reconstruct_thread_turns_for_turns_list(
&items,
self.thread_watch_manager
.loaded_status_for_thread(&thread_uuid.to_string())
.await,
has_live_in_progress_turn,
has_live_running_thread,
active_turn,
);
let page = paginate_thread_turns(
turns,
@@ -4166,6 +4105,63 @@ impl CodexMessageProcessor {
})
}
async fn load_thread_turns_list_history(
&self,
thread_id: ThreadId,
) -> Result<Vec<RolloutItem>, ThreadReadViewError> {
match self
.thread_store
.read_thread(StoreReadThreadParams {
thread_id,
include_archived: true,
include_history: true,
})
.await
{
Ok(stored_thread) => {
let history = stored_thread.history.ok_or_else(|| {
ThreadReadViewError::Internal(format!(
"thread store did not return history for thread {thread_id}"
))
})?;
return Ok(history.items);
}
Err(ThreadStoreError::InvalidRequest { message })
if message == format!("no rollout found for thread id {thread_id}") => {}
Err(ThreadStoreError::ThreadNotFound {
thread_id: missing_thread_id,
}) if missing_thread_id == thread_id => {}
Err(ThreadStoreError::InvalidRequest { message }) => {
return Err(ThreadReadViewError::InvalidRequest(message));
}
Err(err) => {
return Err(ThreadReadViewError::Internal(format!(
"failed to read thread: {err}"
)));
}
}
let thread = self
.thread_manager
.get_thread(thread_id)
.await
.map_err(|_| {
ThreadReadViewError::InvalidRequest(format!("thread not loaded: {thread_id}"))
})?;
let config_snapshot = thread.config_snapshot().await;
if config_snapshot.ephemeral {
return Err(ThreadReadViewError::InvalidRequest(
"ephemeral threads do not support thread/turns/list".to_string(),
));
}
thread
.load_history(/*include_archived*/ true)
.await
.map(|history| history.items)
.map_err(|err| thread_turns_list_history_load_error(thread_id, err))
}
pub(crate) fn thread_created_receiver(&self) -> broadcast::Receiver<ThreadId> {
self.thread_manager.subscribe_thread_created()
}
@@ -4722,12 +4718,11 @@ impl CodexMessageProcessor {
let (mut thread, history) =
thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd);
if include_turns && let Some(history) = history {
populate_thread_turns(
populate_thread_turns_from_history(
&mut thread,
ThreadTurnSource::HistoryItems(&history.items),
&history.items,
/*active_turn*/ None,
)
.await?;
)?;
}
Ok(thread)
}
@@ -4761,8 +4756,33 @@ impl CodexMessageProcessor {
let thread = match thread_history {
InitialHistory::Resumed(resumed) => {
let fallback_provider = config_snapshot.model_provider_id.as_str();
if let Some(mut stored_thread) = resume_source_thread {
stored_thread.history = None;
if let Some(stored_thread) = resume_source_thread {
let stored_thread =
if let Some(rollout_path) = stored_thread.rollout_path.clone() {
self.thread_store
.read_thread_by_rollout_path(StoreReadThreadByRolloutPathParams {
rollout_path,
include_archived: true,
include_history: false,
})
.await
.unwrap_or(StoredThread {
history: None,
..stored_thread
})
} else {
self.thread_store
.read_thread(StoreReadThreadParams {
thread_id: stored_thread.thread_id,
include_archived: true,
include_history: false,
})
.await
.unwrap_or(StoredThread {
history: None,
..stored_thread
})
};
Ok(thread_from_stored_thread(
stored_thread,
fallback_provider,
@@ -4809,12 +4829,11 @@ impl CodexMessageProcessor {
thread.path = Some(rollout_path.to_path_buf());
if include_turns {
let history_items = thread_history.get_rollout_items();
populate_thread_turns(
populate_thread_turns_from_history(
&mut thread,
ThreadTurnSource::HistoryItems(&history_items),
&history_items,
/*active_turn*/ None,
)
.await?;
)?;
}
self.attach_thread_name(thread_id, &mut thread).await;
Ok(thread)
@@ -4991,12 +5010,11 @@ impl CodexMessageProcessor {
thread.preview = preview_from_rollout_items(&history_items);
thread.forked_from_id = Some(source_thread_id.to_string());
if include_turns {
populate_thread_turns(
populate_thread_turns_from_history(
&mut thread,
ThreadTurnSource::HistoryItems(&history_items),
&history_items,
/*active_turn*/ None,
)
.await
.map_err(internal_error)?;
}
thread
@@ -5150,7 +5168,7 @@ impl CodexMessageProcessor {
sort_key: StoreThreadSortKey,
sort_direction: SortDirection,
filters: ThreadListFilters,
) -> Result<(Vec<ConversationSummary>, Option<String>), JSONRPCErrorError> {
) -> Result<(Vec<StoredThread>, Option<String>), JSONRPCErrorError> {
let ThreadListFilters {
model_providers,
source_kinds,
@@ -5175,7 +5193,6 @@ impl CodexMessageProcessor {
}
None => Some(vec![self.config.model_provider_id.clone()]),
};
let fallback_provider = self.config.model_provider_id.clone();
let (allowed_sources_vec, source_kind_filter) = compute_source_filters(source_kinds);
let allowed_sources = allowed_sources_vec.as_slice();
let store_sort_direction = match sort_direction {
@@ -5204,20 +5221,21 @@ impl CodexMessageProcessor {
let mut filtered = Vec::with_capacity(page.items.len());
for it in page.items {
let Some(summary) = summary_from_stored_thread(it, fallback_provider.as_str())
else {
continue;
};
let source = with_thread_spawn_agent_metadata(
it.source.clone(),
it.agent_nickname.clone(),
it.agent_role.clone(),
);
if source_kind_filter
.as_ref()
.is_none_or(|filter| source_kind_matches(&summary.source, filter))
.is_none_or(|filter| source_kind_matches(&source, filter))
&& cwd_filters.as_ref().is_none_or(|expected_cwds| {
expected_cwds.iter().any(|expected_cwd| {
path_utils::paths_match_after_normalization(&summary.cwd, expected_cwd)
path_utils::paths_match_after_normalization(&it.cwd, expected_cwd)
})
})
{
filtered.push(summary);
filtered.push(it);
if filtered.len() >= remaining {
break;
}
@@ -8317,12 +8335,11 @@ async fn handle_pending_thread_resume_request(
let connection_id = request_id.connection_id;
let mut thread = pending.thread_summary;
if pending.include_turns
&& let Err(message) = populate_thread_turns(
&& let Err(message) = populate_thread_turns_from_history(
&mut thread,
ThreadTurnSource::HistoryItems(&pending.history_items),
&pending.history_items,
active_turn.as_ref(),
)
.await
{
outgoing
.send_error(request_id, internal_error(message))
@@ -8482,18 +8499,12 @@ async fn send_thread_goal_snapshot_notification(
}
}
enum ThreadTurnSource<'a> {
HistoryItems(&'a [RolloutItem]),
}
async fn populate_thread_turns(
fn populate_thread_turns_from_history(
thread: &mut Thread,
turn_source: ThreadTurnSource<'_>,
items: &[RolloutItem],
active_turn: Option<&Turn>,
) -> std::result::Result<(), String> {
let mut turns = match turn_source {
ThreadTurnSource::HistoryItems(items) => build_turns_from_rollout_items(items),
};
let mut turns = build_turns_from_rollout_items(items);
if let Some(active_turn) = active_turn {
merge_turn_history_with_active_turn(&mut turns, active_turn.clone());
}
@@ -8959,31 +8970,6 @@ async fn title_from_state_db(config: &Config, thread_id: ThreadId) -> Option<Str
.flatten()
}
async fn thread_titles_by_ids(
config: &Config,
thread_ids: &HashSet<ThreadId>,
) -> HashMap<ThreadId, String> {
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<StateDbHandle> {
StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone())
.await
@@ -9067,6 +9053,27 @@ fn thread_store_resume_read_error(err: ThreadStoreError) -> JSONRPCErrorError {
}
}
fn thread_turns_list_history_load_error(
thread_id: ThreadId,
err: ThreadStoreError,
) -> ThreadReadViewError {
match err {
ThreadStoreError::InvalidRequest { message }
if message.starts_with("failed to resolve rollout path `") =>
{
ThreadReadViewError::InvalidRequest(format!(
"thread {thread_id} is not materialized yet; thread/turns/list is unavailable before first user message"
))
}
ThreadStoreError::InvalidRequest { message } => {
ThreadReadViewError::InvalidRequest(message)
}
err => ThreadReadViewError::Internal(format!(
"failed to load thread history for thread {thread_id}: {err}"
)),
}
}
fn conversation_summary_thread_id_read_error(
conversation_id: ThreadId,
err: ThreadStoreError,
@@ -9726,18 +9733,14 @@ pub(crate) fn summary_to_thread(
}
fn thread_backwards_cursor_for_sort_key(
summary: &ConversationSummary,
thread: &StoredThread,
sort_key: StoreThreadSortKey,
sort_direction: SortDirection,
) -> Option<String> {
let timestamp = match sort_key {
StoreThreadSortKey::CreatedAt => summary.timestamp.as_deref(),
StoreThreadSortKey::UpdatedAt => summary
.updated_at
.as_deref()
.or(summary.timestamp.as_deref()),
StoreThreadSortKey::CreatedAt => thread.created_at,
StoreThreadSortKey::UpdatedAt => thread.updated_at,
};
let timestamp = parse_datetime(timestamp)?;
// The state DB stores unique millisecond timestamps. Offset the reverse cursor by one
// millisecond so the opposite-direction query includes the page anchor.
let timestamp = match sort_direction {
@@ -9874,6 +9877,27 @@ fn reconstruct_thread_turns_from_rollout_items(
turns
}
fn reconstruct_thread_turns_for_turns_list(
items: &[RolloutItem],
loaded_status: ThreadStatus,
has_live_running_thread: bool,
active_turn: Option<Turn>,
) -> Vec<Turn> {
let has_live_in_progress_turn = has_live_running_thread
|| active_turn
.as_ref()
.is_some_and(|turn| matches!(turn.status, TurnStatus::InProgress));
let mut turns = reconstruct_thread_turns_from_rollout_items(
items,
loaded_status,
has_live_in_progress_turn,
);
if let Some(active_turn) = active_turn {
merge_turn_history_with_active_turn(&mut turns, active_turn);
}
turns
}
fn normalize_thread_turns_status(
turns: &mut [Turn],
loaded_status: ThreadStatus,
@@ -10032,6 +10056,42 @@ mod tests {
assert!(err.contains("my_tool"), "unexpected error: {err}");
}
#[test]
fn thread_turns_list_merges_in_progress_active_turn_before_agent_status_running() {
let persisted_items = vec![RolloutItem::EventMsg(EventMsg::UserMessage(
codex_protocol::protocol::UserMessageEvent {
message: "persisted".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
},
))];
let active_turn = Turn {
id: "live-turn".to_string(),
items: vec![ThreadItem::UserMessage {
id: "live-user-message".to_string(),
content: vec![V2UserInput::Text {
text: "live".to_string(),
text_elements: Vec::new(),
}],
}],
error: None,
status: TurnStatus::InProgress,
started_at: None,
completed_at: None,
duration_ms: None,
};
let turns = reconstruct_thread_turns_for_turns_list(
&persisted_items,
ThreadStatus::Idle,
/*has_live_running_thread*/ false,
Some(active_turn.clone()),
);
assert_eq!(turns.last(), Some(&active_turn));
}
#[test]
fn validate_dynamic_tools_rejects_empty_namespace() {
let tools = vec![ApiDynamicToolSpec {