run: one fail-closed device transaction with translating circles

Split BodyManager action upload into host-only _prepare_action_upload
(validation + pinned staging) and _upload_prepared_action_async (H2D);
_upload_action_async stays the checked wrapper for direct callers.
Simulation.run prepares outside the failure boundary, then runs action
upload, obs/diagnostic zeroing, step group, readback and sync as one
transaction: with translating circles or attached moving tables any
exception fails every circle and the Simulation (also for run(0) and an
empty registry). The inner _run_moving catch is removed (its only caller
is run). Stationary runs keep their failure policy; pending/hold-consume
and run(0) success behaviour unchanged. No kernel change.

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

docs/moving_bodies.md

@@ -170,9 +170,9 @@ acts as a sliding wall. One moving step consumes one target.
 | `run(1)` without pending | hold: `c_old == c_new`, no cover/uncover, links at `applied`, translational `u_w = 0` |
 | `run(n > 1)` with pending | `ValueError` before mutation (single-step-only moving API) |
 | `run(n > 1)`, no pending | allowed: hold for n steps, translational `u_w = 0`, spin only |
-| `run(0)` | no motion upload, no commit; `pending` kept |
-| validation failure | `ValueError`, nothing mutated, retry allowed |
-| upload/launch/sync failure after enqueue | simulation unusable; `applied` and `pending` are not reported as device truth; no re-entry |
+| `run(0)` | no motion upload, no commit; `pending` kept; action upload and obs/diagnostic zeroing still run |
+| validation failure (incl. action validation and pinned staging in `run`) | `ValueError`/error before any device work, nothing mutated, retry allowed |
+| failure of any device work in `run` (action upload, obs/diagnostic zeroing, record upload, launch, readback, sync), also for `run(0)` | simulation unusable, every translating circle failed; `applied` and `pending` are not reported as device truth; no re-entry |
 
 Displacement norm: Euclidean per-step `|c_new - c_old|_2 ≤ 0.5` lattice
 cells (declared bound; CANDIDATE value).

src/CelerisLab/body/manager.py

@@ -629,10 +629,17 @@ class ObjectManager:
         cuda.memcpy_dtoh_async(self.obs_pinned, self.obs_gpu, stream)
 
     def _upload_action_async(self, stream: cuda.Stream) -> None:
-        """Async H2D of host action array to action_gpu.
+        """Checked async H2D of host action array to action_gpu."""
+        if self._prepare_action_upload():
+            self._upload_prepared_action_async(stream)
+
+    def _prepare_action_upload(self) -> bool:
+        """Validate the action array and stage it in pinned memory; no device work.
 
         Single gate for every action path (setters or direct ``action`` edits):
         surface-actuated and Jet-driven bodies must keep zero motion slots.
