Add exec-server process RPC implementation

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

codex-rs/Cargo.lock

@@ -2003,6 +2003,26 @@ dependencies = [
  "wiremock",
 ]
 
+[[package]]
+name = "codex-exec-server"
+version = "0.0.0"
+dependencies = [
+ "anyhow",
+ "base64 0.22.1",
+ "clap",
+ "codex-app-server-protocol",
+ "codex-utils-cargo-bin",
+ "codex-utils-pty",
+ "futures",
+ "pretty_assertions",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+ "tokio-tungstenite",
+ "tracing",
+]
+
 [[package]]
 name = "codex-execpolicy"
 version = "0.0.0"

codex-rs/Cargo.toml

@@ -26,6 +26,7 @@ members = [
     "hooks",
     "secrets",
     "exec",
+    "exec-server",
     "execpolicy",
     "execpolicy-legacy",
     "keyring-store",

codex-rs/exec-server/BUILD.bazel

@@ -0,0 +1,7 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "exec-server",
+    crate_name = "codex_exec_server",
+    test_tags = ["no-sandbox"],
+)

codex-rs/exec-server/Cargo.toml

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

codex-rs/exec-server/README.md

@@ -0,0 +1,286 @@
+# codex-exec-server
+
+`codex-exec-server` is a small standalone stdio JSON-RPC server for spawning
+and controlling subprocesses through `codex-utils-pty`.
+
+This PR intentionally lands only the standalone binary, client, wire protocol,
+and docs. Exec and filesystem methods are stubbed server-side here and are
+implemented in follow-up PRs.
+
+It currently provides:
+
+- a standalone binary: `codex-exec-server`
+- a Rust client: `ExecServerClient`
+- a small protocol module with shared request/response types
+
+This crate is intentionally narrow. It is not wired into the main Codex CLI or
+unified-exec in this PR; it is only the standalone transport layer.
+
+## Transport
+
+The server speaks newline-delimited JSON-RPC 2.0 over stdio.
+
+- `stdin`: one JSON-RPC message per line
+- `stdout`: one JSON-RPC message per line
+- `stderr`: reserved for logs / process errors
+
+Like the app-server transport, messages on the wire omit the `"jsonrpc":"2.0"`
+field and use the shared `codex-app-server-protocol` envelope types.
+
+The current protocol version is:
+
+```text
+exec-server.v0
+```
+
+## Lifecycle
+
+Each connection follows this sequence:
+
+1. Send `initialize`.
+2. Wait for the `initialize` response.
+3. Send `initialized`.
+4. Call exec or filesystem RPCs once the follow-up implementation PRs land.
+
+If the server receives any notification other than `initialized`, it replies
+with an error using request id `-1`.
+
+If the stdio connection closes, the server terminates any remaining managed
+processes before exiting.
+
+## API
+
+### `initialize`
+
+Initial handshake request.
+
+Request params:
+
+```json
+{
+  "clientName": "my-client"
+}
+```
+
+Response:
+
+```json
+{
+  "protocolVersion": "exec-server.v0"
+}
+```
+
+### `initialized`
+
+Handshake acknowledgement notification sent by the client after a successful
+`initialize` response.
+
+Params are currently ignored. Sending any other notification method is treated
+as an invalid request.
+
+### `command/exec`
+
+Starts a new managed process.
+
+Request params:
+
+```json
+{
+  "processId": "proc-1",
+  "argv": ["bash", "-lc", "printf 'hello\\n'"],
+  "cwd": "/absolute/working/directory",
+  "env": {
+    "PATH": "/usr/bin:/bin"
+  },
+  "tty": true,
+  "outputBytesCap": 16384,
+  "arg0": null
+}
+```
+
+Field definitions:
+
+- `processId`: caller-chosen stable id for this process within the connection.
+- `argv`: command vector. It must be non-empty.
+- `cwd`: absolute working directory used for the child process.
+- `env`: environment variables passed to the child process.
+- `tty`: when `true`, spawn a PTY-backed interactive process; when `false`,
+  spawn a pipe-backed process with closed stdin.
+- `outputBytesCap`: maximum retained stdout/stderr bytes per stream for the
+  in-memory buffer. Defaults to `codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP`.
+- `arg0`: optional argv0 override forwarded to `codex-utils-pty`.
+
+Response:
+
+```json
+{
+  "processId": "proc-1",
+  "running": true,
+  "exitCode": null,
+  "stdout": null,
+  "stderr": null
+}
+```
+
+Behavior notes:
+
+- Reusing an existing `processId` is rejected.
+- PTY-backed processes accept later writes through `command/exec/write`.
+- Pipe-backed processes are launched with stdin closed and reject writes.
+- Output is streamed asynchronously via `command/exec/outputDelta`.
+- Exit is reported asynchronously via `command/exec/exited`.
+
+### `command/exec/write`
+
+Writes raw bytes to a running PTY-backed process stdin.
+
+Request params:
+
+```json
+{
+  "processId": "proc-1",
+  "chunk": "aGVsbG8K"
+}
+```
+
+`chunk` is base64-encoded raw bytes. In the example above it is `hello\n`.
+
+Response:
+
+```json
+{
+  "accepted": true
+}
+```
+
+Behavior notes:
+
+- Writes to an unknown `processId` are rejected.
+- Writes to a non-PTY process are rejected because stdin is already closed.
+
+### `command/exec/terminate`
+
+Terminates a running managed process.
+
+Request params:
+
+```json
+{
+  "processId": "proc-1"
+}
+```
+
+Response:
+
+```json
+{
+  "running": true
+}
+```
+
+If the process is already unknown or already removed, the server responds with:
+
+```json
+{
+  "running": false
+}
+```
+
+## Notifications
+
+### `command/exec/outputDelta`
+
+Streaming output chunk from a running process.
+
+Params:
+
+```json
+{
+  "processId": "proc-1",
+  "stream": "stdout",
+  "chunk": "aGVsbG8K"
+}
+```
+
+Fields:
+
+- `processId`: process identifier
+- `stream`: `"stdout"` or `"stderr"`
+- `chunk`: base64-encoded output bytes
+
+### `command/exec/exited`
+
+Final process exit notification.
+
+Params:
+
+```json
+{
+  "processId": "proc-1",
+  "exitCode": 0
+}
+```
+
+## Errors
+
+The server returns JSON-RPC errors with these codes:
+
+- `-32600`: invalid request
+- `-32602`: invalid params
+- `-32603`: internal error
+
+Typical error cases:
+
+- unknown method
+- malformed params
+- empty `argv`
+- duplicate `processId`
+- writes to unknown processes
+- writes to non-PTY processes
+
+## Rust surface
+
+The crate exports:
+
+- `ExecServerClient`
+- `ExecServerLaunchCommand`
+- `ExecServerProcess`
+- `ExecServerError`
+- protocol structs such as `ExecParams`, `ExecResponse`,
+  `WriteParams`, `TerminateParams`, `ExecOutputDeltaNotification`, and
+  `ExecExitedNotification`
+- `run_main()` for embedding the stdio server in a binary
+
+## Example session
+
+Initialize:
+
+```json
+{"id":1,"method":"initialize","params":{"clientName":"example-client"}}
+{"id":1,"result":{"protocolVersion":"exec-server.v0"}}
+{"method":"initialized","params":{}}
+```
+
+Start a process:
+
+```json
+{"id":2,"method":"command/exec","params":{"processId":"proc-1","argv":["bash","-lc","printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"],"cwd":"/tmp","env":{"PATH":"/usr/bin:/bin"},"tty":true,"outputBytesCap":4096,"arg0":null}}
+{"id":2,"result":{"processId":"proc-1","running":true,"exitCode":null,"stdout":null,"stderr":null}}
+{"method":"command/exec/outputDelta","params":{"processId":"proc-1","stream":"stdout","chunk":"cmVhZHkK"}}
+```
+
+Write to the process:
+
+```json
+{"id":3,"method":"command/exec/write","params":{"processId":"proc-1","chunk":"aGVsbG8K"}}
+{"id":3,"result":{"accepted":true}}
+{"method":"command/exec/outputDelta","params":{"processId":"proc-1","stream":"stdout","chunk":"ZWNobzpoZWxsbwo="}}
+```
+
+Terminate it:
+
+```json
+{"id":4,"method":"command/exec/terminate","params":{"processId":"proc-1"}}
+{"id":4,"result":{"running":true}}
+{"method":"command/exec/exited","params":{"processId":"proc-1","exitCode":0}}
+```

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

@@ -0,0 +1,23 @@
+use clap::Parser;
+use codex_exec_server::ExecServerTransport;
+
+#[derive(Debug, Parser)]
+struct ExecServerArgs {
+    /// Transport endpoint URL. Supported values: `stdio://` (default),
+    /// `ws://IP:PORT`.
+    #[arg(
+        long = "listen",
+        value_name = "URL",
+        default_value = ExecServerTransport::DEFAULT_LISTEN_URL
+    )]
+    listen: ExecServerTransport,
+}
+
+#[tokio::main]
+async fn main() {
+    let args = ExecServerArgs::parse();
+    if let Err(err) = codex_exec_server::run_main_with_transport(args.listen).await {
+        eprintln!("{err}");
+        std::process::exit(1);
+    }
+}

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

