diff --git a/codex-rs/core/src/git_info_tests.rs b/codex-rs/core/src/git_info_tests.rs index bebcd0f25a..d050441a5f 100644 --- a/codex-rs/core/src/git_info_tests.rs +++ b/codex-rs/core/src/git_info_tests.rs @@ -12,7 +12,7 @@ use codex_exec_server::RemoveOptions; use codex_git_utils::GitInfo; use codex_git_utils::GitSha; use codex_git_utils::collect_git_info; -use codex_git_utils::get_has_changes; +use codex_git_utils::get_has_changes_in_repo; use codex_git_utils::git_diff_to_remote; use codex_git_utils::recent_commits; use codex_git_utils::resolve_root_git_project_for_trust; @@ -425,14 +425,20 @@ async fn test_collect_git_info_with_branch() { #[tokio::test] async fn test_get_has_changes_non_git_directory_returns_none() { let temp_dir = TempDir::new().expect("Failed to create temp dir"); - assert_eq!(get_has_changes(temp_dir.path()).await, None); + assert_eq!( + get_has_changes_in_repo(temp_dir.path(), temp_dir.path()).await, + None + ); } #[tokio::test] async fn test_get_has_changes_clean_repo_returns_false() { let temp_dir = TempDir::new().expect("Failed to create temp dir"); let repo_path = create_test_git_repo(&temp_dir).await; - assert_eq!(get_has_changes(&repo_path).await, Some(false)); + assert_eq!( + get_has_changes_in_repo(&repo_path, &repo_path).await, + Some(false) + ); } #[tokio::test] @@ -441,7 +447,10 @@ async fn test_get_has_changes_with_tracked_change_returns_true() { let repo_path = create_test_git_repo(&temp_dir).await; fs::write(repo_path.join("test.txt"), "updated tracked file").expect("write tracked file"); - assert_eq!(get_has_changes(&repo_path).await, Some(true)); + assert_eq!( + get_has_changes_in_repo(&repo_path, &repo_path).await, + Some(true) + ); } #[tokio::test] @@ -450,7 +459,10 @@ async fn test_get_has_changes_with_untracked_change_returns_true() { let repo_path = create_test_git_repo(&temp_dir).await; fs::write(repo_path.join("new_file.txt"), "untracked").expect("write untracked file"); - assert_eq!(get_has_changes(&repo_path).await, Some(true)); + assert_eq!( + get_has_changes_in_repo(&repo_path, &repo_path).await, + Some(true) + ); } #[cfg(unix)] @@ -490,7 +502,10 @@ async fn test_get_has_changes_ignores_configured_hooks_path() { fs::write(repo_path.join("test.txt"), "test content").expect("refresh tracked file"); - assert_eq!(get_has_changes(&repo_path).await, Some(false)); + assert_eq!( + get_has_changes_in_repo(&repo_path, &repo_path).await, + Some(false) + ); assert!( !marker_path.exists(), "metadata collection should not invoke configured hook directories" diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index 8423e1af5e..7f6567b642 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -1,5 +1,7 @@ use std::collections::BTreeMap; use std::collections::HashMap; +use std::path::Path; +use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; use std::sync::OnceLock; @@ -21,7 +23,7 @@ use crate::responses_metadata::subagent_metadata_kind; use crate::sandbox_tags::permission_profile_sandbox_tag; use codex_git_utils::get_git_remote_urls_assume_git_repo; use codex_git_utils::get_git_repo_root; -use codex_git_utils::get_has_changes; +use codex_git_utils::get_has_changes_in_repo; use codex_git_utils::get_head_commit_hash; use codex_protocol::ThreadId; use codex_protocol::ToolName; @@ -89,7 +91,7 @@ pub async fn detached_memory_responses_metadata( #[derive(Debug)] pub(crate) struct TurnMetadataState { cwd: AbsolutePathBuf, - repo_root: Option, + repo_root: Option, session_id: String, thread_id: String, forked_from_thread_id: Option, @@ -123,7 +125,7 @@ impl TurnMetadataState { windows_sandbox_level: WindowsSandboxLevel, enforce_managed_network: bool, ) -> Self { - let repo_root = get_git_repo_root(&cwd).map(|root| root.to_string_lossy().into_owned()); + let repo_root = get_git_repo_root(&cwd); let sandbox = Some( permission_profile_sandbox_tag( permission_profile, @@ -318,17 +320,20 @@ impl TurnMetadataState { let state = Arc::clone(self); *task_guard = Some(tokio::spawn(async move { - let workspace_git_metadata = state.fetch_workspace_git_metadata().await; let Some(repo_root) = state.repo_root.clone() else { return; }; + let workspace_git_metadata = state.fetch_workspace_git_metadata(&repo_root).await; if workspace_git_metadata.is_empty() { return; } let mut workspaces = BTreeMap::new(); - workspaces.insert(repo_root, workspace_git_metadata.into()); + workspaces.insert( + repo_root.to_string_lossy().into_owned(), + workspace_git_metadata.into(), + ); *state .enriched_workspaces .write() @@ -346,11 +351,11 @@ impl TurnMetadataState { } } - async fn fetch_workspace_git_metadata(&self) -> WorkspaceGitMetadata { + async fn fetch_workspace_git_metadata(&self, repo_root: &Path) -> WorkspaceGitMetadata { let (head_commit_hash, associated_remote_urls, has_changes) = tokio::join!( get_head_commit_hash(&self.cwd), get_git_remote_urls_assume_git_repo(&self.cwd), - get_has_changes(&self.cwd), + get_has_changes_in_repo(&self.cwd, repo_root), ); let latest_git_commit_hash = head_commit_hash.map(|sha| sha.0); @@ -363,11 +368,13 @@ impl TurnMetadataState { } async fn memory_workspaces(cwd: &AbsolutePathBuf) -> BTreeMap { - let repo_root = get_git_repo_root(cwd).map(|root| root.to_string_lossy().into_owned()); + let Some(repo_root) = get_git_repo_root(cwd) else { + return BTreeMap::new(); + }; let (head_commit_hash, associated_remote_urls, has_changes) = tokio::join!( get_head_commit_hash(cwd), get_git_remote_urls_assume_git_repo(cwd), - get_has_changes(cwd), + get_has_changes_in_repo(cwd, &repo_root), ); let workspace_git_metadata = WorkspaceGitMetadata { associated_remote_urls, @@ -375,10 +382,11 @@ async fn memory_workspaces(cwd: &AbsolutePathBuf) -> BTreeMap &str { value.strip_suffix(".git").unwrap_or(value) } -pub async fn get_has_changes(cwd: &Path) -> Option { - let git = Path::new("git"); - let fsmonitor = detect_local_fsmonitor_override(git, cwd).await; - let output = - run_git_command_with_timeout_from(git, &["status", "--porcelain"], cwd, fsmonitor).await?; - if !output.status.success() { - return None; - } - - Some(!output.stdout.is_empty()) -} - fn parse_git_remote_urls(stdout: &str) -> Option> { let mut remotes = BTreeMap::new(); for line in stdout.lines() { @@ -411,12 +399,15 @@ impl crate::FsmonitorProbeRunner for LocalFsmonitorProbeRunner<'_> { } } -async fn detect_local_fsmonitor_override(git: &Path, cwd: &Path) -> crate::FsmonitorOverride { +pub(crate) async fn detect_local_fsmonitor_override( + git: &Path, + cwd: &Path, +) -> crate::FsmonitorOverride { let mut runner = LocalFsmonitorProbeRunner { git, cwd }; crate::detect_fsmonitor_override(&mut runner).await } -async fn run_git_command_with_timeout_from( +pub(crate) async fn run_git_command_with_timeout_from( git: &Path, args: &[&str], cwd: &Path, diff --git a/codex-rs/git-utils/src/lib.rs b/codex-rs/git-utils/src/lib.rs index 7acd069d91..1c75e7e7da 100644 --- a/codex-rs/git-utils/src/lib.rs +++ b/codex-rs/git-utils/src/lib.rs @@ -7,6 +7,7 @@ mod git_process; mod info; mod operations; mod platform; +mod status; /// Git configuration that rejects implicitly discovered bare repositories while /// preserving repositories selected explicitly through `GIT_DIR` or `--git-dir`. @@ -40,10 +41,10 @@ pub use info::default_branch_name; pub use info::get_git_remote_urls; pub use info::get_git_remote_urls_assume_git_repo; pub use info::get_git_repo_root; -pub use info::get_has_changes; pub use info::get_head_commit_hash; pub use info::git_diff_to_remote; pub use info::local_git_branches; pub use info::recent_commits; pub use info::resolve_root_git_project_for_trust; pub use platform::create_symlink; +pub use status::get_has_changes_in_repo; diff --git a/codex-rs/git-utils/src/status.rs b/codex-rs/git-utils/src/status.rs new file mode 100644 index 0000000000..ba24d52e37 --- /dev/null +++ b/codex-rs/git-utils/src/status.rs @@ -0,0 +1,83 @@ +use std::collections::HashMap; +use std::future::Future; +use std::path::Path; +use std::path::PathBuf; +use std::sync::Mutex; +use std::sync::OnceLock; + +use futures::FutureExt; +use futures::future::BoxFuture; +use futures::future::WeakShared; + +use crate::info::detect_local_fsmonitor_override; +use crate::info::run_git_command_with_timeout_from; + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct GitStatusKey { + git: PathBuf, + repo_root: PathBuf, +} + +type GitStatusFuture = BoxFuture<'static, Option>; + +fn git_status_runs() -> &'static Mutex>> { + static RUNS: OnceLock>>> = + OnceLock::new(); + RUNS.get_or_init(|| Mutex::new(HashMap::new())) +} + +pub async fn get_has_changes_in_repo(cwd: &Path, repo_root: &Path) -> Option { + let git = PathBuf::from("git"); + let cwd = cwd.to_path_buf(); + let key = git_status_key(git.clone(), repo_root).await; + share_git_status_run(key, move || async move { + let fsmonitor = detect_local_fsmonitor_override(&git, &cwd).await; + let output = + run_git_command_with_timeout_from(&git, &["status", "--porcelain"], &cwd, fsmonitor) + .await?; + output.status.success().then_some(!output.stdout.is_empty()) + }) + .await +} + +async fn git_status_key(git: PathBuf, repo_root: &Path) -> GitStatusKey { + let repo_root = tokio::fs::canonicalize(repo_root) + .await + .unwrap_or_else(|_| repo_root.to_path_buf()); + GitStatusKey { git, repo_root } +} + +async fn share_git_status_run(key: GitStatusKey, run: F) -> Option +where + F: FnOnce() -> Fut, + Fut: Future> + Send + 'static, +{ + let result = { + let mut runs = git_status_runs() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(result) = runs + .get(&key) + .and_then(WeakShared::upgrade) + .filter(|result| result.peek().is_none()) + { + result + } else { + runs.retain(|_, result| { + result + .upgrade() + .is_some_and(|result| result.peek().is_none()) + }); + let result = run().boxed().shared(); + if let Some(weak_result) = result.downgrade() { + runs.insert(key, weak_result); + } + result + } + }; + result.await +} + +#[cfg(test)] +#[path = "status_tests.rs"] +mod tests; diff --git a/codex-rs/git-utils/src/status_tests.rs b/codex-rs/git-utils/src/status_tests.rs new file mode 100644 index 0000000000..6da16af542 --- /dev/null +++ b/codex-rs/git-utils/src/status_tests.rs @@ -0,0 +1,302 @@ +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::time::Duration; + +use futures::future::WeakShared; +use futures::future::join_all; +use pretty_assertions::assert_eq; +use tempfile::TempDir; + +use super::GitStatusKey; +use super::get_has_changes_in_repo; +use super::git_status_key; +use super::git_status_runs; +use super::share_git_status_run; + +#[tokio::test] +async fn status_requests_from_sibling_directories_use_the_known_worktree_root() { + let repository = TempDir::new().expect("create repository"); + gix::init(repository.path()).expect("initialize repository"); + let first_directory = repository.path().join("first"); + let second_directory = repository.path().join("second"); + std::fs::create_dir(&first_directory).expect("create first directory"); + std::fs::create_dir(&second_directory).expect("create second directory"); + + let results = tokio::join!( + get_has_changes_in_repo(&first_directory, repository.path()), + get_has_changes_in_repo(&second_directory, repository.path()), + ); + + assert_eq!(results, (Some(false), Some(false))); +} + +#[cfg(unix)] +#[tokio::test] +async fn status_keys_coalesce_symlink_aliases_of_the_same_worktree() { + let temp_dir = TempDir::new().expect("create temp directory"); + let repository = temp_dir.path().join("repository"); + let repository_alias = temp_dir.path().join("repository-alias"); + std::fs::create_dir(&repository).expect("create repository"); + std::os::unix::fs::symlink(&repository, &repository_alias).expect("create repository alias"); + + let key = git_status_key(PathBuf::from("git"), &repository).await; + let alias_key = git_status_key(PathBuf::from("git"), &repository_alias).await; + + assert_eq!(key, alias_key); +} + +#[tokio::test] +async fn concurrent_status_requests_share_one_repository_scan() { + let repository = TempDir::new().expect("create repository"); + let key = GitStatusKey { + git: PathBuf::from("git"), + repo_root: repository.path().to_path_buf(), + }; + let scan_count = Arc::new(AtomicUsize::new(0)); + + let results = join_all((0..32).map(|_| { + let key = key.clone(); + let scan_count = Arc::clone(&scan_count); + share_git_status_run(key, move || async move { + scan_count.fetch_add(1, Ordering::Relaxed); + tokio::time::sleep(Duration::from_millis(25)).await; + Some(true) + }) + })) + .await; + + assert_eq!(results, vec![Some(true); 32]); + assert_eq!(scan_count.load(Ordering::Relaxed), 1); +} + +#[tokio::test] +async fn status_requests_do_not_share_different_repositories() { + let first_repository = TempDir::new().expect("create first repository"); + let second_repository = TempDir::new().expect("create second repository"); + let scan_count = Arc::new(AtomicUsize::new(0)); + let first_key = git_status_key(PathBuf::from("git"), first_repository.path()).await; + let second_key = git_status_key(PathBuf::from("git"), second_repository.path()).await; + + let first_scan_count = Arc::clone(&scan_count); + let second_scan_count = Arc::clone(&scan_count); + let results = tokio::join!( + share_git_status_run(first_key, move || async move { + first_scan_count.fetch_add(1, Ordering::Relaxed); + tokio::time::sleep(Duration::from_millis(25)).await; + Some(false) + }), + share_git_status_run(second_key, move || async move { + second_scan_count.fetch_add(1, Ordering::Relaxed); + tokio::time::sleep(Duration::from_millis(25)).await; + Some(true) + }), + ); + + assert_eq!(results, (Some(false), Some(true))); + assert_eq!(scan_count.load(Ordering::Relaxed), 2); +} + +#[tokio::test] +async fn completed_status_requests_start_fresh_scans_and_prune_expired_entries() { + let first_repository = TempDir::new().expect("create first repository"); + let first_key = GitStatusKey { + git: PathBuf::from("git"), + repo_root: first_repository.path().to_path_buf(), + }; + let scan_count = Arc::new(AtomicUsize::new(0)); + + let first_scan_count = Arc::clone(&scan_count); + assert_eq!( + share_git_status_run(first_key.clone(), move || async move { + first_scan_count.fetch_add(1, Ordering::Relaxed); + Some(true) + }) + .await, + Some(true) + ); + + let second_scan_count = Arc::clone(&scan_count); + assert_eq!( + share_git_status_run(first_key.clone(), move || async move { + second_scan_count.fetch_add(1, Ordering::Relaxed); + Some(false) + }) + .await, + Some(false) + ); + assert_eq!(scan_count.load(Ordering::Relaxed), 2); + + let second_repository = TempDir::new().expect("create second repository"); + let third_scan_count = Arc::clone(&scan_count); + assert_eq!( + share_git_status_run( + GitStatusKey { + git: PathBuf::from("git"), + repo_root: second_repository.path().to_path_buf(), + }, + move || async move { + third_scan_count.fetch_add(1, Ordering::Relaxed); + Some(true) + }, + ) + .await, + Some(true) + ); + assert_eq!(scan_count.load(Ordering::Relaxed), 3); + assert!( + !git_status_runs() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .contains_key(&first_key) + ); +} + +#[tokio::test] +async fn completed_status_requests_are_not_reused_while_a_consumer_retains_them() { + let repository = TempDir::new().expect("create repository"); + let key = GitStatusKey { + git: PathBuf::from("git"), + repo_root: repository.path().to_path_buf(), + }; + let scan_count = Arc::new(AtomicUsize::new(0)); + let (release_sender, release_receiver) = tokio::sync::oneshot::channel(); + + let first_key = key.clone(); + let first_scan_count = Arc::clone(&scan_count); + let first = tokio::spawn(async move { + share_git_status_run(first_key, move || async move { + first_scan_count.fetch_add(1, Ordering::Relaxed); + release_receiver.await.ok()?; + Some(true) + }) + .await + }); + + tokio::task::yield_now().await; + let retained_result = git_status_runs() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&key) + .and_then(WeakShared::upgrade) + .expect("retain the pending shared status scan"); + + release_sender.send(()).expect("release shared status scan"); + assert_eq!(first.await.expect("status task completed"), Some(true)); + assert_eq!(retained_result.peek(), Some(&Some(true))); + + let second_scan_count = Arc::clone(&scan_count); + let result = share_git_status_run(key, move || async move { + second_scan_count.fetch_add(1, Ordering::Relaxed); + Some(false) + }) + .await; + + assert_eq!(result, Some(false)); + assert_eq!(scan_count.load(Ordering::Relaxed), 2); + assert_eq!(retained_result.peek(), Some(&Some(true))); +} + +#[tokio::test] +async fn canceling_one_consumer_preserves_the_shared_status_scan() { + let repository = TempDir::new().expect("create repository"); + let key = GitStatusKey { + git: PathBuf::from("git"), + repo_root: repository.path().to_path_buf(), + }; + let scan_count = Arc::new(AtomicUsize::new(0)); + let (release_sender, release_receiver) = tokio::sync::oneshot::channel(); + + let first_key = key.clone(); + let first_scan_count = Arc::clone(&scan_count); + let first = tokio::spawn(async move { + share_git_status_run(first_key, move || async move { + first_scan_count.fetch_add(1, Ordering::Relaxed); + release_receiver.await.ok()?; + Some(true) + }) + .await + }); + + tokio::task::yield_now().await; + + let second_key = key.clone(); + let second_scan_count = Arc::clone(&scan_count); + let second = tokio::spawn(async move { + share_git_status_run(second_key, move || async move { + second_scan_count.fetch_add(1, Ordering::Relaxed); + Some(false) + }) + .await + }); + + tokio::task::yield_now().await; + second.abort(); + assert!( + second + .await + .expect_err("consumer was canceled") + .is_cancelled() + ); + assert_eq!(scan_count.load(Ordering::Relaxed), 1); + + release_sender.send(()).expect("release shared status scan"); + assert_eq!(first.await.expect("status task completed"), Some(true)); + assert!( + git_status_runs() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&key) + .and_then(WeakShared::upgrade) + .is_none() + ); +} + +#[tokio::test] +async fn canceling_the_last_consumer_allows_a_fresh_status_scan() { + let repository = TempDir::new().expect("create repository"); + let key = GitStatusKey { + git: PathBuf::from("git"), + repo_root: repository.path().to_path_buf(), + }; + let scan_count = Arc::new(AtomicUsize::new(0)); + let first_key = key.clone(); + let first_scan_count = Arc::clone(&scan_count); + + let first = tokio::spawn(async move { + share_git_status_run(first_key, move || async move { + first_scan_count.fetch_add(1, Ordering::Relaxed); + std::future::pending::>().await + }) + .await + }); + + tokio::task::yield_now().await; + assert_eq!(scan_count.load(Ordering::Relaxed), 1); + first.abort(); + assert!( + first + .await + .expect_err("consumer was canceled") + .is_cancelled() + ); + + let second_scan_count = Arc::clone(&scan_count); + let result = share_git_status_run(key.clone(), move || async move { + second_scan_count.fetch_add(1, Ordering::Relaxed); + Some(false) + }) + .await; + + assert_eq!(result, Some(false)); + assert_eq!(scan_count.load(Ordering::Relaxed), 2); + assert!( + git_status_runs() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&key) + .and_then(WeakShared::upgrade) + .is_none() + ); +}