Recent Historical Checkpoints
@@ -1811,6 +1814,7 @@
function dashboard() {
return {
stats: {},
+ lifetimeStats: {},
historyStats: {},
healthy: true,
version: 'loading',
@@ -1824,8 +1828,10 @@
expandedRows: {},
pollInterval: null,
statsPollMs: 5000,
+ lifetimePollMs: 30000,
historyPollMs: 30000,
feedPollMs: 5000,
+ lastLifetimeFetchMs: 0,
lastHistoryFetchMs: 0,
lastFeedFetchMs: 0,
feedOpen: false,
@@ -1858,6 +1864,9 @@
await this.fetchStats();
const now = Date.now();
+ if (this.viewMode === 'lifetime' && (force || now - this.lastLifetimeFetchMs >= this.lifetimePollMs)) {
+ await this.fetchLifetimeStats();
+ }
if (this.viewMode === 'history' && (force || now - this.lastHistoryFetchMs >= this.historyPollMs)) {
await this.fetchHistoryStats();
}
@@ -1868,6 +1877,9 @@
async setViewMode(mode) {
this.viewMode = mode;
+ if (mode === 'lifetime') {
+ await this.fetchLifetimeStats();
+ }
if (mode === 'history') {
await this.fetchHistoryStats();
}
@@ -1914,6 +1926,17 @@
}
},
+ async fetchLifetimeStats() {
+ try {
+ const response = await fetch('/stats-lifetime');
+ if (response.ok) {
+ this.lifetimeStats = await response.json();
+ this.lastLifetimeFetchMs = Date.now();
+ }
+ } catch (e) {
+ console.error('Failed to fetch lifetime stats:', e);
+ }
+ },
async fetchHistoryStats() {
try {
const response = await fetch('/stats-history');
@@ -2054,28 +2077,12 @@
// --- Formatting ---
- hasNumber(n) {
- return typeof n === 'number' && Number.isFinite(n);
- },
-
formatNumber(n) {
if (n >= 1000000) return (n / 1000000).toFixed(1) + 'M';
if (n >= 1000) return (n / 1000).toFixed(1) + 'k';
return n.toString();
},
- formatOptionalNumber(n) {
- return this.hasNumber(n) ? this.formatNumber(n) : 'unknown';
- },
-
- formatOptionalPercent(n) {
- return this.hasNumber(n) ? n.toFixed(0) + '%' : 'unknown';
- },
-
- formatOptionalMs(n) {
- return this.hasNumber(n) ? n.toFixed(0) + 'ms' : 'unknown';
- },
-
formatCurrency(n) {
if (n < 0) return '-' + this.formatCurrency(-n);
if (n >= 1000) return (n / 1000).toFixed(1) + 'k';
diff --git a/headroom/proxy/forwarded_headers.py b/headroom/proxy/forwarded_headers.py
index 50ade8589..9541d8a60 100644
--- a/headroom/proxy/forwarded_headers.py
+++ b/headroom/proxy/forwarded_headers.py
@@ -72,7 +72,9 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
__all__ = [
+ "TRUSTED_DASHBOARD_CLIENT_CIDRS_ENV",
"TRUSTED_GATEWAY_CIDRS_ENV",
+ "load_trusted_dashboard_client_cidrs",
"load_trusted_gateway_cidrs",
"peer_is_trusted_gateway",
"resolve_client_ip",
@@ -82,6 +84,7 @@ __all__ = [
#: Environment variable that holds the comma-separated CIDR allow-list.
TRUSTED_GATEWAY_CIDRS_ENV = "HEADROOM_PROXY_TRUSTED_GATEWAY_CIDRS"
+TRUSTED_DASHBOARD_CLIENT_CIDRS_ENV = "HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS"
def _parse_cidr_list(
@@ -115,6 +118,20 @@ def load_trusted_gateway_cidrs(
return _parse_cidr_list(raw)
+def load_trusted_dashboard_client_cidrs(
+ raw: str | None = None,
+) -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]:
+ """Parse the Dashboard client CIDR allow-list from its environment variable."""
+ if raw is None:
+ raw = os.environ.get(TRUSTED_DASHBOARD_CLIENT_CIDRS_ENV, "")
+ try:
+ return _parse_cidr_list(raw)
+ except ValueError as exc:
+ raise ValueError(
+ f"Invalid {TRUSTED_DASHBOARD_CLIENT_CIDRS_ENV} entry: {exc}"
+ ) from exc
+
+
def _normalize_ip(
host: str,
) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
diff --git a/headroom/proxy/loopback_guard.py b/headroom/proxy/loopback_guard.py
index 95ec843ba..21f36ff37 100644
--- a/headroom/proxy/loopback_guard.py
+++ b/headroom/proxy/loopback_guard.py
@@ -50,6 +50,7 @@ except ImportError: # pragma: no cover - fastapi is a hard dep in practice
__all__ = [
"LOOPBACK_HOSTS",
+ "is_ip_literal_host_header",
"is_loopback_host",
"is_loopback_host_header",
"require_loopback",
@@ -128,6 +129,46 @@ def is_loopback_host_header(header_value: str | None) -> bool:
return is_loopback_host(host_part)
+def is_ip_literal_host_header(header_value: str | None) -> bool:
+ """Return whether ``Host:`` contains an IPv4 or bracketed IPv6 literal.
+
+ Dashboard clients may use a non-loopback server address, but retaining an
+ IP-literal Host requirement prevents DNS-rebinding requests from using an
+ attacker-controlled hostname. Ports are accepted in normal HTTP forms.
+ """
+ if not header_value:
+ return False
+
+ candidate = header_value.strip()
+ if not candidate or "/" in candidate or "@" in candidate:
+ return False
+
+ if candidate.startswith("["):
+ closing = candidate.find("]")
+ if closing == -1 or candidate.count("[") != 1 or candidate.count("]") != 1:
+ return False
+ host_part = candidate[1:closing]
+ suffix = candidate[closing + 1 :]
+ if suffix and (not suffix.startswith(":") or not suffix[1:].isdigit()):
+ return False
+ try:
+ return isinstance(ipaddress.ip_address(host_part), ipaddress.IPv6Address)
+ except ValueError:
+ return False
+
+ if candidate.count(":") == 1:
+ host_part, port = candidate.rsplit(":", 1)
+ if not port.isdigit():
+ return False
+ else:
+ host_part = candidate
+
+ try:
+ return isinstance(ipaddress.ip_address(host_part), ipaddress.IPv4Address)
+ except ValueError:
+ return False
+
+
def require_loopback(request: Request) -> None: # type: ignore[valid-type]
"""FastAPI dependency: 404 any non-loopback caller.
diff --git a/headroom/proxy/persistent_metrics.py b/headroom/proxy/persistent_metrics.py
new file mode 100644
index 000000000..2aeab96e7
--- /dev/null
+++ b/headroom/proxy/persistent_metrics.py
@@ -0,0 +1,470 @@
+"""Pure, bounded aggregate state for the durable Dashboard Lifetime view."""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Callable
+from copy import deepcopy
+from datetime import datetime, timezone
+from typing import Any
+
+SCHEMA_VERSION = 5
+MAX_PROVIDER_VALUES = 32
+MAX_STACK_VALUES = 64
+MAX_TRACKED_MODELS = 200
+MAX_EXPOSED_MODELS = 100
+MAX_LABEL_LENGTH = 128
+
+KNOWN_MISS_REASONS = frozenset({"ttl_expiry", "prefix_change", "unknown"})
+KNOWN_WASTE_SIGNALS = frozenset(
+ {
+ "json_noise",
+ "html_noise",
+ "base64",
+ "whitespace",
+ "dynamic_date",
+ "repetition",
+ "reread",
+ "reread_compressed",
+ }
+)
+
+
+def utc_now() -> datetime:
+ """Return the current UTC time without sub-second noise in persisted state."""
+
+ return datetime.now(timezone.utc).replace(microsecond=0)
+
+
+def _to_iso(value: datetime | None) -> str | None:
+ if value is None:
+ return None
+ return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _coerce_int(value: Any) -> int:
+ try:
+ return max(int(value), 0)
+ except (TypeError, ValueError, OverflowError):
+ return 0
+
+
+def _coerce_float(value: Any) -> float:
+ try:
+ result = float(value)
+ except (TypeError, ValueError, OverflowError):
+ return 0.0
+ return result if math.isfinite(result) and result >= 0 else 0.0
+
+
+def _label(value: Any) -> str:
+ if not isinstance(value, str):
+ return "other"
+ value = value.strip()
+ return value[:MAX_LABEL_LENGTH] if value else "other"
+
+
+def _model_entry(raw: Any = None) -> dict[str, Any]:
+ raw = raw if isinstance(raw, dict) else {}
+ return {
+ "requests": _coerce_int(raw.get("requests")),
+ "input_tokens": _coerce_int(raw.get("input_tokens")),
+ "output_tokens": _coerce_int(raw.get("output_tokens")),
+ "attempted_input_tokens": _coerce_int(raw.get("attempted_input_tokens")),
+ "tokens_saved": _coerce_int(raw.get("tokens_saved")),
+ "last_activity_at": raw.get("last_activity_at")
+ if isinstance(raw.get("last_activity_at"), str)
+ else None,
+ }
+
+
+def _empty_state() -> dict[str, Any]:
+ return {
+ "started_at": None,
+ "last_activity_at": None,
+ "full_fidelity_started_at": None,
+ "requests": {
+ "total": 0,
+ "cached": 0,
+ "failed": 0,
+ "rate_limited": 0,
+ "by_provider": {},
+ "by_stack": {},
+ },
+ "tokens": {"input": 0, "output": 0, "attempted_input": 0, "saved": 0},
+ "prefix_cache": {
+ "requests": 0,
+ "hit_requests": 0,
+ "cache_read_tokens": 0,
+ "cache_write_tokens": 0,
+ "cache_write_5m_tokens": 0,
+ "cache_write_1h_tokens": 0,
+ "uncached_input_tokens": 0,
+ "bust_count": 0,
+ "bust_tokens": 0,
+ "misses_by_reason": {},
+ "by_provider": {},
+ },
+ "cost": {"input_usd": 0.0, "compression_savings_usd": 0.0, "cache_savings_usd": 0.0},
+ "waste_signals": {},
+ "models": {"tracked": {}, "other": _model_entry()},
+ "persistence": {"last_saved_at": None},
+ }
+
+
+def _dict_or_empty(value: Any) -> dict[Any, Any]:
+ return value if isinstance(value, dict) else {}
+
+
+class PersistentMetricsState:
+ """In-memory Lifetime aggregate with deterministic, bounded dimensions."""
+
+ def __init__(
+ self,
+ raw: dict[str, Any] | None = None,
+ *,
+ now: Callable[[], datetime] = utc_now,
+ ) -> None:
+ self._now = now
+ self._state = self._normalize(raw)
+ self._compact_models()
+
+ def _normalize(self, raw: dict[str, Any] | None) -> dict[str, Any]:
+ source = raw if isinstance(raw, dict) else {}
+ result = _empty_state()
+ for timestamp_key in ("started_at", "last_activity_at", "full_fidelity_started_at"):
+ value = source.get(timestamp_key)
+ if isinstance(value, str):
+ result[timestamp_key] = value
+
+ raw_requests = _dict_or_empty(source.get("requests"))
+ for key in ("total", "cached", "failed", "rate_limited"):
+ result["requests"][key] = _coerce_int(raw_requests.get(key))
+ result["requests"]["by_provider"] = self._normalize_count_map(
+ raw_requests.get("by_provider"), MAX_PROVIDER_VALUES
+ )
+ result["requests"]["by_stack"] = self._normalize_count_map(
+ raw_requests.get("by_stack"), MAX_STACK_VALUES
+ )
+
+ raw_tokens = _dict_or_empty(source.get("tokens"))
+ for key in ("input", "output", "attempted_input", "saved"):
+ result["tokens"][key] = _coerce_int(raw_tokens.get(key))
+
+ raw_cache = _dict_or_empty(source.get("prefix_cache"))
+ for key in (
+ "requests",
+ "hit_requests",
+ "cache_read_tokens",
+ "cache_write_tokens",
+ "cache_write_5m_tokens",
+ "cache_write_1h_tokens",
+ "uncached_input_tokens",
+ "bust_count",
+ "bust_tokens",
+ ):
+ result["prefix_cache"][key] = _coerce_int(raw_cache.get(key))
+ result["prefix_cache"]["by_provider"] = self._normalize_count_map(
+ raw_cache.get("by_provider"), MAX_PROVIDER_VALUES
+ )
+ result["prefix_cache"]["misses_by_reason"] = self._normalize_enum_map(
+ raw_cache.get("misses_by_reason"), KNOWN_MISS_REASONS
+ )
+
+ raw_cost = _dict_or_empty(source.get("cost"))
+ for key in ("input_usd", "compression_savings_usd", "cache_savings_usd"):
+ result["cost"][key] = round(_coerce_float(raw_cost.get(key)), 6)
+ result["waste_signals"] = self._normalize_enum_map(
+ source.get("waste_signals"), KNOWN_WASTE_SIGNALS
+ )
+
+ raw_models = _dict_or_empty(source.get("models"))
+ raw_tracked = _dict_or_empty(raw_models.get("tracked"))
+ for name, entry in raw_tracked.items():
+ normalized_name = self._model_name(name)
+ if normalized_name == "other":
+ self._merge_model_entry(result["models"]["other"], _model_entry(entry))
+ continue
+ result["models"]["tracked"][normalized_name] = _model_entry(entry)
+ self._merge_model_entry(result["models"]["other"], _model_entry(raw_models.get("other")))
+ raw_persistence = _dict_or_empty(source.get("persistence"))
+ if isinstance(raw_persistence.get("last_saved_at"), str):
+ result["persistence"]["last_saved_at"] = raw_persistence["last_saved_at"]
+ return result
+
+ @staticmethod
+ def _normalize_count_map(raw: Any, limit: int) -> dict[str, int]:
+ result: dict[str, int] = {}
+ if not isinstance(raw, dict):
+ return result
+ for key, value in raw.items():
+ label = _label(key)
+ result[label] = result.get(label, 0) + _coerce_int(value)
+ PersistentMetricsState._compact_count_map(result, limit)
+ return result
+
+ @staticmethod
+ def _normalize_enum_map(raw: Any, allowed: frozenset[str]) -> dict[str, int]:
+ result: dict[str, int] = {}
+ if not isinstance(raw, dict):
+ return result
+ for key, value in raw.items():
+ label = key if isinstance(key, str) and key in allowed else "unknown"
+ result[label] = result.get(label, 0) + _coerce_int(value)
+ return result
+
+ @staticmethod
+ def _compact_count_map(values: dict[str, int], limit: int) -> None:
+ named = [key for key in values if key != "other"]
+ while len(named) > limit:
+ evicted = min(named, key=lambda key: (values[key], key))
+ values["other"] = values.get("other", 0) + values.pop(evicted)
+ named.remove(evicted)
+
+ def _record_activity(self) -> str:
+ timestamp = _to_iso(self._now())
+ if self._state["started_at"] is None:
+ self._state["started_at"] = timestamp
+ if self._state["full_fidelity_started_at"] is None:
+ self._state["full_fidelity_started_at"] = timestamp
+ self._state["last_activity_at"] = timestamp
+ return timestamp or ""
+
+ @staticmethod
+ def _increment_count(values: dict[str, int], label: str, limit: int) -> None:
+ values[label] = values.get(label, 0) + 1
+ PersistentMetricsState._compact_count_map(values, limit)
+
+ @staticmethod
+ def _model_name(value: Any) -> str:
+ """Never retain the legacy unknown model bucket as a named model."""
+
+ label = _label(value)
+ return "other" if label.lower() == "unknown" else label
+
+ @staticmethod
+ def _merge_model_entry(destination: dict[str, Any], source: dict[str, Any]) -> None:
+ for key in (
+ "requests",
+ "input_tokens",
+ "output_tokens",
+ "attempted_input_tokens",
+ "tokens_saved",
+ ):
+ destination[key] += _coerce_int(source.get(key))
+ if destination["last_activity_at"] is None or (
+ source["last_activity_at"] is not None
+ and source["last_activity_at"] > destination["last_activity_at"]
+ ):
+ destination["last_activity_at"] = source["last_activity_at"]
+
+ @staticmethod
+ def _model_rank(item: tuple[str, dict[str, Any]]) -> tuple[int, str, str]:
+ name, entry = item
+ observed_tokens = entry["input_tokens"] + entry["output_tokens"]
+ return (-observed_tokens, entry["last_activity_at"] or "", name)
+
+ def _compact_models(self) -> None:
+ tracked = self._state["models"]["tracked"]
+ if len(tracked) <= MAX_TRACKED_MODELS:
+ return
+ ranked = sorted(tracked.items(), key=self._model_rank)
+ kept = dict(ranked[:MAX_EXPOSED_MODELS])
+ other = self._state["models"]["other"]
+ for _, entry in ranked[MAX_EXPOSED_MODELS:]:
+ self._merge_model_entry(other, entry)
+ self._state["models"]["tracked"] = kept
+
+ def _record_model(
+ self,
+ *,
+ model: str | None,
+ timestamp: str,
+ input_tokens: int,
+ output_tokens: int,
+ attempted_input_tokens: int,
+ tokens_saved: int,
+ ) -> None:
+ name = self._model_name(model)
+ models = self._state["models"]
+ entry = (
+ models["other"]
+ if name == "other"
+ else models["tracked"].setdefault(name, _model_entry())
+ )
+ entry["requests"] += 1
+ entry["input_tokens"] += input_tokens
+ entry["output_tokens"] += output_tokens
+ entry["attempted_input_tokens"] += attempted_input_tokens
+ entry["tokens_saved"] += tokens_saved
+ entry["last_activity_at"] = timestamp
+ self._compact_models()
+
+ def record_request(
+ self,
+ *,
+ provider: str | None,
+ stack: str | None,
+ model: str | None,
+ input_tokens: Any = 0,
+ output_tokens: Any = 0,
+ attempted_input_tokens: Any = 0,
+ tokens_saved: Any = 0,
+ cached: bool = False,
+ record_stack: bool = True,
+ cache_read_tokens: Any = 0,
+ cache_write_tokens: Any = 0,
+ cache_write_5m_tokens: Any = 0,
+ cache_write_1h_tokens: Any = 0,
+ uncached_input_tokens: Any = 0,
+ input_usd: Any = 0.0,
+ compression_savings_usd: Any = 0.0,
+ cache_savings_usd: Any = 0.0,
+ waste_signals: dict[str, Any] | None = None,
+ ) -> None:
+ """Accumulate one completed top-level request after coercing all deltas."""
+
+ timestamp = self._record_activity()
+ input_delta = _coerce_int(input_tokens)
+ output_delta = _coerce_int(output_tokens)
+ attempted_delta = _coerce_int(attempted_input_tokens)
+ saved_delta = _coerce_int(tokens_saved)
+ provider_label = _label(provider)
+ stack_label = _label(stack)
+
+ requests = self._state["requests"]
+ requests["total"] += 1
+ requests["cached"] += int(bool(cached))
+ self._increment_count(requests["by_provider"], provider_label, MAX_PROVIDER_VALUES)
+ if record_stack:
+ self._increment_count(requests["by_stack"], stack_label, MAX_STACK_VALUES)
+
+ tokens = self._state["tokens"]
+ tokens["input"] += input_delta
+ tokens["output"] += output_delta
+ tokens["attempted_input"] += attempted_delta
+ tokens["saved"] += saved_delta
+
+ cache = self._state["prefix_cache"]
+ cache["requests"] += 1
+ cache["hit_requests"] += int(bool(cached))
+ cache["cache_read_tokens"] += _coerce_int(cache_read_tokens)
+ cache["cache_write_tokens"] += _coerce_int(cache_write_tokens)
+ cache["cache_write_5m_tokens"] += _coerce_int(cache_write_5m_tokens)
+ cache["cache_write_1h_tokens"] += _coerce_int(cache_write_1h_tokens)
+ cache["uncached_input_tokens"] += _coerce_int(uncached_input_tokens)
+ self._increment_count(cache["by_provider"], provider_label, MAX_PROVIDER_VALUES)
+
+ cost = self._state["cost"]
+ cost["input_usd"] = round(cost["input_usd"] + _coerce_float(input_usd), 6)
+ cost["compression_savings_usd"] = round(
+ cost["compression_savings_usd"] + _coerce_float(compression_savings_usd), 6
+ )
+ cost["cache_savings_usd"] = round(
+ cost["cache_savings_usd"] + _coerce_float(cache_savings_usd), 6
+ )
+
+ if isinstance(waste_signals, dict):
+ for name, token_count in waste_signals.items():
+ bucket = name if isinstance(name, str) and name in KNOWN_WASTE_SIGNALS else "other"
+ self._state["waste_signals"][bucket] = self._state["waste_signals"].get(
+ bucket, 0
+ ) + _coerce_int(token_count)
+
+ self._record_model(
+ model=model,
+ timestamp=timestamp,
+ input_tokens=input_delta,
+ output_tokens=output_delta,
+ attempted_input_tokens=attempted_delta,
+ tokens_saved=saved_delta,
+ )
+
+ def record_stack(self, stack: str | None) -> None:
+ """Accumulate the existing inbound stack label without adding a request."""
+
+ if stack is None:
+ return
+ self._record_activity()
+ self._increment_count(self._state["requests"]["by_stack"], _label(stack), MAX_STACK_VALUES)
+
+ def record_failed(self, *, provider: str | None = None, model: str | None = None) -> None:
+ """Record a failed request without changing the completed-request denominator."""
+
+ self._record_activity()
+ self._state["requests"]["failed"] += 1
+
+ def record_rate_limited(self, *, provider: str | None = None, model: str | None = None) -> None:
+ """Record a rate-limited request without redefining total request semantics."""
+
+ self._record_activity()
+ self._state["requests"]["rate_limited"] += 1
+
+ def record_cache_bust(self, *, tokens_lost: Any = 0) -> None:
+ self._record_activity()
+ cache = self._state["prefix_cache"]
+ cache["bust_count"] += 1
+ cache["bust_tokens"] += _coerce_int(tokens_lost)
+
+ def record_cache_miss(self, *, provider: str | None, reason: str | None) -> None:
+ self._record_activity()
+ bucket = reason if isinstance(reason, str) and reason in KNOWN_MISS_REASONS else "unknown"
+ misses = self._state["prefix_cache"]["misses_by_reason"]
+ misses[bucket] = misses.get(bucket, 0) + 1
+
+ def set_last_saved_at(self, value: str | None) -> None:
+ self._state["persistence"]["last_saved_at"] = value
+
+ def to_dict(self) -> dict[str, Any]:
+ """Return the persisted form without derived values or I/O metadata."""
+
+ return deepcopy(self._state)
+
+ @staticmethod
+ def _percent(numerator: int, denominator: int) -> float | None:
+ if denominator <= 0:
+ return None
+ return round(numerator / denominator * 100, 6)
+
+ def _by_model_snapshot(self) -> dict[str, dict[str, Any]]:
+ ranked = sorted(self._state["models"]["tracked"].items(), key=self._model_rank)
+ visible = ranked[:MAX_EXPOSED_MODELS]
+ other = _model_entry(self._state["models"]["other"])
+ for _, entry in ranked[MAX_EXPOSED_MODELS:]:
+ self._merge_model_entry(other, entry)
+ result = {name: deepcopy(entry) for name, entry in visible}
+ result["other"] = other
+ return result
+
+ def snapshot(self, *, persistence: dict[str, Any]) -> dict[str, Any]:
+ """Return an API-safe aggregate and derive all percentages at read time."""
+
+ cache = self._state["prefix_cache"]
+ cache_write_total = cache["cache_write_1h_tokens"] + cache["cache_write_5m_tokens"]
+ tokens = self._state["tokens"]
+ return {
+ "scope": "lifetime",
+ "schema_version": SCHEMA_VERSION,
+ "generated_at": _to_iso(self._now()),
+ "started_at": self._state["started_at"],
+ "last_activity_at": self._state["last_activity_at"],
+ "full_fidelity_started_at": self._state["full_fidelity_started_at"],
+ "requests": deepcopy(self._state["requests"]),
+ "tokens": {
+ **deepcopy(tokens),
+ "token_savings_percent": self._percent(tokens["saved"], tokens["attempted_input"]),
+ },
+ "prefix_cache": {
+ **deepcopy(cache),
+ "cache_hit_rate": self._percent(cache["hit_requests"], cache["requests"]),
+ "ttl_1h_percent": self._percent(cache["cache_write_1h_tokens"], cache_write_total),
+ "ttl_5m_percent": self._percent(cache["cache_write_5m_tokens"], cache_write_total),
+ },
+ "cost": deepcopy(self._state["cost"]),
+ "waste_signals": deepcopy(self._state["waste_signals"]),
+ "by_model": self._by_model_snapshot(),
+ "persistence": {
+ **deepcopy(persistence),
+ "last_saved_at": self._state["persistence"]["last_saved_at"],
+ },
+ }
diff --git a/headroom/proxy/prometheus_metrics.py b/headroom/proxy/prometheus_metrics.py
index 56e20e0ab..1aa704b2e 100644
--- a/headroom/proxy/prometheus_metrics.py
+++ b/headroom/proxy/prometheus_metrics.py
@@ -446,6 +446,7 @@ class PrometheusMetrics:
):
return
self.requests_by_stack[slug] += 1
+ self.savings_tracker.record_lifetime_stack(slug)
def record_compression(
self,
@@ -732,6 +733,24 @@ class PrometheusMetrics:
if len(self.savings_history) > 500:
self.savings_history = self.savings_history[-500:]
+ self.savings_tracker.record_lifetime_request(
+ persist=False,
+ provider=provider,
+ stack=None,
+ record_stack=False,
+ model=model,
+ input_tokens=input_tokens,
+ output_tokens=output_tokens,
+ attempted_input_tokens=attempted_input_tokens,
+ tokens_saved=tokens_saved,
+ cached=cached,
+ 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,
+ waste_signals=waste_signals,
+ )
total_input_tokens, total_input_cost_usd = self._current_savings_tracker_totals()
self.savings_tracker.record_request(
model=model,
@@ -828,6 +847,7 @@ class PrometheusMetrics:
async with self._lock:
self.cache_bust_tokens_lost += tokens_lost
self.cache_bust_count += 1
+ self.savings_tracker.record_lifetime_cache_bust(tokens_lost=tokens_lost)
self._get_otel_metrics().record_proxy_cache_bust(tokens_lost=tokens_lost)
async def record_cache_miss_attribution(self, provider: str, reason: str) -> None:
@@ -841,6 +861,7 @@ class PrometheusMetrics:
"""
async with self._lock:
self.cache_miss_attribution_by_provider[provider][reason] += 1
+ self.savings_tracker.record_lifetime_cache_miss(provider=provider, reason=reason)
# ------------------------------------------------------------------
# Unit 3: WS session lifecycle gauges / histogram
@@ -886,11 +907,13 @@ class PrometheusMetrics:
async def record_rate_limited(self, *, provider: str | None = None, model: str | None = None):
async with self._lock:
self.requests_rate_limited += 1
+ self.savings_tracker.record_lifetime_rate_limited(provider=provider, model=model)
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.savings_tracker.record_lifetime_failed(provider=provider, model=model)
self._get_otel_metrics().record_proxy_failed(provider=provider, model=model)
async def export(self) -> str:
diff --git a/headroom/proxy/savings_tracker.py b/headroom/proxy/savings_tracker.py
index 725aafeb8..85698d5e5 100644
--- a/headroom/proxy/savings_tracker.py
+++ b/headroom/proxy/savings_tracker.py
@@ -22,6 +22,7 @@ from typing import Any
from headroom import paths as _paths
from headroom.proxy import project_name_policy
+from headroom.proxy.persistent_metrics import PersistentMetricsState
PROJECT_NAME_MAX_LENGTH = project_name_policy.PROJECT_NAME_MAX_LENGTH
sanitize_project_name = project_name_policy.sanitize_project_name
@@ -31,7 +32,7 @@ logger = logging.getLogger(__name__)
HEADROOM_SAVINGS_PATH_ENV_VAR = _paths.HEADROOM_SAVINGS_PATH_ENV
DEFAULT_SAVINGS_DIR = ".headroom"
DEFAULT_SAVINGS_FILE = "proxy_savings.json"
-SCHEMA_VERSION = 4
+SCHEMA_VERSION = 5
DEFAULT_MAX_HISTORY_POINTS = 5000
DEFAULT_MAX_PROJECTS = 50
DEFAULT_MAX_HISTORY_AGE_DAYS = 365
@@ -546,7 +547,13 @@ class SavingsTracker:
self._save_flush_every = max(_coerce_int(save_flush_every, 1), 1)
self._since_save = 0
self._lock = threading.Lock()
+ self._persistence_healthy = True
+ self._persistence_error: str | None = None
+ self._needs_schema_save = False
self._state = self._load_state()
+ self._persistent_metrics = PersistentMetricsState(
+ self._state.pop("lifetime_metrics", None)
+ )
@property
def storage_path(self) -> str:
@@ -784,6 +791,66 @@ class SavingsTracker:
self._maybe_save_locked()
return True
+ def record_lifetime_request(self, *, persist: bool = True, **metrics: Any) -> None:
+ """Record one completed request in the durable Lifetime aggregate."""
+
+ model = _normalize_model(metrics.get("model"))
+ input_tokens = _coerce_int(metrics.get("input_tokens"))
+ cache_read_tokens = _coerce_int(metrics.get("cache_read_tokens"))
+ cache_write_tokens = _coerce_int(metrics.get("cache_write_tokens"))
+ uncached_input_tokens = _coerce_int(metrics.get("uncached_input_tokens"))
+ metrics.setdefault(
+ "input_usd",
+ _estimate_input_cost_usd(
+ model,
+ input_tokens,
+ cache_read_tokens=cache_read_tokens,
+ cache_write_tokens=cache_write_tokens,
+ uncached_input_tokens=uncached_input_tokens,
+ ),
+ )
+ metrics.setdefault(
+ "compression_savings_usd",
+ _estimate_compression_savings_usd(model, _coerce_int(metrics.get("tokens_saved"))),
+ )
+ metrics.setdefault("cache_savings_usd", _estimate_cache_savings_usd(model, cache_read_tokens))
+ with self._lock:
+ self._persistent_metrics.record_request(**metrics)
+ if persist:
+ self._maybe_save_locked()
+
+ def record_lifetime_stack(self, stack: str | None) -> None:
+ """Mirror the existing inbound stack dimension without another save."""
+
+ with self._lock:
+ self._persistent_metrics.record_stack(stack)
+
+ def record_lifetime_failed(self, *, provider: str | None = None, model: str | None = None) -> None:
+ """Record a failed proxy request without changing legacy history."""
+
+ with self._lock:
+ self._persistent_metrics.record_failed(provider=provider, model=model)
+ self._maybe_save_locked()
+
+ def record_lifetime_rate_limited(
+ self, *, provider: str | None = None, model: str | None = None
+ ) -> None:
+ """Record a rate-limited proxy request without changing legacy history."""
+
+ with self._lock:
+ self._persistent_metrics.record_rate_limited(provider=provider, model=model)
+ self._maybe_save_locked()
+
+ def record_lifetime_cache_bust(self, *, tokens_lost: int) -> None:
+ with self._lock:
+ self._persistent_metrics.record_cache_bust(tokens_lost=tokens_lost)
+ self._maybe_save_locked()
+
+ def record_lifetime_cache_miss(self, *, provider: str | None, reason: str | None) -> None:
+ with self._lock:
+ self._persistent_metrics.record_cache_miss(provider=provider, reason=reason)
+ self._maybe_save_locked()
+
def _record_project_locked(
self,
project: str | None,
@@ -891,6 +958,24 @@ class SavingsTracker:
)
result[model] = view
return result
+ def lifetime_response(self) -> dict[str, Any]:
+ """Return the durable aggregate used only by ``/stats-lifetime``."""
+
+ with self._lock:
+ response = self._persistent_metrics.snapshot(
+ persistence={
+ "enabled": not self._stateless,
+ "healthy": self._persistence_healthy,
+ "error": (
+ "Lifetime metrics unavailable in stateless mode"
+ if self._stateless
+ else self._persistence_error
+ ),
+ "pending_records": 0 if self._stateless else self._since_save,
+ }
+ )
+ response["projects"] = self._projects_snapshot_locked()
+ return response
def stats_preview(self, recent_points: int = 20) -> dict[str, Any]:
"""Return a compact preview for `/stats`."""
@@ -1031,10 +1116,30 @@ class SavingsTracker:
raw = json.load(f)
except (json.JSONDecodeError, OSError) as e:
logger.warning("Failed to load savings history from %s: %s", self._path, e)
+ self._persistence_healthy = False
+ self._persistence_error = str(e)
+ self._preserve_corrupt_file()
return self._default_state()
+ if not isinstance(raw, dict):
+ self._persistence_healthy = False
+ self._persistence_error = "Savings state root must be a JSON object"
+ self._preserve_corrupt_file()
+ return self._default_state()
+ if _coerce_int(raw.get("schema_version")) != SCHEMA_VERSION:
+ self._needs_schema_save = True
return self._sanitize_state(raw)
+ def _preserve_corrupt_file(self) -> None:
+ """Best-effort retention of unreadable state before starting fresh."""
+
+ timestamp = _to_utc_iso(_utc_now()).replace(":", "-")
+ corrupt_path = self._path.with_name(f"{self._path.name}.corrupt-{timestamp}")
+ try:
+ self._path.replace(corrupt_path)
+ except OSError:
+ logger.warning("Could not preserve corrupt savings history at %s", self._path)
+
def _sanitize_state(self, raw: Any) -> dict[str, Any]:
if not isinstance(raw, dict):
return self._default_state()
@@ -1101,6 +1206,12 @@ class SavingsTracker:
"projects": _normalize_projects(raw.get("projects")),
"by_model": _normalize_by_model(raw.get("by_model")),
}
+ raw_lifetime_metrics = raw.get("lifetime_metrics")
+ if isinstance(raw_lifetime_metrics, dict):
+ state["lifetime_metrics"] = raw_lifetime_metrics
+ else:
+ state["lifetime_metrics"] = self._migrate_v4_lifetime_metrics(state)
+ self._needs_schema_save = True
if normalized_history:
reference_time = _parse_timestamp(normalized_history[-1]["timestamp"]) or _utc_now()
@@ -1115,6 +1226,49 @@ class SavingsTracker:
return state
+ def _migrate_v4_lifetime_metrics(self, state: dict[str, Any]) -> dict[str, Any]:
+ """Seed the v5 aggregate from the best information available in v4."""
+
+ history = state["history"]
+ display_session = state["display_session"]
+ started_at = (
+ history[0]["timestamp"]
+ if history
+ else display_session.get("started_at") or _to_utc_iso(_utc_now())
+ )
+ last_activity_at = (
+ history[-1]["timestamp"] if history else display_session.get("last_activity_at")
+ )
+ legacy = state["lifetime"]
+ return {
+ "started_at": started_at,
+ "last_activity_at": last_activity_at,
+ "full_fidelity_started_at": _to_utc_iso(_utc_now()),
+ "requests": {"total": legacy["requests"]},
+ "tokens": {
+ "input": legacy["total_input_tokens"],
+ "attempted_input": legacy["total_input_tokens"] + legacy["tokens_saved"],
+ "saved": legacy["tokens_saved"],
+ },
+ "prefix_cache": {"cache_read_tokens": legacy["cache_read_tokens"]},
+ "cost": {
+ "input_usd": legacy["total_input_cost_usd"],
+ "compression_savings_usd": legacy["compression_savings_usd"],
+ "cache_savings_usd": legacy["cache_savings_usd"],
+ },
+ "models": {
+ "tracked": {},
+ "other": {
+ "requests": legacy["requests"],
+ "input_tokens": legacy["total_input_tokens"],
+ "attempted_input_tokens": legacy["total_input_tokens"]
+ + legacy["tokens_saved"],
+ "tokens_saved": legacy["tokens_saved"],
+ "last_activity_at": last_activity_at,
+ },
+ },
+ }
+
def _trim_history_locked(self, reference_time: datetime | None = None) -> None:
history = self._state["history"]
if not history:
@@ -1190,7 +1344,7 @@ class SavingsTracker:
recent requests. No-op when nothing is buffered.
"""
with self._lock:
- if self._since_save > 0:
+ if self._since_save > 0 or self._needs_schema_save:
self._save_locked()
def _maybe_save_locked(self) -> None:
@@ -1206,9 +1360,13 @@ class SavingsTracker:
if self._stateless:
# Stateless mode: live counters stay in memory; nothing is persisted.
self._since_save = 0
+ self._needs_schema_save = False
return
try:
self._path.parent.mkdir(parents=True, exist_ok=True)
+ saved_at = _to_utc_iso(_utc_now())
+ lifetime_metrics = self._persistent_metrics.to_dict()
+ lifetime_metrics["persistence"]["last_saved_at"] = saved_at
payload = {
"schema_version": SCHEMA_VERSION,
"lifetime": self._state["lifetime"],
@@ -1216,6 +1374,7 @@ class SavingsTracker:
"history": self._state["history"],
"projects": self._state.get("projects", {}),
"by_model": self._state.get("by_model", {}),
+ "lifetime_metrics": lifetime_metrics,
}
json_data = json.dumps(payload, indent=2)
@@ -1255,9 +1414,15 @@ class SavingsTracker:
# Reset only after a durable write. A failed save leaves the counter
# untouched so the next record retries instead of waiting a full window.
+ self._persistent_metrics.set_last_saved_at(saved_at)
+ self._persistence_healthy = True
+ self._persistence_error = None
+ self._needs_schema_save = False
self._since_save = 0
except OSError as e:
logger.warning("Failed to save savings history to %s: %s", self._path, e)
+ self._persistence_healthy = False
+ self._persistence_error = str(e)
def _display_session_snapshot_locked(
self,
diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py
index 454d6c3cc..7131251cc 100644
--- a/headroom/proxy/server.py
+++ b/headroom/proxy/server.py
@@ -28,6 +28,7 @@ import asyncio
import concurrent.futures
import contextlib
import hmac
+import ipaddress
import json
import logging
import math
@@ -40,6 +41,7 @@ from dataclasses import fields, is_dataclass, replace
from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast
+from urllib.parse import urlsplit
if TYPE_CHECKING:
from ..backends.base import Backend
@@ -2036,6 +2038,77 @@ def _request_is_loopback(request: Request) -> bool:
return peer_is_trusted_gateway(client_host, load_trusted_gateway_cidrs())
+def _request_can_view_dashboard_metadata(
+ request: Request,
+ trusted_dashboard_client_cidrs: tuple[
+ ipaddress.IPv4Network | ipaddress.IPv6Network, ...
+ ],
+) -> bool:
+ """Authorize sensitive ``/stats`` metadata without widening admin access."""
+ if _request_is_loopback(request):
+ return True
+
+ from headroom.proxy.forwarded_headers import peer_is_trusted_gateway, resolve_client_ip
+ from headroom.proxy.loopback_guard import is_ip_literal_host_header
+
+ try:
+ host_header = request.headers.get("host")
+ except AttributeError:
+ return False
+ if not is_ip_literal_host_header(host_header):
+ return False
+
+ # CIDR authorization makes this endpoint usable by a remote dashboard, but
+ # it must not let an unrelated site read sensitive metadata through a
+ # victim's browser. Native CLI clients usually send neither header, so
+ # absence remains valid. If either browser provenance header is present,
+ # require it to identify this exact scheme/host/port.
+ if not _request_has_same_origin_or_no_provenance(request, host_header):
+ return False
+
+ return peer_is_trusted_gateway(
+ resolve_client_ip(request),
+ trusted_dashboard_client_cidrs,
+ )
+
+
+def _request_has_same_origin_or_no_provenance(
+ request: Request, host_header: str
+) -> bool:
+ """Accept no browser provenance, otherwise require same-origin headers."""
+
+ from headroom.proxy.forwarded_headers import trusted_forwarded_headers
+
+ forwarded_proto = trusted_forwarded_headers(request)["proto"]
+ request_scheme = forwarded_proto or request.url.scheme
+ expected_origin = _normalized_http_origin(f"{request_scheme}://{host_header}")
+ if expected_origin is None:
+ return False
+
+ for header_name in ("origin", "referer"):
+ header_value = request.headers.get(header_name)
+ if header_value and _normalized_http_origin(header_value) != expected_origin:
+ return False
+ return True
+
+
+def _normalized_http_origin(value: str) -> tuple[str, str, int] | None:
+ """Return a normalized HTTP(S) origin tuple."""
+
+ try:
+ parsed = urlsplit(value.strip())
+ port = parsed.port
+ except ValueError:
+ return None
+
+ scheme = parsed.scheme.lower()
+ if scheme not in {"http", "https"} or not parsed.hostname:
+ return None
+ if port is None:
+ port = 80 if scheme == "http" else 443
+ return scheme, parsed.hostname.lower(), port
+
+
_is_known_websocket_callback_failure = is_known_websocket_callback_failure
@@ -2047,6 +2120,11 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
if not FASTAPI_AVAILABLE:
raise ImportError("FastAPI required. Install: pip install fastapi uvicorn httpx")
+ from headroom.proxy.forwarded_headers import load_trusted_dashboard_client_cidrs
+
+ # Parse once at startup so invalid operator configuration fails loudly.
+ trusted_dashboard_client_cidrs = load_trusted_dashboard_client_cidrs()
+
from contextlib import asynccontextmanager
# Always-on file logging to ~/.headroom/logs/ for `headroom perf` analysis.
@@ -2993,7 +3071,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
def _build_recent_request_payload(limit: int = RECENT_REQUEST_LOG_WINDOW) -> dict[str, Any]:
recent_request_logs = proxy.logger.get_recent(limit) if proxy.logger else []
dashboard_recent_requests = []
- for log in recent_request_logs:
+ for log in reversed(recent_request_logs):
token_accounting_status = _recent_request_token_accounting_status(log)
dashboard_recent_requests.append(
{
@@ -3021,7 +3099,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
"tool_schema_saved_tokens": _tool_schema_saved_from_tags(log.get("tags")),
}
)
- dashboard_recent_requests = dashboard_recent_requests[-10:]
+ dashboard_recent_requests = dashboard_recent_requests[:25]
return {
"request_logs": recent_request_logs[-10:],
"recent_requests": dashboard_recent_requests,
@@ -3659,7 +3737,10 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
only for loopback callers — the local dashboard. Network callers still
get the aggregate counters but never the per-request metadata.
"""
- include_sensitive = _request_is_loopback(request)
+ include_sensitive = _request_can_view_dashboard_metadata(
+ request,
+ trusted_dashboard_client_cidrs,
+ )
if cached:
payload = dict(await _get_cached_stats_payload())
if include_sensitive:
@@ -3676,6 +3757,21 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
payload.pop("request_logs", None)
return payload
+ @app.get("/stats-lifetime")
+ async def stats_lifetime(request: Request):
+ """Return persisted lifetime aggregates with sensitive fields gated."""
+ payload = dict(proxy.metrics.savings_tracker.lifetime_response())
+ include_sensitive = _request_can_view_dashboard_metadata(
+ request,
+ trusted_dashboard_client_cidrs,
+ )
+ if not include_sensitive:
+ payload.pop("projects", None)
+ persistence = payload.get("persistence")
+ if isinstance(persistence, dict):
+ payload["persistence"] = {**persistence, "error": None}
+ return payload
+
@app.post("/stats/reset", dependencies=[Depends(_require_loopback)])
async def stats_reset():
"""Reset in-memory proxy stats for local test/debug isolation."""
diff --git a/tests/test_forwarded_headers.py b/tests/test_forwarded_headers.py
index 006378e80..19d9cff5c 100644
--- a/tests/test_forwarded_headers.py
+++ b/tests/test_forwarded_headers.py
@@ -19,7 +19,9 @@ from fastapi.testclient import TestClient
from starlette.datastructures import Headers, State
from headroom.proxy.forwarded_headers import (
+ TRUSTED_DASHBOARD_CLIENT_CIDRS_ENV,
TRUSTED_GATEWAY_CIDRS_ENV,
+ load_trusted_dashboard_client_cidrs,
load_trusted_gateway_cidrs,
peer_is_trusted_gateway,
resolve_client_ip,
@@ -115,6 +117,36 @@ def test_load_cidrs_reads_env_by_default(monkeypatch: pytest.MonkeyPatch) -> Non
assert [str(c) for c in cidrs] == ["10.0.0.0/8"]
+def test_load_dashboard_client_cidrs_uses_their_own_env(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.setenv(
+ TRUSTED_DASHBOARD_CLIENT_CIDRS_ENV,
+ "100.90.0.5/32, fd7a:115c:a1e0::/48",
+ )
+
+ cidrs = load_trusted_dashboard_client_cidrs()
+
+ assert [str(cidr) for cidr in cidrs] == [
+ "100.90.0.5/32",
+ "fd7a:115c:a1e0::/48",
+ ]
+
+
+def test_load_dashboard_client_cidrs_unset_is_empty(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.delenv(TRUSTED_DASHBOARD_CLIENT_CIDRS_ENV, raising=False)
+ assert load_trusted_dashboard_client_cidrs() == ()
+
+
+def test_load_dashboard_client_cidrs_empty_and_malformed_values() -> None:
+ assert load_trusted_dashboard_client_cidrs("") == ()
+ assert load_trusted_dashboard_client_cidrs(" ") == ()
+ with pytest.raises(ValueError, match=TRUSTED_DASHBOARD_CLIENT_CIDRS_ENV):
+ load_trusted_dashboard_client_cidrs("100.90.0.5/32,not-a-cidr")
+
+
# ──────────────────────────────────────────────────────────────────
# Membership check (peer_is_trusted_gateway)
# ──────────────────────────────────────────────────────────────────
diff --git a/tests/test_persistent_metrics.py b/tests/test_persistent_metrics.py
new file mode 100644
index 000000000..23a1d45ff
--- /dev/null
+++ b/tests/test_persistent_metrics.py
@@ -0,0 +1,155 @@
+"""Tests for durable, aggregate-only proxy Lifetime metrics."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+
+from headroom.proxy.persistent_metrics import PersistentMetricsState
+
+FIXED_NOW = datetime(2026, 7, 14, 8, 30, tzinfo=timezone.utc)
+
+
+def _new_state() -> PersistentMetricsState:
+ return PersistentMetricsState(now=lambda: FIXED_NOW)
+
+
+def test_snapshot_accumulates_request_token_cache_cost_and_waste_metrics() -> None:
+ state = _new_state()
+
+ state.record_request(
+ provider="anthropic",
+ stack="codex",
+ model="claude-test",
+ input_tokens=100,
+ output_tokens=20,
+ attempted_input_tokens=150,
+ tokens_saved=50,
+ cached=True,
+ cache_read_tokens=80,
+ cache_write_tokens=40,
+ cache_write_5m_tokens=10,
+ cache_write_1h_tokens=30,
+ uncached_input_tokens=20,
+ input_usd=0.4,
+ compression_savings_usd=0.2,
+ cache_savings_usd=0.1,
+ waste_signals={"repetition": 7},
+ )
+ state.record_failed(provider="anthropic", model="claude-test")
+ state.record_rate_limited(provider="anthropic", model="claude-test")
+ state.record_cache_bust(tokens_lost=9)
+ state.record_cache_miss(provider="anthropic", reason="prefix_change")
+
+ snapshot = state.snapshot(persistence={"enabled": True, "healthy": True})
+
+ assert snapshot["scope"] == "lifetime"
+ assert snapshot["requests"] == {
+ "total": 1,
+ "cached": 1,
+ "failed": 1,
+ "rate_limited": 1,
+ "by_provider": {"anthropic": 1},
+ "by_stack": {"codex": 1},
+ }
+ assert snapshot["tokens"] == {
+ "input": 100,
+ "output": 20,
+ "attempted_input": 150,
+ "saved": 50,
+ "token_savings_percent": pytest.approx(50 / 150 * 100),
+ }
+ assert snapshot["prefix_cache"]["requests"] == 1
+ assert snapshot["prefix_cache"]["hit_requests"] == 1
+ assert snapshot["prefix_cache"]["cache_read_tokens"] == 80
+ assert snapshot["prefix_cache"]["cache_write_tokens"] == 40
+ assert snapshot["prefix_cache"]["cache_hit_rate"] == 100.0
+ assert snapshot["prefix_cache"]["ttl_1h_percent"] == 75.0
+ assert snapshot["prefix_cache"]["ttl_5m_percent"] == 25.0
+ assert snapshot["prefix_cache"]["bust_count"] == 1
+ assert snapshot["prefix_cache"]["bust_tokens"] == 9
+ assert snapshot["prefix_cache"]["misses_by_reason"] == {"prefix_change": 1}
+ assert snapshot["cost"] == {
+ "input_usd": 0.4,
+ "compression_savings_usd": 0.2,
+ "cache_savings_usd": 0.1,
+ }
+ assert snapshot["waste_signals"] == {"repetition": 7}
+ assert snapshot["by_model"]["claude-test"]["input_tokens"] == 100
+
+
+def test_snapshot_uses_null_for_ratios_without_a_denominator() -> None:
+ snapshot = _new_state().snapshot(persistence={"enabled": True, "healthy": True})
+
+ assert snapshot["tokens"]["token_savings_percent"] is None
+ assert snapshot["prefix_cache"]["cache_hit_rate"] is None
+ assert snapshot["prefix_cache"]["ttl_1h_percent"] is None
+ assert snapshot["prefix_cache"]["ttl_5m_percent"] is None
+
+
+def test_candidate_models_remain_available_until_the_two_hundred_and_first_model() -> None:
+ state = _new_state()
+ for index in range(200):
+ state.record_request(
+ provider="provider",
+ stack="stack",
+ model=f"model-{index:03}",
+ input_tokens=index + 1,
+ )
+
+ persisted = state.to_dict()
+ snapshot = state.snapshot(persistence={"enabled": True, "healthy": True})
+
+ assert len(persisted["models"]["tracked"]) == 200
+ assert "model-000" not in snapshot["by_model"]
+ assert snapshot["by_model"]["other"]["input_tokens"] == sum(range(1, 101))
+
+
+def test_two_hundred_and_first_model_permanently_compacts_non_top_candidates() -> None:
+ state = _new_state()
+ for index in range(201):
+ state.record_request(
+ provider="provider",
+ stack="stack",
+ model=f"model-{index:03}",
+ input_tokens=index + 1,
+ )
+
+ persisted = state.to_dict()
+ snapshot = state.snapshot(persistence={"enabled": True, "healthy": True})
+
+ assert len(persisted["models"]["tracked"]) == 100
+ assert set(snapshot["by_model"]) == {
+ *(f"model-{index:03}" for index in range(101, 201)),
+ "other",
+ }
+ assert snapshot["by_model"]["other"]["input_tokens"] == sum(range(1, 102))
+
+
+def test_state_normalizes_invalid_values_and_unknown_dimension_labels() -> None:
+ state = PersistentMetricsState(
+ {
+ "requests": {"total": "not-a-number"},
+ "tokens": {"input": float("nan"), "output": -3},
+ "models": {"tracked": {"unknown": {"input_tokens": "7"}}},
+ },
+ now=lambda: FIXED_NOW,
+ )
+ state.record_request(
+ provider=" ",
+ stack=None,
+ model=" ",
+ input_tokens=-1,
+ output_tokens=float("inf"),
+ waste_signals={"unrecognized": 9},
+ )
+
+ snapshot = state.snapshot(persistence={"enabled": True, "healthy": True})
+
+ assert snapshot["tokens"]["input"] == 0
+ assert snapshot["tokens"]["output"] == 0
+ assert snapshot["requests"]["by_provider"] == {"other": 1}
+ assert snapshot["requests"]["by_stack"] == {"other": 1}
+ assert snapshot["by_model"]["other"]["input_tokens"] == 7
+ assert snapshot["waste_signals"] == {"other": 9}
diff --git a/tests/test_persistent_metrics_integration.py b/tests/test_persistent_metrics_integration.py
new file mode 100644
index 000000000..fdb9f2406
--- /dev/null
+++ b/tests/test_persistent_metrics_integration.py
@@ -0,0 +1,49 @@
+"""Tests forwarding existing proxy metric events to Lifetime storage."""
+
+from __future__ import annotations
+
+import asyncio
+
+from headroom.proxy.prometheus_metrics import PrometheusMetrics
+from headroom.proxy.savings_tracker import SavingsTracker
+
+
+def test_runtime_metric_events_feed_lifetime_without_resetting_runtime_counters(tmp_path) -> None:
+ tracker = SavingsTracker(path=str(tmp_path / "proxy_savings.json"), save_flush_every=25)
+ metrics = PrometheusMetrics(savings_tracker=tracker)
+
+ metrics.record_stack("codex")
+ asyncio.run(
+ metrics.record_request(
+ provider="anthropic",
+ model="claude-test",
+ input_tokens=10,
+ output_tokens=3,
+ tokens_saved=2,
+ latency_ms=1,
+ cached=True,
+ attempted_input_tokens=12,
+ cache_read_tokens=5,
+ cache_write_1h_tokens=2,
+ waste_signals={"repetition": 4},
+ )
+ )
+ asyncio.run(metrics.record_failed(provider="anthropic", model="claude-test"))
+ asyncio.run(metrics.record_rate_limited(provider="anthropic", model="claude-test"))
+ asyncio.run(metrics.record_cache_bust(tokens_lost=7))
+ asyncio.run(metrics.record_cache_miss_attribution("anthropic", "prefix_change"))
+
+ lifetime = tracker.lifetime_response()
+
+ assert lifetime["requests"]["total"] == 1
+ assert lifetime["requests"]["cached"] == 1
+ assert lifetime["requests"]["failed"] == 1
+ assert lifetime["requests"]["rate_limited"] == 1
+ assert lifetime["requests"]["by_provider"] == {"anthropic": 1}
+ assert lifetime["requests"]["by_stack"] == {"codex": 1}
+ assert lifetime["tokens"]["output"] == 3
+ assert lifetime["tokens"]["attempted_input"] == 12
+ assert lifetime["prefix_cache"]["bust_tokens"] == 7
+ assert lifetime["prefix_cache"]["misses_by_reason"] == {"prefix_change": 1}
+ assert lifetime["waste_signals"] == {"repetition": 4}
+ assert metrics.requests_total == 1
diff --git a/tests/test_persistent_metrics_persistence.py b/tests/test_persistent_metrics_persistence.py
new file mode 100644
index 000000000..a55f2dc91
--- /dev/null
+++ b/tests/test_persistent_metrics_persistence.py
@@ -0,0 +1,87 @@
+"""Tests schema v5 persistence around the pure Lifetime aggregate."""
+
+from __future__ import annotations
+
+import json
+
+from headroom.proxy.savings_tracker import SavingsTracker
+
+
+def test_savings_tracker_migrates_v4_lifetime_to_v5_metrics_and_preserves_legacy_state(tmp_path):
+ path = tmp_path / "proxy_savings.json"
+ legacy_state = {
+ "schema_version": 4,
+ "lifetime": {
+ "requests": 7,
+ "tokens_saved": 20,
+ "compression_savings_usd": 0.5,
+ "cache_read_tokens": 5,
+ "cache_savings_usd": 0.2,
+ "total_input_tokens": 80,
+ "total_input_cost_usd": 1.5,
+ },
+ "display_session": {
+ "requests": 2,
+ "tokens_saved": 4,
+ "compression_savings_usd": 0.1,
+ "cache_read_tokens": 1,
+ "cache_savings_usd": 0.01,
+ "total_input_tokens": 10,
+ "total_input_cost_usd": 0.2,
+ "started_at": "2026-07-01T00:00:00Z",
+ "last_activity_at": "2026-07-02T00:00:00Z",
+ },
+ "history": [
+ {
+ "timestamp": "2026-07-02T00:00:00Z",
+ "total_tokens_saved": 20,
+ "compression_savings_usd": 0.5,
+ "total_input_tokens": 80,
+ "total_input_cost_usd": 1.5,
+ }
+ ],
+ "projects": {"keep-me": {"requests": 1}},
+ }
+ path.write_text(json.dumps(legacy_state), encoding="utf-8")
+
+ tracker = SavingsTracker(path=str(path), save_flush_every=25)
+ lifetime = tracker.lifetime_response()
+
+ assert lifetime["schema_version"] == 5
+ assert lifetime["requests"]["total"] == 7
+ assert lifetime["tokens"]["input"] == 80
+ assert lifetime["tokens"]["attempted_input"] == 100
+ assert lifetime["tokens"]["saved"] == 20
+ assert lifetime["prefix_cache"]["cache_read_tokens"] == 5
+ assert lifetime["cost"] == {
+ "input_usd": 1.5,
+ "compression_savings_usd": 0.5,
+ "cache_savings_usd": 0.2,
+ }
+ assert lifetime["by_model"]["other"]["input_tokens"] == 80
+
+ tracker.flush()
+ saved = json.loads(path.read_text(encoding="utf-8"))
+ assert saved["schema_version"] == 5
+ assert saved["lifetime"] == legacy_state["lifetime"]
+ assert saved["display_session"]["requests"] == 2
+ assert saved["projects"]["keep-me"]["requests"] == 1
+ assert saved["lifetime_metrics"]["models"]["other"]["input_tokens"] == 80
+ assert isinstance(saved["lifetime_metrics"]["persistence"]["last_saved_at"], str)
+
+
+def test_lifetime_response_reports_stateless_mode_without_writing(tmp_path):
+ path = tmp_path / "proxy_savings.json"
+ tracker = SavingsTracker(path=str(path), stateless=True, save_flush_every=1)
+
+ tracker.record_lifetime_request(provider="openai", stack="codex", model="gpt-test", input_tokens=3)
+
+ response = tracker.lifetime_response()
+ assert response["persistence"] == {
+ "enabled": False,
+ "healthy": True,
+ "error": "Lifetime metrics unavailable in stateless mode",
+ "pending_records": 0,
+ "last_saved_at": None,
+ }
+ assert path.exists() is False
diff --git a/tests/test_proxy_loopback_gating.py b/tests/test_proxy_loopback_gating.py
index 73b0fb73b..89387a0ff 100644
--- a/tests/test_proxy_loopback_gating.py
+++ b/tests/test_proxy_loopback_gating.py
@@ -16,6 +16,7 @@ from fastapi.testclient import TestClient
from headroom.cache.backends import InMemoryBackend
from headroom.cache.compression_store import get_compression_store, reset_compression_store
+from headroom.proxy.loopback_guard import is_ip_literal_host_header
from headroom.proxy.server import ProxyConfig, create_app
GATED = [
@@ -75,6 +76,57 @@ def test_loopback_caller_allowed(method: str, path: str) -> None:
# CCR data endpoints — cached session content, gated to 404 off-loopback (#1227).
+def test_stats_lifetime_route_uses_dashboard_metadata_access_policy(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.setenv(
+ "HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS",
+ "100.90.0.5/32",
+ )
+ app = _make_app()
+ expected = {
+ "requests": {"total": 7},
+ "projects": {"headroom": {"requests": 3}},
+ "persistence": {
+ "enabled": True,
+ "healthy": False,
+ "error": "D:/private/proxy_savings.json: access denied",
+ },
+ }
+ monkeypatch.setattr(
+ app.state.proxy.metrics.savings_tracker,
+ "lifetime_response",
+ lambda: expected,
+ )
+
+ network = TestClient(app).get("/stats-lifetime")
+ assert network.status_code == 200, network.text
+ assert network.json() == {
+ "requests": {"total": 7},
+ "persistence": {
+ "enabled": True,
+ "healthy": False,
+ "error": None,
+ },
+ }
+
+ loopback = TestClient(
+ app,
+ base_url="http://127.0.0.1",
+ client=("127.0.0.1", 12345),
+ ).get("/stats-lifetime")
+ assert loopback.status_code == 200, loopback.text
+ assert loopback.json() == expected
+
+ trusted_dashboard = TestClient(
+ app,
+ base_url="http://100.82.0.2:8787",
+ client=("100.90.0.5", 12345),
+ ).get("/stats-lifetime")
+ assert trusted_dashboard.status_code == 200, trusted_dashboard.text
+ assert trusted_dashboard.json() == expected
+
+
CCR_GATED = [
("post", "/v1/retrieve"),
("get", "/v1/retrieve/stats"),
@@ -112,6 +164,22 @@ def test_dns_rebinding_host_header_rejected() -> None:
assert resp.status_code == 404, resp.text
+@pytest.mark.parametrize(
+ "host_header",
+ ["100.82.0.2", "100.82.0.2:8787", "[fd7a:115c:a1e0::2]", "[fd7a:115c:a1e0::2]:8787"],
+)
+def test_ip_literal_host_header_accepts_ip_addresses(host_header: str) -> None:
+ assert is_ip_literal_host_header(host_header) is True
+
+
+@pytest.mark.parametrize(
+ "host_header",
+ [None, "", "attacker.example", "localhost", "user@100.82.0.2", "100.82.0.2/path", "[fd7a::1"],
+)
+def test_ip_literal_host_header_rejects_non_addresses(host_header: str | None) -> None:
+ assert is_ip_literal_host_header(host_header) is False
+
+
def _client(*, loopback: bool) -> TestClient:
app = _make_app()
if loopback:
@@ -185,3 +253,198 @@ def test_stats_metadata_served_to_trusted_gateway_peer(
rebind = TestClient(app, base_url="http://attacker.example", client=(gateway_ip, 54321))
payload = rebind.get("/stats").json()
assert "recent_requests" not in payload
+
+
+@pytest.mark.parametrize("cached", [False, True])
+def test_dashboard_client_cidr_grants_stats_metadata_for_ip_literal_host(
+ monkeypatch: pytest.MonkeyPatch,
+ cached: bool,
+) -> None:
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS", "100.90.0.5/32")
+ app = _make_app()
+ client = TestClient(
+ app,
+ base_url="http://100.82.0.2:8787",
+ client=("100.90.0.5", 12345),
+ )
+
+ payload = client.get("/stats", params={"cached": int(cached)}).json()
+
+ assert "recent_requests" in payload
+ assert "request_logs" in payload
+ assert "config" in payload
+
+
+@pytest.mark.parametrize(
+ "headers",
+ [
+ {"origin": "http://100.82.0.2:8787"},
+ {"referer": "http://100.82.0.2:8787/dashboard"},
+ ],
+)
+@pytest.mark.parametrize("cached", [False, True])
+def test_dashboard_client_cidr_grants_stats_metadata_to_same_origin_browser(
+ monkeypatch: pytest.MonkeyPatch, headers: dict[str, str], cached: bool
+) -> None:
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS", "100.90.0.5/32")
+ client = TestClient(
+ _make_app(),
+ base_url="http://100.82.0.2:8787",
+ client=("100.90.0.5", 12345),
+ )
+
+ payload = client.get(
+ "/stats", params={"cached": int(cached)}, headers=headers
+ ).json()
+
+ assert "recent_requests" in payload
+ assert "request_logs" in payload
+ assert "config" in payload
+
+
+@pytest.mark.parametrize(
+ "headers",
+ [
+ {"origin": "http://attacker.example"},
+ {"referer": "http://attacker.example/dashboard"},
+ ],
+)
+@pytest.mark.parametrize("cached", [False, True])
+def test_dashboard_client_cidr_hides_stats_metadata_from_cross_origin_browser(
+ monkeypatch: pytest.MonkeyPatch, headers: dict[str, str], cached: bool
+) -> None:
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS", "100.90.0.5/32")
+ client = TestClient(
+ _make_app(),
+ base_url="http://100.82.0.2:8787",
+ client=("100.90.0.5", 12345),
+ )
+
+ response = client.get(
+ "/stats", params={"cached": int(cached)}, headers=headers
+ )
+ payload = response.json()
+
+ assert response.status_code == 200
+ assert "tokens" in payload
+ assert "recent_requests" not in payload
+ assert "request_logs" not in payload
+ assert "config" not in payload
+
+
+def test_dashboard_client_cidr_only_uses_forwarded_proto_from_trusted_gateway(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS", "100.90.0.5/32")
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_GATEWAY_CIDRS", "172.18.0.0/16")
+ client = TestClient(
+ _make_app(),
+ base_url="http://100.82.0.2:8787",
+ client=("172.18.0.1", 12345),
+ )
+
+ payload = client.get(
+ "/stats",
+ headers={
+ "origin": "https://100.82.0.2:8787",
+ "x-forwarded-for": "100.90.0.5",
+ "x-forwarded-proto": "https",
+ },
+ ).json()
+
+ assert "recent_requests" in payload
+ assert "request_logs" in payload
+ assert "config" in payload
+
+ spoofed = TestClient(
+ _make_app(),
+ base_url="http://100.82.0.2:8787",
+ client=("100.90.0.5", 12345),
+ ).get(
+ "/stats",
+ headers={
+ "origin": "https://100.82.0.2:8787",
+ "x-forwarded-proto": "https",
+ },
+ ).json()
+
+ assert "recent_requests" not in spoofed
+ assert "request_logs" not in spoofed
+ assert "config" not in spoofed
+
+
+def test_dashboard_client_cidr_rejects_unlisted_clients_and_hostname_hosts(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS", "100.90.0.5/32")
+ app = _make_app()
+
+ unlisted = TestClient(
+ app,
+ base_url="http://100.82.0.2:8787",
+ client=("100.90.0.6", 12345),
+ ).get("/stats").json()
+ hostname = TestClient(
+ app,
+ base_url="http://100.82.0.2:8787",
+ client=("100.90.0.5", 12345),
+ ).get("/stats", headers={"host": "attacker.example"}).json()
+
+ for payload in (unlisted, hostname):
+ assert "recent_requests" not in payload
+ assert "request_logs" not in payload
+ assert "config" not in payload
+
+
+def test_dashboard_client_cidr_only_accepts_forwarded_client_from_trusted_gateway(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS", "100.90.0.5/32")
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_GATEWAY_CIDRS", "172.18.0.0/16")
+ app = _make_app()
+
+ trusted = TestClient(
+ app,
+ base_url="http://100.82.0.2:8787",
+ client=("172.18.0.1", 12345),
+ ).get("/stats", headers={"x-forwarded-for": "100.90.0.5"}).json()
+ forged = TestClient(
+ app,
+ base_url="http://100.82.0.2:8787",
+ client=("198.51.100.10", 12345),
+ ).get("/stats", headers={"x-forwarded-for": "100.90.0.5"}).json()
+
+ assert "recent_requests" in trusted
+ assert "recent_requests" not in forged
+
+
+def test_dashboard_client_cidr_normalizes_ipv4_mapped_ipv6(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS", "100.90.0.0/24")
+ app = _make_app()
+ payload = TestClient(
+ app,
+ base_url="http://100.82.0.2:8787",
+ client=("::ffff:100.90.0.5", 12345),
+ ).get("/stats").json()
+
+ assert "recent_requests" in payload
+
+
+def test_dashboard_client_cidr_does_not_expand_other_management_endpoints(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.setenv("HEADROOM_PROXY_TRUSTED_DASHBOARD_CLIENT_CIDRS", "100.90.0.5/32")
+ client = TestClient(
+ _make_app(),
+ base_url="http://100.82.0.2:8787",
+ client=("100.90.0.5", 12345),
+ )
+
+ health = client.get("/health")
+ assert health.status_code == 200
+ assert "config" not in health.json()
+ assert client.get("/admin/upstream").status_code == 404
+ assert client.get("/debug/tasks").status_code == 404
+ assert client.post("/stats/reset").status_code == 404
diff --git a/tests/test_proxy_project_savings.py b/tests/test_proxy_project_savings.py
index 2ea4c08a8..c39884aac 100644
--- a/tests/test_proxy_project_savings.py
+++ b/tests/test_proxy_project_savings.py
@@ -122,6 +122,7 @@ def test_tracker_accumulates_per_project_and_persists(tmp_path):
# Survives a restart via the persisted JSON state.
reloaded = SavingsTracker(path=str(path))
assert reloaded.stats_preview()["projects"]["api"]["tokens_saved"] == 500
+ assert reloaded.lifetime_response()["projects"]["api"]["tokens_saved"] == 500
def test_tracker_migrates_v2_state_without_projects(tmp_path):
diff --git a/tests/test_proxy_stats_recent_requests.py b/tests/test_proxy_stats_recent_requests.py
index 654632e8f..debbde158 100644
--- a/tests/test_proxy_stats_recent_requests.py
+++ b/tests/test_proxy_stats_recent_requests.py
@@ -81,7 +81,7 @@ def test_stats_refreshes_recent_requests_when_cached() -> None:
assert second_response.status_code == 200
second_payload = second_response.json()
- assert second_payload["recent_requests"][-1]["model"] == "claude-sonnet"
+ assert second_payload["recent_requests"][0]["model"] == "claude-sonnet"
assert second_payload["request_logs"][-1]["model"] == "claude-sonnet"
@@ -148,18 +148,18 @@ def test_stats_recent_requests_includes_token_incomplete_requests() -> None:
assert response.status_code == 200
payload = response.json()
assert [req["model"] for req in payload["recent_requests"]] == [
- "claude-haiku",
- "claude-haiku",
"claude-sonnet",
+ "claude-haiku",
+ "claude-haiku",
]
- assert payload["recent_requests"][0]["input_tokens_optimized"] is None
- assert payload["recent_requests"][0]["token_accounting_status"] == "missing"
- assert payload["recent_requests"][0]["has_exact_tokens"] is False
+ assert payload["recent_requests"][0]["token_accounting_status"] == "complete"
+ assert payload["recent_requests"][0]["has_exact_tokens"] is True
assert payload["recent_requests"][1]["output_tokens"] is None
assert payload["recent_requests"][1]["token_accounting_status"] == "partial"
assert payload["recent_requests"][1]["tokens_saved"] == 0
- assert payload["recent_requests"][2]["token_accounting_status"] == "complete"
- assert payload["recent_requests"][2]["has_exact_tokens"] is True
+ assert payload["recent_requests"][2]["input_tokens_optimized"] is None
+ assert payload["recent_requests"][2]["token_accounting_status"] == "missing"
+ assert payload["recent_requests"][2]["has_exact_tokens"] is False
assert payload["summary"]["uncompressed_requests"]["unknown_token_accounting"] == 2
assert payload["request_logs"][-1]["model"] == "claude-sonnet"