feat(policy): decay P_alive from idle time near cache TTL (#856 P3b) (#1028)

## Description

#856 P3b (umbrella #904), the idle-timer-compaction increment after P2
(#905), P2b (#944), and P3a (#1015), all merged.

Anthropic prompt-cache entries live in a ~5-minute TTL tier (the basis
for the 1.25× write multiplier). As a session goes idle the cached
suffix approaches lapse, so **P_alive** — the probability the cache
still survives to the next turn — decays toward 0. When P_alive → 0 the
net-cost penalty term `P_alive·(w−r)·(S+ΔT)` vanishes and a deep edit
near lapse is free to make: the suffix is about to be rebuilt cold
regardless. P2/P3a fed the break-even gate a **static**
`HEADROOM_NET_COST_P_ALIVE` constant; this derives P_alive from an idle
signal when one is available.

Flag-gated under `HEADROOM_NET_COST_POLICY` (the same flag as
P2/P2b/P3a), default **off**.

## Type of Change

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

## Changes Made

- `ContentRouter.apply`: reads an optional `idle_seconds` kwarg and
derives `P_alive = max(0, 1 − idle_s / ttl)` **once per request** (idle
is a per-request property, like `frozen_message_count`), passing it to
the gate as `p_alive_override`. Absent/malformed `idle_seconds` → `None`
→ the P2 env-constant path is preserved exactly.
- `ContentRouter._net_cost_allows`: new `p_alive_override` param. When
set it replaces the `HEADROOM_NET_COST_P_ALIVE` constant (clamped to
[0,1]); otherwise unchanged. An admit made under a decayed (`< 1.0`)
idle P_alive emits the `router:netcost_idle_compaction` marker and the
`netcost_idle_admitted` counter (independent of the P3a batch marker;
both may apply).
- Cache TTL: module default 300s (Anthropic tier), overridable via
`HEADROOM_NET_COST_CACHE_TTL_SECONDS`, with malformed/non-positive
guards. Explicitly **distinct** from
`PrefixFreezeConfig.session_ttl_seconds` (tracker cleanup, 600s).
- `PrefixCacheTracker.seconds_since_activity()`: exposes the idle signal
for the proxy handlers to plumb (see Additional Notes).

## 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
25 passed in 2.03s

$ pytest tests/ -k "content_router or netcost or router or prefix_tracker or prefix" -q
245 passed, 8 skipped, 6252 deselected, 1 warning in 24.47s

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

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

$ mypy headroom/transforms/content_router.py headroom/cache/prefix_tracker.py
Success: no issues found in 2 source files
```

## Real Behavior Proof

- Environment: local macOS, repo .venv, Python 3.11.9, gpt-4o tokenizer
fixture
- Exact command / steps: drive `ContentRouter.apply()` on the P2
"blocked" baseline (a modest tool-dump shave, ΔT≈5K, under a ~120K-token
cached suffix — rejected at the default P_alive=1.0), varying only
`idle_seconds`.
- Observed result: `idle_seconds=295` (TTL 300) → P_alive≈0.017, penalty
collapses, the edit is admitted and `router:netcost_idle_compaction` is
emitted; `idle_seconds=0` → P_alive=1.0, byte-identical to the constant
baseline (still blocked, `netcost:skip:` emitted, no idle marker);
absent/malformed `idle_seconds` → env-constant path (blocked);
`HEADROOM_NET_COST_CACHE_TTL_SECONDS=60` with `idle_seconds=59` → unlock
(custom TTL controls the decay).
- Not tested: live proxy traffic — deferred to the default-on milestone
per #904 (ships default-off to gather telemetry first).

## Review Readiness

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

## Additional Notes

**Proxy wiring is a deliberate follow-up**, mirroring how P2 shipped
P_alive as an unplumbed constant and gathered telemetry before
default-on. The gate already honors `idle_seconds` via kwarg and
`PrefixCacheTracker.seconds_since_activity()` exposes the value; the
remaining step is for the provider handlers (`handlers/anthropic.py`,
`handlers/openai.py`) to pass it alongside the existing
`frozen_message_count` kwarg (`pipeline.apply` already forwards
`**kwargs` to `transform.apply`, so no pipeline change is needed). One
wiring caveat is documented on `seconds_since_activity()`:
`SessionTrackerStore.get_or_create` refreshes `_last_activity` on
access, so the handler must read idle before fetching the tracker for
the current request. Kept out of this PR for reviewability and because
it touches ~10 call sites across both providers.

---------

Co-authored-by: JD Davis <mxjerrett@gmail.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

headroom/cache/prefix_tracker.py

@@ -215,6 +215,24 @@ class PrefixCacheTracker:
         """Check if this tracker has been idle beyond TTL."""
         return (time.time() - self._last_activity) > self.config.session_ttl_seconds
 
+    def seconds_since_activity(self) -> float:
+        """Wall-clock seconds since this tracker last saw activity.
+
+        #856 P3b feeds this to the net-cost gate as an idle signal: as it
+        approaches the provider's prompt-cache TTL (~300s for Anthropic),
+        P_alive decays toward 0 and deep edits near cache lapse become free.
+        Distinct from :attr:`is_expired`, which uses the much longer
+        session-tracker *cleanup* TTL (``session_ttl_seconds``), not the cache
+        TTL.
+
+        Wiring caveat: ``SessionTrackerStore.get_or_create`` refreshes
+        ``_last_activity`` on access, so a caller that wants the idle gap
+        since the *previous turn's response* must read this before fetching
+        the tracker for the current request (or the store must capture it at
+        fetch time). ``update_from_response`` is the per-turn activity stamp.
+        """
+        return max(0.0, time.time() - self._last_activity)
+
     @property
     def stats(self) -> FreezeStats:
         """Return stats for dashboard/metrics."""

headroom/transforms/content_router.py

@@ -216,6 +216,45 @@ def _create_content_signature(
         return None
 
 
+# #856 P3b: Anthropic prompt-cache entries live in a 5-minute TTL tier (the
+# basis for the 1.25x write multiplier). As a session goes idle the cached
+# suffix approaches lapse, so P_alive — the probability the cache survives to
+# the next turn — decays toward 0. When P_alive hits 0 the net-cost penalty
+# term vanishes and a deep edit near lapse is free to make (the suffix is
+# about to be rebuilt cold anyway). This is the cache TTL, NOT the
+# session-tracker cleanup TTL (``PrefixFreezeConfig.session_ttl_seconds``).
+_NET_COST_CACHE_TTL_SECONDS = 300.0
+
+
+def _net_cost_cache_ttl_seconds() -> float:
+    """Provider cache TTL (seconds) used to decay P_alive from idle time.
+
+    Defaults to Anthropic's 5-minute tier; overridable via
+    ``HEADROOM_NET_COST_CACHE_TTL_SECONDS`` for other providers/tiers. A
+    malformed or non-positive value falls back to the default with a warning
+    rather than producing a divide-by-zero or negative TTL (same posture as
+    the other ``HEADROOM_NET_COST_*`` env guards).
+    """
+    raw = os.environ.get("HEADROOM_NET_COST_CACHE_TTL_SECONDS", "")
+    if not raw:
+        return _NET_COST_CACHE_TTL_SECONDS
+    try:
+        ttl = float(raw)
+    except ValueError:
+        logger.warning(
+            "HEADROOM_NET_COST_CACHE_TTL_SECONDS malformed; using default %s",
+            _NET_COST_CACHE_TTL_SECONDS,
+        )
+        return _NET_COST_CACHE_TTL_SECONDS
+    if not math.isfinite(ttl) or ttl <= 0.0:
+        logger.warning(
+            "HEADROOM_NET_COST_CACHE_TTL_SECONDS invalid; using default %s",
+            _NET_COST_CACHE_TTL_SECONDS,
+        )
+        return _NET_COST_CACHE_TTL_SECONDS
+    return ttl
+
+
 def _gain_bucket(gain: float) -> str:
     """Quantize a net-cost gain into a coarse magnitude band for markers.
 
@@ -2029,6 +2068,7 @@ class ContentRouter(Transform):
         route_counts: dict[str, int],
         transforms_applied: list[str],
         batch_state: dict[str, int | None] | None = None,
+        p_alive_override: float | None = None,
     ) -> bool:
         """Break-even gate for one candidate mutation (#856 P2, flag-gated).
 
@@ -2057,6 +2097,18 @@ class ContentRouter(Transform):
         mutated shallower slot. Each batch admission emits the
         ``router:netcost_batch_admit`` marker and the ``netcost_batch_admitted``
         counter for telemetry.
+
+        #856 P3b (idle-timer compaction): ``p_alive_override``, when supplied
+        by the caller, replaces the static ``HEADROOM_NET_COST_P_ALIVE``
+        constant. It is derived in ``apply`` from how long the session has
+        been idle relative to the provider cache TTL
+        (``max(0, 1 − idle_s / ttl)``). As the cached suffix nears lapse
+        P_alive → 0, the ``P_alive·(w−r)·(S+ΔT)`` penalty vanishes, and edits
+        that would lose to a warm suffix become free — the suffix is about to
+        be rebuilt cold regardless. ``None`` preserves the P2 env-constant
+        behaviour. An admit made under a decayed (``< 1.0``) idle P_alive emits
+        the ``router:netcost_idle_compaction`` marker and the
+        ``netcost_idle_admitted`` counter.
         """
         delta_t = max(0, original_tokens - compressed_tokens)
         # Batch reclaim: if a shallower slot was already admitted, its
@@ -2086,23 +2138,32 @@ class ContentRouter(Transform):
             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")
+        # #856 P3b: an idle-derived override takes precedence over the static
+        # env constant. ``net_mutation_gain`` clamps p_alive to [0, 1]
+        # internally, but clamp here too so the value logged/branched on below
+        # matches what the formula uses.
+        idle_derived = p_alive_override is not None
+        if p_alive_override is not None:
+            p_alive = min(max(p_alive_override, 0.0), 1.0)
+        else:
+            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 batch_reclaim=%s -> %s",
+            "idle_derived=%s gain=%.0f batch_reclaim=%s -> %s",
             slot_idx,
             delta_t,
             suffix,
             reads,
             p_alive,
+            idle_derived,
             gain,
             batch_reclaim,
             "mutate" if allowed else "skip",
@@ -2110,6 +2171,13 @@ class ContentRouter(Transform):
         if allowed:
             route_counts.setdefault("netcost_allowed", 0)
             route_counts["netcost_allowed"] += 1
+            if idle_derived and p_alive < 1.0:
+                # Admitted under an idle-decayed P_alive: the cached suffix is
+                # near TTL lapse, so its invalidation penalty is discounted.
+                # Independent of batch reclaim — both markers may apply.
+                route_counts.setdefault("netcost_idle_admitted", 0)
+                route_counts["netcost_idle_admitted"] += 1
+                transforms_applied.append("router:netcost_idle_compaction")
             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).
@@ -2313,12 +2381,27 @@ class ContentRouter(Transform):
         # 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}
+        # #856 P3b (idle-timer compaction): if the caller supplies how long the
+        # session has been idle, decay P_alive from it once per request and
+        # pass it to the gate. Absent/malformed → None → the gate keeps the P2
+        # 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
         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
                 )
+            idle_seconds = kwargs.get("idle_seconds")
+            if idle_seconds is not None:
+                try:
+                    idle_f = float(idle_seconds)
+                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 / ttl)
 
         # Tasks: list of (slot_index, content, context, bias, content_key)
         _PendingTask = tuple[int, str, str, float, int]
@@ -2512,6 +2595,7 @@ class ContentRouter(Transform):
                         route_counts=route_counts,
                         transforms_applied=transforms_applied,
                         batch_state=netcost_batch_state,
+                        p_alive_override=netcost_p_alive_override,
                     ):
                         # Net-cost gate: mutation would cost more in cache
                         # invalidation than it saves — leave untouched.
@@ -2594,6 +2678,7 @@ class ContentRouter(Transform):
                         route_counts=route_counts,
                         transforms_applied=transforms_applied,
                         batch_state=netcost_batch_state,
+                        p_alive_override=netcost_p_alive_override,
                     ):
                         result_slots[slot_idx] = message
                         continue
@@ -2651,6 +2736,8 @@ class ContentRouter(Transform):
             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")
+        if route_counts.get("netcost_idle_admitted"):
+            parts.append(f"{route_counts['netcost_idle_admitted']} netcost idle-admitted")
         cs = self._cache.stats
         if cs["cache_size"] > 0 or cs["cache_skip_size"] > 0:
             parts.append(

tests/test_netcost_gate.py

@@ -354,3 +354,60 @@ class TestNetCostBatchReclaim:
         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
+
+
+class TestNetCostIdleCompaction:
+    """#856 P3b: derive P_alive from idle time. As the session goes idle the
+    cached suffix nears TTL lapse, P_alive -> 0, the net-cost penalty term
+    vanishes, and edits that lose to a warm suffix become free.
+
+    Baseline shape (mirrors TestNetCostGate.test_flag_on_blocks...): a modest
+    tool-dump shave under a huge cached suffix is BLOCKED at the default
+    P_alive=1.0. These tests vary only the idle signal.
+    """
+
+    def test_idle_near_ttl_unlocks_blocked_edit(self, router, tokenizer, monkeypatch):
+        # idle ~= cache TTL (default 300s) -> P_alive ~= 0 -> penalty ~= 0 ->
+        # the otherwise-blocked deep edit is admitted and marked.
+        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, idle_seconds=295.0)
+        assert _tool_slot_compressed(result, messages)
+        assert "router:netcost_idle_compaction" in result.transforms_applied
+        assert not any(t.startswith("netcost:skip:") for t in result.transforms_applied)
+
+    def test_idle_zero_matches_constant_baseline(self, router, tokenizer, monkeypatch):
+        # idle=0 -> P_alive=1.0, identical to the env-constant default: the
+        # edit stays blocked and no idle marker is emitted.
+        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, idle_seconds=0.0)
+        assert not _tool_slot_compressed(result, messages)
+        assert "router:netcost_idle_compaction" not in result.transforms_applied
+        assert any(t.startswith("netcost:skip:") for t in result.transforms_applied)
+
+    def test_idle_absent_uses_env_constant(self, router, tokenizer, monkeypatch):
+        # No idle_seconds kwarg -> override is None -> P2 env-constant path.
+        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 "router:netcost_idle_compaction" not in result.transforms_applied
+
+    def test_malformed_idle_falls_back_to_constant(self, router, tokenizer, monkeypatch):
+        # Non-numeric idle_seconds is ignored (override stays None), so the
+        # gate keeps the constant behaviour rather than crashing the request.
+        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, idle_seconds="soon")
+        assert not _tool_slot_compressed(result, messages)
+        assert "router:netcost_idle_compaction" not in result.transforms_applied
+
+    def test_custom_ttl_env_controls_decay(self, router, tokenizer, monkeypatch):
+        # A shorter TTL makes the same idle fully decay P_alive -> unlock.
+        monkeypatch.setenv("HEADROOM_NET_COST_POLICY", "1")
+        monkeypatch.setenv("HEADROOM_NET_COST_CACHE_TTL_SECONDS", "60")
+        messages = _messages(_tool_json(300), suffix_filler_words=40000)
+        result = router.apply([dict(m) for m in messages], tokenizer, idle_seconds=59.0)
+        assert _tool_slot_compressed(result, messages)
+        assert "router:netcost_idle_compaction" in result.transforms_applied