diff --git a/codex-rs/app-server/tests/suite/v2/guardian_sync_session_tests.rs b/codex-rs/app-server/tests/suite/v2/guardian_sync_session_tests.rs index 4012493637..48fc7b52fb 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_sync_session_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_sync_session_tests.rs @@ -57,6 +57,84 @@ enum ReviewEnding { Interrupt, } +#[tokio::test] +async fn managed_reviewer_refreshes_global_instructions_before_reuse() -> Result<()> { + skip_if_no_network!(Ok(())); + let checks = ["initial", "unchanged", "updated", "still-updated"]; + let mut responses = Vec::new(); + for check in checks { + for body in [ + tool_response(TEST_SERVER_NAME, TEST_TOOL_NAME, &[check]), + assistant_response(ALLOW)?, + assistant_response("Done.")?, + ] { + responses.push(vec![StreamingSseChunk { gate: None, body }]); + } + } + let (server, _completions) = start_streaming_sse_server(responses).await; + let (mcp_url, mcp_server) = start_mcp_server(/*sensitive_action*/ None).await?; + let codex_home = TempDir::new()?; + let instructions_path = codex_home.path().join("AGENTS.md"); + sync_config(server.uri()) + .with_extra_config(&format!( + "[mcp_servers.{TEST_SERVER_NAME}]\nurl = \"{mcp_url}/mcp\"\ndefault_tools_approval_mode = \"prompt\"" + )) + .write(codex_home.path())?; + let mut app = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized_with_timeout(TIMEOUT) + .await?; + let old_instructions = "Keep this project private."; + let new_instructions = "Ask before sharing this project's source code."; + std::fs::write(&instructions_path, old_instructions)?; + let parent = app.start_thread(ThreadStartParams::default()).await?.thread; + let mut reviewer_ids = Vec::new(); + for (index, instructions) in [ + old_instructions, + old_instructions, + new_instructions, + new_instructions, + ] + .into_iter() + .enumerate() + { + std::fs::write(&instructions_path, instructions)?; + let request_id = app + .send_turn_start_request(TurnStartParams { + thread_id: parent.id.clone(), + input: vec![UserInput::Text { + text: format!("Run the {} check.", checks[index]), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let _: TurnStartResponse = timeout(TIMEOUT, app.read_response(request_id)).await??; + let completed: TurnCompletedNotification = + timeout(TIMEOUT, app.read_notification("turn/completed")).await??; + assert_eq!(completed.thread_id, parent.id); + assert_eq!(completed.turn.status, TurnStatus::Completed); + + let requests = server.requests().await; + assert_eq!(requests.len(), (index + 1) * 3); + let review: Value = serde_json::from_slice(&requests[index * 3 + 1])?; + assert!(review["input"].to_string().contains(instructions)); + reviewer_ids.push( + review["client_metadata"]["thread_id"] + .as_str() + .expect("reviewer ID") + .to_owned(), + ); + } + assert_eq!(reviewer_ids[0], reviewer_ids[1]); + assert_ne!(reviewer_ids[1], reviewer_ids[2]); + assert_eq!(reviewer_ids[2], reviewer_ids[3]); + + app.shutdown_gracefully().await?; + mcp_server.abort(); + Ok(()) +} + #[test_case(ReviewEnding::Complete; "completed reviews")] #[test_case(ReviewEnding::Interrupt; "cancelled concurrent reviews")] #[tokio::test] diff --git a/codex-rs/codex-home/src/instructions/mod.rs b/codex-rs/codex-home/src/instructions/mod.rs index 16f8a062b5..a1952b26a3 100644 --- a/codex-rs/codex-home/src/instructions/mod.rs +++ b/codex-rs/codex-home/src/instructions/mod.rs @@ -1,4 +1,7 @@ +use std::collections::HashSet; use std::io; +use std::sync::Arc; +use std::sync::Mutex; use codex_extension_api::Instructions; use codex_extension_api::LoadUserInstructionsFuture; @@ -13,12 +16,25 @@ const LOCAL_AGENTS_MD_FILENAME: &str = "AGENTS.override.md"; #[derive(Clone, Debug)] pub struct CodexHomeUserInstructionsProvider { codex_home: AbsolutePathBuf, + // Cached instructions and warning history are shared across provider clones. + state: Arc>, +} + +#[derive(Debug, Default)] +struct InstructionsState { + // Failed refreshes retain the last good value; confirmed absence still clears it. + last_successful_instructions: Option, + // Report ongoing failures once, then again if they return after recovery. + active_warnings: HashSet, } impl CodexHomeUserInstructionsProvider { /// Creates a provider rooted at the supplied absolute Codex home directory. pub fn new(codex_home: AbsolutePathBuf) -> Self { - Self { codex_home } + Self { + codex_home, + state: Arc::new(Mutex::new(InstructionsState::default())), + } } async fn load_from_codex_home(&self) -> LoadedUserInstructions { @@ -69,7 +85,26 @@ impl CodexHomeUserInstructionsProvider { impl UserInstructionsProvider for CodexHomeUserInstructionsProvider { fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_> { - Box::pin(self.load_from_codex_home()) + Box::pin(async move { + let mut loaded = self.load_from_codex_home().await; + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if loaded.instructions.is_some() || loaded.warnings.is_empty() { + state + .last_successful_instructions + .clone_from(&loaded.instructions); + } else { + loaded.instructions = state.last_successful_instructions.clone(); + } + let previous_warnings = std::mem::take(&mut state.active_warnings); + loaded.warnings.retain(|warning| { + state.active_warnings.insert(warning.clone()) + && !previous_warnings.contains(warning) + }); + loaded + }) } } diff --git a/codex-rs/codex-home/src/instructions/tests.rs b/codex-rs/codex-home/src/instructions/tests.rs index 478681b528..bebe897851 100644 --- a/codex-rs/codex-home/src/instructions/tests.rs +++ b/codex-rs/codex-home/src/instructions/tests.rs @@ -110,6 +110,7 @@ async fn directory_override_falls_back_to_default() { #[tokio::test] async fn recoverable_override_read_error_warns_and_falls_back_to_default() { let home = TempDir::new().expect("temp dir"); + let provider = provider(&home); let override_path = home.path().join(LOCAL_AGENTS_MD_FILENAME); create_symlink_loop(&override_path); fs::write(home.path().join(DEFAULT_AGENTS_MD_FILENAME), "default").expect("write default"); @@ -120,11 +121,70 @@ async fn recoverable_override_read_error_warns_and_falls_back_to_default() { ); assert_eq!( - provider(&home).load_user_instructions().await, + provider.load_user_instructions().await, + expected( + &home, + DEFAULT_AGENTS_MD_FILENAME, + "default", + vec![warning.clone()] + ) + ); + assert_eq!( + provider.clone().load_user_instructions().await, + expected(&home, DEFAULT_AGENTS_MD_FILENAME, "default", Vec::new()) + ); + + fs::remove_file(&override_path).expect("remove broken symlink"); + assert_eq!( + provider.load_user_instructions().await, + expected(&home, DEFAULT_AGENTS_MD_FILENAME, "default", Vec::new()) + ); + create_symlink_loop(&override_path); + assert_eq!( + provider.load_user_instructions().await, expected(&home, DEFAULT_AGENTS_MD_FILENAME, "default", vec![warning]) ); } +#[tokio::test] +async fn failed_refresh_retains_last_success_until_recovery_or_removal() { + let home = TempDir::new().expect("temp dir"); + let provider = provider(&home); + let path = home.path().join(DEFAULT_AGENTS_MD_FILENAME); + fs::write(&path, "initial").expect("write initial instructions"); + let initial = provider.load_user_instructions().await; + + fs::remove_file(&path).expect("remove initial file"); + create_symlink_loop(&path); + let failed = provider.load_user_instructions().await; + assert_eq!(failed.instructions, initial.instructions); + assert_eq!(failed.warnings.len(), 1); + assert_eq!(provider.load_user_instructions().await, initial); + + fs::remove_file(&path).expect("remove broken symlink"); + fs::write(&path, "recovered").expect("restore instructions"); + assert_eq!( + provider.load_user_instructions().await, + expected(&home, DEFAULT_AGENTS_MD_FILENAME, "recovered", Vec::new()), + ); + + fs::write(&path, " \n").expect("clear instructions"); + assert_eq!( + provider.load_user_instructions().await, + LoadedUserInstructions::default(), + ); + fs::remove_file(&path).expect("remove cleared instructions"); + create_symlink_loop(&path); + let failed_after_clear = provider.load_user_instructions().await; + assert_eq!(failed_after_clear.instructions, None); + assert_eq!(failed_after_clear.warnings.len(), 1); + fs::remove_file(&path).expect("remove broken symlink"); + assert_eq!( + provider.load_user_instructions().await, + LoadedUserInstructions::default(), + ); +} + #[tokio::test] async fn invalid_utf8_is_lossy() { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/core/src/agents_md.rs b/codex-rs/core/src/agents_md.rs index 431834e1da..37678f3e34 100644 --- a/codex-rs/core/src/agents_md.rs +++ b/codex-rs/core/src/agents_md.rs @@ -316,6 +316,15 @@ impl LoadedAgentsMd { } } + pub(crate) fn with_user_instructions( + mut self, + user_instructions: Option, + ) -> Option { + self.user_instructions = + user_instructions.filter(|instructions| !instructions.text.trim().is_empty()); + (!self.is_empty()).then_some(self) + } + /// Creates source-less user instructions for tests. /// /// This cannot be gated with `#[cfg(test)]` because integration tests diff --git a/codex-rs/core/src/agents_md_manager.rs b/codex-rs/core/src/agents_md_manager.rs index 4ac0005cee..3bb500192b 100644 --- a/codex-rs/core/src/agents_md_manager.rs +++ b/codex-rs/core/src/agents_md_manager.rs @@ -1,18 +1,36 @@ +//! Resolves global and repository instructions into one validated snapshot. +//! Refreshes are serialized; the state lock is never held while calling providers. + use crate::agents_md::LoadedAgentsMd; use crate::agents_md::load_project_instructions; use crate::config::Config; use crate::environment_selection::TurnEnvironmentSnapshot; use codex_extension_api::Instructions; +use codex_extension_api::UserInstructionsProvider; use codex_protocol::config_types::TrustLevel; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result as CodexResult; use codex_protocol::protocol::TurnEnvironmentSelection; -use std::io; use std::sync::Arc; use tokio::sync::Mutex; +use tokio::sync::Semaphore; -/// Owns the inputs and cached result of AGENTS.md discovery for a session. +/// Owns instruction sources, refresh serialization, and the applied snapshot. pub(crate) struct AgentsMdManager { - user_instructions: Option, - cache: Mutex, + refresh_lock: Semaphore, + state: Mutex, +} + +/// Root runtimes retain providers; subagents inherit only their applied snapshots. +#[derive(Clone, Default)] +pub(crate) struct SessionInstructions { + pub(crate) user: Option, + pub(crate) user_provider: Option>, +} + +struct AgentsMdState { + instructions: SessionInstructions, + cache: AgentsMdCache, } #[derive(Default)] @@ -23,53 +41,105 @@ struct AgentsMdCache { } impl AgentsMdManager { - pub(crate) fn new(user_instructions: Option) -> Self { + pub(crate) fn new(mut instructions: SessionInstructions) -> Self { + instructions.user = normalize_instructions(instructions.user); Self { - user_instructions: user_instructions - .filter(|instructions| !instructions.text.trim().is_empty()), - cache: Mutex::new(AgentsMdCache::default()), + refresh_lock: Semaphore::new(/*permits*/ 1), + state: Mutex::new(AgentsMdState { + instructions, + cache: AgentsMdCache::default(), + }), } } + /// Resolves and validates one coherent snapshot for startup or a request boundary. + /// Providers own fetching and caching. Warnings survive validation failures; + /// only validated snapshots replace the applied host instructions. #[tracing::instrument(name = "agents_md.refresh", skip_all)] pub(crate) async fn refresh( &self, config: &Config, environments: &TurnEnvironmentSnapshot, - ) -> io::Result<()> { + ) -> (CodexResult>>, Vec) { + // Serialize overlapping captures without blocking reads of the applied snapshot. + let Ok(_refresh_guard) = self.refresh_lock.acquire().await else { + return ( + Err(CodexErr::Fatal( + "instruction refresh semaphore closed".to_string(), + )), + Vec::new(), + ); + }; let selections = environments .turn_environments() .map(|environment| environment.selection.clone()) .collect::>(); let active_project_trust_level = config.active_project.trust_level; - { - let mut cache = self.cache.lock().await; - if cache.selections.as_ref() == Some(&selections) - && cache.active_project_trust_level == active_project_trust_level - { - return Ok(()); + let (mut instructions, cached, refresh_repository) = { + let mut state = self.state.lock().await; + let refresh_repository = state.cache.selections.as_ref() != Some(&selections) + || state.cache.active_project_trust_level != active_project_trust_level; + if refresh_repository { + // Tightened read permissions must not leave inaccessible instructions visible, + // even if discovery fails or the caller cancels the refresh. + state.cache = AgentsMdCache::default(); } - cache.selections = None; - cache.active_project_trust_level = None; - cache.loaded = None; + ( + state.instructions.clone(), + state.cache.loaded.clone(), + refresh_repository, + ) + }; + let mut warnings = Vec::new(); + if let Some(provider) = &instructions.user_provider { + let loaded = provider.load_user_instructions().await; + instructions.user = normalize_instructions(loaded.instructions); + warnings.extend(loaded.warnings); } - let loaded = - load_project_instructions(config, self.user_instructions.clone(), environments) - .await? - .map(Arc::new); - let mut cache = self.cache.lock().await; - cache.selections = Some(selections); - cache.active_project_trust_level = active_project_trust_level; - cache.loaded = loaded; - Ok(()) + let result = async { + if !refresh_repository { + let state = self.state.lock().await; + if state.instructions.user == instructions.user { + return Ok(cached); + } + } + + let loaded = if refresh_repository { + load_project_instructions(config, /*user_instructions*/ None, environments) + .await? + .unwrap_or_default() + } else { + cached.as_deref().cloned().unwrap_or_default() + } + .with_user_instructions(instructions.user.clone()) + .map(Arc::new); + let mut state = self.state.lock().await; + state.instructions = instructions; + state.cache = AgentsMdCache { + selections: Some(selections), + active_project_trust_level, + loaded: loaded.clone(), + }; + Ok(loaded) + } + .await; + (result, warnings) } pub(crate) async fn get_loaded(&self) -> Option> { - self.cache.lock().await.loaded.clone() + self.state.lock().await.cache.loaded.clone() } - pub(crate) fn user_instructions(&self) -> Option { - self.user_instructions.clone() + pub(crate) async fn inherited_instructions(&self) -> SessionInstructions { + let state = self.state.lock().await; + SessionInstructions { + user: state.instructions.user.clone(), + ..Default::default() + } } } + +fn normalize_instructions(instructions: Option) -> Option { + instructions.filter(|instructions| !instructions.text.trim().is_empty()) +} diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 4a3e9c4a5d..f9304384c4 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -3,7 +3,6 @@ use std::sync::Arc; use async_channel::Receiver; use async_channel::Sender; use codex_async_utils::OrCancelExt; -use codex_extension_api::LoadedUserInstructions; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; @@ -74,10 +73,7 @@ pub(crate) async fn run_codex_thread_interactive( let conversation_history = initial_history.unwrap_or(InitialHistory::New); let forked_from_thread_id = conversation_history.forked_from_id(); - let user_instructions = LoadedUserInstructions { - instructions: parent_session.user_instructions().await, - warnings: Vec::new(), - }; + let instructions = parent_session.inherited_instructions().await; let session_source = SessionSource::SubAgent(subagent_source.clone()); let is_guardian_reviewer = crate::guardian::is_basic_session_source(&session_source); let extensions = if isolation == codex_extension_api::SessionIsolation::Isolated { @@ -93,7 +89,7 @@ pub(crate) async fn run_codex_thread_interactive( let (session, io) = Session::spawn(SessionSpawnArgs { config, allow_provider_model_fallback: false, - user_instructions, + instructions, installation_id: parent_session.installation_id.clone(), auth_manager, models_manager, diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 7e0dd0c5e2..22e1ff5965 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -52,6 +52,7 @@ use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; use tracing::warn; +use crate::agents_md_manager::SessionInstructions; use crate::codex_delegate::run_codex_thread_interactive; use crate::config::Config; use crate::config::Constrained; @@ -234,7 +235,7 @@ pub struct GuardianReviewSessionReuseKey { impl GuardianReviewSessionReuseKey { fn from_spawn_config( spawn_config: &Config, - user_instructions: Option, + instructions: SessionInstructions, parent_history_version: u64, context_mode: GuardianContextMode, ) -> Self { @@ -262,7 +263,7 @@ impl GuardianReviewSessionReuseKey { permissions: spawn_config.permissions.clone(), developer_instructions: spawn_config.developer_instructions.clone(), base_instructions: spawn_config.base_instructions.clone(), - user_instructions, + user_instructions: instructions.user, compact_prompt: spawn_config.compact_prompt.clone(), cwd: PathUri::from_abs_path(&spawn_config.cwd), mcp_servers: spawn_config.mcp_servers.clone(), diff --git a/codex-rs/core/src/guardian/review_session_factory.rs b/codex-rs/core/src/guardian/review_session_factory.rs index b70ee4ff8a..c7be2363fd 100644 --- a/codex-rs/core/src/guardian/review_session_factory.rs +++ b/codex-rs/core/src/guardian/review_session_factory.rs @@ -32,7 +32,7 @@ impl PreparedSession { let parent_compaction = context_policy.parent_compaction(history, compaction_model_hash)?; let mut key = GuardianReviewSessionReuseKey::from_spawn_config( &config, - parent.user_instructions().await, + parent.inherited_instructions().await, history.history_version(), parent.guardian_context_mode, ) @@ -134,6 +134,11 @@ impl ReviewerSessionFactory for PreparedSession { .await? } }; + let inherited = session.inherited_instructions().await; + let context = GuardianReviewSessionReuseKey { + user_instructions: inherited.user, + ..context + }; Ok(GuardianReviewSession { session, io, diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index 176faac1ad..d0239d7e74 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -1,7 +1,11 @@ use super::super::prompt::BUNDLED_GUARDIAN_POLICY_TEMPLATE; use super::super::prompt::guardian_policy_prompt_with_config_and_template; use super::*; +use crate::agents_md_manager::AgentsMdManager; use crate::context_manager::ContextManager; +use codex_guardian_reviewer::ReviewerRequest; +use codex_guardian_reviewer::ReviewerSession; +use codex_guardian_reviewer::ReviewerSessionFactory; use codex_history::CodexHarnessMetadata; use codex_history::ResponseItemEnvelope; use codex_protocol::openai_models::AutoReviewMessages; @@ -56,7 +60,7 @@ async fn run_review_preserves_evidence_during_parent_compaction() { let (mut reviewer, tx_event, rx_sub) = test_review_session().await; reviewer.reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( ¶ms.spawn_config, - parent.user_instructions().await, + parent.inherited_instructions().await, params.parent_history.history_version(), parent.guardian_context_mode, ) @@ -138,7 +142,7 @@ async fn test_review_session() -> ( let (_agent_status_tx, agent_status) = tokio::sync::watch::channel(AgentStatus::PendingInit); let reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( session.get_config().await.as_ref(), - session.user_instructions().await, + session.inherited_instructions().await, session.clone_history().await.history_version(), GuardianContextMode::Legacy, ); @@ -275,6 +279,57 @@ async fn spawned_guardian_session_preserves_windows_sandbox_proxy_settings() { manager.shutdown().await; } +#[tokio::test] +async fn spawned_guardian_reuse_key_matches_inherited_instructions() { + let mut params = test_review_params().await; + let latest_global = Some(Instructions { + text: "latest global instructions".to_string(), + source: params.spawn_config.codex_home.join("AGENTS.md"), + }); + let parent = Arc::get_mut(&mut params.parent_session).expect("unshared parent session"); + parent.services.agents_md_manager = Arc::new(AgentsMdManager::new(SessionInstructions { + user: latest_global.clone(), + ..Default::default() + })); + // Reproduce an update between reuse-key capture and reviewer creation. + let stale_key = GuardianReviewSessionReuseKey::from_spawn_config( + ¶ms.spawn_config, + SessionInstructions { + user: Some(Instructions { + text: "previous global instructions".to_string(), + source: params.spawn_config.codex_home.join("AGENTS.md"), + }), + ..Default::default() + }, + /*parent_history_version*/ 0, + parent.guardian_context_mode, + ); + let expected_key = GuardianReviewSessionReuseKey { + user_instructions: latest_global.clone(), + ..stale_key.clone() + }; + let prepared = factory::prepare_review(params) + .await + .expect("prepare review"); + let review = prepared + .factory() + .spawn( + stale_key, + GuardianReviewSessionKind::TrunkNew, + /*snapshot*/ None, + CancellationToken::new(), + ) + .await + .expect("spawn reviewer after instruction update"); + + assert_eq!(review.reuse_key, expected_key); + assert_eq!( + review.session.inherited_instructions().await.user, + latest_global + ); + review.shutdown().await; +} + #[tokio::test] async fn guardian_review_session_config_change_invalidates_cached_session() { let parent_config = crate::config::test_config().await; @@ -290,7 +345,7 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { .expect("cached guardian config"); let cached_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( &cached_spawn_config, - /*user_instructions*/ None, + SessionInstructions::default(), /*parent_history_version*/ 0, GuardianContextMode::Legacy, ); @@ -310,7 +365,7 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { .expect("next guardian config"); let next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( &next_spawn_config, - /*user_instructions*/ None, + SessionInstructions::default(), /*parent_history_version*/ 0, GuardianContextMode::Legacy, ); @@ -324,7 +379,7 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { cached_reuse_key, GuardianReviewSessionReuseKey::from_spawn_config( &cached_spawn_config, - /*user_instructions*/ None, + SessionInstructions::default(), /*parent_history_version*/ 0, GuardianContextMode::Legacy, ) @@ -334,7 +389,7 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { cached_reuse_key, GuardianReviewSessionReuseKey::from_spawn_config( &cached_spawn_config, - /*user_instructions*/ None, + SessionInstructions::default(), /*parent_history_version*/ 1, GuardianContextMode::Legacy, ) @@ -370,13 +425,13 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { assert_ne!( GuardianReviewSessionReuseKey::from_spawn_config( &compaction_enabled_config, - /*user_instructions*/ None, + SessionInstructions::default(), /*parent_history_version*/ 0, GuardianContextMode::Legacy, ), GuardianReviewSessionReuseKey::from_spawn_config( &compaction_enabled_config, - /*user_instructions*/ None, + SessionInstructions::default(), /*parent_history_version*/ 1, GuardianContextMode::Legacy, ) @@ -489,7 +544,7 @@ async fn guardian_review_session_compact_scope_change_invalidates_cached_session .expect("cached guardian config"); let cached_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( &cached_spawn_config, - /*user_instructions*/ None, + SessionInstructions::default(), /*parent_history_version*/ 0, GuardianContextMode::Legacy, ); @@ -509,7 +564,7 @@ async fn guardian_review_session_compact_scope_change_invalidates_cached_session .expect("next guardian config"); let next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( &next_spawn_config, - /*user_instructions*/ None, + SessionInstructions::default(), /*parent_history_version*/ 0, GuardianContextMode::Legacy, ); @@ -909,10 +964,18 @@ async fn run_review_on_reused_session_waits_for_submitted_turn() { #[tokio::test] async fn run_review_removes_trunk_when_event_stream_is_broken() { let (mut review_session, tx_event, rx_sub) = test_review_session().await; - let params = test_review_params().await; + let mut params = test_review_params().await; + let parent = Arc::get_mut(&mut params.parent_session).expect("unshared parent session"); + parent.services.agents_md_manager = Arc::new(AgentsMdManager::new(SessionInstructions { + user: Some(Instructions { + text: "parent global instructions".to_string(), + source: params.spawn_config.codex_home.join("AGENTS.md"), + }), + ..Default::default() + })); review_session.reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( ¶ms.spawn_config, - params.parent_session.user_instructions().await, + params.parent_session.inherited_instructions().await, params .parent_session .clone_history() diff --git a/codex-rs/core/src/guardian/review_session_threads.rs b/codex-rs/core/src/guardian/review_session_threads.rs index e2cc14ca28..dd3d793140 100644 --- a/codex-rs/core/src/guardian/review_session_threads.rs +++ b/codex-rs/core/src/guardian/review_session_threads.rs @@ -76,7 +76,7 @@ impl ManagedReviewerThreads { environments: Some(context.environments().to_selections()), inherited_environments: Some(context.environments().clone()), user_instructions: Some(LoadedUserInstructions { - instructions: parent.user_instructions().await, + instructions: parent.inherited_instructions().await.user, warnings: Vec::new(), }), client_mcp_extensions: parent.services.client_mcp_extensions.clone(), diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 777c7834ef..d78802086b 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -15,6 +15,7 @@ use crate::agent::agent_status_from_event; use crate::agent::status::is_final; use crate::agent_communication::AgentCommunicationContext; use crate::agent_communication::AgentCommunicationKind; +use crate::agents_md_manager::SessionInstructions; use crate::attestation::AttestationProvider; use crate::compact; use crate::compact::CompactedHistoryMetadata; @@ -68,7 +69,6 @@ use codex_exec_server::EnvironmentManager; use codex_execpolicy::prefix_rule_migration; use codex_extension_api::ConversationHistorySnapshot; use codex_extension_api::ExtensionDataInit; -use codex_extension_api::LoadedUserInstructions; use codex_extension_api::PromptSlot; use codex_extension_api::TurnContextContributionInput; use codex_features::FEATURES; @@ -426,7 +426,7 @@ pub(crate) enum ForkPersistence { pub(crate) struct SessionSpawnArgs { pub(crate) config: Config, pub(crate) allow_provider_model_fallback: bool, - pub(crate) user_instructions: LoadedUserInstructions, + pub(crate) instructions: SessionInstructions, pub(crate) installation_id: String, pub(crate) auth_manager: Arc, pub(crate) models_manager: SharedModelsManager, @@ -529,9 +529,9 @@ impl Session { async fn spawn_internal(args: SessionSpawnArgs) -> CodexResult<(Arc, SessionIo)> { let SessionSpawnArgs { - mut config, + config, allow_provider_model_fallback, - user_instructions, + instructions, installation_id, auth_manager, models_manager, @@ -574,14 +574,6 @@ impl Session { let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_event, rx_event) = async_channel::unbounded(); - let LoadedUserInstructions { - instructions: user_instructions, - warnings: user_instruction_provider_warnings, - } = user_instructions; - // TODO(anp) pull startup_warnings out of Config - config - .startup_warnings - .extend(user_instruction_provider_warnings); let isolation = thread_extension_init .get::() .map(|policy| *policy) @@ -822,7 +814,7 @@ impl Session { session_configuration, &environment_selections, config.clone(), - user_instructions, + instructions, installation_id, auth_manager.clone(), models_manager.clone(), @@ -857,7 +849,10 @@ impl Session { .await .map_err(|e| { error!("Failed to create session: {e:#}"); - map_session_init_error(&e, &config.codex_home) + match e.downcast::() { + Ok(error) => error, + Err(error) => map_session_init_error(&error, &config.codex_home), + } })?; if let Some(message) = initial_service_tier_warning { session @@ -1928,8 +1923,21 @@ impl Session { .clone() } - pub(crate) async fn user_instructions(&self) -> Option { - self.services.agents_md_manager.user_instructions() + pub(crate) async fn inherited_instructions(&self) -> SessionInstructions { + self.services + .agents_md_manager + .inherited_instructions() + .await + } + + pub(crate) async fn emit_instruction_warnings(&self, warnings: Vec) { + for message in warnings { + self.send_event_raw(Event { + id: INITIAL_SUBMIT_ID.to_owned(), + msg: EventMsg::Warning(WarningEvent { message }), + }) + .await; + } } pub(crate) async fn provider(&self) -> ModelProviderInfo { @@ -3712,11 +3720,14 @@ impl Session { let session_telemetry = settings.telemetry(&turn_context.session_telemetry); // Keep selections fixed for the turn while allowing their startup work to finish. let environments = turn_context.environments.refresh_readiness(); - self.services + let (loaded_agents_md, warnings) = self + .services .agents_md_manager .refresh(&turn_context.config, &environments) + .or_cancel(cancellation_token) .await?; - let loaded_agents_md = self.services.agents_md_manager.get_loaded().await; + self.emit_instruction_warnings(warnings).await; + let loaded_agents_md = loaded_agents_md?; let selected_capability_roots = self .resolve_selected_capability_roots_for_step(&environments) .await; diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 047eca6d84..101fdccb94 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -6,6 +6,7 @@ use super::step_settings::StepSettingsConstraints; use super::step_settings::StepSettingsUpdate; use super::*; use crate::agents_md_manager::AgentsMdManager; +use crate::agents_md_manager::SessionInstructions; use crate::config::ConstraintError; use crate::context::GuardianContextMode; use crate::environment_selection::ThreadEnvironments; @@ -667,7 +668,7 @@ impl Session { mut session_configuration: SessionConfiguration, environment_selections: &[TurnEnvironmentSelection], config: Arc, - user_instructions: Option, + instructions: SessionInstructions, installation_id: String, auth_manager: Arc, models_manager: SharedModelsManager, @@ -1300,7 +1301,7 @@ impl Session { &session_configuration.inferred_environment_config(), ); let resolved_environments = turn_environments.snapshot().await; - let agents_md_manager = Arc::new(AgentsMdManager::new(user_instructions)); + let agents_md_manager = Arc::new(AgentsMdManager::new(instructions)); let plugin_skill_warmup = warm_plugins_and_skills_for_session_init( Arc::clone(&config), Arc::clone(&plugins_manager), @@ -1318,13 +1319,20 @@ impl Session { "session_init.thread_name_lookup", otel.name = "session_init.thread_name_lookup", )); - let (agents_md_result, plugin_skill_errors, thread_name) = tokio::join!( + let (instruction_refresh, plugin_skill_errors, thread_name) = tokio::join!( agents_md_manager.refresh(config.as_ref(), &resolved_environments), plugin_skill_warmup, thread_name_lookup, ); + let (agents_md_result, instruction_warnings) = instruction_refresh; // TODO(anp): Present AGENTS.md discovery errors more clearly to the user. agents_md_result?; + post_session_configured_events.extend( + instruction_warnings.into_iter().map(|message| Event { + id: INITIAL_SUBMIT_ID.to_owned(), + msg: EventMsg::Warning(WarningEvent { message }), + }), + ); for err in &plugin_skill_errors { error!( "failed to load skill {}: {}", diff --git a/codex-rs/core/src/session/step_activation_tests.rs b/codex-rs/core/src/session/step_activation_tests.rs index f5887b10b0..828a5d36b5 100644 --- a/codex-rs/core/src/session/step_activation_tests.rs +++ b/codex-rs/core/src/session/step_activation_tests.rs @@ -1,4 +1,6 @@ use super::*; +use crate::agents_md_manager::AgentsMdManager; +use crate::agents_md_manager::SessionInstructions; use crate::guardian::BUNDLED_GUARDIAN_POLICY; use crate::session::handlers::submission_loop; use crate::session::step_context::StepContext; @@ -15,6 +17,10 @@ use codex_config::ConfigRequirementsToml; use codex_config::ConfigRequirementsWithSources; use codex_config::RequirementSource; use codex_config::Sourced; +use codex_extension_api::Instructions; +use codex_extension_api::LoadUserInstructionsFuture; +use codex_extension_api::LoadedUserInstructions; +use codex_extension_api::UserInstructionsProvider; use codex_http_client::HttpClientFactory; use codex_login::AuthManager; use codex_models_manager::ModelsManagerConfig; @@ -41,6 +47,8 @@ use codex_protocol::protocol::TurnAbortReason; use pretty_assertions::assert_eq; use std::collections::BTreeSet; use std::sync::Mutex as StdMutex; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; use std::time::Duration; use test_case::test_case; use tokio::sync::Notify; @@ -52,6 +60,95 @@ use tokio_util::sync::CancellationToken; const MODEL_A: &str = "step-activation-a"; const MODEL_B: &str = "step-activation-b"; +#[tokio::test] +async fn instruction_refresh_serializes_reads_and_releases_on_cancellation() { + struct GatedInstructionsProvider { + instructions: StdMutex>, + block_next_read: AtomicBool, + read_started: Notify, + } + + impl UserInstructionsProvider for GatedInstructionsProvider { + fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_> { + let instructions = self + .instructions + .lock() + .expect("instruction snapshot") + .clone(); + let block = self.block_next_read.swap(/*val*/ false, Ordering::SeqCst); + Box::pin(async move { + if block { + self.read_started.notify_one(); + std::future::pending::<()>().await; + } + LoadedUserInstructions { + instructions, + warnings: Vec::new(), + } + }) + } + } + + let (mut session, turn) = make_session_and_context().await; + let turn = Arc::new(turn); + let source = turn.config.codex_home.join("AGENTS.md"); + let instructions = |text: &str| { + Some(Instructions { + text: text.to_string(), + source: source.clone(), + }) + }; + let provider = Arc::new(GatedInstructionsProvider { + instructions: StdMutex::new(instructions("initial")), + block_next_read: AtomicBool::new(/*v*/ false), + read_started: Notify::new(), + }); + session.services.agents_md_manager = Arc::new(AgentsMdManager::new(SessionInstructions { + user_provider: Some(provider.clone()), + ..Default::default() + })); + let session = Arc::new(session); + session + .services + .agents_md_manager + .refresh(&turn.config, &turn.environments) + .await + .0 + .expect("install initial provider"); + *provider.instructions.lock().expect("instruction snapshot") = instructions("cancelled"); + provider + .block_next_read + .store(/*val*/ true, Ordering::SeqCst); + + let cancellation = CancellationToken::new(); + let first_capture = session.capture_step_context(Arc::clone(&turn), &cancellation); + tokio::pin!(first_capture); + tokio::select! { + _ = &mut first_capture => panic!("capture must wait for provider"), + () = provider.read_started.notified() => {} + } + + *provider.instructions.lock().expect("instruction snapshot") = instructions("latest"); + let next_cancellation = CancellationToken::new(); + let next_capture = session.capture_step_context(turn, &next_cancellation); + tokio::pin!(next_capture); + assert!(futures::poll!(&mut next_capture).is_pending()); + assert_eq!( + session.inherited_instructions().await.user, + instructions("initial") + ); + + cancellation.cancel(); + assert!(first_capture.await.is_err()); + next_capture + .await + .expect("cancelled read releases the refresh guard"); + assert_eq!( + session.inherited_instructions().await.user, + instructions("latest") + ); +} + fn activation_models() -> Vec { let model = bundled_models_response() .expect("bundled models") diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 1ed7f88979..b77be1ae0d 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6,6 +6,7 @@ pub(crate) use super::step_settings::tests::update_selected_settings_for_test; use super::turn_context::TurnEnvironment; use super::*; use crate::agents_md_manager::AgentsMdManager; +use crate::agents_md_manager::SessionInstructions; use crate::compact::InitialContextInjection; use crate::config::ConfigBuilder; use crate::config::ConfigOverrides; @@ -6431,7 +6432,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { session_configuration, /*environment_selections*/ &[], Arc::clone(&config), - /*user_instructions*/ None, + SessionInstructions::default(), "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, @@ -6673,7 +6674,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { tool_approvals: Mutex::new(ApprovalStore::default()), runtime_handle: tokio::runtime::Handle::current(), skills_service, - agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)), + agents_md_manager: Arc::new(AgentsMdManager::new(SessionInstructions::default())), plugins_manager, mcp_manager, extensions: Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), @@ -6912,7 +6913,7 @@ async fn make_session_with_config_and_rx( session_configuration, &default_environments, Arc::clone(&config), - /*user_instructions*/ None, + SessionInstructions::default(), "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, @@ -7041,7 +7042,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx( session_configuration, &default_environments, Arc::clone(&config), - /*user_instructions*/ None, + SessionInstructions::default(), "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, @@ -8847,7 +8848,7 @@ where tool_approvals: Mutex::new(ApprovalStore::default()), runtime_handle: tokio::runtime::Handle::current(), skills_service, - agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)), + agents_md_manager: Arc::new(AgentsMdManager::new(SessionInstructions::default())), plugins_manager, mcp_manager, extensions: Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 5a4c79a98a..e474b32621 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -1262,7 +1262,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { let (session, io) = Session::spawn(SessionSpawnArgs { config, allow_provider_model_fallback: false, - user_instructions: Default::default(), + instructions: Default::default(), installation_id: "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 7fc5f3ac72..9ce622f540 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -1,5 +1,6 @@ use crate::CodexAppsToolsCache; use crate::agent::AgentControl; +use crate::agents_md_manager::SessionInstructions; use crate::attestation::AttestationProvider; use crate::codex_thread::CodexThread; use crate::config::Config; @@ -1652,53 +1653,37 @@ impl ThreadManagerState { resolve_multi_agent_version(initial_history, inherited_multi_agent_version) } - /// Resolves the provider snapshot for a newly spawned runtime. + /// Selects instruction sources for a newly spawned runtime without loading providers. /// - /// Loads a fresh provider snapshot for: - /// - fresh root threads; - /// - cold resumes; - /// - root forks. - /// - /// Uses an existing snapshot for: - /// - subagents, which inherit from their parent without invoking the - /// provider; - /// - running resumes and compaction paths, which retain the live session. - /// - /// Provider warnings only apply to fresh loads. If a parent runtime is no - /// longer available, its child starts without provider instructions rather - /// than loading independently. - async fn user_instructions_for_spawn( + /// Fresh roots, cold resumes, and root forks retain the global provider. The + /// session's AgentsMdManager loads it at startup and subsequent context captures. + /// Subagents inherit their live parent's applied snapshot without its provider; + /// if that parent is unavailable, they start without inherited instructions. + /// Warm resumes retain the existing session and do not call this function. + async fn instructions_for_spawn( &self, session_source: &SessionSource, parent_thread_id: Option, forked_from_thread_id: Option, - ) -> LoadedUserInstructions { - let is_root_agent = !session_source.is_non_root_agent(); - if is_root_agent { - return self - .user_instructions_provider - .load_user_instructions() - .await; + ) -> SessionInstructions { + if !session_source.is_non_root_agent() { + return SessionInstructions { + user_provider: Some(Arc::clone(&self.user_instructions_provider)), + ..Default::default() + }; } - let inherited_thread_id = match session_source { SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, .. }) => Some(*parent_thread_id), _ => parent_thread_id.or(forked_from_thread_id), }; - let instructions = match inherited_thread_id { - // The spawn path retains only thread IDs, so look up the live - // runtime again here to inherit its user instructions. + match inherited_thread_id { Some(thread_id) => match self.get_thread(thread_id).await { - Ok(thread) => thread.session.user_instructions().await, - Err(_) => None, + Ok(thread) => thread.session.inherited_instructions().await, + Err(_) => SessionInstructions::default(), }, - None => None, - }; - LoadedUserInstructions { - instructions, - warnings: Vec::new(), + None => SessionInstructions::default(), } } @@ -1920,7 +1905,7 @@ impl ThreadManagerState { user_shell_override, } = request; let StartThreadOptions { - config, + mut config, allow_provider_model_fallback, initial_history, history_mode, @@ -1996,41 +1981,41 @@ impl ThreadManagerState { threads.remove(&resumed.conversation_id); } } - let ( - user_instructions, - inherited_exec_policy, - extensions, - mcp_manager, - multi_agent_version, - ) = if isolation == codex_extension_api::SessionIsolation::Isolated { - ( - LoadedUserInstructions::default(), - None, - empty_extension_registry(), - Arc::new(McpManager::new(Arc::clone(&self.plugins_manager))), - Some(MultiAgentVersion::Disabled), - ) - } else { - ( - self.user_instructions_for_spawn( - &session_source, - parent_thread_id, - forked_from_thread_id, + let (instructions, inherited_exec_policy, extensions, mcp_manager, multi_agent_version) = + if isolation == codex_extension_api::SessionIsolation::Isolated { + ( + SessionInstructions::default(), + None, + empty_extension_registry(), + Arc::new(McpManager::new(Arc::clone(&self.plugins_manager))), + Some(MultiAgentVersion::Disabled), ) - .await, - inherited_exec_policy, - Arc::clone(&self.extensions), - Arc::clone(&self.mcp_manager), - self.initial_multi_agent_version_for_spawn( - &initial_history, - Some(&session_source), - parent_thread_id, - forked_from_thread_id, + } else { + ( + self.instructions_for_spawn( + &session_source, + parent_thread_id, + forked_from_thread_id, + ) + .await, + inherited_exec_policy, + Arc::clone(&self.extensions), + Arc::clone(&self.mcp_manager), + self.initial_multi_agent_version_for_spawn( + &initial_history, + Some(&session_source), + parent_thread_id, + forked_from_thread_id, + ) + .await, ) - .await, - ) - }; - let user_instructions = supplied_user_instructions.unwrap_or(user_instructions); + }; + let mut instructions = instructions; + if let Some(supplied) = supplied_user_instructions { + instructions.user = supplied.instructions; + instructions.user_provider = None; + config.startup_warnings.extend(supplied.warnings); + } let parent_rollout_thread_trace = self .parent_rollout_thread_trace_for_source(&session_source, &initial_history) .await; @@ -2064,7 +2049,7 @@ impl ThreadManagerState { let (session, io) = Session::spawn(SessionSpawnArgs { config, allow_provider_model_fallback, - user_instructions, + instructions, installation_id: self.installation_id.clone(), auth_manager, models_manager: Arc::clone(&self.models_manager), diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index f4a41b8482..88fe058d5e 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -1233,7 +1233,7 @@ async fn spawn_internal_session_preserves_parent_lineage_without_forking_history .session .set_multi_agent_version_if_unset(MultiAgentVersion::V2); assert_eq!( - parent.thread.session.user_instructions().await, + parent.thread.session.inherited_instructions().await.user, Some(parent_instructions) ); assert_eq!( @@ -1317,7 +1317,15 @@ async fn spawn_internal_session_preserves_parent_lineage_without_forking_history Some(parent.thread_id) ); assert_eq!(reviewer.session_configured.forked_from_id, None); - assert!(reviewer.thread.session.user_instructions().await.is_none()); + assert!( + reviewer + .thread + .session + .inherited_instructions() + .await + .user + .is_none() + ); assert!( reviewer .thread diff --git a/codex-rs/core/tests/suite/agents_md.rs b/codex-rs/core/tests/suite/agents_md.rs index e0ca380b37..4979b9df78 100644 --- a/codex-rs/core/tests/suite/agents_md.rs +++ b/codex-rs/core/tests/suite/agents_md.rs @@ -6,6 +6,8 @@ use codex_core::TurnInputRequest; use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::LOCAL_ENVIRONMENT_ID; use codex_exec_server::REMOTE_ENVIRONMENT_ID; +use codex_extension_api::LoadUserInstructionsFuture; +use codex_extension_api::UserInstructionsProvider; use codex_features::Feature; use codex_history::RolloutItem; use codex_home::CodexHomeUserInstructionsProvider; @@ -46,9 +48,14 @@ use pretty_assertions::assert_eq; use serde_json::json; use std::path::Path; use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; use std::time::Duration; use tempfile::TempDir; +#[path = "agents_md_refresh.rs"] +mod refresh; + const GLOBAL_AGENTS_FILENAME: &str = "AGENTS.md"; const GLOBAL_AGENTS_OVERRIDE_FILENAME: &str = "AGENTS.override.md"; const GLOBAL_INSTRUCTIONS: &str = "global instructions"; @@ -62,6 +69,24 @@ const SPAWN_CHILD_PROMPT: &str = "inspect inherited global instructions"; const SPAWN_FRESH_PARENT_PROMPT: &str = "spawn a child with fresh context"; const SPAWN_PARENT_PROMPT: &str = "spawn a child with the parent context"; const SPAWN_SEED_PROMPT: &str = "seed parent history"; +const PROVIDER_WARNING: &str = "global instruction source unavailable; using fallback"; + +struct WarningInstructionsProvider { + inner: CodexHomeUserInstructionsProvider, + warning_active: AtomicBool, +} + +impl UserInstructionsProvider for WarningInstructionsProvider { + fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_> { + Box::pin(async move { + let mut loaded = self.inner.load_user_instructions().await; + if self.warning_active.load(Ordering::SeqCst) { + loaded.warnings = vec![PROVIDER_WARNING.to_string()]; + } + loaded + }) + } +} async fn agents_instructions(mut builder: TestCodexBuilder) -> Result { let server = start_mock_server().await; @@ -135,26 +160,6 @@ fn expected_provider_only_instruction_fragment(contents: &str) -> String { format!("# AGENTS.md instructions\n\n\n{contents}\n") } -fn assert_instruction_replacement_once( - requests: &[responses::ResponsesRequest], - initial_contents: &str, - replacement_contents: &str, -) { - let initial = expected_provider_only_instruction_fragment(initial_contents); - let replacement = expected_provider_only_instruction_fragment(&format!( - "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{replacement_contents}" - )); - assert_eq!(instruction_fragments(&requests[0]), vec![initial.clone()]); - assert_eq!( - instruction_fragments(&requests[1]), - vec![initial.clone(), replacement.clone()] - ); - assert_eq!( - instruction_fragments(&requests[2]), - vec![initial, replacement] - ); -} - fn assert_single_instruction_fragment(request: &responses::ResponsesRequest, expected: &str) { assert_eq!(instruction_fragments(request), vec![expected.to_string()]); } @@ -734,7 +739,7 @@ async fn symlinked_writable_root_reports_sandbox_failure_instead_of_session_corr .replace(&canonical_home_path, "$CODEX_HOME") .replace(&home_path, "$CODEX_HOME"); insta::assert_snapshot!(error, @" - Fatal error: Failed to initialize session: failed to load AGENTS.md instructions for environment `local`: failed to prepare fs sandbox: failed to prepare Seatbelt sandbox: writable root $CODEX_HOME/visualizations contains symlink component $CODEX_HOME/visualizations; symlinked writable roots are not supported. + failed to load AGENTS.md instructions for environment `local`: failed to prepare fs sandbox: failed to prepare Seatbelt sandbox: writable root $CODEX_HOME/visualizations contains symlink component $CODEX_HOME/visualizations; symlinked writable roots are not supported. If this writable root is at or beneath CODEX_HOME and you trust its symlink targets, set `allow_symlinked_codex_home = true` at the top level of `$CODEX_HOME/config.toml` (normally `~/.codex/config.toml`) on the execution host, then restart Codex or its executor. This opt-out trusts targets outside CODEX_HOME and targets changed between commands. It does not apply to other writable roots. "); @@ -757,23 +762,38 @@ async fn tightening_environment_read_permissions_invalidates_cached_project_inst sse(vec![ev_response_created("resp1"), ev_completed("resp1")]), ) .await; - let mut builder = test_codex().with_workspace_setup(|cwd, fs| async move { - fs.write_file( - &executor_path_uri(cwd.join(GLOBAL_AGENTS_FILENAME))?, - PROJECT_INSTRUCTIONS.as_bytes().to_vec(), - Default::default(), - /*sandbox*/ None, - ) - .await?; - Ok(()) + let home = Arc::new(TempDir::new()?); + let global_source = write_global_file(&home, GLOBAL_AGENTS_FILENAME, GLOBAL_INSTRUCTIONS)?; + let provider = Arc::new(WarningInstructionsProvider { + inner: CodexHomeUserInstructionsProvider::new(home.path().to_path_buf().abs()), + warning_active: AtomicBool::new(/*v*/ false), }); + let mut builder = test_codex() + .with_home(home) + .with_user_instructions_provider(provider.clone()) + .with_workspace_setup(|cwd, fs| async move { + fs.write_file( + &executor_path_uri(cwd.join(GLOBAL_AGENTS_FILENAME))?, + PROJECT_INSTRUCTIONS.as_bytes().to_vec(), + Default::default(), + /*sandbox*/ None, + ) + .await?; + Ok(()) + }); let test = builder.build_with_auto_env(&server).await?; assert_eq!( test.codex.instruction_sources().await, - vec![test.workspace_path_uri(GLOBAL_AGENTS_FILENAME)?] + vec![ + PathUri::from_abs_path(&global_source), + test.workspace_path_uri(GLOBAL_AGENTS_FILENAME)? + ] ); + provider + .warning_active + .store(/*val*/ true, Ordering::SeqCst); let mut file_system_policy = FileSystemSandboxPolicy::read_only(); file_system_policy.entries.push(FileSystemSandboxEntry::new( test.config.cwd.join(GLOBAL_AGENTS_FILENAME).into(), @@ -799,11 +819,18 @@ async fn tightening_environment_read_permissions_invalidates_cached_project_inst ) .await?; - let EventMsg::Error(error) = - wait_for_event(&test.codex, |event| matches!(event, EventMsg::Error(_))).await + let mut warnings_before_error = Vec::new(); + let EventMsg::Error(error) = wait_for_event(&test.codex, |event| { + if let EventMsg::Warning(warning) = event { + warnings_before_error.push(warning.message.clone()); + } + matches!(event, EventMsg::Error(_)) + }) + .await else { unreachable!(); }; + assert_eq!(warnings_before_error, vec![PROVIDER_WARNING.to_string()]); assert!( error.message.contains("AGENTS.md"), "turn should report the unreadable project instructions: {}", @@ -958,8 +985,8 @@ async fn fresh_thread_composes_global_before_project_and_reports_sources() -> Re ); test.submit_turn("second turn").await?; - // Assert the running thread keeps its original rendering and structured prefix even though - // both files at the reported source paths now contain different text. + // The global provider refreshes, while repository discovery keeps its cached snapshot. + // Append the changed instructions without rewriting the earlier model input. let requests = response_mock.requests(); assert_eq!(requests.len(), 2); let expected_contents = @@ -970,7 +997,16 @@ async fn fresh_thread_composes_global_before_project_and_reports_sources() -> Re ); let fragments = instruction_fragments(&requests[0]); assert_eq!(fragments, vec![expected_fragment.clone()]); - assert_single_instruction_fragment(&requests[1], &expected_fragment); + let updated_fragment = expected_instruction_fragment( + &test.executor_environment().selection().cwd, + &format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{NEW_GLOBAL_INSTRUCTIONS}\n\n{PROJECT_SEPARATOR}\n\n{PROJECT_INSTRUCTIONS}" + ), + ); + assert_eq!( + instruction_fragments(&requests[1]), + vec![expected_fragment, updated_fragment] + ); let rendered = fragments .into_iter() .next() @@ -996,7 +1032,7 @@ async fn fresh_thread_composes_global_before_project_and_reports_sources() -> Re assert_eq!( test.codex.instruction_sources().await, creation_sources, - "ordinary turns retain the creation-time source list" + "same-path global refresh preserves source paths and composition order" ); let first_input = requests[0].input(); let second_input = requests[1].input(); @@ -1077,7 +1113,7 @@ async fn multi_environment_project_instructions_share_one_byte_budget() -> Resul } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn multi_environment_thread_loads_every_project_and_keeps_creation_snapshot() -> Result<()> { +async fn multi_environment_thread_refreshes_global_and_keeps_repository_snapshot() -> Result<()> { skip_if_no_network!(Ok(())); skip_if_no_remote_env!(Ok(())); @@ -1154,7 +1190,7 @@ async fn multi_environment_thread_loads_every_project_and_keeps_creation_snapsho submit_thread_turn(&thread.thread, "first multi-environment turn").await?; - write_global_file( + let new_global_source = write_global_file( home.as_ref(), GLOBAL_AGENTS_OVERRIDE_FILENAME, NEW_GLOBAL_INSTRUCTIONS, @@ -1186,12 +1222,19 @@ async fn multi_environment_thread_loads_every_project_and_keeps_creation_snapsho let requests = response_mock.requests(); assert_eq!(requests.len(), 2); assert_single_instruction_fragment(&requests[0], &expected); - assert_single_instruction_fragment(&requests[1], &expected); - assert_eq!(provider.load_count(), 2); + let replacement = expected_provider_only_instruction_fragment(&format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{}", + contents.replace(GLOBAL_INSTRUCTIONS, NEW_GLOBAL_INSTRUCTIONS), + )); + assert_eq!( + instruction_fragments(&requests[1]), + vec![expected, replacement] + ); + assert_eq!(provider.load_count(), 4); assert_eq!( thread.thread.instruction_sources().await, vec![ - PathUri::from_abs_path(&global_source), + PathUri::from_abs_path(&new_global_source), executor_path_uri(&remote_source)?, PathUri::from_host_native_path(&local_source)?, ] @@ -1200,6 +1243,87 @@ async fn multi_environment_thread_loads_every_project_and_keeps_creation_snapsho Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn global_instruction_warnings_reappear_only_after_recovery() -> Result<()> { + let server = responses::start_mock_server().await; + let response_mock = responses::mount_sse_sequence( + &server, + (0..4) + .map(|index| { + responses::sse(vec![ + responses::ev_response_created(&format!("response-{index}")), + responses::ev_completed(&format!("response-{index}")), + ]) + }) + .collect(), + ) + .await; + let home = Arc::new(TempDir::new()?); + write_global_file(home.as_ref(), GLOBAL_AGENTS_FILENAME, GLOBAL_INSTRUCTIONS)?; + let override_path = home.path().join(GLOBAL_AGENTS_OVERRIDE_FILENAME); + let create_unreadable_override = || { + #[cfg(unix)] + { + std::os::unix::fs::symlink(GLOBAL_AGENTS_OVERRIDE_FILENAME, &override_path) + } + #[cfg(windows)] + { + std::os::windows::fs::symlink_file(GLOBAL_AGENTS_OVERRIDE_FILENAME, &override_path) + } + }; + create_unreadable_override()?; + let read_error = std::fs::read(&override_path).expect_err("symlink loop must be unreadable"); + let expected_warning = format!( + "Failed to read global AGENTS.md instructions from `{}`: {read_error}", + override_path.display() + ); + let mut builder = test_codex().with_home(Arc::clone(&home)); + let test = builder.build_with_auto_env(&server).await?; + wait_for_event( + &test.codex, + |event| matches!(event, EventMsg::Warning(warning) if warning.message == expected_warning), + ) + .await; + + for (warning_active, expected_warnings) in [ + (true, Vec::new()), + (false, Vec::new()), + (true, vec![expected_warning]), + (true, Vec::new()), + ] { + std::fs::remove_file(&override_path)?; + if warning_active { + create_unreadable_override()?; + } else { + std::fs::write(&override_path, "")?; + } + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "check the instructions".to_string(), + text_elements: Vec::new(), + }])) + .await?; + let mut warnings = Vec::new(); + wait_for_event(&test.codex, |event| { + if let EventMsg::Warning(warning) = event { + warnings.push(warning.message.clone()); + } + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + assert_eq!(warnings, expected_warnings); + } + let requests = response_mock.requests(); + assert_eq!(requests.len(), 4); + for request in &requests { + assert_single_instruction_fragment( + request, + &expected_provider_only_instruction_fragment(GLOBAL_INSTRUCTIONS), + ); + } + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn invalid_utf8_global_instructions_are_lossy() -> Result<()> { let server = responses::start_mock_server().await; @@ -1253,6 +1377,7 @@ async fn cold_resume_invalidates_deleted_legacy_agents_md_once() -> Result<()> { responses::ev_response_created("second-resumed-response"), responses::ev_completed("second-resumed-response"), ]), + responses::sse_completed("refreshed-resumed-response"), ], ) .await; @@ -1303,9 +1428,11 @@ async fn cold_resume_invalidates_deleted_legacy_agents_md_once() -> Result<()> { resumed.submit_turn("continue resumed thread").await?; resumed.submit_turn("continue again").await?; + write_global_file(&home, GLOBAL_AGENTS_FILENAME, NEW_GLOBAL_INSTRUCTIONS)?; + resumed.submit_turn("refresh after resume").await?; let requests = response_mock.requests(); - assert_eq!(requests.len(), 3); + assert_eq!(requests.len(), 4); let initial_input = requests[0].input(); let resumed_input = requests[1].input(); assert_eq!( @@ -1322,7 +1449,15 @@ async fn cold_resume_invalidates_deleted_legacy_agents_md_once() -> Result<()> { instruction_fragments(&requests[1]), vec![initial.clone(), removal.clone()] ); - assert_eq!(instruction_fragments(&requests[2]), vec![initial, removal]); + assert_eq!( + instruction_fragments(&requests[2]), + vec![initial.clone(), removal.clone()] + ); + let replacement = expected_provider_only_instruction_fragment(NEW_GLOBAL_INSTRUCTIONS); + assert_eq!( + instruction_fragments(&requests[3]), + vec![initial, removal, replacement] + ); Ok(()) } @@ -1346,6 +1481,7 @@ async fn fork_injects_changed_agents_md_once() -> Result<()> { responses::ev_response_created("second-fork-response"), responses::ev_completed("second-fork-response"), ]), + responses::sse_completed("refreshed-fork-response"), ], ) .await; @@ -1406,10 +1542,16 @@ async fn fork_injects_changed_agents_md_once() -> Result<()> { submit_thread_turn(&forked.thread, "continue fork").await?; submit_thread_turn(&forked.thread, "continue fork again").await?; + write_global_file( + &home, + GLOBAL_AGENTS_OVERRIDE_FILENAME, + "instructions changed after fork", + )?; + submit_thread_turn(&forked.thread, "refresh after fork").await?; // Assert the forked model request replays the parent's exact structured history. let requests = response_mock.requests(); - assert_eq!(requests.len(), 3); + assert_eq!(requests.len(), 4); let parent_input = requests[0].input(); let fork_input = requests[1].input(); assert_eq!( @@ -1417,23 +1559,38 @@ async fn fork_injects_changed_agents_md_once() -> Result<()> { Some(parent_input.as_slice()), "fork should replay the parent's original structured input prefix" ); - assert_instruction_replacement_once( - &requests, - OLD_GLOBAL_INSTRUCTIONS, - NEW_GLOBAL_INSTRUCTIONS, + let initial = expected_provider_only_instruction_fragment(OLD_GLOBAL_INSTRUCTIONS); + let replacement = expected_provider_only_instruction_fragment(&format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{NEW_GLOBAL_INSTRUCTIONS}" + )); + assert_eq!(instruction_fragments(&requests[0]), vec![initial.clone()]); + assert_eq!( + instruction_fragments(&requests[1]), + vec![initial.clone(), replacement.clone()] + ); + assert_eq!( + instruction_fragments(&requests[2]), + vec![initial.clone(), replacement.clone()] + ); + let refreshed = expected_provider_only_instruction_fragment( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\ninstructions changed after fork", + ); + assert_eq!( + instruction_fragments(&requests[3]), + vec![initial, replacement, refreshed] ); Ok(()) } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn forked_subagent_replays_one_creation_time_global_instruction_fragment() -> Result<()> { +async fn forked_subagent_replays_parent_applied_global_instructions() -> Result<()> { skip_if_no_network!(Ok(())); run_subagent_global_instruction_case(/*fork_context*/ true).await } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn fresh_subagent_uses_creation_time_instructions_without_parent_history() -> Result<()> { +async fn fresh_subagent_uses_parent_applied_instructions_without_parent_history() -> Result<()> { skip_if_no_network!(Ok(())); run_subagent_global_instruction_case(/*fork_context*/ false).await } @@ -1553,20 +1710,32 @@ async fn run_subagent_global_instruction_case(fork_context: bool) -> Result<()> .await .map_err(|_| anyhow!("timed out waiting for the subagent request"))?; - // Assert parent and child report and render the parent's creation-time snapshot exactly once. + // The parent refreshes global instructions before spawning. The child inherits + // that applied snapshot without independently loading the global provider. let expected_fragment = expected_provider_only_instruction_fragment(OLD_GLOBAL_INSTRUCTIONS); assert_single_instruction_fragment(&seed_request, &expected_fragment); - assert_single_instruction_fragment(&spawn_request, &expected_fragment); - assert_single_instruction_fragment(&child_request, &expected_fragment); + let replacement = expected_provider_only_instruction_fragment(&format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{NEW_GLOBAL_INSTRUCTIONS}" + )); + let inherited_fragments = vec![expected_fragment, replacement]; + assert_eq!(instruction_fragments(&spawn_request), inherited_fragments); + if fork_context { + assert_eq!(instruction_fragments(&child_request), inherited_fragments); + } else { + assert_single_instruction_fragment( + &child_request, + &expected_provider_only_instruction_fragment(NEW_GLOBAL_INSTRUCTIONS), + ); + } assert_eq!( test.codex.instruction_sources().await, - vec![PathUri::from_abs_path(&source)], - "running parent retains the creation-time global source after spawning" + vec![PathUri::from_abs_path(&new_source)], + "parent reports the refreshed global source" ); assert_eq!( child_thread.instruction_sources().await, - vec![PathUri::from_abs_path(&source)], - "subagent reports the parent's creation-time source" + vec![PathUri::from_abs_path(&new_source)], + "subagent reports the parent's applied global source" ); if fork_context { let seed_input = seed_request.input(); @@ -1596,5 +1765,35 @@ async fn run_subagent_global_instruction_case(fork_context: bool) -> Result<()> ); } + wait_for_event(&child_thread, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + write_global_file( + &home, + GLOBAL_AGENTS_OVERRIDE_FILENAME, + "instructions changed after child creation", + )?; + let child_follow_up = + responses::mount_sse_once(&server, responses::sse_completed("child-follow-up")).await; + submit_thread_turn(&child_thread, "continue with inherited instructions").await?; + assert_eq!( + instruction_fragments(&child_follow_up.single_request()), + instruction_fragments(&child_request), + "children keep the parent's snapshot even after the global source changes", + ); + let parent_follow_up = + responses::mount_sse_once(&server, responses::sse_completed("parent-refresh")).await; + test.submit_turn("refresh the parent independently").await?; + let refreshed = expected_provider_only_instruction_fragment( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\ninstructions changed after child creation", + ); + let mut parent_fragments = instruction_fragments(&spawn_request); + parent_fragments.push(refreshed); + assert_eq!( + instruction_fragments(&parent_follow_up.single_request()), + parent_fragments + ); + Ok(()) } diff --git a/codex-rs/core/tests/suite/agents_md_refresh.rs b/codex-rs/core/tests/suite/agents_md_refresh.rs new file mode 100644 index 0000000000..dfd6c7f4e6 --- /dev/null +++ b/codex-rs/core/tests/suite/agents_md_refresh.rs @@ -0,0 +1,254 @@ +//! Exercises global-instruction refreshes through real turns and provider reads. + +use super::*; +use codex_protocol::request_user_input::RequestUserInputAnswer; +use codex_protocol::request_user_input::RequestUserInputResponse; +use pretty_assertions::assert_eq; +use std::collections::HashMap; +use tokio::sync::Notify; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_global_read_keeps_instructions_until_recovery() -> Result<()> { + let server = start_mock_server().await; + let requests = responses::mount_sse_sequence( + &server, + ["initial", "failed-read", "recovered"] + .map(responses::sse_completed) + .to_vec(), + ) + .await; + let home = Arc::new(TempDir::new()?); + let source = write_global_file(&home, GLOBAL_AGENTS_FILENAME, GLOBAL_INSTRUCTIONS)?; + let mut builder = test_codex().with_home(Arc::clone(&home)); + let test = builder.build_with_auto_env(&server).await?; + test.submit_turn("initial instructions").await?; + + std::fs::remove_file(&source)?; + #[cfg(unix)] + std::os::unix::fs::symlink(GLOBAL_AGENTS_FILENAME, &source)?; + #[cfg(windows)] + std::os::windows::fs::symlink_file(GLOBAL_AGENTS_FILENAME, &source)?; + test.submit_turn("keep instructions through the read failure") + .await?; + assert_eq!( + test.codex.instruction_sources().await, + vec![PathUri::from_abs_path(&source)], + ); + + std::fs::remove_file(&source)?; + write_global_file(&home, GLOBAL_AGENTS_FILENAME, NEW_GLOBAL_INSTRUCTIONS)?; + test.submit_turn("load recovered instructions").await?; + let initial = expected_provider_only_instruction_fragment(GLOBAL_INSTRUCTIONS); + let replacement = expected_provider_only_instruction_fragment(&format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{NEW_GLOBAL_INSTRUCTIONS}" + )); + assert_eq!( + requests + .requests() + .iter() + .map(instruction_fragments) + .collect::>(), + vec![ + vec![initial.clone()], + vec![initial.clone()], + vec![initial, replacement] + ], + ); + Ok(()) +} + +#[test_case::test_case(None; "deleted")] +#[test_case::test_case(Some(" \n"); "blank")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn live_global_removal_preserves_repository_instructions( + contents: Option<&str>, +) -> Result<()> { + let server = start_mock_server().await; + let requests = responses::mount_sse_sequence( + &server, + ["initial", "removed", "unchanged"] + .map(responses::sse_completed) + .to_vec(), + ) + .await; + let home = Arc::new(TempDir::new()?); + let source = write_global_file(&home, GLOBAL_AGENTS_FILENAME, GLOBAL_INSTRUCTIONS)?; + let mut builder = test_codex() + .with_home(Arc::clone(&home)) + .with_workspace_setup(|cwd, fs| async move { + fs.write_file( + &executor_path_uri(cwd.join(GLOBAL_AGENTS_FILENAME))?, + PROJECT_INSTRUCTIONS.as_bytes().to_vec(), + Default::default(), + /*sandbox*/ None, + ) + .await?; + Ok(()) + }); + let test = builder.build_with_auto_env(&server).await?; + test.submit_turn("initial instructions").await?; + match contents { + Some(contents) => std::fs::write(&source, contents)?, + None => std::fs::remove_file(&source)?, + } + test.submit_turn("remove global instructions").await?; + test.submit_turn("keep repository instructions").await?; + assert_eq!( + test.codex.instruction_sources().await, + vec![test.workspace_path_uri(GLOBAL_AGENTS_FILENAME)?], + ); + let cwd = &test.executor_environment().selection().cwd; + let initial = expected_instruction_fragment( + cwd, + &format!("{GLOBAL_INSTRUCTIONS}\n\n{PROJECT_SEPARATOR}\n\n{PROJECT_INSTRUCTIONS}"), + ); + let replacement = expected_instruction_fragment( + cwd, + &format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{PROJECT_INSTRUCTIONS}" + ), + ); + assert_eq!( + requests + .requests() + .iter() + .map(instruction_fragments) + .collect::>(), + vec![ + vec![initial.clone()], + vec![initial.clone(), replacement.clone()], + vec![initial, replacement] + ], + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn global_instructions_refresh_after_a_tool_in_the_same_turn() -> Result<()> { + let server = start_mock_server().await; + let requests = responses::mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("waiting"), + responses::ev_function_call("pause", "request_user_input", &json!({ + "questions": [{"id": "continue", "header": "Continue", "question": "Continue?", + "options": [{"label": "Yes", "description": "Continue the turn."}, + {"label": "No", "description": "Stop the turn."}]}] + }).to_string()), + ev_completed("waiting"), + ]), + responses::sse_completed("continued"), + ], + ) + .await; + let home = Arc::new(TempDir::new()?); + write_global_file(&home, GLOBAL_AGENTS_FILENAME, GLOBAL_INSTRUCTIONS)?; + let mut builder = test_codex() + .with_home(Arc::clone(&home)) + .with_config(|config| { + config + .features + .enable(Feature::DefaultModeRequestUserInput) + .expect("test config should allow request-user-input feature"); + }); + let test = builder.build_with_auto_env(&server).await?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "ask before continuing".to_string(), + text_elements: Vec::new(), + }])) + .await?; + let EventMsg::RequestUserInput(request) = wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::RequestUserInput(_)) + }) + .await + else { + unreachable!() + }; + write_global_file(&home, GLOBAL_AGENTS_FILENAME, NEW_GLOBAL_INSTRUCTIONS)?; + test.codex + .submit(Op::UserInputAnswer { + id: request.turn_id, + response: RequestUserInputResponse { + answers: HashMap::from([( + "continue".to_string(), + RequestUserInputAnswer { + answers: vec!["Yes".to_string()], + }, + )]), + }, + }) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let initial = expected_provider_only_instruction_fragment(GLOBAL_INSTRUCTIONS); + let replacement = expected_provider_only_instruction_fragment(&format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{NEW_GLOBAL_INSTRUCTIONS}" + )); + assert_eq!( + requests + .requests() + .iter() + .map(instruction_fragments) + .collect::>(), + vec![vec![initial.clone()], vec![initial, replacement]], + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn interrupting_a_provider_read_allows_the_next_turn_to_refresh() -> Result<()> { + struct GatedProvider { + inner: CodexHomeUserInstructionsProvider, + block_next: AtomicBool, + started: Notify, + } + impl UserInstructionsProvider for GatedProvider { + fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_> { + Box::pin(async move { + if self.block_next.swap(/*val*/ false, Ordering::SeqCst) { + self.started.notify_one(); + std::future::pending::<()>().await; + } + self.inner.load_user_instructions().await + }) + } + } + let server = start_mock_server().await; + let request = mount_sse_once(&server, responses::sse_completed("next-turn")).await; + let home = Arc::new(TempDir::new()?); + write_global_file(&home, GLOBAL_AGENTS_FILENAME, GLOBAL_INSTRUCTIONS)?; + let provider = Arc::new(GatedProvider { + inner: CodexHomeUserInstructionsProvider::new(home.path().to_path_buf().abs()), + block_next: AtomicBool::new(/*v*/ false), + started: Notify::new(), + }); + let mut builder = test_codex() + .with_home(Arc::clone(&home)) + .with_user_instructions_provider(provider.clone()); + let test = builder.build_with_auto_env(&server).await?; + provider.block_next.store(/*val*/ true, Ordering::SeqCst); + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "interrupt this blocked read".to_string(), + text_elements: Vec::new(), + }])) + .await?; + tokio::time::timeout(Duration::from_secs(10), provider.started.notified()).await?; + test.codex.submit(Op::Interrupt).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnAborted(_)) + }) + .await; + assert!(request.requests().is_empty()); + write_global_file(&home, GLOBAL_AGENTS_FILENAME, NEW_GLOBAL_INSTRUCTIONS)?; + test.submit_turn("use the latest instructions").await?; + assert_single_instruction_fragment( + &request.single_request(), + &expected_provider_only_instruction_fragment(NEW_GLOBAL_INSTRUCTIONS), + ); + Ok(()) +} diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 082da177f7..033126a16e 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -1841,11 +1841,11 @@ async fn includes_user_instructions_message_in_request() { .with_pre_build_hook(|home| { std::fs::write(home.join("AGENTS.md"), "be nice").expect("write global instructions"); }); - let codex = builder + let test = builder .build(&server) .await - .expect("create new conversation") - .codex; + .expect("create new conversation"); + let codex = test.codex.clone(); codex .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { @@ -2878,11 +2878,11 @@ async fn includes_developer_instructions_message_in_request() { .with_config(|config| { config.developer_instructions = Some("be useful".to_string()); }); - let codex = builder + let test = builder .build(&server) .await - .expect("create new conversation") - .codex; + .expect("create new conversation"); + let codex = test.codex.clone(); codex .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index e01ff2d4ab..4cc257ac2a 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -5127,7 +5127,7 @@ async fn snapshot_request_shape_manual_compact_without_previous_user_messages() } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn manual_compaction_keeps_the_creation_time_global_instructions() -> Result<()> { +async fn manual_compaction_refreshes_global_instructions_for_next_turn() -> Result<()> { // Set up an initial turn, a manual compaction response, and a post-compaction turn. let server = responses::start_mock_server().await; let response_mock = responses::mount_sse_sequence( @@ -5188,25 +5188,25 @@ async fn manual_compaction_keeps_the_creation_time_global_instructions() -> Resu .await; test.submit_turn("after compact").await?; - // Assert ordinary and compact turns keep the old rendering even though the reported source - // path now contains new text. + // Compaction summarizes the existing history; the next turn injects the refreshed instructions. let requests = response_mock.requests(); assert_eq!(requests.len(), 3); - let expected_fragment = expected_instruction_fragment(OLD_GLOBAL_INSTRUCTIONS); - assert_single_instruction_fragment(&requests[0], &expected_fragment); - assert_single_instruction_fragment(&requests[1], &expected_fragment); - assert_single_instruction_fragment(&requests[2], &expected_fragment); + let old_fragment = expected_instruction_fragment(OLD_GLOBAL_INSTRUCTIONS); + let new_fragment = expected_instruction_fragment(NEW_GLOBAL_INSTRUCTIONS); + assert_single_instruction_fragment(&requests[0], &old_fragment); + assert_single_instruction_fragment(&requests[1], &old_fragment); + assert_single_instruction_fragment(&requests[2], &new_fragment); assert_eq!( test.codex.instruction_sources().await, vec![PathUri::from_abs_path(&source)], - "thread retains the creation-time global source after compaction" + "refreshing same-path instructions preserves their source" ); Ok(()) } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn mid_turn_compaction_keeps_the_creation_time_global_instructions() -> Result<()> { +async fn mid_turn_compaction_uses_refreshed_global_instructions() -> Result<()> { // Set up a turn that crosses the auto-compaction limit and a post-compaction response. let server = responses::start_mock_server().await; let response_mock = responses::mount_sse_sequence( @@ -5261,24 +5261,24 @@ async fn mid_turn_compaction_keeps_the_creation_time_global_instructions() -> Re assert_ne!(source, new_source); test.submit_turn("trigger mid-turn compaction").await?; - // Assert the initial, compact, and resumed requests all keep the old snapshot and source. + // The next request boundary selects the override; compaction and continuation retain it. let requests = response_mock.requests(); assert_eq!(requests.len(), 3); - let expected_fragment = expected_instruction_fragment(OLD_GLOBAL_INSTRUCTIONS); + let expected_fragment = expected_instruction_fragment(NEW_GLOBAL_INSTRUCTIONS); assert_single_instruction_fragment(&requests[0], &expected_fragment); assert_single_instruction_fragment(&requests[1], &expected_fragment); assert_single_instruction_fragment(&requests[2], &expected_fragment); assert_eq!( test.codex.instruction_sources().await, - vec![PathUri::from_abs_path(&source)], - "thread retains the creation-time global source after mid-turn compaction" + vec![PathUri::from_abs_path(&new_source)], + "thread reports the refreshed global override after mid-turn compaction" ); Ok(()) } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn remote_v2_compaction_keeps_creation_time_instructions_after_same_path_mutation() +async fn remote_v2_compaction_refreshes_instructions_and_preserves_them_on_cold_resume() -> Result<()> { skip_if_no_network!(Ok(())); @@ -5330,14 +5330,14 @@ async fn remote_v2_compaction_keeps_creation_time_instructions_after_same_path_m test.submit_turn("after remote v2 compaction").await?; test.codex.flush_rollout().await?; - // Assert the compact request, installed replacement history, and follow-up all keep the - // creation-time item despite the file-backed source now containing new text. + // Compaction summarizes the existing history; the follow-up injects the refreshed instructions. let requests = response_mock.requests(); assert_eq!(requests.len(), 3); let old_fragment = expected_instruction_fragment(OLD_GLOBAL_INSTRUCTIONS); + let new_fragment = expected_instruction_fragment(NEW_GLOBAL_INSTRUCTIONS); assert_single_instruction_fragment(&requests[0], &old_fragment); assert_single_instruction_fragment(&requests[1], &old_fragment); - assert_single_instruction_fragment(&requests[2], &old_fragment); + assert_single_instruction_fragment(&requests[2], &new_fragment); assert_eq!( requests[1].input().last(), Some(&json!({"type": "compaction_trigger"})), @@ -5379,17 +5379,10 @@ async fn remote_v2_compaction_keeps_creation_time_instructions_after_same_path_m .submit_turn("after remote v2 compaction cold resume") .await?; - // Cold resume replays the persisted old context, then appends the newly loaded instructions as - // an explicit replacement. + // Cold resume replays the refreshed context without appending unchanged instructions again. let requests = response_mock.requests(); assert_eq!(requests.len(), 4); - let replacement_fragment = expected_instruction_fragment(&format!( - "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{NEW_GLOBAL_INSTRUCTIONS}" - )); - assert_eq!( - instruction_fragments(&requests[3]), - vec![old_fragment.clone(), replacement_fragment] - ); + assert_single_instruction_fragment(&requests[3], &new_fragment); let resumed_input = requests[3].input(); assert_eq!( resumed_input.get(..replacement_history.len()), diff --git a/codex-rs/core/tests/suite/prompt_caching.rs b/codex-rs/core/tests/suite/prompt_caching.rs index b677137b54..ef5ae7050d 100644 --- a/codex-rs/core/tests/suite/prompt_caching.rs +++ b/codex-rs/core/tests/suite/prompt_caching.rs @@ -172,6 +172,8 @@ async fn prompt_tools_are_consistent_across_requests( .await; let TestCodex { + // Keep the file-backed instructions alive across request-boundary refreshes. + home: _home, codex, config, thread_manager, @@ -325,7 +327,9 @@ async fn gpt_5_tools_without_apply_patch_append_apply_patch_instructions() -> an ) .await; - let TestCodex { codex, .. } = test_codex() + let TestCodex { + home: _home, codex, .. + } = test_codex() .with_pre_build_hook(write_global_instructions) .with_config(|config| { config @@ -393,7 +397,12 @@ async fn prefixes_context_and_instructions_once_and_consistently_across_requests ) .await; - let TestCodex { codex, config, .. } = test_codex() + let TestCodex { + home: _home, + codex, + config, + .. + } = test_codex() .with_pre_build_hook(write_global_instructions) .with_config(|config| { config @@ -482,7 +491,12 @@ async fn overrides_turn_context_but_keeps_cached_prefix_and_key_constant() -> an ) .await; - let TestCodex { codex, config, .. } = test_codex() + let TestCodex { + home: _home, + codex, + config, + .. + } = test_codex() .with_pre_build_hook(write_global_instructions) .with_config(|config| { config @@ -750,7 +764,9 @@ async fn per_turn_overrides_keep_cached_prefix_and_key_constant() -> anyhow::Res ) .await; - let TestCodex { codex, .. } = test_codex() + let TestCodex { + home: _home, codex, .. + } = test_codex() .with_pre_build_hook(write_global_instructions) .with_config(|config| { config @@ -878,6 +894,7 @@ async fn send_user_turn_with_no_changes_does_not_send_environment_context() -> a .await; let TestCodex { + home: _home, codex, config, session_configured, @@ -1011,6 +1028,7 @@ async fn send_user_turn_with_changes_sends_environment_context() -> anyhow::Resu ) .await; let TestCodex { + home: _home, codex, config, session_configured, diff --git a/codex-rs/ext/extension-api/src/user_instructions.rs b/codex-rs/ext/extension-api/src/user_instructions.rs index 67b5e07b87..5e262722b3 100644 --- a/codex-rs/ext/extension-api/src/user_instructions.rs +++ b/codex-rs/ext/extension-api/src/user_instructions.rs @@ -23,7 +23,8 @@ pub struct Instructions { pub struct LoadedUserInstructions { /// Loaded instructions, or `None` when the provider has no applicable text. pub instructions: Option, - /// Recoverable loading problems that should be surfaced during startup. + /// Recoverable loading problems that should be surfaced to the host. + /// Providers own suppression of recurring warnings; Core forwards each returned warning. pub warnings: Vec, } @@ -31,11 +32,13 @@ pub struct LoadedUserInstructions { pub type LoadUserInstructionsFuture<'a> = Pin + Send + 'a>>; -/// Loads the user instructions that apply when a root thread runtime starts. +/// Loads global user instructions shared across root threads. /// +/// Core reads this provider at startup and when capturing model-request context. +/// Implementations own fetching and caching, so repeated reads should be cheap. /// Implementations should return any recoverable loading problems as warnings /// while still returning usable fallback instructions when available. pub trait UserInstructionsProvider: Send + Sync { - /// Loads the snapshot to use for a newly created root runtime. + /// Loads the current global snapshot for a root runtime. fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_>; }