2b.1b: B1 collective-free pooled-L2 write-through (default-off)

The B1 write-through: every CP rank runs the IDENTICAL deterministic CpSharedL2PageAllocator.reserve over the replicated event stream -> identical placement with NO broadcast/gather; commit rides the writing_check ReduceOp.MIN frontier (mark_object_committed) instead of a per-(rank,layer,payload) gather quorum.

cache_controller.py: _reserve_write_cp_shared_l2 stripped to every-rank reserve (dropped preflight#2/rank0-gate/broadcast#3/adopt/missing-payloads; per-object contract kept MF3; idempotent abort-on-partial SF3); _submit_write_cp_layer_states drops the 3 gather-commit calls (per-layer D2H + all-payload done-gate MF1 + per-node ack kept); submit_write_cp_all_layer drops both fallback gathers (zero-owned keeps its ack MF2).

hiradix_cache.py: 3-way merge of l2_pooling shared-L2 init (clean, 0-conflict; Phase-1 clock + E1 + reserve-budget preserved) builds the allocator/slab when the flag is on and passes it to HiCacheController WITHOUT collective fns. The #5 deadlock removed: shared_l2_capacity reserve-miss now SKIPS the backup (reactive, rank-uniform) instead of _evict_host_for_physical_slots(synchronize_across_ranks=True) -- the deterministic shared-pool evict is 2b.2. mark_object_committed wired at _commit_pending_backup (the MIN commit). New _cp_assert_placement_replicated (SGLANG_CP_HICACHE_PLACEMENT_ASSERT) in check_hicache_events: MIN/MAX-reduce placement_digest across the CP group, fail-loud on divergence (Theorem 1 runtime gate; rank-uniform entry, cannot deadlock).

environ.py: SGLANG_CP_HICACHE_PLACEMENT_ASSERT (EnvBool, default off).

Tests: 8-rank reserve-determinism (identical placement across 8 ranks + divergence-detectable) + full pool suite 91/91 + import smoke on g0033 syh-dev-new. TWO opus adversarial reviews both SHIP (cache_controller strip + hiradix wiring).

NOTES: default-off (--enable-cp-shared-physical-l2-hicache); validated for write_through policy (prod default). Capacity-miss currently SKIPS (B1 shared-pool evict = 2b.2) so the flag must stay off in real runs until 2b.2. Dead gather island (3 fns + 3 imports, opus-confirmed inert) pending GC follow-up. Design+proof: docs_internal/cp_hicache_2b_pooled_l2_b1_design.md sec 2.4/9.

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

python/sglang/srt/environ.py

@@ -389,6 +389,10 @@ class Envs:
     # during bring-up so E1 measures the baseline (reactive drops, not waits);
     # flip on once validated (E3). Opt-in.
     SGLANG_CP_HICACHE_RESERVE_ADMISSION = EnvBool(False)
+    # Phase 2b (B1): runtime cross-rank assert that the pooled-L2 allocator placement
+    # is replicated (Theorem 1). When on, MIN/MAX-reduce placement_digest across the
+    # CP group each tick and fail loud on divergence. On in CI/bring-up, off in prod.
+    SGLANG_CP_HICACHE_PLACEMENT_ASSERT = EnvBool(False)
 
     # Mooncake KV Transfer
     SGLANG_MOONCAKE_CUSTOM_MEM_POOL = EnvStr(None)

test/registered/unit/mem_cache/test_cp_shared_l2_pool.py

@@ -2862,5 +2862,69 @@ class TestCpSharedL2MarkObjectCommitted(unittest.TestCase):
         self.assertFalse(a.is_committed("n"))
 
 
+class TestCpSharedL2EightRankReserveDeterminism(unittest.TestCase):
+    """B1 (2b.1b) write-through end-to-end determinism: every CP rank runs the
+    IDENTICAL deterministic reserve over the replicated event stream, so all 8
+    ranks reach identical placement with NO broadcast (the controller's
+    _reserve_write_cp_shared_l2 strips to exactly this loop). Models 8 ranks as 8
+    independent allocators applying the same reserve/commit/release sequence --
+    the runtime placement_digest assert is the production analog of this check."""
+
+    CP = 8
+
+    def make_rank_allocators(self, pages=64):
+        return [
+            _cp_shared_l2_pool.CpSharedL2PageAllocator(
+                pages_per_payload={PAYLOAD_TARGET_KV: pages, PAYLOAD_DRAFT_KV: pages},
+                slab_ids_by_payload={PAYLOAD_TARGET_KV: 10, PAYLOAD_DRAFT_KV: 11},
+                expected_ranks=range(self.CP),
+                expected_layers=(0, 1),
+                required_payloads=(PAYLOAD_TARGET_KV,),
+            )
+            for _ in range(self.CP)
+        ]
+
+    def _reserve_on_all(self, allocs, node_id, payloads, num_pages):
+        # the B1 every-rank reserve loop, run identically on all ranks
+        for a in allocs:
+            for p in payloads:
+                a.reserve(f"cp_hicache_node:{node_id}", p, num_pages)
+
+    def test_eight_ranks_reach_identical_placement(self):
+        allocs = self.make_rank_allocators()
+        for nid, (payloads, npages) in enumerate(
+            [
+                ((PAYLOAD_TARGET_KV,), 3),
+                ((PAYLOAD_TARGET_KV, PAYLOAD_DRAFT_KV), 2),
+                ((PAYLOAD_TARGET_KV,), 5),
+            ]
+        ):
+            self._reserve_on_all(allocs, nid, payloads, npages)
+        self.assertEqual(
+            len({a.placement_digest() for a in allocs}),
+            1,
+            "all 8 ranks must compute identical placement with no broadcast",
+        )
+
+    def test_commit_then_release_stays_identical(self):
+        allocs = self.make_rank_allocators()
+        self._reserve_on_all(allocs, 0, (PAYLOAD_TARGET_KV,), 4)
+        for a in allocs:  # MIN-commit fires on every rank with the same object_key
+            a.mark_object_committed("cp_hicache_node:0")
+        self.assertEqual(len({a.placement_digest() for a in allocs}), 1)
+        for a in allocs:  # deterministic evict on every rank
+            a.release("cp_hicache_node:0")
+        self.assertEqual(len({a.placement_digest() for a in allocs}), 1)
+
+    def test_one_rank_divergence_is_detectable(self):
+        # A reserve that fires on one rank but not the others (a non-replicated
+        # trigger) makes the placement_digests diverge -- exactly what the runtime
+        # cross-rank assert catches.
+        allocs = self.make_rank_allocators()
+        self._reserve_on_all(allocs, 0, (PAYLOAD_TARGET_KV,), 3)
+        allocs[3].reserve("cp_hicache_node:1", PAYLOAD_TARGET_KV, 1)  # rank 3 only
+        self.assertGreater(len({a.placement_digest() for a in allocs}), 1)
+
+
 if __name__ == "__main__":
     unittest.main()