Contract: name the in-EDGE half-cell projection; repair DFG runner current-checkout Jet path

- No NODE<->EDGE input conversion still leaves the root-edge -> fine-local
  half-cell projection (prepare_fine_boundary, fine_hits_to_edge_frame); say so.
- run_refined_dfg.py (jet_scenario != none, current checkout only): pass
  level=1 to FixedPatchStepper.set_jet_amplitude on reversal; use
  blocked_coarse_tables there. jet_scenario=none keeps the pinned source's
  blocked_base_tables; manifests and results unchanged.

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

docs/compatibility.md

@@ -3,6 +3,13 @@
 Changes that can make a previously accepted setup or call behave differently.
 Each entry names the commit, the affected entry points and the observable result.
 
+## Redundant periodic stepper alias removed
+
+`PeriodicPatchStepper` is no longer exported from `lbm.refinement_stepper`.
+Use `FixedPatchStepper` with the same arguments, including explicit
+`level_blocks`. The old name was an alias for that exact class; periodic
+boundaries remain the default. Existing imports of the old name must be updated.
+
 ## 87c3a31: cut-link ownership is validated for every body
 
 `validate_cut_link_ownership` (`body/preprocess/flag_overlay.py`) now runs for
@@ -33,3 +40,18 @@ or negative amplitude (the aperture test is geometric). Bodies with neither
 `SurfaceActuation` nor aperture links still raise
 `ValueError("Body has no surface actuation")`. The value is unchanged: the latest
 step's wall/aperture exchange, not total nozzle force.
+
+## 7afb5ad: `Simulation` fails closed after a Jet amplitude device-write failure
+
+`Simulation.set_jet_amplitude` previously only documented that a device error
+leaves the per-link velocity column undefined; the instance stayed usable.
+Now any error after input validation (internal stream creation or the upload)
+marks the Simulation failed. Every later device-touching public method,
+including the `stream` property (guard added in the follow-up repair commit on
+`claude/jet-update-consistency`), raises
+`RuntimeError("Simulation failed (...); close() and recreate it")`. The host
+amplitude keeps its previous value. `close()` still works and stays retryable;
+cached host readers and host staging calls are not blocked. Invalid input
+(`ValueError`/`IndexError`) is unchanged: it raises before any change and the
+instance stays usable. `FixedPatchStepper.set_jet_amplitude` already failed the
+instance (status `failed`) and is unchanged.

docs/development/projected_pair_explosion.md

@@ -4,7 +4,7 @@
 G-to-G explosion reconstruction. The omitted/default `"constant"` selector retains
 the same kernel, launch arguments and device allocations. Projected mode adds two
 compact tables per adjacent pair; recursion, collision, BFL, restriction, buffer
-ownership and synchronized publication are unchanged. `remap_child` preserves it.
+ownership and synchronized publication are unchanged. `remap_finest` preserves it.
 
 Runtime eligibility is periodic-root, fully blocked D2Q9/SRT/FP32, body/Jet-free,
 without LES; the fixed-patch backend has no forcing interface. Shifted and unshifted

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/development/src_coherence.md

@@ -0,0 +1,77 @@
+# src coherence stage (from main 914906e)
+
+Owner direction 2026-10-28: unified style over historical compatibility, inside `src`.
+No numerical change is intended; kernel text was moved verbatim (PTX/bit parity is
+pending a GPU run).
+
+## Conventions (defined once, in `lbm.hierarchy.Hierarchy`)
+
+- Level `l = 0 .. depth-1` (0 = root). Pair `hierarchy.pairs[l-1]` builds level `l`.
+- Array shapes are `(ny, nx)`; rectangles, origins and centers are `(x, y)`.
+- Coarse/fine name the two members (parent/child) of ANY pair, not absolute L0/L1:
+  `fine_shape`, `fine_origin`, `prepare_fine_boundary`, `blocked_fine_tables`,
+  `blocked_coarse_tables`.
+- Refined levels (l >= 1) of FixedPatchStepper place bodies/Jets in the root EDGE frame
+  (root cell `i` = `[i, i+1)`). Uniform `Simulation` objects AND the static circles of a
+  single-level (depth 1) FixedPatchStepper use the integer-NODE frame. Nothing converts
+  between them. This concerns input placement only: within the EDGE frame, refined
+  levels still project root-edge coordinates to fine-local cell indices with the
+  half-cell shift (`prepare_fine_boundary`: `(v - o)/dx - .5`; hits back via `fine_hits_to_edge_frame`
+  `origin + dx*(hit + .5)`), using `edge_frame`/`fine_origin` below.
+- `Hierarchy.edge_frame(l)` -> `((ox, oy), h)`: cell `(i, j)` of level `l`'s allocated
+  array (ghost band and holes included) covers `[o + h*i, o + h*(i+1))`. A scalar
+  FixedPatch2D window carries the half-coarse-cell ghost shift `bounds - HALO/2` in
+  `fine_origin`; a set's level-wide grid has origin 0. Level 1 boundaries are built from
+  the root config (the wall-aware fields live there); deeper ones from the parent frame.
+- `FixedPatch2D.fine_level_offset` (was `fine_level_origin`) is an array INDEX offset
+  `(y, x)`, not an `(x, y)` origin.
+
+## FixedPatchStepper migration map
+
+| removed / changed | replacement |
+|---|---|
+| `patch`, `child_patch` | `hierarchy.pairs[0]`, `hierarchy.pairs[1]` |
+| `base_layout`, `fine_layout`, `child_layout` (stepper properties) | `layouts[0]`, `layouts[1]`, `layouts[2]` (`FixedPatchSet2D.fine_layout(block)` is unchanged) |
+| `fine_boundary`, `child_boundary` | `boundaries[1]`, `boundaries[2]` (host geometry; `None` = no bodies/Jets) |
+| `_jets` | `jets[1]` |
+| `_boundaries[l]` (dict from L1, level keys) | `_device_boundaries[l]` (list from L0, `None` at L0 unless single-level static circles) |
+| `_level_boundaries` (private, b4eacf7) | `_device_boundaries` |
+| `_fine_tables`, `_actuated`, `_jet_cache` | `_device_boundaries[1].tables` / `.actuated` / `.jet_cache` |
+| `_remap_cfg` (private) | `_root_cfg` |
+| `refinement_tables.blocked_base_tables` (also via `refinement_stepper`) | `refinement_tables.blocked_coarse_tables` (import from `refinement_tables`; the stepper no longer re-exports) |
+| `_fine_leaf`, `_child_leaf` | `leaf[1]`, `leaf[2]` (copies) |
+| `_middle_kind` | `hierarchy.kinds[1]` |
+| `_boundary_hosts` (private) | `_boundaries` (private list behind `boundaries`) |
+| `remap_child(rects, child_objects=...)` | `remap_finest(rects, objects=...)` |
+| `set_jet_amplitude(i, a)` (implicit level 1) | `set_jet_amplitude(i, a, level=l)` (required) |
+| `read_raw_exchange(b)` / `read_exchange_window(b)` (implicit level 1) | `read_raw_exchange(b, level=l)` / `read_exchange_window(b, level=l)` (required) |
+| `_level_name(l)` "base/fine/child" messages | `"level l"` |
+| `refinement.MIN_PATCH_GAP` | deleted (no gap policy exists) |
+| `params.cuh HaloPlan`, `collide_trt_no_force` | deleted (zero references) |
+
+Kept: `step_count`, constructor keywords (`refinements`, `level_blocks`, `level_objects`,
+`level_jets`, `coarse_boundary`, `explosion`), `Simulation.set_jet_amplitude(i, a)` (one level),
+`RemapTopology.from_config(..., child_objects=, child_jets=)` (internal remap helper; "child" =
+the replaced level relative to its parent).
+
+## Responsibilities
+
+- Construction: `_check_levels` (inputs), `_kernel_configs` (per-level kernel configs),
+  `Hierarchy.edge_frame` (geometry frames), `_build_level` / `_build_pair` / `_upload_boundary`
+  (device). Observation reads share `_observed_boundary` + `_summarize`.
+- Blocked kernels: `operators/block_collision.cuh block_collide` is the one collision extension
+  point of `BlockStep` and `BlockChannelStep`. MRT shares one path; SRT keeps its two explicit
+  arithmetic variants (level: unclamped, no-force, shifted direct-momentum feq; channel: uniform
+  OneStep arithmetic with omega clamp), selected by `BLOCK_ROOT_BC`.
+- `explosion="projected_linear"` stays an optional capability; `"constant"` stays default.
+
+## Not done / open
+
+- Two per-level config derivations stay distinct on purpose: `Hierarchy.configs`
+  (`_middle_config`, wall fields reset on every refined level, geometry validation) vs.
+  `FixedPatchStepper._kernel_configs` (reset only on level 1 of a slot_reservoir root;
+  generates each level's config.h). Folding them needs a config.h/PTX diff gate.
+- No single-`Simulation` facade over FixedPatchStepper: the two own different state models
+  (uniform node frame, body sync, forcing, checkpoint vs. immutable blocked hierarchy, edge
+  frame, remap-as-new-instance). A facade would only forward calls; no concrete need found.
+- Tests, examples and validation callers are migrated separately (pinned runners stay frozen).

docs/fixed_three_level_multipatch.md

@@ -61,7 +61,7 @@ Stationary whole-circle L2 bodies and Jets (`level_objects[2]`, `level_jets[2]`)
 coordinates and a single L2 boundary object. Body IDs are physical object-list
 positions, not region IDs or caller `obj_id`. Complete boundary stencils must
 avoid the transfer footprint. Change a child Jet with
-`set_jet_amplitude(index, value, level=2)`; L1 bodies/Jets remain unsupported.
+`set_jet_amplitude(index, value, level=2)` (`level` is required); L1 bodies/Jets remain unsupported.
 
 ## Verification
 

docs/lifecycle.md

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

docs/supported_settings.md

@@ -10,12 +10,13 @@ measured; they are not general bounds.
 
 | Setting | Supported | Guard |
 |---|---|---|
-| Dimension / lattice | 2D D2Q9 only (3D refinement is future work) | `lbm/refinement.py` `FixedPatch2D.from_config`; `#error` guards in `lbm/kernels/blocks.cu` and `lbm/kernels/refinement.cu` |
+| Dimension / lattice | 2D D2Q9 only (3D refinement is future work) | `lbm/refinement.py` `FixedPatch2D.from_config`; `#error` guards in `lbm/kernels/blocked.cu` |
 | Precision / streaming | FP32 storage, double buffer | same guards |
 | 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) |
+| Shifted (ddf_shifting) storage | supported (every level is blocked); one representation on all levels | `lbm/refinement_stepper.py` `FixedPatchStepper._check_storage` |
 | 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` |
+| Refinement storage | blocked on every level: `level_blocks` is required, one tile `(by, bx)` per level dividing that level; a whole-level tile (`(ny, nx)` at the root, the fine window shape for one rectangle) always divides. Dense and mixed dense/blocked storage were removed (parity of D/D, D/B and B/B recorded at 1573942 and 2469af7) | `lbm/refinement_stepper.py` `_check_levels`, `FixedPatchStepper.__init__` |
 
 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`.
@@ -34,7 +35,7 @@ Tests: `tests/unit/test_refinement_mrt.py`, `tests/validation/multigrid/test_ref
 | `slot_reservoir` | on-site velocity bottom slot, reservoir top, velocity 0, 2 <= slot width < nx, patch rows 4..ny-4 | `lbm/refinement_stepper.py` `FixedPatchStepper.__init__`; `lbm/refinement.py` `prepare_fine_boundary` |
 | `channel` | all levels blocked; Zou-He inlet, pressure-NEQ outlet, stationary bounce-back walls; transfers away from the closure | `lbm/channel.py` `validate_channel_config`, `validate_channel_transfers` |
 | Explosion | `constant` (default); `projected_linear` only periodic, all-blocked SRT, no bodies/Jets, no LES (experimental) | `lbm/refinement_stepper.py` `FixedPatchStepper.__init__` |
-| L2 runtime remap (`remap_child`) | periodic three-level L2 sets only, no L0/L1 bodies, identical pinned boundary | `lbm/refinement_stepper.py` `FixedPatchStepper.remap_child` |
+| L2 runtime remap (`remap_finest`) | periodic three-level L2 sets only, no L0/L1 bodies, identical pinned boundary | `lbm/refinement_stepper.py` `FixedPatchStepper.remap_finest` |
 
 Tests: `tests/unit/test_refinement_three_level_multipatch.py`, `tests/validation/multigrid/test_refinement_multipatch.py`,
 `test_refinement_three_level.py`, `test_refinement_three_level_multipatch_gpu.py`, `test_refinement_blocked_base.py`,
@@ -83,6 +84,6 @@ Tests: `tests/unit/test_jets.py`, `test_jets_review.py`, `test_shifted_jets.py`,
 
 ## Not supported in this release
 
-3D, dynamic/automatic AMR (tagging, time-varying hierarchies beyond `remap_child`),
+3D, dynamic/automatic AMR (tagging, time-varying hierarchies beyond `remap_finest`),
 moving bodies on refined levels, pressure Jets, Jet checkpoints, LES or TRT on refined
 levels, partial (non-dividing) tiles.

docs/synchronized_leaf_remap.md

@@ -33,7 +33,7 @@ origin is retained. Physical lower edges are `origin + spacing*(x,y)` on L0,
 `origin + spacing*(parent.fine_origin + (x,y)/4)` on L2.
 
 Factory options `coarse_boundary` and `objects` must declare `"periodic"` and
-an empty collection; L2 bodies use `child_objects` as described below.
+an empty collection; L2 bodies use `objects` as described below.
 LBMConfig's channel settings do not select this operator's boundary mode.
 Geometry metadata is factory-created;
 manual mutation/forging of metadata is unsupported. Caller certifies the
@@ -77,7 +77,7 @@ PYTHONPATH=src python -m unittest discover -s tests/unit -p 'test_synchronized_r
 
 ## Explicit runtime continuation
 
-`replacement = current.remap_child(rectangles)` returns a new ready
+`replacement = current.remap_finest(rectangles)` returns a new ready
 `FixedPatchStepper` at the **same absolute step_count**. It supports only
 periodic three-level runs with no L0/L1 bodies and an existing L2
 `FixedPatchSet2D`; stationary L2 bodies/Jets follow the support contract below.
@@ -99,7 +99,7 @@ resources; an actual old-device synchronization/read failure marks the old
 stepper failed, consistent with its existing failure semantics.
 
 ```python
-replacement = current.remap_child([(20, 16, 36, 40)])
+replacement = current.remap_finest([(20, 16, 36, 40)])
 # Explicitly choose the returned continuation only after success.
 current.close()
 current = replacement
@@ -119,7 +119,7 @@ Per-population conservation is not an advancement target under collision.
 
 ### Bounded pinned-body extension
 
-`replacement = current.remap_child(rectangles, child_objects=original_objects)`
+`replacement = current.remap_finest(rectangles, objects=original_objects)`
 also accepts stationary Circles inside the L2 union. Each body's **physical
 boundary, complete BFL support and current per-link velocity must remain
 identical**. Rectangles are not pinned: they may be split, merged or resized,

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

examples/fixed_periodic_patch.py

@@ -16,7 +16,7 @@ from CelerisLab.config import load_lbm_config
 from CelerisLab.cuda.context import CudaContext
 from CelerisLab.lbm.equilibrium import compute_feq_d2q9, compute_macro_from_ddf
 from CelerisLab.lbm.refinement import ACTIVE
-from CelerisLab.lbm.refinement_stepper import PeriodicPatchStepper
+from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
 
 
 def shear_equilibrium(shape, origin, dx, period):
@@ -48,7 +48,8 @@ def main():
                   nx=16, ny=16, viscosity=.1, velocity=1e-3, rho=1.0,
                   lattice_model="D2Q9", collision="SRT", streaming="double_buffer",
                   store_precision="FP32", ddf_shifting=False, les_enabled=False)
-    with CudaContext(0) as context, PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)]) as run:
+    with CudaContext(0) as context, FixedPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)],
+                                                           level_blocks=[(16, 16), (24, 24)]) as run:
         p = run.patch
         coarse = shear_equilibrium(p.coarse_shape, (0, 0), 1.0, cfg.ny)
         fine = shear_equilibrium(p.fine_shape, p.fine_origin, .5, cfg.ny)

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.
 
@@ -7,8 +7,7 @@ build_module runs, in a private directory it removes afterwards:
   1. build_dir = generate_config(lbm_cfg, n_objects) — isolated source snapshot
   2. compile_kernel(arch, source=entry, build_dir=build_dir) — nvcc -ptx
   3. load_module(ptx_path)              — PyCUDA module from PTX
-Entries: uniform.cu (uniform grid), blocked.cu (blocked level), refinement.cu
-(dense refinement level).
+Entries: uniform.cu (uniform grid), blocked.cu (every blocked level, single or refined).
 """
 
 from __future__ import annotations
@@ -281,9 +280,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/hierarchy.py

@@ -21,7 +21,24 @@ from .refinement import (ACTIVE, OUTER_GHOST, FixedPatch2D, FixedPatchSet2D,
 @dataclass(frozen=True, eq=False)
 class Hierarchy:
     """configs[l]: validated LBMConfig of level l; pairs[l]: tables of (l, l+1);
-    kinds[l]: composed per-cell kind (child coverage OUTER on every non-finest level)."""
+    kinds[l]: composed per-cell kind (child coverage OUTER on every non-finest level).
+
+    Index conventions (the single place they are defined):
+      * level l = 0 (root) .. depth-1; pair index p = l-1 builds level l from level p;
+      * array shapes are (ny, nx); rectangles, origins and centers are (x, y);
+      * a pair's two members are its COARSE (parent) and FINE (child) level, for every
+        pair, not absolute L0/L1 (fine_shape, fine_origin, blocked_coarse_tables, ...);
+      * refined levels (l >= 1) of FixedPatchStepper place bodies and Jets in the root
+        EDGE frame: root cell i covers [i, i+1), center i+.5. Uniform Simulation objects
+        and the static circles of a single-level (depth 1) FixedPatchStepper use the
+        integer-NODE frame (CircleGeometry.build_flag_mask at integer (x, y)). No
+        NODE<->EDGE input conversion is applied anywhere; inside the EDGE frame, refined
+        boundaries still project root-edge coordinates to fine-local cell indices with
+        the half-cell shift ((v - o)/h - .5, see prepare_fine_boundary);
+      * cell (i, j) of level l's allocated array (ghost band and holes included) covers
+        [o + h*i, o + h*(i+1)) with (o, h) = edge_frame(l); level 1 is built from the root
+        config directly, deeper levels from their parent frame (see edge_frame).
+    """
     configs: tuple
     pairs: tuple
     kinds: tuple
@@ -30,6 +47,18 @@ class Hierarchy:
     def depth(self):
         return len(self.kinds)
 
+    def edge_frame(self, level):
+        """(origin (x, y), spacing) of level ``level``'s array in the root edge frame.
+
+        Each pair's fine_origin is in parent cells: zero for a set (level-wide grid),
+        the shifted window origin for a scalar FixedPatch2D.
+        """
+        origin, spacing = (0.0, 0.0), 1.0
+        for pair in self.pairs[:level]:
+            origin = tuple(o + spacing * f for o, f in zip(origin, pair.fine_origin))
+            spacing /= 2
+        return origin, spacing
+
 
 def is_rectangle(value):
     value = tuple(value)
@@ -86,14 +115,14 @@ def _check_chain(coarse, fine, pair):
 def leaf_partition(h):
     """Base-cell fraction published as ACTIVE over all levels (exact: powers of two).
 
-    Window (scalar) levels are offset by their fine_level_origin first.
+    Window (scalar) levels are offset by their fine_level_offset first.
     """
     base = np.zeros(h.kinds[0].shape)
     oy = ox = 0
     for level, kind in enumerate(h.kinds):
         if level:
             pair = h.pairs[level - 1]
-            py, px = (pair.fine_level_origin if isinstance(pair, FixedPatch2D) else (0, 0))
+            py, px = (pair.fine_level_offset if isinstance(pair, FixedPatch2D) else (0, 0))
             oy, ox = 2 * oy + py, 2 * ox + px
         y, x = np.nonzero(kind == ACTIVE)
         np.add.at(base, ((y + oy) >> level, (x + ox) >> level), 0.25 ** level)

src/CelerisLab/lbm/kernels/blocked.cu

@@ -34,9 +34,7 @@
 #if BLOCK_ROOT_BC == 1
 #include "boundary/block_wall.cuh"
 #endif
-#if BLOCK_ROOT_BC != 2
 #include "operators/block_collision.cuh"
-#endif
 
 // Kernels, in definition order: transfers, boundary staging, then the one full
 // step selected for this module (BlockStep, or BlockChannelStep for a channel root).

src/CelerisLab/lbm/kernels/core/params.cuh

@@ -44,16 +44,4 @@ struct RigidBodyControl2D {
     int   control_mode;           // 0=force, 1=velocity, 2=displacement
 };
 
-// ============================================================================
-// Halo plan (multi-GPU / AMR 预留)
-// ============================================================================
-struct HaloPlan {
-    int       face_id;
-    int       peer_gpu;
-    int       level;
-    size_t    count;
-    int*      send_idx;
-    int*      recv_idx;
-};
-
 #endif // CELERIS_CORE_PARAMS_CUH

src/CelerisLab/lbm/kernels/operators/block_collision.cuh

@@ -1,4 +1,10 @@
-// Collision used by periodic/fine and slot/reservoir BlockStep.
+// The one collision extension point of the blocked steps (BlockStep, BlockChannelStep).
+// Compile-time selection: MRT shares one path; SRT keeps two explicit arithmetic
+// variants, unchanged from their original kernels:
+//   level  (BLOCK_ROOT_BC 0/1): unclamped omega, no forcing term; shifted storage
+//          builds feq from the retained drho and direct momentum (numerical variant);
+//   channel (BLOCK_ROOT_BC 2): matches uniform OneStep SRT arithmetic (omega clamp,
+//          zero Guo forcing term, compute_feq).
 #ifndef CELERIS_OPERATORS_BLOCK_COLLISION_CUH
 #define CELERIS_OPERATORS_BLOCK_COLLISION_CUH
 
@@ -11,7 +17,13 @@
 
 __device__ __forceinline__ void block_collide(float* f) {
 #if COLLISION_MODEL == 2
-    collide_mrt_unforced(f);  // SRT path below unchanged.
+    collide_mrt_unforced(f);
+#elif BLOCK_ROOT_BC == 2
+    float rho, ux, uy, feq[NQ], forcing[NQ] = {};
+    compute_rho_u(f, rho, ux, uy);
+    compute_feq(rho, ux, uy, feq);
+    const float omega = fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega));
+    collide_srt(f, feq, forcing, omega);
 #else
     float rho, ux, uy, feq[NQ];
     compute_rho_u(f, rho, ux, uy);

src/CelerisLab/lbm/kernels/operators/collision_trt.cuh

@@ -57,25 +57,4 @@ __device__ __forceinline__ void collide_trt(float* __restrict__ f,
     }
 }
 
-__device__ __forceinline__ void collide_trt_no_force(float* __restrict__ f,
-                                                     const float* __restrict__ feq,
-                                                     float omega)
-{
-    const float wp = omega;
-    const float wm = compute_omega_minus(wp);
-
-    f[0] = f[0] - wp * (f[0] - feq[0]);
-
-    #pragma unroll
-    for (int i = 1; i < NQ; i += 2) {
-        const int ib = i + 1;
-        const float fi = f[i], fb = f[ib];
-        const float feqi = feq[i], feqb = feq[ib];
-        const float delta_s = 0.5f * ((fi - feqi) + (fb - feqb));
-        const float delta_a = 0.5f * ((fi - feqi) - (fb - feqb));
-        f[i]  = fi  - wp * delta_s - wm * delta_a;
-        f[ib] = fb  - wp * delta_s + wm * delta_a;
-    }
-}
-
 #endif // CELERIS_OPERATORS_COLLISION_TRT_CUH

src/CelerisLab/lbm/kernels/refinement.cu

@@ -1,40 +0,0 @@
-// Fixed, unforced internal 2:1 patch. Separate compilation entry;
-// default OneStep/channel kernels are deliberately not included.
-#include <stdint.h>
-#include <math.h>
-#include "config.h"
-#include "core/precision.cuh"
-#include "core/descriptors.cuh"
-#include "core/layout.cuh"
-#include "core/params.cuh"
-#include "core/flags.cuh"
-#include "core/obs.cuh"
-#include "core/cell_kind.cuh"
-#include "operators/macro.cuh"
-#include "operators/collision_srt.cuh"
-#if COLLISION_MODEL == 2
-#include "operators/collision_mrt.cuh"
-#endif
-#include "streaming/pull_double_buffer.cuh"
-#include "boundary/planar_velocity.cuh"
-#include "boundary/curved_boundary.cuh"
-
-extern "C" {
-#include "step/curved_double.cu"
-}
-
-#if DIM != 2 || NQ != 9 || (COLLISION_MODEL != 0 && COLLISION_MODEL != 2) || STORE_PRECISION != 0 || USE_DDF_SHIFTING || USE_LES || STREAMING_MODEL != 0
-#error "Refinement requires unshifted FP32 D2Q9 SRT or MRT double buffers without LES"
-#endif
-
-// Enabled only in the validation build; no diagnostic arrays in normal builds.
-#ifdef CELERIS_REFINEMENT_AUDIT
-extern "C" {
-__device__ unsigned int collision_counts[TOTAL_CELLS];
-__device__ unsigned int explosion_counts[TOTAL_CELLS];
-__device__ unsigned int incoming_counts[NQ * TOTAL_CELLS];
-}
-#endif
-
-// Kernels: patch collision helper, PatchExplode, PatchFineStep, PatchCoarseStep.
-#include "step/patch_step.cu"

src/CelerisLab/lbm/kernels/step/block_channel.cu

@@ -48,18 +48,7 @@ extern "C" __global__ void BlockChannelStep(
         } else {
             apply_wall_bb_d2q9_rows(gy, d_block_channel.ny - 2, f, in + base, k);
         }
-        if (!wall && !outlet) {
-#if COLLISION_MODEL == 2
-            collide_mrt_unforced(f);
-#else
-            // Match uniform SRT arithmetic, including its shifted equilibrium path.
-            float rho, ux, uy, feq[NQ], forcing[NQ] = {};
-            compute_rho_u(f, rho, ux, uy);
-            compute_feq(rho, ux, uy, feq);
-            const float omega = fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega));
-            collide_srt(f, feq, forcing, omega);
-#endif
-        }
+        if (!wall && !outlet) block_collide(f);
     }
     stream_pull_store(k, f, out + base);
 }

src/CelerisLab/lbm/kernels/step/block_transfer.cu

@@ -16,8 +16,8 @@ extern "C" __global__ void BlockHaloCopy(
 }
 
 // Fixed-patch explosion into blocked fine storage: target = owner block cell
-// id of a patch-ghost cell; donors are parent block cell ids. Dense parent
-// storage is the one-window case with coarse_cells = full domain size.
+// id of a patch-ghost cell; donors are parent block cell ids and coarse_cells
+// is the parent's block window size.
 extern "C" __global__ void BlockExplode(
     const fpxx* coarse, fpxx* fine, const uint64_t* target, const uint64_t* donor,
     unsigned int count, unsigned long coarse_cells)
@@ -66,7 +66,7 @@ extern "C" __global__ void BlockExplodeProjected(
 }
 
 // Targets are receiver-local SOURCE slots for the immediately following pull.
-// Keep child summation order identical to PatchCoarseStep; no collision here.
+// Child summation order ((c0 + c1) + c2) + c3 is part of the numerics; no collision here.
 extern "C" __global__ void BlockCoalesce(
     const fpxx* fine, fpxx* coarse, const uint64_t* target, const uint64_t* children,
     unsigned int count)

src/CelerisLab/lbm/kernels/step/patch_step.cu

@@ -1,95 +0,0 @@
-// CelerisLab/lbm/kernels/step/patch_step.cu
-// Dense fixed-patch collision, explosion and the fine/coarse steps.
-// Part of the refinement.cu compilation entry; relies on its includes, guards
-// and the optional CELERIS_REFINEMENT_AUDIT counters declared there.
-
-__device__ __forceinline__ void patch_collide(float* f, unsigned long k) {
-#if COLLISION_MODEL == 2
-    collide_mrt_unforced(f);
-#else
-    float rho, ux, uy, feq[NQ];
-    compute_rho_u(f, rho, ux, uy);
-    compute_feq(rho, ux, uy, feq);
-    collide_srt_no_force(f, feq, d_params.omega);
-#endif
-#ifdef CELERIS_REFINEMENT_AUDIT
-    ++collision_counts[k];
-#endif
-}
-
-extern "C" __global__ void PatchExplode(
-    const float* coarse, float* fine, const uint32_t* targets,
-    const uint32_t* donors, unsigned int count, unsigned long coarse_cells)
-{
-    unsigned int row = blockIdx.x * blockDim.x + threadIdx.x;
-    if (row >= count) return;
-    for (unsigned int q = 0; q < NQ; ++q)
-        fine[index_f(targets[row], q)] = coarse[q * coarse_cells + donors[row]];
-#ifdef CELERIS_REFINEMENT_AUDIT
-    ++explosion_counts[targets[row]];
-#endif
-}
-
-extern "C" __global__ void PatchFineStep(const float* in, float* out, const uint16_t* flags) {
-    unsigned int x, y; unsigned long k;
-    index_from_thread(x, y, k);
-    if (x >= NX || y >= NY) return;
-    float f[NQ];
-    // Outer halo is invalid after EACH hop, including when reusing the buffer
-    // that held explosion data. Inner halo only streams; it never collides.
-    if (x < 2 || y < 2 || x >= NX - 2 || y >= NY - 2
-            || (flags != nullptr && is_solid(flags[k]))) {
-        fill_invalid(f);
-    } else {
-        unsigned long j[NQ];
-        compute_neighbors(x, y, j);  // These interior targets never wrap.
-        stream_pull_load(k, f, in, j);
-        if (x >= 4 && y >= 4 && x < NX - 4 && y < NY - 4)
-            patch_collide(f, k);
-    }
-    stream_pull_store(k, f, out);
-}
-
-extern "C" __global__ void PatchCoarseStep(
-    const float* in, float* out, const float* fine,
-    const uint64_t* targets, const uint64_t* children, unsigned int count,
-    unsigned int x0, unsigned int y0, unsigned int x1, unsigned int y1,
-    unsigned int wall_aware)
-{
-    unsigned int x, y; unsigned long k;
-    index_from_thread(x, y, k);
-    if (x >= NX || y >= NY) return;
-    // Both buffers were poisoned at initialization; covered cells stay invalid.
-    if (x >= x0 && x < x1 && y >= y0 && y < y1) return;
-    if (wall_aware && (y == 0 || y == NY - 1)) return;
-    unsigned long j[NQ];
-    compute_neighbors(x, y, j);
-    float f[NQ];
-    stream_pull_load(k, f, in, j);
-    if (x >= x0 - 1 && x <= x1 && y >= y0 - 1 && y <= y1) {
-        for (unsigned int q = 1; q < NQ; ++q) {
-            uint64_t slot = index_f(k, q);
-            unsigned int lo = 0, hi = count;
-            while (lo < hi) {
-                unsigned int mid = lo + (hi - lo) / 2;
-                if (targets[mid] < slot) lo = mid + 1;
-                else hi = mid;
-            }
-            if (lo < count && targets[lo] == slot) {
-                const uint64_t* s = children + 4ul * lo;
-                f[q] = 0.25f * (fine[s[0]] + fine[s[1]] + fine[s[2]] + fine[s[3]]);
-#ifdef CELERIS_REFINEMENT_AUDIT
-                ++incoming_counts[slot];
-#endif
-            }
-        }
-    }
-    // Includes the eight grazing-corner links with UNCOVERED coarse donors.
-    // All replacement arrivals must be in place before computing rho/u.
-    if (wall_aware) {
-        if (y == 1) reconstruct_bottom_velocity(f, 0.0f, bottom_wall_normal_velocity(x));
-        else if (y == NY - 2) reconstruct_top_reservoir(f, RHO);
-    }
-    patch_collide(f, k);
-    stream_pull_store(k, f, out);
-}

src/CelerisLab/lbm/refinement.py

@@ -115,8 +115,9 @@ class FixedPatch2D:
         return self.bounds[0] - HALO / 2, self.bounds[1] - HALO / 2
 
     @property
-    def fine_level_origin(self) -> tuple[int, int]:
-        """(y, x) of this window's first cell on the level-wide 2:1 fine grid."""
+    def fine_level_offset(self) -> tuple[int, int]:
+        """Array index (y, x) of this window's first cell on the level-wide 2:1 fine grid
+        (an index offset, not an (x, y) origin)."""
         return 2 * self.bounds[1] - HALO, 2 * self.bounds[0] - HALO
 
     @property
@@ -151,7 +152,6 @@ class FixedPatch2D:
             raise ValueError("levels are not synchronized: fine_steps must equal 2*coarse_steps")
 
 
-MIN_PATCH_GAP = 4  # Legacy exported constant; union transfers impose no gap policy.
 
 
 @dataclass(frozen=True, eq=False)

src/CelerisLab/lbm/refinement_stepper.py

@@ -1,10 +1,11 @@
 # CelerisLab/lbm/refinement_stepper.py
-"""Opt-in fixed patch backend with optional stationary fine-interior circles.
+"""Blocked fixed 2:1 refinement hierarchy (any depth) with optional stationary circles/Jets.
 
 The caller owns the CudaContext and must close this backend before that context.
 Only synchronized postcollision_g leaf populations are public. No checkpoint,
-forcing or runtime parameter mutation is offered. Bounded manual periodic
-L2 set replacement is available at completed base intervals via remap_child.
+forcing or runtime parameter mutation besides Jet amplitudes is offered. A bounded
+manual periodic level-2 set replacement is available at completed base intervals
+via remap_finest. Level/pair/shape/frame conventions: lbm.hierarchy.Hierarchy.
 """
 from dataclasses import dataclass, replace
 from numbers import Integral
@@ -14,15 +15,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 .hierarchy import build_hierarchy
 from .refinement import ACTIVE, HALO, FixedPatch2D, FixedPatchSet2D, prepare_fine_boundary
-from .refinement_tables import (blocked_base_tables, blocked_fine_tables,  # noqa: F401 (re-export)
+from .refinement_tables import (blocked_coarse_tables, blocked_fine_tables,
                                 blocked_kind, projected_explosion_tables, _prepare_blocked_boundary,
+                                _BlockedBoundary,
                                 BLOCKED_BOUNDARY_DEVICE_COLUMNS)
 from .descriptors import SOLID
 from .raw_exchange import summarize_exchange
@@ -61,30 +63,50 @@ class _Level:
     boundary: object = None
 
 
-def _level_name(level):
-    return ("base", "fine", "child")[level] if level < 3 else f"level {level}"
+def _kernel_configs(cfg, pairs, wall_aware):
+    """Per-level kernel LBMConfig (window shape, viscosity doubling per level).
 
+    Deliberately distinct from Hierarchy.configs (refinement._middle_config), which
+    resets the wall fields on EVERY refined level for geometry validation. Here only
+    level 1 of a slot_reservoir root resets them (internal bounce-back/velocity ghost
+    setup); other levels inherit their parent's fields. These configs generate each
+    level's compiled config.h, so folding the two derivations is a numerical-parity
+    change (config.h/PTX diff gate), not a style one.
+    """
+    configs = [replace(cfg)]
+    for index, pair in enumerate(pairs):
+        parent = configs[-1]
+        level = replace(parent, nx=pair.fine_shape[1], ny=pair.fine_shape[0],
+                        viscosity=2 * parent.viscosity)
+        if index == 0 and wall_aware:
+            level = replace(level, y_wall_bc="bounce_back", top_wall_bc="velocity",
+                            wall_slot_width=0, wall_normal_velocity=0.0, velocity=0.0)
+        configs.append(level)
+    return configs
+
+
+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 (not isinstance(block, tuple) or len(block) != 2
+                or any(isinstance(v, (bool, np.bool_)) or not isinstance(v, Integral) or v <= 0
+                       for v in block)):
+            raise ValueError(f"level_blocks[{level}] must be a tile (by, bx) of positive integers; "
+                             "a whole-level tile ((ny, nx) at the root, the fine window shape "
+                             "for one rectangle) always divides its level")
+    if depth == 1:  # one blocked level: the root carries the static circles
+        if jets[0]:
+            raise ValueError("a single level 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)):
-        raise ValueError("a blocked level requires blocked refined levels below it")
-    if not is_rectangle(refinements[0]) and (blocks[0] is None or blocks[1] is None):
-        raise ValueError("several L1 patches require level_blocks[0] and level_blocks[1]")
     if depth > 2:
-        if any(block is None for block in blocks):
-            raise ValueError("three or more levels must all be blocked")
         if any(objects[level] or jets[level] for level in range(1, depth - 1)):
             raise ValueError("bodies/Jets are supported only on the finest level of a nested hierarchy")
 
@@ -96,12 +118,17 @@ class FixedPatchStepper:
     (x0, y0, x1, y1) (FixedPatch2D, a fine window with its own origin) or a
     sequence of rectangles (FixedPatchSet2D, a dependency-validated union on one
     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).
+    see lbm.hierarchy. ``level_blocks`` (required), ``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) in the integer-NODE frame, like Simulation
+    objects (see Hierarchy for frames).
 
     ``coarse_boundary`` is ``periodic`` (default) or ``slot_reservoir``:
     a bottom on-site slot and top fixed-density reservoir from ``cfg``.
-    ``channel`` requires all-blocked levels and matches uniform SRT Zou-He
+    ``channel`` matches uniform SRT Zou-He
     inlet, pressure-NEQ outlet and stationary bounce-back y walls. Boundary
     ghosts evolve internally but public downloads exclude them; initialization
     must supply their finite positive-density state. Transfers must avoid the
@@ -109,7 +136,7 @@ class FixedPatchStepper:
     Periodic mode does not apply the wall/slot boundary fields in ``cfg``.
     Outer walls remain coarse; fine halo donors must be ordinary fluid.
 
-    Bodies are fixed circles in root (L0) physical coordinates; velocities are
+    Bodies of refined levels are fixed circles in root (L0) physical coordinates; velocities are
     in root lattice units (dx=dt=1). Independent local copies are projected at
     fine cell centers; their complete stencil must be active and disjoint from
     the actual two-hop transfer footprint. These use the edge/cell-center frame
@@ -124,14 +151,13 @@ class FixedPatchStepper:
     stream. On a device/launch failure the instance becomes unusable: close and
     recreate it, rather than pretending a partially advanced buffer rolled back.
 
-    A ``level_blocks`` entry (by, bx) stores that level in blocked.cu storage:
-    the ENTIRE level shape, including the four-cell patch ghost band, is
-    partitioned by a non-periodic BlockLayout2D with one-cell same-level halos.
-    Patch kinds stay per cell (OUTER invalid, INNER stream only, ACTIVE
-    collide); halos are gathered before each hop. Public arrays keep the dense
-    shapes; only the allocation differs. A blocked level requires blocked
-    refined levels below it; several L1 rectangles and depth >= 3 require all
-    levels blocked. Blocked bodies are stationary circles, optionally
+    Every level is stored in blocked.cu storage with its ``level_blocks`` tile
+    (by, bx): the ENTIRE level shape, including the four-cell patch ghost band,
+    is partitioned by a BlockLayout2D with one-cell same-level halos; the tile
+    must divide the level (a whole-level tile always does). Patch kinds stay
+    per cell (OUTER invalid, INNER stream only, ACTIVE collide); halos are
+    gathered before each hop. Public arrays are dense (9, ny, nx); only the
+    allocation is blocked. Bodies are stationary circles, optionally
     surface-actuated: prepare_fine_boundary decides legality on the dense grid,
     then cut links are staged receiver-local per hop: halo -> BFL -> step, same
     buffer. A blocked parent gathers halos before receiver-local coalescence
@@ -140,8 +166,7 @@ class FixedPatchStepper:
     ``observe_raw_exchange=True`` records blocked physical cut-link exchange
     without changing populations. read_raw_exchange reports the last complete
     base interval or last substep in base-lattice units per unit depth. This
-    is boundary momentum reaction, not a validated nozzle/stress load. Dense
-    body levels are not supported by this diagnostic.
+    is boundary momentum reaction, not a validated nozzle/stress load.
     ``observe_window_exchange=True`` (requires observe_raw_exchange) adds a second
     per-link FP64 ledger accumulated by the same kernel between
     start_exchange_window() and read_exchange_window(); disabled by default.
@@ -158,11 +183,18 @@ class FixedPatchStepper:
     lattice velocity); every whole circle and its BFL stencil must fit strictly
     away from that level's transfer footprint.
 
-    ``explosion="projected_linear"`` selects experimental transverse linear
-    reconstruction on each adjacent pair. Default "constant" is unchanged.
-    Projected mode requires periodic root, all-blocked SRT levels without
-    bodies/Jets or LES; this backend supplies no forcing. Geometry is generic,
-    but union/multilevel GPU and matched-physics qualification remain pending.
+    ``explosion`` selects the parent->child reconstruction of every pair:
+    ``"constant"`` (default) or ``"projected_linear"``, an optional transverse
+    linear reconstruction (Schornbaum-Ruede/Chen style) that is GPU-confirmed
+    more accurate than constant only on periodic two-level single-rectangle
+    Taylor-Green cases; union/multilevel GPU and matched-physics qualification
+    remain pending. It requires a periodic root and SRT levels without
+    bodies/Jets or LES; this backend supplies no forcing.
+
+    Per-level read-only state: ``hierarchy.pairs`` (pair l-1 builds level l),
+    ``layouts``, ``boundaries`` (host geometry), ``jets`` and ``leaf`` masks,
+    each indexed by level. ``set_jet_amplitude``, ``read_raw_exchange`` and
+    ``read_exchange_window`` take an explicit ``level=``.
 
     One recursive hop advances every level: a child explodes from its parent's
     current hop source, takes two hops, then the parent gathers halos, stages
@@ -172,30 +204,37 @@ class FixedPatchStepper:
                  level_blocks=None, level_objects=None, level_jets=None,
                  observe_raw_exchange=False, observe_window_exchange=False, explosion="constant"):
         if HALO != 4:
-            raise RuntimeError("refinement.cu requires a four-cell fine halo")
+            raise RuntimeError("the refinement transfers require a four-cell fine halo")
         if coarse_boundary not in ("periodic", "slot_reservoir", "channel"):
             raise ValueError("coarse_boundary must be periodic, slot_reservoir or channel")
         depth = len(refinements) + 1
-        level_blocks = [None] * depth if level_blocks is None else list(level_blocks)
+        if level_blocks is None:
+            raise ValueError("level_blocks is required: one tile (by, bx) per level; a whole-level "
+                             "tile ((ny, nx) at the root, the fine window shape for one rectangle) "
+                             "always divides its level")
+        level_blocks = list(level_blocks)
         level_objects = [()] * depth if level_objects is None else list(level_objects)
         level_jets = [()] * depth if level_jets is None else list(level_jets)
         level_objects = [tuple(o) for o in level_objects]
         level_jets = [tuple(j) for j in level_jets]
-        level_blocks = [None if b is None else tuple(b) for b in level_blocks]
+        level_blocks = [tuple(b) if isinstance(b, (list, tuple)) else b for b in level_blocks]
         _check_levels(refinements, level_blocks, level_objects, level_jets)
         depth = len(refinements) + 1
         if explosion not in ("constant", "projected_linear"):
             raise ValueError("explosion must be constant or projected_linear")
         if explosion == "projected_linear" and (
-                coarse_boundary != "periodic" or any(b is None for b in level_blocks)
+                coarse_boundary != "periodic"
                 or any(level_objects) or any(level_jets)
                 or cfg.collision != "SRT" or cfg.les_enabled):
-            raise ValueError("projected explosion requires periodic all-blocked SRT levels "
+            raise ValueError("projected explosion requires periodic SRT levels "
                              "without bodies/Jets or LES")
         self._explosion = explosion
         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)
@@ -212,56 +251,37 @@ class FixedPatchStepper:
                 raise ValueError("slot_reservoir requires y0 >= 4 and y1 <= ny-4")
         self._layouts = []
         for level, block in enumerate(level_blocks):
-            layout = None
-            if block is not None:
-                if level == 0:
-                    layout = BlockLayout2D.build(cfg.ny, cfg.nx, *block,
-                                                 periodic=(True, not self._wall_aware))
-                elif isinstance(pairs[level - 1], FixedPatchSet2D):
-                    layout = pairs[level - 1].fine_layout(block)
-                else:
-                    layout = BlockLayout2D.build(*pairs[level - 1].fine_shape, *block,
-                                                 periodic=(False, False))
-                if layout.nblocks > MAX_GRID_DIM or layout.local_shape[0] > MAX_GRID_DIM:
-                    raise ValueError(f"{_level_name(level)} blocks exceed launch limits")
+            if level == 0:
+                layout = BlockLayout2D.build(cfg.ny, cfg.nx, *block,
+                                             periodic=(True, not self._wall_aware))
+            elif isinstance(pairs[level - 1], FixedPatchSet2D):
+                layout = pairs[level - 1].fine_layout(block)
+            else:
+                layout = BlockLayout2D.build(*pairs[level - 1].fine_shape, *block,
+                                             periodic=(False, False))
+            if layout.nblocks > MAX_GRID_DIM or layout.local_shape[0] > MAX_GRID_DIM:
+                raise ValueError(f"level {level} blocks exceed launch limits")
             self._layouts.append(layout)
         if self._channel:
-            base = blocked_base_tables(pairs[0], self._layouts[0], np.argsort(pairs[0].coalesce_coarse))
+            base = blocked_coarse_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]
-        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],
-                                            viscosity=2 * parent.viscosity))
-        # Per-level boundary geometry in the root edge frame (level l spacing 2**-l).
-        self._boundary_hosts = [None]
-        origin, spacing = (0.0, 0.0), 1.0
-        spacings = [spacing]
+        self._level_cfgs = _kernel_configs(cfg, pairs, self._wall_aware)
+        # Host boundary geometry per level in the root edge frame (Hierarchy.edge_frame).
+        self._boundaries = [None]
         for level in range(1, depth):
             pair = pairs[level - 1]
-            if level == 1:
+            if level == 1:  # the root config carries the wall-aware boundary fields
                 host = prepare_fine_boundary(cfg, pair, level_objects[1],
                                              wall_aware=self._wall_aware, jets=level_jets[1],
                                              _shifted_blocked=self._ddf_shifting)
             else:
+                origin, spacing = self.hierarchy.edge_frame(level - 1)
                 host = prepare_fine_boundary(self._level_cfgs[level - 1], pair, level_objects[level],
                                              jets=level_jets[level], parent_origin=origin,
                                              parent_spacing=spacing, _shifted_blocked=self._ddf_shifting)
-            self._boundary_hosts.append(host)
-            origin = tuple(o + spacing * f for o, f in zip(origin, pair.fine_origin))
-            spacing /= 2
-            spacings.append(spacing)
+            self._boundaries.append(host)
         if not isinstance(observe_raw_exchange, bool):
             raise ValueError("observe_raw_exchange must be bool")
-        if observe_raw_exchange and any(b is not None and layout is None
-                                      for b, layout in zip(self._boundary_hosts, self._layouts)):
-            raise ValueError("raw exchange currently requires blocked body levels")
         self._observe_raw_exchange = observe_raw_exchange
         if not isinstance(observe_window_exchange, bool):
             raise ValueError("observe_window_exchange must be bool")
@@ -271,25 +291,31 @@ class FixedPatchStepper:
         self._window_valid = False
         self._window_start = None
         self._exchange_valid = False
-        self._exchange_spacing = tuple(spacings)
+        self._exchange_spacing = tuple(self.hierarchy.edge_frame(l)[1] for l in range(depth))
         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] + [
-            _prepare_blocked_boundary(self._boundary_hosts[level], self._layouts[level],
+        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._device_boundaries = [root] + [
+            _prepare_blocked_boundary(self._boundaries[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]
+            host = self._boundaries[level]
             if host is not None:
                 self._leaf[level] &= (host.flags.reshape(pairs[level - 1].fine_shape) & SOLID) == 0
         self._state = [mask.copy() for mask in self._leaf]
         if self._channel:
             self._leaf[0] &= channel_masks((cfg.ny, cfg.nx))[0]
-        # Private constructor inputs for the bounded manual L2 continuation.
-        self._remap_cfg = replace(cfg)
+        # Private constructor inputs for remap_finest.
+        self._root_cfg = replace(cfg)
         self._level_blocks = level_blocks
         self.ctx = context
         self._lock = RLock()
@@ -298,47 +324,20 @@ class FixedPatchStepper:
         self._allocations = []
         self._levels, self._pairs = [], []
         self._stream = None
-        self._fine_flags = self._fine_curved = self._curved = None
-        self._link_tables = {}
         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))
                 for index, pair in enumerate(pairs):
                     self._pairs.append(self._build_pair(index, pair))
                 for level in self._levels:
-                    b = level.boundary
-                    if b is not None:
-                        b.flags = self._upload(b.tables["flags"])
-                        b.device = {k: self._upload(v) for k, v in b.tables.items()
-                                    if k in BLOCKED_BOUNDARY_DEVICE_COLUMNS}
-                        if self._observe_raw_exchange:
-                            b.device.update(rx=self._upload(b.tables["rx"]),
-                                            ry=self._upload(b.tables["ry"]))
-                            count = len(b.tables["cell"])
-                            b.last_raw = self._allocate(count * 4 * 4)
-                            b.interval_integral = self._allocate(count * 4 * 8)
-                            if self._observe_window:
-                                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:
-                    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:
-                    from .curved_links import CurvedLinkSoA
-                    boundary = self._boundary_hosts[1]
-                    self._fine_flags = self._upload(boundary.flags)
-                    # SoA owns its allocations; never also register them in _allocations.
-                    self._fine_curved = CurvedLinkSoA()
-                    self._fine_curved.assign_host(*boundary.curved_host)
-                    self._fine_curved.assign_surface(boundary.objects, fine_cfg, boundary.flags,
-                                                     jets=self._level_jets[1],
-                                                     hits=boundary.curved_hits)
-                    self._fine_curved.upload()
-                    self._curved = self._levels[1].module.get_function("CurvedBoundaryKernel")
+                    if level.boundary is not None:
+                        self._upload_boundary(level)
             except BaseException:
                 try:
                     self.close()
@@ -346,89 +345,100 @@ class FixedPatchStepper:
                     pass  # Preserve the original construction failure.
                 raise
 
+    def _upload_boundary(self, level):
+        """Device columns, BFL kernel and (when observed) the exchange ledgers of one level."""
+        b = level.boundary
+        b.flags = self._upload(b.tables["flags"])
+        b.device = {k: self._upload(v) for k, v in b.tables.items()
+                    if k in BLOCKED_BOUNDARY_DEVICE_COLUMNS}
+        if self._observe_raw_exchange:
+            b.device.update(rx=self._upload(b.tables["rx"]), ry=self._upload(b.tables["ry"]))
+            count = len(b.tables["cell"])
+            b.last_raw = self._allocate(count * 4 * 4)
+            b.interval_integral = self._allocate(count * 4 * 8)
+            if self._observe_window:
+                b.window_integral = self._allocate(count * 4 * 8)
+            b.accumulate = level.module.get_function("BlockAccumulateExchange")
+        b.kernel = level.module.get_function("BlockCurvedBoundary")
+
+    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):
+            total += 2 * 4 * int(np.prod(layout.blocked_shape))
+            total += layout.halo_dst.nbytes + layout.halo_src.nbytes
+            b = self._device_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
-        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)
-            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")
-        host = self._boundary_hosts[index]
-        n_objects = len(host.objects) if name == "PatchFineStep" and host is not None else 0
-        block_root_bc = 0
-        if index == 0 and layout is not None:
-            block_root_bc = 2 if self._channel else 1 if self._wall_aware else 0
-        module = compiler.build_module(config, source, arch=self.ctx.sm_arch, n_objects=n_objects,
+        tau = (pairs[0].tau_coarse if pairs else 3 * config.viscosity + .5) if index == 0 \
+            else pairs[index - 1].tau_fine
+        config = level_config(config, layout, tau)
+        name = "BlockChannelStep" if index == 0 and self._channel else "BlockStep"
+        block_root_bc = (2 if self._channel else 1 if self._wall_aware else 0) if index == 0 else 0
+        module = compiler.build_module(config, "blocked.cu", arch=self.ctx.sm_arch,
                                        block_root_bc=block_root_bc)
-        compiler.set_params(module, config.omega, n_objects=n_objects)
+        compiler.set_params(module, config.omega)
         shape = (9, *self.hierarchy.kinds[index].shape)
-        size = int(np.prod(shape if layout is None else layout.blocked_shape)) * 4
-        grid = ((config.nx + self._block[0] - 1) // self._block[0], config.ny,
-                1 if layout is None else layout.nblocks)
+        size = int(np.prod(layout.blocked_shape)) * 4
+        grid = ((config.nx + self._block[0] - 1) // self._block[0], config.ny, layout.nblocks)
         level = _Level(shape, module, self._allocate(size), self._allocate(size),
                        module.get_function(name), grid, layout)
-        level.boundary = self._level_boundaries[index]
-        if layout is not None:
-            level.halo_copy = module.get_function("BlockHaloCopy")
-            if len(layout.halo_dst):
-                level.halo_dst = self._upload(layout.halo_dst)
-                level.halo_src = self._upload(layout.halo_src)
-            if index == 0 and self._channel:
-                cfg = self._remap_cfg
-                ptr, size = module.get_global("d_block_channel")
-                if size != struct.calcsize("=IIf"):
-                    raise RuntimeError("Block channel host/device layout mismatch")
-                cuda.memcpy_htod(ptr, struct.pack("=IIf", cfg.nx, cfg.ny, cfg.velocity))
-            if index == 0 and self._wall_aware:
-                cfg = self._remap_cfg
-                ptr, size = module.get_global("d_block_wall")
-                if size != struct.calcsize("=IIIf"):
-                    raise RuntimeError("Block wall host/device layout mismatch")
-                cuda.memcpy_htod(ptr, struct.pack("=IIIf", cfg.nx, cfg.ny,
-                                                cfg.wall_slot_width, cfg.wall_normal_velocity))
+        level.boundary = self._device_boundaries[index]
+        level.halo_copy = module.get_function("BlockHaloCopy")
+        if len(layout.halo_dst):
+            level.halo_dst = self._upload(layout.halo_dst)
+            level.halo_src = self._upload(layout.halo_src)
+        if index == 0 and self._channel:
+            cfg = self._root_cfg
+            ptr, size = module.get_global("d_block_channel")
+            if size != struct.calcsize("=IIf"):
+                raise RuntimeError("Block channel host/device layout mismatch")
+            cuda.memcpy_htod(ptr, struct.pack("=IIf", cfg.nx, cfg.ny, cfg.velocity))
+        if index == 0 and self._wall_aware:
+            cfg = self._root_cfg
+            ptr, size = module.get_global("d_block_wall")
+            if size != struct.calcsize("=IIIf"):
+                raise RuntimeError("Block wall host/device layout mismatch")
+            cuda.memcpy_htod(ptr, struct.pack("=IIIf", cfg.nx, cfg.ny,
+                                            cfg.wall_slot_width, cfg.wall_normal_velocity))
         return level
 
     def _build_pair(self, index, pair):
         """Upload the transfer tables of levels (index, index+1) and the child's block kinds."""
         parent, child = self._levels[index], self._levels[index + 1]
         order = np.argsort(pair.coalesce_coarse)
