Unify savings attribution across stats, perf, metrics, and dashboard (#2976)

## Summary

Adds a small provider-neutral savings attribution seam. Named sources
can attach realized or projected token/USD deltas to a request without
changing headline arithmetic or introducing private-package inventory
into OSS.

Also fixes the Anthropic buffered lifecycle so normal successful
responses run response hooks, applies stream-safety filtering, includes
tool savings in per-model perf totals, and surfaces the same breakdown
in request logs, `/stats`, `headroom perf`, Prometheus, OTEL, and the
dashboard.

## Why

Request-local savings were split between canonical token deltas,
process-global extension counters, and tool-only tags. This made correct
headline totals possible while losing attribution in perf, recent
requests, metrics, and the dashboard. Normal Anthropic responses also
skipped response hooks unless CCR ran.

## Validation

- 74 focused tests passed: turn hooks, OpenAI hook lifecycle, outcome
funnel, perf formats, and tool-search repair
- Ruff passes on all changed Python files
- Existing compression-observability suite: 11 passed; 2 tokenizer-cache
tests require network access to fetch the tiktoken vocabulary

## Compatibility

No named private packages or private inventory are encoded in OSS.
Existing hooks remain source-compatible because all new TurnContext
fields are optional.

---------

Co-authored-by: Tejas Chopra <tejas@Tejass-MacBook-Pro.local>

headroom/cli/perf.py

@@ -79,6 +79,8 @@ def perf(hours: float, raw: bool, output_format: str) -> None:
                 "tokens_before",
                 "tokens_after",
                 "tokens_saved",
+                "message_tokens_saved",
+                "tool_tokens_saved",
                 "savings_pct",
                 "list_price_per_mtok",
             ]

headroom/dashboard/templates/dashboard.html

@@ -270,20 +270,23 @@
                         </template>
                     </div>
 
-                    <!-- Tool-schema deferral: a COMPONENT of the Tokens Saved headline above
-                         (stats.tokens.saved is all-layers), not a rival metric. Labelled as
-                         such so a tool-heavy session doesn't read as "0 saved + some other
-                         number". Only rendered when there's a saving to show. -->
-                    <template x-if="(stats.savings?.by_layer?.tool_search?.tokens || 0) > 0">
+                    <!-- Dynamic attribution for any named savings source. These
+                         rows explain the headline; they are not added to it. -->
+                    <template x-for="row in (stats.savings?.by_source || [])" :key="row.source + ':' + row.realized">
                         <div class="bg-surface rounded-lg p-4 border border-border">
-                            <div class="text-xs text-gray-500 uppercase tracking-wide mb-1">Tokens Saved · Tool Schemas</div>
-                            <div class="flex items-baseline gap-2">
-                                <span class="text-3xl font-light tabular-nums text-emerald-400" x-text="formatNumber(stats.savings?.by_layer?.tool_search?.tokens || 0)"></span>
+                            <div class="text-xs text-gray-500 uppercase tracking-wide mb-1"
+                                 x-text="'Savings · ' + row.source.replaceAll('_', ' ')"></div>
+                            <div class="flex items-baseline gap-2" x-show="(row.tokens || 0) > 0">
+                                <span class="text-3xl font-light tabular-nums text-emerald-400" x-text="formatNumber(row.tokens || 0)"></span>
                                 <span class="text-sm text-gray-400">tokens</span>
                             </div>
-                            <div class="mt-1 text-xs text-gray-600 leading-relaxed"
-                                 x-text="formatNumber(stats.savings?.by_layer?.tool_search?.requests || 0) + ' calls · tool schemas deferred (recent window)'">
+                            <div class="flex items-baseline gap-2" x-show="(row.usd || 0) !== 0">
+                                <span class="text-2xl font-light tabular-nums"
+                                      :class="row.usd >= 0 ? 'text-emerald-400' : 'text-red-400'"
+                                      x-text="(row.usd >= 0 ? '$' : '-$') + formatCurrency(Math.abs(row.usd || 0))"></span>
                             </div>
+                            <div class="mt-1 text-xs text-gray-600 leading-relaxed"
+                                 x-text="formatNumber(row.events || 0) + ' calls · ' + (row.realized ? 'realized' : 'projected')"></div>
                         </div>
                     </template>
                 </div>
@@ -934,6 +937,17 @@
                                                         </div>
                                                     </div>
                                                 </template>
+                                                <template x-if="(req.savings_breakdown || []).length > 0">
+                                                    <div class="mt-3">
+                                                        <div class="text-xs text-gray-500 uppercase tracking-wide mb-1">Savings Attribution</div>
+                                                        <div class="flex flex-wrap gap-1">
+                                                            <template x-for="item in req.savings_breakdown" :key="item.source + ':' + item.tokens + ':' + item.usd">
+                                                                <span class="px-2 py-0.5 bg-border rounded text-xs font-mono"
+                                                                      x-text="item.source + (item.tokens ? ' · ' + formatNumber(item.tokens) + ' tok' : '') + (item.usd ? ' · $' + formatCurrency(item.usd) : '') + (item.realized ? '' : ' · projected')"></span>
+                                                            </template>
+                                                        </div>
+                                                    </div>
+                                                </template>
                                                 <!-- Waste Signals for this request -->
                                                 <template x-if="req.waste_signals && Object.keys(req.waste_signals).length > 0">
                                                     <div class="mt-3">

headroom/observability/__init__.py

