From 9c9def76ece2003a3a9a8aa02342115e0289c9c7 Mon Sep 17 00:00:00 2001 From: Tom Wiltzius Date: Fri, 17 Apr 2026 14:48:57 -0700 Subject: [PATCH] codex: add app-server thread store adapter --- .../app-server/src/codex_message_processor.rs | 207 ++---------- .../thread_store_adapter.rs | 308 ++++++++++++++++++ 2 files changed, 328 insertions(+), 187 deletions(-) create mode 100644 codex-rs/app-server/src/codex_message_processor/thread_store_adapter.rs diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index d49df6fcbb..3324ad1ef5 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -327,10 +327,8 @@ use codex_thread_store::ListThreadsParams as StoreListThreadsParams; use codex_thread_store::LocalThreadStore; use codex_thread_store::ReadThreadParams as StoreReadThreadParams; use codex_thread_store::SortDirection as StoreSortDirection; -use codex_thread_store::StoredThread; use codex_thread_store::ThreadMetadataPatch as StoreThreadMetadataPatch; use codex_thread_store::ThreadSortKey as StoreThreadSortKey; -use codex_thread_store::ThreadStore; use codex_thread_store::ThreadStoreError; use codex_thread_store::UpdateThreadMetadataParams as StoreUpdateThreadMetadataParams; use codex_utils_absolute_path::AbsolutePathBuf; @@ -367,6 +365,7 @@ use codex_app_server_protocol::ServerRequest; mod apps_list_helpers; mod plugin_app_helpers; mod plugin_mcp_oauth; +mod thread_store_adapter; mod token_usage_replay; use crate::filters::compute_source_filters; @@ -374,6 +373,7 @@ use crate::filters::source_kind_matches; use crate::thread_state::ThreadListenerCommand; use crate::thread_state::ThreadState; use crate::thread_state::ThreadStateManager; +use thread_store_adapter::ThreadStoreAdapter; use token_usage_replay::latest_token_usage_turn_id_for_thread_path; use token_usage_replay::latest_token_usage_turn_id_from_rollout_items; use token_usage_replay::latest_token_usage_turn_id_from_rollout_path; @@ -463,7 +463,7 @@ pub(crate) struct CodexMessageProcessor { analytics_events_client: AnalyticsEventsClient, arg0_paths: Arg0DispatchPaths, config: Arc, - thread_store: LocalThreadStore, + thread_store_adapter: ThreadStoreAdapter, cli_overrides: Arc>>, runtime_feature_enablement: Arc>>, cloud_requirements: Arc>, @@ -714,7 +714,11 @@ impl CodexMessageProcessor { outgoing: outgoing.clone(), analytics_events_client, arg0_paths, - thread_store: LocalThreadStore::new(codex_rollout::RolloutConfig::from_view(&config)), + thread_store_adapter: ThreadStoreAdapter::new( + LocalThreadStore::new(codex_rollout::RolloutConfig::from_view(&config)), + config.model_provider_id.clone(), + config.cwd.clone(), + ), config, cli_overrides, runtime_feature_enablement, @@ -2750,7 +2754,7 @@ impl CodexMessageProcessor { let thread_id_str = thread_id.to_string(); if let Err(err) = self - .thread_store + .thread_store_adapter .read_thread(StoreReadThreadParams { thread_id, include_archived: false, @@ -2766,7 +2770,7 @@ impl CodexMessageProcessor { self.prepare_thread_for_archive(thread_id).await; match self - .thread_store + .thread_store_adapter .archive_thread(StoreArchiveThreadParams { thread_id }) .await { @@ -2897,7 +2901,7 @@ impl CodexMessageProcessor { } if let Err(err) = self - .thread_store + .thread_store_adapter .update_thread_metadata(StoreUpdateThreadMetadataParams { thread_id, patch: StoreThreadMetadataPatch { @@ -2967,7 +2971,7 @@ impl CodexMessageProcessor { } if let Err(err) = self - .thread_store + .thread_store_adapter .update_thread_metadata(StoreUpdateThreadMetadataParams { thread_id, patch: StoreThreadMetadataPatch { @@ -3352,21 +3356,11 @@ impl CodexMessageProcessor { } }; - let fallback_provider = self.config.model_provider_id.clone(); let result = self - .thread_store + .thread_store_adapter .unarchive_thread(StoreArchiveThreadParams { thread_id }) .await - .map_err(|err| thread_store_archive_error("unarchive", err)) - .and_then(|stored_thread| { - summary_from_stored_thread(stored_thread, fallback_provider.as_str()) - .map(|summary| summary_to_thread(summary, &self.config.cwd)) - .ok_or_else(|| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to read unarchived thread {thread_id}"), - data: None, - }) - }); + .map_err(|err| thread_store_archive_error("unarchive", err)); match result { Ok(mut thread) => { @@ -3803,9 +3797,8 @@ impl CodexMessageProcessor { thread_id: ThreadId, include_turns: bool, ) -> Result, ThreadReadViewError> { - let fallback_provider = self.config.model_provider_id.as_str(); match self - .thread_store + .thread_store_adapter .read_thread(StoreReadThreadParams { thread_id, include_archived: true, @@ -3813,14 +3806,7 @@ impl CodexMessageProcessor { }) .await { - Ok(stored_thread) => { - let (mut thread, history) = - thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); - if include_turns && let Some(history) = history { - thread.turns = build_turns_from_rollout_items(&history.items); - } - Ok(Some(thread)) - } + Ok(thread) => Ok(Some(thread)), Err(ThreadStoreError::InvalidRequest { message }) if message == format!("no rollout found for thread id {thread_id}") => { @@ -5100,7 +5086,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 { @@ -5111,8 +5096,8 @@ impl CodexMessageProcessor { while remaining > 0 { let page_size = remaining.min(THREAD_LIST_MAX_LIMIT); let page = self - .thread_store - .list_threads(StoreListThreadsParams { + .thread_store_adapter + .list_thread_summaries(StoreListThreadsParams { page_size, cursor: cursor_obj.clone(), sort_key, @@ -5125,17 +5110,8 @@ impl CodexMessageProcessor { .await .map_err(thread_store_list_error)?; - let mut candidate_summaries = 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; - }; - candidate_summaries.push(summary); - } - - let mut filtered = Vec::with_capacity(candidate_summaries.len()); - for summary in candidate_summaries { + let mut filtered = Vec::with_capacity(page.items.len()); + for summary in page.items { if source_kind_filter .as_ref() .is_none_or(|filter| source_kind_matches(&summary.source, filter)) @@ -9380,56 +9356,6 @@ fn thread_store_write_error(operation: &str, err: ThreadStoreError) -> JSONRPCEr } } -fn thread_from_stored_thread( - thread: StoredThread, - fallback_provider: &str, - fallback_cwd: &AbsolutePathBuf, -) -> (Thread, Option) { - let path = thread.rollout_path; - let git_info = thread.git_info.map(|info| ApiGitInfo { - sha: info.commit_hash.map(|sha| sha.0), - branch: info.branch, - origin_url: info.repository_url, - }); - let cwd = AbsolutePathBuf::relative_to_current_dir(path_utils::normalize_for_native_workdir( - thread.cwd, - )) - .unwrap_or_else(|err| { - warn!("failed to normalize thread cwd while reading stored thread: {err}"); - fallback_cwd.clone() - }); - let source = with_thread_spawn_agent_metadata( - thread.source, - thread.agent_nickname.clone(), - thread.agent_role.clone(), - ); - let history = thread.history; - let thread = Thread { - id: thread.thread_id.to_string(), - forked_from_id: thread.forked_from_id.map(|id| id.to_string()), - preview: thread.first_user_message.unwrap_or(thread.preview), - ephemeral: false, - model_provider: if thread.model_provider.is_empty() { - fallback_provider.to_string() - } else { - thread.model_provider - }, - created_at: thread.created_at.timestamp(), - updated_at: thread.updated_at.timestamp(), - status: ThreadStatus::NotLoaded, - path, - cwd, - cli_version: thread.cli_version, - agent_nickname: source.get_nickname(), - agent_role: source.get_agent_role(), - source: source.into(), - git_info, - name: thread.name, - turns: Vec::new(), - }; - (thread, history) -} - fn thread_store_archive_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError { match err { ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { @@ -9445,49 +9371,6 @@ fn thread_store_archive_error(operation: &str, err: ThreadStoreError) -> JSONRPC } } -fn summary_from_stored_thread( - thread: StoredThread, - fallback_provider: &str, -) -> Option { - let path = thread.rollout_path?; - let source = with_thread_spawn_agent_metadata( - thread.source, - thread.agent_nickname.clone(), - thread.agent_role.clone(), - ); - let git_info = thread.git_info.map(|git| ConversationGitInfo { - sha: git.commit_hash.map(|sha| sha.0), - branch: git.branch, - origin_url: git.repository_url, - }); - Some(ConversationSummary { - conversation_id: thread.thread_id, - path, - preview: thread.first_user_message.unwrap_or(thread.preview), - // Preserve millisecond precision from the thread store so thread/list cursors - // round-trip the same ordering key used by pagination queries. - timestamp: Some( - thread - .created_at - .to_rfc3339_opts(SecondsFormat::Millis, true), - ), - updated_at: Some( - thread - .updated_at - .to_rfc3339_opts(SecondsFormat::Millis, true), - ), - model_provider: if thread.model_provider.is_empty() { - fallback_provider.to_string() - } else { - thread.model_provider - }, - cwd: thread.cwd, - cli_version: thread.cli_version, - source, - git_info, - }) -} - #[allow(clippy::too_many_arguments)] fn summary_from_state_db_metadata( conversation_id: ThreadId, @@ -10076,11 +9959,8 @@ mod tests { use codex_app_server_protocol::ToolRequestUserInputParams; use codex_protocol::ThreadId; use codex_protocol::openai_models::ReasoningEffort; - use codex_protocol::protocol::AskForApproval; - use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; - use codex_thread_store::StoredThread; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use pretty_assertions::assert_eq; @@ -10130,53 +10010,6 @@ mod tests { validate_dynamic_tools(&tools).expect("valid schema"); } - #[test] - fn summary_from_stored_thread_preserves_millisecond_precision() { - let created_at = - DateTime::parse_from_rfc3339("2025-01-02T03:04:05.678Z").expect("valid timestamp"); - let updated_at = - DateTime::parse_from_rfc3339("2025-01-02T03:04:06.789Z").expect("valid timestamp"); - let thread_id = - ThreadId::from_string("00000000-0000-0000-0000-000000000123").expect("valid thread"); - let stored_thread = StoredThread { - thread_id, - rollout_path: Some(PathBuf::from("/tmp/thread.jsonl")), - forked_from_id: None, - preview: "preview".to_string(), - name: None, - model_provider: "openai".to_string(), - model: None, - reasoning_effort: None, - created_at: created_at.with_timezone(&Utc), - updated_at: updated_at.with_timezone(&Utc), - archived_at: None, - cwd: PathBuf::from("/tmp"), - cli_version: "0.0.0".to_string(), - source: SessionSource::Cli, - agent_nickname: None, - agent_role: None, - agent_path: None, - git_info: None, - approval_mode: AskForApproval::OnRequest, - sandbox_policy: SandboxPolicy::new_read_only_policy(), - token_usage: None, - first_user_message: Some("first user message".to_string()), - history: None, - }; - - let summary = - summary_from_stored_thread(stored_thread, "fallback").expect("summary should exist"); - - assert_eq!( - summary.timestamp.as_deref(), - Some("2025-01-02T03:04:05.678Z") - ); - assert_eq!( - summary.updated_at.as_deref(), - Some("2025-01-02T03:04:06.789Z") - ); - } - #[test] fn config_load_error_marks_cloud_requirements_failures_for_relogin() { let err = std::io::Error::other(CloudRequirementsLoadError::new( diff --git a/codex-rs/app-server/src/codex_message_processor/thread_store_adapter.rs b/codex-rs/app-server/src/codex_message_processor/thread_store_adapter.rs new file mode 100644 index 0000000000..f2786226a0 --- /dev/null +++ b/codex-rs/app-server/src/codex_message_processor/thread_store_adapter.rs @@ -0,0 +1,308 @@ +use super::with_thread_spawn_agent_metadata; +use chrono::SecondsFormat; +use codex_app_server_protocol::ConversationGitInfo; +use codex_app_server_protocol::ConversationSummary; +use codex_app_server_protocol::GitInfo as ApiGitInfo; +use codex_app_server_protocol::Thread; +use codex_app_server_protocol::ThreadStatus; +use codex_app_server_protocol::build_turns_from_rollout_items; +use codex_core::path_utils; +use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams; +use codex_thread_store::ListThreadsParams as StoreListThreadsParams; +use codex_thread_store::LocalThreadStore; +use codex_thread_store::ReadThreadParams as StoreReadThreadParams; +use codex_thread_store::StoredThread; +use codex_thread_store::ThreadPage; +use codex_thread_store::ThreadStore; +use codex_thread_store::ThreadStoreResult; +use codex_thread_store::UpdateThreadMetadataParams as StoreUpdateThreadMetadataParams; +use codex_utils_absolute_path::AbsolutePathBuf; +use tracing::warn; + +/// App-server boundary around `ThreadStore`. Adapts `StoredThread` to `Thread`. +pub(crate) struct ThreadStoreAdapter { + store: S, + fallback_provider: String, + fallback_cwd: AbsolutePathBuf, +} + +pub(crate) struct ThreadSummaryPage { + pub(crate) items: Vec, + pub(crate) next_cursor: Option, +} + +impl ThreadStoreAdapter { + pub(crate) fn new(store: S, fallback_provider: String, fallback_cwd: AbsolutePathBuf) -> Self { + Self { + store, + fallback_provider, + fallback_cwd, + } + } +} + +impl ThreadStoreAdapter +where + S: ThreadStore, +{ + pub(crate) async fn read_thread( + &self, + params: StoreReadThreadParams, + ) -> ThreadStoreResult { + let include_turns = params.include_history; + let stored_thread = self.store.read_thread(params).await?; + Ok(self.thread_from_stored_thread(stored_thread, include_turns)) + } + + pub(crate) async fn list_thread_summaries( + &self, + params: StoreListThreadsParams, + ) -> ThreadStoreResult { + let page = self.store.list_threads(params).await?; + Ok(self.thread_summary_page_from_store(page)) + } + + pub(crate) async fn update_thread_metadata( + &self, + params: StoreUpdateThreadMetadataParams, + ) -> ThreadStoreResult { + let stored_thread = self.store.update_thread_metadata(params).await?; + Ok(self.thread_from_stored_thread(stored_thread, /*include_turns*/ false)) + } + + pub(crate) async fn archive_thread( + &self, + params: StoreArchiveThreadParams, + ) -> ThreadStoreResult<()> { + self.store.archive_thread(params).await + } + + pub(crate) async fn unarchive_thread( + &self, + params: StoreArchiveThreadParams, + ) -> ThreadStoreResult { + let stored_thread = self.store.unarchive_thread(params).await?; + Ok(self.thread_from_stored_thread(stored_thread, /*include_turns*/ false)) + } +} + +impl ThreadStoreAdapter { + fn thread_from_stored_thread( + &self, + stored_thread: StoredThread, + include_turns: bool, + ) -> Thread { + let path = stored_thread.rollout_path; + let git_info = stored_thread.git_info.map(|info| ApiGitInfo { + sha: info.commit_hash.map(|sha| sha.0), + branch: info.branch, + origin_url: info.repository_url, + }); + let cwd = AbsolutePathBuf::relative_to_current_dir( + path_utils::normalize_for_native_workdir(stored_thread.cwd), + ) + .unwrap_or_else(|err| { + warn!("failed to normalize thread cwd while reading stored thread: {err}"); + self.fallback_cwd.clone() + }); + let source = with_thread_spawn_agent_metadata( + stored_thread.source, + stored_thread.agent_nickname.clone(), + stored_thread.agent_role.clone(), + ); + let mut thread = Thread { + id: stored_thread.thread_id.to_string(), + forked_from_id: stored_thread.forked_from_id.map(|id| id.to_string()), + preview: stored_thread + .first_user_message + .unwrap_or(stored_thread.preview), + ephemeral: false, + model_provider: if stored_thread.model_provider.is_empty() { + self.fallback_provider.clone() + } else { + stored_thread.model_provider + }, + created_at: stored_thread.created_at.timestamp(), + updated_at: stored_thread.updated_at.timestamp(), + status: ThreadStatus::NotLoaded, + path, + cwd, + cli_version: stored_thread.cli_version, + agent_nickname: source.get_nickname(), + agent_role: source.get_agent_role(), + source: source.into(), + git_info, + name: stored_thread.name, + turns: Vec::new(), + }; + if include_turns && let Some(history) = stored_thread.history { + thread.turns = build_turns_from_rollout_items(&history.items); + } + thread + } + + fn thread_summary_page_from_store(&self, page: ThreadPage) -> ThreadSummaryPage { + let items = page + .items + .into_iter() + .filter_map(|stored_thread| self.summary_from_stored_thread(stored_thread)) + .collect(); + ThreadSummaryPage { + items, + next_cursor: page.next_cursor, + } + } + + fn summary_from_stored_thread( + &self, + stored_thread: StoredThread, + ) -> Option { + let path = stored_thread.rollout_path?; + let source = with_thread_spawn_agent_metadata( + stored_thread.source, + stored_thread.agent_nickname.clone(), + stored_thread.agent_role.clone(), + ); + let git_info = stored_thread.git_info.map(|git| ConversationGitInfo { + sha: git.commit_hash.map(|sha| sha.0), + branch: git.branch, + origin_url: git.repository_url, + }); + Some(ConversationSummary { + conversation_id: stored_thread.thread_id, + path, + preview: stored_thread + .first_user_message + .unwrap_or(stored_thread.preview), + // Preserve millisecond precision from the thread store so thread/list cursors + // round-trip the same ordering key used by pagination queries. + timestamp: Some( + stored_thread + .created_at + .to_rfc3339_opts(SecondsFormat::Millis, true), + ), + updated_at: Some( + stored_thread + .updated_at + .to_rfc3339_opts(SecondsFormat::Millis, true), + ), + model_provider: if stored_thread.model_provider.is_empty() { + self.fallback_provider.clone() + } else { + stored_thread.model_provider + }, + cwd: stored_thread.cwd, + cli_version: stored_thread.cli_version, + source, + git_info, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::DateTime; + use chrono::Utc; + use codex_protocol::ThreadId; + use codex_protocol::protocol::AskForApproval; + use codex_protocol::protocol::EventMsg; + use codex_protocol::protocol::RolloutItem; + use codex_protocol::protocol::SandboxPolicy; + use codex_protocol::protocol::SessionSource; + use codex_protocol::protocol::UserMessageEvent; + use codex_thread_store::StoredThreadHistory; + use pretty_assertions::assert_eq; + use std::path::PathBuf; + + #[test] + fn summary_preserves_millisecond_precision() { + let created_at = + DateTime::parse_from_rfc3339("2025-01-02T03:04:05.678Z").expect("valid timestamp"); + let updated_at = + DateTime::parse_from_rfc3339("2025-01-02T03:04:06.789Z").expect("valid timestamp"); + let stored_thread = stored_thread( + created_at.with_timezone(&Utc), + updated_at.with_timezone(&Utc), + ); + + let adapter = ThreadStoreAdapter::new( + (), + "fallback".to_string(), + AbsolutePathBuf::from_absolute_path_checked("/fallback").expect("absolute path"), + ); + let summary = adapter + .summary_from_stored_thread(stored_thread) + .expect("summary should exist"); + + assert_eq!( + summary.timestamp.as_deref(), + Some("2025-01-02T03:04:05.678Z") + ); + assert_eq!( + summary.updated_at.as_deref(), + Some("2025-01-02T03:04:06.789Z") + ); + } + + #[test] + fn thread_conversion_includes_history_when_requested() { + let created_at = + DateTime::parse_from_rfc3339("2025-01-02T03:04:05.678Z").expect("valid timestamp"); + let updated_at = + DateTime::parse_from_rfc3339("2025-01-02T03:04:06.789Z").expect("valid timestamp"); + let mut stored_thread = stored_thread( + created_at.with_timezone(&Utc), + updated_at.with_timezone(&Utc), + ); + stored_thread.history = Some(StoredThreadHistory { + thread_id: stored_thread.thread_id, + items: vec![RolloutItem::EventMsg(EventMsg::UserMessage( + UserMessageEvent { + message: "hello".to_string(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + }, + ))], + }); + + let adapter = ThreadStoreAdapter::new( + (), + "fallback".to_string(), + AbsolutePathBuf::from_absolute_path_checked("/fallback").expect("absolute path"), + ); + let thread = adapter.thread_from_stored_thread(stored_thread, /*include_turns*/ true); + + assert_eq!(thread.turns.len(), 1); + } + + fn stored_thread(created_at: DateTime, updated_at: DateTime) -> StoredThread { + StoredThread { + thread_id: ThreadId::from_string("00000000-0000-0000-0000-000000000123") + .expect("valid thread"), + rollout_path: Some(PathBuf::from("/tmp/thread.jsonl")), + forked_from_id: None, + preview: "preview".to_string(), + name: None, + model_provider: "openai".to_string(), + model: None, + reasoning_effort: None, + created_at, + updated_at, + archived_at: None, + cwd: PathBuf::from("/tmp"), + cli_version: "0.0.0".to_string(), + source: SessionSource::Cli, + agent_nickname: None, + agent_role: None, + agent_path: None, + git_info: None, + approval_mode: AskForApproval::OnRequest, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + token_usage: None, + first_user_message: Some("first user message".to_string()), + history: None, + } + } +}