lbm: guard direct stepper entry while translating circles exist

LBMStepper.step raises before any launch or buffer swap when the owner has
set 'blocked'; Simulation sets it when a translating circle is declared, at
initialize and after stepper recreation. Tests: direct-stepper guard (zero
launches, no swap, unchanged step count), spin/multi-step/checkpoint run
variants, run(0) keeps the pending target, static runs proceed, mixed
moving/static packing parity with the moving id absent, failure refusal of
checkpoint calls, no retry and no re-entry after close.

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

tests/unit/test_moving_bodies.py

@@ -103,27 +103,70 @@ def moving_simulation(monkeypatch, *, moving=True):
                           center=(50., 40.), radius=10.)
     writes = []
     sim._failed, sim._initialized = None, True
-    sim.bodies = SimpleNamespace(objects=[obj], count=1, jets=[],
+    sim.bodies = SimpleNamespace(objects=[obj], count=1, jets=[], action_gpu=None, obs_gpu=None,
                                  has_pending_edit=lambda: False,
                                  _upload_action_async=lambda s: writes.append("action"),
-                                 zero_obs_async=lambda s: writes.append("obs"))
-    sim.stepper = SimpleNamespace(step=lambda *a, **k: writes.append("step"))
+                                 zero_obs_async=lambda s: writes.append("obs"),
+                                 increment_obs_steps=lambda n: writes.append("count"),
+                                 download_obs_full_async=lambda s: writes.append("download"))
+    sim.stepper = SimpleNamespace(step=lambda n, **k: writes.append(("step", n)), blocked=None)
+    sim._assert_runtime_contracts = lambda: None
     return sim, writes
 
 
+STREAM = SimpleNamespace(synchronize=lambda: None)
+
+
 def test_positive_steps_rejected_before_device_writes(monkeypatch):
     sim, writes = moving_simulation(monkeypatch)
     sim.bodies.objects[0].motion.set_target((0., .1), (0., .01))
-    for call in (lambda: sim.run(1), lambda: sim.run(5), lambda: sim.step(),
-                 lambda: sim.run(3, checkpoint_interval=1)):
+    sim.bodies.objects[0].motion.applied = (0., 0.)
+    for call in (lambda: sim.run(1, stream=STREAM), lambda: sim.run(5, stream=STREAM),
+                 lambda: sim.step(), lambda: sim.run(3, stream=STREAM, checkpoint_interval=1)):
         with pytest.raises(RuntimeError, match="moving kernels"):
             call()
     with pytest.raises(RuntimeError, match="moving kernels"):  # nominal at rest as well
         sim.bodies.objects[0].motion.pending = None
         sim.run(1)
+    sim.bodies.objects[0].state = SimpleNamespace(omega=.02)  # spin does not lift the guard
+    with pytest.raises(RuntimeError, match="moving kernels"):
+        sim.run(2, stream=STREAM)
     assert writes == [] and sim.bodies.objects[0].motion.applied == (0., 0.)
 
 
+def test_run_zero_keeps_pending_and_static_runs_proceed(monkeypatch):
+    sim, writes = moving_simulation(monkeypatch)
+    motion = sim.bodies.objects[0].motion
+    motion.set_target((0., .1), (0., .01))
+    sim.run(0, stream=STREAM)  # existing action/obs work only; no motion upload or commit
+    assert motion.pending == ((0., .1), (0., .01)) and motion.applied == (0., 0.)
+    assert ("step", 0) in writes
+    static, writes = moving_simulation(monkeypatch, moving=False)
+    static.save_checkpoint = lambda *a: writes.append("checkpoint")
+    static.run(3, stream=STREAM)
+    static.run(4, stream=STREAM, checkpoint_interval=2)
+    assert writes.count(("step", 3)) == 1 and writes.count(("step", 2)) == 2
+
+
+def test_direct_stepper_entry_is_guarded():
+    from CelerisLab.lbm.stepper import LBMStepper
+    launches = []
+    module = SimpleNamespace(get_function=lambda name: lambda *a, **k: launches.append(name))
+    cfg = SimpleNamespace(streaming="double_buffer", threads_per_block=128, is_d2q9=True,
+                          nx=64, ny=32, nz=1)
+    field = SimpleNamespace(ddf_gpu="ddf", temp_gpu="temp", n_curved=0)
+    stepper = LBMStepper(field, module, cfg)
+    assert stepper.blocked is None
+    stepper.blocked = "translating circles need the moving kernels"
+    for n in (1, 3):
+        with pytest.raises(RuntimeError, match="moving kernels"):
+            stepper.step(n, action_gpu=None, obs_gpu=None)
+    assert launches == [] and (field.ddf_gpu, field.temp_gpu) == ("ddf", "temp")
+    assert stepper.step_count == 0
+    stepper.step(0, action_gpu=None, obs_gpu=None)  # zero steps stay a no-op
+    assert launches == [] and stepper.step_count == 0
+
+
 def test_moving_body_rejects_topology_and_checkpoint(monkeypatch):
     sim, _ = moving_simulation(monkeypatch)
     for call in (sim.sync_bodies, sim.save_checkpoint, lambda: sim.load_checkpoint("x.h5")):