+        Returns False when there is nothing to upload. Restaging assumes no
+        copy from ``_action_pinned`` is in flight (``run`` ends with a sync).
         """
         slot = 3*(self.cfg.dim if self.cfg else 2)
         for obj in self.objects:
@@ -643,10 +650,14 @@ class ObjectManager:
             self.validate_jet_motion(obj.obj_id, {"vx": self.action[base], "vy": self.action[base+1],
                                                   "omega": self.action[base+slot-1]})
         if self.action_gpu is None or self.action.size == 0:
-            return
+            return False
         if self._action_pinned is None or self._action_pinned.size != self.action.size:
             self._action_pinned = cuda.pagelocked_empty(self.action.size, np.float32)
         self._action_pinned[:] = self.action
+        return True
+
+    def _upload_prepared_action_async(self, stream: cuda.Stream) -> None:
+        """Enqueue H2D of the staged ``_action_pinned`` (see ``_prepare_action_upload``)."""
         cuda.memcpy_htod_async(self.action_gpu, self._action_pinned, stream)
 
     @property

src/CelerisLab/body/motion.py

@@ -99,7 +99,12 @@ class MovingCircle:
         return self.centre(self.applied), self.centre(offset), velocity
 
     def advance(self, upload: Callable, step: Callable, synchronize: Callable) -> None:
-        """Upload the record, run one step and commit only after a successful sync.
+        """Single-circle transaction helper (unit tests): upload, step, sync, commit.
+
+        Not the production path: ``Simulation._run_moving`` batches all circles,
+        adds the ``no_donor`` gate and uses ``commit``/``fail`` directly.
+
+        Upload the record, run one step and commit only after a successful sync.
 
         Any exception after the upload is attempted leaves device geometry
         unknown: the circle is marked failed, never reports applied state again,

src/CelerisLab/lbm/stepper.py

@@ -21,7 +21,7 @@ class LBMStepper:
 
         self._esopull = (cfg.streaming == "esopull")
         # Set by the owner when a step would be wrong (translating circles
-        # excluded from the packed links before the moving kernels exist).
+        # without attached moving tables, a pending target, or a failed step).
         self.blocked: str | None = None
         self.moving = None  # MovingTables attached by the owner (translating circles)
 
@@ -39,8 +39,10 @@ class LBMStepper:
         self.apply_esopull_curved_fn = module.get_function("ApplyEsoPullCurvedKernel")
         self.sensor_fn = module.get_function("SensorKernel")
         self.force_region_fn = module.get_function("ForceRegionKernel")
-        self.moving_fn = None if self._esopull else module.get_function("MovingCircleLinks")
-        self.moving_cells_fn = None if self._esopull else module.get_function("MovingCircleCells")
+        # Moving-circle kernels exist only in D2Q9 double-buffer modules.
+        moving = not self._esopull and cfg.is_d2q9
+        self.moving_fn = module.get_function("MovingCircleLinks") if moving else None
+        self.moving_cells_fn = module.get_function("MovingCircleCells") if moving else None
 
         tpb = cfg.threads_per_block
         self.block = (tpb, 1, 1)

src/CelerisLab/simulation.py

@@ -216,6 +216,12 @@ class Simulation:
         self._failed = f"moving circle step failed: {exc!r}"
         self.stepper.blocked = self._failed  # direct stepper calls are refused as well
 
+    def _fail_moving_run(self, exc: BaseException) -> None:
+        # Device work was attempted with translating circles (or their tables) present.
+        for obj in self._moving_bodies():
+            obj.motion.fail(exc)
+        self._fail_moving(exc)  # also when tables are attached but no circle is registered
+
     def _attach_moving(self) -> None:
         # Direct stepper calls stay blocked until the moving tables exist.
         tables = getattr(self, "_moving_tables", None)
@@ -242,7 +248,8 @@ class Simulation:
         self._require_usable()
         tables = self.stepper.moving
         if tables is None:
-            return np.zeros((0, 6), np.float32)
+            from .body.motion import MovingTables
+            return np.zeros((0, MovingTables.DIAG_FLOATS), np.float32)
         self.stream.synchronize()
         return tables.download_diagnostics()
 
@@ -733,9 +740,20 @@ class Simulation:
         if stream is None:
             stream = self.stream
 
-        # Async upload action
-        if upload_act and self.bodies.count > 0:
-            self.bodies._upload_action_async(stream)
+        # Host-only action validation and staging: a rejection here stays retryable.
+        staged = upload_act and self.bodies.count > 0 and self.bodies._prepare_action_upload()
+        watched = bool(moving) or self.stepper.moving is not None
+        try:
+            self._run_steps(steps, stream, staged, sync_obs, zero_obs, checkpoint_interval)
+        except BaseException as exc:
+            if watched:  # device work was attempted: motion state is unknown
+                self._fail_moving_run(exc)
+            raise
+
+    def _run_steps(self, steps, stream, staged, sync_obs, zero_obs, checkpoint_interval):
+        """Device part of ``run``: action upload, obs zeroing, step group, sync."""
+        if staged:
+            self.bodies._upload_prepared_action_async(stream)
 
         # Zero obs segments before step group
         if zero_obs:
@@ -782,30 +800,28 @@ class Simulation:
     def _run_moving(self, steps, stream, sync_obs):
         """Steps with translating circles: hold, or one advancing step that commits
         each pending target only after synchronization and a zero ``no_donor`` count
-        (docs/moving_bodies.md 4a, 5). Any failure leaves the Simulation unusable."""
+        (docs/moving_bodies.md 4a, 5). ``run`` fails the Simulation on any exception."""
         tables = self.stepper.moving
         circles = [(obj.motion, obj.obj_id) for obj in self._moving_bodies()]
