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
This commit is contained in:
acrognale-oai authored and copyberry committed 2026-07-22 10:47:51 +00:00
1 parent 9fc715c086
commit 33d394c69e
12 files changed
+691 -21

No files matched your search

+4
View File
@@ -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::is_mcp_tool_approval_question_id;
use crate::mcp_tool_call::lookup_mcp_tool_metadata; use crate::mcp_tool_call::lookup_mcp_tool_metadata;
use crate::mcp_tool_call::mcp_approvals_reviewer; use crate::mcp_tool_call::mcp_approvals_reviewer;
use crate::session::GitEnrichmentPolicy;
use crate::session::SUBMISSION_CHANNEL_CAPACITY; use crate::session::SUBMISSION_CHANNEL_CAPACITY;
use crate::session::SessionIo; use crate::session::SessionIo;
use crate::session::SessionSpawnArgs; use crate::session::SessionSpawnArgs;
@@ -82,6 +83,7 @@ pub(crate) async fn run_codex_thread_interactive(
cancel_token: CancellationToken, cancel_token: CancellationToken,
subagent_source: SubAgentSource, subagent_source: SubAgentSource,
initial_history: Option<InitialHistory>, initial_history: Option<InitialHistory>,
git_enrichment_policy: GitEnrichmentPolicy,
) -> Result<(Arc<Session>, SessionIo), CodexErr> { ) -> Result<(Arc<Session>, SessionIo), CodexErr> {
let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
let (tx_ops, rx_ops) = 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(), attestation_provider: parent_session.services.attestation_provider.clone(),
external_time_provider: Some(Arc::clone(&parent_session.services.time_provider)), external_time_provider: Some(Arc::clone(&parent_session.services.time_provider)),
inherited_multi_agent_version: Some(MultiAgentVersion::Disabled), inherited_multi_agent_version: Some(MultiAgentVersion::Disabled),
git_enrichment_policy,
})) }))
.or_cancel(&cancel_token) .or_cancel(&cancel_token)
.await??; .await??;
@@ -217,6 +220,7 @@ pub(crate) async fn run_codex_thread_one_shot(
child_cancel.clone(), child_cancel.clone(),
subagent_source, subagent_source,
initial_history, initial_history,
GitEnrichmentPolicy::Fresh,
)) ))
.await?; .await?;
@@ -210,6 +210,7 @@ async fn run_codex_thread_interactive_respects_pre_cancelled_spawn() {
cancel_token, cancel_token,
SubAgentSource::Review, SubAgentSource::Review,
/*initial_history*/ None, /*initial_history*/ None,
crate::session::GitEnrichmentPolicy::Fresh,
), ),
) )
.await .await
@@ -43,6 +43,7 @@ use crate::config::NetworkProxySpec;
use crate::config::Permissions; use crate::config::Permissions;
use crate::context::ContextualUserFragment; use crate::context::ContextualUserFragment;
use crate::context::GuardianFollowupReviewReminder; use crate::context::GuardianFollowupReviewReminder;
use crate::session::GitEnrichmentPolicy;
use crate::session::SessionIo; use crate::session::SessionIo;
use crate::session::session::Session; use crate::session::session::Session;
use crate::session::turn_context::TurnContext; use crate::session::turn_context::TurnContext;
@@ -678,6 +679,7 @@ async fn spawn_guardian_review_session(
cancel_token.clone(), cancel_token.clone(),
SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()), SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()),
initial_history, initial_history,
GitEnrichmentPolicy::Skip,
)) ))
.await?; .await?;
+9
View File
@@ -405,6 +405,12 @@ pub(crate) struct SessionIo {
pub(crate) type SessionLoopTermination = Shared<BoxFuture<'static, ()>>; pub(crate) type SessionLoopTermination = Shared<BoxFuture<'static, ()>>;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum GitEnrichmentPolicy {
Fresh,
Skip,
}
pub(crate) struct SessionSpawnArgs { pub(crate) struct SessionSpawnArgs {
pub(crate) config: Config, pub(crate) config: Config,
pub(crate) allow_provider_model_fallback: bool, pub(crate) allow_provider_model_fallback: bool,
@@ -445,6 +451,7 @@ pub(crate) struct SessionSpawnArgs {
pub(crate) attestation_provider: Option<Arc<dyn AttestationProvider>>, pub(crate) attestation_provider: Option<Arc<dyn AttestationProvider>>,
pub(crate) external_time_provider: Option<Arc<dyn TimeProvider>>, pub(crate) external_time_provider: Option<Arc<dyn TimeProvider>>,
pub(crate) inherited_multi_agent_version: Option<MultiAgentVersion>, pub(crate) inherited_multi_agent_version: Option<MultiAgentVersion>,
pub(crate) git_enrichment_policy: GitEnrichmentPolicy,
} }
pub(crate) fn resolve_multi_agent_version( pub(crate) fn resolve_multi_agent_version(
@@ -533,6 +540,7 @@ impl Session {
attestation_provider, attestation_provider,
external_time_provider, external_time_provider,
inherited_multi_agent_version, inherited_multi_agent_version,
git_enrichment_policy,
} = args; } = args;
let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
let (tx_event, rx_event) = async_channel::unbounded(); let (tx_event, rx_event) = async_channel::unbounded();
@@ -720,6 +728,7 @@ impl Session {
attestation_provider, attestation_provider,
external_time_provider, external_time_provider,
multi_agent_version, multi_agent_version,
git_enrichment_policy,
)) ))
.await .await
.map_err(|e| { .map_err(|e| {
+3
View File
@@ -44,6 +44,7 @@ pub(crate) struct Session {
pub(crate) input_queue: InputQueue, pub(crate) input_queue: InputQueue,
pub(crate) guardian_review_session: GuardianReviewSessionManager, pub(crate) guardian_review_session: GuardianReviewSessionManager,
pub(crate) services: SessionServices, pub(crate) services: SessionServices,
pub(super) git_enrichment_policy: GitEnrichmentPolicy,
pub(super) next_internal_sub_id: AtomicU64, pub(super) next_internal_sub_id: AtomicU64,
} }
@@ -501,6 +502,7 @@ impl Session {
attestation_provider: Option<Arc<dyn AttestationProvider>>, attestation_provider: Option<Arc<dyn AttestationProvider>>,
external_time_provider: Option<Arc<dyn TimeProvider>>, external_time_provider: Option<Arc<dyn TimeProvider>>,
multi_agent_version: Option<MultiAgentVersion>, multi_agent_version: Option<MultiAgentVersion>,
git_enrichment_policy: GitEnrichmentPolicy,
) -> anyhow::Result<Arc<Self>> { ) -> anyhow::Result<Arc<Self>> {
debug!( debug!(
"Configuring session: model={}; provider={:?}", "Configuring session: model={}; provider={:?}",
@@ -1148,6 +1150,7 @@ impl Session {
input_queue: InputQueue::new(), input_queue: InputQueue::new(),
guardian_review_session: GuardianReviewSessionManager::default(), guardian_review_session: GuardianReviewSessionManager::default(),
services, services,
git_enrichment_policy,
next_internal_sub_id: AtomicU64::new(0), next_internal_sub_id: AtomicU64::new(0),
}); });
if let Some(network_policy_decider_session) = network_policy_decider_session { if let Some(network_policy_decider_session) = network_policy_decider_session {
+5
View File
@@ -5270,6 +5270,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() {
/*attestation_provider*/ None, /*attestation_provider*/ None,
/*external_time_provider*/ None, /*external_time_provider*/ None,
Some(config.multi_agent_version_from_features()), Some(config.multi_agent_version_from_features()),
GitEnrichmentPolicy::Fresh,
) )
.await; .await;
@@ -5535,6 +5536,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
input_queue: super::input_queue::InputQueue::new(), input_queue: super::input_queue::InputQueue::new(),
guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(),
services, services,
git_enrichment_policy: GitEnrichmentPolicy::Fresh,
next_internal_sub_id: AtomicU64::new(0), next_internal_sub_id: AtomicU64::new(0),
}; };
@@ -5657,6 +5659,7 @@ async fn make_session_with_config_and_rx(
/*attestation_provider*/ None, /*attestation_provider*/ None,
/*external_time_provider*/ None, /*external_time_provider*/ None,
Some(config.multi_agent_version_from_features()), Some(config.multi_agent_version_from_features()),
GitEnrichmentPolicy::Fresh,
) )
.await?; .await?;
@@ -5771,6 +5774,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx(
/*attestation_provider*/ None, /*attestation_provider*/ None,
/*external_time_provider*/ None, /*external_time_provider*/ None,
Some(config.multi_agent_version_from_features()), Some(config.multi_agent_version_from_features()),
GitEnrichmentPolicy::Fresh,
) )
.await?; .await?;
@@ -7694,6 +7698,7 @@ where
input_queue: super::input_queue::InputQueue::new(), input_queue: super::input_queue::InputQueue::new(),
guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(),
services, services,
git_enrichment_policy: GitEnrichmentPolicy::Fresh,
next_internal_sub_id: AtomicU64::new(0), next_internal_sub_id: AtomicU64::new(0),
}); });
@@ -765,6 +765,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
attestation_provider: None, attestation_provider: None,
external_time_provider: None, external_time_provider: None,
inherited_multi_agent_version: None, inherited_multi_agent_version: None,
git_enrichment_policy: GitEnrichmentPolicy::Skip,
}) })
.await .await
.expect("spawn guardian subagent"); .expect("spawn guardian subagent");
+8 -4
View File
@@ -673,6 +673,7 @@ impl Session {
session_configuration, session_configuration,
final_output_json_schema, final_output_json_schema,
TurnMultiAgentRuntime::ResolveAndStore, TurnMultiAgentRuntime::ResolveAndStore,
self.git_enrichment_policy,
) )
.await .await
} }
@@ -687,6 +688,7 @@ impl Session {
session_configuration, session_configuration,
/*final_output_json_schema*/ None, /*final_output_json_schema*/ None,
TurnMultiAgentRuntime::Preview, TurnMultiAgentRuntime::Preview,
GitEnrichmentPolicy::Skip,
) )
.await .await
} }
@@ -698,6 +700,7 @@ impl Session {
session_configuration: SessionConfiguration, session_configuration: SessionConfiguration,
final_output_json_schema: Option<Option<Value>>, final_output_json_schema: Option<Option<Value>>,
multi_agent_runtime: TurnMultiAgentRuntime, multi_agent_runtime: TurnMultiAgentRuntime,
git_enrichment_policy: GitEnrichmentPolicy,
) -> Arc<TurnContext> { ) -> Arc<TurnContext> {
let turn_environments = self.services.turn_environments.snapshot().await; let turn_environments = self.services.turn_environments.snapshot().await;
let primary_turn_environment = turn_environments.primary(); let primary_turn_environment = turn_environments.primary();
@@ -792,10 +795,11 @@ impl Session {
turn_context.final_output_json_schema = final_schema; turn_context.final_output_json_schema = final_schema;
} }
let turn_context = Arc::new(turn_context); let turn_context = Arc::new(turn_context);
if turn_context if git_enrichment_policy == GitEnrichmentPolicy::Fresh
.environments && turn_context
.single_local_environment_cwd() .environments
.is_some() .single_local_environment_cwd()
.is_some()
{ {
turn_context.turn_metadata_state.spawn_git_enrichment_task(); turn_context.turn_metadata_state.spawn_git_enrichment_task();
} }
+2
View File
@@ -10,6 +10,7 @@ use crate::environment_selection::TurnEnvironmentSnapshot;
use crate::environment_selection::default_thread_environment_selections; use crate::environment_selection::default_thread_environment_selections;
use crate::mcp::McpManager; use crate::mcp::McpManager;
use crate::rollout::truncation; use crate::rollout::truncation;
use crate::session::GitEnrichmentPolicy;
use crate::session::INITIAL_SUBMIT_ID; use crate::session::INITIAL_SUBMIT_ID;
use crate::session::SessionIo; use crate::session::SessionIo;
use crate::session::SessionSpawnArgs; use crate::session::SessionSpawnArgs;
@@ -1685,6 +1686,7 @@ impl ThreadManagerState {
attestation_provider: self.attestation_provider.clone(), attestation_provider: self.attestation_provider.clone(),
external_time_provider: self.external_time_provider.clone(), external_time_provider: self.external_time_provider.clone(),
inherited_multi_agent_version: multi_agent_version, inherited_multi_agent_version: multi_agent_version,
git_enrichment_policy: GitEnrichmentPolicy::Fresh,
})) }))
.await?; .await?;
let new_thread = self let new_thread = self
+154 -17
View File
@@ -20,6 +20,7 @@ use core_test_support::PathExt;
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
use serde_json::Value; use serde_json::Value;
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tempfile::TempDir; use tempfile::TempDir;
use tokio::process::Command; use tokio::process::Command;
@@ -109,6 +110,25 @@ async fn create_clean_git_repo(repo_name: &str) -> (TempDir, AbsolutePathBuf) {
(temp_dir, repo_path) (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] #[tokio::test]
async fn detached_memory_responses_metadata_omits_turn_identity() { async fn detached_memory_responses_metadata_omits_turn_identity() {
let (_temp_dir, repo_path) = create_clean_git_repo("repo-東京").await; 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(); state.spawn_git_enrichment_task();
let json = wait_for_git_enrichment(&state).await;
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");
assert_eq!( assert_eq!(
json["forked_from_thread_id"].as_str(), 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")); 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::<Vec<_>>();
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());
}
+501
View File
@@ -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<String> {
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<Value> {
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::<Result<Vec<_>>>()?;
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(())
}
+1
View File
@@ -64,6 +64,7 @@ mod exec_policy;
mod extension_sandbox; mod extension_sandbox;
mod external_auth; mod external_auth;
mod fork_thread; mod fork_thread;
mod git_enrichment;
#[cfg(not(target_os = "windows"))] #[cfg(not(target_os = "windows"))]
mod guardian_review; mod guardian_review;
#[cfg(not(target_os = "windows"))] #[cfg(not(target_os = "windows"))]