@@ -0,0 +1,428 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use codex_app_server_protocol::JSONRPCNotification;
+use serde_json::Value;
+use tokio::io::AsyncRead;
+use tokio::io::AsyncWrite;
+use tokio::sync::broadcast;
+use tokio::sync::mpsc;
+use tokio::time::timeout;
+use tokio_tungstenite::connect_async;
+use tracing::debug;
+use tracing::warn;
+
+use crate::client_api::ExecServerClientConnectOptions;
+use crate::client_api::ExecServerEvent;
+use crate::client_api::RemoteExecServerConnectArgs;
+use crate::connection::JsonRpcConnection;
+use crate::protocol::EXEC_EXITED_METHOD;
+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::EXEC_WRITE_METHOD;
+use crate::protocol::ExecExitedNotification;
+use crate::protocol::ExecOutputDeltaNotification;
+use crate::protocol::ExecParams;
+use crate::protocol::ExecResponse;
+use crate::protocol::INITIALIZE_METHOD;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::InitializeParams;
+use crate::protocol::InitializeResponse;
+use crate::protocol::ReadParams;
+use crate::protocol::ReadResponse;
+use crate::protocol::TerminateParams;
+use crate::protocol::TerminateResponse;
+use crate::protocol::WriteParams;
+use crate::protocol::WriteResponse;
+use crate::rpc::RpcCallError;
+use crate::rpc::RpcClient;
+use crate::rpc::RpcClientEvent;
+use crate::rpc::RpcNotificationSender;
+use crate::rpc::RpcServerOutboundMessage;
+
+mod local_backend;
+use local_backend::LocalBackend;
+
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+const INITIALIZE_TIMEOUT: Duration = Duration::from_secs(10);
+
+impl Default for ExecServerClientConnectOptions {
+    fn default() -> Self {
+        Self {
+            client_name: "codex-core".to_string(),
+            initialize_timeout: INITIALIZE_TIMEOUT,
+        }
+    }
+}
+
+impl From<RemoteExecServerConnectArgs> for ExecServerClientConnectOptions {
+    fn from(value: RemoteExecServerConnectArgs) -> Self {
+        Self {
+            client_name: value.client_name,
+            initialize_timeout: value.initialize_timeout,
+        }
+    }
+}
+
+impl RemoteExecServerConnectArgs {
+    pub fn new(websocket_url: String, client_name: String) -> Self {
+        Self {
+            websocket_url,
+            client_name,
+            connect_timeout: CONNECT_TIMEOUT,
+            initialize_timeout: INITIALIZE_TIMEOUT,
+        }
+    }
+}
+
+enum ClientBackend {
+    Remote(RpcClient),
+    InProcess(LocalBackend),
+}
+
+impl ClientBackend {
+    fn as_local(&self) -> Option<&LocalBackend> {
+        match self {
+            ClientBackend::Remote(_) => None,
+            ClientBackend::InProcess(backend) => Some(backend),
+        }
+    }
+
+    fn as_remote(&self) -> Option<&RpcClient> {
+        match self {
+            ClientBackend::Remote(client) => Some(client),
+            ClientBackend::InProcess(_) => None,
+        }
+    }
+}
+
+struct Inner {
+    backend: ClientBackend,
+    events_tx: broadcast::Sender<ExecServerEvent>,
+    reader_task: tokio::task::JoinHandle<()>,
+}
+
+impl Drop for Inner {
+    fn drop(&mut self) {
+        if let Some(backend) = self.backend.as_local()
+            && let Ok(handle) = tokio::runtime::Handle::try_current()
+        {
+            let backend = backend.clone();
+            handle.spawn(async move {
+                backend.shutdown().await;
+            });
+        }
+        self.reader_task.abort();
+    }
+}
+
+#[derive(Clone)]
+pub struct ExecServerClient {
+    inner: Arc<Inner>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ExecServerError {
+    #[error("failed to spawn exec-server: {0}")]
+    Spawn(#[source] std::io::Error),
+    #[error("timed out connecting to exec-server websocket `{url}` after {timeout:?}")]
+    WebSocketConnectTimeout { url: String, timeout: Duration },
+    #[error("failed to connect to exec-server websocket `{url}`: {source}")]
+    WebSocketConnect {
+        url: String,
+        #[source]
+        source: tokio_tungstenite::tungstenite::Error,
+    },
+    #[error("timed out waiting for exec-server initialize handshake after {timeout:?}")]
+    InitializeTimedOut { timeout: Duration },
+    #[error("exec-server transport closed")]
+    Closed,
+    #[error("failed to serialize or deserialize exec-server JSON: {0}")]
+    Json(#[from] serde_json::Error),
+    #[error("exec-server protocol error: {0}")]
+    Protocol(String),
+    #[error("exec-server rejected request ({code}): {message}")]
+    Server { code: i64, message: String },
+}
+
+impl ExecServerClient {
+    pub async fn connect_in_process(
+        options: ExecServerClientConnectOptions,
+    ) -> Result<Self, ExecServerError> {
+        let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<RpcServerOutboundMessage>(256);
+        let backend = LocalBackend::new(crate::server::ExecServerHandler::new(
+            RpcNotificationSender::new(outgoing_tx),
+        ));
+        let inner = Arc::new_cyclic(|weak| {
+            let weak = weak.clone();
+            let reader_task = tokio::spawn(async move {
+                while let Some(message) = outgoing_rx.recv().await {
+                    if let Some(inner) = weak.upgrade()
+                        && let Err(err) = handle_in_process_outbound_message(&inner, message).await
+                    {
+                        warn!(
+                            "in-process exec-server client closing after unexpected response: {err}"
+                        );
+                        return;
+                    }
+                }
+            });
+
+            Inner {
+                backend: ClientBackend::InProcess(backend),
+                events_tx: broadcast::channel(256).0,
+                reader_task,
+            }
+        });
+
+        let client = Self { inner };
+        client.initialize(options).await?;
+        Ok(client)
+    }
+
+    pub async fn connect_stdio<R, W>(
+        stdin: W,
+        stdout: R,
+        options: ExecServerClientConnectOptions,
+    ) -> Result<Self, ExecServerError>
+    where
+        R: AsyncRead + Unpin + Send + 'static,
+        W: AsyncWrite + Unpin + Send + 'static,
+    {
+        Self::connect(
+            JsonRpcConnection::from_stdio(stdout, stdin, "exec-server stdio".to_string()),
+            options,
+        )
+        .await
+    }
+
+    pub async fn connect_websocket(
+        args: RemoteExecServerConnectArgs,
+    ) -> Result<Self, 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,
+            })?;
+
+        Self::connect(
+            JsonRpcConnection::from_websocket(
+                stream,
+                format!("exec-server websocket {websocket_url}"),
+            ),
+            args.into(),
+        )
+        .await
+    }
+
+    pub fn event_receiver(&self) -> broadcast::Receiver<ExecServerEvent> {
+        self.inner.events_tx.subscribe()
+    }
+
+    pub async fn initialize(
+        &self,
+        options: ExecServerClientConnectOptions,
+    ) -> Result<InitializeResponse, ExecServerError> {
+        let ExecServerClientConnectOptions {
+            client_name,
+            initialize_timeout,
+        } = options;
+
+        timeout(initialize_timeout, async {
+            let response = if let Some(backend) = self.inner.backend.as_local() {
+                backend.initialize().await?
+            } else {
+                let params = InitializeParams { client_name };
+                let Some(remote) = self.inner.backend.as_remote() else {
+                    return Err(ExecServerError::Protocol(
+                        "remote backend missing during initialize".to_string(),
+                    ));
+                };
+                remote.call(INITIALIZE_METHOD, &params).await?
+            };
+            self.notify_initialized().await?;
+            Ok(response)
+        })
+        .await
+        .map_err(|_| ExecServerError::InitializeTimedOut {
+            timeout: initialize_timeout,
+        })?
+    }
+
+    pub async fn exec(&self, params: ExecParams) -> Result<ExecResponse, ExecServerError> {
+        if let Some(backend) = self.inner.backend.as_local() {
+            return backend.exec(params).await;
+        }
+        let Some(remote) = self.inner.backend.as_remote() else {
+            return Err(ExecServerError::Protocol(
+                "remote backend missing during exec".to_string(),
+            ));
+        };
+        remote.call(EXEC_METHOD, &params).await.map_err(Into::into)
+    }
+
+    pub async fn read(&self, params: ReadParams) -> Result<ReadResponse, ExecServerError> {
+        if let Some(backend) = self.inner.backend.as_local() {
+            return backend.exec_read(params).await;
+        }
+        let Some(remote) = self.inner.backend.as_remote() else {
+            return Err(ExecServerError::Protocol(
+                "remote backend missing during read".to_string(),
+            ));
+        };
+        remote
+            .call(EXEC_READ_METHOD, &params)
+            .await
+            .map_err(Into::into)
+    }
+
+    pub async fn write(
+        &self,
+        process_id: &str,
+        chunk: Vec<u8>,
+    ) -> Result<WriteResponse, ExecServerError> {
+        let params = WriteParams {
+            process_id: process_id.to_string(),
+            chunk: chunk.into(),
+        };
+        if let Some(backend) = self.inner.backend.as_local() {
+            return backend.exec_write(params).await;
+        }
+        let Some(remote) = self.inner.backend.as_remote() else {
+            return Err(ExecServerError::Protocol(
+                "remote backend missing during write".to_string(),
+            ));
+        };
+        remote
+            .call(EXEC_WRITE_METHOD, &params)
+            .await
+            .map_err(Into::into)
+    }
+
+    pub async fn terminate(&self, process_id: &str) -> Result<TerminateResponse, ExecServerError> {
+        let params = TerminateParams {
+            process_id: process_id.to_string(),
+        };
+        if let Some(backend) = self.inner.backend.as_local() {
+            return backend.terminate(params).await;
+        }
+        let Some(remote) = self.inner.backend.as_remote() else {
+            return Err(ExecServerError::Protocol(
+                "remote backend missing during terminate".to_string(),
+            ));
+        };
+        remote
+            .call(EXEC_TERMINATE_METHOD, &params)
+            .await
+            .map_err(Into::into)
+    }
+
+    async fn connect(
+        connection: JsonRpcConnection,
+        options: ExecServerClientConnectOptions,
+    ) -> Result<Self, ExecServerError> {
+        let (rpc_client, mut events_rx) = RpcClient::new(connection);
+        let inner = Arc::new_cyclic(|weak| {
+            let weak = weak.clone();
+            let reader_task = tokio::spawn(async move {
+                while let Some(event) = events_rx.recv().await {
+                    match event {
+                        RpcClientEvent::Notification(notification) => {
+                            if let Some(inner) = weak.upgrade()
+                                && let Err(err) =
+                                    handle_server_notification(&inner, notification).await
+                            {
+                                warn!("exec-server client closing after protocol error: {err}");
+                                return;
+                            }
+                        }
+                        RpcClientEvent::Disconnected { reason } => {
+                            if let Some(reason) = reason {
+                                warn!("exec-server client transport disconnected: {reason}");
+                            }
+                            return;
+                        }
+                    }
+                }
+            });
+
+            Inner {
+                backend: ClientBackend::Remote(rpc_client),
+                events_tx: broadcast::channel(256).0,
+                reader_task,
+            }
+        });
+
+        let client = Self { inner };
+        client.initialize(options).await?;
+        Ok(client)
+    }
+
+    async fn notify_initialized(&self) -> Result<(), ExecServerError> {
+        match &self.inner.backend {
+            ClientBackend::Remote(client) => client
+                .notify(INITIALIZED_METHOD, &serde_json::json!({}))
+                .await
+                .map_err(ExecServerError::Json),
+            ClientBackend::InProcess(backend) => backend.initialized().await,
+        }
+    }
+}
+
+impl From<RpcCallError> for ExecServerError {
+    fn from(value: RpcCallError) -> Self {
+        match value {
+            RpcCallError::Closed => Self::Closed,
+            RpcCallError::Json(err) => Self::Json(err),
+            RpcCallError::Server(error) => Self::Server {
+                code: error.code,
+                message: error.message,
+            },
+        }
+    }
+}
+
+async fn handle_in_process_outbound_message(
+    inner: &Arc<Inner>,
+    message: RpcServerOutboundMessage,
+) -> Result<(), ExecServerError> {
+    match message {
+        RpcServerOutboundMessage::Response { .. } | RpcServerOutboundMessage::Error { .. } => Err(
+            ExecServerError::Protocol("unexpected in-process RPC response".to_string()),
+        ),
+        RpcServerOutboundMessage::Notification(notification) => {
+            handle_server_notification(inner, notification).await
+        }
+    }
+}
+
+async fn handle_server_notification(
+    inner: &Arc<Inner>,
+    notification: JSONRPCNotification,
+) -> Result<(), ExecServerError> {
+    match notification.method.as_str() {
+        EXEC_OUTPUT_DELTA_METHOD => {
+            let params: ExecOutputDeltaNotification =
+                serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
+            let _ = inner.events_tx.send(ExecServerEvent::OutputDelta(params));
+        }
+        EXEC_EXITED_METHOD => {
+            let params: ExecExitedNotification =
+                serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
+            let _ = inner.events_tx.send(ExecServerEvent::Exited(params));
+        }
+        other => {
+            debug!("ignoring unknown exec-server notification: {other}");
+        }
+    }
+    Ok(())
+}

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

