2b dead-island GC: collapse pooled-L2 allocator to the B1 commit model

The 2b.1b adopt-then-strip left the entire option-A commit-quorum apparatus
orphaned: B1's writing_check ReduceOp.MIN frontier (mark_object_committed) is
the all-ranks-done consensus, so the per-(rank,layer,payload) gather quorum it
replaced is information-redundant (design sec 2.4 Corollary). This GC removes
the apparatus and collapses the CpSharedL2PageAllocator commit model to exactly
{_ranges_by_object, _committed_objects, _free_by_payload} -- precisely what
placement_digest() hashes -- so the cross-rank digest assert now covers the
whole model and the dead quorum is structurally unrepresentable.

Removed (all proven dead under B1, confirmed by a full tree caller-sweep +
opus adversarial review = SHIP):
- allocator: commit_layer, _has_full_commit, _expected_layers_for_object_payload,
  adopt_reserved_range, set_object_required_payloads; fields _commits_by_object,
  _expected_ranks, _adopted_ranges, _required_payloads_by_object,
  _expected_layers_by_object, _required_payloads, _expected_layers; ctor args
  expected_ranks/expected_layers/required_payloads; module fns
  broadcast_cp_shared_l2_decision, gather_cp_shared_l2_{preflight,commits,commit}.
  reserve/split_committed_object/_drop_object simplified to ranges+committed-bit.
- cache_controller: the 3 dead _commit_cp_shared_l2_* fns, 3 imports, all 5
  option-A ctor params (cp_shared_l2_{cpu_group,source_rank,broadcast_fn,
  preflight_fn,commit_fn}), the per-object commit-contract builder/setter/call.
- hiradix: allocator construction updated (3 args dropped); live _split_node
  caller and _get_cp_shared_l2_rank_and_group@167 intact.
- tests: split tests migrated to mark_object_committed; 4 dead-symbol tests removed.

Lost __init__ payload-has-slab validation is covered by reserve()'s own
ValueError (fail-loud at first write). Default (flag-off) path unchanged: the
allocator is never constructed when enable_cp_shared_physical_l2_hicache=False.
Net -580 LOC. Validated: 87/87 pool suite (syh-dev-new, torch) + 31/31 core
classes locally + edited-module import smoke. Out of scope (flagged separately):
the CpSharedL2NodeMetadata.{required_payloads,committed_payload_layers} vestige.

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

test/registered/unit/mem_cache/test_cp_shared_l2_pool.py

@@ -963,20 +963,10 @@ class TestCpSharedHostSlabPrimitives(unittest.TestCase):
 
 
 class TestCpSharedL2PageAllocator(unittest.TestCase):
-    def make_allocator(
-        self,
-        *,
-        pages=8,
-        ranks=(0, 1, 2, 3),
-        layers=(0, 1),
-        payloads=(PAYLOAD_TARGET_KV,),
-    ):
+    def make_allocator(self, *, pages=8):
         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=ranks,
-            expected_layers=layers,
-            required_payloads=payloads,
         )
 
 
@@ -1058,9 +1048,8 @@ class TestCpSharedL2PageAllocator(unittest.TestCase):
     def test_split_committed_object_preserves_live_capacity_accounting(self):
         allocator = self.make_allocator(pages=8)
         original = allocator.reserve("node-12", PAYLOAD_TARGET_KV, 4)
-        for layer in (0, 1):
-            for rank in (0, 1, 2, 3):
-                allocator.commit_layer("node-12", PAYLOAD_TARGET_KV, layer, rank)
+        # B1: commit rides the writing_check MIN frontier -> mark_object_committed.
+        allocator.mark_object_committed("node-12")
         self.assertTrue(allocator.is_committed("node-12"))
 
         parent, child = allocator.split_committed_object(
@@ -1234,11 +1223,9 @@ class TestCpSharedL2PageAllocator(unittest.TestCase):
                     }
                 ]
             },
-            expected_ranks=(0,),
-            expected_layers=(0,),
         )
         original = allocator.reserve("node-global", PAYLOAD_TARGET_KV, 4)
-        self.assertTrue(allocator.commit_layer("node-global", PAYLOAD_TARGET_KV, 0, 0))
+        allocator.mark_object_committed("node-global")
 
         parent, child = allocator.split_committed_object(
             "node-global",
@@ -1324,112 +1311,9 @@ class TestCpSharedL2PageAllocator(unittest.TestCase):
                 },
             )
 
-    def test_adopt_reserved_range_validates_known_slab_and_namespace(self):
-        allocator = _cp_shared_l2_pool.CpSharedL2PageAllocator(
-            slabs_by_payload={
-                PAYLOAD_TARGET_KV: [
-                    {
-                        "payload_kind": PAYLOAD_TARGET_KV,
-                        "slab_id": 3,
-                        "global_base_page": 50,
-                        "num_pages": 4,
-                    }
-                ]
-            }
-        )
-
-        allocator.adopt_reserved_range(
-            CpSharedL2ObjectRange(
-                object_key="adopted",
-                payload_kind=PAYLOAD_TARGET_KV,
-                slab_id=3,
-                base_page=51,
-                num_pages=2,
-                generation=9,
-            )
-        )
-        self.assertEqual(allocator.free_pages(PAYLOAD_TARGET_KV), 4)
-        with self.assertRaisesRegex(ValueError, "known slab"):
-            allocator.adopt_reserved_range(
-                CpSharedL2ObjectRange(
-                    object_key="bad-slab",
-                    payload_kind=PAYLOAD_TARGET_KV,
-                    slab_id=4,
-                    base_page=50,
-                    num_pages=1,
-                    generation=9,
-                )
-            )
-        with self.assertRaisesRegex(ValueError, "within slab"):
-            allocator.adopt_reserved_range(
-                CpSharedL2ObjectRange(
-                    object_key="bad-range",
-                    payload_kind=PAYLOAD_TARGET_KV,
-                    slab_id=3,
-                    base_page=49,
-                    num_pages=1,
-                    generation=9,
-                )
-            )
-
     def test_object_range_still_has_no_numa_node_field(self):
         self.assertNotIn("numa_node", CpSharedL2ObjectRange.__dataclass_fields__)
 
