fix: validate passive boundary ownership and expose Jet exchange

tests/unit/test_state_regressions.py

@@ -238,6 +238,81 @@ def test_overlapping_force_regions_rejected(device):
         bodies.build_compact_lists(bodies.build_flags(field.flag))
 
 
+@pytest.mark.parametrize("second_x,error", [(14., "unique cut-link writers"),
+                                           (17., "active fluid receivers")])
+def test_passive_overlap_rejected_before_publishing_lists(second_x, error):
+    cfg = LBMConfig(nx=32, ny=24)
+    cfg.validate()
+    bodies = ObjectManager(cfg.nx, cfg.ny, cfg.nz, cfg.nq, cfg)
+    for x in (14., second_x):
+        bodies.add(obj(x=x, y=12., r=4.3))
+    bodies.add(obj(x=4., y=4., r=2., is_sensor=True))
+    counts = np.full(bodies.count, 91, np.int32)
+    hits = []
+    flags = bodies.build_flags(np.full(cfg.nx * cfg.ny, FLUID, np.uint16))
+    for domain_flags in (flags, None):
+        with pytest.raises(ValueError, match=error):
+            bodies.build_compact_lists_for(bodies.objects, domain_flags,
+                                          sensor_cell_counts_out=counts, hits_out=hits)
+        np.testing.assert_array_equal(counts, 91)
+        assert hits == []
+
+
+@pytest.mark.parametrize("centers", [((10., 12.), (22., 12.)),
+                                    ((10.5, 12.), (19.5, 12.))])
+def test_separated_passive_bodies_keep_valid_writers(centers):
+    # The second case has a subcell physical gap with a resolved fluid column.
+    cfg = LBMConfig(nx=32, ny=24)
+    cfg.validate()
+    bodies = ObjectManager(cfg.nx, cfg.ny, cfg.nz, cfg.nq, cfg)
+    for x, y in centers:
+        bodies.add(obj(x=x, y=y, r=4.3))
+    flags = bodies.build_flags(np.full(cfg.nx * cfg.ny, FLUID, np.uint16))
+    indices, directions, _, ids, *_ = bodies.build_compact_lists(flags)
+    assert set(ids) == {0, 1}
+    assert np.all(flags[indices] & FLUID) and not np.any(flags[indices] & SOLID)
+    assert len(np.unique(indices.astype(np.uint64) * cfg.nq + directions)) == len(indices)
+
+
+@pytest.mark.parametrize("entry", ["sync", "initialize", "runtime_edit"])
+@pytest.mark.parametrize("second_x", [5., 6.])
+def test_passive_overlap_failure_keeps_live_state(device, entry, second_x):
+    from CelerisLab.simulation import Simulation
+    field = make_field(device)
+    bodies = manager(field, [obj(), obj(x=9., y=3., r=1., is_sensor=True)])
+    bodies.sync_to_gpu(field)
+    overlapping = obj(x=second_x)
+    if entry == "runtime_edit":
+        bodies.stage_add(overlapping)
+    else:
+        bodies.add(overlapping)
+    objects = bodies.objects
+    flags, curved, counts = field.flag.copy(), field.curved.fluid_idx.copy(), bodies.sensor_cell_counts.copy()
+    action = bodies.action.copy()
+    buffers = [(block, bytes(block.data), block.freed) for block in device.blocks]
+    writes = device.writes
+    sim = SimpleNamespace(bodies=bodies, field=field, lbm_cfg=field.cfg,
+                          _initialized=entry == "runtime_edit",
+                          _assert_allocation_contract=lambda: None,
+                          stream=SimpleNamespace(synchronize=device.synchronize))
+    with pytest.raises(ValueError, match="active fluid receivers|unique cut-link writers"):
+        if entry == "sync":
+            bodies.sync_to_gpu(field)
+        elif entry == "initialize":
+            Simulation.initialize(sim)
+        else:
+            Simulation.sync_bodies(sim)
+    assert bodies.objects == objects
+    assert bodies.has_pending_edit() == (entry == "runtime_edit")
+    assert sim._initialized == (entry == "runtime_edit")
+    assert device.writes == writes and len(device.blocks) == len(buffers)
+    assert all(bytes(block.data) == data and block.freed == freed for block, data, freed in buffers)
+    np.testing.assert_array_equal(field.flag, flags)
+    np.testing.assert_array_equal(field.curved.fluid_idx, curved)
+    np.testing.assert_array_equal(bodies.sensor_cell_counts, counts)
+    np.testing.assert_array_equal(bodies.action, action)
+
+
 def test_action_upload_and_scratch_are_pinned(device):
     field = make_field(device, streaming="esopull")
     bodies = manager(field, [obj(is_force_region=True)])