diff --git a/CHANGELOG.md b/CHANGELOG.md index 86f85f070..8f3cac956 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -111,6 +111,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * **transforms:** first-class C# support in `CodeAwareCompressor` via the tree-sitter `csharp` grammar already shipped in the pinned `tree-sitter-language-pack` — no new dependencies ([#1664](https://github.com/headroomlabs-ai/headroom/issues/1664)). Parity with Java/C++/Rust: signatures preserved verbatim, method/constructor/destructor/operator/local-function bodies compressed; block-scoped and file-scoped namespaces, records, structs, interfaces, and enums handled; C#-distinctive auto-detection. Preprocessor conditionals (`#if`…`#endif`) are preserved verbatim as opaque regions (blocks wrapping only `using` directives stay with the imports), `#region` markers no longer swallow the following line during class-member extraction, and top-of-file license banners / `#region License` headers stay on top instead of being relocated below the code. Real-repo runs: 16.1% tokens saved on Newtonsoft.Json (945 files), 37.8% on Polly (797 files), output syntax-valid for 1742/1742 files. * **proxy:** add provider-only HTTP proxy routing via `--http-proxy` and `HEADROOM_HTTP_PROXY`. Upstream LLM provider calls can now use an HTTP proxy without setting process-wide `HTTP_PROXY`/`HTTPS_PROXY` variables that are inherited by tool executions; proxied provider clients use HTTP/1.1 so HTTPS provider APIs can tunnel through CONNECT. * **proxy:** add output shaping for OpenAI Responses traffic on `/v1/responses` HTTP requests and Codex WebSocket `response.create` frames, with stable output-savings holdout keys and counted WS token strata for the experiment. +* **stats:** per-bucket output-shaping savings in `/stats-history`. Each `series` bucket (hourly/daily/weekly/monthly) now carries `output_tokens_saved_delta` and `output_savings_usd_delta` alongside the existing compression deltas, sourced from a per-request synthetic-control estimate (`SavingsRecorder.estimate_request_savings`) threaded through `record_request` into the rollup. Lets dashboards chart output-shaping savings over time as a distinct series — previously it existed only as a single global aggregate. Additive and backward-compatible: pre-feature checkpoints default the new fields to 0 ([#1816](https://github.com/headroomlabs-ai/headroom/issues/1816)). * **observability:** the `headroom.compression.pipeline` span now also carries the OpenTelemetry GenAI semantic-convention attribute `gen_ai.request.model` alongside the existing `headroom.*` attributes, so Headroom's traces group and filter by the standard `gen_ai.*` schema in any OTel-native backend (Grafana, Datadog, etc.). Purely additive; no existing attribute changed. `gen_ai.operation.name`, `gen_ai.provider.name`, and `gen_ai.usage.*` are deliberately deferred (they need per-caller operation threading, reliable upstream-provider resolution, and response-path usage respectively). * **wrap:** `headroom wrap claude --1m` preserves the 1M context window. Behind a custom `ANTHROPIC_BASE_URL` (the proxy) Claude Code drops the `context-1m` beta header and caps the window at 200k for entitled subscription users; the opt-in flag sets `ANTHROPIC_MODEL=[1m]` on the launched process so the 1M window activates through Headroom. A model already selected via `ANTHROPIC_MODEL` is preserved (only the `[1m]` suffix is appended) ([#1158](https://github.com/chopratejas/headroom/issues/1158)). * **learn:** weight loops in `headroom learn`. A new loop detector (`headroom/learn/loops.py`) recognizes repeated tool-call patterns — including RTK re-fetch loops, where RTK's output truncation makes the agent re-run larger-limit variants of a *successful* command — collapses output-limit variants to one signature, measures the wasted tokens, surfaces loops as a highest-priority digest section, and weights loop guardrails above one-off rules by their measured waste. Previously loops had no special weight and a no-failure re-fetch loop was skipped entirely. Adds an RTK-loop eval (`benchmarks/rtk_loop_learn_eval.py`) that reproduces a loop, runs it through Learn, and asserts the generated guardrail ranks first and prevents re-triggering. diff --git a/headroom/proxy/outcome.py b/headroom/proxy/outcome.py index 750a2c0de..311a365c9 100644 --- a/headroom/proxy/outcome.py +++ b/headroom/proxy/outcome.py @@ -357,11 +357,16 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None: # tags each request's (arm, stratum) onto ``transforms_applied``; feed the # observed output tokens to the recorder so it can produce an honest # reduction estimate. Best-effort: never let bookkeeping break a response. + output_tokens_saved_est = 0 if any(str(t).startswith("output_shaper:") for t in outcome.transforms_applied): try: from headroom.proxy.output_savings import get_recorder - get_recorder().record_from_labels(outcome.transforms_applied, outcome.output_tokens) + _rec = get_recorder() + _rec.record_from_labels(outcome.transforms_applied, outcome.output_tokens) + output_tokens_saved_est = _rec.estimate_request_savings( + outcome.transforms_applied, outcome.output_tokens + ) except Exception: # pragma: no cover - defensive pass @@ -388,6 +393,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None: cache_write_1h_tokens=outcome.cache_write_1h_tokens, uncached_input_tokens=outcome.uncached_input_tokens, attempted_input_tokens=outcome.attempted_input_tokens, + output_tokens_saved=output_tokens_saved_est, project=project, client=outcome.client, ) diff --git a/headroom/proxy/output_savings.py b/headroom/proxy/output_savings.py index 0d2bd2c97..b0ee6baf3 100644 --- a/headroom/proxy/output_savings.py +++ b/headroom/proxy/output_savings.py @@ -387,6 +387,26 @@ class SavingsRecorder: return True return False + def estimate_request_savings(self, labels: Any, output_tokens: int) -> int: + """Per-request output tokens saved, for the savings rollup. + + For a treatment request, the synthetic-control estimate + ``max(0, baseline_mean(stratum) - output_tokens)``; 0 for control, + unknown strata, or when no shaping label is present. Read-only: + unlike ``record_from_labels`` it does not mutate the ledger, so the + two compose without double-counting.""" + for label in labels or (): + parsed = parse_stratum_label(str(label)) + if parsed is None: + continue + arm, key = parsed + if arm != "treatment": + return 0 + with self._lock: + mean, _var, n = self._ledger.baseline.lookup(key) + return max(0, int(round(mean - output_tokens))) if n > 0 else 0 + return 0 + def _reload_baseline_locked(self) -> None: """Adopt the on-disk baseline written by ``learn --verbosity --apply``. diff --git a/headroom/proxy/prometheus_metrics.py b/headroom/proxy/prometheus_metrics.py index 1aa704b2e..94f005fe3 100644 --- a/headroom/proxy/prometheus_metrics.py +++ b/headroom/proxy/prometheus_metrics.py @@ -637,6 +637,7 @@ class PrometheusMetrics: cache_write_1h_tokens: int = 0, uncached_input_tokens: int = 0, attempted_input_tokens: int = 0, + output_tokens_saved: int = 0, project: str | None = None, client: str | None = None, ): @@ -763,6 +764,7 @@ class PrometheusMetrics: uncached_input_tokens=uncached_input_tokens, total_input_tokens=total_input_tokens, total_input_cost_usd=total_input_cost_usd, + output_tokens_saved=output_tokens_saved, ) # Also append to the durable, multi-process savings ledger so diff --git a/headroom/proxy/savings_tracker.py b/headroom/proxy/savings_tracker.py index 85698d5e5..55afa0744 100644 --- a/headroom/proxy/savings_tracker.py +++ b/headroom/proxy/savings_tracker.py @@ -39,6 +39,8 @@ DEFAULT_MAX_HISTORY_AGE_DAYS = 365 DEFAULT_MAX_RESPONSE_HISTORY_POINTS = 500 DEFAULT_DISPLAY_SESSION_INACTIVITY_MINUTES = 60 DEFAULT_FALLBACK_INPUT_COST_PER_TOKEN = 3.0 / 1_000_000 +# Blended output price used only when litellm cannot price the model. +DEFAULT_FALLBACK_OUTPUT_COST_PER_TOKEN = 15.0 / 1_000_000 LITELLM_AVAILABLE = importlib.util.find_spec("litellm") is not None litellm: Any | None = None @@ -221,6 +223,28 @@ def _estimate_compression_savings_usd(model: str, tokens_saved: int) -> float: return float(tokens_saved) * float(DEFAULT_FALLBACK_INPUT_COST_PER_TOKEN) +def _estimate_output_savings_usd(model: str, tokens_saved: int) -> float: + """Estimate output-shaping savings in USD from saved *output* tokens. + + Mirrors ``_estimate_compression_savings_usd`` but prices at the model's + output rate, since the shaper reduces generated (output) tokens, not input. + """ + litellm = _get_litellm_module() + if tokens_saved <= 0: + return 0.0 + if litellm is None: + return float(tokens_saved) * float(DEFAULT_FALLBACK_OUTPUT_COST_PER_TOKEN) + try: + resolved = _resolve_litellm_model(model) + info = litellm.model_cost.get(resolved, {}) + output_cost_per_token = info.get("output_cost_per_token") + if not output_cost_per_token: + raise RuntimeError("output cost unavailable") + return float(tokens_saved) * float(output_cost_per_token) + except Exception: + return float(tokens_saved) * float(DEFAULT_FALLBACK_OUTPUT_COST_PER_TOKEN) + + def _estimate_cache_savings_usd(model: str, cache_read_tokens: int) -> float: """Estimate cache-read savings in USD — the discount delta vs list price. @@ -329,6 +353,8 @@ def _normalize_history_entry(entry: Any) -> dict[str, Any] | None: cache_savings_usd = 0.0 total_input_tokens = 0 total_input_cost_usd = 0.0 + output_tokens_saved = 0 + output_savings_usd = 0.0 provider = PROVIDER_UNKNOWN model = MODEL_UNKNOWN @@ -343,6 +369,8 @@ def _normalize_history_entry(entry: Any) -> dict[str, Any] | None: cache_savings_usd = _coerce_float(entry.get("cache_savings_usd")) total_input_tokens = _coerce_int(entry.get("total_input_tokens")) total_input_cost_usd = _coerce_float(entry.get("total_input_cost_usd")) + output_tokens_saved = _coerce_int(entry.get("output_tokens_saved")) + output_savings_usd = _coerce_float(entry.get("output_savings_usd")) provider = _normalize_provider(entry.get("provider")) model = _normalize_model(entry.get("model")) elif isinstance(entry, list | tuple) and len(entry) >= 2: @@ -370,6 +398,8 @@ def _normalize_history_entry(entry: Any) -> dict[str, Any] | None: "cache_savings_usd": round(cache_savings_usd, 6), "total_input_tokens": total_input_tokens, "total_input_cost_usd": round(total_input_cost_usd, 6), + "output_tokens_saved": output_tokens_saved, + "output_savings_usd": round(output_savings_usd, 6), } @@ -634,6 +664,7 @@ class SavingsTracker: model: str, input_tokens: int, tokens_saved: int, + output_tokens_saved: int = 0, provider: str | None = None, project: str | None = None, cache_read_tokens: int = 0, @@ -657,6 +688,8 @@ class SavingsTracker: delta_tokens_saved = _coerce_int(tokens_saved) delta_input_tokens = _coerce_int(input_tokens) delta_savings_usd = _estimate_compression_savings_usd(model, delta_tokens_saved) + delta_output_tokens_saved = max(_coerce_int(output_tokens_saved), 0) + delta_output_savings_usd = _estimate_output_savings_usd(model, delta_output_tokens_saved) delta_cache_read_tokens = _coerce_int(cache_read_tokens) delta_cache_savings_usd = _estimate_cache_savings_usd(model, delta_cache_read_tokens) delta_input_cost_usd = _estimate_input_cost_usd( @@ -711,6 +744,13 @@ class SavingsTracker: ) lifetime["total_input_tokens"] = next_total_input_tokens lifetime["total_input_cost_usd"] = next_total_input_cost_usd + lifetime["output_tokens_saved"] = ( + lifetime.get("output_tokens_saved", 0) + delta_output_tokens_saved + ) + lifetime["output_savings_usd"] = round( + lifetime.get("output_savings_usd", 0.0) + delta_output_savings_usd, + 6, + ) session = self._state["display_session"] last_activity = _parse_timestamp(session.get("last_activity_at")) @@ -771,8 +811,12 @@ class SavingsTracker: # not lossy-compressed, to keep Bedrock's prompt cache warm. Gating # on tokens_saved alone silently dropped every history point on # those requests even though real cache-read savings occurred. - # Append whenever either mechanism produced a saving. - if delta_tokens_saved > 0 or delta_cache_read_tokens > 0: + # Append whenever any savings mechanism produced a saving. + if ( + delta_tokens_saved > 0 + or delta_cache_read_tokens > 0 + or delta_output_tokens_saved > 0 + ): self._state["history"].append( { "timestamp": _to_utc_iso(timestamp_dt), @@ -784,6 +828,8 @@ class SavingsTracker: "cache_savings_usd": lifetime["cache_savings_usd"], "total_input_tokens": lifetime["total_input_tokens"], "total_input_cost_usd": lifetime["total_input_cost_usd"], + "output_tokens_saved": lifetime.get("output_tokens_saved", 0), + "output_savings_usd": lifetime.get("output_savings_usd", 0.0), } ) self._trim_history_locked(reference_time=timestamp_dt) @@ -1059,6 +1105,8 @@ class SavingsTracker: "total_input_tokens", "total_input_cost_usd_delta", "total_input_cost_usd", + "output_tokens_saved_delta", + "output_savings_usd_delta", ] buffer = StringIO() @@ -1478,6 +1526,8 @@ class SavingsTracker: prev_total_usd = 0.0 prev_total_input_tokens = 0 prev_total_input_cost_usd = 0.0 + prev_output_tokens = 0 + prev_output_usd = 0.0 for point in history: timestamp = _parse_timestamp(point["timestamp"]) @@ -1491,6 +1541,8 @@ class SavingsTracker: total_usd = _coerce_float(point.get("compression_savings_usd")) total_input_tokens = _coerce_int(point.get("total_input_tokens")) total_input_cost_usd = _coerce_float(point.get("total_input_cost_usd")) + total_output_tokens = _coerce_int(point.get("output_tokens_saved")) + total_output_usd = _coerce_float(point.get("output_savings_usd")) delta_tokens = max(total_tokens_saved - prev_total_tokens, 0) delta_usd = max(total_usd - prev_total_usd, 0.0) delta_input_tokens = max(total_input_tokens - prev_total_input_tokens, 0) @@ -1499,10 +1551,15 @@ class SavingsTracker: 0.0, ) + delta_output_tokens = max(total_output_tokens - prev_output_tokens, 0) + delta_output_usd = max(total_output_usd - prev_output_usd, 0.0) + prev_total_tokens = total_tokens_saved prev_total_usd = total_usd prev_total_input_tokens = total_input_tokens prev_total_input_cost_usd = total_input_cost_usd + prev_output_tokens = total_output_tokens + prev_output_usd = total_output_usd entry = aggregated.setdefault( bucket_key, @@ -1516,6 +1573,8 @@ class SavingsTracker: "total_input_tokens": total_input_tokens, "total_input_cost_usd_delta": 0.0, "total_input_cost_usd": total_input_cost_usd, + "output_tokens_saved_delta": 0, + "output_savings_usd_delta": 0.0, "by_provider": {}, "by_model": {}, }, @@ -1534,6 +1593,11 @@ class SavingsTracker: entry["compression_savings_usd"] = round(total_usd, 6) entry["total_input_tokens"] = total_input_tokens entry["total_input_cost_usd"] = round(total_input_cost_usd, 6) + entry["output_tokens_saved_delta"] += delta_output_tokens + entry["output_savings_usd_delta"] = round( + entry["output_savings_usd_delta"] + delta_output_usd, + 6, + ) # Attribute this checkpoint's delta to the provider that produced # it. Each checkpoint comes from a single request, so its delta is diff --git a/tests/test_output_shaping_rollup.py b/tests/test_output_shaping_rollup.py new file mode 100644 index 000000000..e9dea2f57 --- /dev/null +++ b/tests/test_output_shaping_rollup.py @@ -0,0 +1,88 @@ +"""Per-bucket output-shaping savings in the /stats-history rollup. + +Covers the feature that lets a downstream dashboard stack output-shaping +savings as a distinct daily segment: SavingsTracker.record_request accepts a +per-request output_tokens_saved, accumulates it into each time bucket as +output_tokens_saved_delta / output_savings_usd_delta, and the read-only +SavingsRecorder.estimate_request_savings supplies that per-request number. +""" + +from __future__ import annotations + +from headroom.proxy.output_savings import ( + SavingsRecorder, + stratum_key, + stratum_label, +) +from headroom.proxy.savings_tracker import SavingsTracker + + +def test_record_request_buckets_output_shaping_savings(tmp_path): + tracker = SavingsTracker(path=str(tmp_path / "s.json")) + + # Request with both compression and output-shaping savings. + tracker.record_request( + model="claude-opus-4-8", + input_tokens=1000, + tokens_saved=100, + output_tokens_saved=5000, + timestamp="2026-03-27T09:00:00Z", + ) + # Output-shaping-ONLY request (no compression) must still checkpoint, else + # its output savings would be dropped from the rollup. + tracker.record_request( + model="claude-opus-4-8", + input_tokens=1000, + tokens_saved=0, + output_tokens_saved=3000, + timestamp="2026-03-27T09:30:00Z", + ) + + daily = tracker.history_response()["series"]["daily"] + assert len(daily) == 1 + assert daily[0]["output_tokens_saved_delta"] == 8000 + assert daily[0]["output_savings_usd_delta"] > 0.0 + # Compression axis stays independent. + assert daily[0]["tokens_saved"] == 100 + + +def test_record_request_without_output_savings_is_backward_compatible(tmp_path): + tracker = SavingsTracker(path=str(tmp_path / "s.json")) + tracker.record_request( + model="gpt-4o", + input_tokens=8192, + tokens_saved=4096, + timestamp="2026-03-27T09:00:00Z", + ) + daily = tracker.history_response()["series"]["daily"] + assert daily[0]["output_tokens_saved_delta"] == 0 + assert daily[0]["output_savings_usd_delta"] == 0.0 + + +def _key() -> str: + return stratum_key(turn_kind="code", input_tokens=8000, model="claude-opus-4-8", has_tools=True) + + +def test_estimate_request_savings_treatment_uses_baseline(tmp_path): + rec = SavingsRecorder(str(tmp_path / "o.json"), flush_every=1) + key = _key() + for _ in range(5): + rec._ledger.baseline.observe(key, 1000) # baseline mean ~1000 + + # Treatment request that emitted 600 -> saved ~400 vs the baseline. + saved = rec.estimate_request_savings([stratum_label("treatment", key)], 600) + assert saved == 400 + + +def test_estimate_request_savings_zero_for_control_and_unknown(tmp_path): + rec = SavingsRecorder(str(tmp_path / "o.json"), flush_every=1) + key = _key() + for _ in range(5): + rec._ledger.baseline.observe(key, 1000) + + # Control arm is unshaped -> no attributable saving. + assert rec.estimate_request_savings([stratum_label("control", key)], 600) == 0 + # No shaping label at all. + assert rec.estimate_request_savings(["something-else"], 600) == 0 + # Treatment but output exceeded the baseline -> clamped to 0, never negative. + assert rec.estimate_request_savings([stratum_label("treatment", key)], 5000) == 0 diff --git a/tests/test_proxy_savings_history.py b/tests/test_proxy_savings_history.py index 64db13086..a7e4acec7 100644 --- a/tests/test_proxy_savings_history.py +++ b/tests/test_proxy_savings_history.py @@ -82,6 +82,8 @@ def test_savings_tracker_helpers_normalize_inputs_and_paths(tmp_path, monkeypatc "cache_savings_usd": 0.0, "total_input_tokens": 0, "total_input_cost_usd": 0.0, + "output_tokens_saved": 0, + "output_savings_usd": 0.0, } assert savings_tracker_module._normalize_history_entry({"timestamp": "bad"}) is None assert savings_tracker_module._normalize_history_entry(object()) is None @@ -145,6 +147,8 @@ def test_savings_tracker_sanitizes_legacy_state_and_applies_retention(tmp_path): "cache_savings_usd": 0.0, "total_input_tokens": 0, "total_input_cost_usd": 0.0, + "output_tokens_saved": 0, + "output_savings_usd": 0.0, } ] assert snapshot["retention"] == { @@ -1288,7 +1292,8 @@ def test_stats_history_csv_export_is_frontend_friendly(tmp_path, monkeypatch): assert lines[0] == ( "timestamp,tokens_saved,compression_savings_usd_delta,total_tokens_saved," "compression_savings_usd,total_input_tokens_delta,total_input_tokens," - "total_input_cost_usd_delta,total_input_cost_usd" + "total_input_cost_usd_delta,total_input_cost_usd," + "output_tokens_saved_delta,output_savings_usd_delta" ) assert len(lines) >= 2 assert "total_tokens_saved" in lines[0]