Add exec-server transport for unified exec

codex-rs/Cargo.lock

@@ -1651,6 +1651,7 @@ dependencies = [
  "codex-config",
  "codex-core",
  "codex-exec",
+ "codex-exec-server",
  "codex-execpolicy",
  "codex-login",
  "codex-mcp-server",
@@ -1837,6 +1838,7 @@ dependencies = [
  "codex-client",
  "codex-config",
  "codex-connectors",
+ "codex-exec-server",
  "codex-execpolicy",
  "codex-file-search",
  "codex-git",
@@ -1985,6 +1987,23 @@ dependencies = [
  "wiremock",
 ]
 
+[[package]]
+name = "codex-exec-server"
+version = "0.0.0"
+dependencies = [
+ "anyhow",
+ "base64 0.22.1",
+ "codex-app-server-protocol",
+ "codex-utils-cargo-bin",
+ "codex-utils-pty",
+ "pretty_assertions",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+ "tracing",
+]
+
 [[package]]
 name = "codex-execpolicy"
 version = "0.0.0"

codex-rs/Cargo.toml

@@ -25,6 +25,7 @@ members = [
     "hooks",
     "secrets",
     "exec",
+    "exec-server",
     "execpolicy",
     "execpolicy-legacy",
     "keyring-store",
@@ -103,6 +104,7 @@ codex-connectors = { path = "connectors" }
 codex-config = { path = "config" }
 codex-core = { path = "core" }
 codex-exec = { path = "exec" }
+codex-exec-server = { path = "exec-server" }
 codex-execpolicy = { path = "execpolicy" }
 codex-experimental-api-macros = { path = "codex-experimental-api-macros" }
 codex-feedback = { path = "feedback" }

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-exec-server = { workspace = true }
 codex-execpolicy = { workspace = true }
 codex-login = { workspace = true }
 codex-mcp-server = { workspace = true }

codex-rs/cli/src/main.rs

@@ -20,6 +20,7 @@ use codex_cloud_tasks::Cli as CloudTasksCli;
 use codex_exec::Cli as ExecCli;
 use codex_exec::Command as ExecCommand;
 use codex_exec::ReviewArgs;
+use codex_exec_server::run_main as run_exec_server_main;
 use codex_execpolicy::ExecPolicyCheckCommand;
 use codex_responses_api_proxy::Args as ResponsesApiProxyArgs;
 use codex_state::StateRuntime;
@@ -144,6 +145,10 @@ enum Subcommand {
     #[clap(hide = true, name = "stdio-to-uds")]
     StdioToUds(StdioToUdsCommand),
 
+    /// Internal: run the exec-server stdio JSON-RPC daemon.
+    #[clap(hide = true, name = "exec-server")]
+    ExecServer,
+
     /// Inspect feature flags.
     Features(FeaturesCli),
 }
@@ -781,6 +786,11 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
             tokio::task::spawn_blocking(move || codex_stdio_to_uds::run(socket_path.as_path()))
                 .await??;
         }
+        Some(Subcommand::ExecServer) => {
+            run_exec_server_main()
+                .await
+                .map_err(|err| anyhow::anyhow!(err.to_string()))?;
+        }
         Some(Subcommand::Features(FeaturesCli { sub })) => match sub {
             FeaturesSubcommand::List => {
                 // Respect root-level `-c` overrides plus top-level flags like `--profile`.

codex-rs/core/Cargo.toml

@@ -37,6 +37,7 @@ codex-config = { workspace = true }
 codex-shell-command = { workspace = true }
 codex-skills = { workspace = true }
 codex-execpolicy = { workspace = true }
+codex-exec-server = { workspace = true }
 codex-file-search = { workspace = true }
 codex-git = { workspace = true }
 codex-hooks = { workspace = true }

codex-rs/core/src/codex.rs

@@ -58,6 +58,8 @@ use chrono::Local;
 use chrono::Utc;
 use codex_app_server_protocol::McpServerElicitationRequest;
 use codex_app_server_protocol::McpServerElicitationRequestParams;
+use codex_exec_server::ExecServerClient;
+use codex_exec_server::ExecServerLaunchCommand;
 use codex_hooks::HookEvent;
 use codex_hooks::HookEventAfterAgent;
 use codex_hooks::HookPayload;
@@ -175,6 +177,60 @@ use crate::error::Result as CodexResult;
 use crate::exec::StreamOutput;
 use codex_config::CONFIG_TOML_FILE;
 
+const CODEX_EXEC_SERVER_EXE_ENV_VAR: &str = "CODEX_EXEC_SERVER_EXE";
+
+fn resolve_exec_server_launch_command() -> CodexResult<ExecServerLaunchCommand> {
+    if let Some(override_path) = std::env::var_os(CODEX_EXEC_SERVER_EXE_ENV_VAR) {
+        return Ok(ExecServerLaunchCommand {
+            program: PathBuf::from(override_path),
+            args: Vec::new(),
+        });
+    }
+
+    let exec_server_binary_name = if cfg!(windows) {
+        "codex-exec-server.exe"
+    } else {
+        "codex-exec-server"
+    };
+    if let Ok(current_exe) = std::env::current_exe()
+        && let Some(parent) = current_exe.parent()
+    {
+        let sibling = parent.join(exec_server_binary_name);
+        if sibling.is_file() {
+            return Ok(ExecServerLaunchCommand {
+                program: sibling,
+                args: Vec::new(),
+            });
+        }
+
+        let codex_binary = parent.join(if cfg!(windows) { "codex.exe" } else { "codex" });
+        if codex_binary.is_file() {
+            return Ok(ExecServerLaunchCommand {
+                program: codex_binary,
+                args: vec!["exec-server".to_string()],
+            });
+        }
+    }
+
+    if let Ok(program) = which::which(exec_server_binary_name) {
+        return Ok(ExecServerLaunchCommand {
+            program,
+            args: Vec::new(),
+        });
+    }
+
+    if let Ok(program) = which::which(if cfg!(windows) { "codex.exe" } else { "codex" }) {
+        return Ok(ExecServerLaunchCommand {
+            program,
+            args: vec!["exec-server".to_string()],
+        });
+    }
+
+    Err(CodexErr::Fatal(format!(
+        "failed to resolve exec-server binary; set {CODEX_EXEC_SERVER_EXE_ENV_VAR} or install codex-exec-server"
+    )))
+}
+
 mod rollout_reconstruction;
 #[cfg(test)]
 mod rollout_reconstruction_tests;
@@ -1722,6 +1778,18 @@ impl Session {
             });
         }
 
+        let exec_server_client = if config.features.enabled(Feature::ExecServer) {
+            Some(Arc::new(
+                ExecServerClient::spawn(resolve_exec_server_launch_command()?)
+                    .await
+                    .map_err(|err| {
+                        CodexErr::Fatal(format!("failed to start exec-server: {err}"))
+                    })?,
+            ))
+        } else {
+            None
+        };
+
         let services = SessionServices {
             // Initialize the MCP connection manager with an uninitialized
             // instance. It will be replaced with one created via
@@ -1736,6 +1804,7 @@ impl Session {
             mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()),
             unified_exec_manager: UnifiedExecProcessManager::new(
                 config.background_terminal_max_timeout,
+                exec_server_client,
             ),
             shell_zsh_path: config.zsh_path.clone(),
             main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(),

codex-rs/core/src/codex_tests.rs

@@ -2123,6 +2123,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
         mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()),
         unified_exec_manager: UnifiedExecProcessManager::new(
             config.background_terminal_max_timeout,
+            None,
         ),
         shell_zsh_path: None,
         main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(),
@@ -2795,6 +2796,7 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx(
         mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()),
         unified_exec_manager: UnifiedExecProcessManager::new(
             config.background_terminal_max_timeout,
+            None,
         ),
         shell_zsh_path: None,
         main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(),

codex-rs/core/src/features.rs

@@ -93,6 +93,8 @@ pub enum Feature {
     JsReplToolsOnly,
     /// Use the single unified PTY-backed exec tool.
     UnifiedExec,
+    /// Route unified-exec process realization through the exec-server daemon.
+    ExecServer,
     /// Route shell tool execution through the zsh exec bridge.
     ShellZshFork,
     /// Include the freeform apply_patch tool.
@@ -540,6 +542,12 @@ pub const FEATURES: &[FeatureSpec] = &[
         stage: Stage::Stable,
         default_enabled: !cfg!(windows),
     },
+    FeatureSpec {
+        id: Feature::ExecServer,
+        key: "experimental_exec_server",
+        stage: Stage::UnderDevelopment,
+        default_enabled: false,
+    },
     FeatureSpec {
         id: Feature::ShellZshFork,
         key: "shell_zsh_fork",

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

@@ -46,6 +46,7 @@ use std::path::PathBuf;
 
 #[derive(Clone, Debug)]
 pub struct UnifiedExecRequest {
+    pub process_id: i32,
     pub command: Vec<String>,
     pub cwd: PathBuf,
     pub env: HashMap<String, String>,
@@ -228,6 +229,7 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecProcess> for UnifiedExecRunt
                     return self
                         .manager
                         .open_session_with_exec_env(
+                            req.process_id,
                             &prepared.exec_request,
                             req.tty,
                             prepared.spawn_lifecycle,
@@ -264,7 +266,12 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecProcess> for UnifiedExecRunt
             .env_for(spec, req.network.as_ref())
             .map_err(|err| ToolError::Codex(err.into()))?;
         self.manager
-            .open_session_with_exec_env(&exec_env, req.tty, Box::new(NoopSpawnLifecycle))
+            .open_session_with_exec_env(
+                req.process_id,
+                &exec_env,
+                req.tty,
+                Box::new(NoopSpawnLifecycle),
+            )
             .await
             .map_err(|err| match err {
                 UnifiedExecError::SandboxDenied { output, .. } => {

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

@@ -27,6 +27,7 @@ use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::Weak;
 
+use codex_exec_server::ExecServerClient;
 use codex_network_proxy::NetworkProxy;
 use codex_protocol::models::PermissionProfile;
 use rand::Rng;
@@ -123,21 +124,26 @@ impl ProcessStore {
 pub(crate) struct UnifiedExecProcessManager {
     process_store: Mutex<ProcessStore>,
     max_write_stdin_yield_time_ms: u64,
+    exec_server_client: Option<Arc<ExecServerClient>>,
 }
 
 impl UnifiedExecProcessManager {
-    pub(crate) fn new(max_write_stdin_yield_time_ms: u64) -> Self {
+    pub(crate) fn new(
+        max_write_stdin_yield_time_ms: u64,
+        exec_server_client: Option<Arc<ExecServerClient>>,
+    ) -> Self {
         Self {
             process_store: Mutex::new(ProcessStore::default()),
             max_write_stdin_yield_time_ms: max_write_stdin_yield_time_ms
                 .max(MIN_EMPTY_YIELD_TIME_MS),
+            exec_server_client,
         }
     }
 }
 
 impl Default for UnifiedExecProcessManager {
     fn default() -> Self {
-        Self::new(DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS)
+        Self::new(DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS, None)
     }
 }
 

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

@@ -8,6 +8,9 @@ use crate::protocol::SandboxPolicy;
 use crate::tools::context::ExecCommandToolOutput;
 use crate::unified_exec::ExecCommandRequest;
 use crate::unified_exec::WriteStdinRequest;
+use codex_exec_server::ExecServerClient;
+use codex_exec_server::ExecServerLaunchCommand;
+use codex_utils_cargo_bin::cargo_bin;
 use core_test_support::skip_if_sandbox;
 use std::sync::Arc;
 use tokio::time::Duration;
@@ -82,6 +85,54 @@ async fn write_stdin(
         .await
 }
 
+async fn exec_command_with_manager(
+    manager: &UnifiedExecProcessManager,
+    session: &Arc<Session>,
+    turn: &Arc<TurnContext>,
+    cmd: &str,
+    yield_time_ms: u64,
+) -> Result<ExecCommandToolOutput, UnifiedExecError> {
+    let context =
+        UnifiedExecContext::new(Arc::clone(session), Arc::clone(turn), "call".to_string());
+    let process_id = manager.allocate_process_id().await;
+
+    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,
+                network: None,
+                tty: true,
+                sandbox_permissions: SandboxPermissions::UseDefault,
+                additional_permissions: None,
+                additional_permissions_preapproved: false,
+                justification: None,
+                prefix_rule: None,
+            },
+            &context,
+        )
+        .await
+}
+
+async fn write_stdin_with_manager(
+    manager: &UnifiedExecProcessManager,
+    process_id: i32,
+    input: &str,
+    yield_time_ms: u64,
+) -> Result<ExecCommandToolOutput, UnifiedExecError> {
+    manager
+        .write_stdin(WriteStdinRequest {
+            process_id,
+            input,
+            yield_time_ms,
+            max_output_tokens: None,
+        })
+        .await
+}
+
 #[test]
 fn push_chunk_preserves_prefix_and_suffix() {
     let mut buffer = HeadTailBuffer::default();
@@ -341,3 +392,46 @@ async fn reusing_completed_process_returns_unknown_process() -> anyhow::Result<(
 
     Ok(())
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn unified_exec_can_route_through_exec_server() -> anyhow::Result<()> {
+    skip_if_sandbox!(Ok(()));
+
+    let manager = UnifiedExecProcessManager::new(
+        DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS,
+        Some(Arc::new(
+            ExecServerClient::spawn(ExecServerLaunchCommand {
+                program: cargo_bin("codex-exec-server")?,
+                args: Vec::new(),
+            })
+            .await?,
+        )),
+    );
+    let (session, turn) = test_session_and_turn().await;
+
+    let open_shell = exec_command_with_manager(&manager, &session, &turn, "bash -i", 2_500).await?;
+    let process_id = open_shell.process_id.expect("expected process_id");
+
+    write_stdin_with_manager(
+        &manager,
+        process_id,
+        "export CODEX_INTERACTIVE_SHELL_VAR=codex-remote\n",
+        2_500,
+    )
+    .await?;
+
+    let output = write_stdin_with_manager(
+        &manager,
+        process_id,
+        "echo $CODEX_INTERACTIVE_SHELL_VAR\n",
+        2_500,
+    )
+    .await?;
+    assert!(
+        output.truncated_output().contains("codex-remote"),
+        "expected remote exec-server backed session state"
+    );
+
+    manager.terminate_all_processes().await;
+    Ok(())
+}

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

@@ -18,6 +18,7 @@ use crate::exec::StreamOutput;
 use crate::exec::is_likely_sandbox_denied;
 use crate::truncate::TruncationPolicy;
 use crate::truncate::formatted_truncate_text;
+use codex_exec_server::ExecServerProcess;
 use codex_utils_pty::ExecCommandSession;
 use codex_utils_pty::SpawnedPty;
 
@@ -40,6 +41,49 @@ pub(crate) trait SpawnLifecycle: std::fmt::Debug + Send + Sync {
 
 pub(crate) type SpawnLifecycleHandle = Box<dyn SpawnLifecycle>;
 
+trait ExecProcessControl: std::fmt::Debug + Send + Sync {
+    fn writer_sender(&self) -> mpsc::Sender<Vec<u8>>;
+    fn has_exited(&self) -> bool;
+    fn exit_code(&self) -> Option<i32>;
+    fn terminate(&self);
+}
+
+impl ExecProcessControl for ExecCommandSession {
+    fn writer_sender(&self) -> mpsc::Sender<Vec<u8>> {
+        self.writer_sender()
+    }
+
+    fn has_exited(&self) -> bool {
+        self.has_exited()
+    }
+
+    fn exit_code(&self) -> Option<i32> {
+        self.exit_code()
+    }
+
+    fn terminate(&self) {
+        self.terminate();
+    }
+}
+
+impl ExecProcessControl for ExecServerProcess {
+    fn writer_sender(&self) -> mpsc::Sender<Vec<u8>> {
+        self.writer_sender()
+    }
+
+    fn has_exited(&self) -> bool {
+        self.has_exited()
+    }
+
+    fn exit_code(&self) -> Option<i32> {
+        self.exit_code()
+    }
+
+    fn terminate(&self) {
+        self.terminate();
+    }
+}
+
 #[derive(Debug, Default)]
 pub(crate) struct NoopSpawnLifecycle;
 
@@ -56,7 +100,7 @@ pub(crate) struct OutputHandles {
 
 #[derive(Debug)]
 pub(crate) struct UnifiedExecProcess {
-    process_handle: ExecCommandSession,
+    process_handle: Box<dyn ExecProcessControl>,
     output_rx: broadcast::Receiver<Vec<u8>>,
     output_buffer: OutputBuffer,
     output_notify: Arc<Notify>,
@@ -70,8 +114,8 @@ pub(crate) struct UnifiedExecProcess {
 }
 
 impl UnifiedExecProcess {
-    pub(super) fn new(
-        process_handle: ExecCommandSession,
+    fn build(
+        process_handle: Box<dyn ExecProcessControl>,
         initial_output_rx: tokio::sync::broadcast::Receiver<Vec<u8>>,
         sandbox_type: SandboxType,
         spawn_lifecycle: SpawnLifecycleHandle,
@@ -122,6 +166,34 @@ impl UnifiedExecProcess {
         }
     }
 
+    pub(super) fn new(
+        process_handle: ExecCommandSession,
+        initial_output_rx: tokio::sync::broadcast::Receiver<Vec<u8>>,
+        sandbox_type: SandboxType,
+        spawn_lifecycle: SpawnLifecycleHandle,
+    ) -> Self {
+        Self::build(
+            Box::new(process_handle),
+            initial_output_rx,
+            sandbox_type,
+            spawn_lifecycle,
+        )
+    }
+
+    pub(super) fn from_remote(
+        process_handle: ExecServerProcess,
+        sandbox_type: SandboxType,
+        spawn_lifecycle: SpawnLifecycleHandle,
+    ) -> Self {
+        let output_rx = process_handle.output_receiver();
+        Self::build(
+            Box::new(process_handle),
+            output_rx,
+            sandbox_type,
+            spawn_lifecycle,
+        )
+    }
+
     pub(super) fn writer_sender(&self) -> mpsc::Sender<Vec<u8>> {
         self.process_handle.writer_sender()
     }

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

@@ -50,6 +50,7 @@ use crate::unified_exec::process::OutputBuffer;
 use crate::unified_exec::process::OutputHandles;
 use crate::unified_exec::process::SpawnLifecycleHandle;
 use crate::unified_exec::process::UnifiedExecProcess;
+use codex_exec_server::ExecParams as ExecServerExecParams;
 
 const UNIFIED_EXEC_ENV: [(&str, &str); 10] = [
     ("NO_COLOR", "1"),
@@ -529,10 +530,32 @@ impl UnifiedExecProcessManager {
 
     pub(crate) async fn open_session_with_exec_env(
         &self,
+        process_id: i32,
         env: &ExecRequest,
         tty: bool,
         mut spawn_lifecycle: SpawnLifecycleHandle,
     ) -> Result<UnifiedExecProcess, UnifiedExecError> {
+        if let Some(exec_server_client) = self.exec_server_client.as_ref() {
+            let process = exec_server_client
+                .start_process(ExecServerExecParams {
+                    process_id: process_id.to_string(),
+                    argv: env.command.clone(),
+                    cwd: env.cwd.clone(),
+                    env: env.env.clone(),
+                    tty,
+                    output_bytes_cap: crate::unified_exec::UNIFIED_EXEC_OUTPUT_MAX_BYTES,
+                    arg0: env.arg0.clone(),
+                })
+                .await
+                .map_err(|err| UnifiedExecError::create_process(err.to_string()))?;
+            spawn_lifecycle.after_spawn();
+            return Ok(UnifiedExecProcess::from_remote(
+                process,
+                env.sandbox,
+                spawn_lifecycle,
+            ));
+        }
+
         let (program, args) = env
             .command
             .split_first()
@@ -598,6 +621,7 @@ impl UnifiedExecProcessManager {
             })
             .await;
         let req = UnifiedExecToolRequest {
+            process_id: request.process_id,
             command: request.command.clone(),
             cwd,
             env,

codex-rs/exec-server/Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "codex-exec-server"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[[bin]]
+name = "codex-exec-server"
+path = "src/bin/codex-exec-server.rs"
+
+[lints]
+workspace = true
+
+[dependencies]
+base64 = { workspace = true }
+codex-app-server-protocol = { workspace = true }
+codex-utils-pty = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = [
+    "io-std",
+    "io-util",
+    "macros",
+    "process",
+    "rt-multi-thread",
+    "sync",
+    "time",
+] }
+tracing = { workspace = true }
+
+[dev-dependencies]
+anyhow = { workspace = true }
+codex-utils-cargo-bin = { workspace = true }
+pretty_assertions = { workspace = true }

codex-rs/exec-server/src/bin/codex-exec-server.rs

@@ -0,0 +1,7 @@
+#[tokio::main]
+async fn main() {
+    if let Err(err) = codex_exec_server::run_main().await {
+        eprintln!("{err}");
+        std::process::exit(1);
+    }
+}

codex-rs/exec-server/src/client.rs

@@ -0,0 +1,504 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::process::Stdio;
+use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::AtomicI64;
+use std::sync::atomic::Ordering;
+
+use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use serde_json::Value;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::process::Child;
+use tokio::process::Command;
+use tokio::sync::Mutex;
+use tokio::sync::broadcast;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+use tracing::debug;
+use tracing::warn;
+
+use crate::protocol::EXEC_EXITED_METHOD;
+use crate::protocol::EXEC_METHOD;
+use crate::protocol::EXEC_OUTPUT_DELTA_METHOD;
+use crate::protocol::EXEC_TERMINATE_METHOD;
+use crate::protocol::EXEC_WRITE_METHOD;
+use crate::protocol::ExecExitedNotification;
+use crate::protocol::ExecOutputDeltaNotification;
+use crate::protocol::ExecParams;
+use crate::protocol::ExecResponse;
+use crate::protocol::INITIALIZE_METHOD;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::InitializeParams;
+use crate::protocol::InitializeResponse;
+use crate::protocol::TerminateParams;
+use crate::protocol::TerminateResponse;
+use crate::protocol::WriteParams;
+use crate::protocol::WriteResponse;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ExecServerLaunchCommand {
+    pub program: PathBuf,
+    pub args: Vec<String>,
+}
+
+pub struct ExecServerProcess {
+    process_id: String,
+    pid: Option<u32>,
+    output_rx: broadcast::Receiver<Vec<u8>>,
+    writer_tx: mpsc::Sender<Vec<u8>>,
+    status: Arc<RemoteProcessStatus>,
+    client: ExecServerClient,
+}
+
+impl std::fmt::Debug for ExecServerProcess {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ExecServerProcess")
+            .field("process_id", &self.process_id)
+            .field("pid", &self.pid)
+            .field("has_exited", &self.has_exited())
+            .field("exit_code", &self.exit_code())
+            .finish()
+    }
+}
+
+impl ExecServerProcess {
+    pub fn writer_sender(&self) -> mpsc::Sender<Vec<u8>> {
+        self.writer_tx.clone()
+    }
+
+    pub fn output_receiver(&self) -> broadcast::Receiver<Vec<u8>> {
+        self.output_rx.resubscribe()
+    }
+
+    pub fn has_exited(&self) -> bool {
+        self.status.has_exited()
+    }
+
+    pub fn exit_code(&self) -> Option<i32> {
+        self.status.exit_code()
+    }
+
+    pub fn pid(&self) -> Option<u32> {
+        self.pid
+    }
+
+    pub fn terminate(&self) {
+        self.status.mark_exited(None);
+        let client = self.client.clone();
+        let process_id = self.process_id.clone();
+        tokio::spawn(async move {
+            let _ = client.terminate_process(&process_id).await;
+        });
+    }
+}
+
+impl std::fmt::Debug for RemoteProcessStatus {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RemoteProcessStatus")
+            .field("exited", &self.has_exited())
+            .field("exit_code", &self.exit_code())
+            .finish()
+    }
+}
+
+struct RemoteProcessStatus {
+    exited: AtomicBool,
+    exit_code: StdMutex<Option<i32>>,
+}
+
+impl RemoteProcessStatus {
+    fn new() -> Self {
+        Self {
+            exited: AtomicBool::new(false),
+            exit_code: StdMutex::new(None),
+        }
+    }
+
+    fn has_exited(&self) -> bool {
+        self.exited.load(Ordering::SeqCst)
+    }
+
+    fn exit_code(&self) -> Option<i32> {
+        self.exit_code.lock().ok().and_then(|guard| *guard)
+    }
+
+    fn mark_exited(&self, exit_code: Option<i32>) {
+        self.exited.store(true, Ordering::SeqCst);
+        if let Ok(mut guard) = self.exit_code.lock() {
+            *guard = exit_code;
+        }
+    }
+}
+
+struct RegisteredProcess {
+    output_tx: broadcast::Sender<Vec<u8>>,
+    status: Arc<RemoteProcessStatus>,
+}
+
+struct Inner {
+    child: StdMutex<Option<Child>>,
+    write_tx: mpsc::UnboundedSender<JSONRPCMessage>,
+    pending: Mutex<HashMap<RequestId, oneshot::Sender<Result<Value, JSONRPCErrorError>>>>,
+    processes: Mutex<HashMap<String, RegisteredProcess>>,
+    next_request_id: AtomicI64,
+    reader_task: JoinHandle<()>,
+    writer_task: JoinHandle<()>,
+}
+
+impl Drop for Inner {
+    fn drop(&mut self) {
+        self.reader_task.abort();
+        self.writer_task.abort();
+        if let Ok(mut child_guard) = self.child.lock()
+            && let Some(child) = child_guard.as_mut()
+        {
+            let _ = child.start_kill();
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct ExecServerClient {
+    inner: Arc<Inner>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ExecServerError {
+    #[error("failed to spawn exec-server: {0}")]
+    Spawn(#[source] std::io::Error),
+    #[error("exec-server transport closed")]
+    Closed,
+    #[error("failed to serialize or deserialize exec-server JSON: {0}")]
+    Json(#[from] serde_json::Error),
+    #[error("exec-server protocol error: {0}")]
+    Protocol(String),
+    #[error("exec-server rejected request ({code}): {message}")]
+    Server { code: i64, message: String },
+}
+
+impl ExecServerClient {
+    pub async fn spawn(command: ExecServerLaunchCommand) -> Result<Self, ExecServerError> {
+        let mut child = Command::new(&command.program);
+        child.args(&command.args);
+        child.stdin(Stdio::piped());
+        child.stdout(Stdio::piped());
+        child.stderr(Stdio::inherit());
+        child.kill_on_drop(true);
+
+        let mut child = child.spawn().map_err(ExecServerError::Spawn)?;
+        let stdin = child.stdin.take().ok_or_else(|| {
+            ExecServerError::Protocol("exec-server stdin was not captured".to_string())
+        })?;
+        let stdout = child.stdout.take().ok_or_else(|| {
+            ExecServerError::Protocol("exec-server stdout was not captured".to_string())
+        })?;
+
+        let (write_tx, mut write_rx) = mpsc::unbounded_channel::<JSONRPCMessage>();
+        let writer_task = tokio::spawn(async move {
+            let mut stdin = stdin;
+            while let Some(message) = write_rx.recv().await {
+                let encoded = match serde_json::to_vec(&message) {
+                    Ok(encoded) => encoded,
+                    Err(err) => {
+                        warn!("failed to encode exec-server message: {err}");
+                        break;
+                    }
+                };
+                if stdin.write_all(&encoded).await.is_err() {
+                    break;
+                }
+                if stdin.write_all(b"\n").await.is_err() {
+                    break;
+                }
+                if stdin.flush().await.is_err() {
+                    break;
+                }
+            }
+        });
+
+        let pending = Mutex::new(HashMap::<
+            RequestId,
+            oneshot::Sender<Result<Value, JSONRPCErrorError>>,
+        >::new());
+        let processes = Mutex::new(HashMap::<String, RegisteredProcess>::new());
+        let inner = Arc::new_cyclic(move |weak| {
+            let weak = weak.clone();
+            let reader_task = tokio::spawn(async move {
+                let mut lines = BufReader::new(stdout).lines();
+                loop {
+                    let Some(inner) = weak.upgrade() else {
+                        break;
+                    };
+                    let next_line = lines.next_line().await;
+                    match next_line {
+                        Ok(Some(line)) => {
+                            if line.trim().is_empty() {
+                                continue;
+                            }
+                            match serde_json::from_str::<JSONRPCMessage>(&line) {
+                                Ok(message) => {
+                                    if let Err(err) = handle_server_message(&inner, message).await {
+                                        warn!("failed to handle exec-server message: {err}");
+                                        break;
+                                    }
+                                }
+                                Err(err) => {
+                                    warn!("failed to parse exec-server message: {err}");
+                                    break;
+                                }
+                            }
+                        }
+                        Ok(None) => break,
+                        Err(err) => {
+                            warn!("failed to read exec-server stdout: {err}");
+                            break;
+                        }
+                    }
+                }
+                if let Some(inner) = weak.upgrade() {
+                    handle_transport_shutdown(&inner).await;
+                }
+            });
+
+            Inner {
+                child: StdMutex::new(Some(child)),
+                write_tx,
+                pending,
+                processes,
+                next_request_id: AtomicI64::new(1),
+                reader_task,
+                writer_task,
+            }
+        });
+
+        let client = Self { inner };
+        client.initialize().await?;
+        Ok(client)
+    }
+
+    pub async fn start_process(
+        &self,
+        params: ExecParams,
+    ) -> Result<ExecServerProcess, ExecServerError> {
+        let process_id = params.process_id.clone();
+        let status = Arc::new(RemoteProcessStatus::new());
+        let (output_tx, output_rx) = broadcast::channel(256);
+        self.inner.processes.lock().await.insert(
+            process_id.clone(),
+            RegisteredProcess {
+                output_tx,
+                status: Arc::clone(&status),
+            },
+        );
+
+        let (writer_tx, mut writer_rx) = mpsc::channel::<Vec<u8>>(128);
+        let client = self.clone();
+        let write_process_id = process_id.clone();
+        tokio::spawn(async move {
+            while let Some(chunk) = writer_rx.recv().await {
+                let request = WriteParams {
+                    process_id: write_process_id.clone(),
+                    chunk: chunk.into(),
+                };
+                if client.write_process(request).await.is_err() {
+                    break;
+                }
+            }
+        });
+
+        let response = match self.request::<_, ExecResponse>(EXEC_METHOD, &params).await {
+            Ok(response) => response,
+            Err(err) => {
+                self.inner.processes.lock().await.remove(&process_id);
+                return Err(err);
+            }
+        };
+
+        if let Some(exit_code) = response.exit_code {
+            status.mark_exited(Some(exit_code));
+        }
+
+        Ok(ExecServerProcess {
+            process_id,
+            pid: response.pid,
+            output_rx,
+            writer_tx,
+            status,
+            client: self.clone(),
+        })
+    }
+
+    async fn initialize(&self) -> Result<(), ExecServerError> {
+        let _: InitializeResponse = self
+            .request(
+                INITIALIZE_METHOD,
+                &InitializeParams {
+                    client_name: "codex-core".to_string(),
+                },
+            )
+            .await?;
+        self.notify(INITIALIZED_METHOD, &serde_json::json!({}))
+            .await
+    }
+
+    async fn write_process(&self, params: WriteParams) -> Result<WriteResponse, ExecServerError> {
+        self.request(EXEC_WRITE_METHOD, &params).await
+    }
+
+    async fn terminate_process(
+        &self,
+        process_id: &str,
+    ) -> Result<TerminateResponse, ExecServerError> {
+        self.request(
+            EXEC_TERMINATE_METHOD,
+            &TerminateParams {
+                process_id: process_id.to_string(),
+            },
+        )
+        .await
+    }
+
+    async fn notify<P: Serialize>(&self, method: &str, params: &P) -> Result<(), ExecServerError> {
+        let params = serde_json::to_value(params)?;
+        self.inner
+            .write_tx
+            .send(JSONRPCMessage::Notification(JSONRPCNotification {
+                method: method.to_string(),
+                params: Some(params),
+            }))
+            .map_err(|_| ExecServerError::Closed)
+    }
+
+    async fn request<P, R>(&self, method: &str, params: &P) -> Result<R, ExecServerError>
+    where
+        P: Serialize,
+        R: DeserializeOwned,
+    {
+        let request_id =
+            RequestId::Integer(self.inner.next_request_id.fetch_add(1, Ordering::SeqCst));
+        let (response_tx, response_rx) = oneshot::channel();
+        self.inner
+            .pending
+            .lock()
+            .await
+            .insert(request_id.clone(), response_tx);
+
+        let params = serde_json::to_value(params)?;
+        let message = JSONRPCMessage::Request(JSONRPCRequest {
+            id: request_id.clone(),
+            method: method.to_string(),
+            params: Some(params),
+            trace: None,
+        });
+
+        if self.inner.write_tx.send(message).is_err() {
+            self.inner.pending.lock().await.remove(&request_id);
+            return Err(ExecServerError::Closed);
+        }
+
+        let result = response_rx.await.map_err(|_| ExecServerError::Closed)?;
+        match result {
+            Ok(value) => serde_json::from_value(value).map_err(ExecServerError::from),
+            Err(error) => Err(ExecServerError::Server {
+                code: error.code,
+                message: error.message,
+            }),
+        }
+    }
+}
+
+async fn handle_server_message(
+    inner: &Arc<Inner>,
+    message: JSONRPCMessage,
+) -> Result<(), ExecServerError> {
+    match message {
+        JSONRPCMessage::Response(JSONRPCResponse { id, result }) => {
+            if let Some(tx) = inner.pending.lock().await.remove(&id) {
+                let _ = tx.send(Ok(result));
+            }
+        }
+        JSONRPCMessage::Error(JSONRPCError { id, error }) => {
+            if let Some(tx) = inner.pending.lock().await.remove(&id) {
+                let _ = tx.send(Err(error));
+            }
+        }
+        JSONRPCMessage::Notification(notification) => {
+            handle_server_notification(inner, notification).await?;
+        }
+        JSONRPCMessage::Request(request) => {
+            return Err(ExecServerError::Protocol(format!(
+                "unexpected exec-server request from child: {}",
+                request.method
+            )));
+        }
+    }
+
+    Ok(())
+}
+
+async fn handle_server_notification(
+    inner: &Arc<Inner>,
+    notification: JSONRPCNotification,
+) -> Result<(), ExecServerError> {
+    match notification.method.as_str() {
+        EXEC_OUTPUT_DELTA_METHOD => {
+            let params: ExecOutputDeltaNotification =
+                serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
+            let chunk = params.chunk.into_inner();
+            let processes = inner.processes.lock().await;
+            if let Some(process) = processes.get(&params.process_id) {
+                let _ = process.output_tx.send(chunk);
+            }
+        }
+        EXEC_EXITED_METHOD => {
+            let params: ExecExitedNotification =
+                serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
+            let mut processes = inner.processes.lock().await;
+            if let Some(process) = processes.remove(&params.process_id) {
+                process.status.mark_exited(Some(params.exit_code));
+            }
+        }
+        other => {
+            debug!("ignoring unknown exec-server notification: {other}");
+        }
+    }
+    Ok(())
+}
+
+async fn handle_transport_shutdown(inner: &Arc<Inner>) {
+    let pending = {
+        let mut pending = inner.pending.lock().await;
+        pending.drain().map(|(_, tx)| tx).collect::<Vec<_>>()
+    };
+    for tx in pending {
+        let _ = tx.send(Err(JSONRPCErrorError {
+            code: -32000,
+            data: None,
+            message: "exec-server transport closed".to_string(),
+        }));
+    }
+
+    let processes = {
+        let mut processes = inner.processes.lock().await;
+        processes
+            .drain()
+            .map(|(_, process)| process)
+            .collect::<Vec<_>>()
+    };
+    for process in processes {
+        process.status.mark_exited(None);
+    }
+}

codex-rs/exec-server/src/lib.rs

@@ -0,0 +1,20 @@
+mod client;
+mod protocol;
+mod server;
+
+pub use client::ExecServerClient;
+pub use client::ExecServerError;
+pub use client::ExecServerLaunchCommand;
+pub use client::ExecServerProcess;
+pub use protocol::ExecExitedNotification;
+pub use protocol::ExecOutputDeltaNotification;
+pub use protocol::ExecOutputStream;
+pub use protocol::ExecParams;
+pub use protocol::ExecResponse;
+pub use protocol::InitializeParams;
+pub use protocol::InitializeResponse;
+pub use protocol::TerminateParams;
+pub use protocol::TerminateResponse;
+pub use protocol::WriteParams;
+pub use protocol::WriteResponse;
+pub use server::run_main;

codex-rs/exec-server/src/protocol.rs

@@ -0,0 +1,144 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP;
+use serde::Deserialize;
+use serde::Serialize;
+
+pub const INITIALIZE_METHOD: &str = "initialize";
+pub const INITIALIZED_METHOD: &str = "initialized";
+pub const EXEC_METHOD: &str = "command/exec";
+pub const EXEC_WRITE_METHOD: &str = "command/exec/write";
+pub const EXEC_TERMINATE_METHOD: &str = "command/exec/terminate";
+pub const EXEC_OUTPUT_DELTA_METHOD: &str = "command/exec/outputDelta";
+pub const EXEC_EXITED_METHOD: &str = "command/exec/exited";
+pub const PROTOCOL_VERSION: &str = "exec-server.v0";
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct ByteChunk(#[serde(with = "base64_bytes")] pub Vec<u8>);
+
+impl ByteChunk {
+    pub fn into_inner(self) -> Vec<u8> {
+        self.0
+    }
+}
+
+impl From<Vec<u8>> for ByteChunk {
+    fn from(value: Vec<u8>) -> Self {
+        Self(value)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct InitializeParams {
+    pub client_name: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct InitializeResponse {
+    pub protocol_version: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecParams {
+    pub process_id: String,
+    pub argv: Vec<String>,
+    pub cwd: PathBuf,
+    pub env: HashMap<String, String>,
+    pub tty: bool,
+    #[serde(default = "default_output_bytes_cap")]
+    pub output_bytes_cap: usize,
+    pub arg0: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecResponse {
+    pub process_id: String,
+    pub pid: Option<u32>,
+    pub running: bool,
+    pub exit_code: Option<i32>,
+    pub stdout: Option<ByteChunk>,
+    pub stderr: Option<ByteChunk>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WriteParams {
+    pub process_id: String,
+    pub chunk: ByteChunk,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WriteResponse {
+    pub accepted: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TerminateParams {
+    pub process_id: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TerminateResponse {
+    pub running: bool,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum ExecOutputStream {
+    Stdout,
+    Stderr,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecOutputDeltaNotification {
+    pub process_id: String,
+    pub stream: ExecOutputStream,
+    pub chunk: ByteChunk,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecExitedNotification {
+    pub process_id: String,
+    pub exit_code: i32,
+}
+
+fn default_output_bytes_cap() -> usize {
+    DEFAULT_OUTPUT_BYTES_CAP
+}
+
+mod base64_bytes {
+    use super::BASE64_STANDARD;
+    use base64::Engine as _;
+    use serde::Deserialize;
+    use serde::Deserializer;
+    use serde::Serializer;
+
+    pub fn serialize<S>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&BASE64_STANDARD.encode(bytes))
+    }
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let encoded = String::deserialize(deserializer)?;
+        BASE64_STANDARD
+            .decode(encoded)
+            .map_err(serde::de::Error::custom)
+    }
+}

codex-rs/exec-server/src/server.rs

@@ -0,0 +1,422 @@
+use std::collections::HashMap;
+use std::collections::VecDeque;
+use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
+
+use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use codex_utils_pty::ExecCommandSession;
+use codex_utils_pty::TerminalSize;
+use serde::Serialize;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::io::BufWriter;
+use tokio::sync::Mutex;
+
+use crate::protocol::EXEC_EXITED_METHOD;
+use crate::protocol::EXEC_METHOD;
+use crate::protocol::EXEC_OUTPUT_DELTA_METHOD;
+use crate::protocol::EXEC_TERMINATE_METHOD;
+use crate::protocol::EXEC_WRITE_METHOD;
+use crate::protocol::ExecExitedNotification;
+use crate::protocol::ExecOutputDeltaNotification;
+use crate::protocol::ExecOutputStream;
+use crate::protocol::ExecParams;
+use crate::protocol::ExecResponse;
+use crate::protocol::INITIALIZE_METHOD;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::InitializeResponse;
+use crate::protocol::PROTOCOL_VERSION;
+use crate::protocol::TerminateParams;
+use crate::protocol::TerminateResponse;
+use crate::protocol::WriteParams;
+use crate::protocol::WriteResponse;
+
+struct RunningProcess {
+    session: ExecCommandSession,
+    tty: bool,
+    stdout_buffer: Arc<StdMutex<BoundedBytesBuffer>>,
+    stderr_buffer: Arc<StdMutex<BoundedBytesBuffer>>,
+}
+
+#[derive(Debug)]
+struct BoundedBytesBuffer {
+    max_bytes: usize,
+    bytes: VecDeque<u8>,
+}
+
+impl BoundedBytesBuffer {
+    fn new(max_bytes: usize) -> Self {
+        Self {
+            max_bytes,
+            bytes: VecDeque::with_capacity(max_bytes.min(8192)),
+        }
+    }
+
+    fn push_chunk(&mut self, chunk: &[u8]) {
+        if self.max_bytes == 0 {
+            return;
+        }
+        for byte in chunk {
+            self.bytes.push_back(*byte);
+            if self.bytes.len() > self.max_bytes {
+                self.bytes.pop_front();
+            }
+        }
+    }
+
+    fn snapshot(&self) -> Vec<u8> {
+        self.bytes.iter().copied().collect()
+    }
+}
+
+pub async fn run_main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let writer = Arc::new(Mutex::new(BufWriter::new(tokio::io::stdout())));
+    let processes = Arc::new(Mutex::new(HashMap::<String, RunningProcess>::new()));
+    let mut lines = BufReader::new(tokio::io::stdin()).lines();
+
+    while let Some(line) = lines.next_line().await? {
+        if line.trim().is_empty() {
+            continue;
+        }
+
+        let message = serde_json::from_str::<JSONRPCMessage>(&line)?;
+        if let JSONRPCMessage::Request(request) = message {
+            handle_request(request, &writer, &processes).await;
+            continue;
+        }
+
+        if let JSONRPCMessage::Notification(notification) = message {
+            if notification.method != INITIALIZED_METHOD {
+                send_error(
+                    &writer,
+                    RequestId::Integer(-1),
+                    invalid_request(format!(
+                        "unexpected notification method: {}",
+                        notification.method
+                    )),
+                )
+                .await;
+            }
+            continue;
+        }
+    }
+
+    let remaining = {
+        let mut processes = processes.lock().await;
+        processes
+            .drain()
+            .map(|(_, process)| process)
+            .collect::<Vec<_>>()
+    };
+    for process in remaining {
+        process.session.terminate();
+    }
+
+    Ok(())
+}
+
+async fn handle_request(
+    request: JSONRPCRequest,
+    writer: &Arc<Mutex<BufWriter<tokio::io::Stdout>>>,
+    processes: &Arc<Mutex<HashMap<String, RunningProcess>>>,
+) {
+    let response = match request.method.as_str() {
+        INITIALIZE_METHOD => serde_json::to_value(InitializeResponse {
+            protocol_version: PROTOCOL_VERSION.to_string(),
+        })
+        .map_err(|err| internal_error(err.to_string())),
+        EXEC_METHOD => handle_exec_request(request.params, writer, processes).await,
+        EXEC_WRITE_METHOD => handle_write_request(request.params, processes).await,
+        EXEC_TERMINATE_METHOD => handle_terminate_request(request.params, processes).await,
+        other => Err(invalid_request(format!("unknown method: {other}"))),
+    };
+
+    match response {
+        Ok(result) => {
+            send_response(
+                writer,
+                JSONRPCResponse {
+                    id: request.id,
+                    result,
+                },
+            )
+            .await;
+        }
+        Err(err) => {
+            send_error(writer, request.id, err).await;
+        }
+    }
+}
+
+async fn handle_exec_request(
+    params: Option<serde_json::Value>,
+    writer: &Arc<Mutex<BufWriter<tokio::io::Stdout>>>,
+    processes: &Arc<Mutex<HashMap<String, RunningProcess>>>,
+) -> Result<serde_json::Value, JSONRPCErrorError> {
+    let params: ExecParams = serde_json::from_value(params.unwrap_or(serde_json::Value::Null))
+        .map_err(|err| invalid_params(err.to_string()))?;
+
+    let (program, args) = params
+        .argv
+        .split_first()
+        .ok_or_else(|| invalid_params("argv must not be empty".to_string()))?;
+
+    let spawned = if params.tty {
+        codex_utils_pty::spawn_pty_process(
+            program,
+            args,
+            params.cwd.as_path(),
+            &params.env,
+            &params.arg0,
+            TerminalSize::default(),
+        )
+        .await
+    } else {
+        codex_utils_pty::spawn_pipe_process_no_stdin(
+            program,
+            args,
+            params.cwd.as_path(),
+            &params.env,
+            &params.arg0,
+        )
+        .await
+    }
+    .map_err(|err| internal_error(err.to_string()))?;
+
+    let pid = spawned.session.pid();
+    let stdout_buffer = Arc::new(StdMutex::new(BoundedBytesBuffer::new(
+        params.output_bytes_cap,
+    )));
+    let stderr_buffer = Arc::new(StdMutex::new(BoundedBytesBuffer::new(
+        params.output_bytes_cap,
+    )));
+
+    let process_id = params.process_id.clone();
+    {
+        let mut process_map = processes.lock().await;
+        if process_map.contains_key(&process_id) {
+            spawned.session.terminate();
+            return Err(invalid_request(format!(
+                "process {} already exists",
+                params.process_id
+            )));
+        }
+        process_map.insert(
+            process_id.clone(),
+            RunningProcess {
+                session: spawned.session,
+                tty: params.tty,
+                stdout_buffer: Arc::clone(&stdout_buffer),
+                stderr_buffer: Arc::clone(&stderr_buffer),
+            },
+        );
+    }
+
+    tokio::spawn(stream_output(
+        process_id.clone(),
+        ExecOutputStream::Stdout,
+        spawned.stdout_rx,
+        Arc::clone(writer),
+        Arc::clone(&stdout_buffer),
+    ));
+    tokio::spawn(stream_output(
+        process_id.clone(),
+        ExecOutputStream::Stderr,
+        spawned.stderr_rx,
+        Arc::clone(writer),
+        Arc::clone(&stderr_buffer),
+    ));
+    tokio::spawn(watch_exit(
+        process_id.clone(),
+        spawned.exit_rx,
+        Arc::clone(writer),
+        Arc::clone(processes),
+    ));
+
+    serde_json::to_value(ExecResponse {
+        process_id,
+        pid,
+        running: true,
+        exit_code: None,
+        stdout: None,
+        stderr: None,
+    })
+    .map_err(|err| internal_error(err.to_string()))
+}
+
+async fn handle_write_request(
+    params: Option<serde_json::Value>,
+    processes: &Arc<Mutex<HashMap<String, RunningProcess>>>,
+) -> Result<serde_json::Value, JSONRPCErrorError> {
+    let params: WriteParams = serde_json::from_value(params.unwrap_or(serde_json::Value::Null))
+        .map_err(|err| invalid_params(err.to_string()))?;
+
+    let writer_tx = {
+        let process_map = processes.lock().await;
+        let process = process_map
+            .get(&params.process_id)
+            .ok_or_else(|| invalid_request(format!("unknown process id {}", params.process_id)))?;
+        if !process.tty {
+            return Err(invalid_request(format!(
+                "stdin is closed for process {}",
+                params.process_id
+            )));
+        }
+        process.session.writer_sender()
+    };
+
+    writer_tx
+        .send(params.chunk.into_inner())
+        .await
+        .map_err(|_| internal_error("failed to write to process stdin".to_string()))?;
+
+    serde_json::to_value(WriteResponse { accepted: true })
+        .map_err(|err| internal_error(err.to_string()))
+}
+
+async fn handle_terminate_request(
+    params: Option<serde_json::Value>,
+    processes: &Arc<Mutex<HashMap<String, RunningProcess>>>,
+) -> Result<serde_json::Value, JSONRPCErrorError> {
+    let params: TerminateParams = serde_json::from_value(params.unwrap_or(serde_json::Value::Null))
+        .map_err(|err| invalid_params(err.to_string()))?;
+
+    let process = {
+        let mut process_map = processes.lock().await;
+        process_map.remove(&params.process_id)
+    };
+
+    if let Some(process) = process {
+        process.session.terminate();
+        serde_json::to_value(TerminateResponse { running: true })
+            .map_err(|err| internal_error(err.to_string()))
+    } else {
+        serde_json::to_value(TerminateResponse { running: false })
+            .map_err(|err| internal_error(err.to_string()))
+    }
+}
+
+async fn stream_output(
+    process_id: String,
+    stream: ExecOutputStream,
+    mut receiver: tokio::sync::mpsc::Receiver<Vec<u8>>,
+    writer: Arc<Mutex<BufWriter<tokio::io::Stdout>>>,
+    buffer: Arc<StdMutex<BoundedBytesBuffer>>,
+) {
+    while let Some(chunk) = receiver.recv().await {
+        if let Ok(mut guard) = buffer.lock() {
+            guard.push_chunk(&chunk);
+        }
+        let notification = ExecOutputDeltaNotification {
+            process_id: process_id.clone(),
+            stream,
+            chunk: chunk.into(),
+        };
+        if send_notification(&writer, EXEC_OUTPUT_DELTA_METHOD, &notification)
+            .await
+            .is_err()
+        {
+            break;
+        }
+    }
+}
+
+async fn watch_exit(
+    process_id: String,
+    exit_rx: tokio::sync::oneshot::Receiver<i32>,
+    writer: Arc<Mutex<BufWriter<tokio::io::Stdout>>>,
+    processes: Arc<Mutex<HashMap<String, RunningProcess>>>,
+) {
+    let exit_code = exit_rx.await.unwrap_or(-1);
+    let removed = {
+        let mut processes = processes.lock().await;
+        processes.remove(&process_id)
+    };
+    if let Some(process) = removed {
+        let _ = process.stdout_buffer.lock().map(|buffer| buffer.snapshot());
+        let _ = process.stderr_buffer.lock().map(|buffer| buffer.snapshot());
+    }
+    let _ = send_notification(
+        &writer,
+        EXEC_EXITED_METHOD,
+        &ExecExitedNotification {
+            process_id,
+            exit_code,
+        },
+    )
+    .await;
+}
+
+async fn send_response(
+    writer: &Arc<Mutex<BufWriter<tokio::io::Stdout>>>,
+    response: JSONRPCResponse,
+) {
+    let _ = send_message(writer, JSONRPCMessage::Response(response)).await;
+}
+
+async fn send_error(
+    writer: &Arc<Mutex<BufWriter<tokio::io::Stdout>>>,
+    id: RequestId,
+    error: JSONRPCErrorError,
+) {
+    let _ = send_message(writer, JSONRPCMessage::Error(JSONRPCError { error, id })).await;
+}
+
+async fn send_notification<T: Serialize>(
+    writer: &Arc<Mutex<BufWriter<tokio::io::Stdout>>>,
+    method: &str,
+    params: &T,
+) -> Result<(), serde_json::Error> {
+    send_message(
+        writer,
+        JSONRPCMessage::Notification(JSONRPCNotification {
+            method: method.to_string(),
+            params: Some(serde_json::to_value(params)?),
+        }),
+    )
+    .await
+    .map_err(serde_json::Error::io)
+}
+
+async fn send_message(
+    writer: &Arc<Mutex<BufWriter<tokio::io::Stdout>>>,
+    message: JSONRPCMessage,
+) -> std::io::Result<()> {
+    let encoded =
+        serde_json::to_vec(&message).map_err(|err| std::io::Error::other(err.to_string()))?;
+    let mut writer = writer.lock().await;
+    writer.write_all(&encoded).await?;
+    writer.write_all(b"\n").await?;
+    writer.flush().await
+}
+
+fn invalid_request(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32600,
+        data: None,
+        message,
+    }
+}
+
+fn invalid_params(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32602,
+        data: None,
+        message,
+    }
+}
+
+fn internal_error(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32603,
+        data: None,
+        message,
+    }
+}

codex-rs/exec-server/tests/stdio_smoke.rs

@@ -0,0 +1,141 @@
+#![cfg(unix)]
+
+use std::process::Stdio;
+use std::time::Duration;
+
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use codex_exec_server::ExecParams;
+use codex_exec_server::ExecServerClient;
+use codex_exec_server::ExecServerLaunchCommand;
+use codex_exec_server::InitializeParams;
+use codex_exec_server::InitializeResponse;
+use codex_utils_cargo_bin::cargo_bin;
+use pretty_assertions::assert_eq;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::process::Command;
+use tokio::sync::broadcast;
+use tokio::time::timeout;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_accepts_initialize_over_stdio() -> anyhow::Result<()> {
+    let binary = cargo_bin("codex-exec-server")?;
+    let mut child = Command::new(binary);
+    child.stdin(Stdio::piped());
+    child.stdout(Stdio::piped());
+    child.stderr(Stdio::inherit());
+    let mut child = child.spawn()?;
+
+    let mut stdin = child.stdin.take().expect("stdin");
+    let stdout = child.stdout.take().expect("stdout");
+    let mut stdout = BufReader::new(stdout).lines();
+
+    let initialize = JSONRPCMessage::Request(JSONRPCRequest {
+        id: RequestId::Integer(1),
+        method: "initialize".to_string(),
+        params: Some(serde_json::to_value(InitializeParams {
+            client_name: "exec-server-test".to_string(),
+        })?),
+        trace: None,
+    });
+    stdin
+        .write_all(format!("{}\n", serde_json::to_string(&initialize)?).as_bytes())
+        .await?;
+
+    let response_line = timeout(Duration::from_secs(5), stdout.next_line()).await??;
+    let response_line = response_line.expect("response line");
+    let response: JSONRPCMessage = serde_json::from_str(&response_line)?;
+    let JSONRPCMessage::Response(JSONRPCResponse { id, result }) = response else {
+        panic!("expected initialize response");
+    };
+    assert_eq!(id, RequestId::Integer(1));
+    let initialize_response: InitializeResponse = serde_json::from_value(result)?;
+    assert_eq!(initialize_response.protocol_version, "exec-server.v0");
+
+    let initialized = JSONRPCMessage::Notification(JSONRPCNotification {
+        method: "initialized".to_string(),
+        params: Some(serde_json::json!({})),
+    });
+    stdin
+        .write_all(format!("{}\n", serde_json::to_string(&initialized)?).as_bytes())
+        .await?;
+
+    child.start_kill()?;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_client_streams_output_and_accepts_writes() -> anyhow::Result<()> {
+    let mut env = std::collections::HashMap::new();
+    if let Some(path) = std::env::var_os("PATH") {
+        env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
+    }
+
+    let client = ExecServerClient::spawn(ExecServerLaunchCommand {
+        program: cargo_bin("codex-exec-server")?,
+        args: Vec::new(),
+    })
+    .await?;
+
+    let process = client
+        .start_process(ExecParams {
+            process_id: "2001".to_string(),
+            argv: vec![
+                "bash".to_string(),
+                "-lc".to_string(),
+                "printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"
+                    .to_string(),
+            ],
+            cwd: std::env::current_dir()?,
+            env,
+            tty: true,
+            output_bytes_cap: 4096,
+            arg0: None,
+        })
+        .await?;
+
+    let mut output = process.output_receiver();
+    assert!(
+        recv_until_contains(&mut output, "ready")
+            .await?
+            .contains("ready"),
+        "expected initial ready output"
+    );
+
+    process
+        .writer_sender()
+        .send(b"hello\n".to_vec())
+        .await
+        .expect("write should succeed");
+
+    assert!(
+        recv_until_contains(&mut output, "echo:hello")
+            .await?
+            .contains("echo:hello"),
+        "expected echoed output"
+    );
+
+    process.terminate();
+    Ok(())
+}
+
+async fn recv_until_contains(
+    output: &mut broadcast::Receiver<Vec<u8>>,
+    needle: &str,
+) -> anyhow::Result<String> {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+    let mut collected = String::new();
+    loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        let chunk = timeout(remaining, output.recv()).await??;
+        collected.push_str(&String::from_utf8_lossy(&chunk));
+        if collected.contains(needle) {
+            return Ok(collected);
+        }
+    }
+}

codex-rs/utils/pty/src/pipe.rs

@@ -238,6 +238,7 @@ async fn spawn_process_with_stdin_mode(
         wait_handle,
         exit_status,
         exit_code,
+        Some(pid),
         None,
     );
 

codex-rs/utils/pty/src/process.rs

@@ -79,6 +79,7 @@ pub struct ProcessHandle {
     wait_handle: StdMutex<Option<JoinHandle<()>>>,
     exit_status: Arc<AtomicBool>,
     exit_code: Arc<StdMutex<Option<i32>>>,
+    pid: Option<u32>,
     // PtyHandles must be preserved because the process will receive Control+C if the
     // slave is closed
     _pty_handles: StdMutex<Option<PtyHandles>>,
@@ -101,6 +102,7 @@ impl ProcessHandle {
         wait_handle: JoinHandle<()>,
         exit_status: Arc<AtomicBool>,
         exit_code: Arc<StdMutex<Option<i32>>>,
+        pid: Option<u32>,
         pty_handles: Option<PtyHandles>,
     ) -> Self {
         Self {
@@ -112,6 +114,7 @@ impl ProcessHandle {
             wait_handle: StdMutex::new(Some(wait_handle)),
             exit_status,
             exit_code,
+            pid,
             _pty_handles: StdMutex::new(pty_handles),
         }
     }
@@ -139,6 +142,11 @@ impl ProcessHandle {
         self.exit_code.lock().ok().and_then(|guard| *guard)
     }
 
+    /// Returns the OS process ID when known.
+    pub fn pid(&self) -> Option<u32> {
+        self.pid
+    }
+
     /// Resize the PTY in character cells.
     pub fn resize(&self, size: TerminalSize) -> anyhow::Result<()> {
         let handles = self

codex-rs/utils/pty/src/pty.rs

@@ -159,11 +159,12 @@ async fn spawn_process_portable(
     }
 
     let mut child = pair.slave.spawn_command(command_builder)?;
+    let pid = child.process_id();
     #[cfg(unix)]
     // portable-pty establishes the spawned PTY child as a new session leader on
     // Unix, so PID == PGID and we can reuse the pipe backend's process-group
     // hard-kill semantics for descendants.
-    let process_group_id = child.process_id();
+    let process_group_id = pid;
     let killer = child.clone_killer();
 
     let (writer_tx, mut writer_rx) = mpsc::channel::<Vec<u8>>(128);
@@ -241,6 +242,7 @@ async fn spawn_process_portable(
         wait_handle,
         exit_status,
         exit_code,
+        pid,
         Some(handles),
     );
 
@@ -394,6 +396,7 @@ async fn spawn_process_preserving_fds(
         wait_handle,
         exit_status,
         exit_code,
+        Some(process_group_id),
         Some(handles),
     );