lifecycle: Simulation jet update failure fails the instance like FixedPatchStepper

Simulation.set_jet_amplitude validated before upload and committed host
state after it, but after an uncertain device write nothing stopped reuse.
Record the failure in Simulation._failed and refuse every public device
call except close() (FixedPatchStepper already uses status 'failed').
Invalid input stays side-effect free and retryable. Adds host fault tests
for both owners and states the rule in docs/lifecycle.md.

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

src/CelerisLab/simulation.py

@@ -47,6 +47,7 @@ class Simulation:
         # Internal stream for async DTOH coordination
         self._stream: Optional[cuda.Stream] = None
         self._closed = False
+        self._failed = None  # reason after an uncertain device write; only close() remains
 
         # Load configs
         self.lbm_cfg = load_lbm_config(lbm_config_path)
@@ -172,15 +173,26 @@ class Simulation:
     def set_jet_amplitude(self, index: int, amplitude: float) -> None:
         """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.
+        Invalid input raises before any change. Any error after validation
+        leaves the host amplitude unchanged but the device column undefined,
+        so the Simulation is marked failed: every later call except close()
+        raises; close and recreate it.
         """
+        self._require_usable()
         jet = self.bodies.jets[index].with_amplitude(amplitude)
         if self._initialized:
-            self.field.curved.set_jet_amplitude(index, amplitude, self.stream)
+            try:
+                self.field.curved.set_jet_amplitude(index, amplitude, self.stream)
+            except BaseException as exc:
+                self._failed = f"jet amplitude upload failed: {exc!r}"
+                raise
         self.bodies.jets[index] = jet  # later geometry syncs keep the new amplitude
 
+    def _require_usable(self):
+        failed = getattr(self, "_failed", None)
+        if failed is not None:
+            raise RuntimeError(f"Simulation failed ({failed}); close() and recreate it")
+
     def remove_body(self, id: int) -> None:
         """Remove a body from the simulation.
 
@@ -338,6 +350,7 @@ class Simulation:
         Unlike read_body this is not accumulated/time-averaged. It resets after
         geometry sync or checkpoint load; advance before interpreting it.
         """
+        self._require_usable()
         obj = self.bodies.get(id)
         curved = self.field.curved
         has_aperture = (curved.aperture is not None
@@ -365,6 +378,7 @@ class Simulation:
             BodyTelemetry with fields ``force``, ``torque``, ``sensor``.
             ``sensor`` is a dim-length array of zeros for non-sensor bodies.
         """
+        self._require_usable()
         if self.bodies.obs_pinned is None:
             raise RuntimeError("No obs buffer. Call run() first.")
         if stream is None:
@@ -384,6 +398,7 @@ class Simulation:
         This is the most efficient readback for DRL control loops
         (single contiguous D2H copy, no Python per-body overhead).
         """
+        self._require_usable()
         if self.bodies.obs_pinned is None:
             raise RuntimeError("No obs buffer. Call run() first.")
         if stream is None:
@@ -446,6 +461,7 @@ class Simulation:
     @context_bound
     def initialize(self):
         """Initialize flow field and sync objects to GPU."""
+        self._require_usable()
         self._assert_allocation_contract()
         # Validate the complete geometry before flags, populations or modules
         # change. Compact-list construction publishes no outputs on failure.
@@ -478,6 +494,7 @@ class Simulation:
         In EsoPull mode, the backing layout is decoded to physical DDF on
         GPU for the patched cells only, patched on host, then encoded back.
         """
+        self._require_usable()
         if not self._initialized:
             raise RuntimeError("Call initialize() before sync_bodies()")
 
@@ -573,6 +590,7 @@ class Simulation:
                 sensor) on GPU and reset the step accumulator before stepping.
             checkpoint_interval: Nonnegative integer; if >0, save every N steps.
         """
+        self._require_usable()
         if not self._initialized:
             raise RuntimeError("Call initialize() first")
         steps = operator.index(steps)
@@ -637,10 +655,12 @@ class Simulation:
     @context_bound
     def get_macroscopic(self) -> Dict[str, np.ndarray]:
         """Download DDF and return rho, ux, uy [, uz]."""
+        self._require_usable()
         return self.field.get_macroscopic(step_id=self.stepper.step_count)
 
     @context_bound
     def get_ddf(self) -> np.ndarray:
+        self._require_usable()
         self.field.download_ddf(step_id=self.stepper.step_count)
         return self.field.ddf.copy()
 
@@ -654,23 +674,27 @@ class Simulation:
 
         Accepted: omega, fx, fy, fz, n_objects.
         """
+        self._require_usable()
         self.field.update_params(**kwargs)
 
     # -- Snapshots -----------------------------------------------------------
     @context_bound
     def snapshot(self):
         """Snapshot the flow field; restore keeps the current clock and controls."""
+        self._require_usable()
         self.field.snapshot(step_id=self.stepper.step_count)
 
     @context_bound
     def restore(self):
         """Restore the saved flow field at the current streaming parity."""
+        self._require_usable()
         self.field.restore(step_id=self.stepper.step_count)
 
     # -- Checkpoint / Restore (HDF5) ----------------------------------------
     @context_bound
     def save_checkpoint(self, path: str = None):
         """Save full simulation state to an HDF5 file."""
+        self._require_usable()
         from .common.checkpoint import save_checkpoint
         if self.bodies.jets:
             raise NotImplementedError("Jets are not checkpointed yet")
@@ -680,6 +704,7 @@ class Simulation:
     @context_bound
     def load_checkpoint(self, path: str):
         """Restore simulation state from an HDF5 checkpoint."""
+        self._require_usable()
         from .common.checkpoint import load_checkpoint
         if not self._initialized:
             self.initialize()