feat: exec-server prep for unified exec (#15691)

This PR partially rebase `unified_exec` on the `exec-server` and adapt
the `exec-server` accordingly.

## What changed in `exec-server`

1. Replaced the old "broadcast-driven; process-global" event model with
process-scoped session events. The goal is to be able to have dedicated
handler for each process.
2. Add to protocol contract to support explicit lifecycle status and
stream ordering:
- `WriteResponse` now returns `WriteStatus` (Accepted, UnknownProcess,
StdinClosed, Starting) instead of a bool.
  - Added seq fields to output/exited notifications.
  - Added terminal process/closed notification.
3. Demultiplexed remote notifications into per-process channels. Same as
for the event sys
4. Local and remote backends now both implement ExecBackend.
5. Local backend wraps internal process ID/operations into per-process
ExecProcess objects.
6. Remote backend registers a session channel before launch and
unregisters on failed launch.

## What changed in `unified_exec`

1. Added unified process-state model and backend-neutral process
wrapper. This will probably disappear in the future, but it makes it
easier to keep the work flowing on both side.
- `UnifiedExecProcess` now handles both local PTY sessions and remote
exec-server processes through a shared `ProcessHandle`.
- Added `ProcessState` to track has_exited, exit_code, and terminal
failure message consistently across backends.
2. Routed write and lifecycle handling through process-level methods.

## Some rationals

1. The change centralizes execution transport in exec-server while
preserving policy and orchestration ownership in core, avoiding
duplicated launch approval logic. This comes from internal discussion.
2. Session-scoped events remove coupling/cross-talk between processes
and make stream ordering and terminal state explicit (seq, closed,
failed).
3. The failure-path surfacing (remote launch failures, write failures,
transport disconnects) makes command tool output and cleanup behavior
deterministic

## Follow-ups:
* Unify the concept of thread ID behind an obfuscated struct
* FD handling
* Full zsh-fork compatibility
* Full network sandboxing compatibility
* Handle ws disconnection

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

@@ -1,6 +1,8 @@
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 
+use arc_swap::ArcSwap;
 use codex_app_server_protocol::FsCopyParams;
 use codex_app_server_protocol::FsCopyResponse;
 use codex_app_server_protocol::FsCreateDirectoryParams;
@@ -17,22 +19,25 @@ use codex_app_server_protocol::FsWriteFileParams;
 use codex_app_server_protocol::FsWriteFileResponse;
 use codex_app_server_protocol::JSONRPCNotification;
 use serde_json::Value;
-use tokio::sync::broadcast;
+use tokio::sync::Mutex;
+use tokio::sync::watch;
+
 use tokio::time::timeout;
 use tokio_tungstenite::connect_async;
 use tracing::debug;
-use tracing::warn;
 
+use crate::ProcessId;
 use crate::client_api::ExecServerClientConnectOptions;
 use crate::client_api::RemoteExecServerConnectArgs;
 use crate::connection::JsonRpcConnection;
-use crate::process::ExecServerEvent;
+use crate::protocol::EXEC_CLOSED_METHOD;
 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::ExecClosedNotification;
 use crate::protocol::ExecExitedNotification;
 use crate::protocol::ExecOutputDeltaNotification;
 use crate::protocol::ExecParams;
@@ -90,9 +95,29 @@ impl RemoteExecServerConnectArgs {
     }
 }
 
+pub(crate) struct SessionState {
+    wake_tx: watch::Sender<u64>,
+    failure: Mutex<Option<String>>,
+}
+
+#[derive(Clone)]
+pub(crate) struct Session {
+    client: ExecServerClient,
+    process_id: ProcessId,
+    state: Arc<SessionState>,
+}
+
 struct Inner {
     client: RpcClient,
-    events_tx: broadcast::Sender<ExecServerEvent>,
+    // The remote transport delivers one shared notification stream for every
+    // process on the connection. Keep a local process_id -> session registry so
+    // we can turn those connection-global notifications into process wakeups
+    // without making notifications the source of truth for output delivery.
+    sessions: ArcSwap<HashMap<String, Arc<SessionState>>>,
+    // ArcSwap makes reads cheap on the hot notification path, but writes still
+    // need serialization so concurrent register/remove operations do not
+    // overwrite each other's copy-on-write updates.
+    sessions_write_lock: Mutex<()>,
     reader_task: tokio::task::JoinHandle<()>,
 }
 
@@ -158,10 +183,6 @@ impl ExecServerClient {
         .await
     }
 