@@ -7,9 +7,11 @@ from .metrics import (
     get_otel_meter,
     get_otel_metrics,
     get_otel_metrics_status,
+    register_otel_metric_attribute_provider,
     reset_otel_metrics,
     set_otel_metrics,
     shutdown_otel_metrics,
+    unregister_otel_metric_attribute_provider,
 )
 from .tracing import (
     HeadroomTracer,
@@ -29,6 +31,7 @@ __all__ = [
     "get_otel_meter",
     "get_otel_metrics",
     "get_otel_metrics_status",
+    "register_otel_metric_attribute_provider",
     "HeadroomTracer",
     "LangfuseTracingConfig",
     "configure_langfuse_tracing",
@@ -40,4 +43,5 @@ __all__ = [
     "set_headroom_tracer",
     "shutdown_headroom_tracing",
     "shutdown_otel_metrics",
+    "unregister_otel_metric_attribute_provider",
 ]

headroom/observability/metrics.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import logging
 import os
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass, field
 from threading import Lock
 from typing import Any, Literal
@@ -26,6 +27,67 @@ _global_metrics: HeadroomOtelMetrics | None = None
 _owned_meter_provider: Any | None = None
 _owned_metrics_config: OTelMetricsConfig | None = None
 
+MetricAttributeProvider = Callable[[], Mapping[str, Any]]
+_metric_attribute_providers: list[MetricAttributeProvider] = []
+_metric_attribute_providers_lock = Lock()
+_MAX_DYNAMIC_ATTRIBUTES = 16
+_MAX_DYNAMIC_ATTRIBUTE_LENGTH = 256
+
+
+def register_otel_metric_attribute_provider(
+    provider: MetricAttributeProvider,
+) -> MetricAttributeProvider:
+    """Add request-scoped attributes to every Headroom OTEL datapoint.
+
+    Extensions use this narrow seam for dimensions such as tenant, team, or
+    user identity without coupling the OSS metrics layer to an auth package.
+    Providers run in the request context and must return content-free scalar
+    labels. A failing provider is ignored so observability cannot break traffic.
+    """
+
+    with _metric_attribute_providers_lock:
+        if provider not in _metric_attribute_providers:
+            _metric_attribute_providers.append(provider)
+    return provider
+
+
+def unregister_otel_metric_attribute_provider(provider: MetricAttributeProvider) -> None:
+    """Remove a previously registered request-attribute provider."""
+
+    with _metric_attribute_providers_lock:
+        try:
+            _metric_attribute_providers.remove(provider)
+        except ValueError:
+            pass
+
+
+def _dynamic_metric_attributes() -> dict[str, Any]:
+    with _metric_attribute_providers_lock:
+        providers = tuple(_metric_attribute_providers)
+
+    resolved: dict[str, Any] = {}
+    for provider in providers:
+        try:
+            attributes = provider()
+        except Exception:
+            logger.debug("OTEL metric attribute provider failed", exc_info=True)
+            continue
+        if not isinstance(attributes, Mapping):
+            continue
+        for raw_key, raw_value in attributes.items():
+            if len(resolved) >= _MAX_DYNAMIC_ATTRIBUTES:
+                return resolved
+            key = str(raw_key).strip()
+            if not key or raw_value is None or raw_value == "":
+                continue
+            if not isinstance(raw_value, (str, bool, int, float)):
+                continue
+            value = raw_value
+            if isinstance(value, str):
+                value = value[:_MAX_DYNAMIC_ATTRIBUTE_LENGTH]
+            resolved[key[:_MAX_DYNAMIC_ATTRIBUTE_LENGTH]] = value
+    return resolved
+
 
 def _headroom_version() -> str:
     return get_version()
@@ -163,6 +225,25 @@ class HeadroomOtelMetrics:
             description="Output tokens returned by upstream providers.",
             unit="1",
         )
+        self._proxy_attempted_input_tokens = self._meter.create_counter(
+            "headroom.proxy.tokens.attempted_input",
+            description=(
+                "Input tokens Headroom attempted to optimize before compression."
+            ),
+            unit="1",
+        )
+        self._proxy_output_saved_tokens = self._meter.create_counter(
+            "headroom.proxy.tokens.output_saved",
+            description="Estimated output tokens avoided by Headroom optimization.",
+            unit="1",
+        )
+        self._proxy_savings_usd = self._meter.create_counter(
+            "headroom.proxy.savings.usd",
+            description=(
+                "Estimated savings in USD by distinct Headroom or provider-cache layer."
+            ),
+            unit="USD",
+        )
         self._proxy_saved_tokens = self._meter.create_counter(
             "headroom.proxy.tokens.saved",
             description=(
@@ -265,6 +346,21 @@ class HeadroomOtelMetrics:
             description="Waste tokens detected in compressed inputs.",
             unit="1",
         )
+        self._savings_attribution_events = self._meter.create_counter(
+            "headroom.savings.attribution.events",
+            description="Per-request savings attribution events.",
+            unit="1",
+        )
+        self._savings_attributed_tokens = self._meter.create_counter(
+            "headroom.savings.attributed.tokens",
+            description="Tokens attributed to a named savings source.",
+            unit="1",
+        )
+        self._savings_attributed_usd = self._meter.create_up_down_counter(
+            "headroom.savings.attributed.usd",
+            description="Attributed cost delta; negative values represent added cost.",
+            unit="USD",
+        )
 
         # Backing values updated by record_subscription_window()
         self._sub_5h_util_val: float = 0.0
@@ -337,7 +433,9 @@ class HeadroomOtelMetrics:
 
     @staticmethod
     def _attrs(**attrs: Any) -> dict[str, Any]:
-        filtered: dict[str, Any] = {}
+        # Dynamic request dimensions are deliberately lower precedence than
+        # canonical instrument dimensions (provider/model/source/etc.).
+        filtered = _dynamic_metric_attributes()
         for key, value in attrs.items():
             if value is None or value == "":
                 continue
@@ -362,8 +460,21 @@ class HeadroomOtelMetrics:
         cache_write_5m_tokens: int = 0,
         cache_write_1h_tokens: int = 0,
         uncached_input_tokens: int = 0,
+        attempted_input_tokens: int = 0,
+        output_tokens_saved: int = 0,
+        savings_usd: Mapping[str, float] | None = None,
+        project: str | None = None,
+        client: str | None = None,
     ) -> None:
-        attrs = self._attrs(provider=provider, model=model, cached=cached)
+        attrs = self._attrs(
+            provider=provider,
+            model=model,
+            cached=cached,
+            **{
+                "headroom.project": project,
+                "headroom.client": client,
+            },
+        )
 
         self._proxy_requests.add(1, attrs)
         if cached:
@@ -371,6 +482,17 @@ class HeadroomOtelMetrics:
 
         self._proxy_input_tokens.add(max(input_tokens, 0), attrs)
         self._proxy_output_tokens.add(max(output_tokens, 0), attrs)
+        if attempted_input_tokens > 0:
+            self._proxy_attempted_input_tokens.add(attempted_input_tokens, attrs)
+        if output_tokens_saved > 0:
+            self._proxy_output_saved_tokens.add(output_tokens_saved, attrs)
+        for source, value in (savings_usd or {}).items():
+            amount = max(float(value or 0.0), 0.0)
+            if amount:
+                self._proxy_savings_usd.add(
+                    amount,
+                    {**attrs, "source": str(source)[:64], "estimated": True},
+                )
         compression_saved = max(tokens_saved, 0)
         tool_schema_saved = max(tool_search_saved, 0)
         self._proxy_saved_tokens.add(compression_saved + tool_schema_saved, attrs)
@@ -402,6 +524,21 @@ class HeadroomOtelMetrics:
     def record_proxy_failed(self, *, provider: str | None = None, model: str | None = None) -> None:
         self._proxy_failed_requests.add(1, self._attrs(provider=provider, model=model))
 
+    def record_savings_attribution(self, items: list[dict[str, Any]]) -> None:
+        for item in items:
+            attrs = self._attrs(
+                source=str(item.get("source") or "other")[:64],
+                realized=bool(item.get("realized", True)),
+                estimated=bool(item.get("estimated", False)),
+            )
+            self._savings_attribution_events.add(1, attrs)
+            saved = max(0, int(item.get("tokens", 0) or 0))
+            if saved:
+                self._savings_attributed_tokens.add(saved, attrs)
+            cost = float(item.get("usd", 0.0) or 0.0)
+            if cost:
+                self._savings_attributed_usd.add(cost, attrs)
+
     def record_proxy_rate_limited(
         self,
         *,

headroom/perf/analyzer.py

@@ -134,6 +134,16 @@ def _parse_kv(kv_str: str) -> dict[str, str]:
     return result
 
 
+def _decode_perf_savings(value: str) -> list[dict[str, object]]:
+    # Local import keeps the analyzer usable against old logs/install layouts.
+    try:
+        from headroom.proxy.savings_attribution import decode
+
+        return decode(value)
+    except Exception:
+        return []
+
+
 @dataclass
 class PerfRecord:
     """A single parsed PERF log entry."""
@@ -156,6 +166,7 @@ class PerfRecord:
     tokens_out: int = 0
     ttfb_ms: float = 0.0
     stages: dict[str, float] = field(default_factory=dict)
+    savings_breakdown: list[dict[str, object]] = field(default_factory=list)
 
 
 @dataclass
@@ -353,6 +364,7 @@ def parse_log_files(last_n_hours: float = 168.0) -> PerfReport:
                                 tokens_after=int(kv.get("tok_after", 0)),
                                 tokens_saved=int(kv.get("tok_saved", 0)),
                                 tool_saved=int(kv.get("tool_saved", 0)),
+                                savings_breakdown=_decode_perf_savings(kv.get("savings", "none")),
                                 cache_read=int(kv.get("cache_read", 0)),
                                 cache_write=int(kv.get("cache_write", 0)),
                                 cache_hit_pct=int(kv.get("cache_hit_pct", 0)),
@@ -752,6 +764,7 @@ PERF_RECORD_FIELDS = [
     "tokens_out",
     "ttfb_ms",
     "stages",
+    "savings_breakdown",
 ]
 
 
@@ -1001,7 +1014,9 @@ def build_perf_summary(report: PerfReport) -> dict:
     for model, recs in sorted(by_model_groups.items()):
         m_before = sum(r.tokens_before for r in recs)
         m_after = sum(r.tokens_after for r in recs)
-        m_saved = sum(r.tokens_saved for r in recs)
+        m_message_saved = sum(r.tokens_saved for r in recs)
+        m_tool_saved = sum(r.tool_saved for r in recs)
+        m_saved = m_message_saved + m_tool_saved
         by_model.append(
             {
                 "model": model,
@@ -1009,7 +1024,9 @@ def build_perf_summary(report: PerfReport) -> dict:
                 "tokens_before": m_before,
                 "tokens_after": m_after,
                 "tokens_saved": m_saved,
-                "savings_pct": _pct(m_saved, m_before),
+                "message_tokens_saved": m_message_saved,
+                "tool_tokens_saved": m_tool_saved,
+                "savings_pct": _pct(m_saved, m_before + m_tool_saved),
                 "list_price_per_mtok": _get_list_price(model),
             }
         )
@@ -1033,6 +1050,37 @@ def build_perf_summary(report: PerfReport) -> dict:
             }
         )
 
