/headroom/proxy/prometheus_metrics.py

"""Prometheus-compatible metrics for the Headroom proxy.

Tracks request counts, token usage, latency, overhead, TTFB,
per-transform timing, waste signals, prefix cache stats, and
cumulative savings history.

Extracted from server.py for maintainability.
"""

from __future__ import annotations

import asyncio
import logging
import threading
from collections import defaultdict
from datetime import datetime
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from headroom.observability import HeadroomOtelMetrics
    from headroom.proxy.cost import CostTracker

from headroom import savings_ledger
from headroom.observability import get_otel_metrics
from headroom.proxy.savings_tracker import SavingsTracker

logger = logging.getLogger("headroom.proxy")


def _escape_label_value(value: str) -> str:
    return value.replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"')


def _format_labels(labels: dict[str, str] | None = None) -> str:
    if not labels:
        return ""

    rendered = ",".join(
        f'{key}="{_escape_label_value(str(value))}"' for key, value in sorted(labels.items())
    )
    return f"{{{rendered}}}"


def _append_metric(
    lines: list[str],
    *,
    name: str,
    metric_type: str,
    help_text: str,
    value: int | float,
    labels: dict[str, str] | None = None,
) -> None:
    lines.extend(
        [
            f"# HELP {name} {help_text}",
            f"# TYPE {name} {metric_type}",
            f"{name}{_format_labels(labels)} {value}",
            "",
        ]
    )


# The proxy persists savings state on every request. Batch that write so a busy
# event loop isn't blocked re-serializing + fsyncing the whole history each time;
# the tracker still flushes on graceful shutdown (see HeadroomProxy.shutdown).
PROXY_SAVINGS_FLUSH_EVERY = 25


