From 0524ee7b43a930f1c01b5ee4e5c424a996629ea9 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:53:33 +0800 Subject: [PATCH 1/3] fix: preserve paginated history offsets --- README.md | 8 ++--- src/output.rs | 18 ++++++++++-- src/rollout.rs | 30 +++++++++++++++++++ src/state_db.rs | 40 ++++++++++++++++++++++--- src/sync.rs | 48 ++++++++++++++++++++---------- src/tests.rs | 77 ++++++++++++++++++++++++++++++++++++++++++++++++- 6 files changed, 194 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index a13344d..5e505e0 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,7 +123,7 @@ 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 的最近会话排序被工具污染 +- `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 对齐 - `--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,7 +240,7 @@ 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 +- `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 - `--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..db313c9 100644 --- a/src/output.rs +++ b/src/output.rs @@ -163,6 +163,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 +952,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 +1047,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..ac66cea 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 { @@ -333,6 +338,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 +373,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 +405,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 +799,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..5c6c5f3 100644 --- a/src/state_db.rs +++ b/src/state_db.rs @@ -6,6 +6,9 @@ use rusqlite::OptionalExtension; use rusqlite::TransactionBehavior; use rusqlite::backup::Backup; use rusqlite::backup::StepResult; +use rusqlite::params_from_iter; +use rusqlite::types::Value; +use std::collections::HashSet; use std::fs; use std::path::Path; use std::path::PathBuf; @@ -88,7 +91,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 +108,30 @@ 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 { + let mut excluded_thread_ids = excluded_thread_ids.iter().collect::>(); + excluded_thread_ids.sort_unstable(); + let placeholders = std::iter::repeat_n("?", excluded_thread_ids.len()) + .collect::>() + .join(", "); + let sql = format!( + "UPDATE threads SET model_provider = ? WHERE model_provider <> ? AND id NOT IN ({placeholders})" + ); + let mut values = Vec::with_capacity(excluded_thread_ids.len() + 2); + values.push(Value::Text(provider.to_string())); + values.push(Value::Text(provider.to_string())); + values.extend( + excluded_thread_ids + .into_iter() + .map(|thread_id| Value::Text(thread_id.clone())), + ); + transaction.execute(sql.as_str(), params_from_iter(values.iter()))? as u64 + }; transaction.commit()?; Ok((changed_rows, total_rows)) diff --git a/src/sync.rs b/src/sync.rs index c39eda8..ad8aa82 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,7 @@ 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) rollout_journal_path: Option, pub(crate) elapsed: Duration, } @@ -186,10 +187,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 } @@ -406,6 +407,7 @@ fn reconcile_stores_core( checked_rollouts: 0, prepared_rollouts: 0, skipped_rollouts: 0, + blocked_rollouts: 0, rollout_journal_path: None, elapsed: started.elapsed(), }); @@ -463,6 +465,7 @@ fn reconcile_stores_core( checked_rollouts: 0, prepared_rollouts: 0, skipped_rollouts: 0, + blocked_rollouts: 0, rollout_journal_path: None, elapsed: started.elapsed(), }); @@ -540,7 +543,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(); @@ -576,6 +584,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 +596,7 @@ 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, rollout_journal_path: rollout_summary.journal_path, elapsed: started.elapsed(), }) @@ -615,17 +627,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 +651,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..6ebf0b9 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -1541,7 +1541,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 +1682,80 @@ 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); + assert_rollout_times(&rollout_path, original_mtime)?; + 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\"")); + Ok(()) +} + #[test] fn mismatched_scope_followup_repairs_matching_sqlite_rollout_after_sqlite_changes() -> Result<()> { let dir = tempfile::tempdir()?; @@ -1856,6 +1930,7 @@ fn watch_prints_non_full_summary_even_when_counts_are_unchanged() { checked_rollouts: 0, prepared_rollouts: 0, skipped_rollouts: 0, + blocked_rollouts: 0, rollout_journal_path: None, elapsed: Duration::ZERO, }; From 80a7620c4d4a02d88ccf70431ed8e3146110aded Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Tue, 21 Jul 2026 12:06:22 +0800 Subject: [PATCH 2/3] fix: harden paginated watch retries --- README.md | 4 +- src/output.rs | 3 +- src/rollout.rs | 17 ++++++- src/state_db.rs | 43 ++++++++++-------- src/sync.rs | 32 ++++++++++++++ src/tests.rs | 115 +++++++++++++++++++++++++++++++++++++++++++++++- src/watch.rs | 39 +++++++++++++--- 7 files changed, 223 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index 5e505e0..a786f95 100644 --- a/README.md +++ b/README.md @@ -124,7 +124,7 @@ codex-threadripper install-service - `sync` 会在 SQLite 状态库旁边的 `backups/` 目录里创建备份，然后更新 `state_5.sqlite` 里的 `model_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 对齐 +- `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` @@ -241,7 +241,7 @@ Every command accepts `--provider ` to force the target bucket, or `-- - `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 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 +- `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 db313c9..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), diff --git a/src/rollout.rs b/src/rollout.rs index ac66cea..ec65206 100644 --- a/src/rollout.rs +++ b/src/rollout.rs @@ -222,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, @@ -230,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, diff --git a/src/state_db.rs b/src/state_db.rs index 5c6c5f3..6b4b0f2 100644 --- a/src/state_db.rs +++ b/src/state_db.rs @@ -6,8 +6,6 @@ use rusqlite::OptionalExtension; use rusqlite::TransactionBehavior; use rusqlite::backup::Backup; use rusqlite::backup::StepResult; -use rusqlite::params_from_iter; -use rusqlite::types::Value; use std::collections::HashSet; use std::fs; use std::path::Path; @@ -114,23 +112,30 @@ pub(crate) fn reconcile_sqlite_in_place_excluding( [provider], )? as u64 } else { - let mut excluded_thread_ids = excluded_thread_ids.iter().collect::>(); - excluded_thread_ids.sort_unstable(); - let placeholders = std::iter::repeat_n("?", excluded_thread_ids.len()) - .collect::>() - .join(", "); - let sql = format!( - "UPDATE threads SET model_provider = ? WHERE model_provider <> ? AND id NOT IN ({placeholders})" - ); - let mut values = Vec::with_capacity(excluded_thread_ids.len() + 2); - values.push(Value::Text(provider.to_string())); - values.push(Value::Text(provider.to_string())); - values.extend( - excluded_thread_ids - .into_iter() - .map(|thread_id| Value::Text(thread_id.clone())), - ); - transaction.execute(sql.as_str(), params_from_iter(values.iter()))? as u64 + 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()?; diff --git a/src/sync.rs b/src/sync.rs index ad8aa82..a492fa2 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -176,6 +176,7 @@ pub(crate) struct MultiReconcileSummary { 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, } @@ -283,6 +284,7 @@ pub(crate) fn reconcile_all_stores_with_backup( provider_override, profile_override, rollout_scope, + &HashSet::new(), padding_bytes, backfill_wait, true, @@ -306,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( @@ -313,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, @@ -328,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, @@ -408,6 +435,7 @@ fn reconcile_stores_core( prepared_rollouts: 0, skipped_rollouts: 0, blocked_rollouts: 0, + blocked_thread_ids: HashSet::new(), rollout_journal_path: None, elapsed: started.elapsed(), }); @@ -466,6 +494,7 @@ fn reconcile_stores_core( prepared_rollouts: 0, skipped_rollouts: 0, blocked_rollouts: 0, + blocked_thread_ids: HashSet::new(), rollout_journal_path: None, elapsed: started.elapsed(), }); @@ -496,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, @@ -568,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, @@ -597,6 +628,7 @@ fn reconcile_stores_core( 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(), }) diff --git a/src/tests.rs b/src/tests.rs index 6ebf0b9..518c2ae 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; @@ -1753,6 +1756,75 @@ fn paginated_growth_blocks_only_matching_rollout_and_sqlite_row() -> Result<()> 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(()) } @@ -1931,11 +2003,52 @@ fn watch_prints_non_full_summary_even_when_counts_are_unchanged() { 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"), + &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"), &summary)); + 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( From 939571f476a9e721eb978fec0458644b5a5cc620 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Tue, 21 Jul 2026 12:17:56 +0800 Subject: [PATCH 3/3] test: make blocked rollout timestamp assertion portable --- src/tests.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/tests.rs b/src/tests.rs index 518c2ae..9273ede 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -1741,7 +1741,9 @@ fn paginated_growth_blocks_only_matching_rollout_and_sqlite_row() -> Result<()> assert_eq!(summary.blocked_rollouts, 1); assert_eq!(summary.total_changed_rows(), 1); assert_eq!(fs::read(&rollout_path)?, original); - assert_rollout_times(&rollout_path, original_mtime)?; + 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'",