-    pub fn event_receiver(&self) -> broadcast::Receiver<ExecServerEvent> {
-        self.inner.events_tx.subscribe()
-    }
-
     pub async fn initialize(
         &self,
         options: ExecServerClientConnectOptions,
@@ -307,6 +328,25 @@ impl ExecServerClient {
             .map_err(Into::into)
     }
 
+    pub(crate) async fn register_session(
+        &self,
+        process_id: &str,
+    ) -> Result<Session, ExecServerError> {
+        let state = Arc::new(SessionState::new());
+        self.inner
+            .insert_session(process_id, Arc::clone(&state))
+            .await?;
+        Ok(Session {
+            client: self.clone(),
+            process_id: process_id.to_string().into(),
+            state,
+        })
+    }
+
+    pub(crate) async fn unregister_session(&self, process_id: &str) {
+        self.inner.remove_session(process_id).await;
+    }
+
     async fn connect(
         connection: JsonRpcConnection,
         options: ExecServerClientConnectOptions,
@@ -322,13 +362,18 @@ impl ExecServerClient {
                                 && let Err(err) =
                                     handle_server_notification(&inner, notification).await
                             {
-                                warn!("exec-server client closing after protocol error: {err}");
+                                fail_all_sessions(
+                                    &inner,
+                                    format!("exec-server notification handling failed: {err}"),
+                                )
+                                .await;
                                 return;
                             }
                         }
                         RpcClientEvent::Disconnected { reason } => {
-                            if let Some(reason) = reason {
-                                warn!("exec-server client transport disconnected: {reason}");
+                            if let Some(inner) = weak.upgrade() {
+                                fail_all_sessions(&inner, disconnected_message(reason.as_deref()))
+                                    .await;
                             }
                             return;
                         }
@@ -338,7 +383,8 @@ impl ExecServerClient {
 
             Inner {
                 client: rpc_client,
-                events_tx: broadcast::channel(256).0,
+                sessions: ArcSwap::from_pointee(HashMap::new()),
+                sessions_write_lock: Mutex::new(()),
                 reader_task,
             }
         });
@@ -370,6 +416,177 @@ impl From<RpcCallError> for ExecServerError {
     }
 }
 
+impl SessionState {
+    fn new() -> Self {
+        let (wake_tx, _wake_rx) = watch::channel(0);
+        Self {
+            wake_tx,
+            failure: Mutex::new(None),
+        }
+    }
+
+    pub(crate) fn subscribe(&self) -> watch::Receiver<u64> {
+        self.wake_tx.subscribe()
+    }
+
+    fn note_change(&self, seq: u64) {
+        let next = (*self.wake_tx.borrow()).max(seq);
+        let _ = self.wake_tx.send(next);
+    }
+
+    async fn set_failure(&self, message: String) {
+        let mut failure = self.failure.lock().await;
+        if failure.is_none() {
+            *failure = Some(message);
+        }
+        drop(failure);
+        let next = (*self.wake_tx.borrow()).saturating_add(1);
+        let _ = self.wake_tx.send(next);
+    }
+
+    async fn failed_response(&self) -> Option<ReadResponse> {
+        self.failure
+            .lock()
+            .await
+            .clone()
+            .map(|message| self.synthesized_failure(message))
+    }
+
+    fn synthesized_failure(&self, message: String) -> ReadResponse {
+        let next_seq = (*self.wake_tx.borrow()).saturating_add(1);
+        ReadResponse {
+            chunks: Vec::new(),
+            next_seq,
+            exited: true,
+            exit_code: None,
+            closed: true,
+            failure: Some(message),
+        }
+    }
+}
+
+impl Session {
+    pub(crate) fn process_id(&self) -> &ProcessId {
+        &self.process_id
+    }
+
+    pub(crate) fn subscribe_wake(&self) -> watch::Receiver<u64> {
+        self.state.subscribe()
+    }
+
+    pub(crate) async fn read(
+        &self,
+        after_seq: Option<u64>,
+        max_bytes: Option<usize>,
+        wait_ms: Option<u64>,
+    ) -> Result<ReadResponse, ExecServerError> {
+        if let Some(response) = self.state.failed_response().await {
+            return Ok(response);
+        }
+
+        match self
+            .client
+            .read(ReadParams {
+                process_id: self.process_id.to_string(),
+                after_seq,
+                max_bytes,
+                wait_ms,
+            })
+            .await
+        {
+            Ok(response) => Ok(response),
+            Err(err) if is_transport_closed_error(&err) => {
+                let message = disconnected_message(/*reason*/ None);
+                self.state.set_failure(message.clone()).await;
+                Ok(self.state.synthesized_failure(message))
+            }
+            Err(err) => Err(err),
+        }
+    }
+
+    pub(crate) async fn write(&self, chunk: Vec<u8>) -> Result<WriteResponse, ExecServerError> {
+        self.client.write(&self.process_id, chunk).await
+    }
+
+    pub(crate) async fn terminate(&self) -> Result<(), ExecServerError> {
+        self.client.terminate(&self.process_id).await?;
+        Ok(())
+    }
+
+    pub(crate) async fn unregister(&self) {
+        self.client.unregister_session(&self.process_id).await;
+    }
+}
+
+impl Inner {
+    fn get_session(&self, process_id: &str) -> Option<Arc<SessionState>> {
+        self.sessions.load().get(process_id).cloned()
+    }
+
+    async fn insert_session(
+        &self,
+        process_id: &str,
+        session: Arc<SessionState>,
+    ) -> Result<(), ExecServerError> {
+        let _sessions_write_guard = self.sessions_write_lock.lock().await;
+        let sessions = self.sessions.load();
+        if sessions.contains_key(process_id) {
+            return Err(ExecServerError::Protocol(format!(
+                "session already registered for process {process_id}"
+            )));
+        }
+        let mut next_sessions = sessions.as_ref().clone();
+        next_sessions.insert(process_id.to_string(), session);
+        self.sessions.store(Arc::new(next_sessions));
+        Ok(())
+    }
+
+    async fn remove_session(&self, process_id: &str) -> Option<Arc<SessionState>> {
+        let _sessions_write_guard = self.sessions_write_lock.lock().await;
+        let sessions = self.sessions.load();
+        let session = sessions.get(process_id).cloned();
+        session.as_ref()?;
+        let mut next_sessions = sessions.as_ref().clone();
+        next_sessions.remove(process_id);
+        self.sessions.store(Arc::new(next_sessions));
+        session
+    }
+
+    async fn take_all_sessions(&self) -> HashMap<String, Arc<SessionState>> {
+        let _sessions_write_guard = self.sessions_write_lock.lock().await;
+        let sessions = self.sessions.load();
+        let drained_sessions = sessions.as_ref().clone();
+        self.sessions.store(Arc::new(HashMap::new()));
+        drained_sessions
+    }
+}
+
+fn disconnected_message(reason: Option<&str>) -> String {
+    match reason {
+        Some(reason) => format!("exec-server transport disconnected: {reason}"),
+        None => "exec-server transport disconnected".to_string(),
+    }
+}
+
+fn is_transport_closed_error(error: &ExecServerError) -> bool {
+    matches!(error, ExecServerError::Closed)
+        || matches!(
+            error,
+            ExecServerError::Server {
+                code: -32000,
+                message,
+            } if message == "JSON-RPC transport closed"
+        )
+}
+
+async fn fail_all_sessions(inner: &Arc<Inner>, message: String) {
+    let sessions = inner.take_all_sessions().await;
+
+    for (_, session) in sessions {
+        session.set_failure(message.clone()).await;
+    }
+}
+
 async fn handle_server_notification(
     inner: &Arc<Inner>,
     notification: JSONRPCNotification,
@@ -378,12 +595,26 @@ async fn handle_server_notification(
         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));
+            if let Some(session) = inner.get_session(&params.process_id) {
+                session.note_change(params.seq);
+            }
         }
         EXEC_EXITED_METHOD => {
             let params: ExecExitedNotification =
                 serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
-            let _ = inner.events_tx.send(ExecServerEvent::Exited(params));
+            if let Some(session) = inner.get_session(&params.process_id) {
+                session.note_change(params.seq);
+            }
+        }
+        EXEC_CLOSED_METHOD => {
+            let params: ExecClosedNotification =
+                serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
+            // Closed is the terminal lifecycle event for this process, so drop
+            // the routing entry before forwarding it.
+            let session = inner.remove_session(&params.process_id).await;
+            if let Some(session) = session {
+                session.note_change(params.seq);
+            }
         }
         other => {
             debug!("ignoring unknown exec-server notification: {other}");
@@ -391,3 +622,153 @@ async fn handle_server_notification(
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use codex_app_server_protocol::JSONRPCMessage;
+    use codex_app_server_protocol::JSONRPCNotification;
+    use codex_app_server_protocol::JSONRPCResponse;
+    use pretty_assertions::assert_eq;
+    use tokio::io::AsyncBufReadExt;
+    use tokio::io::AsyncWrite;
+    use tokio::io::AsyncWriteExt;
+    use tokio::io::BufReader;
+    use tokio::io::duplex;
+    use tokio::sync::mpsc;
+    use tokio::time::Duration;
+    use tokio::time::timeout;
+
+    use super::ExecServerClient;
+    use super::ExecServerClientConnectOptions;
+    use crate::connection::JsonRpcConnection;
+    use crate::protocol::EXEC_EXITED_METHOD;
+    use crate::protocol::EXEC_OUTPUT_DELTA_METHOD;
+    use crate::protocol::ExecExitedNotification;
+    use crate::protocol::ExecOutputDeltaNotification;
+    use crate::protocol::ExecOutputStream;
+    use crate::protocol::INITIALIZE_METHOD;
+    use crate::protocol::INITIALIZED_METHOD;
+    use crate::protocol::InitializeResponse;
+
+    async fn read_jsonrpc_line<R>(lines: &mut tokio::io::Lines<BufReader<R>>) -> JSONRPCMessage
+    where
+        R: tokio::io::AsyncRead + Unpin,
+    {
+        let line = timeout(Duration::from_secs(1), lines.next_line())
+            .await
+            .expect("json-rpc read should not time out")
+            .expect("json-rpc read should succeed")
+            .expect("json-rpc connection should stay open");
+        serde_json::from_str(&line).expect("json-rpc line should parse")
+    }
+
+    async fn write_jsonrpc_line<W>(writer: &mut W, message: JSONRPCMessage)
+    where
+        W: AsyncWrite + Unpin,
+    {
+        let encoded = serde_json::to_string(&message).expect("json-rpc message should serialize");
+        writer
+            .write_all(format!("{encoded}\n").as_bytes())
+            .await
+            .expect("json-rpc line should write");
+    }
+
+    #[tokio::test]
+    async fn wake_notifications_do_not_block_other_sessions() {
+        let (client_stdin, server_reader) = duplex(1 << 20);
+        let (mut server_writer, client_stdout) = duplex(1 << 20);
+        let (notifications_tx, mut notifications_rx) = mpsc::channel(16);
+        let server = tokio::spawn(async move {
+            let mut lines = BufReader::new(server_reader).lines();
+            let initialize = read_jsonrpc_line(&mut lines).await;
+            let request = match initialize {
+                JSONRPCMessage::Request(request) if request.method == INITIALIZE_METHOD => request,
+                other => panic!("expected initialize request, got {other:?}"),
+            };
+            write_jsonrpc_line(
+                &mut server_writer,
+                JSONRPCMessage::Response(JSONRPCResponse {
+                    id: request.id,
+                    result: serde_json::to_value(InitializeResponse {})
+                        .expect("initialize response should serialize"),
+                }),
+            )
+            .await;
+
+            let initialized = read_jsonrpc_line(&mut lines).await;
+            match initialized {
+                JSONRPCMessage::Notification(notification)
+                    if notification.method == INITIALIZED_METHOD => {}
+                other => panic!("expected initialized notification, got {other:?}"),
+            }
+
+            while let Some(message) = notifications_rx.recv().await {
+                write_jsonrpc_line(&mut server_writer, message).await;
+            }
+        });
+
+        let client = ExecServerClient::connect(
+            JsonRpcConnection::from_stdio(
+                client_stdout,
+                client_stdin,
+                "test-exec-server-client".to_string(),
+            ),
+            ExecServerClientConnectOptions::default(),
+        )
+        .await
+        .expect("client should connect");
+
+        let _noisy_session = client
+            .register_session("noisy")
+            .await
+            .expect("noisy session should register");
+        let quiet_session = client
+            .register_session("quiet")
+            .await
+            .expect("quiet session should register");
+        let mut quiet_wake_rx = quiet_session.subscribe_wake();
+
+        for seq in 0..=4096 {
+            notifications_tx
+                .send(JSONRPCMessage::Notification(JSONRPCNotification {
+                    method: EXEC_OUTPUT_DELTA_METHOD.to_string(),
+                    params: Some(
+                        serde_json::to_value(ExecOutputDeltaNotification {
+                            process_id: "noisy".to_string(),
+                            seq,
+                            stream: ExecOutputStream::Stdout,
+                            chunk: b"x".to_vec().into(),
+                        })
+                        .expect("output notification should serialize"),
+                    ),
+                }))
+                .await
+                .expect("output notification should queue");
+        }
+
+        notifications_tx
+            .send(JSONRPCMessage::Notification(JSONRPCNotification {
+                method: EXEC_EXITED_METHOD.to_string(),
+                params: Some(
+                    serde_json::to_value(ExecExitedNotification {
+                        process_id: "quiet".to_string(),
+                        seq: 1,
+                        exit_code: 17,
+                    })
+                    .expect("exit notification should serialize"),
+                ),
+            }))
+            .await
+            .expect("exit notification should queue");
+
+        timeout(Duration::from_secs(1), quiet_wake_rx.changed())
+            .await
+            .expect("quiet session should receive wake before timeout")
+            .expect("quiet wake channel should stay open");
+        assert_eq!(*quiet_wake_rx.borrow(), 1);
+
+        drop(notifications_tx);
+        drop(client);
+        server.await.expect("server task should finish");
+    }
+}

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

@@ -8,14 +8,14 @@ use crate::RemoteExecServerConnectArgs;
 use crate::file_system::ExecutorFileSystem;
 use crate::local_file_system::LocalFileSystem;
 use crate::local_process::LocalProcess;
-use crate::process::ExecProcess;
+use crate::process::ExecBackend;
 use crate::remote_file_system::RemoteFileSystem;
 use crate::remote_process::RemoteProcess;
 
 pub const CODEX_EXEC_SERVER_URL_ENV_VAR: &str = "CODEX_EXEC_SERVER_URL";
 
 pub trait ExecutorEnvironment: Send + Sync {
-    fn get_executor(&self) -> Arc<dyn ExecProcess>;
+    fn get_exec_backend(&self) -> Arc<dyn ExecBackend>;
 }
 
 #[derive(Debug, Default)]
@@ -56,7 +56,7 @@ impl EnvironmentManager {
 pub struct Environment {
     exec_server_url: Option<String>,
     remote_exec_server_client: Option<ExecServerClient>,
-    executor: Arc<dyn ExecProcess>,
+    exec_backend: Arc<dyn ExecBackend>,
 }
 
 impl Default for Environment {
@@ -72,7 +72,7 @@ impl Default for Environment {
         Self {
             exec_server_url: None,
             remote_exec_server_client: None,
-            executor: Arc::new(local_process),
+            exec_backend: Arc::new(local_process),
         }
     }
 }
@@ -102,24 +102,24 @@ impl Environment {
             None
         };
 
-        let executor: Arc<dyn ExecProcess> = if let Some(client) = remote_exec_server_client.clone()
-        {
-            Arc::new(RemoteProcess::new(client))
-        } else {
-            let local_process = LocalProcess::default();
-            local_process
-                .initialize()
-                .map_err(|err| ExecServerError::Protocol(err.message))?;
-            local_process
-                .initialized()
-                .map_err(ExecServerError::Protocol)?;
-            Arc::new(local_process)
-        };
+        let exec_backend: Arc<dyn ExecBackend> =
+            if let Some(client) = remote_exec_server_client.clone() {
+                Arc::new(RemoteProcess::new(client))
+            } else {
+                let local_process = LocalProcess::default();
+                local_process
+                    .initialize()
+                    .map_err(|err| ExecServerError::Protocol(err.message))?;
+                local_process
+                    .initialized()
+                    .map_err(ExecServerError::Protocol)?;
+                Arc::new(local_process)
+            };
 
         Ok(Self {
             exec_server_url,
             remote_exec_server_client,
-            executor,
+            exec_backend,
         })
     }
 
@@ -127,8 +127,8 @@ impl Environment {
         self.exec_server_url.as_deref()
     }
 
-    pub fn get_executor(&self) -> Arc<dyn ExecProcess> {
-        Arc::clone(&self.executor)
+    pub fn get_exec_backend(&self) -> Arc<dyn ExecBackend> {
+        Arc::clone(&self.exec_backend)
     }
 
     pub fn get_filesystem(&self) -> Arc<dyn ExecutorFileSystem> {
@@ -148,8 +148,8 @@ fn normalize_exec_server_url(exec_server_url: Option<String>) -> Option<String>
 }
 
 impl ExecutorEnvironment for Environment {
-    fn get_executor(&self) -> Arc<dyn ExecProcess> {
-        Arc::clone(&self.executor)
+    fn get_exec_backend(&self) -> Arc<dyn ExecBackend> {
+        Arc::clone(&self.exec_backend)
     }
 }
 
@@ -193,7 +193,7 @@ mod tests {
         let environment = Environment::default();
 
         let response = environment
-            .get_executor()
+            .get_exec_backend()
             .start(crate::ExecParams {
                 process_id: "default-env-proc".to_string(),
                 argv: vec!["true".to_string()],
@@ -205,11 +205,6 @@ mod tests {
             .await
             .expect("start process");
 
-        assert_eq!(
-            response,
-            crate::ExecResponse {
-                process_id: "default-env-proc".to_string(),
-            }
-        );
+        assert_eq!(response.process.process_id().as_str(), "default-env-proc");
     }
 }

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

@@ -41,8 +41,11 @@ pub use file_system::FileMetadata;
 pub use file_system::FileSystemResult;
 pub use file_system::ReadDirectoryEntry;
 pub use file_system::RemoveOptions;
+pub use process::ExecBackend;
 pub use process::ExecProcess;
-pub use process::ExecServerEvent;
+pub use process::ProcessId;
+pub use process::StartedExecProcess;
+pub use protocol::ExecClosedNotification;
 pub use protocol::ExecExitedNotification;
 pub use protocol::ExecOutputDeltaNotification;
 pub use protocol::ExecOutputStream;
@@ -56,6 +59,7 @@ pub use protocol::TerminateParams;
 pub use protocol::TerminateResponse;
 pub use protocol::WriteParams;
 pub use protocol::WriteResponse;
+pub use protocol::WriteStatus;
 pub use server::DEFAULT_LISTEN_URL;
 pub use server::ExecServerListenUrlParseError;
 pub use server::run_main;

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

@@ -11,13 +11,16 @@ use codex_utils_pty::ExecCommandSession;
 use codex_utils_pty::TerminalSize;
 use tokio::sync::Mutex;
 use tokio::sync::Notify;
-use tokio::sync::broadcast;
 use tokio::sync::mpsc;
-use tracing::warn;
+use tokio::sync::watch;
 
+use crate::ExecBackend;
 use crate::ExecProcess;
 use crate::ExecServerError;
-use crate::ExecServerEvent;
+use crate::ProcessId;
+use crate::StartedExecProcess;
+use crate::protocol::EXEC_CLOSED_METHOD;
+use crate::protocol::ExecClosedNotification;
 use crate::protocol::ExecExitedNotification;
 use crate::protocol::ExecOutputDeltaNotification;
 use crate::protocol::ExecOutputStream;
@@ -31,6 +34,7 @@ use crate::protocol::TerminateParams;
 use crate::protocol::TerminateResponse;
 use crate::protocol::WriteParams;
 use crate::protocol::WriteResponse;
+use crate::protocol::WriteStatus;
 use crate::rpc::RpcNotificationSender;
 use crate::rpc::RpcServerOutboundMessage;
 use crate::rpc::internal_error;
@@ -38,7 +42,6 @@ use crate::rpc::invalid_params;
 use crate::rpc::invalid_request;
 
 const RETAINED_OUTPUT_BYTES_PER_PROCESS: usize = 1024 * 1024;
-const EVENT_CHANNEL_CAPACITY: usize = 256;
 const NOTIFICATION_CHANNEL_CAPACITY: usize = 256;
 #[cfg(test)]
 const EXITED_PROCESS_RETENTION: Duration = Duration::from_millis(25);
@@ -59,7 +62,10 @@ struct RunningProcess {
     retained_bytes: usize,
     next_seq: u64,
     exit_code: Option<i32>,
+    wake_tx: watch::Sender<u64>,
     output_notify: Arc<Notify>,
+    open_streams: usize,
+    closed: bool,
 }
 
 enum ProcessEntry {
@@ -69,7 +75,6 @@ enum ProcessEntry {
 
 struct Inner {
     notifications: RpcNotificationSender,
-    events_tx: broadcast::Sender<ExecServerEvent>,
     processes: Mutex<HashMap<String, ProcessEntry>>,
     initialize_requested: AtomicBool,
     initialized: AtomicBool,
@@ -80,6 +85,12 @@ pub(crate) struct LocalProcess {
     inner: Arc<Inner>,
 }
 
+struct LocalExecProcess {
+    process_id: ProcessId,
+    backend: LocalProcess,
+    wake_tx: watch::Sender<u64>,
+}
+
 impl Default for LocalProcess {
     fn default() -> Self {
         let (outgoing_tx, mut outgoing_rx) =
@@ -94,7 +105,6 @@ impl LocalProcess {
         Self {
             inner: Arc::new(Inner {
                 notifications,
-                events_tx: broadcast::channel(EVENT_CHANNEL_CAPACITY).0,
                 processes: Mutex::new(HashMap::new()),
                 initialize_requested: AtomicBool::new(false),
                 initialized: AtomicBool::new(false),
@@ -152,10 +162,12 @@ impl LocalProcess {
         Ok(())
     }
 
-    pub(crate) async fn exec(&self, params: ExecParams) -> Result<ExecResponse, JSONRPCErrorError> {
+    async fn start_process(
+        &self,
+        params: ExecParams,
+    ) -> Result<(ExecResponse, watch::Sender<u64>), JSONRPCErrorError> {
         self.require_initialized_for("exec")?;
         let process_id = params.process_id.clone();
-
         let (program, args) = params
             .argv
             .split_first()
@@ -203,6 +215,7 @@ impl LocalProcess {
         };
 
         let output_notify = Arc::new(Notify::new());
+        let (wake_tx, _wake_rx) = watch::channel(0);
         {
             let mut process_map = self.inner.processes.lock().await;
             process_map.insert(
@@ -214,7 +227,10 @@ impl LocalProcess {
                     retained_bytes: 0,
                     next_seq: 1,
                     exit_code: None,
+                    wake_tx: wake_tx.clone(),
                     output_notify: Arc::clone(&output_notify),
+                    open_streams: 2,
+                    closed: false,
                 })),
             );
         }
@@ -248,7 +264,13 @@ impl LocalProcess {
             output_notify,
         ));
 
-        Ok(ExecResponse { process_id })
+        Ok((ExecResponse { process_id }, wake_tx))
+    }
+
+    pub(crate) async fn exec(&self, params: ExecParams) -> Result<ExecResponse, JSONRPCErrorError> {
+        self.start_process(params)
+            .await
+            .map(|(response, _)| response)
     }
 
     pub(crate) async fn exec_read(
@@ -256,6 +278,7 @@ impl LocalProcess {
         params: ReadParams,
     ) -> Result<ReadResponse, JSONRPCErrorError> {
         self.require_initialized_for("exec")?;
+        let _process_id = params.process_id.clone();
         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));
@@ -300,6 +323,8 @@ impl LocalProcess {
                         next_seq,
                         exited: process.exit_code.is_some(),
                         exit_code: process.exit_code,
+                        closed: process.closed,
+                        failure: None,
                     },
                     Arc::clone(&process.output_notify),
                 )
@@ -309,6 +334,11 @@ impl LocalProcess {
                 || response.exited
                 || tokio::time::Instant::now() >= deadline
             {
+                let _total_bytes: usize = response
+                    .chunks
+                    .iter()
+                    .map(|chunk| chunk.chunk.0.len())
+                    .sum();
                 return Ok(response);
             }
 
@@ -325,22 +355,24 @@ impl LocalProcess {
         params: WriteParams,
     ) -> Result<WriteResponse, JSONRPCErrorError> {
         self.require_initialized_for("exec")?;
+        let _process_id = params.process_id.clone();
+        let _input_bytes = params.chunk.0.len();
         let writer_tx = {
             let process_map = self.inner.processes.lock().await;
-            let process = process_map.get(&params.process_id).ok_or_else(|| {
-                invalid_request(format!("unknown process id {}", params.process_id))
-            })?;
+            let Some(process) = process_map.get(&params.process_id) else {
+                return Ok(WriteResponse {
+                    status: WriteStatus::UnknownProcess,
+                });
+            };
             let ProcessEntry::Running(process) = process else {
-                return Err(invalid_request(format!(
-                    "process id {} is starting",
-                    params.process_id
-                )));
+                return Ok(WriteResponse {
+                    status: WriteStatus::Starting,
+                });
             };
             if !process.tty {
-                return Err(invalid_request(format!(
-                    "stdin is closed for process {}",
-                    params.process_id
-                )));
+                return Ok(WriteResponse {
+                    status: WriteStatus::StdinClosed,
+                });
             }
             process.session.writer_sender()
         };
@@ -350,7 +382,9 @@ impl LocalProcess {
             .await
             .map_err(|_| internal_error("failed to write to process stdin".to_string()))?;
 
-        Ok(WriteResponse { accepted: true })
+        Ok(WriteResponse {
+            status: WriteStatus::Accepted,
+        })
     }
 
     pub(crate) async fn terminate_process(
@@ -358,6 +392,7 @@ impl LocalProcess {
         params: TerminateParams,
     ) -> Result<TerminateResponse, JSONRPCErrorError> {
         self.require_initialized_for("exec")?;
+        let _process_id = params.process_id.clone();
         let running = {
             let process_map = self.inner.processes.lock().await;
             match process_map.get(&params.process_id) {
@@ -377,13 +412,68 @@ impl LocalProcess {
 }
 
 #[async_trait]
-impl ExecProcess for LocalProcess {
-    async fn start(&self, params: ExecParams) -> Result<ExecResponse, ExecServerError> {
-        self.exec(params).await.map_err(map_handler_error)
+impl ExecBackend for LocalProcess {
+    async fn start(&self, params: ExecParams) -> Result<StartedExecProcess, ExecServerError> {
+        let (response, wake_tx) = self
+            .start_process(params)
+            .await
+            .map_err(map_handler_error)?;
+        Ok(StartedExecProcess {
+            process: Arc::new(LocalExecProcess {
+                process_id: response.process_id.into(),
+                backend: self.clone(),
+                wake_tx,
+            }),
+        })
+    }
+}
+
+#[async_trait]
+impl ExecProcess for LocalExecProcess {
+    fn process_id(&self) -> &ProcessId {
+        &self.process_id
     }
 
-    async fn read(&self, params: ReadParams) -> Result<ReadResponse, ExecServerError> {
-        self.exec_read(params).await.map_err(map_handler_error)
+    fn subscribe_wake(&self) -> watch::Receiver<u64> {
+        self.wake_tx.subscribe()
+    }
+
+    async fn read(
+        &self,
+        after_seq: Option<u64>,
+        max_bytes: Option<usize>,
+        wait_ms: Option<u64>,
+    ) -> Result<ReadResponse, ExecServerError> {
+        self.backend
+            .read(&self.process_id, after_seq, max_bytes, wait_ms)
+            .await
+    }
+
+    async fn write(&self, chunk: Vec<u8>) -> Result<WriteResponse, ExecServerError> {
+        self.backend.write(&self.process_id, chunk).await
+    }
+
+    async fn terminate(&self) -> Result<(), ExecServerError> {
+        self.backend.terminate(&self.process_id).await
+    }
+}
+
+impl LocalProcess {
+    async fn read(
+        &self,
+        process_id: &str,
+        after_seq: Option<u64>,
+        max_bytes: Option<usize>,
+        wait_ms: Option<u64>,
+    ) -> Result<ReadResponse, ExecServerError> {
+        self.exec_read(ReadParams {
+            process_id: process_id.to_string(),
+            after_seq,
+            max_bytes,
+            wait_ms,
+        })
+        .await
+        .map_err(map_handler_error)
     }
 
     async fn write(
@@ -399,16 +489,13 @@ impl ExecProcess for LocalProcess {
         .map_err(map_handler_error)
     }
 
-    async fn terminate(&self, process_id: &str) -> Result<TerminateResponse, ExecServerError> {
+    async fn terminate(&self, process_id: &str) -> Result<(), ExecServerError> {
         self.terminate_process(TerminateParams {
             process_id: process_id.to_string(),
         })
         .await
-        .map_err(map_handler_error)
-    }
-
-    fn subscribe_events(&self) -> broadcast::Receiver<ExecServerEvent> {
-        self.inner.events_tx.subscribe()
+        .map_err(map_handler_error)?;
+        Ok(())
     }
 }
 
@@ -427,6 +514,7 @@ async fn stream_output(
     output_notify: Arc<Notify>,
 ) {
     while let Some(chunk) = receiver.recv().await {
+        let _chunk_len = chunk.len();
         let notification = {
             let mut processes = inner.processes.lock().await;
             let Some(entry) = processes.get_mut(&process_id) else {
@@ -448,21 +536,16 @@ async fn stream_output(
                     break;
                 };
                 process.retained_bytes = process.retained_bytes.saturating_sub(evicted.chunk.len());
-                warn!(
-                    "retained output cap exceeded for process {process_id}; dropping oldest output"
-                );
             }
+            let _ = process.wake_tx.send(seq);
             ExecOutputDeltaNotification {
                 process_id: process_id.clone(),
+                seq,
                 stream,
                 chunk: chunk.into(),
             }
         };
         output_notify.notify_waiters();
-        let _ = inner
-            .events_tx
-            .send(ExecServerEvent::OutputDelta(notification.clone()));
-
         if inner
             .notifications
             .notify(crate::protocol::EXEC_OUTPUT_DELTA_METHOD, &notification)
@@ -472,6 +555,8 @@ async fn stream_output(
             break;
         }
     }
+
+    finish_output_stream(process_id, inner).await;
 }
 
 async fn watch_exit(
@@ -481,29 +566,35 @@ async fn watch_exit(
     output_notify: Arc<Notify>,
 ) {
     let exit_code = exit_rx.await.unwrap_or(-1);
-    {
+    let notification = {
         let mut processes = inner.processes.lock().await;
         if let Some(ProcessEntry::Running(process)) = processes.get_mut(&process_id) {
+            let seq = process.next_seq;
+            process.next_seq += 1;
             process.exit_code = Some(exit_code);
+            let _ = process.wake_tx.send(seq);
+            Some(ExecExitedNotification {
+                process_id: process_id.clone(),
+                seq,
+                exit_code,
+            })
+        } else {
+            None
         }
-    }
-    output_notify.notify_waiters();
-    let notification = ExecExitedNotification {
-        process_id: process_id.clone(),
-        exit_code,
     };
-    let _ = inner
-        .events_tx
-        .send(ExecServerEvent::Exited(notification.clone()));
-    if inner
-        .notifications
-        .notify(crate::protocol::EXEC_EXITED_METHOD, &notification)
-        .await
-        .is_err()
+    output_notify.notify_waiters();
+    if let Some(notification) = notification
+        && inner
+            .notifications
+            .notify(crate::protocol::EXEC_EXITED_METHOD, &notification)
+            .await
+            .is_err()
     {
         return;
     }
 
+    maybe_emit_closed(process_id.clone(), Arc::clone(&inner)).await;
+
     tokio::time::sleep(EXITED_PROCESS_RETENTION).await;
     let mut processes = inner.processes.lock().await;
     if matches!(
@@ -513,3 +604,51 @@ async fn watch_exit(
         processes.remove(&process_id);
     }
 }
+
+async fn finish_output_stream(process_id: String, inner: Arc<Inner>) {
+    {
+        let mut processes = inner.processes.lock().await;
+        let Some(ProcessEntry::Running(process)) = processes.get_mut(&process_id) else {
+            return;
+        };
+
+        if process.open_streams > 0 {
+            process.open_streams -= 1;
+        }
+    }
+
+    maybe_emit_closed(process_id, inner).await;
+}
+
+async fn maybe_emit_closed(process_id: String, inner: Arc<Inner>) {
+    let notification = {
+        let mut processes = inner.processes.lock().await;
+        let Some(ProcessEntry::Running(process)) = processes.get_mut(&process_id) else {
+            return;
+        };
+
+        if process.closed || process.open_streams != 0 || process.exit_code.is_none() {
+            return;
+        }
+
+        process.closed = true;
+        let seq = process.next_seq;
+        process.next_seq += 1;
+        let _ = process.wake_tx.send(seq);
+        Some(ExecClosedNotification {
+            process_id: process_id.clone(),
+            seq,
+        })
+    };
+
+    let Some(notification) = notification else {
+        return;
+    };
+
+    if inner
+        .notifications
+        .notify(EXEC_CLOSED_METHOD, &notification)
+        .await
+        .is_err()
+    {}
+}

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

@@ -1,35 +1,77 @@
+use std::fmt;
+use std::ops::Deref;
+use std::sync::Arc;
+
 use async_trait::async_trait;
-use tokio::sync::broadcast;
+use tokio::sync::watch;
 
 use crate::ExecServerError;
-use crate::protocol::ExecExitedNotification;
-use crate::protocol::ExecOutputDeltaNotification;
 use crate::protocol::ExecParams;
-use crate::protocol::ExecResponse;
-use crate::protocol::ReadParams;
 use crate::protocol::ReadResponse;
-use crate::protocol::TerminateResponse;
 use crate::protocol::WriteResponse;
 
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum ExecServerEvent {
-    OutputDelta(ExecOutputDeltaNotification),
-    Exited(ExecExitedNotification),
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct ProcessId(String);
+
+pub struct StartedExecProcess {
+    pub process: Arc<dyn ExecProcess>,
+}
+
+impl ProcessId {
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+
+    pub fn into_inner(self) -> String {
+        self.0
+    }
+}
+
+impl Deref for ProcessId {
+    type Target = str;
+
+    fn deref(&self) -> &Self::Target {
+        self.as_str()
+    }
+}
+
+impl AsRef<str> for ProcessId {
+    fn as_ref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl fmt::Display for ProcessId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+impl From<String> for ProcessId {
+    fn from(value: String) -> Self {
+        Self(value)
+    }
 }
 
 #[async_trait]
 pub trait ExecProcess: Send + Sync {
-    async fn start(&self, params: ExecParams) -> Result<ExecResponse, ExecServerError>;
+    fn process_id(&self) -> &ProcessId;
 
-    async fn read(&self, params: ReadParams) -> Result<ReadResponse, ExecServerError>;
+    fn subscribe_wake(&self) -> watch::Receiver<u64>;
 
-    async fn write(
+    async fn read(
         &self,
-        process_id: &str,
-        chunk: Vec<u8>,
-    ) -> Result<WriteResponse, ExecServerError>;
+        after_seq: Option<u64>,
+        max_bytes: Option<usize>,
+        wait_ms: Option<u64>,
+    ) -> Result<ReadResponse, ExecServerError>;
 
-    async fn terminate(&self, process_id: &str) -> Result<TerminateResponse, ExecServerError>;
+    async fn write(&self, chunk: Vec<u8>) -> Result<WriteResponse, ExecServerError>;
 
-    fn subscribe_events(&self) -> broadcast::Receiver<ExecServerEvent>;
+    async fn terminate(&self) -> Result<(), ExecServerError>;
+}
+
+#[async_trait]
+pub trait ExecBackend: Send + Sync {
+    async fn start(&self, params: ExecParams) -> Result<StartedExecProcess, ExecServerError>;
 }

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

@@ -13,6 +13,7 @@ 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 EXEC_CLOSED_METHOD: &str = "process/closed";
 pub const FS_READ_FILE_METHOD: &str = "fs/readFile";
 pub const FS_WRITE_FILE_METHOD: &str = "fs/writeFile";
 pub const FS_CREATE_DIRECTORY_METHOD: &str = "fs/createDirectory";
@@ -90,6 +91,8 @@ pub struct ReadResponse {
     pub next_seq: u64,
     pub exited: bool,
     pub exit_code: Option<i32>,
+    pub closed: bool,
+    pub failure: Option<String>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -99,10 +102,19 @@ pub struct WriteParams {
     pub chunk: ByteChunk,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum WriteStatus {
+    Accepted,
+    UnknownProcess,
+    StdinClosed,
+    Starting,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct WriteResponse {
-    pub accepted: bool,
+    pub status: WriteStatus,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -129,6 +141,7 @@ pub enum ExecOutputStream {
 #[serde(rename_all = "camelCase")]
 pub struct ExecOutputDeltaNotification {
     pub process_id: String,
+    pub seq: u64,
     pub stream: ExecOutputStream,
     pub chunk: ByteChunk,
 }
@@ -137,9 +150,17 @@ pub struct ExecOutputDeltaNotification {
 #[serde(rename_all = "camelCase")]
 pub struct ExecExitedNotification {
     pub process_id: String,
+    pub seq: u64,
     pub exit_code: i32,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecClosedNotification {
+    pub process_id: String,
+    pub seq: u64,
+}
+
 mod base64_bytes {
     use super::BASE64_STANDARD;
     use base64::Engine as _;

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

@@ -1,16 +1,17 @@
+use std::sync::Arc;
+
 use async_trait::async_trait;
-use tokio::sync::broadcast;
+use tokio::sync::watch;
 use tracing::trace;
 
+use crate::ExecBackend;
 use crate::ExecProcess;
-use crate::ExecServerClient;
 use crate::ExecServerError;
-use crate::ExecServerEvent;
+use crate::StartedExecProcess;
+use crate::client::ExecServerClient;
+use crate::client::Session;
 use crate::protocol::ExecParams;
-use crate::protocol::ExecResponse;
-use crate::protocol::ReadParams;
 use crate::protocol::ReadResponse;
-use crate::protocol::TerminateResponse;
 use crate::protocol::WriteResponse;
 
 #[derive(Clone)]
@@ -18,6 +19,10 @@ pub(crate) struct RemoteProcess {
     client: ExecServerClient,
 }
 
+struct RemoteExecProcess {
+    session: Session,
+}
+
 impl RemoteProcess {
     pub(crate) fn new(client: ExecServerClient) -> Self {
         trace!("remote process new");
@@ -26,33 +31,56 @@ impl RemoteProcess {
 }
 
 #[async_trait]
-impl ExecProcess for RemoteProcess {
-    async fn start(&self, params: ExecParams) -> Result<ExecResponse, ExecServerError> {
-        trace!("remote process start");
-        self.client.exec(params).await
-    }
+impl ExecBackend for RemoteProcess {
+    async fn start(&self, params: ExecParams) -> Result<StartedExecProcess, ExecServerError> {
+        let process_id = params.process_id.clone();
+        let session = self.client.register_session(&process_id).await?;
+        if let Err(err) = self.client.exec(params).await {
+            session.unregister().await;
+            return Err(err);
+        }
 
-    async fn read(&self, params: ReadParams) -> Result<ReadResponse, ExecServerError> {
-        trace!("remote process read");
-        self.client.read(params).await
+        Ok(StartedExecProcess {
+            process: Arc::new(RemoteExecProcess { session }),
+        })
     }
+}
 
-    async fn write(
+#[async_trait]
+impl ExecProcess for RemoteExecProcess {
+    fn process_id(&self) -> &crate::ProcessId {
+        self.session.process_id()
+    }
+
+    fn subscribe_wake(&self) -> watch::Receiver<u64> {
+        self.session.subscribe_wake()
+    }
+
+    async fn read(
         &self,
-        process_id: &str,
-        chunk: Vec<u8>,
-    ) -> Result<WriteResponse, ExecServerError> {
-        trace!("remote process write");
-        self.client.write(process_id, chunk).await
+        after_seq: Option<u64>,
+        max_bytes: Option<usize>,
+        wait_ms: Option<u64>,
+    ) -> Result<ReadResponse, ExecServerError> {
+        self.session.read(after_seq, max_bytes, wait_ms).await
+    }
+
+    async fn write(&self, chunk: Vec<u8>) -> Result<WriteResponse, ExecServerError> {
+        trace!("exec process write");
+        self.session.write(chunk).await
     }
 
-    async fn terminate(&self, process_id: &str) -> Result<TerminateResponse, ExecServerError> {
-        trace!("remote process terminate");
-        self.client.terminate(process_id).await
+    async fn terminate(&self) -> Result<(), ExecServerError> {
+        trace!("exec process terminate");
+        self.session.terminate().await
     }
+}
 
-    fn subscribe_events(&self) -> broadcast::Receiver<ExecServerEvent> {
-        trace!("remote process subscribe_events");
-        self.client.event_receiver()
+impl Drop for RemoteExecProcess {
+    fn drop(&mut self) {
+        let session = self.session.clone();
+        tokio::spawn(async move {
+            session.unregister().await;
+        });
     }
 }

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

@@ -19,7 +19,6 @@ 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;
@@ -192,12 +191,12 @@ impl RpcClient {
                         if let Err(err) =
                             handle_server_message(&pending_for_reader, &event_tx, message).await
                         {
-                            warn!("JSON-RPC client closing after protocol error: {err}");
+                            let _ = err;
                             break;
                         }
                     }
                     JsonRpcConnectionEvent::MalformedMessage { reason } => {
-                        warn!("JSON-RPC client closing after malformed message: {reason}");
+                        let _ = reason;
                         break;
                     }
                     JsonRpcConnectionEvent::Disconnected { reason } => {