Refactor: extract 7 modules from server.py (Steps 2-4)

server.py: 8778 → 7412 lines (-1366, -15.5%)

Extracted modules:
- cost.py (629 lines): CostTracker, build_prefix_cache_stats, merge_cost_stats
- prometheus_metrics.py (312 lines): PrometheusMetrics
- semantic_cache.py (142 lines): SemanticCache
- rate_limiter.py (101 lines): TokenBucketRateLimiter
- request_logger.py (108 lines): RequestLogger
- helpers.py (195 lines): _read_request_json, constants, lazy loaders
- models.py (199 lines): ProxyConfig, RequestLog, CacheEntry (from Step 1)

All existing imports via headroom.proxy.server continue to work
through re-exports. Updated test patches to target new module paths.

181 tests pass, 0 regressions.

headroom/proxy/cost.py

@@ -0,0 +1,629 @@
+"""Cost tracking and budget management for the Headroom proxy.
+
+Contains the CostTracker class and cost-related helper functions
+for prefix cache statistics, cost merging, and session summaries.
+
+Extracted from server.py for maintainability.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections import deque
+from datetime import datetime, timedelta
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from headroom.proxy.prometheus_metrics import PrometheusMetrics
+
+# Try to import LiteLLM for pricing
+try:
+    import litellm
+
+    LITELLM_AVAILABLE = True
+except ImportError:
+    LITELLM_AVAILABLE = False
+
+logger = logging.getLogger("headroom.proxy")
+
+# Provider-specific cache discount multipliers (what fraction of input price)
+# Used to calculate dollar savings from prefix caching
+_CACHE_ECONOMICS = {
+    "anthropic": {
+        "read_multiplier": 0.1,
+        "write_multiplier": 1.25,
+        "label": "Explicit breakpoints, 5-min TTL",
+    },
+    "openai": {
+        "read_multiplier": 0.5,
+        "write_multiplier": 1.0,
+        "label": "Automatic, no TTL control",
+    },
+    "gemini": {
+        "read_multiplier": 0.1,
+        "write_multiplier": 1.0,
+        "label": "Explicit cachedContent, configurable TTL",
+    },
+    "bedrock": {
+        "read_multiplier": 0.1,
+        "write_multiplier": 1.25,
+        "label": "Same as Anthropic (Bedrock)",
+    },
+}
+
+
+def _summarize_transforms(transforms: list[str]) -> str:
+    """Collapse repeated transforms into counted summary.
+
+    e.g. ['router:excluded:tool', 'router:excluded:tool', 'read_lifecycle:stale']
+      → 'router:excluded:tool*2 read_lifecycle:stale'
+    """
+    if not transforms:
+        return "none"
+    counts: dict[str, int] = {}
+    for t in transforms:
+        counts[t] = counts.get(t, 0) + 1
+    parts = [f"{k}*{v}" if v > 1 else k for k, v in counts.items()]
+    return " ".join(parts)
+
+
+def build_prefix_cache_stats(
+    metrics: PrometheusMetrics,
+    cost_tracker: CostTracker | None,
+) -> dict:
+    """Build provider-aware prefix cache statistics for the dashboard."""
+    by_provider = {}
+    totals = {
+        "cache_read_tokens": 0,
+        "cache_write_tokens": 0,
+        "requests": 0,
+        "hit_requests": 0,
+        "bust_count": 0,
+        "bust_write_tokens": 0,
+        "savings_usd": 0.0,
+        "write_premium_usd": 0.0,
+    }
+
+    for provider, pc in metrics.cache_by_provider.items():
+        if pc["requests"] == 0:
+            continue
+
+        econ = _CACHE_ECONOMICS.get(provider, _CACHE_ECONOMICS["anthropic"])
+        read_mult: float = econ["read_multiplier"]  # type: ignore[assignment]
+        write_mult: float = econ["write_multiplier"]  # type: ignore[assignment]
+
+        # Get the base input price per token for the most-used model on this provider
+        input_price_per_token = None
+        if cost_tracker:
+            for model_name in cost_tracker._tokens_sent_by_model:
+                # Match model to provider
+                _openai_prefixes = ("gpt", "o1", "o3", "o4")
+                is_match = (
+                    (provider == "anthropic" and "claude" in model_name)
+                    or (provider == "openai" and any(p in model_name for p in _openai_prefixes))
+                    or (provider == "gemini" and "gemini" in model_name)
+                    or (provider == "bedrock" and "claude" in model_name)
+                )
+                if is_match:
+                    price_per_1m = cost_tracker._get_list_price(model_name)
+                    if price_per_1m:
+                        input_price_per_token = price_per_1m / 1_000_000
+                        break
+
+        # Calculate savings:
+        # Cache reads save (1.0 - read_mult) per token vs uncached input price.
+        # Cache write premium is NOT deducted — it's baseline cost that the
+        # client (e.g. Claude Code) pays regardless of Headroom. We track it
+        # for observability but don't penalise our savings number.
+        read_tokens: int = pc["cache_read_tokens"]  # type: ignore[assignment]
+        write_tokens: int = pc["cache_write_tokens"]  # type: ignore[assignment]
+        savings_usd = 0.0
+        write_premium_usd = 0.0
+
+        if input_price_per_token:
+            # Savings from reads: tokens * price * (1.0 - read_multiplier)
+            savings_usd = read_tokens * input_price_per_token * (1.0 - read_mult)
+            # Write premium (observability only — not subtracted from savings)
+            if write_mult > 1.0:
+                write_premium_usd = write_tokens * input_price_per_token * (write_mult - 1.0)
+
+        hit_rate = round(pc["hit_requests"] / pc["requests"] * 100, 1) if pc["requests"] > 0 else 0
+
+        provider_stats = {
+            "cache_read_tokens": read_tokens,
+            "cache_write_tokens": write_tokens,
+            "requests": pc["requests"],
+            "hit_requests": pc["hit_requests"],
+            "hit_rate": hit_rate,
+            "bust_count": pc["bust_count"],
+            "bust_write_tokens": pc["bust_write_tokens"],
+            "read_discount": f"{(1.0 - read_mult) * 100:.0f}%",
+            "write_premium": f"{(write_mult - 1.0) * 100:.0f}%" if write_mult > 1.0 else "none",
+            "savings_usd": round(savings_usd, 4),
+            "write_premium_usd": round(write_premium_usd, 4),
+            "net_savings_usd": round(savings_usd, 4),
+            "label": str(econ["label"]),
+        }
+        by_provider[provider] = provider_stats
+
+        # Accumulate totals
+        totals["cache_read_tokens"] += read_tokens
+        totals["cache_write_tokens"] += write_tokens
+        totals["requests"] += pc["requests"]
+        totals["hit_requests"] += pc["hit_requests"]
+        totals["bust_count"] += pc["bust_count"]
+        totals["bust_write_tokens"] += pc["bust_write_tokens"]
+        totals["savings_usd"] += savings_usd
+        totals["write_premium_usd"] += write_premium_usd
+
+    totals["net_savings_usd"] = round(totals["savings_usd"], 4)
+    totals["savings_usd"] = round(totals["savings_usd"], 4)
+    totals["write_premium_usd"] = round(totals["write_premium_usd"], 4)
+    totals["hit_rate"] = (
+        round(totals["hit_requests"] / totals["requests"] * 100, 1) if totals["requests"] > 0 else 0
+    )
+
+    return {
+        "by_provider": by_provider,
+        "totals": totals,
+        "prefix_freeze": {
+            "busts_avoided": metrics.prefix_freeze_busts_avoided,
+            "tokens_preserved": metrics.prefix_freeze_tokens_preserved,
+            "compression_foregone_tokens": metrics.prefix_freeze_compression_foregone,
+            "net_benefit_tokens": (
+                metrics.prefix_freeze_tokens_preserved - metrics.prefix_freeze_compression_foregone
+            ),
+        },
+        "attribution": (
+            "Prefix caching is performed by the LLM provider (Anthropic, OpenAI). "
+            "Headroom reports cache stats as observed from API responses. "
+            "CacheAligner and prefix freeze improve cache hit rates by stabilizing "
+            "the message prefix, but baseline caching happens without Headroom."
+        ),
+    }
+
+
+def merge_cost_stats(
+    cost_stats: dict | None,
+    cache_stats: dict,
+    cli_tokens_avoided: int = 0,
+) -> dict | None:
+    """Merge compression, cache, and CLI savings into cost stats.
+
+    Each savings layer is reported separately with its own scope:
+    - savings_usd: compression savings at model list price (monotonic)
+    - cache_savings_usd: prefix cache discount from provider (separate)
+    - cli_tokens_avoided: tokens filtered by rtk (token count only, no $ estimate)
+
+    The hero metric (savings_usd) is ONLY compression savings priced at
+    the model's published input rate. Cache and CLI are shown separately.
+    This avoids the non-monotonic moving-average repricing bug (#83).
+    """
+    if cost_stats is None:
+        return None
+
+    cache_net = cache_stats.get("totals", {}).get("net_savings_usd", 0.0)
+    compression_savings = cost_stats.get("savings_usd", 0.0)
+
+    return {
+        **cost_stats,
+        "savings_usd": round(compression_savings, 4),
+        "compression_savings_usd": round(compression_savings, 4),
+        "cache_savings_usd": round(cache_net, 4),
+        "cli_tokens_avoided": cli_tokens_avoided,
+    }
+
+
+def build_session_summary(
+    proxy: Any,
+    metrics: Any,
+    prefix_cache_stats: dict,
+    cli_tokens_avoided: int,
+    total_tokens_before: int,
+) -> dict[str, Any]:
+    """Build a human-readable session summary from metrics and request logs.
+
+    This is the headline view users see first in /stats — designed to answer
+    "is Headroom working?" at a glance.
+    """
+    # Analyze per-request compression from the logger
+    compressed_requests: list[dict] = []
+    uncompressed_reasons: dict[str, int] = {
+        "prefix_frozen": 0,
+        "too_small": 0,
+        "passthrough": 0,
+        "no_compressible_content": 0,
+    }
+
+    if proxy.logger:
+        for entry in proxy.logger._logs:
+            if entry.model and "count_tokens" in entry.model:
+                uncompressed_reasons["passthrough"] += 1
+                continue
+            if entry.tokens_saved > 0 and entry.savings_percent > 0:
+                compressed_requests.append(
+                    {
+                        "savings_pct": round(entry.savings_percent, 1),
+                        "tokens_saved": entry.tokens_saved,
+                        "original": entry.input_tokens_original,
+                        "optimized": entry.input_tokens_optimized,
+                    }
+                )
+            elif entry.input_tokens_original > 0:
+                # Categorize why it wasn't compressed
+                transforms = entry.transforms_applied or []
+                if not transforms:
+                    # Pipeline returned unchanged — likely all frozen
+                    uncompressed_reasons["prefix_frozen"] += 1
+                elif all("excluded" in t or "protected" in t for t in transforms):
+                    uncompressed_reasons["no_compressible_content"] += 1
+                elif entry.input_tokens_original < 500:
+                    uncompressed_reasons["too_small"] += 1
+                else:
+                    uncompressed_reasons["prefix_frozen"] += 1
+
+    # Compute compression stats for requests that DID compress
+    avg_compression = 0.0
+    best_compression = 0.0
+    best_detail = ""
+    if compressed_requests:
+        avg_compression = round(
+            sum(r["savings_pct"] for r in compressed_requests) / len(compressed_requests),
+            1,
+        )
+        best = max(compressed_requests, key=lambda r: r["savings_pct"])
+        best_compression = best["savings_pct"]
+        best_detail = f"{best['original']:,} → {best['optimized']:,} tokens"
+
+    # Cost summary — savings_usd is compression savings at model list price (monotonic)
+    cost_stats = proxy.cost_tracker.stats() if proxy.cost_tracker else {}
+    cost_with = cost_stats.get("cost_with_headroom_usd", 0.0)
+    compression_savings = cost_stats.get("savings_usd", 0.0)
+    cache_net = prefix_cache_stats.get("totals", {}).get("net_savings_usd", 0.0)
+    total_saved_usd = round(compression_savings, 2)
+    cost_without = cost_with + compression_savings
+    savings_pct_cost = round(total_saved_usd / cost_without * 100, 1) if cost_without > 0 else 0.0
+
+    # Primary models used
+    models = dict(metrics.requests_by_model)
+    primary_model = max(models, key=lambda k: models[k]) if models else "unknown"
+    api_requests = sum(v for k, v in models.items() if "count_tokens" not in k)
+
+    # Build the summary
+    summary: dict[str, Any] = {
+        "mode": proxy.config.mode,
+        "api_requests": api_requests,
+        "primary_model": primary_model,
+        "compression": {
+            "requests_compressed": len(compressed_requests),
+            "avg_compression_pct": avg_compression,
+            "best_compression_pct": best_compression,
+            "best_detail": best_detail,
+            "total_tokens_removed": metrics.tokens_saved_total,
+        },
+        "uncompressed_requests": {k: v for k, v in uncompressed_reasons.items() if v > 0},
+        "cost": {
+            "without_headroom_usd": round(cost_without, 2),
+            "with_headroom_usd": round(cost_with, 2),
+            "total_saved_usd": total_saved_usd,
+            "savings_pct": savings_pct_cost,
+            "breakdown": {
+                "cache_savings_usd": round(cache_net, 2),
+                "compression_savings_usd": round(compression_savings, 2),
+            },
+        },
+    }
+
+    # Add tip if token_headroom mode would help
+    if proxy.config.mode == "cost_savings" and uncompressed_reasons["prefix_frozen"] > 10:
+        summary["tip"] = (
+            "Most requests are prefix-frozen. Set HEADROOM_MODE=token_headroom "
+            "to compress frozen messages and extend your session by ~25-35%."
+        )
+
+    return summary
+
+
+class CostTracker:
+    """Track costs and enforce budgets.
+
+    Cost history is automatically pruned to prevent unbounded memory growth:
+    - Entries older than 24 hours are removed
+    - Maximum of 100,000 entries are kept
+
+    Uses LiteLLM's community-maintained pricing database for accurate costs.
+    See: https://github.com/BerriAI/litellm/blob/main/model_prices_and_context_window.json
+    """
+
+    MAX_COST_ENTRIES = 100_000
+    COST_RETENTION_HOURS = 24
+
+    def __init__(self, budget_limit_usd: float | None = None, budget_period: str = "daily"):
+        self.budget_limit_usd = budget_limit_usd
+        self.budget_period = budget_period
+
+        # Cost tracking - using deque for efficient left-side removal
+        self._costs: deque[tuple[datetime, float]] = deque(maxlen=self.MAX_COST_ENTRIES)
+        self._last_prune_time: datetime = datetime.now()
+
+        # Token savings per model (exact, no dollar estimation)
+        self._tokens_saved_by_model: dict[str, int] = {}
+        self._tokens_sent_by_model: dict[str, int] = {}
+        self._requests_by_model: dict[str, int] = {}
+
+        # API-reported cache breakdown per model (for accurate cost calculation)
+        self._api_cache_read_by_model: dict[str, int] = {}
+        self._api_cache_write_by_model: dict[str, int] = {}
+        self._api_uncached_by_model: dict[str, int] = {}
+
+    # Cache resolved model names to avoid repeated litellm lookups.
+    # This is critical: litellm.cost_per_token() is synchronous and can block
+    # the async event loop if it triggers I/O (lazy model info download).
+    _resolved_model_cache: dict[str, str] = {}
+
+    @classmethod
+    def _resolve_litellm_model(cls, model: str) -> str:
+        """Resolve model name to one LiteLLM recognizes, adding provider prefix if needed.
+
+        Results are cached per model name to avoid blocking the event loop
+        with repeated synchronous litellm lookups.
+        """
+        if model in cls._resolved_model_cache:
+            return cls._resolved_model_cache[model]
+
+        resolved = cls._resolve_litellm_model_uncached(model)
+        cls._resolved_model_cache[model] = resolved
+        return resolved
+
+    @staticmethod
+    def _resolve_litellm_model_uncached(model: str) -> str:
+        """Uncached resolution — called once per unique model name."""
+        if not LITELLM_AVAILABLE:
+            return model
+
+        # Try as-is first
+        try:
+            litellm.cost_per_token(model=model, prompt_tokens=1, completion_tokens=0)
+            return model
+        except Exception:
+            pass
+
+        # Try with provider prefix
+        prefixes = {
+            "claude-": "anthropic/",
+            "gpt-": "openai/",
+            "o1-": "openai/",
+            "o3-": "openai/",
+            "o4-": "openai/",
+            "gemini-": "google/",
+        }
+        for pattern, prefix in prefixes.items():
+            if model.startswith(pattern):
+                prefixed = f"{prefix}{model}"
+                try:
+                    litellm.cost_per_token(model=prefixed, prompt_tokens=1, completion_tokens=0)
+                    return prefixed
+                except Exception:
+                    break
+
+        return model
+
+    def estimate_cost(
+        self,
+        model: str,
+        input_tokens: int,
+        output_tokens: int,
+        cache_read_tokens: int = 0,
+        cache_write_tokens: int = 0,
+    ) -> float | None:
+        """Estimate cost in USD using LiteLLM's pricing database.
+
+        LiteLLM natively handles cache_read and cache_creation pricing
+        for all providers (Anthropic, OpenAI, Google, etc.) in a single call.
+
+        Args:
+            model: Model name for pricing lookup
+            input_tokens: Non-cached input tokens (excludes cache_read)
+            output_tokens: Output tokens
+            cache_read_tokens: Tokens served from cache (~10% of input rate)
+            cache_write_tokens: Tokens written to cache (~125% of input rate)
+        """
+        if not LITELLM_AVAILABLE:
+            logger.warning("LiteLLM not available - cannot calculate costs")
+            return None
+
+        try:
+            resolved_model = self._resolve_litellm_model(model)
+
+            # litellm.cost_per_token handles all token types natively:
+            # prompt_tokens at input rate, cache_read at ~10%, cache_creation at ~125%
+            input_cost, output_cost = litellm.cost_per_token(
+                model=resolved_model,
+                prompt_tokens=input_tokens,
+                completion_tokens=output_tokens,
+                cache_read_input_tokens=cache_read_tokens,
+                cache_creation_input_tokens=cache_write_tokens,
+            )
+
+            total_cost = input_cost + output_cost
+            return float(total_cost) if total_cost > 0 else None
+
+        except Exception as e:
+            logger.warning(f"Failed to get pricing for model {model}: {e}")
+            return None
+
+    def _prune_old_costs(self):
+        """Remove cost entries older than retention period.
+
+        Called periodically (every 5 minutes) to prevent unbounded memory growth.
+        The deque maxlen provides a hard cap, but time-based pruning keeps
+        memory usage proportional to actual traffic patterns.
+        """
+        now = datetime.now()
+        # Only prune every 5 minutes to avoid overhead
+        if (now - self._last_prune_time).total_seconds() < 300:
+            return
+
+        self._last_prune_time = now
+        cutoff = now - timedelta(hours=self.COST_RETENTION_HOURS)
+
+        # Remove entries from the left (oldest) while they're older than cutoff
+        while self._costs and self._costs[0][0] < cutoff:
+            self._costs.popleft()
+
+    def record_tokens(
+        self,
+        model: str,
+        tokens_saved: int,
+        tokens_sent: int,
+        cache_read_tokens: int = 0,
+        cache_write_tokens: int = 0,
+        uncached_tokens: int = 0,
+    ):
+        """Record token counts per model.
+
+        Args:
+            model: Model name.
+            tokens_saved: Tokens removed by compression (Headroom's count).
+            tokens_sent: Compressed message tokens sent (Headroom's count).
+            cache_read_tokens: Cache read tokens from API response usage.
+            cache_write_tokens: Cache write tokens from API response usage.
+            uncached_tokens: Non-cached input tokens from API response usage.
+        """
+        self._tokens_saved_by_model[model] = (
+            self._tokens_saved_by_model.get(model, 0) + tokens_saved
+        )
+        self._tokens_sent_by_model[model] = self._tokens_sent_by_model.get(model, 0) + tokens_sent
+        self._requests_by_model[model] = self._requests_by_model.get(model, 0) + 1
+        self._api_cache_read_by_model[model] = (
+            self._api_cache_read_by_model.get(model, 0) + cache_read_tokens
+        )
+        self._api_cache_write_by_model[model] = (
+            self._api_cache_write_by_model.get(model, 0) + cache_write_tokens
+        )
+        self._api_uncached_by_model[model] = (
+            self._api_uncached_by_model.get(model, 0) + uncached_tokens
+        )
+
+    def get_period_cost(self) -> float:
+        """Get cost for current budget period."""
+        now = datetime.now()
+
+        if self.budget_period == "hourly":
+            cutoff = now - timedelta(hours=1)
+        elif self.budget_period == "daily":
+            cutoff = now.replace(hour=0, minute=0, second=0, microsecond=0)
+        else:  # monthly
+            cutoff = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
+
+        return sum(cost for ts, cost in self._costs if ts >= cutoff)
+
+    def check_budget(self) -> tuple[bool, float]:
+        """Check if within budget. Returns (allowed, remaining)."""
+        if self.budget_limit_usd is None:
+            return True, float("inf")
+
+        period_cost = self.get_period_cost()
+        remaining = self.budget_limit_usd - period_cost
+        return remaining > 0, max(0, remaining)
+
+    def _get_list_price(self, model: str) -> float | None:
+        """Get list input price per 1M tokens for a model."""
+        if not LITELLM_AVAILABLE:
+            return None
+        try:
+            resolved = self._resolve_litellm_model(model)
+            info = litellm.model_cost.get(resolved, {})
+            cost_per_token = info.get("input_cost_per_token")
+            return cost_per_token * 1_000_000 if cost_per_token else None
+        except Exception:
+            return None
+
+    def _get_cache_prices(self, model: str) -> tuple[float, float, float] | None:
+        """Get per-token prices for cache read, cache write, and uncached input.
+
+        Returns (cache_read, cache_write, uncached) per-token costs, or None
+        if pricing is unavailable. Uses LiteLLM's native cache pricing data.
+        """
+        if not LITELLM_AVAILABLE:
+            return None
+        try:
+            resolved = self._resolve_litellm_model(model)
+            info = litellm.model_cost.get(resolved, {})
+            uncached = info.get("input_cost_per_token")
+            if not uncached:
+                return None
+            cache_read = info.get("cache_read_input_token_cost", uncached)
+            cache_write = info.get("cache_creation_input_token_cost", uncached)
+            return (cache_read, cache_write, uncached)
+        except Exception:
+            return None
+
+    def stats(self) -> dict:
+        """Get token statistics per model."""
+        per_model = {}
+        total_saved = 0
+        for model in sorted(self._tokens_saved_by_model.keys()):
+            saved = self._tokens_saved_by_model[model]
+            sent = self._tokens_sent_by_model.get(model, 0)
+            reqs = self._requests_by_model.get(model, 0)
+            total_saved += saved
+            per_model[model] = {
+                "requests": reqs,
+                "tokens_saved": saved,
+                "tokens_sent": sent,
+                "reduction_pct": round(saved / (saved + sent) * 100, 1)
+                if (saved + sent) > 0
+                else 0,
+            }
+
+        # Compute actual input cost using API-reported cache breakdown and
+        # LiteLLM's per-category pricing (cache reads discounted, writes at
+        # premium, uncached at list). Falls back to list price when cache
+        # data is unavailable.
+        cost_with_headroom = 0.0
+        total_billed_input_tokens = 0
+        total_input_tokens = 0
+        for model in self._tokens_saved_by_model:
+            saved = self._tokens_saved_by_model[model]
+            sent = self._tokens_sent_by_model.get(model, 0)
+            cr = self._api_cache_read_by_model.get(model, 0)
+            cw = self._api_cache_write_by_model.get(model, 0)
+            uncached = self._api_uncached_by_model.get(model, 0)
+            total_input_tokens += sent
+
+            prices = self._get_cache_prices(model)
+            if prices:
+                cr_price, cw_price, uncached_price = prices
+                if cr + cw + uncached > 0:
+                    # Use API's real cache breakdown with LiteLLM pricing
+                    model_cost = cr * cr_price + cw * cw_price + uncached * uncached_price
+                    billed_tokens = cr + cw + uncached
+                else:
+                    # No cache data from API — fall back to list price
+                    model_cost = sent * uncached_price
+                    billed_tokens = sent
+                cost_with_headroom += model_cost
+                total_billed_input_tokens += billed_tokens
+
+        # Compression savings: price saved tokens at the model's list input price.
+        # This is simple, monotonic, and transparent — each saved token is valued
+        # at the published $/token rate for its model. Not affected by cache mix.
+        savings_usd = 0.0
+        for model in self._tokens_saved_by_model:
+            saved = self._tokens_saved_by_model[model]
+            if saved <= 0:
+                continue
+            prices = self._get_cache_prices(model)
+            if prices:
+                _cr_price, _cw_price, uncached_price = prices
+                savings_usd += saved * uncached_price
+
+        return {
+            "total_tokens_saved": total_saved,
+            "total_input_tokens": total_input_tokens,
+            "total_input_cost_usd": round(cost_with_headroom, 4),
+            "per_model": per_model,
+            "cost_with_headroom_usd": round(cost_with_headroom, 4),
+            "savings_usd": round(savings_usd, 4),
+        }

headroom/proxy/helpers.py

@@ -0,0 +1,195 @@
+"""Top-level helper functions and constants for the Headroom proxy.
+
+Contains lazy loaders, file logging setup, request body decompression,
+and safety-limit constants.
+
+Extracted from server.py for maintainability.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from fastapi import Request
+
+logger = logging.getLogger("headroom.proxy")
+
+# Maximum request body size (100MB - increased to support image-heavy requests)
+MAX_REQUEST_BODY_SIZE = 100 * 1024 * 1024
+
+# Maximum SSE buffer size (10MB - prevents memory exhaustion from malformed streams)
+MAX_SSE_BUFFER_SIZE = 10 * 1024 * 1024
+
+# Maximum message array length (prevents DoS from deeply nested payloads)
+MAX_MESSAGE_ARRAY_LENGTH = 10000
+
+# Compression pipeline timeout in seconds
+COMPRESSION_TIMEOUT_SECONDS = 30
+
+# Maximum compression cache sessions (prevents unbounded memory growth)
+MAX_COMPRESSION_CACHE_SESSIONS = 500
+
+# Image compression (lazy-loaded to avoid heavy dependencies at startup)
+_image_compressor = None
+
+
+def _get_image_compressor():
+    """Lazy load image compressor to avoid startup overhead."""
+    global _image_compressor
+    if _image_compressor is None:
+        try:
+            from headroom.image import ImageCompressor
+
+            _image_compressor = ImageCompressor()
+            logger.info("Image compression enabled (model: chopratejas/technique-router)")
+        except ImportError as e:
+            logger.warning(f"Image compression not available: {e}")
+            _image_compressor = False  # Mark as unavailable
+    return _image_compressor if _image_compressor else None
+
+
+# Always-on file logging to ~/.headroom/logs/ for `headroom perf` analysis
+_HEADROOM_LOG_DIR = Path.home() / ".headroom" / "logs"
+
+
+def _setup_file_logging() -> None:
+    """Add a RotatingFileHandler to the headroom root logger.
+
+    Writes to ~/.headroom/logs/proxy.log with automatic rotation:
+    - Rotates at 10 MB
+    - Keeps 5 backups (~50 MB max)
+    """
+    from logging.handlers import RotatingFileHandler
+
+    try:
+        _HEADROOM_LOG_DIR.mkdir(parents=True, exist_ok=True)
+        log_path = _HEADROOM_LOG_DIR / "proxy.log"
+        handler = RotatingFileHandler(
+            log_path,
+            maxBytes=10 * 1024 * 1024,  # 10 MB
+            backupCount=5,
+            encoding="utf-8",
+        )
+        handler.setLevel(logging.INFO)
+        handler.setFormatter(
+            logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+        )
+        # Attach to the headroom root logger so all sub-loggers are captured
+        logging.getLogger("headroom").addHandler(handler)
+    except OSError:
+        # Non-fatal: can't write logs (read-only fs, permissions, etc.)
+        pass
+
+
+def _get_rtk_stats() -> dict[str, Any] | None:
+    """Get rtk (Rust Token Killer) savings stats if rtk is installed.
+
+    Reads from rtk's tracking database via `rtk gain --format json`.
+    Returns None if rtk is not installed.
+    """
+    import shutil
+    import subprocess as _sp
+
+    rtk_bin = shutil.which("rtk")
+    if not rtk_bin:
+        # Check headroom-managed install
+        rtk_managed = Path.home() / ".headroom" / "bin" / "rtk"
+        if rtk_managed.exists():
+            rtk_bin = str(rtk_managed)
+        else:
+            return None
+
+    try:
+        result = _sp.run(
+            [rtk_bin, "gain", "--format", "json"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+        if result.returncode == 0 and result.stdout.strip():
+            data = json.loads(result.stdout)
+            summary = data.get("summary", {})
+            return {
+                "installed": True,
+                "total_commands": summary.get("total_commands", 0),
+                "tokens_saved": summary.get("total_saved", 0),
+                "avg_savings_pct": summary.get("avg_savings_pct", 0.0),
+            }
+    except Exception:
+        pass
+
+    return {"installed": True, "total_commands": 0, "tokens_saved": 0, "avg_savings_pct": 0.0}
+
+
+async def _read_request_json(request: Request) -> dict[str, Any]:
+    """Read and parse JSON from a request, handling compressed bodies.
+
+    Clients like OpenAI Codex may send zstd, gzip, or deflate-compressed
+    request bodies.  Starlette's ``request.json()`` does not decompress
+    automatically, causing a UnicodeDecodeError on compressed bytes.
+
+    This helper inspects ``Content-Encoding``, decompresses if needed,
+    then JSON-decodes the result.  It raises ``ValueError`` on any
+    decompression or parse failure so callers can return a clean 400.
+    """
+    encoding = (request.headers.get("content-encoding") or "").lower().strip()
+    raw = await request.body()
+
+    if encoding in ("zstd", "zstandard"):
+        try:
+            import zstandard
+
+            dctx = zstandard.ZstdDecompressor()
+            # Use stream_reader for streaming zstd frames (no content size in header).
+            # Plain decompress() fails when the frame header omits the size, which
+            # is common with clients like OpenAI Codex.
+            reader = dctx.stream_reader(raw)
+            raw = reader.read()
+            reader.close()
+        except ImportError:
+            raise ValueError(
+                "Request body is zstd-compressed but the 'zstandard' package is not installed. "
+                "Install it with: pip install zstandard"
+            ) from None
+        except Exception as exc:
+            raise ValueError(f"Failed to decompress zstd request body: {exc}") from exc
+    elif encoding == "gzip":
+        import gzip as _gzip
+
+        try:
+            raw = _gzip.decompress(raw)
+        except Exception as exc:
+            raise ValueError(f"Failed to decompress gzip request body: {exc}") from exc
+    elif encoding == "deflate":
+        import zlib
+
+        try:
+            raw = zlib.decompress(raw)
+        except Exception as exc:
+            raise ValueError(f"Failed to decompress deflate request body: {exc}") from exc
+    elif encoding == "br":
+        try:
+            import brotli
+
+            raw = brotli.decompress(raw)
+        except ImportError:
+            raise ValueError(
+                "Request body is brotli-compressed but the 'brotli' package is not installed."
+            ) from None
+        except Exception as exc:
+            raise ValueError(f"Failed to decompress brotli request body: {exc}") from exc
+    elif encoding and encoding != "identity":
+        raise ValueError(f"Unsupported Content-Encoding: {encoding}")
+
+    # Decode and parse JSON
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ValueError(f"Request body is not valid UTF-8 (possibly compressed?): {exc}") from exc
+
+    result: dict[str, Any] = json.loads(text)
+    return result

headroom/proxy/prometheus_metrics.py

@@ -0,0 +1,312 @@
+"""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
+from collections import defaultdict
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from headroom.proxy.cost import CostTracker
+
+from headroom.proxy.savings_tracker import SavingsTracker
+
+logger = logging.getLogger("headroom.proxy")
+
+
+class PrometheusMetrics:
+    """Prometheus-compatible metrics."""
+
+    def __init__(
+        self,
+        savings_tracker: SavingsTracker | None = None,
+        cost_tracker: CostTracker | None = None,
+    ):
+        self.requests_total = 0
+        self.requests_by_provider: dict[str, int] = defaultdict(int)
+        self.requests_by_model: dict[str, int] = defaultdict(int)
+        self.requests_cached = 0
+        self.requests_rate_limited = 0
+        self.requests_failed = 0
+
+        self.tokens_input_total = 0
+        self.tokens_output_total = 0
+        self.tokens_saved_total = 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)
+
+        # Aggregate waste signals
+        self.waste_signals_total: 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,
+                "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
+
+        # Cumulative savings history (timestamp → cumulative tokens saved)
+        self.savings_history: list[tuple[str, int]] = []
+        self.savings_tracker = savings_tracker or SavingsTracker()
+        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()
+
+    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
+
+    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,
+        uncached_input_tokens: int = 0,
+    ):
+        """Record metrics for a request."""
+        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
+
+            # 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["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,
+                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,
+            )
+
+    async def record_rate_limited(self):
+        async with self._lock:
+            self.requests_rate_limited += 1
+
+    async def record_failed(self):
+        async with self._lock:
+            self.requests_failed += 1
+
+    async def export(self) -> str:
+        """Export metrics in Prometheus format."""
+        async with self._lock:
+            lines = [
+                "# HELP headroom_requests_total Total number of requests",
+                "# TYPE headroom_requests_total counter",
+                f"headroom_requests_total {self.requests_total}",
+                "",
+                "# HELP headroom_requests_cached_total Cached request count",
+                "# TYPE headroom_requests_cached_total counter",
+                f"headroom_requests_cached_total {self.requests_cached}",
+                "",
+                "# HELP headroom_requests_rate_limited_total Rate limited requests",
+                "# TYPE headroom_requests_rate_limited_total counter",
+                f"headroom_requests_rate_limited_total {self.requests_rate_limited}",
+                "",
+                "# HELP headroom_requests_failed_total Failed requests",
+                "# TYPE headroom_requests_failed_total counter",
+                f"headroom_requests_failed_total {self.requests_failed}",
+                "",
+                "# HELP headroom_tokens_input_total Total input tokens",
+                "# TYPE headroom_tokens_input_total counter",
+                f"headroom_tokens_input_total {self.tokens_input_total}",
+                "",
+                "# HELP headroom_tokens_output_total Total output tokens",
+                "# TYPE headroom_tokens_output_total counter",
+                f"headroom_tokens_output_total {self.tokens_output_total}",
+                "",
+                "# HELP headroom_tokens_saved_total Tokens saved by optimization",
+                "# TYPE headroom_tokens_saved_total counter",
+                f"headroom_tokens_saved_total {self.tokens_saved_total}",
+                "",
+                "# HELP headroom_latency_ms_sum Sum of request latencies",
+                "# TYPE headroom_latency_ms_sum counter",
+                f"headroom_latency_ms_sum {self.latency_sum_ms:.2f}",
+            ]
+
+            # Per-provider metrics
+            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}')
+
+            # Per-model metrics
+            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}')
+
+            return "\n".join(lines)

headroom/proxy/rate_limiter.py

@@ -0,0 +1,101 @@
+"""Token bucket rate limiter for the Headroom proxy.
+
+Rate limits requests and token usage per API key or IP address.
+
+Extracted from server.py for maintainability.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from collections import defaultdict
+
+from headroom.proxy.models import RateLimitState
+
+logger = logging.getLogger("headroom.proxy")
+
+# Maximum rate limiter buckets (prevents DoS via spoofed API keys)
+MAX_RATE_LIMITER_BUCKETS = 1000
+
+
+class TokenBucketRateLimiter:
+    """Token bucket rate limiter for requests and tokens."""
+
+    def __init__(
+        self,
+        requests_per_minute: int = 60,
+        tokens_per_minute: int = 100000,
+    ):
+        self.requests_per_minute = requests_per_minute
+        self.tokens_per_minute = tokens_per_minute
+
+        # Per-key buckets (key = API key or IP)
+        self._request_buckets: dict[str, RateLimitState] = defaultdict(
+            lambda: RateLimitState(tokens=requests_per_minute, last_update=time.time())
+        )
+        self._token_buckets: dict[str, RateLimitState] = defaultdict(
+            lambda: RateLimitState(tokens=tokens_per_minute, last_update=time.time())
+        )
+        self._lock = asyncio.Lock()
+
+    async def _cleanup_stale_buckets(self) -> None:
+        """Remove buckets that haven't been used in the last 10 minutes."""
+        now = time.time()
+        stale_threshold = now - 600  # 10 minutes
+        stale_keys = [
+            k for k, v in self._request_buckets.items() if v.last_update < stale_threshold
+        ]
+        for k in stale_keys:
+            del self._request_buckets[k]
+            self._token_buckets.pop(k, None)
+        if stale_keys:
+            logger.debug(f"Cleaned up {len(stale_keys)} stale rate limiter buckets")
+
+    def _refill(self, state: RateLimitState, rate_per_minute: float) -> float:
+        """Refill bucket based on elapsed time."""
+        now = time.time()
+        elapsed = now - state.last_update
+        refill = elapsed * (rate_per_minute / 60.0)
+        state.tokens = min(rate_per_minute, state.tokens + refill)
+        state.last_update = now
+        return state.tokens
+
+    async def check_request(self, key: str = "default") -> tuple[bool, float]:
+        """Check if request is allowed. Returns (allowed, wait_seconds)."""
+        async with self._lock:
+            # Prevent unbounded bucket growth from spoofed keys
+            if len(self._request_buckets) > MAX_RATE_LIMITER_BUCKETS:
+                await self._cleanup_stale_buckets()
+            state = self._request_buckets[key]
+            available = self._refill(state, self.requests_per_minute)
+
+            if available >= 1:
+                state.tokens -= 1
+                return True, 0
+
+            wait_seconds = (1 - available) * (60.0 / self.requests_per_minute)
+            return False, wait_seconds
+
+    async def check_tokens(self, key: str, token_count: int) -> tuple[bool, float]:
+        """Check if token usage is allowed."""
+        async with self._lock:
+            state = self._token_buckets[key]
+            available = self._refill(state, self.tokens_per_minute)
+
+            if available >= token_count:
+                state.tokens -= token_count
+                return True, 0
+
+            wait_seconds = (token_count - available) * (60.0 / self.tokens_per_minute)
+            return False, wait_seconds
+
+    async def stats(self) -> dict:
+        """Get rate limiter statistics."""
+        async with self._lock:
+            return {
+                "requests_per_minute": self.requests_per_minute,
+                "tokens_per_minute": self.tokens_per_minute,
+                "active_keys": len(self._request_buckets),
+            }

headroom/proxy/request_logger.py

@@ -0,0 +1,108 @@
+"""Request logger for the Headroom proxy.
+
+Logs requests to an in-memory deque and optionally to a JSONL file.
+
+Extracted from server.py for maintainability.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from collections import deque
+from dataclasses import asdict
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ..memory.tracker import ComponentStats
+
+from headroom.proxy.models import RequestLog
+
+
+class RequestLogger:
+    """Log requests to JSONL file.
+
+    Uses a deque with max 10,000 entries to prevent unbounded memory growth.
+    """
+
+    MAX_LOG_ENTRIES = 10_000
+
+    def __init__(self, log_file: str | None = None, log_full_messages: bool = False):
+        self.log_file = Path(log_file) if log_file else None
+        self.log_full_messages = log_full_messages
+        # Use deque with maxlen for automatic FIFO eviction
+        self._logs: deque[RequestLog] = deque(maxlen=self.MAX_LOG_ENTRIES)
+
+        if self.log_file:
+            self.log_file.parent.mkdir(parents=True, exist_ok=True)
+
+    def log(self, entry: RequestLog):
+        """Log a request. Oldest entries are automatically removed when limit reached."""
+        self._logs.append(entry)
+
+        if self.log_file:
+            with open(self.log_file, "a") as f:
+                log_dict = asdict(entry)
+                if not self.log_full_messages:
+                    log_dict.pop("request_messages", None)
+                    log_dict.pop("response_content", None)
+                f.write(json.dumps(log_dict) + "\n")
+
+    def get_recent(self, n: int = 100) -> list[dict]:
+        """Get recent log entries."""
+        # Convert deque to list for slicing (deque doesn't support slicing)
+        entries = list(self._logs)[-n:]
+        return [
+            {
+                k: v
+                for k, v in asdict(e).items()
+                if k not in ("request_messages", "response_content")
+            }
+            for e in entries
+        ]
+
+    def stats(self) -> dict:
+        """Get logging statistics."""
+        return {
+            "total_logged": len(self._logs),
+            "log_file": str(self.log_file) if self.log_file else None,
+        }
+
+    def get_memory_stats(self) -> ComponentStats:
+        """Get memory statistics for the MemoryTracker.
+
+        Returns:
+            ComponentStats with current memory usage.
+        """
+        from ..memory.tracker import ComponentStats
+
+        # Calculate size
+        size_bytes = sys.getsizeof(self._logs)
+
+        for log_entry in self._logs:
+            size_bytes += sys.getsizeof(log_entry)
+            # Add string fields
+            if log_entry.request_id:
+                size_bytes += len(log_entry.request_id)
+            if log_entry.provider:
+                size_bytes += len(log_entry.provider)
+            if log_entry.model:
+                size_bytes += len(log_entry.model)
+            if log_entry.error:
+                size_bytes += len(log_entry.error)
+            # Messages and response can be large
+            if log_entry.request_messages:
+                size_bytes += sys.getsizeof(log_entry.request_messages)
+            if log_entry.response_content:
+                size_bytes += len(log_entry.response_content)
+
+        return ComponentStats(
+            name="request_logger",
+            entry_count=len(self._logs),
+            size_bytes=size_bytes,
+            budget_bytes=None,
+            hits=0,
+            misses=0,
+            evictions=0,
+        )

headroom/proxy/semantic_cache.py

@@ -0,0 +1,142 @@
+"""Semantic cache for the Headroom proxy.
+
+Simple semantic cache based on message content hash with LRU eviction.
+
+Extracted from server.py for maintainability.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import sys
+from collections import OrderedDict
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ..memory.tracker import ComponentStats
+
+from headroom.proxy.models import CacheEntry
+
+
+class SemanticCache:
+    """Simple semantic cache based on message content hash.
+
+    Uses OrderedDict for O(1) LRU eviction instead of list with O(n) pop(0).
+    """
+
+    def __init__(self, max_entries: int = 1000, ttl_seconds: int = 3600):
+        self.max_entries = max_entries
+        self.ttl_seconds = ttl_seconds
+        # OrderedDict maintains insertion order and supports O(1) move_to_end/popitem
+        self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
+        self._lock = asyncio.Lock()
+
+    def _compute_key(self, messages: list[dict], model: str) -> str:
+        """Compute cache key from messages and model."""
+        # Normalize messages for consistent hashing
+        normalized = json.dumps(
+            {
+                "model": model,
+                "messages": messages,
+            },
+            sort_keys=True,
+        )
+        return hashlib.sha256(normalized.encode()).hexdigest()[:32]
+
+    async def get(self, messages: list[dict], model: str) -> CacheEntry | None:
+        """Get cached response if exists and not expired."""
+        key = self._compute_key(messages, model)
+        async with self._lock:
+            entry = self._cache.get(key)
+
+            if entry is None:
+                return None
+
+            # Check expiration
+            age = (datetime.now() - entry.created_at).total_seconds()
+            if age > entry.ttl_seconds:
+                del self._cache[key]
+                return None
+
+            entry.hit_count += 1
+            # Move to end for LRU (O(1) operation)
+            self._cache.move_to_end(key)
+            return entry
+
+    async def set(
+        self,
+        messages: list[dict],
+        model: str,
+        response_body: bytes,
+        response_headers: dict[str, str],
+        tokens_saved: int = 0,
+    ):
+        """Cache a response."""
+        key = self._compute_key(messages, model)
+
+        async with self._lock:
+            # If key already exists, remove it first to update position
+            if key in self._cache:
+                del self._cache[key]
+
+            # Evict oldest entries if at capacity (LRU) - O(1) with popitem
+            while len(self._cache) >= self.max_entries:
+                self._cache.popitem(last=False)  # Remove oldest (first) entry
+
+            self._cache[key] = CacheEntry(
+                response_body=response_body,
+                response_headers=response_headers,
+                created_at=datetime.now(),
+                ttl_seconds=self.ttl_seconds,
+                tokens_saved_per_hit=tokens_saved,
+            )
+
+    async def stats(self) -> dict:
+        """Get cache statistics."""
+        async with self._lock:
+            total_hits = sum(e.hit_count for e in self._cache.values())
+            return {
+                "entries": len(self._cache),
+                "max_entries": self.max_entries,
+                "total_hits": total_hits,
+                "ttl_seconds": self.ttl_seconds,
+            }
+
+    async def clear(self):
+        """Clear all cache entries."""
+        async with self._lock:
+            self._cache.clear()
+
+    def get_memory_stats(self) -> ComponentStats:
+        """Get memory statistics for the MemoryTracker.
+
+        Returns:
+            ComponentStats with current memory usage.
+        """
+        from ..memory.tracker import ComponentStats
+
+        # Calculate size - this is sync but we access _cache directly
+        # Note: This is a rough estimate, not perfectly accurate under async load
+        size_bytes = sys.getsizeof(self._cache)
+        total_hits = 0
+
+        for entry in self._cache.values():
+            size_bytes += sys.getsizeof(entry)
+            size_bytes += len(entry.response_body)
+            size_bytes += sys.getsizeof(entry.response_headers)
+            for k, v in entry.response_headers.items():
+                size_bytes += len(k) + len(v)
+            total_hits += entry.hit_count
+
+        return ComponentStats(
+            name="semantic_cache",
+            entry_count=len(self._cache),
+            size_bytes=size_bytes,
+            budget_bytes=None,
+            hits=total_hits,
+            misses=0,  # Would need to track this separately
+            evictions=0,  # Would need to track this separately
+        )

headroom/proxy/server.py

@@ -25,22 +25,19 @@ from __future__ import annotations
 
 import argparse
 import asyncio
-import hashlib
 import json
 import logging
 import os
 import random
 import sys
 import time
-from collections import OrderedDict, defaultdict, deque
-from dataclasses import asdict
-from datetime import datetime, timedelta
+from datetime import datetime
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal
 
 if TYPE_CHECKING:
     from ..cache.compression_cache import CompressionCache
-    from ..memory.tracker import ComponentStats, MemoryTracker
+    from ..memory.tracker import MemoryTracker
 
 import contextlib
 
@@ -88,11 +85,37 @@ from headroom.config import (
 )
 from headroom.dashboard import get_dashboard_html
 from headroom.providers import AnthropicProvider, OpenAIProvider
+
+# =============================================================================
+# Extracted modules (re-exported for backward compatibility)
+# =============================================================================
+from headroom.proxy.cost import (
+    _CACHE_ECONOMICS,  # noqa: F401
+    CostTracker,  # noqa: F401
+    _summarize_transforms,  # noqa: F401
+)
+from headroom.proxy.cost import build_prefix_cache_stats as _build_prefix_cache_stats  # noqa: F401
+from headroom.proxy.cost import build_session_summary as _build_session_summary  # noqa: F401
+from headroom.proxy.cost import merge_cost_stats as _merge_cost_stats  # noqa: F401
+from headroom.proxy.helpers import (
+    COMPRESSION_TIMEOUT_SECONDS,  # noqa: F401
+    MAX_COMPRESSION_CACHE_SESSIONS,  # noqa: F401
+    MAX_MESSAGE_ARRAY_LENGTH,  # noqa: F401
+    MAX_REQUEST_BODY_SIZE,  # noqa: F401
+    MAX_SSE_BUFFER_SIZE,  # noqa: F401
+    _get_image_compressor,  # noqa: F401
+    _get_rtk_stats,  # noqa: F401
+    _read_request_json,  # noqa: F401
+    _setup_file_logging,  # noqa: F401
+)
 from headroom.proxy.memory_handler import MemoryConfig, MemoryHandler
 
 # Data models (extracted to headroom/proxy/models.py for maintainability)
 from headroom.proxy.models import CacheEntry, ProxyConfig, RateLimitState, RequestLog  # noqa: F401
-from headroom.proxy.savings_tracker import SavingsTracker
+from headroom.proxy.prometheus_metrics import PrometheusMetrics  # noqa: F401
+from headroom.proxy.rate_limiter import TokenBucketRateLimiter  # noqa: F401
+from headroom.proxy.request_logger import RequestLogger  # noqa: F401
+from headroom.proxy.semantic_cache import SemanticCache  # noqa: F401
 from headroom.telemetry import get_telemetry_collector
 from headroom.telemetry.toin import get_toin
 from headroom.tokenizers import get_tokenizer
@@ -111,33 +134,6 @@ from headroom.transforms import (
 )
 from headroom.utils import extract_user_query
 
-# Image compression (lazy-loaded to avoid heavy dependencies at startup)
-_image_compressor = None
-
-
-def _get_image_compressor():
-    """Lazy load image compressor to avoid startup overhead."""
-    global _image_compressor
-    if _image_compressor is None:
-        try:
-            from headroom.image import ImageCompressor
-
-            _image_compressor = ImageCompressor()
-            logger.info("Image compression enabled (model: chopratejas/technique-router)")
-        except ImportError as e:
-            logger.warning(f"Image compression not available: {e}")
-            _image_compressor = False  # Mark as unavailable
-    return _image_compressor if _image_compressor else None
-
-
-# Try to import LiteLLM for pricing
-try:
-    import litellm
-
-    LITELLM_AVAILABLE = True
-except ImportError:
-    LITELLM_AVAILABLE = False
-
 logging.basicConfig(
     level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 )
@@ -147,464 +143,13 @@ logger = logging.getLogger("headroom.proxy")
 _HEADROOM_LOG_DIR = Path.home() / ".headroom" / "logs"
 
 
-def _setup_file_logging() -> None:
-    """Add a RotatingFileHandler to the headroom root logger.
-
-    Writes to ~/.headroom/logs/proxy.log with automatic rotation:
-    - Rotates at 10 MB
-    - Keeps 5 backups (~50 MB max)
-    """
-    from logging.handlers import RotatingFileHandler
-
-    try:
-        _HEADROOM_LOG_DIR.mkdir(parents=True, exist_ok=True)
-        log_path = _HEADROOM_LOG_DIR / "proxy.log"
-        handler = RotatingFileHandler(
-            log_path,
-            maxBytes=10 * 1024 * 1024,  # 10 MB
-            backupCount=5,
-            encoding="utf-8",
-        )
-        handler.setLevel(logging.INFO)
-        handler.setFormatter(
-            logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
-        )
-        # Attach to the headroom root logger so all sub-loggers are captured
-        logging.getLogger("headroom").addHandler(handler)
-    except OSError:
-        # Non-fatal: can't write logs (read-only fs, permissions, etc.)
-        pass
-
-
 _setup_file_logging()
 
 
-def _summarize_transforms(transforms: list[str]) -> str:
-    """Collapse repeated transforms into counted summary.
-
-    e.g. ['router:excluded:tool', 'router:excluded:tool', 'read_lifecycle:stale']
-      → 'router:excluded:tool*2 read_lifecycle:stale'
-    """
-    if not transforms:
-        return "none"
-    counts: dict[str, int] = {}
-    for t in transforms:
-        counts[t] = counts.get(t, 0) + 1
-    parts = [f"{k}*{v}" if v > 1 else k for k, v in counts.items()]
-    return " ".join(parts)
-
-
-# Provider-specific cache discount multipliers (what fraction of input price)
-# Used to calculate dollar savings from prefix caching
-_CACHE_ECONOMICS = {
-    "anthropic": {
-        "read_multiplier": 0.1,
-        "write_multiplier": 1.25,
-        "label": "Explicit breakpoints, 5-min TTL",
-    },
-    "openai": {
-        "read_multiplier": 0.5,
-        "write_multiplier": 1.0,
-        "label": "Automatic, no TTL control",
-    },
-    "gemini": {
-        "read_multiplier": 0.1,
-        "write_multiplier": 1.0,
-        "label": "Explicit cachedContent, configurable TTL",
-    },
-    "bedrock": {
-        "read_multiplier": 0.1,
-        "write_multiplier": 1.25,
-        "label": "Same as Anthropic (Bedrock)",
-    },
-}
-
-
-def _get_rtk_stats() -> dict[str, Any] | None:
-    """Get rtk (Rust Token Killer) savings stats if rtk is installed.
-
-    Reads from rtk's tracking database via `rtk gain --format json`.
-    Returns None if rtk is not installed.
-    """
-    import shutil
-    import subprocess as _sp
-
-    rtk_bin = shutil.which("rtk")
-    if not rtk_bin:
-        # Check headroom-managed install
-        rtk_managed = Path.home() / ".headroom" / "bin" / "rtk"
-        if rtk_managed.exists():
-            rtk_bin = str(rtk_managed)
-        else:
-            return None
-
-    try:
-        result = _sp.run(
-            [rtk_bin, "gain", "--format", "json"],
-            capture_output=True,
-            text=True,
-            timeout=5,
-        )
-        if result.returncode == 0 and result.stdout.strip():
-            data = json.loads(result.stdout)
-            summary = data.get("summary", {})
-            return {
-                "installed": True,
-                "total_commands": summary.get("total_commands", 0),
-                "tokens_saved": summary.get("total_saved", 0),
-                "avg_savings_pct": summary.get("avg_savings_pct", 0.0),
-            }
-    except Exception:
-        pass
-
-    return {"installed": True, "total_commands": 0, "tokens_saved": 0, "avg_savings_pct": 0.0}
-
-
-def _build_prefix_cache_stats(
-    metrics: PrometheusMetrics,
-    cost_tracker: CostTracker | None,
-) -> dict:
-    """Build provider-aware prefix cache statistics for the dashboard."""
-    by_provider = {}
-    totals = {
-        "cache_read_tokens": 0,
-        "cache_write_tokens": 0,
-        "requests": 0,
-        "hit_requests": 0,
-        "bust_count": 0,
-        "bust_write_tokens": 0,
-        "savings_usd": 0.0,
-        "write_premium_usd": 0.0,
-    }
-
-    for provider, pc in metrics.cache_by_provider.items():
-        if pc["requests"] == 0:
-            continue
-
-        econ = _CACHE_ECONOMICS.get(provider, _CACHE_ECONOMICS["anthropic"])
-        read_mult: float = econ["read_multiplier"]  # type: ignore[assignment]
-        write_mult: float = econ["write_multiplier"]  # type: ignore[assignment]
-
-        # Get the base input price per token for the most-used model on this provider
-        input_price_per_token = None
-        if cost_tracker:
-            for model_name in cost_tracker._tokens_sent_by_model:
-                # Match model to provider
-                _openai_prefixes = ("gpt", "o1", "o3", "o4")
-                is_match = (
-                    (provider == "anthropic" and "claude" in model_name)
-                    or (provider == "openai" and any(p in model_name for p in _openai_prefixes))
-                    or (provider == "gemini" and "gemini" in model_name)
-                    or (provider == "bedrock" and "claude" in model_name)
-                )
-                if is_match:
-                    price_per_1m = cost_tracker._get_list_price(model_name)
-                    if price_per_1m:
-                        input_price_per_token = price_per_1m / 1_000_000
-                        break
-
-        # Calculate savings:
-        # Cache reads save (1.0 - read_mult) per token vs uncached input price.
-        # Cache write premium is NOT deducted — it's baseline cost that the
-        # client (e.g. Claude Code) pays regardless of Headroom. We track it
-        # for observability but don't penalise our savings number.
-        read_tokens: int = pc["cache_read_tokens"]  # type: ignore[assignment]
-        write_tokens: int = pc["cache_write_tokens"]  # type: ignore[assignment]
-        savings_usd = 0.0
-        write_premium_usd = 0.0
-
-        if input_price_per_token:
-            # Savings from reads: tokens * price * (1.0 - read_multiplier)
-            savings_usd = read_tokens * input_price_per_token * (1.0 - read_mult)
-            # Write premium (observability only — not subtracted from savings)
-            if write_mult > 1.0:
-                write_premium_usd = write_tokens * input_price_per_token * (write_mult - 1.0)
-
-        hit_rate = round(pc["hit_requests"] / pc["requests"] * 100, 1) if pc["requests"] > 0 else 0
-
-        provider_stats = {
-            "cache_read_tokens": read_tokens,
-            "cache_write_tokens": write_tokens,
-            "requests": pc["requests"],
-            "hit_requests": pc["hit_requests"],
-            "hit_rate": hit_rate,
-            "bust_count": pc["bust_count"],
-            "bust_write_tokens": pc["bust_write_tokens"],
-            "read_discount": f"{(1.0 - read_mult) * 100:.0f}%",
-            "write_premium": f"{(write_mult - 1.0) * 100:.0f}%" if write_mult > 1.0 else "none",
-            "savings_usd": round(savings_usd, 4),
-            "write_premium_usd": round(write_premium_usd, 4),
-            "net_savings_usd": round(savings_usd, 4),
-            "label": str(econ["label"]),
-        }
-        by_provider[provider] = provider_stats
-
-        # Accumulate totals
-        totals["cache_read_tokens"] += read_tokens
-        totals["cache_write_tokens"] += write_tokens
-        totals["requests"] += pc["requests"]
-        totals["hit_requests"] += pc["hit_requests"]
-        totals["bust_count"] += pc["bust_count"]
-        totals["bust_write_tokens"] += pc["bust_write_tokens"]
-        totals["savings_usd"] += savings_usd
-        totals["write_premium_usd"] += write_premium_usd
-
-    totals["net_savings_usd"] = round(totals["savings_usd"], 4)
-    totals["savings_usd"] = round(totals["savings_usd"], 4)
-    totals["write_premium_usd"] = round(totals["write_premium_usd"], 4)
-    totals["hit_rate"] = (
-        round(totals["hit_requests"] / totals["requests"] * 100, 1) if totals["requests"] > 0 else 0
-    )
-
-    return {
-        "by_provider": by_provider,
-        "totals": totals,
-        "prefix_freeze": {
-            "busts_avoided": metrics.prefix_freeze_busts_avoided,
-            "tokens_preserved": metrics.prefix_freeze_tokens_preserved,
-            "compression_foregone_tokens": metrics.prefix_freeze_compression_foregone,
-            "net_benefit_tokens": (
-                metrics.prefix_freeze_tokens_preserved - metrics.prefix_freeze_compression_foregone
-            ),
-        },
-        "attribution": (
-            "Prefix caching is performed by the LLM provider (Anthropic, OpenAI). "
-            "Headroom reports cache stats as observed from API responses. "
-            "CacheAligner and prefix freeze improve cache hit rates by stabilizing "
-            "the message prefix, but baseline caching happens without Headroom."
-        ),
-    }
-
-
-def _merge_cost_stats(
-    cost_stats: dict | None,
-    cache_stats: dict,
-    cli_tokens_avoided: int = 0,
-) -> dict | None:
-    """Merge compression, cache, and CLI savings into cost stats.
-
-    Each savings layer is reported separately with its own scope:
-    - savings_usd: compression savings at model list price (monotonic)
-    - cache_savings_usd: prefix cache discount from provider (separate)
-    - cli_tokens_avoided: tokens filtered by rtk (token count only, no $ estimate)
-
-    The hero metric (savings_usd) is ONLY compression savings priced at
-    the model's published input rate. Cache and CLI are shown separately.
-    This avoids the non-monotonic moving-average repricing bug (#83).
-    """
-    if cost_stats is None:
-        return None
-
-    cache_net = cache_stats.get("totals", {}).get("net_savings_usd", 0.0)
-    compression_savings = cost_stats.get("savings_usd", 0.0)
-
-    return {
-        **cost_stats,
-        "savings_usd": round(compression_savings, 4),
-        "compression_savings_usd": round(compression_savings, 4),
-        "cache_savings_usd": round(cache_net, 4),
-        "cli_tokens_avoided": cli_tokens_avoided,
-    }
-
-
-def _build_session_summary(
-    proxy: HeadroomProxy,
-    metrics: Any,
-    prefix_cache_stats: dict,
-    cli_tokens_avoided: int,
-    total_tokens_before: int,
-) -> dict[str, Any]:
-    """Build a human-readable session summary from metrics and request logs.
-
-    This is the headline view users see first in /stats — designed to answer
-    "is Headroom working?" at a glance.
-    """
-    # Analyze per-request compression from the logger
-    compressed_requests: list[dict] = []
-    uncompressed_reasons: dict[str, int] = {
-        "prefix_frozen": 0,
-        "too_small": 0,
-        "passthrough": 0,
-        "no_compressible_content": 0,
-    }
-
-    if proxy.logger:
-        for entry in proxy.logger._logs:
-            if entry.model and "count_tokens" in entry.model:
-                uncompressed_reasons["passthrough"] += 1
-                continue
-            if entry.tokens_saved > 0 and entry.savings_percent > 0:
-                compressed_requests.append(
-                    {
-                        "savings_pct": round(entry.savings_percent, 1),
-                        "tokens_saved": entry.tokens_saved,
-                        "original": entry.input_tokens_original,
-                        "optimized": entry.input_tokens_optimized,
-                    }
-                )
-            elif entry.input_tokens_original > 0:
-                # Categorize why it wasn't compressed
-                transforms = entry.transforms_applied or []
-                if not transforms:
-                    # Pipeline returned unchanged — likely all frozen
-                    uncompressed_reasons["prefix_frozen"] += 1
-                elif all("excluded" in t or "protected" in t for t in transforms):
-                    uncompressed_reasons["no_compressible_content"] += 1
-                elif entry.input_tokens_original < 500:
-                    uncompressed_reasons["too_small"] += 1
-                else:
-                    uncompressed_reasons["prefix_frozen"] += 1
-
-    # Compute compression stats for requests that DID compress
-    avg_compression = 0.0
-    best_compression = 0.0
-    best_detail = ""
-    if compressed_requests:
-        avg_compression = round(
-            sum(r["savings_pct"] for r in compressed_requests) / len(compressed_requests),
-            1,
-        )
-        best = max(compressed_requests, key=lambda r: r["savings_pct"])
-        best_compression = best["savings_pct"]
-        best_detail = f"{best['original']:,} → {best['optimized']:,} tokens"
-
-    # Cost summary — savings_usd is compression savings at model list price (monotonic)
-    cost_stats = proxy.cost_tracker.stats() if proxy.cost_tracker else {}
-    cost_with = cost_stats.get("cost_with_headroom_usd", 0.0)
-    compression_savings = cost_stats.get("savings_usd", 0.0)
-    cache_net = prefix_cache_stats.get("totals", {}).get("net_savings_usd", 0.0)
-    total_saved_usd = round(compression_savings, 2)
-    cost_without = cost_with + compression_savings
-    savings_pct_cost = round(total_saved_usd / cost_without * 100, 1) if cost_without > 0 else 0.0
-
-    # Primary models used
-    models = dict(metrics.requests_by_model)
-    primary_model = max(models, key=lambda k: models[k]) if models else "unknown"
-    api_requests = sum(v for k, v in models.items() if "count_tokens" not in k)
-
-    # Build the summary
-    summary: dict[str, Any] = {
-        "mode": proxy.config.mode,
-        "api_requests": api_requests,
-        "primary_model": primary_model,
-        "compression": {
-            "requests_compressed": len(compressed_requests),
-            "avg_compression_pct": avg_compression,
-            "best_compression_pct": best_compression,
-            "best_detail": best_detail,
-            "total_tokens_removed": metrics.tokens_saved_total,
-        },
-        "uncompressed_requests": {k: v for k, v in uncompressed_reasons.items() if v > 0},
-        "cost": {
-            "without_headroom_usd": round(cost_without, 2),
-            "with_headroom_usd": round(cost_with, 2),
-            "total_saved_usd": total_saved_usd,
-            "savings_pct": savings_pct_cost,
-            "breakdown": {
-                "cache_savings_usd": round(cache_net, 2),
-                "compression_savings_usd": round(compression_savings, 2),
-            },
-        },
-    }
-
-    # Add tip if token_headroom mode would help
-    if proxy.config.mode == "cost_savings" and uncompressed_reasons["prefix_frozen"] > 10:
-        summary["tip"] = (
-            "Most requests are prefix-frozen. Set HEADROOM_MODE=token_headroom "
-            "to compress frozen messages and extend your session by ~25-35%."
-        )
-
-    return summary
-
-
-# Maximum request body size (100MB - increased to support image-heavy requests)
-MAX_REQUEST_BODY_SIZE = 100 * 1024 * 1024
-
-# Maximum SSE buffer size (10MB - prevents memory exhaustion from malformed streams)
-MAX_SSE_BUFFER_SIZE = 10 * 1024 * 1024
-
-# Maximum message array length (prevents DoS from deeply nested payloads)
-MAX_MESSAGE_ARRAY_LENGTH = 10000
-
-
-async def _read_request_json(request: Request) -> dict[str, Any]:
-    """Read and parse JSON from a request, handling compressed bodies.
-
-    Clients like OpenAI Codex may send zstd, gzip, or deflate-compressed
-    request bodies.  Starlette's ``request.json()`` does not decompress
-    automatically, causing a UnicodeDecodeError on compressed bytes.
-
-    This helper inspects ``Content-Encoding``, decompresses if needed,
-    then JSON-decodes the result.  It raises ``ValueError`` on any
-    decompression or parse failure so callers can return a clean 400.
-    """
-    encoding = (request.headers.get("content-encoding") or "").lower().strip()
-    raw = await request.body()
-
-    if encoding in ("zstd", "zstandard"):
-        try:
-            import zstandard
-
-            dctx = zstandard.ZstdDecompressor()
-            # Use stream_reader for streaming zstd frames (no content size in header).
-            # Plain decompress() fails when the frame header omits the size, which
-            # is common with clients like OpenAI Codex.
-            reader = dctx.stream_reader(raw)
-            raw = reader.read()
-            reader.close()
-        except ImportError:
-            raise ValueError(
-                "Request body is zstd-compressed but the 'zstandard' package is not installed. "
-                "Install it with: pip install zstandard"
-            ) from None
-        except Exception as exc:
-            raise ValueError(f"Failed to decompress zstd request body: {exc}") from exc
-    elif encoding == "gzip":
-        import gzip as _gzip
-
-        try:
-            raw = _gzip.decompress(raw)
-        except Exception as exc:
-            raise ValueError(f"Failed to decompress gzip request body: {exc}") from exc
-    elif encoding == "deflate":
-        import zlib
-
-        try:
-            raw = zlib.decompress(raw)
-        except Exception as exc:
-            raise ValueError(f"Failed to decompress deflate request body: {exc}") from exc
-    elif encoding == "br":
-        try:
-            import brotli
-
-            raw = brotli.decompress(raw)
-        except ImportError:
-            raise ValueError(
-                "Request body is brotli-compressed but the 'brotli' package is not installed."
-            ) from None
-        except Exception as exc:
-            raise ValueError(f"Failed to decompress brotli request body: {exc}") from exc
-    elif encoding and encoding != "identity":
-        raise ValueError(f"Unsupported Content-Encoding: {encoding}")
-
-    # Decode and parse JSON
-    try:
-        text = raw.decode("utf-8")
-    except UnicodeDecodeError as exc:
-        raise ValueError(f"Request body is not valid UTF-8 (possibly compressed?): {exc}") from exc
-
-    result: dict[str, Any] = json.loads(text)
-    return result
-
-
-# Maximum compression cache sessions (prevents unbounded memory growth)
-MAX_COMPRESSION_CACHE_SESSIONS = 500
-
 # Maximum rate limiter buckets (prevents DoS via spoofed API keys)
 MAX_RATE_LIMITER_BUCKETS = 1000
 
 # Compression pipeline timeout in seconds
-COMPRESSION_TIMEOUT_SECONDS = 30
 
 
 # =============================================================================
@@ -612,917 +157,6 @@ COMPRESSION_TIMEOUT_SECONDS = 30
 # =============================================================================
 
 
-class SemanticCache:
-    """Simple semantic cache based on message content hash.
-
-    Uses OrderedDict for O(1) LRU eviction instead of list with O(n) pop(0).
-    """
-
-    def __init__(self, max_entries: int = 1000, ttl_seconds: int = 3600):
-        self.max_entries = max_entries
-        self.ttl_seconds = ttl_seconds
-        # OrderedDict maintains insertion order and supports O(1) move_to_end/popitem
-        self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
-        self._lock = asyncio.Lock()
-
-    def _compute_key(self, messages: list[dict], model: str) -> str:
-        """Compute cache key from messages and model."""
-        # Normalize messages for consistent hashing
-        normalized = json.dumps(
-            {
-                "model": model,
-                "messages": messages,
-            },
-            sort_keys=True,
-        )
-        return hashlib.sha256(normalized.encode()).hexdigest()[:32]
-
-    async def get(self, messages: list[dict], model: str) -> CacheEntry | None:
-        """Get cached response if exists and not expired."""
-        key = self._compute_key(messages, model)
-        async with self._lock:
-            entry = self._cache.get(key)
-
-            if entry is None:
-                return None
-
-            # Check expiration
-            age = (datetime.now() - entry.created_at).total_seconds()
-            if age > entry.ttl_seconds:
-                del self._cache[key]
-                return None
-
-            entry.hit_count += 1
-            # Move to end for LRU (O(1) operation)
-            self._cache.move_to_end(key)
-            return entry
-
-    async def set(
-        self,
-        messages: list[dict],
-        model: str,
-        response_body: bytes,
-        response_headers: dict[str, str],
-        tokens_saved: int = 0,
-    ):
-        """Cache a response."""
-        key = self._compute_key(messages, model)
-
-        async with self._lock:
-            # If key already exists, remove it first to update position
-            if key in self._cache:
-                del self._cache[key]
-
-            # Evict oldest entries if at capacity (LRU) - O(1) with popitem
-            while len(self._cache) >= self.max_entries:
-                self._cache.popitem(last=False)  # Remove oldest (first) entry
-
-            self._cache[key] = CacheEntry(
-                response_body=response_body,
-                response_headers=response_headers,
-                created_at=datetime.now(),
-                ttl_seconds=self.ttl_seconds,
-                tokens_saved_per_hit=tokens_saved,
-            )
-
-    async def stats(self) -> dict:
-        """Get cache statistics."""
-        async with self._lock:
-            total_hits = sum(e.hit_count for e in self._cache.values())
-            return {
-                "entries": len(self._cache),
-                "max_entries": self.max_entries,
-                "total_hits": total_hits,
-                "ttl_seconds": self.ttl_seconds,
-            }
-
-    async def clear(self):
-        """Clear all cache entries."""
-        async with self._lock:
-            self._cache.clear()
-
-    def get_memory_stats(self) -> ComponentStats:
-        """Get memory statistics for the MemoryTracker.
-
-        Returns:
-            ComponentStats with current memory usage.
-        """
-        from ..memory.tracker import ComponentStats
-
-        # Calculate size - this is sync but we access _cache directly
-        # Note: This is a rough estimate, not perfectly accurate under async load
-        size_bytes = sys.getsizeof(self._cache)
-        total_hits = 0
-
-        for entry in self._cache.values():
-            size_bytes += sys.getsizeof(entry)
-            size_bytes += len(entry.response_body)
-            size_bytes += sys.getsizeof(entry.response_headers)
-            for k, v in entry.response_headers.items():
-                size_bytes += len(k) + len(v)
-            total_hits += entry.hit_count
-
-        return ComponentStats(
-            name="semantic_cache",
-            entry_count=len(self._cache),
-            size_bytes=size_bytes,
-            budget_bytes=None,
-            hits=total_hits,
-            misses=0,  # Would need to track this separately
-            evictions=0,  # Would need to track this separately
-        )
-
-
-# =============================================================================
-# Rate Limiting
-# =============================================================================
-
-
-class TokenBucketRateLimiter:
-    """Token bucket rate limiter for requests and tokens."""
-
-    def __init__(
-        self,
-        requests_per_minute: int = 60,
-        tokens_per_minute: int = 100000,
-    ):
-        self.requests_per_minute = requests_per_minute
-        self.tokens_per_minute = tokens_per_minute
-
-        # Per-key buckets (key = API key or IP)
-        self._request_buckets: dict[str, RateLimitState] = defaultdict(
-            lambda: RateLimitState(tokens=requests_per_minute, last_update=time.time())
-        )
-        self._token_buckets: dict[str, RateLimitState] = defaultdict(
-            lambda: RateLimitState(tokens=tokens_per_minute, last_update=time.time())
-        )
-        self._lock = asyncio.Lock()
-
-    async def _cleanup_stale_buckets(self) -> None:
-        """Remove buckets that haven't been used in the last 10 minutes."""
-        now = time.time()
-        stale_threshold = now - 600  # 10 minutes
-        stale_keys = [
-            k for k, v in self._request_buckets.items() if v.last_update < stale_threshold
-        ]
-        for k in stale_keys:
-            del self._request_buckets[k]
-            self._token_buckets.pop(k, None)
-        if stale_keys:
-            logger.debug(f"Cleaned up {len(stale_keys)} stale rate limiter buckets")
-
-    def _refill(self, state: RateLimitState, rate_per_minute: float) -> float:
-        """Refill bucket based on elapsed time."""
-        now = time.time()
-        elapsed = now - state.last_update
-        refill = elapsed * (rate_per_minute / 60.0)
-        state.tokens = min(rate_per_minute, state.tokens + refill)
-        state.last_update = now
-        return state.tokens
-
-    async def check_request(self, key: str = "default") -> tuple[bool, float]:
-        """Check if request is allowed. Returns (allowed, wait_seconds)."""
-        async with self._lock:
-            # Prevent unbounded bucket growth from spoofed keys
-            if len(self._request_buckets) > MAX_RATE_LIMITER_BUCKETS:
-                await self._cleanup_stale_buckets()
-            state = self._request_buckets[key]
-            available = self._refill(state, self.requests_per_minute)
-
-            if available >= 1:
-                state.tokens -= 1
-                return True, 0
-
-            wait_seconds = (1 - available) * (60.0 / self.requests_per_minute)
-            return False, wait_seconds
-
-    async def check_tokens(self, key: str, token_count: int) -> tuple[bool, float]:
-        """Check if token usage is allowed."""
-        async with self._lock:
-            state = self._token_buckets[key]
-            available = self._refill(state, self.tokens_per_minute)
-
-            if available >= token_count:
-                state.tokens -= token_count
-                return True, 0
-
-            wait_seconds = (token_count - available) * (60.0 / self.tokens_per_minute)
-            return False, wait_seconds
-
-    async def stats(self) -> dict:
-        """Get rate limiter statistics."""
-        async with self._lock:
-            return {
-                "requests_per_minute": self.requests_per_minute,
-                "tokens_per_minute": self.tokens_per_minute,
-                "active_keys": len(self._request_buckets),
-            }
-
-
-# =============================================================================
-# Cost Tracking
-# =============================================================================
-
-
-class CostTracker:
-    """Track costs and enforce budgets.
-
-    Cost history is automatically pruned to prevent unbounded memory growth:
-    - Entries older than 24 hours are removed
-    - Maximum of 100,000 entries are kept
-
-    Uses LiteLLM's community-maintained pricing database for accurate costs.
-    See: https://github.com/BerriAI/litellm/blob/main/model_prices_and_context_window.json
-    """
-
-    MAX_COST_ENTRIES = 100_000
-    COST_RETENTION_HOURS = 24
-
-    def __init__(self, budget_limit_usd: float | None = None, budget_period: str = "daily"):
-        self.budget_limit_usd = budget_limit_usd
-        self.budget_period = budget_period
-
-        # Cost tracking - using deque for efficient left-side removal
-        self._costs: deque[tuple[datetime, float]] = deque(maxlen=self.MAX_COST_ENTRIES)
-        self._last_prune_time: datetime = datetime.now()
-
-        # Token savings per model (exact, no dollar estimation)
-        self._tokens_saved_by_model: dict[str, int] = {}
-        self._tokens_sent_by_model: dict[str, int] = {}
-        self._requests_by_model: dict[str, int] = {}
-
-        # API-reported cache breakdown per model (for accurate cost calculation)
-        self._api_cache_read_by_model: dict[str, int] = {}
-        self._api_cache_write_by_model: dict[str, int] = {}
-        self._api_uncached_by_model: dict[str, int] = {}
-
-    # Cache resolved model names to avoid repeated litellm lookups.
-    # This is critical: litellm.cost_per_token() is synchronous and can block
-    # the async event loop if it triggers I/O (lazy model info download).
-    _resolved_model_cache: dict[str, str] = {}
-
-    @classmethod
-    def _resolve_litellm_model(cls, model: str) -> str:
-        """Resolve model name to one LiteLLM recognizes, adding provider prefix if needed.
-
-        Results are cached per model name to avoid blocking the event loop
-        with repeated synchronous litellm lookups.
-        """
-        if model in cls._resolved_model_cache:
-            return cls._resolved_model_cache[model]
-
-        resolved = cls._resolve_litellm_model_uncached(model)
-        cls._resolved_model_cache[model] = resolved
-        return resolved
-
-    @staticmethod
-    def _resolve_litellm_model_uncached(model: str) -> str:
-        """Uncached resolution — called once per unique model name."""
-        if not LITELLM_AVAILABLE:
-            return model
-
-        # Try as-is first
-        try:
-            litellm.cost_per_token(model=model, prompt_tokens=1, completion_tokens=0)
-            return model
-        except Exception:
-            pass
-
-        # Try with provider prefix
-        prefixes = {
-            "claude-": "anthropic/",
-            "gpt-": "openai/",
-            "o1-": "openai/",
-            "o3-": "openai/",
-            "o4-": "openai/",
-            "gemini-": "google/",
-        }
-        for pattern, prefix in prefixes.items():
-            if model.startswith(pattern):
-                prefixed = f"{prefix}{model}"
-                try:
-                    litellm.cost_per_token(model=prefixed, prompt_tokens=1, completion_tokens=0)
-                    return prefixed
-                except Exception:
-                    break
-
-        return model
-
-    def estimate_cost(
-        self,
-        model: str,
-        input_tokens: int,
-        output_tokens: int,
-        cache_read_tokens: int = 0,
-        cache_write_tokens: int = 0,
-    ) -> float | None:
-        """Estimate cost in USD using LiteLLM's pricing database.
-
-        LiteLLM natively handles cache_read and cache_creation pricing
-        for all providers (Anthropic, OpenAI, Google, etc.) in a single call.
-
-        Args:
-            model: Model name for pricing lookup
-            input_tokens: Non-cached input tokens (excludes cache_read)
-            output_tokens: Output tokens
-            cache_read_tokens: Tokens served from cache (~10% of input rate)
-            cache_write_tokens: Tokens written to cache (~125% of input rate)
-        """
-        if not LITELLM_AVAILABLE:
-            logger.warning("LiteLLM not available - cannot calculate costs")
-            return None
-
-        try:
-            resolved_model = self._resolve_litellm_model(model)
-
-            # litellm.cost_per_token handles all token types natively:
-            # prompt_tokens at input rate, cache_read at ~10%, cache_creation at ~125%
-            input_cost, output_cost = litellm.cost_per_token(
-                model=resolved_model,
-                prompt_tokens=input_tokens,
-                completion_tokens=output_tokens,
-                cache_read_input_tokens=cache_read_tokens,
-                cache_creation_input_tokens=cache_write_tokens,
-            )
-
-            total_cost = input_cost + output_cost
-            return float(total_cost) if total_cost > 0 else None
-
-        except Exception as e:
-            logger.warning(f"Failed to get pricing for model {model}: {e}")
-            return None
-
-    def _prune_old_costs(self):
-        """Remove cost entries older than retention period.
-
-        Called periodically (every 5 minutes) to prevent unbounded memory growth.
-        The deque maxlen provides a hard cap, but time-based pruning keeps
-        memory usage proportional to actual traffic patterns.
-        """
-        now = datetime.now()
-        # Only prune every 5 minutes to avoid overhead
-        if (now - self._last_prune_time).total_seconds() < 300:
-            return
-
-        self._last_prune_time = now
-        cutoff = now - timedelta(hours=self.COST_RETENTION_HOURS)
-
-        # Remove entries from the left (oldest) while they're older than cutoff
-        while self._costs and self._costs[0][0] < cutoff:
-            self._costs.popleft()
-
-    def record_tokens(
-        self,
-        model: str,
-        tokens_saved: int,
-        tokens_sent: int,
-        cache_read_tokens: int = 0,
-        cache_write_tokens: int = 0,
-        uncached_tokens: int = 0,
-    ):
-        """Record token counts per model.
-
-        Args:
-            model: Model name.
-            tokens_saved: Tokens removed by compression (Headroom's count).
-            tokens_sent: Compressed message tokens sent (Headroom's count).
-            cache_read_tokens: Cache read tokens from API response usage.
-            cache_write_tokens: Cache write tokens from API response usage.
-            uncached_tokens: Non-cached input tokens from API response usage.
-        """
-        self._tokens_saved_by_model[model] = (
-            self._tokens_saved_by_model.get(model, 0) + tokens_saved
-        )
-        self._tokens_sent_by_model[model] = self._tokens_sent_by_model.get(model, 0) + tokens_sent
-        self._requests_by_model[model] = self._requests_by_model.get(model, 0) + 1
-        self._api_cache_read_by_model[model] = (
-            self._api_cache_read_by_model.get(model, 0) + cache_read_tokens
-        )
-        self._api_cache_write_by_model[model] = (
-            self._api_cache_write_by_model.get(model, 0) + cache_write_tokens
-        )
-        self._api_uncached_by_model[model] = (
-            self._api_uncached_by_model.get(model, 0) + uncached_tokens
-        )
-
-    def get_period_cost(self) -> float:
-        """Get cost for current budget period."""
-        now = datetime.now()
-
-        if self.budget_period == "hourly":
-            cutoff = now - timedelta(hours=1)
-        elif self.budget_period == "daily":
-            cutoff = now.replace(hour=0, minute=0, second=0, microsecond=0)
-        else:  # monthly
-            cutoff = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
-
-        return sum(cost for ts, cost in self._costs if ts >= cutoff)
-
-    def check_budget(self) -> tuple[bool, float]:
-        """Check if within budget. Returns (allowed, remaining)."""
-        if self.budget_limit_usd is None:
-            return True, float("inf")
-
-        period_cost = self.get_period_cost()
-        remaining = self.budget_limit_usd - period_cost
-        return remaining > 0, max(0, remaining)
-
-    def _get_list_price(self, model: str) -> float | None:
-        """Get list input price per 1M tokens for a model."""
-        if not LITELLM_AVAILABLE:
-            return None
-        try:
-            resolved = self._resolve_litellm_model(model)
-            info = litellm.model_cost.get(resolved, {})
-            cost_per_token = info.get("input_cost_per_token")
-            return cost_per_token * 1_000_000 if cost_per_token else None
-        except Exception:
-            return None
-
-    def _get_cache_prices(self, model: str) -> tuple[float, float, float] | None:
-        """Get per-token prices for cache read, cache write, and uncached input.
-
-        Returns (cache_read, cache_write, uncached) per-token costs, or None
-        if pricing is unavailable. Uses LiteLLM's native cache pricing data.
-        """
-        if not LITELLM_AVAILABLE:
-            return None
-        try:
-            resolved = self._resolve_litellm_model(model)
-            info = litellm.model_cost.get(resolved, {})
-            uncached = info.get("input_cost_per_token")
-            if not uncached:
-                return None
-            cache_read = info.get("cache_read_input_token_cost", uncached)
-            cache_write = info.get("cache_creation_input_token_cost", uncached)
-            return (cache_read, cache_write, uncached)
-        except Exception:
-            return None
-
-    def stats(self) -> dict:
-        """Get token statistics per model."""
-        per_model = {}
-        total_saved = 0
-        for model in sorted(self._tokens_saved_by_model.keys()):
-            saved = self._tokens_saved_by_model[model]
-            sent = self._tokens_sent_by_model.get(model, 0)
-            reqs = self._requests_by_model.get(model, 0)
-            total_saved += saved
-            per_model[model] = {
-                "requests": reqs,
-                "tokens_saved": saved,
-                "tokens_sent": sent,
-                "reduction_pct": round(saved / (saved + sent) * 100, 1)
-                if (saved + sent) > 0
-                else 0,
-            }
-
-        # Compute actual input cost using API-reported cache breakdown and
-        # LiteLLM's per-category pricing (cache reads discounted, writes at
-        # premium, uncached at list). Falls back to list price when cache
-        # data is unavailable.
-        cost_with_headroom = 0.0
-        total_billed_input_tokens = 0
-        total_input_tokens = 0
-        for model in self._tokens_saved_by_model:
-            saved = self._tokens_saved_by_model[model]
-            sent = self._tokens_sent_by_model.get(model, 0)
-            cr = self._api_cache_read_by_model.get(model, 0)
-            cw = self._api_cache_write_by_model.get(model, 0)
-            uncached = self._api_uncached_by_model.get(model, 0)
-            total_input_tokens += sent
-
-            prices = self._get_cache_prices(model)
-            if prices:
-                cr_price, cw_price, uncached_price = prices
-                if cr + cw + uncached > 0:
-                    # Use API's real cache breakdown with LiteLLM pricing
-                    model_cost = cr * cr_price + cw * cw_price + uncached * uncached_price
-                    billed_tokens = cr + cw + uncached
-                else:
-                    # No cache data from API — fall back to list price
-                    model_cost = sent * uncached_price
-                    billed_tokens = sent
-                cost_with_headroom += model_cost
-                total_billed_input_tokens += billed_tokens
-
-        # Compression savings: price saved tokens at the model's list input price.
-        # This is simple, monotonic, and transparent — each saved token is valued
-        # at the published $/token rate for its model. Not affected by cache mix.
-        savings_usd = 0.0
-        for model in self._tokens_saved_by_model:
-            saved = self._tokens_saved_by_model[model]
-            if saved <= 0:
-                continue
-            prices = self._get_cache_prices(model)
-            if prices:
-                _cr_price, _cw_price, uncached_price = prices
-                savings_usd += saved * uncached_price
-
-        return {
-            "total_tokens_saved": total_saved,
-            "total_input_tokens": total_input_tokens,
-            "total_input_cost_usd": round(cost_with_headroom, 4),
-            "per_model": per_model,
-            "cost_with_headroom_usd": round(cost_with_headroom, 4),
-            "savings_usd": round(savings_usd, 4),
-        }
-
-
-# =============================================================================
-# Prometheus Metrics
-# =============================================================================
-
-
-class PrometheusMetrics:
-    """Prometheus-compatible metrics."""
-
-    def __init__(
-        self,
-        savings_tracker: SavingsTracker | None = None,
-        cost_tracker: CostTracker | None = None,
-    ):
-        self.requests_total = 0
-        self.requests_by_provider: dict[str, int] = defaultdict(int)
-        self.requests_by_model: dict[str, int] = defaultdict(int)
-        self.requests_cached = 0
-        self.requests_rate_limited = 0
-        self.requests_failed = 0
-
-        self.tokens_input_total = 0
-        self.tokens_output_total = 0
-        self.tokens_saved_total = 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)
-
-        # Aggregate waste signals
-        self.waste_signals_total: 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,
-                "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
-
-        # Cumulative savings history (timestamp → cumulative tokens saved)
-        self.savings_history: list[tuple[str, int]] = []
-        self.savings_tracker = savings_tracker or SavingsTracker()
-        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()
-
-    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
-
-    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,
-        uncached_input_tokens: int = 0,
-    ):
-        """Record metrics for a request."""
-        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
-
-            # 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["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)
-            from datetime import datetime
-
-            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,
-                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,
-            )
-
-    async def record_rate_limited(self):
-        async with self._lock:
-            self.requests_rate_limited += 1
-
-    async def record_failed(self):
-        async with self._lock:
-            self.requests_failed += 1
-
-    async def export(self) -> str:
-        """Export metrics in Prometheus format."""
-        async with self._lock:
-            lines = [
-                "# HELP headroom_requests_total Total number of requests",
-                "# TYPE headroom_requests_total counter",
-                f"headroom_requests_total {self.requests_total}",
-                "",
-                "# HELP headroom_requests_cached_total Cached request count",
-                "# TYPE headroom_requests_cached_total counter",
-                f"headroom_requests_cached_total {self.requests_cached}",
-                "",
-                "# HELP headroom_requests_rate_limited_total Rate limited requests",
-                "# TYPE headroom_requests_rate_limited_total counter",
-                f"headroom_requests_rate_limited_total {self.requests_rate_limited}",
-                "",
-                "# HELP headroom_requests_failed_total Failed requests",
-                "# TYPE headroom_requests_failed_total counter",
-                f"headroom_requests_failed_total {self.requests_failed}",
-                "",
-                "# HELP headroom_tokens_input_total Total input tokens",
-                "# TYPE headroom_tokens_input_total counter",
-                f"headroom_tokens_input_total {self.tokens_input_total}",
-                "",
-                "# HELP headroom_tokens_output_total Total output tokens",
-                "# TYPE headroom_tokens_output_total counter",
-                f"headroom_tokens_output_total {self.tokens_output_total}",
-                "",
-                "# HELP headroom_tokens_saved_total Tokens saved by optimization",
-                "# TYPE headroom_tokens_saved_total counter",
-                f"headroom_tokens_saved_total {self.tokens_saved_total}",
-                "",
-                "# HELP headroom_latency_ms_sum Sum of request latencies",
-                "# TYPE headroom_latency_ms_sum counter",
-                f"headroom_latency_ms_sum {self.latency_sum_ms:.2f}",
-            ]
-
-            # Per-provider metrics
-            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}')
-
-            # Per-model metrics
-            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}')
-
-            return "\n".join(lines)
-
-
-# =============================================================================
-# Request Logger
-# =============================================================================
-
-
-class RequestLogger:
-    """Log requests to JSONL file.
-
-    Uses a deque with max 10,000 entries to prevent unbounded memory growth.
-    """
-
-    MAX_LOG_ENTRIES = 10_000
-
-    def __init__(self, log_file: str | None = None, log_full_messages: bool = False):
-        self.log_file = Path(log_file) if log_file else None
-        self.log_full_messages = log_full_messages
-        # Use deque with maxlen for automatic FIFO eviction
-        self._logs: deque[RequestLog] = deque(maxlen=self.MAX_LOG_ENTRIES)
-
-        if self.log_file:
-            self.log_file.parent.mkdir(parents=True, exist_ok=True)
-
-    def log(self, entry: RequestLog):
-        """Log a request. Oldest entries are automatically removed when limit reached."""
-        self._logs.append(entry)
-
-        if self.log_file:
-            with open(self.log_file, "a") as f:
-                log_dict = asdict(entry)
-                if not self.log_full_messages:
-                    log_dict.pop("request_messages", None)
-                    log_dict.pop("response_content", None)
-                f.write(json.dumps(log_dict) + "\n")
-
-    def get_recent(self, n: int = 100) -> list[dict]:
-        """Get recent log entries."""
-        # Convert deque to list for slicing (deque doesn't support slicing)
-        entries = list(self._logs)[-n:]
-        return [
-            {
-                k: v
-                for k, v in asdict(e).items()
-                if k not in ("request_messages", "response_content")
-            }
-            for e in entries
-        ]
-
-    def stats(self) -> dict:
-        """Get logging statistics."""
-        return {
-            "total_logged": len(self._logs),
-            "log_file": str(self.log_file) if self.log_file else None,
-        }
-
-    def get_memory_stats(self) -> ComponentStats:
-        """Get memory statistics for the MemoryTracker.
-
-        Returns:
-            ComponentStats with current memory usage.
-        """
-        from ..memory.tracker import ComponentStats
-
-        # Calculate size
-        size_bytes = sys.getsizeof(self._logs)
-
-        for log_entry in self._logs:
-            size_bytes += sys.getsizeof(log_entry)
-            # Add string fields
-            if log_entry.request_id:
-                size_bytes += len(log_entry.request_id)
-            if log_entry.provider:
-                size_bytes += len(log_entry.provider)
-            if log_entry.model:
-                size_bytes += len(log_entry.model)
-            if log_entry.error:
-                size_bytes += len(log_entry.error)
-            # Messages and response can be large
-            if log_entry.request_messages:
-                size_bytes += sys.getsizeof(log_entry.request_messages)
-            if log_entry.response_content:
-                size_bytes += len(log_entry.response_content)
-
-        return ComponentStats(
-            name="request_logger",
-            entry_count=len(self._logs),
-            size_bytes=size_bytes,
-            budget_bytes=None,
-            hits=0,
-            misses=0,
-            evictions=0,
-        )
-
-
-# =============================================================================
-# Main Proxy
-# =============================================================================
-
-
 class HeadroomProxy:
     """Production-ready Headroom optimization proxy."""
 

