feat: net-cost cache mutation formula on CompressionPolicy (#856 P1) (#857)

Closes #856

**P1 of the #856 phased plan** — pure functions, zero behavior change.
(Closing keyword links the issue; if P2 hasn't started when this merges,
reopen #856 or it remains the design record for the P2/P3 follow-up
PRs.)

## What

Adds the break-even decision rule for deep (pre-cache-marker) edits to
`CompressionPolicy`:

```
gain = ΔT · (w + r·(R−1)) − P_alive · (w − r) · S
```

- `net_mutation_gain()`, `should_mutate_deep()` (gain > 0),
`break_even_reads()` (R = ((w−r)/r)·(S/ΔT−1) ≈ 11.5·S/ΔT) on the Rust
struct (source of truth) and the Python hand-mirror, following the
existing F2.1/F2.2 parity pattern.
- `CACHE_WRITE_MULTIPLIER = 1.25` / `CACHE_READ_MULTIPLIER = 0.1` public
constants (Anthropic 5-minute tier).
- Inputs clamped (`expected_reads ≥ 0`, `p_alive ∈ [0,1]`); methods take
`&self`/`self` so a follow-up can add per-mode margins.
- The formula derives the existing Subscription live-zone policy as its
S=0 special case rather than contradicting it.

**No callers yet.** P2 (consuming this in `TransformPipeline` behind
`HEADROOM_NET_COST_POLICY`, replacing the binary `live_zone_only` gate,
with decision telemetry) is specified in #856 and awaits maintainer
direction — this PR just lands the audited arithmetic both dispatchers
will share.

## Tests

Golden-value parity: 6 new Rust unit tests and 7 new Python tests assert
the **identical scenario numbers** (loss −53 200 for a 2K shave under a
50K warm suffix at R=10; win +61 000 for a 50K shave under a 10K suffix
at R=3; S=0 always profitable; P_alive=0 always profitable — the
idle-timer window; clamping; break-even 276 reads for the 2K/50K
anchor). A drift on either side trips the pair loudly, same contract as
the existing field-map parity test.

- `cargo test -p headroom-core --lib compression_policy`: 12 passed (6
existing + 6 new)
- `pytest tests/test_compression_policy.py`: 17 passed (10 existing + 7
new)
- `cargo fmt --check`, `cargo clippy -p headroom-core` clean; `ruff
check` + `ruff format --check` clean

## Real behavior proof

Not applicable in the runtime sense — this PR intentionally adds **no
runtime behavior** (pure functions, no call sites). The arithmetic is
validated against the research anchors above in both languages' test
suites; live decision telemetry arrives with P2 where the formula first
gates real traffic.

## Out of scope

P2 (flag-gated pipeline consumption + telemetry), P3 (deep-edit
batching, idle-timer compaction near TTL lapse), retiring the deprecated
`volatile_token_threshold`/`max_lossy_ratio` fields — all tracked in
#856.

---------

Co-authored-by: Ash Rhodes <ashley.rhodes@king.com>

crates/headroom-core/src/compression_policy.rs

@@ -130,6 +130,16 @@ pub(crate) const MAX_LOSSY_RATIO_PAYG: f32 = 0.45;
 /// Subscription: conservative cap at 25%. Cache stability over savings.
 pub(crate) const MAX_LOSSY_RATIO_SUBSCRIPTION: f32 = 0.25;
 
+/// Anthropic prompt-cache write multiplier: a `cache_creation` token
+/// costs 1.25× a plain input token (5-minute TTL tier). Input to the
+/// net-cost mutation formula (#856).
+pub const CACHE_WRITE_MULTIPLIER: f32 = 1.25;
+
+/// 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;
+
 /// Per-auth-mode policy that downstream compression stages consult.
 ///
 /// `Copy` because the struct is small POD (two `bool`s + a `u32` + an
@@ -224,6 +234,79 @@ impl CompressionPolicy {
     pub fn live_zone_compression_enabled(&self) -> bool {
         true
     }
+
+    /// Net gain (in plain-input-token cost units) of a mutation that
+    /// removes `delta_t` tokens from a message whose cached suffix is
+    /// `suffix_tokens` long (#856).
+    ///
+    /// Mutating message K invalidates every cached token after it: the
+    /// suffix is re-written once at the write multiplier instead of
+    /// being read at the read multiplier, costing
+    /// `P_alive · (w − r) · S`. In exchange, `delta_t` tokens are gone
+    /// from the current write and every one of the `expected_reads`
+    /// remaining reads of the chain, saving `ΔT · (w + r·(R − 1))`.
+    ///
+    ///   gain = ΔT · (w + r·(R − 1))  −  P_alive · (w − r) · S
+    ///
+    /// Sanity anchors (Anthropic w=1.25, r=0.1), matching the unit
+    /// tests below: a 2K shave under a 50K warm suffix needs ~276
+    /// remaining reads to pay off (rarely profitable); a 50K shave
+    /// under a 10K suffix is profitable from the first write (its
+    /// break-even read count is negative); a live-zone edit (S = 0)
+    /// is always profitable.
+    ///
+    /// Takes `&self` so a follow-up can apply per-mode margins; today
+    /// the arithmetic is mode-independent. Inputs are clamped:
+    /// `expected_reads` to `>= 0` (NaN → 0), `p_alive` to `[0, 1]`
+    /// (NaN → 1, the conservative full-penalty assumption).
+    pub fn net_mutation_gain(
+        &self,
+        delta_t: u32,
+        suffix_tokens: u32,
+        expected_reads: f32,
+        p_alive: f32,
+    ) -> f32 {
+        let w = 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.
+        let reads = expected_reads.max(0.0);
+        let alive = if p_alive.is_nan() {
+            1.0
+        } else {
+            p_alive.clamp(0.0, 1.0)
+        };
+        (delta_t as f32) * (w + r * (reads - 1.0)) - alive * (w - r) * (suffix_tokens as f32)
+    }
+
+    /// Decision form of [`Self::net_mutation_gain`]: mutate iff the
+    /// gain is strictly positive.
+    pub fn should_mutate_deep(
+        &self,
+        delta_t: u32,
+        suffix_tokens: u32,
+        expected_reads: f32,
+        p_alive: f32,
+    ) -> bool {
+        self.net_mutation_gain(delta_t, suffix_tokens, expected_reads, p_alive) > 0.0
+    }
+
+    /// Remaining-read count at which a warm-cache (P_alive = 1)
+    /// mutation breaks even:
+    ///
+    ///   R = ((w − r) / r) · (S/ΔT − 1)   ≈ 11.5 · S/ΔT  for S ≫ ΔT
+    ///
+    /// Useful for decision telemetry ("this edit pays off if the
+    /// 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 {
+        if delta_t == 0 {
+            return 0.0;
+        }
+        let w = CACHE_WRITE_MULTIPLIER;
+        let r = CACHE_READ_MULTIPLIER;
+        ((w - r) / r) * ((suffix_tokens as f32) / (delta_t as f32) - 1.0)
+    }
 }
 
 #[cfg(test)]
