diff --git a/headroom/perf/analyzer.py b/headroom/perf/analyzer.py
index b296e59f6..e05997fac 100644
--- a/headroom/perf/analyzer.py
+++ b/headroom/perf/analyzer.py
@@ -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",
 ]
 
 
diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py
index 79045fa84..5abb1b074 100644
--- a/headroom/proxy/handlers/anthropic.py
+++ b/headroom/proxy/handlers/anthropic.py
@@ -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,
diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py
index c011b4984..4886593d4 100644
--- a/headroom/proxy/handlers/openai.py
+++ b/headroom/proxy/handlers/openai.py
@@ -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,
diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py
index ecd57f242..a20801c2e 100644
--- a/headroom/proxy/helpers.py
+++ b/headroom/proxy/helpers.py
@@ -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,
diff --git a/headroom/proxy/outcome.py b/headroom/proxy/outcome.py
index d96190b4d..bc594f81e 100644
--- a/headroom/proxy/outcome.py
+++ b/headroom/proxy/outcome.py
@@ -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}"
     )
diff --git a/tests/test_anthropic_pre_upstream_backpressure.py b/tests/test_anthropic_pre_upstream_backpressure.py
index 777ee7c6a..be51e48a0 100644
--- a/tests/test_anthropic_pre_upstream_backpressure.py
+++ b/tests/test_anthropic_pre_upstream_backpressure.py
@@ -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
diff --git a/tests/test_proxy_response_cache_replay.py b/tests/test_proxy_response_cache_replay.py
new file mode 100644
index 000000000..776d2569d
--- /dev/null
+++ b/tests/test_proxy_response_cache_replay.py
@@ -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"