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); +}