From f1c34d336cf35db341153c1c65e8c15219398340 Mon Sep 17 00:00:00 2001 From: Parideboy Date: Thu, 13 Aug 2026 18:46:55 +0200 Subject: [PATCH] fix(proxy/anthropic): don't buffer a CCR stream when passthrough discards the stream flip (#2953) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description Fixes #2952. Since `dc163bcd` (#2254), a Claude Code session with extended thinking on dies from the turn where the first signed `thinking` block enters history: ``` API Error: API returned an empty or malformed response (HTTP 200) — check for a proxy or gateway intercepting the request ``` `select_outbound_body` forwards the client's original bytes whenever the body carries a signed `thinking` / `redacted_thinking` block, and it decides that **before** it looks at `body_mutated` — so every edit the handler made is discarded. The buffered-CCR path depends on exactly such an edit: it sets `body["stream"] = False` (`anthropic.py:3073-3078`) so the reply arrives as one JSON document it can scan for `headroom_retrieve` calls. With the flip discarded, upstream streams, `response.json()` fails, SSE resynthesis is skipped, and the client is handed a 200 it cannot read. From the reporter's `proxy.log`, the turn that breaks — note `body_bytes` equals the inbound `content_length` byte for byte, and `source=passthrough` despite two recorded mutations: ``` CCR: stream:true request has headroom_retrieve available; using buffered stream:false upstream request event=outbound_request forwarder=anthropic_messages body_bytes=132017 body_mutated=true mutation_reasons=structural_diff_vs_original,ccr_streaming_retrieve_buffered_non_stream source=passthrough PERF ... msgs=6 tok_saved=3575 tool_saved=16064 tok_out=0 total_ms=9623 ``` This is a different failure from #2251 (a 400 from Anthropic). Signed thinking blocks still leave as original bytes here, so that fix is untouched. ## 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 Primary fix, `headroom/proxy/handlers/anthropic.py`: - Gate `buffered_stream_ccr` on the outbound body actually being ours to change, via a new `outbound_body_is_client_bytes()` predicate that mirrors the passthrough branch. Those turns take the plain streaming path instead, which is the coherent outcome: the injected retrieve tool is itself a discarded mutation there, so the model was never going to see it. One INFO line records the choice. Three follow-on defenses, each of which independently kept the failure alive or invisible: - A non-JSON 200 on the buffered path now logs at WARNING (it was DEBUG, which is why nothing in `proxy.log` looked wrong) and the upstream SSE is relayed to the client verbatim, instead of falling through to a plain `Response` that `_BufferedCCRResponse` can only turn into a bare `event: error` once its 1 s keepalive has committed headers. - The semantic cache no longer stores a body that did not parse as JSON, and drops the stored `content-type` on the hit path. The cache key has no `stream` component, so a cached SSE body was replayed to buffered callers for the full 3600 s TTL — that replay is the request the reporter actually saw the error on (a 2 ms `PERF ... transforms=none` cache hit). - `select_outbound_body` now reports the mutations passthrough discarded (`dropped_mutations` / `dropped_mutation_reasons`), and `log_outbound_request` logs them as `event=outbound_body_mutations_dropped` at WARNING. Without it, `PERF` reports savings and tool injections that never reached the wire and nothing contradicts it. `prepare_outbound_body_bytes` keeps its two-value shape, so existing callers are unchanged. ## 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 New `tests/test_ccr_buffered_stream_signed_thinking.py` covers the gate (signed-thinking history takes the streaming path and still leaves as `stream: true`; the same request without thinking blocks still takes the buffered path with `stream: false`), the SSE relay, and the cache guards. The relay case is parametrized on upstream latency because the failure only reaches its worst form past the 1 s keepalive, where a plain `Response` has no `body_iterator` left to forward. Verified red before green — with the source changes stashed and the tests in place: ```text FAILED tests/test_ccr_buffered_stream_signed_thinking.py::test_signed_thinking_history_skips_the_buffered_ccr_path[True-True] FAILED tests/test_ccr_buffered_stream_signed_thinking.py::test_buffered_ccr_relays_an_unexpected_sse_reply_and_does_not_cache_it[prompt] FAILED tests/test_ccr_buffered_stream_signed_thinking.py::test_buffered_ccr_relays_an_unexpected_sse_reply_and_does_not_cache_it[past-keepalive] FAILED tests/test_ccr_buffered_stream_signed_thinking.py::test_cache_hit_never_replays_a_foreign_content_type ========================= 4 failed, 1 passed in 8.88s ========================= ``` With the fix applied: ```text $ python -m pytest tests/test_proxy_byte_faithful_forwarding.py tests/test_ccr_buffered_stream_signed_thinking.py -q 55 passed in 11.40s $ python -m pytest tests/test_compression_cache.py tests/test_ccr_inline_resolve_handlers.py \ tests/test_ccr_sqlite_backend.py tests/test_anthropic_stage_timings.py \ tests/test_backend_nonstreaming_cache_metrics.py tests/test_cache_mode_cold_start.py \ tests/test_cache_breakpoint_diagnostics.py -q 87 passed, 1 skipped in 17.57s $ python -m ruff check . All checks passed! $ python -m mypy headroom --ignore-missing-imports --python-version 3.13 Found 12 errors in 3 files (checked 517 source files) # all 12 pre-existing in headroom/ccr/mcp_server.py and headroom/memory/mcp_server.py # (local mcp package version); none in the four files this PR touches ``` ## Real Behavior Proof - Environment: headroom 0.35.0-dev source checkout, Python 3.13.11, Windows 11, Claude Code CLI 2.1.228 against api.anthropic.com (claude-opus-5), proxy run as `headroom proxy --port 8787 --memory --code-aware --mode token` - Exact command / steps: reproduced from the reporter's `~/.headroom/logs/proxy.log` — request `hr_1786551079_000005` shows `source=passthrough` with `body_bytes` identical to the inbound `content_length` while `mutation_reasons` contains `ccr_streaming_retrieve_buffered_non_stream`, then `tok_out=0`; request `hr_1786551089_000006` is the 2 ms `transforms=none` semantic-cache hit that returned the poisoned SSE body to a caller asking for JSON. The preceding turn (`...000004`, no thinking block yet) was `source=canonical` with `tok_out=341`. Then: pytest suites above, with the red/green stash comparison - Observed result: with the gate in place the thinking-bearing turn goes down `_stream_response` and the forwarded body still says `"stream": true`, matching the bytes passthrough will send; a buffered turn whose upstream answers with SSE reaches the client as a stream and writes nothing to the semantic cache; a cache entry can no longer hand a caller a content-type from a differently-shaped request - Not tested: a live end-to-end Claude Code session against Anthropic with the patched proxy (the reproduction here is the reporter's proxy log plus handler-level tests); `/v1/responses`, which has the same `stream`-flip pattern (`openai.py:5479-5487`) but carries `input` rather than `messages`, so `has_signed_thinking_blocks` never fires there and I left it alone ## 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 Deliberately out of scope, worth a separate issue: option (b) of #2251 — a canonical re-serialization that preserves signed blocks byte-for-byte so compression and tool injection survive a thinking-bearing history rather than being silently dropped. #2251 reports a 400 even on a no-op re-encode whose only transform was `tool_search_deferral`, which hints the signature covers `tools` too, so getting it wrong would re-break every multi-turn thinking session. That needs validating against the live API, not guessing. Until then the new WARNING at least makes the dropped work visible. Also unfixed by design: the savings accounting itself. A passthrough turn still books `tok_saved` / `tool_saved` in `PERF` for bytes that never shipped; correcting the numbers is a wider change than this bug needs. Co-authored-by: Claude Opus 5 Co-authored-by: JD Davis --- headroom/proxy/body_forwarding.py | 53 ++++- headroom/proxy/handlers/anthropic.py | 111 ++++++++- headroom/proxy/helpers.py | 16 ++ headroom/proxy/server.py | 7 +- ...est_ccr_buffered_stream_signed_thinking.py | 219 ++++++++++++++++++ tests/test_proxy_byte_faithful_forwarding.py | 109 ++++++++- 6 files changed, 504 insertions(+), 11 deletions(-) create mode 100644 tests/test_ccr_buffered_stream_signed_thinking.py diff --git a/headroom/proxy/body_forwarding.py b/headroom/proxy/body_forwarding.py index f93c68af9..4648fce6c 100644 --- a/headroom/proxy/body_forwarding.py +++ b/headroom/proxy/body_forwarding.py @@ -32,6 +32,15 @@ class OutboundBody: content: bytes source: OutboundBodySource + #: True when byte-faithful passthrough won over a mutated body, so every + #: edit the handler made to ``body`` was discarded before the wire. Callers + #: that gated a downstream decision on their own mutation (for example + #: flipping ``stream`` to False to buffer a reply) MUST consult this — the + #: request upstream actually sees is the client's original one. + dropped_mutations: bool = False + #: The ``BodyMutationTracker`` reasons discarded alongside it, when the + #: caller supplied them. Empty when the caller passed no reason list. + dropped_mutation_reasons: tuple[str, ...] = () def get_python_forwarder_mode() -> PythonForwarderMode: @@ -104,11 +113,23 @@ def select_outbound_body( original_body_bytes: bytes | None, body_mutated: bool, forwarder_mode: PythonForwarderMode | None = None, + mutation_reasons: list[str] | None = None, ) -> OutboundBody: - """Select the exact bytes to forward upstream.""" + """Select the exact bytes to forward upstream. + + ``mutation_reasons`` is optional and only used for reporting: when the + signed-thinking passthrough overrides a mutated body, the discarded reasons + are echoed back on the result so the call site can log what never made it + upstream instead of silently claiming the edit landed. + """ mode = forwarder_mode if forwarder_mode is not None else get_python_forwarder_mode() if original_body_bytes is not None and has_signed_thinking_blocks(body): - return OutboundBody(content=original_body_bytes, source="passthrough") + return OutboundBody( + content=original_body_bytes, + source="passthrough", + dropped_mutations=body_mutated, + dropped_mutation_reasons=tuple(mutation_reasons or ()) if body_mutated else (), + ) if mode == "legacy_json_kwarg": content = json.dumps(body, separators=(", ", ": "), ensure_ascii=True).encode("utf-8") @@ -125,12 +146,38 @@ def prepare_outbound_body_bytes( original_body_bytes: bytes | None, body_mutated: bool, forwarder_mode: PythonForwarderMode | None = None, + mutation_reasons: list[str] | None = None, ) -> tuple[bytes, OutboundBodySource]: - """Compatibility tuple wrapper around :func:`select_outbound_body`.""" + """Compatibility tuple wrapper around :func:`select_outbound_body`. + + Keeps the two-value shape its existing callers unpack. Call + :func:`select_outbound_body` directly when you need the dropped-mutation + reporting. + """ outbound = select_outbound_body( body=body, original_body_bytes=original_body_bytes, body_mutated=body_mutated, forwarder_mode=forwarder_mode, + mutation_reasons=mutation_reasons, ) return outbound.content, outbound.source + + +def outbound_body_is_client_bytes( + *, + body: dict[str, Any], + original_body_bytes: bytes | None, +) -> bool: + """Return whether the wire body will be the client's original bytes. + + A handler that changes ``body`` to steer its own upstream call — the + ``stream`` flip that buys a buffered reply is the load-bearing case — has to + know that the signed-thinking passthrough will throw that change away and + send the client's bytes verbatim. Asking before acting is cheaper than + discovering it from a reply in the wrong wire format. + + Mirrors the first branch of :func:`select_outbound_body`; the forwarder mode + is deliberately not consulted because that branch overrides it too. + """ + return original_body_bytes is not None and has_signed_thinking_blocks(body) diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py index 5426d0cf5..c08da0fa1 100644 --- a/headroom/proxy/handlers/anthropic.py +++ b/headroom/proxy/handlers/anthropic.py @@ -73,6 +73,20 @@ def _strip_index_from_content_blocks(content: Any) -> None: _strip_index_from_content_blocks(block.get("content")) +def _looks_like_sse_response(response: httpx.Response) -> bool: + """Return whether an upstream reply is a Server-Sent Events stream. + + Trusts the declared content-type first and falls back to sniffing the + leading bytes for an SSE field, because a gateway in front of Anthropic may + relay the stream under a vaguer type. + """ + content_type = (response.headers.get("content-type") or "").lower() + if "text/event-stream" in content_type: + return True + head = response.content[:64].lstrip() + return head.startswith(b"event:") or head.startswith(b"data:") + + class AnthropicHandlerMixin: """Mixin providing Anthropic API handler methods for HeadroomProxy.""" @@ -1054,6 +1068,11 @@ class AnthropicHandlerMixin: response_headers = dict(cached.response_headers) response_headers.pop("content-encoding", None) response_headers.pop("content-length", None) + # Drop the stored content-type too. Starlette lets an + # explicit header win over ``media_type``, so keeping the + # producing request's type would let a cache entry hand this + # caller a wire format it never asked for (#2952). + response_headers.pop("content-type", None) # Unit 4: release the pre-upstream semaphore on cache # hit — no upstream call will happen. @@ -3111,13 +3130,37 @@ class AnthropicHandlerMixin: ccr_response_handler_enabled = bool( self.ccr_response_handler and getattr(ccr_handler_config, "enabled", True) ) - buffered_stream_ccr = bool( + # A body carrying signed thinking blocks leaves as the client's + # original bytes (see ``select_outbound_body``), which throws + # away every edit made here — including the ``stream`` flip + # below. Taking the buffered path anyway asks upstream for a + # stream:true reply and then tries to read it as buffered JSON: + # the parse fails, SSE resynthesis is skipped, and the client + # gets a 200 with no usable body (#2952). The retrieve tool is + # itself an injected (and equally discarded) mutation on these + # turns, so the plain streaming path is the coherent choice. + from headroom.proxy.body_forwarding import outbound_body_is_client_bytes + + outbound_locked_to_client_bytes = outbound_body_is_client_bytes( + body=body, + original_body_bytes=original_body_bytes, + ) + wants_buffered_stream_ccr = bool( stream and ccr_response_handler_enabled and self._has_headroom_retrieve_tool( tools if tools is not None else body.get("tools") ) ) + buffered_stream_ccr = ( + wants_buffered_stream_ccr and not outbound_locked_to_client_bytes + ) + if wants_buffered_stream_ccr and outbound_locked_to_client_bytes: + logger.info( + f"[{request_id}] CCR: signed thinking blocks force byte-faithful " + "passthrough, so a stream:false flip could not reach upstream; " + "using the plain streaming path instead of buffered retrieval" + ) if buffered_stream_ccr: if body.get("stream") is not False: body["stream"] = False @@ -3400,9 +3443,23 @@ class AnthropicHandlerMixin: try: resp_json = response.json() except (json.JSONDecodeError, ValueError) as e: - logger.debug( - f"[{request_id}] Failed to parse response JSON for CCR handling: {e}" - ) + # DEBUG is right for the buffered non-stream path, where + # an unparseable body is just "no CCR handling". On the + # buffered-stream path it means the reply came back in a + # wire format we did not ask for, and every downstream + # step (retrieval, SSE resynthesis, usage accounting) + # silently no-ops — that has to be visible (#2952). + if buffered_stream_ccr: + logger.warning( + f"[{request_id}] CCR: buffered stream:false request got a " + f"non-JSON {response.status_code} reply " + f"(content-type={response.headers.get('content-type')!r}): {e}" + ) + else: + logger.debug( + f"[{request_id}] Failed to parse response JSON for CCR " + f"handling: {e}" + ) # CCR Response Handling: Handle headroom_retrieve tool calls automatically if ( @@ -3730,7 +3787,13 @@ class AnthropicHandlerMixin: # Cache response under the SAME key it was looked up by: # cache_lookup_messages is the raw pre-mutation snapshot, not # the live (compressed/hooked) `messages` (#327). - if self.cache and response.status_code == 200: + # ``resp_json`` is None when the reply did not parse as + # JSON — an SSE stream, most often. Caching those bytes + # poisons the entry for every later caller that shares + # the key: the cache key has no ``stream`` component, so + # a buffered request would be answered with a stream it + # cannot read (#2952). + if self.cache and response.status_code == 200 and resp_json is not None: await self.cache.set( cache_lookup_messages, model, @@ -3893,6 +3956,44 @@ class AnthropicHandlerMixin: f"[{request_id}] Security response scan error: {sec_err}" ) + if ( + buffered_stream_ccr + and response.status_code == 200 + and not resp_json + and _looks_like_sse_response(response) + ): + # Upstream streamed instead of buffering, so there is + # nothing to resynthesize — but the client asked for a + # stream and this already is one. Relay it verbatim + # rather than falling through to a plain Response the + # _BufferedCCRResponse wrapper can only turn into a bare + # error event (#2952). + logger.warning( + f"[{request_id}] CCR: relaying the upstream SSE reply verbatim; " + "server-side retrieval was skipped for this turn" + ) + relay_headers = { + k: v + for k, v in response_headers.items() + if k.lower() + not in ( + "content-encoding", + "content-length", + "transfer-encoding", + "content-type", + ) + } + relayed_sse = response.content + + async def _upstream_sse_relay(): + yield relayed_sse + + return StreamingResponse( + _upstream_sse_relay(), + media_type="text/event-stream", + headers=relay_headers, + ) + if buffered_stream_ccr and response.status_code == 200 and resp_json: sse_headers = { k: v diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py index 1092c58a5..b9cce8a23 100644 --- a/headroom/proxy/helpers.py +++ b/headroom/proxy/helpers.py @@ -327,12 +327,18 @@ def log_outbound_request( mutation_reasons: list[str], request_id: str | None, source: str, + dropped_mutation_reasons: tuple[str, ...] | list[str] | None = None, ) -> None: """Structured log line for every outbound forwarder call. Per realignment build constraints: every cache-affecting decision is logged. Never includes ``Authorization``/``x-api-key`` content or full body bytes. + + ``dropped_mutation_reasons`` records edits that byte-faithful passthrough + discarded before the wire. That is a WARNING, not a detail: the line above + reports the transforms Headroom *decided* on, and without this the operator + reads savings and injections that the upstream never saw. """ logger.info( "event=outbound_request forwarder=%s method=%s path=%s body_bytes=%d " @@ -346,6 +352,16 @@ def log_outbound_request( source, request_id or "", ) + if dropped_mutation_reasons: + logger.warning( + "event=outbound_body_mutations_dropped forwarder=%s source=%s " + "dropped_mutation_reasons=%s request_id=%s (signed thinking blocks force " + "byte-faithful passthrough, so these body edits did NOT reach upstream)", + forwarder, + source, + ",".join(dropped_mutation_reasons), + request_id or "", + ) def count_cache_breakpoints( diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py index 1c9fae65e..bd2dc3acb 100644 --- a/headroom/proxy/server.py +++ b/headroom/proxy/server.py @@ -2128,16 +2128,18 @@ class HeadroomProxy( construct their body from scratch, so canonical serialization is correct and original bytes do not exist). """ - from headroom.proxy.body_forwarding import prepare_outbound_body_bytes + from headroom.proxy.body_forwarding import select_outbound_body from headroom.proxy.helpers import log_outbound_request last_error = None reasons = list(mutation_reasons or []) - outbound_bytes, source = prepare_outbound_body_bytes( + outbound = select_outbound_body( body=body, original_body_bytes=original_body_bytes, body_mutated=body_mutated, + mutation_reasons=reasons, ) + outbound_bytes, source = outbound.content, outbound.source outbound_headers = {**headers, "content-type": "application/json"} log_outbound_request( @@ -2149,6 +2151,7 @@ class HeadroomProxy( mutation_reasons=reasons, request_id=request_id, source=source, + dropped_mutation_reasons=outbound.dropped_mutation_reasons, ) post_kwargs: dict = {"content": outbound_bytes, "headers": outbound_headers} diff --git a/tests/test_ccr_buffered_stream_signed_thinking.py b/tests/test_ccr_buffered_stream_signed_thinking.py new file mode 100644 index 000000000..b90a599d2 --- /dev/null +++ b/tests/test_ccr_buffered_stream_signed_thinking.py @@ -0,0 +1,219 @@ +"""Buffered-CCR streaming vs. byte-faithful passthrough (issue #2952). + +The buffered-CCR path is the one place the Anthropic handler changes the +request *for its own benefit*: it flips ``stream`` to False so the reply comes +back as one JSON document it can inspect for ``headroom_retrieve`` calls, then +resynthesizes SSE for the client. + +That only works if the flip reaches the wire. When conversation history carries +a signed ``thinking`` block, ``select_outbound_body`` forwards the client's +original bytes instead — ``"stream": true`` and all — so upstream streams, the +JSON parse fails, resynthesis is skipped, and the client is left with a 200 and +nothing it can read. These tests pin the three defenses: don't take the path, +survive the reply if we somehow do, and never cache a body in the wrong format. +""" + +from __future__ import annotations + +import asyncio +import json +from datetime import datetime + +import pytest + +fastapi = pytest.importorskip("fastapi") +httpx = pytest.importorskip("httpx") + +from fastapi.testclient import TestClient # noqa: E402 + +from headroom.proxy.models import CacheEntry # noqa: E402 +from headroom.proxy.server import ProxyConfig, create_app # noqa: E402 + +RETRIEVE_TOOL = { + "name": "headroom_retrieve", + "description": "Retrieve original content", + "input_schema": {"type": "object", "properties": {}}, +} + +SIGNED_THINKING_TURN = { + "role": "assistant", + "content": [ + { + "type": "thinking", + "thinking": "private reasoning", + "signature": "sig-abc123", + }, + {"type": "text", "text": "Answered."}, + ], +} + +SSE_BODY = ( + b'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1"}}\n\n' + b'event: message_stop\ndata: {"type":"message_stop"}\n\n' +) + + +def _config() -> ProxyConfig: + return ProxyConfig( + optimize=False, + cache_enabled=True, + rate_limit_enabled=False, + memory_enabled=False, + ) + + +def _body(*, with_thinking: bool) -> dict: + messages: list[dict] = [{"role": "user", "content": "hi"}] + if with_thinking: + messages.append(SIGNED_THINKING_TURN) + messages.append({"role": "user", "content": "continue"}) + return { + "model": "claude-sonnet-4-20250514", + "max_tokens": 64, + "stream": True, + "tools": [RETRIEVE_TOOL], + "messages": messages, + } + + +def _headers() -> dict[str, str]: + return {"Authorization": "Bearer test-key", "x-api-key": "test-key"} + + +@pytest.mark.parametrize( + ("with_thinking", "expect_plain_streaming"), + [(True, True), (False, False)], +) +def test_signed_thinking_history_skips_the_buffered_ccr_path( + with_thinking: bool, expect_plain_streaming: bool +) -> None: + """The buffered path is only chosen when the stream:false flip can land.""" + calls: dict[str, object] = {} + + async def fake_stream_response(url, headers, body, *args, **kwargs): # noqa: ANN001 + calls["stream_body"] = body + return fastapi.responses.StreamingResponse(iter([SSE_BODY]), media_type="text/event-stream") + + async def fake_retry(method, url, headers, req_body, *args, **kwargs): # noqa: ANN001 + calls["buffered_body"] = json.loads(json.dumps(req_body)) + return httpx.Response( + 200, + json={ + "id": "msg_1", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-20250514", + "content": [{"type": "text", "text": "ok"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 10, "output_tokens": 5}, + }, + headers={"content-type": "application/json"}, + ) + + app = create_app(_config()) + with TestClient(app) as client: + client.app.state.proxy._stream_response = fake_stream_response + client.app.state.proxy._retry_request = fake_retry + resp = client.post( + "/v1/messages", json=_body(with_thinking=with_thinking), headers=_headers() + ) + + assert resp.status_code == 200, resp.text + if expect_plain_streaming: + # Passthrough is locked in, so we must not pretend we can buffer. + assert "stream_body" in calls, "expected the plain streaming path" + assert "buffered_body" not in calls + # The turn still leaves as a streaming request, matching the bytes + # that passthrough will actually forward. + assert calls["stream_body"]["stream"] is True + else: + assert "buffered_body" in calls, "expected the buffered CCR path" + assert calls["buffered_body"]["stream"] is False + + +@pytest.mark.parametrize("upstream_delay", [0.0, 1.2], ids=["prompt", "past-keepalive"]) +def test_buffered_ccr_relays_an_unexpected_sse_reply_and_does_not_cache_it( + upstream_delay: float, +) -> None: + """A 200 SSE reply on the buffered path reaches the client as a stream. + + The delay matters: ``_BufferedCCRResponse`` commits SSE response headers + after a 1 s keepalive, and past that point it can only forward a result + that exposes a ``body_iterator``. A plain ``Response`` there degrades to a + bare ``event: error`` — which is what a real (multi-second) Anthropic turn + hit in #2952. + """ + + async def fake_retry(method, url, headers, req_body, *args, **kwargs): # noqa: ANN001 + if upstream_delay: + await asyncio.sleep(upstream_delay) + return httpx.Response(200, content=SSE_BODY, headers={"content-type": "text/event-stream"}) + + app = create_app(_config()) + with TestClient(app) as client: + proxy = client.app.state.proxy + proxy._retry_request = fake_retry + resp = client.post("/v1/messages", json=_body(with_thinking=False), headers=_headers()) + + assert resp.status_code == 200, resp.text + assert resp.headers["content-type"].startswith("text/event-stream") + assert b"message_start" in resp.content + # Caching SSE bytes under a key with no `stream` component is what + # served a stream to a buffered caller in the first place. + assert proxy.cache is not None + assert len(proxy.cache._cache) == 0, "an unparseable body must never be cached" + + +def test_cache_hit_never_replays_a_foreign_content_type() -> None: + """A cache entry cannot hand a caller a wire format it did not ask for.""" + body = { + "model": "claude-sonnet-4-20250514", + "max_tokens": 64, + "stream": False, + "messages": [{"role": "user", "content": "hi"}], + } + payload = json.dumps( + { + "id": "msg_1", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-20250514", + "content": [{"type": "text", "text": "cached"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 1, "output_tokens": 1}, + } + ).encode() + + async def fail_retry(*args, **kwargs): # noqa: ANN001, ANN002, ANN003 + raise AssertionError("upstream must not be called on a cache hit") + + app = create_app(_config()) + with TestClient(app) as client: + proxy = client.app.state.proxy + proxy._retry_request = fail_retry + key = proxy.cache._compute_key( + body["messages"], + body["model"], + system=None, + tools=None, + tool_choice=None, + temperature=None, + top_p=None, + top_k=None, + max_tokens=64, + stop=None, + thinking=None, + output_config=None, + ) + proxy.cache._cache[key] = CacheEntry( + response_body=payload, + response_headers={"content-type": "text/event-stream"}, + created_at=datetime.now(), + ttl_seconds=3600, + ) + + resp = client.post("/v1/messages", json=body, headers=_headers()) + + assert resp.status_code == 200, resp.text + assert resp.headers["content-type"].startswith("application/json") + assert resp.json()["content"][0]["text"] == "cached" diff --git a/tests/test_proxy_byte_faithful_forwarding.py b/tests/test_proxy_byte_faithful_forwarding.py index 7fc2dbd69..eacfdc0bd 100644 --- a/tests/test_proxy_byte_faithful_forwarding.py +++ b/tests/test_proxy_byte_faithful_forwarding.py @@ -34,6 +34,7 @@ from headroom.proxy.body_forwarding import ( BodyMutationTracker, OutboundBody, get_python_forwarder_mode, + outbound_body_is_client_bytes, prepare_outbound_body_bytes, select_outbound_body, serialize_body_canonical, @@ -252,7 +253,113 @@ def test_signed_thinking_history_overrides_legacy_encoder() -> None: forwarder_mode="legacy_json_kwarg", ) - assert outbound == OutboundBody(content=original, source="passthrough") + assert outbound == OutboundBody(content=original, source="passthrough", dropped_mutations=True) + + +def test_signed_thinking_passthrough_reports_the_mutations_it_discarded() -> None: + """Passthrough silently winning over a mutated body is what hid #2952.""" + body = { + "stream": False, + "messages": [ + { + "role": "assistant", + "content": [{"type": "thinking", "signature": "sig123"}], + } + ], + } + original = json.dumps({**body, "stream": True}).encode("utf-8") + + outbound = select_outbound_body( + body=body, + original_body_bytes=original, + body_mutated=True, + forwarder_mode="byte_faithful", + mutation_reasons=["ccr_streaming_retrieve_buffered_non_stream"], + ) + + assert outbound.source == "passthrough" + assert outbound.dropped_mutations is True + assert outbound.dropped_mutation_reasons == ("ccr_streaming_retrieve_buffered_non_stream",) + + +def test_signed_thinking_passthrough_reports_nothing_when_body_unmutated() -> None: + body = { + "messages": [ + { + "role": "assistant", + "content": [{"type": "thinking", "signature": "sig123"}], + } + ] + } + original = json.dumps(body).encode("utf-8") + + outbound = select_outbound_body( + body=body, + original_body_bytes=original, + body_mutated=False, + forwarder_mode="byte_faithful", + mutation_reasons=["irrelevant"], + ) + + assert outbound.source == "passthrough" + assert outbound.dropped_mutations is False + assert outbound.dropped_mutation_reasons == () + + +def test_canonical_path_reports_no_dropped_mutations() -> None: + body = {"messages": [{"role": "user", "content": "hi"}]} + + outbound = select_outbound_body( + body=body, + original_body_bytes=b'{"messages": []}', + body_mutated=True, + forwarder_mode="byte_faithful", + mutation_reasons=["compression"], + ) + + assert outbound.source == "canonical" + assert outbound.dropped_mutations is False + assert outbound.dropped_mutation_reasons == () + + +@pytest.mark.parametrize( + ("original_body_bytes", "expected"), + [(b'{"messages": []}', True), (None, False)], +) +def test_outbound_body_is_client_bytes_matches_selection( + original_body_bytes: bytes | None, expected: bool +) -> None: + """Handlers gate on this before mutating a body for their own upstream call.""" + body = { + "messages": [ + { + "role": "assistant", + "content": [{"type": "thinking", "signature": "sig123"}], + } + ] + } + + assert ( + outbound_body_is_client_bytes(body=body, original_body_bytes=original_body_bytes) + is expected + ) + outbound = select_outbound_body( + body=body, + original_body_bytes=original_body_bytes, + body_mutated=True, + forwarder_mode="byte_faithful", + ) + assert (outbound.source == "passthrough") is expected + + +def test_outbound_body_is_client_bytes_false_without_thinking_blocks() -> None: + assert ( + outbound_body_is_client_bytes( + body={"messages": [{"role": "user", "content": "hi"}]}, + original_body_bytes=b'{"messages": []}', + ) + is False + ) def test_prepare_outbound_no_original_bytes_uses_canonical() -> None: