Reject non-integral, bool and out-of-range block-layout indices

One _index helper guards block_cell (block, y, x), slot (block_cell, q)
and map_cut_links (fluid_idx, dir) before addressing; previously e.g.
block_cell(-1, 0, 0) returned -49 and fractional values were truncated.
Adds focused rejection tests and Astra's one-block seam reproducer
(build(4,8,4,8), receiver (2,0), dir 2 -> west-halo slot 90).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/CelerisLab/lbm/blocks.py

@@ -41,6 +41,14 @@ def _readonly(values, dtype):
     return array
 
 
+def _index(value, upper, name, lower=0):
+    """Integer index or index array in [lower, upper); rejects bools and floats."""
+    array = np.asarray(value)
+    if array.dtype.kind not in "iu" or (array.size and (array.min() < lower or array.max() >= upper)):
+        raise ValueError(f"{name} must be integer index(es) in [{lower}, {upper})")
+    return array
+
+
 def _positive_int(value, name):
     if isinstance(value, (bool, np.bool_)) or not isinstance(value, Integral) or value <= 0:
         raise ValueError(f"{name} must be a positive integer")
@@ -146,8 +154,10 @@ class BlockLayout2D:
     def slot(self, block_cell, q):
         """Flat blocked population slot of block cell id(s) for direction q."""
         cells = int(np.prod(self.local_shape))
-        b, c = np.divmod(np.asarray(block_cell, dtype=np.uint64), np.uint64(cells))
-        return (b * np.uint64(NQ) + np.asarray(q, dtype=np.uint64)) * np.uint64(cells) + c
+        block_cell = _index(block_cell, self.nblocks * cells, "block_cell").astype(np.uint64)
+        q = _index(q, NQ, "q").astype(np.uint64)
+        b, c = np.divmod(block_cell, np.uint64(cells))
+        return (b * np.uint64(NQ) + q) * np.uint64(cells) + c
 
     def block_cell(self, block, y, x):
         """Address of in-grid cell (y, x) in ``block``'s own storage, core or halo.
@@ -160,15 +170,17 @@ class BlockLayout2D:
         by, bx = self.block
         ly, lx = self.local_shape
         ny, nx = self.shape
+        block = int(_index(block, self.nblocks, "block"))
+        y, x = int(_index(y, ny, "y")), int(_index(x, nx, "x"))
         y0, x0 = self.origins[block]
-        ry, rx = int(y) - int(y0) + self.halo, int(x) - int(x0) + self.halo
+        ry, rx = y - int(y0) + self.halo, x - int(x0) + self.halo
         if self.periodic[1] and not 0 <= ry < ly:
             ry = next((r for r in (ry - ny, ry + ny) if 0 <= r < ly), ry)
         if self.periodic[0] and not 0 <= rx < lx:
             rx = next((r for r in (rx - nx, rx + nx) if 0 <= r < lx), rx)
-        if not (0 <= ry < ly and 0 <= rx < lx and 0 <= y < ny and 0 <= x < nx):
+        if not (0 <= ry < ly and 0 <= rx < lx):
             raise ValueError("cell is not represented in this block's storage")
-        return int(block) * ly * lx + ry * lx + rx
+        return block * ly * lx + ry * lx + rx
 
     def _neighbor(self, y, x, dy, dx):
         ny, nx = self.shape
@@ -193,9 +205,10 @@ class BlockLayout2D:
         cells = int(np.prod(self.local_shape))
         rows, seen, writers = [], set(), set()
         for link in links:
-            r, q = int(link.fluid_idx), int(link.dir)
-            if not (0 <= r < ny * nx and 1 <= q < NQ) or (r, q) in seen:
-                raise ValueError("cut link is invalid or repeated")
+            r = int(_index(link.fluid_idx, ny * nx, "fluid_idx"))
+            q = int(_index(link.dir, NQ, "dir", lower=1))
+            if (r, q) in seen:
+                raise ValueError("cut link is repeated")
             seen.add((r, q))
             y, x = divmod(r, nx)
             receiver = int(self.owner[y, x])

tests/unit/test_block_layout.py

@@ -257,6 +257,28 @@ class BlockLayoutTests(unittest.TestCase):
         with self.assertRaises(ValueError):
             layout.map_cut_links([CutLink(fluid_idx=5 * 16 + 5, dir=0, q=.6, body_id=0)])
 
+    def test_public_indices_are_rejected_not_truncated(self):
+        layout = BlockLayout2D.build(4, 8, 4, 8)  # one block, nx=8, ny=4
+        for args in ((-1, 0, 0), (1, 0, 0), (True, 0, 0), (0.0, 0, 0), (0, 0.5, 0),
+                     (0, 1, 2.0), (0, -1, 0), (0, 4, 0), (0, 0, 8), (0, False, 0)):
+            with self.subTest(block_cell=args), self.assertRaises(ValueError):
+                layout.block_cell(*args)
+        cells = int(np.prod(layout.local_shape))
+        for args in ((-1, 1), (cells, 1), (True, 1), (1.0, 1), (0, NQ), (0, -1), (0, 1.0),
+                     (0, True), (np.array([0, cells]), 1)):
+            with self.subTest(slot=args), self.assertRaises(ValueError):
+                layout.slot(*args)
+        for idx, d in ((5.7, 1), (True, 1), (-1, 1), (32, 1), (5, 1.0), (5, 9), (5, True), (5, 0)):
+            with self.subTest(fluid_idx=idx, dir=d), self.assertRaises(ValueError):
+                layout.map_cut_links([CutLink(fluid_idx=idx, dir=d, q=.6, body_id=0)])
+
+    def test_one_block_seam_reproducer(self):
+        # Receiver (2,0) -x wraps to x=7; the write must land in the west halo.
+        layout = BlockLayout2D.build(4, 8, 4, 8)
+        mapped = layout.map_cut_links([CutLink(fluid_idx=2 * 8, dir=2, q=.6, body_id=0)])
+        lx = layout.local_shape[1]
+        self.assertEqual(int(mapped.write[0]), int(layout.slot(3 * lx + 0, 1)))
+
     def test_tables_are_readonly(self):
         layout = BlockLayout2D.build(6, 8, 3, 4)
         for array in (layout.origins, layout.owner, layout.halo_dst, layout.halo_src, layout.halo_kind):