diff --git a/apps/codex-plus-manager/src-tauri/src/commands.rs b/apps/codex-plus-manager/src-tauri/src/commands.rs index c70022a21..d98c7bb06 100644 --- a/apps/codex-plus-manager/src-tauri/src/commands.rs +++ b/apps/codex-plus-manager/src-tauri/src/commands.rs @@ -1916,6 +1916,13 @@ pub fn delete_local_session(request: DeleteLocalSessionRequest) -> CommandResult candidate_paths.push(path); } } + for path in codex_plus_core::codex_sqlite::codex_thread_reference_db_paths_from_home( + &codex_plus_core::codex_sqlite::default_codex_home_dir(), + ) { + if !candidate_paths.iter().any(|candidate| candidate == &path) { + candidate_paths.push(path); + } + } log_manager_event( "manager.delete_local_session.start", json!({ @@ -2314,24 +2321,24 @@ fn merge_manual_provider_sync_targets( #[tauri::command] pub async fn preview_session_index_cleanup() -> CommandResult { - let result = tauri::async_runtime::spawn_blocking(|| { - codex_plus_data::preview_session_index_cleanup(None) - }) - .await - .map_err(|error| anyhow::anyhow!("session index cleanup preview task failed: {error}")) - .and_then(|result| result); + let result = + tauri::async_runtime::spawn_blocking(|| codex_plus_data::preview_historical_cleanup(None)) + .await + .map_err(|error| anyhow::anyhow!("session index cleanup preview task failed: {error}")) + .and_then(|result| result); match result { Ok(preview) => ok( &format!( - "发现 {} 条仅存在于任务索引中的候选记录。", + "发现 {} 条已失去正文来源的历史残留记录。", preview.candidates.len() ), json!({ "snapshotSha256": preview.snapshot_sha256, + "catalogRevision": preview.catalog_revision, "candidates": preview.candidates, }), ), - Err(error) => failed(&format!("预览失效任务索引失败：{error}"), json!({})), + Err(error) => failed(&format!("预览历史已删除会话残留失败：{error}"), json!({})), } } @@ -2341,18 +2348,27 @@ pub async fn apply_session_index_cleanup( thread_ids: Vec, ) -> CommandResult { let result = tauri::async_runtime::spawn_blocking(move || { - codex_plus_data::apply_session_index_cleanup(None, &snapshot_sha256, &thread_ids) + codex_plus_data::apply_historical_cleanup(None, &snapshot_sha256, &thread_ids) }) .await .map_err(|error| anyhow::anyhow!("session index cleanup task failed: {error}")); match result { Ok(Ok(cleanup)) => ok( &format!( - "已清理 {} 条失效任务索引；原索引已完整备份。", - cleanup.pruned_entries + "历史残留已清理：目录 {}、时间线 {}、任务索引 {}、全局状态 {}；已创建可撤销备份。", + cleanup.catalog_rows, + cleanup.timeline_rows, + cleanup.session_index_entries, + cleanup.global_state_references + cleanup.global_state_backup_references, ), json!({ - "prunedEntries": cleanup.pruned_entries, + "prunedEntries": cleanup.session_index_entries, + "catalogRows": cleanup.catalog_rows, + "timelineRows": cleanup.timeline_rows, + "sessionIndexEntries": cleanup.session_index_entries, + "globalStateReferences": cleanup.global_state_references, + "globalStateBackupReferences": cleanup.global_state_backup_references, + "skipped": cleanup.skipped, "backupDir": cleanup.backup_dir, }), ), @@ -2363,11 +2379,46 @@ pub async fn apply_session_index_cleanup( .map(|path| format!(" 备份目录：{}。", path.to_string_lossy())) .unwrap_or_default(); failed( - &format!("清理失效任务索引失败：{}{backup_hint}", error.message), - json!({ "backupDir": error.backup_dir }), + &format!("清理历史已删除会话残留失败：{}{backup_hint}", error.message), + json!({ + "backupDir": error.backup_dir, + "catalogRows": error.partial_result.catalog_rows, + "timelineRows": error.partial_result.timeline_rows, + "sessionIndexEntries": error.partial_result.session_index_entries, + "globalStateReferences": error.partial_result.global_state_references, + "globalStateBackupReferences": error.partial_result.global_state_backup_references, + "skipped": error.partial_result.skipped, + "failureReason": error.message, + }), ) } - Err(error) => failed(&format!("清理失效任务索引失败：{error}"), json!({})), + Err(error) => failed(&format!("清理历史已删除会话残留失败：{error}"), json!({})), + } +} + +#[tauri::command] +pub async fn undo_session_index_cleanup(backup_dir: String) -> CommandResult { + let result = tauri::async_runtime::spawn_blocking(move || { + codex_plus_data::undo_historical_cleanup(None, std::path::Path::new(&backup_dir)) + }) + .await + .map_err(|error| anyhow::anyhow!("historical cleanup undo task failed: {error}")); + match result { + Ok(Ok(restored)) => ok( + "已从备份恢复历史会话目录、时间线、任务索引和全局状态引用。", + json!({ + "catalogRows": restored.catalog_rows, + "timelineRows": restored.timeline_rows, + "sessionIndexEntries": restored.session_index_entries, + "globalStateReferences": restored.global_state_references, + "globalStateBackupReferences": restored.global_state_backup_references, + }), + ), + Ok(Err(error)) => failed( + &format!("撤销历史残留清理失败：{}", error.message), + json!({ "backupDir": error.backup_dir }), + ), + Err(error) => failed(&format!("撤销历史残留清理失败：{error}"), json!({})), } } @@ -5554,6 +5605,96 @@ mod tests { assert_eq!(thread_count(&legacy_db, "t1"), 0); } + #[test] + fn delete_local_session_removes_thread_rollout_and_catalog_reference() { + 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 codex_home = temp.path().join("codex-home"); + let sqlite_dir = codex_home.join("sqlite"); + std::fs::create_dir_all(&sqlite_dir).unwrap(); + let thread_db = sqlite_dir.join("state_5.sqlite"); + let catalog_db = sqlite_dir.join("codex-dev.db"); + let rollout_path = codex_home.join("rollout.jsonl"); + std::fs::write(&rollout_path, "{\"type\":\"message\"}\n").unwrap(); + create_minimal_thread_db(&thread_db, "t1", "Current Thread", 100); + rusqlite::Connection::open(&thread_db) + .unwrap() + .execute( + "UPDATE threads SET rollout_path = ?1 WHERE id = 't1'", + [rollout_path.to_string_lossy().to_string()], + ) + .unwrap(); + let catalog = rusqlite::Connection::open(&catalog_db).unwrap(); + catalog + .execute("CREATE TABLE automation_runs (thread_id TEXT NOT NULL)", []) + .unwrap(); + catalog + .execute( + "CREATE TABLE local_thread_catalog (host_id TEXT NOT NULL, thread_id TEXT NOT NULL, display_title TEXT NOT NULL, PRIMARY KEY (host_id, thread_id))", + [], + ) + .unwrap(); + catalog + .execute( + "INSERT INTO local_thread_catalog VALUES ('local', 't1', 'Current Thread')", + [], + ) + .unwrap(); + catalog + .execute( + "CREATE TABLE local_thread_catalog_metadata (id INTEGER PRIMARY KEY, catalog_revision INTEGER NOT NULL DEFAULT 0)", + [], + ) + .unwrap(); + catalog + .execute( + "INSERT INTO local_thread_catalog_metadata VALUES (1, 12)", + [], + ) + .unwrap(); + drop(catalog); + + unsafe { + std::env::set_var("CODEX_HOME", &codex_home); + } + let result = delete_local_session(DeleteLocalSessionRequest { + session_id: "t1".to_string(), + title: "Current Thread".to_string(), + db_path: Some(thread_db.to_string_lossy().to_string()), + }); + restore_codex_home(previous_codex_home); + + assert_eq!(result.status, "ok"); + assert_eq!( + result.payload.status, + codex_plus_core::models::DeleteStatus::LocalDeleted + ); + assert_eq!(thread_count(&thread_db, "t1"), 0); + assert!(!rollout_path.exists()); + let catalog = rusqlite::Connection::open(&catalog_db).unwrap(); + assert_eq!( + catalog + .query_row( + "SELECT COUNT(*) FROM local_thread_catalog WHERE thread_id = 't1'", + [], + |row| row.get::<_, i64>(0), + ) + .unwrap(), + 0 + ); + assert_eq!( + catalog + .query_row( + "SELECT catalog_revision FROM local_thread_catalog_metadata WHERE id = 1", + [], + |row| row.get::<_, i64>(0), + ) + .unwrap(), + 13 + ); + } + fn create_minimal_thread_db(path: &Path, id: &str, title: &str, updated_at_ms: i64) { let db = rusqlite::Connection::open(path).unwrap(); db.execute( diff --git a/apps/codex-plus-manager/src-tauri/src/lib.rs b/apps/codex-plus-manager/src-tauri/src/lib.rs index da987661a..bcdd87e72 100644 --- a/apps/codex-plus-manager/src-tauri/src/lib.rs +++ b/apps/codex-plus-manager/src-tauri/src/lib.rs @@ -51,6 +51,7 @@ pub fn run() { let mut main_window_builder = tauri::WebviewWindowBuilder::new(app, "main", tauri::WebviewUrl::App(url.into())) .title("Codex++ 管理工具") + .center() .inner_size(1180.0, 820.0) .min_inner_size(960.0, 720.0); if let Some(icon) = app.default_window_icon().cloned() { @@ -104,6 +105,7 @@ pub fn run() { commands::load_provider_sync_targets, commands::preview_session_index_cleanup, commands::apply_session_index_cleanup, + commands::undo_session_index_cleanup, commands::sync_providers_now, commands::load_ads, commands::refresh_script_market, diff --git a/apps/codex-plus-manager/src-tauri/tests/windows_subsystem.rs b/apps/codex-plus-manager/src-tauri/tests/windows_subsystem.rs index 3cda81083..00490e78d 100644 --- a/apps/codex-plus-manager/src-tauri/tests/windows_subsystem.rs +++ b/apps/codex-plus-manager/src-tauri/tests/windows_subsystem.rs @@ -69,6 +69,14 @@ fn manager_close_minimizes_to_tray_without_confirmation() { assert!(app_tsx.contains("manager_hide_to_tray")); } +#[test] +fn manager_centers_window_on_startup() { + let lib_rs = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/lib.rs")) + .expect("read manager lib.rs"); + + assert!(lib_rs.contains(".center()")); +} + #[test] fn manager_queues_codexplusplus_provider_urls_for_confirmation_on_startup() { let main_rs = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/main.rs")) diff --git a/apps/codex-plus-manager/src/App.tsx b/apps/codex-plus-manager/src/App.tsx index 2a7aadfaa..71721c54b 100644 --- a/apps/codex-plus-manager/src/App.tsx +++ b/apps/codex-plus-manager/src/App.tsx @@ -593,15 +593,24 @@ type SessionIndexCleanupCandidate = { id: string; threadName: string; updatedAt: string; + workspace: string; + sources: string[]; }; type SessionIndexCleanupPreviewPayload = { snapshotSha256: string; + catalogRevision: number; candidates: SessionIndexCleanupCandidate[]; }; type SessionIndexCleanupApplyPayload = { prunedEntries?: number; + catalogRows?: number; + timelineRows?: number; + sessionIndexEntries?: number; + globalStateReferences?: number; + globalStateBackupReferences?: number; + skipped?: number; backupDir?: string | null; }; @@ -898,6 +907,7 @@ export function App() { candidates: SessionIndexCleanupCandidate[]; resolve: (selectedIds: string[] | null) => void; } | null>(null); + const [historicalCleanupBackupDir, setHistoricalCleanupBackupDir] = useState(null); const [overview, setOverview] = useState(null); const [settings, setSettings] = useState(null); const [relay, setRelay] = useState(null); @@ -2108,27 +2118,34 @@ export function App() { if (!preview) { cleanupFailure = { status: "failed", - message: t("幽灵任务索引处理失败，请查看错误提示后重试。"), + message: t("历史已删除会话残留扫描失败，请查看错误提示后重试。"), }; } else if (isSuccessStatus(preview.status) && preview.candidates.length > 0) { const selectedIds = await selectSessionIndexCleanupCandidates(preview.candidates); if (selectedIds?.length) { - const cleanup = await run(() => - call>("apply_session_index_cleanup", { - snapshotSha256: preview.snapshotSha256, - threadIds: selectedIds, - }), + const confirmed = await confirmSessionDelete( + t("确认清理历史残留"), + tf("将清理所选 {0} 个已失去正文来源的会话索引，并创建可撤销备份。执行前必须完全退出 Codex App / ChatGPT。是否继续？", [selectedIds.length]), ); - if (cleanup && isSuccessStatus(cleanup.status)) { - finalResult = { - ...result, - prunedSessionIndexEntries: cleanup.prunedEntries ?? 0, - }; - } else { - cleanupFailure = cleanup ?? { - status: "failed", - message: t("幽灵任务索引处理失败，请查看错误提示后重试。"), - }; + if (confirmed) { + const cleanup = await run(() => + call>("apply_session_index_cleanup", { + snapshotSha256: preview.snapshotSha256, + threadIds: selectedIds, + }), + ); + if (cleanup && isSuccessStatus(cleanup.status)) { + setHistoricalCleanupBackupDir(cleanup.backupDir ?? null); + finalResult = { + ...result, + prunedSessionIndexEntries: cleanup.prunedEntries ?? 0, + }; + } else { + cleanupFailure = cleanup ?? { + status: "failed", + message: t("历史已删除会话残留清理失败，请查看错误提示后重试。"), + }; + } } } } else if (!isSuccessStatus(preview.status)) { @@ -2158,7 +2175,7 @@ export function App() { } await refreshProviderSyncTargets(true); const noticeTitle = - completion.noticeKind === "cleanup" ? t("清理幽灵任务索引") : t("历史会话修复"); + completion.noticeKind === "cleanup" ? t("清理历史已删除会话残留") : t("历史会话修复"); showNotice( noticeTitle, completion.result.message, @@ -2177,6 +2194,27 @@ export function App() { } }; + const undoHistoricalCleanup = async () => { + if (!historicalCleanupBackupDir) return; + const confirmed = await confirmSessionDelete( + t("撤销历史残留清理"), + t("将从最近一次备份恢复目录、时间线、任务索引和全局状态引用。若已出现同 ID 的新会话，系统会拒绝覆盖。是否继续？"), + ); + if (!confirmed) return; + const result = await run(() => + call>("undo_session_index_cleanup", { + backupDir: historicalCleanupBackupDir, + }), + ); + if (result) { + showResultNotice(t("撤销历史残留清理"), result); + if (isSuccessStatus(result.status)) { + setHistoricalCleanupBackupDir(null); + await refreshLocalSessions(true); + } + } + }; + const applyRelayInjection = async (silent = false) => { const settingsResult = await run(() => call("save_settings", { settings: settingsForm })); if (settingsResult) { @@ -2736,6 +2774,8 @@ export function App() { } }, syncProvidersNow, + undoHistoricalCleanup, + canUndoHistoricalCleanup: Boolean(historicalCleanupBackupDir), refreshProviderSyncTargets, setProviderSyncTarget: (provider: string) => { setSelectedProviderSyncTarget(provider); @@ -2800,7 +2840,7 @@ export function App() { disableWatcher: () => watcherAction("disable_watcher"), toggleTheme: () => setTheme((current) => (current === "dark" ? "light" : "dark")), }), - [route, launchForm, settingsForm, settings, overview, removeOwnedData, update, updateInstallProgress.active, logs, diagnostics, theme, relayFiles, localSessions, zedRemoteProjects, selectedProviderSyncTarget, envConflicts, relayEnvironment, ccsProviders, dreamSkinLibrary, dreamSkinMarket, dreamSkinCommunity, selectedDreamSkinTheme, savedDreamSkinThemeDraft, dreamSkinThemeDraft, dreamSkinDraftDirty, pendingDreamSkinRestart], + [route, launchForm, settingsForm, settings, overview, removeOwnedData, update, updateInstallProgress.active, logs, diagnostics, theme, relayFiles, localSessions, zedRemoteProjects, selectedProviderSyncTarget, envConflicts, relayEnvironment, ccsProviders, dreamSkinLibrary, dreamSkinMarket, dreamSkinCommunity, selectedDreamSkinTheme, savedDreamSkinThemeDraft, dreamSkinThemeDraft, dreamSkinDraftDirty, pendingDreamSkinRestart, historicalCleanupBackupDir], ); const hasUpdate = update?.updateAvailable === true; @@ -3097,6 +3137,8 @@ type Actions = { saveDreamSkinScreenshot: () => Promise; saveManualCodexAppPath: () => Promise; syncProvidersNow: () => Promise; + undoHistoricalCleanup: () => Promise; + canUndoHistoricalCleanup: boolean; refreshProviderSyncTargets: (silent?: boolean) => Promise; setProviderSyncTarget: (provider: string) => void; setLaunchMode: (launchMode: LaunchMode) => Promise; @@ -5034,6 +5076,12 @@ function SessionsScreen({ {providerSyncProgress.active ? t("正在修复…") : t("立刻修复历史会话")} + {actions.canUndoHistoricalCleanup ? ( + + ) : null}
@@ -7379,9 +7427,9 @@ function SessionIndexCleanupDialog({
-

{t("清理幽灵任务索引")}

+

{t("历史已删除会话清理")}

- {tf("发现 {0} 条仅存在于 session_index.jsonl、未在本地数据库或 rollout 中找到来源的候选记录。它们也可能是云端或尚未落盘的任务，请逐项核对。任务标题仅用于预览，实际按 thread ID 与数据来源判断。清理前请先完全退出 Codex App / ChatGPT。", [request.candidates.length])} + {tf("发现 {0} 条已失去 threads、自动化记录和 rollout 正文来源的本地残留索引。请逐项核对；标题仅用于预览，清理严格按 thread ID 与残留来源执行。这不是普通会话删除。清理前请完全退出 Codex App / ChatGPT。", [request.candidates.length])}

@@ -7407,7 +7455,9 @@ function SessionIndexCleanupDialog({ {candidate.threadName || t("未命名任务")} {candidate.id} - {candidate.updatedAt} + {candidate.updatedAt || t("未知更新时间")} + {candidate.workspace ? {tf("项目/工作区：{0}", [candidate.workspace])} : null} + {tf("残留来源：{0}", [candidate.sources.join("、")])} ))} diff --git a/crates/codex-plus-data/src/historical_cleanup.rs b/crates/codex-plus-data/src/historical_cleanup.rs new file mode 100644 index 000000000..62e5a034c --- /dev/null +++ b/crates/codex-plus-data/src/historical_cleanup.rs @@ -0,0 +1,1141 @@ +use base64::Engine; +use rusqlite::types::{ToSqlOutput, Value as SqlValue, ValueRef}; +use rusqlite::{Connection, ToSql, params_from_iter}; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value, json}; +use sha2::{Digest, Sha256}; +use std::collections::{BTreeMap, BTreeSet, HashSet}; +use std::fs; +use std::path::{Path, PathBuf}; + +const FILE_NAMES: [&str; 3] = [ + "session_index.jsonl", + ".codex-global-state.json", + ".codex-global-state.json.bak", +]; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct HistoricalCleanupCandidate { + pub id: String, + pub thread_name: String, + pub updated_at: String, + pub workspace: String, + pub sources: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct HistoricalCleanupPreview { + pub snapshot_sha256: String, + pub catalog_revision: i64, + pub candidates: Vec, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct HistoricalCleanupResult { + pub catalog_rows: usize, + pub timeline_rows: usize, + pub session_index_entries: usize, + pub global_state_references: usize, + pub global_state_backup_references: usize, + pub skipped: usize, + pub backup_dir: Option, +} + +#[derive(Debug, thiserror::Error)] +#[error("{message}")] +pub struct HistoricalCleanupError { + pub message: String, + pub backup_dir: Option, + pub partial_result: HistoricalCleanupResult, +} + +pub type HistoricalCleanupApplyResult = + Result>; + +#[derive(Clone)] +struct FileSnapshot { + name: String, + path: PathBuf, + existed: bool, + bytes: Vec, +} + +#[derive(Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct DbRows { + path: String, + catalog_rows: Vec>, + timeline_rows: Vec>, +} + +struct CleanupPlan { + snapshot_sha256: String, + catalog_revision: i64, + candidates: Vec, + files: Vec, + db_rows: Vec, +} + +#[derive(Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct BackupFileEntry { + name: String, + existed: bool, + original_sha256: String, + post_cleanup_sha256: String, +} + +#[derive(Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct CleanupManifest { + version: u32, + namespace: String, + codex_home: String, + created_at: String, + snapshot_sha256: String, + selected_ids: Vec, + selected_candidates: Vec, + files: Vec, + databases: Vec, + deleted_counts: HistoricalCleanupResult, +} + +#[derive(Clone)] +struct OwnedSqlValue(SqlValue); + +impl ToSql for OwnedSqlValue { + fn to_sql(&self) -> rusqlite::Result> { + Ok(ToSqlOutput::Owned(self.0.clone())) + } +} + +pub fn preview_historical_cleanup(home: Option<&Path>) -> anyhow::Result { + let home = home + .map(Path::to_path_buf) + .unwrap_or_else(codex_plus_core::codex_home::default_codex_home_dir); + let plan = build_plan(&home)?; + Ok(HistoricalCleanupPreview { + snapshot_sha256: plan.snapshot_sha256, + catalog_revision: plan.catalog_revision, + candidates: plan.candidates, + }) +} + +pub fn apply_historical_cleanup( + home: Option<&Path>, + expected_snapshot_sha256: &str, + selected_ids: &[String], +) -> HistoricalCleanupApplyResult { + let require_stopped_app = home.is_none(); + ensure_codex_stopped(require_stopped_app, None)?; + let home = home + .map(Path::to_path_buf) + .unwrap_or_else(codex_plus_core::codex_home::default_codex_home_dir); + let lock = acquire_cleanup_lock(&home)?; + let result = apply_locked( + &home, + expected_snapshot_sha256, + selected_ids, + require_stopped_app, + ); + let _ = fs::remove_dir(&lock); + result +} + +fn apply_locked( + home: &Path, + expected_snapshot_sha256: &str, + selected_ids: &[String], + require_stopped_app: bool, +) -> HistoricalCleanupApplyResult { + let plan = build_plan(home).map_err(|error| cleanup_error(error, None))?; + if plan.snapshot_sha256 != expected_snapshot_sha256 { + return Err(cleanup_error( + "SQLite、任务索引或全局状态已在预览后发生变化；本次清理已中止，请重新预览", + None, + )); + } + let candidate_ids = plan + .candidates + .iter() + .map(|candidate| candidate.id.as_str()) + .collect::>(); + let selected = selected_ids + .iter() + .map(|id| id.trim()) + .filter(|id| !id.is_empty()) + .map(ToString::to_string) + .collect::>(); + if selected.is_empty() { + return Ok(HistoricalCleanupResult::default()); + } + if selected + .iter() + .any(|id| !candidate_ids.contains(id.as_str())) + { + return Err(cleanup_error( + "确认列表已过期或包含非候选会话；本次清理未执行，请重新预览", + None, + )); + } + ensure_codex_stopped(require_stopped_app, None)?; + + let selected_set = selected.iter().cloned().collect::>(); + let (next_files, file_counts) = + cleaned_files(&plan.files, &selected_set).map_err(|error| cleanup_error(error, None))?; + let selected_db_rows = selected_database_rows(&plan.db_rows, &selected_set); + let mut result = HistoricalCleanupResult { + session_index_entries: file_counts[0], + global_state_references: file_counts[1], + global_state_backup_references: file_counts[2], + ..HistoricalCleanupResult::default() + }; + result.catalog_rows = selected_db_rows + .iter() + .map(|rows| rows.catalog_rows.len()) + .sum(); + result.timeline_rows = selected_db_rows + .iter() + .map(|rows| rows.timeline_rows.len()) + .sum(); + let backup_dir = create_backup( + home, + &plan, + &selected, + &next_files, + &selected_db_rows, + &result, + )?; + result.backup_dir = Some(backup_dir.clone()); + let mut completed = HistoricalCleanupResult { + backup_dir: Some(backup_dir.clone()), + ..HistoricalCleanupResult::default() + }; + + let current = + build_plan(home).map_err(|error| cleanup_error(error, Some(backup_dir.clone())))?; + if current.snapshot_sha256 != plan.snapshot_sha256 { + return Err(cleanup_error( + "数据在备份过程中发生变化；未继续写入，请重新预览", + Some(backup_dir), + )); + } + + for (snapshot, next) in plan.files.iter().zip(next_files.iter()) { + if snapshot.bytes == *next { + continue; + } + codex_plus_core::settings::atomic_write(&snapshot.path, next).map_err(|error| { + cleanup_error_with_progress(error, Some(backup_dir.clone()), completed.clone()) + })?; + match snapshot.name.as_str() { + "session_index.jsonl" => completed.session_index_entries = file_counts[0], + ".codex-global-state.json" => completed.global_state_references = file_counts[1], + ".codex-global-state.json.bak" => { + completed.global_state_backup_references = file_counts[2]; + } + _ => {} + } + } + for rows in &selected_db_rows { + delete_database_rows(rows, &selected_set).map_err(|error| { + cleanup_error_with_progress( + format!("数据库 {} 清理失败：{error}", rows.path), + Some(backup_dir.clone()), + completed.clone(), + ) + })?; + completed.catalog_rows += rows.catalog_rows.len(); + completed.timeline_rows += rows.timeline_rows.len(); + } + Ok(result) +} + +pub fn undo_historical_cleanup( + home: Option<&Path>, + backup_dir: &Path, +) -> HistoricalCleanupApplyResult { + let require_stopped_app = home.is_none(); + ensure_codex_stopped(require_stopped_app, Some(backup_dir.to_path_buf()))?; + let home = home + .map(Path::to_path_buf) + .unwrap_or_else(codex_plus_core::codex_home::default_codex_home_dir); + let backup_dir = validate_backup_dir(&home, backup_dir)?; + let manifest: CleanupManifest = serde_json::from_slice( + &fs::read(backup_dir.join("manifest.json")) + .map_err(|error| cleanup_error(error, Some(backup_dir.clone())))?, + ) + .map_err(|error| cleanup_error(error, Some(backup_dir.clone())))?; + let selected = manifest + .selected_ids + .iter() + .cloned() + .collect::>(); + let live = collect_real_thread_ids(&home, &database_paths(&home)) + .map_err(|error| cleanup_error(error, Some(backup_dir.clone())))?; + if selected.iter().any(|id| live.contains(id)) { + return Err(cleanup_error( + "检测到同 ID 的新会话或 rollout，撤销已拒绝，未覆盖新内容", + Some(backup_dir), + )); + } + preflight_restore_databases(&manifest.databases, &selected) + .map_err(|error| cleanup_error(error, Some(backup_dir.clone())))?; + for file in &manifest.files { + let path = home.join(&file.name); + let current = fs::read(&path).unwrap_or_default(); + if sha256_hex(¤t) != file.post_cleanup_sha256 { + return Err(cleanup_error( + format!("{} 已在清理后发生变化，撤销已拒绝", file.name), + Some(backup_dir), + )); + } + } + for rows in &manifest.databases { + restore_database_rows(rows) + .map_err(|error| cleanup_error(error, Some(backup_dir.clone())))?; + } + for file in &manifest.files { + let path = home.join(&file.name); + if file.existed { + let bytes = fs::read(backup_dir.join(&file.name)) + .map_err(|error| cleanup_error(error, Some(backup_dir.clone())))?; + codex_plus_core::settings::atomic_write(&path, &bytes) + .map_err(|error| cleanup_error(error, Some(backup_dir.clone())))?; + } else if path.exists() { + fs::remove_file(&path) + .map_err(|error| cleanup_error(error, Some(backup_dir.clone())))?; + } + } + Ok(manifest.deleted_counts) +} + +fn build_plan(home: &Path) -> anyhow::Result { + let paths = database_paths(home); + let mut live_ids = collect_real_thread_ids(home, &paths)?; + let mut db_rows = Vec::new(); + let mut candidates = BTreeMap::::new(); + let mut catalog_revision = 0_i64; + for path in &paths { + if !path.exists() { + continue; + } + let db = Connection::open(path)?; + let host_id = local_host_id(&db)?; + live_ids.extend(remote_catalog_thread_ids(&db, host_id.as_deref())?); + let catalog_rows = select_thread_rows(&db, "local_thread_catalog", host_id.as_deref())?; + let timeline_rows = select_thread_rows(&db, "thread_timeline_ledger", host_id.as_deref())?; + catalog_revision += catalog_revision_value(&db)?; + for row in &catalog_rows { + let id = row_string(row, "thread_id"); + let source_detail = row_string(row, "source_detail"); + if !source_detail.is_empty() && Path::new(&source_detail).is_file() { + live_ids.insert(id.clone()); + } + merge_candidate(&mut candidates, row, "catalog"); + } + for row in &timeline_rows { + merge_candidate(&mut candidates, row, "timeline"); + } + db_rows.push(DbRows { + path: path.to_string_lossy().to_string(), + catalog_rows, + timeline_rows, + }); + } + let files = FILE_NAMES + .iter() + .map(|name| file_snapshot(home, name)) + .collect::>>()?; + add_file_sources(&files, &mut candidates)?; + candidates.retain(|id, _| !live_ids.contains(id)); + for candidate in candidates.values_mut() { + candidate.sources.sort(); + candidate.sources.dedup(); + } + let candidates = candidates.into_values().collect::>(); + let fingerprint = json!({ + "liveIds": live_ids.into_iter().collect::>(), + "databases": db_rows, + "files": files.iter().map(|file| json!({"name": file.name, "existed": file.existed, "sha256": sha256_hex(&file.bytes)})).collect::>(), + "catalogRevision": catalog_revision, + "candidates": candidates, + }); + Ok(CleanupPlan { + snapshot_sha256: sha256_hex(&serde_json::to_vec(&fingerprint)?), + catalog_revision, + candidates, + files, + db_rows, + }) +} + +fn database_paths(home: &Path) -> Vec { + codex_plus_core::codex_sqlite::codex_thread_reference_db_paths_from_home(home) +} + +fn collect_real_thread_ids(home: &Path, paths: &[PathBuf]) -> anyhow::Result> { + let mut ids = HashSet::new(); + for root_name in ["sessions", "archived_sessions"] { + collect_rollout_ids(&home.join(root_name), &mut ids)?; + } + for path in paths { + if !path.exists() { + continue; + } + let db = Connection::open(path)?; + for (table, column) in [ + ("threads", "id"), + ("automation_runs", "thread_id"), + ("inbox_items", "thread_id"), + ("sessions", "id"), + ("messages", "session_id"), + ("thread_dynamic_tools", "thread_id"), + ("thread_goals", "thread_id"), + ("stage1_outputs", "thread_id"), + ("agent_job_items", "assigned_thread_id"), + ] { + if !table_columns(&db, table)?.contains(column) { + continue; + } + let sql = + format!("SELECT DISTINCT {column} FROM {table} WHERE COALESCE({column}, '') <> ''"); + let mut stmt = db.prepare(&sql)?; + for id in stmt.query_map([], |row| row.get::<_, String>(0))? { + ids.insert(id?); + } + } + } + Ok(ids) +} + +fn collect_rollout_ids(root: &Path, ids: &mut HashSet) -> anyhow::Result<()> { + if !root.exists() { + return Ok(()); + } + for entry in fs::read_dir(root)? { + let path = entry?.path(); + if path.is_dir() { + collect_rollout_ids(&path, ids)?; + continue; + } + if rollout_id_from_name(&path).is_none() { + continue; + } + let text = match fs::read_to_string(&path) { + Ok(text) => text, + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::PermissionDenied | std::io::ErrorKind::WouldBlock + ) => + { + continue; + } + Err(error) => return Err(error.into()), + }; + for line in text.lines() { + let Ok(value) = serde_json::from_str::(line) else { + continue; + }; + if value.get("type").and_then(Value::as_str) == Some("session_meta") + && let Some(id) = value.pointer("/payload/id").and_then(Value::as_str) + { + ids.insert(id.to_string()); + } + } + if let Some(id) = rollout_id_from_name(&path) { + ids.insert(id); + } + } + Ok(()) +} + +fn rollout_id_from_name(path: &Path) -> Option { + let name = path.file_name()?.to_str()?; + let stem = name.strip_prefix("rollout-")?.strip_suffix(".jsonl")?; + (stem.len() >= 36).then(|| stem[stem.len() - 36..].to_string()) +} + +fn file_snapshot(home: &Path, name: &str) -> anyhow::Result { + let path = home.join(name); + let existed = path.exists(); + let bytes = if existed { + fs::read(&path)? + } else { + Vec::new() + }; + Ok(FileSnapshot { + name: name.to_string(), + path, + existed, + bytes, + }) +} + +fn add_file_sources( + files: &[FileSnapshot], + candidates: &mut BTreeMap, +) -> anyhow::Result<()> { + for file in files { + if !file.existed || file.name != "session_index.jsonl" { + continue; + } + let text = std::str::from_utf8(&file.bytes)?; + for line in text.lines() { + let Ok(value) = serde_json::from_str::(line) else { + continue; + }; + let Some(object) = value.as_object() else { + continue; + }; + if object.len() != 3 + || !["id", "thread_name", "updated_at"] + .iter() + .all(|key| object.contains_key(*key)) + { + continue; + } + let Some(id) = value + .get("id") + .and_then(Value::as_str) + .filter(|id| !id.trim().is_empty()) + else { + continue; + }; + let candidate = + candidates + .entry(id.to_string()) + .or_insert_with(|| HistoricalCleanupCandidate { + id: id.to_string(), + thread_name: String::new(), + updated_at: String::new(), + workspace: String::new(), + sources: Vec::new(), + }); + candidate.sources.push("session_index".to_string()); + fill_candidate_from_index(candidate, &value); + } + } + let known = candidates.keys().cloned().collect::>(); + for file in files { + if !file.existed || file.name == "session_index.jsonl" { + continue; + } + let value: Value = serde_json::from_slice(&file.bytes)?; + for id in &known { + if has_structural_reference(&value, id) + && let Some(candidate) = candidates.get_mut(id) + { + candidate.sources.push(if file.name.ends_with(".bak") { + "global_state_bak".to_string() + } else { + "global_state".to_string() + }); + } + } + } + Ok(()) +} + +fn merge_candidate( + candidates: &mut BTreeMap, + row: &Map, + source: &str, +) { + let id = row_string(row, "thread_id"); + if id.is_empty() { + return; + } + let candidate = candidates + .entry(id.clone()) + .or_insert_with(|| HistoricalCleanupCandidate { + id, + thread_name: String::new(), + updated_at: String::new(), + workspace: String::new(), + sources: Vec::new(), + }); + candidate.sources.push(source.to_string()); + for key in ["display_title", "title"] { + if candidate.thread_name.is_empty() { + candidate.thread_name = row_string(row, key); + } + } + for key in ["source_updated_at", "updated_at", "updated_at_ms"] { + if candidate.updated_at.is_empty() { + candidate.updated_at = row.get(key).map(value_text).unwrap_or_default(); + } + } + if candidate.workspace.is_empty() { + candidate.workspace = row_string(row, "cwd"); + } +} + +fn fill_candidate_from_index(candidate: &mut HistoricalCleanupCandidate, value: &Value) { + if candidate.thread_name.is_empty() { + candidate.thread_name = value + .get("thread_name") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + } + if candidate.updated_at.is_empty() { + candidate.updated_at = value + .get("updated_at") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + } +} + +fn cleaned_files( + files: &[FileSnapshot], + selected: &HashSet, +) -> anyhow::Result<(Vec>, [usize; 3])> { + let mut next = Vec::new(); + let mut counts = [0; 3]; + for (index, file) in files.iter().enumerate() { + if !file.existed { + next.push(Vec::new()); + } else if file.name == "session_index.jsonl" { + let text = std::str::from_utf8(&file.bytes)?; + let mut output = String::new(); + for segment in text.split_inclusive('\n') { + let (line, ending) = split_line_ending(segment); + let remove = serde_json::from_str::(line) + .ok() + .and_then(|value| { + value + .get("id") + .and_then(Value::as_str) + .map(ToString::to_string) + }) + .is_some_and(|id| selected.contains(&id)); + if remove { + counts[index] += 1; + } else { + output.push_str(line); + output.push_str(ending); + } + } + next.push(output.into_bytes()); + } else { + let mut value: Value = serde_json::from_slice(&file.bytes)?; + counts[index] = prune_structural_references(&mut value, selected); + next.push(serde_json::to_vec_pretty(&value)?); + } + } + Ok((next, counts)) +} + +fn has_structural_reference(value: &Value, id: &str) -> bool { + match value { + Value::Object(map) => map.iter().any(|(key, value)| { + key.contains(id) || value.as_str() == Some(id) || has_structural_reference(value, id) + }), + Value::Array(items) => items + .iter() + .any(|value| value.as_str() == Some(id) || has_structural_reference(value, id)), + _ => false, + } +} + +fn prune_structural_references(value: &mut Value, selected: &HashSet) -> usize { + match value { + Value::Object(map) => { + let remove = map + .iter() + .filter(|(key, value)| { + selected.iter().any(|id| key.contains(id)) + || value.as_str().is_some_and(|text| selected.contains(text)) + }) + .map(|(key, _)| key.clone()) + .collect::>(); + let mut count = remove.len(); + for key in remove { + map.remove(&key); + } + count += map + .values_mut() + .map(|value| prune_structural_references(value, selected)) + .sum::(); + count + } + Value::Array(items) => { + let before = items.len(); + items.retain(|value| !value.as_str().is_some_and(|text| selected.contains(text))); + before - items.len() + + items + .iter_mut() + .map(|value| prune_structural_references(value, selected)) + .sum::() + } + _ => 0, + } +} + +fn selected_database_rows(rows: &[DbRows], selected: &HashSet) -> Vec { + rows.iter() + .filter_map(|rows| { + let catalog_rows = rows + .catalog_rows + .iter() + .filter(|row| selected.contains(&row_string(row, "thread_id"))) + .cloned() + .collect::>(); + let timeline_rows = rows + .timeline_rows + .iter() + .filter(|row| selected.contains(&row_string(row, "thread_id"))) + .cloned() + .collect::>(); + (!catalog_rows.is_empty() || !timeline_rows.is_empty()).then(|| DbRows { + path: rows.path.clone(), + catalog_rows, + timeline_rows, + }) + }) + .collect() +} + +fn create_backup( + home: &Path, + plan: &CleanupPlan, + selected: &BTreeSet, + next_files: &[Vec], + databases: &[DbRows], + counts: &HistoricalCleanupResult, +) -> Result> { + let root = home.join("backups_state/history-cleanup"); + let mut dir = root.join(chrono::Utc::now().format("%Y%m%d-%H%M%S%.3f").to_string()); + let mut suffix = 0; + while dir.exists() { + suffix += 1; + dir = root.join(format!( + "{}-{suffix}", + chrono::Utc::now().format("%Y%m%d-%H%M%S%.3f") + )); + } + fs::create_dir_all(&dir).map_err(|error| cleanup_error(error, None))?; + let mut files = Vec::new(); + for (snapshot, next) in plan.files.iter().zip(next_files.iter()) { + if snapshot.existed { + fs::write(dir.join(&snapshot.name), &snapshot.bytes) + .map_err(|error| cleanup_error(error, Some(dir.clone())))?; + } + files.push(BackupFileEntry { + name: snapshot.name.clone(), + existed: snapshot.existed, + original_sha256: sha256_hex(&snapshot.bytes), + post_cleanup_sha256: sha256_hex(next), + }); + } + let manifest = CleanupManifest { + version: 1, + namespace: "codex-plus-history-cleanup".to_string(), + codex_home: home.to_string_lossy().to_string(), + created_at: chrono::Utc::now().to_rfc3339(), + snapshot_sha256: plan.snapshot_sha256.clone(), + selected_ids: selected.iter().cloned().collect(), + selected_candidates: plan + .candidates + .iter() + .filter(|candidate| selected.contains(&candidate.id)) + .cloned() + .collect(), + files, + databases: databases.to_vec(), + deleted_counts: counts.clone(), + }; + let manifest_bytes = serde_json::to_vec_pretty(&manifest) + .map_err(|error| cleanup_error(error, Some(dir.clone())))?; + fs::write(dir.join("manifest.json"), manifest_bytes) + .map_err(|error| cleanup_error(error, Some(dir.clone())))?; + Ok(dir) +} + +fn delete_database_rows(rows: &DbRows, selected: &HashSet) -> anyhow::Result<()> { + let mut db = Connection::open(&rows.path)?; + let tx = db.transaction()?; + let host = local_host_id(&tx)?; + let mut catalog_deleted = 0; + for table in ["local_thread_catalog", "thread_timeline_ledger"] { + if !table_columns(&tx, table)?.contains("thread_id") { + continue; + } + for id in selected { + let count = if table_columns(&tx, table)?.contains("host_id") { + tx.execute( + &format!("DELETE FROM {table} WHERE host_id = ?1 AND thread_id = ?2"), + (host.as_deref().unwrap_or("local"), id), + )? + } else { + tx.execute(&format!("DELETE FROM {table} WHERE thread_id = ?1"), [id])? + }; + if table == "local_thread_catalog" { + catalog_deleted += count; + } + } + } + increment_catalog_revision(&tx, catalog_deleted)?; + tx.commit()?; + Ok(()) +} + +fn preflight_restore_databases(rows: &[DbRows], selected: &HashSet) -> anyhow::Result<()> { + for rows in rows { + let db = Connection::open(&rows.path)?; + for table in ["local_thread_catalog", "thread_timeline_ledger"] { + if !table_columns(&db, table)?.contains("thread_id") { + continue; + } + for id in selected { + let exists: i64 = db.query_row( + &format!("SELECT COUNT(*) FROM {table} WHERE thread_id = ?1"), + [id], + |row| row.get(0), + )?; + if exists > 0 { + anyhow::bail!("restore conflict: {table} already contains {id}"); + } + } + } + let source_ids = real_ids_in_db(&db)?; + if selected.iter().any(|id| source_ids.contains(id)) { + anyhow::bail!("restore conflict: a real session with the same ID exists"); + } + } + Ok(()) +} + +fn restore_database_rows(rows: &DbRows) -> anyhow::Result<()> { + let mut db = Connection::open(&rows.path)?; + let tx = db.transaction()?; + for row in &rows.catalog_rows { + insert_row(&tx, "local_thread_catalog", row)?; + } + for row in &rows.timeline_rows { + insert_row(&tx, "thread_timeline_ledger", row)?; + } + increment_catalog_revision(&tx, rows.catalog_rows.len())?; + tx.commit()?; + Ok(()) +} + +fn validate_backup_dir(home: &Path, dir: &Path) -> Result> { + let root = fs::canonicalize(home.join("backups_state/history-cleanup")) + .map_err(|error| cleanup_error(error, None))?; + let dir = fs::canonicalize(dir).map_err(|error| cleanup_error(error, None))?; + if !dir.starts_with(&root) { + return Err(cleanup_error("备份目录不属于历史残留清理", None)); + } + Ok(dir) +} + +fn acquire_cleanup_lock(home: &Path) -> Result> { + let lock = home.join("tmp/history-cleanup.lock"); + if let Some(parent) = lock.parent() { + fs::create_dir_all(parent).map_err(|error| cleanup_error(error, None))?; + } + fs::create_dir(&lock).map_err(|_| cleanup_error("已有历史残留清理正在运行", None))?; + Ok(lock) +} + +fn ensure_codex_stopped( + required: bool, + backup: Option, +) -> Result<(), Box> { + if !required { + return Ok(()); + } + let pids = codex_plus_core::watcher::find_session_index_cleanup_blocking_processes(); + if pids.is_empty() { + return Ok(()); + } + Err(cleanup_error( + format!( + "Codex App / ChatGPT 仍在运行（进程：{}）；请完全退出后重新预览", + pids.iter() + .map(u32::to_string) + .collect::>() + .join(", ") + ), + backup, + )) +} + +fn cleanup_error( + message: impl std::fmt::Display, + backup_dir: Option, +) -> Box { + Box::new(HistoricalCleanupError { + message: message.to_string(), + backup_dir, + partial_result: HistoricalCleanupResult::default(), + }) +} + +fn cleanup_error_with_progress( + message: impl std::fmt::Display, + backup_dir: Option, + partial_result: HistoricalCleanupResult, +) -> Box { + Box::new(HistoricalCleanupError { + message: message.to_string(), + backup_dir, + partial_result, + }) +} + +fn select_thread_rows( + db: &Connection, + table: &str, + host: Option<&str>, +) -> anyhow::Result>> { + let columns = table_columns(db, table)?; + if !columns.contains("thread_id") { + return Ok(Vec::new()); + } + let (sql, args): (String, Vec) = if columns.contains("host_id") { + ( + format!( + "SELECT * FROM {table} WHERE host_id = ?1 AND COALESCE(thread_id, '') <> '' ORDER BY thread_id" + ), + vec![OwnedSqlValue(SqlValue::Text( + host.unwrap_or("local").to_string(), + ))], + ) + } else { + ( + format!("SELECT * FROM {table} WHERE COALESCE(thread_id, '') <> '' ORDER BY thread_id"), + Vec::new(), + ) + }; + select_rows(db, &sql, &args) +} + +fn remote_catalog_thread_ids( + db: &Connection, + local_host: Option<&str>, +) -> anyhow::Result> { + let columns = table_columns(db, "local_thread_catalog")?; + if !columns.contains("thread_id") || !columns.contains("host_id") { + return Ok(HashSet::new()); + } + let mut stmt = db.prepare( + "SELECT DISTINCT thread_id FROM local_thread_catalog + WHERE host_id <> ?1 AND COALESCE(thread_id, '') <> ''", + )?; + Ok(stmt + .query_map([local_host.unwrap_or("local")], |row| { + row.get::<_, String>(0) + })? + .collect::>>()?) +} + +fn select_rows( + db: &Connection, + sql: &str, + args: &[OwnedSqlValue], +) -> anyhow::Result>> { + let mut stmt = db.prepare(sql)?; + let columns = stmt + .column_names() + .iter() + .map(|name| name.to_string()) + .collect::>(); + let refs = args + .iter() + .map(|value| value as &dyn ToSql) + .collect::>(); + Ok(stmt + .query_map(refs.as_slice(), |row| { + let mut map = Map::new(); + for (index, column) in columns.iter().enumerate() { + map.insert(column.clone(), sql_to_json(row.get_ref(index)?)); + } + Ok(map) + })? + .collect::>>()?) +} + +fn table_columns(db: &Connection, table: &str) -> anyhow::Result> { + let mut stmt = db.prepare(&format!( + "PRAGMA table_info(\"{}\")", + table.replace('"', "\"\"") + ))?; + Ok(stmt + .query_map([], |row| row.get::<_, String>(1))? + .collect::>>()?) +} + +fn local_host_id(db: &Connection) -> anyhow::Result> { + let columns = table_columns(db, "local_thread_catalog_hosts")?; + if !columns.contains("host_id") { + return Ok(Some("local".to_string())); + } + let sql = if columns.contains("host_kind") { + "SELECT host_id FROM local_thread_catalog_hosts WHERE LOWER(COALESCE(host_kind, '')) = 'local' ORDER BY host_id LIMIT 1" + } else { + "SELECT host_id FROM local_thread_catalog_hosts WHERE host_id = 'local' LIMIT 1" + }; + Ok(db + .query_row(sql, [], |row| row.get::<_, String>(0)) + .ok() + .or_else(|| Some("local".to_string()))) +} + +fn catalog_revision_value(db: &Connection) -> anyhow::Result { + if !table_columns(db, "local_thread_catalog_metadata")?.contains("catalog_revision") { + return Ok(0); + } + Ok(db.query_row( + "SELECT COALESCE(MAX(catalog_revision), 0) FROM local_thread_catalog_metadata", + [], + |row| row.get(0), + )?) +} + +fn increment_catalog_revision(db: &Connection, amount: usize) -> anyhow::Result<()> { + if amount == 0 + || !table_columns(db, "local_thread_catalog_metadata")?.contains("catalog_revision") + { + return Ok(()); + } + let updated = db.execute( + "UPDATE local_thread_catalog_metadata SET catalog_revision = catalog_revision + ?1", + [amount as i64], + )?; + if updated == 0 { + let columns = table_columns(db, "local_thread_catalog_metadata")?; + if columns.len() == 1 { + db.execute( + "INSERT INTO local_thread_catalog_metadata (catalog_revision) VALUES (?1)", + [amount as i64], + )?; + } else if columns.contains("id") { + db.execute( + "INSERT INTO local_thread_catalog_metadata (id, catalog_revision) VALUES (1, ?1)", + [amount as i64], + )?; + } else if columns.contains("host_id") { + db.execute( + "INSERT INTO local_thread_catalog_metadata (host_id, catalog_revision) VALUES (?1, ?2)", + (local_host_id(db)?.as_deref().unwrap_or("local"), amount as i64), + )?; + } + } + Ok(()) +} + +fn real_ids_in_db(db: &Connection) -> anyhow::Result> { + let mut ids = HashSet::new(); + for (table, column) in [ + ("threads", "id"), + ("automation_runs", "thread_id"), + ("inbox_items", "thread_id"), + ("sessions", "id"), + ("messages", "session_id"), + ("thread_dynamic_tools", "thread_id"), + ("thread_goals", "thread_id"), + ("stage1_outputs", "thread_id"), + ("agent_job_items", "assigned_thread_id"), + ] { + if !table_columns(db, table)?.contains(column) { + continue; + } + let mut stmt = db.prepare(&format!( + "SELECT {column} FROM {table} WHERE COALESCE({column}, '') <> ''" + ))?; + for id in stmt.query_map([], |row| row.get::<_, String>(0))? { + ids.insert(id?); + } + } + Ok(ids) +} + +fn insert_row(db: &Connection, table: &str, row: &Map) -> anyhow::Result<()> { + let columns = row.keys().collect::>(); + let quoted = columns + .iter() + .map(|column| format!("\"{}\"", column.replace('"', "\"\""))) + .collect::>() + .join(", "); + let marks = (1..=columns.len()) + .map(|index| format!("?{index}")) + .collect::>() + .join(", "); + let values = columns + .iter() + .map(|column| OwnedSqlValue(json_to_sql(&row[*column]))) + .collect::>(); + db.execute( + &format!("INSERT INTO {table} ({quoted}) VALUES ({marks})"), + params_from_iter(values), + )?; + Ok(()) +} + +fn sql_to_json(value: ValueRef<'_>) -> Value { + match value { + ValueRef::Null => Value::Null, + ValueRef::Integer(v) => json!(v), + ValueRef::Real(v) => json!(v), + ValueRef::Text(v) => Value::String(String::from_utf8_lossy(v).to_string()), + ValueRef::Blob(v) => { + json!({"__sqlite_blob_b64": base64::engine::general_purpose::STANDARD.encode(v)}) + } + } +} + +fn json_to_sql(value: &Value) -> SqlValue { + match value { + Value::Null => SqlValue::Null, + Value::Bool(v) => SqlValue::Integer(i64::from(*v)), + Value::Number(v) if v.is_i64() => SqlValue::Integer(v.as_i64().unwrap()), + Value::Number(v) => SqlValue::Real(v.as_f64().unwrap_or_default()), + Value::String(v) => SqlValue::Text(v.clone()), + Value::Object(map) if map.len() == 1 && map.contains_key("__sqlite_blob_b64") => { + let bytes = map + .get("__sqlite_blob_b64") + .and_then(Value::as_str) + .and_then(|encoded| { + base64::engine::general_purpose::STANDARD + .decode(encoded) + .ok() + }) + .unwrap_or_default(); + SqlValue::Blob(bytes) + } + other => SqlValue::Text(other.to_string()), + } +} + +fn row_string(row: &Map, key: &str) -> String { + row.get(key) + .and_then(Value::as_str) + .unwrap_or_default() + .to_string() +} +fn value_text(value: &Value) -> String { + value + .as_str() + .map(ToString::to_string) + .unwrap_or_else(|| value.to_string()) +} +fn sha256_hex(bytes: &[u8]) -> String { + format!("{:x}", Sha256::digest(bytes)) +} +fn split_line_ending(segment: &str) -> (&str, &str) { + if let Some(line) = segment.strip_suffix("\r\n") { + (line, "\r\n") + } else if let Some(line) = segment.strip_suffix('\n') { + (line, "\n") + } else { + (segment, "") + } +} diff --git a/crates/codex-plus-data/src/lib.rs b/crates/codex-plus-data/src/lib.rs index 26aee77f5..13d44da27 100644 --- a/crates/codex-plus-data/src/lib.rs +++ b/crates/codex-plus-data/src/lib.rs @@ -1,9 +1,15 @@ pub mod backup; +pub mod historical_cleanup; pub mod markdown; pub mod provider_sync; pub mod storage; pub use backup::BackupStore; +pub use historical_cleanup::{ + HistoricalCleanupCandidate, HistoricalCleanupError, HistoricalCleanupPreview, + HistoricalCleanupResult, apply_historical_cleanup, preview_historical_cleanup, + undo_historical_cleanup, +}; pub use markdown::{MarkdownExportService, export_markdown_from_paths}; pub use provider_sync::{ ProviderSyncResult, ProviderSyncStatus, ProviderSyncTargetList, ProviderSyncTargetOption, diff --git a/crates/codex-plus-data/src/storage.rs b/crates/codex-plus-data/src/storage.rs index ab790e816..7822de560 100644 --- a/crates/codex-plus-data/src/storage.rs +++ b/crates/codex-plus-data/src/storage.rs @@ -21,27 +21,71 @@ pub fn delete_local_from_paths( ); let mut deleted_count = 0usize; let mut backup_tokens = Vec::new(); + let mut failures = Vec::new(); for db_path in db_paths { let adapter = SQLiteStorageAdapter::new(db_path, backup_store.clone()); let candidate_result = adapter.delete_local(session); - if matches!(candidate_result.status, DeleteStatus::LocalDeleted) { - deleted_count += 1; - if let Some(token) = candidate_result.undo_token.as_ref() { - backup_tokens.push(token.clone()); + match candidate_result.status { + DeleteStatus::LocalDeleted | DeleteStatus::Partial => { + deleted_count += 1; + if let Some(token) = candidate_result.undo_token.as_ref() { + backup_tokens.push(token.clone()); + } + if matches!(candidate_result.status, DeleteStatus::Partial) { + failures.push(candidate_result.message.clone()); + } + result = candidate_result; + } + DeleteStatus::Failed if is_benign_delete_miss(&candidate_result.message) => { + if deleted_count == 0 && failures.is_empty() { + result = candidate_result; + } } - result = candidate_result; - } else if deleted_count == 0 { - result = candidate_result; + DeleteStatus::Failed => { + failures.push(candidate_result.message.clone()); + if deleted_count == 0 { + result = candidate_result; + } + } + _ => {} } } - if deleted_count > 1 { - result.message = format!("已从 {deleted_count} 个本地存储删除"); - result.undo_token = Some(json!(backup_tokens).to_string()); - result.backup_path = None; + if deleted_count > 0 { + result.status = if failures.is_empty() { + DeleteStatus::LocalDeleted + } else { + DeleteStatus::Partial + }; + result.message = if failures.is_empty() { + if deleted_count > 1 { + format!("已从 {deleted_count} 个本地存储删除") + } else { + "已从本地存储删除".to_string() + } + } else { + format!( + "已从 {deleted_count} 个本地存储删除，但部分清理失败：{}", + failures.join("; ") + ) + }; + result.undo_token = match backup_tokens.as_slice() { + [] => None, + [token] => Some(token.clone()), + _ => Some(json!(backup_tokens).to_string()), + }; + if backup_tokens.len() > 1 { + result.backup_path = None; + } } result } +fn is_benign_delete_miss(message: &str) -> bool { + message.contains("not found in local storage") + || message.starts_with("Database not found:") + || message == "Unsupported local storage schema" +} + pub fn move_codex_thread_workspace_from_paths( db_paths: impl IntoIterator, backup_store: BackupStore, @@ -72,6 +116,7 @@ enum SchemaKind { GenericSessions, CodexThreads, CodexAutomationRuns, + CodexThreadReferences, } fn sqlite_limit(limit: usize) -> i64 { @@ -134,6 +179,9 @@ impl SQLiteStorageAdapter { Some(SchemaKind::CodexAutomationRuns) => { self.delete_codex_automation_run(&mut db, session) } + Some(SchemaKind::CodexThreadReferences) => { + self.delete_codex_thread_references(&mut db, session) + } None => Ok(failed( &session.session_id, "Unsupported local storage schema".to_string(), @@ -598,6 +646,7 @@ impl SQLiteStorageAdapter { "assigned_thread_id = ?1", &[&thread_id], )?; + backup_codex_thread_reference_rows(db, &mut tables, &thread_id)?; let file_backups = rollout_file_backups(tables.get("threads").and_then(Value::as_array)); if !file_backups.is_empty() { tables.insert("__files".to_string(), Value::Array(file_backups.clone())); @@ -625,6 +674,7 @@ impl SQLiteStorageAdapter { [&thread_id], )?; } + delete_codex_thread_reference_rows(&tx, &thread_id)?; tx.execute("DELETE FROM threads WHERE id = ?1", [&thread_id])?; tx.commit()?; Ok(()) @@ -683,6 +733,7 @@ impl SQLiteStorageAdapter { "thread_id = ?1", &[&thread_id], )?; + backup_codex_thread_reference_rows(db, &mut tables, &thread_id)?; if tables.values().all(|rows| { rows.as_array() .map(|items| items.is_empty()) @@ -701,6 +752,7 @@ impl SQLiteStorageAdapter { let tx = db.transaction()?; delete_related_rows(&tx, "automation_runs", "thread_id = ?1", &[&thread_id])?; delete_related_rows(&tx, "inbox_items", "thread_id = ?1", &[&thread_id])?; + delete_codex_thread_reference_rows(&tx, &thread_id)?; tx.commit()?; Ok(()) })(); @@ -714,6 +766,117 @@ impl SQLiteStorageAdapter { } Ok(local_deleted(&thread_id, &token, &backup_path)) } + + fn delete_codex_thread_references( + &self, + db: &mut Connection, + session: &SessionRef, + ) -> anyhow::Result { + let thread_id = normalize_codex_thread_id(&session.session_id); + let mut tables = Map::new(); + backup_codex_thread_reference_rows(db, &mut tables, &thread_id)?; + if tables.values().all(|rows| { + rows.as_array() + .map(|items| items.is_empty()) + .unwrap_or(true) + }) { + return Ok(failed( + &session.session_id, + "Thread not found in local storage".to_string(), + )); + } + let token = + self.backup_store + .write_backup(&thread_id, &self.db_path, Value::Object(tables))?; + let backup_path = self.backup_store.path_for(&token); + let delete_result = (|| -> anyhow::Result<()> { + let tx = db.transaction()?; + delete_codex_thread_reference_rows(&tx, &thread_id)?; + tx.commit()?; + Ok(()) + })(); + if let Err(err) = delete_result { + return Ok(failed_with_undo( + &thread_id, + err.to_string(), + &token, + Some(&backup_path), + )); + } + Ok(local_deleted(&thread_id, &token, &backup_path)) + } +} + +fn backup_codex_thread_reference_rows( + db: &Connection, + tables: &mut Map, + thread_id: &str, +) -> anyhow::Result<()> { + if has_columns(db, "local_thread_catalog", &["thread_id"])? { + backup_related_rows( + db, + tables, + "local_thread_catalog", + "thread_id = ?1", + &[&thread_id], + )?; + } + if has_columns(db, "thread_timeline_ledger", &["thread_id"])? { + backup_related_rows( + db, + tables, + "thread_timeline_ledger", + "thread_id = ?1", + &[&thread_id], + )?; + } + Ok(()) +} + +fn delete_codex_thread_reference_rows(db: &Connection, thread_id: &str) -> anyhow::Result<()> { + let deleted_catalog_rows = if has_columns(db, "local_thread_catalog", &["thread_id"])? { + db.execute( + "DELETE FROM local_thread_catalog WHERE thread_id = ?1", + [thread_id], + )? + } else { + 0 + }; + if has_columns(db, "thread_timeline_ledger", &["thread_id"])? { + delete_related_rows( + db, + "thread_timeline_ledger", + "thread_id = ?1", + &[&thread_id], + )?; + } + if deleted_catalog_rows > 0 { + increment_local_catalog_revision(db, deleted_catalog_rows)?; + } + Ok(()) +} + +fn increment_local_catalog_revision(db: &Connection, amount: usize) -> anyhow::Result<()> { + if amount == 0 || !has_columns(db, "local_thread_catalog_metadata", &["catalog_revision"])? { + return Ok(()); + } + let affected = db.execute( + "UPDATE local_thread_catalog_metadata SET catalog_revision = catalog_revision + ?1", + [amount as i64], + )?; + if affected == 0 + && has_columns( + db, + "local_thread_catalog_metadata", + &["id", "catalog_revision"], + )? + { + db.execute( + "INSERT INTO local_thread_catalog_metadata (id, catalog_revision) VALUES (1, ?1)", + [amount as i64], + )?; + } + Ok(()) } fn optional_column_expression<'a>( @@ -908,6 +1071,12 @@ fn restore_backups( let mut db = Connection::open_with_flags(&source_db, OpenFlags::SQLITE_OPEN_READ_WRITE)?; let tx = db.transaction()?; restore_rows(&tx, tables)?; + let restored_catalog_rows = tables + .get("local_thread_catalog") + .and_then(Value::as_array) + .map(Vec::len) + .unwrap_or(0); + increment_local_catalog_revision(&tx, restored_catalog_rows)?; tx.commit()?; if let Some(files) = tables.get("__files").and_then(Value::as_array) { for file in files { @@ -1000,6 +1169,11 @@ fn schema_kind(db: &Connection) -> anyhow::Result> { if has_table(db, "automation_runs")? && has_columns(db, "automation_runs", &["thread_id"])? { return Ok(Some(SchemaKind::CodexAutomationRuns)); } + if has_columns(db, "local_thread_catalog", &["thread_id"])? + || has_columns(db, "thread_timeline_ledger", &["thread_id"])? + { + return Ok(Some(SchemaKind::CodexThreadReferences)); + } Ok(None) } @@ -1056,6 +1230,8 @@ fn validate_restore_tables(tables: &Map) -> anyhow::Result<()> { "agent_job_items", "automation_runs", "inbox_items", + "local_thread_catalog", + "thread_timeline_ledger", "__files", ]; for table in tables.keys() { @@ -1127,6 +1303,8 @@ fn restore_conflict_key_columns<'a>(table: &str, row: &'a Map) -> "thread_goals" => &["thread_id", "goal"], "thread_spawn_edges" => &["parent_thread_id", "child_thread_id"], "stage1_outputs" => &["thread_id"], + "local_thread_catalog" => &["host_id", "thread_id"], + "thread_timeline_ledger" => &["host_id", "thread_id", "sequence", "record_id"], _ => &[], }; let keys = wanted diff --git a/crates/codex-plus-data/tests/historical_cleanup.rs b/crates/codex-plus-data/tests/historical_cleanup.rs new file mode 100644 index 000000000..0f8914e46 --- /dev/null +++ b/crates/codex-plus-data/tests/historical_cleanup.rs @@ -0,0 +1,376 @@ +use codex_plus_data::{ + apply_historical_cleanup, preview_historical_cleanup, undo_historical_cleanup, +}; +use rusqlite::Connection; +use serde_json::{Value, json}; +use std::fs; +use std::path::{Path, PathBuf}; +use tempfile::tempdir; + +const GHOST: &str = "019f8d0f-068d-7b11-86ce-727daba1f76b"; + +fn catalog_db(home: &Path) -> PathBuf { + let path = home.join("sqlite/codex-dev.db"); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + let db = Connection::open(&path).unwrap(); + db.execute_batch( + "CREATE TABLE local_thread_catalog ( + host_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + display_title TEXT, + cwd TEXT, + source_updated_at TEXT, + source_detail TEXT, + PRIMARY KEY(host_id, thread_id) + ); + CREATE TABLE thread_timeline_ledger ( + host_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + updated_at TEXT, + PRIMARY KEY(host_id, thread_id) + ); + CREATE TABLE local_thread_catalog_metadata (catalog_revision INTEGER NOT NULL); + CREATE TABLE local_thread_catalog_sync_state (host_id TEXT PRIMARY KEY, sync_cursor TEXT); + INSERT INTO local_thread_catalog_metadata VALUES (7);", + ) + .unwrap(); + db.execute( + "INSERT INTO local_thread_catalog_sync_state VALUES ('local', 'cursor-keep')", + [], + ) + .unwrap(); + path +} + +fn add_ghost(db_path: &Path, source_detail: &str) { + let db = Connection::open(db_path).unwrap(); + db.execute( + "INSERT INTO local_thread_catalog VALUES ('local', ?1, '旧会话标题', 'D:/work/demo', '2026-08-01T12:00:00Z', ?2)", + (GHOST, source_detail), + ) + .unwrap(); + db.execute( + "INSERT INTO thread_timeline_ledger VALUES ('local', ?1, '2026-08-01T12:00:00Z')", + [GHOST], + ) + .unwrap(); +} + +fn write_index_and_global_state(home: &Path) { + fs::write( + home.join("session_index.jsonl"), + format!( + "{}\n{}\n", + json!({"id": GHOST, "thread_name": "旧会话标题", "updated_at": "2026-08-01T12:00:00Z"}), + json!({"id": "keep", "thread_name": "保留"}) + ), + ) + .unwrap(); + let state = json!({ + "threadBindings": {(GHOST): {"selected": true}, "keep": "keep"}, + "activeThread": GHOST, + "recentThreads": [GHOST, "keep"], + "promptHistory": [format!("普通提示文本提到了 {GHOST}，不应删除")] + }); + fs::write( + home.join(".codex-global-state.json"), + serde_json::to_vec_pretty(&state).unwrap(), + ) + .unwrap(); + fs::write( + home.join(".codex-global-state.json.bak"), + serde_json::to_vec_pretty(&state).unwrap(), + ) + .unwrap(); +} + +fn count(db_path: &Path, table: &str, id: &str) -> i64 { + Connection::open(db_path) + .unwrap() + .query_row( + &format!("SELECT COUNT(*) FROM {table} WHERE thread_id = ?1"), + [id], + |row| row.get(0), + ) + .unwrap() +} + +fn rollout(home: &Path, root: &str, id: &str) { + let path = home + .join(root) + .join(format!("2026/08/rollout-2026-08-01T00-00-00-{id}.jsonl")); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write( + path, + format!( + "{}\n", + json!({"type": "session_meta", "payload": {"id": id}}) + ), + ) + .unwrap(); +} + +#[test] +fn historical_shell_is_cleaned_safely_and_can_be_undone() { + let temp = tempdir().unwrap(); + let home = temp.path(); + let db_path = catalog_db(home); + add_ghost(&db_path, "D:/missing/rollout.jsonl"); + write_index_and_global_state(home); + + let preview = preview_historical_cleanup(Some(home)).unwrap(); + assert_eq!(preview.catalog_revision, 7); + assert_eq!(preview.candidates.len(), 1); + assert_eq!(preview.candidates[0].id, GHOST); + assert_eq!(preview.candidates[0].workspace, "D:/work/demo"); + assert_eq!( + preview.candidates[0].sources, + [ + "catalog", + "global_state", + "global_state_bak", + "session_index", + "timeline" + ] + ); + + let result = + apply_historical_cleanup(Some(home), &preview.snapshot_sha256, &[GHOST.to_string()]) + .unwrap(); + assert_eq!(result.catalog_rows, 1); + assert_eq!(result.timeline_rows, 1); + assert_eq!(result.session_index_entries, 1); + assert!(result.global_state_references >= 3); + assert_eq!(count(&db_path, "local_thread_catalog", GHOST), 0); + assert_eq!(count(&db_path, "thread_timeline_ledger", GHOST), 0); + let revision: i64 = Connection::open(&db_path) + .unwrap() + .query_row( + "SELECT catalog_revision FROM local_thread_catalog_metadata", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(revision, 8); + let sync_cursor: String = Connection::open(&db_path) + .unwrap() + .query_row( + "SELECT sync_cursor FROM local_thread_catalog_sync_state WHERE host_id = 'local'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(sync_cursor, "cursor-keep"); + let state: Value = + serde_json::from_slice(&fs::read(home.join(".codex-global-state.json")).unwrap()).unwrap(); + let expected_prompt = format!("普通提示文本提到了 {GHOST}，不应删除"); + assert_eq!( + state.pointer("/promptHistory/0").and_then(Value::as_str), + Some(expected_prompt.as_str()) + ); + assert!( + !state + .to_string() + .contains(&format!("\"activeThread\":\"{GHOST}\"")) + ); + + let backup = result.backup_dir.unwrap(); + assert!(backup.join("manifest.json").is_file()); + undo_historical_cleanup(Some(home), &backup).unwrap(); + assert_eq!(count(&db_path, "local_thread_catalog", GHOST), 1); + assert_eq!(count(&db_path, "thread_timeline_ledger", GHOST), 1); + assert!( + fs::read_to_string(home.join("session_index.jsonl")) + .unwrap() + .contains(GHOST) + ); + let restored: Value = + serde_json::from_slice(&fs::read(home.join(".codex-global-state.json")).unwrap()).unwrap(); + assert_eq!( + restored.pointer("/activeThread").and_then(Value::as_str), + Some(GHOST) + ); +} + +#[test] +fn real_database_thread_is_not_a_candidate() { + let temp = tempdir().unwrap(); + let home = temp.path(); + let db_path = catalog_db(home); + add_ghost(&db_path, ""); + let state = Connection::open(home.join("state_5.sqlite")).unwrap(); + state + .execute("CREATE TABLE threads (id TEXT PRIMARY KEY)", []) + .unwrap(); + state + .execute("INSERT INTO threads VALUES (?1)", [GHOST]) + .unwrap(); + assert!( + preview_historical_cleanup(Some(home)) + .unwrap() + .candidates + .is_empty() + ); +} + +#[test] +fn message_body_source_is_not_a_candidate() { + let temp = tempdir().unwrap(); + let home = temp.path(); + let db_path = catalog_db(home); + add_ghost(&db_path, ""); + let state = Connection::open(home.join("state_5.sqlite")).unwrap(); + state + .execute("CREATE TABLE messages (session_id TEXT, body TEXT)", []) + .unwrap(); + state + .execute("INSERT INTO messages VALUES (?1, '正文')", [GHOST]) + .unwrap(); + assert!( + preview_historical_cleanup(Some(home)) + .unwrap() + .candidates + .is_empty() + ); +} + +#[test] +fn legacy_session_index_only_shell_remains_cleanable() { + let temp = tempdir().unwrap(); + let home = temp.path(); + write_index_and_global_state(home); + let preview = preview_historical_cleanup(Some(home)).unwrap(); + assert_eq!(preview.candidates.len(), 1); + assert_eq!(preview.candidates[0].id, GHOST); + assert_eq!( + preview.candidates[0].sources, + ["global_state", "global_state_bak", "session_index"] + ); +} + +#[test] +fn remote_host_catalog_thread_is_not_a_candidate() { + let temp = tempdir().unwrap(); + let home = temp.path(); + let db_path = catalog_db(home); + Connection::open(&db_path) + .unwrap() + .execute( + "INSERT INTO local_thread_catalog VALUES ('remote-ssh', ?1, '远程会话', '', '', '')", + [GHOST], + ) + .unwrap(); + write_index_and_global_state(home); + assert!( + preview_historical_cleanup(Some(home)) + .unwrap() + .candidates + .is_empty() + ); +} + +#[test] +fn active_and_archived_rollouts_are_not_candidates() { + for root in ["sessions", "archived_sessions"] { + let temp = tempdir().unwrap(); + let home = temp.path(); + let db_path = catalog_db(home); + add_ghost(&db_path, ""); + rollout(home, root, GHOST); + assert!( + preview_historical_cleanup(Some(home)) + .unwrap() + .candidates + .is_empty(), + "{root}" + ); + } +} + +#[test] +fn changed_source_rejects_apply_and_undo_rejects_same_id_conflict() { + let temp = tempdir().unwrap(); + let home = temp.path(); + let db_path = catalog_db(home); + add_ghost(&db_path, ""); + write_index_and_global_state(home); + let preview = preview_historical_cleanup(Some(home)).unwrap(); + fs::write(home.join("session_index.jsonl"), "{\"id\":\"changed\"}\n").unwrap(); + let error = + apply_historical_cleanup(Some(home), &preview.snapshot_sha256, &[GHOST.to_string()]) + .unwrap_err(); + assert!(error.message.contains("发生变化")); + + write_index_and_global_state(home); + let preview = preview_historical_cleanup(Some(home)).unwrap(); + let result = + apply_historical_cleanup(Some(home), &preview.snapshot_sha256, &[GHOST.to_string()]) + .unwrap(); + let state = Connection::open(home.join("state_5.sqlite")).unwrap(); + state + .execute("CREATE TABLE threads (id TEXT PRIMARY KEY)", []) + .unwrap(); + state + .execute("INSERT INTO threads VALUES (?1)", [GHOST]) + .unwrap(); + let error = + undo_historical_cleanup(Some(home), result.backup_dir.as_ref().unwrap()).unwrap_err(); + assert!(error.message.contains("同 ID")); + assert_eq!(count(&db_path, "local_thread_catalog", GHOST), 0); +} + +#[test] +fn sqlite_failure_is_reported_and_backup_remains_available() { + let temp = tempdir().unwrap(); + let home = temp.path(); + let db_path = catalog_db(home); + add_ghost(&db_path, ""); + write_index_and_global_state(home); + let db = Connection::open(&db_path).unwrap(); + db.execute_batch( + "CREATE TRIGGER reject_catalog_delete BEFORE DELETE ON local_thread_catalog + BEGIN SELECT RAISE(ABORT, 'blocked'); END;", + ) + .unwrap(); + drop(db); + let preview = preview_historical_cleanup(Some(home)).unwrap(); + let error = + apply_historical_cleanup(Some(home), &preview.snapshot_sha256, &[GHOST.to_string()]) + .unwrap_err(); + assert!( + error + .backup_dir + .as_ref() + .is_some_and(|path| path.join("manifest.json").is_file()) + ); + assert_eq!(count(&db_path, "local_thread_catalog", GHOST), 1); + assert_eq!(count(&db_path, "thread_timeline_ledger", GHOST), 1); +} + +#[test] +fn json_write_failure_is_reported_before_database_changes() { + let temp = tempdir().unwrap(); + let home = temp.path(); + let db_path = catalog_db(home); + add_ghost(&db_path, ""); + write_index_and_global_state(home); + fs::create_dir(home.join("session_index.jsonl.tmp")).unwrap(); + + let preview = preview_historical_cleanup(Some(home)).unwrap(); + let error = + apply_historical_cleanup(Some(home), &preview.snapshot_sha256, &[GHOST.to_string()]) + .unwrap_err(); + assert!( + error + .backup_dir + .as_ref() + .is_some_and(|path| path.join("manifest.json").is_file()) + ); + assert_eq!(count(&db_path, "local_thread_catalog", GHOST), 1); + assert_eq!(count(&db_path, "thread_timeline_ledger", GHOST), 1); + assert!( + fs::read_to_string(home.join("session_index.jsonl")) + .unwrap() + .contains(GHOST) + ); +} diff --git a/crates/codex-plus-data/tests/storage_adapter.rs b/crates/codex-plus-data/tests/storage_adapter.rs index 9fd1dfc70..8047a6d0b 100644 --- a/crates/codex-plus-data/tests/storage_adapter.rs +++ b/crates/codex-plus-data/tests/storage_adapter.rs @@ -86,6 +86,40 @@ fn create_codex_thread_db(path: &Path, rollout_path: &Path) { .unwrap(); } +fn create_codex_catalog_db(path: &Path, revision: i64) { + let db = Connection::open(path).unwrap(); + db.execute( + "CREATE TABLE local_thread_catalog (host_id TEXT NOT NULL, thread_id TEXT NOT NULL, display_title TEXT NOT NULL, PRIMARY KEY (host_id, thread_id))", + [], + ) + .unwrap(); + db.execute( + "INSERT INTO local_thread_catalog VALUES ('local', 't1', 'Codex Thread')", + [], + ) + .unwrap(); + db.execute( + "CREATE TABLE local_thread_catalog_metadata (id INTEGER PRIMARY KEY, catalog_revision INTEGER NOT NULL DEFAULT 0)", + [], + ) + .unwrap(); + db.execute( + "INSERT INTO local_thread_catalog_metadata VALUES (1, ?1)", + [revision], + ) + .unwrap(); + db.execute( + "CREATE TABLE thread_timeline_ledger (host_id TEXT NOT NULL, thread_id TEXT NOT NULL, sequence INTEGER NOT NULL, record_id TEXT NOT NULL, payload_json TEXT NOT NULL, PRIMARY KEY (host_id, thread_id, sequence, record_id))", + [], + ) + .unwrap(); + db.execute( + "INSERT INTO thread_timeline_ledger VALUES ('local', 't1', 1, 'r1', '{}')", + [], + ) + .unwrap(); +} + fn thread_count(path: &Path, id: &str) -> i64 { let db = Connection::open(path).unwrap(); db.query_row("SELECT COUNT(*) FROM threads WHERE id = ?1", [id], |row| { @@ -507,6 +541,124 @@ fn delete_local_from_paths_undo_restores_duplicate_threads_and_shared_rollout_to ); } +#[test] +fn delete_local_from_paths_removes_and_restores_codex_catalog_references() { + let tmp = tempdir().unwrap(); + let thread_db = tmp.path().join("state_5.sqlite"); + let catalog_db = tmp.path().join("codex-dev.db"); + let rollout = tmp.path().join("rollout.jsonl"); + fs::write(&rollout, "{\"type\":\"message\"}\n").unwrap(); + create_codex_thread_db(&thread_db, &rollout); + create_codex_catalog_db(&catalog_db, 7); + let backups = BackupStore::new(tmp.path().join("backups")); + + let deleted = delete_local_from_paths( + vec![thread_db.clone(), catalog_db.clone()], + backups.clone(), + &session("t1", "Codex Thread"), + ); + + assert_eq!(deleted.status, DeleteStatus::LocalDeleted); + assert_eq!(thread_count(&thread_db, "t1"), 0); + assert!(!rollout.exists()); + let catalog = Connection::open(&catalog_db).unwrap(); + for table in ["local_thread_catalog", "thread_timeline_ledger"] { + assert_eq!( + catalog + .query_row( + &format!("SELECT COUNT(*) FROM {table} WHERE thread_id = 't1'"), + [], + |row| row.get::<_, i64>(0), + ) + .unwrap(), + 0 + ); + } + assert_eq!(catalog_revision(&catalog), 8); + drop(catalog); + + let token = deleted.undo_token.as_deref().unwrap(); + let restored = SQLiteStorageAdapter::new(&thread_db, backups) + .with_allowed_db_paths(vec![thread_db.clone(), catalog_db.clone()]) + .undo(token); + + assert_eq!(restored.status, DeleteStatus::Undone); + assert_eq!(thread_count(&thread_db, "t1"), 1); + assert!(rollout.exists()); + let catalog = Connection::open(&catalog_db).unwrap(); + for table in ["local_thread_catalog", "thread_timeline_ledger"] { + assert_eq!( + catalog + .query_row( + &format!("SELECT COUNT(*) FROM {table} WHERE thread_id = 't1'"), + [], + |row| row.get::<_, i64>(0), + ) + .unwrap(), + 1 + ); + } + assert_eq!(catalog_revision(&catalog), 9); +} + +#[test] +fn delete_local_from_paths_reports_partial_failure_and_keeps_successful_backup_undoable() { + let tmp = tempdir().unwrap(); + let thread_db = tmp.path().join("state_5.sqlite"); + let catalog_db = tmp.path().join("codex-dev.db"); + let rollout = tmp.path().join("rollout.jsonl"); + fs::write(&rollout, "{\"type\":\"message\"}\n").unwrap(); + create_codex_thread_db(&thread_db, &rollout); + create_codex_catalog_db(&catalog_db, 7); + Connection::open(&catalog_db) + .unwrap() + .execute( + "CREATE TRIGGER fail_catalog_delete BEFORE DELETE ON local_thread_catalog BEGIN SELECT RAISE(ABORT, 'catalog busy'); END", + [], + ) + .unwrap(); + let backups = BackupStore::new(tmp.path().join("backups")); + + let deleted = delete_local_from_paths( + vec![thread_db.clone(), catalog_db.clone()], + backups.clone(), + &session("t1", "Codex Thread"), + ); + + assert_eq!(deleted.status, DeleteStatus::Partial); + assert!(deleted.message.contains("catalog busy")); + assert_eq!(thread_count(&thread_db, "t1"), 0); + assert!(!rollout.exists()); + assert_eq!( + Connection::open(&catalog_db) + .unwrap() + .query_row( + "SELECT COUNT(*) FROM local_thread_catalog WHERE thread_id = 't1'", + [], + |row| row.get::<_, i64>(0), + ) + .unwrap(), + 1 + ); + + let token = deleted.undo_token.as_deref().unwrap(); + let restored = SQLiteStorageAdapter::new(&thread_db, backups) + .with_allowed_db_paths(vec![thread_db.clone(), catalog_db]) + .undo(token); + assert_eq!(restored.status, DeleteStatus::Undone); + assert_eq!(thread_count(&thread_db, "t1"), 1); + assert!(rollout.exists()); +} + +fn catalog_revision(db: &Connection) -> i64 { + db.query_row( + "SELECT catalog_revision FROM local_thread_catalog_metadata WHERE id = 1", + [], + |row| row.get(0), + ) + .unwrap() +} + #[test] fn grouped_undo_preflights_all_databases_before_restoring_any() { let tmp = tempdir().unwrap();