openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 17:48:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add exec-server sandbox launch intent

Browse Source
This commit is contained in:
starr-openai
2026-05-29 18:23:29 -07:00
parent a5a94ee5a7
commit de8a76beee
25 changed files with 843 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/exec-server/src/client.rs
+23
View File
@@ -41,6 +41,7 @@ use crate::protocol::ExecExitedNotification;
use crate::protocol::ExecOutputDeltaNotification;
use crate::protocol::ExecParams;
use crate::protocol::ExecResponse;
use crate::protocol::ExecServerCapability;
use crate::protocol::FS_COPY_METHOD;
use crate::protocol::FS_CREATE_DIRECTORY_METHOD;
use crate::protocol::FS_GET_METADATA_METHOD;
@@ -175,6 +176,7 @@ struct Inner {
http_body_streams_write_lock: Mutex<()>,
http_body_stream_next_id: AtomicU64,
session_id: std::sync::RwLock<Option<String>>,
capabilities: std::sync::RwLock<Vec<ExecServerCapability>>,
reader_task: tokio::task::JoinHandle<()>,
}
@@ -341,6 +343,14 @@ impl ExecServerClient {
.unwrap_or_else(std::sync::PoisonError::into_inner);
*session_id = Some(response.session_id.clone());
}
{
let mut capabilities = self
.inner
.capabilities
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner);
*capabilities = response.capabilities.clone();
}
self.notify_initialized().await?;
Ok(response)
})
@@ -459,6 +469,14 @@ impl ExecServerClient {
.clone()
}
pub fn supports(&self, capability: ExecServerCapability) -> bool {
self.inner
.capabilities
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.contains(&capability)
}
fn is_disconnected(&self) -> bool {
self.inner.disconnected.get().is_some() || self.inner.client.is_disconnected()
}
@@ -510,6 +528,7 @@ impl ExecServerClient {
http_body_streams_write_lock: Mutex::new(()),
http_body_stream_next_id: AtomicU64::new(1),
session_id: std::sync::RwLock::new(None),
capabilities: std::sync::RwLock::new(Vec::new()),
reader_task,
}
});
@@ -1047,6 +1066,7 @@ mod tests {
id: request.id,
result: serde_json::to_value(InitializeResponse {
session_id: session_id.to_string(),
capabilities: Vec::new(),
})
.expect("initialize response should serialize"),
}),
@@ -1280,6 +1300,7 @@ mod tests {
id: request.id,
result: serde_json::to_value(InitializeResponse {
session_id: "session-1".to_string(),
capabilities: Vec::new(),
})
.expect("initialize response should serialize"),
}),
@@ -1423,6 +1444,7 @@ mod tests {
id: request.id,
result: serde_json::to_value(InitializeResponse {
session_id: "session-1".to_string(),
capabilities: Vec::new(),
})
.expect("initialize response should serialize"),
}),
@@ -1560,6 +1582,7 @@ mod tests {
id: request.id,
result: serde_json::to_value(InitializeResponse {
session_id: "session-1".to_string(),
capabilities: Vec::new(),
})
.expect("initialize response should serialize"),
}),
codex-rs/exec-server/src/environment.rs
+17 -1
View File
@@ -2,6 +2,7 @@ use std::collections::HashMap;
use std::sync::Arc;
use std::sync::RwLock;
use crate::ExecServerCapability;
use crate::ExecServerError;
use crate::ExecServerRuntimePaths;
use crate::ExecutorFileSystem;
@@ -289,6 +290,7 @@ pub struct Environment {
exec_backend: Arc<dyn ExecBackend>,
filesystem: Arc<dyn ExecutorFileSystem>,
http_client: Arc<dyn HttpClient>,
remote_client: Option<LazyRemoteExecServerClient>,
local_runtime_paths: Option<ExecServerRuntimePaths>,
}
@@ -301,6 +303,7 @@ impl Environment {
exec_backend: Arc::new(LocalProcess::default()),
filesystem: Arc::new(LocalFileSystem::unsandboxed()),
http_client: Arc::new(ReqwestHttpClient),
remote_client: None,
local_runtime_paths: None,
}
}
@@ -359,6 +362,7 @@ impl Environment {
local_runtime_paths.clone(),
)),
http_client: Arc::new(ReqwestHttpClient),
remote_client: None,
local_runtime_paths: Some(local_runtime_paths),
}
}
@@ -394,7 +398,8 @@ impl Environment {
remote_transport: Some(remote_transport),
exec_backend,
filesystem,
http_client: Arc::new(client),
http_client: Arc::new(client.clone()),
remote_client: Some(client),
local_runtime_paths,
}
}
@@ -420,6 +425,16 @@ impl Environment {
Arc::clone(&self.http_client)
}
pub async fn supports_exec_server_capability(
&self,
capability: ExecServerCapability,
) -> Result<bool, ExecServerError> {
let Some(client) = &self.remote_client else {
return Ok(false);
};
Ok(client.get().await?.supports(capability))
}
pub fn get_filesystem(&self) -> Arc<dyn ExecutorFileSystem> {
Arc::clone(&self.filesystem)
}
@@ -806,6 +821,7 @@ mod tests {
tty: false,
pipe_stdin: false,
arg0: None,
launch: crate::ExecLaunch::Materialized,
})
.await
.expect("start process");
codex-rs/exec-server/src/lib.rs
+5
View File
@@ -12,6 +12,7 @@ mod local_file_system;
mod local_process;
mod process;
mod process_id;
mod process_sandbox;
mod protocol;
mod relay;
mod relay_proto;
@@ -58,10 +59,14 @@ pub use process_id::ProcessId;
pub use protocol::ExecClosedNotification;
pub use protocol::ExecEnvPolicy;
pub use protocol::ExecExitedNotification;
pub use protocol::ExecLaunch;
pub use protocol::ExecOutputDeltaNotification;
pub use protocol::ExecOutputStream;
pub use protocol::ExecParams;
pub use protocol::ExecResponse;
pub use protocol::ExecSandboxIntent;
pub use protocol::ExecSandboxMode;
pub use protocol::ExecServerCapability;
pub use protocol::FsCopyParams;
pub use protocol::FsCopyResponse;
pub use protocol::FsCreateDirectoryParams;
codex-rs/exec-server/src/local_process.rs
+43 -2
View File
@@ -24,6 +24,7 @@ use crate::ExecServerError;
use crate::ProcessId;
use crate::StartedExecProcess;
use crate::process::ExecProcessEventLog;
use crate::process_sandbox::ProcessSandboxTransformer;
use crate::protocol::EXEC_CLOSED_METHOD;
use crate::protocol::ExecClosedNotification;
use crate::protocol::ExecEnvPolicy;
@@ -84,6 +85,7 @@ enum ProcessEntry {
struct Inner {
notifications: std::sync::RwLock<Option<RpcNotificationSender>>,
processes: Mutex<HashMap<ProcessId, ProcessEntry>>,
sandbox: Option<ProcessSandboxTransformer>,
}
#[derive(Clone)]
@@ -113,6 +115,20 @@ impl LocalProcess {
inner: Arc::new(Inner {
notifications: std::sync::RwLock::new(Some(notifications)),
processes: Mutex::new(HashMap::new()),
sandbox: None,
}),
}
}
pub(crate) fn with_runtime_paths(
notifications: RpcNotificationSender,
runtime_paths: crate::ExecServerRuntimePaths,
) -> Self {
Self {
inner: Arc::new(Inner {
notifications: std::sync::RwLock::new(Some(notifications)),
processes: Mutex::new(HashMap::new()),
sandbox: Some(ProcessSandboxTransformer::new(runtime_paths)),
}),
}
}
@@ -146,6 +162,21 @@ impl LocalProcess {
&self,
params: ExecParams,
) -> Result<(ExecResponse, watch::Sender<u64>, ExecProcessEventLog), JSONRPCErrorError> {
let materialized = match (&self.inner.sandbox, &params.launch) {
(Some(sandbox), _) => sandbox.materialize(params)?,
(None, crate::protocol::ExecLaunch::Materialized) => {
crate::process_sandbox::MaterializedProcess {
params,
sandbox: None,
}
}
(None, crate::protocol::ExecLaunch::SandboxIntent { .. }) => {
return Err(crate::rpc::invalid_request(
"sandbox intent requires exec-server runtime paths".to_string(),
));
}
};
let params = materialized.params;
let process_id = params.process_id.clone();
let (program, args) = params
.argv
@@ -259,7 +290,14 @@ impl LocalProcess {
output_notify,
));
Ok((ExecResponse { process_id }, wake_tx, events))
Ok((
ExecResponse {
process_id,
sandbox: materialized.sandbox,
},
wake_tx,
events,
))
}
pub(crate) async fn exec(&self, params: ExecParams) -> Result<ExecResponse, JSONRPCErrorError> {
@@ -406,7 +444,7 @@ impl LocalProcess {
}
}
fn child_env(params: &ExecParams) -> HashMap<String, String> {
pub(crate) fn child_env(params: &ExecParams) -> HashMap<String, String> {
let Some(env_policy) = &params.env_policy else {
return params.env.clone();
};
@@ -444,6 +482,7 @@ impl ExecBackend for LocalProcess {
.await
.map_err(map_handler_error)?;
Ok(StartedExecProcess {
sandbox: response.sandbox,
process: Arc::new(LocalExecProcess {
process_id: response.process_id,
backend: self.clone(),
@@ -706,6 +745,7 @@ fn notification_sender(inner: &Inner) -> Option<RpcNotificationSender> {
#[cfg(test)]
mod tests {
use super::*;
use crate::protocol::ExecLaunch;
use codex_protocol::config_types::ShellEnvironmentPolicyInherit;
use codex_utils_pty::ProcessDriver;
use pretty_assertions::assert_eq;
@@ -722,6 +762,7 @@ mod tests {
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
}
}
codex-rs/exec-server/src/process.rs
+2
View File
@@ -12,9 +12,11 @@ use crate::protocol::ExecParams;
use crate::protocol::ProcessOutputChunk;
use crate::protocol::ReadResponse;
use crate::protocol::WriteResponse;
use codex_sandboxing::SandboxType;
pub struct StartedExecProcess {
pub process: Arc<dyn ExecProcess>,
pub sandbox: Option<SandboxType>,
}
/// Pushed process events for consumers that want to follow process output as it
codex-rs/exec-server/src/process_sandbox.rs
+181
View File
@@ -0,0 +1,181 @@
use codex_app_server_protocol::JSONRPCErrorError;
use codex_sandboxing::SandboxCommand;
use codex_sandboxing::SandboxManager;
use codex_sandboxing::SandboxTransformRequest;
use codex_sandboxing::SandboxType;
use codex_utils_absolute_path::AbsolutePathBuf;
use crate::ExecServerRuntimePaths;
use crate::protocol::ExecLaunch;
use crate::protocol::ExecParams;
use crate::protocol::ExecSandboxMode;
use crate::rpc::invalid_request;
#[derive(Clone, Debug)]
pub(crate) struct ProcessSandboxTransformer {
runtime_paths: ExecServerRuntimePaths,
}
pub(crate) struct MaterializedProcess {
pub(crate) params: ExecParams,
pub(crate) sandbox: Option<SandboxType>,
}
impl ProcessSandboxTransformer {
pub(crate) fn new(runtime_paths: ExecServerRuntimePaths) -> Self {
Self { runtime_paths }
}
pub(crate) fn materialize(
&self,
mut params: ExecParams,
) -> Result<MaterializedProcess, JSONRPCErrorError> {
let ExecLaunch::SandboxIntent { intent } = params.launch.clone() else {
return Ok(MaterializedProcess {
params,
sandbox: None,
});
};
let (program, args) = params
.argv
.split_first()
.ok_or_else(|| invalid_request("sandbox intent argv must not be empty".to_string()))?;
let cwd = AbsolutePathBuf::from_absolute_path(params.cwd.as_path())
.map_err(|err| invalid_request(format!("sandbox intent cwd is invalid: {err}")))?;
let sandbox = executor_sandbox_type(intent.sandbox, intent.windows_sandbox_level)?;
let command = SandboxCommand {
program: program.clone().into(),
args: args.to_vec(),
cwd,
env: crate::local_process::child_env(&params),
additional_permissions: intent.additional_permissions,
};
let request = SandboxManager::new()
.transform(SandboxTransformRequest {
command,
permissions: &intent.permissions,
sandbox,
enforce_managed_network: false,
network: None,
sandbox_policy_cwd: intent.sandbox_policy_cwd.as_path(),
codex_linux_sandbox_exe: self.runtime_paths.codex_linux_sandbox_exe.as_deref(),
use_legacy_landlock: intent.use_legacy_landlock,
windows_sandbox_level: intent.windows_sandbox_level,
windows_sandbox_private_desktop: intent.windows_sandbox_private_desktop,
})
.map_err(|err| invalid_request(format!("failed to prepare process sandbox: {err}")))?;
params.argv = request.command;
params.cwd = request.cwd.to_path_buf();
params.env_policy = None;
params.env = request.env;
params.arg0 = request.arg0;
params.launch = ExecLaunch::Materialized;
Ok(MaterializedProcess {
params,
sandbox: Some(sandbox),
})
}
}
fn executor_sandbox_type(
sandbox: ExecSandboxMode,
windows_sandbox_level: codex_protocol::config_types::WindowsSandboxLevel,
) -> Result<SandboxType, JSONRPCErrorError> {
match sandbox {
ExecSandboxMode::None => Ok(SandboxType::None),
ExecSandboxMode::Platform => {
let sandbox = codex_sandboxing::get_platform_sandbox(
windows_sandbox_level
!= codex_protocol::config_types::WindowsSandboxLevel::Disabled,
)
.ok_or_else(|| {
invalid_request(
"process sandbox intent requires a platform sandbox on the executor"
.to_string(),
)
})?;
if sandbox == SandboxType::WindowsRestrictedToken {
return Err(invalid_request(
"process sandbox intent does not support WindowsRestrictedToken yet"
.to_string(),
));
}
Ok(sandbox)
}
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::models::PermissionProfile;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use super::ProcessSandboxTransformer;
use crate::ExecServerRuntimePaths;
use crate::ProcessId;
use crate::protocol::ExecLaunch;
use crate::protocol::ExecParams;
use crate::protocol::ExecSandboxIntent;
use crate::protocol::ExecSandboxMode;
#[test]
fn platform_sandbox_intent_uses_executor_platform() {
let sandbox =
super::executor_sandbox_type(ExecSandboxMode::Platform, WindowsSandboxLevel::Disabled);
match codex_sandboxing::get_platform_sandbox(/*windows_sandbox_enabled*/ false) {
Some(expected) => assert_eq!(sandbox.expect("platform sandbox"), expected),
None => assert!(sandbox.is_err()),
}
}
#[test]
fn materializes_sandbox_intent_with_executor_runtime_paths() {
let cwd = AbsolutePathBuf::from_absolute_path(
std::env::current_dir().expect("current dir").as_path(),
)
.expect("absolute cwd");
let params = ExecParams {
process_id: ProcessId::from("sandbox-intent"),
argv: vec!["true".to_string()],
cwd: cwd.to_path_buf(),
env_policy: None,
env: HashMap::from([("PATH".to_string(), "/usr/bin".to_string())]),
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::SandboxIntent {
intent: ExecSandboxIntent {
sandbox: ExecSandboxMode::None,
permissions: PermissionProfile::Disabled,
sandbox_policy_cwd: cwd,
use_legacy_landlock: false,
windows_sandbox_level: WindowsSandboxLevel::Disabled,
windows_sandbox_private_desktop: false,
additional_permissions: None,
},
},
};
let transformer = ProcessSandboxTransformer::new(
ExecServerRuntimePaths::new(
std::env::current_exe().expect("current exe"),
/*codex_linux_sandbox_exe*/ None,
)
.expect("runtime paths"),
);
let materialized = transformer.materialize(params).expect("materialize");
assert_eq!(materialized.params.launch, ExecLaunch::Materialized);
assert_eq!(materialized.params.argv, vec!["true".to_string()]);
assert_eq!(materialized.params.env_policy, None);
assert_eq!(
materialized.sandbox,
Some(codex_sandboxing::SandboxType::None)
);
}
}
codex-rs/exec-server/src/protocol.rs
+97
View File
@@ -4,6 +4,10 @@ use std::path::PathBuf;
use crate::FileSystemSandboxContext;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_protocol::config_types::ShellEnvironmentPolicyInherit;
use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::models::AdditionalPermissionProfile;
use codex_protocol::models::PermissionProfile;
use codex_sandboxing::SandboxType;
use codex_utils_absolute_path::AbsolutePathBuf;
use serde::Deserialize;
use serde::Serialize;
@@ -59,6 +63,50 @@ pub struct InitializeParams {
#[serde(rename_all = "camelCase")]
pub struct InitializeResponse {
pub session_id: String,
#[serde(default)]
pub capabilities: Vec<ExecServerCapability>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum ExecServerCapability {
ProcessStartSandboxIntent,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum ExecSandboxMode {
None,
Platform,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExecSandboxIntent {
pub sandbox: ExecSandboxMode,
pub permissions: PermissionProfile,
pub sandbox_policy_cwd: AbsolutePathBuf,
pub use_legacy_landlock: bool,
pub windows_sandbox_level: WindowsSandboxLevel,
pub windows_sandbox_private_desktop: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub additional_permissions: Option<AdditionalPermissionProfile>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum ExecLaunch {
#[default]
Materialized,
SandboxIntent {
intent: ExecSandboxIntent,
},
}
impl ExecLaunch {
fn is_materialized(&self) -> bool {
matches!(self, Self::Materialized)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -77,6 +125,8 @@ pub struct ExecParams {
#[serde(default)]
pub pipe_stdin: bool,
pub arg0: Option<String>,
#[serde(default, skip_serializing_if = "ExecLaunch::is_materialized")]
pub launch: ExecLaunch,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -93,6 +143,8 @@ pub struct ExecEnvPolicy {
#[serde(rename_all = "camelCase")]
pub struct ExecResponse {
pub process_id: ProcessId,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub sandbox: Option<SandboxType>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -398,9 +450,54 @@ mod base64_bytes {
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use super::ExecLaunch;
use super::ExecParams;
use super::ExecResponse;
use super::HttpRequestParams;
use super::InitializeResponse;
use crate::ProcessId;
use pretty_assertions::assert_eq;
#[test]
fn initialize_response_defaults_missing_capabilities() {
let response: InitializeResponse =
serde_json::from_value(serde_json::json!({ "sessionId": "session-1" }))
.expect("initialize response should deserialize");
assert_eq!(response.capabilities, Vec::new());
}
#[test]
fn materialized_exec_params_omit_launch() {
let params = ExecParams {
process_id: ProcessId::from("proc-1"),
argv: vec!["true".to_string()],
cwd: std::env::current_dir().expect("cwd"),
env_policy: None,
env: HashMap::new(),
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
};
let value = serde_json::to_value(params).expect("exec params should serialize");
assert_eq!(value.get("launch"), None);
}
#[test]
fn exec_response_defaults_missing_sandbox() {
let response: ExecResponse = serde_json::from_value(serde_json::json!({
"processId": "proc-1",
}))
.expect("legacy exec response should deserialize");
assert_eq!(response.sandbox, None);
}
#[test]
fn http_request_timeout_treats_omitted_and_null_as_no_timeout() {
let omitted: HttpRequestParams = serde_json::from_value(serde_json::json!({
codex-rs/exec-server/src/remote_process.rs
+8 -4
View File
@@ -37,12 +37,16 @@ impl ExecBackend for RemoteProcess {
let process_id = params.process_id.clone();
let client = self.client.get().await?;
let session = client.register_session(&process_id).await?;
if let Err(err) = client.exec(params).await {
session.unregister().await;
return Err(err);
}
let response = match client.exec(params).await {
Ok(response) => response,
Err(err) => {
session.unregister().await;
return Err(err);
}
};
Ok(StartedExecProcess {
sandbox: response.sandbox,
process: Arc::new(RemoteExecProcess { session }),
})
}
codex-rs/exec-server/src/server/handler.rs
+12 -1
View File
@@ -123,7 +123,10 @@ impl ExecServerHandler {
.session
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(session);
Ok(InitializeResponse { session_id })
Ok(InitializeResponse {
session_id,
capabilities: process_start_capabilities(),
})
}
pub(crate) fn initialized(&self) -> Result<(), String> {
@@ -343,5 +346,13 @@ impl ExecServerHandler {
}
}
fn process_start_capabilities() -> Vec<crate::protocol::ExecServerCapability> {
if cfg!(windows) {
Vec::new()
} else {
vec![crate::protocol::ExecServerCapability::ProcessStartSandboxIntent]
}
}
#[cfg(test)]
mod tests;
codex-rs/exec-server/src/server/handler/tests.rs
+6 -4
View File
@@ -9,6 +9,7 @@ use uuid::Uuid;
use super::ExecServerHandler;
use crate::ExecServerRuntimePaths;
use crate::ProcessId;
use crate::protocol::ExecLaunch;
use crate::protocol::ExecParams;
use crate::protocol::InitializeParams;
use crate::protocol::ReadParams;
@@ -32,6 +33,7 @@ fn exec_params_with_argv(process_id: &str, argv: Vec<String>) -> ExecParams {
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
}
}
@@ -77,7 +79,7 @@ fn test_runtime_paths() -> ExecServerRuntimePaths {
async fn initialized_handler() -> Arc<ExecServerHandler> {
let (outgoing_tx, _outgoing_rx) = mpsc::channel(16);
let registry = SessionRegistry::new();
let registry = SessionRegistry::new(test_runtime_paths());
let handler = Arc::new(ExecServerHandler::new(
registry,
RpcNotificationSender::new(outgoing_tx),
@@ -155,7 +157,7 @@ async fn terminate_reports_false_after_process_exit() {
#[tokio::test]
async fn long_poll_read_fails_after_session_resume() {
let (first_tx, _first_rx) = mpsc::channel(16);
let registry = SessionRegistry::new();
let registry = SessionRegistry::new(test_runtime_paths());
let first_handler = Arc::new(ExecServerHandler::new(
Arc::clone(&registry),
RpcNotificationSender::new(first_tx),
@@ -228,7 +230,7 @@ async fn long_poll_read_fails_after_session_resume() {
#[tokio::test]
async fn active_session_resume_is_rejected() {
let (first_tx, _first_rx) = mpsc::channel(16);
let registry = SessionRegistry::new();
let registry = SessionRegistry::new(test_runtime_paths());
let first_handler = Arc::new(ExecServerHandler::new(
Arc::clone(&registry),
RpcNotificationSender::new(first_tx),
@@ -272,7 +274,7 @@ async fn active_session_resume_is_rejected() {
async fn output_and_exit_are_retained_after_notification_receiver_closes() {
let (outgoing_tx, outgoing_rx) = mpsc::channel(16);
let handler = Arc::new(ExecServerHandler::new(
SessionRegistry::new(),
SessionRegistry::new(test_runtime_paths()),
RpcNotificationSender::new(outgoing_tx),
test_runtime_paths(),
));
codex-rs/exec-server/src/server/process_handler.rs
+5 -2
View File
@@ -17,9 +17,12 @@ pub(crate) struct ProcessHandler {
}
impl ProcessHandler {
pub(crate) fn new(notifications: RpcNotificationSender) -> Self {
pub(crate) fn new(
notifications: RpcNotificationSender,
runtime_paths: crate::ExecServerRuntimePaths,
) -> Self {
Self {
process: LocalProcess::new(notifications),
process: LocalProcess::with_runtime_paths(notifications, runtime_paths),
}
}
codex-rs/exec-server/src/server/processor.rs
+4 -2
View File
@@ -26,7 +26,7 @@ pub(crate) struct ConnectionProcessor {
impl ConnectionProcessor {
pub(crate) fn new(runtime_paths: ExecServerRuntimePaths) -> Self {
Self {
session_registry: SessionRegistry::new(),
session_registry: SessionRegistry::new(runtime_paths.clone()),
runtime_paths,
}
}
@@ -213,6 +213,7 @@ mod tests {
use crate::protocol::EXEC_METHOD;
use crate::protocol::EXEC_READ_METHOD;
use crate::protocol::EXEC_TERMINATE_METHOD;
use crate::protocol::ExecLaunch;
use crate::protocol::ExecParams;
use crate::protocol::ExecResponse;
use crate::protocol::INITIALIZE_METHOD;
@@ -226,7 +227,7 @@ mod tests {
#[tokio::test]
async fn transport_disconnect_detaches_session_during_in_flight_read() {
let registry = SessionRegistry::new();
let registry = SessionRegistry::new(test_runtime_paths());
let (mut first_writer, mut first_lines, first_task) =
spawn_test_connection(Arc::clone(&registry), "first");
@@ -402,6 +403,7 @@ mod tests {
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
}
}
codex-rs/exec-server/src/server/session_registry.rs
+5 -10
View File
@@ -7,6 +7,7 @@ use codex_app_server_protocol::JSONRPCErrorError;
use tokio::sync::Mutex;
use uuid::Uuid;
use crate::ExecServerRuntimePaths;
use crate::rpc::RpcNotificationSender;
use crate::rpc::invalid_request;
use crate::server::process_handler::ProcessHandler;
@@ -18,6 +19,7 @@ const DETACHED_SESSION_TTL: Duration = Duration::from_secs(10);
pub(crate) struct SessionRegistry {
sessions: Mutex<HashMap<String, Arc<SessionEntry>>>,
runtime_paths: ExecServerRuntimePaths,
}
struct SessionEntry {
@@ -49,9 +51,10 @@ pub(crate) struct SessionHandle {
}
impl SessionRegistry {
pub(crate) fn new() -> Arc<Self> {
pub(crate) fn new(runtime_paths: ExecServerRuntimePaths) -> Arc<Self> {
Arc::new(Self {
sessions: Mutex::new(HashMap::new()),
runtime_paths,
})
}
@@ -94,7 +97,7 @@ impl SessionRegistry {
let session_id = Uuid::new_v4().to_string();
let entry = Arc::new(SessionEntry::new(
session_id.clone(),
ProcessHandler::new(notifications),
ProcessHandler::new(notifications, self.runtime_paths.clone()),
connection_id,
));
sessions.insert(session_id, Arc::clone(&entry));
@@ -136,14 +139,6 @@ impl SessionRegistry {
}
}
impl Default for SessionRegistry {
fn default() -> Self {
Self {
sessions: Mutex::new(HashMap::new()),
}
}
}
impl SessionEntry {
fn new(session_id: String, process: ProcessHandler, connection_id: ConnectionId) -> Self {
Self {
codex-rs/exec-server/tests/exec_process.rs
+19 -8
View File
@@ -8,6 +8,7 @@ use anyhow::Context;
use anyhow::Result;
use codex_exec_server::Environment;
use codex_exec_server::ExecBackend;
use codex_exec_server::ExecLaunch;
use codex_exec_server::ExecOutputStream;
use codex_exec_server::ExecParams;
use codex_exec_server::ExecProcess;
@@ -79,6 +80,7 @@ async fn assert_exec_process_starts_and_exits(use_remote: bool) -> Result<()> {
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
assert_eq!(session.process.process_id().as_str(), "proc-1");
@@ -220,11 +222,12 @@ async fn assert_exec_process_streams_output(use_remote: bool) -> Result<()> {
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
assert_eq!(session.process.process_id().as_str(), process_id);
let StartedExecProcess { process } = session;
let StartedExecProcess { process, .. } = session;
let wake_rx = process.subscribe_wake();
let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;
assert_eq!(output, "session output\n");
@@ -251,11 +254,12 @@ async fn assert_exec_process_pushes_events(use_remote: bool) -> Result<()> {
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
assert_eq!(session.process.process_id().as_str(), process_id);
let StartedExecProcess { process } = session;
let StartedExecProcess { process, .. } = session;
let actual = collect_process_event_snapshots(process).await?;
assert_eq!(
actual,
@@ -298,11 +302,12 @@ async fn assert_exec_process_replays_events_after_close(use_remote: bool) -> Res
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
assert_eq!(session.process.process_id().as_str(), process_id);
let StartedExecProcess { process } = session;
let StartedExecProcess { process, .. } = session;
let wake_rx = process.subscribe_wake();
let read_result = collect_process_output_from_reads(Arc::clone(&process), wake_rx).await?;
assert_eq!(
@@ -346,11 +351,12 @@ async fn assert_exec_process_retains_output_after_exit_until_streams_close(
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
assert_eq!(session.process.process_id().as_str(), process_id);
let StartedExecProcess { process } = session;
let StartedExecProcess { process, .. } = session;
let exit_response = timeout(
Duration::from_secs(2),
@@ -419,13 +425,14 @@ async fn assert_exec_process_write_then_read(use_remote: bool) -> Result<()> {
tty: true,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
assert_eq!(session.process.process_id().as_str(), process_id);
tokio::time::sleep(Duration::from_millis(200)).await;
session.process.write(b"hello\n".to_vec()).await?;
let StartedExecProcess { process } = session;
let StartedExecProcess { process, .. } = session;
let wake_rx = process.subscribe_wake();
let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;
@@ -456,6 +463,7 @@ async fn assert_exec_process_write_then_read_without_tty(use_remote: bool) -> Re
tty: false,
pipe_stdin: true,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
assert_eq!(session.process.process_id().as_str(), process_id);
@@ -463,7 +471,7 @@ async fn assert_exec_process_write_then_read_without_tty(use_remote: bool) -> Re
tokio::time::sleep(Duration::from_millis(200)).await;
let write_response = session.process.write(b"hello\n".to_vec()).await?;
assert_eq!(write_response.status, WriteStatus::Accepted);
let StartedExecProcess { process } = session;
let StartedExecProcess { process, .. } = session;
let wake_rx = process.subscribe_wake();
let actual = collect_process_output_from_reads(process, wake_rx).await?;
@@ -489,13 +497,14 @@ async fn assert_exec_process_rejects_write_without_pipe_stdin(use_remote: bool)
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
assert_eq!(session.process.process_id().as_str(), process_id);
let write_response = session.process.write(b"ignored\n".to_vec()).await?;
assert_eq!(write_response.status, WriteStatus::StdinClosed);
let StartedExecProcess { process } = session;
let StartedExecProcess { process, .. } = session;
let wake_rx = process.subscribe_wake();
let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;
@@ -524,12 +533,13 @@ async fn assert_exec_process_preserves_queued_events_before_subscribe(
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
tokio::time::sleep(Duration::from_millis(200)).await;
let StartedExecProcess { process } = session;
let StartedExecProcess { process, .. } = session;
let wake_rx = process.subscribe_wake();
let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;
assert_eq!(output, "queued output\n");
@@ -558,6 +568,7 @@ async fn remote_exec_process_reports_transport_disconnect() -> Result<()> {
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::Materialized,
})
.await?;
codex-rs/exec-server/tests/http_client.rs
+1
View File
@@ -1013,6 +1013,7 @@ impl JsonRpcPeer {
request.id,
InitializeResponse {
session_id: "session-1".to_string(),
capabilities: Vec::new(),
},
)
.await?;
codex-rs/exec-server/tests/process.rs
+176 -3
View File
@@ -4,7 +4,12 @@ mod common;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCResponse;
use codex_exec_server::ExecLaunch;
use codex_exec_server::ExecParams;
use codex_exec_server::ExecResponse;
use codex_exec_server::ExecSandboxIntent;
use codex_exec_server::ExecSandboxMode;
use codex_exec_server::ExecServerCapability;
use codex_exec_server::InitializeParams;
use codex_exec_server::InitializeResponse;
use codex_exec_server::ProcessId;
@@ -12,6 +17,9 @@ use codex_exec_server::ReadResponse;
use codex_exec_server::TerminateResponse;
use codex_exec_server::WriteResponse;
use codex_exec_server::WriteStatus;
use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::models::PermissionProfile;
use codex_utils_absolute_path::AbsolutePathBuf;
use common::exec_server::exec_server;
use pretty_assertions::assert_eq;
@@ -27,7 +35,7 @@ async fn exec_server_starts_process_over_websocket() -> anyhow::Result<()> {
})?,
)
.await?;
let _ = server
let initialize_response = server
.wait_for_event(|event| {
matches!(
event,
@@ -35,6 +43,16 @@ async fn exec_server_starts_process_over_websocket() -> anyhow::Result<()> {
)
})
.await?;
let JSONRPCMessage::Response(JSONRPCResponse { result, .. }) = initialize_response else {
panic!("expected initialize response");
};
let initialize_response: InitializeResponse = serde_json::from_value(result)?;
let expected_capabilities = if cfg!(windows) {
Vec::new()
} else {
vec![ExecServerCapability::ProcessStartSandboxIntent]
};
assert_eq!(initialize_response.capabilities, expected_capabilities);
server
.send_notification("initialized", serde_json::json!({}))
@@ -70,7 +88,161 @@ async fn exec_server_starts_process_over_websocket() -> anyhow::Result<()> {
assert_eq!(
process_start_response,
ExecResponse {
process_id: ProcessId::from("proc-1")
process_id: ProcessId::from("proc-1"),
sandbox: None,
}
);
server.shutdown().await?;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_materializes_process_sandbox_intent() -> anyhow::Result<()> {
let mut server = exec_server().await?;
let initialize_id = server
.send_request(
"initialize",
serde_json::to_value(InitializeParams {
client_name: "exec-server-test".to_string(),
resume_session_id: None,
})?,
)
.await?;
let _ = server
.wait_for_event(|event| {
matches!(
event,
JSONRPCMessage::Response(JSONRPCResponse { id, .. }) if id == &initialize_id
)
})
.await?;
server
.send_notification("initialized", serde_json::json!({}))
.await?;
let cwd = AbsolutePathBuf::from_absolute_path(std::env::current_dir()?.as_path())?;
let process_start_id = server
.send_request(
"process/start",
serde_json::to_value(ExecParams {
process_id: ProcessId::from("proc-sandbox-intent"),
argv: vec!["true".to_string()],
cwd: cwd.to_path_buf(),
env_policy: None,
env: Default::default(),
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::SandboxIntent {
intent: ExecSandboxIntent {
sandbox: ExecSandboxMode::None,
permissions: PermissionProfile::Disabled,
sandbox_policy_cwd: cwd.clone(),
use_legacy_landlock: false,
windows_sandbox_level: WindowsSandboxLevel::Disabled,
windows_sandbox_private_desktop: false,
additional_permissions: None,
},
},
})?,
)
.await?;
let response = server
.wait_for_event(|event| {
matches!(
event,
JSONRPCMessage::Response(JSONRPCResponse { id, .. }) if id == &process_start_id
)
})
.await?;
let JSONRPCMessage::Response(JSONRPCResponse { result, .. }) = response else {
panic!("expected process/start response");
};
let process_start_response: ExecResponse = serde_json::from_value(result)?;
assert_eq!(
process_start_response,
ExecResponse {
process_id: ProcessId::from("proc-sandbox-intent"),
sandbox: Some(codex_sandboxing::SandboxType::None),
}
);
server.shutdown().await?;
Ok(())
}
#[cfg(not(target_os = "windows"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_materializes_platform_process_sandbox_intent() -> anyhow::Result<()> {
let mut server = exec_server().await?;
let initialize_id = server
.send_request(
"initialize",
serde_json::to_value(InitializeParams {
client_name: "exec-server-test".to_string(),
resume_session_id: None,
})?,
)
.await?;
let _ = server
.wait_for_event(|event| {
matches!(
event,
JSONRPCMessage::Response(JSONRPCResponse { id, .. }) if id == &initialize_id
)
})
.await?;
server
.send_notification("initialized", serde_json::json!({}))
.await?;
let cwd = AbsolutePathBuf::from_absolute_path(std::env::current_dir()?.as_path())?;
let process_start_id = server
.send_request(
"process/start",
serde_json::to_value(ExecParams {
process_id: ProcessId::from("proc-platform-sandbox-intent"),
argv: vec!["true".to_string()],
cwd: cwd.to_path_buf(),
env_policy: None,
env: Default::default(),
tty: false,
pipe_stdin: false,
arg0: None,
launch: ExecLaunch::SandboxIntent {
intent: ExecSandboxIntent {
sandbox: ExecSandboxMode::Platform,
permissions: PermissionProfile::Disabled,
sandbox_policy_cwd: cwd,
use_legacy_landlock: false,
windows_sandbox_level: WindowsSandboxLevel::Disabled,
windows_sandbox_private_desktop: false,
additional_permissions: None,
},
},
})?,
)
.await?;
let response = server
.wait_for_event(|event| {
matches!(
event,
JSONRPCMessage::Response(JSONRPCResponse { id, .. }) if id == &process_start_id
)
})
.await?;
let JSONRPCMessage::Response(JSONRPCResponse { result, .. }) = response else {
panic!("expected process/start response");
};
let process_start_response: ExecResponse = serde_json::from_value(result)?;
assert_eq!(
process_start_response,
ExecResponse {
process_id: ProcessId::from("proc-platform-sandbox-intent"),
sandbox: codex_sandboxing::get_platform_sandbox(/*windows_sandbox_enabled*/ false),
}
);
@@ -135,7 +307,8 @@ async fn exec_server_defaults_omitted_pipe_stdin_to_closed_stdin() -> anyhow::Re
assert_eq!(
process_start_response,
ExecResponse {
process_id: ProcessId::from("proc-default-stdin")
process_id: ProcessId::from("proc-default-stdin"),
sandbox: None,
}
);