fix: handle UTF-8 split across SSE chunks (#135)

Author: Excelius-Wang

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:

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'])