From 33d394c69e23e8906e1d0b9a3807f10ea4b7c294 Mon Sep 17 00:00:00 2001 From: acrognale-oai Date: Wed, 22 Jul 2026 10:42:45 +0000 Subject: [PATCH] Skip Git enrichment for prewarm and Guardian turns (#34728) ## Why Startup prewarms and Guardian review sessions do not need workspace Git metadata. Collecting it for those requests performs redundant work, while user turns still need a fresh view of the repository. ## What changed - Add a per-session Git enrichment policy. - Skip enrichment for startup prewarm, preview, and Guardian review turns. - Keep fresh enrichment enabled for regular sessions and one-shot delegated turns. ## Testing - Cover prewarm and Guardian requests omitting workspace metadata while subsequent user turns observe repository changes. - Cover concurrent repositories and worktrees retaining distinct metadata. - Verify concurrent enrichment is coalesced and canceled enrichment can be retried. GitOrigin-RevId: 4a68615288bb568c75f2e487d9c0776e3e80f191 --- codex-rs/core/src/codex_delegate.rs | 4 + codex-rs/core/src/codex_delegate_tests.rs | 1 + codex-rs/core/src/guardian/review_session.rs | 2 + codex-rs/core/src/session/mod.rs | 9 + 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 | 12 +- codex-rs/core/src/thread_manager.rs | 2 + codex-rs/core/src/turn_metadata_tests.rs | 171 +++++- codex-rs/core/tests/suite/git_enrichment.rs | 501 ++++++++++++++++++ codex-rs/core/tests/suite/mod.rs | 1 + 12 files changed, 691 insertions(+), 21 deletions(-) create mode 100644 codex-rs/core/tests/suite/git_enrichment.rs diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index eba2bc6f8b..bc16482174 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -46,6 +46,7 @@ use crate::mcp_tool_call::build_guardian_mcp_tool_review_request; use crate::mcp_tool_call::is_mcp_tool_approval_question_id; use crate::mcp_tool_call::lookup_mcp_tool_metadata; use crate::mcp_tool_call::mcp_approvals_reviewer; +use crate::session::GitEnrichmentPolicy; use crate::session::SUBMISSION_CHANNEL_CAPACITY; use crate::session::SessionIo; use crate::session::SessionSpawnArgs; @@ -82,6 +83,7 @@ pub(crate) async fn run_codex_thread_interactive( cancel_token: CancellationToken, subagent_source: SubAgentSource, initial_history: Option, + git_enrichment_policy: GitEnrichmentPolicy, ) -> Result<(Arc, SessionIo), CodexErr> { let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_ops, rx_ops) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); @@ -133,6 +135,7 @@ pub(crate) async fn run_codex_thread_interactive( attestation_provider: parent_session.services.attestation_provider.clone(), external_time_provider: Some(Arc::clone(&parent_session.services.time_provider)), inherited_multi_agent_version: Some(MultiAgentVersion::Disabled), + git_enrichment_policy, })) .or_cancel(&cancel_token) .await??; @@ -217,6 +220,7 @@ pub(crate) async fn run_codex_thread_one_shot( child_cancel.clone(), subagent_source, initial_history, + GitEnrichmentPolicy::Fresh, )) .await?; diff --git a/codex-rs/core/src/codex_delegate_tests.rs b/codex-rs/core/src/codex_delegate_tests.rs index 98adcfd385..c74a01c584 100644 --- a/codex-rs/core/src/codex_delegate_tests.rs +++ b/codex-rs/core/src/codex_delegate_tests.rs @@ -210,6 +210,7 @@ async fn run_codex_thread_interactive_respects_pre_cancelled_spawn() { cancel_token, SubAgentSource::Review, /*initial_history*/ None, + crate::session::GitEnrichmentPolicy::Fresh, ), ) .await diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 054d28ac56..dfc5457143 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -43,6 +43,7 @@ use crate::config::NetworkProxySpec; use crate::config::Permissions; use crate::context::ContextualUserFragment; use crate::context::GuardianFollowupReviewReminder; +use crate::session::GitEnrichmentPolicy; use crate::session::SessionIo; use crate::session::session::Session; use crate::session::turn_context::TurnContext; @@ -678,6 +679,7 @@ async fn spawn_guardian_review_session( cancel_token.clone(), SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()), initial_history, + GitEnrichmentPolicy::Skip, )) .await?; diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 01a7d05881..c77aa11095 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -405,6 +405,12 @@ pub(crate) struct SessionIo { pub(crate) type SessionLoopTermination = Shared>; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum GitEnrichmentPolicy { + Fresh, + Skip, +} + pub(crate) struct SessionSpawnArgs { pub(crate) config: Config, pub(crate) allow_provider_model_fallback: bool, @@ -445,6 +451,7 @@ pub(crate) struct SessionSpawnArgs { pub(crate) attestation_provider: Option>, pub(crate) external_time_provider: Option>, pub(crate) inherited_multi_agent_version: Option, + pub(crate) git_enrichment_policy: GitEnrichmentPolicy, } pub(crate) fn resolve_multi_agent_version( @@ -533,6 +540,7 @@ impl Session { attestation_provider, external_time_provider, inherited_multi_agent_version, + git_enrichment_policy, } = args; let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_event, rx_event) = async_channel::unbounded(); @@ -720,6 +728,7 @@ impl Session { attestation_provider, external_time_provider, multi_agent_version, + git_enrichment_policy, )) .await .map_err(|e| { diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 30c508045c..b6394f8ed0 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -44,6 +44,7 @@ pub(crate) struct Session { pub(crate) input_queue: InputQueue, pub(crate) guardian_review_session: GuardianReviewSessionManager, pub(crate) services: SessionServices, + pub(super) git_enrichment_policy: GitEnrichmentPolicy, pub(super) next_internal_sub_id: AtomicU64, } @@ -501,6 +502,7 @@ impl Session { attestation_provider: Option>, external_time_provider: Option>, multi_agent_version: Option, + git_enrichment_policy: GitEnrichmentPolicy, ) -> anyhow::Result> { debug!( "Configuring session: model={}; provider={:?}", @@ -1148,6 +1150,7 @@ impl Session { input_queue: InputQueue::new(), guardian_review_session: GuardianReviewSessionManager::default(), services, + git_enrichment_policy, next_internal_sub_id: AtomicU64::new(0), }); if let Some(network_policy_decider_session) = network_policy_decider_session { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index b2d13e4cf7..fe15ddac5b 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -5270,6 +5270,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { /*attestation_provider*/ None, /*external_time_provider*/ None, Some(config.multi_agent_version_from_features()), + GitEnrichmentPolicy::Fresh, ) .await; @@ -5535,6 +5536,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { input_queue: super::input_queue::InputQueue::new(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, + git_enrichment_policy: GitEnrichmentPolicy::Fresh, next_internal_sub_id: AtomicU64::new(0), }; @@ -5657,6 +5659,7 @@ async fn make_session_with_config_and_rx( /*attestation_provider*/ None, /*external_time_provider*/ None, Some(config.multi_agent_version_from_features()), + GitEnrichmentPolicy::Fresh, ) .await?; @@ -5771,6 +5774,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx( /*attestation_provider*/ None, /*external_time_provider*/ None, Some(config.multi_agent_version_from_features()), + GitEnrichmentPolicy::Fresh, ) .await?; @@ -7694,6 +7698,7 @@ where input_queue: super::input_queue::InputQueue::new(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, + git_enrichment_policy: GitEnrichmentPolicy::Fresh, next_internal_sub_id: AtomicU64::new(0), }); diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index d6c103a951..c09b53b353 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -765,6 +765,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { attestation_provider: None, external_time_provider: None, inherited_multi_agent_version: None, + git_enrichment_policy: GitEnrichmentPolicy::Skip, }) .await .expect("spawn guardian subagent"); diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index fbb6f96493..a30a453540 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -673,6 +673,7 @@ impl Session { session_configuration, final_output_json_schema, TurnMultiAgentRuntime::ResolveAndStore, + self.git_enrichment_policy, ) .await } @@ -687,6 +688,7 @@ impl Session { session_configuration, /*final_output_json_schema*/ None, TurnMultiAgentRuntime::Preview, + GitEnrichmentPolicy::Skip, ) .await } @@ -698,6 +700,7 @@ impl Session { session_configuration: SessionConfiguration, final_output_json_schema: Option>, multi_agent_runtime: TurnMultiAgentRuntime, + git_enrichment_policy: GitEnrichmentPolicy, ) -> Arc { let turn_environments = self.services.turn_environments.snapshot().await; let primary_turn_environment = turn_environments.primary(); @@ -792,10 +795,11 @@ impl Session { turn_context.final_output_json_schema = final_schema; } let turn_context = Arc::new(turn_context); - if turn_context - .environments - .single_local_environment_cwd() - .is_some() + if git_enrichment_policy == GitEnrichmentPolicy::Fresh + && turn_context + .environments + .single_local_environment_cwd() + .is_some() { turn_context.turn_metadata_state.spawn_git_enrichment_task(); } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 70599398bb..4bfd964cb2 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -10,6 +10,7 @@ use crate::environment_selection::TurnEnvironmentSnapshot; use crate::environment_selection::default_thread_environment_selections; use crate::mcp::McpManager; use crate::rollout::truncation; +use crate::session::GitEnrichmentPolicy; use crate::session::INITIAL_SUBMIT_ID; use crate::session::SessionIo; use crate::session::SessionSpawnArgs; @@ -1685,6 +1686,7 @@ impl ThreadManagerState { attestation_provider: self.attestation_provider.clone(), external_time_provider: self.external_time_provider.clone(), inherited_multi_agent_version: multi_agent_version, + git_enrichment_policy: GitEnrichmentPolicy::Fresh, })) .await?; let new_thread = self diff --git a/codex-rs/core/src/turn_metadata_tests.rs b/codex-rs/core/src/turn_metadata_tests.rs index 788a785184..99eb4d3753 100644 --- a/codex-rs/core/src/turn_metadata_tests.rs +++ b/codex-rs/core/src/turn_metadata_tests.rs @@ -20,6 +20,7 @@ use core_test_support::PathExt; use pretty_assertions::assert_eq; use serde_json::Value; use std::collections::HashMap; +use std::sync::Arc; use std::time::Duration; use tempfile::TempDir; use tokio::process::Command; @@ -109,6 +110,25 @@ async fn create_clean_git_repo(repo_name: &str) -> (TempDir, AbsolutePathBuf) { (temp_dir, repo_path) } +async fn wait_for_git_enrichment(state: &TurnMetadataState) -> Value { + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let header = test_turn_metadata_header(state); + let json: Value = serde_json::from_str(&header).expect("json"); + if json + .get("workspaces") + .and_then(Value::as_object) + .is_some_and(|workspaces| !workspaces.is_empty()) + { + return json; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("git enrichment should complete") +} + #[tokio::test] async fn detached_memory_responses_metadata_omits_turn_identity() { let (_temp_dir, repo_path) = create_clean_git_repo("repo-東京").await; @@ -762,23 +782,7 @@ async fn turn_metadata_state_preserves_lineage_after_git_enrichment() { ); state.spawn_git_enrichment_task(); - - let json = tokio::time::timeout(Duration::from_secs(2), async { - loop { - let header = test_turn_metadata_header(&state); - let json: Value = serde_json::from_str(&header).expect("json"); - if json - .get("workspaces") - .and_then(Value::as_object) - .is_some_and(|workspaces| !workspaces.is_empty()) - { - return json; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - }) - .await - .expect("git enrichment should complete"); + let json = wait_for_git_enrichment(&state).await; assert_eq!( json["forked_from_thread_id"].as_str(), @@ -790,3 +794,136 @@ async fn turn_metadata_state_preserves_lineage_after_git_enrichment() { ); assert_eq!(json["subagent_kind"].as_str(), Some("thread_spawn")); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn turn_metadata_state_coalesces_concurrent_git_enrichment() { + let (_temp_dir, repo_path) = create_clean_git_repo("repo").await; + let head = Command::new("git") + .args(["rev-parse", "HEAD"]) + .current_dir(&repo_path) + .output() + .await + .expect("git rev-parse HEAD"); + let head = String::from_utf8(head.stdout) + .expect("commit hash") + .trim() + .to_string(); + let permission_profile = PermissionProfile::read_only(); + 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(), + &permission_profile, + WindowsSandboxLevel::Disabled, + /*enforce_managed_network*/ false, + )); + let barrier = Arc::new(tokio::sync::Barrier::new(8)); + let tasks = (0..8) + .map(|_| { + let state = Arc::clone(&state); + let barrier = Arc::clone(&barrier); + tokio::spawn(async move { + barrier.wait().await; + state.spawn_git_enrichment_task(); + state + .enrichment_task + .lock() + .expect("enrichment task lock") + .as_ref() + .expect("enrichment task") + .id() + }) + }) + .collect::>(); + let mut task_ids = Vec::new(); + for task in tasks { + task_ids.push(task.await.expect("spawn task")); + } + assert!(task_ids.iter().all(|task_id| *task_id == task_ids[0])); + + let json = wait_for_git_enrichment(state.as_ref()).await; + assert_eq!( + json["workspaces"], + serde_json::json!({ + repo_path.to_string_lossy().as_ref(): { + "latest_git_commit_hash": head, + "has_changes": false, + } + }) + ); +} + +#[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; + let permission_profile = PermissionProfile::read_only(); + let state = 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, + &permission_profile, + WindowsSandboxLevel::Disabled, + /*enforce_managed_network*/ false, + ); + state.spawn_git_enrichment_task(); + state.cancel_git_enrichment_task(); + assert!( + state + .enrichment_task + .lock() + .expect("enrichment task lock") + .is_none() + ); + assert!(state.current_workspaces().is_empty()); + + state.spawn_git_enrichment_task(); + let json = wait_for_git_enrichment(&state).await; + assert_eq!( + json["workspaces"].as_object().map(serde_json::Map::len), + Some(1) + ); + + let invalid_repo = TempDir::new().expect("invalid repo"); + std::fs::create_dir(invalid_repo.path().join(".git")).expect("invalid git directory"); + let invalid_state = 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-b".to_string(), + invalid_repo.path().abs(), + &permission_profile, + WindowsSandboxLevel::Disabled, + /*enforce_managed_network*/ false, + ); + invalid_state.spawn_git_enrichment_task(); + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if invalid_state + .enrichment_task + .lock() + .expect("enrichment task lock") + .as_ref() + .is_some_and(tokio::task::JoinHandle::is_finished) + { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("failed git enrichment should complete"); + assert!(invalid_state.current_workspaces().is_empty()); +} diff --git a/codex-rs/core/tests/suite/git_enrichment.rs b/codex-rs/core/tests/suite/git_enrichment.rs new file mode 100644 index 0000000000..f1b6fb3758 --- /dev/null +++ b/codex-rs/core/tests/suite/git_enrichment.rs @@ -0,0 +1,501 @@ +use anyhow::Context; +use anyhow::Result; +use anyhow::ensure; +use codex_core::StartThreadOptions; +#[cfg(not(target_os = "windows"))] +use codex_core::config::Constrained; +#[cfg(not(target_os = "windows"))] +use codex_core::sandboxing::SandboxPermissions; +#[cfg(not(target_os = "windows"))] +use codex_protocol::config_types::ApprovalsReviewer; +#[cfg(not(target_os = "windows"))] +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::InitialHistory; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::ThreadSource; +use codex_protocol::user_input::UserInput; +use core_test_support::PathBufExt; +use core_test_support::responses::ev_assistant_message; +use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_function_call; +use core_test_support::responses::ev_response_created; +use core_test_support::responses::start_websocket_server; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use std::path::Path; +use std::process::Command; +use std::time::Duration; +use tempfile::TempDir; + +const ORIGIN_URL: &str = "https://example.invalid/cxa5426/repo.git"; + +fn run_git(cwd: &Path, args: &[&str]) -> Result { + let output = Command::new("git") + .args(args) + .current_dir(cwd) + .output() + .with_context(|| format!("run git {}", args.join(" ")))?; + ensure!( + output.status.success(), + "git {} failed: {}", + args.join(" "), + String::from_utf8_lossy(&output.stderr) + ); + Ok(String::from_utf8(output.stdout)?.trim().to_string()) +} + +fn create_git_repo() -> Result<(TempDir, String)> { + let repo = TempDir::new()?; + run_git(repo.path(), &["init", "-q", "-b", "main"])?; + std::fs::write(repo.path().join("README.md"), "initial\n")?; + run_git(repo.path(), &["add", "README.md"])?; + run_git( + repo.path(), + &[ + "-c", + "user.name=CXA-5426", + "-c", + "user.email=cxa5426@example.invalid", + "commit", + "-q", + "-m", + "initial", + ], + )?; + run_git(repo.path(), &["remote", "add", "origin", ORIGIN_URL])?; + let head = run_git(repo.path(), &["rev-parse", "HEAD"])?; + Ok((repo, head)) +} + +fn turn_metadata(request: &Value) -> Result { + serde_json::from_str( + request["client_metadata"]["x-codex-turn-metadata"] + .as_str() + .context("turn metadata")?, + ) + .context("valid turn metadata") +} + +fn expected_workspace(repo: &Path, head: &str, has_changes: bool) -> Value { + json!({ + repo.to_string_lossy().as_ref(): { + "associated_remote_urls": {"origin": ORIGIN_URL}, + "latest_git_commit_hash": head, + "has_changes": has_changes, + } + }) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn startup_prewarm_skips_git_enrichment_and_user_turn_observes_fresh_state() -> Result<()> { + skip_if_no_network!(Ok(())); + + let (repo, head) = create_git_repo()?; + let server = start_websocket_server(vec![vec![ + vec![ev_response_created("warm-1"), ev_completed("warm-1")], + vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "done"), + ev_completed("resp-1"), + ], + ]]) + .await; + let cwd = repo.path().to_path_buf(); + let mut builder = test_codex().with_config(move |config| { + config.cwd = cwd.abs(); + }); + let test = builder.build_with_websocket_server(&server).await?; + + let prewarm = tokio::time::timeout( + Duration::from_secs(5), + server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0), + ) + .await? + .body_json(); + assert!(turn_metadata(&prewarm)?.get("workspaces").is_none()); + + std::fs::write(repo.path().join("untracked.txt"), "dirty\n")?; + test.submit_turn("inspect the workspace").await?; + let turn = server + .single_connection() + .get(1) + .context("turn request")? + .body_json(); + assert_eq!( + turn_metadata(&turn)?["workspaces"], + expected_workspace(repo.path(), &head, /*has_changes*/ true) + ); + + test.codex.shutdown_and_wait().await?; + server.shutdown().await; + Ok(()) +} + +#[cfg(not(target_os = "windows"))] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_prewarm_and_review_skip_redundant_git_enrichment() -> Result<()> { + skip_if_no_network!(Ok(())); + + let (repo, head) = create_git_repo()?; + let tool_args = json!({ + "cmd": "true", + "sandbox_permissions": SandboxPermissions::RequireEscalated, + "justification": "Exercise Guardian approval routing.", + }) + .to_string(); + let server = start_websocket_server(vec![ + vec![vec![ev_response_created("warm-1"), ev_completed("warm-1")]], + vec![vec![ev_response_created("warm-2"), ev_completed("warm-2")]], + vec![vec![ + ev_response_created("approval-request"), + ev_function_call("approval-call", "exec_command", &tool_args), + ev_completed("approval-request"), + ]], + vec![vec![ + ev_response_created("guardian-review"), + ev_completed("guardian-review"), + ]], + ]) + .await; + let cwd = repo.path().to_path_buf(); + let mut builder = test_codex().with_config(move |config| { + config.cwd = cwd.abs(); + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + }); + let test = builder.build_with_websocket_server(&server).await?; + + let (first, second) = tokio::time::timeout(Duration::from_secs(5), async { + tokio::join!( + server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0), + server.wait_for_request(/*connection_index*/ 1, /*request_index*/ 0) + ) + }) + .await?; + for prewarm in [first.body_json(), second.body_json()] { + assert_eq!(prewarm["generate"].as_bool(), Some(false)); + assert!(turn_metadata(&prewarm)?.get("workspaces").is_none()); + } + + std::fs::write(repo.path().join("untracked.txt"), "dirty\n")?; + test.codex + .submit( + vec![UserInput::Text { + text: "run a command that requires Guardian review".into(), + text_elements: Vec::new(), + }] + .into(), + ) + .await?; + let (user_turn, guardian_turn) = tokio::time::timeout(Duration::from_secs(5), async { + tokio::join!( + server.wait_for_request(/*connection_index*/ 2, /*request_index*/ 0), + server.wait_for_request(/*connection_index*/ 3, /*request_index*/ 0) + ) + }) + .await?; + let user_turn = user_turn.body_json(); + let guardian_turn = guardian_turn.body_json(); + assert_eq!( + turn_metadata(&user_turn)?["workspaces"], + expected_workspace(repo.path(), &head, /*has_changes*/ true) + ); + assert_eq!( + guardian_turn["client_metadata"]["x-openai-subagent"].as_str(), + Some("guardian") + ); + assert!(turn_metadata(&guardian_turn)?.get("workspaces").is_none()); + + test.codex.shutdown_and_wait().await?; + server.shutdown().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn ephemeral_system_thread_prewarm_skips_and_turn_observes_fresh_state() -> Result<()> { + skip_if_no_network!(Ok(())); + + let (repo, head) = create_git_repo()?; + let server = start_websocket_server(vec![ + vec![vec![ev_response_created("warm-1"), ev_completed("warm-1")]], + vec![ + vec![ev_response_created("warm-2"), ev_completed("warm-2")], + vec![ + ev_response_created("resp-1"), + ev_function_call( + "wait-for-git", + "test_sync_tool", + r#"{"sleep_after_ms":1000}"#, + ), + ev_completed("resp-1"), + ], + vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-2", "done"), + ev_completed("resp-2"), + ], + ], + ]) + .await; + let cwd = repo.path().to_path_buf(); + let mut builder = test_codex() + .with_model("test-gpt-5.1-codex") + .with_config(move |config| { + config.cwd = cwd.abs(); + }); + let test = builder.build_with_websocket_server(&server).await?; + tokio::time::timeout( + Duration::from_secs(5), + server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0), + ) + .await?; + + let mut config = test.config.clone(); + config.ephemeral = true; + let environments = test + .thread_manager + .default_environment_selections(&config.cwd, &config.workspace_roots); + let system_thread = test + .thread_manager + .start_thread_with_options(StartThreadOptions { + config, + allow_provider_model_fallback: false, + initial_history: InitialHistory::New, + history_mode: None, + session_source: None, + thread_source: Some(ThreadSource::Feature("system".to_string())), + dynamic_tools: Vec::new(), + metrics_service_name: None, + parent_trace: None, + environments, + thread_extension_init: Default::default(), + supports_openai_form_elicitation: false, + }) + .await?; + let prewarm = tokio::time::timeout( + Duration::from_secs(5), + server.wait_for_request(/*connection_index*/ 1, /*request_index*/ 0), + ) + .await? + .body_json(); + assert!(turn_metadata(&prewarm)?.get("workspaces").is_none()); + + std::fs::write(repo.path().join("untracked.txt"), "dirty\n")?; + system_thread + .thread + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "generate a thread title".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + wait_for_event(system_thread.thread.as_ref(), |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let turn = server + .connections() + .get(1) + .and_then(|connection| connection.get(2)) + .context("system turn follow-up request")? + .body_json(); + assert_eq!( + turn_metadata(&turn)?["workspaces"], + expected_workspace(repo.path(), &head, /*has_changes*/ true) + ); + + system_thread.thread.shutdown_and_wait().await?; + test.codex.shutdown_and_wait().await?; + server.shutdown().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn concurrent_turns_keep_distinct_worktree_and_repository_metadata() -> Result<()> { + skip_if_no_network!(Ok(())); + + let (repo, repo_head) = create_git_repo()?; + let worktree_parent = TempDir::new()?; + let worktree = worktree_parent.path().join("linked-worktree"); + run_git( + repo.path(), + &[ + "worktree", + "add", + "-q", + "-b", + "feature/cxa5426", + worktree.to_str().context("worktree path")?, + ], + )?; + std::fs::write(worktree.join("branch.txt"), "branch commit\n")?; + run_git(&worktree, &["add", "branch.txt"])?; + run_git( + &worktree, + &[ + "-c", + "user.name=CXA-5426", + "-c", + "user.email=cxa5426@example.invalid", + "commit", + "-q", + "-m", + "branch commit", + ], + )?; + let worktree_head = run_git(&worktree, &["rev-parse", "HEAD"])?; + let (other_repo, other_head) = create_git_repo()?; + + let server = start_websocket_server(vec![ + vec![vec![ev_response_created("warm-1"), ev_completed("warm-1")]], + vec![vec![ev_response_created("warm-2"), ev_completed("warm-2")]], + vec![vec![ev_response_created("warm-3"), ev_completed("warm-3")]], + vec![vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "done"), + ev_completed("resp-1"), + ]], + vec![vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-2", "done"), + ev_completed("resp-2"), + ]], + vec![vec![ + ev_response_created("resp-3"), + ev_assistant_message("msg-3", "done"), + ev_completed("resp-3"), + ]], + ]) + .await; + + let repo_cwd = repo.path().to_path_buf(); + let mut builder = test_codex().with_config(move |config| { + config.cwd = repo_cwd.abs(); + }); + let test = builder.build_with_websocket_server(&server).await?; + + let mut worktree_config = test.config.clone(); + worktree_config.cwd = worktree.abs(); + let worktree_environments = test + .thread_manager + .default_environment_selections(&worktree_config.cwd, &worktree_config.workspace_roots); + let worktree_thread = test + .thread_manager + .start_thread_with_options(StartThreadOptions { + config: worktree_config, + allow_provider_model_fallback: false, + initial_history: InitialHistory::New, + history_mode: None, + session_source: None, + thread_source: Some(ThreadSource::User), + dynamic_tools: Vec::new(), + metrics_service_name: None, + parent_trace: None, + environments: worktree_environments, + thread_extension_init: Default::default(), + supports_openai_form_elicitation: false, + }) + .await?; + + let mut other_config = test.config.clone(); + other_config.cwd = other_repo.path().to_path_buf().abs(); + let other_environments = test + .thread_manager + .default_environment_selections(&other_config.cwd, &other_config.workspace_roots); + let other_thread = test + .thread_manager + .start_thread_with_options(StartThreadOptions { + config: other_config, + allow_provider_model_fallback: false, + initial_history: InitialHistory::New, + history_mode: None, + session_source: None, + thread_source: Some(ThreadSource::User), + dynamic_tools: Vec::new(), + metrics_service_name: None, + parent_trace: None, + environments: other_environments, + thread_extension_init: Default::default(), + supports_openai_form_elicitation: false, + }) + .await?; + + tokio::time::timeout(Duration::from_secs(5), async { + tokio::join!( + server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0), + server.wait_for_request(/*connection_index*/ 1, /*request_index*/ 0), + server.wait_for_request(/*connection_index*/ 2, /*request_index*/ 0) + ) + }) + .await?; + + std::fs::write(repo.path().join("untracked.txt"), "dirty\n")?; + std::fs::write(other_repo.path().join("untracked.txt"), "dirty\n")?; + let user_turn = |prompt: &str| Op::UserInput { + items: vec![UserInput::Text { + text: prompt.to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }; + tokio::try_join!( + test.codex.submit(user_turn("inspect the main worktree")), + worktree_thread + .thread + .submit(user_turn("inspect the linked worktree")), + other_thread + .thread + .submit(user_turn("inspect the other repository")) + )?; + tokio::join!( + wait_for_event(test.codex.as_ref(), |event| matches!( + event, + EventMsg::TurnComplete(_) + )), + wait_for_event(worktree_thread.thread.as_ref(), |event| matches!( + event, + EventMsg::TurnComplete(_) + )), + wait_for_event(other_thread.thread.as_ref(), |event| matches!( + event, + EventMsg::TurnComplete(_) + )) + ); + + let mut actual_workspaces = server + .connections() + .into_iter() + .skip(3) + .map(|connection| { + let request = connection.first().context("turn request")?.body_json(); + Ok(turn_metadata(&request)?["workspaces"].clone()) + }) + .collect::>>()?; + let mut expected_workspaces = vec![ + expected_workspace(repo.path(), &repo_head, /*has_changes*/ true), + expected_workspace(&worktree, &worktree_head, /*has_changes*/ false), + expected_workspace(other_repo.path(), &other_head, /*has_changes*/ true), + ]; + actual_workspaces.sort_by_key(Value::to_string); + expected_workspaces.sort_by_key(Value::to_string); + assert_eq!(actual_workspaces, expected_workspaces); + + worktree_thread.thread.shutdown_and_wait().await?; + other_thread.thread.shutdown_and_wait().await?; + test.codex.shutdown_and_wait().await?; + server.shutdown().await; + Ok(()) +} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index aeb570c032..ca09b01ce9 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -64,6 +64,7 @@ mod exec_policy; mod extension_sandbox; mod external_auth; mod fork_thread; +mod git_enrichment; #[cfg(not(target_os = "windows"))] mod guardian_review; #[cfg(not(target_os = "windows"))]