@@ -0,0 +1,95 @@
+use std::sync::Arc;
+
+use crate::protocol::ExecParams;
+use crate::protocol::ExecResponse;
+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 super::ExecServerError;
+
+#[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> {
+        self.handler
+            .initialize()
+            .map_err(|error| ExecServerError::Server {
+                code: error.code,
+                message: error.message,
+            })
+    }
+
+    pub(super) async fn initialized(&self) -> Result<(), ExecServerError> {
+        self.handler
+            .initialized()
+            .map_err(ExecServerError::Protocol)
+    }
+
+    pub(super) async fn exec(&self, params: ExecParams) -> Result<ExecResponse, ExecServerError> {
+        self.handler
+            .exec(params)
+            .await
+            .map_err(|error| ExecServerError::Server {
+                code: error.code,
+                message: error.message,
+            })
+    }
+
+    pub(super) async fn exec_read(
+        &self,
+        params: ReadParams,
+    ) -> Result<ReadResponse, ExecServerError> {
+        self.handler
+            .exec_read(params)
+            .await
+            .map_err(|error| ExecServerError::Server {
+                code: error.code,
+                message: error.message,
+            })
+    }
+
+    pub(super) async fn exec_write(
+        &self,
+        params: WriteParams,
+    ) -> Result<WriteResponse, ExecServerError> {
+        self.handler
+            .exec_write(params)
+            .await
+            .map_err(|error| ExecServerError::Server {
+                code: error.code,
+                message: error.message,
+            })
+    }
+
+    pub(super) async fn terminate(
+        &self,
+        params: TerminateParams,
+    ) -> Result<TerminateResponse, ExecServerError> {
+        self.handler
+            .terminate(params)
+            .await
+            .map_err(|error| ExecServerError::Server {
+                code: error.code,
+                message: error.message,
+            })
+    }
+}

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

@@ -0,0 +1,27 @@
+use std::time::Duration;
+
+use crate::protocol::ExecExitedNotification;
+use crate::protocol::ExecOutputDeltaNotification;
+
+/// Connection options for any exec-server client transport.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ExecServerClientConnectOptions {
+    pub client_name: String,
+    pub initialize_timeout: Duration,
+}
+
+/// WebSocket connection arguments for a remote exec-server.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RemoteExecServerConnectArgs {
+    pub websocket_url: String,
+    pub client_name: String,
+    pub connect_timeout: Duration,
+    pub initialize_timeout: Duration,
+}
+
+/// Connection-level server events.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ExecServerEvent {
+    OutputDelta(ExecOutputDeltaNotification),
+    Exited(ExecExitedNotification),
+}

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

@@ -0,0 +1,266 @@
+use codex_app_server_protocol::JSONRPCMessage;
+use futures::SinkExt;
+use futures::StreamExt;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncRead;
+use tokio::io::AsyncWrite;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::io::BufWriter;
+use tokio::sync::mpsc;
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::tungstenite::Message;
+
+pub(crate) const CHANNEL_CAPACITY: usize = 128;
+
+#[derive(Debug)]
+pub(crate) enum JsonRpcConnectionEvent {
+    Message(JSONRPCMessage),
+    Disconnected { reason: Option<String> },
+}
+
+pub(crate) struct JsonRpcConnection {
+    outgoing_tx: mpsc::Sender<JSONRPCMessage>,
+    incoming_rx: mpsc::Receiver<JsonRpcConnectionEvent>,
+    task_handles: Vec<tokio::task::JoinHandle<()>>,
+}
+
+impl JsonRpcConnection {
+    pub(crate) fn from_stdio<R, W>(reader: R, writer: W, connection_label: String) -> Self
+    where
+        R: AsyncRead + Unpin + Send + 'static,
+        W: AsyncWrite + Unpin + Send + 'static,
+    {
+        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(CHANNEL_CAPACITY);
+        let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY);
+
+        let reader_label = connection_label.clone();
+        let incoming_tx_for_reader = incoming_tx.clone();
+        let reader_task = tokio::spawn(async move {
+            let mut lines = BufReader::new(reader).lines();
+            loop {
+                match lines.next_line().await {
+                    Ok(Some(line)) => {
+                        if line.trim().is_empty() {
+                            continue;
+                        }
+                        match serde_json::from_str::<JSONRPCMessage>(&line) {
+                            Ok(message) => {
+                                if incoming_tx_for_reader
+                                    .send(JsonRpcConnectionEvent::Message(message))
+                                    .await
+                                    .is_err()
+                                {
+                                    break;
+                                }
+                            }
+                            Err(err) => {
+                                send_disconnected(
+                                    &incoming_tx_for_reader,
+                                    Some(format!(
+                                        "failed to parse JSON-RPC message from {reader_label}: {err}"
+                                    )),
+                                )
+                                .await;
+                                break;
+                            }
+                        }
+                    }
+                    Ok(None) => {
+                        send_disconnected(&incoming_tx_for_reader, /*reason*/ None).await;
+                        break;
+                    }
+                    Err(err) => {
+                        send_disconnected(
+                            &incoming_tx_for_reader,
+                            Some(format!(
+                                "failed to read JSON-RPC message from {reader_label}: {err}"
+                            )),
+                        )
+                        .await;
+                        break;
+                    }
+                }
+            }
+        });
+
+        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 {
+                    send_disconnected(
+                        &incoming_tx,
+                        Some(format!(
+                            "failed to write JSON-RPC message to {connection_label}: {err}"
+                        )),
+                    )
+                    .await;
+                    break;
+                }
+            }
+        });
+
+        Self {
+            outgoing_tx,
+            incoming_rx,
+            task_handles: vec![reader_task, writer_task],
+        }
+    }
+
+    pub(crate) fn from_websocket<S>(stream: WebSocketStream<S>, connection_label: String) -> Self
+    where
+        S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+    {
+        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(CHANNEL_CAPACITY);
+        let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY);
+        let (mut websocket_writer, mut websocket_reader) = stream.split();
+
+        let reader_label = connection_label.clone();
+        let incoming_tx_for_reader = incoming_tx.clone();
+        let reader_task = tokio::spawn(async move {
+            loop {
+                match websocket_reader.next().await {
+                    Some(Ok(Message::Text(text))) => {
+                        match serde_json::from_str::<JSONRPCMessage>(text.as_ref()) {
+                            Ok(message) => {
+                                if incoming_tx_for_reader
+                                    .send(JsonRpcConnectionEvent::Message(message))
+                                    .await
+                                    .is_err()
+                                {
+                                    break;
+                                }
+                            }
+                            Err(err) => {
+                                send_disconnected(
+                                    &incoming_tx_for_reader,
+                                    Some(format!(
+                                        "failed to parse websocket JSON-RPC message from {reader_label}: {err}"
+                                    )),
+                                )
+                                .await;
+                                break;
+                            }
+                        }
+                    }
+                    Some(Ok(Message::Binary(bytes))) => {
+                        match serde_json::from_slice::<JSONRPCMessage>(bytes.as_ref()) {
+                            Ok(message) => {
+                                if incoming_tx_for_reader
+                                    .send(JsonRpcConnectionEvent::Message(message))
+                                    .await
+                                    .is_err()
+                                {
+                                    break;
+                                }
+                            }
+                            Err(err) => {
+                                send_disconnected(
+                                    &incoming_tx_for_reader,
+                                    Some(format!(
+                                        "failed to parse websocket JSON-RPC message from {reader_label}: {err}"
+                                    )),
+                                )
+                                .await;
+                                break;
+                            }
+                        }
+                    }
+                    Some(Ok(Message::Close(_))) => {
+                        send_disconnected(&incoming_tx_for_reader, /*reason*/ None).await;
+                        break;
+                    }
+                    Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {}
+                    Some(Ok(_)) => {}
+                    Some(Err(err)) => {
+                        send_disconnected(
+                            &incoming_tx_for_reader,
+                            Some(format!(
+                                "failed to read websocket JSON-RPC message from {reader_label}: {err}"
+                            )),
+                        )
+                        .await;
+                        break;
+                    }
+                    None => {
+                        send_disconnected(&incoming_tx_for_reader, /*reason*/ None).await;
+                        break;
+                    }
+                }
+            }
+        });
+
+        let writer_task = tokio::spawn(async move {
+            while let Some(message) = outgoing_rx.recv().await {
+                match serialize_jsonrpc_message(&message) {
+                    Ok(encoded) => {
+                        if let Err(err) = websocket_writer.send(Message::Text(encoded.into())).await
+                        {
+                            send_disconnected(
+                                &incoming_tx,
+                                Some(format!(
+                                    "failed to write websocket JSON-RPC message to {connection_label}: {err}"
+                                )),
+                            )
+                            .await;
+                            break;
+                        }
+                    }
+                    Err(err) => {
+                        send_disconnected(
+                            &incoming_tx,
+                            Some(format!(
+                                "failed to serialize JSON-RPC message for {connection_label}: {err}"
+                            )),
+                        )
+                        .await;
+                        break;
+                    }
+                }
+            }
+        });
+
+        Self {
+            outgoing_tx,
+            incoming_rx,
+            task_handles: vec![reader_task, writer_task],
+        }
+    }
+
+    pub(crate) fn into_parts(
+        self,
+    ) -> (
+        mpsc::Sender<JSONRPCMessage>,
+        mpsc::Receiver<JsonRpcConnectionEvent>,
+        Vec<tokio::task::JoinHandle<()>>,
+    ) {
+        (self.outgoing_tx, self.incoming_rx, self.task_handles)
+    }
+}
+
+async fn send_disconnected(
+    incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>,
+    reason: Option<String>,
+) {
+    let _ = incoming_tx
+        .send(JsonRpcConnectionEvent::Disconnected { reason })
+        .await;
+}
+
+async fn write_jsonrpc_line_message<W>(
+    writer: &mut BufWriter<W>,
+    message: &JSONRPCMessage,
+) -> std::io::Result<()>
+where
+    W: AsyncWrite + Unpin,
+{
+    let encoded =
+        serialize_jsonrpc_message(message).map_err(|err| std::io::Error::other(err.to_string()))?;
+    writer.write_all(encoded.as_bytes()).await?;
+    writer.write_all(b"\n").await?;
+    writer.flush().await
+}
+
+fn serialize_jsonrpc_message(message: &JSONRPCMessage) -> Result<String, serde_json::Error> {
+    serde_json::to_string(message)
+}

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

