[6/6] Fail exec client operations after disconnect (#18027 )

## Summary - Reject new exec-server client operations once the transport has disconnected. - Convert pending RPC calls into closed errors instead of synthetic server errors. - Cover pending read and later write behavior after remote executor disconnect. ## Verification - `just fmt` - `cargo check -p codex-exec-server` ## Stack ```text @ #18027 [6/6] Fail exec client operations after disconnect │ o #18212 [5/6] Wire executor-backed MCP stdio │ o #18087 [4/6] Abstract MCP stdio server launching │ o #18020 [3/6] Add pushed exec process events │ o #18086 [2/6] Support piped stdin in exec process API │ o #18085 [1/6] Add MCP server environment config │ o main ``` --------- Co-authored-by: Codex <noreply@openai.com>
Fix exec inheritance of root shared flags (#18630 )
2026-05-09 22:02:32 +00:00 · 2026-04-20 23:24:06 +00:00 · 2026-04-20 16:12:17 -07:00
11 changed files with 597 additions and 260 deletions
--- a/codex-rs/cli/src/main.rs
+++ b/codex-rs/cli/src/main.rs
@@ -717,6 +717,9 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
                root_remote_auth_token_env.as_deref(),
                "exec",
            )?;
+            exec_cli
+                .shared
+                .inherit_exec_root_options(&interactive.shared);
            prepend_config_flags(
                &mut exec_cli.config_overrides,
                root_config_overrides.clone(),
@@ -1241,6 +1244,7 @@ async fn run_debug_prompt_input_command(
    interactive: TuiCli,
    arg0_paths: Arg0DispatchPaths,
 ) -> anyhow::Result<()> {
+    let shared = interactive.shared.into_inner();
    let mut cli_kv_overrides = root_config_overrides
        .parse_overrides()
        .map_err(anyhow::Error::msg)?;
@@ -1251,38 +1255,38 @@ async fn run_debug_prompt_input_command(
        ));
    }

-    let approval_policy = if interactive.full_auto {
+    let approval_policy = if shared.full_auto {
        Some(AskForApproval::OnRequest)
-    } else if interactive.dangerously_bypass_approvals_and_sandbox {
+    } else if shared.dangerously_bypass_approvals_and_sandbox {
        Some(AskForApproval::Never)
    } else {
        interactive.approval_policy.map(Into::into)
    };
-    let sandbox_mode = if interactive.full_auto {
+    let sandbox_mode = if shared.full_auto {
        Some(codex_protocol::config_types::SandboxMode::WorkspaceWrite)
-    } else if interactive.dangerously_bypass_approvals_and_sandbox {
+    } else if shared.dangerously_bypass_approvals_and_sandbox {
        Some(codex_protocol::config_types::SandboxMode::DangerFullAccess)
    } else {
-        interactive.sandbox_mode.map(Into::into)
+        shared.sandbox_mode.map(Into::into)
    };
    let overrides = ConfigOverrides {
-        model: interactive.model,
-        config_profile: interactive.config_profile,
+        model: shared.model,
+        config_profile: shared.config_profile,
        approval_policy,
        sandbox_mode,
-        cwd: interactive.cwd,
+        cwd: shared.cwd,
        codex_self_exe: arg0_paths.codex_self_exe,
        codex_linux_sandbox_exe: arg0_paths.codex_linux_sandbox_exe,
        main_execve_wrapper_exe: arg0_paths.main_execve_wrapper_exe,
-        show_raw_agent_reasoning: interactive.oss.then_some(true),
+        show_raw_agent_reasoning: shared.oss.then_some(true),
        ephemeral: Some(true),
-        additional_writable_roots: interactive.add_dir,
+        additional_writable_roots: shared.add_dir,
        ..Default::default()
    };
    let config =
        Config::load_with_cli_overrides_and_harness_overrides(cli_kv_overrides, overrides).await?;

-    let mut input = interactive
+    let mut input = shared
        .images
        .into_iter()
        .chain(cmd.images)
@@ -1553,40 +1557,24 @@ fn finalize_fork_interactive(
 /// root-level flags. Only overrides fields explicitly set on the subcommand-scoped
 /// CLI. Also appends `-c key=value` overrides with highest precedence.
 fn merge_interactive_cli_flags(interactive: &mut TuiCli, subcommand_cli: TuiCli) {
-    if let Some(model) = subcommand_cli.model {
-        interactive.model = Some(model);
-    }
-    if subcommand_cli.oss {
-        interactive.oss = true;
-    }
-    if let Some(profile) = subcommand_cli.config_profile {
-        interactive.config_profile = Some(profile);
-    }
-    if let Some(sandbox) = subcommand_cli.sandbox_mode {
-        interactive.sandbox_mode = Some(sandbox);
-    }
-    if let Some(approval) = subcommand_cli.approval_policy {
+    let TuiCli {
+        shared,
+        approval_policy,
+        web_search,
+        prompt,
+        config_overrides,
+        ..
+    } = subcommand_cli;
+    interactive
+        .shared
+        .apply_subcommand_overrides(shared.into_inner());
+    if let Some(approval) = approval_policy {
        interactive.approval_policy = Some(approval);
    }
-    if subcommand_cli.full_auto {
-        interactive.full_auto = true;
-    }
-    if subcommand_cli.dangerously_bypass_approvals_and_sandbox {
-        interactive.dangerously_bypass_approvals_and_sandbox = true;
-    }
-    if let Some(cwd) = subcommand_cli.cwd {
-        interactive.cwd = Some(cwd);
-    }
-    if subcommand_cli.web_search {
+    if web_search {
        interactive.web_search = true;
    }
-    if !subcommand_cli.images.is_empty() {
-        interactive.images = subcommand_cli.images;
-    }
-    if !subcommand_cli.add_dir.is_empty() {
-        interactive.add_dir.extend(subcommand_cli.add_dir);
-    }
-    if let Some(prompt) = subcommand_cli.prompt {
+    if let Some(prompt) = prompt {
        // Normalize CRLF/CR to LF so CLI-provided text can't leak `\r` into TUI state.
        interactive.prompt = Some(prompt.replace("\r\n", "\n").replace('\r', "\n"));
    }
@@ -1594,7 +1582,7 @@ fn merge_interactive_cli_flags(interactive: &mut TuiCli, subcommand_cli: TuiCli)
    interactive
        .config_overrides
        .raw_overrides
-        .extend(subcommand_cli.config_overrides.raw_overrides);
+        .extend(config_overrides.raw_overrides);
 }

 fn print_completion(cmd: CompletionCommand) {
@@ -1714,6 +1702,19 @@ mod tests {
        assert_eq!(args.prompt.as_deref(), Some("re-review"));
    }

+    #[test]
+    fn dangerous_bypass_conflicts_with_approval_policy() {
+        let err = MultitoolCli::try_parse_from([
+            "codex",
+            "--dangerously-bypass-approvals-and-sandbox",
+            "--ask-for-approval",
+            "on-request",
+        ])
+        .expect_err("conflicting permission flags should be rejected");
+
+        assert_eq!(err.kind(), clap::error::ErrorKind::ArgumentConflict);
+    }
+
    fn app_server_from_args(args: &[&str]) -> AppServerCommand {
        let cli = MultitoolCli::try_parse_from(args).expect("parse");
        let Subcommand::AppServer(app_server) = cli.subcommand.expect("app-server present") else {
--- a/codex-rs/exec-server/src/client.rs
+++ b/codex-rs/exec-server/src/client.rs
@@ -2,6 +2,7 @@ use std::collections::BTreeMap;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::Mutex as StdMutex;
+use std::sync::OnceLock;
 use std::time::Duration;

 use arc_swap::ArcSwap;
@@ -140,6 +141,10 @@ struct Inner {
    // need serialization so concurrent register/remove operations do not
    // overwrite each other's copy-on-write updates.
    sessions_write_lock: Mutex<()>,
+    // Once the transport closes, every executor operation should fail quickly
+    // with the same canonical message. This client never reconnects, so the
+    // latch only moves from unset to set once.
+    disconnected: OnceLock<String>,
    session_id: std::sync::RwLock<Option<String>>,
    reader_task: tokio::task::JoinHandle<()>,
 }
@@ -171,6 +176,8 @@ pub enum ExecServerError {
    InitializeTimedOut { timeout: Duration },
    #[error("exec-server transport closed")]
    Closed,
+    #[error("{0}")]
+    Disconnected(String),
    #[error("failed to serialize or deserialize exec-server JSON: {0}")]
    Json(#[from] serde_json::Error),
    #[error("exec-server protocol error: {0}")]
@@ -246,19 +253,11 @@ impl ExecServerClient {
    }

    pub async fn exec(&self, params: ExecParams) -> Result<ExecResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(EXEC_METHOD, &params)
-            .await
-            .map_err(Into::into)
+        self.call(EXEC_METHOD, &params).await
    }

    pub async fn read(&self, params: ReadParams) -> Result<ReadResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(EXEC_READ_METHOD, &params)
-            .await
-            .map_err(Into::into)
+        self.call(EXEC_READ_METHOD, &params).await
    }

    pub async fn write(
@@ -266,107 +265,73 @@ impl ExecServerClient {
        process_id: &ProcessId,
        chunk: Vec<u8>,
    ) -> Result<WriteResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(
-                EXEC_WRITE_METHOD,
-                &WriteParams {
-                    process_id: process_id.clone(),
-                    chunk: chunk.into(),
-                },
-            )
-            .await
-            .map_err(Into::into)
+        self.call(
+            EXEC_WRITE_METHOD,
+            &WriteParams {
+                process_id: process_id.clone(),
+                chunk: chunk.into(),
+            },
+        )
+        .await
    }

    pub async fn terminate(
        &self,
        process_id: &ProcessId,
    ) -> Result<TerminateResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(
-                EXEC_TERMINATE_METHOD,
-                &TerminateParams {
-                    process_id: process_id.clone(),
-                },
-            )
-            .await
-            .map_err(Into::into)
+        self.call(
+            EXEC_TERMINATE_METHOD,
+            &TerminateParams {
+                process_id: process_id.clone(),
+            },
+        )
+        .await
    }

    pub async fn fs_read_file(
        &self,
        params: FsReadFileParams,
    ) -> Result<FsReadFileResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(FS_READ_FILE_METHOD, &params)
-            .await
-            .map_err(Into::into)
+        self.call(FS_READ_FILE_METHOD, &params).await
    }

    pub async fn fs_write_file(
        &self,
        params: FsWriteFileParams,
    ) -> Result<FsWriteFileResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(FS_WRITE_FILE_METHOD, &params)
-            .await
-            .map_err(Into::into)
+        self.call(FS_WRITE_FILE_METHOD, &params).await
    }

    pub async fn fs_create_directory(
        &self,
        params: FsCreateDirectoryParams,
    ) -> Result<FsCreateDirectoryResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(FS_CREATE_DIRECTORY_METHOD, &params)
-            .await
-            .map_err(Into::into)
+        self.call(FS_CREATE_DIRECTORY_METHOD, &params).await
    }

    pub async fn fs_get_metadata(
        &self,
        params: FsGetMetadataParams,
    ) -> Result<FsGetMetadataResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(FS_GET_METADATA_METHOD, &params)
-            .await
-            .map_err(Into::into)
+        self.call(FS_GET_METADATA_METHOD, &params).await
    }

    pub async fn fs_read_directory(
        &self,
        params: FsReadDirectoryParams,
    ) -> Result<FsReadDirectoryResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(FS_READ_DIRECTORY_METHOD, &params)
-            .await
-            .map_err(Into::into)
+        self.call(FS_READ_DIRECTORY_METHOD, &params).await
    }

    pub async fn fs_remove(
        &self,
        params: FsRemoveParams,
    ) -> Result<FsRemoveResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(FS_REMOVE_METHOD, &params)
-            .await
-            .map_err(Into::into)
+        self.call(FS_REMOVE_METHOD, &params).await
    }

    pub async fn fs_copy(&self, params: FsCopyParams) -> Result<FsCopyResponse, ExecServerError> {
-        self.inner
-            .client
-            .call(FS_COPY_METHOD, &params)
-            .await
-            .map_err(Into::into)
+        self.call(FS_COPY_METHOD, &params).await
    }

    pub(crate) async fn register_session(
@@ -411,18 +376,21 @@ impl ExecServerClient {
                                && let Err(err) =
                                    handle_server_notification(&inner, notification).await
                            {
-                                fail_all_sessions(
+                                let message = record_disconnected(
                                    &inner,
                                    format!("exec-server notification handling failed: {err}"),
-                                )
-                                .await;
+                                );
+                                fail_all_sessions(&inner, message).await;
                                return;
                            }
                        }
                        RpcClientEvent::Disconnected { reason } => {
                            if let Some(inner) = weak.upgrade() {
-                                fail_all_sessions(&inner, disconnected_message(reason.as_deref()))
-                                    .await;
+                                let message = record_disconnected(
+                                    &inner,
+                                    disconnected_message(reason.as_deref()),
+                                );
+                                fail_all_sessions(&inner, message).await;
                            }
                            return;
                        }
@@ -434,6 +402,7 @@ impl ExecServerClient {
                client: rpc_client,
                sessions: ArcSwap::from_pointee(HashMap::new()),
                sessions_write_lock: Mutex::new(()),
+                disconnected: OnceLock::new(),
                session_id: std::sync::RwLock::new(None),
                reader_task,
            }
@@ -451,6 +420,36 @@ impl ExecServerClient {
            .await
            .map_err(ExecServerError::Json)
    }
+
+    async fn call<P, T>(&self, method: &str, params: &P) -> Result<T, ExecServerError>
+    where
+        P: serde::Serialize,
+        T: serde::de::DeserializeOwned,
+    {
+        // Reject new work before allocating a JSON-RPC request id. MCP tool
+        // calls, process writes, and fs operations all pass through here, so
+        // this is the shared low-level failure path after executor disconnect.
+        if let Some(error) = self.inner.disconnected_error() {
+            return Err(error);
+        }
+
+        match self.inner.client.call(method, params).await {
+            Ok(response) => Ok(response),
+            Err(error) => {
+                let error = ExecServerError::from(error);
+                if is_transport_closed_error(&error) {
+                    // A call can race with disconnect after the preflight
+                    // check. Only the reader task drains sessions so queued
+                    // process notifications stay ordered before disconnect.
+                    let message = disconnected_message(/*reason*/ None);
+                    let message = record_disconnected(&self.inner, message);
+                    Err(ExecServerError::Disconnected(message))
+                } else {
+                    Err(error)
+                }
+            }
+        }
+    }
 }

 impl From<RpcCallError> for ExecServerError {
@@ -630,6 +629,20 @@ impl Session {
 }

 impl Inner {
+    fn disconnected_error(&self) -> Option<ExecServerError> {
+        self.disconnected
+            .get()
+            .cloned()
+            .map(ExecServerError::Disconnected)
+    }
+
+    fn set_disconnected(&self, message: String) -> Option<String> {
+        match self.disconnected.set(message.clone()) {
+            Ok(()) => Some(message),
+            Err(_) => None,
+        }
+    }
+
    fn get_session(&self, process_id: &ProcessId) -> Option<Arc<SessionState>> {
        self.sessions.load().get(process_id).cloned()
    }
@@ -640,6 +653,12 @@ impl Inner {
        session: Arc<SessionState>,
    ) -> Result<(), ExecServerError> {
        let _sessions_write_guard = self.sessions_write_lock.lock().await;
+        // Do not register a process session that can never receive executor
+        // notifications. Without this check, remote MCP startup could create a
+        // dead session and wait for process output that will never arrive.
+        if let Some(error) = self.disconnected_error() {
+            return Err(error);
+        }
        let sessions = self.sessions.load();
        if sessions.contains_key(process_id) {
            return Err(ExecServerError::Protocol(format!(
@@ -680,20 +699,36 @@ fn disconnected_message(reason: Option<&str>) -> 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"
-        )
+    matches!(
+        error,
+        ExecServerError::Closed | ExecServerError::Disconnected(_)
+    ) || matches!(
+        error,
+        ExecServerError::Server {
+            code: -32000,
+            message,
+        } if message == "JSON-RPC transport closed"
+    )
+}
+
+fn record_disconnected(inner: &Arc<Inner>, message: String) -> String {
+    // The first observer records the canonical disconnect reason. Session
+    // draining stays with the reader task so it can preserve notification
+    // ordering before publishing the terminal failure.
+    if let Some(message) = inner.set_disconnected(message.clone()) {
+        message
+    } else {
+        inner.disconnected.get().cloned().unwrap_or(message)
+    }
 }

 async fn fail_all_sessions(inner: &Arc<Inner>, message: String) {
    let sessions = inner.take_all_sessions().await;

    for (_, session) in sessions {
+        // Sessions synthesize a closed read response and emit a pushed Failed
+        // event. That covers both polling consumers and streaming consumers
+        // such as executor-backed MCP stdio.
        session.set_failure(message.clone()).await;
    }
 }
--- a/codex-rs/exec-server/src/remote_file_system.rs
+++ b/codex-rs/exec-server/src/remote_file_system.rs
@@ -195,9 +195,46 @@ fn map_remote_error(error: ExecServerError) -> io::Error {
            io::Error::new(io::ErrorKind::InvalidInput, message)
        }
        ExecServerError::Server { message, .. } => io::Error::other(message),
-        ExecServerError::Closed => {
+        ExecServerError::Closed | ExecServerError::Disconnected(_) => {
            io::Error::new(io::ErrorKind::BrokenPipe, "exec-server transport closed")
        }
        _ => io::Error::other(error.to_string()),
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use pretty_assertions::assert_eq;
+
+    use super::*;
+
+    #[test]
+    fn transport_errors_map_to_broken_pipe() {
+        let errors = [
+            ExecServerError::Closed,
+            ExecServerError::Disconnected("exec-server transport disconnected".to_string()),
+        ];
+
+        let mapped_errors = errors
+            .into_iter()
+            .map(|error| {
+                let error = map_remote_error(error);
+                (error.kind(), error.to_string())
+            })
+            .collect::<Vec<_>>();
+
+        assert_eq!(
+            mapped_errors,
+            vec![
+                (
+                    io::ErrorKind::BrokenPipe,
+                    "exec-server transport closed".to_string()
+                ),
+                (
+                    io::ErrorKind::BrokenPipe,
+                    "exec-server transport closed".to_string()
+                ),
+            ]
+        );
+    }
+}
--- a/codex-rs/exec-server/src/rpc.rs
+++ b/codex-rs/exec-server/src/rpc.rs
@@ -18,12 +18,23 @@ use serde_json::Value;
 use tokio::sync::Mutex;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
+use tokio::sync::watch;
 use tokio::task::JoinHandle;

 use crate::connection::JsonRpcConnection;
 use crate::connection::JsonRpcConnectionEvent;

-type PendingRequest = oneshot::Sender<Result<Value, JSONRPCErrorError>>;
+#[derive(Debug)]
+pub(crate) enum RpcCallError {
+    /// The underlying JSON-RPC transport closed before this call completed.
+    Closed,
+    /// The response bytes were valid JSON-RPC but not the expected result type.
+    Json(serde_json::Error),
+    /// The executor returned a JSON-RPC error response for this call.
+    Server(JSONRPCErrorError),
+}
+
+type PendingRequest = oneshot::Sender<Result<Value, RpcCallError>>;
 type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
 type RequestRoute<S> =
    Box<dyn Fn(Arc<S>, JSONRPCRequest) -> BoxFuture<RpcServerOutboundMessage> + Send + Sync>;
@@ -172,6 +183,10 @@ where
 pub(crate) struct RpcClient {
    write_tx: mpsc::Sender<JSONRPCMessage>,
    pending: Arc<Mutex<HashMap<RequestId, PendingRequest>>>,
+    // Shared transport state from `JsonRpcConnection`. Calls use this to fail
+    // immediately when the socket closes, even if no JSON-RPC error response
+    // can be delivered for their request id.
+    disconnected_rx: watch::Receiver<bool>,
    next_request_id: AtomicI64,
    transport_tasks: Vec<JoinHandle<()>>,
    reader_task: JoinHandle<()>,
@@ -179,8 +194,7 @@ pub(crate) struct RpcClient {

 impl RpcClient {
    pub(crate) fn new(connection: JsonRpcConnection) -> (Self, mpsc::Receiver<RpcClientEvent>) {
-        let (write_tx, mut incoming_rx, _disconnected_rx, transport_tasks) =
-            connection.into_parts();
+        let (write_tx, mut incoming_rx, disconnected_rx, transport_tasks) = connection.into_parts();
        let pending = Arc::new(Mutex::new(HashMap::<RequestId, PendingRequest>::new()));
        let (event_tx, event_rx) = mpsc::channel(128);

@@ -218,6 +232,7 @@ impl RpcClient {
            Self {
                write_tx,
                pending,
+                disconnected_rx,
                next_request_id: AtomicI64::new(1),
                transport_tasks,
                reader_task,
@@ -253,10 +268,16 @@ impl RpcClient {
    {
        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 mut pending = self.pending.lock().await;
+            // Registering the pending request and checking disconnect must be
+            // atomic with the reader's drain_pending path. Otherwise a call
+            // can sneak in after the drain and wait forever.
+            if *self.disconnected_rx.borrow() {
+                return Err(RpcCallError::Closed);
+            }
+            pending.insert(request_id.clone(), response_tx);
+        }

        let params = match serde_json::to_value(params) {
            Ok(params) => params,
@@ -280,10 +301,17 @@ impl RpcClient {
            return Err(RpcCallError::Closed);
        }

-        let result = response_rx.await.map_err(|_| RpcCallError::Closed)?;
+        // Do not race in-flight requests directly against the transport-close
+        // watch value. The connection reader receives JSON-RPC messages and
+        // the terminal disconnect event on one ordered queue, then drains any
+        // still-pending requests. Awaiting this receiver preserves that order:
+        // responses already read before EOF still win, and truly pending calls
+        // are failed once the reader observes the disconnect.
+        let result: Result<Value, RpcCallError> =
+            response_rx.await.map_err(|_| RpcCallError::Closed)?;
        let response = match result {
            Ok(response) => response,
-            Err(error) => return Err(RpcCallError::Server(error)),
+            Err(error) => return Err(error),
        };
        serde_json::from_value(response).map_err(RpcCallError::Json)
    }
@@ -304,13 +332,6 @@ impl Drop for RpcClient {
    }
 }

-#[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> {
@@ -417,7 +438,7 @@ async fn handle_server_message(
        }
        JSONRPCMessage::Error(JSONRPCError { id, error }) => {
            if let Some(pending) = pending.lock().await.remove(&id) {
-                let _ = pending.send(Err(error));
+                let _ = pending.send(Err(RpcCallError::Server(error)));
            }
        }
        JSONRPCMessage::Notification(notification) => {
@@ -445,11 +466,7 @@ async fn drain_pending(pending: &Mutex<HashMap<RequestId, PendingRequest>>) {
            .collect::<Vec<_>>()
    };
    for pending in pending {
-        let _ = pending.send(Err(JSONRPCErrorError {
-            code: -32000,
-            data: None,
-            message: "JSON-RPC transport closed".to_string(),
-        }));
+        let _ = pending.send(Err(RpcCallError::Closed));
    }
 }

--- a/codex-rs/exec-server/tests/exec_process.rs
+++ b/codex-rs/exec-server/tests/exec_process.rs
@@ -4,6 +4,7 @@ mod common;

 use std::sync::Arc;

+use anyhow::Context;
 use anyhow::Result;
 use codex_exec_server::Environment;
 use codex_exec_server::ExecBackend;
@@ -484,6 +485,16 @@ async fn remote_exec_process_reports_transport_disconnect() -> Result<()> {

    let process = Arc::clone(&session.process);
    let mut events = process.subscribe_events();
+    let process_for_pending_read = Arc::clone(&process);
+    let pending_read = tokio::spawn(async move {
+        process_for_pending_read
+            .read(
+                /*after_seq*/ None,
+                /*max_bytes*/ None,
+                /*wait_ms*/ Some(60_000),
+            )
+            .await
+    });
    let server = context
        .server
        .as_mut()
@@ -499,6 +510,15 @@ async fn remote_exec_process_reports_transport_disconnect() -> Result<()> {
        "unexpected failure event: {event_message}"
    );

+    let pending_response = timeout(Duration::from_secs(2), pending_read).await???;
+    let pending_message = pending_response
+        .failure
+        .expect("pending read should surface disconnect as a failure");
+    assert!(
+        pending_message.starts_with("exec-server transport disconnected"),
+        "unexpected pending failure message: {pending_message}"
+    );
+
    let mut wake_rx = process.subscribe_wake();
    let response = read_process_until_change(process, &mut wake_rx, /*after_seq*/ None).await?;
    let message = response
@@ -513,6 +533,20 @@ async fn remote_exec_process_reports_transport_disconnect() -> Result<()> {
        "disconnect should close the process session"
    );

+    let write_result = timeout(
+        Duration::from_secs(2),
+        session.process.write(b"hello".to_vec()),
+    )
+    .await
+    .context("timed out waiting for write after disconnect")?;
+    let write_error = write_result.expect_err("write after disconnect should fail");
+    assert!(
+        write_error
+            .to_string()
+            .starts_with("exec-server transport disconnected"),
+        "unexpected write error: {write_error}"
+    );
+
    Ok(())
 }

--- a/codex-rs/exec/src/cli.rs
+++ b/codex-rs/exec/src/cli.rs
@@ -3,6 +3,7 @@ use clap::FromArgMatches;
 use clap::Parser;
 use clap::ValueEnum;
 use codex_utils_cli::CliConfigOverrides;
+use codex_utils_cli::SharedCliOptions;
 use std::path::PathBuf;

 #[derive(Parser, Debug)]
@@ -15,65 +16,13 @@ pub struct Cli {
    #[command(subcommand)]
    pub command: Option<Command>,

-    /// Optional image(s) to attach to the initial prompt.
-    #[arg(
-        long = "image",
-        short = 'i',
-        value_name = "FILE",
-        value_delimiter = ',',
-        num_args = 1..
-    )]
-    pub images: Vec<PathBuf>,
-
-    /// Model the agent should use.
-    #[arg(long, short = 'm', global = true)]
-    pub model: Option<String>,
-
-    /// Use open-source provider.
-    #[arg(long = "oss", default_value_t = false)]
-    pub oss: bool,
-
-    /// Specify which local provider to use (lmstudio or ollama).
-    /// If not specified with --oss, will use config default or show selection.
-    #[arg(long = "local-provider")]
-    pub oss_provider: Option<String>,
-
-    /// Select the sandbox policy to use when executing model-generated shell
-    /// commands.
-    #[arg(long = "sandbox", short = 's', value_enum)]
-    pub sandbox_mode: Option<codex_utils_cli::SandboxModeCliArg>,
-
-    /// Configuration profile from config.toml to specify default options.
-    #[arg(long = "profile", short = 'p')]
-    pub config_profile: Option<String>,
-
-    /// Convenience alias for low-friction sandboxed automatic execution (--sandbox workspace-write).
-    #[arg(long = "full-auto", default_value_t = false, global = true)]
-    pub full_auto: bool,
-
-    /// Skip all confirmation prompts and execute commands without sandboxing.
-    /// EXTREMELY DANGEROUS. Intended solely for running in environments that are externally sandboxed.
-    #[arg(
-        long = "dangerously-bypass-approvals-and-sandbox",
-        alias = "yolo",
-        default_value_t = false,
-        global = true,
-        conflicts_with = "full_auto"
-    )]
-    pub dangerously_bypass_approvals_and_sandbox: bool,
-
-    /// Tell the agent to use the specified directory as its working root.
-    #[clap(long = "cd", short = 'C', value_name = "DIR")]
-    pub cwd: Option<PathBuf>,
+    #[clap(flatten)]
+    pub shared: ExecSharedCliOptions,

    /// Allow running Codex outside a Git repository.
    #[arg(long = "skip-git-repo-check", global = true, default_value_t = false)]
    pub skip_git_repo_check: bool,

-    /// Additional directories that should be writable alongside the primary workspace.
-    #[arg(long = "add-dir", value_name = "DIR", value_hint = clap::ValueHint::DirPath)]
-    pub add_dir: Vec<PathBuf>,
-
    /// Run without persisting session files to disk.
    #[arg(long = "ephemeral", global = true, default_value_t = false)]
    pub ephemeral: bool,
@@ -122,6 +71,71 @@ pub struct Cli {
    pub prompt: Option<String>,
 }

+impl std::ops::Deref for Cli {
+    type Target = SharedCliOptions;
+
+    fn deref(&self) -> &Self::Target {
+        &self.shared.0
+    }
+}
+
+impl std::ops::DerefMut for Cli {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.shared.0
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct ExecSharedCliOptions(SharedCliOptions);
+
+impl ExecSharedCliOptions {
+    pub fn into_inner(self) -> SharedCliOptions {
+        self.0
+    }
+}
+
+impl std::ops::Deref for ExecSharedCliOptions {
+    type Target = SharedCliOptions;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl std::ops::DerefMut for ExecSharedCliOptions {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+impl Args for ExecSharedCliOptions {
+    fn augment_args(cmd: clap::Command) -> clap::Command {
+        mark_exec_global_args(SharedCliOptions::augment_args(cmd))
+    }
+
+    fn augment_args_for_update(cmd: clap::Command) -> clap::Command {
+        mark_exec_global_args(SharedCliOptions::augment_args_for_update(cmd))
+    }
+}
+
+impl FromArgMatches for ExecSharedCliOptions {
+    fn from_arg_matches(matches: &clap::ArgMatches) -> Result<Self, clap::Error> {
+        SharedCliOptions::from_arg_matches(matches).map(Self)
+    }
+
+    fn update_from_arg_matches(&mut self, matches: &clap::ArgMatches) -> Result<(), clap::Error> {
+        self.0.update_from_arg_matches(matches)
+    }
+}
+
+fn mark_exec_global_args(cmd: clap::Command) -> clap::Command {
+    cmd.mut_arg("model", |arg| arg.global(true))
+        .mut_arg("full_auto", |arg| arg.global(true))
+        .mut_arg("dangerously_bypass_approvals_and_sandbox", |arg| {
+            arg.global(true)
+        })
+}
+
 #[derive(Debug, clap::Subcommand)]
 pub enum Command {
    /// Resume a previous session by id or pick the most recent with --last.
--- a/codex-rs/exec/src/lib.rs
+++ b/codex-rs/exec/src/lib.rs
@@ -86,6 +86,7 @@ use codex_protocol::protocol::SessionSource;
 use codex_protocol::user_input::UserInput;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_absolute_path::canonicalize_existing_preserving_symlinks;
+use codex_utils_cli::SharedCliOptions;
 use codex_utils_oss::ensure_oss_provider_ready;
 use codex_utils_oss::get_default_model_for_oss_provider;
 use event_processor_with_human_output::EventProcessorWithHumanOutput;
@@ -219,27 +220,31 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result

    let Cli {
        command,
-        images,
-        model: model_cli_arg,
-        oss,
-        oss_provider,
-        config_profile,
-        full_auto,
-        dangerously_bypass_approvals_and_sandbox,
-        cwd,
+        shared,
        skip_git_repo_check,
-        add_dir,
        ephemeral,
        ignore_user_config,
        ignore_rules,
        color,
        last_message_file,
        json: json_mode,
-        sandbox_mode: sandbox_mode_cli_arg,
        prompt,
        output_schema: output_schema_path,
        config_overrides,
    } = cli;
+    let shared = shared.into_inner();
+    let SharedCliOptions {
+        images,
+        model: model_cli_arg,
+        oss,
+        oss_provider,
+        config_profile,
+        sandbox_mode: sandbox_mode_cli_arg,
+        full_auto,
+        dangerously_bypass_approvals_and_sandbox,
+        cwd,
+        add_dir,
+    } = shared;

    let (_stdout_with_ansi, stderr_with_ansi) = match color {
        cli::Color::Always => (true, true),
--- a/codex-rs/tui/src/cli.rs
+++ b/codex-rs/tui/src/cli.rs
@@ -1,8 +1,9 @@
+use clap::Args;
+use clap::FromArgMatches;
 use clap::Parser;
-use clap::ValueHint;
 use codex_utils_cli::ApprovalModeCliArg;
 use codex_utils_cli::CliConfigOverrides;
-use std::path::PathBuf;
+use codex_utils_cli::SharedCliOptions;

 #[derive(Parser, Debug)]
 #[command(version)]
@@ -11,10 +12,6 @@ pub struct Cli {
    #[arg(value_name = "PROMPT", value_hint = clap::ValueHint::Other)]
    pub prompt: Option<String>,

-    /// Optional image(s) to attach to the initial prompt.
-    #[arg(long = "image", short = 'i', value_name = "FILE", value_delimiter = ',', num_args = 1..)]
-    pub images: Vec<PathBuf>,
-
    // Internal controls set by the top-level `codex resume` subcommand.
    // These are not exposed as user flags on the base `codex` command.
    #[clap(skip)]
@@ -53,60 +50,17 @@ pub struct Cli {
    #[clap(skip)]
    pub fork_show_all: bool,

-    /// Model the agent should use.
-    #[arg(long, short = 'm')]
-    pub model: Option<String>,
-
-    /// Convenience flag to select the local open source model provider. Equivalent to -c
-    /// model_provider=oss; verifies a local LM Studio or Ollama server is running.
-    #[arg(long = "oss", default_value_t = false)]
-    pub oss: bool,
-
-    /// Specify which local provider to use (lmstudio or ollama).
-    /// If not specified with --oss, will use config default or show selection.
-    #[arg(long = "local-provider")]
-    pub oss_provider: Option<String>,
-
-    /// Configuration profile from config.toml to specify default options.
-    #[arg(long = "profile", short = 'p')]
-    pub config_profile: Option<String>,
-
-    /// Select the sandbox policy to use when executing model-generated shell
-    /// commands.
-    #[arg(long = "sandbox", short = 's')]
-    pub sandbox_mode: Option<codex_utils_cli::SandboxModeCliArg>,
+    #[clap(flatten)]
+    pub shared: TuiSharedCliOptions,

    /// Configure when the model requires human approval before executing a command.
    #[arg(long = "ask-for-approval", short = 'a')]
    pub approval_policy: Option<ApprovalModeCliArg>,

-    /// Convenience alias for low-friction sandboxed automatic execution (-a on-request, --sandbox workspace-write).
-    #[arg(long = "full-auto", default_value_t = false)]
-    pub full_auto: bool,
-
-    /// Skip all confirmation prompts and execute commands without sandboxing.
-    /// EXTREMELY DANGEROUS. Intended solely for running in environments that are externally sandboxed.
-    #[arg(
-        long = "dangerously-bypass-approvals-and-sandbox",
-        alias = "yolo",
-        default_value_t = false,
-        conflicts_with_all = ["approval_policy", "full_auto"]
-    )]
-    pub dangerously_bypass_approvals_and_sandbox: bool,
-
-    /// Tell the agent to use the specified directory as its working root.
-    /// In remote mode, the path is forwarded to the server and resolved there.
-    #[clap(long = "cd", short = 'C', value_name = "DIR")]
-    pub cwd: Option<PathBuf>,
-
    /// Enable live web search. When enabled, the native Responses `web_search` tool is available to the model (no per‑call approval).
    #[arg(long = "search", default_value_t = false)]
    pub web_search: bool,

-    /// Additional directories that should be writable alongside the primary workspace.
-    #[arg(long = "add-dir", value_name = "DIR", value_hint = ValueHint::DirPath)]
-    pub add_dir: Vec<PathBuf>,
-
    /// Disable alternate screen mode
    ///
    /// Runs the TUI in inline mode, preserving terminal scrollback history. This is useful
@@ -118,3 +72,66 @@ pub struct Cli {
    #[clap(skip)]
    pub config_overrides: CliConfigOverrides,
 }
+
+impl std::ops::Deref for Cli {
+    type Target = SharedCliOptions;
+
+    fn deref(&self) -> &Self::Target {
+        &self.shared.0
+    }
+}
+
+impl std::ops::DerefMut for Cli {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.shared.0
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct TuiSharedCliOptions(SharedCliOptions);
+
+impl TuiSharedCliOptions {
+    pub fn into_inner(self) -> SharedCliOptions {
+        self.0
+    }
+}
+
+impl std::ops::Deref for TuiSharedCliOptions {
+    type Target = SharedCliOptions;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl std::ops::DerefMut for TuiSharedCliOptions {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+impl Args for TuiSharedCliOptions {
+    fn augment_args(cmd: clap::Command) -> clap::Command {
+        mark_tui_args(SharedCliOptions::augment_args(cmd))
+    }
+
+    fn augment_args_for_update(cmd: clap::Command) -> clap::Command {
+        mark_tui_args(SharedCliOptions::augment_args_for_update(cmd))
+    }
+}
+
+impl FromArgMatches for TuiSharedCliOptions {
+    fn from_arg_matches(matches: &clap::ArgMatches) -> Result<Self, clap::Error> {
+        SharedCliOptions::from_arg_matches(matches).map(Self)
+    }
+
+    fn update_from_arg_matches(&mut self, matches: &clap::ArgMatches) -> Result<(), clap::Error> {
+        self.0.update_from_arg_matches(matches)
+    }
+}
+
+fn mark_tui_args(cmd: clap::Command) -> clap::Command {
+    cmd.mut_arg("dangerously_bypass_approvals_and_sandbox", |arg| {
+        arg.conflicts_with("approval_policy")
+    })
+}
--- a/codex-rs/tui/src/lib.rs
+++ b/codex-rs/tui/src/lib.rs
@@ -1387,10 +1387,11 @@ async fn run_ratatui_app(

    let Cli {
        prompt,
-        images,
+        shared,
        no_alt_screen,
        ..
    } = cli;
+    let images = shared.into_inner().images;

    let use_alt_screen = determine_alt_screen_mode(no_alt_screen, config.tui_alternate_screen);
    tui.set_alt_screen_enabled(use_alt_screen);
--- a/codex-rs/utils/cli/src/lib.rs
+++ b/codex-rs/utils/cli/src/lib.rs
@@ -2,8 +2,10 @@ mod approval_mode_cli_arg;
 mod config_override;
 pub(crate) mod format_env_display;
 mod sandbox_mode_cli_arg;
+mod shared_options;

 pub use approval_mode_cli_arg::ApprovalModeCliArg;
 pub use config_override::CliConfigOverrides;
 pub use format_env_display::format_env_display;
 pub use sandbox_mode_cli_arg::SandboxModeCliArg;
+pub use shared_options::SharedCliOptions;
--- a/codex-rs/utils/cli/src/shared_options.rs
+++ b/codex-rs/utils/cli/src/shared_options.rs
@@ -0,0 +1,174 @@
+//! Shared command-line flags used by both interactive and non-interactive Codex entry points.
+
+use crate::SandboxModeCliArg;
+use clap::Args;
+use std::path::PathBuf;
+
+#[derive(Args, Debug, Default)]
+pub struct SharedCliOptions {
+    /// Optional image(s) to attach to the initial prompt.
+    #[arg(
+        long = "image",
+        short = 'i',
+        value_name = "FILE",
+        value_delimiter = ',',
+        num_args = 1..
+    )]
+    pub images: Vec<PathBuf>,
+
+    /// Model the agent should use.
+    #[arg(long, short = 'm')]
+    pub model: Option<String>,
+
+    /// Use open-source provider.
+    #[arg(long = "oss", default_value_t = false)]
+    pub oss: bool,
+
+    /// Specify which local provider to use (lmstudio or ollama).
+    /// If not specified with --oss, will use config default or show selection.
+    #[arg(long = "local-provider")]
+    pub oss_provider: Option<String>,
+
+    /// Configuration profile from config.toml to specify default options.
+    #[arg(long = "profile", short = 'p')]
+    pub config_profile: Option<String>,
+
+    /// Select the sandbox policy to use when executing model-generated shell
+    /// commands.
+    #[arg(long = "sandbox", short = 's')]
+    pub sandbox_mode: Option<SandboxModeCliArg>,
+
+    /// Convenience alias for low-friction sandboxed automatic execution.
+    #[arg(long = "full-auto", default_value_t = false)]
+    pub full_auto: bool,
+
+    /// Skip all confirmation prompts and execute commands without sandboxing.
+    /// EXTREMELY DANGEROUS. Intended solely for running in environments that are externally sandboxed.
+    #[arg(
+        long = "dangerously-bypass-approvals-and-sandbox",
+        alias = "yolo",
+        default_value_t = false,
+        conflicts_with = "full_auto"
+    )]
+    pub dangerously_bypass_approvals_and_sandbox: bool,
+
+    /// Tell the agent to use the specified directory as its working root.
+    #[clap(long = "cd", short = 'C', value_name = "DIR")]
+    pub cwd: Option<PathBuf>,
+
+    /// Additional directories that should be writable alongside the primary workspace.
+    #[arg(long = "add-dir", value_name = "DIR", value_hint = clap::ValueHint::DirPath)]
+    pub add_dir: Vec<PathBuf>,
+}
+
+impl SharedCliOptions {
+    pub fn inherit_exec_root_options(&mut self, root: &Self) {
+        let self_selected_sandbox_mode = self.sandbox_mode.is_some()
+            || self.full_auto
+            || self.dangerously_bypass_approvals_and_sandbox;
+        let Self {
+            images,
+            model,
+            oss,
+            oss_provider,
+            config_profile,
+            sandbox_mode,
+            full_auto,
+            dangerously_bypass_approvals_and_sandbox,
+            cwd,
+            add_dir,
+        } = self;
+        let Self {
+            images: root_images,
+            model: root_model,
+            oss: root_oss,
+            oss_provider: root_oss_provider,
+            config_profile: root_config_profile,
+            sandbox_mode: root_sandbox_mode,
+            full_auto: root_full_auto,
+            dangerously_bypass_approvals_and_sandbox: root_dangerously_bypass_approvals_and_sandbox,
+            cwd: root_cwd,
+            add_dir: root_add_dir,
+        } = root;
+
+        if model.is_none() {
+            model.clone_from(root_model);
+        }
+        if *root_oss {
+            *oss = true;
+        }
+        if oss_provider.is_none() {
+            oss_provider.clone_from(root_oss_provider);
+        }
+        if config_profile.is_none() {
+            config_profile.clone_from(root_config_profile);
+        }
+        if sandbox_mode.is_none() {
+            *sandbox_mode = *root_sandbox_mode;
+        }
+        if !self_selected_sandbox_mode {
+            *full_auto = *root_full_auto;
+            *dangerously_bypass_approvals_and_sandbox =
+                *root_dangerously_bypass_approvals_and_sandbox;
+        }
+        if cwd.is_none() {
+            cwd.clone_from(root_cwd);
+        }
+        if !root_images.is_empty() {
+            let mut merged_images = root_images.clone();
+            merged_images.append(images);
+            *images = merged_images;
+        }
+        if !root_add_dir.is_empty() {
+            let mut merged_add_dir = root_add_dir.clone();
+            merged_add_dir.append(add_dir);
+            *add_dir = merged_add_dir;
+        }
+    }
+
+    pub fn apply_subcommand_overrides(&mut self, subcommand: Self) {
+        let subcommand_selected_sandbox_mode = subcommand.sandbox_mode.is_some()
+            || subcommand.full_auto
+            || subcommand.dangerously_bypass_approvals_and_sandbox;
+        let Self {
+            images,
+            model,
+            oss,
+            oss_provider,
+            config_profile,
+            sandbox_mode,
+            full_auto,
+            dangerously_bypass_approvals_and_sandbox,
+            cwd,
+            add_dir,
+        } = subcommand;
+
+        if let Some(model) = model {
+            self.model = Some(model);
+        }
+        if oss {
+            self.oss = true;
+        }
+        if let Some(oss_provider) = oss_provider {
+            self.oss_provider = Some(oss_provider);
+        }
+        if let Some(config_profile) = config_profile {
+            self.config_profile = Some(config_profile);
+        }
+        if subcommand_selected_sandbox_mode {
+            self.sandbox_mode = sandbox_mode;
+            self.full_auto = full_auto;
+            self.dangerously_bypass_approvals_and_sandbox =
+                dangerously_bypass_approvals_and_sandbox;
+        }
+        if let Some(cwd) = cwd {
+            self.cwd = Some(cwd);
+        }
+        if !images.is_empty() {
+            self.images = images;
+        }
+        if !add_dir.is_empty() {
+            self.add_dir.extend(add_dir);
+        }
+    }
+}