From f2f80ef442ff84612004bad064991283eb17cdb5 Mon Sep 17 00:00:00 2001 From: sayan-oai Date: Wed, 24 Jun 2026 22:57:42 -0700 Subject: [PATCH] core: make AGENTS.md react to environment changes (#29810) ## Why With deferred executors, a turn can begin before a remote environment attaches. AGENTS.md discovery previously ran only during session setup, so instructions from a later environment never reached the model or the session instruction sources. WorldState persistence has now landed, so this uses the durable model-visible baseline directly instead of carrying a temporary resume/fork compatibility path. ## What - Add an `AgentsMdManager` in `SessionServices` to own host instructions, loaded state, and refresh caching. - When `DeferredExecutor` is enabled, refresh AGENTS.md when attached environment selections change and freeze the result in the corresponding `StepContext`. - Represent AGENTS.md as a persisted WorldState section for every session, with bounded initial, replacement, and removal updates. - Remove duplicate AGENTS.md state and rendering from `SessionConfiguration` and `TurnContext`. - Build initial context, per-request updates, and compaction context from the same step-scoped value. - On resume and fork, compare current instructions with the restored WorldState baseline and inject a replacement exactly once when they differ. Builds on #29833, #29835, and #29837. ## Tests - Covers a remote environment becoming ready mid-turn, with AGENTS.md appearing on the next request exactly once and updating canonical instruction sources. - Covers full, unchanged, replaced, and removed AGENTS.md WorldState rendering. - Covers changed instructions across cold resume and fork without duplicate reinjection. - Covers remote-v2 compaction retaining creation-time instructions in the live session and cold resume appending one replacement when the source changed. - Ran focused `codex-core` AGENTS.md, WorldState, and context-update test suites. --- codex-rs/core/src/agents_md.rs | 10 +- codex-rs/core/src/agents_md_manager.rs | 53 ++++++ codex-rs/core/src/agents_md_tests.rs | 21 ++- codex-rs/core/src/codex_thread.rs | 8 +- codex-rs/core/src/compact_tests.rs | 1 + codex-rs/core/src/compact_token_budget.rs | 20 +-- .../core/src/context/world_state/agents_md.rs | 75 ++++++++ .../context/world_state/agents_md_tests.rs | 73 ++++++++ codex-rs/core/src/context/world_state/mod.rs | 2 + codex-rs/core/src/guardian/tests.rs | 3 - codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/prompt_debug.rs | 16 +- codex-rs/core/src/session/mod.rs | 85 ++++----- codex-rs/core/src/session/review.rs | 1 - .../session/rollout_reconstruction_tests.rs | 4 +- codex-rs/core/src/session/session.rs | 18 +- codex-rs/core/src/session/step_context.rs | 15 +- codex-rs/core/src/session/tests.rs | 120 ++++++++----- .../core/src/session/tests/guardian_tests.rs | 1 + codex-rs/core/src/session/turn.rs | 17 +- codex-rs/core/src/session/turn_context.rs | 7 - codex-rs/core/src/session/world_state.rs | 44 +++-- codex-rs/core/src/session_startup_prewarm.rs | 1 + codex-rs/core/src/state/service.rs | 2 + codex-rs/core/src/thread_manager_tests.rs | 1 + .../src/thread_rollout_truncation_tests.rs | 2 + codex-rs/core/src/tools/spec_plan_tests.rs | 6 +- codex-rs/core/tests/suite/agents_md.rs | 75 ++++---- codex-rs/core/tests/suite/compact.rs | 12 +- codex-rs/core/tests/suite/remote_env.rs | 164 +++++++++++++++++- 30 files changed, 639 insertions(+), 219 deletions(-) create mode 100644 codex-rs/core/src/agents_md_manager.rs create mode 100644 codex-rs/core/src/context/world_state/agents_md.rs create mode 100644 codex-rs/core/src/context/world_state/agents_md_tests.rs diff --git a/codex-rs/core/src/agents_md.rs b/codex-rs/core/src/agents_md.rs index b2b3cfea2..1c01c8873 100644 --- a/codex-rs/core/src/agents_md.rs +++ b/codex-rs/core/src/agents_md.rs @@ -16,7 +16,6 @@ //! 3. We do **not** walk past the project root. use crate::config::Config; -use crate::context::ContextualUserFragment; use crate::context::UserInstructions as ContextUserInstructions; use crate::environment_selection::TurnEnvironmentSnapshot; use codex_config::ConfigLayerSource; @@ -378,8 +377,7 @@ impl LoadedAgentsMd { output } - /// Returns the complete model-visible contextual user fragment. - pub(crate) fn render(&self) -> String { + pub(crate) fn contextual_user_fragment(&self) -> ContextUserInstructions { // One contributing project environment retains the legacy cwd wrapper. With two or more, // the body labels every contributing environment itself, so the outer cwd is omitted. let directory = if self.has_multiple_project_environments() { @@ -392,12 +390,6 @@ impl LoadedAgentsMd { directory, text: self.text(), } - .render() - } - - /// Returns the host-provided user instructions. - pub(crate) fn user_instructions(&self) -> Option<&UserInstructions> { - self.user_instructions.as_ref() } /// Returns the AGENTS.md files that supplied instruction entries. diff --git a/codex-rs/core/src/agents_md_manager.rs b/codex-rs/core/src/agents_md_manager.rs new file mode 100644 index 000000000..1e3a3f291 --- /dev/null +++ b/codex-rs/core/src/agents_md_manager.rs @@ -0,0 +1,53 @@ +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::UserInstructions; +use codex_protocol::protocol::TurnEnvironmentSelection; +use std::sync::Arc; +use tokio::sync::Mutex; + +/// Owns the inputs and cached result of AGENTS.md discovery for a session. +pub(crate) struct AgentsMdManager { + user_instructions: Option, + cache: Mutex, +} + +#[derive(Default)] +struct AgentsMdCache { + selections: Option>, + loaded: Option>, +} + +impl AgentsMdManager { + pub(crate) fn new(user_instructions: Option) -> Self { + Self { + user_instructions: user_instructions + .filter(|instructions| !instructions.text.trim().is_empty()), + cache: Mutex::new(AgentsMdCache::default()), + } + } + + pub(crate) async fn refresh(&self, config: &Config, environments: &TurnEnvironmentSnapshot) { + let selections = environments.to_selections(); + if self.cache.lock().await.selections.as_ref() == Some(&selections) { + return; + } + + 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.loaded = loaded; + } + + pub(crate) async fn get_loaded(&self) -> Option> { + self.cache.lock().await.loaded.clone() + } + + pub(crate) fn user_instructions(&self) -> Option { + self.user_instructions.clone() + } +} diff --git a/codex-rs/core/src/agents_md_tests.rs b/codex-rs/core/src/agents_md_tests.rs index 6551347aa..d93e4634d 100644 --- a/codex-rs/core/src/agents_md_tests.rs +++ b/codex-rs/core/src/agents_md_tests.rs @@ -1,5 +1,6 @@ use super::*; use crate::config::ConfigBuilder; +use crate::context::ContextualUserFragment; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::session::turn_context::TurnEnvironment; use codex_config::ConfigLayerEntry; @@ -345,7 +346,7 @@ fn foreign_agents_md_uses_environment_native_paths() { }; assert_eq!( - loaded.render(), + loaded.contextual_user_fragment().render(), format!( "# AGENTS.md instructions for {rendered_cwd} @@ -388,7 +389,7 @@ fn multi_environment_agents_md_renders_mixed_path_conventions() { }; assert_eq!( - loaded.render(), + loaded.contextual_user_fragment().render(), r#"# AGENTS.md instructions @@ -931,7 +932,10 @@ secondary doc"#, {inner} "# ); - assert_eq!(loaded.render(), expected_fragment); + assert_eq!( + loaded.contextual_user_fragment().render(), + expected_fragment + ); assert_eq!( loaded.sources().collect::>(), vec![ @@ -972,7 +976,10 @@ async fn secondary_only_project_doc_uses_single_contributor_layout() { "# AGENTS.md instructions for {}\n\n\n{inner}\n", secondary.path().display() ); - assert_eq!(loaded.render(), expected_fragment); + assert_eq!( + loaded.contextual_user_fragment().render(), + expected_fragment + ); } #[tokio::test] @@ -998,7 +1005,10 @@ async fn primary_only_project_doc_preserves_legacy_layout_with_multiple_bound_en "# AGENTS.md instructions for {}\n\n\n{inner}\n", primary.path().display() ); - assert_eq!(loaded.render(), expected_fragment); + assert_eq!( + loaded.contextual_user_fragment().render(), + expected_fragment + ); } #[tokio::test] @@ -1276,7 +1286,6 @@ async fn instruction_sources_include_global_before_agents_md_docs() { }], }; assert_eq!(loaded, expected); - assert_eq!(loaded.user_instructions(), cfg.user_instructions.as_ref()); assert_eq!( loaded.sources().collect::>(), vec![ diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 0cd1ca25b..25f47ce45 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -466,9 +466,15 @@ impl CodexThread { let turn_context = self.codex.session.new_default_turn().await; if self.codex.session.reference_context_item().await.is_none() { + // This history-only API runs without run_turn, so it owns its initial step. + let step_context = self + .codex + .session + .capture_step_context(Arc::clone(&turn_context)) + .await; self.codex .session - .record_context_updates_and_set_reference_context_item(turn_context.as_ref()) + .record_context_updates_and_set_reference_context_item(step_context.as_ref()) .await; } self.codex diff --git a/codex-rs/core/src/compact_tests.rs b/codex-rs/core/src/compact_tests.rs index 19ca75b18..16b9ae16f 100644 --- a/codex-rs/core/src/compact_tests.rs +++ b/codex-rs/core/src/compact_tests.rs @@ -12,6 +12,7 @@ async fn process_compacted_history_with_test_session( previous_turn_settings: Option<&PreviousTurnSettings>, ) -> (Vec, Vec) { let (session, turn_context) = crate::session::tests::make_session_and_context().await; + let turn_context = Arc::new(turn_context); session .set_previous_turn_settings(previous_turn_settings.cloned()) .await; diff --git a/codex-rs/core/src/compact_token_budget.rs b/codex-rs/core/src/compact_token_budget.rs index 0ff7506bc..710e7cefe 100644 --- a/codex-rs/core/src/compact_token_budget.rs +++ b/codex-rs/core/src/compact_token_budget.rs @@ -7,6 +7,7 @@ use crate::hook_runtime::PreCompactHookOutcome; use crate::hook_runtime::run_post_compact_hooks; use crate::hook_runtime::run_pre_compact_hooks; use crate::session::session::Session; +use crate::session::step_context::StepContext; use crate::session::turn_context::TurnContext; use codex_analytics::CompactionTrigger; use codex_protocol::error::CodexErr; @@ -34,10 +35,9 @@ pub(crate) async fn run_manual_compact_task( }); sess.send_event(&turn_context, start_event).await; - let world_state = Arc::new( - sess.build_world_state_for_environments(&turn_context, &turn_context.environments) - .await, - ); + // Manual compaction runs outside run_turn, so it captures its own current step. + let step_context = sess.capture_step_context(Arc::clone(&turn_context)).await; + let world_state = Arc::new(sess.build_world_state_for_step(&step_context).await); run_compact_task_inner(&sess, &turn_context, world_state, CompactionTrigger::Manual).await } @@ -48,17 +48,17 @@ pub(crate) async fn run_manual_compact_task( /// observe the same lifecycle as local or remote compaction. pub(crate) async fn run_inline_auto_compact_task( sess: Arc, - turn_context: Arc, + step_context: Arc, initial_context_injection: InitialContextInjection, ) -> CodexResult<()> { + let turn_context = &step_context.turn; let world_state = match initial_context_injection { InitialContextInjection::BeforeLastUserMessage(world_state) => world_state, - InitialContextInjection::DoNotInject => Arc::new( - sess.build_world_state_for_environments(&turn_context, &turn_context.environments) - .await, - ), + InitialContextInjection::DoNotInject => { + Arc::new(sess.build_world_state_for_step(&step_context).await) + } }; - run_compact_task_inner(&sess, &turn_context, world_state, CompactionTrigger::Auto).await + run_compact_task_inner(&sess, turn_context, world_state, CompactionTrigger::Auto).await } async fn run_compact_task_inner( diff --git a/codex-rs/core/src/context/world_state/agents_md.rs b/codex-rs/core/src/context/world_state/agents_md.rs new file mode 100644 index 000000000..f713b2aff --- /dev/null +++ b/codex-rs/core/src/context/world_state/agents_md.rs @@ -0,0 +1,75 @@ +use super::WorldStateSection; +use crate::agents_md::LoadedAgentsMd; +use crate::context::ContextualUserFragment; +use crate::context::UserInstructions; +use serde::Deserialize; +use serde::Serialize; + +const REPLACEMENT_NOTICE: &str = + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions."; +const REMOVAL_NOTICE: &str = "The previously provided AGENTS.md instructions no longer apply."; + +/// The AGENTS.md instructions currently visible to the model. +#[derive(Clone, Debug, Default)] +pub(crate) struct AgentsMdState { + instructions: Option, +} + +/// Persisted model-visible AGENTS.md state, without filesystem provenance. +#[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq, Serialize)] +pub(crate) struct AgentsMdSnapshot { + directory: Option, + text: Option, +} + +impl AgentsMdState { + pub(crate) fn new(loaded: Option<&LoadedAgentsMd>) -> Self { + Self { + instructions: loaded.map(LoadedAgentsMd::contextual_user_fragment), + } + } +} + +impl WorldStateSection for AgentsMdState { + const ID: &'static str = "agents_md"; + type Snapshot = AgentsMdSnapshot; + + fn snapshot(&self) -> Self::Snapshot { + match &self.instructions { + Some(instructions) => AgentsMdSnapshot { + directory: instructions.directory.clone(), + text: Some(instructions.text.clone()), + }, + None => AgentsMdSnapshot::default(), + } + } + + fn render_diff( + &self, + previous: Option<&Self::Snapshot>, + ) -> Option> { + let current = self.snapshot(); + if previous == Some(¤t) { + return None; + } + + let previous_instructions = previous.and_then(|state| state.text.as_ref()); + let instructions = match (&self.instructions, previous_instructions) { + (Some(instructions), Some(_)) => UserInstructions { + directory: instructions.directory.clone(), + text: format!("{REPLACEMENT_NOTICE}\n\n{}", instructions.text), + }, + (Some(instructions), None) => instructions.clone(), + (None, Some(_)) => UserInstructions { + directory: None, + text: REMOVAL_NOTICE.to_string(), + }, + (None, None) => return None, + }; + Some(Box::new(instructions)) + } +} + +#[cfg(test)] +#[path = "agents_md_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/context/world_state/agents_md_tests.rs b/codex-rs/core/src/context/world_state/agents_md_tests.rs new file mode 100644 index 000000000..f4c171b74 --- /dev/null +++ b/codex-rs/core/src/context/world_state/agents_md_tests.rs @@ -0,0 +1,73 @@ +use super::*; +use crate::context::world_state::WorldState; +use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseItem; +use pretty_assertions::assert_eq; +use serde_json::json; + +#[test] +fn renders_full_state_and_omits_unchanged_state() { + let loaded = LoadedAgentsMd::from_text_for_testing("use the project formatter"); + let mut state = WorldState::default(); + state.add_section(AgentsMdState::new(Some(&loaded))); + + assert_eq!( + vec![user_message( + "# AGENTS.md instructions\n\n\nuse the project formatter\n", + )], + render_fragments(state.render_full()), + ); + assert_eq!( + Vec::::new(), + render_fragments(state.render_diff(&state.snapshot())) + ); + assert_eq!( + state.snapshot().into_value(), + json!({"agents_md": {"text": "use the project formatter"}}), + ); +} + +#[test] +fn changed_and_removed_state_supersedes_previous_instructions() { + let previous_loaded = LoadedAgentsMd::from_text_for_testing("old instructions"); + let mut previous = WorldState::default(); + previous.add_section(AgentsMdState::new(Some(&previous_loaded))); + + let current_loaded = LoadedAgentsMd::from_text_for_testing("new instructions"); + let mut current = WorldState::default(); + current.add_section(AgentsMdState::new(Some(¤t_loaded))); + assert_eq!( + vec![user_message( + "# AGENTS.md instructions\n\n\nThese AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\nnew instructions\n", + )], + render_fragments(current.render_diff(&previous.snapshot())), + ); + + let mut removed = WorldState::default(); + removed.add_section(AgentsMdState::default()); + assert_eq!( + vec![user_message( + "# AGENTS.md instructions\n\n\nThe previously provided AGENTS.md instructions no longer apply.\n", + )], + render_fragments(removed.render_diff(¤t.snapshot())), + ); +} + +fn render_fragments(fragments: Vec>) -> Vec { + fragments + .into_iter() + .map(ContextualUserFragment::into_boxed_response_item) + .collect() +} + +fn user_message(text: &str) -> ResponseItem { + ResponseItem::Message { + id: None, + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: text.to_string(), + }], + phase: None, + internal_chat_message_metadata_passthrough: None, + } +} diff --git a/codex-rs/core/src/context/world_state/mod.rs b/codex-rs/core/src/context/world_state/mod.rs index a6f4e0c5f..541efbe74 100644 --- a/codex-rs/core/src/context/world_state/mod.rs +++ b/codex-rs/core/src/context/world_state/mod.rs @@ -1,3 +1,4 @@ +mod agents_md; mod environment; use crate::context::ContextualUserFragment; @@ -9,6 +10,7 @@ use serde_json::Value; use std::collections::BTreeMap; use std::fmt; +pub(crate) use agents_md::AgentsMdState; pub(crate) use environment::EnvironmentsState; trait ErasedWorldStateSection: Send + Sync { diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index bf6a47889..7dc4d8544 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -248,7 +248,6 @@ async fn guardian_test_session_turn_and_rx( turn_mut.config = Arc::clone(&config); turn_mut.provider = create_model_provider(config.model_provider.clone(), turn_mut.auth_manager.clone()); - turn_mut.user_instructions = None; (session, turn, rx) } @@ -280,7 +279,6 @@ async fn guardian_test_session_and_turn_with_base_url( session.services.models_manager = models_manager; turn.config = Arc::clone(&config); turn.provider = create_model_provider(config.model_provider.clone(), turn.auth_manager.clone()); - turn.user_instructions = None; (Arc::new(session), Arc::new(turn)) } @@ -2305,7 +2303,6 @@ async fn guardian_review_surfaces_responses_api_errors_in_rejection_reason() -> turn_mut.config = Arc::clone(&config); turn_mut.provider = create_model_provider(config.model_provider.clone(), turn_mut.auth_manager.clone()); - turn_mut.user_instructions = None; seed_guardian_parent_history(&session, &turn).await; diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 64148a1d4..ad5f54123 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -130,6 +130,7 @@ pub type NewConversation = NewThread; #[deprecated(note = "use CodexThread")] pub type CodexConversation = CodexThread; pub(crate) mod agents_md; +mod agents_md_manager; pub use agents_md::DEFAULT_AGENTS_MD_FILENAME; pub use agents_md::LOCAL_AGENTS_MD_FILENAME; pub use agents_md::LoadedAgentsMd; diff --git a/codex-rs/core/src/prompt_debug.rs b/codex-rs/core/src/prompt_debug.rs index f5941bc10..a866116a2 100644 --- a/codex-rs/core/src/prompt_debug.rs +++ b/codex-rs/core/src/prompt_debug.rs @@ -3,7 +3,6 @@ use std::sync::Arc; use codex_exec_server::EnvironmentManager; use codex_exec_server::ExecServerRuntimePaths; use codex_extension_api::UserInstructionsProvider; -use codex_features::Feature; use codex_login::AuthManager; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; @@ -78,8 +77,9 @@ pub(crate) async fn build_prompt_input_from_session( input: Vec, ) -> CodexResult> { let turn_context = sess.new_default_turn().await; - let world_state = sess - .record_context_updates_and_set_reference_context_item(turn_context.as_ref()) + // Prompt debugging builds a standalone request without entering run_turn. + let step_context = sess.capture_step_context(Arc::clone(&turn_context)).await; + sess.record_context_updates_and_set_reference_context_item(step_context.as_ref()) .await; if !input.is_empty() { @@ -88,16 +88,6 @@ pub(crate) async fn build_prompt_input_from_session( .await; } - let step_context = sess.capture_step_context(Arc::clone(&turn_context)).await; - if turn_context - .config - .features - .enabled(Feature::DeferredExecutor) - { - sess.record_step_environment_context_if_changed(&world_state, step_context.as_ref()) - .await; - } - let prompt_input = sess .clone_history() .await diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 52ce20c90..be12e61e0 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -14,7 +14,6 @@ use crate::agent::AgentControl; use crate::agent::AgentStatus; use crate::agent::agent_status_from_event; use crate::agent::status::is_final; -use crate::agents_md::LoadedAgentsMd; use crate::attestation::AttestationProvider; use crate::build_available_skills; use crate::compact; @@ -248,7 +247,6 @@ use self::turn::collect_explicit_app_ids_from_skill_items; use self::turn::realtime_text_for_event; use self::turn_context::TurnContext; use self::turn_context::TurnSkillsContext; -use self::world_state::build_world_state_from_environment_snapshot; #[cfg(test)] mod rollout_reconstruction_tests; @@ -308,7 +306,6 @@ pub(crate) struct PreviousTurnSettings { #[cfg(test)] use crate::SkillMetadata; use crate::SkillsService; -use crate::agents_md::load_project_instructions; use crate::exec_policy::ExecPolicyUpdateError; use crate::guardian::GuardianReviewSessionManager; use crate::mcp::McpManager; @@ -627,7 +624,6 @@ impl Codex { model_reasoning_summary: config.model_reasoning_summary, service_tier, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md: None, personality: config.personality, base_instructions, compact_prompt: config.compact_prompt.clone(), @@ -842,10 +838,11 @@ impl Codex { } pub(crate) async fn instruction_sources(&self) -> Vec { - let state = self.session.state.lock().await; - state - .session_configuration - .loaded_agents_md + self.session + .services + .agents_md_manager + .get_loaded() + .await .as_ref() .map_or_else(Vec::new, |instructions| instructions.sources().collect()) } @@ -1542,13 +1539,7 @@ impl Session { } pub(crate) async fn user_instructions(&self) -> Option { - let state = self.state.lock().await; - state - .session_configuration - .loaded_agents_md - .as_ref() - .and_then(LoadedAgentsMd::user_instructions) - .cloned() + self.services.agents_md_manager.user_instructions() } pub(crate) async fn provider(&self) -> ModelProviderInfo { @@ -2776,17 +2767,14 @@ impl Session { self.send_raw_response_items(turn_context, items).await; } - pub(crate) async fn record_step_environment_context_if_changed( + pub(crate) async fn record_step_world_state_if_changed( &self, previous_world_state: &Arc, step_context: &step_context::StepContext, ) -> Arc { let turn_context = step_context.turn.as_ref(); // Render model-visible state from the same step used to build and run tools. - let world_state = Arc::new( - self.build_world_state_for_environments(turn_context, &step_context.environments) - .await, - ); + let world_state = Arc::new(self.build_world_state_for_step(step_context).await); // Derive the model update and persisted patch from the same two snapshots. let previous_snapshot = previous_world_state.snapshot(); let world_state_snapshot = world_state.snapshot(); @@ -2814,21 +2802,37 @@ impl Session { world_state } + /// Captures one request-scoped view of dynamic state. + /// + /// This may refresh filesystem-derived state. Normal turns should call it only from + /// `run_turn` and pass the result down; standalone request or history boundaries may capture + /// their own step. pub(crate) async fn capture_step_context( &self, turn_context: Arc, ) -> Arc { - // Keep the old turn-frozen view unless deferred executors are explicitly enabled. - let environments = if turn_context + let deferred_executor_enabled = turn_context .config .features - .enabled(Feature::DeferredExecutor) - { + .enabled(Feature::DeferredExecutor); + // Keep the old turn-frozen environment view unless deferred executors are enabled. + let environments = if deferred_executor_enabled { self.services.turn_environments.snapshot().await } else { turn_context.environments.clone() }; - Arc::new(StepContext::new(turn_context, environments)) + if deferred_executor_enabled { + self.services + .agents_md_manager + .refresh(&turn_context.config, &environments) + .await; + } + let loaded_agents_md = self.services.agents_md_manager.get_loaded().await; + Arc::new(StepContext::new( + turn_context, + environments, + loaded_agents_md, + )) } pub(crate) async fn record_inter_agent_communication( @@ -3088,26 +3092,6 @@ impl Session { items } - pub(crate) async fn build_world_state_for_environments( - &self, - turn_context: &TurnContext, - environments: &TurnEnvironmentSnapshot, - ) -> WorldState { - let environment_subagents = if turn_context.config.include_environment_context { - self.services - .agent_control - .format_environment_context_subagents(self.thread_id) - .await - } else { - String::new() - }; - build_world_state_from_environment_snapshot( - turn_context, - environments, - &environment_subagents, - ) - } - pub(crate) async fn build_initial_context_with_world_state( &self, turn_context: &TurnContext, @@ -3311,9 +3295,6 @@ impl Session { ); } } - if let Some(user_instructions) = turn_context.user_instructions.as_deref() { - contextual_user_sections.push(user_instructions.to_string()); - } // This is full-context metadata. Steady-state context diffs should not re-emit it. if turn_context.config.features.enabled(Feature::TokenBudget) && turn_context.model_context_window().is_some() @@ -3519,8 +3500,9 @@ impl Session { #[instrument(level = "trace", skip_all)] pub(crate) async fn record_context_updates_and_set_reference_context_item( &self, - turn_context: &TurnContext, + step_context: &StepContext, ) -> Arc { + let turn_context = step_context.turn.as_ref(); let reference_context_item = { let state = self.state.lock().await; state.reference_context_item() @@ -3528,10 +3510,7 @@ impl Session { let turn_context_item = turn_context.to_turn_context_item(); let turn_context_changed = reference_context_item.as_ref() != Some(&turn_context_item); let should_inject_full_context = reference_context_item.is_none(); - let world_state = Arc::new( - self.build_world_state_for_environments(turn_context, &turn_context.environments) - .await, - ); + let world_state = Arc::new(self.build_world_state_for_step(step_context).await); // Full initial context resets the baseline; later turns persist only its changes. let (mut context_items, world_state_item) = if should_inject_full_context { let context_items = self diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index f2482a3ce..1587290d3 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -126,7 +126,6 @@ pub(super) async fn spawn_review_thread( timezone: parent_turn_context.timezone.clone(), app_server_client_name: parent_turn_context.app_server_client_name.clone(), developer_instructions: None, - user_instructions: None, collaboration_mode: parent_turn_context.collaboration_mode.clone(), multi_agent_version: MultiAgentVersion::Disabled, personality: parent_turn_context.personality, diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index 7c53c93fc..a7d2a7efc 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -135,6 +135,7 @@ async fn record_initial_history_reconstructs_typed_inter_agent_message() { #[tokio::test] async fn record_initial_history_restores_world_state_baseline() { let (session, turn_context) = make_session_and_context().await; + let turn_context = Arc::new(turn_context); let world_state = build_world_state_from_turn_context(&session, &turn_context).await; let rollout_items = completed_user_turn_rollout( turn_context.to_turn_context_item(), @@ -150,8 +151,9 @@ async fn record_initial_history_restores_world_state_baseline() { rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")), })) .await; + let step_context = StepContext::for_test(Arc::clone(&turn_context)); session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; assert_eq!(session.clone_history().await.raw_items(), &[]); diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 78abe9df8..5988834ee 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1,6 +1,6 @@ use super::input_queue::InputQueue; use super::*; -use crate::agents_md::LoadedAgentsMd; +use crate::agents_md_manager::AgentsMdManager; use crate::config::ConstraintError; use crate::environment_selection::ThreadEnvironments; use crate::environment_selection::TurnEnvironmentSnapshot; @@ -16,7 +16,6 @@ use codex_protocol::permissions::FileSystemPath; use codex_protocol::permissions::FileSystemSpecialPath; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::ThreadSource; -use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::protocol::TurnEnvironmentSelections; use std::sync::OnceLock; use tokio::sync::Semaphore; @@ -59,10 +58,6 @@ pub(crate) struct SessionConfiguration { /// Developer instructions that supplement the base instructions. pub(super) developer_instructions: Option, - /// Model instructions assembled from provider instructions and discovered - /// AGENTS.md files. - pub(super) loaded_agents_md: Option, - /// Personality preference for the model. pub(super) personality: Option, @@ -863,12 +858,10 @@ impl Session { )); turn_environments.update_selections(session_configuration.environment_selections()); let resolved_environments = turn_environments.snapshot().await; - session_configuration.loaded_agents_md = load_project_instructions( - config.as_ref(), - user_instructions, - &resolved_environments, - ) - .await; + let agents_md_manager = Arc::new(AgentsMdManager::new(user_instructions)); + agents_md_manager + .refresh(config.as_ref(), &resolved_environments) + .await; let plugin_skill_errors = warm_plugins_and_skills_for_session_init( Arc::clone(&config), Arc::clone(&plugins_manager), @@ -1036,6 +1029,7 @@ impl Session { guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), skills_service, + agents_md_manager, plugins_manager: Arc::clone(&plugins_manager), mcp_manager: Arc::clone(&mcp_manager), extensions, diff --git a/codex-rs/core/src/session/step_context.rs b/codex-rs/core/src/session/step_context.rs index b246968a0..bf672db6e 100644 --- a/codex-rs/core/src/session/step_context.rs +++ b/codex-rs/core/src/session/step_context.rs @@ -1,5 +1,6 @@ use std::sync::Arc; +use crate::agents_md::LoadedAgentsMd; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::session::turn_context::TurnContext; @@ -8,10 +9,20 @@ use crate::session::turn_context::TurnContext; pub(crate) struct StepContext { pub(crate) turn: Arc, pub(crate) environments: TurnEnvironmentSnapshot, + /// The canonical AGENTS.md value observed with this environment snapshot. + pub(crate) loaded_agents_md: Option>, } impl StepContext { - pub(crate) fn new(turn: Arc, environments: TurnEnvironmentSnapshot) -> Self { - Self { turn, environments } + pub(crate) fn new( + turn: Arc, + environments: TurnEnvironmentSnapshot, + loaded_agents_md: Option>, + ) -> Self { + Self { + turn, + environments, + loaded_agents_md, + } } } diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 1030a64e6..4553d54a0 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -1,5 +1,6 @@ use super::turn_context::TurnEnvironment; use super::*; +use crate::agents_md_manager::AgentsMdManager; use crate::codex_thread::TryStartTurnIfIdleRejectionReason; use crate::config::ConfigBuilder; use crate::config::ConfigOverrides; @@ -186,7 +187,11 @@ use std::time::Duration as StdDuration; impl StepContext { pub(crate) fn for_test(turn: Arc) -> Arc { let environments = turn.environments.clone(); - Arc::new(Self::new(turn, environments)) + Arc::new(Self::new( + turn, + environments, + /*loaded_agents_md*/ None, + )) } } @@ -2082,6 +2087,7 @@ fn session_meta_item( #[tokio::test] async fn resumed_history_injects_initial_context_on_first_context_update_only() { let (session, turn_context) = make_session_and_context().await; + let turn_context = Arc::new(turn_context); let (rollout_items, mut expected) = sample_rollout(&session, &turn_context).await; session @@ -2095,8 +2101,9 @@ async fn resumed_history_injects_initial_context_on_first_context_update_only() let history_before_seed = session.state.lock().await.clone_history(); assert_eq!(expected, history_before_seed.raw_items()); + let step_context = StepContext::for_test(Arc::clone(&turn_context)); session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; let initial_context = build_initial_context(&session, &turn_context).await; expected.extend(initial_context); @@ -2104,7 +2111,7 @@ async fn resumed_history_injects_initial_context_on_first_context_update_only() assert_eq!(expected, history_after_seed.raw_items()); session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; let history_after_second_seed = session.clone_history().await; assert_eq!( @@ -2715,9 +2722,8 @@ async fn start_new_context_window_assigns_and_persists_item_ids() { .await; let rollout_path = attach_thread_persistence(Arc::get_mut(&mut session).expect("unique session")).await; - let world_state = Arc::new( - build_world_state_from_turn_context(session.as_ref(), turn_context.as_ref()).await, - ); + let world_state = + Arc::new(build_world_state_from_turn_context(session.as_ref(), &turn_context).await); session .start_new_context_window(turn_context.as_ref(), world_state) @@ -3091,7 +3097,7 @@ async fn thread_rollback_drops_last_turn_from_history() { ) .await; - let initial_context = build_initial_context(&sess, tc.as_ref()).await; + let initial_context = build_initial_context(&sess, &tc).await; let turn_1 = vec![ user_message("turn 1 user"), assistant_message("turn 1 assistant"), @@ -3159,7 +3165,7 @@ async fn thread_rollback_clears_history_when_num_turns_exceeds_existing_turns() ) .await; - let initial_context = build_initial_context(&sess, tc.as_ref()).await; + let initial_context = build_initial_context(&sess, &tc).await; let turn_1 = vec![user_message("turn 1 user")]; let mut full_history = Vec::new(); full_history.extend(initial_context.clone()); @@ -3185,7 +3191,7 @@ async fn thread_rollback_clears_history_when_num_turns_exceeds_existing_turns() async fn thread_rollback_fails_without_persisted_thread_history() { let (sess, tc, rx) = make_session_and_context_with_rx().await; - let initial_context = build_initial_context(&sess, tc.as_ref()).await; + let initial_context = build_initial_context(&sess, &tc).await; sess.record_conversation_items(tc.as_ref(), &initial_context) .await; @@ -3595,7 +3601,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() { async fn thread_rollback_fails_when_turn_in_progress() { let (sess, tc, rx) = make_session_and_context_with_rx().await; - let initial_context = build_initial_context(&sess, tc.as_ref()).await; + let initial_context = build_initial_context(&sess, &tc).await; sess.record_conversation_items(tc.as_ref(), &initial_context) .await; @@ -3616,7 +3622,7 @@ async fn thread_rollback_fails_when_turn_in_progress() { async fn thread_rollback_fails_when_num_turns_is_zero() { let (sess, tc, rx) = make_session_and_context_with_rx().await; - let initial_context = build_initial_context(&sess, tc.as_ref()).await; + let initial_context = build_initial_context(&sess, &tc).await; sess.record_conversation_items(tc.as_ref(), &initial_context) .await; @@ -3655,7 +3661,6 @@ async fn set_rate_limits_retains_previous_credits() { collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config @@ -3762,7 +3767,6 @@ async fn set_rate_limits_updates_plan_type_when_present() { collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config @@ -4293,7 +4297,6 @@ pub(crate) async fn make_session_configuration_for_tests() -> SessionConfigurati collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config @@ -5164,7 +5167,6 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config @@ -5243,20 +5245,22 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { assert!(msg.contains("zsh fork feature enabled, but no packaged zsh fork is available")); } -async fn build_initial_context(session: &Session, turn_context: &TurnContext) -> Vec { +async fn build_initial_context( + session: &Session, + turn_context: &Arc, +) -> Vec { let world_state = build_world_state_from_turn_context(session, turn_context).await; session - .build_initial_context_with_world_state(turn_context, &world_state) + .build_initial_context_with_world_state(turn_context.as_ref(), &world_state) .await } pub(crate) async fn build_world_state_from_turn_context( session: &Session, - turn_context: &TurnContext, + turn_context: &Arc, ) -> WorldState { - session - .build_world_state_for_environments(turn_context, &turn_context.environments) - .await + let step_context = StepContext::for_test(Arc::clone(turn_context)); + session.build_world_state_for_step(&step_context).await } // todo: use online model info @@ -5293,7 +5297,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config @@ -5393,6 +5396,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), skills_service, + agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)), plugins_manager, mcp_manager, extensions: Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), @@ -5471,7 +5475,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { "turn_id".to_string(), skills_snapshot, ); - let session = Session { thread_id, installation_id: "11111111-1111-4111-8111-111111111111".to_string(), @@ -5541,7 +5544,6 @@ async fn make_session_with_config_and_rx( collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config @@ -5648,7 +5650,6 @@ async fn make_session_with_history_source_and_agent_control_and_rx( collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config @@ -7372,7 +7373,6 @@ where collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config @@ -7471,6 +7471,7 @@ where guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), skills_service, + agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)), plugins_manager, mcp_manager, extensions: Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), @@ -7549,7 +7550,6 @@ where "turn_id".to_string(), skills_snapshot, )); - let session = Arc::new(Session { thread_id, installation_id: "11111111-1111-4111-8111-111111111111".to_string(), @@ -7718,7 +7718,7 @@ async fn record_context_updates_emits_environment_item_for_network_changes() { current_context.config = Arc::new(config); let update_items = - record_context_update_items(&session, &previous_context, ¤t_context).await; + record_context_update_items(&session, previous_context, current_context).await; let environment_update = user_input_texts(&update_items) .into_iter() @@ -7749,7 +7749,7 @@ async fn record_context_updates_emits_environment_item_for_cwd_changes() { ); let update_items = - record_context_update_items(&session, &previous_context, ¤t_context).await; + record_context_update_items(&session, previous_context, current_context).await; let environment_update = user_input_texts(&update_items) .into_iter() @@ -7776,7 +7776,7 @@ async fn record_context_updates_emits_environment_item_for_time_changes() { current_context.timezone = Some("Europe/Berlin".to_string()); let update_items = - record_context_update_items(&session, &previous_context, ¤t_context).await; + record_context_update_items(&session, previous_context, current_context).await; let environment_update = user_input_texts(&update_items) .into_iter() @@ -7808,7 +7808,7 @@ async fn record_context_updates_omits_environment_item_when_disabled() { ); let update_items = - record_context_update_items(&session, &previous_context, ¤t_context).await; + record_context_update_items(&session, previous_context, current_context).await; let user_texts = user_input_texts(&update_items); assert!( @@ -7821,16 +7821,18 @@ async fn record_context_updates_omits_environment_item_when_disabled() { async fn record_context_update_items( session: &Session, - previous_context: &TurnContext, - current_context: &TurnContext, + previous_context: Arc, + current_context: TurnContext, ) -> Vec { + let previous_step = StepContext::for_test(previous_context); session - .record_context_updates_and_set_reference_context_item(previous_context) + .record_context_updates_and_set_reference_context_item(&previous_step) .await; let previous_len = session.clone_history().await.raw_items().len(); + let current_step = StepContext::for_test(Arc::new(current_context)); session - .record_context_updates_and_set_reference_context_item(current_context) + .record_context_updates_and_set_reference_context_item(¤t_step) .await; let history = session.clone_history().await; history.raw_items()[previous_len..].to_vec() @@ -7930,6 +7932,7 @@ async fn build_settings_update_items_uses_previous_turn_settings_for_realtime_en async fn build_initial_context_uses_previous_realtime_state() { let (session, mut turn_context) = make_session_and_context().await; turn_context.realtime_active = true; + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let developer_texts = developer_input_texts(&initial_context); @@ -8043,6 +8046,7 @@ async fn build_initial_context_includes_prompt_fragments_from_extensions() { .services .thread_extension_data .insert(PromptExtensionTestState); + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let developer_messages = developer_message_texts(&initial_context); @@ -8069,6 +8073,7 @@ async fn build_initial_context_includes_turn_context_fragments_from_extensions() .insert(TurnContextExtensionTestState { expected_model_context_window: Some(50), }); + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let developer_messages = developer_message_texts(&initial_context); @@ -8097,6 +8102,7 @@ async fn record_context_updates_includes_turn_context_fragments_on_steady_state_ }); let mut previous_context_item = turn_context.to_turn_context_item(); previous_context_item.turn_id = Some("previous-turn-id".to_string()); + let turn_context = Arc::new(turn_context); let world_state = build_world_state_from_turn_context(&session, &turn_context).await; { let mut state = session.state.lock().await; @@ -8106,8 +8112,9 @@ async fn record_context_updates_includes_turn_context_fragments_on_steady_state_ .set_world_state_baseline(world_state.snapshot()); } + let step_context = StepContext::for_test(Arc::clone(&turn_context)); session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; let history = session.clone_history().await; @@ -8125,6 +8132,7 @@ async fn record_context_updates_includes_turn_context_fragments_on_steady_state_ async fn build_initial_context_omits_prompt_fragments_without_extension_state() { let (mut session, turn_context) = make_session_and_context().await; session.services.extensions = prompt_extension_test_registry(); + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let developer_messages = developer_message_texts(&initial_context); @@ -8143,7 +8151,7 @@ async fn build_initial_context_adds_multi_agent_v2_root_usage_hint_as_developer_ let (session, turn_context) = make_multi_agent_v2_usage_hint_test_session(/*enable_multi_agent_v2*/ true).await; - let initial_context = build_initial_context(&session, turn_context.as_ref()).await; + let initial_context = build_initial_context(&session, &turn_context).await; let developer_messages = developer_message_texts(&initial_context); assert!( @@ -8181,7 +8189,7 @@ async fn build_initial_context_adds_multi_agent_v2_subagent_usage_hint_as_develo .expect("thread settings should not be shared") .session_source = session_source; - let initial_context = build_initial_context(&session, turn_context.as_ref()).await; + let initial_context = build_initial_context(&session, &turn_context).await; let developer_messages = developer_message_texts(&initial_context); assert!( @@ -8203,7 +8211,7 @@ async fn build_initial_context_omits_multi_agent_v2_usage_hints_when_feature_dis let (session, turn_context) = make_multi_agent_v2_usage_hint_test_session(/*enable_multi_agent_v2*/ false).await; - let initial_context = build_initial_context(&session, turn_context.as_ref()).await; + let initial_context = build_initial_context(&session, &turn_context).await; let developer_messages = developer_message_texts(&initial_context); assert!( @@ -8230,7 +8238,7 @@ async fn build_initial_context_omits_multi_agent_v2_usage_hints_when_hint_is_emp ) .await; - let initial_context = build_initial_context(&session, turn_context.as_ref()).await; + let initial_context = build_initial_context(&session, &turn_context).await; let developer_messages = developer_message_texts(&initial_context); assert!( @@ -8259,6 +8267,7 @@ async fn build_initial_context_omits_default_image_save_location_with_image_hist /*reference_context_item*/ None, ) .await; + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let developer_texts = developer_input_texts(&initial_context); @@ -8273,6 +8282,7 @@ async fn build_initial_context_omits_default_image_save_location_with_image_hist #[tokio::test] async fn build_initial_context_omits_default_image_save_location_without_image_history() { let (session, turn_context) = make_session_and_context().await; + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let developer_texts = developer_input_texts(&initial_context); @@ -8315,6 +8325,7 @@ async fn build_initial_context_trims_skill_metadata_from_context_window_budget() ]; turn_context.model_info.context_window = Some(100); turn_context.turn_skills = TurnSkillsContext::new(HostSkillsSnapshot::new(Arc::new(outcome))); + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let developer_texts = developer_input_texts(&initial_context); @@ -8466,6 +8477,7 @@ async fn build_initial_context_emits_thread_start_skill_warning_on_repeated_buil ]; turn_context.model_info.context_window = Some(100); turn_context.turn_skills = TurnSkillsContext::new(HostSkillsSnapshot::new(Arc::new(outcome))); + let turn_context = Arc::new(turn_context); let _ = build_initial_context(&session, &turn_context).await; let warning_event = timeout(Duration::from_secs(1), rx.recv()) @@ -8598,6 +8610,7 @@ async fn build_initial_context_uses_previous_turn_settings_for_realtime_end() { session .set_previous_turn_settings(Some(previous_turn_settings)) .await; + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let developer_texts = developer_input_texts(&initial_context); assert!( @@ -8621,6 +8634,7 @@ async fn build_initial_context_restates_realtime_start_when_reference_context_is session .set_previous_turn_settings(Some(previous_turn_settings)) .await; + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let developer_texts = developer_input_texts(&initial_context); assert!( @@ -8704,8 +8718,10 @@ async fn turn_context_item_stores_split_file_system_sandbox_policy_when_differen async fn record_context_updates_and_set_reference_context_item_injects_full_context_when_baseline_missing() { let (session, turn_context) = make_session_and_context().await; + let turn_context = Arc::new(turn_context); + let step_context = StepContext::for_test(Arc::clone(&turn_context)); session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; let history = session.clone_history().await; let initial_context = build_initial_context(&session, &turn_context).await; @@ -8723,6 +8739,8 @@ async fn record_context_updates_and_set_reference_context_item_injects_full_cont async fn record_context_updates_and_set_reference_context_item_reinjects_full_context_after_clear() { let (session, turn_context) = make_session_and_context().await; + let turn_context = Arc::new(turn_context); + let step_context = StepContext::for_test(Arc::clone(&turn_context)); let compacted_summary = ResponseItem::Message { id: None, role: "user".to_string(), @@ -8736,7 +8754,7 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co .record_conversation_items(&turn_context, std::slice::from_ref(&compacted_summary)) .await; session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; { let mut state = session.state.lock().await; @@ -8750,7 +8768,7 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co .await; session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; let history = session.clone_history().await; @@ -8773,6 +8791,7 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline .with_model(next_model.to_string(), &session.services.models_manager) .await; let previous_context_item = previous_context.to_turn_context_item(); + let previous_context = Arc::new(previous_context); let world_state = build_world_state_from_turn_context(&session, &previous_context).await; { let mut state = session.state.lock().await; @@ -8788,8 +8807,10 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline .await; assert_eq!(update_items, Vec::new()); + let turn_context = Arc::new(turn_context); + let step_context = StepContext::for_test(Arc::clone(&turn_context)); session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; assert_eq!( @@ -8835,8 +8856,10 @@ async fn record_context_updates_and_set_reference_context_item_persists_split_fi ); let rollout_path = attach_thread_persistence(&mut session).await; + let turn_context = Arc::new(turn_context); + let step_context = StepContext::for_test(Arc::clone(&turn_context)); session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; session.ensure_rollout_materialized().await; session.flush_rollout().await.expect("rollout should flush"); @@ -8869,6 +8892,7 @@ async fn build_initial_context_prepends_model_switch_message() { session .set_previous_turn_settings(Some(previous_turn_settings)) .await; + let turn_context = Arc::new(turn_context); let initial_context = build_initial_context(&session, &turn_context).await; let ResponseItem::Message { role, content, .. } = &initial_context[0] else { @@ -8919,8 +8943,10 @@ async fn record_context_updates_and_set_reference_context_item_persists_full_rei realtime_active: Some(previous_context.realtime_active), })) .await; + let turn_context = Arc::new(turn_context); + let step_context = StepContext::for_test(Arc::clone(&turn_context)); session - .record_context_updates_and_set_reference_context_item(&turn_context) + .record_context_updates_and_set_reference_context_item(&step_context) .await; session.ensure_rollout_materialized().await; session.flush_rollout().await.expect("rollout should flush"); @@ -10195,7 +10221,7 @@ async fn sample_rollout( // Use the same turn_context source as record_initial_history so model_info (and thus // personality_spec) matches reconstruction. let reconstruction_turn = session.new_default_turn().await; - let mut initial_context = build_initial_context(session, reconstruction_turn.as_ref()).await; + let mut initial_context = build_initial_context(session, &reconstruction_turn).await; // Ensure personality_spec is present when Personality is enabled, so expected matches // what reconstruction produces (build_initial_context may omit it when baked into model). if !initial_context.iter().any(|m| { diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 38a7ec039..64c4ba29f 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -530,6 +530,7 @@ async fn process_compacted_history_preserves_separate_guardian_developer_message } turn_context.session_source = guardian_source; turn_context.developer_instructions = Some(guardian_policy.clone()); + let turn_context = Arc::new(turn_context); let world_state = Arc::new(build_world_state_from_turn_context(&session, &turn_context).await); let initial_context_injection = InitialContextInjection::BeforeLastUserMessage(world_state); diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index a8fe50b78..9b3588fc1 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -164,9 +164,11 @@ pub(crate) async fn run_turn( return Ok(None); } + // run_turn owns the step used to seed context and make the first sampling request. + let first_step_context = sess.capture_step_context(Arc::clone(&turn_context)).await; // Keep the exact model-visible state used by this turn and its inline compactions. let mut world_state = sess - .record_context_updates_and_set_reference_context_item(turn_context.as_ref()) + .record_context_updates_and_set_reference_context_item(first_step_context.as_ref()) .await; let Some((injection_items, explicitly_enabled_connectors)) = @@ -214,6 +216,7 @@ pub(crate) async fn run_turn( // 1. At the start of a turn, so the fresh turn input in `input` gets sampled first. // 2. After auto-compact, when model/tool continuation needs to resume before any steer. + let mut next_step_context = Some(first_step_context); loop { // Note that pending_input would be something like a message the user // submitted through the UI while the model was running. Though the UI @@ -237,7 +240,10 @@ pub(crate) async fn run_turn( .await; // Capture once so context, advertised tools, and tool calls share one request view. - let step_context = sess.capture_step_context(Arc::clone(&turn_context)).await; + let step_context = match next_step_context.take() { + Some(step_context) => step_context, + None => sess.capture_step_context(Arc::clone(&turn_context)).await, + }; let sampling_request_result: CodexResult<_> = async { super::time_reminder::maybe_record_current_time_reminder( sess.as_ref(), @@ -252,7 +258,7 @@ pub(crate) async fn run_turn( .enabled(Feature::DeferredExecutor) { world_state = sess - .record_step_environment_context_if_changed(&world_state, step_context.as_ref()) + .record_step_world_state_if_changed(&world_state, step_context.as_ref()) .await; } @@ -796,6 +802,7 @@ async fn run_pre_sampling_compact( .await; // Compact if the configured auto-compaction budget or usable context window is exhausted. if token_status.token_limit_reached { + // Pre-turn compaction runs before run_turn creates the normal sampling step. let step_context = sess.capture_step_context(Arc::clone(turn_context)).await; run_auto_compact( sess, @@ -841,6 +848,7 @@ async fn maybe_run_previous_model_inline_compact( ); if should_compact_for_comp_hash_change { + // This pre-turn request needs a step built from the previous model's turn context. let step_context = sess .capture_step_context(Arc::clone(&previous_model_turn_context)) .await; @@ -881,6 +889,7 @@ async fn maybe_run_previous_model_inline_compact( && previous_model_turn_context.model_info.slug != turn_context.model_info.slug && old_context_window > new_context_window; if should_run { + // This pre-turn request needs a step built from the previous model's turn context. let step_context = sess .capture_step_context(Arc::clone(&previous_model_turn_context)) .await; @@ -916,7 +925,7 @@ async fn run_auto_compact( // instead of consuming a pending `new_context` tool request. crate::compact_token_budget::run_inline_auto_compact_task( Arc::clone(sess), - Arc::clone(turn_context), + step_context, initial_context_injection, ) .await?; diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 9799f68cc..a5ae45895 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -1,5 +1,4 @@ use super::*; -use crate::agents_md::LoadedAgentsMd; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::shell_snapshot::ShellSnapshotFile; use codex_core_skills::HostSkillsSnapshot; @@ -124,7 +123,6 @@ pub struct TurnContext { pub(crate) timezone: Option, pub(crate) app_server_client_name: Option, pub(crate) developer_instructions: Option, - pub(crate) user_instructions: Option, pub(crate) collaboration_mode: CollaborationMode, pub(crate) multi_agent_version: MultiAgentVersion, pub(crate) personality: Option, @@ -273,7 +271,6 @@ impl TurnContext { timezone: self.timezone.clone(), app_server_client_name: self.app_server_client_name.clone(), developer_instructions: self.developer_instructions.clone(), - user_instructions: self.user_instructions.clone(), collaboration_mode, multi_agent_version: self.multi_agent_version, personality: self.personality, @@ -552,10 +549,6 @@ impl Session { timezone: Some(timezone), app_server_client_name: session_configuration.app_server_client_name.clone(), developer_instructions: session_configuration.developer_instructions.clone(), - user_instructions: session_configuration - .loaded_agents_md - .as_ref() - .map(LoadedAgentsMd::render), collaboration_mode: session_configuration.collaboration_mode.clone(), multi_agent_version, personality: session_configuration.personality, diff --git a/codex-rs/core/src/session/world_state.rs b/codex-rs/core/src/session/world_state.rs index c20f3df64..e19e8c2ff 100644 --- a/codex-rs/core/src/session/world_state.rs +++ b/codex-rs/core/src/session/world_state.rs @@ -1,19 +1,35 @@ -use super::turn_context::TurnContext; +use super::session::Session; +use super::step_context::StepContext; +use crate::context::world_state::AgentsMdState; use crate::context::world_state::EnvironmentsState; use crate::context::world_state::WorldState; -use crate::environment_selection::TurnEnvironmentSnapshot; -pub(super) fn build_world_state_from_environment_snapshot( - turn_context: &TurnContext, - environments: &TurnEnvironmentSnapshot, - environment_subagents: &str, -) -> WorldState { - let mut world_state = WorldState::default(); - if turn_context.config.include_environment_context { - world_state.add_section( - EnvironmentsState::from_turn_context_with_environments(turn_context, environments) - .with_subagents(environment_subagents.to_string()), - ); +impl Session { + pub(crate) async fn build_world_state_for_step( + &self, + step_context: &StepContext, + ) -> WorldState { + let turn_context = step_context.turn.as_ref(); + let environment_subagents = if turn_context.config.include_environment_context { + self.services + .agent_control + .format_environment_context_subagents(self.thread_id) + .await + } else { + String::new() + }; + + let mut world_state = WorldState::default(); + world_state.add_section(AgentsMdState::new(step_context.loaded_agents_md.as_deref())); + if turn_context.config.include_environment_context { + world_state.add_section( + EnvironmentsState::from_turn_context_with_environments( + turn_context, + &step_context.environments, + ) + .with_subagents(environment_subagents), + ); + } + world_state } - world_state } diff --git a/codex-rs/core/src/session_startup_prewarm.rs b/codex-rs/core/src/session_startup_prewarm.rs index 8ce5229a3..a5a4c5e43 100644 --- a/codex-rs/core/src/session_startup_prewarm.rs +++ b/codex-rs/core/src/session_startup_prewarm.rs @@ -266,6 +266,7 @@ async fn schedule_startup_prewarm_inner( } let startup_cancellation_token = CancellationToken::new(); let built_tools_started_at = Instant::now(); + // Startup prewarm runs before run_turn and needs its own tool-building snapshot. let step_context = session .capture_step_context(Arc::clone(&startup_turn_context)) .await; diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 853b96e92..cfed7d90d 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -4,6 +4,7 @@ use std::sync::atomic::AtomicBool; use crate::SkillsService; use crate::agent::AgentControl; +use crate::agents_md_manager::AgentsMdManager; use crate::attestation::AttestationProvider; use crate::client::ModelClient; use crate::config::NetworkProxyAuditMetadata; @@ -64,6 +65,7 @@ pub(crate) struct SessionServices { pub(crate) guardian_rejection_circuit_breaker: Mutex, pub(crate) runtime_handle: Handle, pub(crate) skills_service: Arc, + pub(crate) agents_md_manager: Arc, pub(crate) plugins_manager: Arc, pub(crate) mcp_manager: Arc, pub(crate) extensions: Arc>, diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index a886dbd97..f1acc2d97 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -304,6 +304,7 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() { #[tokio::test] async fn ignores_session_prefix_messages_when_truncating() { let (session, turn_context) = make_session_and_context().await; + let turn_context = Arc::new(turn_context); let world_state = build_world_state_from_turn_context(&session, &turn_context).await; let mut items = session .build_initial_context_with_world_state(&turn_context, &world_state) diff --git a/codex-rs/core/src/thread_rollout_truncation_tests.rs b/codex-rs/core/src/thread_rollout_truncation_tests.rs index 3c79a8ac2..0da9763fc 100644 --- a/codex-rs/core/src/thread_rollout_truncation_tests.rs +++ b/codex-rs/core/src/thread_rollout_truncation_tests.rs @@ -7,6 +7,7 @@ use codex_protocol::models::ReasoningItemReasoningSummary; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::ThreadRolledBackEvent; use pretty_assertions::assert_eq; +use std::sync::Arc; fn user_msg(text: &str) -> ResponseItem { ResponseItem::Message { @@ -167,6 +168,7 @@ fn truncates_rollout_from_start_applies_thread_rollback_markers() { #[tokio::test] async fn ignores_session_prefix_messages_when_truncating_rollout_from_start() { let (session, turn_context) = make_session_and_context().await; + let turn_context = Arc::new(turn_context); let world_state = build_world_state_from_turn_context(&session, &turn_context).await; let mut items = session .build_initial_context_with_world_state(&turn_context, &world_state) diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index f89b3e202..c422cc749 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -686,7 +686,11 @@ async fn environment_tools_follow_the_step_context() { let environments = turn.environments.clone(); turn.environments.turn_environments.clear(); - let step_context = Arc::new(StepContext::new(Arc::new(turn), environments)); + let step_context = Arc::new(StepContext::new( + Arc::new(turn), + environments, + /*loaded_agents_md*/ None, + )); let plan = ToolPlanProbe::from_router(ToolRouter::from_context( step_context.as_ref(), diff --git a/codex-rs/core/tests/suite/agents_md.rs b/codex-rs/core/tests/suite/agents_md.rs index fcc675b86..894b76027 100644 --- a/codex-rs/core/tests/suite/agents_md.rs +++ b/codex-rs/core/tests/suite/agents_md.rs @@ -95,6 +95,26 @@ 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()]); } @@ -750,11 +770,8 @@ async fn invalid_utf8_global_instructions_are_lossy() -> Result<()> { Ok(()) } -// TODO(anp): Align cold-resume instruction sources with the historical instructions replayed to -// the model so the API source list and model-visible context describe the same files. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn cold_resume_replays_rendered_instructions_but_reports_current_config_sources() -> Result<()> -{ +async fn cold_resume_injects_changed_agents_md_once() -> Result<()> { // Set up an initial turn and a later cold-resumed turn against the same rollout. let server = responses::start_mock_server().await; let response_mock = responses::mount_sse_sequence( @@ -768,6 +785,10 @@ async fn cold_resume_replays_rendered_instructions_but_reports_current_config_so responses::ev_response_created("resumed-response"), responses::ev_completed("resumed-response"), ]), + responses::sse(vec![ + responses::ev_response_created("second-resumed-response"), + responses::ev_completed("second-resumed-response"), + ]), ], ) .await; @@ -820,9 +841,10 @@ async fn cold_resume_replays_rendered_instructions_but_reports_current_config_so ); resumed.submit_turn("continue resumed thread").await?; + resumed.submit_turn("continue again").await?; let requests = response_mock.requests(); - assert_eq!(requests.len(), 2); + assert_eq!(requests.len(), 3); let initial_input = requests[0].input(); let resumed_input = requests[1].input(); assert_eq!( @@ -830,15 +852,17 @@ async fn cold_resume_replays_rendered_instructions_but_reports_current_config_so Some(initial_input.as_slice()), "cold resume should replay the original structured input prefix" ); - let expected_fragment = expected_provider_only_instruction_fragment(OLD_GLOBAL_INSTRUCTIONS); - assert_single_instruction_fragment(&requests[0], &expected_fragment); - assert_single_instruction_fragment(&requests[1], &expected_fragment); + assert_instruction_replacement_once( + &requests, + OLD_GLOBAL_INSTRUCTIONS, + NEW_GLOBAL_INSTRUCTIONS, + ); Ok(()) } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn fork_replays_rendered_instructions_from_shared_history() -> Result<()> { +async fn fork_injects_changed_agents_md_once() -> Result<()> { // Set up a parent turn and a later fork turn against the parent's rollout. let server = responses::start_mock_server().await; let response_mock = responses::mount_sse_sequence( @@ -852,6 +876,10 @@ async fn fork_replays_rendered_instructions_from_shared_history() -> Result<()> responses::ev_response_created("fork-response"), responses::ev_completed("fork-response"), ]), + responses::sse(vec![ + responses::ev_response_created("second-fork-response"), + responses::ev_completed("second-fork-response"), + ]), ], ) .await; @@ -908,27 +936,12 @@ async fn fork_replays_rendered_instructions_from_shared_history() -> Result<()> "fork config should reflect the newly loaded global source" ); - forked - .thread - .submit(Op::UserInput { - items: vec![UserInput::Text { - text: "continue fork".to_string(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - responsesapi_client_metadata: None, - additional_context: Default::default(), - thread_settings: Default::default(), - }) - .await?; - wait_for_event(&forked.thread, |event| { - matches!(event, EventMsg::TurnComplete(_)) - }) - .await; + submit_thread_turn(&forked.thread, "continue fork").await?; + submit_thread_turn(&forked.thread, "continue fork again").await?; // Assert the forked model request replays the parent's exact structured history. let requests = response_mock.requests(); - assert_eq!(requests.len(), 2); + assert_eq!(requests.len(), 3); let parent_input = requests[0].input(); let fork_input = requests[1].input(); assert_eq!( @@ -936,9 +949,11 @@ async fn fork_replays_rendered_instructions_from_shared_history() -> Result<()> Some(parent_input.as_slice()), "fork should replay the parent's original structured input prefix" ); - let expected_fragment = expected_provider_only_instruction_fragment(OLD_GLOBAL_INSTRUCTIONS); - assert_single_instruction_fragment(&requests[0], &expected_fragment); - assert_single_instruction_fragment(&requests[1], &expected_fragment); + assert_instruction_replacement_once( + &requests, + OLD_GLOBAL_INSTRUCTIONS, + NEW_GLOBAL_INSTRUCTIONS, + ); Ok(()) } diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index 17c101bd0..74c28664f 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -4811,11 +4811,17 @@ async fn remote_v2_compaction_keeps_creation_time_instructions_after_same_path_m .submit_turn("after remote v2 compaction cold resume") .await?; - // Modern replacement-history resume replays the persisted checkpoint and its later old-context - // suffix even though the same source path now contains new text. + // Cold resume replays the persisted old context, then appends the newly loaded instructions as + // an explicit replacement. let requests = response_mock.requests(); assert_eq!(requests.len(), 4); - assert_single_instruction_fragment(&requests[3], &old_fragment); + 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] + ); let resumed_input = requests[3].input(); assert_eq!( resumed_input.get(..replacement_history.len()), diff --git a/codex-rs/core/tests/suite/remote_env.rs b/codex-rs/core/tests/suite/remote_env.rs index a694d0fb2..5f8c50349 100644 --- a/codex-rs/core/tests/suite/remote_env.rs +++ b/codex-rs/core/tests/suite/remote_env.rs @@ -1,5 +1,7 @@ use anyhow::Context; use anyhow::Result; +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use codex_config::types::ApprovalsReviewer; use codex_core::compact::SUMMARIZATION_PROMPT; use codex_core::config::Constrained; @@ -38,6 +40,7 @@ use core_test_support::PathBufExt; use core_test_support::PathExt; use core_test_support::TestTargetOs; use core_test_support::responses::ResponseMock; +use core_test_support::responses::ResponsesRequest; use core_test_support::responses::ev_apply_patch_custom_tool_call; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; @@ -368,7 +371,7 @@ async fn read_exec_server_json(websocket: &mut WebSocketStream) -> Va } } -async fn serve_environment_info(listener: TcpListener) { +async fn accept_initialized_exec_server(listener: TcpListener) -> WebSocketStream { let (stream, _) = listener.accept().await.expect("connection"); let mut websocket = accept_async(stream).await.expect("websocket handshake"); @@ -388,7 +391,11 @@ async fn serve_environment_info(listener: TcpListener) { let initialized = read_exec_server_json(&mut websocket).await; assert_eq!(initialized["method"], "initialized"); - let info = read_exec_server_json(&mut websocket).await; + websocket +} + +async fn send_environment_info(websocket: &mut WebSocketStream) { + let info = read_exec_server_json(websocket).await; assert_eq!(info["method"], "environment/info"); websocket .send(Message::Text( @@ -403,6 +410,64 @@ async fn serve_environment_info(listener: TcpListener) { .expect("environment info response"); } +async fn serve_environment_info(listener: TcpListener) { + let mut websocket = accept_initialized_exec_server(listener).await; + send_environment_info(&mut websocket).await; +} + +async fn serve_environment_with_agents_md( + listener: TcpListener, + contents: &str, + attach: tokio::sync::oneshot::Receiver<()>, + mut shutdown: tokio::sync::oneshot::Receiver<()>, +) -> usize { + let mut websocket = accept_initialized_exec_server(listener).await; + attach.await.expect("attach signal"); + send_environment_info(&mut websocket).await; + + let mut agents_md_reads = 0; + loop { + let request = tokio::select! { + request = read_exec_server_json(&mut websocket) => request, + _ = &mut shutdown => return agents_md_reads, + }; + let is_agents_md = request["params"]["path"] + .as_str() + .is_some_and(|path| path.ends_with("/AGENTS.md")); + let response = match request["method"].as_str() { + Some("fs/getMetadata") if is_agents_md => { + json!({ + "id": request["id"], + "result": { + "isDirectory": false, + "isFile": true, + "isSymlink": false, + "size": contents.len(), + "createdAtMs": 0, + "modifiedAtMs": 0, + } + }) + } + Some("fs/getMetadata") => json!({ + "id": request["id"], + "error": { "code": -32004, "message": "not found" } + }), + Some("fs/readFile") if is_agents_md => { + agents_md_reads += 1; + json!({ + "id": request["id"], + "result": { "dataBase64": BASE64_STANDARD.encode(contents) } + }) + } + method => panic!("unexpected exec-server request: {method:?}"), + }; + websocket + .send(Message::Text(response.to_string().into())) + .await + .expect("filesystem response"); + } +} + fn tool_names(body: &Value) -> Vec { body["tools"] .as_array() @@ -468,6 +533,7 @@ async fn deferred_executor_updates_context_and_tools_after_startup() -> Result<( let mut builder = test_codex() .with_exec_server_url(format!("ws://{}", listener.local_addr()?)) .with_config(|config| { + config.project_doc_max_bytes = 0; config.use_experimental_unified_exec_tool = true; config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); config.approvals_reviewer = ApprovalsReviewer::User; @@ -579,6 +645,99 @@ async fn deferred_executor_updates_context_and_tools_after_startup() -> Result<( Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn deferred_executor_loads_agents_md_when_environment_becomes_ready() -> Result<()> { + const AGENTS_CONTENT: &str = "REMOTE_AGENTS_INSTRUCTIONS"; + + let listener = TcpListener::bind("127.0.0.1:0").await?; + let server = start_mock_server().await; + let response_mock = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("resp-1"), + ev_function_call( + "wait-1", + "wait_for_environment", + &json!({ "environment_id": REMOTE_ENVIRONMENT_ID }).to_string(), + ), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_function_call( + "wait-2", + "wait_for_environment", + &json!({ "environment_id": REMOTE_ENVIRONMENT_ID }).to_string(), + ), + ev_completed("resp-2"), + ]), + sse(vec![ + ev_response_created("resp-3"), + ev_assistant_message("msg-3", "done"), + ev_completed("resp-3"), + ]), + ], + ) + .await; + let mut builder = test_codex() + .with_exec_server_url(format!("ws://{}", listener.local_addr()?)) + .with_config(|config| { + assert!(config.features.enable(Feature::DeferredExecutor).is_ok()); + }); + let (attach_tx, attach_rx) = tokio::sync::oneshot::channel(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel(); + let exec_server = tokio::spawn(serve_environment_with_agents_md( + listener, + AGENTS_CONTENT, + attach_rx, + shutdown_rx, + )); + let test = timeout(Duration::from_secs(5), builder.build(&server)) + .await + .context("thread startup should not wait for the remote environment")??; + + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "load the environment instructions".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + wait_for_response_request_count(&response_mock, /*expected_count*/ 1).await; + let agents_path = PathUri::from_abs_path(&test.config.cwd).join("AGENTS.md")?; + attach_tx.send(()).expect("attach environment"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + shutdown_tx.send(()).expect("stop exec server"); + let agents_md_reads = exec_server.await?; + + let requests = response_mock.requests(); + assert_eq!(requests.len(), 3); + assert_eq!(agents_md_reads, 1); + assert_eq!(agents_md_occurrences(&requests[0], AGENTS_CONTENT), 0); + assert_eq!(agents_md_occurrences(&requests[1], AGENTS_CONTENT), 1); + assert_eq!(agents_md_occurrences(&requests[2], AGENTS_CONTENT), 1); + assert_eq!(test.codex.instruction_sources().await, vec![agents_path]); + + Ok(()) +} + +fn agents_md_occurrences(request: &ResponsesRequest, contents: &str) -> usize { + request + .message_input_texts("user") + .iter() + .filter(|text| text.contains(contents)) + .count() +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn deferred_executor_wait_reports_startup_failure() -> Result<()> { let listener = TcpListener::bind("127.0.0.1:0").await?; @@ -711,6 +870,7 @@ async fn deferred_executor_compaction_preserves_then_updates_environment_once() let mut builder = test_codex() .with_exec_server_url(format!("ws://{}", listener.local_addr()?)) .with_config(|config| { + config.project_doc_max_bytes = 0; assert!(config.features.enable(Feature::DeferredExecutor).is_ok()); assert!( config