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>

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"))
@@ -361,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