+    by_source_groups: dict[tuple[str, bool], dict[str, int | float | str | bool]] = {}
+    for record in records:
+        for item in record.savings_breakdown:
+            source = str(item.get("source") or "other")
+            realized = bool(item.get("realized", True))
+            key = (source, realized)
+            row = by_source_groups.setdefault(
+                key,
+                {
+                    "source": source,
+                    "realized": realized,
+                    "events": 0,
+                    "tokens": 0,
+                    "usd": 0.0,
+                },
+            )
+            row["events"] = int(row["events"]) + 1
+            raw_tokens = item.get("tokens", 0)
+            raw_usd = item.get("usd", 0.0)
+            tokens = int(raw_tokens) if isinstance(raw_tokens, (str, int, float)) else 0
+            usd = float(raw_usd) if isinstance(raw_usd, (str, int, float)) else 0.0
+            row["tokens"] = int(row["tokens"]) + max(0, tokens)
+            row["usd"] = round(
+                float(row["usd"]) + usd,
+                12,
+            )
+    by_source = sorted(
+        by_source_groups.values(),
+        key=lambda row: (-int(row["tokens"]), str(row["source"])),
+    )
+
     return {
         "window_hours": report.requested_hours,
         "actual_window": {
@@ -1056,6 +1104,7 @@ def build_perf_summary(report: PerfReport) -> dict:
         "cache_hit_pct": cache_hit_pct,
         "by_model": by_model,
         "by_transform": by_transform,
+        "by_source": by_source,
         "overhead": build_overhead_summary(report),
         "throughput": calculate_throughput(report),
         "log_files_read": report.log_files_read,

headroom/proxy/handlers/anthropic.py

@@ -46,6 +46,115 @@ from headroom.proxy.outcome import RequestOutcome
 logger = logging.getLogger("headroom.proxy")
 
 
+class _AnthropicTurnHookUsage:
+    """Usage from hook-triggered Anthropic calls the main response omits.
+
+    The handler accounts for the response ultimately returned to the client.
+    A turn hook may make additional, billed calls before selecting that final
+    response, so retain every response and settle by object identity. This is
+    the Anthropic counterpart of OpenAI's ``TurnHookUsage`` and includes the
+    provider's disjoint uncached/read/write input buckets.
+    """
+
+    __slots__ = (
+        "_seen",
+        "input_tokens",
+        "output_tokens",
+        "cache_read_tokens",
+        "cache_write_tokens",
+        "cache_write_5m_tokens",
+        "cache_write_1h_tokens",
+        "extra_calls",
+    )
+
+    def __init__(self) -> None:
+        self._seen: list[tuple[Any, int, int, int, int, int, int]] = []
+        self.input_tokens = 0
+        self.output_tokens = 0
+        self.cache_read_tokens = 0
+        self.cache_write_tokens = 0
+        self.cache_write_5m_tokens = 0
+        self.cache_write_1h_tokens = 0
+        self.extra_calls = 0
+
+    @staticmethod
+    def _int(value: Any) -> int:
+        try:
+            return max(0, int(value or 0))
+        except (TypeError, ValueError):
+            return 0
+
+    def record(self, payload: Any) -> None:
+        usage = payload.get("usage") if isinstance(payload, dict) else None
+        if not isinstance(usage, dict):
+            self._seen.append((payload, 0, 0, 0, 0, 0, 0))
+            return
+        cache_read = self._int(usage.get("cache_read_input_tokens"))
+        cache_write = self._int(usage.get("cache_creation_input_tokens"))
+        creation = usage.get("cache_creation")
+        cache_write_5m = (
+            self._int(creation.get("ephemeral_5m_input_tokens"))
+            if isinstance(creation, dict)
+            else 0
+        )
+        cache_write_1h = (
+            self._int(creation.get("ephemeral_1h_input_tokens"))
+            if isinstance(creation, dict)
+            else 0
+        )
+        self._seen.append(
+            (
+                payload,
+                _anthropic_provider_input_tokens(usage),
+                self._int(usage.get("output_tokens")),
+                cache_read,
+                cache_write,
+                cache_write_5m,
+                cache_write_1h,
+            )
+        )
+
+    def settle(self, final: Any) -> None:
+        self.input_tokens = self.output_tokens = 0
+        self.cache_read_tokens = self.cache_write_tokens = 0
+        self.cache_write_5m_tokens = self.cache_write_1h_tokens = 0
+        self.extra_calls = 0
+        dropped = False
+        for (
+            payload,
+            input_tokens,
+            output_tokens,
+            cache_read,
+            cache_write,
+            cache_write_5m,
+            cache_write_1h,
+        ) in self._seen:
+            if not dropped and payload is final:
+                dropped = True
+                continue
+            self.extra_calls += 1
+            self.input_tokens += input_tokens
+            self.output_tokens += output_tokens
+            self.cache_read_tokens += cache_read
+            self.cache_write_tokens += cache_write
+            self.cache_write_5m_tokens += cache_write_5m
+            self.cache_write_1h_tokens += cache_write_1h
+
+
+def _anthropic_provider_input_tokens(usage: Any) -> int:
+    """Provider-scaled prompt total across Anthropic's disjoint buckets."""
+    if not isinstance(usage, dict):
+        return 0
+    return sum(
+        _AnthropicTurnHookUsage._int(usage.get(key))
+        for key in (
+            "input_tokens",
+            "cache_read_input_tokens",
+            "cache_creation_input_tokens",
+        )
+    )
+
+
 def _strip_streaming_only_content_fields(messages: Any) -> None:
     """Remove streaming-only ``index`` keys from request content blocks, in place.
 
@@ -866,6 +975,9 @@ class AnthropicHandlerMixin:
             # body is undecipherable → 502.
             headers.pop("accept-encoding", None)
             tags = extract_tags(headers)
+            from headroom.proxy.savings_attribution import bind_scope
+
+            bind_scope(tags, request.scope)
             # Identify the harness (codex / claude-code / aider / etc.)
             # from User-Agent or X-Client. Surfaced via the funnel into
             # PERF logs and RequestLog.tags — see RequestOutcome.client.
@@ -2652,6 +2764,9 @@ class AnthropicHandlerMixin:
                     tools = _ts_after
                     tags["tool_search_deferred_tools"] = len(_ts_deferred)
                     tags["tool_search_deferred_tokens"] = _ts_saved_tokens
+                    from headroom.proxy.savings_attribution import record_savings
+
+                    record_savings(tags, "tool_search", tokens=_ts_saved_tokens)
                     transforms_applied.append(
                         f"router:tool_search_deferral:{len(_ts_deferred)}tools:"
                         f"{_ts_saved_tokens}tok"
@@ -2668,6 +2783,7 @@ class AnthropicHandlerMixin:
             )
 
             _pre_hook_tokens: int | None = None
+            _req_ctx: TurnContext | None = None
             if registered_turn_hooks():
                 _req_ctx = TurnContext(
                     provider="anthropic",
@@ -2675,6 +2791,9 @@ class AnthropicHandlerMixin:
                     messages=optimized_messages,
                     tools=body.get("tools"),
                     config=self.config,
+                    tags=tags,
+                    count_messages=tokenizer.count_messages,
+                    count_tools=_count_tool_tokens,
                 )
                 # Snapshot BEFORE the hook (same tokenizer) so we can tell whether the
                 # hook itself folded — comparing against the pipeline's optimized_tokens
@@ -2685,7 +2804,7 @@ class AnthropicHandlerMixin:
                     _pre_hook_tokens = None
                 _th_tools_before = body.get("tools")
                 _th_tok_before = _count_tool_tokens(_th_tools_before) if _th_tools_before else 0
-                run_request_hooks(_req_ctx)
+                run_request_hooks(_req_ctx, stream_safe_only=bool(stream))
                 if _req_ctx.messages is not optimized_messages:
                     optimized_messages = _req_ctx.messages
                     body["messages"] = optimized_messages
@@ -3114,6 +3233,9 @@ class AnthropicHandlerMixin:
                                 output_tokens=output_tokens,
                                 tokens_saved=tokens_saved,
                                 attempted_input_tokens=attempted_input_tokens,
+                                provider_input_tokens=(
+                                    uncached_input_tokens + cr_tokens + cw_tokens
+                                ),
                                 cache_read_tokens=cr_tokens,
                                 cache_write_tokens=cw_tokens,
                                 cache_write_5m_tokens=cw_5m_tokens,
@@ -3612,26 +3734,6 @@ class AnthropicHandlerMixin:
                                 )
                                 # Update response content with final response
                                 resp_json = final_resp_json
-                                # Turn hooks (opt-in extensions) may inspect the turn or
-                                # re-drive the model before we hand back the response.
-                                # Inert when no hook is registered.
-                                from headroom.proxy.turn_hooks import (
-                                    TurnContext,
-                                    run_response_hooks,
-                                )
-
-                                final_resp_json = await run_response_hooks(
-                                    TurnContext(
-                                        provider="anthropic",
-                                        model=str(model),
-                                        messages=optimized_messages,
-                                        tools=tools,
-                                        config=self.config,
-                                    ),
-                                    final_resp_json,
-                                    api_call_fn,
-                                )
-                                resp_json = final_resp_json
                                 # Remove encoding headers since content is now uncompressed JSON
                                 ccr_response_headers = {
                                     k: v
@@ -3745,6 +3847,47 @@ class AnthropicHandlerMixin:
                                 )
                                 # Continue with original response
 
+                        # Buffered response hooks run for every successful turn,
+                        # not only the CCR branch. Reuse the request context so
+                        # observers close the exact turn they opened and a
+                        # search/reload hook can consume its synthetic tool call.
+                        _hook_usage = _AnthropicTurnHookUsage()
+                        if _req_ctx is not None and resp_json and response.status_code == 200:
+                            from headroom.proxy.turn_hooks import run_response_hooks
+
+                            _hook_usage.record(resp_json)
+
+                            async def _turn_hook_call_model(
+                                hook_messages: list[dict[str, Any]],
+                            ) -> dict[str, Any]:
+                                continuation_body = {**body, "messages": hook_messages}
+                                continuation_response = await self._retry_request(
+                                    "POST",
+                                    url,
+                                    headers,
+                                    continuation_body,
+                                    timeout=self._anthropic_buffered_request_timeout(),
+                                )
+                                continuation_json = continuation_response.json()
+                                _hook_usage.record(continuation_json)
+                                return continuation_json
+
+                            hooked_json = await run_response_hooks(
+                                _req_ctx, resp_json, _turn_hook_call_model
+                            )
+                            _hook_usage.settle(hooked_json)
+                            if hooked_json is not resp_json:
+                                resp_json = hooked_json
+                                response = httpx.Response(
+                                    status_code=200,
+                                    content=json.dumps(resp_json).encode(),
+                                    headers={
+                                        key: value
+                                        for key, value in response.headers.items()
+                                        if key.lower() not in ("content-encoding", "content-length")
+                                    },
+                                )
+
                         total_latency = (time.time() - start_time) * 1000
 
                         # Parse response for output token count and cache metrics
@@ -3757,12 +3900,23 @@ class AnthropicHandlerMixin:
                         if resp_json:
                             usage = resp_json.get("usage", {})
                             output_tokens = int(usage.get("output_tokens", 0) or 0)
+                            output_tokens += _hook_usage.output_tokens
                             cr_tokens = int(usage.get("cache_read_input_tokens", 0) or 0)
+                            cr_tokens += _hook_usage.cache_read_tokens
                             cw_tokens = int(usage.get("cache_creation_input_tokens", 0) or 0)
+                            cw_tokens += _hook_usage.cache_write_tokens
                             cw_5m_tokens, cw_1h_tokens = self._extract_anthropic_cache_ttl_metrics(
                                 usage
                             )
+                            cw_5m_tokens += _hook_usage.cache_write_5m_tokens
+                            cw_1h_tokens += _hook_usage.cache_write_1h_tokens
                             uncached_input_tokens = int(usage.get("input_tokens", 0) or 0)
+                            uncached_input_tokens += max(
+                                0,
+                                _hook_usage.input_tokens
+                                - _hook_usage.cache_read_tokens
+                                - _hook_usage.cache_write_tokens,
+                            )
 
                         # Track cache bust: tokens that lost their cache discount due to compression.
                         # If we had X tokens cached last turn and only Y hit cache this turn,
@@ -3894,6 +4048,9 @@ class AnthropicHandlerMixin:
                                 status_code=response.status_code,
                                 original_tokens=original_tokens,
                                 optimized_tokens=optimized_tokens,
+                                provider_input_tokens=(
+                                    uncached_input_tokens + cr_tokens + cw_tokens
+                                ),
                                 output_tokens=output_tokens,
                                 tokens_saved=tokens_saved,
                                 attempted_input_tokens=optimized_tokens + tokens_saved,

headroom/proxy/handlers/openai.py

@@ -2478,6 +2478,7 @@ class OpenAIHandlerMixin:
         request_id: str,
         timing: dict[str, float] | None = None,
         client: str | None = None,
+        savings_tags: dict[str, Any] | None = None,
     ) -> tuple[dict[str, Any], bool, int, list[str], str | None, int, int, int]:
         """Compress an OpenAI Responses payload through the shared router.
 
@@ -2624,6 +2625,23 @@ class OpenAIHandlerMixin:
             working["tools"] = _deferred_tools
             modified = True
             transforms.append("openai:responses:tool_search_deferral")
+            try:
+                from headroom.proxy.savings_attribution import record_savings
+
+                deferred = [
+                    tool
+                    for tool in _deferred_tools
+                    if isinstance(tool, dict) and tool.get("defer_loading")
+                ]
+                record_savings(
+                    savings_tags if savings_tags is not None else {},
+                    "tool_search",
+                    tokens=self.openai_provider.get_token_counter(model).count_text(
+                        _json_debug_dumps(deferred)
+                    ),
+                )
+            except Exception:
+                logger.debug("tool-search savings attribution skipped", exc_info=True)
 
         # Turn hooks (opt-in extensions): a registered hook may inspect or rewrite
         # the outbound tools before we send — the extensible counterpart to the
@@ -2662,6 +2680,17 @@ class OpenAIHandlerMixin:
                 messages=_msgs_before,
                 tools=working.get("tools"),
                 config=getattr(self, "config", None),
+                tags=savings_tags if savings_tags is not None else {},
+                count_messages=lambda value: self.openai_provider.get_token_counter(
+                    model
+                ).count_text(_json_debug_dumps(value)),
+                count_tools=lambda value: (
+                    self.openai_provider.get_token_counter(model).count_text(
+                        _json_debug_dumps(value)
+                    )
+                    if value
+                    else 0
+                ),
             )
             # Streaming turns get fold-only hooks, same rule as the
             # chat-completions path. A hook that defers work to `on_response`
@@ -2832,6 +2861,7 @@ class OpenAIHandlerMixin:
         request_id: str,
         timeout: float = COMPRESSION_TIMEOUT_SECONDS,
         client: str | None = None,
+        savings_tags: dict[str, Any] | None = None,
     ) -> tuple[dict[str, Any], bool, int, list[str], str | None, int, int, int, dict[str, float]]:
         timing: dict[str, float] = {}
 
