diff --git a/codex-rs/core-skills/src/loader_tests.rs b/codex-rs/core-skills/src/loader_tests.rs index 3122dbaf0b..0a4559d3ee 100644 --- a/codex-rs/core-skills/src/loader_tests.rs +++ b/codex-rs/core-skills/src/loader_tests.rs @@ -4,19 +4,34 @@ use codex_config::ConfigLayerEntry; use codex_config::ConfigLayerStack; use codex_config::ConfigRequirements; use codex_config::ConfigRequirementsToml; +use codex_exec_server::CopyOptions; +use codex_exec_server::CreateDirectoryOptions; +use codex_exec_server::ExecutorFileSystem; +use codex_exec_server::ExecutorFileSystemFuture; +use codex_exec_server::FileMetadata; +use codex_exec_server::FileSystemReadStream; +use codex_exec_server::FileSystemSandboxContext; use codex_exec_server::LOCAL_FS; +use codex_exec_server::ReadDirectoryEntry; +use codex_exec_server::RemoveOptions; +use codex_exec_server::WalkOptions; +use codex_exec_server::WalkOutcome; use codex_protocol::protocol::Product; use codex_protocol::protocol::SkillScope; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::PathExt; +use codex_utils_path_uri::PathUri; use dunce::canonicalize as canonicalize_path; use pretty_assertions::assert_eq; use std::fs; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; +use std::sync::Mutex; use tempfile::TempDir; +use tokio::sync::Notify; +use tokio::sync::Semaphore; use toml::Value as TomlValue; const REPO_ROOT_CONFIG_DIR_NAME: &str = ".codex"; @@ -26,6 +41,138 @@ struct TestConfig { config_layer_stack: ConfigLayerStack, } +struct BlockingRepoSkillRootFileSystem { + inner: Arc, + metadata_calls: Arc, +} + +struct BlockingMetadataCalls { + paths: Mutex>, + started: Notify, + release: Semaphore, +} + +impl Default for BlockingMetadataCalls { + fn default() -> Self { + Self { + paths: Mutex::new(Vec::new()), + started: Notify::new(), + release: Semaphore::new(0), + } + } +} + +impl ExecutorFileSystem for BlockingRepoSkillRootFileSystem { + fn canonicalize<'a>( + &'a self, + path: &'a PathUri, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, PathUri> { + self.inner.canonicalize(path, sandbox) + } + + fn read_file<'a>( + &'a self, + path: &'a PathUri, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, Vec> { + self.inner.read_file(path, sandbox) + } + + fn read_file_stream<'a>( + &'a self, + path: &'a PathUri, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, FileSystemReadStream> { + self.inner.read_file_stream(path, sandbox) + } + + fn write_file<'a>( + &'a self, + path: &'a PathUri, + contents: Vec, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, ()> { + self.inner.write_file(path, contents, sandbox) + } + + fn create_directory<'a>( + &'a self, + path: &'a PathUri, + options: CreateDirectoryOptions, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, ()> { + self.inner.create_directory(path, options, sandbox) + } + + fn get_metadata<'a>( + &'a self, + path: &'a PathUri, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, FileMetadata> { + let repo_skill_root_suffix = Path::new(AGENTS_DIR_NAME).join(SKILLS_DIR_NAME); + let Ok(path_abs) = path.to_abs_path() else { + return self.inner.get_metadata(path, sandbox); + }; + if !path_abs.ends_with(repo_skill_root_suffix) { + return self.inner.get_metadata(path, sandbox); + } + + self.metadata_calls + .paths + .lock() + .expect("metadata paths lock") + .push(path.clone()); + self.metadata_calls.started.notify_one(); + Box::pin(async move { + self.metadata_calls + .release + .acquire() + .await + .expect("metadata release semaphore") + .forget(); + self.inner.get_metadata(path, sandbox).await + }) + } + + fn read_directory<'a>( + &'a self, + path: &'a PathUri, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, Vec> { + self.inner.read_directory(path, sandbox) + } + + fn walk<'a>( + &'a self, + path: &'a PathUri, + options: WalkOptions, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, WalkOutcome> { + self.inner.walk(path, options, sandbox) + } + + fn remove<'a>( + &'a self, + path: &'a PathUri, + options: RemoveOptions, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, ()> { + self.inner.remove(path, options, sandbox) + } + + fn copy<'a>( + &'a self, + source_path: &'a PathUri, + destination_path: &'a PathUri, + options: CopyOptions, + sandbox: Option<&'a FileSystemSandboxContext>, + ) -> ExecutorFileSystemFuture<'a, ()> { + self.inner + .copy(source_path, destination_path, options, sandbox) + } +} + async fn make_config(codex_home: &TempDir) -> TestConfig { make_config_for_cwd(codex_home, codex_home.path().to_path_buf()).await } @@ -2050,6 +2197,112 @@ async fn loads_skills_from_all_codex_dirs_under_project_root() { ); } +#[tokio::test] +async fn repo_skill_root_search_limits_concurrent_probes_and_preserves_order() { + const CONCURRENCY_LIMIT: usize = 256; + + let codex_home = tempfile::tempdir().expect("tempdir"); + let repo_dir = tempfile::tempdir().expect("tempdir"); + mark_as_git_repo(repo_dir.path()); + + let mut directories = vec![repo_dir.path().to_path_buf()]; + let mut cwd = repo_dir.path().to_path_buf(); + for _ in 0..CONCURRENCY_LIMIT { + cwd.push("d"); + directories.push(cwd.clone()); + } + fs::create_dir_all(&cwd).expect("nested cwd"); + + let expected_roots = [0, CONCURRENCY_LIMIT / 2, CONCURRENCY_LIMIT] + .map(|index| { + directories[index] + .join(AGENTS_DIR_NAME) + .join(SKILLS_DIR_NAME) + }) + .map(|path| { + fs::create_dir_all(&path).expect("repo skill root"); + path.abs() + }); + let expected_probes = directories + .iter() + .map(|directory| { + PathUri::from_abs_path(&directory.join(AGENTS_DIR_NAME).join(SKILLS_DIR_NAME).abs()) + }) + .collect::>(); + let cfg = make_config_for_cwd(&codex_home, cwd).await; + let metadata_calls = Arc::new(BlockingMetadataCalls::default()); + let fs: Arc = Arc::new(BlockingRepoSkillRootFileSystem { + inner: Arc::clone(&LOCAL_FS), + metadata_calls: Arc::clone(&metadata_calls), + }); + + let assertions = async { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let started = metadata_calls.started.notified(); + if metadata_calls + .paths + .lock() + .expect("metadata paths lock") + .len() + >= CONCURRENCY_LIMIT + { + break; + } + started.await; + } + }) + .await + .expect("initial repo skill root window should start"); + assert_eq!( + metadata_calls + .paths + .lock() + .expect("metadata paths lock") + .as_slice(), + &expected_probes[..CONCURRENCY_LIMIT] + ); + + metadata_calls.release.add_permits(1); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let started = metadata_calls.started.notified(); + if metadata_calls + .paths + .lock() + .expect("metadata paths lock") + .len() + > CONCURRENCY_LIMIT + { + break; + } + started.await; + } + }) + .await + .expect("next repo skill root probe should start"); + assert_eq!( + metadata_calls + .paths + .lock() + .expect("metadata paths lock") + .as_slice(), + expected_probes.as_slice() + ); + + metadata_calls.release.add_permits(expected_probes.len()); + }; + let (roots, ()) = tokio::join!( + super::repo_agents_skill_roots(Some(fs), &cfg.config_layer_stack, &cfg.cwd), + assertions + ); + + assert_eq!( + roots.into_iter().map(|root| root.path).collect::>(), + expected_roots + ); +} + #[tokio::test] async fn loads_skills_from_codex_dir_when_not_git_repo() { let codex_home = tempfile::tempdir().expect("tempdir"); diff --git a/codex-rs/core/src/agents_md_tests.rs b/codex-rs/core/src/agents_md_tests.rs index 89bc5414f4..4a5eef4143 100644 --- a/codex-rs/core/src/agents_md_tests.rs +++ b/codex-rs/core/src/agents_md_tests.rs @@ -34,11 +34,13 @@ use std::sync::Arc; use std::sync::Mutex; use tempfile::TempDir; use tokio::sync::Notify; +use tokio::sync::Semaphore; #[derive(Clone, Copy)] enum InjectedFailure { Metadata(io::ErrorKind), MetadataBlocked, + MetadataBlockedByFilenamePrefix(&'static str), MetadataPending, Read(io::ErrorKind), } @@ -49,11 +51,20 @@ struct FailingFileSystem { metadata_calls: Arc, } -#[derive(Default)] struct MetadataCallCounts { paths: Mutex>, started: Notify, - release: Notify, + release: Semaphore, +} + +impl Default for MetadataCallCounts { + fn default() -> Self { + Self { + paths: Mutex::new(Vec::new()), + started: Notify::new(), + release: Semaphore::new(0), + } + } } impl FailingFileSystem { @@ -113,7 +124,26 @@ impl FailingFileSystem { Err(io::Error::new(kind, "injected metadata failure")) } InjectedFailure::MetadataBlocked if path_abs == self.path => { - self.metadata_calls.release.notified().await; + self.metadata_calls + .release + .acquire() + .await + .expect("metadata release semaphore") + .forget(); + LOCAL_FS.get_metadata(path, sandbox).await + } + InjectedFailure::MetadataBlockedByFilenamePrefix(prefix) + if path_abs + .file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.starts_with(prefix)) => + { + self.metadata_calls + .release + .acquire() + .await + .expect("metadata release semaphore") + .forget(); LOCAL_FS.get_metadata(path, sandbox).await } InjectedFailure::MetadataPending if path_abs == self.path => { @@ -121,6 +151,7 @@ impl FailingFileSystem { } InjectedFailure::Metadata(_) | InjectedFailure::MetadataBlocked + | InjectedFailure::MetadataBlockedByFilenamePrefix(_) | InjectedFailure::MetadataPending | InjectedFailure::Read(_) => LOCAL_FS.get_metadata(path, sandbox).await, } @@ -719,69 +750,107 @@ async fn marker_search_does_not_wait_for_a_higher_ancestor() { } #[tokio::test] -async fn project_root_marker_search_starts_all_ancestor_probes() { - const NESTING_DEPTH: usize = 9; +async fn project_root_marker_search_limits_concurrent_probes_and_preserves_order() { + const CONCURRENCY_LIMIT: usize = 256; let tmp = tempfile::tempdir().expect("tempdir"); - fs::write(tmp.path().join(".git"), "").unwrap(); fs::write(tmp.path().join("AGENTS.md"), "project doc").unwrap(); - let mut nested = tmp.path().to_path_buf(); - for depth in 0..NESTING_DEPTH { - nested.push(format!("nested-{depth}")); - } + let nested = tmp.path().join("nested"); fs::create_dir_all(&nested).unwrap(); + fs::write(nested.join("AGENTS.md"), "nested project doc").unwrap(); - let mut config = make_config(&tmp, /*limit*/ 4096, /*instructions*/ None).await; + let markers = (0..=CONCURRENCY_LIMIT) + .map(|index| format!(".project-root-{index}")) + .collect::>(); + fs::write( + tmp.path() + .join(markers.last().expect("last project root marker")), + "", + ) + .unwrap(); + let marker_refs = markers.iter().map(String::as_str).collect::>(); + + let mut config = make_config_with_project_root_markers( + &tmp, + /*limit*/ 4096, + /*instructions*/ None, + &marker_refs, + ) + .await; config.cwd = nested.abs(); - let expected_probe_count = config.cwd.ancestors().count(); let cwd = PathUri::from_abs_path(&config.cwd); + let expected_initial_probes = markers + .iter() + .map(|marker| cwd.join(marker).expect("project root marker path")) + .collect::>(); + let max_probe_count = markers.len() * config.cwd.ancestors().count(); let metadata_calls = Arc::new(MetadataCallCounts::default()); let fs = FailingFileSystem { - path: config.cwd.join(".git"), - failure: InjectedFailure::MetadataBlocked, + path: config.cwd.join("unused"), + failure: InjectedFailure::MetadataBlockedByFilenamePrefix(".project-root-"), metadata_calls: Arc::clone(&metadata_calls), }; - let search = - tokio::spawn(async move { super::agents_md_paths(&config.config, &cwd, &fs).await }); - tokio::time::timeout(std::time::Duration::from_secs(5), async { - loop { - let started = metadata_calls.started.notified(); - if metadata_calls - .paths - .lock() - .expect("metadata paths lock") - .len() - >= expected_probe_count - { - break; + let assertions = async { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let started = metadata_calls.started.notified(); + if metadata_calls + .paths + .lock() + .expect("metadata paths lock") + .len() + >= CONCURRENCY_LIMIT + { + break; + } + started.await; } - started.await; - } - }) - .await - .expect("initial marker window should start"); - assert_eq!( - metadata_calls - .paths - .lock() - .expect("metadata paths lock") - .len(), - expected_probe_count - ); - - metadata_calls.release.notify_one(); - let paths = tokio::time::timeout(std::time::Duration::from_secs(5), search) + }) .await - .expect("marker search should complete") - .expect("marker search task") - .expect("AGENTS.md discovery"); + .expect("initial marker window should start"); + assert_eq!( + *metadata_calls.paths.lock().expect("metadata paths lock"), + expected_initial_probes[..CONCURRENCY_LIMIT] + ); + + metadata_calls.release.add_permits(1); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let started = metadata_calls.started.notified(); + if metadata_calls + .paths + .lock() + .expect("metadata paths lock") + .len() + > CONCURRENCY_LIMIT + { + break; + } + started.await; + } + }) + .await + .expect("next marker probe should start"); + assert_eq!( + *metadata_calls.paths.lock().expect("metadata paths lock"), + expected_initial_probes + ); + + metadata_calls.release.add_permits(max_probe_count); + }; + let (paths, ()) = tokio::join!( + super::agents_md_paths(&config.config, &cwd, &fs), + assertions + ); + let paths = paths.expect("AGENTS.md discovery"); assert_eq!( paths, - vec![PathUri::from_abs_path( - &tmp.path().join(DEFAULT_AGENTS_MD_FILENAME).abs() - )] + vec![ + PathUri::from_abs_path(&tmp.path().join(DEFAULT_AGENTS_MD_FILENAME).abs()), + PathUri::from_abs_path(&nested.join(DEFAULT_AGENTS_MD_FILENAME).abs()), + ] ); } @@ -854,7 +923,7 @@ async fn agents_md_search_starts_all_directory_probes() { expected_probes.sort(); assert_eq!(actual_probes, expected_probes); - metadata_calls.release.notify_one(); + metadata_calls.release.add_permits(1); let paths = tokio::time::timeout(std::time::Duration::from_secs(5), search) .await .expect("AGENTS.md search should complete")