test: gate V1 side files and V3 legacy receipts; early output rejection

- analyze_jet_cv_continuation: samples are analyzed only if the sibling
  re*_s*.json receipt is the same case with run_status COMPLETE and
  source_recheck UNCHANGED; receipts without run_status (pre-recheck
  runners) are refused unless --legacy-receipts, and labelled
  LEGACY_NO_FINAL_SOURCE_RECHECK. Results carry receipt_gate.
- V3: receipts are schema 2 and need source_recheck UNCHANGED; a missing
  recheck is accepted only for frozen-mode schema 1 receipts whose
  run_case.py hash is one of the pre-recheck runners (9805383, 2cf9853,
  db3d098; hashes checked against git history in a test).
- V1/V3 refuse --output inside the checkout before source checks or CUDA
  (shared source_provenance.reject_output_inside).
- A failing recheck is recorded as RECHECK_FAILED (still invalid); the
  in-run V1 receipt write is tmp+replace.
Negative host tests for each gate. No threshold, solver or evidence edits.

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

README.md

@@ -112,9 +112,8 @@ from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
 cfg = LBMConfig(nx=32, ny=24, viscosity=.1)
 body = SimObject(0, geometry=CircleGeometry(16.0, 12.0, 2.15), center=(16.0, 12.0), radius=2.15,
                  surface_actuation=SurfaceActuation(0.7, 0.85, 0.002))  # angle, half-width, speed
-with CudaContext(0) as ctx, FixedPatchStepper(cfg, context=ctx, refinements=[(8, 6, 24, 18)],
-                                              level_blocks=[None, (16, 20)],
-                                              level_objects=[(), (body,)]) as run:
+with CudaContext(0) as ctx, FixedPatchStepper(cfg, (8, 6, 24, 18), context=ctx,
+                                              fine_objects=[body], fine_block=(16, 20)) as run:
     rest = np.asarray(compute_feq_d2q9(1.0, 0.0, 0.0), np.float32)[:, None, None]
     # (9, ny, nx) postcollision populations per level; non-leaf cells are discarded.
     run.initialize(np.broadcast_to(rest, (9, *run.patch.coarse_shape)),
@@ -125,8 +124,8 @@ with CudaContext(0) as ctx, FixedPatchStepper(cfg, context=ctx, refinements=[(8,
 
 Scope and limits:
 - The examples below use D2Q9, SRT, FP32, double buffer, unshifted, no LES or
-  forcing. Fixed 2:1 hierarchies use level lists of arbitrary depth; two- and
-  three-level examples below use that same constructor.
+  forcing. Fixed 2:1 hierarchies can use level lists of arbitrary depth; the
+  legacy two/three-level forms are shown below.
 - `coarse_boundary="periodic"` (default) or `"slot_reservoir"` (bottom slot, top
   reservoir; patch needs `y0 >= 4`, `y1 <= ny-4`). Periodic mode ignores the wall/slot config fields.
 - Object centers/radii use the coarse **cell-edge** frame (cell `I` spans `[I, I+1)`),
@@ -134,51 +133,50 @@ Scope and limits:
 - Bodies must be stationary circles whose whole boundary stencil lies in the fine
   active region, clear of the coarse/fine transfer cells; violations are rejected
   before any GPU allocation.
-- `level_blocks=[None, (by, bx)]` stores the fine level in fixed-size blocks; `by`/`bx` must
+- `fine_block=(by, bx)` stores the fine level in fixed-size blocks; `by`/`bx` must
   divide the WHOLE fine allocation, i.e. `2*(y1-y0)+8` by `2*(x1-x0)+8` including the
   four-cell ghost band. It gives results identical to dense storage and is
   infrastructure for future block AMR, not a speedup.
 - Several regions: pass a sequence of `(x0, y0, x1, y1)` rectangles as
-  `refinements[0]`, with both levels blocked (`level_blocks`). Transfers are built once
+  `bounds`, with both `fine_block` and `base_block`. Transfers are built once
   from their union; adjacent, overlapping and concave unions are accepted when
   their actual dependencies close. There is no fixed rectangle-gap rule.
   The fine level is then one level-wide grid `(2*ny, 2*nx)`:
   `run.patch.fine_shape` gives it, `initialize`/`download` use it, and only the
   grid-aligned tiles intersecting the union plus its four-cell ghost band are stored.
   Block sizes must divide the level-wide grid; rectangle edges need not align
-  with tiles. Order of rectangles does not matter. L1 bodies/Jets (`level_objects[1]`/`level_jets[1]`) work
+  with tiles. Order of rectangles does not matter. `fine_objects`/`jets` work
   as for one patch: each circle (same coarse cell-edge frame) must lie wholly
   inside the union's active region with
   its boundary stencil clear of transfer cells; a jet may not select links of two
   different bodies, even across regions. Wall-touching patches remain unsupported.
-  For nesting, add further `refinements` entries as shown below.
+  For nesting, use the level lists or legacy child forms below.
 
   ```python
   cfg = LBMConfig(nx=32, ny=16, viscosity=.1)
   rest = np.asarray(compute_feq_d2q9(1.0, 0.0, 0.0), np.float32)[:, None, None]
   with CudaContext(0) as ctx, FixedPatchStepper(
-          cfg, context=ctx, refinements=[((4, 4, 12, 12), (16, 4, 24, 12))],
-          level_blocks=[(4, 4), (4, 4)]) as run:
+          cfg, ((4, 4, 12, 12), (16, 4, 24, 12)), context=ctx,
+          fine_block=(4, 4), base_block=(4, 4)) as run:
       run.initialize(np.broadcast_to(rest, (9, *run.patch.coarse_shape)),
                      np.broadcast_to(rest, (9, *run.patch.fine_shape)))  # (9, 32, 64)
       run.step(32)
       coarse_g, fine_g = run.download()  # coarse: NaN on every covered cell;
                                          # fine: NaN on every non-ACTIVE cell
   ```
-- Three-level nested windows: pass two scalar rectangles in `refinements` —
-  L1 in L0 cells, L2 in **L1 window cells** (the parent's fine grid including its
-  4-cell ghost band) — plus all three `level_blocks` entries. L2 must sit inside
-  L1's ACTIVE cells, away from L1's ghost band and transfer footprint (checked
-  before any GPU allocation). `initialize` and `download` take/return three
-  arrays; each level is NaN on its non-ACTIVE (and child-covered) cells. One
-  coarse step = 2 L1 steps = 4 L2 steps. One L1 window only in this scalar
-  chain. Optional stationary circles and Jets live on L2 via `level_objects[2]`
-  and `level_jets[2]`; L1 bodies/Jets retain their L1 meaning and must be empty
-  in this mode. All object and Jet geometry uses the **L0 edge frame**, not the
-  child bounds' L1 window frame. L2 has dx=dt=0.25, so prescribed lattice
-  velocity is unchanged. Each whole body and its BFL stencil must be inside L2
-  ACTIVE cells, outside the transfer footprint. A Jet may select one physical
-  body across multiple blocks, never two bodies.
+- Legacy three-level scalar chain: add `child=(x0, y0, x1, y1)` in
+  **L1 window cells** (the patch's fine grid including its 4-cell ghost band) and
+  `child_block`, with `fine_block` and `base_block`. L2 must sit inside L1's
+  ACTIVE cells, away from L1's ghost band and transfer footprint (checked before
+  any GPU allocation). `initialize` and `download` take/return three arrays; each
+  level is NaN on its non-ACTIVE (and child-covered) cells. One coarse step = 2 L1
+  steps = 4 L2 steps. One L1 patch only. Optional stationary circles and Jets
+  live on L2 via `child_objects` and `child_jets`; `fine_objects`/`jets` retain
+  their L1 meaning and must be empty in this mode. All object and Jet geometry
+  uses the **L0 edge frame**, not the child bounds' L1 window frame. L2 has
+  dx=dt=0.25, so prescribed lattice velocity is unchanged. Each whole body and
+  its BFL stencil must be inside L2 ACTIVE cells, outside the transfer footprint.
+  A Jet may select one physical body across multiple blocks, never two bodies.
 
   ```python
   from CelerisLab.body.geometry.circle import CircleGeometry
@@ -190,9 +188,9 @@ Scope and limits:
   jet = Jet(Tube((6.5, 8.), 0., 3., .3), Parallel(0.), amplitude=.002)
   rest = np.asarray(compute_feq_d2q9(1.0, 0.0, 0.0), np.float32)[:, None, None]
   with CudaContext(0) as ctx, FixedPatchStepper(
-          cfg, context=ctx, refinements=[(4, 4, 12, 12), (8, 8, 16, 16)],
-          level_blocks=[(4, 4)] * 3, level_objects=[(), (), (body,)],
-          level_jets=[(), (), (jet,)]) as run:
+          cfg, (4, 4, 12, 12), context=ctx, fine_block=(4, 4), base_block=(4, 4),
+          child=(8, 8, 16, 16), child_block=(4, 4),
+          child_objects=[body], child_jets=[jet]) as run:
       shapes = (run.patch.coarse_shape, run.patch.fine_shape, run.child_patch.fine_shape)
       run.initialize(*(np.broadcast_to(rest, (9, *s)) for s in shapes))  # 16², 24², 24²
       run.step(4)
@@ -200,16 +198,16 @@ Scope and limits:
       run.step(4)
       coarse_g, l1_g, l2_g = run.download()  # solids/covered/nonleaves are NaN
   ```
-- Level lists (the only constructor): `refinements[l]` describes level `l+1`
-  in allocated level-`l` integer cells; use a rectangle list for a union, or a
-  scalar rectangle for a single window (a scalar child requires a scalar
-  parent). `level_blocks`, `level_objects` and `level_jets` each have one entry
-  per level, including L0. Three or more levels require all-blocked storage and
-  bodies/Jets only on the finest level. Every adjacent pair must preserve
-  inherited ACTIVE coverage, valid donors/receivers, surviving pulls and the
-  upstream transfer footprint. Level `l` takes `2**l` steps per base interval.
-  This describes implemented topology and scheduling, not accuracy or
-  convergence qualification.
+- Level lists: `refinements[l]` describes level `l+1` in allocated level-`l`
+  integer cells; use a rectangle list for a union, or a scalar rectangle for a
+  single window (a scalar child requires a scalar parent). `level_blocks`,
+  `level_objects` and `level_jets` each have one entry per level, including L0.
+  These lists are mutually exclusive with the legacy patch keywords. Three or
+  more levels require all-blocked storage and bodies/Jets only on the finest
+  level. Every adjacent pair must preserve inherited ACTIVE coverage, valid
+  donors/receivers, surviving pulls and the upstream transfer footprint.
+  Level `l` takes `2**l` steps per base interval. This describes implemented
+  topology and scheduling, not accuracy or convergence qualification.
 - Blocked body levels can enable `observe_raw_exchange=True` and call
   `read_raw_exchange(body_id, level=...)` after a completed base interval.
   It reduces unique cut-link observations into wall/aperture/total force,

docs/development/root_channel_candidate.md

@@ -1,6 +1,6 @@
 # Blocked AMR root channel increment
 
-Opt in with `FixedPatchStepper(..., coarse_boundary="channel", refinements=[bounds], level_blocks=[(by,bx), (fy,fx)])`. Every level must be blocked. The supported boundary is the existing uniform D2Q9/SRT/FP32 double-buffer channel: local Zou-He west inlet, pressure-NEQ east outlet, stationary halfway bounce-back y walls. Uniform and parabolic inlet profiles and shifted/unshifted storage are supported; the existing configuration guards still reject MRT/LES and incompatible storage. This does not change the external DFG MRT reference.
+Opt in with `FixedPatchStepper(..., coarse_boundary="channel", base_block=(by,bx), fine_block=(fy,fx))`, or the existing level-list form. Every level must be blocked. The supported boundary is the existing uniform D2Q9/SRT/FP32 double-buffer channel: local Zou-He west inlet, pressure-NEQ east outlet, stationary halfway bounce-back y walls. Uniform and parabolic inlet profiles and shifted/unshifted storage are supported; the existing configuration guards still reject MRT/LES and incompatible storage. This does not change the external DFG MRT reference.
 
 `step/block_channel.cu` is a small root-only kernel. Fine kernels, transfer kernels and recursive hop scheduling are unchanged. Global coordinates and the global profile height/velocity are separate from block-local storage strides. Corner ownership and collision phases match `OneStep`:
 

docs/fixed_three_level_multipatch.md

@@ -1,8 +1,8 @@
 # Fixed three-level rectangle unions
 
-This page describes three-level `FixedPatchStepper` setups built from the
-`refinements` level list, with rectangle unions on either adjacent pair. The
-same constructor supports arbitrary depth; three or more levels require
+This page describes the legacy three-level entry point of `FixedPatchStepper`,
+with rectangle unions on either adjacent pair. The `refinements` level-list
+entry point also supports arbitrary depth; three or more levels require
 all-blocked storage and bodies/Jets only on the finest level. This is not
 general AMR: there is no per-region timestep, moving geometry, automatic
 dynamic refinement or L1 body support in the three-level form.
@@ -12,15 +12,15 @@ No CUDA kernel changes are involved.
 
 ## Explicit coordinate frames
 
-For a single scalar parent patch, pass a scalar rectangle as `refinements[1]`:
-`(x0,y0,x1,y1)` in half-open integer indices of the allocated L1 window,
-including its ghost band.
+Keep the existing scalar `child=(x0,y0,x1,y1)` for a single scalar parent patch.
+Its coordinates retain their original meaning: half-open integer indices in the
+allocated L1 window, including its ghost band.
 
-A rectangle list as `refinements[1]` accepts a scalar or a set parent and
-interprets every rectangle in the actual allocated L1 array, including ghost
-bands and holes. It never infers region-local coordinates. The physical root
-cell edge corresponding to an L1 index `(x,y)` is
-`parent.fine_origin + (x,y)/2`:
+The new keyword `child_rectangles_in_parent=[...]` is mutually exclusive with
+`child`. It accepts a scalar or a set parent and interprets every rectangle in
+the actual allocated L1 array, including ghost bands and holes. It never infers
+region-local coordinates. The physical root cell edge corresponding to an L1
+index `(x,y)` is `parent.fine_origin + (x,y)/2`:
 
 - A scalar L1 parent has a shifted window origin.
 - A set L1 parent has origin `(0,0)` and public shape `(2*ny,2*nx)`.
@@ -47,9 +47,9 @@ tile edges.
 
 ```python
 run = FixedPatchStepper(
-    cfg, context=ctx,
-    refinements=[[(4, 4, 20, 28), (24, 4, 44, 28)], [(16, 16, 32, 40), (56, 16, 80, 40)]],
-    level_blocks=[(4, 4)] * 3,
+    cfg, [(4, 4, 20, 28), (24, 4, 44, 28)], context=ctx,
+    base_block=(4, 4), fine_block=(4, 4), child_block=(4, 4),
+    child_rectangles_in_parent=[(16, 16, 32, 40), (56, 16, 80, 40)],
 )
 ```
 
@@ -57,7 +57,7 @@ This example uses a 48x32 root. `initialize` and `download` use three Q-major
 dense arrays. Public arrays and masks still scale with the full allocated
 extent; sparse device buffers do not remove that host-memory limit.
 
-Stationary whole-circle L2 bodies and Jets (`level_objects[2]`, `level_jets[2]`) use root edge
+Stationary whole-circle `child_objects` and `child_jets` use root edge
 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

docs/supported_settings.md

@@ -1,87 +0,0 @@
-# Supported settings: 2D Jet + fixed multilevel refinement (candidate c95a7ba)
-
-Every entry names the guard that enforces it (file:line at c95a7ba) and, where one
-exists, an existing test that exercises it. A setting not listed here is not
-claimed. Observations at the end are tied to the exact case in which they were
-measured; they are not general bounds.
-
-## Lattice, storage, collision
-
-| Setting | Supported | Guard |
-|---|---|---|
-| Dimension / lattice | 2D D2Q9 only (3D refinement is future work) | `lbm/refinement.py:65`, `lbm/kernels/blocks.cu:25`, `lbm/kernels/refinement.cu:26` |
-| Precision / streaming | FP32 storage, double buffer | same guards |
-| Collision on refinement levels | SRT or MRT; TRT and LES rejected | `lbm/refinement.py:65` (re-checked for fine boundaries at `lbm/refinement.py:457`) |
-| Shifted (ddf_shifting) storage | only when every level is blocked; dense or mixed rejected | `lbm/refinement_stepper.py:198`, `:469`; `lbm/kernels/refinement.cu:26` (dense patch kernel unshifted only) |
-| Fine relaxation | 1/tau_fine must lie inside the configured omega guards | `lbm/refinement.py:80` |
-| Uniform blocked backend `PeriodicBlockStepper` | doubly periodic, SRT only, stationary passive circles | `lbm/block_stepper.py:84`, `:91`, `:45`-`:67` |
-
-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`.
-
-## Refinement hierarchy (`FixedPatchStepper(cfg, *, context, refinements, ...)`)
-
-| Setting | Supported | Guard |
-|---|---|---|
-| Depth | at least one refinement level; one entry per level in `level_blocks`/`level_objects`/`level_jets` | `lbm/refinement_stepper.py:72`, `:75` |
-| Single L1 rectangle | two-coarse-cell margin to the domain edge | `lbm/refinement.py:73` |
-| Several L1 rectangles (union) | require `level_blocks[0]` and `level_blocks[1]`; transfers must close (dependency checks, no fixed gap rule) | `lbm/refinement_stepper.py:84`; `lbm/refinement.py:283`-`:295` |
-| Nested levels | depth >= 3 must be all blocked; a blocked level requires blocked levels below; a scalar child needs a scalar parent | `lbm/refinement_stepper.py:82`, `:87`; `lbm/hierarchy.py:54` |
-| Nested transfer support | child only over inherited ACTIVE cells, donors/receivers uncovered ACTIVE, no child over the parent restriction footprint | `lbm/refinement.py:391`-`:407` |
-| Tile shape | positive `(by, bx)`; the level allocation must be divisible by the tile (no partial tiles yet) | `lbm/refinement_stepper.py:78`, `lbm/refinement.py:246` (shape); `lbm/blocks.py:134` (divisibility) |
-| Coarse boundary | `periodic`, `slot_reservoir` or `channel` | `lbm/refinement_stepper.py:177` |
-| `slot_reservoir` | on-site velocity bottom slot, reservoir top, velocity 0, 2 <= slot width < nx, patch rows 4..ny-4 | `lbm/refinement_stepper.py:205`, `:212`; `lbm/refinement.py:546` |
-| `channel` | all levels blocked; Zou-He inlet, pressure-NEQ outlet, stationary bounce-back walls; transfers away from the closure | `lbm/channel.py:22`, `:26`, `:43`, `:54` |
-| Explosion | `constant` (default); `projected_linear` only periodic, all-blocked SRT, no bodies/Jets, no LES (experimental) | `lbm/refinement_stepper.py:188`, `:193` |
-| L2 runtime remap (`remap_child`) | periodic three-level L2 sets only, no L0/L1 bodies, identical pinned boundary | `lbm/refinement_stepper.py:557`, `:567`-`:581` |
-
-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`,
-`test_refinement_blocked_fine.py`, `test_channel_root_gpu.py`, `tests/unit/test_channel_root.py`,
-`tests/unit/test_projected_pair_stepper.py`, `test_runtime_child_remap_gpu.py`, `test_pinned_body_remap_gpu.py`.
-
-## Bodies and Jets on refined levels
-
-| Setting | Supported | Guard |
-|---|---|---|
-| Bodies | stationary circles without sensor/force-region roles, entirely inside the patch union, fully resolved, disjoint, stencil away from the two-hop transfer footprint | `lbm/refinement.py:474`-`:542` |
-| Level | not on L0; only on the finest level of a nested hierarchy | `lbm/refinement_stepper.py:80`, `:89` |
-| Jet condition | velocity only; `pressure` is reserved and raises NotImplementedError | `body/jets.py:148`, `:150` |
-| Jet geometry | each Jet selects resolved links of exactly one stationary body; no Jet/Jet or Jet/SurfaceActuation overlap; abs(amplitude) < 1 lattice | `body/jets.py:155`, `:175`, `:177`, `:255`, `:261`, `:263` |
-| Jets with shifted storage | only in the all-blocked stepper with SRT and no LES | `body/jets.py:249` |
-| SurfaceActuation | unshifted storage only | `body/objects.py:155` |
-| Raw boundary exchange (force) | blocked body levels only; the window ledger requires `observe_raw_exchange` | `lbm/refinement_stepper.py:264`, `:269` |
-| Checkpoint with Jets | refused | `simulation.py:687` |
-
-Raw exchange is the cut-link momentum reaction (wall + aperture), not a validated
-nozzle thrust or stress load. Jet amplitude is a lattice velocity, not a flow rate.
-
-Tests: `tests/unit/test_jets.py`, `test_jets_review.py`, `test_shifted_jets.py`, `test_raw_exchange.py`,
-`tests/validation/multigrid/test_refinement_jets.py`, `test_shifted_jets_gpu.py`, `test_raw_exchange_gpu.py`,
-`test_exchange_window.py`, `test_fine_surface_gpu.py`.
-
-## Observations from specific tested cases (not general bounds)
-
-- SRT channel root vs uniform OneStep, strict-parity test
-  `tests/validation/multigrid/test_channel_root_gpu.py::ChannelRootGPU::test_uniform_matched_root`:
-  in one of its 18 sub-cases (2 storage modes x 3 profile/clamp/damp settings x 3 blocks; which
-  one is not recorded), 1 of the 4320 compared values differs, max abs 2.33e-10, max rel 1.77e-7
-  (about 2 ULP). The test remains failed; its cause and any general bound are unresolved.
-- MRT root parity test
-  `tests/validation/multigrid/test_refinement_mrt_gpu.py::RefinementMRTGPU::test_uniform_matched_mrt_root`
-  (12 sub-cases: 2 storage modes x 2 profiles x 3 blocks, default outlet settings) passed bitwise
-  in the Horsea focused GPU run at K1 7813334 (2026-10-21).
-- Periodic Taylor-Green Re 100, measured at 36ffc8b1 (V3 matched pair, N128 depth 2):
-  every per-sample screen passed; the constant explosion's aggregate worst velocity L2,
-  0.004418, exceeds its preregistered 0.002 budget, while projected_linear passed that budget.
-- Curved-wall pressure (pilot harness, exact-q analytic annulus, tau 0.8): the coarse swirl r1
-  case exceeded its terminal density-gain cap (1.105e-4 vs 1e-4), so the frozen 9-case matrix
-  is NOT_ESTABLISHED. No validated fix exists. This is a pilot harness, not production code.
-- DFG 2D-1 Re 20 (V4, SRT, t* 60, pressure-NEQ outlet and local Zou-He inlet): refined and uniform
-  runs were NOT_STEADY under the frozen steady rule, so no interval verdict was given.
-
-## Not supported in this release
-
-3D, dynamic/automatic AMR (tagging, time-varying hierarchies beyond `remap_child`),
-moving bodies on refined levels, pressure Jets, Jet checkpoints, LES or TRT on refined
-levels, partial (non-dividing) tiles.

docs/validation/dfg_re20.md

@@ -46,4 +46,9 @@ CUDA_VISIBLE_DEVICES=<n> python tests/validation/run_dfg_re20.py --case both \
 
 The uniform D80 control separates method error from refinement error. `--plan` writes `plan.json` only. A GPU run writes `plan.json` and `result.json`
 (per case: lattice parameters, every sample, window ranges, status, verdict).
+Receipts: `plan.json` and `result.json` carry `schema` 2 and `provenance` (HEAD, src tree,
+runner hash; clean checkout required, refused if it changes during the run). Receipts
+written before this (schema 1, e.g. evidence 70f6617) are a bare list of cases. Read both
+with `run_dfg_re20.read_result(path)` (`["cases"][0]` replaces the old `[0]`); existing
+evidence scripts stay valid for schema-1 receipts and need this reader for new ones.
 Host contract tests: `tests/unit/test_dfg_re20.py`.

examples/fixed_patch_circle.py

@@ -90,8 +90,8 @@ def main():
     # 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:
+            cfg, BOUNDS, context=context, coarse_boundary=args.mode,
+            fine_objects=[body], fine_block=fine_block) 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.

examples/fixed_patch_jet.py

@@ -148,8 +148,8 @@ def main():
                   samples=[], status='running', last_synchronized_coarse_step=0)
     start, run = time.monotonic(), None
     try:
-        with CudaContext(0) as context, FixedPatchStepper(cfg, context=context,
-                coarse_boundary='slot_reservoir', refinements=[BOUNDS]) as run:
+        with CudaContext(0) as context, FixedPatchStepper(cfg, BOUNDS, context=context,
+                coarse_boundary='slot_reservoir') 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

@@ -48,7 +48,7 @@ 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, PeriodicPatchStepper(cfg, (4, 4, 12, 12), context=context) 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/jets.py

@@ -207,30 +207,19 @@ def surface_link_velocity(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry,
                           jets=(), hits=None, *, _shifted_blocked=False):
     """Per-link prescribed surface velocity (n, 2) FP32 and aperture mask.
 
-    See _surface_links, which also returns the unit-amplitude jet shapes.
-    """
-    return _surface_links(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry, jets, hits,
-                          _shifted_blocked=_shifted_blocked)[:2]
-
-
-def _surface_links(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry,
-                   jets=(), hits=None, *, _shifted_blocked=False):
-    """(velocity, aperture, jet shape, jet id); the last two are None without jets.
-
     Host-only and storage-neutral: rows follow the given physical cut-link
-    order. Returns all None when neither an object carries surface
+    order. Returns (None, None) when neither an object carries surface
     actuation nor a jet is given. Jets need true FP64 ``hits`` (n, 2) and may
     not share a link with each other or with a legacy aperture.
     Shifted storage is privately enabled only by the all-blocked stepper.
     """
     enabled = [obj for obj in objects if obj.surface_actuation is not None]
     if not enabled and not jets:
-        return None, None, None, None
+        return None, None
     from .preprocess.flag_overlay import validate_cut_link_ownership
     validate_cut_link_ownership(flags, fluid_idx, dir, cfg.nq)
     surface_velocity = np.zeros((len(dir), 2), np.float32)
     surface_aperture = np.zeros(len(dir), bool)
-    shape = jet_id = None
     for obj in enabled:
         obj.validate_surface_actuation(cfg)
         selected = body_id == obj.obj_id
@@ -263,4 +252,4 @@ def _surface_links(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry,
             raise ValueError("Jet overlaps a SurfaceActuation aperture")
         surface_velocity[driven] = jet_link_velocity(jets, shape, jet_id)[driven]
         surface_aperture |= driven
-    return surface_velocity, surface_aperture, shape, jet_id
+    return surface_velocity, surface_aperture

src/CelerisLab/cuda/compiler_v2.py

@@ -54,18 +54,14 @@ def _write(path: str, content: str):
 
 
 def generate_config(cfg: LBMConfig, n_objects: int = 0,
-                    *, build_root: Optional[str] = None, block_root_bc: int = 0) -> str:
+                    *, build_root: Optional[str] = None) -> str:
     """Snapshot kernels and render headers in a fresh private directory.
 
     Returns its path. Pass it explicitly to compile_kernel for interleaved
     builds. Legacy generate_config(); compile_kernel() remains context-local.
     The caller owns cleanup_build_dir() once the loaded module is no longer
     needed for recompilation/debugging. No package files are changed.
-    block_root_bc specializes a blocked root module; fine/dense builds use 0.
     """
-    if (isinstance(block_root_bc, bool) or not isinstance(block_root_bc, int)
-            or block_root_bc not in (0, 1, 2)):
-        raise ValueError("block_root_bc must be an integer mode 0, 1 or 2")
     macros = cfg.to_macros()
     if isinstance(n_objects, bool) or not isinstance(n_objects, int) or n_objects < 0:
         raise ValueError("n_objects must be a nonnegative integer")
@@ -75,15 +71,14 @@ def generate_config(cfg: LBMConfig, n_objects: int = 0,
     try:
         shutil.copytree(_KERNEL_DIR, build_dir, dirs_exist_ok=True,
                         ignore=shutil.ignore_patterns("*.ptx", "*.cubin", "*.tmp"))
-        _render_config(macros, n_objects, os.path.join(build_dir, "config"),
-                       block_root_bc=block_root_bc)
+        _render_config(macros, n_objects, os.path.join(build_dir, "config"))
         digest = hashlib.sha256()
         for path in sorted(Path(build_dir).rglob("*")):
             if path.is_file():
                 digest.update(str(path.relative_to(build_dir)).encode())
                 digest.update(path.read_bytes())
         metadata = {"source_sha256": digest.hexdigest(), "config": asdict(cfg),
-                    "n_objects": n_objects, "block_root_bc": block_root_bc}
+                    "n_objects": n_objects}
         _write(os.path.join(build_dir, ".celeris-build"), json.dumps(metadata, sort_keys=True))
     except BaseException:
         shutil.rmtree(build_dir)
@@ -106,7 +101,7 @@ def cleanup_build_dir(build_dir: Optional[str]) -> None:
         _last_ptx_path.set(None)
 
 
-def _render_config(m: dict, n_objects: int, cdir: str, *, block_root_bc: int = 0):
+def _render_config(m: dict, n_objects: int, cdir: str):
     _write(os.path.join(cdir, "config_grid.h"), f"""{_HEADER}\
 // Layer: Global / Grid
 #ifndef CELERIS_CONFIG_GRID_H
@@ -170,9 +165,6 @@ def _render_config(m: dict, n_objects: int, cdir: str, *, block_root_bc: int = 0
 #define STORE_PRECISION  {m['STORE_PRECISION']}
 #define USE_DDF_SHIFTING {m['USE_DDF_SHIFTING']}
 
-// Internal blocked root mode: 0=periodic/fine, 1=slot/reservoir, 2=channel.
-#define BLOCK_ROOT_BC {block_root_bc}
-
 #define USE_LES {m['USE_LES']}
 #define LES_CS  {m['LES_CS']}
 #define LES_CLOSED_FORM {m['LES_CLOSED_FORM']}

src/CelerisLab/lbm/block_stepper.py

@@ -204,21 +204,11 @@ class PeriodicBlockStepper:
                 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.
-        """
+        """Advance n steps: halo refresh, then core pull/collide, per step."""
         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"
@@ -234,9 +224,9 @@ class PeriodicBlockStepper:
                     self._step(self._current, self._spare, np.uint32(self.layout.halo),
                                self._tables.get("flags", np.uintp(0)), np.uintp(0),  # no kind
                                grid=self._grid, **kw)
-                    self._current, self._spare = self._spare, self._current  # host handles only
-                self._stream.synchronize()
-                self._step_count += n
+                    self._stream.synchronize()
+                    self._current, self._spare = self._spare, self._current
+                    self._step_count += 1
                 self._status = "ready"
             except BaseException:
                 self._status = "failed"

src/CelerisLab/lbm/curved_links.py

@@ -94,10 +94,12 @@ class CurvedLinkSoA:
         """
         self.surface_velocity = self.aperture = None
         self.jets, self.jet_shape, self.jet_id = tuple(jets), None, None
-        from ..body.jets import _surface_links
-        self.surface_velocity, self.aperture, self.jet_shape, self.jet_id = _surface_links(
+        self.surface_velocity, self.aperture = surface_link_velocity(
             objects, cfg, flags, self.fluid_idx, self.dir, self.body_id, self.rx, self.ry,
             jets=self.jets, hits=hits)
+        if self.jets:
+            from ..body.jets import jet_link_shapes
+            self.jet_shape, self.jet_id = jet_link_shapes(self.jets, hits)
 
     def set_jet_amplitude(self, index, amplitude, stream=None):
         """Rewrite jet rows from cached shapes; re-upload only the velocity column.

src/CelerisLab/lbm/kernels/blocks.cu

@@ -1,6 +1,5 @@
 // CelerisLab/lbm/kernels/blocks.cu
-// One fixed-size blocked level (lbm/blocks.py layout), specialized at compile
-// time for periodic/fine, slot/reservoir or channel stepping. Separate
+// One periodic level of fixed-size blocks (lbm/blocks.py layout). Separate
 // compilation entry: NX, NY are the block window including halo, so index_f
 // and linear_index address one block; blockIdx.z selects the block, whose
 // populations start at b*NQ*TOTAL_CELLS (block-contiguous, Q-major inside).
@@ -20,18 +19,7 @@
 #endif
 #include "streaming/pull_double_buffer.cuh"
 #include "boundary/curved_boundary.cuh"
-#ifndef BLOCK_ROOT_BC
-#error "Blocks require a generated BLOCK_ROOT_BC mode"
-#elif BLOCK_ROOT_BC < 0 || BLOCK_ROOT_BC > 2
-#error "BLOCK_ROOT_BC must be 0, 1 or 2"
-#endif
-
-#if BLOCK_ROOT_BC == 1
-#include "boundary/block_wall.cuh"
-#endif
-#if BLOCK_ROOT_BC != 2
-#include "operators/block_collision.cuh"
-#endif
+#include "boundary/planar_velocity.cuh"
 
 #if DIM != 2 || NQ != 9 || (COLLISION_MODEL != 0 && COLLISION_MODEL != 2) || STORE_PRECISION != 0 || USE_LES || STREAMING_MODEL != 0
 #error "Blocks require FP32 D2Q9 SRT or MRT double buffers without LES"
@@ -40,6 +28,11 @@
 // Fixed-patch cell kinds, matching refinement.py ACTIVE/INNER_GHOST/OUTER_GHOST.
 constexpr unsigned char KIND_ACTIVE = 0u, KIND_INNER = 1u, KIND_OUTER = 2u;
 
+// Zero-initialized for ordinary blocks/fine levels. Set only for parent walls;
+// NX/NY remain local storage strides, never physical wall coordinates.
+struct BlockWall { unsigned int nx, ny, slot_width; float amplitude; };
+extern "C" { __constant__ BlockWall d_block_wall; }
+
 // Halo refresh: block cell ids b*TOTAL_CELLS+c, sources are always core cells.
 extern "C" __global__ void BlockHaloCopy(
     fpxx* f, const uint64_t* dst, const uint64_t* src, unsigned int count)
@@ -154,7 +147,6 @@ extern "C" __global__ void BlockAccumulateExchange(
     integral[k + 3] += h * h * (double)raw[k + 3];
 }
 
-#if BLOCK_ROOT_BC != 2
 // Pull/collide/store core cells only; halo cells of out are left untouched.
 // flags (block cell ids, nullable): solid cores store NaN and never collide.
 // kind (block cell ids, nullable = all ACTIVE): OUTER skips the pull and
@@ -175,18 +167,40 @@ extern "C" __global__ void BlockStep(const fpxx* in, fpxx* out, unsigned int hal
         compute_neighbors(x, y, j);  // Core neighbors stay inside the window: no wrap.
         stream_pull_load(k, f, in + base, j);
         if (cell_kind == KIND_ACTIVE) {
-#if BLOCK_ROOT_BC == 1
-            block_wall(f, x, y, halo);
+            if (d_block_wall.nx != 0) {
+                const unsigned int bx = NX - 2 * halo, by = NY - 2 * halo;
+                const unsigned int nbx = d_block_wall.nx / bx;
+                const unsigned int gx = (blockIdx.z % nbx) * bx + x - halo;
+                const unsigned int gy = (blockIdx.z / nbx) * by + y - halo;
+                if (gy == 1)
+                    reconstruct_bottom_velocity(f, 0.0f, bottom_wall_normal_velocity(
+                        gx, d_block_wall.nx, d_block_wall.slot_width, d_block_wall.amplitude));
+                else if (gy == d_block_wall.ny - 2) reconstruct_top_reservoir(f, RHO);
+            }
+#if COLLISION_MODEL == 2
+            // Uniform MRT sequence (collide_dispatch, zero forcing); SRT path below unchanged.
+            float rho, ux, uy;
+            compute_rho_u(f, rho, ux, uy);
+            collide_mrt_no_force(f, rho, ux, uy,
+                fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega)));
+#else
+            float rho, ux, uy, feq[NQ];
+            compute_rho_u(f, rho, ux, uy);
+#if USE_DDF_SHIFTING
+            // Preserve the density perturbation before adding the unit reference.
+            const float drho = f[0]+f[1]+f[2]+f[3]+f[4]+f[5]+f[6]+f[7]+f[8];
+            const float jx = f[1]-f[2]+f[5]-f[6]+f[7]-f[8];
+            const float jy = f[3]-f[4]+f[5]-f[6]-f[7]+f[8];
+            compute_feq_from_momentum(drho, jx, jy, ux, uy, feq);
+#else
+            compute_feq(rho, ux, uy, feq);
+#endif
+            collide_srt_no_force(f, feq, d_params.omega);
 #endif
-            block_collide(f);
         }
     }
     stream_pull_store(k, f, out + base);
 }
 
