akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor(proxy): migrate Gemini + Anthropic non-streaming onto outcome funnel

Browse source 
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
This commit is contained in:
chopratejas 2026-05-14 14:39:30 -07:00
parent e898f68b89
commit 7aca00445e
5 changed files with 334 additions and 273 deletions
Download patch file Download diff file Expand all files Collapse all files

315
headroom/proxy/handlers/anthropic.py
View file

@ -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(

63
headroom/proxy/handlers/gemini.py
View file

@ -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:

131
headroom/proxy/outcome.py
View file

@ -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))}"
)

90
headroom/proxy/server.py
View file

@ -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."""

8
tests/test_anthropic_pre_upstream_backpressure.py
View file

@ -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}"