@@ -2861,6 +2891,8 @@ class OpenAIHandlerMixin:
                 "timing": timing,
                 "client": client,
             }
+            if savings_tags is not None:
+                compression_kwargs["savings_tags"] = savings_tags
             while True:
                 try:
                     result = self._compress_openai_responses_payload(
@@ -2872,7 +2904,7 @@ class OpenAIHandlerMixin:
                     unsupported_kwarg = next(
                         (
                             name
-                            for name in ("client", "timing")
+                            for name in ("savings_tags", "client", "timing")
                             if f"unexpected keyword argument '{name}'" in str(exc)
                             and name in compression_kwargs
                         ),
@@ -3076,6 +3108,9 @@ class OpenAIHandlerMixin:
         # if httpx lacks brotli support the response body is undecipherable → 502.
         headers.pop("accept-encoding", None)
         tags = extract_tags(headers)
+        from headroom.proxy.savings_attribution import bind_scope
+
+        bind_scope(tags, request.scope)
         client = classify_client(headers)
         # Surface the image-compression decision (computed earlier) into
         # tags now that the tags dict exists. Same observability pattern
@@ -3943,6 +3978,7 @@ class OpenAIHandlerMixin:
             run_request_hooks,
         )
 
+        _th_ctx: TurnContext | None = None
         if registered_turn_hooks():
             _th_tools_before = body.get("tools")
             _th_tok_before = (
@@ -3956,6 +3992,11 @@ class OpenAIHandlerMixin:
                 messages=body["messages"],
                 tools=_th_tools_before,
                 config=self.config,
+                tags=tags,
+                count_messages=tokenizer.count_messages,
+                count_tools=lambda value: (
+                    tokenizer.count_text(json.dumps(value, default=str)) if value else 0
+                ),
             )
             # Snapshot messages BEFORE the hook (same tokenizer) so we can tell whether
             # the hook itself folded — comparing against optimized_tokens instead
@@ -4483,12 +4524,13 @@ class OpenAIHandlerMixin:
                         # replaces the response, this original is the one nobody
                         # else will read.
                         _hook_usage.record(_hook_resp_json, **CHAT_USAGE_KEYS)
-                        _hook_ctx = _TurnContext(
+                        _hook_ctx = _th_ctx or _TurnContext(
                             provider="openai",
                             model=str(model),
                             messages=body["messages"],
                             tools=body.get("tools"),
                             config=self.config,
+                            tags=tags,
                         )
 
                         async def _hook_call_model(_msgs):
@@ -5026,6 +5068,9 @@ class OpenAIHandlerMixin:
         # to decompress already-decoded JSON and reject it with HTTP 400 (#1542).
         headers.pop("content-encoding", None)
         tags = extract_tags(headers)
+        from headroom.proxy.savings_attribution import bind_scope
+
+        bind_scope(tags, request.scope)
         client = classify_client(headers)
 
         # Learn from the original client payload before memory context or
@@ -5388,6 +5433,7 @@ class OpenAIHandlerMixin:
                     model=model,
                     request_id=request_id,
                     client=client,
+                    savings_tags=tags,
                 )
                 attempted_input_tokens = int(_attempted_tokens)
                 if _transforms:
@@ -5659,6 +5705,7 @@ class OpenAIHandlerMixin:
                                 messages=body.get(_resp_key) or [],
                                 tools=body.get("tools"),
                                 config=self.config,
+                                tags=tags,
                             )
 
                             async def _resp_hook_call_model(

headroom/proxy/models.py

@@ -66,10 +66,14 @@ class RequestLog:
     total_latency_ms: float | None
 
     # Metadata
-    tags: dict[str, str]
+    tags: dict[str, Any]
     cache_hit: bool
     transforms_applied: list[str]
 
+    # Per-request attribution. Headline totals remain authoritative, so these
+    # explanatory rows are never added a second time.
+    savings_breakdown: list[dict[str, Any]] = field(default_factory=list)
+
     # Provider-side cache economics (Anthropic prompt caching, #2438).
     # ``cache_hit`` alone is ambiguous: a call billed cache-*creation* (write)
     # cannot be told apart from a real cache-*read* hit. These raw deltas —

headroom/proxy/outcome.py

@@ -170,7 +170,7 @@ class RequestOutcome:
     # (``original_messages``); otherwise ``request_messages`` carries the sent
     # body for backward compatibility and this stays ``None``.
     compressed_messages: list[dict[str, Any]] | None = None
-    tags: dict[str, str] = field(default_factory=dict)
+    tags: dict[str, Any] = field(default_factory=dict)
     client: str | None = None
     project: str | None = None
 
@@ -399,6 +399,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
     from headroom.proxy.cost import _summarize_transforms
     from headroom.proxy.models import RequestLog
     from headroom.proxy.project_context import get_current_project
+    from headroom.proxy.savings_attribution import encode, from_tags, public_tags
     from headroom.telemetry.session import record_outcome
 
     # GitHub Copilot: requests routed to the Copilot API travel on the OpenAI or
@@ -464,6 +465,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
     # tags and never move tok_before/after; aggregate them into Metrics so the
     # session summary / cost summary / all-layers total can surface the layer.
     tool_search_saved = tool_schema_saved_from_tags(outcome.tags or {})
+    savings_breakdown = from_tags(outcome.tags)
 
     # Billed input volume. Prefer the provider's own count where it reported one
     # — that is what the invoice charges for, and it is the number cache math is
@@ -499,6 +501,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
         client=outcome.client,
         tool_search_saved=tool_search_saved,
         local_input_tokens=outcome.optimized_tokens,
+        savings_attribution=savings_breakdown,
     )
 
     # 2. Cost tracker (optional).
