From 0cbc0e8e5435cd8d743ae537cdbaa70787bfc5b4 Mon Sep 17 00:00:00 2001 From: Abhay Singh Date: Mon, 20 Jul 2026 10:48:47 +0530 Subject: [PATCH] fix(proxy/openai): replay incremental events in buffered Responses SSE (#2410) (#2415) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description Fixes #2410. When a streaming `/v1/responses` request has `headroom_retrieve` available, Headroom forces a non-streaming (`stream:false`) upstream call so CCR retrieval can be resolved server-side, then reconstructs the complete response as SSE for the client. GitHub Copilot returns 200 with real output tokens, but OpenCode shows no assistant response. Root cause: `_openai_responses_to_sse` emitted only two events — `response.created` and `response.completed`: ```python created_response = {**response, "status": "in_progress", "output": []} events = [("response.created", created_response), ("response.completed", response)] ``` Clients that read the whole answer off the terminal `response.completed` event work, but OpenCode and the Vercel AI SDK render output from the **incremental** item/text events (`response.output_item.added`, `response.output_text.delta`, ...). With those absent, the SDK displays nothing. ## Fix Reconstruct the real Responses event sequence: ``` response.created (status in_progress, empty output) response.in_progress for each output item: response.output_item.added (message items start with empty content) for each message content part: response.content_part.added (text blanked) response.output_text.delta (the text) response.output_text.done response.content_part.done response.output_item.done (full item) response.completed (full response) data: [DONE] ``` Non-message items (reasoning, function_call, ...) get `output_item.added` + `output_item.done` with the full item. Every event carries a contiguous `sequence_number`. The terminal `response.completed` still carries the full response, so clients that key off it are unaffected; clients that stream now receive the deltas they need. ## 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`: rewrite `_openai_responses_to_sse` to replay the incremental output-item/content-part/output-text events between `response.created`/`response.in_progress` and `response.completed`. - `tests/test_openai_responses_buffered_sse.py`: new test asserting the incremental `output_text.delta` (visible text), the per-item sequence for message vs non-message items, the empty-output case, and contiguous sequence numbers. ## 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 $ uvx ruff@0.15.17 check headroom/proxy/handlers/openai.py tests/test_openai_responses_buffered_sse.py All checks passed! $ uvx mypy@1.20.2 --ignore-missing-imports headroom/proxy/handlers/openai.py # no errors in the changed file # _openai_responses_to_sse is a pure module-level function, so I ran the new # tests against the real code in the project venv (uv sync): 3 passed. ``` ## 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`. - Exact command / steps: fed the real `_openai_responses_to_sse` a completed response with a reasoning item and a message item whose content is `output_text: "Hello world"`, plus an empty-output response and a function_call-only response. - Observed result: the stream now contains `response.output_text.delta` with `"Hello world"` at `output_index=1, content_index=0`, wrapped by `content_part.added/done` and `output_item.added/done`, with the reasoning and function_call items emitted as `output_item.added/done` and preserved whole; `response.created`/`in_progress` carry empty output while `response.completed` carries the full output; sequence numbers are `0..n`. Ran against the actual module. - Not tested: a live OpenCode -> Copilot Responses round trip; the added tests assert the event stream directly. ## 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 ## Additional Notes `_openai_responses_to_sse` is a pure function, so I verified the fix against the real code in the venv (output above) in addition to the unit tests. This mirrors the incremental replay the Anthropic buffered path already does in `StreamingMixin._response_to_sse` (content_block_start/delta/stop), bringing the Responses buffered-CCR path to the same fidelity. --- headroom/proxy/handlers/openai.py | 82 +++++++++++---- tests/test_openai_responses_buffered_sse.py | 109 ++++++++++++++++++++ 2 files changed, 170 insertions(+), 21 deletions(-) create mode 100644 tests/test_openai_responses_buffered_sse.py diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index 39926a430..f7bbd8eec 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -957,31 +957,71 @@ def _dedup_responses_output_items( def _openai_responses_to_sse(response: dict[str, Any]) -> list[bytes]: - """Convert a complete Responses API JSON body into a minimal SSE stream. + """Convert a complete Responses API JSON body into an SSE stream. - Used only for the buffered-CCR path: the client asked for - ``stream: true`` but we forced a non-streaming upstream call so CCR - retrieval could be resolved server-side. This reconstructs just enough - of the real event sequence (``response.created`` + ``response.completed``) - for Responses API clients that key off the terminal event's full - response object — it does not replay incremental output-item/text - deltas. Mirrors the equivalent simplification in - ``StreamingMixin._response_to_sse`` for the Anthropic buffered path. + Used only for the buffered-CCR path: the client asked for ``stream: true`` + but we forced a non-streaming upstream call so CCR retrieval could be + resolved server-side. We then have to replay the response as SSE. + + Some Responses clients read the whole answer off the terminal + ``response.completed`` event, but others (OpenCode / the Vercel AI SDK) + render output only from the *incremental* item/text events and show nothing + when they are absent (#2410). So reconstruct the real event sequence: + ``response.created`` -> ``response.in_progress`` -> per output item + (``response.output_item.added``, and for message items the + ``response.content_part.added`` / ``response.output_text.delta`` / + ``response.output_text.done`` / ``response.content_part.done`` sequence) -> + ``response.output_item.done`` -> ``response.completed`` -> ``[DONE]``. """ - created_response = {**response, "status": "in_progress", "output": []} events: list[bytes] = [] - for seq, (event_type, event_response) in enumerate( - ( - ("response.created", created_response), - ("response.completed", response), - ) - ): - payload = { - "type": event_type, - "sequence_number": seq, - "response": event_response, - } + seq = 0 + + def _emit(event_type: str, extra: dict[str, Any]) -> None: + nonlocal seq + payload = {"type": event_type, "sequence_number": seq, **extra} events.append(f"event: {event_type}\ndata: {json.dumps(payload)}\n\n".encode()) + seq += 1 + + output_items = response.get("output") or [] + created_response = {**response, "status": "in_progress", "output": []} + _emit("response.created", {"response": created_response}) + _emit("response.in_progress", {"response": created_response}) + + for out_idx, item in enumerate(output_items): + if not isinstance(item, dict): + continue + item_id = item.get("id", f"item_{out_idx}") + + # ``output_item.added`` carries the item shell; message content streams + # via the content-part events below, so start it empty there. + if item.get("type") == "message": + added_item = {k: v for k, v in item.items() if k != "content"} + added_item["content"] = [] + else: + added_item = item + _emit("response.output_item.added", {"output_index": out_idx, "item": added_item}) + + content = item.get("content") + if item.get("type") == "message" and isinstance(content, list): + for c_idx, part in enumerate(content): + if not isinstance(part, dict): + continue + loc = {"item_id": item_id, "output_index": out_idx, "content_index": c_idx} + if part.get("type") in ("output_text", "text"): + text = part.get("text", "") or "" + _emit("response.content_part.added", {**loc, "part": {**part, "text": ""}}) + if text: + _emit("response.output_text.delta", {**loc, "delta": text}) + _emit("response.output_text.done", {**loc, "text": text}) + _emit("response.content_part.done", {**loc, "part": part}) + else: + # Non-text part (e.g. refusal): add + done with the full part. + _emit("response.content_part.added", {**loc, "part": part}) + _emit("response.content_part.done", {**loc, "part": part}) + + _emit("response.output_item.done", {"output_index": out_idx, "item": item}) + + _emit("response.completed", {"response": response}) events.append(b"data: [DONE]\n\n") return events diff --git a/tests/test_openai_responses_buffered_sse.py b/tests/test_openai_responses_buffered_sse.py new file mode 100644 index 000000000..a3271f5dc --- /dev/null +++ b/tests/test_openai_responses_buffered_sse.py @@ -0,0 +1,109 @@ +"""Regression for #2410: the buffered-CCR Responses -> SSE reconstruction must +replay the incremental output-item/text events, not just response.created + +response.completed, so AI-SDK / OpenCode clients render the output.""" + +from __future__ import annotations + +import json + +from headroom.proxy.handlers.openai import _openai_responses_to_sse + + +def _parse(events: list[bytes]) -> list[dict]: + out: list[dict] = [] + for e in events: + s = e.decode() + if s.startswith("data: [DONE]"): + out.append({"type": "[DONE]"}) + continue + out.append(json.loads(s.split("data: ", 1)[1])) + return out + + +def test_responses_sse_replays_incremental_output_text() -> None: + resp = { + "id": "resp_1", + "object": "response", + "status": "completed", + "model": "gpt-5.3-codex", + "output": [ + {"type": "reasoning", "id": "rs_1", "summary": []}, + { + "type": "message", + "id": "msg_1", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "Hello world", "annotations": []}], + }, + ], + "usage": {"input_tokens": 10, "output_tokens": 3}, + } + + parsed = _parse(_openai_responses_to_sse(resp)) + types = [p["type"] for p in parsed] + + assert types[0] == "response.created" + assert types[1] == "response.in_progress" + assert types[-2] == "response.completed" + assert types[-1] == "[DONE]" + + # The visible assistant text is streamed as an output_text.delta. + deltas = [p for p in parsed if p["type"] == "response.output_text.delta"] + assert len(deltas) == 1 + assert deltas[0]["delta"] == "Hello world" + assert deltas[0]["output_index"] == 1 + assert deltas[0]["content_index"] == 0 + + # The message item gets the full content-part sequence; the reasoning item + # gets add/done with no content parts. + assert types.count("response.output_item.added") == 2 + assert types.count("response.output_item.done") == 2 + assert "response.content_part.added" in types + assert "response.output_text.done" in types + assert "response.content_part.done" in types + + # created / in_progress carry an empty output; completed carries the full one. + created = next(p for p in parsed if p["type"] == "response.created") + assert created["response"]["output"] == [] + completed = next(p for p in parsed if p["type"] == "response.completed") + assert completed["response"]["output"] == resp["output"] + + # Sequence numbers are contiguous from 0. + seqs = [p["sequence_number"] for p in parsed if p["type"] != "[DONE]"] + assert seqs == list(range(len(seqs))) + + +def test_responses_sse_empty_output_still_valid() -> None: + resp = {"id": "resp_2", "status": "completed", "output": [], "usage": {}} + types = [p["type"] for p in _parse(_openai_responses_to_sse(resp))] + assert types == ["response.created", "response.in_progress", "response.completed", "[DONE]"] + + +def test_responses_sse_non_message_item_added_and_done() -> None: + resp = { + "id": "resp_3", + "status": "completed", + "output": [ + { + "type": "function_call", + "id": "fc_1", + "call_id": "c1", + "name": "grep", + "arguments": "{}", + } + ], + "usage": {}, + } + parsed = _parse(_openai_responses_to_sse(resp)) + types = [p["type"] for p in parsed] + assert types == [ + "response.created", + "response.in_progress", + "response.output_item.added", + "response.output_item.done", + "response.completed", + "[DONE]", + ] + # The function_call item is preserved whole on added and done. + done = next(p for p in parsed if p["type"] == "response.output_item.done") + assert done["item"]["name"] == "grep"