From ebc2902f33e52b4ff1a1b27a7a5141ec2e25c88d Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Sat, 13 Jun 2026 13:32:10 +0800 Subject: [PATCH] feat: add store filter for multi-store surfaces --- src/cli.rs | 39 +++++++++ src/main.rs | 20 ++++- src/output.rs | 42 +++++++++ src/rollout.rs | 3 +- src/stores.rs | 76 ++++++++++++++++- src/sync.rs | 57 +++++++++++-- src/tests.rs | 228 +++++++++++++++++++++++++++++++++++++++++++++++-- src/watch.rs | 5 ++ 8 files changed, 454 insertions(+), 16 deletions(-) diff --git a/src/cli.rs b/src/cli.rs index 1230156..f75702b 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -11,6 +11,7 @@ use std::path::PathBuf; use crate::codex_config::is_valid_profile_name; use crate::locale::Locale; use crate::output::*; +use crate::stores::StoreFilter; pub(crate) const APP_VERSION: &str = env!("CARGO_PKG_VERSION"); pub(crate) const DEFAULT_POLL_INTERVAL_MS: u64 = 500; @@ -36,6 +37,16 @@ pub(crate) struct Cli { #[arg(long, global = true, value_name = "PROFILE", help = "placeholder")] pub(crate) profile: Option, + #[arg( + long, + global = true, + value_enum, + default_value_t = StoreFilter::All, + value_name = "STORE", + help = "placeholder" + )] + pub(crate) store: StoreFilter, + #[command(subcommand)] pub(crate) command: Command, } @@ -164,6 +175,11 @@ pub(crate) fn localized_command(locale: Locale) -> clap::Command { .help_heading(options_heading(locale)) .value_name(profile_value_name(locale)) }); + command = command.mut_arg("store", |arg| { + arg.help(store_help(locale)) + .help_heading(options_heading(locale)) + .value_name(store_value_name(locale)) + }); command = command.mut_subcommand("status", |sub| { sub.about(status_about(locale)) @@ -306,6 +322,29 @@ pub(crate) fn validate_profile_override(locale: Locale, profile: Option<&str>) - Ok(()) } +pub(crate) fn validate_store_filter_supported( + locale: Locale, + store: StoreFilter, + command: &str, +) -> Result<()> { + if store != StoreFilter::All { + anyhow::bail!(store_filter_unsupported_error(locale, command)); + } + Ok(()) +} + +pub(crate) fn validate_store_filter_rollout_scope( + locale: Locale, + store: StoreFilter, + sqlite_only: bool, + command: &str, +) -> Result<()> { + if store != StoreFilter::All && !sqlite_only { + anyhow::bail!(store_filter_requires_sqlite_only_error(locale, command)); + } + Ok(()) +} + pub(crate) fn validate_provider_override_args(locale: Locale, args: I) -> Result<()> where I: IntoIterator, diff --git a/src/main.rs b/src/main.rs index 79515a9..0469579 100644 --- a/src/main.rs +++ b/src/main.rs @@ -25,6 +25,8 @@ use cli::DEFAULT_BUCKET_PADDING_BYTES; use cli::parse_cli; use cli::validate_profile_override; use cli::validate_provider_override; +use cli::validate_store_filter_rollout_scope; +use cli::validate_store_filter_supported; use locale::Locale; use locale::detect_locale; use output::bucket_switch_complete_title; @@ -70,12 +72,17 @@ fn run() -> Result { match cli.command { Command::Status => { - let summary = - collect_status(&codex_home, cli.provider.as_deref(), cli.profile.as_deref())?; + let summary = collect_status( + &codex_home, + cli.provider.as_deref(), + cli.profile.as_deref(), + cli.store, + )?; print_status(locale, &summary); Ok(ExitCode::SUCCESS) } Command::Sync { sqlite_only } => { + validate_store_filter_rollout_scope(locale, cli.store, sqlite_only, "sync")?; let rollout_scope = if sqlite_only { RolloutScope::None } else { @@ -93,6 +100,7 @@ fn run() -> Result { rollout_scope, DEFAULT_BUCKET_PADDING_BYTES, DEFAULT_BACKFILL_WAIT, + cli.store, progress, )?; print_multi_sync_summary(locale, sync_complete_title(locale), &summary); @@ -103,6 +111,7 @@ fn run() -> Result { } Command::Bucket { command } => match command { BucketCommand::Prepare { padding_bytes } => { + validate_store_filter_supported(locale, cli.store, "bucket prepare")?; let summary = prepare_bucket_padding(&codex_home, cli.profile.as_deref(), padding_bytes)?; print_bucket_prepare_summary(locale, &summary); @@ -112,6 +121,7 @@ fn run() -> Result { target_provider, padding_bytes, } => { + validate_store_filter_rollout_scope(locale, cli.store, false, "bucket switch")?; validate_provider_override(locale, target_provider.as_deref())?; let provider = match target_provider { Some(provider) => Some(provider), @@ -124,6 +134,7 @@ fn run() -> Result { RolloutScope::AllRows, padding_bytes, DEFAULT_BACKFILL_WAIT, + cli.store, Some(RolloutProgressConfig { locale }), )?; print_multi_sync_summary(locale, bucket_switch_complete_title(locale), &summary); @@ -134,11 +145,13 @@ fn run() -> Result { poll_interval_ms, sqlite_only, } => { + validate_store_filter_rollout_scope(locale, cli.store, sqlite_only, "watch")?; run_watch( locale, &codex_home, cli.provider.clone(), cli.profile.clone(), + cli.store, if sqlite_only { RolloutScope::None } else { @@ -149,6 +162,7 @@ fn run() -> Result { Ok(ExitCode::SUCCESS) } Command::PrintServiceConfig { poll_interval_ms } => { + validate_store_filter_supported(locale, cli.store, "print-service-config")?; let exe_path = std::env::current_exe().context(current_exe_error(locale))?; let config = service::render_service_config( exe_path.as_path(), @@ -161,6 +175,7 @@ fn run() -> Result { Ok(ExitCode::SUCCESS) } Command::InstallService { poll_interval_ms } => { + validate_store_filter_supported(locale, cli.store, "install-service")?; install_service( locale, &codex_home, @@ -171,6 +186,7 @@ fn run() -> Result { Ok(ExitCode::SUCCESS) } Command::UninstallService => { + validate_store_filter_supported(locale, cli.store, "uninstall-service")?; uninstall_service(locale, &codex_home)?; Ok(ExitCode::SUCCESS) } diff --git a/src/output.rs b/src/output.rs index bfa1d10..b2916da 100644 --- a/src/output.rs +++ b/src/output.rs @@ -311,6 +311,22 @@ pub(crate) fn provider_value_name(locale: Locale) -> &'static str { } } +pub(crate) fn store_value_name(locale: Locale) -> &'static str { + match locale { + Locale::En => "STORE", + Locale::ZhHans => "STORE", + } +} + +pub(crate) fn store_help(locale: Locale) -> &'static str { + match locale { + Locale::En => { + "Limit which storage surface to operate on: cli, app, configured, or all (default: all); filtered writes require --sqlite-only" + } + Locale::ZhHans => "限定操作的存储面：cli、app、configured 或 all（默认：all）", + } +} + pub(crate) fn profile_value_name(locale: Locale) -> &'static str { match locale { Locale::En => "PROFILE", @@ -380,6 +396,32 @@ pub(crate) fn profile_path_error(locale: Locale) -> &'static str { } } +pub(crate) fn store_filter_unsupported_error(locale: Locale, command: &str) -> String { + match locale { + Locale::En => format!( + "--store is not supported for `{command}`; use --store only with status, sync --sqlite-only, or watch --sqlite-only" + ), + Locale::ZhHans => { + format!( + "--store 不支持 `{command}`；它只能用于 status、sync --sqlite-only 或 watch --sqlite-only" + ) + } + } +} + +pub(crate) fn store_filter_requires_sqlite_only_error(locale: Locale, command: &str) -> String { + match locale { + Locale::En => format!( + "--store on `{command}` can only be used with --sqlite-only because rollout JSONL is shared across stores" + ), + Locale::ZhHans => { + format!( + "`{command}` 使用 --store 时必须同时使用 --sqlite-only，因为 rollout JSONL 在多个库之间共享" + ) + } + } +} + pub(crate) fn help_option_help(locale: Locale) -> &'static str { match locale { Locale::En => "Print help", diff --git a/src/rollout.rs b/src/rollout.rs index bb8e494..a02ae15 100644 --- a/src/rollout.rs +++ b/src/rollout.rs @@ -29,6 +29,7 @@ use crate::output::RolloutProgressSnapshot; use crate::output::rollout_progress_message; use crate::state_db::ensure_sqlite_exists; use crate::state_db::unix_timestamp_millis; +use crate::stores::StoreFilter; use crate::stores::discover_stores; const ROLLOUT_PROGRESS_INTERVAL: Duration = Duration::from_millis(500); @@ -415,7 +416,7 @@ fn prepare_bucket_padding_unlocked( padding_bytes: usize, ) -> Result { let started = Instant::now(); - let store_db_paths = discover_stores(codex_home, profile_override)? + let store_db_paths = discover_stores(codex_home, profile_override, StoreFilter::All)? .into_iter() .map(|store| store.db_path) .collect::>(); diff --git a/src/stores.rs b/src/stores.rs index 6e7aaf4..277f466 100644 --- a/src/stores.rs +++ b/src/stores.rs @@ -61,15 +61,56 @@ pub(crate) struct StoreTarget { pub(crate) db_path: PathBuf, } +/// `--store` selector: narrow which discovered surface(s) a command operates on. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, clap::ValueEnum)] +pub(crate) enum StoreFilter { + /// Every discovered store (default). + #[default] + All, + /// Only the Codex CLI default (`/state_5.sqlite`). + Cli, + /// Only the Codex App store (`/sqlite/state_5.sqlite`). + App, + /// Only an explicit `sqlite_home` / `CODEX_SQLITE_HOME` store. + Configured, +} + +impl StoreFilter { + pub(crate) fn slug(self) -> &'static str { + match self { + StoreFilter::All => "all", + StoreFilter::Cli => "cli", + StoreFilter::App => "app", + StoreFilter::Configured => "configured", + } + } + + pub(crate) fn matches(self, kind: StoreKind) -> bool { + match self { + StoreFilter::All => true, + StoreFilter::Cli => kind == StoreKind::Cli, + StoreFilter::App => kind == StoreKind::App, + StoreFilter::Configured => kind == StoreKind::Configured, + } + } +} + /// Discover every existing `state_5.sqlite` store under `codex_home`, /// canonicalized and de-duplicated. Reads the configured `sqlite_home` / /// `CODEX_SQLITE_HOME` (if any) and then layers the App and CLI defaults. pub(crate) fn discover_stores( codex_home: &Path, profile_override: Option<&str>, + filter: StoreFilter, ) -> Result> { let configured = configured_sqlite_home(codex_home, profile_override)?; - Ok(discover_stores_with(codex_home, configured.as_deref())) + if filter == StoreFilter::All { + return Ok(discover_stores_with(codex_home, configured.as_deref())); + } + let candidates = store_candidates(codex_home, configured.as_deref()) + .into_iter() + .filter(|(kind, _)| filter.matches(*kind)); + Ok(discover_stores_from_candidates(candidates)) } /// Pure core of [`discover_stores`]: builds candidates from an already-resolved @@ -82,6 +123,13 @@ pub(crate) fn discover_stores_with( codex_home: &Path, configured_sqlite_home: Option<&Path>, ) -> Vec { + discover_stores_from_candidates(store_candidates(codex_home, configured_sqlite_home)) +} + +fn store_candidates( + codex_home: &Path, + configured_sqlite_home: Option<&Path>, +) -> Vec<(StoreKind, PathBuf)> { let mut candidates: Vec<(StoreKind, PathBuf)> = Vec::new(); if let Some(dir) = configured_sqlite_home { candidates.push((StoreKind::Configured, dir.join(STATE_DB_FILENAME))); @@ -91,7 +139,13 @@ pub(crate) fn discover_stores_with( codex_home.join(APP_SQLITE_SUBDIR).join(STATE_DB_FILENAME), )); candidates.push((StoreKind::Cli, codex_home.join(STATE_DB_FILENAME))); + candidates +} +fn discover_stores_from_candidates(candidates: I) -> Vec +where + I: IntoIterator, +{ let mut seen: HashSet = HashSet::new(); let mut stores: Vec = Vec::new(); for (kind, path) in candidates { @@ -128,3 +182,23 @@ pub(crate) fn no_store_found_message(locale: Locale, codex_home: &Path) -> Strin ), } } + +/// Error message shown when stores exist, but none match the selected `--store`. +pub(crate) fn no_store_selected_message( + locale: Locale, + codex_home: &Path, + filter: StoreFilter, +) -> String { + match locale { + Locale::En => format!( + "no Codex state database under {} matched --store {}; run `codex-threadripper status --store all` to see detected stores", + codex_home.display(), + filter.slug() + ), + Locale::ZhHans => format!( + "{} 下没有匹配 --store {} 的 Codex 状态库；可运行 `codex-threadripper status --store all` 查看已发现的存储面", + codex_home.display(), + filter.slug() + ), + } +} diff --git a/src/sync.rs b/src/sync.rs index a229413..c39eda8 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -25,10 +25,12 @@ use crate::state_db::read_backfill_status; use crate::state_db::read_backfill_status_with_timeout; use crate::state_db::reconcile_sqlite_in_place; use crate::state_db::unix_timestamp_millis; +use crate::stores::StoreFilter; use crate::stores::StoreKind; use crate::stores::StoreTarget; use crate::stores::discover_stores; use crate::stores::no_store_found_message; +use crate::stores::no_store_selected_message; /// Per-store status for a single discovered `state_5.sqlite` surface. #[derive(Debug)] @@ -55,6 +57,7 @@ pub(crate) fn collect_status( codex_home: &Path, provider_override: Option<&str>, profile_override: Option<&str>, + filter: StoreFilter, ) -> Result { let config_path = codex_home.join("config.toml"); let provider = match provider_override { @@ -62,8 +65,17 @@ pub(crate) fn collect_status( None => read_provider_from_config(codex_home, profile_override)?, }; - let targets = discover_stores(codex_home, profile_override)?; + let targets = discover_stores(codex_home, profile_override, filter)?; if targets.is_empty() { + if filter != StoreFilter::All + && !discover_stores(codex_home, profile_override, StoreFilter::All)?.is_empty() + { + anyhow::bail!(no_store_selected_message( + detect_locale(), + codex_home, + filter + )); + } anyhow::bail!(no_store_found_message(detect_locale(), codex_home)); } let mut stores = Vec::with_capacity(targets.len()); @@ -253,6 +265,7 @@ fn is_sqlite_lock_error(error: &anyhow::Error) -> bool { /// rollout JSONL, backing up each store first. This is the multi-store write /// path for the one-shot `sync` / `bucket switch` commands. A store whose Codex /// backfill is still running after `backfill_wait` is skipped, not written. +#[allow(clippy::too_many_arguments)] pub(crate) fn reconcile_all_stores_with_backup( codex_home: &Path, provider_override: Option<&str>, @@ -260,6 +273,7 @@ pub(crate) fn reconcile_all_stores_with_backup( rollout_scope: RolloutScope, padding_bytes: usize, backfill_wait: Duration, + filter: StoreFilter, progress: Option, ) -> Result { with_threadripper_lock(codex_home, || { @@ -271,6 +285,7 @@ pub(crate) fn reconcile_all_stores_with_backup( padding_bytes, backfill_wait, true, + filter, progress, ) }) @@ -281,12 +296,14 @@ pub(crate) fn reconcile_all_stores_with_backup( /// every time would be wasteful. Supports the incremental `MismatchedRows` scope /// (with an `AllRows` rollout followup when rows change), like the former /// single-store watch path. +#[allow(clippy::too_many_arguments)] pub(crate) fn reconcile_all_stores( codex_home: &Path, provider_override: Option<&str>, profile_override: Option<&str>, rollout_scope: RolloutScope, backfill_wait: Duration, + filter: StoreFilter, progress: Option, ) -> Result { with_threadripper_lock(codex_home, || { @@ -298,6 +315,7 @@ pub(crate) fn reconcile_all_stores( DEFAULT_BUCKET_PADDING_BYTES, backfill_wait, false, + filter, progress, ) }) @@ -312,23 +330,52 @@ fn reconcile_stores_core( padding_bytes: usize, backfill_wait: Duration, backup: bool, + filter: StoreFilter, progress: Option, ) -> Result { let provider = match provider_override { Some(provider) => provider.to_string(), None => read_provider_from_config(codex_home, profile_override)?, }; - let targets = discover_stores(codex_home, profile_override)?; - if targets.is_empty() { + if filter != StoreFilter::All && rollout_scope != RolloutScope::None { + anyhow::bail!( + "--store {} can only be used with --sqlite-only because rollout JSONL is shared across stores", + filter.slug() + ); + } + let all_targets = discover_stores(codex_home, profile_override, StoreFilter::All)?; + if all_targets.is_empty() { anyhow::bail!(no_store_found_message(detect_locale(), codex_home)); } + let targets = if filter == StoreFilter::All { + all_targets.clone() + } else { + discover_stores(codex_home, profile_override, filter)? + }; + if targets.is_empty() { + anyhow::bail!(no_store_selected_message( + detect_locale(), + codex_home, + filter + )); + } let started = Instant::now(); // 0) Backfill guard: a store whose Codex startup-backfill is still running // after the bounded wait is skipped entirely — we neither collect its // (possibly partial) rollout targets nor write its DB, so we never race // the rebuild. - let busy: HashSet = targets + // + // Rollout JSONL is shared across CLI/App/configured stores, so any + // rollout-rewriting scope must guard every discovered store, even if + // `--store` selected just one of them. `--sqlite-only` touches no rollout, + // so it only waits on and skips the selected stores. + let backfill_guard_targets: &[StoreTarget] = if rollout_scope == RolloutScope::None { + targets.as_slice() + } else { + all_targets.as_slice() + }; + let busy: HashSet = backfill_guard_targets .iter() .filter(|target| { wait_for_store_backfill(&target.db_path, backfill_wait) == BackfillReadiness::Busy @@ -498,7 +545,7 @@ fn reconcile_stores_core( }) .collect(); - // 3) MismatchedRows is incremental: the first pass only rewrote rollouts for + // 4) MismatchedRows is incremental: the first pass only rewrote rollouts for // rows the DB had marked mismatched. After flipping those rows, run a full // AllRows rollout pass so any rollout that drifted is corrected too. (busy // is empty here — the whole-round skip above already returned if a backfill diff --git a/src/tests.rs b/src/tests.rs index 3a41c56..80660bb 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -6,6 +6,8 @@ use crate::cli::DEFAULT_POLL_INTERVAL_MS; use crate::cli::localized_command; use crate::cli::validate_provider_override; use crate::cli::validate_provider_override_args; +use crate::cli::validate_store_filter_rollout_scope; +use crate::cli::validate_store_filter_supported; use crate::codex_config::read_provider_from_config; use crate::codex_config::resolve_sqlite_home_from_config; use crate::codex_config::resolve_sqlite_path; @@ -21,7 +23,9 @@ use crate::service::ServiceManager; use crate::state_db::inspect_sqlite_distribution; use crate::state_db::reconcile_sqlite_in_place; use crate::state_db::reconcile_sqlite_with_backup; +use crate::stores::StoreFilter; use crate::stores::StoreKind; +use crate::stores::discover_stores; use crate::stores::discover_stores_with; use crate::sync::MultiReconcileSummary; use crate::sync::ReconcileStatus; @@ -382,6 +386,41 @@ fn discover_dedupes_configured_pointing_at_app_default() -> Result<()> { Ok(()) } +#[test] +fn store_filter_keeps_cli_when_configured_aliases_cli_default() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = dir.path(); + fs::write(home.join("config.toml"), "sqlite_home = \".\"\n")?; + fs::write(home.join("state_5.sqlite"), b"")?; + + let stores = discover_stores(home, None, StoreFilter::Cli)?; + + assert_eq!(stores.len(), 1); + assert_eq!(stores[0].kind, StoreKind::Cli); + let all = discover_stores(home, None, StoreFilter::All)?; + assert_eq!(all.len(), 1); + assert_eq!(all[0].kind, StoreKind::Configured); + Ok(()) +} + +#[test] +fn store_filter_keeps_app_when_configured_aliases_app_default() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = dir.path(); + fs::write(home.join("config.toml"), "sqlite_home = \"sqlite\"\n")?; + fs::create_dir_all(home.join("sqlite"))?; + fs::write(home.join("sqlite").join("state_5.sqlite"), b"")?; + + let stores = discover_stores(home, None, StoreFilter::App)?; + + assert_eq!(stores.len(), 1); + assert_eq!(stores[0].kind, StoreKind::App); + let all = discover_stores(home, None, StoreFilter::All)?; + assert_eq!(all.len(), 1); + assert_eq!(all[0].kind, StoreKind::Configured); + Ok(()) +} + #[test] fn discover_orders_configured_app_cli_when_all_distinct() -> Result<()> { let dir = tempfile::tempdir()?; @@ -437,7 +476,7 @@ fn status_reports_broken_store_without_hiding_healthy_store() -> Result<()> { fs::create_dir_all(app_db.parent().unwrap())?; fs::write(&app_db, b"not a sqlite database")?; - let summary = collect_status(home, Some("openai"), None)?; + let summary = collect_status(home, Some("openai"), None, StoreFilter::All)?; let cli = summary .stores @@ -466,7 +505,7 @@ fn status_reports_missing_configured_store_alongside_default_store() -> Result<( let cli_db = home.join("state_5.sqlite"); seed_sqlite(&cli_db)?; - let summary = collect_status(home, Some("openai"), None)?; + let summary = collect_status(home, Some("openai"), None, StoreFilter::All)?; let configured = summary .stores @@ -490,7 +529,7 @@ fn status_errors_when_every_store_is_broken() -> Result<()> { fs::write(home.join("config.toml"), sqlite_home_config(home))?; fs::write(home.join("state_5.sqlite"), b"not a sqlite database")?; - let err = collect_status(home, Some("openai"), None).unwrap_err(); + let err = collect_status(home, Some("openai"), None, StoreFilter::All).unwrap_err(); assert!(err.to_string().contains("failed to inspect any")); Ok(()) @@ -529,6 +568,7 @@ fn reconcile_all_stores_updates_every_surface_and_dedupes_rollout() -> Result<() RolloutScope::AllRows, DEFAULT_BUCKET_PADDING_BYTES, Duration::from_millis(0), + StoreFilter::All, None, )?; @@ -590,6 +630,7 @@ fn reconcile_all_stores_reports_partial_when_a_store_is_unreadable() -> Result<( RolloutScope::None, DEFAULT_BUCKET_PADDING_BYTES, Duration::from_millis(0), + StoreFilter::All, None, )?; @@ -637,6 +678,7 @@ fn sqlite_only_warning_detects_configured_app_alias() -> Result<()> { RolloutScope::None, DEFAULT_BUCKET_PADDING_BYTES, Duration::from_millis(50), + StoreFilter::All, None, )?; @@ -724,6 +766,7 @@ fn reconcile_all_stores_skips_rollout_when_store_backup_fails() -> Result<()> { RolloutScope::AllRows, DEFAULT_BUCKET_PADDING_BYTES, Duration::from_millis(0), + StoreFilter::All, None, ); fs::set_permissions(app_dir, fs::Permissions::from_mode(original_mode))?; @@ -777,6 +820,7 @@ fn reconcile_all_stores_fails_store_when_rollout_targets_unreadable() -> Result< RolloutScope::AllRows, DEFAULT_BUCKET_PADDING_BYTES, Duration::from_millis(0), + StoreFilter::All, None, )?; @@ -849,6 +893,7 @@ fn reconcile_all_stores_skips_busy_store_in_sqlite_only() -> Result<()> { RolloutScope::None, DEFAULT_BUCKET_PADDING_BYTES, Duration::from_millis(50), + StoreFilter::All, None, )?; @@ -893,6 +938,7 @@ fn reconcile_all_stores_treats_locked_backfill_status_as_busy() -> Result<()> { RolloutScope::None, DEFAULT_BUCKET_PADDING_BYTES, Duration::from_millis(0), + StoreFilter::All, None, )?; let elapsed = started.elapsed(); @@ -942,6 +988,7 @@ fn reconcile_all_stores_skips_whole_round_when_backfill_running() -> Result<()> RolloutScope::AllRows, DEFAULT_BUCKET_PADDING_BYTES, Duration::from_millis(50), + StoreFilter::All, None, )?; @@ -975,6 +1022,7 @@ fn reconcile_all_stores_treats_complete_backfill_as_ready() -> Result<()> { RolloutScope::AllRows, DEFAULT_BUCKET_PADDING_BYTES, Duration::from_millis(50), + StoreFilter::All, None, )?; @@ -1017,6 +1065,7 @@ fn reconcile_all_stores_writes_no_journal_without_backup() -> Result<()> { None, RolloutScope::AllRows, Duration::from_millis(0), + StoreFilter::All, None, )?; @@ -1037,6 +1086,171 @@ fn reconcile_all_stores_writes_no_journal_without_backup() -> Result<()> { Ok(()) } +#[test] +fn store_filter_limits_reconcile_to_selected_surface() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + let cli_db = codex_home.join("state_5.sqlite"); + seed_store_with_provider(&cli_db, "cong")?; + let app_db = codex_home.join("sqlite").join("state_5.sqlite"); + fs::create_dir_all(app_db.parent().unwrap())?; + seed_store_with_provider(&app_db, "cong")?; + + // `--store cli` reconciles only the CLI surface; the App store is untouched. + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::None, + DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(0), + StoreFilter::Cli, + None, + )?; + + assert_eq!(summary.stores.len(), 1); + assert_eq!(summary.stores[0].kind, StoreKind::Cli); + assert_eq!(provider_of(&cli_db)?, "openai"); + assert_eq!(provider_of(&app_db)?, "cong"); + Ok(()) +} + +#[test] +fn store_filter_reconciles_cli_when_configured_aliases_cli_default() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + fs::write(codex_home.join("config.toml"), "sqlite_home = \".\"\n")?; + let cli_db = codex_home.join("state_5.sqlite"); + seed_store_with_provider(&cli_db, "cong")?; + + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::None, + DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(0), + StoreFilter::Cli, + None, + )?; + + assert_eq!(summary.stores.len(), 1); + assert_eq!(summary.stores[0].kind, StoreKind::Cli); + assert_eq!(provider_of(&cli_db)?, "openai"); + Ok(()) +} + +#[test] +fn store_filter_reconciles_app_when_configured_aliases_app_default() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + fs::write(codex_home.join("config.toml"), "sqlite_home = \"sqlite\"\n")?; + let app_db = codex_home.join("sqlite").join("state_5.sqlite"); + fs::create_dir_all(app_db.parent().unwrap())?; + seed_store_with_provider(&app_db, "cong")?; + + let summary = reconcile_all_stores( + codex_home, + Some("openai"), + None, + RolloutScope::None, + Duration::from_millis(0), + StoreFilter::App, + None, + )?; + + assert_eq!(summary.stores.len(), 1); + assert_eq!(summary.stores[0].kind, StoreKind::App); + assert_eq!(provider_of(&app_db)?, "openai"); + Ok(()) +} + +#[test] +fn store_filter_rejects_rollout_writing_scope() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + + let rollout_path = codex_home.join("sessions/2026/05/07/rollout-1.jsonl"); + fs::create_dir_all(rollout_path.parent().unwrap())?; + fs::write( + &rollout_path, + "{\"type\":\"session_meta\",\"payload\":{\"id\":\"1\",\"model_provider\":\"cong\"}} \n", + )?; + + let cli_db = codex_home.join("state_5.sqlite"); + seed_store_with_provider(&cli_db, "cong")?; + { + let connection = Connection::open(&cli_db)?; + connection.execute( + "UPDATE threads SET rollout_path = ?1 WHERE id = '1'", + [rollout_path.display().to_string()], + )?; + } + + let app_db = codex_home.join("sqlite").join("state_5.sqlite"); + fs::create_dir_all(app_db.parent().unwrap())?; + seed_store_with_provider(&app_db, "cong")?; + + let err = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::AllRows, + DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(50), + StoreFilter::Cli, + None, + ) + .unwrap_err(); + + assert!(err.to_string().contains("--sqlite-only")); + assert_eq!(provider_of(&cli_db)?, "cong"); + assert_eq!(provider_of(&app_db)?, "cong"); + let rollout = fs::read_to_string(&rollout_path)?; + assert!(rollout.contains("\"model_provider\":\"cong\"")); + assert!(!rollout.contains("openai")); + Ok(()) +} + +#[test] +fn rejects_store_filter_for_commands_that_do_not_select_stores() { + let err = validate_store_filter_supported(Locale::En, StoreFilter::App, "install-service") + .unwrap_err(); + assert!(err.to_string().contains("--store is not supported")); + assert!(err.to_string().contains("install-service")); + assert!(!err.to_string().contains("bucket switch")); + + validate_store_filter_supported(Locale::En, StoreFilter::All, "install-service").unwrap(); +} + +#[test] +fn rejects_store_filter_for_rollout_writing_commands() { + let err = validate_store_filter_rollout_scope(Locale::En, StoreFilter::Cli, false, "sync") + .unwrap_err(); + assert!(err.to_string().contains("--sqlite-only")); + assert!(err.to_string().contains("sync")); + + validate_store_filter_rollout_scope(Locale::En, StoreFilter::Cli, true, "sync").unwrap(); + validate_store_filter_rollout_scope(Locale::En, StoreFilter::All, false, "sync").unwrap(); +} + +#[test] +fn status_reports_when_store_filter_matches_no_detected_store() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + let cli_db = codex_home.join("state_5.sqlite"); + seed_store_with_provider(&cli_db, "openai")?; + + let err = collect_status(codex_home, Some("openai"), None, StoreFilter::App).unwrap_err(); + + assert!(err.to_string().contains("--store app")); + assert!(!err.to_string().contains("run Codex at least once")); + Ok(()) +} + #[test] fn rejects_blank_provider_override() { let err = validate_provider_override(Locale::En, Some(" ")).unwrap_err(); @@ -1093,6 +1307,7 @@ fn reconcile_all_stores_reports_missing_configured_store_as_failed() -> Result<( None, RolloutScope::None, Duration::from_millis(0), + StoreFilter::All, None, )?; @@ -1397,10 +1612,7 @@ fn durable_sync_skips_longer_provider_without_padding() -> Result<()> { fn mismatched_scope_followup_repairs_matching_sqlite_rollout_after_sqlite_changes() -> Result<()> { let dir = tempfile::tempdir()?; let codex_home = dir.path(); - fs::write( - codex_home.join("config.toml"), - sqlite_home_config(codex_home), - )?; + isolate_process_sqlite_home(codex_home)?; let sqlite_path = codex_home.join("state_5.sqlite"); let rollout_a = codex_home.join("sessions/2026/05/07/rollout-a.jsonl"); let rollout_b = codex_home.join("sessions/2026/05/07/rollout-b.jsonl"); @@ -1435,6 +1647,7 @@ fn mismatched_scope_followup_repairs_matching_sqlite_rollout_after_sqlite_change None, RolloutScope::MismatchedRows, Duration::from_millis(0), + StoreFilter::All, None, )?; @@ -1493,6 +1706,7 @@ fn mismatched_scope_followup_excludes_failed_stores() -> Result<()> { None, RolloutScope::MismatchedRows, Duration::from_millis(0), + StoreFilter::All, None, )?; diff --git a/src/watch.rs b/src/watch.rs index 1960f15..db506df 100644 --- a/src/watch.rs +++ b/src/watch.rs @@ -30,6 +30,7 @@ use crate::output::watch_stopped_message; use crate::output::watcher_disconnected_error; use crate::output::watcher_error_message; use crate::rollout::RolloutScope; +use crate::stores::StoreFilter; use crate::sync::MultiReconcileSummary; use crate::sync::ReconcileStatus; use crate::sync::reconcile_all_stores; @@ -41,6 +42,7 @@ pub(crate) fn run_watch( codex_home: &Path, provider_override: Option, profile_override: Option, + store_filter: StoreFilter, rollout_scope: RolloutScope, poll_interval: Duration, ) -> Result<()> { @@ -89,6 +91,7 @@ pub(crate) fn run_watch( profile_override.as_deref(), full_rollout_scope, Duration::ZERO, + store_filter, None, ) { Ok(summary) => { @@ -118,6 +121,7 @@ pub(crate) fn run_watch( profile_override.as_deref(), full_rollout_scope, Duration::ZERO, + store_filter, None, ) { Ok(summary) => { @@ -150,6 +154,7 @@ pub(crate) fn run_watch( profile_override.as_deref(), periodic_watch_rollout_scope(rollout_scope, poll_count), Duration::ZERO, + store_filter, None, ) { Ok(summary) => {