diff --git a/src-tauri/src/proxy/providers/streaming.rs b/src-tauri/src/proxy/providers/streaming.rs index 65a10260e..d553ba6b9 100644 --- a/src-tauri/src/proxy/providers/streaming.rs +++ b/src-tauri/src/proxy/providers/streaming.rs @@ -93,6 +93,7 @@ pub fn create_anthropic_sse_stream( ) -> impl Stream> + Send { async_stream::stream! { let mut buffer = String::new(); + let mut utf8_remainder: Vec = Vec::new(); let mut message_id = None; let mut current_model = None; let mut next_content_index: u32 = 0; @@ -107,8 +108,7 @@ pub fn create_anthropic_sse_stream( while let Some(chunk) = stream.next().await { match chunk { Ok(bytes) => { - let text = String::from_utf8_lossy(&bytes); - buffer.push_str(&text); + crate::proxy::sse::append_utf8_safe(&mut buffer, &mut utf8_remainder, &bytes); while let Some(pos) = buffer.find("\n\n") { let line = buffer[..pos].to_string(); @@ -750,4 +750,45 @@ mod tests { assert!(deltas.contains(&"{\"a\":")); assert!(deltas.contains(&"1}")); } + + #[tokio::test] + async fn test_streaming_chinese_split_across_chunks_no_replacement_chars() { + // "你好" split across two TCP chunks inside a streaming text delta. + // Before the fix, from_utf8_lossy would produce U+FFFD for each half. + let full = concat!( + "data: {\"id\":\"chatcmpl_3\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"content\":\"你好\"}}]}\n\n", + "data: {\"id\":\"chatcmpl_3\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":2}}\n\n", + "data: [DONE]\n\n" + ); + let bytes = full.as_bytes(); + + // Find "你" in the byte stream and split inside it + let ni_start = bytes.windows(3).position(|w| w == "你".as_bytes()).unwrap(); + let split_point = ni_start + 1; // split after first byte of "你" + + let chunk1 = Bytes::from(bytes[..split_point].to_vec()); + let chunk2 = Bytes::from(bytes[split_point..].to_vec()); + + let upstream = stream::iter(vec![ + Ok::<_, std::io::Error>(chunk1), + Ok::<_, std::io::Error>(chunk2), + ]); + let converted = create_anthropic_sse_stream(upstream); + let chunks: Vec<_> = converted.collect().await; + + let merged = chunks + .into_iter() + .map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string()) + .collect::(); + + // Must contain the original Chinese characters, not replacement chars + assert!( + merged.contains("你好"), + "expected '你好' in output, got replacement chars (U+FFFD)" + ); + assert!( + !merged.contains('\u{FFFD}'), + "output must not contain U+FFFD replacement characters" + ); + } } diff --git a/src-tauri/src/proxy/providers/streaming_responses.rs b/src-tauri/src/proxy/providers/streaming_responses.rs index ea9274ff8..2c4b306e2 100644 --- a/src-tauri/src/proxy/providers/streaming_responses.rs +++ b/src-tauri/src/proxy/providers/streaming_responses.rs @@ -101,6 +101,7 @@ pub fn create_anthropic_sse_stream_from_responses impl Stream> + Send { async_stream::stream! { let mut buffer = String::new(); + let mut utf8_remainder: Vec = Vec::new(); let mut message_id: Option = None; let mut current_model: Option = None; let mut has_sent_message_start = false; @@ -118,8 +119,7 @@ pub fn create_anthropic_sse_stream_from_responses { - let text = String::from_utf8_lossy(&bytes); - buffer.push_str(&text); + crate::proxy::sse::append_utf8_safe(&mut buffer, &mut utf8_remainder, &bytes); // SSE 事件由 \n\n 分隔 while let Some(pos) = buffer.find("\n\n") { @@ -1029,4 +1029,45 @@ mod tests { assert_eq!(text_stops, 1); assert_eq!(text_deltas, vec!["你".to_string(), "好".to_string()]); } + + #[tokio::test] + async fn test_streaming_responses_chinese_split_across_chunks_no_replacement_chars() { + // Chinese text delta split across two TCP chunks. + let full = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_cn\",\"model\":\"gpt-4o\",\"usage\":{\"input_tokens\":5,\"output_tokens\":0}}}\n\n", + "event: response.output_text.delta\n", + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"你好世界\"}\n\n", + "event: response.completed\n", + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":4}}}\n\n" + ); + let bytes = full.as_bytes(); + + // Find "你" and split inside it + let ni_start = bytes.windows(3).position(|w| w == "你".as_bytes()).unwrap(); + let split_point = ni_start + 2; // split after second byte of "你" + + let chunk1 = Bytes::from(bytes[..split_point].to_vec()); + let chunk2 = Bytes::from(bytes[split_point..].to_vec()); + + let upstream = stream::iter(vec![ + Ok::<_, std::io::Error>(chunk1), + Ok::<_, std::io::Error>(chunk2), + ]); + let converted = create_anthropic_sse_stream_from_responses(upstream); + let chunks: Vec<_> = converted.collect().await; + let merged = chunks + .into_iter() + .map(|c| String::from_utf8_lossy(c.unwrap().as_ref()).to_string()) + .collect::(); + + assert!( + merged.contains("你好世界"), + "expected '你好世界' in output, got replacement chars (U+FFFD)" + ); + assert!( + !merged.contains('\u{FFFD}'), + "output must not contain U+FFFD replacement characters" + ); + } } diff --git a/src-tauri/src/proxy/response_handler.rs b/src-tauri/src/proxy/response_handler.rs index 408e66a86..502ed289f 100644 --- a/src-tauri/src/proxy/response_handler.rs +++ b/src-tauri/src/proxy/response_handler.rs @@ -71,6 +71,7 @@ impl StreamHandler { async_stream::stream! { let mut _last_activity = Instant::now(); let mut buffer = String::new(); + let mut utf8_remainder: Vec = Vec::new(); tokio::pin!(stream); @@ -82,8 +83,7 @@ impl StreamHandler { _last_activity = Instant::now(); // 解析 SSE 事件 - let text = String::from_utf8_lossy(&bytes); - buffer.push_str(&text); + crate::proxy::sse::append_utf8_safe(&mut buffer, &mut utf8_remainder, &bytes); // 提取完整事件 while let Some(pos) = buffer.find("\n\n") { diff --git a/src-tauri/src/proxy/response_processor.rs b/src-tauri/src/proxy/response_processor.rs index 609ff393a..3f4fe8425 100644 --- a/src-tauri/src/proxy/response_processor.rs +++ b/src-tauri/src/proxy/response_processor.rs @@ -568,6 +568,7 @@ pub fn create_logged_passthrough_stream( ) -> impl Stream> + Send { async_stream::stream! { let mut buffer = String::new(); + let mut utf8_remainder: Vec = Vec::new(); let mut collector = usage_collector; let mut is_first_chunk = true; @@ -619,8 +620,7 @@ pub fn create_logged_passthrough_stream( ); } is_first_chunk = false; - let text = String::from_utf8_lossy(&bytes); - buffer.push_str(&text); + crate::proxy::sse::append_utf8_safe(&mut buffer, &mut utf8_remainder, &bytes); // 尝试解析并记录完整的 SSE 事件 while let Some(pos) = buffer.find("\n\n") { diff --git a/src-tauri/src/proxy/sse.rs b/src-tauri/src/proxy/sse.rs index 235edce8b..f26a6a711 100644 --- a/src-tauri/src/proxy/sse.rs +++ b/src-tauri/src/proxy/sse.rs @@ -4,9 +4,71 @@ pub(crate) fn strip_sse_field<'a>(line: &'a str, field: &str) -> Option<&'a str> .or_else(|| line.strip_prefix(&format!("{field}:"))) } +/// Append raw bytes to a UTF-8 `String` buffer, correctly handling multi-byte +/// characters that are split across chunk boundaries. +/// +/// `remainder` accumulates trailing bytes from the previous chunk that form an +/// incomplete UTF-8 sequence (at most 3 bytes under normal operation). On each +/// call the remainder is prepended to `new_bytes`, the longest valid UTF-8 +/// prefix is appended to `buffer`, and any trailing incomplete bytes are saved +/// back into `remainder` for the next call. +/// +/// A defensive guard discards `remainder` via lossy conversion if it ever +/// exceeds 3 bytes, which cannot happen with well-formed UTF-8 streams. +pub(crate) fn append_utf8_safe(buffer: &mut String, remainder: &mut Vec, new_bytes: &[u8]) { + // Build the byte slice to decode: prepend any leftover bytes from previous chunk. + let (owned, bytes): (Option>, &[u8]) = if remainder.is_empty() { + (None, new_bytes) + } else { + // Defensive guard: remainder should never exceed 3 bytes (max incomplete + // UTF-8 sequence is 3 bytes: a 4-byte char missing its last byte). If it + // does, the stream is producing genuinely invalid bytes; flush them lossy + // and start fresh. + if remainder.len() > 3 { + buffer.push_str(&String::from_utf8_lossy(remainder)); + remainder.clear(); + (None, new_bytes) + } else { + let mut combined = std::mem::take(remainder); + combined.extend_from_slice(new_bytes); + (Some(combined), &[]) + } + }; + let input = owned.as_deref().unwrap_or(bytes); + + // Decode loop: consume all valid UTF-8 and any genuinely invalid bytes, + // only leaving a trailing incomplete sequence in remainder. + let mut pos = 0; + loop { + match std::str::from_utf8(&input[pos..]) { + Ok(s) => { + buffer.push_str(s); + // Everything consumed – remainder stays empty. + return; + } + Err(e) => { + let valid_up_to = pos + e.valid_up_to(); + buffer.push_str( + // Safety: from_utf8 guarantees [pos..valid_up_to] is valid UTF-8. + std::str::from_utf8(&input[pos..valid_up_to]).unwrap(), + ); + if let Some(invalid_len) = e.error_len() { + // Genuinely invalid byte(s) – emit U+FFFD and continue. + buffer.push('\u{FFFD}'); + pos = valid_up_to + invalid_len; + } else { + // Incomplete trailing sequence – stash for next chunk. + *remainder = input[valid_up_to..].to_vec(); + return; + } + } + } + } +} + #[cfg(test)] mod tests { - use super::strip_sse_field; + use super::{append_utf8_safe, strip_sse_field}; #[test] fn strip_sse_field_accepts_optional_space() { @@ -28,4 +90,215 @@ mod tests { ); assert_eq!(strip_sse_field("id:1", "data"), None); } + + // ------------------------------------------------------------------ + // append_utf8_safe tests + // ------------------------------------------------------------------ + + #[test] + fn ascii_passthrough() { + let mut buf = String::new(); + let mut rem = Vec::new(); + append_utf8_safe(&mut buf, &mut rem, b"hello world"); + assert_eq!(buf, "hello world"); + assert!(rem.is_empty()); + } + + #[test] + fn complete_multibyte_in_single_chunk() { + let mut buf = String::new(); + let mut rem = Vec::new(); + append_utf8_safe(&mut buf, &mut rem, "你好世界".as_bytes()); + assert_eq!(buf, "你好世界"); + assert!(rem.is_empty()); + } + + #[test] + fn split_multibyte_across_two_chunks() { + // "你" = E4 BD A0 (3 bytes) + let bytes = "你".as_bytes(); + assert_eq!(bytes.len(), 3); + + let mut buf = String::new(); + let mut rem = Vec::new(); + + // Chunk 1: first 2 bytes (incomplete) + append_utf8_safe(&mut buf, &mut rem, &bytes[..2]); + assert_eq!(buf, ""); + assert_eq!(rem.len(), 2); + + // Chunk 2: last byte completes the character + append_utf8_safe(&mut buf, &mut rem, &bytes[2..]); + assert_eq!(buf, "你"); + assert!(rem.is_empty()); + } + + #[test] + fn split_four_byte_char_across_chunks() { + // 😀 = F0 9F 98 80 (4 bytes) + let bytes = "😀".as_bytes(); + assert_eq!(bytes.len(), 4); + + let mut buf = String::new(); + let mut rem = Vec::new(); + + // Send 1 byte at a time + append_utf8_safe(&mut buf, &mut rem, &bytes[..1]); + assert_eq!(buf, ""); + assert_eq!(rem.len(), 1); + + append_utf8_safe(&mut buf, &mut rem, &bytes[1..2]); + assert_eq!(buf, ""); + assert_eq!(rem.len(), 2); + + append_utf8_safe(&mut buf, &mut rem, &bytes[2..3]); + assert_eq!(buf, ""); + assert_eq!(rem.len(), 3); + + append_utf8_safe(&mut buf, &mut rem, &bytes[3..]); + assert_eq!(buf, "😀"); + assert!(rem.is_empty()); + } + + #[test] + fn mixed_ascii_and_split_multibyte() { + // "hi你" = 68 69 E4 BD A0 + let all = "hi你".as_bytes(); + assert_eq!(all.len(), 5); + + let mut buf = String::new(); + let mut rem = Vec::new(); + + // Chunk 1: "hi" + first byte of "你" + append_utf8_safe(&mut buf, &mut rem, &all[..3]); + assert_eq!(buf, "hi"); + assert_eq!(rem.len(), 1); + + // Chunk 2: remaining 2 bytes of "你" + append_utf8_safe(&mut buf, &mut rem, &all[3..]); + assert_eq!(buf, "hi你"); + assert!(rem.is_empty()); + } + + #[test] + fn multiple_split_characters_in_sequence() { + let text = "你好"; + let bytes = text.as_bytes(); // E4 BD A0 E5 A5 BD + + let mut buf = String::new(); + let mut rem = Vec::new(); + + // Split in the middle: first char complete + 1 byte of second + append_utf8_safe(&mut buf, &mut rem, &bytes[..4]); + assert_eq!(buf, "你"); + assert_eq!(rem.len(), 1); + + // Remaining 2 bytes complete second char + append_utf8_safe(&mut buf, &mut rem, &bytes[4..]); + assert_eq!(buf, "你好"); + assert!(rem.is_empty()); + } + + #[test] + fn empty_chunks_are_harmless() { + let mut buf = String::new(); + let mut rem = Vec::new(); + + append_utf8_safe(&mut buf, &mut rem, b""); + assert_eq!(buf, ""); + assert!(rem.is_empty()); + + append_utf8_safe(&mut buf, &mut rem, b"ok"); + assert_eq!(buf, "ok"); + + append_utf8_safe(&mut buf, &mut rem, b""); + assert_eq!(buf, "ok"); + } + + #[test] + fn sse_json_with_chinese_split_at_boundary() { + // Simulates an SSE data line with Chinese content split across chunks + let json_line = "data: {\"text\":\"你好\"}\n\n"; + let bytes = json_line.as_bytes(); + + // Find where "你" starts in the byte stream and split there + let ni_start = bytes.windows(3).position(|w| w == "你".as_bytes()).unwrap(); + let split_point = ni_start + 1; // split inside "你" + + let mut buf = String::new(); + let mut rem = Vec::new(); + + append_utf8_safe(&mut buf, &mut rem, &bytes[..split_point]); + append_utf8_safe(&mut buf, &mut rem, &bytes[split_point..]); + + assert_eq!(buf, json_line); + assert!(rem.is_empty()); + + // Verify the buffer can be parsed as SSE with valid JSON + let data = strip_sse_field(buf.lines().next().unwrap(), "data").unwrap(); + let parsed: serde_json::Value = serde_json::from_str(data).unwrap(); + assert_eq!(parsed["text"], "你好"); + } + + #[test] + fn invalid_bytes_flushed_immediately_not_accumulated() { + // 0xFF is never valid in UTF-8 – it should be replaced immediately, + // not stashed in remainder. + let mut buf = String::new(); + let mut rem = Vec::new(); + + // "hi" + invalid byte + "ok" + append_utf8_safe(&mut buf, &mut rem, b"hi\xFFok"); + assert!( + rem.is_empty(), + "remainder should be empty after invalid byte" + ); + assert!(buf.contains("hi"), "valid prefix must be present"); + assert!(buf.contains("ok"), "valid suffix must be present"); + assert!(buf.contains('\u{FFFD}'), "invalid byte must produce U+FFFD"); + } + + #[test] + fn invalid_byte_in_slow_path_flushed_immediately() { + let mut buf = String::new(); + let mut rem = Vec::new(); + + // Prime remainder with an incomplete sequence (first byte of "你") + append_utf8_safe(&mut buf, &mut rem, &"你".as_bytes()[..1]); + assert_eq!(rem.len(), 1); + + // Next chunk starts with an invalid byte – the stale remainder and the + // invalid byte should both be flushed, not accumulated. + append_utf8_safe(&mut buf, &mut rem, b"\xFFworld"); + assert!(rem.is_empty(), "remainder should be empty"); + assert!( + buf.contains("world"), + "valid data after invalid byte must appear" + ); + } + + #[test] + fn defensive_guard_flushes_oversized_remainder() { + let mut buf = String::new(); + let mut rem = Vec::new(); + + // Manually inject 4 invalid bytes into remainder to trigger the >3 guard. + // This can't happen with well-formed UTF-8, but tests the safety net. + rem.extend_from_slice(b"\x80\x80\x80\x80"); + assert_eq!(rem.len(), 4); + + append_utf8_safe(&mut buf, &mut rem, b"hello"); + // The 4 invalid bytes should have been flushed lossy, then "hello" decoded. + assert!(rem.is_empty(), "remainder must be empty after guard flush"); + assert!( + buf.contains("hello"), + "valid data after guard flush must appear" + ); + // The 4 invalid bytes each produce a U+FFFD + let replacement_count = buf.chars().filter(|&c| c == '\u{FFFD}').count(); + assert_eq!( + replacement_count, 4, + "each invalid byte should produce one U+FFFD" + ); + } }