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

Restrict capability roots to captured turn environments (#49280)

Browse Source 
## Why

Stale thread capability roots can reconnect executors that are not selected for the current turn and shadow selected roots with the same ID.

## What changed

Filter out roots whose environments are absent from the turn snapshot before deduplication and readiness resolution. Update app-server capability tests to explicitly select the executor environments used by each turn, including after resume.

## Testing

Add regression coverage for full-access and workspace-write permissions, verifying that stale roots never connect an unselected executor, selected roots with matching IDs remain available, and turns with no environments expose no roots.

GitOrigin-RevId: afe375047cc5e8aad06df83334149510460dcdcc
This commit is contained in:
Michael Zeng
2026-09-29 15:10:17 +00:00
committed by copyberry
parent d515b2f85e
commit 18194bfd35
4 changed files with 171 additions and 401 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs
+33 -391
View File
@@ -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<Child> {
let mut child = Command::new(codex_utils_cargo_bin::cargo_bin("codex")?)
.args(["exec-server", "--listen", url])
codex-rs/core/src/session/mcp.rs
+12 -10
View File
@@ -419,12 +419,6 @@ impl Session {
ready_selected_capability_roots: &[SelectedCapabilityRoot],
environments: &TurnEnvironmentSnapshot,
) -> Option<Arc<ExecutorCapabilityDiscoverySnapshot>> {
// 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<ResolvedSelectedCapabilityRoot> {
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
}
codex-rs/core/tests/suite/remote_env.rs
+2
View File
@@ -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"]
codex-rs/core/tests/suite/remote_env_capability_roots_tests.rs
+124
View File
@@ -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(&"<ready_capability_roots>shared-root</ready_capability_roots>".to_string())
);
Ok(())
}