From 0ee08bc702b9e6a71da24fa0234ea8d7f50630c7 Mon Sep 17 00:00:00 2001 From: starr-openai Date: Sat, 21 Mar 2026 00:58:05 +0000 Subject: [PATCH] refactor(core): add explicit exec-server unified exec runtime path Co-authored-by: Codex --- .../core/src/tools/runtimes/unified_exec.rs | 25 ++- .../core/src/unified_exec/process_manager.rs | 178 ++++++++++-------- 2 files changed, 125 insertions(+), 78 deletions(-) diff --git a/codex-rs/core/src/tools/runtimes/unified_exec.rs b/codex-rs/core/src/tools/runtimes/unified_exec.rs index 3d7ce21173..80da762c4e 100644 --- a/codex-rs/core/src/tools/runtimes/unified_exec.rs +++ b/codex-rs/core/src/tools/runtimes/unified_exec.rs @@ -73,6 +73,7 @@ pub struct UnifiedExecApprovalKey { pub struct UnifiedExecRuntime<'a> { manager: &'a UnifiedExecProcessManager, shell_mode: UnifiedExecShellMode, + use_exec_server: bool, } impl<'a> UnifiedExecRuntime<'a> { @@ -80,6 +81,18 @@ impl<'a> UnifiedExecRuntime<'a> { Self { manager, shell_mode, + use_exec_server: false, + } + } + + pub fn with_exec_server( + manager: &'a UnifiedExecProcessManager, + shell_mode: UnifiedExecShellMode, + ) -> Self { + Self { + manager, + shell_mode, + use_exec_server: true, } } } @@ -193,6 +206,12 @@ impl<'a> ToolRuntime> for UnifiedExecRun attempt: &SandboxAttempt<'_>, ctx: &ToolCtx, ) -> Result, ToolError> { + let use_exec_server = if self.use_exec_server { + true + } else { + req.use_exec_server + }; + let base_command = &req.command; let session_shell = ctx.session.user_shell(); let command = maybe_wrap_shell_lc_with_snapshot( @@ -241,7 +260,7 @@ impl<'a> ToolRuntime> for UnifiedExecRun &prepared.exec_request, req.process_id, req.tty, - req.use_exec_server, + use_exec_server, Some(ctx.session.services.environment.get_executor()), prepared.spawn_lifecycle, ) @@ -281,8 +300,8 @@ impl<'a> ToolRuntime> for UnifiedExecRun &exec_env, req.process_id, req.tty, - req.use_exec_server, - if req.use_exec_server { + use_exec_server, + if use_exec_server { Some(ctx.session.services.environment.get_executor()) } else { None diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index d8b1a93cab..2ef441c77b 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -537,78 +537,102 @@ impl UnifiedExecProcessManager { mut spawn_lifecycle: SpawnLifecycleHandle, ) -> Result, UnifiedExecError> { if use_exec_server { - let executor = executor - .take() - .ok_or_else(|| UnifiedExecError::create_process("exec-server unavailable".to_string()))?; - let response = executor - .start(ExecParams { - process_id: process_id.to_string(), - argv: env.command.clone(), - cwd: env.cwd.clone(), - env: env.env.clone(), - tty, - arg0: env.arg0.clone(), - }) - .await - .map_err(|err| UnifiedExecError::create_process(err.to_string()))?; - let _ = response.process_id; - - let process = Arc::new(ProcessBackend::ExecServer { - process_id: process_id.to_string(), - executor, - output_buffer: Arc::new(tokio::sync::Mutex::new(HeadTailBuffer::default())), - output_notify: Arc::new(Notify::new()), - output_closed: Arc::new(AtomicBool::new(false)), - output_closed_notify: Arc::new(Notify::new()), - output_drained: Arc::new(Notify::new()), - cancellation_token: CancellationToken::new(), - exit_code: Arc::new(RwLock::new(None)), - has_exited: Arc::new(AtomicBool::new(false)), - sandbox_type: env.sandbox, - output_seq: Arc::new(AtomicU64::new(0)), - }); - - spawn_lifecycle.after_spawn(); - Self::spawn_exec_server_output_watcher(Arc::clone(&process)); - - Ok(process) - } else { - let (program, args) = env - .command - .split_first() - .ok_or(UnifiedExecError::MissingCommandLine)?; - let inherited_fds = spawn_lifecycle.inherited_fds(); - - let spawn_result = if tty { - codex_utils_pty::pty::spawn_process_with_inherited_fds( - program, - args, - env.cwd.as_path(), - &env.env, - &env.arg0, - codex_utils_pty::TerminalSize::default(), - &inherited_fds, - ) - .await - } else { - codex_utils_pty::pipe::spawn_process_no_stdin_with_inherited_fds( - program, - args, - env.cwd.as_path(), - &env.env, - &env.arg0, - &inherited_fds, - ) - .await - }; - let spawned = - spawn_result.map_err(|err| UnifiedExecError::create_process(err.to_string()))?; - spawn_lifecycle.after_spawn(); - let process = UnifiedExecProcess::from_spawned(spawned, env.sandbox, spawn_lifecycle).await?; - Ok(Arc::new(ProcessBackend::Local { - process: Arc::new(process), - })) + return self + .open_session_with_exec_server(env, process_id, tty, &mut executor, spawn_lifecycle) + .await; } + + self.open_session_with_local_process(env, process_id, tty, spawn_lifecycle) + .await + } + + async fn open_session_with_local_process( + &self, + env: &ExecRequest, + process_id: i32, + tty: bool, + spawn_lifecycle: SpawnLifecycleHandle, + ) -> Result, UnifiedExecError> { + let (program, args) = env + .command + .split_first() + .ok_or(UnifiedExecError::MissingCommandLine)?; + let inherited_fds = spawn_lifecycle.inherited_fds(); + + let spawn_result = if tty { + codex_utils_pty::pty::spawn_process_with_inherited_fds( + program, + args, + env.cwd.as_path(), + &env.env, + &env.arg0, + codex_utils_pty::TerminalSize::default(), + &inherited_fds, + ) + .await + } else { + codex_utils_pty::pipe::spawn_process_no_stdin_with_inherited_fds( + program, + args, + env.cwd.as_path(), + &env.env, + &env.arg0, + &inherited_fds, + ) + .await + }; + + let spawned = spawn_result.map_err(|err| UnifiedExecError::create_process(err.to_string()))?; + spawn_lifecycle.after_spawn(); + let process = UnifiedExecProcess::from_spawned(spawned, env.sandbox, spawn_lifecycle).await?; + Ok(Arc::new(ProcessBackend::Local { + process: Arc::new(process), + })) + } + + async fn open_session_with_exec_server( + &self, + env: &ExecRequest, + process_id: i32, + tty: bool, + mut executor: &mut Option>, + spawn_lifecycle: SpawnLifecycleHandle, + ) -> Result, UnifiedExecError> { + let executor = executor + .take() + .ok_or_else(|| UnifiedExecError::create_process("exec-server unavailable".to_string()))?; + let response = executor + .start(ExecParams { + process_id: process_id.to_string(), + argv: env.command.clone(), + cwd: env.cwd.clone(), + env: env.env.clone(), + tty, + arg0: env.arg0.clone(), + }) + .await + .map_err(|err| UnifiedExecError::create_process(err.to_string()))?; + let _ = response.process_id; + + let process = Arc::new(ProcessBackend::ExecServer { + process_id: process_id.to_string(), + executor, + output_buffer: Arc::new(tokio::sync::Mutex::new(HeadTailBuffer::default())), + output_notify: Arc::new(Notify::new()), + output_closed: Arc::new(AtomicBool::new(false)), + output_closed_notify: Arc::new(Notify::new()), + output_drained: Arc::new(Notify::new()), + cancellation_token: CancellationToken::new(), + exit_code: Arc::new(RwLock::new(None)), + has_exited: Arc::new(AtomicBool::new(false)), + sandbox_type: env.sandbox, + output_seq: Arc::new(AtomicU64::new(0)), + }); + + spawn_lifecycle.after_spawn(); + Self::spawn_exec_server_output_watcher(Arc::clone(&process)); + + Ok(process) } pub(super) async fn open_session_with_sandbox( @@ -622,10 +646,14 @@ impl UnifiedExecProcessManager { Some(context.session.conversation_id), )); let mut orchestrator = ToolOrchestrator::new(); - let mut runtime = UnifiedExecRuntime::new( - self, - context.turn.tools_config.unified_exec_shell_mode.clone(), - ); + let mut runtime = if context.turn.config.experimental_exec_server_url.is_some() { + UnifiedExecRuntime::with_exec_server( + self, + context.turn.tools_config.unified_exec_shell_mode.clone(), + ) + } else { + UnifiedExecRuntime::new(self, context.turn.tools_config.unified_exec_shell_mode.clone()) + }; let exec_approval_requirement = context .session .services