@@ -524,7 +527,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
     #    dict is not mutated (frozen dataclass + defensive copy).
     request_logger = getattr(handler, "logger", None)
     if request_logger is not None:
-        log_tags = dict(outcome.tags)
+        log_tags = public_tags(outcome.tags)
         if outcome.client:
             log_tags["client"] = outcome.client
         if project:
@@ -551,6 +554,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
                 cache_write_tokens=outcome.cache_write_tokens,
                 uncached_input_tokens=outcome.uncached_input_tokens,
                 transforms_applied=list(outcome.transforms_applied),
+                savings_breakdown=savings_breakdown,
                 waste_signals=outcome.waste_signals,
                 request_messages=outcome.request_messages,
                 compressed_messages=outcome.compressed_messages,
@@ -570,6 +574,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
     # compaction is already inside tok_saved and must not be added twice.
     tool_saved = tool_schema_saved_from_tags(outcome.tags or {})
     total_saved = headline_tokens_saved(outcome.tokens_saved, outcome.tags or {})
+    encoded_savings = encode(savings_breakdown)
     logger.info(
         f"[{outcome.perf_request_id or outcome.request_id}] PERF "
         f"model={outcome.model} msgs={outcome.num_messages} "
@@ -584,6 +589,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
         f"total_ms={outcome.total_latency_ms:.0f} "
         f"tok_out={outcome.output_tokens} "
         f"ttfb_ms={outcome.ttfb_ms:.0f} "
+        f"savings={encoded_savings} "
         f"transforms={_summarize_transforms(list(outcome.transforms_applied))}"
         f"{client_part}"
     )

headroom/proxy/prometheus_metrics.py

@@ -14,7 +14,7 @@ import logging
 import threading
 from collections import defaultdict
 from datetime import datetime
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from headroom.observability import HeadroomOtelMetrics
@@ -22,7 +22,7 @@ if TYPE_CHECKING:
 
 from headroom import savings_ledger
 from headroom.observability import get_otel_metrics
-from headroom.proxy.savings_tracker import SavingsTracker
+from headroom.proxy.savings_tracker import SavingsTracker, estimate_request_savings_usd
 
 logger = logging.getLogger("headroom.proxy")
 
@@ -142,6 +142,8 @@ class PrometheusMetrics:
         # they saved so per-extension contribution is observable via /stats,
         # mirroring the per-strategy compression breakdown above.
         self.extension_savings: dict[str, int] = defaultdict(int)
+        # Named savings attribution; realized and projected rows stay separate.
+        self.savings_by_source: dict[str, dict[str, str | int | float | bool]] = {}
 
         # Fail-open compression failures, keyed by reason ("timeout",
         # "error"). The proxy fails open on any optimization error so the
@@ -350,6 +352,7 @@ class PrometheusMetrics:
             self.compressions_by_strategy.clear()
             self.tokens_saved_by_strategy.clear()
             self.extension_savings.clear()
+            self.savings_by_source.clear()
             with self._obs_counter_lock:
                 self.compression_failed_by_reason.clear()
                 self.kompress_size_gate_by_outcome.clear()
@@ -730,6 +733,7 @@ class PrometheusMetrics:
         client: str | None = None,
         tool_search_saved: int = 0,
         local_input_tokens: int | None = None,
