diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..187cf68 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,24 @@ +name: CI + +on: + pull_request: + push: + branches: + - main + +jobs: + rust: + 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 + with: + components: rustfmt, clippy + - run: cargo fmt --check + - run: cargo test --locked + - run: cargo clippy --locked --all-targets -- -D warnings + - run: python -m py_compile scripts/smoke_release.py diff --git a/README.md b/README.md index 02db215..cd099f7 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 名称只能包含 ASCII 字母、数字、`_` 或 `-`。如果没有同名 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. 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 - `sync` writes a backup next to the SQLite state DB and then updates the `model_provider` column in `state_5.sqlite` diff --git a/scripts/smoke_release.py b/scripts/smoke_release.py index ca6ad80..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 @@ -19,7 +20,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 +43,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 @@ -76,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) @@ -85,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) @@ -126,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) @@ -141,20 +193,85 @@ 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", + str(rollout_a), + 1, + 1, + 1000, + 1000, + "cli", + "user", + "vm", + "/tmp", + "a", + "workspace-write", + "auto", + "0.0.0-test", + "gpt-5-codex", + "medium", + "a", + "a", + ), + ( + "2", + str(rollout_b), + 1, + 1, + 1001, + 1001, + "cli", + "user", + "cp", + "/tmp", + "b", + "workspace-write", + "auto", + "0.0.0-test", + "gpt-5-codex", + "medium", + "b", + "b", + ), + ( + "3", + str(rollout_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() 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" @@ -213,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: ")), @@ -223,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( @@ -269,28 +413,47 @@ 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( """ 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", + str(rollout_path), + 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() -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: diff --git a/src/main.rs b/src/main.rs index 697314c..080357e 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,9 +49,11 @@ 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); +const WATCH_FULL_ROLLOUT_POLL_INTERVALS: u64 = 120; type ProviderDistribution = Vec<(String, u64)>; @@ -83,6 +86,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, } @@ -167,10 +173,28 @@ enum BucketCommand { }, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Default, Deserialize)] struct ConfigToml { model_provider: Option, sqlite_home: Option, + profile: Option, + profiles: Option>, +} + +#[derive(Debug, Deserialize)] +struct ConfigProfileToml { + model_provider: Option, +} + +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)] @@ -245,6 +269,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, @@ -288,16 +323,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"); @@ -334,11 +369,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 +392,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 +400,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 +416,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 +432,7 @@ fn main() -> Result<()> { locale, &codex_home, cli.provider.clone(), + cli.profile.clone(), if sqlite_only { RolloutScope::None } else { @@ -406,6 +447,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 +457,7 @@ fn main() -> Result<()> { locale, &codex_home, cli.provider.as_deref(), + cli.profile.as_deref(), Duration::from_millis(poll_interval_ms), )?; } @@ -440,6 +483,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 +520,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 +577,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 +775,7 @@ fn run_watch( locale: Locale, codex_home: &Path, provider_override: Option, + profile_override: Option, rollout_scope: RolloutScope, poll_interval: Duration, ) -> Result<()> { @@ -717,6 +795,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 || { @@ -738,7 +818,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(), + full_rollout_scope, + ) { Ok(summary) => { print_sync_summary(locale, watch_started_title(locale), &summary); last_provider = Some(summary.provider.clone()); @@ -753,13 +838,19 @@ 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()); 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) { + if touches_config_file(&event, watched_paths.as_slice()) { + match reconcile_once( + codex_home, + provider_override.as_deref(), + profile_override.as_deref(), + full_rollout_scope, + ) { Ok(summary) => { if last_provider.as_deref() != Some(summary.provider.as_str()) || summary.changed_rows > 0 @@ -768,6 +859,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)); @@ -780,7 +873,13 @@ fn run_watch( eprintln!("{}", watcher_error_message(locale, err)); } Err(mpsc::RecvTimeoutError::Timeout) => { - match reconcile_once(codex_home, provider_override.as_deref(), rollout_scope) { + poll_count = poll_count.wrapping_add(1); + match reconcile_once( + codex_home, + provider_override.as_deref(), + profile_override.as_deref(), + periodic_watch_rollout_scope(rollout_scope, poll_count), + ) { Ok(summary) => { if last_provider.as_deref() != Some(summary.provider.as_str()) || summary.changed_rows > 0 @@ -809,19 +908,55 @@ fn run_watch( Ok(()) } -fn touches_config_file(event: ¬ify::Event, config_path: &Path) -> bool { +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, + 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(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 +977,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 +1030,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 +1047,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 +1095,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 +1116,90 @@ fn resolve_sqlite_path(codex_home: &Path) -> Result { .join(STATE_DB_FILENAME)) } -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, - }); +fn read_effective_codex_config( + codex_home: &Path, + profile_override: Option<&str>, +) -> Result { + let mut config = read_codex_config(codex_home)?; + let explicit_profile = profile_override.and_then(trimmed_string); + let config_profile = config.profile.as_deref().and_then(trimmed_string); + + 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)?; } - 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())) + 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()) +} + +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> { + 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())) +} + +fn profile_config_path(codex_home: &Path, profile: &str) -> PathBuf { + codex_home.join(format!("{profile}{PROFILE_CONFIG_SUFFIX}")) +} + +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 +1406,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 +1903,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,13 +1911,21 @@ fn install_service( exe_path.as_path(), codex_home, provider_override, + profile_override, poll_interval, )?; 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(()) @@ -1712,7 +1943,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 { @@ -1786,16 +2017,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", @@ -1920,10 +2151,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" } } } @@ -1963,6 +2194,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 +2229,17 @@ 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. Profile names may contain ASCII letters, digits, '_' or '-'." + } + Locale::ZhHans => { + "优先从 Codex profile 配置解析 model_provider 和 sqlite_home。profile 名称可包含 ASCII 字母、数字、'_' 或 '-'。" + } + } +} + fn provider_empty_error(locale: Locale) -> &'static str { match locale { Locale::En => "provider must contain at least one non-whitespace character", @@ -1998,6 +2247,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", @@ -2298,9 +2563,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()? { @@ -2313,12 +2581,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 { @@ -2549,12 +2832,15 @@ 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; use super::install_next_steps; 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; @@ -2564,6 +2850,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; @@ -2652,15 +2939,85 @@ 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 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()?; - let provider = read_provider_from_config(dir.path())?; + let provider = read_provider_from_config(dir.path(), None)?; assert_eq!(provider, "openai"); Ok(()) } @@ -2669,7 +3026,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(()) } @@ -2680,10 +3037,10 @@ mod tests { 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())?; + let sqlite_path = resolve_sqlite_path(dir.path(), None)?; assert_eq!(sqlite_path, sqlite_home.join("state_5.sqlite")); Ok(()) @@ -2731,16 +3088,42 @@ 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()?; 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"), 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()); @@ -2809,6 +3192,83 @@ mod tests { 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 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, + 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"); + + 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(()) + } + #[test] fn durable_sync_updates_matching_rollout_session_meta() -> Result<()> { let dir = tempfile::tempdir()?; @@ -2934,35 +3394,93 @@ mod tests { 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_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()?; + 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( 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")); } #[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(), - ServiceManager::Launchd, + Some("openai"), + Some("work"), + manager, )?; 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")); + 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(()) } @@ -2975,7 +3493,10 @@ mod tests { 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, @@ -2995,20 +3516,79 @@ mod tests { 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(()) } + 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()); diff --git a/src/service.rs b/src/service.rs index 3dd3c82..57e925a 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") @@ -1084,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; @@ -1112,6 +1191,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 +1216,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); @@ -1150,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()