feat(executor): add executor protocol and runtime crates

codex-rs/Cargo.lock

@@ -1672,6 +1672,7 @@ dependencies = [
  "codex-core",
  "codex-exec",
  "codex-execpolicy",
+ "codex-executor",
  "codex-features",
  "codex-login",
  "codex-mcp-server",
@@ -2068,6 +2069,27 @@ dependencies = [
  "tempfile",
 ]
 
+[[package]]
+name = "codex-executor"
+version = "0.0.0"
+dependencies = [
+ "anyhow",
+ "clap",
+ "codex-executor-protocol",
+ "pretty_assertions",
+ "serde_json",
+ "tokio",
+]
+
+[[package]]
+name = "codex-executor-protocol"
+version = "0.0.0"
+dependencies = [
+ "pretty_assertions",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "codex-experimental-api-macros"
 version = "0.0.0"

codex-rs/Cargo.toml

@@ -26,6 +26,8 @@ members = [
     "hooks",
     "secrets",
     "exec",
+    "executor",
+    "executor-protocol",
     "exec-server",
     "execpolicy",
     "execpolicy-legacy",
@@ -108,6 +110,8 @@ codex-connectors = { path = "connectors" }
 codex-config = { path = "config" }
 codex-core = { path = "core" }
 codex-exec = { path = "exec" }
+codex-executor = { path = "executor" }
+codex-executor-protocol = { path = "executor-protocol" }
 codex-exec-server = { path = "exec-server" }
 codex-execpolicy = { path = "execpolicy" }
 codex-experimental-api-macros = { path = "codex-experimental-api-macros" }

codex-rs/cli/Cargo.toml

@@ -29,6 +29,7 @@ codex-utils-cli = { workspace = true }
 codex-config = { workspace = true }
 codex-core = { workspace = true }
 codex-exec = { workspace = true }
+codex-executor = { workspace = true }
 codex-execpolicy = { workspace = true }
 codex-features = { workspace = true }
 codex-login = { workspace = true }

codex-rs/cli/src/main.rs

@@ -21,6 +21,7 @@ use codex_exec::Cli as ExecCli;
 use codex_exec::Command as ExecCommand;
 use codex_exec::ReviewArgs;
 use codex_execpolicy::ExecPolicyCheckCommand;
+use codex_executor::Cli as ExecutorCli;
 use codex_responses_api_proxy::Args as ResponsesApiProxyArgs;
 use codex_state::StateRuntime;
 use codex_state::state_db_path;
@@ -109,6 +110,9 @@ enum Subcommand {
     /// [experimental] Run the app server or related tooling.
     AppServer(AppServerCommand),
 
+    /// [experimental] Run Codex as an executor.
+    Executor(ExecutorCli),
+
     /// Launch the Codex desktop app (downloads the macOS installer if missing).
     #[cfg(target_os = "macos")]
     App(app_cmd::AppCommand),
@@ -682,6 +686,10 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
                 codex_app_server_protocol::generate_internal_json_schema(&gen_cli.out_dir)?;
             }
         },
+        Some(Subcommand::Executor(executor_cli)) => {
+            reject_remote_mode_for_subcommand(root_remote.as_deref(), "executor")?;
+            codex_executor::run_main(executor_cli).await?;
+        }
         #[cfg(target_os = "macos")]
         Some(Subcommand::App(app_cli)) => {
             reject_remote_mode_for_subcommand(root_remote.as_deref(), "app")?;
@@ -1385,6 +1393,14 @@ mod tests {
         app_server
     }
 
+    fn executor_from_args(args: &[&str]) -> ExecutorCli {
+        let cli = MultitoolCli::try_parse_from(args).expect("parse");
+        let Subcommand::Executor(executor) = cli.subcommand.expect("executor present") else {
+            unreachable!()
+        };
+        executor
+    }
+
     fn sample_exit_info(conversation_id: Option<&str>, thread_name: Option<&str>) -> AppExitInfo {
         let token_usage = TokenUsage {
             output_tokens: 2,
@@ -1622,6 +1638,11 @@ mod tests {
         assert!(app_server.analytics_default_enabled);
     }
 
+    #[test]
+    fn executor_subcommand_parses_without_extra_args() {
+        let _executor = executor_from_args(["codex", "executor"].as_ref());
+    }
+
     #[test]
     fn remote_flag_parses_for_interactive_root() {
         let cli = MultitoolCli::try_parse_from(["codex", "--remote", "ws://127.0.0.1:4500"])

codex-rs/core/src/tools/handlers/unified_exec.rs

@@ -1,9 +1,12 @@
+use crate::exec_env::create_env;
 use crate::function_tool::FunctionCallError;
 use crate::is_safe_command::is_known_safe_command;
+use crate::powershell::prefix_powershell_script_with_utf8;
 use crate::protocol::EventMsg;
 use crate::protocol::TerminalInteractionEvent;
 use crate::sandboxing::SandboxPermissions;
 use crate::shell::Shell;
+use crate::shell::ShellType;
 use crate::shell::get_shell_by_model_provided_path;
 use crate::skills::maybe_emit_implicit_skill_invocation;
 use crate::tools::context::ExecCommandToolOutput;
@@ -18,13 +21,18 @@ use crate::tools::handlers::parse_arguments_with_base_path;
 use crate::tools::handlers::resolve_workdir_base_path;
 use crate::tools::registry::ToolHandler;
 use crate::tools::registry::ToolKind;
+use crate::tools::runtimes::maybe_wrap_shell_lc_with_snapshot;
 use crate::tools::spec::UnifiedExecShellMode;
 use crate::unified_exec::ExecCommandRequest;
 use crate::unified_exec::UnifiedExecContext;
 use crate::unified_exec::UnifiedExecProcessManager;
-use crate::unified_exec::WriteStdinRequest;
+use crate::unified_exec::apply_unified_exec_env;
+use crate::unified_exec::into_wire_exec_approval_requirement;
+use crate::unified_exec::resolve_max_tokens;
 use async_trait::async_trait;
 use codex_features::Feature;
+use codex_protocol::executor::UnifiedExecExecCommandRequest;
+use codex_protocol::executor::UnifiedExecWriteStdinRequest;
 use codex_protocol::models::PermissionProfile;
 use serde::Deserialize;
 use std::path::PathBuf;
@@ -152,7 +160,6 @@ impl ToolHandler for UnifiedExecHandler {
                     args.workdir.as_deref(),
                 )
                 .await;
-                let process_id = manager.allocate_process_id().await;
                 let command = get_command(
                     &args,
                     session.user_shell(),
@@ -163,7 +170,7 @@ impl ToolHandler for UnifiedExecHandler {
                 let command_for_display = codex_shell_command::parse_command::shlex_join(&command);
 
                 let ExecCommandArgs {
-                    workdir,
+                    workdir: _,
                     tty,
                     yield_time_ms,
                     max_output_tokens,
@@ -199,16 +206,13 @@ impl ToolHandler for UnifiedExecHandler {
                     )
                 {
                     let approval_policy = context.turn.approval_policy.value();
-                    manager.release_process_id(process_id).await;
                     return Err(FunctionCallError::RespondToModel(format!(
                         "approval policy is {approval_policy:?}; reject command — you cannot ask for escalated permissions if the approval policy is {approval_policy:?}"
                     )));
                 }
 
-                let workdir = workdir.filter(|value| !value.is_empty());
-
-                let workdir = workdir.map(|dir| context.turn.resolve_path(Some(dir)));
-                let cwd = workdir.clone().unwrap_or(cwd);
+                let explicit_env_overrides = context.turn.shell_environment_policy.r#set.clone();
+                let cwd = cwd.clone();
                 let normalized_additional_permissions = match implicit_granted_permissions(
                     sandbox_permissions,
                     requested_additional_permissions.as_ref(),
@@ -229,7 +233,6 @@ impl ToolHandler for UnifiedExecHandler {
                 ) {
                     Ok(normalized) => normalized,
                     Err(err) => {
-                        manager.release_process_id(process_id).await;
                         return Err(FunctionCallError::RespondToModel(err));
                     }
                 };
@@ -246,7 +249,6 @@ impl ToolHandler for UnifiedExecHandler {
                 )
                 .await?
                 {
-                    manager.release_process_id(process_id).await;
                     return Ok(ExecCommandToolOutput {
                         event_call_id: String::new(),
                         chunk_id: String::new(),
@@ -260,23 +262,61 @@ impl ToolHandler for UnifiedExecHandler {
                     });
                 }
 
+                let exec_approval_requirement = context
+                    .session
+                    .services
+                    .exec_policy
+                    .create_exec_approval_requirement_for_command(
+                        crate::exec_policy::ExecApprovalRequest {
+                            command: &command,
+                            approval_policy: context.turn.approval_policy.value(),
+                            sandbox_policy: context.turn.sandbox_policy.get(),
+                            file_system_sandbox_policy: &context.turn.file_system_sandbox_policy,
+                            sandbox_permissions: if effective_additional_permissions
+                                .permissions_preapproved
+                            {
+                                crate::sandboxing::SandboxPermissions::UseDefault
+                            } else {
+                                effective_additional_permissions.sandbox_permissions
+                            },
+                            prefix_rule: prefix_rule.clone(),
+                        },
+                    )
+                    .await;
+                let mut command = maybe_wrap_shell_lc_with_snapshot(
+                    &command,
+                    session.user_shell().as_ref(),
+                    cwd.as_path(),
+                    &explicit_env_overrides,
+                );
+                if matches!(session.user_shell().shell_type, ShellType::PowerShell) {
+                    command = prefix_powershell_script_with_utf8(&command);
+                }
+                let env = apply_unified_exec_env(create_env(
+                    &context.turn.shell_environment_policy,
+                    Some(context.session.conversation_id),
+                ));
                 manager
                     .exec_command(
                         ExecCommandRequest {
-                            command,
-                            process_id,
-                            yield_time_ms,
-                            max_output_tokens,
-                            workdir,
+                            wire_request: UnifiedExecExecCommandRequest {
+                                command,
+                                cwd,
+                                env,
+                                tty,
+                                yield_time_ms,
+                                max_output_tokens: resolve_max_tokens(max_output_tokens),
+                                sandbox_permissions: effective_additional_permissions
+                                    .sandbox_permissions,
+                                additional_permissions: normalized_additional_permissions,
+                                additional_permissions_preapproved:
+                                    effective_additional_permissions.permissions_preapproved,
+                                justification,
+                                exec_approval_requirement: into_wire_exec_approval_requirement(
+                                    exec_approval_requirement,
+                                ),
+                            },
                             network: context.turn.network.clone(),
-                            tty,
-                            sandbox_permissions: effective_additional_permissions
-                                .sandbox_permissions,
-                            additional_permissions: normalized_additional_permissions,
-                            additional_permissions_preapproved: effective_additional_permissions
-                                .permissions_preapproved,
-                            justification,
-                            prefix_rule,
                         },
                         &context,
                     )
@@ -290,11 +330,11 @@ impl ToolHandler for UnifiedExecHandler {
             "write_stdin" => {
                 let args: WriteStdinArgs = parse_arguments(&arguments)?;
                 let response = manager
-                    .write_stdin(WriteStdinRequest {
+                    .write_stdin(UnifiedExecWriteStdinRequest {
                         process_id: args.session_id,
-                        input: &args.chars,
+                        input: args.chars.clone(),
                         yield_time_ms: args.yield_time_ms,
-                        max_output_tokens: args.max_output_tokens,
+                        max_output_tokens: resolve_max_tokens(args.max_output_tokens),
                     })
                     .await
                     .map_err(|err| {

codex-rs/core/src/tools/runtimes/unified_exec.rs

@@ -11,13 +11,10 @@ use crate::exec::ExecExpiration;
 use crate::guardian::GuardianApprovalRequest;
 use crate::guardian::review_approval_request;
 use crate::guardian::routes_approval_to_guardian;
-use crate::powershell::prefix_powershell_script_with_utf8;
 use crate::sandboxing::SandboxPermissions;
-use crate::shell::ShellType;
 use crate::tools::network_approval::NetworkApprovalMode;
 use crate::tools::network_approval::NetworkApprovalSpec;
 use crate::tools::runtimes::build_command_spec;
-use crate::tools::runtimes::maybe_wrap_shell_lc_with_snapshot;
 use crate::tools::runtimes::shell::zsh_fork_backend;
 use crate::tools::sandboxing::Approvable;
 use crate::tools::sandboxing::ApprovalCtx;
@@ -48,7 +45,6 @@ pub struct UnifiedExecRequest {
     pub command: Vec<String>,
     pub cwd: PathBuf,
     pub env: HashMap<String, String>,
-    pub explicit_env_overrides: HashMap<String, String>,
     pub network: Option<NetworkProxy>,
     pub tty: bool,
     pub sandbox_permissions: SandboxPermissions,
@@ -191,20 +187,7 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecProcess> for UnifiedExecRunt
         attempt: &SandboxAttempt<'_>,
         ctx: &ToolCtx,
     ) -> Result<UnifiedExecProcess, ToolError> {
-        let base_command = &req.command;
-        let session_shell = ctx.session.user_shell();
-        let command = maybe_wrap_shell_lc_with_snapshot(
-            base_command,
-            session_shell.as_ref(),
-            &req.cwd,
-            &req.explicit_env_overrides,
-        );
-        let command = if matches!(session_shell.shell_type, ShellType::PowerShell) {
-            prefix_powershell_script_with_utf8(&command)
-        } else {
-            command
-        };
-
+        let command = req.command.clone();
         let mut env = req.env.clone();
         if let Some(network) = req.network.as_ref() {
             network.apply_to_env(&mut env);

codex-rs/core/src/unified_exec/mod.rs

@@ -4,17 +4,21 @@
 //! - Manages interactive processes (create, reuse, buffer output with caps).
 //! - Uses the shared ToolOrchestrator to handle approval, sandbox selection, and
 //!   retry semantics in a single, descriptive flow.
+//! - Consumes a handler-built executor request so shell/env/output defaults are
+//!   resolved before the process manager runs, while the manager itself owns
+//!   unified-exec session ids.
 //! - Spawns the PTY from a sandbox-transformed `ExecRequest`; on sandbox denial,
 //!   retries without sandbox when policy allows (no re‑prompt thanks to caching).
 //! - Uses the shared `is_likely_sandbox_denied` heuristic to keep denial messages
 //!   consistent with other exec paths.
 //!
 //! Flow at a glance (open process)
-//! 1) Build a small request `{ command, cwd }`.
-//! 2) Orchestrator: approval (bypass/cache/prompt) → select sandbox → run.
-//! 3) Runtime: transform `CommandSpec` -> `ExecRequest` -> spawn PTY.
-//! 4) If denial, orchestrator retries with `SandboxType::None`.
-//! 5) Process handle is returned with streaming output + metadata.
+//! 1) Build a resolved executor request `{ command, cwd, env, approval, ... }`.
+//! 2) Manager allocates a unified-exec session id.
+//! 3) Orchestrator: approval (bypass/cache/prompt) → select sandbox → run.
+//! 4) Runtime: transform `CommandSpec` -> `ExecRequest` -> spawn PTY.
+//! 5) If denial, orchestrator retries with `SandboxType::None`.
+//! 6) Process handle is returned with streaming output + metadata.
 //!
 //! This keeps policy logic and user interaction centralized while the PTY/process
 //! concerns remain isolated here. The implementation is split between:
@@ -23,19 +27,21 @@
 
 use std::collections::HashMap;
 use std::collections::HashSet;
-use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::Weak;
 
 use codex_network_proxy::NetworkProxy;
-use codex_protocol::models::PermissionProfile;
+use codex_protocol::approvals::ExecPolicyAmendment;
+use codex_protocol::executor::UnifiedExecApprovalRequirement;
+use codex_protocol::executor::UnifiedExecExecCommandRequest;
+use codex_protocol::executor::UnifiedExecWriteStdinRequest;
 use rand::Rng;
 use rand::rng;
 use tokio::sync::Mutex;
 
 use crate::codex::Session;
 use crate::codex::TurnContext;
-use crate::sandboxing::SandboxPermissions;
+use crate::tools::sandboxing::ExecApprovalRequirement;
 
 mod async_watcher;
 mod errors;
@@ -63,6 +69,18 @@ pub(crate) const DEFAULT_MAX_OUTPUT_TOKENS: usize = 10_000;
 pub(crate) const UNIFIED_EXEC_OUTPUT_MAX_BYTES: usize = 1024 * 1024; // 1 MiB
 pub(crate) const UNIFIED_EXEC_OUTPUT_MAX_TOKENS: usize = UNIFIED_EXEC_OUTPUT_MAX_BYTES / 4;
 pub(crate) const MAX_UNIFIED_EXEC_PROCESSES: usize = 64;
+const UNIFIED_EXEC_ENV: [(&str, &str); 10] = [
+    ("NO_COLOR", "1"),
+    ("TERM", "dumb"),
+    ("LANG", "C.UTF-8"),
+    ("LC_CTYPE", "C.UTF-8"),
+    ("LC_ALL", "C.UTF-8"),
+    ("COLORTERM", ""),
+    ("PAGER", "cat"),
+    ("GIT_PAGER", "cat"),
+    ("GH_PAGER", "cat"),
+    ("CODEX_CI", "1"),
+];
 
 // Send a warning message to the models when it reaches this number of processes.
 pub(crate) const WARNING_UNIFIED_EXEC_PROCESSES: usize = 60;
@@ -85,27 +103,11 @@ impl UnifiedExecContext {
 
 #[derive(Debug)]
 pub(crate) struct ExecCommandRequest {
-    pub command: Vec<String>,
-    pub process_id: i32,
-    pub yield_time_ms: u64,
-    pub max_output_tokens: Option<usize>,
-    pub workdir: Option<PathBuf>,
+    pub wire_request: UnifiedExecExecCommandRequest,
     pub network: Option<NetworkProxy>,
-    pub tty: bool,
-    pub sandbox_permissions: SandboxPermissions,
-    pub additional_permissions: Option<PermissionProfile>,
-    pub additional_permissions_preapproved: bool,
-    pub justification: Option<String>,
-    pub prefix_rule: Option<Vec<String>>,
 }
 
-#[derive(Debug)]
-pub(crate) struct WriteStdinRequest<'a> {
-    pub process_id: i32,
-    pub input: &'a str,
-    pub yield_time_ms: u64,
-    pub max_output_tokens: Option<usize>,
-}
+pub(crate) type WriteStdinRequest = UnifiedExecWriteStdinRequest;
 
 #[derive(Default)]
 pub(crate) struct ProcessStore {
@@ -160,6 +162,73 @@ pub(crate) fn resolve_max_tokens(max_tokens: Option<usize>) -> usize {
     max_tokens.unwrap_or(DEFAULT_MAX_OUTPUT_TOKENS)
 }
 
+pub(crate) fn apply_unified_exec_env(mut env: HashMap<String, String>) -> HashMap<String, String> {
+    for (key, value) in UNIFIED_EXEC_ENV {
+        env.insert(key.to_string(), value.to_string());
+    }
+    env
+}
+
+pub(crate) fn into_wire_exec_approval_requirement(
+    requirement: ExecApprovalRequirement,
+) -> UnifiedExecApprovalRequirement {
+    match requirement {
+        ExecApprovalRequirement::Skip {
+            bypass_sandbox,
+            proposed_execpolicy_amendment,
+        } => UnifiedExecApprovalRequirement::Skip {
+            bypass_sandbox,
+            proposed_exec_policy_amendment: proposed_execpolicy_amendment,
+        },
+        ExecApprovalRequirement::NeedsApproval {
+            reason,
+            proposed_execpolicy_amendment,
+        } => UnifiedExecApprovalRequirement::NeedsApproval {
+            reason,
+            proposed_exec_policy_amendment: proposed_execpolicy_amendment,
+        },
+        ExecApprovalRequirement::Forbidden { reason } => {
+            UnifiedExecApprovalRequirement::Forbidden { reason }
+        }
+    }
+}
+
+pub(crate) fn from_wire_exec_approval_requirement(
+    requirement: &UnifiedExecApprovalRequirement,
+) -> ExecApprovalRequirement {
+    match requirement {
+        UnifiedExecApprovalRequirement::Skip {
+            bypass_sandbox,
+            proposed_exec_policy_amendment,
+        } => ExecApprovalRequirement::Skip {
+            bypass_sandbox: *bypass_sandbox,
+            proposed_execpolicy_amendment: clone_exec_policy_amendment(
+                proposed_exec_policy_amendment,
+            ),
+        },
+        UnifiedExecApprovalRequirement::NeedsApproval {
+            reason,
+            proposed_exec_policy_amendment,
+        } => ExecApprovalRequirement::NeedsApproval {
+            reason: reason.clone(),
+            proposed_execpolicy_amendment: clone_exec_policy_amendment(
+                proposed_exec_policy_amendment,
+            ),
+        },
+        UnifiedExecApprovalRequirement::Forbidden { reason } => {
+            ExecApprovalRequirement::Forbidden {
+                reason: reason.clone(),
+            }
+        }
+    }
+}
+
+fn clone_exec_policy_amendment(
+    amendment: &Option<ExecPolicyAmendment>,
+) -> Option<ExecPolicyAmendment> {
+    amendment.clone()
+}
+
 pub(crate) fn generate_chunk_id() -> String {
     let mut rng = rng();
     (0..6)

codex-rs/core/src/unified_exec/mod_tests.rs

@@ -3,11 +3,16 @@ use super::*;
 use crate::codex::Session;
 use crate::codex::TurnContext;
 use crate::codex::make_session_and_context;
+use crate::exec_env::create_env;
 use crate::protocol::AskForApproval;
 use crate::protocol::SandboxPolicy;
+use crate::sandboxing::SandboxPermissions;
 use crate::tools::context::ExecCommandToolOutput;
 use crate::unified_exec::ExecCommandRequest;
 use crate::unified_exec::WriteStdinRequest;
+use crate::unified_exec::apply_unified_exec_env;
+use codex_protocol::executor::UnifiedExecApprovalRequirement;
+use codex_protocol::executor::UnifiedExecExecCommandRequest;
 use core_test_support::skip_if_sandbox;
 use std::sync::Arc;
 use tokio::time::Duration;
@@ -35,29 +40,32 @@ async fn exec_command(
 ) -> Result<ExecCommandToolOutput, UnifiedExecError> {
     let context =
         UnifiedExecContext::new(Arc::clone(session), Arc::clone(turn), "call".to_string());
-    let process_id = session
-        .services
-        .unified_exec_manager
-        .allocate_process_id()
-        .await;
 
     session
         .services
         .unified_exec_manager
         .exec_command(
             ExecCommandRequest {
-                command: vec!["bash".to_string(), "-lc".to_string(), cmd.to_string()],
-                process_id,
-                yield_time_ms,
-                max_output_tokens: None,
-                workdir: None,
+                wire_request: UnifiedExecExecCommandRequest {
+                    command: vec!["bash".to_string(), "-lc".to_string(), cmd.to_string()],
+                    cwd: turn.cwd.clone(),
+                    env: apply_unified_exec_env(create_env(
+                        &turn.shell_environment_policy,
+                        Some(session.conversation_id),
+                    )),
+                    tty: true,
+                    yield_time_ms,
+                    max_output_tokens: DEFAULT_MAX_OUTPUT_TOKENS,
+                    sandbox_permissions: SandboxPermissions::UseDefault,
+                    additional_permissions: None,
+                    additional_permissions_preapproved: false,
+                    justification: None,
+                    exec_approval_requirement: UnifiedExecApprovalRequirement::Skip {
+                        bypass_sandbox: false,
+                        proposed_exec_policy_amendment: None,
+                    },
+                },
                 network: None,
-                tty: true,
-                sandbox_permissions: SandboxPermissions::UseDefault,
-                additional_permissions: None,
-                additional_permissions_preapproved: false,
-                justification: None,
-                prefix_rule: None,
             },
             &context,
         )
@@ -75,9 +83,9 @@ async fn write_stdin(
         .unified_exec_manager
         .write_stdin(WriteStdinRequest {
             process_id,
-            input,
+            input: input.to_string(),
             yield_time_ms,
-            max_output_tokens: None,
+            max_output_tokens: DEFAULT_MAX_OUTPUT_TOKENS,
         })
         .await
 }

codex-rs/core/src/unified_exec/process_manager.rs

@@ -1,6 +1,5 @@
 use rand::Rng;
 use std::cmp::Reverse;
-use std::collections::HashMap;
 use std::collections::HashSet;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -13,8 +12,6 @@ use tokio::time::Duration;
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 
-use crate::exec_env::create_env;
-use crate::exec_policy::ExecApprovalRequest;
 use crate::protocol::ExecCommandSource;
 use crate::sandboxing::ExecRequest;
 use crate::tools::context::ExecCommandToolOutput;
@@ -44,6 +41,7 @@ use crate::unified_exec::async_watcher::emit_exec_end_for_unified_exec;
 use crate::unified_exec::async_watcher::spawn_exit_watcher;
 use crate::unified_exec::async_watcher::start_streaming_output;
 use crate::unified_exec::clamp_yield_time;
+use crate::unified_exec::from_wire_exec_approval_requirement;
 use crate::unified_exec::generate_chunk_id;
 use crate::unified_exec::head_tail_buffer::HeadTailBuffer;
 use crate::unified_exec::process::OutputBuffer;
@@ -51,19 +49,6 @@ use crate::unified_exec::process::OutputHandles;
 use crate::unified_exec::process::SpawnLifecycleHandle;
 use crate::unified_exec::process::UnifiedExecProcess;
 
-const UNIFIED_EXEC_ENV: [(&str, &str); 10] = [
-    ("NO_COLOR", "1"),
-    ("TERM", "dumb"),
-    ("LANG", "C.UTF-8"),
-    ("LC_CTYPE", "C.UTF-8"),
-    ("LC_ALL", "C.UTF-8"),
-    ("COLORTERM", ""),
-    ("PAGER", "cat"),
-    ("GIT_PAGER", "cat"),
-    ("GH_PAGER", "cat"),
-    ("CODEX_CI", "1"),
-];
-
 /// Test-only override for deterministic unified exec process IDs.
 ///
 /// In production builds this value should remain at its default (`false`) and
@@ -82,13 +67,6 @@ fn should_use_deterministic_process_ids() -> bool {
     cfg!(test) || deterministic_process_ids_forced_for_tests()
 }
 
-fn apply_unified_exec_env(mut env: HashMap<String, String>) -> HashMap<String, String> {
-    for (key, value) in UNIFIED_EXEC_ENV {
-        env.insert(key.to_string(), value.to_string());
-    }
-    env
-}
-
 struct PreparedProcessHandles {
     writer_tx: mpsc::Sender<Vec<u8>>,
     output_buffer: OutputBuffer,
@@ -103,7 +81,7 @@ struct PreparedProcessHandles {
 }
 
 impl UnifiedExecProcessManager {
-    pub(crate) async fn allocate_process_id(&self) -> i32 {
+    async fn allocate_process_id(&self) -> i32 {
         loop {
             let mut store = self.process_store.lock().await;
 
@@ -130,7 +108,7 @@ impl UnifiedExecProcessManager {
         }
     }
 
-    pub(crate) async fn release_process_id(&self, process_id: i32) {
+    async fn release_process_id(&self, process_id: i32) {
         let removed = {
             let mut store = self.process_store.lock().await;
             store.remove(process_id)
@@ -157,20 +135,16 @@ impl UnifiedExecProcessManager {
         request: ExecCommandRequest,
         context: &UnifiedExecContext,
     ) -> Result<ExecCommandToolOutput, UnifiedExecError> {
-        let cwd = request
-            .workdir
-            .clone()
-            .unwrap_or_else(|| context.turn.cwd.clone());
-        let process = self
-            .open_session_with_sandbox(&request, cwd.clone(), context)
-            .await;
+        let process_id = self.allocate_process_id().await;
+        let cwd = request.wire_request.cwd.clone();
+        let process = self.open_session_with_sandbox(&request, context).await;
 
         let (process, mut deferred_network_approval) = match process {
             Ok((process, deferred_network_approval)) => {
                 (Arc::new(process), deferred_network_approval)
             }
             Err(err) => {
-                self.release_process_id(request.process_id).await;
+                self.release_process_id(process_id).await;
                 return Err(err);
             }
         };
@@ -183,10 +157,10 @@ impl UnifiedExecProcessManager {
             /*turn_diff_tracker*/ None,
         );
         let emitter = ToolEmitter::unified_exec(
-            &request.command,
+            &request.wire_request.command,
             cwd.clone(),
             ExecCommandSource::UnifiedExecStartup,
-            Some(request.process_id.to_string()),
+            Some(process_id.to_string()),
         );
         emitter.emit(event_ctx, ToolEventStage::Begin).await;
 
@@ -202,18 +176,18 @@ impl UnifiedExecProcessManager {
             self.store_process(
                 Arc::clone(&process),
                 context,
-                &request.command,
+                &request.wire_request.command,
                 cwd.clone(),
                 start,
-                request.process_id,
-                request.tty,
+                process_id,
+                request.wire_request.tty,
                 network_approval_id,
                 Arc::clone(&transcript),
             )
             .await;
         }
 
-        let yield_time_ms = clamp_yield_time(request.yield_time_ms);
+        let yield_time_ms = clamp_yield_time(request.wire_request.yield_time_ms);
         // For the initial exec_command call, we both stream output to events
         // (via start_streaming_output above) and collect a snapshot here for
         // the tool response body.
@@ -243,7 +217,6 @@ impl UnifiedExecProcessManager {
 
         let text = String::from_utf8_lossy(&collected).to_string();
         let chunk_id = generate_chunk_id();
-        let process_id = request.process_id;
         let (response_process_id, exit_code) = if process_started_alive {
             match self.refresh_process_state(process_id).await {
                 ProcessStatus::Alive {
@@ -269,7 +242,7 @@ impl UnifiedExecProcessManager {
                 Arc::clone(&context.session),
                 Arc::clone(&context.turn),
                 context.call_id.clone(),
-                request.command.clone(),
+                request.wire_request.command.clone(),
                 cwd.clone(),
                 Some(process_id.to_string()),
                 Arc::clone(&transcript),
@@ -279,7 +252,7 @@ impl UnifiedExecProcessManager {
             )
             .await;
 
-            self.release_process_id(request.process_id).await;
+            self.release_process_id(process_id).await;
             finish_deferred_network_approval(
                 context.session.as_ref(),
                 deferred_network_approval.take(),
@@ -295,11 +268,11 @@ impl UnifiedExecProcessManager {
             chunk_id,
             wall_time,
             raw_output: collected,
-            max_output_tokens: request.max_output_tokens,
+            max_output_tokens: Some(request.wire_request.max_output_tokens),
             process_id: response_process_id,
             exit_code,
             original_token_count: Some(original_token_count),
-            session_command: Some(request.command.clone()),
+            session_command: Some(request.wire_request.command.clone()),
         };
 
         Ok(response)
@@ -307,7 +280,7 @@ impl UnifiedExecProcessManager {
 
     pub(crate) async fn write_stdin(
         &self,
-        request: WriteStdinRequest<'_>,
+        request: WriteStdinRequest,
     ) -> Result<ExecCommandToolOutput, UnifiedExecError> {
         let process_id = request.process_id;
 
@@ -390,7 +363,7 @@ impl UnifiedExecProcessManager {
             chunk_id,
             wall_time,
             raw_output: collected,
-            max_output_tokens: request.max_output_tokens,
+            max_output_tokens: Some(request.max_output_tokens),
             process_id,
             exit_code,
             original_token_count: Some(original_token_count),
@@ -580,48 +553,29 @@ impl UnifiedExecProcessManager {
     pub(super) async fn open_session_with_sandbox(
         &self,
         request: &ExecCommandRequest,
-        cwd: PathBuf,
         context: &UnifiedExecContext,
     ) -> Result<(UnifiedExecProcess, Option<DeferredNetworkApproval>), UnifiedExecError> {
-        let env = apply_unified_exec_env(create_env(
-            &context.turn.shell_environment_policy,
-            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 exec_approval_requirement = context
-            .session
-            .services
-            .exec_policy
-            .create_exec_approval_requirement_for_command(ExecApprovalRequest {
-                command: &request.command,
-                approval_policy: context.turn.approval_policy.value(),
-                sandbox_policy: context.turn.sandbox_policy.get(),
-                file_system_sandbox_policy: &context.turn.file_system_sandbox_policy,
-                sandbox_permissions: if request.additional_permissions_preapproved {
-                    crate::sandboxing::SandboxPermissions::UseDefault
-                } else {
-                    request.sandbox_permissions
-                },
-                prefix_rule: request.prefix_rule.clone(),
-            })
-            .await;
         let req = UnifiedExecToolRequest {
-            command: request.command.clone(),
-            cwd,
-            env,
-            explicit_env_overrides: context.turn.shell_environment_policy.r#set.clone(),
+            command: request.wire_request.command.clone(),
+            cwd: request.wire_request.cwd.clone(),
+            env: request.wire_request.env.clone(),
             network: request.network.clone(),
-            tty: request.tty,
-            sandbox_permissions: request.sandbox_permissions,
-            additional_permissions: request.additional_permissions.clone(),
+            tty: request.wire_request.tty,
+            sandbox_permissions: request.wire_request.sandbox_permissions,
+            additional_permissions: request.wire_request.additional_permissions.clone(),
             #[cfg(unix)]
-            additional_permissions_preapproved: request.additional_permissions_preapproved,
-            justification: request.justification.clone(),
-            exec_approval_requirement,
+            additional_permissions_preapproved: request
+                .wire_request
+                .additional_permissions_preapproved,
+            justification: request.wire_request.justification.clone(),
+            exec_approval_requirement: from_wire_exec_approval_requirement(
+                &request.wire_request.exec_approval_requirement,
+            ),
         };
         let tool_ctx = ToolCtx {
             session: context.session.clone(),

codex-rs/core/src/unified_exec/process_manager_tests.rs

@@ -1,8 +1,11 @@
 use super::*;
 use pretty_assertions::assert_eq;
+use std::collections::HashMap;
 use tokio::time::Duration;
 use tokio::time::Instant;
 
+use crate::unified_exec::apply_unified_exec_env;
+
 #[test]
 fn unified_exec_env_injects_defaults() {
     let env = apply_unified_exec_env(HashMap::new());

codex-rs/docs/orchestrator_executor.md

@@ -0,0 +1,133 @@
+# Orchestrator and Executor Split
+
+## Goal
+
+Codex should be able to route tool calls through a remote executor without
+teaching every tool implementation about remote filesystems, RPC retries, or
+model-specific defaults.
+
+For unified exec, the main rule should be:
+
+- the executor advertises the `exec_command` and `write_stdin` tool contracts
+- the orchestrator forwards tool calls plus any orchestrator-owned overrides
+- the executor owns host-local execution details, session ids, and process
+  lifecycle
+- the executor does not invent model-dependent policy that only the
+  orchestrator can know
+
+## Current Ownership Leak
+
+Today the unified exec path is split across several layers:
+
+- `core/src/tools/handlers/unified_exec.rs` parses model args and normalizes
+  approval-related fields
+- `core/src/unified_exec/process_manager.rs` still derives execution env,
+  approval requirements, and other execution inputs from `TurnContext`
+- `core/src/tools/runtimes/unified_exec.rs` still rewrites commands for shell
+  snapshot and PowerShell handling
+- `core/src/tools/spec.rs` still statically advertises unified exec from the
+  orchestrator side instead of learning it from the executor
+
+That makes the executor boundary fuzzy. A future remote executor would need to
+know more than "handle this tool call" because some request fields are still
+hidden behind local helpers and the tool itself is not yet executor-owned.
+
+## Ownership Rule
+
+The orchestrator owns:
+
+- model sampling
+- tool routing and dynamic tool registration
+- model-facing tool-call parsing
+- shell selection and final argv construction
+- working-directory resolution
+- execution environment construction
+- approval-policy evaluation
+- output-policy selection, including max output truncation for the tool result
+
+The executor owns:
+
+- advertising host-local tool specs such as unified exec
+- spawning and tracking the process
+- streaming and buffering raw output
+- minting unified-exec session ids
+- applying the already selected sandbox attempt
+- returning the requested output slice for the current poll
+- filesystem and host-local resource access that should happen on the executor
+  host
+
+## Unified Exec Wire Contract
+
+The unified exec boundary should be expressed as an executor-advertised tool
+contract plus explicit orchestrator overrides that can be serialized and sent
+over a future executor transport.
+
+`UnifiedExecExecCommandRequest`
+
+- `command`: final argv vector
+- `cwd`: absolute working directory
+- `env`: fully constructed environment for the command
+- `tty`: whether to allocate a PTY
+- `yield_time_ms`: initial poll window
+- `max_output_tokens`: tool-output truncation budget selected by the
+  orchestrator
+- `sandbox_permissions`: already normalized sandbox request
+- `additional_permissions`: already normalized additional permissions
+- `additional_permissions_preapproved`: whether those permissions are already
+  sticky-approved
+- `justification`: optional user-facing approval reason
+- `exec_approval_requirement`: resolved approval plan for this request
+
+The executor's `exec_command` tool result should include:
+
+- `process_id`: stable unified-exec session id chosen by the executor when the
+  session stays open after the initial poll
+- no `process_id` for short-lived commands that exit before the initial poll
+  completes
+
+`UnifiedExecWriteStdinRequest`
+
+- `process_id`: existing unified-exec session id
+- `input`: bytes to write, expressed as text for the tool path
+- `yield_time_ms`: poll window for this write or empty poll
+- `max_output_tokens`: tool-output truncation budget selected by the
+  orchestrator
+
+These are intentionally higher level than `codex-exec-server`'s low-level
+`process/start`, `process/read`, and `process/write` methods. Unified exec is a
+tool contract owned by the executor; the executor transport can map that
+contract onto lower-level process RPCs internally.
+
+## Phase Plan
+
+1. Let the executor advertise unified exec tool specs to the orchestrator,
+   similar to MCP `listTools`.
+2. Define protocol-owned unified-exec request types for executor-owned tool
+   calls plus orchestrator-owned overrides.
+3. Make the executor side mint unified-exec session ids and return them in tool
+   results.
+4. Move the host-local unified-exec implementation behind the executor boundary.
+5. Reuse the same request types for a remote executor transport.
+
+## Crate Split
+
+- `codex-executor-protocol`: shared executor transport types with minimal
+  dependencies
+- `codex-executor`: executor-side receive loop and future transport adapter
+- `codex-cli`: `codex executor` subcommand that launches the executor process
+
+## Follow-ups
+
+This work does not fully remove every host-local concern yet. In particular:
+
+- remote network-proxy binding is still local to the executor path
+- zsh-fork backend setup remains local because it depends on executor-host
+  binaries and inherited file descriptors
+- shell snapshot capture and executor bootstrapping still need a broader
+  context-gathering protocol
+- unified exec is still statically registered by the orchestrator today rather
+  than advertised by the executor
+
+Those should be follow-up steps, but they should build on the same principle:
+the orchestrator chooses only the policy and routing inputs that are truly
+orchestrator-owned, while the executor owns host-local tool behavior.

codex-rs/executor-protocol/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "codex-executor-protocol"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[lib]
+name = "codex_executor_protocol"
+path = "src/lib.rs"
+
+[lints]
+workspace = true
+
+[dependencies]
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+
+[dev-dependencies]
+pretty_assertions = { workspace = true }

codex-rs/executor-protocol/src/lib.rs

@@ -0,0 +1,189 @@
+use serde::Deserialize;
+use serde::Serialize;
+use serde_json::Value as JsonValue;
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecutorToolSpec {
+    pub name: String,
+    pub description: String,
+    pub input_schema: JsonValue,
+}
+
+impl ExecutorToolSpec {
+    pub fn new(
+        name: impl Into<String>,
+        description: impl Into<String>,
+        input_schema: JsonValue,
+    ) -> Self {
+        Self {
+            name: name.into(),
+            description: description.into(),
+            input_schema,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct ListToolsRequest {
+    pub request_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct CallToolRequest {
+    pub request_id: String,
+    pub tool_name: String,
+    pub arguments: JsonValue,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct ShutdownRequest {
+    pub request_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum OrchestratorToExecutorMessage {
+    ListTools(ListToolsRequest),
+    CallTool(CallToolRequest),
+    Shutdown(ShutdownRequest),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct ListToolsResponse {
+    pub request_id: String,
+    pub tools: Vec<ExecutorToolSpec>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum ToolCallContent {
+    InputText { text: String },
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "status", rename_all = "snake_case")]
+pub enum ToolCallOutcome {
+    Success { content: Vec<ToolCallContent> },
+    Error { message: String },
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct CallToolResponse {
+    pub request_id: String,
+    pub tool_name: String,
+    pub outcome: ToolCallOutcome,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct ShutdownResponse {
+    pub request_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct ErrorResponse {
+    pub request_id: Option<String>,
+    pub message: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum ExecutorToOrchestratorMessage {
+    ListToolsResponse(ListToolsResponse),
+    CallToolResponse(CallToolResponse),
+    ShutdownResponse(ShutdownResponse),
+    Error(ErrorResponse),
+}
+
+impl ExecutorToOrchestratorMessage {
+    pub fn error(request_id: Option<String>, message: impl Into<String>) -> Self {
+        Self::Error(ErrorResponse {
+            request_id,
+            message: message.into(),
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::CallToolRequest;
+    use super::ExecutorToOrchestratorMessage;
+    use super::ExecutorToolSpec;
+    use super::ListToolsResponse;
+    use super::OrchestratorToExecutorMessage;
+    use pretty_assertions::assert_eq;
+    use serde_json::json;
+
+    #[test]
+    fn list_tools_response_round_trips_with_camel_case_fields() {
+        let message = ExecutorToOrchestratorMessage::ListToolsResponse(ListToolsResponse {
+            request_id: "req-1".to_string(),
+            tools: vec![ExecutorToolSpec::new(
+                "exec_command",
+                "Run a command",
+                json!({
+                    "type": "object",
+                    "properties": {
+                        "cmd": {"type": "string"},
+                    },
+                    "required": ["cmd"],
+                }),
+            )],
+        });
+
+        let value = serde_json::to_value(&message).expect("serialize");
+        assert_eq!(
+            value,
+            json!({
+                "type": "list_tools_response",
+                "requestId": "req-1",
+                "tools": [{
+                    "name": "exec_command",
+                    "description": "Run a command",
+                    "inputSchema": {
+                        "type": "object",
+                        "properties": {
+                            "cmd": {"type": "string"},
+                        },
+                        "required": ["cmd"],
+                    },
+                }],
+            })
+        );
+    }
+
+    #[test]
+    fn call_tool_request_round_trips() {
+        let value = json!({
+            "type": "call_tool",
+            "requestId": "req-2",
+            "toolName": "exec_command",
+            "arguments": {
+                "cmd": "pwd",
+            },
+        });
+
+        let actual: OrchestratorToExecutorMessage =
+            serde_json::from_value(value.clone()).expect("deserialize");
+
+        assert_eq!(
+            actual,
+            OrchestratorToExecutorMessage::CallTool(CallToolRequest {
+                request_id: "req-2".to_string(),
+                tool_name: "exec_command".to_string(),
+                arguments: json!({
+                    "cmd": "pwd",
+                }),
+            })
+        );
+
+        assert_eq!(serde_json::to_value(actual).expect("serialize"), value);
+    }
+}

codex-rs/executor/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "codex-executor"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[[bin]]
+name = "codex-executor"
+path = "src/main.rs"
+
+[lib]
+name = "codex_executor"
+path = "src/lib.rs"
+
+[lints]
+workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+clap = { workspace = true, features = ["derive"] }
+codex-executor-protocol = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync"] }
+
+[dev-dependencies]
+pretty_assertions = { workspace = true }
+serde_json = { workspace = true }

codex-rs/executor/src/cli.rs

@@ -0,0 +1,5 @@
+use clap::Parser;
+
+#[derive(Debug, Parser, Default)]
+#[command(version)]
+pub struct Cli {}

codex-rs/executor/src/lib.rs

@@ -0,0 +1,249 @@
+mod cli;
+
+pub use cli::Cli;
+
+use anyhow::Context;
+use codex_executor_protocol::CallToolResponse;
+use codex_executor_protocol::ExecutorToOrchestratorMessage;
+use codex_executor_protocol::ExecutorToolSpec;
+use codex_executor_protocol::ListToolsResponse;
+use codex_executor_protocol::OrchestratorToExecutorMessage;
+use codex_executor_protocol::ShutdownResponse;
+use codex_executor_protocol::ToolCallOutcome;
+use tokio::sync::mpsc;
+
+pub async fn run_main(cli: Cli) -> anyhow::Result<()> {
+    let (outbound, inbound) = establish_connection(&cli).await?;
+    Executor::new(outbound, inbound).run().await
+}
+
+async fn establish_connection(
+    _cli: &Cli,
+) -> anyhow::Result<(
+    mpsc::Sender<ExecutorToOrchestratorMessage>,
+    mpsc::Receiver<OrchestratorToExecutorMessage>,
+)> {
+    unimplemented!("executor transport establishment is not implemented yet");
+}
+
+pub struct Executor {
+    outbound: mpsc::Sender<ExecutorToOrchestratorMessage>,
+    inbound: mpsc::Receiver<OrchestratorToExecutorMessage>,
+    tools: Vec<ExecutorToolSpec>,
+}
+
+impl Executor {
+    pub fn new(
+        outbound: mpsc::Sender<ExecutorToOrchestratorMessage>,
+        inbound: mpsc::Receiver<OrchestratorToExecutorMessage>,
+    ) -> Self {
+        Self::with_tools(outbound, inbound, Vec::new())
+    }
+
+    pub fn with_tools(
+        outbound: mpsc::Sender<ExecutorToOrchestratorMessage>,
+        inbound: mpsc::Receiver<OrchestratorToExecutorMessage>,
+        tools: Vec<ExecutorToolSpec>,
+    ) -> Self {
+        Self {
+            outbound,
+            inbound,
+            tools,
+        }
+    }
+
+    pub async fn run(mut self) -> anyhow::Result<()> {
+        while let Some(message) = self.inbound.recv().await {
+            if self.handle_message(message).await? {
+                break;
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn handle_message(
+        &mut self,
+        message: OrchestratorToExecutorMessage,
+    ) -> anyhow::Result<bool> {
+        match message {
+            OrchestratorToExecutorMessage::ListTools(request) => {
+                self.send(ExecutorToOrchestratorMessage::ListToolsResponse(
+                    ListToolsResponse {
+                        request_id: request.request_id,
+                        tools: self.tools.clone(),
+                    },
+                ))
+                .await?;
+                Ok(false)
+            }
+            OrchestratorToExecutorMessage::CallTool(request) => {
+                let outcome = if self.tool_exists(request.tool_name.as_str()) {
+                    ToolCallOutcome::Error {
+                        message: format!(
+                            "tool `{}` is registered but execution is not implemented yet",
+                            request.tool_name
+                        ),
+                    }
+                } else {
+                    ToolCallOutcome::Error {
+                        message: format!("unknown executor tool `{}`", request.tool_name),
+                    }
+                };
+
+                self.send(ExecutorToOrchestratorMessage::CallToolResponse(
+                    CallToolResponse {
+                        request_id: request.request_id,
+                        tool_name: request.tool_name,
+                        outcome,
+                    },
+                ))
+                .await?;
+                Ok(false)
+            }
+            OrchestratorToExecutorMessage::Shutdown(request) => {
+                self.send(ExecutorToOrchestratorMessage::ShutdownResponse(
+                    ShutdownResponse {
+                        request_id: request.request_id,
+                    },
+                ))
+                .await?;
+                Ok(true)
+            }
+        }
+    }
+
+    async fn send(&self, message: ExecutorToOrchestratorMessage) -> anyhow::Result<()> {
+        self.outbound
+            .send(message)
+            .await
+            .context("executor outbound channel closed")
+    }
+
+    fn tool_exists(&self, tool_name: &str) -> bool {
+        self.tools.iter().any(|tool| tool.name == tool_name)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Executor;
+    use codex_executor_protocol::CallToolRequest;
+    use codex_executor_protocol::ExecutorToOrchestratorMessage;
+    use codex_executor_protocol::ExecutorToolSpec;
+    use codex_executor_protocol::ListToolsRequest;
+    use codex_executor_protocol::ListToolsResponse;
+    use codex_executor_protocol::OrchestratorToExecutorMessage;
+    use codex_executor_protocol::ShutdownRequest;
+    use codex_executor_protocol::ShutdownResponse;
+    use codex_executor_protocol::ToolCallOutcome;
+    use pretty_assertions::assert_eq;
+    use serde_json::json;
+    use tokio::sync::mpsc;
+
+    #[tokio::test]
+    async fn executor_lists_registered_tools() {
+        let (outbound_tx, mut outbound_rx) = mpsc::channel(4);
+        let (inbound_tx, inbound_rx) = mpsc::channel(4);
+        let executor = Executor::with_tools(
+            outbound_tx,
+            inbound_rx,
+            vec![ExecutorToolSpec::new(
+                "exec_command",
+                "Run a command",
+                json!({"type": "object"}),
+            )],
+        );
+
+        let task = tokio::spawn(async move { executor.run().await });
+        inbound_tx
+            .send(OrchestratorToExecutorMessage::ListTools(ListToolsRequest {
+                request_id: "req-1".to_string(),
+            }))
+            .await
+            .expect("send request");
+
+        let response = outbound_rx.recv().await.expect("response");
+        assert_eq!(
+            response,
+            ExecutorToOrchestratorMessage::ListToolsResponse(ListToolsResponse {
+                request_id: "req-1".to_string(),
+                tools: vec![ExecutorToolSpec::new(
+                    "exec_command",
+                    "Run a command",
+                    json!({"type": "object"}),
+                )],
+            })
+        );
+
+        drop(inbound_tx);
+        task.await.expect("task join").expect("task result");
+    }
+
+    #[tokio::test]
+    async fn executor_returns_error_for_call_tool_until_handlers_exist() {
+        let (outbound_tx, mut outbound_rx) = mpsc::channel(4);
+        let (inbound_tx, inbound_rx) = mpsc::channel(4);
+        let executor = Executor::with_tools(
+            outbound_tx,
+            inbound_rx,
+            vec![ExecutorToolSpec::new(
+                "exec_command",
+                "Run a command",
+                json!({"type": "object"}),
+            )],
+        );
+
+        let task = tokio::spawn(async move { executor.run().await });
+        inbound_tx
+            .send(OrchestratorToExecutorMessage::CallTool(CallToolRequest {
+                request_id: "req-2".to_string(),
+                tool_name: "exec_command".to_string(),
+                arguments: json!({"cmd": "pwd"}),
+            }))
+            .await
+            .expect("send request");
+
+        let response = outbound_rx.recv().await.expect("response");
+        let ExecutorToOrchestratorMessage::CallToolResponse(response) = response else {
+            panic!("expected call tool response");
+        };
+        assert_eq!(response.request_id, "req-2");
+        assert_eq!(response.tool_name, "exec_command");
+        assert_eq!(
+            response.outcome,
+            ToolCallOutcome::Error {
+                message: "tool `exec_command` is registered but execution is not implemented yet"
+                    .to_string(),
+            }
+        );
+
+        drop(inbound_tx);
+        task.await.expect("task join").expect("task result");
+    }
+
+    #[tokio::test]
+    async fn executor_acknowledges_shutdown_and_exits() {
+        let (outbound_tx, mut outbound_rx) = mpsc::channel(4);
+        let (inbound_tx, inbound_rx) = mpsc::channel(4);
+        let executor = Executor::new(outbound_tx, inbound_rx);
+
+        let task = tokio::spawn(async move { executor.run().await });
+        inbound_tx
+            .send(OrchestratorToExecutorMessage::Shutdown(ShutdownRequest {
+                request_id: "req-3".to_string(),
+            }))
+            .await
+            .expect("send request");
+
+        let response = outbound_rx.recv().await.expect("response");
+        assert_eq!(
+            response,
+            ExecutorToOrchestratorMessage::ShutdownResponse(ShutdownResponse {
+                request_id: "req-3".to_string(),
+            })
+        );
+
+        task.await.expect("task join").expect("task result");
+    }
+}

codex-rs/executor/src/main.rs

@@ -0,0 +1,7 @@
+use clap::Parser;
+use codex_executor::Cli;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    codex_executor::run_main(Cli::parse()).await
+}

codex-rs/protocol/src/executor.rs

@@ -0,0 +1,138 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+
+use crate::approvals::ExecPolicyAmendment;
+use crate::models::PermissionProfile;
+use crate::models::SandboxPermissions;
+use schemars::JsonSchema;
+use serde::Deserialize;
+use serde::Serialize;
+
+/// Approval plan chosen by the orchestrator for a unified-exec command.
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema)]
+#[serde(rename_all = "camelCase", rename_all_fields = "camelCase")]
+pub enum UnifiedExecApprovalRequirement {
+    Skip {
+        bypass_sandbox: bool,
+        proposed_exec_policy_amendment: Option<ExecPolicyAmendment>,
+    },
+    NeedsApproval {
+        reason: Option<String>,
+        proposed_exec_policy_amendment: Option<ExecPolicyAmendment>,
+    },
+    Forbidden {
+        reason: String,
+    },
+}
+
+/// Fully resolved unified-exec startup request suitable for an executor wire.
+///
+/// The executor, not the orchestrator, mints the long-lived session handle for
+/// any interactive process that survives the initial poll.
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct UnifiedExecExecCommandRequest {
+    pub command: Vec<String>,
+    pub cwd: PathBuf,
+    pub env: HashMap<String, String>,
+    pub tty: bool,
+    pub yield_time_ms: u64,
+    pub max_output_tokens: usize,
+    pub sandbox_permissions: SandboxPermissions,
+    pub additional_permissions: Option<PermissionProfile>,
+    pub additional_permissions_preapproved: bool,
+    pub justification: Option<String>,
+    pub exec_approval_requirement: UnifiedExecApprovalRequirement,
+}
+
+/// Fully resolved unified-exec follow-up request suitable for an executor wire.
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct UnifiedExecWriteStdinRequest {
+    pub process_id: i32,
+    pub input: String,
+    pub yield_time_ms: u64,
+    pub max_output_tokens: usize,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::UnifiedExecApprovalRequirement;
+    use super::UnifiedExecExecCommandRequest;
+    use super::UnifiedExecWriteStdinRequest;
+    use crate::models::SandboxPermissions;
+    use pretty_assertions::assert_eq;
+    use serde_json::json;
+    use std::collections::HashMap;
+    use std::path::PathBuf;
+
+    #[test]
+    fn unified_exec_exec_command_request_uses_camel_case() {
+        let request = UnifiedExecExecCommandRequest {
+            command: vec!["bash".to_string(), "-lc".to_string(), "pwd".to_string()],
+            cwd: PathBuf::from("/tmp/example"),
+            env: HashMap::from([("TERM".to_string(), "dumb".to_string())]),
+            tty: true,
+            yield_time_ms: 2_500,
+            max_output_tokens: 4_000,
+            sandbox_permissions: SandboxPermissions::UseDefault,
+            additional_permissions: None,
+            additional_permissions_preapproved: false,
+            justification: Some("Need to inspect the repo".to_string()),
+            exec_approval_requirement: UnifiedExecApprovalRequirement::NeedsApproval {
+                reason: Some("sandbox escalation requested".to_string()),
+                proposed_exec_policy_amendment: None,
+            },
+        };
+
+        let value = serde_json::to_value(&request).expect("serialize request");
+        assert_eq!(
+            value,
+            json!({
+                "command": ["bash", "-lc", "pwd"],
+                "cwd": "/tmp/example",
+                "env": {
+                    "TERM": "dumb",
+                },
+                "tty": true,
+                "yieldTimeMs": 2_500,
+                "maxOutputTokens": 4_000,
+                "sandboxPermissions": "use_default",
+                "additionalPermissions": null,
+                "additionalPermissionsPreapproved": false,
+                "justification": "Need to inspect the repo",
+                "execApprovalRequirement": {
+                    "needsApproval": {
+                        "reason": "sandbox escalation requested",
+                        "proposedExecPolicyAmendment": null,
+                    }
+                }
+            })
+        );
+    }
+
+    #[test]
+    fn unified_exec_write_stdin_request_round_trips() {
+        let value = json!({
+            "processId": 7,
+            "input": "echo hello\n",
+            "yieldTimeMs": 5_000,
+            "maxOutputTokens": 1_500,
+        });
+
+        let request = serde_json::from_value::<UnifiedExecWriteStdinRequest>(value.clone())
+            .expect("deserialize request");
+        assert_eq!(
+            request,
+            UnifiedExecWriteStdinRequest {
+                process_id: 7,
+                input: "echo hello\n".to_string(),
+                yield_time_ms: 5_000,
+                max_output_tokens: 1_500,
+            }
+        );
+
+        let encoded = serde_json::to_value(&request).expect("serialize request");
+        assert_eq!(encoded, value);
+    }
+}

codex-rs/protocol/src/lib.rs

@@ -7,6 +7,7 @@ pub mod approvals;
 pub mod config_types;
 pub mod custom_prompts;
 pub mod dynamic_tools;
+pub mod executor;
 pub mod items;
 pub mod mcp;
 pub mod memory_citation;