Make one more test 2

codex-rs/.config/nextest.toml

@@ -11,3 +11,7 @@ slow-timeout = { period = "1m", terminate-after = 4 }
 [[profile.default.overrides]]
 filter = 'test(approval_matrix_covers_all_modes)'
 slow-timeout = { period = "30s", terminate-after = 2 }
+
+[[profile.default.overrides]]
+filter = 'test(windows_unified_exec)'
+slow-timeout = { period = "30s", terminate-after = 4 }

codex-rs/Cargo.lock

@@ -1213,6 +1213,7 @@ dependencies = [
  "tree-sitter-bash",
  "url",
  "uuid",
+ "vt100",
  "walkdir",
  "which",
  "wildmatch",

codex-rs/core/Cargo.toml

@@ -59,6 +59,7 @@ shlex = { workspace = true }
 similar = { workspace = true }
 strum_macros = { workspace = true }
 url = { workspace = true }
+vt100 = { workspace = true }
 once_cell = { workspace = true }
 regex = { workspace = true }
 tempfile = { workspace = true }

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

@@ -234,13 +234,12 @@ mod tests {
 
         assert_eq!(buffer.total_bytes, UNIFIED_EXEC_OUTPUT_MAX_BYTES);
         let snapshot = buffer.snapshot();
-        assert_eq!(snapshot.len(), 3);
+        assert_eq!(snapshot.len(), 2);
         assert_eq!(
             snapshot.first().unwrap().len(),
             UNIFIED_EXEC_OUTPUT_MAX_BYTES - 2
         );
-        assert_eq!(snapshot.get(2).unwrap(), &vec![b'c']);
-        assert_eq!(snapshot.get(1).unwrap(), &vec![b'b']);
+        assert_eq!(snapshot.get(1).unwrap(), &vec![b'b', b'c']);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]

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

@@ -31,8 +31,25 @@ pub(crate) struct OutputBufferState {
 
 impl OutputBufferState {
     pub(super) fn push_chunk(&mut self, chunk: Vec<u8>) {
-        self.total_bytes = self.total_bytes.saturating_add(chunk.len());
-        self.chunks.push_back(chunk);
+        if chunk.is_empty() {
+            return;
+        }
+
+        // On Windows (especially with ConPTY) output can arrive in many tiny chunks.
+        // Coalesce them to avoid pathological behavior in drain/collect paths.
+        const MAX_COALESCED_CHUNK_BYTES: usize = 8_192;
+
+        let mut chunk = chunk;
+        let chunk_len = chunk.len();
+        self.total_bytes = self.total_bytes.saturating_add(chunk_len);
+
+        if let Some(tail) = self.chunks.back_mut()
+            && tail.len().saturating_add(chunk_len) <= MAX_COALESCED_CHUNK_BYTES
+        {
+            tail.append(&mut chunk);
+        } else {
+            self.chunks.push_back(chunk);
+        }
 
         let mut excess = self
             .total_bytes

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

@@ -62,6 +62,17 @@ const UNIFIED_EXEC_ENV: [(&str, &str); 8] = [
     ("GIT_PAGER", "cat"),
 ];
 
+fn normalize_unified_exec_text(raw: &str) -> String {
+    if cfg!(target_os = "windows") {
+        let mut parser = vt100::Parser::new(40, 120, 1_024);
+        parser.process(raw.as_bytes());
+        parser.screen_mut().set_scrollback(usize::MAX);
+        return parser.screen().contents();
+    }
+
+    raw.to_string()
+}
+
 fn apply_unified_exec_env(mut env: HashMap<String, String>) -> HashMap<String, String> {
     for (key, value) in UNIFIED_EXEC_ENV {
         env.insert(key.to_string(), value.to_string());
@@ -163,7 +174,8 @@ impl UnifiedExecSessionManager {
         .await;
         let wall_time = Instant::now().saturating_duration_since(start);
 
-        let text = String::from_utf8_lossy(&collected).to_string();
+        let raw_text = String::from_utf8_lossy(&collected).to_string();
+        let text = normalize_unified_exec_text(&raw_text);
         let output = formatted_truncate_text(&text, TruncationPolicy::Tokens(max_tokens));
         let has_exited = session.has_exited();
         let exit_code = session.exit_code();
@@ -285,7 +297,8 @@ impl UnifiedExecSessionManager {
         .await;
         let wall_time = Instant::now().saturating_duration_since(start);
 
-        let text = String::from_utf8_lossy(&collected).to_string();
+        let raw_text = String::from_utf8_lossy(&collected).to_string();
+        let text = normalize_unified_exec_text(&raw_text);
         let output = formatted_truncate_text(&text, TruncationPolicy::Tokens(max_tokens));
         let original_token_count = approx_token_count(&text);
         let chunk_id = generate_chunk_id();

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

@@ -30,6 +30,7 @@ use core_test_support::test_codex::test_codex;
 use core_test_support::wait_for_event;
 use core_test_support::wait_for_event_match;
 use core_test_support::wait_for_event_with_timeout;
+use pretty_assertions::assert_eq;
 use regex_lite::Regex;
 use serde_json::Value;
 use serde_json::json;
@@ -343,7 +344,12 @@ async fn unified_exec_emits_exec_command_begin_event() -> Result<()> {
 
     assert_eq!(begin_event.cwd, cwd.path());
 
-    wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await;
+    wait_for_event_with_timeout(
+        &codex,
+        |event| matches!(event, EventMsg::TaskComplete(_)),
+        Duration::from_secs(10),
+    )
+    .await;
 
     Ok(())
 }
@@ -2316,6 +2322,284 @@ async fn unified_exec_prunes_exited_sessions_first() -> Result<()> {
     Ok(())
 }
 
+#[cfg(target_os = "windows")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn windows_unified_exec_escape_output_snapshot() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    skip_if_sandbox!(Ok(()));
+
+    let server = start_mock_server().await;
+
+    let mut builder = test_codex().with_config(|config| {
+        config.use_experimental_unified_exec_tool = true;
+        config.features.enable(Feature::UnifiedExec);
+    });
+    let TestCodex {
+        codex,
+        cwd,
+        session_configured,
+        ..
+    } = builder.build(&server).await?;
+
+    let call_id = "windows-uexec-escapes";
+    let args = json!({
+        "cmd": "echo \"UEXEC-WINDOWS-ESCAPES\"",
+        "yield_time_ms": 10_000,
+    });
+
+    let responses = vec![
+        sse(vec![
+            ev_response_created("resp-1"),
+            ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
+            ev_completed("resp-1"),
+        ]),
+        sse(vec![
+            ev_response_created("resp-2"),
+            ev_assistant_message("msg-1", "done"),
+            ev_completed("resp-2"),
+        ]),
+    ];
+    mount_sse_sequence(&server, responses).await;
+
+    let session_model = session_configured.model.clone();
+
+    codex
+        .submit(Op::UserTurn {
+            items: vec![UserInput::Text {
+                text: "capture unified_exec escape output on Windows".into(),
+            }],
+            final_output_json_schema: None,
+            cwd: cwd.path().to_path_buf(),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::DangerFullAccess,
+            model: session_model,
+            effort: None,
+            summary: ReasoningSummary::Auto,
+        })
+        .await?;
+
+    wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await;
+
+    let requests = server.received_requests().await.expect("recorded requests");
+    assert!(!requests.is_empty(), "expected at least one POST request");
+
+    let bodies = requests
+        .iter()
+        .map(|req| req.body_json::<Value>().expect("request json"))
+        .collect::<Vec<_>>();
+
+    let outputs = collect_tool_outputs(&bodies)?;
+    let parsed = outputs
+        .get(call_id)
+        .expect("missing unified_exec output for Windows escape snapshot");
+
+    let raw_output = parsed.output.as_str();
+
+    assert!(
+        raw_output.contains("UEXEC-WINDOWS-ESCAPES"),
+        "expected marker string in unified_exec output, got {raw_output:?}"
+    );
+    assert!(
+        !raw_output.contains('\u{1b}'),
+        "expected unified_exec output to be stripped of ANSI escape sequences on Windows, got {raw_output:?}"
+    );
+
+    Ok(())
+}
+
+#[cfg(target_os = "windows")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn windows_unified_exec_write_stdin_strips_escapes() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    skip_if_sandbox!(Ok(()));
+
+    let server = start_mock_server().await;
+
+    let mut builder = test_codex().with_config(|config| {
+        config.use_experimental_unified_exec_tool = true;
+        config.features.enable(Feature::UnifiedExec);
+    });
+    let TestCodex {
+        codex,
+        cwd,
+        session_configured,
+        ..
+    } = builder.build(&server).await?;
+
+    let open_call_id = "windows-uexec-open-stdin";
+    let script = "$esc=[char]27; while ($true) { $line=[Console]::In.ReadLine(); if ($null -eq $line) { break }; Write-Output ($esc + '[31mUEXEC-WINDOWS-STDIN' + $esc + '[0m') }";
+    let open_args = json!({
+        "cmd": script,
+        "yield_time_ms": 2_000,
+    });
+
+    let stdin_call_id = "windows-uexec-stdin-escapes";
+    let stdin_args = json!({
+        "chars": "trigger\r\n",
+        "session_id": 1000,
+        "yield_time_ms": 5_000,
+    });
+
+    let responses = vec![
+        sse(vec![
+            ev_response_created("resp-1"),
+            ev_function_call(
+                open_call_id,
+                "exec_command",
+                &serde_json::to_string(&open_args)?,
+            ),
+            ev_completed("resp-1"),
+        ]),
+        sse(vec![
+            ev_response_created("resp-2"),
+            ev_function_call(
+                stdin_call_id,
+                "write_stdin",
+                &serde_json::to_string(&stdin_args)?,
+            ),
+            ev_completed("resp-2"),
+        ]),
+        sse(vec![
+            ev_response_created("resp-3"),
+            ev_assistant_message("msg-1", "done"),
+            ev_completed("resp-3"),
+        ]),
+    ];
+    mount_sse_sequence(&server, responses).await;
+
+    let session_model = session_configured.model.clone();
+
+    codex
+        .submit(Op::UserTurn {
+            items: vec![UserInput::Text {
+                text: "windows write_stdin escape stripping".into(),
+            }],
+            final_output_json_schema: None,
+            cwd: cwd.path().to_path_buf(),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::DangerFullAccess,
+            model: session_model,
+            effort: None,
+            summary: ReasoningSummary::Auto,
+        })
+        .await?;
+
+    wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await;
+
+    let requests = server.received_requests().await.expect("recorded requests");
+    assert!(!requests.is_empty(), "expected at least one POST request");
+
+    let bodies = requests
+        .iter()
+        .map(|req| req.body_json::<Value>().expect("request json"))
+        .collect::<Vec<_>>();
+
+    let outputs = collect_tool_outputs(&bodies)?;
+    let parsed = outputs
+        .get(stdin_call_id)
+        .expect("missing unified_exec write_stdin output on Windows");
+
+    let raw_output = parsed.output.as_str();
+
+    assert!(
+        raw_output.contains("UEXEC-WINDOWS-STDIN"),
+        "expected marker string in write_stdin unified_exec output, got {raw_output:?}"
+    );
+    assert!(
+        !raw_output.contains('\u{1b}'),
+        "expected write_stdin unified_exec output to be stripped of ANSI escape sequences on Windows, got {raw_output:?}"
+    );
+
+    Ok(())
+}
+
+#[cfg(target_os = "windows")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn windows_unified_exec_large_output_strips_escapes() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    skip_if_sandbox!(Ok(()));
+
+    let server = start_mock_server().await;
+
+    let mut builder = test_codex().with_config(|config| {
+        config.use_experimental_unified_exec_tool = true;
+        config.features.enable(Feature::UnifiedExec);
+    });
+    let TestCodex {
+        codex,
+        cwd,
+        session_configured,
+        ..
+    } = builder.build(&server).await?;
+
+    let call_id = "windows-uexec-large-output";
+    let cmd = "1..200 | ForEach-Object { Write-Output \"UEXEC-LARGE-$_\" }";
+    let args = json!({
+        "cmd": cmd,
+        "yield_time_ms": 2000,
+        "max_output_tokens": 50,
+    });
+
+    let responses = vec![
+        sse(vec![
+            ev_response_created("resp-1"),
+            ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
+            ev_completed("resp-1"),
+        ]),
+        sse(vec![
+            ev_response_created("resp-2"),
+            ev_assistant_message("msg-1", "done"),
+            ev_completed("resp-2"),
+        ]),
+    ];
+    mount_sse_sequence(&server, responses).await;
+
+    let session_model = session_configured.model.clone();
+
+    codex
+        .submit(Op::UserTurn {
+            items: vec![UserInput::Text {
+                text: "windows large-output escape stripping".into(),
+            }],
+            final_output_json_schema: None,
+            cwd: cwd.path().to_path_buf(),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::DangerFullAccess,
+            model: session_model,
+            effort: None,
+            summary: ReasoningSummary::Auto,
+        })
+        .await?;
+
+    wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await;
+
+    let requests = server.received_requests().await.expect("recorded requests");
+    assert!(!requests.is_empty(), "expected at least one POST request");
+
+    let bodies = requests
+        .iter()
+        .map(|req| req.body_json::<Value>().expect("request json"))
+        .collect::<Vec<_>>();
+
+    let outputs = collect_tool_outputs(&bodies)?;
+    let parsed = outputs
+        .get(call_id)
+        .expect("missing unified_exec large-output snapshot on Windows");
+
+    let raw_output = parsed.output.as_str();
+
+    assert!(
+        raw_output.contains("UEXEC-LARGE-"),
+        "expected large-output marker lines in unified_exec output, got {raw_output:?}"
+    );
+    assert!(
+        !raw_output.contains('\u{1b}'),
+        "expected unified_exec large-output summary to be stripped of ANSI escape sequences on Windows, got {raw_output:?}"
+    );
+
+    Ok(())
+}
+
 fn assert_command(command: &[String], expected_args: &str, expected_cmd: &str) {
     assert_eq!(command.len(), 3);
     let shell_path = &command[0];

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

@@ -189,7 +189,11 @@ pub async fn spawn_pty_process(
                 let mut guard = writer.lock().await;
                 use std::io::Write;
                 let _ = guard.write_all(&bytes);
-                let _ = guard.flush();
+                // On Windows, flushing ConPTY input pipes can hang indefinitely in some cases.
+                // Writing is sufficient to deliver input to the PTY.
+                if !cfg!(windows) {
+                    let _ = guard.flush();
+                }
             }
         }
     });