From 935ac7710da6d33cf700b3fd953a878f9e1c4535 Mon Sep 17 00:00:00 2001 From: vkg-oai Date: Thu, 10 Sep 2026 23:35:02 +0000 Subject: [PATCH] Refresh global instructions at model-request boundaries (#44675) ## Why Running root threads retained their startup global instructions, so edits to global `AGENTS.md` files did not take effect during an active session. ## What changed - Reload global instructions when capturing model-request context, including after tools within the same turn. Apply changes without repeating unchanged instructions or rediscovering repository instructions when the environment and trust level are unchanged. - Preserve the last successful global instructions on read failures, suppress recurring warnings until recovery, and clear instructions when their source is removed or blank. - Serialize refreshes and allow cancellation without blocking subsequent requests. - Give new subagents the parent's applied instruction snapshot and update Guardian reviewer reuse to account for refreshed instructions. ## Testing Add regression coverage for live edits and removal, read failures and recovery, warning suppression, cancellation, subagent inheritance, and Guardian reviewer reuse. Update resume, fork, and compaction tests to verify refreshed instructions. GitOrigin-RevId: f7e9b399740fa4f45e482c070f6e901ee7cf85cd --- .../suite/v2/guardian_sync_session_tests.rs | 78 +++++ codex-rs/codex-home/src/instructions/mod.rs | 39 ++- codex-rs/codex-home/src/instructions/tests.rs | 62 +++- codex-rs/core/src/agents_md.rs | 9 + codex-rs/core/src/agents_md_manager.rs | 130 ++++++-- codex-rs/core/src/codex_delegate.rs | 8 +- codex-rs/core/src/guardian/review_session.rs | 5 +- .../src/guardian/review_session_factory.rs | 7 +- .../core/src/guardian/review_session_tests.rs | 87 ++++- .../src/guardian/review_session_threads.rs | 2 +- codex-rs/core/src/session/mod.rs | 47 ++- codex-rs/core/src/session/session.rs | 14 +- .../core/src/session/step_activation_tests.rs | 97 ++++++ codex-rs/core/src/session/tests.rs | 11 +- .../core/src/session/tests/guardian_tests.rs | 2 +- codex-rs/core/src/thread_manager.rs | 121 +++---- codex-rs/core/src/thread_manager_tests.rs | 12 +- codex-rs/core/tests/suite/agents_md.rs | 315 ++++++++++++++---- .../core/tests/suite/agents_md_refresh.rs | 254 ++++++++++++++ codex-rs/core/tests/suite/client.rs | 12 +- codex-rs/core/tests/suite/compact.rs | 45 ++- codex-rs/core/tests/suite/prompt_caching.rs | 26 +- .../extension-api/src/user_instructions.rs | 9 +- 23 files changed, 1143 insertions(+), 249 deletions(-) create mode 100644 codex-rs/core/tests/suite/agents_md_refresh.rs 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<'_>; }