feat(telemetry): record provider cache read/write/uncached tokens per request (#2450)

## Description

The per-request JSONL feed (`--log-file`) collapsed all cache signal
into a single `cache_hit: bool`, defined as `cache_read_tokens > 0 or
from_response_cache`. A call that was billed cache-*creation* (write)
with zero reads is therefore indistinguishable from a real cache-*read*
hit. On Claude Code traffic where the proxy pays repeated cache writes,
this hides the real economics from users (the issue's "cache_hit inverts
the user's real economics" telemetry complaint).

The provider-truth counters already ride on `RequestOutcome`,
`cache_read_tokens`, `cache_write_tokens`, `uncached_input_tokens`,
parsed from the upstream response usage on every path
(`handlers/anthropic.py`, `handlers/streaming.py`,
`backends/litellm.py`) but were dropped when the `RequestLog` entry was
constructed in `emit_request_outcome`. This surfaces them per call.

Refs #2438 (Finding 1, telemetry sub-item). The core prompt-cache
preservation regression (Finding 1) and Finding 3 are architectural and
tracked separately.

## Type of Change

- [ ] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [x] Code refactoring (no functional changes)

## Changes Made

- Add `cache_read_tokens`, `cache_write_tokens`, `uncached_input_tokens`
(optional, default `0`) to `RequestLog` (`headroom/proxy/models.py`).
Optional so existing consumers and serialized logs stay backward
compatible.
- Populate the three fields at the single log-emit site in
`emit_request_outcome` (`headroom/proxy/outcome.py`) from the values
already on `RequestOutcome`. `cache_hit` is unchanged.
- Add `tests/test_proxy_cache_telemetry.py`: drive
`emit_request_outcome` through the real proxy funnel with logging
enabled and assert the JSONL entry carries the write/uncached deltas
even when `cache_hit` is False; plus a default-value backward-compat
check.
- Leave `CHANGELOG.md` untouched release-please generates it.

## Testing

- [x] Unit tests pass (`python -m pytest
tests/test_proxy_cache_telemetry.py -q`)
- [x] Linting passes (`ruff check`, `ruff format --check` on the three
changed files)
- [x] Type checking passes (`mypy headroom/proxy/models.py
headroom/proxy/outcome.py --ignore-missing-imports`)
- [x] New tests added for new functionality

### Test Output

```text
$ python -m pytest tests/test_proxy_cache_telemetry.py -q
2 passed, 1 warning in 8.51s

$ ruff check headroom/proxy/models.py headroom/proxy/outcome.py tests/test_proxy_cache_telemetry.py
All checks passed!
```

## Real Behavior Proof

- Environment: Windows 11, Python 3.13, local dev checkout on a branch
off upstream/main
- Exact command / steps: Built a `RequestOutcome` with
`cache_read_tokens=0, cache_write_tokens=800, uncached_input_tokens=200`
and ran it through `emit_request_outcome` against a real proxy app
(`create_app`) with `log_requests=True` and a temp `log_file`, then read
the JSONL back.
- Observed result: The written entry carries `cache_read_tokens=0`,
`cache_write_tokens=800`, `uncached_input_tokens=200` a cache-write-only
call is now distinguishable from a cache-read hit in the log, where
previously only `cache_hit` (False here) was recorded.
- Not tested: A live Anthropic call end to end from this environment,
the funnel is exercised with a synthetic outcome carrying real
provider-usage values instead.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

headroom/proxy/models.py

@@ -69,6 +69,16 @@ class RequestLog:
     cache_hit: bool
     transforms_applied: list[str]
 
+    # 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 —
+    # already carried on RequestOutcome from the upstream response usage —
+    # let the JSONL telemetry reflect true economics (uncached input +
+    # cache_creation), not just the proxy's boolean.
+    cache_read_tokens: int = 0
+    cache_write_tokens: int = 0
+    uncached_input_tokens: int = 0
+
     # Waste signals detected in original messages
     waste_signals: dict[str, int] | None = None
 

headroom/proxy/outcome.py

@@ -454,6 +454,9 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
                 total_latency_ms=outcome.total_latency_ms,
                 tags=log_tags,
                 cache_hit=outcome.cache_hit,
+                cache_read_tokens=outcome.cache_read_tokens,
+                cache_write_tokens=outcome.cache_write_tokens,
+                uncached_input_tokens=outcome.uncached_input_tokens,
                 transforms_applied=list(outcome.transforms_applied),
                 waste_signals=outcome.waste_signals,
                 request_messages=outcome.request_messages,

tests/test_proxy_cache_telemetry.py

@@ -0,0 +1,82 @@
+"""Provider-side cache economics land in the per-request JSONL (#2438).
+
+`cache_hit` alone can't distinguish a call billed cache-*creation* (write)
+from a real cache-*read* hit, which is exactly the telemetry gap the issue
+reported (the proxy stamps `cache_hit: true` while the client is billed
+cache-write tokens). The raw provider deltas already live on RequestOutcome;
+this pins that they survive into the RequestLog feed.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+
+pytest.importorskip("fastapi")
+from fastapi.testclient import TestClient  # noqa: E402
+
+from headroom.proxy.models import RequestLog  # noqa: E402
+from headroom.proxy.outcome import RequestOutcome, emit_request_outcome  # noqa: E402
+from headroom.proxy.server import ProxyConfig, create_app  # noqa: E402
+
+
+def test_request_log_carries_provider_cache_deltas(tmp_path):
+    log_file = tmp_path / "proxy.jsonl"
+    config = ProxyConfig(
+        cache_enabled=False,
+        rate_limit_enabled=False,
+        log_requests=True,
+        log_file=str(log_file),
+    )
+
+    with TestClient(create_app(config)) as client:
+        proxy = client.app.state.proxy
+
+        # A call billed cache-*creation* (write), zero reads: cache_hit would
+        # be False here, but the write/uncached deltas must still be recorded
+        # so the true economics are visible.
+        outcome = RequestOutcome(
+            request_id="req-cache",
+            provider="anthropic",
+            model="claude-sonnet-5",
+            original_tokens=1000,
+            optimized_tokens=1000,
+            output_tokens=20,
+            tokens_saved=0,
+            attempted_input_tokens=1000,
+            cache_read_tokens=0,
+            cache_write_tokens=800,
+            uncached_input_tokens=200,
+        )
+        asyncio.run(emit_request_outcome(proxy, outcome))
+
+    lines = [json.loads(line) for line in log_file.read_text().splitlines() if line.strip()]
+    entry = next(e for e in lines if e["request_id"] == "req-cache")
+    assert entry["cache_read_tokens"] == 0
+    assert entry["cache_write_tokens"] == 800
+    assert entry["uncached_input_tokens"] == 200
+
+
+def test_request_log_cache_delta_fields_default_zero():
+    # Backward-compatible: the new fields are optional and default to 0.
+    entry = RequestLog(
+        request_id="r",
+        timestamp="t",
+        provider="anthropic",
+        model="m",
+        input_tokens_original=0,
+        input_tokens_optimized=0,
+        output_tokens=0,
+        tokens_saved=0,
+        savings_percent=0.0,
+        optimization_latency_ms=0.0,
+        total_latency_ms=None,
+        tags={},
+        cache_hit=False,
+        transforms_applied=[],
+    )
+    assert entry.cache_read_tokens == 0
+    assert entry.cache_write_tokens == 0
+    assert entry.uncached_input_tokens == 0