From 6e72dbbd17408ce2a4afcc3c7fafeb54e79518ff Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Sun, 24 May 2026 22:54:02 +0800 Subject: [PATCH 1/8] feat: resolve provider from Codex profiles --- README.md | 4 + src/main.rs | 348 ++++++++++++++++++++++++++++++++++++++++++++----- src/service.rs | 116 ++++++++++++++--- 3 files changed, 417 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index 02db215..594cbd2 100644 --- a/README.md +++ b/README.md @@ -118,6 +118,8 @@ codex-threadripper install-service `watch` 的默认轮询间隔是 500ms。`watch`、`print-service-config` 和 `install-service` 都支持 `--poll-interval-ms` 来自定义间隔。 +所有命令都支持 `--provider ` 强制指定目标桶，或 `--profile ` 按 Codex v2 profile 读取 `$CODEX_HOME/.config.toml` 里的 `model_provider` / `sqlite_home`。如果没有同名 profile 文件，也会兼容旧的 `[profiles.]` provider 配置。 + ## 它会改什么 - `sync` 会在 SQLite 状态库旁边的 `backups/` 目录里创建备份，然后更新 `state_5.sqlite` 里的 `model_provider` 字段 @@ -233,6 +235,8 @@ These legacy command names still work: `watch` polls every 500ms by default. You can change that with `--poll-interval-ms` on `watch`, `print-service-config`, and `install-service`. +Every command accepts `--provider ` to force the target bucket, or `--profile ` to resolve `model_provider` / `sqlite_home` from Codex's v2 `$CODEX_HOME/.config.toml` profile. If that profile file is missing, legacy `[profiles.]` provider config is still supported. + ## What it changes - `sync` writes a backup next to the SQLite state DB and then updates the `model_provider` column in `state_5.sqlite` diff --git a/src/main.rs b/src/main.rs index 697314c..562e785 100644 --- a/src/main.rs +++ b/src/main.rs @@ -21,6 +21,7 @@ use rusqlite::backup::StepResult; use serde::Deserialize; use serde_json::Value; use serde_json::json; +use std::collections::BTreeMap; use std::ffi::OsStr; use std::fs; use std::fs::File; @@ -48,6 +49,7 @@ const APP_VERSION: &str = env!("CARGO_PKG_VERSION"); const DEFAULT_PROVIDER: &str = "openai"; const CODEX_SQLITE_HOME_ENV: &str = "CODEX_SQLITE_HOME"; const STATE_DB_FILENAME: &str = "state_5.sqlite"; +const PROFILE_CONFIG_SUFFIX: &str = ".config.toml"; const DEFAULT_POLL_INTERVAL_MS: u64 = 500; const DEFAULT_BUCKET_PADDING_BYTES: usize = 256; const ROLLOUT_PROGRESS_INTERVAL: Duration = Duration::from_millis(500); @@ -83,6 +85,9 @@ struct Cli { #[arg(long, global = true, value_name = "PROVIDER", help = "placeholder")] provider: Option, + #[arg(long, global = true, value_name = "PROFILE", help = "placeholder")] + profile: Option, + #[command(subcommand)] command: Command, } @@ -171,6 +176,35 @@ enum BucketCommand { struct ConfigToml { model_provider: Option, sqlite_home: Option, + profile: Option, + profiles: Option>, +} + +#[derive(Debug, Deserialize)] +struct ConfigProfileToml { + model_provider: Option, +} + +impl Default for ConfigToml { + fn default() -> Self { + Self { + model_provider: None, + sqlite_home: None, + profile: None, + profiles: None, + } + } +} + +impl ConfigToml { + fn overlay(&mut self, profile_config: ConfigToml) { + if profile_config.model_provider.is_some() { + self.model_provider = profile_config.model_provider; + } + if profile_config.sqlite_home.is_some() { + self.sqlite_home = profile_config.sqlite_home; + } + } } #[derive(Debug)] @@ -334,11 +368,13 @@ fn main() -> Result<()> { let locale = detect_locale(); let cli = parse_cli(locale)?; validate_provider_override(locale, cli.provider.as_deref())?; + validate_profile_override(locale, cli.profile.as_deref())?; let codex_home = cli.codex_home.unwrap_or_else(default_codex_home); match cli.command { Command::Status => { - let summary = collect_status(&codex_home, cli.provider.as_deref())?; + let summary = + collect_status(&codex_home, cli.provider.as_deref(), cli.profile.as_deref())?; print_status(locale, &summary); } Command::Sync { sqlite_only } => { @@ -355,6 +391,7 @@ fn main() -> Result<()> { let summary = reconcile_once_with_backup_progress( &codex_home, cli.provider.as_deref(), + cli.profile.as_deref(), rollout_scope, progress, )?; @@ -362,7 +399,8 @@ fn main() -> Result<()> { } Command::Bucket { command } => match command { BucketCommand::Prepare { padding_bytes } => { - let summary = prepare_bucket_padding(&codex_home, padding_bytes)?; + let summary = + prepare_bucket_padding(&codex_home, cli.profile.as_deref(), padding_bytes)?; print_bucket_prepare_summary(locale, &summary); } BucketCommand::Switch { @@ -377,6 +415,7 @@ fn main() -> Result<()> { let summary = reconcile_once_with_backup_and_padding( &codex_home, provider.as_deref(), + cli.profile.as_deref(), RolloutScope::AllRows, padding_bytes, Some(RolloutProgressConfig { locale }), @@ -392,6 +431,7 @@ fn main() -> Result<()> { locale, &codex_home, cli.provider.clone(), + cli.profile.clone(), if sqlite_only { RolloutScope::None } else { @@ -406,6 +446,7 @@ fn main() -> Result<()> { exe_path.as_path(), &codex_home, cli.provider.as_deref(), + cli.profile.as_deref(), Duration::from_millis(poll_interval_ms), )?; println!("{config}"); @@ -415,6 +456,7 @@ fn main() -> Result<()> { locale, &codex_home, cli.provider.as_deref(), + cli.profile.as_deref(), Duration::from_millis(poll_interval_ms), )?; } @@ -440,6 +482,26 @@ fn validate_provider_override(locale: Locale, provider: Option<&str>) -> Result< Ok(()) } +fn validate_profile_override(locale: Locale, profile: Option<&str>) -> Result<()> { + let Some(profile) = profile else { + return Ok(()); + }; + let trimmed = profile.trim(); + if trimmed.is_empty() { + anyhow::bail!(profile_empty_error(locale)); + } + if !is_valid_profile_name(trimmed) { + anyhow::bail!(profile_path_error(locale)); + } + Ok(()) +} + +fn is_valid_profile_name(profile: &str) -> bool { + profile + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) +} + fn validate_provider_override_args(locale: Locale, args: I) -> Result<()> where I: IntoIterator, @@ -457,11 +519,20 @@ where } continue; } + if arg == OsStr::new("--profile") { + if let Some(value) = args.next() { + validate_profile_override(locale, value.as_ref().to_str())?; + } + continue; + } let rendered = arg.to_string_lossy(); if let Some(value) = rendered.strip_prefix("--provider=") { validate_provider_override(locale, Some(value))?; } + if let Some(value) = rendered.strip_prefix("--profile=") { + validate_profile_override(locale, Some(value))?; + } } Ok(()) } @@ -505,6 +576,11 @@ fn localized_command(locale: Locale) -> clap::Command { .help_heading(options_heading(locale)) .value_name(provider_value_name(locale)) }); + command = command.mut_arg("profile", |arg| { + arg.help(profile_help(locale)) + .help_heading(options_heading(locale)) + .value_name(profile_value_name(locale)) + }); command = command.mut_subcommand("status", |sub| { sub.about(status_about(locale)) @@ -698,6 +774,7 @@ fn run_watch( locale: Locale, codex_home: &Path, provider_override: Option, + profile_override: Option, rollout_scope: RolloutScope, poll_interval: Duration, ) -> Result<()> { @@ -738,7 +815,12 @@ fn run_watch( } let mut last_provider = None; - match reconcile_once(codex_home, provider_override.as_deref(), rollout_scope) { + match reconcile_once( + codex_home, + provider_override.as_deref(), + profile_override.as_deref(), + rollout_scope, + ) { Ok(summary) => { print_sync_summary(locale, watch_started_title(locale), &summary); last_provider = Some(summary.provider.clone()); @@ -759,7 +841,12 @@ fn run_watch( match rx.recv_timeout(timeout) { Ok(Ok(event)) => { if touches_config_file(&event, &config_path) { - match reconcile_once(codex_home, provider_override.as_deref(), rollout_scope) { + match reconcile_once( + codex_home, + provider_override.as_deref(), + profile_override.as_deref(), + rollout_scope, + ) { Ok(summary) => { if last_provider.as_deref() != Some(summary.provider.as_str()) || summary.changed_rows > 0 @@ -780,7 +867,12 @@ fn run_watch( eprintln!("{}", watcher_error_message(locale, err)); } Err(mpsc::RecvTimeoutError::Timeout) => { - match reconcile_once(codex_home, provider_override.as_deref(), rollout_scope) { + match reconcile_once( + codex_home, + provider_override.as_deref(), + profile_override.as_deref(), + rollout_scope, + ) { Ok(summary) => { if last_provider.as_deref() != Some(summary.provider.as_str()) || summary.changed_rows > 0 @@ -816,12 +908,16 @@ fn touches_config_file(event: ¬ify::Event, config_path: &Path) -> bool { ) && event.paths.iter().any(|path| path == config_path) } -fn collect_status(codex_home: &Path, provider_override: Option<&str>) -> Result { +fn collect_status( + codex_home: &Path, + provider_override: Option<&str>, + profile_override: Option<&str>, +) -> Result { let config_path = codex_home.join("config.toml"); - let sqlite_path = resolve_sqlite_path(codex_home)?; + let sqlite_path = resolve_sqlite_path(codex_home, profile_override)?; let provider = match provider_override { Some(provider) => provider.to_string(), - None => read_provider_from_config(codex_home)?, + None => read_provider_from_config(codex_home, profile_override)?, }; let (total_rows, mismatched_rows, distribution) = inspect_sqlite_distribution(&sqlite_path, provider.as_str())?; @@ -842,22 +938,30 @@ fn collect_status(codex_home: &Path, provider_override: Option<&str>) -> Result< fn reconcile_once( codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, rollout_scope: RolloutScope, ) -> Result { - reconcile_once_with_progress(codex_home, provider_override, rollout_scope, None) + reconcile_once_with_progress( + codex_home, + provider_override, + profile_override, + rollout_scope, + None, + ) } fn reconcile_once_with_progress( codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, rollout_scope: RolloutScope, progress: Option, ) -> Result { let provider = match provider_override { Some(provider) => provider.to_string(), - None => read_provider_from_config(codex_home)?, + None => read_provider_from_config(codex_home, profile_override)?, }; - let sqlite_path = resolve_sqlite_path(codex_home)?; + let sqlite_path = resolve_sqlite_path(codex_home, profile_override)?; let started = Instant::now(); let rollout_summary = reconcile_rollout_metadata_from_sqlite_with_progress( &sqlite_path, @@ -887,12 +991,14 @@ fn reconcile_once_with_progress( fn reconcile_once_with_backup_progress( codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, rollout_scope: RolloutScope, progress: Option, ) -> Result { reconcile_once_with_backup_and_padding( codex_home, provider_override, + profile_override, rollout_scope, DEFAULT_BUCKET_PADDING_BYTES, progress, @@ -902,15 +1008,16 @@ fn reconcile_once_with_backup_progress( fn reconcile_once_with_backup_and_padding( codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, rollout_scope: RolloutScope, padding_bytes: usize, progress: Option, ) -> Result { let provider = match provider_override { Some(provider) => provider.to_string(), - None => read_provider_from_config(codex_home)?, + None => read_provider_from_config(codex_home, profile_override)?, }; - let sqlite_path = resolve_sqlite_path(codex_home)?; + let sqlite_path = resolve_sqlite_path(codex_home, profile_override)?; let started = Instant::now(); let backup_path = create_sqlite_backup_file(&sqlite_path)?; let rollout_journal_path = @@ -949,16 +1056,16 @@ fn reconcile_once_with_backup_and_padding( }) } -fn read_provider_from_config(codex_home: &Path) -> Result { - let parsed = read_codex_config(codex_home)?; +fn read_provider_from_config(codex_home: &Path, profile_override: Option<&str>) -> Result { + let parsed = read_effective_codex_config(codex_home, profile_override)?; Ok(parsed .model_provider .filter(|s| !s.trim().is_empty()) .unwrap_or_else(|| DEFAULT_PROVIDER.to_string())) } -fn resolve_sqlite_path(codex_home: &Path) -> Result { - let parsed = read_codex_config(codex_home)?; +fn resolve_sqlite_path(codex_home: &Path, profile_override: Option<&str>) -> Result { + let parsed = read_effective_codex_config(codex_home, profile_override)?; let current_dir = std::env::current_dir().context("failed to resolve current directory for sqlite_home")?; Ok(resolve_sqlite_home_from_config( @@ -970,18 +1077,75 @@ fn resolve_sqlite_path(codex_home: &Path) -> Result { .join(STATE_DB_FILENAME)) } +fn read_effective_codex_config( + codex_home: &Path, + profile_override: Option<&str>, +) -> Result { + let mut config = read_codex_config(codex_home)?; + let active_profile = profile_override + .and_then(trimmed_string) + .or_else(|| config.profile.as_deref().and_then(trimmed_string)); + + if let Some(profile) = active_profile { + if !is_valid_profile_name(profile.as_str()) { + anyhow::bail!("profile `{profile}` is not a valid Codex profile name"); + } + if let Some(profile_config) = read_profile_v2_config(codex_home, profile.as_str())? { + config.overlay(profile_config); + } else if let Some(profile_config) = config + .profiles + .as_ref() + .and_then(|profiles| profiles.get(profile.as_str())) + { + if let Some(provider) = profile_config.model_provider.as_ref() { + config.model_provider = Some(provider.clone()); + } + } else { + anyhow::bail!( + "profile `{}` was not found; expected {} or [profiles.{}] in {}", + profile, + profile_config_path(codex_home, profile.as_str()).display(), + profile, + codex_home.join("config.toml").display() + ); + } + } + + Ok(config) +} + fn read_codex_config(codex_home: &Path) -> Result { - let config_path = codex_home.join("config.toml"); - if !config_path.exists() { - return Ok(ConfigToml { - model_provider: None, - sqlite_home: None, - }); + read_optional_config_file(codex_home.join("config.toml").as_path()) + .map(|config| config.unwrap_or_default()) +} + +fn read_profile_v2_config(codex_home: &Path, profile: &str) -> Result> { + let path = profile_config_path(codex_home, profile); + read_optional_config_file(path.as_path()) +} + +fn read_optional_config_file(path: &Path) -> Result> { + if !path.exists() { + return Ok(None); } + let raw = std::fs::read_to_string(path) + .with_context(|| format!("failed to read {}", path.display()))?; + toml::from_str(&raw) + .map(Some) + .with_context(|| format!("failed to parse {}", path.display())) +} + +fn profile_config_path(codex_home: &Path, profile: &str) -> PathBuf { + codex_home.join(format!("{profile}{PROFILE_CONFIG_SUFFIX}")) +} - let raw = std::fs::read_to_string(&config_path) - .with_context(|| format!("failed to read {}", config_path.display()))?; - toml::from_str(&raw).with_context(|| format!("failed to parse {}", config_path.display())) +fn trimmed_string(value: &str) -> Option { + let trimmed = value.trim(); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } } fn resolve_sqlite_home_from_config( @@ -1188,8 +1352,12 @@ fn rewrite_rollout_provider_first_line( }) } -fn prepare_bucket_padding(codex_home: &Path, padding_bytes: usize) -> Result { - let sqlite_path = resolve_sqlite_path(codex_home)?; +fn prepare_bucket_padding( + codex_home: &Path, + profile_override: Option<&str>, + padding_bytes: usize, +) -> Result { + let sqlite_path = resolve_sqlite_path(codex_home, profile_override)?; let started = Instant::now(); let targets = rollout_targets_for_scope(&sqlite_path, DEFAULT_PROVIDER, RolloutScope::AllRows)?; let journal_path = codex_home @@ -1681,6 +1849,7 @@ fn install_service( locale: Locale, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> Result<()> { let exe_path = std::env::current_exe().context(current_exe_error(locale))?; @@ -1688,6 +1857,7 @@ fn install_service( exe_path.as_path(), codex_home, provider_override, + profile_override, poll_interval, )?; @@ -1963,6 +2133,13 @@ fn provider_value_name(locale: Locale) -> &'static str { } } +fn profile_value_name(locale: Locale) -> &'static str { + match locale { + Locale::En => "PROFILE", + Locale::ZhHans => "PROFILE", + } +} + fn milliseconds_value_name(locale: Locale) -> &'static str { match locale { Locale::En => "MILLISECONDS", @@ -1991,6 +2168,13 @@ fn provider_help(locale: Locale) -> &'static str { } } +fn profile_help(locale: Locale) -> &'static str { + match locale { + Locale::En => "Resolve model_provider and sqlite_home from a Codex profile config first.", + Locale::ZhHans => "优先从 Codex profile 配置解析 model_provider 和 sqlite_home。", + } +} + fn provider_empty_error(locale: Locale) -> &'static str { match locale { Locale::En => "provider must contain at least one non-whitespace character", @@ -1998,6 +2182,22 @@ fn provider_empty_error(locale: Locale) -> &'static str { } } +fn profile_empty_error(locale: Locale) -> &'static str { + match locale { + Locale::En => "profile must contain at least one non-whitespace character", + Locale::ZhHans => "profile 需要包含至少一个非空白字符", + } +} + +fn profile_path_error(locale: Locale) -> &'static str { + match locale { + Locale::En => { + "profile must be a plain name containing only ASCII letters, digits, '_' or '-'" + } + Locale::ZhHans => "profile 需要是只包含 ASCII 字母、数字、'_' 或 '-' 的普通名称", + } +} + fn help_option_help(locale: Locale) -> &'static str { match locale { Locale::En => "Print help", @@ -2652,15 +2852,69 @@ mod tests { fn reads_provider_from_config() -> Result<()> { let dir = tempfile::tempdir()?; fs::write(dir.path().join("config.toml"), "model_provider = \"vm\"\n")?; - let provider = read_provider_from_config(dir.path())?; + let provider = read_provider_from_config(dir.path(), None)?; assert_eq!(provider, "vm"); Ok(()) } + #[test] + fn reads_provider_from_profile_v2_config() -> Result<()> { + let dir = tempfile::tempdir()?; + fs::write( + dir.path().join("config.toml"), + "model_provider = \"openai\"\n", + )?; + fs::write( + dir.path().join("work.config.toml"), + "model_provider = \"bedrock\"\nsqlite_home = \"work-state\"\n", + )?; + + let provider = read_provider_from_config(dir.path(), Some("work"))?; + let sqlite_path = resolve_sqlite_path(dir.path(), Some("work"))?; + + assert_eq!(provider, "bedrock"); + assert_eq!(sqlite_path, dir.path().join("work-state/state_5.sqlite")); + Ok(()) + } + + #[test] + fn reads_provider_from_selected_legacy_profile() -> Result<()> { + let dir = tempfile::tempdir()?; + fs::write( + dir.path().join("config.toml"), + r#" +model_provider = "openai" +profile = "work" + +[profiles.work] +model_provider = "local" +"#, + )?; + + let provider = read_provider_from_config(dir.path(), None)?; + + assert_eq!(provider, "local"); + Ok(()) + } + + #[test] + fn errors_when_profile_config_is_missing() -> Result<()> { + let dir = tempfile::tempdir()?; + fs::write( + dir.path().join("config.toml"), + "model_provider = \"openai\"\n", + )?; + + let err = read_provider_from_config(dir.path(), Some("missing")).unwrap_err(); + + assert!(err.to_string().contains("profile `missing` was not found")); + Ok(()) + } + #[test] fn defaults_to_openai_without_config() -> Result<()> { let dir = tempfile::tempdir()?; - let provider = read_provider_from_config(dir.path())?; + let provider = read_provider_from_config(dir.path(), None)?; assert_eq!(provider, "openai"); Ok(()) } @@ -2669,7 +2923,7 @@ mod tests { fn defaults_to_openai_with_blank_provider_in_config() -> Result<()> { let dir = tempfile::tempdir()?; fs::write(dir.path().join("config.toml"), "model_provider = \" \"\n")?; - let provider = read_provider_from_config(dir.path())?; + let provider = read_provider_from_config(dir.path(), None)?; assert_eq!(provider, "openai"); Ok(()) } @@ -2683,7 +2937,7 @@ mod tests { format!("sqlite_home = \"{}\"\n", sqlite_home.display()), )?; - let sqlite_path = resolve_sqlite_path(dir.path())?; + let sqlite_path = resolve_sqlite_path(dir.path(), None)?; assert_eq!(sqlite_path, sqlite_home.join("state_5.sqlite")); Ok(()) @@ -2731,6 +2985,32 @@ mod tests { assert!(err.to_string().contains("provider must contain")); } + #[test] + fn rejects_blank_profile_override_in_cli_args_before_help_short_circuit() { + let err = validate_provider_override_args( + Locale::En, + ["codex-threadripper", "--profile", "", "status", "--help"], + ) + .unwrap_err(); + assert!(err.to_string().contains("profile must contain")); + } + + #[test] + fn rejects_path_profile_override_in_cli_args_before_help_short_circuit() { + let err = validate_provider_override_args( + Locale::En, + [ + "codex-threadripper", + "--profile", + "nested/work", + "status", + "--help", + ], + ) + .unwrap_err(); + assert!(err.to_string().contains("profile must be a plain name")); + } + #[test] fn reconcile_once_returns_error_when_db_missing() -> Result<()> { let dir = tempfile::tempdir()?; @@ -2740,7 +3020,7 @@ mod tests { format!("sqlite_home = \"{}\"\n", dir.path().display()), )?; - let err = reconcile_once(dir.path(), Some("openai"), RolloutScope::None).unwrap_err(); + let err = reconcile_once(dir.path(), Some("openai"), None, RolloutScope::None).unwrap_err(); assert!(err.to_string().contains(&sqlite_path.display().to_string())); assert!(!sqlite_path.exists()); @@ -2940,11 +3220,13 @@ mod tests { PathBuf::from("/tmp/codex-threadripper").as_path(), PathBuf::from("/tmp/codex-home").as_path(), Some("openai"), + Some("work"), Duration::from_millis(DEFAULT_POLL_INTERVAL_MS), ); assert!(plist.contains("dev.wangnov.codex-threadripper")); assert!(plist.contains("--codex-home")); assert!(plist.contains("--provider")); + assert!(plist.contains("--profile")); assert!(plist.contains("watch")); } diff --git a/src/service.rs b/src/service.rs index 3dd3c82..08a9064 100644 --- a/src/service.rs +++ b/src/service.rs @@ -74,21 +74,34 @@ pub fn install_service( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> Result { let manager = current_manager(); stop_detached_watch_if_present(codex_home)?; match manager { - ServiceManager::Launchd => { - install_launchd(exe_path, codex_home, provider_override, poll_interval) - } - ServiceManager::SystemdUser => { - install_systemd_user(exe_path, codex_home, provider_override, poll_interval) - } - ServiceManager::WindowsStartup => { - install_windows_startup(exe_path, codex_home, provider_override, poll_interval) - } + ServiceManager::Launchd => install_launchd( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + ), + ServiceManager::SystemdUser => install_systemd_user( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + ), + ServiceManager::WindowsStartup => install_windows_startup( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + ), } } @@ -110,6 +123,7 @@ pub fn render_service_config( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> Result { let manager = current_manager(); @@ -118,15 +132,21 @@ pub fn render_service_config( exe_path, codex_home, provider_override, + profile_override, poll_interval, )), - ServiceManager::SystemdUser => { - build_systemd_bundle(exe_path, codex_home, provider_override, poll_interval) - } + ServiceManager::SystemdUser => build_systemd_bundle( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + ), ServiceManager::WindowsStartup => Ok(build_windows_startup_vbs( exe_path, codex_home, provider_override, + profile_override, poll_interval, )), } @@ -171,6 +191,7 @@ fn install_launchd( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> Result { let plist_path = service_config_path(ServiceManager::Launchd)?; @@ -179,7 +200,13 @@ fn install_launchd( .with_context(|| format!("launchd plist path has no parent: {}", plist_path.display()))?; std::fs::create_dir_all(launch_agents_dir)?; - let plist = build_launchd_plist(exe_path, codex_home, provider_override, poll_interval); + let plist = build_launchd_plist( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + ); std::fs::write(&plist_path, plist) .with_context(|| format!("failed to write {}", plist_path.display()))?; @@ -221,6 +248,7 @@ fn install_systemd_user( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> Result { let config_path = service_config_path(ServiceManager::SystemdUser)?; @@ -238,7 +266,13 @@ fn install_systemd_user( })?; std::fs::create_dir_all(runner_script_dir)?; - let script = build_linux_runner_script(exe_path, codex_home, provider_override, poll_interval); + let script = build_linux_runner_script( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + ); std::fs::write(&runner_script_path, script) .with_context(|| format!("failed to write {}", runner_script_path.display()))?; make_executable(&runner_script_path)?; @@ -249,7 +283,13 @@ fn install_systemd_user( let systemd_started = try_run_systemd_user_unit()?; if !systemd_started { - start_detached_watch(exe_path, codex_home, provider_override, poll_interval)?; + start_detached_watch( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + )?; } Ok(ServiceInstallSummary { @@ -286,6 +326,7 @@ fn install_windows_startup( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> Result { let config_path = service_config_path(ServiceManager::WindowsStartup)?; @@ -297,7 +338,13 @@ fn install_windows_startup( })?; std::fs::create_dir_all(config_dir)?; - let script = build_windows_startup_vbs(exe_path, codex_home, provider_override, poll_interval); + let script = build_windows_startup_vbs( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + ); std::fs::write(&config_path, script) .with_context(|| format!("failed to write {}", config_path.display()))?; @@ -307,7 +354,13 @@ fn install_windows_startup( .with_context(|| format!("failed to remove {}", legacy_cmd_path.display()))?; } - start_detached_watch(exe_path, codex_home, provider_override, poll_interval)?; + start_detached_watch( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + )?; Ok(ServiceInstallSummary { manager: ServiceManager::WindowsStartup, @@ -393,10 +446,17 @@ fn build_systemd_bundle( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> Result { let runner_script_path = linux_runner_script_path()?; - let script = build_linux_runner_script(exe_path, codex_home, provider_override, poll_interval); + let script = build_linux_runner_script( + exe_path, + codex_home, + provider_override, + profile_override, + poll_interval, + ); let unit = build_systemd_unit(&runner_script_path); Ok(format!( "# {}\n{}\n# {}\n{}", @@ -418,6 +478,7 @@ fn build_linux_runner_script( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> String { format!( @@ -426,6 +487,7 @@ fn build_linux_runner_script( exe_path, codex_home, provider_override, + profile_override, poll_interval, ShellFlavor::Sh ) @@ -436,12 +498,14 @@ fn build_windows_startup_vbs( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> String { let command = watch_command_line( exe_path, codex_home, provider_override, + profile_override, poll_interval, ShellFlavor::WindowsProcess, ); @@ -456,6 +520,7 @@ pub fn build_launchd_plist( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> String { let stdout_path = @@ -472,6 +537,10 @@ pub fn build_launchd_plist( arguments.push("--provider".to_string()); arguments.push(xml_escape(provider)); } + if let Some(profile) = profile_override { + arguments.push("--profile".to_string()); + arguments.push(xml_escape(profile)); + } arguments.push("watch".to_string()); arguments.push("--poll-interval-ms".to_string()); arguments.push(poll_interval.as_millis().to_string()); @@ -516,6 +585,7 @@ fn watch_command_line( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, flavor: ShellFlavor, ) -> String { @@ -533,6 +603,10 @@ fn watch_command_line( parts.push("--provider".to_string()); parts.push(quote(provider.to_string())); } + if let Some(profile) = profile_override { + parts.push("--profile".to_string()); + parts.push(quote(profile.to_string())); + } parts.push("watch".to_string()); parts.push("--poll-interval-ms".to_string()); parts.push(poll_interval.as_millis().to_string()); @@ -549,6 +623,7 @@ fn start_detached_watch( exe_path: &Path, codex_home: &Path, provider_override: Option<&str>, + profile_override: Option<&str>, poll_interval: Duration, ) -> Result<()> { let stdout_path = log_path_for(codex_home)?; @@ -574,6 +649,9 @@ fn start_detached_watch( if let Some(provider) = provider_override { command.arg("--provider").arg(provider); } + if let Some(profile) = profile_override { + command.arg("--profile").arg(profile); + } command .arg("watch") @@ -1112,6 +1190,7 @@ mod tests { Path::new(r"C:\Program Files\codex threadripper\codex-threadripper.exe"), Path::new(r"C:\Users\Admin User\.codex"), None, + None, Duration::from_millis(1500), ); let command = shell_run_command(&script); @@ -1136,6 +1215,7 @@ mod tests { Path::new(r"C:\Tools\codex-threadripper.exe"), Path::new(r"C:\Codex"), Some(r#"open ai "beta""#), + None, Duration::from_millis(500), ); let command = shell_run_command(&script); From 8b59b9fd0d5b7f78c8aa0726fab7966f6f4f56c6 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Mon, 25 May 2026 00:07:10 +0800 Subject: [PATCH 2/8] fix: address profile watch review findings --- .github/workflows/ci.yml | 20 +++ src/main.rs | 286 ++++++++++++++++++++++++++++----------- src/service.rs | 31 +++++ 3 files changed, 256 insertions(+), 81 deletions(-) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..ac1221f --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,20 @@ +name: CI + +on: + pull_request: + push: + branches: + - main + +jobs: + rust: + runs-on: ubuntu-22.04 + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + - run: cargo fmt --check + - run: cargo test --locked + - run: cargo clippy --locked --all-targets -- -D warnings + - run: python3 -m py_compile scripts/smoke_release.py diff --git a/src/main.rs b/src/main.rs index 562e785..7c60def 100644 --- a/src/main.rs +++ b/src/main.rs @@ -172,7 +172,7 @@ enum BucketCommand { }, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Default, Deserialize)] struct ConfigToml { model_provider: Option, sqlite_home: Option, @@ -185,17 +185,6 @@ struct ConfigProfileToml { model_provider: Option, } -impl Default for ConfigToml { - fn default() -> Self { - Self { - model_provider: None, - sqlite_home: None, - profile: None, - profiles: None, - } - } -} - impl ConfigToml { fn overlay(&mut self, profile_config: ConfigToml) { if profile_config.model_provider.is_some() { @@ -279,6 +268,17 @@ struct RolloutProgress { printed: bool, } +#[derive(Clone, Copy, Debug)] +struct RolloutProgressSnapshot { + visited_files: u64, + total_files: u64, + checked_files: u64, + changed_files: u64, + prepared_files: u64, + skipped_files: u64, + elapsed: Duration, +} + #[derive(Debug)] struct FirstLine { content: Vec, @@ -322,16 +322,16 @@ impl RolloutProgress { } fn print(&mut self, summary: &RolloutReconcileSummary, final_line: bool) { - let message = rollout_progress_message( - self.config.locale, - self.visited_files, - self.total_files, - summary.checked_files, - summary.changed_files, - summary.prepared_files, - summary.skipped_files, - self.started_at.elapsed(), - ); + let snapshot = RolloutProgressSnapshot { + visited_files: self.visited_files, + total_files: self.total_files, + checked_files: summary.checked_files, + changed_files: summary.changed_files, + prepared_files: summary.prepared_files, + skipped_files: summary.skipped_files, + elapsed: self.started_at.elapsed(), + }; + let message = rollout_progress_message(self.config.locale, &snapshot); let mut stderr = std::io::stderr(); if self.is_terminal { let _ = write!(stderr, "\r{message}\x1b[K"); @@ -794,6 +794,8 @@ fn run_watch( } let config_path = codex_home.join("config.toml"); + let mut watched_paths = watched_config_paths(codex_home, profile_override.as_deref()); + let full_rollout_scope = full_watch_rollout_scope(rollout_scope); let shutdown = Arc::new(AtomicBool::new(false)); let shutdown_for_handler = Arc::clone(&shutdown); ctrlc::set_handler(move || { @@ -819,7 +821,7 @@ fn run_watch( codex_home, provider_override.as_deref(), profile_override.as_deref(), - rollout_scope, + full_rollout_scope, ) { Ok(summary) => { print_sync_summary(locale, watch_started_title(locale), &summary); @@ -840,12 +842,12 @@ fn run_watch( let timeout = next_poll_deadline.saturating_duration_since(Instant::now()); match rx.recv_timeout(timeout) { Ok(Ok(event)) => { - if touches_config_file(&event, &config_path) { + if touches_config_file(&event, watched_paths.as_slice()) { match reconcile_once( codex_home, provider_override.as_deref(), profile_override.as_deref(), - rollout_scope, + full_rollout_scope, ) { Ok(summary) => { if last_provider.as_deref() != Some(summary.provider.as_str()) @@ -855,6 +857,8 @@ fn run_watch( print_sync_summary(locale, config_change_title(locale), &summary); } last_provider = Some(summary.provider.clone()); + watched_paths = + watched_config_paths(codex_home, profile_override.as_deref()); } Err(err) => { eprintln!("{}", watch_reconcile_skipped_message(locale, &err)); @@ -901,11 +905,36 @@ fn run_watch( Ok(()) } -fn touches_config_file(event: ¬ify::Event, config_path: &Path) -> bool { +fn full_watch_rollout_scope(rollout_scope: RolloutScope) -> RolloutScope { + match rollout_scope { + RolloutScope::None => RolloutScope::None, + RolloutScope::MismatchedRows | RolloutScope::AllRows => RolloutScope::AllRows, + } +} + +fn watched_config_paths(codex_home: &Path, profile_override: Option<&str>) -> Vec { + let mut paths = vec![codex_home.join("config.toml")]; + let profile = profile_override.and_then(trimmed_string).or_else(|| { + read_codex_config(codex_home) + .ok() + .and_then(|config| config.profile.as_deref().and_then(trimmed_string)) + }); + if let Some(profile) = profile + && is_valid_profile_name(profile.as_str()) + { + paths.push(profile_config_path(codex_home, profile.as_str())); + } + paths +} + +fn touches_config_file(event: ¬ify::Event, config_paths: &[PathBuf]) -> bool { matches!( event.kind, EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) - ) && event.paths.iter().any(|path| path == config_path) + ) && event + .paths + .iter() + .any(|path| config_paths.iter().any(|config_path| path == config_path)) } fn collect_status( @@ -1082,38 +1111,53 @@ fn read_effective_codex_config( profile_override: Option<&str>, ) -> Result { let mut config = read_codex_config(codex_home)?; - let active_profile = profile_override - .and_then(trimmed_string) - .or_else(|| config.profile.as_deref().and_then(trimmed_string)); + let explicit_profile = profile_override.and_then(trimmed_string); + let config_profile = config.profile.as_deref().and_then(trimmed_string); - if let Some(profile) = active_profile { - if !is_valid_profile_name(profile.as_str()) { - anyhow::bail!("profile `{profile}` is not a valid Codex profile name"); - } - if let Some(profile_config) = read_profile_v2_config(codex_home, profile.as_str())? { - config.overlay(profile_config); - } else if let Some(profile_config) = config - .profiles - .as_ref() - .and_then(|profiles| profiles.get(profile.as_str())) - { - if let Some(provider) = profile_config.model_provider.as_ref() { - config.model_provider = Some(provider.clone()); - } - } else { - anyhow::bail!( - "profile `{}` was not found; expected {} or [profiles.{}] in {}", - profile, - profile_config_path(codex_home, profile.as_str()).display(), - profile, - codex_home.join("config.toml").display() - ); - } + if let Some(profile) = explicit_profile { + apply_profile_config(codex_home, &mut config, profile.as_str(), true)?; + } else if let Some(profile) = config_profile { + apply_profile_config(codex_home, &mut config, profile.as_str(), false)?; } Ok(config) } +fn apply_profile_config( + codex_home: &Path, + config: &mut ConfigToml, + profile: &str, + required: bool, +) -> Result<()> { + if !is_valid_profile_name(profile) { + anyhow::bail!("profile `{profile}` is not a valid Codex profile name"); + } + if let Some(profile_config) = read_profile_v2_config(codex_home, profile)? { + config.overlay(profile_config); + return Ok(()); + } + if let Some(profile_config) = config + .profiles + .as_ref() + .and_then(|profiles| profiles.get(profile)) + { + if let Some(provider) = profile_config.model_provider.as_ref() { + config.model_provider = Some(provider.clone()); + } + return Ok(()); + } + if required { + anyhow::bail!( + "profile `{}` was not found; expected {} or [profiles.{}] in {}", + profile, + profile_config_path(codex_home, profile).display(), + profile, + codex_home.join("config.toml").display() + ); + } + Ok(()) +} + fn read_codex_config(codex_home: &Path) -> Result { read_optional_config_file(codex_home.join("config.toml").as_path()) .map(|config| config.unwrap_or_default()) @@ -1125,11 +1169,11 @@ fn read_profile_v2_config(codex_home: &Path, profile: &str) -> Result Result> { - if !path.exists() { - return Ok(None); - } - let raw = std::fs::read_to_string(path) - .with_context(|| format!("failed to read {}", path.display()))?; + let raw = match std::fs::read_to_string(path) { + Ok(raw) => raw, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(err) => return Err(err).with_context(|| format!("failed to read {}", path.display())), + }; toml::from_str(&raw) .map(Some) .with_context(|| format!("failed to parse {}", path.display())) @@ -1864,7 +1908,14 @@ fn install_service( print_install_service_summary(locale, codex_home, poll_interval, &summary); println!(); println!("{}", next_steps_heading(locale)); - for line in install_next_steps(locale, exe_path.as_path(), codex_home, summary.manager)? { + for line in install_next_steps( + locale, + exe_path.as_path(), + codex_home, + provider_override, + profile_override, + summary.manager, + )? { println!("{line}"); } Ok(()) @@ -1882,7 +1933,7 @@ fn uninstall_service(locale: Locale, codex_home: &Path) -> Result<()> { "{}", run_status_next_step( locale, - &cli_status_command(std::env::current_exe()?, codex_home) + &cli_status_command(std::env::current_exe()?, codex_home, None, None) ) ); } else { @@ -1956,16 +2007,16 @@ fn print_status(locale: Locale, summary: &StatusSummary) { ); } -fn rollout_progress_message( - locale: Locale, - visited_files: u64, - total_files: u64, - checked_files: u64, - changed_files: u64, - prepared_files: u64, - skipped_files: u64, - elapsed: Duration, -) -> String { +fn rollout_progress_message(locale: Locale, snapshot: &RolloutProgressSnapshot) -> String { + let RolloutProgressSnapshot { + visited_files, + total_files, + checked_files, + changed_files, + prepared_files, + skipped_files, + elapsed, + } = *snapshot; match locale { Locale::En => format!( "Rollouts: scanned {visited_files}/{total_files}, checked {checked_files}, updated {changed_files}, prepared {prepared_files}, skipped {skipped_files}, elapsed {} ms", @@ -2090,10 +2141,10 @@ fn root_about(locale: Locale) -> &'static str { fn root_long_about(locale: Locale) -> &'static str { match locale { Locale::En => { - "codex-threadripper is a human-first maintenance tool for Codex thread history.\n\nIt reads the active provider from CODEX_HOME/config.toml and rewrites Codex's SQLite state DB so every thread stays in the same provider bucket. The DB defaults to CODEX_HOME/state_5.sqlite, but sqlite_home and CODEX_SQLITE_HOME are respected. That makes thread lists and resume flows stop fragmenting across providers.\n\nExamples:\n codex-threadripper status\n codex-threadripper sync\n codex-threadripper watch\n codex-threadripper install-service" + "codex-threadripper is a human-first maintenance tool for Codex thread history.\n\nIt reads the effective Codex provider from CODEX_HOME/config.toml, including the selected profile config when present, and rewrites Codex's SQLite state DB so every thread stays in the same provider bucket. The DB defaults to CODEX_HOME/state_5.sqlite, but sqlite_home and CODEX_SQLITE_HOME are respected. That makes thread lists and resume flows stop fragmenting across providers.\n\nExamples:\n codex-threadripper status\n codex-threadripper sync\n codex-threadripper watch\n codex-threadripper install-service" } Locale::ZhHans => { - "codex-threadripper 是一个面向人的 Codex 线程历史维护工具。\n\n它会读取 CODEX_HOME/config.toml 里的当前 provider，并改写 Codex 的 SQLite 状态库，让所有线程始终落在同一个 provider 桶里。状态库默认是 CODEX_HOME/state_5.sqlite，同时会尊重 sqlite_home 和 CODEX_SQLITE_HOME。这样线程列表和 resume 流程就不会再被 provider 切碎。\n\n示例：\n codex-threadripper status\n codex-threadripper sync\n codex-threadripper watch\n codex-threadripper install-service" + "codex-threadripper 是一个面向人的 Codex 线程历史维护工具。\n\n它会读取 CODEX_HOME/config.toml 和已选 profile 配置合成后的有效 provider，并改写 Codex 的 SQLite 状态库，让所有线程始终落在同一个 provider 桶里。状态库默认是 CODEX_HOME/state_5.sqlite，同时会尊重 sqlite_home 和 CODEX_SQLITE_HOME。这样线程列表和 resume 流程就不会再被 provider 切碎。\n\n示例：\n codex-threadripper status\n codex-threadripper sync\n codex-threadripper watch\n codex-threadripper install-service" } } } @@ -2170,8 +2221,12 @@ fn provider_help(locale: Locale) -> &'static str { fn profile_help(locale: Locale) -> &'static str { match locale { - Locale::En => "Resolve model_provider and sqlite_home from a Codex profile config first.", - Locale::ZhHans => "优先从 Codex profile 配置解析 model_provider 和 sqlite_home。", + Locale::En => { + "Resolve model_provider and sqlite_home from a Codex profile config first. Profile names may contain ASCII letters, digits, '_' or '-'." + } + Locale::ZhHans => { + "优先从 Codex profile 配置解析 model_provider 和 sqlite_home。profile 名称可包含 ASCII 字母、数字、'_' 或 '-'。" + } } } @@ -2498,9 +2553,12 @@ fn install_next_steps( locale: Locale, exe_path: &Path, codex_home: &Path, + provider_override: Option<&str>, + profile_override: Option<&str>, manager: ServiceManager, ) -> Result> { - let status_command = cli_status_command(exe_path, codex_home); + let status_command = + cli_status_command(exe_path, codex_home, provider_override, profile_override); let log_path = service::log_path()?; let mut steps = vec![run_status_next_step(locale, &status_command)]; if let Some(command) = service::current_service_inspect_command()? { @@ -2513,12 +2571,27 @@ fn install_next_steps( Ok(steps) } -fn cli_status_command(exe_path: impl AsRef, codex_home: &Path) -> String { - format!( - "{} --codex-home {} status", +fn cli_status_command( + exe_path: impl AsRef, + codex_home: &Path, + provider_override: Option<&str>, + profile_override: Option<&str>, +) -> String { + let mut parts = vec![ shell_quote(exe_path.as_ref().display().to_string()), - shell_quote(codex_home.display().to_string()) - ) + "--codex-home".to_string(), + shell_quote(codex_home.display().to_string()), + ]; + if let Some(provider) = provider_override { + parts.push("--provider".to_string()); + parts.push(shell_quote(provider.to_string())); + } + if let Some(profile) = profile_override { + parts.push("--profile".to_string()); + parts.push(shell_quote(profile.to_string())); + } + parts.push("status".to_string()); + parts.join(" ") } fn shell_quote(input: String) -> String { @@ -2750,6 +2823,7 @@ mod tests { use super::Locale; use super::RolloutScope; use super::detect_locale_from_sources; + use super::full_watch_rollout_scope; use super::inspect_sqlite_distribution; use super::install_next_steps; use super::localized_command; @@ -2764,6 +2838,7 @@ mod tests { use super::resolve_sqlite_path; use super::validate_provider_override; use super::validate_provider_override_args; + use super::watched_config_paths; use crate::service::ServiceManager; use anyhow::Result; use clap::FromArgMatches; @@ -2911,6 +2986,22 @@ model_provider = "local" Ok(()) } + #[test] + fn falls_back_to_root_config_when_selected_profile_is_missing() -> Result<()> { + let dir = tempfile::tempdir()?; + fs::write( + dir.path().join("config.toml"), + "model_provider = \"openai\"\nsqlite_home = \"root-state\"\nprofile = \"missing\"\n", + )?; + + let provider = read_provider_from_config(dir.path(), None)?; + let sqlite_path = resolve_sqlite_path(dir.path(), None)?; + + assert_eq!(provider, "openai"); + assert_eq!(sqlite_path, dir.path().join("root-state/state_5.sqlite")); + Ok(()) + } + #[test] fn defaults_to_openai_without_config() -> Result<()> { let dir = tempfile::tempdir()?; @@ -3214,6 +3305,35 @@ model_provider = "local" Ok(()) } + #[test] + fn watch_promotes_initial_rollout_scan_to_all_rows() { + assert_eq!( + full_watch_rollout_scope(RolloutScope::MismatchedRows), + RolloutScope::AllRows + ); + assert_eq!( + full_watch_rollout_scope(RolloutScope::None), + RolloutScope::None + ); + } + + #[test] + fn watch_tracks_selected_profile_config_file() -> Result<()> { + let dir = tempfile::tempdir()?; + fs::write( + dir.path().join("config.toml"), + "model_provider = \"openai\"\nprofile = \"work\"\n", + )?; + + let implicit = watched_config_paths(dir.path(), None); + let explicit = watched_config_paths(dir.path(), Some("other")); + + assert!(implicit.contains(&dir.path().join("config.toml"))); + assert!(implicit.contains(&dir.path().join("work.config.toml"))); + assert!(explicit.contains(&dir.path().join("other.config.toml"))); + Ok(()) + } + #[test] fn generates_launchd_plist() { let plist = crate::service::build_launchd_plist( @@ -3236,12 +3356,16 @@ model_provider = "local" Locale::ZhHans, PathBuf::from("/tmp/codex threadripper").as_path(), PathBuf::from("/tmp/codex home").as_path(), + Some("openai"), + Some("work"), ServiceManager::Launchd, )?; assert_eq!(steps.len(), 3); assert!(steps[0].contains("运行这条命令查看状态")); assert!( - steps[0].contains("'/tmp/codex threadripper' --codex-home '/tmp/codex home' status") + steps[0].contains( + "'/tmp/codex threadripper' --codex-home '/tmp/codex home' --provider openai --profile work status" + ) ); assert!(steps[1].contains("launchctl print")); assert!(steps[2].contains("tail -f")); diff --git a/src/service.rs b/src/service.rs index 08a9064..57e925a 100644 --- a/src/service.rs +++ b/src/service.rs @@ -1162,6 +1162,7 @@ enum ProcessStatus { #[cfg(test)] mod tests { + use super::build_linux_runner_script; use super::build_windows_startup_vbs; use super::installed_windows_startup_config_path; use anyhow::Result; @@ -1230,6 +1231,36 @@ mod tests { ); } + #[test] + fn linux_runner_script_preserves_profile_override() { + let script = build_linux_runner_script( + Path::new("/opt/codex threadripper/bin/codex-threadripper"), + Path::new("/home/user/.codex"), + None, + Some("work"), + Duration::from_millis(500), + ); + + assert!(script.contains("'/opt/codex threadripper/bin/codex-threadripper'")); + assert!(script.contains("--profile work")); + assert!(script.contains("watch --poll-interval-ms 500")); + } + + #[test] + fn windows_startup_vbs_preserves_profile_override() { + let script = build_windows_startup_vbs( + Path::new(r"C:\Tools\codex-threadripper.exe"), + Path::new(r"C:\Codex"), + None, + Some("work"), + Duration::from_millis(500), + ); + let command = shell_run_command(&script); + + assert!(command.contains("--profile work")); + assert!(command.ends_with("watch --poll-interval-ms 500")); + } + fn shell_run_command(script: &str) -> String { let line = script .lines() From 84979fb7820bac0cdb2b90dc4b7fd0fce16e4a7a Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Mon, 25 May 2026 00:14:08 +0800 Subject: [PATCH 3/8] test: make service next-step test portable --- src/main.rs | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/main.rs b/src/main.rs index 7c60def..6705f09 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3352,13 +3352,14 @@ model_provider = "local" #[test] fn builds_install_next_steps() -> Result<()> { + let manager = crate::service::current_manager(); let steps = install_next_steps( Locale::ZhHans, PathBuf::from("/tmp/codex threadripper").as_path(), PathBuf::from("/tmp/codex home").as_path(), Some("openai"), Some("work"), - ServiceManager::Launchd, + manager, )?; assert_eq!(steps.len(), 3); assert!(steps[0].contains("运行这条命令查看状态")); @@ -3367,8 +3368,11 @@ model_provider = "local" "'/tmp/codex threadripper' --codex-home '/tmp/codex home' --provider openai --profile work status" ) ); - assert!(steps[1].contains("launchctl print")); - assert!(steps[2].contains("tail -f")); + assert!(steps[1].contains(crate::service::manager_name(manager))); + match manager { + ServiceManager::WindowsStartup => assert!(steps[2].contains("Get-Content")), + _ => assert!(steps[2].contains("tail -f")), + } Ok(()) } From 1de9f496612dc4bc5367726f0ed72f66eb086f24 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Mon, 25 May 2026 00:27:53 +0800 Subject: [PATCH 4/8] fix: harden watch and CI coverage --- .github/workflows/ci.yml | 8 ++++++-- README.md | 4 ++-- src/main.rs | 33 ++++++++++++++++++++++++++++++++- 3 files changed, 40 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ac1221f..187cf68 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,7 +8,11 @@ on: jobs: rust: - runs-on: ubuntu-22.04 + strategy: + fail-fast: false + matrix: + os: [ubuntu-22.04, macos-latest, windows-latest] + runs-on: ${{ matrix.os }} steps: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable @@ -17,4 +21,4 @@ jobs: - run: cargo fmt --check - run: cargo test --locked - run: cargo clippy --locked --all-targets -- -D warnings - - run: python3 -m py_compile scripts/smoke_release.py + - run: python -m py_compile scripts/smoke_release.py diff --git a/README.md b/README.md index 594cbd2..cd099f7 100644 --- a/README.md +++ b/README.md @@ -118,7 +118,7 @@ codex-threadripper install-service `watch` 的默认轮询间隔是 500ms。`watch`、`print-service-config` 和 `install-service` 都支持 `--poll-interval-ms` 来自定义间隔。 -所有命令都支持 `--provider ` 强制指定目标桶，或 `--profile ` 按 Codex v2 profile 读取 `$CODEX_HOME/.config.toml` 里的 `model_provider` / `sqlite_home`。如果没有同名 profile 文件，也会兼容旧的 `[profiles.]` provider 配置。 +所有命令都支持 `--provider ` 强制指定目标桶，或 `--profile ` 按 Codex v2 profile 读取 `$CODEX_HOME/.config.toml` 里的 `model_provider` / `sqlite_home`。Profile 名称只能包含 ASCII 字母、数字、`_` 或 `-`。如果没有同名 profile 文件，也会兼容旧的 `[profiles.]` provider 配置。 ## 它会改什么 @@ -235,7 +235,7 @@ These legacy command names still work: `watch` polls every 500ms by default. You can change that with `--poll-interval-ms` on `watch`, `print-service-config`, and `install-service`. -Every command accepts `--provider ` to force the target bucket, or `--profile ` to resolve `model_provider` / `sqlite_home` from Codex's v2 `$CODEX_HOME/.config.toml` profile. If that profile file is missing, legacy `[profiles.]` provider config is still supported. +Every command accepts `--provider ` to force the target bucket, or `--profile ` to resolve `model_provider` / `sqlite_home` from Codex's v2 `$CODEX_HOME/.config.toml` profile. Profile names must contain only ASCII letters, digits, `_`, or `-`. If that profile file is missing, legacy `[profiles.]` provider config is still supported. ## What it changes diff --git a/src/main.rs b/src/main.rs index 6705f09..7e4a26e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -53,6 +53,7 @@ const PROFILE_CONFIG_SUFFIX: &str = ".config.toml"; const DEFAULT_POLL_INTERVAL_MS: u64 = 500; const DEFAULT_BUCKET_PADDING_BYTES: usize = 256; const ROLLOUT_PROGRESS_INTERVAL: Duration = Duration::from_millis(500); +const WATCH_FULL_ROLLOUT_POLL_INTERVALS: u64 = 120; type ProviderDistribution = Vec<(String, u64)>; @@ -837,6 +838,7 @@ fn run_watch( ); let mut next_poll_deadline = Instant::now() + poll_interval; + let mut poll_count = 0_u64; while !shutdown.load(Ordering::Relaxed) { let timeout = next_poll_deadline.saturating_duration_since(Instant::now()); @@ -871,11 +873,12 @@ fn run_watch( eprintln!("{}", watcher_error_message(locale, err)); } Err(mpsc::RecvTimeoutError::Timeout) => { + poll_count = poll_count.wrapping_add(1); match reconcile_once( codex_home, provider_override.as_deref(), profile_override.as_deref(), - rollout_scope, + periodic_watch_rollout_scope(rollout_scope, poll_count), ) { Ok(summary) => { if last_provider.as_deref() != Some(summary.provider.as_str()) @@ -905,6 +908,13 @@ fn run_watch( Ok(()) } +fn periodic_watch_rollout_scope(rollout_scope: RolloutScope, poll_count: u64) -> RolloutScope { + if poll_count.is_multiple_of(WATCH_FULL_ROLLOUT_POLL_INTERVALS) { + return full_watch_rollout_scope(rollout_scope); + } + rollout_scope +} + fn full_watch_rollout_scope(rollout_scope: RolloutScope) -> RolloutScope { match rollout_scope { RolloutScope::None => RolloutScope::None, @@ -2822,6 +2832,7 @@ mod tests { use super::DEFAULT_POLL_INTERVAL_MS; use super::Locale; use super::RolloutScope; + use super::WATCH_FULL_ROLLOUT_POLL_INTERVALS; use super::detect_locale_from_sources; use super::full_watch_rollout_scope; use super::inspect_sqlite_distribution; @@ -2829,6 +2840,7 @@ mod tests { use super::localized_command; use super::parse_apple_languages; use super::parse_locale_tag; + use super::periodic_watch_rollout_scope; use super::read_provider_from_config; use super::reconcile_once; use super::reconcile_rollout_metadata_from_sqlite_with_progress; @@ -3317,6 +3329,25 @@ model_provider = "local" ); } + #[test] + fn watch_periodically_runs_full_rollout_scan() { + assert_eq!( + periodic_watch_rollout_scope(RolloutScope::MismatchedRows, 1), + RolloutScope::MismatchedRows + ); + assert_eq!( + periodic_watch_rollout_scope( + RolloutScope::MismatchedRows, + WATCH_FULL_ROLLOUT_POLL_INTERVALS + ), + RolloutScope::AllRows + ); + assert_eq!( + periodic_watch_rollout_scope(RolloutScope::None, WATCH_FULL_ROLLOUT_POLL_INTERVALS), + RolloutScope::None + ); + } + #[test] fn watch_tracks_selected_profile_config_file() -> Result<()> { let dir = tempfile::tempdir()?; From eddab43eb3bf60a8034634be11ee1421152df407 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Mon, 25 May 2026 00:35:43 +0800 Subject: [PATCH 5/8] test: make sqlite home fixtures portable --- src/main.rs | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/main.rs b/src/main.rs index 7e4a26e..56ea324 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3037,7 +3037,7 @@ model_provider = "local" let sqlite_home = dir.path().join("custom-state"); fs::write( dir.path().join("config.toml"), - format!("sqlite_home = \"{}\"\n", sqlite_home.display()), + sqlite_home_config(&sqlite_home), )?; let sqlite_path = resolve_sqlite_path(dir.path(), None)?; @@ -3120,7 +3120,7 @@ model_provider = "local" let sqlite_path = dir.path().join("state_5.sqlite"); fs::write( dir.path().join("config.toml"), - format!("sqlite_home = \"{}\"\n", dir.path().display()), + sqlite_home_config(dir.path()), )?; let err = reconcile_once(dir.path(), Some("openai"), None, RolloutScope::None).unwrap_err(); @@ -3450,6 +3450,15 @@ model_provider = "local" Ok(()) } + fn sqlite_home_config(path: &Path) -> String { + let path = path + .display() + .to_string() + .replace('\\', "\\\\") + .replace('"', "\\\""); + format!("sqlite_home = \"{path}\"\n") + } + fn assert_rollout_mtime(path: &Path, expected: FileTime) -> Result<()> { let actual = FileTime::from_last_modification_time(&fs::metadata(path)?); assert_eq!(actual.unix_seconds(), expected.unix_seconds()); From 2a55abba47da7180e5ae6511aeb3ed150d32b562 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Sun, 24 May 2026 22:56:54 +0800 Subject: [PATCH 6/8] test: modernize Codex SQLite fixtures --- scripts/smoke_release.py | 149 ++++++++++++++++++++++++++++++++++++--- src/main.rs | 96 +++++++++++++++++++++++-- 2 files changed, 228 insertions(+), 17 deletions(-) diff --git a/scripts/smoke_release.py b/scripts/smoke_release.py index ca6ad80..d5195b3 100644 --- a/scripts/smoke_release.py +++ b/scripts/smoke_release.py @@ -19,7 +19,10 @@ rollout_path TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, + created_at_ms INTEGER, + updated_at_ms INTEGER, source TEXT NOT NULL, + thread_source TEXT, model_provider TEXT NOT NULL, cwd TEXT NOT NULL, title TEXT NOT NULL, @@ -39,8 +42,54 @@ memory_mode TEXT NOT NULL DEFAULT 'enabled', model TEXT, reasoning_effort TEXT, - agent_path TEXT + agent_path TEXT, + preview TEXT NOT NULL DEFAULT '' ); +CREATE INDEX idx_threads_created_at ON threads(created_at DESC, id DESC); +CREATE INDEX idx_threads_updated_at ON threads(updated_at DESC, id DESC); +CREATE INDEX idx_threads_archived ON threads(archived); +CREATE INDEX idx_threads_source ON threads(source); +CREATE INDEX idx_threads_provider ON threads(model_provider); +CREATE INDEX idx_threads_created_at_ms ON threads(created_at_ms DESC, id DESC); +CREATE INDEX idx_threads_updated_at_ms ON threads(updated_at_ms DESC, id DESC); +CREATE INDEX idx_threads_archived_cwd_created_at_ms + ON threads(archived, cwd, created_at_ms DESC, id DESC); +CREATE INDEX idx_threads_archived_cwd_updated_at_ms + ON threads(archived, cwd, updated_at_ms DESC, id DESC); +CREATE TRIGGER threads_created_at_ms_after_insert +AFTER INSERT ON threads +WHEN NEW.created_at_ms IS NULL +BEGIN + UPDATE threads + SET created_at_ms = NEW.created_at * 1000 + WHERE id = NEW.id; +END; +CREATE TRIGGER threads_updated_at_ms_after_insert +AFTER INSERT ON threads +WHEN NEW.updated_at_ms IS NULL +BEGIN + UPDATE threads + SET updated_at_ms = NEW.updated_at * 1000 + WHERE id = NEW.id; +END; +CREATE TRIGGER threads_created_at_ms_after_update +AFTER UPDATE OF created_at ON threads +WHEN NEW.created_at != OLD.created_at + AND NEW.created_at_ms IS OLD.created_at_ms +BEGIN + UPDATE threads + SET created_at_ms = NEW.created_at * 1000 + WHERE id = NEW.id; +END; +CREATE TRIGGER threads_updated_at_ms_after_update +AFTER UPDATE OF updated_at ON threads +WHEN NEW.updated_at != OLD.updated_at + AND NEW.updated_at_ms IS OLD.updated_at_ms +BEGIN + UPDATE threads + SET updated_at_ms = NEW.updated_at * 1000 + WHERE id = NEW.id; +END; """ COMMAND_TIMEOUT_SECONDS = 30 @@ -141,14 +190,72 @@ def prepare_codex_home(codex_home: pathlib.Path) -> None: connection.executemany( """ INSERT INTO threads ( - id, rollout_path, created_at, updated_at, source, model_provider, cwd, title, - sandbox_policy, approval_mode - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + id, rollout_path, created_at, updated_at, created_at_ms, updated_at_ms, + source, thread_source, model_provider, cwd, title, sandbox_policy, approval_mode, + cli_version, model, reasoning_effort, first_user_message, preview + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ - ("1", "/tmp/a", 1, 1, "cli", "vm", "/tmp", "a", "workspace-write", "auto"), - ("2", "/tmp/b", 1, 1, "cli", "cp", "/tmp", "b", "workspace-write", "auto"), - ("3", "/tmp/c", 1, 1, "cli", "openai", "/tmp", "c", "workspace-write", "auto"), + ( + "1", + "/tmp/a", + 1, + 1, + 1000, + 1000, + "cli", + "user", + "vm", + "/tmp", + "a", + "workspace-write", + "auto", + "0.0.0-test", + "gpt-5-codex", + "medium", + "a", + "a", + ), + ( + "2", + "/tmp/b", + 1, + 1, + 1001, + 1001, + "cli", + "user", + "cp", + "/tmp", + "b", + "workspace-write", + "auto", + "0.0.0-test", + "gpt-5-codex", + "medium", + "b", + "b", + ), + ( + "3", + "/tmp/c", + 1, + 1, + 1002, + 1002, + "cli", + "user", + "openai", + "/tmp", + "c", + "workspace-write", + "auto", + "0.0.0-test", + "gpt-5-codex", + "medium", + "c", + "c", + ), ], ) connection.commit() @@ -274,11 +381,31 @@ def insert_dirty_row(sqlite_path: pathlib.Path) -> None: connection.execute( """ INSERT INTO threads ( - id, rollout_path, created_at, updated_at, source, model_provider, cwd, title, - sandbox_policy, approval_mode - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + id, rollout_path, created_at, updated_at, created_at_ms, updated_at_ms, + source, thread_source, model_provider, cwd, title, sandbox_policy, approval_mode, + cli_version, model, reasoning_effort, first_user_message, preview + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, - ("4", "/tmp/d", 1, 1, "cli", "vm", "/tmp", "d", "workspace-write", "auto"), + ( + "4", + "/tmp/d", + 1, + 1, + 1003, + 1003, + "cli", + "user", + "vm", + "/tmp", + "d", + "workspace-write", + "auto", + "0.0.0-test", + "gpt-5-codex", + "medium", + "d", + "d", + ), ) connection.commit() connection.close() diff --git a/src/main.rs b/src/main.rs index 56ea324..e58c5b3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3192,6 +3192,37 @@ model_provider = "local" Ok(()) } + #[test] + fn seed_sqlite_fixture_includes_current_codex_thread_columns() -> Result<()> { + let dir = tempfile::tempdir()?; + let sqlite_path = dir.path().join("state_5.sqlite"); + seed_sqlite(&sqlite_path)?; + let connection = Connection::open(&sqlite_path)?; + + let mut statement = connection.prepare("PRAGMA table_info(threads)")?; + let columns = statement + .query_map([], |row| row.get::<_, String>(1))? + .collect::>>()?; + for column in ["created_at_ms", "updated_at_ms", "thread_source", "preview"] { + assert!(columns.iter().any(|candidate| candidate == column)); + } + + let (created_at_ms, updated_at_ms, thread_source, preview): ( + i64, + i64, + String, + String, + ) = connection.query_row( + "SELECT created_at_ms, updated_at_ms, thread_source, preview FROM threads WHERE id = '1'", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + )?; + assert_eq!((created_at_ms, updated_at_ms), (1000, 1000)); + assert_eq!(thread_source, "user"); + assert_eq!(preview, "a"); + Ok(()) + } + #[test] fn durable_sync_updates_matching_rollout_session_meta() -> Result<()> { let dir = tempfile::tempdir()?; @@ -3416,7 +3447,10 @@ model_provider = "local" rollout_path TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, + created_at_ms INTEGER, + updated_at_ms INTEGER, source TEXT NOT NULL, + thread_source TEXT, model_provider TEXT NOT NULL, cwd TEXT NOT NULL, title TEXT NOT NULL, @@ -3436,15 +3470,65 @@ model_provider = "local" memory_mode TEXT NOT NULL DEFAULT 'enabled', model TEXT, reasoning_effort TEXT, - agent_path TEXT + agent_path TEXT, + preview TEXT NOT NULL DEFAULT '' ); + CREATE INDEX idx_threads_created_at ON threads(created_at DESC, id DESC); + CREATE INDEX idx_threads_updated_at ON threads(updated_at DESC, id DESC); + CREATE INDEX idx_threads_archived ON threads(archived); + CREATE INDEX idx_threads_source ON threads(source); + CREATE INDEX idx_threads_provider ON threads(model_provider); + CREATE INDEX idx_threads_created_at_ms ON threads(created_at_ms DESC, id DESC); + CREATE INDEX idx_threads_updated_at_ms ON threads(updated_at_ms DESC, id DESC); + CREATE INDEX idx_threads_archived_cwd_created_at_ms + ON threads(archived, cwd, created_at_ms DESC, id DESC); + CREATE INDEX idx_threads_archived_cwd_updated_at_ms + ON threads(archived, cwd, updated_at_ms DESC, id DESC); + CREATE TRIGGER threads_created_at_ms_after_insert + AFTER INSERT ON threads + WHEN NEW.created_at_ms IS NULL + BEGIN + UPDATE threads + SET created_at_ms = NEW.created_at * 1000 + WHERE id = NEW.id; + END; + CREATE TRIGGER threads_updated_at_ms_after_insert + AFTER INSERT ON threads + WHEN NEW.updated_at_ms IS NULL + BEGIN + UPDATE threads + SET updated_at_ms = NEW.updated_at * 1000 + WHERE id = NEW.id; + END; + CREATE TRIGGER threads_created_at_ms_after_update + AFTER UPDATE OF created_at ON threads + WHEN NEW.created_at != OLD.created_at + AND NEW.created_at_ms IS OLD.created_at_ms + BEGIN + UPDATE threads + SET created_at_ms = NEW.created_at * 1000 + WHERE id = NEW.id; + END; + CREATE TRIGGER threads_updated_at_ms_after_update + AFTER UPDATE OF updated_at ON threads + WHEN NEW.updated_at != OLD.updated_at + AND NEW.updated_at_ms IS OLD.updated_at_ms + BEGIN + UPDATE threads + SET updated_at_ms = NEW.updated_at * 1000 + WHERE id = NEW.id; + END; INSERT INTO threads ( - id, rollout_path, created_at, updated_at, source, model_provider, cwd, title, - sandbox_policy, approval_mode + id, rollout_path, created_at, updated_at, created_at_ms, updated_at_ms, + source, thread_source, model_provider, cwd, title, sandbox_policy, + approval_mode, cli_version, model, reasoning_effort, first_user_message, preview ) VALUES - ('1', '/tmp/a', 1, 1, 'cli', 'vm', '/tmp', 'a', 'workspace-write', 'auto'), - ('2', '/tmp/b', 1, 1, 'cli', 'cp', '/tmp', 'b', 'workspace-write', 'auto'), - ('3', '/tmp/c', 1, 1, 'cli', 'openai', '/tmp', 'c', 'workspace-write', 'auto'); + ('1', '/tmp/a', 1, 1, 1000, 1000, 'cli', 'user', 'vm', '/tmp', 'a', + 'workspace-write', 'auto', '0.0.0-test', 'gpt-5-codex', 'medium', 'a', 'a'), + ('2', '/tmp/b', 1, 1, 1001, 1001, 'cli', 'user', 'cp', '/tmp', 'b', + 'workspace-write', 'auto', '0.0.0-test', 'gpt-5-codex', 'medium', 'b', 'b'), + ('3', '/tmp/c', 1, 1, 1002, 1002, 'cli', 'user', 'openai', '/tmp', 'c', + 'workspace-write', 'auto', '0.0.0-test', 'gpt-5-codex', 'medium', 'c', 'c'); ", )?; Ok(()) From 34589d4ed1ac5b9d2c3af393f9cb22953c424f03 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Mon, 25 May 2026 00:08:18 +0800 Subject: [PATCH 7/8] test: cover rollout rewrites in smoke fixture --- scripts/smoke_release.py | 80 +++++++++++++++++++++++++++++----------- 1 file changed, 58 insertions(+), 22 deletions(-) diff --git a/scripts/smoke_release.py b/scripts/smoke_release.py index d5195b3..e02f811 100644 --- a/scripts/smoke_release.py +++ b/scripts/smoke_release.py @@ -4,6 +4,7 @@ import os import pathlib import shutil +import signal import sqlite3 import subprocess import sys @@ -125,6 +126,9 @@ def main() -> int: print(f"[smoke] sync passed with backup: {backup_path}") assert_backup_contains_dirty_rows(backup_path) print("[smoke] backup verification passed") + assert_rollout_provider(codex_home, "rollout-a.jsonl", "1", "openai") + assert_rollout_provider(codex_home, "rollout-b.jsonl", "2", "openai") + print("[smoke] rollout metadata verification passed") assert_status(binary_path, codex_home, expected_total=3, expected_mismatched=0) print("[smoke] post-sync status passed") run_service_install(binary_path, codex_home) @@ -134,10 +138,11 @@ def main() -> int: binary_path, codex_home, expected_installed="yes", expected_running="yes" ) print("[smoke] service reached installed=yes running=yes") - insert_dirty_row(codex_home / "state_5.sqlite") - write_new_session(codex_home) + rollout_path = write_new_session(codex_home) + insert_dirty_row(codex_home / "state_5.sqlite", rollout_path) print("[smoke] inserted dirty row and wrote new session") wait_for_reconcile(binary_path, codex_home) + assert_rollout_provider(codex_home, "rollout-d.jsonl", "4", "openai") print("[smoke] background reconcile passed") finally: run_service_uninstall(binary_path, codex_home) @@ -175,14 +180,12 @@ def prepare_codex_home(codex_home: pathlib.Path) -> None: sessions_dir = codex_home / "sessions" / "2026" / "04" / "15" sessions_dir.mkdir(parents=True, exist_ok=True) - (sessions_dir / "rollout-a.jsonl").write_text( - '{"type":"session_meta","payload":{"model_provider":"vm"}}\n', - encoding="utf-8", - ) - (sessions_dir / "rollout-b.jsonl").write_text( - '{"type":"session_meta","payload":{"model_provider":"cp"}}\n', - encoding="utf-8", - ) + rollout_a = sessions_dir / "rollout-a.jsonl" + rollout_b = sessions_dir / "rollout-b.jsonl" + rollout_c = sessions_dir / "rollout-c.jsonl" + write_rollout(rollout_a, "1", "vm") + write_rollout(rollout_b, "2", "cp") + write_rollout(rollout_c, "3", "openai") sqlite_path = codex_home / "state_5.sqlite" connection = sqlite3.connect(sqlite_path) @@ -198,7 +201,7 @@ def prepare_codex_home(codex_home: pathlib.Path) -> None: [ ( "1", - "/tmp/a", + str(rollout_a), 1, 1, 1000, @@ -218,7 +221,7 @@ def prepare_codex_home(codex_home: pathlib.Path) -> None: ), ( "2", - "/tmp/b", + str(rollout_b), 1, 1, 1001, @@ -238,7 +241,7 @@ def prepare_codex_home(codex_home: pathlib.Path) -> None: ), ( "3", - "/tmp/c", + str(rollout_c), 1, 1, 1002, @@ -262,6 +265,13 @@ def prepare_codex_home(codex_home: pathlib.Path) -> None: connection.close() +def write_rollout(path: pathlib.Path, thread_id: str, provider: str) -> None: + path.write_text( + f'{{"type":"session_meta","payload":{{"id":"{thread_id}","model_provider":"{provider}"}}}}\n', + encoding="utf-8", + ) + + def command_env() -> dict[str, str]: env = os.environ.copy() env["CODEX_THREADRIPPER_LANG"] = "en" @@ -320,6 +330,8 @@ def run_sync(binary_path: pathlib.Path, codex_home: pathlib.Path) -> pathlib.Pat output = run_command(binary_path, codex_home, "sync") if "Rows updated: 2" not in output: raise RuntimeError(f"unexpected sync output\n\n{output}") + if "Rollouts updated: 2" not in output: + raise RuntimeError(f"sync did not rewrite rollout metadata\n\n{output}") backup_line = next( (line for line in output.splitlines() if line.startswith("Backup: ")), @@ -330,9 +342,34 @@ def run_sync(binary_path: pathlib.Path, codex_home: pathlib.Path) -> pathlib.Pat backup_path = pathlib.Path(backup_line.removeprefix("Backup: ").strip()) if not backup_path.exists(): raise RuntimeError(f"backup file is missing: {backup_path}") + journal_line = next( + (line for line in output.splitlines() if line.startswith("Rollout first-line journal: ")), + None, + ) + if journal_line is None: + raise RuntimeError(f"missing rollout journal line in sync output\n\n{output}") + journal_path = pathlib.Path(journal_line.removeprefix("Rollout first-line journal: ").strip()) + if not journal_path.exists(): + raise RuntimeError(f"rollout journal file is missing: {journal_path}") + journal = journal_path.read_text(encoding="utf-8") + if '"thread_id":"1"' not in journal or '"thread_id":"2"' not in journal: + raise RuntimeError(f"rollout journal did not include rewritten threads\n\n{journal}") return backup_path +def assert_rollout_provider( + codex_home: pathlib.Path, rollout_name: str, thread_id: str, provider: str +) -> None: + rollout_path = codex_home / "sessions" / "2026" / "04" / "15" / rollout_name + first_line = rollout_path.read_text(encoding="utf-8").splitlines()[0] + expected_id = f'"id":"{thread_id}"' + expected_provider = f'"model_provider":"{provider}"' + if expected_id not in first_line or expected_provider not in first_line: + raise RuntimeError( + f"unexpected rollout first line for {rollout_path}\n\n{first_line}" + ) + + def assert_backup_contains_dirty_rows(backup_path: pathlib.Path) -> None: connection = sqlite3.connect(backup_path) mismatched = connection.execute( @@ -376,7 +413,7 @@ def wait_for_service_status( ) -def insert_dirty_row(sqlite_path: pathlib.Path) -> None: +def insert_dirty_row(sqlite_path: pathlib.Path, rollout_path: pathlib.Path) -> None: connection = sqlite3.connect(sqlite_path) connection.execute( """ @@ -386,9 +423,9 @@ def insert_dirty_row(sqlite_path: pathlib.Path) -> None: cli_version, model, reasoning_effort, first_user_message, preview ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, - ( - "4", - "/tmp/d", + ( + "4", + str(rollout_path), 1, 1, 1003, @@ -411,13 +448,12 @@ def insert_dirty_row(sqlite_path: pathlib.Path) -> None: connection.close() -def write_new_session(codex_home: pathlib.Path) -> None: +def write_new_session(codex_home: pathlib.Path) -> pathlib.Path: sessions_dir = codex_home / "sessions" / "2026" / "04" / "15" sessions_dir.mkdir(parents=True, exist_ok=True) - (sessions_dir / "rollout-c.jsonl").write_text( - '{"type":"session_meta","payload":{"model_provider":"vm"}}\n', - encoding="utf-8", - ) + rollout_path = sessions_dir / "rollout-d.jsonl" + write_rollout(rollout_path, "4", "vm") + return rollout_path def wait_for_reconcile(binary_path: pathlib.Path, codex_home: pathlib.Path) -> None: From e8156ceee2c6f9300f935ba103f8763a4d053352 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Mon, 25 May 2026 00:29:18 +0800 Subject: [PATCH 8/8] test: assert sqlite fixture indexes and triggers --- src/main.rs | 46 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/src/main.rs b/src/main.rs index e58c5b3..080357e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3206,6 +3206,31 @@ model_provider = "local" for column in ["created_at_ms", "updated_at_ms", "thread_source", "preview"] { assert!(columns.iter().any(|candidate| candidate == column)); } + let mut statement = connection.prepare("PRAGMA index_list(threads)")?; + let indexes = statement + .query_map([], |row| row.get::<_, String>(1))? + .collect::>>()?; + for index in [ + "idx_threads_created_at_ms", + "idx_threads_updated_at_ms", + "idx_threads_archived_cwd_created_at_ms", + "idx_threads_archived_cwd_updated_at_ms", + ] { + assert!(indexes.iter().any(|candidate| candidate == index)); + } + let mut statement = + connection.prepare("SELECT name FROM sqlite_master WHERE type = 'trigger'")?; + let triggers = statement + .query_map([], |row| row.get::<_, String>(0))? + .collect::>>()?; + for trigger in [ + "threads_created_at_ms_after_insert", + "threads_updated_at_ms_after_insert", + "threads_created_at_ms_after_update", + "threads_updated_at_ms_after_update", + ] { + assert!(triggers.iter().any(|candidate| candidate == trigger)); + } let (created_at_ms, updated_at_ms, thread_source, preview): ( i64, @@ -3220,6 +3245,27 @@ model_provider = "local" assert_eq!((created_at_ms, updated_at_ms), (1000, 1000)); assert_eq!(thread_source, "user"); assert_eq!(preview, "a"); + + connection.execute( + " + INSERT INTO threads ( + id, rollout_path, created_at, updated_at, source, thread_source, + model_provider, cwd, title, sandbox_policy, approval_mode, cli_version, + model, reasoning_effort, first_user_message, preview + ) VALUES ( + 'trigger-test', '/tmp/trigger-test', 2, 3, 'cli', 'user', + 'openai', '/tmp', 'trigger-test', 'workspace-write', 'auto', + '0.0.0-test', 'gpt-5-codex', 'medium', 'trigger-test', 'trigger-test' + ) + ", + [], + )?; + let trigger_ms: (i64, i64) = connection.query_row( + "SELECT created_at_ms, updated_at_ms FROM threads WHERE id = 'trigger-test'", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + )?; + assert_eq!(trigger_ms, (2000, 3000)); Ok(()) }