fix(proxy): stop cached responses replaying the producing turn's wire framing (#3024)
## Description Closes #3019 A response-cache hit could hand the client an HTTP 200 that the client could not read, and nothing in the logs marked the turn as anything other than normal. Two separate problems combine to produce the reported failure. **The unreadable 200.** A cache entry stores the producing upstream's response headers verbatim. When the entry is replayed, the Anthropic handler removed only `content-encoding`, `content-length` and `content-type` before handing those headers to a brand-new `Response`. Anything else describing how that *other* connection framed its body rode along — most damagingly `transfer-encoding: chunked`. RFC 9112 §6.1 makes `Transfer-Encoding` override `Content-Length`, so the client is told to parse a plain JSON body as chunked frames, finds no valid chunk-size line, and reads an empty body out of a 200. Every other response-forwarding site in the Python proxy already strips that header; the two cache-hit sites were the only ones that did not. **How a CCR turn could put a foreign response in the cache.** On the Anthropic path, `cache.get` is gated on `not stream` but `cache.set` was not, and the cache key has no `stream` component. A CCR buffered-stream conversion takes a request the client sent with `stream: true`, forces `stream: false` upstream, and — unlike every other streaming turn, which returns via `_stream_response` and never touches the cache — falls through to the store site. The stored reply was shaped by that forced flip plus CCR tool injection, and the key cannot distinguish it from an ordinary non-streaming reply, so a later non-streaming caller could be served a response built for a request it never made. This is why the reporters saw the failures pair with CCR activity and stop under `--lossless` / `--no-ccr`. **Why it was invisible.** The cache-hit block emitted no log line at all, and the `PERF` line rendered no field for `RequestOutcome.from_response_cache`. A cache-served turn contacts no upstream, so it has no `outbound_request` line, no upstream stage timings, and all-zero token counters — byte-for-byte what a turn that died would look like. That is why `headroom doctor` reported zero failures while turns were dying. ### Scope note The header fix also lands on the OpenAI cache-hit site, which additionally never received the `content-type` fix from #2952. The `not stream` gate is added to the OpenAI store site too, where it is currently redundant — a streaming chat request returns via `_stream_response` long before that point — purely to state the invariant, since the Anthropic handler had exactly that shape until a buffered-CCR branch began falling through to it. Because the strip list now lives in one shared helper, the OpenAI handler's other five forwarding sites strip the three added headers as well. That is a widening, so it is worth being explicit about: each of those sites builds a fresh fixed-length `Response` (or, at `openai.py:6122`, synthesises SSE) from `response.content`, so replaying the upstream's framing there was the same latent bug, just without a cache to make it outlive the request that produced it. The precedent is already in the file — `openai.py:9865` passes `"transfer-encoding", "connection"` as extra names by hand, which is exactly the gap this PR closes centrally. That call site keeps its now-redundant arguments; removing them is a cleanup for another PR. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) ## Changes Made - Added `sanitize_forwarded_response_headers` to `headroom/proxy/helpers.py`, promoting the private helper that already lived in `headroom/proxy/handlers/openai.py` and extending it with the remaining wire-framing headers (`transfer-encoding`, `connection`, `keep-alive`). Matching is now case-insensitive; surviving headers keep their original casing. `openai.py`'s `_sanitize_forwarded_response_headers` is now a thin alias so its six call sites and the Anthropic handler strip an identical set. - `headroom/proxy/handlers/anthropic.py`: the response-cache hit now sanitises through that helper (passing `content-type` as an extra name, preserving #2952) instead of three hand-rolled `pop` calls. - `headroom/proxy/handlers/openai.py`: the response-cache hit sanitises the same way, gains the `content-type` handling it was missing, and sets `media_type="application/json"` explicitly. - `headroom/proxy/handlers/anthropic.py`: `cache.set` is now gated on `not stream`, mirroring the read gate. `stream` still holds the client's original flag at that point — the buffered-CCR conversion flips `body["stream"]`, never the local variable. - `headroom/proxy/handlers/openai.py`: the same `not stream` gate on its store site, as an invariant guard. - Both cache-hit sites now log `RESPONSE-CACHE-HIT: model=… bytes=… age_s=… hits=…`, following the existing `CACHE-MISS-ATTRIBUTION` line style. - `headroom/proxy/outcome.py`: the `PERF` line appends `cached=1` on a response-cache hit. It is appended only on a hit, so every other PERF line is byte-identical to before and existing parsers are unaffected. - `headroom/perf/analyzer.py`: `PerfRecord.from_response_cache` reads that field, so `headroom perf` can tell a cache-served turn from a dead one. It defaults to `False`, so older logs still parse. `PERF_RECORD_FIELDS` gains the name at the end of the list, which is what `headroom perf --format csv --raw` uses as its column set; appending keeps every existing column at its current position. `--format json --raw` gains the key too. - `tests/test_anthropic_pre_upstream_backpressure.py`: its cache-hit double was a partial hand-rolled stand-in for `CacheEntry` carrying only a body and headers, so it broke once the hit path started reading the entry's age and hit count. It now constructs a real `CacheEntry`, which is what the cache actually returns. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality ### Test Output ```text $ python -m pytest tests/test_proxy_response_cache_replay.py -q tests\test_proxy_response_cache_replay.py ......... [100%] ============================== 9 passed in 4.22s ============================== # Everything that mentions PERF, the sanitiser, cache.set, PerfRecord or # response_headers, plus the whole proxy suite. $ python -m pytest tests/test_proxy/ tests/test_proxy_compression_headers.py \ tests/test_agent_savings.py tests/test_anthropic_pre_upstream_backpressure.py \ tests/test_backend_nonstreaming_cache_metrics.py tests/test_backend_streaming_cache_metrics.py \ tests/test_ccr_buffered_stream_signed_thinking.py tests/test_cli_perf_format.py \ tests/test_codex_ws_compression_scheduler.py tests/test_handler_outcome_tag_invariant.py \ tests/test_openai_codex_ws_lifecycle.py tests/test_provider_codex_images.py \ tests/test_proxy_handlers_batch.py tests/test_proxy_passthrough_transient_retry.py \ tests/test_proxy_response_cache_replay.py tests/test_proxy_semantic_cache_key.py \ tests/test_proxy_streaming_request_logger.py tests/test_request_outcome.py \ tests/test_savings_tool_search_aggregation.py -q ================== 555 passed, 1 skipped in 88.60s (0:01:28) ================== # Full suite, 16 workers. See "Real Behavior Proof" below for how every # failure here was traced to a pre-existing failure or a parallelism flake. $ python -m pytest tests scripts/tests -n 16 -q -p no:randomly --timeout=300 83 failed, 10493 passed, 657 skipped, 80 errors in 437.00s (0:07:17) $ ruff check . All checks passed! $ ruff format --check <the 7 changed files> 7 files already formatted $ python -m mypy headroom --ignore-missing-imports --python-version 3.13 Found 12 errors in 3 files (checked 520 source files) # All 12 are pre-existing MCP-SDK/tomllib drift in release_version.py, # ccr/mcp_server.py and memory/mcp_server.py; identical count before and # after this change, none in the files it touches. ``` ## Real Behavior Proof - Environment: Windows 11, Python 3.13.11, pytest 9.1.1, ruff 0.16.2, branch based on `upstream/main` at `2d88e31a`. - Exact command / steps: Two experiments. (1) Revert-and-rerun: I reverted both fixes in place (dropped the three framing headers from `FRAMING_RESPONSE_HEADERS`, restored `cache.set` to `if self.cache and response.status_code == 200 and resp_json is not None:`), ran `python -m pytest tests/test_proxy_response_cache_replay.py -q`, then restored the fixes and re-ran. (2) Regression sweep: ran the full suite on this branch, then checked out `upstream/main` into a second worktree and re-ran, in that worktree, exactly the tests that failed here and not there. - Observed result: With the fixes reverted, 5 of 9 new tests fail and reproduce both halves of the bug. `test_buffered_ccr_turn_does_not_write_the_response_cache` fails with `AssertionError: Expected mock to not have been awaited. Awaited 1 times.` — a turn the client sent as `stream: true` really does reach `cache.set` through the buffered-CCR branch. `test_cache_hit_replays_a_body_the_client_can_actually_read` fails with `AssertionError: assert 'transfer-encoding' not in {'transfer-encoding': 'chunked', 'connection': 'keep-alive', 'request-id': ..., 'content-length': '228', ...}` — the replayed 200 carries the producing turn's chunked framing alongside a fresh `content-length`, which is the exact framing conflict a client cannot parse. With the fixes restored, all 9 pass, the replayed body arrives intact as `application/json`, and the run logs both `RESPONSE-CACHE-HIT` and a `PERF … cached=1` line. The full suite on this branch gives `83 failed, 10493 passed, 657 skipped, 80 errors`; 33 of those failures were not in my baseline list, so I ran those 33 in the `upstream/main` worktree and 20 failed there identically (Windows-specific: `sqlite:///C:\…` path handling, private-directory permissions, fsync, ONNX thread caps, serena config discovery). Re-running the remaining 13 serially on this branch gave `1 failed, 25 passed` — the other 12 were xdist parallelism flakes, including all four `tests/test_proxy/test_anthropic_ccr_deferred_injection.py` tests, which are the only ones in this change's blast radius and which pass serially. The one real serial failure, `tests/test_savings_ledger_offload.py::test_concurrent_requests_all_land_their_events` (`AssertionError: a concurrent append was lost / assert 23 == 24`), fails the same way on `upstream/main` run serially. The 80 errors are dashboard-template collection errors unrelated to the proxy. Net: no failure attributable to this change. - Not tested: I could not reproduce against live upstream traffic, so I have not confirmed which upstream in the reporters' setups emits `transfer-encoding: chunked`. Anthropic direct is HTTP/2, where the header is forbidden, but any HTTP/1.1 hop (corporate proxy, third-party gateway, local relay) reintroduces it. I have also not measured whether the `not stream` gate reduces the cache hit rate in practice; by construction it can only drop entries that were unsafe to serve. A reporter running unmodified 0.35.0 with `headroom proxy --no-cache` would confirm the cache path is the one involved, and that flag is a lighter workaround than `--lossless` or `--no-ccr` because it keeps CCR and compression enabled. ## Runtime Rollout Safety - Rollout-managed feature(s): none — this is a correctness fix on the always-on response-cache path (`cache_enabled` defaults to `True`). - Minimum rollout channel: stable. - Stable/default behavior changed: yes, in four ways. Replayed cached responses no longer carry the producing upstream's framing headers (or `server`, on the Anthropic side). Forwarded responses on the OpenAI handler's other five sanitiser call sites no longer carry `transfer-encoding`, `connection` or `keep-alive` either, since the strip list is now shared; all five build a fixed-length response from `response.content`, so none of them could legitimately replay that framing. A turn whose client asked for `stream: true` no longer writes the response cache on the Anthropic path. `PERF` lines gain a trailing `cached=1` on a response-cache hit only; all other PERF lines are unchanged. - Kill switch / disable path: `headroom proxy --no-cache` disables the response cache entirely and bypasses every path this PR touches. - Unsafe override required: no. - Qualification impact: low. No public API, config key, CLI flag or wire format changes. Two additive output changes: the `cached=1` PERF field, which `_parse_kv` already handles the same way it handles the existing trailing `client=` field, and a `from_response_cache` column appended to `headroom perf --format csv --raw` (plus the matching key in `--format json --raw`). Anything consuming that CSV positionally keeps working because the column is last; anything reading it by name is unaffected. - Rollback path: revert this commit. It is self-contained with no migration, no persisted state and no schema change; cache entries written before or after behave identically on read. ## 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 is marked N/A: no user-facing surface changes, and the new `cached=` PERF field is additive and self-describing. Relationship to nearby open PRs, since several touch adjacent code: - **#2953** (already merged, unreleased) added the `resp_json is not None` guard at the same Anthropic store site. That stops an SSE *body* being stored; it does not stop a JSON-bodied response storing chunked framing headers, and it does not add the `stream` gate. The two changes are complementary. - **#2959** and **#2968** both touch the buffered-CCR response path but address when and how the status is committed. Neither reaches the cache-hit replay. - **#3013** rewrites CCR into event-level stream splicing and keeps `buffered_stream_ccr` as a fallback, so the store site this PR gates remains reachable. If #3013 lands first I am happy to rebase. `mypy headroom --ignore-missing-imports` reports 12 pre-existing errors in `headroom/release_version.py`, `headroom/ccr/mcp_server.py` and `headroom/memory/mcp_server.py` from MCP SDK version drift in my local environment. None are in the files this PR touches, and the count is identical before and after the change. Co-authored-by: Claude Opus 5 <noreply@anthropic.com> Co-authored-by: JD Davis <mxjerrett@gmail.com>
This commit is contained in:
Parideboy
@@ -167,6 +167,11 @@ class PerfRecord:
|
||||
ttfb_ms: float = 0.0
|
||||
stages: dict[str, float] = field(default_factory=dict)
|
||||
savings_breakdown: list[dict[str, object]] = field(default_factory=list)
|
||||
# True when the proxy answered from its own response cache and never
|
||||
# contacted the upstream. Such a turn has all-zero token counters and no
|
||||
# upstream stage timings, so without this flag it reads as a turn that
|
||||
# did nothing (#3019). Absent from pre-#3019 logs, hence the default.
|
||||
from_response_cache: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -373,6 +378,7 @@ def parse_log_files(last_n_hours: float = 168.0) -> PerfReport:
|
||||
total_ms=float(kv.get("total_ms", 0)),
|
||||
tokens_out=int(kv.get("tok_out", 0)),
|
||||
ttfb_ms=float(kv.get("ttfb_ms", 0)),
|
||||
from_response_cache=kv.get("cached", "0") == "1",
|
||||
stages=stages_by_rid.get(m.group("rid"), {}),
|
||||
)
|
||||
)
|
||||
@@ -765,6 +771,9 @@ PERF_RECORD_FIELDS = [
|
||||
"ttfb_ms",
|
||||
"stages",
|
||||
"savings_breakdown",
|
||||
# Appended last so every existing CSV column keeps its position; a reader
|
||||
# that indexes by name is unaffected either way.
|
||||
"from_response_cache",
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -36,7 +36,11 @@ from headroom.proxy.auth_mode import (
|
||||
from headroom.proxy.compression_decision import CompressionDecision
|
||||
from headroom.proxy.forwarded_headers import resolve_client_ip
|
||||
from headroom.proxy.handlers._debug_dump import _debug_dump_mode, _redact_debug_value
|
||||
from headroom.proxy.helpers import extract_tags, relocate_system_messages_to_top_level
|
||||
from headroom.proxy.helpers import (
|
||||
extract_tags,
|
||||
relocate_system_messages_to_top_level,
|
||||
sanitize_forwarded_response_headers,
|
||||
)
|
||||
from headroom.proxy.image_isolation import run_image_compression_isolated
|
||||
from headroom.proxy.memory_decision import MemoryDecision
|
||||
from headroom.proxy.memory_query import MemoryQuery
|
||||
@@ -1176,15 +1180,32 @@ class AnthropicHandlerMixin:
|
||||
)
|
||||
)
|
||||
|
||||
# Remove compression headers from cached response
|
||||
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)
|
||||
# Strip the stored response's wire-framing headers. The
|
||||
# entry carries whatever the *producing* upstream sent,
|
||||
# and replaying that framing over a different connection
|
||||
# breaks the body: a stale ``transfer-encoding: chunked``
|
||||
# makes the client parse plain JSON as chunked frames and
|
||||
# read nothing out of an HTTP 200 (#3019). ``content-type``
|
||||
# goes too, because Starlette lets an explicit header win
|
||||
# over ``media_type`` and the producing request's type
|
||||
# would hand this caller a wire format it never asked
|
||||
# for (#2952).
|
||||
response_headers = sanitize_forwarded_response_headers(
|
||||
cached.response_headers,
|
||||
"content-type",
|
||||
)
|
||||
|
||||
# A cache hit answers the client without touching the
|
||||
# upstream, so it emits no outbound_request line and no
|
||||
# upstream stage timings. Without this log a served-from-
|
||||
# cache turn is indistinguishable from a turn that died
|
||||
# silently, which is exactly how #3019 stayed invisible.
|
||||
logger.info(
|
||||
f"[{request_id}] RESPONSE-CACHE-HIT: model={model} "
|
||||
f"bytes={len(cached.response_body)} "
|
||||
f"age_s={(datetime.now() - cached.created_at).total_seconds():.0f} "
|
||||
f"hits={cached.hit_count}"
|
||||
)
|
||||
|
||||
# Unit 4: release the pre-upstream semaphore on cache
|
||||
# hit — no upstream call will happen.
|
||||
@@ -3991,7 +4012,24 @@ class AnthropicHandlerMixin:
|
||||
# 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:
|
||||
#
|
||||
# ``not stream`` mirrors the read gate at the cache
|
||||
# lookup above. ``stream`` still holds the *client's*
|
||||
# original flag here — the buffered-CCR conversion
|
||||
# flips ``body["stream"]``, never this variable — so a
|
||||
# turn the client asked to stream is the one case that
|
||||
# can reach this store site with a buffered body. That
|
||||
# body was shaped by a forced ``stream: false`` flip
|
||||
# plus CCR tool injection, and the key cannot tell it
|
||||
# apart from an ordinary non-streaming reply, so
|
||||
# storing it lets a later caller be answered with a
|
||||
# response built for a request it never made (#3019).
|
||||
if (
|
||||
self.cache
|
||||
and not stream
|
||||
and response.status_code == 200
|
||||
and resp_json is not None
|
||||
):
|
||||
await self.cache.set(
|
||||
cache_lookup_messages,
|
||||
model,
|
||||
|
||||
@@ -28,6 +28,7 @@ from headroom.proxy.helpers import (
|
||||
_headroom_bypass_enabled,
|
||||
extract_tags,
|
||||
jitter_delay_ms,
|
||||
sanitize_forwarded_response_headers,
|
||||
)
|
||||
from headroom.proxy.loopback_guard import is_loopback_host
|
||||
from headroom.proxy.stage_timer import StageTimer, emit_stage_timings_log
|
||||
@@ -327,10 +328,10 @@ def _sanitize_forwarded_response_headers(
|
||||
headers: httpx.Headers | dict[str, str],
|
||||
*extra_names: str,
|
||||
) -> dict[str, str]:
|
||||
cleaned = dict(headers)
|
||||
for name in ("content-encoding", "content-length", "server", *extra_names):
|
||||
cleaned.pop(name, None)
|
||||
return cleaned
|
||||
# Thin alias kept for the many call sites in this module; the policy
|
||||
# (and the list of framing headers) lives in one place so the Anthropic
|
||||
# handler strips exactly the same set — see #3019.
|
||||
return sanitize_forwarded_response_headers(headers, *extra_names)
|
||||
|
||||
|
||||
def _resolve_openai_handler_path(
|
||||
@@ -3273,10 +3274,34 @@ class OpenAIHandlerMixin:
|
||||
)
|
||||
)
|
||||
|
||||
# Remove compression headers from cached response
|
||||
response_headers = _sanitize_forwarded_response_headers(cached.response_headers)
|
||||
# Strip the stored response's wire-framing headers, and its
|
||||
# content-type: the entry carries whatever the *producing*
|
||||
# upstream sent, and replaying that framing over a different
|
||||
# connection breaks the body — a stale
|
||||
# ``transfer-encoding: chunked`` makes the client parse plain
|
||||
# JSON as chunked frames and read nothing out of an HTTP 200
|
||||
# (#3019, same reasoning as #2952 on the Anthropic twin).
|
||||
response_headers = _sanitize_forwarded_response_headers(
|
||||
cached.response_headers,
|
||||
"content-type",
|
||||
)
|
||||
|
||||
return Response(content=cached.response_body, headers=response_headers)
|
||||
# A cache hit answers without touching the upstream, so it
|
||||
# emits no outbound_request line and no upstream stage
|
||||
# timings. Log it, or a served-from-cache turn looks exactly
|
||||
# like a turn that died silently (#3019).
|
||||
logger.info(
|
||||
f"[{request_id}] RESPONSE-CACHE-HIT: model={model} "
|
||||
f"bytes={len(cached.response_body)} "
|
||||
f"age_s={(datetime.now() - cached.created_at).total_seconds():.0f} "
|
||||
f"hits={cached.hit_count}"
|
||||
)
|
||||
|
||||
return Response(
|
||||
content=cached.response_body,
|
||||
headers=response_headers,
|
||||
media_type="application/json",
|
||||
)
|
||||
|
||||
# Token counting (offloaded off the event loop — GH #1701)
|
||||
tokenizer, original_tokens = await self._count_tokens_offloaded(model, messages)
|
||||
@@ -4810,7 +4835,16 @@ class OpenAIHandlerMixin:
|
||||
# Cache response under the SAME key it was looked up by:
|
||||
# cache_lookup_messages is the raw pre-mutation snapshot, not
|
||||
# the live (hooked) `messages` (#327).
|
||||
if self.cache and response.status_code == 200:
|
||||
#
|
||||
# ``not stream`` mirrors the read gate at the cache lookup
|
||||
# above. It is currently redundant here — a streaming chat
|
||||
# request returns via ``_stream_response`` well before this
|
||||
# point — but the Anthropic handler had the same shape until a
|
||||
# buffered-CCR branch started falling through to its store
|
||||
# site, which let a response built for a stream:true request
|
||||
# answer a later non-streaming caller (#3019). Stating the
|
||||
# invariant keeps that from being reintroduced silently.
|
||||
if self.cache and not stream and response.status_code == 200:
|
||||
await self.cache.set(
|
||||
cache_lookup_messages,
|
||||
model,
|
||||
|
||||
@@ -317,6 +317,40 @@ def _headroom_bypass_enabled(headers: Any) -> bool:
|
||||
return bypass or passthrough
|
||||
|
||||
|
||||
# Response headers that describe how the *upstream* framed its body on the
|
||||
# wire, not what the payload means. Every one of them is invalid to replay:
|
||||
# Starlette recomputes content-length, and uvicorn owns the connection
|
||||
# framing. Forwarding a stale ``transfer-encoding: chunked`` onto a
|
||||
# fixed-length body is the worst of them — RFC 9112 §6.1 makes
|
||||
# Transfer-Encoding override Content-Length, so the client tries to parse a
|
||||
# plain JSON body as chunked frames, finds no valid chunk-size line, and
|
||||
# reads an empty body out of an HTTP 200 (#3019).
|
||||
FRAMING_RESPONSE_HEADERS: tuple[str, ...] = (
|
||||
"content-encoding",
|
||||
"content-length",
|
||||
"transfer-encoding",
|
||||
"connection",
|
||||
"keep-alive",
|
||||
"server",
|
||||
)
|
||||
|
||||
|
||||
def sanitize_forwarded_response_headers(
|
||||
headers: Any,
|
||||
*extra_names: str,
|
||||
) -> dict[str, str]:
|
||||
"""Drop wire-framing headers before replaying an upstream response.
|
||||
|
||||
Pass any additional header names to strip as ``extra_names`` (for
|
||||
example ``"content-type"`` when the caller sets its own media type).
|
||||
|
||||
Matching is case-insensitive, but the casing of the headers that
|
||||
survive is left untouched.
|
||||
"""
|
||||
drop = {name.lower() for name in (*FRAMING_RESPONSE_HEADERS, *extra_names)}
|
||||
return {key: value for key, value in dict(headers).items() if key.lower() not in drop}
|
||||
|
||||
|
||||
def log_outbound_request(
|
||||
*,
|
||||
forwarder: str,
|
||||
|
||||
@@ -586,6 +586,15 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
|
||||
# line unchanged, and gives ``headroom perf --client X``
|
||||
# parsers a clean key to filter on.
|
||||
client_part = f" client={outcome.client}" if outcome.client else ""
|
||||
# ``cached=1`` marks a turn answered from Headroom's own response cache.
|
||||
# Such a turn never contacts the upstream, so it has no outbound_request
|
||||
# line, no upstream stage timings, and all-zero token counters — which
|
||||
# made it indistinguishable in the logs from a turn that died silently
|
||||
# (#3019). Appended only on a hit, so every other PERF line is unchanged
|
||||
# and existing parsers keep working (``_parse_kv`` reads trailing
|
||||
# key=value pairs after ``transforms=`` the same way it reads
|
||||
# ``client=``).
|
||||
cached_part = " cached=1" if outcome.from_response_cache else ""
|
||||
# Tool-schema DEFERRAL savings can't move tok_before/after (those count messages
|
||||
# only), so a tool-heavy turn shows tok_saved=0 while genuinely saving thousands of
|
||||
# tool-definition tokens. `tool_saved` carries that component and `total_saved` is
|
||||
@@ -611,4 +620,5 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
|
||||
f"savings={encoded_savings} "
|
||||
f"transforms={_summarize_transforms(list(outcome.transforms_applied))}"
|
||||
f"{client_part}"
|
||||
f"{cached_part}"
|
||||
)
|
||||
|
||||
@@ -26,6 +26,7 @@ import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from datetime import datetime
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
@@ -38,7 +39,7 @@ from fastapi.testclient import TestClient
|
||||
|
||||
from headroom.cli.proxy import proxy as proxy_cli
|
||||
from headroom.proxy.handlers.anthropic import AnthropicHandlerMixin
|
||||
from headroom.proxy.models import ProxyConfig
|
||||
from headroom.proxy.models import CacheEntry, ProxyConfig
|
||||
from headroom.proxy.server import HeadroomProxy, create_app
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -870,12 +871,20 @@ class _SecurityBlock:
|
||||
|
||||
|
||||
class _CacheHit:
|
||||
class _Entry:
|
||||
response_headers: dict = {}
|
||||
response_body: bytes = b'{"id":"cached","type":"message","role":"assistant","content":[{"type":"text","text":"hit"}]}'
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._entry = self._Entry()
|
||||
# A real ``CacheEntry`` rather than a hand-rolled stand-in: the
|
||||
# cache-hit path reads more of the entry than just the body (it logs
|
||||
# the entry's age and hit count), and a partial fake drifts out of
|
||||
# sync with it silently.
|
||||
self._entry = CacheEntry(
|
||||
response_body=(
|
||||
b'{"id":"cached","type":"message","role":"assistant",'
|
||||
b'"content":[{"type":"text","text":"hit"}]}'
|
||||
),
|
||||
response_headers={},
|
||||
created_at=datetime.now(),
|
||||
ttl_seconds=3600,
|
||||
)
|
||||
|
||||
async def get(self, _messages, _model, **_kwargs):
|
||||
return self._entry
|
||||
|
||||
@@ -0,0 +1,348 @@
|
||||
"""Regression tests for #3019 — a response-cache hit must not hand the client
|
||||
an unusable HTTP 200.
|
||||
|
||||
Three separate defects met to produce the reported failure:
|
||||
|
||||
1. The cached entry stores the *producing* upstream's response headers
|
||||
verbatim. Replaying ``transfer-encoding: chunked`` onto a fresh
|
||||
fixed-length response makes the client parse plain JSON as chunked frames
|
||||
(RFC 9112 §6.1: Transfer-Encoding overrides Content-Length), so it reads an
|
||||
empty body out of a 200.
|
||||
2. The Anthropic ``cache.set`` had no ``stream`` gate while ``cache.get`` did,
|
||||
and the cache key has no ``stream`` component — so a buffered-CCR turn
|
||||
(client asked for ``stream: true``, upstream forced to ``stream: false``)
|
||||
could store a response that a later non-streaming caller was served.
|
||||
3. Nothing logged the hit, and the PERF line rendered no ``cached=`` field, so
|
||||
a served-from-cache turn was indistinguishable from a turn that died.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
fastapi = pytest.importorskip("fastapi")
|
||||
httpx = pytest.importorskip("httpx")
|
||||
|
||||
from fastapi.testclient import TestClient # noqa: E402
|
||||
|
||||
from headroom.ccr.tool_injection import create_ccr_tool_definition # noqa: E402
|
||||
from headroom.proxy.helpers import sanitize_forwarded_response_headers # noqa: E402
|
||||
from headroom.proxy.models import CacheEntry # noqa: E402
|
||||
from headroom.proxy.outcome import RequestOutcome, emit_request_outcome # noqa: E402
|
||||
from headroom.proxy.server import ProxyConfig, create_app # noqa: E402
|
||||
|
||||
|
||||
class _CapturingHandler(logging.Handler):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(level=logging.INFO)
|
||||
self.records: list[logging.LogRecord] = []
|
||||
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
self.records.append(record)
|
||||
|
||||
def messages(self) -> list[str]:
|
||||
return [record.getMessage() for record in self.records]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def proxy_log_capture():
|
||||
"""Capture ``headroom.proxy`` records.
|
||||
|
||||
``_setup_file_logging`` sets ``propagate = False`` on this logger, so
|
||||
``caplog`` (which hangs off the root) never sees them — the same reason
|
||||
``tests/test_anthropic_stage_timings.py`` attaches its own handler.
|
||||
"""
|
||||
target = logging.getLogger("headroom.proxy")
|
||||
handler = _CapturingHandler()
|
||||
previous_level = target.level
|
||||
target.addHandler(handler)
|
||||
target.setLevel(logging.INFO)
|
||||
try:
|
||||
yield handler
|
||||
finally:
|
||||
target.removeHandler(handler)
|
||||
target.setLevel(previous_level)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 1. The shared header sanitiser
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSanitizeForwardedResponseHeaders:
|
||||
def test_drops_every_wire_framing_header(self):
|
||||
cleaned = sanitize_forwarded_response_headers(
|
||||
{
|
||||
"content-encoding": "gzip",
|
||||
"content-length": "412",
|
||||
"transfer-encoding": "chunked",
|
||||
"connection": "keep-alive",
|
||||
"keep-alive": "timeout=5",
|
||||
"server": "cloudflare",
|
||||
"request-id": "req_abc",
|
||||
"anthropic-ratelimit-requests-remaining": "42",
|
||||
}
|
||||
)
|
||||
assert cleaned == {
|
||||
"request-id": "req_abc",
|
||||
"anthropic-ratelimit-requests-remaining": "42",
|
||||
}
|
||||
|
||||
def test_matches_case_insensitively_but_preserves_surviving_casing(self):
|
||||
cleaned = sanitize_forwarded_response_headers(
|
||||
{"Transfer-Encoding": "chunked", "Request-Id": "req_abc"}
|
||||
)
|
||||
assert cleaned == {"Request-Id": "req_abc"}
|
||||
|
||||
def test_extra_names_are_dropped_too(self):
|
||||
cleaned = sanitize_forwarded_response_headers(
|
||||
{"content-type": "text/event-stream", "request-id": "req_abc"},
|
||||
"content-type",
|
||||
)
|
||||
assert cleaned == {"request-id": "req_abc"}
|
||||
|
||||
def test_accepts_httpx_headers(self):
|
||||
cleaned = sanitize_forwarded_response_headers(
|
||||
httpx.Headers({"transfer-encoding": "chunked", "request-id": "req_abc"})
|
||||
)
|
||||
assert "transfer-encoding" not in cleaned
|
||||
assert cleaned["request-id"] == "req_abc"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 2. Replaying a poisoned cache entry
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _cache_config() -> ProxyConfig:
|
||||
return ProxyConfig(
|
||||
optimize=False,
|
||||
cache_enabled=True,
|
||||
rate_limit_enabled=False,
|
||||
cost_tracking_enabled=False,
|
||||
log_requests=False,
|
||||
image_optimize=False,
|
||||
)
|
||||
|
||||
|
||||
_CACHED_BODY = json.dumps(
|
||||
{
|
||||
"id": "msg_cached",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": "claude-sonnet-4-6",
|
||||
"content": [{"type": "text", "text": "served from cache"}],
|
||||
"stop_reason": "end_turn",
|
||||
"usage": {"input_tokens": 10, "output_tokens": 5},
|
||||
}
|
||||
).encode()
|
||||
|
||||
|
||||
def _poisoned_entry() -> CacheEntry:
|
||||
"""A cache entry carrying the producing upstream's wire framing."""
|
||||
return CacheEntry(
|
||||
response_body=_CACHED_BODY,
|
||||
response_headers={
|
||||
"transfer-encoding": "chunked",
|
||||
"content-length": "999999",
|
||||
"content-encoding": "gzip",
|
||||
"connection": "keep-alive",
|
||||
"content-type": "text/event-stream",
|
||||
"request-id": "req_from_the_producing_turn",
|
||||
},
|
||||
created_at=datetime.now(),
|
||||
ttl_seconds=3600,
|
||||
)
|
||||
|
||||
|
||||
def test_cache_hit_replays_a_body_the_client_can_actually_read(proxy_log_capture):
|
||||
"""The replayed 200 must carry no stale framing and an intact JSON body."""
|
||||
with patch("headroom.proxy.server.AnyLLMBackend"):
|
||||
app = create_app(_cache_config())
|
||||
with TestClient(app) as client:
|
||||
proxy = client.app.state.proxy
|
||||
proxy.cache.get = AsyncMock(return_value=_poisoned_entry())
|
||||
proxy._retry_request = AsyncMock(
|
||||
side_effect=AssertionError("a cache hit must not contact the upstream")
|
||||
)
|
||||
|
||||
resp = client.post(
|
||||
"/v1/messages",
|
||||
headers={"x-api-key": "test-key", "anthropic-version": "2023-06-01"},
|
||||
json={
|
||||
"model": "claude-sonnet-4-6",
|
||||
"max_tokens": 64,
|
||||
"messages": [{"role": "user", "content": "hello"}],
|
||||
},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
# The body survived intact — this is what an empty 200 looked like.
|
||||
assert resp.json()["content"][0]["text"] == "served from cache"
|
||||
|
||||
replayed = {key.lower(): value for key, value in resp.headers.items()}
|
||||
# None of the producing turn's framing may ride along.
|
||||
assert "transfer-encoding" not in replayed
|
||||
assert "content-encoding" not in replayed
|
||||
assert "connection" not in replayed
|
||||
# content-type is the caller's, not the producing turn's (#2952).
|
||||
assert replayed["content-type"] == "application/json"
|
||||
# content-length describes THIS body, not the stored one.
|
||||
assert replayed["content-length"] == str(len(_CACHED_BODY))
|
||||
# Non-framing upstream metadata still passes through.
|
||||
assert replayed["request-id"] == "req_from_the_producing_turn"
|
||||
|
||||
# The hit is no longer silent, and the PERF line marks it as cache-served.
|
||||
messages = proxy_log_capture.messages()
|
||||
assert any("RESPONSE-CACHE-HIT" in message for message in messages)
|
||||
assert any(" PERF " in message and "cached=1" in message for message in messages)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 3. A buffered-CCR turn must not populate the cache
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _ccr_cache_config() -> ProxyConfig:
|
||||
return ProxyConfig(
|
||||
optimize=False,
|
||||
cache_enabled=True,
|
||||
rate_limit_enabled=False,
|
||||
cost_tracking_enabled=False,
|
||||
log_requests=False,
|
||||
ccr_inject_tool=True,
|
||||
ccr_handle_responses=True,
|
||||
ccr_context_tracking=False,
|
||||
image_optimize=False,
|
||||
)
|
||||
|
||||
|
||||
def test_buffered_ccr_turn_does_not_write_the_response_cache():
|
||||
"""A client ``stream: true`` turn is converted to a buffered ``stream:
|
||||
false`` upstream call. Its reply is shaped by that flip plus CCR tool
|
||||
injection, and the cache key has no ``stream`` component — so storing it
|
||||
would let a later non-streaming caller be served a response built for a
|
||||
request it never made (#3019).
|
||||
"""
|
||||
upstream_response = {
|
||||
"id": "msg_buffered",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": "claude-sonnet-4-6",
|
||||
"content": [{"type": "text", "text": "buffered reply"}],
|
||||
"stop_reason": "end_turn",
|
||||
"usage": {
|
||||
"input_tokens": 10,
|
||||
"output_tokens": 5,
|
||||
"cache_read_input_tokens": 0,
|
||||
"cache_creation_input_tokens": 0,
|
||||
},
|
||||
}
|
||||
|
||||
with patch("headroom.proxy.server.AnyLLMBackend"):
|
||||
app = create_app(_ccr_cache_config())
|
||||
with TestClient(app) as client:
|
||||
proxy = client.app.state.proxy
|
||||
proxy._stream_response = AsyncMock(
|
||||
side_effect=AssertionError("buffered CCR must not take the live stream path")
|
||||
)
|
||||
proxy.cache.set = AsyncMock()
|
||||
forwarded_bodies: list[dict] = []
|
||||
|
||||
async def _fake_retry(method, url, headers, body, stream=False, **kwargs): # noqa: ANN001
|
||||
forwarded_bodies.append(json.loads(json.dumps(body)))
|
||||
return httpx.Response(200, json=upstream_response)
|
||||
|
||||
proxy._retry_request = _fake_retry # type: ignore[assignment]
|
||||
|
||||
resp = client.post(
|
||||
"/v1/messages",
|
||||
headers={
|
||||
"x-api-key": "test-key",
|
||||
"anthropic-version": "2023-06-01",
|
||||
"accept": "text/event-stream",
|
||||
},
|
||||
json={
|
||||
"model": "claude-sonnet-4-6",
|
||||
"max_tokens": 64,
|
||||
"stream": True,
|
||||
"tools": [create_ccr_tool_definition("anthropic")],
|
||||
"messages": [{"role": "user", "content": "hello"}],
|
||||
},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200, resp.text
|
||||
# The conversion really happened — otherwise this test proves nothing.
|
||||
assert forwarded_bodies and forwarded_bodies[0]["stream"] is False
|
||||
# ...and nothing was written to the response cache.
|
||||
proxy.cache.set.assert_not_awaited()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 4. The PERF line marks a cache-served turn
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _Metrics:
|
||||
async def record_request(self, **kwargs):
|
||||
return None
|
||||
|
||||
async def record_failed(self, provider):
|
||||
return None
|
||||
|
||||
|
||||
class _Handler:
|
||||
def __init__(self):
|
||||
self.metrics = _Metrics()
|
||||
self.cost_tracker = None
|
||||
self.logger = None
|
||||
|
||||
|
||||
def _perf_line(capture: _CapturingHandler) -> str:
|
||||
for message in capture.messages():
|
||||
if " PERF " in message:
|
||||
return message
|
||||
raise AssertionError("no PERF log line captured")
|
||||
|
||||
|
||||
def _outcome(*, from_response_cache: bool) -> RequestOutcome:
|
||||
return RequestOutcome(
|
||||
request_id="req-1",
|
||||
provider="anthropic",
|
||||
model="claude-sonnet-4-6",
|
||||
original_tokens=0,
|
||||
optimized_tokens=0,
|
||||
output_tokens=0,
|
||||
tokens_saved=0,
|
||||
attempted_input_tokens=0,
|
||||
from_response_cache=from_response_cache,
|
||||
)
|
||||
|
||||
|
||||
def test_perf_line_marks_a_response_cache_hit(proxy_log_capture):
|
||||
asyncio.run(emit_request_outcome(_Handler(), _outcome(from_response_cache=True)))
|
||||
assert "cached=1" in _perf_line(proxy_log_capture)
|
||||
|
||||
|
||||
def test_perf_line_is_unchanged_for_an_ordinary_turn(proxy_log_capture):
|
||||
"""Appended only on a hit, so existing PERF parsers see no new field."""
|
||||
asyncio.run(emit_request_outcome(_Handler(), _outcome(from_response_cache=False)))
|
||||
assert "cached=" not in _perf_line(proxy_log_capture)
|
||||
|
||||
|
||||
def test_perf_analyzer_reads_the_cached_field():
|
||||
from headroom.perf.analyzer import _parse_kv
|
||||
|
||||
parsed = _parse_kv("model=claude-sonnet-4-6 transforms=none client=claude cached=1")
|
||||
assert parsed["cached"] == "1"
|
||||
# ``transforms=`` is parsed last and swallows the rest of the line, so the
|
||||
# new trailing field has to survive that split the way ``client=`` does.
|
||||
assert parsed["client"] == "claude"
|
||||
assert parsed["transforms"] == "none"
|
||||
assert parsed["model"] == "claude-sonnet-4-6"
|
||||
Reference in New Issue
Block a user