-#endif
-
-// Root channel is a replacement full step, selected once by the host.
-#if BLOCK_ROOT_BC == 2
+// Root channel closure; fine and periodic levels keep BlockStep.
 #include "step/block_channel.cu"
-#endif

src/CelerisLab/lbm/kernels/boundary/block_wall.cuh

@@ -1,20 +0,0 @@
-// Physical parent walls; NX/NY remain local storage strides.
-#ifndef CELERIS_BOUNDARY_BLOCK_WALL_CUH
-#define CELERIS_BOUNDARY_BLOCK_WALL_CUH
-
-#include "planar_velocity.cuh"
-
-struct BlockWall { unsigned int nx, ny, slot_width; float amplitude; };
-extern "C" { __constant__ BlockWall d_block_wall; }
-
-__device__ __forceinline__ void block_wall(float* f, unsigned int x, unsigned int y,
-                                         unsigned int halo) {
-    unsigned int gx, gy;
-    block_global_xy(x, y, halo, d_block_wall.nx, gx, gy);
-    if (gy == 1)
-        reconstruct_bottom_velocity(f, 0.0f, bottom_wall_normal_velocity(
-            gx, d_block_wall.nx, d_block_wall.slot_width, d_block_wall.amplitude));
-    else if (gy == d_block_wall.ny - 2) reconstruct_top_reservoir(f, RHO);
-}
-
-#endif

src/CelerisLab/lbm/kernels/config/config_method.h

@@ -8,9 +8,6 @@
 #define STORE_PRECISION  0
 #define USE_DDF_SHIFTING 0
 
-// Internal blocked root mode: 0=periodic/fine, 1=slot/reservoir, 2=channel.
-#define BLOCK_ROOT_BC 0
-
 #define USE_LES 0
 #define LES_CS  0.160000f
 #define LES_CLOSED_FORM 1

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

@@ -53,18 +53,6 @@ __device__ __forceinline__ void index_from_thread(unsigned int& x,
     k = (unsigned long)y * NX + x;
 }
 
-// Global cell of window cell (x, y) in block blockIdx.z of a row-major,
-// full-domain block layout domain_nx cells wide. NX/NY are the local strides.
-__device__ __forceinline__ void block_global_xy(unsigned int x, unsigned int y,
-                                                unsigned int halo, unsigned int domain_nx,
-                                                unsigned int& gx, unsigned int& gy)
-{
-    const unsigned int bx = NX - 2 * halo, by = NY - 2 * halo;
-    const unsigned int nbx = domain_nx / bx;
-    gx = (blockIdx.z % nbx) * bx + x - halo;
-    gy = (blockIdx.z / nbx) * by + y - halo;
-}
-
 // ---------------------------------------------------------------------------
 // Neighbor list (periodic BCs via modular arithmetic)
 // ---------------------------------------------------------------------------

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

@@ -1,24 +0,0 @@
-// Collision used by periodic/fine and slot/reservoir BlockStep.
-#ifndef CELERIS_OPERATORS_BLOCK_COLLISION_CUH
-#define CELERIS_OPERATORS_BLOCK_COLLISION_CUH
-
-__device__ __forceinline__ void block_collide(float* f) {
-#if COLLISION_MODEL == 2
-    collide_mrt_unforced(f);  // SRT path below unchanged.
-#else
-    float rho, ux, uy, feq[NQ];
-    compute_rho_u(f, rho, ux, uy);
-#if USE_DDF_SHIFTING
-    // Preserve the density perturbation before adding the unit reference.
-    const float drho = f[0]+f[1]+f[2]+f[3]+f[4]+f[5]+f[6]+f[7]+f[8];
-    const float jx = f[1]-f[2]+f[5]-f[6]+f[7]-f[8];
-    const float jy = f[3]-f[4]+f[5]-f[6]-f[7]+f[8];
-    compute_feq_from_momentum(drho, jx, jy, ux, uy, feq);
-#else
-    compute_feq(rho, ux, uy, feq);
-#endif
-    collide_srt_no_force(f, feq, d_params.omega);
-#endif
-}
-
-#endif

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

@@ -114,16 +114,6 @@ __device__ __forceinline__ void collide_mrt_no_force(float* __restrict__ g,
     collide_mrt(g, rho, ux, uy, Fin, omega);
 }
 
-// Unforced MRT with the clamped device omega, as in collide_dispatch with zero
-// forcing. Shared by the blocked, channel-root and dense-patch collisions.
-__device__ __forceinline__ void collide_mrt_unforced(float* __restrict__ g)
-{
-    float rho, ux, uy;
-    compute_rho_u(g, rho, ux, uy);
-    collide_mrt_no_force(g, rho, ux, uy,
-        fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega)));
-}
-
 #elif LATTICE_MODEL == LATTICE_D3Q19
 
 __device__ __forceinline__ void collide_mrt(float* __restrict__ g,

src/CelerisLab/lbm/kernels/refinement.cu

@@ -37,7 +37,10 @@ __device__ unsigned int incoming_counts[NQ * TOTAL_CELLS];
 
 __device__ __forceinline__ void patch_collide(float* f, unsigned long k) {
 #if COLLISION_MODEL == 2
-    collide_mrt_unforced(f);
+    float rho, ux, uy;
+    compute_rho_u(f, rho, ux, uy);
+    collide_mrt_no_force(f, rho, ux, uy,
+        fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega)));
 #else
     float rho, ux, uy, feq[NQ];
     compute_rho_u(f, rho, ux, uy);

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

@@ -1,7 +1,3 @@
-#if BLOCK_ROOT_BC != 2
-#error "BlockChannelStep requires BLOCK_ROOT_BC == 2"
-#endif
-
 // Root-only blocked counterpart of OneStep's D2Q9/SRT channel closure.
 // NX/NY are LOCAL strides; all physical boundary decisions use global coordinates.
 #include "../boundary/inlet/common.cuh"
@@ -19,8 +15,10 @@ extern "C" __global__ void BlockChannelStep(
     unsigned int x, y; unsigned long k;
     index_from_thread(x, y, k);
     if (x < halo || y < halo || x >= NX - halo || y >= NY - halo) return;
-    unsigned int gx, gy;
-    block_global_xy(x, y, halo, d_block_channel.nx, gx, gy);
+    const unsigned int bx = NX - 2 * halo, by = NY - 2 * halo;
+    const unsigned int nbx = d_block_channel.nx / bx;
+    const unsigned int gx = (blockIdx.z % nbx) * bx + x - halo;
+    const unsigned int gy = (blockIdx.z / nbx) * by + y - halo;
     const unsigned long base = (unsigned long)blockIdx.z * NQ * TOTAL_CELLS;
     const unsigned long cell = (unsigned long)blockIdx.z * TOTAL_CELLS + k;
     float f[NQ];
@@ -57,7 +55,11 @@ extern "C" __global__ void BlockChannelStep(
         }
         if (!wall && !outlet) {
 #if COLLISION_MODEL == 2
-            collide_mrt_unforced(f);
+            // Match uniform MRT arithmetic (collide_dispatch, zero forcing).
+            float rho, ux, uy;
+            compute_rho_u(f, rho, ux, uy);
+            collide_mrt_no_force(f, rho, ux, uy,
+                fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega)));
 #else
             // Match uniform SRT arithmetic, including its shifted equilibrium path.
             float rho, ux, uy, feq[NQ], forcing[NQ] = {};

src/CelerisLab/lbm/refinement_stepper.py

@@ -65,8 +65,52 @@ def _level_name(level):
     return ("base", "fine", "child")[level] if level < 3 else f"level {level}"
 
 
+def _legacy_levels(bounds, *, fine_block, base_block, child, child_block, child_rectangles_in_parent,
+                   fine_objects, jets, child_objects, child_jets):
+    """Normalise the original two/three-level keywords to per-level lists (original errors)."""
+    if child is not None and child_rectangles_in_parent is not None:
+        raise ValueError("child and child_rectangles_in_parent are mutually exclusive")
+    nested = child is not None or child_rectangles_in_parent is not None
+    bounds = tuple(bounds)
+    if not is_rectangle(bounds) and (fine_block is None or base_block is None):
+        raise ValueError("several patches require fine_block and base_block")
+    fine_objects, jets = tuple(fine_objects), tuple(jets)
+    child_objects, child_jets = tuple(child_objects), tuple(child_jets)
+    if not nested and (child_objects or child_jets):
+        raise ValueError("child_objects/child_jets require a nested child")
+    if base_block is not None and fine_block is None:
+        raise ValueError("base_block requires fine_block")
+    refinements, blocks = [bounds], [base_block, fine_block]
+    objects, level_jets = [(), fine_objects], [(), jets]
+    if nested:
+        if child is not None and not is_rectangle(bounds):
+            raise ValueError("scalar child requires one parent patch; use child_rectangles_in_parent")
+        if fine_block is None or base_block is None or child_block is None:
+            raise ValueError("a nested child needs fine_block, base_block and child_block")
+        if fine_objects or jets:
+            raise ValueError("a nested child supports bodies/Jets only on L2; use child_objects/child_jets")
+        refinements.append(tuple(child) if child is not None else tuple(child_rectangles_in_parent))
+        blocks.append(child_block)
+        objects.append(child_objects)
+        level_jets.append(child_jets)
+    return refinements, blocks, objects, level_jets
+
+
+def _initial_levels(level_g, coarse_g, fine_g, child_g):
+    """One tuple of per-level arrays from initialize()'s positional or legacy named form."""
+    if coarse_g is None and fine_g is None:
+        if child_g is not None and len(level_g) != 2:
+            raise TypeError("child_g needs exactly the coarse and fine arrays before it")
+        return tuple(level_g) + (() if child_g is None else (child_g,))
+    if level_g:
+        raise TypeError("pass level arrays positionally or as coarse_g/fine_g/child_g, not both")
+    if coarse_g is None or fine_g is None:
+        raise TypeError("coarse_g and fine_g are both required in the named form")
+    return (coarse_g, fine_g) + (() if child_g is None else (child_g,))
+
+
 def _check_levels(refinements, blocks, objects, jets):
-    """Candidate limits of the level lists, checked before any device work."""
+    """Candidate limits shared by the list and the legacy entry points."""
     depth = len(refinements) + 1
     if depth < 2:
         raise ValueError("at least one refinement level is required")
@@ -75,13 +119,13 @@ def _check_levels(refinements, blocks, objects, jets):
             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)")
+            raise ValueError(f"{_level_name(level)}_block must be (by, bx)")
     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]")
+        raise ValueError("several patches require fine_block and base_block")
     if depth > 2:
         if any(block is None for block in blocks):
             raise ValueError("three or more levels must all be blocked")
@@ -90,14 +134,7 @@ def _check_levels(refinements, blocks, objects, jets):
 
 
 class FixedPatchStepper:
-    """D2Q9 SRT/MRT FP32 fixed 2:1 refinement hierarchy, built from level lists.
-
-    ``refinements[l]`` gives level l+1 in level-l integer cells: one rectangle
-    (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).
+    """D2Q9/SRT/FP32, one unforced internal 2:1 patch in an x-periodic box.
 
     ``coarse_boundary`` is ``periodic`` (default) or ``slot_reservoir``:
     a bottom on-site slot and top fixed-density reservoir from ``cfg``.
@@ -108,55 +145,68 @@ class FixedPatchStepper:
     actual channel closure footprint. Channel remapping is not supported.
     Periodic mode does not apply the wall/slot boundary fields in ``cfg``.
     Outer walls remain coarse; fine halo donors must be ordinary fluid.
+    ``fine_objects`` accepts fixed circles in coarse physical coordinates;
+    velocities are in coarse 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 (coarse centers I+.5), whereas ordinary Simulation objects use the integer-node frame.
 
-    Bodies 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
-    (coarse centers I+.5), whereas ordinary Simulation objects use the
-    integer-node frame. Jets (body.jets.Jet) use the same frame and drive cut
-    links selected at their true hits; ``set_jet_amplitude(index, amplitude,
-    level=l)`` re-uploads the per-link velocity column only.
-
-    ``initialize(*level_g)`` accepts one physical Q-major (9, ny, nx) array per
-    level, already in postcollision phase. It does not collide or build channel
-    flags. Every ``step()`` publishes only after all levels complete on one
-    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 blocks.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
-    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
-    staging and pull/collision; covered cells are poisoned on every step.
+    ``initialize(coarse_g, fine_g)`` accepts physical Q-major (9,ny,nx) arrays
+    already in postcollision phase. It does not collide or build channel flags.
+    Every ``step()`` publishes only after both levels complete on one stream.
+    On a device/launch failure the instance becomes unusable: close and recreate
+    it, rather than pretending the partially advanced fine buffer rolled back.
 