+        savings_attribution: list[dict[str, Any]] | None = None,
     ):
         """Record metrics for a request.
 
@@ -757,6 +761,13 @@ class PrometheusMetrics:
                 model,
             )
             tokens_saved = 0
+        savings_usd = estimate_request_savings_usd(
+            model,
+            compression_tokens_saved=tokens_saved,
+            tool_schema_tokens_saved=tool_search_saved,
+            output_tokens_saved=output_tokens_saved,
+            cache_read_tokens=cache_read_tokens,
+        )
         async with self._lock:
             self.requests_total += 1
             self.requests_by_provider[provider] += 1
@@ -787,6 +798,26 @@ class PrometheusMetrics:
             self.tokens_output_total += output_tokens
             self.tokens_saved_total += tokens_saved
             self.tool_search_saved_total += max(0, int(tool_search_saved))
+            for item in savings_attribution or ():
+                source = str(item.get("source") or "other")[:64]
+                realized = bool(item.get("realized", True))
+                key = f"{source}:{int(realized)}"
+                row = self.savings_by_source.setdefault(
+                    key,
+                    {
+                        "source": source,
+                        "realized": realized,
+                        "events": 0,
+                        "tokens": 0,
+                        "usd": 0.0,
+                    },
+                )
+                row["events"] = int(row["events"]) + 1
+                row["tokens"] = int(row["tokens"]) + max(0, int(item.get("tokens", 0) or 0))
+                row["usd"] = round(
+                    float(row["usd"]) + float(item.get("usd", 0.0) or 0.0),
+                    12,
+                )
             # See the attribute definition for why this is the right
             # denominator for the active-compression ratio.
             self.attempted_input_tokens_total += max(0, int(attempted_input_tokens))
@@ -891,6 +922,7 @@ class PrometheusMetrics:
                 total_input_tokens=total_input_tokens,
                 total_input_cost_usd=total_input_cost_usd,
                 output_tokens_saved=output_tokens_saved,
+                estimated_savings_usd=savings_usd,
             )
 
         # Also append to the durable, multi-process savings ledger so
@@ -945,7 +977,8 @@ class PrometheusMetrics:
                 source="proxy",
             )
 
-        self._get_otel_metrics().record_proxy_request(
+        otel_metrics = self._get_otel_metrics()
+        otel_metrics.record_proxy_request(
             provider=provider,
             model=model,
             input_tokens=input_tokens,
@@ -961,7 +994,15 @@ class PrometheusMetrics:
             cache_write_5m_tokens=cache_write_5m_tokens,
             cache_write_1h_tokens=cache_write_1h_tokens,
             uncached_input_tokens=uncached_input_tokens,
+            attempted_input_tokens=attempted_input_tokens,
+            output_tokens_saved=output_tokens_saved,
+            savings_usd=savings_usd,
+            project=project,
+            client=client,
         )
+        record_attribution = getattr(otel_metrics, "record_savings_attribution", None)
+        if record_attribution is not None and savings_attribution:
+            record_attribution(savings_attribution)
 
     async def record_stage_timings(
         self,
@@ -1157,6 +1198,47 @@ class PrometheusMetrics:
                 help_text="Tokens saved by optimization",
                 value=self.tokens_saved_total,
             )
+            if self.savings_by_source:
+                lines.extend(
+                    [
+                        "# HELP headroom_savings_attribution_events_total Per-request savings attribution events",
+                        "# TYPE headroom_savings_attribution_events_total counter",
+                    ]
+                )
+                for row in self.savings_by_source.values():
+                    labels = _format_labels(
+                        {"source": str(row["source"]), "realized": str(row["realized"]).lower()}
+                    )
+                    lines.append(
+                        f"headroom_savings_attribution_events_total{labels} {row['events']}"
+                    )
+                lines.extend(
+                    [
+                        "",
+                        "# HELP headroom_savings_attributed_tokens_total Tokens attributed to a savings source",
+                        "# TYPE headroom_savings_attributed_tokens_total counter",
+                    ]
+                )
+                for row in self.savings_by_source.values():
+                    labels = _format_labels(
+                        {"source": str(row["source"]), "realized": str(row["realized"]).lower()}
+                    )
+                    lines.append(
+                        f"headroom_savings_attributed_tokens_total{labels} {row['tokens']}"
+                    )
+                lines.extend(
+                    [
+                        "",
+                        "# HELP headroom_savings_attributed_usd_total Cost savings attributed to a source; may be negative",
+                        "# TYPE headroom_savings_attributed_usd_total gauge",
+                    ]
+                )
+                for row in self.savings_by_source.values():
+                    labels = _format_labels(
+                        {"source": str(row["source"]), "realized": str(row["realized"]).lower()}
+                    )
+                    lines.append(f"headroom_savings_attributed_usd_total{labels} {row['usd']}")
+                lines.append("")
             _append_metric(
                 lines,
                 name="headroom_persistent_savings_requests_total",

headroom/proxy/savings_attribution.py

@@ -0,0 +1,110 @@
+"""Bounded attribution for savings that do not have a built-in metric."""
+
+from __future__ import annotations
+
+import base64
+import json
+import re
+from collections.abc import MutableMapping
+from typing import Any
+
+SAVINGS_ATTRIBUTION_TAG = "_headroom_savings_attribution"
+_NAME_RE = re.compile(r"[^a-z0-9_.-]+")
+MAX_SOURCES = 32
+_SCOPE_KEY = "headroom_savings_attribution"
+
+
+def _source_name(value: object) -> str:
+    name = _NAME_RE.sub("_", str(value or "other").strip().lower()).strip("_.-")
+    return (name or "other")[:64]
+
+
+def _ledger(tags: MutableMapping[str, Any]) -> list[dict[str, Any]]:
+    current = tags.get(SAVINGS_ATTRIBUTION_TAG)
+    if isinstance(current, list):
+        return current
+    current = []
+    tags[SAVINGS_ATTRIBUTION_TAG] = current
+    return current
+
+
+def bind_scope(tags: MutableMapping[str, Any], scope: MutableMapping[str, Any]) -> None:
+    """Share one ledger between ASGI middleware and the request handler."""
+    state = scope.setdefault("state", {})
+    ledger = state.get(_SCOPE_KEY)
+    if not isinstance(ledger, list):
+        ledger = []
+        state[_SCOPE_KEY] = ledger
+    tags[SAVINGS_ATTRIBUTION_TAG] = ledger
+
+
+def record_scope_savings(scope: MutableMapping[str, Any], source: object, **values: Any) -> None:
+    state = scope.setdefault("state", {})
+    ledger = state.get(_SCOPE_KEY)
+    if not isinstance(ledger, list):
+        ledger = []
+        state[_SCOPE_KEY] = ledger
+    record_savings({SAVINGS_ATTRIBUTION_TAG: ledger}, source, **values)
+
+
+def record_savings(
+    tags: MutableMapping[str, Any],
+    source: object,
+    *,
+    tokens: int = 0,
+    usd: float = 0.0,
+    realized: bool = True,
+    estimated: bool = False,
+    details: dict[str, Any] | None = None,
+) -> None:
+    """Attribute savings to a source; this never changes headline totals."""
+    ledger = _ledger(tags)
+    if len(ledger) >= MAX_SOURCES:
+        return
+    item: dict[str, Any] = {
+        "source": _source_name(source),
+        "realized": bool(realized),
+        "estimated": bool(estimated),
+        "tokens": max(0, int(tokens or 0)),
+        "usd": round(float(usd or 0.0), 12),
+    }
+    if details:
+        item["details"] = {
+            _source_name(key): value
+            for key, value in list(details.items())[:12]
+            if isinstance(value, (str, int, float, bool)) or value is None
+        }
+    ledger.append(item)
+
+
+def from_tags(tags: MutableMapping[str, Any] | None) -> list[dict[str, Any]]:
+    raw = (tags or {}).get(SAVINGS_ATTRIBUTION_TAG)
+    if not isinstance(raw, list):
+        return []
+    return [dict(item) for item in raw[:MAX_SOURCES] if isinstance(item, dict)]
+
+
+def public_tags(tags: MutableMapping[str, Any] | None) -> dict[str, Any]:
+    return {key: value for key, value in (tags or {}).items() if key != SAVINGS_ATTRIBUTION_TAG}
+
+
+def encode(items: list[dict[str, Any]]) -> str:
+    if not items:
+        return "none"
+    payload = json.dumps(items, separators=(",", ":"), sort_keys=True).encode()
+    return base64.urlsafe_b64encode(payload).decode().rstrip("=")
+
+
+def decode(value: str) -> list[dict[str, Any]]:
+    if not value or value == "none":
+        return []
+    try:
+        padded = value + "=" * (-len(value) % 4)
+        decoded = json.loads(base64.urlsafe_b64decode(padded).decode())
+    except Exception:
+        return []
+    return (
+        [dict(item) for item in decoded if isinstance(item, dict)]
+        if isinstance(decoded, list)
+        else []
+    )

headroom/proxy/savings_tracker.py

@@ -19,6 +19,7 @@ from datetime import datetime, timedelta, timezone
 from functools import lru_cache
 from io import StringIO
 from pathlib import Path
+from collections.abc import Mapping
 from typing import Any
 
 from headroom import paths as _paths
@@ -328,6 +329,37 @@ def _estimate_cache_savings_usd(model: str, cache_read_tokens: int) -> float:
         return 0.0
 
 
+def estimate_request_savings_usd(
+    model: str,
+    *,
+    compression_tokens_saved: int = 0,
+    tool_schema_tokens_saved: int = 0,
+    output_tokens_saved: int = 0,
+    cache_read_tokens: int = 0,
+) -> dict[str, float]:
+    """Price one request's distinct savings layers for external telemetry.
+
+    The values use the same pricing functions as the built-in dashboard. They
+    stay separate because provider-cache benefit is not caused by compression,
+    and extension attribution can be explanatory rather than additive.
+    """
+
+    return {
+        "compression": _estimate_compression_savings_usd(
+            model, max(_coerce_int(compression_tokens_saved), 0)
+        ),
+        "tool_schema": _estimate_compression_savings_usd(
+            model, max(_coerce_int(tool_schema_tokens_saved), 0)
+        ),
+        "output_shaping": _estimate_output_savings_usd(
+            model, max(_coerce_int(output_tokens_saved), 0)
+        ),
+        "provider_cache": _estimate_cache_savings_usd(
+            model, max(_coerce_int(cache_read_tokens), 0)
+        ),
+    }
+
+
 def _estimate_input_cost_usd(
     model: str,
     input_tokens: int,
@@ -717,6 +749,7 @@ class SavingsTracker:
         uncached_input_tokens: int = 0,
         total_input_tokens: int | None = None,
         total_input_cost_usd: float | None = None,
+        estimated_savings_usd: Mapping[str, float] | None = None,
         timestamp: datetime | str | None = None,
     ) -> bool:
         """Persist a canonical display-session update for every request."""
@@ -732,11 +765,24 @@ class SavingsTracker:
 
         delta_tokens_saved = _coerce_int(tokens_saved)
         delta_input_tokens = _coerce_int(input_tokens)
-        delta_savings_usd = _estimate_compression_savings_usd(model, delta_tokens_saved)
         delta_output_tokens_saved = max(_coerce_int(output_tokens_saved), 0)
-        delta_output_savings_usd = _estimate_output_savings_usd(model, delta_output_tokens_saved)
         delta_cache_read_tokens = _coerce_int(cache_read_tokens)
-        delta_cache_savings_usd = _estimate_cache_savings_usd(model, delta_cache_read_tokens)
+        priced = estimated_savings_usd
+        delta_savings_usd = (
+            max(_coerce_float(priced.get("compression")), 0.0)
+            if priced is not None
+            else _estimate_compression_savings_usd(model, delta_tokens_saved)
+        )
+        delta_output_savings_usd = (
+            max(_coerce_float(priced.get("output_shaping")), 0.0)
+            if priced is not None
+            else _estimate_output_savings_usd(model, delta_output_tokens_saved)
+        )
+        delta_cache_savings_usd = (
+            max(_coerce_float(priced.get("provider_cache")), 0.0)
+            if priced is not None
+            else _estimate_cache_savings_usd(model, delta_cache_read_tokens)
+        )
         delta_input_cost_usd = _estimate_input_cost_usd(
             model,
             delta_input_tokens,

headroom/proxy/server.py

@@ -3806,6 +3806,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
                     "has_exact_tokens": token_accounting_status == "complete",
                     "token_accounting_status": token_accounting_status,
                     "transforms_applied": log.get("transforms_applied", []),
+                    "savings_breakdown": log.get("savings_breakdown", []),
                     "waste_signals": log.get("waste_signals"),
                     "tool_schema_saved_tokens": _tool_schema_saved_from_tags(log.get("tags")),
                 }
@@ -4030,6 +4031,12 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
             "savings": {
                 "total_tokens": total_tokens_all_layers,
                 "per_project": persistent_savings.get("projects", {}),
+                # Attribution only: these rows explain the canonical headline
+                # and are not added to it again.
+                "by_source": sorted(
+                    (dict(row) for row in m.savings_by_source.values()),
+                    key=lambda row: (-int(row.get("tokens", 0)), str(row["source"])),
+                ),
                 "by_layer": {
                     "compression": {
                         "tokens": proxy_compression_tokens,

headroom/proxy/turn_hooks.py

@@ -23,7 +23,7 @@ from __future__ import annotations
 
 import logging
 from collections.abc import Awaitable, Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any, Protocol, runtime_checkable
 
 log = logging.getLogger(__name__)
@@ -47,6 +47,35 @@ class TurnContext:
     messages: list[dict[str, Any]]
     tools: Any = None  # provider-native tools value (list, or None)
     config: Any = None
+    # The handler's live request tags and provider-native counters let the hook
+    # runner attribute savings with the exact same tokenizer as the canonical
+    # RequestOutcome. Optional defaults preserve the public extension API.
+    tags: dict[str, Any] = field(default_factory=dict)
+    count_messages: Callable[[list[dict[str, Any]]], int] | None = None
+    count_tools: Callable[[Any], int] | None = None
+
+    def record_savings(
+        self,
+        source: str,
+        *,
+        tokens: int = 0,
+        usd: float = 0.0,
+        realized: bool = True,
+        estimated: bool = False,
+        details: dict[str, Any] | None = None,
+    ) -> None:
+        """Attribute savings without changing the request's headline totals."""
+        from headroom.proxy.savings_attribution import record_savings
+
+        record_savings(
+            self.tags,
+            source,
+            tokens=tokens,
+            usd=usd,
+            realized=realized,
+            estimated=estimated,
+            details=details,
+        )
 
 
 @runtime_checkable
