Merge 1cbdc479 (claude/step-single-sync) for integration review

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

src/CelerisLab/lbm/block_stepper.py

@@ -204,11 +204,21 @@ class PeriodicBlockStepper:
                 raise
 
     def step(self, n=1):
-        """Advance n steps: halo refresh, then core pull/collide, per step."""
+        """Advance n steps: halo refresh, then core pull/collide, per step.
+
+        All n steps are enqueued in order on the one stream and the host waits
+        once before returning (same launches and bits as n step(1) calls).
+        Asynchronous device errors surface at that single wait, so they are not
+        attributed to the step that caused them; call step(1) repeatedly to
+        localize one. On any failure the instance is failed and step_count keeps
+        its entry value.
+        """
         if isinstance(n, bool) or not isinstance(n, Integral) or n < 0:
             raise ValueError("n must be a nonnegative integer")
         with self._lock, self.ctx.activate():
             self._require("ready")
+            if n == 0:
+                return
             kw = dict(block=self._block, stream=self._stream)
             try:
                 self._status = "advancing"
@@ -224,9 +234,9 @@ class PeriodicBlockStepper:
                     self._step(self._current, self._spare, np.uint32(self.layout.halo),
                                self._tables.get("flags", np.uintp(0)), np.uintp(0),  # no kind
                                grid=self._grid, **kw)
-                    self._stream.synchronize()
-                    self._current, self._spare = self._spare, self._current
-                    self._step_count += 1
+                    self._current, self._spare = self._spare, self._current  # host handles only
+                self._stream.synchronize()
+                self._step_count += n
                 self._status = "ready"
             except BaseException:
                 self._status = "failed"

src/CelerisLab/lbm/refinement_stepper.py

@@ -822,27 +822,41 @@ class FixedPatchStepper:
                             grid=((pairs + self._block[0] - 1) // self._block[0], 1), **kw)
 
     def step(self, n=1):
-        """Advance G→G by n base intervals: level l takes 2**l hops per interval."""
+        """Advance G→G by n base intervals: level l takes 2**l hops per interval.
+
+        All n intervals are enqueued in order on the one stream, and the host waits
+        once, before returning; step(0) launches and waits for nothing. On return
+        the device work is complete, step_count and an open window cover all n
+        intervals, and the raw-exchange ledger holds the final interval only.
+
+        Asynchronous device errors surface at that single wait, so they are not
+        attributed to the interval that caused them; call step(1) repeatedly to
+        localize one (same launches, same bits). On any failure the instance is
+        failed, the exchange ledger and window are invalid, and step_count keeps
+        its value from entry to the failing call.
+        """
         if isinstance(n, bool) or not isinstance(n, Integral) or n < 0:
             raise ValueError("n must be a nonnegative integer")
         with self._lock, self.ctx.activate():
             self._require("ready")
+            if n == 0:
+                return
             root = self._levels[0]
             kw = dict(block=self._block, stream=self._stream)
             try:
                 self._status = "advancing"
+                self._exchange_valid = False
                 for _ in range(n):
-                    if self._observe_raw_exchange:
-                        self._exchange_valid = False
+                    if self._observe_raw_exchange:  # stream-ordered: holds the last interval only
                         for b in self._level_boundaries:
                             if b is not None:
                                 cuda.memset_d8_async(b.interval_integral, 0,
                                                      len(b.tables["cell"]) * 32, self._stream)
                     self._hop(0, root.current, root.spare, **kw)
-                    self._stream.synchronize()
-                    root.current, root.spare = root.spare, root.current
-                    self._step_count += 1
-                    self._exchange_valid = self._observe_raw_exchange
+                    root.current, root.spare = root.spare, root.current  # host handles only
+                self._stream.synchronize()
+                self._step_count += n
+                self._exchange_valid = self._observe_raw_exchange
                 self._status = "ready"
             except BaseException:
                 self._window_valid = False