+    ``fine_block=(by, bx)`` (opt-in) stores the fine level in
+    blocks.cu storage: the ENTIRE fine_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 fine hop. Public
+    arrays keep the dense (9, ny, nx) fine shape; only the allocation differs.
+    Blocked fine_objects are stationary circles, optionally surface-actuated:
+    prepare_fine_boundary decides legality on the dense fine grid, then cut
+    links are staged receiver-local per hop: halo -> BFL -> step, same buffer.
+    ``base_block=(by, bx)`` also blocks the parent; it requires ``fine_block``.
+    Parent halos precede receiver-local coalescence staging and pull/collision;
+    covered cells are explicitly poisoned on every step. Public shapes stay dense.
     ``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.
+    body layouts are not supported by this new diagnostic in this candidate.
     ``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.
+    ``jets`` (body.jets.Jet) use the same edge frame as ``fine_objects`` and
+    drive fine cut links selected at their true hits; ``set_jet_amplitude``
+    re-uploads the per-link velocity column only.
 
-    Nested levels: a child set's coordinates index the allocated parent array,
-    including ghost bands and holes: root edge = parent fine_origin + index / 2
-    (zero origin for a parent set; a scalar parent keeps its shifted window
-    origin). A scalar child requires a scalar parent. Rectangle unions impose no
-    fixed gap rule: adjacent, overlapping and concave unions are accepted when
-    their actual transfer dependencies close (FixedPatchSet2D /
-    nest_patch_set). All regions share one buffer pair per level and the fixed
-    schedule of 2**l hops per base step at level l. Bodies/Jets are supported only on the
-    finest level of a nested hierarchy (L2 dx=dt=.25 in root units, unchanged
-    lattice velocity); every whole circle and its BFL stencil must fit strictly
-    away from that level's transfer footprint.
+    Several patches: pass ``bounds`` as a sequence of (x0, y0, x1, y1)
+    rectangles (FixedPatchSet2D, dependency-validated union). This requires
+    fine_block and base_block; fine_objects/jets are projected once onto the
+    level-wide fine grid (each circle inside the union, body ids = object
+    positions, jet and writer checks level-wide), and
+    uses one level-wide fine grid (2*ny, 2*nx): public fine arrays have that
+    shape, NaN outside the patches' ACTIVE cells. The step sequence is the
+    blocked one above, with unique union transfers in one launch each.
+
+    Three-level separated unions: ``child_rectangles_in_parent=[...]`` is
+    mutually exclusive with scalar ``child`` and accepts either parent form.
+    Coordinates index the allocated L1 array, including ghost bands and holes:
+    root edge = patch.fine_origin + L1 index / 2. A parent set has origin zero;
+    a scalar parent retains its shifted window origin. The child set's public
+    extent is twice the allocated L1 shape, even though storage is sparse.
+    Each pair requires gap >= 4 in its own parent-cell units. All regions share
+    one buffer pair per level and the fixed 1:2:4 schedule; L2-only bodies/Jets
+    below apply unchanged, with object IDs assigned over the whole level.
+
+    Three-level single chain: ``child`` bounds use L1 window cells; all three
+    levels must be blocked. ``child_objects`` and ``child_jets`` use the L0
+    edge frame, with L2 dx=dt=.25 and unchanged lattice velocity. L1 bodies
+    and Jets remain unsupported in a chain. Every whole circle and its BFL
+    stencil must fit strictly away from the L2 transfer footprint. L2 runs
+    halo -> BFL -> step four times per base step. Runtime child Jet changes
+    require ``set_jet_amplitude(index, amplitude, level=2)``.
 
     ``explosion="projected_linear"`` selects experimental transverse linear
     reconstruction on each adjacent pair. Default "constant" is unchanged.
@@ -164,21 +214,43 @@ class FixedPatchStepper:
     bodies/Jets or LES; this backend supplies no forcing. Geometry is generic,
     but union/multilevel GPU and matched-physics qualification remain pending.
 
-    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
-    restriction, applies BFL and steps.
+    Level lists (any depth): ``refinements[l]`` gives level l+1 in level-l
+    integer cells (scalar window or rectangle list, see lbm.hierarchy);
+    ``level_blocks``/``level_objects``/``level_jets`` hold one entry per level,
+    L0 included. They are mutually exclusive with the legacy keywords above,
+    which are normalised to the same lists. 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 restriction, applies BFL and
+    steps. Depth >= 3 must be fully blocked with bodies/Jets on the finest level.
     """
-    def __init__(self, cfg, *, context, refinements, coarse_boundary="periodic",
-                 level_blocks=None, level_objects=None, level_jets=None,
+    def __init__(self, cfg, bounds=None, *, context, coarse_boundary="periodic", fine_objects=(),
+                 fine_block=None, base_block=None, jets=(), child=None, child_block=None,
+                 child_objects=(), child_jets=(), child_rectangles_in_parent=None,
+                 refinements=None, 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")
         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)
-        level_objects = [()] * depth if level_objects is None else list(level_objects)
-        level_jets = [()] * depth if level_jets is None else list(level_jets)
+        if refinements is None:
+            if bounds is None:
+                raise ValueError("bounds or refinements is required")
+            if level_blocks is not None or level_objects is not None or level_jets is not None:
+                raise ValueError("level_blocks/level_objects/level_jets require refinements")
+            refinements, level_blocks, level_objects, level_jets = _legacy_levels(
+                bounds, fine_block=fine_block, base_block=base_block, child=child,
+                child_block=child_block, child_rectangles_in_parent=child_rectangles_in_parent,
+                fine_objects=fine_objects, jets=jets, child_objects=child_objects,
+                child_jets=child_jets)
+        else:
+            if (bounds is not None or child is not None or child_rectangles_in_parent is not None
+                    or fine_block is not None or base_block is not None or child_block is not None
+                    or tuple(fine_objects) or tuple(jets) or tuple(child_objects) or tuple(child_jets)):
+                raise ValueError("refinements and the legacy patch keywords are mutually exclusive")
+            depth = len(refinements) + 1
+            level_blocks = [None] * depth if level_blocks is None else 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]
@@ -362,10 +434,7 @@ class FixedPatchStepper:
             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
-        build = compiler.generate_config(config, n_objects=n_objects, block_root_bc=block_root_bc)
+        build = compiler.generate_config(config, n_objects=n_objects)
         try:
             ptx = compiler.compile_kernel(self.ctx.sm_arch, source=source, build_dir=build)
             module = compiler.load_module(ptx)
@@ -471,13 +540,13 @@ class FixedPatchStepper:
         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):
+    def initialize(self, *level_g, coarse_g=None, fine_g=None, child_g=None):
         """Upload physical postcollision G, encoding before any FP32 rounding.
 
         Shifted input should be physical FP64 to retain small perturbations.
         Public initialization resets the step count; non-leaves are discarded.
         """
-        given = tuple(level_g)
+        given = _initial_levels(level_g, coarse_g, fine_g, child_g)
         with self._lock, self.ctx.activate():
             self._require("new", "ready")
             self._check_storage()
@@ -601,9 +670,10 @@ class FixedPatchStepper:
                 old, new, source, base_count=count, level_steps=(count, 2*count, 4*count),
                 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],
+                cfg, pairs[0].bounds, context=self.ctx, base_block=base_block,
+                fine_block=fine_block, child_block=child_block,
+                child_rectangles_in_parent=new.child.bounds,
+                child_objects=child_objects, child_jets=jets,
                 observe_raw_exchange=self._observe_raw_exchange,
                 observe_window_exchange=self._observe_window, explosion=self._explosion)
             try:
@@ -752,41 +822,27 @@ class FixedPatchStepper:
                             grid=((pairs + self._block[0] - 1) // self._block[0], 1), **kw)
 
     def step(self, n=1):
-        """Advance G→G by n base intervals: level l takes 2**l hops per interval.
-
-        All n intervals are enqueued in order on the one stream, and the host waits
-        once, before returning; step(0) launches and waits for nothing. On return
-        the device work is complete, step_count and an open window cover all n
-        intervals, and the raw-exchange ledger holds the final interval only.
-
-        Asynchronous device errors surface at that single wait, so they are not
-        attributed to the interval that caused them; call step(1) repeatedly to
-        localize one (same launches, same bits). On any failure the instance is
-        failed, the exchange ledger and window are invalid, and step_count keeps
-        its value from entry to the failing call.
-        """
+        """Advance G→G by n base intervals: level l takes 2**l hops per interval."""
         if isinstance(n, bool) or not isinstance(n, Integral) or n < 0:
             raise ValueError("n must be a nonnegative integer")
         with self._lock, self.ctx.activate():
             self._require("ready")
-            if n == 0:
-                return
             root = self._levels[0]
             kw = dict(block=self._block, stream=self._stream)
             try:
                 self._status = "advancing"
-                self._exchange_valid = False
                 for _ in range(n):
-                    if self._observe_raw_exchange:  # stream-ordered: holds the last interval only
+                    if self._observe_raw_exchange:
+                        self._exchange_valid = False
                         for b in self._level_boundaries:
                             if b is not None:
                                 cuda.memset_d8_async(b.interval_integral, 0,
                                                      len(b.tables["cell"]) * 32, self._stream)
                     self._hop(0, root.current, root.spare, **kw)
-                    root.current, root.spare = root.spare, root.current  # host handles only
-                self._stream.synchronize()
-                self._step_count += n
-                self._exchange_valid = self._observe_raw_exchange
+                    self._stream.synchronize()
+                    root.current, root.spare = root.spare, root.current
+                    self._step_count += 1
+                    self._exchange_valid = self._observe_raw_exchange
                 self._status = "ready"
             except BaseException:
                 self._window_valid = False

src/CelerisLab/lbm/refinement_tables.py

@@ -8,7 +8,7 @@ from dataclasses import dataclass, replace
 import numpy as np
 
 from .blocks import blocked_link_tables
-from ..body.jets import _surface_links
+from ..body.jets import jet_link_shapes, surface_link_velocity
 from .descriptors import D2Q9_EX, D2Q9_EY
 from .refinement import ACTIVE, OUTER_GHOST
 
@@ -142,13 +142,11 @@ def _prepare_blocked_boundary(boundary, layout, cfg, jets, *, _shifted_blocked=F
     tables.update(body_id=np.asarray(body_id, np.int32), rx=np.asarray(rx, np.float32),
                   ry=np.asarray(ry, np.float32), aperture=np.zeros(len(dirs), bool))
     actuated = bool(jets) or any(o.surface_actuation is not None for o in boundary.objects)
-    jet_cache = None
     if actuated:
         cfg = replace(cfg)
         cfg.validate()
-        tables["surface_velocity"], tables["aperture"], shape, jet_id = _surface_links(
+        tables["surface_velocity"], tables["aperture"] = surface_link_velocity(
             boundary.objects, cfg, boundary.flags, fluid_idx, dirs, body_id, rx, ry,
             jets=jets, hits=boundary.curved_hits, _shifted_blocked=_shifted_blocked)
-        if jets:
-            jet_cache = (shape, jet_id)
-    return _BlockedBoundary(boundary, tables, jets, actuated, jet_cache)
+    return _BlockedBoundary(boundary, tables, jets, actuated,
+                            jet_link_shapes(jets, boundary.curved_hits) if jets else None)

tests/unit/cpp/channel_root.cpp

@@ -16,7 +16,6 @@
 #define NZ 1
 #define NT 1
 #define COLLISION_MODEL 0
-#define BLOCK_ROOT_BC 2
 #define USE_LES 0
 #define RHO 1.f
 #define U0 .025f

tests/unit/historical_tree.py

@@ -1,46 +0,0 @@
-"""Check historical source manifests against their own recorded commit, not today's checkout."""
-
-import io
-from pathlib import Path
-import subprocess
-import tarfile
-
-ROOT = Path(__file__).resolve().parents[2]
-
-
-def _require(commit):
-    """Missing recorded history is an explicit failure, never a skip or a repin."""
-    found = subprocess.run(
-        ["git", "-C", str(ROOT), "cat-file", "-e", commit + "^{commit}"],
-        capture_output=True,
-    )
-    if found.returncode:
-        raise AssertionError(
-            f"Required history missing: commit {commit} is not in this clone; "
-            "fetch it instead of repinning or substituting"
-        )
-
-
-def tracked(commit, prefix="src"):
-    """Tracked paths under prefix at the recorded commit."""
-    _require(commit)
-    return set(
-        subprocess.check_output(
-            ["git", "-C", str(ROOT), "ls-tree", "-r", "--name-only", commit, prefix],
-            text=True,
-        ).splitlines()
-    )
-
-
-def historical_tree(commit, dest):
-    """Extract the tracked tree of the recorded commit into dest."""
-    _require(commit)
-    archive = subprocess.check_output(
-        ["git", "-C", str(ROOT), "archive", "--format=tar", commit]
-    )
-    dest = Path(dest)
-    dest.mkdir(parents=True, exist_ok=True)
-    extra = {"filter": "data"} if hasattr(tarfile, "data_filter") else {}
-    with tarfile.open(fileobj=io.BytesIO(archive)) as tar:
-        tar.extractall(dest, **extra)
-    return dest

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.load_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_build_contracts.py

@@ -83,32 +83,6 @@ assert abs(sum(compute_feq_d2q9(1.,0.,0.)) - 1.) < 1e-6
 
 
 class BuildContracts(unittest.TestCase):
-    def test_block_root_modes_are_rendered_and_recorded(self):
-        with tempfile.TemporaryDirectory() as root:
-            for mode in (0, 1, 2):
-                with self.subTest(mode=mode):
-                    build = compiler.generate_config(LBMConfig(), build_root=root, block_root_bc=mode)
-                    try:
-                        method = Path(build, "config/config_method.h").read_text()
-                        self.assertIn(f"#define BLOCK_ROOT_BC {mode}\n", method)
-                        metadata = json.loads(Path(build, ".celeris-build").read_text())
-                        self.assertEqual(metadata["block_root_bc"], mode)
-                        self.assertNotIn("block_root_bc", metadata["config"])
-                    finally:
-                        compiler.cleanup_build_dir(build)
-            build = compiler.generate_config(LBMConfig(), build_root=root)
-            try:
-                self.assertEqual(json.loads(Path(build, ".celeris-build").read_text())["block_root_bc"], 0)
-            finally:
-                compiler.cleanup_build_dir(build)
-
-    def test_invalid_block_root_mode_fails_before_snapshot(self):
-        for mode in (-1, 3, True, False, 1.0, "1", None):
-            with self.subTest(mode=mode), patch.object(compiler.tempfile, "mkdtemp") as allocate:
-                with self.assertRaisesRegex(ValueError, "block_root_bc"):
-                    compiler.generate_config(LBMConfig(), block_root_bc=mode)
-                allocate.assert_not_called()
-
     @staticmethod
     def fake_compile(command, check):
         source = Path(command[-3])

tests/unit/test_channel_root.py

@@ -50,7 +50,7 @@ def test_masks_and_minimal_footprint():
 def test_actual_closure_read_conflicts():
     pair, layout, targets = transfer((6, 6, 18, 14))
     validate_channel_transfers(pair, layout, targets)
-    for target in (layout.slot(int(layout.owner[5, -1]) - 1, 2),  # integer under NumPy 1 and 2
+    for target in (layout.slot(layout.owner[5, -1]-1, 2),
                    layout.slot(layout.owner[1, 4], 4),
                    layout.slot(layout.owner[-2, 4], 3)):
         with pytest.raises(ValueError, match="immutable"):
@@ -80,8 +80,8 @@ def test_evolved_ghost_snapshot_and_physical_download():
     host.cuda.memcpy_dtoh.side_effect = lambda a, ptr: np.copyto(a, storage[id(ptr)])
     context = Mock(activate=nullcontext, sm_arch="sm_70")
     cfg = LBMConfig(nx=24, ny=20, viscosity=.1, velocity=.025)
-    backend = host.FixedPatchStepper(cfg, context=context, coarse_boundary="channel",
-                                    refinements=[(6, 6, 18, 14)], level_blocks=[(4, 6), (4, 4)])
+    backend = host.FixedPatchStepper(cfg, (6, 6, 18, 14), context=context,
+                                    coarse_boundary="channel", base_block=(4, 6), fine_block=(4, 4))
     try:
         weights = np.array([4/9]+[1/9]*4+[1/36]*4)[:, None, None]
         arrays = [np.broadcast_to(weights, level.shape).copy() for level in backend._levels]
@@ -97,6 +97,5 @@ def test_evolved_ghost_snapshot_and_physical_download():
         with pytest.raises(ValueError, match="finite"):
             backend.initialize(*arrays)
         assert backend._levels[0].kernel == host.compiler.load_module.return_value.get_function("BlockChannelStep")
-        assert [call.kwargs["block_root_bc"] for call in host.compiler.generate_config.call_args_list] == [2, 0]
     finally:
         backend.close()

tests/unit/test_dfg_re20.py

@@ -75,5 +75,59 @@ class Re20Contract(unittest.TestCase):
         self.assertEqual((out["status"], out["verdict"]), ("INCOMPLETE", None))
 
 
+class Re20Provenance(unittest.TestCase):
+    def test_records_current_checkout(self):
+        record = v.provenance(rr.DEFAULT_SOURCE)
+        self.assertEqual(record["source_mode"], "current_checkout")
+        self.assertEqual(len(record["checkout_revision"]), 40)
+        self.assertEqual(len(record["src_tree"]), 40)
+        self.assertEqual(len(record["runner_sha256"]), 64)
+        self.assertFalse(record["checkout_dirty"])
+        self.assertEqual(v.RESULT_SCHEMA, 2)
+
+    def test_reader_accepts_both_schemas_and_rejects_unknown(self):
+        import json
+        import tempfile
+        from pathlib import Path
+        case = dict(case="refined", status="NOT_STEADY", verdict=None, rows=[])
+        with tempfile.TemporaryDirectory() as d:
+            old, new, bad = Path(d, "old.json"), Path(d, "new.json"), Path(d, "bad.json")
+            old.write_text(json.dumps([case]))
+            new.write_text(json.dumps(dict(schema=2, provenance=dict(src_tree="a" * 40),
+                                           cases=[case])))
+            bad.write_text(json.dumps(dict(schema=3, cases=[case])))
+            self.assertEqual(v.read_result(old), dict(schema=1, provenance=None, cases=[case]))
+            self.assertEqual(v.read_result(new)["cases"][0], case)
+            self.assertEqual(v.read_result(new)["provenance"]["src_tree"], "a" * 40)
+            with self.assertRaisesRegex(ValueError, "unknown DFG Re20 result schema"):
+                v.read_result(bad)
+
+    def test_dirty_checkout_is_refused(self):
+        from unittest.mock import patch
+        # Three git calls in order: HEAD:src, HEAD, status --porcelain.
+        with patch("subprocess.check_output",
+                   side_effect=["e" * 40 + "\n", "f" * 40 + "\n", " M src/x.py\n"]) as git:
+            with self.assertRaisesRegex(RuntimeError, "clean checkout"):
+                v.provenance(rr.DEFAULT_SOURCE)
+        self.assertEqual(git.call_count, 3)
+        self.assertEqual([c.args[0][-1] for c in git.call_args_list[:2]], ["HEAD:src", "HEAD"])
+        self.assertIn("--porcelain", git.call_args_list[2].args[0])
+
+    def test_output_inside_checkout_is_refused_before_any_work(self):
+        import tempfile
+        from pathlib import Path
+        from unittest.mock import patch
+        source = rr.DEFAULT_SOURCE.resolve()
+        for output in (source, source / "re20-out", source / "src" / "x"):
+            argv = ["run_dfg_re20.py", "--plan", "--source", str(source), "--output", str(output)]
+            with patch("sys.argv", argv), \
+                 patch.object(v, "provenance", side_effect=AssertionError("work started")):
+                with self.assertRaisesRegex(SystemExit, "outside the source checkout"):
+                    v.main()
+        self.assertFalse((source / "re20-out").exists())
+        with tempfile.TemporaryDirectory() as d:
+            v.reject_output_inside(Path(d) / "out", source)  # outside: accepted
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_jet_circle_cv.py

@@ -8,7 +8,6 @@ import numpy as np
 ROOT = Path(__file__).resolve().parents[2]
 REPO = ROOT
 sys.path.insert(0, str(ROOT / "tests/validation"))
-sys.path.insert(0, str(ROOT / "tests/unit"))
 sys.path.insert(0, str(REPO / "src"))
 from run_jet_circle_cv import (
     CX,
@@ -20,7 +19,6 @@ from run_jet_circle_cv import (
     cv_from_macros,
     aperture_probe,
     measured_aperture_flux,
-    PIN,
     verify,
 )
 
@@ -168,13 +166,80 @@ class Host(unittest.TestCase):
         self.assertEqual(a["steps"] * 4, b["steps"])
 
     def test_source_pin(self):
-        from tempfile import TemporaryDirectory
-        from historical_tree import historical_tree
+        """Historical integrity: the manifest matches its pinned commit, read with git."""
+        from run_jet_circle_cv import verify_historical
+        from source_provenance import HistoryUnavailable, history_unavailable
+        try:
+            verify_historical(REPO)
+        except HistoryUnavailable as error:
+            history_unavailable(self, error)
 
+    def test_pinned_mode_still_rejects_changed_source(self):
+        """Negative control: the frozen-source guard refuses a tree that differs from PIN."""
+        import json
+        import run_jet_circle_cv as v
+        manifest = json.loads((v.HERE / "data/jet_circle_cv/source_manifest.json").read_text())
+        import hashlib
+        current = {f: hashlib.sha256((REPO / f).read_bytes()).hexdigest()
+                   for f in manifest["files"] if (REPO / f).exists()}
+        if current == manifest["files"]:
+            self.skipTest("working tree equals PIN; no changed source to reject")
         with self.assertRaisesRegex(RuntimeError, "Source hash mismatch"):
             verify(REPO)
-        with TemporaryDirectory() as directory:
-            verify(historical_tree(PIN, directory))
+
+    def test_current_checkout_provenance(self):
+        from run_jet_circle_cv import current_checkout
+        record = current_checkout(REPO)
+        self.assertEqual(record["source_mode"], "current_checkout")
+        self.assertIn("not the frozen", record["source_capture"])
+        self.assertEqual(len(record["src_tree"]), 40)
+        self.assertTrue(record["src_sha256"])
+
+
+    def test_final_source_recheck_completes_or_invalidates_receipt(self):
+        import json
+        import tempfile
+        import run_jet_circle_cv as v
+        recorded = dict(source_mode="current_checkout", checkout_revision="a" * 40,
+                        checkout_dirty=False, src_tree="b" * 40)
+        payload = dict(status="PILOT_DIAGNOSTIC_ONLY", run_status="INCOMPLETE", windows=[1, 2])
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "re100_s1.json"
+            path.write_text(json.dumps(payload))
+            done = v.finish_receipt(path, payload, recorded, lambda: dict(recorded))
+            self.assertEqual(json.loads(path.read_text()), done)
+            self.assertEqual((done["run_status"], done["source_recheck"]), ("COMPLETE", "UNCHANGED"))
+            for recheck, state in ((lambda: dict(recorded, checkout_dirty=True), "CHANGED"),
+                                   (lambda: v.verify(Path(d)), "RECHECK_FAILED")):
+                path.write_text(json.dumps(payload))
+                with self.assertRaisesRegex(RuntimeError, "Source changed during the run"):
+                    v.finish_receipt(path, payload, recorded, recheck)
+                bad = json.loads(path.read_text())
+                self.assertEqual((bad["run_status"], bad["source_recheck"]),
+                                 ("INVALID_SOURCE_CHANGED", state))
+                self.assertEqual(bad["source_after"] is None, state == "RECHECK_FAILED")
+                self.assertEqual(bad["windows"], [1, 2])
+            self.assertFalse(Path(str(path) + ".tmp").exists())
+
+    def test_receipts_written_during_the_run_are_incomplete(self):
+        """Static check (run_case needs CUDA): the in-run payload is INCOMPLETE and only
+        main() -> finish_receipt can mark it COMPLETE."""
+        import inspect
+        import run_jet_circle_cv as v
+        self.assertIn('run_status="INCOMPLETE"', inspect.getsource(v.run_case))
+        self.assertNotIn("COMPLETE\"", inspect.getsource(v.run_case).replace("INCOMPLETE\"", ""))
+        self.assertIn("finish_receipt(", inspect.getsource(v.main))
+
+
+    def test_output_inside_checkout_is_refused_before_source_or_cuda(self):
+        from unittest import mock
+        import run_jet_circle_cv as v
+        argv = ["run_jet_circle_cv.py", "--repo", str(REPO), "--output", str(REPO / "v1-out")]
+        with mock.patch("sys.argv", argv), mock.patch.dict("os.environ", CUDA_VISIBLE_DEVICES="0"), \
+             mock.patch.object(v, "source_identity", side_effect=AssertionError("work started")):
+            with self.assertRaisesRegex(SystemExit, "outside the source checkout"):
+                v.main()
+        self.assertFalse((REPO / "v1-out").exists())
 
 
 if __name__ == "__main__":

tests/unit/test_jet_cv_measurements.py

@@ -151,6 +151,8 @@ class Measurements(unittest.TestCase):
             (path / "samples_s2.json").write_text(
                 json.dumps(dict(case=c, sample_file="samples_s2.npz"))
             )
+            (path / f"re{c['re_d']}_s2.json").write_text(json.dumps(
+                dict(case=c, run_status="COMPLETE", source_recheck="UNCHANGED")))
             np.savez(
                 path / "samples_s2.npz",
                 t_reference=t,
@@ -163,6 +165,7 @@ class Measurements(unittest.TestCase):
                 velocity_error=np.zeros(n),
             )
             result = analyze(path)
+            self.assertEqual(result["receipt_gate"], "COMPLETE_SOURCE_UNCHANGED")
             self.assertEqual(len(result["windows"]), 32)
             np.testing.assert_allclose(
                 result["windows"][0]["normalized_cv_residual"], 0, atol=1e-14
@@ -178,5 +181,36 @@ class Measurements(unittest.TestCase):
         np.testing.assert_allclose(integrate(t, y), [112, 8], atol=1e-15)
 
 
+    def test_receipt_gate_refuses_incomplete_invalid_and_unmarked_legacy(self):
+        import json
+        from analyze_jet_cv_continuation import receipt_gate
+
+        c = case(2, steps=3200)
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp)
+            receipt = path / f"re{c['re_d']}_s2.json"
+            with self.assertRaisesRegex(ValueError, "No sibling run receipt"):
+                receipt_gate(path, c)
+            for fields, message in (
+                (dict(run_status="INCOMPLETE"), "INCOMPLETE"),
+                (dict(run_status="INVALID_SOURCE_CHANGED", source_recheck="CHANGED"), "INVALID"),
+                (dict(run_status="COMPLETE", source_recheck="RECHECK_FAILED"), "RECHECK_FAILED"),
+                (dict(run_status="COMPLETE"), "source_recheck=None"),
+                (dict(), "Legacy run receipt"),
+            ):
+                receipt.write_text(json.dumps(dict(case=c, **fields)))
+                with self.assertRaisesRegex(ValueError, message):
+                    receipt_gate(path, c)
+            receipt.write_text(json.dumps(dict(case=dict(c, scale=1), run_status="COMPLETE",
+                                               source_recheck="UNCHANGED")))
+            with self.assertRaisesRegex(ValueError, "different case"):
+                receipt_gate(path, c)
+            receipt.write_text(json.dumps(dict(case=c)))
+            self.assertEqual(receipt_gate(path, c, legacy=True), "LEGACY_NO_FINAL_SOURCE_RECHECK")
+            receipt.write_text(json.dumps(dict(case=c, run_status="INCOMPLETE")))
+            with self.assertRaisesRegex(ValueError, "INCOMPLETE"):
+                receipt_gate(path, c, legacy=True)  # legacy opt-in never admits new statuses
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_les_flow_runner.py

@@ -13,12 +13,15 @@ from unittest.mock import patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
 from run_uniform_les_flow import source_guard, decode_macros, HERE, HEADER
+from source_provenance import (HistoryUnavailable, historical_files, historical_sha256,
+                               historical_tree, history_unavailable)
 
 ROOT = HERE.parents[1]
-sys.path.insert(0, str(ROOT / "tests/unit"))
-from historical_tree import historical_tree, tracked
+A135252 = "a135252ae78b47b57d7e77d19001700b3bedac8b"
 
-A135 = "a135252ae78b47b57d7e77d19001700b3bedac8b"
+
+def a135252_pins():
+    return json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
 
 
 class GuardTests(unittest.TestCase):
@@ -45,24 +48,35 @@ class GuardTests(unittest.TestCase):
         with self.assertRaisesRegex(RuntimeError, "velocity/profile"):
             decode_macros(invalid, w, cx, cy)
 
-    def test_current_corrected_pins(self):
+    def materialize(self, root):
+        """The pinned a135252 tree, read with git (the guard's own qualification)."""
+        try:
+            historical_files(ROOT, A135252, a135252_pins()["sources"], root)
+        except HistoryUnavailable as error:
+            history_unavailable(self, error)
+
+    def test_a135252_pins_qualify_their_own_commit(self):
         with TemporaryDirectory() as directory:
-            root = historical_tree(A135, directory)
-            with patch("run_uniform_les_flow.subprocess.check_output", return_value="historical-tree\n"):
-                identity = source_guard(root, "corrected", "a135252")
-            # base_commit comes from the pins file; the mock only feeds checkout_commit.
-            self.assertEqual(identity["base_commit"], A135)
-            self.assertEqual(identity["checkout_commit"], "historical-tree")
+            root = Path(directory)
+            self.materialize(root)
+            with patch("run_uniform_les_flow.subprocess.check_output",
+                       return_value="isolated-fixture\n"):
+                self.assertEqual(source_guard(root, "corrected", "a135252")["base_commit"], A135252)
+
+    def test_current_tree_is_not_the_a135252_qualification(self):
+        """Explicit record: the guard refuses today's src wherever it differs from the pin."""
+        pins = a135252_pins()["sources"]
+        changed = sorted(n for n, sha in pins.items() if (ROOT / n).exists()
+                         and hashlib.sha256((ROOT / n).read_bytes()).hexdigest() != sha)
+        if not changed:
+            self.skipTest("current src equals the a135252 pin")
+        with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
+            source_guard(ROOT, "corrected", "a135252")
 
     def test_corrected_and_tamper_guard(self):
-        pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
         with TemporaryDirectory() as directory:
-            origin = historical_tree(A135, Path(directory) / "origin")
-            root = Path(directory) / "fixture"
-            for name in pins["sources"]:
-                dest = root / name
-                dest.parent.mkdir(parents=True, exist_ok=True)
-                shutil.copyfile(origin / name, dest)
+            root = Path(directory)
+            self.materialize(root)
             with patch(
                 "run_uniform_les_flow.subprocess.check_output",
                 return_value="isolated-fixture\n",
@@ -93,28 +107,25 @@ class GuardTests(unittest.TestCase):
         with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
             source_guard(ROOT, "corrected")  # Omission still uses the historical profile.
 
-    def test_current_manifest_pins_actual_corrected_tree(self):
-        pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
-        self.assertEqual(pins["base_commit"], A135)
-        self.assertEqual(set(pins["sources"]), tracked(A135))
-        with TemporaryDirectory() as directory:
-            root = historical_tree(A135, directory)
-            for name, expected in pins["sources"].items():
-                self.assertEqual(hashlib.sha256((root / name).read_bytes()).hexdigest(), expected)
-            self.assertEqual(pins["sources"][HEADER], pins["candidate_header_sha256"])
-            self.assertNotEqual(pins["sources"][HEADER], pins["baseline_header_sha256"])
-            with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
-                source_guard(root, "baseline", "a135252")
+    def test_manifest_pins_the_a135252_corrected_tree(self):
+        pins = a135252_pins()
+        self.assertEqual(pins["base_commit"], A135252)
+        self.assertEqual(pins["sources"][HEADER], pins["candidate_header_sha256"])
+        self.assertNotEqual(pins["sources"][HEADER], pins["baseline_header_sha256"])
+        with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
+            source_guard(ROOT, "baseline", "a135252")
+        try:
+            names = historical_tree(ROOT, A135252)
+            actual = historical_sha256(ROOT, A135252, pins["sources"])
+        except HistoryUnavailable as error:
+            history_unavailable(self, error)
+        self.assertEqual(set(pins["sources"]), set(names))
+        self.assertEqual(actual, pins["sources"])
 
     def test_named_baseline_accepts_only_reversed_header(self):
-        pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
         with TemporaryDirectory() as directory:
-            origin = historical_tree(A135, Path(directory) / "origin")
-            root = Path(directory) / "fixture"
-            for name in pins["sources"]:
-                dest = root / name
-                dest.parent.mkdir(parents=True, exist_ok=True)
-                shutil.copyfile(origin / name, dest)
+            root = Path(directory)
+            self.materialize(root)
             subprocess.run(["git", "apply", "--reverse", str(HERE / "les/data/smagorinsky_normalization.patch")], cwd=root, check=True)
             with patch("run_uniform_les_flow.subprocess.check_output", return_value="baseline-fixture\n"):
                 source_guard(root, "baseline", "a135252")

tests/unit/test_raw_exchange.py

@@ -157,9 +157,8 @@ class RawExchange(unittest.TestCase):
         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)
+            host.FixedPatchStepper(LBMConfig(nx=32, ny=32, viscosity=.1), (4, 4, 28, 28),
+                                   context=ctx, fine_objects=[body(radius=2.15)], observe_raw_exchange=True)
         ctx.activate.assert_not_called()
 
     def test_all_depth_weights(self):

tests/unit/test_re100.py

@@ -5,7 +5,6 @@ import pytest
 
 ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT / "tests/validation"))