@@ -322,4 +405,80 @@ mod tests {
             );
         }
     }
+
+    // --- Net-cost mutation formula (#856). Scenario values are golden:
+    // tests/test_compression_policy.py asserts the identical numbers
+    // against the Python hand-mirror, so a drift in either side trips
+    // the parity pair loudly.
+
+    #[test]
+    fn net_gain_small_shave_deep_suffix_is_loss() {
+        // Shave 2K under a 50K warm suffix at R=10 remaining reads:
+        // 2000·(1.25 + 0.1·9) − 1.0·1.15·50000 = 4300 − 57500 = −53200.
+        let p = CompressionPolicy::for_mode(AuthMode::Payg);
+        let gain = p.net_mutation_gain(2_000, 50_000, 10.0, 1.0);
+        assert!((gain - (-53_200.0)).abs() < 1.0, "gain = {gain}");
+        assert!(!p.should_mutate_deep(2_000, 50_000, 10.0, 1.0));
+    }
+
+    #[test]
+    fn net_gain_big_shave_shallow_suffix_is_win() {
+        // Shave 50K under a 10K warm suffix at R=3:
+        // 50000·(1.25 + 0.1·2) − 1.0·1.15·10000 = 72500 − 11500 = 61000.
+        let p = CompressionPolicy::for_mode(AuthMode::Payg);
+        let gain = p.net_mutation_gain(50_000, 10_000, 3.0, 1.0);
+        assert!((gain - 61_000.0).abs() < 1.0, "gain = {gain}");
+        assert!(p.should_mutate_deep(50_000, 10_000, 3.0, 1.0));
+    }
+
+    #[test]
+    fn net_gain_live_zone_edit_always_profitable() {
+        // S = 0 derives the existing Subscription live-zone policy as a
+        // special case: nothing cached is invalidated, so any positive
+        // shave wins even at R=0 (gain = ΔT·(w − r) > 0).
+        let p = CompressionPolicy::for_mode(AuthMode::Subscription);
+        assert!(p.should_mutate_deep(1, 0, 0.0, 1.0));
+        assert!(p.should_mutate_deep(2_000, 0, 0.0, 1.0));
+    }
+
+    #[test]
+    fn net_gain_cold_cache_ignores_suffix() {
+        // P_alive = 0 (TTL lapsed): no warm suffix to lose, so even the
+        // worst shave/suffix ratio is profitable. This is the idle-timer
+        // compaction window.
+        let p = CompressionPolicy::for_mode(AuthMode::Payg);
+        assert!(p.should_mutate_deep(2_000, 50_000, 0.0, 0.0));
+    }
+
+    #[test]
+    fn net_gain_clamps_out_of_range_inputs() {
+        let p = CompressionPolicy::for_mode(AuthMode::Payg);
+        // Negative reads clamp to 0; p_alive > 1 clamps to 1.
+        let clamped = p.net_mutation_gain(2_000, 50_000, -5.0, 7.0);
+        let reference = p.net_mutation_gain(2_000, 50_000, 0.0, 1.0);
+        assert!((clamped - reference).abs() < f32::EPSILON);
+    }
+
+    #[test]
+    fn net_gain_guards_nan_inputs() {
+        // NaN reads → 0, NaN p_alive → 1: gain stays finite and matches
+        // the conservative reference instead of poisoning the decision.
+        let p = CompressionPolicy::for_mode(AuthMode::Payg);
+        let guarded = p.net_mutation_gain(2_000, 50_000, f32::NAN, f32::NAN);
+        assert!(guarded.is_finite());
+        let reference = p.net_mutation_gain(2_000, 50_000, 0.0, 1.0);
+        assert!((guarded - reference).abs() < f32::EPSILON);
+    }
+
+    #[test]
+    fn break_even_reads_matches_research_anchor() {
+        // R = 11.5·(S/ΔT − 1): 2K shave / 50K suffix → 11.5·24 = 276
+        // (rarely profitable); 50K shave / 10K suffix →
+        // 11.5·(0.2 − 1) < 0 → profitable from the first read.
+        let p = CompressionPolicy::for_mode(AuthMode::Payg);
+        let r = p.break_even_reads(2_000, 50_000);
+        assert!((r - 276.0).abs() < 0.5, "break-even = {r}");
+        assert!(p.break_even_reads(50_000, 10_000) < 0.0);
+        assert_eq!(p.break_even_reads(0, 10_000), 0.0);
+    }
 }

