feat(policy): batch deep edits through one cache-bust (#856 P3a) (#1015)

## Description

#856 P3a (umbrella #904), stacked on the now-merged P2 (#905) and P2b
(#944).

A net-cost mutation at depth K already busts the provider's cached
suffix after K. Every *later* candidate at a deeper slot therefore rides
that same cache invalidation for free — mutating it adds no incremental
cache-bust cost. Today the P2 break-even gate re-charges each candidate
the full invalidated suffix S independently, so a batch of legitimate
deep edits is under-admitted: only the first pays for the bust, yet each
is billed as if it paid alone.

This adds a batch-reclaim floor to the net-cost gate so that once one
net-positive deep edit is admitted at slot K, candidates at slot > K are
admitted on the write/read economics alone (S charged as 0). Flag-gated
under `HEADROOM_NET_COST_POLICY` (the same flag as P2/P2b), default
**off** — telemetry-first before any default-on.

## Type of Change

- [x] New feature (non-breaking change which adds functionality)
- [ ] Bug fix
- [ ] Breaking change
- [ ] Documentation

## Changes Made

- `ContentRouter._net_cost_allows`: new `batch_state` param. When the
candidate sits strictly deeper than `batch_state["floor"]`, S is charged
as 0 via the *same* `net_mutation_gain` formula (conservative — never
admits a mutation the real economics would reject). Full-S admits
open/lower the floor; batch admits never lower it, so a slot only ever
rides free behind a genuinely mutated shallower slot.
- `ContentRouter.apply`: shared per-request `netcost_batch_state` wired
into both gate call sites (cached-result path and parallel-merge path).
- Telemetry: every batch admission emits the
`router:netcost_batch_admit` transform marker and the
`netcost_batch_admitted` route counter; added to the routing summary log
line.
- Tests: 5 new cases in `tests/test_netcost_gate.py`
(`TestNetCostBatchReclaim`).

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] New tests added for new functionality

### Test Output

```text
$ pytest tests/test_netcost_gate.py -q
20 passed in 1.46s

$ pytest tests/ -k "content_router or netcost or router" -q
142 passed, 8 skipped, 6342 deselected, 1 warning in 22.54s

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

$ ruff format --check headroom/transforms/content_router.py tests/test_netcost_gate.py
2 files already formatted

$ mypy headroom/transforms/content_router.py
Success: no issues found in 1 source file
```

## Real Behavior Proof

- Environment: local macOS, repo .venv, Python 3.11.9, gpt-4o tokenizer
fixture
- Exact command / steps: drive `ContentRouter.apply()` on a 5-message
conversation — a huge compressible tool dump at slot 1 (ΔT≈34K) and a
modest dump at slot 2 (ΔT≈5K) followed by a ~12K-token suffix, so slot
2's own break-even S blocks it. Run with `HEADROOM_NET_COST_POLICY=1`,
once with a non-compressible slot 1 (no shallower admit, control) and
once with the slot-1 dump intact (opens the floor).
- Observed result: control → `slot2_compressed=False batch_markers=0
skip_markers=1` (slot 2 correctly blocked on its own S, no floor
opened); floor opened → `slot2_compressed=True batch_markers=1
skip_markers=0` (slot 2 rides slot 1's cache-bust for free,
`router:netcost_batch_admit` emitted). Flag absent → no
`router:netcost_batch_admit` marker ever.
- Not tested: live proxy traffic / real dashboard validation — deferred
to the default-on milestone per #904 (ships default-off precisely to
gather telemetry first). Known limitation logged for follow-up: in a
*warm-cache* request a deep cache-hit slot is gated in pass 1 before a
shallower cache-miss slot can lower the floor in pass 3, so the batch
win can no-op there (never a wrong admit — strictly conservative).

## Review Readiness

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

## Additional Notes

Charging S=0 through the existing formula (rather than blanket-admitting
on `ΔT > 0`) keeps the decision conservative under non-default env
tunables (`HEADROOM_NET_COST_EXPECTED_READS`,
`HEADROOM_NET_COST_P_ALIVE`). P3b will be a separate PR after this
review.

Note: the failing `test` / `test-extras` checks are a **pre-existing
regression on `main`** in `tests/test_cache/test_dynamic_detector.py`
(unrelated to this PR, which only touches `content_router.py`). Fix
tracked in a separate PR; this branch will go green once that lands and
this is rebased.

headroom/transforms/content_router.py

@@ -1970,6 +1970,7 @@ class ContentRouter(Transform):
         suffix_tokens: list[int],
         route_counts: dict[str, int],
         transforms_applied: list[str],
+        batch_state: dict[str, int | None] | None = None,
     ) -> bool:
         """Break-even gate for one candidate mutation (#856 P2, flag-gated).
 
@@ -1982,9 +1983,31 @@ class ContentRouter(Transform):
         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.
+
+        #856 P3a (batch deep edits): a mutation at depth K busts the
+        provider's cached suffix after K, so every *later* candidate at a
+        deeper slot rides that same invalidation for free — mutating it adds
+        no incremental cache-bust cost. ``batch_state["floor"]`` tracks the
+        shallowest slot already admitted as a net-positive mutation. When the
+        current candidate sits strictly deeper than that floor, S is charged
+        as 0 (rather than the full invalidated suffix), so the break-even
+        formula admits it on the write/read economics alone. Charging S=0 via
+        the same ``net_mutation_gain`` (instead of blanket-admitting on
+        ``delta_t > 0``) keeps the decision conservative: it never admits a
+        mutation the real economics would reject. The floor is only set/lowered
+        by full-S admits, so a slot only ever rides free behind a genuinely
+        mutated shallower slot. Each batch admission emits the
+        ``router:netcost_batch_admit`` marker and the ``netcost_batch_admitted``
+        counter for telemetry.
         """
         delta_t = max(0, original_tokens - compressed_tokens)
-        suffix = suffix_tokens[slot_idx + 1]
+        # Batch reclaim: if a shallower slot was already admitted, its
+        # cache-bust already invalidated everything after it, including this
+        # slot — so charge S=0 here. Otherwise S is the full suffix after the
+        # candidate (P2 v1 estimator).
+        floor = batch_state.get("floor") if batch_state is not None else None
+        batch_reclaim = floor is not None and slot_idx > floor
+        suffix = 0 if batch_reclaim else suffix_tokens[slot_idx + 1]
         policy = self._runtime_compression_policy
         if policy is None:
             from .compression_policy import policy_default_payg
@@ -2015,18 +2038,31 @@ class ContentRouter(Transform):
         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",
+            "NetCostPolicy slot=%d delta_t=%d suffix=%d reads=%.1f p_alive=%.2f "
+            "gain=%.0f batch_reclaim=%s -> %s",
             slot_idx,
             delta_t,
             suffix,
             reads,
             p_alive,
             gain,
+            batch_reclaim,
             "mutate" if allowed else "skip",
         )
         if allowed:
             route_counts.setdefault("netcost_allowed", 0)
             route_counts["netcost_allowed"] += 1
+            if batch_reclaim:
+                # Rode a shallower edit's cache-bust for free — telemetry only;
+                # the floor is unchanged (this slot is deeper than the floor).
+                route_counts.setdefault("netcost_batch_admitted", 0)
+                route_counts["netcost_batch_admitted"] += 1
+                transforms_applied.append("router:netcost_batch_admit")
+            elif batch_state is not None:
+                # First/shallower full-S admit — open (or lower) the batch
+                # floor so deeper candidates can reclaim against it.
+                current = batch_state.get("floor")
+                batch_state["floor"] = slot_idx if current is None else min(current, slot_idx)
         else:
             route_counts.setdefault("netcost_skipped", 0)
             route_counts["netcost_skipped"] += 1
@@ -2215,6 +2251,10 @@ class ContentRouter(Transform):
         # token total of every message after the candidate.
         netcost_enabled = os.environ.get("HEADROOM_NET_COST_POLICY") == "1"
         netcost_suffix_tokens: list[int] = []
+        # #856 P3a: shared batch-reclaim state for this request. ``floor`` is
+        # the shallowest slot admitted as a net-positive mutation; once set,
+        # deeper candidates charge S=0 (their cache-bust is already paid).
+        netcost_batch_state: dict[str, int | None] = {"floor": None}
         if netcost_enabled:
             netcost_suffix_tokens = [0] * (num_messages + 1)
             for j in range(num_messages - 1, -1, -1):
@@ -2411,6 +2451,7 @@ class ContentRouter(Transform):
                         suffix_tokens=netcost_suffix_tokens,
                         route_counts=route_counts,
                         transforms_applied=transforms_applied,
+                        batch_state=netcost_batch_state,
                     ):
                         # Net-cost gate: mutation would cost more in cache
                         # invalidation than it saves — leave untouched.
@@ -2492,6 +2533,7 @@ class ContentRouter(Transform):
                         suffix_tokens=netcost_suffix_tokens,
                         route_counts=route_counts,
                         transforms_applied=transforms_applied,
+                        batch_state=netcost_batch_state,
                     ):
                         result_slots[slot_idx] = message
                         continue
@@ -2547,6 +2589,8 @@ class ContentRouter(Transform):
             parts.append(f"{route_counts['cache_hit']} cache hits")
         if route_counts.get("cache_miss"):
             parts.append(f"{route_counts['cache_miss']} cache misses")
+        if route_counts.get("netcost_batch_admitted"):
+            parts.append(f"{route_counts['netcost_batch_admitted']} netcost batch-admitted")
         cs = self._cache.stats
         if cs["cache_size"] > 0 or cs["cache_skip_size"] > 0:
             parts.append(

tests/test_netcost_gate.py

@@ -262,3 +262,95 @@ class TestNetCostFrozenUnlock:
         result = router.apply([dict(m) for m in messages], tokenizer, frozen_message_count=2)
         assert result.messages[1]["content"] == original[1]["content"]
         assert "router:netcost_frozen_unlock" not in result.transforms_applied
+
+
+class TestNetCostBatchReclaim:
+    """#856 P3a: batch deep edits -- once one net-positive edit is admitted at
+    slot K, deeper candidates ride that cache-bust for free (S charged as 0).
+
+    Tests are cache-cold (fresh router per fixture) so every candidate flows
+    through the parallel-merge pass in ascending slot order, which is where the
+    shared batch_state floor is set and then reclaimed.
+    """
+
+    @staticmethod
+    def _convo(slot1: str, slot2: str, filler_words: int) -> list[dict]:
+        # Two consecutive tool dumps (slots 1 and 2) followed by a filler
+        # suffix. Slot 1 is the shallower candidate; slot 2 the deeper one.
+        suffix = "analysis context word " * filler_words
+        return [
+            {"role": "user", "content": "fetch the records"},
+            {"role": "tool", "content": slot1},
+            {"role": "tool", "content": slot2},
+            {"role": "user", "content": suffix},
+            {"role": "user", "content": "summarize"},
+        ]
+
+    @staticmethod
+    def _compressed(result, original, idx: int) -> bool:
+        return result.messages[idx]["content"] != original[idx]["content"]
+
+    def test_flag_off_no_batch_marker(self, router, tokenizer, monkeypatch):
+        # Without the flag the batch path is inert -- no marker, no counter.
+        monkeypatch.delenv("HEADROOM_NET_COST_POLICY", raising=False)
+        messages = self._convo(_tool_json(2000), _tool_json(800), filler_words=5)
+        original = [dict(m) for m in messages]
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert "router:netcost_batch_admit" not in result.transforms_applied
+        # Both deep edits still compress (no gate at all when flag off).
+        assert self._compressed(result, original, 1)
+        assert self._compressed(result, original, 2)
+
+    def test_deeper_edit_rides_free(self, router, tokenizer, monkeypatch):
+        # Slot 1 (huge shave) admits on its own merit and opens the floor;
+        # slot 2 then admits via the batch reclaim path and emits the marker.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        messages = self._convo(_tool_json(2000), _tool_json(800), filler_words=5)
+        original = [dict(m) for m in messages]
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert self._compressed(result, original, 1)
+        assert self._compressed(result, original, 2)
+        markers = [t for t in result.transforms_applied if t == "router:netcost_batch_admit"]
+        # Exactly one deeper slot rode the floor for free.
+        assert len(markers) == 1
+
+    def test_batch_admits_otherwise_blocked_edit(self, router, tokenizer, monkeypatch):
+        # Slot 2 (modest shave, large suffix after it) would be blocked on its
+        # own S, but slot 1's admit already busted the suffix -- so slot 2 rides
+        # free. Pairs with test_no_prior_admit_keeps_block, which shows the same
+        # slot-2 config stays blocked when no shallower edit opens the floor.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        messages = self._convo(_tool_json(2000), _tool_json(300), filler_words=4000)
+        original = [dict(m) for m in messages]
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert self._compressed(result, original, 1)  # floor-setting admit
+        assert self._compressed(result, original, 2)  # rode free
+        assert "router:netcost_batch_admit" in result.transforms_applied
+
+    def test_no_prior_admit_keeps_block(self, router, tokenizer, monkeypatch):
+        # Neither candidate beats its own S (both modest shaves under a huge
+        # suffix), so the floor is never opened and no slot rides free. Guards
+        # against a floor-init / off-by-one bug that would grant a free ride
+        # with no genuine shallower mutation behind it.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        messages = self._convo(_tool_json(300), _tool_json(300), filler_words=40000)
+        original = [dict(m) for m in messages]
+        result = router.apply([dict(m) for m in messages], tokenizer)
+        assert not self._compressed(result, original, 1)
+        assert not self._compressed(result, original, 2)
+        assert "router:netcost_batch_admit" not in result.transforms_applied
+
+    def test_frozen_unlock_and_batch_combine(self, router, tokenizer, monkeypatch):
+        # Two frozen string slots inside the prefix (frozen_message_count=3).
+        # Slot 1 unlocks and sets the floor; slot 2 unlocks AND rides free.
+        # Slot 2 carries both markers; the batch counter must not double-count.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        messages = self._convo(_tool_json(2000), _tool_json(800), filler_words=5)
+        original = [dict(m) for m in messages]
+        result = router.apply([dict(m) for m in messages], tokenizer, frozen_message_count=3)
+        assert self._compressed(result, original, 1)
+        assert self._compressed(result, original, 2)
+        unlocks = [t for t in result.transforms_applied if t == "router:netcost_frozen_unlock"]
+        batch = [t for t in result.transforms_applied if t == "router:netcost_batch_admit"]
+        assert len(unlocks) == 2  # both frozen slots opened
+        assert len(batch) == 1  # only the deeper one rode free -- no double-count