chore: close pipe on non-pty processes (#9369)

Closing the STDIN of piped process when starting them to avoid commands
like `rg` to wait for content on STDIN and hangs for ever

codex-rs/core/src/tools/handlers/unified_exec.rs

@@ -202,7 +202,7 @@ impl ToolHandler for UnifiedExecHandler {
                     })
                     .await
                     .map_err(|err| {
-                        FunctionCallError::RespondToModel(format!("write_stdin failed: {err:?}"))
+                        FunctionCallError::RespondToModel(format!("write_stdin failed: {err}"))
                     })?;
 
                 let interaction = TerminalInteractionEvent {

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

@@ -10,6 +10,10 @@ pub(crate) enum UnifiedExecError {
     UnknownProcessId { process_id: String },
     #[error("failed to write to stdin")]
     WriteToStdin,
+    #[error(
+        "stdin is closed for this session; rerun exec_command with tty=true to keep stdin open"
+    )]
+    StdinClosed,
     #[error("missing command line for unified exec request")]
     MissingCommandLine,
     #[error("Command denied by sandbox: {message}")]

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

@@ -136,6 +136,7 @@ struct ProcessEntry {
     call_id: String,
     process_id: String,
     command: Vec<String>,
+    tty: bool,
     last_used: tokio::time::Instant,
 }
 

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

@@ -74,6 +74,7 @@ struct PreparedProcessHandles {
     cancellation_token: CancellationToken,
     command: Vec<String>,
     process_id: String,
+    tty: bool,
 }
 
 impl UnifiedExecProcessManager {
@@ -218,6 +219,7 @@ impl UnifiedExecProcessManager {
                 cwd.clone(),
                 start,
                 process_id,
+                request.tty,
                 Arc::clone(&transcript),
             )
             .await;
@@ -256,10 +258,14 @@ impl UnifiedExecProcessManager {
             cancellation_token,
             command: session_command,
             process_id,
+            tty,
             ..
         } = self.prepare_process_handles(process_id.as_str()).await?;
 
         if !request.input.is_empty() {
+            if !tty {
+                return Err(UnifiedExecError::StdinClosed);
+            }
             Self::send_input(&writer_tx, request.input.as_bytes()).await?;
             // Give the remote process a brief window to react so that we are
             // more likely to capture its output in the poll below.
@@ -380,6 +386,7 @@ impl UnifiedExecProcessManager {
             cancellation_token,
             command: entry.command.clone(),
             process_id: entry.process_id.clone(),
+            tty: entry.tty,
         })
     }
 
