Migrate archive/unarchive to local ThreadStore (#17892)

# Summary
- implement local ThreadStore archive/unarchive operations
- implement local ThreadStore read_thread operation
- break up the various ThreadStore local method implementations into
separate files
- migrate app-server archive/unarchive and core archive fixture to use
ThreadStore (but not all read operations yet!)
- use the ThreadStore's read operation as a proxy check for thread
persistence/existence in the app server code
- move all other filesystem operations related to archive (path
validation etc) into the local thread store.

# Tests
- add dedicated local store archive/unarchive tests

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<ThreadPage> {
+    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, &params, 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::<Vec<_>>();
+
+    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<codex_rollout::ThreadsPage> {
+    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::<Vec<_>>();
+        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<_>>(),
+            vec![active_id]
+        );
+        assert_eq!(
+            archived
+                .items
+                .iter()
+                .map(|item| item.thread_id)
+                .collect::<Vec<_>>(),
+            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 { .. }));
+    }
+}