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

MrAshRhodes · JerrettDavis · Copilot · web-flow · commit fe4f9ee478f5 · 2026-06-18T11:15:40.000-05:00
## Description headroomlabs-ai#856 P3b (umbrella headroomlabs-ai#904), the idle-timer-compaction increment after P2 (headroomlabs-ai#905), P2b (headroomlabs-ai#944), and P3a (headroomlabs-ai#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 headroomlabs-ai#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>
diff --git a/headroom/cache/prefix_tracker.py b/headroom/cache/prefix_tracker.py
@@ -215,6 +215,24 @@ def is_expired(self) -> bool:
         """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."""
diff --git a/headroom/transforms/content_router.py b/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 @@ def _net_cost_allows(
         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 @@ def _net_cost_allows(
         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,30 +2138,46 @@ def _net_cost_allows(
             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",
         )
         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 @@ def apply(
         # 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 @@ def apply(
                         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 @@ def apply(
                         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 @@ def apply(
             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(
diff --git a/tests/test_netcost_gate.py b/tests/test_netcost_gate.py
@@ -354,3 +354,60 @@ def test_frozen_unlock_and_batch_combine(self, router, tokenizer, monkeypatch):
         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
