diff --git a/apps/codex-plus-launcher/src/main.rs b/apps/codex-plus-launcher/src/main.rs index 59a3b8181..8704be7d7 100644 --- a/apps/codex-plus-launcher/src/main.rs +++ b/apps/codex-plus-launcher/src/main.rs @@ -416,6 +416,7 @@ impl LaunchHooks for LauncherHooks { updated_workspace_roots: 0, skipped_locked_rollout_files: Vec::new(), encrypted_content_warning: None, + repair_audit: codex_plus_data::ProviderSyncAudit::default(), }, completion_error, )); @@ -446,6 +447,7 @@ impl LaunchHooks for LauncherHooks { updated_workspace_roots: 0, skipped_locked_rollout_files: Vec::new(), encrypted_content_warning: None, + repair_audit: codex_plus_data::ProviderSyncAudit::default(), }, None, )); diff --git a/apps/codex-plus-manager/src-tauri/src/commands.rs b/apps/codex-plus-manager/src-tauri/src/commands.rs index 8e7e1b9f5..2d922b996 100644 --- a/apps/codex-plus-manager/src-tauri/src/commands.rs +++ b/apps/codex-plus-manager/src-tauri/src/commands.rs @@ -2900,10 +2900,19 @@ fn provider_sync_command_result( ) -> CommandResult { let succeeded = is_success_sync_status(&sync.status); let success_message = format!( - "供应商已同步一次:{} 个会话文件,{} 行索引,跳过 {} 个占用文件。", + "供应商已同步一次:{} 个会话文件,{} 行索引,跳过 {} 个占用文件。{}", sync.changed_session_files, sync.sqlite_rows_updated, - sync.skipped_locked_rollout_files.len() + sync.skipped_locked_rollout_files.len(), + if sync.repair_audit.catalog_only_sessions > 0 { + format!( + " 审计发现 {} 条仅存在于会话目录的记录,其中 {} 条没有可用恢复来源。", + sync.repair_audit.catalog_only_sessions, + sync.repair_audit.catalog_only_without_recovery_source, + ) + } else { + String::new() + } ); let failure_message = format!("历史会话修复未执行:{}", sync.message); let payload = json!({ @@ -2919,6 +2928,7 @@ fn provider_sync_command_result( "sqliteCatalogRowsRemoved": sync.sqlite_catalog_rows_removed, "updatedWorkspaceRoots": sync.updated_workspace_roots, "encryptedContentWarning": sync.encrypted_content_warning, + "repairAudit": sync.repair_audit, "backupDir": sync.backup_dir, "syncMessage": sync.message, }); @@ -5423,6 +5433,7 @@ mod tests { sqlite_catalog_rows_removed: 0, updated_workspace_roots: 0, encrypted_content_warning: None, + repair_audit: codex_plus_data::ProviderSyncAudit::default(), } } diff --git a/apps/codex-plus-manager/src/App.tsx b/apps/codex-plus-manager/src/App.tsx index 518b5f8ba..a5e278bb1 100644 --- a/apps/codex-plus-manager/src/App.tsx +++ b/apps/codex-plus-manager/src/App.tsx @@ -629,6 +629,12 @@ type ProviderSyncPayload = { updatedWorkspaceRoots?: number; prunedSessionIndexEntries?: number; encryptedContentWarning?: string | null; + repairAudit?: { + catalogOnlySessions: number; + catalogOnlyWithCurrentRollout: number; + catalogOnlyWithBackupDatabase: number; + catalogOnlyWithoutRecoverySource: number; + }; backupDir?: string | null; }; diff --git a/crates/codex-plus-data/src/lib.rs b/crates/codex-plus-data/src/lib.rs index 1f168ddb0..6bdc73186 100644 --- a/crates/codex-plus-data/src/lib.rs +++ b/crates/codex-plus-data/src/lib.rs @@ -6,7 +6,7 @@ pub mod storage; pub use backup::BackupStore; pub use markdown::{MarkdownExportService, export_markdown_from_paths}; pub use provider_sync::{ - ProviderSyncResult, ProviderSyncStatus, ProviderSyncTargetList, ProviderSyncTargetOption, + ProviderSyncAudit, ProviderSyncResult, ProviderSyncStatus, ProviderSyncTargetList, ProviderSyncTargetOption, ProviderSyncTargetSource, SessionIndexCleanupApplyError, SessionIndexCleanupCandidate, SessionIndexCleanupPreview, SessionIndexCleanupResult, apply_session_index_cleanup, load_provider_sync_targets, preview_session_index_cleanup, diff --git a/crates/codex-plus-data/src/provider_sync.rs b/crates/codex-plus-data/src/provider_sync.rs index e764a01a3..088bfd82f 100644 --- a/crates/codex-plus-data/src/provider_sync.rs +++ b/crates/codex-plus-data/src/provider_sync.rs @@ -49,6 +49,17 @@ pub struct ProviderSyncResult { pub sqlite_catalog_rows_removed: usize, pub updated_workspace_roots: usize, pub encrypted_content_warning: Option, + #[serde(default)] + pub repair_audit: ProviderSyncAudit, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ProviderSyncAudit { + pub catalog_only_sessions: usize, + pub catalog_only_with_current_rollout: usize, + pub catalog_only_with_backup_database: usize, + pub catalog_only_without_recovery_source: usize, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -614,6 +625,21 @@ pub fn run_provider_sync_with_target( let sync_result = (|| -> anyhow::Result { let sqlite_paths = provider_sync_db_paths(&home); let thread_kinds = sqlite_provider_sync_thread_kinds(&sqlite_paths)?; + let repair_audit = match audit_provider_sync_state(&home, &sqlite_paths) { + Ok(audit) => audit, + Err(error) => { + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "provider_sync.repair_audit_failed", + json!({ + "error": error.to_string(), + "backup_root": home + .join("backups_state/provider-sync") + .to_string_lossy(), + }), + ); + ProviderSyncAudit::default() + } + }; let collected = collect_session_changes( &home, &target_provider, @@ -670,6 +696,9 @@ pub fn run_provider_sync_with_target( ); synced.skipped_locked_rollout_files = collected.skipped_locked_rollout_files; synced.encrypted_content_warning = encrypted_content_warning; + synced.repair_audit = repair_audit; + synced.message = + provider_sync_message_with_audit(&synced.message, &synced.repair_audit); return Ok(synced); } let backup_dir = create_backup(&home, &target_provider, &rewrite_changes)?; @@ -720,6 +749,8 @@ pub fn run_provider_sync_with_target( synced.sqlite_catalog_rows_removed = sqlite_updates.catalog_remove_rows; synced.updated_workspace_roots = updated_workspace_roots; synced.encrypted_content_warning = encrypted_content_warning; + synced.repair_audit = repair_audit; + synced.message = provider_sync_message_with_audit(&synced.message, &synced.repair_audit); Ok(synced) })(); let _ = release_lock(&lock_dir); @@ -758,9 +789,23 @@ fn result( sqlite_catalog_rows_removed: 0, updated_workspace_roots: 0, encrypted_content_warning: None, + repair_audit: ProviderSyncAudit::default(), } } +fn provider_sync_message_with_audit(message: &str, audit: &ProviderSyncAudit) -> String { + if audit.catalog_only_sessions == 0 { + return message.to_string(); + } + format!( + "{message};审计发现 {} 条仅存在于本地会话目录的记录,其中 {} 条仍有当前 rollout、{} 条只能在历史数据库备份中找到,{} 条没有可用恢复来源;未自动重建缺失的 canonical 会话。", + audit.catalog_only_sessions, + audit.catalog_only_with_current_rollout, + audit.catalog_only_with_backup_database, + audit.catalog_only_without_recovery_source, + ) +} + fn provider_sync_db_paths(home: &Path) -> Vec { let mut paths = codex_plus_core::codex_sqlite::codex_session_db_paths_from_home(home); for path in codex_plus_core::codex_sqlite::codex_thread_reference_db_paths_from_home(home) { @@ -771,6 +816,98 @@ fn provider_sync_db_paths(home: &Path) -> Vec { paths } +fn audit_provider_sync_state( + home: &Path, + sqlite_paths: &[PathBuf], +) -> anyhow::Result { + let mut canonical_thread_ids = HashSet::new(); + let mut catalog_thread_ids = HashSet::new(); + for path in sqlite_paths { + canonical_thread_ids.extend(sqlite_table_ids(path, "threads", "id")?); + catalog_thread_ids.extend(sqlite_user_thread_ids(path)?); + } + + let catalog_only = catalog_thread_ids + .difference(&canonical_thread_ids) + .cloned() + .collect::>(); + if catalog_only.is_empty() { + return Ok(ProviderSyncAudit::default()); + } + + let current_rollout_ids = rollout_files(home)? + .into_iter() + .filter_map(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .and_then(rollout_thread_id_from_filename) + }) + .collect::>(); + let backup_database_ids = backup_database_thread_ids(home)?; + let with_current_rollout = catalog_only + .iter() + .filter(|thread_id| current_rollout_ids.contains(*thread_id)) + .count(); + let with_backup_database = catalog_only + .iter() + .filter(|thread_id| { + !current_rollout_ids.contains(*thread_id) && backup_database_ids.contains(*thread_id) + }) + .count(); + + Ok(ProviderSyncAudit { + catalog_only_sessions: catalog_only.len(), + catalog_only_with_current_rollout: with_current_rollout, + catalog_only_with_backup_database: with_backup_database, + catalog_only_without_recovery_source: catalog_only + .iter() + .filter(|thread_id| { + !current_rollout_ids.contains(*thread_id) + && !backup_database_ids.contains(*thread_id) + }) + .count(), + }) +} + +fn backup_database_thread_ids(home: &Path) -> anyhow::Result> { + let root = home.join("backups_state/provider-sync"); + let mut ids = HashSet::new(); + if !root.exists() { + return Ok(ids); + } + let mut files = Vec::new(); + collect_files_recursive(&root, &mut files)?; + for path in files { + if !matches!( + path.extension().and_then(|value| value.to_str()), + Some("sqlite" | "db") + ) { + continue; + } + if let Ok(thread_ids) = sqlite_table_ids(&path, "threads", "id") { + ids.extend(thread_ids); + } + } + Ok(ids) +} + +fn collect_files_recursive(root: &Path, files: &mut Vec) -> anyhow::Result<()> { + for entry in fs::read_dir(root)? { + let entry = entry?; + let file_type = entry.file_type()?; + if file_type.is_symlink() { + continue; + } + let path = entry.path(); + if file_type.is_dir() { + collect_files_recursive(&path, files)?; + } else if file_type.is_file() { + files.push(path); + } + } + Ok(()) +} + pub fn load_provider_sync_targets(codex_home: Option<&Path>) -> ProviderSyncTargetList { let home = codex_home .map(Path::to_path_buf) @@ -1513,6 +1650,53 @@ fn sqlite_thread_ids(path: &Path) -> anyhow::Result> { Ok(ids) } +fn sqlite_table_ids(path: &Path, table: &str, column: &str) -> anyhow::Result> { + if !path.exists() { + return Ok(HashSet::new()); + } + let db = Connection::open(path)?; + if !table_columns(&db, table)?.contains(column) { + return Ok(HashSet::new()); + } + let sql = format!("SELECT DISTINCT {column} FROM {table} WHERE COALESCE({column}, '') <> ''"); + Ok(db + .prepare(&sql)? + .query_map([], |row| row.get::<_, String>(0))? + .collect::>>()?) +} + +fn sqlite_user_thread_ids(path: &Path) -> anyhow::Result> { + if !path.exists() { + return Ok(HashSet::new()); + } + let db = Connection::open(path)?; + let columns = table_columns(&db, "local_thread_catalog")?; + if !columns.contains("thread_id") { + return Ok(HashSet::new()); + } + let source_kind = text_expr(&columns, "source_kind", "''"); + let thread_source = text_expr(&columns, "thread_source", "NULL"); + let sql = format!( + "SELECT thread_id, {source_kind}, {thread_source} FROM local_thread_catalog WHERE COALESCE(thread_id, '') <> ''" + ); + let mut ids = HashSet::new(); + for row in db.prepare(&sql)?.query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1).unwrap_or_default(), + row.get::<_, Option>(2).unwrap_or(None), + )) + })? { + let (thread_id, source_kind, thread_source) = row?; + if !thread_source_marks_non_root(thread_source.as_deref()) + && !source_marks_non_root_agent(&source_kind) + { + ids.insert(thread_id); + } + } + Ok(ids) +} + fn plan_session_index_cleanup( path: &Path, live_thread_ids: &HashSet, diff --git a/crates/codex-plus-data/tests/provider_sync.rs b/crates/codex-plus-data/tests/provider_sync.rs index 0329daade..7302e1b46 100644 --- a/crates/codex-plus-data/tests/provider_sync.rs +++ b/crates/codex-plus-data/tests/provider_sync.rs @@ -865,6 +865,119 @@ fn provider_sync_repairs_missing_local_thread_catalog_rows_from_threads() { assert_eq!(sync_state.2, 1); } +#[test] +fn provider_sync_audits_catalog_only_sessions_without_claiming_recovery() { + let tmp = tempdir().unwrap(); + let home = tmp.path().join(".codex"); + let sqlite_dir = home.join("sqlite"); + fs::create_dir_all(&sqlite_dir).unwrap(); + fs::write(home.join("config.toml"), "model_provider = \"apigather\"\n").unwrap(); + + let state_db = home.join("state_5.sqlite"); + create_state_db_with_providers(&state_db, &[("canonical", "openai", 0)]); + let current_rollout_id = "01a01579-4a5d-77e3-89c0-751d38ad21f8"; + write_rollout( + &home.join("sessions/2026/08/18").join(format!( + "rollout-2026-08-18T23-24-25-{current_rollout_id}.jsonl" + )), + "custom", + current_rollout_id, + "C:/workspace", + ); + + let catalog_db = sqlite_dir.join("codex-dev.db"); + create_local_thread_catalog_db( + &catalog_db, + &[ + (current_rollout_id, "custom"), + ("backup-only", "custom"), + ("no-source", "custom"), + ("agent-only", "custom"), + ], + ); + Connection::open(&catalog_db) + .unwrap() + .execute( + "UPDATE local_thread_catalog SET thread_source = 'subagent' WHERE thread_id = 'agent-only'", + [], + ) + .unwrap(); + + let backup_db = home.join("backups_state/provider-sync/20260818233010/db/state_5.sqlite"); + fs::create_dir_all(backup_db.parent().unwrap()).unwrap(); + create_state_db_with_providers(&backup_db, &[("backup-only", "custom", 0)]); + fs::write( + home.join("backups_state/provider-sync/20260818233010/db/broken.sqlite"), + b"not a sqlite database", + ) + .unwrap(); + + let result = run_provider_sync(Some(&home)); + + assert_eq!(result.status, ProviderSyncStatus::Synced); + assert_eq!(result.repair_audit.catalog_only_sessions, 3); + assert_eq!(result.repair_audit.catalog_only_with_current_rollout, 1); + assert_eq!(result.repair_audit.catalog_only_with_backup_database, 1); + assert_eq!(result.repair_audit.catalog_only_without_recovery_source, 1); + assert!(result.message.contains("未自动重建缺失的 canonical 会话")); +} + +#[test] +fn provider_sync_continues_when_repair_audit_backup_root_is_not_directory() { + let tmp = tempdir().unwrap(); + let home = tmp.path().join(".codex"); + let sqlite_dir = home.join("sqlite"); + fs::create_dir_all(&sqlite_dir).unwrap(); + fs::write(home.join("config.toml"), "model_provider = \"apigather\"\n").unwrap(); + create_state_db_with_providers(&home.join("state_5.sqlite"), &[]); + create_local_thread_catalog_db( + &sqlite_dir.join("codex-dev.db"), + &[("catalog-only", "apigather")], + ); + + let backup_root = home.join("backups_state/provider-sync"); + fs::create_dir_all(backup_root.parent().unwrap()).unwrap(); + fs::write(&backup_root, b"unreadable backup root").unwrap(); + + let result = run_provider_sync(Some(&home)); + + assert_eq!( + result.status, + ProviderSyncStatus::Synced, + "{}", + result.message + ); + assert_eq!(result.sqlite_rows_updated, 0); + assert_eq!(result.repair_audit.catalog_only_sessions, 0); +} + +#[cfg(unix)] +#[test] +fn provider_sync_repair_audit_skips_cyclic_backup_symlinks() { + use std::os::unix::fs::symlink; + + let tmp = tempdir().unwrap(); + let home = tmp.path().join(".codex"); + let sqlite_dir = home.join("sqlite"); + fs::create_dir_all(&sqlite_dir).unwrap(); + fs::write(home.join("config.toml"), "model_provider = \"apigather\"\n").unwrap(); + create_state_db_with_providers(&home.join("state_5.sqlite"), &[]); + create_local_thread_catalog_db( + &sqlite_dir.join("codex-dev.db"), + &[("catalog-only", "apigather")], + ); + + let backup_root = home.join("backups_state/provider-sync"); + let cycle = backup_root.join("cycle"); + fs::create_dir_all(&backup_root).unwrap(); + symlink(&backup_root, &cycle).unwrap(); + + let result = run_provider_sync(Some(&home)); + + assert_eq!(result.status, ProviderSyncStatus::Synced); + assert_eq!(result.sqlite_rows_updated, 0); +} + #[test] fn provider_sync_catalogs_user_threads_but_skips_subagents() { let tmp = tempdir().unwrap();