fix(proxy): adapt 200 SSE upstream replies on buffered /v1/responses instead of 502 (#2622)

## Description

The buffered HTTP `/v1/responses` path (`_buffered_ccr_operation` in
`headroom/proxy/handlers/openai.py`) assumed every upstream reply to a
`stream: false` request is JSON. Some OpenAI-compatible upstreams answer
with a valid `200 text/event-stream` body carrying Responses API events.
`response.json()` raised `JSONDecodeError`, which is not in the narrow
usage-extraction catch (`KeyError, TypeError, AttributeError`), so it
escaped to the outer handler and the **successful** upstream reply was
converted into a generic `502 proxy_error` — the client loses the
response and typically retries, duplicating paid calls.

The fix classifies the upstream reply at the ingestion boundary by its
declared `Content-Type` (the SSE spec's own discriminator) instead of
parsing by expectation:

- **200 SSE with a terminal `response.completed` event** → the complete
response object is reassembled from that event
(`_openai_responses_from_sse`, the inverse of the existing
`_openai_responses_to_sse`) and swapped in as a synthesized
`application/json` response *before any parsing happens*. Everything
downstream — usage extraction, CCR retrieval handling, memory-tool
handling — runs unmodified.
- **200 SSE without a recognizable terminal event** → the successful
upstream body is forwarded to the client unchanged (sanitized headers)
rather than fabricating a 502. Adapt only when the adaptation is
provably faithful; otherwise pass through.
- **Everything else** (normal JSON replies, non-200s) → byte-identical
pre-existing behavior.

Deliberately *not* done: widening the `except` clause (would leave
`resp_json` unbound and break the downstream pipeline) and body sniffing
(the declared media type is trusted; a mislabeled body keeps today's
behavior).

Closes #2613

## 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/openai.py`: new module-level helper
`_openai_responses_from_sse()` (SSE-spec framing: blank-line event
separation, multi-line `data:` joining, `\r` tolerance, at most one
stripped space, `[DONE]` skipped; returns the terminal event's
`response` object or `None`), placed next to its inverse
`_openai_responses_to_sse()`.
- `headroom/proxy/handlers/openai.py::_buffered_ccr_operation()`:
content-type dispatch for 200 replies immediately after the upstream
response (and after wire-debug capture, so debug logs keep the true
upstream bytes) — adapt SSE→JSON when a terminal event exists, pass
through unchanged when it doesn't.
- `tests/test_openai_codex_routing.py`: two new handler-level tests (see
below).

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
# Both new tests watched failing BEFORE the fix with the exact issue signature:
#   ERROR headroom.proxy:openai.py [req-1] OpenAI responses request failed: JSONDecodeError: Expecting value: line 1 column 1 (char 0)
#   assert 502 == 200

$ pytest tests/test_openai_codex_routing.py -q
24 passed in 2.08s

$ pytest tests/test_openai_codex_routing.py tests/test_ccr_response_handler_openai_responses.py tests/test_codex_responses_passthrough_bytes.py -q
38 passed, 1 warning in 13.80s

$ pytest tests/test_output_shaper_responses.py tests/test_codex_responses_waste_signals.py tests/test_codex_openai_contract_parity.py tests/test_openai_beta_session_sticky.py tests/test_openai_max_completion_tokens.py tests/test_openai_response_cache_key.py tests/test_litellm_openai_passthrough.py -q
61 passed, 1 warning

$ ruff check headroom/proxy/handlers/openai.py tests/test_openai_codex_routing.py
All checks passed!

$ mypy headroom/proxy/handlers/openai.py
Success: no issues found in 1 source file
```

New tests:

- `test_handle_openai_responses_non_stream_adapts_sse_upstream` — 200
SSE with `response.completed` → client gets 200 `application/json` with
the reassembled response.
-
`test_handle_openai_responses_non_stream_passes_through_unparseable_sse`
— 200 SSE with no terminal event → client gets 200 with the body
unchanged, never a 502.

## Real Behavior Proof

- Environment: macOS 26.5 (arm64), Python 3.12 via `uv`, headroom from
source (editable install). Local fake OpenAI-compatible upstream
(`http.server`) that answers every `POST` with `200 text/event-stream`
containing a `response.completed` event and `data: [DONE]` — the
upstream behavior reported in the issue. Proxy started with
`OPENAI_TARGET_API_URL=http://127.0.0.1:9302 headroom proxy --port
<port>`.
- Exact command / steps: same-session A/B against real proxy processes —
identical upstream and identical request, only the checked-out revision
changed:

  ```bash
curl -s -w "\nHTTP_STATUS=%{http_code} CONTENT_TYPE=%{content_type}\n" \
    -X POST http://127.0.0.1:<port>/v1/responses \
-H "content-type: application/json" -H "authorization: Bearer sk-test" \
    -d '{"model":"gpt-5.4","stream":false,"input":"hello"}'
  ```

- Observed result: unpatched `main` converts the successful upstream
reply into the issue's 502; this branch returns the complete response as
JSON. Full captures:

  **Before (unpatched `main`, port 8794):**

  ```
{"error":{"message":"An error occurred while processing your request.
Please try again.","type":"server_error","code":"proxy_error"}}
  HTTP_STATUS=502
  ```

  **After (this branch, port 8795):**

  ```
{"id": "resp_sse_repro", "object": "response", "status": "completed",
"model": "gpt-5.4", "output": [{"type": "message", "id": "msg_1",
"role": "assistant", "content": [{"type": "output_text", "text": "hello
from sse upstream"}]}], "usage": {"input_tokens": 2, "output_tokens":
1}}
  HTTP_STATUS=200 CONTENT_TYPE=application/json
  ```

- Not tested: a wild third-party SSE-answering upstream (the repro uses
a local stub shaped per the issue report); the buffered-stream-CCR
variant of this path against a live upstream (unit-tested only);
Windows.

## 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
- [x] I did **not** edit `CHANGELOG.md` — it is generated by
release-please from my Conventional Commit PR title (a CI guard enforces
this)

## Additional Notes

- Documentation checklist item is N/A — internal proxy behavior fix, no
documented surface changes.
- Known residual (pre-existing, out of this issue's scope): a
**non-200** upstream reply with a non-JSON body (an SSE error stream, a
gateway HTML error page) still follows the old `JSONDecodeError → 502`
path, blurring a meaningful upstream error into a generic 502. This PR
deliberately adapts only declared-SSE **200** replies, where reassembly
from `response.completed` is provably faithful. Happy to file the
non-200 case as a follow-up issue if maintainers want it tracked.
This commit is contained in:
Pragadeesh 2026-08-13 11:46:12 -05:00 committed by GitHub
parent d6d121e399
commit d76fce04a3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 148 additions and 0 deletions

View file

@ -1182,6 +1182,47 @@ def _openai_responses_to_sse(response: dict[str, Any]) -> list[bytes]:
return events
def _openai_responses_from_sse(sse_text: str) -> dict[str, Any] | None:
"""Reassemble the final Responses API JSON body from an SSE stream.
Inverse of ``_openai_responses_to_sse``, for upstreams that answer a
``stream: false`` request with a valid ``200 text/event-stream`` body
(#2613). The terminal ``response.completed`` event carries the complete
response object, so no delta accumulation is needed. Returns ``None``
when no terminal event is present the caller forwards the raw body
unchanged in that case.
"""
completed: dict[str, Any] | None = None
data_lines: list[str] = []
def _consume(lines: list[str]) -> None:
nonlocal completed
if not lines:
return
data_str = "\n".join(lines)
if data_str == "[DONE]":
return
try:
data = json.loads(data_str)
except json.JSONDecodeError:
return
if isinstance(data, dict) and data.get("type") == "response.completed":
response = data.get("response")
if isinstance(response, dict):
completed = response
for raw_line in sse_text.split("\n"):
line = raw_line.rstrip("\r")
if not line:
_consume(data_lines)
data_lines = []
elif line.startswith("data:"):
# Per the SSE spec, strip at most one leading space after the colon.
data_lines.append(line[5:].removeprefix(" "))
_consume(data_lines)
return completed
def _output_shaping_holdout_fraction() -> float:
from headroom.proxy import runtime_env
@ -5674,6 +5715,45 @@ class OpenAIHandlerMixin:
total_latency = (time.time() - start_time) * 1000
# Some OpenAI-compatible upstreams answer a ``stream: false``
# request with a valid 200 SSE body carrying Responses API
# events (#2613). Reassemble the terminal response JSON so
# the buffered path keeps working — previously
# ``response.json()`` raised JSONDecodeError past the
# narrow usage-extraction catch below and the successful
# upstream reply surfaced as a 502 proxy_error.
upstream_content_type = response.headers.get("content-type", "")
if (
response.status_code == 200
and "text/event-stream" in upstream_content_type.lower()
):
completed_json = _openai_responses_from_sse(response.text)
if completed_json is None:
# No terminal event to adapt — forward the
# successful upstream body unchanged rather than
# fabricating a 502.
return Response(
content=response.content,
status_code=response.status_code,
headers=_sanitize_forwarded_response_headers(response.headers),
)
adapted_headers = {
k: v
for k, v in response.headers.items()
if k.lower()
not in (
"content-type",
"content-length",
"content-encoding",
"transfer-encoding",
)
}
response = httpx.Response(
status_code=response.status_code,
content=json.dumps(completed_json).encode(),
headers={**adapted_headers, "content-type": "application/json"},
)
total_input_tokens = original_tokens # fallback
output_tokens = 0
cache_read_tokens = 0

View file

@ -468,6 +468,74 @@ def test_handle_openai_responses_routes_api_key_auth_direct_to_openai(monkeypatc
assert response.status_code == 200
def test_handle_openai_responses_non_stream_adapts_sse_upstream(monkeypatch):
"""A ``stream: false`` request whose upstream replies ``200
text/event-stream`` must be adapted to the terminal response JSON, not
converted into a 502 proxy_error (#2613)."""
import httpx
sse = (
b"event: response.completed\n"
b'data: {"type":"response.completed","response":{"id":"resp_sse_repro",'
b'"output":[],"usage":{"input_tokens":2,"output_tokens":1}}}\n\n'
)
class _SSEUpstreamHandler(_DummyOpenAIHandler):
async def _retry_request(self, method, url, headers, body, **kwargs):
self.captured_request = (method, url, headers, body)
return httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=sse,
)
request = _build_request(
{"model": "gpt-5.4", "stream": False, "input": "hello"},
{"Authorization": "Bearer sk-test"},
)
handler = _SSEUpstreamHandler()
monkeypatch.setattr("headroom.tokenizers.get_tokenizer", lambda model: _DummyTokenizer())
response = anyio.run(handler.handle_openai_responses, request)
assert response.status_code == 200, response.body
payload = json.loads(response.body)
assert payload["id"] == "resp_sse_repro"
assert response.headers["content-type"].startswith("application/json")
def test_handle_openai_responses_non_stream_passes_through_unparseable_sse(monkeypatch):
"""A 200 SSE upstream body with no recognizable terminal response event
must be forwarded as-is never converted into a 502 (#2613)."""
import httpx
sse = b"event: response.weird\ndata: not-json\n\n"
class _SSEUpstreamHandler(_DummyOpenAIHandler):
async def _retry_request(self, method, url, headers, body, **kwargs):
self.captured_request = (method, url, headers, body)
return httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=sse,
)
request = _build_request(
{"model": "gpt-5.4", "stream": False, "input": "hello"},
{"Authorization": "Bearer sk-test"},
)
handler = _SSEUpstreamHandler()
monkeypatch.setattr("headroom.tokenizers.get_tokenizer", lambda model: _DummyTokenizer())
response = anyio.run(handler.handle_openai_responses, request)
assert response.status_code == 200, response.body
assert response.body == sse
assert response.headers["content-type"] == "text/event-stream"
def test_handle_openai_responses_stream_skips_python_compression(monkeypatch):
"""PR-C5: Python no longer compresses /v1/responses (Rust handles it
natively). The streaming forward path must still fire only the