diff --git a/codex-rs/utils/pty/src/tests.rs b/codex-rs/utils/pty/src/tests.rs index 75579ea45f..0c90ad88a3 100644 --- a/codex-rs/utils/pty/src/tests.rs +++ b/codex-rs/utils/pty/src/tests.rs @@ -345,6 +345,13 @@ fn extract_marker_lines(output: &[u8]) -> Vec { .collect() } +fn output_has_marker_line(output: &[u8], marker: &str) -> bool { + normalize_output(output).lines().any(|line| { + let trimmed = line.trim_start(); + trimmed.starts_with(marker) + }) +} + #[tokio::test] // Verifies basic stdout and exit code parity for a simple command. async fn standard_output_parity() -> anyhow::Result<()> { @@ -370,6 +377,76 @@ async fn standard_output_parity() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +// Verifies commands with no output stay empty and match parity. +async fn no_output_parity() -> anyhow::Result<()> { + if cfg!(windows) && !conpty_supported() { + return Ok(()); + } + let command = shell_command( + #[cfg(unix)] + "true", + #[cfg(windows)] + "exit /b 0", + ); + let env = base_env(); + let cwd = temp_dir(); + + let pty = run_pty(&command, &cwd, &env, None).await?; + let piped = run_piped(&command, &cwd, &env, None).await?; + + assert_eq!(normalize_output(&pty.output), ""); + assert_eq!(normalize_output(&piped.output), ""); + assert_parity(&pty, &piped); + Ok(()) +} + +#[tokio::test] +// Verifies exit state and stored exit code match PTY and are consistent with exit_rx. +async fn exit_state_parity() -> anyhow::Result<()> { + if cfg!(windows) && !conpty_supported() { + return Ok(()); + } + let command = shell_command( + #[cfg(unix)] + r#"printf "done\n"; exit 3"#, + #[cfg(windows)] + r#"echo done & exit /b 3"#, + ); + let env = base_env(); + let cwd = temp_dir(); + + let pty_spawned = spawn_pty_process(&command.program, &command.args, &cwd, &env, &None).await?; + let pty_result = tokio::time::timeout( + OUTPUT_TIMEOUT, + collect_output(pty_spawned.output_rx, pty_spawned.exit_rx), + ) + .await??; + let pty_code = match pty_result.exit { + ExitOutcome::Exited(code) => code, + ExitOutcome::Dropped => anyhow::bail!("pty exit dropped"), + }; + assert!(pty_spawned.session.has_exited()); + assert_eq!(pty_spawned.session.exit_code(), Some(pty_code)); + + let piped_spawned = + fallback::spawn_piped_process(&command.program, &command.args, &cwd, &env, &None).await?; + let piped_result = tokio::time::timeout( + OUTPUT_TIMEOUT, + collect_output(piped_spawned.output_rx, piped_spawned.exit_rx), + ) + .await??; + let piped_code = match piped_result.exit { + ExitOutcome::Exited(code) => code, + ExitOutcome::Dropped => anyhow::bail!("piped exit dropped"), + }; + assert!(piped_spawned.session.has_exited()); + assert_eq!(piped_spawned.session.exit_code(), Some(piped_code)); + + assert_eq!(pty_code, piped_code); + Ok(()) +} + #[tokio::test] // Verifies env propagation and working directory parity. async fn env_and_cwd_parity() -> anyhow::Result<()> { @@ -489,6 +566,46 @@ async fn terminate_parity() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +// Verifies terminate can be called twice and piped matches PTY. +async fn terminate_idempotency_parity() -> anyhow::Result<()> { + if cfg!(windows) && !conpty_supported() { + return Ok(()); + } + let command = shell_command( + #[cfg(unix)] + "sleep 60", + #[cfg(windows)] + r#"ping -n 60 127.0.0.1 >nul"#, + ); + let env = base_env(); + let cwd = temp_dir(); + + let pty = spawn_pty_process(&command.program, &command.args, &cwd, &env, &None).await?; + let piped = + fallback::spawn_piped_process(&command.program, &command.args, &cwd, &env, &None).await?; + + tokio::time::sleep(Duration::from_millis(50)).await; + pty.session.terminate(); + pty.session.terminate(); + piped.session.terminate(); + piped.session.terminate(); + + let pty_exit = tokio::time::timeout(TERMINATE_TIMEOUT, pty.exit_rx).await?; + let piped_exit = tokio::time::timeout(TERMINATE_TIMEOUT, piped.exit_rx).await?; + match pty_exit { + Ok(code) => { + let piped_code = piped_exit?; + assert_eq!(code, piped_code); + } + Err(pty_err) => { + let piped_err = piped_exit.unwrap_err(); + assert_eq!(pty_err.to_string(), piped_err.to_string()); + } + } + Ok(()) +} + #[tokio::test] // Verifies empty program errors are consistent across implementations. async fn empty_program_errors() { @@ -607,6 +724,82 @@ async fn multi_command_parity() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +// Verifies late output subscribers receive subsequent output. +async fn output_subscriber_parity() -> anyhow::Result<()> { + if cfg!(windows) && !conpty_supported() { + return Ok(()); + } + let command = shell_repl_command(); + let env = base_env(); + let cwd = temp_dir(); + + let mut pty = spawn_interactive_pty(&command, &cwd, &env).await?; + let mut piped = spawn_interactive_piped(&command, &cwd, &env).await?; + + let mut pty_late_rx = pty.session.output_receiver(); + let mut piped_late_rx = piped.session.output_receiver(); + let mut pty_late_buffer = Vec::new(); + let mut piped_late_buffer = Vec::new(); + + let marker = "CMD_MARKER_SUB"; + send_line(&mut pty.writer, &format!("echo {marker}")).await?; + send_line(&mut piped.writer, &format!("echo {marker}")).await?; + tokio::time::timeout(OUTPUT_TIMEOUT, async { + loop { + if output_has_marker_line(&pty_late_buffer, marker) { + break; + } + match pty_late_rx.recv().await { + Ok(bytes) => pty_late_buffer.extend_from_slice(&bytes), + Err(broadcast::error::RecvError::Closed) => { + anyhow::bail!("output channel closed before receiving {marker}"); + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + anyhow::bail!("output lagged by {skipped} messages"); + } + } + } + Ok::<_, anyhow::Error>(()) + }) + .await??; + tokio::time::timeout(OUTPUT_TIMEOUT, async { + loop { + if output_has_marker_line(&piped_late_buffer, marker) { + break; + } + match piped_late_rx.recv().await { + Ok(bytes) => piped_late_buffer.extend_from_slice(&bytes), + Err(broadcast::error::RecvError::Closed) => { + anyhow::bail!("output channel closed before receiving {marker}"); + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + anyhow::bail!("output lagged by {skipped} messages"); + } + } + } + Ok::<_, anyhow::Error>(()) + }) + .await??; + + send_line(&mut pty.writer, "exit").await?; + send_line(&mut piped.writer, "exit").await?; + + let pty = finish_interactive(pty).await?; + let piped = finish_interactive(piped).await?; + assert_eq!(pty.exit, ExitOutcome::Exited(0)); + assert_eq!(piped.exit, ExitOutcome::Exited(0)); + assert_eq!( + extract_marker_lines(&pty_late_buffer), + vec![marker.to_string()] + ); + assert_eq!( + extract_marker_lines(&piped_late_buffer), + vec![marker.to_string()] + ); + Ok(()) +} + #[tokio::test] // Verifies post-kill write failures and exit errors match. async fn multi_command_after_kill_parity() -> anyhow::Result<()> {