headroom/transforms/compression_policy.py

@@ -18,6 +18,7 @@ docstring (per-mode rationale, why-a-struct, etc.).
 
 from __future__ import annotations
 
+import math
 import os
 from dataclasses import dataclass
 
@@ -54,6 +55,16 @@ _MAX_LOSSY_RATIO_PAYG: float = 0.45
 #: Subscription: conservative cap at 25%. Cache stability over savings.
 _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``.
+CACHE_WRITE_MULTIPLIER: float = 1.25
+
+#: 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
+
 
 @dataclass(frozen=True, slots=True)
 class CompressionPolicy:
@@ -112,6 +123,64 @@ class CompressionPolicy:
     ``smart_crusher.py`` and ``content_router.py`` at the
     ``record_compression`` call sites."""
 
+    def net_mutation_gain(
+        self,
+        delta_t: int,
+        suffix_tokens: int,
+        expected_reads: float,
+        p_alive: float,
+    ) -> float:
+        """Net gain (in plain-input-token cost units) of a mutation that
+        removes ``delta_t`` tokens from a message whose cached suffix is
+        ``suffix_tokens`` long (#856).
+
+        Mirrors ``CompressionPolicy::net_mutation_gain`` in the Rust
+        crate (source of truth — see its docstring for the derivation)::
+
+            gain = dT * (w + r*(R - 1)) - P_alive * (w - r) * S
+
+        Inputs are clamped: ``delta_t``/``suffix_tokens`` to ``>= 0``
+        (the Rust signature takes ``u32``), ``expected_reads`` to
+        ``>= 0`` (NaN → 0), ``p_alive`` to ``[0, 1]`` (NaN → 1, the
+        conservative full-penalty assumption — same as Rust).
+        """
+        w = CACHE_WRITE_MULTIPLIER
+        r = CACHE_READ_MULTIPLIER
+        dt = max(0, delta_t)
+        suffix = max(0, suffix_tokens)
+        # Python max()/min() propagate NaN from the first argument, unlike
+        # f32::max in the Rust source of truth — guard explicitly.
+        reads = 0.0 if math.isnan(expected_reads) else max(expected_reads, 0.0)
+        alive = 1.0 if math.isnan(p_alive) else min(max(p_alive, 0.0), 1.0)
+        return float(dt) * (w + r * (reads - 1.0)) - alive * (w - r) * float(suffix)
+
+    def should_mutate_deep(
+        self,
+        delta_t: int,
+        suffix_tokens: int,
+        expected_reads: float,
+        p_alive: float,
+    ) -> 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
+
+    def break_even_reads(self, delta_t: int, suffix_tokens: int) -> float:
+        """Remaining-read count at which a warm-cache (``p_alive=1``)
+        mutation breaks even::
+
+            R = ((w - r) / r) * (S/dT - 1)   ~= 11.5 * S/dT  for S >> dT
+
+        Returns 0 when ``delta_t`` is ``<= 0`` (no savings — callers
+        gate on ``delta_t > 0``; the Rust signature takes ``u32``).
+        Mirrors the Rust method.
+        """
+        if delta_t <= 0:
+            return 0.0
+        w = CACHE_WRITE_MULTIPLIER
+        r = CACHE_READ_MULTIPLIER
+        return ((w - r) / r) * (float(max(0, suffix_tokens)) / float(delta_t) - 1.0)
+
 
 def policy_for_mode(mode: AuthMode) -> CompressionPolicy:
     """Resolve the F2.1+F2.2 policy for an auth mode.

tests/test_compression_policy.py

@@ -157,3 +157,81 @@ class TestRustParityFieldMap:
             f"and `crates/headroom-core/src/compression_policy.rs` in "
             f"the same commit."
         )
