diff --git a/codex-rs/cli/src/concurrent/mod.rs b/codex-rs/cli/src/concurrent/mod.rs index 4a6f676b71..b34bb45162 100644 --- a/codex-rs/cli/src/concurrent/mod.rs +++ b/codex-rs/cli/src/concurrent/mod.rs @@ -1,13 +1,31 @@ +use std::fs::File; use std::io::Write; use std::path::Path; use std::path::PathBuf; use std::process::Command; use std::process::Stdio; +use std::sync::OnceLock; + +use tokio::process::Command as TokioCommand; +use tokio::sync::Semaphore; use anyhow::Context; use codex_common::CliConfigOverrides; use codex_exec::Cli as ExecCli; +// Serialize git worktree add operations across tasks to avoid repository lock contention. +static GIT_WORKTREE_ADD_SEMAPHORE: OnceLock = OnceLock::new(); + +#[derive(Debug, Clone)] +pub struct ConcurrentRunResult { + pub branch: String, + pub worktree_dir: PathBuf, + pub log_file: Option, + pub exec_exit_code: Option, + pub _had_changes: bool, + pub _applied_changes: Option, +} + fn compute_codex_home() -> PathBuf { if let Ok(val) = std::env::var("CODEX_HOME") { if !val.is_empty() { @@ -76,15 +94,6 @@ fn git_output(repo_dir: &Path, args: &[&str]) -> anyhow::Result { Ok(String::from_utf8_lossy(&out.stdout).trim().to_string()) } -fn git_status_ok(repo_dir: &Path, args: &[&str]) -> anyhow::Result { - let status = Command::new("git") - .args(args) - .current_dir(repo_dir) - .status() - .with_context(|| format!("running git {args:?}"))?; - Ok(status.success()) -} - fn git_capture_stdout(repo_dir: &Path, args: &[&str]) -> anyhow::Result> { let out = Command::new("git") .args(args) @@ -119,7 +128,8 @@ pub async fn run_concurrent_flow( cli_config_overrides: CliConfigOverrides, codex_linux_sandbox_exe: Option, automerge: bool, -) -> anyhow::Result<()> { + quiet: bool, +) -> anyhow::Result { let cwd = std::env::current_dir()?; // Ensure we are in a git repo and find repo root. @@ -146,76 +156,124 @@ pub async fn run_concurrent_flow( .file_name() .map(|s| s.to_string_lossy().to_string()) .unwrap_or_else(|| "repo".to_string()); - codex_home.push(repo_name); + codex_home.push(repo_name.clone()); - // Prepare branch name: codex/ + // Prepare branch name: codex/, retrying with a numeric suffix to avoid races. let slug = slugify_prompt(&prompt, 64); - let mut branch = format!("codex/{slug}"); - - // Ensure uniqueness if branch already exists + let mut branch: String; + let worktree_dir: PathBuf; let mut attempt: u32 = 1; - while git_status_ok(&repo_root, &["rev-parse", "--verify", &branch]).unwrap_or(false) { + loop { + branch = if attempt == 1 { + format!("codex/{slug}") + } else { + format!("codex/{slug}-{attempt}") + }; + + let mut candidate_dir = codex_home.clone(); + candidate_dir.push(&branch); + + // Create parent directories for candidate path. + if let Some(parent) = candidate_dir.parent() { + std::fs::create_dir_all(parent)?; + } + + if !quiet { + println!( + "Creating worktree at {} with branch {}", + candidate_dir.display(), + branch + ); + } + + // Try to add worktree with new branch from current HEAD + let worktree_path_str = candidate_dir.to_string_lossy().to_string(); + let add_status = Command::new("git") + .arg("worktree") + .arg("add") + .arg("-b") + .arg(&branch) + .arg(&worktree_path_str) + .current_dir(&repo_root) + .status()?; + if add_status.success() { + worktree_dir = candidate_dir; + break; + } + attempt += 1; - branch = format!("codex/{slug}-{attempt}"); + if attempt > 50 { + anyhow::bail!("Failed to create git worktree after multiple attempts"); + } + // Retry with a new branch name. } - // Worktree directory path - let worktree_dir = { - let mut p = codex_home.clone(); - p.push(&branch); - p - }; + // Either run codex exec inline (verbose) or as a subprocess with logs redirected. + let mut log_file: Option = None; + let mut exec_exit_code: Option = None; + if quiet { + let exe = std::env::current_exe() + .map_err(|e| anyhow::anyhow!("failed to locate current executable: {e}"))?; - // Create parent directories - if let Some(parent) = worktree_dir.parent() { - std::fs::create_dir_all(parent)?; - } + // Prepare logs directory: $CODEX_HOME/logs/ + let mut logs_dir = compute_codex_home(); + logs_dir.push("logs"); + logs_dir.push(&repo_name); + std::fs::create_dir_all(&logs_dir)?; - println!( - "Creating worktree at {} with branch {}", - worktree_dir.display(), - branch - ); + let sanitized_branch = branch.replace('/', "_"); + let log_path = logs_dir.join(format!("{sanitized_branch}.log")); + let log_f = File::create(&log_path)?; + log_file = Some(log_path.clone()); - // Add worktree with new branch from current HEAD - let worktree_path_str = worktree_dir.to_string_lossy().to_string(); - let add_status = Command::new("git") - .arg("worktree") - .arg("add") - .arg("-b") - .arg(&branch) - .arg(&worktree_path_str) - .current_dir(&repo_root) - .status()?; - if !add_status.success() { - anyhow::bail!("Failed to create git worktree"); - } + let mut cmd = Command::new(exe); + cmd.arg("exec") + .arg("--full-auto") + .arg("--cd") + .arg(worktree_dir.as_os_str()) + .stdout(Stdio::from(log_f.try_clone()?)) + .stderr(Stdio::from(log_f)); - // Build an ExecCli to run in full-auto mode at the worktree directory. - let mut exec_cli = ExecCli { - images: vec![], - model: None, - sandbox_mode: None, - config_profile: None, - full_auto: true, - dangerously_bypass_approvals_and_sandbox: false, - cwd: Some(worktree_dir.clone()), - skip_git_repo_check: false, - config_overrides: CliConfigOverrides::default(), - color: Default::default(), - json: false, - last_message_file: None, - prompt: Some(prompt.clone()), - }; + // Forward any root-level config overrides. + for ov in cli_config_overrides.raw_overrides.iter() { + cmd.arg("-c").arg(ov); + } - // Prepend any root-level config overrides. - super::prepend_config_flags(&mut exec_cli.config_overrides, cli_config_overrides); + // Append the prompt last (positional argument). + cmd.arg(&prompt); - // Run codex exec - if let Err(e) = codex_exec::run_main(exec_cli, codex_linux_sandbox_exe).await { - eprintln!("codex exec failed: {e}"); - // Do not attempt to bring changes on failure; leave worktree for inspection. - return Err(e); + let status = cmd.status()?; + exec_exit_code = status.code(); + if !status.success() && !quiet { + eprintln!("codex exec failed with exit code {exec_exit_code:?}"); + } + } else { + // Build an ExecCli to run in full-auto mode at the worktree directory. + let mut exec_cli = ExecCli { + images: vec![], + model: None, + sandbox_mode: None, + config_profile: None, + full_auto: true, + dangerously_bypass_approvals_and_sandbox: false, + cwd: Some(worktree_dir.clone()), + skip_git_repo_check: false, + config_overrides: CliConfigOverrides::default(), + color: Default::default(), + json: false, + last_message_file: None, + prompt: Some(prompt.clone()), + }; + + // Prepend any root-level config overrides. + super::prepend_config_flags(&mut exec_cli.config_overrides, cli_config_overrides); + + // Run codex exec + if let Err(e) = codex_exec::run_main(exec_cli, codex_linux_sandbox_exe).await { + eprintln!("codex exec failed: {e}"); + // Do not attempt to bring changes on failure; leave worktree for inspection. + return Err(e); + } } // Auto-commit changes in the worktree if any @@ -242,30 +300,43 @@ pub async fn run_concurrent_flow( .status()? .success() { - eprintln!("No commit created (maybe no changes)"); - } else { + if !quiet { + eprintln!("No commit created (maybe no changes)"); + } + } else if !quiet { println!("Committed changes in worktree branch {branch}"); } - } else { + } else if !quiet { println!("No changes detected in worktree; skipping commit."); } if !automerge { - println!( - "Auto-merge disabled; leaving changes in worktree {} on branch {}.", - worktree_dir.display(), - branch - ); - println!( - "You can review and manually merge from that branch into {current_branch} when ready." - ); - println!("Summary: Auto-merge disabled."); - return Ok(()); + if !quiet { + println!( + "Auto-merge disabled; leaving changes in worktree {} on branch {}.", + worktree_dir.display(), + branch + ); + println!( + "You can review and manually merge from that branch into {current_branch} when ready." + ); + println!("Summary: Auto-merge disabled."); + } + return Ok(ConcurrentRunResult { + branch, + worktree_dir, + log_file, + exec_exit_code, + _had_changes: had_changes, + _applied_changes: None, + }); } // Bring the changes into the main working tree as UNSTAGED modifications. // We generate a patch from the original HEAD to the worktree branch tip, then apply with 3-way merge. - println!("Applying changes from {branch} onto {current_branch} as unstaged modifications"); + if !quiet { + println!("Applying changes from {branch} onto {current_branch} as unstaged modifications"); + } let range = format!("{original_head}..{branch}"); let mut diff_bytes = git_capture_stdout(&repo_root, &["diff", "--binary", "--full-index", &range])?; @@ -280,8 +351,17 @@ pub async fn run_concurrent_flow( } if diff_bytes.is_empty() { - println!("Summary: 0 changes detected."); - return Ok(()); + if !quiet { + println!("Summary: 0 changes detected."); + } + return Ok(ConcurrentRunResult { + branch, + worktree_dir, + log_file, + exec_exit_code, + _had_changes: had_changes, + _applied_changes: Some(0), + }); } let changed_files = count_files_in_patch(&diff_bytes); @@ -302,21 +382,27 @@ pub async fn run_concurrent_flow( } let status = child.wait().context("waiting for git apply")?; if !status.success() { - eprintln!( - "Applying changes failed. You can manually inspect {} and apply diffs.", - worktree_dir.display() - ); - println!("Summary: Apply failed."); + if !quiet { + eprintln!( + "Applying changes failed. You can manually inspect {} and apply diffs.", + worktree_dir.display() + ); + println!("Summary: Apply failed."); + } } else { - println!("Changes applied to working tree (unstaged)."); - println!("Summary: Applied {changed_files} files changed."); + if !quiet { + println!("Changes applied to working tree (unstaged)."); + println!("Summary: Applied {changed_files} files changed."); + } // Cleanup: remove the worktree and delete the temporary branch. - println!( - "Cleaning up worktree {} and branch {}", - worktree_dir.display(), - branch - ); + if !quiet { + println!( + "Cleaning up worktree {} and branch {}", + worktree_dir.display(), + branch + ); + } let worktree_path_str = worktree_dir.to_string_lossy().to_string(); let remove_status = Command::new("git") .args(["worktree", "remove", &worktree_path_str]) @@ -327,7 +413,9 @@ pub async fn run_concurrent_flow( // removed } _ => { - eprintln!("git worktree remove failed; retrying with --force"); + if !quiet { + eprintln!("git worktree remove failed; retrying with --force"); + } let _ = Command::new("git") .args(["worktree", "remove", "--force", &worktree_path_str]) .current_dir(&repo_root) @@ -340,13 +428,157 @@ pub async fn run_concurrent_flow( .current_dir(&repo_root) .status(); if let Ok(s) = del_status { - if !s.success() { + if !s.success() && !quiet { eprintln!("Failed to delete branch {branch}"); } - } else { + } else if !quiet { eprintln!("Error running git branch -D {branch}"); } } - Ok(()) + Ok(ConcurrentRunResult { + branch, + worktree_dir, + log_file, + exec_exit_code, + _had_changes: had_changes, + _applied_changes: Some(changed_files), + }) +} + +/// A Send-friendly variant used for best-of-n: run quietly (logs redirected) and do not auto-merge. +/// This intentionally avoids referencing non-Send types from codex-exec. +pub async fn run_concurrent_flow_quiet_no_automerge( + prompt: String, + cli_config_overrides: CliConfigOverrides, + _codex_linux_sandbox_exe: Option, +) -> anyhow::Result { + let cwd = std::env::current_dir()?; + + let repo_root_str = git_output(&cwd, &["rev-parse", "--show-toplevel"]); + let repo_root = match repo_root_str { + Ok(p) => PathBuf::from(p), + Err(err) => { + eprintln!("Not inside a Git repo: {err}"); + std::process::exit(1); + } + }; + + // Capture basic repo info (not used further in quiet/no-automerge flow). + + let mut codex_home = compute_codex_home(); + codex_home.push("worktrees"); + let repo_name = repo_root + .file_name() + .map(|s| s.to_string_lossy().to_string()) + .unwrap_or_else(|| "repo".to_string()); + codex_home.push(repo_name.clone()); + + let slug = slugify_prompt(&prompt, 64); + let mut branch: String; + let worktree_dir: PathBuf; + // Serialize worktree creation to avoid git repo lock contention across tasks. + { + let semaphore = GIT_WORKTREE_ADD_SEMAPHORE.get_or_init(|| Semaphore::new(1)); + let _permit = semaphore.acquire().await.expect("semaphore closed"); + + let mut attempt: u32 = 1; + loop { + branch = if attempt == 1 { + format!("codex/{slug}") + } else { + format!("codex/{slug}-{attempt}") + }; + + let mut candidate_dir = codex_home.clone(); + candidate_dir.push(&branch); + + if let Some(parent) = candidate_dir.parent() { + std::fs::create_dir_all(parent)?; + } + + let worktree_path_str = candidate_dir.to_string_lossy().to_string(); + let add_status = TokioCommand::new("git") + .arg("worktree") + .arg("add") + .arg("-b") + .arg(&branch) + .arg(&worktree_path_str) + .current_dir(&repo_root) + .status() + .await?; + if add_status.success() { + worktree_dir = candidate_dir; + break; + } + attempt += 1; + if attempt > 50 { + anyhow::bail!("Failed to create git worktree after multiple attempts"); + } + } + } + + // Run the CLI in quiet mode (logs redirected). + let exe = std::env::current_exe() + .map_err(|e| anyhow::anyhow!("failed to locate current executable: {e}"))?; + + let mut logs_dir = compute_codex_home(); + logs_dir.push("logs"); + logs_dir.push(&repo_name); + std::fs::create_dir_all(&logs_dir)?; + + let sanitized_branch = branch.replace('/', "_"); + let log_path = logs_dir.join(format!("{sanitized_branch}.log")); + let log_f = File::create(&log_path)?; + let log_file = Some(log_path.clone()); + + let mut cmd = TokioCommand::new(exe); + cmd.arg("exec") + .arg("--full-auto") + .arg("--cd") + .arg(worktree_dir.as_os_str()) + .stdout(Stdio::from(log_f.try_clone()?)) + .stderr(Stdio::from(log_f)); + for ov in cli_config_overrides.raw_overrides.iter() { + cmd.arg("-c").arg(ov); + } + cmd.arg(&prompt); + + let status = cmd.status().await?; + let exec_exit_code = status.code(); + + // Auto-commit changes in the worktree if any + let status_out = TokioCommand::new("git") + .args(["status", "--porcelain"]) + .current_dir(&worktree_dir) + .output() + .await?; + let status_text = String::from_utf8_lossy(&status_out.stdout); + let had_changes = !status_text.trim().is_empty(); + if had_changes { + if !TokioCommand::new("git") + .args(["add", "-A"]) + .current_dir(&worktree_dir) + .status() + .await? + .success() + { + anyhow::bail!("git add failed in worktree"); + } + let commit_message = format!("Codex concurrent: {prompt}"); + let _ = TokioCommand::new("git") + .args(["commit", "-m", &commit_message]) + .current_dir(&worktree_dir) + .status() + .await?; + } + + Ok(ConcurrentRunResult { + branch, + worktree_dir, + log_file, + exec_exit_code, + _had_changes: had_changes, + _applied_changes: None, + }) } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 2f3489199d..fe6100904d 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -44,6 +44,11 @@ struct MultitoolCli { #[arg(long = "automerge", default_value_t = true, action = clap::ArgAction::Set)] pub automerge: bool, + /// Run the same --concurrent prompt N times in separate worktrees and keep them all. + /// Intended to generate multiple candidate solutions without auto-merging. + #[arg(long = "best-of-n", value_name = "N", default_value_t = 1)] + pub best_of_n: usize, + #[clap(flatten)] interactive: TuiCli, @@ -134,16 +139,81 @@ async fn cli_main(codex_linux_sandbox_exe: Option) -> anyhow::Result<() eprintln!("--concurrent cannot be used together with a subcommand"); std::process::exit(2); } - concurrent::run_concurrent_flow( - prompt, - cli.config_overrides, - codex_linux_sandbox_exe, - cli.automerge, - ) - .await?; + let runs = if cli.best_of_n == 0 { 1 } else { cli.best_of_n }; + if runs > 1 { + println!( + "Running best-of-n with {runs} runs; auto-merge will be disabled and worktrees kept." + ); + + // Launch all runs concurrently and collect results as they finish. + let mut join_set = tokio::task::JoinSet::new(); + for _ in 0..runs { + let prompt = prompt.clone(); + let overrides = cli.config_overrides.clone(); + let sandbox = codex_linux_sandbox_exe.clone(); + join_set.spawn(async move { + concurrent::run_concurrent_flow_quiet_no_automerge(prompt, overrides, sandbox) + .await + }); + } + + let mut results: Vec = Vec::with_capacity(runs); + while let Some(join_result) = join_set.join_next().await { + match join_result { + Ok(Ok(res)) => { + println!( + "task finished for branch: {}\n, directory: {}", + res.branch, + res.worktree_dir.display() + ); + results.push(res); + } + Ok(Err(err)) => { + eprintln!("concurrent task failed: {err}"); + } + Err(join_err) => { + eprintln!("failed to join concurrent task: {join_err}"); + } + } + } + + println!("\nBest-of-n summary:"); + for r in &results { + let status = match r.exec_exit_code { + Some(0) => "OK", + Some(_code) => "FAIL", + None => "OK", + }; + let log = r + .log_file + .as_ref() + .map(|p| p.to_string_lossy().to_string()) + .unwrap_or_else(|| "".to_string()); + println!( + "[{status}] branch={} worktree={} log={}", + r.branch, + r.worktree_dir.display(), + log + ); + } + } else { + concurrent::run_concurrent_flow( + prompt, + cli.config_overrides, + codex_linux_sandbox_exe, + cli.automerge, + false, + ) + .await?; + } return Ok(()); } + if cli.best_of_n > 1 { + eprintln!("--best-of-n requires --concurrent "); + std::process::exit(2); + } + match cli.subcommand { None => { let mut tui_cli = cli.interactive;