diff --git a/headroom/agent_savings.py b/headroom/agent_savings.py
new file mode 100644
index 000000000..075633619
--- /dev/null
+++ b/headroom/agent_savings.py
@@ -0,0 +1,226 @@
+"""Shared token-savings profiles for coding agents."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from typing import Protocol
+
+AGENT_90_PROFILE = "agent-90"
+
+
+class CompressConfigLike(Protocol):
+ compress_user_messages: bool
+ compress_system_messages: bool
+ protect_recent: int
+ protect_analysis_context: bool
+ target_ratio: float | None
+ min_tokens_to_compress: int
+
+
+@dataclass(frozen=True)
+class AgentSavingsProfile:
+ """Reusable policy for high-savings agent compression."""
+
+ name: str
+ target_savings: float
+ target_ratio: float
+ compress_user_messages: bool
+ compress_system_messages: bool
+ protect_recent: int
+ protect_analysis_context: bool
+ min_tokens_to_compress: int
+ max_items_after_crush: int
+ smart_crusher_with_compaction: bool
+ force_kompress: bool
+ proxy_mode: str
+ accuracy_guard: str
+
+ @property
+ def savings_percent(self) -> int:
+ return round(self.target_savings * 100)
+
+ def proxy_env(self) -> dict[str, str]:
+ """Return env vars for Headroom proxy/wrapper entry points."""
+
+ return {
+ "HEADROOM_MODE": self.proxy_mode,
+ "HEADROOM_SAVINGS_PROFILE": self.name,
+ "HEADROOM_SAVINGS_TARGET": f"{self.target_savings:.2f}",
+ "HEADROOM_TARGET_RATIO": f"{self.target_ratio:.2f}",
+ "HEADROOM_COMPRESS_USER_MESSAGES": ("1" if self.compress_user_messages else "0"),
+ "HEADROOM_COMPRESS_SYSTEM_MESSAGES": ("1" if self.compress_system_messages else "0"),
+ "HEADROOM_PROTECT_RECENT": str(self.protect_recent),
+ "HEADROOM_PROTECT_ANALYSIS_CONTEXT": ("1" if self.protect_analysis_context else "0"),
+ "HEADROOM_MIN_TOKENS": str(self.min_tokens_to_compress),
+ "HEADROOM_MAX_ITEMS": str(self.max_items_after_crush),
+ "HEADROOM_SMART_CRUSHER_COMPACTION": (
+ "1" if self.smart_crusher_with_compaction else "0"
+ ),
+ "HEADROOM_FORCE_KOMPRESS": "1" if self.force_kompress else "0",
+ "HEADROOM_ACCURACY_GUARD": self.accuracy_guard,
+ }
+
+ def apply_proxy_env_defaults(self, env: dict[str, str]) -> dict[str, str]:
+ """Seed proxy env defaults without overriding explicit user settings."""
+
+ for key, value in self.proxy_env().items():
+ env.setdefault(key, value)
+ return env
+
+
+_PROFILES: dict[str, AgentSavingsProfile] = {
+ AGENT_90_PROFILE: AgentSavingsProfile(
+ name=AGENT_90_PROFILE,
+ target_savings=0.90,
+ target_ratio=0.10,
+ compress_user_messages=True,
+ compress_system_messages=True,
+ protect_recent=2,
+ protect_analysis_context=True,
+ min_tokens_to_compress=120,
+ max_items_after_crush=8,
+ smart_crusher_with_compaction=False,
+ force_kompress=True,
+ proxy_mode="token",
+ accuracy_guard="strict",
+ ),
+ "balanced": AgentSavingsProfile(
+ name="balanced",
+ target_savings=0.70,
+ target_ratio=0.30,
+ compress_user_messages=False,
+ compress_system_messages=False,
+ protect_recent=4,
+ protect_analysis_context=True,
+ min_tokens_to_compress=250,
+ max_items_after_crush=15,
+ smart_crusher_with_compaction=True,
+ force_kompress=False,
+ proxy_mode="token",
+ accuracy_guard="strict",
+ ),
+}
+
+
+def get_agent_savings_profile(name: str | None = None) -> AgentSavingsProfile:
+ """Return a named agent savings profile."""
+
+ key = (name or AGENT_90_PROFILE).strip().lower()
+ try:
+ return _PROFILES[key]
+ except KeyError as exc:
+ valid = ", ".join(sorted(_PROFILES))
+ raise ValueError(f"unknown savings profile {name!r}; expected one of: {valid}") from exc
+
+
+def apply_agent_savings_env_defaults(
+ env: dict[str, str],
+ profile: AgentSavingsProfile | str | None = None,
+) -> dict[str, str]:
+ """Apply agent savings env defaults to a proxy subprocess environment."""
+
+ resolved = (
+ get_agent_savings_profile(profile)
+ if isinstance(profile, str) or profile is None
+ else profile
+ )
+ return resolved.apply_proxy_env_defaults(env)
+
+
+def apply_agent_savings_profile(
+ config: CompressConfigLike,
+ profile: AgentSavingsProfile | str | None = None,
+) -> CompressConfigLike:
+ """Apply a profile to an existing ``CompressConfig``-like object."""
+
+ resolved = (
+ get_agent_savings_profile(profile)
+ if isinstance(profile, str) or profile is None
+ else profile
+ )
+ config.compress_user_messages = resolved.compress_user_messages
+ config.compress_system_messages = resolved.compress_system_messages
+ config.protect_recent = resolved.protect_recent
+ config.protect_analysis_context = resolved.protect_analysis_context
+ config.target_ratio = resolved.target_ratio
+ config.min_tokens_to_compress = resolved.min_tokens_to_compress
+ return config
+
+
+def proxy_pipeline_kwargs(config: object) -> dict[str, object]:
+ """Build per-request pipeline kwargs from proxy config and savings profile.
+
+ The proxy has provider-specific handlers, but the accuracy-sensitive
+ compression knobs should be consistent across Claude, Codex, and Cursor.
+ """
+
+ kwargs: dict[str, object] = {}
+ profile_name = getattr(config, "savings_profile", None)
+ if profile_name:
+ profile = get_agent_savings_profile(str(profile_name))
+ kwargs.update(
+ {
+ "compress_user_messages": profile.compress_user_messages,
+ "compress_system_messages": profile.compress_system_messages,
+ "protect_recent": profile.protect_recent,
+ "protect_analysis_context": profile.protect_analysis_context,
+ "target_ratio": profile.target_ratio,
+ "min_tokens_to_compress": profile.min_tokens_to_compress,
+ "max_items_after_crush": profile.max_items_after_crush,
+ "smart_crusher_with_compaction": profile.smart_crusher_with_compaction,
+ "force_kompress": profile.force_kompress,
+ "read_protection_window": profile.protect_recent,
+ }
+ )
+
+ if getattr(config, "compress_user_messages", False):
+ kwargs["compress_user_messages"] = True
+
+ compress_system_messages = getattr(config, "compress_system_messages", None)
+ if compress_system_messages is not None:
+ kwargs["compress_system_messages"] = bool(compress_system_messages)
+
+ protect_recent = getattr(config, "protect_recent", None)
+ if protect_recent is not None:
+ kwargs["protect_recent"] = int(protect_recent)
+
+ protect_analysis_context = getattr(config, "protect_analysis_context", None)
+ if protect_analysis_context is not None:
+ kwargs["protect_analysis_context"] = bool(protect_analysis_context)
+
+ target_ratio = getattr(config, "target_ratio", None)
+ if target_ratio is not None:
+ kwargs["target_ratio"] = float(target_ratio)
+
+ min_tokens = getattr(config, "min_tokens_to_crush", None)
+ if min_tokens is not None and (not profile_name or int(min_tokens) != 500):
+ kwargs["min_tokens_to_compress"] = int(min_tokens)
+
+ max_items = getattr(config, "max_items_after_crush", None)
+ if max_items is not None and (not profile_name or int(max_items) != 50):
+ kwargs["max_items_after_crush"] = int(max_items)
+
+ smart_crusher_with_compaction = getattr(
+ config,
+ "smart_crusher_with_compaction",
+ None,
+ )
+ if smart_crusher_with_compaction is not None:
+ kwargs["smart_crusher_with_compaction"] = bool(smart_crusher_with_compaction)
+
+ return kwargs
+
+
+def with_target_savings(
+ profile: AgentSavingsProfile,
+ target_savings: float,
+) -> AgentSavingsProfile:
+ """Return a copy of ``profile`` adjusted to a specific savings target."""
+
+ if not 0 < target_savings < 1:
+ raise ValueError("target_savings must be between 0 and 1")
+ return replace(
+ profile,
+ target_savings=target_savings,
+ target_ratio=round(1 - target_savings, 4),
+ )
diff --git a/headroom/cli/agent_savings.py b/headroom/cli/agent_savings.py
new file mode 100644
index 000000000..daf4de996
--- /dev/null
+++ b/headroom/cli/agent_savings.py
@@ -0,0 +1,230 @@
+"""CLI helpers for agent token-savings profiles."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Sequence
+from pathlib import Path
+
+import click
+
+from headroom.agent_savings import get_agent_savings_profile
+
+from .main import main
+
+
+@main.command("agent-savings")
+@click.option(
+ "--profile",
+ default="agent-90",
+ show_default=True,
+ help="Savings profile to render or check.",
+)
+@click.option(
+ "--format",
+ "output_format",
+ type=click.Choice(["shell", "json"]),
+ default="shell",
+ show_default=True,
+ help="Output format for profile environment.",
+)
+@click.option(
+ "--check-perf",
+ is_flag=True,
+ help="Check recent proxy logs against the profile savings target.",
+)
+@click.option(
+ "--hours",
+ type=float,
+ default=24.0,
+ show_default=True,
+ help="Hours of proxy logs to inspect with --check-perf.",
+)
+@click.option(
+ "--accuracy-report",
+ type=click.Path(exists=True, dir_okay=False, path_type=Path),
+ default=None,
+ help="Headroom eval JSON report proving accuracy preservation.",
+)
+@click.option(
+ "--write-smoke-fixture",
+ type=click.Path(file_okay=False, path_type=Path),
+ default=None,
+ help="Write deterministic three-agent PERF/eval fixture into workspace dir.",
+)
+@click.option(
+ "--require-agents",
+ default="",
+ help="Comma-separated clients that must each meet the savings target.",
+)
+@click.option(
+ "--min-accuracy",
+ type=float,
+ default=0.90,
+ show_default=True,
+ help="Minimum accepted accuracy preservation rate.",
+)
+def agent_savings(
+ profile: str,
+ output_format: str,
+ check_perf: bool,
+ hours: float,
+ accuracy_report: Path | None,
+ write_smoke_fixture: Path | None,
+ require_agents: str,
+ min_accuracy: float,
+) -> None:
+ """Render or verify Codex/Claude/Cursor token-savings settings."""
+
+ savings_profile = get_agent_savings_profile(profile)
+ if write_smoke_fixture is not None:
+ eval_path = _write_smoke_fixture(write_smoke_fixture)
+ click.echo(f"Wrote agent-90 smoke fixture to {write_smoke_fixture}")
+ click.echo(
+ "Verify with: HEADROOM_WORKSPACE_DIR="
+ f"{write_smoke_fixture} headroom agent-savings --check-perf "
+ "--hours 0 --require-agents claude,codex,cursor "
+ f"--accuracy-report {eval_path}"
+ )
+ return
+
+ if check_perf or accuracy_report is not None:
+ messages: list[str] = []
+ from headroom.perf.analyzer import build_perf_summary, parse_log_files
+
+ if check_perf:
+ perf_report = parse_log_files(last_n_hours=hours)
+ summary = build_perf_summary(perf_report)
+ measured = float(summary.get("savings_pct", 0.0))
+ target = savings_profile.target_savings * 100
+ if measured < target:
+ raise click.ClickException(
+ f"{measured:.1f}% savings below {target:.1f}% target for {savings_profile.name}"
+ )
+ messages.append(
+ f"{measured:.1f}% savings meets {target:.1f}% target for {savings_profile.name}"
+ )
+ required = _split_required_agents(require_agents)
+ if required:
+ messages.extend(
+ _check_required_agents(
+ perf_report.perf_records,
+ required,
+ target,
+ )
+ )
+
+ if accuracy_report is not None:
+ accuracy = _read_accuracy_rate(accuracy_report)
+ if accuracy < min_accuracy:
+ raise click.ClickException(
+ f"{accuracy * 100:.1f}% accuracy below {min_accuracy * 100:.1f}% target"
+ )
+ messages.append(
+ f"{accuracy * 100:.1f}% accuracy meets {min_accuracy * 100:.1f}% target"
+ )
+
+ click.echo("\n".join(messages))
+ return
+
+ env = savings_profile.proxy_env()
+ if output_format == "json":
+ click.echo(json.dumps(env, indent=2, sort_keys=True))
+ return
+
+ for key, value in env.items():
+ click.echo(f"export {key}={json.dumps(value)}")
+
+
+def _read_accuracy_rate(path: Path) -> float:
+ payload = json.loads(path.read_text(encoding="utf-8"))
+ if isinstance(payload, dict):
+ totals = payload.get("totals")
+ if isinstance(totals, dict) and totals.get("accuracy_rate") is not None:
+ return float(totals["accuracy_rate"])
+ if payload.get("accuracy_preservation_rate") is not None:
+ return float(payload["accuracy_preservation_rate"])
+ raise click.ClickException(
+ f"{path} does not contain totals.accuracy_rate or accuracy_preservation_rate"
+ )
+
+
+def _split_required_agents(raw: str) -> list[str]:
+ return [agent.strip().lower() for agent in raw.split(",") if agent.strip()]
+
+
+def _check_required_agents(
+ records: Sequence[object],
+ required_agents: list[str],
+ target_percent: float,
+) -> list[str]:
+ messages: list[str] = []
+ records_by_agent: dict[str, list[object]] = {}
+ for record in records:
+ client = str(getattr(record, "client", "") or "").strip().lower()
+ if client:
+ records_by_agent.setdefault(client, []).append(record)
+
+ missing = [agent for agent in required_agents if agent not in records_by_agent]
+ if missing:
+ raise click.ClickException("missing required agent traffic: " + ", ".join(missing))
+
+ for agent in required_agents:
+ agent_records = records_by_agent[agent]
+ before = sum(int(getattr(record, "tokens_before", 0)) for record in agent_records)
+ saved = sum(int(getattr(record, "tokens_saved", 0)) for record in agent_records)
+ measured = (saved / before * 100) if before > 0 else 0.0
+ if measured < target_percent:
+ raise click.ClickException(
+ f"{agent}: {measured:.1f}% savings below {target_percent:.1f}% target"
+ )
+ messages.append(f"{agent}: {measured:.1f}% savings meets {target_percent:.1f}% target")
+ return messages
+
+
+def _write_smoke_fixture(workspace: Path) -> Path:
+ logs_dir = workspace / "logs"
+ logs_dir.mkdir(parents=True, exist_ok=True)
+ perf_lines = [
+ _perf_line(
+ "2026-06-10 10:00:00,000", "hr_smoke_claude", "claude-sonnet", "claude", 1000, 80
+ ),
+ _perf_line("2026-06-10 10:01:00,000", "hr_smoke_codex", "gpt-5", "codex", 1000, 90),
+ _perf_line("2026-06-10 10:02:00,000", "hr_smoke_cursor", "gpt-5", "cursor", 1000, 70),
+ ]
+ (logs_dir / "proxy.log").write_text("\n".join(perf_lines) + "\n", encoding="utf-8")
+ eval_path = workspace / "agent-90-eval.json"
+ eval_path.write_text(
+ json.dumps(
+ {
+ "totals": {
+ "cases": 3,
+ "passed": 3,
+ "accuracy_rate": 1.0,
+ "tokens_original": 3000,
+ "tokens_compressed": 240,
+ }
+ },
+ indent=2,
+ )
+ + "\n",
+ encoding="utf-8",
+ )
+ return eval_path
+
+
+def _perf_line(
+ timestamp: str,
+ request_id: str,
+ model: str,
+ client: str,
+ before: int,
+ after: int,
+) -> str:
+ saved = before - after
+ return (
+ f"{timestamp} - headroom.proxy - INFO - [{request_id}] PERF "
+ f"model={model} msgs=3 tok_before={before} tok_after={after} "
+ f"tok_saved={saved} cache_read=0 cache_write=0 cache_hit_pct=0 "
+ f"opt_ms=1 transforms=agent90_smoke client={client}"
+ )
diff --git a/headroom/cli/main.py b/headroom/cli/main.py
index 40e0a18b4..9581c03f4 100644
--- a/headroom/cli/main.py
+++ b/headroom/cli/main.py
@@ -36,6 +36,7 @@ def main(ctx: click.Context) -> None:
def _register_commands() -> None:
"""Register all subcommand groups."""
from . import (
+ agent_savings, # noqa: F401
capture, # noqa: F401
evals, # noqa: F401
init, # noqa: F401
diff --git a/headroom/cli/proxy.py b/headroom/cli/proxy.py
index 44085c1be..0cf5e1ed4 100644
--- a/headroom/cli/proxy.py
+++ b/headroom/cli/proxy.py
@@ -61,6 +61,22 @@ def _get_env_bool(name: str, default: bool) -> bool:
return val.lower() in ("true", "1", "yes", "on")
+def _get_env_bool_optional(name: str) -> bool | None:
+ if name not in os.environ:
+ return None
+ return _get_env_bool(name, False)
+
+
+def _get_env_int_optional(name: str) -> int | None:
+ val = os.environ.get(name)
+ return int(val) if val is not None and val != "" else None
+
+
+def _get_env_float_optional(name: str) -> float | None:
+ val = os.environ.get(name)
+ return float(val) if val is not None and val != "" else None
+
+
def _selected_context_tool() -> str:
raw = os.environ.get(_CONTEXT_TOOL_ENV, "").strip().lower().replace("_", "-")
if not raw:
@@ -773,6 +789,16 @@ def proxy(
ccr_inject_tool=not no_ccr_inject_tool,
ccr_inject_marker=not no_ccr_marker,
ccr_proactive_expansion=not no_ccr_proactive_expansion,
+ compress_user_messages=_get_env_bool("HEADROOM_COMPRESS_USER_MESSAGES", False),
+ min_tokens_to_crush=_get_env_int_optional("HEADROOM_MIN_TOKENS") or 500,
+ max_items_after_crush=_get_env_int_optional("HEADROOM_MAX_ITEMS") or 50,
+ smart_crusher_with_compaction=_get_env_bool_optional("HEADROOM_SMART_CRUSHER_COMPACTION"),
+ savings_profile=os.environ.get("HEADROOM_SAVINGS_PROFILE") or None,
+ target_ratio=_get_env_float_optional("HEADROOM_TARGET_RATIO"),
+ compress_system_messages=_get_env_bool_optional("HEADROOM_COMPRESS_SYSTEM_MESSAGES"),
+ protect_recent=_get_env_int_optional("HEADROOM_PROTECT_RECENT"),
+ protect_analysis_context=_get_env_bool_optional("HEADROOM_PROTECT_ANALYSIS_CONTEXT"),
+ accuracy_guard=os.environ.get("HEADROOM_ACCURACY_GUARD") or None,
# Flatten repeat-flag tuple AND any comma-separated values inside it.
# `--proxy-extension a,b --proxy-extension c` and `HEADROOM_PROXY_EXTENSIONS=a,b,c`
# both yield ["a", "b", "c"]. None when nothing was supplied.
diff --git a/headroom/cli/wrap.py b/headroom/cli/wrap.py
index da00b4064..2d26bcf26 100644
--- a/headroom/cli/wrap.py
+++ b/headroom/cli/wrap.py
@@ -109,6 +109,8 @@ _WRAP_PROXY_TIMEOUT_ML_MODULES = ("torch", "sentence_transformers", "spacy")
# Code, so the agent loop is unaffected).
_TOOL_SEARCH_ENV = "ENABLE_TOOL_SEARCH"
_TOOL_SEARCH_DEFAULT = "true"
+_AGENT_SAVINGS_WRAP_AGENTS = {"claude", "codex", "cursor"}
+_DEFAULT_AGENT_SAVINGS_PROFILE = "agent-90"
def _normalize_tool_search_mode(value: str) -> str:
@@ -198,6 +200,14 @@ def _ml_wrap_extras_detected() -> bool:
return any(_module_available(module_name) for module_name in _WRAP_PROXY_TIMEOUT_ML_MODULES)
+def _wrap_agent_savings_profile(agent_type: str) -> str | None:
+ """Return the savings profile required for agent wrappers, if any."""
+
+ if agent_type not in _AGENT_SAVINGS_WRAP_AGENTS:
+ return None
+ return os.environ.get("HEADROOM_SAVINGS_PROFILE") or _DEFAULT_AGENT_SAVINGS_PROFILE
+
+
def _default_wrap_proxy_timeout_seconds() -> int:
"""Return the default wrap proxy startup timeout for this environment."""
@@ -355,6 +365,11 @@ def _start_proxy(
if agent_type != "unknown":
proxy_env["HEADROOM_AGENT_TYPE"] = agent_type
proxy_env.setdefault("HEADROOM_STACK", f"wrap_{agent_type}")
+ savings_profile = _wrap_agent_savings_profile(agent_type)
+ if savings_profile is not None:
+ from headroom.agent_savings import apply_agent_savings_env_defaults
+
+ apply_agent_savings_env_defaults(proxy_env, savings_profile)
if openai_api_url:
proxy_env["OPENAI_TARGET_API_URL"] = openai_api_url
if anthropic_api_url:
@@ -1906,6 +1921,12 @@ def _ensure_proxy(
missing.append("learn")
if code_graph and not running_config.get("code_graph"):
missing.append("code_graph")
+ expected_savings_profile = helpers._wrap_agent_savings_profile(agent_type)
+ if (
+ expected_savings_profile is not None
+ and running_config.get("savings_profile") != expected_savings_profile
+ ):
+ missing.append("savings-profile")
if openai_api_url:
running_openai_url = _normalize_proxy_api_url(
running_config.get("openai_api_url")
diff --git a/headroom/perf/analyzer.py b/headroom/perf/analyzer.py
index c54fe1336..b364edc4f 100644
--- a/headroom/perf/analyzer.py
+++ b/headroom/perf/analyzer.py
@@ -111,7 +111,14 @@ def _parse_kv(kv_str: str) -> dict[str, str]:
# Handle transforms= specially since its value contains spaces
if "transforms=" in kv_str:
before, transforms_val = kv_str.split("transforms=", 1)
- result["transforms"] = transforms_val.strip()
+ transform_parts: list[str] = []
+ for part in transforms_val.split():
+ if "=" in part:
+ k, v = part.split("=", 1)
+ result[k] = v
+ else:
+ transform_parts.append(part)
+ result["transforms"] = " ".join(transform_parts).strip()
kv_str = before
for part in kv_str.split():
if "=" in part:
@@ -136,6 +143,7 @@ class PerfRecord:
cache_hit_pct: int = 0
optimization_ms: float = 0
transforms: list[str] = field(default_factory=list)
+ client: str = ""
@dataclass
@@ -226,7 +234,8 @@ def parse_log_files(last_n_hours: float = 168.0) -> PerfReport:
report = PerfReport()
report.requested_hours = last_n_hours
- if not LOG_DIR.exists():
+ log_dir = _paths.log_dir()
+ if not log_dir.exists():
return report
cutoff = datetime.now() - timedelta(hours=last_n_hours) if last_n_hours > 0 else None
@@ -250,7 +259,7 @@ def parse_log_files(last_n_hours: float = 168.0) -> PerfReport:
report.newest_kept_ts = ts_str
# Collect log files: proxy.log, proxy.log.1, proxy.log.2, ...
- log_files = sorted(LOG_DIR.glob("proxy.log*"), key=lambda p: p.stat().st_mtime)
+ log_files = sorted(log_dir.glob("proxy.log*"), key=lambda p: p.stat().st_mtime)
for log_file in log_files:
report.log_files_read += 1
@@ -299,6 +308,7 @@ def parse_log_files(last_n_hours: float = 168.0) -> PerfReport:
cache_hit_pct=int(kv.get("cache_hit_pct", 0)),
optimization_ms=float(kv.get("opt_ms", 0)),
transforms=transforms,
+ client=kv.get("client", ""),
)
)
continue
@@ -589,7 +599,7 @@ def format_report(report: PerfReport) -> str:
lines.append(
f"Log files: {report.log_files_read} | Lines parsed: {report.total_lines_parsed:,}"
)
- lines.append(f"Log dir: {LOG_DIR}")
+ lines.append(f"Log dir: {_paths.log_dir()}")
return "\n".join(lines)
@@ -611,6 +621,7 @@ PERF_RECORD_FIELDS = [
"timestamp",
"request_id",
"model",
+ "client",
"num_messages",
"tokens_before",
"tokens_after",
diff --git a/headroom/proxy/models.py b/headroom/proxy/models.py
index 98bfc1bad..0d1a98fef 100644
--- a/headroom/proxy/models.py
+++ b/headroom/proxy/models.py
@@ -113,6 +113,7 @@ class ProxyConfig:
image_optimize: bool = True
min_tokens_to_crush: int = 500
max_items_after_crush: int = 50
+ smart_crusher_with_compaction: bool | None = None
keep_last_turns: int = 4
# CCR Tool Injection
@@ -153,6 +154,14 @@ class ProxyConfig:
# router would otherwise have nothing eligible to compress.
# CLI: --compress-user-messages; env: HEADROOM_COMPRESS_USER_MESSAGES=1.
compress_user_messages: bool = False
+ # Named savings policy shared across Claude/Codex/Cursor proxy handlers.
+ # CLI/env: HEADROOM_SAVINGS_PROFILE=agent-90.
+ savings_profile: str | None = None
+ target_ratio: float | None = None
+ compress_system_messages: bool | None = None
+ protect_recent: int | None = None
+ protect_analysis_context: bool | None = None
+ accuracy_guard: str | None = None
# Extra tool names whose outputs are never compressed, merged with the
# built-in DEFAULT_EXCLUDE_TOOLS. None means built-in defaults only.
diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py
index 24c43ab35..c85e3a1a2 100644
--- a/headroom/proxy/server.py
+++ b/headroom/proxy/server.py
@@ -61,6 +61,7 @@ except ImportError:
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 (
@@ -366,11 +367,20 @@ class HeadroomProxy(
# 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,
tool_profiles=config.tool_profiles,
read_lifecycle=ReadLifecycleConfig(enabled=config.read_lifecycle),
ccr_inject_marker=config.ccr_inject_marker,
+ 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),
+ ),
)
if config.disable_kompress:
router_config.enable_kompress = False
@@ -386,7 +396,7 @@ class HeadroomProxy(
# 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 config.compress_user_messages:
+ if profile_kwargs.get("compress_user_messages"):
router_config.skip_user_messages = False
transforms = [
CacheAligner(CacheAlignerConfig(enabled=False)),
@@ -1697,6 +1707,11 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
"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,
@@ -1710,6 +1725,44 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
"openai_api_url": config.openai_api_url,
"gemini_api_url": config.gemini_api_url,
"cloudcode_api_url": config.cloudcode_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,
"pid": os.getpid(),
}
return payload
diff --git a/headroom/transforms/compression_units.py b/headroom/transforms/compression_units.py
index 62ab046b3..e0c678fdf 100644
--- a/headroom/transforms/compression_units.py
+++ b/headroom/transforms/compression_units.py
@@ -206,6 +206,7 @@ def compress_unit_with_router(
*,
router: ContentRouter,
tokenizer: TokenCounterLike,
+ target_ratio: float | None = None,
) -> UnitCompressionResult:
"""Compress one safe text unit through ContentRouter.
@@ -251,11 +252,19 @@ def compress_unit_with_router(
return _with_reason(reason=f"cache_zone_{unit.cache_zone}")
if len(unit.text) < unit.min_bytes:
return _with_reason(reason="below_unit_floor")
+
+ prior_target_ratio = getattr(router, "_runtime_target_ratio", None)
+ if target_ratio is not None:
+ router._runtime_target_ratio = target_ratio
if _CCR_MARKER_RE.search(unit.text):
- replacement, marker_transforms, router_result = _compress_live_text_with_markers(
- unit,
- router=router,
- )
+ try:
+ replacement, marker_transforms, router_result = _compress_live_text_with_markers(
+ unit,
+ router=router,
+ )
+ finally:
+ if target_ratio is not None:
+ router._runtime_target_ratio = prior_target_ratio
if replacement == unit.text:
return _with_reason(
router_result=router_result,
@@ -287,12 +296,16 @@ def compress_unit_with_router(
reason_category="applied",
)
- router_result = router.compress(
- unit.text,
- context=unit.context,
- question=unit.question,
- bias=unit.bias,
- )
+ try:
+ router_result = router.compress(
+ unit.text,
+ context=unit.context,
+ question=unit.question,
+ bias=unit.bias,
+ )
+ finally:
+ if target_ratio is not None:
+ router._runtime_target_ratio = prior_target_ratio
replacement = router_result.compressed
strategy = router_result.strategy_used.value
if replacement == unit.text:
diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py
index 622bdf3d1..410459c52 100644
--- a/headroom/transforms/content_router.py
+++ b/headroom/transforms/content_router.py
@@ -504,6 +504,8 @@ class ContentRouterConfig:
# CCR (Compress-Cache-Retrieve) settings for SmartCrusher
ccr_enabled: bool = True # Enable CCR marker injection for reversible compression
ccr_inject_marker: bool = True # Add retrieval markers to compressed content
+ smart_crusher_max_items_after_crush: int | None = None
+ smart_crusher_with_compaction: bool = True
# Tag protection: preserve custom/workflow XML tags from text compression.
# When False (default), entire content blocks are
@@ -940,7 +942,12 @@ class ContentRouter(Transform):
# Determine strategy from content analysis
mixed = is_mixed_content(content)
detection = _detect_content(content)
- strategy = self._determine_strategy(content)
+ force_kompress = bool(getattr(self, "_runtime_force_kompress", False))
+ strategy = (
+ CompressionStrategy.KOMPRESS
+ if force_kompress
+ else self._determine_strategy(content)
+ )
if debug_enabled:
_log_router_debug(
"content_router_input",
@@ -948,7 +955,13 @@ class ContentRouter(Transform):
detected_content_type=detection.content_type.value,
detection_confidence=detection.confidence,
selected_strategy=strategy.value,
- selection_reason="mixed_content" if mixed else "content_detection",
+ selection_reason=(
+ "runtime_force_kompress"
+ if force_kompress
+ else "mixed_content"
+ if mixed
+ else "content_detection"
+ ),
)
if strategy == CompressionStrategy.MIXED:
@@ -1565,14 +1578,23 @@ class ContentRouter(Transform):
if self._smart_crusher is None:
try:
from ..config import CCRConfig
- from .smart_crusher import SmartCrusher
+ from .smart_crusher import SmartCrusher, SmartCrusherConfig
# Pass CCR config for marker injection
ccr_config = CCRConfig(
enabled=self.config.ccr_enabled,
inject_retrieval_marker=self.config.ccr_inject_marker,
)
- self._smart_crusher = SmartCrusher(ccr_config=ccr_config)
+ crusher_config = SmartCrusherConfig()
+ if self.config.smart_crusher_max_items_after_crush is not None:
+ crusher_config.max_items_after_crush = (
+ self.config.smart_crusher_max_items_after_crush
+ )
+ self._smart_crusher = SmartCrusher(
+ config=crusher_config,
+ ccr_config=ccr_config,
+ with_compaction=self.config.smart_crusher_with_compaction,
+ )
except ImportError:
logger.debug("SmartCrusher not available")
return self._smart_crusher
@@ -1930,6 +1952,7 @@ class ContentRouter(Transform):
)
# Store runtime options on self for access by _route_and_compress_block
self._runtime_target_ratio: float | None = kwargs.get("target_ratio")
+ self._runtime_force_kompress: bool = bool(kwargs.get("force_kompress", False))
self._runtime_kompress_model: str | None = kwargs.get("kompress_model")
# F2.2: capture the per-request CompressionPolicy so
# ``_record_to_toin`` can gate TOIN writes on
@@ -1969,6 +1992,9 @@ class ContentRouter(Transform):
)
else:
read_protection_window = num_messages # 0.0 = protect all (old behavior)
+ runtime_read_protection_window = kwargs.get("read_protection_window")
+ if runtime_read_protection_window is not None:
+ read_protection_window = max(0, int(runtime_read_protection_window))
# Adaptive compression ratio: scale with context pressure
if model_limit > 0:
diff --git a/tests/test_agent_savings.py b/tests/test_agent_savings.py
new file mode 100644
index 000000000..dd9a13213
--- /dev/null
+++ b/tests/test_agent_savings.py
@@ -0,0 +1,488 @@
+from __future__ import annotations
+
+import json
+from importlib import import_module
+from types import SimpleNamespace
+
+import pytest
+from click.testing import CliRunner
+
+from headroom.agent_savings import (
+ AGENT_90_PROFILE,
+ apply_agent_savings_env_defaults,
+ apply_agent_savings_profile,
+ get_agent_savings_profile,
+ proxy_pipeline_kwargs,
+ with_target_savings,
+)
+from headroom.cli.main import main
+from headroom.compress import CompressConfig, compress
+from headroom.proxy.models import ProxyConfig
+from headroom.transforms.compression_units import (
+ CompressionUnit,
+ compress_unit_with_router,
+)
+from headroom.transforms.content_router import (
+ CompressionStrategy,
+ ContentRouter,
+ ContentRouterConfig,
+ RouterCompressionResult,
+)
+
+compress_module = import_module("headroom.compress")
+
+
+def test_agent_90_profile_sets_accuracy_preserving_compress_config() -> None:
+ cfg = CompressConfig()
+
+ apply_agent_savings_profile(cfg, AGENT_90_PROFILE)
+
+ assert cfg.compress_user_messages is True
+ assert cfg.compress_system_messages is True
+ assert cfg.protect_recent == 2
+ assert cfg.protect_analysis_context is True
+ assert cfg.target_ratio == 0.10
+ assert cfg.min_tokens_to_compress == 120
+
+
+def test_agent_90_profile_exports_cross_agent_proxy_env() -> None:
+ profile = get_agent_savings_profile(AGENT_90_PROFILE)
+
+ env = profile.proxy_env()
+
+ assert env["HEADROOM_MODE"] == "token"
+ assert env["HEADROOM_SAVINGS_PROFILE"] == "agent-90"
+ assert env["HEADROOM_SAVINGS_TARGET"] == "0.90"
+ assert env["HEADROOM_TARGET_RATIO"] == "0.10"
+ assert env["HEADROOM_COMPRESS_USER_MESSAGES"] == "1"
+ assert env["HEADROOM_COMPRESS_SYSTEM_MESSAGES"] == "1"
+ assert env["HEADROOM_MAX_ITEMS"] == "8"
+ assert env["HEADROOM_SMART_CRUSHER_COMPACTION"] == "0"
+ assert env["HEADROOM_FORCE_KOMPRESS"] == "1"
+ assert env["HEADROOM_ACCURACY_GUARD"] == "strict"
+
+
+def test_agent_savings_env_defaults_preserve_user_overrides() -> None:
+ env = {
+ "HEADROOM_TARGET_RATIO": "0.25",
+ "HEADROOM_MAX_ITEMS": "12",
+ }
+
+ apply_agent_savings_env_defaults(env, AGENT_90_PROFILE)
+
+ assert env["HEADROOM_SAVINGS_PROFILE"] == "agent-90"
+ assert env["HEADROOM_TARGET_RATIO"] == "0.25"
+ assert env["HEADROOM_MAX_ITEMS"] == "12"
+ assert env["HEADROOM_SMART_CRUSHER_COMPACTION"] == "0"
+
+
+def test_unknown_agent_savings_profile_lists_valid_profiles() -> None:
+ with pytest.raises(ValueError, match="agent-90"):
+ get_agent_savings_profile("missing")
+
+
+def test_with_target_savings_recomputes_target_ratio() -> None:
+ profile = with_target_savings(get_agent_savings_profile("balanced"), 0.85)
+
+ assert profile.target_savings == 0.85
+ assert profile.target_ratio == 0.15
+
+
+def test_agent_savings_cli_renders_shell_exports() -> None:
+ result = CliRunner().invoke(main, ["agent-savings", "--profile", "agent-90"])
+
+ assert result.exit_code == 0
+ assert 'export HEADROOM_SAVINGS_PROFILE="agent-90"' in result.output
+ assert 'export HEADROOM_SAVINGS_TARGET="0.90"' in result.output
+ assert 'export HEADROOM_ACCURACY_GUARD="strict"' in result.output
+
+
+def test_agent_savings_cli_renders_json() -> None:
+ result = CliRunner().invoke(
+ main,
+ ["agent-savings", "--profile", "agent-90", "--format", "json"],
+ )
+
+ assert result.exit_code == 0
+ assert '"HEADROOM_TARGET_RATIO": "0.10"' in result.output
+
+
+def test_compress_applies_agent_savings_profile_to_pipeline(monkeypatch) -> None:
+ captured: dict[str, object] = {}
+ messages = [{"role": "user", "content": "x" * 500}]
+
+ class Pipeline:
+ def apply(self, **kwargs):
+ captured.update(kwargs)
+ return SimpleNamespace(
+ messages=messages,
+ tokens_before=1000,
+ tokens_after=100,
+ transforms_applied=["test"],
+ )
+
+ monkeypatch.setattr(compress_module, "_get_pipeline", lambda: Pipeline())
+
+ config = CompressConfig()
+ apply_agent_savings_profile(config, AGENT_90_PROFILE)
+
+ result = compress(messages, config=config)
+
+ assert result.compression_ratio == 0.9
+ assert captured["compress_user_messages"] is True
+ assert captured["compress_system_messages"] is True
+ assert captured["protect_recent"] == 2
+ assert captured["protect_analysis_context"] is True
+ assert captured["target_ratio"] == 0.10
+ assert captured["min_tokens_to_compress"] == 120
+
+
+def test_agent_90_profile_applies_to_proxy_config_runtime_kwargs() -> None:
+ config = ProxyConfig(savings_profile="agent-90")
+
+ kwargs = proxy_pipeline_kwargs(config)
+
+ assert kwargs["compress_user_messages"] is True
+ assert kwargs["compress_system_messages"] is True
+ assert kwargs["protect_recent"] == 2
+ assert kwargs["protect_analysis_context"] is True
+ assert kwargs["target_ratio"] == 0.10
+ assert kwargs["min_tokens_to_compress"] == 120
+ assert kwargs["max_items_after_crush"] == 8
+ assert kwargs["smart_crusher_with_compaction"] is False
+ assert kwargs["force_kompress"] is True
+ assert kwargs["read_protection_window"] == 2
+
+
+def test_proxy_explicit_config_overrides_agent_90_profile() -> None:
+ config = ProxyConfig(
+ savings_profile="agent-90",
+ target_ratio=0.25,
+ protect_recent=5,
+ min_tokens_to_crush=300,
+ )
+
+ kwargs = proxy_pipeline_kwargs(config)
+
+ assert kwargs["target_ratio"] == 0.25
+ assert kwargs["protect_recent"] == 5
+ assert kwargs["min_tokens_to_compress"] == 300
+
+
+def test_agent_90_router_uses_ccr_sampling_not_lossless_table() -> None:
+ router = ContentRouter(
+ ContentRouterConfig(
+ smart_crusher_max_items_after_crush=8,
+ smart_crusher_with_compaction=False,
+ )
+ )
+
+ crusher = router._get_smart_crusher()
+
+ assert crusher is not None
+ assert crusher.config.max_items_after_crush == 8
+ assert crusher._with_compaction is False
+
+
+def test_agent_90_router_json_tool_output_reaches_target_with_needle() -> None:
+ needle = "CRITICAL_NEEDLE_42"
+ rows = [
+ {
+ "id": i,
+ "status": "ok",
+ "message": "normal repeated telemetry payload",
+ "value": i % 7,
+ }
+ for i in range(1000)
+ ]
+ rows.append(
+ {
+ "id": 99999,
+ "status": "error",
+ "message": f"{needle} root cause disk full",
+ "value": 999.99,
+ }
+ )
+ router = ContentRouter(
+ ContentRouterConfig(
+ smart_crusher_max_items_after_crush=8,
+ smart_crusher_with_compaction=False,
+ )
+ )
+
+ result = router.compress(json.dumps(rows), question=f"Find {needle}")
+ before = len(result.original.split())
+ after = len(result.compressed.split())
+
+ assert 1 - after / before >= 0.90
+ assert needle in result.compressed
+ assert "< None:
+ captured_config: dict[str, ProxyConfig] = {}
+
+ def mock_run_server(config: ProxyConfig, **kwargs: object) -> None:
+ captured_config["config"] = config
+
+ runner = CliRunner()
+ with pytest.MonkeyPatch.context() as mp:
+ mp.setattr("headroom.proxy.server.run_server", mock_run_server)
+ result = runner.invoke(
+ main,
+ ["proxy"],
+ env={"HEADROOM_SAVINGS_PROFILE": "agent-90"},
+ catch_exceptions=False,
+ )
+
+ assert result.exit_code == 0, result.output
+ config = captured_config["config"]
+ assert config.savings_profile == "agent-90"
+ assert proxy_pipeline_kwargs(config)["target_ratio"] == 0.10
+
+
+def test_unit_router_receives_agent_target_ratio() -> None:
+ seen: dict[str, object] = {}
+
+ class Tokenizer:
+ def count_text(self, text: str) -> int:
+ return len(text.split())
+
+ class Router:
+ _runtime_target_ratio = None
+
+ def compress(self, text: str, **kwargs: object) -> RouterCompressionResult:
+ seen["target_ratio"] = self._runtime_target_ratio
+ return RouterCompressionResult(
+ compressed="short text",
+ original=text,
+ strategy_used=CompressionStrategy.KOMPRESS,
+ strategy_chain=["kompress"],
+ )
+
+ unit = CompressionUnit(
+ text=("long text " * 40) + "\nRetrieve more: hash=abc123\n",
+ provider="openai",
+ endpoint="responses",
+ role="assistant",
+ item_type="message",
+ cache_zone="live",
+ mutable=True,
+ min_bytes=1,
+ metadata={"compress_assistant": "true"},
+ )
+
+ result = compress_unit_with_router(
+ unit,
+ router=Router(),
+ tokenizer=Tokenizer(),
+ target_ratio=0.10,
+ )
+
+ assert result.modified is True
+ assert seen["target_ratio"] == 0.10
+
+
+def test_agent_savings_check_perf_and_accuracy_report_passes(
+ monkeypatch,
+ tmp_path,
+) -> None:
+ from headroom.perf import analyzer
+
+ monkeypatch.setattr(analyzer, "parse_log_files", lambda last_n_hours: object())
+ monkeypatch.setattr(
+ analyzer,
+ "build_perf_summary",
+ lambda report: {"savings_pct": 92.0},
+ )
+ report = tmp_path / "eval.json"
+ report.write_text(json.dumps({"totals": {"accuracy_rate": 1.0}}))
+
+ result = CliRunner().invoke(
+ main,
+ [
+ "agent-savings",
+ "--profile",
+ "agent-90",
+ "--check-perf",
+ "--accuracy-report",
+ str(report),
+ ],
+ )
+
+ assert result.exit_code == 0, result.output
+ assert "92.0% savings meets 90.0%" in result.output
+ assert "100.0% accuracy meets 90.0%" in result.output
+
+
+def test_agent_savings_accuracy_report_below_threshold_fails(
+ monkeypatch,
+ tmp_path,
+) -> None:
+ from headroom.perf import analyzer
+
+ monkeypatch.setattr(analyzer, "parse_log_files", lambda last_n_hours: object())
+ monkeypatch.setattr(
+ analyzer,
+ "build_perf_summary",
+ lambda report: {"savings_pct": 92.0},
+ )
+ report = tmp_path / "eval.json"
+ report.write_text(json.dumps({"totals": {"accuracy_rate": 0.89}}))
+
+ result = CliRunner().invoke(
+ main,
+ [
+ "agent-savings",
+ "--profile",
+ "agent-90",
+ "--check-perf",
+ "--accuracy-report",
+ str(report),
+ ],
+ )
+
+ assert result.exit_code != 0
+ assert "89.0% accuracy below 90.0%" in result.output
+
+
+def test_agent_savings_requires_each_agent_to_meet_target(monkeypatch) -> None:
+ from headroom.perf import analyzer
+ from headroom.perf.analyzer import PerfRecord, PerfReport
+
+ report = PerfReport(
+ perf_records=[
+ PerfRecord(
+ timestamp="2026-06-10 10:00:00,000",
+ request_id="claude-1",
+ model="claude-sonnet",
+ client="claude",
+ tokens_before=1000,
+ tokens_after=80,
+ tokens_saved=920,
+ ),
+ PerfRecord(
+ timestamp="2026-06-10 10:01:00,000",
+ request_id="codex-1",
+ model="gpt-5",
+ client="codex",
+ tokens_before=1000,
+ tokens_after=90,
+ tokens_saved=910,
+ ),
+ PerfRecord(
+ timestamp="2026-06-10 10:02:00,000",
+ request_id="cursor-1",
+ model="gpt-5",
+ client="cursor",
+ tokens_before=1000,
+ tokens_after=70,
+ tokens_saved=930,
+ ),
+ ]
+ )
+ monkeypatch.setattr(analyzer, "parse_log_files", lambda last_n_hours: report)
+
+ result = CliRunner().invoke(
+ main,
+ [
+ "agent-savings",
+ "--check-perf",
+ "--require-agents",
+ "claude,codex,cursor",
+ ],
+ )
+
+ assert result.exit_code == 0, result.output
+ assert "claude: 92.0% savings meets 90.0%" in result.output
+ assert "codex: 91.0% savings meets 90.0%" in result.output
+ assert "cursor: 93.0% savings meets 90.0%" in result.output
+
+
+def test_agent_savings_required_agent_missing_fails(monkeypatch) -> None:
+ from headroom.perf import analyzer
+ from headroom.perf.analyzer import PerfRecord, PerfReport
+
+ report = PerfReport(
+ perf_records=[
+ PerfRecord(
+ timestamp="2026-06-10 10:00:00,000",
+ request_id="claude-1",
+ model="claude-sonnet",
+ client="claude",
+ tokens_before=1000,
+ tokens_after=80,
+ tokens_saved=920,
+ ),
+ PerfRecord(
+ timestamp="2026-06-10 10:01:00,000",
+ request_id="codex-1",
+ model="gpt-5",
+ client="codex",
+ tokens_before=1000,
+ tokens_after=90,
+ tokens_saved=910,
+ ),
+ ]
+ )
+ monkeypatch.setattr(analyzer, "parse_log_files", lambda last_n_hours: report)
+
+ result = CliRunner().invoke(
+ main,
+ [
+ "agent-savings",
+ "--check-perf",
+ "--require-agents",
+ "claude,codex,cursor",
+ ],
+ )
+
+ assert result.exit_code != 0
+ assert "missing required agent traffic: cursor" in result.output
+
+
+def test_agent_savings_writes_three_agent_smoke_fixture(tmp_path) -> None:
+ workspace = tmp_path / "workspace"
+
+ result = CliRunner().invoke(
+ main,
+ ["agent-savings", "--write-smoke-fixture", str(workspace)],
+ )
+
+ assert result.exit_code == 0, result.output
+ assert (workspace / "logs" / "proxy.log").exists()
+ eval_report = workspace / "agent-90-eval.json"
+ assert eval_report.exists()
+ assert "--require-agents claude,codex,cursor" in result.output
+
+
+def test_agent_savings_smoke_fixture_passes_real_gate(tmp_path) -> None:
+ workspace = tmp_path / "workspace"
+ runner = CliRunner()
+
+ write_result = runner.invoke(
+ main,
+ ["agent-savings", "--write-smoke-fixture", str(workspace)],
+ )
+ assert write_result.exit_code == 0, write_result.output
+
+ gate_result = runner.invoke(
+ main,
+ [
+ "agent-savings",
+ "--check-perf",
+ "--hours",
+ "0",
+ "--require-agents",
+ "claude,codex,cursor",
+ "--accuracy-report",
+ str(workspace / "agent-90-eval.json"),
+ ],
+ env={"HEADROOM_WORKSPACE_DIR": str(workspace)},
+ )
+
+ assert gate_result.exit_code == 0, gate_result.output
+ assert "claude: 92.0% savings meets 90.0%" in gate_result.output
+ assert "codex: 91.0% savings meets 90.0%" in gate_result.output
+ assert "cursor: 93.0% savings meets 90.0%" in gate_result.output
+ assert "100.0% accuracy meets 90.0%" in gate_result.output
diff --git a/tests/test_cli/test_wrap_persistent.py b/tests/test_cli/test_wrap_persistent.py
index 4531e5d73..bddfe77f2 100644
--- a/tests/test_cli/test_wrap_persistent.py
+++ b/tests/test_cli/test_wrap_persistent.py
@@ -80,6 +80,7 @@ def test_ensure_proxy_falls_back_when_persistent_manifest_is_stale(monkeypatch)
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
monkeypatch.setattr(wrap_cli, "_recover_persistent_proxy", lambda port: False)
+ monkeypatch.setattr(wrap_cli, "_port_bind_error", lambda port: None)
monkeypatch.setattr(wrap_cli, "_start_proxy", lambda *args, **kwargs: calls.append("start"))
result = wrap_cli._ensure_proxy(8787, False)
@@ -213,6 +214,7 @@ def test_ensure_proxy_restarts_idle_stale_ephemeral_proxy(monkeypatch) -> None:
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: None)
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: len(calls) == 0)
monkeypatch.setattr(wrap_cli, "_query_proxy_health", lambda port: health)
+ monkeypatch.setattr(wrap_cli, "_port_bind_error", lambda port: None)
monkeypatch.setattr(
wrap_cli,
"_kill_proxy_by_pid",
@@ -248,6 +250,7 @@ def test_ensure_proxy_restarts_ephemeral_proxy_for_openai_api_url_mismatch(monke
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: None)
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: len(calls) == 0)
monkeypatch.setattr(wrap_cli, "_query_proxy_health", lambda port: health)
+ monkeypatch.setattr(wrap_cli, "_port_bind_error", lambda port: None)
monkeypatch.setattr(
wrap_cli,
"_kill_proxy_by_pid",
@@ -271,6 +274,81 @@ def test_ensure_proxy_restarts_ephemeral_proxy_for_openai_api_url_mismatch(monke
assert calls[1][2]["openai_api_url"] == "https://api.individual.githubcopilot.com"
+def test_ensure_proxy_restarts_agent_proxy_without_savings_profile(monkeypatch) -> None:
+ calls: list[object] = []
+ health = {
+ "version": wrap_cli._HEADROOM_VERSION,
+ "runtime": {"websocket_sessions": {"active_sessions": 0, "active_relay_tasks": 0}},
+ "config": {"pid": "12345", "memory": False, "learn": False, "code_graph": False},
+ }
+
+ monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: None)
+ monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: len(calls) == 0)
+ monkeypatch.setattr(wrap_cli, "_query_proxy_health", lambda port: health)
+ monkeypatch.setattr(wrap_cli, "_port_bind_error", lambda port: None)
+ monkeypatch.setattr(
+ wrap_cli,
+ "_kill_proxy_by_pid",
+ lambda pid, port: calls.append(("kill", pid, port)) or True,
+ )
+ monkeypatch.setattr(
+ wrap_cli,
+ "_start_proxy",
+ lambda *args, **kwargs: calls.append(("start", args, kwargs)),
+ )
+
+ result = wrap_cli._ensure_proxy(8787, False, agent_type="codex")
+
+ assert result is None
+ assert calls[0] == ("kill", 12345, 8787)
+ assert calls[1][0] == "start"
+
+
+def test_ensure_proxy_reuses_agent_proxy_with_savings_profile(monkeypatch) -> None:
+ health = {
+ "version": wrap_cli._HEADROOM_VERSION,
+ "runtime": {"websocket_sessions": {"active_sessions": 0, "active_relay_tasks": 0}},
+ "config": {
+ "pid": "12345",
+ "memory": False,
+ "learn": False,
+ "code_graph": False,
+ "savings_profile": "agent-90",
+ "target_ratio": 0.10,
+ "compress_user_messages": True,
+ "compress_system_messages": True,
+ "protect_recent": 2,
+ "protect_analysis_context": True,
+ "min_tokens_to_crush": 120,
+ "max_items_after_crush": 8,
+ "smart_crusher_with_compaction": False,
+ "accuracy_guard": "strict",
+ },
+ }
+
+ monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: None)
+ monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: True)
+ monkeypatch.setattr(wrap_cli, "_query_proxy_health", lambda port: health)
+ monkeypatch.setattr(
+ wrap_cli,
+ "_kill_proxy_by_pid",
+ lambda *args, **kwargs: (_ for _ in ()).throw(
+ AssertionError("configured proxy should not restart")
+ ),
+ )
+ monkeypatch.setattr(
+ wrap_cli,
+ "_start_proxy",
+ lambda *args, **kwargs: (_ for _ in ()).throw(
+ AssertionError("replacement proxy should not start")
+ ),
+ )
+
+ result = wrap_cli._ensure_proxy(8787, False, agent_type="cursor")
+
+ assert result is None
+
+
def test_ensure_proxy_leaves_active_stale_ephemeral_proxy_running(monkeypatch) -> None:
health = {
"version": "0.0.1",
diff --git a/tests/test_cli_perf_format.py b/tests/test_cli_perf_format.py
index 6ef9875fb..f2013f63c 100644
--- a/tests/test_cli_perf_format.py
+++ b/tests/test_cli_perf_format.py
@@ -167,12 +167,15 @@ def test_perf_csv_by_model(runner, monkeypatch):
def test_perf_csv_raw_per_record(runner, monkeypatch):
- _patch_report(monkeypatch, _sample_report())
+ report = _sample_report()
+ report.perf_records[0].client = "codex"
+ _patch_report(monkeypatch, report)
result = runner.invoke(main, ["perf", "--format", "csv", "--raw"])
assert result.exit_code == 0, result.output
rows = list(csv.DictReader(io.StringIO(result.output)))
assert len(rows) == 2
assert rows[0]["request_id"] == "hr_1"
+ assert rows[0]["client"] == "codex"
# transforms flattened to a string cell
assert rows[0]["transforms"] == "content_router"
diff --git a/tests/test_proxy_healthchecks.py b/tests/test_proxy_healthchecks.py
index 1d7761b32..3d84d94d9 100644
--- a/tests/test_proxy_healthchecks.py
+++ b/tests/test_proxy_healthchecks.py
@@ -79,9 +79,49 @@ def test_health_preserves_backwards_compatible_config_payload(client):
assert config["memory"] is False
assert config["learn"] is False
assert config["code_graph"] is False
+ assert config["savings_profile"] is None
+ assert config["target_ratio"] is None
+ assert config["max_items_after_crush"] == 50
+ assert config["smart_crusher_with_compaction"] is None
assert isinstance(config["pid"], int)
+def test_health_reports_agent_savings_config():
+ config = ProxyConfig(
+ optimize=False,
+ cache_enabled=False,
+ rate_limit_enabled=False,
+ cost_tracking_enabled=False,
+ savings_profile="agent-90",
+ target_ratio=0.10,
+ compress_user_messages=True,
+ compress_system_messages=True,
+ protect_recent=2,
+ protect_analysis_context=True,
+ min_tokens_to_crush=120,
+ max_items_after_crush=8,
+ smart_crusher_with_compaction=False,
+ accuracy_guard="strict",
+ )
+ app = create_app(config)
+
+ with TestClient(app) as client:
+ response = client.get("/health")
+
+ assert response.status_code == 200
+ reported = response.json()["config"]
+ assert reported["savings_profile"] == "agent-90"
+ assert reported["target_ratio"] == 0.10
+ assert reported["compress_user_messages"] is True
+ assert reported["compress_system_messages"] is True
+ assert reported["protect_recent"] == 2
+ assert reported["protect_analysis_context"] is True
+ assert reported["min_tokens_to_crush"] == 120
+ assert reported["max_items_after_crush"] == 8
+ assert reported["smart_crusher_with_compaction"] is False
+ assert reported["accuracy_guard"] == "strict"
+
+
def test_health_includes_deployment_metadata_when_present(monkeypatch):
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
monkeypatch.setenv("HEADROOM_DEPLOYMENT_PROFILE", "default")
diff --git a/tests/test_transforms/test_content_router.py b/tests/test_transforms/test_content_router.py
index b1547f4b5..f93cf097a 100644
--- a/tests/test_transforms/test_content_router.py
+++ b/tests/test_transforms/test_content_router.py
@@ -8,6 +8,8 @@ Comprehensive tests covering:
- Transform interface: apply(), should_apply() methods
"""
+from types import SimpleNamespace
+
import pytest
from headroom.transforms.content_detector import ContentType
@@ -142,6 +144,53 @@ That's all!
"""
+def test_force_kompress_routes_anthropic_tool_result_to_targeted_kompress(
+ router, tokenizer, monkeypatch
+):
+ captured: dict[str, object] = {}
+
+ class FakeKompress:
+ def compress(self, content, **kwargs):
+ captured.update(kwargs)
+ compressed = " ".join(content.split()[:20])
+ return SimpleNamespace(
+ compressed=compressed,
+ compressed_tokens=len(compressed.split()),
+ )
+
+ monkeypatch.setattr(router, "_get_kompress", lambda: FakeKompress())
+ tool_content = " ".join(
+ f'{{"file":"src/module_{i}.py","line":{i},"text":"repeated search payload"}}'
+ for i in range(160)
+ )
+ messages = [
+ {
+ "role": "user",
+ "content": [
+ {
+ "type": "tool_result",
+ "tool_use_id": "toolu_search_1",
+ "content": tool_content,
+ }
+ ],
+ }
+ ]
+
+ result = router.apply(
+ messages,
+ tokenizer,
+ force_kompress=True,
+ target_ratio=0.10,
+ compress_user_messages=True,
+ min_tokens_to_compress=10,
+ read_protection_window=0,
+ )
+
+ assert result.messages[0]["content"][0]["content"] != tool_content
+ assert result.transforms_applied == ["router:tool_result:kompress"]
+ assert captured["target_ratio"] == 0.10
+
+
# =============================================================================
# TestContentRouterConfig
# =============================================================================
@@ -739,6 +788,47 @@ class TestExcludeTools:
# Verify exclusion was tracked (consistent with OpenAI format)
assert "router:excluded:tool" in result.transforms_applied
+ def test_anthropic_tool_result_runtime_window_allows_old_excluded_tools(self, tokenizer):
+ """Agent profiles can shrink the protected window for Claude tool results."""
+ config = ContentRouterConfig(
+ min_section_tokens=10,
+ min_chars_for_block_compression=10,
+ exclude_tools={"Glob"},
+ )
+ router = ContentRouter(config)
+
+ old_tool_content = generate_search_results(80)
+ messages = [
+ {
+ "role": "assistant",
+ "content": [
+ {
+ "type": "tool_use",
+ "id": "toolu_glob_old",
+ "name": "Glob",
+ "input": {"pattern": "*.py"},
+ }
+ ],
+ },
+ {
+ "role": "user",
+ "content": [
+ {
+ "type": "tool_result",
+ "tool_use_id": "toolu_glob_old",
+ "content": old_tool_content,
+ }
+ ],
+ },
+ {"role": "assistant", "content": "ack"},
+ {"role": "user", "content": "continue"},
+ {"role": "assistant", "content": "ack"},
+ ]
+
+ result = router.apply(messages, tokenizer, read_protection_window=2)
+
+ assert "router:excluded:tool" not in result.transforms_applied
+
def test_mixed_excluded_and_non_excluded_tools(self, tokenizer):
"""Multiple tools in same conversation - only excluded ones pass through."""
config = ContentRouterConfig(