fix: handle UTF-8 multi-byte characters split across stream chunk boundaries (#1923)

* fix: handle UTF-8 multi-byte characters split across stream chunk boundaries

Replace String::from_utf8_lossy with append_utf8_safe in all four SSE
streaming paths. When a multi-byte UTF-8 character (e.g. Chinese, emoji)
is split across TCP chunk boundaries, from_utf8_lossy silently replaces
the incomplete halves with U+FFFD (�). This caused intermittent garbled
output in Claude Code when using the Copilot reverse proxy, because the
format conversion streams reconstruct SSE events from the corrupted buffer.

The new append_utf8_safe function preserves incomplete trailing bytes in
a remainder buffer and merges them with the next chunk before decoding,
ensuring characters are never split during UTF-8 conversion.

Fixes: intermittent U+FFFD replacement characters in Claude Code output
via Copilot proxy (not reproducible with direct Copilot connections like
opencode because they pass through raw bytes without format conversion).

* style: fix cargo fmt formatting in UTF-8 boundary tests

---------

Co-authored-by: Cod1ng <codingts@gmail.com>
Co-authored-by: encodets <encodets@gmail.com>

src-tauri/src/proxy/providers/streaming.rs

@@ -93,6 +93,7 @@ pub fn create_anthropic_sse_stream<E: std::error::Error + Send + 'static>(
 ) -> impl Stream<Item = Result<Bytes, std::io::Error>> + Send {
     async_stream::stream! {
         let mut buffer = String::new();
+        let mut utf8_remainder: Vec<u8> = 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<E: std::error::Error + Send + 'static>(
         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::<String>();
+
+        // 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"
+        );
+    }
 }

src-tauri/src/proxy/providers/streaming_responses.rs

@@ -101,6 +101,7 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
 ) -> impl Stream<Item = Result<Bytes, std::io::Error>> + Send {
     async_stream::stream! {
         let mut buffer = String::new();
+        let mut utf8_remainder: Vec<u8> = Vec::new();
         let mut message_id: Option<String> = None;
         let mut current_model: Option<String> = None;
         let mut has_sent_message_start = false;
@@ -118,8 +119,7 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
         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);
 
                     // 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::<String>();
+
+        assert!(
+            merged.contains("你好世界"),
+            "expected '你好世界' in output, got replacement chars (U+FFFD)"
+        );
+        assert!(
+            !merged.contains('\u{FFFD}'),
+            "output must not contain U+FFFD replacement characters"
+        );
+    }
 }

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<u8> = 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") {

src-tauri/src/proxy/response_processor.rs

@@ -568,6 +568,7 @@ pub fn create_logged_passthrough_stream(
 ) -> impl Stream<Item = Result<Bytes, std::io::Error>> + Send {
     async_stream::stream! {
         let mut buffer = String::new();
+        let mut utf8_remainder: Vec<u8> = 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") {

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<u8>, new_bytes: &[u8]) {
+    // Build the byte slice to decode: prepend any leftover bytes from previous chunk.
+    let (owned, bytes): (Option<Vec<u8>>, &[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"
+        );
+    }
 }