-sys.path.insert(0, str(ROOT / "tests/unit"))
 import re100_helpers as h
 import periodic as p
 
@@ -63,16 +62,31 @@ def test_periodic_synthetic():
     assert not p.assess(rows()[:100])["periodic"]
 
 
-def test_production_unmodified(tmp_path):
-    from historical_tree import historical_tree
-
+def test_manifest_matches_its_pinned_commit():
+    """Historical integrity only: the manifest describes the h.PIN tree (read with git)."""
     m = json.loads(
         (ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text()
     )
     assert m["commit"] == h.PIN
-    origin = historical_tree(m["commit"], tmp_path)
-    for name, digest in m["production_sha256"].items():
-        assert hashlib.sha256((origin / name).read_bytes()).hexdigest() == digest, name
+    import unittest
+    from source_provenance import HistoryUnavailable, history_unavailable, verify_manifest
+    try:
+        verify_manifest(ROOT, m["commit"], m["production_sha256"])
+    except HistoryUnavailable as error:
+        history_unavailable(unittest.TestCase(), error)
+
+
+def test_current_src_is_not_the_pinned_qualification():
+    """Explicit record: today's src differs from h.PIN, so its qualification does not carry over."""
+    m = json.loads(
+        (ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text()
+    )
+    changed = sorted(n for n, d in m["production_sha256"].items()
+                     if not (ROOT / n).exists()
+                     or hashlib.sha256((ROOT / n).read_bytes()).hexdigest() != d)
+    if not changed:
+        pytest.skip("current src equals the pinned tree")
+    assert changed  # e.g. src/CelerisLab/lbm/channel.py since 47b05bc
 
 
 def test_reference():

tests/unit/test_refined_dfg.py

@@ -7,37 +7,38 @@ import numpy as np
 import unittest
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
-sys.path.insert(0, str(Path(__file__).resolve().parent))
 import run_refined_dfg as r
-from historical_tree import historical_tree
-from tempfile import TemporaryDirectory
 
 h, periodic = r.helpers(r.DEFAULT_SOURCE)
 
 
+def historical_manifest(collision="SRT", observation_mode="last_base_step"):
+    """The manifest verify_source would select, without hashing the working tree.
+
+    Receipt/plan logic tests use it; source integrity is checked once in test_source_pin.
+    """
+    name = ("source_manifest_window.json" if observation_mode == "window" else
+            "source_manifest.json" if collision == "SRT" else "source_manifest_mrt.json")
+    return json.loads((r.HERE / "data/refined_dfg" / name).read_text())
+
+
 def test_source_pin():
-    historical = json.loads((r.HERE / "data/refined_dfg/source_manifest.json").read_text())
+    historical = historical_manifest("SRT")
     assert historical["commit"].startswith("cd3a657")
-    mrt_history = json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text())
+    mrt_history = historical_manifest("MRT")
     assert mrt_history["commit"].startswith("3ea131e")
-    with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
-        r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")
-    with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
-        r.verify_source(r.DEFAULT_SOURCE, "MRT")
-    with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
-        r.verify_source(r.DEFAULT_SOURCE, "SRT")
-    with TemporaryDirectory() as directory:
-        for collision, mode, prefix in (
-            ("SRT", "last_base_step", "cd3a657"),
-            ("MRT", "last_base_step", "3ea131e"),
-            ("MRT", "window", "a233c90"),
-        ):
-            name = {"SRT": "source_manifest.json", "MRT": "source_manifest_mrt.json"}[collision]
-            name = "source_manifest_window.json" if mode == "window" else name
-            commit = json.loads((r.HERE / "data/refined_dfg" / name).read_text())["commit"]
-            assert commit.startswith(prefix)
-            root = historical_tree(commit, Path(directory) / commit)
-            assert r.verify_source(root, collision, mode)["commit"] == commit
+    assert historical_manifest("MRT", "window")["commit"].startswith("a233c90")
+    # Negative controls first, so missing history can never hide them: every pinned
+    # mode refuses a working tree that differs from its pin.
+    for collision, mode in (("MRT", "last_base_step"), ("SRT", "last_base_step"), ("MRT", "window")):
+        with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
+            r.verify_source(r.DEFAULT_SOURCE, collision, mode)
+    from source_provenance import HistoryUnavailable, history_unavailable, verify_manifest
+    for manifest in (historical, mrt_history, historical_manifest("MRT", "window")):
+        try:  # historical integrity: each manifest matches the tree of its own commit
+            verify_manifest(r.DEFAULT_SOURCE, manifest["commit"], manifest["sha256"])
+        except HistoryUnavailable as error:
+            history_unavailable(unittest.TestCase(), error)
 
 
 def test_matched_units_and_planes():
@@ -157,47 +158,17 @@ def test_axis_specific_sampling_phase():
     np.testing.assert_allclose(fine_y - ((k + 43) - 0.5) * 0.00125, 0, atol=1e-15)
 
 
-def frozen_manifest(observation_mode):
-    """Frozen MRT manifest JSON as recorded, without hashing the working tree."""
-    name = "source_manifest_window.json" if observation_mode == "window" else "source_manifest_mrt.json"
-    return json.loads((r.HERE / "data/refined_dfg" / name).read_text())
-
-
-def selected_manifest(collision, observation_mode):
-    """verify_source's manifest selection with its working-tree hash comparison neutralised.
-
-    Source identity is owned by test_source_pin; this exercises only which frozen manifest
-    verify_source selects and returns for (collision, observation_mode).
-    """
-    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)
-
-
 def test_full_requires_completed_pilot():
     with unittest.TestCase().assertRaisesRegex(ValueError, "pilot-results"):
-        r.verify_pilot(None, frozen_manifest("window"))
+        r.verify_pilot(None, historical_manifest("MRT", "window"))
 
 
 def pilot_fixture(path, observation_mode="window", jet_scenario="none", case_selection="both", uniform_diameter=80, placement="wide"):
     """Synthetic contract fixture only; never a physical result or run receipt."""
-    # Receipt-logic fixture: the frozen manifest verify_source returns for a matching tree.
-    # Source identity itself is checked only by test_source_pin.
-    manifest = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", jet_scenario)
-                if observation_mode == "window" and jet_scenario != "none" else
-                frozen_manifest(observation_mode))
+    manifest = ((historical_manifest("MRT", "window") if jet_scenario == "none" else
+                 r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", jet_scenario))
+                if observation_mode == "window" else
+                json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text()))
     selection = dict(case_selection=case_selection, uniform_diameter=uniform_diameter, placement=placement)
     (path / "plan.json").write_text(
         json.dumps(
@@ -395,7 +366,7 @@ def test_record_actual_checkout_separately():
     actual = checkout_provenance(r.DEFAULT_SOURCE)
     assert len(actual["checkout_revision"]) == 40
     assert isinstance(actual["checkout_dirty"], bool)
-    assert selected_manifest("MRT", "window")["production_origin_commit"].startswith("d930")
+    assert historical_manifest("MRT", "window")["production_origin_commit"].startswith("d930")
 
 
 def test_window_plan_preserves_physics():
@@ -405,7 +376,7 @@ def test_window_plan_preserves_physics():
         assert baseline.pop("observation_mode") == "last_base_step"
         assert window.pop("observation_mode") == "window"
         assert baseline == window
-        assert selected_manifest(collision, "window")["observation_mode"] == "window"
+        assert historical_manifest(collision, "window")["observation_mode"] == "window"
     with unittest.TestCase().assertRaises(ValueError):
         r.verify_source(r.DEFAULT_SOURCE, "MRT", "unknown")
 

tests/unit/test_refinement_mrt.py

@@ -24,14 +24,11 @@ class RefinementMRTHost(unittest.TestCase):
         """MRT is an additive compile-time branch; SRT statements and other guards remain."""
         blocks = (KERNELS / "blocks.cu").read_text()
         channel = (KERNELS / "step/block_channel.cu").read_text()
-        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()
         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(text, blocks)
         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):
@@ -40,14 +37,8 @@ class RefinementMRTHost(unittest.TestCase):
             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):
-            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)
-            self.assertNotRegex(source, r"compute_rho_u\(f, rho, ux, uy\);\s*collide_mrt")
+        for source in (blocks, channel, refinement):
+            self.assertEqual(source.count("collide_mrt_no_force("), 1)
 
 
 if __name__ == "__main__": unittest.main()

tests/unit/test_refinement_three_level_multipatch.py

@@ -173,10 +173,8 @@ def bodies_and_jets(speed=.002):
     return bodies,jets
 
 
-def levels(parents=PARENTS,children=CHILDREN,blocks=((4,4),)*3,objects=((),(),()),jets=((),(),())):
-    """Level-list constructor keywords for the L0 set, L1 set and L2 set used here."""
-    return dict(refinements=[parents,children],level_blocks=list(blocks),
-                level_objects=list(objects),level_jets=list(jets))
+KW=dict(base_block=(4,4),fine_block=(4,4),child_block=(4,4),
+        child_rectangles_in_parent=CHILDREN)
 
 
 def mock_backend(*, children=CHILDREN, bodies=False, fail_compile=False, fail_upload=False):
@@ -203,8 +201,9 @@ def mock_backend(*, children=CHILDREN, bodies=False, fail_compile=False, fail_up
         module.cuda.memcpy_htod.side_effect=upload
     objects,jets=bodies_and_jets() if bodies else ((),())
     try:
-        run=module.FixedPatchStepper(CFG,context=context,
-            **levels(children=children,objects=((),(),objects),jets=((),(),jets)))
+        run=module.FixedPatchStepper(CFG,PARENTS,context=context,
+            child_objects=objects,child_jets=jets,
+            **dict(KW,child_rectangles_in_parent=children))
     except RuntimeError:
         for allocation in allocations:
             allocation.free.assert_called_once_with()
@@ -215,29 +214,28 @@ def mock_backend(*, children=CHILDREN, bodies=False, fail_compile=False, fail_up
 class ConstructorAndSchedule(unittest.TestCase):
     def test_invalid_inputs_fail_before_context_or_allocations(self):
         context=Mock()
-        b=(4,4)
-        cases=[dict(children=(8,8,16,16)),dict(children=[]),
-               dict(children=[(16,16,32)]),
-               dict(children=[(16.5,16,32,40)]),
-               dict(children=[(16,16,32,24),(16,27,32,55)]),
-               dict(children=[(40,16,48,40)]),
-               dict(children=[(8,16,32,40)]),
-               dict(blocks=(b,b,None)),dict(blocks=(b,None,b)),dict(blocks=(None,b,b)),
-               dict(blocks=(b,b,(5,4))),dict(blocks=(b,(5,4),b)),
-               dict(blocks=(b,b,(4,))),dict(children=(16,16,32,40))]
+        cases=[dict(child=(8,8,16,16)),dict(child_rectangles_in_parent=[]),
+               dict(child_rectangles_in_parent=[(16,16,32)]),
+               dict(child_rectangles_in_parent=[(16.5,16,32,40)]),
+               dict(child_rectangles_in_parent=[(16,16,32,24),(16,27,32,55)]),
+               dict(child_rectangles_in_parent=[(40,16,48,40)]),
+               dict(child_rectangles_in_parent=[(8,16,32,40)]),
+               dict(child_block=None),dict(fine_block=None),dict(base_block=None),
+               dict(child_block=(5,4)),dict(fine_block=(5,4)),
+               dict(child_block=(4,)),dict(child_rectangles_in_parent=(16,16,32,40))]
         for changes in cases:
             with self.subTest(changes=changes),self.assertRaises((ValueError,TypeError)):
-                host.FixedPatchStepper(CFG,context=context,**levels(**changes))
+                host.FixedPatchStepper(CFG,PARENTS,context=context,**dict(KW,**changes))
         with self.assertRaisesRegex(ValueError,'margin'):
-            host.FixedPatchStepper(CFG,context=context,**levels(parents=[(1,4,20,28),(23,4,44,28)]))
+            host.FixedPatchStepper(CFG,[(1,4,20,28),(23,4,44,28)],context=context,**KW)
         objects,jets=bodies_and_jets()
-        with self.assertRaisesRegex(ValueError,'only on the finest level'):
-            host.FixedPatchStepper(CFG,context=context,**levels(objects=((),objects,())))
+        with self.assertRaisesRegex(ValueError,'only on L2'):
+            host.FixedPatchStepper(CFG,PARENTS,context=context,fine_objects=objects,**KW)
         from CelerisLab.body.jets import Jet,Tube,Parallel
         spanning=Jet(Tube((10.5,14.),0.,25.,.3),Parallel(0.),amplitude=0.)
         with self.assertRaisesRegex(ValueError,'one jet'):
-            host.FixedPatchStepper(CFG,context=context,
-                                  **levels(objects=((),(),objects),jets=((),(),[spanning])))
+            host.FixedPatchStepper(CFG,PARENTS,context=context,child_objects=objects,
+                                  child_jets=[spanning],**KW)
         context.activate.assert_not_called()
         host.cuda.mem_alloc.assert_not_called()
 
@@ -315,23 +313,21 @@ class ConstructorAndSchedule(unittest.TestCase):
 
 
 
-class RemovedLegacyAPI(unittest.TestCase):
-    """The pre-level-list constructor/initialize forms are gone and fail loudly."""
+class InitialLevels(unittest.TestCase):
+    """initialize() entry adapter: positional per-level arrays or the legacy names, never both."""
 
-    def test_removed_constructor_forms_raise_type_error(self):
-        import inspect
-        context = Mock()
-        with self.assertRaises(TypeError):  # positional bounds
-            host.FixedPatchStepper(CFG, PARENTS, context=context)
-        removed = dict(bounds=PARENTS, fine_block=(4, 4), base_block=(4, 4), fine_objects=(),
-                       jets=(), child=(8, 8, 16, 16), child_block=(4, 4), child_objects=(),
-                       child_jets=(), child_rectangles_in_parent=CHILDREN)
-        for name, value in removed.items():
-            with self.subTest(keyword=name), self.assertRaises(TypeError):
-                host.FixedPatchStepper(CFG, context=context, **levels(), **{name: value})
-        context.activate.assert_not_called()
-        self.assertEqual(list(inspect.signature(host.FixedPatchStepper.initialize).parameters),
-                         ["self", "level_g"])  # no coarse_g/fine_g/child_g
+    def test_forms(self):
+        a, b, c = object(), object(), object()
+        f = host._initial_levels
+        self.assertEqual(f((a, b), None, None, None), (a, b))
+        self.assertEqual(f((a, b, c), None, None, None), (a, b, c))
+        self.assertEqual(f((a, b), None, None, c), (a, b, c))      # old initialize(c, f, child_g=g)
+        self.assertEqual(f((), a, b, None), (a, b))
+        self.assertEqual(f((), a, b, c), (a, b, c))
+        for args in (((a,), None, None, c), ((a, b, c), None, None, c), ((a,), a, b, None),
+                     ((), a, None, None), ((), None, b, c)):
+            with self.subTest(args=args), self.assertRaises(TypeError):
+                f(*args)
 
 
 if __name__=='__main__':

tests/unit/test_shifted_jets.py

@@ -54,13 +54,14 @@ class ShiftedJets(unittest.TestCase):
                 (Jet(Tube((10.,14.),0.,26.,.3),Parallel(0.)),))]
         for objects,jets in cases:
             with self.subTest(objects=len(objects),jets=len(jets)),self.assertRaises(ValueError):
-                module.FixedPatchStepper(CFG,context=None,refinements=[backend.PARENTS,A],
-                    level_blocks=[(4,4)]*3,level_objects=[(),(),objects],level_jets=[(),(),jets])
+                module.FixedPatchStepper(CFG,backend.PARENTS,context=None,
+                    base_block=(4,4),fine_block=(4,4),child_block=(4,4),
+                    child_rectangles_in_parent=A,child_objects=objects,child_jets=jets)
         for blocks in ((None,None,None),((4,4),None,(4,4))):
             with self.assertRaises(ValueError):
-                module.FixedPatchStepper(CFG,context=None,refinements=[backend.PARENTS,A],
-                    level_blocks=list(blocks),level_objects=[(),(),(body(),)],
-                    level_jets=[(),(),(jet(),)])
+                module.FixedPatchStepper(CFG,backend.PARENTS,context=None,
+                    base_block=blocks[0],fine_block=blocks[1],child_block=blocks[2],
+                    child_rectangles_in_parent=A,child_objects=(body(),),child_jets=(jet(),))
         module.cuda.mem_alloc.assert_not_called()
         module.compiler.compile_kernel.assert_not_called()
 

