feat: audit incomplete historical session repairs

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,
                     ));

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

@@ -2900,10 +2900,19 @@ fn provider_sync_command_result(
 ) -> CommandResult<Value> {
     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(),
         }
     }
 

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;
 };
 

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,

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<String>,
+    #[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<ProviderSyncResult> {
         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<PathBuf> {
     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<PathBuf> {
     paths
 }
 
+fn audit_provider_sync_state(
+    home: &Path,
+    sqlite_paths: &[PathBuf],
+) -> anyhow::Result<ProviderSyncAudit> {
+    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::<HashSet<_>>();
+    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::<HashSet<_>>();
+    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<HashSet<String>> {
+    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<PathBuf>) -> 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<HashSet<String>> {
     Ok(ids)
 }
 
+fn sqlite_table_ids(path: &Path, table: &str, column: &str) -> anyhow::Result<HashSet<String>> {
+    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::<rusqlite::Result<HashSet<_>>>()?)
+}
+
+fn sqlite_user_thread_ids(path: &Path) -> anyhow::Result<HashSet<String>> {
+    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<String>>(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<String>,