Gate action upload on Jet stationarity; document setter device-error handling

ObjectManager._upload_action_async is the single entry for every action
path (setters, direct edits of the public action array, run()). It now
also rejects nonzero vx/vy/omega slots of Jet-driven bodies via
validate_jet_motion, before any device copy. Regression: each motion
slot edited directly is rejected with no device write; zero motion
uploads. Simulation.set_jet_amplitude documents that a device error
requires close/recreate, as for other CUDA exceptions.

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

src/CelerisLab/body/manager.py

@@ -622,11 +622,19 @@ 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."""
+        """Async H2D of host action array to action_gpu.
+
+        Single gate for every action path (setters or direct ``action`` edits):
+        surface-actuated and Jet-driven bodies must keep zero motion slots.
+        """
+        slot = 3*self.cfg.dim
         for obj in self.objects:
             obj.validate_surface_actuation(self.cfg)
-            if obj.surface_actuation is not None and self.action[(obj.obj_id+1)*3*self.cfg.dim-1] != 0:
+            base = obj.obj_id*slot
+            if obj.surface_actuation is not None and self.action[base+slot-1] != 0:
                 raise ValueError("Surface actuation cannot share the rotation action")
+            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
         if self._action_pinned is None or self._action_pinned.size != self.action.size:

src/CelerisLab/simulation.py

@@ -173,7 +173,12 @@ class Simulation:
 
     @context_bound
     def set_jet_amplitude(self, index: int, amplitude: float) -> None:
-        """Change one jet's signed amplitude; re-uploads per-link velocity only."""
+        """Change one jet's signed amplitude; re-uploads per-link velocity only.
+
+        Invalid input changes nothing. A device error leaves the host amplitude
+        unchanged but the device column undefined: treat it like any other CUDA
+        exception here and close/recreate the Simulation before stepping again.
+        """
         jet = self.bodies.jets[index].with_amplitude(amplitude)
         if self._initialized:
             self.field.curved.set_jet_amplitude(index, amplitude, self.stream)

tests/unit/test_jets.py

@@ -253,6 +253,23 @@ def test_set_body_rejects_motion_of_jet_driven_body(device):
     bodies.set_body_state(0, omega=0.)
 
 
+def test_action_upload_rejects_motion_of_jet_driven_body(device):
+    # Direct edits of the public action array are checked at the upload gate.
+    jet = Jet(Sector(CENTER, 0., .6, 3.), Radial(CENTER, inward=False), "quartic", 2e-3)
+    field, bodies = jet_case(device, [jet])
+    stream = SimpleNamespace(synchronize=lambda: None)
+    slot = 3*field.cfg.dim
+    for index in (0, 1, slot - 1):  # vx, vy, omega
+        bodies.action[:] = 0
+        bodies.action[index] = .01
+        writes = device.writes
+        with pytest.raises(ValueError, match="stationary"):
+            bodies._upload_action_async(stream)
+        assert device.writes == writes  # nothing reached the device
+    bodies.action[:] = 0
+    bodies._upload_action_async(stream)  # zero motion stays accepted
+
+
 def test_fine_boundary_keeps_edge_frame_hits_and_checks_jets():
     from CelerisLab.body.objects import SimObject
     from CelerisLab.body.geometry.circle import CircleGeometry