perf(sessions): 缓存会话索引并按页读取详情

apps/codex-plus-manager/src-tauri/src/commands.rs

@@ -4133,7 +4133,33 @@ pub fn dismiss_pending_provider_import() -> CommandResult<PendingProviderImportP
 }
 
 #[tauri::command]
-pub fn list_local_sessions(
+pub async fn list_local_sessions(
+    request: Option<ListLocalSessionsRequest>,
+) -> CommandResult<LocalSessionsPayload> {
+    let offset = request.as_ref().map_or(0, |request| request.offset);
+    let limit = request
+        .as_ref()
+        .map_or(DEFAULT_LOCAL_SESSIONS_PAGE_SIZE, |request| request.limit)
+        .clamp(1, MAX_LOCAL_SESSIONS_PAGE_SIZE);
+    match tauri::async_runtime::spawn_blocking(move || list_local_sessions_blocking(request)).await
+    {
+        Ok(result) => result,
+        Err(error) => failed(
+            &format!("会话查询后台任务失败：{error}"),
+            LocalSessionsPayload {
+                db_path: String::new(),
+                db_paths: Vec::new(),
+                sessions: Vec::new(),
+                offset,
+                limit,
+                has_more: false,
+                total_count: 0,
+            },
+        ),
+    }
+}
+
+fn list_local_sessions_blocking(
     request: Option<ListLocalSessionsRequest>,
 ) -> CommandResult<LocalSessionsPayload> {
     let request = request.unwrap_or(ListLocalSessionsRequest {
@@ -4142,40 +4168,18 @@ pub fn list_local_sessions(
     });
     let offset = request.offset;
     let limit = request.limit.clamp(1, MAX_LOCAL_SESSIONS_PAGE_SIZE);
-    let fetch_limit = offset.saturating_add(limit).saturating_add(1);
     let home = codex_plus_core::codex_sqlite::default_codex_home_dir();
     let db_paths = codex_plus_core::codex_sqlite::codex_session_db_paths_from_home(&home);
-    let mut sessions = Vec::new();
-    let mut session_ids = std::collections::HashSet::new();
-    let mut errors = Vec::new();
-    for db_path in &db_paths {
-        let adapter = local_session_adapter(db_path);
-        match adapter.list_local_session_ids() {
-            Ok(ids) => session_ids.extend(ids),
-            Err(error) if db_path.exists() => {
-                errors.push(format!("{}: {error}", db_path.to_string_lossy()));
-                continue;
-            }
-            Err(_) => continue,
-        }
-        match adapter.list_local_sessions_limited(fetch_limit) {
-            Ok(mut items) => sessions.append(&mut items),
-            Err(error) if db_path.exists() => {
-                errors.push(format!("{}: {error}", db_path.to_string_lossy()));
-            }
-            Err(_) => {}
-        }
-    }
-    sessions.sort_by(|left, right| {
-        right
-            .updated_at_ms
-            .cmp(&left.updated_at_ms)
-            .then_with(|| right.id.cmp(&left.id))
-    });
-    let mut seen_session_ids = std::collections::HashSet::new();
-    sessions.retain(|session| seen_session_ids.insert(session.id.clone()));
-    let has_more = sessions.len() > offset.saturating_add(limit);
-    let sessions = sessions.into_iter().skip(offset).take(limit).collect();
+    static PAGER: std::sync::OnceLock<std::sync::Mutex<codex_plus_data::LocalSessionPager>> =
+        std::sync::OnceLock::new();
+    let pager = PAGER.get_or_init(Default::default);
+    let page = match pager.lock() {
+        Ok(mut pager) => pager.list_page(&db_paths, offset, limit),
+        Err(_) => codex_plus_data::LocalSessionPage {
+            errors: vec!["会话查询缓存不可用，请重启管理窗口。".to_string()],
+            ..Default::default()
+        },
+    };
     let payload = LocalSessionsPayload {
         db_path: db_paths
             .first()
@@ -4185,24 +4189,24 @@ pub fn list_local_sessions(
             .iter()
             .map(|path| path.to_string_lossy().to_string())
             .collect(),
-        sessions,
+        sessions: page.sessions,
         offset,
         limit,
-        has_more,
-        total_count: session_ids.len(),
+        has_more: page.has_more,
+        total_count: page.total_count,
     };
-    let page = offset / limit + 1;
-    if errors.is_empty() {
+    let page_number = (offset / limit).saturating_add(1);
+    if page.errors.is_empty() {
         ok(
             &format!(
-                "已读取第 {page} 页，共 {} 个本地会话。",
+                "已读取第 {page_number} 页，共 {} 个本地会话。",
                 payload.sessions.len()
             ),
             payload,
         )
     } else {
         failed(
-            &format!("读取部分本地会话失败：{}", errors.join("; ")),
+            &format!("读取部分本地会话失败：{}", page.errors.join("; ")),
             payload,
         )
     }
@@ -4288,15 +4292,6 @@ pub fn delete_local_session(request: DeleteLocalSessionRequest) -> CommandResult
     }
 }
 
-fn local_session_adapter(db_path: &Path) -> codex_plus_data::SQLiteStorageAdapter {
-    codex_plus_data::SQLiteStorageAdapter::new(
-        db_path,
-        codex_plus_data::BackupStore::new(
-            codex_plus_core::paths::default_app_state_dir().join("backups"),
-        ),
-    )
-}
-
 /// 归一化「Codex 应用路径」。**无效路径一律丢弃，不落库。**
 ///
 /// 之前的写法是 normalize 成功才覆盖、失败就原样保留，于是误选的路径会被存进
@@ -9218,7 +9213,7 @@ base_url = "https://example.invalid/v1"
         unsafe {
             std::env::set_var("CODEX_HOME", &codex_home);
         }
-        let result = list_local_sessions(None);
+        let result = tauri::async_runtime::block_on(list_local_sessions(None));
 
         assert_eq!(result.status, "ok");
         assert_eq!(result.payload.sessions.len(), 1);
@@ -9239,7 +9234,7 @@ base_url = "https://example.invalid/v1"
             .execute("INSERT INTO threads VALUES ('t3', '', 'Oldest', 50)", [])
             .unwrap();
 
-        let first_page = list_local_sessions(Some(ListLocalSessionsRequest {
+        let first_page = list_local_sessions_blocking(Some(ListLocalSessionsRequest {
             offset: 0,
             limit: 2,
         }));
@@ -9249,7 +9244,7 @@ base_url = "https://example.invalid/v1"
         assert!(first_page.payload.has_more);
         assert_eq!(first_page.payload.total_count, 3);
 
-        let second_page = list_local_sessions(Some(ListLocalSessionsRequest {
+        let second_page = list_local_sessions_blocking(Some(ListLocalSessionsRequest {
             offset: 2,
             limit: 2,
         }));
@@ -9261,6 +9256,40 @@ base_url = "https://example.invalid/v1"
         assert_eq!(second_page.payload.total_count, 3);
     }
 
+    #[test]
+    fn list_local_sessions_returns_partial_payload_and_clamps_extreme_pages() {
+        let _codex_home_guard = lock_codex_home_for_test();
+        let temp = tempfile::tempdir().unwrap();
+        let previous_codex_home = std::env::var_os("CODEX_HOME");
+        let home = temp.path().join("codex-home");
+        let sqlite = home.join("sqlite");
+        std::fs::create_dir_all(&sqlite).unwrap();
+        create_minimal_thread_db(&sqlite.join("state_5.sqlite"), "valid", "Valid", 10);
+        std::fs::write(home.join("state_5.sqlite"), b"not a database").unwrap();
+        unsafe {
+            std::env::set_var("CODEX_HOME", &home);
+        }
+        let first =
+            tauri::async_runtime::block_on(list_local_sessions(Some(ListLocalSessionsRequest {
+                offset: 0,
+                limit: 0,
+            })));
+        let beyond = list_local_sessions_blocking(Some(ListLocalSessionsRequest {
+            offset: usize::MAX,
+            limit: usize::MAX,
+        }));
+        restore_codex_home(previous_codex_home);
+        assert_eq!(first.status, "failed");
+        assert_eq!(first.payload.limit, 1);
+        assert_eq!(first.payload.total_count, 1);
+        assert_eq!(first.payload.sessions[0].id, "valid");
+        assert_eq!(beyond.status, "failed");
+        assert_eq!(beyond.payload.limit, MAX_LOCAL_SESSIONS_PAGE_SIZE);
+        assert_eq!(beyond.payload.total_count, 1);
+        assert!(beyond.payload.sessions.is_empty());
+        assert!(!beyond.payload.has_more);
+    }
+
     #[test]
     fn list_local_sessions_ignores_relation_only_thread_reference_dbs() {
         let _codex_home_guard = lock_codex_home_for_test();
@@ -9287,7 +9316,7 @@ base_url = "https://example.invalid/v1"
         unsafe {
             std::env::set_var("CODEX_HOME", &codex_home);
         }
-        let result = list_local_sessions(None);
+        let result = list_local_sessions_blocking(None);
         restore_codex_home(previous_codex_home);
 
         assert_eq!(result.status, "ok");

crates/codex-plus-data/src/lib.rs

@@ -23,4 +23,6 @@ pub use provider_sync::{
     run_remote_control_session_finalization_for_thread_with_target,
     try_acquire_provider_sync_lifecycle_guard, validate_provider_sync_target,
 };
-pub use storage::{LocalSession, SQLiteStorageAdapter, delete_local_from_paths};
+pub use storage::{
+    LocalSession, LocalSessionPage, LocalSessionPager, SQLiteStorageAdapter, delete_local_from_paths,
+};

crates/codex-plus-data/src/storage.rs

@@ -10,6 +10,9 @@ use std::fs::File;
 use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
 use std::path::{Path, PathBuf};
 
+mod session_paging;
+pub use session_paging::{LocalSessionPage, LocalSessionPager};
+
 pub fn delete_local_from_paths(
     db_paths: impl IntoIterator<Item = PathBuf>,
     backup_store: BackupStore,
@@ -267,11 +270,14 @@ impl SQLiteStorageAdapter {
             return Ok(Vec::new());
         }
         let db = Connection::open(&self.db_path)?;
-        match schema_kind(&db)? {
-            Some(SchemaKind::CodexThreads) => self.list_codex_threads(&db, limit),
-            Some(SchemaKind::CodexAutomationRuns) => self.list_codex_automation_runs(&db, limit),
-            _ => anyhow::bail!("Unsupported local storage schema"),
-        }
+        let query = session_paging::session_query(&db)?;
+        let mut stmt = db.prepare(&format!(
+            "{query} ORDER BY updated_at_ms DESC, id DESC LIMIT ?1"
+        ))?;
+        let rows = stmt.query_map([sqlite_limit(limit)], |row| {
+            session_paging::read_session(row, &self.db_path)
+        })?;
+        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
     }
 
     pub fn list_local_session_ids(&self) -> anyhow::Result<Vec<String>> {
@@ -294,94 +300,6 @@ impl SQLiteStorageAdapter {
         Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
     }
 
-    fn list_codex_threads(
-        &self,
-        db: &Connection,
-        limit: usize,
-    ) -> anyhow::Result<Vec<LocalSession>> {
-        let columns = table_columns(&db, "threads")?
-            .into_iter()
-            .collect::<HashSet<_>>();
-        let title = optional_column_expression(&columns, "title", "''");
-        let cwd = optional_column_expression(&columns, "cwd", "''");
-        let model_provider = optional_column_expression(&columns, "model_provider", "''");
-        let archived = optional_column_expression(&columns, "archived", "0");
-        let updated_at_ms = if columns.contains("updated_at_ms") {
-            "updated_at_ms"
-        } else if columns.contains("updated_at") {
-            "updated_at * 1000"
-        } else if columns.contains("created_at_ms") {
-            "created_at_ms"
-        } else {
-            "NULL"
-        };
-        let rollout_path = optional_column_expression(&columns, "rollout_path", "''");
-        let child_thread_filter = codex_thread_filter(db)?;
-        let sql = format!(
-            "SELECT id, {title}, {cwd}, {model_provider}, {archived}, {updated_at_ms}, {rollout_path}
-             FROM threads
-             {child_thread_filter}
-             ORDER BY COALESCE({updated_at_ms}, 0) DESC, id DESC
-             LIMIT ?1"
-        );
-        let mut stmt = db.prepare(&sql)?;
-        let rows = stmt.query_map([sqlite_limit(limit)], |row| {
-            Ok(LocalSession {
-                id: row.get(0)?,
-                title: row.get::<_, Option<String>>(1)?.unwrap_or_default(),
-                cwd: row.get::<_, Option<String>>(2)?.unwrap_or_default(),
-                model_provider: row.get::<_, Option<String>>(3)?.unwrap_or_default(),
-                archived: row.get::<_, Option<i64>>(4)?.unwrap_or_default() != 0,
-                updated_at_ms: row.get(5)?,
-                rollout_path: row.get::<_, Option<String>>(6)?.unwrap_or_default(),
-                db_path: self.db_path.to_string_lossy().to_string(),
-            })
-        })?;
-        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
-    }
-
-    fn list_codex_automation_runs(
-        &self,
-        db: &Connection,
-        limit: usize,
-    ) -> anyhow::Result<Vec<LocalSession>> {
-        let columns = table_columns(db, "automation_runs")?
-            .into_iter()
-            .collect::<HashSet<_>>();
-        let title = optional_column_expression(&columns, "thread_title", "''");
-        let cwd = optional_column_expression(&columns, "source_cwd", "''");
-        let status = optional_column_expression(&columns, "status", "''");
-        let updated_at = optional_column_expression(&columns, "updated_at", "NULL");
-        let created_at = optional_column_expression(&columns, "created_at", "NULL");
-        let sql = format!(
-            "SELECT thread_id, {title}, {cwd}, {status}, {updated_at}, {created_at}
-             FROM automation_runs
-             WHERE COALESCE(thread_id, '') <> ''
-             ORDER BY COALESCE({updated_at}, {created_at}, 0) DESC, thread_id DESC
-             LIMIT ?1"
-        );
-        let mut stmt = db.prepare(&sql)?;
-        let rows = stmt.query_map([sqlite_limit(limit)], |row| {
-            let updated_at_ms = row
-                .get::<_, Option<i64>>(4)?
-                .or(row.get::<_, Option<i64>>(5)?);
-            Ok(LocalSession {
-                id: row.get(0)?,
-                title: row.get::<_, Option<String>>(1)?.unwrap_or_default(),
-                cwd: row.get::<_, Option<String>>(2)?.unwrap_or_default(),
-                model_provider: String::new(),
-                archived: row
-                    .get::<_, Option<String>>(3)?
-                    .map(|status| status.eq_ignore_ascii_case("archived"))
-                    .unwrap_or(false),
-                updated_at_ms,
-                rollout_path: String::new(),
-                db_path: self.db_path.to_string_lossy().to_string(),
-            })
-        })?;
-        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
-    }
-
     pub fn undo(&self, token: &str) -> DeleteResult {
         let result = (|| -> anyhow::Result<DeleteResult> {
             let backups = undo_backups(&self.backup_store, token)?;