Jets: one jet may drive links of only one body

surface_link_velocity (the shared host check used by CurvedLinkSoA,
ObjectManager sync, prepare_fine_boundary and FixedPatchStepper) now
rejects a jet whose selected cut links carry more than one body id, at
any amplitude, before uploads. Identity is the scene-wide body id of the
physical cut-link list, so one body spanning storage blocks stays valid.
No body selector; empty-selection semantics unchanged.

Tests: two bodies under one jet rejected (amplitude 2e-3 and 0) on the
Simulation/manager path and on the fine path; separate jets on separate
bodies accepted on both; one fine body crossing 2x2 block seams accepted.

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

src/CelerisLab/body/jets.py

@@ -6,7 +6,8 @@ packing), over every link in the scene; it carries no body association.
 Per-link velocity = amplitude * profile(s) * field(hit), where ``s`` in [-1, 1]
 is the region's lateral coordinate. Velocities are lattice units of the level
 whose links are evaluated; callers convert frames/units before evaluation.
-One physical link (receiver, direction) may be driven by at most one jet.
+One physical link (receiver, direction) may be driven by at most one jet, and
+one jet may drive the links of only one body (checked at assignment, any amplitude).
 
 Edge rule: the lateral bound |s| < 1 is exclusive (the profile vanishes
 there for parabolic/quartic); Sector r_min/r_max and Tube 0 <= along <= length
@@ -222,6 +223,11 @@ def surface_link_velocity(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry,
             raise ValueError("Jets require D2Q9/FP32/unshifted/double_buffer")
         shape, jet_id = jet_link_shapes(jets, hits)
         driven = jet_id >= 0
+        for j in range(len(jets)):  # scene-wide body identity, independent of storage blocks
+            bodies = np.unique(body_id[jet_id == j])
+            if len(bodies) > 1:
+                raise ValueError(f"Jet {j} selects cut links of bodies {bodies.tolist()}; "
+                                 "one jet may drive only one body")
         owners = {obj.obj_id: obj for obj in objects}
         for bid in np.unique(body_id[driven]):
             state = vars(owners[int(bid)].state)

tests/unit/test_jets.py

@@ -293,6 +293,16 @@ def test_fine_boundary_keeps_edge_frame_hits_and_checks_jets():
         refinement.prepare_fine_boundary(cfg, patch, [body], jets=[jet, jet])
     with pytest.raises(ValueError, match="need fine boundary objects"):
         refinement.prepare_fine_boundary(cfg, patch, [], jets=[jet])
+    # One fine jet over two physical bodies is rejected; one per body is accepted.
+    A, B = (8.75, 12.25), (15.75, 12.25)
+    pair = [SimObject(i, geometry=CircleGeometry(*c, 1.5), center=c, radius=1.5)
+            for i, c in enumerate((A, B))]
+    span = Jet(Tube((4.5, 12.25), 0., 15., .5), Parallel(0.))
+    with pytest.raises(ValueError, match="one jet may drive only one body"):
+        refinement.prepare_fine_boundary(cfg, patch, pair, jets=[span])
+    refinement.prepare_fine_boundary(cfg, patch, pair, jets=[
+        Jet(Sector(A, 0., .5, 2.), Radial(A, inward=False)),
+        Jet(Sector(B, math.pi, .5, 2.), Radial(B, inward=False))])
     # Jet vs legacy SurfaceActuation overlap on the fine level (root's case).
     legacy = SimObject(0, geometry=CircleGeometry(*C, R), center=C, radius=R,
                        surface_actuation=SurfaceActuation(0., .5, .002))
@@ -310,3 +320,29 @@ def test_action_upload_keeps_cfg_none_fallback(device):
     bodies._upload_action_async(stream)
     bodies.add(circle(radius=2.))
     bodies._upload_action_async(stream)
+
+
+def two_body_case(device, jets):
+    field = make_field(device)
+    bodies = manager(field, [circle(center=(3., 5.), radius=1.6),
+                             circle(center=(8.5, 5.), radius=1.6)])
+    bodies.jets.extend(jets)
+    bodies.sync_to_gpu(field)
+    return field, bodies
+
+
+@pytest.mark.parametrize("amplitude", [2e-3, 0.0])  # geometry checked even at zero amplitude
+def test_one_jet_spanning_two_bodies_rejected(device, amplitude):
+    both = Jet(Tube((0., 5.), 0., 12., .5), Parallel(0.), amplitude=amplitude)
+    with pytest.raises(ValueError, match="one jet may drive only one body"):
+        two_body_case(device, [both])
+
+
+def test_separate_jets_on_separate_bodies_accepted(device):
+    left = Jet(Sector((3., 5.), 0., .6, 2.), Radial((3., 5.), inward=False), amplitude=1e-3)
+    right = Jet(Sector((8.5, 5.), math.pi, .6, 2.), Radial((8.5, 5.), inward=False), amplitude=-1e-3)
+    field, _ = two_body_case(device, [left, right])
+    cl = field.curved
+    for j in (0, 1):
+        assert len(np.unique(cl.body_id[cl.jet_id == j])) == 1
+    assert cl.body_id[cl.jet_id == 0][0] != cl.body_id[cl.jet_id == 1][0]

tests/validation/multigrid/test_refinement_jets.py

@@ -47,6 +47,17 @@ def mocked(objects, jets, fine_block):
 
 
 class JetStepperCPU(unittest.TestCase):
+    def test_one_body_across_blocks_accepted(self):
+        # Fine circle (19.5, 15.5) R4.3 crosses the 2x2 block seams; one body, one jet.
+        import math
+        from CelerisLab.body.jets import Jet, Radial, Sector
+        C = (16.0, 12.0)
+        whole = Jet(Sector(C, .1, math.pi, 3.), Radial(C, inward=False), amplitude=0.)
+        backend, _ = mocked([circle()], [whole], FINE_BLOCKS[1])
+        _, jet_id = backend._jet_cache
+        self.assertGreater(len(np.unique(backend._fine_tables["block"][jet_id == 0])), 1)
+        backend.close()
+
     def test_blocked_jet_velocity_matches_legacy_rows(self):
         legacy, _ = mocked([circle(actuation(SPEED))], (), FINE_BLOCKS[1])
         driven, _ = mocked([circle()], [jet()], FINE_BLOCKS[1])