tests/unit/test_shifted_storage.py

@@ -78,15 +78,13 @@ class ShiftedStorage(unittest.TestCase):
 
     def test_shifted_density_and_tag_validation(self):
         old,new = topology(A),topology(B)
-        a = tuple(np.where(m[None], np.float32(-.001), np.float32(np.nan)).repeat(9,axis=0)
-                  for m in old.leaf_masks)  # FP32 under NumPy 1 and 2 promotion rules
+        a = tuple(np.where(m[None], np.float32(-.001), np.nan).repeat(9,axis=0) for m in old.leaf_masks)
         remap(old,new,a)
         with self.assertRaisesRegex(ValueError,'representation'): remap(old,new,a,False)
         with self.assertRaisesRegex(ValueError,'FP32'): remap(old,new,tuple(x.astype(np.float64) for x in a))
         a[0][:,old.leaf_masks[0]] = -1
         with self.assertRaisesRegex(ValueError,'positive density'): remap(old,new,a)
-        rest = tuple(np.where(m[None],np.float32(0),np.float32(np.nan)).repeat(9,axis=0)
-                     for m in old.leaf_masks)
+        rest = tuple(np.where(m[None],np.float32(0),np.nan).repeat(9,axis=0) for m in old.leaf_masks)
         remap(old,new,rest)
         with self.assertRaisesRegex(ValueError,'positive density'):
             remap_synchronized(old,new,rest,base_count=7,level_steps=(7,14,28),phase='postcollision_g')
@@ -131,10 +129,9 @@ class ShiftedStorage(unittest.TestCase):
     def test_dense_or_mixed_shifted_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'):
-                module.FixedPatchStepper(cfg,context=None,refinements=[(4,4,20,28)],
-                                         level_blocks=blocks)
+        for kwargs in ({},dict(base_block=(4,4)),dict(fine_block=(4,4))):
+            with self.assertRaisesRegex(ValueError,'all-blocked|base_block requires fine_block'):
+                module.FixedPatchStepper(cfg,(4,4,20,28),context=None,**kwargs)
         module.compiler.compile_kernel.assert_not_called()
 
     def test_uniform_block_codec_and_raw_snapshot(self):

tests/unit/test_step_sync.py

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

tests/unit/test_validation_source_provenance.py

@@ -8,19 +8,19 @@ import sys
 
 ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT / "tests/validation"))
-from source_provenance import checkout_provenance
+import pytest
+from source_provenance import (HistoryUnavailable, checkout_provenance, historical_sha256,
+                               historical_tree, history_unavailable, verify_manifest)
+import unittest
 import re100_helpers
 import run_jet_circle_cv
 
-sys.path.insert(0, str(ROOT / "tests/unit"))
-from historical_tree import historical_tree, tracked
-
 ORIGIN = "47b05bc8abfc957fe876b0c63bdb386e3c6cbaa5"
 
 
-def test_manifests_cover_current_source_and_preserve_origin(tmp_path):
-    assert re100_helpers.PIN == run_jet_circle_cv.PIN == ORIGIN
-    manifests = {}
+def test_manifests_match_their_recorded_source_and_preserve_origin():
+    """Historical evidence integrity: each manifest describes the tree of its own commit."""
+    manifests = []
     for group, hashes_key, revision_key in (
         ("dfg_re100", "production_sha256", "commit"),
         ("jet_circle_cv", "files", "revision"),
@@ -30,6 +30,7 @@ def test_manifests_cover_current_source_and_preserve_origin(tmp_path):
                 ROOT / "tests/validation/data" / group / "source_manifest.json"
             ).read_text()
         )
+        # Structural checks need no history and always run.
         assert manifest[revision_key] == ORIGIN
         assert manifest["source_origin_commits"][-1] == ORIGIN
         assert (
@@ -37,16 +38,15 @@ def test_manifests_cover_current_source_and_preserve_origin(tmp_path):
             == "d79e3f6f0a424f5bcc3a1a6d610313e49daf97a2"
         )
         assert "not inherited from d79" in manifest["experiment"]
-        manifests[group] = manifest[hashes_key]
-    # History-dependent checks last, so a missing commit masks nothing above.
-    names = tracked(ORIGIN)
-    origin = historical_tree(ORIGIN, tmp_path)
-    for group, hashes in manifests.items():
-        assert set(hashes) == set(names), group
-        for name, expected in hashes.items():
-            assert (
-                hashlib.sha256((origin / name).read_bytes()).hexdigest() == expected
-            ), name
+        manifests.append(manifest[hashes_key])
+    try:
+        names = historical_tree(ROOT, ORIGIN)
+        for digests in manifests:
+            assert set(digests) == set(names)
+            verify_manifest(ROOT, ORIGIN, digests)
+    except HistoryUnavailable as error:
+        history_unavailable(unittest.TestCase(), error)
+    assert re100_helpers.PIN == run_jet_circle_cv.PIN == ORIGIN
 
 
 def test_runtime_checkout_identity_is_separate_from_source_origin(tmp_path):
@@ -78,3 +78,62 @@ def test_runtime_checkout_identity_is_separate_from_source_origin(tmp_path):
     assert checkout_provenance(tmp_path) == dict(
         checkout_revision=expected, checkout_dirty=True
     )
+
+
+def test_historical_check_rejects_tampering_and_reports_missing_history(tmp_path):
+    with pytest.raises(HistoryUnavailable):
+        historical_sha256(tmp_path, ORIGIN, ["src/x.py"])  # not a repository
+    with pytest.raises(HistoryUnavailable):
+        historical_sha256(ROOT, "0" * 40, ["src/x.py"])  # commit absent
+    manifest = json.loads(
+        (ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text())
+    digests = dict(manifest["production_sha256"])
+    name = sorted(digests)[0]
+    tampered = dict(digests, **{name: ("0" if digests[name][0] != "0" else "1") + digests[name][1:]})
+    try:
+        historical_tree(ROOT, ORIGIN)
+    except HistoryUnavailable as error:
+        history_unavailable(unittest.TestCase(), error)
+    with pytest.raises(RuntimeError, match="Historical manifest mismatch"):
+        verify_manifest(ROOT, ORIGIN, tampered)  # one flipped digest is caught
+    with pytest.raises(ValueError, match="not part of"):
+        historical_sha256(ROOT, ORIGIN, ["src/CelerisLab/does_not_exist.py"])
+
+
+def test_historical_files_reports_unreadable_blob_as_history_unavailable(tmp_path):
+    """A blob missing from the clone is HistoryUnavailable, not a raw CalledProcessError."""
+    import source_provenance
+    from unittest.mock import patch
+
+    missing = subprocess.CompletedProcess(["git", "show"], 128, b"", b"fatal: missing blob")
+    with patch.object(source_provenance, "historical_sha256", return_value={"src/a.py": "0" * 64}), \
+         patch.object(source_provenance.subprocess, "run", return_value=missing):
+        with pytest.raises(HistoryUnavailable, match="blob not in this clone"):
+            source_provenance.historical_files(ROOT, ORIGIN, ["src/a.py"], tmp_path)
+    assert not (tmp_path / "src/a.py").exists()
+
+
+def test_final_source_check_unchanged_changed_and_failed_recheck():
+    from source_provenance import final_source_check
+
+    recorded = dict(revision="a" * 40, dirty=False)
+    assert final_source_check(recorded, lambda: dict(recorded)) == dict(source_recheck="UNCHANGED")
+    changed = final_source_check(recorded, lambda: dict(recorded, dirty=True))
+    assert changed == dict(source_recheck="CHANGED", source_after=dict(recorded, dirty=True),
+                           source_recheck_error=None)
+
+    def fails():
+        raise RuntimeError("Source hash mismatch: src/x.py")
+
+    failed = final_source_check(recorded, fails)
+    assert (failed["source_recheck"], failed["source_after"]) == ("RECHECK_FAILED", None)
+    assert "Source hash mismatch" in failed["source_recheck_error"]
+
+
+def test_reject_output_inside_checkout(tmp_path):
+    from source_provenance import reject_output_inside
+
+    for output in (ROOT, ROOT / "out", ROOT / "src" / "x.json"):
+        with pytest.raises(SystemExit, match="outside the source checkout"):
+            reject_output_inside(output, ROOT)
+    reject_output_inside(tmp_path / "out", ROOT)

tests/validation/analyze_jet_cv_continuation.py

@@ -14,11 +14,35 @@ def integrate(t, y):
     )
 
 
-def analyze(directory):
+def receipt_gate(directory, c, legacy=False):
+    """Side files are used only if the sibling run receipt finished with an unchanged source.
+
+    Receipts from runners before the final source recheck have no run_status; they are
+    refused unless legacy=True (--legacy-receipts), and the result is labelled as such.
+    """
+    path = Path(directory) / f"re{c['re_d']}_s{c['scale']}.json"
+    if not path.exists():
+        raise ValueError(f"No sibling run receipt {path.name}")
+    receipt = json.loads(path.read_text())
+    if receipt.get("case") != c:
+        raise ValueError("Sibling run receipt describes a different case")
+    if "run_status" not in receipt:
+        if not legacy:
+            raise ValueError("Legacy run receipt without run_status/final source recheck; "
+                             "pass --legacy-receipts to analyze it as such")
+        return "LEGACY_NO_FINAL_SOURCE_RECHECK"
+    if receipt["run_status"] != "COMPLETE" or receipt.get("source_recheck") != "UNCHANGED":
+        raise ValueError(f"Run receipt is {receipt['run_status']} "
+                         f"(source_recheck={receipt.get('source_recheck')})")
+    return "COMPLETE_SOURCE_UNCHANGED"
+
+
+def analyze(directory, legacy=False):
     meta_path = next(Path(directory).glob("samples_s*.json"))
     meta = json.loads(meta_path.read_text())
     c = meta["case"]
     s = c["scale"]
+    gate = receipt_gate(directory, c, legacy)
     sample_path = meta_path.parent / meta["sample_file"]
     checksum_verified = False
     if "sample_sha256" in meta:
@@ -97,6 +121,7 @@ def analyze(directory):
         )
     return dict(
         scale=s,
+        receipt_gate=gate,
         complete_schedule=bool(complete_schedule),
         sample_checksum_verified=checksum_verified,
         windows=records,
@@ -115,8 +140,10 @@ def main():
     p = argparse.ArgumentParser(description=__doc__)
     p.add_argument("directories", nargs="+", type=Path)
     p.add_argument("--output", type=Path, required=True)
+    p.add_argument("--legacy-receipts", action="store_true",
+                   help="Accept receipts written before the final source recheck (labelled)")
     a = p.parse_args()
-    results = [analyze(d) for d in a.directories]
+    results = [analyze(d, a.legacy_receipts) for d in a.directories]
     payload = dict(
         status="TRANSIENT_DIAGNOSTIC_COMPARISON",
         results=results,

tests/validation/multigrid/test_block_stepper.py

@@ -192,7 +192,6 @@ class CpuContracts(unittest.TestCase):
         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()

tests/validation/multigrid/test_channel_root_gpu.py

@@ -27,7 +27,7 @@ def state(ny, nx):
 
 def compile_module(cfg, context, source):
     cfg.validate()
-    build = compiler.generate_config(cfg, block_root_bc=2 if source == "blocks.cu" else 0)
+    build = compiler.generate_config(cfg)
     try:
         module = compiler.load_module(compiler.compile_kernel(context.sm_arch, source=source, build_dir=build))
     finally:
@@ -106,10 +106,9 @@ class ChannelRootGPU(unittest.TestCase):
                     cfg = LBMConfig(nx=48, ny=32, viscosity=.1, velocity=.015, ddf_shifting=shifted)
                     reference = None
                     for block in ((32,48), (8,8), (4,1)):
-                        with FixedPatchStepper(cfg, context=context, coarse_boundary="channel",
-                                refinements=[(8,6,40,26)], level_blocks=[block, (8,8)],
-                                level_objects=[(), (body,) if bodies else ()],
-                                level_jets=[(), (jet,) if bodies else ()]) as backend:
+                        with FixedPatchStepper(cfg, (8,6,40,26), context=context,
+                                coarse_boundary="channel", base_block=block, fine_block=(8,8),
+                                fine_objects=(body,) if bodies else (), jets=(jet,) if bodies else ()) as backend:
                             arrays = [state(*level.shape[1:]) for level in backend._levels]
                             backend.initialize(*arrays)
                             backend.step(4)

tests/validation/multigrid/test_exchange_window.py

@@ -25,8 +25,8 @@ EPS = np.finfo(np.float64).eps
 
 
 def build(m, window=True, raw=True):
-    return FixedPatchStepper(config("periodic"), context=m["context"], refinements=[BOUNDS],
-                             level_blocks=[(3, 4), (16, 20)], level_objects=[(), (circle(),)],
+    return FixedPatchStepper(config("periodic"), BOUNDS, context=m["context"], base_block=(3, 4),
+                             fine_block=(16, 20), fine_objects=[circle()],
                              observe_raw_exchange=raw, observe_window_exchange=window)
 
 
@@ -109,11 +109,10 @@ 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)],
-                      level_objects=[(), (rc(),)], level_jets=[(), (jet(),)],
+            kw = dict(context=ctx, fine_block=(16, 20), fine_objects=[rc()], jets=[jet()],
                       observe_raw_exchange=True)
-            with FixedPatchStepper(CFG, observe_window_exchange=True, **kw) as windowed, \
-                 FixedPatchStepper(CFG, **kw) as plain:
+            with FixedPatchStepper(CFG, RB, observe_window_exchange=True, **kw) as windowed, \
+                 FixedPatchStepper(CFG, RB, **kw) as plain:
                 arrays = [state(level.shape) for level in windowed._levels]
                 windowed.initialize(*arrays)
                 plain.initialize(*arrays)
@@ -201,11 +200,11 @@ class ExchangeWindowShiftedMRTGPU(unittest.TestCase):
         from CelerisLab.cuda.context import CudaContext
         cfg = LBMConfig(nx=32, ny=24, viscosity=.1, collision="MRT", ddf_shifting=True)
         with CudaContext(0) as ctx:
-            kw = dict(context=ctx, refinements=[RB], level_blocks=[(12, 8), (16, 20)],
-                      level_objects=[(), (rc(),)], observe_raw_exchange=True)
-            with FixedPatchStepper(cfg, observe_window_exchange=True, **kw) as windowed, \
-                 FixedPatchStepper(cfg, observe_window_exchange=True, **kw) as split, \
-                 FixedPatchStepper(cfg, **kw) as plain:
+            kw = dict(context=ctx, base_block=(12, 8), fine_block=(16, 20), fine_objects=[rc()],
+                      observe_raw_exchange=True)
+            with FixedPatchStepper(cfg, RB, observe_window_exchange=True, **kw) as windowed, \
+                 FixedPatchStepper(cfg, RB, observe_window_exchange=True, **kw) as split, \
+                 FixedPatchStepper(cfg, RB, **kw) as plain:
                 arrays = [state(level.shape) for level in windowed._levels]
                 for run in (windowed, split, plain):
                     run.initialize(*arrays)

tests/validation/multigrid/test_fine_surface_gpu.py

@@ -140,8 +140,7 @@ def run_patch(speed, *, audit=True):
         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,)])
+            backend = implementation.FixedPatchStepper(cfg, BOUNDS, context=context, fine_objects=[source])
         with backend, context.activate():
             c, f = backend._levels
             cl = backend._fine_curved
@@ -282,8 +281,7 @@ def test_fine_surface_four_hops(speed):
 @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_boundary = SimpleNamespace(objects=[object_spec(.002)])
     backend._fine_curved = SimpleNamespace(fluid_idx_gpu=object(),
                                           surface_velocity_gpu=object(), link_exchange_gpu=object())
     setattr(backend._fine_curved, missing, None)
@@ -327,7 +325,6 @@ def test_cross_interface_donor_rejects_before_cuda():
     context.activate.side_effect = AssertionError("must reject before CUDA activation")
     with patch.object(cuda, "mem_alloc") as allocate:
         with pytest.raises(ValueError, match="donor crosses"):
-            implementation.FixedPatchStepper(cfg, context=context, refinements=[BOUNDS],
-                                             level_objects=[(), (source,)])
+            implementation.FixedPatchStepper(cfg, BOUNDS, context=context, fine_objects=[source])
         context.activate.assert_not_called()
         allocate.assert_not_called()

tests/validation/multigrid/test_multiple_pinned_remap_gpu.py

@@ -48,8 +48,9 @@ def make(context):
         for (x, y), amplitude in zip(centers, (.003, -.002))
     )
     run = FixedPatchStepper(
-        CFG, context=context, refinements=[PARENTS, START], level_blocks=[(4, 4)] * 3,
-        level_objects=[(), (), objects], level_jets=[(), (), jets],
+        CFG, PARENTS, context=context, base_block=(4, 4), fine_block=(4, 4),
+        child_block=(4, 4), child_rectangles_in_parent=START,
+        child_objects=objects, child_jets=jets,
     )
     return run, objects
 

tests/validation/multigrid/test_pinned_body_remap_gpu.py

@@ -25,8 +25,9 @@ B = ((16,16,32,40), (60,16,76,40))
 def make(ctx):
     body = SimObject(71, geometry=CircleGeometry(12.,14.,.8),center=(12.,14.),radius=.8)
     jet = Jet(Tube((10.5,14.),0.,3.,.3),Parallel(0.),'parabolic',.003)
-    run = FixedPatchStepper(CFG,context=ctx,refinements=[PARENTS,A],level_blocks=[(4,4)]*3,
-                           level_objects=[(),(),(body,)],level_jets=[(),(),(jet,)])
+    run = FixedPatchStepper(CFG,PARENTS,context=ctx,base_block=(4,4),fine_block=(4,4),
+                           child_block=(4,4),child_rectangles_in_parent=A,
+                           child_objects=(body,),child_jets=(jet,))
     return run,(body,)
 
 

tests/validation/multigrid/test_raw_exchange_gpu.py

@@ -90,18 +90,14 @@ def census(run, index, buffer):
     return totals(dense, leaf), external, independent
 
 
-
-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],
-                level_objects=[(), (circle(),)], level_jets=[(), (jet(),)])
-
 class RawExchangeGPU(unittest.TestCase):
     def test_row_ledger_interval_jet_switch_and_no_population_effect(self):
         with CudaContext(0) as ctx:
             for block in ((32, 40), (16, 20), (8, 10)):
-                with FixedPatchStepper(CFG, context=ctx, **fine(block), observe_raw_exchange=True) as measured, \
-                     FixedPatchStepper(CFG, context=ctx, **fine(block)) as plain:
+                with FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_block=block,
+                        fine_objects=[circle()], jets=[jet()], observe_raw_exchange=True) as measured, \
+                     FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_block=block,
+                        fine_objects=[circle()], jets=[jet()]) as plain:
                     arrays = [state(l.shape) for l in measured._levels]
                     measured.initialize(*arrays)
                     plain.initialize(*arrays)