@@ -0,0 +1,33 @@
+mod client;
+mod client_api;
+mod connection;
+mod local;
+mod protocol;
+mod rpc;
+mod server;
+
+pub use client::ExecServerClient;
+pub use client::ExecServerError;
+pub use client_api::ExecServerClientConnectOptions;
+pub use client_api::ExecServerEvent;
+pub use client_api::RemoteExecServerConnectArgs;
+pub use local::ExecServerLaunchCommand;
+pub use local::SpawnedExecServer;
+pub use local::spawn_local_exec_server;
+pub use protocol::ExecExitedNotification;
+pub use protocol::ExecOutputDeltaNotification;
+pub use protocol::ExecOutputStream;
+pub use protocol::ExecParams;
+pub use protocol::ExecResponse;
+pub use protocol::InitializeParams;
+pub use protocol::InitializeResponse;
+pub use protocol::ReadParams;
+pub use protocol::ReadResponse;
+pub use protocol::TerminateParams;
+pub use protocol::TerminateResponse;
+pub use protocol::WriteParams;
+pub use protocol::WriteResponse;
+pub use server::ExecServerTransport;
+pub use server::ExecServerTransportParseError;
+pub use server::run_main;
+pub use server::run_main_with_transport;

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

@@ -0,0 +1,70 @@
+use std::path::PathBuf;
+use std::process::Stdio;
+use std::sync::Mutex as StdMutex;
+
+use tokio::process::Child;
+use tokio::process::Command;
+
+use crate::client::ExecServerClient;
+use crate::client::ExecServerError;
+use crate::client_api::ExecServerClientConnectOptions;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ExecServerLaunchCommand {
+    pub program: PathBuf,
+    pub args: Vec<String>,
+}
+
+pub struct SpawnedExecServer {
+    client: ExecServerClient,
+    child: StdMutex<Option<Child>>,
+}
+
+impl SpawnedExecServer {
+    pub fn client(&self) -> &ExecServerClient {
+        &self.client
+    }
+}
+
+impl Drop for SpawnedExecServer {
+    fn drop(&mut self) {
+        if let Ok(mut child_guard) = self.child.lock()
+            && let Some(child) = child_guard.as_mut()
+        {
+            let _ = child.start_kill();
+        }
+    }
+}
+
+pub async fn spawn_local_exec_server(
+    command: ExecServerLaunchCommand,
+    options: ExecServerClientConnectOptions,
+) -> Result<SpawnedExecServer, ExecServerError> {
+    let mut child = Command::new(&command.program);
+    child.args(&command.args);
+    child.stdin(Stdio::piped());
+    child.stdout(Stdio::piped());
+    child.stderr(Stdio::inherit());
+    child.kill_on_drop(true);
+
+    let mut child = child.spawn().map_err(ExecServerError::Spawn)?;
+    let stdin = child.stdin.take().ok_or_else(|| {
+        ExecServerError::Protocol("exec-server stdin was not captured".to_string())
+    })?;
+    let stdout = child.stdout.take().ok_or_else(|| {
+        ExecServerError::Protocol("exec-server stdout was not captured".to_string())
+    })?;
+
+    let client = match ExecServerClient::connect_stdio(stdin, stdout, options).await {
+        Ok(client) => client,
+        Err(err) => {
+            let _ = child.start_kill();
+            return Err(err);
+        }
+    };
+
+    Ok(SpawnedExecServer {
+        client,
+        child: StdMutex::new(Some(child)),
+    })
+}

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

