Allow thread instruction providers to share updates with subagents (#46577)

## Why

Subagents inherit applied instruction snapshots, so later provider updates do not reach running descendants. Hosts need an explicit way to share current thread instructions across an agent tree.

## What changed

- Add `ThreadInstructionsProvider::share_with_subagents()`, defaulting to `false`. Opted-in descendants refresh instructions at their next model-request boundary.
- Retain shared providers across root unloads and descendant reloads. Let a resumed root replace the provider for surviving descendants, or preserve their last shared instructions when it supplies a non-sharing provider.
- Keep providers scoped to their root tree and preserve snapshot-only instructions for root forks and isolated sessions, including Guardian reviews.

## Testing

Add coverage for child and grandchild instruction updates and clearing, provider replacement after root restart, lazy descendant reloads, default snapshot inheritance, and isolation between roots and Guardian sessions.

GitOrigin-RevId: 44c9fb7fe9ae1f96156bdf3c47ab6c4a81bd65e9
This commit is contained in:
vinh-gpt
2026-09-19 02:15:20 +00:00
committed by copyberry
parent fd59b42a09
commit e24ac448b3
14 changed files with 864 additions and 45 deletions
+25
View File
@@ -26,6 +26,7 @@ use crate::thread_manager::default_thread_id_generator;
use crate::thread_rollout_truncation::truncate_rollout_to_last_n_fork_turns;
use crate::turn_timing::now_unix_timestamp_ms;
use arc_swap::ArcSwapOption;
use codex_extension_api::ThreadInstructionsProvider;
use codex_history::InitialHistory;
use codex_history::ResumedHistory;
use codex_history::RolloutItem;
@@ -59,6 +60,7 @@ use std::collections::HashMap;
use std::collections::HashSet;
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::Weak;
use tokio::sync::watch;
use tracing::warn;
@@ -108,6 +110,8 @@ pub(crate) struct LocalAgentControl {
rollout_budget: Arc<RolloutBudget>,
/// The user-selected root routing tier, shared by the entire agent tree.
root_service_tier: Arc<ArcSwapOption<String>>,
/// Retains the root's opt-in instruction provider even when the root is unloaded.
shared_thread_instructions_provider: Arc<OnceLock<Arc<dyn ThreadInstructionsProvider>>>,
}
impl Default for LocalAgentControl {
@@ -136,6 +140,7 @@ impl LocalAgentControl {
agent_execution_limiter: Arc::default(),
rollout_budget: Arc::default(),
root_service_tier: Arc::new(ArcSwapOption::from(None)),
shared_thread_instructions_provider: Arc::default(),
};
if let Some(rollout_budget) = rollout_budget {
control.rollout_budget.configure(rollout_budget);
@@ -157,6 +162,26 @@ impl LocalAgentControl {
(self.thread_id_generator)()
}
pub(crate) fn root_thread_instructions_provider(
&self,
root_thread_id: ThreadId,
provider: Option<Arc<dyn ThreadInstructionsProvider>>,
) -> Option<Arc<dyn ThreadInstructionsProvider>> {
let provider = match self.manager.upgrade() {
Some(manager) => manager.shared_thread_instructions_provider(root_thread_id, provider),
None => provider,
};
if let Some(provider) = provider
.as_ref()
.filter(|provider| provider.share_with_subagents())
{
let _ = self
.shared_thread_instructions_provider
.set(Arc::clone(provider));
}
provider
}
/// Send rich user input items to an existing agent thread.
pub(crate) async fn send_input(
&self,
+7 -1
View File
@@ -6,6 +6,7 @@ use crate::agent::types::AgentMetadata;
use crate::agent::types::LiveAgent;
use crate::agent::types::SpawnAgentForkMode;
use crate::agent::types::SpawnAgentOptions;
use crate::agents_md_manager::SessionInstructions;
use crate::codex_thread::CodexThread;
use crate::config::PermissionProfileSnapshot;
use crate::context::ContextualUserFragment;
@@ -576,7 +577,12 @@ impl LocalAgentControl {
{
Some(parent.session.inherited_instructions().await)
} else {
None
self.shared_thread_instructions_provider
.get()
.map(|provider| SessionInstructions {
thread_provider: Some(Arc::clone(provider)),
..Default::default()
})
};
// 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.
+261 -21
View File
@@ -4,7 +4,9 @@ use crate::StateDbHandle;
use crate::ThreadManager;
use crate::agent::agent_status_from_event;
use crate::agent::next_thread_spawn_depth;
use crate::agent::types::AgentMessage;
use crate::agent::types::LiveAgent;
use crate::agent::types::MessageDeliveryMode;
use crate::agent::types::ResolvedMultiAgentV2UsageHints;
use crate::agent::types::SpawnAgentForkMode;
use crate::agent::types::SpawnAgentOptions;
@@ -82,10 +84,12 @@ use codex_thread_store::ThreadStore;
use codex_utils_path_uri::PathUri;
use core_test_support::responses::strip_response_item_ids;
use pretty_assertions::assert_eq;
use std::sync::RwLock;
use tempfile::TempDir;
use tokio::time::Duration;
use tokio::time::sleep;
use tokio::time::timeout;
use tokio_util::sync::CancellationToken;
use toml::Value as TomlValue;
async fn test_config_with_cli_overrides(
@@ -807,9 +811,10 @@ async fn check_v2_agent_reload(route: V2ReloadRoute) {
}),
warnings: Vec::new(),
}),
thread_instructions_provider: Some(Arc::new(StaticThreadInstructionsProvider(
"thread instructions survive parent eviction",
))),
thread_instructions_provider: Some(Arc::new(TestThreadInstructionsProvider {
text: "thread instructions survive parent eviction".into(),
shared: false,
})),
..StartThreadOptions::new(harness.config.clone())
})
.await
@@ -1140,12 +1145,23 @@ 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);
struct TestThreadInstructionsProvider {
text: RwLock<&'static str>,
shared: bool,
}
impl ThreadInstructionsProvider for TestThreadInstructionsProvider {
fn share_with_subagents(&self) -> bool {
self.shared
}
impl ThreadInstructionsProvider for StaticThreadInstructionsProvider {
fn load_thread_instructions(&self) -> LoadInstructionsFuture<'_> {
let instructions = Instructions {
text: self.0.to_string(),
text: self
.text
.read()
.expect("read thread instructions")
.to_string(),
source: None,
};
Box::pin(async move {
@@ -1157,8 +1173,15 @@ impl ThreadInstructionsProvider for StaticThreadInstructionsProvider {
}
}
#[test_case::test_case(false, false; "snapshot")]
#[test_case::test_case(true, false; "shared")]
#[test_case::test_case(false, true; "unshared_grandchild_without_loaded_parent")]
#[test_case::test_case(true, true; "shared_grandchild_without_loaded_parent")]
#[tokio::test]
async fn cold_resume_with_thread_instructions_preserves_lazy_v2_child_inheritance() {
async fn cold_resume_with_thread_instructions_preserves_lazy_v2_child_inheritance(
shared: bool,
nested: bool,
) {
let (home, mut config) = test_config().await;
let _ = config.features.enable(Feature::MultiAgentV2);
let _ = config.features.enable(Feature::Sqlite);
@@ -1166,9 +1189,10 @@ async fn cold_resume_with_thread_instructions_preserves_lazy_v2_child_inheritanc
let parent = harness
.manager
.start_thread(StartThreadOptions {
thread_instructions_provider: Some(Arc::new(StaticThreadInstructionsProvider(
"initial thread instructions",
))),
thread_instructions_provider: Some(Arc::new(TestThreadInstructionsProvider {
text: "initial thread instructions".into(),
shared,
})),
..StartThreadOptions::new(harness.config.clone())
})
.await
@@ -1188,6 +1212,23 @@ async fn cold_resume_with_thread_instructions_preserves_lazy_v2_child_inheritanc
persist_thread_for_tree_resume(&worker, "worker persisted").await;
wait_for_live_thread_spawn_children(control, parent_thread_id, &[worker_thread_id]).await;
let target_thread_id = if nested {
let grandchild_id =
spawn_v2_reload_test_child(control, harness.config.clone(), &worker, "reviewer")
.await
.thread_id;
let grandchild = harness
.manager
.get_thread(grandchild_id)
.await
.expect("grandchild is loaded");
persist_thread_for_tree_resume(&grandchild, "grandchild persisted").await;
wait_for_live_thread_spawn_children(control, worker_thread_id, &[grandchild_id]).await;
grandchild_id
} else {
worker_thread_id
};
let stored_parent = parent
.thread
.read_thread(
@@ -1214,13 +1255,15 @@ async fn cold_resume_with_thread_instructions_preserves_lazy_v2_child_inheritanc
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
harness.state_db.clone(),
);
let provider: Arc<dyn ThreadInstructionsProvider> = Arc::new(TestThreadInstructionsProvider {
text: "resumed thread instructions".into(),
shared,
});
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",
))),
thread_instructions_provider: Some(provider.clone()),
..StartThreadOptions::new(harness.config.clone())
})
.await
@@ -1240,17 +1283,214 @@ async fn cold_resume_with_thread_instructions_preserves_lazy_v2_child_inheritanc
);
assert_thread_not_loaded(&resumed_manager, worker_thread_id).await;
resumed_manager
.ensure_multi_agent_v2_child_loaded(worker_thread_id)
if nested {
let turn = resumed_parent.thread.session.new_default_turn().await;
resumed_parent
.thread
.session
.services
.agent_control
.deliver_message(
parent_thread_id,
&turn,
target_thread_id,
AgentMessage::Plaintext("hello after resume".to_string()),
MessageDeliveryMode::QueueOnly,
)
.await
.expect("message should reload the grandchild");
assert_thread_not_loaded(&resumed_manager, worker_thread_id).await;
} else {
resumed_manager
.ensure_multi_agent_v2_child_loaded(worker_thread_id)
.await
.expect("resume v2 worker on demand");
}
let resumed_target = resumed_manager
.get_thread(target_thread_id)
.await
.expect("resume v2 worker on demand");
let resumed_worker = resumed_manager
.get_thread(worker_thread_id)
.expect("resumed target should be loaded");
let inherited = resumed_target.session.inherited_instructions().await;
let root_provider = resumed_parent
.thread
.session
.inherited_instructions()
.await
.expect("resumed worker should be loaded");
.thread_provider;
assert_eq!(
resumed_worker.session.inherited_instructions().await.thread,
Some(thread_instructions),
inherited
.thread_provider
.as_ref()
.map(|inherited| root_provider
.as_ref()
.is_some_and(|root| Arc::ptr_eq(inherited, root))),
shared.then_some(true),
);
assert_eq!(
inherited.thread,
(!nested || shared).then_some(thread_instructions),
);
}
#[test_case::test_case(false; "snapshot_is_not_shared")]
#[test_case::test_case(true; "shared_provider_survives")]
#[tokio::test]
async fn v2_sibling_reload_preserves_shared_instructions_after_root_unloads(shared: bool) {
let (home, mut config) = test_config().await;
config
.features
.enable(Feature::MultiAgentV2)
.expect("enable multi-agent v2");
config
.features
.enable(Feature::Sqlite)
.expect("enable SQLite");
let harness = AgentControlHarness::new_with_config(home, config).await;
let provider = Arc::new(TestThreadInstructionsProvider {
text: "initial thread instructions".into(),
shared,
});
let root = harness
.manager
.start_thread(StartThreadOptions {
thread_instructions_provider: Some(provider.clone()),
..StartThreadOptions::new(harness.config.clone())
})
.await
.expect("start root");
let expected_provider = root
.thread
.session
.inherited_instructions()
.await
.thread_provider;
let control = &root.thread.session.services.agent_control;
let sender_id =
spawn_v2_reload_test_child(control, harness.config.clone(), &root.thread, "sender")
.await
.thread_id;
let target_id =
spawn_v2_reload_test_child(control, harness.config.clone(), &root.thread, "target")
.await
.thread_id;
let sender = harness
.manager
.get_thread(sender_id)
.await
.expect("loaded sender");
let target = harness
.manager
.get_thread(target_id)
.await
.expect("loaded target");
persist_thread_for_tree_resume(&root.thread, "root persisted").await;
persist_thread_for_tree_resume(&sender, "sender persisted").await;
persist_thread_for_tree_resume(&target, "target persisted").await;
target.shutdown_and_wait().await.expect("shut down target");
assert!(harness.manager.remove_thread(&target_id).await.is_some());
assert!(
harness
.manager
.remove_thread(&root.thread_id)
.await
.is_some()
);
assert_thread_not_loaded(&harness.manager, root.thread_id).await;
drop(root);
drop(target);
*provider.text.write().expect("update thread instructions") =
"updated while target was unloaded";
let sender_turn = sender.session.new_default_turn().await;
sender
.session
.services
.agent_control
.deliver_message(
sender_id,
&sender_turn,
target_id,
AgentMessage::Plaintext("wake the sibling".to_string()),
MessageDeliveryMode::QueueOnly,
)
.await
.expect("reload target from its sibling");
let resumed = harness
.manager
.get_thread(target_id)
.await
.expect("reloaded target");
let inherited = resumed.session.inherited_instructions().await;
assert_eq!(
inherited
.thread_provider
.as_ref()
.map(|provider| expected_provider
.as_ref()
.is_some_and(|root| Arc::ptr_eq(provider, root))),
shared.then_some(true)
);
assert_eq!(
inherited.thread.map(|instructions| instructions.text),
shared.then(|| "updated while target was unloaded".to_string())
);
*provider.text.write().expect("update thread instructions") =
"updated after target was reloaded";
resumed
.session
.capture_step_context(
resumed.session.new_default_turn().await,
&CancellationToken::new(),
)
.await
.expect("capture next request");
assert_eq!(
resumed
.session
.inherited_instructions()
.await
.thread
.map(|instructions| instructions.text),
shared.then(|| "updated after target was reloaded".to_string())
);
}
#[tokio::test]
async fn resumed_root_reuses_or_freezes_surviving_shared_instructions() {
let harness = AgentControlHarness::new().await;
let id = ThreadId::new();
let original = Arc::new(TestThreadInstructionsProvider {
text: "shared snapshot".into(),
shared: true,
});
let shared = harness
.manager
.agent_control()
.root_thread_instructions_provider(id, Some(original.clone()))
.expect("shared provider");
let snapshot = shared.load_thread_instructions().await;
let reused = harness
.manager
.agent_control()
.root_thread_instructions_provider(id, /*provider*/ None)
.expect("reuse the live tree's provider");
assert!(Arc::ptr_eq(&shared, &reused));
let private = Arc::new(TestThreadInstructionsProvider {
text: "private replacement".into(),
shared: false,
});
let root_only = harness
.manager
.agent_control()
.root_thread_instructions_provider(id, Some(private.clone()))
.expect("private root provider");
*original.text.write().expect("update old provider") = "stale shared update";
assert_eq!(shared.load_thread_instructions().await, snapshot);
assert_eq!(
root_only.load_thread_instructions().await,
private.load_thread_instructions().await
);
}
+6 -1
View File
@@ -24,7 +24,7 @@ pub(crate) struct AgentsMdManager {
state: Mutex<AgentsMdState>,
}
/// Root runtimes retain providers; subagents inherit only their applied snapshots.
/// Subagents inherit applied snapshots, plus thread providers that opt into sharing.
#[derive(Clone, Default)]
pub(crate) struct SessionInstructions {
pub(crate) user: Option<Instructions>,
@@ -152,6 +152,11 @@ impl AgentsMdManager {
SessionInstructions {
user: state.instructions.user.clone(),
thread: state.instructions.thread.clone(),
thread_provider: state
.instructions
.thread_provider
.clone()
.filter(|provider| provider.share_with_subagents()),
..Default::default()
}
}
@@ -261,6 +261,17 @@ async fn test_review_params() -> GuardianReviewSessionParams {
#[tokio::test]
async fn spawned_guardian_reuse_key_matches_inherited_instructions() {
struct SharedProvider;
impl codex_extension_api::ThreadInstructionsProvider for SharedProvider {
fn share_with_subagents(&self) -> bool {
true
}
fn load_thread_instructions(&self) -> codex_extension_api::LoadInstructionsFuture<'_> {
panic!("isolated reviewers must not load the parent's provider")
}
}
let mut params = test_review_params().await;
let latest = Some(Instructions {
text: "latest thread instructions".to_string(),
@@ -274,6 +285,7 @@ async fn spawned_guardian_reuse_key_matches_inherited_instructions() {
parent.services.agents_md_manager = Arc::new(AgentsMdManager::new(SessionInstructions {
user: latest_global.clone(),
thread: latest.clone(),
thread_provider: Some(Arc::new(SharedProvider)),
..Default::default()
}));
// Reproduce an update between reuse-key capture and reviewer creation.
@@ -306,7 +318,9 @@ async fn spawned_guardian_reuse_key_matches_inherited_instructions() {
let review = manager.trunk().await.expect("prewarmed reviewer");
assert_eq!(review.reuse_key, expected_key);
assert_eq!(review.session.inherited_instructions().await.thread, latest);
let inherited = review.session.inherited_instructions().await;
assert_eq!(inherited.thread, latest);
assert!(inherited.thread_provider.is_none());
manager.shutdown().await;
}
+10
View File
@@ -591,6 +591,16 @@ impl Session {
.get::<codex_extension_api::SessionIsolation>()
.map(|policy| *policy)
.unwrap_or_default();
// Enforce snapshot-only instructions for both managed and inline isolated sessions.
let instructions = if isolation == codex_extension_api::SessionIsolation::Isolated {
SessionInstructions {
user: instructions.user,
thread: instructions.thread,
..Default::default()
}
} else {
instructions
};
let exec_policy = if isolation == codex_extension_api::SessionIsolation::Isolated {
let managed_policy = config
.config_layer_stack
+11 -5
View File
@@ -737,7 +737,7 @@ impl Session {
mut session_configuration: SessionConfiguration,
environment_selections: &[TurnEnvironmentSelection],
config: Arc<Config>,
instructions: SessionInstructions,
mut instructions: SessionInstructions,
installation_id: String,
auth_manager: Arc<AuthManager>,
models_manager: SharedModelsManager,
@@ -869,6 +869,16 @@ impl Session {
));
}
};
let isolation = thread_extension_init
.get::<codex_extension_api::SessionIsolation>()
.map(|policy| *policy)
.unwrap_or_default();
if !session_configuration.session_source.is_non_root_agent()
&& isolation != codex_extension_api::SessionIsolation::Isolated
{
instructions.thread_provider = agent_control
.root_thread_instructions_provider(thread_id, instructions.thread_provider);
}
// Ephemeral forks reuse cache routing, without sharing storage or lifecycle identity.
let fork_cache_key = match &initial_history {
InitialHistory::Forked(items)
@@ -949,10 +959,6 @@ impl Session {
// Publish the already resolved model before extensions make startup decisions.
// Turn construction refreshes this attachment when the selected model changes.
thread_extension_init.insert(model_info);
let isolation = thread_extension_init
.get::<codex_extension_api::SessionIsolation>()
.map(|policy| *policy)
.unwrap_or_default();
let allowed_tools = thread_extension_init
.get::<codex_extension_api::AllowedTools>()
.or_else(|| {
+15 -1
View File
@@ -1,4 +1,5 @@
mod managed;
mod shared_instructions;
use crate::CodexAppsToolsCache;
use crate::agent::LocalAgentControl;
@@ -394,6 +395,7 @@ pub(crate) struct ResumeThreadWithHistoryOptions {
/// function to require an `Arc<&Self>`.
pub(crate) struct ThreadManagerState {
threads: Arc<RwLock<HashMap<ThreadId, Arc<CodexThread>>>>,
shared_thread_instructions: shared_instructions::SharedThreadInstructionsProviders,
thread_created_tx: broadcast::Sender<ThreadId>,
thread_id_generator: ThreadIdGenerator,
auth_manager: Arc<AuthManager>,
@@ -542,6 +544,7 @@ impl ThreadManager {
Self {
state: Arc::new(ThreadManagerState {
threads: Arc::new(RwLock::new(HashMap::new())),
shared_thread_instructions: Default::default(),
thread_created_tx,
thread_id_generator: default_thread_id_generator(),
models_manager,
@@ -690,6 +693,7 @@ impl ThreadManager {
Self {
state: Arc::new(ThreadManagerState {
threads: Arc::new(RwLock::new(HashMap::new())),
shared_thread_instructions: Default::default(),
thread_created_tx,
thread_id_generator: default_thread_id_generator(),
models_manager: create_model_provider(provider, Some(auth_manager.clone()))
@@ -1566,6 +1570,15 @@ impl ThreadManager {
}
impl ThreadManagerState {
pub(crate) fn shared_thread_instructions_provider(
&self,
root_thread_id: ThreadId,
provider: Option<Arc<dyn ThreadInstructionsProvider>>,
) -> Option<Arc<dyn ThreadInstructionsProvider>> {
self.shared_thread_instructions
.for_root(root_thread_id, provider)
}
pub(crate) fn agent_graph_store(&self) -> Option<Arc<dyn AgentGraphStore>> {
self.agent_graph_store.clone()
}
@@ -1733,7 +1746,8 @@ impl ThreadManagerState {
///
/// 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.
/// subsequent context captures. Subagents inherit applied parent snapshots and
/// any thread provider that explicitly opts into sharing with descendants.
/// 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.
@@ -0,0 +1,112 @@
//! Keeps one replaceable instruction provider per root while its agent tree is alive.
use codex_extension_api::Instructions;
use codex_extension_api::LoadInstructionsFuture;
use codex_extension_api::LoadedUserInstructions;
use codex_extension_api::ThreadInstructionsProvider;
use codex_protocol::ThreadId;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::Weak;
#[derive(Default)]
pub(super) struct SharedThreadInstructionsProviders {
roots: Mutex<HashMap<ThreadId, Weak<SharedThreadInstructionsProvider>>>,
}
impl SharedThreadInstructionsProviders {
pub(super) fn for_root(
&self,
root: ThreadId,
provider: Option<Arc<dyn ThreadInstructionsProvider>>,
) -> Option<Arc<dyn ThreadInstructionsProvider>> {
let mut roots = self
.roots
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
roots.retain(|_, provider| provider.strong_count() > 0);
let existing = roots.get(&root).and_then(Weak::upgrade);
if provider
.as_ref()
.is_some_and(|provider| !provider.share_with_subagents())
{
if let Some(existing) = existing {
existing.replace(/*provider*/ None);
}
return provider;
}
match (existing, provider) {
(Some(existing), provider) => {
if provider.is_some() {
existing.replace(provider);
}
Some(existing)
}
(None, Some(provider)) => {
let shared = Arc::new(SharedThreadInstructionsProvider(Mutex::new(
ProviderState {
provider: Some(provider),
last: None,
},
)));
roots.insert(root, Arc::downgrade(&shared));
Some(shared)
}
(None, None) => None,
}
}
}
struct SharedThreadInstructionsProvider(Mutex<ProviderState>);
struct ProviderState {
provider: Option<Arc<dyn ThreadInstructionsProvider>>,
last: Option<Instructions>,
}
impl SharedThreadInstructionsProvider {
fn replace(&self, provider: Option<Arc<dyn ThreadInstructionsProvider>>) {
self.0
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.provider = provider;
}
}
impl ThreadInstructionsProvider for SharedThreadInstructionsProvider {
fn share_with_subagents(&self) -> bool {
true
}
fn load_thread_instructions(&self) -> LoadInstructionsFuture<'_> {
Box::pin(async move {
let (provider, last) = {
let state = self
.0
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
(state.provider.clone(), state.last.clone())
};
let Some(provider) = provider else {
return LoadedUserInstructions {
instructions: last,
warnings: Vec::new(),
};
};
let loaded = provider.load_thread_instructions().await;
let mut state = self
.0
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if state
.provider
.as_ref()
.is_some_and(|current| Arc::ptr_eq(current, &provider))
{
state.last = loaded.instructions.clone();
}
loaded
})
}
}
+105 -15
View File
@@ -109,9 +109,10 @@ impl UserInstructionsProvider for WarningInstructionsProvider {
}
}
struct RecordingThreadInstructionsProvider {
pub(super) struct RecordingThreadInstructionsProvider {
loaded: Mutex<LoadedUserInstructions>,
load_count: AtomicUsize,
shared: bool,
}
impl RecordingThreadInstructionsProvider {
@@ -122,21 +123,27 @@ impl RecordingThreadInstructionsProvider {
warnings: Vec::new(),
}),
load_count: AtomicUsize::new(0),
shared: false,
}
}
fn with_text(text: impl Into<String>) -> Self {
pub(super) fn with_text(text: impl Into<String>) -> Self {
Self::new(Some(Instructions {
text: text.into(),
source: None,
}))
}
fn load_count(&self) -> usize {
pub(super) fn shared(mut self) -> Self {
self.shared = true;
self
}
pub(super) fn load_count(&self) -> usize {
self.load_count.load(Ordering::SeqCst)
}
fn set_instructions(&self, instructions: Option<Instructions>) {
pub(super) fn set_instructions(&self, instructions: Option<Instructions>) {
self.loaded
.lock()
.expect("instruction snapshot lock")
@@ -152,6 +159,10 @@ impl RecordingThreadInstructionsProvider {
}
impl ThreadInstructionsProvider for RecordingThreadInstructionsProvider {
fn share_with_subagents(&self) -> bool {
self.shared
}
fn load_thread_instructions(&self) -> LoadInstructionsFuture<'_> {
self.load_count.fetch_add(1, Ordering::SeqCst);
let loaded = self
@@ -218,7 +229,7 @@ fn remove_agents_md_world_state_section(rollout_path: &Path) -> Result<()> {
Ok(())
}
fn instruction_fragments(request: &responses::ResponsesRequest) -> Vec<String> {
pub(super) fn instruction_fragments(request: &responses::ResponsesRequest) -> Vec<String> {
request
.message_input_texts("user")
.into_iter()
@@ -231,7 +242,7 @@ fn expected_instruction_fragment(cwd: &PathUri, contents: &str) -> String {
format!("# AGENTS.md instructions for {cwd}\n\n<INSTRUCTIONS>\n{contents}\n</INSTRUCTIONS>")
}
fn expected_provider_only_instruction_fragment(contents: &str) -> String {
pub(super) fn expected_provider_only_instruction_fragment(contents: &str) -> String {
format!("# AGENTS.md instructions\n\n<INSTRUCTIONS>\n{contents}\n</INSTRUCTIONS>")
}
@@ -239,7 +250,10 @@ fn assert_single_instruction_fragment(request: &responses::ResponsesRequest, exp
assert_eq!(instruction_fragments(request), vec![expected.to_string()]);
}
async fn submit_thread_turn(thread: &Arc<codex_core::CodexThread>, prompt: &str) -> Result<()> {
pub(super) async fn submit_thread_turn(
thread: &Arc<codex_core::CodexThread>,
prompt: &str,
) -> Result<()> {
thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: prompt.to_string(),
@@ -250,7 +264,7 @@ async fn submit_thread_turn(thread: &Arc<codex_core::CodexThread>, prompt: &str)
Ok(())
}
async fn persisted_resume_history(
pub(super) async fn persisted_resume_history(
thread: &Arc<codex_core::CodexThread>,
) -> Result<(ThreadId, InitialHistory)> {
thread.ensure_rollout_materialized().await;
@@ -1293,6 +1307,75 @@ async fn thread_provider_refreshes_at_the_next_step_of_an_active_turn() -> Resul
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn isolated_guardian_keeps_applied_thread_instructions() -> Result<()> {
let server = start_mock_server().await;
let test = test_codex()
.with_config(|config| {
config.permissions.approval_policy = codex_core::config::Constrained::allow_any(
codex_protocol::protocol::AskForApproval::OnRequest,
);
config.approvals_reviewer = codex_protocol::config_types::ApprovalsReviewer::AutoReview;
})
.build_with_auto_env(&server)
.await?;
let provider =
Arc::new(RecordingThreadInstructionsProvider::with_text(TASK_USER_INSTRUCTIONS).shared());
let parent = test
.thread_manager
.start_thread(StartThreadOptions {
environments: Some(test.codex.environment_selections().await),
thread_instructions_provider: Some(provider.clone()),
..StartThreadOptions::new(test.config.clone())
})
.await?;
// Publish an update after the parent captured its instructions, before it starts Guardian.
let parent_request = responses::mount_sse_once_match(
&server,
move |_: &wiremock::Request| {
provider.set_instructions(Some(Instructions {
text: UPDATED_TASK_USER_INSTRUCTIONS.to_owned(),
source: None,
}));
true
},
sse(vec![
responses::ev_exec_command_call_with_args(
"action",
&serde_json::json!({
"cmd": "echo reviewed",
"sandbox_permissions": "require_escalated",
"justification": "Check the requested action.",
}),
),
ev_completed("parent-action"),
]),
)
.await;
let remaining = responses::mount_sse_sequence(
&server,
vec![
sse(vec![
responses::ev_assistant_message("decision", r#"{"outcome":"deny"}"#),
ev_completed("guardian-review"),
]),
responses::sse_completed("parent-done"),
],
)
.await;
submit_thread_turn(&parent.thread, "Check the action before executing it.").await?;
let requests = remaining.requests();
let initial = expected_provider_only_instruction_fragment(TASK_USER_INSTRUCTIONS);
assert_single_instruction_fragment(&parent_request.single_request(), &initial);
assert_eq!(
requests[0].body_json()["client_metadata"]["x-openai-subagent"],
"guardian"
);
assert_single_instruction_fragment(&requests[0], &initial);
assert!(requests[1].body_contains_text(UPDATED_TASK_USER_INSTRUCTIONS));
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;
@@ -1406,11 +1489,15 @@ enum InstructionForkSource {
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")]
#[test_case::test_case(InstructionForkSource::LiveRollout, false; "live snapshot")]
#[test_case::test_case(InstructionForkSource::LiveRollout, true; "shared provider stays with source root")]
#[test_case::test_case(InstructionForkSource::OfflineHistory, false; "offline history provider")]
#[test_case::test_case(InstructionForkSource::OfflinePrepared, false; "offline prepared provider")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn fork_preserves_thread_instructions(source: InstructionForkSource) -> Result<()> {
async fn fork_preserves_thread_instructions(
source: InstructionForkSource,
shared: bool,
) -> Result<()> {
let server = start_mock_server().await;
let response_mock = responses::mount_sse_sequence(
&server,
@@ -1434,9 +1521,12 @@ async fn fork_preserves_thread_instructions(source: InstructionForkSource) -> Re
})
.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_provider = RecordingThreadInstructionsProvider::with_text(TASK_USER_INSTRUCTIONS);
let parent_provider = Arc::new(if shared {
parent_provider.shared()
} else {
parent_provider
});
let parent = test
.thread_manager
.start_thread(StartThreadOptions {
+3
View File
@@ -68,6 +68,9 @@ use tokio::sync::oneshot;
const ONE_PIXEL_PNG_BASE64: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==";
#[path = "scenarios_shared_instructions.rs"]
mod shared_instructions;
fn skills_extensions() -> Arc<ExtensionRegistry<Config>> {
let mut extensions = ExtensionRegistryBuilder::<Config>::new();
install(&mut extensions, |config: &Config| SkillsExtensionConfig {
@@ -0,0 +1,156 @@
//! Exercises provider sharing and model-visible updates across parent unload and restart.
use super::super::agents_md::RecordingThreadInstructionsProvider;
use super::super::agents_md::expected_provider_only_instruction_fragment;
use super::super::agents_md::instruction_fragments;
use super::super::agents_md::persisted_resume_history;
use super::super::agents_md::submit_thread_turn;
use super::*;
use codex_extension_api::Instructions;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use core_test_support::responses;
use core_test_support::responses::mount_sse_once;
use pretty_assertions::assert_eq;
const INITIAL: &str = "Ask before sending email.";
const UPDATED: &str = "Never send email.";
#[test_case::test_case(false; "snapshot_by_default")]
#[test_case::test_case(true; "shared_when_opted_in")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn running_descendants_refresh_only_shared_thread_instructions(shared: bool) -> Result<()> {
let server = start_mock_server().await;
let test = test_codex().build_with_auto_env(&server).await?;
let provider = RecordingThreadInstructionsProvider::with_text(INITIAL);
let provider = Arc::new(if shared { provider.shared() } else { provider });
let root = 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?;
let mut parent_id = root.thread_id;
let mut descendants = Vec::new();
for depth in 1..=2 {
let child = test
.thread_manager
.start_thread(StartThreadOptions {
environments: Some(vec![test.executor_environment().selection().clone()]),
session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: parent_id,
depth,
agent_path: None,
agent_nickname: None,
agent_role: None,
})),
..StartThreadOptions::new(test.config.clone())
})
.await?;
parent_id = child.thread_id;
descendants.push(child.thread);
}
let other_provider =
Arc::new(RecordingThreadInstructionsProvider::with_text("Other root rules").shared());
let other_root = test
.thread_manager
.start_thread(StartThreadOptions {
environments: Some(vec![test.executor_environment().selection().clone()]),
thread_instructions_provider: Some(other_provider),
..StartThreadOptions::new(test.config.clone())
})
.await?;
let root_response = mount_sse_once(&server, responses::sse_completed("root-step")).await;
submit_thread_turn(&root.thread, "persist the root before it unloads").await?;
root_response.single_request();
let (_, history) = persisted_resume_history(&root.thread).await?;
root.thread.shutdown_and_wait().await?;
assert!(
test.thread_manager
.remove_thread(&root.thread_id)
.await
.is_some()
);
assert!(
test.thread_manager
.get_thread(root.thread_id)
.await
.is_err()
);
drop(root);
let previous_provider = provider;
let provider = RecordingThreadInstructionsProvider::with_text(INITIAL);
let provider = Arc::new(if shared { provider.shared() } else { provider });
let _restarted_root = test
.thread_manager
.start_thread(StartThreadOptions {
initial_history: history,
environments: Some(vec![test.executor_environment().selection().clone()]),
thread_instructions_provider: Some(provider.clone()),
..StartThreadOptions::new(test.config.clone())
})
.await?;
previous_provider.set_instructions(Some(Instructions {
text: "Old provider must not be used".to_string(),
source: None,
}));
let previous_loads = previous_provider.load_count();
let replacement = format!(
"These AGENTS.md instructions replace all previously provided AGENTS.md instructions.\n\n{UPDATED}"
);
let updates = [
(Some(INITIAL), INITIAL),
(Some(UPDATED), replacement.as_str()),
(
None,
"The previously provided AGENTS.md instructions no longer apply.",
),
(None, ""),
];
let loads_before = provider.load_count();
let mut request_history = Vec::new();
// Both descendants stayed alive across the restart and must switch without another root turn.
for descendant in descendants {
let mut expected = Vec::new();
for (contents, appended_fragment) in updates {
provider.set_instructions(contents.map(|text| Instructions {
text: text.to_owned(),
source: None,
}));
let response = mount_sse_once(&server, responses::sse_completed("child-step")).await;
submit_thread_turn(&descendant, "continue with the current instructions").await?;
if expected.is_empty() || (shared && !appended_fragment.is_empty()) {
expected.push(expected_provider_only_instruction_fragment(
appended_fragment,
));
}
let request = response.single_request();
assert_eq!(instruction_fragments(&request), expected);
request_history.push(request);
}
}
assert_eq!(provider.load_count() > loads_before, shared);
assert_eq!(previous_provider.load_count(), previous_loads);
let other_response = mount_sse_once(&server, responses::sse_completed("other-root-step")).await;
submit_thread_turn(&other_root.thread, "check unrelated rules").await?;
assert_eq!(
instruction_fragments(&other_response.single_request()),
vec![expected_provider_only_instruction_fragment(
"Other root rules"
)]
);
if shared {
insta::assert_snapshot!(
"shared_instructions_update_running_descendants",
context_snapshot::format_request_history_snapshot(
"A host replaces the provider after the parent restarts, then updates and clears instructions for surviving children and grandchildren.",
&request_history,
&ContextSnapshotOptions::default(),
)
);
}
Ok(())
}
@@ -0,0 +1,126 @@
---
source: core/tests/suite/scenarios_shared_instructions.rs
expression: "context_snapshot::format_request_history_snapshot(\"A host replaces the provider after the parent restarts, then updates and clears instructions for surviving children and grandchildren.\",\n&request_history, &ContextSnapshotOptions::default(),)"
---
Scenario: A host replaces the provider after the parent restarts, then updates and clears instructions for surviving children and grandchildren.
## Window 1
-- request 1 (turn) --
00:message/developer:
<permissions instructions>
Filesystem sandboxing defines which files can be read or written. `sandbox_mode` is `read-only`: The san... permits reading files. Network access is restricted. [hash=be7da4425f89053c]
# Escalation Requests
Commands are run outside the sandbox if they are approved by the user, or match an existing rule that al...hell control operators, including but not limited to: [hash=d24ebf5fa4e4b9bb]
- Pipes: |
- Logical operators: &&, ||
<OMITTED 44 LINES; ~790 TOKENS; hash=3a8900343e968d61>
NEVER provide a prefix_rule if your command uses a heredoc or herestring.
### Examples
Good examples of prefixes:
- ["npm", "run", "dev"]
- ["gh", "pr", "check"]
- ["cargo", "test"]
</permissions instructions>
01:message/user[2]:
[01] # AGENTS.md instructions
<INSTRUCTIONS>
Ask before sending email.
</INSTRUCTIONS>
[02] <environment_context>
<cwd><CWD></cwd>
<shell><HOST_SHELL></shell>
<current_date><CURRENT_DATE></current_date>
<timezone><HOST_TIMEZONE></timezone>
<filesystem><workspace_roots><root><CWD></root></workspace_roots><permission_profile type="managed"><f...ntry></file_system></permission_profile></filesystem> [hash=e8c1de60307d0a3d]
<subagents>
- <UUID 1>
</subagents>
</environment_context>
02:message/user:
continue with the current instructions
-- request 2 (turn) --
03:message/user:
# AGENTS.md instructions
<INSTRUCTIONS>
These AGENTS.md instructions replace all previously provided AGENTS.md instructions.
Never send email.
</INSTRUCTIONS>
04:message/user:
continue with the current instructions
-- request 3 (turn) --
05:message/user:
# AGENTS.md instructions
<INSTRUCTIONS>
The previously provided AGENTS.md instructions no longer apply.
</INSTRUCTIONS>
06:message/user:
continue with the current instructions
-- request 4 (turn) --
07:message/user:
continue with the current instructions
## Window 2 (after request 4: input diverged at item 01, settings changed (prompt_cache_key))
-- request 5 (turn) --
00:message/developer:
<permissions instructions>
Filesystem sandboxing defines which files can be read or written. `sandbox_mode` is `read-only`: The san... permits reading files. Network access is restricted. [hash=be7da4425f89053c]
# Escalation Requests
Commands are run outside the sandbox if they are approved by the user, or match an existing rule that al...hell control operators, including but not limited to: [hash=d24ebf5fa4e4b9bb]
- Pipes: |
- Logical operators: &&, ||
<OMITTED 44 LINES; ~790 TOKENS; hash=3a8900343e968d61>
NEVER provide a prefix_rule if your command uses a heredoc or herestring.
### Examples
Good examples of prefixes:
- ["npm", "run", "dev"]
- ["gh", "pr", "check"]
- ["cargo", "test"]
</permissions instructions>
01:message/user[2]:
[01] # AGENTS.md instructions
<INSTRUCTIONS>
Ask before sending email.
</INSTRUCTIONS>
[02] <environment_context>
<cwd><CWD></cwd>
<shell><HOST_SHELL></shell>
<current_date><CURRENT_DATE></current_date>
<timezone><HOST_TIMEZONE></timezone>
<filesystem><workspace_roots><root><CWD></root></workspace_roots><permission_profile type="managed"><f...ntry></file_system></permission_profile></filesystem> [hash=e8c1de60307d0a3d]
</environment_context>
02:message/user:
continue with the current instructions
-- request 6 (turn) --
03:message/user:
# AGENTS.md instructions
<INSTRUCTIONS>
These AGENTS.md instructions replace all previously provided AGENTS.md instructions.
Never send email.
</INSTRUCTIONS>
04:message/user:
continue with the current instructions
-- request 7 (turn) --
05:message/user:
# AGENTS.md instructions
<INSTRUCTIONS>
The previously provided AGENTS.md instructions no longer apply.
</INSTRUCTIONS>
06:message/user:
continue with the current instructions
-- request 8 (turn) --
07:message/user:
continue with the current instructions
@@ -44,6 +44,18 @@ pub type LoadInstructionsFuture<'a> =
pub trait ThreadInstructionsProvider: Send + Sync {
/// Loads the current snapshot for the provider's root thread.
fn load_thread_instructions(&self) -> LoadInstructionsFuture<'_>;
/// Whether descendants should retain this provider instead of only its applied text.
/// Defaults to snapshot inheritance. Opt in only when the same instruction scope and
/// credentials apply to descendants. Descendants share the installed provider; a root
/// resume can replace it for surviving descendants. Providers must coalesce/cache
/// concurrent reads and retain a usable snapshot during outages.
/// If a resumed root supplies a non-sharing provider, existing descendants keep their
/// last shared instructions without receiving any new private updates.
/// Updates are loaded at each descendant's next model-request boundary, not mid-request.
fn share_with_subagents(&self) -> bool {
false
}
}
/// Loads global user instructions shared across root threads.