feat(policy): consume net-cost mutation gate in ContentRouter (#856 P2) (#905)

## Description

Fixes #907. Part of #904 — the **P2 (consume, flag-gated)** item from
#856's phased plan. (#903, which this was stacked on, has merged; this
is now a clean diff.)

`HEADROOM_NET_COST_POLICY=1` (default **off** — flag absent restores
byte-identical current behavior) routes every ContentRouter mutation
candidate through `CompressionPolicy.net_mutation_gain` before
compression is applied, at both decision sites: the result-cache-hit
path and the fresh-compression merge (pass 3).

v1 estimators (as specced in #856): **ΔT** exact (compressed form
already computed); **S** = token total after the slot, precomputed once
as a reverse cumulative sum (O(1) per candidate); **R / P_alive**
env-tunable (`HEADROOM_NET_COST_EXPECTED_READS`=10,
`HEADROOM_NET_COST_P_ALIVE`=1.0). Every decision logs all inputs at INFO
and increments `netcost_allowed`/`netcost_skipped` counters so the flag
can be validated from telemetry before any default-on.

Closes #907.

## Type of Change

- [x] New feature (non-breaking change which adds functionality)

## Changes Made

- Add flag-gated net-cost mutation gate to `ContentRouter` at both
mutation sites (cache-hit + fresh-compress merge).
- Precompute reverse-cumulative suffix token sums once per request for
O(1) S lookups.
- Emit INFO telemetry, `netcost_allowed`/`netcost_skipped` counters, and
a `netcost:skip:<band>` transform marker on blocked slots.
- **Review-response (4eb2307):** reject non-finite env values
(`math.isfinite` guard), count suffix tokens block-aware via
`_netcost_message_tokens()` (was `str(content)`, which miscounted
Anthropic block lists), and bucket the skip marker via `_gain_bucket()`
to bound dashboard cardinality.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] New tests added for new functionality
- [x] New and existing unit tests pass locally with my changes

### Test Output

```text
$ pytest tests/test_netcost_gate.py -q
11 passed in 0.82s

$ pytest tests/ -k "content_router or netcost or router" -q
133 passed, 8 skipped, 6120 deselected in 23.26s

$ ruff check headroom/transforms/content_router.py
All checks passed!
```

## Real Behavior Proof

- Environment: local macOS, repo .venv, Python 3.11.9, gpt-4o tokenizer
fixture
- Exact command / steps: `pytest tests/test_netcost_gate.py -q` then the
router-suite selector above; gate exercised end-to-end through the real
tokenizer + compression path (flag on via monkeypatch)
- Observed result: with R=10/P=1 defaults, a 300-row tool result
followed by a 40k-word suffix is left uncompressed (gate skips,
`netcost:skip:` marker emitted); a 2000-row result with a 5-word suffix
compresses (gate allows). Non-finite env (`inf`/`nan`) falls back to
defaults and still skips.
- Not tested: live proxy traffic / real dashboard validation — deferred
to the default-on milestone per #904 (this ships default-off precisely
to gather that telemetry first).

## Review Readiness

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

## Additional Notes

Cache-hit re-tokenization (`:2312`) and the large integration fixtures
are tracked as follow-ups in the PR review thread; both are intentional
given the flag is default-off. Known v1 limitations (whole-suffix S, no
batch awareness, static P_alive) are tracked in #904 as P2b/P3a/P3b.

PR body updated to satisfy the new PR-governance template gate (#914-era
governance workflow).

---------

Co-authored-by: integration-check <integration@local>

headroom/transforms/content_router.py

@@ -37,6 +37,7 @@ from __future__ import annotations
 import hashlib
 import json
 import logging
+import math
 import os
 import re
 import time
@@ -189,6 +190,72 @@ def _create_content_signature(
         return None
 
 
+def _gain_bucket(gain: float) -> str:
+    """Quantize a net-cost gain into a coarse magnitude band for markers.
+
+    The net-cost gate emits a ``netcost:skip:<band>`` transform marker. Using
+    the raw rounded gain would make every distinct value a unique marker and
+    blow up the cardinality of any ``transforms_applied`` aggregation. Bands
+    keep the signal (rough magnitude + sign) while bounding cardinality to a
+    handful of values. The exact gain is still logged at INFO for debugging.
+    """
+    if not math.isfinite(gain):
+        return "nan"
+    mag = abs(gain)
+    if mag < 100:
+        band = "lt100"
+    elif mag < 1000:
+        band = "lt1k"
+    elif mag < 10000:
+        band = "lt10k"
+    else:
+        band = "gte10k"
+    if gain == 0:
+        return "0"
+    return ("neg_" if gain < 0 else "") + band
+
+
+def _netcost_message_tokens(message: dict[str, Any], tokenizer: Tokenizer) -> int:
+    """Token count of a message for net-cost suffix (S) estimation.
+
+    String content is counted directly. Anthropic block-list content is
+    counted by summing the text-bearing fields (``text`` blocks and
+    ``tool_result`` content) rather than stringifying the whole list, which
+    would count Python ``repr`` punctuation and type names and badly
+    miscount S — the value that drives the break-even gate decision.
+    """
+    content = message.get("content", "")
+    if isinstance(content, str):
+        return tokenizer.count_text(content)
+    if not isinstance(content, list):
+        return tokenizer.count_text(str(content))
+    total = 0
+    for block in content:
+        if not isinstance(block, dict):
+            total += tokenizer.count_text(str(block))
+            continue
+        block_type = block.get("type")
+        if block_type == "text":
+            total += tokenizer.count_text(str(block.get("text", "")))
+        elif block_type == "tool_result":
+            tc = block.get("content", "")
+            if isinstance(tc, str):
+                total += tokenizer.count_text(tc)
+            elif isinstance(tc, list):
+                for sub in tc:
+                    if isinstance(sub, dict) and sub.get("type") == "text":
+                        total += tokenizer.count_text(str(sub.get("text", "")))
+                    else:
+                        total += tokenizer.count_text(str(sub))
+            else:
+                total += tokenizer.count_text(str(tc))
+        else:
+            # Other blocks (image, tool_use input, …) — repr is a rough proxy
+            # but bounded; these rarely dominate a suffix.
+            total += tokenizer.count_text(str(block))
+    return total
+
+
 class CompressionCache:
     """Two-tier compression cache with TTL.
 
@@ -1894,6 +1961,82 @@ class ContentRouter(Transform):
 
         return mapping
 
+    def _net_cost_allows(
+        self,
+        *,
+        slot_idx: int,
+        original_tokens: int,
+        compressed_tokens: int,
+        suffix_tokens: list[int],
+        route_counts: dict[str, int],
+        transforms_applied: list[str],
+    ) -> bool:
+        """Break-even gate for one candidate mutation (#856 P2, flag-gated).
+
+        Consumes ``CompressionPolicy.net_mutation_gain`` with the issue's v1
+        estimators: ΔT is the candidate's exact token saving (the compressed
+        form is already computed when this runs), S is the token total after
+        the slot, and R / P_alive are env-tunable constants
+        (``HEADROOM_NET_COST_EXPECTED_READS``, default 10;
+        ``HEADROOM_NET_COST_P_ALIVE``, default 1.0 — the conservative
+        full-penalty assumption). Every decision is logged with its inputs
+        and counted in ``route_counts`` so the flag can be validated from
+        telemetry before any default-on.
+        """
+        delta_t = max(0, original_tokens - compressed_tokens)
+        suffix = suffix_tokens[slot_idx + 1]
+        policy = self._runtime_compression_policy
+        if policy is None:
+            from .compression_policy import policy_default_payg
+
+            policy = policy_default_payg()
+        # Malformed env values fall back to defaults with a warning rather
+        # than crashing the request path (same posture as the #851 breaker
+        # env guard).
+        # ``float()`` parses "nan"/"inf" without raising, so a non-finite
+        # check is needed in addition to the ValueError guard — otherwise a
+        # malformed-but-parseable value would be logged verbatim (misleading
+        # telemetry) even though ``net_mutation_gain`` clamps it internally.
+        reads, p_alive = 10.0, 1.0
+        try:
+            _reads = float(os.environ.get("HEADROOM_NET_COST_EXPECTED_READS", "") or 10.0)
+            if not math.isfinite(_reads):
+                raise ValueError("non-finite")
+            reads = _reads
+        except ValueError:
+            logger.warning("HEADROOM_NET_COST_EXPECTED_READS malformed; using 10")
+        try:
+            _p_alive = float(os.environ.get("HEADROOM_NET_COST_P_ALIVE", "") or 1.0)
+            if not math.isfinite(_p_alive):
+                raise ValueError("non-finite")
+            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))
+        allowed = gain > 0.0
+        logger.info(
+            "NetCostPolicy slot=%d delta_t=%d suffix=%d reads=%.1f p_alive=%.2f gain=%.0f -> %s",
+            slot_idx,
+            delta_t,
+            suffix,
+            reads,
+            p_alive,
+            gain,
+            "mutate" if allowed else "skip",
+        )
+        if allowed:
+            route_counts.setdefault("netcost_allowed", 0)
+            route_counts["netcost_allowed"] += 1
+        else:
+            route_counts.setdefault("netcost_skipped", 0)
+            route_counts["netcost_skipped"] += 1
+            # Bucket the gain into a coarse magnitude band rather than emitting
+            # the raw value: a distinct numeric gain per skip would explode the
+            # cardinality of any ``transforms_applied`` aggregation. The exact
+            # value is still in the INFO log above for debugging.
+            transforms_applied.append(f"netcost:skip:{_gain_bucket(gain)}")
+        return allowed
+
     def apply(
         self,
         messages: list[dict[str, Any]],
@@ -2066,6 +2209,19 @@ class ContentRouter(Transform):
         # Pre-allocate result slots — None means "pending compression".
         result_slots: list[dict[str, Any] | None] = [None] * num_messages
 
+        # #856 P2 (flag-gated, default off): net-cost mutation gate. Suffix
+        # token sums are precomputed once (reverse cumulative) so each
+        # candidate's S lookup is O(1). v1 estimator per the issue: S is the
+        # token total of every message after the candidate.
+        netcost_enabled = os.environ.get("HEADROOM_NET_COST_POLICY") == "1"
+        netcost_suffix_tokens: list[int] = []
+        if netcost_enabled:
+            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(
+                    messages[j], tokenizer
+                )
+
         # Tasks: list of (slot_index, content, context, bias, content_key)
         _PendingTask = tuple[int, str, str, float, int]
         pending_tasks: list[_PendingTask] = []
@@ -2227,9 +2383,21 @@ class ContentRouter(Transform):
                 cached_compressed, cached_ratio, cached_strategy = cached
                 # Re-check ratio against current min_ratio (shifts with context pressure)
                 if cached_ratio < min_ratio:
-                    result_slots[i] = {**message, "content": cached_compressed}
-                    transforms_applied.append(f"router:{cached_strategy}:{cached_ratio:.2f}")
-                    compressed_details.append(f"{cached_strategy}:{cached_ratio:.2f}")
+                    if netcost_enabled and not self._net_cost_allows(
+                        slot_idx=i,
+                        original_tokens=tokenizer.count_text(content),
+                        compressed_tokens=tokenizer.count_text(cached_compressed),
+                        suffix_tokens=netcost_suffix_tokens,
+                        route_counts=route_counts,
+                        transforms_applied=transforms_applied,
+                    ):
+                        # Net-cost gate: mutation would cost more in cache
+                        # invalidation than it saves — leave untouched.
+                        result_slots[i] = message
+                    else:
+                        result_slots[i] = {**message, "content": cached_compressed}
+                        transforms_applied.append(f"router:{cached_strategy}:{cached_ratio:.2f}")
+                        compressed_details.append(f"{cached_strategy}:{cached_ratio:.2f}")
                 else:
                     # Threshold tightened — no longer qualifies. Move to skip.
                     self._cache.move_to_skip(content_key)
@@ -2272,7 +2440,7 @@ class ContentRouter(Transform):
             compressor_timing["parallel_compress_total"] = parallel_ms
 
             # --- Pass 3: Merge results back (sequential, updates caches) ---
-            for (slot_idx, _, _, _, content_key), (result, compress_ms) in zip(
+            for (slot_idx, task_content, _, _, content_key), (result, compress_ms) in zip(
                 pending_tasks, task_results
             ):
                 message = messages[slot_idx]
@@ -2282,13 +2450,26 @@ class ContentRouter(Transform):
                 )
 
                 if result.compression_ratio < min_ratio:
-                    # Compressed — store in result cache
+                    # Compressed — store in result cache. The cache is still
+                    # warmed when the net-cost gate blocks the slot: the
+                    # gate's verdict is contextual (suffix size), the
+                    # compression result is not.
                     self._cache.put(
                         content_key,
                         result.compressed,
                         result.compression_ratio,
                         result.strategy_used.value,
                     )
+                    if netcost_enabled and not self._net_cost_allows(
+                        slot_idx=slot_idx,
+                        original_tokens=tokenizer.count_text(task_content),
+                        compressed_tokens=tokenizer.count_text(result.compressed),
+                        suffix_tokens=netcost_suffix_tokens,
+                        route_counts=route_counts,
+                        transforms_applied=transforms_applied,
+                    ):
+                        result_slots[slot_idx] = message
+                        continue
                     result_slots[slot_idx] = {**message, "content": result.compressed}
                     transforms_applied.append(
                         f"router:{result.strategy_used.value}:{result.compression_ratio:.2f}"

tests/test_netcost_gate.py

@@ -0,0 +1,192 @@
+"""Net-cost mutation gate in ContentRouter (#856 P2, flag-gated).
+
+``HEADROOM_NET_COST_POLICY=1`` routes every router mutation candidate
+through ``CompressionPolicy.net_mutation_gain`` with the issue's v1
+estimators (exact ΔT, S = token total after the slot, env-tunable R and
+P_alive). Flag off (default) preserves exact current behavior.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from headroom import OpenAIProvider, Tokenizer
+from headroom.transforms.content_router import ContentRouter, ContentRouterConfig
+
+_provider = OpenAIProvider()
+
+
+@pytest.fixture
+def tokenizer() -> Tokenizer:
+    return Tokenizer(_provider.get_token_counter("gpt-4o"), "gpt-4o")
+
+
+@pytest.fixture
+def router() -> ContentRouter:
+    return ContentRouter(ContentRouterConfig())
+
+
+def _tool_json(rows: int) -> str:
+    return json.dumps(
+        [{"id": i, "name": f"item_{i}", "status": "ok", "score": i * 3.14} for i in range(rows)]
+    )
+
+
+def _messages(tool_content: str, suffix_filler_words: int) -> list[dict]:
+    suffix = "analysis context word " * suffix_filler_words
+    return [
+        {"role": "user", "content": "fetch the records"},
+        {"role": "tool", "content": tool_content},
+        {"role": "user", "content": suffix},
+        {"role": "user", "content": "summarize"},
+    ]
+
+
+def _tool_slot_compressed(result, messages) -> bool:
+    return result.messages[1]["content"] != messages[1]["content"]
+
+
+class TestNetCostGate:
+    def test_flag_off_compresses_as_before(self, router, tokenizer, monkeypatch):
+        monkeypatch.delenv("HEADROOM_NET_COST_POLICY", raising=False)
+        messages = _messages(_tool_json(300), suffix_filler_words=4000)
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert _tool_slot_compressed(result, messages)
+        assert not any(t.startswith("netcost:") for t in result.transforms_applied)
+
+    def test_flag_on_blocks_when_suffix_dominates(self, router, tokenizer, monkeypatch):
+        # Big suffix after a modest shave: corrected formula says the cache
+        # invalidation outweighs the saving -> slot left untouched.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        messages = _messages(_tool_json(300), suffix_filler_words=40000)
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert not _tool_slot_compressed(result, messages)
+        assert any(t.startswith("netcost:skip:") for t in result.transforms_applied)
+
+    def test_flag_on_allows_when_shave_dominates(self, router, tokenizer, monkeypatch):
+        # Tiny suffix after a huge shave -> gate allows, compression applies.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        messages = _messages(_tool_json(2000), suffix_filler_words=5)
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert _tool_slot_compressed(result, messages)
+        assert not any(t.startswith("netcost:skip:") for t in result.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.
+        messages = _messages(_tool_json(300), suffix_filler_words=40000)
+        monkeypatch.delenv("HEADROOM_NET_COST_POLICY", raising=False)
+        warm = router.apply([dict(m) for m in messages], tokenizer)
+        assert _tool_slot_compressed(warm, messages)
+
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        gated = router.apply([dict(m) for m in messages], tokenizer)
+        assert not _tool_slot_compressed(gated, messages)
+        assert any(t.startswith("netcost:skip:") for t in gated.transforms_applied)
+
+    def test_malformed_env_falls_back_to_defaults(self, router, tokenizer, monkeypatch):
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        monkeypatch.setenv("HEADROOM_NET_COST_EXPECTED_READS", "lots")
+        monkeypatch.setenv("HEADROOM_NET_COST_P_ALIVE", "warm")
+        messages = _messages(_tool_json(300), suffix_filler_words=40000)
+        # Must not raise; defaults (R=10, P=1) still block this scenario.
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert not _tool_slot_compressed(result, messages)
+
+    def test_p_alive_zero_disables_penalty(self, router, tokenizer, monkeypatch):
+        # Cold cache (P_alive=0): no suffix penalty, mutation always wins.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        monkeypatch.setenv("HEADROOM_NET_COST_P_ALIVE", "0")
+        messages = _messages(_tool_json(300), suffix_filler_words=40000)
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert _tool_slot_compressed(result, messages)
+
+    def test_nonfinite_env_falls_back_to_defaults(self, router, tokenizer, monkeypatch):
+        # ``float("inf")``/``float("nan")`` parse without ValueError; the gate
+        # must reject them and fall back to defaults so telemetry isn't
+        # poisoned. With R=10/P=1 defaults this scenario still skips.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        monkeypatch.setenv("HEADROOM_NET_COST_EXPECTED_READS", "inf")
+        monkeypatch.setenv("HEADROOM_NET_COST_P_ALIVE", "nan")
+        messages = _messages(_tool_json(300), suffix_filler_words=40000)
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert not _tool_slot_compressed(result, messages)
+        # Marker must be a bounded band, never a raw float / "nan".
+        skip_markers = [t for t in result.transforms_applied if t.startswith("netcost:skip:")]
+        assert skip_markers
+        assert all(m.split(":")[-1] in _GAIN_BANDS for m in skip_markers)
+
+
+_GAIN_BANDS = {
+    "0",
+    "lt100",
+    "lt1k",
+    "lt10k",
+    "gte10k",
+    "neg_lt100",
+    "neg_lt1k",
+    "neg_lt10k",
+    "neg_gte10k",
+    "nan",
+}
+
+
+class TestNetCostHelpers:
+    def test_gain_bucket_bands_and_sign(self):
+        from headroom.transforms.content_router import _gain_bucket
+
+        assert _gain_bucket(0) == "0"
+        assert _gain_bucket(50) == "lt100"
+        assert _gain_bucket(500) == "lt1k"
+        assert _gain_bucket(5000) == "lt10k"
+        assert _gain_bucket(50000) == "gte10k"
+        assert _gain_bucket(-50) == "neg_lt100"
+        assert _gain_bucket(-50000) == "neg_gte10k"
+        assert _gain_bucket(float("nan")) == "nan"
+        assert _gain_bucket(float("inf")) == "nan"
+
+    def test_message_tokens_block_list_beats_repr(self, tokenizer):
+        # str(content) over a block list counts repr punctuation/type names;
+        # the block-aware helper counts only the text-bearing payload.
+        from headroom.transforms.content_router import _netcost_message_tokens
+
+        text = "word " * 200
+        block_msg = {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": text},
+                {"type": "image", "source": {"data": "x" * 500}},
+            ],
+        }
+        helper = _netcost_message_tokens(block_msg, tokenizer)
+        text_only = tokenizer.count_text(text)
+        # Helper tracks the text payload closely; the image block adds only a
+        # small repr proxy, far less than stringifying the whole list.
+        assert abs(helper - text_only) < text_only * 0.5
+        assert helper < tokenizer.count_text(str(block_msg["content"]))
+
+    def test_message_tokens_tool_result_blocks(self, tokenizer):
+        from headroom.transforms.content_router import _netcost_message_tokens
+
+        payload = "log line " * 100
+        msg = {
+            "role": "user",
+            "content": [
+                {
+                    "type": "tool_result",
+                    "tool_use_id": "t1",
+                    "content": [{"type": "text", "text": payload}],
+                }
+            ],
+        }
+        assert _netcost_message_tokens(msg, tokenizer) >= tokenizer.count_text(payload) * 0.8
+
+    def test_message_tokens_string_content(self, tokenizer):
+        from headroom.transforms.content_router import _netcost_message_tokens
+
+        s = "plain string content " * 50
+        assert _netcost_message_tokens({"role": "user", "content": s}, tokenizer) == (
+            tokenizer.count_text(s)
+        )