CP HiCache: expose L2 + L3 usage metrics (the mainline storage metrics are dead under CP)

Under CP, enable_storage=False, so StorageMetricsCollector (prefetch/backup tokens + bandwidth) never
emits, and L2 pool usage was debug-log-only (_cp_maybe_log_lane_stats) while L3 was invisible. Add a
CpHiCacheMetricsCollector covering both, gated on enable_metrics alone, per-rank via the tp_rank label.

- CpL3Store.stats(): per-payload slot occupancy (used,total) + cumulative counters (spill_pages,
  reload_pages, gc_reclaimed, write_errors -- incremented on the bg threads) + instantaneous inflight
  (spill/reload) + the four bg queue depths. Counters are monotonic (NOT reset on clear()).
- CpHiCacheMetricsCollector (metrics_collector.py): gauges for instantaneous state
  (sglang:cp_l2_used/total_pages, cp_l3_used/total_slots, cp_l3_inflight_*, cp_l3_queue_depth) +
  counters for cumulative work (cp_l2_objects_committed/evicted_total, cp_l3_spill/reload_pages_total,
  cp_l3_gc_reclaimed_total, cp_l3_write_errors_total). Counters fed via a monotonic .inc(delta) off the
  store's cumulative totals.
- hiradix: create the collector in _maybe_init_cp_l3 (once, per-rank labels from cache_controller);
  push periodically (~1s, per-rank wall-clock -- metrics need no rank-uniformity) from
  check_hicache_events via _cp_maybe_collect_metrics, reading allocator.occupancy_by_payload()/stats()
  + store.stats(). Off the data path.

store.stats() unit-tested (occupancy + spill/reload counters; 10 store tests green). The collector +
wiring are correct-by-pattern (mirror StorageMetricsCollector) and validate on the live server (3.4;
prometheus_client not in the unit env). Directly useful for 3.4: watch L2/L3 fill + GC + reload hits
under cachebench.

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

test/registered/unit/mem_cache/test_cp_l3_store.py

@@ -272,6 +272,26 @@ class TestCpL3Store(unittest.TestCase):
         self.assertEqual(store.exists_prefix([_h(4000 + 0)], ["target_kv"]), 1)  # touched -> warm -> survived
         self.assertEqual(store.exists_prefix([_h(4000 + 1)], ["target_kv"]), 0)  # next-coldest -> reclaimed
 
+    def test_stats_reports_occupancy_and_counters(self):
+        # stats() snapshot: per-payload slot occupancy + cumulative spill/reload/GC counters + inflight/queues.
+        pages = {"target_kv": [(2, _h(2)), (5, _h(5))]}
+        self.store.submit_spill("o", pages, last_access=10)
+        self.assertEqual(self._finish_spill(), [(1, True)])
+        s = self.store.stats()
+        self.assertEqual(s["occupancy"]["target_kv"], (2, self.store.pools["target_kv"].num_slots))
+        self.assertEqual(s["spill_pages_total"], 2)
+        self.assertEqual(s["gc_reclaimed_total"], 0)
+        self.assertEqual(s["ongoing_spill"], 0)
+        self.assertEqual(s["ongoing_reload"], 0)
+        for k in ("reload_pages_total", "write_errors_total", "gather_qsize", "write_qsize",
+                  "reload_qsize", "touch_qsize"):
+            self.assertIn(k, s)
+        # reload the 2 pages -> reload_pages_total advances
+        self.store.submit_reload("o", pages)
+        self.assertTrue(_wait_ack(self.store.ack_reload_qsize, 1))
+        self.store.drain_reload_acks(self.store.ack_reload_qsize())
+        self.assertEqual(self.store.stats()["reload_pages_total"], 2)
+
     def test_clear_resets(self):
         self.store.submit_spill("o", {"target_kv": [(0, _h(0))]}, last_access=1)
         self._finish_spill()