From 7aca00445e23f421d1c8c278bf720cab046da20f Mon Sep 17 00:00:00 2001
From: chopratejas <chopratejas@gmail.com>
Date: Thu, 14 May 2026 14:39:30 -0700
Subject: [PATCH] refactor(proxy): migrate Gemini + Anthropic non-streaming
 onto outcome funnel
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Builds on the RequestOutcome contract introduced in the previous commit.
This commit collapses **8 more record_request sites** across two
providers, demonstrating that the contract works across the
provider-shape diversity it was designed for:

* `handle_gemini_generate_content` (1 site) — read-only cache, no
  write counter, no TTL splits. The funnel's optional fields default
  to 0 for everything Gemini doesn't have; no special-casing needed.
* `handle_gemini_count_tokens` (1 site) — sizing helper, no output
  tokens, no cache. Funnel handles the "minimal observation" shape
  with zero ceremony.
* `handle_anthropic_messages` — **6 sites collapse to 1 funnel call
  per site**, including the response-cache-hit path, the
  Bedrock/Vertex non-streaming backend path, the main native
  Anthropic non-streaming path, and three batch handlers
  (create / passthrough / CCR-processed results).

Bug fixes that fall out of the migration:

* The non-streaming Anthropic main site was missing
  `attempted_input_tokens=` (one of the 7-of-18 sites flagged in the
  P0 audit). Dashboards showing 0% active-savings on non-streaming
  Anthropic traffic will now show the correct ratio (= #454/#455
  silently retired for this surface).
* Bedrock/Vertex non-streaming site was missing cache args entirely,
  hardcoding `cache_hit=False` on RequestLog. Now `cache_hit` is
  derived from the outcome correctly. Cache extraction itself is
  still a follow-up — but the wire shape is now uniform.
* Three batch handlers (create / passthrough / CCR-processed) were
  emitting only `record_request` — no RequestLog, no PERF log. They
  now flow through the canonical funnel so batch traffic appears in
  `headroom perf` and the recent-requests feed for the first time.

Architectural changes:

* **Extracted the funnel from `HeadroomProxy._record_request_outcome`
  into a free function `emit_request_outcome(handler, outcome)`** in
  `outcome.py`. The proxy method becomes a thin two-line wrapper.
  Reason: test dummies (e.g. `_DummyAnthropicHandler` in
  `test_anthropic_pre_upstream_backpressure.py`) need to call the
  funnel from their mixin tests without inheriting from
  `HeadroomProxy`. A free function with structurally-typed `handler`
  arg satisfies both production and test paths without a typing.Protocol
  ceremony.
* **Added `from_response_cache: bool = False` to `RequestOutcome`**
  to model Headroom's semantic-cache hits separately from
  upstream-prompt-cache hits. Both still collapse to the unified
  `cache_hit` derived property for downstream consumers, but
  dashboards can split them. Previously the cache-hit path
  hardcoded `cached=True` to `record_request`; now it's a typed,
  explicit signal.
* **Two batch handlers (`handle_anthropic_batch_passthrough`,
  `handle_anthropic_batch_results`) now allocate a `request_id`** at
  entry. They didn't have one before (they logged
  `request_id=None`), but the funnel requires it. Minor logging
  improvement.

Tests
* `tests/test_anthropic_pre_upstream_backpressure.py::_DummyAnthropicHandler`
  gets a 5-line `_record_request_outcome` that delegates to
  `emit_request_outcome`. Same pattern the dummy uses for
  `_run_compression_in_executor` / `_next_request_id`.
* All 140 streaming/cache/Codex/anthropic/backpressure tests pass:
  - test_request_outcome.py (14)
  - test_backend_streaming_cache_metrics.py (4)
  - test_proxy_streaming_request_logger.py (8)
  - test_proxy_streaming_resilience.py (24)
  - test_proxy_anthropic_cache_stability.py (22)
  - test_anthropic_pre_upstream_backpressure.py (20)
  - test_openai_codex_routing.py (11)
  - test_openai_codex_ws_lifecycle.py (10)
  - test_responses_ws_pyo3_compression.py (27)
* ruff + mypy clean.

Surface impact
* `anthropic.py`: 6 record_request sites → 0 (all go through funnel).
  Net 315 insertions, 273 deletions, but **the insertions are mostly
  comments explaining the migration** — actual code change is closer
  to a net wash. The wins compound in next migrations.
* `gemini.py`: 2 sites → 0. Net +30 LOC (mostly comments).
* `server.py`: −90 LOC (funnel extracted to free function).
* `outcome.py`: +110 LOC (free function + comments).

Remaining migrations from P0 audit §6 (still pending):
* handle_openai_responses_ws (Codex WS, 2 sites)
* handle_openai_chat non-streaming
* handle_openai_responses HTTP
* handle_gemini_stream_generate_content + handle_google_cloudcode_stream
* handle_databricks_invocations
---
 headroom/proxy/handlers/anthropic.py          | 315 +++++++++---------
 headroom/proxy/handlers/gemini.py             |  63 ++--
 headroom/proxy/outcome.py                     | 131 +++++++-
 headroom/proxy/server.py                      |  90 +----
 ...est_anthropic_pre_upstream_backpressure.py |   8 +
 5 files changed, 334 insertions(+), 273 deletions(-)

diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py
index 4f66f1fd2..8df17f320 100644
--- a/headroom/proxy/handlers/anthropic.py
+++ b/headroom/proxy/handlers/anthropic.py
@@ -25,6 +25,7 @@ import httpx
 
 from headroom.pipeline import PipelineStage, summarize_routing_markers
 from headroom.proxy.auth_mode import classify_auth_mode
+from headroom.proxy.outcome import RequestOutcome
 
 logger = logging.getLogger("headroom.proxy")
 
@@ -352,7 +353,6 @@ class AnthropicHandlerMixin:
 
         from headroom.cache.compression_store import get_compression_store
         from headroom.ccr import CCRToolInjector
-        from headroom.proxy.cost import _summarize_transforms
         from headroom.proxy.helpers import (
             MAX_MESSAGE_ARRAY_LENGTH,
             MAX_REQUEST_BODY_SIZE,
@@ -361,7 +361,6 @@ class AnthropicHandlerMixin:
             compute_turn_id,
             read_request_json_with_bytes,
         )
-        from headroom.proxy.models import RequestLog
         from headroom.proxy.modes import is_cache_mode, is_token_mode
         from headroom.tokenizers import get_tokenizer
         from headroom.utils import extract_user_query
@@ -700,14 +699,28 @@ class AnthropicHandlerMixin:
                     )
                     optimization_latency = (time.time() - start_time) * 1000
 
-                    await self.metrics.record_request(
-                        provider="anthropic",
-                        model=model,
-                        input_tokens=0,
-                        output_tokens=0,
-                        tokens_saved=0,  # Savings already counted when response was cached
-                        latency_ms=optimization_latency,
-                        cached=True,
+                    # Response-cache hit: response body came from
+                    # Headroom's semantic cache, not the upstream
+                    # provider. ``from_response_cache=True`` is a
+                    # distinct signal from `cache_read_tokens > 0`
+                    # (which means upstream-prompt-cache hit). Dashboards
+                    # can split the two; the funnel collapses them into
+                    # the single `cached` boolean for Prometheus.
+                    await self._record_request_outcome(
+                        RequestOutcome(
+                            request_id=request_id,
+                            provider="anthropic",
+                            model=model,
+                            original_tokens=0,
+                            optimized_tokens=0,
+                            output_tokens=0,
+                            tokens_saved=0,
+                            attempted_input_tokens=0,
+                            from_response_cache=True,
+                            total_latency_ms=optimization_latency,
+                            overhead_ms=optimization_latency,
+                            num_messages=len(messages),
+                        )
                     )
 
                     # Remove compression headers from cached response
@@ -1610,50 +1623,36 @@ class AnthropicHandlerMixin:
                             )
                         except Exception:
                             attempted_input_tokens = original_tokens
