Skip to content

Commit 2ca270d

Browse files
aibrahim-oaicodex
andauthored
[2/8] Support piped stdin in exec process API (#18086)
## Summary - Add an explicit stdin mode to process/start. - Keep normal non-interactive exec stdin closed while allowing pipe-backed processes. ## Stack ```text o #18027 [8/8] Fail exec client operations after disconnect │ o #18025 [7/8] Cover MCP stdio tests with executor placement │ o #18089 [6/8] Wire remote MCP stdio through executor │ o #18088 [5/8] Add executor process transport for MCP stdio │ o #18087 [4/8] Abstract MCP stdio server launching │ o #18020 [3/8] Add pushed exec process events │ @ #18086 [2/8] Support piped stdin in exec process API │ o #18085 [1/8] Add MCP server environment config │ o main ``` Co-authored-by: Codex <noreply@openai.com>
1 parent 6e72f0d commit 2ca270d

File tree

9 files changed

+215
-7
lines changed

9 files changed

+215
-7
lines changed

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,7 @@ fn exec_server_params_for_request(
152152
env_policy,
153153
env,
154154
tty,
155+
pipe_stdin: false,
155156
arg0: request.arg0.clone(),
156157
}
157158
}

codex-rs/exec-server/README.md

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ Request params:
8585
"PATH": "/usr/bin:/bin"
8686
},
8787
"tty": true,
88+
"pipeStdin": false,
8889
"arg0": null
8990
}
9091
```
@@ -95,8 +96,8 @@ Field definitions:
9596
- `argv`: command vector. It must be non-empty.
9697
- `cwd`: absolute working directory used for the child process.
9798
- `env`: environment variables passed to the child process.
98-
- `tty`: when `true`, spawn a PTY-backed interactive process; when `false`,
99-
spawn a pipe-backed process with closed stdin.
99+
- `tty`: when `true`, spawn a PTY-backed interactive process.
100+
- `pipeStdin`: when `true`, keep non-PTY stdin writable via `process/write`.
100101
- `arg0`: optional argv0 override forwarded to `codex-utils-pty`.
101102

102103
Response:
@@ -111,7 +112,7 @@ Behavior notes:
111112

112113
- Reusing an existing `processId` is rejected.
113114
- PTY-backed processes accept later writes through `process/write`.
114-
- Pipe-backed processes are launched with stdin closed and reject writes.
115+
- Non-PTY processes reject writes unless `pipeStdin` is `true`.
115116
- Output is streamed asynchronously via `process/output`.
116117
- Exit is reported asynchronously via `process/exited`.
117118

@@ -153,7 +154,7 @@ Response:
153154

154155
### `process/write`
155156

156-
Writes raw bytes to a running PTY-backed process stdin.
157+
Writes raw bytes to a running process stdin.
157158

158159
Request params:
159160

@@ -177,7 +178,7 @@ Response:
177178
Behavior notes:
178179

179180
- Writes to an unknown `processId` are rejected.
180-
- Writes to a non-PTY process are rejected because stdin is already closed.
181+
- Writes to a non-PTY process are rejected unless it started with `pipeStdin`.
181182

182183
### `process/terminate`
183184

@@ -325,7 +326,7 @@ Initialize:
325326
Start a process:
326327

327328
```json
328-
{"id":2,"method":"process/start","params":{"processId":"proc-1","argv":["bash","-lc","printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"],"cwd":"/tmp","env":{"PATH":"/usr/bin:/bin"},"tty":true,"arg0":null}}
329+
{"id":2,"method":"process/start","params":{"processId":"proc-1","argv":["bash","-lc","printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"],"cwd":"/tmp","env":{"PATH":"/usr/bin:/bin"},"tty":true,"pipeStdin":false,"arg0":null}}
329330
{"id":2,"result":{"processId":"proc-1"}}
330331
{"method":"process/output","params":{"processId":"proc-1","seq":1,"stream":"stdout","chunk":"cmVhZHkK"}}
331332
```

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -346,6 +346,7 @@ mod tests {
346346
env_policy: None,
347347
env: Default::default(),
348348
tty: false,
349+
pipe_stdin: false,
349350
arg0: None,
350351
})
351352
.await

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

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@ struct RetainedOutputChunk {
5959
struct RunningProcess {
6060
session: ExecCommandSession,
6161
tty: bool,
62+
pipe_stdin: bool,
6263
output: VecDeque<RetainedOutputChunk>,
6364
retained_bytes: usize,
6465
next_seq: u64,
@@ -165,6 +166,15 @@ impl LocalProcess {
165166
TerminalSize::default(),
166167
)
167168
.await
169+
} else if params.pipe_stdin {
170+
codex_utils_pty::spawn_pipe_process(
171+
program,
172+
args,
173+
params.cwd.as_path(),
174+
&env,
175+
&params.arg0,
176+
)
177+
.await
168178
} else {
169179
codex_utils_pty::spawn_pipe_process_no_stdin(
170180
program,
@@ -195,6 +205,7 @@ impl LocalProcess {
195205
ProcessEntry::Running(Box::new(RunningProcess {
196206
session: spawned.session,
197207
tty: params.tty,
208+
pipe_stdin: params.pipe_stdin,
198209
output: VecDeque::new(),
199210
retained_bytes: 0,
200211
next_seq: 1,
@@ -339,7 +350,7 @@ impl LocalProcess {
339350
status: WriteStatus::Starting,
340351
});
341352
};
342-
if !process.tty {
353+
if !process.tty && !process.pipe_stdin {
343354
return Ok(WriteResponse {
344355
status: WriteStatus::StdinClosed,
345356
});
@@ -667,6 +678,7 @@ mod tests {
667678
env_policy: None,
668679
env,
669680
tty: false,
681+
pipe_stdin: false,
670682
arg0: None,
671683
}
672684
}

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,9 @@ pub struct ExecParams {
6969
pub env_policy: Option<ExecEnvPolicy>,
7070
pub env: HashMap<String, String>,
7171
pub tty: bool,
72+
/// Keep non-tty stdin writable through `process/write`.
73+
#[serde(default)]
74+
pub pipe_stdin: bool,
7275
pub arg0: Option<String>,
7376
}
7477

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ fn exec_params_with_argv(process_id: &str, argv: Vec<String>) -> ExecParams {
3030
env_policy: None,
3131
env: inherited_path_env(),
3232
tty: false,
33+
pipe_stdin: false,
3334
arg0: None,
3435
}
3536
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -393,6 +393,7 @@ mod tests {
393393
env_policy: None,
394394
env,
395395
tty: false,
396+
pipe_stdin: false,
396397
arg0: None,
397398
}
398399
}

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

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ use codex_exec_server::ExecProcess;
1212
use codex_exec_server::ProcessId;
1313
use codex_exec_server::ReadResponse;
1414
use codex_exec_server::StartedExecProcess;
15+
use codex_exec_server::WriteStatus;
1516
use pretty_assertions::assert_eq;
1617
use test_case::test_case;
1718
use tokio::sync::watch;
@@ -54,6 +55,7 @@ async fn assert_exec_process_starts_and_exits(use_remote: bool) -> Result<()> {
5455
env_policy: /*env_policy*/ None,
5556
env: Default::default(),
5657
tty: false,
58+
pipe_stdin: false,
5759
arg0: None,
5860
})
5961
.await?;
@@ -131,6 +133,7 @@ async fn assert_exec_process_streams_output(use_remote: bool) -> Result<()> {
131133
env_policy: /*env_policy*/ None,
132134
env: Default::default(),
133135
tty: false,
136+
pipe_stdin: false,
134137
arg0: None,
135138
})
136139
.await?;
@@ -164,6 +167,7 @@ async fn assert_exec_process_write_then_read(use_remote: bool) -> Result<()> {
164167
env_policy: /*env_policy*/ None,
165168
env: Default::default(),
166169
tty: true,
170+
pipe_stdin: false,
167171
arg0: None,
168172
})
169173
.await?;
@@ -184,6 +188,73 @@ async fn assert_exec_process_write_then_read(use_remote: bool) -> Result<()> {
184188
Ok(())
185189
}
186190

191+
async fn assert_exec_process_write_then_read_without_tty(use_remote: bool) -> Result<()> {
192+
let context = create_process_context(use_remote).await?;
193+
let process_id = "proc-stdin-pipe".to_string();
194+
let session = context
195+
.backend
196+
.start(ExecParams {
197+
process_id: process_id.clone().into(),
198+
argv: vec![
199+
"/bin/sh".to_string(),
200+
"-c".to_string(),
201+
"IFS= read line; printf 'from-stdin:%s\\n' \"$line\"".to_string(),
202+
],
203+
cwd: std::env::current_dir()?,
204+
env_policy: /*env_policy*/ None,
205+
env: Default::default(),
206+
tty: false,
207+
pipe_stdin: true,
208+
arg0: None,
209+
})
210+
.await?;
211+
assert_eq!(session.process.process_id().as_str(), process_id);
212+
213+
tokio::time::sleep(Duration::from_millis(200)).await;
214+
let write_response = session.process.write(b"hello\n".to_vec()).await?;
215+
assert_eq!(write_response.status, WriteStatus::Accepted);
216+
let StartedExecProcess { process } = session;
217+
let wake_rx = process.subscribe_wake();
218+
let actual = collect_process_output_from_reads(process, wake_rx).await?;
219+
220+
assert_eq!(actual, ("from-stdin:hello\n".to_string(), Some(0), true));
221+
Ok(())
222+
}
223+
224+
async fn assert_exec_process_rejects_write_without_pipe_stdin(use_remote: bool) -> Result<()> {
225+
let context = create_process_context(use_remote).await?;
226+
let process_id = "proc-stdin-closed".to_string();
227+
let session = context
228+
.backend
229+
.start(ExecParams {
230+
process_id: process_id.clone().into(),
231+
argv: vec![
232+
"/bin/sh".to_string(),
233+
"-c".to_string(),
234+
"sleep 0.3; if IFS= read -r line; then printf 'read:%s\\n' \"$line\"; else printf 'eof\\n'; fi".to_string(),
235+
],
236+
cwd: std::env::current_dir()?,
237+
env_policy: /*env_policy*/ None,
238+
env: Default::default(),
239+
tty: false,
240+
pipe_stdin: false,
241+
arg0: None,
242+
})
243+
.await?;
244+
assert_eq!(session.process.process_id().as_str(), process_id);
245+
246+
let write_response = session.process.write(b"ignored\n".to_vec()).await?;
247+
assert_eq!(write_response.status, WriteStatus::StdinClosed);
248+
let StartedExecProcess { process } = session;
249+
let wake_rx = process.subscribe_wake();
250+
let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;
251+
252+
assert_eq!(output, "eof\n");
253+
assert_eq!(exit_code, Some(0));
254+
assert!(closed);
255+
Ok(())
256+
}
257+
187258
async fn assert_exec_process_preserves_queued_events_before_subscribe(
188259
use_remote: bool,
189260
) -> Result<()> {
@@ -201,6 +272,7 @@ async fn assert_exec_process_preserves_queued_events_before_subscribe(
201272
env_policy: /*env_policy*/ None,
202273
env: Default::default(),
203274
tty: false,
275+
pipe_stdin: false,
204276
arg0: None,
205277
})
206278
.await?;
@@ -234,6 +306,7 @@ async fn remote_exec_process_reports_transport_disconnect() -> Result<()> {
234306
env_policy: /*env_policy*/ None,
235307
env: Default::default(),
236308
tty: false,
309+
pipe_stdin: false,
237310
arg0: None,
238311
})
239312
.await?;
@@ -289,6 +362,24 @@ async fn exec_process_write_then_read(use_remote: bool) -> Result<()> {
289362
assert_exec_process_write_then_read(use_remote).await
290363
}
291364

365+
#[test_case(false ; "local")]
366+
#[test_case(true ; "remote")]
367+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
368+
// Serialize tests that launch a real exec-server process through the full CLI.
369+
#[serial_test::serial(remote_exec_server)]
370+
async fn exec_process_write_then_read_without_tty(use_remote: bool) -> Result<()> {
371+
assert_exec_process_write_then_read_without_tty(use_remote).await
372+
}
373+
374+
#[test_case(false ; "local")]
375+
#[test_case(true ; "remote")]
376+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
377+
// Serialize tests that launch a real exec-server process through the full CLI.
378+
#[serial_test::serial(remote_exec_server)]
379+
async fn exec_process_rejects_write_without_pipe_stdin(use_remote: bool) -> Result<()> {
380+
assert_exec_process_rejects_write_without_pipe_stdin(use_remote).await
381+
}
382+
292383
#[test_case(false ; "local")]
293384
#[test_case(true ; "remote")]
294385
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]

0 commit comments

Comments
 (0)