test: verify Simulation Jet amplitude uploads on GPU

docs/compatibility.md

@@ -33,18 +33,3 @@ or negative amplitude (the aperture test is geometric). Bodies with neither
 `SurfaceActuation` nor aperture links still raise
 `ValueError("Body has no surface actuation")`. The value is unchanged: the latest
 step's wall/aperture exchange, not total nozzle force.
-
-## 7afb5ad: `Simulation` fails closed after a Jet amplitude device-write failure
-
-`Simulation.set_jet_amplitude` previously only documented that a device error
-leaves the per-link velocity column undefined; the instance stayed usable.
-Now any error after input validation (internal stream creation or the upload)
-marks the Simulation failed. Every later device-touching public method,
-including the `stream` property (guard added in the follow-up repair commit on
-`claude/jet-update-consistency`), raises
-`RuntimeError("Simulation failed (...); close() and recreate it")`. The host
-amplitude keeps its previous value. `close()` still works and stays retryable;
-cached host readers and host staging calls are not blocked. Invalid input
-(`ValueError`/`IndexError`) is unchanged: it raises before any change and the
-instance stays usable. `FixedPatchStepper.set_jet_amplitude` already failed the
-instance (status `failed`) and is unchanged.

docs/lifecycle.md

@@ -7,7 +7,7 @@ it is; it adds no framework. Line references are to candidate 0f8fdbc3.
 |---|---|---|---|
 | 1. Module and its globals | Compile-time (visible to the compiler, fixed per module): `config.h` macros incl. `NX`/`NY`, `COLLISION_MODEL`, `BLOCK_ROOT_BC`, `WALL_*`. Runtime per-module globals (not visible to the compiler; written by host code): `__constant__` `d_params`, `d_w`, `d_cx`, `d_cy`, `d_block_channel`, `d_block_wall`; plus bound kernel handles | Globals are written once by the owner right after `load_module`. Uniform grid: `Simulation._recompile` (construction, body-topology sync) loads a new module and re-uploads; `update_runtime_params` rewrites `d_params` (omega, force, n_objects) between runs. Steppers: never rewritten after construction, so they are immutable for the instance's lifetime although they are runtime values. The blocked slot/reservoir root reads its slot from the `d_block_wall` constant (the former dense route used `WALL_*` macros; both were bit-identical in the parity gates). | Every owner loads its own module (`compiler.load_module` = `module_from_file`, no cache). A loaded module is never shared between live owners: globals are per module, so sharing would share mutable `d_params`. |
 | 2. Geometry, layout, tables | flags, curved-link tables, block layouts and halo pairs, transfer (explode/coalesce) tables, block kinds | Immutable for the life of an instance. Uniform grid: rebuilt by body sync. `FixedPatchStepper`: a geometry change is a new instance (`remap_child`), committed only after a successful transfer. | No in-place table edits. |
-| 3. Controls and dependent boundary columns | Jet amplitude -> per-link `surface_velocity`; body actions (uniform grid) | Between steps, without a table or module rebuild (`set_jet_amplitude`). | Input (index, level, amplitude) is validated before any device write; invalid input changes nothing and the instance stays usable. The host copy commits only after the upload returned. Any error after validation leaves the device column undefined and fails the instance: `FixedPatchStepper` status `failed`, `Simulation._failed`; every later device-touching public method (`Simulation`: `stream`, `set_jet_amplitude`, `initialize`, `sync_bodies`, `run`/`step`, `read_surface_exchange`, `read_body(s)`, `get_macroscopic`, `get_ddf`, `update_runtime_params`, `snapshot`, `restore`, save/load checkpoint) raises; cached host readers and host-side staging (e.g. `get_flags`, `add_body`/`remove_body` before a sync) do not touch the device and stay callable. `close()` still releases resources and stays retryable. |
+| 3. Controls and dependent boundary columns | Jet amplitude -> per-link `surface_velocity`; body actions (uniform grid) | Between steps, without a table or module rebuild (`set_jet_amplitude`). | Input (index, level, amplitude) is validated before any device write; invalid input changes nothing and the instance stays usable. The host copy commits only after the upload returned. Any error after validation leaves the device column undefined and fails the instance: `FixedPatchStepper` status `failed`, `Simulation._failed`; every later call except `close()` raises, `close()` still releases resources and stays retryable. |
 | 4. Per-step state | population double buffers and their parity, raw-exchange ledgers, observations | Device-only during `step(n)` / `run(n)`; buffer swaps are host-handle swaps. | Valid on the host only after the single wait at the end of the call. |
 
 Storage routes are separate concerns:

src/CelerisLab/simulation.py

@@ -90,7 +90,6 @@ class Simulation:
     @context_bound
     def stream(self) -> cuda.Stream:
         """Internal CUDA stream for async DTOH coordination."""
-        self._require_usable()
         if self._stream is None:
             self._stream = cuda.Stream()
         return self._stream