@@ -0,0 +1,162 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use serde::Deserialize;
+use serde::Serialize;
+
+pub const INITIALIZE_METHOD: &str = "initialize";
+pub const INITIALIZED_METHOD: &str = "initialized";
+pub const EXEC_METHOD: &str = "process/start";
+pub const EXEC_READ_METHOD: &str = "process/read";
+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 PROTOCOL_VERSION: &str = "exec-server.v0";
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct ByteChunk(#[serde(with = "base64_bytes")] pub Vec<u8>);
+
+impl ByteChunk {
+    pub fn into_inner(self) -> Vec<u8> {
+        self.0
+    }
+}
+
+impl From<Vec<u8>> for ByteChunk {
+    fn from(value: Vec<u8>) -> Self {
+        Self(value)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct InitializeParams {
+    pub client_name: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct InitializeResponse {
+    pub protocol_version: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecParams {
+    /// Client-chosen logical process handle scoped to this connection/session.
+    /// This is a protocol key, not an OS pid.
+    pub process_id: String,
+    pub argv: Vec<String>,
+    pub cwd: PathBuf,
+    pub env: HashMap<String, String>,
+    pub tty: bool,
+    pub arg0: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecResponse {
+    pub process_id: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ReadParams {
+    pub process_id: String,
+    pub after_seq: Option<u64>,
+    pub max_bytes: Option<usize>,
+    pub wait_ms: Option<u64>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ProcessOutputChunk {
+    pub seq: u64,
+    pub stream: ExecOutputStream,
+    pub chunk: ByteChunk,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ReadResponse {
+    pub chunks: Vec<ProcessOutputChunk>,
+    pub next_seq: u64,
+    pub exited: bool,
+    pub exit_code: Option<i32>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WriteParams {
+    pub process_id: String,
+    pub chunk: ByteChunk,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WriteResponse {
+    pub accepted: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TerminateParams {
+    pub process_id: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TerminateResponse {
+    pub running: bool,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum ExecOutputStream {
+    Stdout,
+    Stderr,
+    Pty,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecOutputDeltaNotification {
+    pub process_id: String,
+    pub stream: ExecOutputStream,
+    pub chunk: ByteChunk,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecExitedNotification {
+    pub process_id: String,
+    pub exit_code: i32,
+}
+
+mod base64_bytes {
+    use super::BASE64_STANDARD;
+    use base64::Engine as _;
+    use serde::Deserialize;
+    use serde::Deserializer;
+    use serde::Serializer;
+
+    pub fn serialize<S>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&BASE64_STANDARD.encode(bytes))
+    }
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let encoded = String::deserialize(deserializer)?;
+        BASE64_STANDARD
+            .decode(encoded)
+            .map_err(serde::de::Error::custom)
+    }
+}

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

@@ -0,0 +1,562 @@
+use std::collections::HashMap;
+use std::future::Future;
+use std::pin::Pin;
+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>>;
+type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
+type RequestRoute<S> =
+    Box<dyn Fn(Arc<S>, JSONRPCRequest) -> BoxFuture<RpcServerOutboundMessage> + Send + Sync>;
+type NotificationRoute<S> =
+    Box<dyn Fn(Arc<S>, JSONRPCNotification) -> BoxFuture<Result<(), String>> + Send + Sync>;
+
+#[derive(Debug)]
+pub(crate) enum RpcClientEvent {
+    Notification(JSONRPCNotification),
+    Disconnected { reason: Option<String> },
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) enum RpcServerOutboundMessage {
+    Response {
+        request_id: RequestId,
+        result: Value,
+    },
+    Error {
+        request_id: RequestId,
+        error: JSONRPCErrorError,
+    },
+    #[allow(dead_code)]
+    Notification(JSONRPCNotification),
+}
+
+#[allow(dead_code)]
+#[derive(Clone)]
+pub(crate) struct RpcNotificationSender {
+    outgoing_tx: mpsc::Sender<RpcServerOutboundMessage>,
+}
+
+impl RpcNotificationSender {
+    pub(crate) fn new(outgoing_tx: mpsc::Sender<RpcServerOutboundMessage>) -> Self {
+        Self { outgoing_tx }
+    }
+
+    #[allow(dead_code)]
+    pub(crate) async fn notify<P: Serialize>(
+        &self,
+        method: &str,
+        params: &P,
+    ) -> Result<(), JSONRPCErrorError> {
+        let params = serde_json::to_value(params).map_err(|err| internal_error(err.to_string()))?;
+        self.outgoing_tx
+            .send(RpcServerOutboundMessage::Notification(
+                JSONRPCNotification {
+                    method: method.to_string(),
+                    params: Some(params),
+                },
+            ))
+            .await
+            .map_err(|_| internal_error("RPC connection closed while sending notification".into()))
+    }
+}
+
+pub(crate) struct RpcRouter<S> {
+    request_routes: HashMap<&'static str, RequestRoute<S>>,
+    notification_routes: HashMap<&'static str, NotificationRoute<S>>,
+}
+
+impl<S> Default for RpcRouter<S> {
+    fn default() -> Self {
+        Self {
+            request_routes: HashMap::new(),
+            notification_routes: HashMap::new(),
+        }
+    }
+}
+
+impl<S> RpcRouter<S>
+where
+    S: Send + Sync + 'static,
+{
+    pub(crate) fn new() -> Self {
+        Self::default()
+    }
+
+    pub(crate) fn request<P, R, F, Fut>(&mut self, method: &'static str, handler: F)
+    where
+        P: DeserializeOwned + Send + 'static,
+        R: Serialize + Send + 'static,
+        F: Fn(Arc<S>, P) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<R, JSONRPCErrorError>> + Send + 'static,
+    {
+        self.request_routes.insert(
+            method,
+            Box::new(move |state, request| {
+                let request_id = request.id;
+                let params = request.params;
+                let response =
+                    decode_request_params::<P>(params).map(|params| handler(state, params));
+                Box::pin(async move {
+                    let response = match response {
+                        Ok(response) => response.await,
+                        Err(error) => {
+                            return RpcServerOutboundMessage::Error { request_id, error };
+                        }
+                    };
+                    match response {
+                        Ok(result) => match serde_json::to_value(result) {
+                            Ok(result) => RpcServerOutboundMessage::Response { request_id, result },
+                            Err(err) => RpcServerOutboundMessage::Error {
+                                request_id,
+                                error: internal_error(err.to_string()),
+                            },
+                        },
+                        Err(error) => RpcServerOutboundMessage::Error { request_id, error },
+                    }
+                })
+            }),
+        );
+    }
+
+    pub(crate) fn notification<P, F, Fut>(&mut self, method: &'static str, handler: F)
+    where
+        P: DeserializeOwned + Send + 'static,
+        F: Fn(Arc<S>, P) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), String>> + Send + 'static,
+    {
+        self.notification_routes.insert(
+            method,
+            Box::new(move |state, notification| {
+                let params = decode_notification_params::<P>(notification.params)
+                    .map(|params| handler(state, params));
+                Box::pin(async move {
+                    let handler = match params {
+                        Ok(handler) => handler,
+                        Err(err) => return Err(err),
+                    };
+                    handler.await
+                })
+            }),
+        );
+    }
+
+    pub(crate) fn request_route(&self, method: &str) -> Option<&RequestRoute<S>> {
+        self.request_routes.get(method)
+    }
+
+    pub(crate) fn notification_route(&self, method: &str) -> Option<&NotificationRoute<S>> {
+        self.notification_routes.get(method)
+    }
+}
+
+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)]
+    #[allow(dead_code)]
+    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),
+}
+
+pub(crate) fn encode_server_message(
+    message: RpcServerOutboundMessage,
+) -> Result<JSONRPCMessage, serde_json::Error> {
+    match message {
+        RpcServerOutboundMessage::Response { request_id, result } => {
+            Ok(JSONRPCMessage::Response(JSONRPCResponse {
+                id: request_id,
+                result,
+            }))
+        }
+        RpcServerOutboundMessage::Error { request_id, error } => {
+            Ok(JSONRPCMessage::Error(JSONRPCError {
+                id: request_id,
+                error,
+            }))
+        }
+        RpcServerOutboundMessage::Notification(notification) => {
+            Ok(JSONRPCMessage::Notification(notification))
+        }
+    }
+}
+
+pub(crate) fn invalid_request(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32600,
+        data: None,
+        message,
+    }
+}
+
+pub(crate) fn method_not_found(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32601,
+        data: None,
+        message,
+    }
+}
+
+pub(crate) fn invalid_params(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32602,
+        data: None,
+        message,
+    }
+}
+
+pub(crate) fn internal_error(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32603,
+        data: None,
+        message,
+    }
+}
+
+fn decode_request_params<P>(params: Option<Value>) -> Result<P, JSONRPCErrorError>
+where
+    P: DeserializeOwned,
+{
+    decode_params(params).map_err(|err| invalid_params(err.to_string()))
+}
+
+fn decode_notification_params<P>(params: Option<Value>) -> Result<P, String>
+where
+    P: DeserializeOwned,
+{
+    decode_params(params).map_err(|err| err.to_string())
+}
+
+fn decode_params<P>(params: Option<Value>) -> Result<P, serde_json::Error>
+where
+    P: DeserializeOwned,
+{
+    let params = params.unwrap_or(Value::Null);
+    match serde_json::from_value(params.clone()) {
+        Ok(params) => Ok(params),
+        Err(err) => {
+            if matches!(params, Value::Object(ref map) if map.is_empty()) {
+                serde_json::from_value(Value::Null).map_err(|_| err)
+            } else {
+                Err(err)
+            }
+        }
+    }
+}
+
+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(),
+        }));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use codex_app_server_protocol::JSONRPCMessage;
+    use codex_app_server_protocol::JSONRPCResponse;
+    use pretty_assertions::assert_eq;
+    use tokio::io::AsyncBufReadExt;
+    use tokio::io::AsyncWriteExt;
+    use tokio::io::BufReader;
+    use tokio::time::timeout;
+
+    use super::RpcClient;
+    use crate::connection::JsonRpcConnection;
+
+    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 rpc_client_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 (client, _events_rx) = RpcClient::new(JsonRpcConnection::from_stdio(
+            client_stdout,
+            client_stdin,
+            "test-rpc".to_string(),
+        ));
+
+        let server = tokio::spawn(async move {
+            let mut lines = BufReader::new(server_reader).lines();
+
+            let first = read_jsonrpc_line(&mut lines).await;
+            let second = read_jsonrpc_line(&mut lines).await;
+            let (slow_request, fast_request) = match (first, second) {
+                (
+                    JSONRPCMessage::Request(first_request),
+                    JSONRPCMessage::Request(second_request),
+                ) if first_request.method == "slow" && second_request.method == "fast" => {
+                    (first_request, second_request)
+                }
+                (
+                    JSONRPCMessage::Request(first_request),
+                    JSONRPCMessage::Request(second_request),
+                ) if first_request.method == "fast" && second_request.method == "slow" => {
+                    (second_request, first_request)
+                }
+                _ => panic!("expected slow and fast requests"),
+            };
+
+            write_jsonrpc_line(
+                &mut server_writer,
+                JSONRPCMessage::Response(JSONRPCResponse {
+                    id: fast_request.id,
+                    result: serde_json::json!({ "value": "fast" }),
+                }),
+            )
+            .await;
+            write_jsonrpc_line(
+                &mut server_writer,
+                JSONRPCMessage::Response(JSONRPCResponse {
+                    id: slow_request.id,
+                    result: serde_json::json!({ "value": "slow" }),
+                }),
+            )
+            .await;
+        });
+
+        let slow_params = serde_json::json!({ "n": 1 });
+        let fast_params = serde_json::json!({ "n": 2 });
+        let (slow, fast) = tokio::join!(
+            client.call::<_, serde_json::Value>("slow", &slow_params),
+            client.call::<_, serde_json::Value>("fast", &fast_params),
+        );
+
+        let slow = slow.unwrap_or_else(|err| panic!("slow request failed: {err:?}"));
+        let fast = fast.unwrap_or_else(|err| panic!("fast request failed: {err:?}"));
+        assert_eq!(slow, serde_json::json!({ "value": "slow" }));
+        assert_eq!(fast, serde_json::json!({ "value": "fast" }));
+
+        assert_eq!(client.pending_request_count().await, 0);
+
+        if let Err(err) = server.await {
+            panic!("server task failed: {err}");
+        }
+    }
+}

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

@@ -0,0 +1,18 @@
+mod handler;
+mod processor;
+mod registry;
+mod transport;
+
+pub(crate) use handler::ExecServerHandler;
+pub use transport::ExecServerTransport;
+pub use transport::ExecServerTransportParseError;
+
+pub async fn run_main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    run_main_with_transport(ExecServerTransport::Stdio).await
+}
+
+pub async fn run_main_with_transport(
+    transport: ExecServerTransport,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    transport::run_transport(transport).await
+}

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