class PrometheusMetrics:
    """Prometheus-compatible metrics."""

    def __init__(
        self,
        savings_tracker: SavingsTracker | None = None,
        cost_tracker: CostTracker | None = None,
        otel_metrics: HeadroomOtelMetrics | None = None,
        stateless: bool = False,
    ):
        # Stateless mode: keep live in-memory metrics but never write the
        # durable savings files (proxy_savings.json, savings_events.jsonl).
        self._stateless = stateless
        self.requests_total = 0
        self.requests_by_provider: dict[str, int] = defaultdict(int)
        self.requests_by_model: dict[str, int] = defaultdict(int)
        # Populated via X-Headroom-Stack header (TS SDK adapters, etc.)
        self.requests_by_stack: dict[str, int] = defaultdict(int)
        self.requests_cached = 0
        self.requests_rate_limited = 0
        self.requests_failed = 0
        self.inbound_requests_total = 0
        self.inbound_requests_completed = 0
        self.inbound_requests_active = 0
        self.inbound_requests_by_method: dict[str, int] = defaultdict(int)
        self.inbound_requests_by_path: dict[str, int] = defaultdict(int)
        self.inbound_responses_by_status: dict[str, int] = defaultdict(int)

        self.tokens_input_total = 0
        self.tokens_output_total = 0
        self.tokens_saved_total = 0
        # Sum of tokens we actually attempted to compress across the
        # session: extracted units that passed all gates + tool-schema
        # tokens we ran compaction against. Excludes prefix-frozen
        # content (instructions, user/system messages, prior turns).
        # This is the right denominator for an "active compression
        # ratio" — what fraction of the compressible-eligible tokens
        # did we actually save?
        self.attempted_input_tokens_total = 0

        # Per-strategy compression counters. Populated lazily as we see
        # each strategy tag — no hardcoded list of strategies; the keys
        # come from ContentRouter's `CompressionStrategy.value` and
        # SmartCrusher's literal `"smart_crusher"`. The forcing
        # function for catching strategy-level silent regressions:
        # if SmartCrusher events drop to zero in production, the
        # `headroom_compressions_total{strategy="smart_crusher"}`
        # counter shows it on day 1, not week 3.
        self.compressions_by_strategy: dict[str, int] = defaultdict(int)
        self.tokens_saved_by_strategy: dict[str, int] = defaultdict(int)

        # Codex WebSocket compression observability. These are intentionally
        # aggregate counters/sums, not per-unit storage, so /stats can answer
        # routing questions without growing with traffic volume.
        self.codex_ws_units_total = 0
        self.codex_ws_units_modified_total = 0
        self.codex_ws_units_to_kompress_total = 0
        self.codex_ws_units_kompress_attempted_total = 0
        self.codex_ws_units_by_strategy: dict[str, int] = defaultdict(int)
        self.codex_ws_units_by_category: dict[str, int] = defaultdict(int)
        self.codex_ws_units_by_content_type: dict[str, int] = defaultdict(int)
        self.codex_ws_units_by_text_shape: dict[str, int] = defaultdict(int)
        self.codex_ws_unit_elapsed_ms_sum = 0.0
        self.codex_ws_unit_elapsed_ms_max = 0.0
        self.codex_ws_unit_bytes_sum = 0
        self.codex_ws_unit_tokens_before_sum = 0
        self.codex_ws_unit_tokens_after_sum = 0
        self.codex_ws_unit_tokens_saved_sum = 0

        self.codex_ws_frames_attempted_total = 0
        self.codex_ws_frames_compressed_total = 0
        self.codex_ws_frames_failed_total = 0
        self.codex_ws_frames_to_kompress_total = 0
        self.codex_ws_frames_kompress_attempted_total = 0
        self.codex_ws_frame_elapsed_ms_sum = 0.0
        self.codex_ws_frame_elapsed_ms_max = 0.0
        self.codex_ws_frame_bytes_before_sum = 0
        self.codex_ws_frame_bytes_after_sum = 0
        self.codex_ws_frame_attempted_tokens_sum = 0
        self.codex_ws_frame_tokens_saved_sum = 0

        self.latency_sum_ms = 0.0
        self.latency_min_ms = float("inf")
        self.latency_max_ms = 0.0
        self.latency_count = 0

        # Headroom overhead (optimization time only, excludes LLM)
        self.overhead_sum_ms = 0.0
        self.overhead_min_ms = float("inf")
        self.overhead_max_ms = 0.0
        self.overhead_count = 0

        # Time to first byte (TTFB) from upstream — what the user actually feels
        self.ttfb_sum_ms = 0.0
        self.ttfb_min_ms = float("inf")
        self.ttfb_max_ms = 0.0
        self.ttfb_count = 0

        # Per-transform timing (name → cumulative ms, count)
        self.transform_timing_sum: dict[str, float] = defaultdict(float)
        self.transform_timing_count: dict[str, int] = defaultdict(int)
        self.transform_timing_max: dict[str, float] = defaultdict(float)

        # Per-stage timing (Unit 2). Keyed by ``(path, stage)`` tuples so
        # a single metric name can distinguish between, e.g.,
        # ``openai_responses_ws`` ``upstream_connect`` and
        # ``anthropic_messages`` ``upstream_connect``.
        self.stage_timing_sum: dict[tuple[str, str], float] = defaultdict(float)
        self.stage_timing_count: dict[tuple[str, str], int] = defaultdict(int)
        self.stage_timing_max: dict[tuple[str, str], float] = defaultdict(float)

        # WS session lifecycle (Unit 3). Gauges are live counters updated
        # by the Codex handler on register/deregister + attach_tasks/
        # detach. Histograms record completed-session durations bucketed
        # by termination cause so we can distinguish slow happy-path
        # sessions from long client-hold followed by client_disconnect.
        self.active_ws_sessions: int = 0
        self.active_relay_tasks: int = 0
        self.ws_session_duration_sum_ms: dict[str, float] = defaultdict(float)
        self.ws_session_duration_count: dict[str, int] = defaultdict(int)
        self.ws_session_duration_max_ms: dict[str, float] = defaultdict(float)

        # Aggregate waste signals
        self.waste_signals_total: dict[str, int] = defaultdict(int)

        # Cumulative ContentRouter protection counts. Each routing pass
        # categorises every message — `user_msg`, `system_msg`,
        # `recent_code`, `excluded_tool`, `analysis_ctx`, `small`,
        # `ratio_too_high`, `already_compressed`, `non_string`,
        # `content_blocks`. Surfacing these in `/stats` gives operators a
        # way to diagnose "why is my compression rate low?" — e.g. a high
        # `user_msg` count on OpenAI/Azure traffic explains why most
        # input was protected and never reached the compressor (#454).
        self.router_route_counts: dict[str, int] = defaultdict(int)

        # Provider-specific prefix cache tracking
        # Each provider has different cache economics:
        #   Anthropic: cache_read=0.1x, cache_write=1.25x, explicit breakpoints
        #   OpenAI: cache_read=0.5x, no write penalty, automatic
        #   Google: cache_read=~0.1x, explicit cachedContent API, storage cost
        #   Bedrock: no cache metrics
        self.cache_by_provider: dict[str, dict[str, int | float]] = defaultdict(
            lambda: {
                "cache_read_tokens": 0,
                "cache_write_tokens": 0,
                "cache_write_5m_tokens": 0,
                "cache_write_1h_tokens": 0,
                "cache_write_5m_requests": 0,
                "cache_write_1h_requests": 0,
                "uncached_input_tokens": 0,
                "requests": 0,
                "hit_requests": 0,  # requests with cache_read > 0
                "bust_count": 0,
                "bust_write_tokens": 0,
            }
        )
        # Track per-model cache request count to distinguish cold starts from busts
        self._cache_requests_by_model: dict[str, int] = defaultdict(int)

        # Prefix freeze stats (cache-aware compression)
        self.prefix_freeze_busts_avoided: int = 0
        self.prefix_freeze_tokens_preserved: int = 0
        self.prefix_freeze_compression_foregone: int = 0

        # Cache bust tracking: how many tokens lost their cache discount due to compression
        self.cache_bust_tokens_lost: int = 0
        self.cache_bust_count: int = 0

        # Cache-miss attribution (#1313): when a turn expected a prompt-cache
        # hit but got none, why? Bucketed by reason so operators can tell a
        # TTL lapse (idle longer than the cache lifetime → consider a longer
        # TTL) from a prefix-content change (the cacheable prefix shifted).
        # Reasons are the MISS_* literals from prefix_tracker. Per provider so
        # the dashboard can scope; populated by `record_cache_miss_attribution`.
        self.cache_miss_attribution_by_provider: dict[str, dict[str, int]] = defaultdict(
            lambda: defaultdict(int)
        )

        # Cumulative savings history (timestamp → cumulative tokens saved)
        self.savings_history: list[tuple[str, int]] = []
        self.savings_tracker = savings_tracker or SavingsTracker(
            stateless=stateless, save_flush_every=PROXY_SAVINGS_FLUSH_EVERY
        )
        self.cost_tracker = cost_tracker
        tracker_lifetime = self.savings_tracker.snapshot()["lifetime"]
        self._savings_tracker_input_tokens_offset = max(
            int(tracker_lifetime.get("total_input_tokens", 0) or 0),
            0,
        )
        self._savings_tracker_input_cost_usd_offset = max(
            float(tracker_lifetime.get("total_input_cost_usd", 0.0) or 0.0),
            0.0,
        )

        self._lock = asyncio.Lock()
        # Tiny synchronous critical section for stage-timing triple updates
        # (sum + count + max must move together for a consistent scrape).
        # threading.Lock is cheaper than asyncio.Lock and does NOT contend
        # with the async ``export()`` path — scrapes snapshot these dicts
        # under this lock in a microsecond block, then build the metrics
        # string without holding anything.
        self._stage_timing_lock = threading.Lock()
        self._otel_metrics = otel_metrics

    async def reset_runtime(self) -> None:
        """Reset in-memory request/compression counters for local test/debug use."""
        async with self._lock:
            self.requests_total = 0
            self.requests_by_provider.clear()
            self.requests_by_model.clear()
            self.requests_by_stack.clear()
            self.requests_cached = 0
            self.requests_rate_limited = 0
            self.requests_failed = 0
            self.inbound_requests_total = 0
            self.inbound_requests_completed = 0
            self.inbound_requests_active = 0
            self.inbound_requests_by_method.clear()
            self.inbound_requests_by_path.clear()
            self.inbound_responses_by_status.clear()

            self.tokens_input_total = 0
            self.tokens_output_total = 0
            self.tokens_saved_total = 0
            self.attempted_input_tokens_total = 0

            self.compressions_by_strategy.clear()
            self.tokens_saved_by_strategy.clear()

            self.codex_ws_units_total = 0
            self.codex_ws_units_modified_total = 0
            self.codex_ws_units_to_kompress_total = 0
            self.codex_ws_units_kompress_attempted_total = 0
            self.codex_ws_units_by_strategy.clear()
            self.codex_ws_units_by_category.clear()
            self.codex_ws_units_by_content_type.clear()
            self.codex_ws_units_by_text_shape.clear()
            self.codex_ws_unit_elapsed_ms_sum = 0.0
            self.codex_ws_unit_elapsed_ms_max = 0.0
            self.codex_ws_unit_bytes_sum = 0
            self.codex_ws_unit_tokens_before_sum = 0
            self.codex_ws_unit_tokens_after_sum = 0
            self.codex_ws_unit_tokens_saved_sum = 0

            self.codex_ws_frames_attempted_total = 0
            self.codex_ws_frames_compressed_total = 0
            self.codex_ws_frames_failed_total = 0
            self.codex_ws_frames_to_kompress_total = 0
            self.codex_ws_frames_kompress_attempted_total = 0
            self.codex_ws_frame_elapsed_ms_sum = 0.0
            self.codex_ws_frame_elapsed_ms_max = 0.0
            self.codex_ws_frame_bytes_before_sum = 0
            self.codex_ws_frame_bytes_after_sum = 0
            self.codex_ws_frame_attempted_tokens_sum = 0
            self.codex_ws_frame_tokens_saved_sum = 0

            self.latency_sum_ms = 0.0
            self.latency_min_ms = float("inf")
            self.latency_max_ms = 0.0
            self.latency_count = 0

            self.overhead_sum_ms = 0.0
            self.overhead_min_ms = float("inf")
            self.overhead_max_ms = 0.0
            self.overhead_count = 0

            self.ttfb_sum_ms = 0.0
            self.ttfb_min_ms = float("inf")
            self.ttfb_max_ms = 0.0
            self.ttfb_count = 0

            self.transform_timing_sum.clear()
            self.transform_timing_count.clear()
            self.transform_timing_max.clear()

            self.waste_signals_total.clear()
            self.cache_by_provider.clear()
            self._cache_requests_by_model.clear()

            self.prefix_freeze_busts_avoided = 0
            self.prefix_freeze_tokens_preserved = 0
            self.prefix_freeze_compression_foregone = 0
            self.cache_bust_tokens_lost = 0
            self.cache_bust_count = 0
            self.cache_miss_attribution_by_provider.clear()
            self.savings_history = []

        with self._stage_timing_lock:
            self.stage_timing_sum.clear()
            self.stage_timing_count.clear()
            self.stage_timing_max.clear()
            self.ws_session_duration_sum_ms.clear()
            self.ws_session_duration_count.clear()
            self.ws_session_duration_max_ms.clear()

    def _get_otel_metrics(self) -> HeadroomOtelMetrics:
        return self._otel_metrics or get_otel_metrics()

    def _current_savings_tracker_totals(self) -> tuple[int, float]:
        total_input_tokens = self._savings_tracker_input_tokens_offset + self.tokens_input_total
        total_input_cost_usd = self._savings_tracker_input_cost_usd_offset

        if self.cost_tracker is None:
            return total_input_tokens, total_input_cost_usd

        try:
            cost_stats = self.cost_tracker.stats()
        except Exception:
            logger.debug("Failed to read cost tracker totals for savings history", exc_info=True)
            return total_input_tokens, total_input_cost_usd

        tracked_input_tokens = cost_stats.get("total_input_tokens")
        tracked_input_cost_usd = cost_stats.get("total_input_cost_usd")

        if tracked_input_tokens is not None:
            try:
                total_input_tokens = self._savings_tracker_input_tokens_offset + max(
                    int(tracked_input_tokens),
                    0,
                )
            except (TypeError, ValueError):
                pass

        if tracked_input_cost_usd is not None:
            try:
                total_input_cost_usd = self._savings_tracker_input_cost_usd_offset + max(
                    float(tracked_input_cost_usd),
                    0.0,
                )
            except (TypeError, ValueError):
                pass

        return total_input_tokens, total_input_cost_usd

    def record_stack(self, stack: str | None) -> None:
        """Increment the per-stack request counter.

        ``stack`` is the ``X-Headroom-Stack`` header value (e.g.
        ``adapter_ts_openai``). Called once per inbound request from the
        proxy's stack middleware; a no-op when the header is absent, fails
        validation, or would exceed the cardinality cap.
        """

        from headroom.telemetry.context import MAX_DISTINCT_STACKS, normalize_stack

        slug = normalize_stack(stack)
        if not slug:
            return
        if (
            slug not in self.requests_by_stack
            and len(self.requests_by_stack) >= MAX_DISTINCT_STACKS
        ):
            return
        self.requests_by_stack[slug] += 1

    def record_compression(
        self,
        strategy: str,
        original_tokens: int,
        compressed_tokens: int,
    ) -> None:
        """Implements `headroom.transforms.observability.CompressionObserver`.

        Called once per real compression event by the configured
        transforms (ContentRouter at routing-decision granularity;
        SmartCrusher at message granularity in the legacy direct-
        pipeline path). Increments the per-strategy counters that
        get exported as labelled Prometheus metrics, so silent
        regressions in any single strategy become visible in the
        scrape.

        Synchronous + lock-free: `defaultdict(int)` writes are
        atomic under the GIL for these key types; the proxy serves
        many requests concurrently and the contention here would be
        a single dict write per routing decision.

        Tokens saved is `max(0, original - compressed)` — the
        observer never records "negative savings" even if a
        compressor goofs and emits more tokens than it received.
        """
        self.compressions_by_strategy[strategy] += 1
        saved = original_tokens - compressed_tokens
        if saved > 0:
            self.tokens_saved_by_strategy[strategy] += saved

    def record_router_route_counts(self, counts: dict[str, int]) -> None:
        """Accumulate ContentRouter routing-category counts for a single
        pass. The router emits a dict like ``{"user_msg": 12,
        "recent_code": 4, ...}`` summarising how it categorised each
        message in that request. Adding these into a long-running
        counter gives `/stats` a session-level breakdown so operators
        can see, e.g., that 80% of messages were protected as
        `user_msg` and only 5% reached the compressor (#454).
        """
        for category, count in counts.items():
            if count > 0:
                self.router_route_counts[category] += int(count)

    def record_codex_ws_unit(
        self,
        *,
        strategy: str,
        reason_category: str,
        elapsed_ms: float,
        text_bytes: int,
        tokens_before: int,
        tokens_after: int,
        tokens_saved: int,
        modified: bool,
        strategy_chain: list[str] | None = None,
        content_type: str = "unknown",
        text_shape: str = "unknown",
    ) -> None:
        """Record one Codex WS compression unit decision."""

        strategy = strategy or "unknown"
        reason_category = reason_category or "unknown"
        chain = strategy_chain or []

        self.codex_ws_units_total += 1
        self.codex_ws_units_by_strategy[strategy] += 1
        self.codex_ws_units_by_category[reason_category] += 1
        self.codex_ws_units_by_content_type[content_type or "unknown"] += 1
        self.codex_ws_units_by_text_shape[text_shape or "unknown"] += 1
        if modified:
            self.codex_ws_units_modified_total += 1
        if strategy == "kompress":
            self.codex_ws_units_to_kompress_total += 1
        if "kompress" in chain or strategy == "kompress":
            self.codex_ws_units_kompress_attempted_total += 1

        elapsed_ms = max(0.0, float(elapsed_ms))
        self.codex_ws_unit_elapsed_ms_sum += elapsed_ms
        self.codex_ws_unit_elapsed_ms_max = max(self.codex_ws_unit_elapsed_ms_max, elapsed_ms)
        self.codex_ws_unit_bytes_sum += max(0, int(text_bytes))
        self.codex_ws_unit_tokens_before_sum += max(0, int(tokens_before))
        self.codex_ws_unit_tokens_after_sum += max(0, int(tokens_after))
        self.codex_ws_unit_tokens_saved_sum += max(0, int(tokens_saved))

    def record_codex_ws_frame(
        self,
        *,
        elapsed_ms: float,
        bytes_before: int,
        bytes_after: int = 0,
        attempted_tokens: int = 0,
        tokens_saved: int = 0,
        modified: bool = False,
        failed: bool = False,
        strategy_chain: list[str] | None = None,
        final_strategies: list[str] | None = None,
    ) -> None:
        """Record one Codex WS response.create compression attempt."""

        chain = strategy_chain or []
        strategies = final_strategies or []

        self.codex_ws_frames_attempted_total += 1
        if modified:
            self.codex_ws_frames_compressed_total += 1
        if failed:
            self.codex_ws_frames_failed_total += 1
        if "kompress" in strategies:
            self.codex_ws_frames_to_kompress_total += 1
        if "kompress" in chain or "kompress" in strategies:
            self.codex_ws_frames_kompress_attempted_total += 1

        elapsed_ms = max(0.0, float(elapsed_ms))
        self.codex_ws_frame_elapsed_ms_sum += elapsed_ms
        self.codex_ws_frame_elapsed_ms_max = max(self.codex_ws_frame_elapsed_ms_max, elapsed_ms)
        self.codex_ws_frame_bytes_before_sum += max(0, int(bytes_before))
        self.codex_ws_frame_bytes_after_sum += max(0, int(bytes_after))
        self.codex_ws_frame_attempted_tokens_sum += max(0, int(attempted_tokens))
        self.codex_ws_frame_tokens_saved_sum += max(0, int(tokens_saved))

    def record_inbound_request(self, *, method: str, path: str) -> None:
        self.inbound_requests_total += 1
        self.inbound_requests_active += 1
        self.inbound_requests_by_method[method.upper()] += 1
        self.inbound_requests_by_path[path] += 1

    def record_inbound_response(self, *, status_code: int | str) -> None:
        self.inbound_requests_completed += 1
        self.inbound_requests_active = max(0, self.inbound_requests_active - 1)
        self.inbound_responses_by_status[str(status_code)] += 1

    def record_inbound_aborted(self, *, reason: str) -> None:
        self.inbound_requests_completed += 1
        self.inbound_requests_active = max(0, self.inbound_requests_active - 1)
        self.inbound_responses_by_status[f"aborted:{reason}"] += 1

    def inbound_snapshot(self) -> dict[str, object]:
        return {
            "total": self.inbound_requests_total,
            "completed": self.inbound_requests_completed,
            "active": self.inbound_requests_active,
            "by_method": dict(self.inbound_requests_by_method),
            "by_path": dict(self.inbound_requests_by_path),
            "by_status": dict(self.inbound_responses_by_status),
        }

    async def record_request(
        self,
        provider: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
        tokens_saved: int,
        latency_ms: float,
        cached: bool = False,
        overhead_ms: float = 0,
        ttfb_ms: float = 0,
        pipeline_timing: dict[str, float] | None = None,
        waste_signals: dict[str, int] | None = None,
        cache_read_tokens: int = 0,
        cache_write_tokens: int = 0,
        cache_write_5m_tokens: int = 0,
        cache_write_1h_tokens: int = 0,
        uncached_input_tokens: int = 0,
        attempted_input_tokens: int = 0,
        project: str | None = None,
        client: str | None = None,
    ):
        """Record metrics for a request."""
        # Post-guard invariant (all providers): Headroom never forwards a request
        # larger than the original — handlers revert any inflation before sending
        # (verified clean on the wire). So compression savings are >= 0; a negative
        # here is an intermediate/hook token-count artifact that never reached the
        # model. Clamp so total_tokens_removed / avg_compression_pct reflect the
        # actually-forwarded bytes instead of surfacing spurious negatives.
        if tokens_saved < 0:
            logger.debug(
                "metrics.record: clamping negative tokens_saved=%d to 0 for %s (artifact; wire not inflated)",
                tokens_saved,
                model,
            )
            tokens_saved = 0
        async with self._lock:
            self.requests_total += 1
            self.requests_by_provider[provider] += 1
            self.requests_by_model[model] += 1

            if cached:
                self.requests_cached += 1

            self.tokens_input_total += input_tokens
            self.tokens_output_total += output_tokens
            self.tokens_saved_total += tokens_saved
            # 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))

            # Track provider-specific prefix cache metrics
            if cache_read_tokens > 0 or cache_write_tokens > 0:
                pc = self.cache_by_provider[provider]
                pc["cache_read_tokens"] += cache_read_tokens
                pc["cache_write_tokens"] += cache_write_tokens
                pc["cache_write_5m_tokens"] += cache_write_5m_tokens
                pc["cache_write_1h_tokens"] += cache_write_1h_tokens
                if cache_write_5m_tokens > 0:
                    pc["cache_write_5m_requests"] += 1
                if cache_write_1h_tokens > 0:
                    pc["cache_write_1h_requests"] += 1
                pc["uncached_input_tokens"] += uncached_input_tokens
                pc["requests"] += 1
                if cache_read_tokens > 0:
                    pc["hit_requests"] += 1
                # Model-aware bust detection: the first request for any model
                # is always a cold start (100% write, 0% read) — not a bust.
                # Only flag as bust when a previously-warm model suddenly has
                # high write ratio, indicating prefix invalidation.
                model_req_num = self._cache_requests_by_model[model]
                self._cache_requests_by_model[model] += 1
                if provider == "anthropic" and model_req_num > 0:
                    total_cached = cache_read_tokens + cache_write_tokens
                    if total_cached > 0 and cache_write_tokens > total_cached * 0.5:
                        pc["bust_count"] += 1
                        pc["bust_write_tokens"] += cache_write_tokens

            self.latency_sum_ms += latency_ms
            self.latency_min_ms = min(self.latency_min_ms, latency_ms)
            self.latency_max_ms = max(self.latency_max_ms, latency_ms)
            self.latency_count += 1

            # Track Headroom overhead separately
            if overhead_ms > 0:
                self.overhead_sum_ms += overhead_ms
                self.overhead_min_ms = min(self.overhead_min_ms, overhead_ms)
                self.overhead_max_ms = max(self.overhead_max_ms, overhead_ms)
                self.overhead_count += 1

            # Track TTFB (time to first byte from upstream)
            if ttfb_ms > 0:
                self.ttfb_sum_ms += ttfb_ms
                self.ttfb_min_ms = min(self.ttfb_min_ms, ttfb_ms)
                self.ttfb_max_ms = max(self.ttfb_max_ms, ttfb_ms)
                self.ttfb_count += 1

            # Track per-transform timing
            if pipeline_timing:
                for name, ms in pipeline_timing.items():
                    self.transform_timing_sum[name] += ms
                    self.transform_timing_count[name] += 1
                    self.transform_timing_max[name] = max(self.transform_timing_max[name], ms)

            # Track waste signals
            if waste_signals:
                for signal_name, token_count in waste_signals.items():
                    self.waste_signals_total[signal_name] += token_count

            # Track cumulative savings history (record every request)
            self.savings_history.append((datetime.now().isoformat(), self.tokens_saved_total))
            # Keep last 500 data points
            if len(self.savings_history) > 500:
                self.savings_history = self.savings_history[-500:]

            total_input_tokens, total_input_cost_usd = self._current_savings_tracker_totals()
            self.savings_tracker.record_request(
                model=model,
                input_tokens=input_tokens,
                tokens_saved=tokens_saved,
                provider=provider,
                project=project,
                cache_read_tokens=cache_read_tokens,
                cache_write_tokens=cache_write_tokens,
                uncached_input_tokens=uncached_input_tokens,
                total_input_tokens=total_input_tokens,
                total_input_cost_usd=total_input_cost_usd,
            )

            # Also append to the durable, multi-process savings ledger so
            # `headroom savings` reflects proxy traffic alongside MCP-tool usage.
            # The real upstream model means litellm prices it accurately. The
            # client is the harness classified from the User-Agent / X-Client
            # (claude-code, codex, cursor, ...); it falls back to "proxy" only
            # when the harness is unidentified.
            if tokens_saved > 0 and not self._stateless:
                savings_ledger.record_savings_event(
                    tokens_before=input_tokens,
                    tokens_after=max(input_tokens - tokens_saved, 0),
                    model=model,
                    client=client or "proxy",
                    source="proxy",
                )

        self._get_otel_metrics().record_proxy_request(
            provider=provider,
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            tokens_saved=tokens_saved,
            latency_ms=latency_ms,
            cached=cached,
            overhead_ms=overhead_ms,
            ttfb_ms=ttfb_ms,
            cache_read_tokens=cache_read_tokens,
            cache_write_tokens=cache_write_tokens,
            cache_write_5m_tokens=cache_write_5m_tokens,
            cache_write_1h_tokens=cache_write_1h_tokens,
            uncached_input_tokens=uncached_input_tokens,
        )

    async def record_stage_timings(
        self,
        path: str,
        timings: dict[str, float],
    ) -> None:
        """Record per-stage timings as histogram-style observations.

        ``path`` identifies the code path that emitted the timings (e.g.
        ``openai_responses_ws`` or ``anthropic_messages``). ``timings``
        maps stage names to millisecond durations. Mirrors the
        ``transform_timing_*`` aggregation pattern so the ``/metrics``
        endpoint exposes sum/count/max series per ``(path, stage)``.

        Uses a tiny synchronous ``threading.Lock`` around the triple
        update (sum + count + max) rather than the async
        ``self._lock``: (1) the updates have no awaits, so there is no
        async contention benefit, and (2) the async lock is also held
        by ``export()`` during Prometheus scrapes — which does
        string-building while holding it. Under N concurrent request
        finalizations + an active scrape, callers would queue behind
        the scrape's string-building.
        """
        if not timings:
            return
        with self._stage_timing_lock:
            for stage, ms in timings.items():
                try:
                    ms_val = float(ms)
                except (TypeError, ValueError):
                    continue
                key = (path, stage)
                self.stage_timing_sum[key] += ms_val
                self.stage_timing_count[key] += 1
                if ms_val > self.stage_timing_max[key]:
                    self.stage_timing_max[key] = ms_val

    async def record_cache_bust(self, tokens_lost: int) -> None:
        """Record tokens that lost their cache discount due to compression."""
        async with self._lock:
            self.cache_bust_tokens_lost += tokens_lost
            self.cache_bust_count += 1
        self._get_otel_metrics().record_proxy_cache_bust(tokens_lost=tokens_lost)

    async def record_cache_miss_attribution(self, provider: str, reason: str) -> None:
        """Record why a turn that expected a prompt-cache hit missed instead.

        ``reason`` is one of the MISS_* literals produced by
        ``PrefixCacheTracker.classify_cache_miss`` (``ttl_expiry``,
        ``prefix_change``, ``unknown``). Cold starts and hits are not recorded
        — only actual misses against a previously-cached prefix reach here.
        Bucketed per provider so the dashboard can scope or aggregate.
        """
        async with self._lock:
            self.cache_miss_attribution_by_provider[provider][reason] += 1

    # ------------------------------------------------------------------
    # Unit 3: WS session lifecycle gauges / histogram
    # ------------------------------------------------------------------

    def inc_active_ws_sessions(self) -> None:
        """Increment the live WS session gauge (called on register)."""
        self.active_ws_sessions += 1

    def dec_active_ws_sessions(self) -> None:
        """Decrement the live WS session gauge (called on deregister)."""
        self.active_ws_sessions = max(0, self.active_ws_sessions - 1)

    def inc_active_relay_tasks(self, n: int = 1) -> None:
        """Increment the live relay-task gauge (attach_tasks)."""
        self.active_relay_tasks += n

    def dec_active_relay_tasks(self, n: int = 1) -> None:
        """Decrement the live relay-task gauge (deregister)."""
        self.active_relay_tasks = max(0, self.active_relay_tasks - n)

    def record_ws_session_duration(
        self,
        duration_ms: float,
        cause: str = "unknown",
    ) -> None:
        """Record a completed WS session's duration, bucketed by cause.

        Mirrors the ``stage_timing_*`` shape so ``/metrics`` exposes
        sum/count/max per termination cause. Uses synchronous dict
        updates (no ``_lock``) because Unit 3 callers run on the event
        loop — matching the gauges above.
        """
        try:
            ms_val = float(duration_ms)
        except (TypeError, ValueError):
            return
        self.ws_session_duration_sum_ms[cause] += ms_val
        self.ws_session_duration_count[cause] += 1
        if ms_val > self.ws_session_duration_max_ms[cause]:
            self.ws_session_duration_max_ms[cause] = ms_val

    async def record_rate_limited(self, *, provider: str | None = None, model: str | None = None):
        async with self._lock:
            self.requests_rate_limited += 1
        self._get_otel_metrics().record_proxy_rate_limited(provider=provider, model=model)

    async def record_failed(self, *, provider: str | None = None, model: str | None = None):
        async with self._lock:
            self.requests_failed += 1
        self._get_otel_metrics().record_proxy_failed(provider=provider, model=model)

    async def export(self) -> str:
        """Export metrics in Prometheus format."""
        # Snapshot stage-timing dicts under the tiny synchronous lock so
        # we don't race a concurrent ``record_stage_timings`` and observe
        # an inconsistent (sum, count, max) triple. Freeze into plain
        # dicts so the scrape's string-building below doesn't hold the
        # stage-timing lock during I/O-ish work.
        with self._stage_timing_lock:
            stage_timing_sum_snapshot = dict(self.stage_timing_sum)
            stage_timing_count_snapshot = dict(self.stage_timing_count)
            stage_timing_max_snapshot = dict(self.stage_timing_max)
        async with self._lock:
            lifetime_savings = self.savings_tracker.snapshot()["lifetime"]
            lines: list[str] = []
            _append_metric(
                lines,
                name="headroom_requests_total",
                metric_type="counter",
                help_text="Total number of requests",
                value=self.requests_total,
            )
            _append_metric(
                lines,
                name="headroom_requests_cached_total",
                metric_type="counter",
                help_text="Cached request count",
                value=self.requests_cached,
            )
            _append_metric(
                lines,
                name="headroom_requests_rate_limited_total",
                metric_type="counter",
                help_text="Rate limited requests",
                value=self.requests_rate_limited,
            )
            _append_metric(
                lines,
                name="headroom_requests_failed_total",
                metric_type="counter",
                help_text="Failed requests",
                value=self.requests_failed,
            )
            _append_metric(
                lines,
                name="headroom_inbound_requests_total",
                metric_type="counter",
                help_text="All inbound HTTP requests accepted by the proxy",
                value=self.inbound_requests_total,
            )
            _append_metric(
                lines,
                name="headroom_inbound_requests_completed_total",
                metric_type="counter",
                help_text="Inbound HTTP requests completed or aborted by the proxy",
                value=self.inbound_requests_completed,
            )
            _append_metric(
                lines,
                name="headroom_inbound_requests_active",
                metric_type="gauge",
                help_text="Inbound HTTP requests currently active in the proxy",
                value=self.inbound_requests_active,
            )
            _append_metric(
                lines,
                name="headroom_tokens_input_total",
                metric_type="counter",
                help_text="Total input tokens",
                value=self.tokens_input_total,
            )
            _append_metric(
                lines,
                name="headroom_tokens_output_total",
                metric_type="counter",
                help_text="Total output tokens",
                value=self.tokens_output_total,
            )
            _append_metric(
                lines,
                name="headroom_tokens_saved_total",
                metric_type="counter",
                help_text="Tokens saved by optimization",
                value=self.tokens_saved_total,
            )
            _append_metric(
                lines,
                name="headroom_persistent_savings_requests_total",
                metric_type="counter",
                help_text="Durable lifetime requests recorded by the proxy savings tracker",
                value=lifetime_savings["requests"],
            )
            _append_metric(
                lines,
                name="headroom_persistent_savings_tokens_saved_total",
                metric_type="counter",
                help_text="Durable lifetime input tokens saved by proxy compression",
                value=lifetime_savings["tokens_saved"],
            )
            _append_metric(
                lines,
                name="headroom_persistent_savings_input_tokens_total",
                metric_type="counter",
                help_text="Durable lifetime input tokens recorded by the proxy savings tracker",
                value=lifetime_savings["total_input_tokens"],
            )
            _append_metric(
                lines,
                name="headroom_persistent_savings_input_cost_usd_total",
                metric_type="counter",
                help_text="Durable lifetime input spend in USD estimated by the proxy savings tracker",
                value=lifetime_savings["total_input_cost_usd"],
            )
            _append_metric(
                lines,
                name="headroom_persistent_savings_compression_savings_usd_total",
                metric_type="counter",
                help_text=(
                    "Durable lifetime compression savings in USD estimated by the "
                    "proxy savings tracker"
                ),
                value=lifetime_savings["compression_savings_usd"],
            )
            # NOTE: per-strategy compression breakdown is tracked
            # internally on `self.compressions_by_strategy` and
            # `self.tokens_saved_by_strategy` (populated by
            # `record_compression`) but **deliberately not exported
            # here** as individual Prometheus series. The state is
            # still observable via /stats + tests + programmatic
            # introspection.
            _append_metric(
                lines,
                name="headroom_latency_ms_sum",
                metric_type="counter",
                help_text="Sum of request latencies in milliseconds",
                value=round(self.latency_sum_ms, 2),
            )
            _append_metric(
                lines,
                name="headroom_latency_ms_count",
                metric_type="counter",
                help_text="Count of observed request latencies",
                value=self.latency_count,
            )
            _append_metric(
                lines,
                name="headroom_latency_ms_min",
                metric_type="gauge",
                help_text="Minimum observed request latency in milliseconds",
                value=0 if self.latency_count == 0 else round(self.latency_min_ms, 2),
            )
            _append_metric(
                lines,
                name="headroom_latency_ms_max",
                metric_type="gauge",
                help_text="Maximum observed request latency in milliseconds",
                value=round(self.latency_max_ms, 2),
            )
            _append_metric(
                lines,
                name="headroom_overhead_ms_sum",
                metric_type="counter",
                help_text="Sum of Headroom processing overhead in milliseconds",
                value=round(self.overhead_sum_ms, 2),
            )
            _append_metric(
                lines,
                name="headroom_overhead_ms_count",
                metric_type="counter",
                help_text="Count of observed Headroom overhead samples",
                value=self.overhead_count,
            )
            _append_metric(
                lines,
                name="headroom_overhead_ms_min",
                metric_type="gauge",
                help_text="Minimum observed Headroom overhead in milliseconds",
                value=0 if self.overhead_count == 0 else round(self.overhead_min_ms, 2),
            )
            _append_metric(
                lines,
                name="headroom_overhead_ms_max",
                metric_type="gauge",
                help_text="Maximum observed Headroom overhead in milliseconds",
                value=round(self.overhead_max_ms, 2),
            )
            _append_metric(
                lines,
                name="headroom_ttfb_ms_sum",
                metric_type="counter",
                help_text="Sum of time to first byte in milliseconds",
                value=round(self.ttfb_sum_ms, 2),
            )
            _append_metric(
                lines,
                name="headroom_ttfb_ms_count",
                metric_type="counter",
                help_text="Count of observed time-to-first-byte samples",
                value=self.ttfb_count,
            )
            _append_metric(
                lines,
                name="headroom_ttfb_ms_min",
                metric_type="gauge",
                help_text="Minimum observed time to first byte in milliseconds",
                value=0 if self.ttfb_count == 0 else round(self.ttfb_min_ms, 2),
            )
            _append_metric(
                lines,
                name="headroom_ttfb_ms_max",
                metric_type="gauge",
                help_text="Maximum observed time to first byte in milliseconds",
                value=round(self.ttfb_max_ms, 2),
            )
            _append_metric(
                lines,
                name="headroom_cache_bust_total",
                metric_type="counter",
                help_text="Requests that lost provider cache efficiency because of compression",
                value=self.cache_bust_count,
            )
            _append_metric(
                lines,
                name="headroom_cache_bust_tokens_lost_total",
                metric_type="counter",
                help_text="Tokens that lost provider cache discount because of compression",
                value=self.cache_bust_tokens_lost,
            )

            if self.cache_miss_attribution_by_provider:
                lines.extend(
                    [
                        "# HELP headroom_cache_miss_attribution_total Cache misses on an "
                        "expected-cached prefix, bucketed by reason (ttl_expiry|prefix_change|unknown)",
                        "# TYPE headroom_cache_miss_attribution_total counter",
                    ]
                )
                for _provider, _reasons in self.cache_miss_attribution_by_provider.items():
                    for _reason, _count in _reasons.items():
                        lines.append(
                            f'headroom_cache_miss_attribution_total{{provider="{_provider}",'
                            f'reason="{_reason}"}} {_count}'
                        )
                lines.append("")

            lines.extend(
                [
                    "# HELP headroom_requests_by_provider Requests by provider",
                    "# TYPE headroom_requests_by_provider counter",
                ]
            )
            for provider, count in self.requests_by_provider.items():
                lines.append(f'headroom_requests_by_provider{{provider="{provider}"}} {count}')
            lines.append("")

            lines.extend(
                [
                    "# HELP headroom_requests_by_model Requests by model",
                    "# TYPE headroom_requests_by_model counter",
                ]
            )
            for model, count in self.requests_by_model.items():
                lines.append(f'headroom_requests_by_model{{model="{model}"}} {count}')
            lines.append("")

            if self.transform_timing_sum:
                lines.extend(
                    [
                        "# HELP headroom_transform_timing_ms_sum Sum of transform timing in milliseconds",
                        "# TYPE headroom_transform_timing_ms_sum counter",
                    ]
                )
                for name, total in self.transform_timing_sum.items():
                    lines.append(
                        f'headroom_transform_timing_ms_sum{{transform="{_escape_label_value(name)}"}} {round(total, 2)}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_transform_timing_ms_count Count of transform timing samples",
                        "# TYPE headroom_transform_timing_ms_count counter",
                    ]
                )
                for name, count in self.transform_timing_count.items():
                    lines.append(
                        f'headroom_transform_timing_ms_count{{transform="{_escape_label_value(name)}"}} {count}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_transform_timing_ms_max Maximum transform timing in milliseconds",
                        "# TYPE headroom_transform_timing_ms_max gauge",
                    ]
                )
                for name, max_value in self.transform_timing_max.items():
                    lines.append(
                        f'headroom_transform_timing_ms_max{{transform="{_escape_label_value(name)}"}} {round(max_value, 2)}'
                    )
                lines.append("")

            if stage_timing_sum_snapshot:
                lines.extend(
                    [
                        "# HELP headroom_stage_timing_ms_sum Sum of per-stage handler timings in milliseconds",
                        "# TYPE headroom_stage_timing_ms_sum counter",
                    ]
                )
                for (path_label, stage), total in stage_timing_sum_snapshot.items():
                    lines.append(
                        f'headroom_stage_timing_ms_sum{{path="{_escape_label_value(path_label)}",stage="{_escape_label_value(stage)}"}} {round(total, 2)}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_stage_timing_ms_count Count of per-stage handler timing samples",
                        "# TYPE headroom_stage_timing_ms_count counter",
                    ]
                )
                for (path_label, stage), count in stage_timing_count_snapshot.items():
                    lines.append(
                        f'headroom_stage_timing_ms_count{{path="{_escape_label_value(path_label)}",stage="{_escape_label_value(stage)}"}} {count}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_stage_timing_ms_max Maximum per-stage handler timing in milliseconds",
                        "# TYPE headroom_stage_timing_ms_max gauge",
                    ]
                )
                for (path_label, stage), max_value in stage_timing_max_snapshot.items():
                    lines.append(
                        f'headroom_stage_timing_ms_max{{path="{_escape_label_value(path_label)}",stage="{_escape_label_value(stage)}"}} {round(max_value, 2)}'
                    )
                lines.append("")

            # Unit 3: WS session lifecycle gauges + duration histogram.
            lines.extend(
                [
                    "# HELP headroom_active_ws_sessions Active Codex WebSocket sessions",
                    "# TYPE headroom_active_ws_sessions gauge",
                    f"headroom_active_ws_sessions {self.active_ws_sessions}",
                    "",
                    "# HELP headroom_active_relay_tasks Active Codex WS relay tasks",
                    "# TYPE headroom_active_relay_tasks gauge",
                    f"headroom_active_relay_tasks {self.active_relay_tasks}",
                    "",
                ]
            )
            if self.ws_session_duration_sum_ms:
                lines.extend(
                    [
                        "# HELP headroom_ws_session_duration_ms_sum Sum of Codex WS session durations",
                        "# TYPE headroom_ws_session_duration_ms_sum counter",
                    ]
                )
                for cause, total in self.ws_session_duration_sum_ms.items():
                    lines.append(
                        f'headroom_ws_session_duration_ms_sum{{cause="{_escape_label_value(cause)}"}} {round(total, 2)}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_ws_session_duration_ms_count Count of completed Codex WS sessions",
                        "# TYPE headroom_ws_session_duration_ms_count counter",
                    ]
                )
                for cause, count in self.ws_session_duration_count.items():
                    lines.append(
                        f'headroom_ws_session_duration_ms_count{{cause="{_escape_label_value(cause)}"}} {count}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_ws_session_duration_ms_max Maximum Codex WS session duration",
                        "# TYPE headroom_ws_session_duration_ms_max gauge",
                    ]
                )
                for cause, max_value in self.ws_session_duration_max_ms.items():
                    lines.append(
                        f'headroom_ws_session_duration_ms_max{{cause="{_escape_label_value(cause)}"}} {round(max_value, 2)}'
                    )
                lines.append("")

            if self.waste_signals_total:
                lines.extend(
                    [
                        "# HELP headroom_waste_signal_tokens_total Tokens attributed to detected waste signals",
                        "# TYPE headroom_waste_signal_tokens_total counter",
                    ]
                )
                for signal_name, token_count in self.waste_signals_total.items():
                    lines.append(
                        f'headroom_waste_signal_tokens_total{{signal="{_escape_label_value(signal_name)}"}} {token_count}'
                    )
                lines.append("")

            if self.cache_by_provider:
                lines.extend(
                    [
                        "# HELP headroom_cache_read_tokens_total Provider cache read tokens",
                        "# TYPE headroom_cache_read_tokens_total counter",
                    ]
                )
                for provider, stats in self.cache_by_provider.items():
                    lines.append(
                        f'headroom_cache_read_tokens_total{{provider="{provider}"}} {stats["cache_read_tokens"]}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_cache_write_tokens_total Provider cache write tokens",
                        "# TYPE headroom_cache_write_tokens_total counter",
                    ]
                )
                for provider, stats in self.cache_by_provider.items():
                    lines.append(
                        f'headroom_cache_write_tokens_total{{provider="{provider}"}} {stats["cache_write_tokens"]}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_cache_write_ttl_tokens_total Provider cache write tokens by observed TTL bucket",
                        "# TYPE headroom_cache_write_ttl_tokens_total counter",
                    ]
                )
                for provider, stats in self.cache_by_provider.items():
                    lines.append(
                        f'headroom_cache_write_ttl_tokens_total{{provider="{provider}",ttl="5m"}} {stats["cache_write_5m_tokens"]}'
                    )
                    lines.append(
                        f'headroom_cache_write_ttl_tokens_total{{provider="{provider}",ttl="1h"}} {stats["cache_write_1h_tokens"]}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_cache_write_ttl_requests_total Provider cache write requests by observed TTL bucket",
                        "# TYPE headroom_cache_write_ttl_requests_total counter",
                    ]
                )
                for provider, stats in self.cache_by_provider.items():
                    lines.append(
                        f'headroom_cache_write_ttl_requests_total{{provider="{provider}",ttl="5m"}} {stats["cache_write_5m_requests"]}'
                    )
                    lines.append(
                        f'headroom_cache_write_ttl_requests_total{{provider="{provider}",ttl="1h"}} {stats["cache_write_1h_requests"]}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_uncached_input_tokens_total Input tokens not served from provider cache",
                        "# TYPE headroom_uncached_input_tokens_total counter",
                    ]
                )
                for provider, stats in self.cache_by_provider.items():
                    lines.append(
                        f'headroom_uncached_input_tokens_total{{provider="{provider}"}} {stats["uncached_input_tokens"]}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_provider_cache_requests_total Requests with provider cache observations",
                        "# TYPE headroom_provider_cache_requests_total counter",
                    ]
                )
                for provider, stats in self.cache_by_provider.items():
                    lines.append(
                        f'headroom_provider_cache_requests_total{{provider="{provider}"}} {stats["requests"]}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_provider_cache_hit_requests_total Requests with provider cache reads",
                        "# TYPE headroom_provider_cache_hit_requests_total counter",
                    ]
                )
                for provider, stats in self.cache_by_provider.items():
                    lines.append(
                        f'headroom_provider_cache_hit_requests_total{{provider="{provider}"}} {stats["hit_requests"]}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_provider_cache_bust_total Provider-specific cache bust count",
                        "# TYPE headroom_provider_cache_bust_total counter",
                    ]
                )
                for provider, stats in self.cache_by_provider.items():
                    lines.append(
                        f'headroom_provider_cache_bust_total{{provider="{provider}"}} {stats["bust_count"]}'
                    )
                lines.extend(
                    [
                        "",
                        "# HELP headroom_provider_cache_bust_write_tokens_total Provider cache write tokens attributed to busts",
                        "# TYPE headroom_provider_cache_bust_write_tokens_total counter",
                    ]
                )
                for provider, stats in self.cache_by_provider.items():
                    lines.append(
                        f'headroom_provider_cache_bust_write_tokens_total{{provider="{provider}"}} {stats["bust_write_tokens"]}'
                    )
                lines.append("")

            # Phase G PR-G3 remediation (C3): image-redacted counter
            # lives Python-side because base64 redaction is purely a
            # Python-proxy concern (request_logger.py). The Rust
            # proxy previously held a dead counter for this; that's
            # been removed in favour of this Python export.
            #
            # The counter is read at scrape-time from the module-
            # level redaction tracker rather than mirrored into the
            # PrometheusMetrics instance, so we never lose a count
            # to ordering between RequestLogger setup and metrics
            # init.
            from headroom.proxy.request_logger import redactions_total

            _append_metric(
                lines,
                name="proxy_image_generation_call_log_redacted_total",
                metric_type="counter",
                help_text=(
                    "Count of base64-encoded image payloads redacted from request "
                    "logs by the Python proxy's request logger"
                ),
                value=redactions_total(),
            )

            # Phase G PR-G3 remediation (C4): RTK invocations counter
            # also lives Python-side. RTK is wrapped by the
            # `headroom wrap` CLI (headroom.cli.wrap); the proxy
            # observes invocation counts via a process-local tracker
            # the wrap tail bumps. The Rust proxy previously held a
            # dead counter for this; that's been removed.
            from headroom.cli.wrap_rtk_metrics import rtk_invocation_counts

            counts = rtk_invocation_counts()
            lines.extend(
                [
                    "# HELP wrap_rtk_invocations_total RTK invocations observed via the wrap CLI tail",
                    "# TYPE wrap_rtk_invocations_total counter",
                ]
            )
            if not counts:
                # Emit a zero-row under the sentinel tool name so
                # the family advertises HELP/TYPE on a fresh boot
                # and dashboards can probe it before any RTK
                # invocation has happened. Matches the Rust side's
                # H3 force-zero contract.
                lines.append('wrap_rtk_invocations_total{tool="__init__"} 0')
            else:
                for tool, count in counts.items():
                    safe_tool = _escape_label_value(str(tool))
                    lines.append(f'wrap_rtk_invocations_total{{tool="{safe_tool}"}} {count}')
            lines.append("")

            return "\n".join(lines)