Support piped stdin in exec process API

Add an explicit stdin mode to process/start so non-tty processes can either keep stdin closed or expose a writable pipe.

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

Author: aibrahim-oai

codex-rs/core/src/unified_exec/process_manager.rs

@@ -152,6 +152,7 @@ fn exec_server_params_for_request(
         env_policy,
         env,
         tty,
+        stdin: codex_exec_server::ExecStdinMode::Closed,
         arg0: request.arg0.clone(),
     }
 }

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

@@ -346,6 +346,7 @@ mod tests {
                 env_policy: None,
                 env: Default::default(),
                 tty: false,
+                stdin: crate::ExecStdinMode::Closed,
                 arg0: None,
             })
             .await

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

@@ -48,6 +48,7 @@ pub use protocol::ExecOutputDeltaNotification;
 pub use protocol::ExecOutputStream;
 pub use protocol::ExecParams;
 pub use protocol::ExecResponse;
+pub use protocol::ExecStdinMode;
 pub use protocol::FsCopyParams;
 pub use protocol::FsCopyResponse;
 pub use protocol::FsCreateDirectoryParams;

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

@@ -28,6 +28,7 @@ use crate::protocol::ExecOutputDeltaNotification;
 use crate::protocol::ExecOutputStream;
 use crate::protocol::ExecParams;
 use crate::protocol::ExecResponse;
+use crate::protocol::ExecStdinMode;
 use crate::protocol::ProcessOutputChunk;
 use crate::protocol::ReadParams;
 use crate::protocol::ReadResponse;
@@ -59,6 +60,7 @@ struct RetainedOutputChunk {
 struct RunningProcess {
     session: ExecCommandSession,
     tty: bool,
+    stdin: ExecStdinMode,
     output: VecDeque<RetainedOutputChunk>,
     retained_bytes: usize,
     next_seq: u64,
@@ -165,6 +167,15 @@ impl LocalProcess {
                 TerminalSize::default(),
             )
             .await
+        } else if matches!(params.stdin, ExecStdinMode::Piped) {
+            codex_utils_pty::spawn_pipe_process(
+                program,
+                args,
+                params.cwd.as_path(),
+                &env,
+                &params.arg0,
+            )
+            .await
         } else {
             codex_utils_pty::spawn_pipe_process_no_stdin(
                 program,
@@ -195,6 +206,7 @@ impl LocalProcess {
                 ProcessEntry::Running(Box::new(RunningProcess {
                     session: spawned.session,
                     tty: params.tty,
+                    stdin: params.stdin,
                     output: VecDeque::new(),
                     retained_bytes: 0,
                     next_seq: 1,
@@ -339,7 +351,7 @@ impl LocalProcess {
                     status: WriteStatus::Starting,
                 });
             };
-            if !process.tty {
+            if !process.tty && matches!(process.stdin, ExecStdinMode::Closed) {
                 return Ok(WriteResponse {
                     status: WriteStatus::StdinClosed,
                 });
@@ -667,6 +679,7 @@ mod tests {
             env_policy: None,
             env,
             tty: false,
+            stdin: ExecStdinMode::Closed,
             arg0: None,
         }
     }

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

@@ -69,9 +69,28 @@ pub struct ExecParams {
     pub env_policy: Option<ExecEnvPolicy>,
     pub env: HashMap<String, String>,
     pub tty: bool,
+
+    /// Controls whether the executor keeps a writable stdin pipe for this
+    /// process.
+    ///
+    /// Normal non-interactive exec commands use `Closed` so programs that read
+    /// stdin see EOF immediately. Remote MCP stdio uses `Piped` because rmcp
+    /// must write JSON-RPC request bytes to the child process stdin after the
+    /// process has started.
+    pub stdin: ExecStdinMode,
     pub arg0: Option<String>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum ExecStdinMode {
+    /// Start the process with stdin connected to null/EOF.
+    Closed,
+
+    /// Start the process with stdin open and writable through `process/write`.
+    Piped,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ExecEnvPolicy {

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

@@ -10,6 +10,7 @@ use super::ExecServerHandler;
 use crate::ExecServerRuntimePaths;
 use crate::ProcessId;
 use crate::protocol::ExecParams;
+use crate::protocol::ExecStdinMode;
 use crate::protocol::InitializeParams;
 use crate::protocol::ReadParams;
 use crate::protocol::ReadResponse;
@@ -30,6 +31,7 @@ fn exec_params_with_argv(process_id: &str, argv: Vec<String>) -> ExecParams {
         env_policy: None,
         env: inherited_path_env(),
         tty: false,
+        stdin: ExecStdinMode::Closed,
         arg0: None,
     }
 }

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

@@ -393,6 +393,7 @@ mod tests {
             env_policy: None,
             env,
             tty: false,
+            stdin: crate::protocol::ExecStdinMode::Closed,
             arg0: None,
         }
     }

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

@@ -9,6 +9,7 @@ use codex_exec_server::Environment;
 use codex_exec_server::ExecBackend;
 use codex_exec_server::ExecParams;
 use codex_exec_server::ExecProcess;
+use codex_exec_server::ExecStdinMode;
 use codex_exec_server::ProcessId;
 use codex_exec_server::ReadResponse;
 use codex_exec_server::StartedExecProcess;
@@ -54,6 +55,7 @@ async fn assert_exec_process_starts_and_exits(use_remote: bool) -> Result<()> {
             env_policy: /*env_policy*/ None,
             env: Default::default(),
             tty: false,
+            stdin: ExecStdinMode::Closed,
             arg0: None,
         })
         .await?;
@@ -131,6 +133,7 @@ async fn assert_exec_process_streams_output(use_remote: bool) -> Result<()> {
             env_policy: /*env_policy*/ None,
             env: Default::default(),
             tty: false,
+            stdin: ExecStdinMode::Closed,
             arg0: None,
         })
         .await?;
@@ -160,7 +163,8 @@ async fn assert_exec_process_write_then_read(use_remote: bool) -> Result<()> {
             cwd: std::env::current_dir()?,
             env_policy: /*env_policy*/ None,
             env: Default::default(),
-            tty: true,
+            tty: false,
+            stdin: ExecStdinMode::Piped,
             arg0: None,
         })
         .await?;
@@ -172,10 +176,7 @@ async fn assert_exec_process_write_then_read(use_remote: bool) -> Result<()> {
     let wake_rx = process.subscribe_wake();
     let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;
 
-    assert!(
-        output.contains("from-stdin:hello"),
-        "unexpected output: {output:?}"
-    );
+    assert_eq!(output, "from-stdin:hello\n");
     assert_eq!(exit_code, Some(0));
     assert!(closed);
     Ok(())
@@ -198,6 +199,7 @@ async fn assert_exec_process_preserves_queued_events_before_subscribe(
             env_policy: /*env_policy*/ None,
             env: Default::default(),
             tty: false,
+            stdin: ExecStdinMode::Closed,
             arg0: None,
         })
         .await?;
@@ -231,6 +233,7 @@ async fn remote_exec_process_reports_transport_disconnect() -> Result<()> {
             env_policy: /*env_policy*/ None,
             env: Default::default(),
             tty: false,
+            stdin: ExecStdinMode::Closed,
             arg0: None,
         })
         .await?;