-        fine = None if child.layout is None else blocked_fine_tables(pair, child.layout, order)
-        base = None
-        if parent.layout is not None:
-            if index == 0:
-                base = blocked_base_tables(pair, parent.layout, order, self._wall_aware)
-            else:
-                # Nested pairs keep inherited INNER/OUTER and sparse holes of the parent.
-                base = blocked_base_tables(pair, parent.layout, order,
-                                           composed_kind=self.hierarchy.kinds[index])
-        ghosts = pair.ghost_fine if fine is None else fine[0]
-        donors = pair.ghost_coarse if base is None else base[0]
-        if fine is not None:
-            donors = donors.astype(np.uint64)
-        targets = pair.coalesce_coarse[order] if base is None else base[1]
-        children = pair.coalesce_fine[order] if fine is None else fine[1]
-        result = _Pair(child.module.get_function("PatchExplode" if fine is None else "BlockExplode"),
-                       None if parent.layout is None else parent.module.get_function("BlockCoalesce"),
-                       self._upload(ghosts), self._upload(donors), self._upload(targets),
-                       self._upload(children), len(ghosts), len(targets),
-                       int(np.prod(pair.coarse_shape if parent.layout is None
-                                   else parent.layout.local_shape)))
+        ghosts, children, child_kind = blocked_fine_tables(pair, child.layout, order)
+        if index == 0:
+            donors, targets, parent_kind = blocked_coarse_tables(pair, parent.layout, order,
+                                                               self._wall_aware)
+        else:
+            # Nested pairs keep inherited INNER/OUTER and sparse holes of the parent.
+            donors, targets, parent_kind = blocked_coarse_tables(
+                pair, parent.layout, order, composed_kind=self.hierarchy.kinds[index])
+        result = _Pair(child.module.get_function("BlockExplode"),
+                       parent.module.get_function("BlockCoalesce"),
+                       self._upload(ghosts), self._upload(donors.astype(np.uint64)),
+                       self._upload(targets), self._upload(children), len(ghosts), len(targets),
+                       int(np.prod(parent.layout.local_shape)))
         if self._explosion == "projected_linear":
             neighbors, offsets = projected_explosion_tables(
                 pair, parent.layout, self.hierarchy.kinds[index])
             result.explode = child.module.get_function("BlockExplodeProjected")
             result.neighbors, result.offsets = self._upload(neighbors), self._upload(offsets)
-        if child.layout is not None:
-            child.kind = self._upload(blocked_kind(child.layout, self.hierarchy.kinds[index + 1]))
-        if index == 0 and base is not None:
-            parent.kind = self._upload(base[2])
+        child.kind = self._upload(blocked_kind(child.layout, self.hierarchy.kinds[index + 1]))
+        if index == 0:
+            parent.kind = self._upload(parent_kind)
         return result
 
     def _allocate(self, size):
@@ -461,8 +471,6 @@ class FixedPatchStepper:
     def _check_storage(self):
         if any(bool(c.ddf_shifting) != self._ddf_shifting for c in self._level_cfgs):
             raise ValueError("mixed population representations are unsupported")
-        if self._ddf_shifting and any(layout is None for layout in self._layouts):
-            raise ValueError("shifted populations require all-blocked levels")
 
     def initialize(self, *level_g):
         """Upload physical postcollision G, encoding before any FP32 rounding.
@@ -507,14 +515,13 @@ class FixedPatchStepper:
                         or (values.sum(axis=0, dtype=np.float64) + int(self._ddf_shifting) <= 0).any()):
                     raise ValueError("Leaf populations must be finite with positive density")
             self._mask_nonstate(*arrays)
-            arrays = [a if level.layout is None else level.layout.to_blocked(a)
-                      for a, level in zip(arrays, self._levels)]
+            arrays = [level.layout.to_blocked(a) for a, level in zip(arrays, self._levels)]
             try:
                 for level, array in zip(self._levels, arrays):
                     cuda.memcpy_htod(level.current, array)
-                    cuda.memcpy_htod(level.spare, 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)
+                    # 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)
                 self._step_count = 0
                 self._exchange_valid = False
                 self._window_valid = False
@@ -523,14 +530,15 @@ class FixedPatchStepper:
                 self._status = "failed"
                 raise
 
-    def remap_child(self, rectangles, *, child_objects=()):
-        """Return a NEW ready stepper with an L2 set replaced at this base time.
+    def remap_finest(self, rectangles, *, objects=()):
+        """Return a NEW ready stepper with the level-2 set replaced at this base time.
 
-        Periodic three-level set children only; L0/L1 and block sizes stay fixed.
-        A body run requires its original root-edge-frame ``child_objects``:
+        Periodic three-level hierarchies whose level 2 is a set only; levels 0/1 and
+        block sizes stay fixed. A body run requires its original root-edge-frame
+        level-2 ``objects``:
         stationary circles with identical physical boundary and BFL support.
         Other fluid coverage may change, even in the same connected union.
-        Current L2 Jets (including updated amplitudes)
+        Current level-2 Jets (including updated amplitudes)
         are preserved. Local projected boundary objects are not valid inputs.
         The lock and stream synchronization certify
         current postcollision G and the completed 1:2:4 counts internally.
@@ -548,19 +556,20 @@ class FixedPatchStepper:
             pairs = self.hierarchy.pairs
             if (self._wall_aware or self._channel or self.hierarchy.depth != 3
                     or not isinstance(pairs[1], FixedPatchSet2D)
-                    or self._boundary_hosts[1] is not None
-                    or any(b is not None for b in self._level_boundaries[:2])):
-                raise ValueError("remap_child requires periodic three-level L2 sets with no L0/L1 bodies")
-            cfg = self._remap_cfg
-            child_objects = tuple(child_objects)
-            actual = self._level_boundaries[2]
+                    or self._boundaries[1] is not None
+                    or any(b is not None for b in self._device_boundaries[:2])):
+                raise ValueError("remap_finest requires a periodic three-level hierarchy with a "
+                                 "level-2 set and no level-0/1 bodies")
+            cfg = self._root_cfg
+            objects = tuple(objects)
+            actual = self._device_boundaries[2]
             jets = () if actual is None else actual.jets
             old = RemapTopology.from_config(cfg, pairs[0], pairs[1].bounds,
-                                           child_objects=child_objects, child_jets=jets,
+                                           child_objects=objects, child_jets=jets,
                                            _shifted_blocked=self._ddf_shifting)
-            if (not same_physical_boundary(old.boundary, self._boundary_hosts[2])
-                    or (self._boundary_hosts[2] is not None and actual is None)):
-                raise ValueError("child_objects must reproduce the original root-edge-frame boundary")
+            if (not same_physical_boundary(old.boundary, self._boundaries[2])
+                    or (self._boundaries[2] is not None and actual is None)):
+                raise ValueError("objects must reproduce the original root-edge-frame level-2 boundary")
             if actual is not None:
                 velocity = actual.tables.get("surface_velocity")
                 if velocity is None:
@@ -568,18 +577,17 @@ class FixedPatchStepper:
                                                  _shifted_blocked=self._ddf_shifting)
                 if (not same_physical_boundary(old.boundary, actual.host)
                         or not np.array_equal(old.surface_velocity, velocity)):
-                    raise ValueError("child_objects changed boundary ownership or surface velocity")
+                    raise ValueError("objects changed boundary ownership or surface velocity")
             new = RemapTopology.from_config(cfg, pairs[0], rectangles,
-                                           child_objects=child_objects, child_jets=jets,
+                                           child_objects=objects, child_jets=jets,
                                            _shifted_blocked=self._ddf_shifting)
             if (not same_physical_boundary(old.boundary, new.boundary)
                     or not np.array_equal(old.surface_velocity, new.surface_velocity)):
                 raise ValueError("remap requires the identical pinned physical boundary and BFL support")
-            base_block, fine_block, child_block = self._level_blocks
             # Check sparse block alignment/launch size before any device read/allocation.
-            layout = new.child.fine_layout(child_block)
+            layout = new.child.fine_layout(self._level_blocks[2])
             if layout.nblocks > MAX_GRID_DIM or layout.local_shape[0] > MAX_GRID_DIM:
-                raise ValueError("child blocks exceed launch limits")
+                raise ValueError("level 2 blocks exceed launch limits")
             with self.ctx.activate():
                 try:
                     self._stream.synchronize()
@@ -595,8 +603,8 @@ class FixedPatchStepper:
                 phase="postcollision_g", **options)
             candidate = type(self)(
                 cfg, context=self.ctx, refinements=[pairs[0].bounds, new.child.bounds],
-                level_blocks=[base_block, fine_block, child_block],
-                level_objects=[(), (), child_objects], level_jets=[(), (), jets],
+                level_blocks=list(self._level_blocks),
+                level_objects=[(), (), objects], level_jets=[(), (), jets],
                 observe_raw_exchange=self._observe_raw_exchange,
                 observe_window_exchange=self._observe_window, explosion=self._explosion)
             try:
@@ -613,30 +621,9 @@ class FixedPatchStepper:
                 raise
             return candidate
 
-    def _apply_fine_boundary(self, buffer, **kw):
-        """Stage solid source slots; exchange buffer holds this fine substep only."""
-        cl = self._fine_curved
-        if cl is None:
-            if self._boundary_hosts[1] is not None:
-                raise RuntimeError("Fine boundary device buffers are missing")
-            return
-        if self._curved is None or cl.fluid_idx_gpu is None:
-            raise RuntimeError("Fine boundary device buffers are missing")
-        driven = bool(self._level_jets[1]) or any(
-            obj.surface_actuation is not None for obj in self._boundary_hosts[1].objects)
-        if driven and (cl.surface_velocity_gpu is None or cl.link_exchange_gpu is None):
-            raise RuntimeError("Surface boundary device buffers are missing")
-        self._curved(buffer, cl.fluid_idx_gpu, cl.dir_gpu, cl.q_gpu,
-                     cl.rx_gpu, cl.ry_gpu, cl.rz_gpu, cl.fallback_class_gpu, cl.body_id_gpu,
-                     np.uintp(0), np.uintp(0), np.uint32(cl.count),
-                     cl.surface_velocity_gpu if cl.surface_velocity_gpu is not None else np.uintp(0),
-                     cl.link_exchange_gpu if cl.link_exchange_gpu is not None else np.uintp(0),
-                     grid=((cl.count + self._block[0] - 1) // self._block[0], 1), **kw)
+    def set_jet_amplitude(self, index, amplitude, *, level):
+        """Replace Jet ``index`` of ``level_jets[level]`` between steps; no geometry/table rebuild.
 
-    def set_jet_amplitude(self, index, amplitude, *, level=1):
-        """Replace one jet's amplitude between steps; no geometry/table rebuild.
-
-        ``level=1`` preserves the L1 default; use ``level=2`` for child_jets.
         Invalid input raises before any change. A device/sync error marks the
         instance failed (as step/initialize do); host state commits only after
         a successful upload.
@@ -645,8 +632,8 @@ class FixedPatchStepper:
             self._require("new", "ready")
             if (isinstance(level, bool) or not isinstance(level, Integral)
                     or not 1 <= level < self.hierarchy.depth):
-                raise ValueError("Jet level must name an existing refined level (1 = first)")
-            boundary = self._level_boundaries[level]
+                raise ValueError("Jet level must name an existing refined level (1 .. depth-1)")
+            boundary = self._device_boundaries[level]
             jets = list(self._level_jets[level])
             jets[index] = jets[index].with_amplitude(amplitude)  # validates first
             velocity = None
@@ -656,9 +643,7 @@ class FixedPatchStepper:
                 driven = jet_id >= 0
                 velocity[driven] = jet_link_velocity(jets, shape, jet_id)[driven]
             try:
-                if level == 1 and self._fine_curved is not None:
-                    self._fine_curved.set_jet_amplitude(index, amplitude, self._stream)
-                elif velocity is not None:
+                if velocity is not None:
                     self._stream.synchronize()  # no in-flight step reads the column
                     cuda.memcpy_htod(boundary.device["surface_velocity"], velocity)
             except BaseException:
@@ -676,15 +661,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``."""
@@ -703,19 +685,6 @@ class FixedPatchStepper:
         nested = index + 1 < len(self._levels)
         if nested:
             self._advance(index + 1, src, **kw)
-        if level.layout is None:
-            if index == 0:  # dense root: PatchCoarseStep fuses restriction and collision
-                pair = self._pairs[0]
-                level.kernel(src, dst, self._levels[1].current, pair.targets, pair.children,
-                             np.uint32(pair.n_rows), *(np.uint32(v) for v in self.hierarchy.pairs[0].bounds),
-                             np.uint32(self._wall_aware), grid=level.grid, **kw)
-                return
-            if self._boundary_hosts[index] is not None and self._fine_flags is None:
-                raise RuntimeError("Fine boundary flags are missing")
-            self._apply_fine_boundary(src, **kw)
-            flags = self._fine_flags if self._fine_flags is not None else np.uintp(0)
-            level.kernel(src, dst, flags, grid=level.grid, **kw)
-            return
         # Child donors were read above; no halo gather may overwrite staged arrivals below.
         self._gather_halo(level, src, **kw)
         if nested:
