diff --git a/apps/codex-plus-manager/src-tauri/src/commands.rs b/apps/codex-plus-manager/src-tauri/src/commands.rs index adf4d7797..77b0ffd71 100644 --- a/apps/codex-plus-manager/src-tauri/src/commands.rs +++ b/apps/codex-plus-manager/src-tauri/src/commands.rs @@ -4133,7 +4133,33 @@ pub fn dismiss_pending_provider_import() -> CommandResult, +) -> CommandResult { + 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, ) -> CommandResult { 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::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"); diff --git a/crates/codex-plus-data/src/lib.rs b/crates/codex-plus-data/src/lib.rs index 0ab9e8547..47bb9c583 100644 --- a/crates/codex-plus-data/src/lib.rs +++ b/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, +}; diff --git a/crates/codex-plus-data/src/storage.rs b/crates/codex-plus-data/src/storage.rs index c635c0e83..a5b4afd31 100644 --- a/crates/codex-plus-data/src/storage.rs +++ b/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, 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::>>()?) } pub fn list_local_session_ids(&self) -> anyhow::Result> { @@ -294,94 +300,6 @@ impl SQLiteStorageAdapter { Ok(rows.collect::>>()?) } - fn list_codex_threads( - &self, - db: &Connection, - limit: usize, - ) -> anyhow::Result> { - let columns = table_columns(&db, "threads")? - .into_iter() - .collect::>(); - 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>(1)?.unwrap_or_default(), - cwd: row.get::<_, Option>(2)?.unwrap_or_default(), - model_provider: row.get::<_, Option>(3)?.unwrap_or_default(), - archived: row.get::<_, Option>(4)?.unwrap_or_default() != 0, - updated_at_ms: row.get(5)?, - rollout_path: row.get::<_, Option>(6)?.unwrap_or_default(), - db_path: self.db_path.to_string_lossy().to_string(), - }) - })?; - Ok(rows.collect::>>()?) - } - - fn list_codex_automation_runs( - &self, - db: &Connection, - limit: usize, - ) -> anyhow::Result> { - let columns = table_columns(db, "automation_runs")? - .into_iter() - .collect::>(); - 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>(4)? - .or(row.get::<_, Option>(5)?); - Ok(LocalSession { - id: row.get(0)?, - title: row.get::<_, Option>(1)?.unwrap_or_default(), - cwd: row.get::<_, Option>(2)?.unwrap_or_default(), - model_provider: String::new(), - archived: row - .get::<_, Option>(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::>>()?) - } - pub fn undo(&self, token: &str) -> DeleteResult { let result = (|| -> anyhow::Result { let backups = undo_backups(&self.backup_store, token)?; diff --git a/crates/codex-plus-data/src/storage/session_paging.rs b/crates/codex-plus-data/src/storage/session_paging.rs new file mode 100644 index 000000000..fa8f95acb --- /dev/null +++ b/crates/codex-plus-data/src/storage/session_paging.rs @@ -0,0 +1,353 @@ +use super::{ + LocalSession, SchemaKind, codex_thread_filter, optional_column_expression, schema_kind, + table_columns, +}; +use rusqlite::{Connection, OpenFlags, Row, params_from_iter}; +use std::collections::HashSet; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant, SystemTime}; + +#[cfg(test)] +mod tests; + +// 文件还原可能保留时间戳及 SQLite 计数器，任何指纹命中最多复用 30 秒。 +const INDEX_MAX_AGE: Duration = Duration::from_secs(30); + +#[derive(Default)] +pub struct LocalSessionPage { + pub sessions: Vec, + pub total_count: usize, + pub has_more: bool, + pub errors: Vec, +} + +#[derive(Default)] +pub struct LocalSessionPager { + databases: Vec<(PathBuf, CachedDatabase)>, + // 只缓存 ID、排序键和来源，不缓存可能很长的标题或路径，也不常驻数据库连接。 + order: Vec<(usize, usize)>, + #[cfg(test)] + after_snapshot: Option>, +} + +#[derive(Default)] +struct CachedDatabase { + refreshed_at: Option, + stamp: Option, + query: String, + index: Vec, +} + +struct IndexEntry { + id: String, + updated_at_ms: Option, +} + +#[derive(PartialEq, Eq)] +struct FileStamp { + len: u64, + modified: Option, + created: Option, + header: Vec, +} + +impl FileStamp { + fn read(path: &Path, header_len: u64) -> std::io::Result> { + use std::io::Read; + let file = match std::fs::File::open(path) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error), + }; + let metadata = file.metadata()?; + let mut header = Vec::new(); + file.take(header_len).read_to_end(&mut header)?; + Ok(Some(Self { + len: metadata.len(), + modified: metadata.modified().ok(), + created: metadata.created().ok(), + header, + })) + } +} + +#[derive(PartialEq, Eq)] +struct DatabaseStamp { + main: FileStamp, + wal: Option, + shm: Option, +} + +impl DatabaseStamp { + fn read(path: &Path) -> std::io::Result> { + let Some(main) = FileStamp::read(path, 100)? else { + return Ok(None); + }; + let sidecar = |suffix| { + let mut name = path.as_os_str().to_os_string(); + name.push(suffix); + PathBuf::from(name) + }; + Ok(Some(Self { + main, + // WAL 头包含重用盐值；SHM 的两份 48 字节索引头包含提交帧和变更计数。 + // 只比较原始字节，不解析平台字节序；同时保留 30 秒期限作为兜底。 + wal: FileStamp::read(&sidecar("-wal"), 32)?, + shm: FileStamp::read(&sidecar("-shm"), 96)?, + })) + } +} + +impl LocalSessionPager { + /// 每个库在短读事务内校验索引并读取详情，确保同一库的总数和页面一致。 + /// 独立数据库之间没有全局事务；同一 ID 按最新时间取值，时间相同时保留输入路径顺序。 + pub fn list_page( + &mut self, + paths: &[PathBuf], + offset: usize, + limit: usize, + ) -> LocalSessionPage { + let mut page = LocalSessionPage::default(); + let mut changed = false; + if !self.databases.iter().map(|(path, _)| path).eq(paths) { + self.databases = paths + .iter() + .cloned() + .map(|path| (path, CachedDatabase::default())) + .collect(); + changed = true; + } + + // 每次请求只短暂打开连接。Windows 上即使只读连接也会阻止数据库重命名或删除。 + let mut connections = Vec::with_capacity(paths.len()); + for (path, cached) in &mut self.databases { + let result = (|| -> anyhow::Result<_> { + if !path.try_exists()? { + return Ok(None); + } + let connection = + Connection::open_with_flags(path.as_path(), OpenFlags::SQLITE_OPEN_READ_ONLY)?; + connection.busy_timeout(Duration::from_millis(250))?; + let stamp = DatabaseStamp::read(path)?; + Ok(Some((connection, stamp))) + })(); + match result { + Ok(Some(connection)) => connections.push(Some(connection)), + result => { + changed |= cached.stamp.take().is_some() || !cached.index.is_empty(); + cached.index.clear(); + if let Err(error) = result { + page.errors.push(format!("{}: {error}", path.display())); + } + connections.push(None); + } + } + } + + let mut snapshots = Vec::with_capacity(self.databases.len()); + let mut versions = Vec::with_capacity(self.databases.len()); + for ((path, cached), connection) in self.databases.iter_mut().zip(&connections) { + let Some((connection, before)) = connection else { + snapshots.push(None); + versions.push(None); + continue; + }; + let result = (|| -> anyhow::Result<_> { + let transaction = connection.unchecked_transaction()?; + // 先固定读快照；事务后的同连接 data_version 用来防止缓存绑定到较新的文件指纹。 + transaction.query_row("SELECT COUNT(*) FROM sqlite_schema", [], |_| Ok(()))?; + let version: i64 = + transaction.query_row("PRAGMA data_version", [], |row| row.get(0))?; + #[cfg(test)] + if let Some(after_snapshot) = self.after_snapshot.take() { + after_snapshot(); + } + let stamp = DatabaseStamp::read(path)?; + let reusable = stamp.is_some() + && &stamp == before + && stamp == cached.stamp + && cached + .refreshed_at + .is_some_and(|time| time.elapsed() < INDEX_MAX_AGE); + if !reusable { + changed = true; + let query = session_query(&transaction)?; + let mut statement = + transaction.prepare(&format!("SELECT id, updated_at_ms FROM ({query})"))?; + let index = statement + .query_map([], |row| { + Ok(IndexEntry { + id: row.get(0)?, + updated_at_ms: row.get(1)?, + }) + })? + .collect::>>()?; + cached.index = index; + cached.query = query; + cached.refreshed_at = Some(Instant::now()); + cached.stamp = stamp; + } + Ok((transaction, version)) + })(); + match result { + Ok((transaction, version)) => { + snapshots.push(Some((path, transaction, &cached.index, &cached.query))); + versions.push(Some(version)); + } + Err(error) => { + changed = true; + cached.stamp = None; + cached.index.clear(); + page.errors.push(format!("{}: {error}", path.display())); + snapshots.push(None); + versions.push(None); + } + } + } + + if changed { + self.order.clear(); + for (source, snapshot) in snapshots.iter().enumerate() { + if let Some((_, _, index, _)) = snapshot { + self.order + .extend((0..index.len()).map(|entry| (source, entry))); + } + } + let entry = |&(source, index): &(usize, usize)| { + &snapshots[source] + .as_ref() + .expect("索引仅来自成功的读快照") + .2[index] + }; + self.order.sort_by(|left, right| { + let (left, right) = (entry(left), entry(right)); + right + .updated_at_ms + .cmp(&left.updated_at_ms) + .then_with(|| right.id.cmp(&left.id)) + }); + let mut seen = HashSet::new(); + self.order + .retain(|item| seen.insert(entry(item).id.as_str())); + } + + page.total_count = self.order.len(); + page.has_more = offset.saturating_add(limit) < page.total_count; + let selected = self + .order + .get(offset..offset.saturating_add(limit).min(self.order.len())) + .unwrap_or_default(); + for (source, snapshot) in snapshots.iter().enumerate() { + let Some((path, transaction, index, query)) = snapshot else { + continue; + }; + let ids: Vec<_> = selected + .iter() + .filter(|(db, _)| *db == source) + .map(|(_, entry)| &index[*entry].id) + .collect(); + // 限制参数数量；这个数据层入口也可能被页大小更大的调用者使用。 + for chunk in ids.chunks(100) { + let result = (|| -> anyhow::Result> { + let placeholders = vec!["?"; chunk.len()].join(","); + // automation_runs 可能为同一会话保存多次运行，只解码最新的一条。 + let sql = format!( + "SELECT * FROM ( + SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at_ms DESC) AS session_rank + FROM ({query}) WHERE id IN ({placeholders}) + ) WHERE session_rank = 1 ORDER BY updated_at_ms DESC, id DESC" + ); + let mut statement = transaction.prepare(&sql)?; + let rows = statement + .query_map(params_from_iter(chunk), |row| read_session(row, path))?; + Ok(rows.collect::>>()?) + })(); + match result { + Ok(rows) => page.sessions.extend(rows), + Err(error) => page.errors.push(format!("{}: {error}", path.display())), + } + } + } + page.sessions.sort_by(|left, right| { + right + .updated_at_ms + .cmp(&left.updated_at_ms) + .then_with(|| right.id.cmp(&left.id)) + }); + drop(snapshots); + for (((path, cached), connection), version) in + self.databases.iter_mut().zip(&connections).zip(versions) + { + if let Some((connection, _)) = connection { + let after = connection + .query_row("PRAGMA data_version", [], |row| row.get::<_, i64>(0)) + .ok(); + if after != version || DatabaseStamp::read(path).ok().flatten() != cached.stamp { + // 页面仍属于刚才的读快照，但下次请求必须重新建索引。 + cached.stamp = None; + } + } + } + // 读事务和文件句柄都在返回前释放，不需要后台轮询清理闲置连接。 + drop(connections); + page + } +} + +pub(super) fn read_session(row: &Row<'_>, path: &Path) -> rusqlite::Result { + Ok(LocalSession { + id: row.get(0)?, + title: row.get::<_, Option>(1)?.unwrap_or_default(), + cwd: row.get::<_, Option>(2)?.unwrap_or_default(), + model_provider: row.get::<_, Option>(3)?.unwrap_or_default(), + archived: row.get::<_, Option>(4)?.unwrap_or_default() != 0, + updated_at_ms: row.get(5)?, + rollout_path: row.get::<_, Option>(6)?.unwrap_or_default(), + db_path: path.to_string_lossy().into_owned(), + }) +} + +// 列表、轻量索引和当前页详情共用字段映射及子会话过滤，避免三个入口逐渐偏离。 +pub(super) fn session_query(db: &Connection) -> anyhow::Result { + match schema_kind(db)? { + Some(SchemaKind::CodexThreads) => { + let columns = table_columns(db, "threads")? + .into_iter() + .collect::>(); + let column = |name, fallback| optional_column_expression(&columns, name, fallback); + let title = column("title", "''"); + let cwd = column("cwd", "''"); + let provider = column("model_provider", "''"); + let archived = column("archived", "0"); + let rollout = column("rollout_path", "''"); + let updated = 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 filter = codex_thread_filter(db)?; + Ok(format!( + "SELECT id, {title} AS title, {cwd} AS cwd, {provider} AS model_provider, {archived} AS archived, {updated} AS updated_at_ms, {rollout} AS rollout_path FROM threads {filter}" + )) + } + Some(SchemaKind::CodexAutomationRuns) => { + let columns = table_columns(db, "automation_runs")? + .into_iter() + .collect::>(); + let column = |name, fallback| optional_column_expression(&columns, name, fallback); + let title = column("thread_title", "''"); + let cwd = column("source_cwd", "''"); + let status = column("status", "''"); + let updated = column("updated_at", "NULL"); + let created = column("created_at", "NULL"); + Ok(format!( + "SELECT thread_id AS id, {title} AS title, {cwd} AS cwd, '' AS model_provider, LOWER({status}) = 'archived' AS archived, COALESCE({updated}, {created}) AS updated_at_ms, '' AS rollout_path FROM automation_runs WHERE COALESCE(thread_id, '') <> ''" + )) + } + _ => anyhow::bail!("Unsupported local storage schema"), + } +} diff --git a/crates/codex-plus-data/src/storage/session_paging/tests.rs b/crates/codex-plus-data/src/storage/session_paging/tests.rs new file mode 100644 index 000000000..470173924 --- /dev/null +++ b/crates/codex-plus-data/src/storage/session_paging/tests.rs @@ -0,0 +1,390 @@ +use super::*; +use rusqlite::params; +use tempfile::tempdir; + +fn threads(path: &Path) -> Connection { + let db = Connection::open(path).unwrap(); + db.execute_batch("CREATE TABLE threads (id TEXT PRIMARY KEY, title TEXT, rollout_path TEXT, updated_at_ms INTEGER)").unwrap(); + db +} + +fn insert(db: &Connection, id: &str, title: &str, updated: Option) { + db.execute( + "INSERT INTO threads VALUES (?1, ?2, '', ?3)", + params![id, title, updated], + ) + .unwrap(); +} + +fn ids(page: &LocalSessionPage) -> Vec<&str> { + assert!(page.errors.is_empty(), "{:?}", page.errors); + page.sessions + .iter() + .map(|session| session.id.as_str()) + .collect() +} + +#[test] +fn deep_pages_reuse_index_and_deduplicate_before_slicing() { + let temp = tempdir().unwrap(); + let paths = [ + temp.path().join("current.db"), + temp.path().join("legacy.db"), + ]; + let current = threads(&paths[0]); + let legacy = threads(&paths[1]); + current.execute_batch("BEGIN").unwrap(); + legacy.execute_batch("BEGIN").unwrap(); + for number in 0..1500 { + let id = format!("t{number:04}"); + insert(¤t, &id, "current", Some(number)); + if number % 2 == 0 { + insert(&legacy, &id, "legacy", Some(number + 10000)); + } + } + current.execute_batch("COMMIT").unwrap(); + legacy.execute_batch("COMMIT").unwrap(); + let mut pager = LocalSessionPager::default(); + let expected = pager.list_page(&paths, 0, 2000); + assert_eq!(expected.total_count, 1500); + assert_eq!(expected.sessions.len(), 1500); + assert!(!expected.has_more); + let index_address = pager.databases[0].1.index.as_ptr(); + for offset in [0, 49, 700, 1400, 1490, 1500, usize::MAX] { + let page = pager.list_page(&paths, offset, 50); + assert!(page.errors.is_empty()); + assert_eq!(page.total_count, 1500); + let wanted = expected + .sessions + .iter() + .skip(offset) + .take(50) + .cloned() + .collect::>(); + assert_eq!(page.sessions, wanted); + assert_eq!(page.has_more, offset.saturating_add(50) < 1500); + } + assert_eq!(pager.databases[0].1.index.as_ptr(), index_address); + assert!( + expected + .sessions + .iter() + .take(750) + .all(|row| row.title == "legacy") + ); +} + +#[test] +fn stable_order_handles_null_negative_times_and_source_ties() { + let temp = tempdir().unwrap(); + let paths = [temp.path().join("a.db"), temp.path().join("b.db")]; + let first = threads(&paths[0]); + let second = threads(&paths[1]); + for (id, time) in [ + ("a", Some(0)), + ("b", Some(0)), + ("negative", Some(-1)), + ("unknown", None), + ] { + insert(&first, id, "first", time); + insert(&second, id, "second", time); + } + let mut pager = LocalSessionPager::default(); + let page = pager.list_page(&paths, 0, 10); + assert_eq!(ids(&page), ["b", "a", "negative", "unknown"]); + assert!(page.sessions.iter().all(|session| session.title == "first")); + assert_eq!(ids(&pager.list_page(&paths, 2, 1)), ["negative"]); + let reversed = [paths[1].clone(), paths[0].clone()]; + assert!( + pager + .list_page(&reversed, 0, 10) + .sessions + .iter() + .all(|row| row.title == "second") + ); +} + +#[test] +fn wal_commits_refresh_counts_order_and_details_without_main_file_changes() { + let temp = tempdir().unwrap(); + let paths = [temp.path().join("state.db")]; + let writer = threads(&paths[0]); + writer + .execute_batch("PRAGMA journal_mode=WAL; PRAGMA wal_autocheckpoint=0;") + .unwrap(); + insert(&writer, "a", "old", Some(1)); + let mut pager = LocalSessionPager::default(); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["a"]); + let stamp = FileStamp::read(&paths[0], 100).unwrap().unwrap(); + insert(&writer, "b", "new", Some(2)); + assert!(stamp == FileStamp::read(&paths[0], 100).unwrap().unwrap()); + let page = pager.list_page(&paths, 0, 1); + assert_eq!(ids(&page), ["b"]); + assert_eq!(page.total_count, 2); + assert!(page.has_more); + writer + .execute( + "UPDATE threads SET title = 'edited', updated_at_ms = 3 WHERE id = 'a'", + [], + ) + .unwrap(); + let page = pager.list_page(&paths, 0, 1); + assert_eq!(ids(&page), ["a"]); + assert_eq!(page.sessions[0].title, "edited"); + writer + .execute_batch("BEGIN; DELETE FROM threads; ROLLBACK;") + .unwrap(); + assert_eq!(pager.list_page(&paths, 0, 10).total_count, 2); + writer + .execute("DELETE FROM threads WHERE id = 'a'", []) + .unwrap(); + let page = pager.list_page(&paths, 0, 10); + assert_eq!(ids(&page), ["b"]); + assert_eq!(page.total_count, 1); + writer + .execute_batch("PRAGMA wal_checkpoint(TRUNCATE);") + .unwrap(); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["b"]); +} + +#[test] +fn read_snapshot_version_does_not_skip_a_concurrent_wal_commit() { + let temp = tempdir().unwrap(); + let path = temp.path().join("state.db"); + let writer = threads(&path); + writer.execute_batch("PRAGMA journal_mode=WAL").unwrap(); + let reader = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_ONLY).unwrap(); + let before: i64 = reader + .query_row("PRAGMA data_version", [], |row| row.get(0)) + .unwrap(); + let snapshot = reader.unchecked_transaction().unwrap(); + snapshot + .query_row("SELECT COUNT(*) FROM sqlite_schema", [], |_| Ok(())) + .unwrap(); + insert(&writer, "a", "new", Some(1)); + let during: i64 = snapshot + .query_row("PRAGMA data_version", [], |row| row.get(0)) + .unwrap(); + assert_eq!(before, during); + let count: i64 = snapshot + .query_row("SELECT COUNT(*) FROM threads", [], |row| row.get(0)) + .unwrap(); + assert_eq!(count, 0); + drop(snapshot); + let after: i64 = reader + .query_row("PRAGMA data_version", [], |row| row.get(0)) + .unwrap(); + assert_ne!(before, after); +} + +#[test] +fn automation_duplicates_do_not_starve_later_pages() { + let temp = tempdir().unwrap(); + let paths = [temp.path().join("automation.db")]; + let writer = Connection::open(&paths[0]).unwrap(); + writer.execute_batch("CREATE TABLE automation_runs (thread_id TEXT, thread_title TEXT, source_cwd TEXT, status TEXT, updated_at INTEGER, created_at INTEGER)").unwrap(); + writer.execute_batch("BEGIN").unwrap(); + for number in 0..200 { + writer + .execute( + "INSERT INTO automation_runs VALUES ('same', ?1, '/cwd', 'ARCHIVED', ?2, 1)", + params![format!("run {number}"), number + 10], + ) + .unwrap(); + } + writer.execute_batch("INSERT INTO automation_runs VALUES ('old', 'old', '', '', NULL, 1), ('', 'ignored', '', '', 999, 999)").unwrap(); + writer.execute_batch("COMMIT").unwrap(); + let mut pager = LocalSessionPager::default(); + let first = pager.list_page(&paths, 0, 1); + assert_eq!(ids(&first), ["same"]); + assert_eq!(first.total_count, 2); + assert_eq!(first.sessions[0].title, "run 199"); + assert!(first.sessions[0].archived); + assert!(first.has_more); + let second = pager.list_page(&paths, 1, 1); + assert_eq!(ids(&second), ["old"]); + assert!(!second.has_more); +} + +#[test] +fn schema_changes_and_child_references_invalidate_the_index() { + let temp = tempdir().unwrap(); + let paths = [temp.path().join("state.db")]; + let writer = threads(&paths[0]); + writer.execute_batch("PRAGMA journal_mode=WAL").unwrap(); + insert(&writer, "parent", "parent", Some(1)); + insert(&writer, "child", "child", Some(2)); + let mut pager = LocalSessionPager::default(); + assert_eq!(pager.list_page(&paths, 0, 10).total_count, 2); + writer.execute_batch("CREATE TABLE thread_spawn_edges(child_thread_id TEXT); INSERT INTO thread_spawn_edges VALUES ('child');").unwrap(); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["parent"]); + writer.execute_batch("DELETE FROM thread_spawn_edges; CREATE TABLE agent_job_items(assigned_thread_id TEXT); INSERT INTO agent_job_items VALUES ('child');").unwrap(); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["parent"]); + writer.execute_batch("DROP TABLE agent_job_items").unwrap(); + assert_eq!(pager.list_page(&paths, 0, 10).total_count, 2); +} + +#[test] +fn missing_corrupt_and_locked_databases_keep_healthy_results_and_recover() { + let temp = tempdir().unwrap(); + let paths = [ + temp.path().join("good.db"), + temp.path().join("bad.db"), + temp.path().join("missing.db"), + ]; + let writer = threads(&paths[0]); + insert(&writer, "good", "good", Some(1)); + std::fs::write(&paths[1], b"not a database").unwrap(); + let mut pager = LocalSessionPager::default(); + let partial = pager.list_page(&paths, 0, 10); + assert_eq!(partial.sessions.len(), 1); + assert_eq!(partial.total_count, 1); + assert_eq!(partial.errors.len(), 1); + assert!(partial.errors[0].contains("bad.db")); + // 替换不支持的库后再次查询，不缓存错误或旧的总数。 + pager.list_page(&paths[..1], 0, 10); + std::fs::remove_file(&paths[1]).unwrap(); + let recovered = threads(&paths[1]); + insert(&recovered, "recovered", "recovered", Some(2)); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["recovered", "good"]); + let new = threads(&paths[2]); + insert(&new, "new", "new", Some(3)); + assert_eq!(pager.list_page(&paths, 0, 10).total_count, 3); + recovered.execute_batch("BEGIN EXCLUSIVE").unwrap(); + let partial = pager.list_page(&paths, 0, 10); + assert_eq!(partial.errors.len(), 1); + assert_eq!(partial.total_count, 2); + recovered.execute_batch("ROLLBACK").unwrap(); + assert_eq!(pager.list_page(&paths, 0, 10).total_count, 3); +} + +#[test] +fn detail_decoding_failure_keeps_count_and_other_database_results() { + let temp = tempdir().unwrap(); + let paths = [temp.path().join("good.db"), temp.path().join("bad.db")]; + let good = threads(&paths[0]); + let bad = threads(&paths[1]); + insert(&good, "good", "good", Some(1)); + bad.execute("INSERT INTO threads VALUES ('bad', x'1234', '', 2)", []) + .unwrap(); + let mut pager = LocalSessionPager::default(); + let partial = pager.list_page(&paths, 0, 10); + assert_eq!(partial.sessions.len(), 1); + assert_eq!(partial.total_count, 2); + assert_eq!(partial.errors.len(), 1); + bad.execute("UPDATE threads SET title = 'fixed'", []) + .unwrap(); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["bad", "good"]); +} + +#[test] +fn restoring_same_size_database_is_detected_even_when_timestamps_are_preserved() { + let temp = tempdir().unwrap(); + let paths = [temp.path().join("state.db")]; + let replacement = temp.path().join("restored.db"); + let original = threads(&paths[0]); + insert(&original, "old", "old", Some(1)); + drop(original); + let restored = threads(&replacement); + insert(&restored, "new", "new", Some(2)); + drop(restored); + let mut pager = LocalSessionPager::default(); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["old"]); + let stamp = FileStamp::read(&paths[0], 100).unwrap().unwrap(); + // 备份还原可能保留长度和时间戳，也可能恰好保留 SQLite 的变更计数器。 + std::fs::copy(&replacement, &paths[0]).unwrap(); + let file = std::fs::OpenOptions::new() + .write(true) + .open(&paths[0]) + .unwrap(); + file.set_times(std::fs::FileTimes::new().set_modified(stamp.modified.unwrap())) + .unwrap(); + drop(file); + assert!(stamp == FileStamp::read(&paths[0], 100).unwrap().unwrap()); + pager.databases[0].1.refreshed_at = Some(Instant::now() - INDEX_MAX_AGE); + let page = pager.list_page(&paths, 0, 10); + assert_eq!(ids(&page), ["new"]); + assert_eq!(page.total_count, 1); +} + +#[test] +fn query_releases_file_handles_so_database_can_be_renamed_deleted_and_recreated() { + let temp = tempdir().unwrap(); + let paths = [temp.path().join("state.db")]; + let writer = threads(&paths[0]); + insert(&writer, "old", "old", Some(1)); + drop(writer); + let mut pager = LocalSessionPager::default(); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["old"]); + let backup = temp.path().join("backup.db"); + std::fs::rename(&paths[0], &backup).unwrap(); + let missing = pager.list_page(&paths, 0, 10); + assert_eq!(missing.total_count, 0); + assert!(missing.errors.is_empty()); + let writer = threads(&paths[0]); + insert(&writer, "new", "new", Some(2)); + drop(writer); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["new"]); + std::fs::remove_file(&paths[0]).unwrap(); + std::fs::rename(&backup, &paths[0]).unwrap(); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["old"]); +} + +#[test] +fn commit_after_snapshot_does_not_cache_old_index_under_new_fingerprint() { + let temp = tempdir().unwrap(); + let paths = [temp.path().join("state.db")]; + let writer = threads(&paths[0]); + writer.execute_batch("PRAGMA journal_mode=WAL").unwrap(); + insert(&writer, "a", "old", Some(1)); + let mut pager = LocalSessionPager::default(); + assert_eq!(ids(&pager.list_page(&paths, 0, 10)), ["a"]); + // 保持主写连接存活，避免测试提交后关闭最后一个写连接而触发 checkpoint。 + let concurrent = Connection::open(&paths[0]).unwrap(); + pager.after_snapshot = Some(Box::new(move || insert(&concurrent, "b", "new", Some(2)))); + let snapshot = pager.list_page(&paths, 0, 10); + assert_eq!(ids(&snapshot), ["a"]); + assert_eq!(snapshot.total_count, 1); + assert!(pager.databases[0].1.stamp.is_none()); + let fresh = pager.list_page(&paths, 0, 10); + assert_eq!(ids(&fresh), ["b", "a"]); + assert_eq!(fresh.total_count, 2); +} + +#[test] +fn wal_reuse_with_unchanged_size_and_timestamps_is_detected_by_index_header() { + let temp = tempdir().unwrap(); + let paths = [temp.path().join("state.db")]; + let writer = threads(&paths[0]); + writer + .execute_batch("PRAGMA journal_mode=WAL; PRAGMA wal_autocheckpoint=0") + .unwrap(); + for number in 0..10 { + insert(&writer, &format!("t{number}"), "old", Some(number)); + } + writer + .execute_batch("PRAGMA wal_checkpoint(RESTART)") + .unwrap(); + insert(&writer, "a", "new", Some(11)); + let mut pager = LocalSessionPager::default(); + assert_eq!(pager.list_page(&paths, 0, 20).total_count, 11); + let before = DatabaseStamp::read(&paths[0]).unwrap().unwrap(); + insert(&writer, "b", "new", Some(12)); + for (suffix, stamp) in [ + ("-wal", before.wal.as_ref().unwrap()), + ("-shm", before.shm.as_ref().unwrap()), + ] { + let path = PathBuf::from(format!("{}{suffix}", paths[0].display())); + let file = std::fs::OpenOptions::new().write(true).open(path).unwrap(); + file.set_times(std::fs::FileTimes::new().set_modified(stamp.modified.unwrap())) + .unwrap(); + } + let after = DatabaseStamp::read(&paths[0]).unwrap().unwrap(); + assert!(before.main == after.main); + assert!(before.wal == after.wal); + assert!(before.shm != after.shm); + let fresh = pager.list_page(&paths, 0, 20); + assert!(fresh.errors.is_empty()); + assert_eq!(fresh.total_count, 12); + assert_eq!(fresh.sessions[0].id, "b"); +} diff --git a/docs/session-paging.md b/docs/session-paging.md new file mode 100644 index 000000000..8eb75385e --- /dev/null +++ b/docs/session-paging.md @@ -0,0 +1,19 @@ +# 会话分页 + +管理器的 `list_local_sessions` 在后台线程读取本地会话，接口继续返回 +`offset`、`limit`、`hasMore` 和 `totalCount`。 + +`LocalSessionPager` 缓存会话 ID、更新时间和来源，翻页只读取本页详情。 +跨库同 ID 保留较新的记录，时间相同时保留数据库输入顺序。子会话过滤与 +`SQLiteStorageAdapter` 共用查询，不保存常驻数据库连接。 + +缓存检查数据库及 WAL/SHM 的文件信息和头部，最多复用 30 秒。每个数据库 +使用短读事务；多个数据库之间没有共同快照，因此外部写入期间翻页结果可能变化。 +单库读取失败时返回错误和健康库的结果，不把旧缓存当成成功结果。 + +回归覆盖深分页、跨库去重、WAL 提交、数据库替换、缓存期限、部分失败及文件句柄释放： + +```sh +cargo test -p codex-plus-data storage::session_paging +cargo test -p codex-plus-manager --lib list_local_sessions +```