@@ -150,8 +146,9 @@ class RawExchangeGPU(unittest.TestCase):
         """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:
+            with FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_objects=[circle()], jets=[jet()]) as dense, \
+                 FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_block=(16, 20),
+                    fine_objects=[circle()], jets=[jet()], observe_raw_exchange=True) as blocked:
                 arrays = [state(l.shape) for l in blocked._levels]
                 dense.initialize(*arrays)
                 blocked.initialize(*arrays)
@@ -232,7 +229,8 @@ class RawExchangeGPU(unittest.TestCase):
         layout = BlockLayout2D.build(48, 64, 8, 8)
         with CudaContext(0) as ctx:
             with PeriodicBlockStepper(fine_cfg, layout, context=ctx, objects=[obj]) as uniform, \
-                 FixedPatchStepper(CFG, context=ctx, **fine((16, 20)), observe_raw_exchange=True) as amr:
+                 FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_block=(16, 20),
+                    fine_objects=[circle()], jets=[jet()], 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]

tests/validation/multigrid/test_refinement_blocked_base.py

@@ -83,10 +83,11 @@ def mocked_backend(mode="periodic", fail_alloc=None):
 class BlockedBaseCPU(unittest.TestCase):
     def test_reject_before_cuda(self):
         context = Mock(activate=Mock(side_effect=AssertionError("CUDA reached")))
-        for blocks in ([(12, 8), None], [(7, 8), FINE_BLOCK], [(12,), FINE_BLOCK]):
-            with self.subTest(blocks=blocks), self.assertRaises(ValueError):
-                FixedPatchStepper(config("periodic"), context=context, refinements=[BOUNDS],
-                                  level_blocks=blocks)
+        for kwargs in ({"base_block": (12, 8)},
+                       {"base_block": (7, 8), "fine_block": FINE_BLOCK},
+                       {"base_block": (12,), "fine_block": FINE_BLOCK}):
+            with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
+                FixedPatchStepper(config("periodic"), BOUNDS, context=context, **kwargs)
         context.activate.assert_not_called()
 
     def test_all_arrival_rows_donors_and_kind_masks(self):
@@ -192,10 +193,9 @@ class BlockedBaseCPU(unittest.TestCase):
 
     def test_launch_order_stream_wall_constants_and_buffer_swaps(self):
         with mocked_backend("slot_reservoir") as m:
-            backend = FixedPatchStepper(config("slot_reservoir"), context=m["context"],
-                                        coarse_boundary="slot_reservoir", refinements=[BOUNDS],
-                                        level_blocks=[(3, 4), FINE_BLOCK],
-                                        level_objects=[(), (circle(actuation(.002)),)])
+            backend = FixedPatchStepper(config("slot_reservoir"), BOUNDS, context=m["context"],
+                                        coarse_boundary="slot_reservoir", fine_block=FINE_BLOCK,
+                                        base_block=(3, 4), fine_objects=[circle(actuation(.002))])
             try:
                 backend.initialize(state(NY, NX, 1), state(*backend.patch.fine_shape, 2))
                 coarse, fine = backend._levels
@@ -223,8 +223,6 @@ class BlockedBaseCPU(unittest.TestCase):
                     self.assertIs(collide[4], coarse.kind)
                 walls = [a for ptr, a in m["uploads"] if isinstance(ptr, str) and ptr == "d_block_wall"]
                 self.assertEqual(len(walls), 1)
-                self.assertEqual([call.kwargs["block_root_bc"] for call in
-                                  compiler.generate_config.call_args_list], [1, 0])
                 width, height, slot, speed = struct.unpack("=IIIf", walls[0])
                 self.assertEqual((width, height, slot), (NX, NY, 8))
                 self.assertAlmostEqual(speed, .001)
@@ -238,16 +236,16 @@ class BlockedBaseCPU(unittest.TestCase):
         for fail_at in (1, 6, 10):
             with self.subTest(fail_at=fail_at), mocked_backend(fail_alloc=fail_at) as m:
                 with self.assertRaisesRegex(MemoryError, "injected allocation"):
-                    FixedPatchStepper(config("periodic"), context=m["context"], refinements=[BOUNDS],
-                                      level_blocks=[(3, 4), FINE_BLOCK])
+                    FixedPatchStepper(config("periodic"), BOUNDS, context=m["context"],
+                                      fine_block=FINE_BLOCK, base_block=(3, 4))
                 self.assertEqual(len(m["allocations"]), fail_at)
                 for allocation in m["allocations"]:
                     allocation.free.assert_called_once_with()
 
     def test_failed_parent_stage_cannot_publish_or_resume(self):
         with mocked_backend() as m:
-            backend = FixedPatchStepper(config("periodic"), context=m["context"], refinements=[BOUNDS],
-                                        level_blocks=[(3, 4), FINE_BLOCK])
+            backend = FixedPatchStepper(config("periodic"), BOUNDS, context=m["context"],
+                                        fine_block=FINE_BLOCK, base_block=(3, 4))
             backend.initialize(state(NY, NX, 1), state(*backend.patch.fine_shape, 2))
             m["functions"]["BlockCoalesce"].side_effect = RuntimeError("injected parent stage failure")
             current = backend._levels[0].current
@@ -291,10 +289,9 @@ class BlockedBaseGPU(unittest.TestCase):
                     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:
+                        with FixedPatchStepper(cfg, BOUNDS, context=context, coarse_boundary=mode,
+                                               fine_block=FINE_BLOCK, base_block=base_block,
+                                               fine_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:

tests/validation/multigrid/test_refinement_blocked_fine.py

@@ -105,10 +105,9 @@ class BlockedFineCPU(unittest.TestCase):
         return Mock(activate=Mock(side_effect=AssertionError("CUDA reached")), sm_arch="sm_70")
 
     def test_rejections_before_cuda(self):
-        for kw in ({"level_blocks": [None, (7, 7)]}, {"level_blocks": [None, (16,)]}):
+        for kw in ({"fine_block": (7, 7)}, {"fine_block": (16,)}):
             with self.subTest(kw=kw), self.assertRaises(ValueError):
-                FixedPatchStepper(config("periodic"), context=self.forbidden_context(),
-                                  refinements=[BOUNDS], **kw)
+                FixedPatchStepper(config("periodic"), BOUNDS, context=self.forbidden_context(), **kw)
 
     def test_single_block_layout_allocates_no_empty_halo_tables(self):
         # A 1x1 non-periodic fine layout has zero halo pairs; CUDA rejects mem_alloc(0).
@@ -126,8 +125,8 @@ class BlockedFineCPU(unittest.TestCase):
                 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]])
+            backend = FixedPatchStepper(config("periodic"), BOUNDS, context=context,
+                                        fine_block=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
@@ -178,9 +177,8 @@ class BlockedFineCPU(unittest.TestCase):
                 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_objects=[(), tuple(objects)])
+            backend = FixedPatchStepper(config("periodic"), BOUNDS, context=context,
+                                        fine_objects=objects, fine_block=FINE_BLOCKS[1])
             backend.initialize(state(NY, NX, 1), state(*backend.patch.fine_shape, 2))
         return backend, calls
 
@@ -291,9 +289,8 @@ class BlockedFineGPU(unittest.TestCase):
             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:
+                    with FixedPatchStepper(cfg, BOUNDS, context=context, coarse_boundary=mode,
+                                           fine_block=fine_block) as backend:
                         backend.initialize(g_coarse, g_fine)
                         backend.step(STEPS)
                         self.assertEqual(backend.step_count, STEPS)
@@ -321,9 +318,8 @@ class BlockedFineGPU(unittest.TestCase):
             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:
+                    with FixedPatchStepper(cfg, BOUNDS, context=context, coarse_boundary=mode,
+                                           fine_objects=[circle()], fine_block=fine_block) as backend:
                         backend.initialize(g_coarse, g_fine)
                         backend.step(STEPS)
                         results.append((backend.download(), raw(backend)))
@@ -350,9 +346,9 @@ class BlockedFineGPU(unittest.TestCase):
                 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:
+                        with FixedPatchStepper(cfg, BOUNDS, context=context, coarse_boundary=mode,
+                                               fine_objects=[circle(actuation(speed))],
+                                               fine_block=fine_block) as backend:
                             backend.initialize(g_coarse, g_fine)
                             backend.step(STEPS)
                             results.append((backend.download(), raw(backend)))

tests/validation/multigrid/test_refinement_followup.py

@@ -94,8 +94,7 @@ class FaultGuardsCPU(unittest.TestCase):
                              ("slot_reservoir", (12, 4, 20, 21))):
             with self.subTest(mode=mode, bounds=bounds):
                 with self.assertRaises(ValueError):
-                    implementation.FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
-                                                     refinements=[bounds])
+                    implementation.FixedPatchStepper(cfg, bounds, context=context, coarse_boundary=mode)
                 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.
@@ -103,14 +102,11 @@ class FaultGuardsCPU(unittest.TestCase):
                              ("periodic", (2, 2, 6, 5))):
             with self.subTest(mode=mode, bounds=bounds):
                 with self.assertRaisesRegex(RuntimeError, "validated before CUDA"):
-                    implementation.FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
-                                                     refinements=[bounds])
+                    implementation.FixedPatchStepper(cfg, bounds, context=context, coarse_boundary=mode)
 
     def fake_backend(self):
         backend = PeriodicPatchStepper.__new__(PeriodicPatchStepper)
-        backend._layouts = [None, None]  # dense 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.fine_layout = None  # dense fine storage
         backend.ctx = Mock(activate=nullcontext)
         backend._lock = RLock()
         backend._status = "ready"
@@ -158,23 +154,12 @@ 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)])
+                PeriodicPatchStepper(cfg, (4, 4, 12, 12), context=context)
         p = implementation.FixedPatch2D.from_config(cfg, (4, 4, 12, 12))
-        from CelerisLab.lbm import hierarchy
-        bad = patch.object(implementation.FixedPatch2D, "from_config",
-                           return_value=replace(p, tau_coarse=.9))
-        # 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, \
-                patch.object(cuda, "memcpy_htod") as upload:
-            with self.assertRaisesRegex(RuntimeError, "config chain is inconsistent"):
-                PeriodicPatchStepper(cfg, context=context, refinements=[p.bounds])
-            with patch.object(hierarchy, "_check_chain"), \
-                    self.assertRaisesRegex(RuntimeError, "omega/tau"):
-                PeriodicPatchStepper(cfg, context=context, refinements=[p.bounds])
-            alloc.assert_not_called()
-            upload.assert_not_called()
+        with patch.object(cuda, "Stream"), patch.object(implementation.FixedPatch2D, "from_config",
+                                                       return_value=replace(p, tau_coarse=.9)):
+            with self.assertRaisesRegex(RuntimeError, "omega/tau"):
+                PeriodicPatchStepper(cfg, p.bounds, context=context)
         module = Mock()
         module.get_global.return_value = (Mock(), 16)
         with patch.object(cuda, "Stream"), patch.object(compiler, "generate_config"), \
@@ -182,7 +167,7 @@ class FaultGuardsCPU(unittest.TestCase):
                 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])
+                PeriodicPatchStepper(cfg, p.bounds, context=context)
             upload.assert_not_called()
 
 
@@ -256,7 +241,7 @@ class NormalBuildGPU(unittest.TestCase):
         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],
+                backend = implementation.FixedPatchStepper(cfg, bounds, context=context,
                                                             coarse_boundary="slot_reservoir")
             with backend:
                 self.assertEqual(len(generated), 2)
@@ -326,8 +311,7 @@ class NormalBuildGPU(unittest.TestCase):
                             for a in (initial.coarse_f, initial.fine_f)))
         max_pop = max_drift = max_moment = 0.0
         with CudaContext(0) as context, PeriodicPatchStepper(
-                LBMConfig(nx=16, ny=16, viscosity=.1), context=context,
-                refinements=[(4, 4, 12, 12)]) as backend:
+                LBMConfig(nx=16, ny=16, viscosity=.1), (4, 4, 12, 12), context=context) as backend:
             backend.initialize(*postcollision(state, spec, maps))
             before = totals(backend.download(), maps)
             for step in range(1, 9):
@@ -358,7 +342,7 @@ 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, PeriodicPatchStepper(cfg, bounds, context=context) 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

tests/validation/multigrid/test_refinement_gpu.py

@@ -84,13 +84,13 @@ class CoupledGPU(unittest.TestCase):
 
         cases = (("dense_unshifted", False, {}),
                  ("blocked_shifted", True,
-                  dict(level_blocks=[(4, 4), (6, 6)])))
+                  dict(base_block=(4, 4), fine_block=(6, 6))))
         with CudaContext(0) as context:
             for label, shifted, storage in cases:
                 with self.subTest(storage=label):
                     cfg = LBMConfig(nx=16, ny=16, viscosity=.0192,
                                     ddf_shifting=shifted)
-                    with FixedPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)],
+                    with FixedPatchStepper(cfg, (4, 4, 12, 12), context=context,
                                            **storage) as backend:
                         self.assertAlmostEqual(backend.patch.tau_coarse, spec.tau_coarse)
                         self.assertAlmostEqual(backend.patch.tau_fine, spec.tau_fine)
@@ -148,7 +148,7 @@ class CoupledGPU(unittest.TestCase):
         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:
+                                      (4, 4, 12, 12), context=context) as backend:
                 backend.initialize(*postcollision(ref.make_state(spec), spec, maps))
                 coarse, fine = backend._levels
 
@@ -187,7 +187,7 @@ class CoupledGPU(unittest.TestCase):
 
         with CudaContext(0) as context:
             with patch.object(compiler, "compile_kernel", audit_compile):
-                backend = PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)])
+                backend = PeriodicPatchStepper(cfg, (4, 4, 12, 12), context=context)
             with backend:
                 # Production allocations: four DDF arrays and four compact tables.
                 device_bytes = 2 * 9 * (16**2 + 24**2) * 4
@@ -270,7 +270,7 @@ class CoupledGPU(unittest.TestCase):
                 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:
+            with PeriodicPatchStepper(cfg, (4, 4, 12, 12), context=context) as plain:
                 rest = postcollision(ref.make_state(spec), spec, maps)
                 plain.initialize(*rest)
                 plain.step()

tests/validation/multigrid/test_refinement_jets.py

@@ -41,9 +41,8 @@ def mocked(objects, jets, fine_block):
             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_jets=[(), tuple(jets)])
+        backend = FixedPatchStepper(config("periodic"), BOUNDS, context=context,
+                                    fine_objects=objects, fine_block=fine_block, jets=jets)
     return backend, htod
 
 
@@ -150,9 +149,8 @@ class JetStepperCPU(unittest.TestCase):
     def test_jet_overlapping_legacy_aperture_rejected_before_cuda(self):
         with patch.object(cuda, "Stream", side_effect=AssertionError("CUDA touched")):
             with self.assertRaisesRegex(ValueError, "SurfaceActuation"):
-                FixedPatchStepper(config("periodic"), context=Mock(), refinements=[BOUNDS],
-                                  level_objects=[(), (circle(actuation(SPEED)),)],
-                                  level_jets=[(), (jet(),)])
+                FixedPatchStepper(config("periodic"), BOUNDS, context=Mock(),
+                                  fine_objects=[circle(actuation(SPEED))], jets=[jet()])
 
 
 def run(mode, fine_block, *, objects, jets, switch=None):
@@ -160,9 +158,8 @@ def run(mode, fine_block, *, objects, jets, switch=None):
     cfg = config(mode)
     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_jets=[(), tuple(jets)]) as backend:
+            cfg, BOUNDS, context=context, coarse_boundary=mode, fine_objects=objects,
+            fine_block=fine_block, jets=jets) as backend:
         backend.initialize(state(NY, NX, 1), state(*patch_.fine_shape, 2))
         if switch is None:
             backend.step(STEPS)

tests/validation/multigrid/test_refinement_mrt_gpu.py

@@ -42,9 +42,8 @@ class RefinementMRTGPU(unittest.TestCase):
                                     collision="MRT", ddf_shifting=shifted)
                     reference = None
                     for block in ((32, 48), (8, 8), (4, 1)):
-                        with FixedPatchStepper(cfg, context=context, coarse_boundary=boundary,
-                                               refinements=[(8, 6, 40, 26)], level_blocks=[block, (8, 8)],
-                                               level_objects=[(), (body,)]) as backend:
+                        with FixedPatchStepper(cfg, (8, 6, 40, 26), context=context, coarse_boundary=boundary,
+                                               base_block=block, fine_block=(8, 8), fine_objects=(body,)) as backend:
                             backend.initialize(*[state(*level.shape[1:]) for level in backend._levels])
                             backend.step(8)
                             raw = backend._download_raw()
@@ -64,8 +63,8 @@ class RefinementMRTGPU(unittest.TestCase):
             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:
+                with FixedPatchStepper(cfg, BOUNDS, context=context, fine_block=fine_block,
+                                       base_block=base_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)
@@ -76,9 +75,9 @@ class RefinementMRTGPU(unittest.TestCase):
     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)
-        kw = dict(refinements=[(4, 4, 12, 12), (8, 8, 16, 16)], level_blocks=[(4, 4)] * 3)
+        kw = dict(fine_block=(4, 4), base_block=(4, 4), child=(8, 8, 16, 16), child_block=(4, 4))
         with CudaContext(0) as context:
-            with FixedPatchStepper(cfg, context=context, **kw) as backend:
+            with FixedPatchStepper(cfg, (4, 4, 12, 12), context=context, **kw) as backend:
                 self.assertEqual(len(backend._level_cfgs), 3)
                 for level, level_cfg in enumerate(backend._level_cfgs):
                     checked = replace(level_cfg)

tests/validation/multigrid/test_refinement_multipatch.py

@@ -26,12 +26,6 @@ FACE = ((4, 4, 8, 12), (12, 4, 16, 12))
 S1 = ((4, 4, 12, 12), (20, 4, 28, 12))
 
 
-def levels(bounds, blocks=(None, None), objects=(), jets=(), **rest):
-    """Two-level constructor keywords: one L1 rectangle or set, per-level blocks, L1 bodies/Jets."""
-    return dict(refinements=[bounds], level_blocks=list(blocks),
-                level_objects=[(), tuple(objects)], level_jets=[(), tuple(jets)], **rest)
-
-
 def level_fine(patch_set, windows):
     """Level-wide fine array with each patch window filled from `windows`."""
     out = np.full((9, *patch_set.fine_shape), np.nan, np.float32)
@@ -51,21 +45,21 @@ class MultipatchCPU(unittest.TestCase):
     def test_rejections_before_cuda(self):
         cfg = config("periodic")
         context = object()  # never activated: every case must fail first
-        for kw in (dict(), dict(blocks=(None, (8, 4))), dict(blocks=((4, 4), None)),
-                   dict(blocks=((4, 4), (8, 4)), objects=[circle()]),
-                   dict(blocks=((4, 4), (5, 4)))):
+        for kw in (dict(), dict(fine_block=(8, 4)), dict(base_block=(4, 4)),
+                   dict(fine_block=(8, 4), base_block=(4, 4), fine_objects=[circle()]),
+                   dict(fine_block=(5, 4), base_block=(4, 4))):
             with self.subTest(kw=kw), self.assertRaises(ValueError):
-                FixedPatchStepper(cfg, context=context, **levels(((4, 4, 8, 12), (14, 4, 18, 12)), **kw))
+                FixedPatchStepper(cfg, ((4, 4, 8, 12), (14, 4, 18, 12)), context=context, **kw)
         with self.assertRaisesRegex(ValueError, "margin"):
-            FixedPatchStepper(cfg, context=context,
-                              **levels(((1, 4, 8, 12), (10, 4, 14, 12)), blocks=((4, 4), (8, 4))))
+            FixedPatchStepper(cfg, ((1, 4, 8, 12), (10, 4, 14, 12)), context=context,
+                              fine_block=(8, 4), base_block=(4, 4))
         from CelerisLab.body.jets import Jet, Parallel, Tube
         cfg32 = LBMConfig(nx=32, ny=16, viscosity=.1)
         span = Jet(Tube((5., 8.), 0., 22., .6), Parallel(0.), amplitude=0.)
         with self.assertRaisesRegex(ValueError, "one jet"):  # same obj_id cannot bypass it
-            FixedPatchStepper(cfg32, context=context,
-                              **levels(((2, 2, 14, 14), (18, 2, 30, 14)), blocks=((4, 4), (4, 4)),
-                                       objects=[body(8., 8.), body(24., 8.)], jets=[span]))
+            FixedPatchStepper(cfg32, ((2, 2, 14, 14), (18, 2, 30, 14)), context=context,
+                              fine_block=(4, 4), base_block=(4, 4),
+                              fine_objects=[body(8., 8.), body(24., 8.)], jets=[span])
 
 
 class MultipatchGPU(unittest.TestCase):
@@ -73,7 +67,7 @@ class MultipatchGPU(unittest.TestCase):
 
     def run_steps(self, context, cfg, bounds, coarse, fine, **kw):
         out = []
-        with FixedPatchStepper(cfg, context=context, **levels(bounds, **kw)) as backend:
+        with FixedPatchStepper(cfg, bounds, context=context, **kw) as backend:
             backend.initialize(coarse, fine)
             for step in self.STEPS:
                 backend.step(step - backend.step_count)
@@ -88,7 +82,7 @@ class MultipatchGPU(unittest.TestCase):
                 p = FixedPatch2D.from_config(cfg, BOUNDS)
                 s = FixedPatchSet2D.from_config(cfg, (BOUNDS,))
                 coarse, fine = state(NY, NX, 1), state(*p.fine_shape, 2)
-                kw = dict(coarse_boundary=mode, blocks=((3, 4), (8, 4)))
+                kw = dict(coarse_boundary=mode, fine_block=(8, 4), base_block=(3, 4))
                 old = self.run_steps(context, cfg, BOUNDS, coarse, fine, **kw)
                 new = self.run_steps(context, cfg, (BOUNDS,), coarse, level_fine(s, [fine]), **kw)
                 for step, a, b in zip(self.STEPS, old, new):
@@ -105,7 +99,7 @@ class MultipatchGPU(unittest.TestCase):
         one = FixedPatchSet2D.from_config(cfg16, (S1[0],))
         two = FixedPatchSet2D.from_config(cfg32, S1)
         coarse, fine = state(16, 16, 3), state(*one.patches[0].fine_shape, 4)
-        kw = dict(blocks=((4, 4), (4, 4)))
+        kw = dict(fine_block=(4, 4), base_block=(4, 4))
         with CudaContext(0) as context:
             ref = self.run_steps(context, cfg16, (S1[0],), coarse, level_fine(one, [fine]), **kw)
             tiled = np.concatenate([coarse, coarse], axis=2)
@@ -124,7 +118,7 @@ class MultipatchGPU(unittest.TestCase):
         s = FixedPatchSet2D.from_config(cfg, FACE)
         coarse = state(16, 24, 5)
         fine = level_fine(s, [state(*p.fine_shape, 6 + i) for i, p in enumerate(s.patches)])
-        kw = dict(blocks=((4, 4), (4, 4)))
+        kw = dict(fine_block=(4, 4), base_block=(4, 4))
         leaf = s.fine_kind == ACTIVE
         support = s.fine_layout((4, 4)).owned
         outer = support & (s.fine_kind == OUTER_GHOST)
@@ -170,7 +164,7 @@ class MultipatchBodiesGPU(unittest.TestCase):
 
     def run_steps(self, context, cfg, bounds, coarse, fine, switch=None, **kw):
         out = []
-        with FixedPatchStepper(cfg, context=context, **levels(bounds, **kw)) as backend:
+        with FixedPatchStepper(cfg, bounds, context=context, **kw) as backend:
             backend.initialize(coarse, fine)
             for step in self.STEPS:
                 if switch is not None and step == 8:
@@ -188,7 +182,8 @@ class MultipatchBodiesGPU(unittest.TestCase):
                 p = FixedPatch2D.from_config(cfg, BOUNDS)
                 s = FixedPatchSet2D.from_config(cfg, (BOUNDS,))
                 coarse, fine = state(NY, NX, 1), state(*p.fine_shape, 2)
-                kw = dict(coarse_boundary=mode, blocks=((3, 4), (8, 4)), objects=[circle()])
+                kw = dict(coarse_boundary=mode, fine_block=(8, 4), base_block=(3, 4),
+                          fine_objects=[circle()])
                 old = self.run_steps(context, cfg, BOUNDS, coarse, fine, **kw)
                 new = self.run_steps(context, cfg, (BOUNDS,), coarse, level_fine(s, [fine]), **kw)
                 for step, a, b in zip(self.STEPS, old, new):
@@ -207,17 +202,17 @@ class MultipatchBodiesGPU(unittest.TestCase):
         two = FixedPatchSet2D.from_config(cfg32, self.PATCHES)
         coarse, fine = state(16, 16, 3), state(*one.patches[0].fine_shape, 4)
         jet = lambda x: Jet(Tube((x, 8.), 0., 6., .6), Parallel(0.), amplitude=.002)
-        kw = dict(blocks=((4, 4), (4, 4)))
+        kw = dict(fine_block=(4, 4), base_block=(4, 4))
         with CudaContext(0) as context:
             for jets in ((), (jet(5.),)):
                 ref = self.run_steps(context, cfg16, self.PATCHES[:1], coarse, level_fine(one, [fine]),
                                      switch=[(0, -.001)] if jets else None,
-                                     objects=[body(8., 8.)], jets=jets, **kw)
+                                     fine_objects=[body(8., 8.)], jets=jets, **kw)
                 both = jets and (jet(5.), jet(21.))
                 got = self.run_steps(context, cfg32, self.PATCHES, np.concatenate([coarse, coarse], 2),
                                      level_fine(two, [fine, fine]),
                                      switch=[(0, -.001), (1, -.001)] if jets else None,
-                                     objects=[body(8., 8., 5), body(24., 8., 5)],
+                                     fine_objects=[body(8., 8., 5), body(24., 8., 5)],
                                      jets=both or (), **kw)
                 for step, a, b in zip(self.STEPS, ref, got):
                     label = f"jets={len(jets)} step {step}"

tests/validation/multigrid/test_refinement_three_level.py

@@ -27,16 +27,7 @@ from test_refinement_blocked_base import raw
 
 CFG = LBMConfig(nx=16, ny=16, viscosity=.1)
 L1, L2 = (4, 4, 12, 12), (8, 8, 16, 16)
-B = (4, 4)
-
-
-def levels(l1=L1, l2=L2, blocks=(B, B, B), objects=((), (), ()), jets=((), (), ())):
-    """Level-list keywords for the scalar L1 patch with one scalar L2 child."""
-    return dict(refinements=[l1, l2], level_blocks=list(blocks),
-                level_objects=list(objects), level_jets=list(jets))
-
-
-KW = levels()
+KW = dict(fine_block=(4, 4), base_block=(4, 4), child=L2, child_block=(4, 4))
 
 
 def child_body(spec=None):
@@ -61,29 +52,26 @@ def mocked_child():
             patch.object(compiler, "load_module", return_value=module), \
             patch.object(cuda, "mem_alloc", side_effect=lambda size: Mock()), \
             patch.object(cuda, "memcpy_htod"):
-        return FixedPatchStepper(CFG, context=context,
-                                 **levels(objects=((), (), (child_body(),)), jets=((), (), (child_jet(),))))
+        return FixedPatchStepper(CFG, L1, context=context, child_objects=[child_body()],
+                                 child_jets=[child_jet()], **KW)
 
 
 class ThreeLevelCPU(unittest.TestCase):
     def test_rejections_before_cuda(self):
         context = object()  # never activated: every case must fail first
-        bad = (levels(blocks=(B, B, None)), levels(blocks=(B, None, B)),
-               levels(objects=((), (circle(),), ())), levels(l2=(4, 8, 16, 16)),
-               levels(blocks=(B, B, (5, 4))))
+        bad = (dict(KW, child_block=None), dict(KW, fine_block=None),
+               dict(KW, fine_objects=[circle()]), dict(KW, child=(4, 8, 16, 16)),
+               dict(KW, child_block=(5, 4)))
         for kw in bad:
             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])
-        for kw in (dict(two_level, level_objects=[(), (), (child_body(),)]),
-                   dict(two_level, level_jets=[(), (), (child_jet(),)]),
-                   levels(jets=((), (), (child_jet(),)))):
+                FixedPatchStepper(CFG, L1, context=context, **kw)
+        for kw in (dict(child_objects=[child_body()]), dict(child_jets=[child_jet()]),
+                   dict(KW, child_jets=[child_jet()])):
             with self.assertRaises(ValueError):
-                FixedPatchStepper(CFG, context=context, **kw)
-        with self.assertRaises(ValueError):  # several L1 patches + scalar child: not yet
-            FixedPatchStepper(LBMConfig(nx=32, ny=16, viscosity=.1), context=context,
-                              **levels(l1=((4, 4, 12, 12), (20, 4, 28, 12))))
+                FixedPatchStepper(CFG, L1, context=context, **kw)
+        with self.assertRaises(ValueError):  # several L1 patches + child: not yet
+            FixedPatchStepper(LBMConfig(nx=32, ny=16, viscosity=.1), ((4, 4, 12, 12), (20, 4, 28, 12)),
+                              context=context, **KW)
 
 
     def test_four_halo_bfl_steps_and_fresh_middle_feeds(self):
@@ -168,7 +156,7 @@ class ThreeLevelGPU(unittest.TestCase):
     def test_equilibrium_is_kept_on_every_level(self):
         from CelerisLab.cuda.context import CudaContext
         feq = np.asarray(compute_feq_d2q9(1.0, .03, -.02), np.float32)[:, None, None]
-        with CudaContext(0) as ctx, FixedPatchStepper(CFG, context=ctx, **KW) as run:
+        with CudaContext(0) as ctx, FixedPatchStepper(CFG, L1, context=ctx, **KW) as run:
             shapes = ((16, 16), run.patch.fine_shape, run.child_patch.fine_shape)
             run.initialize(*(np.broadcast_to(feq, (9, *s)).copy() for s in shapes))
             run.step(8)
@@ -181,7 +169,7 @@ class ThreeLevelGPU(unittest.TestCase):
 
     def test_nonequilibrium_masks_and_fresh_middle_feeds(self):
         from CelerisLab.cuda.context import CudaContext
-        with CudaContext(0) as ctx, FixedPatchStepper(CFG, context=ctx, **KW) as run:
+        with CudaContext(0) as ctx, FixedPatchStepper(CFG, L1, context=ctx, **KW) as run:
             p1, p2 = run.patch, run.child_patch
             run.initialize(state(16, 16, 1), state(*p1.fine_shape, 2), state(*p2.fine_shape, 3))
             f = run._levels[1]
@@ -217,9 +205,8 @@ class ThreeLevelBodyGPU(unittest.TestCase):
             CFG, velocity=0., wall_normal_velocity=.001, wall_slot_width=4,
             y_wall_bc="on_site_velocity", top_wall_bc="reservoir")
         with CudaContext(0) as ctx, FixedPatchStepper(
-                cfg, context=ctx, coarse_boundary=mode,
-                **levels(blocks=(B, B, block), objects=((), (), (child_body(spec),)),
-                         jets=((), (), tuple(jets)))) as run:
+                cfg, L1, context=ctx, coarse_boundary=mode, child_objects=[child_body(spec)],
+                child_jets=jets, **dict(KW, child_block=block)) as run:
             run.initialize(*(state(*level.shape[1:], i + 1) for i, level in enumerate(run._levels)))
             snapshots = []
             for stop in (0, 1, 2, 4, 8):

tests/validation/multigrid/test_refinement_three_level_multipatch_gpu.py

@@ -78,10 +78,10 @@ class ThreeLevelMultipatchGPU(unittest.TestCase):
     def run_case(self, ctx, cfg=CFG, parents=PARENTS, children=CHILDREN,
                  mode='periodic', blocks=((4, 4), (4, 4)), objects=(), drive=(), switch=False,
                  scalar_child=None):
-        child = children if scalar_child is None else scalar_child
-        with FixedPatchStepper(config(cfg, mode), context=ctx, coarse_boundary=mode,
-                refinements=[parents, child], level_blocks=[(4, 4), blocks[0], blocks[1]],
-                level_objects=[(), (), objects], level_jets=[(), (), drive]) as run:
+        kwargs = dict(child_rectangles_in_parent=children) if scalar_child is None else dict(child=scalar_child)
+        with FixedPatchStepper(config(cfg, mode), parents, context=ctx, coarse_boundary=mode,
+                base_block=(4, 4), fine_block=blocks[0], child_block=blocks[1],
+                child_objects=objects, child_jets=drive, **kwargs) as run:
             frame = frames(run)
             run.initialize(*(physical_state(level.shape[1:], *f)
                              for level, f in zip(run._levels, frame)))

tests/validation/multigrid/test_runtime_child_remap_gpu.py

@@ -26,8 +26,9 @@ FP64_ATOL = 3e-6
 
 
 def make(ctx):
-    return FixedPatchStepper(CFG, context=ctx, refinements=[PARENTS, A],
-                             level_blocks=[(4, 4)] * 3)
+    return FixedPatchStepper(CFG, PARENTS, context=ctx, base_block=(4, 4),
+                             fine_block=(4, 4), child_block=(4, 4),
+                             child_rectangles_in_parent=A)
 
 
 def initial(run, uniform):

tests/validation/multigrid/test_shifted_jets_gpu.py

@@ -136,8 +136,8 @@ class ShiftedJetsGPU(unittest.TestCase):
                   Parallel(0.),amplitude=.013)
         for rho in (.99,1.01):
             with self.subTest(rho=rho),CudaContext(0) as ctx:
-                with FixedPatchStepper(cfg,context=ctx,refinements=[(8,6,24,18)],level_blocks=[(4,4)]*2,
-                        level_objects=[(),(obj,)],level_jets=[(),(whole,)],observe_raw_exchange=True) as run:
+                with FixedPatchStepper(cfg,(8,6,24,18),context=ctx,base_block=(4,4),fine_block=(4,4),
+                        fine_objects=(obj,),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())
@@ -153,8 +153,7 @@ class ShiftedJetsGPU(unittest.TestCase):
         obj=circle((16.,12.),2.15);j=drive((16.,12.),2.15)
         for rho in (.99,1.01):
             with self.subTest(rho=rho),CudaContext(0) as ctx,ExitStack() as owned:
-                settings=dict(context=ctx,refinements=[(8,6,24,18)],level_blocks=[(4,4)]*2,
-                              level_objects=[(),(obj,)],level_jets=[(),(j,)])
+                settings=dict(bounds=(8,6,24,18),context=ctx,base_block=(4,4),fine_block=(4,4),fine_objects=(obj,),jets=(j,))
                 measured=owned.enter_context(FixedPatchStepper(cfg,observe_raw_exchange=True,**settings))
                 plain=owned.enter_context(FixedPatchStepper(cfg,**settings))
                 measured._initialize_raw(initial(measured,rho));plain._initialize_raw(initial(plain,rho))
@@ -187,9 +186,8 @@ class ShiftedJetsGPU(unittest.TestCase):
         parents=((4,4,20,28),(24,4,44,28));a=((16,16,32,40),(56,16,72,40));b=(a[0],(60,16,76,40))
         obj=circle((12.,14.),.8);j=drive((12.,14.),.8)
         with CudaContext(0) as ctx,ExitStack() as owned:
-            old=owned.enter_context(FixedPatchStepper(cfg,context=ctx,refinements=[parents,a],
-                level_blocks=[(4,4)]*3,level_objects=[(),(),(obj,)],level_jets=[(),(),(j,)],
-                observe_raw_exchange=True))
+            old=owned.enter_context(FixedPatchStepper(cfg,parents,context=ctx,base_block=(4,4),fine_block=(4,4),
+                child_block=(4,4),child_rectangles_in_parent=a,child_objects=(obj,),child_jets=(j,),observe_raw_exchange=True))
             old._initialize_raw(initial(old,.99));ledger=Ledger(self,old,2)
             for speed in (.002,-.001):old.set_jet_amplitude(0,speed,level=2);ledger.step()
             before=old._download_raw();count=old.step_count

tests/validation/multigrid/test_shifted_remap_gpu.py

@@ -22,8 +22,9 @@ B=((20,16,36,40),)
 
 
 def make(ctx):
-    return FixedPatchStepper(CFG,context=ctx,refinements=[PARENTS,A],
-                             level_blocks=[(4,4)]*3)
+    return FixedPatchStepper(CFG,PARENTS,context=ctx,base_block=(4,4),
+                             fine_block=(4,4),child_block=(4,4),
+                             child_rectangles_in_parent=A)
 
 
 def initial(run):

tests/validation/run_dfg_re20.py

@@ -57,6 +57,8 @@ WINDOW_TSTAR = 10.0            # trailing 5 s, 21 samples
 BANDS = dict(cd=(5.57, 5.59), cl=(0.0104, 0.0110), p_diff=(0.1172, 0.1176))
 STEADY_RANGE = {k: (hi - lo) / 10 for k, (lo, hi) in BANDS.items()}
 STEADY_SLOPE = {k: (hi - lo) / 50 for k, (lo, hi) in BANDS.items()}
+# 2: {schema, provenance, cases}; the earlier receipts (schema 1) are a bare list of cases.
+RESULT_SCHEMA = 2
 DEVIATIONS = ("pressure-NEQ outlet (neq_extrap), not do-nothing", "local Zou-He inlet")
 
 _re100_config = rr.config
@@ -106,6 +108,21 @@ def summarize(rows):
                 verdict=inside)
 
 
+def read_result(path):
+    """Cases and provenance from a result.json of either schema.
+
+    Schema 1 (receipts before RESULT_SCHEMA=2, e.g. evidence 70f6617) is a bare list of
+    cases without provenance; schema 2 is {schema, provenance, cases}. Readers that
+    indexed [0] on schema 1 must use read_result(path)["cases"][0].
+    """
+    data = json.loads(Path(path).read_text())
+    if isinstance(data, list):
+        return dict(schema=1, provenance=None, cases=data)
+    if not isinstance(data, dict) or data.get("schema") != RESULT_SCHEMA:
+        raise ValueError(f"unknown DFG Re20 result schema: {data.get('schema') if isinstance(data, dict) else type(data).__name__}")
+    return dict(schema=RESULT_SCHEMA, provenance=data["provenance"], cases=data["cases"])
+
+
 def healthy(row):
     return row["max_mach"] <= 0.2 and row["rho_min"] >= 0.5 and row["rho_max"] <= 1.5
 
@@ -126,8 +143,8 @@ def run_case(a, h, case):
             cfg, pair, body, host = rr.host_geometry(a.source, h, a.collision)
             context = stack.enter_context(CudaContext(0))
             backend = stack.enter_context(FixedPatchStepper(
-                cfg, context=context, coarse_boundary="channel", refinements=[rr.PLACEMENTS["wide"]],
-                level_blocks=[rr.BASE_BLOCK, rr.FINE_BLOCK], level_objects=[(), (body,)],
+                cfg, rr.PLACEMENTS["wide"], context=context, coarse_boundary="channel",
+                base_block=rr.BASE_BLOCK, fine_block=rr.FINE_BLOCK, fine_objects=(body,),
                 observe_raw_exchange=True, observe_window_exchange=True))
             from CelerisLab.lbm.channel import channel_masks
             from CelerisLab.lbm.refinement import ACTIVE
@@ -194,6 +211,27 @@ def plan(collision="SRT"):
                 label="compared against the official reference bands (non-canonical neq_extrap outlet, zou_he_local inlet); not a full benchmark certification")
 
 
+def provenance(source):
+    """Current checkout identity: HEAD, dirty flag, src tree and this runner's hash."""
+    import hashlib
+    import subprocess
+    from source_provenance import checkout_provenance
+    tree = subprocess.check_output(["git", "-C", str(source), "rev-parse", "HEAD:src"],
+                                   text=True).strip()
+    record = dict(source_mode="current_checkout", **checkout_provenance(source), src_tree=tree,
+                  runner_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest())
+    if record["checkout_dirty"]:  # HEAD would not describe the code that runs
+        raise RuntimeError("DFG Re20 runs require a clean checkout")
+    return record
+
+
+def reject_output_inside(output, source):
+    """Receipts inside the checkout would make it dirty and change the recorded source."""
+    output = Path(output).resolve()
+    if output == source or source in output.parents:
+        raise SystemExit(f"--output must be outside the source checkout {source}")
+
+
 def main():
     parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
     parser.add_argument("--case", choices=["refined", "uniform", "both"], default="refined")