@@ -736,13 +705,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.
@@ -771,7 +740,7 @@ class FixedPatchStepper:
                 self._exchange_valid = False
                 for _ in range(n):
                     if self._observe_raw_exchange:  # stream-ordered: holds the last interval only
-                        for b in self._level_boundaries:
+                        for b in self._device_boundaries:
                             if b is not None:
                                 cuda.memset_d8_async(b.interval_integral, 0,
                                                      len(b.tables["cell"]) * 32, self._stream)
@@ -794,7 +763,7 @@ class FixedPatchStepper:
                 raise RuntimeError("window exchange observation is disabled")
             self._window_valid = False
             try:
-                for b in self._level_boundaries:
+                for b in self._device_boundaries:
                     if b is not None and b.window_integral is not None:
                         cuda.memset_d8(b.window_integral, 0, len(b.tables["cell"]) * 32)
             except BaseException:
@@ -803,7 +772,23 @@ class FixedPatchStepper:
             self._window_start = self._step_count
             self._window_valid = True
 
-    def read_exchange_window(self, body_id, *, level=1):
+    def _observed_boundary(self, level, body_id):
+        if (isinstance(level, bool) or not isinstance(level, Integral)
+                or not 0 <= level < len(self._levels)):
+            raise ValueError("level must name an existing level")
+        b = self._device_boundaries[level]
+        if (b is None or isinstance(body_id, bool) or not isinstance(body_id, Integral)
+                or not 0 <= body_id < len(b.host.objects)):
+            raise ValueError("body_id must name a body on this level")
+        return b
+
+    def _summarize(self, b, rows, level, body_id, window, **kw):
+        return summarize_exchange(rows, b.tables["body_id"], b.tables["aperture"],
+                                  body_id=body_id, level=level, h=self._exchange_spacing[level],
+                                  time_end=self._step_count, window=window,
+                                  reference_center=self._exchange_centers[level][body_id], **kw)
+
+    def read_exchange_window(self, body_id, *, level):
         """Non-destructive read of [window start, current step count] in base-lattice units."""
         with self._lock, self.ctx.activate():
             self._require("ready")
@@ -814,26 +799,16 @@ class FixedPatchStepper:
             duration = self._step_count - self._window_start
             if duration < 1:
                 raise RuntimeError("exchange window requires at least one completed base interval")
-            if (isinstance(level, bool) or not isinstance(level, Integral)
-                    or not 0 <= level < len(self._levels)):
-                raise ValueError("level must name an existing level")
-            b = self._level_boundaries[level]
-            if (b is None or isinstance(body_id, bool) or not isinstance(body_id, Integral)
-                    or not 0 <= body_id < len(b.host.objects)):
-                raise ValueError("body_id must name a body on this level")
+            b = self._observed_boundary(level, body_id)
             rows = np.empty((len(b.tables["cell"]), 4), np.float64)
             try:
                 cuda.memcpy_dtoh(rows, b.window_integral)
             except BaseException:
                 self._status = "failed"
                 raise
-            return summarize_exchange(rows, b.tables["body_id"], b.tables["aperture"],
-                                      body_id=body_id, level=level, h=self._exchange_spacing[level],
-                                      time_end=self._step_count, window="window",
-                                      reference_center=self._exchange_centers[level][body_id],
-                                      duration=duration)
+            return self._summarize(b, rows, level, body_id, "window", duration=duration)
 
-    def read_raw_exchange(self, body_id, *, level=1, window="last_base_step"):
+    def read_raw_exchange(self, body_id, *, level, window="last_base_step"):
         """Read raw boundary reaction about its geometric center in the root edge frame.
 
         Repeated reads and step(0) preserve the window. last_substep samples
@@ -847,13 +822,7 @@ class FixedPatchStepper:
                 raise RuntimeError("raw exchange observation is disabled")
             if not self._exchange_valid:
                 raise RuntimeError("raw exchange requires a completed base interval")
-            if (isinstance(level, bool) or not isinstance(level, Integral)
-                    or not 0 <= level < len(self._levels)):
-                raise ValueError("level must name an existing level")
-            b = self._level_boundaries[level]
-            if (b is None or isinstance(body_id, bool) or not isinstance(body_id, Integral)
-                    or not 0 <= body_id < len(b.host.objects)):
-                raise ValueError("body_id must name a body on this level")
+            b = self._observed_boundary(level, body_id)
             if window not in ("last_base_step", "last_substep"):
                 raise ValueError("window must be last_base_step or last_substep")
             count = len(b.tables["cell"])
@@ -863,10 +832,7 @@ class FixedPatchStepper:
             except BaseException:
                 self._status = "failed"
                 raise
-            return summarize_exchange(rows, b.tables["body_id"], b.tables["aperture"],
-                                      body_id=body_id, level=level, h=self._exchange_spacing[level],
-                                      time_end=self._step_count, window=window,
-                                      reference_center=self._exchange_centers[level][body_id])
+            return self._summarize(b, rows, level, body_id, window)
 
     def download(self):
         """Return physical G (FP64 for shifted storage), NaN on non-leaves."""
@@ -880,7 +846,7 @@ class FixedPatchStepper:
         with self._lock, self.ctx.activate():
             self._require("ready")
             self._check_storage()
-            arrays = [np.empty(level.shape if level.layout is None else level.layout.blocked_shape,
+            arrays = [np.empty(level.layout.blocked_shape,
                                np.float32) for level in self._levels]
             try:
                 for level, array in zip(self._levels, arrays):
@@ -888,7 +854,7 @@ class FixedPatchStepper:
             except BaseException:
                 self._status = "failed"
                 raise
-            arrays = [a if level.layout is None else level.layout.to_dense(a)
+            arrays = [level.layout.to_dense(a)
                       for a, level in zip(arrays, self._levels)]
             self._mask_nonstate(*arrays)
             return _StoredPopulations(tuple(arrays), self._ddf_shifting)
@@ -907,21 +873,13 @@ class FixedPatchStepper:
                         self._stream.synchronize()
                 except Exception as exc:
                     error = exc
-                for b in getattr(self, "_level_boundaries", ()):
+                for b in getattr(self, "_device_boundaries", ()):
                     if b is not None:
                         b.flags = b.device = b.kernel = None
                         b.last_raw = b.interval_integral = b.accumulate = None
                         b.window_integral = None
                 self._levels.clear()
                 self._pairs = []
-                if getattr(self, "_fine_curved", None) is not None:
-                    try:
-                        self._fine_curved.free()
-                    except Exception as exc:
-                        error = error or exc
-                    self._fine_curved = None
-                self._curved = self._fine_flags = None
-                self._link_tables = {}
                 for pointer in self._allocations:
                     try:
                         pointer.free()
@@ -942,28 +900,10 @@ class FixedPatchStepper:
             if exc[0] is None:
                 raise
 
-    # Read-only compatibility views over the level lists, kept on purpose for the existing
-    # tests and docs. Internal code must not use them; it reads hierarchy/_layouts/_level_*.
-    patch = property(lambda self: self.hierarchy.pairs[0])
-    child_patch = property(lambda self: self.hierarchy.pairs[1] if self.hierarchy.depth > 2 else None)
-    base_layout = property(lambda self: self._layouts[0])
-    fine_layout = property(lambda self: self._layouts[1])
-    child_layout = property(lambda self: self._layouts[2] if len(self._layouts) > 2 else None)
-    fine_boundary = property(lambda self: self._boundary_hosts[1])
-    child_boundary = property(lambda self: self._boundary_hosts[2] if len(self._boundary_hosts) > 2
-                              else None)
-    _jets = property(lambda self: self._level_jets[1])
-    _boundaries = property(lambda self: {i: b for i, b in enumerate(self._level_boundaries) if i})
-    _fine_tables = property(lambda self: None if self._level_boundaries[1] is None
-                            else self._level_boundaries[1].tables)
-    _actuated = property(lambda self: self._level_boundaries[1] is not None
-                         and self._level_boundaries[1].actuated)
-    _jet_cache = property(lambda self: None if self._level_boundaries[1] is None
-                          else self._level_boundaries[1].jet_cache)
-    _fine_leaf = property(lambda self: self._leaf[1])
-    _child_leaf = property(lambda self: self._leaf[2] if len(self._leaf) > 2 else None)
-    _middle_kind = property(lambda self: self.hierarchy.kinds[1] if self.hierarchy.depth > 2 else None)
-
-
-# Preserve the original entry point and its periodic default.
-PeriodicPatchStepper = FixedPatchStepper
+    # Read-only per-level state, indexed by level (0 = root). Pairs: hierarchy.pairs[l-1].
+    layouts = property(lambda self: tuple(self._layouts))
+    boundaries = property(lambda self: tuple(self._boundaries),
+                          doc="Host boundary geometry per level (None: no bodies/Jets).")
+    jets = property(lambda self: tuple(self._level_jets))
+    leaf = property(lambda self: tuple(m.copy() for m in self._leaf),
+                    doc="Per-level (ny, nx) masks of public (non-NaN) download cells.")

src/CelerisLab/lbm/refinement_tables.py

@@ -36,7 +36,7 @@ def blocked_kind(layout, dense_kind):
     kind[owner[owned]] = np.asarray(dense_kind).ravel()[owned]
     return kind
 
-def blocked_base_tables(patch, layout, order, wall_aware=False, *, composed_kind=None):
+def blocked_coarse_tables(patch, layout, order, wall_aware=False, *, composed_kind=None):
     """Parent donor cells and receiver-local arrival slots, in transfer row order.
 
     Stage arrivals only after child reads and parent halo gather, immediately

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

tests/integration/test_jet_update_gpu.py

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

tests/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_channel_root.py

@@ -7,7 +7,7 @@ from CelerisLab.config import LBMConfig
 from CelerisLab.lbm.blocks import BlockLayout2D
 from CelerisLab.lbm.channel import channel_masks, validate_channel_config, validate_channel_transfers
 from CelerisLab.lbm.refinement import FixedPatch2D
-from CelerisLab.lbm.refinement_tables import blocked_base_tables
+from CelerisLab.lbm.refinement_tables import blocked_coarse_tables
 from .test_numerical_operators import compile_run
 
 
@@ -31,7 +31,7 @@ def transfer(bounds, block=(4, 6)):
     cfg = LBMConfig(nx=24, ny=20, viscosity=.1)
     pair = FixedPatch2D.from_config(cfg, bounds)
     layout = BlockLayout2D.build(20, 24, *block)
-    targets = blocked_base_tables(pair, layout, np.argsort(pair.coalesce_coarse))[1]
+    targets = blocked_coarse_tables(pair, layout, np.argsort(pair.coalesce_coarse))[1]
     return pair, layout, targets
 
 

tests/unit/test_example_tiles.py

@@ -0,0 +1,70 @@
+"""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_periodic_patch_tiles(self):
+        with self.assertRaisesRegex(AssertionError, "CUDA reached"):
+            FixedPatchStepper(LBMConfig(nx=16, ny=16, viscosity=.1, velocity=1e-3), context=forbidden(),
+                              refinements=[(4, 4, 12, 12)], level_blocks=[(16, 16), (24, 24)])
+
+    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_jet_update_failure.py

@@ -0,0 +1,149 @@
+"""Jet amplitude update failure contract (host doubles only, no CUDA numerics).
+
+Both owners validate before any device write, commit host state only after the
+upload returns, and refuse every later call except close() after an uncertain
+device write.
+"""
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+pytest.importorskip("pycuda.driver")
+from CelerisLab import simulation as module
+from CelerisLab.body.jets import Jet, Radial, Sector
+from tests.unit.test_pinned_body_remap import ready
+from tests.unit.test_simulation_lifecycle import make_simulation
+
+BAD = (float("nan"), float("inf"), 1.)
+
+
+class Curved:
+    """Records uploads; ``fail`` raises once the write was attempted."""
+    def __init__(self):
+        self.writes, self.fail = [], False
+
+    def set_jet_amplitude(self, index, amplitude, stream):
+        self.writes.append((index, amplitude))
+        if self.fail:
+            raise RuntimeError("upload failed")
+
+
+def simulation(monkeypatch):
+    sim = make_simulation(monkeypatch)
+    jet = Jet(Sector((6., 5.), 0., .5, 4.), Radial((6., 5.)), "uniform", 1e-3)
+    sim._failed, sim._initialized = None, True
+    sim.bodies, sim.field = SimpleNamespace(jets=[jet]), SimpleNamespace(curved=Curved())
+    return sim, jet
+
+
+def test_simulation_invalid_input_writes_nothing_and_stays_usable(monkeypatch):
+    sim, jet = simulation(monkeypatch)
+    for amplitude in BAD:
+        with pytest.raises(ValueError):
+            sim.set_jet_amplitude(0, amplitude)
+    with pytest.raises(IndexError):
+        sim.set_jet_amplitude(1, 2e-3)
+    assert sim.field.curved.writes == [] and sim.bodies.jets[0] is jet and sim._failed is None
+    sim.set_jet_amplitude(0, 2e-3)  # retry after a rejected input
+    assert sim.field.curved.writes == [(0, 2e-3)] and sim.bodies.jets[0].amplitude == 2e-3
+    sim.close()
+
+
+def test_simulation_upload_failure_keeps_host_and_blocks_reuse(monkeypatch):
+    sim, jet = simulation(monkeypatch)
+    sim.field.curved.fail = True
+    with pytest.raises(RuntimeError, match="upload failed"):
+        sim.set_jet_amplitude(0, 2e-3)
+    assert sim.bodies.jets[0] is jet and sim._failed is not None
+    sim.field.curved.fail = False
+    for call in (lambda: sim.set_jet_amplitude(0, 2e-3), lambda: sim.run(1), sim.step,
+                 sim.sync_bodies, sim.initialize, sim.get_ddf, sim.get_macroscopic,
+                 lambda: sim.read_surface_exchange(0), lambda: sim.read_body(0),
+                 sim.read_bodies, sim.snapshot, sim.restore, sim.save_checkpoint,
+                 lambda: sim.load_checkpoint("x.h5"), lambda: sim.update_runtime_params(omega=1.)):
+        with pytest.raises(RuntimeError, match="Simulation failed"):
+            call()
+    assert sim.field.curved.writes == [(0, 2e-3)]
+    sim.close()  # cleanup still works
+    assert sim._closed and sim.ctx.is_closed
+
+
+def test_simulation_stream_creation_failure_fails_closed(monkeypatch):
+    sim, jet = simulation(monkeypatch)
+    created = []
+    def stream():
+        created.append(1)
+        raise RuntimeError("stream creation failed")
+    monkeypatch.setattr(module.cuda, "Stream", stream, raising=False)
+    sim._stream = None
+    with pytest.raises(RuntimeError, match="stream creation failed"):
+        sim.set_jet_amplitude(0, 2e-3)
+    assert sim.bodies.jets[0] is jet and sim._failed is not None
+    assert sim.field.curved.writes == [] and created == [1]
+    for call in (lambda: sim.stream, lambda: sim.set_jet_amplitude(0, 2e-3)):
+        with pytest.raises(RuntimeError, match="Simulation failed"):
+            call()
+    assert created == [1]  # the public stream property does not retry creation
+    sim.ctx.detach_failure = True
+    with pytest.raises(RuntimeError, match="detach failed"):
+        sim.close()
+    assert not sim._closed
+    sim.close()  # close stays retryable after the failure
+    assert sim._closed and sim.ctx.is_closed
+
+
+def test_simulation_before_initialize_commits_host_only(monkeypatch):
+    sim, _ = simulation(monkeypatch)
+    sim._initialized = False
+    sim.set_jet_amplitude(0, -2e-3)
+    assert sim.field.curved.writes == [] and sim.bodies.jets[0].amplitude == -2e-3
+    sim.close()
+
+
+def stepper_state(run):
+    boundary = run._device_boundaries[2]
+    return boundary.tables["surface_velocity"].copy(), boundary.jets, run._level_jets[2]
+
+
+def test_stepper_invalid_input_writes_nothing_and_stays_ready():
+    run, module, _, _, _ = ready()
+    try:
+        velocity, jets, level_jets = stepper_state(run)
+        module.cuda.memcpy_htod.reset_mock()
+        for args, kw, error in [((0, a), {"level": 2}, ValueError) for a in BAD] + [
+                ((1, .01), {"level": 2}, IndexError), ((0, .01), {"level": 3}, ValueError)]:
+            with pytest.raises(error):
+                run.set_jet_amplitude(*args, **kw)
+        module.cuda.memcpy_htod.assert_not_called()
+        np.testing.assert_array_equal(run._device_boundaries[2].tables["surface_velocity"], velocity)
+        assert (run._device_boundaries[2].jets, run._level_jets[2]) == (jets, level_jets)
+        assert run._status == "ready"
+        run.set_jet_amplitude(0, .017, level=2)
+        assert run._level_jets[2][0].amplitude == .017
+    finally:
+        run.close()
+
+
+@pytest.mark.parametrize("where", ["upload", "synchronize"])
+def test_stepper_device_failure_keeps_host_and_blocks_reuse(where):
+    run, module, _, _, _ = ready()
+    try:
+        velocity, jets, level_jets = stepper_state(run)
+        if where == "upload":
+            module.cuda.memcpy_htod.side_effect = RuntimeError("upload failed")
+        else:
+            run._stream.synchronize.side_effect = RuntimeError("upload failed")
+        with pytest.raises(RuntimeError, match="upload failed"):
+            run.set_jet_amplitude(0, .017, level=2)
+        np.testing.assert_array_equal(run._device_boundaries[2].tables["surface_velocity"], velocity)
+        assert (run._device_boundaries[2].jets, run._level_jets[2]) == (jets, level_jets)
+        assert run._status == "failed"
+        module.cuda.memcpy_htod.side_effect = None
+        run._stream.synchronize.side_effect = None
+        for call in (lambda: run.set_jet_amplitude(0, .017, level=2), run.step, run.download):
+            with pytest.raises(RuntimeError, match="failed"):
+                call()
+    finally:
+        run.close()
+    assert run._status == "closed"

tests/unit/test_jets.py

@@ -194,6 +194,7 @@ def test_public_surface_exchange_for_jet_preserves_aperture_at_zero_and_reverse(
     field, bodies = jet_case(device, [jet])
     sim = SimpleNamespace(bodies=bodies, field=field, _initialized=True,
                           stream=SimpleNamespace(synchronize=device.synchronize))
+    sim._require_usable = Simulation._require_usable.__get__(sim)
     cl = field.curved
     assert bodies.get(0).surface_actuation is None
     aperture, velocity = cl.aperture, cl.surface_velocity.copy()
@@ -218,6 +219,7 @@ def test_public_surface_exchange_keeps_legacy_and_passive_guards(device):
     bodies.sync_to_gpu(field)
     sim = SimpleNamespace(bodies=bodies, field=field,
                           stream=SimpleNamespace(synchronize=device.synchronize))
+    sim._require_usable = Simulation._require_usable.__get__(sim)
     for value in Simulation.read_surface_exchange(sim, 0).values():
         np.testing.assert_array_equal(value, 0.)
     field.curved.free_gpu_columns()

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_projected_pair_stepper.py

@@ -146,7 +146,6 @@ class ProjectedPairDispatch(unittest.TestCase):
         for cfg, changes in [(CFG, {'explosion':'unknown'}),
                              (CFG, {'coarse_boundary':'slot_reservoir'}),
                              (CFG, {'coarse_boundary':'channel'}),
-                             (CFG, {'level_blocks':[None,None], 'refinements':[(6,6,26,26)]}),
                              (CFG, {'level_objects':[(),(object(),)]}),
                              (CFG, {'level_jets':[(),(object(),)]}),
                              (replace(CFG,collision='MRT'), {}),
@@ -155,9 +154,8 @@ class ProjectedPairDispatch(unittest.TestCase):
                 host.FixedPatchStepper(cfg, **dict(options, **changes))
         context.activate.assert_not_called()
         host.cuda.mem_alloc.assert_not_called()
-        # The legacy constant path still allows MRT and dense storage.
-        for options in ({'cfg':replace(CFG,collision='MRT')},
-                        {'level_blocks':[None,None], 'refinements':[(6,6,26,26)]}):
+        # The constant path still allows MRT.
+        for options in ({'cfg':replace(CFG,collision='MRT')},):
             run, _, _, _, _ = backend(**options)
             run.close()
 

tests/unit/test_raw_exchange.py

@@ -85,7 +85,7 @@ class RawExchange(unittest.TestCase):
 
     def test_four_level_schedule_publication_and_fixed_storage(self):
         run, host, calls, arrays = backend()
-        b = run._level_boundaries[3]
+        b = run._device_boundaries[3]
         rows = len(b.tables['cell'])
         self.assertEqual(b.last_raw.size, rows * 16)
         self.assertEqual(b.interval_integral.size, rows * 32)
@@ -128,7 +128,7 @@ class RawExchange(unittest.TestCase):
 
     def test_disabled_has_nulls_and_no_observation_allocation_or_launch(self):
         run, host, calls, _ = backend(False)
-        b = run._level_boundaries[3]
+        b = run._device_boundaries[3]
         self.assertIs(b.last_raw, b.interval_integral)
         self.assertIs(b.interval_integral, None)
         run.step()
@@ -153,15 +153,6 @@ class RawExchange(unittest.TestCase):
         run._stream.synchronize.side_effect = None
         run.close()
 
-    def test_dense_observation_rejected_before_device_access(self):
-        host = host_module()
-        ctx = Mock()
-        with self.assertRaisesRegex(ValueError, 'blocked body'):
-            host.FixedPatchStepper(LBMConfig(nx=32, ny=32, viscosity=.1), context=ctx,
-                                   refinements=[(4, 4, 28, 28)],
-                                   level_objects=[(), (body(radius=2.15),)], observe_raw_exchange=True)
-        ctx.activate.assert_not_called()
-
     def test_all_depth_weights(self):
         for level in (1, 2, 3, 7):
             self._check_weights_split_and_window(level)

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_mrt.py

@@ -27,23 +27,20 @@ class RefinementMRTHost(unittest.TestCase):
         block_collision = (KERNELS / "operators/block_collision.cuh").read_text()
         self.assertIn('#include "operators/block_collision.cuh"', blocks)
         self.assertEqual(blocks.count("block_collide(f);"), 1)
-        refinement = (KERNELS / "refinement.cu").read_text() + (KERNELS / "step/patch_step.cu").read_text()
         for text in ("compute_feq_from_momentum(drho, jx, jy, ux, uy, feq);",
                      "compute_feq(rho, ux, uy, feq);",
                      "collide_srt_no_force(f, feq, d_params.omega);"):
             self.assertIn(text, block_collision)
         self.assertIn("collide_srt(f, feq, forcing, omega);", channel)
-        self.assertIn("collide_srt_no_force(f, feq, d_params.omega);", refinement)
-        for source in (blocks, refinement):
+        for source in (blocks,):
             self.assertIn("(COLLISION_MODEL != 0 && COLLISION_MODEL != 2)", source)
             self.assertIn("USE_LES", source)
             self.assertIn("STORE_PRECISION != 0", source)
             self.assertIn('#if COLLISION_MODEL == 2\n#include "operators/collision_mrt.cuh"\n#endif', source)
-        self.assertIn("USE_DDF_SHIFTING ||", refinement)  # dense/mixed shifted stays rejected
         # One shared unforced MRT body; each blocked/patch path calls it exactly once.
         mrt = (KERNELS / "operators/collision_mrt.cuh").read_text()
         self.assertEqual(mrt.count("void collide_mrt_unforced("), 1)
-        for source in (blocks + "\n" + block_collision, channel, refinement):
+        for source in (blocks + "\n" + block_collision, channel):
             self.assertEqual(source.count("collide_mrt_unforced(f);"), 1)
             self.assertNotIn("collide_mrt_no_force(", source)
             self.assertNotIn("compute_rho_u(f, rho, ux, uy);\n            collide_mrt", source)

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
 
@@ -140,7 +141,7 @@ class Gate(unittest.TestCase):
     def test_sparse_parent_tables_preserve_inherited_kinds(self):
         child,kind=r.nest_patch_set(CFG,self.parent,CHILDREN)
         layout=self.parent.fine_layout((4,4));order=np.argsort(child.coalesce_coarse)
-        donors,targets,bkind=host.blocked_base_tables(child,layout,order,composed_kind=kind)
+        donors,targets,bkind=host.blocked_coarse_tables(child,layout,order,composed_kind=kind)
         self.assertTrue((donors < layout.hole).all())
         self.assertTrue((targets < np.prod(layout.blocked_shape)).all())
         np.testing.assert_array_equal(bkind[layout.owner[layout.owned]],kind[layout.owned])
@@ -157,10 +158,10 @@ class Gate(unittest.TestCase):
         hole=int(np.flatnonzero(~layout.owned.ravel())[0])
         bad=child.ghost_coarse.copy();bad[0]=hole
         with self.assertRaisesRegex(ValueError,'owned uncovered'):
-            host.blocked_base_tables(replace(child,ghost_coarse=bad),layout,order,composed_kind=kind)
+            host.blocked_coarse_tables(replace(child,ghost_coarse=bad),layout,order,composed_kind=kind)
         bad=child.coalesce_coarse.copy();bad[0]=hole
         with self.assertRaisesRegex(ValueError,'owned uncovered'):
-            host.blocked_base_tables(replace(child,coalesce_coarse=bad),layout,order,composed_kind=kind)
+            host.blocked_coarse_tables(replace(child,coalesce_coarse=bad),layout,order,composed_kind=kind)
 
 
 def bodies_and_jets(speed=.002):

tests/unit/test_shifted_jets.py

@@ -67,7 +67,7 @@ class ShiftedJets(unittest.TestCase):
     def test_switch_upload_only_then_raw_remap_keeps_live_velocity(self):
         run,module,allocations,storage,objects=ready();clone=None
         try:
-            boundary=run._level_boundaries[2]
+            boundary=run._device_boundaries[2]
             before=run._download_raw()
             original=boundary.tables['surface_velocity'].copy()
             module.cuda.memcpy_htod.reset_mock()
@@ -79,7 +79,7 @@ class ShiftedJets(unittest.TestCase):
             for a,b in zip(before.values,run._download_raw().values):self.assertEqual(a.tobytes(),b.tobytes())
             clone=run.remap_child(B,child_objects=objects)
             self.assertEqual(clone._level_jets[2][0].amplitude,-jet().amplitude)
-            np.testing.assert_array_equal(clone._level_boundaries[2].tables['surface_velocity'],
+            np.testing.assert_array_equal(clone._device_boundaries[2].tables['surface_velocity'],
                                           boundary.tables['surface_velocity'])
             self.assertEqual(clone.step_count,7)
             saved=run._download_raw()
@@ -92,12 +92,12 @@ class ShiftedJets(unittest.TestCase):
     def test_invalid_amplitude_does_not_upload_or_commit(self):
         run,module,_,_,_=ready()
         try:
-            before=run._level_boundaries[2].tables['surface_velocity'].copy()
+            before=run._device_boundaries[2].tables['surface_velocity'].copy()
             module.cuda.memcpy_htod.reset_mock()
             for amplitude in (float('nan'),float('inf'),1.):
                 with self.assertRaises(ValueError):run.set_jet_amplitude(0,amplitude,level=2)
             module.cuda.memcpy_htod.assert_not_called()
-            np.testing.assert_array_equal(before,run._level_boundaries[2].tables['surface_velocity'])
+            np.testing.assert_array_equal(before,run._device_boundaries[2].tables['surface_velocity'])
         finally:run.close()
 
 

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
@@ -128,34 +125,24 @@ class ShiftedStorage(unittest.TestCase):
             with self.assertRaisesRegex(ValueError,'mixed'): run._download_raw()
         finally: run.close()
 
-    def test_dense_or_mixed_shifted_rejected_before_compilation(self):
+    def test_missing_tiles_rejected_before_compilation(self):
         module=backend.host_module()
         cfg=replace(backend.CFG,ddf_shifting=True)
         for blocks in ([None,None],[(4,4),None],[None,(4,4)]):
-            with self.assertRaisesRegex(ValueError,'all-blocked|blocked refined levels below'):
+            with self.assertRaisesRegex(ValueError,'must be a tile'):
                 module.FixedPatchStepper(cfg,context=None,refinements=[(4,4,20,28)],
                                          level_blocks=blocks)
         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._device_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/unit/test_state_regressions.py

@@ -295,6 +295,7 @@ def test_passive_overlap_failure_keeps_live_state(device, entry, second_x):
                           _initialized=entry == "runtime_edit",
                           _assert_allocation_contract=lambda: None,
                           stream=SimpleNamespace(synchronize=device.synchronize))
+    sim._require_usable = Simulation._require_usable.__get__(sim)
     with pytest.raises(ValueError, match="active fluid receivers|unique cut-link writers"):
         if entry == "sync":
             bodies.sync_to_gpu(field)

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._device_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_exchange_window.py

@@ -59,7 +59,7 @@ class ExchangeWindowCPU(unittest.TestCase):
             stack.enter_context(patch.object(cuda, "memcpy_dtoh",
                                              side_effect=lambda out, src: out.fill(0)))
             backend = build(m)
-            b = backend._level_boundaries[1]
+            b = backend._device_boundaries[1]
             self.assertIsNotNone(b.window_integral)
             with self.assertRaises(RuntimeError):
                 backend.start_exchange_window()  # not initialized
@@ -96,7 +96,7 @@ class ExchangeWindowCPU(unittest.TestCase):
         with mocked_backend() as m, ExitStack() as stack:
             stack.enter_context(patch.object(cuda, "memset_d8_async"))
             backend = build(m, window=False)
-            self.assertIsNone(backend._level_boundaries[1].window_integral)
+            self.assertIsNone(backend._device_boundaries[1].window_integral)
             backend.initialize(*[np.full(l.shape, 1 / 9, np.float32) for l in backend._levels])
             backend.step(1)
             self.assertEqual(len(accumulations(m)), 2)
@@ -109,7 +109,7 @@ class ExchangeWindowGPU(unittest.TestCase):
         from test_raw_exchange_gpu import BOUNDS as RB, CFG, circle as rc, jet, state
         from CelerisLab.cuda.context import CudaContext
         with CudaContext(0) as ctx:
-            kw = dict(context=ctx, refinements=[RB], level_blocks=[None, (16, 20)],
+            kw = dict(context=ctx, refinements=[RB], level_blocks=[(12, 8), (16, 20)],
                       level_objects=[(), (rc(),)], level_jets=[(), (jet(),)],
                       observe_raw_exchange=True)
             with FixedPatchStepper(CFG, observe_window_exchange=True, **kw) as windowed, \
@@ -120,7 +120,7 @@ class ExchangeWindowGPU(unittest.TestCase):
                 windowed.step(2)
                 plain.step(2)
                 windowed.start_exchange_window()
-                b = windowed._level_boundaries[1]
+                b = windowed._device_boundaries[1]
                 count = len(b.tables["cell"])
                 factors = np.array([.25, .25, .125, .25])  # h*dt, h*h*dt, h*h for level-1 h=.5
                 scale = np.zeros(4)
@@ -212,7 +212,7 @@ class ExchangeWindowShiftedMRTGPU(unittest.TestCase):
                     run.step(2)
                     if run is not plain:
                         run.start_exchange_window()
-                b = windowed._level_boundaries[1]
+                b = windowed._device_boundaries[1]
                 count = len(b.tables["cell"])
                 record = dict(scale=np.zeros(4), substeps=0)
                 audit_substeps(windowed, b, count, record)
@@ -228,7 +228,7 @@ class ExchangeWindowShiftedMRTGPU(unittest.TestCase):
                     same_bits(self, a, e)
                 rows = [np.empty((count, 4), np.float64) for _ in range(2)]
                 cuda.memcpy_dtoh(rows[0], b.window_integral)
-                cuda.memcpy_dtoh(rows[1], split._level_boundaries[1].window_integral)
+                cuda.memcpy_dtoh(rows[1], split._device_boundaries[1].window_integral)
                 np.testing.assert_array_equal(rows[0].view(np.uint64), rows[1].view(np.uint64))
 
                 out = windowed.read_exchange_window(0)

tests/validation/multigrid/test_fine_surface_gpu.py

@@ -1,13 +1,17 @@
-"""Two coarse/four fine steps: isolated fine circle, no flux-accuracy claim.
+"""Two coarse/four fine steps: isolated fine circle on blocked storage, no flux-accuracy claim.
 
-Run only through the bounded GPU runner after a fresh resource check. The test
-hooks read actual kernel inputs/outputs; they never change the stepping order.
+Run only through the bounded GPU runner after a fresh resource check. The fine
+step hook reads the actual output buffer; it never changes the stepping order.
+
+Retired with the dense refinement route (accepted losses, see the D2 commit):
+the dense-kernel per-hop flux ledger (fine/coarse transfer + boundary budget),
+the CurvedLinkSoA reflection oracle and the CELERIS_REFINEMENT_AUDIT collision
+counts. The blocked test keeps the four-hop comparison with the uniform solver,
+the raw-exchange mass sign and passive == zero actuation; raw exchange covers
+the boundary part only, not full population flux closure.
 """
 import importlib.util
