mirror of
https://github.com/openai/codex.git
synced 2026-10-20 00:18:49 +00:00
Separate environment requests from runtime selections (#51221)
## What changed Introduce `TurnEnvironmentRequest` for caller-supplied environment input and construct `TurnEnvironmentSelection` at session startup and app-server settings boundaries. Update thread startup, default environments, and inherited environment handoffs to use requests, with explicit conversions when reusing runtime selections. Move the environment types into `protocol::environment` while retaining their `protocol` re-exports. Update test fixtures and existing tests to build request inputs separately from runtime selections. GitOrigin-RevId: 12e7daaece95aae136afae8b055b4dbaebdffc75
This commit is contained in:
committed by
copyberry
parent
28b91c7c31
commit
2dbcab90e2
@@ -461,6 +461,7 @@ use codex_protocol::protocol::ReviewTarget as CoreReviewTarget;
|
||||
use codex_protocol::protocol::SessionConfiguredEvent;
|
||||
#[cfg(test)]
|
||||
use codex_protocol::protocol::SessionMetaLine;
|
||||
use codex_protocol::protocol::TurnEnvironmentRequest;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelections;
|
||||
use codex_protocol::protocol::W3cTraceContext;
|
||||
@@ -637,14 +638,14 @@ fn resolve_request_cwd(cwd: Option<PathBuf>) -> Result<Option<AbsolutePathBuf>,
|
||||
.transpose()
|
||||
}
|
||||
|
||||
fn resolve_turn_environment_selections(
|
||||
fn resolve_turn_environment_requests(
|
||||
thread_manager: &ThreadManager,
|
||||
environments: Option<Vec<TurnEnvironmentParams>>,
|
||||
) -> Result<Option<Vec<TurnEnvironmentSelection>>, JSONRPCErrorError> {
|
||||
) -> Result<Option<Vec<TurnEnvironmentRequest>>, JSONRPCErrorError> {
|
||||
let Some(environments) = environments else {
|
||||
return Ok(None);
|
||||
};
|
||||
let mut selections = Vec::with_capacity(environments.len());
|
||||
let mut requests = Vec::with_capacity(environments.len());
|
||||
for environment in environments {
|
||||
let environment_id = environment.environment_id;
|
||||
let cwd = environment
|
||||
@@ -674,16 +675,21 @@ fn resolve_turn_environment_selections(
|
||||
})
|
||||
.transpose()?
|
||||
.unwrap_or_else(|| vec![cwd.clone()]);
|
||||
selections.push(TurnEnvironmentSelection {
|
||||
requests.push(TurnEnvironmentRequest {
|
||||
environment_id,
|
||||
cwd,
|
||||
workspace_roots,
|
||||
config: EnvironmentConfigState::FromThread,
|
||||
});
|
||||
}
|
||||
validate_environment_ids_and_cwds(&thread_manager.environment_manager(), &selections)
|
||||
.map_err(environment_selection_error)?;
|
||||
Ok(Some(selections))
|
||||
validate_environment_ids_and_cwds(
|
||||
&thread_manager.environment_manager(),
|
||||
requests
|
||||
.iter()
|
||||
.map(|request| (request.environment_id.as_str(), &request.cwd)),
|
||||
)
|
||||
.map_err(environment_selection_error)?;
|
||||
Ok(Some(requests))
|
||||
}
|
||||
|
||||
fn resolve_runtime_workspace_roots(workspace_roots: Vec<AbsolutePathBuf>) -> Vec<AbsolutePathBuf> {
|
||||
|
||||
@@ -1195,8 +1195,8 @@ impl ThreadRequestProcessor {
|
||||
}
|
||||
}
|
||||
let runtime_workspace_roots = runtime_workspace_roots.map(resolve_runtime_workspace_roots);
|
||||
let environments =
|
||||
resolve_turn_environment_selections(self.thread_manager.as_ref(), environments)?;
|
||||
let environment_requests =
|
||||
resolve_turn_environment_requests(self.thread_manager.as_ref(), environments)?;
|
||||
let mut typesafe_overrides = self.build_thread_config_overrides(
|
||||
model,
|
||||
model_provider,
|
||||
@@ -1247,7 +1247,7 @@ impl ThreadRequestProcessor {
|
||||
thread_source.map(Into::into),
|
||||
project_id,
|
||||
daybreak_enabled,
|
||||
environments,
|
||||
environment_requests,
|
||||
service_name,
|
||||
allow_provider_model_fallback,
|
||||
experimental_raw_events,
|
||||
@@ -1329,7 +1329,7 @@ impl ThreadRequestProcessor {
|
||||
thread_source: Option<codex_protocol::protocol::ThreadSource>,
|
||||
project_id: Option<String>,
|
||||
daybreak_enabled: Option<bool>,
|
||||
environment_selections: Option<Vec<TurnEnvironmentSelection>>,
|
||||
environment_requests: Option<Vec<TurnEnvironmentRequest>>,
|
||||
service_name: Option<String>,
|
||||
allow_provider_model_fallback: bool,
|
||||
experimental_raw_events: bool,
|
||||
@@ -1442,10 +1442,10 @@ impl ThreadRequestProcessor {
|
||||
}
|
||||
}
|
||||
|
||||
let environments = environment_selections.unwrap_or_else(|| {
|
||||
let environment_requests = environment_requests.unwrap_or_else(|| {
|
||||
listener_task_context
|
||||
.thread_manager
|
||||
.default_environment_selections(&config.cwd, &config.workspace_roots)
|
||||
.default_environment_requests(&config.cwd, &config.workspace_roots)
|
||||
});
|
||||
let dynamic_tools = dynamic_tools.unwrap_or_default();
|
||||
if !dynamic_tools.is_empty() {
|
||||
@@ -1508,7 +1508,7 @@ impl ThreadRequestProcessor {
|
||||
dynamic_tools,
|
||||
metrics_service_name: service_name,
|
||||
parent_trace: request_trace,
|
||||
environments: Some(environments),
|
||||
environments: Some(environment_requests),
|
||||
thread_extension_init,
|
||||
client_mcp_extensions,
|
||||
..start_options
|
||||
|
||||
@@ -118,7 +118,7 @@ fn map_additional_context(
|
||||
|
||||
#[derive(Default)]
|
||||
struct ThreadEnvironmentOverride {
|
||||
environments: Option<TurnEnvironmentSelections>,
|
||||
environment_requests: Option<TurnEnvironmentSelections>,
|
||||
// Only default-environment updates replace the task's separately persisted root selection.
|
||||
runtime_workspace_roots: Option<Vec<AbsolutePathBuf>>,
|
||||
}
|
||||
@@ -589,8 +589,8 @@ impl TurnRequestProcessor {
|
||||
let runtime_workspace_roots = params
|
||||
.runtime_workspace_roots
|
||||
.map(resolve_runtime_workspace_roots);
|
||||
let environment_selections =
|
||||
resolve_turn_environment_selections(self.thread_manager.as_ref(), params.environments)?;
|
||||
let environment_requests =
|
||||
resolve_turn_environment_requests(self.thread_manager.as_ref(), params.environments)?;
|
||||
|
||||
let additional_context = map_additional_context(params.additional_context);
|
||||
let turn_has_input = !params.input.is_empty();
|
||||
@@ -624,7 +624,7 @@ impl TurnRequestProcessor {
|
||||
thread.as_ref(),
|
||||
cwd,
|
||||
runtime_workspace_roots,
|
||||
environment_selections,
|
||||
environment_requests,
|
||||
)
|
||||
.await;
|
||||
let thread_settings = self
|
||||
@@ -720,30 +720,33 @@ impl TurnRequestProcessor {
|
||||
thread: &CodexThread,
|
||||
cwd: Option<AbsolutePathBuf>,
|
||||
workspace_roots: Option<Vec<AbsolutePathBuf>>,
|
||||
environment_selections: Option<Vec<TurnEnvironmentSelection>>,
|
||||
environment_requests: Option<Vec<TurnEnvironmentRequest>>,
|
||||
) -> ThreadEnvironmentOverride {
|
||||
if cwd.is_none() && workspace_roots.is_none() && environment_selections.is_none() {
|
||||
if cwd.is_none() && workspace_roots.is_none() && environment_requests.is_none() {
|
||||
return ThreadEnvironmentOverride::default();
|
||||
}
|
||||
|
||||
// Explicit environment selections own their roots and pass through unchanged. Top-level
|
||||
// Explicit environment requests own their workspace roots and pass through unchanged. Top-level
|
||||
// `runtimeWorkspaceRoots` is only a compatibility input for default environments.
|
||||
if let Some(environment_selections) = environment_selections {
|
||||
if let Some(environment_requests) = environment_requests {
|
||||
let legacy_fallback_cwd = match cwd {
|
||||
Some(cwd) => cwd,
|
||||
None => match environment_selections
|
||||
None => match environment_requests
|
||||
.iter()
|
||||
.find(|selection| selection.environment_id == LOCAL_ENVIRONMENT_ID)
|
||||
.and_then(|selection| selection.cwd.to_abs_path().ok())
|
||||
.find(|request| request.environment_id == LOCAL_ENVIRONMENT_ID)
|
||||
.and_then(|request| request.cwd.to_abs_path().ok())
|
||||
{
|
||||
Some(cwd) => cwd,
|
||||
None => thread.config_snapshot().await.cwd().clone(),
|
||||
},
|
||||
};
|
||||
return ThreadEnvironmentOverride {
|
||||
environments: Some(TurnEnvironmentSelections::new(
|
||||
environment_requests: Some(TurnEnvironmentSelections::new(
|
||||
legacy_fallback_cwd,
|
||||
environment_selections,
|
||||
environment_requests
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::new)
|
||||
.collect(),
|
||||
)),
|
||||
..Default::default()
|
||||
};
|
||||
@@ -761,13 +764,16 @@ impl TurnRequestProcessor {
|
||||
legacy_fallback_cwd.clone(),
|
||||
),
|
||||
};
|
||||
let environment_selections = self
|
||||
let environment_requests = self
|
||||
.thread_manager
|
||||
.default_environment_selections(&legacy_fallback_cwd, &workspace_roots);
|
||||
.default_environment_requests(&legacy_fallback_cwd, &workspace_roots);
|
||||
ThreadEnvironmentOverride {
|
||||
environments: Some(TurnEnvironmentSelections::new(
|
||||
environment_requests: Some(TurnEnvironmentSelections::new(
|
||||
legacy_fallback_cwd,
|
||||
environment_selections,
|
||||
environment_requests
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::new)
|
||||
.collect(),
|
||||
)),
|
||||
runtime_workspace_roots: Some(workspace_roots),
|
||||
}
|
||||
@@ -783,7 +789,7 @@ impl TurnRequestProcessor {
|
||||
disabled_plugin_ids,
|
||||
environment_override:
|
||||
ThreadEnvironmentOverride {
|
||||
environments,
|
||||
environment_requests,
|
||||
runtime_workspace_roots,
|
||||
},
|
||||
approval_policy,
|
||||
@@ -806,7 +812,7 @@ impl TurnRequestProcessor {
|
||||
|
||||
let collaboration_mode =
|
||||
collaboration_mode.map(|mode| self.normalize_collaboration_mode(mode));
|
||||
let has_environment_override = environments.is_some();
|
||||
let has_environment_override = environment_requests.is_some();
|
||||
// `thread/settings/update` only acknowledges that the update was queued.
|
||||
// Clients that send dependent partial updates should wait for
|
||||
// `thread/settings/updated` or combine the fields in one request.
|
||||
@@ -842,9 +848,9 @@ impl TurnRequestProcessor {
|
||||
)));
|
||||
};
|
||||
let thread_config = thread.config().await;
|
||||
let cwd = environments.as_ref().map_or_else(
|
||||
let cwd = environment_requests.as_ref().map_or_else(
|
||||
|| snapshot.cwd().clone(),
|
||||
|environments| environments.legacy_fallback_cwd.clone(),
|
||||
|environment_requests| environment_requests.legacy_fallback_cwd.clone(),
|
||||
);
|
||||
let config = self
|
||||
.config_manager
|
||||
@@ -876,7 +882,7 @@ impl TurnRequestProcessor {
|
||||
.preview_thread_settings_overrides(CodexThreadSettingsOverrides {
|
||||
turn_extension_init: None,
|
||||
disabled_plugin_ids: disabled_plugin_ids.clone(),
|
||||
environments: environments.clone(),
|
||||
environments: environment_requests.clone(),
|
||||
runtime_workspace_roots: runtime_workspace_roots.clone(),
|
||||
approval_policy,
|
||||
approvals_reviewer,
|
||||
@@ -900,7 +906,7 @@ impl TurnRequestProcessor {
|
||||
|
||||
Ok(codex_protocol::protocol::ThreadSettingsOverrides {
|
||||
disabled_plugin_ids,
|
||||
environments,
|
||||
environments: environment_requests,
|
||||
runtime_workspace_roots,
|
||||
profile_workspace_roots,
|
||||
approval_policy,
|
||||
@@ -932,7 +938,7 @@ impl TurnRequestProcessor {
|
||||
thread.as_ref(),
|
||||
cwd,
|
||||
/*workspace_roots*/ None,
|
||||
/*environment_selections*/ None,
|
||||
/*environment_requests*/ None,
|
||||
)
|
||||
.await;
|
||||
let thread_settings = self
|
||||
|
||||
@@ -156,7 +156,7 @@ async fn user_verification_mcp_round_trip_requires_proof_in_full_access() -> Res
|
||||
OPENAI_ELICITATION_EXTENSION_ID.to_string(),
|
||||
json!({"userVerification": {}}),
|
||||
)]),
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
.await?
|
||||
|
||||
@@ -139,6 +139,7 @@ pub use codex_protocol::protocol::Op;
|
||||
pub use codex_protocol::protocol::SessionConfiguredEvent;
|
||||
pub use codex_protocol::protocol::SessionSource;
|
||||
pub use codex_protocol::protocol::ThreadSettingsOverrides;
|
||||
pub use codex_protocol::protocol::TurnEnvironmentRequest;
|
||||
pub use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
pub use codex_protocol::protocol::W3cTraceContext;
|
||||
pub use codex_protocol::user_input::UserInput;
|
||||
|
||||
@@ -1722,7 +1722,7 @@ async fn pending_environment_failure_reaches_child_and_grandchild() {
|
||||
let root = harness
|
||||
.manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![pending.clone()]),
|
||||
environments: Some(vec![pending.clone().into_request()]),
|
||||
..StartThreadOptions::new(harness.config.clone())
|
||||
})
|
||||
.await
|
||||
|
||||
@@ -11,6 +11,7 @@ use codex_protocol::protocol::Op;
|
||||
use codex_protocol::protocol::SessionSource;
|
||||
use codex_protocol::protocol::SubAgentSource;
|
||||
use codex_protocol::protocol::ThreadSource;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_protocol::user_input::UserInput;
|
||||
use serde_json::Value;
|
||||
use std::time::Duration;
|
||||
@@ -135,7 +136,11 @@ pub(crate) async fn run_codex_thread_interactive(
|
||||
inherited_exec_policy: Some(Arc::clone(&parent_session.services.exec_policy)),
|
||||
parent_rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(),
|
||||
parent_trace: None,
|
||||
environment_selections: parent_environments.to_selections(),
|
||||
environment_requests: parent_environments
|
||||
.to_selections()
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
thread_extension_init,
|
||||
turn_extension_init: Default::default(),
|
||||
client_mcp_extensions: parent_session.services.client_mcp_extensions.clone(),
|
||||
|
||||
@@ -713,7 +713,7 @@ impl CodexThread {
|
||||
fn thread_settings_update(overrides: CodexThreadSettingsOverrides) -> SessionSettingsUpdate {
|
||||
let CodexThreadSettingsOverrides {
|
||||
turn_extension_init,
|
||||
environments,
|
||||
environments: environment_requests,
|
||||
runtime_workspace_roots,
|
||||
profile_workspace_roots,
|
||||
approval_policy,
|
||||
@@ -742,7 +742,7 @@ impl CodexThread {
|
||||
approval_policy,
|
||||
approvals_reviewer,
|
||||
},
|
||||
environments,
|
||||
environments: environment_requests,
|
||||
runtime_workspace_roots,
|
||||
profile_workspace_roots,
|
||||
sandbox_policy,
|
||||
|
||||
@@ -26,6 +26,7 @@ use codex_protocol::protocol::EnvironmentConfigState;
|
||||
use codex_protocol::protocol::EnvironmentConnectionEvent;
|
||||
use codex_protocol::protocol::Event;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::TurnEnvironmentRequest;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_skills_extension::EnvironmentSkillRequirements;
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
@@ -122,15 +123,15 @@ pub(crate) fn combine_selected_capability_roots(
|
||||
combined_roots
|
||||
}
|
||||
|
||||
pub(crate) fn default_thread_environment_selections(
|
||||
pub(crate) fn default_thread_environment_requests(
|
||||
environment_manager: &EnvironmentManager,
|
||||
cwd: &AbsolutePathBuf,
|
||||
workspace_roots: &[AbsolutePathBuf],
|
||||
) -> Vec<TurnEnvironmentSelection> {
|
||||
) -> Vec<TurnEnvironmentRequest> {
|
||||
environment_manager
|
||||
.default_environment_ids()
|
||||
.into_iter()
|
||||
.map(|environment_id| TurnEnvironmentSelection {
|
||||
.map(|environment_id| TurnEnvironmentRequest {
|
||||
environment_id,
|
||||
cwd: PathUri::from_abs_path(cwd),
|
||||
workspace_roots: workspace_roots.iter().map(PathUri::from_abs_path).collect(),
|
||||
@@ -140,30 +141,26 @@ pub(crate) fn default_thread_environment_selections(
|
||||
}
|
||||
|
||||
/// Checks that environment IDs are registered and unique and that working directories are not too long.
|
||||
pub fn validate_environment_ids_and_cwds(
|
||||
pub fn validate_environment_ids_and_cwds<'a>(
|
||||
environment_manager: &EnvironmentManager,
|
||||
environments: &[TurnEnvironmentSelection],
|
||||
environments: impl IntoIterator<Item = (&'a str, &'a PathUri)>,
|
||||
) -> CodexResult<()> {
|
||||
let mut environment_ids = HashSet::with_capacity(environments.len());
|
||||
for environment in environments {
|
||||
if environment.cwd.inferred_native_path_string().len() > MAX_TURN_ENVIRONMENT_CWD_BYTES {
|
||||
let mut environment_ids = HashSet::new();
|
||||
for (environment_id, cwd) in environments {
|
||||
if cwd.inferred_native_path_string().len() > MAX_TURN_ENVIRONMENT_CWD_BYTES {
|
||||
return Err(CodexErr::InvalidRequest(
|
||||
"turn environment working directory exceeds the maximum size".to_string(),
|
||||
));
|
||||
}
|
||||
if !environment_ids.insert(environment.environment_id.as_str()) {
|
||||
if !environment_ids.insert(environment_id) {
|
||||
return Err(CodexErr::InvalidRequest(format!(
|
||||
"duplicate turn environment id `{}`",
|
||||
environment.environment_id
|
||||
"duplicate turn environment id `{environment_id}`"
|
||||
)));
|
||||
}
|
||||
environment_manager
|
||||
.get_environment(&environment.environment_id)
|
||||
.get_environment(environment_id)
|
||||
.ok_or_else(|| {
|
||||
CodexErr::InvalidRequest(format!(
|
||||
"unknown turn environment id `{}`",
|
||||
environment.environment_id
|
||||
))
|
||||
CodexErr::InvalidRequest(format!("unknown turn environment id `{environment_id}`"))
|
||||
})?;
|
||||
}
|
||||
Ok(())
|
||||
@@ -1321,6 +1318,7 @@ mod tests {
|
||||
use codex_protocol::models::ActivePermissionProfile;
|
||||
use codex_protocol::models::PermissionProfile;
|
||||
use codex_protocol::permissions::FileSystemSandboxPolicyContext;
|
||||
use codex_protocol::protocol::TurnEnvironmentRequest;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_protocol::sandbox::SandboxType;
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
@@ -1444,7 +1442,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn default_thread_environment_selections_use_manager_default_id() {
|
||||
async fn default_thread_environment_requests_use_manager_default_id() {
|
||||
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
||||
let cwd_uri = PathUri::from_abs_path(&cwd);
|
||||
let manager = EnvironmentManager::create_for_tests(
|
||||
@@ -1454,8 +1452,8 @@ mod tests {
|
||||
.await;
|
||||
|
||||
assert_eq!(
|
||||
default_thread_environment_selections(&manager, &cwd, std::slice::from_ref(&cwd)),
|
||||
vec![TurnEnvironmentSelection {
|
||||
default_thread_environment_requests(&manager, &cwd, std::slice::from_ref(&cwd)),
|
||||
vec![TurnEnvironmentRequest {
|
||||
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
||||
cwd: cwd_uri.clone(),
|
||||
workspace_roots: vec![cwd_uri],
|
||||
@@ -1465,7 +1463,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn toml_default_thread_environment_selections_include_local_and_remote() {
|
||||
async fn toml_default_thread_environment_requests_include_local_and_remote() {
|
||||
let temp_dir = tempfile::tempdir().expect("tempdir");
|
||||
std::fs::write(
|
||||
temp_dir.path().join("environments.toml"),
|
||||
@@ -1487,15 +1485,15 @@ url = "ws://127.0.0.1:8765"
|
||||
.expect("environment manager");
|
||||
|
||||
assert_eq!(
|
||||
default_thread_environment_selections(&manager, &cwd, std::slice::from_ref(&cwd)),
|
||||
default_thread_environment_requests(&manager, &cwd, std::slice::from_ref(&cwd)),
|
||||
vec![
|
||||
TurnEnvironmentSelection {
|
||||
TurnEnvironmentRequest {
|
||||
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
||||
cwd: cwd_uri.clone(),
|
||||
workspace_roots: vec![cwd_uri.clone()],
|
||||
config: EnvironmentConfigState::FromThread,
|
||||
},
|
||||
TurnEnvironmentSelection {
|
||||
TurnEnvironmentRequest {
|
||||
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
||||
cwd: cwd_uri.clone(),
|
||||
workspace_roots: vec![cwd_uri],
|
||||
@@ -1506,13 +1504,13 @@ url = "ws://127.0.0.1:8765"
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn default_thread_environment_selections_empty_when_default_disabled() {
|
||||
async fn default_thread_environment_requests_empty_when_default_disabled() {
|
||||
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
||||
let manager = environment_manager_without_environments();
|
||||
|
||||
assert_eq!(
|
||||
default_thread_environment_selections(&manager, &cwd, std::slice::from_ref(&cwd)),
|
||||
Vec::<TurnEnvironmentSelection>::new()
|
||||
default_thread_environment_requests(&manager, &cwd, std::slice::from_ref(&cwd)),
|
||||
Vec::<TurnEnvironmentRequest>::new()
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ use super::*;
|
||||
use crate::guardian::input_budget::CheckpointRecovery;
|
||||
use codex_guardian_reviewer::ReviewerPool;
|
||||
use codex_guardian_reviewer::ReviewerRequest;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
|
||||
/// Controls whether selection may reuse a session or must start from the parent checkpoint.
|
||||
#[derive(Clone, Copy, PartialEq, Eq)]
|
||||
@@ -148,7 +149,14 @@ impl PreparedGuardianContext {
|
||||
}),
|
||||
}),
|
||||
initial_history: initial_history.unwrap_or(InitialHistory::New),
|
||||
environments: Some(self.context.environments().to_selections()),
|
||||
environments: Some(
|
||||
self.context
|
||||
.environments()
|
||||
.to_selections()
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
),
|
||||
inherited_environments: Some(self.context.environments().clone()),
|
||||
client_mcp_extensions: self.parent.services.client_mcp_extensions.clone(),
|
||||
..crate::StartThreadOptions::new(config)
|
||||
|
||||
@@ -155,6 +155,7 @@ use codex_protocol::protocol::ThreadSource;
|
||||
use codex_protocol::protocol::TurnAbortReason;
|
||||
use codex_protocol::protocol::TurnContextItem;
|
||||
use codex_protocol::protocol::TurnContextNetworkItem;
|
||||
use codex_protocol::protocol::TurnEnvironmentRequest;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_protocol::protocol::TurnStartedEvent;
|
||||
use codex_protocol::protocol::W3cTraceContext;
|
||||
@@ -474,7 +475,7 @@ pub(crate) struct SessionSpawnArgs {
|
||||
pub(crate) parent_rollout_thread_trace: ThreadTraceContext,
|
||||
pub(crate) user_shell_override: Option<shell::Shell>,
|
||||
pub(crate) parent_trace: Option<W3cTraceContext>,
|
||||
pub(crate) environment_selections: Vec<TurnEnvironmentSelection>,
|
||||
pub(crate) environment_requests: Vec<TurnEnvironmentRequest>,
|
||||
pub(crate) thread_extension_init: ExtensionDataInit,
|
||||
pub(crate) turn_extension_init: ExtensionDataInit,
|
||||
pub(crate) client_mcp_extensions: ClientMcpExtensions,
|
||||
@@ -583,7 +584,7 @@ impl Session {
|
||||
inherited_environments,
|
||||
parent_rollout_thread_trace,
|
||||
parent_trace: _,
|
||||
environment_selections,
|
||||
environment_requests,
|
||||
thread_extension_init,
|
||||
turn_extension_init,
|
||||
client_mcp_extensions,
|
||||
@@ -604,6 +605,10 @@ impl Session {
|
||||
.get::<codex_extension_api::SessionIsolation>()
|
||||
.map(|policy| *policy)
|
||||
.unwrap_or_default();
|
||||
let environment_selections = environment_requests
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::new)
|
||||
.collect::<Vec<_>>();
|
||||
// Enforce snapshot-only instructions for both managed and inline isolated sessions.
|
||||
let instructions = if isolation == codex_extension_api::SessionIsolation::Isolated {
|
||||
SessionInstructions {
|
||||
|
||||
@@ -266,7 +266,7 @@ impl Session {
|
||||
};
|
||||
let TurnSettingsUpdate {
|
||||
approvals_reviewer,
|
||||
environments,
|
||||
environments: environment_requests,
|
||||
model,
|
||||
effort,
|
||||
summary,
|
||||
@@ -286,7 +286,9 @@ impl Session {
|
||||
// progress, finish, or be cancelled while this awaits; we don't hold the update locks here.
|
||||
let prepared = if updates_step_settings {
|
||||
let current_environments = self.services.turn_environments.selections();
|
||||
let proposed = environments.as_deref().unwrap_or(¤t_environments);
|
||||
let proposed = environment_requests
|
||||
.as_deref()
|
||||
.unwrap_or(¤t_environments);
|
||||
self.prepare_step_settings_activation(&turn_context, ¤t, &update, proposed)
|
||||
.await
|
||||
.map(Some)
|
||||
@@ -295,8 +297,9 @@ impl Session {
|
||||
};
|
||||
|
||||
// Validate the environment configuration supplied in the update.
|
||||
let environment_config_validation =
|
||||
environments.as_deref().map(validate_environment_configs);
|
||||
let environment_config_validation = environment_requests
|
||||
.as_deref()
|
||||
.map(validate_environment_configs);
|
||||
|
||||
// Confirm the task we originally targeted is still running.
|
||||
let active = self.active_turn.lock().await;
|
||||
@@ -326,12 +329,16 @@ impl Session {
|
||||
// Environment configuration can arrive before its executor connects. If this update
|
||||
// omits environments, use what the running turn's manager knows now.
|
||||
let current_environments = self.services.turn_environments.selections();
|
||||
let proposed = environments.as_deref().unwrap_or(¤t_environments);
|
||||
let proposed = environment_requests
|
||||
.as_deref()
|
||||
.unwrap_or(¤t_environments);
|
||||
let validation = (|| {
|
||||
if let Some(configs) = environment_config_validation {
|
||||
validate_environment_ids_and_cwds(
|
||||
&self.services.turn_environments.environment_manager(),
|
||||
proposed,
|
||||
proposed
|
||||
.iter()
|
||||
.map(|selection| (selection.environment_id.as_str(), &selection.cwd)),
|
||||
)
|
||||
.map_err(|error| error.to_string())?;
|
||||
ensure_configs_stay_owner_provided(¤t_environments, proposed)
|
||||
@@ -367,7 +374,7 @@ impl Session {
|
||||
.next_step_settings
|
||||
.store(Arc::new(settings));
|
||||
}
|
||||
if environments.is_some() {
|
||||
if environment_requests.is_some() {
|
||||
self.services.turn_environments.update_selections(proposed);
|
||||
}
|
||||
TurnSettingsUpdateOutcome::Applied
|
||||
|
||||
@@ -1321,7 +1321,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
|
||||
parent_rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(),
|
||||
user_shell_override: None,
|
||||
parent_trace: None,
|
||||
environment_selections: Vec::new(),
|
||||
environment_requests: Vec::new(),
|
||||
thread_extension_init,
|
||||
turn_extension_init: Default::default(),
|
||||
client_mcp_extensions: ClientMcpExtensions::default(),
|
||||
|
||||
@@ -50,7 +50,7 @@ pub(super) fn prepare_update(
|
||||
turn_extension_init,
|
||||
} = overrides.into();
|
||||
let ThreadSettingsOverrides {
|
||||
environments,
|
||||
environments: environment_requests,
|
||||
runtime_workspace_roots,
|
||||
profile_workspace_roots,
|
||||
approval_policy,
|
||||
@@ -79,7 +79,7 @@ pub(super) fn prepare_update(
|
||||
approval_policy,
|
||||
approvals_reviewer,
|
||||
},
|
||||
environments,
|
||||
environments: environment_requests,
|
||||
runtime_workspace_roots,
|
||||
profile_workspace_roots,
|
||||
sandbox_policy,
|
||||
|
||||
@@ -16,7 +16,7 @@ use crate::config::Config;
|
||||
use crate::config::ThreadStoreConfig;
|
||||
use crate::current_time::TimeProvider;
|
||||
use crate::environment_selection::TurnEnvironmentSnapshot;
|
||||
use crate::environment_selection::default_thread_environment_selections;
|
||||
use crate::environment_selection::default_thread_environment_requests;
|
||||
use crate::mcp::McpManager;
|
||||
use crate::rollout::truncation;
|
||||
use crate::session::ForkPersistence;
|
||||
@@ -82,6 +82,7 @@ use codex_protocol::protocol::ThreadHistoryMode;
|
||||
use codex_protocol::protocol::ThreadSource;
|
||||
use codex_protocol::protocol::TurnAbortReason;
|
||||
use codex_protocol::protocol::TurnAbortedEvent;
|
||||
use codex_protocol::protocol::TurnEnvironmentRequest;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_protocol::protocol::W3cTraceContext;
|
||||
use codex_rollout::state_db::StateDbHandle;
|
||||
@@ -282,7 +283,7 @@ pub struct StartThreadOptions {
|
||||
pub dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
|
||||
pub metrics_service_name: Option<String>,
|
||||
pub parent_trace: Option<W3cTraceContext>,
|
||||
pub environments: Option<Vec<TurnEnvironmentSelection>>,
|
||||
pub environments: Option<Vec<TurnEnvironmentRequest>>,
|
||||
/// Existing environment bindings captured by an internal caller.
|
||||
pub inherited_environments: Option<TurnEnvironmentSnapshot>,
|
||||
/// Explicit global instructions carried by an internal caller instead of loading them again.
|
||||
@@ -874,12 +875,12 @@ impl ThreadManager {
|
||||
});
|
||||
}
|
||||
|
||||
pub fn default_environment_selections(
|
||||
pub fn default_environment_requests(
|
||||
&self,
|
||||
cwd: &AbsolutePathBuf,
|
||||
workspace_roots: &[AbsolutePathBuf],
|
||||
) -> Vec<TurnEnvironmentSelection> {
|
||||
default_thread_environment_selections(
|
||||
) -> Vec<TurnEnvironmentRequest> {
|
||||
default_thread_environment_requests(
|
||||
self.state.environment_manager.as_ref(),
|
||||
cwd,
|
||||
workspace_roots,
|
||||
@@ -1946,7 +1947,12 @@ impl ThreadManagerState {
|
||||
session_source: Some(session_source),
|
||||
thread_source,
|
||||
metrics_service_name,
|
||||
environments,
|
||||
environments: environments.map(|selections| {
|
||||
selections
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect()
|
||||
}),
|
||||
client_mcp_extensions,
|
||||
dynamic_tools,
|
||||
..StartThreadOptions::new(config)
|
||||
@@ -1985,7 +1991,12 @@ impl ThreadManagerState {
|
||||
initial_history,
|
||||
session_source: Some(session_source),
|
||||
thread_source,
|
||||
environments: environment_selections,
|
||||
environments: environment_selections.map(|selections| {
|
||||
selections
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect()
|
||||
}),
|
||||
client_mcp_extensions,
|
||||
..StartThreadOptions::new(config)
|
||||
};
|
||||
@@ -2020,7 +2031,12 @@ impl ThreadManagerState {
|
||||
history_mode,
|
||||
session_source: Some(session_source),
|
||||
thread_source,
|
||||
environments,
|
||||
environments: environments.map(|selections| {
|
||||
selections
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect()
|
||||
}),
|
||||
thread_extension_init,
|
||||
client_mcp_extensions,
|
||||
..StartThreadOptions::new(config)
|
||||
@@ -2087,7 +2103,7 @@ impl ThreadManagerState {
|
||||
dynamic_tools,
|
||||
metrics_service_name,
|
||||
parent_trace,
|
||||
environments,
|
||||
environments: environment_requests,
|
||||
inherited_environments: captured_environments,
|
||||
user_instructions: supplied_user_instructions,
|
||||
mut thread_extension_init,
|
||||
@@ -2111,18 +2127,22 @@ impl ThreadManagerState {
|
||||
}
|
||||
});
|
||||
thread_extension_init.insert(isolation);
|
||||
let environments = environments
|
||||
let environment_requests = environment_requests
|
||||
.or_else(|| {
|
||||
let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. }) = &session_source
|
||||
else {
|
||||
return None;
|
||||
};
|
||||
inherited_environments
|
||||
.as_ref()
|
||||
.map(TurnEnvironmentSnapshot::inheritable_selections)
|
||||
inherited_environments.as_ref().map(|snapshot| {
|
||||
snapshot
|
||||
.inheritable_selections()
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect()
|
||||
})
|
||||
})
|
||||
.unwrap_or_else(|| {
|
||||
default_thread_environment_selections(
|
||||
default_thread_environment_requests(
|
||||
self.environment_manager.as_ref(),
|
||||
&config.cwd,
|
||||
&config.workspace_roots,
|
||||
@@ -2334,7 +2354,7 @@ impl ThreadManagerState {
|
||||
parent_rollout_thread_trace,
|
||||
user_shell_override,
|
||||
parent_trace,
|
||||
environment_selections: environments,
|
||||
environment_requests,
|
||||
thread_extension_init,
|
||||
turn_extension_init,
|
||||
client_mcp_extensions,
|
||||
|
||||
@@ -1387,7 +1387,12 @@ async fn spawn_internal_session_preserves_parent_lineage_without_forking_history
|
||||
initial_history: InitialHistory::Forked(vec![RolloutItem::ResponseItem(
|
||||
user_msg("parent history must not be inherited").into(),
|
||||
)]),
|
||||
environments: Some(reviewer_environments),
|
||||
environments: Some(
|
||||
reviewer_environments
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
),
|
||||
..StartThreadOptions::new(config)
|
||||
},
|
||||
)
|
||||
@@ -1990,7 +1995,13 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
|
||||
let source = manager
|
||||
.start_thread(StartThreadOptions {
|
||||
history_mode: Some(ThreadHistoryMode::Legacy),
|
||||
environments: Some(environments.clone()),
|
||||
environments: Some(
|
||||
environments
|
||||
.clone()
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
),
|
||||
..StartThreadOptions::new(source_config)
|
||||
})
|
||||
.await
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
use codex_protocol::protocol::TurnEnvironmentRequest;
|
||||
use std::future::Future;
|
||||
use std::io::ErrorKind;
|
||||
use std::mem::swap;
|
||||
@@ -127,8 +128,9 @@ impl UserInstructionsProvider for RecordingUserInstructionsProvider {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn local(cwd: AbsolutePathBuf) -> TurnEnvironmentSelection {
|
||||
TurnEnvironmentSelection {
|
||||
/// Builds environment input without attaching any thread's capability roots.
|
||||
pub fn local_request(cwd: AbsolutePathBuf) -> TurnEnvironmentRequest {
|
||||
TurnEnvironmentRequest {
|
||||
environment_id: codex_exec_server::LOCAL_ENVIRONMENT_ID.to_string(),
|
||||
cwd: PathUri::from_abs_path(&cwd),
|
||||
workspace_roots: vec![PathUri::from_abs_path(&cwd)],
|
||||
@@ -136,6 +138,10 @@ pub fn local(cwd: AbsolutePathBuf) -> TurnEnvironmentSelection {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn local(cwd: AbsolutePathBuf) -> TurnEnvironmentSelection {
|
||||
TurnEnvironmentSelection::new(local_request(cwd))
|
||||
}
|
||||
|
||||
/// Builds explicit environment configuration with the test thread's permissions and shell settings.
|
||||
pub fn environment_config_for_selection(
|
||||
config: &Config,
|
||||
@@ -243,6 +249,16 @@ impl TestEnv {
|
||||
&self.selection
|
||||
}
|
||||
|
||||
/// Requests the fixture's environment without copying a thread's root choices.
|
||||
pub fn request(&self) -> TurnEnvironmentRequest {
|
||||
TurnEnvironmentRequest {
|
||||
environment_id: self.selection.environment_id.clone(),
|
||||
cwd: self.selection.cwd.clone(),
|
||||
workspace_roots: self.selection.workspace_roots.clone(),
|
||||
config: self.selection.config.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn local_cwd_temp_dir(&self) -> Option<Arc<TempDir>> {
|
||||
self.local_cwd_temp_dir.clone()
|
||||
}
|
||||
@@ -889,7 +905,7 @@ impl TestCodexBuilder {
|
||||
.await?
|
||||
}
|
||||
(None, None) => {
|
||||
let environments = if test_env.selection().cwd.infer_path_convention()
|
||||
let environment_requests = if test_env.selection().cwd.infer_path_convention()
|
||||
== Some(PathConvention::Windows)
|
||||
&& PathUri::from_abs_path(&config.cwd) != test_env.selection().cwd
|
||||
{
|
||||
@@ -897,15 +913,15 @@ impl TestCodexBuilder {
|
||||
let mut selection = test_env.selection().clone();
|
||||
selection.cwd = cwd.clone();
|
||||
selection.workspace_roots = vec![cwd];
|
||||
test_env.selection = selection.clone();
|
||||
Some(vec![selection])
|
||||
test_env.selection = selection;
|
||||
Some(vec![test_env.request()])
|
||||
} else {
|
||||
None
|
||||
};
|
||||
Box::pin(thread_manager.start_thread(StartThreadOptions {
|
||||
history_mode: self.history_mode,
|
||||
client_mcp_extensions: client_mcp_extensions(),
|
||||
environments,
|
||||
environments: environment_requests,
|
||||
..StartThreadOptions::new(config.clone())
|
||||
}))
|
||||
.await?
|
||||
@@ -1022,6 +1038,22 @@ pub struct TestCodex {
|
||||
}
|
||||
|
||||
impl TestCodex {
|
||||
/// Starts another test thread with the fixture's current environments and configuration.
|
||||
/// Keeping request construction here lets ordinary tests ignore runtime selection fields.
|
||||
pub async fn start_thread_options(&self) -> StartThreadOptions {
|
||||
StartThreadOptions {
|
||||
environments: Some(
|
||||
self.codex
|
||||
.environment_selections()
|
||||
.await
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
),
|
||||
..StartThreadOptions::new(self.config.clone())
|
||||
}
|
||||
}
|
||||
|
||||
pub fn cwd_path(&self) -> &Path {
|
||||
self.cwd.path()
|
||||
}
|
||||
|
||||
@@ -75,7 +75,7 @@ async fn abort_lifecycle_finishes_before_terminal_event(
|
||||
let thread = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
dynamic_tools: vec![DynamicToolSpec::Function(DynamicToolFunctionSpec {
|
||||
name: "gate".to_string(),
|
||||
description: "Wait for the host.".to_string(),
|
||||
|
||||
@@ -259,7 +259,7 @@ async fn host_threads_preserve_lineage_settings_and_resume_routing() -> anyhow::
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
reserved_thread_id: Some(ThreadId::new()),
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id: root_id,
|
||||
depth: 1,
|
||||
@@ -343,7 +343,7 @@ async fn host_threads_preserve_lineage_settings_and_resume_routing() -> anyhow::
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
reserved_thread_id: Some(reserved_thread_id),
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
..StartThreadOptions::new(child_config.clone())
|
||||
})
|
||||
.await;
|
||||
@@ -402,7 +402,7 @@ async fn host_factory_follows_thread_lifecycle() -> anyhow::Result<()> {
|
||||
let root_id = test.session_configured.thread_id;
|
||||
let manager = &test.thread_manager;
|
||||
let options = || StartThreadOptions {
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
};
|
||||
let internal = manager
|
||||
|
||||
@@ -29,7 +29,7 @@ use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::Op;
|
||||
use codex_protocol::protocol::ThreadHistoryMode;
|
||||
use codex_protocol::protocol::ThreadSettingsOverrides;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_protocol::protocol::TurnEnvironmentRequest;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelections;
|
||||
use codex_protocol::request_user_input::RequestUserInputAnswer;
|
||||
use codex_protocol::request_user_input::RequestUserInputResponse;
|
||||
@@ -1143,7 +1143,7 @@ impl ThreadInstructionsFixture {
|
||||
let thread = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
thread_instructions_provider: Some(provider.clone()),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
@@ -1337,9 +1337,8 @@ async fn isolated_guardian_keeps_applied_thread_instructions() -> Result<()> {
|
||||
let parent = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(test.codex.environment_selections().await),
|
||||
thread_instructions_provider: Some(provider.clone()),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
..test.start_thread_options().await
|
||||
})
|
||||
.await?;
|
||||
// Publish an update after the parent captured its instructions, before it starts Guardian.
|
||||
@@ -1408,9 +1407,7 @@ async fn thread_provider_enforces_its_own_limit_before_startup_and_sampling() ->
|
||||
.test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![
|
||||
fixture.test.executor_environment().selection().clone(),
|
||||
]),
|
||||
environments: Some(vec![fixture.test.executor_environment().request()]),
|
||||
thread_instructions_provider: Some(Arc::new(RecordingThreadInstructionsProvider::new(
|
||||
Some(oversized.clone()),
|
||||
))),
|
||||
@@ -1892,13 +1889,13 @@ async fn multi_environment_project_instructions_share_one_byte_budget() -> Resul
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![
|
||||
TurnEnvironmentSelection {
|
||||
TurnEnvironmentRequest {
|
||||
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
||||
cwd: test.executor_environment().selection().cwd.clone(),
|
||||
workspace_roots: vec![test.executor_environment().selection().cwd.clone()],
|
||||
config: EnvironmentConfigState::FromThread,
|
||||
},
|
||||
TurnEnvironmentSelection {
|
||||
TurnEnvironmentRequest {
|
||||
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
||||
cwd: PathUri::from_host_native_path(local_root.path())?,
|
||||
workspace_roots: vec![PathUri::from_host_native_path(local_root.path())?],
|
||||
@@ -1976,13 +1973,13 @@ async fn multi_environment_thread_refreshes_global_and_keeps_repository_snapshot
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![
|
||||
TurnEnvironmentSelection {
|
||||
TurnEnvironmentRequest {
|
||||
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
||||
cwd: test.executor_environment().selection().cwd.clone(),
|
||||
workspace_roots: vec![test.executor_environment().selection().cwd.clone()],
|
||||
config: EnvironmentConfigState::FromThread,
|
||||
},
|
||||
TurnEnvironmentSelection {
|
||||
TurnEnvironmentRequest {
|
||||
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
||||
cwd: PathUri::from_host_native_path(local_root.path())?,
|
||||
workspace_roots: vec![PathUri::from_host_native_path(local_root.path())?],
|
||||
|
||||
@@ -101,7 +101,7 @@ async fn concurrent_preparation_preserves_warnings_and_errors(
|
||||
let thread = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
dynamic_tools,
|
||||
thread_instructions_provider: Some(observer.clone()),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
|
||||
@@ -1983,8 +1983,7 @@ async fn apply_patch_turn_diff_skips_git_root_when_feature_is_enabled(
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
metrics_service_name: Some(originator.to_string()),
|
||||
environments: Some(test.codex.environment_selections().await),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
..test.start_thread_options().await
|
||||
})
|
||||
.await?
|
||||
.thread;
|
||||
|
||||
@@ -51,9 +51,7 @@ async fn remote_compact_v2_token_estimate_ignores_message_bookkeeping_and_json_e
|
||||
.test()
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![
|
||||
harness.test().executor_environment().selection().clone(),
|
||||
]),
|
||||
environments: Some(vec![harness.test().executor_environment().request()]),
|
||||
..StartThreadOptions::new(harness.test().config.clone())
|
||||
})
|
||||
.await?
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
//! Checkpoint replay preserves retained evidence across migration and independent review rollback.
|
||||
//! A retired managed opt-out cannot disable capture or later checkpoint promotion.
|
||||
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::Result;
|
||||
@@ -73,7 +74,12 @@ pub(super) async fn resume(
|
||||
Ok(test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(environments),
|
||||
environments: Some(
|
||||
environments
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
),
|
||||
initial_history: InitialHistory::Resumed(ResumedHistory {
|
||||
history_revision: None,
|
||||
conversation_id: thread_id,
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
//! Genuine sender context stays reviewer-only and is replaced on every delivery.
|
||||
|
||||
use anyhow::Result;
|
||||
use codex_core::StartThreadOptions;
|
||||
use codex_core::TurnInputRequest;
|
||||
use codex_core::config::Constrained;
|
||||
use codex_protocol::config_types::ApprovalsReviewer;
|
||||
@@ -52,10 +51,7 @@ async fn guardian_receives_sender_user_messages_by_default(
|
||||
let sender = test.session_configured.thread_id;
|
||||
let receiver = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(test.codex.environment_selections().await),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
.start_thread(test.start_thread_options().await)
|
||||
.await?
|
||||
.thread;
|
||||
let done = || responses::sse(vec![responses::ev_completed("done")]);
|
||||
|
||||
@@ -558,8 +558,7 @@ async fn executor_browser_and_computer_use_cleanup_hooks_use_separate_mcp_routes
|
||||
agent_role: None,
|
||||
})),
|
||||
thread_source: Some(ThreadSource::Subagent),
|
||||
environments: Some(fixture.test.codex.environment_selections().await),
|
||||
..StartThreadOptions::new(fixture.test.config.clone())
|
||||
..fixture.test.start_thread_options().await
|
||||
})
|
||||
.await?;
|
||||
fixture.test.codex = child.thread;
|
||||
|
||||
@@ -48,9 +48,11 @@ async fn dropping_startup_cleans_up_while_required_mcp_is_stalled() -> anyhow::R
|
||||
);
|
||||
config.mcp_servers.set(servers)?;
|
||||
let thread_id = fixture.thread_manager.reserve_thread_id();
|
||||
let mut options = StartThreadOptions::new(config);
|
||||
let mut options = StartThreadOptions {
|
||||
config,
|
||||
..fixture.start_thread_options().await
|
||||
};
|
||||
options.reserved_thread_id = Some(thread_id);
|
||||
options.environments = Some(fixture.codex.environment_selections().await);
|
||||
options
|
||||
.thread_extension_init
|
||||
.insert(SessionIsolation::Isolated);
|
||||
@@ -103,10 +105,9 @@ async fn dropping_startup_before_receiving_the_agent_finishes_cleanup() -> anyho
|
||||
}
|
||||
let server = responses::start_mock_server().await;
|
||||
let fixture = test_codex().build_with_auto_env(&server).await?;
|
||||
let mut options = StartThreadOptions::new(fixture.config.clone());
|
||||
let mut options = fixture.start_thread_options().await;
|
||||
let thread_id = fixture.thread_manager.reserve_thread_id();
|
||||
options.reserved_thread_id = Some(thread_id);
|
||||
options.environments = Some(fixture.codex.environment_selections().await);
|
||||
options
|
||||
.thread_extension_init
|
||||
.insert(SessionIsolation::Isolated);
|
||||
@@ -149,8 +150,7 @@ async fn owner_cancellation_closes_agent_and_preserves_history_and_parent() -> a
|
||||
|
||||
let server = responses::start_mock_server().await;
|
||||
let fixture = test_codex().build_with_auto_env(&server).await?;
|
||||
let mut options = StartThreadOptions::new(fixture.config.clone());
|
||||
options.environments = Some(fixture.codex.environment_selections().await);
|
||||
let mut options = fixture.start_thread_options().await;
|
||||
options
|
||||
.thread_extension_init
|
||||
.insert(SessionIsolation::Isolated);
|
||||
@@ -161,9 +161,8 @@ async fn owner_cancellation_closes_agent_and_preserves_history_and_parent() -> a
|
||||
.start_thread_until(options, cancelled.clone().cancelled_owned(), &tasks)
|
||||
.await?;
|
||||
// A failed startup with the same ID must leave the original agent's writer intact.
|
||||
let mut duplicate = StartThreadOptions::new(fixture.config.clone());
|
||||
let mut duplicate = fixture.start_thread_options().await;
|
||||
duplicate.reserved_thread_id = Some(agent.thread_id);
|
||||
duplicate.environments = Some(fixture.codex.environment_selections().await);
|
||||
duplicate
|
||||
.thread_extension_init
|
||||
.insert(SessionIsolation::Isolated);
|
||||
@@ -268,8 +267,7 @@ async fn startup_allowlist_controls_advertising_and_execution(
|
||||
})
|
||||
.build_with_auto_env(&server)
|
||||
.await?;
|
||||
let mut options = StartThreadOptions::new(fixture.config.clone());
|
||||
options.environments = Some(fixture.codex.environment_selections().await);
|
||||
let mut options = fixture.start_thread_options().await;
|
||||
options
|
||||
.thread_extension_init
|
||||
.insert(SessionIsolation::Isolated);
|
||||
@@ -381,8 +379,7 @@ async fn startup_tool_policy_controls_additional_permission_parameters(
|
||||
})
|
||||
.build_with_auto_env(&server)
|
||||
.await?;
|
||||
let mut options = StartThreadOptions::new(fixture.config.clone());
|
||||
options.environments = Some(fixture.codex.environment_selections().await);
|
||||
let mut options = fixture.start_thread_options().await;
|
||||
options
|
||||
.thread_extension_init
|
||||
.insert(SessionIsolation::Isolated);
|
||||
|
||||
@@ -595,7 +595,7 @@ approvals_reviewer = "user"
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
})),
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
.await?
|
||||
|
||||
@@ -244,7 +244,7 @@ async fn hosted_apps_protocol_override_preserves_native_verification() -> anyhow
|
||||
OPENAI_ELICITATION_EXTENSION_ID.to_string(),
|
||||
json!({"userVerification": {}}),
|
||||
)]),
|
||||
environments: Some(vec![fixture.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![fixture.executor_environment().request()]),
|
||||
..StartThreadOptions::new(fixture.config.clone())
|
||||
})
|
||||
.await?
|
||||
|
||||
@@ -12,6 +12,7 @@ use codex_otel::MetricsConfig;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::SessionMeta;
|
||||
use codex_protocol::protocol::SessionMetaLine;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_protocol::user_input::UserInput;
|
||||
use core_test_support::responses;
|
||||
use core_test_support::skip_if_no_network;
|
||||
@@ -119,7 +120,15 @@ async fn recovery_without_a_snapshot_keeps_each_threads_originator() -> Result<(
|
||||
git: None,
|
||||
},
|
||||
)]),
|
||||
environments: Some(fixture.codex.environment_selections().await),
|
||||
environments: Some(
|
||||
fixture
|
||||
.codex
|
||||
.environment_selections()
|
||||
.await
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
),
|
||||
..StartThreadOptions::new(config.clone())
|
||||
})
|
||||
.await?
|
||||
|
||||
@@ -158,7 +158,7 @@ pub(super) async fn mcp_server_elicitation_scenario(
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
session_source: Some(session_source),
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
..StartThreadOptions::new(config)
|
||||
})
|
||||
.await?
|
||||
|
||||
@@ -96,7 +96,7 @@ async fn mcp_user_verification_rejects_configured_servers(source: CapabilitySour
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
client_mcp_extensions,
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
..StartThreadOptions::new(config)
|
||||
})
|
||||
.await?
|
||||
|
||||
@@ -4,6 +4,7 @@ use anyhow::Result;
|
||||
use anyhow::anyhow;
|
||||
use codex_features::Feature;
|
||||
use codex_login::CodexAuth;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use core_test_support::responses::ev_assistant_message;
|
||||
use core_test_support::responses::ev_completed;
|
||||
use core_test_support::responses::ev_function_call_with_namespace;
|
||||
@@ -185,7 +186,14 @@ async fn ephemeral_fork_shares_cache_routing_but_keeps_session_identity() -> Res
|
||||
.fork_legacy_thread(
|
||||
ForkSnapshot::TruncateBeforeNthUserMessage(usize::MAX),
|
||||
StartThreadOptions {
|
||||
environments: Some(test.codex.environment_selections().await),
|
||||
environments: Some(
|
||||
test.codex
|
||||
.environment_selections()
|
||||
.await
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
),
|
||||
..StartThreadOptions::new(config)
|
||||
},
|
||||
test.codex.rollout_path().expect("parent rollout"),
|
||||
|
||||
@@ -696,8 +696,7 @@ async fn conversation_websocket_transports_send_codex_headers_without_creating_a
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
thread_source: thread_source.clone(),
|
||||
environments: Some(test.codex.config_snapshot().await.environments.environments),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
..test.start_thread_options().await
|
||||
})
|
||||
.await?;
|
||||
let codex = &conversation.thread;
|
||||
@@ -848,8 +847,7 @@ async fn conversation_webrtc_frameless_chatgpt_sends_codex_headers_to_backend(
|
||||
|
||||
let mut options = StartThreadOptions {
|
||||
thread_source: thread_source.clone(),
|
||||
environments: Some(test.codex.config_snapshot().await.environments.environments),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
..test.start_thread_options().await
|
||||
};
|
||||
let conversation = if thread_source == Some(ThreadSource::GuardianReview) {
|
||||
options.session_source = Some(SessionSource::Internal(InternalSessionSource::Guardian));
|
||||
|
||||
@@ -77,6 +77,7 @@ use codex_protocol::protocol::Op;
|
||||
use codex_protocol::protocol::ReviewDecision;
|
||||
use codex_protocol::protocol::SandboxPolicy;
|
||||
use codex_protocol::protocol::ThreadSettingsOverrides;
|
||||
use codex_protocol::protocol::TurnEnvironmentRequest;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelections;
|
||||
use codex_protocol::protocol::TurnSettingsUpdate;
|
||||
@@ -1692,7 +1693,7 @@ async fn shared_executor_keeps_ready_capability_roots_scoped_to_each_attachment(
|
||||
let second = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![TurnEnvironmentSelection {
|
||||
environments: Some(vec![TurnEnvironmentRequest {
|
||||
config: EnvironmentConfigState::Ready(EnvironmentConfig {
|
||||
allow_login_shell: true,
|
||||
workspace_roots: selection.workspace_roots.clone(),
|
||||
@@ -1706,7 +1707,7 @@ async fn shared_executor_keeps_ready_capability_roots_scoped_to_each_attachment(
|
||||
network_policy: None,
|
||||
selected_capability_roots: vec![root("startup-root"), root("second-root")],
|
||||
}),
|
||||
..selection.clone()
|
||||
..selection.clone().into_request()
|
||||
}]),
|
||||
thread_extension_init: second_thread_init,
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
@@ -1967,7 +1968,7 @@ async fn pending_attachment_installs_configuration_before_waiting_turn_resumes()
|
||||
};
|
||||
let start_pending_thread = || {
|
||||
test.thread_manager.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![pending_selection.clone()]),
|
||||
environments: Some(vec![pending_selection.clone().into_request()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
};
|
||||
@@ -2416,7 +2417,7 @@ async fn future_pending_environment_can_finish_without_retargeting_the_active_tu
|
||||
let thread = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![active.clone()]),
|
||||
environments: Some(vec![active.clone().into_request()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
.await?
|
||||
@@ -2530,7 +2531,7 @@ async fn active_environment_update_wakes_the_old_wait_with_the_new_selection() -
|
||||
let thread = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![original.clone()]),
|
||||
environments: Some(vec![original.clone().into_request()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
.await?
|
||||
@@ -2799,7 +2800,7 @@ async fn ready_before_selection_resolves_resumed_thread_capability_root_after_wa
|
||||
let resumed = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![selection.clone()]),
|
||||
environments: Some(vec![selection.clone().into_request()]),
|
||||
thread_extension_init,
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
|
||||
@@ -71,7 +71,7 @@ async fn stale_roots_do_not_connect_unselected_executors(
|
||||
let thread = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![selection.clone()]),
|
||||
environments: Some(vec![selection.clone().into_request()]),
|
||||
thread_extension_init,
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
|
||||
@@ -89,7 +89,7 @@ pub(crate) async fn pending_subagent_scenario(
|
||||
let root = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![pending.clone()]),
|
||||
environments: Some(vec![pending.clone().into_request()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
.await?;
|
||||
|
||||
@@ -840,7 +840,7 @@ async fn astra_omits_disabled_executor_and_plugin_skills_from_model_context() ->
|
||||
let thread = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![first_environment]),
|
||||
environments: Some(vec![first_environment.into_request()]),
|
||||
thread_extension_init,
|
||||
disabled_plugin_ids: Some(vec!["plugin-skills".to_string()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
|
||||
@@ -34,7 +34,7 @@ async fn running_descendants_refresh_only_shared_thread_instructions(shared: boo
|
||||
let root = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
thread_instructions_provider: Some(provider.clone()),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
@@ -45,7 +45,7 @@ async fn running_descendants_refresh_only_shared_thread_instructions(shared: boo
|
||||
let child = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id: parent_id,
|
||||
depth,
|
||||
@@ -64,7 +64,7 @@ async fn running_descendants_refresh_only_shared_thread_instructions(shared: boo
|
||||
let other_root = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
thread_instructions_provider: Some(other_provider),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
@@ -94,7 +94,7 @@ async fn running_descendants_refresh_only_shared_thread_instructions(shared: boo
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
initial_history: history,
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
thread_instructions_provider: Some(provider.clone()),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
@@ -202,7 +202,7 @@ async fn guardian_tracks_shared_instruction_updates_in_running_descendants() ->
|
||||
let root = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
thread_instructions_provider: Some(provider),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
@@ -212,7 +212,7 @@ async fn guardian_tracks_shared_instruction_updates_in_running_descendants() ->
|
||||
descendant = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id: descendant.thread_id,
|
||||
depth,
|
||||
|
||||
@@ -60,7 +60,7 @@ async fn descriptions_share_space_without_hiding_late_namespaces() -> Result<()>
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
dynamic_tools,
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
.await?;
|
||||
|
||||
@@ -1122,7 +1122,7 @@ async fn opted_in_executor_provider_skips_host_discovery_but_injects_discovered_
|
||||
// Keep the trace fixture's legacy mode: paginated SQLite workers can close
|
||||
// spans through a different subscriber than this test's scoped collector.
|
||||
history_mode: Some(codex_protocol::protocol::ThreadHistoryMode::Legacy),
|
||||
environments: Some(vec![environment.clone()]),
|
||||
environments: Some(vec![environment.clone().into_request()]),
|
||||
thread_extension_init,
|
||||
..StartThreadOptions::new(executor_config)
|
||||
})
|
||||
@@ -1584,7 +1584,7 @@ async fn executor_skill_tool_reads_references_under_current_permissions(
|
||||
let thread = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![selection]),
|
||||
environments: Some(vec![selection.into_request()]),
|
||||
thread_extension_init,
|
||||
..StartThreadOptions::new(config)
|
||||
})
|
||||
@@ -3125,7 +3125,7 @@ async fn production_turn_keeps_rebalanced_catalogs_stable_after_compaction_and_r
|
||||
let cloud_thread = test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
environments: Some(vec![pending_selection.clone()]),
|
||||
environments: Some(vec![pending_selection.clone().into_request()]),
|
||||
thread_extension_init: thread_extension_init.clone(),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
@@ -3247,7 +3247,7 @@ async fn production_turn_keeps_rebalanced_catalogs_stable_after_compaction_and_r
|
||||
history: Arc::new(history.items),
|
||||
rollout_path: cloud_thread.session_configured.rollout_path.clone(),
|
||||
}),
|
||||
environments: Some(vec![pending_selection.clone()]),
|
||||
environments: Some(vec![pending_selection.clone().into_request()]),
|
||||
thread_extension_init,
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
//! Cancelling partially initialized sessions must release their persistent writers.
|
||||
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::Ordering;
|
||||
@@ -80,7 +81,13 @@ async fn cancelled_resume_releases_writer_while_mcp_startup_is_pending() -> Resu
|
||||
.context("local thread store")?;
|
||||
let resume_options = || StartThreadOptions {
|
||||
initial_history: history.clone(),
|
||||
environments: Some(environments.clone()),
|
||||
environments: Some(
|
||||
environments
|
||||
.clone()
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
};
|
||||
|
||||
@@ -149,7 +156,13 @@ async fn failed_resume_releases_writer_before_returning() -> Result<()> {
|
||||
config.mcp_servers.set(servers)?;
|
||||
let resume_options = || StartThreadOptions {
|
||||
initial_history: history.clone(),
|
||||
environments: Some(environments.clone()),
|
||||
environments: Some(
|
||||
environments
|
||||
.clone()
|
||||
.into_iter()
|
||||
.map(TurnEnvironmentSelection::into_request)
|
||||
.collect(),
|
||||
),
|
||||
..StartThreadOptions::new(config.clone())
|
||||
};
|
||||
|
||||
|
||||
@@ -130,7 +130,7 @@ async fn fresh_v2_subagent_can_call_parent_dynamic_tool_when_enabled() -> Result
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
dynamic_tools: vec![DynamicToolSpec::Function(tool.clone())],
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
})
|
||||
.await?;
|
||||
|
||||
@@ -81,7 +81,7 @@ async fn turn_extension_data_is_captured_for_automatic_turns() -> anyhow::Result
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
turn_extension_init: initial,
|
||||
environments: Some(vec![test.executor_environment().selection().clone()]),
|
||||
environments: Some(vec![test.executor_environment().request()]),
|
||||
..StartThreadOptions::new(config)
|
||||
})
|
||||
.await?
|
||||
@@ -280,8 +280,7 @@ async fn host_drain_allows_spawned_agent_input_but_not_automatic_work() -> anyho
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
})),
|
||||
environments: Some(test.codex.environment_selections().await),
|
||||
..StartThreadOptions::new(test.config.clone())
|
||||
..test.start_thread_options().await
|
||||
})
|
||||
.await?
|
||||
.thread;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
//! Environment attachment authority, including the shared Full Access decision.
|
||||
//! Environment requests, thread selections, and attachment authority.
|
||||
//! Input APIs accept requests; runtime code uses selections constructed by the thread.
|
||||
|
||||
use crate::capabilities::SelectedCapabilityRoot;
|
||||
use crate::config_types::ShellEnvironmentPolicy;
|
||||
@@ -10,8 +11,74 @@ use crate::protocol::AskForApproval;
|
||||
use crate::sandbox::SandboxType;
|
||||
use codex_execpolicy::RequirementsExecPolicy;
|
||||
use codex_network_proxy::EnvironmentNetworkPolicy;
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
use codex_utils_path_uri::PathUri;
|
||||
|
||||
/// An environment requested by a caller, before construction of thread-owned state.
|
||||
///
|
||||
/// Input APIs accept requests so the receiving thread has a construction boundary for
|
||||
/// attaching its state before passing a selection to runtime code.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct TurnEnvironmentRequest {
|
||||
pub environment_id: String,
|
||||
pub cwd: PathUri,
|
||||
pub workspace_roots: Vec<PathUri>,
|
||||
pub config: EnvironmentConfigState,
|
||||
}
|
||||
|
||||
/// An environment selected by a thread for runtime use.
|
||||
/// Constructed from caller input at the receiving thread's startup or settings boundary.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct TurnEnvironmentSelection {
|
||||
pub environment_id: String,
|
||||
pub cwd: PathUri,
|
||||
pub workspace_roots: Vec<PathUri>,
|
||||
pub config: EnvironmentConfigState,
|
||||
}
|
||||
|
||||
impl TurnEnvironmentSelection {
|
||||
/// Constructs runtime state at the receiving thread boundary.
|
||||
pub fn new(request: TurnEnvironmentRequest) -> Self {
|
||||
Self {
|
||||
environment_id: request.environment_id,
|
||||
cwd: request.cwd,
|
||||
workspace_roots: request.workspace_roots,
|
||||
config: request.config,
|
||||
}
|
||||
}
|
||||
|
||||
/// Requests this environment again; the receiving thread supplies its own roots.
|
||||
/// This explicit conversion discards thread-owned state. Test inputs should instead
|
||||
/// construct requests directly so they do not depend on runtime selection fields.
|
||||
pub fn into_request(self) -> TurnEnvironmentRequest {
|
||||
TurnEnvironmentRequest {
|
||||
environment_id: self.environment_id,
|
||||
cwd: self.cwd,
|
||||
workspace_roots: self.workspace_roots,
|
||||
config: self.config,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The environments captured by a thread and its fallback working directory.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct TurnEnvironmentSelections {
|
||||
pub legacy_fallback_cwd: AbsolutePathBuf,
|
||||
pub environments: Vec<TurnEnvironmentSelection>,
|
||||
}
|
||||
|
||||
impl TurnEnvironmentSelections {
|
||||
pub fn new(
|
||||
legacy_fallback_cwd: AbsolutePathBuf,
|
||||
environments: Vec<TurnEnvironmentSelection>,
|
||||
) -> Self {
|
||||
Self {
|
||||
legacy_fallback_cwd,
|
||||
environments,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Configuration supplied for a thread's selected environment.
|
||||
#[allow(clippy::large_enum_variant)]
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
|
||||
@@ -99,6 +99,9 @@ pub use crate::approvals::NetworkPolicyAmendment;
|
||||
pub use crate::approvals::NetworkPolicyRuleAction;
|
||||
pub use crate::environment::EnvironmentConfig;
|
||||
pub use crate::environment::EnvironmentConfigState;
|
||||
pub use crate::environment::TurnEnvironmentRequest;
|
||||
pub use crate::environment::TurnEnvironmentSelection;
|
||||
pub use crate::environment::TurnEnvironmentSelections;
|
||||
pub use crate::environment::has_full_access;
|
||||
pub use crate::legacy_events::HasLegacyEvent;
|
||||
pub use crate::permissions::FileSystemAccessMode;
|
||||
@@ -149,34 +152,6 @@ pub fn strip_user_message_prefix(text: &str) -> &str {
|
||||
}
|
||||
}
|
||||
|
||||
// TODO(anp): Replace `TurnEnvironmentSelection` with `PathUri` once path URIs carry environment
|
||||
// identifiers.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct TurnEnvironmentSelection {
|
||||
pub environment_id: String,
|
||||
pub cwd: PathUri,
|
||||
pub workspace_roots: Vec<PathUri>,
|
||||
pub config: EnvironmentConfigState,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct TurnEnvironmentSelections {
|
||||
pub legacy_fallback_cwd: AbsolutePathBuf,
|
||||
pub environments: Vec<TurnEnvironmentSelection>,
|
||||
}
|
||||
|
||||
impl TurnEnvironmentSelections {
|
||||
pub fn new(
|
||||
legacy_fallback_cwd: AbsolutePathBuf,
|
||||
environments: Vec<TurnEnvironmentSelection>,
|
||||
) -> Self {
|
||||
Self {
|
||||
legacy_fallback_cwd,
|
||||
environments,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema, TS)]
|
||||
#[serde(transparent)]
|
||||
#[ts(type = "string")]
|
||||
@@ -513,7 +488,7 @@ pub struct ThreadSettingsOverrides {
|
||||
pub environments: Option<TurnEnvironmentSelections>,
|
||||
|
||||
/// Updated top-level runtime workspace roots for default environments.
|
||||
/// Explicit environment selections own their roots separately.
|
||||
/// Explicit environment requests own their workspace roots separately.
|
||||
pub runtime_workspace_roots: Option<Vec<AbsolutePathBuf>>,
|
||||
|
||||
/// Updated profile-defined workspace roots for status summaries and
|
||||
|
||||
Reference in New Issue
Block a user