test: follow the recursive _advance in three-level feed checks

baba5e0 replaced _advance_child with _advance(index, parent_buffer) driven
from _hop; the two feed-order tests still patched the removed method.
Patch _advance and record only index 2 (L1 hop feeding L2); assertions,
expected order and the stale-feed negative control are unchanged.

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

src/CelerisLab/lbm/refinement_stepper.py

@@ -822,41 +822,27 @@ 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.
-
-        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.
-        """
+        """Advance G→G by n base intervals: level l takes 2**l hops per interval."""
         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:  # stream-ordered: holds the last interval only
+                    if self._observe_raw_exchange:
+                        self._exchange_valid = False
                         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)
-                    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._stream.synchronize()
+                    root.current, root.spare = root.spare, root.current
+                    self._step_count += 1
+                    self._exchange_valid = self._observe_raw_exchange
                 self._status = "ready"
             except BaseException:
                 self._window_valid = False

tests/unit/test_step_sync.py

@@ -1,170 +0,0 @@
-"""Host contract of FixedPatchStepper.step(n): one stream wait per call (mocked driver).
-
-All n base intervals are enqueued in order on the backend stream and the host
-waits once before step returns. Launch order and buffer arguments must equal n
-step(1) calls; completion metadata, raw-exchange and window ledgers, and failure
-invalidation are published only after that wait.
-"""
-from contextlib import nullcontext
-from unittest.mock import Mock
-import unittest
-
-import numpy as np
-
-from CelerisLab.config import LBMConfig
-from tests.unit.test_raw_exchange import body
-from tests.unit.test_refinement_three_level_multipatch import host_module
-
-BODY_LEVEL, HOPS = 3, 8  # body on L3: 2**3 hops per base interval
-
-
-def backend(window=False):
-    """Four-level host backend; every launch, memset and stream wait goes to one log."""
-    host = host_module()
-    log, allocations = [], []
-
-    def allocate(size):
-        allocations.append(Mock(size=size))
-        return allocations[-1]
-
-    module = Mock()
-    module.get_global.return_value = (Mock(), 20)
-    module.get_function.side_effect = lambda name: Mock(
-        side_effect=lambda *a, **k: log.append((name, a, k)))
-    host.compiler.load_module.return_value = module
-    host.cuda.mem_alloc.side_effect = allocate
-    host.cuda.memcpy_dtoh.side_effect = lambda array, pointer: array.fill(0)
-    host.cuda.memset_d8_async.side_effect = lambda *a: log.append(("memset", a, {}))
-    run = host.FixedPatchStepper(
-        LBMConfig(nx=48, ny=48, viscosity=.02), context=Mock(activate=nullcontext, sm_arch="sm_70"),
-        refinements=[[(6, 6, 26, 26)], [(20, 20, 44, 44)], [(56, 56, 72, 72)]],
-        level_blocks=[(4, 4)] * 4, level_objects=[(), (), (), (body(),)],
-        observe_raw_exchange=True, observe_window_exchange=window)
-    run._stream.synchronize.side_effect = lambda: log.append(("sync", (), {}))
-    run.initialize(*[np.full(level.shape, 1 / 9, np.float32) for level in run._levels])
-    log.clear()
-    return run, log, allocations
-
-
-def signature(log, allocations):
-    """Event names with device-pointer arguments replaced by allocation index."""
-    index = {id(a): i for i, a in enumerate(allocations)}
-    return [(name, tuple(index.get(id(a), "-") for a in args)) for name, args, _ in log]
-
-
-def names(log):
-    return [name for name, _, _ in log]
-
-
-class StepSync(unittest.TestCase):
-    def test_step1_waits_once_after_all_launches(self):
-        run, log, _ = backend()
-        run.step(1)
-        self.assertEqual(names(log).count("sync"), 1)
-        self.assertEqual(names(log)[-1], "sync")
-        self.assertEqual(names(log)[0], "memset")
-        self.assertEqual(names(log).count("BlockStep"), 1 + 2 + 4 + 8)
-        self.assertEqual(run.step_count, 1)
-        run.close()
-
-    def test_step_n_equals_n_step1_launches_with_one_wait(self):
-        single, single_log, single_alloc = backend()
-        for _ in range(3):
-            single.step(1)
-        batched, batched_log, batched_alloc = backend()
-        batched.step(3)
-        expected = [e for e in signature(single_log, single_alloc) if e[0] != "sync"]
-        got = signature(batched_log, batched_alloc)
-        self.assertEqual(got[-1][0], "sync")
-        self.assertEqual(names(batched_log).count("sync"), 1)
-        self.assertEqual(got[:-1], expected)  # same order, same ping-pong buffers
-        self.assertEqual(names(batched_log).count("memset"), 3)
-        self.assertEqual(batched.step_count, single.step_count)
-        index = [{id(a): i for i, a in enumerate(alloc)} for alloc in (single_alloc, batched_alloc)]
-        self.assertEqual(index[1][id(batched._levels[0].current)],
-                         index[0][id(single._levels[0].current)])
-        single.close()
-        batched.close()
-
-    def test_step0_launches_and_waits_for_nothing(self):
-        run, log, _ = backend()
-        run.step(2)
-        log.clear()
-        run.step(0)
-        self.assertEqual(log, [])
-        self.assertEqual(run.step_count, 2)
-        value = run.read_raw_exchange(0, level=BODY_LEVEL)  # step(0) keeps the ledger valid
-        self.assertEqual((value["time_start"], value["time_end"]), (1, 2))
-        run.close()
-
-    def test_completion_metadata_covers_all_intervals(self):
-        run, _, _ = backend()
-        with self.assertRaisesRegex(RuntimeError, "completed"):
-            run.read_raw_exchange(0, level=BODY_LEVEL)
-        run.step(3)
-        value = run.read_raw_exchange(0, level=BODY_LEVEL)
-        self.assertEqual((value["time_start"], value["time_end"], value["sample_count"]),
-                         (2, 3, HOPS))  # step(n) retains only the final base interval
-        run.step(2)
-        self.assertEqual(run.read_raw_exchange(0, level=BODY_LEVEL)["time_end"], 5)
-        run.close()
-
-    def test_window_accounting_across_batched_calls(self):
-        windows = []
-        for calls in ((1, 3), (4,), (1, 1, 1, 1)):
-            run, log, _ = backend(window=True)
-            run.step(1)
-            run.start_exchange_window()
-            log.clear()
-            for n in calls:
-                run.step(n)
-            self.assertEqual(names(log).count("BlockAccumulateExchange"), 2 * 4 * HOPS)
-            self.assertEqual(names(log).count("sync"), len(calls))
-            out = run.read_exchange_window(0, level=BODY_LEVEL)
-            windows.append((out["time_start"], out["time_end"], out["sample_count"]))
-            run.close()
-        self.assertEqual(windows, [(1., 5, 4 * HOPS)] * 3)
-
-    def test_failed_wait_publishes_nothing(self):
-        run, _, _ = backend(window=True)
-        run.step(2)
-        run.start_exchange_window()
-        run._stream.synchronize.side_effect = RuntimeError("device failure")
-        with self.assertRaisesRegex(RuntimeError, "device failure"):
-            run.step(3)
-        self.assertEqual(run.step_count, 2)  # value at entry to the failing call
-        self.assertFalse(run._exchange_valid)
-        self.assertFalse(run._window_valid)
-        for read in (lambda: run.read_raw_exchange(0, level=BODY_LEVEL),
-                     lambda: run.read_exchange_window(0, level=BODY_LEVEL), run.download,
-                     lambda: run.step(1)):
-            with self.assertRaisesRegex(RuntimeError, "failed"):
-                read()
-        run._stream.synchronize.side_effect = None
-        run.close()
-
-    def test_launch_error_mid_batch_skips_wait_and_fails(self):
-        run, log, _ = backend()
-        run.step(1)
-        log.clear()
-        hop, calls = run._hop, []
-
-        def failing(index, *args, **kw):
-            if index == 0:
-                calls.append(1)
-                if len(calls) == 2:
-                    raise RuntimeError("launch failure")
-            return hop(index, *args, **kw)
-        run._hop = failing
-        with self.assertRaisesRegex(RuntimeError, "launch failure"):
-            run.step(3)
-        self.assertNotIn("sync", names(log))
-        self.assertEqual(run.step_count, 1)
-        self.assertFalse(run._exchange_valid)
-        with self.assertRaisesRegex(RuntimeError, "failed"):
-            run.read_raw_exchange(0, level=BODY_LEVEL)
-        run.close()
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/validation/multigrid/test_refinement_three_level.py

@@ -80,11 +80,12 @@ class ThreeLevelCPU(unittest.TestCase):
         f, g = run._levels[1:]
         b = g.boundary
         events, feeds = [], []
-        advance, gather = run._advance_child, run._gather_halo
-        def record_feed(buffer, **kw):
-            self.assertIs(buffer, (f.current, f.spare)[len(feeds)])
-            feeds.append(buffer)
-            advance(buffer, **kw)
+        advance, gather = run._advance, run._gather_halo
+        def record_feed(index, buffer, **kw):  # index 2: L1 hop feeds L2
+            if index == 2:
+                self.assertIs(buffer, (f.current, f.spare)[len(feeds)])
+                feeds.append(buffer)
+            advance(index, buffer, **kw)
         def record_halo(level, buffer, **kw):
             if level is g:
                 events.append(("halo", buffer))
@@ -95,7 +96,7 @@ class ThreeLevelCPU(unittest.TestCase):
             self.assertIs(kind, g.kind)
             events.append(("step", src))
         g.kernel.side_effect = record_step
-        with patch.object(run, "_advance_child", side_effect=record_feed), \
+        with patch.object(run, "_advance", side_effect=record_feed), \
                 patch.object(run, "_gather_halo", side_effect=record_halo):
             run.step()
         self.assertEqual(feeds, [f.current, f.spare])
@@ -105,8 +106,8 @@ class ThreeLevelCPU(unittest.TestCase):
         self.assertGreater(len(np.unique(b.tables["block"])), 1)
         # Negative control: reusing L1's old source at the midpoint must fail.
         feeds.clear()
-        with patch.object(run, "_advance_child", side_effect=lambda buffer, **kw:
-                          record_feed(f.current, **kw)):
+        with patch.object(run, "_advance", side_effect=lambda index, buffer, **kw:
+                          record_feed(index, f.current if index == 2 else buffer, **kw)):
             with self.assertRaises(AssertionError):
                 run.step()
         run.close()
@@ -172,11 +173,12 @@ class ThreeLevelGPU(unittest.TestCase):
             p1, p2 = run.patch, run.child_patch
             run.initialize(state(16, 16, 1), state(*p1.fine_shape, 2), state(*p2.fine_shape, 3))
             f = run._levels[1]
-            feeds, original = [], run._advance_child
-            def record(buffer, **kw):
-                feeds.append(int(buffer))
-                original(buffer, **kw)
-            with patch.object(run, "_advance_child", side_effect=record):
+            feeds, original = [], run._advance
+            def record(index, buffer, **kw):  # index 2: L1 hop feeds L2
+                if index == 2:
+                    feeds.append(int(buffer))
+                original(index, buffer, **kw)
+            with patch.object(run, "_advance", side_effect=record):
                 cur, spare = int(f.current), int(f.spare)
                 run.step(1)
             self.assertEqual(feeds, [cur, spare])  # current G, then the new G of hop 1