-from contextlib import nullcontext
 from pathlib import Path
-from threading import RLock
-from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
 import numpy as np
@@ -18,7 +22,6 @@ 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.context import CudaContext
 from CelerisLab.lbm import refinement_stepper as implementation
 from CelerisLab.lbm.descriptors import FLUID, SOLID
@@ -27,10 +30,8 @@ from CelerisLab.lbm.descriptors import FLUID, SOLID
 ROOT = Path(__file__).resolve().parents[3]
 CX = np.array([0, 1, -1, 0, 0, 1, -1, 1, -1])
 CY = np.array([0, 0, 0, 1, -1, 1, -1, -1, 1])
-OPP = np.array([0, 2, 1, 4, 3, 6, 5, 8, 7])
 W = np.array([4/9]+[1/9]*4+[1/36]*4)
 MOMENTS = np.stack((np.ones(9), CX, CY))
-GEOMETRY = ("fluid_idx", "dir", "q", "body_id", "rx", "ry", "rz", "fallback_class")
 BOUNDS = (4, 4, 20, 20)
 
 
@@ -53,197 +54,6 @@ def object_spec(speed):
                      radius=2.15, surface_actuation=surface(speed))
 
 
-def audit_compiler(original):
-    def compile_private(*args, **kwargs):
-        entry = Path(kwargs["build_dir"])/"fine_surface_audit.cu"
-        entry.write_text('#define CELERIS_REFINEMENT_AUDIT\n#include "refinement.cu"\n')
-        kwargs["source"] = entry.name
-        return original(*args, **kwargs)
-    return compile_private
-
-
-def reflection_oracle(before, cl, speed, *, verify_projection=True):
-    """FP64 BFL from frozen spatial populations, independent arc projection."""
-    flat = before.astype(float).reshape(9, -1)
-    radius = np.hypot(cl.rx.astype(float), cl.ry.astype(float))
-    delta = (np.arctan2(cl.ry.astype(float), cl.rx.astype(float))-.7+np.pi) % (2*np.pi)-np.pi
-    amplitude = 0 if speed is None else speed
-    strength = amplitude*np.maximum(0, 1-(delta/.85)**2)**2
-    velocity = strength[:, None]*np.column_stack((cl.rx, cl.ry))/radius[:, None]
-    if verify_projection and cl.surface_velocity is not None:
-        np.testing.assert_allclose(cl.surface_velocity, velocity, atol=3e-9, rtol=0)
-    incoming = []
-    for i, (k, d, q, fallback) in enumerate(zip(cl.fluid_idx, cl.dir, cl.q, cl.fallback_class)):
-        k, d, q = int(k), int(d), float(q)
-        rho = flat[:, k].sum()
-        dot = CX[d]*velocity[i, 0]+CY[d]*velocity[i, 1]
-        if fallback:
-            value = flat[d, k]-6*W[d]*rho*dot
-        elif q < .5:
-            value = 2*q*flat[d, k]+(1-2*q)*flat[d, k-CX[d]-40*CY[d]]-6*W[d]*rho*dot
-        else:
-            value = flat[d, k]/(2*q)+(1-1/(2*q))*flat[OPP[d], k]-3*W[d]*rho*dot/q
-        incoming.append(value)
-    return np.array(incoming)
-
-
-def fine_flux(before, staged, active, solid):
-    """Population ledger across the fine leaf boundary: solid versus halo.
-
-    Internal fluid-to-fluid streaming cancels pairwise; solid and halo packets
-    are counted in their actual direction, before any collision.
-    """
-    boundary, transfer = np.zeros(9), np.zeros(9)
-    magnitude = 0.0
-    for d, (cx, cy) in enumerate(zip(CX, CY)):
-        incoming = active & ~np.roll(active, (cy, cx), (0, 1))
-        outgoing = active & ~np.roll(active, (-cy, -cx), (0, 1))
-        from_solid = np.roll(solid, (cy, cx), (0, 1))
-        into_solid = np.roll(solid, (-cy, -cx), (0, 1))
-        received = np.roll(staged[d], (cy, cx), (0, 1)).astype(float)
-        sent = before[d].astype(float)
-        for vector, inc, out in ((boundary, incoming & from_solid, outgoing & into_solid),
-                                 (transfer, incoming & ~from_solid, outgoing & ~into_solid)):
-            assert np.isfinite(received[inc]).all() and np.isfinite(sent[out]).all()
-            vector[d] = received[inc].sum()-sent[out].sum()
-            magnitude += np.abs(received[inc]).sum()+np.abs(sent[out]).sum()
-    return boundary, transfer, magnitude
-
-
-def coarse_flux(before, fine, active):
-    """Derive every replacement from physical midpoints, not production tables."""
-    x0, y0, x1, y1 = BOUNDS
-    incoming = np.stack([np.roll(before[d], (CY[d], CX[d]), (0, 1)) for d in range(9)]).astype(float)
-    magnitude = 0.0
-    for y, x in np.argwhere(active):
-        for d in range(1, 9):
-            if x0 <= x+.5-CX[d]/2 <= x1 and y0 <= y+.5-CY[d]/2 <= y1:
-                a, b = 4+2*(x-x0), 4+2*(y-y0)
-                children = fine[d, b:b+2, a:a+2].astype(float)
-                assert children.shape == (2, 2) and np.isfinite(children).all()
-                incoming[d, y, x] = children.sum()/4
-                magnitude += np.abs(children).sum()/4
-    assert np.isfinite(incoming[:, active]).all()
-    # This retains the eight grazing-corner replacements of uncovered donors.
-    flux = (incoming[:, active]-before[:, active].astype(float)).sum(axis=1)
-    magnitude += np.abs(incoming[:, active]).sum()+np.abs(before[:, active]).sum(dtype=float)
-    return flux, magnitude
-
-
-def run_patch(speed, *, audit=True):
-    cfg = LBMConfig(nx=24, ny=24, viscosity=.05, velocity=0, rho=1)
-    coarse_active = np.ones((24, 24), bool)
-    coarse_active[4:20, 4:20] = False
-    source = object_spec(speed)
-    state = vars(source.state).copy()
-    with CudaContext(0) as context:
-        build = (patch.object(compiler, "compile_kernel", audit_compiler(compiler.compile_kernel))
-                 if audit else nullcontext())
-        with build:
-            backend = implementation.FixedPatchStepper(cfg, context=context, refinements=[BOUNDS],
-                                             level_objects=[(), (source,)])
-        with backend, context.activate():
-            c, f = backend._levels
-            cl = backend._fine_curved
-            geometry = {name: getattr(cl, name).copy() for name in GEOMETRY}
-            flags = backend.fine_boundary.flags.reshape(40, 40)
-            solid = (flags & SOLID) != 0
-            active = np.zeros((40, 40), bool)
-            active[4:-4, 4:-4] = True
-            active &= (flags & FLUID) != 0
-            active &= ~solid
-            local = backend.fine_boundary.objects[0]
-            assert local.center == (20., 20.) and local.radius == 4.3
-            assert source.center == (12.25, 12.25) and source.radius == 2.15
-            yy, xx = np.mgrid[:40, :40]
-            roi = active & ((xx-20)**2+(yy-20)**2 <= 9**2)
-            backend.initialize(rest((24, 24)), rest((40, 40)))
-            if audit:
-                for level in backend._levels:
-                    for name in ("collision_counts", "explosion_counts", "incoming_counts"):
-                        pointer, size = level.module.get_global(name)
-                        cuda.memset_d8(pointer, 0, size)
-            initial = backend.download()
-            initial_pop = initial[0][:, coarse_active].sum(axis=1, dtype=float)
-            initial_pop += .25*initial[1][:, active].sum(axis=1, dtype=float)
-            boundary, fine_transfer, coarse_transfer = np.zeros(9), np.zeros(9), np.zeros(9)
-            budget_sum, reflection_error, drive_mass = 0., 0., 0.
-            traces = []
-            original_curved, original_fine, original_coarse = backend._curved, f.kernel, c.kernel
-
-            def curved(*args, **kwargs):
-                nonlocal boundary, fine_transfer, budget_sum, reflection_error, drive_mass
-                backend._stream.synchronize()
-                before = read(args[0], f.shape)
-                expected = reflection_oracle(before, cl, speed)
-                original_curved(*args, **kwargs)
-                backend._stream.synchronize()
-                staged = read(args[0], f.shape)
-                d, k = cl.dir.astype(int), cl.fluid_idx.astype(int)
-                values = staged.reshape(9, -1)[OPP[d], k+CX[d]+40*CY[d]]
-                passive = reflection_oracle(before, cl, None, verify_projection=False)
-                drive_mass += .25*float(np.sum(values-passive))
-                error = float(np.max(np.abs(values-expected)))
-                reflection_error = max(reflection_error, error)
-                assert error <= 1e-7
-                np.testing.assert_array_equal(staged[:, active], before[:, active])
-                b, transfer, magnitude = fine_flux(before, staged, active, solid)
-                boundary += .25*b
-                fine_transfer += .25*transfer
-                budget_sum += .25*magnitude
-
-            def fine(*args, **kwargs):
-                nonlocal budget_sum
-                backend._stream.synchronize()
-                staged = read(args[0], f.shape)
-                original_fine(*args, **kwargs)
-                backend._stream.synchronize()
-                advanced = read(args[1], f.shape)
-                assert np.isfinite(advanced[:, active]).all()
-                assert np.isnan(advanced[:, solid]).all()
-                # Include actual collision operands in the FP32 rounding budget.
-                budget_sum += .25*np.abs(staged[:, active]).sum(dtype=float)
-                traces.append(advanced[:, roi].copy())
-
-            def coarse(*args, **kwargs):
-                nonlocal coarse_transfer, budget_sum
-                backend._stream.synchronize()
-                before, fine_values = read(args[0], c.shape), read(args[2], f.shape)
-                transfer, magnitude = coarse_flux(before, fine_values, coarse_active)
-                coarse_transfer += transfer
-                budget_sum += magnitude
-                return original_coarse(*args, **kwargs)
-
-            with patch.object(backend, "_curved", curved), patch.object(f, "kernel", fine), \
-                    patch.object(c, "kernel", coarse):
-                backend.step(2)
-            actual = backend.download()
-            population = actual[0][:, coarse_active].sum(axis=1, dtype=float)
-            population += .25*actual[1][:, active].sum(axis=1, dtype=float)
-            ledger = boundary+fine_transfer+coarse_transfer
-            error = MOMENTS@(population-initial_pop-ledger)
-            budget = 8*np.finfo(np.float32).eps*budget_sum
-            assert np.isfinite(budget_sum) and budget_sum > 0
-            assert np.max(np.abs(error)) <= budget
-            if speed:
-                assert speed*drive_mass > 0
-            assert len(traces) == 4 and backend.step_count == 2
-            if audit:
-                for level, mask, count in ((c, coarse_active, 2), (f, active, 4)):
-                    pointer, size = level.module.get_global("collision_counts")
-                    counts = read(pointer, mask.shape, np.uint32)
-                    assert size == counts.nbytes
-                    np.testing.assert_array_equal(counts, count*mask)
-            for name, values in geometry.items():
-                np.testing.assert_array_equal(getattr(cl, name), values)
-            assert vars(source.state) == state and source.geometry.center == source.center
-            print(f"speed={speed} audit={audit} BFL={reflection_error:.9g} ledger={error.tolist()} "
-                  f"budget={budget:.9g} boundary={list(MOMENTS@boundary)} "
-                  f"transfer={list(MOMENTS@(fine_transfer+coarse_transfer))} "
-                  f"drive_mass={drive_mass:+.9g}; counts={'PASS' if audit else 'normal build'}", flush=True)
-            return traces, roi, actual
-
-
 def uniform_reference(speed, roi):
     # Reuse the existing explicit config helper without importing unrelated tests.
     spec = importlib.util.spec_from_file_location("fine_surface_cases", ROOT/"tests/integration/_cases.py")
@@ -261,62 +71,59 @@ def uniform_reference(speed, roi):
         return traces
 
 
