2b.1a: placement_digest gate on CpSharedL2PageAllocator (B1 proof obligation)

Add CpSharedL2PageAllocator.placement_digest() -- a deterministic, order-independent SHA-256 of the full replicated allocator state (free list + per-object ranges + committed set). This is the B1 proof-obligation gate from the 2b design (sec 2.4, Theorem 1): the write-through (2b.1) cross-rank assert hashes this each tick and EQ-checks it across CP ranks, turning placement-determinism into a fail-loud runtime invariant -- any reserve/release fired on a subset of ranks or with a per-rank arg changes the digest.

6 unit tests (pure-Python, no torch): determinism, changes-on-reserve, identical-op-sequence => identical-digest, divergent-op => different-digest, reserve+release restores digest (free-list coalescing), placement-order sensitivity. Full suite 81/81 green on g0033 syh-dev-new.

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

python/sglang/srt/mem_cache/cp_shared_l2_pool.py

@@ -10,6 +10,7 @@ orchestration are intentionally left to later phases.
 from __future__ import annotations
 
 import ctypes
+import hashlib
 import inspect
 import mmap
 import operator
@@ -950,6 +951,37 @@ class CpSharedL2PageAllocator:
             for _, length in free_by_slab
         )
 
+    def placement_digest(self) -> str:
+        """Deterministic content hash of the full replicated allocator state.
+
+        The B1 proof-obligation gate (see docs_internal/cp_hicache_2b_pooled_l2_b1_design.md
+        sec 2.4, Theorem 1): two allocators that applied the IDENTICAL op sequence
+        over identical initial state produce identical digests on every CP rank, so
+        the runtime cross-rank assert can prove placement determinism each tick. Any
+        per-rank divergence -- a reserve/release fired on a subset of ranks, or with
+        a per-rank arg -- changes the digest. Order-independent within each container
+        (sorted) so dict/set iteration order cannot perturb it. Hashes the durable
+        placement state: the free list, the per-object ranges, and the committed set.
+        """
+        h = hashlib.sha256()
+        for payload in sorted(self._free_by_payload):
+            h.update(b"P" + payload.encode())
+            for slab_id in sorted(self._free_by_payload[payload]):
+                h.update(b"|s" + str(slab_id).encode() + b":")
+                for base, length in sorted(self._free_by_payload[payload][slab_id]):
+                    h.update(f"{base},{length};".encode())
+        for object_key in sorted(self._ranges_by_object):
+            ranges = self._ranges_by_object[object_key]
+            for payload in sorted(ranges):
+                r = ranges[payload]
+                h.update(
+                    f"#{object_key}/{payload}={r.slab_id},{r.base_page},"
+                    f"{r.num_pages};".encode()
+                )
+        for object_key in sorted(self._committed_objects):
+            h.update(b"@" + object_key.encode() + b";")
+        return h.hexdigest()
+
     def _allocate_contiguous(self, payload_kind: str, num_pages: int) -> tuple[int, int]:
         for slab in self._slabs_by_payload[payload_kind]:
             free_ranges = self._free_by_payload[payload_kind][slab.slab_id]

test/registered/unit/mem_cache/test_cp_shared_l2_pool.py

@@ -2707,5 +2707,82 @@ class TestPatchDSharedHostTensorAllocator(unittest.TestCase):
             self.assertEqual(list(inspect.signature(cls.load_to_device_per_layer).parameters), expected)
 
 
+class TestCpSharedL2PlacementDigest(unittest.TestCase):
+    """B1 proof-obligation gate (2b design sec 2.4, Theorem 1): placement_digest is a
+    deterministic, replicated fingerprint of allocator state -- identical op sequence
+    => identical digest on every rank; any divergence => different digest."""
+
+    def make_allocator(self, pages=16):
+        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=(0, 1, 2, 3),
+            expected_layers=(0, 1),
+            required_payloads=(PAYLOAD_TARGET_KV,),
+        )
+
+    def _apply(self, allocator, ops):
+        for op in ops:
+            if op[0] == "reserve":
+                allocator.reserve(op[1], op[2], op[3])
+            elif op[0] == "release":
+                allocator.release(op[1])
+
+    def test_digest_is_deterministic(self):
+        a = self.make_allocator()
+        a.reserve("x", PAYLOAD_TARGET_KV, 3)
+        self.assertEqual(a.placement_digest(), a.placement_digest())
+
+    def test_digest_changes_on_reserve(self):
+        a = self.make_allocator()
+        before = a.placement_digest()
+        a.reserve("x", PAYLOAD_TARGET_KV, 3)
+        self.assertNotEqual(before, a.placement_digest())
+
+    def test_identical_op_sequence_yields_identical_digest(self):
+        # The core B1 property: two ranks running the same reserve/release sequence
+        # over identical init reach byte-identical allocator state (Theorem 1).
+        ops = [
+            ("reserve", "a", PAYLOAD_TARGET_KV, 3),
+            ("reserve", "b", PAYLOAD_TARGET_KV, 2),
+            ("release", "a"),
+            ("reserve", "c", PAYLOAD_TARGET_KV, 4),
+        ]
+        a, b = self.make_allocator(), self.make_allocator()
+        self._apply(a, ops)
+        self._apply(b, ops)
+        self.assertEqual(a.placement_digest(), b.placement_digest())
+
+    def test_divergent_op_breaks_digest(self):
+        # A reserve fired on one rank but not the other -> digests diverge: exactly
+        # the per-rank divergence the runtime cross-rank assert must catch.
+        a, b = self.make_allocator(), self.make_allocator()
+        a.reserve("a", PAYLOAD_TARGET_KV, 3)
+        b.reserve("a", PAYLOAD_TARGET_KV, 3)
+        a.reserve("extra", PAYLOAD_TARGET_KV, 1)  # divergence
+        self.assertNotEqual(a.placement_digest(), b.placement_digest())
+
+    def test_reserve_then_release_restores_digest(self):
+        # Free-list coalescing is symmetric: reserve+release of the same range
+        # returns to the initial state (and digest), proving _return_range merges.
+        a = self.make_allocator()
+        initial = a.placement_digest()
+        a.reserve("tmp", PAYLOAD_TARGET_KV, 4)
+        a.release("tmp")
+        self.assertEqual(initial, a.placement_digest())
+
+    def test_digest_reflects_placement_order(self):
+        # Reserve a-then-b vs b-then-a yields different placement (first-fit), so
+        # digests differ -- this is WHY B1 requires the reserve order be replicated
+        # across ranks. Confirms the digest captures placement, not dict order.
+        a = self.make_allocator()
+        a.reserve("a", PAYLOAD_TARGET_KV, 2)
+        a.reserve("b", PAYLOAD_TARGET_KV, 2)
+        b = self.make_allocator()
+        b.reserve("b", PAYLOAD_TARGET_KV, 2)
+        b.reserve("a", PAYLOAD_TARGET_KV, 2)
+        self.assertNotEqual(a.placement_digest(), b.placement_digest())
+
+
 if __name__ == "__main__":
     unittest.main()