+
+
+class TestNetCostFormula:
+    """Net-cost mutation formula (#856) — Rust parity.
+
+    Scenario values are golden: the Rust unit tests in
+    ``crates/headroom-core/src/compression_policy.rs`` assert the
+    identical numbers, so a drift in either side trips the parity pair
+    loudly.
+    """
+
+    def test_small_shave_deep_suffix_is_loss(self):
+        # 2000*(1.25 + 0.1*9) - 1.0*1.15*50000 = 4300 - 57500 = -53200.
+        p = policy_for_mode(AuthMode.PAYG)
+        gain = p.net_mutation_gain(2_000, 50_000, 10.0, 1.0)
+        assert abs(gain - (-53_200.0)) < 1.0
+        assert not p.should_mutate_deep(2_000, 50_000, 10.0, 1.0)
+
+    def test_big_shave_shallow_suffix_is_win(self):
+        # 50000*(1.25 + 0.1*2) - 1.0*1.15*10000 = 72500 - 11500 = 61000.
+        p = policy_for_mode(AuthMode.PAYG)
+        gain = p.net_mutation_gain(50_000, 10_000, 3.0, 1.0)
+        assert abs(gain - 61_000.0) < 1.0
+        assert p.should_mutate_deep(50_000, 10_000, 3.0, 1.0)
+
+    def test_live_zone_edit_always_profitable(self):
+        # S = 0 derives the existing Subscription live-zone policy as a
+        # special case of the formula.
+        p = policy_for_mode(AuthMode.SUBSCRIPTION)
+        assert p.should_mutate_deep(1, 0, 0.0, 1.0)
+        assert p.should_mutate_deep(2_000, 0, 0.0, 1.0)
+
+    def test_cold_cache_ignores_suffix(self):
+        # P_alive = 0 (TTL lapsed): the idle-timer compaction window.
+        p = policy_for_mode(AuthMode.PAYG)
+        assert p.should_mutate_deep(2_000, 50_000, 0.0, 0.0)
+
+    def test_clamps_out_of_range_inputs(self):
+        p = policy_for_mode(AuthMode.PAYG)
+        clamped = p.net_mutation_gain(2_000, 50_000, -5.0, 7.0)
+        reference = p.net_mutation_gain(2_000, 50_000, 0.0, 1.0)
+        assert abs(clamped - reference) < 1e-6
+
+    def test_nan_inputs_guarded(self):
+        # NaN reads -> 0, NaN p_alive -> 1 (same as Rust): the gain stays
+        # finite instead of poisoning the mutate decision.
+        import math
+
+        p = policy_for_mode(AuthMode.PAYG)
+        guarded = p.net_mutation_gain(2_000, 50_000, float("nan"), float("nan"))
+        assert math.isfinite(guarded)
+        reference = p.net_mutation_gain(2_000, 50_000, 0.0, 1.0)
+        assert abs(guarded - reference) < 1e-6
+
+    def test_negative_int_inputs_clamped(self):
+        # Rust takes u32 — negative Python ints must not flip the sign of
+        # the result; they clamp to 0.
+        p = policy_for_mode(AuthMode.PAYG)
+        assert p.net_mutation_gain(-2_000, -50_000, 5.0, 1.0) == p.net_mutation_gain(0, 0, 5.0, 1.0)
+        assert p.break_even_reads(-5, 10_000) == 0.0
+        assert p.net_mutation_gain(2_000, -1, 5.0, 1.0) == p.net_mutation_gain(2_000, 0, 5.0, 1.0)
+
+    def test_break_even_reads_matches_research_anchor(self):
+        # R = 11.5*(S/dT - 1): 2K/50K -> 276; 50K/10K -> negative
+        # (profitable from the first read); dT=0 -> 0.
+        p = policy_for_mode(AuthMode.PAYG)
+        assert abs(p.break_even_reads(2_000, 50_000) - 276.0) < 0.5
+        assert p.break_even_reads(50_000, 10_000) < 0.0
+        assert p.break_even_reads(0, 10_000) == 0.0
+
+    def test_constants_match_rust(self):
+        from headroom.transforms.compression_policy import (
+            CACHE_READ_MULTIPLIER,
+            CACHE_WRITE_MULTIPLIER,
+        )
+
+        assert CACHE_WRITE_MULTIPLIER == 1.25
+        assert CACHE_READ_MULTIPLIER == 0.1