+def run_blocked_patch(speed):
+    """All-blocked run (whole-level tiles) of the same case; fine ROI after each fine hop."""
+    cfg = LBMConfig(nx=24, ny=24, viscosity=.05, velocity=0, rho=1)
+    source = object_spec(speed)
+    with CudaContext(0) as context:
+        with implementation.FixedPatchStepper(cfg, context=context, refinements=[BOUNDS],
+                                              level_blocks=[(24, 24), (40, 40)],
+                                              level_objects=[(), (source,)],
+                                              observe_raw_exchange=True) as backend:
+            c, f = backend._levels
+            flags = backend.fine_boundary.flags.reshape(40, 40)
+            active = np.zeros((40, 40), bool)
+            active[4:-4, 4:-4] = True
+            active &= ((flags & FLUID) != 0) & ((flags & SOLID) == 0)
+            yy, xx = np.mgrid[:40, :40]
+            roi = active & ((xx-20)**2+(yy-20)**2 <= 9**2)
+            backend.initialize(rest((24, 24)), rest((40, 40)))
+            traces, original = [], f.kernel
+
+            def fine(*args, **kwargs):
+                original(*args, **kwargs)
+                backend._stream.synchronize()
+                advanced = f.layout.to_dense(read(args[1], f.layout.blocked_shape))
+                assert np.isfinite(advanced[:, active]).all()
+                assert np.isnan(advanced[:, (flags & SOLID) != 0]).all()
+                traces.append(advanced[:, roi].copy())
+
+            with patch.object(f, "kernel", fine):
+                backend.step(2)
+            assert len(traces) == 4 and backend.step_count == 2
+            mass = backend.read_raw_exchange(0)["total"]["mass_into_fluid"]
+            return traces, roi, backend.download(), float(mass)
+
+
 @pytest.mark.parametrize("speed", [0., .002, -.002])
-def test_fine_surface_four_hops(speed):
-    assert Path(CelerisLab.__file__).resolve() == ROOT/"src/CelerisLab/__init__.py"
-    assert Path(compiler._KERNEL_DIR).resolve() == ROOT/"src/CelerisLab/lbm/kernels"
-    assert Path(implementation.__file__).resolve() == ROOT/"src/CelerisLab/lbm/refinement_stepper.py"
-    actual, roi, final = run_patch(speed)
+def test_fine_surface_blocked_matches_uniform_four_hops(speed):
+    """Blocked storage: fine ROI after each of four hops against the uniform 48^2 solver
+    (same tolerances as the dense test); blowing/suction drives mass with its sign;
+    zero actuation equals the passive circle bit for bit."""
+    actual, roi, final, mass = run_blocked_patch(speed)
     reference = uniform_reference(speed, roi)
     for got, wanted in zip(actual, reference):
         np.testing.assert_allclose(got, wanted, atol=3e-6, rtol=0)
         np.testing.assert_allclose(MOMENTS@got.astype(float), MOMENTS@wanted.astype(float),
                                    atol=2e-5, rtol=0)
-    if speed == 0:
-        passive, passive_roi, passive_final = run_patch(None, audit=False)
+    if speed:
+        assert speed*mass > 0
+    else:
+        passive, passive_roi, passive_final, _ = run_blocked_patch(None)
         np.testing.assert_array_equal(passive_roi, roi)
         for got, wanted in zip(actual+list(final), passive+list(passive_final)):
             np.testing.assert_array_equal(got, wanted)
-
-
-@pytest.mark.parametrize("missing", ["surface_velocity_gpu", "link_exchange_gpu"])
-def test_missing_surface_device_buffer_rejects_before_launch(missing):
-    backend = implementation.FixedPatchStepper.__new__(implementation.FixedPatchStepper)
-    backend._boundary_hosts = [None, SimpleNamespace(objects=[object_spec(.002)])]
-    backend._level_jets = [(), ()]
-    backend._fine_curved = SimpleNamespace(fluid_idx_gpu=object(),
-                                          surface_velocity_gpu=object(), link_exchange_gpu=object())
-    setattr(backend._fine_curved, missing, None)
-    backend._curved = Mock()
-    with pytest.raises(RuntimeError, match="Surface boundary device buffers are missing"):
-        backend._apply_fine_boundary(object())
-    backend._curved.assert_not_called()
-
-
-def test_close_frees_fine_boundary_after_stream_failure_once():
-    backend = implementation.FixedPatchStepper.__new__(implementation.FixedPatchStepper)
-    backend.ctx = SimpleNamespace(activate=nullcontext)
-    backend._lock = RLock()
-    backend._status = "ready"
-    stream, soa, first, second = Mock(), Mock(), Mock(), Mock()
-    stream.synchronize.side_effect = RuntimeError("stream synchronization failed")
-    backend._stream = stream
-    backend._fine_curved = soa
-    backend._fine_flags = first
-    backend._curved = Mock()
-    backend._levels = [object(), object()]
-    backend._allocations = [first, second]
-    with pytest.raises(RuntimeError, match="stream synchronization failed"):
-        backend.close()
-    assert backend._status == "closed"
-    assert backend._levels == [] and backend._allocations == []
-    assert backend._fine_curved is None and backend._fine_flags is None
-    assert backend._curved is None and backend._stream is None
-    backend.close()
-    stream.synchronize.assert_called_once_with()
-    soa.free.assert_called_once_with()
-    first.free.assert_called_once_with()
-    second.free.assert_called_once_with()
+    print(f"blocked speed={speed} four hops vs uniform PASS mass_into_fluid={mass:+.9g}", flush=True)
 
 
 def test_cross_interface_donor_rejects_before_cuda():
@@ -328,6 +135,7 @@ def test_cross_interface_donor_rejects_before_cuda():
     with patch.object(cuda, "mem_alloc") as allocate:
         with pytest.raises(ValueError, match="donor crosses"):
             implementation.FixedPatchStepper(cfg, context=context, refinements=[BOUNDS],
+                                             level_blocks=[(24, 24), (40, 40)],
                                              level_objects=[(), (source,)])
         context.activate.assert_not_called()
         allocate.assert_not_called()

tests/validation/multigrid/test_raw_exchange_gpu.py

@@ -92,8 +92,8 @@ def census(run, index, buffer):
 
 
 def fine(block):
-    """Two-level circle+jet construction on BOUNDS; block None keeps the dense fine level."""
-    return dict(refinements=[BOUNDS], level_blocks=[None, block],
+    """Two-level circle+jet construction on BOUNDS, blocked with a (12, 8) root tile."""
+    return dict(refinements=[BOUNDS], level_blocks=[(12, 8), block],
                 level_objects=[(), (circle(),)], level_jets=[(), (jet(),)])
 
 class RawExchangeGPU(unittest.TestCase):
@@ -105,7 +105,7 @@ class RawExchangeGPU(unittest.TestCase):
                     arrays = [state(l.shape) for l in measured._levels]
                     measured.initialize(*arrays)
                     plain.initialize(*arrays)
-                    b = measured._level_boundaries[1]
+                    b = measured._device_boundaries[1]
                     count = len(b.tables['cell'])
                     kernel = measured._levels[1].kernel
                     samples = []
@@ -146,26 +146,6 @@ class RawExchangeGPU(unittest.TestCase):
                         np.testing.assert_array_equal(measured.read_raw_exchange(0)['total']['impulse'],
                                                       obs['total']['impulse'])
 
-    def test_dense_fine_storage_reference(self):
-        """Same fine resolution/initial data with dense storage: a numerical
-        reference for cut-link rows, not a uniform-whole-domain convergence study."""
-        with CudaContext(0) as ctx:
-            with FixedPatchStepper(CFG, context=ctx, **fine(None)) as dense, \
-                 FixedPatchStepper(CFG, context=ctx, **fine((16, 20)), observe_raw_exchange=True) as blocked:
-                arrays = [state(l.shape) for l in blocked._levels]
-                dense.initialize(*arrays)
-                blocked.initialize(*arrays)
-                for speed in (0., .002, 0.):
-                    dense.set_jet_amplitude(0, speed)
-                    blocked.set_jet_amplitude(0, speed)
-                    dense.step(); blocked.step()
-                    for a, e in zip(blocked.download(), dense.download()):
-                        np.testing.assert_array_equal(a, e)
-                    count = dense._fine_curved.count
-                    expected = download(dense._fine_curved.link_exchange_gpu, (count, 3))
-                    actual = download(blocked._level_boundaries[1].last_raw, (count, 4))
-                    np.testing.assert_array_equal(actual[:, :3], expected)
-
     def test_depth_four_interval_storage_and_sampling(self):
         cfg = LBMConfig(nx=48, ny=48, viscosity=.02)
         obj = SimObject(0, geometry=CircleGeometry(16., 16., .5), center=(16., 16.), radius=.5)
@@ -178,7 +158,7 @@ class RawExchangeGPU(unittest.TestCase):
                  FixedPatchStepper(cfg, context=ctx, **settings) as plain:
                 arrays = [state(l.shape, ux=0.) for l in run._levels]
                 run.initialize(*arrays); plain.initialize(*arrays)
-                b = run._level_boundaries[3]
+                b = run._device_boundaries[3]
                 count = len(b.tables['cell'])
                 kernel = run._levels[3].kernel
                 samples, independent_samples = [], []
@@ -224,34 +204,35 @@ 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)))
-                b = amr._level_boundaries[1]
+                b = amr._device_boundaries[1]
                 count = len(b.tables['cell'])
                 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._device_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,12 @@
 
 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 refinement storage was removed: every level is blocked and level_blocks
+is required (TileContractCPU). The retired dense/blocked route parity gates
+(BlockedBaseGPU exact parity, DenseBlockedRouteParity incl. MRT and two
+bodies, the route capability matrix) passed and are recorded at 1573942 and
+2469af7.
 """
 from contextlib import ExitStack, contextmanager, nullcontext
 from dataclasses import replace
@@ -13,12 +19,13 @@ 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 CelerisLab.lbm.refinement_stepper import FixedPatchStepper
+from CelerisLab.lbm.refinement_tables import blocked_coarse_tables
 from test_refinement_blocked_fine import NX, NY, BOUNDS, config, state, circle, actuation
 
 BASE_BLOCKS = ((24, 32), (12, 8), (3, 4), (1, 1))
@@ -73,6 +80,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))))
@@ -101,7 +109,7 @@ class BlockedBaseCPU(unittest.TestCase):
                 leaves[[0, -1]] = False
             for by, bx in BASE_BLOCKS:
                 layout = BlockLayout2D.build(NY, NX, by, bx, periodic=(True, mode == "periodic"))
-                donors, targets, kind = blocked_base_tables(p, layout, order, mode != "periodic")
+                donors, targets, kind = blocked_coarse_tables(p, layout, order, mode != "periodic")
                 ly, lx = layout.local_shape
                 L = ly * lx
                 expected, source_cells = [], []
@@ -131,7 +139,7 @@ class BlockedBaseCPU(unittest.TestCase):
         p = FixedPatch2D.from_config(config("periodic"), BOUNDS)
         order = np.argsort(p.coalesce_coarse)
         layout = BlockLayout2D.build(NY, NX, 3, 4)
-        donors, targets, _ = blocked_base_tables(p, layout, order)
+        donors, targets, _ = blocked_coarse_tables(p, layout, order)
         dense = np.arange(9 * NY * NX, dtype=np.float32).reshape(9, NY, NX)
         initial = layout.to_blocked(dense)
         arrivals = np.arange(len(targets), dtype=np.float32) + 10000
@@ -171,7 +179,7 @@ class BlockedBaseCPU(unittest.TestCase):
         # With one parent block, all eight grazing sources are owner cores.
         # Each is read for four fine ghost children: late explosion corrupts32.
         single = BlockLayout2D.build(NY, NX, NY, NX)
-        donors, targets, _ = blocked_base_tables(p, single, order)
+        donors, targets, _ = blocked_coarse_tables(p, single, order)
         initial = single.to_blocked(dense)
         staged = initial.copy()
         staged.ravel()[targets.astype(np.intp)] = arrivals
@@ -265,62 +273,33 @@ class BlockedBaseCPU(unittest.TestCase):
                 allocation.free.assert_called_once_with()
 
 
-class BlockedBaseGPU(unittest.TestCase):
-    def test_bodyfree_exact_parity_and_raw_invalid_slots(self):
-        self.compare([(None, ((24, 32), (3, 4), (1, 1)), 8)])
+class TileContractCPU(unittest.TestCase):
+    """level_blocks is required: one positive (by, bx) tile per level, checked before CUDA."""
 
-    def test_actuated_circle_exact_parity_and_raw_invalid_slots(self):
-        self.compare([(.002, ((12, 8),), 8), (-.002, ((12, 8),), 8)])
-
-    def test_non_power_two_slot_width_exact_parity(self):
-        self.compare([(None, ((3, 4),), 7), (None, ((3, 4),), 10)], ("slot_reservoir",))
-
-    def compare(self, cases, modes=("periodic", "slot_reservoir")):
-        from CelerisLab.cuda.context import CudaContext
-        expected_source = Path(__file__).resolve().parents[3] / "src/CelerisLab/lbm/refinement_stepper.py"
-        self.assertEqual(Path(refinement_stepper.__file__).resolve(), expected_source)
-        with CudaContext(0) as context:
-            for mode in modes:
-                for speed, layouts, width in cases:
-                    cfg = config(mode)
-                    if mode == "slot_reservoir":
-                        cfg = replace(cfg, wall_slot_width=width)
-                    p = FixedPatch2D.from_config(cfg, BOUNDS)
-                    arrays = state(NY, NX, 1), state(*p.fine_shape, 2)
-                    reference = None
-                    for base_block in (None, *layouts):
-                        label = f"{mode} width={width} speed={speed} base={base_block}"
-                        snapshots = []
-                        with FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
-                                               refinements=[BOUNDS], level_blocks=[base_block, FINE_BLOCK],
-                                               level_objects=[(), () if speed is None
-                                                              else (circle(actuation(speed)),)]) as backend:
-                            backend.initialize(*arrays)
-                            for step in (0, 1, 2, 8):
-                                if step == 1 and base_block is not None:
-                                    # Blocked staging dirties invalid input slots; its output
-                                    # must poison them. Dense parent does not stage there.
-                                    coarse = backend._levels[0]
-                                    spare = coarse.layout.to_blocked(np.full(coarse.shape, 42, np.float32))
-                                    cuda.memcpy_htod(coarse.spare, spare)
-                                backend.step(step - backend.step_count)
-                                public, owner = backend.download(), raw(backend)
-                                self.assertEqual(backend.step_count, step)
-                                self.assertTrue(np.isnan(owner[0][:, 6:18, 8:24]).all(), label)
-                                if mode != "periodic":
-                                    self.assertTrue(np.isnan(owner[0][:, (0, -1), :]).all(), label)
-                                snapshots.append(public + owner)
-                            backend.initialize(*arrays)
-                            self.assertEqual(backend.step_count, 0)
-                            for actual, expected in zip(backend.download() + raw(backend), snapshots[0]):
-                                exact(self, actual, expected, label + " reset")
-                        if reference is None:
-                            reference = snapshots
-                        else:
-                            for step, actual, expected in zip((0, 1, 2, 8), snapshots, reference):
-                                for index, (a, b) in enumerate(zip(actual, expected)):
-                                    exact(self, a, b, f"{label} step={step} array={index}")
-                            print(f"PASS {label}: full finite bits/NaN masks at 0,1,2,8 + reset", flush=True)
+    def test_level_blocks_required_and_validated(self):
+        context = Mock(activate=Mock(side_effect=AssertionError("CUDA reached")))
+        cfg = config("periodic")
+        for label, kw in (("omitted", {}),
+                          ("None list", dict(level_blocks=None)),
+                          ("too few levels", dict(level_blocks=[(12, 8)])),
+                          ("too many levels", dict(level_blocks=[(12, 8), FINE_BLOCK, FINE_BLOCK])),
+                          ("None root", dict(level_blocks=[None, FINE_BLOCK])),
+                          ("None fine", dict(level_blocks=[(12, 8), None])),
+                          ("wrong arity", dict(level_blocks=[(12,), FINE_BLOCK])),
+                          ("non-integer", dict(level_blocks=[(12, 8.0), FINE_BLOCK])),
+                          ("bool", dict(level_blocks=[(True, 8), FINE_BLOCK])),
+                          ("non-positive", dict(level_blocks=[(0, 8), FINE_BLOCK])),
+                          ("non-tuple", dict(level_blocks=[12, FINE_BLOCK])),
+                          ("not dividing", dict(level_blocks=[(7, 8), FINE_BLOCK]))):
+            with self.subTest(label), self.assertRaises(ValueError):
+                FixedPatchStepper(cfg, context=context, refinements=[BOUNDS], **kw)
+        with self.assertRaisesRegex(ValueError, "whole-level tile"):
+            FixedPatchStepper(cfg, context=context, refinements=[BOUNDS])
+        context.activate.assert_not_called()
+        # The documented whole-level tiles are always valid (construction reaches CUDA).
+        whole = [(NY, NX), FixedPatch2D.from_config(cfg, BOUNDS).fine_shape]
+        with self.assertRaisesRegex(AssertionError, "CUDA reached"):
+            FixedPatchStepper(cfg, context=context, refinements=[BOUNDS], level_blocks=whole)
 
 
 if __name__ == "__main__":

tests/validation/multigrid/test_refinement_blocked_fine.py

@@ -1,9 +1,8 @@
-"""Opt-in blocked fine storage for FixedPatchStepper: CPU tables plus GPU gate.
+"""Blocked fine storage for FixedPatchStepper: CPU tables, staging and launch contracts.
 
-Frozen GPU gate (same stream, 8 coarse steps, nonequilibrium start), in both
-periodic and slot_reservoir modes: dense fine storage vs 1x1, 2x2 and 4x2
-fine blocks give bit-identical masked downloads AND bit-identical raw coarse
-and unpacked owner-core fine arrays, including patch-ghost values/NaN masks.
+The former GPU gate (dense fine storage vs 1x1, 2x2 and 4x2 fine blocks,
+bit-identical) was retired with the dense route; its result and the
+dense/blocked route parity are recorded at 1573942 and 2469af7.
 """
 from contextlib import nullcontext
 import unittest
@@ -15,14 +14,15 @@ 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
 from CelerisLab.lbm.descriptors import SOLID
 from CelerisLab.lbm.refinement import (ACTIVE, INNER_GHOST, OUTER_GHOST, FixedPatch2D,
                                        prepare_fine_boundary)
-from CelerisLab.lbm.refinement_stepper import FixedPatchStepper, blocked_fine_tables
+from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
+from CelerisLab.lbm.refinement_tables import blocked_fine_tables
 
 NX, NY, BOUNDS, STEPS = 32, 24, (8, 6, 24, 18), 8
 FINE_BLOCKS = ((32, 40), (16, 20), (16, 10))  # fine window 32x40: 1x1, 2x2, 4x2
@@ -88,15 +88,13 @@ def seam_cells(layout, mapped):
 
 
 def raw(backend):
-    """Unmasked device arrays; blocked fine storage unpacked from owner cores."""
-    coarse, fine = backend._levels
-    out = [np.empty(coarse.shape, np.float32)]
-    out.append(np.empty(fine.shape if backend.fine_layout is None
-                        else backend.fine_layout.blocked_shape, np.float32))
-    cuda.memcpy_dtoh(out[0], coarse.current)
-    cuda.memcpy_dtoh(out[1], fine.current)
-    if backend.fine_layout is not None:
-        out[1] = backend.fine_layout.to_dense(out[1])
+    """Unmasked device arrays of every level; blocked storage unpacked from owner cores."""
+    out = []
+    for level in backend._levels:
+        array = np.empty(level.shape if level.layout is None else level.layout.blocked_shape,
+                         np.float32)
+        cuda.memcpy_dtoh(array, level.current)
+        out.append(array if level.layout is None else level.layout.to_dense(array))
     return out
 
 
@@ -122,12 +120,12 @@ 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"):
             backend = FixedPatchStepper(config("periodic"), context=context, refinements=[BOUNDS],
-                                        level_blocks=[None, FINE_BLOCKS[0]])
+                                        level_blocks=[(12, 8), FINE_BLOCKS[0]])
             self.assertEqual(len(backend.fine_layout.halo_dst), 0)
             self.assertIsNone(backend._levels[1].halo_dst)
             backend._gather_halo(backend._levels[1], Mock())  # no launch without pairs
@@ -172,14 +170,14 @@ 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()), \
                 patch.object(cuda, "memcpy_htod"), \
                 patch.object(curved_links, "CurvedLinkSoA", side_effect=AssertionError("dense SoA built")):
             backend = FixedPatchStepper(config("periodic"), context=context, refinements=[BOUNDS],
-                                        level_blocks=[None, FINE_BLOCKS[1]],
+                                        level_blocks=[(12, 8), FINE_BLOCKS[1]],
                                         level_objects=[(), tuple(objects)])
             backend.initialize(state(NY, NX, 1), state(*backend.patch.fine_shape, 2))
         return backend, calls
@@ -189,8 +187,8 @@ class BlockedFineCPU(unittest.TestCase):
             backend, calls = self.mocked_backend([circle(spec)])
             c, f = backend._levels
             coarse, cur, spare = c.current, f.current, f.spare
-            flags, kind = backend._fine_flags, backend._levels[1].kind
-            velocity = backend._link_tables.get("surface_velocity")
+            flags, kind = backend._device_boundaries[1].flags, backend._levels[1].kind
+            velocity = backend._device_boundaries[1].device.get("surface_velocity")
             self.assertEqual(velocity is None, spec is None)
             backend.step(1)
             backend.close()
@@ -198,8 +196,9 @@ class BlockedFineCPU(unittest.TestCase):
             self.assertEqual(order, [("BlockExplode", coarse),
                                      ("BlockHaloCopy", cur), ("BlockCurvedBoundary", cur), ("BlockStep", cur),
                                      ("BlockHaloCopy", spare), ("BlockCurvedBoundary", spare),
-                                     ("BlockStep", spare), ("PatchCoarseStep", coarse)])
-            steps = [a for name, a in calls if name == "BlockStep"]
+                                     ("BlockStep", spare), ("BlockHaloCopy", coarse),
+                                     ("BlockCoalesce", cur), ("BlockStep", coarse)])
+            steps = [a for name, a in calls if name == "BlockStep"][:2]  # the two fine hops
             self.assertEqual([a[1] for a in steps], [spare, cur])
             for a in steps:
                 self.assertIs(a[3], flags)  # blocked flags, not null
@@ -215,7 +214,7 @@ class BlockedFineCPU(unittest.TestCase):
         for speed in (0.0, .002):  # any spec, even zero amplitude
             backend, calls = self.mocked_backend([circle(actuation(speed))])
             self.assertIsNotNone(backend._fine_tables["surface_velocity"])
-            del backend._link_tables["surface_velocity"]
+            del backend._device_boundaries[1].device["surface_velocity"]
             with self.assertRaisesRegex(RuntimeError, "Surface velocity device buffer is missing"):
                 backend.step(1)
             names = [name for name, _ in calls]
@@ -281,93 +280,5 @@ class BlockedFineCPU(unittest.TestCase):
             self.assertEqual(set(np.unique(kind).tolist()) - {ACTIVE, INNER_GHOST, OUTER_GHOST}, set())
 
 