@@ -402,6 +409,7 @@ impl UnifiedExecProcessManager {
         cwd: PathBuf,
         started_at: Instant,
         process_id: String,
+        tty: bool,
         transcript: Arc<tokio::sync::Mutex<HeadTailBuffer>>,
     ) {
         let entry = ProcessEntry {
@@ -409,6 +417,7 @@ impl UnifiedExecProcessManager {
             call_id: context.call_id.clone(),
             process_id: process_id.clone(),
             command: command.to_vec(),
+            tty,
             last_used: started_at,
         };
         let number_processes = {
@@ -461,7 +470,7 @@ impl UnifiedExecProcessManager {
             )
             .await
         } else {
-            codex_utils_pty::pipe::spawn_process(
+            codex_utils_pty::pipe::spawn_process_no_stdin(
                 program,
                 args,
                 env.cwd.as_path(),

codex-rs/core/tests/suite/unified_exec.rs

@@ -805,6 +805,7 @@ async fn unified_exec_emits_terminal_interaction_for_write_stdin() -> Result<()>
     let open_args = json!({
         "cmd": "/bin/bash -i",
         "yield_time_ms": 200,
+        "tty": true,
     });
 
     let stdin_call_id = "uexec-stdin-delta";
@@ -905,6 +906,7 @@ async fn unified_exec_terminal_interaction_captures_delayed_output() -> Result<(
     let open_args = json!({
         "cmd": "sleep 3 && echo MARKER1 && sleep 3 && echo MARKER2",
         "yield_time_ms": 10,
+        "tty": true,
     });
 
     // Poll stdin three times: first for no output, second after the first marker,
@@ -1966,6 +1968,7 @@ async fn unified_exec_reuses_session_via_stdin() -> Result<()> {
     let first_args = serde_json::json!({
         "cmd": "/bin/cat",
         "yield_time_ms": 200,
+        "tty": true,
     });
 
     let second_call_id = "uexec-stdin";
@@ -2678,6 +2681,7 @@ async fn unified_exec_prunes_exited_sessions_first() -> Result<()> {
     let keep_args = serde_json::json!({
         "cmd": "/bin/cat",
         "yield_time_ms": 250,
+        "tty": true,
     });
 
     let prune_call_id = "uexec-prune-target";
@@ -2685,6 +2689,7 @@ async fn unified_exec_prunes_exited_sessions_first() -> Result<()> {
     let prune_args = serde_json::json!({
         "cmd": "sleep 1",
         "yield_time_ms": 1_250,
+        "tty": true,
     });
 
     let mut events = vec![ev_response_created("resp-prune-1")];

codex-rs/utils/pty/README.md

@@ -6,6 +6,7 @@ Lightweight helpers for spawning interactive processes either under a PTY (pseud
 
 - `spawn_pty_process(program, args, cwd, env, arg0)` → `SpawnedProcess`
 - `spawn_pipe_process(program, args, cwd, env, arg0)` → `SpawnedProcess`
+- `spawn_pipe_process_no_stdin(program, args, cwd, env, arg0)` → `SpawnedProcess`
 - `conpty_supported()` → `bool` (Windows only; always true elsewhere)
 - `ProcessHandle` exposes:
   - `writer_sender()` → `mpsc::Sender<Vec<u8>>` (stdin)
@@ -46,6 +47,7 @@ let exit_code = spawned.exit_rx.await.unwrap_or(-1);
 ```
 
 Swap in `spawn_pipe_process` for a non-TTY subprocess; the rest of the API stays the same.
+Use `spawn_pipe_process_no_stdin` to force stdin closed (commands that read stdin will see EOF immediately).
 
 ## Tests
 

codex-rs/utils/pty/src/lib.rs

@@ -9,6 +9,8 @@ mod win;
 
 /// Spawn a non-interactive process using regular pipes for stdin/stdout/stderr.
 pub use pipe::spawn_process as spawn_pipe_process;
+/// Spawn a non-interactive process using regular pipes, but close stdin immediately.
+pub use pipe::spawn_process_no_stdin as spawn_pipe_process_no_stdin;
 /// Handle for interacting with a spawned process (PTY or pipe).
 pub use process::ProcessHandle;
 /// Bundle of process handles plus output and exit receivers returned by spawn helpers.

codex-rs/utils/pty/src/pipe.rs

@@ -90,13 +90,19 @@ where
     }
 }
 
-/// Spawn a process using regular pipes (no PTY), returning handles for stdin, output, and exit.
-pub async fn spawn_process(
+#[derive(Clone, Copy)]
+enum PipeStdinMode {
+    Piped,
+    Null,
+}
+
+async fn spawn_process_with_stdin_mode(
     program: &str,
     args: &[String],
     cwd: &Path,
     env: &HashMap<String, String>,
     arg0: &Option<String>,
+    stdin_mode: PipeStdinMode,
 ) -> Result<SpawnedProcess> {
     if program.is_empty() {
         anyhow::bail!("missing program for pipe spawn");
@@ -128,7 +134,14 @@ pub async fn spawn_process(
     for arg in args {
         command.arg(arg);
     }
-    command.stdin(Stdio::piped());
+    match stdin_mode {
+        PipeStdinMode::Piped => {
+            command.stdin(Stdio::piped());
+        }
+        PipeStdinMode::Null => {
+            command.stdin(Stdio::null());
+        }
+    }
     command.stdout(Stdio::piped());
     command.stderr(Stdio::piped());
 
@@ -147,18 +160,19 @@ pub async fn spawn_process(
     let (output_tx, _) = broadcast::channel::<Vec<u8>>(256);
     let initial_output_rx = output_tx.subscribe();
 
-    let writer_handle = tokio::spawn({
-        let writer = stdin.map(|w| Arc::new(tokio::sync::Mutex::new(w)));
-        async move {
+    let writer_handle = if let Some(stdin) = stdin {
+        let writer = Arc::new(tokio::sync::Mutex::new(stdin));
+        tokio::spawn(async move {
             while let Some(bytes) = writer_rx.recv().await {
-                if let Some(writer) = &writer {
-                    let mut guard = writer.lock().await;
-                    let _ = guard.write_all(&bytes).await;
-                    let _ = guard.flush().await;
-                }
+                let mut guard = writer.lock().await;
+                let _ = guard.write_all(&bytes).await;
+                let _ = guard.flush().await;
             }
-        }
-    });
+        })
+    } else {
+        drop(writer_rx);
+        tokio::spawn(async {})
+    };
 
     let stdout_handle = stdout.map(|stdout| {
         let output_tx = output_tx.clone();
@@ -230,3 +244,25 @@ pub async fn spawn_process(
         exit_rx,
     })
 }
+
+/// Spawn a process using regular pipes (no PTY), returning handles for stdin, output, and exit.
+pub async fn spawn_process(
+    program: &str,
+    args: &[String],
+    cwd: &Path,
+    env: &HashMap<String, String>,
+    arg0: &Option<String>,
+) -> Result<SpawnedProcess> {
+    spawn_process_with_stdin_mode(program, args, cwd, env, arg0, PipeStdinMode::Piped).await
+}
+
+/// Spawn a process using regular pipes, but close stdin immediately.
+pub async fn spawn_process_no_stdin(
+    program: &str,
+    args: &[String],
+    cwd: &Path,
+    env: &HashMap<String, String>,
+    arg0: &Option<String>,
+) -> Result<SpawnedProcess> {
+    spawn_process_with_stdin_mode(program, args, cwd, env, arg0, PipeStdinMode::Null).await
+}