Keep first exec-server PR initialize-only

Move process-oriented client helpers and exports into the exec follow-up. Keep the first PR focused on spawning the stub server and running the initialize handshake.\n\nCo-authored-by: Codex <noreply@openai.com>

codex-rs/Cargo.lock

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

codex-rs/Cargo.toml

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

codex-rs/exec-server/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,7 @@
+#[tokio::main]
+async fn main() {
+    if let Err(err) = codex_exec_server::run_main().await {
+        eprintln!("{err}");
+        std::process::exit(1);
+    }
+}

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

@@ -0,0 +1,276 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
+use std::sync::atomic::AtomicI64;
+use std::sync::atomic::Ordering;
+
+use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use serde_json::Value;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::process::Child;
+use tokio::sync::Mutex;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+use tracing::warn;
+
+use crate::protocol::INITIALIZE_METHOD;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::InitializeParams;
+use crate::protocol::InitializeResponse;
+use crate::server_process::ExecServerLaunchCommand;
+use crate::server_process::spawn_stdio_exec_server;
+
+struct Inner {
+    child: StdMutex<Option<Child>>,
+    write_tx: mpsc::UnboundedSender<JSONRPCMessage>,
+    pending: Mutex<HashMap<RequestId, oneshot::Sender<Result<Value, JSONRPCErrorError>>>>,
+    next_request_id: AtomicI64,
+    reader_task: JoinHandle<()>,
+    writer_task: JoinHandle<()>,
+}
+
+impl Drop for Inner {
+    fn drop(&mut self) {
+        self.reader_task.abort();
+        self.writer_task.abort();
+        if let Ok(mut child_guard) = self.child.lock()
+            && let Some(child) = child_guard.as_mut()
+        {
+            let _ = child.start_kill();
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct ExecServerClient {
+    inner: Arc<Inner>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ExecServerError {
+    #[error("failed to spawn exec-server: {0}")]
+    Spawn(#[source] std::io::Error),
+    #[error("exec-server transport closed")]
+    Closed,
+    #[error("failed to serialize or deserialize exec-server JSON: {0}")]
+    Json(#[from] serde_json::Error),
+    #[error("exec-server protocol error: {0}")]
+    Protocol(String),
+    #[error("exec-server rejected request ({code}): {message}")]
+    Server { code: i64, message: String },
+}
+
+impl ExecServerClient {
+    pub async fn spawn(command: ExecServerLaunchCommand) -> Result<Self, ExecServerError> {
+        let crate::server_process::SpawnedStdioExecServer {
+            child,
+            stdin,
+            stdout,
+        } = spawn_stdio_exec_server(command)?;
+
+        let (write_tx, mut write_rx) = mpsc::unbounded_channel::<JSONRPCMessage>();
+        let writer_task = tokio::spawn(async move {
+            let mut stdin = stdin;
+            while let Some(message) = write_rx.recv().await {
+                let encoded = match serde_json::to_vec(&message) {
+                    Ok(encoded) => encoded,
+                    Err(err) => {
+                        warn!("failed to encode exec-server message: {err}");
+                        break;
+                    }
+                };
+                if stdin.write_all(&encoded).await.is_err() {
+                    break;
+                }
+                if stdin.write_all(b"\n").await.is_err() {
+                    break;
+                }
+                if stdin.flush().await.is_err() {
+                    break;
+                }
+            }
+        });
+
+        let pending = Mutex::new(HashMap::<
+            RequestId,
+            oneshot::Sender<Result<Value, JSONRPCErrorError>>,
+        >::new());
+        let inner = Arc::new_cyclic(move |weak| {
+            let weak = weak.clone();
+            let reader_task = tokio::spawn(async move {
+                let mut lines = BufReader::new(stdout).lines();
+                loop {
+                    let Some(inner) = weak.upgrade() else {
+                        break;
+                    };
+                    match lines.next_line().await {
+                        Ok(Some(line)) => {
+                            if line.trim().is_empty() {
+                                continue;
+                            }
+
+                            match serde_json::from_str::<JSONRPCMessage>(&line) {
+                                Ok(message) => {
+                                    if let Err(err) = handle_server_message(&inner, message).await {
+                                        warn!("failed to handle exec-server message: {err}");
+                                        break;
+                                    }
+                                }
+                                Err(err) => {
+                                    warn!("failed to parse exec-server message: {err}");
+                                    break;
+                                }
+                            }
+                        }
+                        Ok(None) => break,
+                        Err(err) => {
+                            warn!("failed to read exec-server stdout: {err}");
+                            break;
+                        }
+                    }
+                }
+
+                if let Some(inner) = weak.upgrade() {
+                    fail_pending_requests(&inner).await;
+                }
+            });
+
+            Inner {
+                child: StdMutex::new(Some(child)),
+                write_tx,
+                pending,
+                next_request_id: AtomicI64::new(1),
+                reader_task,
+                writer_task,
+            }
+        });
+
+        let client = Self { inner };
+        client
+            .initialize(InitializeParams {
+                client_name: "codex-core".to_string(),
+            })
+            .await?;
+        client.send_notification(INITIALIZED_METHOD, serde_json::json!({}))?;
+        Ok(client)
+    }
+
+    pub async fn initialize(
+        &self,
+        params: InitializeParams,
+    ) -> Result<InitializeResponse, ExecServerError> {
+        self.send_request(INITIALIZE_METHOD, params).await
+    }
+
+    async fn send_request<P, R>(&self, method: &str, params: P) -> Result<R, ExecServerError>
+    where
+        P: Serialize,
+        R: DeserializeOwned,
+    {
+        let id = RequestId::Integer(self.inner.next_request_id.fetch_add(1, Ordering::SeqCst));
+        let params = serde_json::to_value(params)?;
+        let (tx, rx) = oneshot::channel();
+        self.inner.pending.lock().await.insert(id.clone(), tx);
+
+        if let Err(err) = self
+            .inner
+            .write_tx
+            .send(JSONRPCMessage::Request(JSONRPCRequest {
+                id: id.clone(),
+                method: method.to_string(),
+                params: Some(params),
+                trace: None,
+            }))
+        {
+            let _ = self.inner.pending.lock().await.remove(&id);
+            return Err(ExecServerError::Protocol(format!(
+                "failed to queue exec-server request: {err}"
+            )));
+        }
+
+        let result = rx.await.map_err(|_| ExecServerError::Closed)??;
+        Ok(serde_json::from_value(result)?)
+    }
+
+    fn send_notification<P>(&self, method: &str, params: P) -> Result<(), ExecServerError>
+    where
+        P: Serialize,
+    {
+        let params = serde_json::to_value(params)?;
+        self.inner
+            .write_tx
+            .send(JSONRPCMessage::Notification(JSONRPCNotification {
+                method: method.to_string(),
+                params: Some(params),
+            }))
+            .map_err(|err| {
+                ExecServerError::Protocol(format!(
+                    "failed to queue exec-server notification: {err}"
+                ))
+            })
+    }
+}
+
+impl From<JSONRPCErrorError> for ExecServerError {
+    fn from(error: JSONRPCErrorError) -> Self {
+        Self::Server {
+            code: error.code,
+            message: error.message,
+        }
+    }
+}
+
+async fn handle_server_message(
+    inner: &Inner,
+    message: JSONRPCMessage,
+) -> Result<(), ExecServerError> {
+    match message {
+        JSONRPCMessage::Response(JSONRPCResponse { id, result }) => {
+            if let Some(tx) = inner.pending.lock().await.remove(&id) {
+                let _ = tx.send(Ok(result));
+            }
+            Ok(())
+        }
+        JSONRPCMessage::Error(JSONRPCError { id, error }) => {
+            if let Some(tx) = inner.pending.lock().await.remove(&id) {
+                let _ = tx.send(Err(error));
+                Ok(())
+            } else {
+                Err(ExecServerError::Server {
+                    code: error.code,
+                    message: error.message,
+                })
+            }
+        }
+        JSONRPCMessage::Notification(notification) => Err(ExecServerError::Protocol(format!(
+            "unexpected exec-server notification: {}",
+            notification.method
+        ))),
+        JSONRPCMessage::Request(request) => Err(ExecServerError::Protocol(format!(
+            "unexpected exec-server request: {}",
+            request.method
+        ))),
+    }
+}
+
+async fn fail_pending_requests(inner: &Inner) {
+    let mut pending = inner.pending.lock().await;
+    for (_, tx) in pending.drain() {
+        let _ = tx.send(Err(JSONRPCErrorError {
+            code: -32000,
+            message: "exec-server transport closed".to_string(),
+            data: None,
+        }));
+    }
+}

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

@@ -0,0 +1,11 @@
+mod client;
+mod protocol;
+mod server;
+mod server_process;
+
+pub use client::ExecServerClient;
+pub use client::ExecServerError;
+pub use protocol::InitializeParams;
+pub use protocol::InitializeResponse;
+pub use server::run_main;
+pub use server_process::ExecServerLaunchCommand;

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

@@ -0,0 +1,18 @@
+use serde::Deserialize;
+use serde::Serialize;
+
+pub const INITIALIZE_METHOD: &str = "initialize";
+pub const INITIALIZED_METHOD: &str = "initialized";
+pub const PROTOCOL_VERSION: &str = "exec-server.v0";
+
+#[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,
+}

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

@@ -0,0 +1,137 @@
+use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+
+use crate::protocol::INITIALIZE_METHOD;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::InitializeResponse;
+use crate::protocol::PROTOCOL_VERSION;
+
+pub async fn run_main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let mut stdin = BufReader::new(tokio::io::stdin()).lines();
+    let mut stdout = tokio::io::stdout();
+
+    while let Some(line) = stdin.next_line().await? {
+        if line.trim().is_empty() {
+            continue;
+        }
+
+        let message = serde_json::from_str::<JSONRPCMessage>(&line)?;
+        match message {
+            JSONRPCMessage::Request(request) => {
+                handle_request(request, &mut stdout).await?;
+            }
+            JSONRPCMessage::Notification(notification) => {
+                if notification.method != INITIALIZED_METHOD {
+                    send_error(
+                        &mut stdout,
+                        RequestId::Integer(-1),
+                        invalid_request(format!(
+                            "unexpected notification method: {}",
+                            notification.method
+                        )),
+                    )
+                    .await?;
+                }
+            }
+            JSONRPCMessage::Response(response) => {
+                send_error(
+                    &mut stdout,
+                    response.id,
+                    invalid_request("unexpected response from client".to_string()),
+                )
+                .await?;
+            }
+            JSONRPCMessage::Error(error) => {
+                send_error(
+                    &mut stdout,
+                    error.id,
+                    invalid_request("unexpected error from client".to_string()),
+                )
+                .await?;
+            }
+        }
+    }
+
+    Ok(())
+}
+
+async fn handle_request(
+    request: JSONRPCRequest,
+    stdout: &mut tokio::io::Stdout,
+) -> Result<(), std::io::Error> {
+    match request.method.as_str() {
+        INITIALIZE_METHOD => {
+            let result = serde_json::to_value(InitializeResponse {
+                protocol_version: PROTOCOL_VERSION.to_string(),
+            })
+            .map_err(std::io::Error::other)?;
+
+            send_response(
+                stdout,
+                JSONRPCResponse {
+                    id: request.id,
+                    result,
+                },
+            )
+            .await
+        }
+        method => {
+            send_error(
+                stdout,
+                request.id,
+                method_not_implemented(format!(
+                    "exec-server stub does not implement `{method}` yet"
+                )),
+            )
+            .await
+        }
+    }
+}
+
+async fn send_response(
+    stdout: &mut tokio::io::Stdout,
+    response: JSONRPCResponse,
+) -> Result<(), std::io::Error> {
+    send_message(stdout, &JSONRPCMessage::Response(response)).await
+}
+
+async fn send_error(
+    stdout: &mut tokio::io::Stdout,
+    id: RequestId,
+    error: JSONRPCErrorError,
+) -> Result<(), std::io::Error> {
+    send_message(stdout, &JSONRPCMessage::Error(JSONRPCError { id, error })).await
+}
+
+async fn send_message(
+    stdout: &mut tokio::io::Stdout,
+    message: &JSONRPCMessage,
+) -> Result<(), std::io::Error> {
+    let encoded = serde_json::to_vec(message).map_err(std::io::Error::other)?;
+    stdout.write_all(&encoded).await?;
+    stdout.write_all(b"\n").await?;
+    stdout.flush().await
+}
+
+fn invalid_request(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32600,
+        message,
+        data: None,
+    }
+}
+
+fn method_not_implemented(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32601,
+        message,
+        data: None,
+    }
+}

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

@@ -0,0 +1,46 @@
+use std::path::PathBuf;
+use std::process::Stdio;
+
+use tokio::process::Child;
+use tokio::process::ChildStdin;
+use tokio::process::ChildStdout;
+use tokio::process::Command;
+
+use crate::client::ExecServerError;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ExecServerLaunchCommand {
+    pub program: PathBuf,
+    pub args: Vec<String>,
+}
+
+pub(crate) struct SpawnedStdioExecServer {
+    pub(crate) child: Child,
+    pub(crate) stdin: ChildStdin,
+    pub(crate) stdout: ChildStdout,
+}
+
+pub(crate) fn spawn_stdio_exec_server(
+    command: ExecServerLaunchCommand,
+) -> Result<SpawnedStdioExecServer, 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())
+    })?;
+
+    Ok(SpawnedStdioExecServer {
+        child,
+        stdin,
+        stdout,
+    })
+}

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