-class BlockedFineGPU(unittest.TestCase):
-    def test_dense_and_blocked_fine_storage_are_bit_identical(self):
-        from CelerisLab.cuda.context import CudaContext
-        for mode in ("periodic", "slot_reservoir"):
-            cfg = config(mode)
-            patch = FixedPatch2D.from_config(cfg, BOUNDS)
-            g_coarse, g_fine = state(NY, NX, 1), state(*patch.fine_shape, 2)
-            results = []
-            with CudaContext(0) as context:
-                for fine_block in (None, *FINE_BLOCKS):
-                    with FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
-                                           refinements=[BOUNDS],
-                                           level_blocks=[None, fine_block]) as backend:
-                        backend.initialize(g_coarse, g_fine)
-                        backend.step(STEPS)
-                        self.assertEqual(backend.step_count, STEPS)
-                        results.append((backend.download(), raw(backend)))
-            (dense_public, dense_raw), blocked = results[0], results[1:]
-            kind = patch.fine_kind
-            self.assertTrue(np.isnan(dense_raw[1][:, kind == OUTER_GHOST]).all(), mode)
-            # INNER populations may be NaN after hop 2 (pulled from poisoned OUTER);
-            # their exact pattern is pinned by the dense-vs-blocked comparison below.
-            self.assertTrue(np.isfinite(dense_raw[1][:, kind == ACTIVE]).all(), mode)
-            for fine_block, (public, raw_arrays) in zip(FINE_BLOCKS, blocked):
-                msg = f"{mode} fine_block={fine_block}"
-                for a, b in zip(public + tuple(raw_arrays), dense_public + tuple(dense_raw)):
-                    np.testing.assert_array_equal(a, b, err_msg=msg)  # values and NaN masks
-            print(f"{mode}: dense vs {len(blocked)} fine-block layouts bit-identical "
-                  f"after {STEPS} coarse steps", flush=True)
-
-    def test_passive_circle_dense_and_blocked_fine_storage_are_bit_identical(self):
-        from CelerisLab.cuda.context import CudaContext
-        for mode in ("periodic", "slot_reservoir"):
-            cfg = config(mode)
-            patch, boundary = fine_circle(mode)
-            solid = (boundary.flags.reshape(patch.fine_shape) & SOLID) != 0
-            g_coarse, g_fine = state(NY, NX, 1), state(*patch.fine_shape, 2)
-            results = []
-            with CudaContext(0) as context:
-                for fine_block in (None, *FINE_BLOCKS):
-                    with FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
-                                           refinements=[BOUNDS], level_blocks=[None, fine_block],
-                                           level_objects=[(), (circle(),)]) as backend:
-                        backend.initialize(g_coarse, g_fine)
-                        backend.step(STEPS)
-                        results.append((backend.download(), raw(backend)))
-            (dense_public, dense_raw), blocked = results[0], results[1:]
-            fine = dense_raw[1]
-            self.assertTrue(np.isfinite(fine[:, (patch.fine_kind == ACTIVE) & ~solid]).all(), mode)
-            self.assertTrue(np.isnan(fine[:, solid]).all(), mode)
-            self.assertTrue(np.isnan(fine[:, patch.fine_kind == OUTER_GHOST]).all(), mode)
-            for fine_block, (public, raw_arrays) in zip(FINE_BLOCKS, blocked):
-                msg = f"circle {mode} fine_block={fine_block}"
-                for a, b in zip(public + tuple(raw_arrays), dense_public + tuple(dense_raw)):
-                    np.testing.assert_array_equal(a, b, err_msg=msg)  # values and NaN masks
-            print(f"circle {mode}: dense vs {len(blocked)} fine-block layouts bit-identical "
-                  f"after {STEPS} coarse steps", flush=True)
-
-    def test_actuated_circle_dense_and_blocked_fine_storage_are_bit_identical(self):
-        from CelerisLab.cuda.context import CudaContext
-        for mode in ("periodic", "slot_reservoir"):
-            for speed in (.002, -.002):
-                cfg = config(mode)
-                patch, boundary = fine_circle(mode, actuation(speed))
-                solid = (boundary.flags.reshape(patch.fine_shape) & SOLID) != 0
-                g_coarse, g_fine = state(NY, NX, 1), state(*patch.fine_shape, 2)
-                results = []
-                with CudaContext(0) as context:
-                    for fine_block in (None, *FINE_BLOCKS):
-                        with FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
-                                               refinements=[BOUNDS], level_blocks=[None, fine_block],
-                                               level_objects=[(), (circle(actuation(speed)),)]) as backend:
-                            backend.initialize(g_coarse, g_fine)
-                            backend.step(STEPS)
-                            results.append((backend.download(), raw(backend)))
-                (dense_public, dense_raw), blocked = results[0], results[1:]
-                fine = dense_raw[1]
-                self.assertTrue(np.isfinite(fine[:, (patch.fine_kind == ACTIVE) & ~solid]).all())
-                self.assertTrue(np.isnan(fine[:, solid]).all())
-                self.assertTrue(np.isnan(fine[:, patch.fine_kind == OUTER_GHOST]).all())
-                for fine_block, (public, raw_arrays) in zip(FINE_BLOCKS, blocked):
-                    msg = f"actuated {speed} {mode} fine_block={fine_block}"
-                    for a, b in zip(public + tuple(raw_arrays), dense_public + tuple(dense_raw)):
-                        np.testing.assert_array_equal(a, b, err_msg=msg)  # values and NaN masks
-                print(f"actuated {speed:+.3f} {mode}: dense vs {len(blocked)} fine-block layouts "
-                      f"bit-identical after {STEPS} coarse steps", flush=True)
-
-
 if __name__ == "__main__":
     unittest.main()

tests/validation/multigrid/test_refinement_followup.py

@@ -17,9 +17,10 @@ import pycuda.driver as cuda
 
 import CelerisLab
 from CelerisLab.lbm import refinement_stepper as implementation
+from CelerisLab.lbm.blocks import BlockLayout2D
 from test_refinement_gpu import (ROOT, POP_ATOL, MOMENT_ATOL, ref, postcollision,
                                  totals, LBMConfig, CudaContext, compiler,
-                                 PeriodicPatchStepper)
+                                 FixedPatchStepper)
 
 
 def scalar_interval(coarse, fine, bounds, slot=None):
@@ -83,6 +84,21 @@ def scalar_interval(coarse, fine, bounds, slot=None):
     return result, current
 
 
+def whole_tiles(cfg, bounds):
+    """One whole-level tile per level: (ny, nx) at the root, the fine window below."""
+    return [(cfg.ny, cfg.nx), implementation.FixedPatch2D.from_config(cfg, bounds).fine_shape]
+
+
+def owner_arrays(backend):
+    """Raw owner storage of every level as dense (9, ny, nx), before public masking."""
+    arrays = []
+    for level in backend._levels:
+        blocked = np.empty(level.layout.blocked_shape, np.float32)
+        cuda.memcpy_dtoh(blocked, level.current)
+        arrays.append(level.layout.to_dense(blocked))
+    return arrays
+
+
 class FaultGuardsCPU(unittest.TestCase):
     def test_boundary_mode_and_margin_contract(self):
         cfg = LBMConfig(nx=32, ny=24, viscosity=.1, velocity=0, wall_normal_velocity=.001,
@@ -95,7 +111,7 @@ class FaultGuardsCPU(unittest.TestCase):
             with self.subTest(mode=mode, bounds=bounds):
                 with self.assertRaises(ValueError):
                     implementation.FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
-                                                     refinements=[bounds])
+                                                     refinements=[bounds], level_blocks=[(1, 1)] * 2)
                 context.activate.assert_not_called()
         # Both exact channel limits and the original periodic two-cell margin
         # must reach context activation, with no allocation or device required.
@@ -104,18 +120,19 @@ class FaultGuardsCPU(unittest.TestCase):
             with self.subTest(mode=mode, bounds=bounds):
                 with self.assertRaisesRegex(RuntimeError, "validated before CUDA"):
                     implementation.FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
-                                                     refinements=[bounds])
+                                                     refinements=[bounds], level_blocks=[(1, 1)] * 2)
 
     def fake_backend(self):
-        backend = PeriodicPatchStepper.__new__(PeriodicPatchStepper)
-        backend._layouts = [None, None]  # dense storage on both levels
+        backend = FixedPatchStepper.__new__(FixedPatchStepper)
+        layout = BlockLayout2D.build(1, 1, 1, 1, periodic=(False, False))
+        backend._layouts = [layout, layout]  # blocked storage on both levels
         backend._ddf_shifting = False  # unshifted on every level, as _check_storage expects
         backend._level_cfgs = [LBMConfig(nx=16, ny=16, viscosity=.1)] * 2  # ddf_shifting off
         backend.ctx = Mock(activate=nullcontext)
         backend._lock = RLock()
         backend._status = "ready"
         backend._stream = Mock()
-        backend._levels = [Mock(shape=(9, 1, 1), layout=None), Mock(shape=(9, 1, 1), layout=None)]
+        backend._levels = [Mock(shape=(9, 1, 1), layout=layout), Mock(shape=(9, 1, 1), layout=layout)]
         backend._allocations = [Mock(), Mock()]
         return backend
 
@@ -158,7 +175,8 @@ class FaultGuardsCPU(unittest.TestCase):
         context = Mock(activate=nullcontext, sm_arch="sm_70")
         with patch.object(implementation, "HALO", 2):
             with self.assertRaisesRegex(RuntimeError, "four-cell"):
-                PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)])
+                FixedPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)],
+                                     level_blocks=[(16, 16), (24, 24)])
         p = implementation.FixedPatch2D.from_config(cfg, (4, 4, 12, 12))
         from CelerisLab.lbm import hierarchy
         bad = patch.object(implementation.FixedPatch2D, "from_config",
@@ -166,23 +184,26 @@ 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])
+                FixedPatchStepper(cfg, context=context, refinements=[p.bounds],
+                                     level_blocks=[(16, 16), (24, 24)])
             with patch.object(hierarchy, "_check_chain"), \
                     self.assertRaisesRegex(RuntimeError, "omega/tau"):
-                PeriodicPatchStepper(cfg, context=context, refinements=[p.bounds])
+                FixedPatchStepper(cfg, context=context, refinements=[p.bounds],
+                                     level_blocks=[(16, 16), (24, 24)])
             alloc.assert_not_called()
             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:
             with self.assertRaisesRegex(RuntimeError, "layout mismatch"):
-                PeriodicPatchStepper(cfg, context=context, refinements=[p.bounds])
+                FixedPatchStepper(cfg, context=context, refinements=[p.bounds],
+                                     level_blocks=[(16, 16), (24, 24)])
             upload.assert_not_called()
 
 
@@ -250,23 +271,27 @@ class NormalBuildGPU(unittest.TestCase):
                 parts = line.split()
                 if len(parts) == 3 and parts[0] == "#define":
                     definitions[parts[1]] = parts[2]
-            generated.append((metadata["config"], definitions))
+            generated.append((metadata["config"], definitions, metadata))
             return build
 
         max_error = max_bc_error = 0.0
         with CudaContext(0) as context:
             with patch.object(compiler, "generate_config", side_effect=record_config):
                 backend = implementation.FixedPatchStepper(cfg, context=context, refinements=[bounds],
-                                                            coarse_boundary="slot_reservoir")
+                                                            coarse_boundary="slot_reservoir",
+                                                            level_blocks=whole_tiles(cfg, bounds))
             with backend:
                 self.assertEqual(len(generated), 2)
-                fine_config, fine_defines = generated[1]
-                self.assertEqual((fine_config["nx"], fine_config["ny"]), (fx, fy))
+                fine_config, fine_defines, _ = generated[1]
+                # Blocked kernels see one tile window including the 1-cell halo.
+                self.assertEqual((fine_config["nx"], fine_config["ny"]), (fx+2, fy+2))
                 for name in ("Y_WALL_BC", "TOP_WALL_BC", "WALL_SLOT_WIDTH",
                              "WALL_NORMAL_VELOCITY", "TOP_WALL_NORMAL_VELOCITY"):
                     self.assertEqual(float(fine_defines[name].removesuffix("f")), 0, name)
-                self.assertEqual(int(generated[0][1]["Y_WALL_BC"]), 2)
-                self.assertEqual(int(generated[0][1]["TOP_WALL_BC"]), 1)
+                # The root's physical slot/reservoir walls come from BLOCK_ROOT_BC=1
+                # (module constant d_block_wall); the fine level has none.
+                root_meta, fine_meta = generated[0][2], generated[1][2]  # read before cleanup
+                self.assertEqual((root_meta["block_root_bc"], fine_meta["block_root_bc"]), (1, 0))
                 # Independently derive all target links, including the eight
                 # whose coarse sources are uncovered, from physical midpoints.
                 targets, corners = set(), set()
@@ -286,9 +311,7 @@ class NormalBuildGPU(unittest.TestCase):
                     expected = scalar_interval(*expected, bounds, slot=(amplitude, width))
                     backend.step()
                     self.assertEqual(backend.step_count, step)
-                    raw = [np.empty(a.shape, np.float32) for a in initial]
-                    for level, array in zip(backend._levels, raw):
-                        cuda.memcpy_dtoh(array, level.current)
+                    raw = owner_arrays(backend)
                     # Read raw storage before masking: active NaNs cannot be
                     # hidden by the public download's non-leaf cleanup.
                     self.assertTrue(np.isfinite(raw[0][:, coarse_leaf]).all())
@@ -325,9 +348,9 @@ class NormalBuildGPU(unittest.TestCase):
         state = ref.State(*(a * (1 + 1e-3*rng.standard_normal(a.shape))
                             for a in (initial.coarse_f, initial.fine_f)))
         max_pop = max_drift = max_moment = 0.0
-        with CudaContext(0) as context, PeriodicPatchStepper(
+        with CudaContext(0) as context, FixedPatchStepper(
                 LBMConfig(nx=16, ny=16, viscosity=.1), context=context,
-                refinements=[(4, 4, 12, 12)]) as backend:
+                refinements=[(4, 4, 12, 12)], level_blocks=[(16, 16), (24, 24)]) as backend:
             backend.initialize(*postcollision(state, spec, maps))
             before = totals(backend.download(), maps)
             for step in range(1, 9):
@@ -358,17 +381,17 @@ class NormalBuildGPU(unittest.TestCase):
         fine_mask[4:-4, 4:-4] = True
         initial[1][:, ~fine_mask] = np.nan
         expected = scalar_interval(*[a.astype(np.float64) for a in initial], bounds)
-        with CudaContext(0) as context, PeriodicPatchStepper(cfg, context=context, refinements=[bounds]) as backend:
+        with CudaContext(0) as context, FixedPatchStepper(cfg, context=context, refinements=[bounds],
+                                                             level_blocks=[(13, 17), (14, 16)]) as backend:
             self.assertEqual(backend.patch.fine_shape, (14, 16))
             p = backend.patch
-            allocated = 2 * 9 * (17*13+16*14) * 4 + sum(a.nbytes for a in
-                        (p.ghost_fine, p.ghost_coarse, p.coalesce_coarse, p.coalesce_fine))
+            allocated = 2 * 4 * sum(int(np.prod(level.layout.blocked_shape)) for level in backend._levels) \
+                + sum(a.nbytes for a in (p.ghost_fine, p.ghost_coarse, p.coalesce_coarse, p.coalesce_fine))
             self.assertLess(allocated, 128*1024)
             backend.initialize(*initial)
             backend.step()
             coarse, fine = backend.download()
-            raw_fine = np.empty((9, 14, 16), np.float32)
-            cuda.memcpy_dtoh(raw_fine, backend._levels[1].current)
+            raw_fine = owner_arrays(backend)[1]
             max_error = 0.0
             for got, want in zip((coarse, raw_fine), expected):
                 np.testing.assert_array_equal(np.isnan(got), np.isnan(want))

tests/validation/multigrid/test_refinement_gpu.py

@@ -20,9 +20,9 @@ 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
+from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
 
 ROOT = Path(__file__).resolve().parents[3]
 POP_ATOL = 3e-6
@@ -52,12 +52,11 @@ def totals(arrays, maps):
     return np.array([populations.sum(), populations @ ref.CX, populations @ ref.CY])
 
 
-def global_array(module, name, shape):
-    out = np.empty(shape, dtype=np.uint32)
-    ptr, size = module.get_global(name)
-    assert size == out.nbytes
-    cuda.memcpy_dtoh(out, ptr)
-    return out
+def level_storage(level):
+    """Raw device storage of one blocked level as dense (9, ny, nx)."""
+    blocked = np.empty(level.layout.blocked_shape, np.float32)
+    cuda.memcpy_dtoh(blocked, level.current)
+    return level.layout.to_dense(blocked)
 
 
 class CoupledGPU(unittest.TestCase):
@@ -82,7 +81,7 @@ class CoupledGPU(unittest.TestCase):
         expected = postcollision(next_state, spec, maps)
         arrivals = (next_state.coarse_f, next_state.fine_f)
 
-        cases = (("dense_unshifted", False, {}),
+        cases = (("blocked_unshifted", False, dict(level_blocks=[(16, 16), (24, 24)])),
                  ("blocked_shifted", True,
                   dict(level_blocks=[(4, 4), (6, 6)])))
         with CudaContext(0) as context:
@@ -125,7 +124,7 @@ class CoupledGPU(unittest.TestCase):
                               "wrong-phase controls PASS", flush=True)
 
     def test_cleanup_attempts_every_allocation(self):
-        backend = PeriodicPatchStepper.__new__(PeriodicPatchStepper)
+        backend = FixedPatchStepper.__new__(FixedPatchStepper)
         backend.ctx = Mock(activate=nullcontext)
         backend._lock = RLock()
         backend._status = "failed"
@@ -147,8 +146,9 @@ class CoupledGPU(unittest.TestCase):
         spec = ref.make_spec(collision="srt")
         maps = ref.build_maps(spec)
         with CudaContext(0) as context:
-            with PeriodicPatchStepper(LBMConfig(nx=16, ny=16, viscosity=.1),
-                                      context=context, refinements=[(4, 4, 12, 12)]) as backend:
+            with FixedPatchStepper(LBMConfig(nx=16, ny=16, viscosity=.1),
+                                      context=context, refinements=[(4, 4, 12, 12)],
+                                      level_blocks=[(16, 16), (24, 24)]) as backend:
                 backend.initialize(*postcollision(ref.make_state(spec), spec, maps))
                 coarse, fine = backend._levels
 
@@ -164,122 +164,75 @@ class CoupledGPU(unittest.TestCase):
                 for operation in (backend.download, backend.step):
                     with self.assertRaisesRegex(RuntimeError, "failed"):
                         operation()
-                raw = np.empty(fine.shape, np.float32)
-                cuda.memcpy_dtoh(raw, fine.current)
+                raw = level_storage(fine)
                 self.assertTrue(np.isfinite(raw.ravel()[backend.patch.coalesce_fine]).all())
             self.assertFalse(context.is_closed)
         print("partial failure: two fine hops completed, no coarse step published; cleanup PASS", flush=True)
 
-    def test_bounded_equivalence_and_ownership(self):
-        self.assertEqual(Path(CelerisLab.__file__).resolve(), ROOT / "src/CelerisLab/__init__.py")
-        self.assertEqual(Path(compiler._KERNEL_DIR).resolve(), ROOT / "src/CelerisLab/lbm/kernels")
+    def eight_steps(self, backend, spec, maps):
+        """Rest and seeded nonequilibrium, 8 steps against the scalar reference."""
+        p = backend.patch
+        for case in ("rest", "nonequilibrium_seed1"):
+            with self.subTest(case=case):
+                state = ref.make_state(spec, velocity=(0, 0) if case == "rest" else (.03, -.02))
+                if case != "rest":
+                    rng = np.random.default_rng(1)
+                    state = ref.State(*(a * (1 + 1e-3 * rng.standard_normal(a.shape))
+                                        for a in (state.coarse_f, state.fine_f)))
+                initial = postcollision(state, spec, maps)
+                backend.initialize(*initial)
+                before = totals(backend.download(), maps)
+                max_error = max_drift = max_moment_error = max_corner = 0.0
+                for step in range(1, 9):
+                    state, diagnostic = ref.advance_coarse_once(spec, maps, state)
+                    backend.step()
+                    actual = backend.download()
+                    expected = postcollision(state, spec, maps)
+                    self.assertEqual(backend.step_count, step)
+                    p.check_synchronization(step, state.tick, phase="postcollision_g")
+                    for got, want in zip(actual, expected):
+                        np.testing.assert_array_equal(np.isnan(got), np.isnan(want))
+                        error = float(np.nanmax(np.abs(got - want)))
+                        self.assertLessEqual(error, POP_ATOL)
+                        max_error = max(max_error, error)
+                    drift = float(np.max(np.abs(totals(actual, maps) - before)))
+                    moment_error = float(np.max(np.abs(totals(actual, maps) - totals(expected, maps))))
+                    self.assertLessEqual(drift, MOMENT_ATOL)
+                    self.assertLessEqual(moment_error, MOMENT_ATOL)
+                    max_drift, max_moment_error = max(max_drift, drift), max(max_moment_error, moment_error)
+                    f = backend._levels[1]
+                    raw = level_storage(f)
+                    np.testing.assert_array_equal(np.isfinite(raw), diagnostic["hop2_valid"])
+                    self.assertTrue(np.isfinite(raw.ravel()[p.coalesce_fine]).all())
+                    self.assertEqual(len(diagnostic["corner_packets"]), 8)
+                    for packet in diagnostic["corner_packets"].values():
+                        sx, sy = packet["source_coarse_xy"]
+                        self.assertTrue(maps.coarse_active[sy, sx])
+                        values = [raw[packet["direction"], y, x]
+                                  for y, x in packet["children_fine_yx"]]
+                        error = float(np.max(np.abs(np.array(values) - packet["packets"])))
+                        self.assertLessEqual(error, POP_ATOL)
+                        max_corner = max(max_corner, error)
+                print(f"{case}: steps=8 max_population={max_error:.9g} "
+                      f"max_leaf_drift={max_drift:.9g} max_leaf_error={max_moment_error:.9g} "
+                      f"max_corner_packet={max_corner:.9g}; PASS", flush=True)
+
+    def test_bounded_equivalence_blocked(self):
+        """The same eight-step reference and launch-failure checks on whole-level tiles."""
         spec = ref.make_spec(collision="srt")
         maps = ref.build_maps(spec)
         cfg = LBMConfig(nx=16, ny=16, viscosity=.1)
-        compile_original = compiler.compile_kernel
-
-        def audit_compile(*args, **kwargs):
-            # Only this private build's entry differs. Production source unchanged.
-            entry = Path(kwargs["build_dir"]) / "refinement_audit.cu"
-            entry.write_text('#define CELERIS_REFINEMENT_AUDIT\n#include "refinement.cu"\n')
-            kwargs["source"] = entry.name
-            return compile_original(*args, **kwargs)
-
         with CudaContext(0) as context:
