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.

crates/headroom-core/src/compression_policy.rs

@@ -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

headroom/proxy/handlers/anthropic.py

@@ -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),
                             )
                         ),

headroom/transforms/compression_policy.py

@@ -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))
 

headroom/transforms/content_router.py

@@ -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 / netcost_ttl)
-                    netcost_p_alive_override = max(0.0, 1.0 - idle_f / 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

tests/test_compression_policy.py

@@ -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

tests/test_netcost_gate.py

@@ -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.