@@ -0,0 +1,444 @@
+use std::collections::HashMap;
+use std::collections::VecDeque;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_utils_pty::ExecCommandSession;
+use codex_utils_pty::TerminalSize;
+use tokio::sync::Mutex;
+use tokio::sync::Notify;
+use tracing::warn;
+
+use crate::protocol::ExecExitedNotification;
+use crate::protocol::ExecOutputDeltaNotification;
+use crate::protocol::ExecOutputStream;
+use crate::protocol::ExecParams;
+use crate::protocol::ExecResponse;
+use crate::protocol::InitializeResponse;
+use crate::protocol::PROTOCOL_VERSION;
+use crate::protocol::ProcessOutputChunk;
+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::rpc::RpcNotificationSender;
+use crate::rpc::internal_error;
+use crate::rpc::invalid_params;
+use crate::rpc::invalid_request;
+
+const RETAINED_OUTPUT_BYTES_PER_PROCESS: usize = 1024 * 1024;
+#[cfg(test)]
+const EXITED_PROCESS_RETENTION: Duration = Duration::from_millis(25);
+#[cfg(not(test))]
+const EXITED_PROCESS_RETENTION: Duration = Duration::from_secs(30);
+
+#[derive(Clone)]
+struct RetainedOutputChunk {
+    seq: u64,
+    stream: ExecOutputStream,
+    chunk: Vec<u8>,
+}
+
+struct RunningProcess {
+    session: ExecCommandSession,
+    tty: bool,
+    output: VecDeque<RetainedOutputChunk>,
+    retained_bytes: usize,
+    next_seq: u64,
+    exit_code: Option<i32>,
+    output_notify: Arc<Notify>,
+}
+
+enum ProcessEntry {
+    Starting,
+    Running(Box<RunningProcess>),
+}
+
+pub(crate) struct ExecServerHandler {
+    notifications: RpcNotificationSender,
+    processes: Arc<Mutex<HashMap<String, ProcessEntry>>>,
+    initialize_requested: AtomicBool,
+    initialized: AtomicBool,
+}
+
+impl ExecServerHandler {
+    pub(crate) fn new(notifications: RpcNotificationSender) -> Self {
+        Self {
+            notifications,
+            processes: Arc::new(Mutex::new(HashMap::new())),
+            initialize_requested: AtomicBool::new(false),
+            initialized: AtomicBool::new(false),
+        }
+    }
+
+    pub(crate) async fn shutdown(&self) {
+        let remaining = {
+            let mut processes = self.processes.lock().await;
+            processes
+                .drain()
+                .filter_map(|(_, process)| match process {
+                    ProcessEntry::Starting => None,
+                    ProcessEntry::Running(process) => Some(process),
+                })
+                .collect::<Vec<_>>()
+        };
+        for process in remaining {
+            process.session.terminate();
+        }
+    }
+
+    pub(crate) fn initialize(&self) -> Result<InitializeResponse, JSONRPCErrorError> {
+        if self.initialize_requested.swap(true, Ordering::SeqCst) {
+            return Err(invalid_request(
+                "initialize may only be sent once per connection".to_string(),
+            ));
+        }
+        Ok(InitializeResponse {
+            protocol_version: PROTOCOL_VERSION.to_string(),
+        })
+    }
+
+    pub(crate) fn initialized(&self) -> Result<(), String> {
+        if !self.initialize_requested.load(Ordering::SeqCst) {
+            return Err("received `initialized` notification before `initialize`".into());
+        }
+        self.initialized.store(true, Ordering::SeqCst);
+        Ok(())
+    }
+
+    fn require_initialized(&self) -> Result<(), JSONRPCErrorError> {
+        if !self.initialize_requested.load(Ordering::SeqCst) {
+            return Err(invalid_request(
+                "client must call initialize before using exec methods".to_string(),
+            ));
+        }
+        if !self.initialized.load(Ordering::SeqCst) {
+            return Err(invalid_request(
+                "client must send initialized before using exec methods".to_string(),
+            ));
+        }
+        Ok(())
+    }
+
+    pub(crate) async fn exec(&self, params: ExecParams) -> Result<ExecResponse, JSONRPCErrorError> {
+        self.require_initialized()?;
+        let process_id = params.process_id.clone();
+
+        let (program, args) = params
+            .argv
+            .split_first()
+            .ok_or_else(|| invalid_params("argv must not be empty".to_string()))?;
+
+        {
+            let mut process_map = self.processes.lock().await;
+            if process_map.contains_key(&process_id) {
+                return Err(invalid_request(format!(
+                    "process {process_id} already exists"
+                )));
+            }
+            process_map.insert(process_id.clone(), ProcessEntry::Starting);
+        }
+
+        let spawned_result = if params.tty {
+            codex_utils_pty::spawn_pty_process(
+                program,
+                args,
+                params.cwd.as_path(),
+                &params.env,
+                &params.arg0,
+                TerminalSize::default(),
+            )
+            .await
+        } else {
+            codex_utils_pty::spawn_pipe_process_no_stdin(
+                program,
+                args,
+                params.cwd.as_path(),
+                &params.env,
+                &params.arg0,
+            )
+            .await
+        };
+        let spawned = match spawned_result {
+            Ok(spawned) => spawned,
+            Err(err) => {
+                let mut process_map = self.processes.lock().await;
+                if matches!(process_map.get(&process_id), Some(ProcessEntry::Starting)) {
+                    process_map.remove(&process_id);
+                }
+                return Err(internal_error(err.to_string()));
+            }
+        };
+
+        let output_notify = Arc::new(Notify::new());
+        {
+            let mut process_map = self.processes.lock().await;
+            process_map.insert(
+                process_id.clone(),
+                ProcessEntry::Running(Box::new(RunningProcess {
+                    session: spawned.session,
+                    tty: params.tty,
+                    output: VecDeque::new(),
+                    retained_bytes: 0,
+                    next_seq: 1,
+                    exit_code: None,
+                    output_notify: Arc::clone(&output_notify),
+                })),
+            );
+        }
+
+        tokio::spawn(stream_output(
+            process_id.clone(),
+            if params.tty {
+                ExecOutputStream::Pty
+            } else {
+                ExecOutputStream::Stdout
+            },
+            spawned.stdout_rx,
+            self.notifications.clone(),
+            Arc::clone(&self.processes),
+            Arc::clone(&output_notify),
+        ));
+        tokio::spawn(stream_output(
+            process_id.clone(),
+            if params.tty {
+                ExecOutputStream::Pty
+            } else {
+                ExecOutputStream::Stderr
+            },
+            spawned.stderr_rx,
+            self.notifications.clone(),
+            Arc::clone(&self.processes),
+            Arc::clone(&output_notify),
+        ));
+        tokio::spawn(watch_exit(
+            process_id.clone(),
+            spawned.exit_rx,
+            self.notifications.clone(),
+            Arc::clone(&self.processes),
+            output_notify,
+        ));
+
+        Ok(ExecResponse { process_id })
+    }
+
+    pub(crate) async fn exec_read(
+        &self,
+        params: ReadParams,
+    ) -> Result<ReadResponse, JSONRPCErrorError> {
+        self.require_initialized()?;
+        let after_seq = params.after_seq.unwrap_or(0);
+        let max_bytes = params.max_bytes.unwrap_or(usize::MAX);
+        let wait = Duration::from_millis(params.wait_ms.unwrap_or(0));
+        let deadline = tokio::time::Instant::now() + wait;
+
+        loop {
+            let (response, output_notify) = {
+                let process_map = self.processes.lock().await;
+                let process = process_map.get(&params.process_id).ok_or_else(|| {
+                    invalid_request(format!("unknown process id {}", params.process_id))
+                })?;
+                let ProcessEntry::Running(process) = process else {
+                    return Err(invalid_request(format!(
+                        "process id {} is starting",
+                        params.process_id
+                    )));
+                };
+
+                let mut chunks = Vec::new();
+                let mut total_bytes = 0;
+                let mut next_seq = process.next_seq;
+                for retained in process.output.iter().filter(|chunk| chunk.seq > after_seq) {
+                    let chunk_len = retained.chunk.len();
+                    if !chunks.is_empty() && total_bytes + chunk_len > max_bytes {
+                        break;
+                    }
+                    total_bytes += chunk_len;
+                    chunks.push(ProcessOutputChunk {
+                        seq: retained.seq,
+                        stream: retained.stream,
+                        chunk: retained.chunk.clone().into(),
+                    });
+                    next_seq = retained.seq + 1;
+                    if total_bytes >= max_bytes {
+                        break;
+                    }
+                }
+
+                (
+                    ReadResponse {
+                        chunks,
+                        next_seq,
+                        exited: process.exit_code.is_some(),
+                        exit_code: process.exit_code,
+                    },
+                    Arc::clone(&process.output_notify),
+                )
+            };
+
+            if !response.chunks.is_empty()
+                || response.exited
+                || tokio::time::Instant::now() >= deadline
+            {
+                return Ok(response);
+            }
+
+            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+            if remaining.is_zero() {
+                return Ok(response);
+            }
+            let _ = tokio::time::timeout(remaining, output_notify.notified()).await;
+        }
+    }
+
+    pub(crate) async fn exec_write(
+        &self,
+        params: WriteParams,
+    ) -> Result<WriteResponse, JSONRPCErrorError> {
+        self.require_initialized()?;
+        let writer_tx = {
+            let process_map = self.processes.lock().await;
+            let process = process_map.get(&params.process_id).ok_or_else(|| {
+                invalid_request(format!("unknown process id {}", params.process_id))
+            })?;
+            let ProcessEntry::Running(process) = process else {
+                return Err(invalid_request(format!(
+                    "process id {} is starting",
+                    params.process_id
+                )));
+            };
+            if !process.tty {
+                return Err(invalid_request(format!(
+                    "stdin is closed for process {}",
+                    params.process_id
+                )));
+            }
+            process.session.writer_sender()
+        };
+
+        writer_tx
+            .send(params.chunk.into_inner())
+            .await
+            .map_err(|_| internal_error("failed to write to process stdin".to_string()))?;
+
+        Ok(WriteResponse { accepted: true })
+    }
+
+    pub(crate) async fn terminate(
+        &self,
+        params: TerminateParams,
+    ) -> Result<TerminateResponse, JSONRPCErrorError> {
+        self.require_initialized()?;
+        let running = {
+            let process_map = self.processes.lock().await;
+            match process_map.get(&params.process_id) {
+                Some(ProcessEntry::Running(process)) => {
+                    process.session.terminate();
+                    true
+                }
+                Some(ProcessEntry::Starting) | None => false,
+            }
+        };
+
+        Ok(TerminateResponse { running })
+    }
+}
+
+async fn stream_output(
+    process_id: String,
+    stream: ExecOutputStream,
+    mut receiver: tokio::sync::mpsc::Receiver<Vec<u8>>,
+    notifications: RpcNotificationSender,
+    processes: Arc<Mutex<HashMap<String, ProcessEntry>>>,
+    output_notify: Arc<Notify>,
+) {
+    while let Some(chunk) = receiver.recv().await {
+        let notification = {
+            let mut processes = processes.lock().await;
+            let Some(entry) = processes.get_mut(&process_id) else {
+                break;
+            };
+            let ProcessEntry::Running(process) = entry else {
+                break;
+            };
+            let seq = process.next_seq;
+            process.next_seq += 1;
+            process.retained_bytes += chunk.len();
+            process.output.push_back(RetainedOutputChunk {
+                seq,
+                stream,
+                chunk: chunk.clone(),
+            });
+            while process.retained_bytes > RETAINED_OUTPUT_BYTES_PER_PROCESS {
+                let Some(evicted) = process.output.pop_front() else {
+                    break;
+                };
+                process.retained_bytes = process.retained_bytes.saturating_sub(evicted.chunk.len());
+                warn!(
+                    "retained output cap exceeded for process {process_id}; dropping oldest output"
+                );
+            }
+            ExecOutputDeltaNotification {
+                process_id: process_id.clone(),
+                stream,
+                chunk: chunk.into(),
+            }
+        };
+        output_notify.notify_waiters();
+
+        if notifications
+            .notify(crate::protocol::EXEC_OUTPUT_DELTA_METHOD, &notification)
+            .await
+            .is_err()
+        {
+            break;
+        }
+    }
+}
+
+async fn watch_exit(
+    process_id: String,
+    exit_rx: tokio::sync::oneshot::Receiver<i32>,
+    notifications: RpcNotificationSender,
+    processes: Arc<Mutex<HashMap<String, ProcessEntry>>>,
+    output_notify: Arc<Notify>,
+) {
+    let exit_code = exit_rx.await.unwrap_or(-1);
+    {
+        let mut processes = processes.lock().await;
+        if let Some(ProcessEntry::Running(process)) = processes.get_mut(&process_id) {
+            process.exit_code = Some(exit_code);
+        }
+    }
+    output_notify.notify_waiters();
+    if notifications
+        .notify(
+            crate::protocol::EXEC_EXITED_METHOD,
+            &ExecExitedNotification {
+                process_id: process_id.clone(),
+                exit_code,
+            },
+        )
+        .await
+        .is_err()
+    {
+        return;
+    }
+
+    tokio::time::sleep(EXITED_PROCESS_RETENTION).await;
+    let mut processes = processes.lock().await;
+    if matches!(
+        processes.get(&process_id),
+        Some(ProcessEntry::Running(process)) if process.exit_code == Some(exit_code)
+    ) {
+        processes.remove(&process_id);
+    }
+}
+
+#[cfg(test)]
+mod tests;

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

