examples: fixed_patch_circle reports its blocked tiles (fix unbound fine_block)

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

docs/development/refinement_mrt.md

@@ -25,7 +25,7 @@ under the alternate compile branch.
 - No LES, no external forcing (`d_params` force fields stay zero).
 - Shifted Jets stay SRT-only: `body/jets.py` keeps its existing guard. This extension does not
   provide shifted MRT Jets.
-- `PeriodicBlockStepper` (uniform blocked stepper) remains SRT-only.
+- `PeriodicBlockStepper` (uniform blocked stepper) remained SRT-only; it was removed in 8e1feff and its single-level capability, still SRT-only, is `FixedPatchStepper(refinements=[])`.
 
 ## Evidence status
 CPU: `tests/unit/test_refinement_mrt.py` (host guards, source contract). GPU:

docs/lifecycle.md

@@ -1,11 +1,11 @@
-# Device state lifecycle (uniform Simulation, PeriodicBlockStepper, FixedPatchStepper)
+# Device state lifecycle (uniform Simulation, FixedPatchStepper)
 
 State is grouped by what a change invalidates. This file describes the code as
 it is; it adds no framework. Line references are to candidate 0f8fdbc3.
 
 | Class | Contents | How it changes | Owner rule |
 |---|---|---|---|
-| 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 same quantity can be a macro in one route and a global in another (dense slot root: `WALL_SLOT_WIDTH`/`WALL_NORMAL_VELOCITY` macros; blocked slot root: `d_block_wall`), which can change generated arithmetic. | Every owner loads its own module (`compiler_v2.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`. |
+| 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 same quantity can be a macro in one route and a global in another (dense slot root: `WALL_SLOT_WIDTH`/`WALL_NORMAL_VELOCITY` macros; blocked slot root: `d_block_wall`), which can change generated arithmetic. | 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`). | Host copy commits only after the device upload succeeded; a device error fails the instance. |
 | 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. |
@@ -13,8 +13,8 @@ it is; it adds no framework. Line references are to candidate 0f8fdbc3.
 Storage routes are separate concerns:
 
 - Original uniform grid (`uniform.cu`, `Simulation`): whole-domain `NX`/`NY`.
-- Blocked levels (`blocked.cu`; `PeriodicBlockStepper` and every blocked
-  `FixedPatchStepper` level): `NX`/`NY` are one block window including halo.
+- Blocked levels (`blocked.cu`; every `FixedPatchStepper` level, including the
+  single level of `refinements=[]`): `NX`/`NY` are one block window including halo.
 - Dense refinement levels (`refinement.cu`, `level_blocks[l] is None`).
 
 A single-rectangle refinement (`FixedPatch2D`) is stored in its own fine

docs/supported_settings.md

@@ -15,7 +15,7 @@ measured; they are not general bounds.
 | Collision on refinement levels | SRT or MRT; TRT and LES rejected | `lbm/refinement.py` `FixedPatch2D.from_config` (re-checked for fine boundaries in `prepare_fine_boundary`) |
 | Shifted (ddf_shifting) storage | only when every level is blocked; dense or mixed rejected | `lbm/refinement_stepper.py` `FixedPatchStepper.__init__`, `FixedPatchStepper._check_storage`; `#error` guard in `lbm/kernels/refinement.cu` (dense patch kernel unshifted only) |
 | Fine relaxation | 1/tau_fine must lie inside the configured omega guards | `lbm/refinement.py` `FixedPatch2D.from_config` |
-| Uniform blocked backend `PeriodicBlockStepper` | doubly periodic, SRT only, stationary passive circles | `lbm/block_stepper.py` `PeriodicBlockStepper.__init__`, `_static_circles` |
+| Single blocked level `FixedPatchStepper(refinements=[], level_blocks=[(by, bx)], level_objects=[circles])` (replaces `PeriodicBlockStepper`, removed in 8e1feff) | doubly periodic, SRT only, no LES/Jets/projection/exchange observation, stationary passive circles; `download()` returns a 1-tuple | `lbm/refinement_stepper.py` `_check_levels`, `FixedPatchStepper.__init__`; `lbm/blocks.py` `static_circle_tables` |
 
 Tests: `tests/unit/test_refinement_mrt.py`, `tests/validation/multigrid/test_refinement_mrt_gpu.py`,
 `tests/unit/test_shifted_storage.py`, `tests/unit/test_shifted_block_operators.py`.

examples/fixed_patch_circle.py

@@ -1,9 +1,9 @@
 """Public API demo: fixed 2:1 patch with a stationary, optionally actuated circle.
 
 From this checkout, after checking the selected physical GPU is idle:
-    PYTHONPATH="$PWD/src" CUDA_VISIBLE_DEVICES=0 python examples/fixed_patch_circle.py --fine-block 16 20
-    ... --speed -0.002 --fine-block 16 20 --mode slot_reservoir
-Omit --fine-block for the (equivalent) dense fine storage.
+    PYTHONPATH="$PWD/src" CUDA_VISIBLE_DEVICES=0 python examples/fixed_patch_circle.py
+    ... --speed -0.002 --mode slot_reservoir
+Both levels use blocked storage (--base-block, --fine-block tiles).
 
 A short smoke run (default 128 coarse steps): it shows setup, stepping and
 fluid-only diagnostics, not a developed flow, accuracy or long-time stability.
@@ -71,8 +71,10 @@ def main():
     parser.add_argument("--mode", choices=("periodic", "slot_reservoir"), default="periodic")
     parser.add_argument("--speed", type=float, default=.002,
                         help="normal surface velocity, coarse lattice units (+ blows, - sucks, 0 passive)")
-    parser.add_argument("--fine-block", type=int, nargs=2, metavar=("BY", "BX"),
-                        help="blocked fine storage; must divide the whole fine allocation (32 x 40 here)")
+    parser.add_argument("--base-block", type=int, nargs=2, metavar=("BY", "BX"), default=(12, 8),
+                        help="root tile; must divide the 24 x 32 root")
+    parser.add_argument("--fine-block", type=int, nargs=2, metavar=("BY", "BX"), default=(16, 20),
+                        help="fine tile; must divide the whole fine allocation (32 x 40 here)")
     parser.add_argument("--steps", type=int, default=128)
     args = parser.parse_args()
     if args.steps < 0:
@@ -86,20 +88,20 @@ def main():
     body = SimObject(0, geometry=CircleGeometry(*CENTER, RADIUS), center=CENTER, radius=RADIUS,
                      surface_actuation=spec)
     cfg = config(args.mode)
-    fine_block = tuple(args.fine_block) if args.fine_block else None
     # The caller owns the CUDA context; the stepper must be closed first (the
     # nested with-statement does that).
     with CudaContext(0) as context, FixedPatchStepper(
             cfg, context=context, coarse_boundary=args.mode, refinements=[BOUNDS],
-            level_blocks=[None, fine_block], level_objects=[(), (body,)]) as run:
+            level_blocks=[tuple(args.base_block), tuple(args.fine_block)],
+            level_objects=[(), (body,)]) as run:
         p = run.patch
         # Rest equilibrium is already postcollision G, so no initial collision is
         # needed. Non-leaf entries (covered, ghost, solid) are discarded by initialize.
         rest = np.asarray(compute_feq_d2q9(1.0, 0.0, 0.0), np.float32)[:, None, None]
         run.initialize(np.broadcast_to(rest, (9, *p.coarse_shape)),
                        np.broadcast_to(rest, (9, *p.fine_shape)))
-        storage = f"blocked {fine_block}" if fine_block else "dense"
-        print(f"mode={args.mode} fine storage={storage} speed={args.speed:+g} "
+        storage = f"blocked root {tuple(args.base_block)} fine {tuple(args.fine_block)}"
+        print(f"mode={args.mode} storage={storage} speed={args.speed:+g} "
               f"(fine allocation {p.fine_shape[0]}x{p.fine_shape[1]}, fine origin {p.fine_origin})")
         masks = leaf_masks(run, args.mode)
         start = leaf_diagnostics(*run.download(), masks)

examples/fixed_patch_jet.py

@@ -31,6 +31,7 @@ from CelerisLab.lbm.refinement import ACTIVE
 from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
 
 BOUNDS, AMPLITUDE, WIDTH = (16, 4, 32, 28), .02, 12
+LEVEL_BLOCKS = [(12, 8), (8, 8)]  # blocked storage: 48x48 root, 56x40 fine window
 TIMES, MAX_MACH, MAX_SECONDS = (0, 300, 1200), .1, 120
 
 
@@ -132,7 +133,8 @@ def main():
                   lattice_model='D2Q9', collision='SRT', streaming='double_buffer',
                   store_precision='FP32', ddf_shifting=False, les_enabled=False)
     report = dict(config=dict(nx=48, ny=48, bounds=BOUNDS, W=WIDTH, A=AMPLITUDE, nu=.1,
-                              rho=1, tau_coarse=.8, tau_fine=1.1, coarse_boundary='slot_reservoir'),
+                              rho=1, tau_coarse=.8, tau_fine=1.1, coarse_boundary='slot_reservoir',
+                              level_blocks=LEVEL_BLOCKS),
                   guards=dict(check_every=50, positive_density=True, finite=True,
                               max_Mach=MAX_MACH, max_seconds=MAX_SECONDS,
                               timing='cooperative checks, not an exact interruption deadline'),
@@ -149,7 +151,8 @@ def main():
     start, run = time.monotonic(), None
     try:
         with CudaContext(0) as context, FixedPatchStepper(cfg, context=context,
-                coarse_boundary='slot_reservoir', refinements=[BOUNDS]) as run:
+                coarse_boundary='slot_reservoir', refinements=[BOUNDS],
+                level_blocks=LEVEL_BLOCKS) as run:
             p = run.patch
             masks = [np.ones(p.coarse_shape, bool), p.fine_kind == ACTIVE]
             masks[0][[0, -1]] = False

src/CelerisLab/body/manager.py

@@ -11,7 +11,7 @@ Responsibilities:
 
 Design::
   Packed ``obs`` layout matches ``generate_config(..., n_objects=count)`` and
-  ``config_obs.h`` macros. Host sizing uses :func:`~CelerisLab.cuda.compiler_v2.obs_layout`.
+  ``config_obs.h`` macros. Host sizing uses :func:`~CelerisLab.cuda.compiler.obs_layout`.
 
   Pending edits are lightweight: only the "intent" (add / remove) is stored.
   All derived quantities (flags, compact lists, GPU buffers) are built at
@@ -28,7 +28,7 @@ import pycuda.driver as cuda
 from .objects import SimObject
 from .registry import BodyRegistry
 from .preprocess.flag_overlay import merge_flag_masks, validate_cut_link_ownership
-from ..cuda.compiler_v2 import obs_layout
+from ..cuda.compiler import obs_layout
 from ..cuda.obs_layout import ObsLayout
 from ..lbm.descriptors import (
     FLUID, SOLID,

src/CelerisLab/config.py

@@ -7,7 +7,7 @@ Two JSON files:
   config_body.json – object list
 
 LBMConfig / BodyConfig are plain dataclasses; they translate to
-macro dicts consumed by `cuda/compiler_v2.py` for header generation.
+macro dicts consumed by `cuda/compiler.py` for header generation.
 """
 
 import json

src/CelerisLab/configs/CONFIG.md

@@ -60,14 +60,14 @@ Python `config.py` 只负责读取和校验，不是配置位置。显式配置
 
 **`lbm/kernels/config/*.h` 由编译器根据 `LBMConfig` 自动生成，请勿手改。**
 
-仓库中检入的 `config/*.h` 可能是历史版本。`compiler_v2.generate_config` 会复制 kernel 源码到独立临时构建目录，并在副本中生成配置；不会改写包内头文件或 PTX。返回目录须显式传给 `compile_kernel(build_dir=...)`，编译结果与配置摘要均保存在该目录。`Simulation` 负责目录清理；直接使用编译器的调用者须调用 `cleanup_build_dir()`。跳过生成配置直接编译仓库中的 `uniform.cu` 属不受支持用法。
+仓库中检入的 `config/*.h` 可能是历史版本。`compiler.generate_config` 会复制 kernel 源码到独立临时构建目录，并在副本中生成配置；不会改写包内头文件或 PTX。返回目录须显式传给 `compile_kernel(build_dir=...)`，编译结果与配置摘要均保存在该目录。`Simulation` 负责目录清理；直接使用编译器的调用者须调用 `cleanup_build_dir()`。跳过生成配置直接编译仓库中的 `uniform.cu` 属不受支持用法。
 
 ```
 config_lbm.json
   ↓  load_lbm_config()
 config.py:LBMConfig       (读取 + 校验)
   ↓  to_macros()
-cuda/compiler_v2.py       (独立构建目录中生成 config/*.h)
+cuda/compiler.py          (独立构建目录中生成 config/*.h)
   ↓
 config_grid.h             → NX, NY, NZ, NT, LATTICE_MODEL；
                             DIM / NQ 由 LATTICE_MODEL 推导写入（与 grid.lattice_model 一致，非 JSON 顶层字段）

src/CelerisLab/cuda/__init__.py

@@ -1,4 +1,4 @@
 """CUDA build helpers (compilation itself does not need a live context)."""
-from . import compiler_v2 as compiler
+from . import compiler
 
 __all__ = ["compiler"]

src/CelerisLab/cuda/compiler.py

@@ -1,4 +1,4 @@
-# CelerisLab/cuda/compiler_v2.py
+# CelerisLab/cuda/compiler.py
 """
 Kernel configuration, compilation, and module loading.
 
@@ -281,9 +281,9 @@ def build_module(cfg: LBMConfig, source: str, *, arch: str, n_objects: int = 0,
                  block_root_bc: int = 0):
     """Generate, compile and load one private module; the build directory is removed.
 
-    The single build path of every owner (Simulation, PeriodicBlockStepper,
-    FixedPatchStepper levels). Each call returns a NEW module: callers must not
-    share it, because its globals (d_params, ...) are mutable per module.
+    The single build path of every owner (Simulation, FixedPatchStepper levels).
+    Each call returns a NEW module: callers must not share it, because its
+    globals (d_params, ...) are mutable per module.
     """
     build_dir = generate_config(cfg, n_objects=n_objects, block_root_bc=block_root_bc)
     try:

src/CelerisLab/cuda/obs_layout.py

@@ -2,7 +2,7 @@
 """Packed ``obs`` buffer layout for force / torque / sensor telemetry.
 
 Usage::
-    Host code (e.g. ``ObjectManager``) and ``compiler_v2.generate_config`` must use
+    Host code (e.g. ``ObjectManager``) and ``compiler.generate_config`` must use
     the same ``dim`` and ``n_objects`` so allocation sizes match device macros.
 
 Design::

src/CelerisLab/lbm/block_stepper.py

@@ -1,287 +0,0 @@
-# CelerisLab/lbm/block_stepper.py
-"""Opt-in periodic backend for one BlockLayout2D level, optional static circles.
-
-D2Q9/SRT/FP32 unforced double buffers. Each step refreshes every halo from
-its owner core, then pulls/collides core cells into the spare buffer, then
-swaps. Only halo pair tables are uploaded; dense ownership stays on the host.
-The caller owns the CudaContext and must close this backend before it.
-Optional ``objects`` are stationary passive circles (static linear BFL via
-receiver-local staging: halo refresh -> cut links -> pull, no copy between).
-No force/actuation readout, motion, walls, refinement, checkpoint or runtime
-parameter mutation; wall/slot/inlet fields in cfg are ignored.
-"""
-from dataclasses import replace
-from numbers import Integral
-from threading import RLock
-
-import numpy as np
-import pycuda.driver as cuda
-
-from ..cuda import compiler_v2 as compiler
-from .blocks import NQ, BlockLayout2D, blocked_link_tables
-from .descriptors import D2Q9_EX, D2Q9_EY, FLUID, SOLID
-from .population_storage import _StoredPopulations, _encode_physical, _decode_stored
-
-MAX_GRID_DIM = 65535  # CUDA gridDim.y/z limit (rows, blocks)
-
-
-def _static_circles(layout, objects):
-    """Host tables for stationary circles; rejects before any CUDA work.
-
-    Every periodic neighbor of every fluid core is inspected, including domain
-    edges that CircleGeometry does not enumerate, so clipped or incomplete
-    geometry cannot pass. Each physical link has one receiver-owning block.
-    """
-    from ..body.geometry.circle import CircleGeometry
-
-    ny, nx = layout.shape
-    flags = np.full(ny * nx, FLUID, np.uint16)
-    for obj in objects:
-        if (type(obj.geometry) is not CircleGeometry or obj.is_sensor or obj.is_force_region
-                or obj.surface_actuation is not None
-                or any(getattr(obj.state, name) != 0 for name in ("vx", "vy", "omega"))
-                or any(vars(obj.control).values())):
-            raise ValueError("Block bodies must be stationary passive circles")
-        cx, cy = obj.geometry.center
-        radius = obj.geometry._radius
-        if not np.isfinite([cx, cy, radius]).all() or radius <= 0:
-            raise ValueError("Block circle center/radius must be finite with radius > 0")
-        mask = obj.geometry.build_flag_mask(nx, ny)
-        solid = (mask & SOLID) != 0
-        if not solid.any() or np.any(flags[solid] & SOLID):
-            raise ValueError("Block circles must be resolved and disjoint")
-        flags[solid] = mask[solid]
-    solid = ((flags & SOLID) != 0).reshape(ny, nx)
-    if solid[[0, -1]].any() or solid[:, [0, -1]].any():
-        raise ValueError("Block circles must not touch the domain edge")
-    links = [link for obj in objects
-             for link in obj.geometry.build_cut_links(nx, ny, domain_flags=flags)]
-    def source(r, q):
-        y, x = divmod(r, nx)
-        return (y + D2Q9_EY[q]) % ny, (x + D2Q9_EX[q]) % nx
-    receivers = {(int(link.fluid_idx), int(link.dir)) for link in links}
-    required = {(int(r), q) for r in np.flatnonzero(~solid.ravel())
-                for q in range(1, NQ) if solid[source(r, q)]}
-    if len(receivers) != len(links) or receivers != required:
-        raise ValueError("Cut links must cover every fluid pull from a solid exactly once")
-    tables, _ = blocked_link_tables(
-        layout, flags, np.array([link.fluid_idx for link in links], np.uint32),
-        np.array([link.dir for link in links], np.uint8),
-        np.array([link.q for link in links], np.float32),
-        np.array([link.fallback for link in links], np.uint8))
-    return dict(solid=solid, **tables)
-
-
-class PeriodicBlockStepper:
-    """G->G stepping of Q-major (9, ny, nx) postcollision populations.
-
-    On a device/launch failure the instance becomes unusable: close and
-    recreate it rather than assuming a partially advanced buffer rolled back.
-    """
-    def __init__(self, cfg, layout, *, context, objects=()):
-        if not isinstance(layout, BlockLayout2D) or layout.periodic != (True, True):
-            raise ValueError("PeriodicBlockStepper requires a doubly periodic BlockLayout2D")
-        checked = replace(cfg)
-        checked.validate()
-        if (not checked.is_d2q9 or checked.collision != "SRT"
-                or checked.streaming != "double_buffer" or checked.store_precision != "FP32"
-                or checked.les_enabled
-                or (checked.ny, checked.nx) != layout.shape):
-            raise ValueError("PeriodicBlockStepper requires matching D2Q9/SRT/FP32/"
-                             "double_buffer config without LES")
-        ly, lx = layout.local_shape
-        pairs = len(layout.halo_dst)
-        tpb = checked.threads_per_block
-        if layout.nblocks > MAX_GRID_DIM or ly > MAX_GRID_DIM or not 0 < pairs < 2**31:
-            raise ValueError("block count, block rows or halo pairs exceed launch limits")
-        objects = tuple(objects)  # an empty iterable takes the body-free path
-        self.boundary = _static_circles(layout, objects) if objects else None
-        links = 0 if self.boundary is None else len(self.boundary["cell"])
-        if links >= 2**31:
-            raise ValueError("cut links exceed launch limits")
-        self._ddf_shifting = bool(checked.ddf_shifting)
-        self.layout = layout
-        self.ctx = context
-        self._links = links
-        self._link_grid = ((links + tpb - 1) // tpb, 1)
-        # Wall/slot fields are not applied; neutralize them like the fine patch level.
-        self._local_cfg = replace(checked, nx=lx, ny=ly, y_wall_bc="bounce_back", top_wall_bc="velocity",
-                                  wall_slot_width=0, wall_normal_velocity=0.0, velocity=0.0)
-        self._block = (tpb, 1, 1)
-        self._grid = ((lx + tpb - 1) // tpb, ly, layout.nblocks)
-        self._copy_grid = ((pairs + tpb - 1) // tpb, 1)
-        self._bytes = int(np.prod(layout.blocked_shape)) * 4
-        self._lock = RLock()
-        self._status = "new"
-        self._step_count = 0
-        self._allocations = []
-        self._stream = self._module = self._copy = self._step = self._curved = None
-        self._tables = {}
-        with self.ctx.activate():
-            try:
-                free, _ = cuda.mem_get_info()
-                tables = sum(a.nbytes for k, a in (self.boundary or {}).items() if k != "solid")
-                if 2 * self._bytes + layout.halo_dst.nbytes + layout.halo_src.nbytes + tables > free:
-                    raise MemoryError("blocked buffers exceed free device memory")
-                self._stream = cuda.Stream()
-                config = self._local_cfg
-                config.validate()
-                if not np.isclose(config.omega, 1 / (3 * checked.viscosity + .5), rtol=1e-14, atol=0):
-                    raise RuntimeError("block omega/tau mismatch")
-                module = compiler.build_module(config, "blocked.cu", arch=self.ctx.sm_arch)
-                compiler.set_params(module, config.omega)
-                self._module = module
-                self._current = self._allocate(self._bytes)
-                self._spare = self._allocate(self._bytes)
-                self._dst = self._upload(layout.halo_dst)
-                self._src = self._upload(layout.halo_src)
-                self._copy = module.get_function("BlockHaloCopy")
-                self._step = module.get_function("BlockStep")
-                if self.boundary is not None:
-                    self._tables = {k: self._upload(a) for k, a in self.boundary.items() if k != "solid"}
-                    self._curved = module.get_function("BlockCurvedBoundary")
-            except BaseException:
-                try:
-                    self.close()
-                except Exception:
-                    pass  # Preserve the original construction failure.
-                raise
-
-    def _allocate(self, size):
-        pointer = cuda.mem_alloc(size)
-        self._allocations.append(pointer)
-        return pointer
-
-    def _upload(self, array):
-        array = np.ascontiguousarray(array)
-        pointer = self._allocate(array.nbytes)
-        cuda.memcpy_htod(pointer, array)
-        return pointer
-
-    def _require(self, *states):
-        if self._status not in states:
-            raise RuntimeError(f"Block stepper is {self._status}; expected {' or '.join(states)}")
-
-    @property
-    def step_count(self):
-        return self._step_count
-
-    def initialize(self, g):
-        """Upload dense G_0; the spare buffer starts NaN so stale halos cannot hide."""
-        with self._lock, self.ctx.activate():
-            self._require("new", "ready")
-            dense = np.array(g, dtype=np.float64 if self._ddf_shifting else np.float32,
-                             order="C", copy=True)
-            if dense.shape != (NQ, *self.layout.shape):
-                raise ValueError("Expected a Q-major (9, ny, nx) array")
-            fluid = np.ones(self.layout.shape, bool) if self.boundary is None else ~self.boundary["solid"]
-            if not np.isfinite(dense[:, fluid]).all() or (dense[:, fluid].sum(axis=0) <= 0).any():
-                raise ValueError("Fluid populations must be finite with positive density")
-            dense[:, ~fluid] = np.nan  # Solid cores are deliberately invalid.
-            encoded = _encode_physical(dense, self._ddf_shifting)
-            values = encoded[:, fluid]
-            if (not np.isfinite(values).all()
-                    or (values.sum(axis=0, dtype=np.float64) + int(self._ddf_shifting) <= 0).any()):
-                raise ValueError("Stored fluid populations must be finite with positive density")
-            blocked = self.layout.to_blocked(encoded)
-            spare = np.full_like(blocked, np.nan)
-            try:
-                cuda.memcpy_htod(self._current, blocked)
-                cuda.memcpy_htod(self._spare, spare)
-                self._step_count = 0
-                self._status = "ready"
-            except BaseException:
-                self._status = "failed"
-                raise
-
-    def step(self, n=1):
-        """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"
-                for _ in range(n):
-                    self._copy(self._current, self._dst, self._src,
-                               np.uint32(len(self.layout.halo_dst)), grid=self._copy_grid, **kw)
-                    if self._curved is not None:
-                        t = self._tables
-                        self._curved(self._current, t["block"], t["cell"], t["dir"], t["q"],
-                                     t["fallback"], np.uint32(self._links), np.uintp(0),  # no actuation
-                                     np.uintp(0), np.uintp(0), np.uintp(0),  # no observation
-                                     grid=self._link_grid, **kw)
-                    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._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"
-                raise
-
-    def download(self):
-        """Return physical core populations (FP64 when shifted); no host cache."""
-        snapshot = self._download_raw()
-        return _decode_stored(snapshot.values[0], snapshot.shifted)
-
-    def _download_raw(self):
-        """Return one tagged dense FP32 population array without conversion."""
-        with self._lock, self.ctx.activate():
-            self._require("ready")
-            blocked = np.empty(self.layout.blocked_shape, np.float32)
-            try:
-                cuda.memcpy_dtoh(blocked, self._current)
-            except BaseException:
-                self._status = "failed"
-                raise
-            return _StoredPopulations((self.layout.to_dense(blocked),), self._ddf_shifting)
-
-    def close(self):
-        """Release owned resources; retain them for retry if context activation fails."""
-        with self._lock:
-            if self._status == "closed":
-                return
-            self._status = "failed"
-            with self.ctx.activate():
-                self._status = "closed"
-                error = None
-                try:
-                    if self._stream is not None:
-                        self._stream.synchronize()
-                except Exception as exc:
-                    error = exc
-                self._copy = self._step = self._curved = self._module = None
-                self._tables = {}
-                for pointer in self._allocations:
-                    try:
-                        pointer.free()
-                    except Exception as exc:
-                        error = error or exc
-                self._allocations.clear()
-                self._stream = None
-                if error is not None:
-                    raise error
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, *exc):
-        try:
-            self.close()
-        except Exception:
-            if exc[0] is None:
-                raise

src/CelerisLab/lbm/blocks.py

@@ -31,7 +31,7 @@ receiver-local boundary values: the solid's owner-core copy of that slot may
 be stale, which is distinct from the unique core ownership of stored state.
 No moving geometry is implied.
 """
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from numbers import Integral
 from types import SimpleNamespace
 
@@ -373,3 +373,84 @@ def blocked_link_tables(layout, flags, fluid_idx, dir, q, fallback):
                   dir=columns[1].astype(np.uint8), q=columns[2].astype(np.float32),
                   fallback=columns[3].astype(np.uint8))
     return tables, mapped
+
+
+# One blocked level as FixedPatchStepper builds and launches it (any depth). Raw argument orders of the blocked.cu kernels live only here.
+
+def level_config(cfg, layout, tau):
+    """Kernel config of one blocked level: NX/NY = block window incl. halo.
+
+    Physical wall/slot/inlet fields are neutralized; a blocked root gets them
+    from its module globals (d_block_wall/d_block_channel) instead.
+    """
+    ly, lx = layout.local_shape
+    local = replace(cfg, nx=lx, ny=ly, y_wall_bc="bounce_back", top_wall_bc="velocity",
+                    wall_slot_width=0, wall_normal_velocity=0.0, velocity=0.0)
+    local.validate()
+    if not np.isclose(local.omega, 1 / tau, rtol=1e-14, atol=0):
+        raise RuntimeError("block level omega/tau mismatch")
+    return local
+
+
+def launch_halo_copy(kernel, buffer, dst, src, count, *, block, stream):
+    """BlockHaloCopy over ``count`` uploaded (dst, src) block-cell pairs."""
+    kernel(buffer, dst, src, np.uint32(count),
+           grid=((count + block[0] - 1) // block[0], 1), block=block, stream=stream)
+
+
+def launch_curved_boundary(kernel, buffer, tables, count, *, block, stream,
+                           velocity=None, rx=None, ry=None, raw=None):
+    """BlockCurvedBoundary on uploaded link tables; None = no actuation/observation."""
+    none = np.uintp(0)
+    kernel(buffer, tables["block"], tables["cell"], tables["dir"], tables["q"], tables["fallback"],
+           np.uint32(count), none if velocity is None else velocity,
+           none if rx is None else rx, none if ry is None else ry, none if raw is None else raw,
+           grid=((count + block[0] - 1) // block[0], 1), block=block, stream=stream)
+
+
+def static_circle_tables(layout, objects):
+    """Host tables for stationary circles; rejects before any CUDA work.
+
+    Every periodic neighbor of every fluid core is inspected, including domain
+    edges that CircleGeometry does not enumerate, so clipped or incomplete
+    geometry cannot pass. Each physical link has one receiver-owning block.
+    """
+    from ..body.geometry.circle import CircleGeometry
+    from .descriptors import FLUID, SOLID
+
+    ny, nx = layout.shape
+    flags = np.full(ny * nx, FLUID, np.uint16)
+    for obj in objects:
+        if (type(obj.geometry) is not CircleGeometry or obj.is_sensor or obj.is_force_region
+                or obj.surface_actuation is not None
+                or any(getattr(obj.state, name) != 0 for name in ("vx", "vy", "omega"))
+                or any(vars(obj.control).values())):
+            raise ValueError("Block bodies must be stationary passive circles")
+        cx, cy = obj.geometry.center
+        radius = obj.geometry._radius
+        if not np.isfinite([cx, cy, radius]).all() or radius <= 0:
+            raise ValueError("Block circle center/radius must be finite with radius > 0")
+        mask = obj.geometry.build_flag_mask(nx, ny)
+        solid = (mask & SOLID) != 0
+        if not solid.any() or np.any(flags[solid] & SOLID):
+            raise ValueError("Block circles must be resolved and disjoint")
+        flags[solid] = mask[solid]
+    solid = ((flags & SOLID) != 0).reshape(ny, nx)
+    if solid[[0, -1]].any() or solid[:, [0, -1]].any():
+        raise ValueError("Block circles must not touch the domain edge")
+    links = [link for obj in objects
+             for link in obj.geometry.build_cut_links(nx, ny, domain_flags=flags)]
+    def source(r, q):
+        y, x = divmod(r, nx)
+        return (y + D2Q9_EY[q]) % ny, (x + D2Q9_EX[q]) % nx
+    receivers = {(int(link.fluid_idx), int(link.dir)) for link in links}
+    required = {(int(r), q) for r in np.flatnonzero(~solid.ravel())
+                for q in range(1, NQ) if solid[source(r, q)]}
+    if len(receivers) != len(links) or receivers != required:
+        raise ValueError("Cut links must cover every fluid pull from a solid exactly once")
+    tables, _ = blocked_link_tables(
+        layout, flags, np.array([link.fluid_idx for link in links], np.uint32),
+        np.array([link.dir for link in links], np.uint8),
+        np.array([link.q for link in links], np.float32),
+        np.array([link.fallback for link in links], np.uint8))
+    return dict(solid=solid, **tables)

src/CelerisLab/lbm/refinement_stepper.py

@@ -14,15 +14,16 @@ from threading import RLock
 import numpy as np
 import pycuda.driver as cuda
 
-from ..cuda import compiler_v2 as compiler
-from .blocks import BlockLayout2D
+from ..cuda import compiler
+from .blocks import (BlockLayout2D, launch_curved_boundary, launch_halo_copy, level_config,
+                     static_circle_tables)
 from .channel import channel_masks, validate_channel_config, validate_channel_transfers
 from ..body.jets import jet_link_velocity
-from .block_stepper import MAX_GRID_DIM
 from .hierarchy import build_hierarchy, is_rectangle
 from .refinement import ACTIVE, HALO, FixedPatch2D, FixedPatchSet2D, prepare_fine_boundary
 from .refinement_tables import (blocked_base_tables, blocked_fine_tables,  # noqa: F401 (re-export)
                                 blocked_kind, projected_explosion_tables, _prepare_blocked_boundary,
+                                _BlockedBoundary,
                                 BLOCKED_BOUNDARY_DEVICE_COLUMNS)
 from .descriptors import SOLID
 from .raw_exchange import summarize_exchange
@@ -65,17 +66,21 @@ def _level_name(level):
     return ("base", "fine", "child")[level] if level < 3 else f"level {level}"
 
 
+MAX_GRID_DIM = 65535  # CUDA gridDim.y/z limit (rows, blocks)
+
 def _check_levels(refinements, blocks, objects, jets):
     """Candidate limits of the level lists, checked before any device work."""
     depth = len(refinements) + 1
-    if depth < 2:
-        raise ValueError("at least one refinement level is required")
     for name, values in (("level_blocks", blocks), ("level_objects", objects), ("level_jets", jets)):
         if len(values) != depth:
             raise ValueError(f"{name} needs one entry per level ({depth})")
     for level, block in enumerate(blocks):
         if block is not None and len(block) != 2:
             raise ValueError(f"level_blocks[{level}] must be (by, bx)")
+    if depth == 1:  # one blocked level: the root carries the static circles
+        if blocks[0] is None or jets[0]:
+            raise ValueError("a single level must be blocked and has no Jets")
+        return
     if objects[0] or jets[0]:
         raise ValueError("base-level bodies/Jets are not supported")
     if any(blocks[level - 1] is not None and blocks[level] is None for level in range(1, depth)):
@@ -98,6 +103,9 @@ class FixedPatchStepper:
     level-wide grid, public fine arrays (2*ny, 2*nx), NaN outside ACTIVE cells);
     see lbm.hierarchy. ``level_blocks``, ``level_objects`` and ``level_jets``
     hold one entry per level, L0 included (L0 bodies/Jets are not supported).
+    ``refinements=[]`` is one blocked level: periodic SRT root without LES,
+    Jets or exchange observation; its ``level_objects[0]`` are stationary
+    passive circles (static BFL).
 
     ``coarse_boundary`` is ``periodic`` (default) or ``slot_reservoir``:
     a bottom on-site slot and top fixed-density reservoir from ``cfg``.
@@ -196,6 +204,11 @@ class FixedPatchStepper:
         self._ddf_shifting = bool(cfg.ddf_shifting)
         if self._ddf_shifting and any(b is None for b in level_blocks):
             raise ValueError("shifted populations require all-blocked levels; dense/mixed is unsupported")
+        if depth == 1 and (coarse_boundary != "periodic" or cfg.collision != "SRT" or cfg.les_enabled
+                           or explosion != "constant"
+                           or observe_raw_exchange or observe_window_exchange):
+            raise ValueError("a single level requires a periodic SRT root without LES, projection "
+                             "or exchange observation")
         self._channel = coarse_boundary == "channel"
         if self._channel:
             validate_channel_config(cfg, level_blocks)
@@ -229,12 +242,14 @@ class FixedPatchStepper:
             base = blocked_base_tables(pairs[0], self._layouts[0], np.argsort(pairs[0].coalesce_coarse))
             validate_channel_transfers(pairs[0], self._layouts[0], base[1])
         # Kernel configurations per level (geometry configs live in self.hierarchy).
-        fine_cfg = replace(cfg, nx=pairs[0].fine_shape[1], ny=pairs[0].fine_shape[0],
-                           viscosity=2 * cfg.viscosity)
-        if self._wall_aware:
-            fine_cfg = replace(fine_cfg, y_wall_bc="bounce_back", top_wall_bc="velocity",
-                               wall_slot_width=0, wall_normal_velocity=0.0, velocity=0.0)
-        self._level_cfgs = [replace(cfg), fine_cfg]
+        self._level_cfgs = [replace(cfg)]
+        if pairs:
+            fine_cfg = replace(cfg, nx=pairs[0].fine_shape[1], ny=pairs[0].fine_shape[0],
+                               viscosity=2 * cfg.viscosity)
+            if self._wall_aware:
+                fine_cfg = replace(fine_cfg, y_wall_bc="bounce_back", top_wall_bc="velocity",
+                                   wall_slot_width=0, wall_normal_velocity=0.0, velocity=0.0)
+            self._level_cfgs.append(fine_cfg)
         for pair in pairs[1:]:
             parent = self._level_cfgs[-1]
             self._level_cfgs.append(replace(parent, nx=pair.fine_shape[1], ny=pair.fine_shape[0],
@@ -275,12 +290,18 @@ class FixedPatchStepper:
         self._exchange_centers = tuple(tuple(tuple(o.geometry.center) for o in objects)
                                        for objects in level_objects)
         self._level_jets = level_jets
-        self._level_boundaries = [None] + [
+        root = None
+        if depth == 1 and level_objects[0]:
+            tables = static_circle_tables(self._layouts[0], level_objects[0])
+            root = _BlockedBoundary(None, tables, (), False)
+        self._level_boundaries = [root] + [
             _prepare_blocked_boundary(self._boundary_hosts[level], self._layouts[level],
                                       self._level_cfgs[level], level_jets[level],
                                       _shifted_blocked=self._ddf_shifting)
             for level in range(1, depth)]
         self._leaf = [kind == ACTIVE for kind in self.hierarchy.kinds]
+        if root is not None:
+            self._leaf[0] &= ~root.tables["solid"]
         for level in range(1, depth):
             host = self._boundary_hosts[level]
             if host is not None:
@@ -303,6 +324,9 @@ class FixedPatchStepper:
         self._block = (cfg.threads_per_block, 1, 1)
         with self.ctx.activate():
             try:
+                free, _ = cuda.mem_get_info()
+                if self._planned_bytes() > free:  # before any stream, module or buffer
+                    raise MemoryError("level buffers exceed free device memory")
                 self._stream = cuda.Stream()
                 for level in range(depth):
                     self._levels.append(self._build_level(level))
@@ -324,7 +348,9 @@ class FixedPatchStepper:
                                 b.window_integral = self._allocate(count * 4 * 8)
                             b.accumulate = level.module.get_function("BlockAccumulateExchange")
                         b.kernel = level.module.get_function("BlockCurvedBoundary")
-                if self._level_boundaries[1] is not None:
+                if depth == 1:
+                    pass
+                elif self._level_boundaries[1] is not None:
                     b = self._level_boundaries[1]
                     self._fine_flags, self._link_tables, self._curved = b.flags, b.device, b.kernel
                 elif self._boundary_hosts[1] is not None:
@@ -346,20 +372,39 @@ class FixedPatchStepper:
                     pass  # Preserve the original construction failure.
                 raise
 
+    def _planned_bytes(self):
+        """Both G buffers, halo pairs and boundary columns of every level (transfer tables excluded).
+
+        A fail-fast lower bound only: passing it cannot guarantee that later
+        allocations succeed (other consumers, fragmentation, races).
+        """
+        total = 0
+        for index, layout in enumerate(self._layouts):
+            cells = int(np.prod(layout.blocked_shape if layout is not None
+                                else (9, *self.hierarchy.kinds[index].shape)))
+            total += 2 * 4 * cells
+            if layout is not None:
+                total += layout.halo_dst.nbytes + layout.halo_src.nbytes
+            b = self._level_boundaries[index]
+            if b is not None:
+                total += sum(np.asarray(v).nbytes for k, v in b.tables.items()
+                             if k == "flags" or k in BLOCKED_BOUNDARY_DEVICE_COLUMNS)
+        return total
+
     def _build_level(self, index):
         """Compile one level's step module and allocate its two G buffers."""
         layout, config = self._layouts[index], self._level_cfgs[index]
         pairs = self.hierarchy.pairs
-        tau = pairs[0].tau_coarse if index == 0 else pairs[index - 1].tau_fine
+        tau = (pairs[0].tau_coarse if pairs else 3 * config.viscosity + .5) if index == 0 \
+            else pairs[index - 1].tau_fine
         name, source = ("PatchCoarseStep" if index == 0 else "PatchFineStep"), "refinement.cu"
         if layout is not None:
-            config = replace(config, nx=layout.local_shape[1], ny=layout.local_shape[0],
-                             y_wall_bc="bounce_back", top_wall_bc="velocity",
-                             wall_slot_width=0, wall_normal_velocity=0.0, velocity=0.0)
+            config = level_config(config, layout, tau)
             name, source = ("BlockChannelStep" if index == 0 and self._channel else "BlockStep"), "blocked.cu"
-        config.validate()
-        if not np.isclose(config.omega, 1 / tau, rtol=1e-14, atol=0):
-            raise RuntimeError("level omega/tau mismatch")
+        else:
+            config.validate()
+            if not np.isclose(config.omega, 1 / tau, rtol=1e-14, atol=0):
+                raise RuntimeError("level omega/tau mismatch")
         host = self._boundary_hosts[index]
         n_objects = len(host.objects) if name == "PatchFineStep" and host is not None else 0
         block_root_bc = 0
@@ -512,7 +557,9 @@ class FixedPatchStepper:
             try:
                 for level, array in zip(self._levels, arrays):
                     cuda.memcpy_htod(level.current, array)
-                    cuda.memcpy_htod(level.spare, array)
+                    # A single level starts its spare NaN so stale halos cannot hide.
+                    cuda.memcpy_htod(level.spare, np.full_like(array, np.nan)
+                                     if len(self._levels) == 1 else array)
                 if self._fine_curved is not None and self._fine_curved.link_exchange_gpu is not None:
                     cuda.memset_d8(self._fine_curved.link_exchange_gpu, 0, self._fine_curved.count * 12)
                 self._step_count = 0
@@ -676,15 +723,12 @@ class FixedPatchStepper:
             return
         if b.kernel is None or b.flags is None:
             raise RuntimeError("Boundary device buffers are missing")
-        t, count = b.device, len(b.tables["cell"])
+        t = b.device
         velocity = t.get("surface_velocity")
         if b.actuated and velocity is None:
             raise RuntimeError("Surface velocity device buffer is missing")
-        b.kernel(buffer, t["block"], t["cell"], t["dir"], t["q"], t["fallback"],
-                 np.uint32(count), velocity if velocity is not None else np.uintp(0),
-                 t.get("rx", np.uintp(0)), t.get("ry", np.uintp(0)),
-                 b.last_raw if b.last_raw is not None else np.uintp(0),
-                 grid=((count + self._block[0] - 1) // self._block[0], 1), **kw)
+        launch_curved_boundary(b.kernel, buffer, t, len(b.tables["cell"]), velocity=velocity,
+                               rx=t.get("rx"), ry=t.get("ry"), raw=b.last_raw, **kw)
 
     def _advance(self, index, parent_buffer, **kw):
         """Explode from the given parent buffer, then two hops of level ``index``."""
@@ -736,13 +780,13 @@ class FixedPatchStepper:
                              np.float64(h), np.float64(h),
                              grid=((count + self._block[0] - 1) // self._block[0], 1), **kw)
         flags = np.uintp(0) if level.boundary is None else level.boundary.flags
-        level.kernel(src, dst, np.uint32(level.layout.halo), flags, level.kind, grid=level.grid, **kw)
+        kind = np.uintp(0) if level.kind is None else level.kind  # single level: all ACTIVE
+        level.kernel(src, dst, np.uint32(level.layout.halo), flags, kind, grid=level.grid, **kw)
 
     def _gather_halo(self, level, buffer, **kw):
         pairs = len(level.layout.halo_dst)
         if pairs:
-            level.halo_copy(buffer, level.halo_dst, level.halo_src, np.uint32(pairs),
-                            grid=((pairs + self._block[0] - 1) // self._block[0], 1), **kw)
+            launch_halo_copy(level.halo_copy, buffer, level.halo_dst, level.halo_src, pairs, **kw)
 
     def step(self, n=1):
         """Advance G→G by n base intervals: level l takes 2**l hops per interval.

src/CelerisLab/simulation.py

@@ -25,7 +25,7 @@ import pycuda.driver as cuda
 
 from .config import LBMConfig, BodyConfig, load_lbm_config, load_body_config
 from .cuda.context import CudaContext, context_bound
-from .cuda import compiler_v2 as compiler
+from .cuda import compiler
 from .lbm.field import LBMField
 from .lbm.stepper import LBMStepper
 from .body.objects import SimObject

tests/integration/test_runtime_contracts_gpu.py

@@ -6,14 +6,14 @@ import pycuda.driver as cuda
 import pytest
 
 import CelerisLab
-from CelerisLab.cuda import compiler_v2
+from CelerisLab.cuda import compiler
 from ._cases import make_sim
 
 
 def test_source_provenance():
     root = Path(__file__).resolve().parents[2] / "src/CelerisLab"
     assert Path(CelerisLab.__file__).resolve().parent == root
-    assert Path(compiler_v2.kernel_path("uniform.cu")).resolve() == root / "lbm/kernels/uniform.cu"
+    assert Path(compiler.kernel_path("uniform.cu")).resolve() == root / "lbm/kernels/uniform.cu"
 
 
 def test_two_contexts_interleaved_and_non_lifo_close():

tests/unit/test_block_step_sync.py

@@ -1,104 +0,0 @@
-"""Host contract of PeriodicBlockStepper.step(n): one stream wait per call (mocked driver)."""
-import ast
-from contextlib import nullcontext
-import importlib.util
-from pathlib import Path
-import types
-from unittest.mock import Mock
-import unittest
-
-import numpy as np
-
-from CelerisLab.config import LBMConfig
-from CelerisLab.lbm.blocks import BlockLayout2D
-
-
-def host_module():
-    """block_stepper with pycuda/compiler replaced by mocks; no GPU context."""
-    # Locate the source without importing it: importing block_stepper loads pycuda.driver,
-    # which CPU-only import-boundary tests (test_build_contracts) must never see.
-    path = Path(importlib.util.find_spec("CelerisLab.lbm.block_stepper").origin)
-    tree = ast.parse(path.read_text())
-    tree.body = [node for node in tree.body if not (
-        isinstance(node, ast.Import) and any(a.name == "pycuda.driver" for a in node.names)
-        or isinstance(node, ast.ImportFrom) and node.module == "cuda")]
-    module = types.ModuleType("CelerisLab.lbm._test_host_block_stepper")
-    module.__package__ = "CelerisLab.lbm"
-    module.__dict__.update(cuda=Mock(), compiler=Mock())
-    exec(compile(tree, str(path), "exec"), module.__dict__)
-    return module
-
-
-def backend():
-    host, log = host_module(), []
-    host.cuda.mem_get_info.return_value = (1 << 40, 1 << 40)
-    host.cuda.mem_alloc.side_effect = lambda size: Mock(size=size)  # distinct buffers
-    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[:2])))
-    host.compiler.build_module.return_value = module
-    cfg = LBMConfig(nx=16, ny=16, viscosity=.1)
-    run = host.PeriodicBlockStepper(cfg, BlockLayout2D.build(16, 16, 8, 8),
-                                    context=Mock(activate=nullcontext, sm_arch="sm_70"))
-    run._stream.synchronize.side_effect = lambda: log.append(("sync", ()))
-    run.initialize(np.full((9, 16, 16), 1 / 9))
-    log.clear()
-    return run, log
-
-
-class BlockStepSync(unittest.TestCase):
-    def test_step_n_waits_once_and_matches_n_single_steps(self):
-        many, log_many = backend()
-        many.step(3)
-        single, log_single = backend()
-        for _ in range(3):
-            single.step(1)
-        names = [name for name, _ in log_many]
-        self.assertEqual(names.count("sync"), 1)
-        self.assertEqual(names[-1], "sync")
-        self.assertEqual(names.count("BlockStep"), 3)
-        # Same launch order and buffer roles as three step(1) calls.
-        strip = lambda log: [(n, tuple(id(x) for x in a)) for n, a in log if n != "sync"]
-        roles = lambda run, log: [(n, tuple({id(run._current): "c", id(run._spare): "s"}.get(x, "-")
-                                            for x in ids)) for n, ids in strip(log)]
-        self.assertEqual([n for n, _ in strip(log_many)], [n for n, _ in strip(log_single)])
-        self.assertEqual(roles(many, log_many), roles(single, log_single))
-        self.assertIsNot(many._current, many._spare)
-        self.assertEqual((many.step_count, single.step_count), (3, 3))
-
-    def test_step_zero_launches_and_waits_for_nothing(self):
-        run, log = backend()
-        run.step(0)
-        self.assertEqual((log, run.step_count), ([], 0))
-
-    def test_failure_keeps_entry_step_count(self):
-        run, log = backend()
-        run.step(1)
-        run._stream.synchronize.side_effect = RuntimeError("device fault")
-        with self.assertRaises(RuntimeError):
-            run.step(2)
-        self.assertEqual((run.step_count, run._status), (1, "failed"))
-
-    def test_launch_failure_mid_batch_fails_closed_without_waiting(self):
-        run, log = backend()
-        calls = []
-        step = run._step
-
-        def failing(*args, **kwargs):
-            calls.append(1)
-            if len(calls) == 2:
-                raise RuntimeError("launch failure")
-            return step(*args, **kwargs)
-
-        run._step = failing
-        with self.assertRaises(RuntimeError):
-            run.step(3)
-        self.assertNotIn("sync", [name for name, _ in log])
-        self.assertEqual((run.step_count, run._status), (0, "failed"))
-        with self.assertRaises(RuntimeError):
-            run.download()
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_example_tiles.py

@@ -0,0 +1,65 @@
+"""The public examples run all-blocked: their level tiles must reach the device unchanged.
+
+Host only: a context whose activation raises proves construction (layouts, tables,
+boundary checks) succeeded up to the first CUDA call.
+"""
+import ast
+from dataclasses import replace
+from pathlib import Path
+import unittest
+from unittest.mock import Mock
+
+from CelerisLab.body.geometry.circle import CircleGeometry
+from CelerisLab.body.objects import SimObject
+from CelerisLab.config import LBMConfig
+from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
+
+EXAMPLES = Path(__file__).resolve().parents[2] / "examples"
+
+
+def literal(name, constant):
+    """Module-level literal from an example, read without importing it (no pycuda)."""
+    tree = ast.parse((EXAMPLES / name).read_text())
+    for node in tree.body:
+        if isinstance(node, ast.Assign):
+            for target in node.targets:
+                names = target.elts if isinstance(target, ast.Tuple) else [target]
+                for i, t in enumerate(names):
+                    if isinstance(t, ast.Name) and t.id == constant:
+                        value = node.value.elts[i] if isinstance(target, ast.Tuple) else node.value
+                        return ast.literal_eval(value)
+    raise KeyError(constant)
+
+
+def forbidden():
+    return Mock(activate=Mock(side_effect=AssertionError("CUDA reached")), sm_arch="sm_70")
+
+
+class ExampleTiles(unittest.TestCase):
+    def test_fixed_patch_jet_tiles(self):
+        bounds, blocks = literal("fixed_patch_jet.py", "BOUNDS"), literal("fixed_patch_jet.py", "LEVEL_BLOCKS")
+        width, amplitude = literal("fixed_patch_jet.py", "WIDTH"), literal("fixed_patch_jet.py", "AMPLITUDE")
+        cfg = LBMConfig(nx=48, ny=48, viscosity=.1, velocity=0., wall_slot_width=width,
+                        wall_normal_velocity=amplitude, y_wall_bc="on_site_velocity",
+                        top_wall_bc="reservoir")
+        with self.assertRaisesRegex(AssertionError, "CUDA reached"):
+            FixedPatchStepper(cfg, context=forbidden(), coarse_boundary="slot_reservoir",
+                              refinements=[bounds], level_blocks=[tuple(b) for b in blocks])
+
+    def test_fixed_patch_circle_default_tiles(self):
+        bounds = literal("fixed_patch_circle.py", "BOUNDS")
+        nx, ny = literal("fixed_patch_circle.py", "NX"), literal("fixed_patch_circle.py", "NY")
+        center, radius = literal("fixed_patch_circle.py", "CENTER"), literal("fixed_patch_circle.py", "RADIUS")
+        body = SimObject(0, geometry=CircleGeometry(*center, radius), center=center, radius=radius)
+        periodic = LBMConfig(nx=nx, ny=ny, viscosity=.1)
+        slot = replace(periodic, velocity=0, wall_normal_velocity=.001, wall_slot_width=8,
+                       y_wall_bc="on_site_velocity", top_wall_bc="reservoir")
+        for mode, cfg in (("periodic", periodic), ("slot_reservoir", slot)):
+            with self.subTest(mode=mode), self.assertRaisesRegex(AssertionError, "CUDA reached"):
+                FixedPatchStepper(cfg, context=forbidden(), coarse_boundary=mode,
+                                  refinements=[bounds], level_blocks=[(12, 8), (16, 20)],
+                                  level_objects=[(), (body,)])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_planar_velocity.py

@@ -9,7 +9,7 @@ import pytest
 
 import CelerisLab
 from CelerisLab.config import LBMConfig, load_lbm_config
-from CelerisLab.cuda import compiler_v2 as compiler
+from CelerisLab.cuda import compiler
 from CelerisLab.common.checkpoint import save_checkpoint, load_checkpoint
 from .test_numerical_operators import compile_run
 from .test_state_regressions import device, make_field, manager

tests/unit/test_refined_dfg.py

@@ -164,26 +164,11 @@ def frozen_manifest(observation_mode):
 
 
 def selected_manifest(collision, observation_mode):
-    """verify_source's manifest selection with its working-tree hash comparison neutralised.
+    """The frozen manifest verify_source selects for (collision, observation_mode).
 
-    Source identity is owned by test_source_pin; this exercises only which frozen manifest
-    verify_source selects and returns for (collision, observation_mode).
+    Source identity is owned by test_source_pin; this reads no checkout files.
     """
-    from unittest.mock import patch
-
-    class Digest(str):
-        def __ne__(self, other):
-            return False
-
-    class Hash:
-        def __init__(self, data=b""):
-            pass
-
-        def hexdigest(self):
-            return Digest()
-
-    with patch.object(r.hashlib, "sha256", Hash):
-        return r.verify_source(r.DEFAULT_SOURCE, collision, observation_mode)
+    return r.source_manifest(collision, observation_mode)
 
 
 def test_full_requires_completed_pilot():

tests/unit/test_refinement_three_level_multipatch.py

@@ -22,12 +22,13 @@ def host_module():
     tree.body=[node for node in tree.body if not (
         isinstance(node,ast.Import) and any(a.name=='pycuda.driver' for a in node.names)
         or isinstance(node,ast.ImportFrom) and node.module in (
-            'cuda','block_stepper','curved_links'))]
+            'cuda','curved_links'))]
     module=types.ModuleType('CelerisLab.lbm._test_host_refinement')
     module.__package__='CelerisLab.lbm'
     from CelerisLab.body.jets import surface_link_velocity
-    module.__dict__.update(cuda=Mock(),compiler=Mock(),MAX_GRID_DIM=65535,
+    module.__dict__.update(cuda=Mock(),compiler=Mock(),
                            surface_link_velocity=surface_link_velocity)
+    module.cuda.mem_get_info.return_value=(1<<40,1<<40)  # free-memory preflight passes
     exec(compile(tree,str(path),'exec'),module.__dict__)
     return module
 

tests/unit/test_shifted_storage.py

@@ -1,7 +1,4 @@
 """CPU host contracts; mocked CUDA executes real lifecycle, not numerical kernels."""
-import ast
-import types
-from pathlib import Path
 from contextlib import nullcontext
 from dataclasses import replace
 import unittest
@@ -138,24 +135,14 @@ class ShiftedStorage(unittest.TestCase):
         module.compiler.build_module.assert_not_called()
 
     def test_uniform_block_codec_and_raw_snapshot(self):
-        path = Path(backend.r.__file__).with_name('block_stepper.py')
-        tree = ast.parse(path.read_text())
-        tree.body = [n for n in tree.body if not (
-            isinstance(n, ast.Import) and any(a.name == 'pycuda.driver' for a in n.names)
-            or isinstance(n, ast.ImportFrom) and n.module == 'cuda')]
-        module = types.ModuleType('CelerisLab.lbm._test_block_host')
-        module.__package__ = 'CelerisLab.lbm'
-        cuda, compiler = Mock(), Mock()
-        module.__dict__.update(cuda=cuda,compiler=compiler)
-        exec(compile(tree,str(path),'exec'),module.__dict__)
-        cuda.mem_get_info.return_value=(10**9,10**9)
+        module = backend.host_module()  # single blocked level: FixedPatchStepper(refinements=[])
+        cuda, compiler = module.cuda, module.compiler
         cuda.mem_alloc.side_effect=lambda size: Mock()
         gpu=Mock();gpu.get_global.return_value=(Mock(),20)
         compiler.build_module.return_value=gpu
         context=Mock(activate=nullcontext,sm_arch='sm_70')
-        layout=module.BlockLayout2D.build(16,16,4,4)
         cfg=replace(backend.CFG,nx=16,ny=16,ddf_shifting=True)
-        run=module.PeriodicBlockStepper(cfg,layout,context=context)
+        run=module.FixedPatchStepper(cfg,context=context,refinements=[],level_blocks=[(4,4)])
         storage={}
         cuda.memcpy_htod.side_effect=lambda p,a: storage.__setitem__(id(p),np.array(a,copy=True))
         cuda.memcpy_dtoh.side_effect=lambda a,p: np.copyto(a,storage[id(p)])
@@ -166,7 +153,8 @@ class ShiftedStorage(unittest.TestCase):
             self.assertTrue(raw.shifted)
             self.assertEqual(len(raw.values),1)
             self.assertTrue((raw.values[0][1]==np.float32(2**-35)).all())
-            np.testing.assert_array_equal(run.download(),physical)
+            (down,)=run.download()
+            np.testing.assert_array_equal(down,physical)
         finally: run.close()
 
     def test_passive_shifted_body_raw_remap_preserves_support(self):

tests/unit/test_single_level.py

@@ -0,0 +1,218 @@
+"""Host contract of FixedPatchStepper(refinements=[]): one blocked level (mocked driver).
+
+Per step: halo refresh -> static BFL (circles only) -> BlockStep with a null
+kind, one stream wait per step(n) call, NaN spare after initialize. An upload,
+launch, wait or download failure fails the instance and keeps its step count;
+close retains resources for retry when context activation fails.
+"""
+from contextlib import nullcontext
+from unittest.mock import MagicMock, Mock
+import unittest
+
+import numpy as np
+
+from CelerisLab.body.geometry.circle import CircleGeometry
+from CelerisLab.body.objects import SimObject
+from CelerisLab.config import LBMConfig
+from tests.unit.test_refinement_three_level_multipatch import host_module
+
+
+def circle(center=(15.5, 11.5), radius=4.3):
+    return SimObject(0, geometry=CircleGeometry(*center, radius), center=center, radius=radius)
+
+
+def backend(objects=(), failing_kernel=None, initialize=True):
+    host, log = host_module(), []
+    host.armed = False  # set True to make failing_kernel raise
+
+    def launch(name):
+        def run(*a, **k):
+            log.append((name, a))
+            if name == failing_kernel and host.armed:
+                raise RuntimeError(name + " launch failed")
+        return run
+
+    module = Mock()
+    module.get_global.return_value = (Mock(), 20)
+    module.get_function.side_effect = lambda name: Mock(side_effect=launch(name))
+    host.compiler.build_module.return_value = module
+    host.cuda.mem_alloc.side_effect = lambda size: Mock(size=size)
+    host.cuda.memcpy_htod.side_effect = lambda pointer, array: log.append(("upload", (pointer, array)))
+    run = host.FixedPatchStepper(LBMConfig(nx=32, ny=24, viscosity=.1),
+                                 context=Mock(activate=nullcontext, sm_arch="sm_70"),
+                                 refinements=[], level_blocks=[(12, 16)], level_objects=[objects])
+    run._stream.synchronize.side_effect = lambda: log.append(("sync", ()))
+    log.clear()
+    if initialize:
+        run.initialize(np.full(run._levels[0].shape, 1 / 9, np.float32))
+    return run, log, host
+
+
+def state():
+    return np.full((9, 24, 32), 1 / 9, np.float32)
+
+
+class SingleLevel(unittest.TestCase):
+    def test_spare_starts_nan_and_current_holds_state(self):
+        run, log, _ = backend()
+        level = run._levels[0]
+        uploads = {id(a[0]): a[1] for name, a in log if name == "upload"}
+        self.assertTrue(np.isnan(uploads[id(level.spare)]).all())
+        self.assertTrue(np.isfinite(level.layout.to_dense(uploads[id(level.current)])).all())
+
+    def test_launch_order_per_step(self):
+        for objects, order in (((), ["BlockHaloCopy", "BlockStep"]),
+                               ((circle(),), ["BlockHaloCopy", "BlockCurvedBoundary", "BlockStep"])):
+            with self.subTest(bodies=len(objects)):
+                run, log, _ = backend(objects)
+                log.clear()
+                run.step(2)
+                self.assertEqual([name for name, _ in log], order * 2 + ["sync"])
+                step = [a for name, a in log if name == "BlockStep"][0]
+                self.assertEqual(int(step[4]), 0)  # null kind: every core cell ACTIVE
+
+    def test_circle_solids_are_not_state(self):
+        run, _, _ = backend((circle(),))
+        solid = run._level_boundaries[0].tables["solid"]
+        self.assertTrue(solid.any())
+        np.testing.assert_array_equal(run._state[0], ~solid)
+
+    def test_initialize_upload_failure_fails_instance(self):
+        run, _, host = backend(initialize=False)
+        host.cuda.memcpy_htod.side_effect = RuntimeError("upload failed")
+        with self.assertRaisesRegex(RuntimeError, "upload failed"):
+            run.initialize(state())
+        with self.assertRaisesRegex(RuntimeError, "failed"):
+            run.step(1)
+
+    def test_launch_failure_fails_instance_and_keeps_step_count(self):
+        for kernel, objects in (("BlockHaloCopy", ()), ("BlockCurvedBoundary", (circle(),)),
+                                ("BlockStep", ())):
+            with self.subTest(kernel=kernel):
+                run, log, host = backend(objects, failing_kernel=kernel)
+                run.step(1)
+                host.armed = True
+                log.clear()
+                with self.assertRaisesRegex(RuntimeError, kernel + " launch failed"):
+                    run.step(2)
+                self.assertEqual(run.step_count, 1)
+                self.assertNotIn("sync", [name for name, _ in log])
+                with self.assertRaisesRegex(RuntimeError, "failed"):
+                    run.download()
+
+    def test_free_memory_preflight_before_stream_and_buffers(self):
+        for cfg, options in ((LBMConfig(nx=32, ny=24, viscosity=.1),
+                              dict(refinements=[], level_blocks=[(12, 16)])),
+                             (LBMConfig(nx=48, ny=48, viscosity=.02),  # every level counts
+                              dict(refinements=[[(6, 6, 26, 26)]], level_blocks=[(4, 4)] * 2))):
+            with self.subTest(depth=len(options["refinements"]) + 1):
+                host = host_module()
+                host.cuda.mem_get_info.return_value = (1024, 1 << 30)
+                with self.assertRaisesRegex(MemoryError, "exceed free device memory"):
+                    host.FixedPatchStepper(cfg, context=Mock(activate=nullcontext, sm_arch="sm_70"),
+                                           **options)
+                host.cuda.Stream.assert_not_called()
+                host.cuda.mem_alloc.assert_not_called()
+                host.compiler.build_module.assert_not_called()
+
+    def test_wait_failure_after_a_good_step_keeps_its_count(self):
+        run, _, _ = backend()
+        run.step(1)
+        run._stream.synchronize.side_effect = RuntimeError("device fault")
+        with self.assertRaises(RuntimeError):
+            run.step(2)
+        self.assertEqual((run.step_count, run._status), (1, "failed"))
+
+    def test_final_wait_failure_fails_instance_and_keeps_step_count(self):
+        run, _, _ = backend()
+        run._stream.synchronize.side_effect = RuntimeError("device error")
+        with self.assertRaisesRegex(RuntimeError, "device error"):
+            run.step(3)
+        self.assertEqual(run.step_count, 0)
+        with self.assertRaisesRegex(RuntimeError, "failed"):
+            run.step(1)
+
+    def test_step_n_waits_once_and_matches_n_single_steps(self):
+        many, log_many, _ = backend()
+        many.step(3)
+        single, log_single, _ = backend()
+        for _ in range(3):
+            single.step(1)
+        names = [name for name, _ in log_many]
+        self.assertEqual(names.count("sync"), 1)
+        self.assertEqual(names[-1], "sync")
+        self.assertEqual(names.count("BlockStep"), 3)
+
+        def roles(run, log):  # same launch order and buffer roles as three step(1) calls
+            level = run._levels[0]
+            tags = {id(level.current): "c", id(level.spare): "s"}
+            return [(n, tuple(tags.get(id(x), "-") for x in a)) for n, a in log if n != "sync"]
+        self.assertEqual(roles(many, log_many), roles(single, log_single))
+        self.assertIsNot(many._levels[0].current, many._levels[0].spare)
+        self.assertEqual((many.step_count, single.step_count), (3, 3))
+
+    def test_step_zero_launches_and_waits_for_nothing(self):
+        run, log, _ = backend()
+        log.clear()
+        run.step(0)
+        self.assertEqual((log, run.step_count), ([], 0))
+
+    def test_launch_failure_mid_batch_fails_without_waiting(self):
+        run, log, host = backend()
+        step = run._levels[0].kernel
+        calls = []
+
+        def failing(*args, **kwargs):
+            calls.append(1)
+            if len(calls) == 2:
+                raise RuntimeError("launch failure")
+            return step(*args, **kwargs)
+
+        run._levels[0].kernel = failing
+        log.clear()
+        with self.assertRaises(RuntimeError):
+            run.step(3)
+        self.assertNotIn("sync", [name for name, _ in log])
+        self.assertEqual((run.step_count, run._status), (0, "failed"))
+        with self.assertRaises(RuntimeError):
+            run.download()
+
+    def test_initialize_rejects_bad_state_before_upload(self):
+        run, log, _ = backend(initialize=False)
+        good = state()
+        for bad in (good[:, :-1], np.where(np.arange(32) == 3, np.nan, good), -good):
+            with self.assertRaises(ValueError):
+                run.initialize(bad)
+        self.assertNotIn("upload", [name for name, _ in log])
+        run.initialize(good)
+        self.assertEqual([name for name, _ in log].count("upload"), 2)
+
+    def test_download_failure_fails_instance(self):
+        run, _, host = backend()
+        host.cuda.memcpy_dtoh.side_effect = RuntimeError("read failed")
+        with self.assertRaisesRegex(RuntimeError, "read failed"):
+            run.download()
+        for operation in (run.download, run.step):
+            with self.assertRaisesRegex(RuntimeError, "failed"):
+                operation()
+
+    def test_activation_failure_retains_resources_for_retry(self):
+        run, _, _ = backend()
+        allocations, stream = run._allocations[:], run._stream
+        unavailable = MagicMock()
+        unavailable.__enter__.side_effect = RuntimeError("activation failed")
+        run.ctx.activate = Mock(side_effect=[unavailable, nullcontext()])
+        with self.assertRaisesRegex(RuntimeError, "activation failed"):
+            run.close()
+        self.assertEqual(run._status, "failed")
+        self.assertEqual(run._allocations, allocations)
+        self.assertIs(run._stream, stream)
+        run.close()
+        self.assertEqual(run._status, "closed")
+        self.assertEqual(run._levels, [])
+        for allocation in allocations:
+            allocation.free.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/validation/multigrid/test_block_stepper.py

@@ -1,4 +1,4 @@
-"""Periodic body-free block backend: CPU contract doubles plus one short GPU case.
+"""Single blocked level, FixedPatchStepper(refinements=[]): CPU contracts plus short GPU cases.
 
 Frozen GPU gates (32x24, 8 steps, positive low-Mach non-equilibrium state):
 finite cores; bit-identical 1x1, 2x2 and 4x2 splits; max |G - FP64 oracle|
@@ -11,9 +11,8 @@ Mass/momentum drift is reported, not gated.
 from contextlib import nullcontext
 from pathlib import Path
 import resource
-from threading import RLock
 import unittest
-from unittest.mock import MagicMock, Mock, patch
+from unittest.mock import Mock, patch
 
 import numpy as np
 import pycuda.driver as cuda
@@ -21,10 +20,10 @@ import pycuda.driver as cuda
 from CelerisLab.body.geometry.circle import CircleGeometry
 from CelerisLab.body.objects import SimObject
 from CelerisLab.config import LBMConfig
-from CelerisLab.cuda import compiler_v2 as compiler
-from CelerisLab.lbm import block_stepper as implementation
-from CelerisLab.lbm.blocks import NO_DONOR, BlockLayout2D
-from CelerisLab.lbm.block_stepper import PeriodicBlockStepper
+from CelerisLab.cuda import compiler
+from CelerisLab.lbm import refinement_stepper as implementation
+from CelerisLab.lbm.blocks import NO_DONOR, BlockLayout2D, static_circle_tables
+from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
 from CelerisLab.lbm.descriptors import D2Q9_EX, D2Q9_EY
 
 ROOT = Path(__file__).resolve().parents[3]
@@ -112,72 +111,78 @@ def totals(g):
     return g.sum(), (CX[:, None, None] * g).sum(), (CY[:, None, None] * g).sum()
 
 
+def single(cfg=None, *, context, block=(12, 16), objects=(), **kw):
+    """One periodic blocked level (the former PeriodicBlockStepper capability)."""
+    return FixedPatchStepper(cfg or config(), context=context, refinements=[], level_blocks=[block],
+                             level_objects=[tuple(objects)], **kw)
+
+
 class CpuContracts(unittest.TestCase):
-    """No device: driver doubles prove rejection before allocation and lifecycle."""
+    """No device: driver doubles prove rejection before allocation."""
 
     def forbidden_context(self):
         return Mock(activate=Mock(side_effect=AssertionError("CUDA reached")), sm_arch="sm_70")
 
     def test_unsupported_inputs_rejected_before_cuda(self):
-        layout = BlockLayout2D.build(NY, NX, 12, 16)
-        cases = ((config(), BlockLayout2D.build(NY, NX, 12, 16, periodic=(True, False))),
-                 (config(), BlockLayout2D.build(NY, NX * 2, 12, 16)),
-                 (config(collision="MRT"), layout),
-                 (config(), "layout"))
-        for cfg, lay in cases:
-            with self.subTest(cfg=cfg.collision, layout=getattr(lay, "periodic", lay)):
-                with self.assertRaises(ValueError):
-                    PeriodicBlockStepper(cfg, lay, context=self.forbidden_context())
+        for cfg, kw in ((None, dict(block=None)),
+                        (None, dict(coarse_boundary="channel")),
+                        (None, dict(coarse_boundary="slot_reservoir")),
+                        (config(collision="MRT"), {}),
+                        (config(les_enabled=True), {}),
+                        (None, dict(explosion="projected_linear")),
+                        (None, dict(observe_raw_exchange=True)),
+                        (None, dict(observe_raw_exchange=True, observe_window_exchange=True)),
+                        (None, dict(level_jets=[(object(),)]))):
+            with self.subTest(kw=kw), self.assertRaises(ValueError):
+                single(cfg, context=self.forbidden_context(), **kw)
         moving = circle()
         moving.state.vx = .01
         build_links = CircleGeometry.build_cut_links
         for objects in ([moving], [circle((1.5, 11.5), 2.0)], [circle(), circle()]):
             with self.assertRaises(ValueError):
-                PeriodicBlockStepper(config(), layout, context=self.forbidden_context(), objects=objects)
+                single(context=self.forbidden_context(), objects=objects)
         for center, radius in (((16.0, 12.0), -.1), ((16.0, 12.0), -.5), ((15.5, 11.5), np.nan),
                                ((np.inf, 11.5), 4.3), ((15.5, np.nan), 4.3)):
             with self.subTest(center=center, radius=radius), self.assertRaises(ValueError):
-                PeriodicBlockStepper(config(), layout, context=self.forbidden_context(),
-                                     objects=[circle(center, radius)])
-        with patch.object(implementation, "_static_circles") as prepare:
+                single(context=self.forbidden_context(), objects=[circle(center, radius)])
+        with patch.object(implementation, "static_circle_tables") as prepare:
             with self.assertRaisesRegex(AssertionError, "CUDA reached"):  # body-free path
-                PeriodicBlockStepper(config(), layout, context=self.forbidden_context(), objects=iter(()))
+                single(context=self.forbidden_context(), objects=iter(()))
             prepare.assert_not_called()
+        with self.assertRaisesRegex(AssertionError, "CUDA reached"):
+            single(context=self.forbidden_context(), objects=[circle()])
         with patch.object(CircleGeometry, "build_cut_links",
                           lambda geometry, *a, **k: build_links(geometry, *a, **k)[1:]):
             with self.assertRaisesRegex(ValueError, "cover every fluid pull"):
-                PeriodicBlockStepper(config(), layout, context=self.forbidden_context(),
-                                     objects=[circle()])
+                single(context=self.forbidden_context(), objects=[circle()])
         with patch.object(implementation, "MAX_GRID_DIM", 3):
             with self.assertRaisesRegex(ValueError, "launch limits"):
-                PeriodicBlockStepper(config(), layout, context=self.forbidden_context())
+                single(context=self.forbidden_context())
 
     def test_memory_and_param_layout_rejected_before_allocation(self):
-        layout = BlockLayout2D.build(NY, NX, 12, 16)
         context = Mock(activate=nullcontext, sm_arch="sm_70")
         with patch.object(cuda, "mem_get_info", return_value=(1024, 1 << 30)), \
                 patch.object(cuda, "Stream") as stream, patch.object(cuda, "mem_alloc") as alloc:
-            with self.assertRaises(MemoryError):
-                PeriodicBlockStepper(config(), layout, context=context)
+            with self.assertRaisesRegex(MemoryError, "exceed free device memory"):
+                single(context=context, objects=[circle()])
             stream.assert_not_called()
             alloc.assert_not_called()
         module = Mock()
         module.get_global.return_value = (Mock(), 16)
-        with patch.object(cuda, "mem_get_info", return_value=(1 << 30, 1 << 30)), \
-                patch.object(cuda, "Stream"), patch.object(compiler, "generate_config"), \
+        with patch.object(cuda, "Stream"), patch.object(cuda, "mem_get_info", return_value=(1 << 40, 1 << 40)), patch.object(compiler, "generate_config"), \
                 patch.object(compiler, "compile_kernel") as compile_kernel, \
                 patch.object(compiler, "cleanup_build_dir"), \
                 patch.object(compiler, "load_module", return_value=module), \
                 patch.object(cuda, "mem_alloc") as alloc, patch.object(cuda, "memcpy_htod") as upload:
             with self.assertRaisesRegex(RuntimeError, "layout mismatch"):
-                PeriodicBlockStepper(config(), layout, context=context)
+                single(context=context)
             self.assertEqual(compile_kernel.call_args.kwargs["source"], "blocked.cu")
             alloc.assert_not_called()
             upload.assert_not_called()
 
     def test_static_circle_tables(self):
         layout = BlockLayout2D.build(NY, NX, 12, 16)
-        tables = implementation._static_circles(layout, (circle(),))
+        tables = static_circle_tables(layout, (circle(),))
         links = CircleGeometry(*CENTER, RADIUS).build_cut_links(NX, NY, domain_flags=dense_flags())
         self.assertEqual(len(tables["cell"]), len(links))
         np.testing.assert_array_equal(tables["q"], np.array([l.q for l in links], np.float32))
@@ -185,85 +190,11 @@ class CpuContracts(unittest.TestCase):
         self.assertTrue((tables["flags"][solid_cells] & 2).all())
         self.assertEqual(int(np.count_nonzero(tables["flags"] & 2)), int(tables["solid"].sum()))
 
-    def fake_backend(self):
-        backend = PeriodicBlockStepper.__new__(PeriodicBlockStepper)
-        backend.layout = BlockLayout2D.build(NY, NX, 12, 16)
-        backend.ctx = Mock(activate=nullcontext)
-        backend._lock = RLock()
-        backend._status = "ready"
-        backend._step_count = 0
-        backend._ddf_shifting = False  # unshifted storage, as for the default config
-        backend._stream = Mock()
-        backend._module = Mock()
-        backend._copy = backend._step = Mock()
-        backend.boundary, backend._curved, backend._tables, backend._links = None, None, {}, 0
-        backend._dst, backend._src = Mock(), Mock()
-        backend._block, backend._grid, backend._copy_grid = (32, 1, 1), (1, 1, 1), (1, 1)
-        backend._allocations = [Mock(), Mock()]
-        backend._current, backend._spare = backend._allocations
-        return backend
 
-    def test_initialize_rejects_bad_state_before_upload(self):
-        backend = self.fake_backend()
-        good = initial_state()
-        bad = (good[:, :-1], np.where(np.arange(NX) == 3, np.nan, good), -good)
-        with patch.object(cuda, "memcpy_htod") as upload:
-            for g in bad:
-                with self.assertRaises(ValueError):
-                    backend.initialize(g)
-            upload.assert_not_called()
-            backend.initialize(good)
-            self.assertEqual(upload.call_count, 2)
-            self.assertTrue(np.isnan(upload.call_args_list[1].args[1]).all())
-        backend.close()
-
-    def test_failed_launch_and_download_make_backend_unusable(self):
-        for failure in ("step", "download"):
-            backend = self.fake_backend()
-            if failure == "step":
-                backend._step = Mock(side_effect=RuntimeError("launch failed"))
-                with self.assertRaisesRegex(RuntimeError, "launch failed"):
-                    backend.step()
-            else:
-                with patch.object(cuda, "memcpy_dtoh", side_effect=RuntimeError("read failed")):
-                    with self.assertRaisesRegex(RuntimeError, "read failed"):
-                        backend.download()
-            for operation in (backend.download, backend.step):
-                with self.assertRaisesRegex(RuntimeError, "failed"):
-                    operation()
-            backend.close()
-            for allocation in backend_allocations(backend):
-                allocation.free.assert_called_once()
-
-    def test_activation_failure_retains_resources_for_retry(self):
-        backend = self.fake_backend()
-        allocations, stream, module = backend._allocations[:], backend._stream, backend._module
-        unavailable = MagicMock()
-        unavailable.__enter__.side_effect = RuntimeError("activation failed")
-        backend.ctx.activate = Mock(side_effect=[unavailable, nullcontext()])
-        with self.assertRaisesRegex(RuntimeError, "activation failed"):
-            backend.close()
-        self.assertEqual(backend._status, "failed")
-        self.assertEqual(backend._allocations, allocations)
-        self.assertIs(backend._stream, stream)
-        self.assertIs(backend._module, module)
-        backend.close()
-        self.assertEqual(backend._status, "closed")
-        self.assertIsNone(backend._module)
-        self.assertIsNone(backend._copy)
-        self.assertIsNone(backend._step)
-        for allocation in allocations:
-            allocation.free.assert_called_once()
-
-
-def backend_allocations(backend):
-    return (backend._current, backend._spare)
-
-
-class PeriodicBlockGPU(unittest.TestCase):
+class SingleLevelGPU(unittest.TestCase):
     def setUp(self):
         self.assertEqual(Path(implementation.__file__).resolve(),
-                         ROOT / "src/CelerisLab/lbm/block_stepper.py")
+                         ROOT / "src/CelerisLab/lbm/refinement_stepper.py")
 
     def tearDown(self):
         rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
@@ -274,13 +205,13 @@ class PeriodicBlockGPU(unittest.TestCase):
         g0 = initial_state()
         results = []
         with CudaContext(0) as context:
-            for by, bx in SPLITS:
-                layout = BlockLayout2D.build(NY, NX, by, bx)
-                with PeriodicBlockStepper(config(), layout, context=context) as backend:
+            for split in SPLITS:
+                with single(context=context, block=split) as backend:
                     backend.initialize(g0)
                     backend.step(STEPS)
                     self.assertEqual(backend.step_count, STEPS)
-                    results.append(backend.download())
+                    (g,) = backend.download()
+                    results.append(g)
         for (by, bx), g in zip(SPLITS, results):
             self.assertTrue(np.isfinite(g).all(), (by, bx))
             np.testing.assert_array_equal(g, results[0], err_msg=f"split {(by, bx)}")
@@ -307,16 +238,17 @@ class PeriodicBlockGPU(unittest.TestCase):
         g0 = initial_state()
         results = []
         with CudaContext(0) as context:
-            for by, bx in SPLITS:
-                layout = BlockLayout2D.build(NY, NX, by, bx)
-                with PeriodicBlockStepper(config(), layout, context=context, objects=[circle()]) as backend:
+            for split in SPLITS:
+                with single(context=context, block=split, objects=[circle()]) as backend:
+                    tables = backend._level_boundaries[0].tables
                     for key, values, dtype in (("q", [l.q for l in links], np.float32),
                                                ("dir", [l.dir for l in links], np.uint8),
                                                ("fallback", [l.fallback for l in links], np.uint8)):
-                        np.testing.assert_array_equal(backend.boundary[key], np.array(values, dtype))
+                        np.testing.assert_array_equal(tables[key], np.array(values, dtype))
                     backend.initialize(g0)
                     backend.step(STEPS)
-                    results.append(backend.download())
+                    (g,) = backend.download()
+                    results.append(g)
         for (by, bx), g in zip(SPLITS, results):
             self.assertTrue(np.isfinite(g[:, ~solid]).all(), (by, bx))
             self.assertTrue(np.isnan(g[:, solid]).all(), (by, bx))

tests/validation/multigrid/test_channel_root_gpu.py

@@ -11,7 +11,7 @@ import numpy as np
 import pycuda.driver as cuda
 
 from CelerisLab.config import LBMConfig
-from CelerisLab.cuda import compiler_v2 as compiler
+from CelerisLab.cuda import compiler
 from CelerisLab.cuda.context import CudaContext
 from CelerisLab.lbm.blocks import BlockLayout2D
 from CelerisLab.lbm.descriptors import FLUID, SOLID, BC_WALL, BC_INLET, BC_OUTLET

tests/validation/multigrid/test_fine_surface_gpu.py

@@ -18,7 +18,7 @@ import CelerisLab
 from CelerisLab.body.geometry.circle import CircleGeometry
 from CelerisLab.body.objects import SimObject
 from CelerisLab.config import LBMConfig
-from CelerisLab.cuda import compiler_v2 as compiler
+from CelerisLab.cuda import compiler
 from CelerisLab.cuda.context import CudaContext
 from CelerisLab.lbm import refinement_stepper as implementation
 from CelerisLab.lbm.descriptors import FLUID, SOLID

tests/validation/multigrid/test_raw_exchange_gpu.py

@@ -224,14 +224,12 @@ class RawExchangeGPU(unittest.TestCase):
         Eight local hops are a short-time equivalence check, not convergence.
         The test adapter only enables the standalone kernel's optional inputs.
         """
-        from CelerisLab.lbm.blocks import BlockLayout2D
-        from CelerisLab.lbm.block_stepper import PeriodicBlockStepper
         fine_cfg = replace(CFG, nx=64, ny=48, viscosity=.2)
         obj = SimObject(0, geometry=CircleGeometry(31.5, 23.5, 4.3),
                         center=(31.5, 23.5), radius=4.3)
-        layout = BlockLayout2D.build(48, 64, 8, 8)
         with CudaContext(0) as ctx:
-            with PeriodicBlockStepper(fine_cfg, layout, context=ctx, objects=[obj]) as uniform, \
+            with FixedPatchStepper(fine_cfg, context=ctx, refinements=[], level_blocks=[(8, 8)],
+                                   level_objects=[(obj,)]) as uniform, \
                  FixedPatchStepper(CFG, context=ctx, **fine((16, 20)), observe_raw_exchange=True) as amr:
                 amr.initialize(*(state(l.shape) for l in amr._levels))
                 uniform.initialize(state((9, 48, 64)))
@@ -240,18 +238,21 @@ class RawExchangeGPU(unittest.TestCase):
                 links = obj.geometry.build_cut_links(64, 48)
                 np.testing.assert_array_equal([v.dir for v in links], b.tables['dir'])
                 np.testing.assert_allclose([v.q for v in links], b.tables['q'], atol=2e-7, rtol=0)
-                # Same fine-grid force arms after translation to the same center.
-                rx = uniform._upload(b.tables['rx'])
-                ry = uniform._upload(b.tables['ry'])
+                # Same fine-grid force arms after translation to the same center: the
+                # single level's static BFL gets the AMR Jet column and force arms; the raw
+                # output pointer is substituted at launch (no stepper-side accumulation).
+                root = uniform._level_boundaries[0]
                 velocity = uniform._upload(b.tables['surface_velocity'])
+                root.device.update(surface_velocity=velocity, rx=uniform._upload(b.tables['rx']),
+                                   ry=uniform._upload(b.tables['ry']))
                 raw = uniform._allocate(count*16)
-                original = uniform._curved
+                original = root.kernel
                 samples = []
                 def measured_boundary(*args, **kw):
-                    original(*args[:7], velocity, rx, ry, raw, **kw)
+                    original(*args[:10], raw, **kw)  # BlockCurvedBoundary raw-exchange output
                     uniform._stream.synchronize()
                     samples.append(download(raw, (count, 4)))
-                uniform._curved = measured_boundary
+                root.kernel = measured_boundary
                 for speed in (0., .002, -.001, 0.):
                     amr.set_jet_amplitude(0, speed)
                     cuda.memcpy_htod(velocity, b.tables['surface_velocity'])

tests/validation/multigrid/test_refinement_blocked_base.py

@@ -2,6 +2,34 @@
 
 Run CPU and GPU classes in separate processes. GPU callers must first check
 resources/device availability; this file never selects or reserves a GPU.
+
+Dense/blocked route parity (DenseBlockedRouteParity): the same configuration,
+initial state, body/Jet schedule and tilings on three storage routes, only
+level_blocks varying: D/D [None, None], D/B [None, FINE_BLOCK] (dense root,
+the mixed pattern of current examples) and B/B [(12, 8), FINE_BLOCK]. Gate:
+every public and owner array identical in FP32 bits and NaN masks at steps
+0, 1, 2, 8 and after re-initialize; D/B and B/B raw-exchange ledgers equal.
+No tolerance. A mismatch is reported with route pair/level/step and diagnosed
+(boundary semantics vs arithmetic ordering), never widened.
+
+Route capabilities (rejections checked before CUDA by test_route_capability_matrix):
+
+  combination                                   D/D       D/B       B/B
+  periodic / slot_reservoir root, fine circle   yes       yes       yes   (parity gate)
+  actuated circle, velocity Jet + amplitude     yes       yes       yes   (parity gate)
+  raw/window exchange on a body level           rejected  yes       yes
+  shifted storage                               rejected  rejected  yes
+  projected explosion                           rejected  rejected  yes
+  channel root                                  rejected  rejected  yes
+  several L1 rectangles                         rejected  rejected  yes
+  depth >= 3 with a dense level                 rejected  rejected  yes
+Slot/reservoir walls are compile-time WALL_* macros on a dense root and the
+d_block_wall constant on a blocked root: BC-encoding parity, bit-identical by
+this gate. Dense fine bodies use CurvedLinkSoA + CurvedBoundaryKernel
+(n_objects > 0), blocked ones BlockCurvedBoundary; same links and physics,
+compared through the population rows. Parked here: MRT (covered by
+test_refinement_mrt_gpu::test_dense_versus_blocked_mrt_two_level), shifted and
+depth >= 3 (no dense route to compare), several bodies (later slice).
 """
 from contextlib import ExitStack, contextmanager, nullcontext
 from dataclasses import replace
@@ -13,13 +41,14 @@ from unittest.mock import Mock, patch
 import numpy as np
 import pycuda.driver as cuda
 
-from CelerisLab.cuda import compiler_v2 as compiler
+from CelerisLab.cuda import compiler
 from CelerisLab.lbm import refinement_stepper
 from CelerisLab.lbm.blocks import BlockLayout2D
 from CelerisLab.lbm.descriptors import D2Q9_EX, D2Q9_EY
 from CelerisLab.lbm.refinement import ACTIVE, OUTER_GHOST, FixedPatch2D
 from CelerisLab.lbm.refinement_stepper import FixedPatchStepper, blocked_base_tables
 from test_refinement_blocked_fine import NX, NY, BOUNDS, config, state, circle, actuation
+from test_refinement_jets import SPEED, jet
 
 BASE_BLOCKS = ((24, 32), (12, 8), (3, 4), (1, 1))
 FINE_BLOCK = (16, 20)
@@ -73,6 +102,7 @@ def mocked_backend(mode="periodic", fail_alloc=None):
             stack.enter_context(patch.object(compiler, name))
         stack.enter_context(patch.object(compiler, "load_module", return_value=module))
         stack.enter_context(patch.object(cuda, "Stream", return_value=stream))
+        stack.enter_context(patch.object(cuda, "mem_get_info", return_value=(1 << 40, 1 << 40)))
         stack.enter_context(patch.object(cuda, "mem_alloc", side_effect=allocate))
         stack.enter_context(patch.object(cuda, "memcpy_htod",
                                         side_effect=lambda p, a: uploads.append((p, a))))
@@ -323,5 +353,128 @@ class BlockedBaseGPU(unittest.TestCase):
                             print(f"PASS {label}: full finite bits/NaN masks at 0,1,2,8 + reset", flush=True)
 
 
+ROUTES = (("D/D", [None, None]), ("D/B", [None, FINE_BLOCK]), ("B/B", [(12, 8), FINE_BLOCK]))
+
+
+def same_summary(a, b):
+    """Exact equality of nested exchange summaries (dicts, sequences, arrays, numbers)."""
+    if isinstance(a, dict):
+        return a.keys() == b.keys() and all(same_summary(a[k], b[k]) for k in a)
+    if isinstance(a, (list, tuple)):
+        return len(a) == len(b) and all(same_summary(x, y) for x, y in zip(a, b))
+    a, b = np.asarray(a), np.asarray(b)
+    if a.dtype.kind in "biufc" and b.dtype.kind in "biufc":
+        return a.shape == b.shape and np.array_equal(a, b, equal_nan=True)
+    return a.dtype == b.dtype and np.array_equal(a, b)  # labels such as the window name
+
+
+class DenseBlockedRouteCPU(unittest.TestCase):
+    def test_same_summary_compares_labels_and_every_number(self):
+        ledger = {"window": "last_base_step", "time_start": 3, "total": {"force": [1.0, -2.0],
+                  "torque": 0.5}, "aperture": {"force": np.array([np.nan, 1.0])}}
+        self.assertTrue(same_summary(ledger, replace_value(ledger, ())))
+        for path, value in ((("window",), "last_substep"), (("time_start",), 4),
+                            (("total", "force"), [1.0, -2.000001]), (("total", "torque"), 0.25),
+                            (("aperture", "force"), np.array([0.0, 1.0]))):
+            with self.subTest(path=path):
+                self.assertFalse(same_summary(ledger, replace_value(ledger, path, value)))
+                self.assertFalse(same_summary(replace_value(ledger, path, value), ledger))
+
+    def test_route_capability_matrix(self):
+        context = Mock(activate=Mock(side_effect=AssertionError("CUDA reached")))
+        periodic = config("periodic")
+        inner = (BOUNDS[0] + 2, BOUNDS[1] + 2, BOUNDS[2] - 2, BOUNDS[3] - 2)
+        second = (BOUNDS[0], BOUNDS[1], BOUNDS[0] + 2, BOUNDS[1] + 2)
+        cases = (
+            ("raw exchange on dense body level", periodic, dict(
+                level_blocks=[None, None], level_objects=[(), (circle(),)], observe_raw_exchange=True)),
+            ("shifted D/D", replace(periodic, ddf_shifting=True), dict(level_blocks=[None, None])),
+            ("shifted D/B", replace(periodic, ddf_shifting=True), dict(level_blocks=[None, FINE_BLOCK])),
+            ("projected D/D", periodic, dict(level_blocks=[None, None], explosion="projected_linear")),
+            ("projected D/B", periodic, dict(level_blocks=[None, FINE_BLOCK], explosion="projected_linear")),
+            ("channel D/B", periodic, dict(level_blocks=[None, FINE_BLOCK], coarse_boundary="channel")),
+            ("several L1 rectangles", periodic, dict(refinements=[[BOUNDS, second]], level_blocks=[None, None])),
+            ("depth 3 dense", periodic, dict(refinements=[BOUNDS, inner], level_blocks=[None, None, None])),
+        )
+        for label, cfg, kw in cases:
+            kw.setdefault("refinements", [BOUNDS])
+            with self.subTest(label), self.assertRaises(ValueError):
+                FixedPatchStepper(cfg, context=context, **kw)
+        context.activate.assert_not_called()
+
+
+def replace_value(tree, path, value=None):
+    """Deep copy of a nested dict with the item at ``path`` replaced (no change for ())."""
+    if isinstance(tree, dict):
+        return {k: (value if path == (k,) else replace_value(v, path[1:], value) if path[:1] == (k,)
+                    else replace_value(v, (), None)) for k, v in tree.items()}
+    return np.array(tree, copy=True) if isinstance(tree, np.ndarray) else tree
+
+
+class DenseBlockedRouteParity(unittest.TestCase):
+    """D/D vs D/B vs B/B, unshifted SRT, one circle or one velocity Jet; see module docstring."""
+
+    def run_routes(self, mode, objects=(), jets=(), switch=None):
+        from CelerisLab.cuda.context import CudaContext
+        cfg = config(mode)
+        p = FixedPatch2D.from_config(cfg, BOUNDS)
+        arrays = state(NY, NX, 1), state(*p.fine_shape, 2)
+        results = {}
+        with CudaContext(0) as context:
+            for name, blocks in ROUTES:
+                observe = bool(objects) and blocks[1] is not None
+                snapshots, ledgers = [], []
+                with FixedPatchStepper(cfg, context=context, coarse_boundary=mode, refinements=[BOUNDS],
+                                       level_blocks=blocks, level_objects=[(), tuple(objects)],
+                                       level_jets=[(), tuple(jets)],
+                                       observe_raw_exchange=observe) as backend:
+                    backend.initialize(*arrays)
+                    for step in (0, 1, 2, 8):
+                        if step == 8 and switch is not None:
+                            backend.set_jet_amplitude(0, switch)
+                        backend.step(step - backend.step_count)
+                        snapshots.append(backend.download() + raw(backend))
+                        if observe and step:
+                            ledgers.append(tuple(backend.read_raw_exchange(0, window=window)
+                                                 for window in ("last_base_step", "last_substep")))
+                    backend.initialize(*arrays)
+                    snapshots.append(backend.download() + raw(backend))
+                results[name] = snapshots, ledgers
+        return results
+
+    def check(self, label, results):
+        reference, _ = results["D/D"]
+        for name in ("D/B", "B/B"):
+            for step, actual, expected in zip((0, 1, 2, 8, "reset"), results[name][0], reference):
+                for index, (a, b) in enumerate(zip(actual, expected)):
+                    exact(self, a, b, f"{label} {name} vs D/D step={step} array={index}")
+        if results["D/B"][1]:
+            for step, a, b in zip((1, 2, 8), results["D/B"][1], results["B/B"][1]):
+                self.assertTrue(same_summary(a, b), f"{label} raw exchange D/B vs B/B step={step}")
+        for name, (snapshots, _) in results.items():
+            masses = [float(np.nansum(level.astype(np.float64))) for level in snapshots[3][:2]]
+            print(f"{label} {name}: step 8 mass by level {masses}", flush=True)
+        print(f"PASS {label}: D/D = D/B = B/B bits/NaN masks at 0,1,2,8 + reset", flush=True)
+
+    def test_body_free(self):
+        for mode in ("periodic", "slot_reservoir"):
+            self.check(f"{mode} body-free", self.run_routes(mode))
+
+    def test_passive_circle(self):
+        for mode in ("periodic", "slot_reservoir"):
+            self.check(f"{mode} circle", self.run_routes(mode, objects=[circle()]))
+
+    def test_actuated_circle(self):
+        for mode in ("periodic", "slot_reservoir"):
+            for speed in (SPEED, -SPEED):
+                self.check(f"{mode} actuated {speed}",
+                           self.run_routes(mode, objects=[circle(actuation(speed))]))
+
+    def test_velocity_jet_with_amplitude_switch(self):
+        for mode in ("periodic", "slot_reservoir"):
+            self.check(f"{mode} jet switch",
+                       self.run_routes(mode, objects=[circle()], jets=[jet()], switch=-SPEED))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/validation/multigrid/test_refinement_blocked_fine.py

@@ -15,7 +15,7 @@ import pycuda.driver as cuda
 from CelerisLab.body.geometry.circle import CircleGeometry
 from CelerisLab.body.objects import SimObject, SurfaceActuation
 from CelerisLab.config import LBMConfig
-from CelerisLab.cuda import compiler_v2 as compiler
+from CelerisLab.cuda import compiler
 from CelerisLab.lbm import curved_links
 from CelerisLab.lbm.blocks import NO_DONOR, BlockLayout2D, blocked_link_tables
 from CelerisLab.lbm.descriptors import D2Q9_EX, D2Q9_EY
@@ -122,7 +122,7 @@ class BlockedFineCPU(unittest.TestCase):
                 raise AssertionError("zero-byte device allocation")
             sizes.append(size)
             return Mock()
-        with patch.object(cuda, "Stream"), patch.object(compiler, "generate_config"), \
+        with patch.object(cuda, "Stream"), patch.object(cuda, "mem_get_info", return_value=(1 << 40, 1 << 40)), patch.object(compiler, "generate_config"), \
                 patch.object(compiler, "compile_kernel"), patch.object(compiler, "cleanup_build_dir"), \
                 patch.object(compiler, "load_module", return_value=module), \
                 patch.object(cuda, "mem_alloc", side_effect=alloc), patch.object(cuda, "memcpy_htod"):
@@ -172,7 +172,7 @@ class BlockedFineCPU(unittest.TestCase):
             return functions.setdefault(name, Mock(side_effect=lambda *a, **k: calls.append((name, a))))
         module = Mock(get_function=Mock(side_effect=get_function))
         module.get_global.return_value = (Mock(), 20)
-        with patch.object(cuda, "Stream"), patch.object(compiler, "generate_config"), \
+        with patch.object(cuda, "Stream"), patch.object(cuda, "mem_get_info", return_value=(1 << 40, 1 << 40)), patch.object(compiler, "generate_config"), \
                 patch.object(compiler, "compile_kernel"), patch.object(compiler, "cleanup_build_dir"), \
                 patch.object(compiler, "load_module", return_value=module), \
                 patch.object(cuda, "mem_alloc", side_effect=lambda size: Mock()), \

tests/validation/multigrid/test_refinement_followup.py

@@ -166,7 +166,7 @@ class FaultGuardsCPU(unittest.TestCase):
         # A pair tau inconsistent with its level config is rejected twice over, both before
         # any device allocation or upload: first by the hierarchy chain check, and, if that
         # is bypassed, by the per-level omega/tau check in _build_level.
-        with patch.object(cuda, "Stream"), bad, patch.object(cuda, "mem_alloc") as alloc, \
+        with patch.object(cuda, "Stream"), patch.object(cuda, "mem_get_info", return_value=(1 << 40, 1 << 40)), bad, patch.object(cuda, "mem_alloc") as alloc, \
                 patch.object(cuda, "memcpy_htod") as upload:
             with self.assertRaisesRegex(RuntimeError, "config chain is inconsistent"):
                 PeriodicPatchStepper(cfg, context=context, refinements=[p.bounds])
@@ -177,7 +177,7 @@ class FaultGuardsCPU(unittest.TestCase):
             upload.assert_not_called()
         module = Mock()
         module.get_global.return_value = (Mock(), 16)
-        with patch.object(cuda, "Stream"), patch.object(compiler, "generate_config"), \
+        with patch.object(cuda, "Stream"), patch.object(cuda, "mem_get_info", return_value=(1 << 40, 1 << 40)), patch.object(compiler, "generate_config"), \
                 patch.object(compiler, "compile_kernel"), patch.object(compiler, "cleanup_build_dir"), \
                 patch.object(compiler, "load_module", return_value=module), \
                 patch.object(cuda, "memcpy_htod") as upload:

tests/validation/multigrid/test_refinement_gpu.py

@@ -20,7 +20,7 @@ import pycuda.driver as cuda
 
 import CelerisLab
 from CelerisLab.config import LBMConfig
-from CelerisLab.cuda import compiler_v2 as compiler
+from CelerisLab.cuda import compiler
 from CelerisLab.cuda.context import CudaContext
 from CelerisLab.lbm.refinement_stepper import FixedPatchStepper, PeriodicPatchStepper
 

tests/validation/multigrid/test_refinement_jets.py

@@ -16,7 +16,7 @@ from unittest.mock import Mock, patch
 import numpy as np
 import pycuda.driver as cuda
 
-from CelerisLab.cuda import compiler_v2 as compiler
+from CelerisLab.cuda import compiler
 from CelerisLab.lbm.descriptors import SOLID
 from CelerisLab.lbm.refinement import ACTIVE, FixedPatch2D, prepare_fine_boundary
 from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
@@ -36,7 +36,7 @@ def mocked(objects, jets, fine_block):
     module = Mock(get_function=Mock(return_value=Mock()))
     module.get_global.return_value = (Mock(), 20)
     htod = Mock()
-    with patch.object(cuda, "Stream"), patch.object(compiler, "generate_config"), \
+    with patch.object(cuda, "Stream"), patch.object(cuda, "mem_get_info", return_value=(1 << 40, 1 << 40)), patch.object(compiler, "generate_config"), \
             patch.object(compiler, "compile_kernel"), patch.object(compiler, "cleanup_build_dir"), \
             patch.object(compiler, "load_module", return_value=module), \
             patch.object(cuda, "mem_alloc", side_effect=lambda size: Mock()), \

tests/validation/multigrid/test_refinement_three_level.py

@@ -52,11 +52,11 @@ def child_jet(speed=.002):
 
 
 def mocked_child():
-    from CelerisLab.cuda import compiler_v2 as compiler
+    from CelerisLab.cuda import compiler
     context = Mock(activate=nullcontext, sm_arch="sm_70")
     module = Mock(get_function=Mock(side_effect=lambda name: Mock(name=name)))
     module.get_global.return_value = (Mock(), 20)
-    with patch.object(cuda, "Stream"), patch.object(compiler, "generate_config"), \
+    with patch.object(cuda, "Stream"), patch.object(cuda, "mem_get_info", return_value=(1 << 40, 1 << 40)), patch.object(compiler, "generate_config"), \
             patch.object(compiler, "compile_kernel"), patch.object(compiler, "cleanup_build_dir"), \
             patch.object(compiler, "load_module", return_value=module), \
             patch.object(cuda, "mem_alloc", side_effect=lambda size: Mock()), \

tests/validation/run_refined_dfg.py

@@ -281,16 +281,21 @@ def host_geometry(source, h, collision="SRT", jet_scenario="none", placement="wi
     return cfg, pair, body, host
 
 
-def verify_source(source, collision="SRT", observation_mode="last_base_step", jet_scenario="none", observe_cv=False):
+def source_manifest(collision="SRT", observation_mode="last_base_step"):
+    """The frozen historical manifest for this collision/observation mode (no checkout read)."""
     if collision not in ("SRT", "MRT"):
         raise ValueError("Matched collision must be SRT or MRT")
     if observation_mode not in ("last_base_step", "window"):
         raise ValueError("Unknown refined observation mode")
-    if observe_cv and jet_scenario == "none":
-        raise ValueError("--observe-cv requires an explicit public Jet scenario")
     name = ("source_manifest_window.json" if observation_mode == "window" else
             "source_manifest.json" if collision == "SRT" else "source_manifest_mrt.json")
-    expected = json.loads((HERE / "data/refined_dfg" / name).read_text())
+    return json.loads((HERE / "data/refined_dfg" / name).read_text())
+
+
+def verify_source(source, collision="SRT", observation_mode="last_base_step", jet_scenario="none", observe_cv=False):
+    expected = source_manifest(collision, observation_mode)
+    if observe_cv and jet_scenario == "none":
+        raise ValueError("--observe-cv requires an explicit public Jet scenario")
     if jet_scenario != "none":
         if collision != "MRT" or observation_mode != "window":
             raise ValueError("Public Jet scenarios require MRT and --exchange-window")
@@ -298,7 +303,8 @@ def verify_source(source, collision="SRT", observation_mode="last_base_step", je
         sys.path.insert(0, str(HERE))
         from source_provenance import checkout_provenance
 
-        paths = set(expected["sha256"])
+        # Harness files from the manifest; production files as this checkout lists them.
+        paths = {name for name in expected["sha256"] if not name.startswith("src/")}
         if observe_cv:
             paths.add("tests/validation/run_jet_circle_cv.py")
         paths.update(subprocess.check_output(
@@ -310,7 +316,8 @@ def verify_source(source, collision="SRT", observation_mode="last_base_step", je
                     for name in sorted(paths)},
         )
     for name, digest in expected["sha256"].items():
-        if hashlib.sha256((source / name).read_bytes()).hexdigest() != digest:
+        path = source / name
+        if not path.is_file() or hashlib.sha256(path.read_bytes()).hexdigest() != digest:
             raise RuntimeError("Source mismatch: " + name)
     return expected
 

tests/validation/taylor_green/test_projected_pairs.py

@@ -1,6 +1,6 @@
 """Host geometry, receipt-corruption tests and mocked scheduling; no CUDA/flow."""
 
-from contextlib import nullcontext, redirect_stdout
+from contextlib import contextmanager, nullcontext, redirect_stdout
 from copy import deepcopy
 from dataclasses import asdict, replace
 import hashlib
@@ -24,6 +24,44 @@ from oracle import LIMITS, TIMES, U, rectangles
 REPO = Path(__file__).resolve().parents[3]
 OLD_REVISION = "519473a61c7b319183aa54e55b9f097499dde0bf"
 MOCK_REVISION = "f" * 40
+
+
+@contextmanager
+def historical_package():
+    """Import CelerisLab from NUMERICAL_REVISION, the only tree run_case accepts.
+
+    The harness and its pinned build-capture helper are tooling for that revision
+    (cuda/compiler_v2.py); the current package is restored afterwards.
+    """
+    import tarfile
+    archive = subprocess.run(["git", "-C", str(REPO), "archive", "--format=tar",
+                              NUMERICAL_REVISION, "src/CelerisLab"],
+                             capture_output=True, check=True).stdout
+    before = set(sys.modules)
+    saved = {k: v for k, v in sys.modules.items() if k == "CelerisLab" or k.startswith("CelerisLab.")}
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp).resolve()
+        with tarfile.open(fileobj=io.BytesIO(archive)) as tar:
+            tar.extractall(root)
+        for name in saved:
+            del sys.modules[name]
+        sys.path.insert(0, str(root / "src"))
+        try:
+            yield root
+        finally:
+            sys.path.remove(str(root / "src"))
+            for name in set(sys.modules) - before | set(saved):  # nothing keeps the old tree
+                sys.modules.pop(name, None)
+            sys.modules.update(saved)
+
+
+def historical_runtime_source(repo):
+    """runtime_source as run_case records it for an accepted a135252 checkout."""
+    package = repo / "src/CelerisLab"
+    return dict(checkout=str(repo), module=str(package / "__init__.py"),
+                compiler=str(package / "cuda/compiler_v2.py"), kernel_root=str(package / "lbm/kernels"))
+
+
 RUNTIME = dict(device=dict(name="host mock", pci_bus_id="mock", capability=[7, 0],
                            driver=12040, runtime=[12, 4, 0]), nvcc_path="/mock/nvcc",
                nvcc_version="mock nvcc", python="mock Python", numpy=np.__version__, pycuda="mock")
@@ -60,7 +98,7 @@ def records(early=False):
             reference_harness=run_case.REFERENCE_HARNESS,
             actual_build_sidecars=[build_metadata(n * 2**level) for level in range(depth)],
             runtime_identity=deepcopy(RUNTIME), cuda_visible_devices="0",
-            runtime_source=run_case.runtime_source(REPO), completed=True,
+            runtime_source=historical_runtime_source(REPO), completed=True,
             workload=run_case.workload([k == ACTIVE for k in h.kinds], 16*n),
             rows=[dict(step=s, tstar=s*U/n, **{k: 0.0 for k in LIMITS}) for s in main_steps],
             early_pressure=dict(enabled=early, rows=[
@@ -125,6 +163,10 @@ class ProjectedPairsTests(unittest.TestCase):
                                  frozen[name])
 
     def test_actual_build_capture_rejects_tampered_ptx_and_sidecar(self):
+        with historical_package():
+            self._check_build_capture()
+
+    def _check_build_capture(self):
         from CelerisLab.cuda import compiler_v2
         from tests.validation.multigrid.check_projected_pairs import capture_builds
 
@@ -338,6 +380,7 @@ class ProjectedPairsTests(unittest.TestCase):
             with patch.object(sys, "argv", pinned + ["--dry-run"]), \
                  patch.dict(sys.modules, cuda_modules), patch.dict("os.environ", CUDA_VISIBLE_DEVICES=""), \
                  patch.object(run_case, "verify_source", return_value=source()) as verify, \
+                 patch.object(run_case, "runtime_source", return_value=historical_runtime_source(REPO)), \
                  patch.object(run_case, "runtime_identity", side_effect=AssertionError("CUDA reached")), \
                  redirect_stdout(stdout):
                 self.assertEqual(run_case.main(), 0)
@@ -346,10 +389,12 @@ class ProjectedPairsTests(unittest.TestCase):
             self.assertFalse(output.exists())
 
     def test_runtime_rejects_installed_kernel_tree(self):
-        from CelerisLab.cuda import compiler_v2
-        with patch.object(compiler_v2, "_KERNEL_DIR", "/other/CelerisLab/lbm/kernels"):
-            with self.assertRaisesRegex(RuntimeError, "CUDA source/include root"):
-                run_case.runtime_source(REPO)
+        with historical_package() as repo:
+            from CelerisLab.cuda import compiler_v2
+            self.assertEqual(run_case.runtime_source(repo)["checkout"], str(repo))
+            with patch.object(compiler_v2, "_KERNEL_DIR", "/other/CelerisLab/lbm/kernels"):
+                with self.assertRaisesRegex(RuntimeError, "CUDA source/include root"):
+                    run_case.runtime_source(repo)
 
     def test_existing_output_is_never_overwritten(self):
         with tempfile.TemporaryDirectory() as tmp:
@@ -425,6 +470,7 @@ class ProjectedPairsTests(unittest.TestCase):
                 with patch.dict(sys.modules, modules), patch.object(sys, "argv", argv), \
                      patch.dict("os.environ", CUDA_VISIBLE_DEVICES="mock"), \
                      patch.object(run_case, "verify_source", return_value=source()) as verify, \
+                     patch.object(run_case, "runtime_source", return_value=historical_runtime_source(REPO)), \
                      patch.object(run_case, "runtime_identity", return_value=deepcopy(RUNTIME)), \
                      patch.object(run_case, "build_capture", return_value=nullcontext()), \
                      patch.object(run_case, "metrics", side_effect=metric), redirect_stdout(io.StringIO()):

tests/validation/time_window_dfg.py

@@ -38,7 +38,7 @@ VARIANTS = ("A", "W", "W1")
 # links). Engineering tolerance: 1e-9 of the interval's summed absolute per-step totals.
 INTERVAL_RTOL = 1e-9
 CANDIDATE_FILES = ("src/CelerisLab/lbm/refinement_stepper.py", "src/CelerisLab/lbm/raw_exchange.py",
-                   "src/CelerisLab/lbm/refinement_tables.py", "src/CelerisLab/lbm/kernels/blocks.cu")
+                   "src/CelerisLab/lbm/refinement_tables.py", "src/CelerisLab/lbm/kernels/blocked.cu")
 
 
 def components(obs):