pyproject.toml

@@ -249,6 +249,12 @@ ignore_missing_imports = true
 [[tool.mypy.overrides]]
 module = [
     "headroom.proxy.server",
+    "headroom.proxy.cost",
+    "headroom.proxy.prometheus_metrics",
+    "headroom.proxy.semantic_cache",
+    "headroom.proxy.rate_limiter",
+    "headroom.proxy.request_logger",
+    "headroom.proxy.helpers",
     "headroom.integrations.langchain",
     "headroom.integrations.mcp",
     "headroom.ccr.mcp_server",

tests/test_proxy_streaming_resilience.py

@@ -83,8 +83,8 @@ class TestModelResolutionCaching:
         from headroom.proxy.server import CostTracker
 
         with (
-            patch("headroom.proxy.server.LITELLM_AVAILABLE", True),
-            patch("headroom.proxy.server.litellm") as mock_litellm,
+            patch("headroom.proxy.cost.LITELLM_AVAILABLE", True),
+            patch("headroom.proxy.cost.litellm") as mock_litellm,
         ):
             # First call (bare name) fails, second call (prefixed) succeeds
             mock_litellm.cost_per_token.side_effect = [
@@ -100,8 +100,8 @@ class TestModelResolutionCaching:
         from headroom.proxy.server import CostTracker
 
         with (
-            patch("headroom.proxy.server.LITELLM_AVAILABLE", True),
-            patch("headroom.proxy.server.litellm") as mock_litellm,
+            patch("headroom.proxy.cost.LITELLM_AVAILABLE", True),
+            patch("headroom.proxy.cost.litellm") as mock_litellm,
         ):
             mock_litellm.cost_per_token.side_effect = [
                 Exception("Unknown model"),
@@ -116,8 +116,8 @@ class TestModelResolutionCaching:
         from headroom.proxy.server import CostTracker
 
         with (
-            patch("headroom.proxy.server.LITELLM_AVAILABLE", True),
-            patch("headroom.proxy.server.litellm") as mock_litellm,
+            patch("headroom.proxy.cost.LITELLM_AVAILABLE", True),
+            patch("headroom.proxy.cost.litellm") as mock_litellm,
         ):
             mock_litellm.cost_per_token.side_effect = [
                 Exception("Unknown model"),
@@ -132,8 +132,8 @@ class TestModelResolutionCaching:
         from headroom.proxy.server import CostTracker
 
         with (
-            patch("headroom.proxy.server.LITELLM_AVAILABLE", True),
-            patch("headroom.proxy.server.litellm") as mock_litellm,
+            patch("headroom.proxy.cost.LITELLM_AVAILABLE", True),
+            patch("headroom.proxy.cost.litellm") as mock_litellm,
         ):
             mock_litellm.cost_per_token.side_effect = Exception("Unknown model")
 
@@ -144,7 +144,7 @@ class TestModelResolutionCaching:
         """When litellm is not available, return model as-is."""
         from headroom.proxy.server import CostTracker
 
-        with patch("headroom.proxy.server.LITELLM_AVAILABLE", False):
+        with patch("headroom.proxy.cost.LITELLM_AVAILABLE", False):
             result = CostTracker._resolve_litellm_model_uncached("claude-opus-4-6")
             assert result == "claude-opus-4-6"
 
@@ -153,8 +153,8 @@ class TestModelResolutionCaching:
         from headroom.proxy.server import CostTracker
 
         with (
-            patch("headroom.proxy.server.LITELLM_AVAILABLE", True),
-            patch("headroom.proxy.server.litellm") as mock_litellm,
+            patch("headroom.proxy.cost.LITELLM_AVAILABLE", True),
+            patch("headroom.proxy.cost.litellm") as mock_litellm,
         ):
             mock_litellm.cost_per_token.return_value = (0.001, 0.002)
 
@@ -516,8 +516,8 @@ class TestConcurrentSessionSafety:
         CostTracker._resolved_model_cache["gpt-4o"] = "openai/gpt-4o"
 
         with (
-            patch("headroom.proxy.server.LITELLM_AVAILABLE", True),
-            patch("headroom.proxy.server.litellm") as mock_litellm,
+            patch("headroom.proxy.cost.LITELLM_AVAILABLE", True),
+            patch("headroom.proxy.cost.litellm") as mock_litellm,
         ):
             mock_litellm.cost_per_token.return_value = (0.001, 0.002)
             mock_litellm.get_model_info.return_value = {}
@@ -555,8 +555,8 @@ class TestCostTrackingAccuracy:
         tracker = CostTracker()
 
         with (
-            patch("headroom.proxy.server.LITELLM_AVAILABLE", True),
-            patch("headroom.proxy.server.litellm") as mock_litellm,
+            patch("headroom.proxy.cost.LITELLM_AVAILABLE", True),
+            patch("headroom.proxy.cost.litellm") as mock_litellm,
         ):
             # Setup: $10/M input, $30/M output
             def mock_cost(model, prompt_tokens, completion_tokens, **kwargs):
@@ -601,8 +601,8 @@ class TestCostTrackingAccuracy:
         tracker = CostTracker()
 
         with (
-            patch("headroom.proxy.server.LITELLM_AVAILABLE", True),
-            patch("headroom.proxy.server.litellm") as mock_litellm,
+            patch("headroom.proxy.cost.LITELLM_AVAILABLE", True),
+            patch("headroom.proxy.cost.litellm") as mock_litellm,
         ):
             mock_litellm.cost_per_token.side_effect = (
                 lambda model, prompt_tokens, completion_tokens, **kwargs: (
@@ -623,6 +623,6 @@ class TestCostTrackingAccuracy:
 
         tracker = CostTracker()
 
-        with patch("headroom.proxy.server.LITELLM_AVAILABLE", False):
+        with patch("headroom.proxy.cost.LITELLM_AVAILABLE", False):
             cost = tracker.estimate_cost("gpt-4o", input_tokens=1000, output_tokens=100)
             assert cost is None