From 8436b749a410133c58ec61fe89db07496ca22033 Mon Sep 17 00:00:00 2001 From: Charlie Marsh Date: Tue, 1 Sep 2026 20:13:59 +0000 Subject: [PATCH] Bound Git root discovery for metadata enrichment (#42132) ## Why Git root discovery is optional metadata work, but filesystem probes can block. They should not exhaust Tokio's blocking pool, delay runtime shutdown, or prevent later turns from observing repository changes. ## What changed - Add a shared `GitRootDiscovery` service that coalesces concurrent lookups for the same working directory and limits probes across directories. - Run probes on detached threads, retain in-flight work across caller cancellation, and discard completed results instead of caching them. - Use the service for turn and memory metadata enrichment, abort unused turn enrichment when its state is dropped, and limit memory metadata waits to one second. ## Testing Add coverage for probe sharing, capacity limits, cancellation, fresh discovery, runtime shutdown, memory timeouts, and repositories restored after startup prewarming. GitOrigin-RevId: bca46fc263e7a12a2f69146d8a0b3e7c7e0846cb --- codex-rs/Cargo.lock | 14 ++ codex-rs/Cargo.toml | 2 + .../tests/suite/v2/realtime_conversation.rs | 11 +- codex-rs/core/Cargo.toml | 1 + codex-rs/core/src/codex_delegate.rs | 1 + codex-rs/core/src/session/mod.rs | 4 + codex-rs/core/src/session/review.rs | 3 +- codex-rs/core/src/session/session.rs | 3 + codex-rs/core/src/session/tests.rs | 5 + .../core/src/session/tests/guardian_tests.rs | 1 + codex-rs/core/src/session/turn_context.rs | 4 +- codex-rs/core/src/state/service.rs | 2 + codex-rs/core/src/thread_manager.rs | 9 + codex-rs/core/src/turn_metadata.rs | 83 ++++---- codex-rs/core/src/turn_metadata_tests.rs | 117 ++++++++++- codex-rs/core/tests/suite/git_enrichment.rs | 21 +- codex-rs/memories/write/src/runtime.rs | 1 + codex-rs/utils/git-discovery/BUILD.bazel | 6 + codex-rs/utils/git-discovery/Cargo.toml | 24 +++ .../src/git_root_discovery_tests.rs | 185 ++++++++++++++++++ codex-rs/utils/git-discovery/src/lib.rs | 120 ++++++++++++ 21 files changed, 562 insertions(+), 55 deletions(-) create mode 100644 codex-rs/utils/git-discovery/BUILD.bazel create mode 100644 codex-rs/utils/git-discovery/Cargo.toml create mode 100644 codex-rs/utils/git-discovery/src/git_root_discovery_tests.rs create mode 100644 codex-rs/utils/git-discovery/src/lib.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 298a2845bf..f5cbc1b787 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2857,6 +2857,7 @@ dependencies = [ "codex-utils-audio", "codex-utils-cache", "codex-utils-cargo-bin", + "codex-utils-git-discovery", "codex-utils-home-dir", "codex-utils-image", "codex-utils-output-truncation", @@ -4689,6 +4690,19 @@ version = "0.0.0" name = "codex-utils-fuzzy-match" version = "0.0.0" +[[package]] +name = "codex-utils-git-discovery" +version = "0.0.0" +dependencies = [ + "codex-git-utils", + "codex-utils-absolute-path", + "futures", + "pretty_assertions", + "tempfile", + "tokio", + "tracing", +] + [[package]] name = "codex-utils-home-dir" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index f7aa05a842..523813b9aa 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -111,6 +111,7 @@ members = [ "utils/cargo-bin", "git-utils", "utils/cache", + "utils/git-discovery", "utils/image", "utils/json-to-toml", "utils/home-dir", @@ -272,6 +273,7 @@ codex-utils-cargo-bin = { path = "utils/cargo-bin" } codex-utils-cli = { path = "utils/cli" } codex-utils-elapsed = { path = "utils/elapsed" } codex-utils-fuzzy-match = { path = "utils/fuzzy-match" } +codex-utils-git-discovery = { path = "utils/git-discovery" } codex-utils-home-dir = { path = "utils/home-dir" } codex-utils-image = { path = "utils/image" } codex-utils-json-to-toml = { path = "utils/json-to-toml" } diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index 5cc7bb3a25..c9e0d181c0 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -1096,16 +1096,17 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> { .. } ))); - assert!(matches!( - history.data.last(), - Some(ThreadTimelineEntry::Realtime { + // A background handoff can append timeline entries after realtime closes. + assert!(history.data.iter().any(|entry| matches!( + entry, + ThreadTimelineEntry::Realtime { item: ThreadRealtimeItem { content: ThreadRealtimeItemContent::RealtimeSessionClosed { .. }, .. }, .. - }) - )); + } + ))); let connections = realtime_server.connections(); assert_eq!(connections.len(), 1); diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 9632576e39..eb0137c03e 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -71,6 +71,7 @@ codex-tools = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-audio = { workspace = true } codex-utils-cache = { workspace = true } +codex-utils-git-discovery = { workspace = true } codex-utils-image = { workspace = true } codex-utils-home-dir = { workspace = true } codex-utils-output-truncation = { workspace = true } diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 047a054b89..af194098ee 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -93,6 +93,7 @@ pub(crate) async fn run_codex_thread_interactive( installation_id: parent_session.installation_id.clone(), auth_manager, models_manager, + git_root_discovery: Arc::clone(&parent_session.services.git_root_discovery), environment_manager: parent_session .services .turn_environments diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 3d0af2e645..a6603e99f0 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -169,6 +169,7 @@ use codex_thread_store::ResumeThreadParams; use codex_thread_store::ThreadPersistenceMetadata; use codex_thread_store::ThreadStore; use codex_utils_audio::prepare_response_items as prepare_audio_response_items; +use codex_utils_git_discovery::GitRootDiscovery; use codex_utils_path_uri::PathUri; use futures::future::BoxFuture; use futures::future::Shared; @@ -418,6 +419,7 @@ pub(crate) struct SessionSpawnArgs { pub(crate) installation_id: String, pub(crate) auth_manager: Arc, pub(crate) models_manager: SharedModelsManager, + pub(crate) git_root_discovery: Arc, pub(crate) environment_manager: Arc, pub(crate) skills_service: Arc, pub(crate) plugins_manager: Arc, @@ -521,6 +523,7 @@ impl Session { installation_id, auth_manager, models_manager, + git_root_discovery, environment_manager, skills_service, plugins_manager, @@ -775,6 +778,7 @@ impl Session { installation_id, auth_manager.clone(), models_manager.clone(), + git_root_discovery, model_info, exec_policy, tx_event.clone(), diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index b2be830235..cf7392ca26 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -191,7 +191,8 @@ pub(super) async fn spawn_review_thread( }]; let tc = Arc::new(review_turn_context); if tc.environments.single_local_environment_cwd().is_some() { - tc.turn_metadata_state.spawn_git_enrichment_task(); + tc.turn_metadata_state + .spawn_git_enrichment_task(Arc::clone(&sess.services.git_root_discovery)); } // TODO(ccunningham): Review turns currently rely on `spawn_task` for TurnComplete but do not // emit a parent TurnStarted. Consider giving review a full parent turn lifecycle diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 63530cc313..02f93a89eb 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -32,6 +32,7 @@ use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TurnEnvironmentSelections; use codex_skills::SkillError; +use codex_utils_git_discovery::GitRootDiscovery; use std::sync::OnceLock; use tokio::sync::Semaphore; @@ -636,6 +637,7 @@ impl Session { installation_id: String, auth_manager: Arc, models_manager: SharedModelsManager, + git_root_discovery: Arc, model_info: ModelInfo, exec_policy: Arc, tx_event: Sender, @@ -1411,6 +1413,7 @@ impl Session { .with_legacy_custom_ca_fallback(), session_telemetry, models_manager: Arc::clone(&models_manager), + git_root_discovery, tool_approvals: Mutex::new(ApprovalStore::default()), guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index d6c7010365..4113d14aec 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6327,6 +6327,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, + Arc::default(), model_info, Arc::new(ExecPolicyManager::default()), tx_event, @@ -6539,6 +6540,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { .with_legacy_custom_ca_fallback(), session_telemetry: session_telemetry.clone(), models_manager: Arc::clone(&models_manager), + git_root_discovery: Arc::default(), tool_approvals: Mutex::new(ApprovalStore::default()), guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), @@ -6782,6 +6784,7 @@ async fn make_session_with_config_and_rx( "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, + Arc::default(), model_info, Arc::new(ExecPolicyManager::default()), tx_event, @@ -6908,6 +6911,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx( "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, + Arc::default(), model_info, Arc::new(ExecPolicyManager::default()), tx_event, @@ -8820,6 +8824,7 @@ where .with_legacy_custom_ca_fallback(), session_telemetry: session_telemetry.clone(), models_manager: Arc::clone(&models_manager), + git_root_discovery: Arc::default(), tool_approvals: Mutex::new(ApprovalStore::default()), guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index f3e2970814..611f3b798c 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -1287,6 +1287,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { installation_id: "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, + git_root_discovery: Arc::default(), environment_manager: Arc::new(EnvironmentManager::default_for_tests()), skills_service, plugins_manager, diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 44c62cac67..e5a371e9d0 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -1035,7 +1035,9 @@ impl Session { .single_local_environment_cwd() .is_some() { - turn_context.turn_metadata_state.spawn_git_enrichment_task(); + turn_context + .turn_metadata_state + .spawn_git_enrichment_task(Arc::clone(&self.services.git_root_discovery)); } turn_context } diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 0ef84c9b97..e5f96eaf18 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -40,6 +40,7 @@ use codex_rollout_trace::ThreadTraceContext; use codex_skills_extension::HostSkillsService; use codex_thread_store::LiveThread; use codex_thread_store::ThreadStore; +use codex_utils_git_discovery::GitRootDiscovery; use tokio::runtime::Handle; use tokio::sync::Mutex; @@ -64,6 +65,7 @@ pub(crate) struct SessionServices { /// Upload-only clients shared across turns without logging signed blob URLs. pub(crate) openai_file_upload_client_pool: RouteAwareClientPool, pub(crate) models_manager: SharedModelsManager, + pub(crate) git_root_discovery: Arc, pub(crate) session_telemetry: SessionTelemetry, pub(crate) tool_approvals: Mutex, pub(crate) guardian_rejection_circuit_breaker: Mutex, diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index c316178bde..cfa619697f 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -84,6 +84,7 @@ use codex_thread_store::ThreadStore; use codex_thread_store::ThreadStoreError; use codex_thread_store::UpdateThreadMetadataParams; use codex_utils_absolute_path::AbsolutePathBuf; +use codex_utils_git_discovery::GitRootDiscovery; use futures::StreamExt; use futures::stream::FuturesUnordered; use std::collections::HashMap; @@ -345,6 +346,7 @@ pub(crate) struct ThreadManagerState { thread_id_generator: ThreadIdGenerator, auth_manager: Arc, models_manager: SharedModelsManager, + git_root_discovery: Arc, environment_manager: Arc, starting_mcp_runtimes: std::sync::Mutex>>, skills_service: Arc, @@ -472,6 +474,7 @@ impl ThreadManager { thread_created_tx, thread_id_generator: default_thread_id_generator(), models_manager, + git_root_discovery: Arc::default(), environment_manager, starting_mcp_runtimes: std::sync::Mutex::new(Vec::new()), skills_service, @@ -619,6 +622,7 @@ impl ThreadManager { thread_id_generator: default_thread_id_generator(), models_manager: create_model_provider(provider, Some(auth_manager.clone())) .models_manager(codex_home, /*config_model_catalog*/ None), + git_root_discovery: Arc::default(), environment_manager, starting_mcp_runtimes: std::sync::Mutex::new(Vec::new()), skills_service, @@ -779,6 +783,10 @@ impl ThreadManager { Ok(()) } + pub(crate) fn git_root_discovery(&self) -> Arc { + Arc::clone(&self.state.git_root_discovery) + } + pub fn get_models_manager(&self) -> SharedModelsManager { self.state.models_manager.clone() } @@ -1979,6 +1987,7 @@ impl ThreadManagerState { installation_id: self.installation_id.clone(), auth_manager, models_manager: Arc::clone(&self.models_manager), + git_root_discovery: Arc::clone(&self.git_root_discovery), environment_manager: Arc::clone(&self.environment_manager), skills_service: Arc::clone(&self.skills_service), plugins_manager: Arc::clone(&self.plugins_manager), diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index 9d4ecc2113..56d962c1fc 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -1,6 +1,6 @@ use std::collections::BTreeMap; use std::collections::HashMap; -use std::path::Path; +use std::future::Future; use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; @@ -11,7 +11,7 @@ use std::sync::atomic::Ordering; use serde_json::Value; use tokio::sync::watch; -use tokio::task::JoinHandle; +use tokio_util::task::AbortOnDropHandle; use crate::responses_metadata::AGENT_NAME_KEY; use crate::responses_metadata::CodexResponsesMetadata; @@ -28,7 +28,6 @@ use crate::sandbox_tags::SandboxTags; use crate::sandbox_tags::record_policy_metadata; use codex_git_utils::SanitizedGitUrl; 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_in_repo; use codex_git_utils::get_head_commit_hash; use codex_protocol::AgentPath; @@ -40,6 +39,10 @@ use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadSource; use codex_utils_absolute_path::AbsolutePathBuf; +use codex_utils_git_discovery::GitRootDiscovery; + +// Memory requests await this optional metadata before sending their model request. +const MEMORY_GIT_METADATA_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1); const MODEL_KEY: &str = "model"; const REASONING_EFFORT_KEY: &str = "reasoning_effort"; @@ -79,6 +82,7 @@ impl From for TurnMetadataWorkspace { #[allow(clippy::too_many_arguments)] pub async fn detached_memory_responses_metadata( + thread_manager: &crate::ThreadManager, installation_id: String, session_id: String, thread_id: String, @@ -93,7 +97,11 @@ pub async fn detached_memory_responses_metadata( thread_source: Some(ThreadSource::MemoryConsolidation), subagent_header: subagent_header_value(session_source), sandbox: sandbox.map(ToString::to_string), - workspaces: memory_workspaces(cwd).await, + workspaces: memory_workspaces( + cwd, + thread_manager.git_root_discovery().discover(cwd.clone()), + ) + .await, ..CodexResponsesMetadata::new(installation_id, session_id, thread_id, window_id) }; record_policy_metadata(permission_profile, cwd.as_path(), &mut metadata); @@ -103,7 +111,6 @@ pub async fn detached_memory_responses_metadata( #[derive(Debug)] pub(crate) struct TurnMetadataState { cwd: AbsolutePathBuf, - repo_root: Option, session_id: String, thread_id: String, agent_name: String, @@ -129,7 +136,7 @@ pub(crate) struct TurnMetadataState { responses_api_metadata: RwLock>, responsesapi_client_metadata: RwLock>, user_input_requested_during_turn: AtomicBool, - enrichment_task: Mutex>>, + enrichment_task: Mutex>>, git_enrichment_complete: watch::Sender, } @@ -180,7 +187,6 @@ impl TurnMetadataState { auto_review_enabled: bool, model_info: &ModelInfo, ) -> Self { - let repo_root = get_git_repo_root(&cwd); let sandbox_tags = SandboxTags::new( permission_profile, cwd.as_path(), @@ -193,7 +199,6 @@ impl TurnMetadataState { .to_string(); Self { cwd, - repo_root, session_id, thread_id, agent_name, @@ -461,11 +466,10 @@ impl TurnMetadataState { .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(turn_started_at_unix_ms); } - pub(crate) fn spawn_git_enrichment_task(self: &Arc) { - let Some(repo_root) = self.repo_root.clone() else { - return; - }; - + pub(crate) fn spawn_git_enrichment_task( + self: &Arc, + git_root_discovery: Arc, + ) { let mut task_guard = self .enrichment_task .lock() @@ -475,16 +479,16 @@ impl TurnMetadataState { } self.git_enrichment_complete.send_replace(/*value*/ false); - let state = Arc::clone(self); - *task_guard = Some(tokio::spawn(async move { - let workspace_git_metadata = state.fetch_workspace_git_metadata(&repo_root).await; - - if !workspace_git_metadata.is_empty() { - let mut workspaces = BTreeMap::new(); - workspaces.insert( - repo_root.to_string_lossy().into_owned(), - workspace_git_metadata.into(), - ); + // History-only contexts have no active task to cancel enrichment. Keep the + // state weak across awaits so dropping its last owner aborts this waiter. + let state = Arc::downgrade(self); + let cwd = self.cwd.clone(); + *task_guard = Some(AbortOnDropHandle::new(tokio::spawn(async move { + let workspaces = git_workspaces(&cwd, git_root_discovery.discover(cwd.clone())).await; + let Some(state) = state.upgrade() else { + return; + }; + if !workspaces.is_empty() { *state .enriched_workspaces .write() @@ -492,7 +496,7 @@ impl TurnMetadataState { } state.git_enrichment_complete.send_replace(/*value*/ true); - })); + }))); } pub(crate) async fn wait_for_git_enrichment(&self) { @@ -510,25 +514,22 @@ impl TurnMetadataState { self.git_enrichment_complete.send_replace(/*value*/ true); } } - - 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_in_repo(&self.cwd, repo_root), - ); - let latest_git_commit_hash = head_commit_hash.map(|sha| sha.0); - - WorkspaceGitMetadata { - associated_remote_urls, - latest_git_commit_hash, - has_changes, - } - } } -async fn memory_workspaces(cwd: &AbsolutePathBuf) -> BTreeMap { - let Some(repo_root) = get_git_repo_root(cwd) else { +async fn memory_workspaces( + cwd: &AbsolutePathBuf, + repo_root: impl Future>, +) -> BTreeMap { + tokio::time::timeout(MEMORY_GIT_METADATA_TIMEOUT, git_workspaces(cwd, repo_root)) + .await + .unwrap_or_default() +} + +async fn git_workspaces( + cwd: &AbsolutePathBuf, + repo_root: impl Future>, +) -> BTreeMap { + let Some(repo_root) = repo_root.await else { return BTreeMap::new(); }; let (head_commit_hash, associated_remote_urls, has_changes) = tokio::join!( diff --git a/codex-rs/core/src/turn_metadata_tests.rs b/codex-rs/core/src/turn_metadata_tests.rs index 2e69b43d83..64cb45b8d7 100644 --- a/codex-rs/core/src/turn_metadata_tests.rs +++ b/codex-rs/core/src/turn_metadata_tests.rs @@ -142,7 +142,13 @@ async fn wait_for_git_enrichment(state: &TurnMetadataState) -> Value { async fn detached_memory_responses_metadata_omits_turn_identity() { let (_temp_dir, repo_path) = create_clean_git_repo("repo-東京").await; + let thread_manager = crate::ThreadManager::with_models_provider_for_tests( + codex_login::CodexAuth::from_api_key("test"), + crate::config::test_config().await.model_provider, + ); + let header = detached_memory_responses_metadata( + &thread_manager, String::new(), String::new(), String::new(), @@ -195,7 +201,13 @@ async fn detached_memory_responses_metadata_omits_empty_workspace_metadata() { let temp_dir = TempDir::new().expect("temp dir"); let cwd = temp_dir.path().abs(); + let thread_manager = crate::ThreadManager::with_models_provider_for_tests( + codex_login::CodexAuth::from_api_key("test"), + crate::config::test_config().await.model_provider, + ); + let header = detached_memory_responses_metadata( + &thread_manager, String::new(), String::new(), String::new(), @@ -220,6 +232,21 @@ async fn detached_memory_responses_metadata_omits_empty_workspace_metadata() { ); } +#[tokio::test(start_paused = true)] +async fn memory_workspaces_times_out_pending_git_discovery() { + let temp_dir = TempDir::new().expect("temp dir"); + let cwd = temp_dir.path().abs(); + + let workspaces = tokio::time::timeout( + Duration::from_secs(2), + memory_workspaces(&cwd, std::future::pending()), + ) + .await + .expect("memory metadata should stop waiting for Git discovery"); + + assert!(workspaces.is_empty()); +} + #[test] fn turn_metadata_state_includes_sandbox_metadata() { let temp_dir = TempDir::new().expect("temp dir"); @@ -1105,7 +1132,7 @@ async fn turn_metadata_state_preserves_subagent_parent_after_git_enrichment() { &model_info_from_slug("gpt-5.4"), )); - state.spawn_git_enrichment_task(); + state.spawn_git_enrichment_task(Arc::default()); let json = wait_for_git_enrichment(&state).await; assert!(json.get("forked_from_thread_id").is_none()); @@ -1145,20 +1172,23 @@ async fn turn_metadata_state_coalesces_concurrent_git_enrichment() { /*auto_review_enabled*/ false, &model_info_from_slug("gpt-5.4"), )); + let git_root_discovery = Arc::default(); let barrier = Arc::new(tokio::sync::Barrier::new(8)); let tasks = (0..8) .map(|_| { let state = Arc::clone(&state); let barrier = Arc::clone(&barrier); + let git_root_discovery = Arc::clone(&git_root_discovery); tokio::spawn(async move { barrier.wait().await; - state.spawn_git_enrichment_task(); + state.spawn_git_enrichment_task(git_root_discovery); state .enrichment_task .lock() .expect("enrichment task lock") .as_ref() .expect("enrichment task") + .abort_handle() .id() }) }) @@ -1181,6 +1211,82 @@ async fn turn_metadata_state_coalesces_concurrent_git_enrichment() { ); } +#[tokio::test(flavor = "current_thread")] +async fn dropping_turn_metadata_aborts_unused_git_enrichment() { + let temp_dir = TempDir::new().expect("temp dir"); + let state = Arc::new(TurnMetadataState::new( + "session-a".to_string(), + "thread-a".to_string(), + /*forked_from_thread_id*/ None, + /*parent_thread_id*/ None, + &SessionSource::Exec, + /*thread_source*/ None, + "turn-a".to_string(), + temp_dir.path().abs(), + &PermissionProfile::read_only(), + WindowsSandboxLevel::Disabled, + /*enforce_managed_network*/ false, + /*auto_review_enabled*/ false, + &model_info_from_slug("gpt-5.4"), + )); + let weak_state = Arc::downgrade(&state); + state.spawn_git_enrichment_task(Arc::default()); + let task = state + .enrichment_task + .lock() + .expect("enrichment task lock") + .as_ref() + .expect("enrichment task") + .abort_handle(); + + // Drop the history-only context before its task can run. A strong task-owned + // reference would retain the metadata even after every caller has left. + drop(state); + assert!(weak_state.upgrade().is_none()); + + tokio::time::timeout(Duration::from_secs(2), async { + while !task.is_finished() { + tokio::task::yield_now().await; + } + }) + .await + .expect("dropping turn metadata should abort its enrichment task"); +} + +#[tokio::test] +async fn git_enrichment_discovers_the_repository_when_it_starts() { + let (workspace, repo_path) = create_clean_git_repo("repo").await; + let saved_git = workspace.path().join("saved-git"); + std::fs::rename(repo_path.join(".git"), &saved_git).expect("move Git metadata"); + let state = Arc::new(TurnMetadataState::new( + "session-a".to_string(), + "thread-a".to_string(), + /*forked_from_thread_id*/ None, + /*parent_thread_id*/ None, + &SessionSource::Exec, + /*thread_source*/ None, + "turn-a".to_string(), + repo_path.clone(), + &PermissionProfile::read_only(), + WindowsSandboxLevel::Disabled, + /*enforce_managed_network*/ false, + /*auto_review_enabled*/ false, + &model_info_from_slug("gpt-5.4"), + )); + std::fs::rename(saved_git, repo_path.join(".git")).expect("restore Git metadata"); + + state.spawn_git_enrichment_task(Arc::default()); + let metadata = wait_for_git_enrichment(&state).await; + assert_eq!( + metadata["workspaces"] + .as_object() + .expect("workspace metadata") + .keys() + .collect::>(), + vec![&repo_path.to_string_lossy().into_owned()], + ); +} + #[tokio::test] async fn turn_metadata_state_git_enrichment_cancellation_is_retryable_and_errors_stay_empty() { let (_temp_dir, repo_path) = create_clean_git_repo("repo").await; @@ -1200,7 +1306,8 @@ async fn turn_metadata_state_git_enrichment_cancellation_is_retryable_and_errors /*auto_review_enabled*/ false, &model_info_from_slug("gpt-5.4"), )); - state.spawn_git_enrichment_task(); + let git_root_discovery = Arc::default(); + state.spawn_git_enrichment_task(Arc::clone(&git_root_discovery)); state.cancel_git_enrichment_task(); assert!( state @@ -1214,7 +1321,7 @@ async fn turn_metadata_state_git_enrichment_cancellation_is_retryable_and_errors .expect("cancelled git enrichment should unblock waiters"); assert!(state.current_workspaces().is_empty()); - state.spawn_git_enrichment_task(); + state.spawn_git_enrichment_task(git_root_discovery); let json = wait_for_git_enrichment(&state).await; assert_eq!( json["workspaces"].as_object().map(serde_json::Map::len), @@ -1243,7 +1350,7 @@ async fn turn_metadata_state_git_enrichment_cancellation_is_retryable_and_errors /*auto_review_enabled*/ false, &model_info_from_slug("gpt-5.4"), )); - invalid_state.spawn_git_enrichment_task(); + invalid_state.spawn_git_enrichment_task(Arc::default()); tokio::time::timeout( Duration::from_secs(2), invalid_state.wait_for_git_enrichment(), diff --git a/codex-rs/core/tests/suite/git_enrichment.rs b/codex-rs/core/tests/suite/git_enrichment.rs index 0bad1eb28b..989683fe82 100644 --- a/codex-rs/core/tests/suite/git_enrichment.rs +++ b/codex-rs/core/tests/suite/git_enrichment.rs @@ -92,11 +92,24 @@ fn expected_workspace(repo: &Path, head: &str, has_changes: bool) -> Value { }) } +enum GitMetadataAtStartup { + Present, + RestoredAfterPrewarm, +} + +#[test_case(GitMetadataAtStartup::Present; "git present at startup")] +#[test_case(GitMetadataAtStartup::RestoredAfterPrewarm; "git restored after prewarm")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn startup_prewarm_skips_git_enrichment_and_user_turn_observes_fresh_state() -> Result<()> { +async fn startup_prewarm_skips_git_enrichment_and_user_turn_observes_fresh_state( + git_metadata: GitMetadataAtStartup, +) -> Result<()> { skip_if_no_network!(Ok(())); let (repo, head) = create_git_repo()?; + let saved_git = TempDir::new()?; + if matches!(git_metadata, GitMetadataAtStartup::RestoredAfterPrewarm) { + std::fs::rename(repo.path().join(".git"), saved_git.path().join("git"))?; + } let server = start_websocket_server(vec![vec![ vec![ev_response_created("warm-1"), ev_completed("warm-1")], vec![ @@ -104,7 +117,7 @@ async fn startup_prewarm_skips_git_enrichment_and_user_turn_observes_fresh_state ev_function_call( "wait-for-git", "test_sync_tool", - r#"{"sleep_after_ms":5000}"#, + r#"{"wait_for_git_enrichment":true}"#, ), ev_completed("resp-1"), ], @@ -129,8 +142,12 @@ async fn startup_prewarm_skips_git_enrichment_and_user_turn_observes_fresh_state ) .await? .body_json(); + assert_eq!(prewarm["generate"], json!(false)); assert!(turn_metadata(&prewarm)?.get("workspaces").is_none()); + if matches!(git_metadata, GitMetadataAtStartup::RestoredAfterPrewarm) { + std::fs::rename(saved_git.path().join("git"), repo.path().join(".git"))?; + } std::fs::write(repo.path().join("untracked.txt"), "dirty\n")?; test.submit_turn("inspect the workspace").await?; let turn = server diff --git a/codex-rs/memories/write/src/runtime.rs b/codex-rs/memories/write/src/runtime.rs index 9ce806499e..c4cf8c7d75 100644 --- a/codex-rs/memories/write/src/runtime.rs +++ b/codex-rs/memories/write/src/runtime.rs @@ -269,6 +269,7 @@ impl MemoryStartupContext { let mut client_session = model_client.new_session(); let window_id = format!("{}:0", self.thread_id); let responses_metadata = detached_memory_responses_metadata( + &self.thread_manager, installation_id, session_id_string, self.thread_id.to_string(), diff --git a/codex-rs/utils/git-discovery/BUILD.bazel b/codex-rs/utils/git-discovery/BUILD.bazel new file mode 100644 index 0000000000..8e275e13ed --- /dev/null +++ b/codex-rs/utils/git-discovery/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "git-discovery", + crate_name = "codex_utils_git_discovery", +) diff --git a/codex-rs/utils/git-discovery/Cargo.toml b/codex-rs/utils/git-discovery/Cargo.toml new file mode 100644 index 0000000000..bb5345f9b2 --- /dev/null +++ b/codex-rs/utils/git-discovery/Cargo.toml @@ -0,0 +1,24 @@ +[package] +name = "codex-utils-git-discovery" +version.workspace = true +edition.workspace = true +license.workspace = true + +[lints] +workspace = true + +[dependencies] +codex-git-utils = { workspace = true } +codex-utils-absolute-path = { workspace = true } +futures = { workspace = true, features = ["std"] } +tokio = { workspace = true, features = ["sync"] } +tracing = { workspace = true } + +[dev-dependencies] +futures = { workspace = true, features = ["async-await"] } +pretty_assertions = { workspace = true } +tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "time"] } + +[lib] +doctest = false diff --git a/codex-rs/utils/git-discovery/src/git_root_discovery_tests.rs b/codex-rs/utils/git-discovery/src/git_root_discovery_tests.rs new file mode 100644 index 0000000000..2cc8e14f8b --- /dev/null +++ b/codex-rs/utils/git-discovery/src/git_root_discovery_tests.rs @@ -0,0 +1,185 @@ +//! Verifies shared probe results, capacity retention after cancellation, and +//! fresh discovery after repository changes. Blocked probes must not delay runtime shutdown. + +use super::GitRootDiscovery; +use codex_utils_absolute_path::AbsolutePathBuf; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::mpsc; +use std::time::Duration; +use tokio::time::timeout; + +const PROBE_TIMEOUT: Duration = Duration::from_secs(10); + +/// Simulates a filesystem call that stays blocked until the test releases it. +fn blocked_root_discovery( + capacity: usize, +) -> (Arc, mpsc::Receiver<()>, mpsc::Sender<()>) { + let (entered_tx, entered_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let release_rx = Mutex::new(release_rx); + let discovery = Arc::new(GitRootDiscovery { + capacity, + find_root: Arc::new(move |cwd| { + entered_tx.send(()).expect("signal probe entry"); + let _ = release_rx + .lock() + .expect("release receiver") + .recv_timeout(PROBE_TIMEOUT); + Some(cwd.to_path_buf()) + }), + ..Default::default() + }); + (discovery, entered_rx, release_tx) +} + +#[test] +fn root_discovery_shares_probes_across_callers_and_cancellation() { + const TIMEOUT: Duration = Duration::from_secs(5); + let workspace = tempfile::tempdir().expect("workspace"); + let first_cwd = AbsolutePathBuf::from_absolute_path(workspace.path().join("first")) + .expect("absolute first cwd"); + let second_cwd = AbsolutePathBuf::from_absolute_path(workspace.path().join("second")) + .expect("absolute second cwd"); + let third_cwd = AbsolutePathBuf::from_absolute_path(workspace.path().join("third")) + .expect("absolute third cwd"); + let (discovery, entered_rx, release_tx) = blocked_root_discovery(/*capacity*/ 2); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + + runtime.block_on(async { + let mut first = Box::pin(discovery.discover(first_cwd.clone())); + let mut duplicate = Box::pin(discovery.discover(first_cwd.clone())); + let mut second = Box::pin(discovery.discover(second_cwd.clone())); + assert!(futures::poll!(first.as_mut()).is_pending()); + assert!(futures::poll!(duplicate.as_mut()).is_pending()); + // Sharing the first cwd leaves room for a different directory. + assert!(futures::poll!(second.as_mut()).is_pending()); + entered_rx + .recv_timeout(TIMEOUT) + .expect("first probe started"); + entered_rx + .recv_timeout(TIMEOUT) + .expect("second probe started"); + drop(first); + drop(duplicate); + // Cancelling every waiter does not free capacity while its worker is blocked. + let mut third = Box::pin(discovery.discover(third_cwd.clone())); + assert!(futures::poll!(third.as_mut()).is_pending()); + let mut cancelled = Box::pin(discovery.discover(first_cwd.join("cancelled"))); + assert!(futures::poll!(cancelled.as_mut()).is_pending()); + drop(cancelled); + assert!(entered_rx.try_recv().is_err()); + let mut retry = Box::pin(discovery.discover(first_cwd.clone())); + // The same directory can still join its probe when capacity is full. + assert!(futures::poll!(retry.as_mut()).is_pending()); + let mut follower = Box::pin(discovery.discover(first_cwd.clone())); + assert!(futures::poll!(follower.as_mut()).is_pending()); + + release_tx.send(()).expect("release first probe"); + release_tx.send(()).expect("release second probe"); + release_tx.send(()).expect("release queued probe"); + let roots = timeout(TIMEOUT, async { + (retry.await, follower.await, second.await, third.await) + }) + .await + .expect("running and queued root probes finished"); + assert_eq!( + roots, + ( + Some(first_cwd.to_path_buf()), + Some(first_cwd.to_path_buf()), + Some(second_cwd.to_path_buf()), + Some(third_cwd.to_path_buf()), + ) + ); + entered_rx.try_recv().expect("queued probe started"); + assert!(entered_rx.try_recv().is_err()); + }); +} + +#[tokio::test] +async fn completed_root_discoveries_are_not_cached() { + let workspace = tempfile::tempdir().expect("workspace"); + let cwd = AbsolutePathBuf::from_absolute_path(workspace.path()).expect("absolute cwd"); + let discovery = Arc::new(GitRootDiscovery { + capacity: 1, + ..Default::default() + }); + assert_eq!(discovery.discover(cwd.clone()).await, None); + + std::fs::create_dir(cwd.join(".git")).expect("git directory"); + std::fs::write(cwd.join(".git/HEAD"), "ref: refs/heads/main\n").expect("git HEAD"); + assert_eq!( + discovery.discover(cwd.clone()).await, + Some(cwd.to_path_buf()) + ); + + std::fs::remove_dir_all(cwd.join(".git")).expect("remove Git metadata"); + assert_eq!(discovery.discover(cwd).await, None); +} + +#[tokio::test] +async fn saturated_discovery_does_not_block_other_instances() { + let workspace = tempfile::tempdir().expect("workspace"); + let cwd = AbsolutePathBuf::from_absolute_path(workspace.path()).expect("absolute cwd"); + let (saturated, entered_rx, release_tx) = blocked_root_discovery(/*capacity*/ 1); + let mut blocked = Box::pin(saturated.discover(cwd.clone())); + assert!(futures::poll!(blocked.as_mut()).is_pending()); + entered_rx + .recv_timeout(PROBE_TIMEOUT) + .expect("probe started"); + + let available = Arc::new(GitRootDiscovery::default()); + let root = timeout(Duration::from_secs(5), available.discover(cwd.clone())).await; + release_tx.send(()).expect("release probe"); + assert_eq!(root.expect("independent discovery finishes"), None); + assert_eq!(blocked.await, Some(cwd.to_path_buf())); +} + +#[test] +fn blocked_root_discovery_does_not_delay_runtime_shutdown() { + const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2); + const WORKER_THREADS: usize = 1; + let workspace = tempfile::tempdir().expect("workspace"); + let cwd = AbsolutePathBuf::from_absolute_path(workspace.path()).expect("absolute cwd"); + let (discovery, entered_rx, release_tx) = blocked_root_discovery(/*capacity*/ 1); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(WORKER_THREADS) + .enable_all() + .build() + .expect("runtime"); + let lookup = runtime.spawn({ + let discovery = Arc::clone(&discovery); + async move { discovery.discover(cwd).await } + }); + entered_rx + .recv_timeout(PROBE_TIMEOUT) + .expect("probe started"); + lookup.abort(); + assert!( + runtime + .block_on(lookup) + .expect_err("cancelled lookup") + .is_cancelled() + ); + + let (shutdown_tx, shutdown_rx) = mpsc::channel(); + let shutdown = std::thread::spawn(move || { + drop(runtime); + let _ = shutdown_tx.send(()); + }); + let shutdown_result = shutdown_rx.recv_timeout(SHUTDOWN_TIMEOUT); + let outstanding = discovery + .in_flight + .lock() + .expect("outstanding probes") + .len(); + // Release the probe even if shutdown timed out so a regression cannot hang the test. + release_tx.send(()).expect("release probe"); + shutdown.join().expect("shutdown thread"); + assert_eq!((shutdown_result, outstanding), (Ok(()), 1)); +} diff --git a/codex-rs/utils/git-discovery/src/lib.rs b/codex-rs/utils/git-discovery/src/lib.rs new file mode 100644 index 0000000000..a4febbd727 --- /dev/null +++ b/codex-rs/utils/git-discovery/src/lib.rs @@ -0,0 +1,120 @@ +//! Bounded, shared Git-root discovery for optional metadata. +//! +//! Each instance bounds detached probes and shares one outstanding probe per +//! supplied cwd. Workers retain their entries through caller cancellation and remove +//! them when filesystem work ends; completed results are not cached. Capacity waits +//! are cancellable; keys are compared without filesystem canonicalization. +//! Probes never enter Tokio's blocking pool, so a stuck filesystem cannot keep its +//! runtime shutdown waiting for optional metadata work. + +use std::collections::HashMap; +use std::path::Path; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::Mutex; + +use codex_git_utils::get_git_repo_root; +use codex_utils_absolute_path::AbsolutePathBuf; +use futures::FutureExt; +use futures::future::BoxFuture; +use futures::future::Shared; +use tokio::sync::Notify; +use tokio::sync::oneshot; + +const MAX_CONCURRENT_ROOT_PROBES: usize = 8; + +type SharedRootProbe = Shared>>; +type RootFinder = dyn Fn(&Path) -> Option + Send + Sync; + +/// Shares outstanding metadata probes by cwd and bounds probes across directories. +/// Entries survive caller cancellation but are not cached after filesystem work ends. +pub struct GitRootDiscovery { + capacity: usize, + in_flight: Mutex>, + find_root: Arc, + capacity_changed: Notify, +} + +impl Default for GitRootDiscovery { + fn default() -> Self { + Self { + capacity: MAX_CONCURRENT_ROOT_PROBES, + in_flight: Mutex::new(HashMap::new()), + find_root: Arc::new(get_git_repo_root), + capacity_changed: Notify::new(), + } + } +} + +impl GitRootDiscovery { + /// Joins the cwd's existing probe, waiting for capacity to start a detached + /// worker when necessary. Dropping the future cancels only the wait. + pub async fn discover(self: &Arc, cwd: AbsolutePathBuf) -> Option { + let (result, new_probe) = loop { + // notify_waiters reaches futures created before the capacity check, + // even if they have not been polled yet. + let capacity_changed = self.capacity_changed.notified(); + { + let mut in_flight = self + .in_flight + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(result) = in_flight.get(&cwd) { + break (result.clone(), None); + } + if in_flight.len() < self.capacity { + let (result_tx, result_rx) = oneshot::channel(); + let result = async move { result_rx.await.unwrap_or_default() } + .boxed() + .shared(); + in_flight.insert(cwd.clone(), result.clone()); + let probe = ProbeGuard { + cwd, + discovery: Arc::clone(self), + }; + break (result, Some((probe, result_tx))); + } + } + capacity_changed.await; + }; + if let Some((probe, result_tx)) = new_probe { + // Other waiters for this cwd can now join the new probe. + self.capacity_changed.notify_waiters(); + // A failed spawn drops the closure and its guard, so do not hold the map lock. + // Dropping the handle detaches the thread; Tokio never owns or joins it. + let worker = std::thread::Builder::new() + .name("codex-git-root".to_string()) + .spawn(move || { + let root = (probe.discovery.find_root)(probe.cwd.as_path()); + drop(probe); + let _ = result_tx.send(root); + }); + if let Err(error) = worker { + tracing::warn!(%error, "failed to start optional Git root probe"); + } + } + result.await + } +} + +/// Keeps an outstanding entry owned by its blocking worker, including during +/// cancellation or unwinding, rather than by any caller waiting for the result. +struct ProbeGuard { + cwd: AbsolutePathBuf, + discovery: Arc, +} + +impl Drop for ProbeGuard { + fn drop(&mut self) { + self.discovery + .in_flight + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&self.cwd); + self.discovery.capacity_changed.notify_waiters(); + } +} + +#[cfg(test)] +#[path = "git_root_discovery_tests.rs"] +mod tests;