From 4b1c449c73af6bb50857c175643a0e95166cc8de Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Yui=28=E3=82=86=E3=81=84=29?=
 <132864240+cnYui@users.noreply.github.com>
Date: Thu, 11 Jun 2026 11:16:00 +0900
Subject: [PATCH] [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.
---
 headroom/proxy/helpers.py    | 32 +++++++++++++++++++++++---------
 tests/test_sse_utf8_split.py | 10 ++++++++++
 2 files changed, 33 insertions(+), 9 deletions(-)

diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py
index 1930ec188..baa0e868e 100644
--- a/headroom/proxy/helpers.py
+++ b/headroom/proxy/helpers.py
@@ -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.
diff --git a/tests/test_sse_utf8_split.py b/tests/test_sse_utf8_split.py
index 989506738..256a44f6e 100644
--- a/tests/test_sse_utf8_split.py
+++ b/tests/test_sse_utf8_split.py
@@ -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