@@ -155,6 +198,7 @@ def body(x, y, r, translation=None, **kw):
 def routed_simulation(monkeypatch):
     sim = make_simulation(monkeypatch)
     sim._failed, sim._initialized, sim.lbm_cfg = None, False, Cfg()
+    sim.stepper = SimpleNamespace(blocked=None)
     sim.bodies = ObjectManager(nx=120, ny=80, nz=1, nq=9, cfg=Cfg())
     return sim
 
@@ -164,6 +208,7 @@ def test_add_body_and_set_body_route_targets(monkeypatch):
     bid = sim.add_body("circle", (50., 40.), 10., translation=Translation((-2., 2., -3., 3.), .1))
     still = sim.add_body("circle", (100., 40.), 5.)
     motion = sim.bodies.get(bid).motion
+    assert sim.stepper.blocked is not None  # declaring a moving circle blocks direct steps
     sim.set_body(bid, offset=(0., .3), vy=.05)
     sim.set_body(bid, offset=(0., .4), vy=.05)  # last wins
     for kwargs in (dict(offset=(0., .6)), dict(vy=.05), dict(vx=.05),
@@ -229,14 +274,18 @@ def test_initialize_validates_before_mutation(monkeypatch):
 
 
 def test_moving_circle_excluded_from_static_links():
-    still = [body(100., 40., 5.)]
-    mgr = ObjectManager(nx=120, ny=80, nz=1, nq=9, cfg=Cfg())
-    reference = mgr.build_compact_lists_for(still)
-    with_moving = mgr.build_compact_lists_for(
-        still + [body(50., 40., 10., Translation((0., 0., -1., 1.), .1))])
-    for a, b in zip(reference[:8], with_moving[:8]):
+    def manager(*objects):
+        mgr = ObjectManager(nx=120, ny=80, nz=1, nq=9, cfg=Cfg())
+        ids = [mgr.add(obj) for obj in objects]
+        return mgr, ids
+    reference_mgr, _ = manager(body(100., 40., 5.), body(20., 40., 6.))
+    mixed_mgr, ids = manager(body(100., 40., 5.), body(20., 40., 6.),
+                             body(55., 40., 10., Translation((0., 0., -1., 1.), .1)))
+    reference = reference_mgr.build_compact_lists_for(reference_mgr.objects)
+    mixed = mixed_mgr.build_compact_lists_for(mixed_mgr.objects)
+    for a, b in zip(reference[:8], mixed[:8]):  # static link columns byte-identical
         assert np.asarray(a).tobytes() == np.asarray(b).tobytes()
-    assert len(reference[0]) > 0
+    assert len(reference[0]) > 0 and ids[2] not in set(np.asarray(mixed[3]).tolist())
 
 
 def test_failed_moving_step_marks_simulation_failed(monkeypatch):
@@ -250,7 +299,12 @@ def test_failed_moving_step_marks_simulation_failed(monkeypatch):
     with pytest.raises(RuntimeError, match="sync failed"):
         motion.advance(lambda *_: None, lambda: None, fail)
     assert sim._failed is not None and motion.applied == (0., 0.)
-    for call in (lambda: sim.run(0), lambda: sim.run(1), sim.step, sim.sync_bodies):
+    for call in (lambda: sim.run(0), lambda: sim.run(1), sim.step, sim.sync_bodies,
+                 sim.save_checkpoint, lambda: sim.load_checkpoint("x.h5")):
         with pytest.raises(RuntimeError, match="Simulation failed"):
             call()
-    sim.close()  # close still works; no re-entry afterwards
+    with pytest.raises(RuntimeError, match="unknown"):  # no retry on the failed circle
+        motion.advance(lambda *_: None, lambda: None, lambda: None)
+    sim.close()  # close still works
+    with pytest.raises(Exception):  # no re-entry after close
+        sim.run(1)