From da18f0be3e4c3cdc03ef078a49aa08aa7811c3d2 Mon Sep 17 00:00:00 2001 From: Kayvan Zahiri <123409429+Kayvan-Zahiri@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:27:03 -0700 Subject: [PATCH] fix: preserve UTF-8 characters across SSE chunks --- python/render/client/sse.py | 4 +- .../render/client/tests/test_sse_unicode.py | 51 +++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) create mode 100644 python/render/client/tests/test_sse_unicode.py diff --git a/python/render/client/sse.py b/python/render/client/sse.py index 2830eb65..16c04040 100644 --- a/python/render/client/sse.py +++ b/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) diff --git a/python/render/client/tests/test_sse_unicode.py b/python/render/client/tests/test_sse_unicode.py new file mode 100644 index 00000000..3249e721 --- /dev/null +++ b/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]