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.

headroom/proxy/handlers/openai.py

@@ -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

tests/test_openai_codex_routing.py

@@ -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