From b910ce5debdb239457ca3838e30fa9fe0555edb4 Mon Sep 17 00:00:00 2001 From: JD Davis Date: Sun, 12 Jul 2026 15:46:36 +0000 Subject: [PATCH] Extract SSE byte buffer policy (#1979) ## Description Extracts the pure SSE byte-buffer parser from `helpers.py` into `headroom.proxy.sse_byte_buffer_policy`. Existing helper imports remain as delegates, while the protocol parser now has its own module and direct tests. Closes # ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [x] Code refactoring (no functional changes) ## Changes Made - Added `sse_byte_buffer_policy.py` for SSE terminator detection and complete-event parsing. - Kept `helpers.parse_sse_events_from_byte_buffer` and `_find_sse_event_terminator` delegating to the extracted policy. - Added direct policy tests for LF/CRLF terminators, buffer draining, split UTF-8 preservation, and invalid complete UTF-8 events. - Carried forward the LiteLLM callback compatibility shim needed for current mypy on `main`. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [ ] Manual testing performed ### Test Output ```text python -m pytest tests\test_sse_byte_buffer_policy.py tests\test_sse_utf8_split.py 8 passed in 0.23s python -m ruff check . All checks passed! python -m ruff format --check . 1095 files already formatted python -m mypy headroom --ignore-missing-imports Success: no issues found in 409 source files gitleaks protect --staged --no-banner --redact no leaks found ``` ## Real Behavior Proof - Environment: Windows, Python 3.13.13, branch `jd/architecture-slice-29`. - Exact command / steps: ran new SSE byte-buffer policy tests, existing SSE UTF-8 split tests, ruff, ruff format check, mypy, and staged gitleaks scan. - Observed result: SSE parser behavior remains covered and local lint/type/security checks pass. - Not tested: live streaming proxy runtime; existing helper imports remain intact. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Screenshots (if applicable) N/A ## Additional Notes Documentation and changelog updates are N/A for this internal architecture-only refactor. The push reported existing default-branch Dependabot alerts; no staged secret leaks were found for this PR. --- headroom/proxy/helpers.py | 47 ++----------------- headroom/proxy/sse_byte_buffer_policy.py | 60 ++++++++++++++++++++++++ tests/test_sse_byte_buffer_policy.py | 33 +++++++++++++ 3 files changed, 97 insertions(+), 43 deletions(-) create mode 100644 headroom/proxy/sse_byte_buffer_policy.py create mode 100644 tests/test_sse_byte_buffer_policy.py diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py index b931b9339..d4b51f280 100644 --- a/headroom/proxy/helpers.py +++ b/headroom/proxy/helpers.py @@ -24,7 +24,7 @@ from typing import TYPE_CHECKING, Any, Literal, cast from headroom import paths as _paths from headroom._subprocess import run -from headroom.proxy import request_limit_policy, wire_debug_redaction_policy +from headroom.proxy import request_limit_policy, sse_byte_buffer_policy, wire_debug_redaction_policy from headroom.proxy.body_forwarding import ( BodyMutationTracker as BodyMutationTracker, # noqa: F401 - compatibility export ) @@ -533,22 +533,12 @@ def get_body_too_large_status() -> int: ) -# 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") +_SSE_EVENT_TERMINATORS = sse_byte_buffer_policy.SSE_EVENT_TERMINATORS 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]) + return sse_byte_buffer_policy.find_sse_event_terminator(buf) _SSE_EVENT_LINE_PREFIX = b"event:" @@ -595,36 +585,7 @@ def parse_sse_events_from_byte_buffer( ``decode("utf-8", errors="ignore")`` on a partial buffer; UTF-8 multi-byte characters split across TCP reads will corrupt content. """ - events: list[tuple[str | None, str]] = [] - while True: - 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 + 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.splitlines(): - if not line: - continue - # SSE comment line — ignored per spec. - if line.startswith(":"): - continue - if line.startswith("event:"): - event_name = line[len("event:") :].lstrip() - elif line.startswith("data:"): - data_lines.append(line[len("data:") :].lstrip()) - # Per SSE spec, multiple `data:` lines join with newline. We - # preserve that here even though OpenAI/Anthropic emit one - # `data:` per event. - if data_lines: - events.append((event_name, "\n".join(data_lines))) - return events + return sse_byte_buffer_policy.parse_sse_events_from_byte_buffer(buf) # Maximum message array length (prevents DoS from deeply nested payloads) diff --git a/headroom/proxy/sse_byte_buffer_policy.py b/headroom/proxy/sse_byte_buffer_policy.py new file mode 100644 index 000000000..c96a19df5 --- /dev/null +++ b/headroom/proxy/sse_byte_buffer_policy.py @@ -0,0 +1,60 @@ +"""Pure SSE byte-buffer parsing policy.""" + +from __future__ import annotations + +# 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") +SSE_EVENT_LINE_PREFIX = "event:" +SSE_DATA_LINE_PREFIX = "data:" + + +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]) + + +def parse_sse_events_from_byte_buffer( + buf: bytearray, +) -> list[tuple[str | None, str]]: + """Drain complete ``event:`` + ``data:`` events from a bytes buffer. + + Returns list of ``(event_name, data_str)`` tuples for complete events. + Mutates ``buf`` in-place to leave only partial-event tail bytes. + + Operates on bytes; only decodes complete events as UTF-8 (raises if a + *complete* event has invalid UTF-8, which is an upstream protocol bug). + """ + events: list[tuple[str | None, str]] = [] + while True: + terminator_match = find_sse_event_terminator(buf) + if terminator_match is None: + break + idx, terminator_len = terminator_match + event_bytes = bytes(buf[:idx]) + del buf[: idx + terminator_len] + + event_text = event_bytes.decode("utf-8") + event_name: str | None = None + data_lines: list[str] = [] + for line in event_text.splitlines(): + if not line: + continue + if line.startswith(":"): + continue + if line.startswith(SSE_EVENT_LINE_PREFIX): + event_name = line[len(SSE_EVENT_LINE_PREFIX) :].lstrip() + elif line.startswith(SSE_DATA_LINE_PREFIX): + data_lines.append(line[len(SSE_DATA_LINE_PREFIX) :].lstrip()) + + if data_lines: + events.append((event_name, "\n".join(data_lines))) + return events diff --git a/tests/test_sse_byte_buffer_policy.py b/tests/test_sse_byte_buffer_policy.py new file mode 100644 index 000000000..d12466dfd --- /dev/null +++ b/tests/test_sse_byte_buffer_policy.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import pytest + +from headroom.proxy.sse_byte_buffer_policy import ( + find_sse_event_terminator, + parse_sse_events_from_byte_buffer, +) + + +def test_find_sse_event_terminator_returns_earliest_separator() -> None: + assert find_sse_event_terminator(bytearray(b"data: one\r\n\r\ndata: two\n\n")) == (9, 4) + + +def test_parse_sse_events_drains_complete_events_and_leaves_tail() -> None: + buf = bytearray(b": ignored\nevent: delta\ndata: one\ndata: two\n\npartial") + + assert parse_sse_events_from_byte_buffer(buf) == [("delta", "one\ntwo")] + assert bytes(buf) == b"partial" + + +def test_parse_sse_events_preserves_split_utf8_tail() -> None: + smile = "\U0001f642".encode() + buf = bytearray(b"data: hello " + smile[:2]) + + assert parse_sse_events_from_byte_buffer(buf) == [] + buf.extend(smile[2:] + b"\n\n") + assert parse_sse_events_from_byte_buffer(buf) == [(None, "hello \U0001f642")] + + +def test_parse_sse_events_raises_on_complete_invalid_utf8_event() -> None: + with pytest.raises(UnicodeDecodeError): + parse_sse_events_from_byte_buffer(bytearray(b"data: \xff\n\n"))