From 0b09838b32dd0ca2d16a3da941a0ecc00163ac7a Mon Sep 17 00:00:00 2001 From: Dylan Date: Wed, 12 Nov 2025 15:10:59 -0800 Subject: [PATCH] wip 1 --- codex-rs/core/src/exec_env.rs | 17 ++ .../core/src/tools/handlers/unified_exec.rs | 9 +- .../core/src/unified_exec/session_manager.rs | 31 ++- codex-rs/core/tests/suite/unified_exec.rs | 180 ++++++++++++++---- codex-rs/utils/pty/src/lib.rs | 141 +++++++++++++- 5 files changed, 324 insertions(+), 54 deletions(-) diff --git a/codex-rs/core/src/exec_env.rs b/codex-rs/core/src/exec_env.rs index 11334896bf..c9b2d12ba2 100644 --- a/codex-rs/core/src/exec_env.rs +++ b/codex-rs/core/src/exec_env.rs @@ -25,6 +25,23 @@ where ShellEnvironmentPolicyInherit::All => vars.into_iter().collect(), ShellEnvironmentPolicyInherit::None => HashMap::new(), ShellEnvironmentPolicyInherit::Core => { + #[cfg(target_os = "windows")] + const CORE_VARS: &[&str] = &[ + "COMSPEC", + "HOME", + "LOGNAME", + "PATH", + "PATHEXT", + "SYSTEMROOT", + "TEMP", + "TMP", + "TMPDIR", + "USER", + "USERNAME", + "USERPROFILE", + "WINDIR", + ]; + #[cfg(not(target_os = "windows"))] const CORE_VARS: &[&str] = &[ "HOME", "LOGNAME", "PATH", "SHELL", "USER", "USERNAME", "TMPDIR", "TEMP", "TMP", ]; diff --git a/codex-rs/core/src/tools/handlers/unified_exec.rs b/codex-rs/core/src/tools/handlers/unified_exec.rs index fb94f23d0d..7f2e9a9a76 100644 --- a/codex-rs/core/src/tools/handlers/unified_exec.rs +++ b/codex-rs/core/src/tools/handlers/unified_exec.rs @@ -54,7 +54,14 @@ struct WriteStdinArgs { } fn default_shell() -> String { - "/bin/bash".to_string() + #[cfg(target_os = "windows")] + { + "powershell.exe".to_string() + } + #[cfg(not(target_os = "windows"))] + { + "/bin/bash".to_string() + } } fn default_login() -> bool { diff --git a/codex-rs/core/src/unified_exec/session_manager.rs b/codex-rs/core/src/unified_exec/session_manager.rs index c9763fbc49..1436feb885 100644 --- a/codex-rs/core/src/unified_exec/session_manager.rs +++ b/codex-rs/core/src/unified_exec/session_manager.rs @@ -43,12 +43,7 @@ impl UnifiedExecSessionManager { .workdir .clone() .unwrap_or_else(|| context.turn.cwd.clone()); - let shell_flag = if request.login { "-lc" } else { "-c" }; - let command = vec![ - request.shell.to_string(), - shell_flag.to_string(), - request.command.to_string(), - ]; + let command = Self::build_shell_command(request.shell, request.login, request.command); let session = self .open_session_with_sandbox( @@ -111,6 +106,30 @@ impl UnifiedExecSessionManager { Ok(response) } + fn build_shell_command(shell: &str, login: bool, command: &str) -> Vec { + #[cfg(target_os = "windows")] + { + let _ = login; + let shell_lower = shell.to_ascii_lowercase(); + let flag = if shell_lower.contains("cmd") { + "/C" + } else { + "-Command" + }; + return vec![shell.to_string(), flag.to_string(), command.to_string()]; + } + + #[cfg(not(target_os = "windows"))] + { + let shell_flag = if login { "-lc" } else { "-c" }; + vec![ + shell.to_string(), + shell_flag.to_string(), + command.to_string(), + ] + } + } + pub(crate) async fn write_stdin( &self, request: WriteStdinRequest<'_>, diff --git a/codex-rs/core/tests/suite/unified_exec.rs b/codex-rs/core/tests/suite/unified_exec.rs index 1225b51ccd..856debcfb2 100644 --- a/codex-rs/core/tests/suite/unified_exec.rs +++ b/codex-rs/core/tests/suite/unified_exec.rs @@ -1,4 +1,3 @@ -#![cfg(not(target_os = "windows"))] use std::collections::HashMap; use std::sync::OnceLock; @@ -151,6 +150,104 @@ fn collect_tool_outputs(bodies: &[Value]) -> Result String { + if cfg!(target_os = "windows") { + format!("Write-Output \"{text}\"") + } else { + format!("/bin/echo {text}") + } +} + +fn print_no_newline_command(text: &str) -> String { + if cfg!(target_os = "windows") { + format!("[Console]::Write('{text}')") + } else { + format!("printf '{text}'") + } +} + +fn current_dir_command() -> String { + if cfg!(target_os = "windows") { + "[Environment]::CurrentDirectory".to_string() + } else { + "pwd".to_string() + } +} + +fn ready_command() -> String { + echo_command("ready") +} + +fn cat_like_command() -> String { + if cfg!(target_os = "windows") { + "while (($line = [Console]::In.ReadLine()) -ne $null) { if ($line -eq '__EXIT__') { break }; Write-Output $line }".to_string() + } else { + "/bin/cat".to_string() + } +} + +fn cat_exit_input() -> &'static str { + if cfg!(target_os = "windows") { + "__EXIT__\n" + } else { + "\u{0004}" + } +} + +fn sleep_then_ready_command() -> String { + if cfg!(target_os = "windows") { + "Start-Sleep -Seconds 0.5; Write-Output 'ready'".to_string() + } else { + "sleep 0.5; echo ready".to_string() + } +} + +fn laggy_output_script() -> String { + if cfg!(target_os = "windows") { + concat!( + "$chunk = 'x' * 1048576; ", + "1..4 | ForEach-Object { [Console]::Write($chunk); [Console]::Out.Flush() }; ", + "Start-Sleep -Milliseconds 200; ", + "1..5 | ForEach-Object { Write-Output 'TAIL-MARKER'; Start-Sleep -Milliseconds 50 }; ", + "Start-Sleep -Milliseconds 200", + ) + .to_string() + } else { + r#"python3 - <<'PY' +import sys +import time + +chunk = b'x' * (1 << 20) +for _ in range(4): + sys.stdout.buffer.write(chunk) + sys.stdout.flush() + +time.sleep(0.2) +for _ in range(5): + sys.stdout.write("TAIL-MARKER\n") + sys.stdout.flush() + time.sleep(0.05) + +time.sleep(0.2) +PY +"# + .to_string() + } +} + +fn large_output_script() -> String { + if cfg!(target_os = "windows") { + "1..300 | ForEach-Object { \"line-$_\" }".to_string() + } else { + r#"python3 - <<'PY' +for i in range(300): + print(f"line-{i}") +PY +"# + .to_string() + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn unified_exec_emits_exec_command_begin_event() -> Result<()> { skip_if_no_network!(Ok(())); @@ -171,7 +268,7 @@ async fn unified_exec_emits_exec_command_begin_event() -> Result<()> { let call_id = "uexec-begin-event"; let args = json!({ - "cmd": "/bin/echo hello unified exec".to_string(), + "cmd": echo_command("hello unified exec"), "yield_time_ms": 250, }); @@ -212,9 +309,10 @@ async fn unified_exec_emits_exec_command_begin_event() -> Result<()> { }) .await; + let expected_command = vec![echo_command("hello unified exec")]; assert_eq!( begin_event.command, - vec!["/bin/echo hello unified exec".to_string()] + expected_command ); assert_eq!(begin_event.cwd, cwd.path()); @@ -245,11 +343,17 @@ async fn unified_exec_respects_workdir_override() -> Result<()> { std::fs::create_dir_all(&workdir)?; let call_id = "uexec-workdir"; - let args = json!({ - "cmd": "pwd", + let mut args = json!({ + "cmd": current_dir_command(), "yield_time_ms": 250, "workdir": workdir.to_string_lossy().to_string(), }); + if cfg!(target_os = "windows") { + if let Some(obj) = args.as_object_mut() { + obj.insert("shell".to_string(), json!("cmd.exe")); + obj.insert("login".to_string(), json!(false)); + } + } let responses = vec![ sse(vec![ @@ -297,7 +401,19 @@ async fn unified_exec_respects_workdir_override() -> Result<()> { .get(call_id) .expect("missing exec_command workdir output"); let output_text = output.output.trim(); - let output_canonical = std::fs::canonicalize(output_text)?; + assert!( + !output_text.is_empty(), + "workdir command should produce a path (raw output: {raw:?}, exit_code: {exit_code:?}, session_id: {session_id:?})", + raw = output.output, + exit_code = output.exit_code, + session_id = output.session_id + ); + let output_path = std::path::PathBuf::from(output_text); + assert!( + output_path.exists(), + "workdir output path does not exist: {output_text}" + ); + let output_canonical = std::fs::canonicalize(output_path)?; let expected_canonical = std::fs::canonicalize(&workdir)?; assert_eq!( output_canonical, expected_canonical, @@ -327,7 +443,7 @@ async fn unified_exec_emits_exec_command_end_event() -> Result<()> { let call_id = "uexec-end-event"; let args = json!({ - "cmd": "/bin/echo END-EVENT".to_string(), + "cmd": echo_command("END-EVENT"), "yield_time_ms": 250, }); let poll_call_id = "uexec-end-event-poll"; @@ -413,7 +529,7 @@ async fn unified_exec_emits_output_delta_for_exec_command() -> Result<()> { let call_id = "uexec-delta-1"; let args = json!({ - "cmd": "printf 'HELLO-UEXEC'", + "cmd": print_no_newline_command("HELLO-UEXEC"), "yield_time_ms": 1000, }); @@ -484,7 +600,7 @@ async fn unified_exec_emits_output_delta_for_write_stdin() -> Result<()> { let open_call_id = "uexec-open"; let open_args = json!({ - "cmd": "/bin/bash -i", + "cmd": cat_like_command(), "yield_time_ms": 200, }); @@ -582,8 +698,9 @@ async fn unified_exec_skips_begin_event_for_empty_input() -> Result<()> { } = builder.build(&server).await?; let open_call_id = "uexec-open-session"; + let open_cmd = ready_command(); let open_args = json!({ - "cmd": "/bin/sh -c echo ready".to_string(), + "cmd": open_cmd, "yield_time_ms": 250, }); @@ -654,7 +771,7 @@ async fn unified_exec_skips_begin_event_for_empty_input() -> Result<()> { "expected only the initial command to emit begin event" ); assert_eq!(begin_events[0].call_id, open_call_id); - assert_eq!(begin_events[0].command[0], "/bin/sh -c echo ready"); + assert_eq!(begin_events[0].command[0], open_cmd); Ok(()) } @@ -678,7 +795,7 @@ async fn exec_command_reports_chunk_and_exit_metadata() -> Result<()> { let call_id = "uexec-metadata"; let args = serde_json::json!({ - "cmd": "printf 'abcdefghijklmnopqrstuvwxyz'", + "cmd": print_no_newline_command("abcdefghijklmnopqrstuvwxyz"), "yield_time_ms": 500, "max_output_tokens": 6, }); @@ -788,7 +905,7 @@ async fn write_stdin_returns_exit_metadata_and_clears_session() -> Result<()> { let exit_call_id = "uexec-cat-exit"; let start_args = serde_json::json!({ - "cmd": "/bin/cat", + "cmd": cat_like_command(), "yield_time_ms": 500, }); let send_args = serde_json::json!({ @@ -797,7 +914,7 @@ async fn write_stdin_returns_exit_metadata_and_clears_session() -> Result<()> { "yield_time_ms": 500, }); let exit_args = serde_json::json!({ - "chars": "\u{0004}", + "chars": cat_exit_input(), "session_id": 0, "yield_time_ms": 500, }); @@ -945,7 +1062,7 @@ async fn unified_exec_emits_end_event_when_session_dies_via_stdin() -> Result<() let start_call_id = "uexec-end-on-exit-start"; let start_args = serde_json::json!({ - "cmd": "/bin/cat", + "cmd": cat_like_command(), "yield_time_ms": 200, }); @@ -958,7 +1075,7 @@ async fn unified_exec_emits_end_event_when_session_dies_via_stdin() -> Result<() let exit_call_id = "uexec-end-on-exit"; let exit_args = serde_json::json!({ - "chars": "\u{0004}", + "chars": cat_exit_input(), "session_id": 0, "yield_time_ms": 500, }); @@ -1048,7 +1165,7 @@ async fn unified_exec_reuses_session_via_stdin() -> Result<()> { let first_call_id = "uexec-start"; let first_args = serde_json::json!({ - "cmd": "/bin/cat", + "cmd": cat_like_command(), "yield_time_ms": 200, }); @@ -1155,24 +1272,7 @@ async fn unified_exec_streams_after_lagged_output() -> Result<()> { .. } = builder.build(&server).await?; - let script = r#"python3 - <<'PY' -import sys -import time - -chunk = b'x' * (1 << 20) -for _ in range(4): - sys.stdout.buffer.write(chunk) - sys.stdout.flush() - -time.sleep(0.2) -for _ in range(5): - sys.stdout.write("TAIL-MARKER\n") - sys.stdout.flush() - time.sleep(0.05) - -time.sleep(0.2) -PY -"#; + let script = laggy_output_script(); let first_call_id = "uexec-lag-start"; let first_args = serde_json::json!({ @@ -1282,7 +1382,7 @@ async fn unified_exec_timeout_and_followup_poll() -> Result<()> { let first_call_id = "uexec-timeout"; let first_args = serde_json::json!({ - "cmd": "sleep 0.5; echo ready", + "cmd": sleep_then_ready_command(), "yield_time_ms": 10, }); @@ -1386,11 +1486,7 @@ async fn unified_exec_formats_large_output_summary() -> Result<()> { .. } = builder.build(&server).await?; - let script = r#"python3 - <<'PY' -for i in range(300): - print(f"line-{i}") -PY -"#; + let script = large_output_script(); let call_id = "uexec-large-output"; let args = serde_json::json!({ @@ -1473,7 +1569,7 @@ async fn unified_exec_runs_under_sandbox() -> Result<()> { let call_id = "uexec"; let args = serde_json::json!({ - "cmd": "echo 'hello'", + "cmd": echo_command("hello"), "yield_time_ms": 500, }); diff --git a/codex-rs/utils/pty/src/lib.rs b/codex-rs/utils/pty/src/lib.rs index 14cc430760..0969af1a15 100644 --- a/codex-rs/utils/pty/src/lib.rs +++ b/codex-rs/utils/pty/src/lib.rs @@ -16,8 +16,8 @@ use tokio::sync::oneshot; use tokio::sync::Mutex as TokioMutex; use tokio::task::JoinHandle; -#[derive(Debug)] pub struct ExecCommandSession { + master: Box, writer_tx: mpsc::Sender>, output_tx: broadcast::Sender>, killer: StdMutex>>, @@ -28,9 +28,19 @@ pub struct ExecCommandSession { exit_code: Arc>>, } +impl std::fmt::Debug for ExecCommandSession { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ExecCommandSession") + .field("exit_status", &self.exit_status) + .field("exit_code", &self.exit_code) + .finish() + } +} + impl ExecCommandSession { #[allow(clippy::too_many_arguments)] pub fn new( + master: Box, writer_tx: mpsc::Sender>, output_tx: broadcast::Sender>, killer: Box, @@ -43,6 +53,7 @@ impl ExecCommandSession { let initial_output_rx = output_tx.subscribe(); ( Self { + master, writer_tx, output_tx, killer: StdMutex::new(Some(killer)), @@ -125,9 +136,22 @@ pub async fn spawn_pty_process( pixel_height: 0, })?; - let mut command_builder = CommandBuilder::new(arg0.as_ref().unwrap_or(&program.to_string())); + let master = pair.master; + let mut slave = pair.slave; + + let mut command_builder = CommandBuilder::new(program); + let _ = arg0; command_builder.cwd(cwd); + #[cfg(not(target_os = "windows"))] command_builder.env_clear(); + #[cfg(target_os = "windows")] + { + // Keep the inherited Windows environment to avoid missing critical + // variables that cause console hosts to fail to initialize. + for (key, value) in std::env::vars() { + command_builder.env(key, value); + } + } for arg in args { command_builder.arg(arg); } @@ -135,13 +159,33 @@ pub async fn spawn_pty_process( command_builder.env(key, value); } - let mut child = pair.slave.spawn_command(command_builder)?; + #[cfg(all(test, target_os = "windows"))] + eprintln!( + "spawn_pty_process env keys: {:?}", + env.keys().cloned().collect::>() + ); + + #[cfg(target_os = "windows")] + { + // Ensure core OS variables are present even if the provided env map + // was minimized. + for key in ["SystemRoot", "WINDIR", "COMSPEC", "PATHEXT", "PATH"] { + if !env.contains_key(key) { + if let Ok(value) = std::env::var(key) { + command_builder.env(key, value); + } + } + } + } + + let mut child = slave.spawn_command(command_builder)?; + drop(slave); let killer = child.clone_killer(); let (writer_tx, mut writer_rx) = mpsc::channel::>(128); let (output_tx, _) = broadcast::channel::>(256); - let mut reader = pair.master.try_clone_reader()?; + let mut reader = master.try_clone_reader()?; let output_tx_clone = output_tx.clone(); let reader_handle: JoinHandle<()> = tokio::task::spawn_blocking(move || { let mut buf = [0u8; 8_192]; @@ -161,7 +205,7 @@ pub async fn spawn_pty_process( } }); - let writer = pair.master.take_writer()?; + let writer = master.take_writer()?; let writer = Arc::new(TokioMutex::new(writer)); let writer_handle: JoinHandle<()> = tokio::spawn({ let writer = Arc::clone(&writer); @@ -193,6 +237,7 @@ pub async fn spawn_pty_process( }); let (session, output_rx) = ExecCommandSession::new( + master, writer_tx, output_tx, killer, @@ -209,3 +254,89 @@ pub async fn spawn_pty_process( exit_rx, }) } + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + #[tokio::test] + #[cfg(target_os = "windows")] + async fn spawn_cmd_succeeds() { + let mut env: HashMap = std::env::vars().collect(); + if let Some(system_root) = env.get("SystemRoot").cloned() { + let base_paths = vec![ + format!(r"{system_root}\system32"), + system_root.clone(), + format!(r"{system_root}\System32\Wbem"), + format!(r"{system_root}\System32\WindowsPowerShell\v1.0"), + ]; + env.insert("PATH".to_string(), base_paths.join(";")); + } + let cwd = std::env::current_dir().expect("current_dir"); + eprintln!( + "SystemRoot={:?} ComSpec={:?} PATH={:?}", + env.get("SystemRoot"), + env.get("ComSpec"), + env.get("PATH").map(|p| p.split(';').take(3).collect::>()) + ); + + let comspec = std::env::var("ComSpec").unwrap_or_else(|_| "cmd.exe".to_string()); + let mut spawned = spawn_pty_process( + &comspec, + &["/C".to_string(), "exit 0".to_string()], + &cwd, + &env, + &None, + ) + .await + .expect("spawn cmd"); + + let mut output_rx = spawned.output_rx; + let first_chunk = output_rx.try_recv().ok(); + eprintln!( + "first_chunk = {:?}", + first_chunk + .as_ref() + .map(|bytes| String::from_utf8_lossy(bytes)) + ); + + let status = spawned.exit_rx.await.expect("exit status"); + assert_eq!(status, 0, "cmd.exe should exit successfully"); + + // Drain any output to avoid broadcast warnings. + while output_rx.try_recv().is_ok() {} + } + + #[test] + #[cfg(target_os = "windows")] + fn spawn_cmd_blocking() { + let pty_system = native_pty_system(); + let mut pair = pty_system + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .expect("open pty"); + + let mut cmd = CommandBuilder::new( + std::env::var("ComSpec").unwrap_or_else(|_| "C:\\windows\\system32\\cmd.exe".into()), + ); + cmd.arg("/C"); + cmd.arg("exit 0"); + + let mut child = pair + .slave + .spawn_command(cmd) + .expect("spawn blocking cmd"); + drop(pair.slave); + + // Explicitly close stdin so the child can exit cleanly. + drop(pair.master.take_writer().expect("writer")); + + let status = child.wait().expect("wait for child"); + assert_eq!(status.exit_code(), 0, "cmd.exe exit code"); + } +}