diff --git a/README.md b/README.md index a13344d..a786f95 100644 --- a/README.md +++ b/README.md @@ -103,7 +103,7 @@ codex-threadripper install-service - `status`：查看当前 provider、SQLite 里各 provider 的线程分布，以及后台服务的运行状态 - `sync`：立即执行一次收敛。执行前会在 SQLite 状态库旁边的 `backups/` 目录里写一份带时间戳的备份；如果 rollout 首行需要改写，也会写 compact journal -- `bucket switch `：把全部历史切到指定 provider 可见桶；首行有 padding 时原地 patch，旧的冷 rollout 需要扩容时会安全重写并补 padding +- `bucket switch `：把全部历史切到指定 provider 可见桶；首行有 padding 时原地 patch，Legacy 模式下旧的冷 rollout 需要扩容时会安全重写并补 padding；分页历史需要扩容时会保持 rollout 和对应 SQLite 行不变并返回部分成功，避免破坏 Codex 的历史偏移 - `bucket prepare`：检查 rollout 首行 padding；需要扩容首行的文件会被跳过，避免替换仍可能被 Codex 写入的日志文件 - `watch`：持续监听 `config.toml` 的变化，同时定时收敛新增的线程记录 - `print-service-config`：打印当前平台的后台服务配置内容（launchd plist / systemd unit / Windows 隐藏 VBS 启动器） @@ -123,8 +123,8 @@ codex-threadripper install-service ## 它会改什么 - `sync` 会在 SQLite 状态库旁边的 `backups/` 目录里创建备份，然后更新 `state_5.sqlite` 里的 `model_provider` 字段 -- `sync` / `bucket switch` 也会改写 rollout JSONL 首行里的 provider 可见桶，并写入 compact journal；有足够 padding 时会原地 patch，旧的冷 rollout 需要扩容时会通过临时文件安全重写并补 padding；刚写过、仍可能被 Codex 写入的 rollout 会被跳过，待稍后重跑。改写后会恢复原本的文件访问时间和修改时间，避免 Codex.app 的最近会话排序被工具污染 -- `watch` 在运行期间会持续处理新写入的线程记录，保持数据库与 rollout 首行里的 provider 可见桶和当前 provider 对齐 +- `sync` / `bucket switch` 也会改写 rollout JSONL 首行里的 provider 可见桶，并写入 compact journal；有足够 padding 时会原地 patch，Legacy 模式下旧的冷 rollout 需要扩容时会通过临时文件安全重写并补 padding；刚写过、仍可能被 Codex 写入的 rollout 会被跳过，待稍后重跑。最新版 Codex 的分页历史会保存 rollout 字节偏移，因此需要增长首行的分页 rollout 会连同对应 SQLite provider 行一起保持不变，并以部分成功退出，避免留下失效的历史投影。改写后会恢复原本的文件访问时间和修改时间，避免 Codex.app 的最近会话排序被工具污染 +- `watch` 在运行期间会持续处理新写入的线程记录，保持数据库与 rollout 首行里的 provider 可见桶和当前 provider 对齐；稳定受阻的分页 rollout 会被缓存，避免每 500ms 重读和重复刷日志，并在配置变化或每 60 秒全量扫描时重新检查 - `--sqlite-only` 只更新 SQLite，适合只关心数据库索引的场景；当前 Codex.app 也会读取 rollout JSONL 元数据，所以日常可见性修复不建议使用它 - 默认状态库路径是 `CODEX_HOME/state_5.sqlite`；如果 Codex 配置了 `sqlite_home` 或环境变量 `CODEX_SQLITE_HOME`，`codex-threadripper` 会跟随 Codex 使用对应目录下的 `state_5.sqlite` @@ -220,7 +220,7 @@ codex-threadripper install-service - `status`: show the active provider, the per-provider thread counts from SQLite, and whether the background service is running - `sync`: reconcile once right now, writing a timestamped backup to the `backups/` directory beside the SQLite state DB; rollout first-line changes also get a compact journal -- `bucket switch `: move all history into the requested provider visibility bucket; prepared first lines are patched in place, while older cold rollouts that need first-line growth are safely rewritten with padding +- `bucket switch `: move history into the requested provider visibility bucket; prepared first lines are patched in place, while older cold legacy rollouts that need first-line growth are safely rewritten with padding. Paginated rollouts that need growth are left unchanged together with their SQLite rows, and the command reports a partial result so Codex history offsets stay valid - `bucket prepare`: inspect rollout first-line padding; files that would require first-line growth are skipped so the tool does not replace logs Codex may still be writing - `watch`: keep watching `config.toml` for provider changes and periodically reconcile any newly added thread rows - `print-service-config`: print the platform-specific service config (launchd plist, systemd unit, or Windows hidden VBS launcher) without installing it @@ -240,8 +240,8 @@ Every command accepts `--provider ` to force the target bucket, or `-- ## What it changes - `sync` writes a backup next to the SQLite state DB and then updates the `model_provider` column in `state_5.sqlite` -- `sync` / `bucket switch` also rewrite the provider visibility bucket in rollout JSONL first lines and write a compact journal. Prepared first lines are patched in place; older cold rollouts that need first-line growth are safely rewritten through a temporary file and padded for future switches; recently written rollouts that Codex may still be writing are skipped and can be retried later. After each rewrite, the original file access and modification times are restored so Codex.app's recent-thread ordering is not polluted by the tool -- `watch` keeps both SQLite and rollout first-line provider buckets aligned with the active provider as new threads are written +- `sync` / `bucket switch` also rewrite the provider visibility bucket in rollout JSONL first lines and write a compact journal. Prepared first lines are patched in place; older cold legacy rollouts that need first-line growth are safely rewritten through a temporary file and padded for future switches; recently written rollouts that Codex may still be writing are skipped and can be retried later. Current Codex paginated history persists rollout byte offsets, so a paginated rollout that would need first-line growth is left unchanged together with its SQLite provider row and the command exits partially instead of invalidating the history projection. After each rewrite, the original file access and modification times are restored so Codex.app's recent-thread ordering is not polluted by the tool +- `watch` keeps both SQLite and rollout first-line provider buckets aligned with the active provider as new threads are written. Stable blocked paginated rollouts are cached instead of being reread and logged every 500 ms, then reevaluated after config changes or during the 60-second full scan - `--sqlite-only` updates SQLite only. Use it only when you deliberately do not need rollout JSONL metadata; current Codex.app builds also read rollout metadata for visibility - The default state DB is `CODEX_HOME/state_5.sqlite`; if Codex uses `sqlite_home` or `CODEX_SQLITE_HOME`, `codex-threadripper` follows that directory and uses its `state_5.sqlite` diff --git a/src/output.rs b/src/output.rs index 1305014..6611067 100644 --- a/src/output.rs +++ b/src/output.rs @@ -138,7 +138,8 @@ pub(crate) fn print_multi_sync_summary( status_target_provider_label(locale), summary.provider ); - if summary.checked_rollouts > 0 || summary.changed_rollouts > 0 { + if summary.checked_rollouts > 0 || summary.changed_rollouts > 0 || summary.blocked_rollouts > 0 + { println!( "{}: {}", sync_rollouts_checked_label(locale), @@ -163,6 +164,13 @@ pub(crate) fn print_multi_sync_summary( summary.skipped_rollouts ); } + if summary.blocked_rollouts > 0 { + println!( + "{}: {}", + sync_paginated_rollouts_blocked_label(locale), + summary.blocked_rollouts + ); + } } println!(); @@ -945,10 +953,10 @@ pub(crate) fn reconcile_status_line(locale: Locale, status: ReconcileStatus) -> (ReconcileStatus::Full, Locale::En) => "Result: all stores updated.".to_string(), (ReconcileStatus::Full, Locale::ZhHans) => "结果：所有库均已更新。".to_string(), (ReconcileStatus::Partial, Locale::En) => { - "Result: PARTIAL — some stores updated, at least one was skipped or failed (see above). Re-run after it is resolved.".to_string() + "Result: PARTIAL — at least one store or paginated rollout was left unchanged (see above). Re-run after it is resolved.".to_string() } (ReconcileStatus::Partial, Locale::ZhHans) => { - "结果：部分成功 —— 部分库已更新，至少一个被跳过或失败（见上）。解决后请重跑。".to_string() + "结果：部分成功 —— 至少一个存储面或分页 rollout 保持未变（见上）。解决后请重跑。".to_string() } (ReconcileStatus::Failed, Locale::En) => { "Result: FAILED — no store could be updated.".to_string() @@ -1040,6 +1048,13 @@ pub(crate) fn sync_rollouts_skipped_label(locale: Locale) -> &'static str { } } +pub(crate) fn sync_paginated_rollouts_blocked_label(locale: Locale) -> &'static str { + match locale { + Locale::En => "Paginated rollouts kept unchanged to preserve history offsets", + Locale::ZhHans => "为保护历史偏移而保持未变的分页 rollout", + } +} + pub(crate) fn sync_elapsed_label(locale: Locale) -> &'static str { match locale { Locale::En => "Elapsed", diff --git a/src/rollout.rs b/src/rollout.rs index 4ed73cd..ec65206 100644 --- a/src/rollout.rs +++ b/src/rollout.rs @@ -55,6 +55,10 @@ pub(crate) struct RolloutReconcileSummary { pub(crate) changed_files: u64, pub(crate) prepared_files: u64, pub(crate) skipped_files: u64, + /// Thread IDs whose paginated rollouts require a length-changing rewrite. + /// Callers must leave the matching SQLite rows untouched so the rollout and + /// provider index cannot diverge while Codex owns byte-offset checkpoints. + pub(crate) blocked_thread_ids: HashSet, pub(crate) journal_path: Option, } @@ -154,6 +158,7 @@ struct RolloutPatchOutcome { changed: bool, prepared: bool, skipped: bool, + block_sqlite: bool, } struct RolloutChangeJournal { @@ -217,6 +222,7 @@ pub(crate) fn reconcile_rollouts_for_stores( store_db_paths: &[PathBuf], provider: &str, scope: RolloutScope, + known_blocked_thread_ids: &HashSet, journal_path: Option<&Path>, padding_bytes: usize, progress: Option, @@ -225,13 +231,27 @@ pub(crate) fn reconcile_rollouts_for_stores( return Ok(MultiStoreRolloutOutcome::default()); } let (targets, failed_stores) = rollout_targets_for_store_paths(store_db_paths, provider, scope); - let summary = reconcile_rollout_metadata_files( + let mut cached_blocked_thread_ids = HashSet::new(); + let targets = targets + .into_iter() + .filter(|target| { + if known_blocked_thread_ids.contains(&target.thread_id) { + cached_blocked_thread_ids.insert(target.thread_id.clone()); + false + } else { + true + } + }) + .collect::>(); + let mut summary = reconcile_rollout_metadata_files( targets.as_slice(), provider, journal_path, padding_bytes, progress, )?; + summary.skipped_files += cached_blocked_thread_ids.len() as u64; + summary.blocked_thread_ids.extend(cached_blocked_thread_ids); Ok(MultiStoreRolloutOutcome { summary, failed_stores, @@ -333,6 +353,9 @@ fn reconcile_rollout_metadata_files( if outcome.skipped { summary.skipped_files += 1; } + if outcome.block_sqlite { + summary.blocked_thread_ids.insert(target.thread_id.clone()); + } if let Some(progress) = progress.as_mut() { progress.tick(&summary); } @@ -365,6 +388,7 @@ fn rewrite_rollout_provider_first_line( }); }; let old_provider = session_meta_provider(&value).unwrap_or("").to_string(); + let paginated_history = session_meta_history_is_paginated(&value); if !set_session_meta_provider(&mut value, provider) { return Ok(RolloutPatchOutcome::default()); @@ -396,6 +420,19 @@ fn rewrite_rollout_provider_first_line( }); } + // Codex paginated history persists a byte offset into the rollout. Growing + // the first line would shift every later record while leaving that + // checkpoint stale. An in-place replacement above is safe because it keeps + // the exact byte length; a growing rewrite must be left untouched, along + // with the matching SQLite provider row. + if paginated_history { + return Ok(RolloutPatchOutcome { + skipped: true, + block_sqlite: true, + ..RolloutPatchOutcome::default() + }); + } + if !rollout_is_old_enough(&original_metadata, ROLLOUT_GROWTH_REWRITE_MIN_AGE) { return Ok(RolloutPatchOutcome { skipped: true, @@ -777,6 +814,14 @@ fn session_meta_provider(value: &Value) -> Option<&str> { .and_then(Value::as_str) } +fn session_meta_history_is_paginated(value: &Value) -> bool { + value + .get("payload") + .and_then(|payload| payload.get("history_mode")) + .and_then(Value::as_str) + == Some("paginated") +} + fn set_session_meta_provider(value: &mut Value, provider: &str) -> bool { let Some(payload) = value.get_mut("payload").and_then(Value::as_object_mut) else { return false; diff --git a/src/state_db.rs b/src/state_db.rs index dcd983a..6b4b0f2 100644 --- a/src/state_db.rs +++ b/src/state_db.rs @@ -6,6 +6,7 @@ use rusqlite::OptionalExtension; use rusqlite::TransactionBehavior; use rusqlite::backup::Backup; use rusqlite::backup::StepResult; +use std::collections::HashSet; use std::fs; use std::path::Path; use std::path::PathBuf; @@ -88,7 +89,16 @@ pub(crate) fn read_backfill_status_with_timeout( Ok(status) } +#[cfg(test)] pub(crate) fn reconcile_sqlite_in_place(sqlite_path: &Path, provider: &str) -> Result<(u64, u64)> { + reconcile_sqlite_in_place_excluding(sqlite_path, provider, &HashSet::new()) +} + +pub(crate) fn reconcile_sqlite_in_place_excluding( + sqlite_path: &Path, + provider: &str, + excluded_thread_ids: &HashSet, +) -> Result<(u64, u64)> { ensure_sqlite_exists(sqlite_path)?; let mut connection = Connection::open(sqlite_path) .with_context(|| format!("failed to open {}", sqlite_path.display()))?; @@ -96,10 +106,37 @@ pub(crate) fn reconcile_sqlite_in_place(sqlite_path: &Path, provider: &str) -> R let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let total_rows: u64 = transaction.query_row("SELECT COUNT(*) FROM threads", [], |row| row.get(0))?; - let changed_rows = transaction.execute( - "UPDATE threads SET model_provider = ?1 WHERE model_provider <> ?1", - [provider], - )? as u64; + let changed_rows = if excluded_thread_ids.is_empty() { + transaction.execute( + "UPDATE threads SET model_provider = ?1 WHERE model_provider <> ?1", + [provider], + )? as u64 + } else { + transaction.execute_batch( + "CREATE TEMP TABLE IF NOT EXISTS threadripper_excluded_threads ( + id TEXT PRIMARY KEY + ) WITHOUT ROWID; + DELETE FROM temp.threadripper_excluded_threads;", + )?; + { + let mut insert = transaction + .prepare("INSERT INTO temp.threadripper_excluded_threads (id) VALUES (?1)")?; + for thread_id in excluded_thread_ids { + insert.execute([thread_id])?; + } + } + transaction.execute( + "UPDATE threads + SET model_provider = ?1 + WHERE model_provider <> ?1 + AND NOT EXISTS ( + SELECT 1 + FROM temp.threadripper_excluded_threads AS excluded + WHERE excluded.id = threads.id + )", + [provider], + )? as u64 + }; transaction.commit()?; Ok((changed_rows, total_rows)) diff --git a/src/sync.rs b/src/sync.rs index c39eda8..a492fa2 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -23,7 +23,7 @@ use crate::state_db::create_sqlite_backup_file_in; use crate::state_db::inspect_sqlite_distribution; 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::reconcile_sqlite_in_place_excluding; use crate::state_db::unix_timestamp_millis; use crate::stores::StoreFilter; use crate::stores::StoreKind; @@ -175,6 +175,8 @@ pub(crate) struct MultiReconcileSummary { pub(crate) checked_rollouts: u64, pub(crate) prepared_rollouts: u64, pub(crate) skipped_rollouts: u64, + pub(crate) blocked_rollouts: u64, + pub(crate) blocked_thread_ids: HashSet, pub(crate) rollout_journal_path: Option, pub(crate) elapsed: Duration, } @@ -186,10 +188,10 @@ impl MultiReconcileSummary { .iter() .filter(|store| matches!(store.outcome, StoreOutcome::Updated { .. })) .count(); - if updated == self.stores.len() { - ReconcileStatus::Full - } else if updated == 0 { + if updated == 0 { ReconcileStatus::Failed + } else if updated == self.stores.len() && self.blocked_rollouts == 0 { + ReconcileStatus::Full } else { ReconcileStatus::Partial } @@ -282,6 +284,7 @@ pub(crate) fn reconcile_all_stores_with_backup( provider_override, profile_override, rollout_scope, + &HashSet::new(), padding_bytes, backfill_wait, true, @@ -305,6 +308,29 @@ pub(crate) fn reconcile_all_stores( backfill_wait: Duration, filter: StoreFilter, progress: Option, +) -> Result { + reconcile_all_stores_with_known_blocked( + codex_home, + provider_override, + profile_override, + rollout_scope, + backfill_wait, + filter, + progress, + &HashSet::new(), + ) +} + +#[allow(clippy::too_many_arguments)] +pub(crate) fn reconcile_all_stores_with_known_blocked( + codex_home: &Path, + provider_override: Option<&str>, + profile_override: Option<&str>, + rollout_scope: RolloutScope, + backfill_wait: Duration, + filter: StoreFilter, + progress: Option, + known_blocked_thread_ids: &HashSet, ) -> Result { with_threadripper_lock(codex_home, || { reconcile_stores_core( @@ -312,6 +338,7 @@ pub(crate) fn reconcile_all_stores( provider_override, profile_override, rollout_scope, + known_blocked_thread_ids, DEFAULT_BUCKET_PADDING_BYTES, backfill_wait, false, @@ -327,6 +354,7 @@ fn reconcile_stores_core( provider_override: Option<&str>, profile_override: Option<&str>, rollout_scope: RolloutScope, + known_blocked_thread_ids: &HashSet, padding_bytes: usize, backfill_wait: Duration, backup: bool, @@ -406,6 +434,8 @@ fn reconcile_stores_core( checked_rollouts: 0, prepared_rollouts: 0, skipped_rollouts: 0, + blocked_rollouts: 0, + blocked_thread_ids: HashSet::new(), rollout_journal_path: None, elapsed: started.elapsed(), }); @@ -463,6 +493,8 @@ fn reconcile_stores_core( checked_rollouts: 0, prepared_rollouts: 0, skipped_rollouts: 0, + blocked_rollouts: 0, + blocked_thread_ids: HashSet::new(), rollout_journal_path: None, elapsed: started.elapsed(), }); @@ -493,6 +525,7 @@ fn reconcile_stores_core( ready_db_paths.as_slice(), provider.as_str(), rollout_scope, + known_blocked_thread_ids, rollout_journal_path.as_deref(), padding_bytes, progress, @@ -540,7 +573,12 @@ fn reconcile_stores_core( } else { None }; - reconcile_single_store(target, provider.as_str(), backup_path) + reconcile_single_store( + target, + provider.as_str(), + backup_path, + &rollout_summary.blocked_thread_ids, + ) } }) .collect(); @@ -560,6 +598,7 @@ fn reconcile_stores_core( followup_db_paths.as_slice(), provider.as_str(), RolloutScope::AllRows, + &rollout_summary.blocked_thread_ids, None, padding_bytes, None, @@ -576,6 +615,9 @@ fn reconcile_stores_core( rollout_summary.changed_files += followup.summary.changed_files; rollout_summary.prepared_files += followup.summary.prepared_files; rollout_summary.skipped_files += followup.summary.skipped_files; + rollout_summary + .blocked_thread_ids + .extend(followup.summary.blocked_thread_ids); } Ok(MultiReconcileSummary { @@ -585,6 +627,8 @@ fn reconcile_stores_core( checked_rollouts: rollout_summary.checked_files, prepared_rollouts: rollout_summary.prepared_files, skipped_rollouts: rollout_summary.skipped_files, + blocked_rollouts: rollout_summary.blocked_thread_ids.len() as u64, + blocked_thread_ids: rollout_summary.blocked_thread_ids, rollout_journal_path: rollout_summary.journal_path, elapsed: started.elapsed(), }) @@ -615,17 +659,19 @@ fn reconcile_single_store( target: &StoreTarget, provider: &str, backup_path: Option, + excluded_thread_ids: &HashSet, ) -> StoreReconcileResult { - let outcome = match reconcile_single_store_inner(target, provider, backup_path) { - Ok((changed_rows, total_rows, backup_path)) => StoreOutcome::Updated { - changed_rows, - total_rows, - backup_path, - }, - Err(error) => StoreOutcome::Failed { - error: error.to_string(), - }, - }; + let outcome = + match reconcile_single_store_inner(target, provider, backup_path, excluded_thread_ids) { + Ok((changed_rows, total_rows, backup_path)) => StoreOutcome::Updated { + changed_rows, + total_rows, + backup_path, + }, + Err(error) => StoreOutcome::Failed { + error: error.to_string(), + }, + }; StoreReconcileResult { kind: target.kind, db_path: target.db_path.clone(), @@ -637,7 +683,9 @@ fn reconcile_single_store_inner( target: &StoreTarget, provider: &str, backup_path: Option, + excluded_thread_ids: &HashSet, ) -> Result<(u64, u64, Option)> { - let (changed_rows, total_rows) = reconcile_sqlite_in_place(&target.db_path, provider)?; + let (changed_rows, total_rows) = + reconcile_sqlite_in_place_excluding(&target.db_path, provider, excluded_thread_ids)?; Ok((changed_rows, total_rows, backup_path)) } diff --git a/src/tests.rs b/src/tests.rs index eb84517..9273ede 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -22,6 +22,7 @@ use crate::rollout::reconcile_rollout_metadata_from_sqlite_with_progress; 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_in_place_excluding; use crate::state_db::reconcile_sqlite_with_backup; use crate::stores::StoreFilter; use crate::stores::StoreKind; @@ -34,6 +35,7 @@ use crate::sync::StoreReconcileResult; use crate::sync::collect_status; use crate::sync::reconcile_all_stores; use crate::sync::reconcile_all_stores_with_backup; +use crate::sync::reconcile_all_stores_with_known_blocked; use crate::watch::WATCH_FULL_ROLLOUT_POLL_INTERVALS; use crate::watch::full_watch_rollout_scope; use crate::watch::periodic_watch_rollout_scope; @@ -45,6 +47,7 @@ use clap::error::ErrorKind; use filetime::FileTime; use filetime::set_file_times; use rusqlite::Connection; +use std::collections::HashSet; use std::fs; use std::path::Path; use std::path::PathBuf; @@ -1541,7 +1544,7 @@ fn durable_sync_patches_shorter_provider_in_place() -> Result<()> { fs::write( &rollout_path, concat!( - "{\"type\":\"session_meta\",\"payload\":{\"id\":\"1\",\"model_provider\":\"openai\"}} \n", + "{\"type\":\"session_meta\",\"payload\":{\"id\":\"1\",\"model_provider\":\"openai\",\"history_mode\":\"paginated\"}} \n", "{\"type\":\"event_msg\",\"payload\":{\"type\":\"user_message\",\"message\":\"hi\"}}\n", ), )?; @@ -1682,6 +1685,151 @@ fn durable_sync_rewrites_cold_longer_provider_with_padding() -> Result<()> { Ok(()) } +#[test] +fn paginated_growth_blocks_only_matching_rollout_and_sqlite_row() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + let sqlite_path = codex_home.join("state_5.sqlite"); + let rollout_path = codex_home.join("sessions/2026/07/21/rollout-paginated.jsonl"); + let legacy_rollout_path = codex_home.join("sessions/2026/07/21/rollout-legacy.jsonl"); + fs::create_dir_all(rollout_path.parent().unwrap())?; + fs::write( + &rollout_path, + concat!( + "{\"type\":\"session_meta\",\"payload\":{\"id\":\"1\",\"model_provider\":\"vm\",\"history_mode\":\"paginated\"}}\n", + "{\"timestamp\":\"2026-07-21T00:00:00Z\",\"type\":\"event_msg\",\"payload\":{\"type\":\"user_message\",\"message\":\"hi\"},\"ordinal\":0}\n", + ), + )?; + fs::write( + &legacy_rollout_path, + "{\"type\":\"session_meta\",\"payload\":{\"id\":\"2\",\"model_provider\":\"vm\",\"history_mode\":\"legacy\"}} \n", + )?; + let original = fs::read(&rollout_path)?; + let original_mtime = FileTime::from_unix_time(1_700_000_300, 0); + set_file_times(&rollout_path, original_mtime, original_mtime)?; + + seed_sqlite(&sqlite_path)?; + let connection = Connection::open(&sqlite_path)?; + connection.execute( + "UPDATE threads SET model_provider = 'openai', rollout_path = ''", + [], + )?; + connection.execute( + "UPDATE threads SET rollout_path = ?1, model_provider = 'vm' WHERE id = '1'", + [rollout_path.display().to_string()], + )?; + connection.execute( + "UPDATE threads SET rollout_path = ?1, model_provider = 'vm' WHERE id = '2'", + [legacy_rollout_path.display().to_string()], + )?; + drop(connection); + + let summary = reconcile_all_stores( + codex_home, + Some("openai"), + None, + RolloutScope::AllRows, + Duration::ZERO, + StoreFilter::All, + None, + )?; + + assert_eq!(summary.status(), ReconcileStatus::Partial); + assert_eq!(summary.changed_rollouts, 1); + assert_eq!(summary.skipped_rollouts, 1); + assert_eq!(summary.blocked_rollouts, 1); + assert_eq!(summary.total_changed_rows(), 1); + assert_eq!(fs::read(&rollout_path)?, original); + let metadata = fs::metadata(&rollout_path)?; + let actual_mtime = FileTime::from_last_modification_time(&metadata); + assert_eq!(actual_mtime.unix_seconds(), original_mtime.unix_seconds()); + let connection = Connection::open(&sqlite_path)?; + let blocked_provider: String = connection.query_row( + "SELECT model_provider FROM threads WHERE id = '1'", + [], + |row| row.get(0), + )?; + let updated_provider: String = connection.query_row( + "SELECT model_provider FROM threads WHERE id = '2'", + [], + |row| row.get(0), + )?; + assert_eq!(blocked_provider, "vm"); + assert_eq!(updated_provider, "openai"); + assert!(fs::read_to_string(&legacy_rollout_path)?.contains("\"model_provider\":\"openai\"")); + + let known_blocked_thread_ids = summary.blocked_thread_ids.clone(); + fs::write( + &rollout_path, + "cached blocked rollout must not be reparsed\n", + )?; + let cached_summary = reconcile_all_stores_with_known_blocked( + codex_home, + Some("openai"), + None, + RolloutScope::MismatchedRows, + Duration::ZERO, + StoreFilter::All, + None, + &known_blocked_thread_ids, + )?; + assert_eq!(cached_summary.status(), ReconcileStatus::Partial); + assert_eq!(cached_summary.changed_rollouts, 0); + assert_eq!(cached_summary.blocked_thread_ids, known_blocked_thread_ids); + assert_eq!(cached_summary.total_changed_rows(), 0); + Ok(()) +} + +#[test] +fn sqlite_exclusions_scale_past_the_variable_limit() -> Result<()> { + const BLOCKED_ROWS: usize = 32_767; + + let dir = tempfile::tempdir()?; + let sqlite_path = dir.path().join("state_5.sqlite"); + let mut connection = Connection::open(&sqlite_path)?; + connection.execute_batch( + "CREATE TABLE threads ( + id TEXT PRIMARY KEY, + model_provider TEXT NOT NULL + );", + )?; + let transaction = connection.transaction()?; + let mut blocked_thread_ids = HashSet::with_capacity(BLOCKED_ROWS); + { + let mut insert = + transaction.prepare("INSERT INTO threads (id, model_provider) VALUES (?1, 'vm')")?; + for index in 0..BLOCKED_ROWS { + let thread_id = format!("blocked-{index}"); + insert.execute([thread_id.as_str()])?; + blocked_thread_ids.insert(thread_id); + } + insert.execute(["allowed"])?; + } + transaction.commit()?; + drop(connection); + + let (changed_rows, total_rows) = + reconcile_sqlite_in_place_excluding(&sqlite_path, "openai", &blocked_thread_ids)?; + + assert_eq!(changed_rows, 1); + assert_eq!(total_rows, BLOCKED_ROWS as u64 + 1); + let connection = Connection::open(&sqlite_path)?; + let blocked_openai: u64 = connection.query_row( + "SELECT COUNT(*) FROM threads WHERE id LIKE 'blocked-%' AND model_provider = 'openai'", + [], + |row| row.get(0), + )?; + let allowed_provider: String = connection.query_row( + "SELECT model_provider FROM threads WHERE id = 'allowed'", + [], + |row| row.get(0), + )?; + assert_eq!(blocked_openai, 0); + assert_eq!(allowed_provider, "openai"); + Ok(()) +} + #[test] fn mismatched_scope_followup_repairs_matching_sqlite_rollout_after_sqlite_changes() -> Result<()> { let dir = tempfile::tempdir()?; @@ -1856,11 +2004,53 @@ fn watch_prints_non_full_summary_even_when_counts_are_unchanged() { checked_rollouts: 0, prepared_rollouts: 0, skipped_rollouts: 0, + blocked_rollouts: 0, + blocked_thread_ids: HashSet::new(), rollout_journal_path: None, elapsed: Duration::ZERO, }; - assert!(watch_should_print_summary(Some("openai"), &summary)); + assert!(watch_should_print_summary( + Some("openai"), + &HashSet::new(), + &summary + )); +} + +#[test] +fn watch_suppresses_an_unchanged_paginated_block_summary() { + let blocked_thread_ids = HashSet::from(["blocked".to_string()]); + let summary = MultiReconcileSummary { + provider: "openai".to_string(), + stores: vec![StoreReconcileResult { + kind: StoreKind::Cli, + db_path: PathBuf::from("/tmp/state_5.sqlite"), + outcome: StoreOutcome::Updated { + changed_rows: 0, + total_rows: 1, + backup_path: None, + }, + }], + changed_rollouts: 0, + checked_rollouts: 0, + prepared_rollouts: 0, + skipped_rollouts: 1, + blocked_rollouts: 1, + blocked_thread_ids: blocked_thread_ids.clone(), + rollout_journal_path: None, + elapsed: Duration::ZERO, + }; + + assert!(!watch_should_print_summary( + Some("openai"), + &blocked_thread_ids, + &summary + )); + assert!(watch_should_print_summary( + Some("openai"), + &HashSet::new(), + &summary + )); } #[test] diff --git a/src/watch.rs b/src/watch.rs index db506df..e2ec772 100644 --- a/src/watch.rs +++ b/src/watch.rs @@ -6,6 +6,7 @@ use notify::EventKind; use notify::RecommendedWatcher; use notify::RecursiveMode; use notify::Watcher; +use std::collections::HashSet; use std::fs::OpenOptions; use std::path::Path; use std::path::PathBuf; @@ -32,8 +33,8 @@ 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; +use crate::sync::reconcile_all_stores_with_known_blocked; pub(crate) const WATCH_FULL_ROLLOUT_POLL_INTERVALS: u64 = 120; @@ -85,6 +86,7 @@ pub(crate) fn run_watch( } let mut last_provider = None; + let mut known_blocked_thread_ids = HashSet::new(); match reconcile_all_stores( codex_home, provider_override.as_deref(), @@ -97,6 +99,7 @@ pub(crate) fn run_watch( Ok(summary) => { print_multi_sync_summary(locale, watch_started_title(locale), &summary); last_provider = Some(summary.provider.clone()); + known_blocked_thread_ids = summary.blocked_thread_ids.clone(); } Err(err) => { eprintln!("{}", watch_initial_reconcile_error_message(locale, &err)); @@ -125,7 +128,11 @@ pub(crate) fn run_watch( None, ) { Ok(summary) => { - if watch_should_print_summary(last_provider.as_deref(), &summary) { + if watch_should_print_summary( + last_provider.as_deref(), + &known_blocked_thread_ids, + &summary, + ) { print_multi_sync_summary( locale, config_change_title(locale), @@ -133,6 +140,7 @@ pub(crate) fn run_watch( ); } last_provider = Some(summary.provider.clone()); + known_blocked_thread_ids = summary.blocked_thread_ids.clone(); watched_paths = watched_config_paths(codex_home, profile_override.as_deref()); } @@ -148,17 +156,29 @@ pub(crate) fn run_watch( } Err(mpsc::RecvTimeoutError::Timeout) => { poll_count = poll_count.wrapping_add(1); - match reconcile_all_stores( + let periodic_scope = periodic_watch_rollout_scope(rollout_scope, poll_count); + let empty_blocked_thread_ids = HashSet::new(); + let blocked_thread_ids = if periodic_scope == RolloutScope::MismatchedRows { + &known_blocked_thread_ids + } else { + &empty_blocked_thread_ids + }; + match reconcile_all_stores_with_known_blocked( codex_home, provider_override.as_deref(), profile_override.as_deref(), - periodic_watch_rollout_scope(rollout_scope, poll_count), + periodic_scope, Duration::ZERO, store_filter, None, + blocked_thread_ids, ) { Ok(summary) => { - if watch_should_print_summary(last_provider.as_deref(), &summary) { + if watch_should_print_summary( + last_provider.as_deref(), + &known_blocked_thread_ids, + &summary, + ) { print_multi_sync_summary( locale, background_reconcile_title(locale), @@ -166,6 +186,7 @@ pub(crate) fn run_watch( ); } last_provider = Some(summary.provider.clone()); + known_blocked_thread_ids = summary.blocked_thread_ids.clone(); } Err(err) => { eprintln!("{}", watch_reconcile_skipped_message(locale, &err)); @@ -184,12 +205,18 @@ pub(crate) fn run_watch( pub(crate) fn watch_should_print_summary( last_provider: Option<&str>, + known_blocked_thread_ids: &HashSet, summary: &MultiReconcileSummary, ) -> bool { - summary.status() != ReconcileStatus::Full + let store_not_updated = summary + .stores + .iter() + .any(|store| !matches!(store.outcome, crate::sync::StoreOutcome::Updated { .. })); + store_not_updated || last_provider != Some(summary.provider.as_str()) || summary.total_changed_rows() > 0 || summary.changed_rollouts > 0 + || known_blocked_thread_ids != &summary.blocked_thread_ids } pub(crate) fn periodic_watch_rollout_scope(