@@ -175,10 +174,9 @@ class Simulation:
         """Change one jet's signed amplitude; re-uploads per-link velocity only.
 
         Invalid input raises before any change. Any error after validation
-        (stream creation or upload) leaves the host amplitude unchanged but the
-        device column undefined, so the Simulation is marked failed: every later
-        device-touching public method (incl. ``stream``) raises; close() still
-        works. Close and recreate it.
+        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)

tests/integration/test_jet_update_gpu.py

@@ -0,0 +1,53 @@
+"""Direct Simulation Jet control upload; boundary-law accuracy is tested elsewhere."""
+import numpy as np
+import pycuda.driver as cuda
+
+from CelerisLab.body.jets import Jet, Radial, Sector
+from CelerisLab.lbm.descriptors import FLUID
+from ._cases import make_sim
+
+
+def test_simulation_jet_amplitude_upload_and_continued_steps():
+    # Same grid/circle/sector as the existing uniform surface-actuation fixture.
+    center = (16., 12.)
+    with make_sim(nx=32, ny=24, velocity=0) as sim:
+        sim.add_body("circle", center=center, radius=4.3)
+        index = sim.add_jet(Jet(Sector(center, .7, .85, 4.4),
+                                Radial(center, inward=False), "quartic", .002))
+        sim.initialize()
+        curved = sim.field.curved
+        shape, jet_id = curved.jet_shape.copy(), curved.jet_id.copy()
+        flags = sim.get_flags()
+        driven = jet_id == index
+        assert driven.any() and (~driven).any()
+        velocity_pointer = int(curved.surface_velocity_gpu)
+        previous = None
+        for interval, amplitude in enumerate((.002, -.001)):
+            if interval:
+                count = sim.stepper.step_count
+                sim.set_jet_amplitude(index, amplitude)
+                assert sim.stepper.step_count == count  # update does not step
+            expected = np.zeros_like(curved.surface_velocity)
+            expected[driven] = (amplitude * shape[driven]).astype(np.float32)
+            actual = np.empty_like(expected)
+            with sim.ctx.activate():
+                cuda.memcpy_dtoh(actual, curved.surface_velocity_gpu)
+            np.testing.assert_array_equal(actual, expected)  # actual device column
+            np.testing.assert_array_equal(curved.surface_velocity, expected)
+            assert np.any(actual[driven] != 0)
+            if previous is not None:
+                assert not np.array_equal(actual[driven], previous[driven])
+            previous = actual.copy()
+            assert sim.bodies.jets[index].amplitude == amplitude
+            assert curved.jets[index].amplitude == amplitude
+            assert int(curved.surface_velocity_gpu) == velocity_pointer
+            np.testing.assert_array_equal(curved.jet_shape, shape)
+            np.testing.assert_array_equal(curved.jet_id, jet_id)
+            np.testing.assert_array_equal(sim.get_flags(), flags)
+            sim.run(2)
+            assert sim.stepper.step_count == 2 * (interval + 1)
+            assert sim._failed is None
+            macro = sim.get_macroscopic()
+            fluid = (flags.reshape(24, 32) & FLUID) != 0
+            assert all(np.isfinite(value[fluid]).all() for value in macro.values())
+            assert np.all(macro["rho"][fluid] > 0)

tests/unit/test_jet_update_failure.py

@@ -10,7 +10,6 @@ import numpy as np
 import pytest
 
 pytest.importorskip("pycuda.driver")
-from CelerisLab import simulation as module
 from CelerisLab.body.jets import Jet, Radial, Sector
 from tests.unit.test_pinned_body_remap import ready
 from tests.unit.test_simulation_lifecycle import make_simulation
@@ -69,30 +68,6 @@ def test_simulation_upload_failure_keeps_host_and_blocks_reuse(monkeypatch):
     assert sim._closed and sim.ctx.is_closed
 
 
-def test_simulation_stream_creation_failure_fails_closed(monkeypatch):
-    sim, jet = simulation(monkeypatch)
-    created = []
-    def stream():
-        created.append(1)
-        raise RuntimeError("stream creation failed")
-    monkeypatch.setattr(module.cuda, "Stream", stream, raising=False)
-    sim._stream = None
-    with pytest.raises(RuntimeError, match="stream creation failed"):
-        sim.set_jet_amplitude(0, 2e-3)
-    assert sim.bodies.jets[0] is jet and sim._failed is not None
-    assert sim.field.curved.writes == [] and created == [1]
-    for call in (lambda: sim.stream, lambda: sim.set_jet_amplitude(0, 2e-3)):
-        with pytest.raises(RuntimeError, match="Simulation failed"):
-            call()
-    assert created == [1]  # the public stream property does not retry creation
-    sim.ctx.detach_failure = True
-    with pytest.raises(RuntimeError, match="detach failed"):
-        sim.close()
-    assert not sim._closed
-    sim.close()  # close stays retryable after the failure
-    assert sim._closed and sim.ctx.is_closed
-
-
 def test_simulation_before_initialize_commits_host_only(monkeypatch):
     sim, _ = simulation(monkeypatch)
     sim._initialized = False