-    def test_commit_requires_all_expected_ranks_layers_and_payloads_and_is_idempotent(self):
-        allocator = self.make_allocator(
-            pages=10,
-            ranks=(0, 1),
-            layers=(0, 1),
-            payloads=(PAYLOAD_TARGET_KV, PAYLOAD_DRAFT_KV),
-        )
-        allocator.reserve("obj", PAYLOAD_TARGET_KV, 2)
-        allocator.reserve("obj", PAYLOAD_DRAFT_KV, 1)
-
-        self.assertFalse(allocator.commit_layer("obj", PAYLOAD_TARGET_KV, 0, 0))
-        self.assertFalse(allocator.is_committed("obj"))
-        self.assertFalse(allocator.commit_layer("obj", PAYLOAD_TARGET_KV, 0, 0))
-        self.assertFalse(allocator.commit_layer("obj", PAYLOAD_TARGET_KV, 0, 1))
-        self.assertFalse(allocator.commit_layer("obj", PAYLOAD_TARGET_KV, 1, 0))
-        self.assertFalse(allocator.commit_layer("obj", PAYLOAD_TARGET_KV, 1, 1))
-        self.assertFalse(allocator.commit_layer("obj", PAYLOAD_DRAFT_KV, 0, 0))
-        self.assertFalse(allocator.commit_layer("obj", PAYLOAD_DRAFT_KV, 0, 1))
-        self.assertFalse(allocator.commit_layer("obj", PAYLOAD_DRAFT_KV, 1, 0))
-
-        self.assertTrue(allocator.commit_layer("obj", PAYLOAD_DRAFT_KV, 1, 1))
-        self.assertTrue(allocator.is_committed("obj"))
-        self.assertTrue(allocator.commit_layer("obj", PAYLOAD_DRAFT_KV, 1, 1))
-
-    def test_object_commit_contract_supports_payload_specific_layer_sets(self):
-        allocator = self.make_allocator(
-            pages=10,
-            ranks=(0,),
-            layers=(0, 1),
-            payloads=(PAYLOAD_TARGET_KV,),
-        )
-        allocator.reserve("obj-draft-small", PAYLOAD_TARGET_KV, 2)
-        allocator.reserve("obj-draft-small", PAYLOAD_DRAFT_KV, 2)
-        allocator.set_object_required_payloads(
-            "obj-draft-small",
-            (PAYLOAD_TARGET_KV, PAYLOAD_DRAFT_KV),
-            expected_layers_by_payload={
-                PAYLOAD_TARGET_KV: range(2),
-                PAYLOAD_DRAFT_KV: range(1),
-            },
-        )
-
-        self.assertFalse(
-            allocator.commit_layer("obj-draft-small", PAYLOAD_TARGET_KV, 0, 0)
-        )
-        self.assertFalse(
-            allocator.commit_layer("obj-draft-small", PAYLOAD_TARGET_KV, 1, 0)
-        )
-        self.assertTrue(
-            allocator.commit_layer("obj-draft-small", PAYLOAD_DRAFT_KV, 0, 0)
-        )
-        self.assertTrue(allocator.is_committed("obj-draft-small"))
-        with self.assertRaisesRegex(ValueError, "unexpected layer_id"):
-            allocator.commit_layer("obj-draft-small", PAYLOAD_DRAFT_KV, 1, 0)
-
     def test_same_node_helper_accepts_local_and_failfasts_for_remote_rank(self):
         self.assertEqual(
             _cp_shared_l2_pool.require_cp_shared_l2_same_node(
@@ -1486,27 +1370,6 @@ class TestCpSharedL2PageAllocator(unittest.TestCase):
         self.assertEqual(calls[0], ("get_global_rank", "cp-group", 0))
         self.assertEqual(calls[1][0:3], ("broadcast", 5, "cp-group"))
 
-    def test_broadcast_helpers_use_injected_function_without_distributed_init(self):
-        calls = []
-
-        def fake_broadcast(object_list, src, group):
-            calls.append((src, group, list(object_list)))
-            if object_list[0] is None:
-                object_list[0] = {"decision": "commit", "object_key": "obj"}
-
-        decision_rank0 = _cp_shared_l2_pool.broadcast_cp_shared_l2_decision(
-            {"decision": "commit", "object_key": "obj"},
-            cp_cpu_group="group",
-            rank=0,
-            broadcast_fn=fake_broadcast,
-        )
-        decision_rank1 = _cp_shared_l2_pool.broadcast_cp_shared_l2_decision(
-            None, cp_cpu_group="group", rank=1, broadcast_fn=fake_broadcast
-        )
-
-        self.assertEqual(decision_rank0["decision"], "commit")
-        self.assertEqual(decision_rank1["object_key"], "obj")
-        self.assertEqual(len(calls), 2)
 
 
 
@@ -2716,9 +2579,6 @@ class TestCpSharedL2PlacementDigest(unittest.TestCase):
         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):
@@ -2793,9 +2653,6 @@ class TestCpSharedL2MarkObjectCommitted(unittest.TestCase):
         return _cp_shared_l2_pool.CpSharedL2PageAllocator(
             pages_per_payload={PAYLOAD_TARGET_KV: pages},
             slab_ids_by_payload={PAYLOAD_TARGET_KV: 10},
-            expected_ranks=(0, 1, 2, 3),
-            expected_layers=(0, 1),
-            required_payloads=(PAYLOAD_TARGET_KV,),
         )
 
     def test_mark_commits_without_per_layer_quorum(self):
@@ -2836,8 +2693,8 @@ class TestCpSharedL2MarkObjectCommitted(unittest.TestCase):
         self.assertEqual(a.placement_digest(), b.placement_digest())
 
     def test_split_of_marked_committed_object_is_coherent(self):
-        # MF4: split a B1-committed object (no per-layer commit facts) -> children
-        # are committed, pages preserved, no error from the empty _commits_by_object.
+        # B1: split a marked-committed object -> children are committed, pages
+        # preserved (the allocator carries only ranges + the committed bit).
         a = self.make_allocator()
         a.reserve("child", PAYLOAD_TARGET_KV, 6)
         a.mark_object_committed("child")
@@ -2877,9 +2734,6 @@ class TestCpSharedL2EightRankReserveDeterminism(unittest.TestCase):
             _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)
         ]