From 3749d1eff7df4ac9a0e2d32099813ba56500b930 Mon Sep 17 00:00:00 2001 From: sayan-oai Date: Mon, 28 Sep 2026 20:21:39 +0000 Subject: [PATCH] Preserve pending environments when spawning subagents (#49075) ## 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 --- codex-rs/core/src/agent/control/spawn.rs | 18 +- codex-rs/core/src/agent/control_tests.rs | 69 ++++ codex-rs/core/src/agent/types.rs | 5 +- codex-rs/core/src/environment_selection.rs | 220 +++++++++---- .../environment_selection_snapshot_tests.rs | 67 ++++ codex-rs/core/src/session/environment.rs | 79 ++++- codex-rs/core/src/session/session.rs | 7 +- codex-rs/core/src/thread_manager.rs | 31 +- .../src/tools/handlers/multi_agents/spawn.rs | 2 +- .../tools/handlers/multi_agents_v2/spawn.rs | 2 +- codex-rs/core/tests/suite/remote_env.rs | 2 + .../tests/suite/remote_env_spawn_tests.rs | 306 ++++++++++++++++++ codex-rs/core/tests/suite/scenarios.rs | 56 ++++ ...subagent_inherits_pending_environment.snap | 128 ++++++++ 14 files changed, 899 insertions(+), 93 deletions(-) create mode 100644 codex-rs/core/tests/suite/remote_env_spawn_tests.rs create mode 100644 codex-rs/core/tests/suite/snapshots/all__suite__scenarios__subagent_inherits_pending_environment.snap diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 9b642981a3..277ce112bd 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -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?; diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index ab18abebb8..dbd89a5ccf 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -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; diff --git a/codex-rs/core/src/agent/types.rs b/codex-rs/core/src/agent/types.rs index 81dcd5354e..8d612d6beb 100644 --- a/codex-rs/core/src/agent/types.rs +++ b/codex-rs/core/src/agent/types.rs @@ -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, pub root_turn_id: Option, - pub environments: Option>, + /// Exact environment bindings from the step that requested the child. + pub environments: Option, pub multi_agent_v2_usage_hints: Option, pub cyber_access_program: Option, } diff --git a/codex-rs/core/src/environment_selection.rs b/codex-rs/core/src/environment_selection.rs index eaa699adda..a1b5098651 100644 --- a/codex-rs/core/src/environment_selection.rs +++ b/codex-rs/core/src/environment_selection.rs @@ -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>; type TurnEnvironmentResolution = Shared>; type PendingConfigurationResult = Result; +pub(crate) type PendingConfiguration = Shared>; // 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>>, resolution: TurnEnvironmentResolution, - pending_completion: Option>, + // Executor retries keep this session's wait for its first accepted configuration. + pending_completion: Option>>, + // Descendants can await the original attachment's first accepted config or failure. + owner_config_result: Option, } #[derive(Clone)] @@ -204,6 +208,8 @@ pub(crate) struct StartingTurnEnvironment { pub(crate) selection: TurnEnvironmentSelection, config_origin: EnvironmentConfigOrigin, resolution: TurnEnvironmentResolution, + environment: Arc, + owner_config_result: Option, } /// 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 { + 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::>(); 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, + owner_config_result: Option, + pending_completion: Option>>, + 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 { self.state @@ -738,7 +816,7 @@ impl ThreadEnvironments { environment: Arc, local_shell: Shell, shell_snapshot: ShellSnapshot, - configuration_ready: Option>, + configuration_ready: Option, ) -> 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 { + 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 { 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, })], }; diff --git a/codex-rs/core/src/environment_selection_snapshot_tests.rs b/codex-rs/core/src/environment_selection_snapshot_tests.rs index 8302c52923..6e95c6ee5d 100644 --- a/codex-rs/core/src/environment_selection_snapshot_tests.rs +++ b/codex-rs/core/src/environment_selection_snapshot_tests.rs @@ -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"); diff --git a/codex-rs/core/src/session/environment.rs b/codex-rs/core/src/session/environment.rs index 28f339c12e..494a621890 100644 --- a/codex-rs/core/src/session/environment.rs +++ b/codex-rs/core/src/session/environment.rs @@ -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, + 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, + 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 diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 62be2d34e2..77085befc4 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -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 { diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 6a0c7d2cb0..ba195d8e0c 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -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( diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index 9bfc5571c4..5a7bd67b87 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -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, }, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index ff00a422e6..4562cc4ebc 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs @@ -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, }, diff --git a/codex-rs/core/tests/suite/remote_env.rs b/codex-rs/core/tests/suite/remote_env.rs index c80b6696f1..e3cfd43865 100644 --- a/codex-rs/core/tests/suite/remote_env.rs +++ b/codex-rs/core/tests/suite/remote_env.rs @@ -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; diff --git a/codex-rs/core/tests/suite/remote_env_spawn_tests.rs b/codex-rs/core/tests/suite/remote_env_spawn_tests.rs new file mode 100644 index 0000000000..eee1da7395 --- /dev/null +++ b/codex-rs/core/tests/suite/remote_env_spawn_tests.rs @@ -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> { + 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::>>()?; + 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::(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(()) +} diff --git a/codex-rs/core/tests/suite/scenarios.rs b/codex-rs/core/tests/suite/scenarios.rs index a342d9e90f..475140e94c 100644 --- a/codex-rs/core/tests/suite/scenarios.rs +++ b/codex-rs/core/tests/suite/scenarios.rs @@ -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::>(); + 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).*?", + "", + ), + ( + r"(?s).*?", + "", + ), + ( + 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 ${2}", + ), + ( + r#"(environment_id"\s*:\s*"|", + ), + ] { + 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(())); diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__subagent_inherits_pending_environment.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__subagent_inherits_pending_environment.snap new file mode 100644 index 0000000000..9ddf428957 --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__subagent_inherits_pending_environment.snap @@ -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] + [02] +01:message/developer: + You are `/root`, the primary agent in a team of agents collaborating to fulfill the user's goals. + +02:message/developer: + Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work. [hash=2b6e037067b4390c] +03:message/user: + + + + + starting + + + + + +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] + [02] +01:message/developer: + You are an agent in a team of agents collaborating to complete a task. + +02:message/developer: + Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work. [hash=2b6e037067b4390c] +03:message/user: + + + + + starting + + + + + +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] + [02] +01:message/developer: + You are an agent in a team of agents collaborating to complete a task. + +02:message/developer: + Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work. [hash=2b6e037067b4390c] +03:message/user: + + + + + starting + + + + + +04:agent_message +05:function_call/functions.wait_for_environment:{"environment_id":""} +06:function_call_output:{"environment_id":"","status":"ready"} +07:message/user: + + + + + + [hash=e8c1de60307d0a3d] + + +## 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] + [02] +01:message/developer: + You are `/root`, the primary agent in a team of agents collaborating to fulfill the user's goals. + +02:message/developer: + Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work. [hash=2b6e037067b4390c] +03:message/user: + + + + + starting + + + + + +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: + + + + + + [hash=e8c1de60307d0a3d] + + + +