@@ -0,0 +1,74 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use pretty_assertions::assert_eq;
+use tokio::sync::mpsc;
+
+use super::ExecServerHandler;
+use crate::protocol::ExecParams;
+use crate::protocol::InitializeResponse;
+use crate::protocol::PROTOCOL_VERSION;
+use crate::rpc::RpcNotificationSender;
+
+fn exec_params(process_id: &str) -> ExecParams {
+    let mut env = HashMap::new();
+    if let Some(path) = std::env::var_os("PATH") {
+        env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
+    }
+    ExecParams {
+        process_id: process_id.to_string(),
+        argv: vec![
+            "bash".to_string(),
+            "-lc".to_string(),
+            "sleep 0.1".to_string(),
+        ],
+        cwd: std::env::current_dir().expect("cwd"),
+        env,
+        tty: false,
+        arg0: None,
+    }
+}
+
+async fn initialized_handler() -> Arc<ExecServerHandler> {
+    let (outgoing_tx, _outgoing_rx) = mpsc::channel(16);
+    let handler = Arc::new(ExecServerHandler::new(RpcNotificationSender::new(
+        outgoing_tx,
+    )));
+    assert_eq!(
+        handler.initialize().expect("initialize"),
+        InitializeResponse {
+            protocol_version: PROTOCOL_VERSION.to_string(),
+        }
+    );
+    handler.initialized().expect("initialized");
+    handler
+}
+
+#[tokio::test]
+async fn duplicate_process_ids_allow_only_one_successful_start() {
+    let handler = initialized_handler().await;
+    let first_handler = Arc::clone(&handler);
+    let second_handler = Arc::clone(&handler);
+
+    let (first, second) = tokio::join!(
+        first_handler.exec(exec_params("proc-1")),
+        second_handler.exec(exec_params("proc-1")),
+    );
+
+    let (successes, failures): (Vec<_>, Vec<_>) =
+        [first, second].into_iter().partition(Result::is_ok);
+    assert_eq!(successes.len(), 1);
+    assert_eq!(failures.len(), 1);
+
+    let error = failures
+        .into_iter()
+        .next()
+        .expect("one failed request")
+        .expect_err("expected duplicate process error");
+    assert_eq!(error.code, -32600);
+    assert_eq!(error.message, "process proc-1 already exists");
+
+    tokio::time::sleep(Duration::from_millis(150)).await;
+    handler.shutdown().await;
+}

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

@@ -0,0 +1,106 @@
+use std::sync::Arc;
+
+use tokio::sync::mpsc;
+use tracing::debug;
+use tracing::warn;
+
+use crate::connection::CHANNEL_CAPACITY;
+use crate::connection::JsonRpcConnection;
+use crate::connection::JsonRpcConnectionEvent;
+use crate::rpc::RpcNotificationSender;
+use crate::rpc::RpcServerOutboundMessage;
+use crate::rpc::encode_server_message;
+use crate::rpc::method_not_found;
+use crate::server::ExecServerHandler;
+use crate::server::registry::build_router;
+
+pub(crate) async fn run_connection(connection: JsonRpcConnection) {
+    let router = Arc::new(build_router());
+    let (json_outgoing_tx, mut incoming_rx, _connection_tasks) = connection.into_parts();
+    let (outgoing_tx, mut outgoing_rx) =
+        mpsc::channel::<RpcServerOutboundMessage>(CHANNEL_CAPACITY);
+    let notifications = RpcNotificationSender::new(outgoing_tx.clone());
+    let handler = Arc::new(ExecServerHandler::new(notifications));
+
+    let outbound_task = tokio::spawn(async move {
+        while let Some(message) = outgoing_rx.recv().await {
+            let json_message = match encode_server_message(message) {
+                Ok(json_message) => json_message,
+                Err(err) => {
+                    warn!("failed to serialize exec-server outbound message: {err}");
+                    break;
+                }
+            };
+            if json_outgoing_tx.send(json_message).await.is_err() {
+                break;
+            }
+        }
+    });
+
+    while let Some(event) = incoming_rx.recv().await {
+        match event {
+            JsonRpcConnectionEvent::Message(message) => match message {
+                codex_app_server_protocol::JSONRPCMessage::Request(request) => {
+                    if let Some(route) = router.request_route(request.method.as_str()) {
+                        let route = route(handler.clone(), request);
+                        let outgoing_tx = outgoing_tx.clone();
+                        tokio::spawn(async move {
+                            let message = route.await;
+                            let _ = outgoing_tx.send(message).await;
+                        });
+                    } else if outgoing_tx
+                        .send(RpcServerOutboundMessage::Error {
+                            request_id: request.id,
+                            error: method_not_found(format!(
+                                "exec-server stub does not implement `{}` yet",
+                                request.method
+                            )),
+                        })
+                        .await
+                        .is_err()
+                    {
+                        break;
+                    }
+                }
+                codex_app_server_protocol::JSONRPCMessage::Notification(notification) => {
+                    let Some(route) = router.notification_route(notification.method.as_str())
+                    else {
+                        warn!(
+                            "closing exec-server connection after unexpected notification: {}",
+                            notification.method
+                        );
+                        break;
+                    };
+                    if let Err(err) = route(handler.clone(), notification).await {
+                        warn!("closing exec-server connection after protocol error: {err}");
+                        break;
+                    }
+                }
+                codex_app_server_protocol::JSONRPCMessage::Response(response) => {
+                    warn!(
+                        "closing exec-server connection after unexpected client response: {:?}",
+                        response.id
+                    );
+                    break;
+                }
+                codex_app_server_protocol::JSONRPCMessage::Error(error) => {
+                    warn!(
+                        "closing exec-server connection after unexpected client error: {:?}",
+                        error.id
+                    );
+                    break;
+                }
+            },
+            JsonRpcConnectionEvent::Disconnected { reason } => {
+                if let Some(reason) = reason {
+                    debug!("exec-server connection disconnected: {reason}");
+                }
+                break;
+            }
+        }
+    }
+
+    handler.shutdown().await;
+    drop(outgoing_tx);
+    let _ = outbound_task.await;
+}

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

@@ -0,0 +1,52 @@
+use std::sync::Arc;
+
+use crate::protocol::EXEC_METHOD;
+use crate::protocol::EXEC_READ_METHOD;
+use crate::protocol::EXEC_TERMINATE_METHOD;
+use crate::protocol::EXEC_WRITE_METHOD;
+use crate::protocol::ExecParams;
+use crate::protocol::INITIALIZE_METHOD;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::InitializeParams;
+use crate::protocol::ReadParams;
+use crate::protocol::TerminateParams;
+use crate::protocol::WriteParams;
+use crate::rpc::RpcRouter;
+use crate::server::ExecServerHandler;
+
+pub(crate) fn build_router() -> RpcRouter<ExecServerHandler> {
+    let mut router = RpcRouter::new();
+    router.request(
+        INITIALIZE_METHOD,
+        |handler: Arc<ExecServerHandler>, _params: InitializeParams| async move {
+            handler.initialize()
+        },
+    );
+    router.notification(
+        INITIALIZED_METHOD,
+        |handler: Arc<ExecServerHandler>, (): ()| async move { handler.initialized() },
+    );
+    router.request(
+        EXEC_METHOD,
+        |handler: Arc<ExecServerHandler>, params: ExecParams| async move { handler.exec(params).await },
+    );
+    router.request(
+        EXEC_READ_METHOD,
+        |handler: Arc<ExecServerHandler>, params: ReadParams| async move {
+            handler.exec_read(params).await
+        },
+    );
+    router.request(
+        EXEC_WRITE_METHOD,
+        |handler: Arc<ExecServerHandler>, params: WriteParams| async move {
+            handler.exec_write(params).await
+        },
+    );
+    router.request(
+        EXEC_TERMINATE_METHOD,
+        |handler: Arc<ExecServerHandler>, params: TerminateParams| async move {
+            handler.terminate(params).await
+        },
+    );
+    router
+}

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