-        try:
-            advancing = tables.upload_records(circles, stream)
-            self.stepper._step(steps, action_gpu=self.bodies.action_gpu,
-                               obs_gpu=self.bodies.obs_gpu, stream=stream, advance=advancing)
-            self.bodies.increment_obs_steps(steps)
-            if sync_obs:
-                self.bodies.download_obs_full_async(stream)
-            stream.synchronize()
+        pending = any(motion.pending is not None for motion, _ in circles)
+        advancing = tables.upload_records(circles, stream)
+        self.stepper._step(steps, action_gpu=self.bodies.action_gpu,
+                           obs_gpu=self.bodies.obs_gpu, stream=stream, advance=advancing)
+        self.bodies.increment_obs_steps(steps)
+        if sync_obs:
+            self.bodies.download_obs_full_async(stream)
+        stream.synchronize()
+        if advancing:
+            no_donor = tables.download_diagnostics()[:, tables.DIAG.index("no_donor")]
+            if np.any(no_donor > 0):
+                raise RuntimeError("translating circle refill found no donor cell")
+        if pending:  # a target equal to the applied offset is consumed as a hold step
+            for motion, _ in circles:
+                motion.commit()
             if advancing:
-                no_donor = tables.download_diagnostics()[:, tables.DIAG.index("no_donor")]
-                if np.any(no_donor > 0):
-                    raise RuntimeError("translating circle refill found no donor cell")
-                for motion, _ in circles:
-                    motion.commit()
                 tables.upload_records(circles, stream)  # first hold: old = new = applied, u = 0
                 stream.synchronize()
-                self._attach_moving()
-        except BaseException as exc:
-            for motion, _ in circles:  # device geometry/state unknown after any failure
-                motion.fail(exc)
-            raise
+            self._attach_moving()
 
     def step(self, n: int = 1):
         """Advance *n* steps (convenience for interactive use)."""

tests/unit/test_moving_bodies.py

@@ -113,7 +113,8 @@ def moving_simulation(monkeypatch, *, moving=True):
     sim._failed, sim._initialized = None, True
     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"),
+                                 _prepare_action_upload=lambda: True,
+                                 _upload_prepared_action_async=lambda s: writes.append("action"),
                                  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"))
@@ -297,6 +298,44 @@ def test_run_one_advances_and_commits_after_sync(monkeypatch):
     assert ("step", 3, False) in writes and motion.applied == (0., .25)
 
 
+def test_zero_motion_target_is_consumed_by_run_one(monkeypatch):
+    sim, writes, motion = advancing_simulation(monkeypatch)
+    motion.set_target((0., 0.), (0., 0.))          # equal to applied: a hold target
+    sim._attach_moving()
+    assert sim.stepper.blocked == module.PENDING_BLOCKED
+    sim.run(1, stream=STREAM)
+    assert motion.pending is None and motion.applied == (0., 0.)
+    assert sim.stepper.blocked is None and ("step", 1, False) in writes
+    writes.clear()
+    sim.run(3, stream=STREAM)                      # no stale latch
+    assert ("step", 3, False) in writes
+
+
+@pytest.mark.parametrize("streaming,d2q9,bound", [("double_buffer", True, True),
+                                                  ("double_buffer", False, False),
+                                                  ("esopull", True, False)])
+def test_moving_kernels_bound_only_for_d2q9_double_buffer(monkeypatch, streaming, d2q9, bound):
+    from CelerisLab.lbm import stepper as stepper_module
+    monkeypatch.setattr(stepper_module.cuda, "mem_alloc", lambda n: object())
+    moving = {"MovingCircleLinks", "MovingCircleCells"}
+
+    def get_function(name):
+        if name in moving and not (d2q9 and streaming == "double_buffer"):
+            raise KeyError(name)                 # absent from D3Q19 / EsoPull modules
+        return lambda *a, **k: None
+    cfg = SimpleNamespace(streaming=streaming, threads_per_block=128, is_d2q9=d2q9,
+                          nx=16, ny=8, nz=4, nq=9 if d2q9 else 19)
+    stepper = stepper_module.LBMStepper(SimpleNamespace(), SimpleNamespace(get_function=get_function), cfg)
+    assert (stepper.moving_fn is not None) == bound and (stepper.moving_cells_fn is not None) == bound
+
+
+def test_moving_diagnostics_empty_shape_matches_layout(monkeypatch):
+    from CelerisLab.body.motion import MovingTables
+    sim = make_simulation(monkeypatch)
+    sim._failed, sim.stepper = None, SimpleNamespace(moving=None)
+    assert sim.moving_diagnostics().shape == (0, MovingTables.DIAG_FLOATS)
+
+
 @pytest.mark.parametrize("failure", ("upload", "first_sync", "no_donor", "hold_upload",
                                      "hold_sync"))
 def test_failed_advance_fails_simulation(monkeypatch, failure):
