CP HiCache: replicated-clock SLRU eviction, collective-safe (Phase 1)

Make CP shared-KV HiCache eviction collective-free and scan-resistant by
removing every per-rank wall-clock input from the eviction/admission decisions
(any such input desyncs the must-be-replicated victim set / batch across the 8
CP ranks and can deadlock the collective-coupled writeback/load-back).

- Replicated logical clock: TreeNode.next_access_time() (a process-global
  monotonic counter, mirrors mamba/swa's get_last_access_time) replaces
  time.monotonic() as the source of last_access_time at every radix bump site
  (radix_cache __init__/match/insert; hiradix CP match/insert/_insert_helper_host;
  reset() zeroes it). creation_time/pin_expiry intentionally stay wall-clock.
  Because match/insert events are replicated (reqs broadcast from rank 0 over the
  replicated tree), last_access_time is now identical on every rank. Unique ints
  also give a strict total order (no LRU tie ambiguity). No duration arithmetic
  reads last_access_time, so non-CP LRU is unchanged; mamba/swa use their own
  TreeNode and are untouched.
- CpReplicatedSLRUStrategy = (is_protected[hit>=2], last_access_time, node.id):
  scan-resistant (cold one-shots stay probationary, evicted before reused/
  protected prefixes), recency-within-segment ages out cold (not LFU), node.id
  is the deterministic total-order tiebreak (heaps never compare TreeNodes).
  Overridden for CP so it reaches every get_priority eviction site; the host
  write-admission key _cp_host_evict_key is switched from FIFO-by-id to it.
- Co-fixes for the same per-rank-wall-clock class: pin_prefix is forbidden under
  CP (its pin_expiry victim-eligibility check is per-rank wall-clock; SLRU
  scan-resistance covers the benefit); the affinity head_age_s admission input is
  disabled (=0.0, relying on the replicated head_defer_count bound); and a
  server_args fail-fast guards CP shared-KV against SGLANG_REQ_WAITING_TIMEOUT>0
  (its waiting-queue pruning is per-rank wall-clock).

Tests: test_evict_policy.py adds TestCpReplicatedSLRUStrategy (scan-resistance,
recency, total-order, full ordering) -- 30 passed in the dev-cu13 container;
test_radix_cache_unit.py adds logical-clock determinism/total-order tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

test/registered/unit/mem_cache/test_evict_policy.py

@@ -8,6 +8,7 @@ import unittest
 from unittest.mock import MagicMock
 
 from sglang.srt.mem_cache.evict_policy import (
+    CpReplicatedSLRUStrategy,
     FIFOStrategy,
     FILOStrategy,
     LFUStrategy,
@@ -24,6 +25,7 @@ def _make_node(**kwargs):
     node.hit_count = kwargs.get("hit_count", 0)
     node.creation_time = kwargs.get("creation_time", 0.0)
     node.priority = kwargs.get("priority", 0)
+    node.id = kwargs.get("id", 0)
     return node
 
 
@@ -214,5 +216,74 @@ class TestEvictionOrdering(unittest.TestCase):
         self.assertEqual(actual, expected)
 
 
+class TestCpReplicatedSLRUStrategy(unittest.TestCase):
+    """CP shared-KV eviction strategy: scan-resistant SLRU + a strict total order.
+
+    Value = (is_protected, last_access_time, node.id). last_access_time is the
+    replicated logical clock under CP; node.id is the final tiebreak.
+    """
+
+    def setUp(self):
+        self.strategy = CpReplicatedSLRUStrategy(protected_threshold=2)
+
+    def test_priority_shape_probationary(self):
+        node = _make_node(hit_count=1, last_access_time=5.0, id=7)
+        self.assertEqual(self.strategy.get_priority(node), (0, 5.0, 7))
+
+    def test_priority_shape_protected(self):
+        node = _make_node(hit_count=2, last_access_time=5.0, id=3)
+        self.assertEqual(self.strategy.get_priority(node), (1, 5.0, 3))
+
+    def test_scan_resistance_probationary_before_protected(self):
+        # A cold one-shot (hit_count=1, even if just touched) is evicted before
+        # any reused/protected prefix (hit_count>=2, even if much older). This is
+        # what shields the hot working set from a cold long-request injector scan.
+        cold_oneshot = _make_node(hit_count=1, last_access_time=100.0, id=1)
+        hot_prefix = _make_node(hit_count=50, last_access_time=1.0, id=2)
+        self.assertLess(
+            self.strategy.get_priority(cold_oneshot),
+            self.strategy.get_priority(hot_prefix),
+        )
+
+    def test_recency_within_segment_ages_out_cold(self):
+        # Within the protected segment, the LESS-recent node is evicted first
+        # (recency, NOT frequency) -> a now-cold high-hit-count prefix ages out,
+        # which pure LFU would wrongly pin.
+        stale_high_freq = _make_node(hit_count=1000, last_access_time=1.0, id=1)
+        recent_low_freq = _make_node(hit_count=2, last_access_time=10.0, id=2)
+        self.assertLess(
+            self.strategy.get_priority(stale_high_freq),
+            self.strategy.get_priority(recent_low_freq),
+        )
+
+    def test_node_id_makes_strict_total_order(self):
+        # Nodes touched in one match can share a logical last_access_time; node.id
+        # breaks the tie so the value is a STRICT total order (heaps never fall
+        # through to comparing TreeNode objects -> deterministic across CP ranks).
+        a = _make_node(hit_count=1, last_access_time=5.0, id=3)
+        b = _make_node(hit_count=1, last_access_time=5.0, id=7)
+        self.assertNotEqual(
+            self.strategy.get_priority(a), self.strategy.get_priority(b)
+        )
+        self.assertLess(
+            self.strategy.get_priority(a), self.strategy.get_priority(b)
+        )
+
+    def test_default_threshold_is_2(self):
+        self.assertEqual(CpReplicatedSLRUStrategy().protected_threshold, 2)
+
+    def test_full_eviction_ordering(self):
+        nodes = [
+            _make_node(hit_count=5, last_access_time=1.0, id=10),  # protected, old
+            _make_node(hit_count=1, last_access_time=9.0, id=11),  # probationary, new
+            _make_node(hit_count=1, last_access_time=2.0, id=12),  # probationary, old
+            _make_node(hit_count=1, last_access_time=2.0, id=4),  # probationary, old, lower id
+            _make_node(hit_count=3, last_access_time=8.0, id=13),  # protected, new
+        ]
+        order = sorted(nodes, key=self.strategy.get_priority)
+        # probationary first (by last_access, then id), then protected (by last_access)
+        self.assertEqual([n.id for n in order], [4, 12, 11, 10, 13])
+
+
 if __name__ == "__main__":
     unittest.main()