@@ -0,0 +1,166 @@
+use std::net::SocketAddr;
+use std::str::FromStr;
+
+use tokio::net::TcpListener;
+use tokio_tungstenite::accept_async;
+use tracing::warn;
+
+use crate::connection::JsonRpcConnection;
+use crate::server::processor::run_connection;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ExecServerTransport {
+    Stdio,
+    WebSocket { bind_address: SocketAddr },
+}
+
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub enum ExecServerTransportParseError {
+    UnsupportedListenUrl(String),
+    InvalidWebSocketListenUrl(String),
+}
+
+impl std::fmt::Display for ExecServerTransportParseError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ExecServerTransportParseError::UnsupportedListenUrl(listen_url) => write!(
+                f,
+                "unsupported --listen URL `{listen_url}`; expected `stdio://` or `ws://IP:PORT`"
+            ),
+            ExecServerTransportParseError::InvalidWebSocketListenUrl(listen_url) => write!(
+                f,
+                "invalid websocket --listen URL `{listen_url}`; expected `ws://IP:PORT`"
+            ),
+        }
+    }
+}
+
+impl std::error::Error for ExecServerTransportParseError {}
+
+impl ExecServerTransport {
+    pub const DEFAULT_LISTEN_URL: &str = "stdio://";
+
+    pub fn from_listen_url(listen_url: &str) -> Result<Self, ExecServerTransportParseError> {
+        if listen_url == Self::DEFAULT_LISTEN_URL {
+            return Ok(Self::Stdio);
+        }
+
+        if let Some(socket_addr) = listen_url.strip_prefix("ws://") {
+            let bind_address = socket_addr.parse::<SocketAddr>().map_err(|_| {
+                ExecServerTransportParseError::InvalidWebSocketListenUrl(listen_url.to_string())
+            })?;
+            return Ok(Self::WebSocket { bind_address });
+        }
+
+        Err(ExecServerTransportParseError::UnsupportedListenUrl(
+            listen_url.to_string(),
+        ))
+    }
+}
+
+impl FromStr for ExecServerTransport {
+    type Err = ExecServerTransportParseError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Self::from_listen_url(s)
+    }
+}
+
+pub(crate) async fn run_transport(
+    transport: ExecServerTransport,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    match transport {
+        ExecServerTransport::Stdio => {
+            run_connection(JsonRpcConnection::from_stdio(
+                tokio::io::stdin(),
+                tokio::io::stdout(),
+                "exec-server stdio".to_string(),
+            ))
+            .await;
+            Ok(())
+        }
+        ExecServerTransport::WebSocket { bind_address } => {
+            run_websocket_listener(bind_address).await
+        }
+    }
+}
+
+async fn run_websocket_listener(
+    bind_address: SocketAddr,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let listener = TcpListener::bind(bind_address).await?;
+    let local_addr = listener.local_addr()?;
+    print_websocket_startup_banner(local_addr);
+
+    loop {
+        let (stream, peer_addr) = listener.accept().await?;
+        tokio::spawn(async move {
+            match accept_async(stream).await {
+                Ok(websocket) => {
+                    run_connection(JsonRpcConnection::from_websocket(
+                        websocket,
+                        format!("exec-server websocket {peer_addr}"),
+                    ))
+                    .await;
+                }
+                Err(err) => {
+                    warn!(
+                        "failed to accept exec-server websocket connection from {peer_addr}: {err}"
+                    );
+                }
+            }
+        });
+    }
+}
+
+#[allow(clippy::print_stderr)]
+fn print_websocket_startup_banner(addr: SocketAddr) {
+    eprintln!("codex-exec-server listening on ws://{addr}");
+}
+
+#[cfg(test)]
+mod tests {
+    use pretty_assertions::assert_eq;
+
+    use super::ExecServerTransport;
+
+    #[test]
+    fn exec_server_transport_parses_stdio_listen_url() {
+        let transport =
+            ExecServerTransport::from_listen_url(ExecServerTransport::DEFAULT_LISTEN_URL)
+                .expect("stdio listen URL should parse");
+        assert_eq!(transport, ExecServerTransport::Stdio);
+    }
+
+    #[test]
+    fn exec_server_transport_parses_websocket_listen_url() {
+        let transport = ExecServerTransport::from_listen_url("ws://127.0.0.1:1234")
+            .expect("websocket listen URL should parse");
+        assert_eq!(
+            transport,
+            ExecServerTransport::WebSocket {
+                bind_address: "127.0.0.1:1234".parse().expect("valid socket address"),
+            }
+        );
+    }
+
+    #[test]
+    fn exec_server_transport_rejects_invalid_websocket_listen_url() {
+        let err = ExecServerTransport::from_listen_url("ws://localhost:1234")
+            .expect_err("hostname bind address should be rejected");
+        assert_eq!(
+            err.to_string(),
+            "invalid websocket --listen URL `ws://localhost:1234`; expected `ws://IP:PORT`"
+        );
+    }
+
+    #[test]
+    fn exec_server_transport_rejects_unsupported_listen_url() {
+        let err = ExecServerTransport::from_listen_url("http://127.0.0.1:1234")
+            .expect_err("unsupported scheme should fail");
+        assert_eq!(
+            err.to_string(),
+            "unsupported --listen URL `http://127.0.0.1:1234`; expected `stdio://` or `ws://IP:PORT`"
+        );
+    }
+}

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

@@ -0,0 +1,235 @@
+#![cfg(unix)]
+
+use std::process::Stdio;
+use std::time::Duration;
+
+use anyhow::Context;
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use codex_exec_server::ExecOutputStream;
+use codex_exec_server::ExecParams;
+use codex_exec_server::ExecServerClient;
+use codex_exec_server::ExecServerClientConnectOptions;
+use codex_exec_server::ExecServerEvent;
+use codex_exec_server::ExecServerLaunchCommand;
+use codex_exec_server::InitializeParams;
+use codex_exec_server::InitializeResponse;
+use codex_exec_server::RemoteExecServerConnectArgs;
+use codex_exec_server::spawn_local_exec_server;
+use codex_utils_cargo_bin::cargo_bin;
+use pretty_assertions::assert_eq;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::process::Command;
+use tokio::sync::broadcast;
+use tokio::time::timeout;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_accepts_initialize_over_stdio() -> anyhow::Result<()> {
+    let binary = cargo_bin("codex-exec-server")?;
+    let mut child = Command::new(binary);
+    child.stdin(Stdio::piped());
+    child.stdout(Stdio::piped());
+    child.stderr(Stdio::inherit());
+    let mut child = child.spawn()?;
+
+    let mut stdin = child.stdin.take().expect("stdin");
+    let stdout = child.stdout.take().expect("stdout");
+    let mut stdout = BufReader::new(stdout).lines();
+
+    let initialize = JSONRPCMessage::Request(JSONRPCRequest {
+        id: RequestId::Integer(1),
+        method: "initialize".to_string(),
+        params: Some(serde_json::to_value(InitializeParams {
+            client_name: "exec-server-test".to_string(),
+        })?),
+        trace: None,
+    });
+    stdin
+        .write_all(format!("{}\n", serde_json::to_string(&initialize)?).as_bytes())
+        .await?;
+
+    let response_line = timeout(Duration::from_secs(5), stdout.next_line()).await??;
+    let response_line = response_line.expect("response line");
+    let response: JSONRPCMessage = serde_json::from_str(&response_line)?;
+    let JSONRPCMessage::Response(JSONRPCResponse { id, result }) = response else {
+        panic!("expected initialize response");
+    };
+    assert_eq!(id, RequestId::Integer(1));
+    let initialize_response: InitializeResponse = serde_json::from_value(result)?;
+    assert_eq!(initialize_response.protocol_version, "exec-server.v0");
+
+    let initialized = JSONRPCMessage::Notification(JSONRPCNotification {
+        method: "initialized".to_string(),
+        params: Some(serde_json::json!({})),
+    });
+    stdin
+        .write_all(format!("{}\n", serde_json::to_string(&initialized)?).as_bytes())
+        .await?;
+
+    child.start_kill()?;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_client_streams_output_and_accepts_writes() -> anyhow::Result<()> {
+    let mut env = std::collections::HashMap::new();
+    if let Some(path) = std::env::var_os("PATH") {
+        env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
+    }
+
+    let server = spawn_local_exec_server(
+        ExecServerLaunchCommand {
+            program: cargo_bin("codex-exec-server")?,
+            args: Vec::new(),
+        },
+        ExecServerClientConnectOptions {
+            client_name: "exec-server-test".to_string(),
+            initialize_timeout: Duration::from_secs(5),
+        },
+    )
+    .await?;
+
+    let client = server.client();
+    let mut events = client.event_receiver();
+    let response = client
+        .exec(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec![
+                "bash".to_string(),
+                "-lc".to_string(),
+                "printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"
+                    .to_string(),
+            ],
+            cwd: std::env::current_dir()?,
+            env,
+            tty: true,
+            arg0: None,
+        })
+        .await?;
+    let process_id = response.process_id;
+
+    let (stream, ready_output) = recv_until_contains(&mut events, &process_id, "ready").await?;
+    assert_eq!(stream, ExecOutputStream::Pty);
+    assert!(
+        ready_output.contains("ready"),
+        "expected initial ready output"
+    );
+
+    client.write(&process_id, b"hello\n".to_vec()).await?;
+
+    let (stream, echoed_output) =
+        recv_until_contains(&mut events, &process_id, "echo:hello").await?;
+    assert_eq!(stream, ExecOutputStream::Pty);
+    assert!(
+        echoed_output.contains("echo:hello"),
+        "expected echoed output"
+    );
+
+    client.terminate(&process_id).await?;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_client_connects_over_websocket() -> anyhow::Result<()> {
+    let mut env = std::collections::HashMap::new();
+    if let Some(path) = std::env::var_os("PATH") {
+        env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
+    }
+
+    let binary = cargo_bin("codex-exec-server")?;
+    let mut child = Command::new(binary);
+    child.args(["--listen", "ws://127.0.0.1:0"]);
+    child.stdin(Stdio::null());
+    child.stdout(Stdio::null());
+    child.stderr(Stdio::piped());
+    let mut child = child.spawn()?;
+    let stderr = child.stderr.take().expect("stderr");
+    let mut stderr_lines = BufReader::new(stderr).lines();
+    let websocket_url = read_websocket_url(&mut stderr_lines).await?;
+
+    let client = ExecServerClient::connect_websocket(RemoteExecServerConnectArgs {
+        websocket_url,
+        client_name: "exec-server-test".to_string(),
+        connect_timeout: Duration::from_secs(5),
+        initialize_timeout: Duration::from_secs(5),
+    })
+    .await?;
+
+    let mut events = client.event_receiver();
+    let response = client
+        .exec(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec![
+                "bash".to_string(),
+                "-lc".to_string(),
+                "printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"
+                    .to_string(),
+            ],
+            cwd: std::env::current_dir()?,
+            env,
+            tty: true,
+            arg0: None,
+        })
+        .await?;
+    let process_id = response.process_id;
+
+    let (stream, ready_output) = recv_until_contains(&mut events, &process_id, "ready").await?;
+    assert_eq!(stream, ExecOutputStream::Pty);
+    assert!(
+        ready_output.contains("ready"),
+        "expected initial ready output"
+    );
+
+    client.write(&process_id, b"hello\n".to_vec()).await?;
+
+    let (stream, echoed_output) =
+        recv_until_contains(&mut events, &process_id, "echo:hello").await?;
+    assert_eq!(stream, ExecOutputStream::Pty);
+    assert!(
+        echoed_output.contains("echo:hello"),
+        "expected echoed output"
+    );
+
+    client.terminate(&process_id).await?;
+    child.start_kill()?;
+    Ok(())
+}
+
+async fn read_websocket_url<R>(lines: &mut tokio::io::Lines<BufReader<R>>) -> anyhow::Result<String>
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    let line = timeout(Duration::from_secs(5), lines.next_line()).await??;
+    let line = line.context("missing websocket startup banner")?;
+    let websocket_url = line
+        .split_whitespace()
+        .find(|part| part.starts_with("ws://"))
+        .context("missing websocket URL in startup banner")?;
+    Ok(websocket_url.to_string())
+}
+
+async fn recv_until_contains(
+    events: &mut broadcast::Receiver<ExecServerEvent>,
+    process_id: &str,
+    needle: &str,
+) -> anyhow::Result<(ExecOutputStream, String)> {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+    let mut collected = String::new();
+    loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        let event = timeout(remaining, events.recv()).await??;
+        if let ExecServerEvent::OutputDelta(output_event) = event
+            && output_event.process_id == process_id
+        {
+            collected.push_str(&String::from_utf8_lossy(&output_event.chunk.into_inner()));
+            if collected.contains(needle) {
+                return Ok((output_event.stream, collected));
+            }
+        }
+    }
+}