-                        await self.metrics.record_request(
-                            provider=_backend_name,
-                            model=model,
-                            input_tokens=optimized_tokens,
-                            output_tokens=output_tokens,
-                            tokens_saved=tokens_saved,
-                            latency_ms=total_latency,
-                            cached=False,
-                            overhead_ms=optimization_latency,
-                            pipeline_timing=pipeline_timing,
-                            attempted_input_tokens=attempted_input_tokens,
-                        )
-
-                        if self.cost_tracker:
-                            self.cost_tracker.record_tokens(model, tokens_saved, optimized_tokens)
-
-                        # Log request
-                        if self.logger:
-                            self.logger.log(
-                                RequestLog(
-                                    request_id=request_id,
-                                    timestamp=datetime.now().isoformat(),
-                                    provider=_backend_name,
-                                    model=model,
-                                    input_tokens_original=original_tokens,
-                                    input_tokens_optimized=optimized_tokens,
-                                    output_tokens=output_tokens,
-                                    tokens_saved=tokens_saved,
-                                    savings_percent=(tokens_saved / original_tokens * 100)
-                                    if original_tokens > 0
-                                    else 0,
-                                    optimization_latency_ms=optimization_latency,
-                                    total_latency_ms=total_latency,
-                                    tags=tags,
-                                    cache_hit=False,
-                                    transforms_applied=transforms_applied,
-                                    request_messages=body.get("messages")
-                                    if self.config.log_full_messages
-                                    else None,
-                                    turn_id=compute_turn_id(
-                                        model, body.get("system"), body.get("messages")
-                                    ),
-                                )
+                        # Backend (Bedrock / Vertex) non-streaming.
+                        # Cache metrics aren't extracted from the backend
+                        # response here yet — that's a follow-up. The
+                        # funnel passes 0s for the cache fields, which
+                        # is the same observable behaviour as the
+                        # pre-refactor code (which also omitted them).
+                        await self._record_request_outcome(
+                            RequestOutcome(
+                                request_id=request_id,
+                                provider=_backend_name,
+                                model=model,
+                                original_tokens=original_tokens,
+                                optimized_tokens=optimized_tokens,
+                                output_tokens=output_tokens,
+                                tokens_saved=tokens_saved,
+                                attempted_input_tokens=attempted_input_tokens,
+                                total_latency_ms=total_latency,
+                                overhead_ms=optimization_latency,
+                                pipeline_timing=pipeline_timing,
+                                transforms_applied=tuple(transforms_applied),
+                                num_messages=len(body.get("messages", [])),
+                                tags=tags,
+                                turn_id=compute_turn_id(
+                                    model, body.get("system"), body.get("messages")
+                                ),
+                                request_messages=body.get("messages")
+                                if self.config.log_full_messages
+                                else None,
                             )
+                        )
 
                         return JSONResponse(
                             status_code=backend_response.status_code,
@@ -2087,18 +2086,6 @@ class AnthropicHandlerMixin:
                         original_messages=next_original_messages,
                     )
 
-                    if self.cost_tracker:
-                        self.cost_tracker.record_tokens(
-                            model,
-                            tokens_saved,
-                            optimized_tokens,
-                            cache_read_tokens=cr_tokens,
-                            cache_write_tokens=cw_tokens,
-                            cache_write_5m_tokens=cw_5m_tokens,
-                            cache_write_1h_tokens=cw_1h_tokens,
-                            uncached_tokens=uncached_input_tokens,
-                        )
-
                     # Cache response
                     if self.cache and response.status_code == 200:
                         await self.cache.set(
@@ -2109,26 +2096,10 @@ class AnthropicHandlerMixin:
                             tokens_saved=tokens_saved,
                         )
 
-                    # Record metrics — use optimized_tokens (what we sent), not API's
-                    # input_tokens which is just the non-cached portion with prompt caching
-                    await self.metrics.record_request(
-                        provider="anthropic",
-                        model=model,
-                        input_tokens=optimized_tokens,
-                        output_tokens=output_tokens,
-                        tokens_saved=tokens_saved,
-                        latency_ms=total_latency,
-                        overhead_ms=optimization_latency,
-                        pipeline_timing=pipeline_timing,
-                        waste_signals=waste_signals_dict,
-                        cache_read_tokens=cr_tokens,
-                        cache_write_tokens=cw_tokens,
-                        cache_write_5m_tokens=cw_5m_tokens,
-                        cache_write_1h_tokens=cw_1h_tokens,
-                        uncached_input_tokens=uncached_input_tokens,
-                    )
-
-                    # Subscription tracker: update headroom contribution counters
+                    # Subscription tracker: update headroom contribution
+                    # counters. Provider-specific OAuth/subscription
+                    # accounting — stays outside the funnel (different
+                    # concern, only fires for Bearer-not-sk-ant tokens).
                     if _auth_header.startswith("Bearer ") and not _auth_header.startswith(
                         "Bearer sk-ant-api"
                     ):
@@ -2144,56 +2115,53 @@ class AnthropicHandlerMixin:
                                 tokens_saved_cache_reads=cr_tokens,
                             )
 
