diff --git a/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs b/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs index 985684648a..13c983ab9f 100644 --- a/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs +++ b/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs @@ -6,7 +6,6 @@ use anyhow::Result; use app_test_support::ChatGptAuthFixture; use app_test_support::TestAppServer; use app_test_support::create_final_assistant_message_sse_response; -use app_test_support::create_request_user_input_sse_response; use app_test_support::to_response; use app_test_support::write_chatgpt_auth; use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url; @@ -14,11 +13,8 @@ use codex_app_server_protocol::AppInfo; use codex_app_server_protocol::CapabilityRootLocation; use codex_app_server_protocol::EnvironmentAddResponse; use codex_app_server_protocol::EnvironmentInfoResponse; -use codex_app_server_protocol::ListMcpServerStatusParams; -use codex_app_server_protocol::ListMcpServerStatusResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::SelectedCapabilityRoot; -use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; @@ -26,14 +22,9 @@ use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnEnvironmentParams; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; -use codex_app_server_protocol::TurnSteerParams; -use codex_app_server_protocol::TurnSteerResponse; use codex_app_server_protocol::UserInput; use codex_config::types::AuthCredentialsStoreMode; use codex_exec_server::LOCAL_ENVIRONMENT_ID; -use codex_protocol::config_types::CollaborationMode; -use codex_protocol::config_types::ModeKind; -use codex_protocol::config_types::Settings; use codex_protocol::protocol::PLUGINS_INSTRUCTIONS_OPEN_TAG; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_path_uri::PathUri; @@ -143,11 +134,18 @@ async fn selected_plugin_mcp_startup_respects_explicit_mentions( .send_turn_start_request(TurnStartParams { thread_id: thread_id.clone(), input: vec![input], - environments: Some(vec![TurnEnvironmentParams { - environment_id: LOCAL_ENVIRONMENT_ID.to_string(), - cwd: fixture.environment_cwd.into(), - runtime_workspace_roots: None, - }]), + environments: Some(vec![ + TurnEnvironmentParams { + environment_id: LOCAL_ENVIRONMENT_ID.to_string(), + cwd: fixture.environment_cwd.clone().into(), + runtime_workspace_roots: None, + }, + TurnEnvironmentParams { + environment_id: EXECUTOR_ID.to_string(), + cwd: fixture.environment_cwd.into(), + runtime_workspace_roots: None, + }, + ]), ..Default::default() }) .await?; @@ -270,6 +268,7 @@ async fn managed_plugins_requirement_disables_selected_executor_plugin_capabilit &thread_id, "Inspect the disabled selected plugin capabilities", fixture.environment_cwd, + &[LOCAL_ENVIRONMENT_ID, EXECUTOR_ID], ) .await?; @@ -286,7 +285,7 @@ async fn managed_plugins_requirement_disables_selected_executor_plugin_capabilit } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn selected_capability_stack_tracks_environment_availability_and_resume() -> Result<()> { +async fn selected_capability_stack_tracks_environment_selection_and_resume() -> Result<()> { let responses_server = responses::start_mock_server().await; let (apps_url, apps_server_handle) = start_apps_server_with_delays( vec![AppInfo { @@ -381,6 +380,7 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() &thread_id, "Inspect the current capabilities", fixture.environment_cwd.clone(), + &[LOCAL_ENVIRONMENT_ID], ) .await?; let initial_requests = response_mock.requests(); @@ -389,14 +389,14 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() let mut exec_server = spawn_exec_server(fixture.codex_home.path(), &fixture.exec_server_url).await?; add_environment(&mut app_server, &fixture.exec_server_url).await?; - wait_for_selected_mcp_server(&mut app_server, &thread_id).await?; - // A skill mention alone does not wait for MCP startup. + // The next turn selects the executor; its server mention waits for MCP startup. run_turn( &mut app_server, &thread_id, &format!("Use ${SKILL_NAME} and call [${MCP_SERVER_NAME}](mcp://{MCP_SERVER_NAME})"), fixture.environment_cwd.clone(), + &[LOCAL_ENVIRONMENT_ID, EXECUTOR_ID], ) .await?; let first_mcp_pid = wait_for_pid_file(&fixture.pid_file).await?; @@ -406,6 +406,7 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() &thread_id, "Continue with the same selected capabilities", fixture.environment_cwd.clone(), + &[LOCAL_ENVIRONMENT_ID, EXECUTOR_ID], ) .await?; assert_eq!(first_mcp_pid, wait_for_pid_file(&fixture.pid_file).await?); @@ -440,6 +441,7 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() &thread_id, "Inspect capabilities while the selected executor is unavailable", fixture.environment_cwd.clone(), + &[LOCAL_ENVIRONMENT_ID], ) .await?; let requests = response_mock.requests(); @@ -452,7 +454,6 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() exec_server = spawn_exec_server(fixture.codex_home.path(), &fixture.exec_server_url).await?; add_environment(&mut app_server, &fixture.exec_server_url).await?; - wait_for_selected_mcp_server(&mut app_server, &thread_id).await?; run_turn( &mut app_server, @@ -461,6 +462,7 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() "Use ${SKILL_NAME} with [${MCP_SERVER_NAME}](mcp://{MCP_SERVER_NAME}) after reattaching the selected executor" ), fixture.environment_cwd, + &[LOCAL_ENVIRONMENT_ID, EXECUTOR_ID], ) .await?; let resumed_mcp_pid = wait_for_pid_file(&fixture.pid_file).await?; @@ -502,340 +504,6 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() Ok(()) } -#[derive(Clone, Copy)] -enum MentionTiming { - Unmentioned, - Initial, - InitialBeforeRestart, - Steered, -} - -#[derive(Clone, Copy)] -enum MentionTarget { - Plugin, - Server, -} - -#[test_case(MentionTiming::Unmentioned, MentionTarget::Plugin; "without a mention")] -#[test_case(MentionTiming::Initial, MentionTarget::Plugin; "with an initial plugin mention")] -#[test_case(MentionTiming::InitialBeforeRestart, MentionTarget::Plugin; "with an initial plugin mention across a restart")] -#[test_case(MentionTiming::Steered, MentionTarget::Plugin; "with a steered plugin mention")] -#[test_case(MentionTiming::Initial, MentionTarget::Server; "with an initial server mention")] -#[test_case(MentionTiming::Steered, MentionTarget::Server; "with a steered server mention")] -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn selected_capabilities_become_available_between_samples_in_one_turn( - mention_timing: MentionTiming, - mention_target: MentionTarget, -) -> Result<()> { - const USER_INPUT_CALL_ID: &str = "pause-for-environment"; - const MCP_USER_INPUT_CALL_ID: &str = "pause-for-mcp"; - - let mention_link = match mention_target { - MentionTarget::Plugin => format!("[@executor-demo](plugin://{PLUGIN_ID})"), - MentionTarget::Server => format!("[${MCP_SERVER_NAME}](mcp://{MCP_SERVER_NAME})"), - }; - let samples_before_attach = if matches!(mention_timing, MentionTiming::Steered) { - 2 - } else { - 1 - }; - let continue_answers = json!({ "answers": { "confirm_path": { "answers": ["yes"] } } }); - let responses_server = responses::start_mock_server().await; - let (apps_url, apps_server_handle) = start_apps_server_with_delays( - vec![AppInfo { - id: CONNECTOR_ID.to_string(), - name: "Calendar".to_string(), - description: None, - logo_url: None, - logo_url_dark: None, - icon_assets: None, - icon_dark_assets: None, - distribution_channel: None, - branding: None, - app_metadata: None, - labels: None, - install_url: None, - is_accessible: false, - is_enabled: true, - plugin_display_names: Vec::new(), - }], - vec![connector_tool(CONNECTOR_ID, "Calendar")?], - Duration::ZERO, - Duration::ZERO, - ) - .await?; - let fixture = selected_capability_fixture(&responses_server.uri(), &apps_url)?; - let initialize_barrier = fixture.block_mcp_startup()?; - let stop_hook_barrier = if matches!(mention_timing, MentionTiming::InitialBeforeRestart) { - let barrier = fixture.codex_home.path().join("allow-stop-hook"); - let hook_path = fixture.codex_home.path().join("wait-for-stop.py"); - std::fs::write( - &hook_path, - "import json\nimport sys\nimport time\nfrom pathlib import Path\n\n\ - json.load(sys.stdin)\nwhile not Path(sys.argv[1]).exists():\n time.sleep(0.01)\n", - )?; - let command = toml::Value::String(format!( - "python3 \"{}\" \"{}\"", - hook_path.display(), - barrier.display() - )); - std::fs::write( - fixture.codex_home.path().join("requirements.toml"), - format!( - "[[hooks.Stop]]\n[[hooks.Stop.hooks]]\ntype = 'command'\ncommand = {command}\ntimeout = 60\n" - ), - )?; - let config_path = fixture.codex_home.path().join("config.toml"); - let config = std::fs::read_to_string(&config_path)? - .replace("[features]\n", "[features]\nhooks = true\n"); - std::fs::write(config_path, config)?; - Some(barrier) - } else { - None - }; - let response_mock = responses::mount_sse_sequence( - &responses_server, - vec![ - if stop_hook_barrier.is_some() { - create_final_assistant_message_sse_response("Waiting for the executor")? - } else { - create_request_user_input_sse_response(USER_INPUT_CALL_ID)? - }, - create_request_user_input_sse_response(MCP_USER_INPUT_CALL_ID)?, - responses::sse(vec![ - responses::ev_response_created("environment-ready-call"), - responses::ev_function_call_with_namespace( - MCP_CALL_ID, - &format!("mcp__{MCP_SERVER_NAME}"), - "echo", - &json!({ - "message": "same turn", - "env_var": EXECUTOR_ENV_NAME, - }) - .to_string(), - ), - responses::ev_completed("environment-ready-call"), - ]), - responses::sse(vec![ - responses::ev_response_created("same-turn-done"), - responses::ev_assistant_message("same-turn-message", "Done"), - responses::ev_completed("same-turn-done"), - ]), - ], - ) - .await; - - let mut app_server = TestAppServer::builder() - .with_codex_home(fixture.codex_home.path()) - // This fixture owns environments.toml and selects its environments explicitly. - .without_auto_env() - .build() - .await?; - timeout(READ_TIMEOUT, app_server.initialize()).await??; - let thread_id = start_thread( - &mut app_server, - fixture.selected_root, - fixture.environment_cwd.clone(), - ) - .await?; - let turn_start_id = app_server - .send_turn_start_request(TurnStartParams { - thread_id: thread_id.clone(), - input: vec![UserInput::Text { - text: match mention_timing { - MentionTiming::Unmentioned | MentionTiming::Steered => { - "Use the executor when it becomes ready.".to_string() - } - MentionTiming::Initial | MentionTiming::InitialBeforeRestart => { - format!("Use {mention_link} when its executor becomes ready.") - } - }, - text_elements: Vec::new(), - }], - environments: Some(vec![TurnEnvironmentParams { - environment_id: LOCAL_ENVIRONMENT_ID.to_string(), - cwd: fixture.environment_cwd.into(), - runtime_workspace_roots: None, - }]), - collaboration_mode: Some(CollaborationMode { - mode: ModeKind::Plan, - settings: Settings { - model: "mock-model".to_string(), - reasoning_effort: None, - developer_instructions: None, - }, - }), - ..Default::default() - }) - .await?; - let response = timeout( - READ_TIMEOUT, - app_server.read_stream_until_response_message(RequestId::Integer(turn_start_id)), - ) - .await??; - let TurnStartResponse { turn } = to_response(response)?; - - let request_id = if stop_hook_barrier.is_some() { - // Steering during a successful Stop hook restarts run_turn in the same RegularTask. - timeout( - READ_TIMEOUT, - app_server.read_stream_until_matching_notification( - "Stop hook started", - |notification| { - notification.method == "hook/started" - && notification.params.as_ref().is_some_and(|params| { - params["turnId"] == turn.id && params["run"]["eventName"] == "stop" - }) - }, - ), - ) - .await??; - None - } else { - let request = - timeout(READ_TIMEOUT, app_server.read_stream_until_request_message()).await??; - let ServerRequest::ToolRequestUserInput { request_id, .. } = request else { - panic!("expected request_user_input, got {request:?}"); - }; - Some(request_id) - }; - let requests = response_mock.requests(); - assert_eq!(1, requests.len()); - assert_selected_capabilities_absent(&requests[0]); - - if !matches!(mention_timing, MentionTiming::Unmentioned) { - let steer_request_id = app_server - .send_turn_steer_request(TurnSteerParams { - thread_id, - input: vec![UserInput::Text { - text: if matches!(mention_timing, MentionTiming::Steered) { - format!("Use {mention_link} now.") - } else { - "Continue with the requested tool when it becomes available.".to_string() - }, - text_elements: Vec::new(), - }], - expected_turn_id: turn.id.clone(), - ..Default::default() - }) - .await?; - let response = timeout( - READ_TIMEOUT, - app_server.read_stream_until_response_message(RequestId::Integer(steer_request_id)), - ) - .await??; - let response: TurnSteerResponse = to_response(response)?; - assert_eq!(response, TurnSteerResponse { turn_id: turn.id }); - } - let request_id = if matches!(mention_timing, MentionTiming::Steered) { - // Consume the mention while the executor is absent, then sample once more. - app_server - .send_response( - request_id.expect("steering pauses before attaching the executor"), - continue_answers.clone(), - ) - .await?; - let request = - timeout(READ_TIMEOUT, app_server.read_stream_until_request_message()).await??; - let ServerRequest::ToolRequestUserInput { request_id, .. } = request else { - panic!("expected request_user_input, got {request:?}"); - }; - let requests = response_mock.requests(); - assert_eq!(2, requests.len()); - assert_selected_capabilities_absent(&requests[1]); - Some(request_id) - } else { - request_id - }; - let mut exec_server = - spawn_exec_server(fixture.codex_home.path(), &fixture.exec_server_url).await?; - add_environment(&mut app_server, &fixture.exec_server_url).await?; - if let Some(request_id) = request_id { - app_server - .send_response(request_id, continue_answers.clone()) - .await?; - } - if let Some(stop_hook_barrier) = stop_hook_barrier { - std::fs::write(stop_hook_barrier, "ready")?; - } - let mcp_pid = wait_for_pid_file(&fixture.pid_file).await?; - if !matches!(mention_timing, MentionTiming::Unmentioned) { - // An explicit mention must still require startup after the executor attaches. - tokio::time::sleep(Duration::from_secs(3)).await; - assert_eq!( - response_mock.requests().len(), - samples_before_attach, - "the explicit mention must keep later samples waiting for MCP startup" - ); - std::fs::write(&initialize_barrier, "ready")?; - } - let request_id = if matches!(mention_timing, MentionTiming::Steered) { - None - } else { - let request = - timeout(READ_TIMEOUT, app_server.read_stream_until_request_message()).await??; - let ServerRequest::ToolRequestUserInput { request_id, .. } = request else { - panic!("expected request_user_input, got {request:?}"); - }; - let requests = response_mock.requests(); - assert_eq!(2, requests.len()); - assert_selected_skill_catalog_available(&requests[1]); - if matches!(mention_timing, MentionTiming::Unmentioned) { - assert!( - requests[1] - .tool_by_name(&format!("mcp__{MCP_SERVER_NAME}"), "echo") - .is_none() - ); - std::fs::write(&initialize_barrier, "ready")?; - } else { - assert_selected_plugin_tools(&requests[1]); - } - Some(request_id) - }; - timeout( - READ_TIMEOUT, - app_server.read_stream_until_matching_notification("selected MCP ready", |notification| { - notification.method == "mcpServer/startupStatus/updated" - && notification.params.as_ref().is_some_and(|params| { - params["name"] == MCP_SERVER_NAME && params["status"] == "ready" - }) - }), - ) - .await??; - if let Some(request_id) = request_id { - app_server - .send_response(request_id, continue_answers) - .await?; - } - timeout( - READ_TIMEOUT, - app_server.read_stream_until_notification_message("turn/completed"), - ) - .await??; - - let requests = response_mock.requests(); - assert_eq!(4, requests.len()); - for request in &requests[samples_before_attach..] { - assert_selected_skill_catalog_available(request); - assert_plugin_guidance_count(request, /*expected_count*/ 0); - } - for request in &requests[2..] { - assert_selected_plugin_tools(request); - } - let output = requests[3].function_call_output(MCP_CALL_ID); - let output = output["output"] - .as_str() - .expect("MCP function output should be text"); - assert!(output.contains("ECHOING: same turn")); - assert!(output.contains(EXECUTOR_ENV_VALUE)); - assert_eq!(mcp_pid, wait_for_pid_file(&fixture.pid_file).await?); - - exec_server.kill().await?; - apps_server_handle.abort(); - let _ = apps_server_handle.await; - Ok(()) -} - struct SelectedCapabilityFixture { codex_home: TempDir, _plugin: TempDir, @@ -875,10 +543,11 @@ fn selected_capability_fixture( "mcp_oauth_credentials_store = \"file\"\nmodel_provider = \"mock_provider\"", 1, ); + // These scenarios attach executors between turns, so each turn waits for attachment resolution. std::fs::write( config_path, format!( - "{config}\n[features]\napps = true\ndeferred_executor = true\nexecutor_capability_discovery = true\n\n[skills]\ninclude_instructions = true\n" + "{config}\n[features]\napps = true\nexecutor_capability_discovery = true\n\n[skills]\ninclude_instructions = true\n" ), )?; write_chatgpt_auth( @@ -1090,6 +759,7 @@ async fn run_turn( thread_id: &str, text: &str, environment_cwd: AbsolutePathBuf, + environment_ids: &[&str], ) -> Result<()> { let request_id = app_server .send_turn_start_request(TurnStartParams { @@ -1098,11 +768,16 @@ async fn run_turn( text: text.to_string(), text_elements: Vec::new(), }], - environments: Some(vec![TurnEnvironmentParams { - environment_id: LOCAL_ENVIRONMENT_ID.to_string(), - cwd: environment_cwd.into(), - runtime_workspace_roots: None, - }]), + environments: Some( + environment_ids + .iter() + .map(|environment_id| TurnEnvironmentParams { + environment_id: (*environment_id).to_string(), + cwd: environment_cwd.clone().into(), + runtime_workspace_roots: None, + }) + .collect(), + ), ..Default::default() }) .await?; @@ -1152,39 +827,6 @@ async fn add_environment(app_server: &mut TestAppServer, exec_server_url: &str) Ok(()) } -async fn wait_for_selected_mcp_server( - app_server: &mut TestAppServer, - thread_id: &str, -) -> Result<()> { - timeout(READ_TIMEOUT, async { - loop { - let request_id = app_server - .send_list_mcp_server_status_request(ListMcpServerStatusParams { - server_name: None, - cursor: None, - limit: None, - detail: None, - thread_id: Some(thread_id.to_string()), - }) - .await?; - let response = app_server - .read_stream_until_response_message(RequestId::Integer(request_id)) - .await?; - let response: ListMcpServerStatusResponse = to_response(response)?; - if response - .data - .iter() - .any(|server| server.name == MCP_SERVER_NAME) - { - return Ok::<_, anyhow::Error>(()); - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - }) - .await??; - Ok(()) -} - async fn spawn_exec_server(codex_home: &std::path::Path, url: &str) -> Result { let mut child = Command::new(codex_utils_cargo_bin::cargo_bin("codex")?) .args(["exec-server", "--listen", url]) diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index 765fe5eda6..889ae89a2e 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -419,12 +419,6 @@ impl Session { ready_selected_capability_roots: &[SelectedCapabilityRoot], environments: &TurnEnvironmentSnapshot, ) -> Option> { - // Capability roots can currently be selected independently of turn environments, so a - // root may be ready when there is no primary `TurnEnvironment`. Keep using the thread - // policy in that case so restricted discovery fails closed below. Once every selected - // root belongs to a thread/environment attachment whose `EnvironmentConfig` is installed - // before the root becomes ready, discovery can use the root owner's policy and this - // fallback can be removed. let restricted_file_system = environments.primary().map_or_else( || { !config @@ -499,6 +493,7 @@ impl Session { &self, environments: &TurnEnvironmentSnapshot, ) -> Vec { + let captured_environments = environments.captured_environments(); let thread_root_count = self.services.selected_capability_roots.len(); let mut root_locations_by_id = HashMap::new(); let mut selected_capability_roots = Vec::new(); @@ -515,6 +510,16 @@ impl Session { }), ); for (index, root) in combined_roots.into_iter().enumerate() { + let CapabilityRootLocation::Environment { environment_id, .. } = &root.location; + // Filter before readiness resolution, which can reconnect a historical executor. + if !captured_environments.contains_key(environment_id) { + tracing::warn!( + root_id = root.id, + environment_id, + "ignoring capability root without a captured turn environment" + ); + continue; + } if let Some(kept_location) = root_locations_by_id.get(&root.id) { if kept_location != &root.location { tracing::warn!( @@ -542,10 +547,7 @@ impl Session { self.services .turn_environments .environment_manager() - .resolve_selected_capability_roots( - &selected_capability_roots, - &environments.captured_environments(), - ) + .resolve_selected_capability_roots(&selected_capability_roots, &captured_environments) .await } diff --git a/codex-rs/core/tests/suite/remote_env.rs b/codex-rs/core/tests/suite/remote_env.rs index e3cfd43865..4b8c919240 100644 --- a/codex-rs/core/tests/suite/remote_env.rs +++ b/codex-rs/core/tests/suite/remote_env.rs @@ -1,3 +1,5 @@ +#[path = "remote_env_capability_roots_tests.rs"] +mod capability_roots; #[path = "guardian_environments_tests.rs"] mod guardian_environments; #[path = "remote_env_spawn_tests.rs"] diff --git a/codex-rs/core/tests/suite/remote_env_capability_roots_tests.rs b/codex-rs/core/tests/suite/remote_env_capability_roots_tests.rs new file mode 100644 index 0000000000..17627b6c59 --- /dev/null +++ b/codex-rs/core/tests/suite/remote_env_capability_roots_tests.rs @@ -0,0 +1,124 @@ +//! Capability discovery stays within the environments captured by the current turn. + +use super::*; +use pretty_assertions::assert_eq; +use test_case::test_case; + +#[test_case(PermissionProfile::Disabled; "full access")] +#[test_case(PermissionProfile::workspace_write(); "workspace write")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn stale_roots_do_not_connect_unselected_executors( + permission_profile: PermissionProfile, +) -> Result<()> { + let server = start_mock_server().await; + let observed_roots = Arc::new(Mutex::new(Vec::new())); + let mut extensions = ExtensionRegistryBuilder::new(); + extensions.prompt_contributor(Arc::new(ReadyCapabilityRootsTestExtension { + observed_roots: Some(Arc::clone(&observed_roots)), + })); + let test = test_codex() + .with_extensions(Arc::new(extensions.build())) + .with_config(move |config| { + config + .permissions + .set_permission_profile(permission_profile) + .expect("thread permissions should be configurable"); + config + .features + .enable(Feature::ExecutorCapabilityDiscovery) + .expect("capability discovery should be configurable"); + }) + .build_with_auto_env(&server) + .await?; + let mut selection = test + .codex + .environment_selections() + .await + .into_iter() + .next() + .context("thread should select its executor environment")?; + let selected_root = SelectedCapabilityRoot { + id: "shared-root".to_string(), + location: CapabilityRootLocation::Environment { + environment_id: selection.environment_id.clone(), + path: selection.cwd.clone(), + }, + }; + let stale_root = SelectedCapabilityRoot { + id: selected_root.id.clone(), + location: CapabilityRootLocation::Environment { + environment_id: "previous-executor".to_string(), + path: selection.cwd.clone(), + }, + }; + let provider = Arc::new(FailingNoiseConnectProvider::default()); + let previous_executor = test + .thread_manager + .environment_manager() + .report_environment_provisioning_status( + "previous-executor".to_string(), + Ok(EnvironmentReadyInfo { + selected_capability_roots: vec![stale_root.clone()], + }), + provider.clone(), + )? + .context("previous executor should remain registered")?; + let mut environment_config = environment_config_for_selection(&test.config, &selection); + environment_config.selected_capability_roots = vec![selected_root.clone()]; + selection.config = EnvironmentConfigState::Ready(environment_config); + let mut thread_extension_init = ExtensionDataInit::new(); + thread_extension_init.insert(vec![stale_root]); + let thread = test + .thread_manager + .start_thread(StartThreadOptions { + environments: Some(vec![selection.clone()]), + thread_extension_init, + ..StartThreadOptions::new(test.config.clone()) + }) + .await? + .thread; + let response_mock = mount_sse_sequence( + &server, + vec![ + sse(vec![ev_completed("selected-environment")]), + sse(vec![ev_completed("no-environments")]), + ], + ) + .await; + for (environments, expected_roots) in [ + (vec![selection], vec![selected_root]), + (Vec::new(), Vec::new()), + ] { + thread + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "continue in the selected environments".to_string(), + text_elements: Vec::new(), + }]) + .with_thread_settings(ThreadSettingsOverrides { + environments: Some(TurnEnvironmentSelections::new( + test.config.cwd.clone(), + environments, + )), + ..Default::default() + }), + ) + .await?; + wait_for_event(&thread, |event| matches!(event, EventMsg::TurnComplete(_))).await; + assert_eq!( + observed_roots.lock().expect("observed roots").last(), + Some(&expected_roots) + ); + } + + assert_eq!(provider.calls.load(Ordering::Relaxed), 0); + assert!(!previous_executor.startup_finished()); + let requests = response_mock.requests(); + assert_eq!(requests.len(), 2); + assert!( + requests[0] + .message_input_texts("user") + .contains(&"shared-root".to_string()) + ); + Ok(()) +}