github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 07:42:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add a provider for thread-scoped instructions (#44701)

Browse Source 
## What changed

- Expose `ThreadInstructionsProvider` through `StartThreadOptions`. Load its snapshot at startup and model-request boundaries, composing it after global instructions and before repository instructions. Empty or blank output clears only the thread contribution.
- Reject thread instructions exceeding 10,000 estimated tokens independently of the repository instruction budget. Allow host-provided instructions without a filesystem source and rename the shared future type to `LoadInstructionsFuture`.
- Retain the provider across warm resumes; require hosts to supply it again for cold resumes and offline forks. Live forks and subagents inherit applied snapshots without inheriting the source thread's provider, including when the parent is unloaded during setup or reload.
- Include thread instructions in guardian reviewer inheritance and reuse decisions.

## Testing

Add coverage for composition and clearing, refresh within an active turn, size rejection before sampling, cancellation-safe refresh, fork and resume behavior, parent eviction, and reviewer reuse invalidation.

GitOrigin-RevId: 7be9a523cbbfd67704067dfd526188dad89a3c88
This commit is contained in:
vkg-oai
2026-09-11 01:57:54 +00:00
committed by copyberry
parent dc55274818
commit fc948f8c47
21 changed files with 1303 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/codex-home/src/instructions/mod.rs
+3 -3
View File
@@ -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
codex-rs/codex-home/src/instructions/tests.rs
+4 -2
View File
@@ -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,
}
codex-rs/core-api/src/lib.rs
+2 -1
View File
@@ -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;
codex-rs/core/src/agent/control/spawn.rs
+13 -1
View File
@@ -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,
})
codex-rs/core/src/agent/control_tests.rs
+139
View File
@@ -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::<ThreadId>::new());
assert_eq!(report.timed_out, Vec::<ThreadId>::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;
codex-rs/core/src/agents_md.rs
+31 -6
View File
@@ -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<Instructions>,
/// Thread-scoped instructions supplied by the host.
thread_instructions: Option<Instructions>,
/// Ordered instructions and their provenance.
entries: Vec<InstructionEntry>,
}
@@ -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<Instructions>,
thread_instructions: Option<Instructions>,
) -> Option<Self> {
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<Item = PathUri> + '_ {
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()
codex-rs/core/src/agents_md_manager.rs
+35 -3
View File
@@ -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<Instructions>,
pub(crate) thread: Option<Instructions>,
pub(crate) user_provider: Option<Arc<dyn UserInstructionsProvider>>,
pub(crate) thread_provider: Option<Arc<dyn ThreadInstructionsProvider>>,
}
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<Instructions>) -> Option<Instructions> {
instructions.filter(|instructions| !instructions.text.trim().is_empty())
}
codex-rs/core/src/agents_md_tests.rs
+18 -9
View File
@@ -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<_>>(),
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()),
codex-rs/core/src/guardian/review_session.rs
+2
View File
@@ -222,6 +222,7 @@ pub struct GuardianReviewSessionReuseKey {
developer_instructions: Option<String>,
base_instructions: Option<String>,
user_instructions: Option<Instructions>,
thread_instructions: Option<Instructions>,
compact_prompt: Option<String>,
cwd: PathUri,
mcp_servers: Constrained<HashMap<String, McpServerConfig>>,
@@ -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(),
codex-rs/core/src/guardian/review_session_factory.rs
+1
View File
@@ -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 {
codex-rs/core/src/guardian/review_session_tests.rs
+28 -10
View File
@@ -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(
&params.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()
}));
codex-rs/core/src/guardian/review_session_threads.rs
+15 -16
View File
@@ -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(
codex-rs/core/src/session/step_activation_tests.rs
+127 -29
View File
@@ -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<Option<Instructions>>,
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<Option<Instructions>>,
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<ModelInfo> {
let model = bundled_models_response()
.expect("bundled models")
codex-rs/core/src/test_support.rs
+2 -2
View File
@@ -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() })
}
}
codex-rs/core/src/thread_manager.rs
+90 -31
View File
@@ -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<Arc<dyn ThreadInstructionsProvider>>,
pub allow_provider_model_fallback: bool,
pub initial_history: InitialHistory,
pub history_mode: Option<ThreadHistoryMode>,
@@ -242,7 +256,7 @@ pub struct StartThreadOptions {
pub environments: Option<Vec<TurnEnvironmentSelection>>,
/// Existing environment bindings captured by an internal caller.
pub inherited_environments: Option<TurnEnvironmentSnapshot>,
/// 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<LoadedUserInstructions>,
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<ThreadId>,
fork_persistence: ForkPersistence,
inherited_environments: Option<TurnEnvironmentSnapshot>,
inherited_instructions: Option<SessionInstructions>,
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
user_shell_override: Option<crate::shell::Shell>,
}
@@ -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<ThreadId>,
pub(crate) environment_selections: Option<Vec<TurnEnvironmentSelection>>,
pub(crate) inherited_environments: Option<TurnEnvironmentSnapshot>,
pub(crate) inherited_instructions: Option<SessionInstructions>,
pub(crate) inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
pub(crate) client_mcp_extensions: Option<ClientMcpExtensions>,
}
@@ -978,8 +996,13 @@ impl ThreadManager {
parent_thread_id: ThreadId,
options: StartThreadOptions,
) -> CodexResult<NewThread> {
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<SessionInstructions>,
) -> CodexResult<NewThread> {
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<S>(
&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<ThreadId>,
forked_from_thread_id: Option<ThreadId>,
thread_provider: Option<Arc<dyn ThreadInstructionsProvider>>,
) -> 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),
codex-rs/core/src/thread_manager_tests.rs
+108 -26
View File
@@ -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
codex-rs/core/tests/common/test_codex.rs
+2 -2
View File
@@ -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()
}
codex-rs/core/tests/suite/agents_md.rs
+658 -2
View File
@@ -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<LoadedUserInstructions>,
load_count: AtomicUsize,
}
impl RecordingThreadInstructionsProvider {
fn new(instructions: Option<Instructions>) -> Self {
Self {
loaded: Mutex::new(LoadedUserInstructions {
instructions,
warnings: Vec::new(),
}),
load_count: AtomicUsize::new(0),
}
}
fn with_text(text: impl Into<String>) -> 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<Instructions>) {
self.loaded
.lock()
.expect("instruction snapshot lock")
.instructions = instructions;
}
fn set_warnings(&self, warnings: Vec<String>) {
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<String> {
let server = start_mock_server().await;
let resp_mock = mount_sse_once(
@@ -175,6 +250,32 @@ async fn submit_thread_turn(thread: &Arc<codex_core::CodexThread>, prompt: &str)
Ok(())
}
async fn persisted_resume_history(
thread: &Arc<codex_core::CodexThread>,
) -> 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<codex_core::CodexThread>,
provider: Arc<RecordingThreadInstructionsProvider>,
}
impl ThreadInstructionsFixture {
async fn new(server: &wiremock::MockServer) -> Result<Self> {
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![
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![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<dyn ThreadInstructionsProvider>),
..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![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<dyn ThreadInstructionsProvider>),
Some(cold_provider.clone() as Arc<dyn ThreadInstructionsProvider>),
] {
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.
codex-rs/core/tests/suite/agents_md_refresh.rs
+1 -1
View File
@@ -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();
codex-rs/ext/extension-api/src/lib.rs
+2 -1
View File
@@ -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;
codex-rs/ext/extension-api/src/user_instructions.rs
+22 -8
View File
@@ -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<AbsolutePathBuf>,
}
/// Result of loading host-provided user instructions.
@@ -28,10 +28,24 @@ pub struct LoadedUserInstructions {
pub warnings: Vec<String>,
}
/// Future returned by a [`UserInstructionsProvider`].
pub type LoadUserInstructionsFuture<'a> =
/// Future returned by an instruction provider.
pub type LoadInstructionsFuture<'a> =
Pin<Box<dyn Future<Output = LoadedUserInstructions> + 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<'_>;
}