fix(proxy/streaming): tolerate malformed content in _response_to_sse (#2481)

## Description

`StreamingMixin._response_to_sse` rebuilds an Anthropic SSE stream from
a buffered response dict. It iterated the content and read usage with no
type guards:

```python
for idx, block in enumerate(response.get("content", [])):
    if block.get("type") == "text":
        ...
...
"usage": {"output_tokens": response.get("usage", {}).get("output_tokens", 0)},
```

`response` here is provider- and reconstruction-controlled.
`.get("content", [])` only falls back when the key is absent, so a
present-but-null `content` returns `None` and `enumerate(None)` raises
`TypeError`. A non-list `content` (e.g. a bare string) makes
`block.get(...)` raise `AttributeError`, and a null element inside the
list hits the same `AttributeError`. `response.get("usage",
{}).get(...)` breaks the same way on `usage: null`.

This matters because the Anthropic buffered CCR path calls it inside an
`except ValueError` guard only:

```python
try:
    sse_events = self._response_to_sse(resp_json, "anthropic")
except ValueError as sse_err:
    ...
```

A `TypeError`/`AttributeError` from any of the shapes above escapes that
guard and 500s the streamed request. The sibling
`_record_ccr_feedback_from_response` in the same class already guards
`content` for list-ness and skips non-dict blocks, so this closes the
asymmetry.

## Fix

Coerce `content` to a list before iterating (non-list becomes empty),
skip any non-dict block, and coerce a non-dict `usage` to `{}` before
reading `output_tokens`. Well-formed responses render byte-for-byte as
before.

## Type of Change

- [x] 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
- [ ] Code refactoring (no functional changes)

## Changes Made

- `headroom/proxy/handlers/streaming.py`: list-guard `content`, skip
non-dict blocks, and dict-guard `usage` in `_response_to_sse`.
- `tests/test_sse_thinking_blocks.py`: regression rendering responses
with null/non-list content, a null block element, and null usage, plus a
check that a valid block alongside a null element still renders.

## 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_thinking_blocks.py -q
14 passed

# with the fix reverted, the new test fails with
# TypeError: 'NoneType' object is not iterable

$ uvx ruff@0.15.17 check headroom/proxy/handlers/streaming.py tests/test_sse_thinking_blocks.py
All checks passed!
$ uvx mypy@1.20.2 --ignore-missing-imports headroom/proxy/handlers/streaming.py
Success: no issues found in 1 source file
```

## Real Behavior Proof

- Environment: Windows 11, Python 3.12, project venv (`uv sync --extra
proxy`), `uvx ruff@0.15.17` / `uvx mypy@1.20.2`, pytest in the venv.
- Exact command / steps: called
`StreamingMixin()._response_to_sse(response, "anthropic")` with four
malformed bodies (`content: null`, `content: "not-a-list"`, `content:
[null, {text}]`, `usage: null`); then reverted `streaming.py` and re-ran
the same inputs.
- Observed result: with the fix each body produces a well-formed SSE
envelope (message_start ... message_stop) and the valid block alongside
a null element still emits its text_delta; with the fix reverted the
`content: null` body raises `TypeError: 'NoneType' object is not
iterable` and the others raise `AttributeError`. Ran against the actual
module via `tests/test_sse_thinking_blocks.py`.
- Not tested: a live upstream returning a malformed buffered response
end to end through the CCR path.

## 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
This commit is contained in:
Abhay Singh 2026-07-22 18:41:00 +05:30 committed by GitHub
parent 7524854da7
commit 77b26c093c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 49 additions and 3 deletions

View file

@ -565,8 +565,17 @@ class StreamingMixin:
}
events.append(f"event: message_start\ndata: {json.dumps(msg_start)}\n\n".encode())
# Content blocks
for idx, block in enumerate(response.get("content", [])):
# Content blocks. `content` is provider/reconstruction-controlled, so a
# present-but-null value or a non-list would crash `enumerate`, and a
# non-dict element would crash `block.get(...)`. Guard both, matching the
# sibling `_record_ccr_feedback_from_response` below. This is reached from
# a call site (anthropic.py buffered CCR path) that only catches
# ValueError, so an unguarded TypeError/AttributeError would 500 the
# streamed request.
content = response.get("content")
for idx, block in enumerate(content if isinstance(content, list) else []):
if not isinstance(block, dict):
continue
# content_block_start
if block.get("type") == "text":
block_start = {
@ -680,10 +689,13 @@ class StreamingMixin:
msg_delta_payload["stop_reason"] = response["stop_reason"]
if "stop_details" in response:
msg_delta_payload["stop_details"] = response["stop_details"]
usage = response.get("usage")
if not isinstance(usage, dict):
usage = {}
msg_delta = {
"type": "message_delta",
"delta": msg_delta_payload,
"usage": {"output_tokens": response.get("usage", {}).get("output_tokens", 0)},
"usage": {"output_tokens": usage.get("output_tokens", 0)},
}
events.append(f"event: message_delta\ndata: {json.dumps(msg_delta)}\n\n".encode())

View file

@ -324,6 +324,40 @@ def test_response_to_sse_emits_unknown_content_block_verbatim() -> None:
assert not any(ev["type"] == "content_block_delta" for ev in events)
def test_response_to_sse_tolerates_malformed_content_and_usage() -> None:
# `_response_to_sse` runs on provider/reconstruction-controlled JSON and is
# reached from a call site (anthropic.py buffered CCR path) that only catches
# ValueError, so a present-but-null `content`/`usage` or a non-dict block must
# not raise a TypeError/AttributeError that would 500 the streamed request.
# The sibling `_record_ccr_feedback_from_response` guards `content` the same
# way. Each of these once crashed the unguarded loop.
parser = _Parser()
for response in (
{"content": None, "usage": {"output_tokens": 5}},
{"content": "not-a-list"},
{"content": [None, {"type": "text", "text": "hi"}]},
{"content": [], "usage": None},
):
sse_text = b"".join(parser._response_to_sse(response, "anthropic")).decode("utf-8")
# Always a well-formed envelope, regardless of the malformed body.
assert "event: message_start" in sse_text
assert "event: message_stop" in sse_text
# The one valid block alongside a null element is still rendered.
sse_text = b"".join(
parser._response_to_sse({"content": [None, {"type": "text", "text": "hi"}]}, "anthropic")
).decode("utf-8")
events = _sse_events(sse_text)
text_deltas = [
ev
for ev in events
if ev["type"] == "content_block_delta" and ev["delta"].get("type") == "text_delta"
]
assert len(text_deltas) == 1
assert text_deltas[0]["delta"]["text"] == "hi"
def test_response_to_sse_emits_server_tool_use_without_delta() -> None:
parser = _Parser()
server_tool_use = {