mirror of
https://github.com/openai/codex.git
synced 2026-11-05 20:10:33 +00:00
Adam Perry @ OpenAI
7e8143a897
## Why Removing an executor must detach its selection even when older turn snapshots still reference it. Connection events queued for a removed attachment must not leak into a later selection of the same executor. ## What changed - Introduce `EnvironmentAttachment` with a detach signal shared by turn snapshots and a guard owned by the selection. - Preserve attachments across workspace and configuration changes, while giving reselections and child threads independent attachment lifetimes. - Stop forwarding connection events on detach and discard buffered events from detached attachments. - Activate initial attachments after `SessionConfigured`, using configured bindings for startup discovery. ## Testing Add coverage for discarding buffered connection events on reselection and extend snapshot tests to verify immediate detachment, fresh attachments on reselection, and independent parent and child lifetimes. GitOrigin-RevId: 6cf882e0d17a296030c13d80c6e2866c80e04638
2699 lines
108 KiB
Rust
2699 lines
108 KiB
Rust
use std::collections::HashMap;
|
|
use std::collections::HashSet;
|
|
use std::fmt;
|
|
use std::future::Future;
|
|
use std::sync::Arc;
|
|
use std::sync::Mutex;
|
|
use std::sync::PoisonError;
|
|
|
|
use async_channel::Receiver;
|
|
use async_channel::Sender;
|
|
use codex_exec_server::Environment;
|
|
use codex_exec_server::EnvironmentInfo;
|
|
use codex_exec_server::EnvironmentManager;
|
|
use codex_exec_server::ExecServerError;
|
|
use codex_exec_server::SelectedCapabilityRootsStatus;
|
|
use codex_protocol::capabilities::CapabilityRootLocation;
|
|
use codex_protocol::capabilities::EnvironmentCapabilityRoots;
|
|
use codex_protocol::capabilities::SelectedCapabilityRoot;
|
|
use codex_protocol::error::CodexErr;
|
|
use codex_protocol::error::Result as CodexResult;
|
|
use codex_protocol::models::PermissionProfile;
|
|
use codex_protocol::protocol::AskForApproval;
|
|
use codex_protocol::protocol::EnvironmentConfig;
|
|
use codex_protocol::protocol::EnvironmentConfigState;
|
|
use codex_protocol::protocol::Event;
|
|
use codex_protocol::protocol::TurnEnvironmentRequest;
|
|
use codex_protocol::protocol::TurnEnvironmentSelection;
|
|
use codex_skills_extension::EnvironmentSkillRequirements;
|
|
use codex_utils_absolute_path::AbsolutePathBuf;
|
|
use codex_utils_path_uri::PathUri;
|
|
use futures::FutureExt;
|
|
use futures::StreamExt;
|
|
use futures::future::BoxFuture;
|
|
use futures::future::Shared;
|
|
use futures::stream::BoxStream;
|
|
use tokio::sync::watch;
|
|
use tokio_util::sync::CancellationToken;
|
|
use tokio_util::sync::DropGuard;
|
|
use tokio_util::task::AbortOnDropHandle;
|
|
use tracing::Instrument;
|
|
use tracing::instrument::WithSubscriber;
|
|
|
|
use crate::session::ThreadEnvironmentDefaults;
|
|
use crate::session::turn_context::ShellSnapshotCache;
|
|
use crate::session::turn_context::ShellSnapshotTask;
|
|
use crate::session::turn_context::TurnEnvironment;
|
|
use crate::shell::Shell;
|
|
use crate::shell_snapshot::ShellSnapshot;
|
|
use crate::shell_snapshot::SnapshotCredentialBrokerState;
|
|
|
|
mod attachment;
|
|
pub(crate) use attachment::EnvironmentAttachment;
|
|
|
|
// Reject pathological selected cwd values at the environment-selection boundary.
|
|
const MAX_TURN_ENVIRONMENT_CWD_BYTES: usize = 8 * 1024;
|
|
|
|
/// Records whether a normalized config should follow later thread setting updates.
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
pub(crate) enum EnvironmentConfigOrigin {
|
|
Thread,
|
|
Owner,
|
|
}
|
|
|
|
impl EnvironmentConfigOrigin {
|
|
/// Reconstructs the input form so another attachment boundary preserves config ownership.
|
|
pub(crate) fn into_input_selection(
|
|
self,
|
|
mut selection: TurnEnvironmentSelection,
|
|
) -> TurnEnvironmentSelection {
|
|
if self == Self::Thread {
|
|
selection.config = EnvironmentConfigState::FromThread;
|
|
}
|
|
selection
|
|
}
|
|
|
|
pub(crate) fn selected_capability_roots(
|
|
self,
|
|
environment: &Environment,
|
|
config: &EnvironmentConfig,
|
|
) -> Vec<SelectedCapabilityRoot> {
|
|
match self {
|
|
Self::Thread => environment.selected_capability_roots(),
|
|
Self::Owner => config.selected_capability_roots.clone(),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Combines persisted thread roots with attachment roots in selection order.
|
|
///
|
|
/// A matching root from the current attachment refreshes its persisted location. Root IDs from
|
|
/// other environments and persisted roots absent from the attachment keep their existing location.
|
|
pub(crate) fn combine_selected_capability_roots(
|
|
thread_roots: &[SelectedCapabilityRoot],
|
|
sources: impl IntoIterator<Item = Vec<SelectedCapabilityRoot>>,
|
|
) -> Vec<SelectedCapabilityRoot> {
|
|
let attachment_roots = sources.into_iter().flatten().collect::<Vec<_>>();
|
|
let mut combined_roots = thread_roots
|
|
.iter()
|
|
.map(|thread_root| {
|
|
let CapabilityRootLocation::Environment {
|
|
environment_id: thread_environment_id,
|
|
..
|
|
} = &thread_root.location;
|
|
attachment_roots
|
|
.iter()
|
|
.find(|attachment_root| {
|
|
if attachment_root.id != thread_root.id {
|
|
return false;
|
|
}
|
|
let CapabilityRootLocation::Environment {
|
|
environment_id: attachment_environment_id,
|
|
..
|
|
} = &attachment_root.location;
|
|
attachment_environment_id == thread_environment_id
|
|
})
|
|
.unwrap_or(thread_root)
|
|
.clone()
|
|
})
|
|
.collect::<Vec<_>>();
|
|
combined_roots.extend(attachment_roots);
|
|
combined_roots
|
|
}
|
|
|
|
pub(crate) fn default_thread_environment_requests(
|
|
environment_manager: &EnvironmentManager,
|
|
cwd: &AbsolutePathBuf,
|
|
workspace_roots: &[AbsolutePathBuf],
|
|
) -> Vec<TurnEnvironmentRequest> {
|
|
environment_manager
|
|
.default_environment_ids()
|
|
.into_iter()
|
|
.map(|environment_id| TurnEnvironmentRequest {
|
|
environment_id,
|
|
cwd: PathUri::from_abs_path(cwd),
|
|
workspace_roots: workspace_roots.iter().map(PathUri::from_abs_path).collect(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Checks that environment IDs are registered and unique and that working directories are not too long.
|
|
pub fn validate_environment_ids_and_cwds<'a>(
|
|
environment_manager: &EnvironmentManager,
|
|
environments: impl IntoIterator<Item = (&'a str, &'a PathUri)>,
|
|
) -> CodexResult<()> {
|
|
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_id) {
|
|
return Err(CodexErr::InvalidRequest(format!(
|
|
"duplicate turn environment id `{environment_id}`"
|
|
)));
|
|
}
|
|
environment_manager
|
|
.get_environment(environment_id)
|
|
.ok_or_else(|| {
|
|
CodexErr::InvalidRequest(format!("unknown turn environment id `{environment_id}`"))
|
|
})?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
type TurnEnvironmentResult = Result<ResolvedEnvironment, Arc<ExecServerError>>;
|
|
type TurnEnvironmentResolution = Shared<BoxFuture<'static, TurnEnvironmentResult>>;
|
|
type PendingConfigurationResult = Result<EnvironmentConfig, String>;
|
|
pub(crate) type PendingConfiguration = Shared<BoxFuture<'static, PendingConfigurationResult>>;
|
|
|
|
fn pending_configuration(
|
|
completion: &watch::Sender<Option<PendingConfigurationResult>>,
|
|
) -> PendingConfiguration {
|
|
let mut receiver = completion.subscribe();
|
|
async move {
|
|
receiver
|
|
.wait_for(Option::is_some)
|
|
.await
|
|
.ok()
|
|
.and_then(|result| result.clone())
|
|
.unwrap_or_else(|| Err("environment configuration was canceled".to_string()))
|
|
}
|
|
.boxed()
|
|
.shared()
|
|
}
|
|
|
|
// Shared startup result used to build each turn's environment with its own config
|
|
// without restarting the connection or shell resolution.
|
|
#[derive(Clone)]
|
|
struct ResolvedEnvironment {
|
|
shell: Option<Shell>,
|
|
user_home_dir: Option<PathUri>,
|
|
executor_platform_os: Option<String>,
|
|
temporary_directories: Option<Vec<PathUri>>,
|
|
shell_snapshot: ShellSnapshotTask,
|
|
shell_snapshot_builder: ShellSnapshot,
|
|
shell_snapshot_cache: ShellSnapshotCache,
|
|
shell_snapshot_v2_supported: bool,
|
|
installed_config: Option<EnvironmentConfig>,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct SelectedTurnEnvironment {
|
|
selection: TurnEnvironmentSelection,
|
|
config_origin: EnvironmentConfigOrigin,
|
|
attachment: Arc<EnvironmentAttachment>,
|
|
// Only selections share this guard. A captured turn must not postpone detach.
|
|
detach_on_drop: Arc<DropGuard>,
|
|
resolution: TurnEnvironmentResolution,
|
|
// Executor retries keep this session's wait for its first accepted configuration.
|
|
pending_completion: Option<watch::Sender<Option<PendingConfigurationResult>>>,
|
|
// Inherited owner only; otherwise snapshots expose this attachment's own completion.
|
|
owner_config_result: Option<PendingConfiguration>,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub(crate) struct StartingTurnEnvironment {
|
|
pub(crate) selection: TurnEnvironmentSelection,
|
|
config_origin: EnvironmentConfigOrigin,
|
|
resolution: TurnEnvironmentResolution,
|
|
attachment: Arc<EnvironmentAttachment>,
|
|
owner_config_result: Option<PendingConfiguration>,
|
|
}
|
|
|
|
/// Uses the current thread defaults when requested and records who owns later config updates.
|
|
fn resolve_selection_config(
|
|
mut selection: TurnEnvironmentSelection,
|
|
defaults: &ThreadEnvironmentDefaults,
|
|
) -> (TurnEnvironmentSelection, EnvironmentConfigOrigin) {
|
|
let (config, origin) = match selection.config {
|
|
EnvironmentConfigState::FromThread => (
|
|
EnvironmentConfigState::Ready(defaults.for_selection(&selection)),
|
|
EnvironmentConfigOrigin::Thread,
|
|
),
|
|
config @ (EnvironmentConfigState::Ready(_)
|
|
| EnvironmentConfigState::Pending
|
|
| EnvironmentConfigState::Failed(_)) => (config, EnvironmentConfigOrigin::Owner),
|
|
};
|
|
selection.config = config;
|
|
(selection, origin)
|
|
}
|
|
|
|
impl SelectedTurnEnvironment {
|
|
// Pending configuration belongs to each attachment set; completed startup can be shared.
|
|
fn for_turn(
|
|
&self,
|
|
attachment: Arc<EnvironmentAttachment>,
|
|
detach_on_drop: Arc<DropGuard>,
|
|
local_shell: &Shell,
|
|
shell_snapshot: &ShellSnapshot,
|
|
) -> Self {
|
|
if matches!(self.selection.config, EnvironmentConfigState::Pending) {
|
|
return ThreadEnvironments::start_environment(
|
|
self.selection.clone(),
|
|
self.config_origin,
|
|
attachment,
|
|
detach_on_drop,
|
|
self.owner_config_result.clone(),
|
|
/*pending_completion*/ None,
|
|
local_shell,
|
|
shell_snapshot,
|
|
);
|
|
}
|
|
Self {
|
|
attachment,
|
|
detach_on_drop,
|
|
..self.clone()
|
|
}
|
|
}
|
|
|
|
fn refresh_thread_config(&mut self, defaults: &ThreadEnvironmentDefaults) {
|
|
if self.config_origin == EnvironmentConfigOrigin::Thread {
|
|
self.selection.config =
|
|
EnvironmentConfigState::Ready(defaults.for_selection(&self.selection));
|
|
}
|
|
}
|
|
}
|
|
|
|
impl fmt::Debug for StartingTurnEnvironment {
|
|
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
formatter
|
|
.debug_struct("StartingTurnEnvironment")
|
|
.field("selection", &self.selection)
|
|
.field("resolved", &self.resolution.peek().is_some())
|
|
.finish_non_exhaustive()
|
|
}
|
|
}
|
|
|
|
impl StartingTurnEnvironment {
|
|
pub(crate) fn owner_configuration(&self) -> Option<PendingConfiguration> {
|
|
if matches!(self.selection.config, EnvironmentConfigState::Pending) {
|
|
self.owner_config_result.clone()
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
#[tracing::instrument(
|
|
name = "environments.wait_until_ready",
|
|
skip_all,
|
|
fields(environment_id = %self.selection.environment_id)
|
|
)]
|
|
pub(crate) async fn wait_until_ready(&self) -> Result<(), Arc<ExecServerError>> {
|
|
self.resolution.clone().await.map(|_| ())
|
|
}
|
|
}
|
|
|
|
pub(crate) struct ThreadEnvironments {
|
|
environment_manager: Arc<EnvironmentManager>,
|
|
local_shell: Shell,
|
|
shell_snapshot: ShellSnapshot,
|
|
non_blocking_snapshots: bool,
|
|
state: Mutex<ThreadEnvironmentsState>,
|
|
// Retain configured bindings independently of active-turn membership and cancellation.
|
|
configured: Mutex<ThreadEnvironmentsState>,
|
|
}
|
|
|
|
struct ThreadEnvironmentsState {
|
|
thread_defaults: ThreadEnvironmentDefaults,
|
|
environments: Vec<SelectedTurnEnvironment>,
|
|
connection_events: Option<ConnectionEventStreams>,
|
|
}
|
|
|
|
/// Forwards the active attachments' streams to their thread's event channel.
|
|
struct ConnectionEventStreams {
|
|
streams: Sender<BoxStream<'static, Event>>,
|
|
_task: AbortOnDropHandle<()>,
|
|
}
|
|
|
|
impl ConnectionEventStreams {
|
|
fn new(tx_event: Sender<Event>) -> Self {
|
|
let (streams, incoming) = async_channel::unbounded::<BoxStream<'static, Event>>();
|
|
let task = tokio::spawn(async move {
|
|
let events = incoming
|
|
.flatten_unordered(/*limit*/ None)
|
|
.take_until(tx_event.closed());
|
|
tokio::pin!(events);
|
|
while let Some(event) = events.next().await {
|
|
if tx_event.send(event).await.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
});
|
|
Self {
|
|
streams,
|
|
_task: AbortOnDropHandle::new(task),
|
|
}
|
|
}
|
|
|
|
fn add(&self, events: Option<Receiver<Event>>, detached: &CancellationToken) {
|
|
if let Some(events) = events {
|
|
// A closed receiver can still yield events queued before detach.
|
|
let events = events.take_until(detached.clone().cancelled_owned());
|
|
let _ = self.streams.try_send(events.boxed());
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Copy)]
|
|
enum StartupRetry<'a> {
|
|
AfterSelectionUpdate,
|
|
OnNewTurn(&'a [TurnEnvironmentSelection]),
|
|
}
|
|
|
|
impl StartupRetry<'_> {
|
|
fn should_restart(self, environment: &SelectedTurnEnvironment) -> bool {
|
|
if matches!(
|
|
environment.selection.config,
|
|
EnvironmentConfigState::Failed(_)
|
|
) {
|
|
return matches!(self, Self::AfterSelectionUpdate);
|
|
}
|
|
// Avoid cloning a resolved environment on every healthy turn.
|
|
if environment.resolution.peek().is_some_and(Result::is_ok) {
|
|
return false;
|
|
}
|
|
match environment.resolution.clone().now_or_never() {
|
|
Some(Err(error)) => match self {
|
|
Self::AfterSelectionUpdate => true,
|
|
Self::OnNewTurn(selected) => {
|
|
error.is_retryable_preparation_error()
|
|
&& selected
|
|
.iter()
|
|
.any(|s| s.has_same_workspace(&environment.selection))
|
|
}
|
|
},
|
|
Some(Ok(_)) | None => false,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl ThreadEnvironments {
|
|
pub(crate) fn new(
|
|
environment_manager: Arc<EnvironmentManager>,
|
|
local_shell: Shell,
|
|
thread_defaults: ThreadEnvironmentDefaults,
|
|
shell_snapshot: ShellSnapshot,
|
|
current: TurnEnvironmentSnapshot,
|
|
non_blocking_snapshots: bool,
|
|
) -> Self {
|
|
let environments: Vec<_> = current
|
|
.environments
|
|
.into_iter()
|
|
.filter_map(|inherited| {
|
|
// Reuse completed startup without resolving a Pending owner's configuration.
|
|
let inherited = if let TurnEnvironmentState::Starting(starting) = &inherited
|
|
&& let EnvironmentConfigState::Ready(_) = &starting.selection.config
|
|
&& let Some(Ok(resolved)) = starting.resolution.clone().now_or_never()
|
|
{
|
|
TurnEnvironmentState::from_resolution(starting.clone(), Some(Ok(resolved)))
|
|
} else {
|
|
inherited
|
|
};
|
|
let environment = match inherited {
|
|
TurnEnvironmentState::Ready(environment) => environment,
|
|
TurnEnvironmentState::Starting(inherited) => {
|
|
let (selection, config_origin) = resolve_selection_config(
|
|
inherited
|
|
.config_origin
|
|
.into_input_selection(inherited.selection.clone()),
|
|
&thread_defaults,
|
|
);
|
|
let (attachment, detach_on_drop, _) = EnvironmentAttachment::new(
|
|
&selection.environment_id,
|
|
Arc::clone(&inherited.attachment.environment),
|
|
);
|
|
return Some(Self::start_environment(
|
|
selection,
|
|
config_origin,
|
|
attachment,
|
|
Arc::new(detach_on_drop),
|
|
inherited.owner_configuration(),
|
|
/*pending_completion*/ None,
|
|
&local_shell,
|
|
&shell_snapshot,
|
|
));
|
|
}
|
|
TurnEnvironmentState::Failed { .. } => return None,
|
|
};
|
|
let selection = environment.selection;
|
|
let config_origin = environment.config_origin;
|
|
let selected_environment = &environment.attachment.environment;
|
|
let inherited_snapshot = if !selected_environment.is_remote()
|
|
&& shell_snapshot.should_rebuild_inherited()
|
|
{
|
|
futures::future::ready(None).boxed().shared()
|
|
} else {
|
|
environment.shell_snapshot
|
|
};
|
|
let resolution: TurnEnvironmentResolution =
|
|
futures::future::ready(Ok(ResolvedEnvironment {
|
|
shell: environment.shell,
|
|
user_home_dir: environment.user_home_dir,
|
|
executor_platform_os: environment.executor_platform_os,
|
|
temporary_directories: environment.temporary_directories,
|
|
shell_snapshot: inherited_snapshot,
|
|
shell_snapshot_builder: shell_snapshot.clone(),
|
|
shell_snapshot_cache: Arc::default(),
|
|
shell_snapshot_v2_supported: environment.shell_snapshot_v2_supported
|
|
&& (selected_environment.is_remote()
|
|
|| !shell_snapshot.should_rebuild_inherited()),
|
|
installed_config: None,
|
|
}))
|
|
.boxed()
|
|
.shared();
|
|
let (attachment, detach_on_drop, _) = EnvironmentAttachment::new(
|
|
&selection.environment_id,
|
|
Arc::clone(selected_environment),
|
|
);
|
|
let mut inherited_environment = SelectedTurnEnvironment {
|
|
selection,
|
|
config_origin,
|
|
attachment,
|
|
detach_on_drop: Arc::new(detach_on_drop),
|
|
resolution,
|
|
pending_completion: None,
|
|
owner_config_result: None,
|
|
};
|
|
// Child threads re-infer thread-owned config while preserving owner config.
|
|
inherited_environment.refresh_thread_config(&thread_defaults);
|
|
Some(inherited_environment)
|
|
})
|
|
.collect();
|
|
Self {
|
|
environment_manager,
|
|
local_shell,
|
|
shell_snapshot,
|
|
non_blocking_snapshots,
|
|
state: Mutex::new(ThreadEnvironmentsState {
|
|
thread_defaults: thread_defaults.clone(),
|
|
environments: Vec::new(),
|
|
connection_events: None,
|
|
}),
|
|
configured: Mutex::new(ThreadEnvironmentsState {
|
|
thread_defaults,
|
|
environments,
|
|
connection_events: None,
|
|
}),
|
|
}
|
|
}
|
|
|
|
/// Attach the initial active selection after `SessionConfigured` so clients see only later
|
|
/// connection changes. Startup discovery uses the configured bindings until this point.
|
|
pub(crate) fn activate_configured(&self, tx_event: Sender<Event>) {
|
|
let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
|
|
let configured = self
|
|
.configured
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner);
|
|
let connection_events = ConnectionEventStreams::new(tx_event);
|
|
state.environments = configured
|
|
.environments
|
|
.iter()
|
|
.map(|environment| {
|
|
let (attachment, detach_on_drop, events) = EnvironmentAttachment::new(
|
|
&environment.selection.environment_id,
|
|
Arc::clone(&environment.attachment.environment),
|
|
);
|
|
connection_events.add(events, &attachment.detached);
|
|
environment.for_turn(
|
|
attachment,
|
|
Arc::new(detach_on_drop),
|
|
&self.local_shell,
|
|
&self.shell_snapshot,
|
|
)
|
|
})
|
|
.collect();
|
|
state.connection_events = Some(connection_events);
|
|
}
|
|
|
|
fn start_shell_snapshot_task(
|
|
shell_snapshot: ShellSnapshot,
|
|
environment: Arc<Environment>,
|
|
cwd: PathUri,
|
|
shell: Option<Shell>,
|
|
config: &EnvironmentConfigState,
|
|
) -> ShellSnapshotTask {
|
|
// Protected snapshots require the command's sandbox and are captured lazily.
|
|
if shell_snapshot.should_rebuild_inherited() {
|
|
return futures::future::ready(None).boxed().shared();
|
|
}
|
|
let EnvironmentConfigState::Ready(config) = config else {
|
|
return futures::future::ready(None).boxed().shared();
|
|
};
|
|
let shell_snapshot = shell_snapshot
|
|
.build(
|
|
environment,
|
|
cwd,
|
|
shell,
|
|
/*allow_login_shell*/ true,
|
|
config.shell_environment_policy.clone(),
|
|
/*sandbox*/ None,
|
|
)
|
|
.boxed()
|
|
.shared();
|
|
drop(tokio::spawn(
|
|
shell_snapshot
|
|
.clone()
|
|
.in_current_span()
|
|
.with_current_subscriber(),
|
|
));
|
|
shell_snapshot
|
|
}
|
|
|
|
/// Updates the selected list before waking work that was waiting on the previous selection.
|
|
pub(crate) fn update_selections(&self, environments: &[TurnEnvironmentSelection]) {
|
|
self.apply_selections(environments, StartupRetry::AfterSelectionUpdate);
|
|
}
|
|
|
|
/// A new turn can retry a failed connection even when it keeps the same workspace.
|
|
pub(crate) fn activate_selections(&self, environments: &[TurnEnvironmentSelection]) {
|
|
let retry = StartupRetry::OnNewTurn(environments);
|
|
if environments != self.selections() {
|
|
self.update_selections(environments);
|
|
} else if self
|
|
.state
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner)
|
|
.environments
|
|
.iter()
|
|
.any(|environment| retry.should_restart(environment))
|
|
{
|
|
self.apply_selections(environments, retry);
|
|
}
|
|
}
|
|
|
|
fn apply_selections(&self, environments: &[TurnEnvironmentSelection], retry: StartupRetry<'_>) {
|
|
{
|
|
let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
|
|
let configured = self
|
|
.configured
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner);
|
|
self.update_selections_with_bindings(
|
|
&mut state,
|
|
environments,
|
|
&configured.environments,
|
|
retry,
|
|
);
|
|
}
|
|
// Retry configured bindings with their own settings, keeping turn-only changes temporary.
|
|
self.apply_configured_selections(&self.configured_snapshot_now().all_selections(), retry);
|
|
}
|
|
|
|
/// Retains configured bindings without changing the active turn's attachments.
|
|
pub(crate) fn update_configured_selections(&self, environments: &[TurnEnvironmentSelection]) {
|
|
self.apply_configured_selections(environments, StartupRetry::AfterSelectionUpdate);
|
|
}
|
|
|
|
fn apply_configured_selections(
|
|
&self,
|
|
environments: &[TurnEnvironmentSelection],
|
|
retry: StartupRetry<'_>,
|
|
) {
|
|
let mut configured = self
|
|
.configured
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner);
|
|
self.update_selections_with_bindings(&mut configured, environments, &[], retry);
|
|
}
|
|
|
|
fn update_selections_with_bindings(
|
|
&self,
|
|
state: &mut ThreadEnvironmentsState,
|
|
environments: &[TurnEnvironmentSelection],
|
|
bindings: &[SelectedTurnEnvironment],
|
|
retry: StartupRetry<'_>,
|
|
) {
|
|
let ThreadEnvironmentsState {
|
|
thread_defaults,
|
|
environments: current,
|
|
connection_events,
|
|
} = state;
|
|
let connection_events = connection_events.as_ref();
|
|
let mut seen_environment_ids = HashSet::with_capacity(environments.len());
|
|
let mut next = Vec::with_capacity(environments.len());
|
|
let mut pending_completions = Vec::new();
|
|
// Workspace and configuration changes preserve the attachment to the same executor.
|
|
let attachment_for =
|
|
|environment_id: &str,
|
|
environment: &Arc<Environment>,
|
|
connection_events: Option<&ConnectionEventStreams>| {
|
|
current
|
|
.iter()
|
|
.find(|previous| {
|
|
previous.selection.environment_id == environment_id
|
|
&& Arc::ptr_eq(&previous.attachment.environment, environment)
|
|
})
|
|
.map(|previous| {
|
|
(
|
|
Arc::clone(&previous.attachment),
|
|
Arc::clone(&previous.detach_on_drop),
|
|
)
|
|
})
|
|
.unwrap_or_else(|| {
|
|
let (attachment, guard, events) =
|
|
EnvironmentAttachment::new(environment_id, Arc::clone(environment));
|
|
if let Some(connection_events) = connection_events {
|
|
connection_events.add(events, &attachment.detached);
|
|
}
|
|
(attachment, Arc::new(guard))
|
|
})
|
|
};
|
|
for selected_environment in environments {
|
|
if !seen_environment_ids.insert(selected_environment.environment_id.as_str()) {
|
|
continue;
|
|
}
|
|
let current_environment = current.iter().find(|environment| {
|
|
environment
|
|
.selection
|
|
.has_same_workspace(selected_environment)
|
|
});
|
|
let previous = current_environment.or_else(|| {
|
|
bindings.iter().find(|environment| {
|
|
environment
|
|
.selection
|
|
.has_same_workspace(selected_environment)
|
|
})
|
|
});
|
|
let (selected_environment, config_origin) =
|
|
resolve_selection_config(selected_environment.clone(), thread_defaults);
|
|
if let Some(environment) = previous {
|
|
let failed = retry.should_restart(environment);
|
|
let restarting_as_pending =
|
|
matches!(selected_environment.config, EnvironmentConfigState::Pending)
|
|
&& !matches!(
|
|
environment.selection.config,
|
|
EnvironmentConfigState::Pending
|
|
);
|
|
let reuse_startup = !failed && !restarting_as_pending;
|
|
let mut environment = if current_environment.is_some() {
|
|
environment.clone()
|
|
} else {
|
|
let (attachment, detach_on_drop) = attachment_for(
|
|
&selected_environment.environment_id,
|
|
&environment.attachment.environment,
|
|
connection_events.filter(|_| reuse_startup),
|
|
);
|
|
environment.for_turn(
|
|
attachment,
|
|
detach_on_drop,
|
|
&self.local_shell,
|
|
&self.shell_snapshot,
|
|
)
|
|
};
|
|
if let Some(completion) = environment.pending_completion.take() {
|
|
match &selected_environment.config {
|
|
EnvironmentConfigState::Ready(config) => {
|
|
pending_completions.push((completion, Ok(config.clone())));
|
|
}
|
|
EnvironmentConfigState::Failed(error) => {
|
|
pending_completions.push((completion, Err(error.clone())));
|
|
}
|
|
EnvironmentConfigState::FromThread | EnvironmentConfigState::Pending => {
|
|
environment.pending_completion = Some(completion);
|
|
}
|
|
}
|
|
}
|
|
if reuse_startup {
|
|
let shell_settings_changed = matches!(
|
|
(&environment.selection.config, &selected_environment.config),
|
|
(EnvironmentConfigState::Ready(previous), EnvironmentConfigState::Ready(current))
|
|
if previous.allow_login_shell != current.allow_login_shell
|
|
|| previous.shell_environment_policy != current.shell_environment_policy
|
|
);
|
|
environment.selection = selected_environment;
|
|
environment.config_origin = config_origin;
|
|
if shell_settings_changed
|
|
&& !environment.attachment.environment.is_remote()
|
|
&& self.shell_snapshot.should_rebuild_inherited()
|
|
&& let Some(Ok(resolved)) = environment.resolution.clone().now_or_never()
|
|
{
|
|
self.restart_shell_snapshot(&mut environment, resolved);
|
|
}
|
|
next.push(environment);
|
|
continue;
|
|
}
|
|
}
|
|
|
|
let environment_id = &selected_environment.environment_id;
|
|
let Some(environment) = self.environment_manager.get_environment(environment_id) else {
|
|
tracing::warn!("skipping unknown turn environment `{environment_id}`");
|
|
continue;
|
|
};
|
|
let (attachment, detach_on_drop) =
|
|
attachment_for(environment_id, &environment, connection_events);
|
|
let pending = previous.filter(|previous| {
|
|
matches!(selected_environment.config, EnvironmentConfigState::Pending)
|
|
&& previous.pending_completion.is_some()
|
|
});
|
|
let selected = Self::start_environment(
|
|
selected_environment,
|
|
config_origin,
|
|
attachment,
|
|
detach_on_drop,
|
|
pending.and_then(|previous| previous.owner_config_result.clone()),
|
|
current_environment.and_then(|previous| previous.pending_completion.clone()),
|
|
&self.local_shell,
|
|
&self.shell_snapshot,
|
|
);
|
|
next.push(selected);
|
|
}
|
|
// Install the new selection before dropping the guards for removed attachments.
|
|
drop(std::mem::replace(current, next));
|
|
|
|
// Publish after installing the new list, including pending bindings replaced by retries.
|
|
for (completion, result) in pending_completions {
|
|
completion.send_if_modified(|current| {
|
|
if current.is_some() {
|
|
return false;
|
|
}
|
|
*current = Some(result);
|
|
true
|
|
});
|
|
}
|
|
}
|
|
|
|
#[allow(clippy::too_many_arguments)]
|
|
fn start_environment(
|
|
selection: TurnEnvironmentSelection,
|
|
config_origin: EnvironmentConfigOrigin,
|
|
attachment: Arc<EnvironmentAttachment>,
|
|
detach_on_drop: Arc<DropGuard>,
|
|
owner_config_result: Option<PendingConfiguration>,
|
|
pending_completion: Option<watch::Sender<Option<PendingConfigurationResult>>>,
|
|
local_shell: &Shell,
|
|
shell_snapshot: &ShellSnapshot,
|
|
) -> SelectedTurnEnvironment {
|
|
let (pending_completion, configuration_ready) =
|
|
if matches!(selection.config, EnvironmentConfigState::Pending) {
|
|
let sender =
|
|
pending_completion.unwrap_or_else(|| watch::channel(/*init*/ None).0);
|
|
let result = pending_configuration(&sender);
|
|
(Some(sender), Some(result))
|
|
} else {
|
|
(None, None)
|
|
};
|
|
// The resolver waits for this Session; descendants can follow the original owner.
|
|
let (task, resolution) = Self::resolve_environment(
|
|
selection.clone(),
|
|
Arc::clone(&attachment.environment),
|
|
local_shell.clone(),
|
|
shell_snapshot.clone(),
|
|
configuration_ready,
|
|
)
|
|
.remote_handle();
|
|
drop(tokio::spawn(
|
|
task.in_current_span().with_current_subscriber(),
|
|
));
|
|
SelectedTurnEnvironment {
|
|
selection,
|
|
config_origin,
|
|
attachment,
|
|
detach_on_drop,
|
|
resolution: resolution.boxed().shared(),
|
|
pending_completion,
|
|
owner_config_result,
|
|
}
|
|
}
|
|
|
|
/// Projects canonical selections back to caller input form without losing config ownership.
|
|
pub(crate) fn selections(&self) -> Vec<TurnEnvironmentSelection> {
|
|
self.state
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner)
|
|
.environments
|
|
.iter()
|
|
.map(|environment| {
|
|
environment
|
|
.config_origin
|
|
.into_input_selection(environment.selection.clone())
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Returns installed owner configuration without treating pending attachments as ready.
|
|
pub(crate) fn primary_config_for(
|
|
selections: &[TurnEnvironmentSelection],
|
|
) -> Option<&EnvironmentConfig> {
|
|
match &selections.first()?.config {
|
|
EnvironmentConfigState::Ready(config) => Some(config),
|
|
EnvironmentConfigState::FromThread
|
|
| EnvironmentConfigState::Pending
|
|
| EnvironmentConfigState::Failed(_) => None,
|
|
}
|
|
}
|
|
|
|
pub(crate) fn primary_workspace_roots_for(
|
|
selections: &[TurnEnvironmentSelection],
|
|
) -> Vec<AbsolutePathBuf> {
|
|
selections
|
|
.first()
|
|
.map(|selection| {
|
|
selection
|
|
.workspace_roots
|
|
.iter()
|
|
.filter_map(|workspace_root| workspace_root.to_abs_path().ok())
|
|
.collect()
|
|
})
|
|
.unwrap_or_default()
|
|
}
|
|
|
|
/// Changes the defaults inherited by all selected and later-added environments. A step that
|
|
/// already captured its environments keeps the previous values.
|
|
pub(crate) fn set_active_thread_defaults(&self, defaults: EnvironmentConfig) {
|
|
self.set_thread_defaults(&self.state, defaults);
|
|
}
|
|
|
|
pub(crate) fn set_configured_thread_defaults(&self, defaults: EnvironmentConfig) {
|
|
self.set_thread_defaults(&self.configured, defaults);
|
|
}
|
|
|
|
fn set_thread_defaults(
|
|
&self,
|
|
state: &Mutex<ThreadEnvironmentsState>,
|
|
defaults: EnvironmentConfig,
|
|
) {
|
|
let mut state = state.lock().unwrap_or_else(PoisonError::into_inner);
|
|
let ThreadEnvironmentsState {
|
|
thread_defaults,
|
|
environments,
|
|
..
|
|
} = &mut *state;
|
|
if thread_defaults.common == defaults {
|
|
return;
|
|
}
|
|
thread_defaults.common = defaults;
|
|
for environment in environments.iter_mut() {
|
|
let shell_settings_changed = environment.config_origin
|
|
== EnvironmentConfigOrigin::Thread
|
|
&& matches!(
|
|
&environment.selection.config,
|
|
EnvironmentConfigState::Ready(previous)
|
|
if previous.allow_login_shell != thread_defaults.common.allow_login_shell
|
|
|| previous.shell_environment_policy != thread_defaults.common.shell_environment_policy
|
|
);
|
|
environment.refresh_thread_config(thread_defaults);
|
|
if shell_settings_changed
|
|
&& !environment.attachment.environment.is_remote()
|
|
&& self.shell_snapshot.should_rebuild_inherited()
|
|
&& let Some(Ok(resolved)) = environment.resolution.clone().now_or_never()
|
|
{
|
|
self.restart_shell_snapshot(environment, resolved);
|
|
}
|
|
}
|
|
}
|
|
|
|
pub(crate) fn set_snapshot_credential_broker(&self, broker: SnapshotCredentialBrokerState) {
|
|
let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
|
|
let mut configured = self
|
|
.configured
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner);
|
|
if !self.shell_snapshot.set_credential_broker(broker) {
|
|
return;
|
|
}
|
|
|
|
for selected in state
|
|
.environments
|
|
.iter_mut()
|
|
.chain(&mut configured.environments)
|
|
{
|
|
if !selected.attachment.environment.is_remote()
|
|
&& let Some(Ok(resolved)) = selected.resolution.clone().now_or_never()
|
|
{
|
|
self.restart_shell_snapshot(selected, resolved);
|
|
}
|
|
}
|
|
}
|
|
|
|
fn restart_shell_snapshot(
|
|
&self,
|
|
selected: &mut SelectedTurnEnvironment,
|
|
resolved: ResolvedEnvironment,
|
|
) {
|
|
let shell_snapshot = Self::start_shell_snapshot_task(
|
|
self.shell_snapshot.clone(),
|
|
Arc::clone(&selected.attachment.environment),
|
|
selected.selection.cwd.clone(),
|
|
resolved.shell.clone(),
|
|
&selected.selection.config,
|
|
);
|
|
selected.resolution = futures::future::ready(Ok(ResolvedEnvironment {
|
|
shell_snapshot,
|
|
shell_snapshot_cache: Arc::default(),
|
|
shell_snapshot_v2_supported: cfg!(unix)
|
|
&& !self.shell_snapshot.should_rebuild_inherited(),
|
|
..resolved
|
|
}))
|
|
.boxed()
|
|
.shared();
|
|
}
|
|
|
|
/// Combines persisted thread roots with installed attachment roots, keeping
|
|
/// thread roots first and hiding attachments that are not ready yet.
|
|
pub(crate) fn inspect_selected_capability_roots(
|
|
&self,
|
|
thread_selected_capability_roots: &[SelectedCapabilityRoot],
|
|
) -> SelectedCapabilityRootsStatus {
|
|
let state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
|
|
let environments = &state.environments;
|
|
let mut selected_capability_roots = combine_selected_capability_roots(
|
|
thread_selected_capability_roots,
|
|
environments.iter().filter_map(|environment| {
|
|
let EnvironmentConfigState::Ready(config) = &environment.selection.config else {
|
|
return None;
|
|
};
|
|
Some(
|
|
environment
|
|
.config_origin
|
|
.selected_capability_roots(&environment.attachment.environment, config),
|
|
)
|
|
}),
|
|
);
|
|
let mut seen_root_ids = HashSet::with_capacity(selected_capability_roots.len());
|
|
selected_capability_roots.retain(|root| seen_root_ids.insert(root.id.clone()));
|
|
|
|
let mut status = self
|
|
.environment_manager
|
|
.inspect_selected_capability_roots(&selected_capability_roots);
|
|
status.ready_roots.retain(|root| {
|
|
let CapabilityRootLocation::Environment { environment_id, .. } = &root.location;
|
|
environments
|
|
.iter()
|
|
.find(|environment| &environment.selection.environment_id == environment_id)
|
|
.is_none_or(|environment| {
|
|
matches!(environment.resolution.clone().now_or_never(), Some(Ok(_)))
|
|
})
|
|
});
|
|
status
|
|
}
|
|
|
|
#[tracing::instrument(
|
|
name = "environments.resolve",
|
|
skip_all,
|
|
fields(
|
|
environment_id = %selection.environment_id,
|
|
remote = environment.is_remote(),
|
|
configuration_pending = configuration_ready.is_some(),
|
|
)
|
|
)]
|
|
async fn resolve_environment(
|
|
selection: TurnEnvironmentSelection,
|
|
environment: Arc<Environment>,
|
|
local_shell: Shell,
|
|
shell_snapshot: ShellSnapshot,
|
|
configuration_ready: Option<PendingConfiguration>,
|
|
) -> TurnEnvironmentResult {
|
|
let environment_id = &selection.environment_id;
|
|
if let EnvironmentConfigState::Failed(error) = &selection.config {
|
|
return Err(Arc::new(ExecServerError::Protocol(error.clone())));
|
|
}
|
|
|
|
// Wait for the shared executor connection.
|
|
let connection_ready = async {
|
|
environment.wait_until_ready().await.map_err(|error| {
|
|
tracing::warn!("turn environment `{environment_id}` failed to start: {error}");
|
|
Arc::new(error)
|
|
})
|
|
};
|
|
// Wait for this thread to accept the configuration, if pending.
|
|
let configuration_ready = async move {
|
|
let Some(configuration_ready) = configuration_ready else {
|
|
return Ok(None);
|
|
};
|
|
configuration_ready
|
|
.await
|
|
.map(Some)
|
|
.map_err(|error| Arc::new(ExecServerError::Protocol(error)))
|
|
};
|
|
// Resolve the attachment only after both prerequisites are ready.
|
|
let ((), installed_config) = tokio::try_join!(connection_ready, configuration_ready)?;
|
|
let executor_platform_os;
|
|
let (shell, user_home_dir, temporary_dirs, snapshot_v2) = if environment.is_remote() {
|
|
match environment.info().await {
|
|
Ok(info) => {
|
|
executor_platform_os = info.platform_os;
|
|
let user_home_dir = info.user_home_dir;
|
|
let temporary_directories = info.temporary_directories;
|
|
let shell_snapshot_v2_supported = info.capabilities.shell_snapshot_v2;
|
|
let shell = match Shell::from_environment_shell_info(info.shell) {
|
|
Ok(shell) => Some(shell),
|
|
Err(err) => {
|
|
tracing::warn!(
|
|
"failed to resolve shell for environment `{environment_id}`: {err}"
|
|
);
|
|
None
|
|
}
|
|
};
|
|
(
|
|
shell,
|
|
user_home_dir,
|
|
temporary_directories,
|
|
shell_snapshot_v2_supported,
|
|
)
|
|
}
|
|
Err(err) => {
|
|
executor_platform_os = None;
|
|
tracing::warn!("failed to get info for environment `{environment_id}`: {err}");
|
|
(None, None, None, false)
|
|
}
|
|
}
|
|
} else {
|
|
executor_platform_os = Some(std::env::consts::OS.to_string());
|
|
(
|
|
Some(local_shell),
|
|
PathUri::from_host_native_path("~").ok(),
|
|
Some(EnvironmentInfo::local_temporary_directories()),
|
|
cfg!(unix),
|
|
)
|
|
};
|
|
let shell_snapshot_builder = shell_snapshot.clone();
|
|
let snapshot_config = installed_config
|
|
.clone()
|
|
.map(EnvironmentConfigState::Ready)
|
|
.unwrap_or(selection.config);
|
|
let task = Self::start_shell_snapshot_task(
|
|
shell_snapshot,
|
|
Arc::clone(&environment),
|
|
selection.cwd,
|
|
shell.clone(),
|
|
&snapshot_config,
|
|
);
|
|
let shell_snapshot_v2_supported = snapshot_v2
|
|
&& (environment.is_remote() || !shell_snapshot_builder.should_rebuild_inherited());
|
|
Ok(ResolvedEnvironment {
|
|
shell,
|
|
user_home_dir,
|
|
executor_platform_os,
|
|
temporary_directories: temporary_dirs,
|
|
shell_snapshot: task,
|
|
shell_snapshot_builder,
|
|
shell_snapshot_cache: Arc::default(),
|
|
shell_snapshot_v2_supported,
|
|
installed_config,
|
|
})
|
|
}
|
|
|
|
/// Captures the selected list immediately, then returns a future that may wait for setup.
|
|
/// This lets callers release their locks before waiting without accidentally using a newer
|
|
/// selection. The trace still covers the setup wait itself.
|
|
pub(crate) fn snapshot(
|
|
&self,
|
|
) -> impl Future<Output = TurnEnvironmentSnapshot> + Send + 'static {
|
|
self.resolve_snapshot(self.snapshot_now())
|
|
}
|
|
|
|
/// Copies the current selection without waiting for any executor to connect.
|
|
pub(crate) fn snapshot_now(&self) -> TurnEnvironmentSnapshot {
|
|
Self::snapshot_state(&self.state).refresh_readiness()
|
|
}
|
|
|
|
/// Captures configured bindings without resolving readiness or losing config ownership.
|
|
/// Inheritance and eviction retain these bindings; execution and validation resolve a view.
|
|
pub(crate) fn configured_snapshot_now(&self) -> TurnEnvironmentSnapshot {
|
|
Self::snapshot_state(&self.configured)
|
|
}
|
|
|
|
fn snapshot_state(state: &Mutex<ThreadEnvironmentsState>) -> TurnEnvironmentSnapshot {
|
|
// Keep the startup results, but not the senders that would keep removed Pending work alive.
|
|
let environments = state
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner)
|
|
.environments
|
|
.iter()
|
|
.map(|environment| {
|
|
TurnEnvironmentState::Starting(StartingTurnEnvironment {
|
|
selection: environment.selection.clone(),
|
|
config_origin: environment.config_origin,
|
|
resolution: environment.resolution.clone(),
|
|
attachment: Arc::clone(&environment.attachment),
|
|
owner_config_result: environment.owner_config_result.clone().or_else(|| {
|
|
environment
|
|
.pending_completion
|
|
.as_ref()
|
|
.map(pending_configuration)
|
|
}),
|
|
})
|
|
})
|
|
.collect();
|
|
TurnEnvironmentSnapshot { environments }
|
|
}
|
|
|
|
/// Applies the same startup policy to an already captured selection without adopting newer ones.
|
|
#[tracing::instrument(
|
|
name = "environments.snapshot",
|
|
skip_all,
|
|
fields(
|
|
environment_count = snapshot.environments.len(),
|
|
non_blocking = non_blocking_snapshots,
|
|
)
|
|
)]
|
|
pub(crate) fn resolve_snapshot(
|
|
&self,
|
|
snapshot: TurnEnvironmentSnapshot,
|
|
) -> impl Future<Output = TurnEnvironmentSnapshot> + Send + 'static {
|
|
let non_blocking_snapshots = self.non_blocking_snapshots;
|
|
async move {
|
|
if !non_blocking_snapshots {
|
|
for starting in snapshot.starting() {
|
|
if !matches!(starting.selection.config, EnvironmentConfigState::Pending) {
|
|
let _ = starting.wait_until_ready().await;
|
|
}
|
|
}
|
|
}
|
|
// An earlier environment may have become ready while we waited for a later one.
|
|
snapshot.refresh_readiness()
|
|
}
|
|
}
|
|
|
|
pub(crate) fn environment_manager(&self) -> Arc<EnvironmentManager> {
|
|
Arc::clone(&self.environment_manager)
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub(crate) enum TurnEnvironmentState {
|
|
Ready(TurnEnvironment),
|
|
Starting(StartingTurnEnvironment),
|
|
/// Unavailable for execution, but still selected when evaluating permissions.
|
|
Failed {
|
|
// Keep the input form so connection failure doesn't hide whether config comes from the thread.
|
|
selection: TurnEnvironmentSelection,
|
|
required_skills: Vec<String>,
|
|
error: String,
|
|
},
|
|
}
|
|
|
|
impl TurnEnvironmentState {
|
|
fn from_resolution(
|
|
starting: StartingTurnEnvironment,
|
|
resolved: Option<TurnEnvironmentResult>,
|
|
) -> Self {
|
|
let required_skills = starting.attachment.environment.required_skills().to_vec();
|
|
if let EnvironmentConfigState::Failed(error) = &starting.selection.config {
|
|
let error = error.clone();
|
|
return Self::Failed {
|
|
required_skills,
|
|
selection: starting
|
|
.config_origin
|
|
.into_input_selection(starting.selection),
|
|
error,
|
|
};
|
|
}
|
|
match resolved {
|
|
Some(Ok(environment)) => {
|
|
let mut selection = starting.selection;
|
|
if matches!(selection.config, EnvironmentConfigState::Pending) {
|
|
let Some(config) = environment.installed_config else {
|
|
return Self::Failed {
|
|
required_skills,
|
|
selection: starting.config_origin.into_input_selection(selection),
|
|
error: "Environment configuration was not supplied.".to_string(),
|
|
};
|
|
};
|
|
selection.config = EnvironmentConfigState::Ready(config);
|
|
}
|
|
let mut turn_environment = TurnEnvironment::from_attachment(
|
|
selection,
|
|
starting.config_origin,
|
|
starting.attachment,
|
|
environment.shell,
|
|
);
|
|
turn_environment.executor_platform_os = environment.executor_platform_os;
|
|
turn_environment.shell_snapshot = environment.shell_snapshot;
|
|
turn_environment.shell_snapshot_builder =
|
|
Some(Box::new(environment.shell_snapshot_builder));
|
|
turn_environment.shell_snapshot_cache = environment.shell_snapshot_cache;
|
|
turn_environment.shell_snapshot_v2_supported =
|
|
environment.shell_snapshot_v2_supported;
|
|
turn_environment.user_home_dir = environment.user_home_dir;
|
|
turn_environment.temporary_directories = environment.temporary_directories;
|
|
Self::Ready(turn_environment)
|
|
}
|
|
Some(Err(err)) => {
|
|
tracing::debug!(
|
|
environment_id = %starting.selection.environment_id,
|
|
"skipping failed turn environment: {err}"
|
|
);
|
|
Self::Failed {
|
|
required_skills,
|
|
selection: starting
|
|
.config_origin
|
|
.into_input_selection(starting.selection),
|
|
error: err.to_string(),
|
|
}
|
|
}
|
|
None => Self::Starting(starting),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Existing environment bindings captured for a turn. Internal child threads can
|
|
/// retain these bindings without resolving a different set of environments.
|
|
#[derive(Clone, Debug, Default)]
|
|
pub struct TurnEnvironmentSnapshot {
|
|
// Keep every selected environment, including failures, in its original order.
|
|
pub(crate) environments: Vec<TurnEnvironmentState>,
|
|
}
|
|
|
|
impl TurnEnvironmentSnapshot {
|
|
/// Keeps the original root order for child threads and saved history.
|
|
pub(crate) fn selected_capability_roots(&self) -> Vec<SelectedCapabilityRoot> {
|
|
EnvironmentCapabilityRoots::collect(self.environments.iter().map(|environment| {
|
|
let selection = match environment {
|
|
TurnEnvironmentState::Ready(environment) => &environment.selection,
|
|
TurnEnvironmentState::Starting(environment) => &environment.selection,
|
|
TurnEnvironmentState::Failed { selection, .. } => selection,
|
|
};
|
|
&selection.selected_capability_roots
|
|
}))
|
|
}
|
|
|
|
pub(crate) fn has_full_access(
|
|
&self,
|
|
approval_policy: AskForApproval,
|
|
thread_profile: &PermissionProfile,
|
|
) -> bool {
|
|
codex_protocol::protocol::has_full_access(
|
|
approval_policy,
|
|
thread_profile,
|
|
self.refresh_readiness()
|
|
.environments
|
|
.iter()
|
|
.map(|environment| match environment {
|
|
TurnEnvironmentState::Ready(environment) => &environment.selection.config,
|
|
TurnEnvironmentState::Starting(_) | TurnEnvironmentState::Failed { .. } => {
|
|
&EnvironmentConfigState::Pending
|
|
}
|
|
}),
|
|
)
|
|
}
|
|
|
|
/// Promotes completed startup work without adopting newer thread selections.
|
|
pub(crate) fn refresh_readiness(&self) -> Self {
|
|
let environments = self
|
|
.environments
|
|
.iter()
|
|
.map(|environment| match environment {
|
|
TurnEnvironmentState::Ready(environment) => {
|
|
TurnEnvironmentState::Ready(environment.clone())
|
|
}
|
|
TurnEnvironmentState::Starting(environment) => {
|
|
TurnEnvironmentState::from_resolution(
|
|
environment.clone(),
|
|
environment.resolution.clone().now_or_never(),
|
|
)
|
|
}
|
|
TurnEnvironmentState::Failed { .. } => environment.clone(),
|
|
})
|
|
.collect();
|
|
Self { environments }
|
|
}
|
|
|
|
pub(crate) fn turn_environments(&self) -> impl Iterator<Item = &TurnEnvironment> {
|
|
self.environments.iter().filter_map(|environment| {
|
|
let TurnEnvironmentState::Ready(environment) = environment else {
|
|
return None;
|
|
};
|
|
Some(environment)
|
|
})
|
|
}
|
|
|
|
pub(crate) fn starting(&self) -> impl Iterator<Item = &StartingTurnEnvironment> {
|
|
self.environments.iter().filter_map(|environment| {
|
|
let TurnEnvironmentState::Starting(environment) = environment else {
|
|
return None;
|
|
};
|
|
Some(environment)
|
|
})
|
|
}
|
|
|
|
pub(crate) fn ready_environment_handles(&self) -> HashMap<String, Arc<Environment>> {
|
|
self.turn_environments()
|
|
.map(|environment| {
|
|
(
|
|
environment.selection.environment_id.clone(),
|
|
Arc::clone(environment.environment()),
|
|
)
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Maps each captured environment to its exact ready handle, or `None` when it was starting.
|
|
pub(crate) fn captured_environments(&self) -> HashMap<String, Option<Arc<Environment>>> {
|
|
self.turn_environments()
|
|
.map(|environment| {
|
|
(
|
|
environment.selection.environment_id.clone(),
|
|
Some(Arc::clone(environment.environment())),
|
|
)
|
|
})
|
|
.chain(
|
|
self.starting()
|
|
.map(|environment| (environment.selection.environment_id.clone(), None)),
|
|
)
|
|
.collect()
|
|
}
|
|
|
|
pub(crate) fn primary(&self) -> Option<&TurnEnvironment> {
|
|
self.turn_environments().next()
|
|
}
|
|
|
|
/// Returns the first selection's executor paths, even when setup is not ready yet.
|
|
pub(crate) fn primary_workspace_root_uris(&self) -> &[PathUri] {
|
|
let selection = match self.environments.first() {
|
|
Some(TurnEnvironmentState::Ready(environment)) => &environment.selection,
|
|
Some(TurnEnvironmentState::Starting(environment)) => &environment.selection,
|
|
Some(TurnEnvironmentState::Failed { selection, .. }) => selection,
|
|
None => return &[],
|
|
};
|
|
&selection.workspace_roots
|
|
}
|
|
|
|
/// Returns the primary environment's resolved permissions, or the provided fallback.
|
|
pub(crate) fn permission_profile_or_else(
|
|
&self,
|
|
fallback: impl FnOnce() -> PermissionProfile,
|
|
) -> PermissionProfile {
|
|
self.primary()
|
|
.map(TurnEnvironment::permission_profile_with_workspace_roots)
|
|
.unwrap_or_else(fallback)
|
|
}
|
|
|
|
pub(crate) fn local(&self) -> Option<&TurnEnvironment> {
|
|
self.turn_environments()
|
|
.find(|environment| !environment.environment().is_remote())
|
|
}
|
|
|
|
pub(crate) fn local_environment_cwd(&self) -> Option<AbsolutePathBuf> {
|
|
self.local_selection()
|
|
.and_then(|selection| selection.cwd.to_abs_path().ok())
|
|
}
|
|
|
|
/// The local workspace remains host-owned while its executor is still starting.
|
|
pub(crate) fn local_selection(&self) -> Option<&TurnEnvironmentSelection> {
|
|
self.environments
|
|
.iter()
|
|
.find_map(|environment| match environment {
|
|
TurnEnvironmentState::Ready(environment)
|
|
if !environment.environment().is_remote() =>
|
|
{
|
|
Some(&environment.selection)
|
|
}
|
|
TurnEnvironmentState::Ready(_) => None,
|
|
TurnEnvironmentState::Starting(environment)
|
|
if environment.selection.environment_id
|
|
== codex_exec_server::LOCAL_ENVIRONMENT_ID =>
|
|
{
|
|
Some(&environment.selection)
|
|
}
|
|
TurnEnvironmentState::Starting(_) | TurnEnvironmentState::Failed { .. } => None,
|
|
})
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn primary_environment(&self) -> Option<Arc<codex_exec_server::Environment>> {
|
|
self.primary()
|
|
.map(|environment| Arc::clone(environment.environment()))
|
|
}
|
|
|
|
pub(crate) fn to_selections(&self) -> Vec<TurnEnvironmentSelection> {
|
|
self.turn_environments()
|
|
.map(TurnEnvironment::selection)
|
|
.collect()
|
|
}
|
|
|
|
/// Captures requirements from the selected registration even when connection setup fails.
|
|
pub(crate) fn required_skills(&self) -> impl Iterator<Item = EnvironmentSkillRequirements<'_>> {
|
|
self.environments.iter().filter_map(|environment| {
|
|
let (selection, required) = match environment {
|
|
TurnEnvironmentState::Ready(environment) => (
|
|
&environment.selection,
|
|
environment.environment().required_skills(),
|
|
),
|
|
TurnEnvironmentState::Starting(environment) => (
|
|
&environment.selection,
|
|
environment.attachment.environment.required_skills(),
|
|
),
|
|
TurnEnvironmentState::Failed {
|
|
selection,
|
|
required_skills,
|
|
..
|
|
} => (selection, required_skills.as_slice()),
|
|
};
|
|
if required.is_empty() {
|
|
None
|
|
} else {
|
|
Some(EnvironmentSkillRequirements {
|
|
environment_id: &selection.environment_id,
|
|
skill_names: required,
|
|
})
|
|
}
|
|
})
|
|
}
|
|
|
|
/// Requests for native children retain ready and starting attachments from their spawning step.
|
|
pub(crate) fn inheritable_requests(&self) -> Vec<TurnEnvironmentRequest> {
|
|
self.environments
|
|
.iter()
|
|
.filter_map(|environment| match environment {
|
|
TurnEnvironmentState::Ready(environment) => Some(environment.selection()),
|
|
TurnEnvironmentState::Starting(environment) => Some(
|
|
environment
|
|
.config_origin
|
|
.into_input_selection(environment.selection.clone()),
|
|
),
|
|
TurnEnvironmentState::Failed { .. } => None,
|
|
})
|
|
.map(TurnEnvironmentSelection::into_request)
|
|
.collect()
|
|
}
|
|
|
|
/// Returns every captured selection, including those still starting or unable to connect.
|
|
pub(crate) fn all_selections(&self) -> Vec<TurnEnvironmentSelection> {
|
|
self.environments
|
|
.iter()
|
|
.map(|environment| match environment {
|
|
TurnEnvironmentState::Ready(environment) => environment.selection(),
|
|
TurnEnvironmentState::Starting(environment) => environment
|
|
.config_origin
|
|
.into_input_selection(environment.selection.clone()),
|
|
TurnEnvironmentState::Failed { selection, .. } => selection.clone(),
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
pub(crate) fn single_local_environment(&self) -> Option<&TurnEnvironment> {
|
|
if self.starting().next().is_some() {
|
|
return None;
|
|
}
|
|
let mut environments = self.turn_environments();
|
|
let environment = environments.next()?;
|
|
if environments.next().is_some() {
|
|
return None;
|
|
}
|
|
|
|
(!environment.environment().is_remote()).then_some(environment)
|
|
}
|
|
|
|
pub(crate) fn single_local_environment_cwd(&self) -> Option<AbsolutePathBuf> {
|
|
// TODO(anp): Migrate local-environment consumers to PathUri so this compatibility
|
|
// conversion can be removed.
|
|
self.single_local_environment()?.cwd().to_abs_path().ok()
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "environment_selection_snapshot_tests.rs"]
|
|
mod snapshot_tests;
|
|
|
|
#[cfg(test)]
|
|
#[path = "environment_connection_tests.rs"]
|
|
mod connection_tests;
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::time::Duration;
|
|
|
|
use crate::config::PermissionProfileSnapshot;
|
|
use codex_exec_server::Environment;
|
|
use codex_exec_server::ExecServerRuntimeOptions;
|
|
use codex_exec_server::LOCAL_ENVIRONMENT_ID;
|
|
use codex_exec_server::REMOTE_ENVIRONMENT_ID;
|
|
use codex_exec_server_test_support::environment_manager_without_environments;
|
|
use codex_http_client::HttpClientFactory;
|
|
use codex_http_client::OutboundProxyPolicy;
|
|
use codex_protocol::config_types::WindowsSandboxLevel;
|
|
use codex_protocol::models::ActivePermissionProfile;
|
|
use codex_protocol::models::PermissionProfile;
|
|
use codex_protocol::permissions::FileSystemSandboxPolicyContext;
|
|
use codex_protocol::protocol::EnvironmentConnectionEvent;
|
|
use codex_protocol::protocol::EventMsg;
|
|
use codex_protocol::protocol::TurnEnvironmentRequest;
|
|
use codex_protocol::protocol::TurnEnvironmentSelection;
|
|
use codex_protocol::sandbox::SandboxType;
|
|
use codex_utils_absolute_path::AbsolutePathBuf;
|
|
use codex_utils_path_uri::PathUri;
|
|
use futures::SinkExt;
|
|
use futures::StreamExt;
|
|
use pretty_assertions::assert_eq;
|
|
use serde_json::Value;
|
|
use tokio::net::TcpListener;
|
|
use tokio::net::TcpStream;
|
|
use tokio::time::timeout;
|
|
use tokio_tungstenite::WebSocketStream;
|
|
use tokio_tungstenite::accept_async;
|
|
use tokio_tungstenite::tungstenite::Message;
|
|
use tracing::Level;
|
|
use tracing_subscriber::fmt::format::FmtSpan;
|
|
use tracing_test::internal::MockWriter;
|
|
|
|
use super::*;
|
|
|
|
pub(super) fn test_environment_config() -> EnvironmentConfig {
|
|
EnvironmentConfig {
|
|
environment_hooks: Default::default(),
|
|
allow_login_shell: true,
|
|
workspace_roots: Vec::new(),
|
|
windows_sandbox_level: WindowsSandboxLevel::Disabled,
|
|
windows_sandbox_type: codex_protocol::sandbox::SandboxType::None,
|
|
use_legacy_landlock: false,
|
|
permission_profile: PermissionProfileSnapshot::legacy(PermissionProfile::read_only()),
|
|
shell_environment_policy: Default::default(),
|
|
exec_policy: None,
|
|
mcp_policy: None,
|
|
network_policy: None,
|
|
selected_capability_roots: Vec::new(),
|
|
}
|
|
}
|
|
|
|
async fn resolve_turn_environments(
|
|
environment_manager: Arc<EnvironmentManager>,
|
|
selections: &[TurnEnvironmentSelection],
|
|
) -> Arc<ThreadEnvironments> {
|
|
let turn_environments = Arc::new(ThreadEnvironments::new(
|
|
environment_manager,
|
|
crate::shell::default_user_shell(),
|
|
ThreadEnvironmentDefaults::new(test_environment_config(), SandboxType::None),
|
|
ShellSnapshot::disabled(),
|
|
TurnEnvironmentSnapshot::default(),
|
|
/*non_blocking_snapshots*/ false,
|
|
));
|
|
turn_environments.update_selections(selections);
|
|
turn_environments.snapshot().await;
|
|
turn_environments
|
|
}
|
|
|
|
fn test_runtime_paths() -> ExecServerRuntimeOptions {
|
|
ExecServerRuntimeOptions::new(
|
|
std::env::current_exe().expect("current exe"),
|
|
/*codex_linux_sandbox_exe*/ None,
|
|
)
|
|
.expect("runtime paths")
|
|
}
|
|
|
|
pub(super) async fn read_websocket_json(websocket: &mut WebSocketStream<TcpStream>) -> Value {
|
|
loop {
|
|
match timeout(std::time::Duration::from_secs(5), websocket.next())
|
|
.await
|
|
.expect("websocket read should not time out")
|
|
.expect("websocket should stay open")
|
|
.expect("websocket frame should read")
|
|
{
|
|
Message::Text(text) => {
|
|
return serde_json::from_str(text.as_ref()).expect("valid JSON-RPC message");
|
|
}
|
|
Message::Binary(bytes) => {
|
|
return serde_json::from_slice(bytes.as_ref()).expect("valid JSON-RPC message");
|
|
}
|
|
Message::Ping(_) | Message::Pong(_) => {}
|
|
other => panic!("expected JSON-RPC message, got {other:?}"),
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn serve_environment_info(listener: TcpListener) {
|
|
let (stream, _) = listener.accept().await.expect("connection");
|
|
let mut websocket = accept_async(stream).await.expect("websocket handshake");
|
|
|
|
let initialize = read_websocket_json(&mut websocket).await;
|
|
assert_eq!(initialize["method"], "initialize");
|
|
websocket
|
|
.send(Message::Text(
|
|
serde_json::json!({
|
|
"id": initialize["id"],
|
|
"result": { "sessionId": "test-session" }
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("initialize response");
|
|
let initialized = read_websocket_json(&mut websocket).await;
|
|
assert_eq!(initialized["method"], "initialized");
|
|
|
|
let info = read_websocket_json(&mut websocket).await;
|
|
assert_eq!(info["method"], "environment/info");
|
|
websocket
|
|
.send(Message::Text(
|
|
serde_json::json!({
|
|
"id": info["id"],
|
|
"result": {
|
|
"shell": { "name": "zsh", "path": "/bin/zsh" },
|
|
"userHomeDir": "file:///home/remote",
|
|
"platformOs": "windows",
|
|
"temporaryDirectories": ["file:///tmp/remote"],
|
|
"capabilities": {"windowsMxc": true},
|
|
}
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("environment info response");
|
|
}
|
|
|
|
#[tokio::test]
|
|
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(
|
|
Some("ws://127.0.0.1:8765".to_string()),
|
|
Some(test_runtime_paths()),
|
|
)
|
|
.await;
|
|
|
|
assert_eq!(
|
|
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],
|
|
config: EnvironmentConfigState::FromThread,
|
|
}]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
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"),
|
|
r#"
|
|
[[environments]]
|
|
id = "remote"
|
|
url = "ws://127.0.0.1:8765"
|
|
"#,
|
|
)
|
|
.expect("write environments.toml");
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let cwd_uri = PathUri::from_abs_path(&cwd);
|
|
let manager = EnvironmentManager::from_codex_home(
|
|
temp_dir.path(),
|
|
Some(test_runtime_paths()),
|
|
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
|
|
)
|
|
.await
|
|
.expect("environment manager");
|
|
|
|
assert_eq!(
|
|
default_thread_environment_requests(&manager, &cwd, std::slice::from_ref(&cwd)),
|
|
vec![
|
|
TurnEnvironmentRequest {
|
|
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
|
cwd: cwd_uri.clone(),
|
|
workspace_roots: vec![cwd_uri.clone()],
|
|
config: EnvironmentConfigState::FromThread,
|
|
},
|
|
TurnEnvironmentRequest {
|
|
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
|
cwd: cwd_uri.clone(),
|
|
workspace_roots: vec![cwd_uri],
|
|
config: EnvironmentConfigState::FromThread,
|
|
},
|
|
]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
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_requests(&manager, &cwd, std::slice::from_ref(&cwd)),
|
|
Vec::<TurnEnvironmentRequest>::new()
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn local_environment_uses_configured_shell_and_login_policy() {
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let local_shell = Shell {
|
|
shell_type: crate::shell::ShellType::Zsh,
|
|
shell_path: std::path::PathBuf::from("/configured/zsh"),
|
|
};
|
|
let expected_config = EnvironmentConfig {
|
|
environment_hooks: Default::default(),
|
|
allow_login_shell: false,
|
|
workspace_roots: Vec::new(),
|
|
windows_sandbox_level: WindowsSandboxLevel::Disabled,
|
|
windows_sandbox_type: codex_protocol::sandbox::SandboxType::None,
|
|
use_legacy_landlock: false,
|
|
permission_profile: PermissionProfileSnapshot::active_with_profile_workspace_roots(
|
|
PermissionProfile::read_only(),
|
|
ActivePermissionProfile::read_only(),
|
|
vec![cwd.join("profile-root").into()],
|
|
),
|
|
shell_environment_policy: Default::default(),
|
|
exec_policy: None,
|
|
mcp_policy: None,
|
|
network_policy: None,
|
|
selected_capability_roots: Vec::new(),
|
|
};
|
|
let turn_environments = ThreadEnvironments::new(
|
|
Arc::new(EnvironmentManager::default_for_tests()),
|
|
local_shell.clone(),
|
|
ThreadEnvironmentDefaults::new(
|
|
expected_config.clone(),
|
|
expected_config.windows_sandbox_type,
|
|
),
|
|
ShellSnapshot::disabled(),
|
|
TurnEnvironmentSnapshot::default(),
|
|
/*non_blocking_snapshots*/ false,
|
|
);
|
|
turn_environments.update_selections(&[TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
|
cwd: PathUri::from_abs_path(&cwd),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
}]);
|
|
|
|
let snapshot = turn_environments.snapshot().await;
|
|
let environment = snapshot.primary().expect("local environment");
|
|
let expected_user_home_dir = PathUri::from_host_native_path("~").ok();
|
|
let expected_temporary_directories = EnvironmentInfo::local_temporary_directories();
|
|
|
|
assert_eq!(environment.shell.as_ref(), Some(&local_shell));
|
|
assert_eq!(environment.config(), &expected_config);
|
|
assert_eq!(
|
|
environment
|
|
.sandbox_context(/*additional_permissions*/ None)
|
|
.policy_context(),
|
|
FileSystemSandboxPolicyContext {
|
|
cwd: environment.cwd(),
|
|
workspace_roots: &[],
|
|
user_home_dir: expected_user_home_dir.as_ref(),
|
|
temporary_directories: Some(expected_temporary_directories.as_slice()),
|
|
}
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn resolve_environment_selections_keeps_first_duplicate_id() {
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let cwd_uri = PathUri::from_abs_path(&cwd);
|
|
let manager = Arc::new(EnvironmentManager::default_for_tests());
|
|
let first = TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
|
cwd: cwd_uri.clone(),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
};
|
|
|
|
let resolved = resolve_turn_environments(
|
|
manager,
|
|
&[
|
|
first.clone(),
|
|
TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
|
cwd: cwd_uri.join("other").expect("other cwd URI"),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
},
|
|
],
|
|
)
|
|
.await;
|
|
|
|
assert_eq!(resolved.snapshot().await.to_selections(), vec![first]);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn resolved_environment_selections_use_first_selection_as_primary() {
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let selected_cwd = cwd.join("selected");
|
|
let selected_cwd_uri = PathUri::from_abs_path(&selected_cwd);
|
|
let manager = Arc::new(EnvironmentManager::default_for_tests());
|
|
|
|
let resolved = resolve_turn_environments(
|
|
Arc::clone(&manager),
|
|
&[TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: "local".to_string(),
|
|
cwd: selected_cwd_uri,
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
}],
|
|
)
|
|
.await;
|
|
|
|
let resolved = resolved.snapshot().await;
|
|
assert_eq!(
|
|
resolved
|
|
.primary()
|
|
.expect("primary environment")
|
|
.selection
|
|
.environment_id,
|
|
"local"
|
|
);
|
|
assert_eq!(
|
|
resolved.primary().expect("primary environment").shell,
|
|
Some(
|
|
Shell::from_environment_shell_info(
|
|
manager
|
|
.get_environment("local")
|
|
.expect("local environment")
|
|
.info()
|
|
.await
|
|
.expect("local environment info")
|
|
.shell
|
|
)
|
|
.expect("resolved shell")
|
|
)
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn unresolved_environment_selections_are_skipped() {
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let cwd_uri = PathUri::from_abs_path(&cwd);
|
|
let manager = Arc::new(EnvironmentManager::default_for_tests());
|
|
let local = TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
|
cwd: cwd_uri.clone(),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
};
|
|
|
|
let resolved = resolve_turn_environments(
|
|
manager,
|
|
&[
|
|
TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: "missing".to_string(),
|
|
cwd: cwd_uri,
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
},
|
|
local.clone(),
|
|
],
|
|
)
|
|
.await;
|
|
|
|
assert_eq!(resolved.snapshot().await.to_selections(), vec![local]);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn blocking_snapshot_waits_for_starting_environment() {
|
|
let buffer: &'static std::sync::Mutex<Vec<u8>> =
|
|
Box::leak(Box::new(std::sync::Mutex::new(Vec::new())));
|
|
let subscriber = tracing_subscriber::fmt()
|
|
.with_ansi(false)
|
|
.with_max_level(Level::TRACE)
|
|
.with_span_events(FmtSpan::NEW)
|
|
.with_writer(MockWriter::new(buffer))
|
|
.finish();
|
|
// Avoid tracing-core's single-dispatch path caching no interest when another
|
|
// test first reaches these shared callsites without a subscriber.
|
|
let _interest_cache_guard =
|
|
tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
|
|
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
|
|
|
|
let listener = TcpListener::bind("127.0.0.1:0")
|
|
.await
|
|
.expect("bind websocket listener");
|
|
let manager = Arc::new(
|
|
EnvironmentManager::create_for_tests(
|
|
Some(format!(
|
|
"ws://{}",
|
|
listener.local_addr().expect("listener address")
|
|
)),
|
|
Some(test_runtime_paths()),
|
|
)
|
|
.await,
|
|
);
|
|
let selection = TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
|
cwd: PathUri::from_abs_path(&AbsolutePathBuf::current_dir().expect("cwd")),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
};
|
|
let environments = Arc::new(ThreadEnvironments::new(
|
|
manager,
|
|
crate::shell::default_user_shell(),
|
|
ThreadEnvironmentDefaults::new(test_environment_config(), SandboxType::None),
|
|
ShellSnapshot::disabled(),
|
|
TurnEnvironmentSnapshot::default(),
|
|
/*non_blocking_snapshots*/ false,
|
|
));
|
|
environments.update_selections(std::slice::from_ref(&selection));
|
|
// Exercise first-use callsite registration from a thread without a subscriber.
|
|
std::thread::spawn({
|
|
let environments = Arc::clone(&environments);
|
|
move || assert!(environments.snapshot().now_or_never().is_none())
|
|
})
|
|
.join()
|
|
.expect("unsubscribed snapshot");
|
|
let snapshot_task = tokio::spawn({
|
|
let environments = Arc::clone(&environments);
|
|
async move { environments.snapshot().await }
|
|
});
|
|
tokio::task::yield_now().await;
|
|
assert!(!snapshot_task.is_finished());
|
|
|
|
let server = tokio::spawn(serve_environment_info(listener));
|
|
let snapshot = timeout(Duration::from_secs(5), snapshot_task)
|
|
.await
|
|
.expect("snapshot should finish after the environment starts")
|
|
.expect("snapshot task");
|
|
|
|
assert!(snapshot.starting().next().is_none());
|
|
assert_eq!(snapshot.to_selections(), vec![selection]);
|
|
server.await.expect("server task");
|
|
|
|
let logs = String::from_utf8(buffer.lock().expect("buffer lock").clone())
|
|
.expect("UTF-8 tracing output");
|
|
assert!(
|
|
logs.contains(
|
|
"environments.snapshot{environment_count=1 non_blocking=false}:environments.wait_until_ready{environment_id=remote}"
|
|
),
|
|
"blocking snapshot should contain its environment wait span: {logs}"
|
|
);
|
|
assert!(
|
|
logs.contains(
|
|
"environments.resolve{environment_id=remote remote=true configuration_pending=false}:exec_server.environment.wait_until_ready{remote=true}"
|
|
),
|
|
"environment resolution should contain the executor connection wait span: {logs}"
|
|
);
|
|
assert!(
|
|
logs.contains(
|
|
"environments.resolve{environment_id=remote remote=true configuration_pending=false}:exec_server.environment.info{remote=true}"
|
|
),
|
|
"environment resolution should contain the remote environment-info span: {logs}"
|
|
);
|
|
}
|
|
|
|
#[test_case::test_case(SandboxType::None, SandboxType::None, WindowsSandboxLevel::Disabled; "legacy")]
|
|
#[test_case::test_case(SandboxType::WindowsRestrictedToken, SandboxType::WindowsMxc, WindowsSandboxLevel::Elevated; "local preference")]
|
|
#[test_case::test_case(SandboxType::WindowsMxc, SandboxType::WindowsMxc, WindowsSandboxLevel::Disabled; "explicit MXC")]
|
|
#[tokio::test]
|
|
async fn snapshot_refreshes_readiness_in_selection_order(
|
|
configured_sandbox: SandboxType,
|
|
local_sandbox: SandboxType,
|
|
legacy_level: WindowsSandboxLevel,
|
|
) {
|
|
let listener = TcpListener::bind("127.0.0.1:0")
|
|
.await
|
|
.expect("bind websocket listener");
|
|
let manager = Arc::new(
|
|
EnvironmentManager::create_for_tests_with_local(
|
|
Some(format!(
|
|
"ws://{}",
|
|
listener.local_addr().expect("listener address")
|
|
)),
|
|
test_runtime_paths(),
|
|
)
|
|
.await,
|
|
);
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let expected_config = EnvironmentConfig {
|
|
environment_hooks: Default::default(),
|
|
allow_login_shell: false,
|
|
workspace_roots: Vec::new(),
|
|
windows_sandbox_level: legacy_level,
|
|
windows_sandbox_type: configured_sandbox,
|
|
use_legacy_landlock: false,
|
|
permission_profile: PermissionProfileSnapshot::active_with_profile_workspace_roots(
|
|
PermissionProfile::read_only(),
|
|
ActivePermissionProfile::read_only(),
|
|
vec![cwd.join("profile-root").into()],
|
|
),
|
|
shell_environment_policy: Default::default(),
|
|
exec_policy: None,
|
|
mcp_policy: None,
|
|
network_policy: None,
|
|
selected_capability_roots: Vec::new(),
|
|
};
|
|
let expected_local_config = EnvironmentConfig {
|
|
windows_sandbox_type: local_sandbox,
|
|
..expected_config.clone()
|
|
};
|
|
let cwd = PathUri::from_abs_path(&cwd);
|
|
let remote = TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
|
cwd: cwd.clone(),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
};
|
|
let local = TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
|
cwd,
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
};
|
|
let turn_environments = ThreadEnvironments::new(
|
|
Arc::clone(&manager),
|
|
crate::shell::default_user_shell(),
|
|
ThreadEnvironmentDefaults::new(test_environment_config(), local_sandbox),
|
|
ShellSnapshot::disabled(),
|
|
TurnEnvironmentSnapshot::default(),
|
|
/*non_blocking_snapshots*/ true,
|
|
);
|
|
turn_environments.update_selections(std::slice::from_ref(&local));
|
|
let resolution = turn_environments.state.lock().unwrap().environments[0]
|
|
.resolution
|
|
.clone();
|
|
resolution.await.expect("local environment should resolve");
|
|
turn_environments.update_selections(&[remote.clone(), local.clone()]);
|
|
let initial = turn_environments.snapshot().await;
|
|
assert_eq!(
|
|
initial.primary().expect("local environment").config(),
|
|
&EnvironmentConfig {
|
|
windows_sandbox_type: local_sandbox,
|
|
..test_environment_config()
|
|
},
|
|
);
|
|
assert_eq!(
|
|
initial.starting().next().unwrap().selection.config,
|
|
EnvironmentConfigState::Ready(test_environment_config()),
|
|
);
|
|
turn_environments.set_active_thread_defaults(expected_config.clone());
|
|
|
|
let starting = turn_environments.snapshot().await;
|
|
let resolved_remote = TurnEnvironmentSelection {
|
|
config: EnvironmentConfigState::Ready(expected_config.clone()),
|
|
..remote.clone()
|
|
};
|
|
assert_eq!(
|
|
starting
|
|
.turn_environments()
|
|
.map(TurnEnvironment::selection)
|
|
.collect::<Vec<_>>(),
|
|
vec![local.clone()]
|
|
);
|
|
assert_eq!(
|
|
starting
|
|
.turn_environments()
|
|
.map(|environment| environment.config().clone())
|
|
.collect::<Vec<_>>(),
|
|
vec![expected_local_config.clone()]
|
|
);
|
|
assert_eq!(
|
|
starting
|
|
.starting()
|
|
.map(|environment| environment.selection.clone())
|
|
.collect::<Vec<_>>(),
|
|
vec![resolved_remote.clone()]
|
|
);
|
|
assert_eq!(starting.to_selections(), vec![local.clone()]);
|
|
assert_eq!(
|
|
starting.all_selections(),
|
|
vec![remote.clone(), local.clone()]
|
|
);
|
|
assert!(starting.single_local_environment().is_none());
|
|
|
|
let next_config = EnvironmentConfig {
|
|
environment_hooks: Default::default(),
|
|
permission_profile: PermissionProfileSnapshot::legacy(PermissionProfile::Disabled),
|
|
..test_environment_config()
|
|
};
|
|
turn_environments.set_active_thread_defaults(next_config.clone());
|
|
let next_starting = turn_environments.snapshot().await;
|
|
assert!(
|
|
!next_starting.has_full_access(AskForApproval::Never, &PermissionProfile::Disabled)
|
|
);
|
|
|
|
let server = tokio::spawn(serve_environment_info(listener));
|
|
timeout(
|
|
std::time::Duration::from_secs(5),
|
|
starting
|
|
.starting()
|
|
.next()
|
|
.expect("starting environment")
|
|
.resolution
|
|
.clone(),
|
|
)
|
|
.await
|
|
.expect("environment resolution should finish")
|
|
.expect("environment resolution should succeed");
|
|
assert!(next_starting.has_full_access(AskForApproval::Never, &PermissionProfile::Disabled));
|
|
assert!(!starting.has_full_access(AskForApproval::Never, &PermissionProfile::Disabled));
|
|
let attached = starting.refresh_readiness();
|
|
|
|
assert!(attached.starting().next().is_none());
|
|
let child = ThreadEnvironments::new(
|
|
manager,
|
|
crate::shell::default_user_shell(),
|
|
ThreadEnvironmentDefaults::new(next_config.clone(), local_sandbox),
|
|
ShellSnapshot::disabled(),
|
|
attached.clone(),
|
|
/*non_blocking_snapshots*/ true,
|
|
);
|
|
child.activate_configured(async_channel::unbounded().0);
|
|
assert_eq!(
|
|
child
|
|
.snapshot()
|
|
.await
|
|
.turn_environments()
|
|
.map(|environment| environment.config().clone())
|
|
.collect::<Vec<_>>(),
|
|
next_starting
|
|
.refresh_readiness()
|
|
.turn_environments()
|
|
.map(|environment| environment.config().clone())
|
|
.collect::<Vec<_>>(),
|
|
);
|
|
assert_eq!(
|
|
attached
|
|
.turn_environments()
|
|
.map(TurnEnvironment::selection)
|
|
.collect::<Vec<_>>(),
|
|
vec![remote.clone(), local.clone()]
|
|
);
|
|
assert_eq!(
|
|
attached
|
|
.turn_environments()
|
|
.map(|environment| environment.config().clone())
|
|
.collect::<Vec<_>>(),
|
|
vec![expected_config, expected_local_config]
|
|
);
|
|
assert_eq!(
|
|
attached
|
|
.turn_environments()
|
|
.map(|environment| environment.executor_platform_os.as_deref())
|
|
.collect::<Vec<_>>(),
|
|
vec![Some("windows"), Some(std::env::consts::OS)]
|
|
);
|
|
assert_eq!(attached.to_selections(), vec![remote, local]);
|
|
let environment = attached.primary().expect("remote environment");
|
|
let expected_user_home_dir = PathUri::parse("file:///home/remote").expect("remote home");
|
|
let expected_temporary_directories =
|
|
[PathUri::parse("file:///tmp/remote").expect("remote temporary directory")];
|
|
assert_eq!(
|
|
environment
|
|
.sandbox_context(/*additional_permissions*/ None)
|
|
.policy_context(),
|
|
FileSystemSandboxPolicyContext {
|
|
cwd: environment.cwd(),
|
|
workspace_roots: &[],
|
|
user_home_dir: Some(&expected_user_home_dir),
|
|
temporary_directories: Some(expected_temporary_directories.as_slice()),
|
|
}
|
|
);
|
|
assert_eq!(
|
|
next_starting
|
|
.refresh_readiness()
|
|
.turn_environments()
|
|
.map(|environment| environment.config().clone())
|
|
.collect::<Vec<_>>(),
|
|
vec![
|
|
next_config.clone(),
|
|
EnvironmentConfig {
|
|
windows_sandbox_type: local_sandbox,
|
|
..next_config
|
|
}
|
|
]
|
|
);
|
|
server.await.expect("server task");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn failed_resolution_is_replaced_from_the_environment_manager() {
|
|
let manager = Arc::new(
|
|
EnvironmentManager::create_for_tests(
|
|
Some("http://example.com".to_string()),
|
|
Some(test_runtime_paths()),
|
|
)
|
|
.await,
|
|
);
|
|
let selection = TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
|
cwd: PathUri::from_abs_path(&AbsolutePathBuf::current_dir().expect("cwd")),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
};
|
|
let environments = ThreadEnvironments::new(
|
|
Arc::clone(&manager),
|
|
crate::shell::default_user_shell(),
|
|
ThreadEnvironmentDefaults::new(test_environment_config(), SandboxType::None),
|
|
ShellSnapshot::disabled(),
|
|
TurnEnvironmentSnapshot::default(),
|
|
/*non_blocking_snapshots*/ true,
|
|
);
|
|
environments.update_selections(std::slice::from_ref(&selection));
|
|
let starting = environments.snapshot().await;
|
|
let failed_resolution = environments.state.lock().unwrap().environments[0]
|
|
.resolution
|
|
.clone();
|
|
let error = failed_resolution
|
|
.clone()
|
|
.await
|
|
.err()
|
|
.expect("environment should fail to start");
|
|
// Failed selections must not turn an empty executable snapshot into Full Access.
|
|
for snapshot in [starting.refresh_readiness(), environments.snapshot().await] {
|
|
assert!(!snapshot.has_full_access(AskForApproval::Never, &PermissionProfile::Disabled));
|
|
assert_eq!(snapshot.all_selections(), vec![selection.clone()]);
|
|
assert!(snapshot.ready_environment_handles().is_empty());
|
|
}
|
|
let selected_root = SelectedCapabilityRoot {
|
|
id: "failed-root".to_string(),
|
|
location: CapabilityRootLocation::Environment {
|
|
environment_id: selection.environment_id.clone(),
|
|
path: selection.cwd.clone(),
|
|
},
|
|
};
|
|
assert_eq!(
|
|
environments.inspect_selected_capability_roots(&[selected_root]),
|
|
SelectedCapabilityRootsStatus {
|
|
ready_roots: Vec::new(),
|
|
warnings: vec![format!(
|
|
"selected capability environment `{}` is unavailable: {error}",
|
|
selection.environment_id
|
|
)],
|
|
}
|
|
);
|
|
|
|
let listener = TcpListener::bind("127.0.0.1:0")
|
|
.await
|
|
.expect("bind replacement listener");
|
|
manager
|
|
.upsert_environment(
|
|
REMOTE_ENVIRONMENT_ID.to_string(),
|
|
format!("ws://{}", listener.local_addr().expect("listener address")),
|
|
/*connect_timeout*/ None,
|
|
)
|
|
.expect("replacement environment");
|
|
let next_config = EnvironmentConfig {
|
|
environment_hooks: Default::default(),
|
|
allow_login_shell: false,
|
|
..test_environment_config()
|
|
};
|
|
environments.set_active_thread_defaults(next_config.clone());
|
|
assert!(
|
|
failed_resolution
|
|
.ptr_eq(&environments.state.lock().unwrap().environments[0].resolution),
|
|
"updating environment config must not retry a failed environment"
|
|
);
|
|
environments.update_selections(std::slice::from_ref(&selection));
|
|
|
|
let replacement = environments.snapshot().await;
|
|
let replacement = replacement
|
|
.starting()
|
|
.next()
|
|
.expect("expected the replacement environment to be starting");
|
|
assert_eq!(
|
|
replacement.selection,
|
|
TurnEnvironmentSelection {
|
|
config: EnvironmentConfigState::Ready(next_config),
|
|
..selection.clone()
|
|
}
|
|
);
|
|
assert_eq!(environments.selections(), vec![selection]);
|
|
assert!(!failed_resolution.ptr_eq(&replacement.resolution));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn replacement_environment_events_follow_selected_environment() {
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let first_listener = TcpListener::bind("127.0.0.1:0")
|
|
.await
|
|
.expect("bind first listener");
|
|
let manager = Arc::new(
|
|
EnvironmentManager::create_for_tests(
|
|
Some(format!(
|
|
"ws://{}",
|
|
first_listener.local_addr().expect("first listener address")
|
|
)),
|
|
Some(test_runtime_paths()),
|
|
)
|
|
.await,
|
|
);
|
|
let selection = TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
|
cwd: PathUri::from_abs_path(&cwd),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
};
|
|
let (tx_event, rx_event) = async_channel::unbounded();
|
|
let environments = Arc::new(ThreadEnvironments::new(
|
|
Arc::clone(&manager),
|
|
crate::shell::default_user_shell(),
|
|
ThreadEnvironmentDefaults::new(test_environment_config(), SandboxType::None),
|
|
ShellSnapshot::disabled(),
|
|
TurnEnvironmentSnapshot::default(),
|
|
/*non_blocking_snapshots*/ true,
|
|
));
|
|
environments.activate_configured(tx_event);
|
|
environments.update_selections(std::slice::from_ref(&selection));
|
|
let initial_snapshot = environments.snapshot().await;
|
|
let second_listener = TcpListener::bind("127.0.0.1:0")
|
|
.await
|
|
.expect("bind second listener");
|
|
manager
|
|
.upsert_environment(
|
|
REMOTE_ENVIRONMENT_ID.to_string(),
|
|
format!(
|
|
"ws://{}",
|
|
second_listener
|
|
.local_addr()
|
|
.expect("second listener address")
|
|
),
|
|
/*connect_timeout*/ None,
|
|
)
|
|
.expect("replace environment");
|
|
|
|
environments.update_selections(std::slice::from_ref(&selection));
|
|
let reused_snapshot = environments.snapshot().await;
|
|
environments.update_selections(&[TurnEnvironmentSelection {
|
|
cwd: PathUri::from_abs_path(&cwd.join("changed")),
|
|
..selection
|
|
}]);
|
|
let changed_snapshot = environments.snapshot().await;
|
|
|
|
let initial = initial_snapshot
|
|
.starting()
|
|
.next()
|
|
.expect("initial environment");
|
|
let reused = reused_snapshot
|
|
.starting()
|
|
.next()
|
|
.expect("reused environment");
|
|
let changed = changed_snapshot
|
|
.starting()
|
|
.next()
|
|
.expect("changed environment");
|
|
assert!(initial.resolution.ptr_eq(&reused.resolution));
|
|
assert!(!reused.resolution.ptr_eq(&changed.resolution));
|
|
assert!(Arc::ptr_eq(&initial.attachment, &reused.attachment));
|
|
assert!(initial.attachment.detached.is_cancelled());
|
|
assert!(!changed.attachment.detached.is_cancelled());
|
|
|
|
serve_environment_info(first_listener).await;
|
|
assert!(
|
|
timeout(Duration::from_millis(250), rx_event.recv())
|
|
.await
|
|
.is_err(),
|
|
"old environment event should not be forwarded"
|
|
);
|
|
|
|
serve_environment_info(second_listener).await;
|
|
let event = timeout(Duration::from_secs(5), rx_event.recv())
|
|
.await
|
|
.expect("replacement environment event")
|
|
.expect("event channel");
|
|
let event = match event.msg {
|
|
EventMsg::EnvironmentConnected(event) => event,
|
|
other => panic!("expected connected event, got {other:?}"),
|
|
};
|
|
assert_eq!(
|
|
event,
|
|
EnvironmentConnectionEvent {
|
|
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
|
}
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn inherited_environment_reuses_parent_handle_and_uses_child_config() {
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let selection = TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
|
cwd: PathUri::from_abs_path(&cwd),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
};
|
|
let inherited_environment = Arc::new(
|
|
Environment::create_for_tests(Some("ws://127.0.0.1:8765".to_string()))
|
|
.expect("inherited environment"),
|
|
);
|
|
let mut inherited = TurnEnvironment::new(
|
|
TurnEnvironmentSelection {
|
|
config: EnvironmentConfigState::Ready(test_environment_config()),
|
|
..selection.clone()
|
|
},
|
|
EnvironmentConfigOrigin::Thread,
|
|
Arc::clone(&inherited_environment),
|
|
/*shell*/ None,
|
|
);
|
|
inherited.user_home_dir =
|
|
Some(PathUri::parse("file:///home/inherited").expect("home directory"));
|
|
inherited.temporary_directories = Some(vec![
|
|
PathUri::parse("file:///tmp/inherited").expect("temporary directory"),
|
|
]);
|
|
let manager = Arc::new(environment_manager_without_environments());
|
|
manager
|
|
.upsert_environment(
|
|
REMOTE_ENVIRONMENT_ID.to_string(),
|
|
"ws://127.0.0.1:9876".to_string(),
|
|
/*connect_timeout*/ None,
|
|
)
|
|
.expect("replacement environment");
|
|
let child_config = EnvironmentConfig {
|
|
environment_hooks: Default::default(),
|
|
allow_login_shell: false,
|
|
workspace_roots: Vec::new(),
|
|
windows_sandbox_level: WindowsSandboxLevel::Disabled,
|
|
windows_sandbox_type: codex_protocol::sandbox::SandboxType::None,
|
|
use_legacy_landlock: false,
|
|
permission_profile: PermissionProfileSnapshot::active_with_profile_workspace_roots(
|
|
PermissionProfile::read_only(),
|
|
ActivePermissionProfile::read_only(),
|
|
vec![cwd.join("child-profile-root").into()],
|
|
),
|
|
shell_environment_policy: Default::default(),
|
|
exec_policy: None,
|
|
mcp_policy: None,
|
|
network_policy: None,
|
|
selected_capability_roots: Vec::new(),
|
|
};
|
|
let environments = ThreadEnvironments::new(
|
|
manager,
|
|
crate::shell::default_user_shell(),
|
|
ThreadEnvironmentDefaults::new(child_config.clone(), child_config.windows_sandbox_type),
|
|
ShellSnapshot::disabled(),
|
|
TurnEnvironmentSnapshot {
|
|
environments: vec![TurnEnvironmentState::Ready(inherited)],
|
|
},
|
|
/*non_blocking_snapshots*/ false,
|
|
);
|
|
|
|
environments.update_selections(std::slice::from_ref(&selection));
|
|
let snapshot = environments.snapshot().await;
|
|
|
|
let inherited = snapshot.primary().expect("inherited environment");
|
|
assert!(Arc::ptr_eq(inherited.environment(), &inherited_environment));
|
|
assert_eq!(inherited.config(), &child_config);
|
|
assert_eq!(
|
|
inherited
|
|
.sandbox_context(/*additional_permissions*/ None)
|
|
.user_home_dir,
|
|
Some(PathUri::parse("file:///home/inherited").expect("home directory")),
|
|
);
|
|
assert_eq!(
|
|
inherited
|
|
.sandbox_context(/*additional_permissions*/ None)
|
|
.temporary_directories
|
|
.as_deref(),
|
|
Some(
|
|
[PathUri::parse("file:///tmp/inherited").expect("temporary directory")].as_slice()
|
|
)
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn owner_environment_config_is_inherited_and_reset_for_new_cwd() {
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let selection = TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
|
cwd: PathUri::from_abs_path(&cwd),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
};
|
|
let root = |id: &str| SelectedCapabilityRoot {
|
|
id: id.to_string(),
|
|
location: CapabilityRootLocation::Environment {
|
|
environment_id: selection.environment_id.clone(),
|
|
path: selection.cwd.clone(),
|
|
},
|
|
};
|
|
let manager = Arc::new(EnvironmentManager::default_for_tests());
|
|
let parent = ThreadEnvironments::new(
|
|
Arc::clone(&manager),
|
|
crate::shell::default_user_shell(),
|
|
ThreadEnvironmentDefaults::new(test_environment_config(), SandboxType::WindowsMxc),
|
|
ShellSnapshot::disabled(),
|
|
TurnEnvironmentSnapshot::default(),
|
|
/*non_blocking_snapshots*/ false,
|
|
);
|
|
parent.update_selections(std::slice::from_ref(&selection));
|
|
parent.snapshot().await;
|
|
let parent_owner_config = EnvironmentConfig {
|
|
environment_hooks: Default::default(),
|
|
allow_login_shell: false,
|
|
workspace_roots: selection.workspace_roots.clone(),
|
|
windows_sandbox_level: WindowsSandboxLevel::Disabled,
|
|
windows_sandbox_type: codex_protocol::sandbox::SandboxType::None,
|
|
use_legacy_landlock: false,
|
|
permission_profile: PermissionProfileSnapshot::legacy(PermissionProfile::read_only()),
|
|
shell_environment_policy: Default::default(),
|
|
exec_policy: None,
|
|
mcp_policy: None,
|
|
network_policy: None,
|
|
selected_capability_roots: vec![root("parent-root")],
|
|
};
|
|
let mut owner_selection = selection.clone();
|
|
owner_selection.config = EnvironmentConfigState::Ready(parent_owner_config.clone());
|
|
parent.update_selections(std::slice::from_ref(&owner_selection));
|
|
|
|
let child_thread_config = EnvironmentConfig {
|
|
windows_sandbox_type: SandboxType::WindowsMxc,
|
|
permission_profile: PermissionProfileSnapshot::legacy(
|
|
PermissionProfile::workspace_write(),
|
|
),
|
|
..test_environment_config()
|
|
};
|
|
parent.set_active_thread_defaults(child_thread_config.clone());
|
|
assert_eq!(
|
|
parent
|
|
.snapshot()
|
|
.await
|
|
.primary()
|
|
.expect("owner-configured environment")
|
|
.config(),
|
|
&parent_owner_config
|
|
);
|
|
let child = ThreadEnvironments::new(
|
|
manager,
|
|
crate::shell::default_user_shell(),
|
|
ThreadEnvironmentDefaults::new(child_thread_config.clone(), SandboxType::WindowsMxc),
|
|
ShellSnapshot::disabled(),
|
|
parent.snapshot().await,
|
|
/*non_blocking_snapshots*/ false,
|
|
);
|
|
child.activate_configured(async_channel::unbounded().0);
|
|
let child_snapshot = child.snapshot().await;
|
|
let child_environment = child_snapshot.primary().expect("child environment");
|
|
|
|
let cleared_parent_owner_config = EnvironmentConfig {
|
|
environment_hooks: Default::default(),
|
|
selected_capability_roots: Vec::new(),
|
|
..parent_owner_config.clone()
|
|
};
|
|
owner_selection.config = EnvironmentConfigState::Ready(cleared_parent_owner_config.clone());
|
|
parent.update_selections(std::slice::from_ref(&owner_selection));
|
|
let cleared_snapshot = parent.snapshot().await;
|
|
assert_eq!(
|
|
cleared_snapshot
|
|
.primary()
|
|
.expect("environment with cleared roots")
|
|
.config(),
|
|
&cleared_parent_owner_config
|
|
);
|
|
assert_eq!(child_environment.config(), &parent_owner_config);
|
|
|
|
let changed_selection = TurnEnvironmentSelection {
|
|
cwd: PathUri::from_abs_path(&cwd.join("changed")),
|
|
..selection
|
|
};
|
|
parent.update_selections(std::slice::from_ref(&changed_selection));
|
|
let changed_snapshot = parent.snapshot().await;
|
|
let changed_environment = changed_snapshot.primary().expect("changed environment");
|
|
assert_eq!(changed_environment.config(), &child_thread_config);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn single_local_environment_cwd_requires_exactly_one_local_environment() {
|
|
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
|
|
let cwd_uri = PathUri::from_abs_path(&cwd);
|
|
let remote_cwd_uri = PathUri::from_abs_path(&cwd.join("remote-cwd"));
|
|
let local_manager = Arc::new(EnvironmentManager::default_for_tests());
|
|
let local = resolve_turn_environments(
|
|
Arc::clone(&local_manager),
|
|
&[TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
|
cwd: cwd_uri.clone(),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
}],
|
|
)
|
|
.await;
|
|
let local = local.snapshot().await;
|
|
let remote_environment = Arc::new(
|
|
Environment::create_for_tests(Some("ws://127.0.0.1:8765".to_string()))
|
|
.expect("remote environment"),
|
|
);
|
|
let remote = TurnEnvironmentSnapshot {
|
|
environments: vec![TurnEnvironmentState::Ready(TurnEnvironment::new(
|
|
TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
|
cwd: remote_cwd_uri.clone(),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::Ready(test_environment_config()),
|
|
},
|
|
EnvironmentConfigOrigin::Thread,
|
|
remote_environment.clone(),
|
|
/*shell*/ None,
|
|
))],
|
|
};
|
|
let multiple = TurnEnvironmentSnapshot {
|
|
environments: vec![
|
|
TurnEnvironmentState::Ready(TurnEnvironment::new(
|
|
TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: REMOTE_ENVIRONMENT_ID.to_string(),
|
|
cwd: remote_cwd_uri,
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::Ready(test_environment_config()),
|
|
},
|
|
EnvironmentConfigOrigin::Thread,
|
|
remote_environment,
|
|
/*shell*/ None,
|
|
)),
|
|
TurnEnvironmentState::Ready(local.primary().expect("local environment").clone()),
|
|
],
|
|
};
|
|
|
|
assert_eq!(local.single_local_environment_cwd(), Some(cwd.clone()));
|
|
assert_eq!(remote.single_local_environment_cwd(), None);
|
|
assert_eq!(multiple.single_local_environment_cwd(), None);
|
|
assert_eq!(multiple.local_environment_cwd(), Some(cwd));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn local_environment_cwd_uses_starting_local_selection() {
|
|
let cwd = AbsolutePathBuf::current_dir()
|
|
.expect("cwd")
|
|
.join("starting-local");
|
|
let (attachment, _guard, _) = EnvironmentAttachment::new(
|
|
LOCAL_ENVIRONMENT_ID,
|
|
Arc::new(
|
|
Environment::create_for_tests(/*exec_server_url*/ None).expect("local environment"),
|
|
),
|
|
);
|
|
let snapshot = TurnEnvironmentSnapshot {
|
|
environments: vec![TurnEnvironmentState::Starting(StartingTurnEnvironment {
|
|
selection: TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
|
cwd: PathUri::from_abs_path(&cwd),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::Ready(test_environment_config()),
|
|
},
|
|
config_origin: EnvironmentConfigOrigin::Thread,
|
|
resolution: futures::future::pending().boxed().shared(),
|
|
attachment,
|
|
owner_config_result: None,
|
|
})],
|
|
};
|
|
|
|
assert_eq!(snapshot.local_environment_cwd(), Some(cwd));
|
|
}
|
|
}
|