"""Headroom Proxy Server - Production Ready. A full-featured LLM proxy with optimization, caching, rate limiting, and observability. Features: - Context optimization (SmartCrusher, CacheAligner — live-zone-only after Phase B) - Semantic caching (save costs on repeated queries) - Rate limiting (token bucket) - Retry with exponential backoff - Cost tracking and budgets - Request tagging and metadata - Provider fallback - Prometheus metrics - Full request/response logging Usage: python -m headroom.proxy.server --port 8787 # With Claude Code: ANTHROPIC_BASE_URL=http://localhost:8787 claude """ from __future__ import annotations import argparse import asyncio import concurrent.futures import contextlib import hmac import json import logging import os import sys import threading import time from collections.abc import Callable from dataclasses import fields, is_dataclass, replace from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast if TYPE_CHECKING: from ..backends.base import Backend from ..cache.compression_cache import CompressionCache from ..memory.tracker import MemoryTracker from .outcome import RequestOutcome import httpx try: import uvicorn from fastapi import Depends, FastAPI, HTTPException, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse FASTAPI_AVAILABLE = True except ImportError: FASTAPI_AVAILABLE = False # Add parent to path for imports sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from headroom._version import __version__ from headroom.agent_savings import proxy_pipeline_kwargs from headroom.cache.compression_feedback import get_compression_feedback from headroom.cache.compression_store import format_retrieval_miss_detail, get_compression_store from headroom.ccr import ( CCR_TOOL_NAME, # Batch processing CCRResponseHandler, CCRToolInjector, ContextTracker, ContextTrackerConfig, ResponseHandlerConfig, parse_tool_call, ) from headroom.config import ( DEFAULT_EXCLUDE_TOOLS, CacheAlignerConfig, ReadLifecycleConfig, ) from headroom.dashboard import get_dashboard_html from headroom.observability import ( LangfuseTracingConfig, OTelMetricsConfig, configure_langfuse_tracing, configure_otel_metrics, get_langfuse_tracing_status, get_otel_metrics_status, shutdown_headroom_tracing, shutdown_otel_metrics, ) from headroom.offline import apply_offline_env, is_offline from headroom.pipeline import PipelineExtensionManager, PipelineStage from headroom.providers.proxy_routes import register_provider_routes from headroom.providers.registry import ( DEFAULT_ANTHROPIC_API_URL, DEFAULT_CLOUDCODE_API_URL, DEFAULT_GEMINI_API_URL, DEFAULT_OPENAI_API_URL, DEFAULT_VERTEX_API_URL, build_proxy_provider_runtime, create_proxy_backend, format_backend_status, resolve_api_targets, ) from headroom.proxy import runtime_env from headroom.proxy.audit import is_auditable_path, record_admin_action from headroom.proxy.auth_mode import should_stamp_codex_client from headroom.proxy.background_compression import BackgroundCompressor # ============================================================================= # Extracted modules (re-exported for backward compatibility) # ============================================================================= from headroom.proxy.cost import ( _CACHE_ECONOMICS, # noqa: F401 CostTracker, # noqa: F401 _summarize_transforms, # noqa: F401 build_prefix_cache_stats, # noqa: F401 build_session_summary, # noqa: F401 merge_cost_stats, # noqa: F401 ) from headroom.proxy.helpers import ( COMPRESSION_TIMEOUT_SECONDS, # noqa: F401 EAGER_PRELOAD_TIMEOUT_SECONDS, MAX_COMPRESSION_CACHE_SESSIONS, # noqa: F401 MAX_MESSAGE_ARRAY_LENGTH, # noqa: F401 MAX_REQUEST_BODY_SIZE, # noqa: F401 MAX_SSE_BUFFER_SIZE, # noqa: F401 RETRYABLE_OVERLOAD_STATUSES, _get_context_tool_stats, _get_image_compressor, # noqa: F401 _get_rtk_stats, # noqa: F401 _read_request_json, # noqa: F401 _setup_file_logging, # noqa: F401 initialize_context_tool_session_baseline, is_anthropic_auth, # noqa: F401 jitter_delay_ms, retry_after_ms, ) from headroom.proxy.loop_callback_failure_policy import is_known_websocket_callback_failure from headroom.proxy.loopback_guard import is_loopback_host 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.modes import ( PROXY_MODE_CACHE, PROXY_MODE_TOKEN, is_token_mode, normalize_proxy_mode, ) from headroom.proxy.probe_recorder import probe_recorder_from_env from headroom.proxy.project_context import ( classify_project, set_current_project, strip_project_path_prefix, ) 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.savings_tracker import LITELLM_AVAILABLE from headroom.proxy.semantic_cache import SemanticCache # noqa: F401 from headroom.proxy.ssl_context import build_httpx_verify from headroom.proxy.tool_schema_savings_policy import tool_schema_saved_from_tags from headroom.proxy.warmup import WarmupRegistry from headroom.proxy.ws_session_registry import WebSocketSessionRegistry from headroom.subscription.base import get_quota_registry, reset_quota_registry from headroom.subscription.codex_rate_limits import get_codex_rate_limit_state from headroom.subscription.copilot_quota import get_copilot_quota_tracker from headroom.subscription.tracker import ( configure_subscription_tracker, get_subscription_tracker, ) from headroom.telemetry import get_telemetry_collector from headroom.telemetry.beacon import is_telemetry_enabled from headroom.telemetry.toin import get_toin from headroom.transforms import ( CacheAligner, CodeAwareCompressor, CodeCompressorConfig, CompressionStrategy, ContentRouter, ContentRouterConfig, TransformPipeline, is_tree_sitter_available, ) AnyLLMBackend: Any = None LiteLLMBackend: Any = None fcntl: Any = None try: import fcntl as _fcntl fcntl = _fcntl HAS_FCNTL = True except ImportError: HAS_FCNTL = False _build_prefix_cache_stats = build_prefix_cache_stats _build_session_summary = build_session_summary _merge_cost_stats = merge_cost_stats _AGENT_LABELS: dict[str, str] = { "claude": "Claude", "claude-code": "Claude", "claude_cli": "Claude", "claude-code-cli": "Claude", "codex": "Codex", "codex-cli": "Codex", "cursor": "Cursor", "copilot": "GitHub Copilot", "github-copilot": "GitHub Copilot", "aider": "Aider", "zed": "Zed", "opencode": "OpenCode", "openclaw": "OpenClaw", "gemini": "Gemini", "google": "Gemini", "vertex:google": "Gemini", "anthropic": "Claude", "openai": "OpenAI", "unknown": "Unidentified", } _AGENT_SOURCE_PRIORITY: dict[str, int] = { "unknown": 0, "provider": 1, "model": 2, "stack": 3, "client": 4, } def _normalize_agent_key(raw: Any) -> str | None: if raw is None: return None value = str(raw).strip().lower() if not value: return None value = value.replace(" ", "-").replace("_", "-") if value.startswith("wrap-"): value = value.removeprefix("wrap-") if value in {"claude-cli", "claude-code", "claude-code-cli"}: return "claude-code" if value in {"codex-cli", "codex"}: return "codex" if value in {"github-copilot", "copilot"}: return "copilot" if value in {"google", "vertex-google", "vertex:google"}: return "gemini" return value def _agent_label(agent_key: str) -> str: if agent_key in _AGENT_LABELS: return _AGENT_LABELS[agent_key] return agent_key.replace("-", " ").replace("_", " ").title() def _classify_agent_from_log(entry: dict[str, Any]) -> tuple[str, str, str]: raw_tags = entry.get("tags") tags = raw_tags if isinstance(raw_tags, dict) else {} for source, candidate in ( ("client", tags.get("client")), ("stack", tags.get("stack") or tags.get("headroom-stack")), ): key = _normalize_agent_key(candidate) if key: return key, _agent_label(key), source model = str(entry.get("model") or "").lower() if "codex" in model: return "codex", _agent_label("codex"), "model" if "claude" in model: return "claude-code", _agent_label("claude-code"), "model" if "gemini" in model: return "gemini", _agent_label("gemini"), "model" key = _normalize_agent_key(entry.get("provider")) if key: return key, _agent_label(key), "provider" return "unknown", _agent_label("unknown"), "unknown" def _build_agent_usage_summary( logs: list[dict[str, Any]], *, requests_by_provider: dict[str, int], requests_by_model: dict[str, int], global_before_tokens: int, global_after_tokens: int, global_tokens_saved: int, global_output_tokens: int, ) -> dict[str, Any]: agents: dict[str, dict[str, Any]] = {} def _agent_row(agent_key: str, label: str, source: str) -> dict[str, Any]: row = agents.setdefault( agent_key, { "agent": agent_key, "label": label, "source": source, "requests": 0, "before_tokens": 0, "after_tokens": 0, "output_tokens": 0, "tokens_saved": 0, "models": {}, "providers": {}, "has_exact_tokens": False, }, ) if _AGENT_SOURCE_PRIORITY.get(source, 0) > _AGENT_SOURCE_PRIORITY.get( str(row.get("source") or "unknown"), 0 ): row["source"] = source return row for entry in logs: agent_key, label, source = _classify_agent_from_log(entry) row = _agent_row(agent_key, label, source) before = max(0, int(entry.get("input_tokens_original") or 0)) after = max(0, int(entry.get("input_tokens_optimized") or 0)) saved = max(0, int(entry.get("tokens_saved") or 0)) output = max(0, int(entry.get("output_tokens") or 0)) provider = str(entry.get("provider") or "unknown") model = str(entry.get("model") or "unknown") row["requests"] += 1 row["before_tokens"] += before row["after_tokens"] += after row["output_tokens"] += output row["tokens_saved"] += saved row["providers"][provider] = int(row["providers"].get(provider, 0)) + 1 row["models"][model] = int(row["models"].get(model, 0)) + 1 if before > 0 or after > 0 or saved > 0: row["has_exact_tokens"] = True if not agents: inferred_model_counts: dict[str, int] = {} for model, count in requests_by_model.items(): model_lower = str(model).lower() if "codex" in model_lower: key = "codex" elif "claude" in model_lower: key = "claude-code" elif "gemini" in model_lower: key = "gemini" else: continue inferred_model_counts[str(model)] = int(count) provider_request_count = sum(max(0, int(count)) for count in requests_by_provider.values()) inferred_request_count = sum(max(0, count) for count in inferred_model_counts.values()) use_model_fallback = ( inferred_request_count > 0 and inferred_request_count == provider_request_count ) if not use_model_fallback: for provider, count in requests_by_provider.items(): key = _normalize_agent_key(provider) or "unknown" row = _agent_row(key, _agent_label(key), "provider") row["requests"] += int(count) row["providers"][provider] = int(row["providers"].get(provider, 0)) + int(count) for model, count in requests_by_model.items(): model_lower = str(model).lower() if "codex" in model_lower: key = "codex" elif "claude" in model_lower: key = "claude-code" elif "gemini" in model_lower: key = "gemini" else: continue if not use_model_fallback: continue row = _agent_row(key, _agent_label(key), "model") row["requests"] += int(count) row["models"][str(model)] = int(row["models"].get(str(model), 0)) + int(count) rows: list[dict[str, Any]] = [] for row in agents.values(): before = int(row["before_tokens"]) saved = int(row["tokens_saved"]) after = int(row["after_tokens"]) if before == 0 and (after > 0 or saved > 0): before = after + saved savings_percent = round((saved / before) * 100.0, 2) if before else 0.0 row["before_tokens"] = before row["savings_percent"] = savings_percent row["after_percent"] = round((after / before) * 100.0, 2) if before else 0.0 row["share_of_saved_percent"] = ( round((saved / global_tokens_saved) * 100.0, 2) if global_tokens_saved else 0.0 ) row["share_of_requests_percent"] = 0.0 rows.append(row) total_requests = sum(int(row["requests"]) for row in rows) for row in rows: row["share_of_requests_percent"] = ( round((int(row["requests"]) / total_requests) * 100.0, 2) if total_requests else 0.0 ) rows.sort( key=lambda row: ( int(row.get("tokens_saved", 0)), int(row.get("before_tokens", 0)), int(row.get("requests", 0)), ), reverse=True, ) return { "agents": rows, "totals": { "requests": total_requests, "before_tokens": global_before_tokens, "after_tokens": global_after_tokens, "output_tokens": global_output_tokens, "tokens_saved": global_tokens_saved, "savings_percent": ( round((global_tokens_saved / global_before_tokens) * 100.0, 2) if global_before_tokens else 0.0 ), }, "coverage": { "logged_requests": len(logs), "exact_token_rows": sum(1 for row in rows if row.get("has_exact_tokens")), "mode": "request_logs" if logs else "aggregate_fallback", }, } # Suppress "[transformers] PyTorch was not found" warning emitted when # transformers is imported for availability checks (e.g. kompress ONNX probe). # PyTorch is optional in headroom; the warning is not actionable for operators. os.environ.setdefault("TRANSFORMERS_VERBOSITY", "error") logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger("headroom.proxy") LoopExceptionHandler = Callable[[asyncio.AbstractEventLoop, dict[str, Any]], object] class LoopFailureDetails(TypedDict): message: Any | None exception: str | None class LoopHealthState(TypedDict): status: str known_failures: int last_known_failure: LoopFailureDetails | None _MULTI_WORKER_CONFIG_ENV = "HEADROOM_PROXY_CONFIG_JSON" # Env var that opts out of the Rust core deployment smoke test (Hotfix-A0). # Default behavior: hard-fail at startup if `headroom._core` is unimportable # (Finding #2 in HEADROOM_PROXY_LOG_FINDINGS_2026_05_03.md — production # deployment was silently running without the Rust extension and degrading # every compressed request to a Python-only path or a no-op). # # Set to the literal string "false" to start the proxy in degraded # Python-only mode. Any other value (including unset) keeps the # fail-loud behavior. _RUST_CORE_REQUIRED_ENV = "HEADROOM_REQUIRE_RUST_CORE" # sysexits.h(3) — EX_CONFIG. Process supervisors (systemd, k8s, docker) # treat this as a deliberate configuration failure rather than a crash, so # they won't restart-loop on a broken deployment. _EXIT_CONFIG = 78 def _check_rust_core() -> tuple[str, str | None]: """Verify the Rust extension `headroom._core` is loadable at startup. Returns a `(status, error)` tuple: - ``("loaded", None)`` — `headroom._core.hello()` returned the expected sentinel. - ``("disabled", reason)`` — opt-out env var was set; proxy starts in Python-only degraded mode. `reason` carries the underlying import error (or ``None`` if the import actually succeeded). - ``("missing", reason)`` — never returned: this branch calls ``sys.exit(78)`` so the proxy refuses to start. The branch exists only as a typed sentinel for callers that want to reason about all three states (e.g. health endpoints). Behavior is gated by the ``HEADROOM_REQUIRE_RUST_CORE`` env var: any value other than ``"false"`` (case-insensitive) keeps the fail-loud default. """ require = os.environ.get(_RUST_CORE_REQUIRED_ENV, "true").strip().lower() != "false" try: from headroom._core import hello as _rust_hello marker = _rust_hello() except Exception as exc: # ImportError, but also any init-time PyO3 failure reason = f"{type(exc).__name__}: {exc}" if not require: logger.warning( "event=rust_core_disabled reason=%r opt_out_env=%s=false mode=python_only_degraded", reason, _RUST_CORE_REQUIRED_ENV, ) return ("disabled", reason) # Fail loud. Print to stderr in addition to logging so operators # see it even if the logging handler is mis-configured. msg = ( f"FATAL: Rust extension `headroom._core` not loadable.\n" f" error: {reason}\n" f" fix: `make build-wheel && pip install --force-reinstall " f"target/wheels/headroom_*.whl`\n" f" opt-out: set {_RUST_CORE_REQUIRED_ENV}=false to start in " f"degraded Python-only mode\n" ) logger.error("event=rust_core_missing reason=%r action=exit_78", reason) print(msg, file=sys.stderr, flush=True) sys.exit(_EXIT_CONFIG) # Import succeeded; sanity-check the marker so we catch a stale or # mis-linked .so where the symbol name resolves but returns garbage. if marker != "headroom-core": reason = f"unexpected marker {marker!r}" if not require: logger.warning( "event=rust_core_disabled reason=%r opt_out_env=%s=false", reason, _RUST_CORE_REQUIRED_ENV, ) return ("disabled", reason) msg = ( f"FATAL: Rust extension `headroom._core` is loaded but the " f"marker function returned {marker!r}; expected 'headroom-core'.\n" f" fix: rebuild: `make build-wheel && pip install " f"--force-reinstall target/wheels/headroom_*.whl`\n" ) logger.error("event=rust_core_marker_mismatch marker=%r action=exit_78", marker) print(msg, file=sys.stderr, flush=True) sys.exit(_EXIT_CONFIG) logger.info("event=rust_core_loaded marker=%r", marker) return ("loaded", None) # Compression pipeline timeout in seconds from headroom.proxy.handlers import ( # noqa: E402 AnthropicHandlerMixin, BatchHandlerMixin, BedrockHandlerMixin, GeminiHandlerMixin, OpenAIHandlerMixin, StreamingMixin, ) def _apply_stateless_persistence(config: ProxyConfig) -> None: """When the proxy runs stateless, force global persisters to in-memory so no files are written to the workspace. Covers TOIN (the always-on serving writer): it keeps learning patterns in-memory but never reads or writes ``toin.json``. An empty ``storage_path`` makes the backend ``None``, which no-ops load/save/auto-save. The savings subsystem is handled separately via ``PrometheusMetrics(stateless=...)``. Note: setting ``HEADROOM_TOIN_BACKEND=none`` is NOT sufficient on its own — ``ToolIntelligenceNetwork`` falls back to ``config.storage_path`` when no backend is passed, so we must clear the path explicitly here. Concurrency: ``stateless`` is a per-process flag (set once at ``headroom proxy`` launch), never a per-request/per-session value — every session a process serves shares it, and two proxies with different settings run as separate OS processes with independent TOIN singletons. In the rare case where two HeadroomProxy instances with different ``stateless`` settings live in ONE process (e.g. tests), this fails closed: the reset forces the process-global TOIN in-memory, so a stateless proxy never persists (the safe direction). A co-resident stateful proxy would then also stop persisting TOIN — acceptable, since not-writing can never leak data. """ if not getattr(config, "stateless", False): return from headroom.telemetry.toin import TOINConfig, get_toin, reset_toin # Reset first so this wins regardless of whether the singleton was already # created with a filesystem backend earlier in the process. reset_toin() get_toin(TOINConfig(storage_path="")) def _provider_httpx_client_options( config: ProxyConfig, verify: Any, ) -> tuple[bool, dict[str, Any]]: client_kwargs: dict[str, Any] = { "timeout": httpx.Timeout( connect=config.connect_timeout_seconds, read=config.request_timeout_seconds, write=config.request_timeout_seconds, pool=config.connect_timeout_seconds, ), "limits": httpx.Limits( max_connections=config.max_connections, max_keepalive_connections=config.max_keepalive_connections, keepalive_expiry=config.keepalive_expiry, ), "verify": verify, } if config.http_proxy: client_kwargs["proxy"] = config.http_proxy return config.http2 and not config.http_proxy, client_kwargs class HeadroomProxy( StreamingMixin, AnthropicHandlerMixin, OpenAIHandlerMixin, GeminiHandlerMixin, BatchHandlerMixin, BedrockHandlerMixin, ): """Production-ready Headroom optimization proxy.""" ANTHROPIC_API_URL = DEFAULT_ANTHROPIC_API_URL OPENAI_API_URL = DEFAULT_OPENAI_API_URL GEMINI_API_URL = DEFAULT_GEMINI_API_URL CLOUDCODE_API_URL = DEFAULT_CLOUDCODE_API_URL VERTEX_API_URL = DEFAULT_VERTEX_API_URL def __init__(self, config: ProxyConfig): self.config = config self.config.mode = normalize_proxy_mode(self.config.mode) # Record process-wide stateless mode so module-level persisters # (output-savings recorder, etc.) can skip workspace writes. from headroom import paths as _hr_paths _hr_paths.set_process_stateless(config.stateless) # Stateless: keep TOIN learning in-memory; never touch toin.json. _apply_stateless_persistence(self.config) pipeline_extensions = list(config.pipeline_extensions or []) probe_recorder = probe_recorder_from_env() if probe_recorder is not None: pipeline_extensions.append(probe_recorder) self.pipeline_extensions = PipelineExtensionManager( hooks=config.hooks, extensions=pipeline_extensions, discover=config.discover_pipeline_extensions, ) self.provider_runtime = build_proxy_provider_runtime(config) api_targets = self.provider_runtime.api_targets # Preserve the long-standing proxy compatibility surface while keeping # provider_runtime as the source of truth for resolved upstream targets. HeadroomProxy.ANTHROPIC_API_URL = api_targets.anthropic HeadroomProxy.OPENAI_API_URL = api_targets.openai HeadroomProxy.GEMINI_API_URL = api_targets.gemini HeadroomProxy.CLOUDCODE_API_URL = api_targets.cloudcode HeadroomProxy.VERTEX_API_URL = api_targets.vertex self.anthropic_provider = self.provider_runtime.pipeline_provider("anthropic") self.openai_provider = self.provider_runtime.pipeline_provider("openai") # `metrics` is hoisted ahead of transform construction so the # transforms can receive `self.metrics` as their compression # observer at __init__ time. The forcing function for catching # silent strategy regressions: per-strategy counters increment # only when wired up here, so the wiring is mandatory, not # something we patch in later. (See `RUST_DEV.md` audit notes.) self.cost_tracker = ( CostTracker( budget_limit_usd=config.budget_limit_usd, budget_period=config.budget_period, ) if config.cost_tracking_enabled else None ) self.metrics = PrometheusMetrics(cost_tracker=self.cost_tracker, stateless=config.stateless) # Initialize transforms based on routing mode. # # Phase B PR-B1 retired the IntelligentContextManager / RollingWindow # message-dropping branch. Live-zone-only compression (PR-B2..B7) does # not drop messages — it operates on content blocks within messages — # so the proxy no longer needs a "context manager" transform stage. # Reported via metrics as `_context_manager_status = "passthrough"`. self._context_manager_status = "passthrough" # ContentRouter is the single proxy routing surface. Provider handlers # normalize their request shapes into messages or CompressionUnits, and # the router chooses SmartCrusher, log/search/diff/code, or Kompress. profile_kwargs = proxy_pipeline_kwargs(config) router_config = ContentRouterConfig( enable_code_aware=config.code_aware_enabled, prefer_code_aware_for_code=_get_env_bool("HEADROOM_PREFER_CODE_AWARE_FOR_CODE", True), tool_profiles=config.tool_profiles, read_lifecycle=ReadLifecycleConfig(enabled=config.read_lifecycle), smart_crusher_max_items_after_crush=cast( int | None, profile_kwargs.get("max_items_after_crush"), ), smart_crusher_with_compaction=cast( bool, profile_kwargs.get("smart_crusher_with_compaction", True), ), ccr_inject_marker=config.ccr_inject_marker, force_kompress_all=config.force_kompress_all, lossless=config.lossless, ) # No-CCR lossless mode: compress tool outputs with format-native # lossless compaction and marker-free SmartCrusher, and suppress every # retrieval marker + the retrieve-tool injection so no MCP round-trip is # needed. Mirrors the force_kompress_all wiring precedent. if config.lossless: router_config.lossless = True router_config.smart_crusher_lossless_only = True router_config.ccr_inject_marker = False if hasattr(config, "ccr_inject_tool"): config.ccr_inject_tool = False if config.disable_kompress: router_config.enable_kompress = False # Opt-in restore of the legacy behaviour: send fall-through content # to PASSTHROUGH instead of the default KOMPRESS fallback strategy. if config.disable_kompress_fallback: router_config.fallback_strategy = CompressionStrategy.PASSTHROUGH # `HEADROOM_LOSSLESS_ONLY=1` routes SmartCrusher through strict # marker-free mode: lossless tabular compaction still applies, but # any path that would emit a `<>` marker (row-drop or # opaque-blob offload) leaves the content uncompacted instead — so # the session needs no CCR retrieval round-trips to stay recoverable. if "HEADROOM_LOSSLESS_ONLY" in os.environ: router_config.smart_crusher_lossless_only = _get_env_bool( "HEADROOM_LOSSLESS_ONLY", False ) # A non-None exclude_tools replaces DEFAULT_EXCLUDE_TOOLS in # ContentRouter, so merge rather than assign. if config.exclude_tools: router_config.exclude_tools = set(DEFAULT_EXCLUDE_TOOLS) | config.exclude_tools # protect_tool_results: force-merge named tools into the exclude set # so their results are never lossy-compressed, regardless of mode. if config.protect_tool_results: base = ( router_config.exclude_tools if router_config.exclude_tools is not None else set(DEFAULT_EXCLUDE_TOOLS) ) router_config.exclude_tools = base | config.protect_tool_results # Token mode: allow compression of older excluded-tool results, # and emit search results grouped by file (path once per file # instead of repeated on every match line). if is_token_mode(config.mode): router_config.protect_recent_reads_fraction = 0.3 router_config.search_group_by_file = True if config.protect_tool_results: router_config.protect_recent_reads_fraction = 0.0 # `--compress-user-messages` flips the router's default skip rule. # Off by default for prefix-cache safety; enabled for workloads where # user-message content dominates input (OpenAI/Azure chat with pasted # code/RAG context — see issue #454). if profile_kwargs.get("compress_user_messages"): router_config.skip_user_messages = False # Kompress (lossy ML text compression) is resolved per provider. The # global `disable_kompress` above is the baseline for both; a per- # provider override (disable_kompress_{anthropic,openai}) wins when set. # Only `enable_kompress` differs between providers — routing, tool # exclusion, and read-protection are identical — so when both resolve # the same we reuse ONE ContentRouter instance and the Kompress model # still loads once (startup warmup dedupes transforms by id()). base_kompress_disabled = not router_config.enable_kompress anthropic_kompress_disabled = ( base_kompress_disabled if config.disable_kompress_anthropic is None else config.disable_kompress_anthropic ) openai_kompress_disabled = ( base_kompress_disabled if config.disable_kompress_openai is None else config.disable_kompress_openai ) def _router_config_for(kompress_disabled: bool) -> ContentRouterConfig: if kompress_disabled == base_kompress_disabled: return router_config return replace(router_config, enable_kompress=not kompress_disabled) cache_aligner = CacheAligner(CacheAlignerConfig(enabled=False)) anthropic_router = ContentRouter( _router_config_for(anthropic_kompress_disabled), observer=self.metrics ) openai_router = ( anthropic_router if openai_kompress_disabled == anthropic_kompress_disabled else ContentRouter(_router_config_for(openai_kompress_disabled), observer=self.metrics) ) self._code_aware_status = "lazy" if config.code_aware_enabled else "disabled" _intercept_prefix: list = [] if os.environ.get("HEADROOM_INTERCEPT_ENABLED"): from headroom.proxy.interceptors import ToolResultInterceptorTransform _intercept_prefix = [ToolResultInterceptorTransform()] self.anthropic_pipeline = TransformPipeline( transforms=[*_intercept_prefix, cache_aligner, anthropic_router], provider=self.anthropic_provider, ) self.openai_pipeline = TransformPipeline( transforms=[*_intercept_prefix, cache_aligner, openai_router], provider=self.openai_provider, ) # Initialize components self.cache = ( SemanticCache( max_entries=config.cache_max_entries, ttl_seconds=config.cache_ttl_seconds, ) if config.cache_enabled else None ) self.rate_limiter = ( TokenBucketRateLimiter( requests_per_minute=config.rate_limit_requests_per_minute, tokens_per_minute=config.rate_limit_tokens_per_minute, ) if config.rate_limit_enabled else None ) # `cost_tracker` and `metrics` were hoisted to before transforms so # ContentRouter / SmartCrusher could take `self.metrics` as their # compression observer at __init__ time. # Prefix cache tracking: freeze already-cached messages to avoid # invalidating the provider's prefix cache with our transforms from headroom.cache.prefix_tracker import PrefixFreezeConfig, SessionTrackerStore self.session_tracker_store = SessionTrackerStore( default_config=PrefixFreezeConfig( enabled=config.prefix_freeze_enabled, session_ttl_seconds=config.prefix_freeze_session_ttl, ) ) # Compression cache store for token mode (session-scoped). The dict # itself is mutated under `_compression_caches_lock`; the per-session # `CompressionCache` instances have their own internal lock guarding # `_cache`/`_stable_hashes`/`_first_seen` against concurrent # async-dispatched requests for the same session. self._compression_caches: dict[str, CompressionCache] = {} self._compression_caches_lock = threading.RLock() self.logger = ( RequestLogger( log_file=config.log_file, log_full_messages=config.log_full_messages, ) if config.log_requests else None ) # Enterprise security plugin (loaded dynamically if available + licensed) self.security = None # HTTP client self.http_client: httpx.AsyncClient | None = None # HTTP/1.1-only client for ChatGPT passthrough (Cloudflare challenges # our HTTP/2 fingerprint on its sensitive account endpoints). self.http_client_h1: httpx.AsyncClient | None = None self._shutdown_event: asyncio.Event | None = None # Shared cold-start warmup registry (populated by startup()). # Holds typed slots with loaded / loading / null / error status for # each preloaded heavy asset. Exposed as ``proxy.warmup`` and # serialized by the /debug/warmup route (Unit 5). self.warmup: WarmupRegistry = WarmupRegistry() # Unit 3: live registry of Codex WS sessions. Populated by # ``handle_openai_responses_ws`` on accept; drained in its # outermost ``finally``. Consumed by ``/debug/ws-sessions``. self.ws_sessions: WebSocketSessionRegistry = WebSocketSessionRegistry() # Unit 4: bounded pre-upstream concurrency for the Anthropic HTTP # path. Caps how many ``handle_anthropic_messages`` calls may be # running deep-copy / first-stage compression / memory-context # lookup / upstream connect concurrently. ``/livez``, ``/readyz``, # ``/health``, ``/metrics``, ``/stats``, and the Codex WS path are # intentionally NOT gated by this semaphore. # # A value of ``0`` or negative disables the semaphore (unbounded # mode); this is useful for the Unit 6 counter-factual where we # deliberately reproduce the original starvation. The default is # ``max(2, min(8, os.cpu_count() or 4))``. _pre_upstream_cfg = config.anthropic_pre_upstream_concurrency if _pre_upstream_cfg is None: _pre_upstream_resolved = max(2, min(8, os.cpu_count() or 4)) else: _pre_upstream_resolved = _pre_upstream_cfg self.anthropic_pre_upstream_concurrency: int = _pre_upstream_resolved self.anthropic_pre_upstream_acquire_timeout_seconds = float( config.anthropic_pre_upstream_acquire_timeout_seconds ) self.anthropic_pre_upstream_memory_context_timeout_seconds = float( config.anthropic_pre_upstream_memory_context_timeout_seconds ) if _pre_upstream_resolved > 0: self.anthropic_pre_upstream_sem: asyncio.Semaphore | None = asyncio.Semaphore( _pre_upstream_resolved ) else: self.anthropic_pre_upstream_sem = None # Dedicated compression executor — see C3 in the audit followup. # Replaces ``asyncio.to_thread(...)`` for ``pipeline.apply()`` calls # so that: # 1. Compression work is bounded — CPU-bound Rust runs here, and # bursts cannot starve other ``asyncio.to_thread`` callers # sharing the loop's default executor (file IO, etc.). # 2. Tasks that exceed ``COMPRESSION_TIMEOUT_SECONDS`` and complete # *after* the asyncio future was cancelled are counted in the # ``compression_leaked_threads`` gauge — Python cannot preempt # the worker, so this is the only signal that some pool slots # are sitting on stuck work. _compression_max_cfg = config.compression_max_workers if _compression_max_cfg is None: _compression_max = max(1, os.cpu_count() or 1) else: _compression_max = max(1, _compression_max_cfg) self.compression_max_workers: int = _compression_max self._compression_executor = concurrent.futures.ThreadPoolExecutor( max_workers=_compression_max, thread_name_prefix="headroom-compress", ) # Phase 3 (#1171): off-path background compression. When enabled, a # cold-start-large request (frozen=0 + large live zone) forwards # uncompressed immediately and enqueues the compression here instead of # blocking the request thread under the 30s budget (which leaks a # non-preemptible worker -> executor saturation -> cascade). Default # off (opt-in), fail-open. Per-process, matching _compression_caches. self._background_compression_enabled: bool = os.environ.get( "HEADROOM_BACKGROUND_COMPRESSION", "" ).strip().lower() in ("1", "true", "yes", "on") try: self._background_compression_min_tokens: int = int( os.environ.get("HEADROOM_BACKGROUND_COMPRESSION_MIN_TOKENS", "50000") ) except ValueError: self._background_compression_min_tokens = 50000 # Dedicated single thread: no-timeout background jobs never contend with # the request-path executor (Phase 3, #1171). Lazy -- no thread spawns # until the first off-path job is submitted. self._background_compression_executor = concurrent.futures.ThreadPoolExecutor( max_workers=1, thread_name_prefix="headroom-bg-compress" ) self._background_compressor = BackgroundCompressor(self._run_compression_background) # Gauge: currently-running compression tasks. Mutated under # ``_compression_metrics_lock`` from worker threads + the asyncio # event loop. self._compression_queued: int = 0 self._compression_queued_max: int = 0 self._compression_queue_timeouts: int = 0 self._compression_queue_wait_seconds_total: float = 0.0 self._compression_queue_wait_seconds_max: float = 0.0 self._compression_in_flight: int = 0 # High-water mark for in-flight count. self._compression_in_flight_max: int = 0 self._compression_run_seconds_total: float = 0.0 self._compression_run_seconds_max: float = 0.0 # Counter: threads that finished AFTER their asyncio future hit the # timeout. Stuck-thread leak indicator. self._compression_leaked_threads: int = 0 self._compression_metrics_lock = threading.Lock() # Backend for Anthropic API (direct, LiteLLM, or any-llm) # Supports: "anthropic" (direct), "bedrock", "vertex", "litellm-", or "anyllm" self.anthropic_backend: Backend | None = create_proxy_backend( backend=config.backend, anyllm_provider=config.anyllm_provider, bedrock_region=config.bedrock_region, bedrock_profile=config.bedrock_profile, logger=logger, openai_api_url=config.openai_api_url, anyllm_backend_cls=AnyLLMBackend, litellm_backend_cls=LiteLLMBackend, ) # Request counter for IDs self._request_counter = 0 self._request_counter_lock = asyncio.Lock() # CCR tool injectors (one per provider) self.anthropic_tool_injector = CCRToolInjector( provider="anthropic", inject_tool=config.ccr_inject_tool, inject_system_instructions=config.ccr_inject_system_instructions, ) self.openai_tool_injector = CCRToolInjector( provider="openai", inject_tool=config.ccr_inject_tool, inject_system_instructions=config.ccr_inject_system_instructions, ) # CCR Response Handler (handles CCR tool calls automatically) self.ccr_response_handler = ( CCRResponseHandler( ResponseHandlerConfig( enabled=True, max_retrieval_rounds=config.ccr_max_retrieval_rounds, ) ) if config.ccr_handle_responses else None ) # CCR Context Tracker (tracks compressed content across turns) self.ccr_context_tracker = ( ContextTracker( ContextTrackerConfig( enabled=True, proactive_expansion=config.ccr_proactive_expansion, max_proactive_expansions=config.ccr_max_proactive_expansions, ) ) if config.ccr_context_tracking else None ) # Turn counter for context tracking self._turn_counter = 0 # Memory Handler (persistent user memory) self.memory_handler: MemoryHandler | None = None if config.memory_enabled and config.stateless: # Persistent memory writes a SQLite DB + markdown files to disk, # which stateless mode forbids. Memory is cross-session learning and # is contradictory with an ephemeral/read-only deployment, so we # disable it rather than persist. Run without --stateless to use it. logger.warning( "Memory is disabled in stateless mode (it persists to disk). " "Run without --stateless to enable persistent memory." ) elif config.memory_enabled: # Resolve memory DB path: empty → project-scoped default _mem_db_path = config.memory_db_path if not _mem_db_path: _mem_dir = Path.cwd() / ".headroom" _mem_dir.mkdir(parents=True, exist_ok=True) _mem_db_path = str(_mem_dir / "memory.db") logger.info(f"Memory: Project-scoped DB at {_mem_db_path}") # PR-B6: translate the string-typed ``ProxyConfig.memory_mode`` # into the typed ``MemoryMode`` enum. Unknown values raise # loudly per the no-silent-fallback policy. from headroom.proxy.memory_handler import MemoryMode try: _memory_mode = MemoryMode(config.memory_mode) except ValueError as exc: raise ValueError( f"Invalid memory_mode={config.memory_mode!r}; " f"expected one of {[m.value for m in MemoryMode]}" ) from exc from headroom.memory.storage_router import MemoryStorageMode try: _storage_mode = MemoryStorageMode(config.memory_storage_mode) except ValueError as exc: raise ValueError( f"Invalid memory_storage_mode={config.memory_storage_mode!r}; " f"expected one of {[m.value for m in MemoryStorageMode]}" ) from exc memory_config = MemoryConfig( enabled=True, backend=config.memory_backend, db_path=_mem_db_path, inject_tools=config.memory_inject_tools, use_native_tool=config.memory_use_native_tool, inject_context=config.memory_inject_context, top_k=config.memory_top_k, min_similarity=config.memory_min_similarity, mode=_memory_mode, storage_mode=_storage_mode, project_root_override=config.memory_project_root_override, qdrant_url=config.memory_qdrant_url, qdrant_host=config.memory_qdrant_host, qdrant_port=config.memory_qdrant_port, qdrant_api_key=config.memory_qdrant_api_key, neo4j_uri=config.memory_neo4j_uri, neo4j_user=config.memory_neo4j_user, neo4j_password=config.memory_neo4j_password, bridge_enabled=config.memory_bridge_enabled, bridge_md_paths=config.memory_bridge_md_paths, bridge_md_format=config.memory_bridge_md_format, bridge_auto_import=config.memory_bridge_auto_import, bridge_export_path=config.memory_bridge_export_path, ) self.memory_handler = MemoryHandler( memory_config, agent_type=config.traffic_learning_agent_type, ) # Migration UX (GH #462). When the user is on the new # project-scoped default but a legacy single-file DB exists # with prior memories, surface that clearly so it doesn't # look like an upgrade ate their data. if _storage_mode is MemoryStorageMode.PROJECT: _legacy_path = Path(_mem_db_path) if _legacy_path.exists() and _legacy_path.stat().st_size > 0: logger.info( "event=memory_storage_legacy_detected path=%s mode=project " "hint=pass_--memory-storage=global_to_reach_pre-fix_memories", _legacy_path, ) # The Memory Bridge binds to the single legacy backend at # init time; it doesn't (yet) follow per-project routing. # Warn so users running bridge + project mode aren't # surprised that only the legacy DB syncs with markdown. if config.memory_bridge_enabled and _storage_mode is MemoryStorageMode.PROJECT: logger.warning( "event=memory_bridge_global_only mode=project " "hint=bridge_syncs_only_the_legacy_DB_today_per-project_bridge_follow-up_planned" ) # Usage Reporter (license validation + phone-home for managed/enterprise). # Suppressed entirely in offline mode — the air-gap switch must stop all # egress, including license phone-home, even when a key is configured. self.usage_reporter: UsageReporter | None = None if config.license_key and not (config.offline or is_offline()): from headroom.telemetry.reporter import UsageReporter self.usage_reporter = UsageReporter( license_key=config.license_key, cloud_url=config.license_cloud_url, report_interval=config.license_report_interval, ) # Traffic Learner (live pattern extraction from proxy traffic) # Only activates with --learn flag; requires --memory for backend self.traffic_learner: TrafficLearner | None = None self.traffic_learning_agent_type: str = config.traffic_learning_agent_type if config.traffic_learning_enabled: from headroom.memory.traffic_learner import TrafficLearner self.traffic_learner = TrafficLearner( user_id=os.environ.get("HEADROOM_USER_ID", os.environ.get("USER", "default")), agent_type=config.traffic_learning_agent_type, min_evidence=config.traffic_learning_min_evidence, ) # Code graph file watcher (live reindex on file changes) self.code_graph_watcher: CodeGraphWatcher | None = None # type: ignore[annotation-unchecked] if config.code_graph_watcher: from headroom.graph.watcher import CodeGraphWatcher self.code_graph_watcher = CodeGraphWatcher(project_dir=Path.cwd()) if self.code_graph_watcher.start(): logger.info("Code graph: file watcher started") else: self.code_graph_watcher = None self.pipeline_extensions.emit( PipelineStage.SETUP, operation="proxy.setup", metadata={ "mode": self.config.mode, "optimize": self.config.optimize, "backend": self.config.backend, "memory_enabled": self.config.memory_enabled, }, ) async def _run_compression_in_executor( self, fn, # noqa: ANN001 — caller-supplied no-arg sync callable *, timeout: float, ): """Run a synchronous compression callable on the bounded executor with cancel-aware metrics. Replaces ``asyncio.wait_for(asyncio.to_thread(fn), timeout=...)``. Why a dedicated executor: the proxy's compression path is CPU-bound Rust work that releases the GIL via ``py.allow_threads``. Sharing the loop's default executor (used by ``asyncio.to_thread``) means a burst of slow compressions can starve unrelated ``to_thread`` callers (file IO, etc.). The compression executor is sized independently via ``config.compression_max_workers``. Why "cancel-aware metrics": when ``asyncio.wait_for`` times out, it cancels the *asyncio future*. The underlying ``concurrent.futures.Future`` from ``run_in_executor`` cannot actually cancel a thread that has started — Python has no way to preempt running CPython bytecode or in-flight Rust calls. The worker keeps running to completion, ignored. We detect this by marking the call timed out on the asyncio side and incrementing ``_compression_leaked_threads`` from the worker's ``finally`` block after it eventually finishes. Jobs that time out before a worker starts are removed from the queued gauge instead. Operators can see leaked-thread rate and queue pressure climbing in ``/stats`` before the pool fills up. Args: fn: A no-arg sync callable that runs the compression. Must not raise asyncio Cancellation; if it does, the wrapper still decrements the in-flight gauge but the leaked-thread counter may double-count. timeout: Wall-clock timeout for the asyncio side. The executor worker keeps running past this (Python limitation — see above), but at least the awaiter unblocks. Returns: Whatever ``fn()`` returns. Raises: ``asyncio.TimeoutError`` if the callable doesn't return within ``timeout``. Any exception raised by ``fn`` propagates unchanged. """ loop = asyncio.get_running_loop() queued_at = time.monotonic() state = {"queued": True, "timed_out": False} with self._compression_metrics_lock: self._compression_queued += 1 if self._compression_queued > self._compression_queued_max: self._compression_queued_max = self._compression_queued def _wrapped(): # noqa: ANN202 started_at = time.monotonic() queue_wait = started_at - queued_at with self._compression_metrics_lock: if state["queued"]: self._compression_queued -= 1 state["queued"] = False self._compression_queue_wait_seconds_total += queue_wait if queue_wait > self._compression_queue_wait_seconds_max: self._compression_queue_wait_seconds_max = queue_wait self._compression_in_flight += 1 if self._compression_in_flight > self._compression_in_flight_max: self._compression_in_flight_max = self._compression_in_flight try: return fn() finally: elapsed = time.monotonic() - started_at with self._compression_metrics_lock: self._compression_in_flight -= 1 self._compression_run_seconds_total += elapsed if elapsed > self._compression_run_seconds_max: self._compression_run_seconds_max = elapsed if state["timed_out"]: self._compression_leaked_threads += 1 future = loop.run_in_executor(self._compression_executor, _wrapped) try: return await asyncio.wait_for(future, timeout=timeout) except asyncio.TimeoutError: with self._compression_metrics_lock: state["timed_out"] = True if state["queued"]: self._compression_queued -= 1 state["queued"] = False self._compression_queue_timeouts += 1 raise async def _run_compression_background(self, fn): # noqa: ANN001, ANN201 """Run a compression callable on the shared executor with NO request- coupled deadline (Phase 3 off-path, #1171). Unlike ``_run_compression_in_executor`` there is no ``asyncio.wait_for`` and no leaked-thread accounting: no caller is waiting, so a slow run backs up the background queue rather than starving the request executor. Runs on the dedicated single-thread background executor. """ loop = asyncio.get_running_loop() return await loop.run_in_executor(self._background_compression_executor, fn) def _get_compression_cache(self, session_id: str) -> CompressionCache: """Get or create a CompressionCache for a session. Thread-safe under `_compression_caches_lock`: a concurrent pair of `_get_compression_cache(session_id)` calls (e.g. two async requests for the same conversation) must return the **same** instance, otherwise the per-session cache state splits and the two halves diverge across requests. """ with self._compression_caches_lock: if session_id not in self._compression_caches: from headroom.cache.compression_cache import CompressionCache # Evict oldest caches if at capacity if len(self._compression_caches) >= MAX_COMPRESSION_CACHE_SESSIONS: # Remove oldest quarter to amortize cleanup cost oldest_keys = list(self._compression_caches.keys())[ : MAX_COMPRESSION_CACHE_SESSIONS // 4 ] for key in oldest_keys: del self._compression_caches[key] logger.info( "Evicted %d compression caches (exceeded %d max sessions)", len(oldest_keys), MAX_COMPRESSION_CACHE_SESSIONS, ) self._compression_caches[session_id] = CompressionCache() return self._compression_caches[session_id] def _setup_code_aware(self, config: ProxyConfig, transforms: list) -> str: """Set up code-aware compression if enabled. Args: config: Proxy configuration transforms: Transform list to append to Returns: Status string for logging: 'enabled', 'disabled', 'available', 'unavailable' """ if config.code_aware_enabled: if is_tree_sitter_available(): code_config = CodeCompressorConfig( preserve_imports=True, preserve_signatures=True, preserve_type_annotations=True, ) # CodeAware runs after the content/structure transforms. # Phase B PR-B1 retired the trailing context_manager so we # append rather than insert(-1). transforms.append(CodeAwareCompressor(code_config)) return "enabled" else: logger.warning( "Code-aware compression requested but tree-sitter not installed. " "Install with: pip install headroom-ai[code]" ) return "unavailable" else: if is_tree_sitter_available(): return "available" # Available but not enabled return "disabled" def _eager_preload_transforms(self) -> tuple[dict[str, str], list[dict[str, str]]]: """Eagerly load every compressor/parser/detector once (dedup by ``id()``). Pure load: returns the merged ``eager_status`` plus the per-transform status dicts for the caller to merge into ``self.warmup`` on the main thread (``WarmupRegistry`` is not written off-thread). This runs via ``asyncio.to_thread`` so a slow or hung native model load cannot keep startup from binding the port (#790). """ eager_status: dict[str, str] = {} transform_statuses: list[dict[str, str]] = [] seen_transform_ids: set[int] = set() for pipeline in (self.anthropic_pipeline, self.openai_pipeline): for transform in pipeline.transforms: if id(transform) in seen_transform_ids: continue seen_transform_ids.add(id(transform)) if not hasattr(transform, "eager_load_compressors"): continue try: transform_status = transform.eager_load_compressors() except Exception as exc: logger.warning( "Eager preload failed for %s: %s", type(transform).__name__, exc, ) continue if not isinstance(transform_status, dict): continue # Merge: later writers win only if the key wasn't set. Preload a # transform ONCE — if another pipeline also has # ``eager_load_compressors`` it contributes only new keys. for key, value in transform_status.items(): eager_status.setdefault(key, value) transform_statuses.append(transform_status) return eager_status, transform_statuses async def startup(self): """Initialize async resources.""" self._get_shutdown_event().clear() self.pipeline_extensions.emit( PipelineStage.PRE_START, operation="proxy.startup", metadata={"port": self.config.port, "host": self.config.host}, ) # Resolve TLS verification: a custom CA bundle (corporate PKI) if one # is configured, else a strict-relaxed default context when # HEADROOM_TLS_STRICT=0, else httpx's default strict verification. _verify = build_httpx_verify() _http2, _client_kwargs = _provider_httpx_client_options(self.config, _verify) self.http_client = httpx.AsyncClient(http2=_http2, **_client_kwargs) # Reuse the primary client when HTTP/2 is already off; otherwise keep a # dedicated HTTP/1.1 client for ChatGPT passthrough. self.http_client_h1 = ( self.http_client if not _http2 else httpx.AsyncClient(http2=False, **_client_kwargs) ) logger.info("Headroom Proxy started (version %s)", __version__) logger.info(f"Optimization: {'ENABLED' if self.config.optimize else 'DISABLED'}") self.config.mode = normalize_proxy_mode(self.config.mode) logger.info(f"Mode: {self.config.mode}") if self.config.mode == PROXY_MODE_TOKEN: logger.info(" Prefix freeze: re-freeze after compression") logger.info(" Read protection window: 30%% of excluded-tool messages") logger.info(" CCR TTL: extended for session lifetime") logger.info(" Compression cache: active") if self.config.mode == PROXY_MODE_CACHE: logger.info(" Prefix freeze: strict (all prior turns immutable)") logger.info(" Mutations: latest turn only") logger.info(f"Caching: {'ENABLED' if self.config.cache_enabled else 'DISABLED'}") logger.info(f"Rate Limiting: {'ENABLED' if self.config.rate_limit_enabled else 'DISABLED'}") logger.info( f"Connection Pool: max_connections={self.config.max_connections}, " f"max_keepalive={self.config.max_keepalive_connections}, " f"http2={'ENABLED' if _http2 else 'DISABLED'}" ) # Unit 4 pre-upstream concurrency announcement. Report the resolved # value (auto-detected vs. explicit) so operators can correlate # ``pre_upstream_wait_ms`` log lines with the configured cap. if self.anthropic_pre_upstream_sem is None: logger.info("Anthropic pre-upstream concurrency: unbounded (explicitly disabled)") else: _explicit = self.config.anthropic_pre_upstream_concurrency _origin = "auto-detected" if _explicit is None else "explicit" logger.info( "Anthropic pre-upstream concurrency: %d (%s)", self.anthropic_pre_upstream_concurrency, _origin, ) logger.info( "Anthropic pre-upstream timeouts: acquire=%.1fs compression=%.1fs memory_context=%.1fs", self.anthropic_pre_upstream_acquire_timeout_seconds, float(COMPRESSION_TIMEOUT_SECONDS), self.anthropic_pre_upstream_memory_context_timeout_seconds, ) logger.info("Smart Routing: ENABLED (ContentRouter is always active)") # Eagerly load ALL compressors, parsers, and detectors at startup # This eliminates cold-start latency spikes on first requests. # Iterate BOTH pipelines (Anthropic + OpenAI) and dedupe transforms # by id() so shared-transform instances never load twice. The # resulting status dict is merged into ``self.warmup`` so /debug/warmup # (Unit 5) and /readyz have a single source of truth. self._kompress_status = "not installed" eager_status: dict[str, str] = {} if self.config.optimize: logger.info("Pre-loading compressors and parsers...") # Run the preload OFF the event loop with a bound. The loop body # already swallows per-transform Exceptions, so the only thing that # can still block ASGI lifespan startup (and therefore the socket # bind) is a hang or an uncatchable native stall during a model load # on Windows — the "never opens its port" failure in #790. Capping it # means startup always returns and uvicorn binds; on timeout the # transforms simply fall back to lazy loading on first use. transform_statuses: list[dict[str, str]] = [] try: eager_status, transform_statuses = await asyncio.wait_for( asyncio.to_thread(self._eager_preload_transforms), timeout=EAGER_PRELOAD_TIMEOUT_SECONDS, ) except Exception as exc: logger.warning( "Eager preload exceeded %.0fs or failed (%s); continuing so " "the proxy still binds — transforms load lazily on first use.", EAGER_PRELOAD_TIMEOUT_SECONDS, exc, ) eager_status, transform_statuses = {}, [] # Merge warmup status on the main thread (WarmupRegistry is not # written off-thread). for transform_status in transform_statuses: self.warmup.merge_transform_status(transform_status) # Update internal status from eager loading results if eager_status.get("kompress") == "enabled": self._kompress_status = "enabled" if eager_status.get("code_aware") == "enabled": self._code_aware_status = "enabled" # Log component status if self._kompress_status == "enabled": logger.info("Kompress: ENABLED (ModernBERT token compressor)") elif self.config.optimize: logger.info("Kompress: not installed (pip install headroom-ai[ml] for ML compression)") if self._code_aware_status == "enabled": logger.info("Code-Aware: ENABLED (AST-based compression)") if "tree_sitter" in eager_status: logger.info(f"Tree-Sitter: {eager_status['tree_sitter']}") elif self._code_aware_status == "lazy": logger.info("Code-Aware: LAZY (will load when code content detected)") elif self._code_aware_status == "available": logger.info("Code-Aware: available but disabled (use --code-aware)") elif self._code_aware_status == "unavailable": logger.info("Code-Aware: not installed (pip install headroom-ai[code])") elif self._code_aware_status == "disabled": logger.info("Code-Aware: DISABLED") if eager_status.get("magika") == "enabled": logger.info("Magika: ENABLED (ML content detection)") if self.memory_handler: if ( self.config.memory_backend == "qdrant-neo4j" and not self.config.memory_neo4j_password ): logger.warning( "NEO4J password is not set — using default credentials is insecure in production" ) self.warmup.memory_backend.mark_loading() try: await self.memory_handler.ensure_initialized() except Exception as exc: # pragma: no cover - defensive self.warmup.memory_backend.mark_error(str(exc)) logger.warning("Memory: backend initialization failed (startup continues): %s", exc) memory_status = self.memory_handler.health_status() if memory_status.get("initialized"): self.warmup.memory_backend.mark_loaded( handle=self.memory_handler, backend=memory_status.get("backend"), ) # Force one embed call so the ONNX graph is compiled now, # not lazily during the first request. Best-effort — any # failure is swallowed inside warmup_embedder. self.warmup.memory_embedder.mark_loading() warmed = await self.memory_handler.warmup_embedder() if warmed: self.warmup.memory_embedder.mark_loaded() else: # Not an error — e.g. qdrant-neo4j has no embedder slot # we can reach, or the backend simply exposes no handle. self.warmup.memory_embedder.mark_null() else: if self.warmup.memory_backend.status != "error": self.warmup.memory_backend.mark_null() self.warmup.memory_embedder.mark_null() logger.info( "Memory: ENABLED " f"(backend={memory_status['backend']}, initialized={memory_status['initialized']})" ) else: logger.info("Memory: DISABLED") # CCR status ccr_features = [] if self.config.ccr_inject_tool: ccr_features.append("tool_injection") if self.config.ccr_handle_responses: ccr_features.append("response_handling") if self.config.ccr_context_tracking: ccr_features.append("context_tracking") if self.config.ccr_proactive_expansion: ccr_features.append("proactive_expansion") if ccr_features: logger.info(f"CCR (Compress-Cache-Retrieve): ENABLED ({', '.join(ccr_features)})") else: logger.info("CCR: DISABLED") logger.info(f"Savings history: {self.metrics.savings_tracker.storage_path}") # Reset and rebuild the quota tracker registry for this server instance. # reset_quota_registry() ensures a clean slate when the proxy is restarted # (e.g. in tests that spin up multiple app instances in the same process). reset_quota_registry() registry = get_quota_registry() tracker = configure_subscription_tracker( poll_interval_s=self.config.subscription_poll_interval_s, active_window_s=self.config.subscription_active_window_s, enabled=self.config.subscription_tracking_enabled, ) registry.register(tracker) registry.register(get_codex_rate_limit_state()) registry.register(get_copilot_quota_tracker()) await registry.start_all() if self.config.subscription_tracking_enabled: logger.info( "Subscription tracking: ENABLED " f"(poll_interval={self.config.subscription_poll_interval_s}s, " f"active_window={self.config.subscription_active_window_s}s)" ) else: logger.info("Subscription tracking: DISABLED") copilot_tracker = get_copilot_quota_tracker() if copilot_tracker.is_available(): logger.info("GitHub Copilot quota tracking: ENABLED") else: logger.info( "GitHub Copilot quota tracking: DISABLED " "(set GITHUB_TOKEN or GITHUB_COPILOT_GITHUB_TOKEN to enable)" ) # Log local telemetry status so operators can see it in the log stream. # Nothing is sent externally — telemetry is collected locally only (the # anonymous telemetry beacon was removed); operational metrics export # only to your own OTEL collector via HEADROOM_OTEL_METRICS_*. if is_telemetry_enabled(): logger.info( "Local telemetry: ENABLED (aggregate stats, local only — nothing sent " "externally). Opt out: HEADROOM_TELEMETRY=off or --no-telemetry" ) else: logger.info( "Local telemetry: DISABLED (off by default — opt in: " "HEADROOM_TELEMETRY=on or --telemetry)" ) self.pipeline_extensions.emit( PipelineStage.POST_START, operation="proxy.startup", metadata={ "port": self.config.port, "host": self.config.host, "warmup": self.warmup.to_dict(), }, ) async def shutdown(self): """Cleanup async resources.""" self._get_shutdown_event().set() if self.http_client_h1 and self.http_client_h1 is not self.http_client: await self.http_client_h1.aclose() self.http_client_h1 = None if self.http_client: await self.http_client.aclose() self.http_client = None if self.memory_handler and hasattr(self.memory_handler, "close"): await self.memory_handler.close() with contextlib.suppress(Exception): from headroom.models.ml_models import MLModelRegistry released_models = [] released_models.extend(MLModelRegistry.unload_prefix("technique_router:")) released_models.extend(MLModelRegistry.unload_prefix("siglip:")) if released_models: logger.info("Released image optimizer models: %s", ", ".join(released_models)) # Stop all quota trackers via the registry await get_quota_registry().stop_all() # Persist any savings the tracker's write throttle is still holding, so # a graceful shutdown doesn't drop the last few requests' totals. with contextlib.suppress(Exception): self.metrics.savings_tracker.flush() # Print final stats self._print_summary() def _print_summary(self): """Print session summary.""" m = self.metrics logger.info("=" * 70) logger.info("HEADROOM PROXY SESSION SUMMARY") logger.info("=" * 70) logger.info(f"Total requests: {m.requests_total}") logger.info(f"Cached responses: {m.requests_cached}") logger.info(f"Rate limited: {m.requests_rate_limited}") logger.info(f"Failed: {m.requests_failed}") logger.info(f"Input tokens: {m.tokens_input_total:,}") logger.info(f"Output tokens: {m.tokens_output_total:,}") logger.info(f"Tokens saved: {m.tokens_saved_total:,}") # Active-compression ratio: savings as a fraction of what we # *attempted* to compress (extracted units + tool schema), # NOT the whole request. The full-request denominator is # dominated by frozen prefix bytes (instructions, user msgs, # prior turns) that we never touch — including them collapses # the headline number even on sessions where every attempted # compression succeeded. attempted = getattr(m, "attempted_input_tokens_total", 0) if attempted > 0: # `attempted` is pre-compression; savings rate is plain # saved / attempted. savings_pct = (m.tokens_saved_total / attempted) * 100 logger.info(f"Active compression: {savings_pct:.1f}%") logger.info(f" (attempted tokens: {attempted:,})") if m.tokens_input_total > 0: whole_request_pct = ( m.tokens_saved_total / (m.tokens_input_total + m.tokens_saved_total) ) * 100 logger.info(f"Of total wire traffic: {whole_request_pct:.2f}%") if m.latency_count > 0: avg_latency = m.latency_sum_ms / m.latency_count logger.info(f"Avg latency: {avg_latency:.0f}ms") logger.info("=" * 70) async def _record_request_outcome(self, outcome: RequestOutcome) -> None: """Single funnel for per-request bookkeeping. Thin wrapper around :func:`headroom.proxy.outcome.emit_request_outcome` so call sites can write ``await self._record_request_outcome(outcome)`` (idiomatic) instead of ``await emit_request_outcome(self, outcome)``. The real implementation lives in ``outcome.py`` as a free function so test dummies and provider mixins can call it without inheriting from ``HeadroomProxy``. See ``docs/superpowers/specs/P0-proxy-pipeline-audit.md`` for the divergence catalog this funnel collapses. """ from headroom.proxy.outcome import emit_request_outcome await emit_request_outcome(self, outcome) async def _next_request_id(self) -> str: """Generate unique request ID.""" async with self._request_counter_lock: self._request_counter += 1 return f"hr_{int(time.time())}_{self._request_counter:06d}" def _extract_tags(self, headers: dict) -> dict[str, str]: """Backwards-compat wrapper around :func:`extract_tags`. Handlers call ``extract_tags(headers)`` directly. Kept here for any external caller still using ``proxy._extract_tags(headers)``. """ from headroom.proxy.helpers import extract_tags return extract_tags(headers) def _get_shutdown_event(self) -> asyncio.Event: event = getattr(self, "_shutdown_event", None) if event is None: event = asyncio.Event() self._shutdown_event = event return event async def _wait_for_retry_delay_or_shutdown(self, delay_seconds: float) -> bool: try: await asyncio.wait_for(self._get_shutdown_event().wait(), timeout=delay_seconds) return True except asyncio.TimeoutError: return False def _shutdown_retry_response(self, method: str, url: str) -> httpx.Response: return httpx.Response( 503, request=httpx.Request(method, url), headers={"content-type": "application/json", "retry-after": "0"}, json={ "error": { "type": "shutdown", "message": "Proxy is shutting down; retry backoff cancelled.", } }, ) async def _retry_request( self, method: str, url: str, headers: dict, body: dict, stream: bool = False, *, original_body_bytes: bytes | None = None, body_mutated: bool = True, mutation_reasons: list[str] | None = None, request_id: str | None = None, forwarder_name: str = "server", path_for_log: str | None = None, timeout: httpx.Timeout | float | None = None, ) -> httpx.Response: """Make request with retry and exponential backoff. Byte-faithful forwarding (PR-A3, fixes P0-2): * If ``original_body_bytes`` is provided AND ``body_mutated`` is ``False``, the original bytes are forwarded verbatim. SHA-256 of upstream-received bytes equals client-sent bytes. * Otherwise the body dict is canonically re-serialized via ``serialize_body_canonical`` (compact separators, ensure_ascii=False). * ``HEADROOM_PROXY_PYTHON_FORWARDER_MODE=legacy_json_kwarg`` is an explicit operator opt-in for emergency rollback to the old ``httpx ... json=body`` behavior. The default ``body_mutated=True`` preserves backward compatibility for callers that still pass only ``body`` (e.g. CCR continuations construct their body from scratch, so canonical serialization is correct and original bytes do not exist). """ from headroom.proxy.body_forwarding import prepare_outbound_body_bytes from headroom.proxy.helpers import log_outbound_request last_error = None reasons = list(mutation_reasons or []) outbound_bytes, source = prepare_outbound_body_bytes( body=body, original_body_bytes=original_body_bytes, body_mutated=body_mutated, ) outbound_headers = {**headers, "content-type": "application/json"} log_outbound_request( forwarder=forwarder_name, method=method, path=path_for_log or url, body_bytes_count=len(outbound_bytes), body_mutated=body_mutated, mutation_reasons=reasons, request_id=request_id, source=source, ) post_kwargs: dict = {"content": outbound_bytes, "headers": outbound_headers} if timeout is not None: post_kwargs["timeout"] = timeout for attempt in range(self.config.retry_max_attempts): try: if stream: # For streaming, we return early - retry happens at higher level return await self.http_client.post( # type: ignore[union-attr] url, **post_kwargs ) else: response = await self.http_client.post( # type: ignore[union-attr] url, **post_kwargs ) # Transient overloads (429 rate-limit, 529 overloaded): # retry honoring Retry-After, but return verbatim once # exhausted — a clean overload signal beats a synthesized 5xx # (extends #1221 to 529, Anthropic's overloaded_error). if response.status_code in RETRYABLE_OVERLOAD_STATUSES: if ( not self.config.retry_enabled or attempt >= self.config.retry_max_attempts - 1 ): return response delay_ms = retry_after_ms( response, self.config.retry_max_delay_ms ) or jitter_delay_ms( self.config.retry_base_delay_ms, self.config.retry_max_delay_ms, attempt, ) logger.warning( f"Upstream {response.status_code} (attempt {attempt + 1}), " f"retrying in {delay_ms:.0f}ms" ) if await self._wait_for_retry_delay_or_shutdown(delay_ms / 1000): logger.info( "Shutdown interrupted retry backoff for %s %s", method, path_for_log or "", ) return self._shutdown_retry_response(method, url) continue # Don't retry other client errors (4xx) if 400 <= response.status_code < 500: return response # Retry other server errors (5xx) if response.status_code >= 500: raise httpx.HTTPStatusError( f"Server error: {response.status_code}", request=response.request, response=response, ) return response # httpx.TransportError covers ConnectError, the timeout family, and — # crucially — the protocol errors (Local/RemoteProtocolError, e.g. an # HTTP/2 `StreamReset`) that a poisoned shared h2 connection raises on # every in-flight request. Retrying drops the bad connection and # re-sends on a fresh one instead of collapsing to a 502. (#1639) except (httpx.TransportError, httpx.HTTPStatusError) as e: last_error = e if not self.config.retry_enabled or attempt >= self.config.retry_max_attempts - 1: # On exhaustion, preserve the upstream 5xx status (e.g. 503 # Service Unavailable, 500, 502, 504) so the client can apply # its own retry/backoff. Collapsing every exhausted 5xx into a # generic 502 hides the retryable signal and makes clients give # up. The 429/529 overload statuses are already returned # verbatim by the RETRYABLE_OVERLOAD_STATUSES branch above and # never reach here. ConnectError/TimeoutException carry no # response, so those still raise. if isinstance(e, httpx.HTTPStatusError) and e.response is not None: return e.response raise # Exponential backoff with jitter delay_with_jitter = jitter_delay_ms( self.config.retry_base_delay_ms, self.config.retry_max_delay_ms, attempt, ) logger.warning( f"Request failed (attempt {attempt + 1}), retrying in {delay_with_jitter:.0f}ms: {e}" ) if await self._wait_for_retry_delay_or_shutdown(delay_with_jitter / 1000): logger.info( "Shutdown interrupted retry backoff for %s %s", method, path_for_log or "", ) return self._shutdown_retry_response(method, url) if last_error is None: raise RuntimeError( "retry loop exhausted with no error recorded; retry_max_attempts must be >= 1" ) raise last_error async def _log_toin_stats_periodically(interval_seconds: int = 300) -> None: """Background task that logs TOIN stats periodically. Args: interval_seconds: How often to log stats (default: 5 minutes). """ while True: await asyncio.sleep(interval_seconds) try: toin = get_toin() stats = toin.get_stats() total_compressions = stats.get("total_compressions", 0) if total_compressions > 0: patterns = stats.get("patterns_tracked", 0) retrievals = stats.get("total_retrievals", 0) retrieval_rate = stats.get("global_retrieval_rate", 0.0) logger.info( "TOIN: %d patterns, %d compressions, %d retrievals, %.1f%% retrieval rate", patterns, total_compressions, retrievals, retrieval_rate * 100, ) except Exception as e: logger.debug("Failed to log TOIN stats: %s", e) def _register_memory_components(proxy: HeadroomProxy, tracker: MemoryTracker) -> None: """Register all memory-tracked components with the tracker. This function is idempotent - it checks if components are already registered. Args: proxy: The HeadroomProxy instance. tracker: The MemoryTracker instance. """ # Register compression store (global singleton) if "compression_store" not in tracker.registered_components: store = get_compression_store() tracker.register("compression_store", store.get_memory_stats) # Register semantic cache (instance on proxy) if proxy.cache and "semantic_cache" not in tracker.registered_components: tracker.register("semantic_cache", proxy.cache.get_memory_stats) # Register request logger (instance on proxy) if proxy.logger and "request_logger" not in tracker.registered_components: tracker.register("request_logger", proxy.logger.get_memory_stats) # Register batch context store (global singleton) if "batch_context_store" not in tracker.registered_components: try: from ..ccr.batch_store import get_batch_context_store batch_store = get_batch_context_store() if hasattr(batch_store, "get_memory_stats"): tracker.register("batch_context_store", batch_store.get_memory_stats) except ImportError: pass # Note: graph_store and vector_index are created per-user within the # LocalMemoryBackend, not as global singletons. They would need to be # registered when the memory system is initialized with specific backends. def _request_is_loopback(request: Request) -> bool: """Return True iff the caller is on loopback by *both* peer IP and Host header. Mirrors the two-gate check in :func:`loopback_guard.require_loopback` (loopback client IP + loopback ``Host`` header, the DNS-rebinding defence) but returns a bool instead of raising. Endpoints use it to vary their payload — serving sensitive sub-blocks (upstream URLs, per-request logs) only to loopback callers — rather than 404ing network callers that still have a legitimate use for the non-sensitive aggregate fields. """ from headroom.proxy.loopback_guard import is_loopback_host, is_loopback_host_header client = getattr(request, "client", None) client_host = getattr(client, "host", None) if client is not None else None try: host_header = request.headers.get("host") except AttributeError: host_header = None # The Host-header gate is the DNS-rebinding defence and always applies. if not is_loopback_host_header(host_header): return False # Genuine loopback peer (native run, or curl inside the container). if is_loopback_host(client_host): return True # Containerized dashboards: when Headroom runs in a bridge-network # container, a browser on the host reaches it via the container's # gateway, so ``request.client.host`` is the gateway IP, not 127.0.0.1 # — and the per-request logs / upstream URLs get stripped even though # the operator is local. Treat a peer inside an operator-configured # trusted-gateway CIDR as loopback-equivalent. Opt-in and empty by # default (HEADROOM_PROXY_TRUSTED_GATEWAY_CIDRS), so this is a no-op # unless the operator explicitly allow-lists their container gateway. from headroom.proxy.forwarded_headers import ( load_trusted_gateway_cidrs, peer_is_trusted_gateway, ) return peer_is_trusted_gateway(client_host, load_trusted_gateway_cidrs()) _is_known_websocket_callback_failure = is_known_websocket_callback_failure _tool_schema_saved_from_tags = tool_schema_saved_from_tags def create_app(config: ProxyConfig | None = None) -> FastAPI: """Create FastAPI application.""" if not FASTAPI_AVAILABLE: raise ImportError("FastAPI required. Install: pip install fastapi uvicorn httpx") from contextlib import asynccontextmanager # Always-on file logging to ~/.headroom/logs/ for `headroom perf` analysis. # Installed here (not at module import) so importing headroom.proxy.server # in tests or library contexts does not silently attach a RotatingFileHandler # to the user's live proxy.log. _setup_file_logging() config = config or ProxyConfig() # Air-gap master switch. Propagate config.offline to the env so the # env-based egress predicates (telemetry, update check, license) all honor # it, force HF/transformers offline before any model code loads, and # announce that every outbound path is disabled. if config.offline: os.environ.setdefault("HEADROOM_OFFLINE", "1") if is_offline(): apply_offline_env() logger.warning( "event=proxy_offline_mode air-gap active — all outbound egress disabled " "(telemetry, update check, license reporter, HuggingFace downloads)" ) proxy = HeadroomProxy(config) # cc-switch reconciler (opt-in: HEADROOM_CC_SWITCH_RECONCILE=1). # Keeps Headroom in the request path while cc-switch overwrites # ~/.claude/settings.json on every provider switch. See # headroom/proxy/cc_switch_reconciler.py for the full rationale. from headroom.proxy.cc_switch_reconciler import ( CCSwitchReconciler, reconciler_enabled, ) _cc_reconciler: CCSwitchReconciler | None = None if reconciler_enabled(): _cc_proxy_port = config.port if hasattr(config, "port") else 8787 def _set_anthropic_upstream(url: str) -> None: from headroom.providers.registry import _normalize_api_url HeadroomProxy.ANTHROPIC_API_URL = _normalize_api_url( url, default=DEFAULT_ANTHROPIC_API_URL ) _cc_reconciler = CCSwitchReconciler( proxy_url=f"http://127.0.0.1:{_cc_proxy_port}", default_upstream=DEFAULT_ANTHROPIC_API_URL, set_upstream=_set_anthropic_upstream, ) # Single-worker-owner lock. With uvicorn workers > 1, each worker runs the # lifespan independently. A file lock elects ONE owner worker so that # single-instance background tasks (currently the cc-switch reconciler) run # once across all workers instead of N times. from headroom import paths as _hr_paths _beacon_lock_path = _hr_paths.beacon_lock_path(config.port) _beacon_lock_fd: list = [None] # mutable holder for the lock file descriptor _beacon_is_owner: list = [False] def _try_acquire_beacon_lock() -> bool: """Try to acquire the beacon file lock (non-blocking). Returns True if this process is the beacon owner. """ if not HAS_FCNTL: return True fd = None try: _beacon_lock_path.parent.mkdir(parents=True, exist_ok=True) fd = open(_beacon_lock_path, "w") # noqa: SIM115 fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) fd.write(str(os.getpid())) fd.flush() _beacon_lock_fd[0] = fd return True except OSError: if fd is not None: fd.close() return False def _release_beacon_lock() -> None: """Release the beacon file lock.""" fd = _beacon_lock_fd[0] if fd: try: if HAS_FCNTL: fcntl.flock(fd, fcntl.LOCK_UN) fd.close() except Exception: pass _beacon_lock_fd[0] = None try: _beacon_lock_path.unlink(missing_ok=True) except Exception: pass @asynccontextmanager async def lifespan(app: FastAPI): # type: ignore[no-untyped-def] # Hotfix-A0: Rust core deployment smoke test. Refuse to accept # traffic if the Rust extension is missing unless the operator # explicitly opted out with HEADROOM_REQUIRE_RUST_CORE=false. See # Finding #2 in HEADROOM_PROXY_LOG_FINDINGS_2026_05_03.md. # `_check_rust_core` either returns ("loaded"|"disabled", _) or # calls `sys.exit(78)` — execution past this line implies the # rust_core_status is recorded. _rust_core_status, _rust_core_error = _check_rust_core() app.state.rust_core_status = _rust_core_status app.state.rust_core_error = _rust_core_error configure_otel_metrics(OTelMetricsConfig.from_env(default_service_name="headroom-proxy")) configure_langfuse_tracing( LangfuseTracingConfig.from_env(default_service_name="headroom-proxy") ) app.state.started_at = time.time() app.state.ready = False app.state.startup_error = None await initialize_context_tool_session_baseline() try: try: previous_handler = _install_loop_exception_handler() # Startup await proxy.startup() if config.periodic_toin_stats_enabled: asyncio.create_task(_log_toin_stats_periodically()) if proxy.usage_reporter: await proxy.usage_reporter.start(proxy) if proxy.traffic_learner: await proxy.traffic_learner.start() if proxy._background_compression_enabled: await proxy._background_compressor.start() # Elect the single owner worker (first worker wins the lock). _beacon_is_owner[0] = _try_acquire_beacon_lock() # Only the owner worker runs the reconciler. With uvicorn # workers > 1 each worker runs this lifespan; without this guard # every worker would watch + rewrite settings.json concurrently # and each process would hold its own HeadroomProxy.ANTHROPIC_API_URL, # so workers could disagree on the upstream. if _cc_reconciler is not None and _beacon_is_owner[0]: await _cc_reconciler.start() app.state.ready = True yield except Exception as exc: app.state.startup_error = str(exc) raise finally: loop: asyncio.AbstractEventLoop | None previous: LoopExceptionHandler | None try: loop = asyncio.get_running_loop() previous = previous_handler except RuntimeError: loop = None previous = app.state.previous_loop_exception_handler if loop is not None: loop.set_exception_handler(previous) app.state.ready = False # Shutdown if _cc_reconciler is not None: await _cc_reconciler.stop() if _beacon_is_owner[0]: _release_beacon_lock() if proxy.usage_reporter: await proxy.usage_reporter.stop() if proxy.traffic_learner: await proxy.traffic_learner.stop() if proxy._background_compression_enabled: await proxy._background_compressor.stop() proxy._background_compression_executor.shutdown(wait=False) if proxy.code_graph_watcher: proxy.code_graph_watcher.stop() await proxy.shutdown() shutdown_headroom_tracing() shutdown_otel_metrics() app = FastAPI( title="Headroom Proxy", description="Production-ready LLM optimization proxy", version=__version__, lifespan=lifespan, ) loop_health_state: LoopHealthState = { "status": "healthy", "known_failures": 0, "last_known_failure": None, } app.state.proxy = proxy app.state.started_at = None app.state.ready = False app.state.startup_error = None app.state.loop_callback_health = loop_health_state app.state.loop_exception_handler = None app.state.previous_loop_exception_handler = None # Set by the lifespan startup smoke test (`_check_rust_core`). Default # "missing" means lifespan hasn't run yet — anything reading /health # before startup completes (rare; lifespan runs before the first # request) sees an honest "missing" rather than a stale "loaded". app.state.rust_core_status = "missing" app.state.rust_core_error = None def _iso_utc_now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def _uptime_seconds() -> float: started_at = getattr(app.state, "started_at", None) if not isinstance(started_at, int | float): return 0.0 return round(max(0.0, time.time() - float(started_at)), 3) def _component_health( *, enabled: bool, ready: bool, **details: Any, ) -> dict[str, Any]: status = "disabled" if not enabled else ("healthy" if ready else "unhealthy") return { "enabled": enabled, "ready": (ready if enabled else True), "status": status, **details, } def _health_checks() -> dict[str, dict[str, Any]]: memory_status = ( proxy.memory_handler.health_status() if proxy.memory_handler else { "enabled": False, "backend": None, "initialized": False, "native_tool": False, "bridge_enabled": False, } ) memory_enabled = bool(memory_status.get("enabled", False)) memory_initialized = bool(memory_status.get("initialized", False)) return { "startup": _component_health( enabled=True, ready=bool(getattr(app.state, "ready", False)), error=getattr(app.state, "startup_error", None), ), "http_client": _component_health( enabled=True, ready=proxy.http_client is not None, ), "cache": _component_health( enabled=config.cache_enabled, ready=(proxy.cache is not None), ), "rate_limiter": _component_health( enabled=config.rate_limit_enabled, ready=(proxy.rate_limiter is not None), ), "memory": _component_health( enabled=memory_enabled, ready=memory_initialized, backend=memory_status["backend"], initialized=memory_initialized, native_tool=bool(memory_status.get("native_tool", False)), bridge_enabled=bool(memory_status.get("bridge_enabled", False)), ), "upstream": _component_health( enabled=os.environ.get("HEADROOM_SKIP_UPSTREAM_CHECK", "").strip() != "1", ready=bool(_upstream_check_cache["ok"]), url=_upstream_check_cache["url"], error=_upstream_check_cache["error"], ), "kompress": _component_health( enabled=not config.disable_kompress, ready=proxy.warmup.kompress.status == "loaded", backend=proxy.warmup.kompress.info.get("backend", None), ), } def _runtime_payload() -> dict[str, Any]: ws_registry = getattr(proxy, "ws_sessions", None) ws_active_sessions = ws_registry.active_count() if ws_registry is not None else 0 ws_active_relay_tasks = ( ws_registry.active_relay_task_count() if ws_registry is not None else 0 ) # Snapshot compression executor metrics under their lock (gauges # mutated by worker threads; not safe to read without). with proxy._compression_metrics_lock: _comp_queued = proxy._compression_queued _comp_queued_max = proxy._compression_queued_max _comp_queue_timeouts = proxy._compression_queue_timeouts _comp_queue_wait_total = proxy._compression_queue_wait_seconds_total _comp_queue_wait_max = proxy._compression_queue_wait_seconds_max _comp_in_flight = proxy._compression_in_flight _comp_in_flight_max = proxy._compression_in_flight_max _comp_run_total = proxy._compression_run_seconds_total _comp_run_max = proxy._compression_run_seconds_max _comp_leaked = proxy._compression_leaked_threads return { "anthropic_pre_upstream": { "enabled": proxy.anthropic_pre_upstream_sem is not None, "resolved_concurrency": proxy.anthropic_pre_upstream_concurrency, "source": ( "auto" if config.anthropic_pre_upstream_concurrency is None else "explicit" ), "acquire_timeout_seconds": proxy.anthropic_pre_upstream_acquire_timeout_seconds, "compression_timeout_seconds": float(COMPRESSION_TIMEOUT_SECONDS), "memory_context_timeout_seconds": ( proxy.anthropic_pre_upstream_memory_context_timeout_seconds ), "codex_ws_gated": False, }, "compression_executor": { "max_workers": proxy.compression_max_workers, "queued": _comp_queued, "queued_max": _comp_queued_max, "queue_timeouts_total": _comp_queue_timeouts, "queue_wait_seconds_total": _comp_queue_wait_total, "queue_wait_seconds_max": _comp_queue_wait_max, "running": _comp_in_flight, "in_flight": _comp_in_flight, "in_flight_max": _comp_in_flight_max, "run_seconds_total": _comp_run_total, "run_seconds_max": _comp_run_max, "leaked_threads_total": _comp_leaked, "source": ("auto" if config.compression_max_workers is None else "explicit"), }, "websocket_sessions": { "active_sessions": ws_active_sessions, "active_relay_tasks": ws_active_relay_tasks, }, } def _loop_callback_payload() -> LoopHealthState: return { "status": loop_health_state["status"], "known_failures": loop_health_state["known_failures"], "last_known_failure": loop_health_state["last_known_failure"], } def _install_loop_exception_handler() -> LoopExceptionHandler | None: try: loop = asyncio.get_running_loop() except RuntimeError: return None previous_handler = loop.get_exception_handler() def _loop_exception_handler( _loop: asyncio.AbstractEventLoop, context: dict[str, Any] ) -> None: if _is_known_websocket_callback_failure(context): loop_health_state["status"] = "unhealthy" loop_health_state["known_failures"] += 1 loop_health_state["last_known_failure"] = { "message": context.get("message"), "exception": str(context.get("exception")) if context.get("exception") else None, } return delegate_handler = app.state.previous_loop_exception_handler if delegate_handler is not None: delegate_handler(_loop, context) return _loop.default_exception_handler(context) loop.set_exception_handler(_loop_exception_handler) app.state.loop_exception_handler = _loop_exception_handler app.state.previous_loop_exception_handler = previous_handler return previous_handler def _health_payload(*, include_config: bool) -> dict[str, Any]: checks = _health_checks() # Kompress is an optional soft component: model downloads lazily on # first use, so "not ready" (cold cache) must not degrade overall health. ready = all(check["ready"] for name, check in checks.items() if name != "kompress") payload: dict[str, Any] = { "service": "headroom-proxy", "status": "healthy" if ready else "unhealthy", "ready": ready, "version": __version__, "timestamp": _iso_utc_now(), "uptime_seconds": _uptime_seconds(), "checks": checks, "runtime": _runtime_payload(), # Hotfix-A0: surface rust core load state so operators can alert # on `rust_core != "loaded"` (Finding #2). "rust_core": getattr(app.state, "rust_core_status", "missing"), } rust_core_error = getattr(app.state, "rust_core_error", None) if rust_core_error: payload["rust_core_error"] = rust_core_error deployment_profile = os.environ.get("HEADROOM_DEPLOYMENT_PROFILE") if deployment_profile: payload["deployment"] = { "profile": deployment_profile, "preset": os.environ.get("HEADROOM_DEPLOYMENT_PRESET"), "runtime": os.environ.get("HEADROOM_DEPLOYMENT_RUNTIME"), "supervisor": os.environ.get("HEADROOM_DEPLOYMENT_SUPERVISOR"), "scope": os.environ.get("HEADROOM_DEPLOYMENT_SCOPE"), } if include_config: profile_kwargs = proxy_pipeline_kwargs(config) effective_target_ratio = cast( float | None, profile_kwargs.get("target_ratio", config.target_ratio), ) payload["config"] = { "backend": config.backend, "optimize": config.optimize, "cache": config.cache_enabled, "rate_limit": config.rate_limit_enabled, "disable_kompress": config.disable_kompress, "disable_kompress_fallback": config.disable_kompress_fallback, "disable_kompress_anthropic": config.disable_kompress_anthropic, "disable_kompress_openai": config.disable_kompress_openai, "memory": config.memory_enabled, "learn": config.traffic_learning_enabled, "code_graph": config.code_graph_watcher, "anthropic_api_url": config.anthropic_api_url, "openai_api_url": config.openai_api_url, "gemini_api_url": config.gemini_api_url, "cloudcode_api_url": config.cloudcode_api_url, "vertex_api_url": config.vertex_api_url, "savings_profile": config.savings_profile, "target_ratio": effective_target_ratio, "target_savings_percent": ( round(max(0.0, min(1.0, 1.0 - float(effective_target_ratio))) * 100, 1) if effective_target_ratio is not None else None ), "compress_user_messages": bool( profile_kwargs.get("compress_user_messages", config.compress_user_messages) ), "compress_system_messages": bool( profile_kwargs.get( "compress_system_messages", config.compress_system_messages, ) ), "protect_recent": profile_kwargs.get( "read_protection_window", config.protect_recent, ), "protect_analysis_context": profile_kwargs.get( "protect_analysis_context", config.protect_analysis_context, ), "min_tokens_to_crush": profile_kwargs.get( "min_tokens_to_compress", config.min_tokens_to_crush, ), "max_items_after_crush": profile_kwargs.get( "max_items_after_crush", config.max_items_after_crush, ), "smart_crusher_with_compaction": profile_kwargs.get( "smart_crusher_with_compaction", config.smart_crusher_with_compaction, ), "force_kompress": bool(profile_kwargs.get("force_kompress", False)), "accuracy_guard": config.accuracy_guard, # Live (per-request) env knobs the proxy reads after startup. # Surfaced so `headroom wrap` can see what a reused proxy is # actually using and hot-sync it via /admin/runtime-env. "runtime_env": runtime_env.effective_runtime_env(), "pid": os.getpid(), } return payload # --------------------------------------------------------------------------- # Upstream connectivity check — cached to avoid hammering the upstream on # every /readyz poll. Set HEADROOM_SKIP_UPSTREAM_CHECK=1 to opt out (e.g. # in air-gapped or test environments where the upstream isn't reachable at # startup time). # --------------------------------------------------------------------------- _UPSTREAM_CHECK_TTL = 30.0 # seconds _upstream_check_cache: dict[str, Any] = { "expires_at": 0.0, "ok": True, "error": None, "url": None, } _upstream_check_lock = asyncio.Lock() def _upstream_target_url() -> str: """Return the primary upstream base URL to probe.""" # Use the resolved API target from the provider runtime so we respect # any overrides set by ProxyConfig.anthropic_api_url / env vars. return proxy.provider_runtime.api_targets.anthropic async def _check_upstream() -> None: """Probe the upstream API endpoint and update the cached result. Uses a HEAD request with a 5-second timeout — just enough to verify TLS + TCP reachability without triggering an inference call. """ if os.environ.get("HEADROOM_SKIP_UPSTREAM_CHECK", "").strip() == "1": # Opt-out: treat upstream as always reachable. _upstream_check_cache["ok"] = True _upstream_check_cache["error"] = None _upstream_check_cache["expires_at"] = time.monotonic() + _UPSTREAM_CHECK_TTL return now = time.monotonic() # Fast-path: return if the cached result is still fresh (no lock needed # for a simple float comparison — worst case we re-check twice). if now < _upstream_check_cache["expires_at"]: return async with _upstream_check_lock: # Re-check inside the lock to handle concurrent waiters. if time.monotonic() < _upstream_check_cache["expires_at"]: return url = _upstream_target_url() _upstream_check_cache["url"] = url client = proxy.http_client if client is None: _upstream_check_cache["ok"] = False _upstream_check_cache["error"] = "proxy client not initialised" _upstream_check_cache["expires_at"] = time.monotonic() + _UPSTREAM_CHECK_TTL return try: resp = await client.head(url, timeout=5.0) # Any HTTP response (even 4xx/5xx) means TLS+TCP worked. _ = resp.status_code _upstream_check_cache["ok"] = True _upstream_check_cache["error"] = None except Exception as exc: # noqa: BLE001 _upstream_check_cache["ok"] = False _upstream_check_cache["error"] = str(exc) _upstream_check_cache["expires_at"] = time.monotonic() + _UPSTREAM_CHECK_TTL # CORS: scoped to localhost by default. The old wildcard origin combined # with allow_credentials=True let any web page the user had open read the # proxy's content endpoints (e.g. /v1/retrieve returns raw, uncompressed # tool outputs) via a cross-origin fetch to 127.0.0.1 (CWE-346). # # The default matches any loopback origin on any port via a regex, so it # works regardless of the --port the proxy was started on without the app # needing to know its own bound port (the port lives in the CLI/uvicorn # layer, not in ProxyConfig). Set HEADROOM_CORS_ORIGINS (comma-separated) # to pin an explicit allowlist for Docker or remote-dashboard deployments; # "*" restores the old wildcard behaviour if the operator accepts the risk. _default_loopback_origin_regex = r"https?://(localhost|127\.0\.0\.1|\[::1\])(:\d+)?" _cors_origins_env = os.environ.get("HEADROOM_CORS_ORIGINS", "").strip() if _cors_origins_env: _cors_allow_origins = [o.strip() for o in _cors_origins_env.split(",") if o.strip()] _cors_allow_origin_regex: str | None = None else: _cors_allow_origins = [] _cors_allow_origin_regex = _default_loopback_origin_regex app.add_middleware( CORSMiddleware, allow_origins=_cors_allow_origins, allow_origin_regex=_cors_allow_origin_regex, allow_credentials=False, allow_methods=["GET", "POST"], allow_headers=["Content-Type", "Authorization"], ) # X-Headroom-Stack: SDK adapters (TS openai/anthropic/etc.) tag their # requests so telemetry can segment by integration surface. Registered # before extension middleware so any extension-level auth/guards run # outermost and we don't count requests they reject. @app.middleware("http") async def _record_headroom_stack(request, call_next): started = time.perf_counter() inbound_id = f"inbound-{time.time_ns()}" # Project attribution: an explicit X-Headroom-Project header wins # (claude/codex wraps); otherwise a /p/ base-URL prefix (aider, # Copilot BYOK, Cursor — clients that cannot send custom headers). # The prefix strip mutates the scope, so it must happen before # request.url is first accessed (Starlette caches the URL). prefix_project = strip_project_path_prefix(request.scope) path = request.url.path method = request.method query = request.url.query headers = dict(request.headers.items()) set_current_project(classify_project(headers) or prefix_project) # Path-based Codex identification: stamp X-Client: codex on the # Responses endpoint for callers that don't otherwise classify (e.g. # Codex Desktop, whose User-Agent isn't a known codex UA). Without it # the backend refuses oversized # requests with a 413 on a compression timeout, which Codex treats as a # hard connection failure. Mutating scope["headers"] before call_next # makes every downstream classify_client(headers) read "codex". if should_stamp_codex_client(path, headers): request.scope["headers"].append((b"x-client", b"codex")) client = getattr(request, "client", None) client_addr = "" if client is not None: client_host = getattr(client, "host", None) client_port = getattr(client, "port", None) client_addr = f"{client_host}:{client_port}" if client_port else str(client_host) try: proxy.metrics.record_inbound_request(method=method, path=path) except Exception: logger.debug("record_inbound_request failed", exc_info=True) try: from headroom.proxy.helpers import redact_for_wire_debug safe_headers = redact_for_wire_debug(headers) except Exception: safe_headers = {"redaction_error": True} logger.info( "event=proxy_inbound_request id=%s method=%s path=%s query=%s client=%s " "content_length=%s headers=%s", inbound_id, method, path, query, client_addr, request.headers.get("content-length", ""), json.dumps(safe_headers, ensure_ascii=False, default=str), ) if request.url.path.startswith("/v1/"): stack = request.headers.get("x-headroom-stack") if stack: try: proxy.metrics.record_stack(stack) except Exception: logger.debug("record_stack failed", exc_info=True) try: response = await call_next(request) except asyncio.CancelledError: try: proxy.metrics.record_inbound_aborted(reason="cancelled") except Exception: logger.debug("record_inbound_aborted failed", exc_info=True) logger.info( "event=proxy_inbound_request_aborted id=%s method=%s path=%s reason=cancelled " "duration_ms=%.2f", inbound_id, method, path, (time.perf_counter() - started) * 1000.0, ) raise except Exception as exc: try: proxy.metrics.record_inbound_aborted(reason=type(exc).__name__) except Exception: logger.debug("record_inbound_aborted failed", exc_info=True) logger.error( "event=proxy_inbound_request_aborted id=%s method=%s path=%s reason=%s " "duration_ms=%.2f", inbound_id, method, path, type(exc).__name__, (time.perf_counter() - started) * 1000.0, exc_info=True, ) raise try: proxy.metrics.record_inbound_response(status_code=response.status_code) except Exception: logger.debug("record_inbound_response failed", exc_info=True) logger.info( "event=proxy_inbound_response id=%s method=%s path=%s status=%s duration_ms=%.2f", inbound_id, method, path, response.status_code, (time.perf_counter() - started) * 1000.0, ) return response # ── Security gate (registered last → runs outermost) ────────────────── # Three concerns, kept together because they all wrap every inbound # request: optional inbound auth on the data plane, response security # headers, and an audit trail for state-mutating admin endpoints. _proxy_token = config.proxy_token or os.environ.get("HEADROOM_PROXY_TOKEN") or None # Pre-encode once for constant-time comparison (compare_digest on str raises # TypeError for non-ASCII input, which would turn a 401 into a 500). _proxy_token_bytes = _proxy_token.encode("utf-8") if _proxy_token else b"" # Health/readiness probes must stay reachable without a token so # orchestrators can check a container that binds non-loopback. _AUTH_EXEMPT_PATHS = frozenset({"/health", "/healthz", "/livez", "/readyz"}) # Loud warning when a non-loopback bind has no token configured: that is the # exact shape (e.g. the Docker 0.0.0.0 image) that exposes unauthenticated # /v1/* routes to the surrounding network. if not _proxy_token and not is_loopback_host(getattr(config, "host", None)): logger.warning( "event=proxy_open_bind host=%s — proxy is bound to a non-loopback " "interface with no HEADROOM_PROXY_TOKEN set; the /v1/* data-plane " "routes are reachable WITHOUT authentication. Set HEADROOM_PROXY_TOKEN " "to require a bearer token from non-loopback callers.", getattr(config, "host", None), ) def _apply_security_headers(response) -> None: # setdefault: never clobber a header an upstream/handler already set. response.headers.setdefault("X-Content-Type-Options", "nosniff") response.headers.setdefault("X-Frame-Options", "DENY") response.headers.setdefault("Referrer-Policy", "no-referrer") response.headers.setdefault( "Strict-Transport-Security", "max-age=31536000; includeSubDomains" ) def _extract_proxy_token(headers) -> str | None: auth = str(headers.get("authorization") or "") if auth.lower().startswith("bearer "): return auth[7:].strip() or None raw = headers.get("x-headroom-proxy-token") return str(raw) if raw else None @app.middleware("http") async def _security_gate(request, call_next): # 1) Optional inbound auth. When a token is configured, require it on # non-loopback requests; loopback callers and health probes are # exempt. Loopback is the same trust boundary the admin/debug # endpoints already use (see loopback_guard). if _proxy_token: path = request.url.path client = getattr(request, "client", None) client_host = getattr(client, "host", None) if client is not None else None if path not in _AUTH_EXEMPT_PATHS and not is_loopback_host(client_host): provided = _extract_proxy_token(request.headers) if provided is None or not hmac.compare_digest( provided.encode("utf-8", "replace"), _proxy_token_bytes ): logger.warning( "event=proxy_auth_rejected path=%s client=%s reason=%s", path, client_host, "missing_token" if provided is None else "bad_token", ) rejection = JSONResponse(status_code=401, content={"error": "unauthorized"}) _apply_security_headers(rejection) return rejection response = await call_next(request) _apply_security_headers(response) # 2) Audit trail for admin / state-mutating endpoints. try: if is_auditable_path(request.url.path): record_admin_action( request=request, action="admin_request", status_code=response.status_code, ) except Exception: logger.debug("admin audit emission failed", exc_info=True) return response # Third-party proxy extensions (Enterprise, custom plugins). Discovered via # the `headroom.proxy_extension` entry-point group, but **opt-in only**: # only names listed in config.proxy_extensions (CLI: --proxy-extension, # env: HEADROOM_PROXY_EXTENSIONS) actually get installed. Discovery alone # never runs third-party code. An extension that raises from its install() # is a deliberate fail-closed signal and aborts startup. from headroom.proxy.extensions import install_all as _install_extensions _install_extensions(app, config, enabled=getattr(config, "proxy_extensions", None)) # Health & Metrics @app.get("/livez") async def livez(): callback_state = _loop_callback_payload() healthy = callback_state["status"] == "healthy" return JSONResponse( status_code=200 if healthy else 503, content={ "service": "headroom-proxy", "status": "healthy" if healthy else "unhealthy", "alive": healthy, "version": __version__, "timestamp": _iso_utc_now(), "uptime_seconds": _uptime_seconds(), "loop_health": callback_state, }, ) @app.get("/readyz") async def readyz(): await _check_upstream() payload = _health_payload(include_config=False) return JSONResponse(status_code=200 if payload["ready"] else 503, content=payload) @app.get("/health") async def health(request: Request): await _check_upstream() # /health echoes upstream API URLs + backend config (the `config` # block). That is operational detail an external scanner should not # see, so include it only for loopback callers; network callers get the # same body as /readyz (status + checks, no config). /livez and /readyz # remain the unauthenticated probes for orchestration health. payload = _health_payload(include_config=_request_is_loopback(request)) return JSONResponse(status_code=200, content=payload) # Loopback-only debug introspection (Unit 5). A remote IP gets 404 — # debug endpoints are invisible to external scanners. from headroom.proxy.debug_introspection import ( collect_tasks as _collect_tasks, ) from headroom.proxy.loopback_guard import require_loopback as _require_loopback @app.get("/admin/upstream", dependencies=[Depends(_require_loopback)]) async def get_upstream(): """Current Anthropic upstream + cc-switch reconciler state (loopback-only). Read-only. The upstream is mutated only via the in-process cc-switch reconciler (driven by ~/.claude/settings.json) — there is deliberately no HTTP write route, so a local process cannot redirect credential- bearing traffic to an arbitrary URL through this surface. """ return { "anthropic": HeadroomProxy.ANTHROPIC_API_URL, "cc_switch_reconcile": _cc_reconciler is not None, "captured_upstream": getattr(_cc_reconciler, "current_upstream", None), } @app.get("/debug/tasks", dependencies=[Depends(_require_loopback)]) async def debug_tasks(stack: bool = False): """Enumerate running asyncio tasks. Default is cheap — ``stack_depth`` is ``null`` in every entry so a storm snapshot does not walk 50+ coroutine frames synchronously. Pass ``?stack=true`` to compute ``stack_depth`` for each task (useful for single-shot human debugging). """ ws_registry = getattr(proxy, "ws_sessions", None) return JSONResponse( status_code=200, content=_collect_tasks(ws_registry, with_stack_depth=stack), ) @app.get("/debug/ws-sessions", dependencies=[Depends(_require_loopback)]) async def debug_ws_sessions(): ws_registry = getattr(proxy, "ws_sessions", None) snapshot = ws_registry.snapshot() if ws_registry is not None else [] return JSONResponse(status_code=200, content=snapshot) @app.get("/debug/warmup", dependencies=[Depends(_require_loopback)]) async def debug_warmup(): warmup_registry = getattr(proxy, "warmup", None) payload = warmup_registry.to_dict() if warmup_registry is not None else {} payload["runtime"] = _runtime_payload() return JSONResponse(status_code=200, content=payload) @app.post("/admin/runtime-env", dependencies=[Depends(_require_loopback)]) async def admin_runtime_env(request: Request): """Hot-reload live env knobs (the output-shaper family, the ast-grep read threshold) without restarting the proxy. Live knobs are read from the proxy's *process* environment, so a proxy that ``headroom wrap`` reused — rather than started — never sees values a user exported afterwards. Instead of a disruptive restart (cold ML load, dropped requests, lost caches), ``wrap`` POSTs the values here and the proxy applies them in memory, effective on the next request. Loopback-only. The body is a flat ``{ENV_NAME: "value"}`` map; unknown keys and non-string values are ignored. Returns what was applied plus the resulting live config. Last writer wins (overrides are global to the proxy, which is inherent — every wrapper shares one process). """ try: body = await request.json() except (ValueError, UnicodeDecodeError): body = None if not isinstance(body, dict): return JSONResponse( status_code=400, content={"error": "expected a JSON object of {ENV_NAME: value}"}, ) applied = runtime_env.set_overrides(body) if applied: logger.info("runtime-env hot-reload applied: %s", sorted(applied)) # Record which runtime-env keys changed (the "what" of a config # change) in addition to the generic admin-request audit emitted by # the security middleware. Values are intentionally omitted — keys # alone avoid logging any secret values that were set. record_admin_action( request=request, action="runtime_env_update", status_code=200, details={"changed_keys": sorted(applied)}, ) return JSONResponse( status_code=200, content={"applied": applied, "runtime_env": runtime_env.effective_runtime_env()}, ) @app.get("/dashboard", response_class=HTMLResponse) async def dashboard(): """Serve the Headroom dashboard UI.""" return get_dashboard_html() @app.get("/favicon.ico") async def favicon() -> Response: # Registered before register_provider_routes' catch-all passthrough # route so browsers' automatic favicon requests for /dashboard are # answered locally instead of being tunneled to the wrapped upstream # provider (GH #1787). return Response(status_code=204) DASHBOARD_STATS_CACHE_TTL_SECONDS = 5.0 _stats_snapshot_lock = asyncio.Lock() _stats_snapshot: dict[str, Any] = {"expires_at": 0.0, "value": None} THROUGHPUT_CACHE_TTL_SECONDS = 10.0 _throughput_cache_lock = asyncio.Lock() _throughput_cache: dict[str, Any] = {"expires_at": 0.0, "value": None} RECENT_REQUEST_LOG_WINDOW = 100 def _build_recent_request_payload(limit: int = RECENT_REQUEST_LOG_WINDOW) -> dict[str, Any]: recent_request_logs = proxy.logger.get_recent(limit) if proxy.logger else [] dashboard_recent_requests = [ { "request_id": log.get("request_id"), "timestamp": log.get("timestamp"), "provider": log.get("provider"), "model": log.get("model"), "input_tokens_original": log.get("input_tokens_original"), "input_tokens_optimized": log.get("input_tokens_optimized"), "output_tokens": log.get("output_tokens"), "tokens_saved": log.get("tokens_saved"), "savings_percent": log.get("savings_percent"), "optimization_latency_ms": log.get("optimization_latency_ms"), "total_latency_ms": log.get("total_latency_ms"), "transforms_applied": log.get("transforms_applied", []), "waste_signals": log.get("waste_signals"), "tool_schema_saved_tokens": _tool_schema_saved_from_tags(log.get("tags")), } for log in recent_request_logs if log.get("input_tokens_original") is not None and log.get("input_tokens_optimized") is not None ][-10:] return { "request_logs": recent_request_logs[-10:], "recent_requests": dashboard_recent_requests, } async def _build_stats_payload() -> dict[str, Any]: """Build the full `/stats` response payload. This is the main stats endpoint - it aggregates data from all subsystems: - Request metrics (total, cached, failed, by model/provider) - Token usage and savings - Cost tracking - Canonical persisted display_session metrics for downstream dashboards - Compression (CCR) statistics - Telemetry/TOIN (data flywheel) statistics - Cache and rate limiter stats """ m = proxy.metrics import time async with _throughput_cache_lock: now = time.time() if _throughput_cache["expires_at"] < now or _throughput_cache["value"] is None: def _compute_throughput(): from headroom.perf.analyzer import build_perf_summary, parse_log_files perf_report = parse_log_files(last_n_hours=1.0) return build_perf_summary(perf_report).get("throughput") try: throughput = await asyncio.to_thread(_compute_throughput) _throughput_cache["value"] = throughput _throughput_cache["expires_at"] = now + THROUGHPUT_CACHE_TTL_SECONDS except Exception as e: logger.warning("Failed to calculate throughput for stats: %s", e, exc_info=True) if _throughput_cache["value"] is None: _throughput_cache["value"] = None throughput = _throughput_cache["value"] # Calculate average latency avg_latency_ms = round(m.latency_sum_ms / m.latency_count, 2) if m.latency_count > 0 else 0 min_latency_ms = ( round(m.latency_min_ms, 2) if m.latency_count > 0 and m.latency_min_ms != float("inf") else 0 ) max_latency_ms = round(m.latency_max_ms, 2) if m.latency_count > 0 else 0 # Calculate Headroom overhead (optimization time only, excludes pass-through requests) avg_overhead_ms = ( round(m.overhead_sum_ms / m.overhead_count, 2) if m.overhead_count > 0 else 0 ) min_overhead_ms = ( round(m.overhead_min_ms, 2) if m.overhead_count > 0 and m.overhead_min_ms != float("inf") else 0 ) max_overhead_ms = round(m.overhead_max_ms, 2) if m.overhead_count > 0 else 0 # Calculate TTFB (time to first byte) avg_ttfb_ms = round(m.ttfb_sum_ms / m.ttfb_count, 2) if m.ttfb_count > 0 else 0 min_ttfb_ms = ( round(m.ttfb_min_ms, 2) if m.ttfb_count > 0 and m.ttfb_min_ms != float("inf") else 0 ) max_ttfb_ms = round(m.ttfb_max_ms, 2) if m.ttfb_count > 0 else 0 def _pct(part: int | float, whole: int | float) -> float: return round((float(part) / float(whole)) * 100.0, 2) if whole else 0.0 # Get compression store stats store = get_compression_store() compression_stats = store.get_stats() # Get telemetry/TOIN stats telemetry = get_telemetry_collector() telemetry_stats = telemetry.get_stats() # Get feedback loop stats feedback = get_compression_feedback() feedback_stats = feedback.get_stats() # Build prefix cache stats once (used in both prefix_cache and cost) prefix_cache_stats = _build_prefix_cache_stats(m, proxy.cost_tracker) # Fetch CLI filtering savings from the selected context tool. These # tokens are avoided before they reach model context. cli_filtering_stats = await asyncio.to_thread(_get_context_tool_stats) cli_filtering_tool = ( str(cli_filtering_stats.get("tool", "rtk")) if cli_filtering_stats else "rtk" ) cli_filtering_label = ( str(cli_filtering_stats.get("label", "RTK")) if cli_filtering_stats else "RTK" ) cli_tokens_avoided = ( cli_filtering_stats.get("tokens_saved", 0) if cli_filtering_stats else 0 ) cli_filtering_session = ( cli_filtering_stats.get("session", {}) if cli_filtering_stats else {} ) cli_filtering_lifetime = ( cli_filtering_stats.get("lifetime", {}) if cli_filtering_stats else {} ) rtk_tokens_avoided = cli_tokens_avoided if cli_filtering_tool == "rtk" else 0 lean_ctx_tokens_avoided = cli_tokens_avoided if cli_filtering_tool == "lean-ctx" else 0 cli_filtering_available = bool( cli_filtering_stats and cli_filtering_stats.get("installed", False) ) # Calculate total tokens before Headroom-side reduction. Proxy # compression and the configured context tool both remove tokens before # they reach model context, so dashboard-facing savings combines them. proxy_compression_tokens = m.tokens_saved_total all_layers_tokens_saved = proxy_compression_tokens + cli_tokens_avoided total_tokens_before = m.tokens_input_total + all_layers_tokens_saved proxy_total_before_compression = m.tokens_input_total + proxy_compression_tokens # `attempted_input_tokens` is the compressible-only denominator # (extracted units + tool schema). The "active compression" # ratio is what fraction of the tokens we *tried* to compress # actually got compressed. Excludes prefix-frozen content # (user/system messages, prior turns) we never touched — # otherwise the ratio is dominated by content we deliberately # avoided changing for prefix-cache safety. # `attempted_input_tokens_total` is already pre-compression: it # accumulates `unit.tokens_before` for each eligible unit that # reached the router, plus the original (pre-compaction) tool # schema size. So the savings rate is plain `saved / attempted` # — adding `saved` again would double-count. attempted_input_tokens = getattr(m, "attempted_input_tokens_total", 0) # New-content denominator: what the provider actually billed as # non-cache-read input (uncached + cache-write tokens, summed # across providers from response usage). Unlike # `proxy_total_before_compression`, this does NOT recount the # full transcript on every turn — a long session's history is # served from prefix cache, not re-billed, so it doesn't belong # in a denominator that claims to measure what compression had # any power over. Tokens Headroom removed never reached the # provider at all, so they're added back to form the baseline. _pc_totals = prefix_cache_stats.get("totals", {}) new_input_tokens = int(_pc_totals.get("uncached_input_tokens", 0) or 0) + int( _pc_totals.get("cache_write_tokens", 0) or 0 ) # Build human-readable summary summary = _build_session_summary( proxy, m, prefix_cache_stats, cli_tokens_avoided, total_tokens_before ) # DEBUG: log the summary payload for external upsert consumers try: logger.debug("/stats summary data: %r", summary) except Exception: logger.warning("Failed to log /stats summary payload") # Compression cache stats (token mode). Snapshot the cache list under # the dict lock so a concurrent eviction can't mutate the dict while # we iterate. Each per-session `get_stats()` is independently # thread-safe via the cache's own internal lock. compression_cache_stats: dict = {} if proxy.config.mode == PROXY_MODE_TOKEN and proxy._compression_caches: with proxy._compression_caches_lock: _caches_snapshot = list(proxy._compression_caches.values()) _active_sessions = len(proxy._compression_caches) total_entries = 0 total_hits = 0 total_misses = 0 total_tokens_saved = 0 for cache in _caches_snapshot: s = cache.get_stats() total_entries += s.get("entries", 0) total_hits += s.get("hits", 0) total_misses += s.get("misses", 0) total_tokens_saved += s.get("total_tokens_saved", 0) compression_cache_stats = { "mode": PROXY_MODE_TOKEN, "active_sessions": _active_sessions, "total_entries": total_entries, "total_hits": total_hits, "total_misses": total_misses, "hit_rate": round(total_hits / max(1, total_hits + total_misses) * 100, 1), "total_tokens_saved": total_tokens_saved, } else: compression_cache_stats = {"mode": proxy.config.mode} # Build unified savings summary (all layers) cache_net_usd = prefix_cache_stats.get("totals", {}).get("net_savings_usd", 0.0) total_tokens_all_layers = all_layers_tokens_saved persistent_savings = m.savings_tracker.stats_preview() display_session = persistent_savings.get("display_session", {}) recent_request_logs = proxy.logger.get_recent(10_000) if proxy.logger else [] recent_request_payload = _build_recent_request_payload() # Tool-schema deferral savings: tool-definition tokens kept out of the # model's context by deferring heavy schemas until they're needed # (native tool-search injection + any registered turn-hook tools # rewrite). Attributed to Headroom only — see _tool_schema_saved_from_tags. # Aggregated over the recent request-log window. tool_schema_tokens = 0 tool_schema_requests = 0 for _ts_log in recent_request_logs: _ts_saved = _tool_schema_saved_from_tags(_ts_log.get("tags")) if _ts_saved > 0: tool_schema_tokens += _ts_saved tool_schema_requests += 1 agent_usage = _build_agent_usage_summary( recent_request_logs, requests_by_provider=dict(m.requests_by_provider), requests_by_model=dict(m.requests_by_model), global_before_tokens=proxy_total_before_compression, global_after_tokens=m.tokens_input_total, global_tokens_saved=proxy_compression_tokens, global_output_tokens=m.tokens_output_total, ) # Output-side reduction (counterfactual estimate from the shaper's # ledger). Distinct from input compression above: these are OUTPUT # tokens the model didn't emit because we steered verbosity / routed # effort down. Always labelled estimated-vs-measured + a CI so it's # never mistaken for an exact count. Best-effort — never break /stats. output_reduction: dict[str, Any] = {"available": False} try: from headroom.proxy.output_savings import get_recorder _oest = get_recorder().estimate() if _oest.n_requests > 0: output_reduction = { "available": True, "method": _oest.kind, # "measured" | "estimated" "tokens_saved": round(_oest.tokens_saved), "baseline_tokens": round(_oest.baseline_tokens), "reduction_percent": round(_oest.pct, 1), "ci_low_percent": round(_oest.ci_low_pct, 1), "ci_high_percent": round(_oest.ci_high_pct, 1), "requests": _oest.n_requests, } except Exception: # pragma: no cover - defensive pass return { "summary": summary, "agent_usage": agent_usage, "savings": { "total_tokens": total_tokens_all_layers, "per_project": persistent_savings.get("projects", {}), "by_layer": { "cli_filtering": { "tool": cli_filtering_tool, "label": cli_filtering_label, "available": cli_filtering_available, "tokens": cli_tokens_avoided, "tokens_saved": cli_tokens_avoided, "session": cli_filtering_session, "lifetime": cli_filtering_lifetime, "session_savings_pct": ( cli_filtering_stats.get("session_savings_pct") if cli_filtering_stats else None ), "lifetime_savings_pct": ( cli_filtering_stats.get("lifetime_avg_savings_pct") if cli_filtering_stats else None ), "refresh_interval_seconds": ( cli_filtering_stats.get("refresh_interval_seconds") if cli_filtering_stats else None ), "included_in": "tokens.saved", "description": ( f"Tokens avoided by CLI output filtering ({cli_filtering_label}) " "before reaching context. " "Included in dashboard token savings, but not in dollar savings." ), }, "compression": { "tokens": proxy_compression_tokens, "proxy_tokens": proxy_compression_tokens, "cli_filtering_tokens": cli_tokens_avoided, "rtk_tokens": rtk_tokens_avoided, "lean_ctx_tokens": lean_ctx_tokens_avoided, "all_layers_tokens": all_layers_tokens_saved, "description": ( "Tokens removed by Headroom proxy compression. " "Dashboard token savings also includes CLI context-tool filtering." ), }, "prefix_cache": { "discount_usd": round(cache_net_usd, 4), "description": ( "Cost discount from provider prefix caching. " "Headroom's CacheAligner improves hit rates; " "baseline caching is provider-native." ), }, "output_shaping": { **output_reduction, "description": ( "OUTPUT tokens the model didn't emit because the shaper " "steered verbosity / routed effort down. Counterfactual — " "shown as an estimate (vs a learned baseline) or measured " "(A/B holdout), always with a confidence band." ), }, "tool_search": { "tokens": tool_schema_tokens, "tokens_saved": tool_schema_tokens, "requests": tool_schema_requests, "window": len(recent_request_logs), "description": ( "Tool-definition tokens kept out of the model's context " "by deferring heavy tool schemas until they're searched " "for. Counted only when Headroom performed the deferral — " "not when the client (e.g. Claude Code / Codex) already " "had tool search enabled. Aggregated over the recent " "request window." ), }, }, }, "requests": { "total": m.requests_total, "cached": m.requests_cached, "rate_limited": m.requests_rate_limited, "failed": m.requests_failed, "by_provider": dict(m.requests_by_provider), "by_model": dict(m.requests_by_model), "by_stack": dict(m.requests_by_stack), }, "tokens": { "input": m.tokens_input_total, "output": m.tokens_output_total, "output_saved": output_reduction.get("tokens_saved", 0), "output_reduction_percent": output_reduction.get("reduction_percent", 0), "output_reduction": output_reduction, "saved": all_layers_tokens_saved, "proxy_compression_saved": proxy_compression_tokens, "cli_filtering_saved": cli_tokens_avoided, "rtk_saved": rtk_tokens_avoided, "lean_ctx_saved": lean_ctx_tokens_avoided, "cli_tokens_avoided": cli_tokens_avoided, "proxy_total_before_compression": proxy_total_before_compression, "total_before_compression": total_tokens_before, "all_layers_saved": all_layers_tokens_saved, # Compressible-only denominator: tokens we extracted as # candidates + tool-schema tokens we compacted. Excludes # frozen-prefix content (user msgs, system prompt, prior # turns) that we deliberately don't touch. Already # pre-compression — do NOT add `tokens_saved` again. "proxy_attempted_tokens": attempted_input_tokens, # Active compression: savings as a fraction of what we # *tried* to compress. The number the dashboard headline # should show — it answers "are we doing well *when we # have something to compress?*" rather than diluting the # win by frozen-prefix bytes we never touched. "active_savings_percent": round( (proxy_compression_tokens / attempted_input_tokens * 100) if attempted_input_tokens > 0 else 0, 2, ), # Whole-request ratio kept for transparency. Heavily # diluted by frozen prefix on Codex-style requests # where most input is non-compressible by design. "proxy_savings_percent": round( (proxy_compression_tokens / proxy_total_before_compression * 100) if proxy_total_before_compression > 0 else 0, 2, ), # New-content-relative rate: savings as a fraction of the # input that would have newly entered context (provider- # billed uncached + cache-write tokens, plus the tokens # compression removed before they could be billed). The # whole-request ratios above recount the FULL transcript # every turn, so a 200-turn session counts its history # 200x into the denominator and long-running sessions # (1M-context models never compact) read as ~0% no # matter how well compression performs on new content. # Guarded on new_input_tokens > 0 (not the full sum): the # cache accumulators only see requests with cache # activity, so a deployment with no cache metrics (e.g. # Bedrock) would otherwise divide savings by themselves # and report ~100%. No usage data -> report 0, not a lie. "new_input_tokens": new_input_tokens, "new_input_savings_percent": round( (proxy_compression_tokens / (new_input_tokens + proxy_compression_tokens) * 100) if new_input_tokens > 0 else 0, 2, ), "savings_percent": round( (all_layers_tokens_saved / total_tokens_before * 100) if total_tokens_before > 0 else 0, 2, ), "all_layers_savings_percent": round( (all_layers_tokens_saved / total_tokens_before * 100) if total_tokens_before > 0 else 0, 2, ), }, "latency": { "average_ms": avg_latency_ms, "min_ms": min_latency_ms, "max_ms": max_latency_ms, "total_requests": m.latency_count, }, "overhead": { "average_ms": avg_overhead_ms, "min_ms": min_overhead_ms, "max_ms": max_overhead_ms, }, "ttfb": { "average_ms": avg_ttfb_ms, "min_ms": min_ttfb_ms, "max_ms": max_ttfb_ms, }, "pipeline_timing": { name: { "average_ms": round( m.transform_timing_sum[name] / m.transform_timing_count[name], 2 ), "max_ms": round(m.transform_timing_max[name], 2), "count": m.transform_timing_count[name], } for name in sorted(m.transform_timing_sum.keys()) } if m.transform_timing_sum else {}, "compressions_by_strategy": dict(m.compressions_by_strategy), "tokens_saved_by_strategy": dict(m.tokens_saved_by_strategy), "codex_ws": { "units_total": m.codex_ws_units_total, "units_modified_total": m.codex_ws_units_modified_total, "units_by_strategy": dict(m.codex_ws_units_by_strategy), "units_by_category": dict(m.codex_ws_units_by_category), "units_by_content_type": dict(m.codex_ws_units_by_content_type), "units_by_text_shape": dict(m.codex_ws_units_by_text_shape), "units_to_kompress_total": m.codex_ws_units_to_kompress_total, "units_kompress_attempted_total": m.codex_ws_units_kompress_attempted_total, "units_to_kompress_percent": _pct( m.codex_ws_units_to_kompress_total, m.codex_ws_units_total, ), "units_kompress_attempted_percent": _pct( m.codex_ws_units_kompress_attempted_total, m.codex_ws_units_total, ), "unit_elapsed_ms": { "average": round( m.codex_ws_unit_elapsed_ms_sum / m.codex_ws_units_total, 2, ) if m.codex_ws_units_total else 0.0, "max": round(m.codex_ws_unit_elapsed_ms_max, 2), }, "unit_bytes_sum": m.codex_ws_unit_bytes_sum, "unit_tokens_before_sum": m.codex_ws_unit_tokens_before_sum, "unit_tokens_after_sum": m.codex_ws_unit_tokens_after_sum, "unit_tokens_saved_sum": m.codex_ws_unit_tokens_saved_sum, "frames_attempted_total": m.codex_ws_frames_attempted_total, "frames_compressed_total": m.codex_ws_frames_compressed_total, "frames_failed_total": m.codex_ws_frames_failed_total, "frames_to_kompress_total": m.codex_ws_frames_to_kompress_total, "frames_kompress_attempted_total": (m.codex_ws_frames_kompress_attempted_total), "frames_to_kompress_percent": _pct( m.codex_ws_frames_to_kompress_total, m.codex_ws_frames_attempted_total, ), "frames_kompress_attempted_percent": _pct( m.codex_ws_frames_kompress_attempted_total, m.codex_ws_frames_attempted_total, ), "frame_elapsed_ms": { "average": round( m.codex_ws_frame_elapsed_ms_sum / m.codex_ws_frames_attempted_total, 2, ) if m.codex_ws_frames_attempted_total else 0.0, "max": round(m.codex_ws_frame_elapsed_ms_max, 2), }, "frame_bytes_before_sum": m.codex_ws_frame_bytes_before_sum, "frame_bytes_after_sum": m.codex_ws_frame_bytes_after_sum, "frame_attempted_tokens_sum": m.codex_ws_frame_attempted_tokens_sum, "frame_tokens_saved_sum": m.codex_ws_frame_tokens_saved_sum, }, "waste_signals": dict(m.waste_signals_total) if m.waste_signals_total else {}, # ContentRouter protection categories aggregated across the # session. Lets operators see, e.g., that 80% of messages # were `user_msg` (protected) and only 5% reached the # compressor — explains why compression rate is low and # whether `--compress-user-messages` would help (#454). "router": { "route_counts": dict(m.router_route_counts) if m.router_route_counts else {}, }, "savings_history": m.savings_history[-100:], # Last 100 data points "display_session": display_session, # Whether LiteLLM is importable. Pricing (the "$ Saved" tile) is # derived entirely from LiteLLM's cost tables, and LiteLLM is gated # off on Python >=3.14 in pyproject — so when this is False the # dashboard tells the user to reinstall on 3.13 instead of just # showing $0.00 forever. "litellm_available": LITELLM_AVAILABLE, "persistent_savings": persistent_savings, "prefix_cache": prefix_cache_stats, "cost": _merge_cost_stats( proxy.cost_tracker.stats() if proxy.cost_tracker else None, prefix_cache_stats, cli_tokens_avoided=cli_tokens_avoided, ), "compression": { "ccr_entries": compression_stats.get("entry_count", 0), "ccr_max_entries": compression_stats.get("max_entries", 0), "original_tokens_cached": compression_stats.get("total_original_tokens", 0), "compressed_tokens_cached": compression_stats.get("total_compressed_tokens", 0), "ccr_retrievals": compression_stats.get("total_retrievals", 0), }, "compression_cache": compression_cache_stats, # Always False: the anonymous telemetry beacon was removed, so no # telemetry is ever shipped externally (local collection only). "anon_telemetry_shipping": False, "telemetry": { "enabled": telemetry_stats.get("enabled", False), "total_compressions": telemetry_stats.get("total_compressions", 0), "total_retrievals": telemetry_stats.get("total_retrievals", 0), "global_retrieval_rate": round(telemetry_stats.get("global_retrieval_rate", 0), 4), "tool_signatures_tracked": telemetry_stats.get("tool_signatures_tracked", 0), "avg_compression_ratio": round(telemetry_stats.get("avg_compression_ratio", 0), 4), "avg_token_reduction": round(telemetry_stats.get("avg_token_reduction", 0), 4), }, "otel": get_otel_metrics_status(), "langfuse": get_langfuse_tracing_status(), "feedback_loop": { "tools_tracked": feedback_stats.get("tools_tracked", 0), "total_compressions": feedback_stats.get("total_compressions", 0), "total_retrievals": feedback_stats.get("total_retrievals", 0), "global_retrieval_rate": round(feedback_stats.get("global_retrieval_rate", 0), 4), "tools_with_high_retrieval": sum( 1 for p in feedback_stats.get("tool_patterns", {}).values() if p.get("retrieval_rate", 0) > 0.3 ), }, "toin": get_toin().get_stats(), "context_tool": { "configured": cli_filtering_tool, "label": cli_filtering_label, "available": cli_filtering_available, "stats": cli_filtering_stats, }, "cli_filtering": cli_filtering_stats, "proxy_inbound": proxy.metrics.inbound_snapshot(), "cache": await proxy.cache.stats() if proxy.cache else None, "rate_limiter": await proxy.rate_limiter.stats() if proxy.rate_limiter else None, **recent_request_payload, "log_full_messages": proxy.config.log_full_messages if proxy else False, **get_quota_registry().get_all_stats(), "throughput": throughput, } def _dashboard_config_payload() -> dict[str, Any]: profile_kwargs = proxy_pipeline_kwargs(config) target_ratio = profile_kwargs.get("target_ratio", config.target_ratio) target_savings_percent = None if isinstance(target_ratio, int | float): target_savings_percent = round(max(0.0, min(1.0, 1.0 - float(target_ratio))) * 100, 1) return { "savings_profile": config.savings_profile, "target_ratio": target_ratio, "target_savings_percent": target_savings_percent, "compress_user_messages": bool( profile_kwargs.get("compress_user_messages", config.compress_user_messages) ), "compress_system_messages": bool( profile_kwargs.get("compress_system_messages", config.compress_system_messages) ), "protect_recent": profile_kwargs.get("read_protection_window", config.protect_recent), "protect_analysis_context": config.protect_analysis_context, "min_tokens_to_crush": profile_kwargs.get( "min_tokens_to_compress", config.min_tokens_to_crush ), "max_items_after_crush": profile_kwargs.get( "max_items_after_crush", config.max_items_after_crush ), "smart_crusher_with_compaction": profile_kwargs.get( "smart_crusher_with_compaction", config.smart_crusher_with_compaction, ), "force_kompress": bool(profile_kwargs.get("force_kompress", False)), "accuracy_guard": config.accuracy_guard, } async def _get_cached_stats_payload() -> dict[str, Any]: """Return a short-TTL cached `/stats` snapshot for dashboard polling.""" now = time.monotonic() cached_payload = cast(dict[str, Any] | None, _stats_snapshot.get("value")) if cached_payload is not None and now < float(_stats_snapshot["expires_at"]): return cached_payload async with _stats_snapshot_lock: now = time.monotonic() cached_payload = cast(dict[str, Any] | None, _stats_snapshot.get("value")) if cached_payload is not None and now < float(_stats_snapshot["expires_at"]): return cached_payload payload = await _build_stats_payload() _stats_snapshot["value"] = payload _stats_snapshot["expires_at"] = time.monotonic() + DASHBOARD_STATS_CACHE_TTL_SECONDS return payload @app.get("/stats") async def stats(request: Request, cached: bool = False): """Get comprehensive proxy statistics. This is the main stats endpoint - it aggregates data from all subsystems: - Request metrics (total, cached, failed, by model/provider) - Token usage and savings - Cost tracking - Canonical persisted display_session metrics for downstream dashboards - Compression (CCR) statistics - Telemetry/TOIN (data flywheel) statistics - Cache and rate limiter stats Use ``?cached=1`` for the dashboard fast path. That returns a short-TTL snapshot to avoid rebuilding the full payload on every UI poll. ``recent_requests`` / ``request_logs`` (per-request ids, providers, models, errors) and ``config`` (backend + savings profile) are embedded only for loopback callers — the local dashboard. Network callers still get the aggregate counters but never the per-request metadata. """ include_sensitive = _request_is_loopback(request) if cached: payload = dict(await _get_cached_stats_payload()) if include_sensitive: # Refresh the per-request tail on top of the cached snapshot. payload.update(_build_recent_request_payload()) payload["config"] = _dashboard_config_payload() else: payload = await _build_stats_payload() if include_sensitive: payload["config"] = _dashboard_config_payload() if not include_sensitive: # _build_stats_payload bakes these in; strip for network callers. payload.pop("recent_requests", None) payload.pop("request_logs", None) return payload @app.post("/stats/reset", dependencies=[Depends(_require_loopback)]) async def stats_reset(): """Reset in-memory proxy stats for local test/debug isolation.""" await proxy.metrics.reset_runtime() if proxy.cost_tracker: proxy.cost_tracker.reset_runtime() await initialize_context_tool_session_baseline() async with _stats_snapshot_lock: _stats_snapshot["value"] = None _stats_snapshot["expires_at"] = 0.0 return JSONResponse(status_code=200, content={"status": "reset"}) @app.get("/stats-history") async def stats_history( format: Literal["json", "csv"] = "json", series: Literal["history", "hourly", "daily", "weekly", "monthly"] = "history", history_mode: Literal["compact", "full", "none"] = "compact", ): """Get durable proxy compression history plus display-session state. The JSON payload also carries a ``cli_filtering`` key with live RTK stats. This is a curated subset (``tool``, ``label``, ``available``, ``lifetime``, ``session``) tailored to the Historical tab, not the full ``_get_context_tool_stats()`` payload that ``/stats`` exposes. It is ``None`` only when the stats read hard-fails; when the tool is merely absent, ``cli_filtering`` stays populated with ``available: False`` and zeroed counters so the tab can distinguish "not installed" from "installed, no data yet." """ if format == "csv": filename = f"headroom-stats-history-{series}.csv" return Response( content=proxy.metrics.savings_tracker.export_csv(series=series), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) history = proxy.metrics.savings_tracker.history_response(history_mode=history_mode) # Augment with live RTK/cli-filtering lifetime stats so the Historical # tab can display them. These live in the context-tool's own stats file # and survive proxy restarts — exactly what the Historical tab needs. # Best-effort: if the RTK stats file can't be read (missing, parse error, # IO), fall back to None so the Historical tab stays available instead of # 500ing. The tab hides the card when cli_filtering is None. try: cli_stats = await asyncio.to_thread(_get_context_tool_stats) except Exception: logger.debug("stats-history: RTK stats unavailable", exc_info=True) cli_stats = None if cli_stats: history["cli_filtering"] = { "tool": str(cli_stats.get("tool", "rtk")), "label": str(cli_stats.get("label", "RTK")), "available": bool(cli_stats.get("installed", False)), "lifetime": cli_stats.get("lifetime", {}), "session": cli_stats.get("session", {}), } else: history["cli_filtering"] = None return history @app.get("/transformations/feed", dependencies=[Depends(_require_loopback)]) async def transformations_feed(limit: int = 20): """Get recent message transformations for the live feed. Loopback-only: when ``log_full_messages`` is enabled this returns the full request/response message bodies (prompt content and completions) via ``request_messages`` / ``compressed_messages`` / ``response_content``. With the default ``--host 0.0.0.0`` Docker bind, leaving it open would expose chat history to anyone able to reach the proxy port. The dashboard runs in the user's browser on loopback, so this gate does not break legitimate use. Returns empty list if log_full_messages is disabled (messages are not stored). """ if limit > 100: limit = 100 transformations = [] log_full_messages = proxy.config.log_full_messages if proxy else False if proxy and proxy.logger: logs = proxy.logger.get_recent_with_messages(limit) for log in logs: transformations.append( { "request_id": log.get("request_id"), "timestamp": log.get("timestamp"), "provider": log.get("provider"), "model": log.get("model"), "input_tokens_original": log.get("input_tokens_original"), "input_tokens_optimized": log.get("input_tokens_optimized"), "tokens_saved": log.get("tokens_saved"), "savings_percent": log.get("savings_percent"), "transforms_applied": log.get("transforms_applied", []), "request_messages": log.get("request_messages"), "compressed_messages": log.get("compressed_messages"), "response_content": log.get("response_content"), "turn_id": log.get("turn_id"), } ) return {"transformations": transformations, "log_full_messages": log_full_messages} @app.get("/subscription-window") async def subscription_window(): """Current Anthropic subscription window utilisation and Headroom contribution. Issue #281: the Anthropic OAuth usage API is polled every 5 minutes (aggressive polling risks 429s / OAuth-token flagging), so the cached ``utilization_pct`` lags reality by up to one poll interval. When the user's 5-hour window rolls over between two polls the dashboard would otherwise render the OLD window's percentage. We: 1. Optionally trigger a 60s-floored singleton poll on dashboard load (bounded across users, well within Anthropic tolerance). 2. Render via :meth:`SubscriptionTracker.render_state`, which synthesizes post-reset windows from local transcript-derived token counts when ``now >= window.resets_at``. """ tracker = get_subscription_tracker() if tracker is None: return JSONResponse( status_code=503, content={"error": "Subscription tracking is not enabled"}, ) await tracker.maybe_poll_on_demand() return JSONResponse(content=tracker.render_state()) @app.get("/quota") async def quota(): """Unified quota/rate-limit stats for all registered providers (Anthropic, Codex, Copilot).""" return JSONResponse(content=get_quota_registry().get_all_stats()) @app.get("/metrics") async def metrics(): """Prometheus metrics endpoint.""" return PlainTextResponse( await proxy.metrics.export(), media_type="text/plain; version=0.0.4", ) # Debug endpoints @app.get("/debug/memory", dependencies=[Depends(_require_loopback)]) async def debug_memory(): """Get detailed memory usage statistics. Returns memory usage for all tracked components including: - Process-level memory (RSS, VMS, percent) - Per-component memory usage and budgets - Cache hit/miss statistics - Total tracked vs target budget This endpoint is useful for debugging memory issues and monitoring memory budgets. """ from ..memory.tracker import MemoryTracker tracker = MemoryTracker.get() # Register components if not already registered _register_memory_components(proxy, tracker) report = tracker.get_report() return report.to_dict() @app.post("/cache/clear", dependencies=[Depends(_require_loopback)]) async def clear_cache(): """Clear the response cache. Loopback-only: this mutates server state. With the default ``--host 0.0.0.0`` Docker bind, an unauthenticated POST from any network-reachable client would otherwise let them forcibly evict the proxy's cached completions — a denial-of-service / cost-amplification lever (every cleared entry forces a fresh upstream call). """ if proxy.cache: await proxy.cache.clear() return {"status": "cleared"} return {"status": "cache disabled"} # CCR (Compress-Cache-Retrieve) endpoints @app.post("/v1/retrieve", dependencies=[Depends(_require_loopback)]) async def ccr_retrieve(request: Request): """Retrieve original content from CCR compression cache. This is the "Retrieve" part of CCR (Compress-Cache-Retrieve). When SmartCrusher compresses tool outputs, the original data is cached. LLMs can call this endpoint to get more data if needed. Request body: hash (str): Hash key from compression marker (required) Response: {"hash": "...", "original_content": "...", ...} """ data = await request.json() hash_key = data.get("hash") if not hash_key: raise HTTPException(status_code=400, detail="hash required") store = get_compression_store() entry_status = store.get_entry_status(hash_key, clean_expired=True) if entry_status["status"] != "available": raise HTTPException( status_code=404, detail=format_retrieval_miss_detail(entry_status), ) # Retrieval is by hash: always return the full original content. entry = store.retrieve(hash_key) if entry: return { "hash": hash_key, "original_content": entry.original_content, "original_tokens": entry.original_tokens, "original_item_count": entry.original_item_count, "compressed_item_count": entry.compressed_item_count, "tool_name": entry.tool_name, "retrieval_count": entry.retrieval_count, } raise HTTPException( status_code=404, detail=format_retrieval_miss_detail( store.get_entry_status(hash_key, clean_expired=True) ), ) @app.get("/v1/retrieve/stats", dependencies=[Depends(_require_loopback)]) async def ccr_stats(): """Get CCR compression store statistics.""" store = get_compression_store() stats = store.get_stats() events = store.get_retrieval_events(limit=20) return { "store": stats, "recent_retrievals": [ { "hash": e.hash, "query": e.query, "items_retrieved": e.items_retrieved, "total_items": e.total_items, "tool_name": e.tool_name, "retrieval_type": e.retrieval_type, } for e in events ], } @app.get("/v1/feedback") async def ccr_feedback(): """Get CCR feedback loop statistics and learned patterns. This endpoint exposes the feedback loop's learned patterns for monitoring and debugging. It shows: - Per-tool retrieval rates (high = compress less aggressively) - Common search queries per tool - Queried fields (suggest what to preserve) Use this to understand how well compression is working and whether the feedback loop is adjusting appropriately. """ feedback = get_compression_feedback() stats = feedback.get_stats() return { "feedback": stats, "hints_example": { tool_name: { "hints": { "max_items": hints.max_items if (hints := feedback.get_compression_hints(tool_name)) else 15, "suggested_items": hints.suggested_items if hints else None, "skip_compression": hints.skip_compression if hints else False, "preserve_fields": hints.preserve_fields if hints else [], "reason": hints.reason if hints else "", } } for tool_name in list(stats.get("tool_patterns", {}).keys())[:5] }, } @app.get("/v1/feedback/{tool_name}") async def ccr_feedback_for_tool(tool_name: str): """Get compression hints for a specific tool. Returns feedback-based hints that would be used for compressing this tool's output. """ feedback = get_compression_feedback() hints = feedback.get_compression_hints(tool_name) patterns = feedback.get_all_patterns().get(tool_name) return { "tool_name": tool_name, "hints": { "max_items": hints.max_items, "min_items": hints.min_items, "suggested_items": hints.suggested_items, "aggressiveness": hints.aggressiveness, "skip_compression": hints.skip_compression, "preserve_fields": hints.preserve_fields, "reason": hints.reason, }, "pattern": { "total_compressions": patterns.total_compressions if patterns else 0, "total_retrievals": patterns.total_retrievals if patterns else 0, "retrieval_rate": patterns.retrieval_rate if patterns else 0.0, "full_retrieval_rate": patterns.full_retrieval_rate if patterns else 0.0, "search_rate": patterns.search_rate if patterns else 0.0, "common_queries": list(patterns.common_queries.keys())[:10] if patterns else [], "queried_fields": list(patterns.queried_fields.keys())[:10] if patterns else [], } if patterns else None, } # Telemetry endpoints (Data Flywheel) @app.get("/v1/telemetry") async def telemetry_stats(): """Get telemetry statistics for the data flywheel. This endpoint exposes privacy-preserving telemetry data that powers the data flywheel - learning optimal compression strategies across tool types based on usage patterns. What's collected (anonymized): - Tool output structure patterns (field types, not values) - Compression decisions and ratios - Retrieval patterns (rate, type, not content) - Strategy effectiveness What's NOT collected: - Actual data values - User identifiers - Queries or search terms - File paths or tool names (hashed by default) """ telemetry = get_telemetry_collector() return telemetry.get_stats() @app.get("/v1/telemetry/export") async def telemetry_export(): """Export full telemetry data for aggregation. This endpoint exports all telemetry data in a format suitable for cross-user aggregation. The data is privacy-preserving - no actual values are included, only structural patterns and statistics. Use this for: - Building a central learning service - Sharing learned patterns across instances - Analysis and debugging """ telemetry = get_telemetry_collector() return telemetry.export_stats() @app.post("/v1/telemetry/import") async def telemetry_import(request: Request): """Import telemetry data from another source. This allows merging telemetry from multiple sources for cross-user learning. The imported data is merged with existing statistics. Request body: Telemetry export data from /v1/telemetry/export """ telemetry = get_telemetry_collector() data = await request.json() telemetry.import_stats(data) return {"status": "imported", "current_stats": telemetry.get_stats()} @app.get("/v1/telemetry/tools") async def telemetry_tools(): """Get telemetry statistics for all tracked tool signatures. Returns statistics per tool signature (anonymized), including: - Compression ratios and strategy usage - Retrieval rates (high = compression too aggressive) - Learned recommendations """ telemetry = get_telemetry_collector() all_stats = telemetry.get_all_tool_stats() return { "tool_count": len(all_stats), "tools": {sig_hash: stats.to_dict() for sig_hash, stats in all_stats.items()}, } @app.get("/v1/telemetry/tools/{signature_hash}") async def telemetry_tool_detail(signature_hash: str): """Get detailed telemetry for a specific tool signature. Includes learned recommendations if enough data has been collected. """ telemetry = get_telemetry_collector() stats = telemetry.get_tool_stats(signature_hash) recommendations = telemetry.get_recommendations(signature_hash) if stats is None: raise HTTPException( status_code=404, detail=f"No telemetry found for signature: {signature_hash}" ) return { "signature_hash": signature_hash, "stats": stats.to_dict(), "recommendations": recommendations, } # TOIN (Tool Output Intelligence Network) endpoints @app.get("/v1/toin/stats") async def toin_stats(): """Get overall TOIN statistics. Returns aggregated statistics from the Tool Output Intelligence Network, which learns optimal compression strategies across all tool types. Response includes: - enabled: Whether TOIN is enabled - patterns_tracked: Number of unique tool patterns being tracked - total_compressions: Total compression events recorded - total_retrievals: Total retrieval events recorded - global_retrieval_rate: Overall retrieval rate (high = compression too aggressive) - patterns_with_recommendations: Patterns with enough data for recommendations """ toin = get_toin() return toin.get_stats() @app.get("/v1/toin/patterns") async def toin_patterns(limit: int = 20): """List TOIN patterns with most samples. Returns patterns sorted by sample_size descending. Use this to see which tool types have the most data and their learned behaviors. Query params: limit: Maximum number of patterns to return (default 20) Response includes for each pattern: - hash: Truncated tool signature hash (12 chars) - compressions: Total compression events - retrievals: Total retrieval events - retrieval_rate: Percentage of compressions that triggered retrieval - confidence: Confidence level in recommendations (0.0-1.0) - skip_recommended: Whether TOIN recommends skipping compression - optimal_max_items: Learned optimal max_items setting """ toin = get_toin() exported = toin.export_patterns() patterns_data = exported.get("patterns", {}) # Convert to list and sort by sample_size patterns_list = [] for sig_hash, pattern_dict in patterns_data.items(): sample_size = pattern_dict.get("sample_size", 0) total_compressions = pattern_dict.get("total_compressions", 0) total_retrievals = pattern_dict.get("total_retrievals", 0) retrieval_rate = ( total_retrievals / total_compressions if total_compressions > 0 else 0.0 ) patterns_list.append( { "hash": sig_hash[:12], "compressions": total_compressions, "retrievals": total_retrievals, "retrieval_rate": f"{retrieval_rate:.1%}", "confidence": round(pattern_dict.get("confidence", 0.0), 3), "skip_recommended": pattern_dict.get("skip_compression_recommended", False), "optimal_max_items": pattern_dict.get("optimal_max_items", 20), "sample_size": sample_size, } ) # Sort by sample_size descending patterns_list.sort(key=lambda p: p["sample_size"], reverse=True) # Remove sample_size from output (used only for sorting) for p in patterns_list: del p["sample_size"] return patterns_list[:limit] @app.get("/v1/toin/pattern/{hash_prefix}") async def toin_pattern_detail(hash_prefix: str): """Get detailed TOIN pattern info by hash prefix. Searches for a pattern where the tool signature hash starts with the provided prefix. Returns full pattern details if found. Path params: hash_prefix: Beginning of the tool signature hash (min 4 chars recommended) Response: Full pattern.to_dict() with all learned statistics and recommendations. """ toin = get_toin() exported = toin.export_patterns() patterns_data = exported.get("patterns", {}) # Search for pattern with matching hash prefix for sig_hash, pattern_dict in patterns_data.items(): if sig_hash.startswith(hash_prefix): return pattern_dict raise HTTPException( status_code=404, detail=f"No TOIN pattern found with hash starting with: {hash_prefix}" ) @app.get("/v1/retrieve/{hash_key}", dependencies=[Depends(_require_loopback)]) async def ccr_retrieve_get(hash_key: str): """GET version of CCR retrieve for easier testing.""" store = get_compression_store() entry_status = store.get_entry_status(hash_key, clean_expired=True) if entry_status["status"] != "available": raise HTTPException( status_code=404, detail=format_retrieval_miss_detail(entry_status), ) # Retrieval is by hash: always return the full original content. entry = store.retrieve(hash_key) if entry: return { "hash": hash_key, "original_content": entry.original_content, "original_tokens": entry.original_tokens, "original_item_count": entry.original_item_count, "compressed_item_count": entry.compressed_item_count, "tool_name": entry.tool_name, "retrieval_count": entry.retrieval_count, } raise HTTPException( status_code=404, detail=format_retrieval_miss_detail( store.get_entry_status(hash_key, clean_expired=True) ), ) # CCR Tool Call Handler - for agent frameworks to call when LLM uses headroom_retrieve @app.post("/v1/retrieve/tool_call", dependencies=[Depends(_require_loopback)]) async def ccr_handle_tool_call(request: Request): """Handle a CCR tool call from an LLM response. This endpoint accepts tool call formats from various providers and returns a properly formatted tool result. Agent frameworks can use this to handle CCR tool calls without implementing the retrieval logic themselves. Request body (Anthropic format): { "tool_call": { "id": "toolu_123", "name": "headroom_retrieve", "input": {"hash": "abc123"} }, "provider": "anthropic" } Request body (OpenAI format): { "tool_call": { "id": "call_123", "function": { "name": "headroom_retrieve", "arguments": "{\"hash\": \"abc123\"}" } }, "provider": "openai" } Response: { "tool_result": {...}, # Formatted for the provider "success": true, "data": {...} # Raw retrieval data } """ data = await request.json() tool_call = data.get("tool_call", {}) provider = data.get("provider", "anthropic") # Parse the tool call hash_key = parse_tool_call(tool_call, provider) if hash_key is None: raise HTTPException( status_code=400, detail=f"Invalid tool call or not a {CCR_TOOL_NAME} call" ) # Perform retrieval store = get_compression_store() entry_status = store.get_entry_status(hash_key, clean_expired=True) if entry_status["status"] != "available": retrieval_data = { "error": format_retrieval_miss_detail(entry_status), "hash": hash_key, "status": entry_status["status"], "ttl_seconds": entry_status.get("ttl_seconds", entry_status["default_ttl_seconds"]), } else: # Retrieval is by hash: always return the full original content. entry = store.retrieve(hash_key) if entry: retrieval_data = { "hash": hash_key, "original_content": entry.original_content, "original_item_count": entry.original_item_count, "compressed_item_count": entry.compressed_item_count, } else: miss_status = store.get_entry_status(hash_key, clean_expired=True) retrieval_data = { "error": format_retrieval_miss_detail(miss_status), "hash": hash_key, "status": miss_status["status"], "ttl_seconds": miss_status.get( "ttl_seconds", miss_status["default_ttl_seconds"] ), } # Format tool result for provider tool_call_id = tool_call.get("id", "") result_content = json.dumps(retrieval_data, indent=2) if provider == "anthropic": tool_result = { "type": "tool_result", "tool_use_id": tool_call_id, "content": result_content, } elif provider == "openai": tool_result = { "role": "tool", "tool_call_id": tool_call_id, "content": result_content, } else: tool_result = { "tool_call_id": tool_call_id, "content": result_content, } return { "tool_result": tool_result, "success": "error" not in retrieval_data, "data": retrieval_data, } # Compression-only endpoint (for TypeScript SDK and other HTTP clients) @app.post("/v1/compress", dependencies=[Depends(_require_loopback)]) async def compress_messages(request: Request): return await proxy.handle_compress(request) register_provider_routes(app, proxy) return app def _json_ready(value: Any) -> Any: if is_dataclass(value) and not isinstance(value, type): return {field.name: _json_ready(getattr(value, field.name)) for field in fields(value)} if isinstance(value, dict): return {str(key): _json_ready(item) for key, item in value.items()} if isinstance(value, list | tuple | set): return [_json_ready(item) for item in value] return value def _proxy_config_payload(config: ProxyConfig) -> dict[str, Any]: payload: dict[str, Any] = {} for field in fields(config): value = _json_ready(getattr(config, field.name)) try: json.dumps(value) except TypeError: continue payload[field.name] = value return payload def _proxy_config_from_env() -> ProxyConfig: raw_config = os.environ.get(_MULTI_WORKER_CONFIG_ENV) if raw_config: try: return ProxyConfig(**json.loads(raw_config)) except (TypeError, ValueError, json.JSONDecodeError): logger.warning( "Invalid %s; falling back to HEADROOM_* env vars", _MULTI_WORKER_CONFIG_ENV ) return ProxyConfig( host=_get_env_str("HEADROOM_HOST", "127.0.0.1"), port=_get_env_int("HEADROOM_PORT", 8787), openai_api_url=os.environ.get("OPENAI_TARGET_API_URL"), anthropic_api_url=os.environ.get("ANTHROPIC_TARGET_API_URL"), anthropic_buffered_request_timeout_seconds=_get_env_int( "HEADROOM_ANTHROPIC_BUFFERED_REQUEST_TIMEOUT_SECONDS", 600, min_value=1, ), vertex_api_url=os.environ.get("VERTEX_TARGET_API_URL"), backend=_get_env_str("HEADROOM_BACKEND", "anthropic"), bedrock_region=_get_env_str("HEADROOM_BEDROCK_REGION", "us-west-2"), bedrock_profile=os.environ.get("AWS_PROFILE"), bedrock_api_url=os.environ.get("BEDROCK_TARGET_API_URL"), anyllm_provider=_get_env_str("HEADROOM_ANYLLM_PROVIDER", "openai"), disable_kompress=_get_env_bool("HEADROOM_DISABLE_KOMPRESS", False), disable_kompress_fallback=_get_env_bool("HEADROOM_DISABLE_KOMPRESS_FALLBACK", False), disable_kompress_anthropic=_get_env_optional_bool("HEADROOM_DISABLE_KOMPRESS_ANTHROPIC"), disable_kompress_openai=_get_env_optional_bool("HEADROOM_DISABLE_KOMPRESS_OPENAI"), force_kompress_all=_get_env_bool("HEADROOM_FORCE_KOMPRESS_ALL", False), lossless=_get_env_bool("HEADROOM_LOSSLESS", False), max_connections=_get_env_int("HEADROOM_MAX_CONNECTIONS", 500), max_keepalive_connections=_get_env_int("HEADROOM_MAX_KEEPALIVE", 100), keepalive_expiry=_get_env_float("HEADROOM_KEEPALIVE_EXPIRY", 90.0), http2=_get_env_bool("HEADROOM_HTTP2", True), http_proxy=os.environ.get("HEADROOM_HTTP_PROXY") or None, periodic_toin_stats_enabled=_get_env_bool("HEADROOM_PERIODIC_TOIN_STATS", True), proxy_token=os.environ.get("HEADROOM_PROXY_TOKEN") or None, offline=_get_env_bool("HEADROOM_OFFLINE", False), # Default mode is CACHE (Headroom's coding posture): delta-only compression # at ~0 prefix-cache busts. HEADROOM_MODE overrides. mode=normalize_proxy_mode(_get_env_str("HEADROOM_MODE", PROXY_MODE_CACHE)), # Default savings profile is "coding" so proxy_pipeline_kwargs applies its # posture (compress_user, protect_recent, min_tokens). HEADROOM_SAVINGS_PROFILE # overrides. savings_profile=os.environ.get("HEADROOM_SAVINGS_PROFILE") or "coding", read_maturation=_get_env_bool("HEADROOM_READ_MATURATION", False), read_maturation_quiesce_turns=_get_env_int("HEADROOM_READ_MATURATION_QUIESCE_TURNS", 5), read_maturation_max_hold_turns=_get_env_int("HEADROOM_READ_MATURATION_MAX_HOLD_TURNS", 25), read_maturation_min_size_bytes=_get_env_int( "HEADROOM_READ_MATURATION_MIN_SIZE_BYTES", 2048 ), ) def create_app_from_env() -> FastAPI: # Seed the coding-profile defaults into the process env BEFORE reading config, # so the uvicorn factory launch gets Headroom's out-of-box posture (cache mode, # tool-search, dedupe, read protection, …). setdefault → explicit env wins. from headroom.agent_savings import seed_proxy_env_defaults seed_proxy_env_defaults() return create_app(_proxy_config_from_env()) def _get_code_aware_banner_status(config: ProxyConfig) -> str: """Get code-aware compression status line for banner.""" if config.code_aware_enabled: if is_tree_sitter_available(): return "ENABLED (AST-based)" else: return "NOT INSTALLED (pip install headroom-ai[code])" else: if is_tree_sitter_available(): return "DISABLED (--code-aware or HEADROOM_CODE_AWARE_ENABLED=1 to enable)" return "DISABLED (install headroom-ai[code] to enable)" def run_server( config: ProxyConfig | None = None, workers: int = 1, limit_concurrency: int = 1000, print_banner: bool = True, ): """Run the proxy server. Args: config: Proxy configuration workers: Number of worker processes (use N for multi-core scaling) limit_concurrency: Max concurrent connections before 503 response print_banner: When False, skip the legacy ASCII banner. The Click CLI (`headroom proxy`) prints its own startup banner before calling this — printing a second banner here is the "dual banner" UX issue. Direct `python -m headroom.proxy.server` still gets the banner since it has no other startup output. """ if not FASTAPI_AVAILABLE: print("ERROR: FastAPI required. Install: pip install fastapi uvicorn httpx") sys.exit(1) # Seed the request-time coding-profile toggles (tool-search, dedupe, read # protection, lossless→lossy, effort-router, block-char floor) into the # process env before serving, so downstream per-request readers pick them up. # Done here (not in the CLI command) so unit tests that mock run_server never # mutate os.environ. setdefault → explicit env still wins. MODE / profile are # already resolved into `config` above via their inline defaults. from headroom.agent_savings import seed_proxy_env_defaults seed_proxy_env_defaults() config = config or ProxyConfig() code_aware_status = _get_code_aware_banner_status(config) # Format connection pool info pool_info = f"max={config.max_connections}, keepalive={config.max_keepalive_connections}" http2_status = "ENABLED" if (config.http2 and not config.http_proxy) else "DISABLED" backend_status = format_backend_status( backend=config.backend, anyllm_provider=config.anyllm_provider, bedrock_region=config.bedrock_region, ) # Resolve upstream API targets for display in the banner (#583). api_targets = resolve_api_targets(config.provider_api_overrides) if print_banner: print(f""" ╔══════════════════════════════════════════════════════════════════════╗ ║ HEADROOM PROXY SERVER ║ ╠══════════════════════════════════════════════════════════════════════╣ ║ Version: 1.0.0 ║ ║ Listening: http://{config.host}:{config.port:<5} ║ ║ Workers: {workers:<3} Concurrency Limit: {limit_concurrency:<5} ║ ║ Backend: {backend_status:<59}║ ╠══════════════════════════════════════════════════════════════════════╣ ║ UPSTREAM TARGETS: ║ ║ Anthropic: {api_targets.anthropic:<57}║ ║ OpenAI: {api_targets.openai:<57}║ ║ Gemini: {api_targets.gemini:<57}║ ║ Cloud Code: {api_targets.cloudcode:<57}║ ║ Vertex AI: {api_targets.vertex:<57}║ ╠══════════════════════════════════════════════════════════════════════╣ ║ FEATURES: ║ ║ Optimization: {"ENABLED " if config.optimize else "DISABLED"} ║ ║ Caching: {"ENABLED " if config.cache_enabled else "DISABLED"} (TTL: {config.cache_ttl_seconds}s) ║ ║ Rate Limiting: {"ENABLED " if config.rate_limit_enabled else "DISABLED"} ({config.rate_limit_requests_per_minute} req/min, {config.rate_limit_tokens_per_minute:,} tok/min) ║ ║ Retry: {"ENABLED " if config.retry_enabled else "DISABLED"} (max {config.retry_max_attempts} attempts) ║ ║ Cost Tracking: {"ENABLED " if config.cost_tracking_enabled else "DISABLED"} (budget: {"$" + str(config.budget_limit_usd) + "/" + config.budget_period if config.budget_limit_usd else "unlimited"}) ║ ║ Code-Aware: {code_aware_status:<52}║ ║ HTTP/2: {http2_status:<52}║ ║ Conn Pool: {pool_info:<52}║ ╠══════════════════════════════════════════════════════════════════════╣ ║ USAGE: ║ ║ Claude Code: ANTHROPIC_BASE_URL=http://{config.host}:{config.port} claude ║ ║ Cursor: Set base URL in settings ║ ╠══════════════════════════════════════════════════════════════════════╣ ║ ENDPOINTS: ║ ║ /livez Process liveness ║ ║ /readyz Traffic readiness ║ ║ /health Aggregate health ║ ║ /stats Detailed statistics ║ ║ /metrics Prometheus metrics ║ ║ /cache/clear Clear response cache ║ ║ /v1/retrieve CCR: Retrieve compressed content ║ ║ /v1/retrieve/stats CCR: Compression store stats ║ ║ /v1/retrieve/tool_call CCR: Handle LLM tool calls ║ ║ /v1/feedback CCR: Feedback loop stats & patterns ║ ║ /v1/feedback/{{tool}} CCR: Compression hints for a tool ║ ║ /v1/telemetry Data flywheel: Telemetry stats ║ ║ /v1/telemetry/export Data flywheel: Export for aggregation ║ ║ /v1/telemetry/tools Data flywheel: Per-tool stats ║ ║ /v1/toin/stats TOIN: Overall intelligence stats ║ ║ /v1/toin/patterns TOIN: List learned patterns ║ ║ /v1/toin/pattern/{{hash}} TOIN: Pattern details by hash ║ ╚══════════════════════════════════════════════════════════════════════╝ """) app_target: Any uvicorn_kwargs: dict[str, Any] = {} if sys.platform == "win32": # ProactorEventLoop can close the listening socket on transient # AcceptEx failures (for example WinError 64 from keep-alive RSTs). # The selector loop keeps accept errors scoped to the connection. uvicorn_kwargs["loop"] = "asyncio:SelectorEventLoop" if workers > 1: # CompressionCache and PrefixTracker are always per-worker instance vars. # Python CompressionStore defaults to InMemoryBackend (per-process), so # CCR markers written on worker A are invisible to worker B unless a # cross-worker backend is configured via HEADROOM_CCR_BACKEND. # See RUST_DEV.md -> "Multi-worker deployment -- CCR fragmentation". if os.environ.get("HEADROOM_CCR_BACKEND", "").strip(): logger.warning( "Headroom is running with workers=%d. Compression cache, " "prefix tracker, TOIN state, and CostTracker are all per-process; " "multi-worker deployments produce avoidable cache busts and an " "unstable dashboard 'Proxy $ Saved' hero tile (each /stats poll " "hits a different worker's partial total) when sessions land on " "different workers. Run --workers 1 or place a sticky-session load " "balancer in front of multiple --workers 1 processes. " "See RUST_DEV.md -> 'Multi-worker deployment -- CCR fragmentation'.", workers, ) else: logger.warning( "Headroom is running with workers=%d. The in-memory CCR store, " "compression cache (incl. off-path background compression), prefix " "tracker, TOIN state, and CostTracker are all " "per-process; multi-worker deployments produce silent CCR retrieval " "failures, avoidable cache busts, and an unstable dashboard 'Proxy $ Saved' " "hero tile (each /stats poll hits a different worker's partial total) when " "sessions land on different workers. Set HEADROOM_CCR_BACKEND=sqlite for a " "persistent cross-worker CCR store, run --workers 1, or place a " "sticky-session load balancer in front of multiple --workers 1 processes. " "See RUST_DEV.md -> 'Multi-worker deployment -- CCR fragmentation'.", workers, ) os.environ[_MULTI_WORKER_CONFIG_ENV] = json.dumps(_proxy_config_payload(config)) app_target = "headroom.proxy.server:create_app_from_env" uvicorn_kwargs["factory"] = True else: app_target = create_app(config) uvicorn.run( app_target, host=config.host, port=config.port, log_level="warning", workers=workers if workers > 1 else None, # None = single process (default) limit_concurrency=limit_concurrency, # Defense-in-depth: the loopback guard for /debug/* endpoints trusts # request.client.host. uvicorn's ProxyHeadersMiddleware rewrites that # from X-Forwarded-For when FORWARDED_ALLOW_IPS is broader than the # default. Disabling proxy_headers here guarantees the guard sees the # real peer address regardless of env. proxy_headers=False, **uvicorn_kwargs, ) def _get_env_bool(name: str, default: bool) -> bool: """Get boolean from environment variable.""" val = os.environ.get(name) if val is None: return default return val.lower() in ("true", "1", "yes", "on") def _get_env_optional_bool(name: str) -> bool | None: """Tristate boolean env var: unset/empty -> None, truthy -> True, falsy -> False.""" val = os.environ.get(name) if val is None or val == "": return None return val.lower() in ("true", "1", "yes", "on") def _get_env_int(name: str, default: int, *, min_value: int | None = None) -> int: """Get integer from environment variable.""" val = os.environ.get(name) if val is None: return default try: parsed = int(val) except ValueError: return default if min_value is not None and parsed < min_value: return default return parsed def _positive_int_arg(value: str) -> int: parsed = int(value) if parsed < 1: raise argparse.ArgumentTypeError("must be >= 1") return parsed def _get_env_float(name: str, default: float) -> float: """Get float from environment variable.""" val = os.environ.get(name) if val is None: return default try: return float(val) except ValueError: return default def _get_env_str(name: str, default: str) -> str: """Get string from environment variable.""" return os.environ.get(name, default) def _parse_exclude_tools(cli_excludes: str | None) -> set[str]: """Parse extra never-compress tool names from CLI args and env var. Both --exclude-tools and HEADROOM_EXCLUDE_TOOLS are comma-separated (e.g. "WebSearch,WebFetch"). Each name is added in both original and lowercase form for case-insensitive matching, mirroring DEFAULT_EXCLUDE_TOOLS. Unset/empty -> empty set (DEFAULT_EXCLUDE_TOOLS used unchanged). Entries may contain glob patterns (e.g. "mcp__*"); see config.is_tool_excluded for the matching semantics. """ raw = ",".join(s for s in (cli_excludes, os.environ.get("HEADROOM_EXCLUDE_TOOLS")) if s) names: set[str] = set() for entry in raw.split(","): name = entry.strip() if name: names.add(name) names.add(name.lower()) return names def _parse_csv_tools(raw: str | None) -> set[str]: """Parse a bare CSV tool-name string without merging HEADROOM_EXCLUDE_TOOLS.""" names: set[str] = set() if not raw: return names for entry in raw.split(","): name = entry.strip() if name: names.add(name) names.add(name.lower()) return names def _parse_tool_profiles(cli_profiles: list[str]) -> dict[str, Any]: """Parse tool profiles from CLI args and HEADROOM_TOOL_PROFILES env var. Format: ToolName:level (e.g., Grep:conservative, Bash:moderate) Env var format: comma-separated (e.g., "Grep:conservative,Bash:moderate") Returns: Dict mapping tool names to CompressionProfile instances. """ from headroom.config import PROFILE_PRESETS, CompressionProfile profiles: dict[str, CompressionProfile] = {} raw_entries: list[str] = list(cli_profiles) # Also check env var env_val = os.environ.get("HEADROOM_TOOL_PROFILES", "") if env_val: raw_entries.extend(e.strip() for e in env_val.split(",") if e.strip()) for entry in raw_entries: if ":" not in entry: logger.warning("Invalid tool profile format (expected ToolName:level): %s", entry) continue tool_name, level = entry.split(":", 1) tool_name = tool_name.strip() level = level.strip().lower() if level in PROFILE_PRESETS: profiles[tool_name] = PROFILE_PRESETS[level] else: logger.warning( "Unknown profile level '%s' for tool '%s'. Use: conservative, moderate, aggressive", level, tool_name, ) return profiles if __name__ == "__main__": parser = argparse.ArgumentParser(description="Headroom Proxy Server") # Server parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8787) parser.add_argument( "--openai-api-url", help=f"Custom OpenAI API URL (default: {DEFAULT_OPENAI_API_URL})" ) parser.add_argument( "--anthropic-api-url", help=f"Custom Anthropic API URL (default: {DEFAULT_ANTHROPIC_API_URL})", ) parser.add_argument( "--anthropic-buffered-request-timeout-seconds", type=_positive_int_arg, default=600, help=( "Anthropic buffered read timeout in seconds for non-streaming " "message and batch paths (default: 600)" ), ) parser.add_argument( "--vertex-api-url", help=f"Custom Vertex AI regional API URL (default: {DEFAULT_VERTEX_API_URL})", ) # Backend (anthropic direct, bedrock, openrouter, anyllm, or litellm-) parser.add_argument( "--backend", default="anthropic", help=( "Backend: 'anthropic' (direct), 'bedrock' (AWS), 'openrouter', " "'anyllm' (any-llm), or 'litellm-' (e.g., litellm-hosted_vllm, litellm-vertex)" ), ) parser.add_argument( "--bedrock-region", default="us-west-2", help="AWS region for Bedrock backend (default: us-west-2)", ) parser.add_argument( "--bedrock-profile", help="AWS profile for Bedrock backend (default: use default credentials)", ) parser.add_argument( "--bedrock-api-url", help=( "Custom Bedrock InvokeModel upstream for the /model/{id}/invoke " "passthrough routes — point at a re-signing gateway, not raw AWS " "(env: BEDROCK_TARGET_API_URL)" ), ) parser.add_argument( "--openrouter-api-key", help="OpenRouter API key (or set OPENROUTER_API_KEY env var)", ) parser.add_argument( "--anyllm-provider", default="openai", help="any-llm provider: openai, anthropic, mistral, groq, ollama, bedrock, etc. (default: openai)", ) # Connection pool (scalability) parser.add_argument( "--max-connections", type=int, default=500, help="Max connections to upstream APIs (default: 500)", ) parser.add_argument( "--max-keepalive", type=int, default=100, help="Max keepalive connections (default: 100)" ) parser.add_argument( "--keepalive-expiry", type=float, default=90.0, help="Seconds an idle upstream keep-alive connection is kept open (default: 90)", ) parser.add_argument( "--no-http2", action="store_true", help="Disable HTTP/2 (enabled by default for better throughput)", ) parser.add_argument( "--http-proxy", help=("HTTP proxy URL for upstream provider requests only (env: HEADROOM_HTTP_PROXY)"), ) parser.add_argument( "--workers", type=int, default=1, help="Number of worker processes (default: 1, use N for multi-core)", ) parser.add_argument( "--limit-concurrency", type=int, default=1000, help="Max concurrent connections before 503 (default: 1000)", ) # Optimization parser.add_argument("--no-optimize", action="store_true", help="Disable optimization") parser.add_argument("--min-tokens", type=int, default=500, help="Min tokens to crush") parser.add_argument("--max-items", type=int, default=50, help="Max items after crush") parser.add_argument( "--tool-profile", action="append", default=[], help="Per-tool compression profile: ToolName:level (e.g., Grep:conservative, Bash:moderate, WebFetch:aggressive). " "Can be specified multiple times. Also settable via HEADROOM_TOOL_PROFILES env var.", ) parser.add_argument( "--compress-user-messages", action="store_true", help=( "Opt in to compressing `user` role messages. Default is off because " "user content is typically the subject of the request and is part of " "the prefix-cache zone. Enable this for OpenAI/Azure chat workloads " "where the bulk of input lives in user messages (pasted content, " "RAG context, etc.) and you want the router to consider it eligible. " "Also settable via HEADROOM_COMPRESS_USER_MESSAGES=1." ), ) parser.add_argument( "--disable-kompress", action="store_true", help=( "Disable Kompress ML compression while keeping structural compression enabled. " "Also settable via HEADROOM_DISABLE_KOMPRESS=1." ), ) parser.add_argument( "--disable-kompress-fallback", action="store_true", help=( "With --disable-kompress, route fall-through content to PASSTHROUGH instead of " "the default KOMPRESS fallback (restores legacy --disable-kompress behaviour). " "Also settable via HEADROOM_DISABLE_KOMPRESS_FALLBACK=1." ), ) parser.add_argument( "--disable-kompress-anthropic", dest="disable_kompress_anthropic", action="store_const", const=True, default=None, help=( "Disable Kompress for the Anthropic pipeline only, overriding --disable-kompress. " "Also settable via HEADROOM_DISABLE_KOMPRESS_ANTHROPIC=1." ), ) parser.add_argument( "--enable-kompress-anthropic", dest="disable_kompress_anthropic", action="store_const", const=False, help="Force-enable Kompress for the Anthropic pipeline, overriding --disable-kompress.", ) parser.add_argument( "--disable-kompress-openai", dest="disable_kompress_openai", action="store_const", const=True, default=None, help=( "Disable Kompress for the OpenAI/Codex pipeline only, overriding --disable-kompress. " "Also settable via HEADROOM_DISABLE_KOMPRESS_OPENAI=1." ), ) parser.add_argument( "--enable-kompress-openai", dest="disable_kompress_openai", action="store_const", const=False, help="Force-enable Kompress for the OpenAI/Codex pipeline, overriding --disable-kompress.", ) parser.add_argument( "--force-kompress-all", action="store_true", help=( "Route ALL compressible content through Kompress (kompress-v2-base), " "bypassing per-type compressor selection. Tool ground truth " "(Read/Glob/... and reversibility-gated output) is still never touched. " "Also settable via HEADROOM_FORCE_KOMPRESS_ALL=1." ), ) parser.add_argument( "--lossless", action="store_true", help=( "No-CCR lossless mode: compress LOG/SEARCH/DIFF tool outputs with " "format-native lossless compaction (and marker-free SmartCrusher) " "without emitting any CCR retrieval marker, so no MCP retrieve tool " "is needed. Also settable via HEADROOM_LOSSLESS=1." ), ) parser.add_argument( "--exclude-tools", default=None, help="Comma-separated tool names whose output is never compressed, " "merged with the built-in defaults (e.g., WebSearch,WebFetch). " "Entries may use glob patterns, e.g. 'mcp__*' to exclude every MCP tool. " "Also settable via HEADROOM_EXCLUDE_TOOLS env var.", ) parser.add_argument( "--protect-tool-results", default=None, help="Comma-separated tool names whose results are never lossy-compressed, " "merged with the built-in defaults (e.g. Bash,WebFetch). " "Also settable via HEADROOM_PROTECT_TOOL_RESULTS env var.", ) # Caching parser.add_argument("--no-cache", action="store_true", help="Disable caching") parser.add_argument("--cache-ttl", type=int, default=3600, help="Cache TTL seconds") # Rate limiting parser.add_argument("--no-rate-limit", action="store_true", help="Disable rate limiting") parser.add_argument("--rpm", type=int, default=60, help="Requests per minute") parser.add_argument("--tpm", type=int, default=100000, help="Tokens per minute") # Cost parser.add_argument("--budget", type=float, help="Budget limit in USD") parser.add_argument("--budget-period", choices=["hourly", "daily", "monthly"], default="daily") # Logging parser.add_argument("--log-file", help="Log file path") parser.add_argument("--log-messages", action="store_true", help="Log full messages") # Code-aware compression parser.add_argument( "--code-aware", action="store_true", help="Enable AST-based code compression (requires: pip install headroom-ai[code])", ) parser.add_argument( "--no-code-aware", action="store_true", help="Disable code-aware compression", ) args = parser.parse_args() # Environment variable defaults (HEADROOM_* prefix) # CLI args override env vars, env vars override ProxyConfig defaults env_code_aware = _get_env_bool("HEADROOM_CODE_AWARE_ENABLED", True) env_optimize = _get_env_bool("HEADROOM_OPTIMIZE", True) env_cache = _get_env_bool("HEADROOM_CACHE_ENABLED", True) env_rate_limit = _get_env_bool("HEADROOM_RATE_LIMIT_ENABLED", True) # Determine settings: CLI flags override env vars # --no-X explicitly disables, --X explicitly enables, neither uses env var code_aware_enabled = ( env_code_aware if not (args.code_aware or args.no_code_aware) else (args.code_aware or not args.no_code_aware) ) optimize = env_optimize if not args.no_optimize else False cache_enabled = env_cache if not args.no_cache else False rate_limit_enabled = env_rate_limit if not args.no_rate_limit else False disable_kompress = args.disable_kompress or _get_env_bool("HEADROOM_DISABLE_KOMPRESS", False) disable_kompress_fallback = args.disable_kompress_fallback or _get_env_bool( "HEADROOM_DISABLE_KOMPRESS_FALLBACK", False ) disable_kompress_anthropic = ( args.disable_kompress_anthropic if args.disable_kompress_anthropic is not None else _get_env_optional_bool("HEADROOM_DISABLE_KOMPRESS_ANTHROPIC") ) disable_kompress_openai = ( args.disable_kompress_openai if args.disable_kompress_openai is not None else _get_env_optional_bool("HEADROOM_DISABLE_KOMPRESS_OPENAI") ) force_kompress_all = args.force_kompress_all or _get_env_bool( "HEADROOM_FORCE_KOMPRESS_ALL", False ) lossless = getattr(args, "lossless", False) or _get_env_bool("HEADROOM_LOSSLESS", False) # Set OpenRouter API key from CLI if provided if hasattr(args, "openrouter_api_key") and args.openrouter_api_key: os.environ["OPENROUTER_API_KEY"] = args.openrouter_api_key # Parse per-tool compression profiles from CLI and env var tool_profiles = _parse_tool_profiles(args.tool_profile) # Parse extra never-compress tools from CLI and env var exclude_tools = _parse_exclude_tools(args.exclude_tools) protect_tool_results = _parse_csv_tools( args.protect_tool_results or os.environ.get("HEADROOM_PROTECT_TOOL_RESULTS") ) config = ProxyConfig( host=_get_env_str("HEADROOM_HOST", args.host), port=_get_env_int("HEADROOM_PORT", args.port), openai_api_url=_get_env_str("OPENAI_TARGET_API_URL", args.openai_api_url), anthropic_api_url=_get_env_str("ANTHROPIC_TARGET_API_URL", args.anthropic_api_url), anthropic_buffered_request_timeout_seconds=_get_env_int( "HEADROOM_ANTHROPIC_BUFFERED_REQUEST_TIMEOUT_SECONDS", args.anthropic_buffered_request_timeout_seconds, min_value=1, ), vertex_api_url=_get_env_str("VERTEX_TARGET_API_URL", args.vertex_api_url), # Backend settings backend=_get_env_str("HEADROOM_BACKEND", args.backend), # type: ignore[arg-type] bedrock_region=_get_env_str("HEADROOM_BEDROCK_REGION", args.bedrock_region), bedrock_profile=args.bedrock_profile or os.environ.get("AWS_PROFILE"), bedrock_api_url=_get_env_str("BEDROCK_TARGET_API_URL", args.bedrock_api_url), anyllm_provider=_get_env_str("HEADROOM_ANYLLM_PROVIDER", args.anyllm_provider), optimize=optimize, min_tokens_to_crush=_get_env_int("HEADROOM_MIN_TOKENS", args.min_tokens), max_items_after_crush=_get_env_int("HEADROOM_MAX_ITEMS", args.max_items), smart_crusher_with_compaction=( _get_env_bool("HEADROOM_SMART_CRUSHER_COMPACTION", False) if "HEADROOM_SMART_CRUSHER_COMPACTION" in os.environ else None ), cache_enabled=cache_enabled, cache_ttl_seconds=_get_env_int("HEADROOM_CACHE_TTL", args.cache_ttl), rate_limit_enabled=rate_limit_enabled, rate_limit_requests_per_minute=_get_env_int("HEADROOM_RPM", args.rpm), rate_limit_tokens_per_minute=_get_env_int("HEADROOM_TPM", args.tpm), budget_limit_usd=args.budget, budget_period=args.budget_period, log_file=_get_env_str("HEADROOM_LOG_FILE", args.log_file) if args.log_file else os.environ.get("HEADROOM_LOG_FILE"), log_full_messages=args.log_messages or _get_env_bool("HEADROOM_LOG_MESSAGES", False), code_aware_enabled=code_aware_enabled, disable_kompress=disable_kompress, disable_kompress_fallback=disable_kompress_fallback, disable_kompress_anthropic=disable_kompress_anthropic, disable_kompress_openai=disable_kompress_openai, force_kompress_all=force_kompress_all, lossless=lossless, # Connection pool settings max_connections=_get_env_int("HEADROOM_MAX_CONNECTIONS", args.max_connections), max_keepalive_connections=_get_env_int("HEADROOM_MAX_KEEPALIVE", args.max_keepalive), keepalive_expiry=_get_env_float("HEADROOM_KEEPALIVE_EXPIRY", args.keepalive_expiry), http2=not args.no_http2 and _get_env_bool("HEADROOM_HTTP2", True), http_proxy=_get_env_str("HEADROOM_HTTP_PROXY", args.http_proxy or "") or None, read_maturation=_get_env_bool("HEADROOM_READ_MATURATION", False), read_maturation_quiesce_turns=_get_env_int("HEADROOM_READ_MATURATION_QUIESCE_TURNS", 5), read_maturation_max_hold_turns=_get_env_int("HEADROOM_READ_MATURATION_MAX_HOLD_TURNS", 25), read_maturation_min_size_bytes=_get_env_int( "HEADROOM_READ_MATURATION_MIN_SIZE_BYTES", 2048 ), tool_profiles=tool_profiles if tool_profiles else None, exclude_tools=exclude_tools if exclude_tools else None, protect_tool_results=frozenset(protect_tool_results) if protect_tool_results else frozenset(), mode=normalize_proxy_mode(_get_env_str("HEADROOM_MODE", PROXY_MODE_CACHE)), compress_user_messages=args.compress_user_messages or _get_env_bool("HEADROOM_COMPRESS_USER_MESSAGES", False), savings_profile=os.environ.get("HEADROOM_SAVINGS_PROFILE") or "coding", # Default 0.4 keep-ratio so the Kompress text (prose/code) path compresses # meaningfully out of the box; HEADROOM_TARGET_RATIO overrides. target_ratio=( float(os.environ["HEADROOM_TARGET_RATIO"]) if os.environ.get("HEADROOM_TARGET_RATIO") else 0.4 ), compress_system_messages=( _get_env_bool("HEADROOM_COMPRESS_SYSTEM_MESSAGES", False) if "HEADROOM_COMPRESS_SYSTEM_MESSAGES" in os.environ else None ), protect_recent=( int(os.environ["HEADROOM_PROTECT_RECENT"]) if os.environ.get("HEADROOM_PROTECT_RECENT") else None ), protect_analysis_context=( _get_env_bool("HEADROOM_PROTECT_ANALYSIS_CONTEXT", False) if "HEADROOM_PROTECT_ANALYSIS_CONTEXT" in os.environ else None ), accuracy_guard=os.environ.get("HEADROOM_ACCURACY_GUARD") or None, ) # Get worker and concurrency settings workers = _get_env_int("HEADROOM_WORKERS", args.workers) limit_concurrency = _get_env_int("HEADROOM_LIMIT_CONCURRENCY", args.limit_concurrency) run_server(config, workers=workers, limit_concurrency=limit_concurrency)