diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 05ce2d797..42cb8b2d1 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -244,7 +244,6 @@ use codex_core::plugins::load_plugin_apps; use codex_core::plugins::load_plugin_mcp_servers; use codex_core::read_head_for_summary; use codex_core::read_session_meta_line; -use codex_core::rollout_date_parts; use codex_core::sandboxing::SandboxPermissions; use codex_core::windows_sandbox::WindowsSandboxLevelExt; use codex_core::windows_sandbox::WindowsSandboxSetupMode as CoreWindowsSandboxSetupMode; @@ -319,8 +318,10 @@ use codex_state::StateRuntime; use codex_state::ThreadMetadata; use codex_state::ThreadMetadataBuilder; use codex_state::log_db::LogDbLayer; +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::ThreadSortKey as StoreThreadSortKey; use codex_thread_store::ThreadStore; @@ -331,9 +332,6 @@ use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP; use std::collections::BTreeMap; use std::collections::HashMap; use std::collections::HashSet; -use std::ffi::OsStr; -use std::fs::FileTimes; -use std::fs::OpenOptions; use std::io::Error as IoError; use std::path::Path; use std::path::PathBuf; @@ -343,7 +341,6 @@ use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::time::Duration; use std::time::Instant; -use std::time::SystemTime; use tokio::sync::Mutex; use tokio::sync::broadcast; use tokio::sync::oneshot; @@ -447,6 +444,7 @@ pub(crate) struct CodexMessageProcessor { analytics_events_client: AnalyticsEventsClient, arg0_paths: Arg0DispatchPaths, config: Arc, + thread_store: LocalThreadStore, cli_overrides: Arc>>, runtime_feature_enablement: Arc>>, cloud_requirements: Arc>, @@ -705,6 +703,7 @@ impl CodexMessageProcessor { outgoing: outgoing.clone(), analytics_events_client, arg0_paths, + thread_store: LocalThreadStore::new(codex_rollout::RolloutConfig::from_view(&config)), config, cli_overrides, runtime_feature_enablement, @@ -2717,7 +2716,6 @@ impl CodexMessageProcessor { } async fn thread_archive(&self, request_id: ConnectionRequestId, params: ThreadArchiveParams) { - // TODO(jif) mostly rewrite this using sqlite after phase 1 let thread_id = match ThreadId::from_string(¶ms.thread_id) { Ok(id) => id, Err(err) => { @@ -2731,32 +2729,28 @@ impl CodexMessageProcessor { } }; - let rollout_path = - match find_thread_path_by_id_str(&self.config.codex_home, &thread_id.to_string()).await - { - Ok(Some(p)) => p, - Ok(None) => { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("no rollout found for thread id {thread_id}"), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; - } - Err(err) => { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("failed to locate thread id {thread_id}: {err}"), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; - } - }; - let thread_id_str = thread_id.to_string(); - match self.archive_thread_common(thread_id, &rollout_path).await { + if let Err(err) = self + .thread_store + .read_thread(StoreReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + { + self.outgoing + .send_error(request_id, thread_store_archive_error("archive", err)) + .await; + return; + } + self.prepare_thread_for_archive(thread_id).await; + + match self + .thread_store + .archive_thread(StoreArchiveThreadParams { thread_id }) + .await + { Ok(()) => { let response = ThreadArchiveResponse {}; self.outgoing.send_response(request_id, response).await; @@ -2768,7 +2762,9 @@ impl CodexMessageProcessor { .await; } Err(err) => { - self.outgoing.send_error(request_id, err).await; + self.outgoing + .send_error(request_id, thread_store_archive_error("archive", err)) + .await; } } } @@ -3437,7 +3433,6 @@ impl CodexMessageProcessor { request_id: ConnectionRequestId, params: ThreadUnarchiveParams, ) { - // TODO(jif) mostly rewrite this using sqlite after phase 1 let thread_id = match ThreadId::from_string(¶ms.thread_id) { Ok(id) => id, Err(err) => { @@ -3451,152 +3446,21 @@ impl CodexMessageProcessor { } }; - let archived_path = match find_archived_thread_path_by_id_str( - &self.config.codex_home, - &thread_id.to_string(), - ) - .await - { - Ok(Some(path)) => path, - Ok(None) => { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("no archived rollout found for thread id {thread_id}"), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; - } - Err(err) => { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("failed to locate archived thread id {thread_id}: {err}"), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; - } - }; - - let rollout_path_display = archived_path.display().to_string(); let fallback_provider = self.config.model_provider_id.clone(); - let state_db_ctx = get_state_db(&self.config).await; - let archived_folder = self - .config - .codex_home - .join(codex_core::ARCHIVED_SESSIONS_SUBDIR); - - let result: Result = async { - let canonical_archived_dir = tokio::fs::canonicalize(&archived_folder).await.map_err( - |err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!( - "failed to unarchive thread: unable to resolve archived directory: {err}" - ), - data: None, - }, - )?; - let canonical_rollout_path = tokio::fs::canonicalize(&archived_path).await; - let canonical_rollout_path = if let Ok(path) = canonical_rollout_path - && path.starts_with(&canonical_archived_dir) - { - path - } else { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "rollout path `{rollout_path_display}` must be in archived directory" - ), - data: None, - }); - }; - - let required_suffix = format!("{thread_id}.jsonl"); - let Some(file_name) = canonical_rollout_path.file_name().map(OsStr::to_owned) else { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("rollout path `{rollout_path_display}` missing file name"), - data: None, - }); - }; - if !file_name - .to_string_lossy() - .ends_with(required_suffix.as_str()) - { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "rollout path `{rollout_path_display}` does not match thread id {thread_id}" - ), - data: None, - }); - } - - let Some((year, month, day)) = rollout_date_parts(&file_name) else { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "rollout path `{rollout_path_display}` missing filename timestamp" - ), - data: None, - }); - }; - - let sessions_folder = self.config.codex_home.join(codex_core::SESSIONS_SUBDIR); - let dest_dir = sessions_folder.join(year).join(month).join(day); - let restored_path = dest_dir.join(&file_name); - tokio::fs::create_dir_all(&dest_dir) - .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to unarchive thread: {err}"), - data: None, - })?; - tokio::fs::rename(&canonical_rollout_path, &restored_path) - .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to unarchive thread: {err}"), - data: None, - })?; - tokio::task::spawn_blocking({ - let restored_path = restored_path.clone(); - move || -> std::io::Result<()> { - let times = FileTimes::new().set_modified(SystemTime::now()); - OpenOptions::new() - .append(true) - .open(&restored_path)? - .set_times(times)?; - Ok(()) - } - }) + let result = self + .thread_store + .unarchive_thread(StoreArchiveThreadParams { thread_id }) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to update unarchived thread timestamp: {err}"), - data: None, - })? - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to update unarchived thread timestamp: {err}"), - data: None, - })?; - if let Some(ctx) = state_db_ctx { - let _ = ctx - .mark_unarchived(thread_id, restored_path.as_path()) - .await; - } - let summary = - read_summary_from_rollout(restored_path.as_path(), fallback_provider.as_str()) - .await - .map_err(|err| JSONRPCErrorError { + .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: {err}"), + message: format!("failed to read unarchived thread {thread_id}"), data: None, - })?; - Ok(summary_to_thread(summary, &self.config.cwd)) - } - .await; + }) + }); match result { Ok(mut thread) => { @@ -5103,11 +4967,11 @@ impl CodexMessageProcessor { 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 = LocalThreadStore::new(codex_rollout::RolloutConfig::from_view(&self.config)); while remaining > 0 { let page_size = remaining.min(THREAD_LIST_MAX_LIMIT); - let page = store + let page = self + .thread_store .list_threads(StoreListThreadsParams { page_size, cursor: cursor_obj.clone(), @@ -5944,76 +5808,10 @@ impl CodexMessageProcessor { .await; } - async fn archive_thread_common( - &self, - thread_id: ThreadId, - rollout_path: &Path, - ) -> Result<(), JSONRPCErrorError> { - // Verify rollout_path is under sessions dir. - let rollout_folder = self.config.codex_home.join(codex_core::SESSIONS_SUBDIR); - - let canonical_sessions_dir = match tokio::fs::canonicalize(&rollout_folder).await { - Ok(path) => path, - Err(err) => { - return Err(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!( - "failed to archive thread: unable to resolve sessions directory: {err}" - ), - data: None, - }); - } - }; - let canonical_rollout_path = tokio::fs::canonicalize(rollout_path).await; - let canonical_rollout_path = if let Ok(path) = canonical_rollout_path - && path.starts_with(&canonical_sessions_dir) - { - path - } else { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "rollout path `{}` must be in sessions directory", - rollout_path.display() - ), - data: None, - }); - }; - - // Verify file name matches thread id. - let required_suffix = format!("{thread_id}.jsonl"); - let Some(file_name) = canonical_rollout_path.file_name().map(OsStr::to_owned) else { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "rollout path `{}` missing file name", - rollout_path.display() - ), - data: None, - }); - }; - if !file_name - .to_string_lossy() - .ends_with(required_suffix.as_str()) - { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "rollout path `{}` does not match thread id {thread_id}", - rollout_path.display() - ), - data: None, - }); - } - - let mut state_db_ctx = None; - + async fn prepare_thread_for_archive(&self, thread_id: ThreadId) { // If the thread is active, request shutdown and wait briefly. let removed_conversation = self.thread_manager.remove_thread(&thread_id).await; if let Some(conversation) = removed_conversation { - if let Some(ctx) = conversation.state_db() { - state_db_ctx = Some(ctx); - } info!("thread {thread_id} was active; shutting down"); match Self::wait_for_thread_shutdown(&conversation).await { ThreadShutdownResult::Complete => {} @@ -6028,34 +5826,6 @@ impl CodexMessageProcessor { } } self.finalize_thread_teardown(thread_id).await; - - if state_db_ctx.is_none() { - state_db_ctx = get_state_db(&self.config).await; - } - - // Move the rollout file to archived. - let result: std::io::Result<()> = async move { - let archive_folder = self - .config - .codex_home - .join(codex_core::ARCHIVED_SESSIONS_SUBDIR); - tokio::fs::create_dir_all(&archive_folder).await?; - let archived_path = archive_folder.join(&file_name); - tokio::fs::rename(&canonical_rollout_path, &archived_path).await?; - if let Some(ctx) = state_db_ctx { - let _ = ctx - .mark_archived(thread_id, archived_path.as_path(), Utc::now()) - .await; - } - Ok(()) - } - .await; - - result.map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to archive thread: {err}"), - data: None, - }) } async fn apps_list(&self, request_id: ConnectionRequestId, params: AppsListParams) { @@ -9398,6 +9168,21 @@ fn thread_store_list_error(err: ThreadStoreError) -> JSONRPCErrorError { } } +fn thread_store_archive_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError { + match err { + ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message, + data: None, + }, + err => JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to {operation} thread: {err}"), + data: None, + }, + } +} + fn summary_from_stored_thread( thread: StoredThread, fallback_provider: &str, diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 6fc74b30e..79a96d11e 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -7,7 +7,6 @@ use crate::config::Config; use crate::config::ConfigBuilder; use crate::contextual_user_message::SUBAGENT_NOTIFICATION_OPEN_TAG; use assert_matches::assert_matches; -use chrono::Utc; use codex_features::Feature; use codex_login::CodexAuth; use codex_protocol::AgentPath; @@ -24,6 +23,9 @@ use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::protocol::TurnStartedEvent; +use codex_thread_store::ArchiveThreadParams; +use codex_thread_store::LocalThreadStore; +use codex_thread_store::ThreadStore; use pretty_assertions::assert_eq; use tempfile::TempDir; use tokio::time::Duration; @@ -1658,38 +1660,18 @@ async fn resume_agent_from_rollout_reads_archived_rollout_path() { .await .expect("child thread should exist"); persist_thread_for_tree_resume(&child_thread, "persist before archiving").await; - let rollout_path = child_thread - .rollout_path() - .expect("thread should have rollout path"); - let state_db = child_thread - .state_db() - .expect("thread should have state db handle"); - let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should succeed"); - - let archived_root = harness - .config - .codex_home - .join(crate::ARCHIVED_SESSIONS_SUBDIR); - tokio::fs::create_dir_all(&archived_root) + let store = LocalThreadStore::new(codex_rollout::RolloutConfig::from_view(&harness.config)); + store + .archive_thread(ArchiveThreadParams { + thread_id: child_thread_id, + }) .await - .expect("archived root should exist"); - let archived_rollout_path = archived_root.join( - rollout_path - .file_name() - .expect("rollout file name should be present"), - ); - tokio::fs::rename(&rollout_path, &archived_rollout_path) - .await - .expect("rollout should move to archived path"); - state_db - .mark_archived(child_thread_id, archived_rollout_path.as_path(), Utc::now()) - .await - .expect("state db archive update should succeed"); + .expect("child thread should archive"); let resumed_thread_id = harness .control diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 31fcf4724..de169db1f 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -48,6 +48,7 @@ pub use list::get_threads_in_root; pub use list::parse_cursor; pub use list::read_head_for_summary; pub use list::read_session_meta_line; +pub use list::read_thread_item_from_rollout; pub use list::rollout_date_parts; pub use metadata::builder_from_items; pub use policy::EventPersistenceMode; diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index afa410330..640f0b027 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -756,6 +756,21 @@ async fn build_thread_item( None } +/// Read a single rollout file into the same summary item shape used by thread listing. +/// +/// This is for callers that already resolved a rollout path and need the same +/// metadata/preview extraction as list operations without scanning the whole +/// sessions tree. +pub async fn read_thread_item_from_rollout(path: PathBuf) -> Option { + build_thread_item( + path, + &[], + /*provider_matcher*/ None, + /*updated_at*/ None, + ) + .await +} + /// Collects immediate subdirectories of `parent`, parses their (string) names with `parse`, /// and returns them sorted descending by the parsed key. async fn collect_dirs_desc(parent: &Path, parse: F) -> io::Result> diff --git a/codex-rs/thread-store/src/local/archive_thread.rs b/codex-rs/thread-store/src/local/archive_thread.rs new file mode 100644 index 000000000..4fcea6dba --- /dev/null +++ b/codex-rs/thread-store/src/local/archive_thread.rs @@ -0,0 +1,170 @@ +use chrono::Utc; +use codex_rollout::find_thread_path_by_id_str; + +use super::LocalThreadStore; +use super::helpers::matching_rollout_file_name; +use super::helpers::scoped_rollout_path; +use crate::ArchiveThreadParams; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +pub(super) async fn archive_thread( + store: &LocalThreadStore, + params: ArchiveThreadParams, +) -> ThreadStoreResult<()> { + let thread_id = params.thread_id; + let rollout_path = + find_thread_path_by_id_str(store.config.codex_home.as_path(), &thread_id.to_string()) + .await + .map_err(|err| ThreadStoreError::InvalidRequest { + message: format!("failed to locate thread id {thread_id}: {err}"), + })? + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("no rollout found for thread id {thread_id}"), + })?; + + let canonical_rollout_path = scoped_rollout_path( + store.config.codex_home.join(codex_rollout::SESSIONS_SUBDIR), + rollout_path.as_path(), + "sessions", + )?; + let file_name = matching_rollout_file_name( + canonical_rollout_path.as_path(), + thread_id, + rollout_path.as_path(), + )?; + + let archive_folder = store + .config + .codex_home + .join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR); + std::fs::create_dir_all(&archive_folder).map_err(|err| ThreadStoreError::Internal { + message: format!("failed to archive thread: {err}"), + })?; + let archived_path = archive_folder.join(&file_name); + std::fs::rename(&canonical_rollout_path, &archived_path).map_err(|err| { + ThreadStoreError::Internal { + message: format!("failed to archive thread: {err}"), + } + })?; + + if let Some(ctx) = codex_rollout::state_db::get_state_db(&store.config).await { + let _ = ctx + .mark_archived(thread_id, archived_path.as_path(), Utc::now()) + .await; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use chrono::Utc; + use codex_protocol::ThreadId; + use codex_protocol::protocol::SessionSource; + use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + use uuid::Uuid; + + use super::*; + use crate::ListThreadsParams; + use crate::ThreadSortKey; + use crate::ThreadStore; + use crate::local::LocalThreadStore; + use crate::local::test_support::test_config; + use crate::local::test_support::write_session_file; + + #[tokio::test] + async fn archive_thread_moves_rollout_to_archived_collection() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let uuid = Uuid::from_u128(201); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let active_path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + + store + .archive_thread(ArchiveThreadParams { thread_id }) + .await + .expect("archive thread"); + + assert!(!active_path.exists()); + let archived_path = home + .path() + .join(ARCHIVED_SESSIONS_SUBDIR) + .join(active_path.file_name().expect("file name")); + assert!(archived_path.exists()); + + let archived = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: true, + search_term: None, + }) + .await + .expect("archived listing"); + assert_eq!(archived.items.len(), 1); + assert_eq!(archived.items[0].thread_id, thread_id); + assert_eq!(archived.items[0].rollout_path, Some(archived_path)); + assert_eq!( + archived.items[0].archived_at, + Some(archived.items[0].updated_at) + ); + } + + #[tokio::test] + async fn archive_thread_updates_sqlite_metadata_when_present() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(202); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let active_path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .expect("backfill should be complete"); + let mut builder = codex_state::ThreadMetadataBuilder::new( + thread_id, + active_path.clone(), + Utc::now(), + SessionSource::Cli, + ); + builder.model_provider = Some(config.model_provider_id.clone()); + builder.cwd = home.path().to_path_buf(); + builder.cli_version = Some("test_version".to_string()); + let metadata = builder.build(config.model_provider_id.as_str()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + store + .archive_thread(ArchiveThreadParams { thread_id }) + .await + .expect("archive thread"); + + let archived_path = home + .path() + .join(ARCHIVED_SESSIONS_SUBDIR) + .join(active_path.file_name().expect("file name")); + let updated = runtime + .get_thread(thread_id) + .await + .expect("state db read should succeed") + .expect("thread metadata should exist"); + assert_eq!(updated.rollout_path, archived_path); + assert!(updated.archived_at.is_some()); + } +} diff --git a/codex-rs/thread-store/src/local/helpers.rs b/codex-rs/thread-store/src/local/helpers.rs new file mode 100644 index 000000000..e28ecc661 --- /dev/null +++ b/codex-rs/thread-store/src/local/helpers.rs @@ -0,0 +1,168 @@ +use std::ffi::OsStr; +use std::fs::FileTimes; +use std::fs::OpenOptions; +use std::path::Path; +use std::path::PathBuf; +use std::time::SystemTime; + +use chrono::DateTime; +use chrono::Utc; +use codex_git_utils::GitSha; +use codex_protocol::ThreadId; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::GitInfo; +use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::SessionSource; +use codex_rollout::ThreadItem; + +use crate::StoredThread; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +pub(super) fn scoped_rollout_path( + root: PathBuf, + rollout_path: &Path, + root_name: &str, +) -> ThreadStoreResult { + let canonical_root = + std::fs::canonicalize(&root).map_err(|err| ThreadStoreError::Internal { + message: format!( + "failed to resolve {root_name} directory `{}`: {err}", + root.display() + ), + })?; + let canonical_rollout_path = + std::fs::canonicalize(rollout_path).map_err(|_| ThreadStoreError::InvalidRequest { + message: format!( + "rollout path `{}` must be in {root_name} directory", + rollout_path.display() + ), + })?; + if canonical_rollout_path.starts_with(&canonical_root) { + Ok(canonical_rollout_path) + } else { + Err(ThreadStoreError::InvalidRequest { + message: format!( + "rollout path `{}` must be in {root_name} directory", + rollout_path.display() + ), + }) + } +} + +pub(super) fn matching_rollout_file_name( + rollout_path: &Path, + thread_id: ThreadId, + display_path: &Path, +) -> ThreadStoreResult { + let Some(file_name) = rollout_path.file_name().map(OsStr::to_owned) else { + return Err(ThreadStoreError::InvalidRequest { + message: format!( + "rollout path `{}` missing file name", + display_path.display() + ), + }); + }; + let required_suffix = format!("{thread_id}.jsonl"); + if file_name + .to_string_lossy() + .ends_with(required_suffix.as_str()) + { + Ok(file_name) + } else { + Err(ThreadStoreError::InvalidRequest { + message: format!( + "rollout path `{}` does not match thread id {thread_id}", + display_path.display() + ), + }) + } +} + +pub(super) fn touch_modified_time(path: &Path) -> std::io::Result<()> { + let times = FileTimes::new().set_modified(SystemTime::now()); + OpenOptions::new().append(true).open(path)?.set_times(times) +} + +pub(super) fn stored_thread_from_rollout_item( + item: ThreadItem, + archived: bool, + default_provider: &str, +) -> Option { + let thread_id = item + .thread_id + .or_else(|| thread_id_from_rollout_path(item.path.as_path()))?; + let created_at = parse_rfc3339(item.created_at.as_deref()).unwrap_or_else(Utc::now); + let updated_at = parse_rfc3339(item.updated_at.as_deref()).unwrap_or(created_at); + let archived_at = archived.then_some(updated_at); + let git_info = git_info_from_parts( + item.git_sha.clone(), + item.git_branch.clone(), + item.git_origin_url.clone(), + ); + let source = item.source.unwrap_or(SessionSource::Unknown); + let preview = item.first_user_message.clone().unwrap_or_default(); + + Some(StoredThread { + thread_id, + rollout_path: Some(item.path), + forked_from_id: None, + preview, + name: None, + model_provider: item + .model_provider + .filter(|provider| !provider.is_empty()) + .unwrap_or_else(|| default_provider.to_string()), + model: None, + reasoning_effort: None, + created_at, + updated_at, + archived_at, + cwd: item.cwd.unwrap_or_default(), + cli_version: item.cli_version.unwrap_or_default(), + source, + agent_nickname: item.agent_nickname, + agent_role: item.agent_role, + agent_path: None, + git_info, + approval_mode: AskForApproval::OnRequest, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + token_usage: None, + first_user_message: item.first_user_message, + history: None, + }) +} + +fn parse_rfc3339(value: Option<&str>) -> Option> { + DateTime::parse_from_rfc3339(value?) + .ok() + .map(|dt| dt.with_timezone(&Utc)) +} + +fn git_info_from_parts( + sha: Option, + branch: Option, + origin_url: Option, +) -> Option { + if sha.is_none() && branch.is_none() && origin_url.is_none() { + return None; + } + Some(GitInfo { + commit_hash: sha.as_deref().map(GitSha::new), + branch, + repository_url: origin_url, + }) +} + +fn thread_id_from_rollout_path(path: &Path) -> Option { + let file_name = path.file_name()?.to_str()?; + let stem = file_name.strip_suffix(".jsonl")?; + if stem.len() < 37 { + return None; + } + let uuid_start = stem.len().saturating_sub(36); + if !stem[..uuid_start].ends_with('-') { + return None; + } + ThreadId::from_string(&stem[uuid_start..]).ok() +} diff --git a/codex-rs/thread-store/src/local/list_threads.rs b/codex-rs/thread-store/src/local/list_threads.rs new file mode 100644 index 000000000..35c6f3de4 --- /dev/null +++ b/codex-rs/thread-store/src/local/list_threads.rs @@ -0,0 +1,319 @@ +use codex_rollout::RolloutConfig; +use codex_rollout::RolloutRecorder; +use codex_rollout::parse_cursor; + +use super::LocalThreadStore; +use super::helpers::stored_thread_from_rollout_item; +use crate::ListThreadsParams; +use crate::ThreadPage; +use crate::ThreadSortKey; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +pub(super) async fn list_threads( + store: &LocalThreadStore, + params: ListThreadsParams, +) -> ThreadStoreResult { + let cursor = params + .cursor + .as_deref() + .map(|cursor| { + parse_cursor(cursor).ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("invalid cursor: {cursor}"), + }) + }) + .transpose()?; + let sort_key = match params.sort_key { + ThreadSortKey::CreatedAt => codex_rollout::ThreadSortKey::CreatedAt, + ThreadSortKey::UpdatedAt => codex_rollout::ThreadSortKey::UpdatedAt, + }; + let page = list_rollout_threads(&store.config, ¶ms, cursor.as_ref(), sort_key).await?; + + let next_cursor = page + .next_cursor + .as_ref() + .and_then(|cursor| serde_json::to_value(cursor).ok()) + .and_then(|value| value.as_str().map(str::to_owned)); + let items = page + .items + .into_iter() + .filter_map(|item| { + stored_thread_from_rollout_item( + item, + params.archived, + store.config.model_provider_id.as_str(), + ) + }) + .collect::>(); + + Ok(ThreadPage { items, next_cursor }) +} + +async fn list_rollout_threads( + config: &RolloutConfig, + params: &ListThreadsParams, + cursor: Option<&codex_rollout::Cursor>, + sort_key: codex_rollout::ThreadSortKey, +) -> ThreadStoreResult { + let page = if params.archived { + RolloutRecorder::list_archived_threads( + config, + params.page_size, + cursor, + sort_key, + params.allowed_sources.as_slice(), + params.model_providers.as_deref(), + config.model_provider_id.as_str(), + params.search_term.as_deref(), + ) + .await + } else { + RolloutRecorder::list_threads( + config, + params.page_size, + cursor, + sort_key, + params.allowed_sources.as_slice(), + params.model_providers.as_deref(), + config.model_provider_id.as_str(), + params.search_term.as_deref(), + ) + .await + }; + page.map_err(|err| ThreadStoreError::Internal { + message: format!("failed to list threads: {err}"), + }) +} + +#[cfg(test)] +mod tests { + use chrono::Utc; + use codex_protocol::ThreadId; + use codex_protocol::protocol::SessionSource; + use pretty_assertions::assert_eq; + use std::fs; + use tempfile::TempDir; + use uuid::Uuid; + + use super::*; + use crate::ThreadStore; + use crate::local::LocalThreadStore; + use crate::local::test_support::test_config; + use crate::local::test_support::write_archived_session_file; + use crate::local::test_support::write_session_file; + use crate::local::test_support::write_session_file_with; + + #[tokio::test] + async fn list_threads_uses_default_provider_when_rollout_omits_provider() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + write_session_file_with( + home.path(), + home.path().join("sessions/2025/01/03"), + "2025-01-03T12-00-00", + Uuid::from_u128(102), + "Hello from user", + /*model_provider*/ None, + ) + .expect("session file"); + + let page = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: false, + search_term: None, + }) + .await + .expect("thread listing"); + + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].model_provider, "test-provider"); + } + + #[tokio::test] + async fn list_threads_preserves_sqlite_title_search_results() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(103); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = home.path().join("rollout-title-search.jsonl"); + fs::write(&rollout_path, "").expect("placeholder rollout file"); + + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .expect("backfill should be complete"); + let created_at = Utc::now(); + let mut builder = codex_state::ThreadMetadataBuilder::new( + thread_id, + rollout_path, + created_at, + SessionSource::Cli, + ); + builder.model_provider = Some(config.model_provider_id.clone()); + builder.cwd = home.path().to_path_buf(); + builder.cli_version = Some("test_version".to_string()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.title = "needle title".to_string(); + metadata.first_user_message = Some("plain preview".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let page = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: false, + search_term: Some("needle".to_string()), + }) + .await + .expect("thread listing"); + + let ids = page + .items + .iter() + .map(|item| item.thread_id) + .collect::>(); + assert_eq!(ids, vec![thread_id]); + assert_eq!( + page.items[0].first_user_message.as_deref(), + Some("plain preview") + ); + } + + #[tokio::test] + async fn list_threads_selects_active_or_archived_collection() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let active_uuid = Uuid::from_u128(105); + let archived_uuid = Uuid::from_u128(106); + write_session_file(home.path(), "2025-01-03T12-00-00", active_uuid) + .expect("active session file"); + write_archived_session_file(home.path(), "2025-01-03T13-00-00", archived_uuid) + .expect("archived session file"); + + let active = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: false, + search_term: None, + }) + .await + .expect("active listing"); + let archived = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: true, + search_term: None, + }) + .await + .expect("archived listing"); + + let active_id = ThreadId::from_string(&active_uuid.to_string()).expect("valid thread id"); + let archived_id = + ThreadId::from_string(&archived_uuid.to_string()).expect("valid thread id"); + assert_eq!( + active + .items + .iter() + .map(|item| item.thread_id) + .collect::>(), + vec![active_id] + ); + assert_eq!( + archived + .items + .iter() + .map(|item| item.thread_id) + .collect::>(), + vec![archived_id] + ); + assert_eq!(active.items[0].archived_at, None); + assert_eq!( + archived.items[0].archived_at, + Some(archived.items[0].updated_at) + ); + } + + #[tokio::test] + async fn list_threads_returns_local_rollout_summary() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config); + let uuid = Uuid::from_u128(101); + let path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + + let page = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: vec![SessionSource::Cli], + model_providers: Some(vec!["test-provider".to_string()]), + archived: false, + search_term: None, + }) + .await + .expect("thread listing"); + + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + assert_eq!(page.next_cursor, None); + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].thread_id, thread_id); + assert_eq!(page.items[0].rollout_path, Some(path)); + assert_eq!(page.items[0].preview, "Hello from user"); + assert_eq!( + page.items[0].first_user_message.as_deref(), + Some("Hello from user") + ); + assert_eq!(page.items[0].model_provider, "test-provider"); + assert_eq!(page.items[0].cli_version, "test_version"); + assert_eq!(page.items[0].source, SessionSource::Cli); + } + + #[tokio::test] + async fn list_threads_rejects_invalid_cursor() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + + let err = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: Some("not-a-cursor".to_string()), + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: false, + search_term: None, + }) + .await + .expect_err("invalid cursor should fail"); + + assert!(matches!(err, ThreadStoreError::InvalidRequest { .. })); + } +} diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index cc15033cd..81428cac8 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -1,15 +1,14 @@ +mod archive_thread; +mod helpers; +mod list_threads; +mod read_thread; +mod unarchive_thread; + +#[cfg(test)] +mod test_support; + use async_trait::async_trait; -use chrono::DateTime; -use chrono::Utc; -use codex_git_utils::GitSha; -use codex_protocol::protocol::AskForApproval; -use codex_protocol::protocol::GitInfo; -use codex_protocol::protocol::SandboxPolicy; -use codex_protocol::protocol::SessionSource; use codex_rollout::RolloutConfig; -use codex_rollout::RolloutRecorder; -use codex_rollout::ThreadItem; -use codex_rollout::parse_cursor; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; @@ -23,7 +22,6 @@ use crate::StoredThread; use crate::StoredThreadHistory; use crate::ThreadPage; use crate::ThreadRecorder; -use crate::ThreadSortKey; use crate::ThreadStore; use crate::ThreadStoreError; use crate::ThreadStoreResult; @@ -32,7 +30,7 @@ use crate::UpdateThreadMetadataParams; /// Local filesystem/SQLite-backed implementation of [`ThreadStore`]. #[derive(Clone, Debug)] pub struct LocalThreadStore { - config: RolloutConfig, + pub(super) config: RolloutConfig, } impl LocalThreadStore { @@ -69,44 +67,12 @@ impl ThreadStore for LocalThreadStore { unsupported("load_history") } - async fn read_thread(&self, _params: ReadThreadParams) -> ThreadStoreResult { - unsupported("read_thread") + async fn read_thread(&self, params: ReadThreadParams) -> ThreadStoreResult { + read_thread::read_thread(self, params).await } async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult { - let cursor = params - .cursor - .as_deref() - .map(|cursor| { - parse_cursor(cursor).ok_or_else(|| ThreadStoreError::InvalidRequest { - message: format!("invalid cursor: {cursor}"), - }) - }) - .transpose()?; - let sort_key = match params.sort_key { - ThreadSortKey::CreatedAt => codex_rollout::ThreadSortKey::CreatedAt, - ThreadSortKey::UpdatedAt => codex_rollout::ThreadSortKey::UpdatedAt, - }; - let page = list_rollout_threads(&self.config, ¶ms, cursor.as_ref(), sort_key).await?; - - let next_cursor = page - .next_cursor - .as_ref() - .and_then(|cursor| serde_json::to_value(cursor).ok()) - .and_then(|value| value.as_str().map(str::to_owned)); - let items = page - .items - .into_iter() - .filter_map(|item| { - stored_thread_from_rollout_item( - item, - params.archived, - self.config.model_provider_id.as_str(), - ) - }) - .collect::>(); - - Ok(ThreadPage { items, next_cursor }) + list_threads::list_threads(self, params).await } async fn set_thread_name(&self, _params: SetThreadNameParams) -> ThreadStoreResult<()> { @@ -120,15 +86,15 @@ impl ThreadStore for LocalThreadStore { unsupported("update_thread_metadata") } - async fn archive_thread(&self, _params: ArchiveThreadParams) -> ThreadStoreResult<()> { - unsupported("archive_thread") + async fn archive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreResult<()> { + archive_thread::archive_thread(self, params).await } async fn unarchive_thread( &self, - _params: ArchiveThreadParams, + params: ArchiveThreadParams, ) -> ThreadStoreResult { - unsupported("unarchive_thread") + unarchive_thread::unarchive_thread(self, params).await } } @@ -137,429 +103,3 @@ fn unsupported(operation: &str) -> ThreadStoreResult { message: format!("local thread store does not implement {operation} in this slice"), }) } - -async fn list_rollout_threads( - config: &RolloutConfig, - params: &ListThreadsParams, - cursor: Option<&codex_rollout::Cursor>, - sort_key: codex_rollout::ThreadSortKey, -) -> ThreadStoreResult { - let page = if params.archived { - RolloutRecorder::list_archived_threads( - config, - params.page_size, - cursor, - sort_key, - params.allowed_sources.as_slice(), - params.model_providers.as_deref(), - config.model_provider_id.as_str(), - params.search_term.as_deref(), - ) - .await - } else { - RolloutRecorder::list_threads( - config, - params.page_size, - cursor, - sort_key, - params.allowed_sources.as_slice(), - params.model_providers.as_deref(), - config.model_provider_id.as_str(), - params.search_term.as_deref(), - ) - .await - }; - page.map_err(|err| ThreadStoreError::Internal { - message: format!("failed to list threads: {err}"), - }) -} - -fn stored_thread_from_rollout_item( - item: ThreadItem, - archived: bool, - default_provider: &str, -) -> Option { - let thread_id = item - .thread_id - .or_else(|| thread_id_from_rollout_path(item.path.as_path()))?; - let created_at = parse_rfc3339(item.created_at.as_deref()).unwrap_or_else(Utc::now); - let updated_at = parse_rfc3339(item.updated_at.as_deref()).unwrap_or(created_at); - let archived_at = archived.then_some(updated_at); - let git_info = git_info_from_parts( - item.git_sha.clone(), - item.git_branch.clone(), - item.git_origin_url.clone(), - ); - let source = item.source.unwrap_or(SessionSource::Unknown); - let preview = item.first_user_message.clone().unwrap_or_default(); - - Some(StoredThread { - thread_id, - rollout_path: Some(item.path), - forked_from_id: None, - preview, - name: None, - model_provider: item - .model_provider - .filter(|provider| !provider.is_empty()) - .unwrap_or_else(|| default_provider.to_string()), - model: None, - reasoning_effort: None, - created_at, - updated_at, - archived_at, - cwd: item.cwd.unwrap_or_default(), - cli_version: item.cli_version.unwrap_or_default(), - source, - agent_nickname: item.agent_nickname, - agent_role: item.agent_role, - agent_path: None, - git_info, - approval_mode: AskForApproval::OnRequest, - sandbox_policy: SandboxPolicy::new_read_only_policy(), - token_usage: None, - first_user_message: item.first_user_message, - history: None, - }) -} - -fn parse_rfc3339(value: Option<&str>) -> Option> { - DateTime::parse_from_rfc3339(value?) - .ok() - .map(|dt| dt.with_timezone(&Utc)) -} - -fn git_info_from_parts( - sha: Option, - branch: Option, - origin_url: Option, -) -> Option { - if sha.is_none() && branch.is_none() && origin_url.is_none() { - return None; - } - Some(GitInfo { - commit_hash: sha.as_deref().map(GitSha::new), - branch, - repository_url: origin_url, - }) -} - -fn thread_id_from_rollout_path(path: &std::path::Path) -> Option { - let file_name = path.file_name()?.to_str()?; - let stem = file_name.strip_suffix(".jsonl")?; - if stem.len() < 37 { - return None; - } - let uuid_start = stem.len().saturating_sub(36); - if !stem[..uuid_start].ends_with('-') { - return None; - } - codex_protocol::ThreadId::from_string(&stem[uuid_start..]).ok() -} - -#[cfg(test)] -mod tests { - use std::fs; - use std::io::Write; - use std::path::Path; - use std::path::PathBuf; - - use codex_protocol::ThreadId; - use codex_protocol::protocol::SessionSource; - use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; - use pretty_assertions::assert_eq; - use tempfile::TempDir; - use uuid::Uuid; - - use super::*; - - fn test_config(codex_home: &Path) -> RolloutConfig { - RolloutConfig { - codex_home: codex_home.to_path_buf(), - sqlite_home: codex_home.to_path_buf(), - cwd: codex_home.to_path_buf(), - model_provider_id: "test-provider".to_string(), - generate_memories: true, - } - } - - fn write_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::Result { - write_session_file_with( - root, - root.join("sessions/2025/01/03"), - ts, - uuid, - "Hello from user", - Some("test-provider"), - ) - } - - fn write_archived_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::Result { - write_session_file_with( - root, - root.join(ARCHIVED_SESSIONS_SUBDIR), - ts, - uuid, - "Archived user message", - Some("test-provider"), - ) - } - - fn write_session_file_with( - root: &Path, - day_dir: PathBuf, - ts: &str, - uuid: Uuid, - first_user_message: &str, - model_provider: Option<&str>, - ) -> std::io::Result { - fs::create_dir_all(&day_dir)?; - let path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl")); - let mut file = fs::File::create(&path)?; - let meta = serde_json::json!({ - "timestamp": ts, - "type": "session_meta", - "payload": { - "id": uuid, - "timestamp": ts, - "cwd": root, - "originator": "test_originator", - "cli_version": "test_version", - "source": "cli", - "model_provider": model_provider, - "git": { - "commit_hash": "abcdef", - "branch": "main", - "repository_url": "https://example.com/repo.git" - } - }, - }); - writeln!(file, "{meta}")?; - let user_event = serde_json::json!({ - "timestamp": ts, - "type": "event_msg", - "payload": { - "type": "user_message", - "message": first_user_message, - "kind": "plain", - }, - }); - writeln!(file, "{user_event}")?; - Ok(path) - } - - #[tokio::test] - async fn list_threads_uses_default_provider_when_rollout_omits_provider() { - let home = TempDir::new().expect("temp dir"); - let store = LocalThreadStore::new(test_config(home.path())); - write_session_file_with( - home.path(), - home.path().join("sessions/2025/01/03"), - "2025-01-03T12-00-00", - Uuid::from_u128(102), - "Hello from user", - /*model_provider*/ None, - ) - .expect("session file"); - - let page = store - .list_threads(ListThreadsParams { - page_size: 10, - cursor: None, - sort_key: ThreadSortKey::CreatedAt, - allowed_sources: Vec::new(), - model_providers: None, - archived: false, - search_term: None, - }) - .await - .expect("thread listing"); - - assert_eq!(page.items.len(), 1); - assert_eq!(page.items[0].model_provider, "test-provider"); - } - - #[tokio::test] - async fn list_threads_preserves_sqlite_title_search_results() { - let home = TempDir::new().expect("temp dir"); - let config = test_config(home.path()); - let store = LocalThreadStore::new(config.clone()); - let uuid = Uuid::from_u128(103); - let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); - let rollout_path = home.path().join("rollout-title-search.jsonl"); - fs::write(&rollout_path, "").expect("placeholder rollout file"); - - let runtime = codex_state::StateRuntime::init( - home.path().to_path_buf(), - config.model_provider_id.clone(), - ) - .await - .expect("state db should initialize"); - runtime - .mark_backfill_complete(/*last_watermark*/ None) - .await - .expect("backfill should be complete"); - let created_at = Utc::now(); - let mut builder = codex_state::ThreadMetadataBuilder::new( - thread_id, - rollout_path, - created_at, - SessionSource::Cli, - ); - builder.model_provider = Some(config.model_provider_id.clone()); - builder.cwd = home.path().to_path_buf(); - builder.cli_version = Some("test_version".to_string()); - let mut metadata = builder.build(config.model_provider_id.as_str()); - metadata.title = "needle title".to_string(); - metadata.first_user_message = Some("plain preview".to_string()); - runtime - .upsert_thread(&metadata) - .await - .expect("state db upsert should succeed"); - - let page = store - .list_threads(ListThreadsParams { - page_size: 10, - cursor: None, - sort_key: ThreadSortKey::CreatedAt, - allowed_sources: Vec::new(), - model_providers: None, - archived: false, - search_term: Some("needle".to_string()), - }) - .await - .expect("thread listing"); - - let ids = page - .items - .iter() - .map(|item| item.thread_id) - .collect::>(); - assert_eq!(ids, vec![thread_id]); - assert_eq!( - page.items[0].first_user_message.as_deref(), - Some("plain preview") - ); - } - - #[tokio::test] - async fn list_threads_selects_active_or_archived_collection() { - let home = TempDir::new().expect("temp dir"); - let store = LocalThreadStore::new(test_config(home.path())); - let active_uuid = Uuid::from_u128(105); - let archived_uuid = Uuid::from_u128(106); - write_session_file(home.path(), "2025-01-03T12-00-00", active_uuid) - .expect("active session file"); - write_archived_session_file(home.path(), "2025-01-03T13-00-00", archived_uuid) - .expect("archived session file"); - - let active = store - .list_threads(ListThreadsParams { - page_size: 10, - cursor: None, - sort_key: ThreadSortKey::CreatedAt, - allowed_sources: Vec::new(), - model_providers: None, - archived: false, - search_term: None, - }) - .await - .expect("active listing"); - let archived = store - .list_threads(ListThreadsParams { - page_size: 10, - cursor: None, - sort_key: ThreadSortKey::CreatedAt, - allowed_sources: Vec::new(), - model_providers: None, - archived: true, - search_term: None, - }) - .await - .expect("archived listing"); - - let active_id = ThreadId::from_string(&active_uuid.to_string()).expect("valid thread id"); - let archived_id = - ThreadId::from_string(&archived_uuid.to_string()).expect("valid thread id"); - assert_eq!( - active - .items - .iter() - .map(|item| item.thread_id) - .collect::>(), - vec![active_id] - ); - assert_eq!( - archived - .items - .iter() - .map(|item| item.thread_id) - .collect::>(), - vec![archived_id] - ); - assert_eq!(active.items[0].archived_at, None); - assert_eq!( - archived.items[0].archived_at, - Some(archived.items[0].updated_at) - ); - } - - #[tokio::test] - async fn list_threads_returns_local_rollout_summary() { - let home = TempDir::new().expect("temp dir"); - let config = test_config(home.path()); - let store = LocalThreadStore::new(config); - let uuid = Uuid::from_u128(101); - let path = - write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); - - let page = store - .list_threads(ListThreadsParams { - page_size: 10, - cursor: None, - sort_key: ThreadSortKey::CreatedAt, - allowed_sources: vec![SessionSource::Cli], - model_providers: Some(vec!["test-provider".to_string()]), - archived: false, - search_term: None, - }) - .await - .expect("thread listing"); - - let thread_id = - codex_protocol::ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); - assert_eq!(page.next_cursor, None); - assert_eq!(page.items.len(), 1); - assert_eq!(page.items[0].thread_id, thread_id); - assert_eq!(page.items[0].rollout_path, Some(path)); - assert_eq!(page.items[0].preview, "Hello from user"); - assert_eq!( - page.items[0].first_user_message.as_deref(), - Some("Hello from user") - ); - assert_eq!(page.items[0].model_provider, "test-provider"); - assert_eq!(page.items[0].cli_version, "test_version"); - assert_eq!(page.items[0].source, SessionSource::Cli); - } - - #[tokio::test] - async fn list_threads_rejects_invalid_cursor() { - let home = TempDir::new().expect("temp dir"); - let store = LocalThreadStore::new(test_config(home.path())); - - let err = store - .list_threads(ListThreadsParams { - page_size: 10, - cursor: Some("not-a-cursor".to_string()), - sort_key: ThreadSortKey::CreatedAt, - allowed_sources: Vec::new(), - model_providers: None, - archived: false, - search_term: None, - }) - .await - .expect_err("invalid cursor should fail"); - - assert!(matches!(err, ThreadStoreError::InvalidRequest { .. })); - } -} diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs new file mode 100644 index 000000000..189bc1938 --- /dev/null +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -0,0 +1,138 @@ +use codex_rollout::RolloutRecorder; +use codex_rollout::find_archived_thread_path_by_id_str; +use codex_rollout::find_thread_path_by_id_str; +use codex_rollout::read_thread_item_from_rollout; + +use super::LocalThreadStore; +use super::helpers::stored_thread_from_rollout_item; +use crate::ReadThreadParams; +use crate::StoredThread; +use crate::StoredThreadHistory; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +pub(super) async fn read_thread( + store: &LocalThreadStore, + params: ReadThreadParams, +) -> ThreadStoreResult { + let thread_id = params.thread_id; + let path = if params.include_archived { + match find_thread_path_by_id_str(store.config.codex_home.as_path(), &thread_id.to_string()) + .await + .map_err(|err| ThreadStoreError::InvalidRequest { + message: format!("failed to locate thread id {thread_id}: {err}"), + })? { + Some(path) => Some(path), + None => find_archived_thread_path_by_id_str( + store.config.codex_home.as_path(), + &thread_id.to_string(), + ) + .await + .map_err(|err| ThreadStoreError::InvalidRequest { + message: format!("failed to locate archived thread id {thread_id}: {err}"), + })?, + } + } else { + find_thread_path_by_id_str(store.config.codex_home.as_path(), &thread_id.to_string()) + .await + .map_err(|err| ThreadStoreError::InvalidRequest { + message: format!("failed to locate thread id {thread_id}: {err}"), + })? + } + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("no rollout found for thread id {thread_id}"), + })?; + + let item = read_thread_item_from_rollout(path.clone()) + .await + .ok_or_else(|| ThreadStoreError::Internal { + message: format!("failed to read thread {}", path.display()), + })?; + let archived = item.path.starts_with( + store + .config + .codex_home + .join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR), + ); + let mut thread = + stored_thread_from_rollout_item(item, archived, store.config.model_provider_id.as_str()) + .ok_or_else(|| ThreadStoreError::Internal { + message: format!("failed to read thread id from {}", path.display()), + })?; + if params.include_history { + let (items, _, _) = RolloutRecorder::load_rollout_items(path.as_path()) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to load thread history {}: {err}", path.display()), + })?; + thread.history = Some(StoredThreadHistory { thread_id, items }); + } + Ok(thread) +} + +#[cfg(test)] +mod tests { + use codex_protocol::ThreadId; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + use uuid::Uuid; + + use super::*; + use crate::ThreadStore; + use crate::local::LocalThreadStore; + use crate::local::test_support::test_config; + use crate::local::test_support::write_session_file; + + #[tokio::test] + async fn read_thread_returns_active_rollout_summary() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let uuid = Uuid::from_u128(205); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let active_path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: true, + }) + .await + .expect("read thread"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.rollout_path, Some(active_path)); + assert_eq!(thread.archived_at, None); + assert_eq!(thread.preview, "Hello from user"); + assert_eq!( + thread.history.expect("history should load").thread_id, + thread_id + ); + } + + #[tokio::test] + async fn read_thread_fails_without_rollout() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let uuid = Uuid::from_u128(206); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + + let err = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect_err("read should fail without rollout"); + + let ThreadStoreError::InvalidRequest { message } = err else { + panic!("expected invalid request error"); + }; + assert_eq!( + message, + format!("no rollout found for thread id {thread_id}") + ); + } +} diff --git a/codex-rs/thread-store/src/local/test_support.rs b/codex-rs/thread-store/src/local/test_support.rs new file mode 100644 index 000000000..d8ddcf8dd --- /dev/null +++ b/codex-rs/thread-store/src/local/test_support.rs @@ -0,0 +1,87 @@ +use std::fs; +use std::io::Write; +use std::path::Path; +use std::path::PathBuf; + +use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; +use codex_rollout::RolloutConfig; +use uuid::Uuid; + +pub(super) fn test_config(codex_home: &Path) -> RolloutConfig { + RolloutConfig { + codex_home: codex_home.to_path_buf(), + sqlite_home: codex_home.to_path_buf(), + cwd: codex_home.to_path_buf(), + model_provider_id: "test-provider".to_string(), + generate_memories: true, + } +} + +pub(super) fn write_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::Result { + write_session_file_with( + root, + root.join("sessions/2025/01/03"), + ts, + uuid, + "Hello from user", + Some("test-provider"), + ) +} + +pub(super) fn write_archived_session_file( + root: &Path, + ts: &str, + uuid: Uuid, +) -> std::io::Result { + write_session_file_with( + root, + root.join(ARCHIVED_SESSIONS_SUBDIR), + ts, + uuid, + "Archived user message", + Some("test-provider"), + ) +} + +pub(super) fn write_session_file_with( + root: &Path, + day_dir: PathBuf, + ts: &str, + uuid: Uuid, + first_user_message: &str, + model_provider: Option<&str>, +) -> std::io::Result { + fs::create_dir_all(&day_dir)?; + let path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl")); + let mut file = fs::File::create(&path)?; + let meta = serde_json::json!({ + "timestamp": ts, + "type": "session_meta", + "payload": { + "id": uuid, + "timestamp": ts, + "cwd": root, + "originator": "test_originator", + "cli_version": "test_version", + "source": "cli", + "model_provider": model_provider, + "git": { + "commit_hash": "abcdef", + "branch": "main", + "repository_url": "https://example.com/repo.git" + } + }, + }); + writeln!(file, "{meta}")?; + let user_event = serde_json::json!({ + "timestamp": ts, + "type": "event_msg", + "payload": { + "type": "user_message", + "message": first_user_message, + "kind": "plain", + }, + }); + writeln!(file, "{user_event}")?; + Ok(path) +} diff --git a/codex-rs/thread-store/src/local/unarchive_thread.rs b/codex-rs/thread-store/src/local/unarchive_thread.rs new file mode 100644 index 000000000..17e484bac --- /dev/null +++ b/codex-rs/thread-store/src/local/unarchive_thread.rs @@ -0,0 +1,198 @@ +use codex_rollout::find_archived_thread_path_by_id_str; +use codex_rollout::read_thread_item_from_rollout; +use codex_rollout::rollout_date_parts; + +use super::LocalThreadStore; +use super::helpers::matching_rollout_file_name; +use super::helpers::scoped_rollout_path; +use super::helpers::stored_thread_from_rollout_item; +use super::helpers::touch_modified_time; +use crate::ArchiveThreadParams; +use crate::StoredThread; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +pub(super) async fn unarchive_thread( + store: &LocalThreadStore, + params: ArchiveThreadParams, +) -> ThreadStoreResult { + let thread_id = params.thread_id; + let archived_path = find_archived_thread_path_by_id_str( + store.config.codex_home.as_path(), + &thread_id.to_string(), + ) + .await + .map_err(|err| ThreadStoreError::InvalidRequest { + message: format!("failed to locate archived thread id {thread_id}: {err}"), + })? + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("no archived rollout found for thread id {thread_id}"), + })?; + + let canonical_archived_path = scoped_rollout_path( + store + .config + .codex_home + .join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR), + archived_path.as_path(), + "archived", + )?; + let file_name = matching_rollout_file_name( + canonical_archived_path.as_path(), + thread_id, + archived_path.as_path(), + )?; + let Some((year, month, day)) = rollout_date_parts(&file_name) else { + return Err(ThreadStoreError::InvalidRequest { + message: format!( + "rollout path `{}` missing filename timestamp", + archived_path.display() + ), + }); + }; + + let dest_dir = store + .config + .codex_home + .join(codex_rollout::SESSIONS_SUBDIR) + .join(year) + .join(month) + .join(day); + std::fs::create_dir_all(&dest_dir).map_err(|err| ThreadStoreError::Internal { + message: format!("failed to unarchive thread: {err}"), + })?; + let restored_path = dest_dir.join(&file_name); + std::fs::rename(&canonical_archived_path, &restored_path).map_err(|err| { + ThreadStoreError::Internal { + message: format!("failed to unarchive thread: {err}"), + } + })?; + touch_modified_time(restored_path.as_path()).map_err(|err| ThreadStoreError::Internal { + message: format!("failed to update unarchived thread timestamp: {err}"), + })?; + + if let Some(ctx) = codex_rollout::state_db::get_state_db(&store.config).await { + let _ = ctx + .mark_unarchived(thread_id, restored_path.as_path()) + .await; + } + + let item = read_thread_item_from_rollout(restored_path.clone()) + .await + .ok_or_else(|| ThreadStoreError::Internal { + message: format!( + "failed to read unarchived thread {}", + restored_path.display() + ), + })?; + stored_thread_from_rollout_item( + item, + /*archived*/ false, + store.config.model_provider_id.as_str(), + ) + .ok_or_else(|| ThreadStoreError::Internal { + message: format!( + "failed to read unarchived thread id from {}", + restored_path.display() + ), + }) +} + +#[cfg(test)] +mod tests { + use chrono::Utc; + use codex_protocol::ThreadId; + use codex_protocol::protocol::SessionSource; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + use uuid::Uuid; + + use super::*; + use crate::ThreadStore; + use crate::local::LocalThreadStore; + use crate::local::test_support::test_config; + use crate::local::test_support::write_archived_session_file; + + #[tokio::test] + async fn unarchive_thread_restores_rollout_and_returns_updated_thread() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let uuid = Uuid::from_u128(203); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let archived_path = write_archived_session_file(home.path(), "2025-01-03T13-00-00", uuid) + .expect("archived session file"); + + let thread = store + .unarchive_thread(ArchiveThreadParams { thread_id }) + .await + .expect("unarchive thread"); + + assert!(!archived_path.exists()); + let restored_path = home + .path() + .join("sessions/2025/01/03") + .join(archived_path.file_name().expect("file name")); + assert!(restored_path.exists()); + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.rollout_path, Some(restored_path)); + assert_eq!(thread.archived_at, None); + assert_eq!(thread.preview, "Archived user message"); + assert_eq!( + thread.first_user_message.as_deref(), + Some("Archived user message") + ); + } + + #[tokio::test] + async fn unarchive_thread_updates_sqlite_metadata_when_present() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(204); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let archived_path = write_archived_session_file(home.path(), "2025-01-03T13-00-00", uuid) + .expect("archived session file"); + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .expect("backfill should be complete"); + let mut builder = codex_state::ThreadMetadataBuilder::new( + thread_id, + archived_path.clone(), + Utc::now(), + SessionSource::Cli, + ); + builder.model_provider = Some(config.model_provider_id.clone()); + builder.cwd = home.path().to_path_buf(); + builder.cli_version = Some("test_version".to_string()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.archived_at = Some(metadata.updated_at); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + store + .unarchive_thread(ArchiveThreadParams { thread_id }) + .await + .expect("unarchive thread"); + + let restored_path = home + .path() + .join("sessions/2025/01/03") + .join(archived_path.file_name().expect("file name")); + let updated = runtime + .get_thread(thread_id) + .await + .expect("state db read should succeed") + .expect("thread metadata should exist"); + assert_eq!(updated.rollout_path, restored_path); + assert_eq!(updated.archived_at, None); + } +}