fix: preserve UTF-8 characters across SSE chunks

python/render/client/sse.py

@@ -3,6 +3,7 @@
 This module provides SSE stream parsing functionality for task run events.
 """
 
+import codecs
 import json
 import logging
 from collections.abc import AsyncIterator
@@ -22,13 +23,14 @@ class SSEParser:
     """
 
     def __init__(self) -> None:
+        self._decoder = codecs.getincrementaldecoder("utf-8")(errors="ignore")
         self._buffer = ""
         self._event_data: dict[str, Any] = {}
 
     def feed(self, chunk: bytes) -> list[TaskRunDetails]:
         """Feed a chunk of bytes and return any completed events."""
         results: list[TaskRunDetails] = []
-        self._buffer += chunk.decode("utf-8", errors="ignore")
+        self._buffer += self._decoder.decode(chunk)
 
         while "\n" in self._buffer:
             line, self._buffer = self._buffer.split("\n", 1)

python/render/client/tests/test_sse_unicode.py

@@ -0,0 +1,51 @@
+"""Network chunk boundaries must not alter UTF-8 task results."""
+
+import json
+
+import pytest
+
+from render.client import TaskRunDetails, TaskRunStatus
+from render.client.sse import parse_stream
+from render.client.sse_sync import parse_stream as parse_stream_sync
+
+_RESULT = "café 中文 🚀"
+_DETAILS = TaskRunDetails(
+    id="trn-unicode",
+    task_id="tsk-unicode",
+    status=TaskRunStatus.COMPLETED,
+    results=[_RESULT],
+    input_=[],
+    parent_task_run_id="trn-unicode",
+    root_task_run_id="trn-unicode",
+    retries=0,
+    attempts=[],
+)
+_EVENT = (
+    b"event: task.completed\ndata: "
+    + json.dumps(_DETAILS.to_dict(), ensure_ascii=False).encode()
+    + b"\n\n"
+)
+_START = _EVENT.index(_RESULT.encode())
+_CHUNKS = [
+    [_EVENT[:split], _EVENT[split:]]
+    for split in range(_START + 1, _START + len(_RESULT.encode()))
+] + [[bytes([byte]) for byte in _EVENT]]
+
+
+@pytest.mark.parametrize("chunks", _CHUNKS)
+def test_sync_stream_preserves_split_utf8(chunks):
+    events = list(parse_stream_sync(iter(chunks)))
+    assert len(events) == 1
+    assert events[0].results == [_RESULT]
+
+
+@pytest.mark.parametrize("chunks", _CHUNKS)
+@pytest.mark.asyncio
+async def test_async_stream_preserves_split_utf8(chunks):
+    async def stream():
+        for chunk in chunks:
+            yield chunk
+
+    events = [event async for event in parse_stream(stream())]
+    assert len(events) == 1
+    assert events[0].results == [_RESULT]