fix(policy): price net-cost mutations with the 1h cache-write tier (#2780)

## Description

This fixes the net-cost mutation gate for requests using Anthropic's
1-hour prompt-cache TTL.

The gate previously hardcoded the 5-minute cache-write multiplier of
1.25x. A 1-hour cache write costs 2.0x, so the old calculation
understated the true write penalty and could incorrectly recommend
mutation for 1-hour clients.

Closes #2773

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Added TTL-aware cache-write multiplier selection for 5-minute and
1-hour tiers.
- Threaded the resolved TTL through the content router and compression
policy helpers.
- Preserved the existing 5-minute behavior as the default.
- Added Python and Rust regression coverage for the 1-hour tier.
- Retuned the netcost gate fixtures so the 1-hour write tier flips the
decision in the full ContentRouter path.
- Did not edit CHANGELOG.md.

## Testing

- [x] Unit tests pass (pytest)
- [x] Linting passes (ruff check .)
- [ ] Type checking passes (mypy headroom)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
pytest tests/test_compression_policy.py -q
20 passed

cargo test -p headroom-core --lib compression_policy -- --nocapture
14 passed

pytest tests/test_netcost_gate.py -q
27 passed

Ruff checks and formatting passed.
git diff --check passed.
```

## Real Behavior Proof

- Environment: Linux x86_64 contributor checkout with Python and Rust
test environments.
- Exact command / steps:
  - Ran the Python compression policy test suite.
  - Ran the Rust compression policy unit tests.
- Ran the netcost gate suite, including the 1-hour env and
request-marker cases.
- Exercised the new 1-hour TTL golden case alongside the existing
5-minute cases.
- Observed result: The 1-hour case uses the 2.0x write multiplier and
skips the same candidate that still mutates under 5-minute pricing.
Existing 5-minute behavior remains covered and passing.
- Not tested: A live Anthropic request through the proxy and production
traffic.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I did **not** edit CHANGELOG.md - it is generated by
release-please from my Conventional Commit PR title (a CI guard enforces
this)

## Screenshots (if applicable)

Not applicable for this backend policy fix.

## Additional Notes

Ready for review. CI is green on the current tip.
This commit is contained in:
AxelRay
2026-08-17 05:09:50 +07:00
committed by GitHub
parent 2a8472525d
commit ef7e07e0f5
6 changed files with 262 additions and 30 deletions
+94 -3
View File
@@ -135,11 +135,24 @@ pub(crate) const MAX_LOSSY_RATIO_SUBSCRIPTION: f32 = 0.25;
/// net-cost mutation formula (#856).
pub const CACHE_WRITE_MULTIPLIER: f32 = 1.25;
/// Anthropic prompt-cache write multiplier for the 1-hour TTL tier.
pub const CACHE_WRITE_MULTIPLIER_1H: f32 = 2.0;
/// Anthropic prompt-cache read multiplier: a `cache_read` token costs
/// 0.1× a plain input token. Input to the net-cost mutation formula
/// (#856).
pub const CACHE_READ_MULTIPLIER: f32 = 0.1;
/// Return the cache-write multiplier for a prompt-cache TTL tier.
///
/// Invalid, missing, and non-positive values retain the 5-minute default.
pub fn cache_write_multiplier_for_ttl(ttl_seconds: Option<f32>) -> f32 {
match ttl_seconds {
Some(ttl) if ttl.is_finite() && ttl >= 3_600.0 => CACHE_WRITE_MULTIPLIER_1H,
_ => CACHE_WRITE_MULTIPLIER,
}
}
/// Per-auth-mode policy that downstream compression stages consult.
///
/// `Copy` because the struct is small POD (two `bool`s + a `u32` + an
@@ -269,7 +282,26 @@ impl CompressionPolicy {
expected_reads: f32,
p_alive: f32,
) -> f32 {
let w = CACHE_WRITE_MULTIPLIER;
self.net_mutation_gain_with_write_multiplier(
delta_t,
suffix_tokens,
expected_reads,
p_alive,
None,
)
}
/// Variant of [`Self::net_mutation_gain`] with an explicit cache-write
/// multiplier. `None` uses the 5-minute default.
pub fn net_mutation_gain_with_write_multiplier(
&self,
delta_t: u32,
suffix_tokens: u32,
expected_reads: f32,
p_alive: f32,
write_multiplier: Option<f32>,
) -> f32 {
let w = write_multiplier.unwrap_or(CACHE_WRITE_MULTIPLIER);
let r = CACHE_READ_MULTIPLIER;
// f32::max ignores NaN (returns the other operand), so NaN reads
// land on 0.0; clamp would propagate NaN, so guard alive explicitly.
@@ -299,7 +331,32 @@ impl CompressionPolicy {
expected_reads: f32,
p_alive: f32,
) -> bool {
self.net_mutation_gain(delta_t, suffix_tokens, expected_reads, p_alive) > 0.0
self.should_mutate_deep_with_write_multiplier(
delta_t,
suffix_tokens,
expected_reads,
p_alive,
None,
)
}
/// Variant of [`Self::should_mutate_deep`] with an explicit cache-write
/// multiplier. `None` uses the 5-minute default.
pub fn should_mutate_deep_with_write_multiplier(
&self,
delta_t: u32,
suffix_tokens: u32,
expected_reads: f32,
p_alive: f32,
write_multiplier: Option<f32>,
) -> bool {
self.net_mutation_gain_with_write_multiplier(
delta_t,
suffix_tokens,
expected_reads,
p_alive,
write_multiplier,
) > 0.0
}
/// Remaining-read count at which a warm-cache (P_alive = 1)
@@ -315,10 +372,21 @@ impl CompressionPolicy {
/// session lasts N more turns"). Returns 0 when `delta_t` is 0
/// (no savings — callers gate on `delta_t > 0`).
pub fn break_even_reads(&self, delta_t: u32, suffix_tokens: u32) -> f32 {
self.break_even_reads_with_write_multiplier(delta_t, suffix_tokens, None)
}
/// Variant of [`Self::break_even_reads`] with an explicit cache-write
/// multiplier. `None` uses the 5-minute default.
pub fn break_even_reads_with_write_multiplier(
&self,
delta_t: u32,
suffix_tokens: u32,
write_multiplier: Option<f32>,
) -> f32 {
if delta_t == 0 {
return 0.0;
}
let w = CACHE_WRITE_MULTIPLIER;
let w = write_multiplier.unwrap_or(CACHE_WRITE_MULTIPLIER);
let r = CACHE_READ_MULTIPLIER;
((w - r) / r) * ((suffix_tokens as f32) / (delta_t as f32))
}
@@ -447,6 +515,29 @@ mod tests {
assert!(p.should_mutate_deep(50_000, 10_000, 3.0, 1.0));
}
#[test]
fn net_gain_big_shave_shallow_suffix_is_loss_at_1h_tier() {
let p = CompressionPolicy::for_mode(AuthMode::Payg);
let default_gain = p.net_mutation_gain(50_000, 10_000, 3.0, 1.0);
assert!(default_gain > 0.0, "default gain = {default_gain}");
let gain = p.net_mutation_gain_with_write_multiplier(
50_000,
10_000,
3.0,
1.0,
Some(CACHE_WRITE_MULTIPLIER_1H),
);
assert!((gain - (-4_000.0)).abs() < 1.0, "gain = {gain}");
assert!(!p.should_mutate_deep_with_write_multiplier(
50_000,
10_000,
3.0,
1.0,
Some(CACHE_WRITE_MULTIPLIER_1H),
));
}
#[test]
fn net_gain_no_suffix_edit_profitable_with_reads_remaining() {
// S = 0: nothing cached after the edit is invalidated. Warm-case
+20 -13
View File
@@ -1364,24 +1364,20 @@ class AnthropicHandlerMixin:
# lossless whole-prefix recompaction instead of the byte-identical splice
# (the splice preserves a dead cache) and skips the overlay replay. Both are
# deterministic → the recompacted prefix re-caches byte-stable on warm turns.
from headroom.transforms.cold_prefix import (
anthropic_cache_ttl_seconds,
is_cold_prefix,
)
# Resolve the authoritative request-level prompt-cache tier once.
# The same value drives cold-prefix handling and net-cost pricing.
_cc_ttl = anthropic_cache_ttl_seconds(model, original_client_messages, system_prompt)
_cold_recompact_active = False
if os.environ.get("HEADROOM_COLD_RECOMPACT", "").strip().lower() in (
"1",
"true",
"yes",
):
from headroom.transforms.cold_prefix import (
anthropic_cache_ttl_seconds,
is_cold_prefix,
)
# Read CC's ACTUAL prompt-cache TTL (request cache_control.ttl + the
# DISABLE_/ENABLE_/FORCE_PROMPT_CACHING_* env controls) instead of the
# static 300s guess — a wrong TTL is exactly what busts a warm cache.
# None ⇒ caching is OFF (no cache to bust) ⇒ recompact every turn.
_cc_ttl = anthropic_cache_ttl_seconds(
model, original_client_messages, system_prompt
)
_cold_recompact_active = _cc_ttl is None or is_cold_prefix(
prefix_tracker, ttl_seconds=_cc_ttl
)
@@ -1580,6 +1576,7 @@ class AnthropicHandlerMixin:
biases=biases,
request_id=request_id,
compression_policy=compression_policy,
cache_ttl_seconds=_cc_ttl,
**proxy_pipeline_kwargs(self.config),
),
lambda bg_result: comp_cache.update_from_result(
@@ -1624,6 +1621,7 @@ class AnthropicHandlerMixin:
biases=biases,
request_id=request_id,
compression_policy=compression_policy,
cache_ttl_seconds=_cc_ttl,
skip_kompress=True,
**proxy_pipeline_kwargs(self.config),
),
@@ -1674,6 +1672,7 @@ class AnthropicHandlerMixin:
biases=biases,
request_id=request_id,
compression_policy=compression_policy,
cache_ttl_seconds=_cc_ttl,
**proxy_pipeline_kwargs(self.config),
),
timeout=COMPRESSION_TIMEOUT_SECONDS,
@@ -1715,6 +1714,7 @@ class AnthropicHandlerMixin:
biases=biases,
request_id=request_id,
compression_policy=compression_policy,
cache_ttl_seconds=_cc_ttl,
**proxy_pipeline_kwargs(self.config),
),
timeout=COMPRESSION_TIMEOUT_SECONDS,
@@ -1847,6 +1847,7 @@ class AnthropicHandlerMixin:
biases=biases,
request_id=request_id,
compression_policy=compression_policy,
cache_ttl_seconds=_cc_ttl,
**proxy_pipeline_kwargs(self.config),
),
timeout=COMPRESSION_TIMEOUT_SECONDS,
@@ -4544,6 +4545,8 @@ class AnthropicHandlerMixin:
compressed_requests = []
pipeline_timing: dict[str, float] = {}
from headroom.transforms.cold_prefix import anthropic_cache_ttl_seconds
# Apply compression to each request in the batch
for batch_req in requests_list:
custom_id = batch_req.get("custom_id", "")
@@ -4553,6 +4556,9 @@ class AnthropicHandlerMixin:
messages = params.get("messages", [])
original_messages = copy.deepcopy(messages)
model = params.get("model", "unknown")
cache_ttl_seconds = anthropic_cache_ttl_seconds(
model, original_messages, params.get("system")
)
if not messages or not self.config.optimize:
# No messages or optimization disabled - pass through unchanged
@@ -4589,7 +4595,7 @@ class AnthropicHandlerMixin:
# blocks every other request for the duration; a timeout
# here is caught below and passes the item through.
result = await self._run_compression_in_executor(
lambda messages=messages, model=model, context_limit=context_limit, frozen_message_count=frozen_message_count: (
lambda messages=messages, model=model, context_limit=context_limit, frozen_message_count=frozen_message_count, cache_ttl_seconds=cache_ttl_seconds: (
self.anthropic_pipeline.apply(
messages=messages,
model=model,
@@ -4597,6 +4603,7 @@ class AnthropicHandlerMixin:
context=extract_user_query(messages),
frozen_message_count=frozen_message_count,
request_id=request_id,
cache_ttl_seconds=cache_ttl_seconds,
**proxy_pipeline_kwargs(self.config),
)
),
+44 -11
View File
@@ -58,21 +58,35 @@ _MAX_LOSSY_RATIO_SUBSCRIPTION: float = 0.25
#: Anthropic prompt-cache write multiplier: a ``cache_creation`` token
#: costs 1.25x a plain input token (5-minute TTL tier). Input to the
#: net-cost mutation formula (#856). Mirrors the Rust ``pub const``.
#: ponytail: hardcoded to the 5m tier. A client on Anthropic's 1h cache
#: (ENABLE_PROMPT_CACHING_1H / cache_control.ttl="1h", which Headroom
#: preserves) writes at 2.0x, so its mutations are gated with a ~40%
#: under-stated write penalty. Harmless while the net-cost gate stays
#: default-off (HEADROOM_NET_COST_POLICY); thread the TTL from
#: cold_prefix.anthropic_cache_ttl_seconds through ContentRouter ->
#: net_mutation_gain if that gate is ever turned on.
CACHE_WRITE_MULTIPLIER: float = 1.25
#: Anthropic prompt-cache write multiplier for the 1-hour TTL tier.
CACHE_WRITE_MULTIPLIER_1H: float = 2.0
#: Anthropic prompt-cache read multiplier: a ``cache_read`` token costs
#: 0.1x a plain input token. Input to the net-cost mutation formula
#: (#856). Mirrors the Rust ``pub const``.
CACHE_READ_MULTIPLIER: float = 0.1
def cache_write_multiplier_for_ttl(ttl_seconds: float | int | None) -> float:
"""Return the cache-write multiplier for a prompt-cache TTL tier.
The net-cost gate prefers an authoritative request-level TTL and falls
back to its environment setting when no request TTL is available.
Invalid and non-positive values retain the 5-minute default.
"""
if ttl_seconds is None:
return CACHE_WRITE_MULTIPLIER
try:
ttl = float(ttl_seconds)
except (TypeError, ValueError):
return CACHE_WRITE_MULTIPLIER
if not math.isfinite(ttl) or ttl <= 0.0:
return CACHE_WRITE_MULTIPLIER
return CACHE_WRITE_MULTIPLIER_1H if ttl >= 3600.0 else CACHE_WRITE_MULTIPLIER
@dataclass(frozen=True, slots=True)
class CompressionPolicy:
"""Per-auth-mode policy that downstream compression stages consult.
@@ -136,6 +150,8 @@ class CompressionPolicy:
suffix_tokens: int,
expected_reads: float,
p_alive: float,
*,
write_multiplier: float | None = None,
) -> float:
"""Net gain (in plain-input-token cost units) of a mutation that
removes ``delta_t`` tokens from a message whose cached suffix is
@@ -156,7 +172,7 @@ class CompressionPolicy:
``>= 0`` (NaN → 0), ``p_alive`` to ``[0, 1]`` (NaN → 1, the
conservative full-penalty assumption — same as Rust).
"""
w = CACHE_WRITE_MULTIPLIER
w = CACHE_WRITE_MULTIPLIER if write_multiplier is None else write_multiplier
r = CACHE_READ_MULTIPLIER
dt = max(0, delta_t)
suffix = max(0, suffix_tokens)
@@ -172,12 +188,29 @@ class CompressionPolicy:
suffix_tokens: int,
expected_reads: float,
p_alive: float,
*,
write_multiplier: float | None = None,
) -> bool:
"""Decision form of :meth:`net_mutation_gain`: mutate iff the
gain is strictly positive."""
return self.net_mutation_gain(delta_t, suffix_tokens, expected_reads, p_alive) > 0.0
return (
self.net_mutation_gain(
delta_t,
suffix_tokens,
expected_reads,
p_alive,
write_multiplier=write_multiplier,
)
> 0.0
)
def break_even_reads(self, delta_t: int, suffix_tokens: int) -> float:
def break_even_reads(
self,
delta_t: int,
suffix_tokens: int,
*,
write_multiplier: float | None = None,
) -> float:
"""Remaining-read count at which a warm-cache (``p_alive=1``)
mutation breaks even::
@@ -192,7 +225,7 @@ class CompressionPolicy:
"""
if delta_t <= 0:
return 0.0
w = CACHE_WRITE_MULTIPLIER
w = CACHE_WRITE_MULTIPLIER if write_multiplier is None else write_multiplier
r = CACHE_READ_MULTIPLIER
return ((w - r) / r) * (float(max(0, suffix_tokens)) / float(delta_t))
+29 -3
View File
@@ -65,6 +65,7 @@ from ..tokenizers.base import count_content_blocks
from ..tokenizers.estimator import EstimatingTokenCounter
from . import mixed_content as _mixed_content
from .base import Transform
from .compression_policy import cache_write_multiplier_for_ttl
from .compressor_registry import (
CompressInput,
CompressorDescriptor,
@@ -4529,6 +4530,7 @@ class ContentRouter(Transform):
transforms_applied: list[str],
batch_state: dict[str, int | None] | None = None,
p_alive_override: float | None = None,
write_multiplier: float | None = None,
) -> bool:
"""Break-even gate for one candidate mutation (#856 P2, flag-gated).
@@ -4613,7 +4615,15 @@ class ContentRouter(Transform):
p_alive = _p_alive
except ValueError:
logger.warning("HEADROOM_NET_COST_P_ALIVE malformed; using 1.0")
gain = float(policy.net_mutation_gain(delta_t, suffix, reads, p_alive))
gain = float(
policy.net_mutation_gain(
delta_t,
suffix,
reads,
p_alive,
write_multiplier=write_multiplier,
)
)
allowed = gain > 0.0
logger.info(
"NetCostPolicy slot=%d delta_t=%d suffix=%d reads=%.1f p_alive=%.2f "
@@ -4976,7 +4986,22 @@ class ContentRouter(Transform):
# env-constant behaviour. Derived once here (not per slot) — idle is a
# per-request property, like frozen_message_count.
netcost_p_alive_override: float | None = None
netcost_write_multiplier: float | None = None
if netcost_enabled:
# Prefer the authoritative per-request prompt-cache TTL when the
# caller has one; retain the env setting for other providers and
# legacy callers.
request_ttl = kwargs.get("cache_ttl_seconds")
if request_ttl is None:
netcost_ttl = _net_cost_cache_ttl_seconds()
else:
try:
netcost_ttl = float(request_ttl)
except (TypeError, ValueError):
netcost_ttl = _net_cost_cache_ttl_seconds()
if not math.isfinite(netcost_ttl) or netcost_ttl <= 0.0:
netcost_ttl = _net_cost_cache_ttl_seconds()
netcost_write_multiplier = cache_write_multiplier_for_ttl(netcost_ttl)
netcost_suffix_tokens = [0] * (num_messages + 1)
for j in range(num_messages - 1, -1, -1):
netcost_suffix_tokens[j] = netcost_suffix_tokens[j + 1] + _netcost_message_tokens(
@@ -4989,8 +5014,7 @@ class ContentRouter(Transform):
except (TypeError, ValueError):
idle_f = None
if idle_f is not None and math.isfinite(idle_f) and idle_f >= 0.0:
ttl = _net_cost_cache_ttl_seconds()
netcost_p_alive_override = max(0.0, 1.0 - idle_f / ttl)
netcost_p_alive_override = max(0.0, 1.0 - idle_f / netcost_ttl)
# Tasks: list of (slot_index, content, context, bias, content_key)
_PendingTask = tuple[int, str, str, float, int, bool]
@@ -5294,6 +5318,7 @@ class ContentRouter(Transform):
transforms_applied=transforms_applied,
batch_state=netcost_batch_state,
p_alive_override=netcost_p_alive_override,
write_multiplier=netcost_write_multiplier,
):
# Net-cost gate: mutation would cost more in cache
# invalidation than it saves — leave untouched.
@@ -5471,6 +5496,7 @@ class ContentRouter(Transform):
transforms_applied=transforms_applied,
batch_state=netcost_batch_state,
p_alive_override=netcost_p_alive_override,
write_multiplier=netcost_write_multiplier,
):
result_slots[slot_idx] = message
continue
+21
View File
@@ -19,6 +19,7 @@ import pytest
from headroom.proxy.auth_mode import AuthMode
from headroom.transforms.compression_policy import (
CompressionPolicy,
cache_write_multiplier_for_ttl,
policy_default_payg,
policy_for_mode,
)
@@ -183,6 +184,26 @@ class TestNetCostFormula:
assert abs(gain - 3_500.0) < 1.0
assert p.should_mutate_deep(50_000, 10_000, 3.0, 1.0)
one_hour_gain = p.net_mutation_gain(
50_000,
10_000,
3.0,
1.0,
write_multiplier=2.0,
)
assert abs(one_hour_gain - (-4_000.0)) < 1.0
assert not p.should_mutate_deep(
50_000,
10_000,
3.0,
1.0,
write_multiplier=2.0,
)
def test_cache_write_multiplier_follows_ttl_tier(self):
assert cache_write_multiplier_for_ttl(300) == 1.25
assert cache_write_multiplier_for_ttl(3600) == 2.0
def test_no_suffix_edit_profitable_with_reads_remaining(self):
# S = 0: warm-case saving is the avoided rereads, dT*r*R —
# positive whenever at least one read remains. At R=0 with a
+54
View File
@@ -73,6 +73,60 @@ class TestNetCostGate:
assert _tool_slot_compressed(result, messages)
assert not any(t.startswith("netcost:skip:") for t in result.transforms_applied)
def test_one_hour_ttl_prices_write_tier(self, router, tokenizer, monkeypatch):
# 5-minute pricing still admits this shave, while the larger 1h
# cache-write multiplier must turn the same candidate into a skip.
monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
messages = _messages(_tool_json(300), suffix_filler_words=1000)
monkeypatch.delenv("HEADROOM_NET_COST_CACHE_TTL_SECONDS", raising=False)
five_minute = router.apply([dict(m) for m in messages], tokenizer)
assert _tool_slot_compressed(five_minute, messages)
monkeypatch.setenv("HEADROOM_NET_COST_CACHE_TTL_SECONDS", "3600")
one_hour = router.apply([dict(m) for m in messages], tokenizer)
assert not _tool_slot_compressed(one_hour, messages)
assert any(t.startswith("netcost:skip:") for t in one_hour.transforms_applied)
def test_request_one_hour_marker_overrides_env_fallback(self, router, tokenizer, monkeypatch):
monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
monkeypatch.delenv("HEADROOM_NET_COST_CACHE_TTL_SECONDS", raising=False)
for name in (
"DISABLE_PROMPT_CACHING",
"DISABLE_PROMPT_CACHING_SONNET",
"ENABLE_PROMPT_CACHING_1H",
"FORCE_PROMPT_CACHING_5M",
):
monkeypatch.delenv(name, raising=False)
messages = _messages(_tool_json(300), suffix_filler_words=1000)
messages[0] = {
"role": "user",
"content": [
{
"type": "text",
"text": "fetch the records",
"cache_control": {"type": "ephemeral", "ttl": "1h"},
}
],
}
from headroom.transforms.cold_prefix import anthropic_cache_ttl_seconds
request_ttl = anthropic_cache_ttl_seconds("claude-sonnet-4-6", messages)
assert request_ttl == 3600
five_minute = router.apply([dict(m) for m in messages], tokenizer)
assert _tool_slot_compressed(five_minute, messages)
one_hour = router.apply(
[dict(m) for m in messages],
tokenizer,
cache_ttl_seconds=request_ttl,
)
assert not _tool_slot_compressed(one_hour, messages)
assert any(t.startswith("netcost:skip:") for t in one_hour.transforms_applied)
def test_flag_on_gates_cached_results_too(self, router, tokenizer, monkeypatch):
# First apply warms the result cache with the flag off; second apply
# with the flag on must still gate the cache-hit path.