@@ -111,8 +140,32 @@ def run_request_hooks(ctx: TurnContext, *, stream_safe_only: bool = False) -> No
         fn = getattr(hook, "on_request", None)
         if fn is None:
             continue
+        before_messages = before_tools = None
         try:
+            if ctx.count_messages is not None:
+                before_messages = ctx.count_messages(ctx.messages)
+            if ctx.count_tools is not None:
+                before_tools = ctx.count_tools(ctx.tools)
             fn(ctx)
+            message_saved = (
+                max(0, before_messages - ctx.count_messages(ctx.messages))
+                if before_messages is not None and ctx.count_messages is not None
+                else 0
+            )
+            tool_saved = (
+                max(0, before_tools - ctx.count_tools(ctx.tools))
+                if before_tools is not None and ctx.count_tools is not None
+                else 0
+            )
+            if message_saved or tool_saved:
+                ctx.record_savings(
+                    getattr(hook, "savings_source", getattr(hook, "name", type(hook).__name__)),
+                    tokens=message_saved + tool_saved,
+                    details={
+                        "message_tokens_saved": message_saved,
+                        "tool_tokens_saved": tool_saved,
+                    },
+                )
         except Exception:  # a hook must never break the proxy
             log.exception("turn hook %r on_request failed", getattr(hook, "name", hook))
 

tests/test_observability_metrics.py

@@ -13,8 +13,10 @@ from opentelemetry.sdk.metrics.export import InMemoryMetricReader
 from headroom.observability import (
     HeadroomOtelMetrics,
     get_otel_meter,
+    register_otel_metric_attribute_provider,
     reset_otel_metrics,
     set_otel_metrics,
+    unregister_otel_metric_attribute_provider,
 )
 from headroom.proxy.prometheus_metrics import PrometheusMetrics
 from headroom.telemetry.context import MAX_DISTINCT_MODELS
@@ -61,6 +63,16 @@ def test_headroom_otel_metrics_records_proxy_and_pipeline_metrics() -> None:
         cache_write_5m_tokens=10,
         cache_write_1h_tokens=25,
         uncached_input_tokens=60,
+        attempted_input_tokens=165,
+        output_tokens_saved=8,
+        savings_usd={
+            "compression": 0.001,
+            "tool_schema": 0.0003,
+            "output_shaping": 0.0008,
+            "provider_cache": 0.0002,
+        },
+        project="checkout",
+        client="claude-code",
     )
     otel_metrics.record_proxy_cache_bust(tokens_lost=7)
     otel_metrics.record_pipeline_run(
@@ -103,6 +115,30 @@ def test_headroom_otel_metrics_records_proxy_and_pipeline_metrics() -> None:
     )
     assert tool_schema_point.value == 15
 
