Refactor exec-server client RPC plumbing

Extract a small generic JSON-RPC client core, move remote connection setup into the JSON-RPC backend, and make handler request execution concurrency-safe so out-of-order responses and duplicate process ID races work correctly.

Co-authored-by: Codex <noreply@openai.com>

codex-rs/Cargo.lock

@@ -2011,6 +2011,7 @@ dependencies = [
  "base64 0.22.1",
  "clap",
  "codex-app-server-protocol",
+ "codex-environment",
  "codex-utils-cargo-bin",
  "codex-utils-pty",
  "futures",

codex-rs/exec-server/Cargo.toml

@@ -15,6 +15,7 @@ workspace = true
 base64 = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
 codex-app-server-protocol = { workspace = true }
+codex-environment = { workspace = true }
 codex-utils-pty = { workspace = true }
 futures = { workspace = true }
 serde = { workspace = true, features = ["derive"] }

codex-rs/exec-server/README.md

@@ -121,7 +121,8 @@ Request params:
     "PATH": "/usr/bin:/bin"
   },
   "tty": true,
-  "arg0": null
+  "arg0": null,
+  "sandbox": null
 }
 ```
 
@@ -133,6 +134,9 @@ Field definitions:
 - `tty`: when `true`, spawn a PTY-backed interactive process; when `false`,
   spawn a pipe-backed process with closed stdin.
 - `arg0`: optional argv0 override forwarded to `codex-utils-pty`.
+- `sandbox`: optional sandbox config. Omit it for the current direct-spawn
+  behavior. Explicit `{"mode":"none"}` is accepted; `{"mode":"hostDefault"}`
+  is currently rejected until host-local sandbox materialization is wired up.
 
 Response:
 

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

@@ -0,0 +1,98 @@
+use serde::Serialize;
+use tokio::io::AsyncRead;
+use tokio::io::AsyncWrite;
+use tokio::sync::mpsc;
+use tokio::time::timeout;
+use tokio_tungstenite::connect_async;
+
+use crate::client_api::RemoteExecServerConnectArgs;
+use crate::connection::JsonRpcConnection;
+use crate::rpc::RpcCallError;
+use crate::rpc::RpcClient;
+use crate::rpc::RpcClientEvent;
+
+use super::ExecServerError;
+
+pub(super) struct JsonRpcBackend {
+    rpc: RpcClient,
+}
+
+impl JsonRpcBackend {
+    pub(super) fn connect_stdio<R, W>(stdin: W, stdout: R) -> (Self, mpsc::Receiver<RpcClientEvent>)
+    where
+        R: AsyncRead + Unpin + Send + 'static,
+        W: AsyncWrite + Unpin + Send + 'static,
+    {
+        Self::connect(JsonRpcConnection::from_stdio(
+            stdout,
+            stdin,
+            "exec-server stdio".to_string(),
+        ))
+    }
+
+    pub(super) async fn connect_websocket(
+        args: &RemoteExecServerConnectArgs,
+    ) -> Result<(Self, mpsc::Receiver<RpcClientEvent>), ExecServerError> {
+        let websocket_url = args.websocket_url.clone();
+        let connect_timeout = args.connect_timeout;
+        let (stream, _) = timeout(connect_timeout, connect_async(websocket_url.as_str()))
+            .await
+            .map_err(|_| ExecServerError::WebSocketConnectTimeout {
+                url: websocket_url.clone(),
+                timeout: connect_timeout,
+            })?
+            .map_err(|source| ExecServerError::WebSocketConnect {
+                url: websocket_url.clone(),
+                source,
+            })?;
+
+        Ok(Self::connect(JsonRpcConnection::from_websocket(
+            stream,
+            format!("exec-server websocket {websocket_url}"),
+        )))
+    }
+
+    fn connect(connection: JsonRpcConnection) -> (Self, mpsc::Receiver<RpcClientEvent>) {
+        let (rpc, events_rx) = RpcClient::new(connection);
+        (Self { rpc }, events_rx)
+    }
+
+    pub(super) async fn notify<P: Serialize>(
+        &self,
+        method: &str,
+        params: &P,
+    ) -> Result<(), ExecServerError> {
+        self.rpc
+            .notify(method, params)
+            .await
+            .map_err(|err| match err.classify() {
+                serde_json::error::Category::Io => ExecServerError::Closed,
+                serde_json::error::Category::Syntax
+                | serde_json::error::Category::Data
+                | serde_json::error::Category::Eof => ExecServerError::Json(err),
+            })
+    }
+
+    pub(super) async fn call<P, T>(&self, method: &str, params: &P) -> Result<T, ExecServerError>
+    where
+        P: Serialize,
+        T: serde::de::DeserializeOwned,
+    {
+        self.rpc
+            .call(method, params)
+            .await
+            .map_err(|err| match err {
+                RpcCallError::Closed => ExecServerError::Closed,
+                RpcCallError::Json(err) => ExecServerError::Json(err),
+                RpcCallError::Server(error) => ExecServerError::Server {
+                    code: error.code,
+                    message: error.message,
+                },
+            })
+    }
+
+    #[cfg(test)]
+    pub(super) async fn pending_request_count(&self) -> usize {
+        self.rpc.pending_request_count().await
+    }
+}

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

@@ -0,0 +1,137 @@
+use std::sync::Arc;
+
+use crate::protocol::ExecParams;
+use crate::protocol::ExecResponse;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::InitializeResponse;
+use crate::protocol::ReadParams;
+use crate::protocol::ReadResponse;
+use crate::protocol::TerminateParams;
+use crate::protocol::TerminateResponse;
+use crate::protocol::WriteParams;
+use crate::protocol::WriteResponse;
+use crate::server::ExecServerHandler;
+use codex_app_server_protocol::FsCopyParams;
+use codex_app_server_protocol::FsCopyResponse;
+use codex_app_server_protocol::FsCreateDirectoryParams;
+use codex_app_server_protocol::FsCreateDirectoryResponse;
+use codex_app_server_protocol::FsGetMetadataParams;
+use codex_app_server_protocol::FsGetMetadataResponse;
+use codex_app_server_protocol::FsReadDirectoryParams;
+use codex_app_server_protocol::FsReadDirectoryResponse;
+use codex_app_server_protocol::FsReadFileParams;
+use codex_app_server_protocol::FsReadFileResponse;
+use codex_app_server_protocol::FsRemoveParams;
+use codex_app_server_protocol::FsRemoveResponse;
+use codex_app_server_protocol::FsWriteFileParams;
+use codex_app_server_protocol::FsWriteFileResponse;
+
+use super::ExecServerError;
+use super::server_result_to_client;
+
+#[derive(Clone)]
+pub(super) struct LocalBackend {
+    handler: Arc<ExecServerHandler>,
+}
+
+impl LocalBackend {
+    pub(super) fn new(handler: ExecServerHandler) -> Self {
+        Self {
+            handler: Arc::new(handler),
+        }
+    }
+
+    pub(super) async fn shutdown(&self) {
+        self.handler.shutdown().await;
+    }
+
+    pub(super) async fn initialize(&self) -> Result<InitializeResponse, ExecServerError> {
+        server_result_to_client(self.handler.initialize())
+    }
+
+    pub(super) async fn notify(&self, method: &str) -> Result<(), ExecServerError> {
+        match method {
+            INITIALIZED_METHOD => self
+                .handler
+                .initialized()
+                .map_err(ExecServerError::Protocol),
+            other => Err(ExecServerError::Protocol(format!(
+                "unsupported in-process notification method `{other}`"
+            ))),
+        }
+    }
+
+    pub(super) async fn exec(&self, params: ExecParams) -> Result<ExecResponse, ExecServerError> {
+        server_result_to_client(self.handler.exec(params).await)
+    }
+
+    pub(super) async fn exec_read(
+        &self,
+        params: ReadParams,
+    ) -> Result<ReadResponse, ExecServerError> {
+        server_result_to_client(self.handler.exec_read(params).await)
+    }
+
+    pub(super) async fn exec_write(
+        &self,
+        params: WriteParams,
+    ) -> Result<WriteResponse, ExecServerError> {
+        server_result_to_client(self.handler.exec_write(params).await)
+    }
+
+    pub(super) async fn terminate(
+        &self,
+        params: TerminateParams,
+    ) -> Result<TerminateResponse, ExecServerError> {
+        server_result_to_client(self.handler.terminate(params).await)
+    }
+
+    pub(super) async fn fs_read_file(
+        &self,
+        params: FsReadFileParams,
+    ) -> Result<FsReadFileResponse, ExecServerError> {
+        server_result_to_client(self.handler.fs_read_file(params).await)
+    }
+
+    pub(super) async fn fs_write_file(
+        &self,
+        params: FsWriteFileParams,
+    ) -> Result<FsWriteFileResponse, ExecServerError> {
+        server_result_to_client(self.handler.fs_write_file(params).await)
+    }
+
+    pub(super) async fn fs_create_directory(
+        &self,
+        params: FsCreateDirectoryParams,
+    ) -> Result<FsCreateDirectoryResponse, ExecServerError> {
+        server_result_to_client(self.handler.fs_create_directory(params).await)
+    }
+
+    pub(super) async fn fs_get_metadata(
+        &self,
+        params: FsGetMetadataParams,
+    ) -> Result<FsGetMetadataResponse, ExecServerError> {
+        server_result_to_client(self.handler.fs_get_metadata(params).await)
+    }
+
+    pub(super) async fn fs_read_directory(
+        &self,
+        params: FsReadDirectoryParams,
+    ) -> Result<FsReadDirectoryResponse, ExecServerError> {
+        server_result_to_client(self.handler.fs_read_directory(params).await)
+    }
+
+    pub(super) async fn fs_remove(
+        &self,
+        params: FsRemoveParams,
+    ) -> Result<FsRemoveResponse, ExecServerError> {
+        server_result_to_client(self.handler.fs_remove(params).await)
+    }
+
+    pub(super) async fn fs_copy(
+        &self,
+        params: FsCopyParams,
+    ) -> Result<FsCopyResponse, ExecServerError> {
+        server_result_to_client(self.handler.fs_copy(params).await)
+    }
+}

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

@@ -0,0 +1,72 @@
+use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+
+use tokio::sync::broadcast;
+
+use super::ExecServerClient;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(super) struct ExecServerOutput {
+    pub(super) stream: crate::protocol::ExecOutputStream,
+    pub(super) chunk: Vec<u8>,
+}
+
+pub(super) struct ExecServerProcess {
+    pub(super) process_id: String,
+    pub(super) output_rx: broadcast::Receiver<ExecServerOutput>,
+    pub(super) status: Arc<RemoteProcessStatus>,
+    pub(super) client: ExecServerClient,
+}
+
+impl ExecServerProcess {
+    pub(super) fn output_receiver(&self) -> broadcast::Receiver<ExecServerOutput> {
+        self.output_rx.resubscribe()
+    }
+
+    pub(super) fn has_exited(&self) -> bool {
+        self.status.has_exited()
+    }
+
+    pub(super) fn exit_code(&self) -> Option<i32> {
+        self.status.exit_code()
+    }
+
+    pub(super) fn terminate(&self) {
+        let client = self.client.clone();
+        let process_id = self.process_id.clone();
+        tokio::spawn(async move {
+            let _ = client.terminate(&process_id).await;
+        });
+    }
+}
+
+pub(super) struct RemoteProcessStatus {
+    exited: AtomicBool,
+    exit_code: StdMutex<Option<i32>>,
+}
+
+impl RemoteProcessStatus {
+    pub(super) fn new() -> Self {
+        Self {
+            exited: AtomicBool::new(false),
+            exit_code: StdMutex::new(None),
+        }
+    }
+
+    pub(super) fn has_exited(&self) -> bool {
+        self.exited.load(Ordering::SeqCst)
+    }
+
+    pub(super) fn exit_code(&self) -> Option<i32> {
+        self.exit_code.lock().ok().and_then(|guard| *guard)
+    }
+
+    pub(super) 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;
+        }
+    }
+}

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

@@ -0,0 +1,981 @@
+use std::collections::HashMap;
+use std::time::Duration;
+
+use pretty_assertions::assert_eq;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::time::timeout;
+
+use super::ExecServerClient;
+use super::ExecServerClientConnectOptions;
+use super::ExecServerError;
+use super::ExecServerOutput;
+use crate::protocol::EXEC_METHOD;
+use crate::protocol::EXEC_OUTPUT_DELTA_METHOD;
+use crate::protocol::EXEC_READ_METHOD;
+use crate::protocol::EXEC_TERMINATE_METHOD;
+use crate::protocol::ExecOutputStream;
+use crate::protocol::ExecParams;
+use crate::protocol::INITIALIZE_METHOD;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::PROTOCOL_VERSION;
+use crate::protocol::ReadParams;
+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;
+
+fn test_options() -> ExecServerClientConnectOptions {
+    ExecServerClientConnectOptions {
+        client_name: "test-client".to_string(),
+        initialize_timeout: Duration::from_secs(1),
+    }
+}
+
+async fn read_jsonrpc_line<R>(lines: &mut tokio::io::Lines<BufReader<R>>) -> JSONRPCMessage
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    let next_line = timeout(Duration::from_secs(1), lines.next_line()).await;
+    let line_result = match next_line {
+        Ok(line_result) => line_result,
+        Err(err) => panic!("timed out waiting for JSON-RPC line: {err}"),
+    };
+    let maybe_line = match line_result {
+        Ok(maybe_line) => maybe_line,
+        Err(err) => panic!("failed to read JSON-RPC line: {err}"),
+    };
+    let line = match maybe_line {
+        Some(line) => line,
+        None => panic!("server connection closed before JSON-RPC line arrived"),
+    };
+    match serde_json::from_str::<JSONRPCMessage>(&line) {
+        Ok(message) => message,
+        Err(err) => panic!("failed to parse JSON-RPC line: {err}"),
+    }
+}
+
+async fn write_jsonrpc_line<W>(writer: &mut W, message: JSONRPCMessage)
+where
+    W: tokio::io::AsyncWrite + Unpin,
+{
+    let encoded = match serde_json::to_string(&message) {
+        Ok(encoded) => encoded,
+        Err(err) => panic!("failed to encode JSON-RPC message: {err}"),
+    };
+    if let Err(err) = writer.write_all(format!("{encoded}\n").as_bytes()).await {
+        panic!("failed to write JSON-RPC line: {err}");
+    }
+}
+
+#[tokio::test]
+async fn connect_stdio_performs_initialize_handshake() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (mut server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    let server = tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+
+        let initialize = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(request) = initialize else {
+            panic!("expected initialize request");
+        };
+        assert_eq!(request.method, INITIALIZE_METHOD);
+        assert_eq!(
+            request.params,
+            Some(serde_json::json!({ "clientName": "test-client" }))
+        );
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: request.id,
+                result: serde_json::json!({ "protocolVersion": PROTOCOL_VERSION }),
+            }),
+        )
+        .await;
+
+        let initialized = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Notification(JSONRPCNotification { method, params }) = initialized
+        else {
+            panic!("expected initialized notification");
+        };
+        assert_eq!(method, INITIALIZED_METHOD);
+        assert_eq!(params, Some(serde_json::json!({})));
+    });
+
+    let client = ExecServerClient::connect_stdio(client_stdin, client_stdout, test_options()).await;
+    if let Err(err) = client {
+        panic!("failed to connect test client: {err}");
+    }
+
+    if let Err(err) = server.await {
+        panic!("server task failed: {err}");
+    }
+}
+
+#[tokio::test]
+async fn connect_stdio_matches_out_of_order_responses_by_request_id() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (mut server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    let server = tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+
+        let initialize = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(initialize) = initialize else {
+            panic!("expected initialize request");
+        };
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: initialize.id,
+                result: serde_json::json!({ "protocolVersion": PROTOCOL_VERSION }),
+            }),
+        )
+        .await;
+
+        let initialized = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Notification(_) = initialized else {
+            panic!("expected initialized notification");
+        };
+
+        let first = read_jsonrpc_line(&mut lines).await;
+        let second = read_jsonrpc_line(&mut lines).await;
+        let (read_request, terminate_request) = match (first, second) {
+            (JSONRPCMessage::Request(first_request), JSONRPCMessage::Request(second_request))
+                if first_request.method == EXEC_READ_METHOD
+                    && second_request.method == EXEC_TERMINATE_METHOD =>
+            {
+                (first_request, second_request)
+            }
+            (JSONRPCMessage::Request(first_request), JSONRPCMessage::Request(second_request))
+                if first_request.method == EXEC_TERMINATE_METHOD
+                    && second_request.method == EXEC_READ_METHOD =>
+            {
+                (second_request, first_request)
+            }
+            _ => panic!("expected read and terminate requests"),
+        };
+
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: terminate_request.id,
+                result: serde_json::json!({ "running": false }),
+            }),
+        )
+        .await;
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: read_request.id,
+                result: serde_json::json!({
+                    "chunks": [],
+                    "nextSeq": 1,
+                    "exited": false,
+                    "exitCode": null,
+                }),
+            }),
+        )
+        .await;
+    });
+
+    let client = ExecServerClient::connect_stdio(client_stdin, client_stdout, test_options())
+        .await
+        .unwrap_or_else(|err| panic!("failed to connect test client: {err}"));
+
+    let (read, terminate) = tokio::join!(
+        client.read(ReadParams {
+            process_id: "proc-1".to_string(),
+            after_seq: None,
+            max_bytes: None,
+            wait_ms: Some(0),
+        }),
+        client.terminate("proc-1"),
+    );
+
+    let read = read.unwrap_or_else(|err| panic!("read failed: {err}"));
+    let terminate = terminate.unwrap_or_else(|err| panic!("terminate failed: {err}"));
+    assert_eq!(read.next_seq, 1);
+    assert!(!read.exited);
+    assert!(!terminate.running);
+
+    if let Err(err) = server.await {
+        panic!("server task failed: {err}");
+    }
+}
+
+#[tokio::test]
+async fn connect_in_process_starts_processes_without_jsonrpc_transport() {
+    let client = match ExecServerClient::connect_in_process(test_options()).await {
+        Ok(client) => client,
+        Err(err) => panic!("failed to connect in-process client: {err}"),
+    };
+
+    let process = match client
+        .start_process(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec!["printf".to_string(), "hello".to_string()],
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: false,
+            arg0: None,
+            sandbox: None,
+        })
+        .await
+    {
+        Ok(process) => process,
+        Err(err) => panic!("failed to start in-process child: {err}"),
+    };
+
+    let mut output = process.output_receiver();
+    let output = timeout(Duration::from_secs(1), output.recv())
+        .await
+        .unwrap_or_else(|err| panic!("timed out waiting for process output: {err}"))
+        .unwrap_or_else(|err| panic!("failed to receive process output: {err}"));
+    assert_eq!(
+        output,
+        ExecServerOutput {
+            stream: crate::protocol::ExecOutputStream::Stdout,
+            chunk: b"hello".to_vec(),
+        }
+    );
+}
+
+#[tokio::test]
+async fn connect_in_process_read_returns_retained_output_and_exit_state() {
+    let client = match ExecServerClient::connect_in_process(test_options()).await {
+        Ok(client) => client,
+        Err(err) => panic!("failed to connect in-process client: {err}"),
+    };
+
+    let response = match client
+        .exec(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec!["printf".to_string(), "hello".to_string()],
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: false,
+            arg0: None,
+            sandbox: None,
+        })
+        .await
+    {
+        Ok(response) => response,
+        Err(err) => panic!("failed to start in-process child: {err}"),
+    };
+
+    let process_id = response.process_id.clone();
+    let read = match client
+        .read(ReadParams {
+            process_id: process_id.clone(),
+            after_seq: None,
+            max_bytes: None,
+            wait_ms: Some(1000),
+        })
+        .await
+    {
+        Ok(read) => read,
+        Err(err) => panic!("failed to read in-process child output: {err}"),
+    };
+
+    assert_eq!(read.chunks.len(), 1);
+    assert_eq!(read.chunks[0].seq, 1);
+    assert_eq!(read.chunks[0].stream, ExecOutputStream::Stdout);
+    assert_eq!(read.chunks[0].chunk.clone().into_inner(), b"hello".to_vec());
+    assert_eq!(read.next_seq, 2);
+    let read = if read.exited {
+        read
+    } else {
+        match client
+            .read(ReadParams {
+                process_id,
+                after_seq: Some(read.next_seq - 1),
+                max_bytes: None,
+                wait_ms: Some(1000),
+            })
+            .await
+        {
+            Ok(read) => read,
+            Err(err) => panic!("failed to wait for in-process child exit: {err}"),
+        }
+    };
+    assert!(read.exited);
+    assert_eq!(read.exit_code, Some(0));
+}
+
+#[tokio::test]
+async fn connect_in_process_rejects_invalid_exec_params_from_handler() {
+    let client = match ExecServerClient::connect_in_process(test_options()).await {
+        Ok(client) => client,
+        Err(err) => panic!("failed to connect in-process client: {err}"),
+    };
+
+    let result = client
+        .start_process(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: Vec::new(),
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: false,
+            arg0: None,
+            sandbox: None,
+        })
+        .await;
+
+    match result {
+        Err(ExecServerError::Server { code, message }) => {
+            assert_eq!(code, -32602);
+            assert_eq!(message, "argv must not be empty");
+        }
+        Err(err) => panic!("unexpected in-process exec failure: {err}"),
+        Ok(_) => panic!("expected invalid params error"),
+    }
+}
+
+#[tokio::test]
+async fn connect_in_process_rejects_writes_to_unknown_processes() {
+    let client = match ExecServerClient::connect_in_process(test_options()).await {
+        Ok(client) => client,
+        Err(err) => panic!("failed to connect in-process client: {err}"),
+    };
+
+    let result = client.write("missing", b"input".to_vec()).await;
+
+    match result {
+        Err(ExecServerError::Server { code, message }) => {
+            assert_eq!(code, -32600);
+            assert_eq!(message, "unknown process id missing");
+        }
+        Err(err) => panic!("unexpected in-process write failure: {err}"),
+        Ok(_) => panic!("expected unknown process error"),
+    }
+}
+
+#[tokio::test]
+async fn connect_in_process_terminate_marks_process_exited() {
+    let client = match ExecServerClient::connect_in_process(test_options()).await {
+        Ok(client) => client,
+        Err(err) => panic!("failed to connect in-process client: {err}"),
+    };
+
+    let process = match client
+        .start_process(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec!["sleep".to_string(), "30".to_string()],
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: false,
+            arg0: None,
+            sandbox: None,
+        })
+        .await
+    {
+        Ok(process) => process,
+        Err(err) => panic!("failed to start in-process child: {err}"),
+    };
+
+    if let Err(err) = client.terminate(&process.process_id).await {
+        panic!("failed to terminate in-process child: {err}");
+    }
+
+    timeout(Duration::from_secs(2), async {
+        loop {
+            if process.has_exited() {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|err| panic!("timed out waiting for in-process child to exit: {err}"));
+
+    assert!(process.has_exited());
+}
+
+#[tokio::test]
+async fn dropping_in_process_client_terminates_running_processes() {
+    let marker_path = std::env::temp_dir().join(format!(
+        "codex-exec-server-inprocess-drop-{}-{}",
+        std::process::id(),
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .expect("system time")
+            .as_nanos()
+    ));
+    let _ = std::fs::remove_file(&marker_path);
+
+    {
+        let client = match ExecServerClient::connect_in_process(test_options()).await {
+            Ok(client) => client,
+            Err(err) => panic!("failed to connect in-process client: {err}"),
+        };
+
+        let _ = client
+            .exec(ExecParams {
+                process_id: "proc-1".to_string(),
+                argv: vec![
+                    "/bin/sh".to_string(),
+                    "-c".to_string(),
+                    format!("sleep 2; printf dropped > {}", marker_path.display()),
+                ],
+                cwd: std::env::current_dir().expect("cwd"),
+                env: HashMap::new(),
+                tty: false,
+                arg0: None,
+                sandbox: None,
+            })
+            .await
+            .unwrap_or_else(|err| panic!("failed to start in-process child: {err}"));
+    }
+
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    assert!(
+        !marker_path.exists(),
+        "dropping the in-process client should terminate managed children"
+    );
+    let _ = std::fs::remove_file(&marker_path);
+}
+
+#[tokio::test]
+async fn connect_stdio_returns_initialize_errors() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (mut server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+
+        let initialize = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(request) = initialize else {
+            panic!("expected initialize request");
+        };
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Error(JSONRPCError {
+                id: request.id,
+                error: JSONRPCErrorError {
+                    code: -32600,
+                    message: "rejected".to_string(),
+                    data: None,
+                },
+            }),
+        )
+        .await;
+    });
+
+    let result = ExecServerClient::connect_stdio(client_stdin, client_stdout, test_options()).await;
+
+    match result {
+        Err(ExecServerError::Server { code, message }) => {
+            assert_eq!(code, -32600);
+            assert_eq!(message, "rejected");
+        }
+        Err(err) => panic!("unexpected initialize failure: {err}"),
+        Ok(_) => panic!("expected initialize failure"),
+    }
+}
+
+#[tokio::test]
+async fn start_process_cleans_up_registered_process_after_request_error() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (mut server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+
+        let initialize = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(initialize_request) = initialize else {
+            panic!("expected initialize request");
+        };
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: initialize_request.id,
+                result: serde_json::json!({ "protocolVersion": PROTOCOL_VERSION }),
+            }),
+        )
+        .await;
+
+        let initialized = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Notification(notification) = initialized else {
+            panic!("expected initialized notification");
+        };
+        assert_eq!(notification.method, INITIALIZED_METHOD);
+
+        let exec_request = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(JSONRPCRequest { id, method, .. }) = exec_request else {
+            panic!("expected exec request");
+        };
+        assert_eq!(method, EXEC_METHOD);
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Error(JSONRPCError {
+                id,
+                error: JSONRPCErrorError {
+                    code: -32600,
+                    message: "duplicate process".to_string(),
+                    data: None,
+                },
+            }),
+        )
+        .await;
+    });
+
+    let client =
+        match ExecServerClient::connect_stdio(client_stdin, client_stdout, test_options()).await {
+            Ok(client) => client,
+            Err(err) => panic!("failed to connect test client: {err}"),
+        };
+
+    let result = client
+        .start_process(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: true,
+            arg0: None,
+            sandbox: None,
+        })
+        .await;
+
+    match result {
+        Err(ExecServerError::Server { code, message }) => {
+            assert_eq!(code, -32600);
+            assert_eq!(message, "duplicate process");
+        }
+        Err(err) => panic!("unexpected start_process failure: {err}"),
+        Ok(_) => panic!("expected start_process failure"),
+    }
+
+    assert!(
+        client.pending_request_count().await == 0,
+        "failed requests should not leave pending request state behind"
+    );
+}
+
+#[tokio::test]
+async fn connect_stdio_times_out_during_initialize_handshake() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (_server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+        let _ = read_jsonrpc_line(&mut lines).await;
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    });
+
+    let result = ExecServerClient::connect_stdio(
+        client_stdin,
+        client_stdout,
+        ExecServerClientConnectOptions {
+            client_name: "test-client".to_string(),
+            initialize_timeout: Duration::from_millis(25),
+        },
+    )
+    .await;
+
+    match result {
+        Err(ExecServerError::InitializeTimedOut { timeout }) => {
+            assert_eq!(timeout, Duration::from_millis(25));
+        }
+        Err(err) => panic!("unexpected initialize timeout failure: {err}"),
+        Ok(_) => panic!("expected initialize timeout"),
+    }
+}
+
+#[tokio::test]
+async fn start_process_preserves_output_stream_metadata() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (mut server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+
+        let initialize = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(initialize_request) = initialize else {
+            panic!("expected initialize request");
+        };
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: initialize_request.id,
+                result: serde_json::json!({ "protocolVersion": PROTOCOL_VERSION }),
+            }),
+        )
+        .await;
+
+        let initialized = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Notification(notification) = initialized else {
+            panic!("expected initialized notification");
+        };
+        assert_eq!(notification.method, INITIALIZED_METHOD);
+
+        let exec_request = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(JSONRPCRequest { id, method, .. }) = exec_request else {
+            panic!("expected exec request");
+        };
+        assert_eq!(method, EXEC_METHOD);
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id,
+                result: serde_json::json!({ "processId": "proc-1" }),
+            }),
+        )
+        .await;
+        tokio::time::sleep(Duration::from_millis(25)).await;
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Notification(JSONRPCNotification {
+                method: EXEC_OUTPUT_DELTA_METHOD.to_string(),
+                params: Some(serde_json::json!({
+                    "processId": "proc-1",
+                    "stream": "stderr",
+                    "chunk": "ZXJyb3IK"
+                })),
+            }),
+        )
+        .await;
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    });
+
+    let client =
+        match ExecServerClient::connect_stdio(client_stdin, client_stdout, test_options()).await {
+            Ok(client) => client,
+            Err(err) => panic!("failed to connect test client: {err}"),
+        };
+
+    let process = match client
+        .start_process(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: true,
+            arg0: None,
+            sandbox: None,
+        })
+        .await
+    {
+        Ok(process) => process,
+        Err(err) => panic!("failed to start process: {err}"),
+    };
+
+    let mut output = process.output_receiver();
+    let output = timeout(Duration::from_secs(1), output.recv())
+        .await
+        .unwrap_or_else(|err| panic!("timed out waiting for process output: {err}"))
+        .unwrap_or_else(|err| panic!("failed to receive process output: {err}"));
+    assert_eq!(output.stream, ExecOutputStream::Stderr);
+    assert_eq!(output.chunk, b"error\n".to_vec());
+}
+
+#[tokio::test]
+async fn terminate_does_not_mark_process_exited_before_exit_notification() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (mut server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+
+        let initialize = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(initialize_request) = initialize else {
+            panic!("expected initialize request");
+        };
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: initialize_request.id,
+                result: serde_json::json!({ "protocolVersion": PROTOCOL_VERSION }),
+            }),
+        )
+        .await;
+
+        let initialized = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Notification(notification) = initialized else {
+            panic!("expected initialized notification");
+        };
+        assert_eq!(notification.method, INITIALIZED_METHOD);
+
+        let exec_request = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(JSONRPCRequest { id, method, .. }) = exec_request else {
+            panic!("expected exec request");
+        };
+        assert_eq!(method, EXEC_METHOD);
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id,
+                result: serde_json::json!({ "processId": "proc-1" }),
+            }),
+        )
+        .await;
+
+        let terminate_request = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(JSONRPCRequest { id, method, .. }) = terminate_request else {
+            panic!("expected terminate request");
+        };
+        assert_eq!(method, EXEC_TERMINATE_METHOD);
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id,
+                result: serde_json::json!({ "running": true }),
+            }),
+        )
+        .await;
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    });
+
+    let client =
+        match ExecServerClient::connect_stdio(client_stdin, client_stdout, test_options()).await {
+            Ok(client) => client,
+            Err(err) => panic!("failed to connect test client: {err}"),
+        };
+
+    let process = match client
+        .start_process(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: true,
+            arg0: None,
+            sandbox: None,
+        })
+        .await
+    {
+        Ok(process) => process,
+        Err(err) => panic!("failed to start process: {err}"),
+    };
+
+    process.terminate();
+    tokio::time::sleep(Duration::from_millis(25)).await;
+    assert!(!process.has_exited(), "terminate should not imply exit");
+    assert_eq!(process.exit_code(), None);
+}
+
+#[tokio::test]
+async fn start_process_uses_protocol_process_ids() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (mut server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+
+        let initialize = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(initialize_request) = initialize else {
+            panic!("expected initialize request");
+        };
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: initialize_request.id,
+                result: serde_json::json!({ "protocolVersion": PROTOCOL_VERSION }),
+            }),
+        )
+        .await;
+
+        let initialized = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Notification(notification) = initialized else {
+            panic!("expected initialized notification");
+        };
+        assert_eq!(notification.method, INITIALIZED_METHOD);
+
+        let exec_request = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(JSONRPCRequest { id, method, .. }) = exec_request else {
+            panic!("expected exec request");
+        };
+        assert_eq!(method, EXEC_METHOD);
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id,
+                result: serde_json::json!({ "processId": "other-proc" }),
+            }),
+        )
+        .await;
+    });
+
+    let client =
+        match ExecServerClient::connect_stdio(client_stdin, client_stdout, test_options()).await {
+            Ok(client) => client,
+            Err(err) => panic!("failed to connect test client: {err}"),
+        };
+
+    let process = match client
+        .start_process(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: true,
+            arg0: None,
+            sandbox: None,
+        })
+        .await
+    {
+        Ok(process) => process,
+        Err(err) => panic!("failed to start process: {err}"),
+    };
+
+    assert_eq!(process.process_id, "other-proc");
+}
+
+#[tokio::test]
+async fn start_process_routes_output_for_protocol_process_ids() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (mut server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+
+        let initialize = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(initialize_request) = initialize else {
+            panic!("expected initialize request");
+        };
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: initialize_request.id,
+                result: serde_json::json!({ "protocolVersion": PROTOCOL_VERSION }),
+            }),
+        )
+        .await;
+
+        let initialized = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Notification(notification) = initialized else {
+            panic!("expected initialized notification");
+        };
+        assert_eq!(notification.method, INITIALIZED_METHOD);
+
+        let exec_request = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(JSONRPCRequest { id, method, .. }) = exec_request else {
+            panic!("expected exec request");
+        };
+        assert_eq!(method, EXEC_METHOD);
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id,
+                result: serde_json::json!({ "processId": "proc-1" }),
+            }),
+        )
+        .await;
+        tokio::time::sleep(Duration::from_millis(25)).await;
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Notification(JSONRPCNotification {
+                method: EXEC_OUTPUT_DELTA_METHOD.to_string(),
+                params: Some(serde_json::json!({
+                    "processId": "proc-1",
+                    "stream": "stdout",
+                    "chunk": "YWxpdmUK"
+                })),
+            }),
+        )
+        .await;
+    });
+
+    let client =
+        match ExecServerClient::connect_stdio(client_stdin, client_stdout, test_options()).await {
+            Ok(client) => client,
+            Err(err) => panic!("failed to connect test client: {err}"),
+        };
+
+    let first_process = match client
+        .start_process(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: true,
+            arg0: None,
+            sandbox: None,
+        })
+        .await
+    {
+        Ok(process) => process,
+        Err(err) => panic!("failed to start first process: {err}"),
+    };
+
+    let mut output = first_process.output_receiver();
+    let output = timeout(Duration::from_secs(1), output.recv())
+        .await
+        .unwrap_or_else(|err| panic!("timed out waiting for process output: {err}"))
+        .unwrap_or_else(|err| panic!("failed to receive process output: {err}"));
+    assert_eq!(output.stream, ExecOutputStream::Stdout);
+    assert_eq!(output.chunk, b"alive\n".to_vec());
+}
+
+#[tokio::test]
+async fn transport_shutdown_marks_processes_exited_without_exit_codes() {
+    let (client_stdin, server_reader) = tokio::io::duplex(4096);
+    let (mut server_writer, client_stdout) = tokio::io::duplex(4096);
+
+    tokio::spawn(async move {
+        let mut lines = BufReader::new(server_reader).lines();
+
+        let initialize = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(initialize_request) = initialize else {
+            panic!("expected initialize request");
+        };
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id: initialize_request.id,
+                result: serde_json::json!({ "protocolVersion": PROTOCOL_VERSION }),
+            }),
+        )
+        .await;
+
+        let initialized = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Notification(notification) = initialized else {
+            panic!("expected initialized notification");
+        };
+        assert_eq!(notification.method, INITIALIZED_METHOD);
+
+        let exec_request = read_jsonrpc_line(&mut lines).await;
+        let JSONRPCMessage::Request(JSONRPCRequest { id, method, .. }) = exec_request else {
+            panic!("expected exec request");
+        };
+        assert_eq!(method, EXEC_METHOD);
+        write_jsonrpc_line(
+            &mut server_writer,
+            JSONRPCMessage::Response(JSONRPCResponse {
+                id,
+                result: serde_json::json!({ "processId": "proc-1" }),
+            }),
+        )
+        .await;
+        drop(server_writer);
+    });
+
+    let client =
+        match ExecServerClient::connect_stdio(client_stdin, client_stdout, test_options()).await {
+            Ok(client) => client,
+            Err(err) => panic!("failed to connect test client: {err}"),
+        };
+
+    let process = match client
+        .start_process(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+            cwd: std::env::current_dir().unwrap_or_else(|err| panic!("missing cwd: {err}")),
+            env: HashMap::new(),
+            tty: true,
+            arg0: None,
+            sandbox: None,
+        })
+        .await
+    {
+        Ok(process) => process,
+        Err(err) => panic!("failed to start process: {err}"),
+    };
+
+    let _ = process;
+}

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

@@ -22,6 +22,7 @@ pub(crate) enum JsonRpcConnectionEvent {
 pub(crate) struct JsonRpcConnection {
     outgoing_tx: mpsc::Sender<JSONRPCMessage>,
     incoming_rx: mpsc::Receiver<JsonRpcConnectionEvent>,
+    task_handles: Vec<tokio::task::JoinHandle<()>>,
 }
 
 impl JsonRpcConnection {
@@ -35,7 +36,7 @@ impl JsonRpcConnection {
 
         let reader_label = connection_label.clone();
         let incoming_tx_for_reader = incoming_tx.clone();
-        tokio::spawn(async move {
+        let reader_task = tokio::spawn(async move {
             let mut lines = BufReader::new(reader).lines();
             loop {
                 match lines.next_line().await {
@@ -66,7 +67,7 @@ impl JsonRpcConnection {
                         }
                     }
                     Ok(None) => {
-                        send_disconnected(&incoming_tx_for_reader, None).await;
+                        send_disconnected(&incoming_tx_for_reader, /*reason*/ None).await;
                         break;
                     }
                     Err(err) => {
@@ -83,7 +84,7 @@ impl JsonRpcConnection {
             }
         });
 
-        tokio::spawn(async move {
+        let writer_task = tokio::spawn(async move {
             let mut writer = BufWriter::new(writer);
             while let Some(message) = outgoing_rx.recv().await {
                 if let Err(err) = write_jsonrpc_line_message(&mut writer, &message).await {
@@ -102,6 +103,7 @@ impl JsonRpcConnection {
         Self {
             outgoing_tx,
             incoming_rx,
+            task_handles: vec![reader_task, writer_task],
         }
     }
 
@@ -115,7 +117,7 @@ impl JsonRpcConnection {
 
         let reader_label = connection_label.clone();
         let incoming_tx_for_reader = incoming_tx.clone();
-        tokio::spawn(async move {
+        let reader_task = tokio::spawn(async move {
             loop {
                 match websocket_reader.next().await {
                     Some(Ok(Message::Text(text))) => {
@@ -165,7 +167,7 @@ impl JsonRpcConnection {
                         }
                     }
                     Some(Ok(Message::Close(_))) => {
-                        send_disconnected(&incoming_tx_for_reader, None).await;
+                        send_disconnected(&incoming_tx_for_reader, /*reason*/ None).await;
                         break;
                     }
                     Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {}
@@ -181,14 +183,14 @@ impl JsonRpcConnection {
                         break;
                     }
                     None => {
-                        send_disconnected(&incoming_tx_for_reader, None).await;
+                        send_disconnected(&incoming_tx_for_reader, /*reason*/ None).await;
                         break;
                     }
                 }
             }
         });
 
-        tokio::spawn(async move {
+        let writer_task = tokio::spawn(async move {
             while let Some(message) = outgoing_rx.recv().await {
                 match serialize_jsonrpc_message(&message) {
                     Ok(encoded) => {
@@ -221,6 +223,7 @@ impl JsonRpcConnection {
         Self {
             outgoing_tx,
             incoming_rx,
+            task_handles: vec![reader_task, writer_task],
         }
     }
 
@@ -229,8 +232,9 @@ impl JsonRpcConnection {
     ) -> (
         mpsc::Sender<JSONRPCMessage>,
         mpsc::Receiver<JsonRpcConnectionEvent>,
+        Vec<tokio::task::JoinHandle<()>>,
     ) {
-        (self.outgoing_tx, self.incoming_rx)
+        (self.outgoing_tx, self.incoming_rx, self.task_handles)
     }
 }
 
@@ -323,7 +327,7 @@ mod tests {
         let (connection_writer, reader_from_connection) = tokio::io::duplex(1024);
         let connection =
             JsonRpcConnection::from_stdio(connection_reader, connection_writer, "test".to_string());
-        let (outgoing_tx, mut incoming_rx) = connection.into_parts();
+        let (outgoing_tx, mut incoming_rx, _task_handles) = connection.into_parts();
 
         let incoming_message = JSONRPCMessage::Request(JSONRPCRequest {
             id: RequestId::Integer(7),
@@ -371,7 +375,7 @@ mod tests {
         let (connection_writer, _reader_from_connection) = tokio::io::duplex(1024);
         let connection =
             JsonRpcConnection::from_stdio(connection_reader, connection_writer, "test".to_string());
-        let (_outgoing_tx, mut incoming_rx) = connection.into_parts();
+        let (_outgoing_tx, mut incoming_rx, _task_handles) = connection.into_parts();
 
         if let Err(err) = writer_to_connection.write_all(b"not-json\n").await {
             panic!("failed to write invalid JSON: {err}");
@@ -401,7 +405,7 @@ mod tests {
         let (connection_writer, _reader_from_connection) = tokio::io::duplex(1024);
         let connection =
             JsonRpcConnection::from_stdio(connection_reader, connection_writer, "test".to_string());
-        let (_outgoing_tx, mut incoming_rx) = connection.into_parts();
+        let (_outgoing_tx, mut incoming_rx, _task_handles) = connection.into_parts();
         drop(writer_to_connection);
 
         let event = recv_event(&mut incoming_rx).await;

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

@@ -3,6 +3,7 @@ mod client_api;
 mod connection;
 mod local;
 mod protocol;
+mod rpc;
 mod server;
 
 pub use client::ExecServerClient;

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

@@ -13,6 +13,13 @@ pub const EXEC_WRITE_METHOD: &str = "process/write";
 pub const EXEC_TERMINATE_METHOD: &str = "process/terminate";
 pub const EXEC_OUTPUT_DELTA_METHOD: &str = "process/output";
 pub const EXEC_EXITED_METHOD: &str = "process/exited";
+pub const FS_READ_FILE_METHOD: &str = "fs/readFile";
+pub const FS_WRITE_FILE_METHOD: &str = "fs/writeFile";
+pub const FS_CREATE_DIRECTORY_METHOD: &str = "fs/createDirectory";
+pub const FS_GET_METADATA_METHOD: &str = "fs/getMetadata";
+pub const FS_READ_DIRECTORY_METHOD: &str = "fs/readDirectory";
+pub const FS_REMOVE_METHOD: &str = "fs/remove";
+pub const FS_COPY_METHOD: &str = "fs/copy";
 pub const PROTOCOL_VERSION: &str = "exec-server.v0";
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -54,6 +61,21 @@ pub struct ExecParams {
     pub env: HashMap<String, String>,
     pub tty: bool,
     pub arg0: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sandbox: Option<ExecSandboxConfig>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecSandboxConfig {
+    pub mode: ExecSandboxMode,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum ExecSandboxMode {
+    None,
+    HostDefault,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

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

@@ -0,0 +1,216 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+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::sync::Mutex;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+use tracing::warn;
+
+use crate::connection::JsonRpcConnection;
+use crate::connection::JsonRpcConnectionEvent;
+
+type PendingRequest = oneshot::Sender<Result<Value, JSONRPCErrorError>>;
+
+#[derive(Debug)]
+pub(crate) enum RpcClientEvent {
+    Notification(JSONRPCNotification),
+    Disconnected { reason: Option<String> },
+}
+
+pub(crate) struct RpcClient {
+    write_tx: mpsc::Sender<JSONRPCMessage>,
+    pending: Arc<Mutex<HashMap<RequestId, PendingRequest>>>,
+    next_request_id: AtomicI64,
+    transport_tasks: Vec<JoinHandle<()>>,
+    reader_task: JoinHandle<()>,
+}
+
+impl RpcClient {
+    pub(crate) fn new(connection: JsonRpcConnection) -> (Self, mpsc::Receiver<RpcClientEvent>) {
+        let (write_tx, mut incoming_rx, transport_tasks) = connection.into_parts();
+        let pending = Arc::new(Mutex::new(HashMap::<RequestId, PendingRequest>::new()));
+        let (event_tx, event_rx) = mpsc::channel(128);
+
+        let pending_for_reader = Arc::clone(&pending);
+        let reader_task = tokio::spawn(async move {
+            while let Some(event) = incoming_rx.recv().await {
+                match event {
+                    JsonRpcConnectionEvent::Message(message) => {
+                        if let Err(err) =
+                            handle_server_message(&pending_for_reader, &event_tx, message).await
+                        {
+                            warn!("JSON-RPC client closing after protocol error: {err}");
+                            break;
+                        }
+                    }
+                    JsonRpcConnectionEvent::Disconnected { reason } => {
+                        let _ = event_tx.send(RpcClientEvent::Disconnected { reason }).await;
+                        drain_pending(&pending_for_reader).await;
+                        return;
+                    }
+                }
+            }
+
+            let _ = event_tx
+                .send(RpcClientEvent::Disconnected { reason: None })
+                .await;
+            drain_pending(&pending_for_reader).await;
+        });
+
+        (
+            Self {
+                write_tx,
+                pending,
+                next_request_id: AtomicI64::new(1),
+                transport_tasks,
+                reader_task,
+            },
+            event_rx,
+        )
+    }
+
+    pub(crate) async fn notify<P: Serialize>(
+        &self,
+        method: &str,
+        params: &P,
+    ) -> Result<(), serde_json::Error> {
+        let params = serde_json::to_value(params)?;
+        self.write_tx
+            .send(JSONRPCMessage::Notification(JSONRPCNotification {
+                method: method.to_string(),
+                params: Some(params),
+            }))
+            .await
+            .map_err(|_| {
+                serde_json::Error::io(std::io::Error::new(
+                    std::io::ErrorKind::BrokenPipe,
+                    "JSON-RPC transport closed",
+                ))
+            })
+    }
+
+    pub(crate) async fn call<P, T>(&self, method: &str, params: &P) -> Result<T, RpcCallError>
+    where
+        P: Serialize,
+        T: DeserializeOwned,
+    {
+        let request_id = RequestId::Integer(self.next_request_id.fetch_add(1, Ordering::SeqCst));
+        let (response_tx, response_rx) = oneshot::channel();
+        self.pending
+            .lock()
+            .await
+            .insert(request_id.clone(), response_tx);
+
+        let params = match serde_json::to_value(params) {
+            Ok(params) => params,
+            Err(err) => {
+                self.pending.lock().await.remove(&request_id);
+                return Err(RpcCallError::Json(err));
+            }
+        };
+        if self
+            .write_tx
+            .send(JSONRPCMessage::Request(JSONRPCRequest {
+                id: request_id.clone(),
+                method: method.to_string(),
+                params: Some(params),
+                trace: None,
+            }))
+            .await
+            .is_err()
+        {
+            self.pending.lock().await.remove(&request_id);
+            return Err(RpcCallError::Closed);
+        }
+
+        let result = response_rx.await.map_err(|_| RpcCallError::Closed)?;
+        let response = match result {
+            Ok(response) => response,
+            Err(error) => return Err(RpcCallError::Server(error)),
+        };
+        serde_json::from_value(response).map_err(RpcCallError::Json)
+    }
+
+    #[cfg(test)]
+    pub(crate) async fn pending_request_count(&self) -> usize {
+        self.pending.lock().await.len()
+    }
+}
+
+impl Drop for RpcClient {
+    fn drop(&mut self) {
+        for task in &self.transport_tasks {
+            task.abort();
+        }
+        self.reader_task.abort();
+    }
+}
+
+#[derive(Debug)]
+pub(crate) enum RpcCallError {
+    Closed,
+    Json(serde_json::Error),
+    Server(JSONRPCErrorError),
+}
+
+async fn handle_server_message(
+    pending: &Mutex<HashMap<RequestId, PendingRequest>>,
+    event_tx: &mpsc::Sender<RpcClientEvent>,
+    message: JSONRPCMessage,
+) -> Result<(), String> {
+    match message {
+        JSONRPCMessage::Response(JSONRPCResponse { id, result }) => {
+            if let Some(pending) = pending.lock().await.remove(&id) {
+                let _ = pending.send(Ok(result));
+            }
+        }
+        JSONRPCMessage::Error(JSONRPCError { id, error }) => {
+            if let Some(pending) = pending.lock().await.remove(&id) {
+                let _ = pending.send(Err(error));
+            }
+        }
+        JSONRPCMessage::Notification(notification) => {
+            let _ = event_tx
+                .send(RpcClientEvent::Notification(notification))
+                .await;
+        }
+        JSONRPCMessage::Request(request) => {
+            return Err(format!(
+                "unexpected JSON-RPC request from remote server: {}",
+                request.method
+            ));
+        }
+    }
+
+    Ok(())
+}
+
+async fn drain_pending(pending: &Mutex<HashMap<RequestId, PendingRequest>>) {
+    let pending = {
+        let mut pending = pending.lock().await;
+        pending
+            .drain()
+            .map(|(_, pending)| pending)
+            .collect::<Vec<_>>()
+    };
+    for pending in pending {
+        let _ = pending.send(Err(JSONRPCErrorError {
+            code: -32000,
+            data: None,
+            message: "JSON-RPC transport closed".to_string(),
+        }));
+    }
+}

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

@@ -1,3 +1,4 @@
+mod filesystem;
 mod handler;
 mod processor;
 mod routing;

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

@@ -0,0 +1,170 @@
+use std::io;
+use std::sync::Arc;
+
+use base64::Engine as _;
+use base64::engine::general_purpose::STANDARD;
+use codex_app_server_protocol::FsCopyParams;
+use codex_app_server_protocol::FsCopyResponse;
+use codex_app_server_protocol::FsCreateDirectoryParams;
+use codex_app_server_protocol::FsCreateDirectoryResponse;
+use codex_app_server_protocol::FsGetMetadataParams;
+use codex_app_server_protocol::FsGetMetadataResponse;
+use codex_app_server_protocol::FsReadDirectoryEntry;
+use codex_app_server_protocol::FsReadDirectoryParams;
+use codex_app_server_protocol::FsReadDirectoryResponse;
+use codex_app_server_protocol::FsReadFileParams;
+use codex_app_server_protocol::FsReadFileResponse;
+use codex_app_server_protocol::FsRemoveParams;
+use codex_app_server_protocol::FsRemoveResponse;
+use codex_app_server_protocol::FsWriteFileParams;
+use codex_app_server_protocol::FsWriteFileResponse;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_environment::CopyOptions;
+use codex_environment::CreateDirectoryOptions;
+use codex_environment::Environment;
+use codex_environment::ExecutorFileSystem;
+use codex_environment::RemoveOptions;
+
+use crate::server::routing::internal_error;
+use crate::server::routing::invalid_request;
+
+#[derive(Clone)]
+pub(crate) struct ExecServerFileSystem {
+    file_system: Arc<dyn ExecutorFileSystem>,
+}
+
+impl Default for ExecServerFileSystem {
+    fn default() -> Self {
+        Self {
+            file_system: Arc::new(Environment.get_filesystem()),
+        }
+    }
+}
+
+impl ExecServerFileSystem {
+    pub(crate) async fn read_file(
+        &self,
+        params: FsReadFileParams,
+    ) -> Result<FsReadFileResponse, JSONRPCErrorError> {
+        let bytes = self
+            .file_system
+            .read_file(&params.path)
+            .await
+            .map_err(map_fs_error)?;
+        Ok(FsReadFileResponse {
+            data_base64: STANDARD.encode(bytes),
+        })
+    }
+
+    pub(crate) async fn write_file(
+        &self,
+        params: FsWriteFileParams,
+    ) -> Result<FsWriteFileResponse, JSONRPCErrorError> {
+        let bytes = STANDARD.decode(params.data_base64).map_err(|err| {
+            invalid_request(format!(
+                "fs/writeFile requires valid base64 dataBase64: {err}"
+            ))
+        })?;
+        self.file_system
+            .write_file(&params.path, bytes)
+            .await
+            .map_err(map_fs_error)?;
+        Ok(FsWriteFileResponse {})
+    }
+
+    pub(crate) async fn create_directory(
+        &self,
+        params: FsCreateDirectoryParams,
+    ) -> Result<FsCreateDirectoryResponse, JSONRPCErrorError> {
+        self.file_system
+            .create_directory(
+                &params.path,
+                CreateDirectoryOptions {
+                    recursive: params.recursive.unwrap_or(true),
+                },
+            )
+            .await
+            .map_err(map_fs_error)?;
+        Ok(FsCreateDirectoryResponse {})
+    }
+
+    pub(crate) async fn get_metadata(
+        &self,
+        params: FsGetMetadataParams,
+    ) -> Result<FsGetMetadataResponse, JSONRPCErrorError> {
+        let metadata = self
+            .file_system
+            .get_metadata(&params.path)
+            .await
+            .map_err(map_fs_error)?;
+        Ok(FsGetMetadataResponse {
+            is_directory: metadata.is_directory,
+            is_file: metadata.is_file,
+            created_at_ms: metadata.created_at_ms,
+            modified_at_ms: metadata.modified_at_ms,
+        })
+    }
+
+    pub(crate) async fn read_directory(
+        &self,
+        params: FsReadDirectoryParams,
+    ) -> Result<FsReadDirectoryResponse, JSONRPCErrorError> {
+        let entries = self
+            .file_system
+            .read_directory(&params.path)
+            .await
+            .map_err(map_fs_error)?;
+        Ok(FsReadDirectoryResponse {
+            entries: entries
+                .into_iter()
+                .map(|entry| FsReadDirectoryEntry {
+                    file_name: entry.file_name,
+                    is_directory: entry.is_directory,
+                    is_file: entry.is_file,
+                })
+                .collect(),
+        })
+    }
+
+    pub(crate) async fn remove(
+        &self,
+        params: FsRemoveParams,
+    ) -> Result<FsRemoveResponse, JSONRPCErrorError> {
+        self.file_system
+            .remove(
+                &params.path,
+                RemoveOptions {
+                    recursive: params.recursive.unwrap_or(true),
+                    force: params.force.unwrap_or(true),
+                },
+            )
+            .await
+            .map_err(map_fs_error)?;
+        Ok(FsRemoveResponse {})
+    }
+
+    pub(crate) async fn copy(
+        &self,
+        params: FsCopyParams,
+    ) -> Result<FsCopyResponse, JSONRPCErrorError> {
+        self.file_system
+            .copy(
+                &params.source_path,
+                &params.destination_path,
+                CopyOptions {
+                    recursive: params.recursive,
+                },
+            )
+            .await
+            .map_err(map_fs_error)?;
+        Ok(FsCopyResponse {})
+    }
+}
+
+fn map_fs_error(err: io::Error) -> JSONRPCErrorError {
+    if err.kind() == io::ErrorKind::InvalidInput {
+        invalid_request(err.to_string())
+    } else {
+        internal_error(err.to_string())
+    }
+}

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

@@ -0,0 +1,789 @@
+use std::collections::HashMap;
+use std::collections::VecDeque;
+use std::sync::Arc;
+use std::time::Duration;
+
+use pretty_assertions::assert_eq;
+use tokio::sync::Notify;
+use tokio::time::timeout;
+
+use super::ExecServerHandler;
+use super::ProcessEntry;
+use super::RetainedOutputChunk;
+use super::RunningProcess;
+use crate::protocol::ExecOutputStream;
+use crate::protocol::ExecSandboxConfig;
+use crate::protocol::ExecSandboxMode;
+use crate::protocol::InitializeParams;
+use crate::protocol::InitializeResponse;
+use crate::protocol::PROTOCOL_VERSION;
+use crate::protocol::ReadParams;
+use crate::protocol::TerminateResponse;
+use crate::protocol::WriteParams;
+use crate::server::routing::ExecServerClientNotification;
+use crate::server::routing::ExecServerInboundMessage;
+use crate::server::routing::ExecServerOutboundMessage;
+use crate::server::routing::ExecServerRequest;
+use crate::server::routing::ExecServerResponseMessage;
+use codex_app_server_protocol::RequestId;
+
+async fn recv_outbound(
+    outgoing_rx: &mut tokio::sync::mpsc::Receiver<ExecServerOutboundMessage>,
+) -> ExecServerOutboundMessage {
+    let recv_result = timeout(Duration::from_secs(1), outgoing_rx.recv()).await;
+    let maybe_message = match recv_result {
+        Ok(maybe_message) => maybe_message,
+        Err(err) => panic!("timed out waiting for handler output: {err}"),
+    };
+    match maybe_message {
+        Some(message) => message,
+        None => panic!("handler output channel closed unexpectedly"),
+    }
+}
+
+#[tokio::test]
+async fn initialize_response_reports_protocol_version() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(1);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(1),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("initialize should succeed: {err}");
+    }
+
+    assert_eq!(
+        recv_outbound(&mut outgoing_rx).await,
+        ExecServerOutboundMessage::Response {
+            request_id: RequestId::Integer(1),
+            response: ExecServerResponseMessage::Initialize(InitializeResponse {
+                protocol_version: PROTOCOL_VERSION.to_string(),
+            }),
+        }
+    );
+}
+
+#[tokio::test]
+async fn exec_methods_require_initialize() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(1);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(ExecServerRequest::Exec {
+            request_id: RequestId::Integer(7),
+            params: crate::protocol::ExecParams {
+                process_id: "proc-1".to_string(),
+                argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+                cwd: std::env::current_dir().expect("cwd"),
+                env: HashMap::new(),
+                tty: true,
+                arg0: None,
+                sandbox: None,
+            },
+        }))
+        .await
+    {
+        panic!("request handling should not fail the handler: {err}");
+    }
+
+    let ExecServerOutboundMessage::Error { request_id, error } =
+        recv_outbound(&mut outgoing_rx).await
+    else {
+        panic!("expected invalid-request error");
+    };
+    assert_eq!(request_id, RequestId::Integer(7));
+    assert_eq!(error.code, -32600);
+    assert_eq!(
+        error.message,
+        "client must call initialize before using exec methods"
+    );
+}
+
+#[tokio::test]
+async fn exec_methods_require_initialized_notification_after_initialize() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(2);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(1),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("initialize should succeed: {err}");
+    }
+    let _ = recv_outbound(&mut outgoing_rx).await;
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(ExecServerRequest::Exec {
+            request_id: RequestId::Integer(2),
+            params: crate::protocol::ExecParams {
+                process_id: "proc-1".to_string(),
+                argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+                cwd: std::env::current_dir().expect("cwd"),
+                env: HashMap::new(),
+                tty: true,
+                arg0: None,
+                sandbox: None,
+            },
+        }))
+        .await
+    {
+        panic!("request handling should not fail the handler: {err}");
+    }
+
+    let ExecServerOutboundMessage::Error { request_id, error } =
+        recv_outbound(&mut outgoing_rx).await
+    else {
+        panic!("expected invalid-request error");
+    };
+    assert_eq!(request_id, RequestId::Integer(2));
+    assert_eq!(error.code, -32600);
+    assert_eq!(
+        error.message,
+        "client must send initialized before using exec methods"
+    );
+}
+
+#[tokio::test]
+async fn initialized_before_initialize_is_a_protocol_error() {
+    let (outgoing_tx, _outgoing_rx) = tokio::sync::mpsc::channel(1);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    let result = handler
+        .handle_message(ExecServerInboundMessage::Notification(
+            ExecServerClientNotification::Initialized,
+        ))
+        .await;
+
+    match result {
+        Err(err) => {
+            assert_eq!(
+                err,
+                "received `initialized` notification before `initialize`"
+            );
+        }
+        Ok(()) => panic!("expected protocol error for early initialized notification"),
+    }
+}
+
+#[tokio::test]
+async fn initialize_may_only_be_sent_once_per_connection() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(2);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(1),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("initialize should succeed: {err}");
+    }
+    let _ = recv_outbound(&mut outgoing_rx).await;
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(2),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("duplicate initialize should not fail the handler: {err}");
+    }
+
+    let ExecServerOutboundMessage::Error { request_id, error } =
+        recv_outbound(&mut outgoing_rx).await
+    else {
+        panic!("expected invalid-request error");
+    };
+    assert_eq!(request_id, RequestId::Integer(2));
+    assert_eq!(error.code, -32600);
+    assert_eq!(
+        error.message,
+        "initialize may only be sent once per connection"
+    );
+}
+
+#[tokio::test]
+async fn host_default_sandbox_requests_are_rejected_until_supported() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(3);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(1),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("initialize should succeed: {err}");
+    }
+    let _ = recv_outbound(&mut outgoing_rx).await;
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Notification(
+            ExecServerClientNotification::Initialized,
+        ))
+        .await
+    {
+        panic!("initialized should succeed: {err}");
+    }
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(ExecServerRequest::Exec {
+            request_id: RequestId::Integer(2),
+            params: crate::protocol::ExecParams {
+                process_id: "proc-1".to_string(),
+                argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+                cwd: std::env::current_dir().expect("cwd"),
+                env: HashMap::new(),
+                tty: false,
+                arg0: None,
+                sandbox: Some(ExecSandboxConfig {
+                    mode: ExecSandboxMode::HostDefault,
+                }),
+            },
+        }))
+        .await
+    {
+        panic!("request handling should not fail the handler: {err}");
+    }
+
+    let ExecServerOutboundMessage::Error { request_id, error } =
+        recv_outbound(&mut outgoing_rx).await
+    else {
+        panic!("expected unsupported sandbox error");
+    };
+    assert_eq!(request_id, RequestId::Integer(2));
+    assert_eq!(error.code, -32600);
+    assert_eq!(
+        error.message,
+        "sandbox mode `hostDefault` is not supported by exec-server yet"
+    );
+}
+
+#[tokio::test]
+async fn exec_echoes_client_process_ids() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(4);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(1),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("initialize should succeed: {err}");
+    }
+    let _ = recv_outbound(&mut outgoing_rx).await;
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Notification(
+            ExecServerClientNotification::Initialized,
+        ))
+        .await
+    {
+        panic!("initialized should succeed: {err}");
+    }
+
+    let params = crate::protocol::ExecParams {
+        process_id: "proc-1".to_string(),
+        argv: vec![
+            "bash".to_string(),
+            "-lc".to_string(),
+            "sleep 30".to_string(),
+        ],
+        cwd: std::env::current_dir().expect("cwd"),
+        env: HashMap::new(),
+        tty: false,
+        arg0: None,
+        sandbox: None,
+    };
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(ExecServerRequest::Exec {
+            request_id: RequestId::Integer(2),
+            params: params.clone(),
+        }))
+        .await
+    {
+        panic!("first exec should succeed: {err}");
+    }
+    let ExecServerOutboundMessage::Response {
+        request_id,
+        response: ExecServerResponseMessage::Exec(first_exec),
+    } = recv_outbound(&mut outgoing_rx).await
+    else {
+        panic!("expected first exec response");
+    };
+    assert_eq!(request_id, RequestId::Integer(2));
+    assert_eq!(first_exec.process_id, "proc-1");
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(ExecServerRequest::Exec {
+            request_id: RequestId::Integer(3),
+            params: crate::protocol::ExecParams {
+                process_id: "proc-2".to_string(),
+                argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+                ..params
+            },
+        }))
+        .await
+    {
+        panic!("second exec should succeed: {err}");
+    }
+
+    let ExecServerOutboundMessage::Response {
+        request_id,
+        response: ExecServerResponseMessage::Exec(second_exec),
+    } = recv_outbound(&mut outgoing_rx).await
+    else {
+        panic!("expected second exec response");
+    };
+    assert_eq!(request_id, RequestId::Integer(3));
+    assert_eq!(second_exec.process_id, "proc-2");
+
+    handler.shutdown().await;
+}
+
+#[tokio::test]
+async fn concurrent_duplicate_execs_reserve_process_ids_atomically() {
+    let (outgoing_tx, _outgoing_rx) = tokio::sync::mpsc::channel(4);
+    let handler = Arc::new(ExecServerHandler::new(outgoing_tx));
+    let _ = handler.initialize().expect("initialize should succeed");
+    handler.initialized().expect("initialized should succeed");
+
+    let params = crate::protocol::ExecParams {
+        process_id: "proc-1".to_string(),
+        argv: vec![
+            "bash".to_string(),
+            "-lc".to_string(),
+            "sleep 30".to_string(),
+        ],
+        cwd: std::env::current_dir().expect("cwd"),
+        env: HashMap::new(),
+        tty: false,
+        arg0: None,
+        sandbox: None,
+    };
+
+    let first = {
+        let handler = Arc::clone(&handler);
+        let params = params.clone();
+        tokio::spawn(async move { handler.exec(params).await })
+    };
+    let second = {
+        let handler = Arc::clone(&handler);
+        let params = params.clone();
+        tokio::spawn(async move { handler.exec(params).await })
+    };
+
+    let (first, second) = tokio::join!(first, second);
+    let results = [
+        first.expect("first task should complete"),
+        second.expect("second task should complete"),
+    ];
+    let successes = results.iter().filter(|result| result.is_ok()).count();
+    let duplicate_errors = results
+        .iter()
+        .filter(|result| {
+            result
+                .as_ref()
+                .err()
+                .is_some_and(|err| err.message == "process proc-1 already exists")
+        })
+        .count();
+
+    assert_eq!(successes, 1);
+    assert_eq!(duplicate_errors, 1);
+
+    handler.shutdown().await;
+}
+
+#[tokio::test]
+async fn writes_to_pipe_backed_processes_are_rejected() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(4);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(1),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("initialize should succeed: {err}");
+    }
+    let _ = recv_outbound(&mut outgoing_rx).await;
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Notification(
+            ExecServerClientNotification::Initialized,
+        ))
+        .await
+    {
+        panic!("initialized should succeed: {err}");
+    }
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(ExecServerRequest::Exec {
+            request_id: RequestId::Integer(2),
+            params: crate::protocol::ExecParams {
+                process_id: "proc-1".to_string(),
+                argv: vec![
+                    "bash".to_string(),
+                    "-lc".to_string(),
+                    "sleep 30".to_string(),
+                ],
+                cwd: std::env::current_dir().expect("cwd"),
+                env: HashMap::new(),
+                tty: false,
+                arg0: None,
+                sandbox: None,
+            },
+        }))
+        .await
+    {
+        panic!("exec should succeed: {err}");
+    }
+    let ExecServerOutboundMessage::Response {
+        response: ExecServerResponseMessage::Exec(exec_response),
+        ..
+    } = recv_outbound(&mut outgoing_rx).await
+    else {
+        panic!("expected exec response");
+    };
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Write {
+                request_id: RequestId::Integer(3),
+                params: WriteParams {
+                    process_id: exec_response.process_id,
+                    chunk: b"hello\n".to_vec().into(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("write should not fail the handler: {err}");
+    }
+
+    let ExecServerOutboundMessage::Error { request_id, error } =
+        recv_outbound(&mut outgoing_rx).await
+    else {
+        panic!("expected stdin-closed error");
+    };
+    assert_eq!(request_id, RequestId::Integer(3));
+    assert_eq!(error.code, -32600);
+    assert_eq!(error.message, "stdin is closed for process proc-1");
+
+    handler.shutdown().await;
+}
+
+#[tokio::test]
+async fn writes_to_unknown_processes_are_rejected() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(2);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(1),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("initialize should succeed: {err}");
+    }
+    let _ = recv_outbound(&mut outgoing_rx).await;
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Notification(
+            ExecServerClientNotification::Initialized,
+        ))
+        .await
+    {
+        panic!("initialized should succeed: {err}");
+    }
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Write {
+                request_id: RequestId::Integer(2),
+                params: WriteParams {
+                    process_id: "missing".to_string(),
+                    chunk: b"hello\n".to_vec().into(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("write should not fail the handler: {err}");
+    }
+
+    let ExecServerOutboundMessage::Error { request_id, error } =
+        recv_outbound(&mut outgoing_rx).await
+    else {
+        panic!("expected unknown-process error");
+    };
+    assert_eq!(request_id, RequestId::Integer(2));
+    assert_eq!(error.code, -32600);
+    assert_eq!(error.message, "unknown process id missing");
+}
+
+#[tokio::test]
+async fn terminate_unknown_processes_report_running_false() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(2);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(1),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("initialize should succeed: {err}");
+    }
+    let _ = recv_outbound(&mut outgoing_rx).await;
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Notification(
+            ExecServerClientNotification::Initialized,
+        ))
+        .await
+    {
+        panic!("initialized should succeed: {err}");
+    }
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Terminate {
+                request_id: RequestId::Integer(2),
+                params: crate::protocol::TerminateParams {
+                    process_id: "missing".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("terminate should not fail the handler: {err}");
+    }
+
+    assert_eq!(
+        recv_outbound(&mut outgoing_rx).await,
+        ExecServerOutboundMessage::Response {
+            request_id: RequestId::Integer(2),
+            response: ExecServerResponseMessage::Terminate(TerminateResponse { running: false }),
+        }
+    );
+}
+
+#[tokio::test]
+async fn terminate_keeps_process_ids_reserved() {
+    let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(2);
+    let handler = ExecServerHandler::new(outgoing_tx);
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Initialize {
+                request_id: RequestId::Integer(1),
+                params: InitializeParams {
+                    client_name: "test".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("initialize should succeed: {err}");
+    }
+    let _ = recv_outbound(&mut outgoing_rx).await;
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Notification(
+            ExecServerClientNotification::Initialized,
+        ))
+        .await
+    {
+        panic!("initialized should succeed: {err}");
+    }
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(ExecServerRequest::Exec {
+            request_id: RequestId::Integer(2),
+            params: crate::protocol::ExecParams {
+                process_id: "proc-1".to_string(),
+                argv: vec![
+                    "bash".to_string(),
+                    "-lc".to_string(),
+                    "sleep 30".to_string(),
+                ],
+                cwd: std::env::current_dir().expect("cwd"),
+                env: HashMap::new(),
+                tty: false,
+                arg0: None,
+                sandbox: None,
+            },
+        }))
+        .await
+    {
+        panic!("exec should not fail the handler: {err}");
+    }
+    let _ = recv_outbound(&mut outgoing_rx).await;
+
+    if let Err(err) = handler
+        .handle_message(ExecServerInboundMessage::Request(
+            ExecServerRequest::Terminate {
+                request_id: RequestId::Integer(3),
+                params: crate::protocol::TerminateParams {
+                    process_id: "proc-1".to_string(),
+                },
+            },
+        ))
+        .await
+    {
+        panic!("terminate should not fail the handler: {err}");
+    }
+
+    assert_eq!(
+        recv_outbound(&mut outgoing_rx).await,
+        ExecServerOutboundMessage::Response {
+            request_id: RequestId::Integer(3),
+            response: ExecServerResponseMessage::Terminate(TerminateResponse { running: true }),
+        }
+    );
+
+    assert!(
+        handler.processes.lock().await.contains_key("proc-1"),
+        "terminated ids should stay reserved until exit cleanup removes them"
+    );
+
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
+    loop {
+        if !handler.processes.lock().await.contains_key("proc-1") {
+            break;
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "terminated ids should be removed after the exit retention window"
+        );
+        tokio::time::sleep(Duration::from_millis(25)).await;
+    }
+
+    handler.shutdown().await;
+}
+
+#[tokio::test]
+async fn read_paginates_retained_output_without_skipping_omitted_chunks() {
+    let (outgoing_tx, _outgoing_rx) = tokio::sync::mpsc::channel(1);
+    let handler = ExecServerHandler::new(outgoing_tx);
+    let _ = handler.initialize().expect("initialize should succeed");
+    handler.initialized().expect("initialized should succeed");
+
+    let spawned = codex_utils_pty::spawn_pipe_process_no_stdin(
+        "bash",
+        &["-lc".to_string(), "true".to_string()],
+        std::env::current_dir().expect("cwd").as_path(),
+        &HashMap::new(),
+        &None,
+    )
+    .await
+    .expect("spawn test process");
+    {
+        let mut process_map = handler.processes.lock().await;
+        process_map.insert(
+            "proc-1".to_string(),
+            ProcessEntry::Running(Box::new(RunningProcess {
+                session: spawned.session,
+                tty: false,
+                output: VecDeque::from([
+                    RetainedOutputChunk {
+                        seq: 1,
+                        stream: ExecOutputStream::Stdout,
+                        chunk: b"abc".to_vec(),
+                    },
+                    RetainedOutputChunk {
+                        seq: 2,
+                        stream: ExecOutputStream::Stderr,
+                        chunk: b"def".to_vec(),
+                    },
+                ]),
+                retained_bytes: 6,
+                next_seq: 3,
+                exit_code: None,
+                output_notify: Arc::new(Notify::new()),
+            })),
+        );
+    }
+
+    let first = handler
+        .exec_read(ReadParams {
+            process_id: "proc-1".to_string(),
+            after_seq: Some(0),
+            max_bytes: Some(3),
+            wait_ms: Some(0),
+        })
+        .await
+        .expect("first read should succeed");
+
+    assert_eq!(first.chunks.len(), 1);
+    assert_eq!(first.chunks[0].seq, 1);
+    assert_eq!(first.chunks[0].stream, ExecOutputStream::Stdout);
+    assert_eq!(first.chunks[0].chunk.clone().into_inner(), b"abc".to_vec());
+    assert_eq!(first.next_seq, 2);
+
+    let second = handler
+        .exec_read(ReadParams {
+            process_id: "proc-1".to_string(),
+            after_seq: Some(first.next_seq - 1),
+            max_bytes: Some(3),
+            wait_ms: Some(0),
+        })
+        .await
+        .expect("second read should succeed");
+
+    assert_eq!(second.chunks.len(), 1);
+    assert_eq!(second.chunks[0].seq, 2);
+    assert_eq!(second.chunks[0].stream, ExecOutputStream::Stderr);
+    assert_eq!(second.chunks[0].chunk.clone().into_inner(), b"def".to_vec());
+    assert_eq!(second.next_seq, 3);
+
+    handler.shutdown().await;
+}

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

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use tokio::sync::mpsc;
 use tracing::debug;
 use tracing::warn;
@@ -6,20 +8,16 @@ use crate::connection::CHANNEL_CAPACITY;
 use crate::connection::JsonRpcConnection;
 use crate::connection::JsonRpcConnectionEvent;
 use crate::server::handler::ExecServerHandler;
-use crate::server::routing::ExecServerClientNotification;
-use crate::server::routing::ExecServerInboundMessage;
 use crate::server::routing::ExecServerOutboundMessage;
-use crate::server::routing::ExecServerRequest;
-use crate::server::routing::ExecServerResponseMessage;
 use crate::server::routing::RoutedExecServerMessage;
 use crate::server::routing::encode_outbound_message;
 use crate::server::routing::route_jsonrpc_message;
 
 pub(crate) async fn run_connection(connection: JsonRpcConnection) {
-    let (json_outgoing_tx, mut incoming_rx) = connection.into_parts();
+    let (json_outgoing_tx, mut incoming_rx, _connection_tasks) = connection.into_parts();
     let (outgoing_tx, mut outgoing_rx) =
         mpsc::channel::<ExecServerOutboundMessage>(CHANNEL_CAPACITY);
-    let mut handler = ExecServerHandler::new(outgoing_tx.clone());
+    let handler = Arc::new(ExecServerHandler::new(outgoing_tx.clone()));
 
     let outbound_task = tokio::spawn(async move {
         while let Some(message) = outgoing_rx.recv().await {
@@ -40,11 +38,12 @@ pub(crate) async fn run_connection(connection: JsonRpcConnection) {
         match event {
             JsonRpcConnectionEvent::Message(message) => match route_jsonrpc_message(message) {
                 Ok(RoutedExecServerMessage::Inbound(message)) => {
-                    if let Err(err) = dispatch_to_handler(&mut handler, message, &outgoing_tx).await
-                    {
-                        warn!("closing exec-server connection after protocol error: {err}");
-                        break;
-                    }
+                    let handler = Arc::clone(&handler);
+                    tokio::spawn(async move {
+                        if let Err(err) = handler.handle_message(message).await {
+                            warn!("exec-server request failed after protocol error: {err}");
+                        }
+                    });
                 }
                 Ok(RoutedExecServerMessage::ImmediateOutbound(message)) => {
                     if outgoing_tx.send(message).await.is_err() {
@@ -70,70 +69,3 @@ pub(crate) async fn run_connection(connection: JsonRpcConnection) {
     drop(outgoing_tx);
     let _ = outbound_task.await;
 }
-
-async fn dispatch_to_handler(
-    handler: &mut ExecServerHandler,
-    message: ExecServerInboundMessage,
-    outgoing_tx: &mpsc::Sender<ExecServerOutboundMessage>,
-) -> Result<(), String> {
-    match message {
-        ExecServerInboundMessage::Request(request) => {
-            let outbound = match request {
-                ExecServerRequest::Initialize { request_id, .. } => request_outbound(
-                    request_id,
-                    handler
-                        .initialize()
-                        .map(ExecServerResponseMessage::Initialize),
-                ),
-                ExecServerRequest::Exec { request_id, params } => request_outbound(
-                    request_id,
-                    handler
-                        .exec(params)
-                        .await
-                        .map(ExecServerResponseMessage::Exec),
-                ),
-                ExecServerRequest::Read { request_id, params } => request_outbound(
-                    request_id,
-                    handler
-                        .read(params)
-                        .await
-                        .map(ExecServerResponseMessage::Read),
-                ),
-                ExecServerRequest::Write { request_id, params } => request_outbound(
-                    request_id,
-                    handler
-                        .write(params)
-                        .await
-                        .map(ExecServerResponseMessage::Write),
-                ),
-                ExecServerRequest::Terminate { request_id, params } => request_outbound(
-                    request_id,
-                    handler
-                        .terminate(params)
-                        .await
-                        .map(ExecServerResponseMessage::Terminate),
-                ),
-            };
-            outgoing_tx
-                .send(outbound)
-                .await
-                .map_err(|_| "outbound channel closed".to_string())
-        }
-        ExecServerInboundMessage::Notification(ExecServerClientNotification::Initialized) => {
-            handler.initialized()
-        }
-    }
-}
-
-fn request_outbound(
-    request_id: codex_app_server_protocol::RequestId,
-    result: Result<ExecServerResponseMessage, codex_app_server_protocol::JSONRPCErrorError>,
-) -> ExecServerOutboundMessage {
-    match result {
-        Ok(response) => ExecServerOutboundMessage::Response {
-            request_id,
-            response,
-        },
-        Err(error) => ExecServerOutboundMessage::Error { request_id, error },
-    }
-}

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

@@ -17,6 +17,13 @@ use crate::protocol::ExecExitedNotification;
 use crate::protocol::ExecOutputDeltaNotification;
 use crate::protocol::ExecParams;
 use crate::protocol::ExecResponse;
+use crate::protocol::FS_COPY_METHOD;
+use crate::protocol::FS_CREATE_DIRECTORY_METHOD;
+use crate::protocol::FS_GET_METADATA_METHOD;
+use crate::protocol::FS_READ_DIRECTORY_METHOD;
+use crate::protocol::FS_READ_FILE_METHOD;
+use crate::protocol::FS_REMOVE_METHOD;
+use crate::protocol::FS_WRITE_FILE_METHOD;
 use crate::protocol::INITIALIZE_METHOD;
 use crate::protocol::INITIALIZED_METHOD;
 use crate::protocol::InitializeParams;
@@ -27,6 +34,20 @@ use crate::protocol::TerminateParams;
 use crate::protocol::TerminateResponse;
 use crate::protocol::WriteParams;
 use crate::protocol::WriteResponse;
+use codex_app_server_protocol::FsCopyParams;
+use codex_app_server_protocol::FsCopyResponse;
+use codex_app_server_protocol::FsCreateDirectoryParams;
+use codex_app_server_protocol::FsCreateDirectoryResponse;
+use codex_app_server_protocol::FsGetMetadataParams;
+use codex_app_server_protocol::FsGetMetadataResponse;
+use codex_app_server_protocol::FsReadDirectoryParams;
+use codex_app_server_protocol::FsReadDirectoryResponse;
+use codex_app_server_protocol::FsReadFileParams;
+use codex_app_server_protocol::FsReadFileResponse;
+use codex_app_server_protocol::FsRemoveParams;
+use codex_app_server_protocol::FsRemoveResponse;
+use codex_app_server_protocol::FsWriteFileParams;
+use codex_app_server_protocol::FsWriteFileResponse;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) enum ExecServerInboundMessage {
@@ -56,6 +77,34 @@ pub(crate) enum ExecServerRequest {
         request_id: RequestId,
         params: TerminateParams,
     },
+    FsReadFile {
+        request_id: RequestId,
+        params: FsReadFileParams,
+    },
+    FsWriteFile {
+        request_id: RequestId,
+        params: FsWriteFileParams,
+    },
+    FsCreateDirectory {
+        request_id: RequestId,
+        params: FsCreateDirectoryParams,
+    },
+    FsGetMetadata {
+        request_id: RequestId,
+        params: FsGetMetadataParams,
+    },
+    FsReadDirectory {
+        request_id: RequestId,
+        params: FsReadDirectoryParams,
+    },
+    FsRemove {
+        request_id: RequestId,
+        params: FsRemoveParams,
+    },
+    FsCopy {
+        request_id: RequestId,
+        params: FsCopyParams,
+    },
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -83,6 +132,13 @@ pub(crate) enum ExecServerResponseMessage {
     Read(ReadResponse),
     Write(WriteResponse),
     Terminate(TerminateResponse),
+    FsReadFile(FsReadFileResponse),
+    FsWriteFile(FsWriteFileResponse),
+    FsCreateDirectory(FsCreateDirectoryResponse),
+    FsGetMetadata(FsGetMetadataResponse),
+    FsReadDirectory(FsReadDirectoryResponse),
+    FsRemove(FsRemoveResponse),
+    FsCopy(FsCopyResponse),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -178,6 +234,27 @@ fn route_request(request: JSONRPCRequest) -> Result<RoutedExecServerMessage, Str
         EXEC_TERMINATE_METHOD => Ok(parse_request_params(request, |request_id, params| {
             ExecServerRequest::Terminate { request_id, params }
         })),
+        FS_READ_FILE_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::FsReadFile { request_id, params }
+        })),
+        FS_WRITE_FILE_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::FsWriteFile { request_id, params }
+        })),
+        FS_CREATE_DIRECTORY_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::FsCreateDirectory { request_id, params }
+        })),
+        FS_GET_METADATA_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::FsGetMetadata { request_id, params }
+        })),
+        FS_READ_DIRECTORY_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::FsReadDirectory { request_id, params }
+        })),
+        FS_REMOVE_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::FsRemove { request_id, params }
+        })),
+        FS_COPY_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::FsCopy { request_id, params }
+        })),
         other => Ok(RoutedExecServerMessage::ImmediateOutbound(
             ExecServerOutboundMessage::Error {
                 request_id: request.id,
@@ -224,6 +301,13 @@ fn serialize_response(
         ExecServerResponseMessage::Read(response) => serde_json::to_value(response),
         ExecServerResponseMessage::Write(response) => serde_json::to_value(response),
         ExecServerResponseMessage::Terminate(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::FsReadFile(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::FsWriteFile(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::FsCreateDirectory(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::FsGetMetadata(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::FsReadDirectory(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::FsRemove(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::FsCopy(response) => serde_json::to_value(response),
     }
 }
 
@@ -261,6 +345,8 @@ mod tests {
     use crate::protocol::ExecExitedNotification;
     use crate::protocol::ExecParams;
     use crate::protocol::ExecResponse;
+    use crate::protocol::ExecSandboxConfig;
+    use crate::protocol::ExecSandboxMode;
     use crate::protocol::INITIALIZE_METHOD;
     use crate::protocol::INITIALIZED_METHOD;
     use crate::protocol::InitializeParams;
@@ -407,6 +493,51 @@ mod tests {
                 env: std::collections::HashMap::new(),
                 tty: true,
                 arg0: None,
+                sandbox: None,
+            }
+        );
+    }
+
+    #[test]
+    fn routes_exec_requests_with_optional_sandbox_config() {
+        let cwd = std::env::current_dir().expect("cwd");
+        let routed = route_jsonrpc_message(JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(4),
+            method: EXEC_METHOD.to_string(),
+            params: Some(json!({
+                "processId": "proc-1",
+                "argv": ["bash", "-lc", "true"],
+                "cwd": cwd,
+                "env": {},
+                "tty": true,
+                "arg0": null,
+                "sandbox": {
+                    "mode": "none",
+                },
+            })),
+            trace: None,
+        }))
+        .expect("exec request with sandbox should route");
+
+        let RoutedExecServerMessage::Inbound(ExecServerInboundMessage::Request(
+            ExecServerRequest::Exec { request_id, params },
+        )) = routed
+        else {
+            panic!("expected typed exec request");
+        };
+        assert_eq!(request_id, RequestId::Integer(4));
+        assert_eq!(
+            params,
+            ExecParams {
+                process_id: "proc-1".to_string(),
+                argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+                cwd: std::env::current_dir().expect("cwd"),
+                env: std::collections::HashMap::new(),
+                tty: true,
+                arg0: None,
+                sandbox: Some(ExecSandboxConfig {
+                    mode: ExecSandboxMode::None,
+                }),
             }
         );
     }

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

@@ -75,6 +75,125 @@ async fn exec_server_accepts_initialize_over_stdio() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_accepts_explicit_none_sandbox_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();
+
+    send_initialize_over_stdio(&mut stdin, &mut stdout).await?;
+
+    let exec = JSONRPCMessage::Request(JSONRPCRequest {
+        id: RequestId::Integer(2),
+        method: "process/start".to_string(),
+        params: Some(serde_json::json!({
+            "processId": "proc-1",
+            "argv": ["printf", "sandbox-none"],
+            "cwd": std::env::current_dir()?,
+            "env": {},
+            "tty": false,
+            "arg0": null,
+            "sandbox": {
+                "mode": "none"
+            }
+        })),
+        trace: None,
+    });
+    stdin
+        .write_all(format!("{}\n", serde_json::to_string(&exec)?).as_bytes())
+        .await?;
+
+    let response_line = timeout(Duration::from_secs(5), stdout.next_line()).await??;
+    let response_line = response_line.expect("exec response line");
+    let response: JSONRPCMessage = serde_json::from_str(&response_line)?;
+    let JSONRPCMessage::Response(JSONRPCResponse { id, result }) = response else {
+        panic!("expected process/start response");
+    };
+    assert_eq!(id, RequestId::Integer(2));
+    assert_eq!(result, serde_json::json!({ "processId": "proc-1" }));
+
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+    let mut saw_output = false;
+    while !saw_output {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        let line = timeout(remaining, stdout.next_line()).await??;
+        let line = line.context("missing process notification")?;
+        let message: JSONRPCMessage = serde_json::from_str(&line)?;
+        if let JSONRPCMessage::Notification(JSONRPCNotification { method, params }) = message
+            && method == "process/output"
+        {
+            let params = params.context("missing process/output params")?;
+            assert_eq!(params["processId"], "proc-1");
+            assert_eq!(params["stream"], "stdout");
+            assert_eq!(params["chunk"], "c2FuZGJveC1ub25l");
+            saw_output = true;
+        }
+    }
+
+    child.start_kill()?;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_rejects_host_default_sandbox_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();
+
+    send_initialize_over_stdio(&mut stdin, &mut stdout).await?;
+
+    let exec = JSONRPCMessage::Request(JSONRPCRequest {
+        id: RequestId::Integer(2),
+        method: "process/start".to_string(),
+        params: Some(serde_json::json!({
+            "processId": "proc-1",
+            "argv": ["bash", "-lc", "true"],
+            "cwd": std::env::current_dir()?,
+            "env": {},
+            "tty": false,
+            "arg0": null,
+            "sandbox": {
+                "mode": "hostDefault"
+            }
+        })),
+        trace: None,
+    });
+    stdin
+        .write_all(format!("{}\n", serde_json::to_string(&exec)?).as_bytes())
+        .await?;
+
+    let response_line = timeout(Duration::from_secs(5), stdout.next_line()).await??;
+    let response_line = response_line.expect("exec error line");
+    let response: JSONRPCMessage = serde_json::from_str(&response_line)?;
+    let JSONRPCMessage::Error(codex_app_server_protocol::JSONRPCError { id, error }) = response
+    else {
+        panic!("expected process/start error");
+    };
+    assert_eq!(id, RequestId::Integer(2));
+    assert_eq!(error.code, -32600);
+    assert_eq!(
+        error.message,
+        "sandbox mode `hostDefault` is not supported by exec-server yet"
+    );
+
+    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();
@@ -109,6 +228,7 @@ async fn exec_server_client_streams_output_and_accepts_writes() -> anyhow::Resul
             env,
             tty: true,
             arg0: None,
+            sandbox: None,
         })
         .await?;
     let process_id = response.process_id;
@@ -174,6 +294,7 @@ async fn exec_server_client_connects_over_websocket() -> anyhow::Result<()> {
             env,
             tty: true,
             arg0: None,
+            sandbox: None,
         })
         .await?;
     let process_id = response.process_id;
@@ -248,6 +369,7 @@ async fn websocket_disconnect_terminates_processes_for_that_connection() -> anyh
                 env,
                 tty: false,
                 arg0: None,
+                sandbox: None,
             })
             .await?;
     }
@@ -276,6 +398,48 @@ where
     Ok(websocket_url.to_string())
 }
 
+async fn send_initialize_over_stdio<W, R>(
+    stdin: &mut W,
+    stdout: &mut tokio::io::Lines<BufReader<R>>,
+) -> anyhow::Result<()>
+where
+    W: tokio::io::AsyncWrite + Unpin,
+    R: tokio::io::AsyncRead + Unpin,
+{
+    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
+        .ok_or_else(|| anyhow::anyhow!("missing initialize response line from stdio server"))?;
+    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?;
+
+    Ok(())
+}
+
 async fn recv_until_contains(
     events: &mut broadcast::Receiver<ExecServerEvent>,
     process_id: &str,