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;