-            with patch.object(compiler, "compile_kernel", audit_compile):
-                backend = PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)])
-            with backend:
-                # Production allocations: four DDF arrays and four compact tables.
-                device_bytes = 2 * 9 * (16**2 + 24**2) * 4
-                p = backend.patch
-                device_bytes += sum(a.nbytes for a in (p.ghost_fine, p.ghost_coarse,
-                                                       p.coalesce_coarse, p.coalesce_fine))
-                audit_bytes = sum((2 + 9) * np.prod(level.shape[1:]) * 4
-                                  for level in backend._levels)
-                self.assertLess(device_bytes + audit_bytes, 128 * 1024)
-                print(f"device_arrays={device_bytes} audit_arrays={audit_bytes} bytes", flush=True)
-                for case in ("rest", "nonequilibrium_seed1"):
-                    with self.subTest(case=case):
-                        state = ref.make_state(spec, velocity=(0, 0) if case == "rest" else (.03, -.02))
-                        if case != "rest":
-                            rng = np.random.default_rng(1)
-                            state = ref.State(*(a * (1 + 1e-3 * rng.standard_normal(a.shape))
-                                                for a in (state.coarse_f, state.fine_f)))
-                        initial = postcollision(state, spec, maps)
-                        backend.initialize(*initial)
-                        for level in backend._levels:
-                            for name in ("collision_counts", "explosion_counts", "incoming_counts"):
-                                ptr, size = level.module.get_global(name)
-                                cuda.memset_d8(ptr, 0, size)
-                        before = totals(backend.download(), maps)
-                        max_error = max_drift = max_moment_error = max_corner = 0.0
-                        for step in range(1, 9):
-                            state, diagnostic = ref.advance_coarse_once(spec, maps, state)
-                            backend.step()
-                            actual = backend.download()
-                            expected = postcollision(state, spec, maps)
-                            self.assertEqual(backend.step_count, step)
-                            p.check_synchronization(step, state.tick, phase="postcollision_g")
-                            for got, want in zip(actual, expected):
-                                np.testing.assert_array_equal(np.isnan(got), np.isnan(want))
-                                error = float(np.nanmax(np.abs(got - want)))
-                                self.assertLessEqual(error, POP_ATOL)
-                                max_error = max(max_error, error)
-                            drift = float(np.max(np.abs(totals(actual, maps) - before)))
-                            moment_error = float(np.max(np.abs(totals(actual, maps) - totals(expected, maps))))
-                            self.assertLessEqual(drift, MOMENT_ATOL)
-                            self.assertLessEqual(moment_error, MOMENT_ATOL)
-                            max_drift, max_moment_error = max(max_drift, drift), max(max_moment_error, moment_error)
-                            c, f = backend._levels
-                            np.testing.assert_array_equal(global_array(c.module, "collision_counts", (16, 16)),
-                                                          step * maps.coarse_active)
-                            np.testing.assert_array_equal(global_array(f.module, "collision_counts", (24, 24)),
-                                                          2 * step * maps.fine_active)
-                            np.testing.assert_array_equal(global_array(f.module, "explosion_counts", (24, 24)),
-                                                          step * ~maps.fine_active)
-                            owners = np.zeros((9, 16, 16), np.uint32)
-                            owners.ravel()[p.coalesce_coarse] = step
-                            np.testing.assert_array_equal(global_array(c.module, "incoming_counts", owners.shape), owners)
-                            raw = np.empty(f.shape, np.float32)
-                            cuda.memcpy_dtoh(raw, f.current)
-                            np.testing.assert_array_equal(np.isfinite(raw), diagnostic["hop2_valid"])
-                            self.assertTrue(np.isfinite(raw.ravel()[p.coalesce_fine]).all())
-                            self.assertEqual(len(diagnostic["corner_packets"]), 8)
-                            for packet in diagnostic["corner_packets"].values():
-                                sx, sy = packet["source_coarse_xy"]
-                                self.assertTrue(maps.coarse_active[sy, sx])
-                                values = [raw[packet["direction"], y, x]
-                                          for y, x in packet["children_fine_yx"]]
-                                error = float(np.max(np.abs(np.array(values) - packet["packets"])))
-                                self.assertLessEqual(error, POP_ATOL)
-                                max_corner = max(max_corner, error)
-                        print(f"{case}: steps=8 max_population={max_error:.9g} "
-                              f"max_leaf_drift={max_drift:.9g} max_leaf_error={max_moment_error:.9g} "
-                              f"max_corner_packet={max_corner:.9g}; ownership/counts PASS", flush=True)
-
-                # Fail before any additional time step; no stale synchronized export.
+            with FixedPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)],
+                                      level_blocks=[(16, 16), (24, 24)]) as backend:
+                self.eight_steps(backend, spec, maps)
                 with patch.object(backend._pairs[0], "explode", side_effect=RuntimeError("injected launch failure")):
                     with self.assertRaisesRegex(RuntimeError, "injected"):
                         backend.step()
-                with self.assertRaisesRegex(RuntimeError, "failed"):
-                    backend.download()
-                with self.assertRaisesRegex(RuntimeError, "failed"):
-                    backend.step()
+                for operation in (backend.download, backend.step):
+                    with self.assertRaisesRegex(RuntimeError, "failed"):
+                        operation()
             self.assertFalse(context.is_closed)
-            with self.assertRaisesRegex(RuntimeError, "closed"):
-                backend.download()
-
-            # Compile and execute the actual non-audited entry as well: one step.
-            with PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)]) as plain:
-                rest = postcollision(ref.make_state(spec), spec, maps)
-                plain.initialize(*rest)
-                plain.step()
-                for got, expected in zip(plain.download(), rest):
-                    np.testing.assert_allclose(got, expected, rtol=0, atol=POP_ATOL, equal_nan=True)
-                print("normal-build rest: steps=1 PASS; source/kernel paths pinned", flush=True)
-        rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
-        print(f"host_peak_rss={rss:.3f} MiB", flush=True)
-
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)

tests/validation/multigrid/test_refinement_jets.py

@@ -3,11 +3,13 @@
 Jets use the same coarse cell-edge frame as fine_objects. The equivalent jet
 of Astra's actuated fixture is SurfaceActuation(0.7, 0.85, U).as_jet((16, 12),
 2.15). Frozen GPU gates (8 coarse steps, both coarse boundary modes):
-  G1 jet run: dense vs 1x1/2x2/4x2 fine blocks bit-identical (values + NaN masks);
-  G2 jet vs legacy actuation (dense): max |dG| over finite fine ACTIVE fluid
+  G1 jet run: 1x1/2x2/4x2 fine blocks (blocked (12, 8) root) bit-identical to the
+     FINE_BLOCKS[0] reference (values + NaN masks);
+  G2 jet vs legacy actuation (blocked reference): max |dG| over finite fine ACTIVE fluid
      populations <= 1e-6 (velocity rows differ only by FP32 rounding; reported);
-  G3 amplitude change after 4 steps (+U -> -U), 4 more steps: dense vs blocked
-     bit-identical and the run differs from the unchanged run.
+  G3 amplitude change after 4 steps (+U -> -U), 4 more steps: blocked layouts
+     bit-identical to the FINE_BLOCKS[0] reference and the run differs from the
+     unchanged run.
 """
 from contextlib import nullcontext
 import unittest
@@ -16,7 +18,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
@@ -26,6 +28,11 @@ from test_refinement_blocked_fine import (BOUNDS, FINE_BLOCKS, NX, NY, STEPS, ac
 SPEED = .002
 
 
+def root_and(fine_block):
+    """Blocked levels with a (12, 8) root tile."""
+    return [(12, 8), fine_block]
+
+
 def jet(speed=SPEED):
     return actuation(speed).as_jet((16.0, 12.0), 2.15)
 
@@ -36,13 +43,13 @@ 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()), \
             patch.object(cuda, "memcpy_htod", htod):
         backend = FixedPatchStepper(config("periodic"), context=context, refinements=[BOUNDS],
-                                    level_blocks=[None, fine_block], level_objects=[(), tuple(objects)],
+                                    level_blocks=root_and(fine_block), level_objects=[(), tuple(objects)],
                                     level_jets=[(), tuple(jets)])
     return backend, htod
 
@@ -100,7 +107,7 @@ class JetStepperCPU(unittest.TestCase):
     def test_blocked_amplitude_update_rewrites_only_velocity_column(self):
         backend, htod = mocked([circle()], [jet()], FINE_BLOCKS[1])
         before = backend._fine_tables["surface_velocity"].copy()
-        device = backend._link_tables["surface_velocity"]
+        device = backend._device_boundaries[1].device["surface_velocity"]
         with patch.object(cuda, "memcpy_htod", htod):
             htod.reset_mock()
             backend.set_jet_amplitude(0, -SPEED)
@@ -116,7 +123,7 @@ class JetStepperCPU(unittest.TestCase):
         backend.close()
 
     def test_copy_failure_marks_failed_and_keeps_host_amplitude(self):
-        for fine_block in (None, FINE_BLOCKS[1]):  # dense SoA and blocked tables
+        for fine_block in (FINE_BLOCKS[1],):  # blocked tables
             backend, _ = mocked([circle()], [jet()], fine_block)
             status = backend._status
             for bad, error in ((1.5, ValueError), (float("nan"), ValueError)):
@@ -126,9 +133,7 @@ class JetStepperCPU(unittest.TestCase):
             with self.assertRaises(IndexError):
                 backend.set_jet_amplitude(1, -SPEED)
             self.assertEqual(backend._status, status)
-            dense = backend._fine_curved
-            before = (dense.surface_velocity if dense is not None
-                      else backend._fine_tables["surface_velocity"]).copy()
+            before = backend._fine_tables["surface_velocity"].copy()
             boom = Mock(side_effect=RuntimeError("copy failed"))
             with patch.object(cuda, "memcpy_htod", boom), patch.object(cuda, "memcpy_htod_async", boom), \
                     patch.object(cuda, "pagelocked_empty", side_effect=lambda shape, dtype: np.empty(shape, dtype)):
@@ -136,11 +141,7 @@ class JetStepperCPU(unittest.TestCase):
                     backend.set_jet_amplitude(0, -SPEED)
             self.assertEqual(backend._status, "failed", fine_block)
             self.assertEqual(backend._jets[0].amplitude, SPEED)
-            if dense is not None:
-                self.assertEqual(dense.jets[0].amplitude, SPEED)
-                np.testing.assert_array_equal(dense.surface_velocity, before)
-            else:
-                np.testing.assert_array_equal(backend._fine_tables["surface_velocity"], before)
+            np.testing.assert_array_equal(backend._fine_tables["surface_velocity"], before)
             with self.assertRaisesRegex(RuntimeError, "failed"):
                 backend.step(1)
             with self.assertRaisesRegex(RuntimeError, "failed"):
@@ -151,6 +152,7 @@ class JetStepperCPU(unittest.TestCase):
         with patch.object(cuda, "Stream", side_effect=AssertionError("CUDA touched")):
             with self.assertRaisesRegex(ValueError, "SurfaceActuation"):
                 FixedPatchStepper(config("periodic"), context=Mock(), refinements=[BOUNDS],
+                                  level_blocks=root_and(FINE_BLOCKS[1]),
                                   level_objects=[(), (circle(actuation(SPEED)),)],
                                   level_jets=[(), (jet(),)])
 
@@ -161,7 +163,7 @@ def run(mode, fine_block, *, objects, jets, switch=None):
     patch_ = FixedPatch2D.from_config(cfg, BOUNDS)
     with CudaContext(0) as context, FixedPatchStepper(
             cfg, context=context, coarse_boundary=mode, refinements=[BOUNDS],
-            level_blocks=[None, fine_block], level_objects=[(), tuple(objects)],
+            level_blocks=root_and(fine_block), level_objects=[(), tuple(objects)],
             level_jets=[(), tuple(jets)]) as backend:
         backend.initialize(state(NY, NX, 1), state(*patch_.fine_shape, 2))
         if switch is None:
@@ -179,29 +181,29 @@ class JetStepperGPU(unittest.TestCase):
             patch_, boundary = fine_circle(mode)
             fluid = (patch_.fine_kind == ACTIVE) & (
                 (boundary.flags.reshape(patch_.fine_shape) & SOLID) == 0)
-            dense = run(mode, None, objects=[circle()], jets=[jet()])
-            self.assertTrue(np.isfinite(dense[1][1][:, fluid]).all(), mode)
-            for fine_block in FINE_BLOCKS:
+            reference = run(mode, FINE_BLOCKS[0], objects=[circle()], jets=[jet()])
+            self.assertTrue(np.isfinite(reference[1][1][:, fluid]).all(), mode)
+            for fine_block in FINE_BLOCKS[1:]:
                 public, raw_arrays = run(mode, fine_block, objects=[circle()], jets=[jet()])
-                for a, b in zip(public + tuple(raw_arrays), dense[0] + tuple(dense[1])):
+                for a, b in zip(public + tuple(raw_arrays), reference[0] + tuple(reference[1])):
                     np.testing.assert_array_equal(a, b, err_msg=f"G1 {mode} {fine_block}")
-            legacy = run(mode, None, objects=[circle(actuation(SPEED))], jets=())
-            diff = np.abs(dense[1][1][:, fluid] - legacy[1][1][:, fluid]).max()
+            legacy = run(mode, FINE_BLOCKS[0], objects=[circle(actuation(SPEED))], jets=())
+            diff = np.abs(reference[1][1][:, fluid] - legacy[1][1][:, fluid]).max()
             print(f"G2 {mode}: jet vs legacy fine max |dG| = {diff:.3e}", flush=True)
             self.assertLessEqual(diff, 1e-6, mode)
 
     def test_g3_amplitude_switch_layouts_identical(self):
         for mode in ("periodic", "slot_reservoir"):
-            dense = run(mode, None, objects=[circle()], jets=[jet()], switch=-SPEED)
-            steady = run(mode, None, objects=[circle()], jets=[jet()])
-            self.assertFalse(np.array_equal(dense[1][1], steady[1][1], equal_nan=True), mode)
-            for fine_block in FINE_BLOCKS:
+            reference = run(mode, FINE_BLOCKS[0], objects=[circle()], jets=[jet()], switch=-SPEED)
+            steady = run(mode, FINE_BLOCKS[0], objects=[circle()], jets=[jet()])
+            self.assertFalse(np.array_equal(reference[1][1], steady[1][1], equal_nan=True), mode)
+            for fine_block in FINE_BLOCKS[1:]:
                 public, raw_arrays = run(mode, fine_block, objects=[circle()], jets=[jet()],
                                          switch=-SPEED)
-                for a, b in zip(public + tuple(raw_arrays), dense[0] + tuple(dense[1])):
+                for a, b in zip(public + tuple(raw_arrays), reference[0] + tuple(reference[1])):
                     np.testing.assert_array_equal(a, b, err_msg=f"G3 {mode} {fine_block}")
-            print(f"G3 {mode}: amplitude switch dense vs {len(FINE_BLOCKS)} layouts identical",
-                  flush=True)
+            print(f"G3 {mode}: amplitude switch, {len(FINE_BLOCKS) - 1} layouts "
+                  f"identical to {FINE_BLOCKS[0]}", flush=True)
 
 
 if __name__ == "__main__":

tests/validation/multigrid/test_refinement_mrt_gpu.py

@@ -54,25 +54,6 @@ class RefinementMRTGPU(unittest.TestCase):
                             else:
                                 for a, b in zip(reference, raw.values): np.testing.assert_array_equal(a, b)
 
-    def test_dense_versus_blocked_mrt_two_level(self):
-        """Unshifted MRT: all-dense PatchCoarseStep/PatchFineStep vs mixed/all-blocked storage."""
-        from test_refinement_blocked_base import BASE_BLOCKS, FINE_BLOCK
-        from test_refinement_blocked_fine import NX, NY, BOUNDS
-        import test_refinement_blocked_fine as fixture
-        cfg = LBMConfig(nx=NX, ny=NY, viscosity=.1, collision="MRT")
-        with CudaContext(0) as context:
-            results = {}
-            layouts = ((None, None), *((base, FINE_BLOCK) for base in (None, *BASE_BLOCKS)))
-            for base_block, fine_block in layouts:
-                with FixedPatchStepper(cfg, context=context, refinements=[BOUNDS],
-                                       level_blocks=[base_block, fine_block]) as backend:
-                    backend.initialize(*[fixture.state(*level.shape[1:], seed) for seed, level in enumerate(backend._levels)])
-                    backend.step(4)
-                    results[base_block, fine_block] = raw(backend)
-            for layout in layouts[1:]:
-                for level, (a, b) in enumerate(zip(results[layout], results[None, None])):
-                    exact(self, a, b, f"layout={layout} level={level}")
-
     def test_shifted_all_blocked_three_level_mrt_smoke(self):
         """Per-level shear rate omega_l = 1/(3*2**l*nu + 1/2); finite ACTIVE leaves."""
         cfg = LBMConfig(nx=16, ny=16, viscosity=.1, collision="MRT", ddf_shifting=True)

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()), \
@@ -75,7 +75,7 @@ class ThreeLevelCPU(unittest.TestCase):
             with self.subTest(kw=kw), self.assertRaises(ValueError):
                 FixedPatchStepper(CFG, context=context, **kw)
         # L2 bodies/Jets without an L2 level (one entry too many), and an L2 Jet without a body.
-        two_level = dict(refinements=[L1], level_blocks=[None, None])
+        two_level = dict(refinements=[L1], level_blocks=[B, B])
         for kw in (dict(two_level, level_objects=[(), (), (child_body(),)]),
                    dict(two_level, level_jets=[(), (), (child_jet(),)]),
                    levels(jets=((), (), (child_jet(),)))):

tests/validation/multigrid/test_shifted_jets_gpu.py

@@ -102,7 +102,7 @@ class Ledger:
 
     def step(self):
         run=self.run;self.samples.clear();run.step()
-        b=run._level_boundaries[self.index];h=run._exchange_spacing[self.index]
+        b=run._device_boundaries[self.index];h=run._exchange_spacing[self.index]
         self.test.assertEqual(len(self.samples),2**self.index)
         expected=np.zeros_like(self.samples[0],dtype=np.float64)
         weights=np.array([h*h,h*h,h*h*h,h*h])
@@ -140,7 +140,7 @@ class ShiftedJetsGPU(unittest.TestCase):
                         level_objects=[(),(obj,)],level_jets=[(),(whole,)],observe_raw_exchange=True) as run:
                     run._initialize_raw(initial(run,rho,u=.013,v=0.))
                     expected=run._download_raw();ledger=Ledger(self,run,1)
-                    self.assertTrue(run._level_boundaries[1].tables['aperture'].all())
+                    self.assertTrue(run._device_boundaries[1].tables['aperture'].all())
                     for _ in range(4):ledger.step()
                     error=max(float(abs(a[:,leaf]-b[:,leaf]).max()) for a,b,leaf in
                               zip(run._download_raw().values,expected.values,run._leaf))
@@ -201,7 +201,7 @@ class ShiftedJetsGPU(unittest.TestCase):
             self.exact_raw(before,old._download_raw())
             self.exact_raw(_StoredPopulations(expected.g,True),new._download_raw())
             self.assertEqual(new._level_jets[2][0].amplitude,-.001)
-            np.testing.assert_array_equal(new._level_boundaries[2].tables['surface_velocity'],old._level_boundaries[2].tables['surface_velocity'])
+            np.testing.assert_array_equal(new._device_boundaries[2].tables['surface_velocity'],old._device_boundaries[2].tables['surface_velocity'])
             newledger=Ledger(self,new,2);ledger.step();newledger.step()
             self.assertEqual((old.step_count,new.step_count),(count+1,count+1))
             ledger.report('L2 source');newledger.report('L2 remapped')

tests/validation/run_refined_dfg.py

@@ -81,7 +81,7 @@ def public_jets(center, radius, scenario):
     )
 
 
-def jet_interval(handle, scenario, begin, n, diameter):
+def jet_interval(handle, scenario, begin, n, diameter, level=None):
     """Apply a reversal only at a completed interval boundary, before advancing."""
     if scenario != "none":
         boundaries = [round(t * diameter / SPEED) for t in (250, 260, 270, 280, 290, 300, 305)]
@@ -92,8 +92,10 @@ def jet_interval(handle, scenario, begin, n, diameter):
     if begin < switch:
         return min(n, switch - begin), None
     if begin == switch:
-        handle.set_jet_amplitude(0, -JET_PEAK)
-        handle.set_jet_amplitude(1, JET_PEAK)
+        # FixedPatchStepper needs an explicit Jet level; Simulation has one level.
+        kw = {} if level is None else {"level": level}
+        handle.set_jet_amplitude(0, -JET_PEAK, **kw)
+        handle.set_jet_amplitude(1, JET_PEAK, **kw)
         return n, dict(step=begin, tstar=300.0, amplitudes=[-JET_PEAK, JET_PEAK])
     return n, None
 
@@ -256,7 +258,7 @@ def host_geometry(source, h, collision="SRT", jet_scenario="none", placement="wi
         validate_channel_config,
         validate_channel_transfers,
     )
-    from CelerisLab.lbm.refinement_tables import blocked_base_tables
+    from CelerisLab.lbm import refinement_tables
     from CelerisLab.body.geometry.circle import CircleGeometry
     from CelerisLab.body.objects import SimObject
 
@@ -268,7 +270,11 @@ def host_geometry(source, h, collision="SRT", jet_scenario="none", placement="wi
     pair = FixedPatch2D.from_config(cfg, PLACEMENTS[placement])
     layout = BlockLayout2D.build(cfg.ny, cfg.nx, *BASE_BLOCK)
     validate_channel_config(cfg, [BASE_BLOCK, FINE_BLOCK])
-    targets = blocked_base_tables(pair, layout, np.argsort(pair.coalesce_coarse))[1]
+    # Historical pinned source (jet_scenario none) names it blocked_base_tables;
+    # the current checkout (public Jet scenarios) names it blocked_coarse_tables.
+    coarse_tables = (refinement_tables.blocked_base_tables if jet_scenario == "none"
+                     else refinement_tables.blocked_coarse_tables)
+    targets = coarse_tables(pair, layout, np.argsort(pair.coalesce_coarse))[1]
     validate_channel_transfers(pair, layout, targets)
     BlockLayout2D.build(*pair.fine_shape, *FINE_BLOCK, periodic=(False, False))
     body = SimObject(
@@ -281,16 +287,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 +309,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 +322,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
 
@@ -441,7 +454,7 @@ def run_case(a, h, periodic, case):
                 )
                 n = min(max(1, round(width * p["diameter"] / SPEED)), max_steps - begin)
                 n, change = jet_interval(backend if refined else sim, a.jet_scenario,
-                                         begin, n, p["diameter"])
+                                         begin, n, p["diameter"], level=1 if refined else None)
                 if change is not None:
                     amplitude_changes.append(change)
                 impulse = np.zeros(2)

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):