diff --git a/codex-rs/codex-home/src/instructions/mod.rs b/codex-rs/codex-home/src/instructions/mod.rs index a1952b26a3..aa3a67267f 100644 --- a/codex-rs/codex-home/src/instructions/mod.rs +++ b/codex-rs/codex-home/src/instructions/mod.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use std::sync::Mutex; use codex_extension_api::Instructions; -use codex_extension_api::LoadUserInstructionsFuture; +use codex_extension_api::LoadInstructionsFuture; use codex_extension_api::LoadedUserInstructions; use codex_extension_api::UserInstructionsProvider; use codex_utils_absolute_path::AbsolutePathBuf; @@ -70,7 +70,7 @@ impl CodexHomeUserInstructionsProvider { return LoadedUserInstructions { instructions: Some(Instructions { text: trimmed.to_string(), - source: path, + source: Some(path), }), warnings, }; @@ -84,7 +84,7 @@ impl CodexHomeUserInstructionsProvider { } impl UserInstructionsProvider for CodexHomeUserInstructionsProvider { - fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_> { + fn load_user_instructions(&self) -> LoadInstructionsFuture<'_> { Box::pin(async move { let mut loaded = self.load_from_codex_home().await; let mut state = self diff --git a/codex-rs/codex-home/src/instructions/tests.rs b/codex-rs/codex-home/src/instructions/tests.rs index bebe897851..bb3e9e50f5 100644 --- a/codex-rs/codex-home/src/instructions/tests.rs +++ b/codex-rs/codex-home/src/instructions/tests.rs @@ -27,8 +27,10 @@ fn expected( LoadedUserInstructions { instructions: Some(Instructions { text: text.to_string(), - source: AbsolutePathBuf::try_from(home.path().join(filename)) - .expect("absolute source path"), + source: Some( + AbsolutePathBuf::try_from(home.path().join(filename)) + .expect("absolute source path"), + ), }), warnings, } diff --git a/codex-rs/core-api/src/lib.rs b/codex-rs/core-api/src/lib.rs index ca2b92cb67..a8facc3ae1 100644 --- a/codex-rs/core-api/src/lib.rs +++ b/codex-rs/core-api/src/lib.rs @@ -85,8 +85,9 @@ pub use codex_exec_server::NoiseRendezvousConnectBundle; pub use codex_exec_server::NoiseRendezvousConnectProvider; pub use codex_extension_api::ExtensionRegistryBuilder; pub use codex_extension_api::Instructions; -pub use codex_extension_api::LoadUserInstructionsFuture; +pub use codex_extension_api::LoadInstructionsFuture; pub use codex_extension_api::LoadedUserInstructions; +pub use codex_extension_api::ThreadInstructionsProvider; pub use codex_extension_api::UserInstructionsProvider; pub use codex_extension_api::empty_extension_registry; pub use codex_features::Feature; diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 306e770bcd..3da7924509 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -557,7 +557,17 @@ impl AgentControl { None, ) }; - // Reserving a slot can evict an idle nested parent. Keep its authority captured above. + let inherited_instructions = if let Some((parent, _)) = parent.as_ref() { + Some(parent.session.inherited_instructions().await) + } else if let Some(parent_thread_id) = parent_thread_id + && let Ok(parent) = state.get_thread(parent_thread_id).await + { + Some(parent.session.inherited_instructions().await) + } else { + None + }; + // Reserving a slot can evict an idle nested parent. Capture its instructions + // alongside its authority so the child does not depend on a later live lookup. let residency_slot = self .reserve_v2_residency_slot(&state, &config, Some(thread_id)) .await?; @@ -571,6 +581,7 @@ impl AgentControl { parent_thread_id, environment_selections, inherited_environments, + inherited_instructions, inherited_exec_policy, client_mcp_extensions, }) @@ -1277,6 +1288,7 @@ impl AgentControl { parent_thread_id, environment_selections: None, inherited_environments, + inherited_instructions: None, inherited_exec_policy, client_mcp_extensions: None, }) diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index f8cb0e5457..2981b13b0a 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -18,9 +18,15 @@ use crate::thread_manager::StartThreadOptions; use crate::tools::handlers::multi_agents_common::thread_spawn_source; use assert_matches::assert_matches; use codex_extension_api::ExtensionDataInit; +use codex_extension_api::Instructions; +use codex_extension_api::LoadInstructionsFuture; +use codex_extension_api::LoadedUserInstructions; +use codex_extension_api::ThreadInstructionsProvider; use codex_extension_api::empty_extension_registry; use codex_features::Feature; use codex_history::CompactedItem; +use codex_history::InitialHistory; +use codex_history::ResumedHistory; use codex_history::RolloutItem; use codex_login::AuthManager; use codex_login::CodexAuth; @@ -788,6 +794,16 @@ async fn check_v2_agent_reload(route: V2ReloadRoute) { .start_thread(StartThreadOptions { history_mode: Some(ThreadHistoryMode::Paginated), client_mcp_extensions: client_mcp_extensions.clone(), + user_instructions: Some(LoadedUserInstructions { + instructions: Some(Instructions { + text: "global instructions survive parent eviction".to_string(), + source: None, + }), + warnings: Vec::new(), + }), + thread_instructions_provider: Some(Arc::new(StaticThreadInstructionsProvider( + "thread instructions survive parent eviction", + ))), ..StartThreadOptions::new(harness.config.clone()) }) .await @@ -811,6 +827,9 @@ async fn check_v2_agent_reload(route: V2ReloadRoute) { } }; let parent_thread_id = parent_thread.session.thread_id; + let inherited_instructions = parent_thread.session.inherited_instructions().await; + assert!(inherited_instructions.user.is_some()); + assert!(inherited_instructions.thread.is_some()); let mut child_config = harness.config.clone(); child_config.model = Some("gpt-5.6-luna".to_string()); let spawned_agent = @@ -910,6 +929,12 @@ async fn check_v2_agent_reload(route: V2ReloadRoute) { .get_thread(spawned_agent.thread_id) .await .expect("reloaded child thread should exist"); + let reloaded_instructions = reloaded_child.session.inherited_instructions().await; + assert_eq!( + (reloaded_instructions.user, reloaded_instructions.thread), + (inherited_instructions.user, inherited_instructions.thread), + "reloading a child must retain both parent snapshots, even if residency evicts the parent", + ); if matches!(route, V2ReloadRoute::NestedParent) { let reloaded_turn = reloaded_child.session.new_default_turn().await; assert_eq!( @@ -1104,6 +1129,120 @@ async fn resume_agent_from_rollout_does_not_reopen_v2_descendants() { assert_thread_not_loaded(&resumed_manager, sibling_thread_id).await; } +struct StaticThreadInstructionsProvider(&'static str); + +impl ThreadInstructionsProvider for StaticThreadInstructionsProvider { + fn load_thread_instructions(&self) -> LoadInstructionsFuture<'_> { + let instructions = Instructions { + text: self.0.to_string(), + source: None, + }; + Box::pin(async move { + LoadedUserInstructions { + instructions: Some(instructions), + warnings: Vec::new(), + } + }) + } +} + +#[tokio::test] +async fn cold_resume_with_thread_instructions_preserves_lazy_v2_child_inheritance() { + let (home, mut config) = test_config().await; + let _ = config.features.enable(Feature::MultiAgentV2); + let _ = config.features.enable(Feature::Sqlite); + let harness = AgentControlHarness::new_with_config(home, config).await; + let parent = harness + .manager + .start_thread(StartThreadOptions { + thread_instructions_provider: Some(Arc::new(StaticThreadInstructionsProvider( + "initial thread instructions", + ))), + ..StartThreadOptions::new(harness.config.clone()) + }) + .await + .expect("start parent with thread instructions"); + let parent_thread_id = parent.thread_id; + let control = &parent.thread.session.services.agent_control; + let worker_thread_id = + spawn_v2_reload_test_child(control, harness.config.clone(), &parent.thread, "worker") + .await + .thread_id; + let worker = harness + .manager + .get_thread(worker_thread_id) + .await + .expect("worker should be loaded"); + persist_thread_for_tree_resume(&parent.thread, "parent persisted").await; + persist_thread_for_tree_resume(&worker, "worker persisted").await; + wait_for_live_thread_spawn_children(control, parent_thread_id, &[worker_thread_id]).await; + + let stored_parent = parent + .thread + .read_thread( + /*include_archived*/ true, /*include_history*/ true, + ) + .await + .expect("read parent history"); + let initial_history = InitialHistory::Resumed(ResumedHistory { + conversation_id: parent_thread_id, + history: Arc::new(stored_parent.history.expect("parent history").items), + rollout_path: stored_parent.rollout_path, + }); + let report = harness + .manager + .shutdown_all_threads_bounded(Duration::from_secs(5)) + .await; + assert_eq!(report.submit_failed, Vec::::new()); + assert_eq!(report.timed_out, Vec::::new()); + + let resumed_manager = ThreadManager::with_models_provider_home_and_state_for_tests( + CodexAuth::from_api_key("dummy"), + harness.config.model_provider.clone(), + harness.config.codex_home.to_path_buf(), + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + harness.state_db.clone(), + ); + let resumed_parent = resumed_manager + .start_thread(StartThreadOptions { + initial_history, + session_source: Some(SessionSource::Exec), + thread_instructions_provider: Some(Arc::new(StaticThreadInstructionsProvider( + "resumed thread instructions", + ))), + ..StartThreadOptions::new(harness.config.clone()) + }) + .await + .expect("cold resume parent with updated thread instructions"); + let thread_instructions = Instructions { + text: "resumed thread instructions".to_string(), + source: None, + }; + assert_eq!( + resumed_parent + .thread + .session + .inherited_instructions() + .await + .thread, + Some(thread_instructions.clone()), + ); + assert_thread_not_loaded(&resumed_manager, worker_thread_id).await; + + resumed_manager + .ensure_multi_agent_v2_child_loaded(worker_thread_id) + .await + .expect("resume v2 worker on demand"); + let resumed_worker = resumed_manager + .get_thread(worker_thread_id) + .await + .expect("resumed worker should be loaded"); + assert_eq!( + resumed_worker.session.inherited_instructions().await.thread, + Some(thread_instructions), + ); +} + #[tokio::test] async fn spawn_agent_creates_thread_and_sends_prompt() { let harness = AgentControlHarness::new().await; diff --git a/codex-rs/core/src/agents_md.rs b/codex-rs/core/src/agents_md.rs index 37678f3e34..0bcb2903e2 100644 --- a/codex-rs/core/src/agents_md.rs +++ b/codex-rs/core/src/agents_md.rs @@ -286,9 +286,12 @@ fn candidate_filenames(config: &Config) -> Vec<&str> { /// guidance. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct LoadedAgentsMd { - /// Host-provided user instructions. + /// Account- or home-scoped instructions supplied by the host. user_instructions: Option, + /// Thread-scoped instructions supplied by the host. + thread_instructions: Option, + /// Ordered instructions and their provenance. entries: Vec, } @@ -302,8 +305,9 @@ impl LoadedAgentsMd { Self { user_instructions: Some(Instructions { text: contents, - source: path, + source: Some(path), }), + thread_instructions: None, entries: Vec::new(), } } @@ -312,16 +316,20 @@ impl LoadedAgentsMd { Self { user_instructions: user_instructions .filter(|instructions| !instructions.text.trim().is_empty()), + thread_instructions: None, entries: Vec::new(), } } - pub(crate) fn with_user_instructions( + pub(crate) fn with_instructions( mut self, user_instructions: Option, + thread_instructions: Option, ) -> Option { self.user_instructions = user_instructions.filter(|instructions| !instructions.text.trim().is_empty()); + self.thread_instructions = + thread_instructions.filter(|instructions| !instructions.text.trim().is_empty()); (!self.is_empty()).then_some(self) } @@ -336,6 +344,7 @@ impl LoadedAgentsMd { } Self { user_instructions: None, + thread_instructions: None, entries: vec![InstructionEntry { contents, provenance: InstructionProvenance::Internal, @@ -345,6 +354,7 @@ impl LoadedAgentsMd { fn is_empty(&self) -> bool { self.user_instructions.is_none() + && self.thread_instructions.is_none() && self .entries .iter() @@ -364,7 +374,14 @@ impl LoadedAgentsMd { let mut output = String::new(); let mut has_previous = false; let mut previous_was_project = false; - if let Some(instructions) = &self.user_instructions { + for instructions in self + .user_instructions + .iter() + .chain(self.thread_instructions.iter()) + { + if has_previous { + output.push_str("\n\n"); + } output.push_str(&instructions.text); has_previous = true; } @@ -392,7 +409,14 @@ impl LoadedAgentsMd { let mut output = String::new(); let mut has_previous = false; let mut previous_environment: Option<(&str, &PathUri)> = None; - if let Some(instructions) = &self.user_instructions { + for instructions in self + .user_instructions + .iter() + .chain(self.thread_instructions.iter()) + { + if has_previous { + output.push_str("\n\n"); + } output.push_str(&instructions.text); has_previous = true; } @@ -452,7 +476,8 @@ impl LoadedAgentsMd { pub fn sources(&self) -> impl Iterator + '_ { self.user_instructions .iter() - .map(|instructions| PathUri::from_abs_path(&instructions.source)) + .chain(self.thread_instructions.iter()) + .filter_map(|instructions| instructions.source.as_ref().map(PathUri::from_abs_path)) .chain( self.entries .iter() diff --git a/codex-rs/core/src/agents_md_manager.rs b/codex-rs/core/src/agents_md_manager.rs index 3bb500192b..68fdeeeafb 100644 --- a/codex-rs/core/src/agents_md_manager.rs +++ b/codex-rs/core/src/agents_md_manager.rs @@ -1,4 +1,4 @@ -//! Resolves global and repository instructions into one validated snapshot. +//! Resolves global, thread, and repository instructions into one validated snapshot. //! Refreshes are serialized; the state lock is never held while calling providers. use crate::agents_md::LoadedAgentsMd; @@ -6,11 +6,14 @@ 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::ThreadInstructionsProvider; 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 codex_utils_string::approx_bytes_for_tokens; +use codex_utils_string::approx_tokens_from_byte_count; use std::sync::Arc; use tokio::sync::Mutex; use tokio::sync::Semaphore; @@ -25,7 +28,9 @@ pub(crate) struct AgentsMdManager { #[derive(Clone, Default)] pub(crate) struct SessionInstructions { pub(crate) user: Option, + pub(crate) thread: Option, pub(crate) user_provider: Option>, + pub(crate) thread_provider: Option>, } struct AgentsMdState { @@ -43,6 +48,7 @@ struct AgentsMdCache { impl AgentsMdManager { pub(crate) fn new(mut instructions: SessionInstructions) -> Self { instructions.user = normalize_instructions(instructions.user); + instructions.thread = normalize_instructions(instructions.thread); Self { refresh_lock: Semaphore::new(/*permits*/ 1), state: Mutex::new(AgentsMdState { @@ -96,11 +102,21 @@ impl AgentsMdManager { instructions.user = normalize_instructions(loaded.instructions); warnings.extend(loaded.warnings); } + if let Some(provider) = &instructions.thread_provider { + let loaded = provider.load_thread_instructions().await; + instructions.thread = normalize_instructions(loaded.instructions); + warnings.extend(loaded.warnings); + } let result = async { + if let Some(input) = &instructions.thread { + validate_thread_instruction_size(input.text.len())?; + } if !refresh_repository { let state = self.state.lock().await; - if state.instructions.user == instructions.user { + if state.instructions.user == instructions.user + && state.instructions.thread == instructions.thread + { return Ok(cached); } } @@ -112,7 +128,7 @@ impl AgentsMdManager { } else { cached.as_deref().cloned().unwrap_or_default() } - .with_user_instructions(instructions.user.clone()) + .with_instructions(instructions.user.clone(), instructions.thread.clone()) .map(Arc::new); let mut state = self.state.lock().await; state.instructions = instructions; @@ -135,11 +151,27 @@ impl AgentsMdManager { let state = self.state.lock().await; SessionInstructions { user: state.instructions.user.clone(), + thread: state.instructions.thread.clone(), ..Default::default() } } } +// Bound the new host-provided contribution independently of project_doc_max_bytes, +// which controls repository discovery. Existing global and combined instruction +// size policy is unchanged; reject oversized thread input rather than truncate it. +const MAX_THREAD_INSTRUCTIONS_TOKENS: usize = 10_000; + +fn validate_thread_instruction_size(bytes: usize) -> CodexResult<()> { + if bytes > approx_bytes_for_tokens(MAX_THREAD_INSTRUCTIONS_TOKENS) { + let estimated_tokens = approx_tokens_from_byte_count(bytes); + return Err(CodexErr::InvalidRequest(format!( + "thread instructions exceed the limit of {MAX_THREAD_INSTRUCTIONS_TOKENS} estimated tokens ({estimated_tokens} estimated tokens provided)" + ))); + } + Ok(()) +} + fn normalize_instructions(instructions: Option) -> Option { instructions.filter(|instructions| !instructions.text.trim().is_empty()) } diff --git a/codex-rs/core/src/agents_md_tests.rs b/codex-rs/core/src/agents_md_tests.rs index bfa1f8768a..d2e8f0644a 100644 --- a/codex-rs/core/src/agents_md_tests.rs +++ b/codex-rs/core/src/agents_md_tests.rs @@ -418,6 +418,7 @@ fn foreign_agents_md_uses_environment_native_paths() { let source_path = cwd.join("AGENTS.md").expect("AGENTS.md URI"); let loaded = LoadedAgentsMd { user_instructions: None, + thread_instructions: None, entries: vec![InstructionEntry { contents: "remote instructions".to_string(), provenance: InstructionProvenance::Project { @@ -451,6 +452,7 @@ fn multi_environment_agents_md_renders_mixed_path_conventions() { .expect("Windows AGENTS.md URI"); let loaded = LoadedAgentsMd { user_instructions: None, + thread_instructions: None, entries: vec![ InstructionEntry { contents: "POSIX instructions".to_string(), @@ -509,7 +511,7 @@ async fn make_config(root: &TempDir, limit: usize, instructions: Option<&str>) - let user_instructions = instructions.map(|text| Instructions { text: text.to_owned(), - source: config.codex_home.join(DEFAULT_AGENTS_MD_FILENAME), + source: Some(config.codex_home.join(DEFAULT_AGENTS_MD_FILENAME)), }); TestConfig { config, @@ -558,7 +560,7 @@ async fn make_config_with_project_root_markers( config.project_doc_max_bytes = limit; let user_instructions = instructions.map(|text| Instructions { text: text.to_owned(), - source: config.codex_home.join(DEFAULT_AGENTS_MD_FILENAME), + source: Some(config.codex_home.join(DEFAULT_AGENTS_MD_FILENAME)), }); TestConfig { config, @@ -608,6 +610,7 @@ fn empty_loaded_instructions_are_empty() { fn loaded_instructions_with_only_empty_or_whitespace_entries_are_empty() { let empty = LoadedAgentsMd { user_instructions: None, + thread_instructions: None, entries: vec![InstructionEntry { contents: String::new(), provenance: InstructionProvenance::Internal, @@ -615,6 +618,7 @@ fn loaded_instructions_with_only_empty_or_whitespace_entries_are_empty() { }; let whitespace = LoadedAgentsMd { user_instructions: None, + thread_instructions: None, entries: vec![InstructionEntry { contents: " \n\t".to_string(), provenance: InstructionProvenance::Internal, @@ -686,6 +690,7 @@ async fn total_byte_limit_truncates_later_project_docs() { let loaded = load_agents_md(&config).await.expect("project instructions"); let expected = LoadedAgentsMd { user_instructions: None, + thread_instructions: None, entries: vec![ InstructionEntry { contents: "root".to_string(), @@ -1100,16 +1105,16 @@ async fn zero_byte_limit_disables_docs() { ); } -/// When both system instructions and AGENTS.md docs are present the two -/// should be concatenated with the separator. +/// User instructions precede project docs without consuming their byte budget. #[tokio::test] async fn merges_existing_instructions_with_agents_md() { let tmp = tempfile::tempdir().expect("tempdir"); - fs::write(tmp.path().join("AGENTS.md"), "proj doc").unwrap(); + fs::write(tmp.path().join("AGENTS.md"), "proj document").unwrap(); const INSTRUCTIONS: &str = "base instructions"; + let limit = "proj doc".len(); - let res = get_user_instructions(&make_config(&tmp, /*limit*/ 4096, Some(INSTRUCTIONS)).await) + let res = get_user_instructions(&make_config(&tmp, limit, Some(INSTRUCTIONS)).await) .await .expect("should produce a combined instruction string"); @@ -1173,11 +1178,13 @@ secondary doc"#, loaded.sources().collect::>(), vec![ PathUri::from_abs_path( - &config + config .user_instructions .as_ref() .expect("global instructions") - .source, + .source + .as_ref() + .expect("global instruction source"), ), PathUri::from_abs_path(&primary.path().join("AGENTS.md").abs()), PathUri::from_abs_path(&primary_nested.join("AGENTS.md").abs()), @@ -1376,6 +1383,7 @@ async fn concatenates_root_and_cwd_docs() { let crate_agents = cfg.cwd.join("AGENTS.md"); let expected = LoadedAgentsMd { user_instructions: None, + thread_instructions: None, entries: vec![ InstructionEntry { contents: "root doc".to_string(), @@ -1513,8 +1521,9 @@ async fn instruction_sources_include_global_before_agents_md_docs() { let expected = LoadedAgentsMd { user_instructions: Some(Instructions { text: "global doc".to_string(), - source: global_agents.clone(), + source: Some(global_agents.clone()), }), + thread_instructions: None, entries: vec![InstructionEntry { contents: "project doc".to_string(), provenance: project_provenance(project_agents.clone(), cfg.cwd.clone()), diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 22e1ff5965..f7d38d20c7 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -222,6 +222,7 @@ pub struct GuardianReviewSessionReuseKey { developer_instructions: Option, base_instructions: Option, user_instructions: Option, + thread_instructions: Option, compact_prompt: Option, cwd: PathUri, mcp_servers: Constrained>, @@ -264,6 +265,7 @@ impl GuardianReviewSessionReuseKey { developer_instructions: spawn_config.developer_instructions.clone(), base_instructions: spawn_config.base_instructions.clone(), user_instructions: instructions.user, + thread_instructions: instructions.thread, 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 c7be2363fd..894ee52c14 100644 --- a/codex-rs/core/src/guardian/review_session_factory.rs +++ b/codex-rs/core/src/guardian/review_session_factory.rs @@ -137,6 +137,7 @@ impl ReviewerSessionFactory for PreparedSession { let inherited = session.inherited_instructions().await; let context = GuardianReviewSessionReuseKey { user_instructions: inherited.user, + thread_instructions: inherited.thread, ..context }; Ok(GuardianReviewSession { diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index d0239d7e74..987faf87fd 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -282,22 +282,27 @@ async fn spawned_guardian_session_preserves_windows_sandbox_proxy_settings() { #[tokio::test] async fn spawned_guardian_reuse_key_matches_inherited_instructions() { let mut params = test_review_params().await; + let latest = Some(Instructions { + text: "latest thread instructions".to_string(), + source: None, + }); let latest_global = Some(Instructions { text: "latest global instructions".to_string(), - source: params.spawn_config.codex_home.join("AGENTS.md"), + source: None, }); 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(), + thread: latest.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"), + thread: Some(Instructions { + text: "previous thread instructions".to_string(), + source: None, }), ..Default::default() }, @@ -305,7 +310,8 @@ async fn spawned_guardian_reuse_key_matches_inherited_instructions() { parent.guardian_context_mode, ); let expected_key = GuardianReviewSessionReuseKey { - user_instructions: latest_global.clone(), + user_instructions: latest_global, + thread_instructions: latest.clone(), ..stale_key.clone() }; let prepared = factory::prepare_review(params) @@ -323,10 +329,7 @@ async fn spawned_guardian_reuse_key_matches_inherited_instructions() { .expect("spawn reviewer after instruction update"); assert_eq!(review.reuse_key, expected_key); - assert_eq!( - review.session.inherited_instructions().await.user, - latest_global - ); + assert_eq!(review.session.inherited_instructions().await.thread, latest); review.shutdown().await; } @@ -394,6 +397,17 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { GuardianContextMode::Legacy, ) ); + assert_ne!( + cached_reuse_key.clone(), + GuardianReviewSessionReuseKey { + thread_instructions: Some(Instructions { + text: "updated thread instructions".to_string(), + source: None, + }), + ..cached_reuse_key.clone() + }, + "changing thread instructions must invalidate reviewer history" + ); assert_ne!( cached_reuse_key .clone() @@ -969,7 +983,11 @@ async fn run_review_removes_trunk_when_event_stream_is_broken() { 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"), + source: None, + }), + thread: Some(Instructions { + text: "parent thread instructions".to_string(), + source: None, }), ..Default::default() })); diff --git a/codex-rs/core/src/guardian/review_session_threads.rs b/codex-rs/core/src/guardian/review_session_threads.rs index dd3d793140..5d50f0710d 100644 --- a/codex-rs/core/src/guardian/review_session_threads.rs +++ b/codex-rs/core/src/guardian/review_session_threads.rs @@ -5,7 +5,6 @@ use std::sync::Arc; use std::sync::Weak; use codex_async_utils::OrCancelExt; -use codex_extension_api::LoadedUserInstructions; use codex_history::InitialHistory; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::InternalSessionSource; @@ -75,28 +74,28 @@ impl ManagedReviewerThreads { thread_source: Some(ThreadSource::GuardianReview), environments: Some(context.environments().to_selections()), inherited_environments: Some(context.environments().clone()), - user_instructions: Some(LoadedUserInstructions { - instructions: parent.inherited_instructions().await.user, - warnings: Vec::new(), - }), client_mcp_extensions: parent.services.client_mcp_extensions.clone(), ..StartThreadOptions::new(config) }; - let spawned = match history.unwrap_or(InitialHistory::New) { - InitialHistory::Forked(history) => manager - .fork_internal_session(parent.thread_id(), options, history) - .or_cancel(&cancel) - .await - .map_err(codex_protocol::error::CodexErr::from)??, - InitialHistory::New | InitialHistory::Cleared => manager - .spawn_internal_session(parent.thread_id(), options) - .or_cancel(&cancel) - .await - .map_err(codex_protocol::error::CodexErr::from)??, + let history = match history.unwrap_or(InitialHistory::New) { + InitialHistory::Forked(history) => InitialHistory::Forked(history), + InitialHistory::New | InitialHistory::Cleared => InitialHistory::New, InitialHistory::Resumed(_) => { anyhow::bail!("guardian review forks cannot resume an existing thread") } }; + // Generic internal sessions start isolated. Reviewers explicitly inherit + // the parent's applied instructions, just like standalone delegates. + let spawned = manager + .spawn_internal_session_with_history( + parent.thread_id(), + options, + history, + Some(parent.inherited_instructions().await), + ) + .or_cancel(&cancel) + .await + .map_err(codex_protocol::error::CodexErr::from)??; let thread = spawned.thread; let session = Arc::clone(&thread.session); let io = forward_session_io( diff --git a/codex-rs/core/src/session/step_activation_tests.rs b/codex-rs/core/src/session/step_activation_tests.rs index 828a5d36b5..dcaa598eb6 100644 --- a/codex-rs/core/src/session/step_activation_tests.rs +++ b/codex-rs/core/src/session/step_activation_tests.rs @@ -18,8 +18,9 @@ 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::LoadInstructionsFuture; use codex_extension_api::LoadedUserInstructions; +use codex_extension_api::ThreadInstructionsProvider; use codex_extension_api::UserInstructionsProvider; use codex_http_client::HttpClientFactory; use codex_login::AuthManager; @@ -60,42 +61,54 @@ use tokio_util::sync::CancellationToken; const MODEL_A: &str = "step-activation-a"; const MODEL_B: &str = "step-activation-b"; +struct GatedInstructionsProvider { + instructions: StdMutex>, + block_next_read: AtomicBool, + read_started: Notify, +} + +impl GatedInstructionsProvider { + fn load(&self) -> LoadInstructionsFuture<'_> { + 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(), + } + }) + } +} + +impl UserInstructionsProvider for GatedInstructionsProvider { + fn load_user_instructions(&self) -> LoadInstructionsFuture<'_> { + self.load() + } +} + +impl ThreadInstructionsProvider for GatedInstructionsProvider { + fn load_thread_instructions(&self) -> LoadInstructionsFuture<'_> { + self.load() + } +} + #[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(), + source: Some(source.clone()), }) }; let provider = Arc::new(GatedInstructionsProvider { @@ -149,6 +162,91 @@ async fn instruction_refresh_serializes_reads_and_releases_on_cancellation() { ); } +#[test_case(None; "thread only")] +#[test_case(Some("global instructions"); "with global instructions")] +#[tokio::test] +async fn thread_instruction_refresh_serializes_reads_and_releases_on_cancellation( + global_text: Option<&str>, +) { + let (mut session, mut turn) = make_session_and_context().await; + // This tests provider snapshots, not the checkout's repository instructions. + Arc::make_mut(&mut turn.config).project_doc_max_bytes = 0; + let turn = Arc::new(turn); + let global = global_text.map(|text| Instructions { + text: text.to_string(), + source: Some(turn.config.codex_home.join("AGENTS.md")), + }); + let instructions = |text: &str| { + Some(Instructions { + text: text.to_string(), + source: None, + }) + }; + 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: global.clone(), + thread_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()); + let inherited = session.inherited_instructions().await; + assert_eq!( + (inherited.user, inherited.thread), + (global.clone(), instructions("initial")) + ); + + cancellation.cancel(); + assert!(first_capture.await.is_err()); + let step = next_capture + .await + .expect("cancelled read releases the refresh guard"); + let inherited = session.inherited_instructions().await; + assert_eq!( + (inherited.user, inherited.thread), + (global, instructions("latest")) + ); + let expected = global_text.map_or_else( + || "latest".to_string(), + |global| format!("{global}\n\nlatest"), + ); + assert_eq!( + step.loaded_agents_md + .as_ref() + .expect("captured instructions") + .text(), + expected + ); +} + fn activation_models() -> Vec { let model = bundled_models_response() .expect("bundled models") diff --git a/codex-rs/core/src/test_support.rs b/codex-rs/core/src/test_support.rs index 8ce1f815d7..b93c2ba700 100644 --- a/codex-rs/core/src/test_support.rs +++ b/codex-rs/core/src/test_support.rs @@ -8,7 +8,7 @@ use std::path::PathBuf; use std::sync::Arc; use codex_exec_server::EnvironmentManager; -use codex_extension_api::LoadUserInstructionsFuture; +use codex_extension_api::LoadInstructionsFuture; use codex_extension_api::LoadedUserInstructions; use codex_extension_api::UserInstructionsProvider; use codex_http_client::HttpClientFactory; @@ -69,7 +69,7 @@ pub async fn history_with_tool_call_metadata( pub struct EmptyUserInstructionsProvider; impl UserInstructionsProvider for EmptyUserInstructionsProvider { - fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_> { + fn load_user_instructions(&self) -> LoadInstructionsFuture<'_> { Box::pin(async { LoadedUserInstructions::default() }) } } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 9ce622f540..fe9d373182 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -34,6 +34,7 @@ use codex_exec_server::EnvironmentManager; use codex_extension_api::ExtensionDataInit; use codex_extension_api::ExtensionRegistry; use codex_extension_api::LoadedUserInstructions; +use codex_extension_api::ThreadInstructionsProvider; use codex_extension_api::UserInstructionsProvider; use codex_extension_api::empty_extension_registry; use codex_features::Feature; @@ -231,6 +232,19 @@ pub struct ThreadManager { pub struct StartThreadOptions { pub config: Config, + /// Host-owned provider for this root thread's additional instructions. + /// + /// Core composes these after the global [`UserInstructionsProvider`] + /// snapshot and before repository instructions. `None` or blank provider + /// output clears only the thread contribution. Core retains this provider + /// and reads it when capturing model-request context. Hosts own fetching + /// and caching, and must supply a provider again when cold-resuming a thread + /// or forking an offline source. A live source can supply its applied snapshot + /// to a fork, but its task-bound provider is never inherited by the new root. + /// Registration lasts for the live session. Warm resume ignores this option + /// and reuses the existing provider without loading it; normal model-request + /// boundaries load its latest instructions. + pub thread_instructions_provider: Option>, pub allow_provider_model_fallback: bool, pub initial_history: InitialHistory, pub history_mode: Option, @@ -242,7 +256,7 @@ pub struct StartThreadOptions { pub environments: Option>, /// Existing environment bindings captured by an internal caller. pub inherited_environments: Option, - /// Explicit instructions carried by an internal caller instead of loading them again. + /// Explicit global instructions carried by an internal caller instead of loading them again. pub user_instructions: Option, pub thread_extension_init: ExtensionDataInit, pub client_mcp_extensions: ClientMcpExtensions, @@ -256,6 +270,7 @@ impl StartThreadOptions { pub fn new(config: Config) -> Self { Self { config, + thread_instructions_provider: None, allow_provider_model_fallback: false, initial_history: InitialHistory::New, history_mode: None, @@ -283,6 +298,7 @@ struct ThreadSpawnRequest { forked_from_thread_id: Option, fork_persistence: ForkPersistence, inherited_environments: Option, + inherited_instructions: Option, inherited_exec_policy: Option>, user_shell_override: Option, } @@ -301,6 +317,7 @@ impl ThreadSpawnRequest { forked_from_thread_id: None, fork_persistence: ForkPersistence::Copied, inherited_environments: None, + inherited_instructions: None, inherited_exec_policy: None, user_shell_override: None, } @@ -345,6 +362,7 @@ pub(crate) struct ResumeThreadWithHistoryOptions { pub(crate) parent_thread_id: Option, pub(crate) environment_selections: Option>, pub(crate) inherited_environments: Option, + pub(crate) inherited_instructions: Option, pub(crate) inherited_exec_policy: Option>, pub(crate) client_mcp_extensions: Option, } @@ -978,8 +996,13 @@ impl ThreadManager { parent_thread_id: ThreadId, options: StartThreadOptions, ) -> CodexResult { - self.spawn_internal_session_with_history(parent_thread_id, options, InitialHistory::New) - .await + self.spawn_internal_session_with_history( + parent_thread_id, + options, + InitialHistory::New, + /*inherited_instructions*/ None, + ) + .await } /// Starts an internal session from an explicitly selected, committed history. @@ -996,15 +1019,17 @@ impl ThreadManager { parent_thread_id, options, InitialHistory::Forked(history), + /*inherited_instructions*/ None, ) .await } - async fn spawn_internal_session_with_history( + pub(crate) async fn spawn_internal_session_with_history( &self, parent_thread_id: ThreadId, mut options: StartThreadOptions, history: InitialHistory, + inherited_instructions: Option, ) -> CodexResult { if !matches!(options.session_source, Some(SessionSource::Internal(_))) { return Err(CodexErr::InvalidRequest( @@ -1021,6 +1046,7 @@ impl ThreadManager { ); request.parent_thread_id = Some(parent_thread_id); request.forked_from_thread_id = forked_from_thread_id; + request.inherited_instructions = inherited_instructions; Box::pin(self.state.spawn_thread(request)).await } @@ -1154,14 +1180,6 @@ impl ThreadManager { let (session_source, thread_source) = initial_history .get_resumed_session_sources() .unwrap_or_else(|| (self.state.session_source.clone(), None)); - if let InitialHistory::Resumed(resumed) = &initial_history - && initial_history.get_multi_agent_version() == Some(MultiAgentVersion::V2) - && !session_source.is_non_root_agent() - { - agent_control - .restore_v2_agent_metadata(&config, resumed.conversation_id) - .await; - } let options = StartThreadOptions { initial_history, session_source: Some(session_source), @@ -1319,7 +1337,8 @@ impl ThreadManager { /// Fork an existing thread by snapshotting rollout history according to /// `snapshot` and starting a new thread with identical configuration /// (unless overridden by the caller's options). The new thread has a fresh id. - /// Fork history replaces `options.initial_history`. + /// Fork history replaces `options.initial_history`; hosts can supply the new + /// task's instruction provider through the same options as thread creation. pub async fn fork_thread( &self, snapshot: S, @@ -1420,6 +1439,20 @@ impl ThreadManager { InitialHistory::Forked(_) => history.forked_from_id(), InitialHistory::New | InitialHistory::Cleared => None, }; + // Capture the live source before later startup work can unload it. Keep + // the fork's own providers, not the source's task-bound callback. + let instructions = self + .state + .instructions_for_spawn( + options + .session_source + .as_ref() + .unwrap_or(&self.state.session_source), + /*parent_thread_id*/ None, + source_thread_id, + options.thread_instructions_provider.clone(), + ) + .await; // Capture the source's settings before truncating its model history. options.disabled_plugin_ids = Some(options.disabled_plugin_ids.unwrap_or_else(|| { source_thread_id @@ -1452,6 +1485,7 @@ impl ThreadManager { ThreadSpawnRequest::new(options, Arc::clone(&self.state.auth_manager), agent_control); request.forked_from_thread_id = source_thread_id; request.fork_persistence = fork_persistence; + request.inherited_instructions = Some(instructions); Box::pin(self.state.spawn_thread(request)).await } @@ -1655,35 +1689,41 @@ impl ThreadManagerState { /// Selects instruction sources for a newly spawned runtime without loading providers. /// - /// 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. + /// Fresh roots, cold resumes, and root forks retain the global and any supplied + /// thread provider. The session's AgentsMdManager loads them at startup and + /// subsequent context captures. Subagents inherit only applied parent snapshots. + /// A root fork without a provider preserves the live source's thread snapshot, + /// never its task-bound provider. Hosts must supply a provider for offline forks. /// 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, + thread_provider: Option>, ) -> 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), }; - match inherited_thread_id { + let inherited = match inherited_thread_id { Some(thread_id) => match self.get_thread(thread_id).await { Ok(thread) => thread.session.inherited_instructions().await, Err(_) => SessionInstructions::default(), }, None => SessionInstructions::default(), + }; + if session_source.is_non_root_agent() { + inherited + } else { + SessionInstructions { + user_provider: Some(Arc::clone(&self.user_instructions_provider)), + thread_provider, + thread: inherited.thread, + ..Default::default() + } } } @@ -1813,6 +1853,7 @@ impl ThreadManagerState { parent_thread_id, environment_selections, inherited_environments, + inherited_instructions, inherited_exec_policy, client_mcp_extensions, } = options; @@ -1838,6 +1879,7 @@ impl ThreadManagerState { ThreadSpawnRequest::new(options, Arc::clone(&self.auth_manager), agent_control); request.parent_thread_id = parent_thread_id; request.inherited_environments = inherited_environments; + request.inherited_instructions = inherited_instructions; request.inherited_exec_policy = inherited_exec_policy; Box::pin(self.spawn_thread(request)).await } @@ -1901,11 +1943,13 @@ impl ThreadManagerState { forked_from_thread_id, fork_persistence, inherited_environments, + inherited_instructions, inherited_exec_policy, user_shell_override, } = request; let StartThreadOptions { mut config, + thread_instructions_provider, allow_provider_model_fallback, initial_history, history_mode, @@ -1981,10 +2025,19 @@ impl ThreadManagerState { threads.remove(&resumed.conversation_id); } } + // Both resume entry points must restore identities before children can be loaded lazily. + if let InitialHistory::Resumed(resumed) = &initial_history + && initial_history.get_multi_agent_version() == Some(MultiAgentVersion::V2) + && !session_source.is_non_root_agent() + { + agent_control + .restore_v2_agent_metadata(&config, resumed.conversation_id) + .await; + } let (instructions, inherited_exec_policy, extensions, mcp_manager, multi_agent_version) = if isolation == codex_extension_api::SessionIsolation::Isolated { ( - SessionInstructions::default(), + inherited_instructions.unwrap_or_default(), None, empty_extension_registry(), Arc::new(McpManager::new(Arc::clone(&self.plugins_manager))), @@ -1992,12 +2045,18 @@ impl ThreadManagerState { ) } else { ( - self.instructions_for_spawn( - &session_source, - parent_thread_id, - forked_from_thread_id, - ) - .await, + match inherited_instructions { + Some(instructions) => instructions, + None => { + self.instructions_for_spawn( + &session_source, + parent_thread_id, + forked_from_thread_id, + thread_instructions_provider, + ) + .await + } + }, inherited_exec_policy, Arc::clone(&self.extensions), Arc::clone(&self.mcp_manager), diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 88fe058d5e..131df94c35 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -53,6 +53,103 @@ use wiremock::MockServer; const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111"; +struct ParentInstructionsProvider(codex_extension_api::Instructions); + +impl codex_extension_api::UserInstructionsProvider for ParentInstructionsProvider { + fn load_user_instructions(&self) -> codex_extension_api::LoadInstructionsFuture<'_> { + Box::pin(async move { + codex_extension_api::LoadedUserInstructions { + instructions: Some(self.0.clone()), + warnings: Vec::new(), + } + }) + } +} + +impl codex_extension_api::ThreadInstructionsProvider for ParentInstructionsProvider { + fn load_thread_instructions(&self) -> codex_extension_api::LoadInstructionsFuture<'_> { + codex_extension_api::UserInstructionsProvider::load_user_instructions(self) + } +} + +#[tokio::test] +async fn live_fork_keeps_instructions_when_source_is_unloaded_during_setup() { + let temp_dir = tempdir().expect("tempdir"); + let mut config = test_config().await; + config.codex_home = temp_dir.path().join("codex-home").abs(); + config.cwd = config.codex_home.abs(); + std::fs::create_dir_all(&config.codex_home).expect("create codex home"); + let global = codex_extension_api::Instructions { + text: "global instructions".to_string(), + source: None, + }; + let thread = codex_extension_api::Instructions { + text: "source thread instructions".to_string(), + source: None, + }; + let mut manager = ThreadManager::with_models_provider_and_home_for_tests( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.to_path_buf(), + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + ); + Arc::get_mut(&mut manager.state) + .expect("unshared manager") + .user_instructions_provider = Arc::new(ParentInstructionsProvider(global.clone())); + let source = manager + .start_thread(StartThreadOptions { + environments: Some(Vec::new()), + thread_instructions_provider: Some(Arc::new(ParentInstructionsProvider( + thread.clone(), + ))), + ..StartThreadOptions::new(config.clone()) + }) + .await + .expect("start source"); + let history = InitialHistory::Resumed(ResumedHistory { + conversation_id: source.thread_id, + history: Arc::new(Vec::new()), + rollout_path: None, + }); + + // Queue removal between the first source lookup and subsequent startup + // lookups. Tokio's fair RwLock makes this ordering deterministic. + let fork = manager.fork_thread_from_history( + ForkSnapshot::Interrupted, + StartThreadOptions { + environments: Some(Vec::new()), + ..StartThreadOptions::new(config) + }, + history, + ); + tokio::pin!(fork); + { + let _guard = manager.state.threads.write().await; + let mut context = std::task::Context::from_waker(std::task::Waker::noop()); + assert!(std::future::Future::poll(fork.as_mut(), &mut context).is_pending()); + } + let removal = manager.remove_thread_if_matches(&source.thread_id, &source.thread); + tokio::pin!(removal); + assert!(futures::poll!(&mut removal).is_pending()); + assert!(futures::poll!(&mut fork).is_pending()); + assert!(removal.await.is_some()); + let fork = fork.await.expect("fork survives source removal"); + let instructions = fork.thread.session.inherited_instructions().await; + assert_eq!( + (instructions.user, instructions.thread), + (Some(global), Some(thread)) + ); + source + .thread + .shutdown_and_wait() + .await + .expect("shutdown source"); + fork.thread + .shutdown_and_wait() + .await + .expect("shutdown fork"); +} + /// A thread opt-out wins over a shared client without disabling its siblings. #[tokio::test] async fn thread_analytics_opt_out_overrides_shared_client() { @@ -1149,19 +1246,6 @@ async fn spawn_internal_session_preserves_parent_lineage_without_forking_history } } - struct ParentInstructionsProvider(codex_extension_api::Instructions); - - impl codex_extension_api::UserInstructionsProvider for ParentInstructionsProvider { - fn load_user_instructions(&self) -> codex_extension_api::LoadUserInstructionsFuture<'_> { - Box::pin(async move { - codex_extension_api::LoadedUserInstructions { - instructions: Some(self.0.clone()), - warnings: Vec::new(), - } - }) - } - } - let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); @@ -1197,7 +1281,7 @@ async fn spawn_internal_session_preserves_parent_lineage_without_forking_history let parent_instructions = codex_extension_api::Instructions { text: "parent user instructions must not be inherited".to_string(), - source: config.codex_home.join("AGENTS.md"), + source: Some(config.codex_home.join("AGENTS.md")), }; let mut manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), @@ -1219,11 +1303,12 @@ async fn spawn_internal_session_preserves_parent_lineage_without_forking_history Arc::clone(&manager_state.extensions), manager_state.mcp_manager.codex_apps_tools_cache(), )); - manager_state.user_instructions_provider = - Arc::new(ParentInstructionsProvider(parent_instructions.clone())); + let parent_provider = Arc::new(ParentInstructionsProvider(parent_instructions.clone())); + manager_state.user_instructions_provider = parent_provider.clone(); let parent = manager .start_thread(StartThreadOptions { metrics_service_name: Some("codex_work_desktop".to_string()), + thread_instructions_provider: Some(parent_provider), ..StartThreadOptions::new(config.clone()) }) .await @@ -1232,9 +1317,10 @@ async fn spawn_internal_session_preserves_parent_lineage_without_forking_history .thread .session .set_multi_agent_version_if_unset(MultiAgentVersion::V2); + let parent_snapshot = parent.thread.session.inherited_instructions().await; assert_eq!( - parent.thread.session.inherited_instructions().await.user, - Some(parent_instructions) + (parent_snapshot.user, parent_snapshot.thread), + (Some(parent_instructions.clone()), Some(parent_instructions)) ); assert_eq!( parent @@ -1317,14 +1403,10 @@ 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 - .inherited_instructions() - .await - .user - .is_none() + let reviewer_snapshot = reviewer.thread.session.inherited_instructions().await; + assert_eq!( + (reviewer_snapshot.user, reviewer_snapshot.thread), + (None, None) ); assert!( reviewer diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index 419a4005e5..f81000b450 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -28,7 +28,7 @@ use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::ExecutorFileSystem; use codex_exec_server::RemoveOptions; use codex_extension_api::ExtensionRegistry; -use codex_extension_api::LoadUserInstructionsFuture; +use codex_extension_api::LoadInstructionsFuture; use codex_extension_api::UserInstructionsProvider; use codex_extension_api::empty_extension_registry; use codex_features::Feature; @@ -114,7 +114,7 @@ impl RecordingUserInstructionsProvider { } impl UserInstructionsProvider for RecordingUserInstructionsProvider { - fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_> { + fn load_user_instructions(&self) -> LoadInstructionsFuture<'_> { self.load_count.fetch_add(1, Ordering::SeqCst); self.inner.load_user_instructions() } diff --git a/codex-rs/core/tests/suite/agents_md.rs b/codex-rs/core/tests/suite/agents_md.rs index 4979b9df78..8c0ed2b731 100644 --- a/codex-rs/core/tests/suite/agents_md.rs +++ b/codex-rs/core/tests/suite/agents_md.rs @@ -6,12 +6,19 @@ 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::Instructions; +use codex_extension_api::LoadInstructionsFuture; +use codex_extension_api::LoadedUserInstructions; +use codex_extension_api::ThreadInstructionsProvider; use codex_extension_api::UserInstructionsProvider; use codex_features::Feature; +use codex_history::InitialHistory; +use codex_history::ResumedHistory; use codex_history::RolloutItem; use codex_home::CodexHomeUserInstructionsProvider; +use codex_protocol::ThreadId; use codex_protocol::config_types::TrustLevel; +use codex_protocol::error::CodexErrorDetails; use codex_protocol::models::PermissionProfile; use codex_protocol::permissions::FileSystemAccessMode; use codex_protocol::permissions::FileSystemSandboxEntry; @@ -20,11 +27,19 @@ use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::EnvironmentConfigState; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::Op; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::TurnEnvironmentSelection; +use codex_protocol::protocol::TurnEnvironmentSelections; +use codex_protocol::request_user_input::RequestUserInputAnswer; +use codex_protocol::request_user_input::RequestUserInputResponse; use codex_protocol::user_input::UserInput; +use codex_thread_store::ForkBoundary; +use codex_thread_store::LoadThreadHistoryParams; +use codex_thread_store::PrepareForkParams; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_path_uri::PathUri; +use codex_utils_string::approx_bytes_for_tokens; use core_test_support::PathBufExt; use core_test_support::create_directory_symlink; use core_test_support::load_default_config_for_test; @@ -39,6 +54,7 @@ use core_test_support::skip_if_no_remote_env; use core_test_support::skip_if_sandbox; use core_test_support::skip_if_target_windows; use core_test_support::test_codex::RecordingUserInstructionsProvider; +use core_test_support::test_codex::TestCodex; use core_test_support::test_codex::TestCodexBuilder; use core_test_support::test_codex::executor_path_uri; use core_test_support::test_codex::test_codex; @@ -46,9 +62,12 @@ use core_test_support::test_codex::turn_permission_fields; use core_test_support::wait_for_event; use pretty_assertions::assert_eq; use serde_json::json; +use std::collections::HashMap; use std::path::Path; use std::sync::Arc; +use std::sync::Mutex; use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::time::Duration; use tempfile::TempDir; @@ -64,6 +83,8 @@ const NEW_PROJECT_INSTRUCTIONS: &str = "new project instructions"; const OLD_GLOBAL_INSTRUCTIONS: &str = "old global instructions"; const PROJECT_INSTRUCTIONS: &str = "project instructions"; const PROJECT_SEPARATOR: &str = "--- project-doc ---"; +const TASK_USER_INSTRUCTIONS: &str = "task user instructions"; +const UPDATED_TASK_USER_INSTRUCTIONS: &str = "updated task user instructions"; const SPAWN_CALL_ID: &str = "spawn-global-instructions-child"; const SPAWN_CHILD_PROMPT: &str = "inspect inherited global instructions"; const SPAWN_FRESH_PARENT_PROMPT: &str = "spawn a child with fresh context"; @@ -77,7 +98,7 @@ struct WarningInstructionsProvider { } impl UserInstructionsProvider for WarningInstructionsProvider { - fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_> { + fn load_user_instructions(&self) -> LoadInstructionsFuture<'_> { Box::pin(async move { let mut loaded = self.inner.load_user_instructions().await; if self.warning_active.load(Ordering::SeqCst) { @@ -88,6 +109,60 @@ impl UserInstructionsProvider for WarningInstructionsProvider { } } +struct RecordingThreadInstructionsProvider { + loaded: Mutex, + load_count: AtomicUsize, +} + +impl RecordingThreadInstructionsProvider { + fn new(instructions: Option) -> Self { + Self { + loaded: Mutex::new(LoadedUserInstructions { + instructions, + warnings: Vec::new(), + }), + load_count: AtomicUsize::new(0), + } + } + + fn with_text(text: impl Into) -> Self { + Self::new(Some(Instructions { + text: text.into(), + source: None, + })) + } + + fn load_count(&self) -> usize { + self.load_count.load(Ordering::SeqCst) + } + + fn set_instructions(&self, instructions: Option) { + self.loaded + .lock() + .expect("instruction snapshot lock") + .instructions = instructions; + } + + fn set_warnings(&self, warnings: Vec) { + self.loaded + .lock() + .expect("instruction snapshot lock") + .warnings = warnings; + } +} + +impl ThreadInstructionsProvider for RecordingThreadInstructionsProvider { + fn load_thread_instructions(&self) -> LoadInstructionsFuture<'_> { + self.load_count.fetch_add(1, Ordering::SeqCst); + let loaded = self + .loaded + .lock() + .expect("instruction snapshot lock") + .clone(); + Box::pin(async move { loaded }) + } +} + async fn agents_instructions(mut builder: TestCodexBuilder) -> Result { let server = start_mock_server().await; let resp_mock = mount_sse_once( @@ -175,6 +250,32 @@ async fn submit_thread_turn(thread: &Arc, prompt: &str) Ok(()) } +async fn persisted_resume_history( + thread: &Arc, +) -> Result<(ThreadId, InitialHistory)> { + thread.ensure_rollout_materialized().await; + thread.flush_rollout().await?; + let stored = thread + .read_thread( + /*include_archived*/ true, /*include_history*/ true, + ) + .await?; + let thread_id = stored.thread_id; + Ok(( + thread_id, + InitialHistory::Resumed(ResumedHistory { + conversation_id: thread_id, + history: Arc::new( + stored + .history + .ok_or_else(|| anyhow!("thread history should be loaded"))? + .items, + ), + rollout_path: stored.rollout_path, + }), + )) +} + fn request_body_contains(request: &wiremock::Request, text: &str) -> bool { let is_zstd = request .headers @@ -919,6 +1020,561 @@ async fn loads_user_instructions_without_a_primary_environment() -> Result<()> { Ok(()) } +struct ThreadInstructionsFixture { + test: TestCodex, + thread: Arc, + provider: Arc, +} + +impl ThreadInstructionsFixture { + async fn new(server: &wiremock::MockServer) -> Result { + let home = Arc::new(TempDir::new()?); + write_global_file(home.as_ref(), GLOBAL_AGENTS_FILENAME, GLOBAL_INSTRUCTIONS)?; + let global_provider = Arc::new(RecordingUserInstructionsProvider::new(Arc::new( + CodexHomeUserInstructionsProvider::new(AbsolutePathBuf::try_from( + home.path().to_path_buf(), + )?), + ))); + let mut builder = test_codex() + .with_home(home) + .with_user_instructions_provider(global_provider.clone()) + .with_config(|config| { + config.project_doc_max_bytes = PROJECT_INSTRUCTIONS.len(); + config + .features + .enable(Feature::DefaultModeRequestUserInput) + .expect("test config should allow request-user-input feature"); + }) + .with_workspace_setup(|cwd, fs| async move { + let project_agents_uri = executor_path_uri(cwd.join(GLOBAL_AGENTS_FILENAME))?; + fs.write_file( + &project_agents_uri, + PROJECT_INSTRUCTIONS.as_bytes().to_vec(), + Default::default(), + /*sandbox*/ None, + ) + .await?; + Ok(()) + }); + let test = builder.build_with_auto_env(server).await?; + assert_eq!(global_provider.load_count(), 1); + let provider = Arc::new(RecordingThreadInstructionsProvider::with_text( + TASK_USER_INSTRUCTIONS, + )); + let thread = test + .thread_manager + .start_thread(StartThreadOptions { + environments: Some(vec![test.executor_environment().selection().clone()]), + thread_instructions_provider: Some(provider.clone()), + ..StartThreadOptions::new(test.config.clone()) + }) + .await? + .thread; + Ok(Self { + test, + thread, + provider, + }) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn thread_provider_composes_and_clears_only_its_instructions() -> Result<()> { + let server = start_mock_server().await; + let response_mock = responses::mount_sse_sequence( + &server, + ["initial", "unchanged", "cleared", "unchanged-empty"] + .map(|id| sse(vec![ev_response_created(id), ev_completed(id)])) + .to_vec(), + ) + .await; + let fixture = ThreadInstructionsFixture::new(&server).await?; + let sources = vec![ + PathUri::from_abs_path(&fixture.test.config.codex_home.join(GLOBAL_AGENTS_FILENAME)), + fixture.test.workspace_path_uri(GLOBAL_AGENTS_FILENAME)?, + ]; + assert_eq!(fixture.thread.instruction_sources().await, sources); + for prompt in [ + "inspect task instructions", + "inspect unchanged task instructions", + ] { + submit_thread_turn(&fixture.thread, prompt).await?; + } + for instructions in [ + Some(Instructions { + text: String::new(), + source: None, + }), + None, + ] { + fixture.provider.set_instructions(instructions); + submit_thread_turn(&fixture.thread, "inspect cleared task instructions").await?; + } + assert!(fixture.provider.load_count() > 1); + assert_eq!(fixture.thread.instruction_sources().await, sources); + + let cwd = &fixture.test.executor_environment().selection().cwd; + let initial = expected_instruction_fragment( + cwd, + &format!( + "{GLOBAL_INSTRUCTIONS}\n\n{TASK_USER_INSTRUCTIONS}\n\n{PROJECT_SEPARATOR}\n\n{PROJECT_INSTRUCTIONS}" + ), + ); + let cleared = expected_instruction_fragment( + cwd, + &format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{GLOBAL_INSTRUCTIONS}\n\n{PROJECT_SEPARATOR}\n\n{PROJECT_INSTRUCTIONS}" + ), + ); + assert_eq!( + response_mock + .requests() + .iter() + .map(instruction_fragments) + .collect::>(), + vec![ + vec![initial.clone()], + vec![initial.clone()], + vec![initial.clone(), cleared.clone()], + vec![initial, cleared], + ], + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn thread_provider_refreshes_at_the_next_step_of_an_active_turn() -> Result<()> { + let server = start_mock_server().await; + let response_mock = responses::mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("active-turn"), + responses::ev_function_call( + "pause-for-thread-instructions", + "request_user_input", + &json!({ + "questions": [{ + "id": "continue", + "header": "Continue", + "question": "Continue after updating thread instructions?", + "options": [{ + "label": "Yes (Recommended)", + "description": "Continue the current turn." + }, { + "label": "No", + "description": "Stop the current turn." + }] + }] + }) + .to_string(), + ), + ev_completed("active-turn"), + ]), + sse(vec![ + ev_response_created("updated"), + ev_completed("updated"), + ]), + ], + ) + .await; + let fixture = ThreadInstructionsFixture::new(&server).await?; + fixture + .thread + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "inspect instructions updated during the active turn".to_string(), + text_elements: Vec::new(), + }])) + .await?; + let EventMsg::RequestUserInput(request) = wait_for_event(&fixture.thread, |event| { + matches!(event, EventMsg::RequestUserInput(_)) + }) + .await + else { + unreachable!("wait_for_event should return the request-user-input event") + }; + // Only the host's provider changes; no explicit instruction update or resume call. + fixture.provider.set_instructions(Some(Instructions { + text: UPDATED_TASK_USER_INSTRUCTIONS.to_string(), + source: None, + })); + fixture + .thread + .submit(Op::UserInputAnswer { + id: request.turn_id, + response: RequestUserInputResponse { + answers: HashMap::from([( + "continue".to_string(), + RequestUserInputAnswer { + answers: vec!["Yes (Recommended)".to_string()], + }, + )]), + }, + }) + .await?; + wait_for_event(&fixture.thread, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let cwd = &fixture.test.executor_environment().selection().cwd; + let initial = expected_instruction_fragment( + cwd, + &format!( + "{GLOBAL_INSTRUCTIONS}\n\n{TASK_USER_INSTRUCTIONS}\n\n{PROJECT_SEPARATOR}\n\n{PROJECT_INSTRUCTIONS}" + ), + ); + let updated = expected_instruction_fragment( + cwd, + &format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{GLOBAL_INSTRUCTIONS}\n\n{UPDATED_TASK_USER_INSTRUCTIONS}\n\n{PROJECT_SEPARATOR}\n\n{PROJECT_INSTRUCTIONS}" + ), + ); + assert_eq!( + response_mock + .requests() + .iter() + .map(instruction_fragments) + .collect::>(), + vec![vec![initial.clone()], vec![initial, updated]], + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn thread_provider_enforces_its_own_limit_before_startup_and_sampling() -> Result<()> { + let server = start_mock_server().await; + let response_mock = responses::mount_sse_sequence( + &server, + ["initial", "full-thread-budget"] + .map(|id| sse(vec![ev_response_created(id), ev_completed(id)])) + .to_vec(), + ) + .await; + let fixture = ThreadInstructionsFixture::new(&server).await?; + let oversized = Instructions { + text: "x".repeat(approx_bytes_for_tokens(/*tokens*/ 10_001)), + source: None, + }; + let error = fixture + .test + .thread_manager + .start_thread(StartThreadOptions { + environments: Some(vec![ + fixture.test.executor_environment().selection().clone(), + ]), + thread_instructions_provider: Some(Arc::new(RecordingThreadInstructionsProvider::new( + Some(oversized.clone()), + ))), + ..StartThreadOptions::new(fixture.test.config.clone()) + }) + .await + .err() + .ok_or_else(|| anyhow!("oversized instructions must fail before thread creation"))?; + assert!(matches!( + error.details(), + CodexErrorDetails::InvalidRequest(_) + )); + assert!(error.to_string().contains("10000 estimated tokens")); + + submit_thread_turn(&fixture.thread, "inspect valid instructions").await?; + fixture.provider.set_instructions(Some(oversized)); + fixture + .provider + .set_warnings(vec!["provider refresh warning".to_string()]); + let request_count = response_mock.requests().len(); + fixture + .thread + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "reject oversized instructions before sampling".to_string(), + text_elements: Vec::new(), + }])) + .await?; + let EventMsg::Warning(warning) = wait_for_event(&fixture.thread, |event| { + matches!(event, EventMsg::Warning(_) | EventMsg::Error(_)) + }) + .await + else { + panic!("provider warnings must be emitted before validation errors"); + }; + assert_eq!(warning.message, "provider refresh warning"); + let EventMsg::Error(error) = + wait_for_event(&fixture.thread, |event| matches!(event, EventMsg::Error(_))).await + else { + unreachable!(); + }; + assert!(error.message.contains("10000 estimated tokens")); + wait_for_event(&fixture.thread, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + assert_eq!(response_mock.requests().len(), request_count); + fixture.provider.set_warnings(Vec::new()); + + // The full thread budget is independent of global instructions and wrapping. + // Changing environments must also remove the previously selected repository docs. + let host_text = "x".repeat(approx_bytes_for_tokens(/*tokens*/ 10_000)); + let expected = expected_provider_only_instruction_fragment(&format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{GLOBAL_INSTRUCTIONS}\n\n{host_text}" + )); + fixture.provider.set_instructions(Some(Instructions { + text: host_text, + source: None, + })); + fixture + .thread + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "inspect instructions without the previous environment".to_string(), + text_elements: Vec::new(), + }]) + .with_thread_settings(ThreadSettingsOverrides { + environments: Some(TurnEnvironmentSelections::new( + fixture.test.config.cwd.clone(), + Vec::new(), + )), + ..Default::default() + }), + ) + .await?; + wait_for_event(&fixture.thread, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let requests = response_mock.requests(); + assert_eq!(requests.len(), 2); + assert_eq!(instruction_fragments(&requests[1]).last(), Some(&expected)); + Ok(()) +} + +#[derive(Clone, Copy)] +enum InstructionForkSource { + LiveRollout, + OfflineHistory, + OfflinePrepared, +} + +#[test_case::test_case(InstructionForkSource::LiveRollout; "live snapshot")] +#[test_case::test_case(InstructionForkSource::OfflineHistory; "offline history provider")] +#[test_case::test_case(InstructionForkSource::OfflinePrepared; "offline prepared provider")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn fork_preserves_thread_instructions(source: InstructionForkSource) -> Result<()> { + let server = start_mock_server().await; + let response_mock = responses::mount_sse_sequence( + &server, + ["parent", "fork", "updated-fork"] + .map(|id| sse(vec![ev_response_created(id), ev_completed(id)])) + .to_vec(), + ) + .await; + let history_mode = match source { + InstructionForkSource::LiveRollout | InstructionForkSource::OfflineHistory => { + ThreadHistoryMode::Legacy + } + InstructionForkSource::OfflinePrepared => ThreadHistoryMode::Paginated, + }; + let mut builder = test_codex() + .with_config(|config| { + config + .features + .enable(Feature::Sqlite) + .expect("enable local persistence"); + }) + .with_history_mode(history_mode); + let test = builder.build_with_auto_env(&server).await?; + let parent_provider = Arc::new(RecordingThreadInstructionsProvider::with_text( + TASK_USER_INSTRUCTIONS, + )); + let parent = test + .thread_manager + .start_thread(StartThreadOptions { + environments: Some(Vec::new()), + thread_instructions_provider: Some(parent_provider.clone()), + history_mode: Some(history_mode), + ..StartThreadOptions::new(test.config.clone()) + }) + .await?; + submit_thread_turn(&parent.thread, "persist parent instructions").await?; + parent.thread.ensure_rollout_materialized().await; + parent.thread.flush_rollout().await?; + let parent_id = parent.thread_id; + let rollout_path = parent + .thread + .rollout_path() + .expect("persisted parent rollout"); + let parent_loads = parent_provider.load_count(); + let offline = !matches!(source, InstructionForkSource::LiveRollout); + if offline { + parent.thread.shutdown_and_wait().await?; + test.thread_manager + .remove_thread_if_matches(&parent_id, &parent.thread) + .await; + } + let fork_provider = Arc::new(RecordingThreadInstructionsProvider::with_text( + TASK_USER_INSTRUCTIONS, + )); + let options = StartThreadOptions { + environments: Some(Vec::new()), + thread_instructions_provider: offline + .then(|| fork_provider.clone() as Arc), + ..StartThreadOptions::new(test.config.clone()) + }; + let fork = match source { + InstructionForkSource::LiveRollout => { + test.thread_manager + .fork_thread(ForkSnapshot::Interrupted, options, rollout_path) + .await? + } + InstructionForkSource::OfflineHistory => { + let stored = test + .thread_store + .load_latest_model_context(LoadThreadHistoryParams { + thread_id: parent_id, + include_archived: true, + }) + .await?; + let history = InitialHistory::Resumed(ResumedHistory { + conversation_id: parent_id, + history: Arc::new(stored.items), + rollout_path: None, + }); + test.thread_manager + .fork_thread_from_history(ForkSnapshot::Interrupted, options, history) + .await? + } + InstructionForkSource::OfflinePrepared => { + let prepared = test + .thread_store + .prepare_fork(PrepareForkParams { + thread_id: parent_id, + boundary: ForkBoundary::Latest, + }) + .await?; + test.thread_manager + .fork_prepared_thread(options, prepared) + .await? + } + }; + submit_thread_turn(&fork.thread, "continue with inherited instructions").await?; + // Live forks inherit only a snapshot; offline forks retain their own provider. + // Neither path may call the original task's provider. + for provider in [&parent_provider, &fork_provider] { + provider.set_instructions(Some(Instructions { + text: UPDATED_TASK_USER_INSTRUCTIONS.to_string(), + source: None, + })); + } + submit_thread_turn(&fork.thread, "continue after source update").await?; + assert_eq!(parent_provider.load_count(), parent_loads); + let initial = expected_provider_only_instruction_fragment(TASK_USER_INSTRUCTIONS); + let final_fragments = if offline { + vec![ + initial.clone(), + expected_provider_only_instruction_fragment(&format!( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{UPDATED_TASK_USER_INSTRUCTIONS}" + )), + ] + } else { + vec![initial.clone()] + }; + assert_eq!( + response_mock + .requests() + .iter() + .map(instruction_fragments) + .collect::>(), + vec![vec![initial.clone()], vec![initial], final_fragments] + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn thread_provider_lives_with_its_session_across_resume() -> Result<()> { + let server = start_mock_server().await; + let response_mock = responses::mount_sse_sequence( + &server, + ["updated-live-session", "cold-session"] + .map(|id| sse(vec![ev_response_created(id), ev_completed(id)])) + .to_vec(), + ) + .await; + let mut builder = test_codex(); + let test = builder.build_with_auto_env(&server).await?; + let provider = Arc::new(RecordingThreadInstructionsProvider::new( + /*instructions*/ None, + )); + let started = test + .thread_manager + .start_thread(StartThreadOptions { + environments: Some(Vec::new()), + thread_instructions_provider: Some(provider.clone()), + ..StartThreadOptions::new(test.config.clone()) + }) + .await?; + assert_eq!(provider.load_count(), 1); + let (_, history) = persisted_resume_history(&started.thread).await?; + let cold_provider = Arc::new(RecordingThreadInstructionsProvider::with_text( + "cold session instructions", + )); + provider.set_instructions(Some(Instructions { + text: UPDATED_TASK_USER_INSTRUCTIONS.to_string(), + source: None, + })); + let same_provider = Arc::new(RecordingThreadInstructionsProvider::with_text( + UPDATED_TASK_USER_INSTRUCTIONS, + )); + for supplied in [ + None, + Some(same_provider.clone() as Arc), + Some(cold_provider.clone() as Arc), + ] { + let resumed = test + .thread_manager + .start_thread(StartThreadOptions { + initial_history: history.clone(), + thread_instructions_provider: supplied, + ..StartThreadOptions::new(test.config.clone()) + }) + .await?; + assert!(Arc::ptr_eq(&resumed.thread, &started.thread)); + assert_eq!(provider.load_count(), 1); + assert_eq!(same_provider.load_count(), 0); + assert_eq!(cold_provider.load_count(), 0); + } + + // Reattachment does no loading; the existing Session pulls at the next model step. + submit_thread_turn(&started.thread, "load instructions added after creation").await?; + assert_eq!(provider.load_count(), 2); + assert_eq!(same_provider.load_count(), 0); + let (_, history) = persisted_resume_history(&started.thread).await?; + started.thread.shutdown_and_wait().await?; + let resumed = test + .thread_manager + .start_thread(StartThreadOptions { + initial_history: history, + environments: Some(Vec::new()), + thread_instructions_provider: Some(cold_provider.clone()), + ..StartThreadOptions::new(test.config.clone()) + }) + .await?; + assert!(!Arc::ptr_eq(&resumed.thread, &started.thread)); + assert_eq!(cold_provider.load_count(), 1); + submit_thread_turn(&resumed.thread, "load cold session instructions").await?; + assert_eq!(provider.load_count(), 2); + let requests = response_mock.requests(); + assert_single_instruction_fragment( + &requests[0], + &expected_provider_only_instruction_fragment(UPDATED_TASK_USER_INSTRUCTIONS), + ); + assert_eq!( + instruction_fragments(&requests[1]).last(), + Some(&expected_provider_only_instruction_fragment( + "These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\ncold session instructions", + )), + ); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn fresh_thread_composes_global_before_project_and_reports_sources() -> Result<()> { // Set up one global source, one project source, and two ordinary model turns. diff --git a/codex-rs/core/tests/suite/agents_md_refresh.rs b/codex-rs/core/tests/suite/agents_md_refresh.rs index dfd6c7f4e6..3155f6a14d 100644 --- a/codex-rs/core/tests/suite/agents_md_refresh.rs +++ b/codex-rs/core/tests/suite/agents_md_refresh.rs @@ -207,7 +207,7 @@ async fn interrupting_a_provider_read_allows_the_next_turn_to_refresh() -> Resul started: Notify, } impl UserInstructionsProvider for GatedProvider { - fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_> { + fn load_user_instructions(&self) -> LoadInstructionsFuture<'_> { Box::pin(async move { if self.block_next.swap(/*val*/ false, Ordering::SeqCst) { self.started.notify_one(); diff --git a/codex-rs/ext/extension-api/src/lib.rs b/codex-rs/ext/extension-api/src/lib.rs index 3daadfe87a..d92e5c6a65 100644 --- a/codex-rs/ext/extension-api/src/lib.rs +++ b/codex-rs/ext/extension-api/src/lib.rs @@ -104,6 +104,7 @@ pub use state::ExtensionData; pub use state::ExtensionDataInit; pub use turn_admission::TurnStartAdmission; pub use user_instructions::Instructions; -pub use user_instructions::LoadUserInstructionsFuture; +pub use user_instructions::LoadInstructionsFuture; pub use user_instructions::LoadedUserInstructions; +pub use user_instructions::ThreadInstructionsProvider; pub use user_instructions::UserInstructionsProvider; diff --git a/codex-rs/ext/extension-api/src/user_instructions.rs b/codex-rs/ext/extension-api/src/user_instructions.rs index 5e262722b3..b6c9a4b36f 100644 --- a/codex-rs/ext/extension-api/src/user_instructions.rs +++ b/codex-rs/ext/extension-api/src/user_instructions.rs @@ -5,17 +5,17 @@ use codex_utils_absolute_path::AbsolutePathBuf; /// Instructions supplied by the host. /// -/// `source` must be an absolute filesystem path because the app-server -/// `instructionSources` API currently exposes instruction sources as -/// `AbsolutePathBuf` values. +/// Filesystem-backed instructions retain their absolute source path for the +/// app-server `instructionSources` API. Other host-provided instructions do +/// not report a filesystem source. // TODO(anp): Replace the absolute path with a more general instruction-source // abstraction when non-filesystem providers need first-class attribution. #[derive(Clone, Debug, Eq, PartialEq)] pub struct Instructions { /// Model-visible instruction text. pub text: String, - /// Absolute filesystem path reported through `instructionSources`. - pub source: AbsolutePathBuf, + /// Absolute filesystem path reported through `instructionSources`, if any. + pub source: Option, } /// Result of loading host-provided user instructions. @@ -28,10 +28,24 @@ pub struct LoadedUserInstructions { pub warnings: Vec, } -/// Future returned by a [`UserInstructionsProvider`]. -pub type LoadUserInstructionsFuture<'a> = +/// Future returned by an instruction provider. +pub type LoadInstructionsFuture<'a> = Pin + Send + 'a>>; +/// Loads host-provided instructions that apply to one root thread. +/// +/// These instructions follow the global [`UserInstructionsProvider`] snapshot +/// and precede repository instructions. A result with no instructions or +/// blank instructions clears only the thread-scoped contribution. Core retains +/// the provider and reads it at startup and when capturing model-request context. +/// Implementations own fetching and caching; repeated reads should be cheap and +/// return a coherent snapshot. On a recoverable fetch failure, return the last +/// usable snapshot with warnings rather than an empty result that clears it. +pub trait ThreadInstructionsProvider: Send + Sync { + /// Loads the current snapshot for the provider's root thread. + fn load_thread_instructions(&self) -> LoadInstructionsFuture<'_>; +} + /// Loads global user instructions shared across root threads. /// /// Core reads this provider at startup and when capturing model-request context. @@ -40,5 +54,5 @@ pub type LoadUserInstructionsFuture<'a> = /// while still returning usable fallback instructions when available. pub trait UserInstructionsProvider: Send + Sync { /// Loads the current global snapshot for a root runtime. - fn load_user_instructions(&self) -> LoadUserInstructionsFuture<'_>; + fn load_user_instructions(&self) -> LoadInstructionsFuture<'_>; }