mirror of
https://github.com/openai/codex.git
synced 2026-09-04 15:08:45 +00:00
## Why This stack crosses World State, executor skills, selected plugin metadata, MCP processes, connectors, dynamic environments, and resume. This PR adds two end-to-end scenarios that validate those pieces together. Both tests enable `deferred_executor`, so they exercise the real delayed-environment path. ## Scenario 1: availability across turns and resume ```text 1. Start a thread with one selected plugin root bound to E1. 2. E1 is unavailable. - executor skill is absent - selected MCP is absent - connector has no selected-plugin attribution 3. Start E1 and register the same stable environment ID. 4. Start a new turn. - the executor skill appears through World State - its body beats a colliding host skill - the selected MCP tool is advertised and executes inside E1 - the connector is attributed to the selected plugin 5. Start another turn without changing E1. - the MCP PID stays the same, proving runtime reuse 6. Restart app-server and resume the thread. - durable selected-root intent is restored - skills, MCP, and connector attribution are restored - a new MCP PID proves ephemeral process state was rebuilt ``` ## Scenario 2: availability changes inside one turn ```text 1. Start a turn while E1 is unavailable. 2. The first model sample sees no executor skill, MCP, or selected connector. 3. The turn pauses on request_user_input. 4. Start E1 and register it while that same turn is still active. 5. Continue the turn. 6. The very next model sample sees: - the executor skill catalog - the selected MCP tool - selected-plugin connector attribution 7. The model calls the MCP, and its output proves execution happened inside E1. ``` This second scenario specifically protects the aeon-style behavior: capability state is captured again for every sampling step, not only at the next user turn. ## Scope These are integration tests only. They do not add a combinatorial matrix for unsupported plugin-file mutation, environment generations, transport disconnects, or delayed `required = true` executor MCPs.
482 lines
16 KiB
Rust
482 lines
16 KiB
Rust
use base64::Engine as _;
|
|
use base64::engine::general_purpose::STANDARD;
|
|
use codex_utils_path_uri::PathUri;
|
|
use tokio::io;
|
|
use tracing::trace;
|
|
|
|
use crate::CopyOptions;
|
|
use crate::CreateDirectoryOptions;
|
|
use crate::ExecServerError;
|
|
use crate::ExecutorFileSystem;
|
|
use crate::ExecutorFileSystemFuture;
|
|
use crate::FileMetadata;
|
|
use crate::FileSystemReadStream;
|
|
use crate::FileSystemResult;
|
|
use crate::FileSystemSandboxContext;
|
|
use crate::ReadDirectoryEntry;
|
|
use crate::RemoveOptions;
|
|
use crate::WalkOptions;
|
|
use crate::WalkOutcome;
|
|
use crate::client::LazyRemoteExecServerClient;
|
|
use crate::protocol::FsCanonicalizeParams;
|
|
use crate::protocol::FsCopyParams;
|
|
use crate::protocol::FsCreateDirectoryParams;
|
|
use crate::protocol::FsGetMetadataParams;
|
|
use crate::protocol::FsReadDirectoryParams;
|
|
use crate::protocol::FsReadFileParams;
|
|
use crate::protocol::FsRemoveParams;
|
|
use crate::protocol::FsWalkParams;
|
|
use crate::protocol::FsWriteFileParams;
|
|
|
|
const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
|
|
const METHOD_NOT_FOUND_ERROR_CODE: i64 = -32601;
|
|
const NOT_FOUND_ERROR_CODE: i64 = -32004;
|
|
|
|
#[path = "remote_file_stream.rs"]
|
|
mod file_stream;
|
|
|
|
pub(crate) struct RemoteFileSystem {
|
|
client: LazyRemoteExecServerClient,
|
|
}
|
|
|
|
impl RemoteFileSystem {
|
|
pub(crate) fn new(client: LazyRemoteExecServerClient) -> Self {
|
|
trace!("remote fs new");
|
|
Self { client }
|
|
}
|
|
|
|
async fn canonicalize(
|
|
&self,
|
|
path: &PathUri,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<PathUri> {
|
|
trace!("remote fs canonicalize");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
let response = client
|
|
.fs_canonicalize(FsCanonicalizeParams {
|
|
path: path.clone(),
|
|
sandbox: remote_sandbox_context(sandbox),
|
|
})
|
|
.await
|
|
.map_err(map_remote_error)?;
|
|
Ok(response.path)
|
|
}
|
|
|
|
async fn read_file(
|
|
&self,
|
|
path: &PathUri,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<Vec<u8>> {
|
|
trace!("remote fs read_file");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
let response = client
|
|
.fs_read_file(FsReadFileParams {
|
|
path: path.clone(),
|
|
sandbox: remote_sandbox_context(sandbox),
|
|
})
|
|
.await
|
|
.map_err(map_remote_error)?;
|
|
STANDARD.decode(response.data_base64).map_err(|err| {
|
|
io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
format!("remote fs/readFile returned invalid base64 dataBase64: {err}"),
|
|
)
|
|
})
|
|
}
|
|
|
|
async fn read_file_stream(
|
|
&self,
|
|
path: &PathUri,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<FileSystemReadStream> {
|
|
if sandbox.is_some_and(FileSystemSandboxContext::should_run_in_sandbox) {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::Unsupported,
|
|
"streaming file reads do not support platform sandboxing",
|
|
));
|
|
}
|
|
trace!("remote fs read_file_stream");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
file_stream::open(client, path.clone(), remote_sandbox_context(sandbox)).await
|
|
}
|
|
|
|
async fn write_file(
|
|
&self,
|
|
path: &PathUri,
|
|
contents: Vec<u8>,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<()> {
|
|
trace!("remote fs write_file");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
client
|
|
.fs_write_file(FsWriteFileParams {
|
|
path: path.clone(),
|
|
data_base64: STANDARD.encode(contents),
|
|
sandbox: remote_sandbox_context(sandbox),
|
|
})
|
|
.await
|
|
.map_err(map_remote_error)?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn create_directory(
|
|
&self,
|
|
path: &PathUri,
|
|
options: CreateDirectoryOptions,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<()> {
|
|
trace!("remote fs create_directory");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
client
|
|
.fs_create_directory(FsCreateDirectoryParams {
|
|
path: path.clone(),
|
|
recursive: Some(options.recursive),
|
|
sandbox: remote_sandbox_context(sandbox),
|
|
})
|
|
.await
|
|
.map_err(map_remote_error)?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn get_metadata(
|
|
&self,
|
|
path: &PathUri,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<FileMetadata> {
|
|
trace!("remote fs get_metadata");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
let response = client
|
|
.fs_get_metadata(FsGetMetadataParams {
|
|
path: path.clone(),
|
|
sandbox: remote_sandbox_context(sandbox),
|
|
})
|
|
.await
|
|
.map_err(map_remote_error)?;
|
|
Ok(FileMetadata {
|
|
is_directory: response.is_directory,
|
|
is_file: response.is_file,
|
|
is_symlink: response.is_symlink,
|
|
size: response.size,
|
|
created_at_ms: response.created_at_ms,
|
|
modified_at_ms: response.modified_at_ms,
|
|
})
|
|
}
|
|
|
|
async fn read_directory(
|
|
&self,
|
|
path: &PathUri,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<Vec<ReadDirectoryEntry>> {
|
|
trace!("remote fs read_directory");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
let response = client
|
|
.fs_read_directory(FsReadDirectoryParams {
|
|
path: path.clone(),
|
|
sandbox: remote_sandbox_context(sandbox),
|
|
})
|
|
.await
|
|
.map_err(map_remote_error)?;
|
|
Ok(response
|
|
.entries
|
|
.into_iter()
|
|
.map(|entry| ReadDirectoryEntry {
|
|
file_name: entry.file_name,
|
|
is_directory: entry.is_directory,
|
|
is_file: entry.is_file,
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
async fn walk(
|
|
&self,
|
|
path: &PathUri,
|
|
options: WalkOptions,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<WalkOutcome> {
|
|
trace!("remote fs walk");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
let response = match client
|
|
.fs_walk(FsWalkParams {
|
|
path: path.clone(),
|
|
options,
|
|
sandbox: remote_sandbox_context(sandbox),
|
|
})
|
|
.await
|
|
{
|
|
Ok(response) => response,
|
|
Err(ExecServerError::Server {
|
|
code: METHOD_NOT_FOUND_ERROR_CODE,
|
|
..
|
|
}) => {
|
|
return <Self as ExecutorFileSystem>::walk_via_directory_reads(
|
|
self, path, options, sandbox,
|
|
)
|
|
.await;
|
|
}
|
|
Err(error) => return Err(map_remote_error(error)),
|
|
};
|
|
Ok(response)
|
|
}
|
|
|
|
async fn remove(
|
|
&self,
|
|
path: &PathUri,
|
|
options: RemoveOptions,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<()> {
|
|
trace!("remote fs remove");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
client
|
|
.fs_remove(FsRemoveParams {
|
|
path: path.clone(),
|
|
recursive: Some(options.recursive),
|
|
force: Some(options.force),
|
|
sandbox: remote_sandbox_context(sandbox),
|
|
})
|
|
.await
|
|
.map_err(map_remote_error)?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn copy(
|
|
&self,
|
|
source_path: &PathUri,
|
|
destination_path: &PathUri,
|
|
options: CopyOptions,
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> FileSystemResult<()> {
|
|
trace!("remote fs copy");
|
|
let client = self.client.get().await.map_err(map_remote_error)?;
|
|
client
|
|
.fs_copy(FsCopyParams {
|
|
source_path: source_path.clone(),
|
|
destination_path: destination_path.clone(),
|
|
recursive: options.recursive,
|
|
sandbox: remote_sandbox_context(sandbox),
|
|
})
|
|
.await
|
|
.map_err(map_remote_error)?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
impl ExecutorFileSystem for RemoteFileSystem {
|
|
fn canonicalize<'a>(
|
|
&'a self,
|
|
path: &'a PathUri,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, PathUri> {
|
|
Box::pin(RemoteFileSystem::canonicalize(self, path, sandbox))
|
|
}
|
|
|
|
fn read_file<'a>(
|
|
&'a self,
|
|
path: &'a PathUri,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, Vec<u8>> {
|
|
Box::pin(RemoteFileSystem::read_file(self, path, sandbox))
|
|
}
|
|
|
|
fn read_file_stream<'a>(
|
|
&'a self,
|
|
path: &'a PathUri,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, FileSystemReadStream> {
|
|
Box::pin(RemoteFileSystem::read_file_stream(self, path, sandbox))
|
|
}
|
|
|
|
fn write_file<'a>(
|
|
&'a self,
|
|
path: &'a PathUri,
|
|
contents: Vec<u8>,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, ()> {
|
|
Box::pin(RemoteFileSystem::write_file(self, path, contents, sandbox))
|
|
}
|
|
|
|
fn create_directory<'a>(
|
|
&'a self,
|
|
path: &'a PathUri,
|
|
options: CreateDirectoryOptions,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, ()> {
|
|
Box::pin(RemoteFileSystem::create_directory(
|
|
self, path, options, sandbox,
|
|
))
|
|
}
|
|
|
|
fn get_metadata<'a>(
|
|
&'a self,
|
|
path: &'a PathUri,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, FileMetadata> {
|
|
Box::pin(RemoteFileSystem::get_metadata(self, path, sandbox))
|
|
}
|
|
|
|
fn read_directory<'a>(
|
|
&'a self,
|
|
path: &'a PathUri,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, Vec<ReadDirectoryEntry>> {
|
|
Box::pin(RemoteFileSystem::read_directory(self, path, sandbox))
|
|
}
|
|
|
|
fn walk<'a>(
|
|
&'a self,
|
|
path: &'a PathUri,
|
|
options: WalkOptions,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, WalkOutcome> {
|
|
Box::pin(RemoteFileSystem::walk(self, path, options, sandbox))
|
|
}
|
|
|
|
fn remove<'a>(
|
|
&'a self,
|
|
path: &'a PathUri,
|
|
options: RemoveOptions,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, ()> {
|
|
Box::pin(RemoteFileSystem::remove(self, path, options, sandbox))
|
|
}
|
|
|
|
fn copy<'a>(
|
|
&'a self,
|
|
source_path: &'a PathUri,
|
|
destination_path: &'a PathUri,
|
|
options: CopyOptions,
|
|
sandbox: Option<&'a FileSystemSandboxContext>,
|
|
) -> ExecutorFileSystemFuture<'a, ()> {
|
|
Box::pin(RemoteFileSystem::copy(
|
|
self,
|
|
source_path,
|
|
destination_path,
|
|
options,
|
|
sandbox,
|
|
))
|
|
}
|
|
}
|
|
|
|
fn remote_sandbox_context(
|
|
sandbox: Option<&FileSystemSandboxContext>,
|
|
) -> Option<FileSystemSandboxContext> {
|
|
sandbox
|
|
.cloned()
|
|
.map(FileSystemSandboxContext::drop_cwd_if_unused)
|
|
}
|
|
|
|
fn map_remote_error(error: ExecServerError) -> io::Error {
|
|
match error {
|
|
ExecServerError::Server { code, message } if code == NOT_FOUND_ERROR_CODE => {
|
|
io::Error::new(io::ErrorKind::NotFound, message)
|
|
}
|
|
ExecServerError::Server { code, message } if code == INVALID_REQUEST_ERROR_CODE => {
|
|
io::Error::new(io::ErrorKind::InvalidInput, message)
|
|
}
|
|
ExecServerError::Server { message, .. } => io::Error::other(message),
|
|
ExecServerError::Closed | ExecServerError::Disconnected(_) => {
|
|
io::Error::new(io::ErrorKind::BrokenPipe, "exec-server transport closed")
|
|
}
|
|
_ => io::Error::other(error.to_string()),
|
|
}
|
|
}
|
|
|
|
#[cfg(all(test, any(unix, windows)))]
|
|
#[path = "remote_file_system_path_uri_tests.rs"]
|
|
mod path_uri_tests;
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use codex_protocol::models::PermissionProfile;
|
|
use codex_protocol::permissions::FileSystemAccessMode;
|
|
use codex_protocol::permissions::FileSystemPath;
|
|
use codex_protocol::permissions::FileSystemSandboxEntry;
|
|
use codex_protocol::permissions::FileSystemSandboxPolicy;
|
|
use codex_protocol::permissions::FileSystemSpecialPath;
|
|
use codex_protocol::permissions::NetworkSandboxPolicy;
|
|
use codex_utils_absolute_path::AbsolutePathBuf;
|
|
use codex_utils_path_uri::PathUri;
|
|
use pretty_assertions::assert_eq;
|
|
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn remote_sandbox_context_drops_unused_cwd() {
|
|
let policy = FileSystemSandboxPolicy::restricted(vec![FileSystemSandboxEntry {
|
|
path: FileSystemPath::Path {
|
|
path: absolute_test_path("remote-root"),
|
|
},
|
|
access: FileSystemAccessMode::Read,
|
|
}]);
|
|
let permissions =
|
|
PermissionProfile::from_runtime_permissions(&policy, NetworkSandboxPolicy::Restricted);
|
|
let sandbox_context = FileSystemSandboxContext::from_permission_profile_with_cwd(
|
|
permissions,
|
|
path_uri("host-checkout"),
|
|
);
|
|
|
|
let remote_context =
|
|
remote_sandbox_context(Some(&sandbox_context)).expect("remote sandbox context");
|
|
|
|
assert_eq!(remote_context.cwd, None);
|
|
}
|
|
|
|
#[test]
|
|
fn remote_sandbox_context_preserves_required_cwd() {
|
|
let policy = FileSystemSandboxPolicy::restricted(vec![FileSystemSandboxEntry {
|
|
path: FileSystemPath::Special {
|
|
value: FileSystemSpecialPath::project_roots(/*subpath*/ None),
|
|
},
|
|
access: FileSystemAccessMode::Write,
|
|
}]);
|
|
let permissions =
|
|
PermissionProfile::from_runtime_permissions(&policy, NetworkSandboxPolicy::Restricted);
|
|
let cwd = path_uri("host-checkout");
|
|
let sandbox_context =
|
|
FileSystemSandboxContext::from_permission_profile_with_cwd(permissions, cwd.clone());
|
|
|
|
let remote_context =
|
|
remote_sandbox_context(Some(&sandbox_context)).expect("remote sandbox context");
|
|
|
|
assert_eq!(remote_context.cwd, Some(cwd));
|
|
}
|
|
|
|
#[test]
|
|
fn transport_errors_map_to_broken_pipe() {
|
|
let errors = [
|
|
ExecServerError::Closed,
|
|
ExecServerError::Disconnected("exec-server transport disconnected".to_string()),
|
|
];
|
|
|
|
let mapped_errors = errors
|
|
.into_iter()
|
|
.map(|error| {
|
|
let error = map_remote_error(error);
|
|
(error.kind(), error.to_string())
|
|
})
|
|
.collect::<Vec<_>>();
|
|
|
|
assert_eq!(
|
|
mapped_errors,
|
|
vec![
|
|
(
|
|
io::ErrorKind::BrokenPipe,
|
|
"exec-server transport closed".to_string()
|
|
),
|
|
(
|
|
io::ErrorKind::BrokenPipe,
|
|
"exec-server transport closed".to_string()
|
|
),
|
|
]
|
|
);
|
|
}
|
|
|
|
fn absolute_test_path(name: &str) -> AbsolutePathBuf {
|
|
let path = std::env::temp_dir().join(name);
|
|
AbsolutePathBuf::from_absolute_path(&path).expect("absolute path")
|
|
}
|
|
|
|
fn path_uri(name: &str) -> PathUri {
|
|
PathUri::from_abs_path(&absolute_test_path(name))
|
|
}
|
|
}
|