+    attempted_input = metrics["headroom.proxy.tokens.attempted_input"]
+    attempted_point = _find_point(
+        attempted_input,
+        **{
+            "headroom.project": "checkout",
+            "headroom.client": "claude-code",
+        },
+    )
+    assert attempted_point.value == 165
+
+    output_saved = metrics["headroom.proxy.tokens.output_saved"]
+    output_saved_point = _find_point(
+        output_saved,
+        **{
+            "headroom.project": "checkout",
+            "headroom.client": "claude-code",
+        },
+    )
+    assert output_saved_point.value == 8
+
+    savings_usd = metrics["headroom.proxy.savings.usd"]
+    compression_usd = _find_point(savings_usd, source="compression", estimated=True)
+    assert compression_usd.value == pytest.approx(0.001)
+
     compression_saved = metrics["headroom.compression.tokens.saved"]
     compression_saved_point = _find_point(
         compression_saved,
@@ -176,6 +212,77 @@ def test_get_otel_meter_uses_headrooms_configured_provider() -> None:
         reset_otel_metrics()
 
 
+def test_request_attribute_provider_enriches_core_and_savings_metrics() -> None:
+    reader = InMemoryMetricReader()
+    provider = MeterProvider(metric_readers=[reader])
+    otel_metrics = HeadroomOtelMetrics(meter_provider=provider)
+
+    def identity_attributes() -> dict[str, str]:
+        return {
+            "headroom.org": "acme",
+            "headroom.team": "payments",
+            "headroom.user": "alice",
+            # Canonical call-site dimensions must win over an extension.
+            "model": "must-not-override",
+            "source": "must-not-override",
+        }
+
+    register_otel_metric_attribute_provider(identity_attributes)
+    try:
+        otel_metrics.record_proxy_request(
+            provider="anthropic",
+            model="claude-sonnet-4-5",
+            input_tokens=100,
+            output_tokens=10,
+            tokens_saved=25,
+            latency_ms=20,
+        )
+        otel_metrics.record_savings_attribution(
+            [{"source": "tool_search", "tokens": 20, "usd": 0.001}]
+        )
+
+        metrics = _collect_metrics(reader)
+        request = _find_point(
+            metrics["headroom.proxy.requests"],
+            model="claude-sonnet-4-5",
+            **{
+                "headroom.org": "acme",
+                "headroom.team": "payments",
+                "headroom.user": "alice",
+            },
+        )
+        assert request.value == 1
+        attributed = _find_point(
+            metrics["headroom.savings.attributed.tokens"],
+            source="tool_search",
+            **{"headroom.user": "alice"},
+        )
+        assert attributed.value == 20
+    finally:
+        unregister_otel_metric_attribute_provider(identity_attributes)
+
+
+def test_failing_request_attribute_provider_is_fail_open() -> None:
+    reader = InMemoryMetricReader()
+    provider = MeterProvider(metric_readers=[reader])
+    otel_metrics = HeadroomOtelMetrics(meter_provider=provider)
+
+    def broken_provider() -> dict[str, str]:
+        raise RuntimeError("identity unavailable")
+
+    register_otel_metric_attribute_provider(broken_provider)
+    try:
+        otel_metrics.record_proxy_failed(provider="openai", model="gpt-5")
+        point = _find_point(
+            _collect_metrics(reader)["headroom.proxy.requests.failed"],
+            provider="openai",
+            model="gpt-5",
+        )
+        assert point.value == 1
+    finally:
+        unregister_otel_metric_attribute_provider(broken_provider)
+
+
 @dataclass
 class _SpyMetrics:
     pipeline_calls: list[dict[str, Any]] = field(default_factory=list)
@@ -186,9 +293,13 @@ class _SpyMetrics:
 
 @dataclass
 class _SpyProxyMetrics:
+    request_calls: list[dict[str, Any]] = field(default_factory=list)
     failed_calls: list[dict[str, Any]] = field(default_factory=list)
     rate_limited_calls: list[dict[str, Any]] = field(default_factory=list)
 
+    def record_proxy_request(self, **kwargs: Any) -> None:
+        self.request_calls.append(kwargs)
+
     def record_proxy_failed(self, **kwargs: Any) -> None:
         self.failed_calls.append(kwargs)
 
@@ -250,6 +361,49 @@ async def test_prometheus_metrics_reads_late_configured_otel_metrics() -> None:
         reset_otel_metrics()
 
 
+@pytest.mark.asyncio
+async def test_prometheus_metrics_forwards_savings_drilldown_fields_to_otel(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    expected_usd = {
+        "compression": 0.003,
+        "tool_schema": 0.0,
+        "output_shaping": 0.004,
+        "provider_cache": 0.0,
+    }
+    monkeypatch.setattr(
+        "headroom.proxy.prometheus_metrics.estimate_request_savings_usd",
+        lambda *_args, **_kwargs: expected_usd,
+    )
+    spy = _SpyProxyMetrics()
+    metrics = PrometheusMetrics(stateless=True)
+    set_otel_metrics(spy)  # type: ignore[arg-type]
+
+    try:
+        await metrics.record_request(
+            provider="anthropic",
+            model="claude-sonnet-4-5",
+            input_tokens=90,
+            output_tokens=12,
+            tokens_saved=30,
+            latency_ms=5.0,
+            attempted_input_tokens=120,
+            output_tokens_saved=4,
+            project="checkout",
+            client="claude-code",
+        )
+
+        assert len(spy.request_calls) == 1
+        call = spy.request_calls[0]
+        assert call["attempted_input_tokens"] == 120
+        assert call["output_tokens_saved"] == 4
+        assert call["savings_usd"] == expected_usd
+        assert call["project"] == "checkout"
+        assert call["client"] == "claude-code"
+    finally:
+        reset_otel_metrics()
+
+
 @pytest.mark.asyncio
 async def test_prometheus_metrics_clamps_negative_token_savings() -> None:
     metrics = PrometheusMetrics()

tests/test_proxy_openai.py

@@ -387,7 +387,7 @@ def test_native_responses_route_carries_the_client_decision(
         assert transport.call_count == 1, response.text
         assert response.status_code == 200, response.text
         assert seen, "the Responses compressor was never reached"
-        assert {k: v for k, v in seen[0].items() if k != "timing"} == expected
+        assert {k: v for k, v in seen[0].items() if k not in {"timing", "savings_tags"}} == expected
 
 
 # --- production route: the Codex WebSocket handler ---------------------------

tests/test_proxy_stats_recent_requests.py

@@ -84,6 +84,14 @@ def test_stats_refreshes_recent_requests_when_cached() -> None:
             "input_tokens_optimized": 60,
             "tokens_saved": 40,
             "savings_percent": 40.0,
+            "savings_breakdown": [
+                {
+                    "source": "tool_search",
+                    "tokens": 40,
+                    "usd": 0.0,
+                    "realized": True,
+                }
+            ],
         }
     )
     second_log = FakeLogEntry(
@@ -104,6 +112,14 @@ def test_stats_refreshes_recent_requests_when_cached() -> None:
         first_response = client.get("/stats?cached=1")
         assert first_response.status_code == 200
         assert first_response.json()["recent_requests"][-1]["model"] == "gpt-4.1"
+        assert first_response.json()["recent_requests"][-1]["savings_breakdown"] == [
+            {
+                "source": "tool_search",
+                "tokens": 40,
+                "usd": 0.0,
+                "realized": True,
+            }
+        ]
 
         logger.logs = [first_log, second_log]
         second_response = client.get("/stats?cached=1")

tests/test_turn_hook_usage.py

@@ -23,6 +23,7 @@ import httpx
 import pytest
 import respx
 
+from headroom.proxy.handlers.anthropic import _AnthropicTurnHookUsage
 from headroom.proxy.handlers.openai import (
     CHAT_USAGE_KEYS,
     RESPONSES_USAGE_KEYS,
@@ -149,6 +150,39 @@ def test_never_raises_on_a_shape_it_does_not_recognise() -> None:
     assert (u.input_tokens, u.output_tokens, u.cache_read_tokens) == (0, 0, 0)
 
 
+def test_anthropic_accumulator_includes_disjoint_cache_buckets() -> None:
+    usage = _AnthropicTurnHookUsage()
+    first = {
+        "usage": {
+            "input_tokens": 100,
+            "output_tokens": 10,
+            "cache_read_input_tokens": 50,
+            "cache_creation_input_tokens": 25,
+            "cache_creation": {
+                "ephemeral_5m_input_tokens": 20,
+                "ephemeral_1h_input_tokens": 5,
+            },
+        }
+    }
+    final = {
+        "usage": {
+            "input_tokens": 150,
+            "output_tokens": 20,
+            "cache_read_input_tokens": 70,
+            "cache_creation_input_tokens": 30,
+        }
+    }
+    usage.record(first)
+    usage.record(final)
+    usage.settle(final)
+    assert usage.input_tokens == 175
+    assert usage.output_tokens == 10
+    assert usage.cache_read_tokens == 50
+    assert usage.cache_write_tokens == 25
+    assert usage.cache_write_5m_tokens == 20
+    assert usage.cache_write_1h_tokens == 5
+
+
 # --- handler level: what the unit tests above structurally cannot see -----
 
 
@@ -314,3 +348,58 @@ def test_no_hook_registered_bills_exactly_the_one_call(monkeypatch, _no_hooks) -
     o = outcomes[-1]
     assert o.provider_input_tokens == 100
     assert o.output_tokens == 10
+
+
+@respx.mock
+def test_anthropic_bills_original_plus_hook_redrive(monkeypatch, _no_hooks) -> None:
+    """Anthropic A=175 total input, B=250 -> 425; outputs 10+20."""
+    register_turn_hook(_RedriveOnce())
+    app, outcomes = _app_and_outcomes(monkeypatch)
+
+    def response(ident: str, input_tokens: int, output_tokens: int, read: int, write: int):
+        return {
+            "id": ident,
+            "type": "message",
+            "role": "assistant",
+            "model": "claude-sonnet-4-5",
+            "content": [{"type": "text", "text": ident}],
+            "stop_reason": "end_turn",
+            "usage": {
+                "input_tokens": input_tokens,
+                "output_tokens": output_tokens,
+                "cache_read_input_tokens": read,
+                "cache_creation_input_tokens": write,
+            },
+        }
+
+    sent = iter(
+        [
+            response("a", 100, 10, 50, 25),
+            response("b", 150, 20, 70, 30),
+        ]
+    )
+    respx.post("https://api.anthropic.com/v1/messages").mock(
+        side_effect=lambda request: httpx.Response(200, json=next(sent))
+    )
+
+    with TestClient(app) as client:
+        result = client.post(
+            "/v1/messages",
+            json={
+                "model": "claude-sonnet-4-5",
+                "max_tokens": 64,
+                "messages": [{"role": "user", "content": "hi"}],
+            },
+            headers={
+                "x-api-key": "sk-ant-test",
+                "anthropic-version": "2023-06-01",
+            },
+        )
+
+    assert result.status_code == 200
+    outcome = outcomes[-1]
+    assert outcome.provider_input_tokens == 425
+    assert outcome.output_tokens == 30
+    assert outcome.cache_read_tokens == 120
+    assert outcome.cache_write_tokens == 55
+    assert outcome.uncached_input_tokens == 250

tests/test_turn_hooks.py

@@ -102,6 +102,41 @@ def test_on_request_may_mutate_ctx():
     assert ctx.tools == [{"name": "keep"}]
 
 
+def test_request_runner_attributes_savings_with_handler_counters():
+    class Shrink:
+        name = "internal-hook-name"
+        savings_source = "tool_search"
+
+        def on_request(self, ctx: TurnContext) -> None:
+            ctx.tools = (ctx.tools or [])[:1]
+            ctx.messages[0]["content"] = "short"
+
+    register_turn_hook(Shrink())
+    tags = {}
+    ctx = _ctx(
+        messages=[{"role": "user", "content": "a much longer value"}],
+        tools=[{"name": "keep"}, {"name": "drop"}],
+        tags=tags,
+        count_messages=lambda messages: len(messages[0]["content"]),
+        count_tools=lambda tools: len(tools or []),
+    )
+
+    run_request_hooks(ctx)
+
+    from headroom.proxy.savings_attribution import from_tags
+
+    assert from_tags(tags) == [
+        {
+            "source": "tool_search",
+            "realized": True,
+            "estimated": False,
+            "tokens": 15,
+            "usd": 0.0,
+            "details": {"message_tokens_saved": 14, "tool_tokens_saved": 1},
+        }
+    ]
+
+
 # --- on_response replacement + re-drive loop ---------------------------------