feat(telemetry): surface per-strategy compression counters

The PrometheusMetrics `compressions_by_strategy` and
`tokens_saved_by_strategy` counters (PR #302) were tracked in process
but never exported, because the Prometheus->Supabase pipeline treats
each metric name as a separate column. The data clock for the
code_compressor port-vs-retire decision needed actual production
visibility, not just CI assertions.

Add both dicts to the `/stats` endpoint and nest them under
`pipeline_timing._strategies` in the beacon Supabase payload. The
existing `pipeline_timing` column is JSONB and absorbs the nested
shape -- zero schema change.

`_build_pipeline_timing` is extracted from the inline payload-builder
so the projection has its own unit tests, including a guard that the
`_strategies` sub-key is only emitted when at least one counter is
non-empty.

The Prometheus scrape still does NOT carry these metrics; the existing
`test_prometheus_export_does_not_leak_per_strategy_metrics` guard
remains in place.

headroom/proxy/server.py

@@ -1618,6 +1618,8 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
             }
             if m.transform_timing_sum
             else {},
+            "compressions_by_strategy": dict(m.compressions_by_strategy),
+            "tokens_saved_by_strategy": dict(m.tokens_saved_by_strategy),
             "waste_signals": dict(m.waste_signals_total) if m.waste_signals_total else {},
             "savings_history": m.savings_history[-100:],  # Last 100 data points
             "display_session": display_session,

headroom/telemetry/beacon.py

@@ -44,6 +44,30 @@ _INTERVAL_SECONDS = 300
 _OFF_VALUES = frozenset(("off", "false", "0", "no", "disable", "disabled"))
 
 
+def _build_pipeline_timing(stats: dict) -> dict[str, object]:
+    """Project the /stats `pipeline_timing` JSONB payload for Supabase.
+
+    Flattens transform timing to {name: avg_ms} and, when present, nests
+    ContentRouter strategy counts and per-strategy tokens-saved totals
+    under a `_strategies` sub-key. The Supabase column is JSONB, so the
+    nested shape lands without a schema change.
+    """
+    raw_timing = stats.get("pipeline_timing", {}) or {}
+    pipeline_timing: dict[str, object] = {
+        name: round(info.get("average_ms", 0), 2)
+        for name, info in raw_timing.items()
+        if isinstance(info, dict)
+    }
+    strategies = stats.get("compressions_by_strategy", {}) or {}
+    tokens_by_strategy = stats.get("tokens_saved_by_strategy", {}) or {}
+    if strategies or tokens_by_strategy:
+        pipeline_timing["_strategies"] = {
+            "compressions": dict(strategies),
+            "tokens_saved": dict(tokens_by_strategy),
+        }
+    return pipeline_timing
+
+
 def is_telemetry_enabled() -> bool:
     """Check if telemetry is enabled (on by default, opt out with env var)."""
     val = os.environ.get("HEADROOM_TELEMETRY", "on").lower().strip()
@@ -245,17 +269,13 @@ class TelemetryBeacon:
             logger.debug("Beacon: failed to extract TTFB metrics", exc_info=True)
 
         # --- Pipeline timing breakdown (where is time spent?) ---
-        # Stored as JSONB — variable-shape dict of transform_name → avg_ms.
-        # This is the most valuable data for optimising Headroom internals.
+        # Stored as JSONB — variable-shape dict of transform_name → avg_ms,
+        # plus an optional `_strategies` sub-key carrying ContentRouter strategy
+        # counts and per-strategy tokens-saved totals (zero schema change — the
+        # JSONB column absorbs the nested shape).
         try:
-            raw_timing = stats.get("pipeline_timing", {})
-            if raw_timing:
-                # Flatten to {name: avg_ms} for compact storage
-                pipeline_timing = {
-                    name: round(info.get("average_ms", 0), 2)
-                    for name, info in raw_timing.items()
-                    if isinstance(info, dict)
-                }
+            pipeline_timing = _build_pipeline_timing(stats)
+            if pipeline_timing:
                 payload["pipeline_timing"] = pipeline_timing
         except Exception:
             logger.debug("Beacon: failed to extract pipeline timing", exc_info=True)

tests/test_strategy_stats_supabase.py

@@ -0,0 +1,91 @@
+"""Phase 3e.0: surface per-strategy compression counters.
+
+The internal `compressions_by_strategy` and `tokens_saved_by_strategy`
+counters (PR #302) were tracked in process but never exported, because
+the Prometheus→Supabase pipeline treats each metric name as a column and
+adding columns is operationally expensive.
+
+These tests pin the alternative path:
+
+1. The `/stats` endpoint exposes both dicts on the response root.
+2. The telemetry beacon nests them under `pipeline_timing._strategies`,
+   landing as JSONB inside the existing `pipeline_timing` Supabase
+   column — zero schema change.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+def test_stats_endpoint_exposes_per_strategy_counters() -> None:
+    pytest.importorskip("fastapi")
+    from fastapi.testclient import TestClient
+
+    from headroom.proxy.server import ProxyConfig, create_app
+
+    app = create_app(
+        ProxyConfig(
+            optimize=False,
+            cache_enabled=False,
+            rate_limit_enabled=False,
+            cost_tracking_enabled=False,
+            log_requests=False,
+            ccr_inject_tool=False,
+            ccr_handle_responses=False,
+            ccr_context_tracking=False,
+        )
+    )
+
+    proxy = app.state.proxy
+    proxy.metrics.record_compression("smart_crusher", original_tokens=300, compressed_tokens=80)
+    proxy.metrics.record_compression("smart_crusher", original_tokens=100, compressed_tokens=40)
+    proxy.metrics.record_compression("diff", original_tokens=120, compressed_tokens=70)
+
+    with TestClient(app) as client:
+        response = client.get("/stats")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["compressions_by_strategy"] == {"smart_crusher": 2, "diff": 1}
+    assert body["tokens_saved_by_strategy"] == {
+        "smart_crusher": (300 - 80) + (100 - 40),
+        "diff": 120 - 70,
+    }
+
+
+def test_beacon_nests_strategies_under_pipeline_timing() -> None:
+    from headroom.telemetry.beacon import _build_pipeline_timing
+
+    stats = {
+        "pipeline_timing": {
+            "smart_crusher": {"average_ms": 12.345, "max_ms": 50.0, "count": 10},
+            "diff_compressor": {"average_ms": 5.1, "max_ms": 9.0, "count": 4},
+        },
+        "compressions_by_strategy": {"smart_crusher": 4321, "diff": 87},
+        "tokens_saved_by_strategy": {"smart_crusher": 1_234_567, "diff": 23_456},
+    }
+
+    timing = _build_pipeline_timing(stats)
+
+    assert timing["smart_crusher"] == 12.35
+    assert timing["diff_compressor"] == 5.1
+    assert timing["_strategies"] == {
+        "compressions": {"smart_crusher": 4321, "diff": 87},
+        "tokens_saved": {"smart_crusher": 1_234_567, "diff": 23_456},
+    }
+
+
+def test_beacon_omits_strategies_subkey_when_counters_empty() -> None:
+    from headroom.telemetry.beacon import _build_pipeline_timing
+
+    stats = {
+        "pipeline_timing": {"router": {"average_ms": 1.2}},
+        "compressions_by_strategy": {},
+        "tokens_saved_by_strategy": {},
+    }
+
+    timing = _build_pipeline_timing(stats)
+
+    assert timing == {"router": 1.2}
+    assert "_strategies" not in timing