@@ -323,6 +362,78 @@ def test_failed_advance_fails_simulation(monkeypatch, failure):
         sim.run(1, stream=STREAM)
 
 
+def raising(log, name, site):
+    def call(*a, **k):
+        log.append(name)
+        if name == site:
+            raise RuntimeError(f"{name} failed")
+    return call
+
+
+def two_circle_simulation(monkeypatch, site):
+    # No owner hooks: run itself must fail the Simulation and every circle.
+    sim, writes, first = advancing_simulation(monkeypatch)
+    second = SimpleNamespace(motion=circle(), obj_id=1, center=(80., 40.), radius=10.)
+    sim.bodies.objects.append(second)
+    sim.bodies._upload_prepared_action_async = raising(writes, "action", site)
+    sim.bodies.zero_obs_async = raising(writes, "obs", site)
+    sim.bodies.download_obs_full_async = raising(writes, "download", site)
+    sim.stepper.moving.zero_diagnostics = raising(writes, "zero_diag", site)
+    step = raising(writes, "step", site)
+    sim.stepper._step = lambda n, advance=False, **k: step()
+    return sim, (first, second.motion), SimpleNamespace(synchronize=raising(writes, "sync", site))
+
+
+@pytest.mark.parametrize("steps,site", [(n, s) for n in (0, 1, 3)
+                                        for s in ("action", "obs", "zero_diag", "step", "download", "sync")
+                                        if not (n == 0 and s == "step")])  # run(0) launches nothing
+def test_device_failure_in_run_fails_every_circle(monkeypatch, steps, site):
+    sim, motions, stream = two_circle_simulation(monkeypatch, site)
+    if steps < 2:
+        motions[0].set_target((0., .25), (0., .05))
+    with pytest.raises(RuntimeError, match=f"{site} failed"):
+        sim.run(steps, stream=stream)
+    assert all(m.failed for m in motions) and sim._failed is not None
+    assert sim.stepper.blocked == sim._failed
+    assert motions[0].applied == (0., 0.)  # nothing committed before the first sync
+    with pytest.raises(RuntimeError, match="Simulation failed"):
+        sim.run(1, stream=STREAM)
+
+
+def test_action_rejection_stays_retryable_before_device_work(monkeypatch):
+    sim, writes, motion = advancing_simulation(monkeypatch)
+    motion.set_target((0., .25), (0., .05))
+
+    def reject():
+        raise ValueError("bad action")
+    sim.bodies._prepare_action_upload = reject
+    for steps in (0, 1):
+        with pytest.raises(ValueError, match="bad action"):
+            sim.run(steps, stream=STREAM)
+    assert writes == [] and sim._failed is None and sim.stepper.blocked is None
+    assert not motion.failed and motion.pending is not None
+    sim.bodies._prepare_action_upload = lambda: True
+    sim.run(1, stream=STREAM)
+    assert motion.applied == (0., .25) and motion.pending is None
+
+
+def test_attached_tables_without_circles_fail_closed(monkeypatch):
+    sim, writes, _ = advancing_simulation(monkeypatch)
+    sim.bodies.objects = []
+    sim.stepper.moving.zero_diagnostics = raising(writes, "zero_diag", "zero_diag")
+    with pytest.raises(RuntimeError, match="zero_diag failed"):
+        sim.run(0, stream=STREAM)
+    assert sim._failed is not None and sim.stepper.blocked == sim._failed
+
+
+def test_stationary_run_failure_policy_unchanged(monkeypatch):
+    sim, writes = moving_simulation(monkeypatch, moving=False)
+    sim.bodies.zero_obs_async = raising(writes, "obs", "obs")
+    with pytest.raises(RuntimeError, match="obs failed"):
+        sim.run(2, stream=STREAM)
+    assert sim._failed is None and sim.stepper.blocked is None
+
+
 def test_failed_hold_step_blocks_the_direct_stepper(monkeypatch):
     # No target is pending, so only the failure latch can refuse direct steps.
     from CelerisLab.lbm.stepper import LBMStepper