openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-01 19:33:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use Environment as the exec and fs gateway

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
starr-openai
2026-03-19 18:41:58 +00:00
co-authored by Codex
parent e0a7c18424
commit a4358f2c4f
12 changed files with 239 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/fs_api.rs
+1 -1
View File
@@ -34,7 +34,7 @@ pub(crate) struct FsApi {
impl Default for FsApi {
fn default() -> Self {
Self {
file_system: Arc::new(Environment::default().get_filesystem()),
file_system: Environment::default().get_filesystem(),
}
}
}
codex-rs/core/src/codex.rs
+6 -9
View File
@@ -314,7 +314,7 @@ use crate::turn_timing::TurnTimingState;
use crate::turn_timing::record_turn_ttfm_metric;
use crate::turn_timing::record_turn_ttft_metric;
use crate::unified_exec::UnifiedExecProcessManager;
use crate::unified_exec::unified_exec_session_factory_for_config;
use crate::unified_exec::unified_exec_session_factory_for_environment;
use crate::util::backoff;
use crate::windows_sandbox::WindowsSandboxLevelExt;
use codex_async_utils::OrCancelExt;
@@ -1774,11 +1774,10 @@ impl Session {
});
}
let unified_exec_session_factory = unified_exec_session_factory_for_config(
config.as_ref(),
None,
)
.await?;
let environment =
Arc::new(Environment::create(config.experimental_exec_server_url.clone()).await?);
let unified_exec_session_factory =
unified_exec_session_factory_for_environment(environment.as_ref());
let services = SessionServices {
// Initialize the MCP connection manager with an uninitialized
@@ -1834,9 +1833,7 @@ impl Session {
code_mode_service: crate::tools::code_mode::CodeModeService::new(
config.js_repl_node_path.clone(),
),
environment: Arc::new(
Environment::create(config.experimental_exec_server_url.clone()).await?,
),
environment,
};
let js_repl = Arc::new(JsReplHandle::with_node_path(
config.js_repl_node_path.clone(),
codex-rs/core/src/config/mod.rs
-20
View File
@@ -543,14 +543,6 @@ pub struct Config {
/// If set to `true`, used only the experimental unified exec tool.
pub use_experimental_unified_exec_tool: bool,
/// When `true`, route unified-exec process launches through `codex-exec-server`
/// instead of spawning them directly in-process.
pub experimental_unified_exec_use_exec_server: bool,
/// When `true`, start a session-scoped local `codex-exec-server` subprocess
/// during session startup and route unified-exec calls through it.
pub experimental_unified_exec_spawn_local_exec_server: bool,
/// Maximum poll window for background terminal output (`write_stdin`), in milliseconds.
/// Default: `300000` (5 minutes).
pub background_terminal_max_timeout: u64,
@@ -1521,8 +1513,6 @@ pub struct ConfigToml {
pub experimental_instructions_file: Option<AbsolutePathBuf>,
pub experimental_compact_prompt_file: Option<AbsolutePathBuf>,
pub experimental_use_unified_exec_tool: Option<bool>,
pub experimental_unified_exec_use_exec_server: Option<bool>,
pub experimental_unified_exec_spawn_local_exec_server: Option<bool>,
pub experimental_use_freeform_apply_patch: Option<bool>,
/// Preferred OSS provider for local models, e.g. "lmstudio" or "ollama".
pub oss_provider: Option<String>,
@@ -2492,14 +2482,6 @@ impl Config {
let include_apply_patch_tool_flag = features.enabled(Feature::ApplyPatchFreeform);
let use_experimental_unified_exec_tool = features.enabled(Feature::UnifiedExec);
let experimental_unified_exec_use_exec_server = config_profile
.experimental_unified_exec_use_exec_server
.or(cfg.experimental_unified_exec_use_exec_server)
.unwrap_or(false);
let experimental_unified_exec_spawn_local_exec_server = config_profile
.experimental_unified_exec_spawn_local_exec_server
.or(cfg.experimental_unified_exec_spawn_local_exec_server)
.unwrap_or(false);
let forced_chatgpt_workspace_id =
cfg.forced_chatgpt_workspace_id.as_ref().and_then(|value| {
@@ -2795,8 +2777,6 @@ impl Config {
web_search_mode: constrained_web_search_mode.value,
web_search_config,
use_experimental_unified_exec_tool,
experimental_unified_exec_use_exec_server,
experimental_unified_exec_spawn_local_exec_server,
background_terminal_max_timeout,
ghost_snapshot,
features,
codex-rs/core/src/config/profile.rs
-2
View File
@@ -49,8 +49,6 @@ pub struct ConfigProfile {
pub experimental_compact_prompt_file: Option<AbsolutePathBuf>,
pub include_apply_patch_tool: Option<bool>,
pub experimental_use_unified_exec_tool: Option<bool>,
pub experimental_unified_exec_use_exec_server: Option<bool>,
pub experimental_unified_exec_spawn_local_exec_server: Option<bool>,
pub experimental_use_freeform_apply_patch: Option<bool>,
pub tools_view_image: Option<bool>,
pub tools: Option<ToolsToml>,
codex-rs/core/src/tools/handlers/view_image.rs
-1
View File
@@ -1,5 +1,4 @@
use async_trait::async_trait;
use codex_exec_server::ExecutorFileSystem;
use codex_protocol::models::FunctionCallOutputBody;
use codex_protocol::models::FunctionCallOutputContentItem;
use codex_protocol::models::FunctionCallOutputPayload;
codex-rs/core/src/unified_exec/backend.rs
+8 -59
View File
@@ -1,15 +1,10 @@
use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
use codex_exec_server::Environment;
use codex_exec_server::ExecServerClient;
use codex_exec_server::ExecServerClientConnectOptions;
use codex_exec_server::ExecServerLaunchCommand;
use codex_exec_server::SpawnedExecServer;
use codex_exec_server::spawn_local_exec_server;
use tracing::debug;
use crate::config::Config;
use crate::exec::SandboxType;
use crate::sandboxing::ExecRequest;
use crate::unified_exec::SpawnLifecycleHandle;
@@ -51,32 +46,18 @@ impl UnifiedExecSessionFactory for LocalUnifiedExecSessionFactory {
pub(crate) struct ExecServerUnifiedExecSessionFactory {
client: ExecServerClient,
_spawned_server: Option<Arc<SpawnedExecServer>>,
}
impl std::fmt::Debug for ExecServerUnifiedExecSessionFactory {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ExecServerUnifiedExecSessionFactory")
.field("owns_spawned_server", &self._spawned_server.is_some())
.finish_non_exhaustive()
}
}
impl ExecServerUnifiedExecSessionFactory {
pub(crate) fn from_client(client: ExecServerClient) -> UnifiedExecSessionFactoryHandle {
Arc::new(Self {
client,
_spawned_server: None,
})
}
pub(crate) fn from_spawned_server(
spawned_server: Arc<SpawnedExecServer>,
) -> UnifiedExecSessionFactoryHandle {
Arc::new(Self {
client: spawned_server.client().clone(),
_spawned_server: Some(spawned_server),
})
Arc::new(Self { client })
}
}
@@ -118,45 +99,13 @@ impl UnifiedExecSessionFactory for ExecServerUnifiedExecSessionFactory {
}
}
pub(crate) async fn unified_exec_session_factory_for_config(
config: &Config,
local_exec_server_command: Option<ExecServerLaunchCommand>,
) -> Result<UnifiedExecSessionFactoryHandle, UnifiedExecError> {
if !config.experimental_unified_exec_use_exec_server {
return Ok(local_unified_exec_session_factory());
}
if config.experimental_unified_exec_spawn_local_exec_server {
let command = local_exec_server_command.unwrap_or_else(default_local_exec_server_command);
let spawned_server =
spawn_local_exec_server(command, ExecServerClientConnectOptions::default())
.await
.map_err(|err| UnifiedExecError::create_process(err.to_string()))?;
return Ok(ExecServerUnifiedExecSessionFactory::from_spawned_server(
Arc::new(spawned_server),
));
}
let client = ExecServerClient::connect_in_process(ExecServerClientConnectOptions::default())
.await
.map_err(|err| UnifiedExecError::create_process(err.to_string()))?;
Ok(ExecServerUnifiedExecSessionFactory::from_client(client))
}
fn default_local_exec_server_command() -> ExecServerLaunchCommand {
let binary_name = if cfg!(windows) {
"codex-exec-server.exe"
pub(crate) fn unified_exec_session_factory_for_environment(
environment: &Environment,
) -> UnifiedExecSessionFactoryHandle {
if let Some(client) = environment.exec_server_client() {
ExecServerUnifiedExecSessionFactory::from_client(client)
} else {
"codex-exec-server"
};
let program = std::env::current_exe()
.ok()
.map(|current_exe| current_exe.with_file_name(binary_name))
.filter(|candidate| candidate.exists())
.unwrap_or_else(|| PathBuf::from(binary_name));
ExecServerLaunchCommand {
program,
args: Vec::new(),
local_unified_exec_session_factory()
}
}
codex-rs/core/src/unified_exec/mod.rs
+1 -1
View File
@@ -50,7 +50,7 @@ pub(crate) fn set_deterministic_process_ids_for_tests(enabled: bool) {
pub(crate) use backend::UnifiedExecSessionFactoryHandle;
pub(crate) use backend::local_unified_exec_session_factory;
pub(crate) use backend::unified_exec_session_factory_for_config;
pub(crate) use backend::unified_exec_session_factory_for_environment;
pub(crate) use errors::UnifiedExecError;
pub(crate) use process::NoopSpawnLifecycle;
#[cfg(unix)]
codex-rs/core/src/unified_exec/mod_tests.rs
+9 -48
View File
@@ -11,13 +11,10 @@ use crate::tools::context::ExecCommandToolOutput;
use crate::unified_exec::ExecCommandRequest;
use crate::unified_exec::WriteStdinRequest;
use core_test_support::skip_if_sandbox;
use std::sync::Arc;
use std::path::Path;
use std::path::PathBuf;
use std::process::Command;
use tokio::time::Duration;
use std::sync::Arc;
use tempfile::TempDir;
use toml::Value as TomlValue;
use tokio::time::Duration;
fn test_exec_request(command: Vec<String>, cwd: &Path) -> crate::sandboxing::ExecRequest {
crate::sandboxing::ExecRequest {
@@ -253,61 +250,25 @@ async fn unified_exec_timeouts() -> anyhow::Result<()> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_can_spawn_a_local_exec_server_backend() -> anyhow::Result<()> {
async fn unified_exec_can_use_remote_exec_server_environment() -> anyhow::Result<()> {
skip_if_sandbox!(Ok(()));
let codex_home = TempDir::new()?;
let cwd = TempDir::new()?;
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.cli_overrides(vec![
(
"experimental_unified_exec_use_exec_server".to_string(),
TomlValue::Boolean(true),
),
(
"experimental_unified_exec_spawn_local_exec_server".to_string(),
TomlValue::Boolean(true),
),
])
.harness_overrides(ConfigOverrides {
cwd: Some(cwd.path().to_path_buf()),
..Default::default()
})
.build()
.await?;
let workspace_root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.parent()
.expect("core crate should be under codex-rs")
.to_path_buf();
let cargo = PathBuf::from(env!("CARGO"));
let build_status = Command::new(&cargo)
.current_dir(&workspace_root)
.args([
"build",
"-p",
"codex-exec-server",
"--bin",
"codex-exec-server",
])
.status()?;
assert!(build_status.success(), "failed to build codex-exec-server");
let target_dir = std::env::var_os("CARGO_TARGET_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| workspace_root.join("target"));
let binary_name = if cfg!(windows) {
"codex-exec-server.exe"
} else {
"codex-exec-server"
};
let session_factory = unified_exec_session_factory_for_config(
&config,
Some(codex_exec_server::ExecServerLaunchCommand {
program: target_dir.join("debug").join(binary_name),
args: Vec::new(),
}),
let client = codex_exec_server::ExecServerClient::connect_in_process(
codex_exec_server::ExecServerClientConnectOptions::default(),
)
.await?;
let environment = codex_exec_server::Environment::from_exec_server_client(client);
let session_factory = unified_exec_session_factory_for_environment(&environment);
let manager = UnifiedExecProcessManager::with_session_factory(
DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS,
session_factory,
@@ -319,7 +280,7 @@ async fn unified_exec_can_spawn_a_local_exec_server_backend() -> anyhow::Result<
vec![
"bash".to_string(),
"-c".to_string(),
"printf unified_exec_spawned_exec_server_backend_marker".to_string(),
"printf unified_exec_remote_exec_server_environment_marker".to_string(),
],
cwd.path(),
),
@@ -332,7 +293,7 @@ async fn unified_exec_can_spawn_a_local_exec_server_backend() -> anyhow::Result<
assert_eq!(
String::from_utf8_lossy(&chunk),
"unified_exec_spawned_exec_server_backend_marker"
"unified_exec_remote_exec_server_environment_marker"
);
process.terminate();
codex-rs/exec-server/src/environment.rs
+70 -13
View File
@@ -3,11 +3,13 @@ use crate::ExecServerError;
use crate::RemoteExecServerConnectArgs;
use crate::fs;
use crate::fs::ExecutorFileSystem;
use std::sync::Arc;
#[derive(Clone, Default)]
#[derive(Clone)]
pub struct Environment {
experimental_exec_server_url: Option<String>,
remote_exec_server_client: Option<ExecServerClient>,
exec_server_client: Option<ExecServerClient>,
file_system: Arc<dyn ExecutorFileSystem>,
}
impl std::fmt::Debug for Environment {
@@ -17,19 +19,24 @@ impl std::fmt::Debug for Environment {
"experimental_exec_server_url",
&self.experimental_exec_server_url,
)
.field(
"has_remote_exec_server_client",
&self.remote_exec_server_client.is_some(),
)
.field("has_exec_server_client", &self.exec_server_client.is_some())
.finish()
}
}
impl Environment {
pub fn local() -> Self {
Self {
experimental_exec_server_url: None,
exec_server_client: None,
file_system: Arc::new(fs::LocalFileSystem),
}
}
pub async fn create(
experimental_exec_server_url: Option<String>,
) -> Result<Self, ExecServerError> {
let remote_exec_server_client =
let exec_server_client =
if let Some(websocket_url) = experimental_exec_server_url.as_deref() {
Some(
ExecServerClient::connect_websocket(RemoteExecServerConnectArgs::new(
@@ -42,35 +49,85 @@ impl Environment {
None
};
let file_system: Arc<dyn ExecutorFileSystem> = if let Some(client) = &exec_server_client {
Arc::new(fs::RemoteFileSystem::new(client.clone()))
} else {
Arc::new(fs::LocalFileSystem)
};
Ok(Self {
experimental_exec_server_url,
remote_exec_server_client,
exec_server_client,
file_system,
})
}
pub fn from_exec_server_client(client: ExecServerClient) -> Self {
Self {
experimental_exec_server_url: None,
exec_server_client: Some(client.clone()),
file_system: Arc::new(fs::RemoteFileSystem::new(client)),
}
}
pub fn experimental_exec_server_url(&self) -> Option<&str> {
self.experimental_exec_server_url.as_deref()
}
pub fn remote_exec_server_client(&self) -> Option<&ExecServerClient> {
self.remote_exec_server_client.as_ref()
pub fn exec_server_client(&self) -> Option<ExecServerClient> {
self.exec_server_client.clone()
}
pub fn get_filesystem(&self) -> impl ExecutorFileSystem + use<> {
fs::LocalFileSystem
pub fn get_filesystem(&self) -> Arc<dyn ExecutorFileSystem> {
Arc::clone(&self.file_system)
}
}
impl Default for Environment {
fn default() -> Self {
Self::local()
}
}
#[cfg(test)]
mod tests {
use super::Environment;
use crate::ExecServerClient;
use crate::ExecServerClientConnectOptions;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
#[tokio::test]
async fn create_without_remote_exec_server_url_does_not_connect() {
let environment = Environment::create(None).await.expect("create environment");
assert_eq!(environment.experimental_exec_server_url(), None);
assert!(environment.remote_exec_server_client().is_none());
assert!(environment.exec_server_client().is_none());
}
#[tokio::test]
async fn environment_uses_remote_filesystem_abstraction_when_client_is_provided() {
let client =
ExecServerClient::connect_in_process(ExecServerClientConnectOptions::default())
.await
.expect("connect in-process client");
let environment = Environment::from_exec_server_client(client);
let tempdir = TempDir::new().expect("tempdir");
let path = AbsolutePathBuf::try_from(tempdir.path().join("marker.txt")).expect("path");
environment
.get_filesystem()
.write_file(&path, b"hello".to_vec())
.await
.expect("write file through environment abstraction");
let bytes = environment
.get_filesystem()
.read_file(&path)
.await
.expect("read file through environment abstraction");
assert_eq!(bytes, b"hello");
}
}
codex-rs/exec-server/src/fs.rs
+135
View File
@@ -1,4 +1,13 @@
use async_trait::async_trait;
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD;
use codex_app_server_protocol::FsCopyParams;
use codex_app_server_protocol::FsCreateDirectoryParams;
use codex_app_server_protocol::FsGetMetadataParams;
use codex_app_server_protocol::FsReadDirectoryParams;
use codex_app_server_protocol::FsReadFileParams;
use codex_app_server_protocol::FsRemoveParams;
use codex_app_server_protocol::FsWriteFileParams;
use codex_utils_absolute_path::AbsolutePathBuf;
use std::path::Component;
use std::path::Path;
@@ -7,6 +16,9 @@ use std::time::SystemTime;
use std::time::UNIX_EPOCH;
use tokio::io;
use crate::ExecServerClient;
use crate::ExecServerError;
const MAX_READ_FILE_BYTES: u64 = 512 * 1024 * 1024;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -203,6 +215,129 @@ impl ExecutorFileSystem for LocalFileSystem {
}
}
#[derive(Clone)]
pub(crate) struct RemoteFileSystem {
client: ExecServerClient,
}
impl RemoteFileSystem {
pub(crate) fn new(client: ExecServerClient) -> Self {
Self { client }
}
}
#[async_trait]
impl ExecutorFileSystem for RemoteFileSystem {
async fn read_file(&self, path: &AbsolutePathBuf) -> FileSystemResult<Vec<u8>> {
let response = self
.client
.fs_read_file(FsReadFileParams { path: path.clone() })
.await
.map_err(map_exec_server_error)?;
STANDARD
.decode(response.data_base64)
.map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))
}
async fn write_file(&self, path: &AbsolutePathBuf, contents: Vec<u8>) -> FileSystemResult<()> {
self.client
.fs_write_file(FsWriteFileParams {
path: path.clone(),
data_base64: STANDARD.encode(contents),
})
.await
.map_err(map_exec_server_error)?;
Ok(())
}
async fn create_directory(
&self,
path: &AbsolutePathBuf,
options: CreateDirectoryOptions,
) -> FileSystemResult<()> {
self.client
.fs_create_directory(FsCreateDirectoryParams {
path: path.clone(),
recursive: Some(options.recursive),
})
.await
.map_err(map_exec_server_error)?;
Ok(())
}
async fn get_metadata(&self, path: &AbsolutePathBuf) -> FileSystemResult<FileMetadata> {
let response = self
.client
.fs_get_metadata(FsGetMetadataParams { path: path.clone() })
.await
.map_err(map_exec_server_error)?;
Ok(FileMetadata {
is_directory: response.is_directory,
is_file: response.is_file,
created_at_ms: response.created_at_ms,
modified_at_ms: response.modified_at_ms,
})
}
async fn read_directory(
&self,
path: &AbsolutePathBuf,
) -> FileSystemResult<Vec<ReadDirectoryEntry>> {
let response = self
.client
.fs_read_directory(FsReadDirectoryParams { path: path.clone() })
.await
.map_err(map_exec_server_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 remove(&self, path: &AbsolutePathBuf, options: RemoveOptions) -> FileSystemResult<()> {
self.client
.fs_remove(FsRemoveParams {
path: path.clone(),
recursive: Some(options.recursive),
force: Some(options.force),
})
.await
.map_err(map_exec_server_error)?;
Ok(())
}
async fn copy(
&self,
source_path: &AbsolutePathBuf,
destination_path: &AbsolutePathBuf,
options: CopyOptions,
) -> FileSystemResult<()> {
self.client
.fs_copy(FsCopyParams {
source_path: source_path.clone(),
destination_path: destination_path.clone(),
recursive: options.recursive,
})
.await
.map_err(map_exec_server_error)?;
Ok(())
}
}
fn map_exec_server_error(err: ExecServerError) -> io::Error {
match err {
ExecServerError::Server { code, message } if matches!(code, -32600 | -32602) => {
io::Error::new(io::ErrorKind::InvalidInput, message)
}
other => io::Error::other(other.to_string()),
}
}
fn copy_dir_recursive(source: &Path, target: &Path) -> io::Result<()> {
std::fs::create_dir_all(target)?;
for entry in std::fs::read_dir(source)? {
codex-rs/exec-server/src/lib.rs
+8 -8
View File
@@ -13,14 +13,6 @@ pub use client::ExecServerError;
pub use client_api::ExecServerClientConnectOptions;
pub use client_api::ExecServerEvent;
pub use client_api::RemoteExecServerConnectArgs;
pub use environment::Environment;
pub use fs::CopyOptions;
pub use fs::CreateDirectoryOptions;
pub use fs::ExecutorFileSystem;
pub use fs::FileMetadata;
pub use fs::FileSystemResult;
pub use fs::ReadDirectoryEntry;
pub use fs::RemoveOptions;
pub use codex_app_server_protocol::FsCopyParams;
pub use codex_app_server_protocol::FsCopyResponse;
pub use codex_app_server_protocol::FsCreateDirectoryParams;
@@ -36,6 +28,14 @@ pub use codex_app_server_protocol::FsRemoveParams;
pub use codex_app_server_protocol::FsRemoveResponse;
pub use codex_app_server_protocol::FsWriteFileParams;
pub use codex_app_server_protocol::FsWriteFileResponse;
pub use environment::Environment;
pub use fs::CopyOptions;
pub use fs::CreateDirectoryOptions;
pub use fs::ExecutorFileSystem;
pub use fs::FileMetadata;
pub use fs::FileSystemResult;
pub use fs::ReadDirectoryEntry;
pub use fs::RemoveOptions;
pub use local::ExecServerLaunchCommand;
pub use local::SpawnedExecServer;
pub use local::spawn_local_exec_server;
codex-rs/exec-server/src/server/filesystem.rs
+1 -1
View File
@@ -36,7 +36,7 @@ pub(crate) struct ExecServerFileSystem {
impl Default for ExecServerFileSystem {
fn default() -> Self {
Self {
file_system: Arc::new(Environment.get_filesystem()),
file_system: Environment::default().get_filesystem(),
}
}
}