github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 00:18:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add configured thread environment inheritance for subagents (#55375)

Browse Source 
## Why

Temporary environment attachments for a parent turn should not become a
subagent's persistent environments. Children also need to retain their
configured executor bindings when their runtime is evicted and reloaded.

## What changed

- Add the disabled-by-default `subagent_thread_environments` feature flag.
  When enabled, both subagent implementations inherit configured thread
  environments captured with the invoking step, excluding turn-only attachments.
- Track configured bindings separately from active turn attachments, preserving
  pending configuration ownership and sharing completed executor startup.
- Retain configured selections and bindings across idle eviction. Reload children
  using retained bindings when the parent changes workspace, while continuing to
  validate current parent roots and permissions for matching workspaces.

## Testing

Extend regression coverage for configured, empty, forked, pending, and legacy
inheritance; temporary attachment removal across eviction; and reload behavior
with changed workspaces, deferred startup, and restricted permissions.

GitOrigin-RevId: 15692ecd99deb38002014b44fd859d793bd54373
This commit is contained in:
jif
2026-10-19 20:30:44 +00:00
committed by copyberry
parent 0d337e5fe7
commit e389b2444c
23 changed files with 667 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/config.schema.json
+6
View File
@@ -1222,6 +1222,9 @@
"subagent_default_context_limits": {
"type": "boolean"
},
"subagent_thread_environments": {
"type": "boolean"
},
"system_proxy_fallback": {
"type": "boolean"
},
@@ -7631,6 +7634,9 @@
"subagent_default_context_limits": {
"type": "boolean"
},
"subagent_thread_environments": {
"type": "boolean"
},
"system_proxy_fallback": {
"type": "boolean"
},
codex-rs/core/src/agent/control.rs
+1 -8
View File
@@ -651,14 +651,7 @@ impl LocalAgentControl {
};
let parent_thread = state.get_thread(*parent_thread_id).await.ok()?;
Some(
parent_thread
.session
.services
.turn_environments
.snapshot()
.await,
)
Some(parent_thread.session.inherited_environment_snapshot().await)
}
async fn inherited_exec_policy_for_source(
codex-rs/core/src/agent/control/residency.rs
+11 -1
View File
@@ -4,6 +4,7 @@ use super::LocalAgentControl;
use super::LocalAgentRuntime;
use super::runtime::AgentTreeMembership;
use crate::agent::AgentStatus;
use crate::agent::registry::EvictedEnvironments;
use crate::codex_thread::CodexThread;
use crate::config::Config;
use crate::thread_manager::ThreadManagerState;
@@ -76,7 +77,16 @@ impl ThreadManagerState {
.collect();
let runtime = &thread.session.services.local_agent_runtime;
let _ = runtime.mailboxes.enqueue(thread_id, /*id*/ None, mail);
let environments = thread.environment_selections().await;
// Preserve configured bindings across eviction without retaining turn-only attachments.
// Raw bindings keep the original pending owner configuration alive.
let environments = EvictedEnvironments {
selections: thread.environment_selections().await,
bindings: thread
.session
.services
.turn_environments
.configured_snapshot_now(),
};
let mut threads = manager.threads.write().await;
if threads
.get(&thread_id)
codex-rs/core/src/agent/control/spawn.rs
+39 -14
View File
@@ -362,7 +362,11 @@ impl LocalAgentControl {
{
return Err(CodexErr::ThreadNotFound(thread_id));
}
let mut environment_selections = self.runtime.registry.evicted_environments(thread_id);
let evicted_environments = self.runtime.registry.evicted_environments(thread_id);
let mut environment_selections = evicted_environments
.as_ref()
.map(|environments| environments.selections.clone());
let evicted_environments = evicted_environments.map(|environments| environments.bindings);
let stored_thread = state
.read_stored_thread(ReadThreadParams {
@@ -418,7 +422,8 @@ impl LocalAgentControl {
"cannot resume multi-agent v2 child {thread_id} with the current parent settings"
))
})?;
Some((parent, turn.initial_environments.clone()))
let environments = parent.session.inherited_environment_snapshot().await;
Some((parent, environments))
} else {
None
};
@@ -490,6 +495,11 @@ impl LocalAgentControl {
)));
}
if let Some(selections) = environment_selections.as_mut() {
let parent_selections = parent_environments.all_selections();
let ready_parent = parent_environments.refresh_readiness();
let ready_evicted = evicted_environments
.as_ref()
.map(TurnEnvironmentSnapshot::refresh_readiness);
for selection in selections {
let environment_id = &selection.environment_id;
let invalid_environment = |reason: &str| {
@@ -497,17 +507,22 @@ impl LocalAgentControl {
"cannot resume multi-agent v2 child {thread_id}: cached environment {environment_id} {reason}"
))
};
// Matching the attachment also keeps startup on the captured owner executor.
let owner_environment = parent_environments
// A changed parent selection does not revoke the child's retained binding.
// If the parent still configures this workspace, keep enforcing its current
// roots and permissions rather than restoring superseded authority.
let owner_environments = if parent_selections.iter().any(|parent_selection| {
parent_selection.environment_id == selection.environment_id
&& parent_selection.cwd == selection.cwd
}) {
&ready_parent
} else {
ready_evicted.as_ref().unwrap_or(&ready_parent)
};
let owner_environment = owner_environments
.turn_environments()
.find(|environment| {
let parent_selection = &environment.selection;
parent_selection.environment_id == selection.environment_id
&& parent_selection.cwd == selection.cwd
&& parent_selection.workspace_roots == selection.workspace_roots
})
.find(|environment| environment.selection.has_same_workspace(selection))
.ok_or_else(|| {
invalid_environment("no longer matches a ready parent environment")
invalid_environment("has no matching ready environment binding")
})?;
let owner_config = owner_environment.config();
let child_config = match &selection.config {
@@ -569,14 +584,24 @@ impl LocalAgentControl {
}
}
(
Some(parent_environments.clone()),
Some(
evicted_environments
.as_ref()
.unwrap_or(parent_environments)
.clone(),
),
Some(Arc::clone(&parent.session.services.exec_policy)),
Some(parent.client_mcp_extensions()),
)
} else {
(
self.inherited_environments_for_source(&state, Some(&session_source))
.await,
match &evicted_environments {
Some(environments) => Some(environments.clone()),
None => {
self.inherited_environments_for_source(&state, Some(&session_source))
.await
}
},
self.inherited_exec_policy_for_source(&state, Some(&session_source), &config)
.await,
None,
codex-rs/core/src/agent/control_tests.rs
+6 -3
View File
@@ -1175,9 +1175,12 @@ async fn cancelled_v2_send_finishes_reload_registration() {
.unwrap(),
child_id,
);
assert_eq!(
control.runtime.registry.evicted_environments(child_id),
None
assert!(
control
.runtime
.registry
.evicted_environments(child_id)
.is_none()
);
// Evict and reload again with the single child slot. This also catches a leaked
// pin or pending residency reservation left behind by the cancelled sender.
codex-rs/core/src/agent/registry.rs
+11 -6
View File
@@ -1,4 +1,5 @@
use crate::agent::types::AgentMetadata;
use crate::environment_selection::TurnEnvironmentSnapshot;
use codex_protocol::AgentPath;
use codex_protocol::ThreadId;
use codex_protocol::error::AgentErrorContext;
@@ -37,9 +38,16 @@ struct ActiveAgents {
nickname_reset_count: usize,
}
/// Thread defaults and executor bindings retained while a child's runtime is unloaded.
#[derive(Clone)]
pub(crate) struct EvictedEnvironments {
pub(crate) selections: Vec<TurnEnvironmentSelection>,
pub(crate) bindings: TurnEnvironmentSnapshot,
}
struct RegisteredAgent {
path: String,
evicted_environments: Option<Vec<TurnEnvironmentSelection>>,
evicted_environments: Option<EvictedEnvironments>,
}
impl RegisteredAgent {
@@ -174,7 +182,7 @@ impl AgentRegistry {
pub(crate) fn save_evicted_environments(
&self,
thread_id: ThreadId,
environments: Vec<TurnEnvironmentSelection>,
environments: EvictedEnvironments,
) {
let mut active_agents = self
.active_agents
@@ -185,10 +193,7 @@ impl AgentRegistry {
}
}
pub(crate) fn evicted_environments(
&self,
thread_id: ThreadId,
) -> Option<Vec<TurnEnvironmentSelection>> {
pub(crate) fn evicted_environments(&self, thread_id: ThreadId) -> Option<EvictedEnvironments> {
let active_agents = self
.active_agents
.lock()
codex-rs/core/src/agent/types.rs
+1 -1
View File
@@ -32,7 +32,7 @@ pub struct SpawnAgentOptions {
/// Attribute delegated usage to the turn that initiated it.
pub turn_trigger: Option<String>,
pub root_turn_id: Option<String>,
/// Exact environment bindings from the step that requested the child.
/// Inheritable bindings captured by the step that requested the child.
pub environments: Option<TurnEnvironmentSnapshot>,
/// Plugin choices from the turn requesting the child. Omission inherits the parent's
/// current choices; an explicit list, including an empty one, overrides them and history.
codex-rs/core/src/environment_selection.rs
+182 -54
View File
@@ -165,6 +165,22 @@ type TurnEnvironmentResolution = Shared<BoxFuture<'static, TurnEnvironmentResult
type PendingConfigurationResult = Result<EnvironmentConfig, String>;
pub(crate) type PendingConfiguration = Shared<BoxFuture<'static, PendingConfigurationResult>>;
fn pending_configuration(
completion: &watch::Sender<Option<PendingConfigurationResult>>,
) -> PendingConfiguration {
let mut receiver = completion.subscribe();
async move {
receiver
.wait_for(Option::is_some)
.await
.ok()
.and_then(|result| result.clone())
.unwrap_or_else(|| Err("environment configuration was canceled".to_string()))
}
.boxed()
.shared()
}
// Shared startup result used to build each turn's environment with its own config
// without restarting the connection or shell resolution.
#[derive(Clone)]
@@ -191,7 +207,7 @@ struct SelectedTurnEnvironment {
resolution: TurnEnvironmentResolution,
// Executor retries keep this session's wait for its first accepted configuration.
pending_completion: Option<watch::Sender<Option<PendingConfigurationResult>>>,
// Descendants can await the original attachment's first accepted config or failure.
// Inherited owner only; otherwise snapshots expose this attachment's own completion.
owner_config_result: Option<PendingConfiguration>,
}
@@ -223,6 +239,22 @@ fn resolve_selection_config(
}
impl SelectedTurnEnvironment {
// Pending configuration belongs to each attachment set; completed startup can be shared.
fn for_turn(&self, local_shell: &Shell, shell_snapshot: &ShellSnapshot) -> Self {
if matches!(self.selection.config, EnvironmentConfigState::Pending) {
return ThreadEnvironments::start_environment(
self.selection.clone(),
self.config_origin,
Arc::clone(&self.environment),
self.owner_config_result.clone(),
/*pending_completion*/ None,
local_shell,
shell_snapshot,
);
}
self.clone()
}
fn refresh_thread_config(&mut self, defaults: &ThreadEnvironmentDefaults) {
if self.config_origin == EnvironmentConfigOrigin::Thread {
self.selection.config =
@@ -266,6 +298,8 @@ pub(crate) struct ThreadEnvironments {
shell_snapshot: ShellSnapshot,
non_blocking_snapshots: bool,
state: Mutex<ThreadEnvironmentsState>,
// Retain configured bindings independently of active-turn membership and cancellation.
configured: Mutex<ThreadEnvironmentsState>,
connection_event_tx: OnceLock<Sender<Event>>,
}
@@ -283,10 +317,19 @@ impl ThreadEnvironments {
current: TurnEnvironmentSnapshot,
non_blocking_snapshots: bool,
) -> Self {
let environments = current
let environments: Vec<_> = current
.environments
.into_iter()
.filter_map(|inherited| {
// Reuse completed startup without resolving a Pending owner's configuration.
let inherited = if let TurnEnvironmentState::Starting(starting) = &inherited
&& let EnvironmentConfigState::Ready(_) = &starting.selection.config
&& let Some(Ok(resolved)) = starting.resolution.clone().now_or_never()
{
TurnEnvironmentState::from_resolution(starting.clone(), Some(Ok(resolved)))
} else {
inherited
};
let environment = match inherited {
TurnEnvironmentState::Ready(environment) => environment,
TurnEnvironmentState::Starting(inherited) => {
@@ -349,12 +392,20 @@ impl ThreadEnvironments {
Some(inherited_environment)
})
.collect();
let active = environments
.iter()
.map(|environment| environment.for_turn(&local_shell, &shell_snapshot))
.collect();
Self {
environment_manager,
local_shell,
shell_snapshot,
non_blocking_snapshots,
state: Mutex::new(ThreadEnvironmentsState {
thread_defaults: thread_defaults.clone(),
environments: active,
}),
configured: Mutex::new(ThreadEnvironmentsState {
thread_defaults,
environments,
}),
@@ -398,7 +449,40 @@ impl ThreadEnvironments {
/// Updates the selected list before waking work that was waiting on the previous selection.
pub(crate) fn update_selections(&self, environments: &[TurnEnvironmentSelection]) {
let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
let bindings = self
.configured
.lock()
.unwrap_or_else(PoisonError::into_inner)
.environments
.clone();
self.update_selections_with_bindings(
&self.state,
environments,
&bindings,
self.connection_event_tx.get(),
);
// Retry configured bindings with their own settings, keeping turn-only changes temporary.
self.update_configured_selections(&self.configured_snapshot_now().all_selections());
}
/// Retains configured bindings without changing the active turn's attachments.
pub(crate) fn update_configured_selections(&self, environments: &[TurnEnvironmentSelection]) {
self.update_selections_with_bindings(
&self.configured,
environments,
&[],
/*tx_event*/ None,
);
}
fn update_selections_with_bindings(
&self,
state: &Mutex<ThreadEnvironmentsState>,
environments: &[TurnEnvironmentSelection],
bindings: &[SelectedTurnEnvironment],
tx_event: Option<&Sender<Event>>,
) {
let mut state = state.lock().unwrap_or_else(PoisonError::into_inner);
let ThreadEnvironmentsState {
thread_defaults,
environments: current,
@@ -406,18 +490,52 @@ impl ThreadEnvironments {
let mut seen_environment_ids = HashSet::with_capacity(environments.len());
let mut next = Vec::with_capacity(environments.len());
let mut pending_completions = Vec::new();
// Connection state belongs to the environment instance, not its cwd or roots.
let connection_events_task = |environment_id: &str, environment: &Arc<Environment>| {
current
.iter()
.find(|previous| {
previous.selection.environment_id == environment_id
&& Arc::ptr_eq(&previous.environment, environment)
})
.and_then(|previous| previous.connection_events_task.clone())
.or_else(|| {
Self::spawn_connection_event_listener(
environment.as_ref(),
environment_id.to_owned(),
tx_event?.clone(),
)
})
};
for selected_environment in environments {
if !seen_environment_ids.insert(selected_environment.environment_id.as_str()) {
continue;
}
let previous = current.iter().find(|environment| {
let previous = &environment.selection;
previous.has_same_workspace(selected_environment)
let current_environment = current.iter().find(|environment| {
environment
.selection
.has_same_workspace(selected_environment)
});
let previous = current_environment.or_else(|| {
bindings.iter().find(|environment| {
environment
.selection
.has_same_workspace(selected_environment)
})
});
let (selected_environment, config_origin) =
resolve_selection_config(selected_environment.clone(), thread_defaults);
if let Some(environment) = previous {
let mut environment = environment.clone();
let failed =
matches!(
environment.selection.config,
EnvironmentConfigState::Failed(_)
) || matches!(environment.resolution.clone().now_or_never(), Some(Err(_)));
let mut environment = if current_environment.is_some() {
environment.clone()
} else {
environment.for_turn(&self.local_shell, &self.shell_snapshot)
};
if let Some(completion) = environment.pending_completion.take() {
match &selected_environment.config {
EnvironmentConfigState::Ready(config) => {
@@ -431,11 +549,6 @@ impl ThreadEnvironments {
}
}
}
let failed =
matches!(
environment.selection.config,
EnvironmentConfigState::Failed(_)
) || matches!(environment.resolution.clone().now_or_never(), Some(Err(_)));
let restarting_as_pending =
matches!(selected_environment.config, EnvironmentConfigState::Pending)
&& !matches!(
@@ -459,6 +572,12 @@ impl ThreadEnvironments {
{
self.restart_shell_snapshot(&mut environment, resolved);
}
if environment.connection_events_task.is_none() {
environment.connection_events_task = connection_events_task(
&environment.selection.environment_id,
&environment.environment,
);
}
next.push(environment);
continue;
}
@@ -469,23 +588,7 @@ impl ThreadEnvironments {
tracing::warn!("skipping unknown turn environment `{environment_id}`");
continue;
};
// Connection state belongs to the environment instance, not its cwd or roots.
let connection_events_task = current
.iter()
.find(|previous| {
previous.selection.environment_id.as_str() == environment_id.as_str()
&& Arc::ptr_eq(&previous.environment, &environment)
})
.and_then(|previous| previous.connection_events_task.clone())
.or_else(|| {
self.connection_event_tx.get().and_then(|tx_event| {
Self::spawn_connection_event_listener(
environment.as_ref(),
environment_id.clone(),
tx_event.clone(),
)
})
});
let connection_events_task = connection_events_task(environment_id, &environment);
let pending = previous.filter(|previous| {
matches!(selected_environment.config, EnvironmentConfigState::Pending)
&& previous.pending_completion.is_some()
@@ -495,7 +598,7 @@ impl ThreadEnvironments {
config_origin,
environment,
pending.and_then(|previous| previous.owner_config_result.clone()),
pending.and_then(|previous| previous.pending_completion.clone()),
current_environment.and_then(|previous| previous.pending_completion.clone()),
&self.local_shell,
&self.shell_snapshot,
);
@@ -518,7 +621,13 @@ impl ThreadEnvironments {
// Publish after installing the new list, including pending bindings replaced by retries.
for (completion, result) in pending_completions {
let _ = completion.send_replace(Some(result));
completion.send_if_modified(|current| {
if current.is_some() {
return false;
}
*current = Some(result);
true
});
}
for task in removed_connection_tasks {
@@ -541,25 +650,12 @@ impl ThreadEnvironments {
if matches!(selection.config, EnvironmentConfigState::Pending) {
let sender =
pending_completion.unwrap_or_else(|| watch::channel(/*init*/ None).0);
let mut receiver = sender.subscribe();
let result = async move {
receiver
.wait_for(Option::is_some)
.await
.ok()
.and_then(|result| result.clone())
.unwrap_or_else(
|| Err("environment configuration was canceled".to_string()),
)
}
.boxed()
.shared();
let result = pending_configuration(&sender);
(Some(sender), Some(result))
} else {
(None, None)
};
// The resolver waits for this Session; descendants can follow the original owner.
let owner_config_result = owner_config_result.or_else(|| configuration_ready.clone());
let (task, resolution) = Self::resolve_environment(
selection.clone(),
Arc::clone(&environment),
@@ -627,7 +723,19 @@ impl ThreadEnvironments {
/// Changes the defaults inherited by all selected and later-added environments. A step that
/// already captured its environments keeps the previous values.
pub(crate) fn set_active_thread_defaults(&self, defaults: EnvironmentConfig) {
let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
self.set_thread_defaults(&self.state, defaults);
}
pub(crate) fn set_configured_thread_defaults(&self, defaults: EnvironmentConfig) {
self.set_thread_defaults(&self.configured, defaults);
}
fn set_thread_defaults(
&self,
state: &Mutex<ThreadEnvironmentsState>,
defaults: EnvironmentConfig,
) {
let mut state = state.lock().unwrap_or_else(PoisonError::into_inner);
let ThreadEnvironmentsState {
thread_defaults,
environments,
@@ -658,11 +766,19 @@ impl ThreadEnvironments {
pub(crate) fn set_snapshot_credential_broker(&self, broker: SnapshotCredentialBrokerState) {
let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
let mut configured = self
.configured
.lock()
.unwrap_or_else(PoisonError::into_inner);
if !self.shell_snapshot.set_credential_broker(broker) {
return;
}
for selected in &mut state.environments {
for selected in state
.environments
.iter_mut()
.chain(&mut configured.environments)
{
if !selected.environment.is_remote()
&& let Some(Ok(resolved)) = selected.resolution.clone().now_or_never()
{
@@ -909,23 +1025,35 @@ impl ThreadEnvironments {
/// Copies the current selection without waiting for any executor to connect.
pub(crate) fn snapshot_now(&self) -> TurnEnvironmentSnapshot {
Self::snapshot_state(&self.state).refresh_readiness()
}
/// Captures configured bindings without resolving readiness or losing config ownership.
/// Inheritance and eviction retain these bindings; execution and validation resolve a view.
pub(crate) fn configured_snapshot_now(&self) -> TurnEnvironmentSnapshot {
Self::snapshot_state(&self.configured)
}
fn snapshot_state(state: &Mutex<ThreadEnvironmentsState>) -> TurnEnvironmentSnapshot {
// Keep the startup results, but not the senders that would keep removed Pending work alive.
let environments = self
.state
let environments = state
.lock()
.unwrap_or_else(PoisonError::into_inner)
.environments
.iter()
.map(|environment| {
let starting = StartingTurnEnvironment {
TurnEnvironmentState::Starting(StartingTurnEnvironment {
selection: environment.selection.clone(),
config_origin: environment.config_origin,
resolution: environment.resolution.clone(),
environment: Arc::clone(&environment.environment),
owner_config_result: environment.owner_config_result.clone(),
};
let resolved = starting.resolution.clone().now_or_never();
TurnEnvironmentState::from_resolution(starting, resolved)
owner_config_result: environment.owner_config_result.clone().or_else(|| {
environment
.pending_completion
.as_ref()
.map(pending_configuration)
}),
})
})
.collect();
TurnEnvironmentSnapshot { environments }
codex-rs/core/src/environment_selection_snapshot_tests.rs
+9 -4
View File
@@ -126,8 +126,10 @@ async fn capability_roots_follow_attachments_through_configuration_changes() {
));
}
#[test_case::test_case(false; "standalone startup")]
#[test_case::test_case(true; "configured startup")]
#[tokio::test]
async fn unpolled_snapshot_does_not_delay_canceling_a_removed_environment() {
async fn unpolled_snapshot_does_not_delay_canceling_a_removed_environment(shared_startup: bool) {
let cwd = PathUri::from_abs_path(&AbsolutePathBuf::current_dir().expect("cwd"));
let selection = TurnEnvironmentSelection {
selected_capability_roots: Default::default(),
@@ -144,6 +146,9 @@ async fn unpolled_snapshot_does_not_delay_canceling_a_removed_environment() {
TurnEnvironmentSnapshot::default(),
/*non_blocking_snapshots*/ false,
);
if shared_startup {
environments.update_configured_selections(std::slice::from_ref(&selection));
}
environments.update_selections(&[selection]);
let starting = environments
.snapshot()
@@ -208,7 +213,7 @@ async fn updating_another_environment_retries_the_executor_without_canceling_pen
(
failing.resolution.clone(),
failing.pending_completion.as_ref().unwrap().subscribe(),
failing.owner_config_result.clone().unwrap(),
pending_configuration(failing.pending_completion.as_ref().unwrap()),
)
};
@@ -218,9 +223,9 @@ async fn updating_another_environment_retries_the_executor_without_canceling_pen
local.config = EnvironmentConfigState::Ready(tests::test_environment_config());
environments.update_selections(&[local.clone(), pending.clone()]);
let (second_attempt, _, second_owner) = current();
let (second_attempt, second_config, _) = current();
assert!(!first_attempt.ptr_eq(&second_attempt));
assert!(original_owner.ptr_eq(&second_owner));
assert!(original_config.same_channel(&second_config));
assert!(second_attempt.await.is_err());
assert!(original_config.borrow().is_none());
codex-rs/core/src/session/environment.rs
+6
View File
@@ -23,6 +23,7 @@ use crate::session::session::SessionConfiguration;
use crate::session::session::SessionSettingsUpdate;
/// Defaults for environments that inherit their configuration from the running turn.
#[derive(Clone)]
pub(crate) struct ThreadEnvironmentDefaults {
pub(crate) common: EnvironmentConfig,
// Chosen once when the session starts; it does not apply to remote environments.
@@ -349,6 +350,11 @@ impl Session {
}
}
if update_future {
self.services
.turn_environments
.update_configured_selections(&future);
}
state.session_configuration.environments = future;
if update_current {
// Invalidate MCP before installed configuration can wake a waiting turn.
codex-rs/core/src/session/mod.rs
+33 -2
View File
@@ -2040,6 +2040,14 @@ impl Session {
// Save new environment defaults for future turns. The running turn keeps its own.
windows_sandbox_changed =
state.session_configuration.windows_sandbox_level != updated.windows_sandbox_level;
self.services
.turn_environments
.set_configured_thread_defaults(updated.inferred_environment_config());
if updated.environments != state.session_configuration.environments {
self.services
.turn_environments
.update_configured_selections(&updated.environments);
}
state.session_configuration = updated;
if root_service_tier_changed {
self.services.agent_control.propagate_config_update(
@@ -2113,6 +2121,18 @@ impl Session {
.thread_config_snapshot(state.session_configuration.environments.clone())
}
/// Captures configured child bindings, or the active turn's bindings for legacy inheritance.
pub(crate) async fn inherited_environment_snapshot(&self) -> TurnEnvironmentSnapshot {
let snapshot = {
let _state = self.state.lock().await;
if self.features.enabled(Feature::SubagentThreadEnvironments) {
return self.services.turn_environments.configured_snapshot_now();
}
self.services.turn_environments.snapshot()
};
snapshot.await
}
pub(crate) async fn configured_environment_selections(&self) -> Vec<TurnEnvironmentSelection> {
let state = self.state.lock().await;
state.session_configuration.environments.clone()
@@ -3865,6 +3885,10 @@ impl Session {
.await
}
#[expect(
clippy::await_holding_invalid_type,
reason = "step settings and both environment snapshots must be captured under active_turn then state"
)]
#[tracing::instrument(name = "step_context.capture", level = "info", skip_all)]
async fn capture_step_context_inner(
self: &Arc<Self>,
@@ -3874,16 +3898,21 @@ impl Session {
required_plugins: &HashSet<String>,
) -> CodexResult<Arc<StepContext>> {
// Read the step's model and record its environments together so an update cannot split them.
// Wait for executor startup below, after releasing the lock.
let (mut settings, environments) = {
// Match apply_turn_settings and activate_turn_environments: active_turn before state.
// Wait for executor startup below, after releasing both locks.
let (mut settings, environments, inherited_environments) = {
let _active = self
.active_turn
.lock()
.or_cancel(cancellation_token)
.await?;
let _state = self.state.lock().or_cancel(cancellation_token).await?;
(
turn_context.next_step_settings.load_full(),
self.services.turn_environments.snapshot(),
self.features
.enabled(Feature::SubagentThreadEnvironments)
.then(|| self.services.turn_environments.configured_snapshot_now()),
)
};
if matches!(
@@ -3911,6 +3940,7 @@ impl Session {
);
let session_telemetry = settings.telemetry(&turn_context.session_telemetry);
let environments = environments.or_cancel(cancellation_token).await?;
let inherited_environments = inherited_environments.unwrap_or_else(|| environments.clone());
// Keep both preparation futures off caller stacks while they are live together.
let load_agents_md = Box::pin(async {
let (loaded_agents_md, warnings) = self
@@ -4047,6 +4077,7 @@ impl Session {
turn: turn_context,
extension_data,
environments,
inherited_environments,
executor_capability_discovery,
mcp,
tool_router,
codex-rs/core/src/session/session.rs
+1
View File
@@ -1483,6 +1483,7 @@ impl Session {
inherited_environments.clone(),
config.features.enabled(Feature::DeferredExecutor),
));
turn_environments.update_configured_selections(environment_selections);
turn_environments.update_selections(environment_selections);
session_configuration.environments = turn_environments.selections();
let resolved_environments = turn_environments.snapshot().await;
codex-rs/core/src/session/step_activation_tests.rs
+15 -1
View File
@@ -189,14 +189,27 @@ async fn next_step_waits_for_environment_publication_before_capturing() {
.turn_environments
.update_selections(&selections);
}
{
// Thread attachments publish under state independently of active-turn settings.
// A capture waiting for that publication must not retain the previous attachments.
let mut state = session.state.lock().await;
let mut context = std::task::Context::from_waker(futures::task::noop_waker_ref());
assert!(std::future::Future::poll(capture.as_mut(), &mut context).is_pending());
session
.services
.turn_environments
.update_configured_selections(&selections);
state.session_configuration.environments = selections.clone();
}
let step = capture.await.expect("capture after publication");
assert_eq!(
(
step.settings.model_info.slug.as_str(),
step.environments.primary().map(TurnEnvironment::cwd),
step.inherited_environments.all_selections(),
),
("next-step-model", Some(&next_cwd)),
("next-step-model", Some(&next_cwd), selections),
);
session.abort_all_tasks(TurnAbortReason::Interrupted).await;
}
@@ -409,6 +422,7 @@ async fn activation_fixture(models: Vec<ModelInfo>) -> ActivationFixture {
let (models, lookup) = GatedModelsManager::new(models);
mutable.services.models_manager = models;
for feature in [
Feature::SubagentThreadEnvironments,
Feature::StepModelSwitching,
Feature::FastMode,
Feature::TokenBudget,
codex-rs/core/src/session/step_context.rs
+2
View File
@@ -33,6 +33,8 @@ pub(crate) struct StepContext {
/// Telemetry context tagged with this sampling request's model.
pub(crate) session_telemetry: SessionTelemetry,
pub(crate) environments: TurnEnvironmentSnapshot,
/// Child bindings captured with this step; configured-only when opted in.
pub(crate) inherited_environments: TurnEnvironmentSnapshot,
/// Legacy discovery retained for executor hook materialization in this exact step.
pub(crate) executor_capability_discovery: Option<Arc<ExecutorCapabilityDiscoverySnapshot>>,
/// Keeps the extension inputs used to build this step's tools for its World State as well.
codex-rs/core/src/session/tests.rs
+1
View File
@@ -289,6 +289,7 @@ impl StepContext {
mode_instructions: None,
},
turn: Arc::clone(&turn),
inherited_environments: environments.clone(),
environments,
executor_capability_discovery: None,
extension_data: codex_extension_api::ExtensionData::new(turn.sub_id.clone()),
codex-rs/core/src/thread_spawn.rs
+1 -1
View File
@@ -152,7 +152,7 @@ pub(crate) async fn spawn(
parent_turn_id: Some(turn.sub_id.clone()),
root_turn_id: turn.turn_metadata_state.root_turn_id(),
turn_trigger: turn.turn_metadata_state.current_turn_trigger(),
environments: Some(step_context.environments.clone()),
environments: Some(step_context.inherited_environments.clone()),
disabled_plugin_ids: Some(turn.disabled_plugin_ids.clone()),
multi_agent_v2_usage_hints,
cyber_access_program: turn.cyber_access_program,
codex-rs/core/src/tools/handlers/multi_agents/spawn.rs
+1 -1
View File
@@ -131,7 +131,7 @@ async fn handle_spawn_agent(
parent_turn_id: Some(turn.sub_id.clone()),
root_turn_id: turn.turn_metadata_state.root_turn_id(),
turn_trigger: turn.turn_metadata_state.current_turn_trigger(),
environments: Some(step_context.environments.clone()),
environments: Some(step_context.inherited_environments.clone()),
disabled_plugin_ids: Some(turn.disabled_plugin_ids.clone()),
multi_agent_v2_usage_hints: None,
cyber_access_program: turn.cyber_access_program,
codex-rs/core/tests/suite/agent_execution.rs
+145 -16
View File
@@ -29,8 +29,10 @@ use core_test_support::wait_for_event;
use futures::FutureExt;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::sync::Arc;
use std::time::Duration;
use test_case::test_case;
use tokio::net::TcpListener;
const FIRST_PROMPT: &str = "spawn the first worker";
const FIRST_TASK: &str = "first worker task";
@@ -322,15 +324,23 @@ async fn child_turn_start_preserves_root_attribution() -> Result<()> {
#[derive(Clone, Copy, PartialEq, Eq)]
enum ResidencyReload {
Sender,
SenderChangesWorkspace,
SenderPendingConfiguration,
OwnerChangesWorkspace,
OwnerNarrowsPermissions,
OwnerPreservesStricterChild,
OwnerRevokesWorkspaceRoot,
OwnerPendingConfiguration,
}
#[test_case(ResidencyReload::Sender; "sender preserves stricter child")]
#[test_case(ResidencyReload::SenderChangesWorkspace; "sender retains child workspace")]
#[test_case(ResidencyReload::SenderPendingConfiguration; "sender follows pending owner configuration")]
#[test_case(ResidencyReload::OwnerChangesWorkspace; "owner retains child workspace")]
#[test_case(ResidencyReload::OwnerNarrowsPermissions; "owner narrows cached permissions")]
#[test_case(ResidencyReload::OwnerPreservesStricterChild; "owner preserves stricter child")]
#[test_case(ResidencyReload::OwnerRevokesWorkspaceRoot; "owner revokes cached workspace root")]
#[test_case(ResidencyReload::OwnerPendingConfiguration; "owner pending configuration blocks cached authority")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn v2_residency_reload_preserves_inherited_environment_and_tools(
reload: ResidencyReload,
@@ -374,13 +384,23 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
let mut builder = test_codex()
.with_model("gpt-5.6-sol")
.with_exec_server_url("none")
.with_config(|config| {
for feature in [Feature::Collab, Feature::MultiAgentV2] {
.with_config(move |config| {
for feature in [
Feature::Collab,
Feature::MultiAgentV2,
Feature::SubagentThreadEnvironments,
] {
config
.features
.enable(feature)
.expect("test config should allow feature update");
}
if reload == ResidencyReload::OwnerChangesWorkspace {
config
.features
.enable(Feature::DeferredExecutor)
.expect("enable deferred executor");
}
config.multi_agent_v2.max_concurrent_threads_per_session = 2;
config
.permissions
@@ -390,20 +410,28 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
let test = builder.build_with_remote_and_local_env(&server).await?;
let mut child_environment = test.executor_environment().selection().clone();
let (child_permissions, parent_permissions) = match reload {
ResidencyReload::Sender => (
ResidencyReload::Sender
| ResidencyReload::SenderChangesWorkspace
| ResidencyReload::OwnerChangesWorkspace => (
PermissionProfile::read_only(),
PermissionProfile::workspace_write(),
),
ResidencyReload::SenderPendingConfiguration => (
PermissionProfile::read_only(),
PermissionProfile::read_only(),
),
ResidencyReload::OwnerNarrowsPermissions => {
(PermissionProfile::Disabled, PermissionProfile::read_only())
}
ResidencyReload::OwnerPreservesStricterChild => {
(PermissionProfile::read_only(), PermissionProfile::Disabled)
}
ResidencyReload::OwnerRevokesWorkspaceRoot => (
PermissionProfile::workspace_write(),
PermissionProfile::workspace_write(),
),
ResidencyReload::OwnerRevokesWorkspaceRoot | ResidencyReload::OwnerPendingConfiguration => {
(
PermissionProfile::workspace_write(),
PermissionProfile::workspace_write(),
)
}
};
if reload == ResidencyReload::OwnerRevokesWorkspaceRoot {
child_environment.cwd = test.workspace_path_uri("retained")?;
@@ -460,6 +488,30 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
/*connect_timeout*/ None,
)?;
}
let deferred_executor = if reload == ResidencyReload::OwnerChangesWorkspace {
let listener = TcpListener::bind("127.0.0.1:0").await?;
child_environment.environment_id = "deferred-child".to_string();
test.thread_manager
.environment_manager()
.upsert_environment(
child_environment.environment_id.clone(),
format!("ws://{}", listener.local_addr()?),
/*connect_timeout*/ None,
)?;
let (attach, connection) = tokio::sync::oneshot::channel();
let (shutdown, stop) = tokio::sync::oneshot::channel();
let executor = tokio::spawn(super::remote_env::serve_environment_with_agents_md(
listener, "", connection, stop,
));
Some((attach, shutdown, executor))
} else {
None
};
let pending_owner = reload == ResidencyReload::SenderPendingConfiguration;
let mut initial_environment = child_environment.clone();
if pending_owner {
initial_environment.config = EnvironmentConfigState::Pending;
}
let mut created_threads = test.thread_manager.subscribe_thread_created();
submit_thread_settings(
@@ -467,7 +519,7 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
ThreadSettingsOverrides {
environments: Some(TurnEnvironmentRequests::new(
test.config.cwd.clone(),
vec![child_environment.clone().into_request()],
vec![initial_environment.clone().into_request()],
)),
..Default::default()
},
@@ -481,16 +533,31 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
})
.await;
let evicted_thread = Arc::downgrade(&first_thread);
drop(first_thread);
let mut parent_environment = child_environment.clone();
if matches!(
reload,
ResidencyReload::SenderChangesWorkspace | ResidencyReload::OwnerChangesWorkspace
) {
parent_environment.cwd = test.workspace_path_uri("owner-only-root")?;
}
if reload == ResidencyReload::OwnerRevokesWorkspaceRoot {
parent_environment.workspace_roots.truncate(1);
} else if reload == ResidencyReload::OwnerPendingConfiguration {
parent_environment.config = EnvironmentConfigState::Pending;
} else {
let EnvironmentConfigState::Ready(parent_config) = &mut parent_environment.config else {
unreachable!("child environment config should be ready");
};
parent_config.permission_profile = PermissionProfileSnapshot::legacy(parent_permissions);
}
if reload == ResidencyReload::Sender {
let sender_reload = matches!(
reload,
ResidencyReload::Sender | ResidencyReload::SenderChangesWorkspace
) || pending_owner;
if sender_reload && !pending_owner {
submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
@@ -520,7 +587,17 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
.is_err()
);
if reload != ResidencyReload::Sender {
assert!(
evicted_thread.upgrade().is_none(),
"evicted child must be dropped"
);
// Complete startup only after eviction has captured the child as Starting.
let deferred_executor = deferred_executor.map(|(attach, shutdown, executor)| {
attach.send(()).expect("release executor startup");
(shutdown, executor)
});
if !sender_reload {
submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
@@ -536,13 +613,30 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
test.codex.config_snapshot().await.environments.environments,
vec![parent_environment]
);
let result = test
.thread_manager
.ensure_multi_agent_v2_child_loaded(first_thread_id)
.await;
// Resume is nonblocking: retry while the released executor finishes startup.
let result = tokio::time::timeout(Duration::from_secs(10), async {
loop {
let result = test
.thread_manager
.ensure_multi_agent_v2_child_loaded(first_thread_id)
.await;
if reload != ResidencyReload::OwnerChangesWorkspace
|| !result.as_ref().is_err_and(|err| {
err.to_string()
.contains("has no matching ready environment binding")
})
{
break result;
}
tokio::task::yield_now().await;
}
})
.await
.context("child should resume after deferred startup completes")?;
let expected_error = match reload {
ResidencyReload::OwnerRevokesWorkspaceRoot => {
Some("no longer matches a ready parent environment")
ResidencyReload::OwnerRevokesWorkspaceRoot
| ResidencyReload::OwnerPendingConfiguration => {
Some("has no matching ready environment binding")
}
ResidencyReload::OwnerNarrowsPermissions
| ResidencyReload::OwnerPreservesStricterChild
@@ -551,6 +645,9 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
Some("permissions changed on a remote executor")
}
ResidencyReload::Sender
| ResidencyReload::SenderChangesWorkspace
| ResidencyReload::SenderPendingConfiguration
| ResidencyReload::OwnerChangesWorkspace
| ResidencyReload::OwnerNarrowsPermissions
| ResidencyReload::OwnerPreservesStricterChild => None,
};
@@ -582,6 +679,34 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
matches!(event, EventMsg::TurnComplete(_))
})
.await;
if pending_owner {
assert_eq!(
reloaded_worker.environment_selections().await,
vec![initial_environment.clone()]
);
let EnvironmentConfigState::Ready(config) = &child_environment.config else {
unreachable!("expected resolved owner configuration");
};
test.codex
.environment_ready(&initial_environment, config.clone())
.await?;
let selections = tokio::time::timeout(Duration::from_secs(10), async {
loop {
let selections = reloaded_worker.environment_selections().await;
if selections
.iter()
.all(|selection| selection.config != EnvironmentConfigState::Pending)
{
break selections;
}
tokio::task::yield_now().await;
}
})
.await
.context("reloaded child should receive the original owner's configuration result")?;
assert_eq!(selections, vec![child_environment]);
return Ok(());
}
assert_eq!(
reloaded_worker
.config_snapshot()
@@ -618,6 +743,10 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools(
};
let reloaded_tools = worker_tools(&reloaded_worker_request);
assert!(reloaded_tools.to_string().contains("### `exec_command`"));
if let Some((shutdown, executor)) = deferred_executor {
shutdown.send(()).expect("stop executor fixture");
executor.await?;
}
Ok(())
}
codex-rs/core/tests/suite/remote_env.rs
+1 -1
View File
@@ -1456,7 +1456,7 @@ async fn serve_environment_info(listener: TcpListener) {
send_environment_info(&mut websocket).await;
}
async fn serve_environment_with_agents_md(
pub(super) async fn serve_environment_with_agents_md(
listener: TcpListener,
contents: &str,
attach: tokio::sync::oneshot::Receiver<()>,
codex-rs/core/tests/suite/scenarios_agent_eviction_tests.rs
+64 -2
View File
@@ -2,7 +2,10 @@
use super::*;
use codex_core::ThreadEvictionOutcome;
use codex_exec_server::CreateDirectoryOptions;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::protocol::TurnSettingsUpdate;
use codex_protocol::protocol::TurnSettingsUpdateOutcome;
use core_test_support::ThreadIdle;
use core_test_support::responses;
use core_test_support::submit_thread_settings;
@@ -20,6 +23,12 @@ async fn idle_eviction_preserves_worker_context_for_followup() -> Result<()> {
.with_extensions(Arc::new(extensions.build()))
.with_config(|config| {
configure_scenario_catalog(config);
// This scenario uses client_echo; omit platform-specific shell schemas.
config.features.disable(Feature::ShellTool).unwrap();
config
.features
.enable(Feature::SubagentThreadEnvironments)
.expect("enable configured environment inheritance");
config.features.enable(Feature::Collab).unwrap();
config.features.enable(Feature::MultiAgentV2).unwrap();
config
@@ -45,12 +54,26 @@ async fn idle_eviction_preserves_worker_context_for_followup() -> Result<()> {
}),
defer_loading: false,
})],
environments: Some(vec![test.executor_environment().request()]),
environments: Some(Vec::new()),
..StartThreadOptions::new(test.config.clone())
})
.await?;
test.codex = root.thread;
test.session_configured = root.session_configured;
let configured = test.codex.environment_selections().await;
let mut temporary = test.executor_environment().selection().clone();
temporary.cwd = test.workspace_path_uri("turn-only-workspace")?;
temporary.workspace_roots = vec![temporary.cwd.clone()];
test.fs()
.create_directory(
&temporary.cwd,
CreateDirectoryOptions {
recursive: false,
follow_symlinks: true,
},
/*sandbox*/ None,
)
.await?;
let parent_id = test.session_configured.session_id.to_string();
let mut created = test.thread_manager.subscribe_thread_created();
let mut child_id = None;
@@ -131,6 +154,7 @@ async fn idle_eviction_preserves_worker_context_for_followup() -> Result<()> {
if tool == "send_message" {
let id = child_id.expect("spawned child");
let child = test.thread_manager.get_thread(id).await?;
assert_eq!(child.environment_selections().await, configured);
// Flush submission dispatch so the note is queued before idle cleanup starts.
submit_thread_settings(&child, ThreadSettingsOverrides::default()).await?;
assert_eq!(
@@ -156,12 +180,32 @@ async fn idle_eviction_preserves_worker_context_for_followup() -> Result<()> {
child_id = Some(id);
}
let child = test.thread_manager.get_thread(id).await?;
assert_eq!(child.environment_selections().await, configured);
let call = wait_for_event_match(&child, |event| match event {
EventMsg::DynamicToolCallRequest(call) => Some(call.clone()),
_ => None,
})
.await;
assert_eq!(call.call_id, format!("worker-echo-{tool}"));
assert_eq!(
child
.current_turn_environment_selections(&call.turn_id)
.await,
Some(Vec::new()),
);
// A child-owned temporary attachment must disappear across idle eviction.
let (reply, outcome) = tokio::sync::oneshot::channel();
child
.submit(Op::TurnSettings {
turn_id: call.turn_id,
update: TurnSettingsUpdate {
environments: Some(vec![temporary.clone().into_request()]),
..Default::default()
},
reply,
})
.await?;
assert_eq!(outcome.await?, TurnSettingsUpdateOutcome::Applied);
responses::mount_sse_once_match(
&server,
wiremock::matchers::header("thread-id", id.to_string()),
@@ -193,8 +237,26 @@ async fn idle_eviction_preserves_worker_context_for_followup() -> Result<()> {
requests
.sort_by_key(|request| request.header("thread-id").as_deref() != Some(parent_id.as_str()));
assert_eq!(requests.len(), 10);
let temporary_workspaces: Vec<_> = requests
.iter()
.map(|request| {
request
.message_input_texts("user")
.into_iter()
.rev()
.find(|text| text.starts_with("<environment_context>"))
.expect("environment context for each request")
.contains("turn-only-workspace")
})
.collect();
assert_eq!(
temporary_workspaces,
[
false, false, false, false, false, false, false, true, false, true
]
);
let history = context_snapshot::format_request_history_snapshot(
"A parent delegates an API migration, queues a note while the worker is idle, and follows up after idle eviction. The same worker resumes with its earlier answer and the unread note. Complete model streams are grouped as parent, then worker.",
"A parent with no configured environment delegates an API migration, queues a note while the worker is idle, and follows up after idle eviction. The worker temporarily attaches a workspace, then resumes with no environment, its earlier answer, and the unread note. Complete model streams are grouped as parent, then worker.",
&requests,
&ContextSnapshotOptions::default().rewrite_known_segments(),
);
codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_eviction__idle_eviction_followup.snap
+17 -11
View File
@@ -1,14 +1,14 @@
---
source: core/tests/suite/scenarios_agent_eviction_tests.rs
assertion_line: 183
assertion_line: 275
expression: "format!(\"{history}\\n\\nResumed worker mailbox:\\n{mailbox}\")"
---
Scenario: A parent delegates an API migration, queues a note while the worker is idle, and follows up after idle eviction. The same worker resumes with its earlier answer and the unread note. Complete model streams are grouped as parent, then worker.
Scenario: A parent with no configured environment delegates an API migration, queues a note while the worker is idle, and follows up after idle eviction. The worker temporarily attaches a workspace, then resumes with no environment, its earlier answer, and the unread note. Complete model streams are grouped as parent, then worker.
## Window 1
-- request 1 (turn) --
00:additional_tools/developer (2; hash=122dc70cb8c7e8f0):
- namespace/functions; hash=f859506569b16c70
00:additional_tools/developer (2; hash=7958a01e34d95ac2):
- namespace/functions; hash=30a5fc1fcffb1c04
- custom/exec
- function/wait
- function/request_user_input
@@ -66,8 +66,8 @@ Scenario: A parent delegates an API migration, queues a note while the worker is
## Window 2 (after request 6: input diverged at item 03)
-- request 7 (turn) --
00:additional_tools/developer (2; hash=122dc70cb8c7e8f0):
- namespace/functions; hash=f859506569b16c70
00:additional_tools/developer (2; hash=7958a01e34d95ac2):
- namespace/functions; hash=30a5fc1fcffb1c04
- custom/exec
- function/wait
- function/request_user_input
@@ -108,14 +108,20 @@ Scenario: A parent delegates an API migration, queues a note while the worker is
-- request 8 (turn) --
07:function_call/client_echo:{"message":"Mira owns the API migration."}
08:function_call_output:Mira owns the API migration.
09:message/user:
<ENVIRONMENT_CONTEXT>
-- request 9 (turn) --
09:message/assistant:
10:message/assistant:
Mira owns the API migration.
10:agent_message
11:agent_message
11:message/user:
<ENVIRONMENT_CONTEXT>
12:agent_message
13:agent_message
-- request 10 (turn) --
12:function_call/client_echo:{"message":"I will continue with Mira using the new API."}
13:function_call_output:I will continue with Mira using the new API.
14:function_call/client_echo:{"message":"I will continue with Mira using the new API."}
15:function_call_output:I will continue with Mira using the new API.
16:message/user:
<ENVIRONMENT_CONTEXT>
Resumed worker mailbox:
[
codex-rs/core/tests/suite/step_settings/agent_spawn_tests.rs
+106 -6
View File
@@ -1,17 +1,41 @@
//! Spawned agents inherit the invoking step's model settings after active-turn updates.
//! Children inherit captured model settings and persistent environment attachments.
use super::*;
use codex_exec_server::CreateDirectoryOptions;
use codex_protocol::protocol::EnvironmentConfigState;
use codex_protocol::protocol::TurnEnvironmentRequests;
use codex_protocol::protocol::TurnEnvironmentSelection;
use core_test_support::submit_thread_settings;
use core_test_support::test_codex::environment_config_for_selection;
use pretty_assertions::assert_eq;
use test_case::test_case;
#[test_case(false, None; "v1 inherits captured settings")]
#[test_case(true, None; "v2 inherits captured settings")]
#[test_case(false, Some(ReasoningEffort::Medium); "v1 validates effort against captured model")]
#[test_case(true, Some(ReasoningEffort::Medium); "v2 validates effort against captured model")]
#[derive(Clone, Copy)]
enum EnvironmentInheritanceCase {
Configured,
Legacy,
Empty,
Fork,
Pending,
}
#[test_case(false, None, EnvironmentInheritanceCase::Configured; "v1 inherits captured settings")]
#[test_case(true, None, EnvironmentInheritanceCase::Configured; "v2 inherits captured settings")]
#[test_case(false, Some(ReasoningEffort::Medium), EnvironmentInheritanceCase::Configured; "v1 validates effort against captured model")]
#[test_case(true, Some(ReasoningEffort::Medium), EnvironmentInheritanceCase::Configured; "v2 validates effort against captured model")]
#[test_case(false, None, EnvironmentInheritanceCase::Empty; "v1 excludes turn-only attachments")]
#[test_case(true, None, EnvironmentInheritanceCase::Empty; "v2 excludes turn-only attachments")]
#[test_case(false, None, EnvironmentInheritanceCase::Fork; "v1 fork inherits configured attachments")]
#[test_case(true, None, EnvironmentInheritanceCase::Fork; "v2 fork inherits configured attachments")]
#[test_case(false, None, EnvironmentInheritanceCase::Pending; "v1 retains replaced pending attachment")]
#[test_case(true, None, EnvironmentInheritanceCase::Pending; "v2 retains replaced pending attachment")]
#[test_case(false, None, EnvironmentInheritanceCase::Legacy; "v1 retains turn attachments before rollout")]
#[test_case(true, None, EnvironmentInheritanceCase::Legacy; "v2 retains turn attachments before rollout")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn spawn_inherits_captured_settings_after_a_turn_update(
multi_agent_v2: bool,
requested_effort: Option<ReasoningEffort>,
inheritance: EnvironmentInheritanceCase,
) -> Result<()> {
skip_if_no_network!(Ok(()));
@@ -19,9 +43,16 @@ async fn spawn_inherits_captured_settings_after_a_turn_update(
let mut spawn_arguments = json!({ "message": "Complete the delegated task." });
let namespace = if multi_agent_v2 {
spawn_arguments["task_name"] = json!("worker");
spawn_arguments["fork_turns"] = json!("none");
spawn_arguments["fork_turns"] =
json!(if matches!(inheritance, EnvironmentInheritanceCase::Fork) {
"all"
} else {
"none"
});
"collaboration"
} else {
spawn_arguments["fork_context"] =
json!(matches!(inheritance, EnvironmentInheritanceCase::Fork));
"multi_agent_v1"
};
if let Some(effort) = &requested_effort {
@@ -50,6 +81,16 @@ async fn spawn_inherits_captured_settings_after_a_turn_update(
mount_sse_once(&server, sse_completed("resp-child-notification")).await;
let test = step_settings_test()
.with_config(move |config| {
if !matches!(inheritance, EnvironmentInheritanceCase::Legacy) {
config
.features
.enable(Feature::SubagentThreadEnvironments)
.expect("enable configured environment inheritance");
}
config
.features
.enable(Feature::DeferredExecutor)
.expect("enable deferred executor");
config
.features
.enable(Feature::Collab)
@@ -85,6 +126,56 @@ async fn spawn_inherits_captured_settings_after_a_turn_update(
.build_with_auto_env(&server)
.await?;
let mut created_threads = test.thread_manager.subscribe_thread_created();
if matches!(inheritance, EnvironmentInheritanceCase::Empty) {
submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
environments: Some(TurnEnvironmentRequests::new(
test.config.cwd.clone(),
Vec::new(),
)),
..Default::default()
},
)
.await?;
}
let mut configured = test.codex.environment_selections().await;
if matches!(inheritance, EnvironmentInheritanceCase::Pending) {
configured[0].config = EnvironmentConfigState::Pending;
submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
environments: Some(TurnEnvironmentRequests::new(
test.config.cwd.clone(),
configured
.clone()
.into_iter()
.map(TurnEnvironmentSelection::into_request)
.collect(),
)),
..Default::default()
},
)
.await?;
}
let mut temporary = test.executor_environment().selection().clone();
temporary.cwd = test.workspace_path_uri("turn-only-workspace")?;
temporary.workspace_roots = vec![temporary.cwd.clone()];
temporary.config =
EnvironmentConfigState::Ready(environment_config_for_selection(&test.config, &temporary));
test.fs()
.create_directory(
&temporary.cwd,
CreateDirectoryOptions {
recursive: false,
follow_symlinks: true,
},
/*sandbox*/ None,
)
.await?;
if matches!(inheritance, EnvironmentInheritanceCase::Legacy) {
configured = vec![temporary.clone()];
}
let request = start_paused_turn(&test.codex).await?;
apply_turn_settings(
&test.codex,
@@ -93,10 +184,18 @@ async fn spawn_inherits_captured_settings_after_a_turn_update(
model: Some(MODEL_B.to_string()),
effort: Some(Some(ReasoningEffort::High)),
summary: Some(ReasoningSummary::Detailed),
environments: Some(vec![temporary.into_request()]),
..Default::default()
},
)
.await?;
if matches!(inheritance, EnvironmentInheritanceCase::Pending) {
let config = environment_config_for_selection(&test.config, &configured[0]);
test.codex
.environment_ready(&configured[0], config.clone())
.await?;
configured[0].config = EnvironmentConfigState::Ready(config);
}
answer_paused_turn(&test.codex, &request.turn_id).await?;
let child_thread_id = tokio::time::timeout(
@@ -108,6 +207,7 @@ async fn spawn_inherits_captured_settings_after_a_turn_update(
for thread in [child_thread.as_ref(), test.codex.as_ref()] {
wait_for_event(thread, |event| matches!(event, EventMsg::TurnComplete(_))).await;
}
assert_eq!(child_thread.environment_selections().await, configured);
let expected_effort = requested_effort.unwrap_or(ReasoningEffort::High);
let requests = response_mock.requests();
codex-rs/features/src/lib.rs
+8
View File
@@ -232,6 +232,8 @@ pub enum Feature {
ModelCatalogInContext,
/// Inherit client-defined dynamic tools in fresh V2 subagents.
MultiAgentV2DynamicTools,
/// Inherit configured thread environments instead of temporary turn selections.
SubagentThreadEnvironments,
/// Use each subagent model's default context window and auto-compaction threshold.
SubagentDefaultContextLimits,
/// Keep sampling through reasoning and commentary boundaries when agent mail arrives.
@@ -1512,6 +1514,12 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::SubagentThreadEnvironments,
key: "subagent_thread_environments",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::SubagentDefaultContextLimits,
key: "subagent_default_context_limits",