From c7e536b864a95a71c1aa0c5b6714e38ae0015cd2 Mon Sep 17 00:00:00 2001
From: Excelius-Wang <57819425+Excelius-Wang@users.noreply.github.com>
Date: Mon, 24 Aug 2026 00:21:02 +0800
Subject: [PATCH] fix: handle UTF-8 split across SSE chunks

---
 src/acp/_sse.py        |  6 +++++-
 tests/http/test_sse.py | 11 +++++++++++
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/src/acp/_sse.py b/src/acp/_sse.py
index 455a2d2..c3b5a7f 100644
--- a/src/acp/_sse.py
+++ b/src/acp/_sse.py
@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import codecs
 import json
 from typing import TYPE_CHECKING, Any
 
@@ -62,11 +63,12 @@ async def parse_sse_stream(chunks: AsyncIterator[bytes]) -> AsyncIterator[dict[s
     Multi-line ``data:`` fields are concatenated with newlines per the spec. A
     blank line dispatches the buffered event.
     """
+    decoder = codecs.getincrementaldecoder("utf-8")()
     buffer = ""
     data_lines: list[str] = []
 
     async for chunk in chunks:
-        buffer += chunk.decode("utf-8")
+        buffer += decoder.decode(chunk)
         while "\n" in buffer:
             line, buffer = buffer.split("\n", 1)
             line = line.rstrip("\r")
@@ -78,6 +80,8 @@ async def parse_sse_stream(chunks: AsyncIterator[bytes]) -> AsyncIterator[dict[s
             elif not line.startswith(":"):
                 _append_field(line, data_lines)
 
+    decoder.decode(b"", final=True)
+
     # Flush a trailing event with no terminating blank line.
     event = _decode_event(data_lines)
     if event is not None:
diff --git a/tests/http/test_sse.py b/tests/http/test_sse.py
index 2003c1b..0365829 100644
--- a/tests/http/test_sse.py
+++ b/tests/http/test_sse.py
@@ -35,6 +35,17 @@ async def test_parse_multiple_events_split_across_chunks() -> None:
     assert events == [{"id": 1}, {"id": 2}]
 
 
+@pytest.mark.asyncio
+async def test_parse_multibyte_utf8_split_across_chunks() -> None:
+    frame = 'data: {"text":"你好"}\n\n'.encode()
+    split_at = frame.index("你".encode()) + 1
+    stream = _aiter([frame[:split_at], frame[split_at:]])
+
+    events = [event async for event in parse_sse_stream(stream)]
+
+    assert events == [{"text": "你好"}]
+
+
 @pytest.mark.asyncio
 async def test_parse_ignores_comments_and_other_fields() -> None:
     stream = _aiter([b': keepalive\n\nevent: message\ndata: {"id":7}\n\n'])