-                    # Log request
-                    if self.logger:
-                        self.logger.log(
-                            RequestLog(
-                                request_id=request_id,
-                                timestamp=datetime.now().isoformat(),
-                                provider="anthropic",
-                                model=model,
-                                input_tokens_original=original_tokens,
-                                input_tokens_optimized=optimized_tokens,
-                                output_tokens=output_tokens,
-                                tokens_saved=tokens_saved,
-                                savings_percent=(tokens_saved / original_tokens * 100)
-                                if original_tokens > 0
-                                else 0,
-                                optimization_latency_ms=optimization_latency,
-                                total_latency_ms=total_latency,
-                                tags=tags,
-                                cache_hit=cache_hit,
-                                transforms_applied=transforms_applied,
-                                waste_signals=waste_signals_dict,
-                                request_messages=messages
-                                if self.config.log_full_messages
-                                else None,
-                                turn_id=compute_turn_id(
-                                    model, body.get("system"), body.get("messages")
-                                ),
-                            )
+                    # The pre-refactor PERF emit (above) read raw usage
+                    # off ``resp_usage`` instead of trusting cr_tokens /
+                    # cw_tokens. Both paths land on identical numbers
+                    # (extraction happens just above the cost_tracker
+                    # call), so the funnel uses the already-computed
+                    # values for consistency. Pre-refactor's
+                    # ``cache_hit`` local was correctly derived from
+                    # cache_read>0; the funnel re-derives via the
+                    # outcome property — same result.
+                    #
+                    # ``attempted_input_tokens`` was MISSING from the
+                    # pre-refactor record_request call here (one of the
+                    # 7-of-18 sites the P0 audit flagged). The funnel
+                    # forces it to a value — using
+                    # ``optimized_tokens + tokens_saved`` as the
+                    # fallback denominator, same as the streaming path
+                    # uses (see _finalize_stream_response). Dashboards
+                    # that were showing 0% active-savings on non-
+                    # streaming Anthropic traffic will now show the
+                    # correct ratio.
+                    await self._record_request_outcome(
+                        RequestOutcome(
+                            request_id=request_id,
+                            provider="anthropic",
+                            model=model,
+                            original_tokens=original_tokens,
+                            optimized_tokens=optimized_tokens,
+                            output_tokens=output_tokens,
+                            tokens_saved=tokens_saved,
+                            attempted_input_tokens=optimized_tokens + tokens_saved,
+                            cache_read_tokens=cr_tokens,
+                            cache_write_tokens=cw_tokens,
+                            cache_write_5m_tokens=cw_5m_tokens,
+                            cache_write_1h_tokens=cw_1h_tokens,
+                            uncached_input_tokens=uncached_input_tokens,
+                            total_latency_ms=total_latency,
+                            overhead_ms=optimization_latency,
+                            pipeline_timing=pipeline_timing,
+                            waste_signals=waste_signals_dict,
+                            transforms_applied=tuple(transforms_applied),
+                            num_messages=len(messages),
+                            tags=tags,
+                            turn_id=compute_turn_id(
+                                model, body.get("system"), body.get("messages")
+                            ),
+                            request_messages=messages if self.config.log_full_messages else None,
                         )
-
-                    # Structured perf log line for `headroom perf` analysis
-                    num_msgs = len(messages)
-                    resp_usage = resp_json.get("usage", {}) if resp_json else {}
-                    cr = resp_usage.get("cache_read_input_tokens", 0)
-                    cw = resp_usage.get("cache_creation_input_tokens", 0)
-                    chp = round(cr / (cr + cw) * 100) if (cr + cw) > 0 else 0
-                    timing_str = (
-                        " ".join(f"{k}={v:.0f}ms" for k, v in pipeline_timing.items())
-                        if pipeline_timing
-                        else ""
-                    )
-                    logger.info(
-                        f"[{request_id}] PERF "
-                        f"model={model} msgs={num_msgs} "
-                        f"tok_before={original_tokens} tok_after={optimized_tokens} "
-                        f"tok_saved={tokens_saved} "
-                        f"cache_read={cr} cache_write={cw} cache_hit_pct={chp} "
-                        f"opt_ms={optimization_latency:.0f} "
-                        f"transforms={_summarize_transforms(transforms_applied)}"
-                        f"{' timing=' + timing_str if timing_str else ''}"
                     )
 
                     # Remove compression headers since httpx already decompressed the response
@@ -2509,16 +2477,27 @@ class AnthropicHandlerMixin:
                 path_for_log="/v1/messages/batches",
             )
 
-            # Record metrics
-            await self.metrics.record_request(
-                provider="anthropic",
-                model="batch",
-                input_tokens=total_optimized_tokens,
-                output_tokens=0,
-                tokens_saved=total_tokens_saved,
-                latency_ms=optimization_latency,
-                overhead_ms=optimization_latency,
-                pipeline_timing=pipeline_timing,
+            # Batch create: tokens accumulated across all requests in
+            # the batch. The funnel records it as a single observation
+            # under the synthetic model name "batch" — same as
+            # pre-refactor, just routed through the canonical path so
+            # batch traffic appears in RequestLog + PERF (it didn't
+            # before — sites 4/5/6 were "metrics-only").
+            await self._record_request_outcome(
+                RequestOutcome(
+                    request_id=request_id,
+                    provider="anthropic",
+                    model="batch",
+                    original_tokens=total_original_tokens,
+                    optimized_tokens=total_optimized_tokens,
+                    output_tokens=0,
+                    tokens_saved=total_tokens_saved,
+                    attempted_input_tokens=total_optimized_tokens + total_tokens_saved,
+                    total_latency_ms=optimization_latency,
+                    overhead_ms=optimization_latency,
+                    pipeline_timing=pipeline_timing,
+                    num_messages=len(compressed_requests),
+                )
             )
 
             # Log compression stats
@@ -2588,6 +2567,7 @@ class AnthropicHandlerMixin:
         """
         from fastapi.responses import Response
 
+        request_id = await self._next_request_id()
         start_time = time.time()
         path = request.url.path
         url = f"{self.ANTHROPIC_API_URL}{path}"
@@ -2618,15 +2598,22 @@ class AnthropicHandlerMixin:
             content=body,
         )
 
-        # Track metrics
+        # Batch passthrough: no compression, no transforms — but we
+        # still record the request so dashboards see the upstream call
+        # happened. Same funnel as the other 5 anthropic sites.
         latency_ms = (time.time() - start_time) * 1000
-        await self.metrics.record_request(
-            provider="anthropic",
-            model="passthrough:batches",
-            input_tokens=0,
-            output_tokens=0,
-            tokens_saved=0,
-            latency_ms=latency_ms,
+        await self._record_request_outcome(
+            RequestOutcome(
+                request_id=request_id,
+                provider="anthropic",
+                model="passthrough:batches",
+                original_tokens=0,
+                optimized_tokens=0,
+                output_tokens=0,
+                tokens_saved=0,
+                attempted_input_tokens=0,
+                total_latency_ms=latency_ms,
+            )
         )
 
         # Remove compression headers
@@ -2699,6 +2686,7 @@ class AnthropicHandlerMixin:
 
         from headroom.ccr import BatchResultProcessor, get_batch_context_store
 
+        request_id = await self._next_request_id()
         start_time = time.time()
 
         # Forward request to get raw results
@@ -2785,15 +2773,22 @@ class AnthropicHandlerMixin:
 
         processed_content = "\n".join(processed_lines)
 
-        # Track metrics
+        # Batch results, post-CCR processing. Like the other batch
+        # sites, no token accounting but we record the request so it's
+        # visible in dashboards + headroom perf.
         latency_ms = (time.time() - start_time) * 1000
-        await self.metrics.record_request(
-            provider="anthropic",
-            model="batch:ccr-processed",
-            input_tokens=0,
-            output_tokens=0,
-            tokens_saved=0,
-            latency_ms=latency_ms,
+        await self._record_request_outcome(
+            RequestOutcome(
+                request_id=request_id,
+                provider="anthropic",
+                model="batch:ccr-processed",
+                original_tokens=0,
+                optimized_tokens=0,
+                output_tokens=0,
+                tokens_saved=0,
+                attempted_input_tokens=0,
+                total_latency_ms=latency_ms,
+            )
         )
 
         return Response(
diff --git a/headroom/proxy/handlers/gemini.py b/headroom/proxy/handlers/gemini.py
index 1da75635a..b02ec82a6 100644
--- a/headroom/proxy/handlers/gemini.py
+++ b/headroom/proxy/handlers/gemini.py
@@ -15,6 +15,8 @@ if TYPE_CHECKING:
     from fastapi import Request
     from fastapi.responses import JSONResponse, Response, StreamingResponse
 
+from headroom.proxy.outcome import RequestOutcome
+
 logger = logging.getLogger("headroom.proxy")
 
 DEFAULT_CLOUDCODE_API_URL = "https://cloudcode-pa.googleapis.com"
@@ -466,35 +468,39 @@ class GeminiHandlerMixin:
 
                 uncached_input_tokens = max(0, total_input_tokens - cache_read_tokens)
 
-                if self.cost_tracker:
-                    self.cost_tracker.record_tokens(
-                        model,
-                        tokens_saved,
-                        optimized_tokens,
-                        cache_read_tokens=cache_read_tokens,
-                        uncached_tokens=uncached_input_tokens,
-                    )
-
                 # Eligible-tracking is TODO for Gemini; pass the full
                 # pre-compression request size as the fallback denominator.
                 # This makes Gemini's contribution to the aggregate
                 # active_savings_percent equal its whole-request ratio —
                 # not ideal but coherent until per-part live-zone
                 # tracking exists for this provider.
-                attempted_input_tokens = total_input_tokens + tokens_saved
-                await self.metrics.record_request(
+                #
+                # Gemini reports read-side context-cache only via
+                # ``cachedContentTokenCount``. There is no write counter
+                # in the Gemini response; cache writes happen out-of-band
+                # via the explicit Cache API. cache_write_* fields on the
+                # outcome stay at their 0 defaults — the dataclass
+                # handles "this provider doesn't have this concept"
+                # without per-handler conditionals.
+                outcome = RequestOutcome(
+                    request_id=request_id,
                     provider="gemini",
                     model=model,
-                    input_tokens=total_input_tokens,
+                    original_tokens=original_tokens,
+                    optimized_tokens=total_input_tokens,
                     output_tokens=output_tokens,
                     tokens_saved=tokens_saved,
-                    latency_ms=total_latency,
-                    overhead_ms=optimization_latency,
-                    waste_signals=waste_signals_dict,
+                    attempted_input_tokens=total_input_tokens + tokens_saved,
                     cache_read_tokens=cache_read_tokens,
                     uncached_input_tokens=uncached_input_tokens,
-                    attempted_input_tokens=attempted_input_tokens,
+                    total_latency_ms=total_latency,
+                    overhead_ms=optimization_latency,
+                    waste_signals=waste_signals_dict,
+                    transforms_applied=tuple(transforms_applied),
+                    num_messages=len(body.get("contents", [])),
+                    tags=tags or {},
                 )
+                await self._record_request_outcome(outcome)
 
                 if tokens_saved > 0:
                     logger.info(
@@ -902,15 +908,22 @@ class GeminiHandlerMixin:
 
             # Fallback denominator (see comment on the main gemini
             # record_request site) — pre-comp request size.
-            attempted_input_tokens = compressed_tokens + tokens_saved
-            await self.metrics.record_request(
-                provider="gemini",
-                model=model,
-                input_tokens=compressed_tokens,
-                output_tokens=0,
-                tokens_saved=tokens_saved,
-                latency_ms=total_latency,
-                attempted_input_tokens=attempted_input_tokens,
+            # countTokens is a sizing helper; it never generates output
+            # tokens and never touches cache. The funnel handles the
+            # "nothing to report" shape with all-zero cache defaults.
+            await self._record_request_outcome(
+                RequestOutcome(
+                    request_id=request_id,
+                    provider="gemini",
+                    model=model,
+                    original_tokens=original_tokens,
+                    optimized_tokens=compressed_tokens,
+                    output_tokens=0,
+                    tokens_saved=tokens_saved,
+                    attempted_input_tokens=compressed_tokens + tokens_saved,
+                    total_latency_ms=total_latency,
+                    transforms_applied=tuple(transforms_applied),
+                )
             )
 
             if tokens_saved > 0:
diff --git a/headroom/proxy/outcome.py b/headroom/proxy/outcome.py
index afdfe20ab..a501c4c00 100644
--- a/headroom/proxy/outcome.py
+++ b/headroom/proxy/outcome.py
@@ -7,9 +7,10 @@ disagreed on argument shape — 9 of 18 omitted ``cached=``, 7 of 18
 omitted ``attempted_input_tokens=``, only 4 sites emitted a structured
 PERF log at all. This module is the structural fix: every handler
 converges on building a :class:`RequestOutcome` at end-of-request and
-hands it to :meth:`HeadroomProxy._record_request_outcome`, which owns
-the four downstream effects (Prometheus, cost tracker, request logger,
-PERF log).
+hands it to :func:`emit_request_outcome` (also exposed as
+:meth:`HeadroomProxy._record_request_outcome`), which owns the four
+downstream effects (Prometheus, cost tracker, request logger, PERF
+log).
 
 Note: this is **output unification, not input unification**. Provider
 APIs (Anthropic ``/v1/messages``, OpenAI Responses WS, Gemini
@@ -24,9 +25,13 @@ actually reports.
 
 from __future__ import annotations
 
+import logging
 from dataclasses import dataclass, field
+from datetime import datetime
 from typing import Any
 
+logger = logging.getLogger("headroom.proxy")
+
 
 @dataclass(frozen=True)
 class RequestOutcome:
@@ -75,6 +80,12 @@ class RequestOutcome:
     cache_write_1h_tokens: int = 0
     uncached_input_tokens: int = 0
     cache_inferred: bool = False
+    # Response-cache hit (Headroom's own semantic cache served the
+    # response from a prior call — completely distinct from
+    # upstream-prompt-cache `cache_read_tokens`). True means the proxy
+    # never reached the provider at all. Used to drive the
+    # Prometheus ``cached`` counter and dashboard "response cache" row.
+    from_response_cache: bool = False
 
     # ── Timing ────────────────────────────────────────────────────────
     # total_latency_ms: wall-clock end-to-end for this request
@@ -111,13 +122,18 @@ class RequestOutcome:
 
     @property
     def cache_hit(self) -> bool:
-        """True iff upstream reported any cache read.
+        """True iff EITHER upstream reported a cache read OR the response
+        was served from Headroom's own response cache.
+
+        Two distinct concepts collapsed into one observable boolean for
+        downstream consumers (Prometheus ``cached`` counter, RequestLog
+        ``cache_hit`` flag). The dataclass tracks them separately so
+        dashboards can split them; the derived property unifies them.
 
-        Used by ``RequestLog.cache_hit`` and ``record_request(cached=...)``.
         Pre-refactor 9 of 18 sites hardcoded this to False — this property
         makes "I forgot to compute it" structurally impossible.
         """
-        return self.cache_read_tokens > 0
+        return self.cache_read_tokens > 0 or self.from_response_cache
 
     @property
     def cache_hit_pct(self) -> int:
@@ -144,3 +160,106 @@ class RequestOutcome:
         if self.original_tokens <= 0:
             return 0.0
         return self.tokens_saved / self.original_tokens * 100.0
+
+
+# ── The funnel ───────────────────────────────────────────────────────
+
+
+async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
+    """Single funnel for per-request bookkeeping. The contract.
+
+    Owns the four downstream effects in canonical order:
+
+      1. ``handler.metrics.record_request(...)`` — Prometheus / SavingsTracker
+      2. ``handler.cost_tracker.record_tokens(...)`` — cost dashboard
+         (skipped when cost_tracker is None, i.e. ``--no-cost``)
+      3. ``handler.logger.log(RequestLog(...))`` — per-request log feed
+         (skipped when logger is None, i.e. ``--no-request-logging``)
+      4. structured PERF log line — consumed by ``headroom perf``
+
+    Takes the handler as a free argument rather than ``self`` so this
+    function is callable from:
+    * ``HeadroomProxy._record_request_outcome`` (production)
+    * any test dummy that has the three required attributes
+      (``metrics``, ``cost_tracker``, optionally ``logger``)
+    * any provider handler mixin
+
+    The handler argument is structurally typed (duck-typed); no formal
+    Protocol — the requirement is simply that ``handler.metrics`` exists
+    and is awaitable-compatible. We could lift this to a typing.Protocol
+    if/when another contract surface emerges, but YAGNI.
+    """
+    from headroom.proxy.cost import _summarize_transforms
+    from headroom.proxy.models import RequestLog
+
+    # 1. Prometheus / SavingsTracker.
+    await handler.metrics.record_request(
+        provider=outcome.provider,
+        model=outcome.model,
+        input_tokens=outcome.optimized_tokens,
+        output_tokens=outcome.output_tokens,
+        tokens_saved=outcome.tokens_saved,
+        latency_ms=outcome.total_latency_ms,
+        cached=outcome.cache_hit,
+        overhead_ms=outcome.overhead_ms,
+        ttfb_ms=outcome.ttfb_ms,
+        pipeline_timing=outcome.pipeline_timing,
+        waste_signals=outcome.waste_signals,
+        cache_read_tokens=outcome.cache_read_tokens,
+        cache_write_tokens=outcome.cache_write_tokens,
+        cache_write_5m_tokens=outcome.cache_write_5m_tokens,
+        cache_write_1h_tokens=outcome.cache_write_1h_tokens,
+        uncached_input_tokens=outcome.uncached_input_tokens,
+        attempted_input_tokens=outcome.attempted_input_tokens,
+    )
+
+    # 2. Cost tracker (optional).
+    cost_tracker = getattr(handler, "cost_tracker", None)
+    if cost_tracker is not None:
+        cost_tracker.record_tokens(
+            outcome.model,
+            outcome.tokens_saved,
+            outcome.optimized_tokens,
+            cache_read_tokens=outcome.cache_read_tokens,
+            cache_write_tokens=outcome.cache_write_tokens,
+            cache_write_5m_tokens=outcome.cache_write_5m_tokens,
+            cache_write_1h_tokens=outcome.cache_write_1h_tokens,
+            uncached_tokens=outcome.uncached_input_tokens,
+        )
+
+    # 3. Per-request log (optional).
+    request_logger = getattr(handler, "logger", None)
+    if request_logger is not None:
+        request_logger.log(
+            RequestLog(
+                request_id=outcome.request_id,
+                timestamp=datetime.now().isoformat(),
+                provider=outcome.provider,
+                model=outcome.model,
+                input_tokens_original=outcome.original_tokens,
+                input_tokens_optimized=outcome.optimized_tokens,
+                output_tokens=outcome.output_tokens,
+                tokens_saved=outcome.tokens_saved,
+                savings_percent=outcome.savings_pct,
+                optimization_latency_ms=outcome.overhead_ms,
+                total_latency_ms=outcome.total_latency_ms,
+                tags=outcome.tags,
+                cache_hit=outcome.cache_hit,
+                transforms_applied=list(outcome.transforms_applied),
+                waste_signals=outcome.waste_signals,
+                request_messages=outcome.request_messages,
+                turn_id=outcome.turn_id,
+            )
+        )
+
+    # 4. Structured PERF log line.
+    logger.info(
+        f"[{outcome.request_id}] PERF "
+        f"model={outcome.model} msgs={outcome.num_messages} "
+        f"tok_before={outcome.original_tokens} tok_after={outcome.optimized_tokens} "
+        f"tok_saved={outcome.tokens_saved} "
+        f"cache_read={outcome.cache_read_tokens} cache_write={outcome.cache_write_tokens} "
+        f"cache_hit_pct={outcome.cache_hit_pct} "
+        f"opt_ms={outcome.overhead_ms:.0f} "
+        f"transforms={_summarize_transforms(list(outcome.transforms_applied))}"
+    )
diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py
index 2c6882eb6..41001a7ad 100644
--- a/headroom/proxy/server.py
+++ b/headroom/proxy/server.py
@@ -1153,93 +1153,19 @@ class HeadroomProxy(
     async def _record_request_outcome(self, outcome: RequestOutcome) -> None:
         """Single funnel for per-request bookkeeping.
 
-        Replaces the divergent four-call sequence (``metrics.record_request``
-        + ``cost_tracker.record_tokens`` + ``logger.log(RequestLog(...))``
-        + structured PERF log) that pre-refactor lived inline at 18 sites
-        across four handler files with subtly different argument shapes.
-        Provider handlers build a :class:`RequestOutcome` from local
-        context and call here — the wire shape for metrics/log/PERF is
-        defined exactly once, in this method, and nowhere else.
+        Thin wrapper around :func:`headroom.proxy.outcome.emit_request_outcome`
+        so call sites can write ``await self._record_request_outcome(outcome)``
+        (idiomatic) instead of ``await emit_request_outcome(self, outcome)``.
+        The real implementation lives in ``outcome.py`` as a free function so
+        test dummies and provider mixins can call it without inheriting from
+        ``HeadroomProxy``.
 
         See ``docs/superpowers/specs/P0-proxy-pipeline-audit.md`` for the
         divergence catalog this funnel collapses.
         """
+        from headroom.proxy.outcome import emit_request_outcome
 
-        # 1. Prometheus / SavingsTracker.
-        await self.metrics.record_request(
-            provider=outcome.provider,
-            model=outcome.model,
-            input_tokens=outcome.optimized_tokens,
-            output_tokens=outcome.output_tokens,
-            tokens_saved=outcome.tokens_saved,
-            latency_ms=outcome.total_latency_ms,
-            cached=outcome.cache_hit,
-            overhead_ms=outcome.overhead_ms,
-            ttfb_ms=outcome.ttfb_ms,
-            pipeline_timing=outcome.pipeline_timing,
-            waste_signals=outcome.waste_signals,
-            cache_read_tokens=outcome.cache_read_tokens,
-            cache_write_tokens=outcome.cache_write_tokens,
-            cache_write_5m_tokens=outcome.cache_write_5m_tokens,
-            cache_write_1h_tokens=outcome.cache_write_1h_tokens,
-            uncached_input_tokens=outcome.uncached_input_tokens,
-            attempted_input_tokens=outcome.attempted_input_tokens,
-        )
-
-        # 2. Cost tracker (optional — disabled when proxy --no-cost).
-        if self.cost_tracker:
-            self.cost_tracker.record_tokens(
-                outcome.model,
-                outcome.tokens_saved,
-                outcome.optimized_tokens,
-                cache_read_tokens=outcome.cache_read_tokens,
-                cache_write_tokens=outcome.cache_write_tokens,
-                cache_write_5m_tokens=outcome.cache_write_5m_tokens,
-                cache_write_1h_tokens=outcome.cache_write_1h_tokens,
-                uncached_tokens=outcome.uncached_input_tokens,
-            )
-
-        # 3. Per-request log (optional — disabled when proxy
-        #    --no-request-logging or when logger isn't installed).
-        request_logger = getattr(self, "logger", None)
-        if request_logger is not None:
-            request_logger.log(
-                RequestLog(
-                    request_id=outcome.request_id,
-                    timestamp=datetime.now().isoformat(),
-                    provider=outcome.provider,
-                    model=outcome.model,
-                    input_tokens_original=outcome.original_tokens,
-                    input_tokens_optimized=outcome.optimized_tokens,
-                    output_tokens=outcome.output_tokens,
-                    tokens_saved=outcome.tokens_saved,
-                    savings_percent=outcome.savings_pct,
-                    optimization_latency_ms=outcome.overhead_ms,
-                    total_latency_ms=outcome.total_latency_ms,
-                    tags=outcome.tags,
-                    cache_hit=outcome.cache_hit,
-                    transforms_applied=list(outcome.transforms_applied),
-                    waste_signals=outcome.waste_signals,
-                    request_messages=outcome.request_messages,
-                    turn_id=outcome.turn_id,
-                )
-            )
-
-        # 4. Structured PERF log — consumed by ``headroom perf``.
-        #    Format frozen so the analyzer's key=value parser keeps
-        #    working; future P3 work replaces the free-text shape with a
-        #    structured event, at which point this becomes a presentation
-        #    of the event rather than a parallel log line.
-        logger.info(
-            f"[{outcome.request_id}] PERF "
-            f"model={outcome.model} msgs={outcome.num_messages} "
-            f"tok_before={outcome.original_tokens} tok_after={outcome.optimized_tokens} "
-            f"tok_saved={outcome.tokens_saved} "
-            f"cache_read={outcome.cache_read_tokens} cache_write={outcome.cache_write_tokens} "
-            f"cache_hit_pct={outcome.cache_hit_pct} "
-            f"opt_ms={outcome.overhead_ms:.0f} "
-            f"transforms={_summarize_transforms(list(outcome.transforms_applied))}"
-        )
+        await emit_request_outcome(self, outcome)
 
     async def _next_request_id(self) -> str:
         """Generate unique request ID."""
diff --git a/tests/test_anthropic_pre_upstream_backpressure.py b/tests/test_anthropic_pre_upstream_backpressure.py
index bd198cce7..8b1496118 100644
--- a/tests/test_anthropic_pre_upstream_backpressure.py
+++ b/tests/test_anthropic_pre_upstream_backpressure.py
@@ -215,6 +215,14 @@ class _DummyAnthropicHandler(AnthropicHandlerMixin):
         future = loop.run_in_executor(self._compression_executor, _wrapped)
         return await asyncio.wait_for(future, timeout=timeout)
 
+    async def _record_request_outcome(self, outcome) -> None:  # noqa: ANN001
+        # Mirror of ``HeadroomProxy._record_request_outcome`` for the
+        # mixin tests. Delegates to the free function in ``outcome.py``
+        # so the wire shape is identical to production.
+        from headroom.proxy.outcome import emit_request_outcome
+
+        await emit_request_outcome(self, outcome)
+
     async def _next_request_id(self) -> str:
         # Unique IDs so log assertions remain disambiguated under parallelism.
         return f"req-{id(object()):x}"