@@ -0,0 +1,127 @@
+#![cfg(unix)]
+
+use std::process::Stdio;
+use std::time::Duration;
+
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use codex_exec_server::InitializeParams;
+use codex_exec_server::InitializeResponse;
+use codex_utils_cargo_bin::cargo_bin;
+use pretty_assertions::assert_eq;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::process::Command;
+use tokio::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_stubs_command_exec_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 _ = timeout(Duration::from_secs(5), stdout.next_line()).await??;
+
+    let exec = JSONRPCMessage::Request(JSONRPCRequest {
+        id: RequestId::Integer(2),
+        method: "command/exec".to_string(),
+        params: Some(serde_json::json!({
+            "processId": "proc-1",
+            "argv": ["true"],
+            "cwd": std::env::current_dir()?,
+            "env": {},
+            "tty": false,
+            "arg0": null
+        })),
+        trace: None,
+    });
+    stdin
+        .write_all(format!("{}\n", serde_json::to_string(&exec)?).as_bytes())
+        .await?;
+
+    let response_line = timeout(Duration::from_secs(5), stdout.next_line()).await??;
+    let response_line = response_line.expect("exec response line");
+    let response: JSONRPCMessage = serde_json::from_str(&response_line)?;
+    let JSONRPCMessage::Error(codex_app_server_protocol::JSONRPCError { id, error }) = response
+    else {
+        panic!("expected command/exec stub error");
+    };
+    assert_eq!(id, RequestId::Integer(2));
+    assert_eq!(error.code, -32601);
+    assert_eq!(
+        error.message,
+        "exec-server stub does not implement `command/exec` yet"
+    );
+
+    child.start_kill()?;
+    Ok(())
+}