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.

Author: JerrettDavis

headroom/proxy/helpers.py

@@ -24,7 +24,7 @@
 
 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)

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

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"))