Multipatch: fine_objects and jets on several patches (level-wide projection)

prepare_fine_boundary accepts a FixedPatchSet2D: objects are projected once
onto the level-wide fine grid (FixedPatchSet2D.fine_origin = (0, 0), same
coarse cell-edge -> fine cell transform), each circle must lie inside one
patch, coarse transfer cells are checked against the covered union, and body
ids stay object positions, so the existing jet one-body-per-jet, overlap and
(receiver, dir) writer checks run once over the whole level (caller obj_id
cannot bypass them). The stepper no longer rejects fine_objects/jets with
several patches; blocked link tables, BFL and the jet amplitude cache are
reused unchanged. No kernel change. README/docstring scope updated.

Tests: CPU translation/ids/rejections (outside one patch, one jet over two
bodies at zero amplitude); GPU one body via the new path == single-patch path
(both modes), two bodies (+jets, amplitude switch) == tiled single (bitwise).

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

tests/unit/test_refinement.py

@@ -233,6 +233,56 @@ class FixedPatchSetTests(unittest.TestCase):
                 self.build(self.FACE).fine_layout(block)
 
 
+class PatchSetBoundaryTests(unittest.TestCase):
+    """fine_objects/jets on a level-wide patch set (16x32, patches 4 cells apart)."""
+    PATCHES = ((2, 2, 14, 14), (18, 2, 30, 14))
+
+    @staticmethod
+    def body(x, y, obj_id=0):
+        from CelerisLab.body.geometry.circle import CircleGeometry
+        from CelerisLab.body.objects import SimObject
+        return SimObject(obj_id, geometry=CircleGeometry(x, y, 2.15), center=(x, y), radius=2.15)
+
+    def prepare(self, nx, rects, objects, jets=()):
+        cfg = LBMConfig(nx=nx, ny=16, viscosity=.1)
+        s = refinement.FixedPatchSet2D.from_config(cfg, rects)
+        return s, refinement.prepare_fine_boundary(cfg, s, objects, jets=jets)
+
+    def test_two_bodies_are_the_one_body_case_translated(self):
+        _, one = self.prepare(16, self.PATCHES[:1], [self.body(8., 8.)])
+        _, two = self.prepare(32, self.PATCHES, [self.body(8., 8., 7), self.body(24., 8., 7)])
+        f1, f2 = one.flags.reshape(32, 32), two.flags.reshape(32, 64)
+        np.testing.assert_array_equal(f2[:, :32], f1)
+        np.testing.assert_array_equal(f2[:, 32:], f1)
+        n = len(one.curved_host[0])
+        self.assertEqual(len(two.curved_host[0]), 2 * n)
+        body_id = two.curved_host[3]
+        self.assertEqual(sorted(set(body_id.tolist())), [0, 1])  # positions, not obj_id 7
+        # Translation must reproduce identical link geometry (prerequisite of the
+        # bitwise GPU tiling gate): dirs, q and fallback equal, receivers shifted.
+        fidx, dirs, q, fallback = (two.curved_host[i] for i in (0, 1, 2, 7))
+        for column in (dirs, q, fallback):
+            np.testing.assert_array_equal(column[:n], column[n:])
+        y, x = np.divmod(fidx.astype(np.int64), 64)
+        np.testing.assert_array_equal(y[:n], y[n:])
+        np.testing.assert_array_equal(x[:n] + 32, x[n:])
+
+    def test_body_outside_a_single_patch_rejected(self):
+        for x in (16., 13.):  # in the gap / crossing the patch edge
+            with self.subTest(x=x), self.assertRaises(ValueError):
+                self.prepare(32, self.PATCHES, [self.body(x, 8.)])
+
+    def test_one_jet_over_two_bodies_rejected_even_at_zero_amplitude(self):
+        from CelerisLab.body.jets import Jet, Parallel, Tube
+        bodies = [self.body(8., 8.), self.body(24., 8.)]
+        span = Jet(Tube((5., 8.), 0., 22., .6), Parallel(0.), amplitude=0.)
+        with self.assertRaisesRegex(ValueError, "one jet"):
+            self.prepare(32, self.PATCHES, bodies, jets=[span])
+        own = [Jet(Tube((5., 8.), 0., 6., .6), Parallel(0.), amplitude=.002),
+               Jet(Tube((21., 8.), 0., 6., .6), Parallel(0.), amplitude=.002)]
+        self.prepare(32, self.PATCHES, bodies, jets=own)  # one jet per body: accepted
+
+
 class NestPatchTests(unittest.TestCase):
     """Three-level single chain: L2 bounds in L1 window cells (probe fixture)."""
     def setUp(self):

tests/validation/multigrid/test_refinement_multipatch.py

@@ -53,6 +53,13 @@ class MultipatchCPU(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "closer"):
             FixedPatchStepper(cfg, ((4, 4, 8, 12), (10, 4, 14, 12)), context=context,
                               fine_block=(8, 4), base_block=(4, 4))