@@ -204,8 +242,10 @@ def main():
     parser.add_argument("--max-wall-seconds", type=float)
     a = parser.parse_args()
     a.source = a.source.resolve()
+    reject_output_inside(a.output, a.source)
+    record = dict(plan(a.collision), schema=RESULT_SCHEMA, provenance=provenance(a.source))
     a.output.mkdir(parents=True, exist_ok=False)
-    (a.output / "plan.json").write_text(json.dumps(plan(a.collision), indent=2, allow_nan=False))
+    (a.output / "plan.json").write_text(json.dumps(record, indent=2, allow_nan=False))
     if a.plan:
         return
     if not a.max_wall_seconds or a.max_wall_seconds <= 0 or not os.environ.get("CUDA_VISIBLE_DEVICES"):
@@ -215,7 +255,12 @@ def main():
     a.deadline = time.monotonic() + a.max_wall_seconds
     cases = ["refined", "uniform"] if a.case == "both" else [a.case]
     results = [run_case(a, h, case) for case in cases]
-    (a.output / "result.json").write_text(json.dumps(results, indent=2, allow_nan=False))
+    after = provenance(a.source)
+    if after != record["provenance"]:
+        raise RuntimeError("Source checkout changed during the run")
+    (a.output / "result.json").write_text(json.dumps(dict(schema=RESULT_SCHEMA, provenance=after,
+                                                          cases=results),
+                                                     indent=2, allow_nan=False))
 
 
 if __name__ == "__main__":

tests/validation/run_jet_circle_cv.py

@@ -3,6 +3,7 @@
 
 import argparse
 import hashlib
+import subprocess
 import json
 import os
 from pathlib import Path
@@ -220,6 +221,60 @@ def verify(repo):
     return manifest
 
 
+def verify_historical(repo):
+    """The pinned manifest describes the PIN tree (read with git, not the working tree)."""
+    from source_provenance import verify_manifest
+    manifest = json.loads((HERE / "data/jet_circle_cv/source_manifest.json").read_text())
+    if manifest["revision"] != PIN:
+        raise RuntimeError("Wrong pinned revision")
+    verify_manifest(repo, PIN, manifest["files"])
+    return manifest
+
+
+def current_checkout(repo):
+    """Provenance for a run on the current checkout; never labelled as the pinned source."""
+    files = subprocess.check_output(["git", "-C", str(repo), "ls-files", "src"],
+                                    text=True).splitlines()
+    return dict(source_mode="current_checkout",
+                source_capture="current checkout, not the frozen pinned qualification",
+                source_revision=checkout_provenance(repo)["checkout_revision"],
+                src_tree=subprocess.check_output(["git", "-C", str(repo), "rev-parse",
+                                                  "HEAD:src"], text=True).strip(),
+                src_sha256={f: hashlib.sha256((repo / f).read_bytes()).hexdigest()
+                            for f in files})
+
+
+def source_identity(repo, current):
+    """Recorded source of a run; recomputed after the run by finish_receipt."""
+    if current:
+        provenance = dict(**checkout_provenance(repo), **current_checkout(repo))
+        if provenance["checkout_dirty"]:
+            raise RuntimeError("Current-checkout runs require a clean checkout")
+        return provenance
+    source = verify(repo)
+    return dict(source_experiment=source["experiment"], source_mode="pinned",
+                **checkout_provenance(repo))
+
+
+def finish_receipt(path, payload, provenance, recheck):
+    """Mark the receipt COMPLETE only if the source is unchanged; otherwise INVALID.
+
+    Receipts written during the run carry run_status="INCOMPLETE", so an interrupted
+    run never leaves a complete-looking receipt.
+    """
+    from source_provenance import final_source_check
+    fields = final_source_check(provenance, recheck)
+    unchanged = fields["source_recheck"] == "UNCHANGED"
+    payload = dict(payload, **fields,
+                   run_status="COMPLETE" if unchanged else "INVALID_SOURCE_CHANGED")
+    temporary = Path(str(path) + ".tmp")
+    temporary.write_text(json.dumps(payload, indent=2) + "\n")
+    temporary.replace(path)
+    if not unchanged:
+        raise RuntimeError("Source changed during the run; receipt marked invalid")
+    return payload
+
+
 def run_case(c, ctx, output, provenance, continuation=False):
     from CelerisLab.body.geometry.circle import CircleGeometry
     from CelerisLab.body.objects import SimObject
@@ -279,11 +334,12 @@ def run_case(c, ctx, output, provenance, continuation=False):
         PeriodicBlockStepper(fine_cfg, layout, context=ctx, objects=[fobj]) as uniform,
         FixedPatchStepper(
             cfg,
+            c["bounds"],
             context=ctx,
-            refinements=[c["bounds"]],
-            level_blocks=[(8, 8), (8, 8)],
-            level_objects=[(), (obj,)],
-            level_jets=[(), tuple(jets)],
+            base_block=(8, 8),
+            fine_block=(8, 8),
+            fine_objects=[obj],
+            jets=jets,
             observe_raw_exchange=True,
         ) as amr,
     ):
@@ -529,16 +585,16 @@ def run_case(c, ctx, output, provenance, continuation=False):
                     recorder.flush()
                 payload = dict(
                     case=c,
-                    source_revision=PIN,
-                    **provenance,
+                    **{"source_revision": PIN, **provenance},
                     status="PILOT_DIAGNOSTIC_ONLY",
+                    run_status="INCOMPLETE",  # final source recheck in main() completes it
                     windows=rows,
                     physics_validation_pass=False,
                     precision="FP32 evolution, FP64 host diagnostics",
                 )
-                (output / f"re{c['re_d']}_s{c['scale']}.json").write_text(
-                    json.dumps(payload, indent=2) + "\n"
-                )
+                receipt = output / f"re{c['re_d']}_s{c['scale']}.json"
+                Path(str(receipt) + ".tmp").write_text(json.dumps(payload, indent=2) + "\n")
+                Path(str(receipt) + ".tmp").replace(receipt)
                 print(
                     json.dumps(
                         dict(
@@ -570,6 +626,9 @@ def main():
         help="Fixed measurement campaign: Re100, scale1/2, 3200 reference steps",
     )
     p.add_argument("--plan", action="store_true")
+    p.add_argument("--current-checkout", action="store_true",
+                   help="Run on the current clean checkout and record its provenance "
+                        "instead of requiring the frozen pinned source")
     a = p.parse_args()
     if a.steps <= 0 or a.steps > 3200:
         p.error("steps must be 1..3200")
@@ -594,7 +653,8 @@ def main():
         )
     plan = dict(
         case=c,
-        required_revision=PIN,
+        required_revision=None if a.current_checkout else PIN,
+        source_mode="current_checkout" if a.current_checkout else "pinned",
         required_free_device_mib=1024,
         estimated_peak_device_mib=128 * a.scale**2,
         estimated_output_mib=64 if a.continuation else 2,
@@ -610,10 +670,9 @@ def main():
     ):
         p.error("Root must select one verified idle GPU via CUDA_VISIBLE_DEVICES")
     repo = a.repo.resolve()
-    source = verify(repo)
-    provenance = dict(
-        source_experiment=source["experiment"], **checkout_provenance(repo)
-    )
+    from source_provenance import reject_output_inside
+    reject_output_inside(a.output, repo)
+    provenance = source_identity(repo, a.current_checkout)
     sys.path.insert(0, str(repo / "src"))
     import CelerisLab
 
@@ -629,7 +688,9 @@ def main():
             free, total = cuda.mem_get_info()
         if free < max(1024, 2 * plan["estimated_peak_device_mib"]) * 1024**2:
             raise RuntimeError("Insufficient free device memory")
-        run_case(c, ctx, a.output, provenance, continuation=a.continuation)
+        payload = run_case(c, ctx, a.output, provenance, continuation=a.continuation)
+    finish_receipt(a.output / f"re{c['re_d']}_s{c['scale']}.json", payload, provenance,
+                   lambda: source_identity(repo, a.current_checkout))
     print(
         json.dumps(
             dict(status="COMPLETE_DIAGNOSTIC_ONLY", physics_validation_pass=False)

tests/validation/run_refined_dfg.py

@@ -383,12 +383,13 @@ def run_case(a, h, periodic, case):
             backend = stack.enter_context(
                 FixedPatchStepper(
                     cfg,
+                    PLACEMENTS[a.placement],
                     context=context,
                     coarse_boundary="channel",
-                    refinements=[PLACEMENTS[a.placement]],
-                    level_blocks=[BASE_BLOCK, FINE_BLOCK],
-                    level_objects=[(), (body,)],
-                    level_jets=[(), public_jets((80.5, 81.0), 20.0, a.jet_scenario)],
+                    base_block=BASE_BLOCK,
+                    fine_block=FINE_BLOCK,
+                    fine_objects=(body,),
+                    jets=public_jets((80.5, 81.0), 20.0, a.jet_scenario),
                     observe_raw_exchange=True,
                     **({"observe_window_exchange": True} if a.observation_mode == "window" else {}),
                 )

tests/validation/source_provenance.py

@@ -13,3 +13,99 @@ def checkout_provenance(repo):
         text=True,
     )
     return dict(checkout_revision=revision, checkout_dirty=bool(dirty))
+
+
+def final_source_check(recorded, recheck):
+    """Re-derive the source identity after a run and compare it with the recorded one.
+
+    Returns receipt fields: source_recheck "UNCHANGED"; "CHANGED" with source_after; or
+    "RECHECK_FAILED" (source_after None) with the error. Callers mark a receipt complete
+    only for "UNCHANGED"; anything else invalidates the run.
+    """
+    try:
+        after = recheck()
+    except Exception as error:  # a failed recheck cannot confirm the source: invalid too
+        return dict(source_recheck="RECHECK_FAILED", source_after=None,
+                    source_recheck_error=f"{type(error).__name__}: {error}")
+    if after == recorded:
+        return dict(source_recheck="UNCHANGED")
+    return dict(source_recheck="CHANGED", source_after=after, source_recheck_error=None)
+
+
+def reject_output_inside(output, repo):
+    """Refuse run outputs inside the checkout before any work: they would dirty it."""
+    from pathlib import Path
+    output, repo = Path(output).resolve(), Path(repo).resolve()
+    if output == repo or repo in output.parents:
+        raise SystemExit(f"--output must be outside the source checkout {repo}")
+
+
+class HistoryUnavailable(RuntimeError):
+    """The clone cannot show the recorded commit (no git, shallow clone, installed copy)."""
+
+
+def historical_sha256(repo, commit, paths):
+    """SHA-256 of each path as stored in `commit`, read with git, never from the working tree.
+
+    Raises HistoryUnavailable when git or the commit object is missing, so callers can
+    report an unchecked integrity record explicitly instead of passing or failing silently.
+    """
+    import hashlib
+    git = ["git", "-C", str(repo)]
+    try:
+        subprocess.run([*git, "cat-file", "-e", f"{commit}^{{commit}}"], check=True,
+                       capture_output=True)
+    except (OSError, subprocess.CalledProcessError) as error:
+        raise HistoryUnavailable(f"commit {commit} is not available in {repo}") from error
+    digests = {}
+    for path in paths:
+        listed = subprocess.run([*git, "ls-tree", "--name-only", commit, "--", path],
+                                capture_output=True, text=True)
+        if listed.returncode:
+            raise HistoryUnavailable(f"tree of {commit} is not available in {repo}")
+        if listed.stdout.strip() != path:
+            raise ValueError(f"{path} is not part of {commit}")
+        shown = subprocess.run([*git, "show", f"{commit}:{path}"], capture_output=True)
+        if shown.returncode:  # e.g. a blobless partial clone without network access
+            raise HistoryUnavailable(f"content of {path} at {commit} is not available in {repo}")
+        digests[path] = hashlib.sha256(shown.stdout).hexdigest()
+    return digests
+
+
+def verify_manifest(repo, commit, digests):
+    """Raise RuntimeError naming the first path whose recorded digest differs at `commit`."""
+    actual = historical_sha256(repo, commit, digests)
+    for path, digest in digests.items():
+        if actual[path] != digest:
+            raise RuntimeError(f"Historical manifest mismatch at {commit}: {path}")
+    return actual
+
+
+def historical_tree(repo, commit, prefix="src"):
+    """Tracked file names under `prefix` at `commit` (HistoryUnavailable if missing)."""
+    historical_sha256(repo, commit, [])
+    return subprocess.check_output(["git", "-C", str(repo), "ls-tree", "-r", "--name-only",
+                                    commit, "--", prefix], text=True).splitlines()
+
+
+def historical_files(repo, commit, paths, dest):
+    """Write each path as stored in `commit` under `dest` (HistoryUnavailable if missing)."""
+    from pathlib import Path
+    historical_sha256(repo, commit, paths)  # HistoryUnavailable / ValueError before writing
+    for path in paths:
+        shown = subprocess.run(["git", "-C", str(repo), "show", f"{commit}:{path}"],
+                               capture_output=True)
+        if shown.returncode:
+            raise HistoryUnavailable(f"cannot read {commit}:{path} (blob not in this clone)")
+        target = Path(dest) / path
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_bytes(shown.stdout)
+
+
+def history_unavailable(testcase, error):
+    """Skip with the reason, or fail when CELERIS_REQUIRE_HISTORY=1 (release gating)."""
+    import os
+    message = f"historical integrity NOT checked: {error}"
+    if os.environ.get("CELERIS_REQUIRE_HISTORY") == "1":
+        raise AssertionError(message)
+    testcase.skipTest(message)

tests/validation/taylor_green/assess_projected_pairs.py

@@ -73,7 +73,31 @@ def valid_builds(record, n, depth):
     return True
 
 
-def assess(records, expected_revision=None):
+# run_case.py (sha256) of the runners before the final source recheck (git: 9805383,
+# 2cf9853 = the a135252 qualification runner, db3d098); they wrote schema 1 receipts.
+LEGACY_RUN_CASE_SHA256 = {
+    "9805383": "fbceab8c1aaaf357195a4fb9e6a699a195326e480e274692a277ce3df54ee6d9",
+    "2cf9853": "0789d2e1ffa581078bd11e6251b660b2c268363171fda0c6b26ac41205f1de3a",
+    "db3d098": "618c14ba94584aa9c0b714e0e2064a746d2b882973a13424f09dd7e54c68f7f1",
+}
+
+
+def rechecked(r, current):
+    """Schema 2 receipts need source_recheck UNCHANGED; a missing recheck is accepted only
+    for frozen-mode schema 1 receipts written by a verified legacy runner."""
+    if "invalid" in r:
+        return False
+    if r.get("schema") == 2:
+        return r.get("source_recheck") == "UNCHANGED"
+    harness = r.get("source", {}).get("harness_sha256") or {}
+    return (not current and r.get("schema") == 1 and "source_recheck" not in r
+            and harness.get("run_case.py") in LEGACY_RUN_CASE_SHA256.values())
+
+
+def assess(records, expected_revision=None, current=False):
+    """current=True assesses current-checkout receipts (run_case --current-checkout) with
+    the same gates; it never accepts them as the frozen a135252 qualification, and the
+    frozen mode never accepts current-checkout receipts."""
     indexed, duplicates = {}, []
     for record in records:
         key = tuple(record.get(k) for k in ("re_L", "n", "depth", "explosion"))
@@ -87,11 +111,19 @@ def assess(records, expected_revision=None):
     sources = [r.get("source", {}) for r in records]
     checks["same_source_and_harness"] = bool(sources) and len({
         json.dumps(s, sort_keys=True) for s in sources}) == 1
-    checks["clean_a135252_numerical_source"] = bool(sources) and all(
-        s.get("dirty") is False and s.get("source_tree_sha256") == SOURCE_SHA256
-        and bool(s.get("revision")) and bool(s.get("harness_sha256"))
-        and s.get("build_capture_sha256") == BUILD_CAPTURE_SHA256
-        and (expected_revision is None or s.get("revision") == expected_revision) for s in sources)
+    if current:
+        checks["clean_current_checkout_source"] = bool(sources) and expected_revision is not None and all(
+            s.get("dirty") is False and s.get("source_mode") == "current_checkout"
+            and bool(s.get("source_tree_sha256")) and bool(s.get("harness_sha256"))
+            and s.get("build_capture_sha256") == BUILD_CAPTURE_SHA256
+            and s.get("revision") == expected_revision for s in sources)
+    else:
+        checks["clean_a135252_numerical_source"] = bool(sources) and all(
+            s.get("dirty") is False and s.get("source_tree_sha256") == SOURCE_SHA256
+            and "source_mode" not in s
+            and bool(s.get("revision")) and bool(s.get("harness_sha256"))
+            and s.get("build_capture_sha256") == BUILD_CAPTURE_SHA256
+            and (expected_revision is None or s.get("revision") == expected_revision) for s in sources)
     runtimes = [r.get("runtime_identity") for r in records]
     checks["same_recorded_runtime"] = bool(runtimes) and all(
         isinstance(v, dict) and all(v.get(k) for k in
@@ -125,8 +157,15 @@ def assess(records, expected_revision=None):
             and r.get("mach") == U * math.sqrt(3)
         )
         checks[label + " actual_builds"] = valid_builds(r, n, depth)
-        checks[label + " reference"] = (r.get("numerical_revision") == NUMERICAL_REVISION
-                                         and r.get("reference_harness") == REFERENCE_HARNESS)
+        if current:
+            checks[label + " reference"] = (r.get("numerical_revision") is None
+                                             and r.get("source_mode") == "current_checkout"
+                                             and r.get("reference_harness") == REFERENCE_HARNESS)
+        else:
+            checks[label + " reference"] = (r.get("numerical_revision") == NUMERICAL_REVISION
+                                             and "source_mode" not in r
+                                             and r.get("reference_harness") == REFERENCE_HARNESS)
+        checks[label + " source_recheck"] = rechecked(r, current)
         checks[label + " numerics"] = r.get("numerics") == NUMERICS and r.get("initialization") == INITIALIZATION
         runtime = r.get("runtime_source", {})
         package = Path(runtime.get("checkout", "")) / "src/CelerisLab"
@@ -185,7 +224,10 @@ def assess(records, expected_revision=None):
                                      valid_case_and_controls=valid, baseline_passed=baseline_passed)
     return dict(passed=all(checks.values()), missing=missing, duplicates=duplicates, checks=checks,
                 comparisons=comparisons, scheme_results=scheme_results, early_pressure=early,
-                numerical_revision=NUMERICAL_REVISION, source_tree_sha256=SOURCE_SHA256,
+                source_mode="current_checkout" if current else "frozen",
+                numerical_revision=None if current else NUMERICAL_REVISION,
+                source_tree_sha256=(sources[0].get("source_tree_sha256") if current and sources
+                                    else None if current else SOURCE_SHA256),
                 reference_harness=REFERENCE_HARNESS, expected_revision=expected_revision,
                 floors=FLOORS, baseline_factor=BASELINE_FACTOR,
                 main_statistic="maximum of four post-initial samples at t*=.12,.24,.36,.48",
@@ -199,10 +241,13 @@ if __name__ == "__main__":
     p.add_argument("files", type=Path, nargs="+")
     p.add_argument("--output", type=Path, required=True)
     p.add_argument("--expected-revision", required=True)
+    p.add_argument("--current-checkout", action="store_true",
+                   help="assess current-checkout receipts; never the frozen qualification")
     args = p.parse_args()
     if args.output.exists():
         p.error("Output already exists; choose a new path to retain prior evidence")
-    result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision)
+    result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision,
+                    current=args.current_checkout)
     args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n")
     print(json.dumps(result, indent=2, allow_nan=False))
     raise SystemExit(0 if result["passed"] else 1)

tests/validation/taylor_green/run_case.py

@@ -64,8 +64,16 @@ def provenance(repo):
     )
 
 
-def verify_source(repo, revision):
-    """Exact clean harness commit, fixed production tree, and reviewed capture helper."""
+CURRENT_CAPTURE = "current checkout, not the frozen a135252 qualification"
+
+
+def verify_source(repo, revision, current=False):
+    """Exact clean harness commit, fixed production tree, and reviewed capture helper.
+
+    current=True records the clean checkout itself instead of requiring the frozen
+    numerical tree; such receipts carry source_mode="current_checkout" and are only
+    assessed with assess(..., current=True), never as the frozen qualification.
+    """
     import oracle
     from tests.validation.multigrid import check_projected_pairs as gate
 
@@ -77,6 +85,17 @@ def verify_source(repo, revision):
     if (Path(gate.__file__).resolve() != helper
             or hashlib.sha256(helper.read_bytes()).hexdigest() != BUILD_CAPTURE_SHA256):
         raise RuntimeError("Wrong build-capture helper")
+    if current:
+        result = provenance(repo)
+        if result["revision"] != revision or result["dirty"]:
+            raise ValueError("requires the exact expected clean checkout")
+        result.update(source_mode="current_checkout", source_capture=CURRENT_CAPTURE,
+                      build_capture_sha256=BUILD_CAPTURE_SHA256)
+        return result
+    if subprocess.run(["git", "-C", str(repo), "cat-file", "-e", f"{NUMERICAL_REVISION}^{{commit}}"],
+                      capture_output=True).returncode:
+        raise RuntimeError(f"frozen numerical revision {NUMERICAL_REVISION} is not available "
+                           "in this clone; the frozen qualification cannot be checked")
     if gate.preflight(repo, revision, "candidate") != NUMERICAL_REVISION:
         raise RuntimeError("Wrong production revision")
     result = provenance(repo)
@@ -149,6 +168,8 @@ def main():
     p.add_argument("--depth", type=int, choices=[1, 2, 3], required=True)
     p.add_argument("--output", type=Path, required=True)
     p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant")
+    p.add_argument("--current-checkout", action="store_true",
+                   help="record the clean current checkout instead of the frozen a135252 tree")
     p.add_argument("--early-pressure", action="store_true",
                    help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64")
     args = p.parse_args()
@@ -161,7 +182,9 @@ def main():
         p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required")
     repo = args.repo.resolve()
     sys.path[:0] = [str(repo / "src"), str(repo)]
-    source = verify_source(repo, args.expected_revision)
+    from tests.validation.source_provenance import reject_output_inside
+    reject_output_inside(args.output, repo)
+    source = verify_source(repo, args.expected_revision, current=args.current_checkout)
     runtime = runtime_source(repo)
     if args.dry_run:
         print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True)))
