fix(ccr): send Accept: application/json on a buffered stream:false turn (#3102)
## Description Server-side CCR retrieval flips a `stream: true` turn to `stream: false` so the whole upstream reply is in hand before answering. The **body** was rewritten; the client's `Accept: text/event-stream` was **not**. The request that went on the wire therefore contradicted itself — *"answer as JSON"* in the body, *"I only accept SSE"* in the headers. Anthropic's first-party API tolerates that, which is why this never surfaced against it. GitHub Copilot's Anthropic-compatible gateway does not, and answers with a generic `api_error`. That is the reported shape exactly. An OpenCode session's **first** call succeeds — no marker exists yet, so nothing is buffered. The **second** call is the first to carry a redeemable `<<ccr:…>>` marker, so it is the first to be flipped to buffered, and it fails. The reporter's own logs show the correlation: every failed request carries `mutation_reasons=…,ccr_streaming_retrieve_buffered_non_stream`. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) ## Changes Made - `headroom/proxy/handlers/anthropic.py`: when the buffered CCR path flips `stream` to `false`, the outgoing `Accept` header is set to `application/json` to match. The lookup is case-insensitive and **replaces** the existing header rather than appending, so exactly one `Accept` goes upstream. - `headroom/proxy/handlers/openai.py`: the **same fix on the `/v1/responses` buffered path**, which has an identical `stream: false` flip with no matching `Accept`. This handler is a GitHub Copilot path — it calls `apply_copilot_api_auth` — so leaving it would have left the reported bug live on a route the reporter can hit. Found during self-review, not in the original diff. - Same treatment for the Anthropic CCR continuation request, which is non-streaming for the same reason and previously fixed only `Content-Type`. Its header strip is now case-insensitive for `Content-Type` as well, removing a latent duplicate-header path. - `tests/test_buffered_ccr_accept_header.py`: 6 tests — the buffered turn asks for JSON, exactly one `Accept` survives, mixed-case `Accept` is replaced, a client sending no `Accept` still gets one, a non-buffered streaming turn keeps `text/event-stream` untouched, and the OpenAI `/v1/responses` buffered turn asks for JSON too. ## Testing - [x] Unit tests pass (`pytest`) - [x] New tests added for new functionality - [x] Manual testing performed ### Test Output ```text tests/test_buffered_ccr_accept_header.py ...... [100%] 6 passed CCR-adjacent suites on this branch: tests/test_buffered_ccr_accept_header.py, test_buffered_ccr_salvage.py, test_buffered_ccr_grace_window.py, test_anthropic_streaming_ccr_retrieve.py, test_ccr_buffered_stream_signed_thinking.py 41 passed Full suite on this branch: 3 failed, 11216 passed, 581 skipped in 414.81s ``` The 3 failures are pre-existing and environmental, identical to a plain-`main` baseline run on the same machine: no `cargo` installed (`test_no_native_tls_in_wheel_build_tree`), no `codex` CLI (`test_learn/test_integration.py`), and `test_run_server_installs_cancelled_error_filter`, which fails under full-suite ordering on `main` too. ## Real Behavior Proof - Environment: macOS (darwin 25.4.0), Python 3.12.13, worktree off `main` @ `7ef736fb`, `HEADROOM_SKIP_UPSTREAM_CHECK=1` - Exact command / steps: Drove one streaming `/v1/messages` turn through `create_app()` carrying a redeemable `<<ccr:…>>` marker and `headroom_retrieve` in `tools` (so the buffered path engages), with the client sending `Accept: text/event-stream`, and captured the exact headers and body handed to the upstream call. - Observed result: Before — `body.stream=False` sent together with `accept: text/event-stream`, the self-contradicting request. After — `body.stream=False` with `accept: application/json`, and a turn that is not flipped still sends `accept: text/event-stream` unchanged. Reverting only `headroom/proxy/handlers/anthropic.py` fails 3 of the new tests; reverting the OpenAI hunk alone fails the `/v1/responses` test with `['text/event-stream'] != ['application/json']`. Restoring both passes all 6. - Not tested: No live GitHub Copilot gateway call — I have no Copilot credentials here, so the claim that Copilot rejects the contradictory request is inferred from the reporter's logs plus the header mismatch, not observed against their upstream. Confirmation from @mars-peng-lb on a real OpenCode + Copilot session is still wanted before treating #3078 as fully closed. Separately noted while reviewing, **not fixed here**: `_should_buffer_openai_responses_stream_ccr` has no redeemable-marker requirement, so the `/v1/responses` path still buffers on mere tool presence — the #3071/#3092 narrowing was never mirrored from the Anthropic handler. Worth its own issue. ## Runtime Rollout Safety - Rollout-managed feature(s): None — no rollout channel gates this. - Minimum rollout channel: n/a - Stable/default behavior changed: Only on the buffered CCR path, and only the `Accept` header, which is made consistent with the `stream: false` body already being sent. Non-buffered turns are byte-identical, pinned by a test. - Kill switch / disable path: `--no-ccr` / `HEADROOM_NO_CCR` disables the buffered path entirely (see #3082), as does `ccr_handle_responses=False`. - Unsafe override required: No. - Qualification impact: None. - Rollback path: Revert this commit; the buffered path returns to forwarding the client's `Accept` unchanged. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review Closes #3078 Co-authored-by: Tejas Chopra <tejas@Tejass-MacBook-Pro.local> Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Tejas Chopra
@@ -3501,6 +3501,20 @@ class AnthropicHandlerMixin:
|
||||
body_mutation_tracker.mark_mutated(
|
||||
"ccr_streaming_retrieve_buffered_non_stream"
|
||||
)
|
||||
# The body now asks for a non-streaming reply, so the
|
||||
# client's ``Accept: text/event-stream`` no longer describes
|
||||
# the response being requested. Forwarding it unchanged
|
||||
# sends upstream a self-contradicting request: "answer as
|
||||
# JSON" in the body, "I only accept SSE" in the headers.
|
||||
#
|
||||
# Anthropic tolerates that. Stricter Anthropic-compatible
|
||||
# gateways do not: GitHub Copilot's returns a generic
|
||||
# ``api_error``, which is why a session's first call
|
||||
# succeeded and the next one — the first to carry a
|
||||
# redeemable marker, and so the first to be buffered —
|
||||
# failed (#3078).
|
||||
_accept_key = next((k for k in headers if k.lower() == "accept"), "accept")
|
||||
headers[_accept_key] = "application/json"
|
||||
logger.info(
|
||||
f"[{request_id}] CCR: stream:true request has "
|
||||
"headroom_retrieve available; using buffered stream:false "
|
||||
@@ -3903,10 +3917,16 @@ class AnthropicHandlerMixin:
|
||||
body_mutated=True,
|
||||
)
|
||||
)
|
||||
# A continuation is a non-streaming call, so it
|
||||
# needs a matching Accept for the same reason the
|
||||
# buffered flip above does (#3078).
|
||||
ccr_outbound_headers = {
|
||||
**continuation_headers,
|
||||
"content-type": "application/json",
|
||||
k: v
|
||||
for k, v in continuation_headers.items()
|
||||
if k.lower() not in ("accept", "content-type")
|
||||
}
|
||||
ccr_outbound_headers["content-type"] = "application/json"
|
||||
ccr_outbound_headers["accept"] = "application/json"
|
||||
log_outbound_request(
|
||||
forwarder="anthropic_ccr_continuation",
|
||||
method="POST",
|
||||
|
||||
@@ -5616,6 +5616,12 @@ class OpenAIHandlerMixin:
|
||||
if body.get("stream") is not False:
|
||||
body["stream"] = False
|
||||
body_mutation_tracker.mark_mutated("ccr_streaming_retrieve_buffered_non_stream")
|
||||
# Same contradiction as the Anthropic path: the body now asks for a
|
||||
# non-streaming reply while the client's Accept still says SSE. This
|
||||
# handler serves GitHub Copilot (see apply_copilot_api_auth below),
|
||||
# whose gateway is one of the strict ones (#3078).
|
||||
_accept_key = next((k for k in headers if k.lower() == "accept"), "accept")
|
||||
headers[_accept_key] = "application/json"
|
||||
logger.info(
|
||||
f"[{request_id}] CCR: stream:true /v1/responses request has "
|
||||
"headroom_retrieve available; using buffered stream:false "
|
||||
|
||||
@@ -0,0 +1,238 @@
|
||||
"""A buffered CCR turn must not ask for SSE it no longer wants (#3078).
|
||||
|
||||
Server-side retrieval flips a ``stream: true`` turn to ``stream: false`` so the
|
||||
whole reply is in hand before answering. The body was rewritten; the client's
|
||||
``Accept: text/event-stream`` was not, so the request that went on the wire
|
||||
contradicted itself — "answer as JSON" in the body, "I only accept SSE" in the
|
||||
headers.
|
||||
|
||||
Anthropic tolerates that, which is why it never showed up against the first-party
|
||||
API. GitHub Copilot's Anthropic-compatible gateway does not, and answers with a
|
||||
generic ``api_error``. That produced the reported shape exactly: the first call
|
||||
of an OpenCode session succeeds (no marker yet, so no buffering), and the next
|
||||
one — the first to carry a redeemable marker, and so the first to be flipped —
|
||||
fails.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
fastapi = pytest.importorskip("fastapi")
|
||||
httpx = pytest.importorskip("httpx")
|
||||
|
||||
from fastapi.testclient import TestClient # noqa: E402
|
||||
|
||||
from headroom.cache.backends import InMemoryBackend # noqa: E402
|
||||
from headroom.cache.compression_store import ( # noqa: E402
|
||||
get_compression_store,
|
||||
reset_compression_store,
|
||||
)
|
||||
from headroom.ccr.tool_injection import create_ccr_tool_definition # noqa: E402
|
||||
from headroom.proxy.server import ProxyConfig, create_app # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _store():
|
||||
reset_compression_store()
|
||||
get_compression_store(backend=InMemoryBackend())
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
reset_compression_store()
|
||||
|
||||
|
||||
def _drive(*, with_marker: bool, accept: str | None) -> dict[str, object]:
|
||||
"""Run one streaming turn; report the body `stream` and headers sent upstream."""
|
||||
marker = get_compression_store().store(
|
||||
original=json.dumps({"earlier": "tool output"}),
|
||||
compressed="{}",
|
||||
original_item_count=400,
|
||||
)
|
||||
config = ProxyConfig(
|
||||
optimize=False,
|
||||
cache_enabled=False,
|
||||
rate_limit_enabled=False,
|
||||
memory_enabled=False,
|
||||
ccr_inject_tool=True,
|
||||
ccr_handle_responses=True,
|
||||
ccr_context_tracking=False,
|
||||
image_optimize=False,
|
||||
)
|
||||
seen: dict[str, object] = {}
|
||||
app = create_app(config)
|
||||
with TestClient(app) as client:
|
||||
proxy = client.app.state.proxy
|
||||
|
||||
async def _fake_retry(method, url, headers, body, stream=False, **kwargs): # noqa: ANN001
|
||||
sent = json.loads(body) if isinstance(body, (str, bytes)) else body
|
||||
seen["stream"] = sent.get("stream")
|
||||
seen["headers"] = dict(headers or {})
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"id": "msg_1",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": "claude-sonnet-4-6",
|
||||
"content": [{"type": "text", "text": "ok"}],
|
||||
"stop_reason": "end_turn",
|
||||
"usage": {
|
||||
"input_tokens": 10,
|
||||
"output_tokens": 5,
|
||||
"cache_read_input_tokens": 0,
|
||||
"cache_creation_input_tokens": 0,
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
proxy._retry_request = _fake_retry # type: ignore[assignment]
|
||||
|
||||
# A turn that is *not* flipped never reaches `_retry_request` — the plain
|
||||
# streaming path has its own upstream call — so capture that one too.
|
||||
async def _fake_stream(url, headers, body, *args, **kwargs): # noqa: ANN001
|
||||
from fastapi.responses import StreamingResponse
|
||||
|
||||
sent = json.loads(body) if isinstance(body, (str, bytes)) else body
|
||||
seen["stream"] = sent.get("stream")
|
||||
seen["headers"] = dict(headers or {})
|
||||
|
||||
async def _gen():
|
||||
yield b'event: message_stop\ndata: {"type":"message_stop"}\n\n'
|
||||
|
||||
return StreamingResponse(_gen(), media_type="text/event-stream")
|
||||
|
||||
proxy._stream_response = _fake_stream # type: ignore[assignment]
|
||||
headers = {"x-api-key": "test-key", "anthropic-version": "2023-06-01"}
|
||||
if accept is not None:
|
||||
headers["accept"] = accept
|
||||
content = f"go <<ccr:{marker}>>" if with_marker else "go"
|
||||
client.post(
|
||||
"/v1/messages",
|
||||
json={
|
||||
"model": "claude-sonnet-4-6",
|
||||
"max_tokens": 64,
|
||||
"stream": True,
|
||||
"tools": [create_ccr_tool_definition("anthropic")],
|
||||
"messages": [{"role": "user", "content": content}],
|
||||
},
|
||||
headers=headers,
|
||||
)
|
||||
return seen
|
||||
|
||||
|
||||
def _accepts(headers: dict) -> list[str]:
|
||||
return [v for k, v in headers.items() if k.lower() == "accept"]
|
||||
|
||||
|
||||
def test_buffered_turn_asks_for_json() -> None:
|
||||
seen = _drive(with_marker=True, accept="text/event-stream")
|
||||
|
||||
# Precondition: this turn really was flipped to buffered.
|
||||
assert seen["stream"] is False
|
||||
assert _accepts(seen["headers"]) == ["application/json"] # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_buffered_turn_leaves_exactly_one_accept_header() -> None:
|
||||
"""Replaced, never appended — two Accept values is its own bug."""
|
||||
seen = _drive(with_marker=True, accept="text/event-stream")
|
||||
|
||||
assert len(_accepts(seen["headers"])) == 1 # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_accept_header_is_replaced_regardless_of_casing() -> None:
|
||||
"""Header names are case-insensitive; the SSE value must not survive."""
|
||||
seen = _drive(with_marker=True, accept="TEXT/EVENT-STREAM")
|
||||
|
||||
values = _accepts(seen["headers"]) # type: ignore[arg-type]
|
||||
assert values == ["application/json"]
|
||||
assert not any("event-stream" in v.lower() for v in values)
|
||||
|
||||
|
||||
def test_buffered_turn_without_a_client_accept_still_asks_for_json() -> None:
|
||||
seen = _drive(with_marker=True, accept=None)
|
||||
|
||||
assert seen["stream"] is False
|
||||
assert _accepts(seen["headers"]) == ["application/json"] # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_a_streaming_turn_keeps_its_sse_accept() -> None:
|
||||
"""No marker means no flip, so nothing about the request should change."""
|
||||
seen = _drive(with_marker=False, accept="text/event-stream")
|
||||
|
||||
assert seen["stream"] is not False
|
||||
assert _accepts(seen["headers"]) == ["text/event-stream"] # type: ignore[arg-type]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# The same flip exists on the OpenAI Responses path, which serves Copilot
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _drive_responses(*, accept: str) -> dict[str, object]:
|
||||
"""Run one streaming /v1/responses turn and report what went upstream."""
|
||||
from headroom.ccr import CCR_TOOL_NAME
|
||||
|
||||
config = ProxyConfig(
|
||||
optimize=False,
|
||||
cache_enabled=False,
|
||||
rate_limit_enabled=False,
|
||||
memory_enabled=False,
|
||||
ccr_inject_tool=True,
|
||||
ccr_handle_responses=True,
|
||||
ccr_context_tracking=False,
|
||||
image_optimize=False,
|
||||
)
|
||||
seen: dict[str, object] = {}
|
||||
app = create_app(config)
|
||||
with TestClient(app) as client:
|
||||
proxy = client.app.state.proxy
|
||||
|
||||
async def _fake_retry(method, url, headers, body, stream=False, **kwargs): # noqa: ANN001
|
||||
sent = json.loads(body) if isinstance(body, (str, bytes)) else body
|
||||
seen["stream"] = sent.get("stream")
|
||||
seen["headers"] = dict(headers or {})
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"id": "resp_1",
|
||||
"object": "response",
|
||||
"model": "gpt-4o",
|
||||
"status": "completed",
|
||||
"output": [
|
||||
{
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "ok"}],
|
||||
}
|
||||
],
|
||||
"usage": {"input_tokens": 5, "output_tokens": 2, "total_tokens": 7},
|
||||
},
|
||||
)
|
||||
|
||||
proxy._retry_request = _fake_retry # type: ignore[assignment]
|
||||
client.post(
|
||||
"/v1/responses",
|
||||
json={
|
||||
"model": "gpt-4o",
|
||||
"stream": True,
|
||||
# Responses tool defs are flat, not nested under "function".
|
||||
"tools": [{"type": "function", "name": CCR_TOOL_NAME}],
|
||||
"input": "go",
|
||||
},
|
||||
headers={
|
||||
"authorization": "Bearer test-key",
|
||||
"accept": accept,
|
||||
"content-type": "application/json",
|
||||
},
|
||||
)
|
||||
return seen
|
||||
|
||||
|
||||
def test_responses_buffered_turn_asks_for_json() -> None:
|
||||
"""This handler serves GitHub Copilot, the gateway that rejects the mismatch."""
|
||||
seen = _drive_responses(accept="text/event-stream")
|
||||
|
||||
assert seen.get("stream") is False, "precondition: the turn must be buffered"
|
||||
assert _accepts(seen["headers"]) == ["application/json"] # type: ignore[arg-type]
|
||||
assert len(_accepts(seen["headers"])) == 1 # type: ignore[arg-type]
|
||||
Reference in New Issue
Block a user