+        from CelerisLab.body.jets import Jet, Parallel, Tube
+        cfg32 = LBMConfig(nx=32, ny=16, viscosity=.1)
+        span = Jet(Tube((5., 8.), 0., 22., .6), Parallel(0.), amplitude=0.)
+        with self.assertRaisesRegex(ValueError, "one jet"):  # same obj_id cannot bypass it
+            FixedPatchStepper(cfg32, ((2, 2, 14, 14), (18, 2, 30, 14)), context=context,
+                              fine_block=(4, 4), base_block=(4, 4),
+                              fine_objects=[body(8., 8.), body(24., 8.)], jets=[span])
 
 
 class MultipatchGPU(unittest.TestCase):
@@ -144,5 +151,76 @@ class MultipatchGPU(unittest.TestCase):
             print(f"face gap4 step {step}: weighted leaf mass {mass:.9g} (diagnostic)", flush=True)
 
 
+def body(x, y, obj_id=0):
+    from CelerisLab.body.geometry.circle import CircleGeometry
+    from CelerisLab.body.objects import SimObject
+    return SimObject(obj_id, geometry=CircleGeometry(x, y, 2.15), center=(x, y), radius=2.15)
+
+
+class MultipatchBodiesGPU(unittest.TestCase):
+    """fine_objects/jets on several patches (frozen gates, bitwise)."""
+    PATCHES = ((2, 2, 14, 14), (18, 2, 30, 14))
+    STEPS = (0, 1, 2, 8)
+
+    def run_steps(self, context, cfg, bounds, coarse, fine, switch=None, **kw):
+        out = []
+        with FixedPatchStepper(cfg, bounds, context=context, **kw) as backend:
+            backend.initialize(coarse, fine)
+            for step in self.STEPS:
+                if switch is not None and step == 8:
+                    for index, amplitude in switch:
+                        backend.set_jet_amplitude(index, amplitude)
+                backend.step(step - backend.step_count)
+                out.append(backend.download() + raw(backend))
+        return out
+
+    def test_one_body_matches_single_patch_path(self):
+        from CelerisLab.cuda.context import CudaContext
+        with CudaContext(0) as context:
+            for mode in ("periodic", "slot_reservoir"):
+                cfg = config(mode)
+                p = FixedPatch2D.from_config(cfg, BOUNDS)
+                s = FixedPatchSet2D.from_config(cfg, (BOUNDS,))
+                coarse, fine = state(NY, NX, 1), state(*p.fine_shape, 2)
+                kw = dict(coarse_boundary=mode, fine_block=(8, 4), base_block=(3, 4),
+                          fine_objects=[circle()])
+                old = self.run_steps(context, cfg, BOUNDS, coarse, fine, **kw)
+                new = self.run_steps(context, cfg, (BOUNDS,), coarse, level_fine(s, [fine]), **kw)
+                for step, a, b in zip(self.STEPS, old, new):
+                    label = f"{mode} step {step}"
+                    exact(self, b[0], a[0], label + " coarse")
+                    exact(self, window(s, 0, b[1]), a[1], label + " fine")
+                    exact(self, window(s, 0, b[3]), a[3], label + " raw fine")
+
+    def test_two_bodies_and_jets_equal_tiled_single(self):
+        """Translation copy across a block seam, with jets and an amplitude switch."""
+        from CelerisLab.body.jets import Jet, Parallel, Tube
+        from CelerisLab.cuda.context import CudaContext
+        cfg16 = LBMConfig(nx=16, ny=16, viscosity=.1)
+        cfg32 = replace(cfg16, nx=32)
+        one = FixedPatchSet2D.from_config(cfg16, self.PATCHES[:1])
+        two = FixedPatchSet2D.from_config(cfg32, self.PATCHES)
+        coarse, fine = state(16, 16, 3), state(*one.patches[0].fine_shape, 4)
+        jet = lambda x: Jet(Tube((x, 8.), 0., 6., .6), Parallel(0.), amplitude=.002)
+        kw = dict(fine_block=(4, 4), base_block=(4, 4))
+        with CudaContext(0) as context:
+            for jets in ((), (jet(5.),)):
+                ref = self.run_steps(context, cfg16, self.PATCHES[:1], coarse, level_fine(one, [fine]),
+                                     switch=[(0, -.001)] if jets else None,
+                                     fine_objects=[body(8., 8.)], jets=jets, **kw)
+                both = jets and (jet(5.), jet(21.))
+                got = self.run_steps(context, cfg32, self.PATCHES, np.concatenate([coarse, coarse], 2),
+                                     level_fine(two, [fine, fine]),
+                                     switch=[(0, -.001), (1, -.001)] if jets else None,
+                                     fine_objects=[body(8., 8., 5), body(24., 8., 5)],
+                                     jets=both or (), **kw)
+                for step, a, b in zip(self.STEPS, ref, got):
+                    label = f"jets={len(jets)} step {step}"
+                    exact(self, b[0], np.concatenate([a[0], a[0]], axis=2), label + " coarse")
+                    for i in (0, 1):
+                        exact(self, window(two, i, b[1]), window(one, 0, a[1]), f"{label} fine {i}")
+                        exact(self, window(two, i, b[3]), window(one, 0, a[3]), f"{label} raw fine {i}")
+
+
 if __name__ == "__main__":
     unittest.main()