mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
[codex] fix(proxy): parse CRLF SSE event terminators (#649)
## Summary - support CRLF (`\r\n\r\n`) SSE event terminators in the byte-buffer parser - parse completed SSE events with `splitlines()` so LF and CRLF line endings are handled consistently - add a regression test for CRLF-terminated SSE events ## Why SSE streams may be emitted with CRLF line endings by HTTP stacks. The existing byte-buffer parser only looked for `\n\n`, so a complete CRLF-terminated event could remain buffered and never reach usage/event parsing. ## Validation - `.venv/bin/pytest tests/test_sse_utf8_split.py tests/test_streaming_usage_parser.py -q` - `.venv/bin/ruff check headroom/proxy/helpers.py tests/test_sse_utf8_split.py` ## Risk Low. The change is isolated to complete-event boundary detection and keeps the existing invalid UTF-8 behavior loud for complete events.
This commit is contained in:
parent
0d458e5d45
commit
4b1c449c73
2 changed files with 33 additions and 9 deletions
|
|
@ -669,10 +669,24 @@ def get_body_too_large_status() -> int:
|
|||
return value
|
||||
|
||||
|
||||
# Sentinel used by the SSE byte-buffer helper to mark events that have no
|
||||
# `event:` line. Per the SSE spec the default event name is "message"; we
|
||||
# return ``None`` so callers can decide whether to apply that default.
|
||||
_SSE_EVENT_TERMINATOR = b"\n\n"
|
||||
# SSE byte-buffer helper supports LF and CRLF event separators. Per the SSE
|
||||
# spec the default event name is "message"; we return ``None`` so callers can
|
||||
# decide whether to apply that default.
|
||||
_SSE_EVENT_TERMINATORS = (b"\n\n", b"\r\n\r\n")
|
||||
|
||||
|
||||
def _find_sse_event_terminator(buf: bytearray) -> tuple[int, int] | None:
|
||||
"""Return the earliest complete SSE event terminator in ``buf``."""
|
||||
matches = [
|
||||
(idx, len(terminator))
|
||||
for terminator in _SSE_EVENT_TERMINATORS
|
||||
if (idx := buf.find(terminator)) != -1
|
||||
]
|
||||
if not matches:
|
||||
return None
|
||||
return min(matches, key=lambda match: match[0])
|
||||
|
||||
|
||||
_SSE_EVENT_LINE_PREFIX = b"event:"
|
||||
_SSE_DATA_LINE_PREFIX = b"data:"
|
||||
|
||||
|
|
@ -718,20 +732,20 @@ def parse_sse_events_from_byte_buffer(
|
|||
multi-byte characters split across TCP reads will corrupt content.
|
||||
"""
|
||||
events: list[tuple[str | None, str]] = []
|
||||
terminator = _SSE_EVENT_TERMINATOR
|
||||
while True:
|
||||
idx = buf.find(terminator)
|
||||
if idx == -1:
|
||||
terminator_match = _find_sse_event_terminator(buf)
|
||||
if terminator_match is None:
|
||||
break
|
||||
idx, terminator_len = terminator_match
|
||||
event_bytes = bytes(buf[:idx])
|
||||
# Drain the event + the trailing terminator from the buffer.
|
||||
del buf[: idx + len(terminator)]
|
||||
del buf[: idx + terminator_len]
|
||||
# Decoding the COMPLETE event must succeed. If it doesn't, the
|
||||
# upstream emitted invalid UTF-8 mid-stream — surface loudly.
|
||||
event_text = event_bytes.decode("utf-8")
|
||||
event_name: str | None = None
|
||||
data_lines: list[str] = []
|
||||
for line in event_text.split("\n"):
|
||||
for line in event_text.splitlines():
|
||||
if not line:
|
||||
continue
|
||||
# SSE comment line — ignored per spec.
|
||||
|
|
|
|||
|
|
@ -85,6 +85,16 @@ def test_cjk_split_across_chunks_preserved() -> None:
|
|||
assert parsed["text"] == "日本語テスト"
|
||||
|
||||
|
||||
def test_crlf_terminated_event_is_parsed() -> None:
|
||||
"""SSE permits CRLF event terminators as well as LF terminators."""
|
||||
buf = bytearray(b'event: message\r\ndata: {"ok": true}\r\n\r\n')
|
||||
|
||||
events = parse_sse_events_from_byte_buffer(buf)
|
||||
|
||||
assert events == [("message", '{"ok": true}')]
|
||||
assert buf == bytearray()
|
||||
|
||||
|
||||
def test_complete_event_with_invalid_utf8_raises_loud() -> None:
|
||||
"""Invalid UTF-8 in a *complete* event surfaces loudly (not silent corruption)."""
|
||||
# Build a complete event whose data field has invalid UTF-8 bytes
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue