feat(policy): unlock formula-positive deep edits through the frozen floor (#856 P2b) (#944)

## Description

Part of #904 — the **P2b (Subscription deep-unlock)** item from #856's
phased plan. Builds directly on the P2 gate (#905, now merged); rebased
onto `main` so the diff below is P2b-only
(`headroom/transforms/content_router.py` +39/−5,
`tests/test_netcost_gate.py` +72).

The P2 net-cost gate only governs mutations the router already considers
— messages **above** the `frozen_message_count` floor. The floor itself
stays a hard binary skip: anything in the provider's prefix cache is
left byte-identical no matter how compressible. That leaves the
deep-edit half of #856 on the table — e.g. a ~60K-token stale tool dump
sitting in the frozen prefix with only a small cached suffix after it,
which pays for its cache-bust many times over.

With `HEADROOM_NET_COST_POLICY=1` (default **off**), a
**string-content** frozen message now falls through to the normal
candidate pipeline instead of being skipped at the floor. The existing
P2 break-even gate then decides per candidate: **S** is the full
invalidated suffix after the slot, so the deep edit proceeds only when
`ΔT·(w+r(R−1))` still beats the cache-bust penalty. Flag off restores
byte-identical current behavior.

## Type of Change

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

## Changes Made

- Open the `frozen_message_count` floor in `ContentRouter` under
`HEADROOM_NET_COST_POLICY=1`: string-content frozen messages route to
the existing P2 gate instead of an unconditional skip; the gate's
whole-suffix S already prices the cache-bust correctly for frozen slots.
- **Scope guard:** block-list and non-string frozen content stay frozen
— the gate is wired into the string and parallel-merge paths only, and
the per-block `cache_control` contract in `_process_content_blocks` is
not net-cost aware, so opening them here would mutate cached blocks
ungated.
- Emit a `router:netcost_frozen_unlock` transform marker +
`netcost_frozen_unlocked` route count on actual unlocks, and
`netcost_frozen_considered` for every frozen string slot routed to the
gate — telemetry to validate the flag before any default-on.

## 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
15 passed in 0.96s

$ pytest tests/ -k "content_router or netcost or router" -q
137 passed, 8 skipped, 6251 deselected in 20.89s

$ 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
```

## Real Behavior Proof

- Environment: local macOS, repo .venv, Python 3.11.9, gpt-4o tokenizer
fixture
- Exact command / steps: drive `ContentRouter.apply()` with a 4-message
conversation whose index-1 `tool` message (61,584 tokens) sits inside
the frozen prefix (`frozen_message_count=2`), tiny suffix after; run
once with the flag absent and once with `HEADROOM_NET_COST_POLICY=1`
- Observed result: flag **off** → frozen tool dump left untouched, no
unlock marker; flag **on** → dump compressed (`router:smart_crusher`)
and `router:netcost_frozen_unlock` emitted, while the surrounding user
messages stay `router:protected:user_message`. The 4 new unit tests also
confirm a modest-shave / 40K-suffix frozen slot is *kept* frozen (gate
runs, `netcost:skip:` emitted, no unlock) and block-list frozen content
stays frozen.
- Not tested: live proxy traffic / real dashboard validation — deferred
to the default-on milestone per #904 (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

Net-cost economics are unchanged from P2 — this only widens *which
slots* the same gate may consider. The Subscription deep-unlock story
from #856 is realized without a mode branch: the floor is mode-agnostic
in `ContentRouter`, and the formula is the correct arbiter regardless of
auth mode. Remaining #904 items: P3a (batch deep edits) and P3b
(idle-timer compaction).

headroom/transforms/content_router.py

@@ -2226,13 +2226,34 @@ class ContentRouter(Transform):
         _PendingTask = tuple[int, str, str, float, int]
         pending_tasks: list[_PendingTask] = []
 
+        # #856 P2b (flag-gated, default off): net-cost frozen-floor unlock.
+        # Without the flag, every message in the provider's prefix cache
+        # (index < frozen_message_count) is unconditionally skipped — mutating
+        # one trades a 90% read discount for a 25% write penalty (Anthropic).
+        # That binary floor leaves money on the table: a 50K-token stale tool
+        # dump with only a 10K cached suffix after it pays for itself many
+        # times over. With HEADROOM_NET_COST_POLICY=1 a *string-content*
+        # frozen message instead falls through to the normal candidate
+        # pipeline, where the P2 break-even gate (_net_cost_allows) decides
+        # per candidate: its S is the full invalidated suffix after the slot,
+        # so the deep edit proceeds only when ΔT·(w+r(R-1)) still beats the
+        # cache-bust penalty. Block-list and non-string frozen content stay
+        # frozen — the gate is wired into the string and parallel-merge paths
+        # only, and the per-block cache_control contract in
+        # _process_content_blocks is not net-cost aware, so opening them here
+        # would mutate cached blocks ungated.
+        frozen_unlock_slots: set[int] = set()
         for i, message in enumerate(messages):
-            # Skip frozen messages (in provider's prefix cache).
-            # Modifying these would invalidate the cache, replacing a 90%
-            # read discount with a 25% write penalty (Anthropic).
             if i < frozen_message_count:
-                result_slots[i] = message
-                continue
+                if netcost_enabled and isinstance(message.get("content", ""), str):
+                    # Defer to the break-even gate below instead of skipping.
+                    frozen_unlock_slots.add(i)
+                    route_counts.setdefault("netcost_frozen_considered", 0)
+                    route_counts["netcost_frozen_considered"] += 1
+                else:
+                    # Frozen — byte-identical to preserve the prefix cache.
+                    result_slots[i] = message
+                    continue
 
             role = message.get("role", "")
             content = message.get("content", "")
@@ -2398,6 +2419,10 @@ class ContentRouter(Transform):
                         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 i in frozen_unlock_slots:
+                            transforms_applied.append("router:netcost_frozen_unlock")
+                            route_counts.setdefault("netcost_frozen_unlocked", 0)
+                            route_counts["netcost_frozen_unlocked"] += 1
                 else:
                     # Threshold tightened — no longer qualifies. Move to skip.
                     self._cache.move_to_skip(content_key)
@@ -2477,6 +2502,10 @@ class ContentRouter(Transform):
                     compressed_details.append(
                         f"{result.strategy_used.value}:{result.compression_ratio:.2f}"
                     )
+                    if slot_idx in frozen_unlock_slots:
+                        transforms_applied.append("router:netcost_frozen_unlock")
+                        route_counts.setdefault("netcost_frozen_unlocked", 0)
+                        route_counts["netcost_frozen_unlocked"] += 1
                 else:
                     # Didn't compress — add to skip set
                     self._cache.mark_skip(content_key)

tests/test_netcost_gate.py

@@ -190,3 +190,75 @@ class TestNetCostHelpers:
         assert _netcost_message_tokens({"role": "user", "content": s}, tokenizer) == (
             tokenizer.count_text(s)
         )
+
+
+def _frozen_messages(tool_content: str, suffix_filler_words: int) -> list[dict]:
+    """A short conversation whose compressible tool dump sits *inside* the
+    frozen prefix (index 1, with frozen_message_count=2)."""
+    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"},
+    ]
+
+
+class TestNetCostFrozenUnlock:
+    """#856 P2b: let formula-positive deep edits through the frozen floor."""
+
+    def test_flag_off_frozen_stays_frozen(self, router, tokenizer, monkeypatch):
+        # Default (flag off): a message in the prefix cache is never mutated,
+        # however compressible it is — the binary floor wins.
+        monkeypatch.delenv("HEADROOM_NET_COST_POLICY", raising=False)
+        messages = _frozen_messages(_tool_json(2000), suffix_filler_words=5)
+        result = router.apply([dict(m) for m in messages], tokenizer, frozen_message_count=2)
+        assert not _tool_slot_compressed(result, messages)
+        assert "router:netcost_frozen_unlock" not in result.transforms_applied
+
+    def test_flag_on_unlocks_when_shave_dominates(self, router, tokenizer, monkeypatch):
+        # Huge shave deep in the frozen zone, tiny suffix after -> the
+        # break-even gate clears the deep edit and it proceeds (the "50K
+        # stale dump, 10K suffix" user story).
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        messages = _frozen_messages(_tool_json(2000), suffix_filler_words=5)
+        result = router.apply([dict(m) for m in messages], tokenizer, frozen_message_count=2)
+        assert _tool_slot_compressed(result, messages)
+        assert "router:netcost_frozen_unlock" in result.transforms_applied
+
+    def test_flag_on_keeps_frozen_when_suffix_dominates(self, router, tokenizer, monkeypatch):
+        # Modest shave, big cached suffix -> gate runs on the unlocked slot
+        # but rejects it. The frozen message is left byte-identical and no
+        # unlock marker is emitted, proving the floor opened yet the formula
+        # still protected the cache.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        messages = _frozen_messages(_tool_json(300), suffix_filler_words=40000)
+        result = router.apply([dict(m) for m in messages], tokenizer, frozen_message_count=2)
+        assert not _tool_slot_compressed(result, messages)
+        assert "router:netcost_frozen_unlock" not in result.transforms_applied
+        assert any(t.startswith("netcost:skip:") for t in result.transforms_applied)
+
+    def test_flag_on_block_content_frozen_stays_frozen(self, router, tokenizer, monkeypatch):
+        # The gate is wired into the string and parallel-merge paths only;
+        # block-list frozen content (whose per-block cache_control contract
+        # is not net-cost aware) stays frozen even with a tiny suffix.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        big = "log line of output " * 400
+        messages = [
+            {"role": "user", "content": "fetch"},
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "tool_result",
+                        "tool_use_id": "t1",
+                        "content": [{"type": "text", "text": big}],
+                    }
+                ],
+            },
+            {"role": "user", "content": "summarize"},
+        ]
+        original = [dict(m) for m in messages]
+        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