From f51f674c5716dc0da7e8b9197c416d7b78115af4 Mon Sep 17 00:00:00 2001 From: wsp Date: Thu, 17 Sep 2026 23:45:37 +0800 Subject: [PATCH] fix(terminal): Preserve split UTF-8 output Keep incomplete UTF-8 characters across process reads so Git Bash output is not misidentified as Windows-1252 at buffer boundaries. Decode stdout, stderr and PTY output before sharing text with tool results, streaming progress and background capture. Preserve stateful legacy encoding support and avoid marking read boundaries as EOF during encoding detection. Cover split characters, polling, truncation, capture, legacy encodings and real Git Bash cat. Validation: - 29 terminal execution tests passed - 2 workspace-origin compatibility tests passed - cargo check -p terminal-core passed --- src/crates/services/terminal/AGENTS.md | 1 + src/crates/services/terminal/src/exec.rs | 152 ++++------------ .../terminal/src/exec/output_decoder.rs | 169 ++++++++++++++++++ .../src/exec/output_encoding_tests.rs | 147 +++++++++++++++ 4 files changed, 355 insertions(+), 114 deletions(-) create mode 100644 src/crates/services/terminal/src/exec/output_decoder.rs create mode 100644 src/crates/services/terminal/src/exec/output_encoding_tests.rs diff --git a/src/crates/services/terminal/AGENTS.md b/src/crates/services/terminal/AGENTS.md index 6db99775e5..6fae3347b0 100644 --- a/src/crates/services/terminal/AGENTS.md +++ b/src/crates/services/terminal/AGENTS.md @@ -22,6 +22,7 @@ infrastructure, not a product command or UI layer. ```bash cargo check -p terminal-core +cargo test -p terminal-core --lib exec:: # process output, encoding, polling and lifecycle cargo test -p terminal-core --lib workspace_origin_contract_tests # terminal response compatibility node scripts/check-core-boundaries.mjs ``` diff --git a/src/crates/services/terminal/src/exec.rs b/src/crates/services/terminal/src/exec.rs index 9e1e3d453d..5e08afbed7 100644 --- a/src/crates/services/terminal/src/exec.rs +++ b/src/crates/services/terminal/src/exec.rs @@ -5,8 +5,11 @@ //! while that process is still running so later calls can poll or write stdin. use crate::{shell::invalidate_cached_executable, TerminalError, TerminalResult}; -use chardetng::EncodingDetector; -use encoding_rs::{Encoding, IBM866, WINDOWS_1252}; +mod output_decoder; +#[cfg(test)] +mod output_encoding_tests; + +use output_decoder::OutputDecoder; use portable_pty::{native_pty_system, CommandBuilder, MasterPty, PtySize, SlavePty}; use rand::Rng; use std::collections::{HashMap, VecDeque}; @@ -223,7 +226,7 @@ struct OutputState { } struct OutputInner { - chunks: VecDeque<(u64, Vec)>, + chunks: VecDeque<(u64, String)>, next_seq: u64, retained_bytes: usize, total_output_chars: usize, @@ -779,13 +782,12 @@ impl OutputState { } } - async fn push_chunk(&self, chunk: Vec) { + async fn push_text(&self, chunk: String) { if chunk.is_empty() { return; } - let decoded = bytes_to_string_smart(&chunk); - let decoded_chars = decoded.chars().count(); - let capture_text = self.output_capture_tx.as_ref().map(|_| decoded); + let decoded_chars = chunk.chars().count(); + let capture_text = self.output_capture_tx.as_ref().map(|_| chunk.clone()); { let mut inner = self.inner.lock().await; let seq = inner.next_seq; @@ -846,10 +848,9 @@ impl OutputState { let inner = self.inner.lock().await; for (seq, chunk) in inner.chunks.iter() { if *seq >= cursor.next_seq { - let text = bytes_to_string_smart(chunk); - sink.push_str(&text); + sink.push_str(chunk); if let Some(tx) = output_tx { - let _ = tx.try_send(text); + let _ = tx.try_send(chunk.clone()); } } } @@ -1150,9 +1151,11 @@ async fn spawn_pty_process(request: &ExecCommandRequest) -> TerminalResult Option { } #[cfg(not(unix))] -fn spawn_pipe_reader(mut reader: R, output: Arc) -> JoinHandle<()> +fn spawn_pipe_reader(reader: R, output: Arc) -> JoinHandle<()> where R: AsyncRead + Unpin + Send + 'static, { - tokio::spawn(async move { - let mut buffer = vec![0u8; 8192]; - loop { - match reader.read(&mut buffer).await { - Ok(0) => break, - Ok(n) => output.push_chunk(buffer[..n].to_vec()).await, - Err(ref error) if error.kind() == ErrorKind::Interrupted => continue, - Err(_) => break, - } + tokio::spawn(read_pipe_output(reader, output)) +} + +async fn read_pipe_output(mut reader: R, output: Arc) +where + R: AsyncRead + Unpin, +{ + let mut decoder = OutputDecoder::default(); + let mut buffer = vec![0u8; 8192]; + loop { + match reader.read(&mut buffer).await { + Ok(0) => break, + Ok(n) => output.push_text(decoder.push(&buffer[..n])).await, + Err(ref error) if error.kind() == ErrorKind::Interrupted => continue, + Err(_) => break, } - }) + } + output.push_text(decoder.finish()).await; } #[cfg(not(unix))] @@ -1634,7 +1644,7 @@ where #[cfg(unix)] fn spawn_pipe_reader_with_done( - mut reader: R, + reader: R, output: Arc, done_tx: mpsc::Sender<()>, ) -> JoinHandle<()> @@ -1642,15 +1652,7 @@ where R: AsyncRead + Unpin + Send + 'static, { tokio::spawn(async move { - let mut buffer = vec![0u8; 8192]; - loop { - match reader.read(&mut buffer).await { - Ok(0) => break, - Ok(n) => output.push_chunk(buffer[..n].to_vec()).await, - Err(ref error) if error.kind() == ErrorKind::Interrupted => continue, - Err(_) => break, - } - } + read_pipe_output(reader, output).await; let _ = done_tx.send(()).await; }) } @@ -1746,74 +1748,16 @@ fn input_bytes_for_write(chars: &str, append_enter: bool) -> Vec { bytes } -fn bytes_to_string_smart(bytes: &[u8]) -> String { - if bytes.is_empty() { - return String::new(); - } - - if let Ok(text) = std::str::from_utf8(bytes) { - return text.to_owned(); - } - - decode_bytes(bytes, detect_encoding(bytes)) -} - -fn detect_encoding(bytes: &[u8]) -> &'static Encoding { - let mut detector = EncodingDetector::new(); - detector.feed(bytes, true); - let (encoding, _is_confident) = detector.guess_assess(None, true); - - if encoding == IBM866 && looks_like_windows_1252_punctuation(bytes) { - return WINDOWS_1252; - } - - encoding -} - -fn decode_bytes(bytes: &[u8], encoding: &'static Encoding) -> String { - let (decoded, _, had_errors) = encoding.decode(bytes); - if had_errors { - String::from_utf8_lossy(bytes).into_owned() - } else { - decoded.into_owned() - } -} - -const WINDOWS_1252_PUNCT_BYTES: [u8; 8] = [0x91, 0x92, 0x93, 0x94, 0x95, 0x96, 0x97, 0x99]; - -fn looks_like_windows_1252_punctuation(bytes: &[u8]) -> bool { - let mut saw_extended_punctuation = false; - let mut saw_ascii_word = false; - - for &byte in bytes { - if byte >= 0xA0 { - return false; - } - if (0x80..=0x9F).contains(&byte) { - if !WINDOWS_1252_PUNCT_BYTES.contains(&byte) { - return false; - } - saw_extended_punctuation = true; - } - if byte.is_ascii_alphabetic() { - saw_ascii_word = true; - } - } - - saw_extended_punctuation && saw_ascii_word -} - #[cfg(test)] mod tests { use super::{ - bytes_to_string_smart, input_bytes_for_write, ExecCommandRequest, ExecControlAction, - ExecControlOrigin, ExecControlRequest, ExecProcessLifecycleStatus, ExecProcessManager, + input_bytes_for_write, ExecCommandRequest, ExecControlAction, ExecControlOrigin, + ExecControlRequest, ExecProcessLifecycleStatus, ExecProcessManager, ExecSessionCompletionSource, ExecSessionCompletionStatus, HeadTailText, OutputCursor, OutputState, SendStdinRequest, WriteStdinRequest, }; #[cfg(windows)] use crate::shell::{ShellDetector, ShellType}; - use encoding_rs::GBK; use std::collections::HashMap; #[cfg(windows)] use std::path::PathBuf; @@ -1911,9 +1855,9 @@ mod tests { #[tokio::test] async fn original_output_chars_survives_multibyte_retention_eviction() { let output = OutputState::new(None); - let chunk = "🦀".repeat(32 * 1024).into_bytes(); + let chunk = "🦀".repeat(32 * 1024); for _ in 0..10 { - output.push_chunk(chunk.clone()).await; + output.push_text(chunk.clone()).await; } output.close(Some(0)).await; @@ -2811,26 +2755,6 @@ print("parent_exit", flush=True)"#; ); } - #[test] - fn smart_decode_preserves_utf8() { - assert_eq!(bytes_to_string_smart("小游戏平台".as_bytes()), "小游戏平台"); - } - - #[test] - fn smart_decode_handles_gbk_chinese_output() { - let (encoded, _, had_errors) = GBK.encode("小游戏平台"); - assert!(!had_errors); - assert_eq!(bytes_to_string_smart(&encoded), "小游戏平台"); - } - - #[test] - fn smart_decode_keeps_windows_1252_punctuation() { - assert_eq!( - bytes_to_string_smart(b"\x93\x94 test \x96 dash"), - "\u{201C}\u{201D} test \u{2013} dash" - ); - } - #[cfg(unix)] fn line_count(path: &std::path::Path) -> usize { std::fs::read_to_string(path) diff --git a/src/crates/services/terminal/src/exec/output_decoder.rs b/src/crates/services/terminal/src/exec/output_decoder.rs new file mode 100644 index 0000000000..0a36b81626 --- /dev/null +++ b/src/crates/services/terminal/src/exec/output_decoder.rs @@ -0,0 +1,169 @@ +//! Decode each process stream before merging or retaining its output. + +use chardetng::EncodingDetector; +use encoding_rs::{Decoder, Encoding, IBM866, WINDOWS_1252}; + +/// Pipe reads and PTY frames need not end at a character boundary. Keep the +/// unfinished suffix across reads (and tool polls), rather than guessing a new +/// encoding for an otherwise valid UTF-8 chunk. +#[derive(Default)] +pub(super) struct OutputDecoder { + pending: Vec, + fallback: Option, +} + +impl OutputDecoder { + pub(super) fn push(&mut self, bytes: &[u8]) -> String { + if let Some(decoder) = self.fallback.as_mut() { + return decode_fallback(decoder, bytes, false); + } + + self.pending.extend_from_slice(bytes); + match std::str::from_utf8(&self.pending) { + Ok(text) => { + let text = text.to_owned(); + self.pending.clear(); + text + } + Err(error) if error.error_len().is_none() => { + let boundary = error.valid_up_to(); + let text = String::from_utf8_lossy(&self.pending[..boundary]).into_owned(); + self.pending.drain(..boundary); + text + } + Err(_) => { + // Retain the existing GBK / Windows code-page compatibility, + // but only after a real UTF-8 error, never an incomplete suffix. + // A stateful decoder also preserves split legacy characters. + let mut decoder = detect_encoding(&self.pending).new_decoder_without_bom_handling(); + let text = decode_fallback(&mut decoder, &self.pending, false); + self.pending.clear(); + self.fallback = Some(decoder); + text + } + } + } + + pub(super) fn finish(mut self) -> String { + if let Some(decoder) = self.fallback.as_mut() { + return decode_fallback(decoder, &[], true); + } + // Only EOF makes an unfinished UTF-8 scalar malformed. Do not reinterpret + // it as a different encoding or silently discard it. + String::from_utf8_lossy(&self.pending).into_owned() + } +} + +fn decode_fallback(decoder: &mut Decoder, bytes: &[u8], last: bool) -> String { + let capacity = decoder + .max_utf8_buffer_length(bytes.len()) + .expect("process output decode capacity overflow"); + let mut text = String::with_capacity(capacity); + let (_, read, _) = decoder.decode_to_string(bytes, &mut text, last); + debug_assert_eq!(read, bytes.len()); + text +} + +fn detect_encoding(bytes: &[u8]) -> &'static Encoding { + let mut detector = EncodingDetector::new(); + // This is a stream prefix, not EOF: a legacy multibyte character may also + // straddle the read boundary and must remain a viable encoding candidate. + detector.feed(bytes, false); + let (encoding, _is_confident) = detector.guess_assess(None, true); + + if encoding == IBM866 && looks_like_windows_1252_punctuation(bytes) { + return WINDOWS_1252; + } + + encoding +} + +const WINDOWS_1252_PUNCT_BYTES: [u8; 8] = [0x91, 0x92, 0x93, 0x94, 0x95, 0x96, 0x97, 0x99]; + +fn looks_like_windows_1252_punctuation(bytes: &[u8]) -> bool { + let mut saw_extended_punctuation = false; + let mut saw_ascii_word = false; + + for &byte in bytes { + if byte >= 0xA0 { + return false; + } + if (0x80..=0x9F).contains(&byte) { + if !WINDOWS_1252_PUNCT_BYTES.contains(&byte) { + return false; + } + saw_extended_punctuation = true; + } + if byte.is_ascii_alphabetic() { + saw_ascii_word = true; + } + } + + saw_extended_punctuation && saw_ascii_word +} + +#[cfg(test)] +mod tests { + use super::OutputDecoder; + use encoding_rs::GBK; + + #[test] + fn utf8_survives_every_split_and_single_byte_reads() { + let expected = "ASCII / 中文，é😀 / end"; + for split in 0..=expected.len() { + let mut decoder = OutputDecoder::default(); + let mut text = decoder.push(&expected.as_bytes()[..split]); + text.push_str(&decoder.push(&expected.as_bytes()[split..])); + text.push_str(&decoder.finish()); + assert_eq!(text, expected, "split at byte {split}"); + } + + let mut decoder = OutputDecoder::default(); + let mut text = String::new(); + for byte in expected.as_bytes() { + text.push_str(&decoder.push(&[*byte])); + } + text.push_str(&decoder.finish()); + assert_eq!(text, expected); + } + + #[test] + fn incomplete_utf8_waits_for_more_bytes_and_flushes_only_at_eof() { + let mut decoder = OutputDecoder::default(); + assert_eq!(decoder.push(b"ready\xef\xbc"), "ready"); + assert_eq!(decoder.push(b""), ""); + assert_eq!(decoder.push(b"\x8c"), "，"); + assert_eq!(decoder.push(b"tail\xf0\x9f"), "tail"); + assert_eq!(decoder.finish(), "\u{fffd}"); + } + + #[test] + fn legacy_gbk_output_preserves_decoder_state_across_reads() { + let expected = "小游戏平台小游戏平台"; + let (bytes, _, had_errors) = GBK.encode(expected); + assert!(!had_errors); + let mut decoder = OutputDecoder::default(); + // Supply enough text for detection, ending on a GBK lead byte. + let mut text = decoder.push(&bytes[..9]); + for byte in &bytes[9..] { + text.push_str(&decoder.push(&[*byte])); + } + text.push_str(&decoder.finish()); + assert_eq!(text, expected); + } + + #[test] + fn legacy_windows_1252_punctuation_remains_supported() { + let mut decoder = OutputDecoder::default(); + assert_eq!(decoder.push(b"\x93\x94 test \x96 dash"), "“” test – dash"); + assert_eq!(decoder.finish(), ""); + } + + #[test] + fn legacy_gbk_incomplete_character_is_flushed_at_eof() { + let (bytes, _, _) = GBK.encode("小游戏平台"); + let mut decoder = OutputDecoder::default(); + assert_eq!(decoder.push(&bytes[..9]), "小游戏平"); + assert_eq!(decoder.finish(), "\u{fffd}"); + } +} diff --git a/src/crates/services/terminal/src/exec/output_encoding_tests.rs b/src/crates/services/terminal/src/exec/output_encoding_tests.rs new file mode 100644 index 0000000000..2b8b0caf37 --- /dev/null +++ b/src/crates/services/terminal/src/exec/output_encoding_tests.rs @@ -0,0 +1,147 @@ +use super::{read_pipe_output, OutputCursor, OutputState}; +use std::sync::Arc; +use tokio::io::AsyncWriteExt; +use tokio::sync::mpsc; +use tokio::time::Instant; + +fn initial_cursor() -> OutputCursor { + OutputCursor { + next_seq: 0, + observed_output_chars: 0, + } +} + +#[tokio::test] +async fn pipe_read_boundary_preserves_results_progress_capture_and_counts() { + // The comma's EF BC | 8C bytes straddle the real 8192-byte read buffer. + let expected = format!("{}，后😀\n", "中".repeat(2730)); + let (capture_tx, mut capture_rx) = mpsc::unbounded_channel(); + let output = Arc::new(OutputState::new(Some(capture_tx))); + read_pipe_output(expected.as_bytes(), Arc::clone(&output)).await; + output.close(Some(0)).await; + + let (progress_tx, mut progress_rx) = mpsc::channel(16); + let collected = output + .collect_until(initial_cursor(), Instant::now(), 10_000, Some(&progress_tx)) + .await; + assert_eq!(collected.output, expected); + assert_eq!(collected.original_output_chars, expected.chars().count()); + let mut progress = String::new(); + while let Ok(text) = progress_rx.try_recv() { + progress.push_str(&text); + } + let mut captured = String::new(); + while let Ok(text) = capture_rx.try_recv() { + captured.push_str(&text); + } + assert_eq!(progress, expected); + assert_eq!(captured, expected); + + let truncated = output + .collect_until(initial_cursor(), Instant::now(), 8, None) + .await; + assert_eq!(truncated.original_output_chars, expected.chars().count()); + assert_eq!( + truncated.output, + "中中中中\n... [truncated, middle omitted] ...\n，后😀\n" + ); +} + +#[tokio::test] +async fn pipe_streams_keep_independent_partial_characters_across_polls() { + let (capture_tx, mut capture_rx) = mpsc::unbounded_channel(); + let output = Arc::new(OutputState::new(Some(capture_tx))); + let (mut stdout_writer, stdout) = tokio::io::duplex(64); + let (mut stderr_writer, stderr) = tokio::io::duplex(64); + let stdout_task = tokio::spawn(read_pipe_output(stdout, Arc::clone(&output))); + let stderr_task = tokio::spawn(read_pipe_output(stderr, Arc::clone(&output))); + + stdout_writer.write_all(b"out:\xe4\xb8").await.unwrap(); + assert_eq!(capture_rx.recv().await.unwrap(), "out:"); + stderr_writer.write_all(b"err:\xe6\x96").await.unwrap(); + assert_eq!(capture_rx.recv().await.unwrap(), "err:"); + let first = output + .collect_until(initial_cursor(), Instant::now(), 100, None) + .await; + assert_eq!(first.output, "out:err:"); + assert_eq!(first.original_output_chars, 8); + + stdout_writer.write_all(b"\xad").await.unwrap(); + assert_eq!(capture_rx.recv().await.unwrap(), "中"); + stderr_writer.write_all(b"\x87").await.unwrap(); + assert_eq!(capture_rx.recv().await.unwrap(), "文"); + let second = output + .collect_until(first.cursor, Instant::now(), 100, None) + .await; + assert_eq!(second.output, "中文"); + assert_eq!(second.original_output_chars, 2); + + stdout_writer.write_all(b"\xf0\x9f").await.unwrap(); + drop(stdout_writer); + drop(stderr_writer); + stdout_task.await.unwrap(); + stderr_task.await.unwrap(); + output.close(Some(0)).await; + assert_eq!(capture_rx.recv().await.unwrap(), "\u{fffd}"); + let last = output + .collect_until(second.cursor, Instant::now(), 100, None) + .await; + assert_eq!(last.output, "\u{fffd}"); + assert_eq!(last.original_output_chars, 1); +} + +#[cfg(windows)] +#[tokio::test] +async fn git_bash_cat_preserves_utf8_fixture_through_exec_command() { + use super::{ExecCommandRequest, ExecProcessManager}; + use crate::shell::ShellDetector; + use std::collections::HashMap; + + let Some(shell) = ShellDetector::detect_git_bash() else { + eprintln!("Skipping Git Bash encoding regression: Git Bash is not installed"); + return; + }; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("utf8-boundary.txt"); + let expected = format!("{}，后😀\n", "中".repeat(2730)); + std::fs::write(&path, expected.as_bytes()).unwrap(); + let (capture_tx, mut capture_rx) = mpsc::unbounded_channel(); + let (progress_tx, mut progress_rx) = mpsc::channel(32); + let manager = ExecProcessManager::default(); + let result = manager + .exec_command_streaming( + ExecCommandRequest { + argv: vec![ + shell.path.to_string_lossy().into_owned(), + "-lc".into(), + "cat -- \"$OPENBITFUN_ENCODING_FIXTURE\"".into(), + ], + cwd: dir.path().to_path_buf(), + env: HashMap::from([( + "OPENBITFUN_ENCODING_FIXTURE".into(), + path.to_string_lossy().replace('\\', "/"), + )]), + tty: false, + yield_time_ms: Some(10_000), + max_output_chars: Some(10_000), + lifecycle_tx: None, + output_capture_tx: Some(capture_tx), + }, + progress_tx, + ) + .await + .unwrap(); + assert_eq!(result.exit_code, Some(0)); + assert_eq!(result.output, expected); + assert_eq!(result.original_output_chars, expected.chars().count()); + let mut progress = String::new(); + while let Ok(text) = progress_rx.try_recv() { + progress.push_str(&text); + } + let mut capture = String::new(); + while let Ok(text) = capture_rx.try_recv() { + capture.push_str(&text); + } + assert_eq!(progress, expected); + assert_eq!(capture, expected); +}