@@ -180,9 +203,10 @@ def main():
     masks = [k == ACTIVE for k in h.kinds]
     main_steps, early_steps = sample_steps(args.n, args.early_pressure)
     report = dict(
-        schema=1,
+        schema=2,  # 2: final source recheck (source_recheck) is recorded
         source=source,
-        numerical_revision=NUMERICAL_REVISION,
+        numerical_revision=None if args.current_checkout else NUMERICAL_REVISION,
+        **({"source_mode": "current_checkout"} if args.current_checkout else {}),
         reference_harness=REFERENCE_HARNESS,
         actual_build_sidecars=[],
         runtime_identity=None,
@@ -297,6 +321,17 @@ def main():
                         ),
                         flush=True,
                     )
+    from tests.validation.source_provenance import final_source_check
+    recheck = final_source_check(
+        dict(source=source, runtime_source=runtime),
+        lambda: dict(source=verify_source(repo, args.expected_revision,
+                                          current=args.current_checkout),
+                     runtime_source=runtime_source(repo)))
+    report.update(recheck)
+    if recheck["source_recheck"] != "UNCHANGED":
+        report.update(invalid="SOURCE_CHANGED_DURING_RUN")  # completed stays False
+        save()
+        raise RuntimeError("Source changed during the run; receipt marked invalid")
     report.update(
         completed=True,
         seconds=time.monotonic() - started,

tests/validation/taylor_green/test_projected_pairs.py

@@ -12,13 +12,13 @@ import sys
 import tempfile
 from types import SimpleNamespace
 import unittest
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 import numpy as np
 
 import run_case
 from assess_projected_pairs import (BUILD_SOURCES, CASES, NUMERICS, NUMERICAL_REVISION,
-                                    SOURCE_SHA256, assess)
+                                    LEGACY_RUN_CASE_SHA256, SOURCE_SHA256, assess)
 from oracle import LIMITS, TIMES, U, rectangles
 
 REPO = Path(__file__).resolve().parents[3]
@@ -31,7 +31,7 @@ RUNTIME = dict(device=dict(name="host mock", pci_bus_id="mock", capability=[7, 0
 
 def source():
     return dict(revision=MOCK_REVISION, dirty=False, source_tree_sha256=SOURCE_SHA256,
-                harness_sha256={"run_case.py": "a" * 64},
+                harness_sha256={"run_case.py": LEGACY_RUN_CASE_SHA256["2cf9853"]},
                 build_capture_sha256=run_case.BUILD_CAPTURE_SHA256)
 
 
@@ -53,7 +53,7 @@ def records(early=False):
         h = build_hierarchy(cfg, refinements)
         main_steps, early_steps = run_case.sample_steps(n, early)
         out.append(dict(
-            re_L=re, n=n, depth=depth, explosion=scheme, U=U, mach=float(U*np.sqrt(3)),
+            schema=1, re_L=re, n=n, depth=depth, explosion=scheme, U=U, mach=float(U*np.sqrt(3)),
             viscosity=cfg.viscosity, tau_by_level=[0.5+3*c.viscosity for c in h.configs],
             refinements=refinements, numerics=NUMERICS.copy(), initialization=run_case.INITIALIZATION,
             source=source(), numerical_revision=NUMERICAL_REVISION,
@@ -249,41 +249,19 @@ class ProjectedPairsTests(unittest.TestCase):
         self.assertFalse(assess(incomplete)["comparisons"][3]["passed"])
 
     def test_pinned_build_sources_match_host_generated_inputs(self):
-        """Check the pinned fingerprints using NUMERICAL_REVISION's renderer and kernels.
-        The imported LBMConfig is unchanged from that revision; this is a host
-        build-input check, not a reconstruction of the historical CUDA toolchain."""
-        import importlib.util
-        import tarfile
+        from CelerisLab.cuda import compiler_v2
 
-        kernels = "src/CelerisLab/lbm/kernels"
-        renderer = "src/CelerisLab/cuda/compiler_v2.py"
-        if subprocess.run(["git", "-C", str(REPO), "cat-file", "-e",
-                           f"{NUMERICAL_REVISION}^{{commit}}"], capture_output=True).returncode:
-            self.fail(f"{NUMERICAL_REVISION} is not in this clone; the pinned build "
-                      "sources cannot be checked")
-        archive = subprocess.run(["git", "-C", str(REPO), "archive", "--format=tar",
-                                  NUMERICAL_REVISION, kernels, renderer],
-                                 capture_output=True, check=True).stdout
-        with tempfile.TemporaryDirectory() as snapshot:
-            with tarfile.open(fileobj=io.BytesIO(archive)) as tar:
-                tar.extractall(snapshot)
-            spec = importlib.util.spec_from_file_location(
-                "CelerisLab.cuda._compiler_a135252", Path(snapshot) / renderer)
-            compiler = importlib.util.module_from_spec(spec)
-            spec.loader.exec_module(compiler)
-            with patch.object(compiler, "_KERNEL_DIR", str(Path(snapshot) / kernels)):
-                for size in (128, 256):
-                    with self.subTest(size=size):
-                        cfg = replace(run_case.config(size, 100), nx=10, ny=10)
-                        cfg.validate()
-                        root = compiler.generate_config(cfg)
-                        try:
-                            actual = json.loads((Path(root) / ".celeris-build").read_text())
-                            expected = build_metadata(size)
-                            self.assertEqual(actual, {key: expected[key] for key in
-                                                      ("source_sha256", "config", "n_objects")})
-                        finally:
-                            compiler.cleanup_build_dir(root)
+        for size in (128, 256):
+            cfg = replace(run_case.config(size, 100), nx=10, ny=10)
+            cfg.validate()
+            root = compiler_v2.generate_config(cfg)
+            try:
+                actual = json.loads((Path(root) / ".celeris-build").read_text())
+                expected = build_metadata(size)
+                self.assertEqual(actual, {key: expected[key] for key in
+                                          ("source_sha256", "config", "n_objects")})
+            finally:
+                compiler_v2.cleanup_build_dir(root)
 
     def test_source_preflight_requires_clean_exact_revision_and_tree(self):
         import oracle
@@ -313,6 +291,105 @@ class ProjectedPairsTests(unittest.TestCase):
                 else:
                     self.assertEqual(run_case.verify_source(REPO, expected), source())
 
+    def test_frozen_mode_reports_missing_history_not_a_diff(self):
+        from tests.validation.multigrid import check_projected_pairs as gate
+        missing = SimpleNamespace(returncode=128)
+        with patch.object(run_case.subprocess, "run", return_value=missing), \
+             patch.object(gate, "preflight") as preflight:
+            with self.assertRaisesRegex(RuntimeError, "not available in this clone"):
+                run_case.verify_source(REPO, MOCK_REVISION)
+            preflight.assert_not_called()
+
+    def test_current_checkout_source_is_recorded_not_frozen(self):
+        from tests.validation.multigrid import check_projected_pairs as gate
+        current = dict(source(), source_tree_sha256="c" * 64)
+        with patch.object(run_case, "provenance", return_value=dict(current)), \
+             patch.object(gate, "preflight") as preflight:
+            result = run_case.verify_source(REPO, MOCK_REVISION, current=True)
+            preflight.assert_not_called()
+        self.assertEqual(result["source_mode"], "current_checkout")
+        self.assertEqual(result["source_capture"], run_case.CURRENT_CAPTURE)
+        for bad in (dict(current, dirty=True), dict(current, revision=OLD_REVISION)):
+            with self.subTest(bad=bad), patch.object(run_case, "provenance", return_value=bad):
+                with self.assertRaisesRegex(ValueError, "exact expected clean checkout"):
+                    run_case.verify_source(REPO, MOCK_REVISION, current=True)
+
+    def test_current_and_frozen_receipts_never_cross_assess(self):
+        frozen = records()
+        current = deepcopy(frozen)
+        for r in current:
+            r["source"] = dict(r["source"], source_tree_sha256="c" * 64,
+                               source_mode="current_checkout",
+                               source_capture=run_case.CURRENT_CAPTURE)
+            r["numerical_revision"], r["source_mode"] = None, "current_checkout"
+            r["schema"], r["source_recheck"] = 2, "UNCHANGED"
+        ok = assess(current, MOCK_REVISION, current=True)
+        self.assertTrue(ok["checks"]["clean_current_checkout_source"])
+        self.assertTrue(all(v for k, v in ok["checks"].items() if k.endswith(" reference")))
+        self.assertEqual((ok["source_mode"], ok["numerical_revision"]), ("current_checkout", None))
+        self.assertEqual(ok["source_tree_sha256"], "c" * 64)
+        self.assertFalse(assess(current, MOCK_REVISION)["checks"]["clean_a135252_numerical_source"])
+        self.assertFalse(assess(current, MOCK_REVISION)["passed"])
+        self.assertFalse(assess(frozen, MOCK_REVISION, current=True)["checks"]["clean_current_checkout_source"])
+        self.assertFalse(assess(current, None, current=True)["checks"]["clean_current_checkout_source"])
+        self.assertTrue(assess(frozen, MOCK_REVISION)["checks"]["clean_a135252_numerical_source"])
+
+    def test_assessor_requires_unchanged_final_source(self):
+        frozen = records()  # schema 1 receipts from the verified legacy runner 2cf9853
+        label = lambda r: f"{r['n']}/{r['depth']}/{r['explosion']} source_recheck"
+        self.assertTrue(all(assess(frozen, MOCK_REVISION)["checks"][label(r)] for r in frozen))
+        new = deepcopy(frozen)
+        for r in new:
+            r.update(schema=2, source_recheck="UNCHANGED")
+        self.assertTrue(all(assess(new, MOCK_REVISION)["checks"][label(r)] for r in new))
+        unknown_runner = dict(frozen[0]["source"], harness_sha256={"run_case.py": "a" * 64})
+        for case, mutate in (
+            ("schema 1, unverified runner", dict(source=unknown_runner)),
+            ("schema 2 without recheck", dict(schema=2)),
+            ("schema 2 changed", dict(schema=2, source_recheck="CHANGED")),
+            ("schema 2 recheck failed", dict(schema=2, source_recheck="RECHECK_FAILED")),
+            ("schema 1 with a recheck field", dict(source_recheck="UNCHANGED")),
+            ("unknown schema", dict(schema=3, source_recheck="UNCHANGED")),
+            ("no schema", dict(schema=None)),
+            ("invalid", dict(schema=2, source_recheck="UNCHANGED",
+                             invalid="SOURCE_CHANGED_DURING_RUN")),
+        ):
+            with self.subTest(case):
+                bad = deepcopy(frozen)
+                bad[0].update(mutate)
+                out = assess(bad, MOCK_REVISION)
+                self.assertFalse(out["checks"][label(bad[0])])
+                self.assertFalse(out["passed"])
+        current = deepcopy(frozen)
+        for r in current:
+            r["source"] = dict(r["source"], source_mode="current_checkout")
+            r["numerical_revision"], r["source_mode"] = None, "current_checkout"
+        # Legacy acceptance is frozen-mode only.
+        self.assertFalse(assess(current, MOCK_REVISION, current=True)["checks"][label(current[0])])
+
+    def test_legacy_runner_hashes_are_the_recorded_commits(self):
+        sys.path.insert(0, str(REPO))
+        from tests.validation.source_provenance import (HistoryUnavailable, historical_sha256,
+                                                        history_unavailable)
+        try:
+            for commit, digest in LEGACY_RUN_CASE_SHA256.items():
+                path = "tests/validation/taylor_green/run_case.py"
+                self.assertEqual(historical_sha256(REPO, commit, [path])[path], digest)
+        except HistoryUnavailable as error:
+            history_unavailable(self, error)
+        current = hashlib.sha256((REPO / "tests/validation/taylor_green/run_case.py").read_bytes())
+        self.assertNotIn(current.hexdigest(), LEGACY_RUN_CASE_SHA256.values())
+
+    def test_output_inside_checkout_is_refused_before_source_check(self):
+        argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128", "--depth", "1",
+                "--output", str(REPO / "tg-out.json"), "--expected-revision", MOCK_REVISION,
+                "--dry-run"]
+        with patch.object(sys, "argv", argv), \
+             patch.object(run_case, "verify_source", side_effect=AssertionError("work started")):
+            with self.assertRaisesRegex(SystemExit, "outside the source checkout"):
+                run_case.main()
+        self.assertFalse((REPO / "tg-out.json").exists())
+
     def test_explicit_opt_in_selector_and_dry_run_without_cuda(self):
         with tempfile.TemporaryDirectory() as tmp:
             output = Path(tmp) / "case.json"
@@ -341,7 +418,7 @@ class ProjectedPairsTests(unittest.TestCase):
                  patch.object(run_case, "runtime_identity", side_effect=AssertionError("CUDA reached")), \
                  redirect_stdout(stdout):
                 self.assertEqual(run_case.main(), 0)
-            verify.assert_called_once_with(REPO, MOCK_REVISION)
+            verify.assert_called_once_with(REPO, MOCK_REVISION, current=False)
             self.assertTrue(json.loads(stdout.getvalue())["dry_run"])
             self.assertFalse(output.exists())
 
@@ -373,7 +450,8 @@ class ProjectedPairsTests(unittest.TestCase):
         # Replace the GPU context and steppers before import. No CUDA module or flow is used.
         for early, depth, fail in ((False, 1, None), (False, 2, None), (True, 2, None),
                                    (True, 2, "resource"), (True, 2, "nonfinite"),
-                                   (True, 2, "probe_nonfinite")):
+                                   (True, 2, "probe_nonfinite"),
+                                   (False, 2, "changed"), (False, 1, "recheck_error")):
             calls, selectors = [], []
 
             class FakeRun:
@@ -428,13 +506,31 @@ class ProjectedPairsTests(unittest.TestCase):
                      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()):
-                    if fail:
+                    if fail in ("changed", "recheck_error"):
+                        verify.side_effect = [source(), dict(source(), source_tree_sha256="d" * 64)
+                                              if fail == "changed" else RuntimeError("Wrong numerical source fingerprint")]
+                        with self.assertRaisesRegex(RuntimeError, "Source changed during the run"):
+                            run_case.main()
+                    elif fail:
                         with self.assertRaises(MemoryError if fail == "resource" else FloatingPointError):
                             run_case.main()
                     else:
                         self.assertEqual(run_case.main(), 0)
                 receipt = json.loads(output.read_text())
-                verify.assert_called_once_with(REPO, MOCK_REVISION)
+                rechecked = fail in (None, "changed", "recheck_error")
+                self.assertEqual(verify.call_args_list,
+                                 [call(REPO, MOCK_REVISION, current=False)] * (2 if rechecked else 1))
+                if fail in ("changed", "recheck_error"):
+                    # Full trajectory, but invalid: never a complete-looking receipt.
+                    self.assertEqual(receipt["source_recheck"],
+                                     "CHANGED" if fail == "changed" else "RECHECK_FAILED")
+                    self.assertEqual(receipt["invalid"], "SOURCE_CHANGED_DURING_RUN")
+                    self.assertFalse(receipt["completed"])
+                    self.assertNotIn("screening_pass", receipt)
+                    self.assertEqual(receipt["source_after"] is None, fail == "recheck_error")
+                    continue
+                self.assertEqual(receipt.get("source_recheck"), None if fail else "UNCHANGED")
+                self.assertEqual(receipt["schema"], 2)
                 self.assertEqual(selectors, ["projected_linear" if early else "constant"] if depth == 2 else [])
                 self.assertEqual(receipt["source"], source())
                 self.assertEqual(receipt["runtime_identity"], RUNTIME)

tests/validation/time_refined_dfg.py

@@ -173,9 +173,8 @@ def time_refined(a, h, sched):
         cfg, pair, body, host = r.host_geometry(a.source, h, "MRT")
         context = stack.enter_context(CudaContext(0))
         backend = stack.enter_context(FixedPatchStepper(
-            cfg, context=context, coarse_boundary="channel", refinements=[r.BOUNDS],
-            level_blocks=[r.BASE_BLOCK, r.FINE_BLOCK], level_objects=[(), (body,)],
-            observe_raw_exchange=True))
+            cfg, r.BOUNDS, context=context, coarse_boundary="channel", base_block=r.BASE_BLOCK,
+            fine_block=r.FINE_BLOCK, fine_objects=(body,), observe_raw_exchange=True))
         masks = [channel_masks(pair.coarse_shape)[0] & ~pair.covered,  # as run_refined_dfg
                  (pair.fine_kind == ACTIVE) & ((host.flags.reshape(pair.fine_shape) & SOLID) == 0)]
         fresh = (r.initial_state(pair.coarse_shape, (0.0, 0.0), 1.0, root=True),

tests/validation/time_window_dfg.py

@@ -130,8 +130,8 @@ def main():
         cfg, pair, body, host = r.host_geometry(a.source, h, "MRT")
         context = stack.enter_context(CudaContext(0))
         backend = stack.enter_context(FixedPatchStepper(
-            cfg, context=context, coarse_boundary="channel", refinements=[r.BOUNDS],
-            level_blocks=[r.BASE_BLOCK, r.FINE_BLOCK], level_objects=[(), (body,)], observe_raw_exchange=True,
+            cfg, r.BOUNDS, context=context, coarse_boundary="channel", base_block=r.BASE_BLOCK,
+            fine_block=r.FINE_BLOCK, fine_objects=(body,), observe_raw_exchange=True,
             observe_window_exchange=True))
         masks = [channel_masks(pair.coarse_shape)[0] & ~pair.covered,
                  (pair.fine_kind == ACTIVE) & ((host.flags.reshape(pair.fine_shape) & SOLID) == 0)]