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

Preserve pending environments when spawning subagents (#49075)

Browse Source 
## Why

Spawning a subagent before an environment was ready dropped the starting environment from the child's selections. The child needs to retain that environment and receive its configuration or failure when preparation finishes.

## What changed

- Pass the spawning step's environment snapshot to children, preserving ready and starting environments and configuration ownership.
- Propagate the original owner's first configuration result through descendants, applying each session's validation and leaving independently configured children unchanged.
- Preserve pending configuration waits across executor retries. Later owner updates do not overwrite the child's inherited configuration.

## Testing

Add regression coverage for fresh and forked subagents, readiness through turn-settings updates, failure propagation to children and grandchildren, child configuration precedence, and executor retries. Add a request snapshot showing a child using `wait_for_environment` before execution tools become available.

GitOrigin-RevId: a2afed740391f88d195596cfe28e418332b60d43
This commit is contained in:
sayan-oai
2026-09-28 20:42:04 +00:00
committed by copyberry
parent 22d1d9336f
commit 3749d1eff7
14 changed files with 899 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control/spawn.rs
+13 -5
View File
@@ -676,9 +676,13 @@ impl LocalAgentControl {
.registry
.reserve_spawn_slot(reservation_max_threads)?;
let inheritance = SpawnAgentThreadInheritance {
environments: self
.inherited_environments_for_source(&state, session_source.as_ref())
.await,
environments: match &options.environments {
Some(environments) => Some(environments.clone()),
None => {
self.inherited_environments_for_source(&state, session_source.as_ref())
.await
}
},
exec_policy: self
.inherited_exec_policy_for_source(&state, session_source.as_ref(), &config)
.await,
@@ -735,6 +739,10 @@ impl LocalAgentControl {
} else {
None
};
let environments = options
.environments
.as_ref()
.map(TurnEnvironmentSnapshot::inheritable_selections);
let child_create_started_at = Instant::now();
let new_thread = Box::pin(state.spawn_new_thread_with_source(
config.clone(),
@@ -747,7 +755,7 @@ impl LocalAgentControl {
/*metrics_service_name*/ None,
inheritance.environments,
inheritance.exec_policy,
options.environments.clone(),
environments,
))
.await?;
SpawnedThreadResult {
@@ -1229,7 +1237,7 @@ impl LocalAgentControl {
/*forked_from_thread_id*/ Some(parent_thread_id),
inherited_environments,
inherited_exec_policy,
options.environments.clone(),
/*environments*/ None,
thread_extension_init,
)
.await?;
codex-rs/core/src/agent/control_tests.rs
+69
View File
@@ -85,6 +85,7 @@ use codex_protocol::protocol::TokenUsageRecord;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::protocol::TurnEnvironmentSelections;
use codex_protocol::protocol::TurnStartedEvent;
use codex_thread_store::ArchiveThreadParams;
@@ -1649,6 +1650,74 @@ async fn spawn_agent_creates_thread_and_sends_prompt() {
wait_for_recorded_user_message(thread.as_ref(), "spawned").await;
}
#[tokio::test]
async fn pending_environment_failure_reaches_child_and_grandchild() {
let (home, mut config) = test_config().await;
for feature in [
Feature::DeferredExecutor,
Feature::MultiAgentV2,
Feature::Sqlite,
] {
config.features.enable(feature).expect("enable feature");
}
config.model = Some("gpt-5.6-sol".to_string());
config.agent_max_depth = 2;
let harness = AgentControlHarness::new_with_config(home, config).await;
let cwd = PathUri::from_abs_path(&harness.config.codex_home);
let pending = TurnEnvironmentSelection {
environment_id: codex_exec_server::LOCAL_ENVIRONMENT_ID.to_string(),
cwd: cwd.clone(),
workspace_roots: vec![cwd],
config: EnvironmentConfigState::Pending,
};
let root = harness
.manager
.start_thread(StartThreadOptions {
environments: Some(vec![pending.clone()]),
..StartThreadOptions::new(harness.config.clone())
})
.await
.expect("start root")
.thread;
let control = root
.session
.services
.local_agent_runtime
.control(root.session.session_id());
let mut parent = Arc::clone(&root);
let mut descendants = Vec::new();
for name in ["child", "grandchild"] {
let agent =
spawn_v2_reload_test_child(&control, harness.config.clone(), &parent, name).await;
let thread = harness
.manager
.get_thread(agent.thread_id)
.await
.expect("get descendant");
assert_eq!(thread.environment_selections().await, vec![pending.clone()]);
descendants.push(Arc::clone(&thread));
parent = thread;
}
let error = "root could not prepare the environment";
root.environment_failed(&pending, error.to_string())
.await
.expect("fail root environment");
let failed = TurnEnvironmentSelection {
config: EnvironmentConfigState::Failed(error.to_string()),
..pending
};
timeout(Duration::from_secs(/*secs*/ 5), async {
for thread in descendants {
while thread.environment_selections().await != [failed.clone()] {
sleep(Duration::from_millis(/*millis*/ 10)).await;
}
}
})
.await
.expect("both descendants should receive the root failure");
}
#[tokio::test]
async fn ephemeral_spawn_does_not_persist_agent_graph_edge() {
let (home, mut config) = test_config().await;
codex-rs/core/src/agent/types.rs
+3 -2
View File
@@ -2,10 +2,10 @@
//! Backend operations and local admission policy live in the implementations.
use crate::context::MultiAgentRoleInstructions;
use crate::environment_selection::TurnEnvironmentSnapshot;
use codex_protocol::AgentPath;
use codex_protocol::ThreadId;
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::turn_input::CyberAccessProgram;
/// Registry identity shared by loaded and unloaded agents.
@@ -33,7 +33,8 @@ pub struct SpawnAgentOptions {
/// Attribute delegated usage to the turn that initiated it.
pub turn_trigger: Option<String>,
pub root_turn_id: Option<String>,
pub environments: Option<Vec<TurnEnvironmentSelection>>,
/// Exact environment bindings from the step that requested the child.
pub environments: Option<TurnEnvironmentSnapshot>,
pub multi_agent_v2_usage_hints: Option<ResolvedMultiAgentV2UsageHints>,
pub cyber_access_program: Option<CyberAccessProgram>,
}
codex-rs/core/src/environment_selection.rs
+159 -61
View File
@@ -32,7 +32,7 @@ use codex_utils_path_uri::PathUri;
use futures::FutureExt;
use futures::future::BoxFuture;
use futures::future::Shared;
use tokio::sync::mpsc;
use tokio::sync::watch;
use tokio_util::task::AbortOnDropHandle;
use tracing::Instrument;
use tracing::instrument::WithSubscriber;
@@ -171,6 +171,7 @@ pub fn validate_environment_ids_and_cwds(
type TurnEnvironmentResult = Result<ResolvedEnvironment, Arc<ExecServerError>>;
type TurnEnvironmentResolution = Shared<BoxFuture<'static, TurnEnvironmentResult>>;
type PendingConfigurationResult = Result<EnvironmentConfig, String>;
pub(crate) type PendingConfiguration = Shared<BoxFuture<'static, PendingConfigurationResult>>;
// Shared startup result used to build each turn's environment with its own config
// without restarting the connection or shell resolution.
@@ -196,7 +197,10 @@ struct SelectedTurnEnvironment {
// Selection clones share one listener; the final handle drop aborts it.
connection_events_task: Option<Arc<AbortOnDropHandle<()>>>,
resolution: TurnEnvironmentResolution,
pending_completion: Option<mpsc::Sender<PendingConfigurationResult>>,
// 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.
owner_config_result: Option<PendingConfiguration>,
}
#[derive(Clone)]
@@ -204,6 +208,8 @@ pub(crate) struct StartingTurnEnvironment {
pub(crate) selection: TurnEnvironmentSelection,
config_origin: EnvironmentConfigOrigin,
resolution: TurnEnvironmentResolution,
environment: Arc<Environment>,
owner_config_result: Option<PendingConfiguration>,
}
/// Uses the current thread defaults when requested and records who owns later config updates.
@@ -244,6 +250,14 @@ impl fmt::Debug for StartingTurnEnvironment {
}
impl StartingTurnEnvironment {
pub(crate) fn owner_configuration(&self) -> Option<PendingConfiguration> {
if matches!(self.selection.config, EnvironmentConfigState::Pending) {
self.owner_config_result.clone()
} else {
None
}
}
#[tracing::instrument(
name = "environments.wait_until_ready",
skip_all,
@@ -277,13 +291,30 @@ impl ThreadEnvironments {
current: TurnEnvironmentSnapshot,
non_blocking_snapshots: bool,
) -> Self {
// Reuse only attached environments from the supplied snapshot; drop starting entries.
let environments = current
.environments
.into_iter()
.filter_map(|environment| {
let TurnEnvironmentState::Ready(environment) = environment else {
return None;
.filter_map(|inherited| {
let environment = match inherited {
TurnEnvironmentState::Ready(environment) => environment,
TurnEnvironmentState::Starting(inherited) => {
let (selection, config_origin) = resolve_selection_config(
inherited
.config_origin
.into_input_selection(inherited.selection.clone()),
&thread_defaults,
);
return Some(Self::start_environment(
selection,
config_origin,
Arc::clone(&inherited.environment),
inherited.owner_configuration(),
/*pending_completion*/ None,
&local_shell,
&shell_snapshot,
));
}
TurnEnvironmentState::Failed { .. } => return None,
};
let selection = environment.selection;
let config_origin = environment.config_origin;
@@ -319,6 +350,7 @@ impl ThreadEnvironments {
connection_events_task: None,
resolution,
pending_completion: None,
owner_config_result: None,
};
// Child threads re-infer thread-owned config while preserving owner config.
inherited_environment.refresh_thread_config(&thread_defaults);
@@ -381,16 +413,34 @@ impl ThreadEnvironments {
} = &mut *state;
let mut seen_environment_ids = HashSet::with_capacity(environments.len());
let mut next = Vec::with_capacity(environments.len());
let mut pending_completions = Vec::new();
for selected_environment in environments {
if !seen_environment_ids.insert(selected_environment.environment_id.as_str()) {
continue;
}
if let Some(environment) = current.iter().find(|environment| {
let previous = current.iter().find(|environment| {
let previous = &environment.selection;
previous.environment_id == selected_environment.environment_id
&& previous.cwd == selected_environment.cwd
&& previous.workspace_roots == selected_environment.workspace_roots
}) {
});
let (selected_environment, config_origin) =
resolve_selection_config(selected_environment.clone(), thread_defaults);
if let Some(environment) = previous {
let mut environment = environment.clone();
if let Some(completion) = environment.pending_completion.take() {
match &selected_environment.config {
EnvironmentConfigState::Ready(config) => {
pending_completions.push((completion, Ok(config.clone())));
}
EnvironmentConfigState::Failed(error) => {
pending_completions.push((completion, Err(error.clone())));
}
EnvironmentConfigState::FromThread | EnvironmentConfigState::Pending => {
environment.pending_completion = Some(completion);
}
}
}
let failed =
matches!(
environment.selection.config,
@@ -404,15 +454,12 @@ impl ThreadEnvironments {
);
if !failed && !restarting_as_pending {
let (selected_environment, config_origin) =
resolve_selection_config(selected_environment.clone(), thread_defaults);
let shell_settings_changed = matches!(
(&environment.selection.config, &selected_environment.config),
(EnvironmentConfigState::Ready(previous), EnvironmentConfigState::Ready(current))
if previous.allow_login_shell != current.allow_login_shell
|| previous.shell_environment_policy != current.shell_environment_policy
);
let mut environment = environment.clone();
environment.selection = selected_environment;
environment.config_origin = config_origin;
if shell_settings_changed
@@ -432,8 +479,6 @@ impl ThreadEnvironments {
tracing::warn!("skipping unknown turn environment `{environment_id}`");
continue;
};
let (selected_environment, config_origin) =
resolve_selection_config(selected_environment.clone(), thread_defaults);
// Connection state belongs to the environment instance, not its cwd or roots.
let connection_events_task = current
.iter()
@@ -451,33 +496,20 @@ impl ThreadEnvironments {
)
})
});
let (pending_completion, configuration_ready) =
if matches!(selected_environment.config, EnvironmentConfigState::Pending) {
let (sender, receiver) = mpsc::channel(/*buffer*/ 1);
(Some(sender), Some(receiver))
} else {
(None, None)
};
let (resolution_task, resolution) = Self::resolve_environment(
selected_environment.clone(),
Arc::clone(&environment),
self.local_shell.clone(),
self.shell_snapshot.clone(),
configuration_ready,
)
.remote_handle();
drop(tokio::spawn(
resolution_task.in_current_span().with_current_subscriber(),
));
let resolution = resolution.boxed().shared();
let selected = SelectedTurnEnvironment {
selection: selected_environment,
let pending = previous.filter(|previous| {
matches!(selected_environment.config, EnvironmentConfigState::Pending)
&& previous.pending_completion.is_some()
});
let mut selected = Self::start_environment(
selected_environment,
config_origin,
environment,
connection_events_task,
resolution,
pending_completion,
};
pending.and_then(|previous| previous.owner_config_result.clone()),
pending.and_then(|previous| previous.pending_completion.clone()),
&self.local_shell,
&self.shell_snapshot,
);
selected.connection_events_task = connection_events_task;
next.push(selected);
}
let removed_connection_tasks = current
@@ -494,17 +526,9 @@ impl ThreadEnvironments {
.collect::<Vec<_>>();
let previous = std::mem::replace(current, next);
// Publish owner configuration before waking turns waiting on this attachment.
for environment in current.iter() {
let Some(completion) = &environment.pending_completion else {
continue;
};
let result = match &environment.selection.config {
EnvironmentConfigState::Ready(config) => Ok(config.clone()),
EnvironmentConfigState::Failed(error) => Err(error.clone()),
EnvironmentConfigState::FromThread | EnvironmentConfigState::Pending => continue,
};
let _ = completion.try_send(result);
// Publish after installing the new list, including pending bindings replaced by retries.
for (completion, result) in pending_completions {
let _ = completion.send_replace(Some(result));
}
for task in removed_connection_tasks {
@@ -514,6 +538,60 @@ impl ThreadEnvironments {
drop(previous);
}
fn start_environment(
selection: TurnEnvironmentSelection,
config_origin: EnvironmentConfigOrigin,
environment: Arc<Environment>,
owner_config_result: Option<PendingConfiguration>,
pending_completion: Option<watch::Sender<Option<PendingConfigurationResult>>>,
local_shell: &Shell,
shell_snapshot: &ShellSnapshot,
) -> SelectedTurnEnvironment {
let (pending_completion, configuration_ready) =
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();
(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),
local_shell.clone(),
shell_snapshot.clone(),
configuration_ready,
)
.remote_handle();
drop(tokio::spawn(
task.in_current_span().with_current_subscriber(),
));
SelectedTurnEnvironment {
selection,
config_origin,
environment,
connection_events_task: None,
resolution: resolution.boxed().shared(),
pending_completion,
owner_config_result,
}
}
/// Projects canonical selections back to caller input form without losing config ownership.
pub(crate) fn selections(&self) -> Vec<TurnEnvironmentSelection> {
self.state
@@ -738,7 +816,7 @@ impl ThreadEnvironments {
environment: Arc<Environment>,
local_shell: Shell,
shell_snapshot: ShellSnapshot,
configuration_ready: Option<mpsc::Receiver<PendingConfigurationResult>>,
configuration_ready: Option<PendingConfiguration>,
) -> TurnEnvironmentResult {
let environment_id = &selection.environment_id;
if let EnvironmentConfigState::Failed(error) = &selection.config {
@@ -752,18 +830,15 @@ impl ThreadEnvironments {
Arc::new(error)
})
};
// Wait for this thread attachment's owner configuration, if pending.
// Wait for this thread to accept the configuration, if pending.
let configuration_ready = async move {
let Some(mut configuration_ready) = configuration_ready else {
let Some(configuration_ready) = configuration_ready else {
return Ok(None);
};
match configuration_ready.recv().await {
Some(Ok(config)) => Ok(Some(config)),
Some(Err(error)) => Err(Arc::new(ExecServerError::Protocol(error))),
None => Err(Arc::new(ExecServerError::Protocol(
"environment configuration was canceled".to_string(),
))),
}
configuration_ready
.await
.map(Some)
.map_err(|error| Arc::new(ExecServerError::Protocol(error)))
};
// Resolve the attachment only after both prerequisites are ready.
let ((), installed_config) = tokio::try_join!(connection_ready, configuration_ready)?;
@@ -857,6 +932,8 @@ impl ThreadEnvironments {
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)
@@ -1134,6 +1211,22 @@ impl TurnEnvironmentSnapshot {
.collect()
}
/// Native children retain ready and starting attachments captured by their spawning step.
pub(crate) fn inheritable_selections(&self) -> Vec<TurnEnvironmentSelection> {
self.environments
.iter()
.filter_map(|environment| match environment {
TurnEnvironmentState::Ready(environment) => Some(environment.selection()),
TurnEnvironmentState::Starting(environment) => Some(
environment
.config_origin
.into_input_selection(environment.selection.clone()),
),
TurnEnvironmentState::Failed { .. } => None,
})
.collect()
}
/// Returns every captured selection, including those still starting or unable to connect.
pub(crate) fn all_selections(&self) -> Vec<TurnEnvironmentSelection> {
self.environments
@@ -2297,8 +2390,8 @@ url = "ws://127.0.0.1:8765"
assert_eq!(multiple.local_environment_cwd(), Some(cwd));
}
#[test]
fn local_environment_cwd_uses_starting_local_selection() {
#[tokio::test]
async fn local_environment_cwd_uses_starting_local_selection() {
let cwd = AbsolutePathBuf::current_dir()
.expect("cwd")
.join("starting-local");
@@ -2312,6 +2405,11 @@ url = "ws://127.0.0.1:8765"
},
config_origin: EnvironmentConfigOrigin::Thread,
resolution: futures::future::pending().boxed().shared(),
environment: Arc::new(
Environment::create_for_tests(/*exec_server_url*/ None)
.expect("local environment"),
),
owner_config_result: None,
})],
};
codex-rs/core/src/environment_selection_snapshot_tests.rs
+67
View File
@@ -56,6 +56,73 @@ async fn unpolled_snapshot_does_not_delay_canceling_a_removed_environment() {
));
}
#[tokio::test]
async fn updating_another_environment_retries_the_executor_without_canceling_pending_config() {
let manager = Arc::new(EnvironmentManager::default_for_tests());
manager
.upsert_environment(
"failing".to_string(),
"http://example.com".to_string(),
/*connect_timeout*/ None,
)
.unwrap();
let cwd = PathUri::from_abs_path(&AbsolutePathBuf::current_dir().unwrap());
let mut local = TurnEnvironmentSelection {
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
cwd,
workspace_roots: Vec::new(),
config: EnvironmentConfigState::FromThread,
};
let mut pending = TurnEnvironmentSelection {
environment_id: "failing".to_string(),
config: EnvironmentConfigState::Pending,
..local.clone()
};
let environments = ThreadEnvironments::new(
manager,
crate::shell::default_user_shell(),
ThreadEnvironmentDefaults::new(tests::test_environment_config(), SandboxType::None),
ShellSnapshot::disabled(),
TurnEnvironmentSnapshot::default(),
/*non_blocking_snapshots*/ true,
);
let current = || {
let state = environments.state.lock().unwrap();
let failing = &state.environments[1];
(
failing.resolution.clone(),
failing.pending_completion.as_ref().unwrap().subscribe(),
failing.owner_config_result.clone().unwrap(),
)
};
environments.update_selections(&[local.clone(), pending.clone()]);
let (first_attempt, mut original_config, original_owner) = current();
assert!(first_attempt.clone().await.is_err());
local.config = EnvironmentConfigState::Ready(tests::test_environment_config());
environments.update_selections(&[local.clone(), pending.clone()]);
let (second_attempt, _, second_owner) = current();
assert!(!first_attempt.ptr_eq(&second_attempt));
assert!(original_owner.ptr_eq(&second_owner));
assert!(second_attempt.await.is_err());
assert!(original_config.borrow().is_none());
let config = tests::test_environment_config();
pending.config = EnvironmentConfigState::Ready(config.clone());
environments.update_selections(&[local, pending]);
let received = timeout(
Duration::from_secs(/*secs*/ 5),
original_config.wait_for(Option::is_some),
)
.await
.unwrap()
.unwrap()
.clone();
assert_eq!(received, Some(Ok(config.clone())));
assert_eq!(original_owner.await, Ok(config));
}
#[tokio::test]
async fn credential_refresh_does_not_restore_a_removed_environment() {
let home = tempfile::tempdir().expect("home");
codex-rs/core/src/session/environment.rs
+70 -9
View File
@@ -1,4 +1,5 @@
use std::collections::HashSet;
use std::sync::Arc;
use codex_exec_server::LOCAL_ENVIRONMENT_ID;
use codex_exec_server::MAX_SELECTED_CAPABILITY_ROOTS;
@@ -7,6 +8,7 @@ use codex_execpolicy::Policy;
use codex_protocol::capabilities::CapabilityRootLocation;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::EnvironmentConfig;
use codex_protocol::protocol::EnvironmentConfigState;
use codex_protocol::protocol::TurnEnvironmentSelection;
@@ -151,6 +153,11 @@ fn validate_environment_config(
Ok(())
}
enum ConfigUpdateSource {
Direct,
Inherited,
}
impl Session {
pub(super) fn apply_session_settings(
&self,
@@ -211,8 +218,7 @@ impl Session {
selection: &TurnEnvironmentSelection,
config: EnvironmentConfig,
) -> CodexResult<()> {
validate_environment_config(selection, &config)?;
self.update_environment_configuration(selection, EnvironmentConfigState::Ready(config))
self.update_environment_configuration(selection, Ok(config), ConfigUpdateSource::Direct)
.await
}
@@ -221,10 +227,49 @@ impl Session {
selection: &TurnEnvironmentSelection,
error: String,
) -> CodexResult<()> {
self.update_environment_configuration(selection, EnvironmentConfigState::Failed(error))
self.update_environment_configuration(selection, Err(error), ConfigUpdateSource::Direct)
.await
}
/// Installs the original owner's first result through this session's usual configuration checks.
pub(super) fn follow_inherited_environment_configurations(
self: &Arc<Self>,
inherited: &TurnEnvironmentSnapshot,
selected: &[TurnEnvironmentSelection],
) {
for starting in inherited
.starting()
.filter(|starting| selected.contains(&starting.selection))
{
let Some(configuration) = starting.owner_configuration() else {
continue;
};
let selection = starting.selection.clone();
let session = Arc::downgrade(self);
let mut status = self.agent_status.subscribe();
drop(tokio::spawn(async move {
let result = tokio::select! {
result = configuration => result,
_ = status.wait_for(|status| matches!(status, AgentStatus::Shutdown)) => return,
};
let Some(session) = session.upgrade() else {
return;
};
let apply = |result| {
session.update_environment_configuration(
&selection,
result,
ConfigUpdateSource::Inherited,
)
};
if let Err(error) = apply(result).await {
// Unlike the external owner, no one will retry this child's one-time update.
let _ = apply(Err(error.to_string())).await;
}
}));
}
}
/// Records configuration, or a failure to obtain it, for this environment and workspace.
/// The running turn and future turns may use different environments. Update the matching
/// selection or selections so work already waiting can still finish, and check successful
@@ -232,21 +277,32 @@ impl Session {
async fn update_environment_configuration(
&self,
selection: &TurnEnvironmentSelection,
config: EnvironmentConfigState,
result: Result<EnvironmentConfig, String>,
source: ConfigUpdateSource,
) -> CodexResult<()> {
// Serialize owner callbacks with ordinary thread settings updates.
let mut state = self.state.lock().await;
let mut current = self.services.turn_environments.selections();
let mut future = state.session_configuration.environments.clone();
let inherited = matches!(source, ConfigUpdateSource::Inherited);
let matches = |environment: &TurnEnvironmentSelection| {
environment.environment_id == selection.environment_id
&& environment.cwd == selection.cwd
&& environment.workspace_roots == selection.workspace_roots
&& (!inherited || matches!(environment.config, EnvironmentConfigState::Pending))
};
let (current_environment, future_environment) = match (
let environments = (
current.iter_mut().find(|environment| matches(environment)),
future.iter_mut().find(|environment| matches(environment)),
) {
);
// Ignore the parent's result if neither the running turn nor future turns are still
// waiting for this environment's configuration.
if inherited && matches!(environments, (None, None)) {
return Ok(());
}
if let Ok(config) = &result {
validate_environment_config(selection, config)?;
}
let (current_environment, future_environment) = match environments {
(None, None) => {
return Err(CodexErr::InvalidRequest(format!(
"environment `{}` is not selected on this thread with the requested workspace",
@@ -265,6 +321,11 @@ impl Session {
(Some(current), future) => (Some(current), future),
(None, Some(future)) => (None, Some(future)),
};
let successful = result.is_ok();
let config = match result {
Ok(config) => EnvironmentConfigState::Ready(config),
Err(error) => EnvironmentConfigState::Failed(error),
};
let update_current = current_environment.is_some();
let update_future = future_environment.is_some();
@@ -272,9 +333,9 @@ impl Session {
environment.config = config.clone();
}
if let Some(environment) = future_environment {
environment.config = config.clone();
environment.config = config;
}
if matches!(config, EnvironmentConfigState::Ready(_)) {
if successful {
let validate = |environments: &[TurnEnvironmentSelection]| {
state
.session_configuration
codex-rs/core/src/session/session.rs
+6 -1
View File
@@ -1427,6 +1427,7 @@ impl Session {
} else {
ShellSnapshot::disabled()
};
let inherited_environments = inherited_environments.unwrap_or_default();
let turn_environments = Arc::new(ThreadEnvironments::new(
environment_manager,
default_shell.clone(),
@@ -1435,7 +1436,7 @@ impl Session {
session_configuration.windows_sandbox_type,
),
shell_snapshot,
inherited_environments.unwrap_or_default(),
inherited_environments.clone(),
config.features.enabled(Feature::DeferredExecutor),
));
turn_environments.update_selections(environment_selections);
@@ -1884,6 +1885,10 @@ impl Session {
)
.await?;
sess.start_mcp_prewarm_worker(mcp_prewarm_rx, mcp_auth_changes);
sess.follow_inherited_environment_configurations(
&inherited_environments,
&session_configuration.environments,
);
sess.schedule_startup_prewarm(super::startup_prewarm::PrewarmInput::Base)
.await;
let session_start_source = match &initial_history {
codex-rs/core/src/thread_manager.rs
+18 -13
View File
@@ -1969,16 +1969,11 @@ impl ThreadManagerState {
None => self.client_mcp_extensions_for_child(parent_thread_id).await,
};
let thread_source = initial_history.get_resumed_thread_source();
let environments = environment_selections.or_else(|| {
inherited_environments
.as_ref()
.map(TurnEnvironmentSnapshot::to_selections)
});
let options = StartThreadOptions {
initial_history,
session_source: Some(session_source),
thread_source,
environments,
environments: environment_selections,
client_mcp_extensions,
..StartThreadOptions::new(config)
};
@@ -2092,13 +2087,23 @@ impl ThreadManagerState {
}
});
thread_extension_init.insert(isolation);
let environments = environments.unwrap_or_else(|| {
default_thread_environment_selections(
self.environment_manager.as_ref(),
&config.cwd,
&config.workspace_roots,
)
});
let environments = environments
.or_else(|| {
let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. }) = &session_source
else {
return None;
};
inherited_environments
.as_ref()
.map(TurnEnvironmentSnapshot::inheritable_selections)
})
.unwrap_or_else(|| {
default_thread_environment_selections(
self.environment_manager.as_ref(),
&config.cwd,
&config.workspace_roots,
)
});
let is_resumed_thread = matches!(&initial_history, InitialHistory::Resumed(_));
if reserved_thread_id.is_some() && matches!(&initial_history, InitialHistory::Resumed(_)) {
return Err(CodexErr::InvalidRequest(
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.to_selections()),
environments: Some(step_context.environments.clone()),
multi_agent_v2_usage_hints: None,
cyber_access_program: turn.cyber_access_program,
},
codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs
+1 -1
View File
@@ -199,7 +199,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.to_selections()),
environments: Some(step_context.environments.clone()),
multi_agent_v2_usage_hints,
cyber_access_program: turn.cyber_access_program,
},
codex-rs/core/tests/suite/remote_env.rs
+2
View File
@@ -1,5 +1,7 @@
#[path = "guardian_environments_tests.rs"]
mod guardian_environments;
#[path = "remote_env_spawn_tests.rs"]
pub(super) mod spawn_tests;
use anyhow::Context;
use anyhow::Result;
codex-rs/core/tests/suite/remote_env_spawn_tests.rs
+306
View File
@@ -0,0 +1,306 @@
//! Model coverage for subagents inheriting the first pending environment result.
use super::*;
use core_test_support::responses::sse_response;
use pretty_assertions::assert_eq;
use test_case::test_case;
const SPAWN: &str = "spawn-environment-worker";
const WAIT_CHILD: &str = "wait-for-environment-worker";
const WAIT_ENV: &str = "wait-for-inherited-environment";
const NEXT_TURN: &str = "child-next-turn-after-owner-change";
const FAILURE: &str = "owner could not prepare the subagent environment";
#[derive(Clone, Copy)]
pub(crate) enum PendingSpawnCase {
TurnSettings,
V1Fork,
Failure,
FirstResultOnly,
ChildConfiguredFirst,
}
fn call_output<'a>(body: &'a Value, call_id: &str) -> Option<&'a str> {
let output = body["input"]
.as_array()?
.iter()
.find(|item| item["type"] == "function_call_output" && item["call_id"] == call_id)?;
let output = &output["output"];
output
.as_str()
.or_else(|| output["content"].as_str())
.or_else(|| output[0]["text"].as_str())
}
// The parent and child run concurrently; these stages give the snapshot a stable causal order.
fn request_stage(body: &Value, root_id: &str) -> usize {
if body["client_metadata"]["thread_id"] == root_id {
if call_output(body, WAIT_CHILD).is_some() {
4
} else if call_output(body, SPAWN).is_some() {
1
} else {
0
}
} else if body["input"].as_array().is_some_and(|items| {
items
.iter()
.any(|item| item["role"] == "user" && item["content"].to_string().contains(NEXT_TURN))
}) {
5
} else if call_output(body, WAIT_ENV).is_some() {
3
} else {
2
}
}
pub(crate) async fn pending_subagent_scenario(
case: PendingSpawnCase,
configure_scenario: impl FnOnce(&mut Config) + Send + 'static,
) -> Result<Vec<Value>> {
let v1 = matches!(case, PendingSpawnCase::V1Fork);
let failed = matches!(case, PendingSpawnCase::Failure);
let server = start_mock_server().await;
let test = test_codex_with_wait_for_environment()
.with_config(move |config| {
config.project_doc_max_bytes = 0;
for (feature, enabled) in [
(Feature::DeferredExecutor, true),
(Feature::Collab, true),
(Feature::DeferMailboxPreemption, true),
(Feature::EnableRequestCompression, false),
(Feature::MultiAgentV2, !v1),
] {
config
.features
.set_enabled(feature, enabled)
.expect("configure subagent feature");
}
configure_scenario(config);
})
.build_with_auto_env(&server)
.await?;
let selection = test.executor_environment().selection().clone();
let pending = TurnEnvironmentSelection {
config: EnvironmentConfigState::Pending,
..selection.clone()
};
let root = test
.thread_manager
.start_thread(StartThreadOptions {
environments: Some(vec![pending.clone()]),
..StartThreadOptions::new(test.config.clone())
})
.await?;
let root_id = root.thread_id.to_string();
let matcher_root_id = root_id.clone();
let environment_id = selection.environment_id.clone();
let parent_requested = Arc::new(tokio::sync::Notify::new());
let child_requested = Arc::new(tokio::sync::Notify::new());
let notify_parent_request = Arc::clone(&parent_requested);
let notify_child_request = Arc::clone(&child_requested);
Mock::given(method("POST"))
.and(path("/v1/responses"))
.respond_with(move |request: &wiremock::Request| {
let body: Value = request.body_json().expect("model request body");
let namespace = if v1 {
"multi_agent_v1"
} else {
"collaboration"
};
let tool = |id, namespace, name, args: Value| {
ev_function_call_with_namespace(id, namespace, name, &args.to_string())
};
let stage = request_stage(&body, &matcher_root_id);
let item = match stage {
0 => {
let message = "Wait for the shared workspace to become ready.";
let args = if v1 {
json!({"message": message, "fork_context": true})
} else {
json!({"task_name": "worker", "message": message, "fork_turns": "none"})
};
tool(SPAWN, namespace, "spawn_agent", args)
}
1 => {
notify_parent_request.notify_one();
let mut args = json!({"timeout_ms": 10000});
if v1 {
let output = call_output(&body, SPAWN).expect("spawn result");
let output: Value =
serde_json::from_str(output).expect("spawn result JSON");
args["targets"] = json!([output["agent_id"]]);
}
tool(WAIT_CHILD, namespace, "wait_agent", args)
}
2 => {
notify_child_request.notify_one();
tool(
WAIT_ENV,
"functions",
"wait_for_environment",
json!({"environment_id": environment_id}),
)
}
3..=5 => ev_assistant_message(&format!("message-{stage}"), "Done."),
_ => unreachable!("known request stage"),
};
let response_id = format!("response-{stage}");
sse_response(sse(vec![
ev_response_created(&response_id),
item,
ev_completed(&response_id),
]))
})
.mount(&server)
.await;
let mut created = test.thread_manager.subscribe_thread_created();
let TurnInputSubmission::Started { turn_id } = root
.thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "Delegate waiting for the shared workspace, and report what the worker finds."
.into(),
text_elements: Vec::new(),
}]))
.await?
else {
anyhow::bail!("expected the parent's turn to start");
};
let child_id = timeout(Duration::from_secs(/*secs*/ 10), created.recv()).await??;
let child = test.thread_manager.get_thread(child_id).await?;
timeout(Duration::from_secs(/*secs*/ 10), async {
tokio::join!(parent_requested.notified(), child_requested.notified());
})
.await?;
assert_eq!(child.environment_selections().await, vec![pending.clone()]);
let mut first = environment_config_for_selection(&test.config, &selection);
if test.config.permissions.permission_profile() != &PermissionProfile::read_only() {
first.permission_profile =
PermissionProfileSnapshot::legacy(PermissionProfile::read_only());
}
if matches!(case, PendingSpawnCase::FirstResultOnly) {
first.allow_login_shell = false;
}
let expected = TurnEnvironmentSelection {
config: if failed {
EnvironmentConfigState::Failed(FAILURE.to_string())
} else {
EnvironmentConfigState::Ready(first.clone())
},
..selection
};
if matches!(case, PendingSpawnCase::ChildConfiguredFirst) {
child.environment_ready(&pending, first.clone()).await?;
}
match case {
PendingSpawnCase::TurnSettings => {
let (reply, outcome) = tokio::sync::oneshot::channel();
root.thread
.submit(Op::TurnSettings {
turn_id,
update: TurnSettingsUpdate {
environments: Some(vec![expected.clone()]),
..Default::default()
},
reply,
})
.await?;
assert_eq!(outcome.await?, TurnSettingsUpdateOutcome::Applied);
}
PendingSpawnCase::Failure | PendingSpawnCase::ChildConfiguredFirst => {
root.thread
.environment_failed(&pending, FAILURE.to_string())
.await?
}
PendingSpawnCase::V1Fork | PendingSpawnCase::FirstResultOnly => {
root.thread
.environment_ready(&pending, first.clone())
.await?;
}
}
tokio::join!(
wait_for_event(&root.thread, |event| matches!(
event,
EventMsg::TurnComplete(_)
)),
wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))),
);
if matches!(case, PendingSpawnCase::ChildConfiguredFirst) {
// The inherited callback runs separately; give it time to overwrite the child if unguarded.
tokio::time::sleep(Duration::from_millis(250)).await;
}
assert_eq!(child.environment_selections().await, vec![expected.clone()]);
if matches!(case, PendingSpawnCase::FirstResultOnly) {
let mut later = first;
later.allow_login_shell = true;
root.thread
.environment_ready(&pending, later.clone())
.await?;
assert_eq!(
root.thread.environment_selections().await,
vec![TurnEnvironmentSelection {
config: EnvironmentConfigState::Ready(later),
..pending.clone()
}]
);
child
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: NEXT_TURN.into(),
text_elements: Vec::new(),
}]))
.await?;
wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await;
assert_eq!(child.environment_selections().await, vec![expected]);
}
let mut requests = server
.received_requests()
.await
.context("recorded model requests")?
.into_iter()
.filter(|request| request.url.path() == "/v1/responses")
.map(|request| request.body_json().context("model request body"))
.collect::<Result<Vec<Value>>>()?;
requests.sort_by_key(|body| request_stage(body, &root_id));
let [_, _, starting, resumed, _, next @ ..] = requests.as_slice() else {
anyhow::bail!("missing model request")
};
let starting_tools = tool_names(starting);
assert!(starting_tools.contains(&"wait_for_environment".to_string()));
assert!(!starting_tools.contains(&"exec_command".to_string()));
let output = call_output(resumed, WAIT_ENV).context("child wait result")?;
if failed {
assert!(output.contains(FAILURE));
assert!(!tool_names(resumed).contains(&"exec_command".to_string()));
} else {
assert_eq!(
serde_json::from_str::<Value>(output)?,
json!({"environment_id": pending.environment_id, "status": "ready"})
);
assert!(tool_names(resumed).contains(&"exec_command".to_string()));
}
if matches!(case, PendingSpawnCase::FirstResultOnly) {
let exec = next.first().context("missing next child model request")?["tools"]
.as_array()
.context("child tools")?
.iter()
.find(|tool| tool["name"] == "exec_command")
.context("child execution tool")?;
assert!(exec["parameters"]["properties"].get("login").is_none());
}
Ok(requests)
}
#[test_case(PendingSpawnCase::V1Fork; "v1_fork")]
#[test_case(PendingSpawnCase::FirstResultOnly; "v2_fresh_ignores_later_owner_update_across_turns")]
#[test_case(PendingSpawnCase::Failure; "v2_owner_failure")]
#[test_case(PendingSpawnCase::ChildConfiguredFirst; "v2_child_configuration_wins_over_late_owner_failure")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn spawn_inherits_first_pending_owner_configuration(case: PendingSpawnCase) -> Result<()> {
pending_subagent_scenario(case, |_| {}).await?;
Ok(())
}
codex-rs/core/tests/suite/scenarios.rs
+56
View File
@@ -1553,6 +1553,62 @@ async fn guardian_code_mode_messaging_request_history() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn subagent_waits_for_its_inherited_environment_configuration() -> Result<()> {
skip_if_no_network!(Ok(()));
use super::remote_env::spawn_tests::PendingSpawnCase;
use super::remote_env::spawn_tests::pending_subagent_scenario;
let requests =
pending_subagent_scenario(PendingSpawnCase::TurnSettings, configure_scenario_catalog)
.await?;
let labels = [
"parent delegates while the environment is starting",
"parent waits for the worker",
"child sees the inherited starting environment",
"child receives the ready result and updated environment",
"parent receives the child result",
];
assert_eq!(requests.len(), labels.len());
let entries = requests
.iter()
.zip(labels)
.map(|(request, label)| SnapshotEntry::body(request).labeled(label))
.collect::<Vec<_>>();
let snapshot = context_snapshot::format_context_snapshot(
"A parent spawns a child before the selected environment is configured. The child sees it starting, waits, and receives the ready result after an ordinary parent turn-settings update.",
&entries,
&ContextSnapshotOptions::default(),
);
// Keep every request and its environment context; collapse unrelated guidance repeated per window.
let mut snapshot = snapshot;
for (pattern, replacement) in [
(
r"(?s)<permissions instructions>.*?</permissions instructions>",
"<PERMISSIONS_INSTRUCTIONS>",
),
(
r"(?s)<environments_instructions>.*?</environments_instructions>",
"<ENVIRONMENTS_INSTRUCTIONS>",
),
(
r"(?ms)(^\d+:message/developer:\n You are (?:`/root`, the primary agent|an agent) in a team of agents[^\n]+)\n.*?(\n^\d+:message/developer:)",
"${1}\n <STANDARD_MULTI_AGENT_INSTRUCTIONS>${2}",
),
(
r#"(environment_id"\s*:\s*"|<environment id=")(?:local|remote)"#,
"${1}<ENVIRONMENT>",
),
] {
snapshot = regex_lite::Regex::new(pattern)?
.replace_all(&snapshot, replacement)
.into_owned();
}
insta::assert_snapshot!("subagent_inherits_pending_environment", snapshot);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_checkpoint_migration_request_history() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/core/tests/suite/snapshots/all__suite__scenarios__subagent_inherits_pending_environment.snap
+128
View File
@@ -0,0 +1,128 @@
---
source: core/tests/suite/scenarios.rs
expression: snapshot
---
Scenario: A parent spawns a child before the selected environment is configured. The child sees it starting, waits, and receives the ready result after an ordinary parent turn-settings update.
## Window 1
-- request 1 (request; parent delegates while the environment is starting) --
00:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <ENVIRONMENTS_INSTRUCTIONS>
01:message/developer:
You are `/root`, the primary agent in a team of agents collaborating to fulfill the user's goals.
<STANDARD_MULTI_AGENT_INSTRUCTIONS>
02:message/developer:
<multi_agent_mode>Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work.</multi_agent_mode> [hash=2b6e037067b4390c]
03:message/user:
<environment_context>
<environments>
<environment id="<ENVIRONMENT>">
<cwd><CWD></cwd>
<status>starting</status>
</environment>
</environments>
<current_date><CURRENT_DATE></current_date>
<timezone><HOST_TIMEZONE></timezone>
</environment_context>
04:message/user:
Delegate waiting for the shared workspace, and report what the worker finds.
-- request 2 (request; parent waits for the worker) --
05:function_call/collaboration.spawn_agent:{"fork_turns":"none","message":"Wait for the shared workspace to become ready.","task_name":"worker"}
06:function_call_output:{"task_name":"/root/worker"}
## Window 2 (after request 2: input diverged at item 01, settings changed (tools))
-- request 3 (request; child sees the inherited starting environment) --
00:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <ENVIRONMENTS_INSTRUCTIONS>
01:message/developer:
You are an agent in a team of agents collaborating to complete a task.
<STANDARD_MULTI_AGENT_INSTRUCTIONS>
02:message/developer:
<multi_agent_mode>Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work.</multi_agent_mode> [hash=2b6e037067b4390c]
03:message/user:
<environment_context>
<environments>
<environment id="<ENVIRONMENT>">
<cwd><CWD></cwd>
<status>starting</status>
</environment>
</environments>
<current_date><CURRENT_DATE></current_date>
<timezone><HOST_TIMEZONE></timezone>
</environment_context>
04:agent_message
## Window 3 (after request 3: settings changed (tools))
-- request 4 (request; child receives the ready result and updated environment) --
00:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <ENVIRONMENTS_INSTRUCTIONS>
01:message/developer:
You are an agent in a team of agents collaborating to complete a task.
<STANDARD_MULTI_AGENT_INSTRUCTIONS>
02:message/developer:
<multi_agent_mode>Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work.</multi_agent_mode> [hash=2b6e037067b4390c]
03:message/user:
<environment_context>
<environments>
<environment id="<ENVIRONMENT>">
<cwd><CWD></cwd>
<status>starting</status>
</environment>
</environments>
<current_date><CURRENT_DATE></current_date>
<timezone><HOST_TIMEZONE></timezone>
</environment_context>
04:agent_message
05:function_call/functions.wait_for_environment:{"environment_id":"<ENVIRONMENT>"}
06:function_call_output:{"environment_id":"<ENVIRONMENT>","status":"ready"}
07:message/user:
<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>
## Window 4 (after request 4: input diverged at item 01, settings changed (tools))
-- request 5 (request; parent receives the child result) --
00:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <ENVIRONMENTS_INSTRUCTIONS>
01:message/developer:
You are `/root`, the primary agent in a team of agents collaborating to fulfill the user's goals.
<STANDARD_MULTI_AGENT_INSTRUCTIONS>
02:message/developer:
<multi_agent_mode>Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work.</multi_agent_mode> [hash=2b6e037067b4390c]
03:message/user:
<environment_context>
<environments>
<environment id="<ENVIRONMENT>">
<cwd><CWD></cwd>
<status>starting</status>
</environment>
</environments>
<current_date><CURRENT_DATE></current_date>
<timezone><HOST_TIMEZONE></timezone>
</environment_context>
04:message/user:
Delegate waiting for the shared workspace, and report what the worker finds.
05:function_call/collaboration.spawn_agent:{"fork_turns":"none","message":"Wait for the shared workspace to become ready.","task_name":"worker"}
06:function_call_output:{"task_name":"/root/worker"}
07:function_call/collaboration.wait_agent:{"timeout_ms":10000}
08:function_call_output:{"message":"Wait completed.","timed_out":false}
09:agent_message
10:message/user:
<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>
<agent name="/root/worker" />
</subagents>
</environment_context>